try

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

@@ -1,168 +0,0 @@
----
-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

@@ -1,91 +0,0 @@
-# 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

@@ -1,157 +0,0 @@
-# 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

@@ -1,61 +0,0 @@
-# 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

@@ -1,139 +0,0 @@
-# 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

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

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

@@ -77,8 +77,8 @@ jobs:
             }
 
             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.';
+              ? `### Pyrefly Diff (base → PR)\\n\\`\\`\\`diff\\n${diff}\\n\\`\\`\\``
+              : '### Pyrefly Diff\\nNo changes detected.';
 
             await github.rest.issues.createComment({
               issue_number: prNumber,

.github/workflows/pyrefly-diff.yml

@@ -74,16 +74,7 @@ jobs:
             }
 
             const body = diff.trim()
-              ? [
-                  '### Pyrefly Diff',
-                  '<details>',
-                  '<summary>base → PR</summary>',
-                  '',
-                  '```diff',
-                  diff,
-                  '```',
-                  '</details>',
-                ].join('\n')
+              ? `### Pyrefly Diff (base → PR)\n\`\`\`diff\n${diff}\n\`\`\``
               : '### Pyrefly Diff\nNo changes detected.';
 
             await github.rest.issues.createComment({

.github/workflows/web-tests.yml

@@ -3,22 +3,14 @@ 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 (${{ matrix.shardIndex }}/${{ matrix.shardTotal }})
+    name: Web Tests
     runs-on: ubuntu-latest
-    strategy:
-      fail-fast: false
-      matrix:
-        shardIndex: [1, 2, 3, 4]
-        shardTotal: [4]
     defaults:
       run:
         shell: bash
@@ -47,58 +39,7 @@ jobs:
         run: pnpm install --frozen-lockfile
 
       - name: Run tests
-        run: pnpm vitest run --reporter=blob --shard=${{ matrix.shardIndex }}/${{ matrix.shardTotal }} --coverage
-
-      - name: Upload blob report
-        if: ${{ !cancelled() }}
-        uses: actions/upload-artifact@v6
-        with:
-          name: blob-report-${{ matrix.shardIndex }}
-          path: web/.vitest-reports/*
-          include-hidden-files: true
-          retention-days: 1
-
-  merge-reports:
-    name: Merge Test Reports
-    if: ${{ !cancelled() }}
-    needs: [test]
-    runs-on: ubuntu-latest
-    defaults:
-      run:
-        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
+        run: pnpm test:ci
 
       - name: Coverage Summary
         if: always()

Makefile

@@ -68,9 +68,10 @@ lint:
 	@echo "✅ Linting complete"
 
 type-check:
-	@echo "📝 Running type checks (basedpyright + mypy)..."
+	@echo "📝 Running type checks (basedpyright + mypy + ty)..."
 	@./dev/basedpyright-check $(PATH_TO_CHECK)
 	@uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --check-untyped-defs --disable-error-code=import-untyped .
+	@cd api && uv run ty check
 	@echo "✅ Type checks complete"
 
 test:
@@ -131,7 +132,7 @@ help:
 	@echo "  make format         - Format code with ruff"
 	@echo "  make check          - Check code with ruff"
 	@echo "  make lint           - Format, fix, and lint code (ruff, imports, dotenv)"
-	@echo "  make type-check     - Run type checks (basedpyright, mypy)"
+	@echo "  make type-check     - Run type checks (basedpyright, mypy, ty)"
 	@echo "  make test           - Run backend unit tests (or TARGET_TESTS=./api/tests/<target_tests>)"
 	@echo ""
 	@echo "Docker Build Targets:"

README.md

@@ -1,5 +1,9 @@
 ![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/.importlinter

@@ -50,11 +50,14 @@ 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.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
     # TODO(QuantumGhost): use DI to avoid depending on global DB.
     core.workflow.nodes.human_input.human_input_node -> extensions.ext_database
 
@@ -88,10 +91,11 @@ forbidden_modules =
     core.logging
     core.mcp
     core.memory
+    core.model_manager
     core.moderation
     core.ops
     core.plugin
-    core.model_runtime.prompt
+    core.prompt
     core.provider_manager
     core.rag
     core.repositories
@@ -101,16 +105,22 @@ 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 -> core.helper.ssrf_proxy
+    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.llm.llm_utils -> configs
+    core.workflow.nodes.llm.llm_utils -> core.app.entities.app_invoke_entities
     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
@@ -127,37 +137,52 @@ ignore_imports =
     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.parameter_extractor.parameter_extractor_node -> core.model_runtime.prompt.advanced_prompt_transform
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_runtime.prompt.simple_prompt_transform
+    core.workflow.nodes.llm.node -> core.app.entities.app_invoke_entities
+    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.app.entities.app_invoke_entities
+    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.advanced_prompt_transform
+    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.simple_prompt_transform
     core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_runtime.model_providers.__base.large_language_model
-    core.workflow.nodes.question_classifier.question_classifier_node -> core.model_runtime.prompt.simple_prompt_transform
+    core.workflow.nodes.question_classifier.question_classifier_node -> core.app.entities.app_invoke_entities
+    core.workflow.nodes.question_classifier.question_classifier_node -> core.prompt.advanced_prompt_transform
+    core.workflow.nodes.question_classifier.question_classifier_node -> core.prompt.simple_prompt_transform
     core.workflow.nodes.start.entities -> core.app.app_config.entities
     core.workflow.nodes.start.start_node -> core.app.app_config.entities
     core.workflow.workflow_entry -> core.app.apps.exc
     core.workflow.workflow_entry -> core.app.entities.app_invoke_entities
     core.workflow.workflow_entry -> core.app.workflow.node_factory
+    core.workflow.nodes.datasource.datasource_node -> core.datasource.datasource_manager
+    core.workflow.nodes.datasource.datasource_node -> core.datasource.utils.message_transformer
     core.workflow.nodes.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.nodes.llm.llm_utils -> core.variables.segments
+    core.workflow.nodes.loop.entities -> core.variables.types
     core.workflow.nodes.tool.tool_node -> core.tools.utils.message_transformer
     core.workflow.nodes.tool.tool_node -> models
     core.workflow.nodes.agent.agent_node -> models.model
+    core.workflow.nodes.code.code_node -> core.helper.code_executor.code_node_provider
+    core.workflow.nodes.code.code_node -> core.helper.code_executor.javascript.javascript_code_provider
+    core.workflow.nodes.code.code_node -> core.helper.code_executor.python3.python3_code_provider
+    core.workflow.nodes.code.entities -> core.helper.code_executor.code_executor
+    core.workflow.nodes.datasource.datasource_node -> core.variables.variables
+    core.workflow.nodes.http_request.executor -> core.helper.ssrf_proxy
+    core.workflow.nodes.http_request.node -> core.helper.ssrf_proxy
     core.workflow.nodes.llm.file_saver -> core.helper.ssrf_proxy
     core.workflow.nodes.llm.node -> core.helper.code_executor
     core.workflow.nodes.template_transform.template_renderer -> core.helper.code_executor.code_executor
     core.workflow.nodes.llm.node -> core.llm_generator.output_parser.errors
     core.workflow.nodes.llm.node -> core.llm_generator.output_parser.structured_output
     core.workflow.nodes.llm.node -> core.model_manager
-    core.workflow.nodes.agent.entities -> core.model_runtime.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.llm.entities -> core.model_runtime.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.llm.llm_utils -> core.model_runtime.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.llm.node -> core.model_runtime.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.llm.node -> core.model_runtime.prompt.utils.prompt_message_util
-    core.workflow.nodes.parameter_extractor.entities -> core.model_runtime.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_runtime.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_runtime.prompt.utils.prompt_message_util
-    core.workflow.nodes.question_classifier.entities -> core.model_runtime.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.question_classifier.question_classifier_node -> core.model_runtime.prompt.utils.prompt_message_util
+    core.workflow.nodes.agent.entities -> core.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.llm.entities -> core.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.llm.llm_utils -> core.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.llm.node -> core.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.llm.node -> core.prompt.utils.prompt_message_util
+    core.workflow.nodes.parameter_extractor.entities -> core.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.utils.prompt_message_util
+    core.workflow.nodes.question_classifier.entities -> core.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.question_classifier.question_classifier_node -> core.prompt.utils.prompt_message_util
     core.workflow.nodes.knowledge_index.entities -> core.rag.retrieval.retrieval_methods
     core.workflow.nodes.knowledge_index.knowledge_index_node -> core.rag.retrieval.retrieval_methods
     core.workflow.nodes.knowledge_index.knowledge_index_node -> models.dataset
@@ -169,7 +194,58 @@ ignore_imports =
     core.workflow.nodes.llm.file_saver -> core.tools.signature
     core.workflow.nodes.llm.file_saver -> core.tools.tool_file_manager
     core.workflow.nodes.tool.tool_node -> core.tools.errors
+    core.workflow.conversation_variable_updater -> core.variables
+    core.workflow.graph_engine.entities.commands -> core.variables.variables
+    core.workflow.nodes.agent.agent_node -> core.variables.segments
+    core.workflow.nodes.answer.answer_node -> core.variables
+    core.workflow.nodes.code.code_node -> core.variables.segments
+    core.workflow.nodes.code.code_node -> core.variables.types
+    core.workflow.nodes.code.entities -> core.variables.types
+    core.workflow.nodes.datasource.datasource_node -> core.variables.segments
+    core.workflow.nodes.document_extractor.node -> core.variables
+    core.workflow.nodes.document_extractor.node -> core.variables.segments
+    core.workflow.nodes.http_request.executor -> core.variables.segments
+    core.workflow.nodes.http_request.node -> core.variables.segments
+    core.workflow.nodes.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
+    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.variables
+    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.variables.segments
+    core.workflow.nodes.list_operator.node -> core.variables
+    core.workflow.nodes.list_operator.node -> core.variables.segments
+    core.workflow.nodes.llm.node -> core.variables
+    core.workflow.nodes.loop.loop_node -> core.variables
+    core.workflow.nodes.parameter_extractor.entities -> core.variables.types
+    core.workflow.nodes.parameter_extractor.exc -> core.variables.types
+    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.variables.types
+    core.workflow.nodes.tool.tool_node -> core.variables.segments
+    core.workflow.nodes.tool.tool_node -> core.variables.variables
+    core.workflow.nodes.trigger_webhook.node -> core.variables.types
+    core.workflow.nodes.trigger_webhook.node -> core.variables.variables
+    core.workflow.nodes.variable_aggregator.entities -> core.variables.types
+    core.workflow.nodes.variable_aggregator.variable_aggregator_node -> core.variables.segments
+    core.workflow.nodes.variable_assigner.common.helpers -> core.variables
+    core.workflow.nodes.variable_assigner.common.helpers -> core.variables.consts
+    core.workflow.nodes.variable_assigner.common.helpers -> core.variables.types
+    core.workflow.nodes.variable_assigner.v1.node -> core.variables
+    core.workflow.nodes.variable_assigner.v2.helpers -> core.variables
+    core.workflow.nodes.variable_assigner.v2.node -> core.variables
+    core.workflow.nodes.variable_assigner.v2.node -> core.variables.consts
+    core.workflow.runtime.graph_runtime_state_protocol -> core.variables.segments
+    core.workflow.runtime.read_only_wrappers -> core.variables.segments
+    core.workflow.runtime.variable_pool -> core.variables
+    core.workflow.runtime.variable_pool -> core.variables.consts
+    core.workflow.runtime.variable_pool -> core.variables.segments
+    core.workflow.runtime.variable_pool -> core.variables.variables
+    core.workflow.utils.condition.processor -> core.variables
+    core.workflow.utils.condition.processor -> core.variables.segments
+    core.workflow.variable_loader -> core.variables
+    core.workflow.variable_loader -> core.variables.consts
+    core.workflow.workflow_type_encoder -> core.variables
+    core.workflow.graph_engine.manager -> extensions.ext_redis
     core.workflow.nodes.agent.agent_node -> extensions.ext_database
+    core.workflow.nodes.datasource.datasource_node -> extensions.ext_database
     core.workflow.nodes.knowledge_index.knowledge_index_node -> extensions.ext_database
     core.workflow.nodes.llm.file_saver -> extensions.ext_database
     core.workflow.nodes.llm.llm_utils -> extensions.ext_database
@@ -190,12 +266,7 @@ ignore_imports =
 name = Model Runtime Internal Imports
 type = forbidden
 source_modules =
-    core.model_runtime.callbacks
-    core.model_runtime.entities
-    core.model_runtime.errors
-    core.model_runtime.model_providers
-    core.model_runtime.schema_validators
-    core.model_runtime.utils
+    core.model_runtime
 forbidden_modules =
     configs
     controllers
@@ -225,7 +296,7 @@ forbidden_modules =
     core.moderation
     core.ops
     core.plugin
-    core.model_runtime.prompt
+    core.prompt
     core.provider_manager
     core.rag
     core.repositories

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. Start the worker service (async and scheduler tasks, runs from `api`).
+1. Optional: start the worker service (async tasks, runs from `api`).
 
    ```bash
    ./dev/start-worker
@@ -54,6 +54,86 @@ 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 api_token,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/controllers/console/app/workflow.py

@@ -33,7 +33,6 @@ 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.workflow_fields import workflow_fields, workflow_pagination_fields
@@ -741,7 +740,7 @@ class WorkflowTaskStopApi(Resource):
         AppQueueManager.set_stop_flag_no_user_check(task_id)
 
         # New graph engine command channel mechanism
-        GraphEngineManager(redis_client).send_stop_command(task_id)
+        GraphEngineManager.send_stop_command(task_id)
 
         return {"result": "success"}
 

api/controllers/console/app/workflow_draft_variable.py

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

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

@@ -21,8 +21,8 @@ from controllers.console.app.workflow_draft_variable import (
 from controllers.console.datasets.wraps import get_rag_pipeline
 from controllers.console.wraps import account_initialization_required, setup_required
 from controllers.web.error import InvalidArgumentError, NotFoundError
+from core.variables.types import SegmentType
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
-from core.workflow.variables.types import SegmentType
 from extensions.ext_database import db
 from factories.file_factory import build_from_mapping, build_from_mappings
 from factories.variable_factory import build_segment_with_type

api/controllers/console/explore/trial.py

@@ -44,7 +44,6 @@ 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,
@@ -226,7 +225,7 @@ class TrialAppWorkflowTaskStopApi(TrialAppResource):
         AppQueueManager.set_stop_flag_no_user_check(task_id)
 
         # New graph engine command channel mechanism
-        GraphEngineManager(redis_client).send_stop_command(task_id)
+        GraphEngineManager.send_stop_command(task_id)
 
         return {"result": "success"}
 

api/controllers/console/explore/workflow.py

@@ -23,7 +23,6 @@ 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
@@ -101,6 +100,6 @@ class InstalledAppWorkflowTaskStopApi(InstalledAppResource):
         AppQueueManager.set_stop_flag_no_user_check(task_id)
 
         # New graph engine command channel mechanism
-        GraphEngineManager(redis_client).send_stop_command(task_id)
+        GraphEngineManager.send_stop_command(task_id)
 
         return {"result": "success"}

api/controllers/console/wraps.py

@@ -36,9 +36,9 @@ ERROR_MSG_INVALID_ENCRYPTED_DATA = "Invalid encrypted data"
 ERROR_MSG_INVALID_ENCRYPTED_CODE = "Invalid encrypted code"
 
 
-def account_initialization_required(view: Callable[P, R]) -> Callable[P, R]:
+def account_initialization_required(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         # check account initialization
         current_user, _ = current_account_with_tenant()
         if current_user.status == AccountStatus.UNINITIALIZED:
@@ -214,9 +214,9 @@ def cloud_utm_record(view: Callable[P, R]):
     return decorated
 
 
-def setup_required(view: Callable[P, R]) -> Callable[P, R]:
+def setup_required(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         # check setup
         if (
             dify_config.EDITION == "SELF_HOSTED"

api/controllers/files/image_preview.py

@@ -137,7 +137,7 @@ class FilePreviewApi(Resource):
         if args.as_attachment:
             encoded_filename = quote(upload_file.name)
             response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
-        response.headers["Content-Type"] = "application/octet-stream"
+            response.headers["Content-Type"] = "application/octet-stream"
 
         enforce_download_for_html(
             response,

api/controllers/service_api/app/workflow.py

@@ -31,7 +31,6 @@ from core.model_runtime.errors.invoke import InvokeError
 from core.workflow.enums import WorkflowExecutionStatus
 from core.workflow.graph_engine.manager import GraphEngineManager
 from extensions.ext_database import db
-from extensions.ext_redis import redis_client
 from fields.workflow_app_log_fields import build_workflow_app_log_pagination_model
 from libs import helper
 from libs.helper import OptionalTimestampField, TimestampField
@@ -281,7 +280,7 @@ class WorkflowTaskStopApi(Resource):
         AppQueueManager.set_stop_flag_no_user_check(task_id)
 
         # New graph engine command channel mechanism
-        GraphEngineManager(redis_client).send_stop_command(task_id)
+        GraphEngineManager.send_stop_command(task_id)
 
         return {"result": "success"}
 

api/controllers/web/workflow.py

@@ -24,7 +24,6 @@ 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 models.model import App, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
@@ -122,6 +121,6 @@ class WorkflowTaskStopApi(WebApiResource):
         AppQueueManager.set_stop_flag_no_user_check(task_id)
 
         # New graph engine command channel mechanism
-        GraphEngineManager(redis_client).send_stop_command(task_id)
+        GraphEngineManager.send_stop_command(task_id)
 
         return {"result": "success"}

api/core/agent/base_agent_runner.py

@@ -32,7 +32,7 @@ from core.model_runtime.entities import (
 from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
 from core.model_runtime.entities.model_entities import ModelFeature
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from core.model_runtime.prompt.utils.extract_thread_messages import extract_thread_messages
+from core.prompt.utils.extract_thread_messages import extract_thread_messages
 from core.tools.__base.tool import Tool
 from core.tools.entities.tool_entities import (
     ToolParameter,
@@ -112,7 +112,7 @@ class BaseAgentRunner(AppRunner):
 
         # check if model supports stream tool call
         llm_model = cast(LargeLanguageModel, model_instance.model_type_instance)
-        model_schema = llm_model.get_model_schema(model_instance.model_name, model_instance.credentials)
+        model_schema = llm_model.get_model_schema(model_instance.model, model_instance.credentials)
         features = model_schema.features if model_schema and model_schema.features else []
         self.stream_tool_call = ModelFeature.STREAM_TOOL_CALL in features
         self.files = application_generate_entity.files if ModelFeature.VISION in features else []

api/core/agent/cot_agent_runner.py

@@ -17,8 +17,8 @@ from core.model_runtime.entities.message_entities import (
     ToolPromptMessage,
     UserPromptMessage,
 )
-from core.model_runtime.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
 from core.ops.ops_trace_manager import TraceQueueManager
+from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
 from core.tools.__base.tool import Tool
 from core.tools.entities.tool_entities import ToolInvokeMeta
 from core.tools.tool_engine import ToolEngine
@@ -245,7 +245,7 @@ class CotAgentRunner(BaseAgentRunner, ABC):
             iteration_step += 1
 
         yield LLMResultChunk(
-            model=model_instance.model_name,
+            model=model_instance.model,
             prompt_messages=prompt_messages,
             delta=LLMResultChunkDelta(
                 index=0, message=AssistantPromptMessage(content=final_answer), usage=llm_usage["usage"]
@@ -268,7 +268,7 @@ class CotAgentRunner(BaseAgentRunner, ABC):
         self.queue_manager.publish(
             QueueMessageEndEvent(
                 llm_result=LLMResult(
-                    model=model_instance.model_name,
+                    model=model_instance.model,
                     prompt_messages=prompt_messages,
                     message=AssistantPromptMessage(content=final_answer),
                     usage=llm_usage["usage"] or LLMUsage.empty_usage(),

api/core/agent/fc_agent_runner.py

@@ -21,7 +21,7 @@ from core.model_runtime.entities import (
     UserPromptMessage,
 )
 from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
-from core.model_runtime.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
+from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
 from core.tools.entities.tool_entities import ToolInvokeMeta
 from core.tools.tool_engine import ToolEngine
 from core.workflow.file import file_manager
@@ -178,7 +178,7 @@ class FunctionCallAgentRunner(BaseAgentRunner):
                 )
 
                 yield LLMResultChunk(
-                    model=model_instance.model_name,
+                    model=model_instance.model,
                     prompt_messages=result.prompt_messages,
                     system_fingerprint=result.system_fingerprint,
                     delta=LLMResultChunkDelta(
@@ -308,7 +308,7 @@ class FunctionCallAgentRunner(BaseAgentRunner):
         self.queue_manager.publish(
             QueueMessageEndEvent(
                 llm_result=LLMResult(
-                    model=model_instance.model_name,
+                    model=model_instance.model,
                     prompt_messages=prompt_messages,
                     message=AssistantPromptMessage(content=final_answer),
                     usage=llm_usage["usage"] or LLMUsage.empty_usage(),

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

@@ -5,7 +5,7 @@ from core.app.app_config.entities import (
     PromptTemplateEntity,
 )
 from core.model_runtime.entities.message_entities import PromptMessageRole
-from core.model_runtime.prompt.simple_prompt_transform import ModelMode
+from core.prompt.simple_prompt_transform import ModelMode
 from models.model import AppMode
 
 

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

@@ -32,8 +32,8 @@ from core.app.entities.task_entities import ChatbotAppBlockingResponse, ChatbotA
 from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig, PauseStatePersistenceLayer
 from core.helper.trace_id_helper import extract_external_trace_id_from_args
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
-from core.model_runtime.prompt.utils.get_thread_messages_length import get_thread_messages_length
 from core.ops.ops_trace_manager import TraceQueueManager
+from core.prompt.utils.get_thread_messages_length import get_thread_messages_length
 from core.repositories import DifyCoreRepositoryFactory
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.repositories.draft_variable_repository import (

api/core/app/apps/advanced_chat/app_runner.py

@@ -25,6 +25,7 @@ from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, Workfl
 from core.db.session_factory import session_factory
 from core.moderation.base import ModerationError
 from core.moderation.input_moderation import InputModeration
+from core.variables.variables import Variable
 from core.workflow.enums import WorkflowType
 from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
@@ -33,7 +34,6 @@ from core.workflow.repositories.workflow_node_execution_repository import Workfl
 from core.workflow.runtime import GraphRuntimeState, VariablePool
 from core.workflow.system_variable import SystemVariable
 from core.workflow.variable_loader import VariableLoader
-from core.workflow.variables.variables import Variable
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client

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

@@ -669,14 +669,16 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
     ) -> Generator[StreamResponse, None, None]:
         """Handle retriever resources events."""
         self._message_cycle_manager.handle_retriever_resources(event)
-        yield from ()
+        return
+        yield  # Make this a generator
 
     def _handle_annotation_reply_event(
         self, event: QueueAnnotationReplyEvent, **kwargs
     ) -> Generator[StreamResponse, None, None]:
         """Handle annotation reply events."""
         self._message_cycle_manager.handle_annotation_reply(event)
-        yield from ()
+        return
+        yield  # Make this a generator
 
     def _handle_message_replace_event(
         self, event: QueueMessageReplaceEvent, **kwargs

api/core/app/apps/agent_chat/app_runner.py

@@ -178,7 +178,7 @@ class AgentChatAppRunner(AppRunner):
 
         # change function call strategy based on LLM model
         llm_model = cast(LargeLanguageModel, model_instance.model_type_instance)
-        model_schema = llm_model.get_model_schema(model_instance.model_name, model_instance.credentials)
+        model_schema = llm_model.get_model_schema(model_instance.model, model_instance.credentials)
         if not model_schema:
             raise ValueError("Model schema not found")
 

api/core/app/apps/base_app_queue_manager.py

@@ -122,7 +122,7 @@ class AppQueueManager(ABC):
         """Attach the live graph runtime state reference for downstream consumers."""
         self._graph_runtime_state = graph_runtime_state
 
-    def publish(self, event: AppQueueEvent, pub_from: PublishFrom) -> None:
+    def publish(self, event: AppQueueEvent, pub_from: PublishFrom):
         """
         Publish event to queue
         :param event:

api/core/app/apps/base_app_runner.py

@@ -33,14 +33,10 @@ from core.model_runtime.entities.message_entities import (
 )
 from core.model_runtime.entities.model_entities import ModelPropertyKey
 from core.model_runtime.errors.invoke import InvokeBadRequestError
-from core.model_runtime.prompt.advanced_prompt_transform import AdvancedPromptTransform
-from core.model_runtime.prompt.entities.advanced_prompt_entities import (
-    ChatModelMessage,
-    CompletionModelPromptTemplate,
-    MemoryConfig,
-)
-from core.model_runtime.prompt.simple_prompt_transform import ModelMode, SimplePromptTransform
 from core.moderation.input_moderation import InputModeration
+from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
+from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate, MemoryConfig
+from core.prompt.simple_prompt_transform import ModelMode, SimplePromptTransform
 from core.tools.tool_file_manager import ToolFileManager
 from core.workflow.file.enums import FileTransferMethod, FileType
 from extensions.ext_database import db

api/core/app/apps/common/workflow_response_converter.py

@@ -49,6 +49,7 @@ from core.plugin.impl.datasource import PluginDatasourceManager
 from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.tool_manager import ToolManager
 from core.trigger.trigger_manager import TriggerManager
+from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.workflow.entities.pause_reason import HumanInputRequired
 from core.workflow.entities.workflow_start_reason import WorkflowStartReason
 from core.workflow.enums import (
@@ -61,7 +62,6 @@ from core.workflow.enums import (
 from core.workflow.file import FILE_MODEL_IDENTITY, File
 from core.workflow.runtime import GraphRuntimeState
 from core.workflow.system_variable import SystemVariable
-from core.workflow.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.workflow.workflow_entry import WorkflowEntry
 from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from extensions.ext_database import db

api/core/app/apps/message_based_app_generator.py

@@ -27,7 +27,7 @@ from core.app.entities.task_entities import (
     CompletionAppStreamResponse,
 )
 from core.app.task_pipeline.easy_ui_based_generate_task_pipeline import EasyUIBasedGenerateTaskPipeline
-from core.model_runtime.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from extensions.ext_database import db
 from extensions.ext_redis import get_pubsub_broadcast_channel
 from libs.broadcast_channel.channel import Topic

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

@@ -11,6 +11,7 @@ from core.app.entities.app_invoke_entities import (
 )
 from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from core.app.workflow.node_factory import DifyNodeFactory
+from core.variables.variables import RAGPipelineVariable, RAGPipelineVariableInput
 from core.workflow.entities.graph_init_params import GraphInitParams
 from core.workflow.enums import WorkflowType
 from core.workflow.graph import Graph
@@ -20,7 +21,6 @@ from core.workflow.repositories.workflow_node_execution_repository import Workfl
 from core.workflow.runtime import GraphRuntimeState, VariablePool
 from core.workflow.system_variable import SystemVariable
 from core.workflow.variable_loader import VariableLoader
-from core.workflow.variables.variables import RAGPipelineVariable, RAGPipelineVariableInput
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
 from models.dataset import Document, Pipeline

api/core/app/layers/conversation_variable_persist_layer.py

@@ -1,12 +1,12 @@
 import logging
 
+from core.variables import VariableBase
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID
 from core.workflow.conversation_variable_updater import ConversationVariableUpdater
 from core.workflow.enums import NodeType
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.graph_events import GraphEngineEvent, NodeRunSucceededEvent
 from core.workflow.nodes.variable_assigner.common import helpers as common_helpers
-from core.workflow.variables import VariableBase
 
 logger = logging.getLogger(__name__)
 

api/core/app/llm/__init__.py

@@ -1 +0,0 @@
-"""LLM-related application services."""

api/core/app/llm/model_access.py

@@ -1,110 +0,0 @@
-from __future__ import annotations
-
-from typing import Any
-
-from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.errors.error import ProviderTokenNotInitError
-from core.model_manager import ModelInstance, ModelManager
-from core.model_runtime.entities.model_entities import ModelType
-from core.provider_manager import ProviderManager
-from core.workflow.nodes.llm.entities import ModelConfig
-from core.workflow.nodes.llm.exc import LLMModeRequiredError, ModelNotExistError
-from core.workflow.nodes.llm.protocols import CredentialsProvider, ModelFactory
-
-
-class DifyCredentialsProvider:
-    tenant_id: str
-    provider_manager: ProviderManager
-
-    def __init__(self, tenant_id: str, provider_manager: ProviderManager | None = None) -> None:
-        self.tenant_id = tenant_id
-        self.provider_manager = provider_manager or ProviderManager()
-
-    def fetch(self, provider_name: str, model_name: str) -> dict[str, Any]:
-        provider_configurations = self.provider_manager.get_configurations(self.tenant_id)
-        provider_configuration = provider_configurations.get(provider_name)
-        if not provider_configuration:
-            raise ValueError(f"Provider {provider_name} does not exist.")
-
-        provider_model = provider_configuration.get_provider_model(model_type=ModelType.LLM, model=model_name)
-        if provider_model is None:
-            raise ModelNotExistError(f"Model {model_name} not exist.")
-        provider_model.raise_for_status()
-
-        credentials = provider_configuration.get_current_credentials(model_type=ModelType.LLM, model=model_name)
-        if credentials is None:
-            raise ProviderTokenNotInitError(f"Model {model_name} credentials is not initialized.")
-
-        return credentials
-
-
-class DifyModelFactory:
-    tenant_id: str
-    model_manager: ModelManager
-
-    def __init__(self, tenant_id: str, model_manager: ModelManager | None = None) -> None:
-        self.tenant_id = tenant_id
-        self.model_manager = model_manager or ModelManager()
-
-    def init_model_instance(self, provider_name: str, model_name: str) -> ModelInstance:
-        return self.model_manager.get_model_instance(
-            tenant_id=self.tenant_id,
-            provider=provider_name,
-            model_type=ModelType.LLM,
-            model=model_name,
-        )
-
-
-def build_dify_model_access(tenant_id: str) -> tuple[CredentialsProvider, ModelFactory]:
-    return (
-        DifyCredentialsProvider(tenant_id=tenant_id),
-        DifyModelFactory(tenant_id=tenant_id),
-    )
-
-
-def fetch_model_config(
-    *,
-    node_data_model: ModelConfig,
-    credentials_provider: CredentialsProvider,
-    model_factory: ModelFactory,
-) -> tuple[ModelInstance, ModelConfigWithCredentialsEntity]:
-    if not node_data_model.mode:
-        raise LLMModeRequiredError("LLM mode is required.")
-
-    credentials = credentials_provider.fetch(node_data_model.provider, node_data_model.name)
-    model_instance = model_factory.init_model_instance(node_data_model.provider, node_data_model.name)
-    provider_model_bundle = model_instance.provider_model_bundle
-
-    provider_model = provider_model_bundle.configuration.get_provider_model(
-        model=node_data_model.name,
-        model_type=ModelType.LLM,
-    )
-    if provider_model is None:
-        raise ModelNotExistError(f"Model {node_data_model.name} not exist.")
-    provider_model.raise_for_status()
-
-    completion_params = dict(node_data_model.completion_params)
-    stop = completion_params.pop("stop", [])
-    if not isinstance(stop, list):
-        stop = []
-
-    model_schema = model_instance.model_type_instance.get_model_schema(node_data_model.name, credentials)
-    if not model_schema:
-        raise ModelNotExistError(f"Model {node_data_model.name} not exist.")
-
-    model_instance.provider = node_data_model.provider
-    model_instance.model_name = node_data_model.name
-    model_instance.credentials = credentials
-    model_instance.parameters = completion_params
-    model_instance.stop = tuple(stop)
-
-    return model_instance, ModelConfigWithCredentialsEntity(
-        provider=node_data_model.provider,
-        model=node_data_model.name,
-        model_schema=model_schema,
-        mode=node_data_model.mode,
-        provider_model_bundle=provider_model_bundle,
-        credentials=credentials,
-        parameters=completion_params,
-        stop=stop,
-    )

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -52,10 +52,10 @@ from core.model_runtime.entities.message_entities import (
     TextPromptMessageContent,
 )
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from core.model_runtime.prompt.utils.prompt_message_util import PromptMessageUtil
-from core.model_runtime.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
+from core.prompt.utils.prompt_message_util import PromptMessageUtil
+from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.tools.signature import sign_tool_file
 from core.workflow.file import helpers as file_helpers
 from core.workflow.file.enums import FileTransferMethod

api/core/app/workflow/node_factory.py

@@ -1,20 +1,11 @@
-from collections.abc import Mapping
-from typing import TYPE_CHECKING, Any, cast, final
+from typing import TYPE_CHECKING, final
 
 from typing_extensions import override
 
 from configs import dify_config
-from core.app.llm.model_access import build_dify_model_access
-from core.datasource.datasource_manager import DatasourceManager
-from core.helper.code_executor.code_executor import (
-    CodeExecutionError,
-    CodeExecutor,
-)
+from core.helper.code_executor.code_executor import CodeExecutor
+from core.helper.code_executor.code_node_provider import CodeNodeProvider
 from core.helper.ssrf_proxy import ssrf_proxy
-from core.model_manager import ModelInstance
-from core.model_runtime.entities.model_entities import ModelType
-from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from core.model_runtime.prompt.entities.advanced_prompt_entities import MemoryConfig
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
 from core.tools.tool_file_manager import ToolFileManager
 from core.workflow.entities.graph_config import NodeConfigDict
@@ -22,24 +13,13 @@ from core.workflow.enums import NodeType
 from core.workflow.file.file_manager import file_manager
 from core.workflow.graph.graph import NodeFactory
 from core.workflow.nodes.base.node import Node
-from core.workflow.nodes.code.code_node import CodeNode, WorkflowCodeExecutor
-from core.workflow.nodes.code.entities import CodeLanguage
+from core.workflow.nodes.code.code_node import CodeNode
 from core.workflow.nodes.code.limits import CodeNodeLimits
-from core.workflow.nodes.datasource import DatasourceNode
 from core.workflow.nodes.document_extractor import DocumentExtractorNode, UnstructuredApiConfig
 from core.workflow.nodes.http_request import HttpRequestNode, build_http_request_config
 from core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node import KnowledgeRetrievalNode
-from core.workflow.nodes.llm import llm_utils
-from core.workflow.nodes.llm.entities import ModelConfig
-from core.workflow.nodes.llm.exc import LLMModeRequiredError, ModelNotExistError
-from core.workflow.nodes.llm.node import LLMNode
-from core.workflow.nodes.llm.protocols import PromptMessageMemory
 from core.workflow.nodes.node_mapping import LATEST_VERSION, NODE_TYPE_CLASSES_MAPPING
-from core.workflow.nodes.parameter_extractor.parameter_extractor_node import ParameterExtractorNode
-from core.workflow.nodes.question_classifier.question_classifier_node import QuestionClassifierNode
-from core.workflow.nodes.template_transform.template_renderer import (
-    CodeExecutorJinja2TemplateRenderer,
-)
+from core.workflow.nodes.template_transform.template_renderer import CodeExecutorJinja2TemplateRenderer
 from core.workflow.nodes.template_transform.template_transform_node import TemplateTransformNode
 
 if TYPE_CHECKING:
@@ -47,24 +27,6 @@ if TYPE_CHECKING:
     from core.workflow.runtime import GraphRuntimeState
 
 
-class DefaultWorkflowCodeExecutor:
-    def execute(
-        self,
-        *,
-        language: CodeLanguage,
-        code: str,
-        inputs: Mapping[str, Any],
-    ) -> Mapping[str, Any]:
-        return CodeExecutor.execute_workflow_code_template(
-            language=language,
-            code=code,
-            inputs=inputs,
-        )
-
-    def is_execution_error(self, error: Exception) -> bool:
-        return isinstance(error, CodeExecutionError)
-
-
 @final
 class DifyNodeFactory(NodeFactory):
     """
@@ -81,7 +43,8 @@ class DifyNodeFactory(NodeFactory):
     ) -> None:
         self.graph_init_params = graph_init_params
         self.graph_runtime_state = graph_runtime_state
-        self._code_executor: WorkflowCodeExecutor = DefaultWorkflowCodeExecutor()
+        self._code_executor: type[CodeExecutor] = CodeExecutor
+        self._code_providers: tuple[type[CodeNodeProvider], ...] = CodeNode.default_code_providers()
         self._code_limits = CodeNodeLimits(
             max_string_length=dify_config.CODE_MAX_STRING_LENGTH,
             max_number=dify_config.CODE_MAX_NUMBER,
@@ -112,8 +75,6 @@ class DifyNodeFactory(NodeFactory):
             ssrf_default_max_retries=dify_config.SSRF_DEFAULT_MAX_RETRIES,
         )
 
-        self._llm_credentials_provider, self._llm_model_factory = build_dify_model_access(graph_init_params.tenant_id)
-
     @override
     def create_node(self, node_config: NodeConfigDict) -> Node:
         """
@@ -153,6 +114,7 @@ class DifyNodeFactory(NodeFactory):
                 graph_init_params=self.graph_init_params,
                 graph_runtime_state=self.graph_runtime_state,
                 code_executor=self._code_executor,
+                code_providers=self._code_providers,
                 code_limits=self._code_limits,
             )
 
@@ -178,29 +140,6 @@ class DifyNodeFactory(NodeFactory):
                 file_manager=self._http_request_file_manager,
             )
 
-        if node_type == NodeType.LLM:
-            model_instance = self._build_model_instance_for_llm_node(node_data)
-            memory = self._build_memory_for_llm_node(node_data=node_data, model_instance=model_instance)
-            return LLMNode(
-                id=node_id,
-                config=node_config,
-                graph_init_params=self.graph_init_params,
-                graph_runtime_state=self.graph_runtime_state,
-                credentials_provider=self._llm_credentials_provider,
-                model_factory=self._llm_model_factory,
-                model_instance=model_instance,
-                memory=memory,
-            )
-
-        if node_type == NodeType.DATASOURCE:
-            return DatasourceNode(
-                id=node_id,
-                config=node_config,
-                graph_init_params=self.graph_init_params,
-                graph_runtime_state=self.graph_runtime_state,
-                datasource_manager=DatasourceManager,
-            )
-
         if node_type == NodeType.KNOWLEDGE_RETRIEVAL:
             return KnowledgeRetrievalNode(
                 id=node_id,
@@ -219,85 +158,9 @@ class DifyNodeFactory(NodeFactory):
                 unstructured_api_config=self._document_extractor_unstructured_api_config,
             )
 
-        if node_type == NodeType.QUESTION_CLASSIFIER:
-            model_instance = self._build_model_instance_for_llm_node(node_data)
-            return QuestionClassifierNode(
-                id=node_id,
-                config=node_config,
-                graph_init_params=self.graph_init_params,
-                graph_runtime_state=self.graph_runtime_state,
-                credentials_provider=self._llm_credentials_provider,
-                model_factory=self._llm_model_factory,
-                model_instance=model_instance,
-            )
-
-        if node_type == NodeType.PARAMETER_EXTRACTOR:
-            model_instance = self._build_model_instance_for_llm_node(node_data)
-            return ParameterExtractorNode(
-                id=node_id,
-                config=node_config,
-                graph_init_params=self.graph_init_params,
-                graph_runtime_state=self.graph_runtime_state,
-                credentials_provider=self._llm_credentials_provider,
-                model_factory=self._llm_model_factory,
-                model_instance=model_instance,
-            )
-
         return node_class(
             id=node_id,
             config=node_config,
             graph_init_params=self.graph_init_params,
             graph_runtime_state=self.graph_runtime_state,
         )
-
-    def _build_model_instance_for_llm_node(self, node_data: Mapping[str, Any]) -> ModelInstance:
-        node_data_model = ModelConfig.model_validate(node_data["model"])
-        if not node_data_model.mode:
-            raise LLMModeRequiredError("LLM mode is required.")
-
-        credentials = self._llm_credentials_provider.fetch(node_data_model.provider, node_data_model.name)
-        model_instance = self._llm_model_factory.init_model_instance(node_data_model.provider, node_data_model.name)
-        provider_model_bundle = model_instance.provider_model_bundle
-
-        provider_model = provider_model_bundle.configuration.get_provider_model(
-            model=node_data_model.name,
-            model_type=ModelType.LLM,
-        )
-        if provider_model is None:
-            raise ModelNotExistError(f"Model {node_data_model.name} not exist.")
-        provider_model.raise_for_status()
-
-        completion_params = dict(node_data_model.completion_params)
-        stop = completion_params.pop("stop", [])
-        if not isinstance(stop, list):
-            stop = []
-
-        model_schema = model_instance.model_type_instance.get_model_schema(node_data_model.name, credentials)
-        if not model_schema:
-            raise ModelNotExistError(f"Model {node_data_model.name} not exist.")
-
-        model_instance.provider = node_data_model.provider
-        model_instance.model_name = node_data_model.name
-        model_instance.credentials = credentials
-        model_instance.parameters = completion_params
-        model_instance.stop = tuple(stop)
-        model_instance.model_type_instance = cast(LargeLanguageModel, model_instance.model_type_instance)
-        return model_instance
-
-    def _build_memory_for_llm_node(
-        self,
-        *,
-        node_data: Mapping[str, Any],
-        model_instance: ModelInstance,
-    ) -> PromptMessageMemory | None:
-        raw_memory_config = node_data.get("memory")
-        if raw_memory_config is None:
-            return None
-
-        node_memory = MemoryConfig.model_validate(raw_memory_config)
-        return llm_utils.fetch_memory(
-            variable_pool=self.graph_runtime_state.variable_pool,
-            app_id=self.graph_init_params.app_id,
-            node_data_memory=node_memory,
-            model_instance=model_instance,
-        )

api/core/datasource/datasource_manager.py

@@ -1,39 +1,16 @@
 import logging
-from collections.abc import Generator
 from threading import Lock
-from typing import Any, cast
-
-from sqlalchemy import select
 
 import contexts
 from core.datasource.__base.datasource_plugin import DatasourcePlugin
 from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
-from core.datasource.entities.datasource_entities import (
-    DatasourceMessage,
-    DatasourceProviderType,
-    GetOnlineDocumentPageContentRequest,
-    OnlineDriveDownloadFileRequest,
-)
+from core.datasource.entities.datasource_entities import DatasourceProviderType
 from core.datasource.errors import DatasourceProviderNotFoundError
 from core.datasource.local_file.local_file_provider import LocalFileDatasourcePluginProviderController
-from core.datasource.online_document.online_document_plugin import OnlineDocumentDatasourcePlugin
 from core.datasource.online_document.online_document_provider import OnlineDocumentDatasourcePluginProviderController
-from core.datasource.online_drive.online_drive_plugin import OnlineDriveDatasourcePlugin
 from core.datasource.online_drive.online_drive_provider import OnlineDriveDatasourcePluginProviderController
-from core.datasource.utils.message_transformer import DatasourceFileMessageTransformer
 from core.datasource.website_crawl.website_crawl_provider import WebsiteCrawlDatasourcePluginProviderController
-from core.db.session_factory import session_factory
 from core.plugin.impl.datasource import PluginDatasourceManager
-from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionStatus
-from core.workflow.enums import WorkflowNodeExecutionMetadataKey
-from core.workflow.file import File
-from core.workflow.file.enums import FileTransferMethod, FileType
-from core.workflow.node_events import NodeRunResult, StreamChunkEvent, StreamCompletedEvent
-from core.workflow.repositories.datasource_manager_protocol import DatasourceParameter, OnlineDriveDownloadFileParam
-from factories import file_factory
-from models.model import UploadFile
-from models.tools import ToolFile
-from services.datasource_provider_service import DatasourceProviderService
 
 logger = logging.getLogger(__name__)
 
@@ -126,238 +103,3 @@ class DatasourceManager:
             tenant_id,
             datasource_type,
         ).get_datasource(datasource_name)
-
-    @classmethod
-    def get_icon_url(cls, provider_id: str, tenant_id: str, datasource_name: str, datasource_type: str) -> str:
-        datasource_runtime = cls.get_datasource_runtime(
-            provider_id=provider_id,
-            datasource_name=datasource_name,
-            tenant_id=tenant_id,
-            datasource_type=DatasourceProviderType.value_of(datasource_type),
-        )
-        return datasource_runtime.get_icon_url(tenant_id)
-
-    @classmethod
-    def stream_online_results(
-        cls,
-        *,
-        user_id: str,
-        datasource_name: str,
-        datasource_type: str,
-        provider_id: str,
-        tenant_id: str,
-        provider: str,
-        plugin_id: str,
-        credential_id: str,
-        datasource_param: DatasourceParameter | None = None,
-        online_drive_request: OnlineDriveDownloadFileParam | None = None,
-    ) -> Generator[DatasourceMessage, None, Any]:
-        """
-        Pull-based streaming of domain messages from datasource plugins.
-        Returns a generator that yields DatasourceMessage and finally returns a minimal final payload.
-        Only ONLINE_DOCUMENT and ONLINE_DRIVE are streamable here; other types are handled by nodes directly.
-        """
-        ds_type = DatasourceProviderType.value_of(datasource_type)
-        runtime = cls.get_datasource_runtime(
-            provider_id=provider_id,
-            datasource_name=datasource_name,
-            tenant_id=tenant_id,
-            datasource_type=ds_type,
-        )
-
-        dsp_service = DatasourceProviderService()
-        credentials = dsp_service.get_datasource_credentials(
-            tenant_id=tenant_id,
-            provider=provider,
-            plugin_id=plugin_id,
-            credential_id=credential_id,
-        )
-
-        if ds_type == DatasourceProviderType.ONLINE_DOCUMENT:
-            doc_runtime = cast(OnlineDocumentDatasourcePlugin, runtime)
-            if credentials:
-                doc_runtime.runtime.credentials = credentials
-            if datasource_param is None:
-                raise ValueError("datasource_param is required for ONLINE_DOCUMENT streaming")
-            inner_gen: Generator[DatasourceMessage, None, None] = doc_runtime.get_online_document_page_content(
-                user_id=user_id,
-                datasource_parameters=GetOnlineDocumentPageContentRequest(
-                    workspace_id=datasource_param.workspace_id,
-                    page_id=datasource_param.page_id,
-                    type=datasource_param.type,
-                ),
-                provider_type=ds_type,
-            )
-        elif ds_type == DatasourceProviderType.ONLINE_DRIVE:
-            drive_runtime = cast(OnlineDriveDatasourcePlugin, runtime)
-            if credentials:
-                drive_runtime.runtime.credentials = credentials
-            if online_drive_request is None:
-                raise ValueError("online_drive_request is required for ONLINE_DRIVE streaming")
-            inner_gen = drive_runtime.online_drive_download_file(
-                user_id=user_id,
-                request=OnlineDriveDownloadFileRequest(
-                    id=online_drive_request.id,
-                    bucket=online_drive_request.bucket,
-                ),
-                provider_type=ds_type,
-            )
-        else:
-            raise ValueError(f"Unsupported datasource type for streaming: {ds_type}")
-
-        # Bridge through to caller while preserving generator return contract
-        yield from inner_gen
-        # No structured final data here; node/adapter will assemble outputs
-        return {}
-
-    @classmethod
-    def stream_node_events(
-        cls,
-        *,
-        node_id: str,
-        user_id: str,
-        datasource_name: str,
-        datasource_type: str,
-        provider_id: str,
-        tenant_id: str,
-        provider: str,
-        plugin_id: str,
-        credential_id: str,
-        parameters_for_log: dict[str, Any],
-        datasource_info: dict[str, Any],
-        variable_pool: Any,
-        datasource_param: DatasourceParameter | None = None,
-        online_drive_request: OnlineDriveDownloadFileParam | None = None,
-    ) -> Generator[StreamChunkEvent | StreamCompletedEvent, None, None]:
-        ds_type = DatasourceProviderType.value_of(datasource_type)
-
-        messages = cls.stream_online_results(
-            user_id=user_id,
-            datasource_name=datasource_name,
-            datasource_type=datasource_type,
-            provider_id=provider_id,
-            tenant_id=tenant_id,
-            provider=provider,
-            plugin_id=plugin_id,
-            credential_id=credential_id,
-            datasource_param=datasource_param,
-            online_drive_request=online_drive_request,
-        )
-
-        transformed = DatasourceFileMessageTransformer.transform_datasource_invoke_messages(
-            messages=messages, user_id=user_id, tenant_id=tenant_id, conversation_id=None
-        )
-
-        variables: dict[str, Any] = {}
-        file_out: File | None = None
-
-        for message in transformed:
-            mtype = message.type
-            if mtype in {
-                DatasourceMessage.MessageType.IMAGE_LINK,
-                DatasourceMessage.MessageType.BINARY_LINK,
-                DatasourceMessage.MessageType.IMAGE,
-            }:
-                wanted_ds_type = ds_type in {
-                    DatasourceProviderType.ONLINE_DRIVE,
-                    DatasourceProviderType.ONLINE_DOCUMENT,
-                }
-                if wanted_ds_type and isinstance(message.message, DatasourceMessage.TextMessage):
-                    url = message.message.text
-
-                    datasource_file_id = str(url).split("/")[-1].split(".")[0]
-                    with session_factory.create_session() as session:
-                        stmt = select(ToolFile).where(
-                            ToolFile.id == datasource_file_id, ToolFile.tenant_id == tenant_id
-                        )
-                        datasource_file = session.scalar(stmt)
-                        if not datasource_file:
-                            raise ValueError(
-                                f"ToolFile not found for file_id={datasource_file_id}, tenant_id={tenant_id}"
-                            )
-                        mime_type = datasource_file.mimetype
-                    if datasource_file is not None:
-                        mapping = {
-                            "tool_file_id": datasource_file_id,
-                            "type": file_factory.get_file_type_by_mime_type(mime_type),
-                            "transfer_method": FileTransferMethod.TOOL_FILE,
-                            "url": url,
-                        }
-                        file_out = file_factory.build_from_mapping(mapping=mapping, tenant_id=tenant_id)
-            elif mtype == DatasourceMessage.MessageType.TEXT:
-                assert isinstance(message.message, DatasourceMessage.TextMessage)
-                yield StreamChunkEvent(selector=[node_id, "text"], chunk=message.message.text, is_final=False)
-            elif mtype == DatasourceMessage.MessageType.LINK:
-                assert isinstance(message.message, DatasourceMessage.TextMessage)
-                yield StreamChunkEvent(
-                    selector=[node_id, "text"], chunk=f"Link: {message.message.text}\n", is_final=False
-                )
-            elif mtype == DatasourceMessage.MessageType.VARIABLE:
-                assert isinstance(message.message, DatasourceMessage.VariableMessage)
-                name = message.message.variable_name
-                value = message.message.variable_value
-                if message.message.stream:
-                    assert isinstance(value, str), "stream variable_value must be str"
-                    variables[name] = variables.get(name, "") + value
-                    yield StreamChunkEvent(selector=[node_id, name], chunk=value, is_final=False)
-                else:
-                    variables[name] = value
-            elif mtype == DatasourceMessage.MessageType.FILE:
-                if ds_type == DatasourceProviderType.ONLINE_DRIVE and message.meta:
-                    f = message.meta.get("file")
-                    if isinstance(f, File):
-                        file_out = f
-            else:
-                pass
-
-        yield StreamChunkEvent(selector=[node_id, "text"], chunk="", is_final=True)
-
-        if ds_type == DatasourceProviderType.ONLINE_DRIVE and file_out is not None:
-            variable_pool.add([node_id, "file"], file_out)
-
-        if ds_type == DatasourceProviderType.ONLINE_DOCUMENT:
-            yield StreamCompletedEvent(
-                node_run_result=NodeRunResult(
-                    status=WorkflowNodeExecutionStatus.SUCCEEDED,
-                    inputs=parameters_for_log,
-                    metadata={WorkflowNodeExecutionMetadataKey.DATASOURCE_INFO: datasource_info},
-                    outputs={**variables},
-                )
-            )
-        else:
-            yield StreamCompletedEvent(
-                node_run_result=NodeRunResult(
-                    status=WorkflowNodeExecutionStatus.SUCCEEDED,
-                    inputs=parameters_for_log,
-                    metadata={WorkflowNodeExecutionMetadataKey.DATASOURCE_INFO: datasource_info},
-                    outputs={
-                        "file": file_out,
-                        "datasource_type": ds_type,
-                    },
-                )
-            )
-
-    @classmethod
-    def get_upload_file_by_id(cls, file_id: str, tenant_id: str) -> File:
-        with session_factory.create_session() as session:
-            upload_file = (
-                session.query(UploadFile).where(UploadFile.id == file_id, UploadFile.tenant_id == tenant_id).first()
-            )
-            if not upload_file:
-                raise ValueError(f"UploadFile not found for file_id={file_id}, tenant_id={tenant_id}")
-
-        file_info = File(
-            id=upload_file.id,
-            filename=upload_file.name,
-            extension="." + upload_file.extension,
-            mime_type=upload_file.mime_type,
-            tenant_id=tenant_id,
-            type=FileType.CUSTOM,
-            transfer_method=FileTransferMethod.LOCAL_FILE,
-            remote_url=upload_file.source_url,
-            related_id=upload_file.id,
-            size=upload_file.size,
-            storage_key=upload_file.key,
-            url=upload_file.source_url,
-        )
-        return file_info

api/core/datasource/entities/datasource_entities.py

@@ -379,11 +379,4 @@ class OnlineDriveDownloadFileRequest(BaseModel):
     """
 
     id: str = Field(..., description="The id of the file")
-    bucket: str = Field("", description="The name of the bucket")
-
-    @field_validator("bucket", mode="before")
-    @classmethod
-    def _coerce_bucket(cls, v) -> str:
-        if v is None:
-            return ""
-        return str(v)
+    bucket: str | None = Field(None, description="The name of the bucket")

api/core/helper/code_executor/code_executor.py

@@ -1,5 +1,6 @@
 import logging
 from collections.abc import Mapping
+from enum import StrEnum
 from threading import Lock
 from typing import Any
 
@@ -13,7 +14,6 @@ from core.helper.code_executor.jinja2.jinja2_transformer import Jinja2TemplateTr
 from core.helper.code_executor.python3.python3_transformer import Python3TemplateTransformer
 from core.helper.code_executor.template_transformer import TemplateTransformer
 from core.helper.http_client_pooling import get_pooled_http_client
-from core.workflow.nodes.code.entities import CodeLanguage
 
 logger = logging.getLogger(__name__)
 code_execution_endpoint_url = URL(str(dify_config.CODE_EXECUTION_ENDPOINT))
@@ -40,6 +40,12 @@ class CodeExecutionResponse(BaseModel):
     data: Data
 
 
+class CodeLanguage(StrEnum):
+    PYTHON3 = "python3"
+    JINJA2 = "jinja2"
+    JAVASCRIPT = "javascript"
+
+
 def _build_code_executor_client() -> httpx.Client:
     return httpx.Client(
         verify=CODE_EXECUTION_SSL_VERIFY,

api/core/helper/code_executor/template_transformer.py

@@ -5,7 +5,7 @@ from base64 import b64encode
 from collections.abc import Mapping
 from typing import Any
 
-from core.workflow.variables.utils import dumps_with_segments
+from core.variables.utils import dumps_with_segments
 
 
 class TemplateTransformer(ABC):

api/core/llm_generator/llm_generator.py

@@ -27,10 +27,10 @@ from core.model_runtime.entities.llm_entities import LLMResult
 from core.model_runtime.entities.message_entities import PromptMessage, SystemPromptMessage, UserPromptMessage
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.errors.invoke import InvokeAuthorizationError, InvokeError
-from core.model_runtime.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.ops.utils import measure_time
+from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
 from extensions.ext_database import db
 from extensions.ext_storage import storage

api/core/memory/token_buffer_memory.py

@@ -14,7 +14,7 @@ from core.model_runtime.entities import (
     UserPromptMessage,
 )
 from core.model_runtime.entities.message_entities import PromptMessageContentUnionTypes
-from core.model_runtime.prompt.utils.extract_thread_messages import extract_thread_messages
+from core.prompt.utils.extract_thread_messages import extract_thread_messages
 from core.workflow.file import file_manager
 from extensions.ext_database import db
 from factories import file_factory

api/core/model_manager.py

@@ -1,5 +1,5 @@
 import logging
-from collections.abc import Callable, Generator, Iterable, Mapping, Sequence
+from collections.abc import Callable, Generator, Iterable, Sequence
 from typing import IO, Any, Literal, Optional, Union, cast, overload
 
 from configs import dify_config
@@ -35,12 +35,9 @@ class ModelInstance:
 
     def __init__(self, provider_model_bundle: ProviderModelBundle, model: str):
         self.provider_model_bundle = provider_model_bundle
-        self.model_name = model
+        self.model = model
         self.provider = provider_model_bundle.configuration.provider.provider
         self.credentials = self._fetch_credentials_from_bundle(provider_model_bundle, model)
-        # Runtime LLM invocation fields.
-        self.parameters: Mapping[str, Any] = {}
-        self.stop: Sequence[str] = ()
         self.model_type_instance = self.provider_model_bundle.model_type_instance
         self.load_balancing_manager = self._get_load_balancing_manager(
             configuration=provider_model_bundle.configuration,
@@ -166,7 +163,7 @@ class ModelInstance:
             Union[LLMResult, Generator],
             self._round_robin_invoke(
                 function=self.model_type_instance.invoke,
-                model=self.model_name,
+                model=self.model,
                 credentials=self.credentials,
                 prompt_messages=prompt_messages,
                 model_parameters=model_parameters,
@@ -194,7 +191,7 @@ class ModelInstance:
             int,
             self._round_robin_invoke(
                 function=self.model_type_instance.get_num_tokens,
-                model=self.model_name,
+                model=self.model,
                 credentials=self.credentials,
                 prompt_messages=prompt_messages,
                 tools=tools,
@@ -218,7 +215,7 @@ class ModelInstance:
             EmbeddingResult,
             self._round_robin_invoke(
                 function=self.model_type_instance.invoke,
-                model=self.model_name,
+                model=self.model,
                 credentials=self.credentials,
                 texts=texts,
                 user=user,
@@ -246,7 +243,7 @@ class ModelInstance:
             EmbeddingResult,
             self._round_robin_invoke(
                 function=self.model_type_instance.invoke,
-                model=self.model_name,
+                model=self.model,
                 credentials=self.credentials,
                 multimodel_documents=multimodel_documents,
                 user=user,
@@ -267,7 +264,7 @@ class ModelInstance:
             list[int],
             self._round_robin_invoke(
                 function=self.model_type_instance.get_num_tokens,
-                model=self.model_name,
+                model=self.model,
                 credentials=self.credentials,
                 texts=texts,
             ),
@@ -297,7 +294,7 @@ class ModelInstance:
             RerankResult,
             self._round_robin_invoke(
                 function=self.model_type_instance.invoke,
-                model=self.model_name,
+                model=self.model,
                 credentials=self.credentials,
                 query=query,
                 docs=docs,
@@ -331,7 +328,7 @@ class ModelInstance:
             RerankResult,
             self._round_robin_invoke(
                 function=self.model_type_instance.invoke_multimodal_rerank,
-                model=self.model_name,
+                model=self.model,
                 credentials=self.credentials,
                 query=query,
                 docs=docs,
@@ -355,7 +352,7 @@ class ModelInstance:
             bool,
             self._round_robin_invoke(
                 function=self.model_type_instance.invoke,
-                model=self.model_name,
+                model=self.model,
                 credentials=self.credentials,
                 text=text,
                 user=user,
@@ -376,7 +373,7 @@ class ModelInstance:
             str,
             self._round_robin_invoke(
                 function=self.model_type_instance.invoke,
-                model=self.model_name,
+                model=self.model,
                 credentials=self.credentials,
                 file=file,
                 user=user,
@@ -399,7 +396,7 @@ class ModelInstance:
             Iterable[bytes],
             self._round_robin_invoke(
                 function=self.model_type_instance.invoke,
-                model=self.model_name,
+                model=self.model,
                 credentials=self.credentials,
                 content_text=content_text,
                 user=user,
@@ -472,7 +469,7 @@ class ModelInstance:
         if not isinstance(self.model_type_instance, TTSModel):
             raise Exception("Model type instance is not TTSModel")
         return self.model_type_instance.get_tts_model_voices(
-            model=self.model_name, credentials=self.credentials, language=language
+            model=self.model, credentials=self.credentials, language=language
         )
 
 

api/core/model_runtime/model_providers/__base/large_language_model.py

@@ -83,21 +83,19 @@ def _merge_tool_call_delta(
         tool_call.function.arguments += delta.function.arguments
 
 
-def _build_llm_result_from_chunks(
+def _build_llm_result_from_first_chunk(
     model: str,
     prompt_messages: Sequence[PromptMessage],
     chunks: Iterator[LLMResultChunk],
 ) -> LLMResult:
     """
-    Build a single `LLMResult` by accumulating all returned chunks.
+    Build a single `LLMResult` from the first returned chunk.
 
-    Some models only support streaming output (e.g. Qwen3 open-source edition)
-    and the plugin side may still implement the response via a chunked stream,
-    so all chunks must be consumed and concatenated into a single ``LLMResult``.
+    This is used for `stream=False` because the plugin side may still implement the response via a chunked stream.
 
-    The ``usage`` is taken from the last chunk that carries it, which is the
-    typical convention for streaming responses (the final chunk contains the
-    aggregated token counts).
+    Note:
+        This function always drains the `chunks` iterator after reading the first chunk to ensure any underlying
+        streaming resources are released (e.g., HTTP connections owned by the plugin runtime).
     """
     content = ""
     content_list: list[PromptMessageContentUnionTypes] = []
@@ -106,27 +104,24 @@ def _build_llm_result_from_chunks(
     tools_calls: list[AssistantPromptMessage.ToolCall] = []
 
     try:
-        for chunk in chunks:
-            if isinstance(chunk.delta.message.content, str):
-                content += chunk.delta.message.content
-            elif isinstance(chunk.delta.message.content, list):
-                content_list.extend(chunk.delta.message.content)
+        first_chunk = next(chunks, None)
+        if first_chunk is not None:
+            if isinstance(first_chunk.delta.message.content, str):
+                content += first_chunk.delta.message.content
+            elif isinstance(first_chunk.delta.message.content, list):
+                content_list.extend(first_chunk.delta.message.content)
 
-            if chunk.delta.message.tool_calls:
-                _increase_tool_call(chunk.delta.message.tool_calls, tools_calls)
+            if first_chunk.delta.message.tool_calls:
+                _increase_tool_call(first_chunk.delta.message.tool_calls, tools_calls)
 
-            if chunk.delta.usage:
-                usage = chunk.delta.usage
-            if chunk.system_fingerprint:
-                system_fingerprint = chunk.system_fingerprint
-    except Exception:
-        logger.exception("Error while consuming non-stream plugin chunk iterator.")
-        raise
+            usage = first_chunk.delta.usage or LLMUsage.empty_usage()
+            system_fingerprint = first_chunk.system_fingerprint
     finally:
-        # Drain any remaining chunks to release underlying streaming resources (e.g. HTTP connections).
-        close = getattr(chunks, "close", None)
-        if callable(close):
-            close()
+        try:
+            for _ in chunks:
+                pass
+        except Exception:
+            logger.debug("Failed to drain non-stream plugin chunk iterator.", exc_info=True)
 
     return LLMResult(
         model=model,
@@ -179,7 +174,7 @@ def _normalize_non_stream_plugin_result(
 ) -> LLMResult:
     if isinstance(result, LLMResult):
         return result
-    return _build_llm_result_from_chunks(model=model, prompt_messages=prompt_messages, chunks=result)
+    return _build_llm_result_from_first_chunk(model=model, prompt_messages=prompt_messages, chunks=result)
 
 
 def _increase_tool_call(

api/core/ops/base_trace_instance.py

@@ -14,9 +14,10 @@ class BaseTraceInstance(ABC):
     Base trace instance for ops trace services
     """
 
+    @abstractmethod
     def __init__(self, trace_config: BaseTracingConfig):
         """
-        Initializer for the trace instance.
+        Abstract initializer for the trace instance.
         Distribute trace tasks by matching entities
         """
         self.trace_config = trace_config

api/core/ops/ops_trace_manager.py

@@ -41,8 +41,8 @@ logger = logging.getLogger(__name__)
 
 
 class OpsTraceProviderConfigMap(collections.UserDict[str, dict[str, Any]]):
-    def __getitem__(self, key: str) -> dict[str, Any]:
-        match key:
+    def __getitem__(self, provider: str) -> dict[str, Any]:
+        match provider:
             case TracingProviderEnum.LANGFUSE:
                 from core.ops.entities.config_entity import LangfuseConfig
                 from core.ops.langfuse_trace.langfuse_trace import LangFuseDataTrace
@@ -149,7 +149,7 @@ class OpsTraceProviderConfigMap(collections.UserDict[str, dict[str, Any]]):
                 }
 
             case _:
-                raise KeyError(f"Unsupported tracing provider: {key}")
+                raise KeyError(f"Unsupported tracing provider: {provider}")
 
 
 provider_config_map = OpsTraceProviderConfigMap()

api/core/prompt/advanced_prompt_transform.py

@@ -4,7 +4,6 @@ from typing import cast
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.helper.code_executor.jinja2.jinja2_formatter import Jinja2Formatter
 from core.memory.token_buffer_memory import TokenBufferMemory
-from core.model_manager import ModelInstance
 from core.model_runtime.entities import (
     AssistantPromptMessage,
     PromptMessage,
@@ -14,13 +13,9 @@ from core.model_runtime.entities import (
     UserPromptMessage,
 )
 from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
-from core.model_runtime.prompt.entities.advanced_prompt_entities import (
-    ChatModelMessage,
-    CompletionModelPromptTemplate,
-    MemoryConfig,
-)
-from core.model_runtime.prompt.prompt_transform import PromptTransform
-from core.model_runtime.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate, MemoryConfig
+from core.prompt.prompt_transform import PromptTransform
+from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.workflow.file import file_manager
 from core.workflow.file.models import File
 from core.workflow.runtime import VariablePool
@@ -49,8 +44,7 @@ class AdvancedPromptTransform(PromptTransform):
         context: str | None,
         memory_config: MemoryConfig | None,
         memory: TokenBufferMemory | None,
-        model_config: ModelConfigWithCredentialsEntity | None = None,
-        model_instance: ModelInstance | None = None,
+        model_config: ModelConfigWithCredentialsEntity,
         image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
         prompt_messages = []
@@ -65,7 +59,6 @@ class AdvancedPromptTransform(PromptTransform):
                 memory_config=memory_config,
                 memory=memory,
                 model_config=model_config,
-                model_instance=model_instance,
                 image_detail_config=image_detail_config,
             )
         elif isinstance(prompt_template, list) and all(isinstance(item, ChatModelMessage) for item in prompt_template):
@@ -78,7 +71,6 @@ class AdvancedPromptTransform(PromptTransform):
                 memory_config=memory_config,
                 memory=memory,
                 model_config=model_config,
-                model_instance=model_instance,
                 image_detail_config=image_detail_config,
             )
 
@@ -93,8 +85,7 @@ class AdvancedPromptTransform(PromptTransform):
         context: str | None,
         memory_config: MemoryConfig | None,
         memory: TokenBufferMemory | None,
-        model_config: ModelConfigWithCredentialsEntity | None = None,
-        model_instance: ModelInstance | None = None,
+        model_config: ModelConfigWithCredentialsEntity,
         image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
         """
@@ -120,7 +111,6 @@ class AdvancedPromptTransform(PromptTransform):
                     parser=parser,
                     prompt_inputs=prompt_inputs,
                     model_config=model_config,
-                    model_instance=model_instance,
                 )
 
             if query:
@@ -156,8 +146,7 @@ class AdvancedPromptTransform(PromptTransform):
         context: str | None,
         memory_config: MemoryConfig | None,
         memory: TokenBufferMemory | None,
-        model_config: ModelConfigWithCredentialsEntity | None = None,
-        model_instance: ModelInstance | None = None,
+        model_config: ModelConfigWithCredentialsEntity,
         image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
         """
@@ -209,13 +198,8 @@ class AdvancedPromptTransform(PromptTransform):
 
         prompt_message_contents: list[PromptMessageContentUnionTypes] = []
         if memory and memory_config:
-            prompt_messages = self._append_chat_histories(
-                memory,
-                memory_config,
-                prompt_messages,
-                model_config=model_config,
-                model_instance=model_instance,
-            )
+            prompt_messages = self._append_chat_histories(memory, memory_config, prompt_messages, model_config)
+
             if files and query is not None:
                 for file in files:
                     prompt_message_contents.append(
@@ -292,8 +276,7 @@ class AdvancedPromptTransform(PromptTransform):
         role_prefix: MemoryConfig.RolePrefix,
         parser: PromptTemplateParser,
         prompt_inputs: Mapping[str, str],
-        model_config: ModelConfigWithCredentialsEntity | None = None,
-        model_instance: ModelInstance | None = None,
+        model_config: ModelConfigWithCredentialsEntity,
     ) -> Mapping[str, str]:
         prompt_inputs = dict(prompt_inputs)
         if "#histories#" in parser.variable_keys:
@@ -303,11 +286,7 @@ class AdvancedPromptTransform(PromptTransform):
                 prompt_inputs = {k: inputs[k] for k in parser.variable_keys if k in inputs}
                 tmp_human_message = UserPromptMessage(content=parser.format(prompt_inputs))
 
-                rest_tokens = self._calculate_rest_token(
-                    [tmp_human_message],
-                    model_config=model_config,
-                    model_instance=model_instance,
-                )
+                rest_tokens = self._calculate_rest_token([tmp_human_message], model_config)
 
                 histories = self._get_history_messages_from_memory(
                     memory=memory,

api/core/prompt/agent_history_prompt_transform.py

@@ -10,7 +10,7 @@ from core.model_runtime.entities.message_entities import (
     UserPromptMessage,
 )
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from core.model_runtime.prompt.prompt_transform import PromptTransform
+from core.prompt.prompt_transform import PromptTransform
 
 
 class AgentHistoryPromptTransform(PromptTransform):
@@ -41,15 +41,13 @@ class AgentHistoryPromptTransform(PromptTransform):
         if not self.memory:
             return prompt_messages
 
-        max_token_limit = self._calculate_rest_token(self.prompt_messages, model_config=self.model_config)
+        max_token_limit = self._calculate_rest_token(self.prompt_messages, self.model_config)
 
         model_type_instance = self.model_config.provider_model_bundle.model_type_instance
         model_type_instance = cast(LargeLanguageModel, model_type_instance)
 
         curr_message_tokens = model_type_instance.get_num_tokens(
-            self.model_config.model,
-            self.model_config.credentials,
-            self.history_messages,
+            self.memory.model_instance.model, self.memory.model_instance.credentials, self.history_messages
         )
         if curr_message_tokens <= max_token_limit:
             return self.history_messages
@@ -65,9 +63,7 @@ class AgentHistoryPromptTransform(PromptTransform):
             # a message is start with UserPromptMessage
             if isinstance(prompt_message, UserPromptMessage):
                 curr_message_tokens = model_type_instance.get_num_tokens(
-                    self.model_config.model,
-                    self.model_config.credentials,
-                    prompt_messages,
+                    self.memory.model_instance.model, self.memory.model_instance.credentials, prompt_messages
                 )
                 # if current message token is overflow, drop all the prompts in current message and break
                 if curr_message_tokens > max_token_limit:

api/core/prompt/prompt_transform.py

@@ -4,83 +4,45 @@ from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEnti
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.message_entities import PromptMessage
-from core.model_runtime.entities.model_entities import AIModelEntity, ModelPropertyKey
-from core.model_runtime.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.model_runtime.entities.model_entities import ModelPropertyKey
+from core.prompt.entities.advanced_prompt_entities import MemoryConfig
 
 
 class PromptTransform:
-    def _resolve_model_runtime(
-        self,
-        *,
-        model_config: ModelConfigWithCredentialsEntity | None = None,
-        model_instance: ModelInstance | None = None,
-    ) -> tuple[ModelInstance, AIModelEntity]:
-        if model_instance is None:
-            if model_config is None:
-                raise ValueError("Either model_config or model_instance must be provided.")
-            model_instance = ModelInstance(
-                provider_model_bundle=model_config.provider_model_bundle, model=model_config.model
-            )
-            model_instance.credentials = model_config.credentials
-            model_instance.parameters = model_config.parameters
-            model_instance.stop = model_config.stop
-
-        model_schema = model_instance.model_type_instance.get_model_schema(
-            model=model_instance.model_name,
-            credentials=model_instance.credentials,
-        )
-        if model_schema is None:
-            if model_config is None:
-                raise ValueError("Model schema not found for the provided model instance.")
-            model_schema = model_config.model_schema
-
-        return model_instance, model_schema
-
     def _append_chat_histories(
         self,
         memory: TokenBufferMemory,
         memory_config: MemoryConfig,
         prompt_messages: list[PromptMessage],
-        *,
-        model_config: ModelConfigWithCredentialsEntity | None = None,
-        model_instance: ModelInstance | None = None,
+        model_config: ModelConfigWithCredentialsEntity,
     ) -> list[PromptMessage]:
-        rest_tokens = self._calculate_rest_token(
-            prompt_messages,
-            model_config=model_config,
-            model_instance=model_instance,
-        )
+        rest_tokens = self._calculate_rest_token(prompt_messages, model_config)
         histories = self._get_history_messages_list_from_memory(memory, memory_config, rest_tokens)
         prompt_messages.extend(histories)
 
         return prompt_messages
 
     def _calculate_rest_token(
-        self,
-        prompt_messages: list[PromptMessage],
-        *,
-        model_config: ModelConfigWithCredentialsEntity | None = None,
-        model_instance: ModelInstance | None = None,
+        self, prompt_messages: list[PromptMessage], model_config: ModelConfigWithCredentialsEntity
     ) -> int:
-        model_instance, model_schema = self._resolve_model_runtime(
-            model_config=model_config,
-            model_instance=model_instance,
-        )
-        model_parameters = model_instance.parameters
         rest_tokens = 2000
 
-        model_context_tokens = model_schema.model_properties.get(ModelPropertyKey.CONTEXT_SIZE)
+        model_context_tokens = model_config.model_schema.model_properties.get(ModelPropertyKey.CONTEXT_SIZE)
         if model_context_tokens:
+            model_instance = ModelInstance(
+                provider_model_bundle=model_config.provider_model_bundle, model=model_config.model
+            )
+
             curr_message_tokens = model_instance.get_llm_num_tokens(prompt_messages)
 
             max_tokens = 0
-            for parameter_rule in model_schema.parameter_rules:
+            for parameter_rule in model_config.model_schema.parameter_rules:
                 if parameter_rule.name == "max_tokens" or (
                     parameter_rule.use_template and parameter_rule.use_template == "max_tokens"
                 ):
                     max_tokens = (
-                        model_parameters.get(parameter_rule.name)
-                        or model_parameters.get(parameter_rule.use_template or "")
+                        model_config.parameters.get(parameter_rule.name)
+                        or model_config.parameters.get(parameter_rule.use_template or "")
                     ) or 0
 
             rest_tokens = model_context_tokens - max_tokens - curr_message_tokens

api/core/prompt/simple_prompt_transform.py

@@ -15,9 +15,9 @@ from core.model_runtime.entities.message_entities import (
     TextPromptMessageContent,
     UserPromptMessage,
 )
-from core.model_runtime.prompt.entities.advanced_prompt_entities import MemoryConfig
-from core.model_runtime.prompt.prompt_transform import PromptTransform
-from core.model_runtime.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.prompt.prompt_transform import PromptTransform
+from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.workflow.file import file_manager
 from models.model import AppMode
 
@@ -252,7 +252,7 @@ class SimplePromptTransform(PromptTransform):
         if memory:
             tmp_human_message = UserPromptMessage(content=prompt)
 
-            rest_tokens = self._calculate_rest_token([tmp_human_message], model_config=model_config)
+            rest_tokens = self._calculate_rest_token([tmp_human_message], model_config)
             histories = self._get_history_messages_from_memory(
                 memory=memory,
                 memory_config=MemoryConfig(

api/core/prompt/utils/get_thread_messages_length.py

@@ -1,6 +1,6 @@
 from sqlalchemy import select
 
-from core.model_runtime.prompt.utils.extract_thread_messages import extract_thread_messages
+from core.prompt.utils.extract_thread_messages import extract_thread_messages
 from extensions.ext_database import db
 from models.model import Message
 

api/core/prompt/utils/prompt_message_util.py

@@ -10,7 +10,7 @@ from core.model_runtime.entities import (
     PromptMessageRole,
     TextPromptMessageContent,
 )
-from core.model_runtime.prompt.simple_prompt_transform import ModelMode
+from core.prompt.simple_prompt_transform import ModelMode
 
 
 class PromptMessageUtil:

api/core/rag/datasource/vdb/analyticdb/analyticdb_vector_openapi.py

@@ -192,8 +192,8 @@ class AnalyticdbVectorOpenAPI:
             collection=self._collection_name,
             metrics=self.config.metrics,
             include_values=True,
-            vector=None,
-            content=None,
+            vector=None,  # ty: ignore [invalid-argument-type]
+            content=None,  # ty: ignore [invalid-argument-type]
             top_k=1,
             filter=f"ref_doc_id='{id}'",
         )
@@ -211,7 +211,7 @@ class AnalyticdbVectorOpenAPI:
             namespace=self.config.namespace,
             namespace_password=self.config.namespace_password,
             collection=self._collection_name,
-            collection_data=None,
+            collection_data=None,  # ty: ignore [invalid-argument-type]
             collection_data_filter=f"ref_doc_id IN {ids_str}",
         )
         self._client.delete_collection_data(request)
@@ -225,7 +225,7 @@ class AnalyticdbVectorOpenAPI:
             namespace=self.config.namespace,
             namespace_password=self.config.namespace_password,
             collection=self._collection_name,
-            collection_data=None,
+            collection_data=None,  # ty: ignore [invalid-argument-type]
             collection_data_filter=f"metadata_ ->> '{key}' = '{value}'",
         )
         self._client.delete_collection_data(request)
@@ -249,7 +249,7 @@ class AnalyticdbVectorOpenAPI:
             include_values=kwargs.pop("include_values", True),
             metrics=self.config.metrics,
             vector=query_vector,
-            content=None,
+            content=None,  # ty: ignore [invalid-argument-type]
             top_k=kwargs.get("top_k", 4),
             filter=where_clause,
         )
@@ -285,7 +285,7 @@ class AnalyticdbVectorOpenAPI:
             collection=self._collection_name,
             include_values=kwargs.pop("include_values", True),
             metrics=self.config.metrics,
-            vector=None,
+            vector=None,  # ty: ignore [invalid-argument-type]
             content=query,
             top_k=kwargs.get("top_k", 4),
             filter=where_clause,

api/core/rag/datasource/vdb/couchbase/couchbase_vector.py

@@ -306,7 +306,7 @@ class CouchbaseVector(BaseVector):
     def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]:
         top_k = kwargs.get("top_k", 4)
         try:
-            CBrequest = search.SearchRequest.create(search.QueryStringQuery("text:" + query))
+            CBrequest = search.SearchRequest.create(search.QueryStringQuery("text:" + query))  # ty: ignore [too-many-positional-arguments]
             search_iter = self._scope.search(
                 self._collection_name + "_search", CBrequest, SearchOptions(limit=top_k, fields=["*"])
             )

api/core/rag/embedding/cached_embedding.py

@@ -35,9 +35,7 @@ class CacheEmbedding(Embeddings):
             embedding = (
                 db.session.query(Embedding)
                 .filter_by(
-                    model_name=self._model_instance.model_name,
-                    hash=hash,
-                    provider_name=self._model_instance.provider,
+                    model_name=self._model_instance.model, hash=hash, provider_name=self._model_instance.provider
                 )
                 .first()
             )
@@ -54,7 +52,7 @@ class CacheEmbedding(Embeddings):
             try:
                 model_type_instance = cast(TextEmbeddingModel, self._model_instance.model_type_instance)
                 model_schema = model_type_instance.get_model_schema(
-                    self._model_instance.model_name, self._model_instance.credentials
+                    self._model_instance.model, self._model_instance.credentials
                 )
                 max_chunks = (
                     model_schema.model_properties[ModelPropertyKey.MAX_CHUNKS]
@@ -89,7 +87,7 @@ class CacheEmbedding(Embeddings):
                         hash = helper.generate_text_hash(texts[i])
                         if hash not in cache_embeddings:
                             embedding_cache = Embedding(
-                                model_name=self._model_instance.model_name,
+                                model_name=self._model_instance.model,
                                 hash=hash,
                                 provider_name=self._model_instance.provider,
                                 embedding=pickle.dumps(n_embedding, protocol=pickle.HIGHEST_PROTOCOL),
@@ -116,9 +114,7 @@ class CacheEmbedding(Embeddings):
             embedding = (
                 db.session.query(Embedding)
                 .filter_by(
-                    model_name=self._model_instance.model_name,
-                    hash=file_id,
-                    provider_name=self._model_instance.provider,
+                    model_name=self._model_instance.model, hash=file_id, provider_name=self._model_instance.provider
                 )
                 .first()
             )
@@ -135,7 +131,7 @@ class CacheEmbedding(Embeddings):
             try:
                 model_type_instance = cast(TextEmbeddingModel, self._model_instance.model_type_instance)
                 model_schema = model_type_instance.get_model_schema(
-                    self._model_instance.model_name, self._model_instance.credentials
+                    self._model_instance.model, self._model_instance.credentials
                 )
                 max_chunks = (
                     model_schema.model_properties[ModelPropertyKey.MAX_CHUNKS]
@@ -172,7 +168,7 @@ class CacheEmbedding(Embeddings):
                         file_id = multimodel_documents[i]["file_id"]
                         if file_id not in cache_embeddings:
                             embedding_cache = Embedding(
-                                model_name=self._model_instance.model_name,
+                                model_name=self._model_instance.model,
                                 hash=file_id,
                                 provider_name=self._model_instance.provider,
                                 embedding=pickle.dumps(n_embedding, protocol=pickle.HIGHEST_PROTOCOL),
@@ -194,7 +190,7 @@ class CacheEmbedding(Embeddings):
         """Embed query text."""
         # use doc embedding cache or store if not exists
         hash = helper.generate_text_hash(text)
-        embedding_cache_key = f"{self._model_instance.provider}_{self._model_instance.model_name}_{hash}"
+        embedding_cache_key = f"{self._model_instance.provider}_{self._model_instance.model}_{hash}"
         embedding = redis_client.get(embedding_cache_key)
         if embedding:
             redis_client.expire(embedding_cache_key, 600)
@@ -237,7 +233,7 @@ class CacheEmbedding(Embeddings):
         """Embed multimodal documents."""
         # use doc embedding cache or store if not exists
         file_id = multimodel_document["file_id"]
-        embedding_cache_key = f"{self._model_instance.provider}_{self._model_instance.model_name}_{file_id}"
+        embedding_cache_key = f"{self._model_instance.provider}_{self._model_instance.model}_{file_id}"
         embedding = redis_client.get(embedding_cache_key)
         if embedding:
             redis_client.expire(embedding_cache_key, 600)

api/core/rag/index_processor/index_processor_base.py

@@ -75,15 +75,15 @@ class BaseIndexProcessor(ABC):
         multimodal_documents: list[AttachmentDocument] | None = None,
         with_keywords: bool = True,
         **kwargs,
-    ) -> None:
+    ):
         raise NotImplementedError
 
     @abstractmethod
-    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
+    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
         raise NotImplementedError
 
     @abstractmethod
-    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
+    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
         raise NotImplementedError
 
     @abstractmethod

api/core/rag/index_processor/processor/paragraph_index_processor.py

@@ -115,7 +115,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
         multimodal_documents: list[AttachmentDocument] | None = None,
         with_keywords: bool = True,
         **kwargs,
-    ) -> None:
+    ):
         if dataset.indexing_technique == "high_quality":
             vector = Vector(dataset)
             vector.create(documents)
@@ -130,7 +130,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             else:
                 keyword.add_texts(documents)
 
-    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
+    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
         # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
         # This method is called for actual deletion scenarios (e.g., when segment is deleted).
         # For disable operations, disable_summaries_for_segments is called directly in the task.
@@ -196,7 +196,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
                 docs.append(doc)
         return docs
 
-    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
+    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
         documents: list[Any] = []
         all_multimodal_documents: list[Any] = []
         if isinstance(chunks, list):
@@ -469,7 +469,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
         if not isinstance(result, LLMResult):
             raise ValueError("Expected LLMResult when stream=False")
 
-        summary_content = result.message.get_text_content()
+        summary_content = getattr(result.message, "content", "")
         usage = result.usage
 
         # Deduct quota for summary generation (same as workflow nodes)

api/core/rag/index_processor/processor/parent_child_index_processor.py

@@ -126,7 +126,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
         multimodal_documents: list[AttachmentDocument] | None = None,
         with_keywords: bool = True,
         **kwargs,
-    ) -> None:
+    ):
         if dataset.indexing_technique == "high_quality":
             vector = Vector(dataset)
             for document in documents:
@@ -139,7 +139,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
             if multimodal_documents and dataset.is_multimodal:
                 vector.create_multimodal(multimodal_documents)
 
-    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
+    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
         # node_ids is segment's node_ids
         # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
         # This method is called for actual deletion scenarios (e.g., when segment is deleted).
@@ -272,7 +272,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                     child_nodes.append(child_document)
         return child_nodes
 
-    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
+    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
         parent_childs = ParentChildStructureChunk.model_validate(chunks)
         documents = []
         for parent_child in parent_childs.parent_child_chunks:

api/core/rag/index_processor/processor/qa_index_processor.py

@@ -139,14 +139,14 @@ class QAIndexProcessor(BaseIndexProcessor):
         multimodal_documents: list[AttachmentDocument] | None = None,
         with_keywords: bool = True,
         **kwargs,
-    ) -> None:
+    ):
         if dataset.indexing_technique == "high_quality":
             vector = Vector(dataset)
             vector.create(documents)
             if multimodal_documents and dataset.is_multimodal:
                 vector.create_multimodal(multimodal_documents)
 
-    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
+    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
         # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
         # This method is called for actual deletion scenarios (e.g., when segment is deleted).
         # For disable operations, disable_summaries_for_segments is called directly in the task.
@@ -206,7 +206,7 @@ class QAIndexProcessor(BaseIndexProcessor):
                 docs.append(doc)
         return docs
 
-    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
+    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
         qa_chunks = QAStructureChunk.model_validate(chunks)
         documents = []
         for qa_chunk in qa_chunks.qa_chunks:

api/core/rag/rerank/rerank_model.py

@@ -38,7 +38,7 @@ class RerankModelRunner(BaseRerankRunner):
         is_support_vision = model_manager.check_model_support_vision(
             tenant_id=self.rerank_model_instance.provider_model_bundle.configuration.tenant_id,
             provider=self.rerank_model_instance.provider,
-            model=self.rerank_model_instance.model_name,
+            model=self.rerank_model_instance.model,
             model_type=ModelType.RERANK,
         )
         if not is_support_vision:

api/core/rag/retrieval/dataset_retrieval.py

@@ -29,12 +29,12 @@ from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
 from core.model_runtime.entities.message_entities import PromptMessage, PromptMessageRole, PromptMessageTool
 from core.model_runtime.entities.model_entities import ModelFeature, ModelType
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from core.model_runtime.prompt.advanced_prompt_transform import AdvancedPromptTransform
-from core.model_runtime.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate
-from core.model_runtime.prompt.simple_prompt_transform import ModelMode
 from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.ops.utils import measure_time
+from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
+from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate
+from core.prompt.simple_prompt_transform import ModelMode
 from core.rag.data_post_processor.data_post_processor import DataPostProcessor
 from core.rag.datasource.keyword.jieba.jieba_keyword_table_handler import JiebaKeywordTableHandler
 from core.rag.datasource.retrieval_service import RetrievalService

api/core/rag/retrieval/router/multi_dataset_react_route.py

@@ -5,8 +5,8 @@ from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEnti
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
 from core.model_runtime.entities.message_entities import PromptMessage, PromptMessageRole, PromptMessageTool
-from core.model_runtime.prompt.advanced_prompt_transform import AdvancedPromptTransform
-from core.model_runtime.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate
+from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
+from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate
 from core.rag.retrieval.output_parser.react_output import ReactAction
 from core.rag.retrieval.output_parser.structured_chat import StructuredChatOutputParser
 from core.workflow.nodes.llm import llm_utils

api/core/tools/utils/model_invocation_utils.py

@@ -47,7 +47,7 @@ class ModelInvocationUtils:
             raise InvokeModelError("Model not found")
 
         llm_model = cast(LargeLanguageModel, model_instance.model_type_instance)
-        schema = llm_model.get_model_schema(model_instance.model_name, model_instance.credentials)
+        schema = llm_model.get_model_schema(model_instance.model, model_instance.credentials)
 
         if not schema:
             raise InvokeModelError("No model schema found")

api/core/variables/variables.py

@@ -4,6 +4,8 @@ from uuid import uuid4
 
 from pydantic import BaseModel, Discriminator, Field, Tag
 
+from core.helper import encrypter
+
 from .segments import (
     ArrayAnySegment,
     ArrayBooleanSegment,
@@ -25,14 +27,6 @@ from .segments import (
 from .types import SegmentType
 
 
-def _obfuscated_token(token: str) -> str:
-    if not token:
-        return token
-    if len(token) <= 8:
-        return "*" * 20
-    return token[:6] + "*" * 12 + token[-2:]
-
-
 class VariableBase(Segment):
     """
     A variable is a segment that has a name.
@@ -92,7 +86,7 @@ class SecretVariable(StringVariable):
 
     @property
     def log(self) -> str:
-        return _obfuscated_token(self.value)
+        return encrypter.obfuscated_token(self.value)
 
 
 class NoneVariable(NoneSegment, VariableBase):

api/core/workflow/conversation_variable_updater.py

@@ -1,7 +1,7 @@
 import abc
 from typing import Protocol
 
-from core.workflow.variables import VariableBase
+from core.variables import VariableBase
 
 
 class ConversationVariableUpdater(Protocol):

api/core/workflow/graph_engine/command_channels/redis_channel.py

@@ -7,28 +7,12 @@ Each instance uses a unique key for its command queue.
 """
 
 import json
-from contextlib import AbstractContextManager
-from typing import Any, Protocol, final
+from typing import TYPE_CHECKING, Any, final
 
 from ..entities.commands import AbortCommand, CommandType, GraphEngineCommand, PauseCommand, UpdateVariablesCommand
 
-
-class RedisPipelineProtocol(Protocol):
-    """Minimal Redis pipeline contract used by the command channel."""
-
-    def lrange(self, name: str, start: int, end: int) -> Any: ...
-    def delete(self, *names: str) -> Any: ...
-    def execute(self) -> list[Any]: ...
-    def rpush(self, name: str, *values: str) -> Any: ...
-    def expire(self, name: str, time: int) -> Any: ...
-    def set(self, name: str, value: str, ex: int | None = None) -> Any: ...
-    def get(self, name: str) -> Any: ...
-
-
-class RedisClientProtocol(Protocol):
-    """Redis client contract required by the command channel."""
-
-    def pipeline(self) -> AbstractContextManager[RedisPipelineProtocol]: ...
+if TYPE_CHECKING:
+    from extensions.ext_redis import RedisClientWrapper
 
 
 @final
@@ -42,7 +26,7 @@ class RedisChannel:
 
     def __init__(
         self,
-        redis_client: RedisClientProtocol,
+        redis_client: "RedisClientWrapper",
         channel_key: str,
         command_ttl: int = 3600,
     ) -> None:

api/core/workflow/graph_engine/entities/commands.py

@@ -11,7 +11,7 @@ from typing import Any
 
 from pydantic import BaseModel, Field
 
-from core.workflow.variables.variables import Variable
+from core.variables.variables import Variable
 
 
 class CommandType(StrEnum):

api/core/workflow/graph_engine/graph_engine.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import logging
 import queue
+import threading
 from collections.abc import Generator
 from typing import TYPE_CHECKING, cast, final
 
@@ -76,10 +77,13 @@ class GraphEngine:
         config: GraphEngineConfig = _DEFAULT_CONFIG,
     ) -> None:
         """Initialize the graph engine with all subsystems and dependencies."""
+        # stop event
+        self._stop_event = threading.Event()
 
         # Bind runtime state to current workflow context
         self._graph = graph
         self._graph_runtime_state = graph_runtime_state
+        self._graph_runtime_state.stop_event = self._stop_event
         self._graph_runtime_state.configure(graph=cast("GraphProtocol", graph))
         self._command_channel = command_channel
         self._config = config
@@ -159,6 +163,7 @@ class GraphEngine:
             layers=self._layers,
             execution_context=execution_context,
             config=self._config,
+            stop_event=self._stop_event,
         )
 
         # === Orchestration ===
@@ -189,6 +194,7 @@ class GraphEngine:
             event_handler=self._event_handler_registry,
             execution_coordinator=self._execution_coordinator,
             event_emitter=self._event_manager,
+            stop_event=self._stop_event,
         )
 
         # === Validation ===
@@ -308,6 +314,7 @@ class GraphEngine:
 
     def _start_execution(self, *, resume: bool = False) -> None:
         """Start execution subsystems."""
+        self._stop_event.clear()
         paused_nodes: list[str] = []
         deferred_nodes: list[str] = []
         if resume:
@@ -341,6 +348,7 @@ class GraphEngine:
 
     def _stop_execution(self) -> None:
         """Stop execution subsystems."""
+        self._stop_event.set()
         self._dispatcher.stop()
         self._worker_pool.stop()
         # Don't mark complete here as the dispatcher already does it

api/core/workflow/graph_engine/manager.py

@@ -3,14 +3,13 @@ GraphEngine Manager for sending control commands via Redis channel.
 
 This module provides a simplified interface for controlling workflow executions
 using the new Redis command channel, without requiring user permission checks.
-Callers must provide a Redis client dependency from outside the workflow package.
 """
 
 import logging
 from collections.abc import Sequence
 from typing import final
 
-from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel, RedisClientProtocol
+from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
 from core.workflow.graph_engine.entities.commands import (
     AbortCommand,
     GraphEngineCommand,
@@ -18,6 +17,7 @@ from core.workflow.graph_engine.entities.commands import (
     UpdateVariablesCommand,
     VariableUpdate,
 )
+from extensions.ext_redis import redis_client
 
 logger = logging.getLogger(__name__)
 
@@ -31,12 +31,8 @@ class GraphEngineManager:
     by sending commands through Redis channels, without user validation.
     """
 
-    _redis_client: RedisClientProtocol
-
-    def __init__(self, redis_client: RedisClientProtocol) -> None:
-        self._redis_client = redis_client
-
-    def send_stop_command(self, task_id: str, reason: str | None = None) -> None:
+    @staticmethod
+    def send_stop_command(task_id: str, reason: str | None = None) -> None:
         """
         Send a stop command to a running workflow.
 
@@ -45,31 +41,34 @@ class GraphEngineManager:
             reason: Optional reason for stopping (defaults to "User requested stop")
         """
         abort_command = AbortCommand(reason=reason or "User requested stop")
-        self._send_command(task_id, abort_command)
+        GraphEngineManager._send_command(task_id, abort_command)
 
-    def send_pause_command(self, task_id: str, reason: str | None = None) -> None:
+    @staticmethod
+    def send_pause_command(task_id: str, reason: str | None = None) -> None:
         """Send a pause command to a running workflow."""
 
         pause_command = PauseCommand(reason=reason or "User requested pause")
-        self._send_command(task_id, pause_command)
+        GraphEngineManager._send_command(task_id, pause_command)
 
-    def send_update_variables_command(self, task_id: str, updates: Sequence[VariableUpdate]) -> None:
+    @staticmethod
+    def send_update_variables_command(task_id: str, updates: Sequence[VariableUpdate]) -> None:
         """Send a command to update variables in a running workflow."""
 
         if not updates:
             return
 
         update_command = UpdateVariablesCommand(updates=updates)
-        self._send_command(task_id, update_command)
+        GraphEngineManager._send_command(task_id, update_command)
 
-    def _send_command(self, task_id: str, command: GraphEngineCommand) -> None:
+    @staticmethod
+    def _send_command(task_id: str, command: GraphEngineCommand) -> None:
         """Send a command to the workflow-specific Redis channel."""
 
         if not task_id:
             return
 
         channel_key = f"workflow:{task_id}:commands"
-        channel = RedisChannel(self._redis_client, channel_key)
+        channel = RedisChannel(redis_client, channel_key)
 
         try:
             channel.send_command(command)

api/core/workflow/graph_engine/orchestration/dispatcher.py

@@ -44,6 +44,7 @@ class Dispatcher:
         event_queue: queue.Queue[GraphNodeEventBase],
         event_handler: "EventHandler",
         execution_coordinator: ExecutionCoordinator,
+        stop_event: threading.Event,
         event_emitter: EventManager | None = None,
     ) -> None:
         """
@@ -61,7 +62,7 @@ class Dispatcher:
         self._event_emitter = event_emitter
 
         self._thread: threading.Thread | None = None
-        self._stop_event = threading.Event()
+        self._stop_event = stop_event
         self._start_time: float | None = None
 
     def start(self) -> None:
@@ -69,14 +70,12 @@ class Dispatcher:
         if self._thread and self._thread.is_alive():
             return
 
-        self._stop_event.clear()
         self._start_time = time.time()
         self._thread = threading.Thread(target=self._dispatcher_loop, name="GraphDispatcher", daemon=True)
         self._thread.start()
 
     def stop(self) -> None:
         """Stop the dispatcher thread."""
-        self._stop_event.set()
         if self._thread and self._thread.is_alive():
             self._thread.join(timeout=2.0)
 

api/core/workflow/graph_engine/worker.py

@@ -42,6 +42,7 @@ class Worker(threading.Thread):
         event_queue: queue.Queue[GraphNodeEventBase],
         graph: Graph,
         layers: Sequence[GraphEngineLayer],
+        stop_event: threading.Event,
         worker_id: int = 0,
         execution_context: IExecutionContext | None = None,
     ) -> None:
@@ -62,13 +63,16 @@ class Worker(threading.Thread):
         self._graph = graph
         self._worker_id = worker_id
         self._execution_context = execution_context
-        self._stop_event = threading.Event()
+        self._stop_event = stop_event
         self._layers = layers if layers is not None else []
         self._last_task_time = time.time()
 
     def stop(self) -> None:
-        """Signal the worker to stop processing."""
-        self._stop_event.set()
+        """Worker is controlled via shared stop_event from GraphEngine.
+
+        This method is a no-op retained for backward compatibility.
+        """
+        pass
 
     @property
     def is_idle(self) -> bool:

api/core/workflow/graph_engine/worker_management/worker_pool.py

@@ -37,6 +37,7 @@ class WorkerPool:
         event_queue: queue.Queue[GraphNodeEventBase],
         graph: Graph,
         layers: list[GraphEngineLayer],
+        stop_event: threading.Event,
         config: GraphEngineConfig,
         execution_context: IExecutionContext | None = None,
     ) -> None:
@@ -63,6 +64,7 @@ class WorkerPool:
         self._worker_counter = 0
         self._lock = threading.RLock()
         self._running = False
+        self._stop_event = stop_event
 
         # No longer tracking worker states with callbacks to avoid lock contention
 
@@ -133,6 +135,7 @@ class WorkerPool:
             layers=self._layers,
             worker_id=worker_id,
             execution_context=self._execution_context,
+            stop_event=self._stop_event,
         )
 
         worker.start()

api/core/workflow/nodes/agent/agent_node.py

@@ -25,6 +25,7 @@ from core.tools.entities.tool_entities import (
 )
 from core.tools.tool_manager import ToolManager
 from core.tools.utils.message_transformer import ToolFileMessageTransformer
+from core.variables.segments import ArrayFileSegment, StringSegment
 from core.workflow.enums import (
     NodeType,
     SystemVariableKey,
@@ -43,7 +44,6 @@ from core.workflow.nodes.agent.entities import AgentNodeData, AgentOldVersionMod
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.variable_template_parser import VariableTemplateParser
 from core.workflow.runtime import VariablePool
-from core.workflow.variables.segments import ArrayFileSegment, StringSegment
 from extensions.ext_database import db
 from factories import file_factory
 from factories.agent_factory import get_plugin_agent_strategy

api/core/workflow/nodes/agent/entities.py

@@ -3,7 +3,7 @@ from typing import Any, Literal, Union
 
 from pydantic import BaseModel
 
-from core.model_runtime.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.prompt.entities.advanced_prompt_entities import MemoryConfig
 from core.tools.entities.tool_entities import ToolSelector
 from core.workflow.nodes.base.entities import BaseNodeData
 

api/core/workflow/nodes/answer/answer_node.py

@@ -1,13 +1,13 @@
 from collections.abc import Mapping, Sequence
 from typing import Any
 
+from core.variables import ArrayFileSegment, FileSegment, Segment
 from core.workflow.enums import NodeExecutionType, NodeType, WorkflowNodeExecutionStatus
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.answer.entities import AnswerNodeData
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.template import Template
 from core.workflow.nodes.base.variable_template_parser import VariableTemplateParser
-from core.workflow.variables import ArrayFileSegment, FileSegment, Segment
 
 
 class AnswerNode(Node[AnswerNodeData]):

api/core/workflow/nodes/base/node.py

@@ -302,6 +302,10 @@ class Node(Generic[NodeDataT]):
         """
         raise NotImplementedError
 
+    def _should_stop(self) -> bool:
+        """Check if execution should be stopped."""
+        return self.graph_runtime_state.stop_event.is_set()
+
     def run(self) -> Generator[GraphNodeEventBase, None, None]:
         execution_id = self.ensure_execution_id()
         self._start_at = naive_utc_now()
@@ -370,6 +374,21 @@ class Node(Generic[NodeDataT]):
                     yield event
                 else:
                     yield event
+
+                if self._should_stop():
+                    error_message = "Execution cancelled"
+                    yield NodeRunFailedEvent(
+                        id=self.execution_id,
+                        node_id=self._node_id,
+                        node_type=self.node_type,
+                        start_at=self._start_at,
+                        node_run_result=NodeRunResult(
+                            status=WorkflowNodeExecutionStatus.FAILED,
+                            error=error_message,
+                        ),
+                        error=error_message,
+                    )
+                    return
         except Exception as e:
             logger.exception("Node %s failed to run", self._node_id)
             result = NodeRunResult(