refactor(api): move TokenBufferMemory to model_runtime

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[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()
+    ```

.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

.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)
@@ -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,25 @@
 version: 2
+
+multi-ecosystem-groups:
+  python:
+    schedule:
+      interval: "weekly"  # or whatever schedule you want
+
 updates:
+  - package-ecosystem: "pip"
+    directory: "/api"
+    open-pull-requests-limit: 2
+    patterns: ["*"]
+    schedule:
+      interval: "weekly"
+  - package-ecosystem: "uv"
+    directory: "/api"
+    open-pull-requests-limit: 2
+    patterns: ["*"]
+    schedule:
+      interval: "weekly"
   - 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

.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:
       - "*"
 
@@ -75,9 +76,7 @@ jobs:
         with:
           context: "{{defaultContext}}:${{ matrix.context }}"
           platforms: ${{ matrix.platform }}
-          build-args: |
-            COMMIT_SHA=${{ fromJSON(steps.meta.outputs.json).labels['org.opencontainers.image.revision'] }}
-            ENABLE_PROD_SOURCEMAP=${{ matrix.context == 'web' && github.ref_name == 'deploy/dev' }}
+          build-args: COMMIT_SHA=${{ fromJSON(steps.meta.outputs.json).labels['org.opencontainers.image.revision'] }}
           labels: ${{ steps.meta.outputs.labels }}
           outputs: type=image,name=${{ env[matrix.image_name_env] }},push-by-digest=true,name-canonical=true,push=true
           cache-from: type=gha,scope=${{ matrix.service_name }}

.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,94 @@
+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: Run pyrefly on PR branch
+        run: |
+          uv run --directory api pyrefly check > /tmp/pyrefly_pr.txt 2>&1 || true
+
+      - name: Checkout base branch
+        run: git checkout ${{ github.base_ref }}
+
+      - name: Run pyrefly on base branch
+        run: |
+          uv run --directory api pyrefly check > /tmp/pyrefly_base.txt 2>&1 || true
+
+      - name: Compute diff
+        run: |
+          diff /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/web-tests.yml

@@ -3,14 +3,22 @@ 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
@@ -39,7 +47,58 @@ jobs:
         run: pnpm install --frozen-lockfile
 
       - name: Run tests
-        run: pnpm test:coverage
+        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:
+        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: 24
+          cache: pnpm
+          cache-dependency-path: ./web/pnpm-lock.yaml
+
+      - name: Install dependencies
+        run: pnpm install --frozen-lockfile
+
+      - 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()

.gitignore

@@ -209,7 +209,6 @@ api/.vscode
 .history
 
 .idea/
-web/migration/
 
 # pnpm
 /.pnpm-store

.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,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"
             ],

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

@@ -33,9 +33,6 @@ TRIGGER_URL=http://localhost:5001
 # The time in seconds after the signature is rejected
 FILES_ACCESS_TIMEOUT=300
 
-# Collaboration mode toggle
-ENABLE_COLLABORATION_MODE=false
-
 # Access token expiration time in minutes
 ACCESS_TOKEN_EXPIRE_MINUTES=60
 
@@ -556,6 +553,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
@@ -718,16 +717,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
 
-# Sandbox Dify CLI configuration
-# Directory containing dify CLI binaries (dify-cli-<os>-<arch>). Defaults to api/bin when unset.
-SANDBOX_DIFY_CLI_ROOT=
 
-# CLI API URL for sandbox (dify-sandbox or e2b) to call back to Dify API.
-# This URL must be accessible from the sandbox environment.
-# For local development: use http://localhost:5001 or http://127.0.0.1:5001
-# For Docker deployment: use http://api:5001 (internal Docker network)
-# For external sandbox (e.g., e2b): use a publicly accessible URL
-CLI_API_URL=http://localhost:5001
+# 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

@@ -36,6 +36,8 @@ ignore_imports =
     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
+    # TODO(QuantumGhost): fix the import violation later
+    core.workflow.entities.pause_reason -> core.workflow.nodes.human_input.entities
 
 [importlinter:contract:workflow-infrastructure-dependencies]
 name = Workflow Infrastructure Dependencies
@@ -48,16 +50,13 @@ forbidden_modules =
 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
+    # TODO(QuantumGhost): use DI to avoid depending on global DB.
+    core.workflow.nodes.human_input.human_input_node -> extensions.ext_database
 
 [importlinter:contract:workflow-external-imports]
 name = Workflow External Imports
@@ -89,7 +88,6 @@ forbidden_modules =
     core.logging
     core.mcp
     core.memory
-    core.model_manager
     core.moderation
     core.ops
     core.plugin
@@ -103,40 +101,23 @@ forbidden_modules =
     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.protocols -> 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
@@ -145,9 +126,9 @@ ignore_imports =
     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.human_input.human_input_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
@@ -161,50 +142,18 @@ ignore_imports =
     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
@@ -214,7 +163,6 @@ ignore_imports =
     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
@@ -230,7 +178,6 @@ ignore_imports =
     core.workflow.nodes.knowledge_index.knowledge_index_node -> services.summary_index_service
     core.workflow.nodes.knowledge_index.knowledge_index_node -> tasks.generate_summary_index_task
     core.workflow.nodes.knowledge_index.knowledge_index_node -> core.rag.index_processor.processor.paragraph_index_processor
-    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
@@ -243,11 +190,11 @@ ignore_imports =
     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.human_input.entities -> core.variables.consts
     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
@@ -284,16 +231,14 @@ ignore_imports =
     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.nodes.human_input.human_input_node -> extensions.ext_database
+    core.workflow.nodes.human_input.human_input_node -> core.repositories.human_input_repository
     core.workflow.workflow_entry -> extensions.otel.runtime
     core.workflow.nodes.agent.agent_node -> models
     core.workflow.nodes.base.node -> models.enums
@@ -302,6 +247,11 @@ ignore_imports =
     core.workflow.workflow_entry -> models.enums
     core.workflow.nodes.agent.agent_node -> services
     core.workflow.nodes.tool.tool_node -> services
+    core.workflow.nodes.agent.agent_node -> core.model_runtime.token_buffer_memory
+    core.workflow.nodes.llm.llm_utils -> core.model_runtime.token_buffer_memory
+    core.workflow.nodes.llm.node -> core.model_runtime.token_buffer_memory
+    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_runtime.token_buffer_memory
+    core.workflow.nodes.question_classifier.question_classifier_node -> core.model_runtime.token_buffer_memory
 
 [importlinter:contract:model-runtime-no-internal-imports]
 name = Model Runtime Internal Imports
@@ -354,6 +304,13 @@ ignore_imports =
     core.model_runtime.model_providers.model_provider_factory -> configs
     core.model_runtime.model_providers.model_provider_factory -> extensions.ext_redis
     core.model_runtime.model_providers.model_provider_factory -> models.provider_ids
+    core.model_runtime.token_buffer_memory -> core.app.app_config.features.file_upload.manager
+    core.model_runtime.token_buffer_memory -> core.model_manager
+    core.model_runtime.token_buffer_memory -> core.prompt.utils.extract_thread_messages
+    core.model_runtime.token_buffer_memory -> core.workflow.file.file_manager
+    core.model_runtime.token_buffer_memory -> extensions.ext_database
+    core.model_runtime.token_buffer_memory -> models.model
+    core.model_runtime.token_buffer_memory -> models.workflow
 
 [importlinter:contract:rsc]
 name = RSC

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]
@@ -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/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/agent-notes/configs/feature/__init__.py.md

@@ -1,9 +0,0 @@
-Summary:
-Summary:
-- Application configuration definitions, including file access settings.
-
-Invariants:
-- File access settings drive signed URL expiration and base URLs.
-
-Tests:
-- Config parsing tests under tests/unit_tests/configs.

api/agent-notes/controllers/files/__init__.py.md

@@ -1,9 +0,0 @@
-Summary:
-- Registers file-related API namespaces and routes for files service.
-- Includes app-assets and sandbox archive proxy controllers.
-
-Invariants:
-- files_ns must include all file controller modules to register routes.
-
-Tests:
-- Coverage via controller unit tests and route registration smoke checks.

api/agent-notes/controllers/files/app_assets_download.py.md

@@ -1,14 +0,0 @@
-Summary:
-- App assets download proxy endpoint (signed URL verification, stream from storage).
-
-Invariants:
-- Validates AssetPath fields (UUIDs, asset_type allowlist).
-- Verifies tenant-scoped signature and expiration before reading storage.
-- URL uses expires_at/nonce/sign query params.
-
-Edge Cases:
-- Missing files return NotFound.
-- Invalid signature or expired link returns Forbidden.
-
-Tests:
-- Verify signature validation and invalid/expired cases.

api/agent-notes/controllers/files/app_assets_upload.py.md

@@ -1,13 +0,0 @@
-Summary:
-- App assets upload proxy endpoint (signed URL verification, upload to storage).
-
-Invariants:
-- Validates AssetPath fields (UUIDs, asset_type allowlist).
-- Verifies tenant-scoped signature and expiration before writing storage.
-- URL uses expires_at/nonce/sign query params.
-
-Edge Cases:
-- Invalid signature or expired link returns Forbidden.
-
-Tests:
-- Verify signature validation and invalid/expired cases.

api/agent-notes/controllers/files/sandbox_archive.py.md

@@ -1,14 +0,0 @@
-Summary:
-- Sandbox archive upload/download proxy endpoints (signed URL verification, stream to storage).
-
-Invariants:
-- Validates tenant_id and sandbox_id UUIDs.
-- Verifies tenant-scoped signature and expiration before storage access.
-- URL uses expires_at/nonce/sign query params.
-
-Edge Cases:
-- Missing archive returns NotFound.
-- Invalid signature or expired link returns Forbidden.
-
-Tests:
-- Add unit tests for signature validation if needed.

api/agent-notes/core/app_assets/builder/file_builder.py.md

@@ -1,9 +0,0 @@
-Summary:
-Summary:
-- Collects file assets and emits FileAsset entries with storage keys.
-
-Invariants:
-- Storage keys are derived via AppAssetStorage for draft files.
-
-Tests:
-- Covered by asset build pipeline tests.

api/agent-notes/core/app_assets/builder/skill_builder.py.md

@@ -1,14 +0,0 @@
-Summary:
-Summary:
-- Builds skill artifacts from markdown assets and uploads resolved outputs.
-
-Invariants:
-- Reads draft asset content via AppAssetStorage refs.
-- Writes resolved artifacts via AppAssetStorage refs.
-- FileAsset storage keys are derived via AppAssetStorage.
-
-Edge Cases:
-- Missing or invalid JSON content yields empty skill content/metadata.
-
-Tests:
-- Build pipeline unit tests covering compile/upload paths.

api/agent-notes/core/app_assets/converters.py.md

@@ -1,9 +0,0 @@
-Summary:
-Summary:
-- Converts AppAssetFileTree to FileAsset items for packaging.
-
-Invariants:
-- Storage keys for assets are derived via AppAssetStorage.
-
-Tests:
-- Used in packaging/service tests for asset bundles.

api/agent-notes/core/app_assets/packager/zip_packager.py.md

@@ -1,14 +0,0 @@
-# Zip Packager Notes
-
-## Purpose
-- Builds a ZIP archive of asset contents stored via the configured storage backend.
-
-## Key Decisions
-- Packaging writes assets into an in-memory zip buffer returned as bytes.
-- Asset fetch + zip writing are executed via a thread pool with a lock guarding `ZipFile` writes.
-
-## Edge Cases
-- ZIP writes are serialized by the lock; storage reads still run in parallel.
-
-## Tests/Verification
-- None yet.

api/agent-notes/core/app_assets/parser/asset_parser.py.md

@@ -1,9 +0,0 @@
-Summary:
-Summary:
-- Builds AssetItem entries for asset trees using AssetPath-derived storage keys.
-
-Invariants:
-- Uses AssetPath to compute draft storage keys.
-
-Tests:
-- Covered by asset parsing and packaging tests.

api/agent-notes/core/app_assets/storage.py.md

@@ -1,20 +0,0 @@
-Summary:
-- Defines AssetPath facade + typed asset path classes for app-asset storage access.
-- Maps asset paths to storage keys and generates presigned or signed-proxy URLs.
-- Signs proxy URLs using tenant private keys and enforces expiration.
-- Exposes app_asset_storage singleton for reuse.
-
-Invariants:
-- AssetPathBase fields (tenant_id/app_id/resource_id/node_id) must be UUIDs.
-- AssetPath.from_components enforces valid types and resolved node_id presence.
-- Storage keys are derived internally via AssetPathBase.get_storage_key; callers never supply raw paths.
-- AppAssetStorage.storage returns the cached presign wrapper (not the raw storage).
-
-Edge Cases:
-- Storage backends without presign support must fall back to signed proxy URLs.
-- Signed proxy verification enforces expiration and tenant-scoped signing keys.
-- Upload URLs also fall back to signed proxy endpoints when presign is unsupported.
-- load_or_none treats SilentStorage "File Not Found" bytes as missing.
-
-Tests:
-- Unit tests for ref validation, storage key mapping, and signed URL verification.

api/agent-notes/core/app_bundle/source_zip_extractor.py.md

@@ -1,10 +0,0 @@
-Summary:
-Summary:
-- Extracts asset files from a zip and persists them into app asset storage.
-
-Invariants:
-- Rejects path traversal/absolute/backslash paths.
-- Saves extracted files via AppAssetStorage draft refs.
-
-Tests:
-- Zip security edge cases and tree construction tests.

api/agent-notes/core/sandbox/initializer/app_assets_initializer.py.md

@@ -1,9 +0,0 @@
-Summary:
-Summary:
-- Downloads published app asset zip into sandbox and extracts it.
-
-Invariants:
-- Uses AppAssetStorage to generate download URLs for build zips (internal URL).
-
-Tests:
-- Sandbox initialization integration tests.

api/agent-notes/core/sandbox/initializer/draft_app_assets_initializer.py.md

@@ -1,12 +0,0 @@
-Summary:
-Summary:
-- Downloads draft/resolved assets into sandbox for draft execution.
-
-Invariants:
-- Uses AppAssetStorage to generate download URLs for draft/resolved refs (internal URL).
-
-Edge Cases:
-- No nodes -> returns early.
-
-Tests:
-- Sandbox draft initialization tests.

api/agent-notes/core/sandbox/sandbox.py.md

@@ -1,9 +0,0 @@
-Summary:
-- Sandbox lifecycle wrapper (ready/cancel/fail signals, mount/unmount, release).
-
-Invariants:
-- wait_ready raises with the original initialization error as the cause.
-- release always attempts unmount and environment release, logging failures.
-
-Tests:
-- Covered by sandbox lifecycle/unit tests and workflow execution error handling.

api/agent-notes/core/sandbox/security/__init__.py.md

@@ -1,2 +0,0 @@
-Summary:
-- Sandbox security helper modules.

api/agent-notes/core/sandbox/security/archive_signer.py.md

@@ -1,13 +0,0 @@
-Summary:
-- Generates and verifies signed URLs for sandbox archive upload/download.
-
-Invariants:
-- tenant_id and sandbox_id must be UUIDs.
-- Signatures are tenant-scoped and include operation, expiry, and nonce.
-
-Edge Cases:
-- Missing tenant private key raises ValueError.
-- Expired or tampered signatures are rejected.
-
-Tests:
-- Add unit tests if sandbox archive signature behavior expands.

api/agent-notes/core/sandbox/storage/archive_storage.py.md

@@ -1,12 +0,0 @@
-Summary:
-- Manages sandbox archive uploads/downloads for workspace persistence.
-
-Invariants:
-- Archive storage key is sandbox/<tenant_id>/<sandbox_id>.tar.gz.
-- Signed URLs are tenant-scoped and use external files URL.
-
-Edge Cases:
-- Missing archive skips mount.
-
-Tests:
-- Covered indirectly via sandbox integration tests.

api/agent-notes/core/skill/skill_manager.py.md

@@ -1,9 +0,0 @@
-Summary:
-Summary:
-- Loads/saves skill bundles to app asset storage.
-
-Invariants:
-- Skill bundles use AppAssetStorage refs and JSON serialization.
-
-Tests:
-- Covered by skill bundle build/load unit tests.

api/agent-notes/core/virtual_environment/providers/e2b_sandbox.py.md

@@ -1,16 +0,0 @@
-# E2B Sandbox Provider Notes
-
-## Purpose
-- Implements the E2B-backed `VirtualEnvironment` provider and bootstraps sandbox metadata, file I/O, and command execution.
-
-## Key Decisions
-- Sandbox metadata is gathered during `_construct_environment` using the E2B SDK before returning `Metadata`.
-- Architecture/OS detection uses a single `uname -m -s` call split by whitespace to reduce round-trips.
-- Command execution streams stdout/stderr through `QueueTransportReadCloser`; stdin is unsupported.
-
-## Edge Cases
-- `release_environment` raises when sandbox termination fails.
-- `execute_command` runs in a background thread; consumers must read stdout/stderr until EOF.
-
-## Tests/Verification
-- None yet. Add targeted service tests when behavior changes.

api/agent-notes/services/app_asset_service.py.md

@@ -1,14 +0,0 @@
-Summary:
-- App asset CRUD, publish/build pipeline, and presigned URL generation.
-
-Invariants:
-- Asset storage access goes through AppAssetStorage + AssetPath, using app_asset_storage singleton.
-- Tree operations require tenant/app scoping and lock for mutation.
-- Asset zips are packaged via raw storage with storage keys from AppAssetStorage.
-
-Edge Cases:
-- File nodes larger than preview limit are rejected.
-- Deletion runs asynchronously; storage failures are logged.
-
-Tests:
-- Unit tests for storage URL generation and publish/build flows.

api/agent-notes/services/app_bundle_service.py.md

@@ -1,10 +0,0 @@
-Summary:
-Summary:
-- Imports app bundles, including asset extraction into app asset storage.
-
-Invariants:
-- Asset imports respect zip security checks and tenant/app scoping.
-- Draft asset packaging uses AppAssetStorage for key mapping.
-
-Tests:
-- Bundle import unit tests and zip validation coverage.

api/agent-notes/tests/unit_tests/core/app_assets/test_storage.py.md

@@ -1,6 +0,0 @@
-Summary:
-Summary:
-- Unit tests for AppAssetStorage ref validation, key mapping, and signing.
-
-Tests:
-- Covers valid/invalid refs, signature verify, expiration handling, and proxy URL generation.

api/app.py

@@ -1,5 +1,12 @@
-import os
+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:
@@ -9,15 +16,10 @@ def is_db_command() -> bool:
 
 
 # create app
-flask_app = None
-socketio_app = None
-
 if is_db_command():
     from app_factory import create_migrations_app
 
     app = create_migrations_app()
-    socketio_app = app
-    flask_app = app
 else:
     # Gunicorn and Celery handle monkey patching automatically in production by
     # specifying the `gevent` worker class. Manual monkey patching is not required here.
@@ -28,15 +30,8 @@ else:
 
     from app_factory import create_app
 
-    socketio_app, flask_app = create_app()
-    app = flask_app
-    celery = flask_app.extensions["celery"]
+    app = create_app()
+    celery = cast("Celery", app.extensions["celery"])
 
 if __name__ == "__main__":
-    from gevent import pywsgi
-    from geventwebsocket.handler import WebSocketHandler  # type: ignore[reportMissingTypeStubs]
-
-    host = os.environ.get("HOST", "0.0.0.0")
-    port = int(os.environ.get("PORT", 5001))
-    server = pywsgi.WSGIServer((host, port), socketio_app, handler_class=WebSocketHandler)
-    server.serve_forever()
+    app.run(host="0.0.0.0", port=5001)

api/app_factory.py

@@ -1,7 +1,6 @@
 import logging
 import time
 
-import socketio  # type: ignore[reportMissingTypeStubs]
 from opentelemetry.trace import get_current_span
 from opentelemetry.trace.span import INVALID_SPAN_ID, INVALID_TRACE_ID
 
@@ -9,7 +8,6 @@ from configs import dify_config
 from contexts.wrapper import RecyclableContextVar
 from core.logging.context import init_request_context
 from dify_app import DifyApp
-from extensions.ext_socketio import sio
 
 logger = logging.getLogger(__name__)
 
@@ -62,18 +60,14 @@ def create_flask_app_with_configs() -> DifyApp:
     return dify_app
 
 
-def create_app() -> tuple[socketio.WSGIApp, DifyApp]:
+def create_app() -> DifyApp:
     start_time = time.perf_counter()
     app = create_flask_app_with_configs()
     initialize_extensions(app)
-
-    sio.app = app
-    socketio_app = socketio.WSGIApp(sio, app)
-
     end_time = time.perf_counter()
     if dify_config.DEBUG:
         logger.info("Finished create_app (%s ms)", round((end_time - start_time) * 1000, 2))
-    return socketio_app, app
+    return app
 
 
 def initialize_extensions(app: DifyApp):
@@ -155,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

@@ -23,14 +23,14 @@ from core.rag.datasource.vdb.vector_factory import Vector
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.index_processor.constant.built_in_field import BuiltInField
 from core.rag.models.document import ChildDocument, Document
-from core.sandbox import SandboxBuilder, SandboxType
-from core.tools.utils.system_encryption import encrypt_system_params
+from core.tools.utils.system_oauth_encryption import encrypt_system_oauth_params
 from events.app_event import app_was_created
 from extensions.ext_database import db
 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.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
@@ -55,6 +55,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")
@@ -728,8 +730,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"))
 
@@ -738,12 +747,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")
 
@@ -1451,54 +1464,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:
@@ -1608,7 +1625,7 @@ def remove_orphaned_files_on_storage(force: bool):
             click.echo(click.style(f"- Scanning files on storage path {storage_path}", fg="white"))
             files = storage.scan(path=storage_path, files=True, directories=False)
             all_files_on_storage.extend(files)
-        except FileNotFoundError:
+        except FileNotFoundError as e:
             click.echo(click.style(f"  -> Skipping path {storage_path} as it does not exist.", fg="yellow"))
             continue
         except Exception as e:
@@ -1738,59 +1755,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]
@@ -1813,6 +1789,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 = {
@@ -1856,57 +1876,6 @@ def file_usage(
             click.echo(click.style(f"Use --offset {offset + limit} to see next page", fg="white"))
 
 
-@click.command("setup-sandbox-system-config", help="Setup system-level sandbox provider configuration.")
-@click.option(
-    "--provider-type", prompt=True, type=click.Choice(["e2b", "docker", "local"]), help="Sandbox provider type"
-)
-@click.option("--config", prompt=True, help='Configuration JSON (e.g., {"api_key": "xxx"} for e2b)')
-def setup_sandbox_system_config(provider_type: str, config: str):
-    """
-    Setup system-level sandbox provider configuration.
-
-    Examples:
-        flask setup-sandbox-system-config --provider-type e2b --config '{"api_key": "e2b_xxx"}'
-        flask setup-sandbox-system-config --provider-type docker --config '{"docker_sock": "unix:///var/run/docker.sock"}'
-        flask setup-sandbox-system-config --provider-type local --config '{}'
-    """
-    from models.sandbox import SandboxProviderSystemConfig
-
-    try:
-        click.echo(click.style(f"Validating config: {config}", fg="yellow"))
-        config_dict = TypeAdapter(dict[str, Any]).validate_json(config)
-        click.echo(click.style("Config validated successfully.", fg="green"))
-
-        click.echo(click.style(f"Validating config schema for provider type: {provider_type}", fg="yellow"))
-        SandboxBuilder.validate(SandboxType(provider_type), config_dict)
-        click.echo(click.style("Config schema validated successfully.", fg="green"))
-
-        click.echo(click.style("Encrypting config...", fg="yellow"))
-        click.echo(click.style(f"Using SECRET_KEY: `{dify_config.SECRET_KEY}`", fg="yellow"))
-        encrypted_config = encrypt_system_params(config_dict)
-        click.echo(click.style("Config encrypted successfully.", fg="green"))
-    except Exception as e:
-        click.echo(click.style(f"Error validating/encrypting config: {str(e)}", fg="red"))
-        return
-
-    deleted_count = db.session.query(SandboxProviderSystemConfig).filter_by(provider_type=provider_type).delete()
-    if deleted_count > 0:
-        click.echo(
-            click.style(
-                f"Deleted {deleted_count} existing system config for provider type: {provider_type}", fg="yellow"
-            )
-        )
-
-    system_config = SandboxProviderSystemConfig(
-        provider_type=provider_type,
-        encrypted_config=encrypted_config,
-    )
-    db.session.add(system_config)
-    db.session.commit()
-    click.echo(click.style(f"Sandbox system config setup successfully. id: {system_config.id}", fg="green"))
-    click.echo(click.style(f"Provider type: {provider_type}", fg="green"))
-
-
 @click.command("setup-system-tool-oauth-client", help="Setup system tool oauth client.")
 @click.option("--provider", prompt=True, help="Provider name")
 @click.option("--client-params", prompt=True, help="Client Params")
@@ -1926,7 +1895,7 @@ def setup_system_tool_oauth_client(provider, client_params):
 
         click.echo(click.style(f"Encrypting client params: {client_params}", fg="yellow"))
         click.echo(click.style(f"Using SECRET_KEY: `{dify_config.SECRET_KEY}`", fg="yellow"))
-        oauth_client_params = encrypt_system_params(client_params_dict)
+        oauth_client_params = encrypt_system_oauth_params(client_params_dict)
         click.echo(click.style("Client params encrypted successfully.", fg="green"))
     except Exception as e:
         click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))
@@ -1975,7 +1944,7 @@ def setup_system_trigger_oauth_client(provider, client_params):
 
         click.echo(click.style(f"Encrypting client params: {client_params}", fg="yellow"))
         click.echo(click.style(f"Using SECRET_KEY: `{dify_config.SECRET_KEY}`", fg="yellow"))
-        oauth_client_params = encrypt_system_params(client_params_dict)
+        oauth_client_params = encrypt_system_oauth_params(client_params_dict)
         click.echo(click.style("Client params encrypted successfully.", fg="green"))
     except Exception as e:
         click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))

api/configs/app_config.py

@@ -2,7 +2,6 @@ import logging
 from pathlib import Path
 from typing import Any
 
-from pydantic import Field
 from pydantic.fields import FieldInfo
 from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict, TomlConfigSettingsSource
 
@@ -83,17 +82,6 @@ class DifyConfig(
         extra="ignore",
     )
 
-    SANDBOX_DIFY_CLI_ROOT: str | None = Field(
-        default=None,
-        description=(
-            "Filesystem directory containing dify CLI binaries named dify-cli-<os>-<arch>. "
-            "Defaults to api/bin when unset."
-        ),
-    )
-    DIFY_PORT: int = Field(
-        default=5001,
-        description="Port used by Dify to communicate with the host machine.",
-    )
     # Before adding any config,
     # please consider to arrange it in the proper config group of existed or added
     # for better readability and maintainability.

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):
     """
@@ -248,15 +265,9 @@ class PluginConfig(BaseSettings):
         default=60 * 60,
     )
 
-
-class CliApiConfig(BaseSettings):
-    """
-    Configuration for CLI API (for dify-cli to call back from external sandbox environments)
-    """
-
-    CLI_API_URL: str = Field(
-        description="CLI API URL for external sandbox (e.g., e2b) to call back.",
-        default="http://localhost:5001",
+    PLUGIN_MAX_FILE_SIZE: PositiveInt = Field(
+        description="Maximum allowed size (bytes) for plugin-generated files",
+        default=50 * 1024 * 1024,
     )
 
 
@@ -1145,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,
@@ -1166,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",
@@ -1245,13 +1274,6 @@ class PositionConfig(BaseSettings):
         return {item.strip() for item in self.POSITION_TOOL_EXCLUDES.split(",") if item.strip() != ""}
 
 
-class CollaborationConfig(BaseSettings):
-    ENABLE_COLLABORATION_MODE: bool = Field(
-        description="Whether to enable collaboration mode features across the workspace",
-        default=False,
-    )
-
-
 class LoginConfig(BaseSettings):
     ENABLE_EMAIL_CODE_LOGIN: bool = Field(
         description="whether to enable email code login",
@@ -1297,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):
@@ -1327,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,
@@ -1346,7 +1375,6 @@ class FeatureConfig(
     TriggerConfig,
     AsyncWorkflowConfig,
     PluginConfig,
-    CliApiConfig,
     MarketplaceConfig,
     DataSetConfig,
     EndpointConfig,
@@ -1371,7 +1399,6 @@ class FeatureConfig(
     WorkflowConfig,
     WorkflowNodeExecutionConfig,
     WorkspaceConfig,
-    CollaborationConfig,
     LoginConfig,
     AccountConfig,
     SwaggerUIConfig,

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_pubsub_config.py

@@ -0,0 +1,96 @@
+from typing import Literal, Protocol
+from urllib.parse import quote_plus, urlunparse
+
+from pydantic import 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 Redis pub/sub streaming.
+    """
+
+    PUBSUB_REDIS_URL: str | None = Field(
+        alias="PUBSUB_REDIS_URL",
+        description=(
+            "Redis connection URL for pub/sub streaming events between API "
+            "and celery worker, defaults to url constructed from "
+            "`REDIS_*` configurations"
+        ),
+        default=None,
+    )
+
+    PUBSUB_REDIS_USE_CLUSTERS: bool = Field(
+        description=(
+            "Enable Redis Cluster mode for pub/sub streaming. It's highly "
+            "recommended to enable this for large deployments."
+        ),
+        default=False,
+    )
+
+    PUBSUB_REDIS_CHANNEL_TYPE: Literal["pubsub", "sharded"] = Field(
+        description=(
+            "Pub/sub channel type for streaming events. "
+            "Valid options are:\n"
+            "\n"
+            " - pubsub: for normal Pub/Sub\n"
+            " - sharded: for sharded Pub/Sub\n"
+            "\n"
+            "It's highly recommended to use sharded Pub/Sub AND redis cluster "
+            "for large deployments."
+        ),
+        default="pubsub",
+    )
+
+    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/controllers/cli_api/__init__.py

@@ -1,27 +0,0 @@
-from flask import Blueprint
-from flask_restx import Namespace
-
-from libs.external_api import ExternalApi
-
-bp = Blueprint("cli_api", __name__, url_prefix="/cli/api")
-
-api = ExternalApi(
-    bp,
-    version="1.0",
-    title="CLI API",
-    description="APIs for Dify CLI to call back from external sandbox environments (e.g., e2b)",
-)
-
-# Create namespace
-cli_api_ns = Namespace("cli_api", description="CLI API operations", path="/")
-
-from .dify_cli import cli_api as _plugin
-
-api.add_namespace(cli_api_ns)
-
-__all__ = [
-    "_plugin",
-    "api",
-    "bp",
-    "cli_api_ns",
-]

api/controllers/cli_api/dify_cli/cli_api.py

@@ -1,192 +0,0 @@
-from flask import abort
-from flask_restx import Resource
-from pydantic import BaseModel
-
-from controllers.cli_api import cli_api_ns
-from controllers.cli_api.dify_cli.wraps import get_cli_user_tenant, plugin_data
-from controllers.cli_api.wraps import cli_api_only
-from controllers.console.wraps import setup_required
-from core.app.entities.app_invoke_entities import InvokeFrom
-from core.file.helpers import get_signed_file_url_for_plugin
-from core.plugin.backwards_invocation.app import PluginAppBackwardsInvocation
-from core.plugin.backwards_invocation.base import BaseBackwardsInvocationResponse
-from core.plugin.backwards_invocation.model import PluginModelBackwardsInvocation
-from core.plugin.backwards_invocation.tool import PluginToolBackwardsInvocation
-from core.plugin.entities.request import (
-    RequestInvokeApp,
-    RequestInvokeLLM,
-    RequestInvokeTool,
-    RequestRequestUploadFile,
-)
-from core.sandbox.bash.dify_cli import DifyCliToolConfig
-from core.session.cli_api import CliContext
-from core.skill.entities import ToolInvocationRequest
-from core.tools.entities.tool_entities import ToolProviderType
-from core.tools.tool_manager import ToolManager
-from libs.helper import length_prefixed_response
-from models.account import Account
-from models.model import EndUser, Tenant
-
-
-class FetchToolItem(BaseModel):
-    tool_type: str
-    tool_provider: str
-    tool_name: str
-    credential_id: str | None = None
-
-
-class FetchToolBatchRequest(BaseModel):
-    tools: list[FetchToolItem]
-
-
-@cli_api_ns.route("/invoke/llm")
-class CliInvokeLLMApi(Resource):
-    @cli_api_only
-    @get_cli_user_tenant
-    @setup_required
-    @plugin_data(payload_type=RequestInvokeLLM)
-    def post(
-        self,
-        user_model: Account | EndUser,
-        tenant_model: Tenant,
-        payload: RequestInvokeLLM,
-        cli_context: CliContext,
-    ):
-        def generator():
-            response = PluginModelBackwardsInvocation.invoke_llm(user_model.id, tenant_model, payload)
-            return PluginModelBackwardsInvocation.convert_to_event_stream(response)
-
-        return length_prefixed_response(0xF, generator())
-
-
-@cli_api_ns.route("/invoke/tool")
-class CliInvokeToolApi(Resource):
-    @cli_api_only
-    @get_cli_user_tenant
-    @setup_required
-    @plugin_data(payload_type=RequestInvokeTool)
-    def post(
-        self,
-        user_model: Account | EndUser,
-        tenant_model: Tenant,
-        payload: RequestInvokeTool,
-        cli_context: CliContext,
-    ):
-        tool_type = ToolProviderType.value_of(payload.tool_type)
-
-        request = ToolInvocationRequest(
-            tool_type=tool_type,
-            provider=payload.provider,
-            tool_name=payload.tool,
-            credential_id=payload.credential_id,
-        )
-        if cli_context.tool_access and not cli_context.tool_access.is_allowed(request):
-            abort(403, description=f"Access denied for tool: {payload.provider}/{payload.tool}")
-
-        def generator():
-            return PluginToolBackwardsInvocation.convert_to_event_stream(
-                PluginToolBackwardsInvocation.invoke_tool(
-                    tenant_id=tenant_model.id,
-                    user_id=user_model.id,
-                    tool_type=tool_type,
-                    provider=payload.provider,
-                    tool_name=payload.tool,
-                    tool_parameters=payload.tool_parameters,
-                    credential_id=payload.credential_id,
-                ),
-            )
-
-        return length_prefixed_response(0xF, generator())
-
-
-@cli_api_ns.route("/invoke/app")
-class CliInvokeAppApi(Resource):
-    @cli_api_only
-    @get_cli_user_tenant
-    @setup_required
-    @plugin_data(payload_type=RequestInvokeApp)
-    def post(
-        self,
-        user_model: Account | EndUser,
-        tenant_model: Tenant,
-        payload: RequestInvokeApp,
-        cli_context: CliContext,
-    ):
-        response = PluginAppBackwardsInvocation.invoke_app(
-            app_id=payload.app_id,
-            user_id=user_model.id,
-            tenant_id=tenant_model.id,
-            conversation_id=payload.conversation_id,
-            query=payload.query,
-            stream=payload.response_mode == "streaming",
-            inputs=payload.inputs,
-            files=payload.files,
-        )
-
-        return length_prefixed_response(0xF, PluginAppBackwardsInvocation.convert_to_event_stream(response))
-
-
-@cli_api_ns.route("/upload/file/request")
-class CliUploadFileRequestApi(Resource):
-    @cli_api_only
-    @get_cli_user_tenant
-    @setup_required
-    @plugin_data(payload_type=RequestRequestUploadFile)
-    def post(
-        self,
-        user_model: Account | EndUser,
-        tenant_model: Tenant,
-        payload: RequestRequestUploadFile,
-        cli_context: CliContext,
-    ):
-        url = get_signed_file_url_for_plugin(
-            filename=payload.filename,
-            mimetype=payload.mimetype,
-            tenant_id=tenant_model.id,
-            user_id=user_model.id,
-        )
-        return BaseBackwardsInvocationResponse(data={"url": url}).model_dump()
-
-
-@cli_api_ns.route("/fetch/tools/batch")
-class CliFetchToolsBatchApi(Resource):
-    @cli_api_only
-    @get_cli_user_tenant
-    @setup_required
-    @plugin_data(payload_type=FetchToolBatchRequest)
-    def post(
-        self,
-        user_model: Account | EndUser,
-        tenant_model: Tenant,
-        payload: FetchToolBatchRequest,
-        cli_context: CliContext,
-    ):
-        tools: list[dict] = []
-
-        for item in payload.tools:
-            provider_type = ToolProviderType.value_of(item.tool_type)
-
-            request = ToolInvocationRequest(
-                tool_type=provider_type,
-                provider=item.tool_provider,
-                tool_name=item.tool_name,
-                credential_id=item.credential_id,
-            )
-            if cli_context.tool_access and not cli_context.tool_access.is_allowed(request):
-                abort(403, description=f"Access denied for tool: {item.tool_provider}/{item.tool_name}")
-
-            try:
-                tool_runtime = ToolManager.get_tool_runtime(
-                    tenant_id=tenant_model.id,
-                    provider_type=provider_type,
-                    provider_id=item.tool_provider,
-                    tool_name=item.tool_name,
-                    invoke_from=InvokeFrom.AGENT,
-                    credential_id=item.credential_id,
-                )
-                tool_config = DifyCliToolConfig.create_from_tool(tool_runtime)
-                tools.append(tool_config.model_dump())
-            except Exception:
-                continue
-
-        return BaseBackwardsInvocationResponse(data={"tools": tools}).model_dump()

api/controllers/cli_api/dify_cli/wraps.py

@@ -1,137 +0,0 @@
-from collections.abc import Callable
-from functools import wraps
-from typing import ParamSpec, TypeVar
-
-from flask import current_app, g, request
-from flask_login import user_logged_in
-from pydantic import BaseModel
-from sqlalchemy.orm import Session
-
-from core.session.cli_api import CliApiSession, CliContext
-from extensions.ext_database import db
-from libs.login import current_user
-from models.account import Tenant
-from models.model import DefaultEndUserSessionID, EndUser
-
-P = ParamSpec("P")
-R = TypeVar("R")
-
-
-class TenantUserPayload(BaseModel):
-    tenant_id: str
-    user_id: str
-
-
-def get_user(tenant_id: str, user_id: str | None) -> EndUser:
-    """
-    Get current user
-
-    NOTE: user_id is not trusted, it could be maliciously set to any value.
-    As a result, it could only be considered as an end user id.
-    """
-    if not user_id:
-        user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
-    is_anonymous = user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID
-    try:
-        with Session(db.engine) as session:
-            user_model = None
-
-            if is_anonymous:
-                user_model = (
-                    session.query(EndUser)
-                    .where(
-                        EndUser.session_id == user_id,
-                        EndUser.tenant_id == tenant_id,
-                    )
-                    .first()
-                )
-            else:
-                user_model = (
-                    session.query(EndUser)
-                    .where(
-                        EndUser.id == user_id,
-                        EndUser.tenant_id == tenant_id,
-                    )
-                    .first()
-                )
-
-            if not user_model:
-                user_model = EndUser(
-                    tenant_id=tenant_id,
-                    type="service_api",
-                    is_anonymous=is_anonymous,
-                    session_id=user_id,
-                )
-                session.add(user_model)
-                session.commit()
-                session.refresh(user_model)
-
-    except Exception:
-        raise ValueError("user not found")
-
-    return user_model
-
-
-def get_cli_user_tenant(view_func: Callable[P, R]):
-    @wraps(view_func)
-    def decorated_view(*args: P.args, **kwargs: P.kwargs):
-        session: CliApiSession | None = getattr(g, "cli_api_session", None)
-        if session is None:
-            raise ValueError("session not found")
-
-        user_id = session.user_id
-        tenant_id = session.tenant_id
-        cli_context = CliContext.model_validate(session.context)
-
-        if not user_id:
-            user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
-
-        try:
-            tenant_model = (
-                db.session.query(Tenant)
-                .where(
-                    Tenant.id == tenant_id,
-                )
-                .first()
-            )
-        except Exception:
-            raise ValueError("tenant not found")
-
-        if not tenant_model:
-            raise ValueError("tenant not found")
-
-        kwargs["tenant_model"] = tenant_model
-        kwargs["user_model"] = get_user(tenant_id, user_id)
-        kwargs["cli_context"] = cli_context
-
-        current_app.login_manager._update_request_context_with_user(kwargs["user_model"])  # type: ignore
-        user_logged_in.send(current_app._get_current_object(), user=current_user)  # type: ignore
-
-        return view_func(*args, **kwargs)
-
-    return decorated_view
-
-
-def plugin_data(view: Callable[P, R] | None = None, *, payload_type: type[BaseModel]):
-    def decorator(view_func: Callable[P, R]):
-        @wraps(view_func)
-        def decorated_view(*args: P.args, **kwargs: P.kwargs):
-            try:
-                data = request.get_json()
-            except Exception:
-                raise ValueError("invalid json")
-
-            try:
-                payload = payload_type.model_validate(data)
-            except Exception as e:
-                raise ValueError(f"invalid payload: {str(e)}")
-
-            kwargs["payload"] = payload
-            return view_func(*args, **kwargs)
-
-        return decorated_view
-
-    if view is None:
-        return decorator
-    else:
-        return decorator(view)

api/controllers/cli_api/wraps.py

@@ -1,56 +0,0 @@
-import hashlib
-import hmac
-import time
-from collections.abc import Callable
-from functools import wraps
-from typing import ParamSpec, TypeVar
-
-from flask import abort, g, request
-
-from core.session.cli_api import CliApiSessionManager
-
-P = ParamSpec("P")
-R = TypeVar("R")
-
-SIGNATURE_TTL_SECONDS = 300
-
-
-def _verify_signature(session_secret: str, timestamp: str, body: bytes, signature: str) -> bool:
-    expected = hmac.new(
-        session_secret.encode(),
-        f"{timestamp}.".encode() + body,
-        hashlib.sha256,
-    ).hexdigest()
-    return hmac.compare_digest(f"sha256={expected}", signature)
-
-
-def cli_api_only(view: Callable[P, R]):
-    @wraps(view)
-    def decorated(*args: P.args, **kwargs: P.kwargs):
-        session_id = request.headers.get("X-Cli-Api-Session-Id")
-        timestamp = request.headers.get("X-Cli-Api-Timestamp")
-        signature = request.headers.get("X-Cli-Api-Signature")
-
-        if not session_id or not timestamp or not signature:
-            abort(401)
-
-        try:
-            ts = int(timestamp)
-            if abs(time.time() - ts) > SIGNATURE_TTL_SECONDS:
-                abort(401)
-        except ValueError:
-            abort(401)
-
-        session = CliApiSessionManager().get(session_id)
-        if not session:
-            abort(401)
-
-        body = request.get_data()
-        if not _verify_signature(session.secret, timestamp, body, signature):
-            abort(401)
-
-        g.cli_api_session = session
-
-        return view(*args, **kwargs)
-
-    return decorated

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 core.workflow.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

@@ -32,15 +32,14 @@ for module_name in RESOURCE_MODULES:
 
 # Ensure resource modules are imported so route decorators are evaluated.
 # Import other controllers
-# Sandbox file browser
 from . import (
     admin,
     apikey,
     extension,
     feature,
+    human_input_form,
     init_validate,
     ping,
-    sandbox_files,
     setup,
     spec,
     version,
@@ -52,7 +51,6 @@ from .app import (
     agent,
     annotation,
     app,
-    app_asset,
     audio,
     completion,
     conversation,
@@ -63,11 +61,9 @@ from .app import (
     model_config,
     ops_trace,
     site,
-    skills,
     statistic,
     workflow,
     workflow_app_log,
-    workflow_comment,
     workflow_draft_variable,
     workflow_run,
     workflow_statistic,
@@ -119,7 +115,6 @@ from .explore import (
     saved_message,
     trial,
 )
-from .socketio import workflow as socketio_workflow  # pyright: ignore[reportUnusedImport]
 
 # Import tag controllers
 from .tag import tags
@@ -134,7 +129,6 @@ from .workspace import (
     model_providers,
     models,
     plugin,
-    sandbox_providers,
     tool_providers,
     trigger_providers,
     workspace,
@@ -153,7 +147,6 @@ __all__ = [
     "api",
     "apikey",
     "app",
-    "app_asset",
     "audio",
     "banner",
     "billing",
@@ -179,6 +172,7 @@ __all__ = [
     "forgot_password",
     "generator",
     "hit_testing",
+    "human_input_form",
     "init_validate",
     "installed_app",
     "load_balancing_config",
@@ -202,12 +196,9 @@ __all__ = [
     "rag_pipeline_import",
     "rag_pipeline_workflow",
     "recommended_app",
-    "sandbox_files",
-    "sandbox_providers",
     "saved_message",
     "setup",
     "site",
-    "skills",
     "spec",
     "statistic",
     "tags",
@@ -218,7 +209,6 @@ __all__ = [
     "website",
     "workflow",
     "workflow_app_log",
-    "workflow_comment",
     "workflow_draft_variable",
     "workflow_run",
     "workflow_statistic",

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,6 +1,6 @@
+import logging
 import uuid
 from datetime import datetime
-from enum import StrEnum
 from typing import Any, Literal, TypeAlias
 
 from flask import request
@@ -23,15 +23,14 @@ 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 core.workflow.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
 from models.model import IconType
-from models.workflow_features import WorkflowFeatures
 from services.app_dsl_service import AppDslService, ImportMode
 from services.app_service import AppService
 from services.enterprise.enterprise_service import EnterpriseService
@@ -56,10 +55,7 @@ ALLOW_CREATE_APP_MODES = ["chat", "agent-chat", "advanced-chat", "workflow", "co
 
 register_enum_models(console_ns, IconType)
 
-
-class RuntimeType(StrEnum):
-    CLASSIC = "classic"
-    SANDBOXED = "sandboxed"
+_logger = logging.getLogger(__name__)
 
 
 class AppListQuery(BaseModel):
@@ -126,11 +122,6 @@ class AppExportQuery(BaseModel):
     workflow_id: str | None = Field(default=None, description="Specific workflow ID to export")
 
 
-class AppExportBundleQuery(BaseModel):
-    include_secret: bool = Field(default=False, description="Include secrets in export")
-    workflow_id: str | None = Field(default=None, description="Specific workflow ID to export")
-
-
 class AppNamePayload(BaseModel):
     name: str = Field(..., min_length=1, description="Name to check")
 
@@ -356,7 +347,6 @@ class AppPartial(ResponseModel):
     create_user_name: str | None = None
     author_name: str | None = None
     has_draft_trigger: bool | None = None
-    runtime_type: RuntimeType = RuntimeType.CLASSIC
 
     @computed_field(return_type=str | None)  # type: ignore
     @property
@@ -506,13 +496,13 @@ class AppListApi(Resource):
             str(app.id) for app in app_pagination.items if app.mode in {"workflow", "advanced-chat"}
         ]
         draft_trigger_app_ids: set[str] = set()
-        sandbox_app_ids: set[str] = set()
         if workflow_capable_app_ids:
             draft_workflows = (
                 db.session.execute(
                     select(Workflow).where(
                         Workflow.version == Workflow.VERSION_DRAFT,
                         Workflow.app_id.in_(workflow_capable_app_ids),
+                        Workflow.tenant_id == current_tenant_id,
                     )
                 )
                 .scalars()
@@ -524,21 +514,18 @@ class AppListApi(Resource):
                 NodeType.TRIGGER_PLUGIN,
             }
             for workflow in draft_workflows:
-                # Check sandbox feature
-                if workflow.get_feature(WorkflowFeatures.SANDBOX).enabled:
-                    sandbox_app_ids.add(str(workflow.app_id))
-
+                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:
             app.has_draft_trigger = str(app.id) in draft_trigger_app_ids
-            app.runtime_type = RuntimeType.SANDBOXED if str(app.id) in sandbox_app_ids else RuntimeType.CLASSIC
 
         pagination_model = AppPagination.model_validate(app_pagination, from_attributes=True)
         return pagination_model.model_dump(mode="json"), 200
@@ -673,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)
 
@@ -707,29 +707,6 @@ class AppExportApi(Resource):
         return payload.model_dump(mode="json")
 
 
-@console_ns.route("/apps/<uuid:app_id>/export-bundle")
-class AppExportBundleApi(Resource):
-    @get_app_model
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @edit_permission_required
-    def get(self, app_model):
-        from services.app_bundle_service import AppBundleService
-
-        args = AppExportBundleQuery.model_validate(request.args.to_dict(flat=True))
-        current_user, _ = current_account_with_tenant()
-
-        result = AppBundleService.export_bundle(
-            app_model=app_model,
-            account_id=str(current_user.id),
-            include_secret=args.include_secret,
-            workflow_id=args.workflow_id,
-        )
-
-        return result.model_dump(mode="json")
-
-
 @console_ns.route("/apps/<uuid:app_id>/name")
 class AppNameApi(Resource):
     @console_ns.doc("check_app_name")

api/controllers/console/app/app_asset.py

@@ -1,321 +0,0 @@
-from flask import request
-from flask_restx import Resource
-from pydantic import BaseModel, Field, field_validator
-
-from controllers.console import console_ns
-from controllers.console.app.error import (
-    AppAssetNodeNotFoundError,
-    AppAssetPathConflictError,
-)
-from controllers.console.app.wraps import get_app_model
-from controllers.console.wraps import account_initialization_required, setup_required
-from core.app.entities.app_asset_entities import BatchUploadNode
-from libs.login import current_account_with_tenant, login_required
-from models import App
-from models.model import AppMode
-from services.app_asset_service import AppAssetService
-from services.errors.app_asset import (
-    AppAssetNodeNotFoundError as ServiceNodeNotFoundError,
-)
-from services.errors.app_asset import (
-    AppAssetParentNotFoundError,
-)
-from services.errors.app_asset import (
-    AppAssetPathConflictError as ServicePathConflictError,
-)
-
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-class CreateFolderPayload(BaseModel):
-    name: str = Field(..., min_length=1, max_length=255)
-    parent_id: str | None = None
-
-
-class CreateFilePayload(BaseModel):
-    name: str = Field(..., min_length=1, max_length=255)
-    parent_id: str | None = None
-
-    @field_validator("name", mode="before")
-    @classmethod
-    def strip_name(cls, v: str) -> str:
-        return v.strip() if isinstance(v, str) else v
-
-    @field_validator("parent_id", mode="before")
-    @classmethod
-    def empty_to_none(cls, v: str | None) -> str | None:
-        return v or None
-
-
-class GetUploadUrlPayload(BaseModel):
-    name: str = Field(..., min_length=1, max_length=255)
-    size: int = Field(..., ge=0)
-    parent_id: str | None = None
-
-    @field_validator("name", mode="before")
-    @classmethod
-    def strip_name(cls, v: str) -> str:
-        return v.strip() if isinstance(v, str) else v
-
-    @field_validator("parent_id", mode="before")
-    @classmethod
-    def empty_to_none(cls, v: str | None) -> str | None:
-        return v or None
-
-
-class BatchUploadPayload(BaseModel):
-    children: list[BatchUploadNode] = Field(..., min_length=1)
-
-
-class UpdateFileContentPayload(BaseModel):
-    content: str
-
-
-class RenameNodePayload(BaseModel):
-    name: str = Field(..., min_length=1, max_length=255)
-
-
-class MoveNodePayload(BaseModel):
-    parent_id: str | None = None
-
-
-class ReorderNodePayload(BaseModel):
-    after_node_id: str | None = Field(default=None, description="Place after this node, None for first position")
-
-
-def reg(cls: type[BaseModel]) -> None:
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(CreateFolderPayload)
-reg(CreateFilePayload)
-reg(GetUploadUrlPayload)
-reg(BatchUploadNode)
-reg(BatchUploadPayload)
-reg(UpdateFileContentPayload)
-reg(RenameNodePayload)
-reg(MoveNodePayload)
-reg(ReorderNodePayload)
-
-
-@console_ns.route("/apps/<string:app_id>/assets/tree")
-class AppAssetTreeResource(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def get(self, app_model: App):
-        current_user, _ = current_account_with_tenant()
-        tree = AppAssetService.get_asset_tree(app_model, current_user.id)
-        return {"children": [view.model_dump() for view in tree.transform()]}
-
-
-@console_ns.route("/apps/<string:app_id>/assets/folders")
-class AppAssetFolderResource(Resource):
-    @console_ns.expect(console_ns.models[CreateFolderPayload.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        current_user, _ = current_account_with_tenant()
-        payload = CreateFolderPayload.model_validate(console_ns.payload or {})
-
-        try:
-            node = AppAssetService.create_folder(app_model, current_user.id, payload.name, payload.parent_id)
-            return node.model_dump(), 201
-        except AppAssetParentNotFoundError:
-            raise AppAssetNodeNotFoundError()
-        except ServicePathConflictError:
-            raise AppAssetPathConflictError()
-
-
-@console_ns.route("/apps/<string:app_id>/assets/files/<string:node_id>")
-class AppAssetFileDetailResource(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def get(self, app_model: App, node_id: str):
-        current_user, _ = current_account_with_tenant()
-        try:
-            content = AppAssetService.get_file_content(app_model, current_user.id, node_id)
-            return {"content": content.decode("utf-8", errors="replace")}
-        except ServiceNodeNotFoundError:
-            raise AppAssetNodeNotFoundError()
-
-    @console_ns.expect(console_ns.models[UpdateFileContentPayload.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def put(self, app_model: App, node_id: str):
-        current_user, _ = current_account_with_tenant()
-
-        file = request.files.get("file")
-        if file:
-            content = file.read()
-        else:
-            payload = UpdateFileContentPayload.model_validate(console_ns.payload or {})
-            content = payload.content.encode("utf-8")
-
-        try:
-            node = AppAssetService.update_file_content(app_model, current_user.id, node_id, content)
-            return node.model_dump()
-        except ServiceNodeNotFoundError:
-            raise AppAssetNodeNotFoundError()
-
-
-@console_ns.route("/apps/<string:app_id>/assets/nodes/<string:node_id>")
-class AppAssetNodeResource(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def delete(self, app_model: App, node_id: str):
-        current_user, _ = current_account_with_tenant()
-        try:
-            AppAssetService.delete_node(app_model, current_user.id, node_id)
-            return {"result": "success"}, 200
-        except ServiceNodeNotFoundError:
-            raise AppAssetNodeNotFoundError()
-
-
-@console_ns.route("/apps/<string:app_id>/assets/nodes/<string:node_id>/rename")
-class AppAssetNodeRenameResource(Resource):
-    @console_ns.expect(console_ns.models[RenameNodePayload.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App, node_id: str):
-        current_user, _ = current_account_with_tenant()
-        payload = RenameNodePayload.model_validate(console_ns.payload or {})
-
-        try:
-            node = AppAssetService.rename_node(app_model, current_user.id, node_id, payload.name)
-            return node.model_dump()
-        except ServiceNodeNotFoundError:
-            raise AppAssetNodeNotFoundError()
-        except ServicePathConflictError:
-            raise AppAssetPathConflictError()
-
-
-@console_ns.route("/apps/<string:app_id>/assets/nodes/<string:node_id>/move")
-class AppAssetNodeMoveResource(Resource):
-    @console_ns.expect(console_ns.models[MoveNodePayload.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App, node_id: str):
-        current_user, _ = current_account_with_tenant()
-        payload = MoveNodePayload.model_validate(console_ns.payload or {})
-
-        try:
-            node = AppAssetService.move_node(app_model, current_user.id, node_id, payload.parent_id)
-            return node.model_dump()
-        except ServiceNodeNotFoundError:
-            raise AppAssetNodeNotFoundError()
-        except AppAssetParentNotFoundError:
-            raise AppAssetNodeNotFoundError()
-        except ServicePathConflictError:
-            raise AppAssetPathConflictError()
-
-
-@console_ns.route("/apps/<string:app_id>/assets/nodes/<string:node_id>/reorder")
-class AppAssetNodeReorderResource(Resource):
-    @console_ns.expect(console_ns.models[ReorderNodePayload.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App, node_id: str):
-        current_user, _ = current_account_with_tenant()
-        payload = ReorderNodePayload.model_validate(console_ns.payload or {})
-
-        try:
-            node = AppAssetService.reorder_node(app_model, current_user.id, node_id, payload.after_node_id)
-            return node.model_dump()
-        except ServiceNodeNotFoundError:
-            raise AppAssetNodeNotFoundError()
-
-
-@console_ns.route("/apps/<string:app_id>/assets/files/<string:node_id>/download-url")
-class AppAssetFileDownloadUrlResource(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def get(self, app_model: App, node_id: str):
-        current_user, _ = current_account_with_tenant()
-        try:
-            download_url = AppAssetService.get_file_download_url(app_model, current_user.id, node_id)
-            return {"download_url": download_url}
-        except ServiceNodeNotFoundError:
-            raise AppAssetNodeNotFoundError()
-
-
-@console_ns.route("/apps/<string:app_id>/assets/files/upload")
-class AppAssetFileUploadUrlResource(Resource):
-    @console_ns.expect(console_ns.models[GetUploadUrlPayload.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        current_user, _ = current_account_with_tenant()
-        payload = GetUploadUrlPayload.model_validate(console_ns.payload or {})
-
-        try:
-            node, upload_url = AppAssetService.get_file_upload_url(
-                app_model, current_user.id, payload.name, payload.size, payload.parent_id
-            )
-            return {"node": node.model_dump(), "upload_url": upload_url}, 201
-        except AppAssetParentNotFoundError:
-            raise AppAssetNodeNotFoundError()
-        except ServicePathConflictError:
-            raise AppAssetPathConflictError()
-
-
-@console_ns.route("/apps/<string:app_id>/assets/batch-upload")
-class AppAssetBatchUploadResource(Resource):
-    @console_ns.expect(console_ns.models[BatchUploadPayload.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        """
-        Create nodes from tree structure and return upload URLs.
-
-        Input:
-        {
-            "children": [
-                {"name": "folder1", "node_type": "folder", "children": [
-                    {"name": "file1.txt", "node_type": "file", "size": 1024}
-                ]},
-                {"name": "root.txt", "node_type": "file", "size": 512}
-            ]
-        }
-
-        Output:
-        {
-            "children": [
-                {"id": "xxx", "name": "folder1", "node_type": "folder", "children": [
-                    {"id": "yyy", "name": "file1.txt", "node_type": "file", "size": 1024, "upload_url": "..."}
-                ]},
-                {"id": "zzz", "name": "root.txt", "node_type": "file", "size": 512, "upload_url": "..."}
-            ]
-        }
-        """
-        current_user, _ = current_account_with_tenant()
-        payload = BatchUploadPayload.model_validate(console_ns.payload or {})
-
-        try:
-            result_children = AppAssetService.batch_create_from_tree(app_model, current_user.id, payload.children)
-            return {"children": [child.model_dump() for child in result_children]}, 201
-        except AppAssetParentNotFoundError:
-            raise AppAssetNodeNotFoundError()
-        except ServicePathConflictError:
-            raise AppAssetPathConflictError()

api/controllers/console/app/app_import.py

@@ -51,14 +51,6 @@ class AppImportPayload(BaseModel):
     app_id: str | None = Field(None)
 
 
-class AppImportBundleConfirmPayload(BaseModel):
-    name: str | None = None
-    description: str | None = None
-    icon_type: str | None = None
-    icon: str | None = None
-    icon_background: str | None = None
-
-
 console_ns.schema_model(
     AppImportPayload.__name__, AppImportPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
 )
@@ -147,68 +139,3 @@ class AppImportCheckDependenciesApi(Resource):
             result = import_service.check_dependencies(app_model=app_model)
 
         return result.model_dump(mode="json"), 200
-
-
-@console_ns.route("/apps/imports-bundle/prepare")
-class AppImportBundlePrepareApi(Resource):
-    """Step 1: Get upload URL for bundle import."""
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @edit_permission_required
-    def post(self):
-        from services.app_bundle_service import AppBundleService
-
-        current_user, current_tenant_id = current_account_with_tenant()
-
-        result = AppBundleService.prepare_import(
-            tenant_id=current_tenant_id,
-            account_id=current_user.id,
-        )
-
-        return {"import_id": result.import_id, "upload_url": result.upload_url}, 200
-
-
-@console_ns.route("/apps/imports-bundle/<string:import_id>/confirm")
-class AppImportBundleConfirmApi(Resource):
-    """Step 2: Confirm bundle import after upload."""
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @marshal_with(app_import_model)
-    @cloud_edition_billing_resource_check("apps")
-    @edit_permission_required
-    def post(self, import_id: str):
-        from flask import request
-
-        from core.app.entities.app_bundle_entities import BundleFormatError
-        from services.app_bundle_service import AppBundleService
-
-        current_user, _ = current_account_with_tenant()
-
-        args = AppImportBundleConfirmPayload.model_validate(request.get_json() or {})
-
-        try:
-            result = AppBundleService.confirm_import(
-                import_id=import_id,
-                account=current_user,
-                name=args.name,
-                description=args.description,
-                icon_type=args.icon_type,
-                icon=args.icon,
-                icon_background=args.icon_background,
-            )
-        except BundleFormatError as e:
-            return {"error": str(e)}, 400
-
-        if result.app_id and FeatureService.get_system_features().webapp_auth.enabled:
-            EnterpriseService.WebAppAuth.update_app_access_mode(result.app_id, "private")
-
-        status = result.status
-        if status == ImportStatus.FAILED:
-            return result.model_dump(mode="json"), 400
-        elif status == ImportStatus.PENDING:
-            return result.model_dump(mode="json"), 202
-        return result.model_dump(mode="json"), 200

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,
@@ -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/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/error.py

@@ -110,6 +110,8 @@ class TracingConfigCheckError(BaseHTTPException):
 
 
 class InvokeRateLimitError(BaseHTTPException):
+    """Raised when the Invoke returns rate limit error."""
+
     error_code = "rate_limit_error"
     description = "Rate Limit Error"
     code = 429
@@ -119,21 +121,3 @@ class NeedAddIdsError(BaseHTTPException):
     error_code = "need_add_ids"
     description = "Need to add ids."
     code = 400
-
-
-class AppAssetNodeNotFoundError(BaseHTTPException):
-    error_code = "app_asset_node_not_found"
-    description = "App asset node not found."
-    code = 404
-
-
-class AppAssetFileRequiredError(BaseHTTPException):
-    error_code = "app_asset_file_required"
-    description = "File is required."
-    code = 400
-
-
-class AppAssetPathConflictError(BaseHTTPException):
-    error_code = "app_asset_path_conflict"
-    description = "Path already exists."
-    code = 409

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,15 +11,12 @@ 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.context_models import (
-    AvailableVarPayload,
-    CodeContextPayload,
-    ParameterInfoPayload,
-)
+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 extensions.ext_database import db
@@ -31,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")
 
 
@@ -60,34 +41,6 @@ class InstructionTemplatePayload(BaseModel):
     type: str = Field(..., description="Instruction template type")
 
 
-class ContextGeneratePayload(BaseModel):
-    """Payload for generating extractor code node."""
-
-    language: str = Field(default="python3", description="Code language (python3/javascript)")
-    prompt_messages: list[dict[str, Any]] = Field(
-        ..., description="Multi-turn conversation history, last message is the current instruction"
-    )
-    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
-    available_vars: list[AvailableVarPayload] = Field(..., description="Available variables from upstream nodes")
-    parameter_info: ParameterInfoPayload = Field(..., description="Target parameter metadata from the frontend")
-    code_context: CodeContextPayload = Field(description="Existing code node context for incremental generation")
-
-
-class SuggestedQuestionsPayload(BaseModel):
-    """Payload for generating suggested questions."""
-
-    language: str = Field(
-        default="English", description="Language for generated questions (e.g. English, Chinese, Japanese)"
-    )
-    model_config_data: dict[str, Any] = Field(
-        default_factory=dict,
-        alias="model_config",
-        description="Model configuration (optional, uses system default if not provided)",
-    )
-    available_vars: list[AvailableVarPayload] = Field(..., description="Available variables from upstream nodes")
-    parameter_info: ParameterInfoPayload = Field(..., description="Target parameter metadata from the frontend")
-
-
 def reg(cls: type[BaseModel]):
     console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
 
@@ -97,8 +50,7 @@ reg(RuleCodeGeneratePayload)
 reg(RuleStructuredOutputPayload)
 reg(InstructionGeneratePayload)
 reg(InstructionTemplatePayload)
-reg(ContextGeneratePayload)
-reg(SuggestedQuestionsPayload)
+reg(ModelConfig)
 
 
 @console_ns.route("/rule-generate")
@@ -117,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:
@@ -153,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)
@@ -187,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)
@@ -239,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}"}
@@ -313,70 +263,3 @@ class InstructionGenerationTemplateApi(Resource):
                 return {"data": INSTRUCTION_GENERATE_TEMPLATE_CODE}
             case _:
                 raise ValueError(f"Invalid type: {args.type}")
-
-
-@console_ns.route("/context-generate")
-class ContextGenerateApi(Resource):
-    @console_ns.doc("generate_with_context")
-    @console_ns.doc(description="Generate with multi-turn conversation context")
-    @console_ns.expect(console_ns.models[ContextGeneratePayload.__name__])
-    @console_ns.response(200, "Content generated successfully")
-    @console_ns.response(400, "Invalid request parameters or workflow not found")
-    @console_ns.response(402, "Provider quota exceeded")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self):
-        from core.llm_generator.utils import deserialize_prompt_messages
-
-        args = ContextGeneratePayload.model_validate(console_ns.payload)
-        _, current_tenant_id = current_account_with_tenant()
-
-        try:
-            return LLMGenerator.generate_with_context(
-                tenant_id=current_tenant_id,
-                language=args.language,
-                prompt_messages=deserialize_prompt_messages(args.prompt_messages),
-                model_config=args.model_config_data,
-                available_vars=args.available_vars,
-                parameter_info=args.parameter_info,
-                code_context=args.code_context,
-            )
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-
-
-@console_ns.route("/context-generate/suggested-questions")
-class SuggestedQuestionsApi(Resource):
-    @console_ns.doc("generate_suggested_questions")
-    @console_ns.doc(description="Generate suggested questions for context generation")
-    @console_ns.expect(console_ns.models[SuggestedQuestionsPayload.__name__])
-    @console_ns.response(200, "Questions generated successfully")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self):
-        args = SuggestedQuestionsPayload.model_validate(console_ns.payload)
-        _, current_tenant_id = current_account_with_tenant()
-        try:
-            return LLMGenerator.generate_suggested_questions(
-                tenant_id=current_tenant_id,
-                language=args.language,
-                available_vars=args.available_vars,
-                parameter_info=args.parameter_info,
-                model_config=args.model_config_data,
-            )
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)

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,
@@ -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,11 +207,11 @@ 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,
         "parent_message_id": fields.String,
-        "generation_detail": fields.Raw,
     },
 )
 
@@ -232,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)
@@ -291,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)
 
@@ -357,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
@@ -377,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
@@ -429,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
@@ -475,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/skills.py

@@ -1,83 +0,0 @@
-from flask_restx import Resource
-
-from controllers.console import console_ns
-from controllers.console.app.error import DraftWorkflowNotExist
-from controllers.console.app.wraps import get_app_model
-from controllers.console.wraps import account_initialization_required, current_account_with_tenant, setup_required
-from libs.login import login_required
-from models import App
-from models.model import AppMode
-from services.skill_service import SkillService
-from services.workflow_service import WorkflowService
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/skills")
-class NodeSkillsApi(Resource):
-    """API for retrieving skill references for a specific workflow node."""
-
-    @console_ns.doc("get_node_skills")
-    @console_ns.doc(description="Get skill references for a specific node in the draft workflow")
-    @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @console_ns.response(200, "Node skills retrieved successfully")
-    @console_ns.response(404, "Workflow or node not found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def get(self, app_model: App, node_id: str):
-        """
-        Get skill information for a specific node in the draft workflow.
-
-        Returns information about skill references in the node, including:
-        - skill_references: List of prompt messages marked as skills
-        - tool_references: Aggregated tool references from all skill prompts
-        - file_references: Aggregated file references from all skill prompts
-        """
-        current_user, _ = current_account_with_tenant()
-        workflow_service = WorkflowService()
-        workflow = workflow_service.get_draft_workflow(app_model=app_model)
-
-        if not workflow:
-            raise DraftWorkflowNotExist()
-
-        skill_info = SkillService.get_node_skill_info(
-            app=app_model,
-            workflow=workflow,
-            node_id=node_id,
-            user_id=current_user.id,
-        )
-        return skill_info.model_dump()
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/skills")
-class WorkflowSkillsApi(Resource):
-    """API for retrieving all skill references in a workflow."""
-
-    @console_ns.doc("get_workflow_skills")
-    @console_ns.doc(description="Get all skill references in the draft workflow")
-    @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Workflow skills retrieved successfully")
-    @console_ns.response(404, "Workflow not found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def get(self, app_model: App):
-        """
-        Get skill information for all nodes in the draft workflow that have skill references.
-
-        Returns a list of nodes with their skill information.
-        """
-        current_user, _ = current_account_with_tenant()
-        workflow_service = WorkflowService()
-        workflow = workflow_service.get_draft_workflow(app_model=app_model)
-
-        if not workflow:
-            raise DraftWorkflowNotExist()
-
-        skills_info = SkillService.get_workflow_skills(
-            app=app_model,
-            workflow=workflow,
-            user_id=current_user.id,
-        )
-        return {"nodes": [info.model_dump() for info in skills_info]}

api/controllers/console/app/workflow.py

@@ -20,7 +20,6 @@ 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
@@ -31,12 +30,12 @@ from core.trigger.debug.event_selectors import (
     select_trigger_debug_events,
 )
 from core.workflow.enums import NodeType
+from core.workflow.file.models import File
 from core.workflow.graph_engine.manager import GraphEngineManager
 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.online_user_fields import online_user_list_fields
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
 from libs import helper
 from libs.datetime_utils import naive_utc_now
@@ -45,12 +44,9 @@ from libs.login import current_account_with_tenant, login_required
 from models import App
 from models.model import AppMode
 from models.workflow import Workflow
-from repositories.workflow_collaboration_repository import WORKFLOW_ONLINE_USERS_PREFIX
 from services.app_generate_service import AppGenerateService
 from services.errors.app import WorkflowHashNotEqualError
 from services.errors.llm import InvokeRateLimitError
-from services.workflow.entities import NestedNodeGraphRequest, NestedNodeParameterSchema
-from services.workflow.nested_node_graph_service import NestedNodeGraphService
 from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService
 
 logger = logging.getLogger(__name__)
@@ -165,14 +161,6 @@ class WorkflowUpdatePayload(BaseModel):
     marked_comment: str | None = Field(default=None, max_length=100)
 
 
-class WorkflowFeaturesPayload(BaseModel):
-    features: dict[str, Any] = Field(..., description="Workflow feature configuration")
-
-
-class WorkflowOnlineUsersQuery(BaseModel):
-    workflow_ids: str = Field(..., description="Comma-separated workflow IDs")
-
-
 class DraftWorkflowTriggerRunPayload(BaseModel):
     node_id: str
 
@@ -181,15 +169,6 @@ class DraftWorkflowTriggerRunAllPayload(BaseModel):
     node_ids: list[str]
 
 
-class NestedNodeGraphPayload(BaseModel):
-    """Request payload for generating nested node graph."""
-
-    parent_node_id: str = Field(description="ID of the parent node that uses the extracted value")
-    parameter_key: str = Field(description="Key of the parameter being extracted")
-    context_source: list[str] = Field(description="Variable selector for the context source")
-    parameter_schema: dict[str, Any] = Field(description="Schema of the parameter to extract")
-
-
 def reg(cls: type[BaseModel]):
     console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
 
@@ -205,11 +184,8 @@ reg(DefaultBlockConfigQuery)
 reg(ConvertToWorkflowPayload)
 reg(WorkflowListQuery)
 reg(WorkflowUpdatePayload)
-reg(WorkflowFeaturesPayload)
-reg(WorkflowOnlineUsersQuery)
 reg(DraftWorkflowTriggerRunPayload)
 reg(DraftWorkflowTriggerRunAllPayload)
-reg(NestedNodeGraphPayload)
 
 
 # TODO(QuantumGhost): Refactor existing node run API to handle file parameter parsing
@@ -532,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")
@@ -592,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"}
 
@@ -679,14 +828,13 @@ class PublishedWorkflowApi(Resource):
         """
         Publish workflow
         """
-        from services.app_bundle_service import AppBundleService
-
         current_user, _ = current_account_with_tenant()
 
         args = PublishWorkflowPayload.model_validate(console_ns.payload or {})
 
+        workflow_service = WorkflowService()
         with Session(db.engine) as session:
-            workflow = AppBundleService.publish(
+            workflow = workflow_service.publish_workflow(
                 session=session,
                 app_model=app_model,
                 account=current_user,
@@ -797,31 +945,6 @@ class ConvertToWorkflowApi(Resource):
         }
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/features")
-class WorkflowFeaturesApi(Resource):
-    """Update draft workflow features."""
-
-    @console_ns.expect(console_ns.models[WorkflowFeaturesPayload.__name__])
-    @console_ns.doc("update_workflow_features")
-    @console_ns.doc(description="Update draft workflow features")
-    @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Workflow features updated successfully")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        current_user, _ = current_account_with_tenant()
-
-        args = WorkflowFeaturesPayload.model_validate(console_ns.payload or {})
-        features = args.features
-
-        workflow_service = WorkflowService()
-        workflow_service.update_draft_workflow_features(app_model=app_model, features=features, account=current_user)
-
-        return {"result": "success"}
-
-
 @console_ns.route("/apps/<uuid:app_id>/workflows")
 class PublishedAllWorkflowApi(Resource):
     @console_ns.expect(console_ns.models[WorkflowListQuery.__name__])
@@ -1199,83 +1322,3 @@ class DraftWorkflowTriggerRunAllApi(Resource):
                     "status": "error",
                 }
             ), 400
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/nested-node-graph")
-class NestedNodeGraphApi(Resource):
-    """
-    API for generating Nested Node LLM graph structures.
-
-    This endpoint creates a complete graph structure containing an LLM node
-    configured to extract values from list[PromptMessage] variables.
-    """
-
-    @console_ns.doc("generate_nested_node_graph")
-    @console_ns.doc(description="Generate a Nested Node LLM graph structure")
-    @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[NestedNodeGraphPayload.__name__])
-    @console_ns.response(200, "Nested node graph generated successfully")
-    @console_ns.response(400, "Invalid request parameters")
-    @console_ns.response(403, "Permission denied")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    @edit_permission_required
-    def post(self, app_model: App):
-        """
-        Generate a Nested Node LLM graph structure.
-
-        Returns a complete graph structure containing a single LLM node
-        configured for extracting values from list[PromptMessage] context.
-        """
-
-        payload = NestedNodeGraphPayload.model_validate(console_ns.payload or {})
-
-        parameter_schema = NestedNodeParameterSchema(
-            name=payload.parameter_schema.get("name", payload.parameter_key),
-            type=payload.parameter_schema.get("type", "string"),
-            description=payload.parameter_schema.get("description", ""),
-        )
-
-        request = NestedNodeGraphRequest(
-            parent_node_id=payload.parent_node_id,
-            parameter_key=payload.parameter_key,
-            context_source=payload.context_source,
-            parameter_schema=parameter_schema,
-        )
-
-        with Session(db.engine) as session:
-            service = NestedNodeGraphService(session)
-            response = service.generate_nested_node_graph(tenant_id=app_model.tenant_id, request=request)
-
-        return response.model_dump()
-
-
-@console_ns.route("/apps/workflows/online-users")
-class WorkflowOnlineUsersApi(Resource):
-    @console_ns.expect(console_ns.models[WorkflowOnlineUsersQuery.__name__])
-    @console_ns.doc("get_workflow_online_users")
-    @console_ns.doc(description="Get workflow online users")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @marshal_with(online_user_list_fields)
-    def get(self):
-        args = WorkflowOnlineUsersQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
-
-        workflow_ids = [workflow_id.strip() for workflow_id in args.workflow_ids.split(",") if workflow_id.strip()]
-
-        results = []
-        for workflow_id in workflow_ids:
-            users_json = redis_client.hgetall(f"{WORKFLOW_ONLINE_USERS_PREFIX}{workflow_id}")
-
-            users = []
-            for _, user_info_json in users_json.items():
-                try:
-                    users.append(json.loads(user_info_json))
-                except Exception:
-                    continue
-            results.append({"workflow_id": workflow_id, "users": users})
-
-        return {"data": results}

api/controllers/console/app/workflow_comment.py

@@ -1,317 +0,0 @@
-import logging
-
-from flask_restx import Resource, fields, marshal_with
-from pydantic import BaseModel, Field
-
-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 fields.member_fields import account_with_role_fields
-from fields.workflow_comment_fields import (
-    workflow_comment_basic_fields,
-    workflow_comment_create_fields,
-    workflow_comment_detail_fields,
-    workflow_comment_reply_create_fields,
-    workflow_comment_reply_update_fields,
-    workflow_comment_resolve_fields,
-    workflow_comment_update_fields,
-)
-from libs.login import current_user, login_required
-from models import App
-from services.account_service import TenantService
-from services.workflow_comment_service import WorkflowCommentService
-
-logger = logging.getLogger(__name__)
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-class WorkflowCommentCreatePayload(BaseModel):
-    position_x: float = Field(..., description="Comment X position")
-    position_y: float = Field(..., description="Comment Y position")
-    content: str = Field(..., description="Comment content")
-    mentioned_user_ids: list[str] = Field(default_factory=list, description="Mentioned user IDs")
-
-
-class WorkflowCommentUpdatePayload(BaseModel):
-    content: str = Field(..., description="Comment content")
-    position_x: float | None = Field(default=None, description="Comment X position")
-    position_y: float | None = Field(default=None, description="Comment Y position")
-    mentioned_user_ids: list[str] = Field(default_factory=list, description="Mentioned user IDs")
-
-
-class WorkflowCommentReplyCreatePayload(BaseModel):
-    content: str = Field(..., description="Reply content")
-    mentioned_user_ids: list[str] = Field(default_factory=list, description="Mentioned user IDs")
-
-
-class WorkflowCommentReplyUpdatePayload(BaseModel):
-    content: str = Field(..., description="Reply content")
-    mentioned_user_ids: list[str] = Field(default_factory=list, description="Mentioned user IDs")
-
-
-for model in (
-    WorkflowCommentCreatePayload,
-    WorkflowCommentUpdatePayload,
-    WorkflowCommentReplyCreatePayload,
-    WorkflowCommentReplyUpdatePayload,
-):
-    console_ns.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-workflow_comment_basic_model = console_ns.model("WorkflowCommentBasic", workflow_comment_basic_fields)
-workflow_comment_detail_model = console_ns.model("WorkflowCommentDetail", workflow_comment_detail_fields)
-workflow_comment_create_model = console_ns.model("WorkflowCommentCreate", workflow_comment_create_fields)
-workflow_comment_update_model = console_ns.model("WorkflowCommentUpdate", workflow_comment_update_fields)
-workflow_comment_resolve_model = console_ns.model("WorkflowCommentResolve", workflow_comment_resolve_fields)
-workflow_comment_reply_create_model = console_ns.model(
-    "WorkflowCommentReplyCreate", workflow_comment_reply_create_fields
-)
-workflow_comment_reply_update_model = console_ns.model(
-    "WorkflowCommentReplyUpdate", workflow_comment_reply_update_fields
-)
-workflow_comment_mention_users_model = console_ns.model(
-    "WorkflowCommentMentionUsers",
-    {"users": fields.List(fields.Nested(account_with_role_fields))},
-)
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflow/comments")
-class WorkflowCommentListApi(Resource):
-    """API for listing and creating workflow comments."""
-
-    @console_ns.doc("list_workflow_comments")
-    @console_ns.doc(description="Get all comments for a workflow")
-    @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Comments retrieved successfully", workflow_comment_basic_model)
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model()
-    @marshal_with(workflow_comment_basic_model, envelope="data")
-    def get(self, app_model: App):
-        """Get all comments for a workflow."""
-        comments = WorkflowCommentService.get_comments(tenant_id=current_user.current_tenant_id, app_id=app_model.id)
-
-        return comments
-
-    @console_ns.doc("create_workflow_comment")
-    @console_ns.doc(description="Create a new workflow comment")
-    @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[WorkflowCommentCreatePayload.__name__])
-    @console_ns.response(201, "Comment created successfully", workflow_comment_create_model)
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model()
-    @marshal_with(workflow_comment_create_model)
-    def post(self, app_model: App):
-        """Create a new workflow comment."""
-        payload = WorkflowCommentCreatePayload.model_validate(console_ns.payload or {})
-
-        result = WorkflowCommentService.create_comment(
-            tenant_id=current_user.current_tenant_id,
-            app_id=app_model.id,
-            created_by=current_user.id,
-            content=payload.content,
-            position_x=payload.position_x,
-            position_y=payload.position_y,
-            mentioned_user_ids=payload.mentioned_user_ids,
-        )
-
-        return result, 201
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>")
-class WorkflowCommentDetailApi(Resource):
-    """API for managing individual workflow comments."""
-
-    @console_ns.doc("get_workflow_comment")
-    @console_ns.doc(description="Get a specific workflow comment")
-    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
-    @console_ns.response(200, "Comment retrieved successfully", workflow_comment_detail_model)
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model()
-    @marshal_with(workflow_comment_detail_model)
-    def get(self, app_model: App, comment_id: str):
-        """Get a specific workflow comment."""
-        comment = WorkflowCommentService.get_comment(
-            tenant_id=current_user.current_tenant_id, app_id=app_model.id, comment_id=comment_id
-        )
-
-        return comment
-
-    @console_ns.doc("update_workflow_comment")
-    @console_ns.doc(description="Update a workflow comment")
-    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
-    @console_ns.expect(console_ns.models[WorkflowCommentUpdatePayload.__name__])
-    @console_ns.response(200, "Comment updated successfully", workflow_comment_update_model)
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model()
-    @marshal_with(workflow_comment_update_model)
-    def put(self, app_model: App, comment_id: str):
-        """Update a workflow comment."""
-        payload = WorkflowCommentUpdatePayload.model_validate(console_ns.payload or {})
-
-        result = WorkflowCommentService.update_comment(
-            tenant_id=current_user.current_tenant_id,
-            app_id=app_model.id,
-            comment_id=comment_id,
-            user_id=current_user.id,
-            content=payload.content,
-            position_x=payload.position_x,
-            position_y=payload.position_y,
-            mentioned_user_ids=payload.mentioned_user_ids,
-        )
-
-        return result
-
-    @console_ns.doc("delete_workflow_comment")
-    @console_ns.doc(description="Delete a workflow comment")
-    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
-    @console_ns.response(204, "Comment deleted successfully")
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model()
-    def delete(self, app_model: App, comment_id: str):
-        """Delete a workflow comment."""
-        WorkflowCommentService.delete_comment(
-            tenant_id=current_user.current_tenant_id,
-            app_id=app_model.id,
-            comment_id=comment_id,
-            user_id=current_user.id,
-        )
-
-        return {"result": "success"}, 204
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/resolve")
-class WorkflowCommentResolveApi(Resource):
-    """API for resolving and reopening workflow comments."""
-
-    @console_ns.doc("resolve_workflow_comment")
-    @console_ns.doc(description="Resolve a workflow comment")
-    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
-    @console_ns.response(200, "Comment resolved successfully", workflow_comment_resolve_model)
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model()
-    @marshal_with(workflow_comment_resolve_model)
-    def post(self, app_model: App, comment_id: str):
-        """Resolve a workflow comment."""
-        comment = WorkflowCommentService.resolve_comment(
-            tenant_id=current_user.current_tenant_id,
-            app_id=app_model.id,
-            comment_id=comment_id,
-            user_id=current_user.id,
-        )
-
-        return comment
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies")
-class WorkflowCommentReplyApi(Resource):
-    """API for managing comment replies."""
-
-    @console_ns.doc("create_workflow_comment_reply")
-    @console_ns.doc(description="Add a reply to a workflow comment")
-    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
-    @console_ns.expect(console_ns.models[WorkflowCommentReplyCreatePayload.__name__])
-    @console_ns.response(201, "Reply created successfully", workflow_comment_reply_create_model)
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model()
-    @marshal_with(workflow_comment_reply_create_model)
-    def post(self, app_model: App, comment_id: str):
-        """Add a reply to a workflow comment."""
-        # Validate comment access first
-        WorkflowCommentService.validate_comment_access(
-            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
-        )
-
-        payload = WorkflowCommentReplyCreatePayload.model_validate(console_ns.payload or {})
-
-        result = WorkflowCommentService.create_reply(
-            comment_id=comment_id,
-            content=payload.content,
-            created_by=current_user.id,
-            mentioned_user_ids=payload.mentioned_user_ids,
-        )
-
-        return result, 201
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies/<string:reply_id>")
-class WorkflowCommentReplyDetailApi(Resource):
-    """API for managing individual comment replies."""
-
-    @console_ns.doc("update_workflow_comment_reply")
-    @console_ns.doc(description="Update a comment reply")
-    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID", "reply_id": "Reply ID"})
-    @console_ns.expect(console_ns.models[WorkflowCommentReplyUpdatePayload.__name__])
-    @console_ns.response(200, "Reply updated successfully", workflow_comment_reply_update_model)
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model()
-    @marshal_with(workflow_comment_reply_update_model)
-    def put(self, app_model: App, comment_id: str, reply_id: str):
-        """Update a comment reply."""
-        # Validate comment access first
-        WorkflowCommentService.validate_comment_access(
-            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
-        )
-
-        payload = WorkflowCommentReplyUpdatePayload.model_validate(console_ns.payload or {})
-
-        reply = WorkflowCommentService.update_reply(
-            reply_id=reply_id,
-            user_id=current_user.id,
-            content=payload.content,
-            mentioned_user_ids=payload.mentioned_user_ids,
-        )
-
-        return reply
-
-    @console_ns.doc("delete_workflow_comment_reply")
-    @console_ns.doc(description="Delete a comment reply")
-    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID", "reply_id": "Reply ID"})
-    @console_ns.response(204, "Reply deleted successfully")
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model()
-    def delete(self, app_model: App, comment_id: str, reply_id: str):
-        """Delete a comment reply."""
-        # Validate comment access first
-        WorkflowCommentService.validate_comment_access(
-            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
-        )
-
-        WorkflowCommentService.delete_reply(reply_id=reply_id, user_id=current_user.id)
-
-        return {"result": "success"}, 204
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflow/comments/mention-users")
-class WorkflowCommentMentionUsersApi(Resource):
-    """API for getting mentionable users for workflow comments."""
-
-    @console_ns.doc("workflow_comment_mention_users")
-    @console_ns.doc(description="Get all users in current tenant for mentions")
-    @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Mentionable users retrieved successfully", workflow_comment_mention_users_model)
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model()
-    @marshal_with(workflow_comment_mention_users_model)
-    def get(self, app_model: App):
-        """Get all users in current tenant for mentions."""
-        members = TenantService.get_tenant_members(current_user.current_tenant)
-        return {"users": members}

api/controllers/console/app/workflow_draft_variable.py

@@ -15,18 +15,17 @@ 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, ArrayPromptMessageSegment, FileSegment, Segment
+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 core.workflow.file import helpers as file_helpers
 from extensions.ext_database import db
-from factories import variable_factory
 from factories.file_factory import build_from_mapping, build_from_mappings
-from libs.login import current_account_with_tenant, login_required
+from factories.variable_factory import build_segment_with_type
+from libs.login import login_required
 from models import App, AppMode
 from models.workflow import WorkflowDraftVariable
-from services.sandbox.sandbox_service import SandboxService
 from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
 from services.workflow_service import WorkflowService
 
@@ -44,16 +43,6 @@ class WorkflowDraftVariableUpdatePayload(BaseModel):
     value: Any | None = Field(default=None, description="Variable value")
 
 
-class ConversationVariableUpdatePayload(BaseModel):
-    conversation_variables: list[dict[str, Any]] = Field(
-        ..., description="Conversation variables for the draft workflow"
-    )
-
-
-class EnvironmentVariableUpdatePayload(BaseModel):
-    environment_variables: list[dict[str, Any]] = Field(..., description="Environment variables for the draft workflow")
-
-
 console_ns.schema_model(
     WorkflowDraftVariableListQuery.__name__,
     WorkflowDraftVariableListQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
@@ -62,14 +51,6 @@ console_ns.schema_model(
     WorkflowDraftVariableUpdatePayload.__name__,
     WorkflowDraftVariableUpdatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
 )
-console_ns.schema_model(
-    ConversationVariableUpdatePayload.__name__,
-    ConversationVariableUpdatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-console_ns.schema_model(
-    EnvironmentVariableUpdatePayload.__name__,
-    EnvironmentVariableUpdatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
 
 
 def _convert_values_to_json_serializable_object(value: Segment):
@@ -77,8 +58,6 @@ def _convert_values_to_json_serializable_object(value: Segment):
         return value.value.model_dump()
     elif isinstance(value, ArrayFileSegment):
         return [i.model_dump() for i in value.value]
-    elif isinstance(value, ArrayPromptMessageSegment):
-        return value.to_object()
     elif isinstance(value, SegmentGroup):
         return [_convert_values_to_json_serializable_object(i) for i in value.value]
     else:
@@ -133,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,
@@ -268,8 +247,6 @@ class WorkflowVariableCollectionApi(Resource):
     @console_ns.response(204, "Workflow variables deleted successfully")
     @_api_prerequisite
     def delete(self, app_model: App):
-        current_user, _ = current_account_with_tenant()
-        SandboxService.delete_draft_storage(app_model.tenant_id, current_user.id)
         draft_var_srv = WorkflowDraftVariableService(
             session=db.session(),
         )
@@ -406,7 +383,7 @@ class VariableApi(Resource):
                 if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
                     raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
                 raw_value = build_from_mappings(mappings=raw_value, tenant_id=app_model.tenant_id)
-            new_value = variable_factory.build_segment_with_type(variable.value_type, raw_value)
+            new_value = build_segment_with_type(variable.value_type, raw_value)
         draft_var_srv.update_variable(variable, name=new_name, value=new_value)
         db.session.commit()
         return variable
@@ -499,34 +476,6 @@ class ConversationVariableCollectionApi(Resource):
         db.session.commit()
         return _get_variable_list(app_model, CONVERSATION_VARIABLE_NODE_ID)
 
-    @console_ns.expect(console_ns.models[ConversationVariableUpdatePayload.__name__])
-    @console_ns.doc("update_conversation_variables")
-    @console_ns.doc(description="Update conversation variables for workflow draft")
-    @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Conversation variables updated successfully")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @edit_permission_required
-    @get_app_model(mode=AppMode.ADVANCED_CHAT)
-    def post(self, app_model: App):
-        payload = ConversationVariableUpdatePayload.model_validate(console_ns.payload or {})
-
-        workflow_service = WorkflowService()
-
-        conversation_variables_list = payload.conversation_variables
-        conversation_variables = [
-            variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
-        ]
-
-        workflow_service.update_draft_workflow_conversation_variables(
-            app_model=app_model,
-            account=current_user,
-            conversation_variables=conversation_variables,
-        )
-
-        return {"result": "success"}
-
 
 @console_ns.route("/apps/<uuid:app_id>/workflows/draft/system-variables")
 class SystemVariableCollectionApi(Resource):
@@ -578,31 +527,3 @@ class EnvironmentVariableCollectionApi(Resource):
             )
 
         return {"items": env_vars_list}
-
-    @console_ns.expect(console_ns.models[EnvironmentVariableUpdatePayload.__name__])
-    @console_ns.doc("update_environment_variables")
-    @console_ns.doc(description="Update environment variables for workflow draft")
-    @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Environment variables updated successfully")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @edit_permission_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        payload = EnvironmentVariableUpdatePayload.model_validate(console_ns.payload or {})
-
-        workflow_service = WorkflowService()
-
-        environment_variables_list = payload.environment_variables
-        environment_variables = [
-            variable_factory.build_environment_variable_from_mapping(obj) for obj in environment_variables_list
-        ]
-
-        workflow_service.update_draft_workflow_environment_variables(
-            app_model=app_model,
-            account=current_user,
-            environment_variables=environment_variables,
-        )
-
-        return {"result": "success"}

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 core.workflow.entities.pause_reason import HumanInputRequired
+from core.workflow.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

@@ -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

@@ -55,6 +55,7 @@ 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.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
@@ -820,6 +821,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

@@ -576,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(
@@ -954,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
 
@@ -1339,6 +1339,18 @@ class DocumentGenerateSummaryApi(Resource):
             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

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/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
@@ -38,7 +37,7 @@ from core.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/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,
@@ -43,6 +44,7 @@ from core.errors.error import (
 from core.model_runtime.errors.invoke import InvokeError
 from core.workflow.graph_engine.manager import GraphEngineManager
 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

@@ -23,6 +23,7 @@ from core.errors.error import (
 )
 from core.model_runtime.errors.invoke import InvokeError
 from core.workflow.graph_engine.manager import GraphEngineManager
+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 core.workflow.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/sandbox_files.py

@@ -1,86 +0,0 @@
-from __future__ import annotations
-
-from flask import request
-from flask_restx import Resource, fields
-from pydantic import BaseModel, Field
-
-from controllers.console import console_ns
-from controllers.console.wraps import account_initialization_required, setup_required
-from libs.login import current_account_with_tenant, login_required
-from services.sandbox.sandbox_file_service import SandboxFileService
-
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-class SandboxFileListQuery(BaseModel):
-    path: str | None = Field(default=None, description="Workspace relative path")
-    recursive: bool = Field(default=False, description="List recursively")
-
-
-class SandboxFileDownloadRequest(BaseModel):
-    path: str = Field(..., description="Workspace relative file path")
-
-
-console_ns.schema_model(
-    SandboxFileListQuery.__name__,
-    SandboxFileListQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-console_ns.schema_model(
-    SandboxFileDownloadRequest.__name__,
-    SandboxFileDownloadRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-
-
-SANDBOX_FILE_NODE_FIELDS = {
-    "path": fields.String,
-    "is_dir": fields.Boolean,
-    "size": fields.Raw,
-    "mtime": fields.Raw,
-    "extension": fields.String,
-}
-
-
-SANDBOX_FILE_DOWNLOAD_TICKET_FIELDS = {
-    "download_url": fields.String,
-    "expires_in": fields.Integer,
-    "export_id": fields.String,
-}
-
-
-sandbox_file_node_model = console_ns.model("SandboxFileNode", SANDBOX_FILE_NODE_FIELDS)
-sandbox_file_download_ticket_model = console_ns.model("SandboxFileDownloadTicket", SANDBOX_FILE_DOWNLOAD_TICKET_FIELDS)
-
-
-@console_ns.route("/sandboxes/<string:sandbox_id>/files")
-class SandboxFilesApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @console_ns.expect(console_ns.models[SandboxFileListQuery.__name__])
-    @console_ns.marshal_list_with(sandbox_file_node_model)
-    def get(self, sandbox_id: str):
-        args = SandboxFileListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore[arg-type]
-        _, tenant_id = current_account_with_tenant()
-        return [
-            e.__dict__
-            for e in SandboxFileService.list_files(
-                tenant_id=tenant_id,
-                sandbox_id=sandbox_id,
-                path=args.path,
-                recursive=args.recursive,
-            )
-        ]
-
-
-@console_ns.route("/sandboxes/<string:sandbox_id>/files/download")
-class SandboxFileDownloadApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @console_ns.expect(console_ns.models[SandboxFileDownloadRequest.__name__])
-    @console_ns.marshal_with(sandbox_file_download_ticket_model)
-    def post(self, sandbox_id: str):
-        payload = SandboxFileDownloadRequest.model_validate(console_ns.payload or {})
-        _, tenant_id = current_account_with_tenant()
-        res = SandboxFileService.download_file(tenant_id=tenant_id, sandbox_id=sandbox_id, path=payload.path)
-        return res.__dict__

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()