fix CI

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

.agents/skills/backend-code-review/SKILL.md

@@ -0,0 +1,168 @@
+---
+name: backend-code-review
+description: Review backend code for quality, security, maintainability, and best practices based on established checklist rules. Use when the user requests a review, analysis, or improvement of backend files (e.g., `.py`) under the `api/` directory. Do NOT use for frontend files (e.g., `.tsx`, `.ts`, `.js`). Supports pending-change review, code snippets review, and file-focused review.
+---
+
+# Backend Code Review
+
+## When to use this skill
+
+Use this skill whenever the user asks to **review, analyze, or improve** backend code (e.g., `.py`) under the `api/` directory. Supports the following review modes:
+
+- **Pending-change review**: when the user asks to review current changes (inspect staged/working-tree files slated for commit to get the changes).
+- **Code snippets review**: when the user pastes code snippets (e.g., a function/class/module excerpt) into the chat and asks for a review.
+- **File-focused review**: when the user points to specific files and asks for a review of those files (one file or a small, explicit set of files, e.g., `api/...`, `api/app.py`).
+
+Do NOT use this skill when:
+
+- The request is about frontend code or UI (e.g., `.tsx`, `.ts`, `.js`, `web/`).
+- The user is not asking for a review/analysis/improvement of backend code.
+- The scope is not under `api/` (unless the user explicitly asks to review backend-related changes outside `api/`).
+
+## How to use this skill
+
+Follow these steps when using this skill:
+
+1. **Identify the review mode** (pending-change vs snippet vs file-focused) based on the user’s input. Keep the scope tight: review only what the user provided or explicitly referenced.
+2. Follow the rules defined in **Checklist** to perform the review. If no Checklist rule matches, apply **General Review Rules** as a fallback to perform the best-effort review.
+3. Compose the final output strictly follow the **Required Output Format**.
+
+Notes when using this skill:
+- Always include actionable fixes or suggestions (including possible code snippets).
+- Use best-effort `File:Line` references when a file path and line numbers are available; otherwise, use the most specific identifier you can.
+
+## Checklist
+
+- db schema design: if the review scope includes code/files under `api/models/` or `api/migrations/`, follow [references/db-schema-rule.md](references/db-schema-rule.md) to perform the review
+- architecture: if the review scope involves controller/service/core-domain/libs/model layering, dependency direction, or moving responsibilities across modules, follow [references/architecture-rule.md](references/architecture-rule.md) to perform the review
+- repositories abstraction: if the review scope contains table/model operations (e.g., `select(...)`, `session.execute(...)`, joins, CRUD) and is not under `api/repositories`, `api/core/repositories`, or `api/extensions/*/repositories/`, follow [references/repositories-rule.md](references/repositories-rule.md) to perform the review
+- sqlalchemy patterns: if the review scope involves SQLAlchemy session/query usage, db transaction/crud usage, or raw SQL usage, follow [references/sqlalchemy-rule.md](references/sqlalchemy-rule.md) to perform the review
+
+## General Review Rules
+
+### 1. Security Review
+
+Check for:
+- SQL injection vulnerabilities
+- Server-Side Request Forgery (SSRF)
+- Command injection
+- Insecure deserialization
+- Hardcoded secrets/credentials
+- Improper authentication/authorization
+- Insecure direct object references
+
+### 2. Performance Review
+
+Check for:
+- N+1 queries
+- Missing database indexes
+- Memory leaks
+- Blocking operations in async code
+- Missing caching opportunities
+
+### 3. Code Quality Review
+
+Check for:
+- Code forward compatibility
+- Code duplication (DRY violations)
+- Functions doing too much (SRP violations)
+- Deep nesting / complex conditionals
+- Magic numbers/strings
+- Poor naming
+- Missing error handling
+- Incomplete type coverage
+
+### 4. Testing Review
+
+Check for:
+- Missing test coverage for new code
+- Tests that don't test behavior
+- Flaky test patterns
+- Missing edge cases
+
+## Required Output Format
+
+When this skill invoked, the response must exactly follow one of the two templates:
+
+### Template A (any findings)
+
+```markdown
+# Code Review Summary
+
+Found <X> critical issues need to be fixed:
+
+## 🔴 Critical (Must Fix)
+
+### 1. <brief description of the issue>
+
+FilePath: <path> line <line>
+<relevant code snippet or pointer>
+
+#### Explanation
+
+<detailed explanation and references of the issue>
+
+#### Suggested Fix
+
+1. <brief description of suggested fix>
+2. <code example> (optional, omit if not applicable)
+
+---
+... (repeat for each critical issue) ...
+
+Found <Y> suggestions for improvement:
+
+## 🟡 Suggestions (Should Consider)
+
+### 1. <brief description of the suggestion>
+
+FilePath: <path> line <line>
+<relevant code snippet or pointer>
+
+#### Explanation
+
+<detailed explanation and references of the suggestion>
+
+#### Suggested Fix
+
+1. <brief description of suggested fix>
+2. <code example> (optional, omit if not applicable)
+
+---
+... (repeat for each suggestion) ...
+
+Found <Z> optional nits:
+
+## 🟢 Nits (Optional)
+### 1. <brief description of the nit>
+
+FilePath: <path> line <line>
+<relevant code snippet or pointer>
+
+#### Explanation
+
+<explanation and references of the optional nit>
+
+#### Suggested Fix
+
+- <minor suggestions>
+
+---
+... (repeat for each nits) ...
+
+## ✅ What's Good
+
+- <Positive feedback on good patterns>
+```
+
+- If there are no critical issues or suggestions or option nits or good points, just omit that section.
+- If the issue number is more than 10, summarize as "Found 10+ critical issues/suggestions/optional nits" and only output the first 10 items.
+- Don't compress the blank lines between sections; keep them as-is for readability.
+- If there is any issue requires code changes, append a brief follow-up question to ask whether the user wants to apply the fix(es) after the structured output. For example: "Would you like me to use the Suggested fix(es) to address these issues?"
+
+### Template B (no issues)
+
+```markdown
+## Code Review Summary
+✅ No issues found.
+```

.agents/skills/backend-code-review/references/architecture-rule.md

@@ -0,0 +1,91 @@
+# Rule Catalog — Architecture
+
+## Scope
+- Covers: controller/service/core-domain/libs/model layering, dependency direction, responsibility placement, observability-friendly flow.
+
+## Rules
+
+### Keep business logic out of controllers
+- Category: maintainability
+- Severity: critical
+- Description: Controllers should parse input, call services, and return serialized responses. Business decisions inside controllers make behavior hard to reuse and test.
+- Suggested fix: Move domain/business logic into the service or core/domain layer. Keep controller handlers thin and orchestration-focused.
+- Example:
+  - Bad:
+    ```python
+    @bp.post("/apps/<app_id>/publish")
+    def publish_app(app_id: str):
+        payload = request.get_json() or {}
+        if payload.get("force") and current_user.role != "admin":
+            raise ValueError("only admin can force publish")
+        app = App.query.get(app_id)
+        app.status = "published"
+        db.session.commit()
+        return {"result": "ok"}
+    ```
+  - Good:
+    ```python
+    @bp.post("/apps/<app_id>/publish")
+    def publish_app(app_id: str):
+        payload = PublishRequest.model_validate(request.get_json() or {})
+        app_service.publish_app(app_id=app_id, force=payload.force, actor_id=current_user.id)
+        return {"result": "ok"}
+    ```
+
+### Preserve layer dependency direction
+- Category: best practices
+- Severity: critical
+- Description: Controllers may depend on services, and services may depend on core/domain abstractions. Reversing this direction (for example, core importing controller/web modules) creates cycles and leaks transport concerns into domain code.
+- Suggested fix: Extract shared contracts into core/domain or service-level modules and make upper layers depend on lower, not the reverse.
+- Example:
+  - Bad:
+    ```python
+    # core/policy/publish_policy.py
+    from controllers.console.app import request_context
+
+    def can_publish() -> bool:
+        return request_context.current_user.is_admin
+    ```
+  - Good:
+    ```python
+    # core/policy/publish_policy.py
+    def can_publish(role: str) -> bool:
+        return role == "admin"
+
+    # service layer adapts web/user context to domain input
+    allowed = can_publish(role=current_user.role)
+    ```
+
+### Keep libs business-agnostic
+- Category: maintainability
+- Severity: critical
+- Description: Modules under `api/libs/` should remain reusable, business-agnostic building blocks. They must not encode product/domain-specific rules, workflow orchestration, or business decisions.
+- Suggested fix:
+  - If business logic appears in `api/libs/`, extract it into the appropriate `services/` or `core/` module and keep `libs` focused on generic, cross-cutting helpers.
+  - Keep `libs` dependencies clean: avoid importing service/controller/domain-specific modules into `api/libs/`.
+- Example:
+  - Bad:
+    ```python
+    # api/libs/conversation_filter.py
+    from services.conversation_service import ConversationService
+
+    def should_archive_conversation(conversation, tenant_id: str) -> bool:
+        # Domain policy and service dependency are leaking into libs.
+        service = ConversationService()
+        if service.has_paid_plan(tenant_id):
+            return conversation.idle_days > 90
+        return conversation.idle_days > 30
+    ```
+  - Good:
+    ```python
+    # api/libs/datetime_utils.py (business-agnostic helper)
+    def older_than_days(idle_days: int, threshold_days: int) -> bool:
+        return idle_days > threshold_days
+
+    # services/conversation_service.py (business logic stays in service/core)
+    from libs.datetime_utils import older_than_days
+
+    def should_archive_conversation(conversation, tenant_id: str) -> bool:
+        threshold_days = 90 if has_paid_plan(tenant_id) else 30
+        return older_than_days(conversation.idle_days, threshold_days)
+    ```

.agents/skills/backend-code-review/references/db-schema-rule.md

@@ -0,0 +1,157 @@
+# Rule Catalog — DB Schema Design
+
+## Scope
+- Covers: model/base inheritance, schema boundaries in model properties, tenant-aware schema design, index redundancy checks, dialect portability in models, and cross-database compatibility in migrations.
+- Does NOT cover: session lifecycle, transaction boundaries, and query execution patterns (handled by `sqlalchemy-rule.md`).
+
+## Rules
+
+### Do not query other tables inside `@property`
+- Category: [maintainability, performance]
+- Severity: critical
+- Description: A model `@property` must not open sessions or query other tables. This hides dependencies across models, tightly couples schema objects to data access, and can cause N+1 query explosions when iterating collections.
+- Suggested fix:
+  - Keep model properties pure and local to already-loaded fields.
+  - Move cross-table data fetching to service/repository methods.
+  - For list/batch reads, fetch required related data explicitly (join/preload/bulk query) before rendering derived values.
+- Example:
+  - Bad:
+    ```python
+    class Conversation(TypeBase):
+        __tablename__ = "conversations"
+
+        @property
+        def app_name(self) -> str:
+            with Session(db.engine, expire_on_commit=False) as session:
+                app = session.execute(select(App).where(App.id == self.app_id)).scalar_one()
+                return app.name
+    ```
+  - Good:
+    ```python
+    class Conversation(TypeBase):
+        __tablename__ = "conversations"
+
+        @property
+        def display_title(self) -> str:
+            return self.name or "Untitled"
+
+
+    # Service/repository layer performs explicit batch fetch for related App rows.
+    ```
+
+### Prefer including `tenant_id` in model definitions
+- Category: maintainability
+- Severity: suggestion
+- Description: In multi-tenant domains, include `tenant_id` in schema definitions whenever the entity belongs to tenant-owned data. This improves data isolation safety and keeps future partitioning/sharding strategies practical as data volume grows.
+- Suggested fix:
+  - Add a `tenant_id` column and ensure related unique/index constraints include tenant dimension when applicable.
+  - Propagate `tenant_id` through service/repository contracts to keep access paths tenant-aware.
+  - Exception: if a table is explicitly designed as non-tenant-scoped global metadata, document that design decision clearly.
+- Example:
+  - Bad:
+    ```python
+    from sqlalchemy.orm import Mapped
+
+    class Dataset(TypeBase):
+        __tablename__ = "datasets"
+        id: Mapped[str] = mapped_column(StringUUID, primary_key=True)
+        name: Mapped[str] = mapped_column(sa.String(255), nullable=False)
+    ```
+  - Good:
+    ```python
+    from sqlalchemy.orm import Mapped
+
+    class Dataset(TypeBase):
+        __tablename__ = "datasets"
+        id: Mapped[str] = mapped_column(StringUUID, primary_key=True)
+        tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False, index=True)
+        name: Mapped[str] = mapped_column(sa.String(255), nullable=False)
+    ```
+
+### Detect and avoid duplicate/redundant indexes
+- Category: performance
+- Severity: suggestion
+- Description: Review index definitions for leftmost-prefix redundancy. For example, index `(a, b, c)` can safely cover most lookups for `(a, b)`. Keeping both may increase write overhead and can mislead the optimizer into suboptimal execution plans.
+- Suggested fix:
+  - Before adding an index, compare against existing composite indexes by leftmost-prefix rules.
+  - Drop or avoid creating redundant prefixes unless there is a proven query-pattern need.
+  - Apply the same review standard in both model `__table_args__` and migration index DDL.
+- Example:
+  - Bad:
+    ```python
+    __table_args__ = (
+        sa.Index("idx_msg_tenant_app", "tenant_id", "app_id"),
+        sa.Index("idx_msg_tenant_app_created", "tenant_id", "app_id", "created_at"),
+    )
+    ```
+  - Good:
+    ```python
+    __table_args__ = (
+        # Keep the wider index unless profiling proves a dedicated short index is needed.
+        sa.Index("idx_msg_tenant_app_created", "tenant_id", "app_id", "created_at"),
+    )
+    ```
+
+### Avoid PostgreSQL-only dialect usage in models; wrap in `models.types`
+- Category: maintainability
+- Severity: critical
+- Description: Model/schema definitions should avoid PostgreSQL-only constructs directly in business models. When database-specific behavior is required, encapsulate it in `api/models/types.py` using both PostgreSQL and MySQL dialect implementations, then consume that abstraction from model code.
+- Suggested fix:
+  - Do not directly place dialect-only types/operators in model columns when a portable wrapper can be used.
+  - Add or extend wrappers in `models.types` (for example, `AdjustedJSON`, `LongText`, `BinaryData`) to normalize behavior across PostgreSQL and MySQL.
+- Example:
+  - Bad:
+    ```python
+    from sqlalchemy.dialects.postgresql import JSONB
+    from sqlalchemy.orm import Mapped
+
+    class ToolConfig(TypeBase):
+        __tablename__ = "tool_configs"
+        config: Mapped[dict] = mapped_column(JSONB, nullable=False)
+    ```
+  - Good:
+    ```python
+    from sqlalchemy.orm import Mapped
+
+    from models.types import AdjustedJSON
+
+    class ToolConfig(TypeBase):
+        __tablename__ = "tool_configs"
+        config: Mapped[dict] = mapped_column(AdjustedJSON(), nullable=False)
+    ```
+
+### Guard migration incompatibilities with dialect checks and shared types
+- Category: maintainability
+- Severity: critical
+- Description: Migration scripts under `api/migrations/versions/` must account for PostgreSQL/MySQL incompatibilities explicitly. For dialect-sensitive DDL or defaults, branch on the active dialect (for example, `conn.dialect.name == "postgresql"`), and prefer reusable compatibility abstractions from `models.types` where applicable.
+- Suggested fix:
+  - In migration upgrades/downgrades, bind connection and branch by dialect for incompatible SQL fragments.
+  - Reuse `models.types` wrappers in column definitions when that keeps behavior aligned with runtime models.
+  - Avoid one-dialect-only migration logic unless there is a documented, deliberate compatibility exception.
+- Example:
+  - Bad:
+    ```python
+    with op.batch_alter_table("dataset_keyword_tables") as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "data_source_type",
+                sa.String(255),
+                server_default=sa.text("'database'::character varying"),
+                nullable=False,
+            )
+        )
+    ```
+  - Good:
+    ```python
+    def _is_pg(conn) -> bool:
+        return conn.dialect.name == "postgresql"
+
+
+    conn = op.get_bind()
+    default_expr = sa.text("'database'::character varying") if _is_pg(conn) else sa.text("'database'")
+
+    with op.batch_alter_table("dataset_keyword_tables") as batch_op:
+        batch_op.add_column(
+            sa.Column("data_source_type", sa.String(255), server_default=default_expr, nullable=False)
+        )
+    ```

.agents/skills/backend-code-review/references/repositories-rule.md

@@ -0,0 +1,61 @@
+# Rule Catalog - Repositories Abstraction
+
+## Scope
+- Covers: when to reuse existing repository abstractions, when to introduce new repositories, and how to preserve dependency direction between service/core and infrastructure implementations.
+- Does NOT cover: SQLAlchemy session lifecycle and query-shape specifics (handled by `sqlalchemy-rule.md`), and table schema/migration design (handled by `db-schema-rule.md`).
+
+## Rules
+
+### Introduce repositories abstraction
+- Category: maintainability
+- Severity: suggestion
+- Description: If a table/model already has a repository abstraction, all reads/writes/queries for that table should use the existing repository. If no repository exists, introduce one only when complexity justifies it, such as large/high-volume tables, repeated complex query logic, or likely storage-strategy variation.
+- Suggested fix:
+  - First check  `api/repositories`, `api/core/repositories`, and `api/extensions/*/repositories/` to verify whether the table/model already has a repository abstraction. If it exists, route all operations through it and add missing repository methods instead of bypassing it with ad-hoc SQLAlchemy access.
+  - If no repository exists, add one only when complexity warrants it (for example, repeated complex queries, large data domains, or multiple storage strategies), while preserving dependency direction (service/core depends on abstraction; infra provides implementation).
+- Example:
+  - Bad:
+    ```python
+    # Existing repository is ignored and service uses ad-hoc table queries.
+    class AppService:
+        def archive_app(self, app_id: str, tenant_id: str) -> None:
+            app = self.session.execute(
+                select(App).where(App.id == app_id, App.tenant_id == tenant_id)
+            ).scalar_one()
+            app.archived = True
+            self.session.commit()
+    ```
+  - Good:
+    ```python
+    # Case A: Existing repository must be reused for all table operations.
+    class AppService:
+        def archive_app(self, app_id: str, tenant_id: str) -> None:
+            app = self.app_repo.get_by_id(app_id=app_id, tenant_id=tenant_id)
+            app.archived = True
+            self.app_repo.save(app)
+
+    # If the query is missing, extend the existing abstraction.
+    active_apps = self.app_repo.list_active_for_tenant(tenant_id=tenant_id)
+    ```
+  - Bad:
+    ```python
+    # No repository exists, but large-domain query logic is scattered in service code.
+    class ConversationService:
+        def list_recent_for_app(self, app_id: str, tenant_id: str, limit: int) -> list[Conversation]:
+            ...
+            # many filters/joins/pagination variants duplicated across services
+    ```
+  - Good:
+    ```python
+    # Case B: Introduce repository for large/complex domains or storage variation.
+    class ConversationRepository(Protocol):
+        def list_recent_for_app(self, app_id: str, tenant_id: str, limit: int) -> list[Conversation]: ...
+
+    class SqlAlchemyConversationRepository:
+        def list_recent_for_app(self, app_id: str, tenant_id: str, limit: int) -> list[Conversation]:
+            ...
+
+    class ConversationService:
+        def __init__(self, conversation_repo: ConversationRepository):
+            self.conversation_repo = conversation_repo
+    ```

.agents/skills/backend-code-review/references/sqlalchemy-rule.md

@@ -0,0 +1,139 @@
+# Rule Catalog — SQLAlchemy Patterns
+
+## Scope
+- Covers: SQLAlchemy session and transaction lifecycle, query construction, tenant scoping, raw SQL boundaries, and write-path concurrency safeguards.
+- Does NOT cover: table/model schema and migration design details (handled by `db-schema-rule.md`).
+
+## Rules
+
+### Use Session context manager with explicit transaction control behavior
+- Category: best practices
+- Severity: critical
+- Description: Session and transaction lifecycle must be explicit and bounded on write paths. Missing commits can silently drop intended updates, while ad-hoc or long-lived transactions increase contention, lock duration, and deadlock risk.
+- Suggested fix:
+  - Use **explicit `session.commit()`** after completing a related write unit.
+  - Or use **`session.begin()` context manager** for automatic commit/rollback on a scoped block.
+  - Keep transaction windows short: avoid network I/O, heavy computation, or unrelated work inside the transaction.
+- Example:
+  - Bad:
+    ```python
+    # Missing commit: write may never be persisted.
+    with Session(db.engine, expire_on_commit=False) as session:
+        run = session.get(WorkflowRun, run_id)
+        run.status = "cancelled"
+
+    # Long transaction: external I/O inside a DB transaction.
+    with Session(db.engine, expire_on_commit=False) as session, session.begin():
+        run = session.get(WorkflowRun, run_id)
+        run.status = "cancelled"
+        call_external_api()
+    ```
+  - Good:
+    ```python
+    # Option 1: explicit commit.
+    with Session(db.engine, expire_on_commit=False) as session:
+        run = session.get(WorkflowRun, run_id)
+        run.status = "cancelled"
+        session.commit()
+
+    # Option 2: scoped transaction with automatic commit/rollback.
+    with Session(db.engine, expire_on_commit=False) as session, session.begin():
+        run = session.get(WorkflowRun, run_id)
+        run.status = "cancelled"
+
+    # Keep non-DB work outside transaction scope.
+    call_external_api()
+    ```
+
+### Enforce tenant_id scoping on shared-resource queries
+- Category: security
+- Severity: critical
+- Description: Reads and writes against shared tables must be scoped by `tenant_id` to prevent cross-tenant data leakage or corruption.
+- Suggested fix: Add `tenant_id` predicate to all tenant-owned entity queries and propagate tenant context through service/repository interfaces.
+- Example:
+  - Bad:
+    ```python
+    stmt = select(Workflow).where(Workflow.id == workflow_id)
+    workflow = session.execute(stmt).scalar_one_or_none()
+    ```
+  - Good:
+    ```python
+    stmt = select(Workflow).where(
+        Workflow.id == workflow_id,
+        Workflow.tenant_id == tenant_id,
+    )
+    workflow = session.execute(stmt).scalar_one_or_none()
+    ```
+
+### Prefer SQLAlchemy expressions over raw SQL by default
+- Category: maintainability
+- Severity: suggestion
+- Description: Raw SQL should be exceptional. ORM/Core expressions are easier to evolve, safer to compose, and more consistent with the codebase.
+- Suggested fix: Rewrite straightforward raw SQL into SQLAlchemy `select/update/delete` expressions; keep raw SQL only when required by clear technical constraints.
+- Example:
+  - Bad:
+    ```python
+    row = session.execute(
+        text("SELECT * FROM workflows WHERE id = :id AND tenant_id = :tenant_id"),
+        {"id": workflow_id, "tenant_id": tenant_id},
+    ).first()
+    ```
+  - Good:
+    ```python
+    stmt = select(Workflow).where(
+        Workflow.id == workflow_id,
+        Workflow.tenant_id == tenant_id,
+    )
+    row = session.execute(stmt).scalar_one_or_none()
+    ```
+
+### Protect write paths with concurrency safeguards
+- Category: quality
+- Severity: critical
+- Description: Multi-writer paths without explicit concurrency control can silently overwrite data. Choose the safeguard based on contention level, lock scope, and throughput cost instead of defaulting to one strategy.
+- Suggested fix:
+  - **Optimistic locking**: Use when contention is usually low and retries are acceptable. Add a version (or updated_at) guard in `WHERE` and treat `rowcount == 0` as a conflict.
+  - **Redis distributed lock**: Use when the critical section spans multiple steps/processes (or includes non-DB side effects) and you need cross-worker mutual exclusion.
+  - **SELECT ... FOR UPDATE**: Use when contention is high on the same rows and strict in-transaction serialization is required. Keep transactions short to reduce lock wait/deadlock risk.
+  - In all cases, scope by `tenant_id` and verify affected row counts for conditional writes.
+- Example:
+  - Bad:
+    ```python
+    # No tenant scope, no conflict detection, and no lock on a contested write path.
+    session.execute(update(WorkflowRun).where(WorkflowRun.id == run_id).values(status="cancelled"))
+    session.commit()  # silently overwrites concurrent updates
+    ```
+  - Good:
+    ```python
+    # 1) Optimistic lock (low contention, retry on conflict)
+    result = session.execute(
+        update(WorkflowRun)
+        .where(
+            WorkflowRun.id == run_id,
+            WorkflowRun.tenant_id == tenant_id,
+            WorkflowRun.version == expected_version,
+        )
+        .values(status="cancelled", version=WorkflowRun.version + 1)
+    )
+    if result.rowcount == 0:
+        raise WorkflowStateConflictError("stale version, retry")
+
+    # 2) Redis distributed lock (cross-worker critical section)
+    lock_name = f"workflow_run_lock:{tenant_id}:{run_id}"
+    with redis_client.lock(lock_name, timeout=20):
+        session.execute(
+            update(WorkflowRun)
+            .where(WorkflowRun.id == run_id, WorkflowRun.tenant_id == tenant_id)
+            .values(status="cancelled")
+        )
+        session.commit()
+
+    # 3) Pessimistic lock with SELECT ... FOR UPDATE (high contention)
+    run = session.execute(
+        select(WorkflowRun)
+        .where(WorkflowRun.id == run_id, WorkflowRun.tenant_id == tenant_id)
+        .with_for_update()
+    ).scalar_one()
+    run.status = "cancelled"
+    session.commit()
+    ```

.agents/skills/component-refactoring/SKILL.md

@@ -480,4 +480,4 @@ const useButtonState = () => {
 ### Related Skills
 
 - `frontend-testing` - For testing refactored components
-- `web/testing/testing.md` - Testing specification
+- `web/docs/test.md` - Testing specification

.agents/skills/frontend-testing/SKILL.md

@@ -7,7 +7,7 @@ description: Generate Vitest + React Testing Library tests for Dify frontend com
 
 This skill enables Claude to generate high-quality, comprehensive frontend tests for the Dify project following established conventions and best practices.
 
-> **⚠️ Authoritative Source**: This skill is derived from `web/testing/testing.md`. Use Vitest mock/timer APIs (`vi.*`).
+> **⚠️ Authoritative Source**: This skill is derived from `web/docs/test.md`. Use Vitest mock/timer APIs (`vi.*`).
 
 ## When to Apply This Skill
 
@@ -204,6 +204,16 @@ When assigned to test a directory/path, test **ALL content** within that path:
 
 > See [Test Structure Template](#test-structure-template) for correct import/mock patterns.
 
+### `nuqs` Query State Testing (Required for URL State Hooks)
+
+When a component or hook uses `useQueryState` / `useQueryStates`:
+
+- ✅ Use `NuqsTestingAdapter` (prefer shared helpers in `web/test/nuqs-testing.tsx`)
+- ✅ Assert URL synchronization via `onUrlUpdate` (`searchParams`, `options.history`)
+- ✅ For custom parsers (`createParser`), keep `parse` and `serialize` bijective and add round-trip edge cases (`%2F`, `%25`, spaces, legacy encoded values)
+- ✅ Verify default-clearing behavior (default values should be removed from URL when applicable)
+- ⚠️ Only mock `nuqs` directly when URL behavior is explicitly out of scope for the test
+
 ## Core Principles
 
 ### 1. AAA Pattern (Arrange-Act-Assert)
@@ -309,7 +319,7 @@ For more detailed information, refer to:
 
 ### Primary Specification (MUST follow)
 
-- **`web/testing/testing.md`** - The canonical testing specification. This skill is derived from this document.
+- **`web/docs/test.md`** - The canonical testing specification. This skill is derived from this document.
 
 ### Reference Examples in Codebase
 

.agents/skills/frontend-testing/references/checklist.md

@@ -80,6 +80,9 @@ Use this checklist when generating or reviewing tests for Dify frontend componen
 - [ ] Router mocks match actual Next.js API
 - [ ] Mocks reflect actual component conditional behavior
 - [ ] Only mock: API services, complex context providers, third-party libs
+- [ ] For `nuqs` URL-state tests, wrap with `NuqsTestingAdapter` (prefer `web/test/nuqs-testing.tsx`)
+- [ ] For `nuqs` URL-state tests, assert `onUrlUpdate` payload (`searchParams`, `options.history`)
+- [ ] If custom `nuqs` parser exists, add round-trip tests for encoded edge cases (`%2F`, `%25`, spaces, legacy encoded values)
 
 ### Queries
 

.agents/skills/frontend-testing/references/mocking.md

@@ -125,6 +125,31 @@ describe('Component', () => {
 })
 ```
 
+### 2.1 `nuqs` Query State (Preferred: Testing Adapter)
+
+For tests that validate URL query behavior, use `NuqsTestingAdapter` instead of mocking `nuqs` directly.
+
+```typescript
+import { renderHookWithNuqs } from '@/test/nuqs-testing'
+
+it('should sync query to URL with push history', async () => {
+  const { result, onUrlUpdate } = renderHookWithNuqs(() => useMyQueryState(), {
+    searchParams: '?page=1',
+  })
+
+  act(() => {
+    result.current.setQuery({ page: 2 })
+  })
+
+  await waitFor(() => expect(onUrlUpdate).toHaveBeenCalled())
+  const update = onUrlUpdate.mock.calls[onUrlUpdate.mock.calls.length - 1][0]
+  expect(update.options.history).toBe('push')
+  expect(update.searchParams.get('page')).toBe('2')
+})
+```
+
+Use direct `vi.mock('nuqs')` only when URL synchronization is intentionally out of scope.
+
 ### 3. Portal Components (with Shared State)
 
 ```typescript

.agents/skills/frontend-testing/references/workflow.md

@@ -4,7 +4,7 @@ This guide defines the workflow for generating tests, especially for complex com
 
 ## Scope Clarification
 
-This guide addresses **multi-file workflow** (how to process multiple test files). For coverage requirements within a single test file, see `web/testing/testing.md` § Coverage Goals.
+This guide addresses **multi-file workflow** (how to process multiple test files). For coverage requirements within a single test file, see `web/docs/test.md` § Coverage Goals.
 
 | Scope | Rule |
 |-------|------|

.agents/skills/orpc-contract-first/SKILL.md

@@ -1,43 +1,100 @@
 ---
 name: orpc-contract-first
-description: Guide for implementing oRPC contract-first API patterns in Dify frontend. Triggers when creating new API contracts, adding service endpoints, integrating TanStack Query with typed contracts, or migrating legacy service calls to oRPC. Use for all API layer work in web/contract and web/service directories.
+description: Guide for implementing oRPC contract-first API patterns in Dify frontend. Trigger when creating or updating contracts in web/contract, wiring router composition, integrating TanStack Query with typed contracts, migrating legacy service calls to oRPC, or deciding whether to call queryOptions directly vs extracting a helper or use-* hook in web/service.
 ---
 
 # oRPC Contract-First Development
 
-## Project Structure
+## Intent
 
-```
+- Keep contract as single source of truth in `web/contract/*`.
+- Default query usage: call-site `useQuery(consoleQuery|marketplaceQuery.xxx.queryOptions(...))` when endpoint behavior maps 1:1 to the contract.
+- Keep abstractions minimal and preserve TypeScript inference.
+
+## Minimal Structure
+
+```text
 web/contract/
-├── base.ts           # Base contract (inputStructure: 'detailed')
-├── router.ts         # Router composition & type exports
-├── marketplace.ts    # Marketplace contracts
-└── console/          # Console contracts by domain
-    ├── system.ts
-    └── billing.ts
+├── base.ts
+├── router.ts
+├── marketplace.ts
+└── console/
+    ├── billing.ts
+    └── ...other domains
+web/service/client.ts
 ```
 
-## Workflow
+## Core Workflow
 
-1. **Create contract** in `web/contract/console/{domain}.ts`
-   - Import `base` from `../base` and `type` from `@orpc/contract`
-   - Define route with `path`, `method`, `input`, `output`
+1. Define contract in `web/contract/console/{domain}.ts` or `web/contract/marketplace.ts`
+   - Use `base.route({...}).output(type<...>())` as baseline.
+   - Add `.input(type<...>())` only when request has `params/query/body`.
+   - For `GET` without input, omit `.input(...)` (do not use `.input(type<unknown>())`).
+2. Register contract in `web/contract/router.ts`
+   - Import directly from domain files and nest by API prefix.
+3. Consume from UI call sites via oRPC query utils.
 
-2. **Register in router** at `web/contract/router.ts`
-   - Import directly from domain file (no barrel files)
-   - Nest by API prefix: `billing: { invoices, bindPartnerStack }`
+```typescript
+import { useQuery } from '@tanstack/react-query'
+import { consoleQuery } from '@/service/client'
 
-3. **Create hooks** in `web/service/use-{domain}.ts`
-   - Use `consoleQuery.{group}.{contract}.queryKey()` for query keys
-   - Use `consoleClient.{group}.{contract}()` for API calls
+const invoiceQuery = useQuery(consoleQuery.billing.invoices.queryOptions({
+  staleTime: 5 * 60 * 1000,
+  throwOnError: true,
+  select: invoice => invoice.url,
+}))
+```
 
-## Key Rules
+## Query Usage Decision Rule
+
+1. Default: call site directly uses `*.queryOptions(...)`.
+2. If 3+ call sites share the same extra options (for example `retry: false`), extract a small queryOptions helper, not a `use-*` passthrough hook.
+3. Create `web/service/use-{domain}.ts` only for orchestration:
+   - Combine multiple queries/mutations.
+   - Share domain-level derived state or invalidation helpers.
+
+```typescript
+const invoicesBaseQueryOptions = () =>
+  consoleQuery.billing.invoices.queryOptions({ retry: false })
+
+const invoiceQuery = useQuery({
+  ...invoicesBaseQueryOptions(),
+  throwOnError: true,
+})
+```
+
+## Mutation Usage Decision Rule
+
+1. Default: call mutation helpers from `consoleQuery` / `marketplaceQuery`, for example `useMutation(consoleQuery.billing.bindPartnerStack.mutationOptions(...))`.
+2. If mutation flow is heavily custom, use oRPC clients as `mutationFn` (for example `consoleClient.xxx` / `marketplaceClient.xxx`), instead of generic handwritten non-oRPC mutation logic.
+
+## Key API Guide (`.key` vs `.queryKey` vs `.mutationKey`)
+
+- `.key(...)`:
+  - Use for partial matching operations (recommended for invalidation/refetch/cancel patterns).
+  - Example: `queryClient.invalidateQueries({ queryKey: consoleQuery.billing.key() })`
+- `.queryKey(...)`:
+  - Use for a specific query's full key (exact query identity / direct cache addressing).
+- `.mutationKey(...)`:
+  - Use for a specific mutation's full key.
+  - Typical use cases: mutation defaults registration, mutation-status filtering (`useIsMutating`, `queryClient.isMutating`), or explicit devtools grouping.
+
+## Anti-Patterns
+
+- Do not wrap `useQuery` with `options?: Partial<UseQueryOptions>`.
+- Do not split local `queryKey/queryFn` when oRPC `queryOptions` already exists and fits the use case.
+- Do not create thin `use-*` passthrough hooks for a single endpoint.
+- Reason: these patterns can degrade inference (`data` may become `unknown`, especially around `throwOnError`/`select`) and add unnecessary indirection.
+
+## Contract Rules
 
 - **Input structure**: Always use `{ params, query?, body? }` format
+- **No-input GET**: Omit `.input(...)`; do not use `.input(type<unknown>())`
 - **Path params**: Use `{paramName}` in path, match in `params` object
-- **Router nesting**: Group by API prefix (e.g., `/billing/*` → `billing: {}`)
+- **Router nesting**: Group by API prefix (e.g., `/billing/*` -> `billing: {}`)
 - **No barrel files**: Import directly from specific files
 - **Types**: Import from `@/types/`, use `type<T>()` helper
+- **Mutations**: Prefer `mutationOptions`; use explicit `mutationKey` mainly for defaults/filtering/devtools
 
 ## Type Export
 

.claude/skills/backend-code-review

@@ -0,0 +1 @@
+../../.agents/skills/backend-code-review

.codex/skills/component-refactoring

@@ -1 +0,0 @@
-../../.agents/skills/component-refactoring

.codex/skills/frontend-code-review

@@ -1 +0,0 @@
-../../.agents/skills/frontend-code-review

.codex/skills/frontend-testing

@@ -1 +0,0 @@
-../../.agents/skills/frontend-testing

.codex/skills/orpc-contract-first

@@ -1 +0,0 @@
-../../.agents/skills/orpc-contract-first

.devcontainer/post_create_command.sh

@@ -7,7 +7,7 @@ cd web && pnpm install
 pipx install uv
 
 echo "alias start-api=\"cd $WORKSPACE_ROOT/api && uv run python -m flask run --host 0.0.0.0 --port=5001 --debug\"" >> ~/.bashrc
-echo "alias start-worker=\"cd $WORKSPACE_ROOT/api && uv run python -m celery -A app.celery worker -P threads -c 1 --loglevel INFO -Q dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention\"" >> ~/.bashrc
+echo "alias start-worker=\"cd $WORKSPACE_ROOT/api && uv run python -m celery -A app.celery worker -P threads -c 1 --loglevel INFO -Q dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention\"" >> ~/.bashrc
 echo "alias start-web=\"cd $WORKSPACE_ROOT/web && pnpm dev:inspect\"" >> ~/.bashrc
 echo "alias start-web-prod=\"cd $WORKSPACE_ROOT/web && pnpm build && pnpm start\"" >> ~/.bashrc
 echo "alias start-containers=\"cd $WORKSPACE_ROOT/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env up -d\"" >> ~/.bashrc

.github/CODEOWNERS

@@ -9,6 +9,9 @@
 # CODEOWNERS file
 /.github/CODEOWNERS @laipz8200 @crazywoola
 
+# Agents
+/.agents/skills/ @hyoban
+
 # Docs
 /docs/ @crazywoola
 
@@ -21,6 +24,10 @@
 /api/services/tools/mcp_tools_manage_service.py @Nov1c444
 /api/controllers/mcp/ @Nov1c444
 /api/controllers/console/app/mcp_server.py @Nov1c444
+
+# Backend - Tests
+/api/tests/ @laipz8200 @QuantumGhost
+
 /api/tests/**/*mcp* @Nov1c444
 
 # Backend - Workflow - Engine (Core graph execution engine)
@@ -29,7 +36,7 @@
 /api/core/workflow/graph/ @laipz8200 @QuantumGhost
 /api/core/workflow/graph_events/ @laipz8200 @QuantumGhost
 /api/core/workflow/node_events/ @laipz8200 @QuantumGhost
-/api/core/model_runtime/ @laipz8200 @QuantumGhost
+/api/dify_graph/model_runtime/ @laipz8200 @QuantumGhost
 
 # Backend - Workflow - Nodes (Agent, Iteration, Loop, LLM)
 /api/core/workflow/nodes/agent/ @Nov1c444
@@ -231,6 +238,9 @@
 # Frontend - Base Components
 /web/app/components/base/ @iamjoel @zxhlyh
 
+# Frontend - Base Components Tests
+/web/app/components/base/**/*.spec.tsx @hyoban @CodingOnStar
+
 # Frontend - Utils and Hooks
 /web/utils/classnames.ts @iamjoel @zxhlyh
 /web/utils/time.ts @iamjoel @zxhlyh

.github/dependabot.yml

@@ -1,12 +1,43 @@
 version: 2
+
 updates:
+  - package-ecosystem: "pip"
+    directory: "/api"
+    open-pull-requests-limit: 2
+    schedule:
+      interval: "weekly"
+    groups:
+      python-dependencies:
+        patterns:
+          - "*"
+  - package-ecosystem: "uv"
+    directory: "/api"
+    open-pull-requests-limit: 2
+    schedule:
+      interval: "weekly"
+    groups:
+      uv-dependencies:
+        patterns:
+          - "*"
   - package-ecosystem: "npm"
     directory: "/web"
     schedule:
       interval: "weekly"
     open-pull-requests-limit: 2
-  - package-ecosystem: "uv"
-    directory: "/api"
-    schedule:
-      interval: "weekly"
-    open-pull-requests-limit: 2
+    groups:
+      lexical:
+        patterns:
+          - "lexical"
+          - "@lexical/*"
+      storybook:
+        patterns:
+          - "storybook"
+          - "@storybook/*"
+      npm-dependencies:
+        patterns:
+          - "*"
+        exclude-patterns:
+          - "lexical"
+          - "@lexical/*"
+          - "storybook"
+          - "@storybook/*"

.github/workflows/api-tests.yml

@@ -72,6 +72,7 @@ jobs:
           OPENDAL_FS_ROOT: /tmp/dify-storage
         run: |
           uv run --project api pytest \
+            -n auto \
             --timeout "${PYTEST_TIMEOUT:-180}" \
             api/tests/integration_tests/workflow \
             api/tests/integration_tests/tools \

.github/workflows/autofix.yml

@@ -79,29 +79,6 @@ jobs:
           find . -name "*.py" -type f -exec sed -i.bak -E 's/"([^"]+)" \| None/Optional["\1"]/g; s/'"'"'([^'"'"']+)'"'"' \| None/Optional['"'"'\1'"'"']/g' {} \;
           find . -name "*.py.bak" -type f -delete
 
-      - name: Install pnpm
-        uses: pnpm/action-setup@v4
-        with:
-          package_json_file: web/package.json
-          run_install: false
-
-      - name: Setup Node.js
-        uses: actions/setup-node@v6
-        with:
-          node-version: 24
-          cache: pnpm
-          cache-dependency-path: ./web/pnpm-lock.yaml
-
-      - name: Install web dependencies
-        run: |
-          cd web
-          pnpm install --frozen-lockfile
-
-      - name: ESLint autofix
-        run: |
-          cd web
-          pnpm lint:fix || true
-
       # mdformat breaks YAML front matter in markdown files. Add --exclude for directories containing YAML front matter.
       - name: mdformat
         run: |

.github/workflows/build-push.yml

@@ -8,6 +8,7 @@ on:
       - "build/**"
       - "release/e-*"
       - "hotfix/**"
+      - "feat/hitl-backend"
     tags:
       - "*"
 

.github/workflows/deploy-hitl.yml

@@ -4,8 +4,7 @@ on:
   workflow_run:
     workflows: ["Build and Push API & Web"]
     branches:
-      - "feat/hitl-frontend"
-      - "feat/hitl-backend"
+      - "build/feat/hitl"
     types:
       - completed
 
@@ -14,10 +13,7 @@ jobs:
     runs-on: ubuntu-latest
     if: |
       github.event.workflow_run.conclusion == 'success' &&
-      (
-        github.event.workflow_run.head_branch == 'feat/hitl-frontend' ||
-        github.event.workflow_run.head_branch == 'feat/hitl-backend'
-      )
+      github.event.workflow_run.head_branch == 'build/feat/hitl'
     steps:
       - name: Deploy to server
         uses: appleboy/ssh-action@v1

.github/workflows/pyrefly-diff-comment.yml

@@ -0,0 +1,88 @@
+name: Comment with Pyrefly Diff
+
+on:
+  workflow_run:
+    workflows:
+      - Pyrefly Diff Check
+    types:
+      - completed
+
+permissions: {}
+
+jobs:
+  comment:
+    name: Comment PR with pyrefly diff
+    runs-on: ubuntu-latest
+    permissions:
+      actions: read
+      contents: read
+      issues: write
+      pull-requests: write
+    if: ${{ github.event.workflow_run.conclusion == 'success' && github.event.workflow_run.pull_requests[0].head.repo.full_name != github.repository }}
+    steps:
+      - name: Download pyrefly diff artifact
+        uses: actions/github-script@v8
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          script: |
+            const fs = require('fs');
+            const artifacts = await github.rest.actions.listWorkflowRunArtifacts({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              run_id: ${{ github.event.workflow_run.id }},
+            });
+            const match = artifacts.data.artifacts.find((artifact) =>
+              artifact.name === 'pyrefly_diff'
+            );
+            if (!match) {
+              throw new Error('pyrefly_diff artifact not found');
+            }
+            const download = await github.rest.actions.downloadArtifact({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              artifact_id: match.id,
+              archive_format: 'zip',
+            });
+            fs.writeFileSync('pyrefly_diff.zip', Buffer.from(download.data));
+
+      - name: Unzip artifact
+        run: unzip -o pyrefly_diff.zip
+
+      - name: Post comment
+        uses: actions/github-script@v8
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          script: |
+            const fs = require('fs');
+            let diff = fs.readFileSync('pyrefly_diff.txt', { encoding: 'utf8' });
+            let prNumber = null;
+            try {
+              prNumber = parseInt(fs.readFileSync('pr_number.txt', { encoding: 'utf8' }), 10);
+            } catch (err) {
+              // Fallback to workflow_run payload if artifact is missing or incomplete.
+              const prs = context.payload.workflow_run.pull_requests || [];
+              if (prs.length > 0 && prs[0].number) {
+                prNumber = prs[0].number;
+              }
+            }
+            if (!prNumber) {
+              throw new Error('PR number not found in artifact or workflow_run payload');
+            }
+
+            const MAX_CHARS = 65000;
+            if (diff.length > MAX_CHARS) {
+              diff = diff.slice(0, MAX_CHARS);
+              diff = diff.slice(0, diff.lastIndexOf('\\n'));
+              diff += '\\n\\n... (truncated) ...';
+            }
+
+            const body = diff.trim()
+              ? '### Pyrefly Diff\n<details>\n<summary>base → PR</summary>\n\n```diff\n' + diff + '\n```\n</details>'
+              : '### Pyrefly Diff\nNo changes detected.';
+
+            await github.rest.issues.createComment({
+              issue_number: prNumber,
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              body,
+            });

.github/workflows/pyrefly-diff.yml

@@ -0,0 +1,100 @@
+name: Pyrefly Diff Check
+
+on:
+  pull_request:
+    paths:
+      - 'api/**/*.py'
+
+permissions:
+  contents: read
+
+jobs:
+  pyrefly-diff:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      issues: write
+      pull-requests: write
+    steps:
+      - name: Checkout PR branch
+        uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+
+      - name: Setup Python & UV
+        uses: astral-sh/setup-uv@v5
+        with:
+          enable-cache: true
+
+      - name: Install dependencies
+        run: uv sync --project api --dev
+
+      - name: Prepare diagnostics extractor
+        run: |
+          git show ${{ github.event.pull_request.head.sha }}:api/libs/pyrefly_diagnostics.py > /tmp/pyrefly_diagnostics.py
+
+      - name: Run pyrefly on PR branch
+        run: |
+          uv run --directory api --dev pyrefly check 2>&1 \
+            | uv run --directory api python /tmp/pyrefly_diagnostics.py > /tmp/pyrefly_pr.txt || true
+
+      - name: Checkout base branch
+        run: git checkout ${{ github.base_ref }}
+
+      - name: Run pyrefly on base branch
+        run: |
+          uv run --directory api --dev pyrefly check 2>&1 \
+            | uv run --directory api python /tmp/pyrefly_diagnostics.py > /tmp/pyrefly_base.txt || true
+
+      - name: Compute diff
+        run: |
+          diff -u /tmp/pyrefly_base.txt /tmp/pyrefly_pr.txt > pyrefly_diff.txt || true
+
+      - name: Save PR number
+        run: |
+          echo ${{ github.event.pull_request.number }} > pr_number.txt
+
+      - name: Upload pyrefly diff
+        uses: actions/upload-artifact@v4
+        with:
+          name: pyrefly_diff
+          path: |
+            pyrefly_diff.txt
+            pr_number.txt
+
+      - name: Comment PR with pyrefly diff
+        if: ${{ github.event.pull_request.head.repo.full_name == github.repository }}
+        uses: actions/github-script@v8
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          script: |
+            const fs = require('fs');
+            let diff = fs.readFileSync('pyrefly_diff.txt', { encoding: 'utf8' });
+            const prNumber = context.payload.pull_request.number;
+
+            const MAX_CHARS = 65000;
+            if (diff.length > MAX_CHARS) {
+              diff = diff.slice(0, MAX_CHARS);
+              diff = diff.slice(0, diff.lastIndexOf('\n'));
+              diff += '\n\n... (truncated) ...';
+            }
+
+            const body = diff.trim()
+              ? [
+                  '### Pyrefly Diff',
+                  '<details>',
+                  '<summary>base → PR</summary>',
+                  '',
+                  '```diff',
+                  diff,
+                  '```',
+                  '</details>',
+                ].join('\n')
+              : '### Pyrefly Diff\nNo changes detected.';
+
+            await github.rest.issues.createComment({
+              issue_number: prNumber,
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              body,
+            });

.github/workflows/style.yml

@@ -47,13 +47,9 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         run: uv run --directory api --dev lint-imports
 
-      - name: Run Basedpyright Checks
+      - name: Run Type Checks
         if: steps.changed-files.outputs.any_changed == 'true'
-        run: dev/basedpyright-check
-
-      - name: Run Mypy Type Checks
-        if: steps.changed-files.outputs.any_changed == 'true'
-        run: uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --check-untyped-defs --disable-error-code=import-untyped .
+        run: make type-check
 
       - name: Dotenv check
         if: steps.changed-files.outputs.any_changed == 'true'
@@ -93,7 +89,7 @@ jobs:
         uses: actions/setup-node@v6
         if: steps.changed-files.outputs.any_changed == 'true'
         with:
-          node-version: 24
+          node-version: 22
           cache: pnpm
           cache-dependency-path: ./web/pnpm-lock.yaml
 

.github/workflows/tool-test-sdks.yaml

@@ -28,7 +28,7 @@ jobs:
       - name: Use Node.js
         uses: actions/setup-node@v6
         with:
-          node-version: 24
+          node-version: 22
           cache: ''
           cache-dependency-path: 'pnpm-lock.yaml'
 

.github/workflows/translate-i18n-claude.yml

@@ -57,7 +57,7 @@ jobs:
       - name: Set up Node.js
         uses: actions/setup-node@v6
         with:
-          node-version: 24
+          node-version: 22
           cache: pnpm
           cache-dependency-path: ./web/pnpm-lock.yaml
 

.github/workflows/web-tests.yml

@@ -3,13 +3,65 @@ name: Web Tests
 on:
   workflow_call:
 
+permissions:
+  contents: read
+
 concurrency:
   group: web-tests-${{ github.head_ref || github.run_id }}
   cancel-in-progress: true
 
 jobs:
   test:
-    name: Web Tests
+    name: Web Tests (${{ matrix.shardIndex }}/${{ matrix.shardTotal }})
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        shardIndex: [1, 2, 3, 4]
+        shardTotal: [4]
+    defaults:
+      run:
+        shell: bash
+        working-directory: ./web
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v6
+        with:
+          persist-credentials: false
+
+      - name: Install pnpm
+        uses: pnpm/action-setup@v4
+        with:
+          package_json_file: web/package.json
+          run_install: false
+
+      - name: Setup Node.js
+        uses: actions/setup-node@v6
+        with:
+          node-version: 22
+          cache: pnpm
+          cache-dependency-path: ./web/pnpm-lock.yaml
+
+      - name: Install dependencies
+        run: pnpm install --frozen-lockfile
+
+      - name: Run tests
+        run: pnpm vitest run --reporter=blob --shard=${{ matrix.shardIndex }}/${{ matrix.shardTotal }} --coverage
+
+      - name: Upload blob report
+        if: ${{ !cancelled() }}
+        uses: actions/upload-artifact@v6
+        with:
+          name: blob-report-${{ matrix.shardIndex }}
+          path: web/.vitest-reports/*
+          include-hidden-files: true
+          retention-days: 1
+
+  merge-reports:
+    name: Merge Test Reports
+    if: ${{ !cancelled() }}
+    needs: [test]
     runs-on: ubuntu-latest
     defaults:
       run:
@@ -31,15 +83,22 @@ jobs:
       - name: Setup Node.js
         uses: actions/setup-node@v6
         with:
-          node-version: 24
+          node-version: 22
           cache: pnpm
           cache-dependency-path: ./web/pnpm-lock.yaml
 
       - name: Install dependencies
         run: pnpm install --frozen-lockfile
 
-      - name: Run tests
-        run: pnpm test:coverage
+      - name: Download blob reports
+        uses: actions/download-artifact@v6
+        with:
+          path: web/.vitest-reports
+          pattern: blob-report-*
+          merge-multiple: true
+
+      - name: Merge reports
+        run: pnpm vitest --merge-reports --coverage --silent=passed-only
 
       - name: Coverage Summary
         if: always()
@@ -398,7 +457,7 @@ jobs:
         uses: actions/setup-node@v6
         if: steps.changed-files.outputs.any_changed == 'true'
         with:
-          node-version: 24
+          node-version: 22
           cache: pnpm
           cache-dependency-path: ./web/pnpm-lock.yaml
 

.gitignore

@@ -222,6 +222,7 @@ mise.toml
 
 # AI Assistant
 .roo/
+/.claude/worktrees/
 api/.env.backup
 /clickzetta
 

.vscode/launch.json.template

@@ -37,7 +37,7 @@
                 "-c",
                 "1",
                 "-Q",
-                "dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention",
+                "dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention,workflow_based_app_execution",
                 "--loglevel",
                 "INFO"
             ],

AGENTS.md

@@ -7,7 +7,7 @@ Dify is an open-source platform for developing LLM applications with an intuitiv
 The codebase is split into:
 
 - **Backend API** (`/api`): Python Flask application organized with Domain-Driven Design
-- **Frontend Web** (`/web`): Next.js 15 application using TypeScript and React 19
+- **Frontend Web** (`/web`): Next.js application using TypeScript and React
 - **Docker deployment** (`/docker`): Containerized deployment configurations
 
 ## Backend Workflow
@@ -18,36 +18,7 @@ The codebase is split into:
 
 ## Frontend Workflow
 
-```bash
-cd web
-pnpm lint:fix
-pnpm type-check:tsgo
-pnpm test
-```
-
-### Frontend Linting
-
-ESLint is used for frontend code quality. Available commands:
-
-```bash
-# Lint all files (report only)
-pnpm lint
-
-# Lint and auto-fix issues
-pnpm lint:fix
-
-# Lint specific files or directories
-pnpm lint:fix app/components/base/button/
-pnpm lint:fix app/components/base/button/index.tsx
-
-# Lint quietly (errors only, no warnings)
-pnpm lint:quiet
-
-# Check code complexity
-pnpm lint:complexity
-```
-
-**Important**: Always run `pnpm lint:fix` before committing. The pre-commit hook runs `lint-staged` which only lints staged files.
+- Read `web/AGENTS.md` for details
 
 ## Testing & Quality Practices
 
@@ -58,7 +29,7 @@ pnpm lint:complexity
 
 ## Language Style
 
-- **Python**: Keep type hints on functions and attributes, and implement relevant special methods (e.g., `__repr__`, `__str__`).
+- **Python**: Keep type hints on functions and attributes, and implement relevant special methods (e.g., `__repr__`, `__str__`). Prefer `TypedDict` over `dict` or `Mapping` for type safety and better code documentation.
 - **TypeScript**: Use the strict config, rely on ESLint (`pnpm lint:fix` preferred) plus `pnpm type-check:tsgo`, and avoid `any` types.
 
 ## General Practices

CONTRIBUTING.md

@@ -77,7 +77,7 @@ How we prioritize:
 
 For setting up the frontend service, please refer to our comprehensive [guide](https://github.com/langgenius/dify/blob/main/web/README.md) in the `web/README.md` file. This document provides detailed instructions to help you set up the frontend environment properly.
 
-**Testing**: All React components must have comprehensive test coverage. See [web/testing/testing.md](https://github.com/langgenius/dify/blob/main/web/testing/testing.md) for the canonical frontend testing guidelines and follow every requirement described there.
+**Testing**: All React components must have comprehensive test coverage. See [web/docs/test.md](https://github.com/langgenius/dify/blob/main/web/docs/test.md) for the canonical frontend testing guidelines and follow every requirement described there.
 
 #### Backend
 

Makefile

@@ -68,9 +68,11 @@ lint:
 	@echo "✅ Linting complete"
 
 type-check:
-	@echo "📝 Running type check with basedpyright..."
-	@uv run --directory api --dev basedpyright
-	@echo "✅ Type check complete"
+	@echo "📝 Running type checks (basedpyright + pyrefly + mypy)..."
+	@./dev/basedpyright-check $(PATH_TO_CHECK)
+	@./dev/pyrefly-check-local
+	@uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --check-untyped-defs --disable-error-code=import-untyped .
+	@echo "✅ Type checks complete"
 
 test:
 	@echo "🧪 Running backend unit tests..."
@@ -78,7 +80,7 @@ test:
 		echo "Target: $(TARGET_TESTS)"; \
 		uv run --project api --dev pytest $(TARGET_TESTS); \
 	else \
-		uv run --project api --dev dev/pytest/pytest_unit_tests.sh; \
+		PYTEST_XDIST_ARGS="-n auto" uv run --project api --dev dev/pytest/pytest_unit_tests.sh; \
 	fi
 	@echo "✅ Tests complete"
 
@@ -130,7 +132,7 @@ help:
 	@echo "  make format         - Format code with ruff"
 	@echo "  make check          - Check code with ruff"
 	@echo "  make lint           - Format, fix, and lint code (ruff, imports, dotenv)"
-	@echo "  make type-check     - Run type checking with basedpyright"
+	@echo "  make type-check     - Run type checks (basedpyright, pyrefly, mypy)"
 	@echo "  make test           - Run backend unit tests (or TARGET_TESTS=./api/tests/<target_tests>)"
 	@echo ""
 	@echo "Docker Build Targets:"

README.md

@@ -1,9 +1,5 @@
 ![cover-v5-optimized](./images/GitHub_README_if.png)
 
-<p align="center">
-  📌 <a href="https://dify.ai/blog/introducing-dify-workflow-file-upload-a-demo-on-ai-podcast">Introducing Dify Workflow File Upload: Recreate Google NotebookLM Podcast</a>
-</p>
-
 <p align="center">
   <a href="https://cloud.dify.ai">Dify Cloud</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">Self-hosting</a> ·

api/.env.example

@@ -42,6 +42,8 @@ REFRESH_TOKEN_EXPIRE_DAYS=30
 # redis configuration
 REDIS_HOST=localhost
 REDIS_PORT=6379
+# Optional: limit total connections in connection pool (unset for default)
+# REDIS_MAX_CONNECTIONS=200
 REDIS_USERNAME=
 REDIS_PASSWORD=difyai123456
 REDIS_USE_SSL=false
@@ -553,6 +555,8 @@ WORKFLOW_LOG_CLEANUP_ENABLED=false
 WORKFLOW_LOG_RETENTION_DAYS=30
 # Batch size for workflow log cleanup operations (default: 100)
 WORKFLOW_LOG_CLEANUP_BATCH_SIZE=100
+# Comma-separated list of workflow IDs to clean logs for
+WORKFLOW_LOG_CLEANUP_SPECIFIC_WORKFLOW_IDS=
 
 # App configuration
 APP_MAX_EXECUTION_TIME=1200
@@ -617,6 +621,7 @@ PLUGIN_DAEMON_URL=http://127.0.0.1:5002
 PLUGIN_REMOTE_INSTALL_PORT=5003
 PLUGIN_REMOTE_INSTALL_HOST=localhost
 PLUGIN_MAX_PACKAGE_SIZE=15728640
+PLUGIN_MODEL_SCHEMA_CACHE_TTL=3600
 INNER_API_KEY_FOR_PLUGIN=QaHbTe77CtuXmsfyhR7+vRjI/+XbV1AaFy691iy+kGDv2Jvy0/eAh8Y1
 
 # Marketplace configuration
@@ -714,6 +719,31 @@ ANNOTATION_IMPORT_MAX_CONCURRENT=5
 # Sandbox expired records clean configuration
 SANDBOX_EXPIRED_RECORDS_CLEAN_GRACEFUL_PERIOD=21
 SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_SIZE=1000
+SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_MAX_INTERVAL=200
 SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS=30
 SANDBOX_EXPIRED_RECORDS_CLEAN_TASK_LOCK_TTL=90000
 
+
+# Redis URL used for PubSub between API and
+# celery worker
+# defaults to url constructed from `REDIS_*`
+# configurations
+PUBSUB_REDIS_URL=
+# Pub/sub channel type for streaming events.
+# valid options are:
+#
+# - pubsub: for normal Pub/Sub
+# - sharded: for sharded Pub/Sub
+#
+# It's highly recommended to use sharded Pub/Sub AND redis cluster
+# for large deployments.
+PUBSUB_REDIS_CHANNEL_TYPE=pubsub
+# Whether to use Redis cluster mode while running
+# PubSub.
+#  It's highly recommended to enable this for large deployments.
+PUBSUB_REDIS_USE_CLUSTERS=false
+
+# Whether to Enable human input timeout check task
+ENABLE_HUMAN_INPUT_TIMEOUT_TASK=true
+# Human input timeout check interval in minutes
+HUMAN_INPUT_TIMEOUT_TASK_INTERVAL=1

api/.importlinter

@@ -1,6 +1,7 @@
 [importlinter]
 root_packages =
     core
+    dify_graph
     configs
     controllers
     extensions
@@ -21,49 +22,39 @@ layers =
     runtime
     entities
 containers =
-    core.workflow
+    dify_graph
 ignore_imports =
-    core.workflow.nodes.base.node -> core.workflow.graph_events
-    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph_events
-    core.workflow.nodes.loop.loop_node -> core.workflow.graph_events
+    dify_graph.nodes.base.node -> dify_graph.graph_events
+    dify_graph.nodes.iteration.iteration_node -> dify_graph.graph_events
+    dify_graph.nodes.loop.loop_node -> dify_graph.graph_events
 
-    core.workflow.nodes.iteration.iteration_node -> core.app.workflow.node_factory
-    core.workflow.nodes.loop.loop_node -> core.app.workflow.node_factory
-
-    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph_engine
-    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph
-    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph_engine.command_channels
-    core.workflow.nodes.loop.loop_node -> core.workflow.graph_engine
-    core.workflow.nodes.loop.loop_node -> core.workflow.graph
-    core.workflow.nodes.loop.loop_node -> core.workflow.graph_engine.command_channels
+    dify_graph.nodes.iteration.iteration_node -> dify_graph.graph_engine
+    dify_graph.nodes.loop.loop_node -> dify_graph.graph_engine
+    # TODO(QuantumGhost): fix the import violation later
+    dify_graph.entities.pause_reason -> dify_graph.nodes.human_input.entities
 
 [importlinter:contract:workflow-infrastructure-dependencies]
 name = Workflow Infrastructure Dependencies
 type = forbidden
 source_modules =
-    core.workflow
+    dify_graph
 forbidden_modules =
     extensions.ext_database
     extensions.ext_redis
 allow_indirect_imports = True
 ignore_imports =
-    core.workflow.nodes.agent.agent_node -> extensions.ext_database
-    core.workflow.nodes.datasource.datasource_node -> extensions.ext_database
-    core.workflow.nodes.knowledge_index.knowledge_index_node -> extensions.ext_database
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> extensions.ext_database
-    core.workflow.nodes.llm.file_saver -> extensions.ext_database
-    core.workflow.nodes.llm.llm_utils -> extensions.ext_database
-    core.workflow.nodes.llm.node -> extensions.ext_database
-    core.workflow.nodes.tool.tool_node -> extensions.ext_database
-    core.workflow.graph_engine.command_channels.redis_channel -> extensions.ext_redis
-    core.workflow.graph_engine.manager -> extensions.ext_redis
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> extensions.ext_redis
+    dify_graph.nodes.agent.agent_node -> extensions.ext_database
+    dify_graph.nodes.llm.file_saver -> extensions.ext_database
+    dify_graph.nodes.llm.node -> extensions.ext_database
+    dify_graph.nodes.tool.tool_node -> extensions.ext_database
+    dify_graph.model_runtime.model_providers.__base.ai_model -> extensions.ext_redis
+    dify_graph.model_runtime.model_providers.model_provider_factory -> extensions.ext_redis
 
 [importlinter:contract:workflow-external-imports]
 name = Workflow External Imports
 type = forbidden
 source_modules =
-    core.workflow
+    dify_graph
 forbidden_modules =
     configs
     controllers
@@ -89,7 +80,6 @@ forbidden_modules =
     core.logging
     core.mcp
     core.memory
-    core.model_manager
     core.moderation
     core.ops
     core.plugin
@@ -102,203 +92,63 @@ forbidden_modules =
     core.trigger
     core.variables
 ignore_imports =
-    core.workflow.nodes.loop.loop_node -> core.app.workflow.node_factory
-    core.workflow.graph_engine.command_channels.redis_channel -> extensions.ext_redis
-    core.workflow.workflow_entry -> core.app.workflow.layers.observability
-    core.workflow.nodes.agent.agent_node -> core.model_manager
-    core.workflow.nodes.agent.agent_node -> core.provider_manager
-    core.workflow.nodes.agent.agent_node -> core.tools.tool_manager
-    core.workflow.nodes.code.code_node -> core.helper.code_executor.code_executor
-    core.workflow.nodes.datasource.datasource_node -> models.model
-    core.workflow.nodes.datasource.datasource_node -> models.tools
-    core.workflow.nodes.datasource.datasource_node -> services.datasource_provider_service
-    core.workflow.nodes.document_extractor.node -> configs
-    core.workflow.nodes.document_extractor.node -> core.file.file_manager
-    core.workflow.nodes.document_extractor.node -> core.helper.ssrf_proxy
-    core.workflow.nodes.http_request.entities -> configs
-    core.workflow.nodes.http_request.executor -> configs
-    core.workflow.nodes.http_request.executor -> core.file.file_manager
-    core.workflow.nodes.http_request.node -> configs
-    core.workflow.nodes.http_request.node -> core.tools.tool_file_manager
-    core.workflow.nodes.iteration.iteration_node -> core.app.workflow.node_factory
-    core.workflow.nodes.knowledge_index.knowledge_index_node -> core.rag.index_processor.index_processor_factory
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.rag.datasource.retrieval_service
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.rag.retrieval.dataset_retrieval
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> models.dataset
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> services.feature_service
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.model_runtime.model_providers.__base.large_language_model
-    core.workflow.nodes.llm.llm_utils -> configs
-    core.workflow.nodes.llm.llm_utils -> core.app.entities.app_invoke_entities
-    core.workflow.nodes.llm.llm_utils -> core.file.models
-    core.workflow.nodes.llm.llm_utils -> core.model_manager
-    core.workflow.nodes.llm.llm_utils -> core.model_runtime.model_providers.__base.large_language_model
-    core.workflow.nodes.llm.llm_utils -> models.model
-    core.workflow.nodes.llm.llm_utils -> models.provider
-    core.workflow.nodes.llm.llm_utils -> services.credit_pool_service
-    core.workflow.nodes.llm.node -> core.tools.signature
-    core.workflow.nodes.template_transform.template_transform_node -> configs
-    core.workflow.nodes.tool.tool_node -> core.callback_handler.workflow_tool_callback_handler
-    core.workflow.nodes.tool.tool_node -> core.tools.tool_engine
-    core.workflow.nodes.tool.tool_node -> core.tools.tool_manager
-    core.workflow.workflow_entry -> configs
-    core.workflow.workflow_entry -> models.workflow
-    core.workflow.nodes.agent.agent_node -> core.agent.entities
-    core.workflow.nodes.agent.agent_node -> core.agent.plugin_entities
-    core.workflow.nodes.base.node -> core.app.entities.app_invoke_entities
-    core.workflow.nodes.knowledge_index.knowledge_index_node -> core.app.entities.app_invoke_entities
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.app.app_config.entities
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.app.entities.app_invoke_entities
-    core.workflow.nodes.llm.node -> core.app.entities.app_invoke_entities
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.app.entities.app_invoke_entities
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.advanced_prompt_transform
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.simple_prompt_transform
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_runtime.model_providers.__base.large_language_model
-    core.workflow.nodes.question_classifier.question_classifier_node -> core.app.entities.app_invoke_entities
-    core.workflow.nodes.question_classifier.question_classifier_node -> core.prompt.advanced_prompt_transform
-    core.workflow.nodes.question_classifier.question_classifier_node -> core.prompt.simple_prompt_transform
-    core.workflow.nodes.start.entities -> core.app.app_config.entities
-    core.workflow.nodes.start.start_node -> core.app.app_config.entities
-    core.workflow.workflow_entry -> core.app.apps.exc
-    core.workflow.workflow_entry -> core.app.entities.app_invoke_entities
-    core.workflow.workflow_entry -> core.app.workflow.node_factory
-    core.workflow.nodes.datasource.datasource_node -> core.datasource.datasource_manager
-    core.workflow.nodes.datasource.datasource_node -> core.datasource.utils.message_transformer
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.entities.agent_entities
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.entities.model_entities
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.model_manager
-    core.workflow.nodes.llm.llm_utils -> core.entities.provider_entities
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_manager
-    core.workflow.nodes.question_classifier.question_classifier_node -> core.model_manager
-    core.workflow.node_events.node -> core.file
-    core.workflow.nodes.agent.agent_node -> core.file
-    core.workflow.nodes.datasource.datasource_node -> core.file
-    core.workflow.nodes.datasource.datasource_node -> core.file.enums
-    core.workflow.nodes.document_extractor.node -> core.file
-    core.workflow.nodes.http_request.executor -> core.file.enums
-    core.workflow.nodes.http_request.node -> core.file
-    core.workflow.nodes.http_request.node -> core.file.file_manager
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.file.models
-    core.workflow.nodes.list_operator.node -> core.file
-    core.workflow.nodes.llm.file_saver -> core.file
-    core.workflow.nodes.llm.llm_utils -> core.variables.segments
-    core.workflow.nodes.llm.node -> core.file
-    core.workflow.nodes.llm.node -> core.file.file_manager
-    core.workflow.nodes.llm.node -> core.file.models
-    core.workflow.nodes.loop.entities -> core.variables.types
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.file
-    core.workflow.nodes.protocols -> core.file
-    core.workflow.nodes.question_classifier.question_classifier_node -> core.file.models
-    core.workflow.nodes.tool.tool_node -> core.file
-    core.workflow.nodes.tool.tool_node -> core.tools.utils.message_transformer
-    core.workflow.nodes.tool.tool_node -> models
-    core.workflow.nodes.trigger_webhook.node -> core.file
-    core.workflow.runtime.variable_pool -> core.file
-    core.workflow.runtime.variable_pool -> core.file.file_manager
-    core.workflow.system_variable -> core.file.models
-    core.workflow.utils.condition.processor -> core.file
-    core.workflow.utils.condition.processor -> core.file.file_manager
-    core.workflow.workflow_entry -> core.file.models
-    core.workflow.workflow_type_encoder -> core.file.models
-    core.workflow.nodes.agent.agent_node -> models.model
-    core.workflow.nodes.code.code_node -> core.helper.code_executor.code_node_provider
-    core.workflow.nodes.code.code_node -> core.helper.code_executor.javascript.javascript_code_provider
-    core.workflow.nodes.code.code_node -> core.helper.code_executor.python3.python3_code_provider
-    core.workflow.nodes.code.entities -> core.helper.code_executor.code_executor
-    core.workflow.nodes.datasource.datasource_node -> core.variables.variables
-    core.workflow.nodes.http_request.executor -> core.helper.ssrf_proxy
-    core.workflow.nodes.http_request.node -> core.helper.ssrf_proxy
-    core.workflow.nodes.llm.file_saver -> core.helper.ssrf_proxy
-    core.workflow.nodes.llm.node -> core.helper.code_executor
-    core.workflow.nodes.template_transform.template_renderer -> core.helper.code_executor.code_executor
-    core.workflow.nodes.llm.node -> core.llm_generator.output_parser.errors
-    core.workflow.nodes.llm.node -> core.llm_generator.output_parser.structured_output
-    core.workflow.nodes.llm.node -> core.model_manager
-    core.workflow.nodes.agent.entities -> core.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.prompt.simple_prompt_transform
-    core.workflow.nodes.llm.entities -> core.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.llm.llm_utils -> core.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.llm.node -> core.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.llm.node -> core.prompt.utils.prompt_message_util
-    core.workflow.nodes.parameter_extractor.entities -> core.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.utils.prompt_message_util
-    core.workflow.nodes.question_classifier.entities -> core.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.question_classifier.question_classifier_node -> core.prompt.utils.prompt_message_util
-    core.workflow.nodes.knowledge_index.entities -> core.rag.retrieval.retrieval_methods
-    core.workflow.nodes.knowledge_index.knowledge_index_node -> core.rag.retrieval.retrieval_methods
-    core.workflow.nodes.knowledge_index.knowledge_index_node -> models.dataset
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.rag.retrieval.retrieval_methods
-    core.workflow.nodes.llm.node -> models.dataset
-    core.workflow.nodes.agent.agent_node -> core.tools.utils.message_transformer
-    core.workflow.nodes.llm.file_saver -> core.tools.signature
-    core.workflow.nodes.llm.file_saver -> core.tools.tool_file_manager
-    core.workflow.nodes.tool.tool_node -> core.tools.errors
-    core.workflow.conversation_variable_updater -> core.variables
-    core.workflow.graph_engine.entities.commands -> core.variables.variables
-    core.workflow.nodes.agent.agent_node -> core.variables.segments
-    core.workflow.nodes.answer.answer_node -> core.variables
-    core.workflow.nodes.code.code_node -> core.variables.segments
-    core.workflow.nodes.code.code_node -> core.variables.types
-    core.workflow.nodes.code.entities -> core.variables.types
-    core.workflow.nodes.datasource.datasource_node -> core.variables.segments
-    core.workflow.nodes.document_extractor.node -> core.variables
-    core.workflow.nodes.document_extractor.node -> core.variables.segments
-    core.workflow.nodes.http_request.executor -> core.variables.segments
-    core.workflow.nodes.http_request.node -> core.variables.segments
-    core.workflow.nodes.iteration.iteration_node -> core.variables
-    core.workflow.nodes.iteration.iteration_node -> core.variables.segments
-    core.workflow.nodes.iteration.iteration_node -> core.variables.variables
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.variables
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.variables.segments
-    core.workflow.nodes.list_operator.node -> core.variables
-    core.workflow.nodes.list_operator.node -> core.variables.segments
-    core.workflow.nodes.llm.node -> core.variables
-    core.workflow.nodes.loop.loop_node -> core.variables
-    core.workflow.nodes.parameter_extractor.entities -> core.variables.types
-    core.workflow.nodes.parameter_extractor.exc -> core.variables.types
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.variables.types
-    core.workflow.nodes.tool.tool_node -> core.variables.segments
-    core.workflow.nodes.tool.tool_node -> core.variables.variables
-    core.workflow.nodes.trigger_webhook.node -> core.variables.types
-    core.workflow.nodes.trigger_webhook.node -> core.variables.variables
-    core.workflow.nodes.variable_aggregator.entities -> core.variables.types
-    core.workflow.nodes.variable_aggregator.variable_aggregator_node -> core.variables.segments
-    core.workflow.nodes.variable_assigner.common.helpers -> core.variables
-    core.workflow.nodes.variable_assigner.common.helpers -> core.variables.consts
-    core.workflow.nodes.variable_assigner.common.helpers -> core.variables.types
-    core.workflow.nodes.variable_assigner.v1.node -> core.variables
-    core.workflow.nodes.variable_assigner.v2.helpers -> core.variables
-    core.workflow.nodes.variable_assigner.v2.node -> core.variables
-    core.workflow.nodes.variable_assigner.v2.node -> core.variables.consts
-    core.workflow.runtime.graph_runtime_state_protocol -> core.variables.segments
-    core.workflow.runtime.read_only_wrappers -> core.variables.segments
-    core.workflow.runtime.variable_pool -> core.variables
-    core.workflow.runtime.variable_pool -> core.variables.consts
-    core.workflow.runtime.variable_pool -> core.variables.segments
-    core.workflow.runtime.variable_pool -> core.variables.variables
-    core.workflow.utils.condition.processor -> core.variables
-    core.workflow.utils.condition.processor -> core.variables.segments
-    core.workflow.variable_loader -> core.variables
-    core.workflow.variable_loader -> core.variables.consts
-    core.workflow.workflow_type_encoder -> core.variables
-    core.workflow.graph_engine.manager -> extensions.ext_redis
-    core.workflow.nodes.agent.agent_node -> extensions.ext_database
-    core.workflow.nodes.datasource.datasource_node -> extensions.ext_database
-    core.workflow.nodes.knowledge_index.knowledge_index_node -> extensions.ext_database
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> extensions.ext_database
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> extensions.ext_redis
-    core.workflow.nodes.llm.file_saver -> extensions.ext_database
-    core.workflow.nodes.llm.llm_utils -> extensions.ext_database
-    core.workflow.nodes.llm.node -> extensions.ext_database
-    core.workflow.nodes.tool.tool_node -> extensions.ext_database
-    core.workflow.workflow_entry -> extensions.otel.runtime
-    core.workflow.nodes.agent.agent_node -> models
-    core.workflow.nodes.base.node -> models.enums
-    core.workflow.nodes.llm.llm_utils -> models.provider_ids
-    core.workflow.nodes.llm.node -> models.model
-    core.workflow.workflow_entry -> models.enums
-    core.workflow.nodes.agent.agent_node -> services
-    core.workflow.nodes.tool.tool_node -> services
+    dify_graph.nodes.agent.agent_node -> core.model_manager
+    dify_graph.nodes.agent.agent_node -> core.provider_manager
+    dify_graph.nodes.agent.agent_node -> core.tools.tool_manager
+    dify_graph.nodes.llm.llm_utils -> core.model_manager
+    dify_graph.nodes.llm.protocols -> core.model_manager
+    dify_graph.nodes.llm.llm_utils -> dify_graph.model_runtime.model_providers.__base.large_language_model
+    dify_graph.nodes.llm.node -> core.tools.signature
+    dify_graph.nodes.tool.tool_node -> core.callback_handler.workflow_tool_callback_handler
+    dify_graph.nodes.tool.tool_node -> core.tools.tool_engine
+    dify_graph.nodes.tool.tool_node -> core.tools.tool_manager
+    dify_graph.nodes.agent.agent_node -> core.agent.entities
+    dify_graph.nodes.agent.agent_node -> core.agent.plugin_entities
+    dify_graph.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.app.app_config.entities
+    dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.advanced_prompt_transform
+    dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.simple_prompt_transform
+    dify_graph.nodes.parameter_extractor.parameter_extractor_node -> dify_graph.model_runtime.model_providers.__base.large_language_model
+    dify_graph.nodes.question_classifier.question_classifier_node -> core.prompt.simple_prompt_transform
+    dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.model_manager
+    dify_graph.nodes.question_classifier.question_classifier_node -> core.model_manager
+    dify_graph.nodes.tool.tool_node -> core.tools.utils.message_transformer
+    dify_graph.nodes.tool.tool_node -> models
+    dify_graph.nodes.agent.agent_node -> models.model
+    dify_graph.nodes.llm.file_saver -> core.helper.ssrf_proxy
+    dify_graph.nodes.llm.node -> core.helper.code_executor
+    dify_graph.nodes.llm.node -> core.llm_generator.output_parser.errors
+    dify_graph.nodes.llm.node -> core.llm_generator.output_parser.structured_output
+    dify_graph.nodes.llm.node -> core.model_manager
+    dify_graph.nodes.agent.entities -> core.prompt.entities.advanced_prompt_entities
+    dify_graph.nodes.llm.entities -> core.prompt.entities.advanced_prompt_entities
+    dify_graph.nodes.llm.node -> core.prompt.entities.advanced_prompt_entities
+    dify_graph.nodes.llm.node -> core.prompt.utils.prompt_message_util
+    dify_graph.nodes.parameter_extractor.entities -> core.prompt.entities.advanced_prompt_entities
+    dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.entities.advanced_prompt_entities
+    dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.utils.prompt_message_util
+    dify_graph.nodes.question_classifier.entities -> core.prompt.entities.advanced_prompt_entities
+    dify_graph.nodes.question_classifier.question_classifier_node -> core.prompt.utils.prompt_message_util
+    dify_graph.nodes.knowledge_index.entities -> core.rag.retrieval.retrieval_methods
+    dify_graph.nodes.llm.node -> models.dataset
+    dify_graph.nodes.agent.agent_node -> core.tools.utils.message_transformer
+    dify_graph.nodes.llm.file_saver -> core.tools.signature
+    dify_graph.nodes.llm.file_saver -> core.tools.tool_file_manager
+    dify_graph.nodes.tool.tool_node -> core.tools.errors
+    dify_graph.nodes.agent.agent_node -> extensions.ext_database
+    dify_graph.nodes.llm.file_saver -> extensions.ext_database
+    dify_graph.nodes.llm.node -> extensions.ext_database
+    dify_graph.nodes.tool.tool_node -> extensions.ext_database
+    dify_graph.nodes.agent.agent_node -> models
+    dify_graph.nodes.llm.node -> models.model
+    dify_graph.nodes.agent.agent_node -> services
+    dify_graph.nodes.tool.tool_node -> services
+    dify_graph.model_runtime.model_providers.__base.ai_model -> configs
+    dify_graph.model_runtime.model_providers.__base.ai_model -> extensions.ext_redis
+    dify_graph.model_runtime.model_providers.__base.large_language_model -> configs
+    dify_graph.model_runtime.model_providers.__base.text_embedding_model -> core.entities.embedding_type
+    dify_graph.model_runtime.model_providers.model_provider_factory -> configs
+    dify_graph.model_runtime.model_providers.model_provider_factory -> extensions.ext_redis
+    dify_graph.model_runtime.model_providers.model_provider_factory -> models.provider_ids
 
 [importlinter:contract:rsc]
 name = RSC
@@ -307,7 +157,7 @@ layers =
     graph_engine
     response_coordinator
 containers =
-    core.workflow.graph_engine
+    dify_graph.graph_engine
 
 [importlinter:contract:worker]
 name = Worker
@@ -316,7 +166,7 @@ layers =
     graph_engine
     worker
 containers =
-    core.workflow.graph_engine
+    dify_graph.graph_engine
 
 [importlinter:contract:graph-engine-architecture]
 name = Graph Engine Architecture
@@ -332,28 +182,28 @@ layers =
     worker_management
     domain
 containers =
-    core.workflow.graph_engine
+    dify_graph.graph_engine
 
 [importlinter:contract:domain-isolation]
 name = Domain Model Isolation
 type = forbidden
 source_modules =
-    core.workflow.graph_engine.domain
+    dify_graph.graph_engine.domain
 forbidden_modules =
-    core.workflow.graph_engine.worker_management
-    core.workflow.graph_engine.command_channels
-    core.workflow.graph_engine.layers
-    core.workflow.graph_engine.protocols
+    dify_graph.graph_engine.worker_management
+    dify_graph.graph_engine.command_channels
+    dify_graph.graph_engine.layers
+    dify_graph.graph_engine.protocols
 
 [importlinter:contract:worker-management]
 name = Worker Management
 type = forbidden
 source_modules =
-    core.workflow.graph_engine.worker_management
+    dify_graph.graph_engine.worker_management
 forbidden_modules =
-    core.workflow.graph_engine.orchestration
-    core.workflow.graph_engine.command_processing
-    core.workflow.graph_engine.event_management
+    dify_graph.graph_engine.orchestration
+    dify_graph.graph_engine.command_processing
+    dify_graph.graph_engine.event_management
 
 
 [importlinter:contract:graph-traversal-components]
@@ -363,11 +213,11 @@ layers =
     edge_processor
     skip_propagator
 containers =
-    core.workflow.graph_engine.graph_traversal
+    dify_graph.graph_engine.graph_traversal
 
 [importlinter:contract:command-channels]
 name = Command Channels Independence
 type = independence
 modules =
-    core.workflow.graph_engine.command_channels.in_memory_channel
-    core.workflow.graph_engine.command_channels.redis_channel
+    dify_graph.graph_engine.command_channels.in_memory_channel
+    dify_graph.graph_engine.command_channels.redis_channel

api/.ruff.toml

@@ -53,6 +53,7 @@ select = [
     "S301", # suspicious-pickle-usage, disallow use of `pickle` and its wrappers.
     "S302", # suspicious-marshal-usage, disallow use of `marshal` module
     "S311", # suspicious-non-cryptographic-random-usage,
+    "TID",   # flake8-tidy-imports
 
 ]
 
@@ -88,6 +89,7 @@ ignore = [
     "SIM113",  # enumerate-for-loop
     "SIM117",  # multiple-with-statements
     "SIM210",  # if-expr-with-true-false
+    "TID252",  # allow relative imports from parent modules
 ]
 
 [lint.per-file-ignores]
@@ -98,7 +100,7 @@ ignore = [
 "configs/*" = [
     "N802", # invalid-function-name
 ]
-"core/model_runtime/callbacks/base_callback.py" = ["T201"]
+"dify_graph/model_runtime/callbacks/base_callback.py" = ["T201"]
 "core/workflow/callbacks/workflow_logging_callback.py" = ["T201"]
 "libs/gmpy2_pkcs10aep_cipher.py" = [
     "N803", # invalid-argument-name
@@ -109,10 +111,20 @@ ignore = [
     "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.pyflakes]
 allowed-unused-imports = [
-    "_pytest.monkeypatch",
     "tests.integration_tests",
     "tests.unit_tests",
 ]
+
+[lint.flake8-tidy-imports]
+
+[lint.flake8-tidy-imports.banned-api."flask_restx.reqparse"]
+msg = "Use Pydantic payload/query models instead of reqparse."
+
+[lint.flake8-tidy-imports.banned-api."flask_restx.reqparse.RequestParser"]
+msg = "Use Pydantic payload/query models instead of reqparse."

api/.vscode/launch.json.example

@@ -54,7 +54,7 @@
                 "--loglevel",
                 "DEBUG",
                 "-Q",
-                "dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor"
+                "dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,workflow_based_app_execution,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor"
             ]
         }
     ]

api/AGENTS.md

@@ -62,6 +62,22 @@ This is the default standard for backend code in this repo. Follow it for new co
 
 - Code should usually include type annotations that match the repo’s current Python version (avoid untyped public APIs and “mystery” values).
 - Prefer modern typing forms (e.g. `list[str]`, `dict[str, int]`) and avoid `Any` unless there’s a strong reason.
+- For dictionary-like data with known keys and value types, prefer `TypedDict` over `dict[...]` or `Mapping[...]`.
+- For optional keys in typed payloads, use `NotRequired[...]` (or `total=False` when most fields are optional).
+- Keep `dict[...]` / `Mapping[...]` for truly dynamic key spaces where the key set is unknown.
+
+```python
+from datetime import datetime
+from typing import NotRequired, TypedDict
+
+
+class UserProfile(TypedDict):
+    user_id: str
+    email: str
+    created_at: datetime
+    nickname: NotRequired[str]
+```
+
 - For classes, declare member variables at the top of the class body (before `__init__`) so the class shape is obvious at a glance:
 
 ```python

api/README.md

@@ -42,7 +42,7 @@ The scripts resolve paths relative to their location, so you can run them from a
 
 1. Set up your application by visiting `http://localhost:3000`.
 
-1. Optional: start the worker service (async tasks, runs from `api`).
+1. Start the worker service (async and scheduler tasks, runs from `api`).
 
    ```bash
    ./dev/start-worker
@@ -54,86 +54,6 @@ The scripts resolve paths relative to their location, so you can run them from a
    ./dev/start-beat
    ```
 
-### Manual commands
-
-<details>
-<summary>Show manual setup and run steps</summary>
-
-These commands assume you start from the repository root.
-
-1. Start the docker-compose stack.
-
-   The backend requires middleware, including PostgreSQL, Redis, and Weaviate, which can be started together using `docker-compose`.
-
-   ```bash
-   cp docker/middleware.env.example docker/middleware.env
-   # Use mysql or another vector database profile if you are not using postgres/weaviate.
-   docker compose -f docker/docker-compose.middleware.yaml --profile postgresql --profile weaviate -p dify up -d
-   ```
-
-1. Copy env files.
-
-   ```bash
-   cp api/.env.example api/.env
-   cp web/.env.example web/.env.local
-   ```
-
-1. Install UV if needed.
-
-   ```bash
-   pip install uv
-   # Or on macOS
-   brew install uv
-   ```
-
-1. Install API dependencies.
-
-   ```bash
-   cd api
-   uv sync --group dev
-   ```
-
-1. Install web dependencies.
-
-   ```bash
-   cd web
-   pnpm install
-   cd ..
-   ```
-
-1. Start backend (runs migrations first, in a new terminal).
-
-   ```bash
-   cd api
-   uv run flask db upgrade
-   uv run flask run --host 0.0.0.0 --port=5001 --debug
-   ```
-
-1. Start Dify [web](../web) service (in a new terminal).
-
-   ```bash
-   cd web
-   pnpm dev:inspect
-   ```
-
-1. Set up your application by visiting `http://localhost:3000`.
-
-1. Optional: start the worker service (async tasks, in a new terminal).
-
-   ```bash
-   cd api
-   uv run celery -A app.celery worker -P threads -c 2 --loglevel INFO -Q dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention
-   ```
-
-1. Optional: start Celery Beat (scheduled tasks, in a new terminal).
-
-   ```bash
-   cd api
-   uv run celery -A app.celery beat
-   ```
-
-</details>
-
 ### Environment notes
 
 > [!IMPORTANT]

api/app.py

@@ -1,4 +1,12 @@
+from __future__ import annotations
+
 import sys
+from typing import TYPE_CHECKING, cast
+
+if TYPE_CHECKING:
+    from celery import Celery
+
+    celery: Celery
 
 
 def is_db_command() -> bool:
@@ -23,7 +31,7 @@ else:
     from app_factory import create_app
 
     app = create_app()
-    celery = app.extensions["celery"]
+    celery = cast("Celery", app.extensions["celery"])
 
 if __name__ == "__main__":
     app.run(host="0.0.0.0", port=5001)

api/app_factory.py

@@ -149,7 +149,7 @@ def initialize_extensions(app: DifyApp):
             logger.info("Loaded %s (%s ms)", short_name, round((end_time - start_time) * 1000, 2))
 
 
-def create_migrations_app():
+def create_migrations_app() -> DifyApp:
     app = create_flask_app_with_configs()
     from extensions import ext_database, ext_migrate
 

api/commands.py

@@ -30,6 +30,8 @@ from extensions.ext_redis import redis_client
 from extensions.ext_storage import storage
 from extensions.storage.opendal_storage import OpenDALStorage
 from extensions.storage.storage_type import StorageType
+from libs.datetime_utils import naive_utc_now
+from libs.db_migration_lock import DbMigrationAutoRenewLock
 from libs.helper import email as email_validate
 from libs.password import hash_password, password_pattern, valid_password
 from libs.rsa import generate_key_pair
@@ -54,6 +56,8 @@ from tasks.remove_app_and_related_data_task import delete_draft_variables_batch
 
 logger = logging.getLogger(__name__)
 
+DB_UPGRADE_LOCK_TTL_SECONDS = 60
+
 
 @click.command("reset-password", help="Reset the account password.")
 @click.option("--email", prompt=True, help="Account email to reset password for")
@@ -727,8 +731,15 @@ def create_tenant(email: str, language: str | None = None, name: str | None = No
 @click.command("upgrade-db", help="Upgrade the database")
 def upgrade_db():
     click.echo("Preparing database migration...")
-    lock = redis_client.lock(name="db_upgrade_lock", timeout=60)
+    lock = DbMigrationAutoRenewLock(
+        redis_client=redis_client,
+        name="db_upgrade_lock",
+        ttl_seconds=DB_UPGRADE_LOCK_TTL_SECONDS,
+        logger=logger,
+        log_context="db_migration",
+    )
     if lock.acquire(blocking=False):
+        migration_succeeded = False
         try:
             click.echo(click.style("Starting database migration.", fg="green"))
 
@@ -737,12 +748,16 @@ def upgrade_db():
 
             flask_migrate.upgrade()
 
+            migration_succeeded = True
             click.echo(click.style("Database migration successful!", fg="green"))
 
-        except Exception:
+        except Exception as e:
             logger.exception("Failed to execute database migration")
+            click.echo(click.style(f"Database migration failed: {e}", fg="red"))
+            raise SystemExit(1)
         finally:
-            lock.release()
+            status = "successful" if migration_succeeded else "failed"
+            lock.release_safely(status=status)
     else:
         click.echo("Database migration skipped")
 
@@ -922,6 +937,12 @@ def clear_free_plan_tenant_expired_logs(days: int, batch: int, tenant_ids: list[
     is_flag=True,
     help="Preview cleanup results without deleting any workflow run data.",
 )
+@click.option(
+    "--task-label",
+    default="daily",
+    show_default=True,
+    help="Stable label value used to distinguish multiple cleanup CronJobs in metrics.",
+)
 def clean_workflow_runs(
     before_days: int,
     batch_size: int,
@@ -930,10 +951,13 @@ def clean_workflow_runs(
     start_from: datetime.datetime | None,
     end_before: datetime.datetime | None,
     dry_run: bool,
+    task_label: str,
 ):
     """
     Clean workflow runs and related workflow data for free tenants.
     """
+    from extensions.otel.runtime import flush_telemetry
+
     if (start_from is None) ^ (end_before is None):
         raise click.UsageError("--start-from and --end-before must be provided together.")
 
@@ -953,13 +977,17 @@ def clean_workflow_runs(
     start_time = datetime.datetime.now(datetime.UTC)
     click.echo(click.style(f"Starting workflow run cleanup at {start_time.isoformat()}.", fg="white"))
 
-    WorkflowRunCleanup(
-        days=before_days,
-        batch_size=batch_size,
-        start_from=start_from,
-        end_before=end_before,
-        dry_run=dry_run,
-    ).run()
+    try:
+        WorkflowRunCleanup(
+            days=before_days,
+            batch_size=batch_size,
+            start_from=start_from,
+            end_before=end_before,
+            dry_run=dry_run,
+            task_label=task_label,
+        ).run()
+    finally:
+        flush_telemetry()
 
     end_time = datetime.datetime.now(datetime.UTC)
     elapsed = end_time - start_time
@@ -1450,54 +1478,58 @@ def clear_orphaned_file_records(force: bool):
         all_ids_in_tables = []
         for ids_table in ids_tables:
             query = ""
-            if ids_table["type"] == "uuid":
-                click.echo(
-                    click.style(
-                        f"- Listing file ids in column {ids_table['column']} in table {ids_table['table']}", fg="white"
+            match ids_table["type"]:
+                case "uuid":
+                    click.echo(
+                        click.style(
+                            f"- Listing file ids in column {ids_table['column']} in table {ids_table['table']}",
+                            fg="white",
+                        )
                     )
-                )
-                query = (
-                    f"SELECT {ids_table['column']} FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
-                )
-                with db.engine.begin() as conn:
-                    rs = conn.execute(sa.text(query))
-                for i in rs:
-                    all_ids_in_tables.append({"table": ids_table["table"], "id": str(i[0])})
-            elif ids_table["type"] == "text":
-                click.echo(
-                    click.style(
-                        f"- Listing file-id-like strings in column {ids_table['column']} in table {ids_table['table']}",
-                        fg="white",
+                    c = ids_table["column"]
+                    query = f"SELECT {c} FROM {ids_table['table']} WHERE {c} IS NOT NULL"
+                    with db.engine.begin() as conn:
+                        rs = conn.execute(sa.text(query))
+                    for i in rs:
+                        all_ids_in_tables.append({"table": ids_table["table"], "id": str(i[0])})
+                case "text":
+                    t = ids_table["table"]
+                    click.echo(
+                        click.style(
+                            f"- Listing file-id-like strings in column {ids_table['column']} in table {t}",
+                            fg="white",
+                        )
                     )
-                )
-                query = (
-                    f"SELECT regexp_matches({ids_table['column']}, '{guid_regexp}', 'g') AS extracted_id "
-                    f"FROM {ids_table['table']}"
-                )
-                with db.engine.begin() as conn:
-                    rs = conn.execute(sa.text(query))
-                for i in rs:
-                    for j in i[0]:
-                        all_ids_in_tables.append({"table": ids_table["table"], "id": j})
-            elif ids_table["type"] == "json":
-                click.echo(
-                    click.style(
-                        (
-                            f"- Listing file-id-like JSON string in column {ids_table['column']} "
-                            f"in table {ids_table['table']}"
-                        ),
-                        fg="white",
+                    query = (
+                        f"SELECT regexp_matches({ids_table['column']}, '{guid_regexp}', 'g') AS extracted_id "
+                        f"FROM {ids_table['table']}"
                     )
-                )
-                query = (
-                    f"SELECT regexp_matches({ids_table['column']}::text, '{guid_regexp}', 'g') AS extracted_id "
-                    f"FROM {ids_table['table']}"
-                )
-                with db.engine.begin() as conn:
-                    rs = conn.execute(sa.text(query))
-                for i in rs:
-                    for j in i[0]:
-                        all_ids_in_tables.append({"table": ids_table["table"], "id": j})
+                    with db.engine.begin() as conn:
+                        rs = conn.execute(sa.text(query))
+                    for i in rs:
+                        for j in i[0]:
+                            all_ids_in_tables.append({"table": ids_table["table"], "id": j})
+                case "json":
+                    click.echo(
+                        click.style(
+                            (
+                                f"- Listing file-id-like JSON string in column {ids_table['column']} "
+                                f"in table {ids_table['table']}"
+                            ),
+                            fg="white",
+                        )
+                    )
+                    query = (
+                        f"SELECT regexp_matches({ids_table['column']}::text, '{guid_regexp}', 'g') AS extracted_id "
+                        f"FROM {ids_table['table']}"
+                    )
+                    with db.engine.begin() as conn:
+                        rs = conn.execute(sa.text(query))
+                    for i in rs:
+                        for j in i[0]:
+                            all_ids_in_tables.append({"table": ids_table["table"], "id": j})
+                case _:
+                    pass
         click.echo(click.style(f"Found {len(all_ids_in_tables)} file ids in tables.", fg="white"))
 
     except Exception as e:
@@ -1737,59 +1769,18 @@ def file_usage(
                 if src_filter != src:
                     continue
 
-        if ids_table["type"] == "uuid":
-            # Direct UUID match
-            query = (
-                f"SELECT {ids_table['pk_column']}, {ids_table['column']} "
-                f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
-            )
-            with db.engine.begin() as conn:
-                rs = conn.execute(sa.text(query))
-                for row in rs:
-                    record_id = str(row[0])
-                    ref_file_id = str(row[1])
-                    if ref_file_id not in file_key_map:
-                        continue
-                    storage_key = file_key_map[ref_file_id]
-
-                    # Apply filters
-                    if file_id and ref_file_id != file_id:
-                        continue
-                    if key and not storage_key.endswith(key):
-                        continue
-
-                    # Only collect items within the requested page range
-                    if offset <= total_count < offset + limit:
-                        paginated_usages.append(
-                            {
-                                "src": f"{ids_table['table']}.{ids_table['column']}",
-                                "record_id": record_id,
-                                "file_id": ref_file_id,
-                                "key": storage_key,
-                            }
-                        )
-                    total_count += 1
-
-        elif ids_table["type"] in ("text", "json"):
-            # Extract UUIDs from text/json content
-            column_cast = f"{ids_table['column']}::text" if ids_table["type"] == "json" else ids_table["column"]
-            query = (
-                f"SELECT {ids_table['pk_column']}, {column_cast} "
-                f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
-            )
-            with db.engine.begin() as conn:
-                rs = conn.execute(sa.text(query))
-                for row in rs:
-                    record_id = str(row[0])
-                    content = str(row[1])
-
-                    # Find all UUIDs in the content
-                    import re
-
-                    uuid_pattern = re.compile(guid_regexp, re.IGNORECASE)
-                    matches = uuid_pattern.findall(content)
-
-                    for ref_file_id in matches:
+        match ids_table["type"]:
+            case "uuid":
+                # Direct UUID match
+                query = (
+                    f"SELECT {ids_table['pk_column']}, {ids_table['column']} "
+                    f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
+                )
+                with db.engine.begin() as conn:
+                    rs = conn.execute(sa.text(query))
+                    for row in rs:
+                        record_id = str(row[0])
+                        ref_file_id = str(row[1])
                         if ref_file_id not in file_key_map:
                             continue
                         storage_key = file_key_map[ref_file_id]
@@ -1812,6 +1803,50 @@ def file_usage(
                             )
                         total_count += 1
 
+            case "text" | "json":
+                # Extract UUIDs from text/json content
+                column_cast = f"{ids_table['column']}::text" if ids_table["type"] == "json" else ids_table["column"]
+                query = (
+                    f"SELECT {ids_table['pk_column']}, {column_cast} "
+                    f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
+                )
+                with db.engine.begin() as conn:
+                    rs = conn.execute(sa.text(query))
+                    for row in rs:
+                        record_id = str(row[0])
+                        content = str(row[1])
+
+                        # Find all UUIDs in the content
+                        import re
+
+                        uuid_pattern = re.compile(guid_regexp, re.IGNORECASE)
+                        matches = uuid_pattern.findall(content)
+
+                        for ref_file_id in matches:
+                            if ref_file_id not in file_key_map:
+                                continue
+                            storage_key = file_key_map[ref_file_id]
+
+                            # Apply filters
+                            if file_id and ref_file_id != file_id:
+                                continue
+                            if key and not storage_key.endswith(key):
+                                continue
+
+                            # Only collect items within the requested page range
+                            if offset <= total_count < offset + limit:
+                                paginated_usages.append(
+                                    {
+                                        "src": f"{ids_table['table']}.{ids_table['column']}",
+                                        "record_id": record_id,
+                                        "file_id": ref_file_id,
+                                        "key": storage_key,
+                                    }
+                                )
+                            total_count += 1
+            case _:
+                pass
+
     # Output results
     if output_json:
         result = {
@@ -2577,15 +2612,29 @@ def migrate_oss(
 @click.option(
     "--start-from",
     type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
-    required=True,
+    required=False,
+    default=None,
     help="Lower bound (inclusive) for created_at.",
 )
 @click.option(
     "--end-before",
     type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
-    required=True,
+    required=False,
+    default=None,
     help="Upper bound (exclusive) for created_at.",
 )
+@click.option(
+    "--from-days-ago",
+    type=int,
+    default=None,
+    help="Relative lower bound in days ago (inclusive). Must be used with --before-days.",
+)
+@click.option(
+    "--before-days",
+    type=int,
+    default=None,
+    help="Relative upper bound in days ago (exclusive). Required for relative mode.",
+)
 @click.option("--batch-size", default=1000, show_default=True, help="Batch size for selecting messages.")
 @click.option(
     "--graceful-period",
@@ -2594,33 +2643,99 @@ def migrate_oss(
     help="Graceful period in days after subscription expiration, will be ignored when billing is disabled.",
 )
 @click.option("--dry-run", is_flag=True, default=False, help="Show messages logs would be cleaned without deleting")
+@click.option(
+    "--task-label",
+    default="daily",
+    show_default=True,
+    help="Stable label value used to distinguish multiple cleanup CronJobs in metrics.",
+)
 def clean_expired_messages(
     batch_size: int,
     graceful_period: int,
-    start_from: datetime.datetime,
-    end_before: datetime.datetime,
+    start_from: datetime.datetime | None,
+    end_before: datetime.datetime | None,
+    from_days_ago: int | None,
+    before_days: int | None,
     dry_run: bool,
+    task_label: str,
 ):
     """
     Clean expired messages and related data for tenants based on clean policy.
     """
+    from extensions.otel.runtime import flush_telemetry
+
     click.echo(click.style("clean_messages: start clean messages.", fg="green"))
 
     start_at = time.perf_counter()
 
     try:
+        abs_mode = start_from is not None and end_before is not None
+        rel_mode = before_days is not None
+
+        if abs_mode and rel_mode:
+            raise click.UsageError(
+                "Options are mutually exclusive: use either (--start-from,--end-before) "
+                "or (--from-days-ago,--before-days)."
+            )
+
+        if from_days_ago is not None and before_days is None:
+            raise click.UsageError("--from-days-ago must be used together with --before-days.")
+
+        if (start_from is None) ^ (end_before is None):
+            raise click.UsageError("Both --start-from and --end-before are required when using absolute time range.")
+
+        if not abs_mode and not rel_mode:
+            raise click.UsageError(
+                "You must provide either (--start-from,--end-before) or (--before-days [--from-days-ago])."
+            )
+
+        if rel_mode:
+            assert before_days is not None
+            if before_days < 0:
+                raise click.UsageError("--before-days must be >= 0.")
+            if from_days_ago is not None:
+                if from_days_ago < 0:
+                    raise click.UsageError("--from-days-ago must be >= 0.")
+                if from_days_ago <= before_days:
+                    raise click.UsageError("--from-days-ago must be greater than --before-days.")
+
         # Create policy based on billing configuration
         # NOTE: graceful_period will be ignored when billing is disabled.
         policy = create_message_clean_policy(graceful_period_days=graceful_period)
 
         # Create and run the cleanup service
-        service = MessagesCleanService.from_time_range(
-            policy=policy,
-            start_from=start_from,
-            end_before=end_before,
-            batch_size=batch_size,
-            dry_run=dry_run,
-        )
+        if abs_mode:
+            assert start_from is not None
+            assert end_before is not None
+            service = MessagesCleanService.from_time_range(
+                policy=policy,
+                start_from=start_from,
+                end_before=end_before,
+                batch_size=batch_size,
+                dry_run=dry_run,
+                task_label=task_label,
+            )
+        elif from_days_ago is None:
+            assert before_days is not None
+            service = MessagesCleanService.from_days(
+                policy=policy,
+                days=before_days,
+                batch_size=batch_size,
+                dry_run=dry_run,
+                task_label=task_label,
+            )
+        else:
+            assert before_days is not None
+            assert from_days_ago is not None
+            now = naive_utc_now()
+            service = MessagesCleanService.from_time_range(
+                policy=policy,
+                start_from=now - datetime.timedelta(days=from_days_ago),
+                end_before=now - datetime.timedelta(days=before_days),
+                batch_size=batch_size,
+                dry_run=dry_run,
+                task_label=task_label,
+            )
         stats = service.run()
 
         end_at = time.perf_counter()
@@ -2645,5 +2760,81 @@ def clean_expired_messages(
             )
         )
         raise
+    finally:
+        flush_telemetry()
 
     click.echo(click.style("messages cleanup completed.", fg="green"))
+
+
+@click.command("export-app-messages", help="Export messages for an app to JSONL.GZ.")
+@click.option("--app-id", required=True, help="Application ID to export messages for.")
+@click.option(
+    "--start-from",
+    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    default=None,
+    help="Optional lower bound (inclusive) for created_at.",
+)
+@click.option(
+    "--end-before",
+    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    required=True,
+    help="Upper bound (exclusive) for created_at.",
+)
+@click.option(
+    "--filename",
+    required=True,
+    help="Base filename (relative path). Do not include suffix like .jsonl.gz.",
+)
+@click.option("--use-cloud-storage", is_flag=True, default=False, help="Upload to cloud storage instead of local file.")
+@click.option("--batch-size", default=1000, show_default=True, help="Batch size for cursor pagination.")
+@click.option("--dry-run", is_flag=True, default=False, help="Scan only, print stats without writing any file.")
+def export_app_messages(
+    app_id: str,
+    start_from: datetime.datetime | None,
+    end_before: datetime.datetime,
+    filename: str,
+    use_cloud_storage: bool,
+    batch_size: int,
+    dry_run: bool,
+):
+    if start_from and start_from >= end_before:
+        raise click.UsageError("--start-from must be before --end-before.")
+
+    from services.retention.conversation.message_export_service import AppMessageExportService
+
+    try:
+        validated_filename = AppMessageExportService.validate_export_filename(filename)
+    except ValueError as e:
+        raise click.BadParameter(str(e), param_hint="--filename") from e
+
+    click.echo(click.style(f"export_app_messages: starting export for app {app_id}.", fg="green"))
+    start_at = time.perf_counter()
+
+    try:
+        service = AppMessageExportService(
+            app_id=app_id,
+            end_before=end_before,
+            filename=validated_filename,
+            start_from=start_from,
+            batch_size=batch_size,
+            use_cloud_storage=use_cloud_storage,
+            dry_run=dry_run,
+        )
+        stats = service.run()
+
+        elapsed = time.perf_counter() - start_at
+        click.echo(
+            click.style(
+                f"export_app_messages: completed in {elapsed:.2f}s\n"
+                f"  - Batches: {stats.batches}\n"
+                f"  - Total messages: {stats.total_messages}\n"
+                f"  - Messages with feedback: {stats.messages_with_feedback}\n"
+                f"  - Total feedbacks: {stats.total_feedbacks}",
+                fg="green",
+            )
+        )
+    except Exception as e:
+        elapsed = time.perf_counter() - start_at
+        logger.exception("export_app_messages failed")
+        click.echo(click.style(f"export_app_messages: failed after {elapsed:.2f}s - {e}", fg="red"))
+        raise

api/configs/feature/__init__.py

@@ -1,3 +1,4 @@
+from datetime import timedelta
 from enum import StrEnum
 from typing import Literal
 
@@ -48,6 +49,16 @@ class SecurityConfig(BaseSettings):
         default=5,
     )
 
+    WEB_FORM_SUBMIT_RATE_LIMIT_MAX_ATTEMPTS: PositiveInt = Field(
+        description="Maximum number of web form submissions allowed per IP within the rate limit window",
+        default=30,
+    )
+
+    WEB_FORM_SUBMIT_RATE_LIMIT_WINDOW_SECONDS: PositiveInt = Field(
+        description="Time window in seconds for web form submission rate limiting",
+        default=60,
+    )
+
     LOGIN_DISABLED: bool = Field(
         description="Whether to disable login checks",
         default=False,
@@ -82,6 +93,12 @@ class AppExecutionConfig(BaseSettings):
         default=0,
     )
 
+    HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS: PositiveInt = Field(
+        description="Maximum seconds a workflow run can stay paused waiting for human input before global timeout.",
+        default=int(timedelta(days=7).total_seconds()),
+        ge=1,
+    )
+
 
 class CodeExecutionSandboxConfig(BaseSettings):
     """
@@ -243,6 +260,16 @@ class PluginConfig(BaseSettings):
         default=15728640 * 12,
     )
 
+    PLUGIN_MODEL_SCHEMA_CACHE_TTL: PositiveInt = Field(
+        description="TTL in seconds for caching plugin model schemas in Redis",
+        default=60 * 60,
+    )
+
+    PLUGIN_MAX_FILE_SIZE: PositiveInt = Field(
+        description="Maximum allowed size (bytes) for plugin-generated files",
+        default=50 * 1024 * 1024,
+    )
+
 
 class MarketplaceConfig(BaseSettings):
     """
@@ -1129,6 +1156,14 @@ class CeleryScheduleTasksConfig(BaseSettings):
         description="Enable queue monitor task",
         default=False,
     )
+    ENABLE_HUMAN_INPUT_TIMEOUT_TASK: bool = Field(
+        description="Enable human input timeout check task",
+        default=True,
+    )
+    HUMAN_INPUT_TIMEOUT_TASK_INTERVAL: PositiveInt = Field(
+        description="Human input timeout check interval in minutes",
+        default=1,
+    )
     ENABLE_CHECK_UPGRADABLE_PLUGIN_TASK: bool = Field(
         description="Enable check upgradable plugin task",
         default=True,
@@ -1150,6 +1185,16 @@ class CeleryScheduleTasksConfig(BaseSettings):
         default=0,
     )
 
+    # API token last_used_at batch update
+    ENABLE_API_TOKEN_LAST_USED_UPDATE_TASK: bool = Field(
+        description="Enable periodic batch update of API token last_used_at timestamps",
+        default=True,
+    )
+    API_TOKEN_LAST_USED_UPDATE_INTERVAL: int = Field(
+        description="Interval in minutes for batch updating API token last_used_at (default 30)",
+        default=30,
+    )
+
     # Trigger provider refresh (simple version)
     ENABLE_TRIGGER_PROVIDER_REFRESH_TASK: bool = Field(
         description="Enable trigger provider refresh poller",
@@ -1274,6 +1319,9 @@ class WorkflowLogConfig(BaseSettings):
     WORKFLOW_LOG_CLEANUP_BATCH_SIZE: int = Field(
         default=100, description="Batch size for workflow run log cleanup operations"
     )
+    WORKFLOW_LOG_CLEANUP_SPECIFIC_WORKFLOW_IDS: str = Field(
+        default="", description="Comma-separated list of workflow IDs to clean logs for"
+    )
 
 
 class SwaggerUIConfig(BaseSettings):
@@ -1304,6 +1352,10 @@ class SandboxExpiredRecordsCleanConfig(BaseSettings):
         description="Maximum number of records to process in each batch",
         default=1000,
     )
+    SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_MAX_INTERVAL: PositiveInt = Field(
+        description="Maximum interval in milliseconds between batches",
+        default=200,
+    )
     SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS: PositiveInt = Field(
         description="Retention days for sandbox expired workflow_run records and message records",
         default=30,

api/configs/middleware/__init__.py

@@ -6,6 +6,7 @@ from pydantic import Field, NonNegativeFloat, NonNegativeInt, PositiveFloat, Pos
 from pydantic_settings import BaseSettings
 
 from .cache.redis_config import RedisConfig
+from .cache.redis_pubsub_config import RedisPubSubConfig
 from .storage.aliyun_oss_storage_config import AliyunOSSStorageConfig
 from .storage.amazon_s3_storage_config import S3StorageConfig
 from .storage.azure_blob_storage_config import AzureBlobStorageConfig
@@ -258,11 +259,20 @@ class CeleryConfig(DatabaseConfig):
         description="Password of the Redis Sentinel master.",
         default=None,
     )
+
     CELERY_SENTINEL_SOCKET_TIMEOUT: PositiveFloat | None = Field(
         description="Timeout for Redis Sentinel socket operations in seconds.",
         default=0.1,
     )
 
+    CELERY_TASK_ANNOTATIONS: dict[str, Any] | None = Field(
+        description=(
+            "Annotations for Celery tasks as a JSON mapping of task name -> options "
+            "(for example, rate limits or other task-specific settings)."
+        ),
+        default=None,
+    )
+
     @computed_field
     def CELERY_RESULT_BACKEND(self) -> str | None:
         if self.CELERY_BACKEND in ("database", "rabbitmq"):
@@ -317,6 +327,7 @@ class MiddlewareConfig(
     CeleryConfig,  # Note: CeleryConfig already inherits from DatabaseConfig
     KeywordStoreConfig,
     RedisConfig,
+    RedisPubSubConfig,
     # configs of storage and storage providers
     StorageConfig,
     AliyunOSSStorageConfig,

api/configs/middleware/cache/redis_config.py

@@ -111,3 +111,8 @@ class RedisConfig(BaseSettings):
         description="Enable client side cache in redis",
         default=False,
     )
+
+    REDIS_MAX_CONNECTIONS: PositiveInt | None = Field(
+        description="Maximum connections in the Redis connection pool (unset for library default)",
+        default=None,
+    )

api/configs/middleware/cache/redis_pubsub_config.py

@@ -0,0 +1,111 @@
+from typing import Literal, Protocol
+from urllib.parse import quote_plus, urlunparse
+
+from pydantic import AliasChoices, Field
+from pydantic_settings import BaseSettings
+
+
+class RedisConfigDefaults(Protocol):
+    REDIS_HOST: str
+    REDIS_PORT: int
+    REDIS_USERNAME: str | None
+    REDIS_PASSWORD: str | None
+    REDIS_DB: int
+    REDIS_USE_SSL: bool
+    REDIS_USE_SENTINEL: bool | None
+    REDIS_USE_CLUSTERS: bool
+
+
+class RedisConfigDefaultsMixin:
+    def _redis_defaults(self: RedisConfigDefaults) -> RedisConfigDefaults:
+        return self
+
+
+class RedisPubSubConfig(BaseSettings, RedisConfigDefaultsMixin):
+    """
+    Configuration settings for event transport between API and workers.
+
+    Supported transports:
+    - pubsub: Redis PUBLISH/SUBSCRIBE (at-most-once)
+    - sharded: Redis 7+ Sharded Pub/Sub (at-most-once, better scaling)
+    - streams: Redis Streams (at-least-once, supports late subscribers)
+    """
+
+    PUBSUB_REDIS_URL: str | None = Field(
+        validation_alias=AliasChoices("EVENT_BUS_REDIS_URL", "PUBSUB_REDIS_URL"),
+        description=(
+            "Redis connection URL for streaming events between API and celery worker; "
+            "defaults to URL constructed from `REDIS_*` configurations. Also accepts ENV: EVENT_BUS_REDIS_URL."
+        ),
+        default=None,
+    )
+
+    PUBSUB_REDIS_USE_CLUSTERS: bool = Field(
+        validation_alias=AliasChoices("EVENT_BUS_REDIS_CLUSTERS", "PUBSUB_REDIS_USE_CLUSTERS"),
+        description=(
+            "Enable Redis Cluster mode for pub/sub or streams transport. Recommended for large deployments. "
+            "Also accepts ENV: EVENT_BUS_REDIS_CLUSTERS."
+        ),
+        default=False,
+    )
+
+    PUBSUB_REDIS_CHANNEL_TYPE: Literal["pubsub", "sharded", "streams"] = Field(
+        validation_alias=AliasChoices("EVENT_BUS_REDIS_CHANNEL_TYPE", "PUBSUB_REDIS_CHANNEL_TYPE"),
+        description=(
+            "Event transport type. Options are:\n\n"
+            " - pubsub: normal Pub/Sub (at-most-once)\n"
+            " - sharded: sharded Pub/Sub (at-most-once)\n"
+            " - streams: Redis Streams (at-least-once, recommended to avoid subscriber races)\n\n"
+            "Note: Before enabling 'streams' in production, estimate your expected event volume and retention needs.\n"
+            "Configure Redis memory limits and stream trimming appropriately (e.g., MAXLEN and key expiry) to reduce\n"
+            "the risk of data loss from Redis auto-eviction under memory pressure.\n"
+            "Also accepts ENV: EVENT_BUS_REDIS_CHANNEL_TYPE."
+        ),
+        default="pubsub",
+    )
+
+    PUBSUB_STREAMS_RETENTION_SECONDS: int = Field(
+        validation_alias=AliasChoices("EVENT_BUS_STREAMS_RETENTION_SECONDS", "PUBSUB_STREAMS_RETENTION_SECONDS"),
+        description=(
+            "When using 'streams', expire each stream key this many seconds after the last event is published. "
+            "Also accepts ENV: EVENT_BUS_STREAMS_RETENTION_SECONDS."
+        ),
+        default=600,
+    )
+
+    def _build_default_pubsub_url(self) -> str:
+        defaults = self._redis_defaults()
+        if not defaults.REDIS_HOST or not defaults.REDIS_PORT:
+            raise ValueError("PUBSUB_REDIS_URL must be set when default Redis URL cannot be constructed")
+
+        scheme = "rediss" if defaults.REDIS_USE_SSL else "redis"
+        username = defaults.REDIS_USERNAME or None
+        password = defaults.REDIS_PASSWORD or None
+
+        userinfo = ""
+        if username:
+            userinfo = quote_plus(username)
+        if password:
+            password_part = quote_plus(password)
+            userinfo = f"{userinfo}:{password_part}" if userinfo else f":{password_part}"
+        if userinfo:
+            userinfo = f"{userinfo}@"
+
+        host = defaults.REDIS_HOST
+        port = defaults.REDIS_PORT
+        db = defaults.REDIS_DB
+
+        netloc = f"{userinfo}{host}:{port}"
+        return urlunparse((scheme, netloc, f"/{db}", "", "", ""))
+
+    @property
+    def normalized_pubsub_redis_url(self) -> str:
+        pubsub_redis_url = self.PUBSUB_REDIS_URL
+        if pubsub_redis_url:
+            cleaned = pubsub_redis_url.strip()
+            pubsub_redis_url = cleaned or None
+
+        if pubsub_redis_url:
+            return pubsub_redis_url
+
+        return self._build_default_pubsub_url()

api/configs/middleware/vdb/oceanbase_config.py

@@ -1,3 +1,5 @@
+from typing import Literal
+
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -49,3 +51,43 @@ class OceanBaseVectorConfig(BaseSettings):
         ),
         default="ik",
     )
+
+    OCEANBASE_VECTOR_BATCH_SIZE: PositiveInt = Field(
+        description="Number of documents to insert per batch",
+        default=100,
+    )
+
+    OCEANBASE_VECTOR_METRIC_TYPE: Literal["l2", "cosine", "inner_product"] = Field(
+        description="Distance metric type for vector index: l2, cosine, or inner_product",
+        default="l2",
+    )
+
+    OCEANBASE_HNSW_M: PositiveInt = Field(
+        description="HNSW M parameter (max number of connections per node)",
+        default=16,
+    )
+
+    OCEANBASE_HNSW_EF_CONSTRUCTION: PositiveInt = Field(
+        description="HNSW efConstruction parameter (index build-time search width)",
+        default=256,
+    )
+
+    OCEANBASE_HNSW_EF_SEARCH: int = Field(
+        description="HNSW efSearch parameter (query-time search width, -1 uses server default)",
+        default=-1,
+    )
+
+    OCEANBASE_VECTOR_POOL_SIZE: PositiveInt = Field(
+        description="SQLAlchemy connection pool size",
+        default=5,
+    )
+
+    OCEANBASE_VECTOR_MAX_OVERFLOW: int = Field(
+        description="SQLAlchemy connection pool max overflow connections",
+        default=10,
+    )
+
+    OCEANBASE_HNSW_REFRESH_THRESHOLD: int = Field(
+        description="Minimum number of inserted documents to trigger an automatic HNSW index refresh (0 to disable)",
+        default=1000,
+    )

api/constants/languages.py

@@ -21,6 +21,7 @@ language_timezone_mapping = {
     "th-TH": "Asia/Bangkok",
     "id-ID": "Asia/Jakarta",
     "ar-TN": "Africa/Tunis",
+    "nl-NL": "Europe/Amsterdam",
 }
 
 languages = list(language_timezone_mapping.keys())

api/context/__init__.py

@@ -12,7 +12,7 @@ or any other web framework.
 import contextvars
 from collections.abc import Callable
 
-from core.workflow.context.execution_context import (
+from dify_graph.context.execution_context import (
     ExecutionContext,
     IExecutionContext,
     NullAppContext,

api/context/flask_app_context.py

@@ -10,8 +10,8 @@ from typing import Any, final
 
 from flask import Flask, current_app, g
 
-from core.workflow.context import register_context_capturer
-from core.workflow.context.execution_context import (
+from dify_graph.context import register_context_capturer
+from dify_graph.context.execution_context import (
     AppContext,
     IExecutionContext,
 )

api/contexts/__init__.py

@@ -6,7 +6,6 @@ from contexts.wrapper import RecyclableContextVar
 
 if TYPE_CHECKING:
     from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
-    from core.model_runtime.entities.model_entities import AIModelEntity
     from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
     from core.tools.plugin_tool.provider import PluginToolProviderController
     from core.trigger.provider import PluginTriggerProviderController
@@ -29,12 +28,6 @@ plugin_model_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(
     ContextVar("plugin_model_providers_lock")
 )
 
-plugin_model_schema_lock: RecyclableContextVar[Lock] = RecyclableContextVar(ContextVar("plugin_model_schema_lock"))
-
-plugin_model_schemas: RecyclableContextVar[dict[str, "AIModelEntity"]] = RecyclableContextVar(
-    ContextVar("plugin_model_schemas")
-)
-
 datasource_plugin_providers: RecyclableContextVar[dict[str, "DatasourcePluginProviderController"]] = (
     RecyclableContextVar(ContextVar("datasource_plugin_providers"))
 )

api/controllers/common/fields.py

@@ -4,7 +4,7 @@ from typing import Any, TypeAlias
 
 from pydantic import BaseModel, ConfigDict, computed_field
 
-from core.file import helpers as file_helpers
+from dify_graph.file import helpers as file_helpers
 from models.model import IconType
 
 JSONValue: TypeAlias = str | int | float | bool | None | dict[str, Any] | list[Any]

api/controllers/common/schema.py

@@ -5,8 +5,6 @@ from enum import StrEnum
 from flask_restx import Namespace
 from pydantic import BaseModel, TypeAdapter
 
-from controllers.console import console_ns
-
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
@@ -24,6 +22,9 @@ def register_schema_models(namespace: Namespace, *models: type[BaseModel]) -> No
 
 
 def get_or_create_model(model_name: str, field_def):
+    # Import lazily to avoid circular imports between console controllers and schema helpers.
+    from controllers.console import console_ns
+
     existing = console_ns.models.get(model_name)
     if existing is None:
         existing = console_ns.model(model_name, field_def)

api/controllers/console/__init__.py

@@ -37,6 +37,7 @@ from . import (
     apikey,
     extension,
     feature,
+    human_input_form,
     init_validate,
     ping,
     setup,
@@ -171,6 +172,7 @@ __all__ = [
     "forgot_password",
     "generator",
     "hit_testing",
+    "human_input_form",
     "init_validate",
     "installed_app",
     "load_balancing_config",

api/controllers/console/admin.py

@@ -243,15 +243,13 @@ class InsertExploreBannerApi(Resource):
     def post(self):
         payload = InsertExploreBannerPayload.model_validate(console_ns.payload)
 
-        content = {
-            "category": payload.category,
-            "title": payload.title,
-            "description": payload.description,
-            "img-src": payload.img_src,
-        }
-
         banner = ExporleBanner(
-            content=content,
+            content={
+                "category": payload.category,
+                "title": payload.title,
+                "description": payload.description,
+                "img-src": payload.img_src,
+            },
             link=payload.link,
             sort=payload.sort,
             language=payload.language,

api/controllers/console/apikey.py

@@ -10,6 +10,7 @@ from libs.helper import TimestampField
 from libs.login import current_account_with_tenant, login_required
 from models.dataset import Dataset
 from models.model import ApiToken, App
+from services.api_token_service import ApiTokenCache
 
 from . import console_ns
 from .wraps import account_initialization_required, edit_permission_required, setup_required
@@ -131,6 +132,11 @@ class BaseApiKeyResource(Resource):
         if key is None:
             flask_restx.abort(HTTPStatus.NOT_FOUND, message="API key not found")
 
+        # Invalidate cache before deleting from database
+        # Type assertion: key is guaranteed to be non-None here because abort() raises
+        assert key is not None  # nosec - for type checker only
+        ApiTokenCache.delete(key.token, key.type)
+
         db.session.query(ApiToken).where(ApiToken.id == api_key_id).delete()
         db.session.commit()
 

api/controllers/console/app/annotation.py

@@ -1,10 +1,11 @@
 from typing import Any, Literal
 
 from flask import abort, make_response, request
-from flask_restx import Resource, fields, marshal, marshal_with
-from pydantic import BaseModel, Field, field_validator
+from flask_restx import Resource
+from pydantic import BaseModel, Field, TypeAdapter, field_validator
 
 from controllers.common.errors import NoFileUploadedError, TooManyFilesError
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import (
     account_initialization_required,
@@ -16,9 +17,11 @@ from controllers.console.wraps import (
 )
 from extensions.ext_redis import redis_client
 from fields.annotation_fields import (
-    annotation_fields,
-    annotation_hit_history_fields,
-    build_annotation_model,
+    Annotation,
+    AnnotationExportList,
+    AnnotationHitHistory,
+    AnnotationHitHistoryList,
+    AnnotationList,
 )
 from libs.helper import uuid_value
 from libs.login import login_required
@@ -89,6 +92,14 @@ reg(CreateAnnotationPayload)
 reg(UpdateAnnotationPayload)
 reg(AnnotationReplyStatusQuery)
 reg(AnnotationFilePayload)
+register_schema_models(
+    console_ns,
+    Annotation,
+    AnnotationList,
+    AnnotationExportList,
+    AnnotationHitHistory,
+    AnnotationHitHistoryList,
+)
 
 
 @console_ns.route("/apps/<uuid:app_id>/annotation-reply/<string:action>")
@@ -107,10 +118,11 @@ class AnnotationReplyActionApi(Resource):
     def post(self, app_id, action: Literal["enable", "disable"]):
         app_id = str(app_id)
         args = AnnotationReplyPayload.model_validate(console_ns.payload)
-        if action == "enable":
-            result = AppAnnotationService.enable_app_annotation(args.model_dump(), app_id)
-        elif action == "disable":
-            result = AppAnnotationService.disable_app_annotation(app_id)
+        match action:
+            case "enable":
+                result = AppAnnotationService.enable_app_annotation(args.model_dump(), app_id)
+            case "disable":
+                result = AppAnnotationService.disable_app_annotation(app_id)
         return result, 200
 
 
@@ -201,33 +213,33 @@ class AnnotationApi(Resource):
 
         app_id = str(app_id)
         annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(app_id, page, limit, keyword)
-        response = {
-            "data": marshal(annotation_list, annotation_fields),
-            "has_more": len(annotation_list) == limit,
-            "limit": limit,
-            "total": total,
-            "page": page,
-        }
-        return response, 200
+        annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
+        response = AnnotationList(
+            data=annotation_models,
+            has_more=len(annotation_list) == limit,
+            limit=limit,
+            total=total,
+            page=page,
+        )
+        return response.model_dump(mode="json"), 200
 
     @console_ns.doc("create_annotation")
     @console_ns.doc(description="Create a new annotation for an app")
     @console_ns.doc(params={"app_id": "Application ID"})
     @console_ns.expect(console_ns.models[CreateAnnotationPayload.__name__])
-    @console_ns.response(201, "Annotation created successfully", build_annotation_model(console_ns))
+    @console_ns.response(201, "Annotation created successfully", console_ns.models[Annotation.__name__])
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
     @login_required
     @account_initialization_required
     @cloud_edition_billing_resource_check("annotation")
-    @marshal_with(annotation_fields)
     @edit_permission_required
     def post(self, app_id):
         app_id = str(app_id)
         args = CreateAnnotationPayload.model_validate(console_ns.payload)
         data = args.model_dump(exclude_none=True)
         annotation = AppAnnotationService.up_insert_app_annotation_from_message(data, app_id)
-        return annotation
+        return Annotation.model_validate(annotation, from_attributes=True).model_dump(mode="json")
 
     @setup_required
     @login_required
@@ -264,7 +276,7 @@ class AnnotationExportApi(Resource):
     @console_ns.response(
         200,
         "Annotations exported successfully",
-        console_ns.model("AnnotationList", {"data": fields.List(fields.Nested(build_annotation_model(console_ns)))}),
+        console_ns.models[AnnotationExportList.__name__],
     )
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
@@ -274,7 +286,8 @@ class AnnotationExportApi(Resource):
     def get(self, app_id):
         app_id = str(app_id)
         annotation_list = AppAnnotationService.export_annotation_list_by_app_id(app_id)
-        response_data = {"data": marshal(annotation_list, annotation_fields)}
+        annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
+        response_data = AnnotationExportList(data=annotation_models).model_dump(mode="json")
 
         # Create response with secure headers for CSV export
         response = make_response(response_data, 200)
@@ -289,7 +302,7 @@ class AnnotationUpdateDeleteApi(Resource):
     @console_ns.doc("update_delete_annotation")
     @console_ns.doc(description="Update or delete an annotation")
     @console_ns.doc(params={"app_id": "Application ID", "annotation_id": "Annotation ID"})
-    @console_ns.response(200, "Annotation updated successfully", build_annotation_model(console_ns))
+    @console_ns.response(200, "Annotation updated successfully", console_ns.models[Annotation.__name__])
     @console_ns.response(204, "Annotation deleted successfully")
     @console_ns.response(403, "Insufficient permissions")
     @console_ns.expect(console_ns.models[UpdateAnnotationPayload.__name__])
@@ -298,7 +311,6 @@ class AnnotationUpdateDeleteApi(Resource):
     @account_initialization_required
     @cloud_edition_billing_resource_check("annotation")
     @edit_permission_required
-    @marshal_with(annotation_fields)
     def post(self, app_id, annotation_id):
         app_id = str(app_id)
         annotation_id = str(annotation_id)
@@ -306,7 +318,7 @@ class AnnotationUpdateDeleteApi(Resource):
         annotation = AppAnnotationService.update_app_annotation_directly(
             args.model_dump(exclude_none=True), app_id, annotation_id
         )
-        return annotation
+        return Annotation.model_validate(annotation, from_attributes=True).model_dump(mode="json")
 
     @setup_required
     @login_required
@@ -414,14 +426,7 @@ class AnnotationHitHistoryListApi(Resource):
     @console_ns.response(
         200,
         "Hit histories retrieved successfully",
-        console_ns.model(
-            "AnnotationHitHistoryList",
-            {
-                "data": fields.List(
-                    fields.Nested(console_ns.model("AnnotationHitHistoryItem", annotation_hit_history_fields))
-                )
-            },
-        ),
+        console_ns.models[AnnotationHitHistoryList.__name__],
     )
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
@@ -436,11 +441,14 @@ class AnnotationHitHistoryListApi(Resource):
         annotation_hit_history_list, total = AppAnnotationService.get_annotation_hit_histories(
             app_id, annotation_id, page, limit
         )
-        response = {
-            "data": marshal(annotation_hit_history_list, annotation_hit_history_fields),
-            "has_more": len(annotation_hit_history_list) == limit,
-            "limit": limit,
-            "total": total,
-            "page": page,
-        }
-        return response
+        history_models = TypeAdapter(list[AnnotationHitHistory]).validate_python(
+            annotation_hit_history_list, from_attributes=True
+        )
+        response = AnnotationHitHistoryList(
+            data=history_models,
+            has_more=len(annotation_hit_history_list) == limit,
+            limit=limit,
+            total=total,
+            page=page,
+        )
+        return response.model_dump(mode="json")

api/controllers/console/app/app.py

@@ -1,3 +1,4 @@
+import logging
 import uuid
 from datetime import datetime
 from typing import Any, Literal, TypeAlias
@@ -22,10 +23,10 @@ from controllers.console.wraps import (
     is_admin_or_owner_required,
     setup_required,
 )
-from core.file import helpers as file_helpers
 from core.ops.ops_trace_manager import OpsTraceManager
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
-from core.workflow.enums import NodeType, WorkflowExecutionStatus
+from dify_graph.enums import NodeType, WorkflowExecutionStatus
+from dify_graph.file import helpers as file_helpers
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant, login_required
 from models import App, DatasetPermissionEnum, Workflow
@@ -54,6 +55,8 @@ ALLOW_CREATE_APP_MODES = ["chat", "agent-chat", "advanced-chat", "workflow", "co
 
 register_enum_models(console_ns, IconType)
 
+_logger = logging.getLogger(__name__)
+
 
 class AppListQuery(BaseModel):
     page: int = Field(default=1, ge=1, le=99999, description="Page number (1-99999)")
@@ -499,6 +502,7 @@ class AppListApi(Resource):
                     select(Workflow).where(
                         Workflow.version == Workflow.VERSION_DRAFT,
                         Workflow.app_id.in_(workflow_capable_app_ids),
+                        Workflow.tenant_id == current_tenant_id,
                     )
                 )
                 .scalars()
@@ -510,12 +514,14 @@ class AppListApi(Resource):
                 NodeType.TRIGGER_PLUGIN,
             }
             for workflow in draft_workflows:
+                node_id = None
                 try:
-                    for _, node_data in workflow.walk_nodes():
+                    for node_id, node_data in workflow.walk_nodes():
                         if node_data.get("type") in trigger_node_types:
                             draft_trigger_app_ids.add(str(workflow.app_id))
                             break
                 except Exception:
+                    _logger.exception("error while walking nodes, workflow_id=%s, node_id=%s", workflow.id, node_id)
                     continue
 
         for app in app_pagination.items:
@@ -654,6 +660,19 @@ class AppCopyApi(Resource):
             )
             session.commit()
 
+            # Inherit web app permission from original app
+            if result.app_id and FeatureService.get_system_features().webapp_auth.enabled:
+                try:
+                    # Get the original app's access mode
+                    original_settings = EnterpriseService.WebAppAuth.get_app_access_mode_by_id(app_model.id)
+                    access_mode = original_settings.access_mode
+                except Exception:
+                    # If original app has no settings (old app), default to public to match fallback behavior
+                    access_mode = "public"
+
+                # Apply the same access mode to the copied app
+                EnterpriseService.WebAppAuth.update_app_access_mode(result.app_id, access_mode)
+
             stmt = select(App).where(App.id == result.app_id)
             app = session.scalar(stmt)
 

api/controllers/console/app/audio.py

@@ -6,6 +6,7 @@ from pydantic import BaseModel, Field
 from werkzeug.exceptions import InternalServerError
 
 import services
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
     AppUnavailableError,
@@ -21,7 +22,7 @@ from controllers.console.app.error import (
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from core.model_runtime.errors.invoke import InvokeError
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from libs.login import login_required
 from models import App, AppMode
 from services.audio_service import AudioService
@@ -33,7 +34,6 @@ from services.errors.audio import (
 )
 
 logger = logging.getLogger(__name__)
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
 class TextToSpeechPayload(BaseModel):
@@ -47,13 +47,11 @@ class TextToSpeechVoiceQuery(BaseModel):
     language: str = Field(..., description="Language code")
 
 
-console_ns.schema_model(
-    TextToSpeechPayload.__name__, TextToSpeechPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
-console_ns.schema_model(
-    TextToSpeechVoiceQuery.__name__,
-    TextToSpeechVoiceQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
+class AudioTranscriptResponse(BaseModel):
+    text: str = Field(description="Transcribed text from audio")
+
+
+register_schema_models(console_ns, AudioTranscriptResponse, TextToSpeechPayload, TextToSpeechVoiceQuery)
 
 
 @console_ns.route("/apps/<uuid:app_id>/audio-to-text")
@@ -64,7 +62,7 @@ class ChatMessageAudioApi(Resource):
     @console_ns.response(
         200,
         "Audio transcription successful",
-        console_ns.model("AudioTranscriptResponse", {"text": fields.String(description="Transcribed text from audio")}),
+        console_ns.models[AudioTranscriptResponse.__name__],
     )
     @console_ns.response(400, "Bad request - No audio uploaded or unsupported type")
     @console_ns.response(413, "Audio file too large")

api/controllers/console/app/completion.py

@@ -26,7 +26,7 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.helper.trace_id_helper import get_external_trace_id
-from core.model_runtime.errors.invoke import InvokeError
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import uuid_value
 from libs.login import current_user, login_required

api/controllers/console/app/conversation.py

@@ -89,6 +89,7 @@ status_count_model = console_ns.model(
         "success": fields.Integer,
         "failed": fields.Integer,
         "partial_success": fields.Integer,
+        "paused": fields.Integer,
     },
 )
 
@@ -508,16 +509,19 @@ class ChatConversationApi(Resource):
                 case "created_at" | "-created_at" | _:
                     query = query.where(Conversation.created_at <= end_datetime_utc)
 
-        if args.annotation_status == "annotated":
-            query = query.options(joinedload(Conversation.message_annotations)).join(  # type: ignore
-                MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
-            )
-        elif args.annotation_status == "not_annotated":
-            query = (
-                query.outerjoin(MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id)
-                .group_by(Conversation.id)
-                .having(func.count(MessageAnnotation.id) == 0)
-            )
+        match args.annotation_status:
+            case "annotated":
+                query = query.options(joinedload(Conversation.message_annotations)).join(  # type: ignore
+                    MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
+                )
+            case "not_annotated":
+                query = (
+                    query.outerjoin(MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id)
+                    .group_by(Conversation.id)
+                    .having(func.count(MessageAnnotation.id) == 0)
+                )
+            case "all":
+                pass
 
         if app_model.mode == AppMode.ADVANCED_CHAT:
             query = query.where(Conversation.invoke_from != InvokeFrom.DEBUGGER)
@@ -595,7 +599,12 @@ def _get_conversation(app_model, conversation_id):
     db.session.execute(
         sa.update(Conversation)
         .where(Conversation.id == conversation_id, Conversation.read_at.is_(None))
-        .values(read_at=naive_utc_now(), read_account_id=current_user.id)
+        # Keep updated_at unchanged when only marking a conversation as read.
+        .values(
+            read_at=naive_utc_now(),
+            read_account_id=current_user.id,
+            updated_at=Conversation.updated_at,
+        )
     )
     db.session.commit()
     db.session.refresh(conversation)

api/controllers/console/app/generator.py

@@ -1,5 +1,4 @@
 from collections.abc import Sequence
-from typing import Any
 
 from flask_restx import Resource
 from pydantic import BaseModel, Field
@@ -12,12 +11,14 @@ from controllers.console.app.error import (
     ProviderQuotaExceededError,
 )
 from controllers.console.wraps import account_initialization_required, setup_required
+from core.app.app_config.entities import ModelConfig
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from core.helper.code_executor.code_node_provider import CodeNodeProvider
 from core.helper.code_executor.javascript.javascript_code_provider import JavascriptCodeProvider
 from core.helper.code_executor.python3.python3_code_provider import Python3CodeProvider
+from core.llm_generator.entities import RuleCodeGeneratePayload, RuleGeneratePayload, RuleStructuredOutputPayload
 from core.llm_generator.llm_generator import LLMGenerator
-from core.model_runtime.errors.invoke import InvokeError
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant, login_required
 from models import App
@@ -26,28 +27,13 @@ from services.workflow_service import WorkflowService
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
-class RuleGeneratePayload(BaseModel):
-    instruction: str = Field(..., description="Rule generation instruction")
-    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
-    no_variable: bool = Field(default=False, description="Whether to exclude variables")
-
-
-class RuleCodeGeneratePayload(RuleGeneratePayload):
-    code_language: str = Field(default="javascript", description="Programming language for code generation")
-
-
-class RuleStructuredOutputPayload(BaseModel):
-    instruction: str = Field(..., description="Structured output generation instruction")
-    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
-
-
 class InstructionGeneratePayload(BaseModel):
     flow_id: str = Field(..., description="Workflow/Flow ID")
     node_id: str = Field(default="", description="Node ID for workflow context")
     current: str = Field(default="", description="Current instruction text")
     language: str = Field(default="javascript", description="Programming language (javascript/python)")
     instruction: str = Field(..., description="Instruction for generation")
-    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
+    model_config_data: ModelConfig = Field(..., alias="model_config", description="Model configuration")
     ideal_output: str = Field(default="", description="Expected ideal output")
 
 
@@ -64,6 +50,7 @@ reg(RuleCodeGeneratePayload)
 reg(RuleStructuredOutputPayload)
 reg(InstructionGeneratePayload)
 reg(InstructionTemplatePayload)
+reg(ModelConfig)
 
 
 @console_ns.route("/rule-generate")
@@ -82,12 +69,7 @@ class RuleGenerateApi(Resource):
         _, current_tenant_id = current_account_with_tenant()
 
         try:
-            rules = LLMGenerator.generate_rule_config(
-                tenant_id=current_tenant_id,
-                instruction=args.instruction,
-                model_config=args.model_config_data,
-                no_variable=args.no_variable,
-            )
+            rules = LLMGenerator.generate_rule_config(tenant_id=current_tenant_id, args=args)
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
         except QuotaExceededError:
@@ -118,9 +100,7 @@ class RuleCodeGenerateApi(Resource):
         try:
             code_result = LLMGenerator.generate_code(
                 tenant_id=current_tenant_id,
-                instruction=args.instruction,
-                model_config=args.model_config_data,
-                code_language=args.code_language,
+                args=args,
             )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -152,8 +132,7 @@ class RuleStructuredOutputGenerateApi(Resource):
         try:
             structured_output = LLMGenerator.generate_structured_output(
                 tenant_id=current_tenant_id,
-                instruction=args.instruction,
-                model_config=args.model_config_data,
+                args=args,
             )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -204,23 +183,29 @@ class InstructionGenerateApi(Resource):
                     case "llm":
                         return LLMGenerator.generate_rule_config(
                             current_tenant_id,
-                            instruction=args.instruction,
-                            model_config=args.model_config_data,
-                            no_variable=True,
+                            args=RuleGeneratePayload(
+                                instruction=args.instruction,
+                                model_config=args.model_config_data,
+                                no_variable=True,
+                            ),
                         )
                     case "agent":
                         return LLMGenerator.generate_rule_config(
                             current_tenant_id,
-                            instruction=args.instruction,
-                            model_config=args.model_config_data,
-                            no_variable=True,
+                            args=RuleGeneratePayload(
+                                instruction=args.instruction,
+                                model_config=args.model_config_data,
+                                no_variable=True,
+                            ),
                         )
                     case "code":
                         return LLMGenerator.generate_code(
                             tenant_id=current_tenant_id,
-                            instruction=args.instruction,
-                            model_config=args.model_config_data,
-                            code_language=args.language,
+                            args=RuleCodeGeneratePayload(
+                                instruction=args.instruction,
+                                model_config=args.model_config_data,
+                                code_language=args.language,
+                            ),
                         )
                     case _:
                         return {"error": f"invalid node type: {node_type}"}

api/controllers/console/app/message.py

@@ -7,6 +7,7 @@ from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import exists, select
 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 (
     CompletionRequestError,
@@ -23,7 +24,7 @@ from controllers.console.wraps import (
 )
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from core.model_runtime.errors.invoke import InvokeError
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
 from fields.raws import FilesContainedField
 from libs.helper import TimestampField, uuid_value
@@ -32,10 +33,9 @@ from libs.login import current_account_with_tenant, login_required
 from models.model import AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
 from services.errors.conversation import ConversationNotExistsError
 from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError
-from services.message_service import MessageService
+from services.message_service import MessageService, attach_message_extra_contents
 
 logger = logging.getLogger(__name__)
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
 class ChatMessagesQuery(BaseModel):
@@ -90,13 +90,22 @@ class FeedbackExportQuery(BaseModel):
         raise ValueError("has_comment must be a boolean value")
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+class AnnotationCountResponse(BaseModel):
+    count: int = Field(description="Number of annotations")
 
 
-reg(ChatMessagesQuery)
-reg(MessageFeedbackPayload)
-reg(FeedbackExportQuery)
+class SuggestedQuestionsResponse(BaseModel):
+    data: list[str] = Field(description="Suggested question")
+
+
+register_schema_models(
+    console_ns,
+    ChatMessagesQuery,
+    MessageFeedbackPayload,
+    FeedbackExportQuery,
+    AnnotationCountResponse,
+    SuggestedQuestionsResponse,
+)
 
 # Register models for flask_restx to avoid dict type issues in Swagger
 # Register in dependency order: base models first, then dependent models
@@ -198,6 +207,7 @@ message_detail_model = console_ns.model(
         "created_at": TimestampField,
         "agent_thoughts": fields.List(fields.Nested(agent_thought_model)),
         "message_files": fields.List(fields.Nested(message_file_model)),
+        "extra_contents": fields.List(fields.Raw),
         "metadata": fields.Raw(attribute="message_metadata_dict"),
         "status": fields.String,
         "error": fields.String,
@@ -231,7 +241,7 @@ class ChatMessageListApi(Resource):
     @marshal_with(message_infinite_scroll_pagination_model)
     @edit_permission_required
     def get(self, app_model):
-        args = ChatMessagesQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ChatMessagesQuery.model_validate(request.args.to_dict())
 
         conversation = (
             db.session.query(Conversation)
@@ -290,6 +300,7 @@ class ChatMessageListApi(Resource):
             has_more = False
 
         history_messages = list(reversed(history_messages))
+        attach_message_extra_contents(history_messages)
 
         return InfiniteScrollPagination(data=history_messages, limit=args.limit, has_more=has_more)
 
@@ -356,7 +367,7 @@ class MessageAnnotationCountApi(Resource):
     @console_ns.response(
         200,
         "Annotation count retrieved successfully",
-        console_ns.model("AnnotationCountResponse", {"count": fields.Integer(description="Number of annotations")}),
+        console_ns.models[AnnotationCountResponse.__name__],
     )
     @get_app_model
     @setup_required
@@ -376,9 +387,7 @@ class MessageSuggestedQuestionApi(Resource):
     @console_ns.response(
         200,
         "Suggested questions retrieved successfully",
-        console_ns.model(
-            "SuggestedQuestionsResponse", {"data": fields.List(fields.String(description="Suggested question"))}
-        ),
+        console_ns.models[SuggestedQuestionsResponse.__name__],
     )
     @console_ns.response(404, "Message or conversation not found")
     @setup_required
@@ -428,7 +437,7 @@ class MessageFeedbackExportApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, app_model):
-        args = FeedbackExportQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = FeedbackExportQuery.model_validate(request.args.to_dict())
 
         # Import the service function
         from services.feedback_service import FeedbackService
@@ -474,4 +483,5 @@ class MessageApi(Resource):
         if not message:
             raise NotFound("Message Not Exists.")
 
+        attach_message_extra_contents([message])
         return message

api/controllers/console/app/workflow.py

@@ -20,9 +20,7 @@ from core.app.app_config.features.file_upload.manager import FileUploadConfigMan
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.workflow.app_generator import SKIP_PREPARE_USER_INPUTS_KEY
 from core.app.entities.app_invoke_entities import InvokeFrom
-from core.file.models import File
 from core.helper.trace_id_helper import get_external_trace_id
-from core.model_runtime.utils.encoders import jsonable_encoder
 from core.plugin.impl.exc import PluginInvokeError
 from core.trigger.debug.event_selectors import (
     TriggerDebugEvent,
@@ -30,9 +28,12 @@ from core.trigger.debug.event_selectors import (
     create_event_poller,
     select_trigger_debug_events,
 )
-from core.workflow.enums import NodeType
-from core.workflow.graph_engine.manager import GraphEngineManager
+from dify_graph.enums import NodeType
+from dify_graph.file.models import File
+from dify_graph.graph_engine.manager import GraphEngineManager
+from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from extensions.ext_database import db
+from extensions.ext_redis import redis_client
 from factories import file_factory, variable_factory
 from fields.member_fields import simple_account_fields
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
@@ -507,6 +508,179 @@ class WorkflowDraftRunLoopNodeApi(Resource):
             raise InternalServerError()
 
 
+class HumanInputFormPreviewPayload(BaseModel):
+    inputs: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Values used to fill missing upstream variables referenced in form_content",
+    )
+
+
+class HumanInputFormSubmitPayload(BaseModel):
+    form_inputs: dict[str, Any] = Field(..., description="Values the user provides for the form's own fields")
+    inputs: dict[str, Any] = Field(
+        ...,
+        description="Values used to fill missing upstream variables referenced in form_content",
+    )
+    action: str = Field(..., description="Selected action ID")
+
+
+class HumanInputDeliveryTestPayload(BaseModel):
+    delivery_method_id: str = Field(..., description="Delivery method ID")
+    inputs: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Values used to fill missing upstream variables referenced in form_content",
+    )
+
+
+reg(HumanInputFormPreviewPayload)
+reg(HumanInputFormSubmitPayload)
+reg(HumanInputDeliveryTestPayload)
+
+
+@console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflows/draft/human-input/nodes/<string:node_id>/form/preview")
+class AdvancedChatDraftHumanInputFormPreviewApi(Resource):
+    @console_ns.doc("get_advanced_chat_draft_human_input_form")
+    @console_ns.doc(description="Get human input form preview for advanced chat workflow")
+    @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models[HumanInputFormPreviewPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT])
+    @edit_permission_required
+    def post(self, app_model: App, node_id: str):
+        """
+        Preview human input form content and placeholders
+        """
+        current_user, _ = current_account_with_tenant()
+        args = HumanInputFormPreviewPayload.model_validate(console_ns.payload or {})
+        inputs = args.inputs
+
+        workflow_service = WorkflowService()
+        preview = workflow_service.get_human_input_form_preview(
+            app_model=app_model,
+            account=current_user,
+            node_id=node_id,
+            inputs=inputs,
+        )
+        return jsonable_encoder(preview)
+
+
+@console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflows/draft/human-input/nodes/<string:node_id>/form/run")
+class AdvancedChatDraftHumanInputFormRunApi(Resource):
+    @console_ns.doc("submit_advanced_chat_draft_human_input_form")
+    @console_ns.doc(description="Submit human input form preview for advanced chat workflow")
+    @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models[HumanInputFormSubmitPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT])
+    @edit_permission_required
+    def post(self, app_model: App, node_id: str):
+        """
+        Submit human input form preview
+        """
+        current_user, _ = current_account_with_tenant()
+        args = HumanInputFormSubmitPayload.model_validate(console_ns.payload or {})
+        workflow_service = WorkflowService()
+        result = workflow_service.submit_human_input_form_preview(
+            app_model=app_model,
+            account=current_user,
+            node_id=node_id,
+            form_inputs=args.form_inputs,
+            inputs=args.inputs,
+            action=args.action,
+        )
+        return jsonable_encoder(result)
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/human-input/nodes/<string:node_id>/form/preview")
+class WorkflowDraftHumanInputFormPreviewApi(Resource):
+    @console_ns.doc("get_workflow_draft_human_input_form")
+    @console_ns.doc(description="Get human input form preview for workflow")
+    @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models[HumanInputFormPreviewPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.WORKFLOW])
+    @edit_permission_required
+    def post(self, app_model: App, node_id: str):
+        """
+        Preview human input form content and placeholders
+        """
+        current_user, _ = current_account_with_tenant()
+        args = HumanInputFormPreviewPayload.model_validate(console_ns.payload or {})
+        inputs = args.inputs
+
+        workflow_service = WorkflowService()
+        preview = workflow_service.get_human_input_form_preview(
+            app_model=app_model,
+            account=current_user,
+            node_id=node_id,
+            inputs=inputs,
+        )
+        return jsonable_encoder(preview)
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/human-input/nodes/<string:node_id>/form/run")
+class WorkflowDraftHumanInputFormRunApi(Resource):
+    @console_ns.doc("submit_workflow_draft_human_input_form")
+    @console_ns.doc(description="Submit human input form preview for workflow")
+    @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models[HumanInputFormSubmitPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.WORKFLOW])
+    @edit_permission_required
+    def post(self, app_model: App, node_id: str):
+        """
+        Submit human input form preview
+        """
+        current_user, _ = current_account_with_tenant()
+        workflow_service = WorkflowService()
+        args = HumanInputFormSubmitPayload.model_validate(console_ns.payload or {})
+        result = workflow_service.submit_human_input_form_preview(
+            app_model=app_model,
+            account=current_user,
+            node_id=node_id,
+            form_inputs=args.form_inputs,
+            inputs=args.inputs,
+            action=args.action,
+        )
+        return jsonable_encoder(result)
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/human-input/nodes/<string:node_id>/delivery-test")
+class WorkflowDraftHumanInputDeliveryTestApi(Resource):
+    @console_ns.doc("test_workflow_draft_human_input_delivery")
+    @console_ns.doc(description="Test human input delivery for workflow")
+    @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models[HumanInputDeliveryTestPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.WORKFLOW, AppMode.ADVANCED_CHAT])
+    @edit_permission_required
+    def post(self, app_model: App, node_id: str):
+        """
+        Test human input delivery
+        """
+        current_user, _ = current_account_with_tenant()
+        workflow_service = WorkflowService()
+        args = HumanInputDeliveryTestPayload.model_validate(console_ns.payload or {})
+        workflow_service.test_human_input_delivery(
+            app_model=app_model,
+            account=current_user,
+            node_id=node_id,
+            delivery_method_id=args.delivery_method_id,
+            inputs=args.inputs,
+        )
+        return jsonable_encoder({})
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows/draft/run")
 class DraftWorkflowRunApi(Resource):
     @console_ns.doc("run_draft_workflow")
@@ -567,7 +741,7 @@ class WorkflowTaskStopApi(Resource):
         AppQueueManager.set_stop_flag_no_user_check(task_id)
 
         # New graph engine command channel mechanism
-        GraphEngineManager.send_stop_command(task_id)
+        GraphEngineManager(redis_client).send_stop_command(task_id)
 
         return {"result": "success"}
 

api/controllers/console/app/workflow_app_log.py

@@ -9,7 +9,7 @@ from sqlalchemy.orm import Session
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
-from core.workflow.enums import WorkflowExecutionStatus
+from dify_graph.enums import WorkflowExecutionStatus
 from extensions.ext_database import db
 from fields.workflow_app_log_fields import (
     build_workflow_app_log_pagination_model,

api/controllers/console/app/workflow_draft_variable.py

@@ -15,11 +15,11 @@ from controllers.console.app.error import (
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
 from controllers.web.error import InvalidArgumentError, NotFoundError
-from core.file import helpers as file_helpers
-from core.variables.segment_group import SegmentGroup
-from core.variables.segments import ArrayFileSegment, FileSegment, Segment
-from core.variables.types import SegmentType
-from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from dify_graph.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from dify_graph.file import helpers as file_helpers
+from dify_graph.variables.segment_group import SegmentGroup
+from dify_graph.variables.segments import ArrayFileSegment, FileSegment, Segment
+from dify_graph.variables.types import SegmentType
 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
@@ -112,11 +112,11 @@ _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS = {
     "is_truncated": fields.Boolean(attribute=lambda model: model.file_id is not None),
 }
 
-_WORKFLOW_DRAFT_VARIABLE_FIELDS = dict(
-    _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS,
-    value=fields.Raw(attribute=_serialize_var_value),
-    full_content=fields.Raw(attribute=_serialize_full_content),
-)
+_WORKFLOW_DRAFT_VARIABLE_FIELDS = {
+    **_WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS,
+    "value": fields.Raw(attribute=_serialize_var_value),
+    "full_content": fields.Raw(attribute=_serialize_full_content),
+}
 
 _WORKFLOW_DRAFT_ENV_VARIABLE_FIELDS = {
     "id": fields.String,

api/controllers/console/app/workflow_run.py

@@ -5,10 +5,15 @@ from flask import request
 from flask_restx import Resource, fields, marshal_with
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
 
+from configs import dify_config
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
+from controllers.web.error import NotFoundError
+from dify_graph.entities.pause_reason import HumanInputRequired
+from dify_graph.enums import WorkflowExecutionStatus
 from extensions.ext_database import db
 from fields.end_user_fields import simple_end_user_fields
 from fields.member_fields import simple_account_fields
@@ -27,9 +32,21 @@ from libs.custom_inputs import time_duration
 from libs.helper import uuid_value
 from libs.login import current_user, login_required
 from models import Account, App, AppMode, EndUser, WorkflowArchiveLog, WorkflowRunTriggeredFrom
+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
 
+
+def _build_backstage_input_url(form_token: str | None) -> str | None:
+    if not form_token:
+        return None
+    base_url = dify_config.APP_WEB_URL
+    if not base_url:
+        return None
+    return f"{base_url.rstrip('/')}/form/{form_token}"
+
+
 # Workflow run status choices for filtering
 WORKFLOW_RUN_STATUS_CHOICES = ["running", "succeeded", "failed", "stopped", "partial-succeeded"]
 EXPORT_SIGNED_URL_EXPIRE_SECONDS = 3600
@@ -440,3 +457,68 @@ class WorkflowRunNodeExecutionListApi(Resource):
         )
 
         return {"data": node_executions}
+
+
+@console_ns.route("/workflow/<string:workflow_run_id>/pause-details")
+class ConsoleWorkflowPauseDetailsApi(Resource):
+    """Console API for getting workflow pause details."""
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, workflow_run_id: str):
+        """
+        Get workflow pause details.
+
+        GET /console/api/workflow/<workflow_run_id>/pause-details
+
+        Returns information about why and where the workflow is paused.
+        """
+
+        # Query WorkflowRun to determine if workflow is suspended
+        session_maker = sessionmaker(bind=db.engine)
+        workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker=session_maker)
+
+        workflow_run = db.session.get(WorkflowRun, workflow_run_id)
+        if not workflow_run:
+            raise NotFoundError("Workflow run not found")
+
+        if workflow_run.tenant_id != current_user.current_tenant_id:
+            raise NotFoundError("Workflow run not found")
+
+        # Check if workflow is suspended
+        is_paused = workflow_run.status == WorkflowExecutionStatus.PAUSED
+        if not is_paused:
+            return {
+                "paused_at": None,
+                "paused_nodes": [],
+            }, 200
+
+        pause_entity = workflow_run_repo.get_workflow_pause(workflow_run_id)
+        pause_reasons = pause_entity.get_pause_reasons() if pause_entity else []
+
+        # Build response
+        paused_at = pause_entity.paused_at if pause_entity else None
+        paused_nodes = []
+        response = {
+            "paused_at": paused_at.isoformat() + "Z" if paused_at else None,
+            "paused_nodes": paused_nodes,
+        }
+
+        for reason in pause_reasons:
+            if isinstance(reason, HumanInputRequired):
+                paused_nodes.append(
+                    {
+                        "node_id": reason.node_id,
+                        "node_title": reason.node_title,
+                        "pause_type": {
+                            "type": "human_input",
+                            "form_id": reason.form_id,
+                            "backstage_input_url": _build_backstage_input_url(reason.form_token),
+                        },
+                    }
+                )
+            else:
+                raise AssertionError("unimplemented.")
+
+        return response, 200

api/controllers/console/auth/data_source_oauth.py

@@ -2,9 +2,11 @@ import logging
 
 import httpx
 from flask import current_app, redirect, request
-from flask_restx import Resource, fields
+from flask_restx import Resource
+from pydantic import BaseModel, Field
 
 from configs import dify_config
+from controllers.common.schema import register_schema_models
 from libs.login import login_required
 from libs.oauth_data_source import NotionOAuth
 
@@ -14,6 +16,26 @@ from ..wraps import account_initialization_required, is_admin_or_owner_required,
 logger = logging.getLogger(__name__)
 
 
+class OAuthDataSourceResponse(BaseModel):
+    data: str = Field(description="Authorization URL or 'internal' for internal setup")
+
+
+class OAuthDataSourceBindingResponse(BaseModel):
+    result: str = Field(description="Operation result")
+
+
+class OAuthDataSourceSyncResponse(BaseModel):
+    result: str = Field(description="Operation result")
+
+
+register_schema_models(
+    console_ns,
+    OAuthDataSourceResponse,
+    OAuthDataSourceBindingResponse,
+    OAuthDataSourceSyncResponse,
+)
+
+
 def get_oauth_providers():
     with current_app.app_context():
         notion_oauth = NotionOAuth(
@@ -34,10 +56,7 @@ class OAuthDataSource(Resource):
     @console_ns.response(
         200,
         "Authorization URL or internal setup success",
-        console_ns.model(
-            "OAuthDataSourceResponse",
-            {"data": fields.Raw(description="Authorization URL or 'internal' for internal setup")},
-        ),
+        console_ns.models[OAuthDataSourceResponse.__name__],
     )
     @console_ns.response(400, "Invalid provider")
     @console_ns.response(403, "Admin privileges required")
@@ -101,7 +120,7 @@ class OAuthDataSourceBinding(Resource):
     @console_ns.response(
         200,
         "Data source binding success",
-        console_ns.model("OAuthDataSourceBindingResponse", {"result": fields.String(description="Operation result")}),
+        console_ns.models[OAuthDataSourceBindingResponse.__name__],
     )
     @console_ns.response(400, "Invalid provider or code")
     def get(self, provider: str):
@@ -133,7 +152,7 @@ class OAuthDataSourceSync(Resource):
     @console_ns.response(
         200,
         "Data source sync success",
-        console_ns.model("OAuthDataSourceSyncResponse", {"result": fields.String(description="Operation result")}),
+        console_ns.models[OAuthDataSourceSyncResponse.__name__],
     )
     @console_ns.response(400, "Invalid provider or sync failed")
     @setup_required

api/controllers/console/auth/forgot_password.py

@@ -2,10 +2,11 @@ import base64
 import secrets
 
 from flask import request
-from flask_restx import Resource, fields
+from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy.orm import Session
 
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
     EmailCodeError,
@@ -48,8 +49,31 @@ class ForgotPasswordResetPayload(BaseModel):
         return valid_password(value)
 
 
-for model in (ForgotPasswordSendPayload, ForgotPasswordCheckPayload, ForgotPasswordResetPayload):
-    console_ns.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+class ForgotPasswordEmailResponse(BaseModel):
+    result: str = Field(description="Operation result")
+    data: str | None = Field(default=None, description="Reset token")
+    code: str | None = Field(default=None, description="Error code if account not found")
+
+
+class ForgotPasswordCheckResponse(BaseModel):
+    is_valid: bool = Field(description="Whether code is valid")
+    email: EmailStr = Field(description="Email address")
+    token: str = Field(description="New reset token")
+
+
+class ForgotPasswordResetResponse(BaseModel):
+    result: str = Field(description="Operation result")
+
+
+register_schema_models(
+    console_ns,
+    ForgotPasswordSendPayload,
+    ForgotPasswordCheckPayload,
+    ForgotPasswordResetPayload,
+    ForgotPasswordEmailResponse,
+    ForgotPasswordCheckResponse,
+    ForgotPasswordResetResponse,
+)
 
 
 @console_ns.route("/forgot-password")
@@ -60,14 +84,7 @@ class ForgotPasswordSendEmailApi(Resource):
     @console_ns.response(
         200,
         "Email sent successfully",
-        console_ns.model(
-            "ForgotPasswordEmailResponse",
-            {
-                "result": fields.String(description="Operation result"),
-                "data": fields.String(description="Reset token"),
-                "code": fields.String(description="Error code if account not found"),
-            },
-        ),
+        console_ns.models[ForgotPasswordEmailResponse.__name__],
     )
     @console_ns.response(400, "Invalid email or rate limit exceeded")
     @setup_required
@@ -106,14 +123,7 @@ class ForgotPasswordCheckApi(Resource):
     @console_ns.response(
         200,
         "Code verified successfully",
-        console_ns.model(
-            "ForgotPasswordCheckResponse",
-            {
-                "is_valid": fields.Boolean(description="Whether code is valid"),
-                "email": fields.String(description="Email address"),
-                "token": fields.String(description="New reset token"),
-            },
-        ),
+        console_ns.models[ForgotPasswordCheckResponse.__name__],
     )
     @console_ns.response(400, "Invalid code or token")
     @setup_required
@@ -163,7 +173,7 @@ class ForgotPasswordResetApi(Resource):
     @console_ns.response(
         200,
         "Password reset successfully",
-        console_ns.model("ForgotPasswordResetResponse", {"result": fields.String(description="Operation result")}),
+        console_ns.models[ForgotPasswordResetResponse.__name__],
     )
     @console_ns.response(400, "Invalid token or password mismatch")
     @setup_required

api/controllers/console/auth/oauth_server.py

@@ -8,7 +8,7 @@ from pydantic import BaseModel
 from werkzeug.exceptions import BadRequest, NotFound
 
 from controllers.console.wraps import account_initialization_required, setup_required
-from core.model_runtime.utils.encoders import jsonable_encoder
+from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from models import Account
 from models.model import OAuthProviderApp
@@ -155,43 +155,43 @@ class OAuthServerUserTokenApi(Resource):
             grant_type = OAuthGrantType(payload.grant_type)
         except ValueError:
             raise BadRequest("invalid grant_type")
+        match grant_type:
+            case OAuthGrantType.AUTHORIZATION_CODE:
+                if not payload.code:
+                    raise BadRequest("code is required")
 
-        if grant_type == OAuthGrantType.AUTHORIZATION_CODE:
-            if not payload.code:
-                raise BadRequest("code is required")
+                if payload.client_secret != oauth_provider_app.client_secret:
+                    raise BadRequest("client_secret is invalid")
 
-            if payload.client_secret != oauth_provider_app.client_secret:
-                raise BadRequest("client_secret is invalid")
+                if payload.redirect_uri not in oauth_provider_app.redirect_uris:
+                    raise BadRequest("redirect_uri is invalid")
 
-            if payload.redirect_uri not in oauth_provider_app.redirect_uris:
-                raise BadRequest("redirect_uri is invalid")
+                access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
+                    grant_type, code=payload.code, client_id=oauth_provider_app.client_id
+                )
+                return jsonable_encoder(
+                    {
+                        "access_token": access_token,
+                        "token_type": "Bearer",
+                        "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
+                        "refresh_token": refresh_token,
+                    }
+                )
+            case OAuthGrantType.REFRESH_TOKEN:
+                if not payload.refresh_token:
+                    raise BadRequest("refresh_token is required")
 
-            access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
-                grant_type, code=payload.code, client_id=oauth_provider_app.client_id
-            )
-            return jsonable_encoder(
-                {
-                    "access_token": access_token,
-                    "token_type": "Bearer",
-                    "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
-                    "refresh_token": refresh_token,
-                }
-            )
-        elif grant_type == OAuthGrantType.REFRESH_TOKEN:
-            if not payload.refresh_token:
-                raise BadRequest("refresh_token is required")
-
-            access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
-                grant_type, refresh_token=payload.refresh_token, client_id=oauth_provider_app.client_id
-            )
-            return jsonable_encoder(
-                {
-                    "access_token": access_token,
-                    "token_type": "Bearer",
-                    "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
-                    "refresh_token": refresh_token,
-                }
-            )
+                access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
+                    grant_type, refresh_token=payload.refresh_token, client_id=oauth_provider_app.client_id
+                )
+                return jsonable_encoder(
+                    {
+                        "access_token": access_token,
+                        "token_type": "Bearer",
+                        "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
+                        "refresh_token": refresh_token,
+                    }
+                )
 
 
 @console_ns.route("/oauth/provider/account")

api/controllers/console/datasets/data_source.py

@@ -1,6 +1,6 @@
 import json
 from collections.abc import Generator
-from typing import Any, cast
+from typing import Any, Literal, cast
 
 from flask import request
 from flask_restx import Resource, fields, marshal_with
@@ -157,9 +157,8 @@ class DataSourceApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def patch(self, binding_id, action):
+    def patch(self, binding_id, action: Literal["enable", "disable"]):
         binding_id = str(binding_id)
-        action = str(action)
         with Session(db.engine) as session:
             data_source_binding = session.execute(
                 select(DataSourceOauthBinding).filter_by(id=binding_id)
@@ -167,23 +166,24 @@ class DataSourceApi(Resource):
         if data_source_binding is None:
             raise NotFound("Data source binding not found.")
         # enable binding
-        if action == "enable":
-            if data_source_binding.disabled:
-                data_source_binding.disabled = False
-                data_source_binding.updated_at = naive_utc_now()
-                db.session.add(data_source_binding)
-                db.session.commit()
-            else:
-                raise ValueError("Data source is not disabled.")
-        # disable binding
-        if action == "disable":
-            if not data_source_binding.disabled:
-                data_source_binding.disabled = True
-                data_source_binding.updated_at = naive_utc_now()
-                db.session.add(data_source_binding)
-                db.session.commit()
-            else:
-                raise ValueError("Data source is disabled.")
+        match action:
+            case "enable":
+                if data_source_binding.disabled:
+                    data_source_binding.disabled = False
+                    data_source_binding.updated_at = naive_utc_now()
+                    db.session.add(data_source_binding)
+                    db.session.commit()
+                else:
+                    raise ValueError("Data source is not disabled.")
+            # disable binding
+            case "disable":
+                if not data_source_binding.disabled:
+                    data_source_binding.disabled = True
+                    data_source_binding.updated_at = naive_utc_now()
+                    db.session.add(data_source_binding)
+                    db.session.commit()
+                else:
+                    raise ValueError("Data source is disabled.")
         return {"result": "success"}, 200
 
 

api/controllers/console/datasets/datasets.py

@@ -25,12 +25,12 @@ from controllers.console.wraps import (
 )
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.indexing_runner import IndexingRunner
-from core.model_runtime.entities.model_entities import ModelType
 from core.provider_manager import ProviderManager
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
+from dify_graph.model_runtime.entities.model_entities import ModelType
 from extensions.ext_database import db
 from fields.app_fields import app_detail_kernel_fields, related_app_list
 from fields.dataset_fields import (
@@ -53,8 +53,9 @@ 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.dataset import DatasetPermissionEnum
+from models.dataset import DatasetPermission, DatasetPermissionEnum
 from models.provider_ids import ModelProviderID
+from services.api_token_service import ApiTokenCache
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
 
 # Register models for flask_restx to avoid dict type issues in Swagger
@@ -118,6 +119,14 @@ def _validate_indexing_technique(value: str | None) -> str | None:
     return value
 
 
+def _validate_doc_form(value: str | None) -> str | None:
+    if value is None:
+        return value
+    if value not in Dataset.DOC_FORM_LIST:
+        raise ValueError("Invalid doc_form.")
+    return value
+
+
 class DatasetCreatePayload(BaseModel):
     name: str = Field(..., min_length=1, max_length=40)
     description: str = Field("", max_length=400)
@@ -148,6 +157,7 @@ class DatasetUpdatePayload(BaseModel):
     embedding_model: str | None = None
     embedding_model_provider: str | None = None
     retrieval_model: dict[str, Any] | None = None
+    summary_index_setting: dict[str, Any] | None = None
     partial_member_list: list[dict[str, str]] | None = None
     external_retrieval_model: dict[str, Any] | None = None
     external_knowledge_id: str | None = None
@@ -177,6 +187,14 @@ class IndexingEstimatePayload(BaseModel):
             raise ValueError("indexing_technique is required.")
         return result
 
+    @field_validator("doc_form")
+    @classmethod
+    def validate_doc_form(cls, value: str) -> str:
+        result = _validate_doc_form(value)
+        if result is None:
+            return "text_model"
+        return result
+
 
 class ConsoleDatasetListQuery(BaseModel):
     page: int = Field(default=1, description="Page number")
@@ -288,7 +306,14 @@ class DatasetListApi(Resource):
     @enterprise_license_required
     def get(self):
         current_user, current_tenant_id = current_account_with_tenant()
-        query = ConsoleDatasetListQuery.model_validate(request.args.to_dict())
+        # Convert query parameters to dict, handling list parameters correctly
+        query_params: dict[str, str | list[str]] = dict(request.args.to_dict())
+        # Handle ids and tag_ids as lists (Flask request.args.getlist returns list even for single value)
+        if "ids" in request.args:
+            query_params["ids"] = request.args.getlist("ids")
+        if "tag_ids" in request.args:
+            query_params["tag_ids"] = request.args.getlist("tag_ids")
+        query = ConsoleDatasetListQuery.model_validate(query_params)
         # provider = request.args.get("provider", default="vendor")
         if query.ids:
             datasets, total = DatasetService.get_datasets_by_ids(query.ids, current_tenant_id)
@@ -314,6 +339,18 @@ class DatasetListApi(Resource):
             model_names.append(f"{embedding_model.model}:{embedding_model.provider.provider}")
 
         data = cast(list[dict[str, Any]], marshal(datasets, dataset_detail_fields))
+        dataset_ids = [item["id"] for item in data if item.get("permission") == "partial_members"]
+        partial_members_map: dict[str, list[str]] = {}
+        if dataset_ids:
+            permissions = db.session.execute(
+                select(DatasetPermission.dataset_id, DatasetPermission.account_id).where(
+                    DatasetPermission.dataset_id.in_(dataset_ids)
+                )
+            ).all()
+
+            for dataset_id, account_id in permissions:
+                partial_members_map.setdefault(dataset_id, []).append(account_id)
+
         for item in data:
             # convert embedding_model_provider to plugin standard format
             if item["indexing_technique"] == "high_quality" and item["embedding_model_provider"]:
@@ -327,8 +364,7 @@ class DatasetListApi(Resource):
                 item["embedding_available"] = True
 
             if item.get("permission") == "partial_members":
-                part_users_list = DatasetPermissionService.get_dataset_partial_member_list(item["id"])
-                item.update({"partial_member_list": part_users_list})
+                item.update({"partial_member_list": partial_members_map.get(item["id"], [])})
             else:
                 item.update({"partial_member_list": []})
 
@@ -812,6 +848,11 @@ class DatasetApiDeleteApi(Resource):
         if key is None:
             console_ns.abort(404, message="API key not found")
 
+        # Invalidate cache before deleting from database
+        # Type assertion: key is guaranteed to be non-None here because abort() raises
+        assert key is not None  # nosec - for type checker only
+        ApiTokenCache.delete(key.token, key.type)
+
         db.session.query(ApiToken).where(ApiToken.id == api_key_id).delete()
         db.session.commit()
 

api/controllers/console/datasets/datasets_document.py

@@ -24,11 +24,11 @@ from core.errors.error import (
 )
 from core.indexing_runner import IndexingRunner
 from core.model_manager import ModelManager
-from core.model_runtime.entities.model_entities import ModelType
-from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.plugin.impl.exc import PluginDaemonClientSideError
 from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
+from dify_graph.model_runtime.entities.model_entities import ModelType
+from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
 from extensions.ext_database import db
 from fields.dataset_fields import dataset_fields
 from fields.document_fields import (
@@ -45,6 +45,7 @@ from models.dataset import DocumentPipelineExecutionLog
 from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig, ProcessRule, RetrievalModel
 from services.file_service import FileService
+from tasks.generate_summary_index_task import generate_summary_index_task
 
 from ..app.error import (
     ProviderModelCurrentlyNotSupportError,
@@ -103,6 +104,10 @@ class DocumentRenamePayload(BaseModel):
     name: str
 
 
+class GenerateSummaryPayload(BaseModel):
+    document_list: list[str]
+
+
 class DocumentBatchDownloadZipPayload(BaseModel):
     """Request payload for bulk downloading documents as a zip archive."""
 
@@ -125,6 +130,7 @@ register_schema_models(
     RetrievalModel,
     DocumentRetryPayload,
     DocumentRenamePayload,
+    GenerateSummaryPayload,
     DocumentBatchDownloadZipPayload,
 )
 
@@ -312,6 +318,13 @@ class DatasetDocumentListApi(Resource):
 
         paginated_documents = db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)
         documents = paginated_documents.items
+
+        DocumentService.enrich_documents_with_summary_index_status(
+            documents=documents,
+            dataset=dataset,
+            tenant_id=current_tenant_id,
+        )
+
         if fetch:
             for document in documents:
                 completed_segments = (
@@ -563,63 +576,62 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
             if document.indexing_status in {"completed", "error"}:
                 raise DocumentAlreadyFinishedError()
             data_source_info = document.data_source_info_dict
+            match document.data_source_type:
+                case "upload_file":
+                    if not data_source_info:
+                        continue
+                    file_id = data_source_info["upload_file_id"]
+                    file_detail = (
+                        db.session.query(UploadFile)
+                        .where(UploadFile.tenant_id == current_tenant_id, UploadFile.id == file_id)
+                        .first()
+                    )
 
-            if document.data_source_type == "upload_file":
-                if not data_source_info:
-                    continue
-                file_id = data_source_info["upload_file_id"]
-                file_detail = (
-                    db.session.query(UploadFile)
-                    .where(UploadFile.tenant_id == current_tenant_id, UploadFile.id == file_id)
-                    .first()
-                )
+                    if file_detail is None:
+                        raise NotFound("File not found.")
 
-                if file_detail is None:
-                    raise NotFound("File not found.")
+                    extract_setting = ExtractSetting(
+                        datasource_type=DatasourceType.FILE, upload_file=file_detail, document_model=document.doc_form
+                    )
+                    extract_settings.append(extract_setting)
+                case "notion_import":
+                    if not data_source_info:
+                        continue
+                    extract_setting = ExtractSetting(
+                        datasource_type=DatasourceType.NOTION,
+                        notion_info=NotionInfo.model_validate(
+                            {
+                                "credential_id": data_source_info.get("credential_id"),
+                                "notion_workspace_id": data_source_info["notion_workspace_id"],
+                                "notion_obj_id": data_source_info["notion_page_id"],
+                                "notion_page_type": data_source_info["type"],
+                                "tenant_id": current_tenant_id,
+                            }
+                        ),
+                        document_model=document.doc_form,
+                    )
+                    extract_settings.append(extract_setting)
+                case "website_crawl":
+                    if not data_source_info:
+                        continue
+                    extract_setting = ExtractSetting(
+                        datasource_type=DatasourceType.WEBSITE,
+                        website_info=WebsiteInfo.model_validate(
+                            {
+                                "provider": data_source_info["provider"],
+                                "job_id": data_source_info["job_id"],
+                                "url": data_source_info["url"],
+                                "tenant_id": current_tenant_id,
+                                "mode": data_source_info["mode"],
+                                "only_main_content": data_source_info["only_main_content"],
+                            }
+                        ),
+                        document_model=document.doc_form,
+                    )
+                    extract_settings.append(extract_setting)
 
-                extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.FILE, upload_file=file_detail, document_model=document.doc_form
-                )
-                extract_settings.append(extract_setting)
-
-            elif document.data_source_type == "notion_import":
-                if not data_source_info:
-                    continue
-                extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.NOTION,
-                    notion_info=NotionInfo.model_validate(
-                        {
-                            "credential_id": data_source_info.get("credential_id"),
-                            "notion_workspace_id": data_source_info["notion_workspace_id"],
-                            "notion_obj_id": data_source_info["notion_page_id"],
-                            "notion_page_type": data_source_info["type"],
-                            "tenant_id": current_tenant_id,
-                        }
-                    ),
-                    document_model=document.doc_form,
-                )
-                extract_settings.append(extract_setting)
-            elif document.data_source_type == "website_crawl":
-                if not data_source_info:
-                    continue
-                extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.WEBSITE,
-                    website_info=WebsiteInfo.model_validate(
-                        {
-                            "provider": data_source_info["provider"],
-                            "job_id": data_source_info["job_id"],
-                            "url": data_source_info["url"],
-                            "tenant_id": current_tenant_id,
-                            "mode": data_source_info["mode"],
-                            "only_main_content": data_source_info["only_main_content"],
-                        }
-                    ),
-                    document_model=document.doc_form,
-                )
-                extract_settings.append(extract_setting)
-
-            else:
-                raise ValueError("Data source type not support")
+                case _:
+                    raise ValueError("Data source type not support")
             indexing_runner = IndexingRunner()
             try:
                 response = indexing_runner.indexing_estimate(
@@ -797,6 +809,7 @@ class DocumentApi(DocumentResource):
                 "display_status": document.display_status,
                 "doc_form": document.doc_form,
                 "doc_language": document.doc_language,
+                "need_summary": document.need_summary if document.need_summary is not None else False,
             }
         else:
             dataset_process_rules = DatasetService.get_process_rules(dataset_id)
@@ -832,6 +845,7 @@ class DocumentApi(DocumentResource):
                 "display_status": document.display_status,
                 "doc_form": document.doc_form,
                 "doc_language": document.doc_language,
+                "need_summary": document.need_summary if document.need_summary is not None else False,
             }
 
         return response, 200
@@ -939,23 +953,24 @@ class DocumentProcessingApi(DocumentResource):
         if not current_user.is_dataset_editor:
             raise Forbidden()
 
-        if action == "pause":
-            if document.indexing_status != "indexing":
-                raise InvalidActionError("Document not in indexing state.")
+        match action:
+            case "pause":
+                if document.indexing_status != "indexing":
+                    raise InvalidActionError("Document not in indexing state.")
 
-            document.paused_by = current_user.id
-            document.paused_at = naive_utc_now()
-            document.is_paused = True
-            db.session.commit()
+                document.paused_by = current_user.id
+                document.paused_at = naive_utc_now()
+                document.is_paused = True
+                db.session.commit()
 
-        elif action == "resume":
-            if document.indexing_status not in {"paused", "error"}:
-                raise InvalidActionError("Document not in paused or error state.")
+            case "resume":
+                if document.indexing_status not in {"paused", "error"}:
+                    raise InvalidActionError("Document not in paused or error state.")
 
-            document.paused_by = None
-            document.paused_at = None
-            document.is_paused = False
-            db.session.commit()
+                document.paused_by = None
+                document.paused_at = None
+                document.is_paused = False
+                db.session.commit()
 
         return {"result": "success"}, 200
 
@@ -1255,3 +1270,149 @@ class DocumentPipelineExecutionLogApi(DocumentResource):
             "input_data": log.input_data,
             "datasource_node_id": log.datasource_node_id,
         }, 200
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/generate-summary")
+class DocumentGenerateSummaryApi(Resource):
+    @console_ns.doc("generate_summary_for_documents")
+    @console_ns.doc(description="Generate summary index for documents")
+    @console_ns.doc(params={"dataset_id": "Dataset ID"})
+    @console_ns.expect(console_ns.models[GenerateSummaryPayload.__name__])
+    @console_ns.response(200, "Summary generation started successfully")
+    @console_ns.response(400, "Invalid request or dataset configuration")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @cloud_edition_billing_rate_limit_check("knowledge")
+    def post(self, dataset_id):
+        """
+        Generate summary index for specified documents.
+
+        This endpoint checks if the dataset configuration supports summary generation
+        (indexing_technique must be 'high_quality' and summary_index_setting.enable must be true),
+        then asynchronously generates summary indexes for the provided documents.
+        """
+        current_user, _ = current_account_with_tenant()
+        dataset_id = str(dataset_id)
+
+        # Get dataset
+        dataset = DatasetService.get_dataset(dataset_id)
+        if not dataset:
+            raise NotFound("Dataset not found.")
+
+        # Check permissions
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
+
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        # Validate request payload
+        payload = GenerateSummaryPayload.model_validate(console_ns.payload or {})
+        document_list = payload.document_list
+
+        if not document_list:
+            from werkzeug.exceptions import BadRequest
+
+            raise BadRequest("document_list cannot be empty.")
+
+        # Check if dataset configuration supports summary generation
+        if dataset.indexing_technique != "high_quality":
+            raise ValueError(
+                f"Summary generation is only available for 'high_quality' indexing technique. "
+                f"Current indexing technique: {dataset.indexing_technique}"
+            )
+
+        summary_index_setting = dataset.summary_index_setting
+        if not summary_index_setting or not summary_index_setting.get("enable"):
+            raise ValueError("Summary index is not enabled for this dataset. Please enable it in the dataset settings.")
+
+        # Verify all documents exist and belong to the dataset
+        documents = DocumentService.get_documents_by_ids(dataset_id, document_list)
+
+        if len(documents) != len(document_list):
+            found_ids = {doc.id for doc in documents}
+            missing_ids = set(document_list) - found_ids
+            raise NotFound(f"Some documents not found: {list(missing_ids)}")
+
+        # Update need_summary to True for documents that don't have it set
+        # This handles the case where documents were created when summary_index_setting was disabled
+        documents_to_update = [doc for doc in documents if not doc.need_summary and doc.doc_form != "qa_model"]
+
+        if documents_to_update:
+            document_ids_to_update = [str(doc.id) for doc in documents_to_update]
+            DocumentService.update_documents_need_summary(
+                dataset_id=dataset_id,
+                document_ids=document_ids_to_update,
+                need_summary=True,
+            )
+
+        # Dispatch async tasks for each document
+        for document in documents:
+            # Skip qa_model documents as they don't generate summaries
+            if document.doc_form == "qa_model":
+                logger.info("Skipping summary generation for qa_model document %s", document.id)
+                continue
+
+            # Dispatch async task
+            generate_summary_index_task.delay(dataset_id, document.id)
+            logger.info(
+                "Dispatched summary generation task for document %s in dataset %s",
+                document.id,
+                dataset_id,
+            )
+
+        return {"result": "success"}, 200
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/summary-status")
+class DocumentSummaryStatusApi(DocumentResource):
+    @console_ns.doc("get_document_summary_status")
+    @console_ns.doc(description="Get summary index generation status for a document")
+    @console_ns.doc(params={"dataset_id": "Dataset ID", "document_id": "Document ID"})
+    @console_ns.response(200, "Summary status retrieved successfully")
+    @console_ns.response(404, "Document not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id, document_id):
+        """
+        Get summary index generation status for a document.
+
+        Returns:
+        - total_segments: Total number of segments in the document
+        - summary_status: Dictionary with status counts
+          - completed: Number of summaries completed
+          - generating: Number of summaries being generated
+          - error: Number of summaries with errors
+          - not_started: Number of segments without summary records
+        - summaries: List of summary records with status and content preview
+        """
+        current_user, _ = current_account_with_tenant()
+        dataset_id = str(dataset_id)
+        document_id = str(document_id)
+
+        # Get dataset
+        dataset = DatasetService.get_dataset(dataset_id)
+        if not dataset:
+            raise NotFound("Dataset not found.")
+
+        # Check permissions
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        # Get summary status detail from service
+        from services.summary_index_service import SummaryIndexService
+
+        result = SummaryIndexService.get_document_summary_status_detail(
+            document_id=document_id,
+            dataset_id=dataset_id,
+        )
+
+        return result, 200

api/controllers/console/datasets/datasets_segments.py

@@ -26,7 +26,7 @@ from controllers.console.wraps import (
 )
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.model_manager import ModelManager
-from core.model_runtime.entities.model_entities import ModelType
+from dify_graph.model_runtime.entities.model_entities import ModelType
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.segment_fields import child_chunk_fields, segment_fields
@@ -41,6 +41,17 @@ from services.errors.chunk import ChildChunkIndexingError as ChildChunkIndexingS
 from tasks.batch_create_segment_to_index_task import batch_create_segment_to_index_task
 
 
+def _get_segment_with_summary(segment, dataset_id):
+    """Helper function to marshal segment and add summary information."""
+    from services.summary_index_service import SummaryIndexService
+
+    segment_dict = dict(marshal(segment, segment_fields))
+    # Query summary for this segment (only enabled summaries)
+    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
+
+
 class SegmentListQuery(BaseModel):
     limit: int = Field(default=20, ge=1, le=100)
     status: list[str] = Field(default_factory=list)
@@ -63,6 +74,7 @@ class SegmentUpdatePayload(BaseModel):
     keywords: list[str] | None = None
     regenerate_child_chunks: bool = False
     attachment_ids: list[str] | None = None
+    summary: str | None = None  # Summary content for summary index
 
 
 class BatchImportPayload(BaseModel):
@@ -181,8 +193,25 @@ class DatasetDocumentSegmentListApi(Resource):
 
         segments = db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)
 
+        # Query summaries for all segments in this page (batch query for efficiency)
+        segment_ids = [segment.id for segment in segments.items]
+        summaries = {}
+        if segment_ids:
+            from services.summary_index_service import SummaryIndexService
+
+            summary_records = SummaryIndexService.get_segments_summaries(segment_ids=segment_ids, dataset_id=dataset_id)
+            # Only include enabled summaries (already filtered by service)
+            summaries = {chunk_id: summary.summary_content for chunk_id, summary in summary_records.items()}
+
+        # Add summary to each segment
+        segments_with_summary = []
+        for segment in segments.items:
+            segment_dict = dict(marshal(segment, segment_fields))
+            segment_dict["summary"] = summaries.get(segment.id)
+            segments_with_summary.append(segment_dict)
+
         response = {
-            "data": marshal(segments.items, segment_fields),
+            "data": segments_with_summary,
             "limit": limit,
             "total": segments.total,
             "total_pages": segments.pages,
@@ -328,7 +357,7 @@ class DatasetDocumentSegmentAddApi(Resource):
         payload_dict = payload.model_dump(exclude_none=True)
         SegmentService.segment_create_args_validate(payload_dict, document)
         segment = SegmentService.create_segment(payload_dict, document, dataset)
-        return {"data": marshal(segment, segment_fields), "doc_form": document.doc_form}, 200
+        return {"data": _get_segment_with_summary(segment, dataset_id), "doc_form": document.doc_form}, 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segments/<uuid:segment_id>")
@@ -390,10 +419,12 @@ class DatasetDocumentSegmentUpdateApi(Resource):
         payload = SegmentUpdatePayload.model_validate(console_ns.payload or {})
         payload_dict = payload.model_dump(exclude_none=True)
         SegmentService.segment_create_args_validate(payload_dict, document)
+
+        # Update segment (summary update with change detection is handled in SegmentService.update_segment)
         segment = SegmentService.update_segment(
             SegmentUpdateArgs.model_validate(payload.model_dump(exclude_none=True)), segment, document, dataset
         )
-        return {"data": marshal(segment, segment_fields), "doc_form": document.doc_form}, 200
+        return {"data": _get_segment_with_summary(segment, dataset_id), "doc_form": document.doc_form}, 200
 
     @setup_required
     @login_required

api/controllers/console/datasets/hit_testing.py

@@ -1,6 +1,13 @@
-from flask_restx import Resource
+from flask_restx import Resource, fields
 
 from controllers.common.schema import register_schema_model
+from fields.hit_testing_fields import (
+    child_chunk_fields,
+    document_fields,
+    files_fields,
+    hit_testing_record_fields,
+    segment_fields,
+)
 from libs.login import login_required
 
 from .. import console_ns
@@ -14,13 +21,45 @@ from ..wraps import (
 register_schema_model(console_ns, HitTestingPayload)
 
 
+def _get_or_create_model(model_name: str, field_def):
+    """Get or create a flask_restx model to avoid dict type issues in Swagger."""
+    existing = console_ns.models.get(model_name)
+    if existing is None:
+        existing = console_ns.model(model_name, field_def)
+    return existing
+
+
+# Register models for flask_restx to avoid dict type issues in Swagger
+document_model = _get_or_create_model("HitTestingDocument", document_fields)
+
+segment_fields_copy = segment_fields.copy()
+segment_fields_copy["document"] = fields.Nested(document_model)
+segment_model = _get_or_create_model("HitTestingSegment", segment_fields_copy)
+
+child_chunk_model = _get_or_create_model("HitTestingChildChunk", child_chunk_fields)
+files_model = _get_or_create_model("HitTestingFile", files_fields)
+
+hit_testing_record_fields_copy = hit_testing_record_fields.copy()
+hit_testing_record_fields_copy["segment"] = fields.Nested(segment_model)
+hit_testing_record_fields_copy["child_chunks"] = fields.List(fields.Nested(child_chunk_model))
+hit_testing_record_fields_copy["files"] = fields.List(fields.Nested(files_model))
+hit_testing_record_model = _get_or_create_model("HitTestingRecord", hit_testing_record_fields_copy)
+
+# Response model for hit testing API
+hit_testing_response_fields = {
+    "query": fields.String,
+    "records": fields.List(fields.Nested(hit_testing_record_model)),
+}
+hit_testing_response_model = _get_or_create_model("HitTestingResponse", hit_testing_response_fields)
+
+
 @console_ns.route("/datasets/<uuid:dataset_id>/hit-testing")
 class HitTestingApi(Resource, DatasetsHitTestingBase):
     @console_ns.doc("test_dataset_retrieval")
     @console_ns.doc(description="Test dataset knowledge retrieval")
     @console_ns.doc(params={"dataset_id": "Dataset ID"})
     @console_ns.expect(console_ns.models[HitTestingPayload.__name__])
-    @console_ns.response(200, "Hit testing completed successfully")
+    @console_ns.response(200, "Hit testing completed successfully", model=hit_testing_response_model)
     @console_ns.response(404, "Dataset not found")
     @console_ns.response(400, "Invalid parameters")
     @setup_required

api/controllers/console/datasets/hit_testing_base.py

@@ -19,7 +19,7 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
-from core.model_runtime.errors.invoke import InvokeError
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from fields.hit_testing_fields import hit_testing_record_fields
 from libs.login import current_user
 from models.account import Account

api/controllers/console/datasets/metadata.py

@@ -126,10 +126,11 @@ class DatasetMetadataBuiltInFieldActionApi(Resource):
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user)
 
-        if action == "enable":
-            MetadataService.enable_built_in_field(dataset)
-        elif action == "disable":
-            MetadataService.disable_built_in_field(dataset)
+        match action:
+            case "enable":
+                MetadataService.enable_built_in_field(dataset)
+            case "disable":
+                MetadataService.disable_built_in_field(dataset)
         return {"result": "success"}, 200
 
 

api/controllers/console/datasets/rag_pipeline/datasource_auth.py

@@ -9,9 +9,9 @@ from configs import dify_config
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
-from core.model_runtime.errors.validate import CredentialsValidateFailedError
-from core.model_runtime.utils.encoders import jsonable_encoder
 from core.plugin.impl.oauth import OAuthHandler
+from dify_graph.model_runtime.errors.validate import CredentialsValidateFailedError
+from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from models.provider_ids import DatasourceProviderID
 from services.datasource_provider_service import DatasourceProviderService

api/controllers/console/datasets/rag_pipeline/rag_pipeline_draft_variable.py

@@ -21,8 +21,8 @@ from controllers.console.app.workflow_draft_variable import (
 from controllers.console.datasets.wraps import get_rag_pipeline
 from controllers.console.wraps import account_initialization_required, setup_required
 from controllers.web.error import InvalidArgumentError, NotFoundError
-from core.variables.types import SegmentType
-from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from dify_graph.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from dify_graph.variables.types import SegmentType
 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

api/controllers/console/datasets/rag_pipeline/rag_pipeline_workflow.py

@@ -1,10 +1,9 @@
 import json
 import logging
 from typing import Any, Literal, cast
-from uuid import UUID
 
 from flask import abort, request
-from flask_restx import Resource, marshal_with, reqparse  # type: ignore
+from flask_restx import Resource, marshal_with  # type: ignore
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
@@ -34,11 +33,11 @@ from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpErr
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.pipeline.pipeline_generator import PipelineGenerator
 from core.app.entities.app_invoke_entities import InvokeFrom
-from core.model_runtime.utils.encoders import jsonable_encoder
+from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from extensions.ext_database import db
 from factories import variable_factory
 from libs import helper
-from libs.helper import TimestampField
+from libs.helper import TimestampField, UUIDStrOrEmpty
 from libs.login import current_account_with_tenant, current_user, login_required
 from models import Account
 from models.dataset import Pipeline
@@ -110,7 +109,7 @@ class NodeIdQuery(BaseModel):
 
 
 class WorkflowRunQuery(BaseModel):
-    last_id: UUID | None = None
+    last_id: UUIDStrOrEmpty | None = None
     limit: int = Field(default=20, ge=1, le=100)
 
 
@@ -121,6 +120,10 @@ class DatasourceVariablesPayload(BaseModel):
     start_node_title: str
 
 
+class RagPipelineRecommendedPluginQuery(BaseModel):
+    type: str = "all"
+
+
 register_schema_models(
     console_ns,
     DraftWorkflowSyncPayload,
@@ -135,6 +138,7 @@ register_schema_models(
     NodeIdQuery,
     WorkflowRunQuery,
     DatasourceVariablesPayload,
+    RagPipelineRecommendedPluginQuery,
 )
 
 
@@ -975,11 +979,8 @@ class RagPipelineRecommendedPluginApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("type", type=str, location="args", required=False, default="all")
-        args = parser.parse_args()
-        type = args["type"]
+        query = RagPipelineRecommendedPluginQuery.model_validate(request.args.to_dict())
 
         rag_pipeline_service = RagPipelineService()
-        recommended_plugins = rag_pipeline_service.get_recommended_plugins(type)
+        recommended_plugins = rag_pipeline_service.get_recommended_plugins(query.type)
         return recommended_plugins

api/controllers/console/explore/audio.py

@@ -19,7 +19,7 @@ from controllers.console.app.error import (
 )
 from controllers.console.explore.wraps import InstalledAppResource
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from core.model_runtime.errors.invoke import InvokeError
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from services.audio_service import AudioService
 from services.errors.audio import (
     AudioTooLargeServiceError,

api/controllers/console/explore/completion.py

@@ -24,7 +24,7 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
-from core.model_runtime.errors.invoke import InvokeError
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
 from libs import helper
 from libs.datetime_utils import naive_utc_now

api/controllers/console/explore/message.py

@@ -21,7 +21,7 @@ from controllers.console.explore.error import (
 from controllers.console.explore.wraps import InstalledAppResource
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from core.model_runtime.errors.invoke import InvokeError
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import MessageInfiniteScrollPagination, MessageListItem, SuggestedQuestionsResponse
 from libs import helper

api/controllers/console/explore/parameter.py

@@ -1,3 +1,5 @@
+from typing import Any, cast
+
 from controllers.common import fields
 from controllers.console import console_ns
 from controllers.console.app.error import AppUnavailableError
@@ -23,14 +25,14 @@ class AppParameterApi(InstalledAppResource):
             if workflow is None:
                 raise AppUnavailableError()
 
-            features_dict = workflow.features_dict
+            features_dict: dict[str, Any] = workflow.features_dict
             user_input_form = workflow.user_input_form(to_old_structure=True)
         else:
             app_model_config = app_model.app_model_config
             if app_model_config is None:
                 raise AppUnavailableError()
 
-            features_dict = app_model_config.to_dict()
+            features_dict = cast(dict[str, Any], app_model_config.to_dict())
 
             user_input_form = features_dict.get("user_input_form", [])
 

api/controllers/console/explore/trial.py

@@ -1,15 +1,16 @@
 import logging
-from typing import Any, cast
+from typing import Any, Literal, cast
 
 from flask import request
-from flask_restx import Resource, fields, marshal, marshal_with, reqparse
+from flask_restx import Resource, fields, marshal, marshal_with
+from pydantic import BaseModel
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
 from controllers.common.fields import Parameters as ParametersResponse
 from controllers.common.fields import Site as SiteResponse
 from controllers.common.schema import get_or_create_model
-from controllers.console import api, console_ns
+from controllers.console import console_ns
 from controllers.console.app.error import (
     AppUnavailableError,
     AudioTooLargeError,
@@ -40,9 +41,10 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
-from core.model_runtime.errors.invoke import InvokeError
-from core.workflow.graph_engine.manager import GraphEngineManager
+from dify_graph.graph_engine.manager import GraphEngineManager
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
+from extensions.ext_redis import redis_client
 from fields.app_fields import (
     app_detail_fields_with_site,
     deleted_tool_fields,
@@ -51,7 +53,7 @@ from fields.app_fields import (
     tag_fields,
 )
 from fields.dataset_fields import dataset_fields
-from fields.member_fields import build_simple_account_model
+from fields.member_fields import simple_account_fields
 from fields.workflow_fields import (
     conversation_variable_fields,
     pipeline_variable_fields,
@@ -103,7 +105,7 @@ app_detail_fields_with_site_copy["tags"] = fields.List(fields.Nested(tag_model))
 app_detail_fields_with_site_copy["site"] = fields.Nested(site_model)
 app_detail_with_site_model = get_or_create_model("TrialAppDetailWithSite", app_detail_fields_with_site_copy)
 
-simple_account_model = build_simple_account_model(console_ns)
+simple_account_model = get_or_create_model("SimpleAccount", simple_account_fields)
 conversation_variable_model = get_or_create_model("TrialConversationVariable", conversation_variable_fields)
 pipeline_variable_model = get_or_create_model("TrialPipelineVariable", pipeline_variable_fields)
 
@@ -117,7 +119,56 @@ workflow_fields_copy["rag_pipeline_variables"] = fields.List(fields.Nested(pipel
 workflow_model = get_or_create_model("TrialWorkflow", workflow_fields_copy)
 
 
+# Pydantic models for request validation
+DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
+
+
+class WorkflowRunRequest(BaseModel):
+    inputs: dict
+    files: list | None = None
+
+
+class ChatRequest(BaseModel):
+    inputs: dict
+    query: str
+    files: list | None = None
+    conversation_id: str | None = None
+    parent_message_id: str | None = None
+    retriever_from: str = "explore_app"
+
+
+class TextToSpeechRequest(BaseModel):
+    message_id: str | None = None
+    voice: str | None = None
+    text: str | None = None
+    streaming: bool | None = None
+
+
+class CompletionRequest(BaseModel):
+    inputs: dict
+    query: str = ""
+    files: list | None = None
+    response_mode: Literal["blocking", "streaming"] | None = None
+    retriever_from: str = "explore_app"
+
+
+# Register schemas for Swagger documentation
+console_ns.schema_model(
+    WorkflowRunRequest.__name__, WorkflowRunRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+console_ns.schema_model(
+    ChatRequest.__name__, ChatRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+console_ns.schema_model(
+    TextToSpeechRequest.__name__, TextToSpeechRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+console_ns.schema_model(
+    CompletionRequest.__name__, CompletionRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+
+
 class TrialAppWorkflowRunApi(TrialAppResource):
+    @console_ns.expect(console_ns.models[WorkflowRunRequest.__name__])
     def post(self, trial_app):
         """
         Run workflow
@@ -129,10 +180,8 @@ class TrialAppWorkflowRunApi(TrialAppResource):
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
 
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
-        parser.add_argument("files", type=list, required=False, location="json")
-        args = parser.parse_args()
+        request_data = WorkflowRunRequest.model_validate(console_ns.payload)
+        args = request_data.model_dump()
         assert current_user is not None
         try:
             app_id = app_model.id
@@ -177,12 +226,13 @@ class TrialAppWorkflowTaskStopApi(TrialAppResource):
         AppQueueManager.set_stop_flag_no_user_check(task_id)
 
         # New graph engine command channel mechanism
-        GraphEngineManager.send_stop_command(task_id)
+        GraphEngineManager(redis_client).send_stop_command(task_id)
 
         return {"result": "success"}
 
 
 class TrialChatApi(TrialAppResource):
+    @console_ns.expect(console_ns.models[ChatRequest.__name__])
     @trial_feature_enable
     def post(self, trial_app):
         app_model = trial_app
@@ -190,14 +240,14 @@ class TrialChatApi(TrialAppResource):
         if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, location="json")
-        parser.add_argument("query", type=str, required=True, location="json")
-        parser.add_argument("files", type=list, required=False, location="json")
-        parser.add_argument("conversation_id", type=uuid_value, location="json")
-        parser.add_argument("parent_message_id", type=uuid_value, required=False, location="json")
-        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
-        args = parser.parse_args()
+        request_data = ChatRequest.model_validate(console_ns.payload)
+        args = request_data.model_dump()
+
+        # Validate UUID values if provided
+        if args.get("conversation_id"):
+            args["conversation_id"] = uuid_value(args["conversation_id"])
+        if args.get("parent_message_id"):
+            args["parent_message_id"] = uuid_value(args["parent_message_id"])
 
         args["auto_generate_name"] = False
 
@@ -320,20 +370,16 @@ class TrialChatAudioApi(TrialAppResource):
 
 
 class TrialChatTextApi(TrialAppResource):
+    @console_ns.expect(console_ns.models[TextToSpeechRequest.__name__])
     @trial_feature_enable
     def post(self, trial_app):
         app_model = trial_app
         try:
-            parser = reqparse.RequestParser()
-            parser.add_argument("message_id", type=str, required=False, location="json")
-            parser.add_argument("voice", type=str, location="json")
-            parser.add_argument("text", type=str, location="json")
-            parser.add_argument("streaming", type=bool, location="json")
-            args = parser.parse_args()
+            request_data = TextToSpeechRequest.model_validate(console_ns.payload)
 
-            message_id = args.get("message_id", None)
-            text = args.get("text", None)
-            voice = args.get("voice", None)
+            message_id = request_data.message_id
+            text = request_data.text
+            voice = request_data.voice
             if not isinstance(current_user, Account):
                 raise ValueError("current_user must be an Account instance")
 
@@ -371,19 +417,15 @@ class TrialChatTextApi(TrialAppResource):
 
 
 class TrialCompletionApi(TrialAppResource):
+    @console_ns.expect(console_ns.models[CompletionRequest.__name__])
     @trial_feature_enable
     def post(self, trial_app):
         app_model = trial_app
         if app_model.mode != "completion":
             raise NotCompletionAppError()
 
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, location="json")
-        parser.add_argument("query", type=str, location="json", default="")
-        parser.add_argument("files", type=list, required=False, location="json")
-        parser.add_argument("response_mode", type=str, choices=["blocking", "streaming"], location="json")
-        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
-        args = parser.parse_args()
+        request_data = CompletionRequest.model_validate(console_ns.payload)
+        args = request_data.model_dump()
 
         streaming = args["response_mode"] == "streaming"
         args["auto_generate_name"] = False
@@ -428,7 +470,7 @@ class TrialSitApi(Resource):
     """Resource for trial app sites."""
 
     @trial_feature_enable
-    @get_app_model_with_trial
+    @get_app_model_with_trial(None)
     def get(self, app_model):
         """Retrieve app site info.
 
@@ -450,7 +492,7 @@ class TrialAppParameterApi(Resource):
     """Resource for app variables."""
 
     @trial_feature_enable
-    @get_app_model_with_trial
+    @get_app_model_with_trial(None)
     def get(self, app_model):
         """Retrieve app parameters."""
 
@@ -479,7 +521,7 @@ class TrialAppParameterApi(Resource):
 
 class AppApi(Resource):
     @trial_feature_enable
-    @get_app_model_with_trial
+    @get_app_model_with_trial(None)
     @marshal_with(app_detail_with_site_model)
     def get(self, app_model):
         """Get app detail"""
@@ -492,7 +534,7 @@ class AppApi(Resource):
 
 class AppWorkflowApi(Resource):
     @trial_feature_enable
-    @get_app_model_with_trial
+    @get_app_model_with_trial(None)
     @marshal_with(workflow_model)
     def get(self, app_model):
         """Get workflow detail"""
@@ -511,7 +553,7 @@ class AppWorkflowApi(Resource):
 
 class DatasetListApi(Resource):
     @trial_feature_enable
-    @get_app_model_with_trial
+    @get_app_model_with_trial(None)
     def get(self, app_model):
         page = request.args.get("page", default=1, type=int)
         limit = request.args.get("limit", default=20, type=int)
@@ -529,27 +571,31 @@ class DatasetListApi(Resource):
         return response
 
 
-api.add_resource(TrialChatApi, "/trial-apps/<uuid:app_id>/chat-messages", endpoint="trial_app_chat_completion")
+console_ns.add_resource(TrialChatApi, "/trial-apps/<uuid:app_id>/chat-messages", endpoint="trial_app_chat_completion")
 
-api.add_resource(
+console_ns.add_resource(
     TrialMessageSuggestedQuestionApi,
     "/trial-apps/<uuid:app_id>/messages/<uuid:message_id>/suggested-questions",
     endpoint="trial_app_suggested_question",
 )
 
-api.add_resource(TrialChatAudioApi, "/trial-apps/<uuid:app_id>/audio-to-text", endpoint="trial_app_audio")
-api.add_resource(TrialChatTextApi, "/trial-apps/<uuid:app_id>/text-to-audio", endpoint="trial_app_text")
+console_ns.add_resource(TrialChatAudioApi, "/trial-apps/<uuid:app_id>/audio-to-text", endpoint="trial_app_audio")
+console_ns.add_resource(TrialChatTextApi, "/trial-apps/<uuid:app_id>/text-to-audio", endpoint="trial_app_text")
 
-api.add_resource(TrialCompletionApi, "/trial-apps/<uuid:app_id>/completion-messages", endpoint="trial_app_completion")
+console_ns.add_resource(
+    TrialCompletionApi, "/trial-apps/<uuid:app_id>/completion-messages", endpoint="trial_app_completion"
+)
 
-api.add_resource(TrialSitApi, "/trial-apps/<uuid:app_id>/site")
+console_ns.add_resource(TrialSitApi, "/trial-apps/<uuid:app_id>/site")
 
-api.add_resource(TrialAppParameterApi, "/trial-apps/<uuid:app_id>/parameters", endpoint="trial_app_parameters")
+console_ns.add_resource(TrialAppParameterApi, "/trial-apps/<uuid:app_id>/parameters", endpoint="trial_app_parameters")
 
-api.add_resource(AppApi, "/trial-apps/<uuid:app_id>", endpoint="trial_app")
+console_ns.add_resource(AppApi, "/trial-apps/<uuid:app_id>", endpoint="trial_app")
 
-api.add_resource(TrialAppWorkflowRunApi, "/trial-apps/<uuid:app_id>/workflows/run", endpoint="trial_app_workflow_run")
-api.add_resource(TrialAppWorkflowTaskStopApi, "/trial-apps/<uuid:app_id>/workflows/tasks/<string:task_id>/stop")
+console_ns.add_resource(
+    TrialAppWorkflowRunApi, "/trial-apps/<uuid:app_id>/workflows/run", endpoint="trial_app_workflow_run"
+)
+console_ns.add_resource(TrialAppWorkflowTaskStopApi, "/trial-apps/<uuid:app_id>/workflows/tasks/<string:task_id>/stop")
 
-api.add_resource(AppWorkflowApi, "/trial-apps/<uuid:app_id>/workflows", endpoint="trial_app_workflow")
-api.add_resource(DatasetListApi, "/trial-apps/<uuid:app_id>/datasets", endpoint="trial_app_datasets")
+console_ns.add_resource(AppWorkflowApi, "/trial-apps/<uuid:app_id>/workflows", endpoint="trial_app_workflow")
+console_ns.add_resource(DatasetListApi, "/trial-apps/<uuid:app_id>/datasets", endpoint="trial_app_datasets")

api/controllers/console/explore/workflow.py

@@ -21,8 +21,9 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
-from core.model_runtime.errors.invoke import InvokeError
-from core.workflow.graph_engine.manager import GraphEngineManager
+from dify_graph.graph_engine.manager import GraphEngineManager
+from dify_graph.model_runtime.errors.invoke import InvokeError
+from extensions.ext_redis import redis_client
 from libs import helper
 from libs.login import current_account_with_tenant
 from models.model import AppMode, InstalledApp
@@ -100,6 +101,6 @@ class InstalledAppWorkflowTaskStopApi(InstalledAppResource):
         AppQueueManager.set_stop_flag_no_user_check(task_id)
 
         # New graph engine command channel mechanism
-        GraphEngineManager.send_stop_command(task_id)
+        GraphEngineManager(redis_client).send_stop_command(task_id)
 
         return {"result": "success"}

api/controllers/console/explore/wraps.py

@@ -105,9 +105,9 @@ def trial_app_required(view: Callable[Concatenate[App, P], R] | None = None):
     return decorator
 
 
-def trial_feature_enable(view: Callable[..., R]) -> Callable[..., R]:
+def trial_feature_enable(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_system_features()
         if not features.enable_trial_app:
             abort(403, "Trial app feature is not enabled.")
@@ -116,9 +116,9 @@ def trial_feature_enable(view: Callable[..., R]) -> Callable[..., R]:
     return decorated
 
 
-def explore_banner_enabled(view: Callable[..., R]) -> Callable[..., R]:
+def explore_banner_enabled(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_system_features()
         if not features.enable_explore_banner:
             abort(403, "Explore banner feature is not enabled.")

api/controllers/console/human_input_form.py

@@ -0,0 +1,217 @@
+"""
+Console/Studio Human Input Form APIs.
+"""
+
+import json
+import logging
+from collections.abc import Generator
+
+from flask import Response, jsonify, request
+from flask_restx import Resource, reqparse
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from controllers.console import console_ns
+from controllers.console.wraps import account_initialization_required, setup_required
+from controllers.web.error import InvalidArgumentError, NotFoundError
+from core.app.apps.advanced_chat.app_generator import AdvancedChatAppGenerator
+from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
+from core.app.apps.message_generator import MessageGenerator
+from core.app.apps.workflow.app_generator import WorkflowAppGenerator
+from extensions.ext_database import db
+from libs.login import current_account_with_tenant, login_required
+from models import App
+from models.enums import CreatorUserRole
+from models.human_input import RecipientType
+from models.model import AppMode
+from models.workflow import WorkflowRun
+from repositories.factory import DifyAPIRepositoryFactory
+from services.human_input_service import Form, HumanInputService
+from services.workflow_event_snapshot_service import build_workflow_event_stream
+
+logger = logging.getLogger(__name__)
+
+
+def _jsonify_form_definition(form: Form) -> Response:
+    payload = form.get_definition().model_dump()
+    payload["expiration_time"] = int(form.expiration_time.timestamp())
+    return Response(json.dumps(payload, ensure_ascii=False), mimetype="application/json")
+
+
+@console_ns.route("/form/human_input/<string:form_token>")
+class ConsoleHumanInputFormApi(Resource):
+    """Console API for getting human input form definition."""
+
+    @staticmethod
+    def _ensure_console_access(form: Form):
+        _, current_tenant_id = current_account_with_tenant()
+
+        if form.tenant_id != current_tenant_id:
+            raise NotFoundError("App not found")
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, form_token: str):
+        """
+        Get human input form definition by form token.
+
+        GET /console/api/form/human_input/<form_token>
+        """
+        service = HumanInputService(db.engine)
+        form = service.get_form_definition_by_token_for_console(form_token)
+        if form is None:
+            raise NotFoundError(f"form not found, token={form_token}")
+
+        self._ensure_console_access(form)
+
+        return _jsonify_form_definition(form)
+
+    @account_initialization_required
+    @login_required
+    def post(self, form_token: str):
+        """
+        Submit human input form by form token.
+
+        POST /console/api/form/human_input/<form_token>
+
+        Request body:
+        {
+            "inputs": {
+                "content": "User input content"
+            },
+            "action": "Approve"
+        }
+        """
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, location="json")
+        parser.add_argument("action", type=str, required=True, location="json")
+        args = parser.parse_args()
+        current_user, _ = current_account_with_tenant()
+
+        service = HumanInputService(db.engine)
+        form = service.get_form_by_token(form_token)
+        if form is None:
+            raise NotFoundError(f"form not found, token={form_token}")
+
+        self._ensure_console_access(form)
+
+        recipient_type = form.recipient_type
+        if recipient_type not in {RecipientType.CONSOLE, RecipientType.BACKSTAGE}:
+            raise NotFoundError(f"form not found, token={form_token}")
+        # The type checker is not smart enought to validate the following invariant.
+        # So we need to assert it manually.
+        assert recipient_type is not None, "recipient_type cannot be None here."
+
+        service.submit_form_by_token(
+            recipient_type=recipient_type,
+            form_token=form_token,
+            selected_action_id=args["action"],
+            form_data=args["inputs"],
+            submission_user_id=current_user.id,
+        )
+
+        return jsonify({})
+
+
+@console_ns.route("/workflow/<string:workflow_run_id>/events")
+class ConsoleWorkflowEventsApi(Resource):
+    """Console API for getting workflow execution events after resume."""
+
+    @account_initialization_required
+    @login_required
+    def get(self, workflow_run_id: str):
+        """
+        Get workflow execution events stream after resume.
+
+        GET /console/api/workflow/<workflow_run_id>/events
+
+        Returns Server-Sent Events stream.
+        """
+
+        user, tenant_id = current_account_with_tenant()
+        session_maker = sessionmaker(db.engine)
+        repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
+        workflow_run = repo.get_workflow_run_by_id_and_tenant_id(
+            tenant_id=tenant_id,
+            run_id=workflow_run_id,
+        )
+        if workflow_run is None:
+            raise NotFoundError(f"WorkflowRun not found, id={workflow_run_id}")
+
+        if workflow_run.created_by_role != CreatorUserRole.ACCOUNT:
+            raise NotFoundError(f"WorkflowRun not created by account, id={workflow_run_id}")
+
+        if workflow_run.created_by != user.id:
+            raise NotFoundError(f"WorkflowRun not created by the current account, id={workflow_run_id}")
+
+        with Session(expire_on_commit=False, bind=db.engine) as session:
+            app = _retrieve_app_for_workflow_run(session, workflow_run)
+
+        if workflow_run.finished_at is not None:
+            # TODO(QuantumGhost): should we modify the handling for finished workflow run here?
+            response = WorkflowResponseConverter.workflow_run_result_to_finish_response(
+                task_id=workflow_run.id,
+                workflow_run=workflow_run,
+                creator_user=user,
+            )
+
+            payload = response.model_dump(mode="json")
+            payload["event"] = response.event.value
+
+            def _generate_finished_events() -> Generator[str, None, None]:
+                yield f"data: {json.dumps(payload)}\n\n"
+
+            event_generator = _generate_finished_events
+
+        else:
+            msg_generator = MessageGenerator()
+            if app.mode == AppMode.ADVANCED_CHAT:
+                generator = AdvancedChatAppGenerator()
+            elif app.mode == AppMode.WORKFLOW:
+                generator = WorkflowAppGenerator()
+            else:
+                raise InvalidArgumentError(f"cannot subscribe to workflow run, workflow_run_id={workflow_run.id}")
+
+            include_state_snapshot = request.args.get("include_state_snapshot", "false").lower() == "true"
+
+            def _generate_stream_events():
+                if include_state_snapshot:
+                    return generator.convert_to_event_stream(
+                        build_workflow_event_stream(
+                            app_mode=AppMode(app.mode),
+                            workflow_run=workflow_run,
+                            tenant_id=workflow_run.tenant_id,
+                            app_id=workflow_run.app_id,
+                            session_maker=session_maker,
+                        )
+                    )
+                return generator.convert_to_event_stream(
+                    msg_generator.retrieve_events(AppMode(app.mode), workflow_run.id),
+                )
+
+            event_generator = _generate_stream_events
+
+        return Response(
+            event_generator(),
+            mimetype="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+            },
+        )
+
+
+def _retrieve_app_for_workflow_run(session: Session, workflow_run: WorkflowRun):
+    query = select(App).where(
+        App.id == workflow_run.app_id,
+        App.tenant_id == workflow_run.tenant_id,
+    )
+    app = session.scalars(query).first()
+    if app is None:
+        raise AssertionError(
+            f"App not found for WorkflowRun, workflow_run_id={workflow_run.id}, "
+            f"app_id={workflow_run.app_id}, tenant_id={workflow_run.tenant_id}"
+        )
+
+    return app

api/controllers/console/init_validate.py

@@ -1,87 +1,74 @@
 import os
+from typing import Literal
 
 from flask import session
-from flask_restx import Resource, fields
 from pydantic import BaseModel, Field
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from configs import dify_config
+from controllers.fastopenapi import console_router
 from extensions.ext_database import db
 from models.model import DifySetup
 from services.account_service import TenantService
 
-from . import console_ns
 from .error import AlreadySetupError, InitValidateFailedError
 from .wraps import only_edition_self_hosted
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class InitValidatePayload(BaseModel):
-    password: str = Field(..., max_length=30)
+    password: str = Field(..., max_length=30, description="Initialization password")
 
 
-console_ns.schema_model(
-    InitValidatePayload.__name__,
-    InitValidatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+class InitStatusResponse(BaseModel):
+    status: Literal["finished", "not_started"] = Field(..., description="Initialization status")
+
+
+class InitValidateResponse(BaseModel):
+    result: str = Field(description="Operation result", examples=["success"])
+
+
+@console_router.get(
+    "/init",
+    response_model=InitStatusResponse,
+    tags=["console"],
 )
+def get_init_status() -> InitStatusResponse:
+    """Get initialization validation status."""
+    init_status = get_init_validate_status()
+    if init_status:
+        return InitStatusResponse(status="finished")
+    return InitStatusResponse(status="not_started")
 
 
-@console_ns.route("/init")
-class InitValidateAPI(Resource):
-    @console_ns.doc("get_init_status")
-    @console_ns.doc(description="Get initialization validation status")
-    @console_ns.response(
-        200,
-        "Success",
-        model=console_ns.model(
-            "InitStatusResponse",
-            {"status": fields.String(description="Initialization status", enum=["finished", "not_started"])},
-        ),
-    )
-    def get(self):
-        """Get initialization validation status"""
-        init_status = get_init_validate_status()
-        if init_status:
-            return {"status": "finished"}
-        return {"status": "not_started"}
+@console_router.post(
+    "/init",
+    response_model=InitValidateResponse,
+    tags=["console"],
+    status_code=201,
+)
+@only_edition_self_hosted
+def validate_init_password(payload: InitValidatePayload) -> InitValidateResponse:
+    """Validate initialization password."""
+    tenant_count = TenantService.get_tenant_count()
+    if tenant_count > 0:
+        raise AlreadySetupError()
 
-    @console_ns.doc("validate_init_password")
-    @console_ns.doc(description="Validate initialization password for self-hosted edition")
-    @console_ns.expect(console_ns.models[InitValidatePayload.__name__])
-    @console_ns.response(
-        201,
-        "Success",
-        model=console_ns.model("InitValidateResponse", {"result": fields.String(description="Operation result")}),
-    )
-    @console_ns.response(400, "Already setup or validation failed")
-    @only_edition_self_hosted
-    def post(self):
-        """Validate initialization password"""
-        # is tenant created
-        tenant_count = TenantService.get_tenant_count()
-        if tenant_count > 0:
-            raise AlreadySetupError()
+    if payload.password != os.environ.get("INIT_PASSWORD"):
+        session["is_init_validated"] = False
+        raise InitValidateFailedError()
 
-        payload = InitValidatePayload.model_validate(console_ns.payload)
-        input_password = payload.password
-
-        if input_password != os.environ.get("INIT_PASSWORD"):
-            session["is_init_validated"] = False
-            raise InitValidateFailedError()
-
-        session["is_init_validated"] = True
-        return {"result": "success"}, 201
+    session["is_init_validated"] = True
+    return InitValidateResponse(result="success")
 
 
-def get_init_validate_status():
+def get_init_validate_status() -> bool:
     if dify_config.EDITION == "SELF_HOSTED":
         if os.environ.get("INIT_PASSWORD"):
             if session.get("is_init_validated"):
                 return True
 
             with Session(db.engine) as db_session:
-                return db_session.execute(select(DifySetup)).scalar_one_or_none()
+                return db_session.execute(select(DifySetup)).scalar_one_or_none() is not None
 
     return True

api/controllers/console/remote_files.py

@@ -11,68 +11,59 @@ from controllers.common.errors import (
     RemoteFileUploadError,
     UnsupportedFileTypeError,
 )
-from controllers.common.schema import register_schema_models
-from core.file import helpers as file_helpers
+from controllers.console import console_ns
 from core.helper import ssrf_proxy
+from dify_graph.file import helpers as file_helpers
 from extensions.ext_database import db
 from fields.file_fields import FileWithSignedUrl, RemoteFileInfo
-from libs.login import current_account_with_tenant
+from libs.login import current_account_with_tenant, login_required
 from services.file_service import FileService
 
-from . import console_ns
-
-register_schema_models(console_ns, RemoteFileInfo, FileWithSignedUrl)
-
-
-@console_ns.route("/remote-files/<path:url>")
-class RemoteFileInfoApi(Resource):
-    @console_ns.response(200, "Remote file info", console_ns.models[RemoteFileInfo.__name__])
-    def get(self, url):
-        decoded_url = urllib.parse.unquote(url)
-        resp = ssrf_proxy.head(decoded_url)
-        if resp.status_code != httpx.codes.OK:
-            # failed back to get method
-            resp = ssrf_proxy.get(decoded_url, timeout=3)
-        resp.raise_for_status()
-        info = RemoteFileInfo(
-            file_type=resp.headers.get("Content-Type", "application/octet-stream"),
-            file_length=int(resp.headers.get("Content-Length", 0)),
-        )
-        return info.model_dump(mode="json")
-
 
 class RemoteFileUploadPayload(BaseModel):
     url: str = Field(..., description="URL to fetch")
 
 
-console_ns.schema_model(
-    RemoteFileUploadPayload.__name__,
-    RemoteFileUploadPayload.model_json_schema(ref_template="#/definitions/{model}"),
-)
+@console_ns.route("/remote-files/<path:url>")
+class GetRemoteFileInfo(Resource):
+    @login_required
+    def get(self, url: str):
+        decoded_url = urllib.parse.unquote(url)
+        resp = ssrf_proxy.head(decoded_url)
+        if resp.status_code != httpx.codes.OK:
+            resp = ssrf_proxy.get(decoded_url, timeout=3)
+        resp.raise_for_status()
+        return RemoteFileInfo(
+            file_type=resp.headers.get("Content-Type", "application/octet-stream"),
+            file_length=int(resp.headers.get("Content-Length", 0)),
+        ).model_dump(mode="json")
 
 
 @console_ns.route("/remote-files/upload")
-class RemoteFileUploadApi(Resource):
-    @console_ns.expect(console_ns.models[RemoteFileUploadPayload.__name__])
-    @console_ns.response(201, "Remote file uploaded", console_ns.models[FileWithSignedUrl.__name__])
+class RemoteFileUpload(Resource):
+    @login_required
     def post(self):
-        args = RemoteFileUploadPayload.model_validate(console_ns.payload)
-        url = args.url
+        payload = RemoteFileUploadPayload.model_validate(console_ns.payload)
+        url = payload.url
 
+        # Try to fetch remote file metadata/content first
         try:
             resp = ssrf_proxy.head(url=url)
             if resp.status_code != httpx.codes.OK:
                 resp = ssrf_proxy.get(url=url, timeout=3, follow_redirects=True)
             if resp.status_code != httpx.codes.OK:
+                # Normalize into a user-friendly error message expected by tests
                 raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}")
         except httpx.RequestError as e:
             raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}")
 
         file_info = helpers.guess_file_info_from_response(resp)
 
+        # Enforce file size limit with 400 (Bad Request) per tests' expectation
         if not FileService.is_file_size_within_limit(extension=file_info.extension, file_size=file_info.size):
-            raise FileTooLargeError
+            raise FileTooLargeError()
 
+        # Load content if needed
         content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content
 
         try:
@@ -89,14 +80,17 @@ class RemoteFileUploadApi(Resource):
         except services.errors.file.UnsupportedFileTypeError:
             raise UnsupportedFileTypeError()
 
-        payload = FileWithSignedUrl(
-            id=upload_file.id,
-            name=upload_file.name,
-            size=upload_file.size,
-            extension=upload_file.extension,
-            url=file_helpers.get_signed_file_url(upload_file_id=upload_file.id),
-            mime_type=upload_file.mime_type,
-            created_by=upload_file.created_by,
-            created_at=int(upload_file.created_at.timestamp()),
+        # Success: return created resource with 201 status
+        return (
+            FileWithSignedUrl(
+                id=upload_file.id,
+                name=upload_file.name,
+                size=upload_file.size,
+                extension=upload_file.extension,
+                url=file_helpers.get_signed_file_url(upload_file_id=upload_file.id),
+                mime_type=upload_file.mime_type,
+                created_by=upload_file.created_by,
+                created_at=int(upload_file.created_at.timestamp()),
+            ).model_dump(mode="json"),
+            201,
         )
-        return payload.model_dump(mode="json"), 201

api/controllers/console/setup.py

@@ -42,7 +42,15 @@ class SetupResponse(BaseModel):
     tags=["console"],
 )
 def get_setup_status_api() -> SetupStatusResponse:
-    """Get system setup status."""
+    """Get system setup status.
+
+    NOTE: This endpoint is unauthenticated by design.
+
+    During first-time bootstrap there is no admin account yet, so frontend initialization must be
+    able to query setup progress before any login flow exists.
+
+    Only bootstrap-safe status information should be returned by this endpoint.
+    """
     if dify_config.EDITION == "SELF_HOSTED":
         setup_status = get_setup_status()
         if setup_status and not isinstance(setup_status, bool):
@@ -61,7 +69,12 @@ def get_setup_status_api() -> SetupStatusResponse:
 )
 @only_edition_self_hosted
 def setup_system(payload: SetupRequestPayload) -> SetupResponse:
-    """Initialize system setup with admin account."""
+    """Initialize system setup with admin account.
+
+    NOTE: This endpoint is unauthenticated by design for first-time bootstrap.
+    Access is restricted by deployment mode (`SELF_HOSTED`), one-time setup guards,
+    and init-password validation rather than user session authentication.
+    """
     if get_setup_status():
         raise AlreadySetupError()
 

api/controllers/console/tag/tags.py

@@ -1,17 +1,27 @@
 from typing import Literal
 
 from flask import request
-from flask_restx import Resource, marshal_with
+from flask_restx import Namespace, Resource, fields, marshal_with
 from pydantic import BaseModel, Field
 from werkzeug.exceptions import Forbidden
 
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
-from fields.tag_fields import dataset_tag_fields
 from libs.login import current_account_with_tenant, login_required
 from services.tag_service import TagService
 
+dataset_tag_fields = {
+    "id": fields.String,
+    "name": fields.String,
+    "type": fields.String,
+    "binding_count": fields.String,
+}
+
+
+def build_dataset_tag_fields(api_or_ns: Namespace):
+    return api_or_ns.model("DataSetTag", dataset_tag_fields)
+
 
 class TagBasePayload(BaseModel):
     name: str = Field(description="Tag name", min_length=1, max_length=50)
@@ -110,7 +120,7 @@ class TagUpdateDeleteApi(Resource):
 
         TagService.delete_tag(tag_id)
 
-        return 204
+        return "", 204
 
 
 @console_ns.route("/tag-bindings/create")

api/controllers/console/workspace/account.py

@@ -12,6 +12,7 @@ from sqlalchemy.orm import Session
 
 from configs import dify_config
 from constants.languages import supported_language
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
     EmailAlreadyInUseError,
@@ -37,7 +38,7 @@ from controllers.console.wraps import (
     setup_required,
 )
 from extensions.ext_database import db
-from fields.member_fields import account_fields
+from fields.member_fields import Account as AccountResponse
 from libs.datetime_utils import naive_utc_now
 from libs.helper import EmailStr, TimestampField, extract_remote_ip, timezone
 from libs.login import current_account_with_tenant, login_required
@@ -170,6 +171,12 @@ reg(ChangeEmailSendPayload)
 reg(ChangeEmailValidityPayload)
 reg(ChangeEmailResetPayload)
 reg(CheckEmailUniquePayload)
+register_schema_models(console_ns, AccountResponse)
+
+
+def _serialize_account(account) -> dict:
+    return AccountResponse.model_validate(account, from_attributes=True).model_dump(mode="json")
+
 
 integrate_fields = {
     "provider": fields.String,
@@ -236,11 +243,11 @@ class AccountProfileApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     @enterprise_license_required
     def get(self):
         current_user, _ = current_account_with_tenant()
-        return current_user
+        return _serialize_account(current_user)
 
 
 @console_ns.route("/account/name")
@@ -249,14 +256,14 @@ class AccountNameApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
         args = AccountNamePayload.model_validate(payload)
         updated_account = AccountService.update_account(current_user, name=args.name)
 
-        return updated_account
+        return _serialize_account(updated_account)
 
 
 @console_ns.route("/account/avatar")
@@ -265,7 +272,7 @@ class AccountAvatarApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -273,7 +280,7 @@ class AccountAvatarApi(Resource):
 
         updated_account = AccountService.update_account(current_user, avatar=args.avatar)
 
-        return updated_account
+        return _serialize_account(updated_account)
 
 
 @console_ns.route("/account/interface-language")
@@ -282,7 +289,7 @@ class AccountInterfaceLanguageApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -290,7 +297,7 @@ class AccountInterfaceLanguageApi(Resource):
 
         updated_account = AccountService.update_account(current_user, interface_language=args.interface_language)
 
-        return updated_account
+        return _serialize_account(updated_account)
 
 
 @console_ns.route("/account/interface-theme")
@@ -299,7 +306,7 @@ class AccountInterfaceThemeApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -307,7 +314,7 @@ class AccountInterfaceThemeApi(Resource):
 
         updated_account = AccountService.update_account(current_user, interface_theme=args.interface_theme)
 
-        return updated_account
+        return _serialize_account(updated_account)
 
 
 @console_ns.route("/account/timezone")
@@ -316,7 +323,7 @@ class AccountTimezoneApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -324,7 +331,7 @@ class AccountTimezoneApi(Resource):
 
         updated_account = AccountService.update_account(current_user, timezone=args.timezone)
 
-        return updated_account
+        return _serialize_account(updated_account)
 
 
 @console_ns.route("/account/password")
@@ -333,7 +340,7 @@ class AccountPasswordApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -344,7 +351,7 @@ class AccountPasswordApi(Resource):
         except ServiceCurrentPasswordIncorrectError:
             raise CurrentPasswordIncorrectError()
 
-        return {"result": "success"}
+        return _serialize_account(current_user)
 
 
 @console_ns.route("/account/integrates")
@@ -620,7 +627,7 @@ class ChangeEmailResetApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     def post(self):
         payload = console_ns.payload or {}
         args = ChangeEmailResetPayload.model_validate(payload)
@@ -649,7 +656,7 @@ class ChangeEmailResetApi(Resource):
             email=normalized_new_email,
         )
 
-        return updated_account
+        return _serialize_account(updated_account)
 
 
 @console_ns.route("/account/change-email/check-email-unique")

api/controllers/console/workspace/agent_providers.py

@@ -2,7 +2,7 @@ from flask_restx import Resource, fields
 
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
-from core.model_runtime.utils.encoders import jsonable_encoder
+from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from services.agent_service import AgentService
 

api/controllers/console/workspace/endpoint.py

@@ -1,13 +1,14 @@
 from typing import Any
 
 from flask import request
-from flask_restx import Resource, fields
+from flask_restx import Resource
 from pydantic import BaseModel, Field
 
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
-from core.model_runtime.utils.encoders import jsonable_encoder
 from core.plugin.impl.exc import PluginPermissionDeniedError
+from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from services.plugin.endpoint_service import EndpointService
 
@@ -38,15 +39,53 @@ class EndpointListForPluginQuery(EndpointListQuery):
     plugin_id: str
 
 
+class EndpointCreateResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
+class EndpointListResponse(BaseModel):
+    endpoints: list[dict[str, Any]] = Field(description="Endpoint information")
+
+
+class PluginEndpointListResponse(BaseModel):
+    endpoints: list[dict[str, Any]] = Field(description="Endpoint information")
+
+
+class EndpointDeleteResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
+class EndpointUpdateResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
+class EndpointEnableResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
+class EndpointDisableResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
 def reg(cls: type[BaseModel]):
     console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
 
 
-reg(EndpointCreatePayload)
-reg(EndpointIdPayload)
-reg(EndpointUpdatePayload)
-reg(EndpointListQuery)
-reg(EndpointListForPluginQuery)
+register_schema_models(
+    console_ns,
+    EndpointCreatePayload,
+    EndpointIdPayload,
+    EndpointUpdatePayload,
+    EndpointListQuery,
+    EndpointListForPluginQuery,
+    EndpointCreateResponse,
+    EndpointListResponse,
+    PluginEndpointListResponse,
+    EndpointDeleteResponse,
+    EndpointUpdateResponse,
+    EndpointEnableResponse,
+    EndpointDisableResponse,
+)
 
 
 @console_ns.route("/workspaces/current/endpoints/create")
@@ -57,7 +96,7 @@ class EndpointCreateApi(Resource):
     @console_ns.response(
         200,
         "Endpoint created successfully",
-        console_ns.model("EndpointCreateResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointCreateResponse.__name__],
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -91,9 +130,7 @@ class EndpointListApi(Resource):
     @console_ns.response(
         200,
         "Success",
-        console_ns.model(
-            "EndpointListResponse", {"endpoints": fields.List(fields.Raw(description="Endpoint information"))}
-        ),
+        console_ns.models[EndpointListResponse.__name__],
     )
     @setup_required
     @login_required
@@ -126,9 +163,7 @@ class EndpointListForSinglePluginApi(Resource):
     @console_ns.response(
         200,
         "Success",
-        console_ns.model(
-            "PluginEndpointListResponse", {"endpoints": fields.List(fields.Raw(description="Endpoint information"))}
-        ),
+        console_ns.models[PluginEndpointListResponse.__name__],
     )
     @setup_required
     @login_required
@@ -163,7 +198,7 @@ class EndpointDeleteApi(Resource):
     @console_ns.response(
         200,
         "Endpoint deleted successfully",
-        console_ns.model("EndpointDeleteResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointDeleteResponse.__name__],
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -190,7 +225,7 @@ class EndpointUpdateApi(Resource):
     @console_ns.response(
         200,
         "Endpoint updated successfully",
-        console_ns.model("EndpointUpdateResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointUpdateResponse.__name__],
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -221,7 +256,7 @@ class EndpointEnableApi(Resource):
     @console_ns.response(
         200,
         "Endpoint enabled successfully",
-        console_ns.model("EndpointEnableResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointEnableResponse.__name__],
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -248,7 +283,7 @@ class EndpointDisableApi(Resource):
     @console_ns.response(
         200,
         "Endpoint disabled successfully",
-        console_ns.model("EndpointDisableResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointDisableResponse.__name__],
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required

api/controllers/console/workspace/load_balancing_config.py

@@ -5,8 +5,8 @@ from werkzeug.exceptions import Forbidden
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
-from core.model_runtime.entities.model_entities import ModelType
-from core.model_runtime.errors.validate import CredentialsValidateFailedError
+from dify_graph.model_runtime.entities.model_entities import ModelType
+from dify_graph.model_runtime.errors.validate import CredentialsValidateFailedError
 from libs.login import current_account_with_tenant, login_required
 from models import TenantAccountRole
 from services.model_load_balancing_service import ModelLoadBalancingService

api/controllers/console/workspace/members.py

@@ -1,12 +1,12 @@
 from urllib import parse
 
 from flask import abort, request
-from flask_restx import Resource, fields, marshal_with
-from pydantic import BaseModel, Field
+from flask_restx import Resource
+from pydantic import BaseModel, Field, TypeAdapter
 
 import services
 from configs import dify_config
-from controllers.common.schema import get_or_create_model, register_enum_models
+from controllers.common.schema import register_enum_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
     CannotTransferOwnerToSelfError,
@@ -25,7 +25,7 @@ from controllers.console.wraps import (
     setup_required,
 )
 from extensions.ext_database import db
-from fields.member_fields import account_with_role_fields, account_with_role_list_fields
+from fields.member_fields import AccountWithRole, AccountWithRoleList
 from libs.helper import extract_remote_ip
 from libs.login import current_account_with_tenant, login_required
 from models.account import Account, TenantAccountRole
@@ -69,12 +69,7 @@ reg(OwnerTransferEmailPayload)
 reg(OwnerTransferCheckPayload)
 reg(OwnerTransferPayload)
 register_enum_models(console_ns, TenantAccountRole)
-
-account_with_role_model = get_or_create_model("AccountWithRole", account_with_role_fields)
-
-account_with_role_list_fields_copy = account_with_role_list_fields.copy()
-account_with_role_list_fields_copy["accounts"] = fields.List(fields.Nested(account_with_role_model))
-account_with_role_list_model = get_or_create_model("AccountWithRoleList", account_with_role_list_fields_copy)
+register_schema_models(console_ns, AccountWithRole, AccountWithRoleList)
 
 
 @console_ns.route("/workspaces/current/members")
@@ -84,13 +79,15 @@ class MemberListApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_with_role_list_model)
+    @console_ns.response(200, "Success", console_ns.models[AccountWithRoleList.__name__])
     def get(self):
         current_user, _ = current_account_with_tenant()
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
         members = TenantService.get_tenant_members(current_user.current_tenant)
-        return {"result": "success", "accounts": members}, 200
+        member_models = TypeAdapter(list[AccountWithRole]).validate_python(members, from_attributes=True)
+        response = AccountWithRoleList(accounts=member_models)
+        return response.model_dump(mode="json"), 200
 
 
 @console_ns.route("/workspaces/current/members/invite-email")
@@ -235,13 +232,15 @@ class DatasetOperatorMemberListApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_with_role_list_model)
+    @console_ns.response(200, "Success", console_ns.models[AccountWithRoleList.__name__])
     def get(self):
         current_user, _ = current_account_with_tenant()
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
         members = TenantService.get_dataset_operator_members(current_user.current_tenant)
-        return {"result": "success", "accounts": members}, 200
+        member_models = TypeAdapter(list[AccountWithRole]).validate_python(members, from_attributes=True)
+        response = AccountWithRoleList(accounts=member_models)
+        return response.model_dump(mode="json"), 200
 
 
 @console_ns.route("/workspaces/current/members/send-owner-transfer-confirm-email")