refactor: comment out amplitude user ID check in setUserId function

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

.agents/skills/e2e-cucumber-playwright/SKILL.md

@@ -0,0 +1,79 @@
+---
+name: e2e-cucumber-playwright
+description: Write, update, or review Dify end-to-end tests under `e2e/` that use Cucumber, Gherkin, and Playwright. Use when the task involves `.feature` files, `features/step-definitions/`, `features/support/`, `DifyWorld`, scenario tags, locator/assertion choices, or E2E testing best practices for this repository.
+---
+
+# Dify E2E Cucumber + Playwright
+
+Use this skill for Dify's repository-level E2E suite in `e2e/`. Use [`e2e/AGENTS.md`](../../../e2e/AGENTS.md) as the canonical guide for local architecture and conventions, then apply Playwright/Cucumber best practices only where they fit the current suite.
+
+## Scope
+
+- Use this skill for `.feature` files, Cucumber step definitions, `DifyWorld`, hooks, tags, and E2E review work under `e2e/`.
+- Do not use this skill for Vitest or React Testing Library work under `web/`; use `frontend-testing` instead.
+- Do not use this skill for backend test or API review tasks under `api/`.
+
+## Read Order
+
+1. Read [`e2e/AGENTS.md`](../../../e2e/AGENTS.md) first.
+2. Read only the files directly involved in the task:
+   - target `.feature` files under `e2e/features/`
+   - related step files under `e2e/features/step-definitions/`
+   - `e2e/features/support/hooks.ts` and `e2e/features/support/world.ts` when session lifecycle or shared state matters
+   - `e2e/scripts/run-cucumber.ts` and `e2e/cucumber.config.ts` when tags or execution flow matter
+3. Read [`references/playwright-best-practices.md`](references/playwright-best-practices.md) only when locator, assertion, isolation, or waiting choices are involved.
+4. Read [`references/cucumber-best-practices.md`](references/cucumber-best-practices.md) only when scenario wording, step granularity, tags, or expression design are involved.
+5. Re-check official docs with Context7 before introducing a new Playwright or Cucumber pattern.
+
+## Local Rules
+
+- `e2e/` uses Cucumber for scenarios and Playwright as the browser layer.
+- `DifyWorld` is the per-scenario context object. Type `this` as `DifyWorld` and use `async function`, not arrow functions.
+- Keep glue organized by capability under `e2e/features/step-definitions/`; use `common/` only for broadly reusable steps.
+- Browser session behavior comes from `features/support/hooks.ts`:
+  - default: authenticated session with shared storage state
+  - `@unauthenticated`: clean browser context
+  - `@authenticated`: readability/selective-run tag only unless implementation changes
+  - `@fresh`: only for `e2e:full*` flows
+- Do not import Playwright Test runner patterns that bypass the current Cucumber + `DifyWorld` architecture unless the task is explicitly about changing that architecture.
+
+## Workflow
+
+1. Rebuild local context.
+   - Inspect the target feature area.
+   - Reuse an existing step when wording and behavior already match.
+   - Add a new step only for a genuinely new user action or assertion.
+   - Keep edits close to the current capability folder unless the step is broadly reusable.
+2. Write behavior-first scenarios.
+   - Describe user-observable behavior, not DOM mechanics.
+   - Keep each scenario focused on one workflow or outcome.
+   - Keep scenarios independent and re-runnable.
+3. Write step definitions in the local style.
+   - Keep one step to one user-visible action or one assertion.
+   - Prefer Cucumber Expressions such as `{string}` and `{int}`.
+   - Scope locators to stable containers when the page has repeated elements.
+   - Avoid page-object layers or extra helper abstractions unless repeated complexity clearly justifies them.
+4. Use Playwright in the local style.
+   - Prefer user-facing locators: `getByRole`, `getByLabel`, `getByPlaceholder`, `getByText`, then `getByTestId` for explicit contracts.
+   - Use web-first `expect(...)` assertions.
+   - Do not use `waitForTimeout`, manual polling, or raw visibility checks when a locator action or retrying assertion already expresses the behavior.
+5. Validate narrowly.
+   - Run the narrowest tagged scenario or flow that exercises the change.
+   - Run `pnpm -C e2e check`.
+   - Broaden verification only when the change affects hooks, tags, setup, or shared step semantics.
+
+## Review Checklist
+
+- Does the scenario describe behavior rather than implementation?
+- Does it fit the current session model, tags, and `DifyWorld` usage?
+- Should an existing step be reused instead of adding a new one?
+- Are locators user-facing and assertions web-first?
+- Does the change introduce hidden coupling across scenarios, tags, or instance state?
+- Does it document or implement behavior that differs from the real hooks or configuration?
+
+Lead findings with correctness, flake risk, and architecture drift.
+
+## References
+
+- [`references/playwright-best-practices.md`](references/playwright-best-practices.md)
+- [`references/cucumber-best-practices.md`](references/cucumber-best-practices.md)

.agents/skills/e2e-cucumber-playwright/agents/openai.yaml

@@ -0,0 +1,4 @@
+interface:
+  display_name: "E2E Cucumber + Playwright"
+  short_description: "Write and review Dify E2E scenarios."
+  default_prompt: "Use $e2e-cucumber-playwright to write or review a Dify E2E scenario under e2e/."

.agents/skills/e2e-cucumber-playwright/references/cucumber-best-practices.md

@@ -0,0 +1,93 @@
+# Cucumber Best Practices For Dify E2E
+
+Use this reference when writing or reviewing Gherkin scenarios, step definitions, parameter expressions, and step reuse in Dify's `e2e/` suite.
+
+Official sources:
+
+- https://cucumber.io/docs/guides/10-minute-tutorial/
+- https://cucumber.io/docs/cucumber/step-definitions/
+- https://cucumber.io/docs/cucumber/cucumber-expressions/
+
+## What Matters Most
+
+### 1. Treat scenarios as executable specifications
+
+Cucumber scenarios should describe examples of behavior, not test implementation recipes.
+
+Apply it like this:
+
+- write what the user does and what should happen
+- avoid UI-internal wording such as selector details, DOM structure, or component names
+- keep language concrete enough that the scenario reads like living documentation
+
+### 2. Keep scenarios focused
+
+A scenario should usually prove one workflow or business outcome. If a scenario wanders across several unrelated behaviors, split it.
+
+In Dify's suite, this means:
+
+- one capability-focused scenario per feature path
+- no long setup chains when existing bootstrap or reusable steps already cover them
+- no hidden dependency on another scenario's side effects
+
+### 3. Reuse steps, but only when behavior really matches
+
+Good reuse reduces duplication. Bad reuse hides meaning.
+
+Prefer reuse when:
+
+- the user action is genuinely the same
+- the expected outcome is genuinely the same
+- the wording stays natural across features
+
+Write a new step when:
+
+- the behavior is materially different
+- reusing the old wording would make the scenario misleading
+- a supposedly generic step would become an implementation-detail wrapper
+
+### 4. Prefer Cucumber Expressions
+
+Use Cucumber Expressions for parameters unless regex is clearly necessary.
+
+Common examples:
+
+- `{string}` for labels, names, and visible text
+- `{int}` for counts
+- `{float}` for decimal values
+- `{word}` only when the value is truly a single token
+
+Keep expressions readable. If a step needs complicated parsing logic, first ask whether the scenario wording should be simpler.
+
+### 5. Keep step definitions thin and meaningful
+
+Step definitions are glue between Gherkin and automation, not a second abstraction language.
+
+For Dify:
+
+- type `this` as `DifyWorld`
+- use `async function`
+- keep each step to one user-visible action or assertion
+- rely on `DifyWorld` and existing support code for shared context
+- avoid leaking cross-scenario state
+
+### 6. Use tags intentionally
+
+Tags should communicate run scope or session semantics, not become ad hoc metadata.
+
+In Dify's current suite:
+
+- capability tags group related scenarios
+- `@unauthenticated` changes session behavior
+- `@authenticated` is descriptive/selective, not a behavior switch by itself
+- `@fresh` belongs to reset/full-install flows only
+
+If a proposed tag implies behavior, verify that hooks or runner configuration actually implement it.
+
+## Review Questions
+
+- Does the scenario read like a real example of product behavior?
+- Are the steps behavior-oriented instead of implementation-oriented?
+- Is a reused step still truthful in this feature?
+- Is a new tag documenting real behavior, or inventing semantics that the suite does not implement?
+- Would a new reader understand the outcome without opening the step-definition file?

.agents/skills/e2e-cucumber-playwright/references/playwright-best-practices.md

@@ -0,0 +1,96 @@
+# Playwright Best Practices For Dify E2E
+
+Use this reference when writing or reviewing locator, assertion, isolation, or synchronization logic for Dify's Cucumber-based E2E suite.
+
+Official sources:
+
+- https://playwright.dev/docs/best-practices
+- https://playwright.dev/docs/locators
+- https://playwright.dev/docs/test-assertions
+- https://playwright.dev/docs/browser-contexts
+
+## What Matters Most
+
+### 1. Keep scenarios isolated
+
+Playwright's model is built around clean browser contexts so one test does not leak into another. In Dify's suite, that principle maps to per-scenario session setup in `features/support/hooks.ts` and `DifyWorld`.
+
+Apply it like this:
+
+- do not depend on another scenario having run first
+- do not persist ad hoc scenario state outside `DifyWorld`
+- do not couple ordinary scenarios to `@fresh` behavior
+- when a flow needs special auth/session semantics, express that through the existing tag model or explicit hook changes
+
+### 2. Prefer user-facing locators
+
+Playwright recommends built-in locators that reflect what users perceive on the page.
+
+Preferred order in this repository:
+
+1. `getByRole`
+2. `getByLabel`
+3. `getByPlaceholder`
+4. `getByText`
+5. `getByTestId` when an explicit test contract is the most stable option
+
+Avoid raw CSS/XPath selectors unless no stable user-facing contract exists and adding one is not practical.
+
+Also remember:
+
+- repeated content usually needs scoping to a stable container
+- exact text matching is often too brittle when role/name or label already exists
+- `getByTestId` is acceptable when semantics are weak but the contract is intentional
+
+### 3. Use web-first assertions
+
+Playwright assertions auto-wait and retry. Prefer them over manual state inspection.
+
+Prefer:
+
+- `await expect(page).toHaveURL(...)`
+- `await expect(locator).toBeVisible()`
+- `await expect(locator).toBeHidden()`
+- `await expect(locator).toBeEnabled()`
+- `await expect(locator).toHaveText(...)`
+
+Avoid:
+
+- `expect(await locator.isVisible()).toBe(true)`
+- custom polling loops for DOM state
+- `waitForTimeout` as synchronization
+
+If a condition genuinely needs custom retry logic, use Playwright's polling/assertion tools deliberately and keep that choice local and explicit.
+
+### 4. Let actions wait for actionability
+
+Locator actions already wait for the element to be actionable. Do not preface every click/fill with extra timing logic unless the action needs a specific visible/ready assertion for clarity.
+
+Good pattern:
+
+- assert a meaningful visible state when that is part of the behavior
+- then click/fill/select via locator APIs
+
+Bad pattern:
+
+- stack arbitrary waits before every action
+- wait on unstable implementation details instead of the visible state the user cares about
+
+### 5. Match debugging to the current suite
+
+Playwright's wider ecosystem supports traces and rich debugging tools. Dify's current suite already captures:
+
+- full-page screenshots
+- page HTML
+- console errors
+- page errors
+
+Use the existing artifact flow by default. If a task is specifically about improving diagnostics, confirm the change fits the current Cucumber architecture before importing broader Playwright tooling.
+
+## Review Questions
+
+- Would this locator survive DOM refactors that do not change user-visible behavior?
+- Is this assertion using Playwright's retrying semantics?
+- Is any explicit wait masking a real readiness problem?
+- Does this code preserve per-scenario isolation?
+- Is a new abstraction really needed, or does it bypass the existing `DifyWorld` + step-definition model?

.claude/skills/e2e-cucumber-playwright

@@ -0,0 +1 @@
+../../.agents/skills/e2e-cucumber-playwright

.github/workflows/docker-build.yml

@@ -6,14 +6,7 @@ on:
       - "main"
     paths:
       - api/Dockerfile
-      - web/docker/**
       - web/Dockerfile
-      - packages/**
-      - package.json
-      - pnpm-lock.yaml
-      - pnpm-workspace.yaml
-      - .npmrc
-      - .nvmrc
 
 concurrency:
   group: docker-build-${{ github.head_ref || github.run_id }}

.github/workflows/main-ci.yml

@@ -92,6 +92,7 @@ jobs:
             vdb:
               - 'api/core/rag/datasource/**'
               - 'api/tests/integration_tests/vdb/**'
+              - 'api/providers/vdb/*/tests/**'
               - '.github/workflows/vdb-tests.yml'
               - '.github/workflows/expose_service_ports.sh'
               - 'docker/.env.example'

.github/workflows/vdb-tests-full.yml

@@ -89,7 +89,7 @@ jobs:
           cp api/tests/integration_tests/.env.example api/tests/integration_tests/.env
 
 #      - name: Check VDB Ready (TiDB)
-#        run: uv run --project api python api/tests/integration_tests/vdb/tidb_vector/check_tiflash_ready.py
+#        run: uv run --project api python api/providers/vdb/tidb-vector/tests/integration_tests/check_tiflash_ready.py
 
       - name: Test Vector Stores
         run: uv run --project api bash dev/pytest/pytest_vdb.sh

.github/workflows/vdb-tests.yml

@@ -81,12 +81,12 @@ jobs:
           cp api/tests/integration_tests/.env.example api/tests/integration_tests/.env
 
 #      - name: Check VDB Ready (TiDB)
-#        run: uv run --project api python api/tests/integration_tests/vdb/tidb_vector/check_tiflash_ready.py
+#        run: uv run --project api python api/providers/vdb/tidb-vector/tests/integration_tests/check_tiflash_ready.py
 
       - name: Test Vector Stores
         run: |
           uv run --project api pytest --timeout "${PYTEST_TIMEOUT:-180}" \
-            api/tests/integration_tests/vdb/chroma \
-            api/tests/integration_tests/vdb/pgvector \
-            api/tests/integration_tests/vdb/qdrant \
-            api/tests/integration_tests/vdb/weaviate
+            api/providers/vdb/vdb-chroma/tests/integration_tests \
+            api/providers/vdb/vdb-pgvector/tests/integration_tests \
+            api/providers/vdb/vdb-qdrant/tests/integration_tests \
+            api/providers/vdb/vdb-weaviate/tests/integration_tests

api/.ruff.toml

@@ -69,8 +69,6 @@ ignore = [
     "FURB152", # math-constant
     "UP007",   # non-pep604-annotation
     "UP032",   # f-string
-    "UP045",   # non-pep604-annotation-optional
-    "B005",    # strip-with-multi-characters
     "B006",    # mutable-argument-default
     "B007",    # unused-loop-control-variable
     "B026",    # star-arg-unpacking-after-keyword-arg
@@ -84,7 +82,6 @@ ignore = [
     "SIM102",  # collapsible-if
     "SIM103",  # needless-bool
     "SIM105",  # suppressible-exception
-    "SIM107",  # return-in-try-except-finally
     "SIM108",  # if-else-block-instead-of-if-exp
     "SIM113",  # enumerate-for-loop
     "SIM117",  # multiple-with-statements
@@ -93,29 +90,16 @@ ignore = [
 ]
 
 [lint.per-file-ignores]
-"__init__.py" = [
-    "F401", # unused-import
-    "F811", # redefined-while-unused
-]
 "configs/*" = [
     "N802", # invalid-function-name
 ]
-"graphon/model_runtime/callbacks/base_callback.py" = ["T201"]
-"core/workflow/callbacks/workflow_logging_callback.py" = ["T201"]
 "libs/gmpy2_pkcs10aep_cipher.py" = [
     "N803", # invalid-argument-name
 ]
 "tests/*" = [
-    "F811", # redefined-while-unused
     "T201", # allow print in tests,
     "S110", # allow ignoring exceptions in tests code (currently)
-
 ]
-"controllers/console/explore/trial.py" = ["TID251"]
-"controllers/console/human_input_form.py" = ["TID251"]
-"controllers/web/human_input_form.py" = ["TID251"]
-
-[lint.flake8-tidy-imports]
 
 [lint.flake8-tidy-imports.banned-api."flask_restx.reqparse"]
 msg = "Use Pydantic payload/query models instead of reqparse."

api/Dockerfile

@@ -21,8 +21,9 @@ RUN apt-get update \
           # for building gmpy2
           libmpfr-dev libmpc-dev
 
-# Install Python dependencies
+# Install Python dependencies (workspace members under providers/vdb/)
 COPY pyproject.toml uv.lock ./
+COPY providers ./providers
 RUN uv sync --locked --no-dev
 
 # production stage

api/commands/vector.py

@@ -341,11 +341,10 @@ def add_qdrant_index(field: str):
             click.echo(click.style("No dataset collection bindings found.", fg="red"))
             return
         import qdrant_client
+        from dify_vdb_qdrant.qdrant_vector import PathQdrantParams, QdrantConfig
         from qdrant_client.http.exceptions import UnexpectedResponse
         from qdrant_client.http.models import PayloadSchemaType
 
-        from core.rag.datasource.vdb.qdrant.qdrant_vector import PathQdrantParams, QdrantConfig
-
         for binding in bindings:
             if dify_config.QDRANT_URL is None:
                 raise ValueError("Qdrant URL is required.")

api/configs/feature/__init__.py

@@ -1366,6 +1366,32 @@ class SandboxExpiredRecordsCleanConfig(BaseSettings):
     )
 
 
+class EvaluationConfig(BaseSettings):
+    """
+    Configuration for evaluation runtime
+    """
+
+    EVALUATION_FRAMEWORK: str = Field(
+        description="Evaluation framework to use (ragas/deepeval/none)",
+        default="none",
+    )
+
+    EVALUATION_MAX_CONCURRENT_RUNS: PositiveInt = Field(
+        description="Maximum number of concurrent evaluation runs per tenant",
+        default=3,
+    )
+
+    EVALUATION_MAX_DATASET_ROWS: PositiveInt = Field(
+        description="Maximum number of rows allowed in an evaluation dataset",
+        default=500,
+    )
+
+    EVALUATION_TASK_TIMEOUT: PositiveInt = Field(
+        description="Timeout in seconds for a single evaluation task",
+        default=3600,
+    )
+
+
 class FeatureConfig(
     # place the configs in alphabet order
     AppExecutionConfig,
@@ -1378,6 +1404,7 @@ class FeatureConfig(
     MarketplaceConfig,
     DataSetConfig,
     EndpointConfig,
+    EvaluationConfig,
     FileAccessConfig,
     FileUploadConfig,
     HttpConfig,

api/configs/middleware/__init__.py

@@ -160,6 +160,16 @@ class DatabaseConfig(BaseSettings):
         default="",
     )
 
+    DB_SESSION_TIMEZONE_OVERRIDE: str = Field(
+        description=(
+            "PostgreSQL session timezone override injected via startup options."
+            " Default is 'UTC' for out-of-the-box consistency."
+            " Set to empty string to disable app-level timezone injection, for example when using RDS Proxy"
+            " together with a database-side default timezone."
+        ),
+        default="UTC",
+    )
+
     @computed_field  # type: ignore[prop-decorator]
     @property
     def SQLALCHEMY_DATABASE_URI_SCHEME(self) -> str:
@@ -227,12 +237,13 @@ class DatabaseConfig(BaseSettings):
         connect_args: dict[str, str] = {}
         # Use the dynamic SQLALCHEMY_DATABASE_URI_SCHEME property
         if self.SQLALCHEMY_DATABASE_URI_SCHEME.startswith("postgresql"):
-            timezone_opt = "-c timezone=UTC"
-            if options:
-                merged_options = f"{options} {timezone_opt}"
-            else:
-                merged_options = timezone_opt
-            connect_args = {"options": merged_options}
+            merged_options = options.strip()
+            session_timezone_override = self.DB_SESSION_TIMEZONE_OVERRIDE.strip()
+            if session_timezone_override:
+                timezone_opt = f"-c timezone={session_timezone_override}"
+                merged_options = f"{merged_options} {timezone_opt}".strip() if merged_options else timezone_opt
+            if merged_options:
+                connect_args = {"options": merged_options}
 
         result: SQLAlchemyEngineOptionsDict = {
             "pool_size": self.SQLALCHEMY_POOL_SIZE,

api/configs/middleware/vdb/hologres_config.py

@@ -1,4 +1,3 @@
-from holo_search_sdk.types import BaseQuantizationType, DistanceType, TokenizerType
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
@@ -42,17 +41,17 @@ class HologresConfig(BaseSettings):
         default="public",
     )
 
-    HOLOGRES_TOKENIZER: TokenizerType = Field(
+    HOLOGRES_TOKENIZER: str = Field(
         description="Tokenizer for full-text search index (e.g., 'jieba', 'ik', 'standard', 'simple').",
         default="jieba",
     )
 
-    HOLOGRES_DISTANCE_METHOD: DistanceType = Field(
+    HOLOGRES_DISTANCE_METHOD: str = Field(
         description="Distance method for vector index (e.g., 'Cosine', 'Euclidean', 'InnerProduct').",
         default="Cosine",
     )
 
-    HOLOGRES_BASE_QUANTIZATION_TYPE: BaseQuantizationType = Field(
+    HOLOGRES_BASE_QUANTIZATION_TYPE: str = Field(
         description="Base quantization type for vector index (e.g., 'rabitq', 'sq8', 'fp16', 'fp32').",
         default="rabitq",
     )

api/configs/middleware/vdb/iris_config.py

@@ -1,5 +1,7 @@
 """Configuration for InterSystems IRIS vector database."""
 
+from typing import Any
+
 from pydantic import Field, PositiveInt, model_validator
 from pydantic_settings import BaseSettings
 
@@ -64,7 +66,7 @@ class IrisVectorConfig(BaseSettings):
 
     @model_validator(mode="before")
     @classmethod
-    def validate_config(cls, values: dict) -> dict:
+    def validate_config(cls, values: dict[str, Any]) -> dict[str, Any]:
         """Validate IRIS configuration values.
 
         Args:

api/controllers/common/controller_schemas.py

@@ -1,4 +1,5 @@
 from typing import Any, Literal
+from uuid import UUID
 
 from pydantic import BaseModel, Field, model_validator
 
@@ -23,9 +24,9 @@ class ConversationRenamePayload(BaseModel):
 
 
 class MessageListQuery(BaseModel):
-    conversation_id: UUIDStrOrEmpty
-    first_id: UUIDStrOrEmpty | None = None
-    limit: int = Field(default=20, ge=1, le=100)
+    conversation_id: UUIDStrOrEmpty = Field(description="Conversation UUID")
+    first_id: UUIDStrOrEmpty | None = Field(default=None, description="First message ID for pagination")
+    limit: int = Field(default=20, ge=1, le=100, description="Number of messages to return (1-100)")
 
 
 class MessageFeedbackPayload(BaseModel):
@@ -69,11 +70,35 @@ class WorkflowUpdatePayload(BaseModel):
     marked_comment: str | None = Field(default=None, max_length=100)
 
 
+# --- Dataset schemas ---
+
+
+DOCUMENT_BATCH_DOWNLOAD_ZIP_MAX_DOCS = 100
+
+
+class ChildChunkCreatePayload(BaseModel):
+    content: str
+
+
+class ChildChunkUpdatePayload(BaseModel):
+    content: str
+
+
+class DocumentBatchDownloadZipPayload(BaseModel):
+    """Request payload for bulk downloading documents as a zip archive."""
+
+    document_ids: list[UUID] = Field(..., min_length=1, max_length=DOCUMENT_BATCH_DOWNLOAD_ZIP_MAX_DOCS)
+
+
+class MetadataUpdatePayload(BaseModel):
+    name: str
+
+
 # --- Audio schemas ---
 
 
 class TextToAudioPayload(BaseModel):
-    message_id: str | None = None
-    voice: str | None = None
-    text: str | None = None
-    streaming: bool | None = None
+    message_id: str | None = Field(default=None, description="Message ID")
+    voice: str | None = Field(default=None, description="Voice to use for TTS")
+    text: str | None = Field(default=None, description="Text to convert to audio")
+    streaming: bool | None = Field(default=None, description="Enable streaming response")

api/controllers/console/__init__.py

@@ -107,6 +107,9 @@ from .datasets.rag_pipeline import (
     rag_pipeline_workflow,
 )
 
+# Import evaluation controllers
+from .evaluation import evaluation
+
 # Import explore controllers
 from .explore import (
     banner,
@@ -117,6 +120,9 @@ from .explore import (
     trial,
 )
 
+# Import snippet controllers
+from .snippets import snippet_workflow, snippet_workflow_draft_variable
+
 # Import tag controllers
 from .tag import tags
 
@@ -130,6 +136,7 @@ from .workspace import (
     model_providers,
     models,
     plugin,
+    snippets,
     tool_providers,
     trigger_providers,
     workspace,
@@ -167,6 +174,7 @@ __all__ = [
     "datasource_content_preview",
     "email_register",
     "endpoint",
+    "evaluation",
     "extension",
     "external",
     "feature",
@@ -201,6 +209,9 @@ __all__ = [
     "saved_message",
     "setup",
     "site",
+    "snippet_workflow",
+    "snippet_workflow_draft_variable",
+    "snippets",
     "spec",
     "statistic",
     "tags",

api/controllers/console/apikey.py

@@ -1,12 +1,16 @@
+from datetime import datetime
+
 import flask_restx
-from flask_restx import Resource, fields, marshal_with
+from flask_restx import Resource
 from flask_restx._http import HTTPStatus
+from pydantic import field_validator
 from sqlalchemy import delete, func, select
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import Forbidden
 
+from controllers.common.schema import register_schema_models
 from extensions.ext_database import db
-from libs.helper import TimestampField
+from fields.base import ResponseModel
 from libs.login import current_account_with_tenant, login_required
 from models.dataset import Dataset
 from models.enums import ApiTokenType
@@ -16,21 +20,31 @@ from services.api_token_service import ApiTokenCache
 from . import console_ns
 from .wraps import account_initialization_required, edit_permission_required, setup_required
 
-api_key_fields = {
-    "id": fields.String,
-    "type": fields.String,
-    "token": fields.String,
-    "last_used_at": TimestampField,
-    "created_at": TimestampField,
-}
 
-api_key_item_model = console_ns.model("ApiKeyItem", api_key_fields)
+def _to_timestamp(value: datetime | int | None) -> int | None:
+    if isinstance(value, datetime):
+        return int(value.timestamp())
+    return value
 
-api_key_list = {"data": fields.List(fields.Nested(api_key_item_model), attribute="items")}
 
-api_key_list_model = console_ns.model(
-    "ApiKeyList", {"data": fields.List(fields.Nested(api_key_item_model), attribute="items")}
-)
+class ApiKeyItem(ResponseModel):
+    id: str
+    type: str
+    token: str
+    last_used_at: int | None = None
+    created_at: int | None = None
+
+    @field_validator("last_used_at", "created_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return _to_timestamp(value)
+
+
+class ApiKeyList(ResponseModel):
+    data: list[ApiKeyItem]
+
+
+register_schema_models(console_ns, ApiKeyItem, ApiKeyList)
 
 
 def _get_resource(resource_id, tenant_id, resource_model):
@@ -54,7 +68,6 @@ class BaseApiKeyListResource(Resource):
     token_prefix: str | None = None
     max_keys = 10
 
-    @marshal_with(api_key_list_model)
     def get(self, resource_id):
         assert self.resource_id_field is not None, "resource_id_field must be set"
         resource_id = str(resource_id)
@@ -66,9 +79,8 @@ class BaseApiKeyListResource(Resource):
                 ApiToken.type == self.resource_type, getattr(ApiToken, self.resource_id_field) == resource_id
             )
         ).all()
-        return {"items": keys}
+        return ApiKeyList.model_validate({"data": keys}, from_attributes=True).model_dump(mode="json")
 
-    @marshal_with(api_key_item_model)
     @edit_permission_required
     def post(self, resource_id):
         assert self.resource_id_field is not None, "resource_id_field must be set"
@@ -100,7 +112,7 @@ class BaseApiKeyListResource(Resource):
         api_token.type = self.resource_type
         db.session.add(api_token)
         db.session.commit()
-        return api_token, 201
+        return ApiKeyItem.model_validate(api_token, from_attributes=True).model_dump(mode="json"), 201
 
 
 class BaseApiKeyResource(Resource):
@@ -147,7 +159,7 @@ class AppApiKeyListResource(BaseApiKeyListResource):
     @console_ns.doc("get_app_api_keys")
     @console_ns.doc(description="Get all API keys for an app")
     @console_ns.doc(params={"resource_id": "App ID"})
-    @console_ns.response(200, "Success", api_key_list_model)
+    @console_ns.response(200, "API keys retrieved successfully", console_ns.models[ApiKeyList.__name__])
     def get(self, resource_id):  # type: ignore
         """Get all API keys for an app"""
         return super().get(resource_id)
@@ -155,7 +167,7 @@ class AppApiKeyListResource(BaseApiKeyListResource):
     @console_ns.doc("create_app_api_key")
     @console_ns.doc(description="Create a new API key for an app")
     @console_ns.doc(params={"resource_id": "App ID"})
-    @console_ns.response(201, "API key created successfully", api_key_item_model)
+    @console_ns.response(201, "API key created successfully", console_ns.models[ApiKeyItem.__name__])
     @console_ns.response(400, "Maximum keys exceeded")
     def post(self, resource_id):  # type: ignore
         """Create a new API key for an app"""
@@ -187,7 +199,7 @@ class DatasetApiKeyListResource(BaseApiKeyListResource):
     @console_ns.doc("get_dataset_api_keys")
     @console_ns.doc(description="Get all API keys for a dataset")
     @console_ns.doc(params={"resource_id": "Dataset ID"})
-    @console_ns.response(200, "Success", api_key_list_model)
+    @console_ns.response(200, "API keys retrieved successfully", console_ns.models[ApiKeyList.__name__])
     def get(self, resource_id):  # type: ignore
         """Get all API keys for a dataset"""
         return super().get(resource_id)
@@ -195,7 +207,7 @@ class DatasetApiKeyListResource(BaseApiKeyListResource):
     @console_ns.doc("create_dataset_api_key")
     @console_ns.doc(description="Create a new API key for a dataset")
     @console_ns.doc(params={"resource_id": "Dataset ID"})
-    @console_ns.response(201, "API key created successfully", api_key_item_model)
+    @console_ns.response(201, "API key created successfully", console_ns.models[ApiKeyItem.__name__])
     @console_ns.response(400, "Maximum keys exceeded")
     def post(self, resource_id):  # type: ignore
         """Create a new API key for a dataset"""

api/controllers/console/app/advanced_prompt_template.py

@@ -5,7 +5,7 @@ from pydantic import BaseModel, Field
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from libs.login import login_required
-from services.advanced_prompt_template_service import AdvancedPromptTemplateService
+from services.advanced_prompt_template_service import AdvancedPromptTemplateArgs, AdvancedPromptTemplateService
 
 
 class AdvancedPromptTemplateQuery(BaseModel):
@@ -35,5 +35,10 @@ class AdvancedPromptTemplateList(Resource):
     @account_initialization_required
     def get(self):
         args = AdvancedPromptTemplateQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
-
-        return AdvancedPromptTemplateService.get_prompt(args.model_dump())
+        prompt_args: AdvancedPromptTemplateArgs = {
+            "app_mode": args.app_mode,
+            "model_mode": args.model_mode,
+            "model_name": args.model_name,
+            "has_context": args.has_context,
+        }
+        return AdvancedPromptTemplateService.get_prompt(prompt_args)

api/controllers/console/app/app.py

@@ -338,6 +338,7 @@ class AppPartial(ResponseModel):
     create_user_name: str | None = None
     author_name: str | None = None
     has_draft_trigger: bool | None = None
+    workflow_type: str | None = None
 
     @computed_field(return_type=str | None)  # type: ignore
     @property
@@ -372,6 +373,7 @@ class AppDetail(ResponseModel):
     updated_by: str | None = None
     updated_at: int | None = None
     access_mode: str | None = None
+    workflow_type: str | None = None
     tags: list[Tag] = Field(default_factory=list)
 
     @field_validator("created_at", "updated_at", mode="before")
@@ -514,6 +516,17 @@ class AppListApi(Resource):
         for app in app_pagination.items:
             app.has_draft_trigger = str(app.id) in draft_trigger_app_ids
 
+        workflow_ids = [str(app.workflow_id) for app in app_pagination.items if app.workflow_id]
+        workflow_type_map: dict[str, str] = {}
+        if workflow_ids:
+            rows = db.session.execute(
+                select(Workflow.id, Workflow.type).where(Workflow.id.in_(workflow_ids))
+            ).all()
+            workflow_type_map = {str(row.id): row.type for row in rows}
+
+        for app in app_pagination.items:
+            app.workflow_type = workflow_type_map.get(str(app.workflow_id)) if app.workflow_id else None
+
         pagination_model = AppPagination.model_validate(app_pagination, from_attributes=True)
         return pagination_model.model_dump(mode="json"), 200
 
@@ -560,6 +573,14 @@ class AppApi(Resource):
             app_setting = EnterpriseService.WebAppAuth.get_app_access_mode_by_id(app_id=str(app_model.id))
             app_model.access_mode = app_setting.access_mode
 
+        if app_model.workflow_id:
+            row = db.session.execute(
+                select(Workflow.type).where(Workflow.id == app_model.workflow_id)
+            ).scalar()
+            app_model.workflow_type = row if row else None
+        else:
+            app_model.workflow_type = None
+
         response_model = AppDetailWithSite.model_validate(app_model, from_attributes=True)
         return response_model.model_dump(mode="json")
 

api/controllers/console/app/app_import.py

@@ -1,7 +1,8 @@
-from flask_restx import Resource, fields, marshal_with
+from flask_restx import Resource
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import sessionmaker
 
+from controllers.common.schema import register_schema_models
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import (
     account_initialization_required,
@@ -10,35 +11,15 @@ from controllers.console.wraps import (
     setup_required,
 )
 from extensions.ext_database import db
-from fields.app_fields import (
-    app_import_check_dependencies_fields,
-    app_import_fields,
-    leaked_dependency_fields,
-)
 from libs.login import current_account_with_tenant, login_required
 from models.model import App
-from services.app_dsl_service import AppDslService
+from services.app_dsl_service import AppDslService, Import
 from services.enterprise.enterprise_service import EnterpriseService
-from services.entities.dsl_entities import ImportStatus
+from services.entities.dsl_entities import CheckDependenciesResult, ImportStatus
 from services.feature_service import FeatureService
 
 from .. import console_ns
 
-# Register models for flask_restx to avoid dict type issues in Swagger
-# Register base model first
-leaked_dependency_model = console_ns.model("LeakedDependency", leaked_dependency_fields)
-
-app_import_model = console_ns.model("AppImport", app_import_fields)
-
-# For nested models, need to replace nested dict with registered model
-app_import_check_dependencies_fields_copy = app_import_check_dependencies_fields.copy()
-app_import_check_dependencies_fields_copy["leaked_dependencies"] = fields.List(fields.Nested(leaked_dependency_model))
-app_import_check_dependencies_model = console_ns.model(
-    "AppImportCheckDependencies", app_import_check_dependencies_fields_copy
-)
-
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class AppImportPayload(BaseModel):
     mode: str = Field(..., description="Import mode")
@@ -52,18 +33,18 @@ class AppImportPayload(BaseModel):
     app_id: str | None = Field(None)
 
 
-console_ns.schema_model(
-    AppImportPayload.__name__, AppImportPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
+register_schema_models(console_ns, AppImportPayload, Import, CheckDependenciesResult)
 
 
 @console_ns.route("/apps/imports")
 class AppImportApi(Resource):
     @console_ns.expect(console_ns.models[AppImportPayload.__name__])
+    @console_ns.response(200, "Import completed", console_ns.models[Import.__name__])
+    @console_ns.response(202, "Import pending confirmation", console_ns.models[Import.__name__])
+    @console_ns.response(400, "Import failed", console_ns.models[Import.__name__])
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(app_import_model)
     @cloud_edition_billing_resource_check("apps")
     @edit_permission_required
     def post(self):
@@ -104,10 +85,11 @@ class AppImportApi(Resource):
 
 @console_ns.route("/apps/imports/<string:import_id>/confirm")
 class AppImportConfirmApi(Resource):
+    @console_ns.response(200, "Import confirmed", console_ns.models[Import.__name__])
+    @console_ns.response(400, "Import failed", console_ns.models[Import.__name__])
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(app_import_model)
     @edit_permission_required
     def post(self, import_id):
         # Check user role first
@@ -128,11 +110,11 @@ class AppImportConfirmApi(Resource):
 
 @console_ns.route("/apps/imports/<string:app_id>/check-dependencies")
 class AppImportCheckDependenciesApi(Resource):
+    @console_ns.response(200, "Dependencies checked", console_ns.models[CheckDependenciesResult.__name__])
     @setup_required
     @login_required
     @get_app_model
     @account_initialization_required
-    @marshal_with(app_import_check_dependencies_model)
     @edit_permission_required
     def get(self, app_model: App):
         with sessionmaker(db.engine).begin() as session:

api/controllers/console/app/mcp_server.py

@@ -1,39 +1,68 @@
 import json
+from datetime import datetime
+from typing import Any
 
-from flask_restx import Resource, marshal_with
-from pydantic import BaseModel, Field
+from flask_restx import Resource
+from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import select
 from werkzeug.exceptions import NotFound
 
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
 from extensions.ext_database import db
-from fields.app_fields import app_server_fields
+from fields.base import ResponseModel
 from libs.login import current_account_with_tenant, login_required
 from models.enums import AppMCPServerStatus
 from models.model import AppMCPServer
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
-# Register model for flask_restx to avoid dict type issues in Swagger
-app_server_model = console_ns.model("AppServer", app_server_fields)
+def _to_timestamp(value: datetime | int | None) -> int | None:
+    if isinstance(value, datetime):
+        return int(value.timestamp())
+    return value
 
 
 class MCPServerCreatePayload(BaseModel):
     description: str | None = Field(default=None, description="Server description")
-    parameters: dict = Field(..., description="Server parameters configuration")
+    parameters: dict[str, Any] = Field(..., description="Server parameters configuration")
 
 
 class MCPServerUpdatePayload(BaseModel):
     id: str = Field(..., description="Server ID")
     description: str | None = Field(default=None, description="Server description")
-    parameters: dict = Field(..., description="Server parameters configuration")
+    parameters: dict[str, Any] = Field(..., description="Server parameters configuration")
     status: str | None = Field(default=None, description="Server status")
 
 
-for model in (MCPServerCreatePayload, MCPServerUpdatePayload):
-    console_ns.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+class AppMCPServerResponse(ResponseModel):
+    id: str
+    name: str
+    server_code: str
+    description: str
+    status: str
+    parameters: dict[str, Any] | list[Any] | str
+    created_at: int | None = None
+    updated_at: int | None = None
+
+    @field_validator("parameters", mode="before")
+    @classmethod
+    def _parse_json_string(cls, value: Any) -> Any:
+        if isinstance(value, str):
+            try:
+                return json.loads(value)
+            except (json.JSONDecodeError, TypeError):
+                return value
+        return value
+
+    @field_validator("created_at", "updated_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return _to_timestamp(value)
+
+
+register_schema_models(console_ns, MCPServerCreatePayload, MCPServerUpdatePayload, AppMCPServerResponse)
 
 
 @console_ns.route("/apps/<uuid:app_id>/server")
@@ -41,27 +70,27 @@ class AppMCPServerController(Resource):
     @console_ns.doc("get_app_mcp_server")
     @console_ns.doc(description="Get MCP server configuration for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "MCP server configuration retrieved successfully", app_server_model)
+    @console_ns.response(200, "Server configuration", console_ns.models[AppMCPServerResponse.__name__])
     @login_required
     @account_initialization_required
     @setup_required
     @get_app_model
-    @marshal_with(app_server_model)
     def get(self, app_model):
         server = db.session.scalar(select(AppMCPServer).where(AppMCPServer.app_id == app_model.id).limit(1))
-        return server
+        if server is None:
+            return {}
+        return AppMCPServerResponse.model_validate(server, from_attributes=True).model_dump(mode="json")
 
     @console_ns.doc("create_app_mcp_server")
     @console_ns.doc(description="Create MCP server configuration for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
     @console_ns.expect(console_ns.models[MCPServerCreatePayload.__name__])
-    @console_ns.response(201, "MCP server configuration created successfully", app_server_model)
+    @console_ns.response(200, "Server created", console_ns.models[AppMCPServerResponse.__name__])
     @console_ns.response(403, "Insufficient permissions")
     @account_initialization_required
     @get_app_model
     @login_required
     @setup_required
-    @marshal_with(app_server_model)
     @edit_permission_required
     def post(self, app_model):
         _, current_tenant_id = current_account_with_tenant()
@@ -82,20 +111,19 @@ class AppMCPServerController(Resource):
         )
         db.session.add(server)
         db.session.commit()
-        return server
+        return AppMCPServerResponse.model_validate(server, from_attributes=True).model_dump(mode="json")
 
     @console_ns.doc("update_app_mcp_server")
     @console_ns.doc(description="Update MCP server configuration for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
     @console_ns.expect(console_ns.models[MCPServerUpdatePayload.__name__])
-    @console_ns.response(200, "MCP server configuration updated successfully", app_server_model)
+    @console_ns.response(200, "Server updated", console_ns.models[AppMCPServerResponse.__name__])
     @console_ns.response(403, "Insufficient permissions")
     @console_ns.response(404, "Server not found")
     @get_app_model
     @login_required
     @setup_required
     @account_initialization_required
-    @marshal_with(app_server_model)
     @edit_permission_required
     def put(self, app_model):
         payload = MCPServerUpdatePayload.model_validate(console_ns.payload or {})
@@ -118,7 +146,7 @@ class AppMCPServerController(Resource):
             except ValueError:
                 raise ValueError("Invalid status")
         db.session.commit()
-        return server
+        return AppMCPServerResponse.model_validate(server, from_attributes=True).model_dump(mode="json")
 
 
 @console_ns.route("/apps/<uuid:server_id>/server/refresh")
@@ -126,13 +154,12 @@ class AppMCPServerRefreshController(Resource):
     @console_ns.doc("refresh_app_mcp_server")
     @console_ns.doc(description="Refresh MCP server configuration and regenerate server code")
     @console_ns.doc(params={"server_id": "Server ID"})
-    @console_ns.response(200, "MCP server refreshed successfully", app_server_model)
+    @console_ns.response(200, "Server refreshed", console_ns.models[AppMCPServerResponse.__name__])
     @console_ns.response(403, "Insufficient permissions")
     @console_ns.response(404, "Server not found")
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(app_server_model)
     @edit_permission_required
     def get(self, server_id):
         _, current_tenant_id = current_account_with_tenant()
@@ -145,4 +172,4 @@ class AppMCPServerRefreshController(Resource):
             raise NotFound()
         server.server_code = AppMCPServer.generate_server_code(16)
         db.session.commit()
-        return server
+        return AppMCPServerResponse.model_validate(server, from_attributes=True).model_dump(mode="json")

api/controllers/console/app/site.py

@@ -1,11 +1,12 @@
 from typing import Literal
 
-from flask_restx import Resource, marshal_with
+from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import select
 from werkzeug.exceptions import NotFound
 
 from constants.languages import supported_language
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import (
@@ -15,13 +16,11 @@ from controllers.console.wraps import (
     setup_required,
 )
 from extensions.ext_database import db
-from fields.app_fields import app_site_fields
+from fields.base import ResponseModel
 from libs.datetime_utils import naive_utc_now
 from libs.login import current_account_with_tenant, login_required
 from models import Site
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class AppSiteUpdatePayload(BaseModel):
     title: str | None = Field(default=None)
@@ -49,13 +48,26 @@ class AppSiteUpdatePayload(BaseModel):
         return supported_language(value)
 
 
-console_ns.schema_model(
-    AppSiteUpdatePayload.__name__,
-    AppSiteUpdatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
+class AppSiteResponse(ResponseModel):
+    app_id: str
+    access_token: str | None = Field(default=None, validation_alias="code")
+    code: str | None = None
+    title: str
+    icon: str | None = None
+    icon_background: str | None = None
+    description: str | None = None
+    default_language: str
+    customize_domain: str | None = None
+    copyright: str | None = None
+    privacy_policy: str | None = None
+    custom_disclaimer: str | None = None
+    customize_token_strategy: str
+    prompt_public: bool
+    show_workflow_steps: bool
+    use_icon_as_answer_icon: bool
 
-# Register model for flask_restx to avoid dict type issues in Swagger
-app_site_model = console_ns.model("AppSite", app_site_fields)
+
+register_schema_models(console_ns, AppSiteUpdatePayload, AppSiteResponse)
 
 
 @console_ns.route("/apps/<uuid:app_id>/site")
@@ -64,7 +76,7 @@ class AppSite(Resource):
     @console_ns.doc(description="Update application site configuration")
     @console_ns.doc(params={"app_id": "Application ID"})
     @console_ns.expect(console_ns.models[AppSiteUpdatePayload.__name__])
-    @console_ns.response(200, "Site configuration updated successfully", app_site_model)
+    @console_ns.response(200, "Site configuration updated successfully", console_ns.models[AppSiteResponse.__name__])
     @console_ns.response(403, "Insufficient permissions")
     @console_ns.response(404, "App not found")
     @setup_required
@@ -72,7 +84,6 @@ class AppSite(Resource):
     @edit_permission_required
     @account_initialization_required
     @get_app_model
-    @marshal_with(app_site_model)
     def post(self, app_model):
         args = AppSiteUpdatePayload.model_validate(console_ns.payload or {})
         current_user, _ = current_account_with_tenant()
@@ -106,7 +117,7 @@ class AppSite(Resource):
         site.updated_at = naive_utc_now()
         db.session.commit()
 
-        return site
+        return AppSiteResponse.model_validate(site, from_attributes=True).model_dump(mode="json")
 
 
 @console_ns.route("/apps/<uuid:app_id>/site/access-token-reset")
@@ -114,7 +125,7 @@ class AppSiteAccessTokenReset(Resource):
     @console_ns.doc("reset_app_site_access_token")
     @console_ns.doc(description="Reset access token for application site")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Access token reset successfully", app_site_model)
+    @console_ns.response(200, "Access token reset successfully", console_ns.models[AppSiteResponse.__name__])
     @console_ns.response(403, "Insufficient permissions (admin/owner required)")
     @console_ns.response(404, "App or site not found")
     @setup_required
@@ -122,7 +133,6 @@ class AppSiteAccessTokenReset(Resource):
     @is_admin_or_owner_required
     @account_initialization_required
     @get_app_model
-    @marshal_with(app_site_model)
     def post(self, app_model):
         current_user, _ = current_account_with_tenant()
         site = db.session.scalar(select(Site).where(Site.app_id == app_model.id).limit(1))
@@ -135,4 +145,4 @@ class AppSiteAccessTokenReset(Resource):
         site.updated_at = naive_utc_now()
         db.session.commit()
 
-        return site
+        return AppSiteResponse.model_validate(site, from_attributes=True).model_dump(mode="json")

api/controllers/console/app/workflow.py

@@ -1,7 +1,7 @@
 import json
 import logging
 from collections.abc import Sequence
-from typing import Any
+from typing import Any, Literal
 
 from flask import abort, request
 from flask_restx import Resource, fields, marshal_with
@@ -46,7 +46,7 @@ from libs.helper import TimestampField, uuid_value
 from libs.login import current_account_with_tenant, login_required
 from models import App
 from models.model import AppMode
-from models.workflow import Workflow
+from models.workflow import Workflow, WorkflowType
 from services.app_generate_service import AppGenerateService
 from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError, WorkflowNotFoundError
 from services.errors.llm import InvokeRateLimitError
@@ -150,6 +150,24 @@ class ConvertToWorkflowPayload(BaseModel):
     icon_background: str | None = None
 
 
+class WorkflowListQuery(BaseModel):
+    page: int = Field(default=1, ge=1, le=99999)
+    limit: int = Field(default=10, ge=1, le=100)
+    user_id: str | None = None
+    named_only: bool = False
+    keyword: str | None = Field(default=None, max_length=255)
+
+
+class WorkflowUpdatePayload(BaseModel):
+    marked_name: str | None = Field(default=None, max_length=20)
+    marked_comment: str | None = Field(default=None, max_length=100)
+
+
+class WorkflowTypeConvertQuery(BaseModel):
+    target_type: Literal["workflow", "evaluation"]
+
+
+
 class DraftWorkflowTriggerRunPayload(BaseModel):
     node_id: str
 
@@ -173,6 +191,7 @@ reg(DefaultBlockConfigQuery)
 reg(ConvertToWorkflowPayload)
 reg(WorkflowListQuery)
 reg(WorkflowUpdatePayload)
+reg(WorkflowTypeConvertQuery)
 reg(DraftWorkflowTriggerRunPayload)
 reg(DraftWorkflowTriggerRunAllPayload)
 
@@ -845,6 +864,54 @@ class PublishedWorkflowApi(Resource):
         }
 
 
+@console_ns.route("/apps/<uuid:app_id>/workflows/publish/evaluation")
+class EvaluationPublishedWorkflowApi(Resource):
+    @console_ns.doc("publish_evaluation_workflow")
+    @console_ns.doc(description="Publish draft workflow as evaluation workflow")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.expect(console_ns.models[PublishWorkflowPayload.__name__])
+    @console_ns.response(200, "Evaluation workflow published successfully")
+    @console_ns.response(400, "Invalid workflow or unsupported node type")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    @edit_permission_required
+    def post(self, app_model: App):
+        """
+        Publish draft workflow as evaluation workflow.
+
+        Evaluation workflows cannot include trigger or human-input nodes.
+        """
+        current_user, _ = current_account_with_tenant()
+        args = PublishWorkflowPayload.model_validate(console_ns.payload or {})
+
+        workflow_service = WorkflowService()
+        with Session(db.engine) as session:
+            workflow = workflow_service.publish_evaluation_workflow(
+                session=session,
+                app_model=app_model,
+                account=current_user,
+                marked_name=args.marked_name or "",
+                marked_comment=args.marked_comment or "",
+            )
+
+            # Keep workflow_id aligned with the latest published workflow.
+            app_model_in_session = session.get(App, app_model.id)
+            if app_model_in_session:
+                app_model_in_session.workflow_id = workflow.id
+                app_model_in_session.updated_by = current_user.id
+                app_model_in_session.updated_at = naive_utc_now()
+
+            workflow_created_at = TimestampField().format(workflow.created_at)
+            session.commit()
+
+        return {
+            "result": "success",
+            "created_at": workflow_created_at,
+        }
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows/default-workflow-block-configs")
 class DefaultBlockConfigsApi(Resource):
     @console_ns.doc("get_default_block_configs")
@@ -1016,6 +1083,51 @@ class DraftWorkflowRestoreApi(Resource):
         }
 
 
+@console_ns.route("/apps/<uuid:app_id>/workflows/convert-type")
+class WorkflowTypeConvertApi(Resource):
+    @console_ns.doc("convert_published_workflow_type")
+    @console_ns.doc(description="Convert current effective published workflow type in-place")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.expect(console_ns.models[WorkflowTypeConvertQuery.__name__])
+    @console_ns.response(200, "Workflow type converted successfully")
+    @console_ns.response(400, "Invalid workflow type or unsupported workflow graph")
+    @console_ns.response(404, "Workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    @edit_permission_required
+    def post(self, app_model: App):
+        current_user, _ = current_account_with_tenant()
+        args = WorkflowTypeConvertQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        target_type = WorkflowType.value_of(args.target_type)
+
+        workflow_service = WorkflowService()
+        with Session(db.engine) as session:
+            try:
+                workflow = workflow_service.convert_published_workflow_type(
+                    session=session,
+                    app_model=app_model,
+                    target_type=target_type,
+                    account=current_user,
+                )
+            except WorkflowNotFoundError as exc:
+                raise NotFound(str(exc)) from exc
+            except IsDraftWorkflowError as exc:
+                raise BadRequest(str(exc)) from exc
+            except ValueError as exc:
+                raise BadRequest(str(exc)) from exc
+
+            session.commit()
+
+        return {
+            "result": "success",
+            "workflow_id": workflow.id,
+            "type": workflow.type.value,
+            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
+        }
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows/<string:workflow_id>")
 class WorkflowByIdApi(Resource):
     @console_ns.doc("update_workflow_by_id")

api/controllers/console/app/workflow_app_log.py

@@ -87,7 +87,7 @@ class WorkflowAppLogApi(Resource):
 
         # get paginate workflow app logs
         workflow_app_service = WorkflowAppService()
-        with sessionmaker(db.engine).begin() as session:
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
             workflow_app_log_pagination = workflow_app_service.get_paginate_workflow_app_logs(
                 session=session,
                 app_model=app_model,
@@ -124,7 +124,7 @@ class WorkflowArchivedLogApi(Resource):
         args = WorkflowAppLogQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
 
         workflow_app_service = WorkflowAppService()
-        with sessionmaker(db.engine).begin() as session:
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
             workflow_app_log_pagination = workflow_app_service.get_paginate_workflow_archive_logs(
                 session=session,
                 app_model=app_model,

api/controllers/console/app/workflow_run.py

@@ -36,7 +36,7 @@ from models import Account, App, AppMode, EndUser, WorkflowArchiveLog, WorkflowR
 from models.workflow import WorkflowRun
 from repositories.factory import DifyAPIRepositoryFactory
 from services.retention.workflow_run.constants import ARCHIVE_BUNDLE_NAME
-from services.workflow_run_service import WorkflowRunService
+from services.workflow_run_service import WorkflowRunListArgs, WorkflowRunService
 
 
 def _build_backstage_input_url(form_token: str | None) -> str | None:
@@ -214,7 +214,11 @@ class AdvancedChatAppWorkflowRunListApi(Resource):
         Get advanced chat app workflow run list
         """
         args_model = WorkflowRunListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
-        args = args_model.model_dump(exclude_none=True)
+        args: WorkflowRunListArgs = {"limit": args_model.limit}
+        if args_model.last_id is not None:
+            args["last_id"] = args_model.last_id
+        if args_model.status is not None:
+            args["status"] = args_model.status
 
         # Default to DEBUGGING if not specified
         triggered_from = (
@@ -356,7 +360,11 @@ class WorkflowRunListApi(Resource):
         Get workflow run list
         """
         args_model = WorkflowRunListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
-        args = args_model.model_dump(exclude_none=True)
+        args: WorkflowRunListArgs = {"limit": args_model.limit}
+        if args_model.last_id is not None:
+            args["last_id"] = args_model.last_id
+        if args_model.status is not None:
+            args["status"] = args_model.status
 
         # Default to DEBUGGING for workflow if not specified (backward compatibility)
         triggered_from = (

api/controllers/console/app/workflow_trigger.py

@@ -64,7 +64,7 @@ class WebhookTriggerApi(Resource):
 
         node_id = args.node_id
 
-        with sessionmaker(db.engine).begin() as session:
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
             # Get webhook trigger for this app and node
             webhook_trigger = session.scalar(
                 select(WorkflowWebhookTrigger)
@@ -95,7 +95,7 @@ class AppTriggersApi(Resource):
         assert isinstance(current_user, Account)
         assert current_user.current_tenant_id is not None
 
-        with sessionmaker(db.engine).begin() as session:
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
             # Get all triggers for this app using select API
             triggers = (
                 session.execute(

api/controllers/console/auth/activate.py

@@ -1,8 +1,9 @@
 from flask import request
-from flask_restx import Resource, fields
+from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
 
 from constants.languages import supported_language
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.error import AlreadyActivateError
 from extensions.ext_database import db
@@ -11,8 +12,6 @@ from libs.helper import EmailStr, timezone
 from models import AccountStatus
 from services.account_service import RegisterService
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class ActivateCheckQuery(BaseModel):
     workspace_id: str | None = Field(default=None)
@@ -39,8 +38,16 @@ class ActivatePayload(BaseModel):
         return timezone(value)
 
 
-for model in (ActivateCheckQuery, ActivatePayload):
-    console_ns.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+class ActivationCheckResponse(BaseModel):
+    is_valid: bool = Field(description="Whether token is valid")
+    data: dict | None = Field(default=None, description="Activation data if valid")
+
+
+class ActivationResponse(BaseModel):
+    result: str = Field(description="Operation result")
+
+
+register_schema_models(console_ns, ActivateCheckQuery, ActivatePayload, ActivationCheckResponse, ActivationResponse)
 
 
 @console_ns.route("/activate/check")
@@ -51,13 +58,7 @@ class ActivateCheckApi(Resource):
     @console_ns.response(
         200,
         "Success",
-        console_ns.model(
-            "ActivationCheckResponse",
-            {
-                "is_valid": fields.Boolean(description="Whether token is valid"),
-                "data": fields.Raw(description="Activation data if valid"),
-            },
-        ),
+        console_ns.models[ActivationCheckResponse.__name__],
     )
     def get(self):
         args = ActivateCheckQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
@@ -95,12 +96,7 @@ class ActivateApi(Resource):
     @console_ns.response(
         200,
         "Account activated successfully",
-        console_ns.model(
-            "ActivationResponse",
-            {
-                "result": fields.String(description="Operation result"),
-            },
-        ),
+        console_ns.models[ActivationResponse.__name__],
     )
     @console_ns.response(400, "Already activated or invalid token")
     def post(self):

api/controllers/console/auth/login.py

@@ -1,7 +1,10 @@
+import logging
+
 import flask_login
 from flask import make_response, request
 from flask_restx import Resource
 from pydantic import BaseModel, Field
+from werkzeug.exceptions import Unauthorized
 
 import services
 from configs import dify_config
@@ -42,12 +45,13 @@ from libs.token import (
 )
 from services.account_service import AccountService, InvitationDetailDict, RegisterService, TenantService
 from services.billing_service import BillingService
-from services.entities.auth_entities import LoginPayloadBase
+from services.entities.auth_entities import LoginFailureReason, LoginPayloadBase
 from services.errors.account import AccountRegisterError
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
 from services.feature_service import FeatureService
 
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
+logger = logging.getLogger(__name__)
 
 
 class LoginPayload(LoginPayloadBase):
@@ -91,10 +95,12 @@ class LoginApi(Resource):
         normalized_email = request_email.lower()
 
         if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(normalized_email):
+            _log_console_login_failure(email=normalized_email, reason=LoginFailureReason.ACCOUNT_IN_FREEZE)
             raise AccountInFreezeError()
 
         is_login_error_rate_limit = AccountService.is_login_error_rate_limit(normalized_email)
         if is_login_error_rate_limit:
+            _log_console_login_failure(email=normalized_email, reason=LoginFailureReason.LOGIN_RATE_LIMITED)
             raise EmailPasswordLoginLimitError()
 
         invite_token = args.invite_token
@@ -110,14 +116,20 @@ class LoginApi(Resource):
                 invitee_email = data.get("email") if data else None
                 invitee_email_normalized = invitee_email.lower() if isinstance(invitee_email, str) else invitee_email
                 if invitee_email_normalized != normalized_email:
+                    _log_console_login_failure(
+                        email=normalized_email,
+                        reason=LoginFailureReason.INVALID_INVITATION_EMAIL,
+                    )
                     raise InvalidEmailError()
             account = _authenticate_account_with_case_fallback(
                 request_email, normalized_email, args.password, invite_token
             )
         except services.errors.account.AccountLoginError:
+            _log_console_login_failure(email=normalized_email, reason=LoginFailureReason.ACCOUNT_BANNED)
             raise AccountBannedError()
         except services.errors.account.AccountPasswordError as exc:
             AccountService.add_login_error_rate_limit(normalized_email)
+            _log_console_login_failure(email=normalized_email, reason=LoginFailureReason.INVALID_CREDENTIALS)
             raise AuthenticationFailedError() from exc
         # SELF_HOSTED only have one workspace
         tenants = TenantService.get_join_tenants(account)
@@ -240,20 +252,27 @@ class EmailCodeLoginApi(Resource):
 
         token_data = AccountService.get_email_code_login_data(args.token)
         if token_data is None:
+            _log_console_login_failure(email=user_email, reason=LoginFailureReason.INVALID_EMAIL_CODE_TOKEN)
             raise InvalidTokenError()
 
         token_email = token_data.get("email")
         normalized_token_email = token_email.lower() if isinstance(token_email, str) else token_email
         if normalized_token_email != user_email:
+            _log_console_login_failure(email=user_email, reason=LoginFailureReason.EMAIL_CODE_EMAIL_MISMATCH)
             raise InvalidEmailError()
 
         if token_data["code"] != args.code:
+            _log_console_login_failure(email=user_email, reason=LoginFailureReason.INVALID_EMAIL_CODE)
             raise EmailCodeError()
 
         AccountService.revoke_email_code_login_token(args.token)
         try:
             account = _get_account_with_case_fallback(original_email)
+        except Unauthorized as exc:
+            _log_console_login_failure(email=user_email, reason=LoginFailureReason.ACCOUNT_BANNED)
+            raise AccountBannedError() from exc
         except AccountRegisterError:
+            _log_console_login_failure(email=user_email, reason=LoginFailureReason.ACCOUNT_IN_FREEZE)
             raise AccountInFreezeError()
         if account:
             tenants = TenantService.get_join_tenants(account)
@@ -279,6 +298,7 @@ class EmailCodeLoginApi(Resource):
             except WorkSpaceNotAllowedCreateError:
                 raise NotAllowedCreateWorkspace()
             except AccountRegisterError:
+                _log_console_login_failure(email=user_email, reason=LoginFailureReason.ACCOUNT_IN_FREEZE)
                 raise AccountInFreezeError()
             except WorkspacesLimitExceededError:
                 raise WorkspacesLimitExceeded()
@@ -336,3 +356,12 @@ def _authenticate_account_with_case_fallback(
         if original_email == normalized_email:
             raise
         return AccountService.authenticate(normalized_email, password, invite_token)
+
+
+def _log_console_login_failure(*, email: str, reason: LoginFailureReason) -> None:
+    logger.warning(
+        "Console login failed: email=%s reason=%s ip_address=%s",
+        email,
+        reason,
+        extract_remote_ip(request),
+    )

api/controllers/console/billing/billing.py

@@ -1,4 +1,6 @@
 import base64
+import json
+from datetime import UTC, datetime, timedelta
 from typing import Literal
 
 from flask import request
@@ -10,6 +12,7 @@ from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, only_edition_cloud, setup_required
 from enums.cloud_plan import CloudPlan
+from extensions.ext_redis import redis_client
 from libs.login import current_account_with_tenant, login_required
 from services.billing_service import BillingService
 
@@ -77,3 +80,39 @@ class PartnerTenants(Resource):
             raise BadRequest("Invalid partner information")
 
         return BillingService.sync_partner_tenants_bindings(current_user.id, decoded_partner_key, click_id)
+
+
+_DEBUG_KEY = "billing:debug"
+_DEBUG_TTL = timedelta(days=7)
+
+
+class DebugDataPayload(BaseModel):
+    type: str = Field(..., min_length=1, description="Data type key")
+    data: str = Field(..., min_length=1, description="Data value to append")
+
+
+@console_ns.route("/billing/debug/data")
+class DebugData(Resource):
+    def post(self):
+        body = DebugDataPayload.model_validate(request.get_json(force=True))
+        item = json.dumps({
+            "type": body.type,
+            "data": body.data,
+            "createTime": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
+        })
+        redis_client.lpush(_DEBUG_KEY, item)
+        redis_client.expire(_DEBUG_KEY, _DEBUG_TTL)
+        return {"result": "ok"}, 201
+
+    def get(self):
+        recent = request.args.get("recent", 10, type=int)
+        items = redis_client.lrange(_DEBUG_KEY, 0, recent - 1)
+        return {
+            "data": [
+                json.loads(item.decode("utf-8") if isinstance(item, bytes) else item) for item in items
+            ]
+        }
+
+    def delete(self):
+        redis_client.delete(_DEBUG_KEY)
+        return {"result": "ok"}

api/controllers/console/datasets/datasets.py

@@ -1,20 +1,20 @@
+import json
 from typing import Any, cast
+from urllib.parse import quote
 
-from flask import request
+from flask import Response, request
 from flask_restx import Resource, fields, marshal, marshal_with
 from graphon.model_runtime.entities.model_entities import ModelType
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import func, select
-from werkzeug.exceptions import Forbidden, NotFound
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import BadRequest, Forbidden, NotFound
 
 import services
 from configs import dify_config
 from controllers.common.schema import get_or_create_model, register_schema_models
 from controllers.console import console_ns
-from controllers.console.apikey import (
-    api_key_item_model,
-    api_key_list_model,
-)
+from controllers.console.apikey import ApiKeyItem, ApiKeyList
 from controllers.console.app.error import ProviderNotInitializeError
 from controllers.console.datasets.error import DatasetInUseError, DatasetNameDuplicateError, IndexingEstimateError
 from controllers.console.wraps import (
@@ -25,6 +25,7 @@ from controllers.console.wraps import (
     setup_required,
 )
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
+from core.evaluation.entities.evaluation_entity import EvaluationCategory, EvaluationConfigData, EvaluationRunRequest
 from core.indexing_runner import IndexingRunner
 from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
 from core.rag.datasource.vdb.vector_type import VectorType
@@ -33,6 +34,7 @@ from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from extensions.ext_database import db
+from extensions.ext_storage import storage
 from fields.app_fields import app_detail_kernel_fields, related_app_list
 from fields.dataset_fields import (
     content_fields,
@@ -53,12 +55,19 @@ from fields.dataset_fields import (
 )
 from fields.document_fields import document_status_fields
 from libs.login import current_account_with_tenant, login_required
-from models import ApiToken, Dataset, Document, DocumentSegment, UploadFile
+from models import ApiToken, Dataset, Document, DocumentSegment, EvaluationRun, EvaluationTargetType, UploadFile
 from models.dataset import DatasetPermission, DatasetPermissionEnum
 from models.enums import ApiTokenType, SegmentStatus
 from models.provider_ids import ModelProviderID
 from services.api_token_service import ApiTokenCache
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
+from services.errors.evaluation import (
+    EvaluationDatasetInvalidError,
+    EvaluationFrameworkNotConfiguredError,
+    EvaluationMaxConcurrentRunsError,
+    EvaluationNotFoundError,
+)
+from services.evaluation_service import EvaluationService
 
 # Register models for flask_restx to avoid dict type issues in Swagger
 dataset_base_model = get_or_create_model("DatasetBase", dataset_fields)
@@ -785,23 +794,23 @@ class DatasetApiKeyApi(Resource):
 
     @console_ns.doc("get_dataset_api_keys")
     @console_ns.doc(description="Get dataset API keys")
-    @console_ns.response(200, "API keys retrieved successfully", api_key_list_model)
+    @console_ns.response(200, "API keys retrieved successfully", console_ns.models[ApiKeyList.__name__])
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(api_key_list_model)
     def get(self):
         _, current_tenant_id = current_account_with_tenant()
         keys = db.session.scalars(
             select(ApiToken).where(ApiToken.type == self.resource_type, ApiToken.tenant_id == current_tenant_id)
         ).all()
-        return {"items": keys}
+        return ApiKeyList.model_validate({"data": keys}, from_attributes=True).model_dump(mode="json")
 
+    @console_ns.response(200, "API key created successfully", console_ns.models[ApiKeyItem.__name__])
+    @console_ns.response(400, "Maximum keys exceeded")
     @setup_required
     @login_required
     @is_admin_or_owner_required
     @account_initialization_required
-    @marshal_with(api_key_item_model)
     def post(self):
         _, current_tenant_id = current_account_with_tenant()
 
@@ -828,7 +837,7 @@ class DatasetApiKeyApi(Resource):
         api_token.type = self.resource_type
         db.session.add(api_token)
         db.session.commit()
-        return api_token, 200
+        return ApiKeyItem.model_validate(api_token, from_attributes=True).model_dump(mode="json"), 200
 
 
 @console_ns.route("/datasets/api-keys/<uuid:api_key_id>")
@@ -986,3 +995,432 @@ class DatasetAutoDisableLogApi(Resource):
         if dataset is None:
             raise NotFound("Dataset not found.")
         return DatasetService.get_dataset_auto_disable_logs(dataset_id_str), 200
+
+
+# ---- Knowledge Base Retrieval Evaluation ----
+
+
+def _serialize_dataset_evaluation_run(run: EvaluationRun) -> dict[str, Any]:
+    return {
+        "id": run.id,
+        "tenant_id": run.tenant_id,
+        "target_type": run.target_type,
+        "target_id": run.target_id,
+        "evaluation_config_id": run.evaluation_config_id,
+        "status": run.status,
+        "dataset_file_id": run.dataset_file_id,
+        "result_file_id": run.result_file_id,
+        "total_items": run.total_items,
+        "completed_items": run.completed_items,
+        "failed_items": run.failed_items,
+        "progress": run.progress,
+        "metrics_summary": json.loads(run.metrics_summary) if run.metrics_summary else {},
+        "error": run.error,
+        "created_by": run.created_by,
+        "started_at": int(run.started_at.timestamp()) if run.started_at else None,
+        "completed_at": int(run.completed_at.timestamp()) if run.completed_at else None,
+        "created_at": int(run.created_at.timestamp()) if run.created_at else None,
+    }
+
+
+def _serialize_dataset_evaluation_run_item(item: Any) -> dict[str, Any]:
+    return {
+        "id": item.id,
+        "item_index": item.item_index,
+        "inputs": item.inputs_dict,
+        "expected_output": item.expected_output,
+        "actual_output": item.actual_output,
+        "metrics": item.metrics_list,
+        "judgment": item.judgment_dict,
+        "metadata": item.metadata_dict,
+        "error": item.error,
+        "overall_score": item.overall_score,
+    }
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation/template/download")
+class DatasetEvaluationTemplateDownloadApi(Resource):
+    @console_ns.doc("download_dataset_evaluation_template")
+    @console_ns.response(200, "Template file streamed as XLSX attachment")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, dataset_id):
+        """Download evaluation dataset template for knowledge base retrieval."""
+        current_user, _ = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        xlsx_content, filename = EvaluationService.generate_retrieval_dataset_template()
+        encoded_filename = quote(filename)
+        response = Response(
+            xlsx_content,
+            mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        )
+        response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
+        response.headers["Content-Length"] = str(len(xlsx_content))
+        return response
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation")
+class DatasetEvaluationDetailApi(Resource):
+    @console_ns.doc("get_dataset_evaluation_config")
+    @console_ns.response(200, "Evaluation configuration retrieved")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id):
+        """Get evaluation configuration for the knowledge base."""
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            config = EvaluationService.get_evaluation_config(
+                session, current_tenant_id, "dataset", dataset_id_str
+            )
+
+        if config is None:
+            return {
+                "evaluation_model": None,
+                "evaluation_model_provider": None,
+                "default_metrics": None,
+                "customized_metrics": None,
+                "judgment_config": None,
+            }
+
+        return {
+            "evaluation_model": config.evaluation_model,
+            "evaluation_model_provider": config.evaluation_model_provider,
+            "default_metrics": config.default_metrics_list,
+            "customized_metrics": config.customized_metrics_dict,
+            "judgment_config": config.judgment_config_dict,
+        }
+
+    @console_ns.doc("save_dataset_evaluation_config")
+    @console_ns.response(200, "Evaluation configuration saved")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def put(self, dataset_id):
+        """Save evaluation configuration for the knowledge base."""
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        body = request.get_json(force=True)
+        try:
+            config_data = EvaluationConfigData.model_validate(body)
+        except Exception as e:
+            raise BadRequest(f"Invalid request body: {e}")
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            config = EvaluationService.save_evaluation_config(
+                session=session,
+                tenant_id=current_tenant_id,
+                target_type="dataset",
+                target_id=dataset_id_str,
+                account_id=str(current_user.id),
+                data=config_data,
+            )
+
+        return {
+            "evaluation_model": config.evaluation_model,
+            "evaluation_model_provider": config.evaluation_model_provider,
+            "default_metrics": config.default_metrics_list,
+            "customized_metrics": config.customized_metrics_dict,
+            "judgment_config": config.judgment_config_dict,
+        }
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation/run")
+class DatasetEvaluationRunApi(Resource):
+    @console_ns.doc("start_dataset_evaluation_run")
+    @console_ns.response(200, "Evaluation run started")
+    @console_ns.response(400, "Invalid request")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, dataset_id):
+        """Start an evaluation run for the knowledge base retrieval."""
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        body = request.get_json(force=True)
+        if not body:
+            raise BadRequest("Request body is required.")
+
+        try:
+            run_request = EvaluationRunRequest.model_validate(body)
+        except Exception as e:
+            raise BadRequest(f"Invalid request body: {e}")
+
+        upload_file = (
+            db.session.query(UploadFile).filter_by(id=run_request.file_id, tenant_id=current_tenant_id).first()
+        )
+        if not upload_file:
+            raise NotFound("Dataset file not found.")
+
+        try:
+            dataset_content = storage.load_once(upload_file.key)
+        except Exception:
+            raise BadRequest("Failed to read dataset file.")
+
+        if not dataset_content:
+            raise BadRequest("Dataset file is empty.")
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                evaluation_run = EvaluationService.start_evaluation_run(
+                    session=session,
+                    tenant_id=current_tenant_id,
+                    target_type=EvaluationTargetType.KNOWLEDGE_BASE,
+                    target_id=dataset_id_str,
+                    account_id=str(current_user.id),
+                    dataset_file_content=dataset_content,
+                    run_request=run_request,
+                )
+                return _serialize_dataset_evaluation_run(evaluation_run), 200
+        except EvaluationFrameworkNotConfiguredError as e:
+            return {"message": str(e.description)}, 400
+        except EvaluationNotFoundError as e:
+            return {"message": str(e.description)}, 404
+        except EvaluationMaxConcurrentRunsError as e:
+            return {"message": str(e.description)}, 429
+        except EvaluationDatasetInvalidError as e:
+            return {"message": str(e.description)}, 400
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation/logs")
+class DatasetEvaluationLogsApi(Resource):
+    @console_ns.doc("get_dataset_evaluation_logs")
+    @console_ns.response(200, "Evaluation logs retrieved")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id):
+        """Get evaluation run history for the knowledge base."""
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        page = request.args.get("page", 1, type=int)
+        page_size = request.args.get("page_size", 20, type=int)
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            runs, total = EvaluationService.get_evaluation_runs(
+                session=session,
+                tenant_id=current_tenant_id,
+                target_type="dataset",
+                target_id=dataset_id_str,
+                page=page,
+                page_size=page_size,
+            )
+
+        return {
+            "data": [_serialize_dataset_evaluation_run(run) for run in runs],
+            "total": total,
+            "page": page,
+            "page_size": page_size,
+        }
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation/runs/<uuid:run_id>")
+class DatasetEvaluationRunDetailApi(Resource):
+    @console_ns.doc("get_dataset_evaluation_run_detail")
+    @console_ns.response(200, "Evaluation run detail retrieved")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset or run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id, run_id):
+        """Get evaluation run detail including per-item results."""
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        run_id_str = str(run_id)
+        page = request.args.get("page", 1, type=int)
+        page_size = request.args.get("page_size", 50, type=int)
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                run = EvaluationService.get_evaluation_run_detail(
+                    session=session,
+                    tenant_id=current_tenant_id,
+                    run_id=run_id_str,
+                )
+                items, total_items = EvaluationService.get_evaluation_run_items(
+                    session=session,
+                    run_id=run_id_str,
+                    page=page,
+                    page_size=page_size,
+                )
+                return {
+                    "run": _serialize_dataset_evaluation_run(run),
+                    "items": {
+                        "data": [_serialize_dataset_evaluation_run_item(item) for item in items],
+                        "total": total_items,
+                        "page": page,
+                        "page_size": page_size,
+                    },
+                }
+        except EvaluationNotFoundError as e:
+            return {"message": str(e.description)}, 404
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation/runs/<uuid:run_id>/cancel")
+class DatasetEvaluationRunCancelApi(Resource):
+    @console_ns.doc("cancel_dataset_evaluation_run")
+    @console_ns.response(200, "Evaluation run cancelled")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset or run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, dataset_id, run_id):
+        """Cancel a running knowledge base evaluation."""
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        run_id_str = str(run_id)
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                run = EvaluationService.cancel_evaluation_run(
+                    session=session,
+                    tenant_id=current_tenant_id,
+                    run_id=run_id_str,
+                )
+                return _serialize_dataset_evaluation_run(run)
+        except EvaluationNotFoundError as e:
+            return {"message": str(e.description)}, 404
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation/metrics")
+class DatasetEvaluationMetricsApi(Resource):
+    @console_ns.doc("get_dataset_evaluation_metrics")
+    @console_ns.response(200, "Available retrieval metrics retrieved")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id):
+        """Get available evaluation metrics for knowledge base retrieval."""
+        current_user, _ = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        return {
+            "metrics": EvaluationService.get_supported_metrics(EvaluationCategory.KNOWLEDGE_BASE)
+        }
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation/files/<uuid:file_id>")
+class DatasetEvaluationFileDownloadApi(Resource):
+    @console_ns.doc("download_dataset_evaluation_file")
+    @console_ns.response(200, "File download URL generated")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset or file not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id, file_id):
+        """Download evaluation test file or result file for the knowledge base."""
+        from core.workflow.file import helpers as file_helpers
+
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        file_id_str = str(file_id)
+        with Session(db.engine, expire_on_commit=False) as session:
+            stmt = select(UploadFile).where(
+                UploadFile.id == file_id_str,
+                UploadFile.tenant_id == current_tenant_id,
+            )
+            upload_file = session.execute(stmt).scalar_one_or_none()
+
+        if not upload_file:
+            raise NotFound("File not found.")
+
+        download_url = file_helpers.get_signed_file_url(upload_file_id=upload_file.id, as_attachment=True)
+
+        return {
+            "id": upload_file.id,
+            "name": upload_file.name,
+            "size": upload_file.size,
+            "extension": upload_file.extension,
+            "mime_type": upload_file.mime_type,
+            "created_at": int(upload_file.created_at.timestamp()) if upload_file.created_at else None,
+            "download_url": download_url,
+        }

api/controllers/console/datasets/datasets_document.py

@@ -4,7 +4,6 @@ from argparse import ArgumentTypeError
 from collections.abc import Sequence
 from contextlib import ExitStack
 from typing import Any, Literal, cast
-from uuid import UUID
 
 import sqlalchemy as sa
 from flask import request, send_file
@@ -16,6 +15,7 @@ from sqlalchemy import asc, desc, func, select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
+from controllers.common.controller_schemas import DocumentBatchDownloadZipPayload
 from controllers.common.schema import get_or_create_model, register_schema_models
 from controllers.console import console_ns
 from core.errors.error import (
@@ -71,9 +71,6 @@ from ..wraps import (
 
 logger = logging.getLogger(__name__)
 
-# NOTE: Keep constants near the top of the module for discoverability.
-DOCUMENT_BATCH_DOWNLOAD_ZIP_MAX_DOCS = 100
-
 
 # Register models for flask_restx to avoid dict type issues in Swagger
 dataset_model = get_or_create_model("Dataset", dataset_fields)
@@ -110,12 +107,6 @@ class GenerateSummaryPayload(BaseModel):
     document_list: list[str]
 
 
-class DocumentBatchDownloadZipPayload(BaseModel):
-    """Request payload for bulk downloading documents as a zip archive."""
-
-    document_ids: list[UUID] = Field(..., min_length=1, max_length=DOCUMENT_BATCH_DOWNLOAD_ZIP_MAX_DOCS)
-
-
 class DocumentDatasetListParam(BaseModel):
     page: int = Field(1, title="Page", description="Page number.")
     limit: int = Field(20, title="Limit", description="Page size.")

api/controllers/console/datasets/datasets_segments.py

@@ -10,6 +10,7 @@ from werkzeug.exceptions import Forbidden, NotFound
 
 import services
 from configs import dify_config
+from controllers.common.controller_schemas import ChildChunkCreatePayload, ChildChunkUpdatePayload
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import ProviderNotInitializeError
@@ -82,14 +83,6 @@ class BatchImportPayload(BaseModel):
     upload_file_id: str
 
 
-class ChildChunkCreatePayload(BaseModel):
-    content: str
-
-
-class ChildChunkUpdatePayload(BaseModel):
-    content: str
-
-
 class ChildChunkBatchUpdatePayload(BaseModel):
     chunks: list[ChildChunkUpdateArgs]
 

api/controllers/console/datasets/metadata.py

@@ -1,9 +1,9 @@
 from typing import Literal
 
 from flask_restx import Resource, marshal_with
-from pydantic import BaseModel
 from werkzeug.exceptions import NotFound
 
+from controllers.common.controller_schemas import MetadataUpdatePayload
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, enterprise_license_required, setup_required
@@ -18,11 +18,6 @@ from services.entities.knowledge_entities.knowledge_entities import (
 )
 from services.metadata_service import MetadataService
 
-
-class MetadataUpdatePayload(BaseModel):
-    name: str
-
-
 register_schema_models(
     console_ns, MetadataArgs, MetadataOperationData, MetadataUpdatePayload, DocumentMetadataOperation, MetadataDetail
 )

api/controllers/console/evaluation/__init__.py

@@ -0,0 +1 @@
+# Evaluation controller module

api/controllers/console/evaluation/evaluation.py

@@ -0,0 +1,869 @@
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable
+from functools import wraps
+from typing import TYPE_CHECKING, ParamSpec, TypeVar, Union
+from urllib.parse import quote
+
+from flask import Response, request
+from flask_restx import Resource, fields, marshal
+from pydantic import BaseModel
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import BadRequest, Forbidden, NotFound
+
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
+from controllers.console.app.workflow import WorkflowListQuery
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+)
+from core.evaluation.entities.evaluation_entity import EvaluationCategory, EvaluationConfigData, EvaluationRunRequest
+from extensions.ext_database import db
+from extensions.ext_storage import storage
+from fields.member_fields import simple_account_fields
+from graphon.file import helpers as file_helpers
+from libs.helper import TimestampField
+from libs.login import current_account_with_tenant, login_required
+from models import App, Dataset
+from models.model import UploadFile
+from models.snippet import CustomizedSnippet
+from services.errors.evaluation import (
+    EvaluationDatasetInvalidError,
+    EvaluationFrameworkNotConfiguredError,
+    EvaluationMaxConcurrentRunsError,
+    EvaluationNotFoundError,
+)
+from services.evaluation_service import EvaluationService
+from services.workflow_service import WorkflowService
+
+if TYPE_CHECKING:
+    from models.evaluation import EvaluationRun, EvaluationRunItem
+
+logger = logging.getLogger(__name__)
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+# Valid evaluation target types
+EVALUATE_TARGET_TYPES = {"app", "snippets"}
+
+
+class VersionQuery(BaseModel):
+    """Query parameters for version endpoint."""
+
+    version: str
+
+
+register_schema_models(
+    console_ns,
+    VersionQuery,
+)
+
+
+# Response field definitions
+file_info_fields = {
+    "id": fields.String,
+    "name": fields.String,
+}
+
+evaluation_log_fields = {
+    "created_at": TimestampField,
+    "created_by": fields.String,
+    "test_file": fields.Nested(
+        console_ns.model(
+            "EvaluationTestFile",
+            file_info_fields,
+        )
+    ),
+    "result_file": fields.Nested(
+        console_ns.model(
+            "EvaluationResultFile",
+            file_info_fields,
+        ),
+        allow_null=True,
+    ),
+    "version": fields.String,
+}
+
+evaluation_log_list_model = console_ns.model(
+    "EvaluationLogList",
+    {
+        "data": fields.List(fields.Nested(console_ns.model("EvaluationLog", evaluation_log_fields))),
+    },
+)
+
+evaluation_default_metric_node_info_fields = {
+    "node_id": fields.String,
+    "type": fields.String,
+    "title": fields.String,
+}
+evaluation_default_metric_item_fields = {
+    "metric": fields.String,
+    "value_type": fields.String,
+    "node_info_list": fields.List(
+        fields.Nested(
+            console_ns.model("EvaluationDefaultMetricNodeInfo", evaluation_default_metric_node_info_fields),
+        ),
+    ),
+}
+
+customized_metrics_fields = {
+    "evaluation_workflow_id": fields.String,
+    "input_fields": fields.Raw,
+    "output_fields": fields.Raw,
+}
+
+judgment_condition_fields = {
+    "variable_selector": fields.List(fields.String),
+    "comparison_operator": fields.String,
+    "value": fields.String,
+}
+
+judgment_config_fields = {
+    "logical_operator": fields.String,
+    "conditions": fields.List(fields.Nested(console_ns.model("JudgmentCondition", judgment_condition_fields))),
+}
+
+evaluation_detail_fields = {
+    "evaluation_model": fields.String,
+    "evaluation_model_provider": fields.String,
+    "default_metrics": fields.List(
+        fields.Nested(console_ns.model("EvaluationDefaultMetricItem_Detail", evaluation_default_metric_item_fields)),
+        allow_null=True,
+    ),
+    "customized_metrics": fields.Nested(
+        console_ns.model("EvaluationCustomizedMetrics", customized_metrics_fields),
+        allow_null=True,
+    ),
+    "judgment_config": fields.Nested(
+        console_ns.model("EvaluationJudgmentConfig", judgment_config_fields),
+        allow_null=True,
+    ),
+}
+
+evaluation_detail_model = console_ns.model("EvaluationDetail", evaluation_detail_fields)
+
+available_evaluation_workflow_list_fields = {
+    "id": fields.String,
+    "app_id": fields.String,
+    "app_name": fields.String,
+    "type": fields.String,
+    "version": fields.String,
+    "marked_name": fields.String,
+    "marked_comment": fields.String,
+    "hash": fields.String,
+    "created_by": fields.Nested(simple_account_fields),
+    "created_at": TimestampField,
+    "updated_by": fields.Nested(simple_account_fields, allow_null=True),
+    "updated_at": TimestampField,
+}
+
+available_evaluation_workflow_pagination_fields = {
+    "items": fields.List(fields.Nested(available_evaluation_workflow_list_fields)),
+    "page": fields.Integer,
+    "limit": fields.Integer,
+    "has_more": fields.Boolean,
+}
+
+available_evaluation_workflow_pagination_model = console_ns.model(
+    "AvailableEvaluationWorkflowPagination",
+    available_evaluation_workflow_pagination_fields,
+)
+
+evaluation_default_metrics_response_model = console_ns.model(
+    "EvaluationDefaultMetricsResponse",
+    {
+        "default_metrics": fields.List(
+            fields.Nested(console_ns.model("EvaluationDefaultMetricItem", evaluation_default_metric_item_fields)),
+        ),
+    },
+)
+
+
+def get_evaluation_target(view_func: Callable[P, R]):
+    """
+    Decorator to resolve polymorphic evaluation target (app or snippet).
+
+    Validates the target_type parameter and fetches the corresponding
+    model (App or CustomizedSnippet) with tenant isolation.
+    """
+
+    @wraps(view_func)
+    def decorated_view(*args: P.args, **kwargs: P.kwargs):
+        target_type = kwargs.get("evaluate_target_type")
+        target_id = kwargs.get("evaluate_target_id")
+
+        if target_type not in EVALUATE_TARGET_TYPES:
+            raise NotFound(f"Invalid evaluation target type: {target_type}")
+
+        _, current_tenant_id = current_account_with_tenant()
+
+        target_id = str(target_id)
+
+        # Remove path parameters
+        del kwargs["evaluate_target_type"]
+        del kwargs["evaluate_target_id"]
+
+        target: Union[App, CustomizedSnippet, Dataset] | None = None
+
+        if target_type == "app":
+            target = db.session.query(App).where(App.id == target_id, App.tenant_id == current_tenant_id).first()
+        elif target_type == "snippets":
+            target = (
+                db.session.query(CustomizedSnippet)
+                .where(CustomizedSnippet.id == target_id, CustomizedSnippet.tenant_id == current_tenant_id)
+                .first()
+            )
+        elif target_type == "knowledge":
+            target = (db.session.query(Dataset)
+                      .where(Dataset.id == target_id, Dataset.tenant_id == current_tenant_id)
+                      .first())
+
+        if not target:
+            raise NotFound(f"{str(target_type)} not found")
+
+        kwargs["target"] = target
+        kwargs["target_type"] = target_type
+
+        return view_func(*args, **kwargs)
+
+    return decorated_view
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/dataset-template/download")
+class EvaluationDatasetTemplateDownloadApi(Resource):
+    @console_ns.doc("download_evaluation_dataset_template")
+    @console_ns.response(200, "Template file streamed as XLSX attachment")
+    @console_ns.response(400, "Invalid target type or excluded app mode")
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    @edit_permission_required
+    def post(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Download evaluation dataset template.
+
+        Generates an XLSX template based on the target's input parameters
+        and streams it directly as a file attachment.
+        """
+        try:
+            xlsx_content, filename = EvaluationService.generate_dataset_template(
+                target=target,
+                target_type=target_type,
+            )
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        encoded_filename = quote(filename)
+        response = Response(
+            xlsx_content,
+            mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        )
+        response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
+        response.headers["Content-Length"] = str(len(xlsx_content))
+        return response
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation")
+class EvaluationDetailApi(Resource):
+    @console_ns.doc("get_evaluation_detail")
+    @console_ns.response(200, "Evaluation details retrieved successfully", evaluation_detail_model)
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Get evaluation configuration for the target.
+
+        Returns evaluation configuration including model settings,
+        metrics config, and judgement conditions.
+        """
+        _, current_tenant_id = current_account_with_tenant()
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            config = EvaluationService.get_evaluation_config(session, current_tenant_id, target_type, str(target.id))
+
+        if config is None:
+            return {
+                "evaluation_model": None,
+                "evaluation_model_provider": None,
+                "default_metrics": None,
+                "customized_metrics": None,
+                "judgment_config": None,
+            }
+
+        return {
+            "evaluation_model": config.evaluation_model,
+            "evaluation_model_provider": config.evaluation_model_provider,
+            "default_metrics": config.default_metrics_list,
+            "customized_metrics": config.customized_metrics_dict,
+            "judgment_config": config.judgment_config_dict,
+        }
+
+    @console_ns.doc("save_evaluation_detail")
+    @console_ns.response(200, "Evaluation configuration saved successfully")
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    @edit_permission_required
+    def put(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Save evaluation configuration for the target.
+        """
+        current_account, current_tenant_id = current_account_with_tenant()
+        body = request.get_json(force=True)
+
+        try:
+            config_data = EvaluationConfigData.model_validate(body)
+        except Exception as e:
+            raise BadRequest(f"Invalid request body: {e}")
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            config = EvaluationService.save_evaluation_config(
+                session=session,
+                tenant_id=current_tenant_id,
+                target_type=target_type,
+                target_id=str(target.id),
+                account_id=str(current_account.id),
+                data=config_data,
+            )
+
+        return {
+            "evaluation_model": config.evaluation_model,
+            "evaluation_model_provider": config.evaluation_model_provider,
+            "default_metrics": config.default_metrics_list,
+            "customized_metrics": config.customized_metrics_dict,
+            "judgment_config": config.judgment_config_dict,
+        }
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/logs")
+class EvaluationLogsApi(Resource):
+    @console_ns.doc("get_evaluation_logs")
+    @console_ns.response(200, "Evaluation logs retrieved successfully")
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Get evaluation run history for the target.
+
+        Returns a paginated list of evaluation runs.
+        """
+        _, current_tenant_id = current_account_with_tenant()
+        page = request.args.get("page", 1, type=int)
+        page_size = request.args.get("page_size", 20, type=int)
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            runs, total = EvaluationService.get_evaluation_runs(
+                session=session,
+                tenant_id=current_tenant_id,
+                target_type=target_type,
+                target_id=str(target.id),
+                page=page,
+                page_size=page_size,
+            )
+
+        return {
+            "data": [_serialize_evaluation_run(run) for run in runs],
+            "total": total,
+            "page": page,
+            "page_size": page_size,
+        }
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/run")
+class EvaluationRunApi(Resource):
+    @console_ns.doc("start_evaluation_run")
+    @console_ns.response(200, "Evaluation run started")
+    @console_ns.response(400, "Invalid request")
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    @edit_permission_required
+    def post(self, target: Union[App, CustomizedSnippet, Dataset], target_type: str):
+        """
+        Start an evaluation run.
+
+        Expects JSON body with:
+        - file_id: uploaded dataset file ID
+        - evaluation_model: evaluation model name
+        - evaluation_model_provider: evaluation model provider
+        - default_metrics: list of default metric objects
+        - customized_metrics: customized metrics object (optional)
+        - judgment_config: judgment conditions config (optional)
+        """
+        current_account, current_tenant_id = current_account_with_tenant()
+
+        body = request.get_json(force=True)
+        if not body:
+            raise BadRequest("Request body is required.")
+
+        # Validate and parse request body
+        try:
+            run_request = EvaluationRunRequest.model_validate(body)
+        except Exception as e:
+            raise BadRequest(f"Invalid request body: {e}")
+
+        # Load dataset file
+        upload_file = (
+            db.session.query(UploadFile).filter_by(id=run_request.file_id, tenant_id=current_tenant_id).first()
+        )
+        if not upload_file:
+            raise NotFound("Dataset file not found.")
+
+        try:
+            dataset_content = storage.load_once(upload_file.key)
+        except Exception:
+            raise BadRequest("Failed to read dataset file.")
+
+        if not dataset_content:
+            raise BadRequest("Dataset file is empty.")
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                evaluation_run = EvaluationService.start_evaluation_run(
+                    session=session,
+                    tenant_id=current_tenant_id,
+                    target_type=target_type,
+                    target_id=str(target.id),
+                    account_id=str(current_account.id),
+                    dataset_file_content=dataset_content,
+                    run_request=run_request,
+                )
+                return _serialize_evaluation_run(evaluation_run), 200
+        except EvaluationFrameworkNotConfiguredError as e:
+            return {"message": str(e.description)}, 400
+        except EvaluationNotFoundError as e:
+            return {"message": str(e.description)}, 404
+        except EvaluationMaxConcurrentRunsError as e:
+            return {"message": str(e.description)}, 429
+        except EvaluationDatasetInvalidError as e:
+            return {"message": str(e.description)}, 400
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/runs/<uuid:run_id>")
+class EvaluationRunDetailApi(Resource):
+    @console_ns.doc("get_evaluation_run_detail")
+    @console_ns.response(200, "Evaluation run detail retrieved")
+    @console_ns.response(404, "Run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str, run_id: str):
+        """
+        Get evaluation run detail including items.
+        """
+        _, current_tenant_id = current_account_with_tenant()
+        run_id = str(run_id)
+        page = request.args.get("page", 1, type=int)
+        page_size = request.args.get("page_size", 50, type=int)
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                run = EvaluationService.get_evaluation_run_detail(
+                    session=session,
+                    tenant_id=current_tenant_id,
+                    run_id=run_id,
+                )
+                items, total_items = EvaluationService.get_evaluation_run_items(
+                    session=session,
+                    run_id=run_id,
+                    page=page,
+                    page_size=page_size,
+                )
+
+                return {
+                    "run": _serialize_evaluation_run(run),
+                    "items": {
+                        "data": [_serialize_evaluation_run_item(item) for item in items],
+                        "total": total_items,
+                        "page": page,
+                        "page_size": page_size,
+                    },
+                }
+        except EvaluationNotFoundError as e:
+            return {"message": str(e.description)}, 404
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/runs/<uuid:run_id>/cancel")
+class EvaluationRunCancelApi(Resource):
+    @console_ns.doc("cancel_evaluation_run")
+    @console_ns.response(200, "Evaluation run cancelled")
+    @console_ns.response(404, "Run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    @edit_permission_required
+    def post(self, target: Union[App, CustomizedSnippet], target_type: str, run_id: str):
+        """Cancel a running evaluation."""
+        _, current_tenant_id = current_account_with_tenant()
+        run_id = str(run_id)
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                run = EvaluationService.cancel_evaluation_run(
+                    session=session,
+                    tenant_id=current_tenant_id,
+                    run_id=run_id,
+                )
+                return _serialize_evaluation_run(run)
+        except EvaluationNotFoundError as e:
+            return {"message": str(e.description)}, 404
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/metrics")
+class EvaluationMetricsApi(Resource):
+    @console_ns.doc("get_evaluation_metrics")
+    @console_ns.response(200, "Available metrics retrieved")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Get available evaluation metrics for the current framework.
+        """
+        result = {}
+        for category in EvaluationCategory:
+            result[category.value] = EvaluationService.get_supported_metrics(category)
+        return {"metrics": result}
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/default-metrics")
+class EvaluationDefaultMetricsApi(Resource):
+    @console_ns.doc(
+        "get_evaluation_default_metrics_with_nodes",
+        description=(
+            "List default metrics supported by the current evaluation framework with matching nodes "
+            "from the target's published workflow only (draft is ignored)."
+        ),
+    )
+    @console_ns.response(
+        200,
+        "Default metrics and node candidates for the published workflow",
+        evaluation_default_metrics_response_model,
+    )
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        default_metrics = EvaluationService.get_default_metrics_with_nodes_for_published_target(
+            target=target,
+            target_type=target_type,
+        )
+        return {"default_metrics": [m.model_dump() for m in default_metrics]}
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/node-info")
+class EvaluationNodeInfoApi(Resource):
+    @console_ns.doc("get_evaluation_node_info")
+    @console_ns.response(200, "Node info grouped by metric")
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def post(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """Return workflow/snippet node info grouped by requested metrics.
+
+        Request body (JSON):
+            - metrics: list[str] | None  – metric names to query; omit or pass
+              an empty list to get all nodes under key ``"all"``.
+
+        Response:
+            ``{metric_or_all: [{"node_id": ..., "type": ..., "title": ...}, ...]}``
+        """
+        body = request.get_json(silent=True) or {}
+        metrics: list[str] | None = body.get("metrics") or None
+
+        result = EvaluationService.get_nodes_for_metrics(
+            target=target,
+            target_type=target_type,
+            metrics=metrics,
+        )
+        return result
+
+
+@console_ns.route("/evaluation/available-metrics")
+class EvaluationAvailableMetricsApi(Resource):
+    @console_ns.doc("get_available_evaluation_metrics")
+    @console_ns.response(200, "Available metrics list")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        """Return the centrally-defined list of evaluation metrics."""
+        return {"metrics": EvaluationService.get_available_metrics()}
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/files/<uuid:file_id>")
+class EvaluationFileDownloadApi(Resource):
+    @console_ns.doc("download_evaluation_file")
+    @console_ns.response(200, "File download URL generated successfully")
+    @console_ns.response(404, "Target or file not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str, file_id: str):
+        """
+        Download evaluation test file or result file.
+
+        Looks up the specified file, verifies it belongs to the same tenant,
+        and returns file info and download URL.
+        """
+        file_id = str(file_id)
+        _, current_tenant_id = current_account_with_tenant()
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            stmt = select(UploadFile).where(
+                UploadFile.id == file_id,
+                UploadFile.tenant_id == current_tenant_id,
+            )
+            upload_file = session.execute(stmt).scalar_one_or_none()
+
+        if not upload_file:
+            raise NotFound("File not found")
+
+        download_url = file_helpers.get_signed_file_url(upload_file_id=upload_file.id, as_attachment=True)
+
+        return {
+            "id": upload_file.id,
+            "name": upload_file.name,
+            "size": upload_file.size,
+            "extension": upload_file.extension,
+            "mime_type": upload_file.mime_type,
+            "created_at": int(upload_file.created_at.timestamp()) if upload_file.created_at else None,
+            "download_url": download_url,
+        }
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/version")
+class EvaluationVersionApi(Resource):
+    @console_ns.doc("get_evaluation_version_detail")
+    @console_ns.expect(console_ns.models.get(VersionQuery.__name__))
+    @console_ns.response(200, "Version details retrieved successfully")
+    @console_ns.response(404, "Target or version not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Get evaluation target version details.
+
+        Returns the workflow graph for the specified version.
+        """
+        version = request.args.get("version")
+
+        if not version:
+            return {"message": "version parameter is required"}, 400
+
+        graph = {}
+        if target_type == "snippets" and isinstance(target, CustomizedSnippet):
+            graph = target.graph_dict
+
+        return {
+            "graph": graph,
+        }
+
+
+@console_ns.route("/workspaces/current/available-evaluation-workflows")
+class AvailableEvaluationWorkflowsApi(Resource):
+    @console_ns.expect(console_ns.models[WorkflowListQuery.__name__])
+    @console_ns.doc("list_available_evaluation_workflows")
+    @console_ns.doc(description="List published evaluation workflows in the current workspace (all apps)")
+    @console_ns.response(
+        200,
+        "Available evaluation workflows retrieved",
+        available_evaluation_workflow_pagination_model,
+    )
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def get(self):
+        """List published evaluation-type workflows for the current tenant (cross-app)."""
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        args = WorkflowListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        page = args.page
+        limit = args.limit
+        user_id = args.user_id
+        named_only = args.named_only
+        keyword = args.keyword
+
+        if user_id and user_id != current_user.id:
+            raise Forbidden()
+
+        workflow_service = WorkflowService()
+        with Session(db.engine) as session:
+            workflows, has_more = workflow_service.list_published_evaluation_workflows(
+                session=session,
+                tenant_id=current_tenant_id,
+                page=page,
+                limit=limit,
+                user_id=user_id,
+                named_only=named_only,
+                keyword=keyword,
+            )
+
+            app_ids = {w.app_id for w in workflows}
+            if app_ids:
+                apps = session.scalars(select(App).where(App.id.in_(app_ids))).all()
+                app_names = {a.id: a.name for a in apps}
+            else:
+                app_names = {}
+
+        items = []
+        for wf in workflows:
+            items.append(
+                {
+                    "id": wf.id,
+                    "app_id": wf.app_id,
+                    "app_name": app_names.get(wf.app_id, ""),
+                    "type": wf.type.value,
+                    "version": wf.version,
+                    "marked_name": wf.marked_name,
+                    "marked_comment": wf.marked_comment,
+                    "hash": wf.unique_hash,
+                    "created_by": wf.created_by_account,
+                    "created_at": wf.created_at,
+                    "updated_by": wf.updated_by_account,
+                    "updated_at": wf.updated_at,
+                }
+            )
+
+        return (
+            marshal(
+                {"items": items, "page": page, "limit": limit, "has_more": has_more},
+                available_evaluation_workflow_pagination_fields,
+            ),
+            200,
+        )
+
+
+@console_ns.route("/workspaces/current/evaluation-workflows/<string:workflow_id>/associated-targets")
+class EvaluationWorkflowAssociatedTargetsApi(Resource):
+    @console_ns.doc("list_evaluation_workflow_associated_targets")
+    @console_ns.doc(
+        description="List targets (apps / snippets / knowledge bases) that use the given workflow as customized metrics"
+    )
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def get(self, workflow_id: str):
+        """Return all evaluation targets that reference this workflow as customized metrics."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        with Session(db.engine) as session:
+            configs = EvaluationService.list_targets_by_customized_workflow(
+                session=session,
+                tenant_id=current_tenant_id,
+                customized_workflow_id=workflow_id,
+            )
+
+            target_ids_by_type: dict[str, list[str]] = {}
+            for cfg in configs:
+                target_ids_by_type.setdefault(cfg.target_type, []).append(cfg.target_id)
+
+            app_names: dict[str, str] = {}
+            if "app" in target_ids_by_type:
+                apps = session.scalars(select(App).where(App.id.in_(target_ids_by_type["app"]))).all()
+                app_names = {a.id: a.name for a in apps}
+
+            snippet_names: dict[str, str] = {}
+            if "snippets" in target_ids_by_type:
+                snippets = session.scalars(
+                    select(CustomizedSnippet).where(CustomizedSnippet.id.in_(target_ids_by_type["snippets"]))
+                ).all()
+                snippet_names = {s.id: s.name for s in snippets}
+
+            dataset_names: dict[str, str] = {}
+            if "knowledge_base" in target_ids_by_type:
+                datasets = session.scalars(
+                    select(Dataset).where(Dataset.id.in_(target_ids_by_type["knowledge_base"]))
+                ).all()
+                dataset_names = {d.id: d.name for d in datasets}
+
+        items = []
+        for cfg in configs:
+            name = ""
+            if cfg.target_type == "app":
+                name = app_names.get(cfg.target_id, "")
+            elif cfg.target_type == "snippets":
+                name = snippet_names.get(cfg.target_id, "")
+            elif cfg.target_type == "knowledge_base":
+                name = dataset_names.get(cfg.target_id, "")
+
+            items.append(
+                {
+                    "target_type": cfg.target_type,
+                    "target_id": cfg.target_id,
+                    "target_name": name,
+                }
+            )
+
+        return {"items": items}, 200
+
+
+# ---- Serialization Helpers ----
+
+
+def _serialize_evaluation_run(run: EvaluationRun) -> dict[str, object]:
+    return {
+        "id": run.id,
+        "tenant_id": run.tenant_id,
+        "target_type": run.target_type,
+        "target_id": run.target_id,
+        "evaluation_config_id": run.evaluation_config_id,
+        "status": run.status,
+        "dataset_file_id": run.dataset_file_id,
+        "result_file_id": run.result_file_id,
+        "total_items": run.total_items,
+        "completed_items": run.completed_items,
+        "failed_items": run.failed_items,
+        "progress": run.progress,
+        "metrics_summary": run.metrics_summary_dict,
+        "error": run.error,
+        "created_by": run.created_by,
+        "started_at": int(run.started_at.timestamp()) if run.started_at else None,
+        "completed_at": int(run.completed_at.timestamp()) if run.completed_at else None,
+        "created_at": int(run.created_at.timestamp()) if run.created_at else None,
+    }
+
+
+def _serialize_evaluation_run_item(item: EvaluationRunItem) -> dict[str, object]:
+    return {
+        "id": item.id,
+        "item_index": item.item_index,
+        "inputs": item.inputs_dict,
+        "expected_output": item.expected_output,
+        "actual_output": item.actual_output,
+        "metrics": item.metrics_list,
+        "judgment": item.judgment_dict,
+        "metadata": item.metadata_dict,
+        "error": item.error,
+        "overall_score": item.overall_score,
+    }

api/controllers/console/notification.py

@@ -1,3 +1,4 @@
+from collections.abc import Mapping
 from typing import TypedDict
 
 from flask import request
@@ -13,6 +14,14 @@ from services.billing_service import BillingService
 _FALLBACK_LANG = "en-US"
 
 
+class NotificationLangContent(TypedDict, total=False):
+    lang: str
+    title: str
+    subtitle: str
+    body: str
+    titlePicUrl: str
+
+
 class NotificationItemDict(TypedDict):
     notification_id: str | None
     frequency: str | None
@@ -28,9 +37,11 @@ class NotificationResponseDict(TypedDict):
     notifications: list[NotificationItemDict]
 
 
-def _pick_lang_content(contents: dict, lang: str) -> dict:
+def _pick_lang_content(contents: Mapping[str, NotificationLangContent], lang: str) -> NotificationLangContent:
     """Return the single LangContent for *lang*, falling back to English."""
-    return contents.get(lang) or contents.get(_FALLBACK_LANG) or next(iter(contents.values()), {})
+    return (
+        contents.get(lang) or contents.get(_FALLBACK_LANG) or next(iter(contents.values()), NotificationLangContent())
+    )
 
 
 class DismissNotificationPayload(BaseModel):
@@ -71,7 +82,7 @@ class NotificationApi(Resource):
 
         notifications: list[NotificationItemDict] = []
         for notification in result.get("notifications") or []:
-            contents: dict = notification.get("contents") or {}
+            contents: Mapping[str, NotificationLangContent] = notification.get("contents") or {}
             lang_content = _pick_lang_content(contents, lang)
             item: NotificationItemDict = {
                 "notification_id": notification.get("notificationId"),

api/controllers/console/snippets/payloads.py

@@ -0,0 +1,135 @@
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field, field_validator
+
+
+class SnippetListQuery(BaseModel):
+    """Query parameters for listing snippets."""
+
+    page: int = Field(default=1, ge=1, le=99999)
+    limit: int = Field(default=20, ge=1, le=100)
+    keyword: str | None = None
+    is_published: bool | None = Field(default=None, description="Filter by published status")
+    creators: list[str] | None = Field(default=None, description="Filter by creator account IDs")
+
+    @field_validator("creators", mode="before")
+    @classmethod
+    def parse_creators(cls, value: object) -> list[str] | None:
+        """Normalize creators filter from query string or list input."""
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return [creator.strip() for creator in value.split(",") if creator.strip()] or None
+        if isinstance(value, list):
+            return [str(creator).strip() for creator in value if str(creator).strip()] or None
+        return None
+
+
+class IconInfo(BaseModel):
+    """Icon information model."""
+
+    icon: str | None = None
+    icon_type: Literal["emoji", "image"] | None = None
+    icon_background: str | None = None
+    icon_url: str | None = None
+
+
+class InputFieldDefinition(BaseModel):
+    """Input field definition for snippet parameters."""
+
+    default: str | None = None
+    hint: bool | None = None
+    label: str | None = None
+    max_length: int | None = None
+    options: list[str] | None = None
+    placeholder: str | None = None
+    required: bool | None = None
+    type: str | None = None  # e.g., "text-input"
+
+
+class CreateSnippetPayload(BaseModel):
+    """Payload for creating a new snippet."""
+
+    name: str = Field(..., min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=2000)
+    type: Literal["node", "group"] = "node"
+    icon_info: IconInfo | None = None
+    graph: dict[str, Any] | None = None
+    input_fields: list[InputFieldDefinition] | None = Field(default_factory=list)
+
+
+class UpdateSnippetPayload(BaseModel):
+    """Payload for updating a snippet."""
+
+    name: str | None = Field(default=None, min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=2000)
+    icon_info: IconInfo | None = None
+
+
+class SnippetDraftSyncPayload(BaseModel):
+    """Payload for syncing snippet draft workflow."""
+
+    graph: dict[str, Any]
+    hash: str | None = None
+    conversation_variables: list[dict[str, Any]] | None = Field(
+        default=None,
+        description="Ignored. Snippet workflows do not persist conversation variables.",
+    )
+    input_fields: list[dict[str, Any]] | None = None
+
+
+class WorkflowRunQuery(BaseModel):
+    """Query parameters for workflow runs."""
+
+    last_id: str | None = None
+    limit: int = Field(default=20, ge=1, le=100)
+
+
+class SnippetDraftRunPayload(BaseModel):
+    """Payload for running snippet draft workflow."""
+
+    inputs: dict[str, Any]
+    files: list[dict[str, Any]] | None = None
+
+
+class SnippetDraftNodeRunPayload(BaseModel):
+    """Payload for running a single node in snippet draft workflow."""
+
+    inputs: dict[str, Any]
+    query: str = ""
+    files: list[dict[str, Any]] | None = None
+
+
+class SnippetIterationNodeRunPayload(BaseModel):
+    """Payload for running an iteration node in snippet draft workflow."""
+
+    inputs: dict[str, Any] | None = None
+
+
+class SnippetLoopNodeRunPayload(BaseModel):
+    """Payload for running a loop node in snippet draft workflow."""
+
+    inputs: dict[str, Any] | None = None
+
+
+class PublishWorkflowPayload(BaseModel):
+    """Payload for publishing snippet workflow."""
+
+    knowledge_base_setting: dict[str, Any] | None = None
+
+
+class SnippetImportPayload(BaseModel):
+    """Payload for importing snippet from DSL."""
+
+    mode: str = Field(..., description="Import mode: yaml-content or yaml-url")
+    yaml_content: str | None = Field(default=None, description="YAML content (required for yaml-content mode)")
+    yaml_url: str | None = Field(default=None, description="YAML URL (required for yaml-url mode)")
+    name: str | None = Field(default=None, description="Override snippet name")
+    description: str | None = Field(default=None, description="Override snippet description")
+    snippet_id: str | None = Field(default=None, description="Snippet ID to update (optional)")
+
+
+class IncludeSecretQuery(BaseModel):
+    """Query parameter for including secret variables in export."""
+
+    include_secret: str = Field(default="false", description="Whether to include secret variables")

api/controllers/console/snippets/snippet_workflow.py

@@ -0,0 +1,534 @@
+import logging
+from collections.abc import Callable
+from functools import wraps
+from typing import ParamSpec, TypeVar
+
+from flask import request
+from flask_restx import Resource, marshal_with
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import InternalServerError, NotFound
+
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
+from controllers.console.app.error import DraftWorkflowNotExist, DraftWorkflowNotSync
+from controllers.console.app.workflow import workflow_model
+from controllers.console.app.workflow_run import (
+    workflow_run_detail_model,
+    workflow_run_node_execution_list_model,
+    workflow_run_node_execution_model,
+    workflow_run_pagination_model,
+)
+from controllers.console.snippets.payloads import (
+    PublishWorkflowPayload,
+    SnippetDraftNodeRunPayload,
+    SnippetDraftRunPayload,
+    SnippetDraftSyncPayload,
+    SnippetIterationNodeRunPayload,
+    SnippetLoopNodeRunPayload,
+    WorkflowRunQuery,
+)
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+)
+from core.app.apps.base_app_queue_manager import AppQueueManager
+from core.app.entities.app_invoke_entities import InvokeFrom
+from extensions.ext_database import db
+from extensions.ext_redis import redis_client
+from graphon.graph_engine.manager import GraphEngineManager
+from libs import helper
+from libs.helper import TimestampField
+from libs.login import current_account_with_tenant, login_required
+from models.snippet import CustomizedSnippet
+from services.errors.app import WorkflowHashNotEqualError
+from services.snippet_generate_service import SnippetGenerateService
+from services.snippet_service import SnippetService
+
+logger = logging.getLogger(__name__)
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+# Register Pydantic models with Swagger
+register_schema_models(
+    console_ns,
+    SnippetDraftSyncPayload,
+    SnippetDraftNodeRunPayload,
+    SnippetDraftRunPayload,
+    SnippetIterationNodeRunPayload,
+    SnippetLoopNodeRunPayload,
+    WorkflowRunQuery,
+    PublishWorkflowPayload,
+)
+
+
+class SnippetNotFoundError(Exception):
+    """Snippet not found error."""
+
+    pass
+
+
+def get_snippet(view_func: Callable[P, R]):
+    """Decorator to fetch and validate snippet access."""
+
+    @wraps(view_func)
+    def decorated_view(*args: P.args, **kwargs: P.kwargs):
+        if not kwargs.get("snippet_id"):
+            raise ValueError("missing snippet_id in path parameters")
+
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet_id = str(kwargs.get("snippet_id"))
+        del kwargs["snippet_id"]
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=snippet_id,
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        kwargs["snippet"] = snippet
+
+        return view_func(*args, **kwargs)
+
+    return decorated_view
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft")
+class SnippetDraftWorkflowApi(Resource):
+    @console_ns.doc("get_snippet_draft_workflow")
+    @console_ns.response(200, "Draft workflow retrieved successfully", workflow_model)
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    @marshal_with(workflow_model)
+    def get(self, snippet: CustomizedSnippet):
+        """Get draft workflow for snippet."""
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_draft_workflow(snippet=snippet)
+
+        if not workflow:
+            raise DraftWorkflowNotExist()
+
+        db.session.expunge(workflow)
+        workflow.conversation_variables = []
+        return workflow
+
+    @console_ns.doc("sync_snippet_draft_workflow")
+    @console_ns.expect(console_ns.models.get(SnippetDraftSyncPayload.__name__))
+    @console_ns.response(200, "Draft workflow synced successfully")
+    @console_ns.response(400, "Hash mismatch")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet):
+        """Sync draft workflow for snippet."""
+        current_user, _ = current_account_with_tenant()
+
+        payload = SnippetDraftSyncPayload.model_validate(console_ns.payload or {})
+
+        try:
+            snippet_service = SnippetService()
+            workflow = snippet_service.sync_draft_workflow(
+                snippet=snippet,
+                graph=payload.graph,
+                unique_hash=payload.hash,
+                account=current_user,
+                input_fields=payload.input_fields,
+            )
+        except WorkflowHashNotEqualError:
+            raise DraftWorkflowNotSync()
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        return {
+            "result": "success",
+            "hash": workflow.unique_hash,
+            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/config")
+class SnippetDraftConfigApi(Resource):
+    @console_ns.doc("get_snippet_draft_config")
+    @console_ns.response(200, "Draft config retrieved successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def get(self, snippet: CustomizedSnippet):
+        """Get snippet draft workflow configuration limits."""
+        return {
+            "parallel_depth_limit": 3,
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/publish")
+class SnippetPublishedWorkflowApi(Resource):
+    @console_ns.doc("get_snippet_published_workflow")
+    @console_ns.response(200, "Published workflow retrieved successfully", workflow_model)
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    @marshal_with(workflow_model)
+    def get(self, snippet: CustomizedSnippet):
+        """Get published workflow for snippet."""
+        if not snippet.is_published:
+            return None
+
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_published_workflow(snippet=snippet)
+
+        return workflow
+
+    @console_ns.doc("publish_snippet_workflow")
+    @console_ns.expect(console_ns.models.get(PublishWorkflowPayload.__name__))
+    @console_ns.response(200, "Workflow published successfully")
+    @console_ns.response(400, "No draft workflow found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet):
+        """Publish snippet workflow."""
+        current_user, _ = current_account_with_tenant()
+        snippet_service = SnippetService()
+
+        with Session(db.engine) as session:
+            snippet = session.merge(snippet)
+            try:
+                workflow = snippet_service.publish_workflow(
+                    session=session,
+                    snippet=snippet,
+                    account=current_user,
+                )
+                workflow_created_at = TimestampField().format(workflow.created_at)
+                session.commit()
+            except ValueError as e:
+                return {"message": str(e)}, 400
+
+        return {
+            "result": "success",
+            "created_at": workflow_created_at,
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/default-workflow-block-configs")
+class SnippetDefaultBlockConfigsApi(Resource):
+    @console_ns.doc("get_snippet_default_block_configs")
+    @console_ns.response(200, "Default block configs retrieved successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def get(self, snippet: CustomizedSnippet):
+        """Get default block configurations for snippet workflow."""
+        snippet_service = SnippetService()
+        return snippet_service.get_default_block_configs()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs")
+class SnippetWorkflowRunsApi(Resource):
+    @console_ns.doc("list_snippet_workflow_runs")
+    @console_ns.response(200, "Workflow runs retrieved successfully", workflow_run_pagination_model)
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_pagination_model)
+    def get(self, snippet: CustomizedSnippet):
+        """List workflow runs for snippet."""
+        query = WorkflowRunQuery.model_validate(
+            {
+                "last_id": request.args.get("last_id"),
+                "limit": request.args.get("limit", type=int, default=20),
+            }
+        )
+        args = {
+            "last_id": query.last_id,
+            "limit": query.limit,
+        }
+
+        snippet_service = SnippetService()
+        result = snippet_service.get_snippet_workflow_runs(snippet=snippet, args=args)
+
+        return result
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/<uuid:run_id>")
+class SnippetWorkflowRunDetailApi(Resource):
+    @console_ns.doc("get_snippet_workflow_run_detail")
+    @console_ns.response(200, "Workflow run detail retrieved successfully", workflow_run_detail_model)
+    @console_ns.response(404, "Workflow run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_detail_model)
+    def get(self, snippet: CustomizedSnippet, run_id):
+        """Get workflow run detail for snippet."""
+        run_id = str(run_id)
+
+        snippet_service = SnippetService()
+        workflow_run = snippet_service.get_snippet_workflow_run(snippet=snippet, run_id=run_id)
+
+        if not workflow_run:
+            raise NotFound("Workflow run not found")
+
+        return workflow_run
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/<uuid:run_id>/node-executions")
+class SnippetWorkflowRunNodeExecutionsApi(Resource):
+    @console_ns.doc("list_snippet_workflow_run_node_executions")
+    @console_ns.response(200, "Node executions retrieved successfully", workflow_run_node_execution_list_model)
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_node_execution_list_model)
+    def get(self, snippet: CustomizedSnippet, run_id):
+        """List node executions for a workflow run."""
+        run_id = str(run_id)
+
+        snippet_service = SnippetService()
+        node_executions = snippet_service.get_snippet_workflow_run_node_executions(
+            snippet=snippet,
+            run_id=run_id,
+        )
+
+        return {"data": node_executions}
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/nodes/<string:node_id>/run")
+class SnippetDraftNodeRunApi(Resource):
+    @console_ns.doc("run_snippet_draft_node")
+    @console_ns.doc(description="Run a single node in snippet draft workflow (single-step debugging)")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models.get(SnippetDraftNodeRunPayload.__name__))
+    @console_ns.response(200, "Node run completed successfully", workflow_run_node_execution_model)
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_node_execution_model)
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Run a single node in snippet draft workflow.
+
+        Executes a specific node with provided inputs for single-step debugging.
+        Returns the node execution result including status, outputs, and timing.
+        """
+        current_user, _ = current_account_with_tenant()
+        payload = SnippetDraftNodeRunPayload.model_validate(console_ns.payload or {})
+
+        user_inputs = payload.inputs
+
+        # Get draft workflow for file parsing
+        snippet_service = SnippetService()
+        draft_workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if not draft_workflow:
+            raise NotFound("Draft workflow not found")
+
+        files = SnippetGenerateService.parse_files(draft_workflow, payload.files)
+
+        workflow_node_execution = SnippetGenerateService.run_draft_node(
+            snippet=snippet,
+            node_id=node_id,
+            user_inputs=user_inputs,
+            account=current_user,
+            query=payload.query,
+            files=files,
+        )
+
+        return workflow_node_execution
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/nodes/<string:node_id>/last-run")
+class SnippetDraftNodeLastRunApi(Resource):
+    @console_ns.doc("get_snippet_draft_node_last_run")
+    @console_ns.doc(description="Get last run result for a node in snippet draft workflow")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.response(200, "Node last run retrieved successfully", workflow_run_node_execution_model)
+    @console_ns.response(404, "Snippet, draft workflow, or node last run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_node_execution_model)
+    def get(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Get the last run result for a specific node in snippet draft workflow.
+
+        Returns the most recent execution record for the given node,
+        including status, inputs, outputs, and timing information.
+        """
+        snippet_service = SnippetService()
+        draft_workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if not draft_workflow:
+            raise NotFound("Draft workflow not found")
+
+        node_exec = snippet_service.get_snippet_node_last_run(
+            snippet=snippet,
+            workflow=draft_workflow,
+            node_id=node_id,
+        )
+        if node_exec is None:
+            raise NotFound("Node last run not found")
+
+        return node_exec
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/iteration/nodes/<string:node_id>/run")
+class SnippetDraftRunIterationNodeApi(Resource):
+    @console_ns.doc("run_snippet_draft_iteration_node")
+    @console_ns.doc(description="Run draft workflow iteration node for snippet")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models.get(SnippetIterationNodeRunPayload.__name__))
+    @console_ns.response(200, "Iteration node run started successfully (SSE stream)")
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Run a draft workflow iteration node for snippet.
+
+        Iteration nodes execute their internal sub-graph multiple times over an input list.
+        Returns an SSE event stream with iteration progress and results.
+        """
+        current_user, _ = current_account_with_tenant()
+        args = SnippetIterationNodeRunPayload.model_validate(console_ns.payload or {}).model_dump(exclude_none=True)
+
+        try:
+            response = SnippetGenerateService.generate_single_iteration(
+                snippet=snippet, user=current_user, node_id=node_id, args=args, streaming=True
+            )
+
+            return helper.compact_generate_response(response)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/loop/nodes/<string:node_id>/run")
+class SnippetDraftRunLoopNodeApi(Resource):
+    @console_ns.doc("run_snippet_draft_loop_node")
+    @console_ns.doc(description="Run draft workflow loop node for snippet")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models.get(SnippetLoopNodeRunPayload.__name__))
+    @console_ns.response(200, "Loop node run started successfully (SSE stream)")
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Run a draft workflow loop node for snippet.
+
+        Loop nodes execute their internal sub-graph repeatedly until a condition is met.
+        Returns an SSE event stream with loop progress and results.
+        """
+        current_user, _ = current_account_with_tenant()
+        args = SnippetLoopNodeRunPayload.model_validate(console_ns.payload or {})
+
+        try:
+            response = SnippetGenerateService.generate_single_loop(
+                snippet=snippet, user=current_user, node_id=node_id, args=args, streaming=True
+            )
+
+            return helper.compact_generate_response(response)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/run")
+class SnippetDraftWorkflowRunApi(Resource):
+    @console_ns.doc("run_snippet_draft_workflow")
+    @console_ns.expect(console_ns.models.get(SnippetDraftRunPayload.__name__))
+    @console_ns.response(200, "Draft workflow run started successfully (SSE stream)")
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet):
+        """
+        Run draft workflow for snippet.
+
+        Executes the snippet's draft workflow with the provided inputs
+        and returns an SSE event stream with execution progress and results.
+        """
+        current_user, _ = current_account_with_tenant()
+
+        payload = SnippetDraftRunPayload.model_validate(console_ns.payload or {})
+        args = payload.model_dump(exclude_none=True)
+
+        try:
+            response = SnippetGenerateService.generate(
+                snippet=snippet,
+                user=current_user,
+                args=args,
+                invoke_from=InvokeFrom.DEBUGGER,
+                streaming=True,
+            )
+
+            return helper.compact_generate_response(response)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/tasks/<string:task_id>/stop")
+class SnippetWorkflowTaskStopApi(Resource):
+    @console_ns.doc("stop_snippet_workflow_task")
+    @console_ns.response(200, "Task stopped successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, task_id: str):
+        """
+        Stop a running snippet workflow task.
+
+        Uses both the legacy stop flag mechanism and the graph engine
+        command channel for backward compatibility.
+        """
+        # Stop using both mechanisms for backward compatibility
+        # Legacy stop flag mechanism (without user check)
+        AppQueueManager.set_stop_flag_no_user_check(task_id)
+
+        # New graph engine command channel mechanism
+        GraphEngineManager(redis_client).send_stop_command(task_id)
+
+        return {"result": "success"}

api/controllers/console/snippets/snippet_workflow_draft_variable.py

@@ -0,0 +1,319 @@
+"""
+Snippet draft workflow variable APIs.
+
+Mirrors console app routes under /apps/.../workflows/draft/variables for snippet scope,
+using CustomizedSnippet.id as WorkflowDraftVariable.app_id (same invariant as snippet execution).
+
+Snippet workflows do not expose system variables (`node_id == sys`) or conversation variables
+(`node_id == conversation`): paginated list queries exclude those rows; single-variable GET/PATCH/DELETE/reset
+reject them; `GET .../system-variables` and `GET .../conversation-variables` return empty lists for API parity.
+Other routes mirror `workflow_draft_variable` app APIs under `/snippets/...`.
+"""
+
+from collections.abc import Callable
+from functools import wraps
+from typing import Any, ParamSpec, TypeVar
+
+from flask import Response, request
+from flask_restx import Resource, marshal, marshal_with
+from sqlalchemy.orm import Session
+
+from controllers.console import console_ns
+from controllers.console.app.error import DraftWorkflowNotExist
+from controllers.console.app.workflow_draft_variable import (
+    WorkflowDraftVariableListQuery,
+    WorkflowDraftVariableUpdatePayload,
+    _ensure_variable_access,
+    _file_access_controller,
+    validate_node_id,
+    workflow_draft_variable_list_model,
+    workflow_draft_variable_list_without_value_model,
+    workflow_draft_variable_model,
+)
+from controllers.console.snippets.snippet_workflow import get_snippet
+from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
+from controllers.web.error import InvalidArgumentError, NotFoundError
+from core.workflow.variable_prefixes import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from extensions.ext_database import db
+from factories.file_factory import build_from_mapping, build_from_mappings
+from factories.variable_factory import build_segment_with_type
+from graphon.variables.types import SegmentType
+from libs.login import current_user, login_required
+from models.snippet import CustomizedSnippet
+from models.workflow import WorkflowDraftVariable
+from services.snippet_service import SnippetService
+from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+_SNIPPET_EXCLUDED_DRAFT_VARIABLE_NODE_IDS: frozenset[str] = frozenset(
+    {SYSTEM_VARIABLE_NODE_ID, CONVERSATION_VARIABLE_NODE_ID}
+)
+
+
+def _ensure_snippet_draft_variable_row_allowed(
+    *,
+    variable: WorkflowDraftVariable,
+    variable_id: str,
+) -> None:
+    """Snippet scope only supports canvas-node draft variables; treat sys/conversation rows as not found."""
+    if variable.node_id in _SNIPPET_EXCLUDED_DRAFT_VARIABLE_NODE_IDS:
+        raise NotFoundError(description=f"variable not found, id={variable_id}")
+
+
+def _snippet_draft_var_prerequisite(f: Callable[P, R]) -> Callable[P, R]:
+    """Setup, auth, snippet resolution, and tenant edit permission (same stack as snippet workflow APIs)."""
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    @wraps(f)
+    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
+        return f(*args, **kwargs)
+
+    return wrapper
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/variables")
+class SnippetWorkflowVariableCollectionApi(Resource):
+    @console_ns.expect(console_ns.models[WorkflowDraftVariableListQuery.__name__])
+    @console_ns.doc("get_snippet_workflow_variables")
+    @console_ns.doc(description="List draft workflow variables without values (paginated, snippet scope)")
+    @console_ns.response(
+        200,
+        "Workflow variables retrieved successfully",
+        workflow_draft_variable_list_without_value_model,
+    )
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_list_without_value_model)
+    def get(self, snippet: CustomizedSnippet) -> WorkflowDraftVariableList:
+        args = WorkflowDraftVariableListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+
+        snippet_service = SnippetService()
+        if snippet_service.get_draft_workflow(snippet=snippet) is None:
+            raise DraftWorkflowNotExist()
+
+        with Session(bind=db.engine, expire_on_commit=False) as session:
+            draft_var_srv = WorkflowDraftVariableService(session=session)
+            workflow_vars = draft_var_srv.list_variables_without_values(
+                app_id=snippet.id,
+                page=args.page,
+                limit=args.limit,
+                user_id=current_user.id,
+                exclude_node_ids=_SNIPPET_EXCLUDED_DRAFT_VARIABLE_NODE_IDS,
+            )
+
+        return workflow_vars
+
+    @console_ns.doc("delete_snippet_workflow_variables")
+    @console_ns.doc(description="Delete all draft workflow variables for the current user (snippet scope)")
+    @console_ns.response(204, "Workflow variables deleted successfully")
+    @_snippet_draft_var_prerequisite
+    def delete(self, snippet: CustomizedSnippet) -> Response:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        draft_var_srv.delete_user_workflow_variables(snippet.id, user_id=current_user.id)
+        db.session.commit()
+        return Response("", 204)
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/nodes/<string:node_id>/variables")
+class SnippetNodeVariableCollectionApi(Resource):
+    @console_ns.doc("get_snippet_node_variables")
+    @console_ns.doc(description="Get variables for a specific node (snippet draft workflow)")
+    @console_ns.response(200, "Node variables retrieved successfully", workflow_draft_variable_list_model)
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_list_model)
+    def get(self, snippet: CustomizedSnippet, node_id: str) -> WorkflowDraftVariableList:
+        validate_node_id(node_id)
+        with Session(bind=db.engine, expire_on_commit=False) as session:
+            draft_var_srv = WorkflowDraftVariableService(session=session)
+            node_vars = draft_var_srv.list_node_variables(snippet.id, node_id, user_id=current_user.id)
+
+        return node_vars
+
+    @console_ns.doc("delete_snippet_node_variables")
+    @console_ns.doc(description="Delete all variables for a specific node (snippet draft workflow)")
+    @console_ns.response(204, "Node variables deleted successfully")
+    @_snippet_draft_var_prerequisite
+    def delete(self, snippet: CustomizedSnippet, node_id: str) -> Response:
+        validate_node_id(node_id)
+        srv = WorkflowDraftVariableService(db.session())
+        srv.delete_node_variables(snippet.id, node_id, user_id=current_user.id)
+        db.session.commit()
+        return Response("", 204)
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/variables/<uuid:variable_id>")
+class SnippetVariableApi(Resource):
+    @console_ns.doc("get_snippet_workflow_variable")
+    @console_ns.doc(description="Get a specific draft workflow variable (snippet scope)")
+    @console_ns.response(200, "Variable retrieved successfully", workflow_draft_variable_model)
+    @console_ns.response(404, "Variable not found")
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_model)
+    def get(self, snippet: CustomizedSnippet, variable_id: str) -> WorkflowDraftVariable:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        variable = _ensure_variable_access(
+            variable=draft_var_srv.get_variable(variable_id=variable_id),
+            app_id=snippet.id,
+            variable_id=variable_id,
+        )
+        _ensure_snippet_draft_variable_row_allowed(variable=variable, variable_id=variable_id)
+        return variable
+
+    @console_ns.doc("update_snippet_workflow_variable")
+    @console_ns.doc(description="Update a draft workflow variable (snippet scope)")
+    @console_ns.expect(console_ns.models[WorkflowDraftVariableUpdatePayload.__name__])
+    @console_ns.response(200, "Variable updated successfully", workflow_draft_variable_model)
+    @console_ns.response(404, "Variable not found")
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_model)
+    def patch(self, snippet: CustomizedSnippet, variable_id: str) -> WorkflowDraftVariable:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        args_model = WorkflowDraftVariableUpdatePayload.model_validate(console_ns.payload or {})
+
+        variable = _ensure_variable_access(
+            variable=draft_var_srv.get_variable(variable_id=variable_id),
+            app_id=snippet.id,
+            variable_id=variable_id,
+        )
+        _ensure_snippet_draft_variable_row_allowed(variable=variable, variable_id=variable_id)
+
+        new_name = args_model.name
+        raw_value = args_model.value
+        if new_name is None and raw_value is None:
+            return variable
+
+        new_value = None
+        if raw_value is not None:
+            if variable.value_type == SegmentType.FILE:
+                if not isinstance(raw_value, dict):
+                    raise InvalidArgumentError(description=f"expected dict for file, got {type(raw_value)}")
+                raw_value = build_from_mapping(
+                    mapping=raw_value,
+                    tenant_id=snippet.tenant_id,
+                    access_controller=_file_access_controller,
+                )
+            elif variable.value_type == SegmentType.ARRAY_FILE:
+                if not isinstance(raw_value, list):
+                    raise InvalidArgumentError(description=f"expected list for files, got {type(raw_value)}")
+                if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
+                    raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
+                raw_value = build_from_mappings(
+                    mappings=raw_value,
+                    tenant_id=snippet.tenant_id,
+                    access_controller=_file_access_controller,
+                )
+            new_value = build_segment_with_type(variable.value_type, raw_value)
+        draft_var_srv.update_variable(variable, name=new_name, value=new_value)
+        db.session.commit()
+        return variable
+
+    @console_ns.doc("delete_snippet_workflow_variable")
+    @console_ns.doc(description="Delete a draft workflow variable (snippet scope)")
+    @console_ns.response(204, "Variable deleted successfully")
+    @console_ns.response(404, "Variable not found")
+    @_snippet_draft_var_prerequisite
+    def delete(self, snippet: CustomizedSnippet, variable_id: str) -> Response:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        variable = _ensure_variable_access(
+            variable=draft_var_srv.get_variable(variable_id=variable_id),
+            app_id=snippet.id,
+            variable_id=variable_id,
+        )
+        _ensure_snippet_draft_variable_row_allowed(variable=variable, variable_id=variable_id)
+        draft_var_srv.delete_variable(variable)
+        db.session.commit()
+        return Response("", 204)
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/variables/<uuid:variable_id>/reset")
+class SnippetVariableResetApi(Resource):
+    @console_ns.doc("reset_snippet_workflow_variable")
+    @console_ns.doc(description="Reset a draft workflow variable to its default value (snippet scope)")
+    @console_ns.response(200, "Variable reset successfully", workflow_draft_variable_model)
+    @console_ns.response(204, "Variable reset (no content)")
+    @console_ns.response(404, "Variable not found")
+    @_snippet_draft_var_prerequisite
+    def put(self, snippet: CustomizedSnippet, variable_id: str) -> Response | Any:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        snippet_service = SnippetService()
+        draft_workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if draft_workflow is None:
+            raise NotFoundError(
+                f"Draft workflow not found, snippet_id={snippet.id}",
+            )
+        variable = _ensure_variable_access(
+            variable=draft_var_srv.get_variable(variable_id=variable_id),
+            app_id=snippet.id,
+            variable_id=variable_id,
+        )
+        _ensure_snippet_draft_variable_row_allowed(variable=variable, variable_id=variable_id)
+
+        resetted = draft_var_srv.reset_variable(draft_workflow, variable)
+        db.session.commit()
+        if resetted is None:
+            return Response("", 204)
+        return marshal(resetted, workflow_draft_variable_model)
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/conversation-variables")
+class SnippetConversationVariableCollectionApi(Resource):
+    @console_ns.doc("get_snippet_conversation_variables")
+    @console_ns.doc(
+        description="Conversation variables are not used in snippet workflows; returns an empty list for API parity"
+    )
+    @console_ns.response(200, "Conversation variables retrieved successfully", workflow_draft_variable_list_model)
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_list_model)
+    def get(self, snippet: CustomizedSnippet) -> WorkflowDraftVariableList:
+        return WorkflowDraftVariableList(variables=[])
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/system-variables")
+class SnippetSystemVariableCollectionApi(Resource):
+    @console_ns.doc("get_snippet_system_variables")
+    @console_ns.doc(
+        description="System variables are not used in snippet workflows; returns an empty list for API parity"
+    )
+    @console_ns.response(200, "System variables retrieved successfully", workflow_draft_variable_list_model)
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_list_model)
+    def get(self, snippet: CustomizedSnippet) -> WorkflowDraftVariableList:
+        return WorkflowDraftVariableList(variables=[])
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/environment-variables")
+class SnippetEnvironmentVariableCollectionApi(Resource):
+    @console_ns.doc("get_snippet_environment_variables")
+    @console_ns.doc(description="Get environment variables from snippet draft workflow graph")
+    @console_ns.response(200, "Environment variables retrieved successfully")
+    @console_ns.response(404, "Draft workflow not found")
+    @_snippet_draft_var_prerequisite
+    def get(self, snippet: CustomizedSnippet) -> dict[str, list[dict[str, Any]]]:
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if workflow is None:
+            raise DraftWorkflowNotExist()
+
+        env_vars_list: list[dict[str, Any]] = []
+        for v in workflow.environment_variables:
+            env_vars_list.append(
+                {
+                    "id": v.id,
+                    "type": "env",
+                    "name": v.name,
+                    "description": v.description,
+                    "selector": v.selector,
+                    "value_type": v.value_type.exposed_type().value,
+                    "value": v.value,
+                    "edited": False,
+                    "visible": True,
+                    "editable": True,
+                }
+            )
+
+        return {"items": env_vars_list}

api/controllers/console/workspace/__init__.py

@@ -35,22 +35,24 @@ def plugin_permission_required(
                     return view(*args, **kwargs)
 
                 if install_required:
-                    if permission.install_permission == TenantPluginPermission.InstallPermission.NOBODY:
-                        raise Forbidden()
-                    if permission.install_permission == TenantPluginPermission.InstallPermission.ADMINS:
-                        if not user.is_admin_or_owner:
+                    match permission.install_permission:
+                        case TenantPluginPermission.InstallPermission.NOBODY:
                             raise Forbidden()
-                    if permission.install_permission == TenantPluginPermission.InstallPermission.EVERYONE:
-                        pass
+                        case TenantPluginPermission.InstallPermission.ADMINS:
+                            if not user.is_admin_or_owner:
+                                raise Forbidden()
+                        case TenantPluginPermission.InstallPermission.EVERYONE:
+                            pass
 
                 if debug_required:
-                    if permission.debug_permission == TenantPluginPermission.DebugPermission.NOBODY:
-                        raise Forbidden()
-                    if permission.debug_permission == TenantPluginPermission.DebugPermission.ADMINS:
-                        if not user.is_admin_or_owner:
+                    match permission.debug_permission:
+                        case TenantPluginPermission.DebugPermission.NOBODY:
                             raise Forbidden()
-                    if permission.debug_permission == TenantPluginPermission.DebugPermission.EVERYONE:
-                        pass
+                        case TenantPluginPermission.DebugPermission.ADMINS:
+                            if not user.is_admin_or_owner:
+                                raise Forbidden()
+                        case TenantPluginPermission.DebugPermission.EVERYONE:
+                            pass
 
             return view(*args, **kwargs)
 

api/controllers/console/workspace/account.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import Literal
+from typing import Any, Literal
 
 import pytz
 from flask import request
@@ -174,7 +174,7 @@ reg(CheckEmailUniquePayload)
 register_schema_models(console_ns, AccountResponse)
 
 
-def _serialize_account(account) -> dict:
+def _serialize_account(account) -> dict[str, Any]:
     return AccountResponse.model_validate(account, from_attributes=True).model_dump(mode="json")
 
 

api/controllers/console/workspace/snippets.py

@@ -0,0 +1,380 @@
+import logging
+from urllib.parse import quote
+
+from flask import Response, request
+from flask_restx import Resource, marshal
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import NotFound
+
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
+from controllers.console.snippets.payloads import (
+    CreateSnippetPayload,
+    IncludeSecretQuery,
+    SnippetImportPayload,
+    SnippetListQuery,
+    UpdateSnippetPayload,
+)
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+)
+from extensions.ext_database import db
+from fields.snippet_fields import snippet_fields, snippet_list_fields, snippet_pagination_fields
+from libs.login import current_account_with_tenant, login_required
+from models.snippet import SnippetType
+from services.app_dsl_service import ImportStatus
+from services.snippet_dsl_service import SnippetDslService
+from services.snippet_service import SnippetService
+
+logger = logging.getLogger(__name__)
+
+# Register Pydantic models with Swagger
+register_schema_models(
+    console_ns,
+    SnippetListQuery,
+    CreateSnippetPayload,
+    UpdateSnippetPayload,
+    SnippetImportPayload,
+    IncludeSecretQuery,
+)
+
+# Create namespace models for marshaling
+snippet_model = console_ns.model("Snippet", snippet_fields)
+snippet_list_model = console_ns.model("SnippetList", snippet_list_fields)
+snippet_pagination_model = console_ns.model("SnippetPagination", snippet_pagination_fields)
+
+
+@console_ns.route("/workspaces/current/customized-snippets")
+class CustomizedSnippetsApi(Resource):
+    @console_ns.doc("list_customized_snippets")
+    @console_ns.expect(console_ns.models.get(SnippetListQuery.__name__))
+    @console_ns.response(200, "Snippets retrieved successfully", snippet_pagination_model)
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        """List customized snippets with pagination and search."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        query_params = request.args.to_dict()
+        query = SnippetListQuery.model_validate(query_params)
+
+        snippets, total, has_more = SnippetService.get_snippets(
+            tenant_id=current_tenant_id,
+            page=query.page,
+            limit=query.limit,
+            keyword=query.keyword,
+            is_published=query.is_published,
+            creators=query.creators,
+        )
+
+        return {
+            "data": marshal(snippets, snippet_list_fields),
+            "page": query.page,
+            "limit": query.limit,
+            "total": total,
+            "has_more": has_more,
+        }, 200
+
+    @console_ns.doc("create_customized_snippet")
+    @console_ns.expect(console_ns.models.get(CreateSnippetPayload.__name__))
+    @console_ns.response(201, "Snippet created successfully", snippet_model)
+    @console_ns.response(400, "Invalid request or name already exists")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self):
+        """Create a new customized snippet."""
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        payload = CreateSnippetPayload.model_validate(console_ns.payload or {})
+
+        try:
+            snippet_type = SnippetType(payload.type)
+        except ValueError:
+            snippet_type = SnippetType.NODE
+
+        try:
+            snippet = SnippetService.create_snippet(
+                tenant_id=current_tenant_id,
+                name=payload.name,
+                description=payload.description,
+                snippet_type=snippet_type,
+                icon_info=payload.icon_info.model_dump() if payload.icon_info else None,
+                input_fields=[f.model_dump() for f in payload.input_fields] if payload.input_fields else None,
+                account=current_user,
+            )
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        return marshal(snippet, snippet_fields), 201
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>")
+class CustomizedSnippetDetailApi(Resource):
+    @console_ns.doc("get_customized_snippet")
+    @console_ns.response(200, "Snippet retrieved successfully", snippet_model)
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, snippet_id: str):
+        """Get customized snippet details."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        return marshal(snippet, snippet_fields), 200
+
+    @console_ns.doc("update_customized_snippet")
+    @console_ns.expect(console_ns.models.get(UpdateSnippetPayload.__name__))
+    @console_ns.response(200, "Snippet updated successfully", snippet_model)
+    @console_ns.response(400, "Invalid request or name already exists")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def patch(self, snippet_id: str):
+        """Update customized snippet."""
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        payload = UpdateSnippetPayload.model_validate(console_ns.payload or {})
+        update_data = payload.model_dump(exclude_unset=True)
+
+        if "icon_info" in update_data and update_data["icon_info"] is not None:
+            update_data["icon_info"] = payload.icon_info.model_dump() if payload.icon_info else None
+
+        if not update_data:
+            return {"message": "No valid fields to update"}, 400
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                snippet = session.merge(snippet)
+                snippet = SnippetService.update_snippet(
+                    session=session,
+                    snippet=snippet,
+                    account_id=current_user.id,
+                    data=update_data,
+                )
+                session.commit()
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        return marshal(snippet, snippet_fields), 200
+
+    @console_ns.doc("delete_customized_snippet")
+    @console_ns.response(204, "Snippet deleted successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def delete(self, snippet_id: str):
+        """Delete customized snippet."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        with Session(db.engine) as session:
+            snippet = session.merge(snippet)
+            SnippetService.delete_snippet(
+                session=session,
+                snippet=snippet,
+            )
+            session.commit()
+
+        return "", 204
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>/export")
+class CustomizedSnippetExportApi(Resource):
+    @console_ns.doc("export_customized_snippet")
+    @console_ns.doc(description="Export snippet configuration as DSL")
+    @console_ns.doc(params={"snippet_id": "Snippet ID to export"})
+    @console_ns.response(200, "Snippet exported successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def get(self, snippet_id: str):
+        """Export snippet as DSL."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        # Get include_secret parameter
+        query = IncludeSecretQuery.model_validate(request.args.to_dict())
+
+        with Session(db.engine) as session:
+            export_service = SnippetDslService(session)
+            result = export_service.export_snippet_dsl(snippet=snippet, include_secret=query.include_secret == "true")
+
+        # Set filename with .snippet extension
+        filename = f"{snippet.name}.snippet"
+        encoded_filename = quote(filename)
+        
+        response = Response(
+            result,
+            mimetype="application/x-yaml",
+        )
+        response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
+        response.headers["Content-Type"] = "application/x-yaml"
+        
+        return response
+
+
+@console_ns.route("/workspaces/current/customized-snippets/imports")
+class CustomizedSnippetImportApi(Resource):
+    @console_ns.doc("import_customized_snippet")
+    @console_ns.doc(description="Import snippet from DSL")
+    @console_ns.expect(console_ns.models.get(SnippetImportPayload.__name__))
+    @console_ns.response(200, "Snippet imported successfully")
+    @console_ns.response(202, "Import pending confirmation")
+    @console_ns.response(400, "Import failed")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self):
+        """Import snippet from DSL."""
+        current_user, _ = current_account_with_tenant()
+        payload = SnippetImportPayload.model_validate(console_ns.payload or {})
+
+        with Session(db.engine) as session:
+            import_service = SnippetDslService(session)
+            result = import_service.import_snippet(
+                account=current_user,
+                import_mode=payload.mode,
+                yaml_content=payload.yaml_content,
+                yaml_url=payload.yaml_url,
+                snippet_id=payload.snippet_id,
+                name=payload.name,
+                description=payload.description,
+            )
+            session.commit()
+
+        # Return appropriate status code based on result
+        status = result.status
+        if status == ImportStatus.FAILED:
+            return result.model_dump(mode="json"), 400
+        elif status == ImportStatus.PENDING:
+            return result.model_dump(mode="json"), 202
+        return result.model_dump(mode="json"), 200
+
+
+@console_ns.route("/workspaces/current/customized-snippets/imports/<string:import_id>/confirm")
+class CustomizedSnippetImportConfirmApi(Resource):
+    @console_ns.doc("confirm_snippet_import")
+    @console_ns.doc(description="Confirm a pending snippet import")
+    @console_ns.doc(params={"import_id": "Import ID to confirm"})
+    @console_ns.response(200, "Import confirmed successfully")
+    @console_ns.response(400, "Import failed")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self, import_id: str):
+        """Confirm a pending snippet import."""
+        current_user, _ = current_account_with_tenant()
+
+        with Session(db.engine) as session:
+            import_service = SnippetDslService(session)
+            result = import_service.confirm_import(import_id=import_id, account=current_user)
+            session.commit()
+
+        if result.status == ImportStatus.FAILED:
+            return result.model_dump(mode="json"), 400
+        return result.model_dump(mode="json"), 200
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>/check-dependencies")
+class CustomizedSnippetCheckDependenciesApi(Resource):
+    @console_ns.doc("check_snippet_dependencies")
+    @console_ns.doc(description="Check dependencies for a snippet")
+    @console_ns.doc(params={"snippet_id": "Snippet ID"})
+    @console_ns.response(200, "Dependencies checked successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def get(self, snippet_id: str):
+        """Check dependencies for a snippet."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        with Session(db.engine) as session:
+            import_service = SnippetDslService(session)
+            result = import_service.check_dependencies(snippet=snippet)
+
+        return result.model_dump(mode="json"), 200
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>/use-count/increment")
+class CustomizedSnippetUseCountIncrementApi(Resource):
+    @console_ns.doc("increment_snippet_use_count")
+    @console_ns.doc(description="Increment snippet use count by 1")
+    @console_ns.doc(params={"snippet_id": "Snippet ID"})
+    @console_ns.response(200, "Use count incremented successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self, snippet_id: str):
+        """Increment snippet use count when it is inserted into a workflow."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        with Session(db.engine) as session:
+            snippet = session.merge(snippet)
+            SnippetService.increment_use_count(session=session, snippet=snippet)
+            session.commit()
+            session.refresh(snippet)
+
+        return {"result": "success", "use_count": snippet.use_count}, 200

api/controllers/console/wraps.py

@@ -20,7 +20,7 @@ from models.account import AccountStatus
 from models.dataset import RateLimitLog
 from models.model import DifySetup
 from services.feature_service import FeatureService, LicenseStatus
-from services.operation_service import OperationService
+from services.operation_service import OperationService, UtmInfo
 
 from .error import NotInitValidateError, NotSetupError, UnauthorizedAndForceLogout
 
@@ -205,7 +205,7 @@ def cloud_utm_record[**P, R](view: Callable[P, R]) -> Callable[P, R]:
                 utm_info = request.cookies.get("utm_info")
 
                 if utm_info:
-                    utm_info_dict: dict = json.loads(utm_info)
+                    utm_info_dict: UtmInfo = json.loads(utm_info)
                     OperationService.record_utm(current_tenant_id, utm_info_dict)
 
         return view(*args, **kwargs)

api/controllers/mcp/mcp.py

@@ -2,7 +2,7 @@ from typing import Any, Union
 
 from flask import Response
 from flask_restx import Resource
-from graphon.variables.input_entities import VariableEntity
+from graphon.variables.input_entities import VariableEntity, VariableEntityType
 from pydantic import BaseModel, Field, ValidationError
 from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
@@ -158,14 +158,20 @@ class MCPAppApi(Resource):
         except ValidationError as e:
             raise MCPRequestError(mcp_types.INVALID_PARAMS, f"Invalid user_input_form: {str(e)}")
 
-    def _convert_user_input_form(self, raw_form: list[dict]) -> list[VariableEntity]:
+    def _convert_user_input_form(self, raw_form: list[dict[str, Any]]) -> list[VariableEntity]:
         """Convert raw user input form to VariableEntity objects"""
         return [self._create_variable_entity(item) for item in raw_form]
 
-    def _create_variable_entity(self, item: dict) -> VariableEntity:
+    def _create_variable_entity(self, item: dict[str, Any]) -> VariableEntity:
         """Create a single VariableEntity from raw form item"""
-        variable_type = item.get("type", "") or list(item.keys())[0]
-        variable = item[variable_type]
+        variable_type_raw: str = item.get("type", "") or list(item.keys())[0]
+        try:
+            variable_type = VariableEntityType(variable_type_raw)
+        except ValueError as e:
+            raise MCPRequestError(
+                mcp_types.INVALID_PARAMS, f"Invalid user_input_form variable type: {variable_type_raw}"
+            ) from e
+        variable = item[variable_type_raw]
 
         return VariableEntity(
             type=variable_type,
@@ -178,7 +184,7 @@ class MCPAppApi(Resource):
             json_schema=variable.get("json_schema"),
         )
 
-    def _parse_mcp_request(self, args: dict) -> mcp_types.ClientRequest | mcp_types.ClientNotification:
+    def _parse_mcp_request(self, args: dict[str, Any]) -> mcp_types.ClientRequest | mcp_types.ClientNotification:
         """Parse and validate MCP request"""
         try:
             return mcp_types.ClientRequest.model_validate(args)

api/controllers/service_api/app/audio.py

@@ -3,10 +3,10 @@ import logging
 from flask import request
 from flask_restx import Resource
 from graphon.model_runtime.errors.invoke import InvokeError
-from pydantic import BaseModel, Field
 from werkzeug.exceptions import InternalServerError
 
 import services
+from controllers.common.controller_schemas import TextToAudioPayload
 from controllers.common.schema import register_schema_model
 from controllers.service_api import service_api_ns
 from controllers.service_api.app.error import (
@@ -86,13 +86,6 @@ class AudioApi(Resource):
             raise InternalServerError()
 
 
-class TextToAudioPayload(BaseModel):
-    message_id: str | None = Field(default=None, description="Message ID")
-    voice: str | None = Field(default=None, description="Voice to use for TTS")
-    text: str | None = Field(default=None, description="Text to convert to audio")
-    streaming: bool | None = Field(default=None, description="Enable streaming response")
-
-
 register_schema_model(service_api_ns, TextToAudioPayload)
 
 

api/controllers/service_api/dataset/document.py

@@ -10,6 +10,7 @@ from sqlalchemy import desc, func, select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
+from controllers.common.controller_schemas import DocumentBatchDownloadZipPayload
 from controllers.common.errors import (
     FilenameNotExistsError,
     FileTooLargeError,
@@ -100,15 +101,6 @@ class DocumentListQuery(BaseModel):
     status: str | None = Field(default=None, description="Document status filter")
 
 
-DOCUMENT_BATCH_DOWNLOAD_ZIP_MAX_DOCS = 100
-
-
-class DocumentBatchDownloadZipPayload(BaseModel):
-    """Request payload for bulk downloading uploaded documents as a ZIP archive."""
-
-    document_ids: list[UUID] = Field(..., min_length=1, max_length=DOCUMENT_BATCH_DOWNLOAD_ZIP_MAX_DOCS)
-
-
 register_enum_models(service_api_ns, RetrievalMethod)
 
 register_schema_models(

api/controllers/service_api/dataset/metadata.py

@@ -2,9 +2,9 @@ from typing import Literal
 
 from flask_login import current_user
 from flask_restx import marshal
-from pydantic import BaseModel
 from werkzeug.exceptions import NotFound
 
+from controllers.common.controller_schemas import MetadataUpdatePayload
 from controllers.common.schema import register_schema_model, register_schema_models
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import DatasetApiResource, cloud_edition_billing_rate_limit_check
@@ -18,11 +18,6 @@ from services.entities.knowledge_entities.knowledge_entities import (
 )
 from services.metadata_service import MetadataService
 
-
-class MetadataUpdatePayload(BaseModel):
-    name: str
-
-
 register_schema_model(service_api_ns, MetadataUpdatePayload)
 register_schema_models(
     service_api_ns,

api/controllers/service_api/dataset/segment.py

@@ -8,6 +8,7 @@ from sqlalchemy import select
 from werkzeug.exceptions import NotFound
 
 from configs import dify_config
+from controllers.common.controller_schemas import ChildChunkCreatePayload, ChildChunkUpdatePayload
 from controllers.common.schema import register_schema_models
 from controllers.service_api import service_api_ns
 from controllers.service_api.app.error import ProviderNotInitializeError
@@ -32,25 +33,25 @@ from services.errors.chunk import ChildChunkIndexingError as ChildChunkIndexingS
 from services.summary_index_service import SummaryIndexService
 
 
-def _marshal_segment_with_summary(segment, dataset_id: str) -> dict:
+def _marshal_segment_with_summary(segment, dataset_id: str) -> dict[str, Any]:
     """Marshal a single segment and enrich it with summary content."""
-    segment_dict = dict(marshal(segment, segment_fields))  # type: ignore[arg-type]
+    segment_dict: dict[str, Any] = dict(marshal(segment, segment_fields))  # type: ignore[arg-type]
     summary = SummaryIndexService.get_segment_summary(segment_id=segment.id, dataset_id=dataset_id)
     segment_dict["summary"] = summary.summary_content if summary else None
     return segment_dict
 
 
-def _marshal_segments_with_summary(segments, dataset_id: str) -> list[dict]:
+def _marshal_segments_with_summary(segments, dataset_id: str) -> list[dict[str, Any]]:
     """Marshal multiple segments and enrich them with summary content (batch query)."""
     segment_ids = [segment.id for segment in segments]
-    summaries: dict = {}
+    summaries: dict[str, str | None] = {}
     if segment_ids:
         summary_records = SummaryIndexService.get_segments_summaries(segment_ids=segment_ids, dataset_id=dataset_id)
         summaries = {chunk_id: record.summary_content for chunk_id, record in summary_records.items()}
 
-    result = []
+    result: list[dict[str, Any]] = []
     for segment in segments:
-        segment_dict = dict(marshal(segment, segment_fields))  # type: ignore[arg-type]
+        segment_dict: dict[str, Any] = dict(marshal(segment, segment_fields))  # type: ignore[arg-type]
         segment_dict["summary"] = summaries.get(segment.id)
         result.append(segment_dict)
     return result
@@ -69,20 +70,12 @@ class SegmentUpdatePayload(BaseModel):
     segment: SegmentUpdateArgs
 
 
-class ChildChunkCreatePayload(BaseModel):
-    content: str
-
-
 class ChildChunkListQuery(BaseModel):
     limit: int = Field(default=20, ge=1)
     keyword: str | None = None
     page: int = Field(default=1, ge=1)
 
 
-class ChildChunkUpdatePayload(BaseModel):
-    content: str
-
-
 register_schema_models(
     service_api_ns,
     SegmentCreatePayload,

api/controllers/web/human_input_form.py

@@ -5,6 +5,7 @@ Web App Human Input Form APIs.
 import json
 import logging
 from datetime import datetime
+from typing import Any, NotRequired, TypedDict
 
 from flask import Response, request
 from flask_restx import Resource
@@ -58,10 +59,19 @@ def _to_timestamp(value: datetime) -> int:
     return int(value.timestamp())
 
 
+class FormDefinitionPayload(TypedDict):
+    form_content: Any
+    inputs: Any
+    resolved_default_values: dict[str, str]
+    user_actions: Any
+    expiration_time: int
+    site: NotRequired[dict]
+
+
 def _jsonify_form_definition(form: Form, site_payload: dict | None = None) -> Response:
     """Return the form payload (optionally with site) as a JSON response."""
     definition_payload = form.get_definition().model_dump()
-    payload = {
+    payload: FormDefinitionPayload = {
         "form_content": definition_payload["rendered_content"],
         "inputs": definition_payload["inputs"],
         "resolved_default_values": _stringify_default_values(definition_payload["default_values"]),
@@ -92,7 +102,7 @@ class HumanInputFormApi(Resource):
         _FORM_ACCESS_RATE_LIMITER.increment_rate_limit(ip_address)
 
         service = HumanInputService(db.engine)
-        # TODO(QuantumGhost): forbid submision for form tokens
+        # TODO(QuantumGhost): forbid submission for form tokens
         # that are only for console.
         form = service.get_form_by_token(form_token)
 

api/controllers/web/login.py

@@ -1,7 +1,10 @@
+import logging
+
 from flask import make_response, request
 from flask_restx import Resource
 from jwt import InvalidTokenError
 from pydantic import BaseModel, Field, field_validator
+from werkzeug.exceptions import Unauthorized
 
 import services
 from configs import dify_config
@@ -20,7 +23,7 @@ from controllers.console.wraps import (
 )
 from controllers.web import web_ns
 from controllers.web.wraps import decode_jwt_token
-from libs.helper import EmailStr
+from libs.helper import EmailStr, extract_remote_ip
 from libs.passport import PassportService
 from libs.password import valid_password
 from libs.token import (
@@ -29,9 +32,11 @@ from libs.token import (
 )
 from services.account_service import AccountService
 from services.app_service import AppService
-from services.entities.auth_entities import LoginPayloadBase
+from services.entities.auth_entities import LoginFailureReason, LoginPayloadBase
 from services.webapp_auth_service import WebAppAuthService
 
+logger = logging.getLogger(__name__)
+
 
 class LoginPayload(LoginPayloadBase):
     @field_validator("password")
@@ -76,14 +81,18 @@ class LoginApi(Resource):
     def post(self):
         """Authenticate user and login."""
         payload = LoginPayload.model_validate(web_ns.payload or {})
+        normalized_email = payload.email.lower()
 
         try:
             account = WebAppAuthService.authenticate(payload.email, payload.password)
         except services.errors.account.AccountLoginError:
+            _log_web_login_failure(email=normalized_email, reason=LoginFailureReason.ACCOUNT_BANNED)
             raise AccountBannedError()
         except services.errors.account.AccountPasswordError:
+            _log_web_login_failure(email=normalized_email, reason=LoginFailureReason.INVALID_CREDENTIALS)
             raise AuthenticationFailedError()
         except services.errors.account.AccountNotFoundError:
+            _log_web_login_failure(email=normalized_email, reason=LoginFailureReason.ACCOUNT_NOT_FOUND)
             raise AuthenticationFailedError()
 
         token = WebAppAuthService.login(account=account)
@@ -212,21 +221,30 @@ class EmailCodeLoginApi(Resource):
 
         token_data = WebAppAuthService.get_email_code_login_data(payload.token)
         if token_data is None:
+            _log_web_login_failure(email=user_email, reason=LoginFailureReason.INVALID_EMAIL_CODE_TOKEN)
             raise InvalidTokenError()
 
         token_email = token_data.get("email")
         if not isinstance(token_email, str):
+            _log_web_login_failure(email=user_email, reason=LoginFailureReason.EMAIL_CODE_EMAIL_MISMATCH)
             raise InvalidEmailError()
         normalized_token_email = token_email.lower()
         if normalized_token_email != user_email:
+            _log_web_login_failure(email=user_email, reason=LoginFailureReason.EMAIL_CODE_EMAIL_MISMATCH)
             raise InvalidEmailError()
 
         if token_data["code"] != payload.code:
+            _log_web_login_failure(email=user_email, reason=LoginFailureReason.INVALID_EMAIL_CODE)
             raise EmailCodeError()
 
         WebAppAuthService.revoke_email_code_login_token(payload.token)
-        account = WebAppAuthService.get_user_through_email(token_email)
+        try:
+            account = WebAppAuthService.get_user_through_email(token_email)
+        except Unauthorized as exc:
+            _log_web_login_failure(email=user_email, reason=LoginFailureReason.ACCOUNT_BANNED)
+            raise AccountBannedError() from exc
         if not account:
+            _log_web_login_failure(email=user_email, reason=LoginFailureReason.ACCOUNT_NOT_FOUND)
             raise AuthenticationFailedError()
 
         token = WebAppAuthService.login(account=account)
@@ -234,3 +252,12 @@ class EmailCodeLoginApi(Resource):
         response = make_response({"result": "success", "data": {"access_token": token}})
         # set_access_token_to_cookie(request, response, token, samesite="None", httponly=False)
         return response
+
+
+def _log_web_login_failure(*, email: str, reason: LoginFailureReason) -> None:
+    logger.warning(
+        "Web login failed: email=%s reason=%s ip_address=%s",
+        email,
+        reason,
+        extract_remote_ip(request),
+    )

api/controllers/web/message.py

@@ -3,10 +3,10 @@ from typing import Literal
 
 from flask import request
 from graphon.model_runtime.errors.invoke import InvokeError
-from pydantic import BaseModel, Field, TypeAdapter, field_validator
+from pydantic import BaseModel, Field, TypeAdapter
 from werkzeug.exceptions import InternalServerError, NotFound
 
-from controllers.common.controller_schemas import MessageFeedbackPayload
+from controllers.common.controller_schemas import MessageFeedbackPayload, MessageListQuery
 from controllers.common.schema import register_schema_models
 from controllers.web import web_ns
 from controllers.web.error import (
@@ -25,7 +25,6 @@ from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotIni
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import SuggestedQuestionsResponse, WebMessageInfiniteScrollPagination, WebMessageListItem
 from libs import helper
-from libs.helper import uuid_value
 from models.enums import FeedbackRating
 from models.model import AppMode
 from services.app_generate_service import AppGenerateService
@@ -41,19 +40,6 @@ from services.message_service import MessageService
 logger = logging.getLogger(__name__)
 
 
-class MessageListQuery(BaseModel):
-    conversation_id: str = Field(description="Conversation UUID")
-    first_id: str | None = Field(default=None, description="First message ID for pagination")
-    limit: int = Field(default=20, ge=1, le=100, description="Number of messages to return (1-100)")
-
-    @field_validator("conversation_id", "first_id")
-    @classmethod
-    def validate_uuid(cls, value: str | None) -> str | None:
-        if value is None:
-            return value
-        return uuid_value(value)
-
-
 class MessageMoreLikeThisQuery(BaseModel):
     response_mode: Literal["blocking", "streaming"] = Field(
         description="Response mode",

api/controllers/web/passport.py

@@ -1,5 +1,6 @@
 import uuid
 from datetime import UTC, datetime, timedelta
+from typing import Any
 
 from flask import make_response, request
 from flask_restx import Resource
@@ -103,21 +104,23 @@ class PassportResource(Resource):
         return response
 
 
-def decode_enterprise_webapp_user_id(jwt_token: str | None):
+def decode_enterprise_webapp_user_id(jwt_token: str | None) -> dict[str, Any] | None:
     """
     Decode the enterprise user session from the Authorization header.
     """
     if not jwt_token:
         return None
 
-    decoded = PassportService().verify(jwt_token)
+    decoded: dict[str, Any] = PassportService().verify(jwt_token)
     source = decoded.get("token_source")
     if not source or source != "webapp_login_token":
         raise Unauthorized("Invalid token source. Expected 'webapp_login_token'.")
     return decoded
 
 
-def exchange_token_for_existing_web_user(app_code: str, enterprise_user_decoded: dict, auth_type: WebAppAuthType):
+def exchange_token_for_existing_web_user(
+    app_code: str, enterprise_user_decoded: dict[str, Any], auth_type: WebAppAuthType
+):
     """
     Exchange a token for an existing web user session.
     """

api/controllers/web/site.py

@@ -1,4 +1,4 @@
-from typing import cast
+from typing import Any, cast
 
 from flask_restx import fields, marshal, marshal_with
 from sqlalchemy import select
@@ -113,12 +113,12 @@ class AppSiteInfo:
             }
 
 
-def serialize_site(site: Site) -> dict:
+def serialize_site(site: Site) -> dict[str, Any]:
     """Serialize Site model using the same schema as AppSiteApi."""
-    return cast(dict, marshal(site, AppSiteApi.site_fields))
+    return cast(dict[str, Any], marshal(site, AppSiteApi.site_fields))
 
 
-def serialize_app_site_payload(app_model: App, site: Site, end_user_id: str | None) -> dict:
+def serialize_app_site_payload(app_model: App, site: Site, end_user_id: str | None) -> dict[str, Any]:
     can_replace_logo = FeatureService.get_features(app_model.tenant_id).can_replace_logo
     app_site_info = AppSiteInfo(app_model.tenant, app_model, site, end_user_id, can_replace_logo)
-    return cast(dict, marshal(app_site_info, AppSiteApi.app_fields))
+    return cast(dict[str, Any], marshal(app_site_info, AppSiteApi.app_fields))

api/core/agent/plugin_entities.py

@@ -84,7 +84,7 @@ class AgentStrategyEntity(BaseModel):
     identity: AgentStrategyIdentity
     parameters: list[AgentStrategyParameter] = Field(default_factory=list)
     description: I18nObject = Field(..., description="The description of the agent strategy")
-    output_schema: dict | None = None
+    output_schema: dict[str, Any] | None = None
     features: list[AgentFeature] | None = None
     meta_version: str | None = None
     # pydantic configs

api/core/app/app_config/common/sensitive_word_avoidance/manager.py

@@ -22,8 +22,8 @@ class SensitiveWordAvoidanceConfigManager:
 
     @classmethod
     def validate_and_set_defaults(
-        cls, tenant_id: str, config: dict, only_structure_validate: bool = False
-    ) -> tuple[dict, list[str]]:
+        cls, tenant_id: str, config: dict[str, Any], only_structure_validate: bool = False
+    ) -> tuple[dict[str, Any], list[str]]:
         if not config.get("sensitive_word_avoidance"):
             config["sensitive_word_avoidance"] = {"enabled": False}
 

api/core/app/app_config/easy_ui_based_app/dataset/manager.py

@@ -138,7 +138,9 @@ class DatasetConfigManager:
             )
 
     @classmethod
-    def validate_and_set_defaults(cls, tenant_id: str, app_mode: AppMode, config: dict) -> tuple[dict, list[str]]:
+    def validate_and_set_defaults(
+        cls, tenant_id: str, app_mode: AppMode, config: dict[str, Any]
+    ) -> tuple[dict[str, Any], list[str]]:
         """
         Validate and set defaults for dataset feature
 
@@ -172,7 +174,7 @@ class DatasetConfigManager:
         return config, ["agent_mode", "dataset_configs", "dataset_query_variable"]
 
     @classmethod
-    def extract_dataset_config_for_legacy_compatibility(cls, tenant_id: str, app_mode: AppMode, config: dict):
+    def extract_dataset_config_for_legacy_compatibility(cls, tenant_id: str, app_mode: AppMode, config: dict[str, Any]):
         """
         Extract dataset config for legacy compatibility
 

api/core/app/app_config/easy_ui_based_app/model_config/manager.py

@@ -41,7 +41,7 @@ class ModelConfigManager:
         )
 
     @classmethod
-    def validate_and_set_defaults(cls, tenant_id: str, config: Mapping[str, Any]) -> tuple[dict, list[str]]:
+    def validate_and_set_defaults(cls, tenant_id: str, config: Mapping[str, Any]) -> tuple[dict[str, Any], list[str]]:
         """
         Validate and set defaults for model config
 
@@ -108,7 +108,7 @@ class ModelConfigManager:
         return dict(config), ["model"]
 
     @classmethod
-    def validate_model_completion_params(cls, cp: dict):
+    def validate_model_completion_params(cls, cp: dict[str, Any]):
         # model.completion_params
         if not isinstance(cp, dict):
             raise ValueError("model.completion_params must be of object type")

api/core/app/app_config/easy_ui_based_app/prompt_template/manager.py

@@ -65,7 +65,7 @@ class PromptTemplateConfigManager:
             )
 
     @classmethod
-    def validate_and_set_defaults(cls, app_mode: AppMode, config: dict) -> tuple[dict, list[str]]:
+    def validate_and_set_defaults(cls, app_mode: AppMode, config: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
         """
         Validate pre_prompt and set defaults for prompt feature
         depending on the config['model']
@@ -130,7 +130,7 @@ class PromptTemplateConfigManager:
         return config, ["prompt_type", "pre_prompt", "chat_prompt_config", "completion_prompt_config"]
 
     @classmethod
-    def validate_post_prompt_and_set_defaults(cls, config: dict):
+    def validate_post_prompt_and_set_defaults(cls, config: dict[str, Any]):
         """
         Validate post_prompt and set defaults for prompt feature
 

api/core/app/app_config/easy_ui_based_app/variables/manager.py

@@ -1,5 +1,5 @@
 import re
-from typing import cast
+from typing import Any, cast
 
 from graphon.variables.input_entities import VariableEntity, VariableEntityType
 
@@ -82,7 +82,7 @@ class BasicVariablesConfigManager:
         return variable_entities, external_data_variables
 
     @classmethod
-    def validate_and_set_defaults(cls, tenant_id: str, config: dict) -> tuple[dict, list[str]]:
+    def validate_and_set_defaults(cls, tenant_id: str, config: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
         """
         Validate and set defaults for user input form
 
@@ -99,7 +99,7 @@ class BasicVariablesConfigManager:
         return config, related_config_keys
 
     @classmethod
-    def validate_variables_and_set_defaults(cls, config: dict) -> tuple[dict, list[str]]:
+    def validate_variables_and_set_defaults(cls, config: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
         """
         Validate and set defaults for user input form
 
@@ -164,7 +164,9 @@ class BasicVariablesConfigManager:
         return config, ["user_input_form"]
 
     @classmethod
-    def validate_external_data_tools_and_set_defaults(cls, tenant_id: str, config: dict) -> tuple[dict, list[str]]:
+    def validate_external_data_tools_and_set_defaults(
+        cls, tenant_id: str, config: dict[str, Any]
+    ) -> tuple[dict[str, Any], list[str]]:
         """
         Validate and set defaults for external data fetch feature
 

api/core/app/app_config/features/file_upload/manager.py

@@ -30,7 +30,7 @@ class FileUploadConfigManager:
                 return FileUploadConfig.model_validate(file_upload_dict)
 
     @classmethod
-    def validate_and_set_defaults(cls, config: dict) -> tuple[dict, list[str]]:
+    def validate_and_set_defaults(cls, config: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
         """
         Validate and set defaults for file upload feature
 

api/core/app/app_config/features/more_like_this/manager.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from pydantic import BaseModel, ConfigDict, Field, ValidationError
 
 
@@ -13,7 +15,7 @@ class AppConfigModel(BaseModel):
 
 class MoreLikeThisConfigManager:
     @classmethod
-    def convert(cls, config: dict) -> bool:
+    def convert(cls, config: dict[str, Any]) -> bool:
         """
         Convert model config to model config
 
@@ -23,7 +25,7 @@ class MoreLikeThisConfigManager:
         return AppConfigModel.model_validate(validated_config).more_like_this.enabled
 
     @classmethod
-    def validate_and_set_defaults(cls, config: dict) -> tuple[dict, list[str]]:
+    def validate_and_set_defaults(cls, config: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
         try:
             return AppConfigModel.model_validate(config).model_dump(), ["more_like_this"]
         except ValidationError:

api/core/app/app_config/features/opening_statement/manager.py

@@ -1,6 +1,9 @@
+from typing import Any
+
+
 class OpeningStatementConfigManager:
     @classmethod
-    def convert(cls, config: dict) -> tuple[str, list]:
+    def convert(cls, config: dict[str, Any]) -> tuple[str, list[str]]:
         """
         Convert model config to model config
 
@@ -15,7 +18,7 @@ class OpeningStatementConfigManager:
         return opening_statement, suggested_questions_list
 
     @classmethod
-    def validate_and_set_defaults(cls, config: dict) -> tuple[dict, list[str]]:
+    def validate_and_set_defaults(cls, config: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
         """
         Validate and set defaults for opening statement feature
 

api/core/app/app_config/features/retrieval_resource/manager.py

@@ -1,6 +1,9 @@
+from typing import Any
+
+
 class RetrievalResourceConfigManager:
     @classmethod
-    def convert(cls, config: dict) -> bool:
+    def convert(cls, config: dict[str, Any]) -> bool:
         show_retrieve_source = False
         retriever_resource_dict = config.get("retriever_resource")
         if retriever_resource_dict:
@@ -10,7 +13,7 @@ class RetrievalResourceConfigManager:
         return show_retrieve_source
 
     @classmethod
-    def validate_and_set_defaults(cls, config: dict) -> tuple[dict, list[str]]:
+    def validate_and_set_defaults(cls, config: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
         """
         Validate and set defaults for retriever resource feature
 

api/core/app/app_config/features/speech_to_text/manager.py

@@ -1,6 +1,9 @@
+from typing import Any
+
+
 class SpeechToTextConfigManager:
     @classmethod
-    def convert(cls, config: dict) -> bool:
+    def convert(cls, config: dict[str, Any]) -> bool:
         """
         Convert model config to model config
 
@@ -15,7 +18,7 @@ class SpeechToTextConfigManager:
         return speech_to_text
 
     @classmethod
-    def validate_and_set_defaults(cls, config: dict) -> tuple[dict, list[str]]:
+    def validate_and_set_defaults(cls, config: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
         """
         Validate and set defaults for speech to text feature
 

api/core/app/app_config/features/suggested_questions_after_answer/manager.py

@@ -1,6 +1,9 @@
+from typing import Any
+
+
 class SuggestedQuestionsAfterAnswerConfigManager:
     @classmethod
-    def convert(cls, config: dict) -> bool:
+    def convert(cls, config: dict[str, Any]) -> bool:
         """
         Convert model config to model config
 
@@ -15,7 +18,7 @@ class SuggestedQuestionsAfterAnswerConfigManager:
         return suggested_questions_after_answer
 
     @classmethod
-    def validate_and_set_defaults(cls, config: dict) -> tuple[dict, list[str]]:
+    def validate_and_set_defaults(cls, config: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
         """
         Validate and set defaults for suggested questions feature
 

api/core/app/app_config/features/text_to_speech/manager.py

@@ -1,9 +1,11 @@
+from typing import Any
+
 from core.app.app_config.entities import TextToSpeechEntity
 
 
 class TextToSpeechConfigManager:
     @classmethod
-    def convert(cls, config: dict):
+    def convert(cls, config: dict[str, Any]):
         """
         Convert model config to model config
 
@@ -22,7 +24,7 @@ class TextToSpeechConfigManager:
         return text_to_speech
 
     @classmethod
-    def validate_and_set_defaults(cls, config: dict) -> tuple[dict, list[str]]:
+    def validate_and_set_defaults(cls, config: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
         """
         Validate and set defaults for text to speech feature
 

api/core/app/apps/pipeline/generate_response_converter.py

@@ -1,5 +1,5 @@
 from collections.abc import Generator
-from typing import cast
+from typing import Any, cast
 
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
 from core.app.entities.task_entities import (
@@ -17,7 +17,7 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
     _blocking_response_type = WorkflowAppBlockingResponse
 
     @classmethod
-    def convert_blocking_full_response(cls, blocking_response: WorkflowAppBlockingResponse) -> dict:  # type: ignore[override]
+    def convert_blocking_full_response(cls, blocking_response: WorkflowAppBlockingResponse) -> dict[str, Any]:  # type: ignore[override]
         """
         Convert blocking full response.
         :param blocking_response: blocking response
@@ -26,7 +26,7 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
         return dict(blocking_response.model_dump())
 
     @classmethod
-    def convert_blocking_simple_response(cls, blocking_response: WorkflowAppBlockingResponse) -> dict:  # type: ignore[override]
+    def convert_blocking_simple_response(cls, blocking_response: WorkflowAppBlockingResponse) -> dict[str, Any]:  # type: ignore[override]
         """
         Convert blocking simple response.
         :param blocking_response: blocking response

api/core/app/apps/pipeline/pipeline_config_manager.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from core.app.app_config.base_app_config_manager import BaseAppConfigManager
 from core.app.app_config.common.sensitive_word_avoidance.manager import SensitiveWordAvoidanceConfigManager
 from core.app.app_config.entities import RagPipelineVariableEntity, WorkflowUIBasedAppConfig
@@ -34,7 +36,9 @@ class PipelineConfigManager(BaseAppConfigManager):
         return pipeline_config
 
     @classmethod
-    def config_validate(cls, tenant_id: str, config: dict, only_structure_validate: bool = False) -> dict:
+    def config_validate(
+        cls, tenant_id: str, config: dict[str, Any], only_structure_validate: bool = False
+    ) -> dict[str, Any]:
         """
         Validate for pipeline config
 

api/core/app/apps/pipeline/pipeline_generator.py

@@ -782,7 +782,7 @@ class PipelineGenerator(BaseAppGenerator):
         user_id: str,
         all_files: list,
         datasource_info: Mapping[str, Any],
-        next_page_parameters: dict | None = None,
+        next_page_parameters: dict[str, Any] | None = None,
     ):
         """
         Get files in a folder.

api/core/app/apps/workflow/app_generator.py

@@ -14,7 +14,7 @@ from graphon.runtime import GraphRuntimeState
 from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from pydantic import ValidationError
 from sqlalchemy import select
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session, sessionmaker
 
 import contexts
 from configs import dify_config
@@ -54,6 +54,25 @@ logger = logging.getLogger(__name__)
 
 
 class WorkflowAppGenerator(BaseAppGenerator):
+    @staticmethod
+    def _ensure_snippet_start_node_in_worker(*, session: Session, workflow: Workflow) -> Workflow:
+        """Re-apply snippet virtual Start injection after worker reloads workflow from DB."""
+        if workflow.type != "snippet":
+            return workflow
+
+        from models.snippet import CustomizedSnippet
+        from services.snippet_generate_service import SnippetGenerateService
+
+        snippet = session.scalar(
+            select(CustomizedSnippet).where(
+                CustomizedSnippet.id == workflow.app_id,
+                CustomizedSnippet.tenant_id == workflow.tenant_id,
+            )
+        )
+        if snippet is None:
+            return workflow
+        return SnippetGenerateService.ensure_start_node_for_worker(workflow, snippet)
+
     @staticmethod
     def _should_prepare_user_inputs(args: Mapping[str, Any]) -> bool:
         return not bool(args.get(SKIP_PREPARE_USER_INPUTS_KEY))
@@ -557,6 +576,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 if workflow is None:
                     raise ValueError("Workflow not found")
 
+                workflow = self._ensure_snippet_start_node_in_worker(session=session, workflow=workflow)
+
                 # Determine system_user_id based on invocation source
                 is_external_api_call = application_generate_entity.invoke_from in {
                     InvokeFrom.WEB_APP,

api/core/app/apps/workflow/generate_task_pipeline.py

@@ -682,15 +682,16 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
 
     def _save_workflow_app_log(self, *, session: Session, workflow_run_id: str | None):
         invoke_from = self._application_generate_entity.invoke_from
-        if invoke_from == InvokeFrom.SERVICE_API:
-            created_from = WorkflowAppLogCreatedFrom.SERVICE_API
-        elif invoke_from == InvokeFrom.EXPLORE:
-            created_from = WorkflowAppLogCreatedFrom.INSTALLED_APP
-        elif invoke_from == InvokeFrom.WEB_APP:
-            created_from = WorkflowAppLogCreatedFrom.WEB_APP
-        else:
-            # not save log for debugging
-            return
+        match invoke_from:
+            case InvokeFrom.SERVICE_API:
+                created_from = WorkflowAppLogCreatedFrom.SERVICE_API
+            case InvokeFrom.EXPLORE:
+                created_from = WorkflowAppLogCreatedFrom.INSTALLED_APP
+            case InvokeFrom.WEB_APP:
+                created_from = WorkflowAppLogCreatedFrom.WEB_APP
+            case InvokeFrom.DEBUGGER | InvokeFrom.TRIGGER | InvokeFrom.PUBLISHED_PIPELINE | InvokeFrom.VALIDATION:
+                # not save log for debugging
+                return
 
         if not workflow_run_id:
             return

api/core/app/entities/task_entities.py

@@ -521,7 +521,7 @@ class IterationNodeStartStreamResponse(StreamResponse):
         node_type: str
         title: str
         created_at: int
-        extras: dict = Field(default_factory=dict)
+        extras: dict[str, Any] = Field(default_factory=dict)
         metadata: Mapping = {}
         inputs: Mapping = {}
         inputs_truncated: bool = False
@@ -547,7 +547,7 @@ class IterationNodeNextStreamResponse(StreamResponse):
         title: str
         index: int
         created_at: int
-        extras: dict = Field(default_factory=dict)
+        extras: dict[str, Any] = Field(default_factory=dict)
 
     event: StreamEvent = StreamEvent.ITERATION_NEXT
     workflow_run_id: str
@@ -571,7 +571,7 @@ class IterationNodeCompletedStreamResponse(StreamResponse):
         outputs: Mapping | None = None
         outputs_truncated: bool = False
         created_at: int
-        extras: dict | None = None
+        extras: dict[str, Any] | None = None
         inputs: Mapping | None = None
         inputs_truncated: bool = False
         status: WorkflowNodeExecutionStatus
@@ -602,7 +602,7 @@ class LoopNodeStartStreamResponse(StreamResponse):
         node_type: str
         title: str
         created_at: int
-        extras: dict = Field(default_factory=dict)
+        extras: dict[str, Any] = Field(default_factory=dict)
         metadata: Mapping = {}
         inputs: Mapping = {}
         inputs_truncated: bool = False
@@ -653,7 +653,7 @@ class LoopNodeCompletedStreamResponse(StreamResponse):
         outputs: Mapping | None = None
         outputs_truncated: bool = False
         created_at: int
-        extras: dict | None = None
+        extras: dict[str, Any] | None = None
         inputs: Mapping | None = None
         inputs_truncated: bool = False
         status: WorkflowNodeExecutionStatus

api/core/datasource/entities/api_entities.py

@@ -14,7 +14,7 @@ class DatasourceApiEntity(BaseModel):
     description: I18nObject
     parameters: list[DatasourceParameter] | None = None
     labels: list[str] = Field(default_factory=list)
-    output_schema: dict | None = None
+    output_schema: dict[str, Any] | None = None
 
 
 ToolProviderTypeApiLiteral = Literal["builtin", "api", "workflow"] | None
@@ -30,7 +30,7 @@ class DatasourceProviderApiEntityDict(TypedDict):
     icon: str | dict
     label: I18nObjectDict
     type: str
-    team_credentials: dict | None
+    team_credentials: dict[str, Any] | None
     is_team_authorization: bool
     allow_delete: bool
     datasources: list[Any]
@@ -45,8 +45,8 @@ class DatasourceProviderApiEntity(BaseModel):
     icon: str | dict
     label: I18nObject  # label
     type: str
-    masked_credentials: dict | None = None
-    original_credentials: dict | None = None
+    masked_credentials: dict[str, Any] | None = None
+    original_credentials: dict[str, Any] | None = None
     is_team_authorization: bool = False
     allow_delete: bool = True
     plugin_id: str | None = Field(default="", description="The plugin id of the datasource")

api/core/datasource/entities/datasource_entities.py

@@ -129,7 +129,7 @@ class DatasourceEntity(BaseModel):
     identity: DatasourceIdentity
     parameters: list[DatasourceParameter] = Field(default_factory=list)
     description: I18nObject = Field(..., description="The label of the datasource")
-    output_schema: dict | None = None
+    output_schema: dict[str, Any] | None = None
 
     @field_validator("parameters", mode="before")
     @classmethod
@@ -192,7 +192,7 @@ class DatasourceInvokeMeta(BaseModel):
 
     time_cost: float = Field(..., description="The time cost of the tool invoke")
     error: str | None = None
-    tool_config: dict | None = None
+    tool_config: dict[str, Any] | None = None
 
     @classmethod
     def empty(cls) -> DatasourceInvokeMeta:
@@ -242,7 +242,7 @@ class OnlineDocumentPage(BaseModel):
 
     page_id: str = Field(..., description="The page id")
     page_name: str = Field(..., description="The page title")
-    page_icon: dict | None = Field(None, description="The page icon")
+    page_icon: dict[str, Any] | None = Field(None, description="The page icon")
     type: str = Field(..., description="The type of the page")
     last_edited_time: str = Field(..., description="The last edited time")
     parent_id: str | None = Field(None, description="The parent page id")
@@ -301,7 +301,7 @@ class GetWebsiteCrawlRequest(BaseModel):
     Get website crawl request
     """
 
-    crawl_parameters: dict = Field(..., description="The crawl parameters")
+    crawl_parameters: dict[str, Any] = Field(..., description="The crawl parameters")
 
 
 class WebSiteInfoDetail(BaseModel):
@@ -358,7 +358,7 @@ class OnlineDriveFileBucket(BaseModel):
     bucket: str | None = Field(None, description="The file bucket")
     files: list[OnlineDriveFile] = Field(..., description="The file list")
     is_truncated: bool = Field(False, description="Whether the result is truncated")
-    next_page_parameters: dict | None = Field(None, description="Parameters for fetching the next page")
+    next_page_parameters: dict[str, Any] | None = Field(None, description="Parameters for fetching the next page")
 
 
 class OnlineDriveBrowseFilesRequest(BaseModel):
@@ -369,7 +369,7 @@ class OnlineDriveBrowseFilesRequest(BaseModel):
     bucket: str | None = Field(None, description="The file bucket")
     prefix: str = Field(..., description="The parent folder ID")
     max_keys: int = Field(20, description="Page size for pagination")
-    next_page_parameters: dict | None = Field(None, description="Parameters for fetching the next page")
+    next_page_parameters: dict[str, Any] | None = Field(None, description="Parameters for fetching the next page")
 
 
 class OnlineDriveBrowseFilesResponse(BaseModel):

api/core/entities/knowledge_entities.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from pydantic import BaseModel, Field, field_validator
 
 
@@ -37,7 +39,7 @@ class PipelineDocument(BaseModel):
     id: str
     position: int
     data_source_type: str
-    data_source_info: dict | None = None
+    data_source_info: dict[str, Any] | None = None
     name: str
     indexing_status: str
     error: str | None = None

api/core/entities/provider_configuration.py

@@ -6,6 +6,7 @@ import re
 from collections import defaultdict
 from collections.abc import Iterator, Sequence
 from json import JSONDecodeError
+from typing import Any
 
 from graphon.model_runtime.entities.model_entities import AIModelEntity, FetchFrom, ModelType
 from graphon.model_runtime.entities.provider_entities import (
@@ -111,7 +112,7 @@ class ProviderConfiguration(BaseModel):
             return ModelProviderFactory(model_runtime=self._bound_model_runtime)
         return create_plugin_model_provider_factory(tenant_id=self.tenant_id)
 
-    def get_current_credentials(self, model_type: ModelType, model: str) -> dict | None:
+    def get_current_credentials(self, model_type: ModelType, model: str) -> dict[str, Any] | None:
         """
         Get current credentials.
 
@@ -233,7 +234,7 @@ class ProviderConfiguration(BaseModel):
 
         return session.execute(stmt).scalar_one_or_none()
 
-    def _get_specific_provider_credential(self, credential_id: str) -> dict | None:
+    def _get_specific_provider_credential(self, credential_id: str) -> dict[str, Any] | None:
         """
         Get a specific provider credential by ID.
         :param credential_id: Credential ID
@@ -297,7 +298,7 @@ class ProviderConfiguration(BaseModel):
             stmt = stmt.where(ProviderCredential.id != exclude_id)
         return session.execute(stmt).scalar_one_or_none() is not None
 
-    def get_provider_credential(self, credential_id: str | None = None) -> dict | None:
+    def get_provider_credential(self, credential_id: str | None = None) -> dict[str, Any] | None:
         """
         Get provider credentials.
 
@@ -317,7 +318,9 @@ class ProviderConfiguration(BaseModel):
             else [],
         )
 
-    def validate_provider_credentials(self, credentials: dict, credential_id: str = "", session: Session | None = None):
+    def validate_provider_credentials(
+        self, credentials: dict[str, Any], credential_id: str = "", session: Session | None = None
+    ):
         """
         Validate custom credentials.
         :param credentials: provider credentials
@@ -447,7 +450,7 @@ class ProviderConfiguration(BaseModel):
             provider_names.append(model_provider_id.provider_name)
         return provider_names
 
-    def create_provider_credential(self, credentials: dict, credential_name: str | None):
+    def create_provider_credential(self, credentials: dict[str, Any], credential_name: str | None):
         """
         Add custom provider credentials.
         :param credentials: provider credentials
@@ -515,7 +518,7 @@ class ProviderConfiguration(BaseModel):
 
     def update_provider_credential(
         self,
-        credentials: dict,
+        credentials: dict[str, Any],
         credential_id: str,
         credential_name: str | None,
     ):
@@ -760,7 +763,7 @@ class ProviderConfiguration(BaseModel):
 
     def _get_specific_custom_model_credential(
         self, model_type: ModelType, model: str, credential_id: str
-    ) -> dict | None:
+    ) -> dict[str, Any] | None:
         """
         Get a specific provider credential by ID.
         :param credential_id: Credential ID
@@ -832,7 +835,9 @@ class ProviderConfiguration(BaseModel):
             stmt = stmt.where(ProviderModelCredential.id != exclude_id)
         return session.execute(stmt).scalar_one_or_none() is not None
 
-    def get_custom_model_credential(self, model_type: ModelType, model: str, credential_id: str | None) -> dict | None:
+    def get_custom_model_credential(
+        self, model_type: ModelType, model: str, credential_id: str | None
+    ) -> dict[str, Any] | None:
         """
         Get custom model credentials.
 
@@ -872,7 +877,7 @@ class ProviderConfiguration(BaseModel):
         self,
         model_type: ModelType,
         model: str,
-        credentials: dict,
+        credentials: dict[str, Any],
         credential_id: str = "",
         session: Session | None = None,
     ):
@@ -939,7 +944,7 @@ class ProviderConfiguration(BaseModel):
                 return _validate(new_session)
 
     def create_custom_model_credential(
-        self, model_type: ModelType, model: str, credentials: dict, credential_name: str | None
+        self, model_type: ModelType, model: str, credentials: dict[str, Any], credential_name: str | None
     ) -> None:
         """
         Create a custom model credential.
@@ -1002,7 +1007,12 @@ class ProviderConfiguration(BaseModel):
                 raise
 
     def update_custom_model_credential(
-        self, model_type: ModelType, model: str, credentials: dict, credential_name: str | None, credential_id: str
+        self,
+        model_type: ModelType,
+        model: str,
+        credentials: dict[str, Any],
+        credential_name: str | None,
+        credential_id: str,
     ) -> None:
         """
         Update a custom model credential.
@@ -1412,7 +1422,9 @@ class ProviderConfiguration(BaseModel):
         # Get model instance of LLM
         return model_provider_factory.get_model_type_instance(provider=self.provider.provider, model_type=model_type)
 
-    def get_model_schema(self, model_type: ModelType, model: str, credentials: dict | None) -> AIModelEntity | None:
+    def get_model_schema(
+        self, model_type: ModelType, model: str, credentials: dict[str, Any] | None
+    ) -> AIModelEntity | None:
         """
         Get model schema
         """
@@ -1471,7 +1483,7 @@ class ProviderConfiguration(BaseModel):
 
         return secret_input_form_variables
 
-    def obfuscated_credentials(self, credentials: dict, credential_form_schemas: list[CredentialFormSchema]):
+    def obfuscated_credentials(self, credentials: dict[str, Any], credential_form_schemas: list[CredentialFormSchema]):
         """
         Obfuscated credentials.
 

api/core/entities/provider_entities.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from enum import StrEnum, auto
-from typing import Union
+from typing import Any, Union
 
 from graphon.model_runtime.entities.model_entities import ModelType
 from pydantic import BaseModel, ConfigDict, Field
@@ -88,7 +88,7 @@ class SystemConfiguration(BaseModel):
     enabled: bool
     current_quota_type: ProviderQuotaType | None = None
     quota_configurations: list[QuotaConfiguration] = []
-    credentials: dict | None = None
+    credentials: dict[str, Any] | None = None
 
 
 class CustomProviderConfiguration(BaseModel):
@@ -96,7 +96,7 @@ class CustomProviderConfiguration(BaseModel):
     Model class for provider custom configuration.
     """
 
-    credentials: dict
+    credentials: dict[str, Any]
     current_credential_id: str | None = None
     current_credential_name: str | None = None
     available_credentials: list[CredentialConfiguration] = []
@@ -109,7 +109,7 @@ class CustomModelConfiguration(BaseModel):
 
     model: str
     model_type: ModelType
-    credentials: dict | None
+    credentials: dict[str, Any] | None
     current_credential_id: str | None = None
     current_credential_name: str | None = None
     available_model_credentials: list[CredentialConfiguration] = []
@@ -145,7 +145,7 @@ class ModelLoadBalancingConfiguration(BaseModel):
 
     id: str
     name: str
-    credentials: dict
+    credentials: dict[str, Any]
     credential_source_type: str | None = None
     credential_id: str | None = None
 

api/core/evaluation/base_evaluation_instance.py

@@ -0,0 +1,279 @@
+import logging
+from abc import ABC, abstractmethod
+from collections.abc import Mapping
+from typing import Any
+
+from core.evaluation.entities.evaluation_entity import (
+    CustomizedMetrics,
+    EvaluationCategory,
+    EvaluationItemInput,
+    EvaluationItemResult,
+    EvaluationMetric,
+    NodeInfo,
+)
+from graphon.node_events.base import NodeRunResult
+
+logger = logging.getLogger(__name__)
+
+
+class BaseEvaluationInstance(ABC):
+    """Abstract base class for evaluation framework adapters."""
+
+    @abstractmethod
+    def evaluate_llm(
+        self,
+        items: list[EvaluationItemInput],
+        metric_names: list[str],
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        """Evaluate LLM outputs using the configured framework."""
+        ...
+
+    @abstractmethod
+    def evaluate_retrieval(
+        self,
+        items: list[EvaluationItemInput],
+        metric_names: list[str],
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        """Evaluate retrieval quality using the configured framework."""
+        ...
+
+    @abstractmethod
+    def evaluate_agent(
+        self,
+        items: list[EvaluationItemInput],
+        metric_names: list[str],
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        """Evaluate agent outputs using the configured framework."""
+        ...
+
+    @abstractmethod
+    def get_supported_metrics(self, category: EvaluationCategory) -> list[str]:
+        """Return the list of supported metric names for a given evaluation category."""
+        ...
+
+    def evaluate_with_customized_workflow(
+        self,
+        node_run_result_mapping_list: list[dict[str, NodeRunResult]],
+        customized_metrics: CustomizedMetrics,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        """Evaluate using a published workflow as the evaluator.
+
+        The evaluator workflow's output variables are treated as metrics:
+        each output variable name becomes a metric name, and its value
+        becomes the score.
+
+        Args:
+            node_run_result_mapping_list: One mapping per test-data item,
+                where each mapping is ``{node_id: NodeRunResult}`` from the
+                target execution.
+            customized_metrics: Contains ``evaluation_workflow_id`` (the
+                published evaluator workflow) and ``input_fields`` (value
+                sources for the evaluator's input variables).
+            tenant_id: Tenant scope.
+
+        Returns:
+            A list of ``EvaluationItemResult`` with metrics extracted from
+            the evaluator workflow's output variables.
+        """
+        from sqlalchemy.orm import Session
+
+        from core.app.apps.workflow.app_generator import WorkflowAppGenerator
+        from core.app.entities.app_invoke_entities import InvokeFrom
+        from core.evaluation.runners import get_service_account_for_app
+        from models.engine import db
+        from models.model import App
+        from services.workflow_service import WorkflowService
+
+        workflow_id = customized_metrics.evaluation_workflow_id
+        if not workflow_id:
+            raise ValueError("customized_metrics must contain 'evaluation_workflow_id' for customized evaluator")
+
+        # Load the evaluator workflow resources using a dedicated session
+        with Session(db.engine, expire_on_commit=False) as session, session.begin():
+            app = session.query(App).filter_by(id=workflow_id, tenant_id=tenant_id).first()
+            if not app:
+                raise ValueError(f"Evaluation workflow app {workflow_id} not found in tenant {tenant_id}")
+            service_account = get_service_account_for_app(session, workflow_id)
+
+        workflow_service = WorkflowService()
+        published_workflow = workflow_service.get_published_workflow(app_model=app)
+        if not published_workflow:
+            raise ValueError(f"No published workflow found for evaluation app {workflow_id}")
+
+        eval_results: list[EvaluationItemResult] = []
+        for idx, node_run_result_mapping in enumerate(node_run_result_mapping_list):
+            try:
+                workflow_inputs = self._build_workflow_inputs(
+                    customized_metrics.input_fields,
+                    node_run_result_mapping,
+                )
+
+                generator = WorkflowAppGenerator()
+                response: Mapping[str, Any] = generator.generate(
+                    app_model=app,
+                    workflow=published_workflow,
+                    user=service_account,
+                    args={"inputs": workflow_inputs},
+                    invoke_from=InvokeFrom.SERVICE_API,
+                    streaming=False,
+                    call_depth=0,
+                )
+
+                metrics = self._extract_workflow_metrics(response, workflow_id)
+                eval_results.append(
+                    EvaluationItemResult(
+                        index=idx,
+                        metrics=metrics,
+                    )
+                )
+            except Exception:
+                logger.exception(
+                    "Customized evaluator failed for item %d with workflow %s",
+                    idx,
+                    workflow_id,
+                )
+                eval_results.append(EvaluationItemResult(index=idx))
+
+        return eval_results
+
+    @staticmethod
+    def _build_workflow_inputs(
+        input_fields: dict[str, Any],
+        node_run_result_mapping: dict[str, NodeRunResult],
+    ) -> dict[str, Any]:
+        """Build customized workflow inputs by resolving value sources.
+
+        Each entry in ``input_fields`` maps a workflow input variable name
+        to its value source, which can be:
+
+          - **Constant**: a plain string without ``{{#…#}}`` used as-is.
+          - **Expression**: a string containing one or more
+            ``{{#node_id.output_key#}}`` selectors (same format as
+            ``VariableTemplateParser``) resolved from
+            ``node_run_result_mapping``.
+
+        """
+        from graphon.nodes.base.variable_template_parser import REGEX as VARIABLE_REGEX
+
+        workflow_inputs: dict[str, Any] = {}
+
+        for field_name, value_source in input_fields.items():
+            if not isinstance(value_source, str):
+                # Non-string values (numbers, bools, dicts) are used directly.
+                workflow_inputs[field_name] = value_source
+                continue
+
+            # Check if the entire value is a single expression.
+            full_match = VARIABLE_REGEX.fullmatch(value_source)
+            if full_match:
+                workflow_inputs[field_name] = resolve_variable_selector(
+                    full_match.group(1),
+                    node_run_result_mapping,
+                )
+            elif VARIABLE_REGEX.search(value_source):
+                # Mixed template: interpolate all expressions as strings.
+                workflow_inputs[field_name] = VARIABLE_REGEX.sub(
+                    lambda m: str(resolve_variable_selector(m.group(1), node_run_result_mapping)),
+                    value_source,
+                )
+            else:
+                # Plain constant — no expression markers.
+                workflow_inputs[field_name] = value_source
+
+        return workflow_inputs
+
+    @staticmethod
+    def _extract_workflow_metrics(
+        response: Mapping[str, object],
+        evaluation_workflow_id: str,
+    ) -> list[EvaluationMetric]:
+        """Extract evaluation metrics from workflow output variables.
+
+        Each metric's ``node_info`` is set with *evaluation_workflow_id* as
+        the ``node_id``, so that judgment conditions can reference customized
+        metrics via ``variable_selector: [evaluation_workflow_id, metric_name]``.
+        """
+        metrics: list[EvaluationMetric] = []
+        node_info = NodeInfo(node_id=evaluation_workflow_id, type="customized", title="customized")
+
+        data = response.get("data")
+        if not isinstance(data, Mapping):
+            logger.warning("Unexpected workflow response format: missing 'data' dict")
+            return metrics
+
+        outputs = data.get("outputs")
+        if not isinstance(outputs, dict):
+            logger.warning("Unexpected workflow response format: 'outputs' is not a dict")
+            return metrics
+
+        for key, raw_value in outputs.items():
+            if not isinstance(key, str):
+                continue
+            metrics.append(EvaluationMetric(name=key, value=raw_value, node_info=node_info))
+
+        return metrics
+
+
+def resolve_variable_selector(
+    selector_raw: str,
+    node_run_result_mapping: dict[str, NodeRunResult],
+) -> object:
+    """
+    Resolve a ``#node_id.output_key#`` selector against node run results.
+    """
+    #
+    cleaned = selector_raw.strip("#")
+    parts = cleaned.split(".")
+
+    if len(parts) < 2:
+        logger.warning(
+            "Selector '%s' must have at least node_id.output_key",
+            selector_raw,
+        )
+        return ""
+
+    node_id = parts[0]
+    output_path = parts[1:]
+
+    node_result = node_run_result_mapping.get(node_id)
+    if not node_result or not node_result.outputs:
+        logger.warning(
+            "Selector '%s': node '%s' not found or has no outputs",
+            selector_raw,
+            node_id,
+        )
+        return ""
+
+    # Traverse the output path to support nested keys.
+    current: object = node_result.outputs
+    for key in output_path:
+        if isinstance(current, Mapping):
+            next_val = current.get(key)
+            if next_val is None:
+                logger.warning(
+                    "Selector '%s': key '%s' not found in node '%s' outputs",
+                    selector_raw,
+                    key,
+                    node_id,
+                )
+                return ""
+            current = next_val
+        else:
+            logger.warning(
+                "Selector '%s': cannot traverse into non-dict value at key '%s'",
+                selector_raw,
+                key,
+            )
+            return ""
+
+    return current if current is not None else ""

api/core/evaluation/entities/config_entity.py

@@ -0,0 +1,27 @@
+from enum import StrEnum
+
+from pydantic import BaseModel
+
+
+class EvaluationFrameworkEnum(StrEnum):
+    RAGAS = "ragas"
+    DEEPEVAL = "deepeval"
+    NONE = "none"
+
+
+class BaseEvaluationConfig(BaseModel):
+    """Base configuration for evaluation frameworks."""
+
+    pass
+
+
+class RagasConfig(BaseEvaluationConfig):
+    """RAGAS-specific configuration."""
+
+    pass
+
+
+class DeepEvalConfig(BaseEvaluationConfig):
+    """DeepEval-specific configuration."""
+
+    pass

api/core/evaluation/entities/evaluation_entity.py

@@ -0,0 +1,226 @@
+from enum import StrEnum
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+from core.evaluation.entities.judgment_entity import JudgmentConfig, JudgmentResult
+
+
+class EvaluationCategory(StrEnum):
+    LLM = "llm"
+    RETRIEVAL = "knowledge_retrieval"
+    AGENT = "agent"
+    WORKFLOW = "workflow"
+    SNIPPET = "snippet"
+    KNOWLEDGE_BASE = "knowledge_base"
+
+
+class EvaluationMetricName(StrEnum):
+    """Canonical metric names shared across all evaluation frameworks.
+
+    Each framework maps these names to its own internal implementation.
+    A framework that does not support a given metric should log a warning
+    and skip it rather than raising an error.
+
+    ── LLM / general text-quality metrics ──────────────────────────────────
+    FAITHFULNESS
+        Measures whether every claim in the model's response is grounded in
+        the provided retrieved context. A high score means the answer
+        contains no hallucinated content — each statement can be traced back
+        to a passage in the context.
+        Required fields: user_input, response, retrieved_contexts.
+
+    ANSWER_RELEVANCY
+        Measures how well the model's response addresses the user's question.
+        A high score means the answer stays on-topic; a low score indicates
+        irrelevant content or a failure to answer the actual question.
+        Required fields: user_input, response.
+
+    ANSWER_CORRECTNESS
+        Measures the factual accuracy and completeness of the model's answer
+        relative to a ground-truth reference. It combines semantic similarity
+        with key-fact coverage, so both meaning and content matter.
+        Required fields: user_input, response, reference (expected_output).
+
+    SEMANTIC_SIMILARITY
+        Measures the cosine similarity between the model's response and the
+        reference answer in an embedding space. It evaluates whether the two
+        texts convey the same meaning, independent of factual correctness.
+        Required fields: response, reference (expected_output).
+
+    ── Retrieval-quality metrics ────────────────────────────────────────────
+    CONTEXT_PRECISION
+        Measures the proportion of retrieved context chunks that are actually
+        relevant to the question (precision). A high score means the retrieval
+        pipeline returns little noise.
+        Required fields: user_input, reference, retrieved_contexts.
+
+    CONTEXT_RECALL
+        Measures the proportion of ground-truth information that is covered by
+        the retrieved context chunks (recall). A high score means the retrieval
+        pipeline does not miss important supporting evidence.
+        Required fields: user_input, reference, retrieved_contexts.
+
+    CONTEXT_RELEVANCE
+        Measures how relevant each individual retrieved chunk is to the query.
+        Similar to CONTEXT_PRECISION but evaluated at the chunk level rather
+        than against a reference answer.
+        Required fields: user_input, retrieved_contexts.
+
+    ── Agent-quality metrics ────────────────────────────────────────────────
+    TOOL_CORRECTNESS
+        Measures the correctness of the tool calls made by the agent during
+        task execution — both the choice of tool and the arguments passed.
+        A high score means the agent's tool-use strategy matches the expected
+        behavior.
+        Required fields: actual tool calls vs. expected tool calls.
+
+    TASK_COMPLETION
+        Measures whether the agent ultimately achieves the user's stated goal.
+        It evaluates the reasoning chain, intermediate steps, and final output
+        holistically; a high score means the task was fully accomplished.
+        Required fields: user_input, actual_output.
+    """
+
+    # LLM / general text-quality metrics
+    FAITHFULNESS = "faithfulness"
+    ANSWER_RELEVANCY = "answer_relevancy"
+    ANSWER_CORRECTNESS = "answer_correctness"
+    SEMANTIC_SIMILARITY = "semantic_similarity"
+
+    # Retrieval-quality metrics
+    CONTEXT_PRECISION = "context_precision"
+    CONTEXT_RECALL = "context_recall"
+    CONTEXT_RELEVANCE = "context_relevance"
+
+    # Agent-quality metrics
+    TOOL_CORRECTNESS = "tool_correctness"
+    TASK_COMPLETION = "task_completion"
+
+
+# Per-category canonical metric lists used by get_supported_metrics().
+LLM_METRIC_NAMES: list[EvaluationMetricName] = [
+    EvaluationMetricName.FAITHFULNESS,        # Every claim is grounded in context; no hallucinations
+    EvaluationMetricName.ANSWER_RELEVANCY,    # Response stays on-topic and addresses the question
+    EvaluationMetricName.ANSWER_CORRECTNESS,  # Factual accuracy and completeness vs. reference
+    EvaluationMetricName.SEMANTIC_SIMILARITY,  # Semantic closeness to the reference answer
+]
+
+RETRIEVAL_METRIC_NAMES: list[EvaluationMetricName] = [
+    EvaluationMetricName.CONTEXT_PRECISION,  # Fraction of retrieved chunks that are relevant (precision)
+    EvaluationMetricName.CONTEXT_RECALL,     # Fraction of ground-truth info covered by retrieval (recall)
+    EvaluationMetricName.CONTEXT_RELEVANCE,  # Per-chunk relevance to the query
+]
+
+AGENT_METRIC_NAMES: list[EvaluationMetricName] = [
+    EvaluationMetricName.TOOL_CORRECTNESS,  # Correct tool selection and arguments
+    EvaluationMetricName.TASK_COMPLETION,   # Whether the agent fully achieves the user's goal
+]
+
+WORKFLOW_METRIC_NAMES: list[EvaluationMetricName] = [
+    EvaluationMetricName.FAITHFULNESS,
+    EvaluationMetricName.ANSWER_RELEVANCY,
+    EvaluationMetricName.ANSWER_CORRECTNESS,
+]
+
+METRIC_NODE_TYPE_MAPPING: dict[str, str] = {
+    **{m.value: "llm" for m in LLM_METRIC_NAMES},
+    **{m.value: "knowledge-retrieval" for m in RETRIEVAL_METRIC_NAMES},
+    **{m.value: "agent" for m in AGENT_METRIC_NAMES},
+}
+
+METRIC_VALUE_TYPE_MAPPING: dict[str, str] = {
+    EvaluationMetricName.FAITHFULNESS: "number",
+    EvaluationMetricName.ANSWER_RELEVANCY: "number",
+    EvaluationMetricName.ANSWER_CORRECTNESS: "number",
+    EvaluationMetricName.SEMANTIC_SIMILARITY: "number",
+    EvaluationMetricName.CONTEXT_PRECISION: "number",
+    EvaluationMetricName.CONTEXT_RECALL: "number",
+    EvaluationMetricName.CONTEXT_RELEVANCE: "number",
+    EvaluationMetricName.TOOL_CORRECTNESS: "number",
+    EvaluationMetricName.TASK_COMPLETION: "number",
+}
+
+
+class NodeInfo(BaseModel):
+    node_id: str
+    type: str
+    title: str
+
+
+class EvaluationMetric(BaseModel):
+    name: str
+    value: Any
+    details: dict[str, Any] = Field(default_factory=dict)
+    node_info: NodeInfo | None = None
+
+
+class EvaluationItemInput(BaseModel):
+    index: int
+    inputs: dict[str, Any]
+    output: str
+    expected_output: str | None = None
+    context: list[str] | None = None
+
+
+class EvaluationDatasetInput(BaseModel):
+    index: int
+    inputs: dict[str, Any]
+    expected_output: str | None = None
+
+
+class EvaluationItemResult(BaseModel):
+    index: int
+    actual_output: str | None = None
+    metrics: list[EvaluationMetric] = Field(default_factory=list)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+    judgment: JudgmentResult = Field(default_factory=JudgmentResult)
+    error: str | None = None
+
+
+class DefaultMetric(BaseModel):
+    metric: str
+    value_type: str = ""
+    node_info_list: list[NodeInfo]
+
+
+class CustomizedMetricOutputField(BaseModel):
+    variable: str
+    value_type: str
+
+
+class CustomizedMetrics(BaseModel):
+    evaluation_workflow_id: str
+    input_fields: dict[str, Any]
+    output_fields: list[CustomizedMetricOutputField]
+
+
+class EvaluationConfigData(BaseModel):
+    """Structured data for saving evaluation configuration."""
+
+    evaluation_model: str = ""
+    evaluation_model_provider: str = ""
+    default_metrics: list[DefaultMetric] = Field(default_factory=list)
+    customized_metrics: CustomizedMetrics | None = None
+    judgment_config: JudgmentConfig | None = None
+
+
+class EvaluationRunRequest(EvaluationConfigData):
+    """Request body for starting an evaluation run."""
+
+    file_id: str
+
+
+class EvaluationRunData(BaseModel):
+    """Serializable data for Celery task."""
+
+    evaluation_run_id: str
+    tenant_id: str
+    target_type: str
+    target_id: str
+    evaluation_model_provider: str
+    evaluation_model: str
+    default_metrics: list[DefaultMetric] = Field(default_factory=list)
+    customized_metrics: CustomizedMetrics | None = None
+    judgment_config: JudgmentConfig | None = None
+    input_list: list[EvaluationDatasetInput]

api/core/evaluation/entities/judgment_entity.py

@@ -0,0 +1,96 @@
+"""Judgment condition entities for evaluation metric assessment.
+
+Condition structure mirrors the workflow if-else ``Condition`` model from
+``graphon.utils.condition.entities``.  The left-hand side uses
+``variable_selector`` — a two-element list ``[node_id, metric_name]`` — to
+uniquely identify an evaluation metric (different nodes may produce metrics
+with the same name).
+
+Operators reuse ``SupportedComparisonOperator`` from the workflow engine so
+that type semantics stay consistent across the platform.
+
+Typical usage::
+
+    judgment_config = JudgmentConfig(
+        logical_operator="and",
+        conditions=[
+            JudgmentCondition(
+                variable_selector=["node_abc", "faithfulness"],
+                comparison_operator=">",
+                value="0.8",
+            )
+        ],
+    )
+"""
+
+from collections.abc import Sequence
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+from graphon.utils.condition.entities import SupportedComparisonOperator
+
+
+class JudgmentCondition(BaseModel):
+    """A single judgment condition that checks one metric value.
+
+    Mirrors ``graphon.utils.condition.entities.Condition`` with the left-hand
+    side being a metric selector instead of a workflow variable selector.
+
+    Attributes:
+        variable_selector: ``[node_id, metric_name]`` identifying the metric.
+        comparison_operator: Reuses workflow's ``SupportedComparisonOperator``.
+        value: The comparison target (right side).  For unary operators such
+            as ``empty`` or ``null`` this can be ``None``.
+    """
+
+    variable_selector: list[str]
+    comparison_operator: SupportedComparisonOperator
+    value: str | Sequence[str] | bool | None = None
+
+
+class JudgmentConfig(BaseModel):
+    """A group of judgment conditions combined with a logical operator.
+
+    Attributes:
+        logical_operator: How to combine condition results — "and" requires
+            all conditions to pass, "or" requires at least one.
+        conditions: The list of individual conditions to evaluate.
+    """
+
+    logical_operator: Literal["and", "or"] = "and"
+    conditions: list[JudgmentCondition] = Field(default_factory=list)
+
+
+class JudgmentConditionResult(BaseModel):
+    """Result of evaluating a single judgment condition.
+
+    Attributes:
+        variable_selector: ``[node_id, metric_name]`` that was checked.
+        comparison_operator: The operator that was applied.
+        expected_value: The resolved comparison value.
+        actual_value: The actual metric value that was evaluated.
+        passed: Whether this individual condition passed.
+        error: Error message if the condition evaluation failed.
+    """
+
+    variable_selector: list[str]
+    comparison_operator: str
+    expected_value: Any = None
+    actual_value: Any = None
+    passed: bool = False
+    error: str | None = None
+
+
+class JudgmentResult(BaseModel):
+    """Overall result of evaluating all judgment conditions for one item.
+
+    Attributes:
+        passed: Whether the overall judgment passed (based on logical_operator).
+        logical_operator: The logical operator used to combine conditions.
+        condition_results: Detailed result for each individual condition.
+    """
+
+    passed: bool = False
+    logical_operator: Literal["and", "or"] = "and"
+    condition_results: list[JudgmentConditionResult] = Field(default_factory=list)

api/core/evaluation/evaluation_manager.py

@@ -0,0 +1,61 @@
+import collections
+import logging
+from typing import Any
+
+from configs import dify_config
+from core.evaluation.base_evaluation_instance import BaseEvaluationInstance
+from core.evaluation.entities.config_entity import EvaluationFrameworkEnum
+from core.evaluation.entities.evaluation_entity import EvaluationCategory
+
+logger = logging.getLogger(__name__)
+
+
+class EvaluationFrameworkConfigMap(collections.UserDict[str, dict[str, Any]]):
+    """Registry mapping framework enum -> {config_class, evaluator_class}."""
+
+    def __getitem__(self, framework: str) -> dict[str, Any]:
+        match framework:
+            case EvaluationFrameworkEnum.RAGAS:
+                from core.evaluation.entities.config_entity import RagasConfig
+                from core.evaluation.frameworks.ragas.ragas_evaluator import RagasEvaluator
+
+                return {
+                    "config_class": RagasConfig,
+                    "evaluator_class": RagasEvaluator,
+                }
+            case EvaluationFrameworkEnum.DEEPEVAL:
+                raise NotImplementedError("DeepEval adapter is not yet implemented.")
+            case _:
+                raise ValueError(f"Unknown evaluation framework: {framework}")
+
+
+evaluation_framework_config_map = EvaluationFrameworkConfigMap()
+
+
+class EvaluationManager:
+    """Factory for evaluation instances based on global configuration."""
+
+    @staticmethod
+    def get_evaluation_instance() -> BaseEvaluationInstance | None:
+        """Create and return an evaluation instance based on EVALUATION_FRAMEWORK env var."""
+        framework = dify_config.EVALUATION_FRAMEWORK
+        if not framework or framework == EvaluationFrameworkEnum.NONE:
+            return None
+
+        try:
+            config_map = evaluation_framework_config_map[framework]
+            evaluator_class = config_map["evaluator_class"]
+            config_class = config_map["config_class"]
+            config = config_class()
+            return evaluator_class(config)
+        except Exception:
+            logger.exception("Failed to create evaluation instance for framework: %s", framework)
+            return None
+
+    @staticmethod
+    def get_supported_metrics(category: EvaluationCategory) -> list[str]:
+        """Return supported metrics for the current framework and given category."""
+        instance = EvaluationManager.get_evaluation_instance()
+        if instance is None:
+            return []
+        return instance.get_supported_metrics(category)

api/core/evaluation/frameworks/deepeval/__init__.py

@@ -0,0 +1 @@
+

api/core/evaluation/frameworks/deepeval/deepeval_evaluator.py

@@ -0,0 +1,299 @@
+import logging
+from typing import Any
+
+from core.evaluation.base_evaluation_instance import BaseEvaluationInstance
+from core.evaluation.entities.config_entity import DeepEvalConfig
+from core.evaluation.entities.evaluation_entity import (
+    AGENT_METRIC_NAMES,
+    LLM_METRIC_NAMES,
+    RETRIEVAL_METRIC_NAMES,
+    WORKFLOW_METRIC_NAMES,
+    EvaluationCategory,
+    EvaluationItemInput,
+    EvaluationItemResult,
+    EvaluationMetric,
+    EvaluationMetricName,
+)
+from core.evaluation.frameworks.ragas.ragas_model_wrapper import DifyModelWrapper
+
+logger = logging.getLogger(__name__)
+
+# Maps canonical EvaluationMetricName to the corresponding deepeval metric class name.
+# deepeval metric field requirements (LLMTestCase fields):
+#   - faithfulness:       input, actual_output, retrieval_context
+#   - answer_relevancy:   input, actual_output
+#   - context_precision:  input, actual_output, expected_output, retrieval_context
+#   - context_recall:     input, actual_output, expected_output, retrieval_context
+#   - context_relevance:  input, actual_output, retrieval_context
+#   - tool_correctness:   input, actual_output, expected_tools
+#   - task_completion:    input, actual_output
+# Metrics not listed here are unsupported by deepeval and will be skipped.
+_DEEPEVAL_METRIC_MAP: dict[EvaluationMetricName, str] = {
+    EvaluationMetricName.FAITHFULNESS: "FaithfulnessMetric",
+    EvaluationMetricName.ANSWER_RELEVANCY: "AnswerRelevancyMetric",
+    EvaluationMetricName.CONTEXT_PRECISION: "ContextualPrecisionMetric",
+    EvaluationMetricName.CONTEXT_RECALL: "ContextualRecallMetric",
+    EvaluationMetricName.CONTEXT_RELEVANCE: "ContextualRelevancyMetric",
+    EvaluationMetricName.TOOL_CORRECTNESS: "ToolCorrectnessMetric",
+    EvaluationMetricName.TASK_COMPLETION: "TaskCompletionMetric",
+}
+
+
+class DeepEvalEvaluator(BaseEvaluationInstance):
+    """DeepEval framework adapter for evaluation."""
+
+    def __init__(self, config: DeepEvalConfig):
+        self.config = config
+
+    def get_supported_metrics(self, category: EvaluationCategory) -> list[str]:
+        match category:
+            case EvaluationCategory.LLM:
+                candidates = LLM_METRIC_NAMES
+            case EvaluationCategory.RETRIEVAL:
+                candidates = RETRIEVAL_METRIC_NAMES
+            case EvaluationCategory.AGENT:
+                candidates = AGENT_METRIC_NAMES
+            case EvaluationCategory.WORKFLOW | EvaluationCategory.SNIPPET:
+                candidates = WORKFLOW_METRIC_NAMES
+            case _:
+                return []
+        return [m for m in candidates if m in _DEEPEVAL_METRIC_MAP]
+
+    def evaluate_llm(
+        self,
+        items: list[EvaluationItemInput],
+        metric_names: list[str],
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        return self._evaluate(items, metric_names, model_provider, model_name, tenant_id, EvaluationCategory.LLM)
+
+    def evaluate_retrieval(
+        self,
+        items: list[EvaluationItemInput],
+        metric_names: list[str],
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        return self._evaluate(items, metric_names, model_provider, model_name, tenant_id, EvaluationCategory.RETRIEVAL)
+
+    def evaluate_agent(
+        self,
+        items: list[EvaluationItemInput],
+        metric_names: list[str],
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        return self._evaluate(items, metric_names, model_provider, model_name, tenant_id, EvaluationCategory.AGENT)
+
+    def evaluate_workflow(
+        self,
+        items: list[EvaluationItemInput],
+        metric_names: list[str],
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        return self._evaluate(items, metric_names, model_provider, model_name, tenant_id, EvaluationCategory.WORKFLOW)
+
+    def _evaluate(
+        self,
+        items: list[EvaluationItemInput],
+        metric_names: list[str],
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+        category: EvaluationCategory,
+    ) -> list[EvaluationItemResult]:
+        """Core evaluation logic using DeepEval."""
+        model_wrapper = DifyModelWrapper(model_provider, model_name, tenant_id)
+        requested_metrics = metric_names or self.get_supported_metrics(category)
+
+        try:
+            return self._evaluate_with_deepeval(items, requested_metrics, category)
+        except ImportError:
+            logger.warning("DeepEval not installed, falling back to simple evaluation")
+            return self._evaluate_simple(items, requested_metrics, model_wrapper)
+
+    def _evaluate_with_deepeval(
+        self,
+        items: list[EvaluationItemInput],
+        requested_metrics: list[str],
+        category: EvaluationCategory,
+    ) -> list[EvaluationItemResult]:
+        """Evaluate using DeepEval library.
+
+        Builds LLMTestCase differently per category:
+        - LLM/Workflow: input=prompt, actual_output=output, retrieval_context=context
+        - Retrieval: input=query, actual_output=output, expected_output, retrieval_context=context
+        - Agent: input=query, actual_output=output
+        """
+        metric_pairs = _build_deepeval_metrics(requested_metrics)
+        if not metric_pairs:
+            logger.warning("No valid DeepEval metrics found for: %s", requested_metrics)
+            return [EvaluationItemResult(index=item.index) for item in items]
+
+        results: list[EvaluationItemResult] = []
+        for item in items:
+            test_case = self._build_test_case(item, category)
+            metrics: list[EvaluationMetric] = []
+            for canonical_name, metric in metric_pairs:
+                try:
+                    metric.measure(test_case)
+                    if metric.score is not None:
+                        metrics.append(EvaluationMetric(name=canonical_name, value=float(metric.score)))
+                except Exception:
+                    logger.exception(
+                        "Failed to compute metric %s for item %d",
+                        canonical_name,
+                        item.index,
+                    )
+            results.append(EvaluationItemResult(index=item.index, metrics=metrics))
+        return results
+
+    @staticmethod
+    def _build_test_case(item: EvaluationItemInput, category: EvaluationCategory) -> Any:
+        """Build a deepeval LLMTestCase with the correct fields per category."""
+        from deepeval.test_case import LLMTestCase
+
+        user_input = _format_input(item.inputs, category)
+
+        match category:
+            case EvaluationCategory.LLM | EvaluationCategory.WORKFLOW:
+                # faithfulness needs: input, actual_output, retrieval_context
+                # answer_relevancy needs: input, actual_output
+                return LLMTestCase(
+                    input=user_input,
+                    actual_output=item.output,
+                    expected_output=item.expected_output or None,
+                    retrieval_context=item.context or None,
+                )
+            case EvaluationCategory.RETRIEVAL:
+                # contextual_precision/recall needs: input, actual_output, expected_output, retrieval_context
+                return LLMTestCase(
+                    input=user_input,
+                    actual_output=item.output or "",
+                    expected_output=item.expected_output or "",
+                    retrieval_context=item.context or [],
+                )
+            case _:
+                return LLMTestCase(
+                    input=user_input,
+                    actual_output=item.output,
+                )
+
+    def _evaluate_simple(
+        self,
+        items: list[EvaluationItemInput],
+        requested_metrics: list[str],
+        model_wrapper: DifyModelWrapper,
+    ) -> list[EvaluationItemResult]:
+        """Simple LLM-as-judge fallback when DeepEval is not available."""
+        results: list[EvaluationItemResult] = []
+        for item in items:
+            metrics: list[EvaluationMetric] = []
+            for m_name in requested_metrics:
+                try:
+                    score = self._judge_with_llm(model_wrapper, m_name, item)
+                    metrics.append(EvaluationMetric(name=m_name, value=score))
+                except Exception:
+                    logger.exception("Failed to compute metric %s for item %d", m_name, item.index)
+            results.append(EvaluationItemResult(index=item.index, metrics=metrics))
+        return results
+
+    def _judge_with_llm(
+        self,
+        model_wrapper: DifyModelWrapper,
+        metric_name: str,
+        item: EvaluationItemInput,
+    ) -> float:
+        """Use the LLM to judge a single metric for a single item."""
+        prompt = self._build_judge_prompt(metric_name, item)
+        response = model_wrapper.invoke(prompt)
+        return self._parse_score(response)
+
+    @staticmethod
+    def _build_judge_prompt(metric_name: str, item: EvaluationItemInput) -> str:
+        """Build a scoring prompt for the LLM judge."""
+        parts = [
+            f"Evaluate the following on the metric '{metric_name}' using a scale of 0.0 to 1.0.",
+            f"\nInput: {item.inputs}",
+            f"\nOutput: {item.output}",
+        ]
+        if item.expected_output:
+            parts.append(f"\nExpected Output: {item.expected_output}")
+        if item.context:
+            parts.append(f"\nContext: {'; '.join(item.context)}")
+        parts.append("\nRespond with ONLY a single floating point number between 0.0 and 1.0, nothing else.")
+        return "\n".join(parts)
+
+    @staticmethod
+    def _parse_score(response: str) -> float:
+        """Parse a float score from LLM response."""
+        import re
+
+        cleaned = response.strip()
+        try:
+            score = float(cleaned)
+            return max(0.0, min(1.0, score))
+        except ValueError:
+            match = re.search(r"(\d+\.?\d*)", cleaned)
+            if match:
+                score = float(match.group(1))
+                return max(0.0, min(1.0, score))
+            return 0.0
+
+
+def _format_input(inputs: dict[str, Any], category: EvaluationCategory) -> str:
+    """Extract the user-facing input string from the inputs dict."""
+    match category:
+        case EvaluationCategory.LLM | EvaluationCategory.WORKFLOW:
+            return str(inputs.get("prompt", ""))
+        case EvaluationCategory.RETRIEVAL:
+            return str(inputs.get("query", ""))
+        case _:
+            return str(next(iter(inputs.values()), "")) if inputs else ""
+
+
+def _build_deepeval_metrics(requested_metrics: list[str]) -> list[tuple[str, Any]]:
+    """Build DeepEval metric instances from canonical metric names.
+
+    Returns a list of (canonical_name, metric_instance) pairs so that callers
+    can record the canonical name rather than the framework-internal class name.
+    """
+    try:
+        from deepeval.metrics import (
+            AnswerRelevancyMetric,
+            ContextualPrecisionMetric,
+            ContextualRecallMetric,
+            ContextualRelevancyMetric,
+            FaithfulnessMetric,
+            TaskCompletionMetric,
+            ToolCorrectnessMetric,
+        )
+
+        # Maps canonical name → deepeval metric class
+        deepeval_class_map: dict[str, Any] = {
+            EvaluationMetricName.FAITHFULNESS: FaithfulnessMetric,
+            EvaluationMetricName.ANSWER_RELEVANCY: AnswerRelevancyMetric,
+            EvaluationMetricName.CONTEXT_PRECISION: ContextualPrecisionMetric,
+            EvaluationMetricName.CONTEXT_RECALL: ContextualRecallMetric,
+            EvaluationMetricName.CONTEXT_RELEVANCE: ContextualRelevancyMetric,
+            EvaluationMetricName.TOOL_CORRECTNESS: ToolCorrectnessMetric,
+            EvaluationMetricName.TASK_COMPLETION: TaskCompletionMetric,
+        }
+
+        pairs: list[tuple[str, Any]] = []
+        for name in requested_metrics:
+            metric_class = deepeval_class_map.get(name)
+            if metric_class:
+                pairs.append((name, metric_class(threshold=0.5)))
+            else:
+                logger.warning("Metric '%s' is not supported by DeepEval, skipping", name)
+        return pairs
+    except ImportError:
+        logger.warning("DeepEval metrics not available")
+        return []

api/core/evaluation/frameworks/ragas/ragas_evaluator.py

@@ -0,0 +1,312 @@
+import logging
+from typing import Any
+
+from core.evaluation.base_evaluation_instance import BaseEvaluationInstance
+from core.evaluation.entities.config_entity import RagasConfig
+from core.evaluation.entities.evaluation_entity import (
+    AGENT_METRIC_NAMES,
+    LLM_METRIC_NAMES,
+    RETRIEVAL_METRIC_NAMES,
+    WORKFLOW_METRIC_NAMES,
+    EvaluationCategory,
+    EvaluationItemInput,
+    EvaluationItemResult,
+    EvaluationMetric,
+    EvaluationMetricName,
+)
+from core.evaluation.frameworks.ragas.ragas_model_wrapper import DifyModelWrapper
+
+logger = logging.getLogger(__name__)
+
+# Maps canonical EvaluationMetricName to the corresponding ragas metric class.
+# Metrics not listed here are unsupported by ragas and will be skipped.
+_RAGAS_METRIC_MAP: dict[EvaluationMetricName, str] = {
+    EvaluationMetricName.FAITHFULNESS: "Faithfulness",
+    EvaluationMetricName.ANSWER_RELEVANCY: "AnswerRelevancy",
+    EvaluationMetricName.ANSWER_CORRECTNESS: "AnswerCorrectness",
+    EvaluationMetricName.SEMANTIC_SIMILARITY: "SemanticSimilarity",
+    EvaluationMetricName.CONTEXT_PRECISION: "ContextPrecision",
+    EvaluationMetricName.CONTEXT_RECALL: "ContextRecall",
+    EvaluationMetricName.CONTEXT_RELEVANCE: "ContextRelevance",
+    EvaluationMetricName.TOOL_CORRECTNESS: "ToolCallAccuracy",
+}
+
+
+class RagasEvaluator(BaseEvaluationInstance):
+    """RAGAS framework adapter for evaluation."""
+
+    def __init__(self, config: RagasConfig):
+        self.config = config
+
+    def get_supported_metrics(self, category: EvaluationCategory) -> list[str]:
+        match category:
+            case EvaluationCategory.LLM:
+                candidates = LLM_METRIC_NAMES
+            case EvaluationCategory.RETRIEVAL:
+                candidates = RETRIEVAL_METRIC_NAMES
+            case EvaluationCategory.AGENT:
+                candidates = AGENT_METRIC_NAMES
+            case EvaluationCategory.WORKFLOW | EvaluationCategory.SNIPPET:
+                candidates = WORKFLOW_METRIC_NAMES
+            case _:
+                return []
+        return [m for m in candidates if m in _RAGAS_METRIC_MAP]
+
+    def evaluate_llm(
+        self,
+        items: list[EvaluationItemInput],
+        metric_names: list[str],
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        return self._evaluate(items, metric_names, model_provider, model_name, tenant_id, EvaluationCategory.LLM)
+
+    def evaluate_retrieval(
+        self,
+        items: list[EvaluationItemInput],
+        metric_names: list[str],
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        return self._evaluate(items, metric_names, model_provider, model_name, tenant_id, EvaluationCategory.RETRIEVAL)
+
+    def evaluate_agent(
+        self,
+        items: list[EvaluationItemInput],
+        metric_names: list[str],
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        return self._evaluate(items, metric_names, model_provider, model_name, tenant_id, EvaluationCategory.AGENT)
+
+    def evaluate_workflow(
+        self,
+        items: list[EvaluationItemInput],
+        metric_names: list[str],
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        return self._evaluate(items, metric_names, model_provider, model_name, tenant_id, EvaluationCategory.WORKFLOW)
+
+    def _evaluate(
+        self,
+        items: list[EvaluationItemInput],
+        metric_names: list[str],
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+        category: EvaluationCategory,
+    ) -> list[EvaluationItemResult]:
+        """Core evaluation logic using RAGAS."""
+        model_wrapper = DifyModelWrapper(model_provider, model_name, tenant_id)
+        requested_metrics = metric_names or self.get_supported_metrics(category)
+
+        try:
+            return self._evaluate_with_ragas(items, requested_metrics, model_wrapper, category)
+        except ImportError:
+            logger.warning("RAGAS not installed, falling back to simple evaluation")
+            return self._evaluate_simple(items, requested_metrics, model_wrapper)
+
+    def _evaluate_with_ragas(
+        self,
+        items: list[EvaluationItemInput],
+        requested_metrics: list[str],
+        model_wrapper: DifyModelWrapper,
+        category: EvaluationCategory,
+    ) -> list[EvaluationItemResult]:
+        """Evaluate using RAGAS library.
+
+        Builds SingleTurnSample differently per category to match ragas requirements:
+        - LLM/Workflow: user_input=prompt, response=output, reference=expected_output
+        - Retrieval: user_input=query, reference=expected_output, retrieved_contexts=context
+        - Agent: Not supported via EvaluationDataset (requires message-based API)
+        """
+        from ragas import evaluate as ragas_evaluate
+        from ragas.dataset_schema import EvaluationDataset
+
+        samples: list[Any] = []
+        for item in items:
+            sample = self._build_sample(item, category)
+            samples.append(sample)
+
+        dataset = EvaluationDataset(samples=samples)
+
+        ragas_metrics = self._build_ragas_metrics(requested_metrics)
+        if not ragas_metrics:
+            logger.warning("No valid RAGAS metrics found for: %s", requested_metrics)
+            return [EvaluationItemResult(index=item.index) for item in items]
+
+        try:
+            result = ragas_evaluate(
+                dataset=dataset,
+                metrics=ragas_metrics,
+            )
+
+            results: list[EvaluationItemResult] = []
+            result_df = result.to_pandas()
+            for i, item in enumerate(items):
+                metrics: list[EvaluationMetric] = []
+                for m_name in requested_metrics:
+                    if m_name in result_df.columns:
+                        score = result_df.iloc[i][m_name]
+                        if score is not None and not (isinstance(score, float) and score != score):
+                            metrics.append(EvaluationMetric(name=m_name, value=float(score)))
+                results.append(EvaluationItemResult(index=item.index, metrics=metrics))
+            return results
+        except Exception:
+            logger.exception("RAGAS evaluation failed, falling back to simple evaluation")
+            return self._evaluate_simple(items, requested_metrics, model_wrapper)
+
+    @staticmethod
+    def _build_sample(item: EvaluationItemInput, category: EvaluationCategory) -> Any:
+        """Build a ragas SingleTurnSample with the correct fields per category.
+
+        ragas metric field requirements:
+        - faithfulness:       user_input, response, retrieved_contexts
+        - answer_relevancy:   user_input, response
+        - answer_correctness: user_input, response, reference
+        - semantic_similarity: user_input, response, reference
+        - context_precision:  user_input, reference, retrieved_contexts
+        - context_recall:     user_input, reference, retrieved_contexts
+        - context_relevance:  user_input, retrieved_contexts
+        """
+        from ragas.dataset_schema import SingleTurnSample
+
+        user_input = _format_input(item.inputs, category)
+
+        match category:
+            case EvaluationCategory.LLM:
+                # response = actual LLM output, reference = expected output
+                return SingleTurnSample(
+                    user_input=user_input,
+                    response=item.output,
+                    reference=item.expected_output or "",
+                    retrieved_contexts=item.context or [],
+                )
+            case EvaluationCategory.RETRIEVAL:
+                # context_precision/recall only need reference + retrieved_contexts
+                return SingleTurnSample(
+                    user_input=user_input,
+                    reference=item.expected_output or "",
+                    retrieved_contexts=item.context or [],
+                )
+            case _:
+                return SingleTurnSample(
+                    user_input=user_input,
+                    response=item.output,
+                )
+
+    def _evaluate_simple(
+        self,
+        items: list[EvaluationItemInput],
+        requested_metrics: list[str],
+        model_wrapper: DifyModelWrapper,
+    ) -> list[EvaluationItemResult]:
+        """Simple LLM-as-judge fallback when RAGAS is not available."""
+        results: list[EvaluationItemResult] = []
+        for item in items:
+            metrics: list[EvaluationMetric] = []
+            for m_name in requested_metrics:
+                try:
+                    score = self._judge_with_llm(model_wrapper, m_name, item)
+                    metrics.append(EvaluationMetric(name=m_name, value=score))
+                except Exception:
+                    logger.exception("Failed to compute metric %s for item %d", m_name, item.index)
+            results.append(EvaluationItemResult(index=item.index, metrics=metrics))
+        return results
+
+    def _judge_with_llm(
+        self,
+        model_wrapper: DifyModelWrapper,
+        metric_name: str,
+        item: EvaluationItemInput,
+    ) -> float:
+        """Use the LLM to judge a single metric for a single item."""
+        prompt = self._build_judge_prompt(metric_name, item)
+        response = model_wrapper.invoke(prompt)
+        return self._parse_score(response)
+
+    @staticmethod
+    def _build_judge_prompt(metric_name: str, item: EvaluationItemInput) -> str:
+        """Build a scoring prompt for the LLM judge."""
+        parts = [
+            f"Evaluate the following on the metric '{metric_name}' using a scale of 0.0 to 1.0.",
+            f"\nInput: {item.inputs}",
+            f"\nOutput: {item.output}",
+        ]
+        if item.expected_output:
+            parts.append(f"\nExpected Output: {item.expected_output}")
+        if item.context:
+            parts.append(f"\nContext: {'; '.join(item.context)}")
+        parts.append("\nRespond with ONLY a single floating point number between 0.0 and 1.0, nothing else.")
+        return "\n".join(parts)
+
+    @staticmethod
+    def _parse_score(response: str) -> float:
+        """Parse a float score from LLM response."""
+        import re
+
+        cleaned = response.strip()
+        try:
+            score = float(cleaned)
+            return max(0.0, min(1.0, score))
+        except ValueError:
+            match = re.search(r"(\d+\.?\d*)", cleaned)
+            if match:
+                score = float(match.group(1))
+                return max(0.0, min(1.0, score))
+            return 0.0
+
+    @staticmethod
+    def _build_ragas_metrics(requested_metrics: list[str]) -> list[Any]:
+        """Build RAGAS metric instances from canonical metric names."""
+        try:
+            from ragas.metrics.collections import (
+                AnswerCorrectness,
+                AnswerRelevancy,
+                ContextPrecision,
+                ContextRecall,
+                ContextRelevance,
+                Faithfulness,
+                SemanticSimilarity,
+                ToolCallAccuracy,
+            )
+
+            # Maps canonical name → ragas metric class
+            ragas_class_map: dict[str, Any] = {
+                EvaluationMetricName.FAITHFULNESS: Faithfulness,
+                EvaluationMetricName.ANSWER_RELEVANCY: AnswerRelevancy,
+                EvaluationMetricName.ANSWER_CORRECTNESS: AnswerCorrectness,
+                EvaluationMetricName.SEMANTIC_SIMILARITY: SemanticSimilarity,
+                EvaluationMetricName.CONTEXT_PRECISION: ContextPrecision,
+                EvaluationMetricName.CONTEXT_RECALL: ContextRecall,
+                EvaluationMetricName.CONTEXT_RELEVANCE: ContextRelevance,
+                EvaluationMetricName.TOOL_CORRECTNESS: ToolCallAccuracy,
+            }
+
+            metrics = []
+            for name in requested_metrics:
+                metric_class = ragas_class_map.get(name)
+                if metric_class:
+                    metrics.append(metric_class())
+                else:
+                    logger.warning("Metric '%s' is not supported by RAGAS, skipping", name)
+            return metrics
+        except ImportError:
+            logger.warning("RAGAS metrics not available")
+            return []
+
+
+def _format_input(inputs: dict[str, Any], category: EvaluationCategory) -> str:
+    """Extract the user-facing input string from the inputs dict."""
+    match category:
+        case EvaluationCategory.LLM | EvaluationCategory.WORKFLOW:
+            return str(inputs.get("prompt", ""))
+        case EvaluationCategory.RETRIEVAL:
+            return str(inputs.get("query", ""))
+        case _:
+            return str(next(iter(inputs.values()), "")) if inputs else ""

api/core/evaluation/frameworks/ragas/ragas_model_wrapper.py

@@ -0,0 +1,48 @@
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+class DifyModelWrapper:
+    """Wraps Dify's model invocation interface for use by RAGAS as an LLM judge.
+
+    RAGAS requires an LLM to compute certain metrics (faithfulness, answer_relevancy, etc.).
+    This wrapper bridges Dify's ModelInstance to a callable that RAGAS can use.
+    """
+
+    def __init__(self, model_provider: str, model_name: str, tenant_id: str):
+        self.model_provider = model_provider
+        self.model_name = model_name
+        self.tenant_id = tenant_id
+
+    def _get_model_instance(self) -> Any:
+        from core.model_manager import ModelManager
+        from core.model_runtime.entities.model_entities import ModelType
+
+        model_manager = ModelManager()
+        model_instance = model_manager.get_model_instance(
+            tenant_id=self.tenant_id,
+            provider=self.model_provider,
+            model_type=ModelType.LLM,
+            model=self.model_name,
+        )
+        return model_instance
+
+    def invoke(self, prompt: str) -> str:
+        """Invoke the model with a text prompt and return the text response."""
+        from core.model_runtime.entities.message_entities import (
+            SystemPromptMessage,
+            UserPromptMessage,
+        )
+
+        model_instance = self._get_model_instance()
+        result = model_instance.invoke_llm(
+            prompt_messages=[
+                SystemPromptMessage(content="You are an evaluation judge. Answer precisely and concisely."),
+                UserPromptMessage(content=prompt),
+            ],
+            model_parameters={"temperature": 0.0, "max_tokens": 2048},
+            stream=False,
+        )
+        return result.message.content

api/core/evaluation/judgment/processor.py

@@ -0,0 +1,160 @@
+"""Judgment condition processor for evaluation metrics.
+
+Evaluates pass/fail judgment conditions against evaluation metric values.
+Each condition uses ``variable_selector`` (``[node_id, metric_name]``) to
+look up the metric value, then delegates the actual comparison to the
+workflow condition engine (``graphon.utils.condition.processor``).
+
+The processor is intentionally decoupled from evaluation frameworks and
+runners.  It operates on plain ``dict`` mappings and can be invoked
+anywhere that already has per-item metric results.
+"""
+
+import logging
+from collections.abc import Sequence
+from typing import Any, cast
+
+from core.evaluation.entities.judgment_entity import (
+    JudgmentCondition,
+    JudgmentConditionResult,
+    JudgmentConfig,
+    JudgmentResult,
+)
+from graphon.utils.condition.entities import SupportedComparisonOperator
+from graphon.utils.condition.processor import _evaluate_condition  # pyright: ignore[reportPrivateUsage]
+
+logger = logging.getLogger(__name__)
+
+_UNARY_OPERATORS = frozenset({"null", "not null", "empty", "not empty"})
+
+
+class JudgmentProcessor:
+    @staticmethod
+    def evaluate(
+        metric_values: dict[tuple[str, str], Any],
+        config: JudgmentConfig,
+    ) -> JudgmentResult:
+        """Evaluate all judgment conditions against the given metric values.
+
+        Args:
+            metric_values: Mapping of ``(node_id, metric_name)`` → metric
+                value (e.g. ``{("node_abc", "faithfulness"): 0.85}``).
+            config: The judgment configuration with logical_operator and
+                conditions.
+
+        Returns:
+            JudgmentResult with overall pass/fail and per-condition details.
+        """
+        if not config.conditions:
+            return JudgmentResult(
+                passed=True,
+                logical_operator=config.logical_operator,
+                condition_results=[],
+            )
+
+        condition_results: list[JudgmentConditionResult] = []
+
+        for condition in config.conditions:
+            result = JudgmentProcessor._evaluate_single_condition(metric_values, condition)
+            condition_results.append(result)
+
+            if config.logical_operator == "and" and not result.passed:
+                return JudgmentResult(
+                    passed=False,
+                    logical_operator=config.logical_operator,
+                    condition_results=condition_results,
+                )
+            if config.logical_operator == "or" and result.passed:
+                return JudgmentResult(
+                    passed=True,
+                    logical_operator=config.logical_operator,
+                    condition_results=condition_results,
+                )
+
+        if config.logical_operator == "and":
+            final_passed = all(r.passed for r in condition_results)
+        else:
+            final_passed = any(r.passed for r in condition_results)
+
+        return JudgmentResult(
+            passed=final_passed,
+            logical_operator=config.logical_operator,
+            condition_results=condition_results,
+        )
+
+    @staticmethod
+    def _evaluate_single_condition(
+        metric_values: dict[tuple[str, str], Any],
+        condition: JudgmentCondition,
+    ) -> JudgmentConditionResult:
+        """Evaluate a single judgment condition.
+
+        Steps:
+          1. Extract ``(node_id, metric_name)`` from ``variable_selector``.
+          2. Look up the metric value from ``metric_values``.
+          3. Delegate comparison to the workflow condition engine.
+        """
+        selector = condition.variable_selector
+        if len(selector) < 2:
+            return JudgmentConditionResult(
+                variable_selector=selector,
+                comparison_operator=condition.comparison_operator,
+                expected_value=condition.value,
+                actual_value=None,
+                passed=False,
+                error=f"variable_selector must have at least 2 elements, got {len(selector)}",
+            )
+
+        node_id, metric_name = selector[0], selector[1]
+        actual_value = metric_values.get((node_id, metric_name))
+
+        if actual_value is None and condition.comparison_operator not in _UNARY_OPERATORS:
+            return JudgmentConditionResult(
+                variable_selector=selector,
+                comparison_operator=condition.comparison_operator,
+                expected_value=condition.value,
+                actual_value=None,
+                passed=False,
+                error=f"Metric '{metric_name}' on node '{node_id}' not found in evaluation results",
+            )
+
+        try:
+            expected = condition.value
+            # Numeric operators need the actual value coerced to int/float
+            # so that the workflow engine's numeric assertions work correctly.
+            coerced_actual: object = actual_value
+            if (
+                condition.comparison_operator in {"=", "≠", ">", "<", "≥", "≤"}
+                and actual_value is not None
+                and not isinstance(actual_value, (int, float, bool))
+            ):
+                coerced_actual = float(actual_value)
+
+            passed = _evaluate_condition(
+                operator=cast(SupportedComparisonOperator, condition.comparison_operator),
+                value=coerced_actual,
+                expected=cast(str | Sequence[str] | bool | Sequence[bool] | None, expected),
+            )
+
+            return JudgmentConditionResult(
+                variable_selector=selector,
+                comparison_operator=condition.comparison_operator,
+                expected_value=expected,
+                actual_value=actual_value,
+                passed=passed,
+            )
+        except Exception as e:
+            logger.warning(
+                "Judgment condition evaluation failed for [%s, %s]: %s",
+                node_id,
+                metric_name,
+                str(e),
+            )
+            return JudgmentConditionResult(
+                variable_selector=selector,
+                comparison_operator=condition.comparison_operator,
+                expected_value=condition.value,
+                actual_value=actual_value,
+                passed=False,
+                error=str(e),
+            )

api/core/evaluation/runners/__init__.py

@@ -0,0 +1,52 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from models import Account, App, CustomizedSnippet, TenantAccountJoin
+
+
+def get_service_account_for_app(session: Session, app_id: str) -> Account:
+    """Get the creator account for an app with tenant context set up.
+
+    This follows the same pattern as BaseTraceInstance.get_service_account_with_tenant().
+    """
+    app = session.scalar(select(App).where(App.id == app_id))
+    if not app:
+        raise ValueError(f"App with id {app_id} not found")
+
+    if not app.created_by:
+        raise ValueError(f"App with id {app_id} has no creator")
+
+    account = session.scalar(select(Account).where(Account.id == app.created_by))
+    if not account:
+        raise ValueError(f"Creator account not found for app {app_id}")
+
+    current_tenant = session.query(TenantAccountJoin).filter_by(account_id=account.id, current=True).first()
+    if not current_tenant:
+        raise ValueError(f"Current tenant not found for account {account.id}")
+
+    account.set_tenant_id(current_tenant.tenant_id)
+    return account
+
+
+def get_service_account_for_snippet(session: Session, snippet_id: str) -> Account:
+    """Get the creator account for a snippet with tenant context set up.
+
+    Mirrors :func:`get_service_account_for_app` but queries CustomizedSnippet.
+    """
+    snippet = session.scalar(select(CustomizedSnippet).where(CustomizedSnippet.id == snippet_id))
+    if not snippet:
+        raise ValueError(f"Snippet with id {snippet_id} not found")
+
+    if not snippet.created_by:
+        raise ValueError(f"Snippet with id {snippet_id} has no creator")
+
+    account = session.scalar(select(Account).where(Account.id == snippet.created_by))
+    if not account:
+        raise ValueError(f"Creator account not found for snippet {snippet_id}")
+
+    current_tenant = session.query(TenantAccountJoin).filter_by(account_id=account.id, current=True).first()
+    if not current_tenant:
+        raise ValueError(f"Current tenant not found for account {account.id}")
+
+    account.set_tenant_id(current_tenant.tenant_id)
+    return account

api/core/evaluation/runners/agent_evaluation_runner.py

@@ -0,0 +1,62 @@
+import logging
+from collections.abc import Mapping
+from typing import Any
+
+from core.evaluation.base_evaluation_instance import BaseEvaluationInstance
+from core.evaluation.entities.evaluation_entity import (
+    DefaultMetric,
+    EvaluationItemInput,
+    EvaluationItemResult,
+)
+from core.evaluation.runners.base_evaluation_runner import BaseEvaluationRunner
+from graphon.node_events import NodeRunResult
+
+logger = logging.getLogger(__name__)
+
+
+class AgentEvaluationRunner(BaseEvaluationRunner):
+    """Runner for agent evaluation: collects tool calls and final output."""
+
+    def __init__(self, evaluation_instance: BaseEvaluationInstance):
+        super().__init__(evaluation_instance)
+
+    def evaluate_metrics(
+        self,
+        node_run_result_list: list[NodeRunResult],
+        default_metric: DefaultMetric,
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        """Compute agent evaluation metrics."""
+        if not node_run_result_list:
+            return []
+        merged_items = self._merge_results_into_items(node_run_result_list)
+        return self.evaluation_instance.evaluate_agent(
+            merged_items, [default_metric.metric], model_provider, model_name, tenant_id
+        )
+
+    @staticmethod
+    def _merge_results_into_items(items: list[NodeRunResult]) -> list[EvaluationItemInput]:
+        """Create EvaluationItemInput list from NodeRunResult for agent evaluation."""
+        merged = []
+        for i, item in enumerate(items):
+            output = _extract_agent_output(item.outputs)
+            merged.append(
+                EvaluationItemInput(
+                    index=i,
+                    inputs=dict(item.inputs),
+                    output=output,
+                )
+            )
+        return merged
+
+
+def _extract_agent_output(outputs: Mapping[str, Any]) -> str:
+    """Extract the primary output text from agent NodeRunResult.outputs."""
+    if "answer" in outputs:
+        return str(outputs["answer"])
+    if "text" in outputs:
+        return str(outputs["text"])
+    values = list(outputs.values())
+    return str(values[0]) if values else ""

api/core/evaluation/runners/base_evaluation_runner.py

@@ -0,0 +1,51 @@
+"""Base evaluation runner.
+
+Provides the abstract interface for metric computation. Each concrete runner
+(LLM, Retrieval, Agent, Workflow, Snippet) implements ``evaluate_metrics``
+to compute scores for a specific node type.
+
+Orchestration (merging results from multiple runners, applying judgment, and
+persisting to the database) is handled by the evaluation task, not the runner.
+"""
+
+import logging
+from abc import ABC, abstractmethod
+
+from core.evaluation.base_evaluation_instance import BaseEvaluationInstance
+from core.evaluation.entities.evaluation_entity import (
+    DefaultMetric,
+    EvaluationItemResult,
+)
+from graphon.node_events import NodeRunResult
+
+logger = logging.getLogger(__name__)
+
+
+class BaseEvaluationRunner(ABC):
+    """Abstract base class for evaluation runners.
+
+    Runners are stateless metric calculators: they receive node execution
+    results and a metric specification, then return scored results.  They
+    do **not** touch the database or apply judgment logic.
+    """
+
+    def __init__(self, evaluation_instance: BaseEvaluationInstance):
+        self.evaluation_instance = evaluation_instance
+
+    @abstractmethod
+    def evaluate_metrics(
+        self,
+        node_run_result_list: list[NodeRunResult],
+        default_metric: DefaultMetric,
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        """Compute evaluation metrics on the collected results.
+
+        The returned ``EvaluationItemResult.index`` values are positional
+        (0-based) and correspond to the order of *node_run_result_list*.
+        The caller is responsible for mapping them back to the original
+        dataset indices.
+        """
+        ...

api/core/evaluation/runners/llm_evaluation_runner.py

@@ -0,0 +1,83 @@
+import logging
+from collections.abc import Mapping
+from typing import Any
+
+from core.evaluation.base_evaluation_instance import BaseEvaluationInstance
+from core.evaluation.entities.evaluation_entity import (
+    DefaultMetric,
+    EvaluationItemInput,
+    EvaluationItemResult,
+)
+from core.evaluation.runners.base_evaluation_runner import BaseEvaluationRunner
+from graphon.node_events import NodeRunResult
+
+logger = logging.getLogger(__name__)
+
+
+class LLMEvaluationRunner(BaseEvaluationRunner):
+    """Runner for LLM evaluation: extracts prompts/outputs then evaluates."""
+
+    def __init__(self, evaluation_instance: BaseEvaluationInstance):
+        super().__init__(evaluation_instance)
+
+    def evaluate_metrics(
+        self,
+        node_run_result_list: list[NodeRunResult],
+        default_metric: DefaultMetric,
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        """Use the evaluation instance to compute LLM metrics."""
+        if not node_run_result_list:
+            return []
+        merged_items = self._merge_results_into_items(node_run_result_list)
+        return self.evaluation_instance.evaluate_llm(
+            merged_items, [default_metric.metric], model_provider, model_name, tenant_id
+        )
+
+    @staticmethod
+    def _merge_results_into_items(
+        items: list[NodeRunResult],
+    ) -> list[EvaluationItemInput]:
+        """Create new items from NodeRunResult for ragas evaluation.
+
+        Extracts prompts from process_data and concatenates them into a single
+        string with role prefixes (e.g. "system: ...\nuser: ...\nassistant: ...").
+        The last assistant message in outputs is used as the actual output.
+        """
+        merged = []
+        for i, item in enumerate(items):
+            prompt = _format_prompts(item.process_data.get("prompts", []))
+            output = _extract_llm_output(item.outputs)
+            merged.append(
+                EvaluationItemInput(
+                    index=i,
+                    inputs={"prompt": prompt},
+                    output=output,
+                )
+            )
+        return merged
+
+
+def _format_prompts(prompts: list[dict[str, Any]]) -> str:
+    """Concatenate a list of prompt messages into a single string for evaluation.
+
+    Each message is formatted as "role: text" and joined with newlines.
+    """
+    parts: list[str] = []
+    for msg in prompts:
+        role = msg.get("role", "unknown")
+        text = msg.get("text", "")
+        parts.append(f"{role}: {text}")
+    return "\n".join(parts)
+
+
+def _extract_llm_output(outputs: Mapping[str, Any]) -> str:
+    """Extract the LLM output text from NodeRunResult.outputs."""
+    if "text" in outputs:
+        return str(outputs["text"])
+    if "answer" in outputs:
+        return str(outputs["answer"])
+    values = list(outputs.values())
+    return str(values[0]) if values else ""

api/core/evaluation/runners/retrieval_evaluation_runner.py

@@ -0,0 +1,61 @@
+import logging
+from typing import Any
+
+from core.evaluation.base_evaluation_instance import BaseEvaluationInstance
+from core.evaluation.entities.evaluation_entity import (
+    DefaultMetric,
+    EvaluationItemInput,
+    EvaluationItemResult,
+)
+from core.evaluation.runners.base_evaluation_runner import BaseEvaluationRunner
+from graphon.node_events import NodeRunResult
+
+logger = logging.getLogger(__name__)
+
+
+class RetrievalEvaluationRunner(BaseEvaluationRunner):
+    """Runner for retrieval evaluation: performs knowledge base retrieval, then evaluates."""
+
+    def __init__(self, evaluation_instance: BaseEvaluationInstance):
+        super().__init__(evaluation_instance)
+
+    def evaluate_metrics(
+        self,
+        node_run_result_list: list[NodeRunResult],
+        default_metric: DefaultMetric,
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        """Compute retrieval evaluation metrics."""
+        if not node_run_result_list:
+            return []
+
+        merged_items = []
+        for i, node_result in enumerate(node_run_result_list):
+            outputs = node_result.outputs
+            query = self._extract_query(dict(node_result.inputs))
+            result_list = outputs.get("result", [])
+            contexts = [item.get("content", "") for item in result_list if item.get("content")]
+            output = "\n---\n".join(contexts)
+
+            merged_items.append(
+                EvaluationItemInput(
+                    index=i,
+                    inputs={"query": query},
+                    output=output,
+                    context=contexts,
+                )
+            )
+
+        return self.evaluation_instance.evaluate_retrieval(
+            merged_items, [default_metric.metric], model_provider, model_name, tenant_id
+        )
+
+    @staticmethod
+    def _extract_query(inputs: dict[str, Any]) -> str:
+        for key in ("query", "question", "input", "text"):
+            if key in inputs:
+                return str(inputs[key])
+        values = list(inputs.values())
+        return str(values[0]) if values else ""

api/core/evaluation/runners/snippet_evaluation_runner.py

@@ -0,0 +1,68 @@
+"""Runner for Snippet evaluation.
+
+Snippets are essentially workflows, so we reuse ``evaluate_workflow`` from
+the evaluation instance for metric computation.
+"""
+
+import logging
+from collections.abc import Mapping
+from typing import Any
+
+from core.evaluation.base_evaluation_instance import BaseEvaluationInstance
+from core.evaluation.entities.evaluation_entity import (
+    DefaultMetric,
+    EvaluationItemInput,
+    EvaluationItemResult,
+)
+from core.evaluation.runners.base_evaluation_runner import BaseEvaluationRunner
+from graphon.node_events import NodeRunResult
+
+logger = logging.getLogger(__name__)
+
+
+class SnippetEvaluationRunner(BaseEvaluationRunner):
+    """Runner for snippet evaluation: evaluates a published Snippet workflow."""
+
+    def __init__(self, evaluation_instance: BaseEvaluationInstance):
+        super().__init__(evaluation_instance)
+
+    def evaluate_metrics(
+        self,
+        node_run_result_list: list[NodeRunResult],
+        default_metric: DefaultMetric,
+        model_provider: str,
+        model_name: str,
+        tenant_id: str,
+    ) -> list[EvaluationItemResult]:
+        """Compute evaluation metrics for snippet outputs."""
+        if not node_run_result_list:
+            return []
+        merged_items = self._merge_results_into_items(node_run_result_list)
+        return self.evaluation_instance.evaluate_workflow(
+            merged_items, [default_metric.metric], model_provider, model_name, tenant_id
+        )
+
+    @staticmethod
+    def _merge_results_into_items(items: list[NodeRunResult]) -> list[EvaluationItemInput]:
+        """Create EvaluationItemInput list from NodeRunResult for snippet evaluation."""
+        merged = []
+        for i, item in enumerate(items):
+            output = _extract_snippet_output(item.outputs)
+            merged.append(
+                EvaluationItemInput(
+                    index=i,
+                    inputs=dict(item.inputs),
+                    output=output,
+                )
+            )
+        return merged
+
+
+def _extract_snippet_output(outputs: Mapping[str, Any]) -> str:
+    """Extract the primary output text from snippet NodeRunResult.outputs."""
+    if "answer" in outputs:
+        return str(outputs["answer"])
+    if "text" in outputs:
+        return str(outputs["text"])
+    values = list(outputs.values())
+    return str(values[0]) if values else ""