fix: list multiple @mentions

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: 非法操作 <hjlarry@163.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

.agent/skills

@@ -0,0 +1 @@
+../.claude/skills

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

@@ -0,0 +1,46 @@
+---
+name: orpc-contract-first
+description: Guide for implementing oRPC contract-first API patterns in Dify frontend. Triggers when creating new API contracts, adding service endpoints, integrating TanStack Query with typed contracts, or migrating legacy service calls to oRPC. Use for all API layer work in web/contract and web/service directories.
+---
+
+# oRPC Contract-First Development
+
+## Project Structure
+
+```
+web/contract/
+├── base.ts           # Base contract (inputStructure: 'detailed')
+├── router.ts         # Router composition & type exports
+├── marketplace.ts    # Marketplace contracts
+└── console/          # Console contracts by domain
+    ├── system.ts
+    └── billing.ts
+```
+
+## Workflow
+
+1. **Create contract** in `web/contract/console/{domain}.ts`
+   - Import `base` from `../base` and `type` from `@orpc/contract`
+   - Define route with `path`, `method`, `input`, `output`
+
+2. **Register in router** at `web/contract/router.ts`
+   - Import directly from domain file (no barrel files)
+   - Nest by API prefix: `billing: { invoices, bindPartnerStack }`
+
+3. **Create hooks** in `web/service/use-{domain}.ts`
+   - Use `consoleQuery.{group}.{contract}.queryKey()` for query keys
+   - Use `consoleClient.{group}.{contract}()` for API calls
+
+## Key Rules
+
+- **Input structure**: Always use `{ params, query?, body? }` format
+- **Path params**: Use `{paramName}` in path, match in `params` object
+- **Router nesting**: Group by API prefix (e.g., `/billing/*` → `billing: {}`)
+- **No barrel files**: Import directly from specific files
+- **Types**: Import from `@/types/`, use `type<T>()` helper
+
+## Type Export
+
+```typescript
+export type ConsoleInputs = InferContractRouterInputs<typeof consoleRouterContract>
+```

.github/workflows/style.yml

@@ -90,7 +90,7 @@ jobs:
         uses: actions/setup-node@v6
         if: steps.changed-files.outputs.any_changed == 'true'
         with:
-          node-version: 22
+          node-version: 24
           cache: pnpm
           cache-dependency-path: ./web/pnpm-lock.yaml
 

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

@@ -16,10 +16,6 @@ jobs:
     name: unit test for Node.js SDK
     runs-on: ubuntu-latest
 
-    strategy:
-      matrix:
-        node-version: [16, 18, 20, 22]
-
     defaults:
       run:
         working-directory: sdks/nodejs-client
@@ -29,10 +25,10 @@ jobs:
         with:
           persist-credentials: false
 
-      - name: Use Node.js ${{ matrix.node-version }}
+      - name: Use Node.js
         uses: actions/setup-node@v6
         with:
-          node-version: ${{ matrix.node-version }}
+          node-version: 24
           cache: ''
           cache-dependency-path: 'pnpm-lock.yaml'
 

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

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

.github/workflows/web-tests.yml

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

.gitignore

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

api/.env.example

@@ -417,6 +417,8 @@ SMTP_USERNAME=123
 SMTP_PASSWORD=abc
 SMTP_USE_TLS=true
 SMTP_OPPORTUNISTIC_TLS=false
+# Optional: override the local hostname used for SMTP HELO/EHLO
+SMTP_LOCAL_HOSTNAME=
 # Sendgid configuration
 SENDGRID_API_KEY=
 # Sentry configuration
@@ -713,3 +715,4 @@ ANNOTATION_IMPORT_MAX_CONCURRENT=5
 SANDBOX_EXPIRED_RECORDS_CLEAN_GRACEFUL_PERIOD=21
 SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_SIZE=1000
 SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS=30
+

api/commands.py

@@ -3,6 +3,7 @@ import datetime
 import json
 import logging
 import secrets
+import time
 from typing import Any
 
 import click
@@ -46,6 +47,8 @@ from services.clear_free_plan_tenant_expired_logs import ClearFreePlanTenantExpi
 from services.plugin.data_migration import PluginDataMigration
 from services.plugin.plugin_migration import PluginMigration
 from services.plugin.plugin_service import PluginService
+from services.retention.conversation.messages_clean_policy import create_message_clean_policy
+from services.retention.conversation.messages_clean_service import MessagesCleanService
 from services.retention.workflow_run.clear_free_plan_expired_workflow_run_logs import WorkflowRunCleanup
 from tasks.remove_app_and_related_data_task import delete_draft_variables_batch
 
@@ -2172,3 +2175,79 @@ def migrate_oss(
             except Exception as e:
                 db.session.rollback()
                 click.echo(click.style(f"Failed to update DB storage_type: {str(e)}", fg="red"))
+
+
+@click.command("clean-expired-messages", help="Clean expired messages.")
+@click.option(
+    "--start-from",
+    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    required=True,
+    help="Lower bound (inclusive) for created_at.",
+)
+@click.option(
+    "--end-before",
+    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    required=True,
+    help="Upper bound (exclusive) for created_at.",
+)
+@click.option("--batch-size", default=1000, show_default=True, help="Batch size for selecting messages.")
+@click.option(
+    "--graceful-period",
+    default=21,
+    show_default=True,
+    help="Graceful period in days after subscription expiration, will be ignored when billing is disabled.",
+)
+@click.option("--dry-run", is_flag=True, default=False, help="Show messages logs would be cleaned without deleting")
+def clean_expired_messages(
+    batch_size: int,
+    graceful_period: int,
+    start_from: datetime.datetime,
+    end_before: datetime.datetime,
+    dry_run: bool,
+):
+    """
+    Clean expired messages and related data for tenants based on clean policy.
+    """
+    click.echo(click.style("clean_messages: start clean messages.", fg="green"))
+
+    start_at = time.perf_counter()
+
+    try:
+        # Create policy based on billing configuration
+        # NOTE: graceful_period will be ignored when billing is disabled.
+        policy = create_message_clean_policy(graceful_period_days=graceful_period)
+
+        # Create and run the cleanup service
+        service = MessagesCleanService.from_time_range(
+            policy=policy,
+            start_from=start_from,
+            end_before=end_before,
+            batch_size=batch_size,
+            dry_run=dry_run,
+        )
+        stats = service.run()
+
+        end_at = time.perf_counter()
+        click.echo(
+            click.style(
+                f"clean_messages: completed successfully\n"
+                f"  - Latency: {end_at - start_at:.2f}s\n"
+                f"  - Batches processed: {stats['batches']}\n"
+                f"  - Total messages scanned: {stats['total_messages']}\n"
+                f"  - Messages filtered: {stats['filtered_messages']}\n"
+                f"  - Messages deleted: {stats['total_deleted']}",
+                fg="green",
+            )
+        )
+    except Exception as e:
+        end_at = time.perf_counter()
+        logger.exception("clean_messages failed")
+        click.echo(
+            click.style(
+                f"clean_messages: failed after {end_at - start_at:.2f}s - {str(e)}",
+                fg="red",
+            )
+        )
+        raise
+
+    click.echo(click.style("messages cleanup completed.", fg="green"))

api/configs/feature/__init__.py

@@ -949,6 +949,12 @@ class MailConfig(BaseSettings):
         default=False,
     )
 
+    SMTP_LOCAL_HOSTNAME: str | None = Field(
+        description="Override the local hostname used in SMTP HELO/EHLO. "
+        "Useful behind NAT or when the default hostname causes rejections.",
+        default=None,
+    )
+
     EMAIL_SEND_IP_LIMIT_PER_MINUTE: PositiveInt = Field(
         description="Maximum number of emails allowed to be sent from the same IP address in a minute",
         default=50,

api/configs/middleware/storage/volcengine_tos_storage_config.py

@@ -4,7 +4,7 @@ from pydantic_settings import BaseSettings
 
 class VolcengineTOSStorageConfig(BaseSettings):
     """
-    Configuration settings for Volcengine Tinder Object Storage (TOS)
+    Configuration settings for Volcengine Torch Object Storage (TOS)
     """
 
     VOLCENGINE_TOS_BUCKET_NAME: str | None = Field(

api/controllers/console/app/conversation.py

@@ -592,9 +592,12 @@ def _get_conversation(app_model, conversation_id):
     if not conversation:
         raise NotFound("Conversation Not Exists.")
 
-    if not conversation.read_at:
-        conversation.read_at = naive_utc_now()
-        conversation.read_account_id = current_user.id
-        db.session.commit()
+    db.session.execute(
+        sa.update(Conversation)
+        .where(Conversation.id == conversation_id, Conversation.read_at.is_(None))
+        .values(read_at=naive_utc_now(), read_account_id=current_user.id)
+    )
+    db.session.commit()
+    db.session.refresh(conversation)
 
     return conversation

api/controllers/console/app/generator.py

@@ -55,6 +55,35 @@ class InstructionTemplatePayload(BaseModel):
     type: str = Field(..., description="Instruction template type")
 
 
+class ContextGeneratePayload(BaseModel):
+    """Payload for generating extractor code node."""
+
+    workflow_id: str = Field(..., description="Workflow ID")
+    node_id: str = Field(..., description="Current tool/llm node ID")
+    parameter_name: str = Field(..., description="Parameter name to generate code for")
+    language: str = Field(default="python3", description="Code language (python3/javascript)")
+    prompt_messages: list[dict[str, Any]] = Field(
+        ..., description="Multi-turn conversation history, last message is the current instruction"
+    )
+    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
+
+
+class SuggestedQuestionsPayload(BaseModel):
+    """Payload for generating suggested questions."""
+
+    workflow_id: str = Field(..., description="Workflow ID")
+    node_id: str = Field(..., description="Current tool/llm node ID")
+    parameter_name: str = Field(..., description="Parameter name")
+    language: str = Field(
+        default="English", description="Language for generated questions (e.g. English, Chinese, Japanese)"
+    )
+    model_config_data: dict[str, Any] | None = Field(
+        default=None,
+        alias="model_config",
+        description="Model configuration (optional, uses system default if not provided)",
+    )
+
+
 def reg(cls: type[BaseModel]):
     console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
 
@@ -64,6 +93,8 @@ reg(RuleCodeGeneratePayload)
 reg(RuleStructuredOutputPayload)
 reg(InstructionGeneratePayload)
 reg(InstructionTemplatePayload)
+reg(ContextGeneratePayload)
+reg(SuggestedQuestionsPayload)
 
 
 @console_ns.route("/rule-generate")
@@ -278,3 +309,74 @@ class InstructionGenerationTemplateApi(Resource):
                 return {"data": INSTRUCTION_GENERATE_TEMPLATE_CODE}
             case _:
                 raise ValueError(f"Invalid type: {args.type}")
+
+
+@console_ns.route("/context-generate")
+class ContextGenerateApi(Resource):
+    @console_ns.doc("generate_with_context")
+    @console_ns.doc(description="Generate with multi-turn conversation context")
+    @console_ns.expect(console_ns.models[ContextGeneratePayload.__name__])
+    @console_ns.response(200, "Content generated successfully")
+    @console_ns.response(400, "Invalid request parameters or workflow not found")
+    @console_ns.response(402, "Provider quota exceeded")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        from core.llm_generator.utils import deserialize_prompt_messages
+
+        args = ContextGeneratePayload.model_validate(console_ns.payload)
+        _, current_tenant_id = current_account_with_tenant()
+
+        prompt_messages = deserialize_prompt_messages(args.prompt_messages)
+
+        try:
+            return LLMGenerator.generate_with_context(
+                tenant_id=current_tenant_id,
+                workflow_id=args.workflow_id,
+                node_id=args.node_id,
+                parameter_name=args.parameter_name,
+                language=args.language,
+                prompt_messages=prompt_messages,
+                model_config=args.model_config_data,
+            )
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+
+
+@console_ns.route("/context-generate/suggested-questions")
+class SuggestedQuestionsApi(Resource):
+    @console_ns.doc("generate_suggested_questions")
+    @console_ns.doc(description="Generate suggested questions for context generation")
+    @console_ns.expect(console_ns.models[SuggestedQuestionsPayload.__name__])
+    @console_ns.response(200, "Questions generated successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        args = SuggestedQuestionsPayload.model_validate(console_ns.payload)
+        _, current_tenant_id = current_account_with_tenant()
+
+        try:
+            return LLMGenerator.generate_suggested_questions(
+                tenant_id=current_tenant_id,
+                workflow_id=args.workflow_id,
+                node_id=args.node_id,
+                parameter_name=args.parameter_name,
+                language=args.language,
+                model_config=args.model_config_data,
+            )
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)

api/controllers/console/datasets/datasets_document.py

@@ -7,7 +7,7 @@ from typing import Literal, cast
 import sqlalchemy as sa
 from flask import request
 from flask_restx import Resource, fields, marshal, marshal_with
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from sqlalchemy import asc, desc, select
 from werkzeug.exceptions import Forbidden, NotFound
 
@@ -104,6 +104,15 @@ class DocumentRenamePayload(BaseModel):
     name: str
 
 
+class DocumentDatasetListParam(BaseModel):
+    page: int = Field(1, title="Page", description="Page number.")
+    limit: int = Field(20, title="Limit", description="Page size.")
+    search: str | None = Field(None, alias="keyword", title="Search", description="Search keyword.")
+    sort_by: str = Field("-created_at", alias="sort", title="SortBy", description="Sort by field.")
+    status: str | None = Field(None, title="Status", description="Document status.")
+    fetch_val: str = Field("false", alias="fetch")
+
+
 register_schema_models(
     console_ns,
     KnowledgeConfig,
@@ -225,14 +234,16 @@ class DatasetDocumentListApi(Resource):
     def get(self, dataset_id):
         current_user, current_tenant_id = current_account_with_tenant()
         dataset_id = str(dataset_id)
-        page = request.args.get("page", default=1, type=int)
-        limit = request.args.get("limit", default=20, type=int)
-        search = request.args.get("keyword", default=None, type=str)
-        sort = request.args.get("sort", default="-created_at", type=str)
-        status = request.args.get("status", default=None, type=str)
+        raw_args = request.args.to_dict()
+        param = DocumentDatasetListParam.model_validate(raw_args)
+        page = param.page
+        limit = param.limit
+        search = param.search
+        sort = param.sort_by
+        status = param.status
         # "yes", "true", "t", "y", "1" convert to True, while others convert to False.
         try:
-            fetch_val = request.args.get("fetch", default="false")
+            fetch_val = param.fetch_val
             if isinstance(fetch_val, bool):
                 fetch = fetch_val
             else:

api/controllers/console/datasets/external.py

@@ -81,7 +81,7 @@ class ExternalKnowledgeApiPayload(BaseModel):
 class ExternalDatasetCreatePayload(BaseModel):
     external_knowledge_api_id: str
     external_knowledge_id: str
-    name: str = Field(..., min_length=1, max_length=40)
+    name: str = Field(..., min_length=1, max_length=100)
     description: str | None = Field(None, max_length=400)
     external_retrieval_model: dict[str, object] | None = None
 

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

@@ -24,7 +24,7 @@ from core.app.layers.conversation_variable_persist_layer import ConversationVari
 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 VariableUnion
+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
@@ -149,8 +149,8 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                 system_variables=system_inputs,
                 user_inputs=inputs,
                 environment_variables=self._workflow.environment_variables,
-                # Based on the definition of `VariableUnion`,
-                # `list[Variable]` can be safely used as `list[VariableUnion]` since they are compatible.
+                # Based on the definition of `Variable`,
+                # `VariableBase` instances can be safely used as `Variable` since they are compatible.
                 conversation_variables=conversation_variables,
             )
 
@@ -318,7 +318,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             trace_manager=app_generate_entity.trace_manager,
         )
 
-    def _initialize_conversation_variables(self) -> list[VariableUnion]:
+    def _initialize_conversation_variables(self) -> list[Variable]:
         """
         Initialize conversation variables for the current conversation.
 
@@ -343,7 +343,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             conversation_variables = [var.to_variable() for var in existing_variables]
 
             session.commit()
-            return cast(list[VariableUnion], conversation_variables)
+            return cast(list[Variable], conversation_variables)
 
     def _load_existing_conversation_variables(self, session: Session) -> list[ConversationVariable]:
         """

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

@@ -82,7 +82,7 @@ class AdvancedChatAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json"))
+                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
             yield response_chunk
 
     @classmethod
@@ -110,7 +110,7 @@ class AdvancedChatAppGenerateResponseConverter(AppGenerateResponseConverter):
             }
 
             if isinstance(sub_stream_response, MessageEndStreamResponse):
-                sub_stream_response_dict = sub_stream_response.model_dump(mode="json")
+                sub_stream_response_dict = sub_stream_response.model_dump(mode="json", exclude_none=True)
                 metadata = sub_stream_response_dict.get("metadata", {})
                 sub_stream_response_dict["metadata"] = cls._get_simple_metadata(metadata)
                 response_chunk.update(sub_stream_response_dict)
@@ -120,6 +120,6 @@ class AdvancedChatAppGenerateResponseConverter(AppGenerateResponseConverter):
             elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
                 response_chunk.update(sub_stream_response.to_ignore_detail_dict())
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json"))
+                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
 
             yield response_chunk

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

@@ -81,7 +81,7 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json"))
+                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
             yield response_chunk
 
     @classmethod
@@ -109,7 +109,7 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter):
             }
 
             if isinstance(sub_stream_response, MessageEndStreamResponse):
-                sub_stream_response_dict = sub_stream_response.model_dump(mode="json")
+                sub_stream_response_dict = sub_stream_response.model_dump(mode="json", exclude_none=True)
                 metadata = sub_stream_response_dict.get("metadata", {})
                 sub_stream_response_dict["metadata"] = cls._get_simple_metadata(metadata)
                 response_chunk.update(sub_stream_response_dict)
@@ -117,6 +117,6 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json"))
+                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
 
             yield response_chunk

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

@@ -81,7 +81,7 @@ class ChatAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json"))
+                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
             yield response_chunk
 
     @classmethod
@@ -109,7 +109,7 @@ class ChatAppGenerateResponseConverter(AppGenerateResponseConverter):
             }
 
             if isinstance(sub_stream_response, MessageEndStreamResponse):
-                sub_stream_response_dict = sub_stream_response.model_dump(mode="json")
+                sub_stream_response_dict = sub_stream_response.model_dump(mode="json", exclude_none=True)
                 metadata = sub_stream_response_dict.get("metadata", {})
                 sub_stream_response_dict["metadata"] = cls._get_simple_metadata(metadata)
                 response_chunk.update(sub_stream_response_dict)
@@ -117,6 +117,6 @@ class ChatAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json"))
+                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
 
             yield response_chunk

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

@@ -70,6 +70,8 @@ class _NodeSnapshot:
     """Empty string means the node is not executing inside an iteration."""
     loop_id: str = ""
     """Empty string means the node is not executing inside a loop."""
+    mention_parent_id: str = ""
+    """Empty string means the node is not an extractor node."""
 
 
 class WorkflowResponseConverter:
@@ -131,6 +133,7 @@ class WorkflowResponseConverter:
             start_at=event.start_at,
             iteration_id=event.in_iteration_id or "",
             loop_id=event.in_loop_id or "",
+            mention_parent_id=event.in_mention_parent_id or "",
         )
         node_execution_id = NodeExecutionId(event.node_execution_id)
         self._node_snapshots[node_execution_id] = snapshot
@@ -287,6 +290,7 @@ class WorkflowResponseConverter:
                 created_at=int(snapshot.start_at.timestamp()),
                 iteration_id=event.in_iteration_id,
                 loop_id=event.in_loop_id,
+                mention_parent_id=event.in_mention_parent_id,
                 agent_strategy=event.agent_strategy,
             ),
         )
@@ -373,6 +377,7 @@ class WorkflowResponseConverter:
                 files=self.fetch_files_from_node_outputs(event.outputs or {}),
                 iteration_id=event.in_iteration_id,
                 loop_id=event.in_loop_id,
+                mention_parent_id=event.in_mention_parent_id,
             ),
         )
 
@@ -422,6 +427,7 @@ class WorkflowResponseConverter:
                 files=self.fetch_files_from_node_outputs(event.outputs or {}),
                 iteration_id=event.in_iteration_id,
                 loop_id=event.in_loop_id,
+                mention_parent_id=event.in_mention_parent_id,
                 retry_index=event.retry_index,
             ),
         )

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

@@ -79,7 +79,7 @@ class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json"))
+                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
             yield response_chunk
 
     @classmethod
@@ -106,7 +106,7 @@ class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter):
             }
 
             if isinstance(sub_stream_response, MessageEndStreamResponse):
-                sub_stream_response_dict = sub_stream_response.model_dump(mode="json")
+                sub_stream_response_dict = sub_stream_response.model_dump(mode="json", exclude_none=True)
                 metadata = sub_stream_response_dict.get("metadata", {})
                 if not isinstance(metadata, dict):
                     metadata = {}
@@ -116,6 +116,6 @@ class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json"))
+                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
 
             yield response_chunk

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

@@ -60,7 +60,7 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(cast(dict, data))
             else:
-                response_chunk.update(sub_stream_response.model_dump())
+                response_chunk.update(sub_stream_response.model_dump(exclude_none=True))
             yield response_chunk
 
     @classmethod
@@ -91,5 +91,5 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
             elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
                 response_chunk.update(cast(dict, sub_stream_response.to_ignore_detail_dict()))
             else:
-                response_chunk.update(sub_stream_response.model_dump())
+                response_chunk.update(sub_stream_response.model_dump(exclude_none=True))
             yield response_chunk

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

@@ -60,7 +60,7 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json"))
+                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
             yield response_chunk
 
     @classmethod
@@ -91,5 +91,5 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
             elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
                 response_chunk.update(sub_stream_response.to_ignore_detail_dict())
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json"))
+                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
             yield response_chunk

api/core/app/apps/workflow_app_runner.py

@@ -385,6 +385,7 @@ class WorkflowBasedAppRunner:
                     start_at=event.start_at,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
+                    in_mention_parent_id=event.in_mention_parent_id,
                     inputs=inputs,
                     process_data=process_data,
                     outputs=outputs,
@@ -405,6 +406,7 @@ class WorkflowBasedAppRunner:
                     start_at=event.start_at,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
+                    in_mention_parent_id=event.in_mention_parent_id,
                     agent_strategy=event.agent_strategy,
                     provider_type=event.provider_type,
                     provider_id=event.provider_id,
@@ -428,6 +430,7 @@ class WorkflowBasedAppRunner:
                     execution_metadata=execution_metadata,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
+                    in_mention_parent_id=event.in_mention_parent_id,
                 )
             )
         elif isinstance(event, NodeRunFailedEvent):
@@ -444,6 +447,7 @@ class WorkflowBasedAppRunner:
                     execution_metadata=event.node_run_result.metadata,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
+                    in_mention_parent_id=event.in_mention_parent_id,
                 )
             )
         elif isinstance(event, NodeRunExceptionEvent):
@@ -460,6 +464,7 @@ class WorkflowBasedAppRunner:
                     execution_metadata=event.node_run_result.metadata,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
+                    in_mention_parent_id=event.in_mention_parent_id,
                 )
             )
         elif isinstance(event, NodeRunStreamChunkEvent):
@@ -469,6 +474,7 @@ class WorkflowBasedAppRunner:
                     from_variable_selector=list(event.selector),
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
+                    in_mention_parent_id=event.in_mention_parent_id,
                 )
             )
         elif isinstance(event, NodeRunRetrieverResourceEvent):
@@ -477,6 +483,7 @@ class WorkflowBasedAppRunner:
                     retriever_resources=event.retriever_resources,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
+                    in_mention_parent_id=event.in_mention_parent_id,
                 )
             )
         elif isinstance(event, NodeRunAgentLogEvent):

api/core/app/entities/queue_entities.py

@@ -190,6 +190,8 @@ class QueueTextChunkEvent(AppQueueEvent):
     """iteration id if node is in iteration"""
     in_loop_id: str | None = None
     """loop id if node is in loop"""
+    in_mention_parent_id: str | None = None
+    """parent node id if this is an extractor node event"""
 
 
 class QueueAgentMessageEvent(AppQueueEvent):
@@ -229,6 +231,8 @@ class QueueRetrieverResourcesEvent(AppQueueEvent):
     """iteration id if node is in iteration"""
     in_loop_id: str | None = None
     """loop id if node is in loop"""
+    in_mention_parent_id: str | None = None
+    """parent node id if this is an extractor node event"""
 
 
 class QueueAnnotationReplyEvent(AppQueueEvent):
@@ -306,6 +310,8 @@ class QueueNodeStartedEvent(AppQueueEvent):
     node_run_index: int = 1  # FIXME(-LAN-): may not used
     in_iteration_id: str | None = None
     in_loop_id: str | None = None
+    in_mention_parent_id: str | None = None
+    """parent node id if this is an extractor node event"""
     start_at: datetime
     agent_strategy: AgentNodeStrategyInit | None = None
 
@@ -328,6 +334,8 @@ class QueueNodeSucceededEvent(AppQueueEvent):
     """iteration id if node is in iteration"""
     in_loop_id: str | None = None
     """loop id if node is in loop"""
+    in_mention_parent_id: str | None = None
+    """parent node id if this is an extractor node event"""
     start_at: datetime
 
     inputs: Mapping[str, object] = Field(default_factory=dict)
@@ -383,6 +391,8 @@ class QueueNodeExceptionEvent(AppQueueEvent):
     """iteration id if node is in iteration"""
     in_loop_id: str | None = None
     """loop id if node is in loop"""
+    in_mention_parent_id: str | None = None
+    """parent node id if this is an extractor node event"""
     start_at: datetime
 
     inputs: Mapping[str, object] = Field(default_factory=dict)
@@ -407,6 +417,8 @@ class QueueNodeFailedEvent(AppQueueEvent):
     """iteration id if node is in iteration"""
     in_loop_id: str | None = None
     """loop id if node is in loop"""
+    in_mention_parent_id: str | None = None
+    """parent node id if this is an extractor node event"""
     start_at: datetime
 
     inputs: Mapping[str, object] = Field(default_factory=dict)

api/core/app/entities/task_entities.py

@@ -262,6 +262,7 @@ class NodeStartStreamResponse(StreamResponse):
         extras: dict[str, object] = Field(default_factory=dict)
         iteration_id: str | None = None
         loop_id: str | None = None
+        mention_parent_id: str | None = None
         agent_strategy: AgentNodeStrategyInit | None = None
 
     event: StreamEvent = StreamEvent.NODE_STARTED
@@ -285,6 +286,7 @@ class NodeStartStreamResponse(StreamResponse):
                 "extras": {},
                 "iteration_id": self.data.iteration_id,
                 "loop_id": self.data.loop_id,
+                "mention_parent_id": self.data.mention_parent_id,
             },
         }
 
@@ -320,6 +322,7 @@ class NodeFinishStreamResponse(StreamResponse):
         files: Sequence[Mapping[str, Any]] | None = []
         iteration_id: str | None = None
         loop_id: str | None = None
+        mention_parent_id: str | None = None
 
     event: StreamEvent = StreamEvent.NODE_FINISHED
     workflow_run_id: str
@@ -349,6 +352,7 @@ class NodeFinishStreamResponse(StreamResponse):
                 "files": [],
                 "iteration_id": self.data.iteration_id,
                 "loop_id": self.data.loop_id,
+                "mention_parent_id": self.data.mention_parent_id,
             },
         }
 
@@ -384,6 +388,7 @@ class NodeRetryStreamResponse(StreamResponse):
         files: Sequence[Mapping[str, Any]] | None = []
         iteration_id: str | None = None
         loop_id: str | None = None
+        mention_parent_id: str | None = None
         retry_index: int = 0
 
     event: StreamEvent = StreamEvent.NODE_RETRY
@@ -414,6 +419,7 @@ class NodeRetryStreamResponse(StreamResponse):
                 "files": [],
                 "iteration_id": self.data.iteration_id,
                 "loop_id": self.data.loop_id,
+                "mention_parent_id": self.data.mention_parent_id,
                 "retry_index": self.data.retry_index,
             },
         }

api/core/app/layers/conversation_variable_persist_layer.py

@@ -1,6 +1,6 @@
 import logging
 
-from core.variables import Variable
+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
@@ -44,7 +44,7 @@ class ConversationVariablePersistenceLayer(GraphEngineLayer):
             if selector[0] != CONVERSATION_VARIABLE_NODE_ID:
                 continue
             variable = self.graph_runtime_state.variable_pool.get(selector)
-            if not isinstance(variable, Variable):
+            if not isinstance(variable, VariableBase):
                 logger.warning(
                     "Conversation variable not found in variable pool. selector=%s",
                     selector,

api/core/helper/ssrf_proxy.py

@@ -33,6 +33,10 @@ class MaxRetriesExceededError(ValueError):
     pass
 
 
+request_error = httpx.RequestError
+max_retries_exceeded_error = MaxRetriesExceededError
+
+
 def _create_proxy_mounts() -> dict[str, httpx.HTTPTransport]:
     return {
         "http://": httpx.HTTPTransport(

api/core/llm_generator/llm_generator.py

@@ -1,8 +1,8 @@
 import json
 import logging
 import re
-from collections.abc import Sequence
-from typing import Protocol, cast
+from collections.abc import Mapping, Sequence
+from typing import Any, Protocol, cast
 
 import json_repair
 
@@ -398,6 +398,488 @@ class LLMGenerator:
             logger.exception("Failed to invoke LLM model, model: %s", model_config.get("name"))
             return {"output": "", "error": f"An unexpected error occurred: {str(e)}"}
 
+    @classmethod
+    def generate_with_context(
+        cls,
+        tenant_id: str,
+        workflow_id: str,
+        node_id: str,
+        parameter_name: str,
+        language: str,
+        prompt_messages: list[PromptMessage],
+        model_config: dict,
+    ) -> dict:
+        """
+        Generate extractor code node based on conversation context.
+
+        Args:
+            tenant_id: Tenant/workspace ID
+            workflow_id: Workflow ID
+            node_id: Current tool/llm node ID
+            parameter_name: Parameter name to generate code for
+            language: Code language (python3/javascript)
+            prompt_messages: Multi-turn conversation history (last message is instruction)
+            model_config: Model configuration (provider, name, completion_params)
+
+        Returns:
+            dict with CodeNodeData format:
+            - variables: Input variable selectors
+            - code_language: Code language
+            - code: Generated code
+            - outputs: Output definitions
+            - message: Explanation
+            - error: Error message if any
+        """
+        from sqlalchemy import select
+        from sqlalchemy.orm import Session
+
+        from services.workflow_service import WorkflowService
+
+        # Get workflow
+        with Session(db.engine) as session:
+            stmt = select(App).where(App.id == workflow_id)
+            app = session.scalar(stmt)
+            if not app:
+                return cls._error_response(f"App {workflow_id} not found")
+
+            workflow = WorkflowService().get_draft_workflow(app_model=app)
+            if not workflow:
+                return cls._error_response(f"Workflow for app {workflow_id} not found")
+
+        # Get upstream nodes via edge backtracking
+        upstream_nodes = cls._get_upstream_nodes(workflow.graph_dict, node_id)
+
+        # Get current node info
+        current_node = cls._get_node_by_id(workflow.graph_dict, node_id)
+        if not current_node:
+            return cls._error_response(f"Node {node_id} not found")
+
+        # Get parameter info
+        parameter_info = cls._get_parameter_info(
+            tenant_id=tenant_id,
+            node_data=current_node.get("data", {}),
+            parameter_name=parameter_name,
+        )
+
+        # Build system prompt
+        system_prompt = cls._build_extractor_system_prompt(
+            upstream_nodes=upstream_nodes,
+            current_node=current_node,
+            parameter_info=parameter_info,
+            language=language,
+        )
+
+        # Construct complete prompt_messages with system prompt
+        complete_messages: list[PromptMessage] = [
+            SystemPromptMessage(content=system_prompt),
+            *prompt_messages,
+        ]
+
+        from core.llm_generator.output_parser.structured_output import invoke_llm_with_structured_output
+
+        # Get model instance and schema
+        provider = model_config.get("provider", "")
+        model_name = model_config.get("name", "")
+        model_instance = ModelManager().get_model_instance(
+            tenant_id=tenant_id,
+            model_type=ModelType.LLM,
+            provider=provider,
+            model=model_name,
+        )
+
+        model_schema = model_instance.model_type_instance.get_model_schema(model_name, model_instance.credentials)
+        if not model_schema:
+            return cls._error_response(f"Model schema not found for {model_name}")
+
+        model_parameters = model_config.get("completion_params", {})
+        json_schema = cls._get_code_node_json_schema()
+
+        try:
+            response = invoke_llm_with_structured_output(
+                provider=provider,
+                model_schema=model_schema,
+                model_instance=model_instance,
+                prompt_messages=complete_messages,
+                json_schema=json_schema,
+                model_parameters=model_parameters,
+                stream=False,
+                tenant_id=tenant_id,
+            )
+
+            return cls._parse_code_node_output(
+                response.structured_output, language, parameter_info.get("type", "string")
+            )
+
+        except InvokeError as e:
+            return cls._error_response(str(e))
+        except Exception as e:
+            logger.exception("Failed to generate with context, model: %s", model_config.get("name"))
+            return cls._error_response(f"An unexpected error occurred: {str(e)}")
+
+    @classmethod
+    def _error_response(cls, error: str) -> dict:
+        """Return error response in CodeNodeData format."""
+        return {
+            "variables": [],
+            "code_language": "python3",
+            "code": "",
+            "outputs": {},
+            "message": "",
+            "error": error,
+        }
+
+    @classmethod
+    def generate_suggested_questions(
+        cls,
+        tenant_id: str,
+        workflow_id: str,
+        node_id: str,
+        parameter_name: str,
+        language: str,
+        model_config: dict | None = None,
+    ) -> dict:
+        """
+        Generate suggested questions for context generation.
+
+        Returns dict with questions array and error field.
+        """
+        from sqlalchemy import select
+        from sqlalchemy.orm import Session
+
+        from core.llm_generator.output_parser.structured_output import invoke_llm_with_structured_output
+        from services.workflow_service import WorkflowService
+
+        # Get workflow context (reuse existing logic)
+        with Session(db.engine) as session:
+            stmt = select(App).where(App.id == workflow_id)
+            app = session.scalar(stmt)
+            if not app:
+                return {"questions": [], "error": f"App {workflow_id} not found"}
+
+            workflow = WorkflowService().get_draft_workflow(app_model=app)
+            if not workflow:
+                return {"questions": [], "error": f"Workflow for app {workflow_id} not found"}
+
+        upstream_nodes = cls._get_upstream_nodes(workflow.graph_dict, node_id)
+        current_node = cls._get_node_by_id(workflow.graph_dict, node_id)
+        if not current_node:
+            return {"questions": [], "error": f"Node {node_id} not found"}
+
+        parameter_info = cls._get_parameter_info(
+            tenant_id=tenant_id,
+            node_data=current_node.get("data", {}),
+            parameter_name=parameter_name,
+        )
+
+        # Build prompt
+        system_prompt = cls._build_suggested_questions_prompt(
+            upstream_nodes=upstream_nodes,
+            current_node=current_node,
+            parameter_info=parameter_info,
+            language=language,
+        )
+
+        prompt_messages: list[PromptMessage] = [
+            SystemPromptMessage(content=system_prompt),
+        ]
+
+        # Get model instance - use default if model_config not provided
+        model_manager = ModelManager()
+        if model_config:
+            provider = model_config.get("provider", "")
+            model_name = model_config.get("name", "")
+            model_instance = model_manager.get_model_instance(
+                tenant_id=tenant_id,
+                model_type=ModelType.LLM,
+                provider=provider,
+                model=model_name,
+            )
+        else:
+            model_instance = model_manager.get_default_model_instance(
+                tenant_id=tenant_id,
+                model_type=ModelType.LLM,
+            )
+            model_name = model_instance.model
+
+        model_schema = model_instance.model_type_instance.get_model_schema(model_name, model_instance.credentials)
+        if not model_schema:
+            return {"questions": [], "error": f"Model schema not found for {model_name}"}
+
+        completion_params = model_config.get("completion_params", {}) if model_config else {}
+        model_parameters = {**completion_params, "max_tokens": 256}
+        json_schema = cls._get_suggested_questions_json_schema()
+
+        try:
+            response = invoke_llm_with_structured_output(
+                provider=model_instance.provider,
+                model_schema=model_schema,
+                model_instance=model_instance,
+                prompt_messages=prompt_messages,
+                json_schema=json_schema,
+                model_parameters=model_parameters,
+                stream=False,
+                tenant_id=tenant_id,
+            )
+
+            questions = response.structured_output.get("questions", []) if response.structured_output else []
+            return {"questions": questions, "error": ""}
+
+        except InvokeError as e:
+            return {"questions": [], "error": str(e)}
+        except Exception as e:
+            logger.exception("Failed to generate suggested questions, model: %s", model_name)
+            return {"questions": [], "error": f"An unexpected error occurred: {str(e)}"}
+
+    @classmethod
+    def _build_suggested_questions_prompt(
+        cls,
+        upstream_nodes: list[dict],
+        current_node: dict,
+        parameter_info: dict,
+        language: str = "English",
+    ) -> str:
+        """Build minimal prompt for suggested questions generation."""
+        # Simplify upstream nodes to reduce tokens
+        sources = [f"{n['title']}({','.join(n.get('outputs', {}).keys())})" for n in upstream_nodes[:5]]
+        param_type = parameter_info.get("type", "string")
+        param_desc = parameter_info.get("description", "")[:100]
+
+        return f"""Suggest 3 code generation questions for extracting data.
+Sources: {", ".join(sources)}
+Target: {parameter_info.get("name")}({param_type}) - {param_desc}
+Output 3 short, practical questions in {language}."""
+
+    @classmethod
+    def _get_suggested_questions_json_schema(cls) -> dict:
+        """Return JSON Schema for suggested questions."""
+        return {
+            "type": "object",
+            "properties": {
+                "questions": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "minItems": 3,
+                    "maxItems": 3,
+                    "description": "3 suggested questions",
+                },
+            },
+            "required": ["questions"],
+        }
+
+    @classmethod
+    def _get_code_node_json_schema(cls) -> dict:
+        """Return JSON Schema for structured output."""
+        return {
+            "type": "object",
+            "properties": {
+                "variables": {
+                    "type": "array",
+                    "items": {
+                        "type": "object",
+                        "properties": {
+                            "variable": {"type": "string", "description": "Variable name in code"},
+                            "value_selector": {
+                                "type": "array",
+                                "items": {"type": "string"},
+                                "description": "Path like [node_id, output_name]",
+                            },
+                        },
+                        "required": ["variable", "value_selector"],
+                    },
+                },
+                "code": {"type": "string", "description": "Generated code with main function"},
+                "outputs": {
+                    "type": "object",
+                    "additionalProperties": {
+                        "type": "object",
+                        "properties": {"type": {"type": "string"}},
+                    },
+                    "description": "Output definitions, key is output name",
+                },
+                "explanation": {"type": "string", "description": "Brief explanation of the code"},
+            },
+            "required": ["variables", "code", "outputs", "explanation"],
+        }
+
+    @classmethod
+    def _get_upstream_nodes(cls, graph_dict: Mapping[str, Any], node_id: str) -> list[dict]:
+        """
+        Get all upstream nodes via edge backtracking.
+
+        Traverses the graph backwards from node_id to collect all reachable nodes.
+        """
+        from collections import defaultdict
+
+        nodes = {n["id"]: n for n in graph_dict.get("nodes", [])}
+        edges = graph_dict.get("edges", [])
+
+        # Build reverse adjacency list
+        reverse_adj: dict[str, list[str]] = defaultdict(list)
+        for edge in edges:
+            reverse_adj[edge["target"]].append(edge["source"])
+
+        # BFS to find all upstream nodes
+        visited: set[str] = set()
+        queue = [node_id]
+        upstream: list[dict] = []
+
+        while queue:
+            current = queue.pop(0)
+            for source in reverse_adj.get(current, []):
+                if source not in visited:
+                    visited.add(source)
+                    queue.append(source)
+                    if source in nodes:
+                        upstream.append(cls._extract_node_info(nodes[source]))
+
+        return upstream
+
+    @classmethod
+    def _get_node_by_id(cls, graph_dict: Mapping[str, Any], node_id: str) -> dict | None:
+        """Get node by ID from graph."""
+        for node in graph_dict.get("nodes", []):
+            if node["id"] == node_id:
+                return node
+        return None
+
+    @classmethod
+    def _extract_node_info(cls, node: dict) -> dict:
+        """Extract minimal node info with outputs based on node type."""
+        node_type = node["data"]["type"]
+        node_data = node.get("data", {})
+
+        # Build outputs based on node type (only type, no description to reduce tokens)
+        outputs: dict[str, str] = {}
+        match node_type:
+            case "start":
+                for var in node_data.get("variables", []):
+                    name = var.get("variable", var.get("name", ""))
+                    outputs[name] = var.get("type", "string")
+            case "llm":
+                outputs["text"] = "string"
+            case "code":
+                for name, output in node_data.get("outputs", {}).items():
+                    outputs[name] = output.get("type", "string")
+            case "http-request":
+                outputs = {"body": "string", "status_code": "number", "headers": "object"}
+            case "knowledge-retrieval":
+                outputs["result"] = "array[object]"
+            case "tool":
+                outputs = {"text": "string", "json": "object"}
+            case _:
+                outputs["output"] = "string"
+
+        info: dict = {
+            "id": node["id"],
+            "title": node_data.get("title", node["id"]),
+            "outputs": outputs,
+        }
+        # Only include description if not empty
+        desc = node_data.get("desc", "")
+        if desc:
+            info["desc"] = desc
+
+        return info
+
+    @classmethod
+    def _get_parameter_info(cls, tenant_id: str, node_data: dict, parameter_name: str) -> dict:
+        """Get parameter info from tool schema using ToolManager."""
+        default_info = {"name": parameter_name, "type": "string", "description": ""}
+
+        if node_data.get("type") != "tool":
+            return default_info
+
+        try:
+            from core.app.entities.app_invoke_entities import InvokeFrom
+            from core.tools.entities.tool_entities import ToolProviderType
+            from core.tools.tool_manager import ToolManager
+
+            provider_type_str = node_data.get("provider_type", "")
+            provider_type = ToolProviderType(provider_type_str) if provider_type_str else ToolProviderType.BUILT_IN
+
+            tool_runtime = ToolManager.get_tool_runtime(
+                provider_type=provider_type,
+                provider_id=node_data.get("provider_id", ""),
+                tool_name=node_data.get("tool_name", ""),
+                tenant_id=tenant_id,
+                invoke_from=InvokeFrom.DEBUGGER,
+            )
+
+            parameters = tool_runtime.get_merged_runtime_parameters()
+            for param in parameters:
+                if param.name == parameter_name:
+                    return {
+                        "name": param.name,
+                        "type": param.type.value if hasattr(param.type, "value") else str(param.type),
+                        "description": param.llm_description
+                        or (param.human_description.en_US if param.human_description else ""),
+                        "required": param.required,
+                    }
+        except Exception as e:
+            logger.debug("Failed to get parameter info from ToolManager: %s", e)
+
+        return default_info
+
+    @classmethod
+    def _build_extractor_system_prompt(
+        cls,
+        upstream_nodes: list[dict],
+        current_node: dict,
+        parameter_info: dict,
+        language: str,
+    ) -> str:
+        """Build system prompt for extractor code generation."""
+        upstream_json = json.dumps(upstream_nodes, indent=2, ensure_ascii=False)
+        param_type = parameter_info.get("type", "string")
+        return f"""You are a code generator for workflow automation.
+
+Generate {language} code to extract/transform upstream node outputs for the target parameter.
+
+## Upstream Nodes
+{upstream_json}
+
+## Target
+Node: {current_node["data"].get("title", current_node["id"])}
+Parameter: {parameter_info.get("name")} ({param_type}) - {parameter_info.get("description", "")}
+
+## Requirements
+- Write a main function that returns type: {param_type}
+- Use value_selector format: ["node_id", "output_name"]
+"""
+
+    @classmethod
+    def _parse_code_node_output(cls, content: Mapping[str, Any] | None, language: str, parameter_type: str) -> dict:
+        """
+        Parse structured output to CodeNodeData format.
+
+        Args:
+            content: Structured output dict from invoke_llm_with_structured_output
+            language: Code language
+            parameter_type: Expected parameter type
+
+        Returns dict with variables, code_language, code, outputs, message, error.
+        """
+        if content is None:
+            return cls._error_response("Empty or invalid response from LLM")
+
+        # Validate and normalize variables
+        variables = [
+            {"variable": v.get("variable", ""), "value_selector": v.get("value_selector", [])}
+            for v in content.get("variables", [])
+            if isinstance(v, dict)
+        ]
+
+        outputs = content.get("outputs", {"result": {"type": parameter_type}})
+
+        return {
+            "variables": variables,
+            "code_language": language,
+            "code": content.get("code", ""),
+            "outputs": outputs,
+            "message": content.get("explanation", ""),
+            "error": "",
+        }
+
     @staticmethod
     def instruction_modify_legacy(
         tenant_id: str, flow_id: str, current: str, instruction: str, model_config: dict, ideal_output: str | None

api/core/llm_generator/output_parser/file_ref.py

@@ -0,0 +1,188 @@
+"""
+File reference detection and conversion for structured output.
+
+This module provides utilities to:
+1. Detect file reference fields in JSON Schema (format: "dify-file-ref")
+2. Convert file ID strings to File objects after LLM returns
+"""
+
+import uuid
+from collections.abc import Mapping
+from typing import Any
+
+from core.file import File
+from core.variables.segments import ArrayFileSegment, FileSegment
+from factories.file_factory import build_from_mapping
+
+FILE_REF_FORMAT = "dify-file-ref"
+
+
+def is_file_ref_property(schema: dict) -> bool:
+    """Check if a schema property is a file reference."""
+    return schema.get("type") == "string" and schema.get("format") == FILE_REF_FORMAT
+
+
+def detect_file_ref_fields(schema: Mapping[str, Any], path: str = "") -> list[str]:
+    """
+    Recursively detect file reference fields in schema.
+
+    Args:
+        schema: JSON Schema to analyze
+        path: Current path in the schema (used for recursion)
+
+    Returns:
+        List of JSON paths containing file refs, e.g., ["image_id", "files[*]"]
+    """
+    file_ref_paths: list[str] = []
+    schema_type = schema.get("type")
+
+    if schema_type == "object":
+        for prop_name, prop_schema in schema.get("properties", {}).items():
+            current_path = f"{path}.{prop_name}" if path else prop_name
+
+            if is_file_ref_property(prop_schema):
+                file_ref_paths.append(current_path)
+            elif isinstance(prop_schema, dict):
+                file_ref_paths.extend(detect_file_ref_fields(prop_schema, current_path))
+
+    elif schema_type == "array":
+        items_schema = schema.get("items", {})
+        array_path = f"{path}[*]" if path else "[*]"
+
+        if is_file_ref_property(items_schema):
+            file_ref_paths.append(array_path)
+        elif isinstance(items_schema, dict):
+            file_ref_paths.extend(detect_file_ref_fields(items_schema, array_path))
+
+    return file_ref_paths
+
+
+def convert_file_refs_in_output(
+    output: Mapping[str, Any],
+    json_schema: Mapping[str, Any],
+    tenant_id: str,
+) -> dict[str, Any]:
+    """
+    Convert file ID strings to File objects based on schema.
+
+    Args:
+        output: The structured_output from LLM result
+        json_schema: The original JSON schema (to detect file ref fields)
+        tenant_id: Tenant ID for file lookup
+
+    Returns:
+        Output with file references converted to File objects
+    """
+    file_ref_paths = detect_file_ref_fields(json_schema)
+    if not file_ref_paths:
+        return dict(output)
+
+    result = _deep_copy_dict(output)
+
+    for path in file_ref_paths:
+        _convert_path_in_place(result, path.split("."), tenant_id)
+
+    return result
+
+
+def _deep_copy_dict(obj: Mapping[str, Any]) -> dict[str, Any]:
+    """Deep copy a mapping to a mutable dict."""
+    result: dict[str, Any] = {}
+    for key, value in obj.items():
+        if isinstance(value, Mapping):
+            result[key] = _deep_copy_dict(value)
+        elif isinstance(value, list):
+            result[key] = [_deep_copy_dict(item) if isinstance(item, Mapping) else item for item in value]
+        else:
+            result[key] = value
+    return result
+
+
+def _convert_path_in_place(obj: dict, path_parts: list[str], tenant_id: str) -> None:
+    """Convert file refs at the given path in place, wrapping in Segment types."""
+    if not path_parts:
+        return
+
+    current = path_parts[0]
+    remaining = path_parts[1:]
+
+    # Handle array notation like "files[*]"
+    if current.endswith("[*]"):
+        key = current[:-3] if current != "[*]" else None
+        target = obj.get(key) if key else obj
+
+        if isinstance(target, list):
+            if remaining:
+                # Nested array with remaining path - recurse into each item
+                for item in target:
+                    if isinstance(item, dict):
+                        _convert_path_in_place(item, remaining, tenant_id)
+            else:
+                # Array of file IDs - convert all and wrap in ArrayFileSegment
+                files: list[File] = []
+                for item in target:
+                    file = _convert_file_id(item, tenant_id)
+                    if file is not None:
+                        files.append(file)
+                # Replace the array with ArrayFileSegment
+                if key:
+                    obj[key] = ArrayFileSegment(value=files)
+        return
+
+    if not remaining:
+        # Leaf node - convert the value and wrap in FileSegment
+        if current in obj:
+            file = _convert_file_id(obj[current], tenant_id)
+            if file is not None:
+                obj[current] = FileSegment(value=file)
+            else:
+                obj[current] = None
+    else:
+        # Recurse into nested object
+        if current in obj and isinstance(obj[current], dict):
+            _convert_path_in_place(obj[current], remaining, tenant_id)
+
+
+def _convert_file_id(file_id: Any, tenant_id: str) -> File | None:
+    """
+    Convert a file ID string to a File object.
+
+    Tries multiple file sources in order:
+    1. ToolFile (files generated by tools/workflows)
+    2. UploadFile (files uploaded by users)
+    """
+    if not isinstance(file_id, str):
+        return None
+
+    # Validate UUID format
+    try:
+        uuid.UUID(file_id)
+    except ValueError:
+        return None
+
+    # Try ToolFile first (files generated by tools/workflows)
+    try:
+        return build_from_mapping(
+            mapping={
+                "transfer_method": "tool_file",
+                "tool_file_id": file_id,
+            },
+            tenant_id=tenant_id,
+        )
+    except ValueError:
+        pass
+
+    # Try UploadFile (files uploaded by users)
+    try:
+        return build_from_mapping(
+            mapping={
+                "transfer_method": "local_file",
+                "upload_file_id": file_id,
+            },
+            tenant_id=tenant_id,
+        )
+    except ValueError:
+        pass
+
+    # File not found in any source
+    return None

api/core/llm_generator/output_parser/structured_output.py

@@ -8,6 +8,7 @@ import json_repair
 from pydantic import TypeAdapter, ValidationError
 
 from core.llm_generator.output_parser.errors import OutputParserError
+from core.llm_generator.output_parser.file_ref import convert_file_refs_in_output
 from core.llm_generator.prompts import STRUCTURED_OUTPUT_PROMPT
 from core.model_manager import ModelInstance
 from core.model_runtime.callbacks.base_callback import Callback
@@ -57,6 +58,7 @@ def invoke_llm_with_structured_output(
     stream: Literal[True],
     user: str | None = None,
     callbacks: list[Callback] | None = None,
+    tenant_id: str | None = None,
 ) -> Generator[LLMResultChunkWithStructuredOutput, None, None]: ...
 @overload
 def invoke_llm_with_structured_output(
@@ -72,6 +74,7 @@ def invoke_llm_with_structured_output(
     stream: Literal[False],
     user: str | None = None,
     callbacks: list[Callback] | None = None,
+    tenant_id: str | None = None,
 ) -> LLMResultWithStructuredOutput: ...
 @overload
 def invoke_llm_with_structured_output(
@@ -87,6 +90,7 @@ def invoke_llm_with_structured_output(
     stream: bool = True,
     user: str | None = None,
     callbacks: list[Callback] | None = None,
+    tenant_id: str | None = None,
 ) -> LLMResultWithStructuredOutput | Generator[LLMResultChunkWithStructuredOutput, None, None]: ...
 def invoke_llm_with_structured_output(
     *,
@@ -101,20 +105,28 @@ def invoke_llm_with_structured_output(
     stream: bool = True,
     user: str | None = None,
     callbacks: list[Callback] | None = None,
+    tenant_id: str | None = None,
 ) -> LLMResultWithStructuredOutput | Generator[LLMResultChunkWithStructuredOutput, None, None]:
     """
-    Invoke large language model with structured output
-    1. This method invokes model_instance.invoke_llm with json_schema
-    2. Try to parse the result as structured output
+    Invoke large language model with structured output.
 
+    This method invokes model_instance.invoke_llm with json_schema and parses
+    the result as structured output.
+
+    :param provider: model provider name
+    :param model_schema: model schema entity
+    :param model_instance: model instance to invoke
     :param prompt_messages: prompt messages
-    :param json_schema: json schema
+    :param json_schema: json schema for structured output
     :param model_parameters: model parameters
     :param tools: tools for tool calling
     :param stop: stop words
     :param stream: is stream response
     :param user: unique user id
     :param callbacks: callbacks
+    :param tenant_id: tenant ID for file reference conversion. When provided and
+                      json_schema contains file reference fields (format: "dify-file-ref"),
+                      file IDs in the output will be automatically converted to File objects.
     :return: full response or stream response chunk generator result
     """
 
@@ -153,8 +165,18 @@ def invoke_llm_with_structured_output(
                 f"Failed to parse structured output, LLM result is not a string: {llm_result.message.content}"
             )
 
+        structured_output = _parse_structured_output(llm_result.message.content)
+
+        # Convert file references if tenant_id is provided
+        if tenant_id is not None:
+            structured_output = convert_file_refs_in_output(
+                output=structured_output,
+                json_schema=json_schema,
+                tenant_id=tenant_id,
+            )
+
         return LLMResultWithStructuredOutput(
-            structured_output=_parse_structured_output(llm_result.message.content),
+            structured_output=structured_output,
             model=llm_result.model,
             message=llm_result.message,
             usage=llm_result.usage,
@@ -186,8 +208,18 @@ def invoke_llm_with_structured_output(
                     delta=event.delta,
                 )
 
+            structured_output = _parse_structured_output(result_text)
+
+            # Convert file references if tenant_id is provided
+            if tenant_id is not None:
+                structured_output = convert_file_refs_in_output(
+                    output=structured_output,
+                    json_schema=json_schema,
+                    tenant_id=tenant_id,
+                )
+
             yield LLMResultChunkWithStructuredOutput(
-                structured_output=_parse_structured_output(result_text),
+                structured_output=structured_output,
                 model=model_schema.model,
                 prompt_messages=prompt_messages,
                 system_fingerprint=system_fingerprint,

api/core/llm_generator/utils.py

@@ -0,0 +1,45 @@
+"""Utility functions for LLM generator."""
+
+from core.model_runtime.entities.message_entities import (
+    AssistantPromptMessage,
+    PromptMessage,
+    PromptMessageRole,
+    SystemPromptMessage,
+    ToolPromptMessage,
+    UserPromptMessage,
+)
+
+
+def deserialize_prompt_messages(messages: list[dict]) -> list[PromptMessage]:
+    """
+    Deserialize list of dicts to list[PromptMessage].
+
+    Expected format:
+        [
+            {"role": "user", "content": "..."},
+            {"role": "assistant", "content": "..."},
+        ]
+    """
+    result: list[PromptMessage] = []
+    for msg in messages:
+        role = PromptMessageRole.value_of(msg["role"])
+        content = msg.get("content", "")
+
+        match role:
+            case PromptMessageRole.USER:
+                result.append(UserPromptMessage(content=content))
+            case PromptMessageRole.ASSISTANT:
+                result.append(AssistantPromptMessage(content=content))
+            case PromptMessageRole.SYSTEM:
+                result.append(SystemPromptMessage(content=content))
+            case PromptMessageRole.TOOL:
+                result.append(ToolPromptMessage(content=content, tool_call_id=msg.get("tool_call_id", "")))
+
+    return result
+
+
+def serialize_prompt_messages(messages: list[PromptMessage]) -> list[dict]:
+    """
+    Serialize list[PromptMessage] to list of dicts.
+    """
+    return [{"role": msg.role.value, "content": msg.content} for msg in messages]

api/core/memory/README.md

@@ -0,0 +1,434 @@
+# Memory Module
+
+This module provides memory management for LLM conversations, enabling context retention across dialogue turns.
+
+## Overview
+
+The memory module contains two types of memory implementations:
+
+1. **TokenBufferMemory** - Conversation-level memory (existing)
+2. **NodeTokenBufferMemory** - Node-level memory (to be implemented, **Chatflow only**)
+
+> **Note**: `NodeTokenBufferMemory` is only available in **Chatflow** (advanced-chat mode).
+> This is because it requires both `conversation_id` and `node_id`, which are only present in Chatflow.
+> Standard Workflow mode does not have `conversation_id` and therefore cannot use node-level memory.
+
+```
+┌─────────────────────────────────────────────────────────────────────────────┐
+│                           Memory Architecture                               │
+├─────────────────────────────────────────────────────────────────────────────┤
+│                                                                             │
+│  ┌─────────────────────────────────────────────────────────────────────-┐   │
+│  │                      TokenBufferMemory                               │   │
+│  │  Scope: Conversation                                                 │   │
+│  │  Storage: Database (Message table)                                   │   │
+│  │  Key: conversation_id                                                │   │
+│  └─────────────────────────────────────────────────────────────────────-┘   │
+│                                                                             │
+│  ┌─────────────────────────────────────────────────────────────────────-┐   │
+│  │                    NodeTokenBufferMemory                             │   │
+│  │  Scope: Node within Conversation                                     │   │
+│  │  Storage: Object Storage (JSON file)                                 │   │
+│  │  Key: (app_id, conversation_id, node_id)                             │   │
+│  └─────────────────────────────────────────────────────────────────────-┘   │
+│                                                                             │
+└─────────────────────────────────────────────────────────────────────────────┘
+```
+
+---
+
+## TokenBufferMemory (Existing)
+
+### Purpose
+
+`TokenBufferMemory` retrieves conversation history from the `Message` table and converts it to `PromptMessage` objects for LLM context.
+
+### Key Features
+
+- **Conversation-scoped**: All messages within a conversation are candidates
+- **Thread-aware**: Uses `parent_message_id` to extract only the current thread (supports regeneration scenarios)
+- **Token-limited**: Truncates history to fit within `max_token_limit`
+- **File support**: Handles `MessageFile` attachments (images, documents, etc.)
+
+### Data Flow
+
+```
+Message Table                    TokenBufferMemory              LLM
+     │                                  │                        │
+     │  SELECT * FROM messages          │                        │
+     │  WHERE conversation_id = ?       │                        │
+     │  ORDER BY created_at DESC        │                        │
+     ├─────────────────────────────────▶│                        │
+     │                                  │                        │
+     │                    extract_thread_messages()              │
+     │                                  │                        │
+     │                    build_prompt_message_with_files()      │
+     │                                  │                        │
+     │                    truncate by max_token_limit            │
+     │                                  │                        │
+     │                                  │  Sequence[PromptMessage]
+     │                                  ├───────────────────────▶│
+     │                                  │                        │
+```
+
+### Thread Extraction
+
+When a user regenerates a response, a new thread is created:
+
+```
+Message A (user)
+    └── Message A' (assistant)
+            └── Message B (user)
+                    └── Message B' (assistant)
+            └── Message A'' (assistant, regenerated)  ← New thread
+                    └── Message C (user)
+                            └── Message C' (assistant)
+```
+
+`extract_thread_messages()` traces back from the latest message using `parent_message_id` to get only the current thread: `[A, A'', C, C']`
+
+### Usage
+
+```python
+from core.memory.token_buffer_memory import TokenBufferMemory
+
+memory = TokenBufferMemory(conversation=conversation, model_instance=model_instance)
+history = memory.get_history_prompt_messages(max_token_limit=2000, message_limit=100)
+```
+
+---
+
+## NodeTokenBufferMemory (To Be Implemented)
+
+### Purpose
+
+`NodeTokenBufferMemory` provides **node-scoped memory** within a conversation. Each LLM node in a workflow can maintain its own independent conversation history.
+
+### Use Cases
+
+1. **Multi-LLM Workflows**: Different LLM nodes need separate context
+2. **Iterative Processing**: An LLM node in a loop needs to accumulate context across iterations
+3. **Specialized Agents**: Each agent node maintains its own dialogue history
+
+### Design Decisions
+
+#### Storage: Object Storage for Messages (No New Database Table)
+
+| Aspect                    | Database             | Object Storage     |
+| ------------------------- | -------------------- | ------------------ |
+| Cost                      | High                 | Low                |
+| Query Flexibility         | High                 | Low                |
+| Schema Changes            | Migration required   | None               |
+| Consistency with existing | ConversationVariable | File uploads, logs |
+
+**Decision**: Store message data in object storage, but still use existing database tables for file metadata.
+
+**What is stored in Object Storage:**
+
+- Message content (text)
+- Message metadata (role, token_count, created_at)
+- File references (upload_file_id, tool_file_id, etc.)
+- Thread relationships (message_id, parent_message_id)
+
+**What still requires Database queries:**
+
+- File reconstruction: When reading node memory, file references are used to query
+  `UploadFile` / `ToolFile` tables via `file_factory.build_from_mapping()` to rebuild
+  complete `File` objects with storage_key, mime_type, etc.
+
+**Why this hybrid approach:**
+
+- No database migration required (no new tables)
+- Message data may be large, object storage is cost-effective
+- File metadata is already in database, no need to duplicate
+- Aligns with existing storage patterns (file uploads, logs)
+
+#### Storage Key Format
+
+```
+node_memory/{app_id}/{conversation_id}/{node_id}.json
+```
+
+#### Data Structure
+
+```json
+{
+  "version": 1,
+  "messages": [
+    {
+      "message_id": "msg-001",
+      "parent_message_id": null,
+      "role": "user",
+      "content": "Analyze this image",
+      "files": [
+        {
+          "type": "image",
+          "transfer_method": "local_file",
+          "upload_file_id": "file-uuid-123",
+          "belongs_to": "user"
+        }
+      ],
+      "token_count": 15,
+      "created_at": "2026-01-07T10:00:00Z"
+    },
+    {
+      "message_id": "msg-002",
+      "parent_message_id": "msg-001",
+      "role": "assistant",
+      "content": "This is a landscape image...",
+      "files": [],
+      "token_count": 50,
+      "created_at": "2026-01-07T10:00:01Z"
+    }
+  ]
+}
+```
+
+### Thread Support
+
+Node memory also supports thread extraction (for regeneration scenarios):
+
+```python
+def _extract_thread(
+    self,
+    messages: list[NodeMemoryMessage],
+    current_message_id: str
+) -> list[NodeMemoryMessage]:
+    """
+    Extract messages belonging to the thread of current_message_id.
+    Similar to extract_thread_messages() in TokenBufferMemory.
+    """
+    ...
+```
+
+### File Handling
+
+Files are stored as references (not full metadata):
+
+```python
+class NodeMemoryFile(BaseModel):
+    type: str                        # image, audio, video, document, custom
+    transfer_method: str             # local_file, remote_url, tool_file
+    upload_file_id: str | None       # for local_file
+    tool_file_id: str | None         # for tool_file
+    url: str | None                  # for remote_url
+    belongs_to: str                  # user / assistant
+```
+
+When reading, files are rebuilt using `file_factory.build_from_mapping()`.
+
+### API Design
+
+```python
+class NodeTokenBufferMemory:
+    def __init__(
+        self,
+        app_id: str,
+        conversation_id: str,
+        node_id: str,
+        model_instance: ModelInstance,
+    ):
+        """
+        Initialize node-level memory.
+
+        :param app_id: Application ID
+        :param conversation_id: Conversation ID
+        :param node_id: Node ID in the workflow
+        :param model_instance: Model instance for token counting
+        """
+        ...
+
+    def add_messages(
+        self,
+        message_id: str,
+        parent_message_id: str | None,
+        user_content: str,
+        user_files: Sequence[File],
+        assistant_content: str,
+        assistant_files: Sequence[File],
+    ) -> None:
+        """
+        Append a dialogue turn (user + assistant) to node memory.
+        Call this after LLM node execution completes.
+
+        :param message_id: Current message ID (from Message table)
+        :param parent_message_id: Parent message ID (for thread tracking)
+        :param user_content: User's text input
+        :param user_files: Files attached by user
+        :param assistant_content: Assistant's text response
+        :param assistant_files: Files generated by assistant
+        """
+        ...
+
+    def get_history_prompt_messages(
+        self,
+        current_message_id: str,
+        tenant_id: str,
+        max_token_limit: int = 2000,
+        file_upload_config: FileUploadConfig | None = None,
+    ) -> Sequence[PromptMessage]:
+        """
+        Retrieve history as PromptMessage sequence.
+
+        :param current_message_id: Current message ID (for thread extraction)
+        :param tenant_id: Tenant ID (for file reconstruction)
+        :param max_token_limit: Maximum tokens for history
+        :param file_upload_config: File upload configuration
+        :return: Sequence of PromptMessage for LLM context
+        """
+        ...
+
+    def flush(self) -> None:
+        """
+        Persist buffered changes to object storage.
+        Call this at the end of node execution.
+        """
+        ...
+
+    def clear(self) -> None:
+        """
+        Clear all messages in this node's memory.
+        """
+        ...
+```
+
+### Data Flow
+
+```
+Object Storage                  NodeTokenBufferMemory           LLM Node
+     │                                  │                           │
+     │                                  │◀── get_history_prompt_messages()
+     │  storage.load(key)               │                           │
+     │◀─────────────────────────────────┤                           │
+     │                                  │                           │
+     │  JSON data                       │                           │
+     ├─────────────────────────────────▶│                           │
+     │                                  │                           │
+     │                    _extract_thread()                         │
+     │                                  │                           │
+     │                    _rebuild_files() via file_factory         │
+     │                                  │                           │
+     │                    _build_prompt_messages()                  │
+     │                                  │                           │
+     │                    _truncate_by_tokens()                     │
+     │                                  │                           │
+     │                                  │  Sequence[PromptMessage]  │
+     │                                  ├──────────────────────────▶│
+     │                                  │                           │
+     │                                  │◀── LLM execution complete │
+     │                                  │                           │
+     │                                  │◀── add_messages()         │
+     │                                  │                           │
+     │  storage.save(key, data)         │                           │
+     │◀─────────────────────────────────┤                           │
+     │                                  │                           │
+```
+
+### Integration with LLM Node
+
+```python
+# In LLM Node execution
+
+# 1. Fetch memory based on mode
+if node_data.memory and node_data.memory.mode == MemoryMode.NODE:
+    # Node-level memory (Chatflow only)
+    memory = fetch_node_memory(
+        variable_pool=variable_pool,
+        app_id=app_id,
+        node_id=self.node_id,
+        node_data_memory=node_data.memory,
+        model_instance=model_instance,
+    )
+elif node_data.memory and node_data.memory.mode == MemoryMode.CONVERSATION:
+    # Conversation-level memory (existing behavior)
+    memory = fetch_memory(
+        variable_pool=variable_pool,
+        app_id=app_id,
+        node_data_memory=node_data.memory,
+        model_instance=model_instance,
+    )
+else:
+    memory = None
+
+# 2. Get history for context
+if memory:
+    if isinstance(memory, NodeTokenBufferMemory):
+        history = memory.get_history_prompt_messages(
+            current_message_id=current_message_id,
+            tenant_id=tenant_id,
+            max_token_limit=max_token_limit,
+        )
+    else:  # TokenBufferMemory
+        history = memory.get_history_prompt_messages(
+            max_token_limit=max_token_limit,
+        )
+    prompt_messages = [*history, *current_messages]
+else:
+    prompt_messages = current_messages
+
+# 3. Call LLM
+response = model_instance.invoke(prompt_messages)
+
+# 4. Append to node memory (only for NodeTokenBufferMemory)
+if isinstance(memory, NodeTokenBufferMemory):
+    memory.add_messages(
+        message_id=message_id,
+        parent_message_id=parent_message_id,
+        user_content=user_input,
+        user_files=user_files,
+        assistant_content=response.content,
+        assistant_files=response_files,
+    )
+    memory.flush()
+```
+
+### Configuration
+
+Add to `MemoryConfig` in `core/workflow/nodes/llm/entities.py`:
+
+```python
+class MemoryMode(StrEnum):
+    CONVERSATION = "conversation"  # Use TokenBufferMemory (default, existing behavior)
+    NODE = "node"                  # Use NodeTokenBufferMemory (new, Chatflow only)
+
+class MemoryConfig(BaseModel):
+    # Existing fields
+    role_prefix: RolePrefix | None = None
+    window: MemoryWindowConfig | None = None
+    query_prompt_template: str | None = None
+
+    # Memory mode (new)
+    mode: MemoryMode = MemoryMode.CONVERSATION
+```
+
+**Mode Behavior:**
+
+| Mode           | Memory Class          | Scope                    | Availability  |
+| -------------- | --------------------- | ------------------------ | ------------- |
+| `conversation` | TokenBufferMemory     | Entire conversation      | All app modes |
+| `node`         | NodeTokenBufferMemory | Per-node in conversation | Chatflow only |
+
+> When `mode=node` is used in a non-Chatflow context (no conversation_id), it should
+> fall back to no memory or raise a configuration error.
+
+---
+
+## Comparison
+
+| Feature        | TokenBufferMemory        | NodeTokenBufferMemory     |
+| -------------- | ------------------------ | ------------------------- |
+| Scope          | Conversation             | Node within Conversation  |
+| Storage        | Database (Message table) | Object Storage (JSON)     |
+| Thread Support | Yes                      | Yes                       |
+| File Support   | Yes (via MessageFile)    | Yes (via file references) |
+| Token Limit    | Yes                      | Yes                       |
+| Use Case       | Standard chat apps       | Complex workflows         |
+
+---
+
+## Future Considerations
+
+1. **Cleanup Task**: Add a Celery task to clean up old node memory files
+2. **Concurrency**: Consider Redis lock for concurrent node executions
+3. **Compression**: Compress large memory files to reduce storage costs
+4. **Extension**: Other nodes (Agent, Tool) may also benefit from node-level memory

api/core/memory/__init__.py

@@ -0,0 +1,15 @@
+from core.memory.base import BaseMemory
+from core.memory.node_token_buffer_memory import (
+    NodeMemoryData,
+    NodeMemoryFile,
+    NodeTokenBufferMemory,
+)
+from core.memory.token_buffer_memory import TokenBufferMemory
+
+__all__ = [
+    "BaseMemory",
+    "NodeMemoryData",
+    "NodeMemoryFile",
+    "NodeTokenBufferMemory",
+    "TokenBufferMemory",
+]

api/core/memory/base.py

@@ -0,0 +1,83 @@
+"""
+Base memory interfaces and types.
+
+This module defines the common protocol for memory implementations.
+"""
+
+from abc import ABC, abstractmethod
+from collections.abc import Sequence
+
+from core.model_runtime.entities import ImagePromptMessageContent, PromptMessage
+
+
+class BaseMemory(ABC):
+    """
+    Abstract base class for memory implementations.
+
+    Provides a common interface for both conversation-level and node-level memory.
+    """
+
+    @abstractmethod
+    def get_history_prompt_messages(
+        self,
+        *,
+        max_token_limit: int = 2000,
+        message_limit: int | None = None,
+    ) -> Sequence[PromptMessage]:
+        """
+        Get history prompt messages.
+
+        :param max_token_limit: Maximum tokens for history
+        :param message_limit: Maximum number of messages
+        :return: Sequence of PromptMessage for LLM context
+        """
+        pass
+
+    def get_history_prompt_text(
+        self,
+        human_prefix: str = "Human",
+        ai_prefix: str = "Assistant",
+        max_token_limit: int = 2000,
+        message_limit: int | None = None,
+    ) -> str:
+        """
+        Get history prompt as formatted text.
+
+        :param human_prefix: Prefix for human messages
+        :param ai_prefix: Prefix for assistant messages
+        :param max_token_limit: Maximum tokens for history
+        :param message_limit: Maximum number of messages
+        :return: Formatted history text
+        """
+        from core.model_runtime.entities import (
+            PromptMessageRole,
+            TextPromptMessageContent,
+        )
+
+        prompt_messages = self.get_history_prompt_messages(
+            max_token_limit=max_token_limit,
+            message_limit=message_limit,
+        )
+
+        string_messages = []
+        for m in prompt_messages:
+            if m.role == PromptMessageRole.USER:
+                role = human_prefix
+            elif m.role == PromptMessageRole.ASSISTANT:
+                role = ai_prefix
+            else:
+                continue
+
+            if isinstance(m.content, list):
+                inner_msg = ""
+                for content in m.content:
+                    if isinstance(content, TextPromptMessageContent):
+                        inner_msg += f"{content.data}\n"
+                    elif isinstance(content, ImagePromptMessageContent):
+                        inner_msg += "[image]\n"
+                string_messages.append(f"{role}: {inner_msg.strip()}")
+            else:
+                message = f"{role}: {m.content}"
+                string_messages.append(message)
+
+        return "\n".join(string_messages)

api/core/memory/node_token_buffer_memory.py

@@ -0,0 +1,353 @@
+"""
+Node-level Token Buffer Memory for Chatflow.
+
+This module provides node-scoped memory within a conversation.
+Each LLM node in a workflow can maintain its own independent conversation history.
+
+Note: This is only available in Chatflow (advanced-chat mode) because it requires
+both conversation_id and node_id.
+
+Design:
+- Storage is indexed by workflow_run_id (each execution stores one turn)
+- Thread tracking leverages Message table's parent_message_id structure
+- On read: query Message table for current thread, then filter Node Memory by workflow_run_ids
+"""
+
+import logging
+from collections.abc import Sequence
+
+from pydantic import BaseModel
+from sqlalchemy import select
+
+from core.file import File, FileTransferMethod
+from core.memory.base import BaseMemory
+from core.model_manager import ModelInstance
+from core.model_runtime.entities import (
+    AssistantPromptMessage,
+    ImagePromptMessageContent,
+    PromptMessage,
+    TextPromptMessageContent,
+    UserPromptMessage,
+)
+from core.prompt.utils.extract_thread_messages import extract_thread_messages
+from extensions.ext_database import db
+from extensions.ext_storage import storage
+from models.model import Message
+
+logger = logging.getLogger(__name__)
+
+
+class NodeMemoryFile(BaseModel):
+    """File reference stored in node memory."""
+
+    type: str  # image, audio, video, document, custom
+    transfer_method: str  # local_file, remote_url, tool_file
+    upload_file_id: str | None = None
+    tool_file_id: str | None = None
+    url: str | None = None
+
+
+class NodeMemoryTurn(BaseModel):
+    """A single dialogue turn (user + assistant) in node memory."""
+
+    user_content: str = ""
+    user_files: list[NodeMemoryFile] = []
+    assistant_content: str = ""
+    assistant_files: list[NodeMemoryFile] = []
+
+
+class NodeMemoryData(BaseModel):
+    """Root data structure for node memory storage."""
+
+    version: int = 1
+    # Key: workflow_run_id, Value: dialogue turn
+    turns: dict[str, NodeMemoryTurn] = {}
+
+
+class NodeTokenBufferMemory(BaseMemory):
+    """
+    Node-level Token Buffer Memory.
+
+    Provides node-scoped memory within a conversation. Each LLM node can maintain
+    its own independent conversation history, stored in object storage.
+
+    Key design: Thread tracking is delegated to Message table's parent_message_id.
+    Storage is indexed by workflow_run_id for easy filtering.
+
+    Storage key format: node_memory/{app_id}/{conversation_id}/{node_id}.json
+    """
+
+    def __init__(
+        self,
+        app_id: str,
+        conversation_id: str,
+        node_id: str,
+        tenant_id: str,
+        model_instance: ModelInstance,
+    ):
+        """
+        Initialize node-level memory.
+
+        :param app_id: Application ID
+        :param conversation_id: Conversation ID
+        :param node_id: Node ID in the workflow
+        :param tenant_id: Tenant ID for file reconstruction
+        :param model_instance: Model instance for token counting
+        """
+        self.app_id = app_id
+        self.conversation_id = conversation_id
+        self.node_id = node_id
+        self.tenant_id = tenant_id
+        self.model_instance = model_instance
+        self._storage_key = f"node_memory/{app_id}/{conversation_id}/{node_id}.json"
+        self._data: NodeMemoryData | None = None
+        self._dirty = False
+
+    def _load(self) -> NodeMemoryData:
+        """Load data from object storage."""
+        if self._data is not None:
+            return self._data
+
+        try:
+            raw = storage.load_once(self._storage_key)
+            self._data = NodeMemoryData.model_validate_json(raw)
+        except Exception:
+            # File not found or parse error, start fresh
+            self._data = NodeMemoryData()
+
+        return self._data
+
+    def _save(self) -> None:
+        """Save data to object storage."""
+        if self._data is not None:
+            storage.save(self._storage_key, self._data.model_dump_json().encode("utf-8"))
+            self._dirty = False
+
+    def _file_to_memory_file(self, file: File) -> NodeMemoryFile:
+        """Convert File object to NodeMemoryFile reference."""
+        return NodeMemoryFile(
+            type=file.type.value if hasattr(file.type, "value") else str(file.type),
+            transfer_method=(
+                file.transfer_method.value if hasattr(file.transfer_method, "value") else str(file.transfer_method)
+            ),
+            upload_file_id=file.related_id if file.transfer_method == FileTransferMethod.LOCAL_FILE else None,
+            tool_file_id=file.related_id if file.transfer_method == FileTransferMethod.TOOL_FILE else None,
+            url=file.remote_url if file.transfer_method == FileTransferMethod.REMOTE_URL else None,
+        )
+
+    def _memory_file_to_mapping(self, memory_file: NodeMemoryFile) -> dict:
+        """Convert NodeMemoryFile to mapping for file_factory."""
+        mapping: dict = {
+            "type": memory_file.type,
+            "transfer_method": memory_file.transfer_method,
+        }
+        if memory_file.upload_file_id:
+            mapping["upload_file_id"] = memory_file.upload_file_id
+        if memory_file.tool_file_id:
+            mapping["tool_file_id"] = memory_file.tool_file_id
+        if memory_file.url:
+            mapping["url"] = memory_file.url
+        return mapping
+
+    def _rebuild_files(self, memory_files: list[NodeMemoryFile]) -> list[File]:
+        """Rebuild File objects from NodeMemoryFile references."""
+        if not memory_files:
+            return []
+
+        from factories import file_factory
+
+        files = []
+        for mf in memory_files:
+            try:
+                mapping = self._memory_file_to_mapping(mf)
+                file = file_factory.build_from_mapping(mapping=mapping, tenant_id=self.tenant_id)
+                files.append(file)
+            except Exception as e:
+                logger.warning("Failed to rebuild file from memory: %s", e)
+                continue
+        return files
+
+    def _build_prompt_message(
+        self,
+        role: str,
+        content: str,
+        files: list[File],
+        detail: ImagePromptMessageContent.DETAIL = ImagePromptMessageContent.DETAIL.HIGH,
+    ) -> PromptMessage:
+        """Build PromptMessage from content and files."""
+        from core.file import file_manager
+
+        if not files:
+            if role == "user":
+                return UserPromptMessage(content=content)
+            else:
+                return AssistantPromptMessage(content=content)
+
+        # Build multimodal content
+        prompt_contents: list = []
+        for file in files:
+            try:
+                prompt_content = file_manager.to_prompt_message_content(file, image_detail_config=detail)
+                prompt_contents.append(prompt_content)
+            except Exception as e:
+                logger.warning("Failed to convert file to prompt content: %s", e)
+                continue
+
+        prompt_contents.append(TextPromptMessageContent(data=content))
+
+        if role == "user":
+            return UserPromptMessage(content=prompt_contents)
+        else:
+            return AssistantPromptMessage(content=prompt_contents)
+
+    def _get_thread_workflow_run_ids(self) -> list[str]:
+        """
+        Get workflow_run_ids for the current thread by querying Message table.
+
+        Returns workflow_run_ids in chronological order (oldest first).
+        """
+        # Query messages for this conversation
+        stmt = (
+            select(Message).where(Message.conversation_id == self.conversation_id).order_by(Message.created_at.desc())
+        )
+        messages = db.session.scalars(stmt.limit(500)).all()
+
+        if not messages:
+            return []
+
+        # Extract thread messages using existing logic
+        thread_messages = extract_thread_messages(messages)
+
+        # For newly created message, its answer is temporarily empty, skip it
+        if thread_messages and not thread_messages[0].answer and thread_messages[0].answer_tokens == 0:
+            thread_messages.pop(0)
+
+        # Reverse to get chronological order, extract workflow_run_ids
+        workflow_run_ids = []
+        for msg in reversed(thread_messages):
+            if msg.workflow_run_id:
+                workflow_run_ids.append(msg.workflow_run_id)
+
+        return workflow_run_ids
+
+    def add_messages(
+        self,
+        workflow_run_id: str,
+        user_content: str,
+        user_files: Sequence[File] | None = None,
+        assistant_content: str = "",
+        assistant_files: Sequence[File] | None = None,
+    ) -> None:
+        """
+        Add a dialogue turn to node memory.
+        Call this after LLM node execution completes.
+
+        :param workflow_run_id: Current workflow execution ID
+        :param user_content: User's text input
+        :param user_files: Files attached by user
+        :param assistant_content: Assistant's text response
+        :param assistant_files: Files generated by assistant
+        """
+        data = self._load()
+
+        # Convert files to memory file references
+        user_memory_files = [self._file_to_memory_file(f) for f in (user_files or [])]
+        assistant_memory_files = [self._file_to_memory_file(f) for f in (assistant_files or [])]
+
+        # Store the turn indexed by workflow_run_id
+        data.turns[workflow_run_id] = NodeMemoryTurn(
+            user_content=user_content,
+            user_files=user_memory_files,
+            assistant_content=assistant_content,
+            assistant_files=assistant_memory_files,
+        )
+
+        self._dirty = True
+
+    def get_history_prompt_messages(
+        self,
+        *,
+        max_token_limit: int = 2000,
+        message_limit: int | None = None,
+    ) -> Sequence[PromptMessage]:
+        """
+        Retrieve history as PromptMessage sequence.
+
+        Thread tracking is handled by querying Message table's parent_message_id structure.
+
+        :param max_token_limit: Maximum tokens for history
+        :param message_limit: unused, for interface compatibility
+        :return: Sequence of PromptMessage for LLM context
+        """
+        # message_limit is unused in NodeTokenBufferMemory (uses token limit instead)
+        _ = message_limit
+        detail = ImagePromptMessageContent.DETAIL.HIGH
+        data = self._load()
+
+        if not data.turns:
+            return []
+
+        # Get workflow_run_ids for current thread from Message table
+        thread_workflow_run_ids = self._get_thread_workflow_run_ids()
+
+        if not thread_workflow_run_ids:
+            return []
+
+        # Build prompt messages in thread order
+        prompt_messages: list[PromptMessage] = []
+        for wf_run_id in thread_workflow_run_ids:
+            turn = data.turns.get(wf_run_id)
+            if not turn:
+                # This workflow execution didn't have node memory stored
+                continue
+
+            # Build user message
+            user_files = self._rebuild_files(turn.user_files) if turn.user_files else []
+            user_msg = self._build_prompt_message(
+                role="user",
+                content=turn.user_content,
+                files=user_files,
+                detail=detail,
+            )
+            prompt_messages.append(user_msg)
+
+            # Build assistant message
+            assistant_files = self._rebuild_files(turn.assistant_files) if turn.assistant_files else []
+            assistant_msg = self._build_prompt_message(
+                role="assistant",
+                content=turn.assistant_content,
+                files=assistant_files,
+                detail=detail,
+            )
+            prompt_messages.append(assistant_msg)
+
+        if not prompt_messages:
+            return []
+
+        # Truncate by token limit
+        try:
+            current_tokens = self.model_instance.get_llm_num_tokens(prompt_messages)
+            while current_tokens > max_token_limit and len(prompt_messages) > 1:
+                prompt_messages.pop(0)
+                current_tokens = self.model_instance.get_llm_num_tokens(prompt_messages)
+        except Exception as e:
+            logger.warning("Failed to count tokens for truncation: %s", e)
+
+        return prompt_messages
+
+    def flush(self) -> None:
+        """
+        Persist buffered changes to object storage.
+        Call this at the end of node execution.
+        """
+        if self._dirty:
+            self._save()
+
+    def clear(self) -> None:
+        """Clear all messages in this node's memory."""
+        self._data = NodeMemoryData()
+        self._save()
+
+    def exists(self) -> bool:
+        """Check if node memory exists in storage."""
+        return storage.exists(self._storage_key)

api/core/memory/token_buffer_memory.py

@@ -5,12 +5,12 @@ from sqlalchemy.orm import sessionmaker
 
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.file import file_manager
+from core.memory.base import BaseMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities import (
     AssistantPromptMessage,
     ImagePromptMessageContent,
     PromptMessage,
-    PromptMessageRole,
     TextPromptMessageContent,
     UserPromptMessage,
 )
@@ -24,7 +24,7 @@ from repositories.api_workflow_run_repository import APIWorkflowRunRepository
 from repositories.factory import DifyAPIRepositoryFactory
 
 
-class TokenBufferMemory:
+class TokenBufferMemory(BaseMemory):
     def __init__(
         self,
         conversation: Conversation,
@@ -115,10 +115,14 @@ class TokenBufferMemory:
                 return AssistantPromptMessage(content=prompt_message_contents)
 
     def get_history_prompt_messages(
-        self, max_token_limit: int = 2000, message_limit: int | None = None
+        self,
+        *,
+        max_token_limit: int = 2000,
+        message_limit: int | None = None,
     ) -> Sequence[PromptMessage]:
         """
         Get history prompt messages.
+
         :param max_token_limit: max token limit
         :param message_limit: message limit
         """
@@ -200,44 +204,3 @@ class TokenBufferMemory:
                 curr_message_tokens = self.model_instance.get_llm_num_tokens(prompt_messages)
 
         return prompt_messages
-
-    def get_history_prompt_text(
-        self,
-        human_prefix: str = "Human",
-        ai_prefix: str = "Assistant",
-        max_token_limit: int = 2000,
-        message_limit: int | None = None,
-    ) -> str:
-        """
-        Get history prompt text.
-        :param human_prefix: human prefix
-        :param ai_prefix: ai prefix
-        :param max_token_limit: max token limit
-        :param message_limit: message limit
-        :return:
-        """
-        prompt_messages = self.get_history_prompt_messages(max_token_limit=max_token_limit, message_limit=message_limit)
-
-        string_messages = []
-        for m in prompt_messages:
-            if m.role == PromptMessageRole.USER:
-                role = human_prefix
-            elif m.role == PromptMessageRole.ASSISTANT:
-                role = ai_prefix
-            else:
-                continue
-
-            if isinstance(m.content, list):
-                inner_msg = ""
-                for content in m.content:
-                    if isinstance(content, TextPromptMessageContent):
-                        inner_msg += f"{content.data}\n"
-                    elif isinstance(content, ImagePromptMessageContent):
-                        inner_msg += "[image]\n"
-
-                string_messages.append(f"{role}: {inner_msg.strip()}")
-            else:
-                message = f"{role}: {m.content}"
-                string_messages.append(message)
-
-        return "\n".join(string_messages)

api/core/ops/aliyun_trace/aliyun_trace.py

@@ -55,7 +55,7 @@ from core.ops.entities.trace_entity import (
     ToolTraceInfo,
     WorkflowTraceInfo,
 )
-from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
+from core.repositories import DifyCoreRepositoryFactory
 from core.workflow.entities import WorkflowNodeExecution
 from core.workflow.enums import NodeType, WorkflowNodeExecutionMetadataKey
 from extensions.ext_database import db
@@ -275,7 +275,7 @@ class AliyunDataTrace(BaseTraceInstance):
         service_account = self.get_service_account_with_tenant(app_id)
 
         session_factory = sessionmaker(bind=db.engine)
-        workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
+        workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
             session_factory=session_factory,
             user=service_account,
             app_id=app_id,

api/core/plugin/impl/endpoint.py

@@ -1,5 +1,6 @@
 from core.plugin.entities.endpoint import EndpointEntityWithInstance
 from core.plugin.impl.base import BasePluginClient
+from core.plugin.impl.exc import PluginDaemonInternalServerError
 
 
 class PluginEndpointClient(BasePluginClient):
@@ -70,18 +71,27 @@ class PluginEndpointClient(BasePluginClient):
     def delete_endpoint(self, tenant_id: str, user_id: str, endpoint_id: str):
         """
         Delete the given endpoint.
+
+        This operation is idempotent: if the endpoint is already deleted (record not found),
+        it will return True instead of raising an error.
         """
-        return self._request_with_plugin_daemon_response(
-            "POST",
-            f"plugin/{tenant_id}/endpoint/remove",
-            bool,
-            data={
-                "endpoint_id": endpoint_id,
-            },
-            headers={
-                "Content-Type": "application/json",
-            },
-        )
+        try:
+            return self._request_with_plugin_daemon_response(
+                "POST",
+                f"plugin/{tenant_id}/endpoint/remove",
+                bool,
+                data={
+                    "endpoint_id": endpoint_id,
+                },
+                headers={
+                    "Content-Type": "application/json",
+                },
+            )
+        except PluginDaemonInternalServerError as e:
+            # Make delete idempotent: if record is not found, consider it a success
+            if "record not found" in str(e.description).lower():
+                return True
+            raise
 
     def enable_endpoint(self, tenant_id: str, user_id: str, endpoint_id: str):
         """

api/core/prompt/advanced_prompt_transform.py

@@ -5,7 +5,7 @@ from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEnti
 from core.file import file_manager
 from core.file.models import File
 from core.helper.code_executor.jinja2.jinja2_formatter import Jinja2Formatter
-from core.memory.token_buffer_memory import TokenBufferMemory
+from core.memory.base import BaseMemory
 from core.model_runtime.entities import (
     AssistantPromptMessage,
     PromptMessage,
@@ -43,7 +43,7 @@ class AdvancedPromptTransform(PromptTransform):
         files: Sequence[File],
         context: str | None,
         memory_config: MemoryConfig | None,
-        memory: TokenBufferMemory | None,
+        memory: BaseMemory | None,
         model_config: ModelConfigWithCredentialsEntity,
         image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
@@ -84,7 +84,7 @@ class AdvancedPromptTransform(PromptTransform):
         files: Sequence[File],
         context: str | None,
         memory_config: MemoryConfig | None,
-        memory: TokenBufferMemory | None,
+        memory: BaseMemory | None,
         model_config: ModelConfigWithCredentialsEntity,
         image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
@@ -145,7 +145,7 @@ class AdvancedPromptTransform(PromptTransform):
         files: Sequence[File],
         context: str | None,
         memory_config: MemoryConfig | None,
-        memory: TokenBufferMemory | None,
+        memory: BaseMemory | None,
         model_config: ModelConfigWithCredentialsEntity,
         image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
@@ -270,7 +270,7 @@ class AdvancedPromptTransform(PromptTransform):
 
     def _set_histories_variable(
         self,
-        memory: TokenBufferMemory,
+        memory: BaseMemory,
         memory_config: MemoryConfig,
         raw_prompt: str,
         role_prefix: MemoryConfig.RolePrefix,

api/core/prompt/entities/advanced_prompt_entities.py

@@ -1,3 +1,4 @@
+from enum import StrEnum
 from typing import Literal
 
 from pydantic import BaseModel
@@ -5,6 +6,13 @@ from pydantic import BaseModel
 from core.model_runtime.entities.message_entities import PromptMessageRole
 
 
+class MemoryMode(StrEnum):
+    """Memory mode for LLM nodes."""
+
+    CONVERSATION = "conversation"  # Use TokenBufferMemory (default, existing behavior)
+    NODE = "node"  # Use NodeTokenBufferMemory (Chatflow only)
+
+
 class ChatModelMessage(BaseModel):
     """
     Chat Message.
@@ -48,3 +56,4 @@ class MemoryConfig(BaseModel):
     role_prefix: RolePrefix | None = None
     window: WindowConfig
     query_prompt_template: str | None = None
+    mode: MemoryMode = MemoryMode.CONVERSATION

api/core/prompt/prompt_transform.py

@@ -1,7 +1,7 @@
 from typing import Any
 
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.memory.token_buffer_memory import TokenBufferMemory
+from core.memory.base import BaseMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.message_entities import PromptMessage
 from core.model_runtime.entities.model_entities import ModelPropertyKey
@@ -11,7 +11,7 @@ from core.prompt.entities.advanced_prompt_entities import MemoryConfig
 class PromptTransform:
     def _append_chat_histories(
         self,
-        memory: TokenBufferMemory,
+        memory: BaseMemory,
         memory_config: MemoryConfig,
         prompt_messages: list[PromptMessage],
         model_config: ModelConfigWithCredentialsEntity,
@@ -52,7 +52,7 @@ class PromptTransform:
 
     def _get_history_messages_from_memory(
         self,
-        memory: TokenBufferMemory,
+        memory: BaseMemory,
         memory_config: MemoryConfig,
         max_token_limit: int,
         human_prefix: str | None = None,
@@ -73,7 +73,7 @@ class PromptTransform:
         return memory.get_history_prompt_text(**kwargs)
 
     def _get_history_messages_list_from_memory(
-        self, memory: TokenBufferMemory, memory_config: MemoryConfig, max_token_limit: int
+        self, memory: BaseMemory, memory_config: MemoryConfig, max_token_limit: int
     ) -> list[PromptMessage]:
         """Get memory messages."""
         return list(

api/core/tools/tool_manager.py

@@ -1047,6 +1047,8 @@ class ToolManager:
                         continue
                     tool_input = ToolNodeData.ToolInput.model_validate(tool_configurations.get(parameter.name, {}))
                     if tool_input.type == "variable":
+                        if not isinstance(tool_input.value, list):
+                            raise ToolParameterError(f"Invalid variable selector for {parameter.name}")
                         variable = variable_pool.get(tool_input.value)
                         if variable is None:
                             raise ToolParameterError(f"Variable {tool_input.value} does not exist")
@@ -1056,6 +1058,11 @@ class ToolManager:
                     elif tool_input.type == "mixed":
                         segment_group = variable_pool.convert_template(str(tool_input.value))
                         parameter_value = segment_group.text
+                    elif tool_input.type == "mention":
+                        # Mention type not supported in agent mode
+                        raise ToolParameterError(
+                            f"Mention type not supported in agent for parameter '{parameter.name}'"
+                        )
                     else:
                         raise ToolParameterError(f"Unknown tool input type '{tool_input.type}'")
                     runtime_parameters[parameter.name] = parameter_value

api/core/tools/workflow_as_tool/tool.py

@@ -7,8 +7,8 @@ from typing import Any, cast
 
 from flask import has_request_context
 from sqlalchemy import select
-from sqlalchemy.orm import Session
 
+from core.db.session_factory import session_factory
 from core.file import FILE_MODEL_IDENTITY, File, FileTransferMethod
 from core.model_runtime.entities.llm_entities import LLMUsage, LLMUsageMetadata
 from core.tools.__base.tool import Tool
@@ -20,7 +20,6 @@ from core.tools.entities.tool_entities import (
     ToolProviderType,
 )
 from core.tools.errors import ToolInvokeError
-from extensions.ext_database import db
 from factories.file_factory import build_from_mapping
 from libs.login import current_user
 from models import Account, Tenant
@@ -230,30 +229,32 @@ class WorkflowTool(Tool):
         """
         Resolve user from database (worker/Celery context).
         """
+        with session_factory.create_session() as session:
+            tenant_stmt = select(Tenant).where(Tenant.id == self.runtime.tenant_id)
+            tenant = session.scalar(tenant_stmt)
+            if not tenant:
+                return None
+
+            user_stmt = select(Account).where(Account.id == user_id)
+            user = session.scalar(user_stmt)
+            if user:
+                user.current_tenant = tenant
+                session.expunge(user)
+                return user
+
+            end_user_stmt = select(EndUser).where(EndUser.id == user_id, EndUser.tenant_id == tenant.id)
+            end_user = session.scalar(end_user_stmt)
+            if end_user:
+                session.expunge(end_user)
+                return end_user
 
-        tenant_stmt = select(Tenant).where(Tenant.id == self.runtime.tenant_id)
-        tenant = db.session.scalar(tenant_stmt)
-        if not tenant:
             return None
 
-        user_stmt = select(Account).where(Account.id == user_id)
-        user = db.session.scalar(user_stmt)
-        if user:
-            user.current_tenant = tenant
-            return user
-
-        end_user_stmt = select(EndUser).where(EndUser.id == user_id, EndUser.tenant_id == tenant.id)
-        end_user = db.session.scalar(end_user_stmt)
-        if end_user:
-            return end_user
-
-        return None
-
     def _get_workflow(self, app_id: str, version: str) -> Workflow:
         """
         get the workflow by app id and version
         """
-        with Session(db.engine, expire_on_commit=False) as session, session.begin():
+        with session_factory.create_session() as session, session.begin():
             if not version:
                 stmt = (
                     select(Workflow)
@@ -265,22 +266,24 @@ class WorkflowTool(Tool):
                 stmt = select(Workflow).where(Workflow.app_id == app_id, Workflow.version == version)
                 workflow = session.scalar(stmt)
 
-        if not workflow:
-            raise ValueError("workflow not found or not published")
+            if not workflow:
+                raise ValueError("workflow not found or not published")
 
-        return workflow
+            session.expunge(workflow)
+            return workflow
 
     def _get_app(self, app_id: str) -> App:
         """
         get the app by app id
         """
         stmt = select(App).where(App.id == app_id)
-        with Session(db.engine, expire_on_commit=False) as session, session.begin():
+        with session_factory.create_session() as session, session.begin():
             app = session.scalar(stmt)
-        if not app:
-            raise ValueError("app not found")
+            if not app:
+                raise ValueError("app not found")
 
-        return app
+            session.expunge(app)
+            return app
 
     def _transform_args(self, tool_parameters: dict) -> tuple[dict, list[dict]]:
         """

api/core/variables/__init__.py

@@ -4,6 +4,7 @@ from .segments import (
     ArrayFileSegment,
     ArrayNumberSegment,
     ArrayObjectSegment,
+    ArrayPromptMessageSegment,
     ArraySegment,
     ArrayStringSegment,
     FileSegment,
@@ -20,6 +21,7 @@ from .variables import (
     ArrayFileVariable,
     ArrayNumberVariable,
     ArrayObjectVariable,
+    ArrayPromptMessageVariable,
     ArrayStringVariable,
     ArrayVariable,
     FileVariable,
@@ -30,6 +32,7 @@ from .variables import (
     SecretVariable,
     StringVariable,
     Variable,
+    VariableBase,
 )
 
 __all__ = [
@@ -41,6 +44,8 @@ __all__ = [
     "ArrayNumberVariable",
     "ArrayObjectSegment",
     "ArrayObjectVariable",
+    "ArrayPromptMessageSegment",
+    "ArrayPromptMessageVariable",
     "ArraySegment",
     "ArrayStringSegment",
     "ArrayStringVariable",
@@ -62,4 +67,5 @@ __all__ = [
     "StringSegment",
     "StringVariable",
     "Variable",
+    "VariableBase",
 ]

api/core/variables/segments.py

@@ -6,6 +6,7 @@ from typing import Annotated, Any, TypeAlias
 from pydantic import BaseModel, ConfigDict, Discriminator, Tag, field_validator
 
 from core.file import File
+from core.model_runtime.entities import PromptMessage
 
 from .types import SegmentType
 
@@ -208,6 +209,15 @@ class ArrayBooleanSegment(ArraySegment):
     value: Sequence[bool]
 
 
+class ArrayPromptMessageSegment(ArraySegment):
+    value_type: SegmentType = SegmentType.ARRAY_PROMPT_MESSAGE
+    value: Sequence[PromptMessage]
+
+    def to_object(self):
+        """Convert to JSON-serializable format for database storage and frontend."""
+        return [msg.model_dump() for msg in self.value]
+
+
 def get_segment_discriminator(v: Any) -> SegmentType | None:
     if isinstance(v, Segment):
         return v.value_type
@@ -232,7 +242,7 @@ def get_segment_discriminator(v: Any) -> SegmentType | None:
 # - All variants in `SegmentUnion` must inherit from the `Segment` class.
 # - The union must include all non-abstract subclasses of `Segment`, except:
 #   - `SegmentGroup`, which is not added to the variable pool.
-#   - `Variable` and its subclasses, which are handled by `VariableUnion`.
+#   - `VariableBase` and its subclasses, which are handled by `Variable`.
 SegmentUnion: TypeAlias = Annotated[
     (
         Annotated[NoneSegment, Tag(SegmentType.NONE)]
@@ -248,6 +258,7 @@ SegmentUnion: TypeAlias = Annotated[
         | Annotated[ArrayObjectSegment, Tag(SegmentType.ARRAY_OBJECT)]
         | Annotated[ArrayFileSegment, Tag(SegmentType.ARRAY_FILE)]
         | Annotated[ArrayBooleanSegment, Tag(SegmentType.ARRAY_BOOLEAN)]
+        | Annotated[ArrayPromptMessageSegment, Tag(SegmentType.ARRAY_PROMPT_MESSAGE)]
     ),
     Discriminator(get_segment_discriminator),
 ]

api/core/variables/types.py

@@ -45,6 +45,7 @@ class SegmentType(StrEnum):
     ARRAY_OBJECT = "array[object]"
     ARRAY_FILE = "array[file]"
     ARRAY_BOOLEAN = "array[boolean]"
+    ARRAY_PROMPT_MESSAGE = "array[message]"
 
     NONE = "none"
 

api/core/variables/utils.py

@@ -3,8 +3,10 @@ from typing import Any
 
 import orjson
 
+from core.model_runtime.entities import PromptMessage
+
 from .segment_group import SegmentGroup
-from .segments import ArrayFileSegment, FileSegment, Segment
+from .segments import ArrayFileSegment, ArrayPromptMessageSegment, FileSegment, Segment
 
 
 def to_selector(node_id: str, name: str, paths: Iterable[str] = ()) -> Sequence[str]:
@@ -16,7 +18,7 @@ def to_selector(node_id: str, name: str, paths: Iterable[str] = ()) -> Sequence[
 
 def segment_orjson_default(o: Any):
     """Default function for orjson serialization of Segment types"""
-    if isinstance(o, ArrayFileSegment):
+    if isinstance(o, (ArrayFileSegment, ArrayPromptMessageSegment)):
         return [v.model_dump() for v in o.value]
     elif isinstance(o, FileSegment):
         return o.value.model_dump()
@@ -24,6 +26,8 @@ def segment_orjson_default(o: Any):
         return [segment_orjson_default(seg) for seg in o.value]
     elif isinstance(o, Segment):
         return o.value
+    elif isinstance(o, PromptMessage):
+        return o.model_dump()
     raise TypeError(f"Object of type {type(o).__name__} is not JSON serializable")
 
 

api/core/variables/variables.py

@@ -12,6 +12,7 @@ from .segments import (
     ArrayFileSegment,
     ArrayNumberSegment,
     ArrayObjectSegment,
+    ArrayPromptMessageSegment,
     ArraySegment,
     ArrayStringSegment,
     BooleanSegment,
@@ -27,7 +28,7 @@ from .segments import (
 from .types import SegmentType
 
 
-class Variable(Segment):
+class VariableBase(Segment):
     """
     A variable is a segment that has a name.
 
@@ -45,23 +46,23 @@ class Variable(Segment):
     selector: Sequence[str] = Field(default_factory=list)
 
 
-class StringVariable(StringSegment, Variable):
+class StringVariable(StringSegment, VariableBase):
     pass
 
 
-class FloatVariable(FloatSegment, Variable):
+class FloatVariable(FloatSegment, VariableBase):
     pass
 
 
-class IntegerVariable(IntegerSegment, Variable):
+class IntegerVariable(IntegerSegment, VariableBase):
     pass
 
 
-class ObjectVariable(ObjectSegment, Variable):
+class ObjectVariable(ObjectSegment, VariableBase):
     pass
 
 
-class ArrayVariable(ArraySegment, Variable):
+class ArrayVariable(ArraySegment, VariableBase):
     pass
 
 
@@ -89,16 +90,16 @@ class SecretVariable(StringVariable):
         return encrypter.obfuscated_token(self.value)
 
 
-class NoneVariable(NoneSegment, Variable):
+class NoneVariable(NoneSegment, VariableBase):
     value_type: SegmentType = SegmentType.NONE
     value: None = None
 
 
-class FileVariable(FileSegment, Variable):
+class FileVariable(FileSegment, VariableBase):
     pass
 
 
-class BooleanVariable(BooleanSegment, Variable):
+class BooleanVariable(BooleanSegment, VariableBase):
     pass
 
 
@@ -110,6 +111,10 @@ class ArrayBooleanVariable(ArrayBooleanSegment, ArrayVariable):
     pass
 
 
+class ArrayPromptMessageVariable(ArrayPromptMessageSegment, ArrayVariable):
+    pass
+
+
 class RAGPipelineVariable(BaseModel):
     belong_to_node_id: str = Field(description="belong to which node id, shared means public")
     type: str = Field(description="variable type, text-input, paragraph, select, number,  file, file-list")
@@ -139,13 +144,13 @@ class RAGPipelineVariableInput(BaseModel):
     value: Any
 
 
-# The `VariableUnion`` type is used to enable serialization and deserialization with Pydantic.
-# Use `Variable` for type hinting when serialization is not required.
+# The `Variable` type is used to enable serialization and deserialization with Pydantic.
+# Use `VariableBase` for type hinting when serialization is not required.
 #
 # Note:
-# - All variants in `VariableUnion` must inherit from the `Variable` class.
-# - The union must include all non-abstract subclasses of `Segment`, except:
-VariableUnion: TypeAlias = Annotated[
+# - All variants in `Variable` must inherit from the `VariableBase` class.
+# - The union must include all non-abstract subclasses of `VariableBase`.
+Variable: TypeAlias = Annotated[
     (
         Annotated[NoneVariable, Tag(SegmentType.NONE)]
         | Annotated[StringVariable, Tag(SegmentType.STRING)]
@@ -160,6 +165,7 @@ VariableUnion: TypeAlias = Annotated[
         | Annotated[ArrayObjectVariable, Tag(SegmentType.ARRAY_OBJECT)]
         | Annotated[ArrayFileVariable, Tag(SegmentType.ARRAY_FILE)]
         | Annotated[ArrayBooleanVariable, Tag(SegmentType.ARRAY_BOOLEAN)]
+        | Annotated[ArrayPromptMessageVariable, Tag(SegmentType.ARRAY_PROMPT_MESSAGE)]
         | Annotated[SecretVariable, Tag(SegmentType.SECRET)]
     ),
     Discriminator(get_segment_discriminator),

api/core/workflow/conversation_variable_updater.py

@@ -1,7 +1,7 @@
 import abc
 from typing import Protocol
 
-from core.variables import Variable
+from core.variables import VariableBase
 
 
 class ConversationVariableUpdater(Protocol):
@@ -20,12 +20,12 @@ class ConversationVariableUpdater(Protocol):
     """
 
     @abc.abstractmethod
-    def update(self, conversation_id: str, variable: "Variable"):
+    def update(self, conversation_id: str, variable: "VariableBase"):
         """
         Updates the value of the specified conversation variable in the underlying storage.
 
         :param conversation_id: The ID of the conversation to update. Typically references `ConversationVariable.id`.
-        :param variable: The `Variable` instance containing the updated value.
+        :param variable: The `VariableBase` instance containing the updated value.
         """
         pass
 

api/core/workflow/enums.py

@@ -63,6 +63,7 @@ class NodeType(StrEnum):
     TRIGGER_SCHEDULE = "trigger-schedule"
     TRIGGER_PLUGIN = "trigger-plugin"
     HUMAN_INPUT = "human-input"
+    GROUP = "group"
 
     @property
     def is_trigger_node(self) -> bool:
@@ -252,6 +253,7 @@ class WorkflowNodeExecutionMetadataKey(StrEnum):
     LOOP_VARIABLE_MAP = "loop_variable_map"  # single loop variable output
     DATASOURCE_INFO = "datasource_info"
     COMPLETED_REASON = "completed_reason"  # completed reason for loop node
+    MENTION_PARENT_ID = "mention_parent_id"  # parent node id for extractor nodes
 
 
 class WorkflowNodeExecutionStatus(StrEnum):

api/core/workflow/graph/graph.py

@@ -307,7 +307,14 @@ class Graph:
         if not node_configs:
             raise ValueError("Graph must have at least one node")
 
-        node_configs = [node_config for node_config in node_configs if node_config.get("type", "") != "custom-note"]
+        # Filter out UI-only node types:
+        # - custom-note: top-level type (node_config.type == "custom-note")
+        # - group: data-level type (node_config.data.type == "group")
+        node_configs = [
+            node_config
+            for node_config in node_configs
+            if node_config.get("type", "") != "custom-note" and node_config.get("data", {}).get("type", "") != "group"
+        ]
 
         # Parse node configurations
         node_configs_map = cls._parse_node_configs(node_configs)

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

@@ -11,7 +11,7 @@ from typing import Any
 
 from pydantic import BaseModel, Field
 
-from core.variables.variables import VariableUnion
+from core.variables.variables import Variable
 
 
 class CommandType(StrEnum):
@@ -46,7 +46,7 @@ class PauseCommand(GraphEngineCommand):
 class VariableUpdate(BaseModel):
     """Represents a single variable update instruction."""
 
-    value: VariableUnion = Field(description="New variable value")
+    value: Variable = Field(description="New variable value")
 
 
 class UpdateVariablesCommand(GraphEngineCommand):

api/core/workflow/graph_engine/event_management/event_handlers.py

@@ -93,8 +93,8 @@ class EventHandler:
         Args:
             event: The event to handle
         """
-        # Events in loops or iterations are always collected
-        if event.in_loop_id or event.in_iteration_id:
+        # Events in loops, iterations, or extractor groups are always collected
+        if event.in_loop_id or event.in_iteration_id or event.in_mention_parent_id:
             self._event_collector.collect(event)
             return
         return self._dispatch(event)
@@ -125,6 +125,11 @@ class EventHandler:
         Args:
             event: The node started event
         """
+        # Check if this is an extractor node (has parent_node_id)
+        if self._is_extractor_node(event.node_id):
+            self._handle_extractor_node_started(event)
+            return
+
         # Track execution in domain model
         node_execution = self._graph_execution.get_or_create_node_execution(event.node_id)
         is_initial_attempt = node_execution.retry_count == 0
@@ -164,6 +169,11 @@ class EventHandler:
         Args:
             event: The node succeeded event
         """
+        # Check if this is an extractor node (has parent_node_id)
+        if self._is_extractor_node(event.node_id):
+            self._handle_extractor_node_success(event)
+            return
+
         # Update domain model
         node_execution = self._graph_execution.get_or_create_node_execution(event.node_id)
         node_execution.mark_taken()
@@ -226,6 +236,11 @@ class EventHandler:
         Args:
             event: The node failed event
         """
+        # Check if this is an extractor node (has parent_node_id)
+        if self._is_extractor_node(event.node_id):
+            self._handle_extractor_node_failed(event)
+            return
+
         # Update domain model
         node_execution = self._graph_execution.get_or_create_node_execution(event.node_id)
         node_execution.mark_failed(event.error)
@@ -345,3 +360,57 @@ class EventHandler:
                     self._graph_runtime_state.set_output("answer", value)
             else:
                 self._graph_runtime_state.set_output(key, value)
+
+    def _is_extractor_node(self, node_id: str) -> bool:
+        """
+        Check if node_id represents an extractor node (has parent_node_id).
+
+        Extractor nodes extract values from list[PromptMessage] for their parent node.
+        They have a parent_node_id field pointing to their parent node.
+        """
+        node = self._graph.nodes.get(node_id)
+        if node is None:
+            return False
+        return node.node_data.is_extractor_node
+
+    def _handle_extractor_node_started(self, event: NodeRunStartedEvent) -> None:
+        """
+        Handle extractor node started event.
+
+        Extractor nodes don't need full execution tracking, just collect the event.
+        """
+        # Track in response coordinator for stream ordering
+        self._response_coordinator.track_node_execution(event.node_id, event.id)
+
+        # Collect the event
+        self._event_collector.collect(event)
+
+    def _handle_extractor_node_success(self, event: NodeRunSucceededEvent) -> None:
+        """
+        Handle extractor node success event.
+
+        Extractor nodes need special handling:
+        - Store outputs in variable pool (for reference by other nodes)
+        - Accumulate token usage
+        - Collect the event for logging
+        - Do NOT process edges or enqueue next nodes (parent node handles that)
+        """
+        self._accumulate_node_usage(event.node_run_result.llm_usage)
+
+        # Store outputs in variable pool
+        self._store_node_outputs(event.node_id, event.node_run_result.outputs)
+
+        # Collect the event
+        self._event_collector.collect(event)
+
+    def _handle_extractor_node_failed(self, event: NodeRunFailedEvent) -> None:
+        """
+        Handle extractor node failed event.
+
+        Extractor node failures are collected for logging,
+        but the parent node is responsible for handling the error.
+        """
+        self._accumulate_node_usage(event.node_run_result.llm_usage)
+
+        # Collect the event for logging
+        self._event_collector.collect(event)

api/core/workflow/graph_engine/layers/persistence.py

@@ -68,6 +68,7 @@ class _NodeRuntimeSnapshot:
     predecessor_node_id: str | None
     iteration_id: str | None
     loop_id: str | None
+    mention_parent_id: str | None
     created_at: datetime
 
 
@@ -230,6 +231,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
         metadata = {
             WorkflowNodeExecutionMetadataKey.ITERATION_ID: event.in_iteration_id,
             WorkflowNodeExecutionMetadataKey.LOOP_ID: event.in_loop_id,
+            WorkflowNodeExecutionMetadataKey.MENTION_PARENT_ID: event.in_mention_parent_id,
         }
 
         domain_execution = WorkflowNodeExecution(
@@ -256,6 +258,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
             predecessor_node_id=event.predecessor_node_id,
             iteration_id=event.in_iteration_id,
             loop_id=event.in_loop_id,
+            mention_parent_id=event.in_mention_parent_id,
             created_at=event.start_at,
         )
         self._node_snapshots[event.id] = snapshot

api/core/workflow/graph_events/base.py

@@ -21,6 +21,12 @@ class GraphNodeEventBase(GraphEngineEvent):
     """iteration id if node is in iteration"""
     in_loop_id: str | None = None
     """loop id if node is in loop"""
+    in_mention_parent_id: str | None = None
+    """Parent node id if this is an extractor node event.
+    
+    When set, indicates this event belongs to an extractor node that
+    is extracting values for the specified parent node.
+    """
 
     # The version of the node, or "1" if not specified.
     node_version: str = "1"

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

@@ -12,11 +12,14 @@ from sqlalchemy.orm import Session
 from core.agent.entities import AgentToolEntity
 from core.agent.plugin_entities import AgentStrategyParameter
 from core.file import File, FileTransferMethod
+from core.memory.base import BaseMemory
+from core.memory.node_token_buffer_memory import NodeTokenBufferMemory
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance, ModelManager
 from core.model_runtime.entities.llm_entities import LLMUsage, LLMUsageMetadata
 from core.model_runtime.entities.model_entities import AIModelEntity, ModelType
 from core.model_runtime.utils.encoders import jsonable_encoder
+from core.prompt.entities.advanced_prompt_entities import MemoryMode
 from core.provider_manager import ProviderManager
 from core.tools.entities.tool_entities import (
     ToolIdentity,
@@ -136,6 +139,9 @@ class AgentNode(Node[AgentNodeData]):
             )
             return
 
+        # Fetch memory for node memory saving
+        memory = self._fetch_memory_for_save()
+
         try:
             yield from self._transform_message(
                 messages=message_stream,
@@ -149,6 +155,7 @@ class AgentNode(Node[AgentNodeData]):
                 node_type=self.node_type,
                 node_id=self._node_id,
                 node_execution_id=self.id,
+                memory=memory,
             )
         except PluginDaemonClientSideError as e:
             transform_error = AgentMessageTransformError(
@@ -395,8 +402,20 @@ class AgentNode(Node[AgentNodeData]):
             icon = None
         return icon
 
-    def _fetch_memory(self, model_instance: ModelInstance) -> TokenBufferMemory | None:
-        # get conversation id
+    def _fetch_memory(self, model_instance: ModelInstance) -> BaseMemory | None:
+        """
+        Fetch memory based on configuration mode.
+
+        Returns TokenBufferMemory for conversation mode (default),
+        or NodeTokenBufferMemory for node mode (Chatflow only).
+        """
+        node_data = self.node_data
+        memory_config = node_data.memory
+
+        if not memory_config:
+            return None
+
+        # get conversation id (required for both modes in Chatflow)
         conversation_id_variable = self.graph_runtime_state.variable_pool.get(
             ["sys", SystemVariableKey.CONVERSATION_ID]
         )
@@ -404,16 +423,26 @@ class AgentNode(Node[AgentNodeData]):
             return None
         conversation_id = conversation_id_variable.value
 
-        with Session(db.engine, expire_on_commit=False) as session:
-            stmt = select(Conversation).where(Conversation.app_id == self.app_id, Conversation.id == conversation_id)
-            conversation = session.scalar(stmt)
-
-            if not conversation:
-                return None
-
-        memory = TokenBufferMemory(conversation=conversation, model_instance=model_instance)
-
-        return memory
+        # Return appropriate memory type based on mode
+        if memory_config.mode == MemoryMode.NODE:
+            # Node-level memory (Chatflow only)
+            return NodeTokenBufferMemory(
+                app_id=self.app_id,
+                conversation_id=conversation_id,
+                node_id=self._node_id,
+                tenant_id=self.tenant_id,
+                model_instance=model_instance,
+            )
+        else:
+            # Conversation-level memory (default)
+            with Session(db.engine, expire_on_commit=False) as session:
+                stmt = select(Conversation).where(
+                    Conversation.app_id == self.app_id, Conversation.id == conversation_id
+                )
+                conversation = session.scalar(stmt)
+                if not conversation:
+                    return None
+            return TokenBufferMemory(conversation=conversation, model_instance=model_instance)
 
     def _fetch_model(self, value: dict[str, Any]) -> tuple[ModelInstance, AIModelEntity | None]:
         provider_manager = ProviderManager()
@@ -457,6 +486,47 @@ class AgentNode(Node[AgentNodeData]):
         else:
             return [tool for tool in tools if tool.get("type") != ToolProviderType.MCP]
 
+    def _fetch_memory_for_save(self) -> BaseMemory | None:
+        """
+        Fetch memory instance for saving node memory.
+        This is a simplified version that doesn't require model_instance.
+        """
+        from core.model_manager import ModelManager
+        from core.model_runtime.entities.model_entities import ModelType
+
+        node_data = self.node_data
+        if not node_data.memory:
+            return None
+
+        # Get conversation_id
+        conversation_id_var = self.graph_runtime_state.variable_pool.get(["sys", SystemVariableKey.CONVERSATION_ID])
+        if not isinstance(conversation_id_var, StringSegment):
+            return None
+        conversation_id = conversation_id_var.value
+
+        # Return appropriate memory type based on mode
+        if node_data.memory.mode == MemoryMode.NODE:
+            # For node memory, we need a model_instance for token counting
+            # Use a simple default model for this purpose
+            try:
+                model_instance = ModelManager().get_default_model_instance(
+                    tenant_id=self.tenant_id,
+                    model_type=ModelType.LLM,
+                )
+            except Exception:
+                return None
+
+            return NodeTokenBufferMemory(
+                app_id=self.app_id,
+                conversation_id=conversation_id,
+                node_id=self._node_id,
+                tenant_id=self.tenant_id,
+                model_instance=model_instance,
+            )
+        else:
+            # Conversation-level memory doesn't need saving here
+            return None
+
     def _transform_message(
         self,
         messages: Generator[ToolInvokeMessage, None, None],
@@ -467,6 +537,7 @@ class AgentNode(Node[AgentNodeData]):
         node_type: NodeType,
         node_id: str,
         node_execution_id: str,
+        memory: BaseMemory | None = None,
     ) -> Generator[NodeEventBase, None, None]:
         """
         Convert ToolInvokeMessages into tuple[plain_text, files]
@@ -711,6 +782,21 @@ class AgentNode(Node[AgentNodeData]):
                 is_final=True,
             )
 
+        # Save to node memory if in node memory mode
+        from core.workflow.nodes.llm import llm_utils
+
+        # Get user query from sys.query
+        user_query_var = self.graph_runtime_state.variable_pool.get(["sys", SystemVariableKey.QUERY])
+        user_query = user_query_var.text if user_query_var else ""
+
+        llm_utils.save_node_memory(
+            memory=memory,
+            variable_pool=self.graph_runtime_state.variable_pool,
+            user_query=user_query,
+            assistant_response=text,
+            assistant_files=files,
+        )
+
         yield StreamCompletedEvent(
             node_run_result=NodeRunResult(
                 status=WorkflowNodeExecutionStatus.SUCCEEDED,

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

@@ -1,4 +1,10 @@
-from .entities import BaseIterationNodeData, BaseIterationState, BaseLoopNodeData, BaseLoopState, BaseNodeData
+from .entities import (
+    BaseIterationNodeData,
+    BaseIterationState,
+    BaseLoopNodeData,
+    BaseLoopState,
+    BaseNodeData,
+)
 from .usage_tracking_mixin import LLMUsageTrackingMixin
 
 __all__ = [

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

@@ -175,6 +175,16 @@ class BaseNodeData(ABC, BaseModel):
     default_value: list[DefaultValue] | None = None
     retry_config: RetryConfig = RetryConfig()
 
+    # Parent node ID when this node is used as an extractor.
+    # If set, this node is an "attached" extractor node that extracts values
+    # from list[PromptMessage] for the parent node's parameters.
+    parent_node_id: str | None = None
+
+    @property
+    def is_extractor_node(self) -> bool:
+        """Check if this node is an extractor node (has parent_node_id)."""
+        return self.parent_node_id is not None
+
     @property
     def default_value_dict(self) -> dict[str, Any]:
         if self.default_value:

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

@@ -270,10 +270,87 @@ class Node(Generic[NodeDataT]):
         """Check if execution should be stopped."""
         return self.graph_runtime_state.stop_event.is_set()
 
+    def _find_extractor_node_configs(self) -> list[dict[str, Any]]:
+        """
+        Find all extractor node configurations that have parent_node_id == self._node_id.
+
+        Returns:
+            List of node configuration dicts for extractor nodes
+        """
+        nodes = self.graph_config.get("nodes", [])
+        extractor_configs = []
+        for node_config in nodes:
+            node_data = node_config.get("data", {})
+            if node_data.get("parent_node_id") == self._node_id:
+                extractor_configs.append(node_config)
+        return extractor_configs
+
+    def _execute_extractor_nodes(self) -> Generator[GraphNodeEventBase, None, None]:
+        """
+        Execute all extractor nodes associated with this node.
+
+        Extractor nodes are nodes with parent_node_id == self._node_id.
+        They are executed before the main node to extract values from list[PromptMessage].
+        """
+        from core.workflow.nodes.node_mapping import LATEST_VERSION, NODE_TYPE_CLASSES_MAPPING
+
+        extractor_configs = self._find_extractor_node_configs()
+        logger.debug("[Extractor] Found %d extractor nodes for parent '%s'", len(extractor_configs), self._node_id)
+        if not extractor_configs:
+            return
+
+        for config in extractor_configs:
+            node_id = config.get("id")
+            node_data = config.get("data", {})
+            node_type_str = node_data.get("type")
+
+            if not node_id or not node_type_str:
+                continue
+
+            # Get node class
+            try:
+                node_type = NodeType(node_type_str)
+            except ValueError:
+                continue
+
+            node_mapping = NODE_TYPE_CLASSES_MAPPING.get(node_type)
+            if not node_mapping:
+                continue
+
+            node_version = str(node_data.get("version", "1"))
+            node_cls = node_mapping.get(node_version) or node_mapping.get(LATEST_VERSION)
+            if not node_cls:
+                continue
+
+            # Instantiate and execute the extractor node
+            extractor_node = node_cls(
+                id=node_id,
+                config=config,
+                graph_init_params=self._graph_init_params,
+                graph_runtime_state=self.graph_runtime_state,
+            )
+
+            # Execute and process extractor node events
+            for event in extractor_node.run():
+                # Tag event with parent node id for stream ordering and history tracking
+                if isinstance(event, GraphNodeEventBase):
+                    event.in_mention_parent_id = self._node_id
+
+                if isinstance(event, NodeRunSucceededEvent):
+                    # Store extractor node outputs in variable pool
+                    outputs: Mapping[str, Any] = event.node_run_result.outputs
+                    for variable_name, variable_value in outputs.items():
+                        self.graph_runtime_state.variable_pool.add((node_id, variable_name), variable_value)
+                if not isinstance(event, NodeRunStreamChunkEvent):
+                    yield event
+
     def run(self) -> Generator[GraphNodeEventBase, None, None]:
         execution_id = self.ensure_execution_id()
         self._start_at = naive_utc_now()
 
+        # Step 1: Execute associated extractor nodes before main node execution
+        yield from self._execute_extractor_nodes()
+
         # Create and push start event with required fields
         start_event = NodeRunStartedEvent(
             id=execution_id,

api/core/workflow/nodes/http_request/executor.py

@@ -17,6 +17,7 @@ from core.helper import ssrf_proxy
 from core.variables.segments import ArrayFileSegment, FileSegment
 from core.workflow.runtime import VariablePool
 
+from ..protocols import FileManagerProtocol, HttpClientProtocol
 from .entities import (
     HttpRequestNodeAuthorization,
     HttpRequestNodeData,
@@ -78,6 +79,8 @@ class Executor:
         timeout: HttpRequestNodeTimeout,
         variable_pool: VariablePool,
         max_retries: int = dify_config.SSRF_DEFAULT_MAX_RETRIES,
+        http_client: HttpClientProtocol = ssrf_proxy,
+        file_manager: FileManagerProtocol = file_manager,
     ):
         # If authorization API key is present, convert the API key using the variable pool
         if node_data.authorization.type == "api-key":
@@ -104,6 +107,8 @@ class Executor:
         self.data = None
         self.json = None
         self.max_retries = max_retries
+        self._http_client = http_client
+        self._file_manager = file_manager
 
         # init template
         self.variable_pool = variable_pool
@@ -200,7 +205,7 @@ class Executor:
                     if file_variable is None:
                         raise FileFetchError(f"cannot fetch file with selector {file_selector}")
                     file = file_variable.value
-                    self.content = file_manager.download(file)
+                    self.content = self._file_manager.download(file)
                 case "x-www-form-urlencoded":
                     form_data = {
                         self.variable_pool.convert_template(item.key).text: self.variable_pool.convert_template(
@@ -239,7 +244,7 @@ class Executor:
                             ):
                                 file_tuple = (
                                     file.filename,
-                                    file_manager.download(file),
+                                    self._file_manager.download(file),
                                     file.mime_type or "application/octet-stream",
                                 )
                                 if key not in files:
@@ -332,19 +337,18 @@ class Executor:
         do http request depending on api bundle
         """
         _METHOD_MAP = {
-            "get": ssrf_proxy.get,
-            "head": ssrf_proxy.head,
-            "post": ssrf_proxy.post,
-            "put": ssrf_proxy.put,
-            "delete": ssrf_proxy.delete,
-            "patch": ssrf_proxy.patch,
+            "get": self._http_client.get,
+            "head": self._http_client.head,
+            "post": self._http_client.post,
+            "put": self._http_client.put,
+            "delete": self._http_client.delete,
+            "patch": self._http_client.patch,
         }
         method_lc = self.method.lower()
         if method_lc not in _METHOD_MAP:
             raise InvalidHttpMethodError(f"Invalid http method {self.method}")
 
         request_args = {
-            "url": self.url,
             "data": self.data,
             "files": self.files,
             "json": self.json,
@@ -357,8 +361,12 @@ class Executor:
         }
         # request_args = {k: v for k, v in request_args.items() if v is not None}
         try:
-            response: httpx.Response = _METHOD_MAP[method_lc](**request_args, max_retries=self.max_retries)
-        except (ssrf_proxy.MaxRetriesExceededError, httpx.RequestError) as e:
+            response: httpx.Response = _METHOD_MAP[method_lc](
+                url=self.url,
+                **request_args,
+                max_retries=self.max_retries,
+            )
+        except (self._http_client.max_retries_exceeded_error, self._http_client.request_error) as e:
             raise HttpRequestNodeError(str(e)) from e
         # FIXME: fix type ignore, this maybe httpx type issue
         return response

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

@@ -1,10 +1,11 @@
 import logging
 import mimetypes
-from collections.abc import Mapping, Sequence
-from typing import Any
+from collections.abc import Callable, Mapping, Sequence
+from typing import TYPE_CHECKING, Any
 
 from configs import dify_config
-from core.file import File, FileTransferMethod
+from core.file import File, FileTransferMethod, file_manager
+from core.helper import ssrf_proxy
 from core.tools.tool_file_manager import ToolFileManager
 from core.variables.segments import ArrayFileSegment
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
@@ -13,6 +14,7 @@ from core.workflow.nodes.base import variable_template_parser
 from core.workflow.nodes.base.entities import VariableSelector
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.http_request.executor import Executor
+from core.workflow.nodes.protocols import FileManagerProtocol, HttpClientProtocol
 from factories import file_factory
 
 from .entities import (
@@ -30,10 +32,35 @@ HTTP_REQUEST_DEFAULT_TIMEOUT = HttpRequestNodeTimeout(
 
 logger = logging.getLogger(__name__)
 
+if TYPE_CHECKING:
+    from core.workflow.entities import GraphInitParams
+    from core.workflow.runtime import GraphRuntimeState
+
 
 class HttpRequestNode(Node[HttpRequestNodeData]):
     node_type = NodeType.HTTP_REQUEST
 
+    def __init__(
+        self,
+        id: str,
+        config: Mapping[str, Any],
+        graph_init_params: "GraphInitParams",
+        graph_runtime_state: "GraphRuntimeState",
+        *,
+        http_client: HttpClientProtocol = ssrf_proxy,
+        tool_file_manager_factory: Callable[[], ToolFileManager] = ToolFileManager,
+        file_manager: FileManagerProtocol = file_manager,
+    ) -> None:
+        super().__init__(
+            id=id,
+            config=config,
+            graph_init_params=graph_init_params,
+            graph_runtime_state=graph_runtime_state,
+        )
+        self._http_client = http_client
+        self._tool_file_manager_factory = tool_file_manager_factory
+        self._file_manager = file_manager
+
     @classmethod
     def get_default_config(cls, filters: Mapping[str, object] | None = None) -> Mapping[str, object]:
         return {
@@ -71,6 +98,8 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
                 timeout=self._get_request_timeout(self.node_data),
                 variable_pool=self.graph_runtime_state.variable_pool,
                 max_retries=0,
+                http_client=self._http_client,
+                file_manager=self._file_manager,
             )
             process_data["request"] = http_executor.to_log()
 
@@ -199,7 +228,7 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
         mime_type = (
             content_disposition_type or content_type or mimetypes.guess_type(filename)[0] or "application/octet-stream"
         )
-        tool_file_manager = ToolFileManager()
+        tool_file_manager = self._tool_file_manager_factory()
 
         tool_file = tool_file_manager.create_file_by_raw(
             user_id=self.user_id,

api/core/workflow/nodes/iteration/iteration_node.py

@@ -11,7 +11,7 @@ from typing_extensions import TypeIs
 from core.model_runtime.entities.llm_entities import LLMUsage
 from core.variables import IntegerVariable, NoneSegment
 from core.variables.segments import ArrayAnySegment, ArraySegment
-from core.variables.variables import VariableUnion
+from core.variables.variables import Variable
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID
 from core.workflow.enums import (
     NodeExecutionType,
@@ -240,7 +240,7 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
                         datetime,
                         list[GraphNodeEventBase],
                         object | None,
-                        dict[str, VariableUnion],
+                        dict[str, Variable],
                         LLMUsage,
                     ]
                 ],
@@ -308,7 +308,7 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
         item: object,
         flask_app: Flask,
         context_vars: contextvars.Context,
-    ) -> tuple[datetime, list[GraphNodeEventBase], object | None, dict[str, VariableUnion], LLMUsage]:
+    ) -> tuple[datetime, list[GraphNodeEventBase], object | None, dict[str, Variable], LLMUsage]:
         """Execute a single iteration in parallel mode and return results."""
         with preserve_flask_contexts(flask_app=flask_app, context_vars=context_vars):
             iter_start_at = datetime.now(UTC).replace(tzinfo=None)
@@ -515,11 +515,11 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
 
         return variable_mapping
 
-    def _extract_conversation_variable_snapshot(self, *, variable_pool: VariablePool) -> dict[str, VariableUnion]:
+    def _extract_conversation_variable_snapshot(self, *, variable_pool: VariablePool) -> dict[str, Variable]:
         conversation_variables = variable_pool.variable_dictionary.get(CONVERSATION_VARIABLE_NODE_ID, {})
         return {name: variable.model_copy(deep=True) for name, variable in conversation_variables.items()}
 
-    def _sync_conversation_variables_from_snapshot(self, snapshot: dict[str, VariableUnion]) -> None:
+    def _sync_conversation_variables_from_snapshot(self, snapshot: dict[str, Variable]) -> None:
         parent_pool = self.graph_runtime_state.variable_pool
         parent_conversations = parent_pool.variable_dictionary.get(CONVERSATION_VARIABLE_NODE_ID, {})
 

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

@@ -1,7 +1,7 @@
 from collections.abc import Mapping, Sequence
-from typing import Any, Literal
+from typing import Annotated, Any, Literal, TypeAlias
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 from core.model_runtime.entities import ImagePromptMessageContent, LLMMode
 from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate, MemoryConfig
@@ -58,9 +58,28 @@ class LLMNodeCompletionModelPromptTemplate(CompletionModelPromptTemplate):
     jinja2_text: str | None = None
 
 
+class PromptMessageContext(BaseModel):
+    """Context variable reference in prompt template.
+
+    YAML/JSON format: { "$context": ["node_id", "variable_name"] }
+    This will be expanded to list[PromptMessage] at runtime.
+    """
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    value_selector: Sequence[str] = Field(alias="$context")
+
+
+# Union type for prompt template items (static message or context variable reference)
+PromptTemplateItem: TypeAlias = Annotated[
+    LLMNodeChatModelMessage | PromptMessageContext,
+    Field(discriminator=None),
+]
+
+
 class LLMNodeData(BaseNodeData):
     model: ModelConfig
-    prompt_template: Sequence[LLMNodeChatModelMessage] | LLMNodeCompletionModelPromptTemplate
+    prompt_template: Sequence[PromptTemplateItem] | LLMNodeCompletionModelPromptTemplate
     prompt_config: PromptConfig = Field(default_factory=PromptConfig)
     memory: MemoryConfig | None = None
     context: ContextConfig

api/core/workflow/nodes/llm/llm_utils.py

@@ -8,12 +8,13 @@ from configs import dify_config
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.entities.provider_entities import ProviderQuotaType, QuotaUnit
 from core.file.models import File
-from core.memory.token_buffer_memory import TokenBufferMemory
+from core.memory import NodeTokenBufferMemory, TokenBufferMemory
+from core.memory.base import BaseMemory
 from core.model_manager import ModelInstance, ModelManager
 from core.model_runtime.entities.llm_entities import LLMUsage
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from core.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.prompt.entities.advanced_prompt_entities import MemoryConfig, MemoryMode
 from core.variables.segments import ArrayAnySegment, ArrayFileSegment, FileSegment, NoneSegment, StringSegment
 from core.workflow.enums import SystemVariableKey
 from core.workflow.nodes.llm.entities import ModelConfig
@@ -86,25 +87,100 @@ def fetch_files(variable_pool: VariablePool, selector: Sequence[str]) -> Sequenc
 
 
 def fetch_memory(
-    variable_pool: VariablePool, app_id: str, node_data_memory: MemoryConfig | None, model_instance: ModelInstance
-) -> TokenBufferMemory | None:
+    variable_pool: VariablePool,
+    app_id: str,
+    tenant_id: str,
+    node_data_memory: MemoryConfig | None,
+    model_instance: ModelInstance,
+    node_id: str = "",
+) -> BaseMemory | None:
+    """
+    Fetch memory based on configuration mode.
+
+    Returns TokenBufferMemory for conversation mode (default),
+    or NodeTokenBufferMemory for node mode (Chatflow only).
+
+    :param variable_pool: Variable pool containing system variables
+    :param app_id: Application ID
+    :param tenant_id: Tenant ID
+    :param node_data_memory: Memory configuration
+    :param model_instance: Model instance for token counting
+    :param node_id: Node ID in the workflow (required for node mode)
+    :return: Memory instance or None if not applicable
+    """
     if not node_data_memory:
         return None
 
-    # get conversation id
+    # Get conversation_id from variable pool (required for both modes in Chatflow)
     conversation_id_variable = variable_pool.get(["sys", SystemVariableKey.CONVERSATION_ID])
     if not isinstance(conversation_id_variable, StringSegment):
         return None
     conversation_id = conversation_id_variable.value
 
-    with Session(db.engine, expire_on_commit=False) as session:
-        stmt = select(Conversation).where(Conversation.app_id == app_id, Conversation.id == conversation_id)
-        conversation = session.scalar(stmt)
-        if not conversation:
+    # Return appropriate memory type based on mode
+    if node_data_memory.mode == MemoryMode.NODE:
+        # Node-level memory (Chatflow only)
+        if not node_id:
             return None
+        return NodeTokenBufferMemory(
+            app_id=app_id,
+            conversation_id=conversation_id,
+            node_id=node_id,
+            tenant_id=tenant_id,
+            model_instance=model_instance,
+        )
+    else:
+        # Conversation-level memory (default)
+        with Session(db.engine, expire_on_commit=False) as session:
+            stmt = select(Conversation).where(Conversation.app_id == app_id, Conversation.id == conversation_id)
+            conversation = session.scalar(stmt)
+            if not conversation:
+                return None
+        return TokenBufferMemory(conversation=conversation, model_instance=model_instance)
 
-    memory = TokenBufferMemory(conversation=conversation, model_instance=model_instance)
-    return memory
+
+def save_node_memory(
+    memory: BaseMemory | None,
+    variable_pool: VariablePool,
+    user_query: str,
+    assistant_response: str,
+    user_files: Sequence["File"] | None = None,
+    assistant_files: Sequence["File"] | None = None,
+) -> None:
+    """
+    Save dialogue turn to node memory if applicable.
+
+    This function handles the storage logic for NodeTokenBufferMemory.
+    For TokenBufferMemory (conversation-level), no action is taken as it uses
+    the Message table which is managed elsewhere.
+
+    :param memory: Memory instance (NodeTokenBufferMemory or TokenBufferMemory)
+    :param variable_pool: Variable pool containing system variables
+    :param user_query: User's input text
+    :param assistant_response: Assistant's response text
+    :param user_files: Files attached by user (optional)
+    :param assistant_files: Files generated by assistant (optional)
+    """
+    if not isinstance(memory, NodeTokenBufferMemory):
+        return
+
+    # Get workflow_run_id as the key for this execution
+    workflow_run_id_var = variable_pool.get(["sys", SystemVariableKey.WORKFLOW_EXECUTION_ID])
+    if not isinstance(workflow_run_id_var, StringSegment):
+        return
+
+    workflow_run_id = workflow_run_id_var.value
+    if not workflow_run_id:
+        return
+
+    memory.add_messages(
+        workflow_run_id=workflow_run_id,
+        user_content=user_query,
+        user_files=list(user_files) if user_files else None,
+        assistant_content=assistant_response,
+        assistant_files=list(assistant_files) if assistant_files else None,
+    )
+    memory.flush()
 
 
 def deduct_llm_quota(tenant_id: str, model_instance: ModelInstance, usage: LLMUsage):

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

@@ -7,7 +7,7 @@ import logging
 import re
 import time
 from collections.abc import Generator, Mapping, Sequence
-from typing import TYPE_CHECKING, Any, Literal
+from typing import TYPE_CHECKING, Any, Literal, cast
 
 from sqlalchemy import select
 
@@ -16,10 +16,11 @@ from core.file import File, FileTransferMethod, FileType, file_manager
 from core.helper.code_executor import CodeExecutor, CodeLanguage
 from core.llm_generator.output_parser.errors import OutputParserError
 from core.llm_generator.output_parser.structured_output import invoke_llm_with_structured_output
-from core.memory.token_buffer_memory import TokenBufferMemory
+from core.memory.base import BaseMemory
 from core.model_manager import ModelInstance, ModelManager
 from core.model_runtime.entities import (
     ImagePromptMessageContent,
+    MultiModalPromptMessageContent,
     PromptMessage,
     PromptMessageContentType,
     TextPromptMessageContent,
@@ -51,6 +52,7 @@ from core.rag.entities.citation_metadata import RetrievalSourceMetadata
 from core.tools.signature import sign_upload_file
 from core.variables import (
     ArrayFileSegment,
+    ArrayPromptMessageSegment,
     ArraySegment,
     FileSegment,
     NoneSegment,
@@ -87,6 +89,7 @@ from .entities import (
     LLMNodeCompletionModelPromptTemplate,
     LLMNodeData,
     ModelConfig,
+    PromptMessageContext,
 )
 from .exc import (
     InvalidContextStructureError,
@@ -159,8 +162,9 @@ class LLMNode(Node[LLMNodeData]):
         variable_pool = self.graph_runtime_state.variable_pool
 
         try:
-            # init messages template
-            self.node_data.prompt_template = self._transform_chat_messages(self.node_data.prompt_template)
+            # Parse prompt template to separate static messages and context references
+            prompt_template = self.node_data.prompt_template
+            static_messages, context_refs, template_order = self._parse_prompt_template()
 
             # fetch variables and fetch values from variable pool
             inputs = self._fetch_inputs(node_data=self.node_data)
@@ -208,8 +212,10 @@ class LLMNode(Node[LLMNodeData]):
             memory = llm_utils.fetch_memory(
                 variable_pool=variable_pool,
                 app_id=self.app_id,
+                tenant_id=self.tenant_id,
                 node_data_memory=self.node_data.memory,
                 model_instance=model_instance,
+                node_id=self._node_id,
             )
 
             query: str | None = None
@@ -220,21 +226,40 @@ class LLMNode(Node[LLMNodeData]):
                 ):
                     query = query_variable.text
 
-            prompt_messages, stop = LLMNode.fetch_prompt_messages(
-                sys_query=query,
-                sys_files=files,
-                context=context,
-                memory=memory,
-                model_config=model_config,
-                prompt_template=self.node_data.prompt_template,
-                memory_config=self.node_data.memory,
-                vision_enabled=self.node_data.vision.enabled,
-                vision_detail=self.node_data.vision.configs.detail,
-                variable_pool=variable_pool,
-                jinja2_variables=self.node_data.prompt_config.jinja2_variables,
-                tenant_id=self.tenant_id,
-                context_files=context_files,
-            )
+            # Get prompt messages
+            prompt_messages: Sequence[PromptMessage]
+            stop: Sequence[str] | None
+            if isinstance(prompt_template, list) and context_refs:
+                prompt_messages, stop = self._build_prompt_messages_with_context(
+                    context_refs=context_refs,
+                    template_order=template_order,
+                    static_messages=static_messages,
+                    query=query,
+                    files=files,
+                    context=context,
+                    memory=memory,
+                    model_config=model_config,
+                    context_files=context_files,
+                )
+            else:
+                prompt_messages, stop = LLMNode.fetch_prompt_messages(
+                    sys_query=query,
+                    sys_files=files,
+                    context=context,
+                    memory=memory,
+                    model_config=model_config,
+                    prompt_template=cast(
+                        Sequence[LLMNodeChatModelMessage] | LLMNodeCompletionModelPromptTemplate,
+                        self.node_data.prompt_template,
+                    ),
+                    memory_config=self.node_data.memory,
+                    vision_enabled=self.node_data.vision.enabled,
+                    vision_detail=self.node_data.vision.configs.detail,
+                    variable_pool=variable_pool,
+                    jinja2_variables=self.node_data.prompt_config.jinja2_variables,
+                    tenant_id=self.tenant_id,
+                    context_files=context_files,
+                )
 
             # handle invoke result
             generator = LLMNode.invoke_llm(
@@ -250,6 +275,7 @@ class LLMNode(Node[LLMNodeData]):
                 node_id=self._node_id,
                 node_type=self.node_type,
                 reasoning_format=self.node_data.reasoning_format,
+                tenant_id=self.tenant_id,
             )
 
             structured_output: LLMStructuredOutput | None = None
@@ -301,12 +327,25 @@ class LLMNode(Node[LLMNodeData]):
                 "reasoning_content": reasoning_content,
                 "usage": jsonable_encoder(usage),
                 "finish_reason": finish_reason,
+                "context": self._build_context(prompt_messages, clean_text),
             }
             if structured_output:
                 outputs["structured_output"] = structured_output.structured_output
             if self._file_outputs:
                 outputs["files"] = ArrayFileSegment(value=self._file_outputs)
 
+            # Write to Node Memory if in node memory mode
+            # Resolve the query template to get actual user content
+            actual_query = variable_pool.convert_template(query or "").text
+            llm_utils.save_node_memory(
+                memory=memory,
+                variable_pool=variable_pool,
+                user_query=actual_query,
+                assistant_response=clean_text,
+                user_files=files,
+                assistant_files=self._file_outputs,
+            )
+
             # Send final chunk event to indicate streaming is complete
             yield StreamChunkEvent(
                 selector=[self._node_id, "text"],
@@ -367,6 +406,7 @@ class LLMNode(Node[LLMNodeData]):
         node_id: str,
         node_type: NodeType,
         reasoning_format: Literal["separated", "tagged"] = "tagged",
+        tenant_id: str | None = None,
     ) -> Generator[NodeEventBase | LLMStructuredOutput, None, None]:
         model_schema = model_instance.model_type_instance.get_model_schema(
             node_data_model.name, model_instance.credentials
@@ -390,6 +430,7 @@ class LLMNode(Node[LLMNodeData]):
                 stop=list(stop or []),
                 stream=True,
                 user=user_id,
+                tenant_id=tenant_id,
             )
         else:
             request_start_time = time.perf_counter()
@@ -566,6 +607,48 @@ class LLMNode(Node[LLMNodeData]):
         # Separated mode: always return clean text and reasoning_content
         return clean_text, reasoning_content or ""
 
+    @staticmethod
+    def _build_context(
+        prompt_messages: Sequence[PromptMessage],
+        assistant_response: str,
+    ) -> list[PromptMessage]:
+        """
+        Build context from prompt messages and assistant response.
+        Excludes system messages and includes the current LLM response.
+        Returns list[PromptMessage] for use with ArrayPromptMessageSegment.
+
+        Note: Multi-modal content base64 data is truncated to avoid storing large data in context.
+        """
+        context_messages: list[PromptMessage] = [
+            LLMNode._truncate_multimodal_content(m) for m in prompt_messages if m.role != PromptMessageRole.SYSTEM
+        ]
+        context_messages.append(AssistantPromptMessage(content=assistant_response))
+        return context_messages
+
+    @staticmethod
+    def _truncate_multimodal_content(message: PromptMessage) -> PromptMessage:
+        """
+        Truncate multi-modal content base64 data in a message to avoid storing large data.
+        Preserves the PromptMessage structure for ArrayPromptMessageSegment compatibility.
+        """
+        content = message.content
+        if content is None or isinstance(content, str):
+            return message
+
+        # Process list content, truncating multi-modal base64 data
+        new_content: list[PromptMessageContentUnionTypes] = []
+        for item in content:
+            if isinstance(item, MultiModalPromptMessageContent):
+                # Truncate base64_data similar to prompt_messages_to_prompt_for_saving
+                truncated_base64 = ""
+                if item.base64_data:
+                    truncated_base64 = item.base64_data[:10] + "...[TRUNCATED]..." + item.base64_data[-10:]
+                new_content.append(item.model_copy(update={"base64_data": truncated_base64}))
+            else:
+                new_content.append(item)
+
+        return message.model_copy(update={"content": new_content})
+
     def _transform_chat_messages(
         self, messages: Sequence[LLMNodeChatModelMessage] | LLMNodeCompletionModelPromptTemplate, /
     ) -> Sequence[LLMNodeChatModelMessage] | LLMNodeCompletionModelPromptTemplate:
@@ -581,6 +664,106 @@ class LLMNode(Node[LLMNodeData]):
 
         return messages
 
+    def _parse_prompt_template(
+        self,
+    ) -> tuple[list[LLMNodeChatModelMessage], list[PromptMessageContext], list[tuple[int, str]]]:
+        """
+        Parse prompt_template to separate static messages and context references.
+
+        Returns:
+            Tuple of (static_messages, context_refs, template_order)
+            - static_messages: list of LLMNodeChatModelMessage
+            - context_refs: list of PromptMessageContext
+            - template_order: list of (index, type) tuples preserving original order
+        """
+        prompt_template = self.node_data.prompt_template
+        static_messages: list[LLMNodeChatModelMessage] = []
+        context_refs: list[PromptMessageContext] = []
+        template_order: list[tuple[int, str]] = []
+
+        if isinstance(prompt_template, list):
+            for idx, item in enumerate(prompt_template):
+                if isinstance(item, PromptMessageContext):
+                    context_refs.append(item)
+                    template_order.append((idx, "context"))
+                else:
+                    static_messages.append(item)
+                    template_order.append((idx, "static"))
+            # Transform static messages for jinja2
+            if static_messages:
+                self.node_data.prompt_template = self._transform_chat_messages(static_messages)
+
+        return static_messages, context_refs, template_order
+
+    def _build_prompt_messages_with_context(
+        self,
+        *,
+        context_refs: list[PromptMessageContext],
+        template_order: list[tuple[int, str]],
+        static_messages: list[LLMNodeChatModelMessage],
+        query: str | None,
+        files: Sequence[File],
+        context: str | None,
+        memory: BaseMemory | None,
+        model_config: ModelConfigWithCredentialsEntity,
+        context_files: list[File],
+    ) -> tuple[list[PromptMessage], Sequence[str] | None]:
+        """
+        Build prompt messages by combining static messages and context references in DSL order.
+
+        Returns:
+            Tuple of (prompt_messages, stop_sequences)
+        """
+        variable_pool = self.graph_runtime_state.variable_pool
+
+        # Build a map from context index to its messages
+        context_messages_map: dict[int, list[PromptMessage]] = {}
+        context_idx = 0
+        for idx, type_ in template_order:
+            if type_ == "context":
+                ctx_ref = context_refs[context_idx]
+                ctx_var = variable_pool.get(ctx_ref.value_selector)
+                if ctx_var is None:
+                    raise VariableNotFoundError(f"Variable {'.'.join(ctx_ref.value_selector)} not found")
+                if not isinstance(ctx_var, ArrayPromptMessageSegment):
+                    raise InvalidVariableTypeError(f"Variable {'.'.join(ctx_ref.value_selector)} is not array[message]")
+                context_messages_map[idx] = list(ctx_var.value)
+                context_idx += 1
+
+        # Process static messages
+        static_prompt_messages: Sequence[PromptMessage] = []
+        stop: Sequence[str] | None = None
+        if static_messages:
+            static_prompt_messages, stop = LLMNode.fetch_prompt_messages(
+                sys_query=query,
+                sys_files=files,
+                context=context,
+                memory=memory,
+                model_config=model_config,
+                prompt_template=cast(Sequence[LLMNodeChatModelMessage], self.node_data.prompt_template),
+                memory_config=self.node_data.memory,
+                vision_enabled=self.node_data.vision.enabled,
+                vision_detail=self.node_data.vision.configs.detail,
+                variable_pool=variable_pool,
+                jinja2_variables=self.node_data.prompt_config.jinja2_variables,
+                tenant_id=self.tenant_id,
+                context_files=context_files,
+            )
+
+        # Combine messages according to original DSL order
+        combined_messages: list[PromptMessage] = []
+        static_msg_iter = iter(static_prompt_messages)
+        for idx, type_ in template_order:
+            if type_ == "context":
+                combined_messages.extend(context_messages_map[idx])
+            else:
+                if msg := next(static_msg_iter, None):
+                    combined_messages.append(msg)
+        # Append any remaining static messages (e.g., memory messages)
+        combined_messages.extend(static_msg_iter)
+
+        return combined_messages, stop
+
     def _fetch_jinja_inputs(self, node_data: LLMNodeData) -> dict[str, str]:
         variables: dict[str, Any] = {}
 
@@ -778,7 +961,7 @@ class LLMNode(Node[LLMNodeData]):
         sys_query: str | None = None,
         sys_files: Sequence[File],
         context: str | None = None,
-        memory: TokenBufferMemory | None = None,
+        memory: BaseMemory | None = None,
         model_config: ModelConfigWithCredentialsEntity,
         prompt_template: Sequence[LLMNodeChatModelMessage] | LLMNodeCompletionModelPromptTemplate,
         memory_config: MemoryConfig | None = None,
@@ -1337,7 +1520,7 @@ def _calculate_rest_token(
 
 def _handle_memory_chat_mode(
     *,
-    memory: TokenBufferMemory | None,
+    memory: BaseMemory | None,
     memory_config: MemoryConfig | None,
     model_config: ModelConfigWithCredentialsEntity,
 ) -> Sequence[PromptMessage]:
@@ -1354,7 +1537,7 @@ def _handle_memory_chat_mode(
 
 def _handle_memory_completion_mode(
     *,
-    memory: TokenBufferMemory | None,
+    memory: BaseMemory | None,
     memory_config: MemoryConfig | None,
     model_config: ModelConfigWithCredentialsEntity,
 ) -> str:

api/core/workflow/nodes/node_factory.py

@@ -1,16 +1,21 @@
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from typing import TYPE_CHECKING, final
 
 from typing_extensions import override
 
 from configs import dify_config
+from core.file import file_manager
+from core.helper import ssrf_proxy
 from core.helper.code_executor.code_executor import CodeExecutor
 from core.helper.code_executor.code_node_provider import CodeNodeProvider
+from core.tools.tool_file_manager import ToolFileManager
 from core.workflow.enums import NodeType
 from core.workflow.graph import NodeFactory
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.code.code_node import CodeNode
 from core.workflow.nodes.code.limits import CodeNodeLimits
+from core.workflow.nodes.http_request.node import HttpRequestNode
+from core.workflow.nodes.protocols import FileManagerProtocol, HttpClientProtocol
 from core.workflow.nodes.template_transform.template_renderer import (
     CodeExecutorJinja2TemplateRenderer,
     Jinja2TemplateRenderer,
@@ -43,6 +48,9 @@ class DifyNodeFactory(NodeFactory):
         code_providers: Sequence[type[CodeNodeProvider]] | None = None,
         code_limits: CodeNodeLimits | None = None,
         template_renderer: Jinja2TemplateRenderer | None = None,
+        http_request_http_client: HttpClientProtocol = ssrf_proxy,
+        http_request_tool_file_manager_factory: Callable[[], ToolFileManager] = ToolFileManager,
+        http_request_file_manager: FileManagerProtocol = file_manager,
     ) -> None:
         self.graph_init_params = graph_init_params
         self.graph_runtime_state = graph_runtime_state
@@ -61,6 +69,9 @@ class DifyNodeFactory(NodeFactory):
             max_object_array_length=dify_config.CODE_MAX_OBJECT_ARRAY_LENGTH,
         )
         self._template_renderer = template_renderer or CodeExecutorJinja2TemplateRenderer()
+        self._http_request_http_client = http_request_http_client
+        self._http_request_tool_file_manager_factory = http_request_tool_file_manager_factory
+        self._http_request_file_manager = http_request_file_manager
 
     @override
     def create_node(self, node_config: dict[str, object]) -> Node:
@@ -113,6 +124,7 @@ class DifyNodeFactory(NodeFactory):
                 code_providers=self._code_providers,
                 code_limits=self._code_limits,
             )
+
         if node_type == NodeType.TEMPLATE_TRANSFORM:
             return TemplateTransformNode(
                 id=node_id,
@@ -122,6 +134,17 @@ class DifyNodeFactory(NodeFactory):
                 template_renderer=self._template_renderer,
             )
 
+        if node_type == NodeType.HTTP_REQUEST:
+            return HttpRequestNode(
+                id=node_id,
+                config=node_config,
+                graph_init_params=self.graph_init_params,
+                graph_runtime_state=self.graph_runtime_state,
+                http_client=self._http_request_http_client,
+                tool_file_manager_factory=self._http_request_tool_file_manager_factory,
+                file_manager=self._http_request_file_manager,
+            )
+
         return node_class(
             id=node_id,
             config=node_config,

api/core/workflow/nodes/parameter_extractor/parameter_extractor_node.py

@@ -7,7 +7,7 @@ from typing import Any, cast
 
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.file import File
-from core.memory.token_buffer_memory import TokenBufferMemory
+from core.memory.base import BaseMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities import ImagePromptMessageContent
 from core.model_runtime.entities.llm_entities import LLMUsage
@@ -145,8 +145,10 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         memory = llm_utils.fetch_memory(
             variable_pool=variable_pool,
             app_id=self.app_id,
+            tenant_id=self.tenant_id,
             node_data_memory=node_data.memory,
             model_instance=model_instance,
+            node_id=self._node_id,
         )
 
         if (
@@ -244,6 +246,14 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         # transform result into standard format
         result = self._transform_result(data=node_data, result=result or {})
 
+        # Save to node memory if in node memory mode
+        llm_utils.save_node_memory(
+            memory=memory,
+            variable_pool=variable_pool,
+            user_query=query,
+            assistant_response=json.dumps(result, ensure_ascii=False),
+        )
+
         return NodeRunResult(
             status=WorkflowNodeExecutionStatus.SUCCEEDED,
             inputs=inputs,
@@ -299,7 +309,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         query: str,
         variable_pool: VariablePool,
         model_config: ModelConfigWithCredentialsEntity,
-        memory: TokenBufferMemory | None,
+        memory: BaseMemory | None,
         files: Sequence[File],
         vision_detail: ImagePromptMessageContent.DETAIL | None = None,
     ) -> tuple[list[PromptMessage], list[PromptMessageTool]]:
@@ -381,7 +391,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         query: str,
         variable_pool: VariablePool,
         model_config: ModelConfigWithCredentialsEntity,
-        memory: TokenBufferMemory | None,
+        memory: BaseMemory | None,
         files: Sequence[File],
         vision_detail: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
@@ -419,7 +429,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         query: str,
         variable_pool: VariablePool,
         model_config: ModelConfigWithCredentialsEntity,
-        memory: TokenBufferMemory | None,
+        memory: BaseMemory | None,
         files: Sequence[File],
         vision_detail: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
@@ -453,7 +463,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         query: str,
         variable_pool: VariablePool,
         model_config: ModelConfigWithCredentialsEntity,
-        memory: TokenBufferMemory | None,
+        memory: BaseMemory | None,
         files: Sequence[File],
         vision_detail: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
@@ -681,7 +691,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         node_data: ParameterExtractorNodeData,
         query: str,
         variable_pool: VariablePool,
-        memory: TokenBufferMemory | None,
+        memory: BaseMemory | None,
         max_token_limit: int = 2000,
     ) -> list[ChatModelMessage]:
         model_mode = ModelMode(node_data.model.mode)
@@ -708,7 +718,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         node_data: ParameterExtractorNodeData,
         query: str,
         variable_pool: VariablePool,
-        memory: TokenBufferMemory | None,
+        memory: BaseMemory | None,
         max_token_limit: int = 2000,
     ):
         model_mode = ModelMode(node_data.model.mode)

api/core/workflow/nodes/protocols.py

@@ -0,0 +1,29 @@
+from typing import Protocol
+
+import httpx
+
+from core.file import File
+
+
+class HttpClientProtocol(Protocol):
+    @property
+    def max_retries_exceeded_error(self) -> type[Exception]: ...
+
+    @property
+    def request_error(self) -> type[Exception]: ...
+
+    def get(self, url: str, max_retries: int = ..., **kwargs: object) -> httpx.Response: ...
+
+    def head(self, url: str, max_retries: int = ..., **kwargs: object) -> httpx.Response: ...
+
+    def post(self, url: str, max_retries: int = ..., **kwargs: object) -> httpx.Response: ...
+
+    def put(self, url: str, max_retries: int = ..., **kwargs: object) -> httpx.Response: ...
+
+    def delete(self, url: str, max_retries: int = ..., **kwargs: object) -> httpx.Response: ...
+
+    def patch(self, url: str, max_retries: int = ..., **kwargs: object) -> httpx.Response: ...
+
+
+class FileManagerProtocol(Protocol):
+    def download(self, f: File, /) -> bytes: ...

api/core/workflow/nodes/question_classifier/question_classifier_node.py

@@ -4,7 +4,7 @@ from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Any
 
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.memory.token_buffer_memory import TokenBufferMemory
+from core.memory.base import BaseMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities import LLMUsage, ModelPropertyKey, PromptMessageRole
 from core.model_runtime.utils.encoders import jsonable_encoder
@@ -96,8 +96,10 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         memory = llm_utils.fetch_memory(
             variable_pool=variable_pool,
             app_id=self.app_id,
+            tenant_id=self.tenant_id,
             node_data_memory=node_data.memory,
             model_instance=model_instance,
+            node_id=self._node_id,
         )
         # fetch instruction
         node_data.instruction = node_data.instruction or ""
@@ -203,6 +205,14 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
                 "usage": jsonable_encoder(usage),
             }
 
+            # Save to node memory if in node memory mode
+            llm_utils.save_node_memory(
+                memory=memory,
+                variable_pool=variable_pool,
+                user_query=query or "",
+                assistant_response=f"class_name: {category_name}, class_id: {category_id}",
+            )
+
             return NodeRunResult(
                 status=WorkflowNodeExecutionStatus.SUCCEEDED,
                 inputs=variables,
@@ -312,7 +322,7 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         self,
         node_data: QuestionClassifierNodeData,
         query: str,
-        memory: TokenBufferMemory | None,
+        memory: BaseMemory | None,
         max_token_limit: int = 2000,
     ):
         model_mode = ModelMode(node_data.model.mode)

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

@@ -1,11 +1,63 @@
-from typing import Any, Literal, Union
+import re
+from collections.abc import Sequence
+from typing import Any, Literal, Self, Union
 
-from pydantic import BaseModel, field_validator
+from pydantic import BaseModel, field_validator, model_validator
 from pydantic_core.core_schema import ValidationInfo
 
 from core.tools.entities.tool_entities import ToolProviderType
 from core.workflow.nodes.base.entities import BaseNodeData
 
+# Pattern to match mention value format: {{@node.context@}}instruction
+# The placeholder {{@node.context@}} must appear at the beginning
+# Format: {{@agent_node_id.context@}} where agent_node_id is dynamic, context is fixed
+MENTION_VALUE_PATTERN = re.compile(r"^\{\{@([a-zA-Z0-9_]+)\.context@\}\}(.*)$", re.DOTALL)
+
+
+def parse_mention_value(value: str) -> tuple[str, str]:
+    """Parse mention value into (node_id, instruction).
+
+    Args:
+        value: The mention value string like "{{@llm.context@}}extract keywords"
+
+    Returns:
+        Tuple of (node_id, instruction)
+
+    Raises:
+        ValueError: If value format is invalid
+    """
+    match = MENTION_VALUE_PATTERN.match(value)
+    if not match:
+        raise ValueError(
+            "For mention type, value must start with {{@node.context@}} placeholder, "
+            "e.g., '{{@llm.context@}}extract keywords'"
+        )
+    return match.group(1), match.group(2)
+
+
+class MentionConfig(BaseModel):
+    """Configuration for extracting value from context variable.
+
+    Used when a tool parameter needs to be extracted from list[PromptMessage]
+    context using an extractor LLM node.
+
+    Note: instruction is embedded in the value field as "{{@node.context@}}instruction"
+    """
+
+    # ID of the extractor LLM node
+    extractor_node_id: str
+
+    # Output variable selector from extractor node
+    # e.g., ["text"], ["structured_output", "query"]
+    output_selector: Sequence[str]
+
+    # Strategy when output is None
+    null_strategy: Literal["raise_error", "use_default"] = "raise_error"
+
+    # Default value when null_strategy is "use_default"
+    # Type should match the parameter's expected type
+    default_value: Any = None
+
 
 class ToolEntity(BaseModel):
     provider_id: str
@@ -35,7 +87,9 @@ class ToolNodeData(BaseNodeData, ToolEntity):
     class ToolInput(BaseModel):
         # TODO: check this type
         value: Union[Any, list[str]]
-        type: Literal["mixed", "variable", "constant"]
+        type: Literal["mixed", "variable", "constant", "mention"]
+        # Required config for mention type, extracting value from context variable
+        mention_config: MentionConfig | None = None
 
         @field_validator("type", mode="before")
         @classmethod
@@ -48,6 +102,9 @@ class ToolNodeData(BaseNodeData, ToolEntity):
 
             if typ == "mixed" and not isinstance(value, str):
                 raise ValueError("value must be a string")
+            elif typ == "mention":
+                # Skip here, will be validated in model_validator
+                pass
             elif typ == "variable":
                 if not isinstance(value, list):
                     raise ValueError("value must be a list")
@@ -58,6 +115,26 @@ class ToolNodeData(BaseNodeData, ToolEntity):
                 raise ValueError("value must be a string, int, float, bool or dict")
             return typ
 
+        @model_validator(mode="after")
+        def check_mention_type(self) -> Self:
+            """Validate mention type with mention_config."""
+            if self.type != "mention":
+                return self
+
+            value = self.value
+            if value is None:
+                return self
+
+            if not isinstance(value, str):
+                raise ValueError("value must be a string for mention type")
+            # For mention type, value must match format: {{@node.context@}}instruction
+            # This will raise ValueError if format is invalid
+            parse_mention_value(value)
+            # mention_config is required for mention type
+            if self.mention_config is None:
+                raise ValueError("mention_config is required for mention type")
+            return self
+
     tool_parameters: dict[str, ToolInput]
     # The version of the tool parameter.
     # If this value is None, it indicates this is a previous version

api/core/workflow/nodes/tool/tool_node.py

@@ -1,7 +1,10 @@
+import logging
 from collections.abc import Generator, Mapping, Sequence
 from typing import TYPE_CHECKING, Any
 
 from sqlalchemy import select
+
+logger = logging.getLogger(__name__)
 from sqlalchemy.orm import Session
 
 from core.callback_handler.workflow_tool_callback_handler import DifyWorkflowCallbackHandler
@@ -184,6 +187,7 @@ class ToolNode(Node[ToolNodeData]):
             tool_parameters (Sequence[ToolParameter]): The list of tool parameters.
             variable_pool (VariablePool): The variable pool containing the variables.
             node_data (ToolNodeData): The data associated with the tool node.
+            for_log (bool): Whether to generate parameters for logging.
 
         Returns:
             Mapping[str, Any]: A dictionary containing the generated parameters.
@@ -199,14 +203,37 @@ class ToolNode(Node[ToolNodeData]):
                 continue
             tool_input = node_data.tool_parameters[parameter_name]
             if tool_input.type == "variable":
-                variable = variable_pool.get(tool_input.value)
+                if not isinstance(tool_input.value, list):
+                    raise ToolParameterError(f"Invalid variable selector for parameter '{parameter_name}'")
+                selector = tool_input.value
+                variable = variable_pool.get(selector)
                 if variable is None:
                     if parameter.required:
-                        raise ToolParameterError(f"Variable {tool_input.value} does not exist")
+                        raise ToolParameterError(f"Variable {selector} does not exist")
                     continue
                 parameter_value = variable.value
+            elif tool_input.type == "mention":
+                # Mention type: get value from extractor node's output
+                if tool_input.mention_config is None:
+                    raise ToolParameterError(
+                        f"mention_config is required for mention type parameter '{parameter_name}'"
+                    )
+                mention_config = tool_input.mention_config.model_dump()
+                try:
+                    parameter_value, found = variable_pool.resolve_mention(
+                        mention_config, parameter_name=parameter_name
+                    )
+                    if not found and parameter.required:
+                        raise ToolParameterError(
+                            f"Extractor output not found for required parameter '{parameter_name}'"
+                        )
+                    if not found:
+                        continue
+                except ValueError as e:
+                    raise ToolParameterError(str(e)) from e
             elif tool_input.type in {"mixed", "constant"}:
-                segment_group = variable_pool.convert_template(str(tool_input.value))
+                template = str(tool_input.value)
+                segment_group = variable_pool.convert_template(template)
                 parameter_value = segment_group.log if for_log else segment_group.text
             else:
                 raise ToolParameterError(f"Unknown tool input type '{tool_input.type}'")
@@ -488,8 +515,12 @@ class ToolNode(Node[ToolNodeData]):
                 for selector in selectors:
                     result[selector.variable] = selector.value_selector
             elif input.type == "variable":
-                selector_key = ".".join(input.value)
-                result[f"#{selector_key}#"] = input.value
+                if isinstance(input.value, list):
+                    selector_key = ".".join(input.value)
+                    result[f"#{selector_key}#"] = input.value
+            elif input.type == "mention":
+                # Mention type: value is handled by extractor node, no direct variable reference
+                pass
             elif input.type == "constant":
                 pass
 

api/core/workflow/nodes/variable_assigner/v1/node.py

@@ -1,7 +1,7 @@
 from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Any
 
-from core.variables import SegmentType, Variable
+from core.variables import SegmentType, VariableBase
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID
 from core.workflow.entities import GraphInitParams
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
@@ -73,7 +73,7 @@ class VariableAssignerNode(Node[VariableAssignerData]):
         assigned_variable_selector = self.node_data.assigned_variable_selector
         # Should be String, Number, Object, ArrayString, ArrayNumber, ArrayObject
         original_variable = self.graph_runtime_state.variable_pool.get(assigned_variable_selector)
-        if not isinstance(original_variable, Variable):
+        if not isinstance(original_variable, VariableBase):
             raise VariableOperatorNodeError("assigned variable not found")
 
         match self.node_data.write_mode:

api/core/workflow/nodes/variable_assigner/v2/node.py

@@ -2,7 +2,7 @@ import json
 from collections.abc import Mapping, MutableMapping, Sequence
 from typing import TYPE_CHECKING, Any
 
-from core.variables import SegmentType, Variable
+from core.variables import SegmentType, VariableBase
 from core.variables.consts import SELECTORS_LENGTH
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
@@ -118,7 +118,7 @@ class VariableAssignerNode(Node[VariableAssignerNodeData]):
                 # ==================== Validation Part
 
                 # Check if variable exists
-                if not isinstance(variable, Variable):
+                if not isinstance(variable, VariableBase):
                     raise VariableNotFoundError(variable_selector=item.variable_selector)
 
                 # Check if operation is supported
@@ -192,7 +192,7 @@ class VariableAssignerNode(Node[VariableAssignerNodeData]):
 
         for selector in updated_variable_selectors:
             variable = self.graph_runtime_state.variable_pool.get(selector)
-            if not isinstance(variable, Variable):
+            if not isinstance(variable, VariableBase):
                 raise VariableNotFoundError(variable_selector=selector)
             process_data[variable.name] = variable.value
 
@@ -213,7 +213,7 @@ class VariableAssignerNode(Node[VariableAssignerNodeData]):
     def _handle_item(
         self,
         *,
-        variable: Variable,
+        variable: VariableBase,
         operation: Operation,
         value: Any,
     ):

api/core/workflow/runtime/variable_pool.py

@@ -9,10 +9,10 @@ from typing import Annotated, Any, Union, cast
 from pydantic import BaseModel, Field
 
 from core.file import File, FileAttribute, file_manager
-from core.variables import Segment, SegmentGroup, Variable
+from core.variables import Segment, SegmentGroup, VariableBase
 from core.variables.consts import SELECTORS_LENGTH
 from core.variables.segments import FileSegment, ObjectSegment
-from core.variables.variables import RAGPipelineVariableInput, VariableUnion
+from core.variables.variables import RAGPipelineVariableInput, Variable
 from core.workflow.constants import (
     CONVERSATION_VARIABLE_NODE_ID,
     ENVIRONMENT_VARIABLE_NODE_ID,
@@ -32,7 +32,7 @@ class VariablePool(BaseModel):
     # The first element of the selector is the node id, it's the first-level key in the dictionary.
     # Other elements of the selector are the keys in the second-level dictionary. To get the key, we hash the
     # elements of the selector except the first one.
-    variable_dictionary: defaultdict[str, Annotated[dict[str, VariableUnion], Field(default_factory=dict)]] = Field(
+    variable_dictionary: defaultdict[str, Annotated[dict[str, Variable], Field(default_factory=dict)]] = Field(
         description="Variables mapping",
         default=defaultdict(dict),
     )
@@ -46,13 +46,13 @@ class VariablePool(BaseModel):
         description="System variables",
         default_factory=SystemVariable.empty,
     )
-    environment_variables: Sequence[VariableUnion] = Field(
+    environment_variables: Sequence[Variable] = Field(
         description="Environment variables.",
-        default_factory=list[VariableUnion],
+        default_factory=list[Variable],
     )
-    conversation_variables: Sequence[VariableUnion] = Field(
+    conversation_variables: Sequence[Variable] = Field(
         description="Conversation variables.",
-        default_factory=list[VariableUnion],
+        default_factory=list[Variable],
     )
     rag_pipeline_variables: list[RAGPipelineVariableInput] = Field(
         description="RAG pipeline variables.",
@@ -105,7 +105,7 @@ class VariablePool(BaseModel):
                 f"got {len(selector)} elements"
             )
 
-        if isinstance(value, Variable):
+        if isinstance(value, VariableBase):
             variable = value
         elif isinstance(value, Segment):
             variable = variable_factory.segment_to_variable(segment=value, selector=selector)
@@ -114,9 +114,9 @@ class VariablePool(BaseModel):
             variable = variable_factory.segment_to_variable(segment=segment, selector=selector)
 
         node_id, name = self._selector_to_keys(selector)
-        # Based on the definition of `VariableUnion`,
-        # `list[Variable]` can be safely used as `list[VariableUnion]` since they are compatible.
-        self.variable_dictionary[node_id][name] = cast(VariableUnion, variable)
+        # Based on the definition of `Variable`,
+        # `VariableBase` instances can be safely used as `Variable` since they are compatible.
+        self.variable_dictionary[node_id][name] = cast(Variable, variable)
 
     @classmethod
     def _selector_to_keys(cls, selector: Sequence[str]) -> tuple[str, str]:
@@ -268,6 +268,58 @@ class VariablePool(BaseModel):
                 continue
             self.add(selector, value)
 
+    def resolve_mention(
+        self,
+        mention_config: Mapping[str, Any],
+        /,
+        *,
+        parameter_name: str = "",
+    ) -> tuple[Any, bool]:
+        """
+        Resolve a mention parameter value from an extractor node's output.
+
+        Mention parameters reference values extracted by an extractor LLM node
+        from list[PromptMessage] context.
+
+        Args:
+            mention_config: A dict containing:
+                - extractor_node_id: ID of the extractor LLM node
+                - output_selector: Selector path for the output variable (e.g., ["text"])
+                - null_strategy: "raise_error" or "use_default"
+                - default_value: Value to use when null_strategy is "use_default"
+            parameter_name: Name of the parameter being resolved (for error messages)
+
+        Returns:
+            Tuple of (resolved_value, found):
+                - resolved_value: The extracted value, or default_value if not found
+                - found: True if value was found, False if using default
+
+        Raises:
+            ValueError: If extractor_node_id is missing, or if null_strategy is
+                       "raise_error" and the value is not found
+        """
+        extractor_node_id = mention_config.get("extractor_node_id")
+        if not extractor_node_id:
+            raise ValueError(f"Missing extractor_node_id for mention parameter '{parameter_name}'")
+
+        output_selector = list(mention_config.get("output_selector", []))
+        null_strategy = mention_config.get("null_strategy", "raise_error")
+        default_value = mention_config.get("default_value")
+
+        # Build full selector: [extractor_node_id, ...output_selector]
+        full_selector = [extractor_node_id] + output_selector
+        variable = self.get(full_selector)
+
+        if variable is None:
+            if null_strategy == "use_default":
+                return default_value, False
+            raise ValueError(
+                f"Extractor node '{extractor_node_id}' output '{'.'.join(output_selector)}' "
+                f"not found for parameter '{parameter_name}'"
+            )
+
+        return variable.value, True
+
     @classmethod
     def empty(cls) -> VariablePool:
         """Create an empty variable pool."""

api/core/workflow/variable_loader.py

@@ -2,7 +2,7 @@ import abc
 from collections.abc import Mapping, Sequence
 from typing import Any, Protocol
 
-from core.variables import Variable
+from core.variables import VariableBase
 from core.variables.consts import SELECTORS_LENGTH
 from core.workflow.runtime import VariablePool
 
@@ -26,7 +26,7 @@ class VariableLoader(Protocol):
     """
 
     @abc.abstractmethod
-    def load_variables(self, selectors: list[list[str]]) -> list[Variable]:
+    def load_variables(self, selectors: list[list[str]]) -> list[VariableBase]:
         """Load variables based on the provided selectors. If the selectors are empty,
         this method should return an empty list.
 
@@ -36,7 +36,7 @@ class VariableLoader(Protocol):
         :param: selectors: a list of string list, each inner list should have at least two elements:
             - the first element is the node ID,
             - the second element is the variable name.
-        :return: a list of Variable objects that match the provided selectors.
+        :return: a list of VariableBase objects that match the provided selectors.
         """
         pass
 
@@ -46,7 +46,7 @@ class _DummyVariableLoader(VariableLoader):
     Serves as a placeholder when no variable loading is needed.
     """
 
-    def load_variables(self, selectors: list[list[str]]) -> list[Variable]:
+    def load_variables(self, selectors: list[list[str]]) -> list[VariableBase]:
         return []
 
 

api/core/workflow/workflow_entry.py

@@ -189,8 +189,7 @@ class WorkflowEntry:
             )
 
         try:
-            # run node
-            generator = node.run()
+            generator = cls._traced_node_run(node)
         except Exception as e:
             logger.exception(
                 "error while running node, workflow_id=%s, node_id=%s, node_type=%s, node_version=%s",
@@ -323,8 +322,7 @@ class WorkflowEntry:
                 tenant_id=tenant_id,
             )
 
-            # run node
-            generator = node.run()
+            generator = cls._traced_node_run(node)
 
             return node, generator
         except Exception as e:
@@ -430,3 +428,26 @@ class WorkflowEntry:
                         input_value = current_variable.value | input_value
 
                 variable_pool.add([variable_node_id] + variable_key_list, input_value)
+
+    @staticmethod
+    def _traced_node_run(node: Node) -> Generator[GraphNodeEventBase, None, None]:
+        """
+        Wraps a node's run method with OpenTelemetry tracing and returns a generator.
+        """
+        # Wrap node.run() with ObservabilityLayer hooks to produce node-level spans
+        layer = ObservabilityLayer()
+        layer.on_graph_start()
+        node.ensure_execution_id()
+
+        def _gen():
+            error: Exception | None = None
+            layer.on_node_run_start(node)
+            try:
+                yield from node.run()
+            except Exception as exc:
+                error = exc
+                raise
+            finally:
+                layer.on_node_run_end(node, error)
+
+        return _gen()

api/events/event_handlers/__init__.py

@@ -6,6 +6,7 @@ from .create_site_record_when_app_created import handle as handle_create_site_re
 from .delete_tool_parameters_cache_when_sync_draft_workflow import (
     handle as handle_delete_tool_parameters_cache_when_sync_draft_workflow,
 )
+from .queue_credential_sync_when_tenant_created import handle as handle_queue_credential_sync_when_tenant_created
 from .sync_plugin_trigger_when_app_created import handle as handle_sync_plugin_trigger_when_app_created
 from .sync_webhook_when_app_created import handle as handle_sync_webhook_when_app_created
 from .sync_workflow_schedule_when_app_published import handle as handle_sync_workflow_schedule_when_app_published
@@ -30,6 +31,7 @@ __all__ = [
     "handle_create_installed_app_when_app_created",
     "handle_create_site_record_when_app_created",
     "handle_delete_tool_parameters_cache_when_sync_draft_workflow",
+    "handle_queue_credential_sync_when_tenant_created",
     "handle_sync_plugin_trigger_when_app_created",
     "handle_sync_webhook_when_app_created",
     "handle_sync_workflow_schedule_when_app_published",

api/events/event_handlers/queue_credential_sync_when_tenant_created.py

@@ -0,0 +1,19 @@
+from configs import dify_config
+from events.tenant_event import tenant_was_created
+from services.enterprise.workspace_sync import WorkspaceSyncService
+
+
+@tenant_was_created.connect
+def handle(sender, **kwargs):
+    """Queue credential sync when a tenant/workspace is created."""
+    # Only queue sync tasks if plugin manager (enterprise feature) is enabled
+    if not dify_config.ENTERPRISE_ENABLED:
+        return
+
+    tenant = sender
+
+    # Determine source from kwargs if available, otherwise use generic
+    source = kwargs.get("source", "tenant_created")
+
+    # Queue credential sync task to Redis for enterprise backend to process
+    WorkspaceSyncService.queue_credential_sync(tenant.id, source=source)

api/extensions/ext_commands.py

@@ -4,6 +4,7 @@ from dify_app import DifyApp
 def init_app(app: DifyApp):
     from commands import (
         add_qdrant_index,
+        clean_expired_messages,
         clean_workflow_runs,
         cleanup_orphaned_draft_variables,
         clear_free_plan_tenant_expired_logs,
@@ -58,6 +59,7 @@ def init_app(app: DifyApp):
         transform_datasource_credentials,
         install_rag_pipeline_plugins,
         clean_workflow_runs,
+        clean_expired_messages,
     ]
     for cmd in cmds_to_register:
         app.cli.add_command(cmd)

api/extensions/ext_logstore.py

@@ -10,6 +10,7 @@ import os
 
 from dotenv import load_dotenv
 
+from configs import dify_config
 from dify_app import DifyApp
 
 logger = logging.getLogger(__name__)
@@ -19,12 +20,17 @@ def is_enabled() -> bool:
     """
     Check if logstore extension is enabled.
 
+    Logstore is considered enabled when:
+    1. All required Aliyun SLS environment variables are set
+    2. At least one repository configuration points to a logstore implementation
+
     Returns:
-        True if all required Aliyun SLS environment variables are set, False otherwise
+        True if logstore should be initialized, False otherwise
     """
     # Load environment variables from .env file
     load_dotenv()
 
+    # Check if Aliyun SLS connection parameters are configured
     required_vars = [
         "ALIYUN_SLS_ACCESS_KEY_ID",
         "ALIYUN_SLS_ACCESS_KEY_SECRET",
@@ -33,24 +39,32 @@ def is_enabled() -> bool:
         "ALIYUN_SLS_PROJECT_NAME",
     ]
 
-    all_set = all(os.environ.get(var) for var in required_vars)
+    sls_vars_set = all(os.environ.get(var) for var in required_vars)
 
-    if not all_set:
-        logger.info("Logstore extension disabled: required Aliyun SLS environment variables not set")
+    if not sls_vars_set:
+        return False
 
-    return all_set
+    # Check if any repository configuration points to logstore implementation
+    repository_configs = [
+        dify_config.CORE_WORKFLOW_EXECUTION_REPOSITORY,
+        dify_config.CORE_WORKFLOW_NODE_EXECUTION_REPOSITORY,
+        dify_config.API_WORKFLOW_NODE_EXECUTION_REPOSITORY,
+        dify_config.API_WORKFLOW_RUN_REPOSITORY,
+    ]
+
+    uses_logstore = any("logstore" in config.lower() for config in repository_configs)
+
+    if not uses_logstore:
+        return False
+
+    logger.info("Logstore extension enabled: SLS variables set and repository configured to use logstore")
+    return True
 
 
 def init_app(app: DifyApp):
     """
     Initialize logstore on application startup.
-
-    This function:
-    1. Creates Aliyun SLS project if it doesn't exist
-    2. Creates logstores (workflow_execution, workflow_node_execution) if they don't exist
-    3. Creates indexes with field configurations based on PostgreSQL table structures
-
-    This operation is idempotent and only executes once during application startup.
+    If initialization fails, the application continues running without logstore features.
 
     Args:
         app: The Dify application instance
@@ -58,17 +72,23 @@ def init_app(app: DifyApp):
     try:
         from extensions.logstore.aliyun_logstore import AliyunLogStore
 
-        logger.info("Initializing logstore...")
+        logger.info("Initializing Aliyun SLS Logstore...")
 
-        # Create logstore client and initialize project/logstores/indexes
+        # Create logstore client and initialize resources
         logstore_client = AliyunLogStore()
         logstore_client.init_project_logstore()
 
-        # Attach to app for potential later use
         app.extensions["logstore"] = logstore_client
 
         logger.info("Logstore initialized successfully")
+
     except Exception:
-        logger.exception("Failed to initialize logstore")
-        # Don't raise - allow application to continue even if logstore init fails
-        # This ensures that the application can still run if logstore is misconfigured
+        logger.exception(
+            "Logstore initialization failed. Configuration: endpoint=%s, region=%s, project=%s, timeout=%ss. "
+            "Application will continue but logstore features will NOT work.",
+            os.environ.get("ALIYUN_SLS_ENDPOINT"),
+            os.environ.get("ALIYUN_SLS_REGION"),
+            os.environ.get("ALIYUN_SLS_PROJECT_NAME"),
+            os.environ.get("ALIYUN_SLS_CHECK_CONNECTIVITY_TIMEOUT", "30"),
+        )
+        # Don't raise - allow application to continue even if logstore setup fails

api/extensions/logstore/aliyun_logstore.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import logging
 import os
+import socket
 import threading
 import time
 from collections.abc import Sequence
@@ -179,9 +180,18 @@ class AliyunLogStore:
         self.region: str = os.environ.get("ALIYUN_SLS_REGION", "")
         self.project_name: str = os.environ.get("ALIYUN_SLS_PROJECT_NAME", "")
         self.logstore_ttl: int = int(os.environ.get("ALIYUN_SLS_LOGSTORE_TTL", 365))
-        self.log_enabled: bool = os.environ.get("SQLALCHEMY_ECHO", "false").lower() == "true"
+        self.log_enabled: bool = (
+            os.environ.get("SQLALCHEMY_ECHO", "false").lower() == "true"
+            or os.environ.get("LOGSTORE_SQL_ECHO", "false").lower() == "true"
+        )
         self.pg_mode_enabled: bool = os.environ.get("LOGSTORE_PG_MODE_ENABLED", "true").lower() == "true"
 
+        # Get timeout configuration
+        check_timeout = int(os.environ.get("ALIYUN_SLS_CHECK_CONNECTIVITY_TIMEOUT", 30))
+
+        # Pre-check endpoint connectivity to prevent indefinite hangs
+        self._check_endpoint_connectivity(self.endpoint, check_timeout)
+
         # Initialize SDK client
         self.client = LogClient(
             self.endpoint, self.access_key_id, self.access_key_secret, auth_version=AUTH_VERSION_4, region=self.region
@@ -199,6 +209,49 @@ class AliyunLogStore:
 
         self.__class__._initialized = True
 
+    @staticmethod
+    def _check_endpoint_connectivity(endpoint: str, timeout: int) -> None:
+        """
+        Check if the SLS endpoint is reachable before creating LogClient.
+        Prevents indefinite hangs when the endpoint is unreachable.
+
+        Args:
+            endpoint: SLS endpoint URL
+            timeout: Connection timeout in seconds
+
+        Raises:
+            ConnectionError: If endpoint is not reachable
+        """
+        # Parse endpoint URL to extract hostname and port
+        from urllib.parse import urlparse
+
+        parsed_url = urlparse(endpoint if "://" in endpoint else f"http://{endpoint}")
+        hostname = parsed_url.hostname
+        port = parsed_url.port or (443 if parsed_url.scheme == "https" else 80)
+
+        if not hostname:
+            raise ConnectionError(f"Invalid endpoint URL: {endpoint}")
+
+        sock = None
+        try:
+            # Create socket and set timeout
+            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            sock.settimeout(timeout)
+            sock.connect((hostname, port))
+        except Exception as e:
+            # Catch all exceptions and provide clear error message
+            error_type = type(e).__name__
+            raise ConnectionError(
+                f"Cannot connect to {hostname}:{port} (timeout={timeout}s): [{error_type}] {e}"
+            ) from e
+        finally:
+            # Ensure socket is properly closed
+            if sock:
+                try:
+                    sock.close()
+                except Exception:  # noqa: S110
+                    pass  # Ignore errors during cleanup
+
     @property
     def supports_pg_protocol(self) -> bool:
         """Check if PG protocol is supported and enabled."""
@@ -220,19 +273,16 @@ class AliyunLogStore:
         try:
             self._use_pg_protocol = self._pg_client.init_connection()
             if self._use_pg_protocol:
-                logger.info("Successfully connected to project %s using PG protocol", self.project_name)
+                logger.info("Using PG protocol for project %s", self.project_name)
                 # Check if scan_index is enabled for all logstores
                 self._check_and_disable_pg_if_scan_index_disabled()
                 return True
             else:
-                logger.info("PG connection failed for project %s. Will use SDK mode.", self.project_name)
+                logger.info("Using SDK mode for project %s", self.project_name)
                 return False
         except Exception as e:
-            logger.warning(
-                "Failed to establish PG connection for project %s: %s. Will use SDK mode.",
-                self.project_name,
-                str(e),
-            )
+            logger.info("Using SDK mode for project %s", self.project_name)
+            logger.debug("PG connection details: %s", str(e))
             self._use_pg_protocol = False
             return False
 
@@ -246,10 +296,6 @@ class AliyunLogStore:
         if self._use_pg_protocol:
             return
 
-        logger.info(
-            "Attempting delayed PG connection for newly created project %s ...",
-            self.project_name,
-        )
         self._attempt_pg_connection_init()
         self.__class__._pg_connection_timer = None
 
@@ -284,11 +330,7 @@ class AliyunLogStore:
         if project_is_new:
             # For newly created projects, schedule delayed PG connection
             self._use_pg_protocol = False
-            logger.info(
-                "Project %s is newly created. Will use SDK mode and schedule PG connection attempt in %d seconds.",
-                self.project_name,
-                self.__class__._pg_connection_delay,
-            )
+            logger.info("Using SDK mode for project %s (newly created)", self.project_name)
             if self.__class__._pg_connection_timer is not None:
                 self.__class__._pg_connection_timer.cancel()
             self.__class__._pg_connection_timer = threading.Timer(
@@ -299,7 +341,6 @@ class AliyunLogStore:
             self.__class__._pg_connection_timer.start()
         else:
             # For existing projects, attempt PG connection immediately
-            logger.info("Project %s already exists. Attempting PG connection...", self.project_name)
             self._attempt_pg_connection_init()
 
     def _check_and_disable_pg_if_scan_index_disabled(self) -> None:
@@ -318,9 +359,9 @@ class AliyunLogStore:
             existing_config = self.get_existing_index_config(logstore_name)
             if existing_config and not existing_config.scan_index:
                 logger.info(
-                    "Logstore %s has scan_index=false, USE SDK mode for read/write operations. "
-                    "PG protocol requires scan_index to be enabled.",
+                    "Logstore %s requires scan_index enabled, using SDK mode for project %s",
                     logstore_name,
+                    self.project_name,
                 )
                 self._use_pg_protocol = False
                 # Close PG connection if it was initialized
@@ -748,7 +789,6 @@ class AliyunLogStore:
             reverse=reverse,
         )
 
-        # Log query info if SQLALCHEMY_ECHO is enabled
         if self.log_enabled:
             logger.info(
                 "[LogStore] GET_LOGS | logstore=%s | project=%s | query=%s | "
@@ -770,7 +810,6 @@ class AliyunLogStore:
             for log in logs:
                 result.append(log.get_contents())
 
-            # Log result count if SQLALCHEMY_ECHO is enabled
             if self.log_enabled:
                 logger.info(
                     "[LogStore] GET_LOGS RESULT | logstore=%s | returned_count=%d",
@@ -845,7 +884,6 @@ class AliyunLogStore:
                 query=full_query,
             )
 
-            # Log query info if SQLALCHEMY_ECHO is enabled
             if self.log_enabled:
                 logger.info(
                     "[LogStore-SDK] EXECUTE_SQL | logstore=%s | project=%s | from_time=%d | to_time=%d | full_query=%s",
@@ -853,8 +891,7 @@ class AliyunLogStore:
                     self.project_name,
                     from_time,
                     to_time,
-                    query,
-                    sql,
+                    full_query,
                 )
 
             try:
@@ -865,7 +902,6 @@ class AliyunLogStore:
                 for log in logs:
                     result.append(log.get_contents())
 
-                # Log result count if SQLALCHEMY_ECHO is enabled
                 if self.log_enabled:
                     logger.info(
                         "[LogStore-SDK] EXECUTE_SQL RESULT | logstore=%s | returned_count=%d",

api/extensions/logstore/aliyun_logstore_pg.py

@@ -7,8 +7,7 @@ from contextlib import contextmanager
 from typing import Any
 
 import psycopg2
-import psycopg2.pool
-from psycopg2 import InterfaceError, OperationalError
+from sqlalchemy import create_engine
 
 from configs import dify_config
 
@@ -16,11 +15,7 @@ logger = logging.getLogger(__name__)
 
 
 class AliyunLogStorePG:
-    """
-    PostgreSQL protocol support for Aliyun SLS LogStore.
-
-    Handles PG connection pooling and operations for regions that support PG protocol.
-    """
+    """PostgreSQL protocol support for Aliyun SLS LogStore using SQLAlchemy connection pool."""
 
     def __init__(self, access_key_id: str, access_key_secret: str, endpoint: str, project_name: str):
         """
@@ -36,24 +31,11 @@ class AliyunLogStorePG:
         self._access_key_secret = access_key_secret
         self._endpoint = endpoint
         self.project_name = project_name
-        self._pg_pool: psycopg2.pool.SimpleConnectionPool | None = None
+        self._engine: Any = None  # SQLAlchemy Engine
         self._use_pg_protocol = False
 
     def _check_port_connectivity(self, host: str, port: int, timeout: float = 2.0) -> bool:
-        """
-        Check if a TCP port is reachable using socket connection.
-
-        This provides a fast check before attempting full database connection,
-        preventing long waits when connecting to unsupported regions.
-
-        Args:
-            host: Hostname or IP address
-            port: Port number
-            timeout: Connection timeout in seconds (default: 2.0)
-
-        Returns:
-            True if port is reachable, False otherwise
-        """
+        """Fast TCP port check to avoid long waits on unsupported regions."""
         try:
             sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
             sock.settimeout(timeout)
@@ -65,166 +47,101 @@ class AliyunLogStorePG:
             return False
 
     def init_connection(self) -> bool:
-        """
-        Initialize PostgreSQL connection pool for SLS PG protocol support.
-
-        Attempts to connect to SLS using PostgreSQL protocol. If successful, sets
-        _use_pg_protocol to True and creates a connection pool. If connection fails
-        (region doesn't support PG protocol or other errors), returns False.
-
-        Returns:
-            True if PG protocol is supported and initialized, False otherwise
-        """
+        """Initialize SQLAlchemy connection pool with pool_recycle and TCP keepalive support."""
         try:
-            # Extract hostname from endpoint (remove protocol if present)
             pg_host = self._endpoint.replace("http://", "").replace("https://", "")
 
-            # Get pool configuration
-            pg_max_connections = int(os.environ.get("ALIYUN_SLS_PG_MAX_CONNECTIONS", 10))
+            # Pool configuration
+            pool_size = int(os.environ.get("ALIYUN_SLS_PG_POOL_SIZE", 5))
+            max_overflow = int(os.environ.get("ALIYUN_SLS_PG_MAX_OVERFLOW", 5))
+            pool_recycle = int(os.environ.get("ALIYUN_SLS_PG_POOL_RECYCLE", 3600))
+            pool_pre_ping = os.environ.get("ALIYUN_SLS_PG_POOL_PRE_PING", "false").lower() == "true"
 
-            logger.debug(
-                "Check PG protocol connection to SLS: host=%s, project=%s",
-                pg_host,
-                self.project_name,
-            )
+            logger.debug("Check PG protocol connection to SLS: host=%s, project=%s", pg_host, self.project_name)
 
-            # Fast port connectivity check before attempting full connection
-            # This prevents long waits when connecting to unsupported regions
+            # Fast port check to avoid long waits
             if not self._check_port_connectivity(pg_host, 5432, timeout=1.0):
-                logger.info(
-                    "USE SDK mode for read/write operations, host=%s",
-                    pg_host,
-                )
+                logger.debug("Using SDK mode for host=%s", pg_host)
                 return False
 
-            # Create connection pool
-            self._pg_pool = psycopg2.pool.SimpleConnectionPool(
-                minconn=1,
-                maxconn=pg_max_connections,
-                host=pg_host,
-                port=5432,
-                database=self.project_name,
-                user=self._access_key_id,
-                password=self._access_key_secret,
-                sslmode="require",
-                connect_timeout=5,
-                application_name=f"Dify-{dify_config.project.version}",
+            # Build connection URL
+            from urllib.parse import quote_plus
+
+            username = quote_plus(self._access_key_id)
+            password = quote_plus(self._access_key_secret)
+            database_url = (
+                f"postgresql+psycopg2://{username}:{password}@{pg_host}:5432/{self.project_name}?sslmode=require"
             )
 
-            # Note: Skip test query because SLS PG protocol only supports SELECT/INSERT on actual tables
-            # Connection pool creation success already indicates connectivity
+            # Create SQLAlchemy engine with connection pool
+            self._engine = create_engine(
+                database_url,
+                pool_size=pool_size,
+                max_overflow=max_overflow,
+                pool_recycle=pool_recycle,
+                pool_pre_ping=pool_pre_ping,
+                pool_timeout=30,
+                connect_args={
+                    "connect_timeout": 5,
+                    "application_name": f"Dify-{dify_config.project.version}-fixautocommit",
+                    "keepalives": 1,
+                    "keepalives_idle": 60,
+                    "keepalives_interval": 10,
+                    "keepalives_count": 5,
+                },
+            )
 
             self._use_pg_protocol = True
             logger.info(
-                "PG protocol initialized successfully for SLS project=%s. Will use PG for read/write operations.",
+                "PG protocol initialized for SLS project=%s (pool_size=%d, pool_recycle=%ds)",
                 self.project_name,
+                pool_size,
+                pool_recycle,
             )
             return True
 
         except Exception as e:
-            # PG connection failed - fallback to SDK mode
             self._use_pg_protocol = False
-            if self._pg_pool:
+            if self._engine:
                 try:
-                    self._pg_pool.closeall()
+                    self._engine.dispose()
                 except Exception:
-                    logger.debug("Failed to close PG connection pool during cleanup, ignoring")
-            self._pg_pool = None
+                    logger.debug("Failed to dispose engine during cleanup, ignoring")
+            self._engine = None
 
-            logger.info(
-                "PG protocol connection failed (region may not support PG protocol): %s. "
-                "Falling back to SDK mode for read/write operations.",
-                str(e),
-            )
-            return False
-
-    def _is_connection_valid(self, conn: Any) -> bool:
-        """
-        Check if a connection is still valid.
-
-        Args:
-            conn: psycopg2 connection object
-
-        Returns:
-            True if connection is valid, False otherwise
-        """
-        try:
-            # Check if connection is closed
-            if conn.closed:
-                return False
-
-            # Quick ping test - execute a lightweight query
-            # For SLS PG protocol, we can't use SELECT 1 without FROM,
-            # so we just check the connection status
-            with conn.cursor() as cursor:
-                cursor.execute("SELECT 1")
-                cursor.fetchone()
-            return True
-        except Exception:
+            logger.debug("Using SDK mode for region: %s", str(e))
             return False
 
     @contextmanager
     def _get_connection(self):
-        """
-        Context manager to get a PostgreSQL connection from the pool.
+        """Get connection from SQLAlchemy pool. Pool handles recycle, invalidation, and keepalive automatically."""
+        if not self._engine:
+            raise RuntimeError("SQLAlchemy engine is not initialized")
 
-        Automatically validates and refreshes stale connections.
-
-        Note: Aliyun SLS PG protocol does not support transactions, so we always
-        use autocommit mode.
-
-        Yields:
-            psycopg2 connection object
-
-        Raises:
-            RuntimeError: If PG pool is not initialized
-        """
-        if not self._pg_pool:
-            raise RuntimeError("PG connection pool is not initialized")
-
-        conn = self._pg_pool.getconn()
+        connection = self._engine.raw_connection()
         try:
-            # Validate connection and get a fresh one if needed
-            if not self._is_connection_valid(conn):
-                logger.debug("Connection is stale, marking as bad and getting a new one")
-                # Mark connection as bad and get a new one
-                self._pg_pool.putconn(conn, close=True)
-                conn = self._pg_pool.getconn()
-
-            # Aliyun SLS PG protocol does not support transactions, always use autocommit
-            conn.autocommit = True
-            yield conn
+            connection.autocommit = True  # SLS PG protocol does not support transactions
+            yield connection
+        except Exception:
+            raise
         finally:
-            # Return connection to pool (or close if it's bad)
-            if self._is_connection_valid(conn):
-                self._pg_pool.putconn(conn)
-            else:
-                self._pg_pool.putconn(conn, close=True)
+            connection.close()
 
     def close(self) -> None:
-        """Close the PostgreSQL connection pool."""
-        if self._pg_pool:
+        """Dispose SQLAlchemy engine and close all connections."""
+        if self._engine:
             try:
-                self._pg_pool.closeall()
-                logger.info("PG connection pool closed")
+                self._engine.dispose()
+                logger.info("SQLAlchemy engine disposed")
             except Exception:
-                logger.exception("Failed to close PG connection pool")
+                logger.exception("Failed to dispose engine")
 
     def _is_retriable_error(self, error: Exception) -> bool:
-        """
-        Check if an error is retriable (connection-related issues).
-
-        Args:
-            error: Exception to check
-
-        Returns:
-            True if the error is retriable, False otherwise
-        """
-        # Retry on connection-related errors
-        if isinstance(error, (OperationalError, InterfaceError)):
+        """Check if error is retriable (connection-related issues)."""
+        # Check for psycopg2 connection errors directly
+        if isinstance(error, (psycopg2.OperationalError, psycopg2.InterfaceError)):
             return True
 
-        # Check error message for specific connection issues
         error_msg = str(error).lower()
         retriable_patterns = [
             "connection",
@@ -234,34 +151,18 @@ class AliyunLogStorePG:
             "reset by peer",
             "no route to host",
             "network",
+            "operational error",
+            "interface error",
         ]
         return any(pattern in error_msg for pattern in retriable_patterns)
 
     def put_log(self, logstore: str, contents: Sequence[tuple[str, str]], log_enabled: bool = False) -> None:
-        """
-        Write log to SLS using PostgreSQL protocol with automatic retry.
-
-        Note: SLS PG protocol only supports INSERT (not UPDATE). This uses append-only
-        writes with log_version field for versioning, same as SDK implementation.
-
-        Args:
-            logstore: Name of the logstore table
-            contents: List of (field_name, value) tuples
-            log_enabled: Whether to enable logging
-
-        Raises:
-            psycopg2.Error: If database operation fails after all retries
-        """
+        """Write log to SLS using INSERT with automatic retry (3 attempts with exponential backoff)."""
         if not contents:
             return
 
-        # Extract field names and values from contents
         fields = [field_name for field_name, _ in contents]
         values = [value for _, value in contents]
-
-        # Build INSERT statement with literal values
-        # Note: Aliyun SLS PG protocol doesn't support parameterized queries,
-        # so we need to use mogrify to safely create literal values
         field_list = ", ".join([f'"{field}"' for field in fields])
 
         if log_enabled:
@@ -272,67 +173,40 @@ class AliyunLogStorePG:
                 len(contents),
             )
 
-        # Retry configuration
         max_retries = 3
-        retry_delay = 0.1  # Start with 100ms
+        retry_delay = 0.1
 
         for attempt in range(max_retries):
             try:
                 with self._get_connection() as conn:
                     with conn.cursor() as cursor:
-                        # Use mogrify to safely convert values to SQL literals
                         placeholders = ", ".join(["%s"] * len(fields))
                         values_literal = cursor.mogrify(f"({placeholders})", values).decode("utf-8")
                         insert_sql = f'INSERT INTO "{logstore}" ({field_list}) VALUES {values_literal}'
                         cursor.execute(insert_sql)
-                # Success - exit retry loop
                 return
 
             except psycopg2.Error as e:
-                # Check if error is retriable
                 if not self._is_retriable_error(e):
-                    # Not a retriable error (e.g., data validation error), fail immediately
-                    logger.exception(
-                        "Failed to put logs to logstore %s via PG protocol (non-retriable error)",
-                        logstore,
-                    )
+                    logger.exception("Failed to put logs to logstore %s (non-retriable error)", logstore)
                     raise
 
-                # Retriable error - log and retry if we have attempts left
                 if attempt < max_retries - 1:
                     logger.warning(
-                        "Failed to put logs to logstore %s via PG protocol (attempt %d/%d): %s. Retrying...",
+                        "Failed to put logs to logstore %s (attempt %d/%d): %s. Retrying...",
                         logstore,
                         attempt + 1,
                         max_retries,
                         str(e),
                     )
                     time.sleep(retry_delay)
-                    retry_delay *= 2  # Exponential backoff
+                    retry_delay *= 2
                 else:
-                    # Last attempt failed
-                    logger.exception(
-                        "Failed to put logs to logstore %s via PG protocol after %d attempts",
-                        logstore,
-                        max_retries,
-                    )
+                    logger.exception("Failed to put logs to logstore %s after %d attempts", logstore, max_retries)
                     raise
 
     def execute_sql(self, sql: str, logstore: str, log_enabled: bool = False) -> list[dict[str, Any]]:
-        """
-        Execute SQL query using PostgreSQL protocol with automatic retry.
-
-        Args:
-            sql: SQL query string
-            logstore: Name of the logstore (for logging purposes)
-            log_enabled: Whether to enable logging
-
-        Returns:
-            List of result rows as dictionaries
-
-        Raises:
-            psycopg2.Error: If database operation fails after all retries
-        """
+        """Execute SQL query with automatic retry (3 attempts with exponential backoff)."""
         if log_enabled:
             logger.info(
                 "[LogStore-PG] EXECUTE_SQL | logstore=%s | project=%s | sql=%s",
@@ -341,20 +215,16 @@ class AliyunLogStorePG:
                 sql,
             )
 
-        # Retry configuration
         max_retries = 3
-        retry_delay = 0.1  # Start with 100ms
+        retry_delay = 0.1
 
         for attempt in range(max_retries):
             try:
                 with self._get_connection() as conn:
                     with conn.cursor() as cursor:
                         cursor.execute(sql)
-
-                        # Get column names from cursor description
                         columns = [desc[0] for desc in cursor.description]
 
-                        # Fetch all results and convert to list of dicts
                         result = []
                         for row in cursor.fetchall():
                             row_dict = {}
@@ -372,36 +242,31 @@ class AliyunLogStorePG:
                         return result
 
             except psycopg2.Error as e:
-                # Check if error is retriable
                 if not self._is_retriable_error(e):
-                    # Not a retriable error (e.g., SQL syntax error), fail immediately
                     logger.exception(
-                        "Failed to execute SQL query on logstore %s via PG protocol (non-retriable error): sql=%s",
+                        "Failed to execute SQL on logstore %s (non-retriable error): sql=%s",
                         logstore,
                         sql,
                     )
                     raise
 
-                # Retriable error - log and retry if we have attempts left
                 if attempt < max_retries - 1:
                     logger.warning(
-                        "Failed to execute SQL query on logstore %s via PG protocol (attempt %d/%d): %s. Retrying...",
+                        "Failed to execute SQL on logstore %s (attempt %d/%d): %s. Retrying...",
                         logstore,
                         attempt + 1,
                         max_retries,
                         str(e),
                     )
                     time.sleep(retry_delay)
-                    retry_delay *= 2  # Exponential backoff
+                    retry_delay *= 2
                 else:
-                    # Last attempt failed
                     logger.exception(
-                        "Failed to execute SQL query on logstore %s via PG protocol after %d attempts: sql=%s",
+                        "Failed to execute SQL on logstore %s after %d attempts: sql=%s",
                         logstore,
                         max_retries,
                         sql,
                     )
                     raise
 
-        # This line should never be reached due to raise above, but makes type checker happy
         return []

api/extensions/logstore/repositories/__init__.py

@@ -0,0 +1,29 @@
+"""
+LogStore repository utilities.
+"""
+
+from typing import Any
+
+
+def safe_float(value: Any, default: float = 0.0) -> float:
+    """
+    Safely convert a value to float, handling 'null' strings and None.
+    """
+    if value is None or value in {"null", ""}:
+        return default
+    try:
+        return float(value)
+    except (ValueError, TypeError):
+        return default
+
+
+def safe_int(value: Any, default: int = 0) -> int:
+    """
+    Safely convert a value to int, handling 'null' strings and None.
+    """
+    if value is None or value in {"null", ""}:
+        return default
+    try:
+        return int(float(value))
+    except (ValueError, TypeError):
+        return default

api/extensions/logstore/repositories/logstore_api_workflow_node_execution_repository.py

@@ -14,6 +14,8 @@ from typing import Any
 from sqlalchemy.orm import sessionmaker
 
 from extensions.logstore.aliyun_logstore import AliyunLogStore
+from extensions.logstore.repositories import safe_float, safe_int
+from extensions.logstore.sql_escape import escape_identifier, escape_logstore_query_value
 from models.workflow import WorkflowNodeExecutionModel
 from repositories.api_workflow_node_execution_repository import DifyAPIWorkflowNodeExecutionRepository
 
@@ -52,9 +54,8 @@ def _dict_to_workflow_node_execution_model(data: dict[str, Any]) -> WorkflowNode
     model.created_by_role = data.get("created_by_role") or ""
     model.created_by = data.get("created_by") or ""
 
-    # Numeric fields with defaults
-    model.index = int(data.get("index", 0))
-    model.elapsed_time = float(data.get("elapsed_time", 0))
+    model.index = safe_int(data.get("index", 0))
+    model.elapsed_time = safe_float(data.get("elapsed_time", 0))
 
     # Optional fields
     model.workflow_run_id = data.get("workflow_run_id")
@@ -130,6 +131,12 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep
             node_id,
         )
         try:
+            # Escape parameters to prevent SQL injection
+            escaped_tenant_id = escape_identifier(tenant_id)
+            escaped_app_id = escape_identifier(app_id)
+            escaped_workflow_id = escape_identifier(workflow_id)
+            escaped_node_id = escape_identifier(node_id)
+
             # Check if PG protocol is supported
             if self.logstore_client.supports_pg_protocol:
                 # Use PG protocol with SQL query (get latest version of each record)
@@ -138,10 +145,10 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep
                         SELECT *, 
                             ROW_NUMBER() OVER (PARTITION BY id ORDER BY log_version DESC) as rn
                         FROM "{AliyunLogStore.workflow_node_execution_logstore}"
-                        WHERE tenant_id = '{tenant_id}' 
-                          AND app_id = '{app_id}' 
-                          AND workflow_id = '{workflow_id}' 
-                          AND node_id = '{node_id}'
+                        WHERE tenant_id = '{escaped_tenant_id}' 
+                          AND app_id = '{escaped_app_id}' 
+                          AND workflow_id = '{escaped_workflow_id}' 
+                          AND node_id = '{escaped_node_id}'
                           AND __time__ > 0
                     ) AS subquery WHERE rn = 1
                     LIMIT 100
@@ -153,7 +160,8 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep
             else:
                 # Use SDK with LogStore query syntax
                 query = (
-                    f"tenant_id: {tenant_id} and app_id: {app_id} and workflow_id: {workflow_id} and node_id: {node_id}"
+                    f"tenant_id: {escaped_tenant_id} and app_id: {escaped_app_id} "
+                    f"and workflow_id: {escaped_workflow_id} and node_id: {escaped_node_id}"
                 )
                 from_time = 0
                 to_time = int(time.time())  # now
@@ -227,6 +235,11 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep
             workflow_run_id,
         )
         try:
+            # Escape parameters to prevent SQL injection
+            escaped_tenant_id = escape_identifier(tenant_id)
+            escaped_app_id = escape_identifier(app_id)
+            escaped_workflow_run_id = escape_identifier(workflow_run_id)
+
             # Check if PG protocol is supported
             if self.logstore_client.supports_pg_protocol:
                 # Use PG protocol with SQL query (get latest version of each record)
@@ -235,9 +248,9 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep
                         SELECT *, 
                             ROW_NUMBER() OVER (PARTITION BY id ORDER BY log_version DESC) as rn
                         FROM "{AliyunLogStore.workflow_node_execution_logstore}"
-                        WHERE tenant_id = '{tenant_id}' 
-                          AND app_id = '{app_id}' 
-                          AND workflow_run_id = '{workflow_run_id}'
+                        WHERE tenant_id = '{escaped_tenant_id}' 
+                          AND app_id = '{escaped_app_id}' 
+                          AND workflow_run_id = '{escaped_workflow_run_id}'
                           AND __time__ > 0
                     ) AS subquery WHERE rn = 1
                     LIMIT 1000
@@ -248,7 +261,10 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep
                 )
             else:
                 # Use SDK with LogStore query syntax
-                query = f"tenant_id: {tenant_id} and app_id: {app_id} and workflow_run_id: {workflow_run_id}"
+                query = (
+                    f"tenant_id: {escaped_tenant_id} and app_id: {escaped_app_id} "
+                    f"and workflow_run_id: {escaped_workflow_run_id}"
+                )
                 from_time = 0
                 to_time = int(time.time())  # now
 
@@ -313,16 +329,24 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep
         """
         logger.debug("get_execution_by_id: execution_id=%s, tenant_id=%s", execution_id, tenant_id)
         try:
+            # Escape parameters to prevent SQL injection
+            escaped_execution_id = escape_identifier(execution_id)
+
             # Check if PG protocol is supported
             if self.logstore_client.supports_pg_protocol:
                 # Use PG protocol with SQL query (get latest version of record)
-                tenant_filter = f"AND tenant_id = '{tenant_id}'" if tenant_id else ""
+                if tenant_id:
+                    escaped_tenant_id = escape_identifier(tenant_id)
+                    tenant_filter = f"AND tenant_id = '{escaped_tenant_id}'"
+                else:
+                    tenant_filter = ""
+
                 sql_query = f"""
                     SELECT * FROM (
                         SELECT *, 
                             ROW_NUMBER() OVER (PARTITION BY id ORDER BY log_version DESC) as rn
                         FROM "{AliyunLogStore.workflow_node_execution_logstore}"
-                        WHERE id = '{execution_id}' {tenant_filter} AND __time__ > 0
+                        WHERE id = '{escaped_execution_id}' {tenant_filter} AND __time__ > 0
                     ) AS subquery WHERE rn = 1
                     LIMIT 1
                 """
@@ -332,10 +356,14 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep
                 )
             else:
                 # Use SDK with LogStore query syntax
+                # Note: Values must be quoted in LogStore query syntax to prevent injection
                 if tenant_id:
-                    query = f"id: {execution_id} and tenant_id: {tenant_id}"
+                    query = (
+                        f"id:{escape_logstore_query_value(execution_id)} "
+                        f"and tenant_id:{escape_logstore_query_value(tenant_id)}"
+                    )
                 else:
-                    query = f"id: {execution_id}"
+                    query = f"id:{escape_logstore_query_value(execution_id)}"
 
                 from_time = 0
                 to_time = int(time.time())  # now

api/extensions/logstore/repositories/logstore_api_workflow_run_repository.py

@@ -10,6 +10,7 @@ Key Features:
 - Optimized deduplication using finished_at IS NOT NULL filter
 - Window functions only when necessary (running status queries)
 - Multi-tenant data isolation and security
+- SQL injection prevention via parameter escaping
 """
 
 import logging
@@ -22,6 +23,8 @@ from typing import Any, cast
 from sqlalchemy.orm import sessionmaker
 
 from extensions.logstore.aliyun_logstore import AliyunLogStore
+from extensions.logstore.repositories import safe_float, safe_int
+from extensions.logstore.sql_escape import escape_identifier, escape_logstore_query_value, escape_sql_string
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from models.enums import WorkflowRunTriggeredFrom
 from models.workflow import WorkflowRun
@@ -63,10 +66,9 @@ def _dict_to_workflow_run(data: dict[str, Any]) -> WorkflowRun:
     model.created_by_role = data.get("created_by_role") or ""
     model.created_by = data.get("created_by") or ""
 
-    # Numeric fields with defaults
-    model.total_tokens = int(data.get("total_tokens", 0))
-    model.total_steps = int(data.get("total_steps", 0))
-    model.exceptions_count = int(data.get("exceptions_count", 0))
+    model.total_tokens = safe_int(data.get("total_tokens", 0))
+    model.total_steps = safe_int(data.get("total_steps", 0))
+    model.exceptions_count = safe_int(data.get("exceptions_count", 0))
 
     # Optional fields
     model.graph = data.get("graph")
@@ -101,7 +103,8 @@ def _dict_to_workflow_run(data: dict[str, Any]) -> WorkflowRun:
     if model.finished_at and model.created_at:
         model.elapsed_time = (model.finished_at - model.created_at).total_seconds()
     else:
-        model.elapsed_time = float(data.get("elapsed_time", 0))
+        # Use safe conversion to handle 'null' strings and None values
+        model.elapsed_time = safe_float(data.get("elapsed_time", 0))
 
     return model
 
@@ -165,16 +168,26 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
             status,
         )
         # Convert triggered_from to list if needed
-        if isinstance(triggered_from, WorkflowRunTriggeredFrom):
+        if isinstance(triggered_from, (WorkflowRunTriggeredFrom, str)):
             triggered_from_list = [triggered_from]
         else:
             triggered_from_list = list(triggered_from)
 
-        # Build triggered_from filter
-        triggered_from_filter = " OR ".join([f"triggered_from='{tf.value}'" for tf in triggered_from_list])
+        # Escape parameters to prevent SQL injection
+        escaped_tenant_id = escape_identifier(tenant_id)
+        escaped_app_id = escape_identifier(app_id)
 
-        # Build status filter
-        status_filter = f"AND status='{status}'" if status else ""
+        # Build triggered_from filter with escaped values
+        # Support both enum and string values for triggered_from
+        triggered_from_filter = " OR ".join(
+            [
+                f"triggered_from='{escape_sql_string(tf.value if isinstance(tf, WorkflowRunTriggeredFrom) else tf)}'"
+                for tf in triggered_from_list
+            ]
+        )
+
+        # Build status filter with escaped value
+        status_filter = f"AND status='{escape_sql_string(status)}'" if status else ""
 
         # Build last_id filter for pagination
         # Note: This is simplified. In production, you'd need to track created_at from last record
@@ -188,8 +201,8 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
             SELECT * FROM (
                 SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY log_version DESC) AS rn
                 FROM {AliyunLogStore.workflow_execution_logstore}
-                WHERE tenant_id='{tenant_id}'
-                  AND app_id='{app_id}'
+                WHERE tenant_id='{escaped_tenant_id}'
+                  AND app_id='{escaped_app_id}'
                   AND ({triggered_from_filter})
                   {status_filter}
                   {last_id_filter}
@@ -232,6 +245,11 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
         logger.debug("get_workflow_run_by_id: tenant_id=%s, app_id=%s, run_id=%s", tenant_id, app_id, run_id)
 
         try:
+            # Escape parameters to prevent SQL injection
+            escaped_run_id = escape_identifier(run_id)
+            escaped_tenant_id = escape_identifier(tenant_id)
+            escaped_app_id = escape_identifier(app_id)
+
             # Check if PG protocol is supported
             if self.logstore_client.supports_pg_protocol:
                 # Use PG protocol with SQL query (get latest version of record)
@@ -240,7 +258,10 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
                         SELECT *, 
                             ROW_NUMBER() OVER (PARTITION BY id ORDER BY log_version DESC) as rn
                         FROM "{AliyunLogStore.workflow_execution_logstore}"
-                        WHERE id = '{run_id}' AND tenant_id = '{tenant_id}' AND app_id = '{app_id}' AND __time__ > 0
+                        WHERE id = '{escaped_run_id}' 
+                          AND tenant_id = '{escaped_tenant_id}' 
+                          AND app_id = '{escaped_app_id}' 
+                          AND __time__ > 0
                     ) AS subquery WHERE rn = 1
                     LIMIT 100
                 """
@@ -250,7 +271,12 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
                 )
             else:
                 # Use SDK with LogStore query syntax
-                query = f"id: {run_id} and tenant_id: {tenant_id} and app_id: {app_id}"
+                # Note: Values must be quoted in LogStore query syntax to prevent injection
+                query = (
+                    f"id:{escape_logstore_query_value(run_id)} "
+                    f"and tenant_id:{escape_logstore_query_value(tenant_id)} "
+                    f"and app_id:{escape_logstore_query_value(app_id)}"
+                )
                 from_time = 0
                 to_time = int(time.time())  # now
 
@@ -323,6 +349,9 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
         logger.debug("get_workflow_run_by_id_without_tenant: run_id=%s", run_id)
 
         try:
+            # Escape parameter to prevent SQL injection
+            escaped_run_id = escape_identifier(run_id)
+
             # Check if PG protocol is supported
             if self.logstore_client.supports_pg_protocol:
                 # Use PG protocol with SQL query (get latest version of record)
@@ -331,7 +360,7 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
                         SELECT *, 
                             ROW_NUMBER() OVER (PARTITION BY id ORDER BY log_version DESC) as rn
                         FROM "{AliyunLogStore.workflow_execution_logstore}"
-                        WHERE id = '{run_id}' AND __time__ > 0
+                        WHERE id = '{escaped_run_id}' AND __time__ > 0
                     ) AS subquery WHERE rn = 1
                     LIMIT 100
                 """
@@ -341,7 +370,8 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
                 )
             else:
                 # Use SDK with LogStore query syntax
-                query = f"id: {run_id}"
+                # Note: Values must be quoted in LogStore query syntax
+                query = f"id:{escape_logstore_query_value(run_id)}"
                 from_time = 0
                 to_time = int(time.time())  # now
 
@@ -410,6 +440,11 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
             triggered_from,
             status,
         )
+        # Escape parameters to prevent SQL injection
+        escaped_tenant_id = escape_identifier(tenant_id)
+        escaped_app_id = escape_identifier(app_id)
+        escaped_triggered_from = escape_sql_string(triggered_from)
+
         # Build time range filter
         time_filter = ""
         if time_range:
@@ -418,6 +453,8 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
 
         # If status is provided, simple count
         if status:
+            escaped_status = escape_sql_string(status)
+
             if status == "running":
                 # Running status requires window function
                 sql = f"""
@@ -425,9 +462,9 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
                     FROM (
                         SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY log_version DESC) AS rn
                         FROM {AliyunLogStore.workflow_execution_logstore}
-                        WHERE tenant_id='{tenant_id}'
-                          AND app_id='{app_id}'
-                          AND triggered_from='{triggered_from}'
+                        WHERE tenant_id='{escaped_tenant_id}'
+                          AND app_id='{escaped_app_id}'
+                          AND triggered_from='{escaped_triggered_from}'
                           AND status='running'
                           {time_filter}
                     ) t
@@ -438,10 +475,10 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
                 sql = f"""
                     SELECT COUNT(DISTINCT id) as count
                     FROM {AliyunLogStore.workflow_execution_logstore}
-                    WHERE tenant_id='{tenant_id}'
-                      AND app_id='{app_id}'
-                      AND triggered_from='{triggered_from}'
-                      AND status='{status}'
+                    WHERE tenant_id='{escaped_tenant_id}'
+                      AND app_id='{escaped_app_id}'
+                      AND triggered_from='{escaped_triggered_from}'
+                      AND status='{escaped_status}'
                       AND finished_at IS NOT NULL
                       {time_filter}
                 """
@@ -467,13 +504,14 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
         # No status filter - get counts grouped by status
         # Use optimized query for finished runs, separate query for running
         try:
+            # Escape parameters (already escaped above, reuse variables)
             # Count finished runs grouped by status
             finished_sql = f"""
                 SELECT status, COUNT(DISTINCT id) as count
                 FROM {AliyunLogStore.workflow_execution_logstore}
-                WHERE tenant_id='{tenant_id}'
-                  AND app_id='{app_id}'
-                  AND triggered_from='{triggered_from}'
+                WHERE tenant_id='{escaped_tenant_id}'
+                  AND app_id='{escaped_app_id}'
+                  AND triggered_from='{escaped_triggered_from}'
                   AND finished_at IS NOT NULL
                   {time_filter}
                 GROUP BY status
@@ -485,9 +523,9 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
                 FROM (
                     SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY log_version DESC) AS rn
                     FROM {AliyunLogStore.workflow_execution_logstore}
-                    WHERE tenant_id='{tenant_id}'
-                      AND app_id='{app_id}'
-                      AND triggered_from='{triggered_from}'
+                    WHERE tenant_id='{escaped_tenant_id}'
+                      AND app_id='{escaped_app_id}'
+                      AND triggered_from='{escaped_triggered_from}'
                       AND status='running'
                       {time_filter}
                 ) t
@@ -546,7 +584,13 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
         logger.debug(
             "get_daily_runs_statistics: tenant_id=%s, app_id=%s, triggered_from=%s", tenant_id, app_id, triggered_from
         )
-        # Build time range filter
+
+        # Escape parameters to prevent SQL injection
+        escaped_tenant_id = escape_identifier(tenant_id)
+        escaped_app_id = escape_identifier(app_id)
+        escaped_triggered_from = escape_sql_string(triggered_from)
+
+        # Build time range filter (datetime.isoformat() is safe)
         time_filter = ""
         if start_date:
             time_filter += f" AND __time__ >= to_unixtime(from_iso8601_timestamp('{start_date.isoformat()}'))"
@@ -557,9 +601,9 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
         sql = f"""
             SELECT DATE(from_unixtime(__time__)) as date, COUNT(DISTINCT id) as runs
             FROM {AliyunLogStore.workflow_execution_logstore}
-            WHERE tenant_id='{tenant_id}'
-              AND app_id='{app_id}'
-              AND triggered_from='{triggered_from}'
+            WHERE tenant_id='{escaped_tenant_id}'
+              AND app_id='{escaped_app_id}'
+              AND triggered_from='{escaped_triggered_from}'
               AND finished_at IS NOT NULL
               {time_filter}
             GROUP BY date
@@ -601,7 +645,13 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
             app_id,
             triggered_from,
         )
-        # Build time range filter
+
+        # Escape parameters to prevent SQL injection
+        escaped_tenant_id = escape_identifier(tenant_id)
+        escaped_app_id = escape_identifier(app_id)
+        escaped_triggered_from = escape_sql_string(triggered_from)
+
+        # Build time range filter (datetime.isoformat() is safe)
         time_filter = ""
         if start_date:
             time_filter += f" AND __time__ >= to_unixtime(from_iso8601_timestamp('{start_date.isoformat()}'))"
@@ -611,9 +661,9 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
         sql = f"""
             SELECT DATE(from_unixtime(__time__)) as date, COUNT(DISTINCT created_by) as terminal_count
             FROM {AliyunLogStore.workflow_execution_logstore}
-            WHERE tenant_id='{tenant_id}'
-              AND app_id='{app_id}'
-              AND triggered_from='{triggered_from}'
+            WHERE tenant_id='{escaped_tenant_id}'
+              AND app_id='{escaped_app_id}'
+              AND triggered_from='{escaped_triggered_from}'
               AND finished_at IS NOT NULL
               {time_filter}
             GROUP BY date
@@ -655,7 +705,13 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
             app_id,
             triggered_from,
         )
-        # Build time range filter
+
+        # Escape parameters to prevent SQL injection
+        escaped_tenant_id = escape_identifier(tenant_id)
+        escaped_app_id = escape_identifier(app_id)
+        escaped_triggered_from = escape_sql_string(triggered_from)
+
+        # Build time range filter (datetime.isoformat() is safe)
         time_filter = ""
         if start_date:
             time_filter += f" AND __time__ >= to_unixtime(from_iso8601_timestamp('{start_date.isoformat()}'))"
@@ -665,9 +721,9 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
         sql = f"""
             SELECT DATE(from_unixtime(__time__)) as date, SUM(total_tokens) as token_count
             FROM {AliyunLogStore.workflow_execution_logstore}
-            WHERE tenant_id='{tenant_id}'
-              AND app_id='{app_id}'
-              AND triggered_from='{triggered_from}'
+            WHERE tenant_id='{escaped_tenant_id}'
+              AND app_id='{escaped_app_id}'
+              AND triggered_from='{escaped_triggered_from}'
               AND finished_at IS NOT NULL
               {time_filter}
             GROUP BY date
@@ -709,7 +765,13 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
             app_id,
             triggered_from,
         )
-        # Build time range filter
+
+        # Escape parameters to prevent SQL injection
+        escaped_tenant_id = escape_identifier(tenant_id)
+        escaped_app_id = escape_identifier(app_id)
+        escaped_triggered_from = escape_sql_string(triggered_from)
+
+        # Build time range filter (datetime.isoformat() is safe)
         time_filter = ""
         if start_date:
             time_filter += f" AND __time__ >= to_unixtime(from_iso8601_timestamp('{start_date.isoformat()}'))"
@@ -726,9 +788,9 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository):
                     created_by,
                     COUNT(DISTINCT id) AS interactions
                 FROM {AliyunLogStore.workflow_execution_logstore}
-                WHERE tenant_id='{tenant_id}'
-                  AND app_id='{app_id}'
-                  AND triggered_from='{triggered_from}'
+                WHERE tenant_id='{escaped_tenant_id}'
+                  AND app_id='{escaped_app_id}'
+                  AND triggered_from='{escaped_triggered_from}'
                   AND finished_at IS NOT NULL
                   {time_filter}
                 GROUP BY date, created_by

api/extensions/logstore/repositories/logstore_workflow_execution_repository.py

@@ -10,6 +10,7 @@ from sqlalchemy.orm import sessionmaker
 from core.repositories.sqlalchemy_workflow_execution_repository import SQLAlchemyWorkflowExecutionRepository
 from core.workflow.entities import WorkflowExecution
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
+from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from extensions.logstore.aliyun_logstore import AliyunLogStore
 from libs.helper import extract_tenant_id
 from models import (
@@ -22,18 +23,6 @@ from models.enums import WorkflowRunTriggeredFrom
 logger = logging.getLogger(__name__)
 
 
-def to_serializable(obj):
-    """
-    Convert non-JSON-serializable objects into JSON-compatible formats.
-
-    - Uses `to_dict()` if it's a callable method.
-    - Falls back to string representation.
-    """
-    if hasattr(obj, "to_dict") and callable(obj.to_dict):
-        return obj.to_dict()
-    return str(obj)
-
-
 class LogstoreWorkflowExecutionRepository(WorkflowExecutionRepository):
     def __init__(
         self,
@@ -79,7 +68,7 @@ class LogstoreWorkflowExecutionRepository(WorkflowExecutionRepository):
 
         # Control flag for dual-write (write to both LogStore and SQL database)
         # Set to True to enable dual-write for safe migration, False to use LogStore only
-        self._enable_dual_write = os.environ.get("LOGSTORE_DUAL_WRITE_ENABLED", "true").lower() == "true"
+        self._enable_dual_write = os.environ.get("LOGSTORE_DUAL_WRITE_ENABLED", "false").lower() == "true"
 
         # Control flag for whether to write the `graph` field to LogStore.
         # If LOGSTORE_ENABLE_PUT_GRAPH_FIELD is "true", write the full `graph` field;
@@ -113,6 +102,9 @@ class LogstoreWorkflowExecutionRepository(WorkflowExecutionRepository):
         # Generate log_version as nanosecond timestamp for record versioning
         log_version = str(time.time_ns())
 
+        # Use WorkflowRuntimeTypeConverter to handle complex types (Segment, File, etc.)
+        json_converter = WorkflowRuntimeTypeConverter()
+
         logstore_model = [
             ("id", domain_model.id_),
             ("log_version", log_version),  # Add log_version field for append-only writes
@@ -127,19 +119,19 @@ class LogstoreWorkflowExecutionRepository(WorkflowExecutionRepository):
             ("version", domain_model.workflow_version),
             (
                 "graph",
-                json.dumps(domain_model.graph, ensure_ascii=False, default=to_serializable)
+                json.dumps(json_converter.to_json_encodable(domain_model.graph), ensure_ascii=False)
                 if domain_model.graph and self._enable_put_graph_field
                 else "{}",
             ),
             (
                 "inputs",
-                json.dumps(domain_model.inputs, ensure_ascii=False, default=to_serializable)
+                json.dumps(json_converter.to_json_encodable(domain_model.inputs), ensure_ascii=False)
                 if domain_model.inputs
                 else "{}",
             ),
             (
                 "outputs",
-                json.dumps(domain_model.outputs, ensure_ascii=False, default=to_serializable)
+                json.dumps(json_converter.to_json_encodable(domain_model.outputs), ensure_ascii=False)
                 if domain_model.outputs
                 else "{}",
             ),

api/extensions/logstore/repositories/logstore_workflow_node_execution_repository.py

@@ -24,6 +24,8 @@ from core.workflow.enums import NodeType
 from core.workflow.repositories.workflow_node_execution_repository import OrderConfig, WorkflowNodeExecutionRepository
 from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from extensions.logstore.aliyun_logstore import AliyunLogStore
+from extensions.logstore.repositories import safe_float, safe_int
+from extensions.logstore.sql_escape import escape_identifier
 from libs.helper import extract_tenant_id
 from models import (
     Account,
@@ -73,7 +75,7 @@ def _dict_to_workflow_node_execution(data: dict[str, Any]) -> WorkflowNodeExecut
         node_execution_id=data.get("node_execution_id"),
         workflow_id=data.get("workflow_id", ""),
         workflow_execution_id=data.get("workflow_run_id"),
-        index=int(data.get("index", 0)),
+        index=safe_int(data.get("index", 0)),
         predecessor_node_id=data.get("predecessor_node_id"),
         node_id=data.get("node_id", ""),
         node_type=NodeType(data.get("node_type", "start")),
@@ -83,7 +85,7 @@ def _dict_to_workflow_node_execution(data: dict[str, Any]) -> WorkflowNodeExecut
         outputs=outputs,
         status=status,
         error=data.get("error"),
-        elapsed_time=float(data.get("elapsed_time", 0.0)),
+        elapsed_time=safe_float(data.get("elapsed_time", 0.0)),
         metadata=domain_metadata,
         created_at=created_at,
         finished_at=finished_at,
@@ -147,7 +149,7 @@ class LogstoreWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
 
         # Control flag for dual-write (write to both LogStore and SQL database)
         # Set to True to enable dual-write for safe migration, False to use LogStore only
-        self._enable_dual_write = os.environ.get("LOGSTORE_DUAL_WRITE_ENABLED", "true").lower() == "true"
+        self._enable_dual_write = os.environ.get("LOGSTORE_DUAL_WRITE_ENABLED", "false").lower() == "true"
 
     def _to_logstore_model(self, domain_model: WorkflowNodeExecution) -> Sequence[tuple[str, str]]:
         logger.debug(
@@ -274,16 +276,34 @@ class LogstoreWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
         Save or update the inputs, process_data, or outputs associated with a specific
         node_execution record.
 
-        For LogStore implementation, this is similar to save() since we always write
-        complete records. We append a new record with updated data fields.
+        For LogStore implementation, this is a no-op for the LogStore write because save()
+        already writes all fields including inputs, process_data, and outputs. The caller
+        typically calls save() first to persist status/metadata, then calls save_execution_data()
+        to persist data fields. Since LogStore writes complete records atomically, we don't
+        need a separate write here to avoid duplicate records.
+
+        However, if dual-write is enabled, we still need to call the SQL repository's
+        save_execution_data() method to properly update the SQL database.
 
         Args:
             execution: The NodeExecution instance with data to save
         """
-        logger.debug("save_execution_data: id=%s, node_execution_id=%s", execution.id, execution.node_execution_id)
-        # In LogStore, we simply write a new complete record with the data
-        # The log_version timestamp will ensure this is treated as the latest version
-        self.save(execution)
+        logger.debug(
+            "save_execution_data: no-op for LogStore (data already saved by save()): id=%s, node_execution_id=%s",
+            execution.id,
+            execution.node_execution_id,
+        )
+        # No-op for LogStore: save() already writes all fields including inputs, process_data, and outputs
+        # Calling save() again would create a duplicate record in the append-only LogStore
+
+        # Dual-write to SQL database if enabled (for safe migration)
+        if self._enable_dual_write:
+            try:
+                self.sql_repository.save_execution_data(execution)
+                logger.debug("Dual-write: saved node execution data to SQL database: id=%s", execution.id)
+            except Exception:
+                logger.exception("Failed to dual-write node execution data to SQL database: id=%s", execution.id)
+                # Don't raise - LogStore write succeeded, SQL is just a backup
 
     def get_by_workflow_run(
         self,
@@ -292,8 +312,8 @@ class LogstoreWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
     ) -> Sequence[WorkflowNodeExecution]:
         """
         Retrieve all NodeExecution instances for a specific workflow run.
-        Uses LogStore SQL query with finished_at IS NOT NULL filter for deduplication.
-        This ensures we only get the final version of each node execution.
+        Uses LogStore SQL query with window function to get the latest version of each node execution.
+        This ensures we only get the most recent version of each node execution record.
         Args:
             workflow_run_id: The workflow run ID
             order_config: Optional configuration for ordering results
@@ -304,16 +324,19 @@ class LogstoreWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
             A list of NodeExecution instances
 
         Note:
-            This method filters by finished_at IS NOT NULL to avoid duplicates from
-            version updates. For complete history including intermediate states,
-            a different query strategy would be needed.
+            This method uses ROW_NUMBER() window function partitioned by node_execution_id
+            to get the latest version (highest log_version) of each node execution.
         """
         logger.debug("get_by_workflow_run: workflow_run_id=%s, order_config=%s", workflow_run_id, order_config)
-        # Build SQL query with deduplication using finished_at IS NOT NULL
-        # This optimization avoids window functions for common case where we only
-        # want the final state of each node execution
+        # Build SQL query with deduplication using window function
+        # ROW_NUMBER() OVER (PARTITION BY node_execution_id ORDER BY log_version DESC)
+        # ensures we get the latest version of each node execution
 
-        # Build ORDER BY clause
+        # Escape parameters to prevent SQL injection
+        escaped_workflow_run_id = escape_identifier(workflow_run_id)
+        escaped_tenant_id = escape_identifier(self._tenant_id)
+
+        # Build ORDER BY clause for outer query
         order_clause = ""
         if order_config and order_config.order_by:
             order_fields = []
@@ -327,16 +350,23 @@ class LogstoreWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
             if order_fields:
                 order_clause = "ORDER BY " + ", ".join(order_fields)
 
-        sql = f"""
-            SELECT *
-            FROM {AliyunLogStore.workflow_node_execution_logstore}
-            WHERE workflow_run_id='{workflow_run_id}'
-              AND tenant_id='{self._tenant_id}'
-              AND finished_at IS NOT NULL
-        """
-
+        # Build app_id filter for subquery
+        app_id_filter = ""
         if self._app_id:
-            sql += f" AND app_id='{self._app_id}'"
+            escaped_app_id = escape_identifier(self._app_id)
+            app_id_filter = f" AND app_id='{escaped_app_id}'"
+
+        # Use window function to get latest version of each node execution
+        sql = f"""
+            SELECT * FROM (
+                SELECT *, ROW_NUMBER() OVER (PARTITION BY node_execution_id ORDER BY log_version DESC) AS rn
+                FROM {AliyunLogStore.workflow_node_execution_logstore}
+                WHERE workflow_run_id='{escaped_workflow_run_id}'
+                  AND tenant_id='{escaped_tenant_id}'
+                  {app_id_filter}
+            ) t
+            WHERE rn = 1
+        """
 
         if order_clause:
             sql += f" {order_clause}"

api/extensions/logstore/sql_escape.py

@@ -0,0 +1,134 @@
+"""
+SQL Escape Utility for LogStore Queries
+
+This module provides escaping utilities to prevent injection attacks in LogStore queries.
+
+LogStore supports two query modes:
+1. PG Protocol Mode: Uses SQL syntax with single quotes for strings
+2. SDK Mode: Uses LogStore query syntax (key: value) with double quotes
+
+Key Security Concerns:
+- Prevent tenant A from accessing tenant B's data via injection
+- SLS queries are read-only, so we focus on data access control
+- Different escaping strategies for SQL vs LogStore query syntax
+"""
+
+
+def escape_sql_string(value: str) -> str:
+    """
+    Escape a string value for safe use in SQL queries.
+
+    This function escapes single quotes by doubling them, which is the standard
+    SQL escaping method. This prevents SQL injection by ensuring that user input
+    cannot break out of string literals.
+
+    Args:
+        value: The string value to escape
+
+    Returns:
+        Escaped string safe for use in SQL queries
+
+    Examples:
+        >>> escape_sql_string("normal_value")
+        "normal_value"
+        >>> escape_sql_string("value' OR '1'='1")
+        "value'' OR ''1''=''1"
+        >>> escape_sql_string("tenant's_id")
+        "tenant''s_id"
+
+    Security:
+        - Prevents breaking out of string literals
+        - Stops injection attacks like: ' OR '1'='1
+        - Protects against cross-tenant data access
+    """
+    if not value:
+        return value
+
+    # Escape single quotes by doubling them (standard SQL escaping)
+    # This prevents breaking out of string literals in SQL queries
+    return value.replace("'", "''")
+
+
+def escape_identifier(value: str) -> str:
+    """
+    Escape an identifier (tenant_id, app_id, run_id, etc.) for safe SQL use.
+
+    This function is for PG protocol mode (SQL syntax).
+    For SDK mode, use escape_logstore_query_value() instead.
+
+    Args:
+        value: The identifier value to escape
+
+    Returns:
+        Escaped identifier safe for use in SQL queries
+
+    Examples:
+        >>> escape_identifier("550e8400-e29b-41d4-a716-446655440000")
+        "550e8400-e29b-41d4-a716-446655440000"
+        >>> escape_identifier("tenant_id' OR '1'='1")
+        "tenant_id'' OR ''1''=''1"
+
+    Security:
+        - Prevents SQL injection via identifiers
+        - Stops cross-tenant access attempts
+        - Works for UUIDs, alphanumeric IDs, and similar identifiers
+    """
+    # For identifiers, use the same escaping as strings
+    # This is simple and effective for preventing injection
+    return escape_sql_string(value)
+
+
+def escape_logstore_query_value(value: str) -> str:
+    """
+    Escape value for LogStore query syntax (SDK mode).
+
+    LogStore query syntax rules:
+    1. Keywords (and/or/not) are case-insensitive
+    2. Single quotes are ordinary characters (no special meaning)
+    3. Double quotes wrap values: key:"value"
+    4. Backslash is the escape character:
+       - \" for double quote inside value
+       - \\ for backslash itself
+    5. Parentheses can change query structure
+
+    To prevent injection:
+    - Wrap value in double quotes to treat special chars as literals
+    - Escape backslashes and double quotes using backslash
+
+    Args:
+        value: The value to escape for LogStore query syntax
+
+    Returns:
+        Quoted and escaped value safe for LogStore query syntax (includes the quotes)
+
+    Examples:
+        >>> escape_logstore_query_value("normal_value")
+        '"normal_value"'
+        >>> escape_logstore_query_value("value or field:evil")
+        '"value or field:evil"'  # 'or' and ':' are now literals
+        >>> escape_logstore_query_value('value"test')
+        '"value\\"test"'  # Internal double quote escaped
+        >>> escape_logstore_query_value('value\\test')
+        '"value\\\\test"'  # Backslash escaped
+
+    Security:
+        - Prevents injection via and/or/not keywords
+        - Prevents injection via colons (:)
+        - Prevents injection via parentheses
+        - Protects against cross-tenant data access
+
+    Note:
+        Escape order is critical: backslash first, then double quotes.
+        Otherwise, we'd double-escape the escape character itself.
+    """
+    if not value:
+        return '""'
+
+    # IMPORTANT: Escape backslashes FIRST, then double quotes
+    # This prevents double-escaping (e.g., " -> \" -> \\" incorrectly)
+    escaped = value.replace("\\", "\\\\")  # \ -> \\
+    escaped = escaped.replace('"', '\\"')  # " -> \"
+
+    # Wrap in double quotes to treat as literal string
+    # This prevents and/or/not/:/() from being interpreted as operators
+    return f'"{escaped}"'

api/factories/file_factory.py

@@ -115,7 +115,18 @@ def build_from_mappings(
     # TODO(QuantumGhost): Performance concern - each mapping triggers a separate database query.
     # Implement batch processing to reduce database load when handling multiple files.
     # Filter out None/empty mappings to avoid errors
-    valid_mappings = [m for m in mappings if m and m.get("transfer_method")]
+    def is_valid_mapping(m: Mapping[str, Any]) -> bool:
+        if not m or not m.get("transfer_method"):
+            return False
+        # For REMOTE_URL transfer method, ensure url or remote_url is provided and not None
+        transfer_method = m.get("transfer_method")
+        if transfer_method == FileTransferMethod.REMOTE_URL:
+            url = m.get("url") or m.get("remote_url")
+            if not url:
+                return False
+        return True
+
+    valid_mappings = [m for m in mappings if is_valid_mapping(m)]
     files = [
         build_from_mapping(
             mapping=mapping,

api/factories/variable_factory.py

@@ -4,6 +4,7 @@ from uuid import uuid4
 
 from configs import dify_config
 from core.file import File
+from core.model_runtime.entities import PromptMessage
 from core.variables.exc import VariableError
 from core.variables.segments import (
     ArrayAnySegment,
@@ -11,6 +12,7 @@ from core.variables.segments import (
     ArrayFileSegment,
     ArrayNumberSegment,
     ArrayObjectSegment,
+    ArrayPromptMessageSegment,
     ArraySegment,
     ArrayStringSegment,
     BooleanSegment,
@@ -29,6 +31,7 @@ from core.variables.variables import (
     ArrayFileVariable,
     ArrayNumberVariable,
     ArrayObjectVariable,
+    ArrayPromptMessageVariable,
     ArrayStringVariable,
     BooleanVariable,
     FileVariable,
@@ -38,7 +41,7 @@ from core.variables.variables import (
     ObjectVariable,
     SecretVariable,
     StringVariable,
-    Variable,
+    VariableBase,
 )
 from core.workflow.constants import (
     CONVERSATION_VARIABLE_NODE_ID,
@@ -61,6 +64,7 @@ SEGMENT_TO_VARIABLE_MAP = {
     ArrayFileSegment: ArrayFileVariable,
     ArrayNumberSegment: ArrayNumberVariable,
     ArrayObjectSegment: ArrayObjectVariable,
+    ArrayPromptMessageSegment: ArrayPromptMessageVariable,
     ArrayStringSegment: ArrayStringVariable,
     BooleanSegment: BooleanVariable,
     FileSegment: FileVariable,
@@ -72,25 +76,25 @@ SEGMENT_TO_VARIABLE_MAP = {
 }
 
 
-def build_conversation_variable_from_mapping(mapping: Mapping[str, Any], /) -> Variable:
+def build_conversation_variable_from_mapping(mapping: Mapping[str, Any], /) -> VariableBase:
     if not mapping.get("name"):
         raise VariableError("missing name")
     return _build_variable_from_mapping(mapping=mapping, selector=[CONVERSATION_VARIABLE_NODE_ID, mapping["name"]])
 
 
-def build_environment_variable_from_mapping(mapping: Mapping[str, Any], /) -> Variable:
+def build_environment_variable_from_mapping(mapping: Mapping[str, Any], /) -> VariableBase:
     if not mapping.get("name"):
         raise VariableError("missing name")
     return _build_variable_from_mapping(mapping=mapping, selector=[ENVIRONMENT_VARIABLE_NODE_ID, mapping["name"]])
 
 
-def build_pipeline_variable_from_mapping(mapping: Mapping[str, Any], /) -> Variable:
+def build_pipeline_variable_from_mapping(mapping: Mapping[str, Any], /) -> VariableBase:
     if not mapping.get("variable"):
         raise VariableError("missing variable")
     return mapping["variable"]
 
 
-def _build_variable_from_mapping(*, mapping: Mapping[str, Any], selector: Sequence[str]) -> Variable:
+def _build_variable_from_mapping(*, mapping: Mapping[str, Any], selector: Sequence[str]) -> VariableBase:
     """
     This factory function is used to create the environment variable or the conversation variable,
     not support the File type.
@@ -100,7 +104,7 @@ def _build_variable_from_mapping(*, mapping: Mapping[str, Any], selector: Sequen
     if (value := mapping.get("value")) is None:
         raise VariableError("missing value")
 
-    result: Variable
+    result: VariableBase
     match value_type:
         case SegmentType.STRING:
             result = StringVariable.model_validate(mapping)
@@ -134,7 +138,7 @@ def _build_variable_from_mapping(*, mapping: Mapping[str, Any], selector: Sequen
         raise VariableError(f"variable size {result.size} exceeds limit {dify_config.MAX_VARIABLE_SIZE}")
     if not result.selector:
         result = result.model_copy(update={"selector": selector})
-    return cast(Variable, result)
+    return cast(VariableBase, result)
 
 
 def build_segment(value: Any, /) -> Segment:
@@ -156,7 +160,13 @@ def build_segment(value: Any, /) -> Segment:
         return ObjectSegment(value=value)
     if isinstance(value, File):
         return FileSegment(value=value)
+    if isinstance(value, PromptMessage):
+        # Single PromptMessage should be wrapped in a list
+        return ArrayPromptMessageSegment(value=[value])
     if isinstance(value, list):
+        # Check if all items are PromptMessage
+        if value and all(isinstance(item, PromptMessage) for item in value):
+            return ArrayPromptMessageSegment(value=value)
         items = [build_segment(item) for item in value]
         types = {item.value_type for item in items}
         if all(isinstance(item, ArraySegment) for item in items):
@@ -200,6 +210,7 @@ _segment_factory: Mapping[SegmentType, type[Segment]] = {
     SegmentType.ARRAY_OBJECT: ArrayObjectSegment,
     SegmentType.ARRAY_FILE: ArrayFileSegment,
     SegmentType.ARRAY_BOOLEAN: ArrayBooleanSegment,
+    SegmentType.ARRAY_PROMPT_MESSAGE: ArrayPromptMessageSegment,
 }
 
 
@@ -285,8 +296,8 @@ def segment_to_variable(
     id: str | None = None,
     name: str | None = None,
     description: str = "",
-) -> Variable:
-    if isinstance(segment, Variable):
+) -> VariableBase:
+    if isinstance(segment, VariableBase):
         return segment
     name = name or selector[-1]
     id = id or str(uuid4())
@@ -297,7 +308,7 @@ def segment_to_variable(
 
     variable_class = SEGMENT_TO_VARIABLE_MAP[segment_type]
     return cast(
-        Variable,
+        VariableBase,
         variable_class(
             id=id,
             name=name,

api/fields/message_fields.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from datetime import datetime
 from typing import TypeAlias
+from uuid import uuid4
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
@@ -20,8 +21,8 @@ class SimpleFeedback(ResponseModel):
 
 
 class RetrieverResource(ResponseModel):
-    id: str
-    message_id: str
+    id: str = Field(default_factory=lambda: str(uuid4()))
+    message_id: str = Field(default_factory=lambda: str(uuid4()))
     position: int
     dataset_id: str | None = None
     dataset_name: str | None = None

api/fields/workflow_fields.py

@@ -1,7 +1,7 @@
 from flask_restx import fields
 
 from core.helper import encrypter
-from core.variables import SecretVariable, SegmentType, Variable
+from core.variables import SecretVariable, SegmentType, VariableBase
 from fields.member_fields import simple_account_fields
 from libs.helper import TimestampField
 
@@ -21,7 +21,7 @@ class EnvironmentVariableField(fields.Raw):
                 "value_type": value.value_type.value,
                 "description": value.description,
             }
-        if isinstance(value, Variable):
+        if isinstance(value, VariableBase):
             return {
                 "id": value.id,
                 "name": value.name,

api/libs/smtp.py

@@ -3,6 +3,8 @@ import smtplib
 from email.mime.multipart import MIMEMultipart
 from email.mime.text import MIMEText
 
+from configs import dify_config
+
 logger = logging.getLogger(__name__)
 
 
@@ -19,20 +21,21 @@ class SMTPClient:
         self.opportunistic_tls = opportunistic_tls
 
     def send(self, mail: dict):
-        smtp = None
+        smtp: smtplib.SMTP | None = None
+        local_host = dify_config.SMTP_LOCAL_HOSTNAME
         try:
-            if self.use_tls:
-                if self.opportunistic_tls:
-                    smtp = smtplib.SMTP(self.server, self.port, timeout=10)
-                    # Send EHLO command with the HELO domain name as the server address
-                    smtp.ehlo(self.server)
-                    smtp.starttls()
-                    # Resend EHLO command to identify the TLS session
-                    smtp.ehlo(self.server)
-                else:
-                    smtp = smtplib.SMTP_SSL(self.server, self.port, timeout=10)
+            if self.use_tls and not self.opportunistic_tls:
+                # SMTP with SSL (implicit TLS)
+                smtp = smtplib.SMTP_SSL(self.server, self.port, timeout=10, local_hostname=local_host)
             else:
-                smtp = smtplib.SMTP(self.server, self.port, timeout=10)
+                # Plain SMTP or SMTP with STARTTLS (explicit TLS)
+                smtp = smtplib.SMTP(self.server, self.port, timeout=10, local_hostname=local_host)
+
+            assert smtp is not None
+            if self.use_tls and self.opportunistic_tls:
+                smtp.ehlo(self.server)
+                smtp.starttls()
+                smtp.ehlo(self.server)
 
             # Only authenticate if both username and password are non-empty
             if self.username and self.password and self.username.strip() and self.password.strip():

api/migrations/versions/2026_01_12_1729-3334862ee907_feat_add_created_at_id_index_to_messages.py

@@ -0,0 +1,33 @@
+"""feat: add created_at id index to messages
+
+Revision ID: 3334862ee907
+Revises: 905527cc8fd3
+Create Date: 2026-01-12 17:29:44.846544
+
+"""
+from alembic import op
+import models as models
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '3334862ee907'
+down_revision = '905527cc8fd3'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('messages', schema=None) as batch_op:
+        batch_op.create_index('message_created_at_id_idx', ['created_at', 'id'], unique=False)
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('messages', schema=None) as batch_op:
+        batch_op.drop_index('message_created_at_id_idx')
+
+    # ### end Alembic commands ###

api/models/dataset.py

@@ -1149,7 +1149,7 @@ class DatasetCollectionBinding(TypeBase):
     )
 
 
-class TidbAuthBinding(Base):
+class TidbAuthBinding(TypeBase):
     __tablename__ = "tidb_auth_bindings"
     __table_args__ = (
         sa.PrimaryKeyConstraint("id", name="tidb_auth_bindings_pkey"),
@@ -1158,7 +1158,13 @@ class TidbAuthBinding(Base):
         sa.Index("tidb_auth_bindings_created_at_idx", "created_at"),
         sa.Index("tidb_auth_bindings_status_idx", "status"),
     )
-    id: Mapped[str] = mapped_column(StringUUID, primary_key=True, default=lambda: str(uuid4()))
+    id: Mapped[str] = mapped_column(
+        StringUUID,
+        primary_key=True,
+        insert_default=lambda: str(uuid4()),
+        default_factory=lambda: str(uuid4()),
+        init=False,
+    )
     tenant_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
     cluster_id: Mapped[str] = mapped_column(String(255), nullable=False)
     cluster_name: Mapped[str] = mapped_column(String(255), nullable=False)
@@ -1166,7 +1172,9 @@ class TidbAuthBinding(Base):
     status: Mapped[str] = mapped_column(sa.String(255), nullable=False, server_default=sa.text("'CREATING'"))
     account: Mapped[str] = mapped_column(String(255), nullable=False)
     password: Mapped[str] = mapped_column(String(255), nullable=False)
-    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp())
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.current_timestamp(), init=False
+    )
 
 
 class Whitelist(TypeBase):

api/models/model.py

@@ -968,6 +968,7 @@ class Message(Base):
         Index("message_workflow_run_id_idx", "conversation_id", "workflow_run_id"),
         Index("message_created_at_idx", "created_at"),
         Index("message_app_mode_idx", "app_mode"),
+        Index("message_created_at_id_idx", "created_at", "id"),
     )
 
     id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuid4()))
@@ -1447,7 +1448,7 @@ class MessageAnnotation(Base):
         return account
 
 
-class AppAnnotationHitHistory(Base):
+class AppAnnotationHitHistory(TypeBase):
     __tablename__ = "app_annotation_hit_histories"
     __table_args__ = (
         sa.PrimaryKeyConstraint("id", name="app_annotation_hit_histories_pkey"),
@@ -1457,17 +1458,19 @@ class AppAnnotationHitHistory(Base):
         sa.Index("app_annotation_hit_histories_message_idx", "message_id"),
     )
 
-    id = mapped_column(StringUUID, default=lambda: str(uuid4()))
-    app_id = mapped_column(StringUUID, nullable=False)
+    id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuid4()), init=False)
+    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     annotation_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
-    source = mapped_column(LongText, nullable=False)
-    question = mapped_column(LongText, nullable=False)
-    account_id = mapped_column(StringUUID, nullable=False)
-    created_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
-    score = mapped_column(Float, nullable=False, server_default=sa.text("0"))
-    message_id = mapped_column(StringUUID, nullable=False)
-    annotation_question = mapped_column(LongText, nullable=False)
-    annotation_content = mapped_column(LongText, nullable=False)
+    source: Mapped[str] = mapped_column(LongText, nullable=False)
+    question: Mapped[str] = mapped_column(LongText, nullable=False)
+    account_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    created_at: Mapped[datetime] = mapped_column(
+        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
+    )
+    score: Mapped[float] = mapped_column(Float, nullable=False, server_default=sa.text("0"))
+    message_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    annotation_question: Mapped[str] = mapped_column(LongText, nullable=False)
+    annotation_content: Mapped[str] = mapped_column(LongText, nullable=False)
 
     @property
     def account(self):
@@ -2083,7 +2086,7 @@ class TraceAppConfig(TypeBase):
         }
 
 
-class TenantCreditPool(Base):
+class TenantCreditPool(TypeBase):
     __tablename__ = "tenant_credit_pools"
     __table_args__ = (
         sa.PrimaryKeyConstraint("id", name="tenant_credit_pool_pkey"),
@@ -2091,14 +2094,20 @@ class TenantCreditPool(Base):
         sa.Index("tenant_credit_pool_pool_type_idx", "pool_type"),
     )
 
-    id = mapped_column(StringUUID, primary_key=True, server_default=text("uuid_generate_v4()"))
-    tenant_id = mapped_column(StringUUID, nullable=False)
-    pool_type = mapped_column(String(40), nullable=False, default="trial", server_default="trial")
-    quota_limit = mapped_column(BigInteger, nullable=False, default=0)
-    quota_used = mapped_column(BigInteger, nullable=False, default=0)
-    created_at = mapped_column(sa.DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"))
-    updated_at = mapped_column(
-        sa.DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
+    id: Mapped[str] = mapped_column(StringUUID, primary_key=True, server_default=text("uuid_generate_v4()"), init=False)
+    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    pool_type: Mapped[str] = mapped_column(String(40), nullable=False, default="trial", server_default="trial")
+    quota_limit: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    quota_used: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    created_at: Mapped[datetime] = mapped_column(
+        sa.DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"), init=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        sa.DateTime,
+        nullable=False,
+        server_default=func.current_timestamp(),
+        onupdate=func.current_timestamp(),
+        init=False,
     )
 
     @property

api/models/workflow.py

@@ -1,11 +1,9 @@
-from __future__ import annotations
-
 import json
 import logging
 from collections.abc import Generator, Mapping, Sequence
 from datetime import datetime
 from enum import StrEnum
-from typing import TYPE_CHECKING, Any, Union, cast
+from typing import TYPE_CHECKING, Any, Optional, Union, cast
 from uuid import uuid4
 
 import sqlalchemy as sa
@@ -46,7 +44,7 @@ if TYPE_CHECKING:
 
 from constants import DEFAULT_FILE_NUMBER_LIMITS, HIDDEN_VALUE
 from core.helper import encrypter
-from core.variables import SecretVariable, Segment, SegmentType, Variable
+from core.variables import SecretVariable, Segment, SegmentType, VariableBase
 from factories import variable_factory
 from libs import helper
 
@@ -69,7 +67,7 @@ class WorkflowType(StrEnum):
     RAG_PIPELINE = "rag-pipeline"
 
     @classmethod
-    def value_of(cls, value: str) -> WorkflowType:
+    def value_of(cls, value: str) -> "WorkflowType":
         """
         Get value of given mode.
 
@@ -82,7 +80,7 @@ class WorkflowType(StrEnum):
         raise ValueError(f"invalid workflow type value {value}")
 
     @classmethod
-    def from_app_mode(cls, app_mode: Union[str, AppMode]) -> WorkflowType:
+    def from_app_mode(cls, app_mode: Union[str, "AppMode"]) -> "WorkflowType":
         """
         Get workflow type from app mode.
 
@@ -178,12 +176,12 @@ class Workflow(Base):  # bug
         graph: str,
         features: str,
         created_by: str,
-        environment_variables: Sequence[Variable],
-        conversation_variables: Sequence[Variable],
+        environment_variables: Sequence[VariableBase],
+        conversation_variables: Sequence[VariableBase],
         rag_pipeline_variables: list[dict],
         marked_name: str = "",
         marked_comment: str = "",
-    ) -> Workflow:
+    ) -> "Workflow":
         workflow = Workflow()
         workflow.id = str(uuid4())
         workflow.tenant_id = tenant_id
@@ -447,7 +445,7 @@ class Workflow(Base):  # bug
 
         # decrypt secret variables value
         def decrypt_func(
-            var: Variable,
+            var: VariableBase,
         ) -> StringVariable | IntegerVariable | FloatVariable | SecretVariable:
             if isinstance(var, SecretVariable):
                 return var.model_copy(update={"value": encrypter.decrypt_token(tenant_id=tenant_id, token=var.value)})
@@ -463,7 +461,7 @@ class Workflow(Base):  # bug
         return decrypted_results
 
     @environment_variables.setter
-    def environment_variables(self, value: Sequence[Variable]):
+    def environment_variables(self, value: Sequence[VariableBase]):
         if not value:
             self._environment_variables = "{}"
             return
@@ -487,7 +485,7 @@ class Workflow(Base):  # bug
                 value[i] = origin_variables_dictionary[variable.id].model_copy(update={"name": variable.name})
 
         # encrypt secret variables value
-        def encrypt_func(var: Variable) -> Variable:
+        def encrypt_func(var: VariableBase) -> VariableBase:
             if isinstance(var, SecretVariable):
                 return var.model_copy(update={"value": encrypter.encrypt_token(tenant_id=tenant_id, token=var.value)})
             else:
@@ -517,7 +515,7 @@ class Workflow(Base):  # bug
         return result
 
     @property
-    def conversation_variables(self) -> Sequence[Variable]:
+    def conversation_variables(self) -> Sequence[VariableBase]:
         # TODO: find some way to init `self._conversation_variables` when instance created.
         if self._conversation_variables is None:
             self._conversation_variables = "{}"
@@ -527,7 +525,7 @@ class Workflow(Base):  # bug
         return results
 
     @conversation_variables.setter
-    def conversation_variables(self, value: Sequence[Variable]):
+    def conversation_variables(self, value: Sequence[VariableBase]):
         self._conversation_variables = json.dumps(
             {var.name: var.model_dump() for var in value},
             ensure_ascii=False,
@@ -622,7 +620,7 @@ class WorkflowRun(Base):
     finished_at: Mapped[datetime | None] = mapped_column(DateTime)
     exceptions_count: Mapped[int] = mapped_column(sa.Integer, server_default=sa.text("0"), nullable=True)
 
-    pause: Mapped[WorkflowPause | None] = orm.relationship(
+    pause: Mapped[Optional["WorkflowPause"]] = orm.relationship(
         "WorkflowPause",
         primaryjoin="WorkflowRun.id == foreign(WorkflowPause.workflow_run_id)",
         uselist=False,
@@ -692,7 +690,7 @@ class WorkflowRun(Base):
         }
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> WorkflowRun:
+    def from_dict(cls, data: dict[str, Any]) -> "WorkflowRun":
         return cls(
             id=data.get("id"),
             tenant_id=data.get("tenant_id"),
@@ -844,7 +842,7 @@ class WorkflowNodeExecutionModel(Base):  # This model is expected to have `offlo
     created_by: Mapped[str] = mapped_column(StringUUID)
     finished_at: Mapped[datetime | None] = mapped_column(DateTime)
 
-    offload_data: Mapped[list[WorkflowNodeExecutionOffload]] = orm.relationship(
+    offload_data: Mapped[list["WorkflowNodeExecutionOffload"]] = orm.relationship(
         "WorkflowNodeExecutionOffload",
         primaryjoin="WorkflowNodeExecutionModel.id == foreign(WorkflowNodeExecutionOffload.node_execution_id)",
         uselist=True,
@@ -854,13 +852,13 @@ class WorkflowNodeExecutionModel(Base):  # This model is expected to have `offlo
 
     @staticmethod
     def preload_offload_data(
-        query: Select[tuple[WorkflowNodeExecutionModel]] | orm.Query[WorkflowNodeExecutionModel],
+        query: Select[tuple["WorkflowNodeExecutionModel"]] | orm.Query["WorkflowNodeExecutionModel"],
     ):
         return query.options(orm.selectinload(WorkflowNodeExecutionModel.offload_data))
 
     @staticmethod
     def preload_offload_data_and_files(
-        query: Select[tuple[WorkflowNodeExecutionModel]] | orm.Query[WorkflowNodeExecutionModel],
+        query: Select[tuple["WorkflowNodeExecutionModel"]] | orm.Query["WorkflowNodeExecutionModel"],
     ):
         return query.options(
             orm.selectinload(WorkflowNodeExecutionModel.offload_data).options(
@@ -935,7 +933,7 @@ class WorkflowNodeExecutionModel(Base):  # This model is expected to have `offlo
                     )
         return extras
 
-    def _get_offload_by_type(self, type_: ExecutionOffLoadType) -> WorkflowNodeExecutionOffload | None:
+    def _get_offload_by_type(self, type_: ExecutionOffLoadType) -> Optional["WorkflowNodeExecutionOffload"]:
         return next(iter([i for i in self.offload_data if i.type_ == type_]), None)
 
     @property
@@ -1049,7 +1047,7 @@ class WorkflowNodeExecutionOffload(Base):
         back_populates="offload_data",
     )
 
-    file: Mapped[UploadFile | None] = orm.relationship(
+    file: Mapped[Optional["UploadFile"]] = orm.relationship(
         foreign_keys=[file_id],
         lazy="raise",
         uselist=False,
@@ -1067,7 +1065,7 @@ class WorkflowAppLogCreatedFrom(StrEnum):
     INSTALLED_APP = "installed-app"
 
     @classmethod
-    def value_of(cls, value: str) -> WorkflowAppLogCreatedFrom:
+    def value_of(cls, value: str) -> "WorkflowAppLogCreatedFrom":
         """
         Get value of given mode.
 
@@ -1184,7 +1182,7 @@ class ConversationVariable(TypeBase):
     )
 
     @classmethod
-    def from_variable(cls, *, app_id: str, conversation_id: str, variable: Variable) -> ConversationVariable:
+    def from_variable(cls, *, app_id: str, conversation_id: str, variable: VariableBase) -> "ConversationVariable":
         obj = cls(
             id=variable.id,
             app_id=app_id,
@@ -1193,7 +1191,7 @@ class ConversationVariable(TypeBase):
         )
         return obj
 
-    def to_variable(self) -> Variable:
+    def to_variable(self) -> VariableBase:
         mapping = json.loads(self.data)
         return variable_factory.build_conversation_variable_from_mapping(mapping)
 
@@ -1291,7 +1289,7 @@ class WorkflowDraftVariable(Base):
     # which may differ from the original value's type. Typically, they are the same,
     # but in cases where the structurally truncated  value still exceeds the size limit,
     # text slicing is applied, and the `value_type` is converted to `STRING`.
-    value_type: Mapped[SegmentType] = mapped_column(EnumText(SegmentType, length=20))
+    value_type: Mapped[SegmentType] = mapped_column(EnumText(SegmentType, length=21))
 
     # The variable's value serialized as a JSON string
     #
@@ -1337,7 +1335,7 @@ class WorkflowDraftVariable(Base):
     )
 
     # Relationship to WorkflowDraftVariableFile
-    variable_file: Mapped[WorkflowDraftVariableFile | None] = orm.relationship(
+    variable_file: Mapped[Optional["WorkflowDraftVariableFile"]] = orm.relationship(
         foreign_keys=[file_id],
         lazy="raise",
         uselist=False,
@@ -1507,7 +1505,7 @@ class WorkflowDraftVariable(Base):
         node_execution_id: str | None,
         description: str = "",
         file_id: str | None = None,
-    ) -> WorkflowDraftVariable:
+    ) -> "WorkflowDraftVariable":
         variable = WorkflowDraftVariable()
         variable.id = str(uuid4())
         variable.created_at = naive_utc_now()
@@ -1530,7 +1528,7 @@ class WorkflowDraftVariable(Base):
         name: str,
         value: Segment,
         description: str = "",
-    ) -> WorkflowDraftVariable:
+    ) -> "WorkflowDraftVariable":
         variable = cls._new(
             app_id=app_id,
             node_id=CONVERSATION_VARIABLE_NODE_ID,
@@ -1551,7 +1549,7 @@ class WorkflowDraftVariable(Base):
         value: Segment,
         node_execution_id: str,
         editable: bool = False,
-    ) -> WorkflowDraftVariable:
+    ) -> "WorkflowDraftVariable":
         variable = cls._new(
             app_id=app_id,
             node_id=SYSTEM_VARIABLE_NODE_ID,
@@ -1574,7 +1572,7 @@ class WorkflowDraftVariable(Base):
         visible: bool = True,
         editable: bool = True,
         file_id: str | None = None,
-    ) -> WorkflowDraftVariable:
+    ) -> "WorkflowDraftVariable":
         variable = cls._new(
             app_id=app_id,
             node_id=node_id,
@@ -1665,12 +1663,12 @@ class WorkflowDraftVariableFile(Base):
 
     # The `value_type` field records the type of the original value.
     value_type: Mapped[SegmentType] = mapped_column(
-        EnumText(SegmentType, length=20),
+        EnumText(SegmentType, length=21),
         nullable=False,
     )
 
     # Relationship to UploadFile
-    upload_file: Mapped[UploadFile] = orm.relationship(
+    upload_file: Mapped["UploadFile"] = orm.relationship(
         foreign_keys=[upload_file_id],
         lazy="raise",
         uselist=False,
@@ -1737,7 +1735,7 @@ class WorkflowPause(DefaultFieldsMixin, Base):
     state_object_key: Mapped[str] = mapped_column(String(length=255), nullable=False)
 
     # Relationship to WorkflowRun
-    workflow_run: Mapped[WorkflowRun] = orm.relationship(
+    workflow_run: Mapped["WorkflowRun"] = orm.relationship(
         foreign_keys=[workflow_run_id],
         # require explicit preloading.
         lazy="raise",
@@ -1793,7 +1791,7 @@ class WorkflowPauseReason(DefaultFieldsMixin, Base):
     )
 
     @classmethod
-    def from_entity(cls, pause_reason: PauseReason) -> WorkflowPauseReason:
+    def from_entity(cls, pause_reason: PauseReason) -> "WorkflowPauseReason":
         if isinstance(pause_reason, HumanInputRequired):
             return cls(
                 type_=PauseReasonType.HUMAN_INPUT_REQUIRED, form_id=pause_reason.form_id, node_id=pause_reason.node_id

api/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "dify-api"
-version = "1.11.3"
+version = "1.11.4"
 requires-python = ">=3.11,<3.13"
 
 dependencies = [