log expiration_date=-1

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

.github/workflows/style.yml

@@ -65,6 +65,9 @@ jobs:
     defaults:
       run:
         working-directory: ./web
+    permissions:
+      checks: write
+      pull-requests: read
 
     steps:
       - name: Checkout code
@@ -103,7 +106,15 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         working-directory: ./web
         run: |
-          pnpm run lint
+          pnpm run lint:report
+        continue-on-error: true
+
+      # - name: Annotate Code
+      #   if: steps.changed-files.outputs.any_changed == 'true' && github.event_name == 'pull_request'
+      #   uses: DerLev/eslint-annotations@51347b3a0abfb503fc8734d5ae31c4b151297fae
+      #   with:
+      #     eslint-report: web/eslint_report.json
+      #     github-token: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Web type check
         if: steps.changed-files.outputs.any_changed == 'true'

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
 
@@ -878,12 +881,25 @@ def clear_free_plan_tenant_expired_logs(days: int, batch: int, tenant_ids: list[
     is_flag=True,
     help="Preview cleanup results without deleting any workflow run data.",
 )
+@click.option(
+    "--log-sql",
+    is_flag=True,
+    help="Log SQL statements and timings for cleanup queries.",
+)
+@click.option(
+    "--log-sql-min-ms",
+    default=0,
+    show_default=True,
+    help="Only log SQL statements slower than N milliseconds (0 logs all).",
+)
 def clean_workflow_runs(
     days: int,
     batch_size: int,
     start_from: datetime.datetime | None,
     end_before: datetime.datetime | None,
     dry_run: bool,
+    log_sql: bool,
+    log_sql_min_ms: int,
 ):
     """
     Clean workflow runs and related workflow data for free tenants.
@@ -900,6 +916,8 @@ def clean_workflow_runs(
         start_from=start_from,
         end_before=end_before,
         dry_run=dry_run,
+        log_sql=log_sql,
+        log_sql_min_ms=log_sql_min_ms,
     ).run()
 
     end_time = datetime.datetime.now(datetime.UTC)
@@ -2172,3 +2190,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/core/model_manager.py

@@ -21,7 +21,6 @@ from core.model_runtime.model_providers.__base.speech2text_model import Speech2T
 from core.model_runtime.model_providers.__base.text_embedding_model import TextEmbeddingModel
 from core.model_runtime.model_providers.__base.tts_model import TTSModel
 from core.provider_manager import ProviderManager
-from core.workflow.utils.generator_timeout import with_first_token_timeout
 from extensions.ext_redis import redis_client
 from models.provider import ProviderType
 from services.enterprise.plugin_manager_service import PluginCredentialType
@@ -110,7 +109,6 @@ class ModelInstance:
         stream: Literal[True] = True,
         user: str | None = None,
         callbacks: list[Callback] | None = None,
-        first_token_timeout: float | None = None,
     ) -> Generator: ...
 
     @overload
@@ -123,7 +121,6 @@ class ModelInstance:
         stream: Literal[False] = False,
         user: str | None = None,
         callbacks: list[Callback] | None = None,
-        first_token_timeout: float | None = None,
     ) -> LLMResult: ...
 
     @overload
@@ -136,7 +133,6 @@ class ModelInstance:
         stream: bool = True,
         user: str | None = None,
         callbacks: list[Callback] | None = None,
-        first_token_timeout: float | None = None,
     ) -> Union[LLMResult, Generator]: ...
 
     def invoke_llm(
@@ -148,7 +144,6 @@ class ModelInstance:
         stream: bool = True,
         user: str | None = None,
         callbacks: list[Callback] | None = None,
-        first_token_timeout: float | None = None,
     ) -> Union[LLMResult, Generator]:
         """
         Invoke large language model
@@ -160,31 +155,26 @@ class ModelInstance:
         :param stream: is stream response
         :param user: unique user id
         :param callbacks: callbacks
-        :param first_token_timeout: timeout in seconds for receiving first token (streaming only)
         :return: full response or stream response chunk generator result
         """
         if not isinstance(self.model_type_instance, LargeLanguageModel):
             raise Exception("Model type instance is not LargeLanguageModel")
-
-        result = self._round_robin_invoke(
-            function=self.model_type_instance.invoke,
-            model=self.model,
-            credentials=self.credentials,
-            prompt_messages=prompt_messages,
-            model_parameters=model_parameters,
-            tools=tools,
-            stop=stop,
-            stream=stream,
-            user=user,
-            callbacks=callbacks,
+        return cast(
+            Union[LLMResult, Generator],
+            self._round_robin_invoke(
+                function=self.model_type_instance.invoke,
+                model=self.model,
+                credentials=self.credentials,
+                prompt_messages=prompt_messages,
+                model_parameters=model_parameters,
+                tools=tools,
+                stop=stop,
+                stream=stream,
+                user=user,
+                callbacks=callbacks,
+            ),
         )
 
-        # Apply first token timeout wrapper for streaming responses
-        if stream and first_token_timeout and first_token_timeout > 0 and isinstance(result, Generator):
-            result = with_first_token_timeout(result, first_token_timeout)
-
-        return cast(Union[LLMResult, Generator], result)
-
     def get_llm_num_tokens(
         self, prompt_messages: Sequence[PromptMessage], tools: Sequence[PromptMessageTool] | None = None
     ) -> int:

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

@@ -23,22 +23,10 @@ class RetryConfig(BaseModel):
     retry_interval: int = 0  # retry interval in milliseconds
     retry_enabled: bool = False  # whether retry is enabled
 
-    # First token timeout for LLM nodes (milliseconds), 0 means no timeout
-    first_token_timeout: int = 0
-
-    @property
-    def first_token_timeout_seconds(self) -> float:
-        return self.first_token_timeout / 1000
-
     @property
     def retry_interval_seconds(self) -> float:
         return self.retry_interval / 1000
 
-    @property
-    def has_first_token_timeout(self) -> bool:
-        """Check if first token timeout should be applied (retry enabled and timeout > 0)."""
-        return self.retry_enabled and self.first_token_timeout > 0
-
 
 class VariableSelector(BaseModel):
     """

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

@@ -237,13 +237,6 @@ class LLMNode(Node[LLMNodeData]):
             )
 
             # handle invoke result
-            # Get first token timeout from retry config if enabled (convert ms to seconds)
-            first_token_timeout = (
-                self.node_data.retry_config.first_token_timeout_seconds
-                if self.node_data.retry_config.has_first_token_timeout
-                else None
-            )
-
             generator = LLMNode.invoke_llm(
                 node_data_model=self.node_data.model,
                 model_instance=model_instance,
@@ -257,7 +250,6 @@ class LLMNode(Node[LLMNodeData]):
                 node_id=self._node_id,
                 node_type=self.node_type,
                 reasoning_format=self.node_data.reasoning_format,
-                first_token_timeout=first_token_timeout,
             )
 
             structured_output: LLMStructuredOutput | None = None
@@ -375,7 +367,6 @@ class LLMNode(Node[LLMNodeData]):
         node_id: str,
         node_type: NodeType,
         reasoning_format: Literal["separated", "tagged"] = "tagged",
-        first_token_timeout: float | None = None,
     ) -> Generator[NodeEventBase | LLMStructuredOutput, None, None]:
         model_schema = model_instance.model_type_instance.get_model_schema(
             node_data_model.name, model_instance.credentials
@@ -409,7 +400,6 @@ class LLMNode(Node[LLMNodeData]):
                 stop=list(stop or []),
                 stream=True,
                 user=user_id,
-                first_token_timeout=first_token_timeout,
             )
 
         return LLMNode.handle_invoke_result(

api/core/workflow/utils/generator_timeout.py

@@ -1,56 +0,0 @@
-"""
-Generator timeout utilities for workflow nodes.
-
-Provides timeout wrappers for streaming generators, primarily used for
-LLM response streaming where we need to enforce time-to-first-token limits.
-"""
-
-import time
-from collections.abc import Generator
-from typing import TypeVar
-
-T = TypeVar("T")
-
-
-class FirstTokenTimeoutError(Exception):
-    """Raised when a generator fails to yield its first item within the configured timeout."""
-
-    def __init__(self, timeout_ms: int):
-        self.timeout_ms = timeout_ms
-        super().__init__(f"Generator timed out after {timeout_ms}ms without yielding first item")
-
-
-def with_first_token_timeout(
-    generator: Generator[T, None, None],
-    timeout_seconds: float,
-) -> Generator[T, None, None]:
-    """
-    Wrap a generator with first token timeout monitoring.
-
-    Only monitors the time until the FIRST item is yielded.
-    Once the first item arrives, timeout monitoring stops and
-    subsequent items are yielded without timeout checks.
-
-    Args:
-        generator: The source generator to wrap
-        timeout_seconds: Maximum time to wait for first item (in seconds)
-
-    Yields:
-        Items from the source generator
-
-    Raises:
-        FirstTokenTimeoutError: If first item doesn't arrive within timeout
-    """
-    start_time = time.monotonic()
-
-    # Handle first item separately to check timeout only once
-    try:
-        first_item = next(generator)
-        if time.monotonic() - start_time > timeout_seconds:
-            raise FirstTokenTimeoutError(int(timeout_seconds * 1000))
-        yield first_item
-    except StopIteration:
-        return
-
-    # Yield remaining items without timeout checks
-    yield from generator

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

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 = [

api/repositories/sqlalchemy_api_workflow_node_execution_repository.py

@@ -9,7 +9,7 @@ from collections.abc import Sequence
 from datetime import datetime
 from typing import TypedDict, cast
 
-from sqlalchemy import asc, delete, desc, func, select, tuple_
+from sqlalchemy import asc, delete, desc, func, select
 from sqlalchemy.engine import CursorResult
 from sqlalchemy.orm import Session, sessionmaker
 
@@ -328,39 +328,14 @@ class DifyAPISQLAlchemyWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecut
         """
         Delete node executions (and offloads) for the given workflow runs using indexed columns.
 
-        Uses the composite index on (tenant_id, app_id, workflow_id, triggered_from, workflow_run_id)
-        by filtering on those columns with tuple IN.
+        Uses the workflow_run_id index to target executions by run id.
         """
         if not runs:
             return 0, 0
 
-        tuple_values = [
-            (
-                run["tenant_id"],
-                run["app_id"],
-                run["workflow_id"],
-                DifyAPISQLAlchemyWorkflowNodeExecutionRepository._map_run_triggered_from_to_node_triggered_from(
-                    run["triggered_from"]
-                ),
-                run["run_id"],
-            )
-            for run in runs
-        ]
-
-        node_execution_ids = session.scalars(
-            select(WorkflowNodeExecutionModel.id).where(
-                tuple_(
-                    WorkflowNodeExecutionModel.tenant_id,
-                    WorkflowNodeExecutionModel.app_id,
-                    WorkflowNodeExecutionModel.workflow_id,
-                    WorkflowNodeExecutionModel.triggered_from,
-                    WorkflowNodeExecutionModel.workflow_run_id,
-                ).in_(tuple_values)
-            )
-        ).all()
-
-        if not node_execution_ids:
-            return 0, 0
+        run_ids = [run["run_id"] for run in runs]
+        run_id_filter = WorkflowNodeExecutionModel.workflow_run_id.in_(run_ids)
+        node_execution_ids = select(WorkflowNodeExecutionModel.id).where(run_id_filter)
 
         offloads_deleted = (
             cast(
@@ -378,7 +353,7 @@ class DifyAPISQLAlchemyWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecut
             cast(
                 CursorResult,
                 session.execute(
-                    delete(WorkflowNodeExecutionModel).where(WorkflowNodeExecutionModel.id.in_(node_execution_ids))
+                    delete(WorkflowNodeExecutionModel).where(run_id_filter)
                 ),
             ).rowcount
             or 0
@@ -394,38 +369,18 @@ class DifyAPISQLAlchemyWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecut
         if not runs:
             return 0, 0
 
-        tuple_values = [
-            (
-                run["tenant_id"],
-                run["app_id"],
-                run["workflow_id"],
-                DifyAPISQLAlchemyWorkflowNodeExecutionRepository._map_run_triggered_from_to_node_triggered_from(
-                    run["triggered_from"]
-                ),
-                run["run_id"],
-            )
-            for run in runs
-        ]
-        tuple_filter = tuple_(
-            WorkflowNodeExecutionModel.tenant_id,
-            WorkflowNodeExecutionModel.app_id,
-            WorkflowNodeExecutionModel.workflow_id,
-            WorkflowNodeExecutionModel.triggered_from,
-            WorkflowNodeExecutionModel.workflow_run_id,
-        ).in_(tuple_values)
+        run_ids = [run["run_id"] for run in runs]
+        run_id_filter = WorkflowNodeExecutionModel.workflow_run_id.in_(run_ids)
 
         node_executions_count = (
-            session.scalar(select(func.count()).select_from(WorkflowNodeExecutionModel).where(tuple_filter)) or 0
+            session.scalar(select(func.count()).select_from(WorkflowNodeExecutionModel).where(run_id_filter)) or 0
         )
+        node_execution_ids = select(WorkflowNodeExecutionModel.id).where(run_id_filter)
         offloads_count = (
             session.scalar(
                 select(func.count())
                 .select_from(WorkflowNodeExecutionOffload)
-                .join(
-                    WorkflowNodeExecutionModel,
-                    WorkflowNodeExecutionOffload.node_execution_id == WorkflowNodeExecutionModel.id,
-                )
-                .where(tuple_filter)
+                .where(WorkflowNodeExecutionOffload.node_execution_id.in_(node_execution_ids))
             )
             or 0
         )

api/schedule/clean_messages.py

@@ -1,90 +1,62 @@
-import datetime
 import logging
 import time
 
 import click
-from sqlalchemy.exc import SQLAlchemyError
 
 import app
 from configs import dify_config
-from enums.cloud_plan import CloudPlan
-from extensions.ext_database import db
-from extensions.ext_redis import redis_client
-from models.model import (
-    App,
-    Message,
-    MessageAgentThought,
-    MessageAnnotation,
-    MessageChain,
-    MessageFeedback,
-    MessageFile,
-)
-from models.web import SavedMessage
-from services.feature_service import FeatureService
+from services.retention.conversation.messages_clean_policy import create_message_clean_policy
+from services.retention.conversation.messages_clean_service import MessagesCleanService
 
 logger = logging.getLogger(__name__)
 
 
-@app.celery.task(queue="dataset")
+@app.celery.task(queue="retention")
 def clean_messages():
-    click.echo(click.style("Start clean messages.", fg="green"))
-    start_at = time.perf_counter()
-    plan_sandbox_clean_message_day = datetime.datetime.now() - datetime.timedelta(
-        days=dify_config.PLAN_SANDBOX_CLEAN_MESSAGE_DAY_SETTING
-    )
-    while True:
-        try:
-            # Main query with join and filter
-            messages = (
-                db.session.query(Message)
-                .where(Message.created_at < plan_sandbox_clean_message_day)
-                .order_by(Message.created_at.desc())
-                .limit(100)
-                .all()
-            )
+    """
+    Clean expired messages based on clean policy.
 
-        except SQLAlchemyError:
-            raise
-        if not messages:
-            break
-        for message in messages:
-            app = db.session.query(App).filter_by(id=message.app_id).first()
-            if not app:
-                logger.warning(
-                    "Expected App record to exist, but none was found, app_id=%s, message_id=%s",
-                    message.app_id,
-                    message.id,
-                )
-                continue
-            features_cache_key = f"features:{app.tenant_id}"
-            plan_cache = redis_client.get(features_cache_key)
-            if plan_cache is None:
-                features = FeatureService.get_features(app.tenant_id)
-                redis_client.setex(features_cache_key, 600, features.billing.subscription.plan)
-                plan = features.billing.subscription.plan
-            else:
-                plan = plan_cache.decode()
-            if plan == CloudPlan.SANDBOX:
-                # clean related message
-                db.session.query(MessageFeedback).where(MessageFeedback.message_id == message.id).delete(
-                    synchronize_session=False
-                )
-                db.session.query(MessageAnnotation).where(MessageAnnotation.message_id == message.id).delete(
-                    synchronize_session=False
-                )
-                db.session.query(MessageChain).where(MessageChain.message_id == message.id).delete(
-                    synchronize_session=False
-                )
-                db.session.query(MessageAgentThought).where(MessageAgentThought.message_id == message.id).delete(
-                    synchronize_session=False
-                )
-                db.session.query(MessageFile).where(MessageFile.message_id == message.id).delete(
-                    synchronize_session=False
-                )
-                db.session.query(SavedMessage).where(SavedMessage.message_id == message.id).delete(
-                    synchronize_session=False
-                )
-                db.session.query(Message).where(Message.id == message.id).delete()
-                db.session.commit()
-    end_at = time.perf_counter()
-    click.echo(click.style(f"Cleaned messages from db success latency: {end_at - start_at}", fg="green"))
+    This task uses MessagesCleanService to efficiently clean messages in batches.
+    The behavior depends on BILLING_ENABLED configuration:
+    - BILLING_ENABLED=True: only delete messages from sandbox tenants (with whitelist/grace period)
+    - BILLING_ENABLED=False: delete all messages within the time range
+    """
+    click.echo(click.style("clean_messages: start clean messages.", fg="green"))
+    start_at = time.perf_counter()
+
+    try:
+        # Create policy based on billing configuration
+        policy = create_message_clean_policy(
+            graceful_period_days=dify_config.SANDBOX_EXPIRED_RECORDS_CLEAN_GRACEFUL_PERIOD,
+        )
+
+        # Create and run the cleanup service
+        service = MessagesCleanService.from_days(
+            policy=policy,
+            days=dify_config.SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS,
+            batch_size=dify_config.SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_SIZE,
+        )
+        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

api/services/billing_service.py

@@ -42,6 +42,7 @@ class BillingService:
         params = {"tenant_id": tenant_id}
 
         billing_info = cls._send_request("GET", "/subscription/info", params=params)
+        cls._log_expiration_date_if_missing(tenant_id, billing_info, "get_info")
         return billing_info
 
     @classmethod
@@ -281,6 +282,7 @@ class BillingService:
                     try:
                         subscription_plan = subscription_adapter.validate_python(plan)
                         results[tenant_id] = subscription_plan
+                        cls._log_expiration_date_if_missing(tenant_id, subscription_plan, "get_plan_bulk")
                     except Exception:
                         logger.exception(
                             "get_plan_bulk: failed to validate subscription plan for tenant(%s)", tenant_id
@@ -296,6 +298,21 @@ class BillingService:
     def _make_plan_cache_key(cls, tenant_id: str) -> str:
         return f"{cls._PLAN_CACHE_KEY_PREFIX}{tenant_id}"
 
+    @staticmethod
+    def _log_expiration_date_if_missing(tenant_id: str, payload: dict, source: str) -> None:
+        expiration_date = None
+        if isinstance(payload, dict):
+            if "expiration_date" in payload:
+                expiration_date = payload.get("expiration_date")
+            elif isinstance(payload.get("subscription"), dict):
+                expiration_date = payload["subscription"].get("expiration_date")
+        if expiration_date == -1:
+            logger.warning(
+                "billing %s: tenant %s returned expiration_date=-1",
+                source,
+                tenant_id,
+            )
+
     @classmethod
     def get_plan_bulk_with_cache(cls, tenant_ids: Sequence[str]) -> dict[str, SubscriptionPlan]:
         """

api/services/enterprise/workspace_sync.py

@@ -0,0 +1,58 @@
+import json
+import logging
+import uuid
+from datetime import UTC, datetime
+
+from redis import RedisError
+
+from extensions.ext_redis import redis_client
+
+logger = logging.getLogger(__name__)
+
+WORKSPACE_SYNC_QUEUE = "enterprise:workspace:sync:queue"
+WORKSPACE_SYNC_PROCESSING = "enterprise:workspace:sync:processing"
+
+
+class WorkspaceSyncService:
+    """Service to publish workspace sync tasks to Redis queue for enterprise backend consumption"""
+
+    @staticmethod
+    def queue_credential_sync(workspace_id: str, *, source: str) -> bool:
+        """
+        Queue a credential sync task for a newly created workspace.
+
+        This publishes a task to Redis that will be consumed by the enterprise backend
+        worker to sync credentials with the plugin-manager.
+
+        Args:
+            workspace_id: The workspace/tenant ID to sync credentials for
+            source: Source of the sync request (for debugging/tracking)
+
+        Returns:
+            bool: True if task was queued successfully, False otherwise
+        """
+        try:
+            task = {
+                "task_id": str(uuid.uuid4()),
+                "workspace_id": workspace_id,
+                "retry_count": 0,
+                "created_at": datetime.now(UTC).isoformat(),
+                "source": source,
+            }
+
+            # Push to Redis list (queue) - LPUSH adds to the head, worker consumes from tail with RPOP
+            redis_client.lpush(WORKSPACE_SYNC_QUEUE, json.dumps(task))
+
+            logger.info(
+                "Queued credential sync task for workspace %s, task_id: %s, source: %s",
+                workspace_id,
+                task["task_id"],
+                source,
+            )
+            return True
+
+        except (RedisError, TypeError) as e:
+            logger.error("Failed to queue credential sync for workspace %s: %s", workspace_id, str(e), exc_info=True)
+            # Don't raise - we don't want to fail workspace creation if queueing fails
+            # The scheduled task will catch it later
+            return False

api/services/retention/conversation/messages_clean_policy.py

@@ -0,0 +1,216 @@
+import datetime
+import logging
+from abc import ABC, abstractmethod
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass
+
+from configs import dify_config
+from enums.cloud_plan import CloudPlan
+from services.billing_service import BillingService, SubscriptionPlan
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class SimpleMessage:
+    id: str
+    app_id: str
+    created_at: datetime.datetime
+
+
+class MessagesCleanPolicy(ABC):
+    """
+    Abstract base class for message cleanup policies.
+
+    A policy determines which messages from a batch should be deleted.
+    """
+
+    @abstractmethod
+    def filter_message_ids(
+        self,
+        messages: Sequence[SimpleMessage],
+        app_to_tenant: dict[str, str],
+    ) -> Sequence[str]:
+        """
+        Filter messages and return IDs of messages that should be deleted.
+
+        Args:
+            messages: Batch of messages to evaluate
+            app_to_tenant: Mapping from app_id to tenant_id
+
+        Returns:
+            List of message IDs that should be deleted
+        """
+        ...
+
+
+class BillingDisabledPolicy(MessagesCleanPolicy):
+    """
+    Policy for community or enterpriseedition (billing disabled).
+
+    No special filter logic, just return all message ids.
+    """
+
+    def filter_message_ids(
+        self,
+        messages: Sequence[SimpleMessage],
+        app_to_tenant: dict[str, str],
+    ) -> Sequence[str]:
+        return [msg.id for msg in messages]
+
+
+class BillingSandboxPolicy(MessagesCleanPolicy):
+    """
+    Policy for sandbox plan tenants in cloud edition (billing enabled).
+
+    Filters messages based on sandbox plan expiration rules:
+    - Skip tenants in the whitelist
+    - Only delete messages from sandbox plan tenants
+    - Respect grace period after subscription expiration
+    - Safe default: if tenant mapping or plan is missing, do NOT delete
+    """
+
+    def __init__(
+        self,
+        plan_provider: Callable[[Sequence[str]], dict[str, SubscriptionPlan]],
+        graceful_period_days: int = 21,
+        tenant_whitelist: Sequence[str] | None = None,
+        current_timestamp: int | None = None,
+    ) -> None:
+        self._graceful_period_days = graceful_period_days
+        self._tenant_whitelist: Sequence[str] = tenant_whitelist or []
+        self._plan_provider = plan_provider
+        self._current_timestamp = current_timestamp
+
+    def filter_message_ids(
+        self,
+        messages: Sequence[SimpleMessage],
+        app_to_tenant: dict[str, str],
+    ) -> Sequence[str]:
+        """
+        Filter messages based on sandbox plan expiration rules.
+
+        Args:
+            messages: Batch of messages to evaluate
+            app_to_tenant: Mapping from app_id to tenant_id
+
+        Returns:
+            List of message IDs that should be deleted
+        """
+        if not messages or not app_to_tenant:
+            return []
+
+        # Get unique tenant_ids and fetch subscription plans
+        tenant_ids = list(set(app_to_tenant.values()))
+        tenant_plans = self._plan_provider(tenant_ids)
+
+        if not tenant_plans:
+            return []
+
+        # Apply sandbox deletion rules
+        return self._filter_expired_sandbox_messages(
+            messages=messages,
+            app_to_tenant=app_to_tenant,
+            tenant_plans=tenant_plans,
+        )
+
+    def _filter_expired_sandbox_messages(
+        self,
+        messages: Sequence[SimpleMessage],
+        app_to_tenant: dict[str, str],
+        tenant_plans: dict[str, SubscriptionPlan],
+    ) -> list[str]:
+        """
+        Filter messages that should be deleted based on sandbox plan expiration.
+
+        A message should be deleted if:
+        1. It belongs to a sandbox tenant AND
+        2. Either:
+           a) The tenant has no previous subscription (expiration_date == -1), OR
+           b) The subscription expired more than graceful_period_days ago
+
+        Args:
+            messages: List of message objects with id and app_id attributes
+            app_to_tenant: Mapping from app_id to tenant_id
+            tenant_plans: Mapping from tenant_id to subscription plan info
+
+        Returns:
+            List of message IDs that should be deleted
+        """
+        current_timestamp = self._current_timestamp
+        if current_timestamp is None:
+            current_timestamp = int(datetime.datetime.now(datetime.UTC).timestamp())
+
+        sandbox_message_ids: list[str] = []
+        graceful_period_seconds = self._graceful_period_days * 24 * 60 * 60
+
+        for msg in messages:
+            # Get tenant_id for this message's app
+            tenant_id = app_to_tenant.get(msg.app_id)
+            if not tenant_id:
+                continue
+
+            # Skip tenant messages in whitelist
+            if tenant_id in self._tenant_whitelist:
+                continue
+
+            # Get subscription plan for this tenant
+            tenant_plan = tenant_plans.get(tenant_id)
+            if not tenant_plan:
+                continue
+
+            plan = str(tenant_plan["plan"])
+            expiration_date = int(tenant_plan["expiration_date"])
+
+            # Only process sandbox plans
+            if plan != CloudPlan.SANDBOX:
+                continue
+
+            # Case 1: No previous subscription (-1 means never had a paid subscription)
+            if expiration_date == -1:
+                sandbox_message_ids.append(msg.id)
+                continue
+
+            # Case 2: Subscription expired beyond grace period
+            if current_timestamp - expiration_date > graceful_period_seconds:
+                sandbox_message_ids.append(msg.id)
+
+        return sandbox_message_ids
+
+
+def create_message_clean_policy(
+    graceful_period_days: int = 21,
+    current_timestamp: int | None = None,
+) -> MessagesCleanPolicy:
+    """
+    Factory function to create the appropriate message clean policy.
+
+    Determines which policy to use based on BILLING_ENABLED configuration:
+    - If BILLING_ENABLED is True: returns BillingSandboxPolicy
+    - If BILLING_ENABLED is False: returns BillingDisabledPolicy
+
+    Args:
+        graceful_period_days: Grace period in days after subscription expiration (default: 21)
+        current_timestamp: Current Unix timestamp for testing (default: None, uses current time)
+    """
+    if not dify_config.BILLING_ENABLED:
+        logger.info("create_message_clean_policy: billing disabled, using BillingDisabledPolicy")
+        return BillingDisabledPolicy()
+
+    # Billing enabled - fetch whitelist from BillingService
+    tenant_whitelist = BillingService.get_expired_subscription_cleanup_whitelist()
+    plan_provider = BillingService.get_plan_bulk_with_cache
+
+    logger.info(
+        "create_message_clean_policy: billing enabled, using BillingSandboxPolicy "
+        "(graceful_period_days=%s, whitelist=%s)",
+        graceful_period_days,
+        tenant_whitelist,
+    )
+
+    return BillingSandboxPolicy(
+        plan_provider=plan_provider,
+        graceful_period_days=graceful_period_days,
+        tenant_whitelist=tenant_whitelist,
+        current_timestamp=current_timestamp,
+    )

api/services/retention/conversation/messages_clean_service.py

@@ -0,0 +1,334 @@
+import datetime
+import logging
+import random
+from collections.abc import Sequence
+from typing import cast
+
+from sqlalchemy import delete, select
+from sqlalchemy.engine import CursorResult
+from sqlalchemy.orm import Session
+
+from extensions.ext_database import db
+from models.model import (
+    App,
+    AppAnnotationHitHistory,
+    DatasetRetrieverResource,
+    Message,
+    MessageAgentThought,
+    MessageAnnotation,
+    MessageChain,
+    MessageFeedback,
+    MessageFile,
+)
+from models.web import SavedMessage
+from services.retention.conversation.messages_clean_policy import (
+    MessagesCleanPolicy,
+    SimpleMessage,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class MessagesCleanService:
+    """
+    Service for cleaning expired messages based on retention policies.
+
+    Compatible with non cloud edition (billing disabled): all messages in the time range will be deleted.
+    If billing is enabled: only sandbox plan tenant messages are deleted (with whitelist and grace period support).
+    """
+
+    def __init__(
+        self,
+        policy: MessagesCleanPolicy,
+        end_before: datetime.datetime,
+        start_from: datetime.datetime | None = None,
+        batch_size: int = 1000,
+        dry_run: bool = False,
+    ) -> None:
+        """
+        Initialize the service with cleanup parameters.
+
+        Args:
+            policy: The policy that determines which messages to delete
+            end_before: End time (exclusive) of the range
+            start_from: Optional start time (inclusive) of the range
+            batch_size: Number of messages to process per batch
+            dry_run: Whether to perform a dry run (no actual deletion)
+        """
+        self._policy = policy
+        self._end_before = end_before
+        self._start_from = start_from
+        self._batch_size = batch_size
+        self._dry_run = dry_run
+
+    @classmethod
+    def from_time_range(
+        cls,
+        policy: MessagesCleanPolicy,
+        start_from: datetime.datetime,
+        end_before: datetime.datetime,
+        batch_size: int = 1000,
+        dry_run: bool = False,
+    ) -> "MessagesCleanService":
+        """
+        Create a service instance for cleaning messages within a specific time range.
+
+        Time range is [start_from, end_before).
+
+        Args:
+            policy: The policy that determines which messages to delete
+            start_from: Start time (inclusive) of the range
+            end_before: End time (exclusive) of the range
+            batch_size: Number of messages to process per batch
+            dry_run: Whether to perform a dry run (no actual deletion)
+
+        Returns:
+            MessagesCleanService instance
+
+        Raises:
+            ValueError: If start_from >= end_before or invalid parameters
+        """
+        if start_from >= end_before:
+            raise ValueError(f"start_from ({start_from}) must be less than end_before ({end_before})")
+
+        if batch_size <= 0:
+            raise ValueError(f"batch_size ({batch_size}) must be greater than 0")
+
+        logger.info(
+            "clean_messages: start_from=%s, end_before=%s, batch_size=%s, policy=%s",
+            start_from,
+            end_before,
+            batch_size,
+            policy.__class__.__name__,
+        )
+
+        return cls(
+            policy=policy,
+            end_before=end_before,
+            start_from=start_from,
+            batch_size=batch_size,
+            dry_run=dry_run,
+        )
+
+    @classmethod
+    def from_days(
+        cls,
+        policy: MessagesCleanPolicy,
+        days: int = 30,
+        batch_size: int = 1000,
+        dry_run: bool = False,
+    ) -> "MessagesCleanService":
+        """
+        Create a service instance for cleaning messages older than specified days.
+
+        Args:
+            policy: The policy that determines which messages to delete
+            days: Number of days to look back from now
+            batch_size: Number of messages to process per batch
+            dry_run: Whether to perform a dry run (no actual deletion)
+
+        Returns:
+            MessagesCleanService instance
+
+        Raises:
+            ValueError: If invalid parameters
+        """
+        if days < 0:
+            raise ValueError(f"days ({days}) must be greater than or equal to 0")
+
+        if batch_size <= 0:
+            raise ValueError(f"batch_size ({batch_size}) must be greater than 0")
+
+        end_before = datetime.datetime.now() - datetime.timedelta(days=days)
+
+        logger.info(
+            "clean_messages: days=%s, end_before=%s, batch_size=%s, policy=%s",
+            days,
+            end_before,
+            batch_size,
+            policy.__class__.__name__,
+        )
+
+        return cls(policy=policy, end_before=end_before, start_from=None, batch_size=batch_size, dry_run=dry_run)
+
+    def run(self) -> dict[str, int]:
+        """
+        Execute the message cleanup operation.
+
+        Returns:
+            Dict with statistics: batches, filtered_messages, total_deleted
+        """
+        return self._clean_messages_by_time_range()
+
+    def _clean_messages_by_time_range(self) -> dict[str, int]:
+        """
+        Clean messages within a time range using cursor-based pagination.
+
+        Time range is [start_from, end_before)
+
+        Steps:
+        1. Iterate messages using cursor pagination (by created_at, id)
+        2. Query app_id -> tenant_id mapping
+        3. Delegate to policy to determine which messages to delete
+        4. Batch delete messages and their relations
+
+        Returns:
+            Dict with statistics: batches, filtered_messages, total_deleted
+        """
+        stats = {
+            "batches": 0,
+            "total_messages": 0,
+            "filtered_messages": 0,
+            "total_deleted": 0,
+        }
+
+        # Cursor-based pagination using (created_at, id) to avoid infinite loops
+        # and ensure proper ordering with time-based filtering
+        _cursor: tuple[datetime.datetime, str] | None = None
+
+        logger.info(
+            "clean_messages: start cleaning messages (dry_run=%s), start_from=%s, end_before=%s",
+            self._dry_run,
+            self._start_from,
+            self._end_before,
+        )
+
+        while True:
+            stats["batches"] += 1
+
+            # Step 1: Fetch a batch of messages using cursor
+            with Session(db.engine, expire_on_commit=False) as session:
+                msg_stmt = (
+                    select(Message.id, Message.app_id, Message.created_at)
+                    .where(Message.created_at < self._end_before)
+                    .order_by(Message.created_at, Message.id)
+                    .limit(self._batch_size)
+                )
+
+                if self._start_from:
+                    msg_stmt = msg_stmt.where(Message.created_at >= self._start_from)
+
+                # Apply cursor condition: (created_at, id) > (last_created_at, last_message_id)
+                # This translates to:
+                #   created_at > last_created_at OR (created_at = last_created_at AND id > last_message_id)
+                if _cursor:
+                    # Continuing from previous batch
+                    msg_stmt = msg_stmt.where(
+                        (Message.created_at > _cursor[0])
+                        | ((Message.created_at == _cursor[0]) & (Message.id > _cursor[1]))
+                    )
+
+                raw_messages = list(session.execute(msg_stmt).all())
+                messages = [
+                    SimpleMessage(id=msg_id, app_id=app_id, created_at=msg_created_at)
+                    for msg_id, app_id, msg_created_at in raw_messages
+                ]
+
+                # Track total messages fetched across all batches
+                stats["total_messages"] += len(messages)
+
+                if not messages:
+                    logger.info("clean_messages (batch %s): no more messages to process", stats["batches"])
+                    break
+
+                # Update cursor to the last message's (created_at, id)
+                _cursor = (messages[-1].created_at, messages[-1].id)
+
+                # Step 2: Extract app_ids and query tenant_ids
+                app_ids = list({msg.app_id for msg in messages})
+
+                if not app_ids:
+                    logger.info("clean_messages (batch %s): no app_ids found, skip", stats["batches"])
+                    continue
+
+                app_stmt = select(App.id, App.tenant_id).where(App.id.in_(app_ids))
+                apps = list(session.execute(app_stmt).all())
+
+            if not apps:
+                logger.info("clean_messages (batch %s): no apps found, skip", stats["batches"])
+                continue
+
+            # Build app_id -> tenant_id mapping
+            app_to_tenant: dict[str, str] = {app.id: app.tenant_id for app in apps}
+
+            # Step 3: Delegate to policy to determine which messages to delete
+            message_ids_to_delete = self._policy.filter_message_ids(messages, app_to_tenant)
+
+            if not message_ids_to_delete:
+                logger.info("clean_messages (batch %s): no messages to delete, skip", stats["batches"])
+                continue
+
+            stats["filtered_messages"] += len(message_ids_to_delete)
+
+            # Step 4: Batch delete messages and their relations
+            if not self._dry_run:
+                with Session(db.engine, expire_on_commit=False) as session:
+                    # Delete related records first
+                    self._batch_delete_message_relations(session, message_ids_to_delete)
+
+                    # Delete messages
+                    delete_stmt = delete(Message).where(Message.id.in_(message_ids_to_delete))
+                    delete_result = cast(CursorResult, session.execute(delete_stmt))
+                    messages_deleted = delete_result.rowcount
+                    session.commit()
+
+                    stats["total_deleted"] += messages_deleted
+
+                    logger.info(
+                        "clean_messages (batch %s): processed %s messages, deleted %s messages",
+                        stats["batches"],
+                        len(messages),
+                        messages_deleted,
+                    )
+            else:
+                # Log random sample of message IDs that would be deleted (up to 10)
+                sample_size = min(10, len(message_ids_to_delete))
+                sampled_ids = random.sample(list(message_ids_to_delete), sample_size)
+
+                logger.info(
+                    "clean_messages (batch %s, dry_run): would delete %s messages, sampling %s ids:",
+                    stats["batches"],
+                    len(message_ids_to_delete),
+                    sample_size,
+                )
+                for msg_id in sampled_ids:
+                    logger.info("clean_messages (batch %s, dry_run) sample: message_id=%s", stats["batches"], msg_id)
+
+        logger.info(
+            "clean_messages completed: total batches: %s, total messages: %s, filtered messages: %s, total deleted: %s",
+            stats["batches"],
+            stats["total_messages"],
+            stats["filtered_messages"],
+            stats["total_deleted"],
+        )
+
+        return stats
+
+    @staticmethod
+    def _batch_delete_message_relations(session: Session, message_ids: Sequence[str]) -> None:
+        """
+        Batch delete all related records for given message IDs.
+
+        Args:
+            session: Database session
+            message_ids: List of message IDs to delete relations for
+        """
+        if not message_ids:
+            return
+
+        # Delete all related records in batch
+        session.execute(delete(MessageFeedback).where(MessageFeedback.message_id.in_(message_ids)))
+
+        session.execute(delete(MessageAnnotation).where(MessageAnnotation.message_id.in_(message_ids)))
+
+        session.execute(delete(MessageChain).where(MessageChain.message_id.in_(message_ids)))
+
+        session.execute(delete(MessageAgentThought).where(MessageAgentThought.message_id.in_(message_ids)))
+
+        session.execute(delete(MessageFile).where(MessageFile.message_id.in_(message_ids)))
+
+        session.execute(delete(SavedMessage).where(SavedMessage.message_id.in_(message_ids)))
+
+        session.execute(delete(AppAnnotationHitHistory).where(AppAnnotationHitHistory.message_id.in_(message_ids)))
+
+        session.execute(delete(DatasetRetrieverResource).where(DatasetRetrieverResource.message_id.in_(message_ids)))

api/services/retention/workflow_run/clear_free_plan_expired_workflow_run_logs.py

@@ -1,8 +1,11 @@
 import datetime
 import logging
+import time
 from collections.abc import Iterable, Sequence
+from contextlib import contextmanager
 
 import click
+from sqlalchemy import event
 from sqlalchemy.orm import Session, sessionmaker
 
 from configs import dify_config
@@ -28,6 +31,8 @@ class WorkflowRunCleanup:
         end_before: datetime.datetime | None = None,
         workflow_run_repo: APIWorkflowRunRepository | None = None,
         dry_run: bool = False,
+        log_sql: bool = False,
+        log_sql_min_ms: int = 0,
     ):
         if (start_from is None) ^ (end_before is None):
             raise ValueError("start_from and end_before must be both set or both omitted.")
@@ -45,6 +50,8 @@ class WorkflowRunCleanup:
         self.batch_size = batch_size
         self._cleanup_whitelist: set[str] | None = None
         self.dry_run = dry_run
+        self.log_sql = log_sql
+        self.log_sql_min_ms = max(0, log_sql_min_ms)
         self.free_plan_grace_period_days = dify_config.SANDBOX_EXPIRED_RECORDS_CLEAN_GRACEFUL_PERIOD
         self.workflow_run_repo: APIWorkflowRunRepository
         if workflow_run_repo:
@@ -56,6 +63,38 @@ class WorkflowRunCleanup:
             session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
             self.workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
 
+    @contextmanager
+    def _sql_logger(self):
+        if not self.log_sql:
+            yield
+            return
+
+        def _before_cursor_execute(conn, cursor, statement, parameters, context, executemany) -> None:
+            context._dify_sql_start_time = time.monotonic()
+            context._dify_sql_statement = statement
+            context._dify_sql_parameters = parameters
+
+        def _after_cursor_execute(conn, cursor, statement, parameters, context, executemany) -> None:
+            start = getattr(context, "_dify_sql_start_time", None)
+            if start is None:
+                return
+            elapsed_ms = (time.monotonic() - start) * 1000
+            if elapsed_ms < self.log_sql_min_ms:
+                return
+            logged_statement = getattr(context, "_dify_sql_statement", statement)
+            logged_parameters = getattr(context, "_dify_sql_parameters", parameters)
+            click.echo(f"[sql] {elapsed_ms:.1f} ms {logged_statement}")
+            if logged_parameters:
+                click.echo(f"[sql] params: {logged_parameters}")
+
+        event.listen(db.engine, "before_cursor_execute", _before_cursor_execute)
+        event.listen(db.engine, "after_cursor_execute", _after_cursor_execute)
+        try:
+            yield
+        finally:
+            event.remove(db.engine, "before_cursor_execute", _before_cursor_execute)
+            event.remove(db.engine, "after_cursor_execute", _after_cursor_execute)
+
     def run(self) -> None:
         click.echo(
             click.style(
@@ -74,74 +113,95 @@ class WorkflowRunCleanup:
         batch_index = 0
         last_seen: tuple[datetime.datetime, str] | None = None
 
-        while True:
-            run_rows = self.workflow_run_repo.get_runs_batch_by_time_range(
-                start_from=self.window_start,
-                end_before=self.window_end,
-                last_seen=last_seen,
-                batch_size=self.batch_size,
-            )
-            if not run_rows:
-                break
+        with self._sql_logger():
+            while True:
+                batch_start = time.monotonic()
+                run_rows = self.workflow_run_repo.get_runs_batch_by_time_range(
+                    start_from=self.window_start,
+                    end_before=self.window_end,
+                    last_seen=last_seen,
+                    batch_size=self.batch_size,
+                )
+                fetch_ms = (time.monotonic() - batch_start) * 1000
+                if not run_rows:
+                    break
 
-            batch_index += 1
-            last_seen = (run_rows[-1].created_at, run_rows[-1].id)
-            tenant_ids = {row.tenant_id for row in run_rows}
-            free_tenants = self._filter_free_tenants(tenant_ids)
-            free_runs = [row for row in run_rows if row.tenant_id in free_tenants]
-            paid_or_skipped = len(run_rows) - len(free_runs)
+                batch_index += 1
+                last_seen = (run_rows[-1].created_at, run_rows[-1].id)
+                tenant_ids = {row.tenant_id for row in run_rows}
+                billing_start = time.monotonic()
+                free_tenants = self._filter_free_tenants(tenant_ids)
+                billing_ms = (time.monotonic() - billing_start) * 1000
+                free_runs = [row for row in run_rows if row.tenant_id in free_tenants]
+                paid_or_skipped = len(run_rows) - len(free_runs)
 
-            if not free_runs:
+                if self.log_sql:
+                    click.echo(
+                        click.style(
+                            f"[batch #{batch_index}] fetch_ms={fetch_ms:.1f} billing_ms={billing_ms:.1f}",
+                            fg="white",
+                        )
+                    )
+
+                if not free_runs:
+                    skipped_message = (
+                        f"[batch #{batch_index}] skipped (no sandbox runs in batch, "
+                        f"{paid_or_skipped} paid/unknown)"
+                    )
+                    click.echo(
+                        click.style(
+                            skipped_message,
+                            fg="yellow",
+                        )
+                    )
+                    continue
+
+                total_runs_targeted += len(free_runs)
+
+                if self.dry_run:
+                    count_start = time.monotonic()
+                    batch_counts = self.workflow_run_repo.count_runs_with_related(
+                        free_runs,
+                        count_node_executions=self._count_node_executions,
+                        count_trigger_logs=self._count_trigger_logs,
+                    )
+                    count_ms = (time.monotonic() - count_start) * 1000
+                    if self.log_sql:
+                        click.echo(click.style(f"[batch #{batch_index}] count_ms={count_ms:.1f}", fg="white"))
+                    if related_totals is not None:
+                        for key in related_totals:
+                            related_totals[key] += batch_counts.get(key, 0)
+                    sample_ids = ", ".join(run.id for run in free_runs[:5])
+                    click.echo(
+                        click.style(
+                            f"[batch #{batch_index}] would delete {len(free_runs)} runs "
+                            f"(sample ids: {sample_ids}) and skip {paid_or_skipped} paid/unknown",
+                            fg="yellow",
+                        )
+                    )
+                    continue
+
+                try:
+                    counts = self.workflow_run_repo.delete_runs_with_related(
+                        free_runs,
+                        delete_node_executions=self._delete_node_executions,
+                        delete_trigger_logs=self._delete_trigger_logs,
+                    )
+                except Exception:
+                    logger.exception("Failed to delete workflow runs batch ending at %s", last_seen[0])
+                    raise
+
+                total_runs_deleted += counts["runs"]
                 click.echo(
                     click.style(
-                        f"[batch #{batch_index}] skipped (no sandbox runs in batch, {paid_or_skipped} paid/unknown)",
-                        fg="yellow",
+                        f"[batch #{batch_index}] deleted runs: {counts['runs']} "
+                        f"(nodes {counts['node_executions']}, offloads {counts['offloads']}, "
+                        f"app_logs {counts['app_logs']}, trigger_logs {counts['trigger_logs']}, "
+                        f"pauses {counts['pauses']}, pause_reasons {counts['pause_reasons']}); "
+                        f"skipped {paid_or_skipped} paid/unknown",
+                        fg="green",
                     )
                 )
-                continue
-
-            total_runs_targeted += len(free_runs)
-
-            if self.dry_run:
-                batch_counts = self.workflow_run_repo.count_runs_with_related(
-                    free_runs,
-                    count_node_executions=self._count_node_executions,
-                    count_trigger_logs=self._count_trigger_logs,
-                )
-                if related_totals is not None:
-                    for key in related_totals:
-                        related_totals[key] += batch_counts.get(key, 0)
-                sample_ids = ", ".join(run.id for run in free_runs[:5])
-                click.echo(
-                    click.style(
-                        f"[batch #{batch_index}] would delete {len(free_runs)} runs "
-                        f"(sample ids: {sample_ids}) and skip {paid_or_skipped} paid/unknown",
-                        fg="yellow",
-                    )
-                )
-                continue
-
-            try:
-                counts = self.workflow_run_repo.delete_runs_with_related(
-                    free_runs,
-                    delete_node_executions=self._delete_node_executions,
-                    delete_trigger_logs=self._delete_trigger_logs,
-                )
-            except Exception:
-                logger.exception("Failed to delete workflow runs batch ending at %s", last_seen[0])
-                raise
-
-            total_runs_deleted += counts["runs"]
-            click.echo(
-                click.style(
-                    f"[batch #{batch_index}] deleted runs: {counts['runs']} "
-                    f"(nodes {counts['node_executions']}, offloads {counts['offloads']}, "
-                    f"app_logs {counts['app_logs']}, trigger_logs {counts['trigger_logs']}, "
-                    f"pauses {counts['pauses']}, pause_reasons {counts['pause_reasons']}); "
-                    f"skipped {paid_or_skipped} paid/unknown",
-                    fg="green",
-                )
-            )
 
         if self.dry_run:
             if self.window_start:

api/tests/unit_tests/core/workflow/nodes/llm/test_first_token_timeout.py

@@ -1,395 +0,0 @@
-"""Tests for LLM Node first token timeout retry functionality."""
-
-import time
-from collections.abc import Generator
-from unittest import mock
-
-import pytest
-
-from core.model_runtime.entities.llm_entities import LLMResultChunk, LLMResultChunkDelta
-from core.model_runtime.entities.message_entities import AssistantPromptMessage
-from core.workflow.nodes.base.entities import RetryConfig
-from core.workflow.utils.generator_timeout import FirstTokenTimeoutError, with_first_token_timeout
-
-
-class TestRetryConfigFirstTokenTimeout:
-    """Test cases for RetryConfig first token timeout fields."""
-
-    def test_default_values(self):
-        """Test that first token timeout fields have correct default values."""
-        config = RetryConfig()
-
-        assert config.first_token_timeout == 0
-        assert config.has_first_token_timeout is False
-
-    def test_has_first_token_timeout_when_retry_enabled_and_positive(self):
-        """Test has_first_token_timeout returns True when retry enabled with positive timeout."""
-        config = RetryConfig(
-            retry_enabled=True,
-            first_token_timeout=3000,  # 3000ms = 3s
-        )
-
-        assert config.has_first_token_timeout is True
-        assert config.first_token_timeout_seconds == 3.0
-
-    def test_has_first_token_timeout_when_retry_disabled(self):
-        """Test has_first_token_timeout returns False when retry is disabled."""
-        config = RetryConfig(
-            retry_enabled=False,
-            first_token_timeout=60,
-        )
-
-        assert config.has_first_token_timeout is False
-
-    def test_has_first_token_timeout_when_zero_timeout(self):
-        """Test has_first_token_timeout returns False when timeout is 0."""
-        config = RetryConfig(
-            retry_enabled=True,
-            first_token_timeout=0,
-        )
-
-        assert config.has_first_token_timeout is False
-
-    def test_backward_compatibility(self):
-        """Test that existing workflows without first_token_timeout work correctly."""
-        old_config_data = {
-            "max_retries": 3,
-            "retry_interval": 1000,
-            "retry_enabled": True,
-        }
-
-        config = RetryConfig.model_validate(old_config_data)
-
-        assert config.max_retries == 3
-        assert config.retry_interval == 1000
-        assert config.retry_enabled is True
-        assert config.first_token_timeout == 0
-        # has_first_token_timeout is False because timeout is 0
-        assert config.has_first_token_timeout is False
-
-    def test_full_config_serialization(self):
-        """Test that full config can be serialized and deserialized."""
-        config = RetryConfig(
-            max_retries=5,
-            retry_interval=2000,
-            retry_enabled=True,
-            first_token_timeout=120,
-        )
-
-        config_dict = config.model_dump()
-        restored_config = RetryConfig.model_validate(config_dict)
-
-        assert restored_config.max_retries == 5
-        assert restored_config.retry_interval == 2000
-        assert restored_config.retry_enabled is True
-        assert restored_config.first_token_timeout == 120
-        assert restored_config.has_first_token_timeout is True
-
-
-class TestWithFirstTokenTimeout:
-    """Test cases for with_first_token_timeout function."""
-
-    @staticmethod
-    def _create_mock_chunk(text: str = "test") -> LLMResultChunk:
-        """Helper to create a mock LLMResultChunk."""
-        return LLMResultChunk(
-            model="test-model",
-            prompt_messages=[],
-            delta=LLMResultChunkDelta(
-                index=0,
-                message=AssistantPromptMessage(content=text),
-            ),
-        )
-
-    def test_first_token_arrives_within_timeout(self):
-        """Test that chunks are yielded normally when first token arrives in time."""
-
-        def mock_generator() -> Generator[LLMResultChunk, None, None]:
-            yield self._create_mock_chunk("Hello")
-            yield self._create_mock_chunk(" world")
-
-        wrapped = with_first_token_timeout(mock_generator(), timeout_seconds=10)
-        chunks = list(wrapped)
-
-        assert len(chunks) == 2
-
-    def test_first_token_timeout_raises_error(self, monkeypatch):
-        """Test that timeout error is raised when first token doesn't arrive in time."""
-        call_count = 0
-
-        def mock_monotonic():
-            nonlocal call_count
-            call_count += 1
-            # First call: start_time = 0
-            # Second call (when checking): current_time = 11 (exceeds 10 second timeout)
-            if call_count == 1:
-                return 0.0
-            return 11.0
-
-        monkeypatch.setattr(time, "monotonic", mock_monotonic)
-
-        def slow_generator() -> Generator[LLMResultChunk, None, None]:
-            # This chunk arrives "after timeout"
-            yield self._create_mock_chunk("Late token")
-
-        wrapped = with_first_token_timeout(slow_generator(), timeout_seconds=10)
-
-        with pytest.raises(FirstTokenTimeoutError) as exc_info:
-            list(wrapped)
-
-        # Error message shows milliseconds (10 seconds = 10000ms)
-        assert "10000ms" in str(exc_info.value)
-
-    def test_no_timeout_check_after_first_token(self, monkeypatch):
-        """Test that subsequent chunks are not subject to timeout after first token received."""
-        call_count = 0
-
-        def mock_monotonic():
-            nonlocal call_count
-            call_count += 1
-            if call_count == 1:
-                return 0.0  # start_time
-            elif call_count == 2:
-                return 5.0  # first token arrives at 5s (within 10s timeout)
-            else:
-                # Subsequent calls simulate long delays for remaining chunks
-                # These should NOT trigger timeout because first token already received
-                return 100.0 + call_count
-
-        monkeypatch.setattr(time, "monotonic", mock_monotonic)
-
-        def generator_with_slow_subsequent_chunks() -> Generator[LLMResultChunk, None, None]:
-            yield self._create_mock_chunk("First")
-            yield self._create_mock_chunk("Second")
-            yield self._create_mock_chunk("Third")
-
-        wrapped = with_first_token_timeout(
-            generator_with_slow_subsequent_chunks(),
-            timeout_seconds=10,
-        )
-
-        # Should not raise, even though "time" passes beyond timeout after first token
-        chunks = list(wrapped)
-        assert len(chunks) == 3
-
-    def test_empty_generator_no_error(self):
-        """Test that empty generator doesn't raise timeout error (no chunks to check)."""
-
-        def empty_generator() -> Generator[LLMResultChunk, None, None]:
-            return
-            yield  # unreachable, but makes this a generator
-
-        wrapped = with_first_token_timeout(empty_generator(), timeout_seconds=10)
-        chunks = list(wrapped)
-
-        assert chunks == []
-
-    def test_exact_timeout_boundary(self, monkeypatch):
-        """Test behavior at exact timeout boundary (should not raise when equal)."""
-        call_count = 0
-
-        def mock_monotonic():
-            nonlocal call_count
-            call_count += 1
-            if call_count == 1:
-                return 0.0
-            # Exactly at boundary: current_time - start_time = 10, timeout_seconds = 10
-            # Since we check > not >=, this should NOT raise
-            return 10.0
-
-        monkeypatch.setattr(time, "monotonic", mock_monotonic)
-
-        def generator() -> Generator[LLMResultChunk, None, None]:
-            yield self._create_mock_chunk("Token at boundary")
-
-        wrapped = with_first_token_timeout(generator(), timeout_seconds=10)
-
-        # Should not raise because 10 is not > 10
-        chunks = list(wrapped)
-        assert len(chunks) == 1
-
-    def test_just_over_timeout_boundary(self, monkeypatch):
-        """Test behavior just over timeout boundary (should raise)."""
-        call_count = 0
-
-        def mock_monotonic():
-            nonlocal call_count
-            call_count += 1
-            if call_count == 1:
-                return 0.0
-            # Just over boundary
-            return 10.001
-
-        monkeypatch.setattr(time, "monotonic", mock_monotonic)
-
-        def generator() -> Generator[LLMResultChunk, None, None]:
-            yield self._create_mock_chunk("Late token")
-
-        wrapped = with_first_token_timeout(generator(), timeout_seconds=10)
-
-        with pytest.raises(FirstTokenTimeoutError):
-            list(wrapped)
-
-
-class TestLLMNodeInvokeLLMWithTimeout:
-    """Test cases for LLMNode.invoke_llm with first_token_timeout parameter."""
-
-    def test_invoke_llm_without_timeout(self):
-        """Test invoke_llm works normally when first_token_timeout is None."""
-        from core.workflow.nodes.llm.node import LLMNode
-
-        with mock.patch.object(LLMNode, "handle_invoke_result") as mock_handle:
-            mock_handle.return_value = iter([])
-
-            # Mock model_instance.invoke_llm to return empty generator
-            mock_model_instance = mock.MagicMock()
-            mock_model_instance.invoke_llm.return_value = iter([])
-            mock_model_instance.model_type_instance.get_model_schema.return_value = mock.MagicMock()
-
-            mock_node_data_model = mock.MagicMock()
-            mock_node_data_model.completion_params = {}
-
-            result = LLMNode.invoke_llm(
-                node_data_model=mock_node_data_model,
-                model_instance=mock_model_instance,
-                prompt_messages=[],
-                user_id="test-user",
-                structured_output_enabled=False,
-                structured_output=None,
-                file_saver=mock.MagicMock(),
-                file_outputs=[],
-                node_id="test-node",
-                node_type=mock.MagicMock(),
-                first_token_timeout=None,  # No timeout
-            )
-
-            list(result)  # Consume generator
-            mock_handle.assert_called_once()
-
-    def test_invoke_llm_with_timeout_passes_to_model_instance(self):
-        """Test invoke_llm passes first_token_timeout to model_instance.invoke_llm."""
-        from core.workflow.nodes.llm.node import LLMNode
-
-        with mock.patch.object(LLMNode, "handle_invoke_result") as mock_handle:
-            mock_handle.return_value = iter([])
-
-            mock_model_instance = mock.MagicMock()
-            mock_model_instance.invoke_llm.return_value = iter([])
-            mock_model_instance.model_type_instance.get_model_schema.return_value = mock.MagicMock()
-
-            mock_node_data_model = mock.MagicMock()
-            mock_node_data_model.completion_params = {}
-
-            result = LLMNode.invoke_llm(
-                node_data_model=mock_node_data_model,
-                model_instance=mock_model_instance,
-                prompt_messages=[],
-                user_id="test-user",
-                structured_output_enabled=False,
-                structured_output=None,
-                file_saver=mock.MagicMock(),
-                file_outputs=[],
-                node_id="test-node",
-                node_type=mock.MagicMock(),
-                first_token_timeout=60,  # With timeout
-            )
-
-            list(result)  # Consume generator
-
-            # Verify model_instance.invoke_llm was called with first_token_timeout
-            mock_model_instance.invoke_llm.assert_called_once()
-            call_kwargs = mock_model_instance.invoke_llm.call_args.kwargs
-            assert call_kwargs.get("first_token_timeout") == 60
-
-    def test_invoke_llm_with_zero_timeout_passes_zero(self):
-        """Test invoke_llm passes zero timeout to model_instance."""
-        from core.workflow.nodes.llm.node import LLMNode
-
-        with mock.patch.object(LLMNode, "handle_invoke_result") as mock_handle:
-            mock_handle.return_value = iter([])
-
-            mock_model_instance = mock.MagicMock()
-            mock_model_instance.invoke_llm.return_value = iter([])
-            mock_model_instance.model_type_instance.get_model_schema.return_value = mock.MagicMock()
-
-            mock_node_data_model = mock.MagicMock()
-            mock_node_data_model.completion_params = {}
-
-            result = LLMNode.invoke_llm(
-                node_data_model=mock_node_data_model,
-                model_instance=mock_model_instance,
-                prompt_messages=[],
-                user_id="test-user",
-                structured_output_enabled=False,
-                structured_output=None,
-                file_saver=mock.MagicMock(),
-                file_outputs=[],
-                node_id="test-node",
-                node_type=mock.MagicMock(),
-                first_token_timeout=0,  # Zero timeout
-            )
-
-            list(result)  # Consume generator
-
-            # Verify model_instance.invoke_llm was called with zero timeout
-            mock_model_instance.invoke_llm.assert_called_once()
-            call_kwargs = mock_model_instance.invoke_llm.call_args.kwargs
-            assert call_kwargs.get("first_token_timeout") == 0
-
-
-class TestRetryConfigIntegration:
-    """Integration tests for RetryConfig with LLM node data."""
-
-    def test_retry_config_in_node_data(self):
-        """Test RetryConfig can be properly configured in LLMNodeData."""
-        from core.model_runtime.entities.llm_entities import LLMMode
-        from core.workflow.nodes.llm.entities import ContextConfig, LLMNodeData, ModelConfig
-
-        node_data = LLMNodeData(
-            title="Test LLM",
-            model=ModelConfig(
-                provider="openai",
-                name="gpt-4",
-                mode=LLMMode.CHAT,
-                completion_params={},
-            ),
-            prompt_template=[],
-            context=ContextConfig(enabled=False),
-            structured_output_enabled=False,
-            retry_config=RetryConfig(
-                max_retries=3,
-                retry_interval=1000,
-                retry_enabled=True,
-                first_token_timeout=3000,  # 3000ms = 3s
-            ),
-        )
-
-        assert node_data.retry_config.max_retries == 3
-        assert node_data.retry_config.retry_enabled is True
-        assert node_data.retry_config.first_token_timeout == 3000
-        assert node_data.retry_config.first_token_timeout_seconds == 3.0
-        assert node_data.retry_config.has_first_token_timeout is True
-
-    def test_default_retry_config_in_node_data(self):
-        """Test default RetryConfig in LLMNodeData."""
-        from core.model_runtime.entities.llm_entities import LLMMode
-        from core.workflow.nodes.llm.entities import ContextConfig, LLMNodeData, ModelConfig
-
-        node_data = LLMNodeData(
-            title="Test LLM",
-            model=ModelConfig(
-                provider="openai",
-                name="gpt-4",
-                mode=LLMMode.CHAT,
-                completion_params={},
-            ),
-            prompt_template=[],
-            context=ContextConfig(enabled=False),
-            structured_output_enabled=False,
-        )
-
-        # Should have default RetryConfig
-        assert node_data.retry_config.max_retries == 0
-        assert node_data.retry_config.retry_enabled is False
-        assert node_data.retry_config.first_token_timeout == 0
-        assert node_data.retry_config.has_first_token_timeout is False

api/tests/unit_tests/services/test_messages_clean_service.py

@@ -0,0 +1,627 @@
+import datetime
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from enums.cloud_plan import CloudPlan
+from services.retention.conversation.messages_clean_policy import (
+    BillingDisabledPolicy,
+    BillingSandboxPolicy,
+    SimpleMessage,
+    create_message_clean_policy,
+)
+from services.retention.conversation.messages_clean_service import MessagesCleanService
+
+
+def make_simple_message(msg_id: str, app_id: str) -> SimpleMessage:
+    """Helper to create a SimpleMessage with a fixed created_at timestamp."""
+    return SimpleMessage(id=msg_id, app_id=app_id, created_at=datetime.datetime(2024, 1, 1))
+
+
+def make_plan_provider(tenant_plans: dict) -> MagicMock:
+    """Helper to create a mock plan_provider that returns the given tenant_plans."""
+    provider = MagicMock()
+    provider.return_value = tenant_plans
+    return provider
+
+
+class TestBillingSandboxPolicyFilterMessageIds:
+    """Unit tests for BillingSandboxPolicy.filter_message_ids method."""
+
+    # Fixed timestamp for deterministic tests
+    CURRENT_TIMESTAMP = 1000000
+    GRACEFUL_PERIOD_DAYS = 8
+    GRACEFUL_PERIOD_SECONDS = GRACEFUL_PERIOD_DAYS * 24 * 60 * 60
+
+    def test_missing_tenant_mapping_excluded(self):
+        """Test that messages with missing app-to-tenant mapping are excluded."""
+        # Arrange
+        messages = [
+            make_simple_message("msg1", "app1"),
+            make_simple_message("msg2", "app2"),
+        ]
+        app_to_tenant = {}  # No mapping
+        tenant_plans = {"tenant1": {"plan": CloudPlan.SANDBOX, "expiration_date": -1}}
+        plan_provider = make_plan_provider(tenant_plans)
+
+        policy = BillingSandboxPolicy(
+            plan_provider=plan_provider,
+            graceful_period_days=self.GRACEFUL_PERIOD_DAYS,
+            current_timestamp=self.CURRENT_TIMESTAMP,
+        )
+
+        # Act
+        result = policy.filter_message_ids(messages, app_to_tenant)
+
+        # Assert
+        assert list(result) == []
+
+    def test_missing_tenant_plan_excluded(self):
+        """Test that messages with missing tenant plan are excluded (safe default)."""
+        # Arrange
+        messages = [
+            make_simple_message("msg1", "app1"),
+            make_simple_message("msg2", "app2"),
+        ]
+        app_to_tenant = {"app1": "tenant1", "app2": "tenant2"}
+        tenant_plans = {}  # No plans
+        plan_provider = make_plan_provider(tenant_plans)
+
+        policy = BillingSandboxPolicy(
+            plan_provider=plan_provider,
+            graceful_period_days=self.GRACEFUL_PERIOD_DAYS,
+            current_timestamp=self.CURRENT_TIMESTAMP,
+        )
+
+        # Act
+        result = policy.filter_message_ids(messages, app_to_tenant)
+
+        # Assert
+        assert list(result) == []
+
+    def test_non_sandbox_plan_excluded(self):
+        """Test that messages from non-sandbox plans (PROFESSIONAL/TEAM) are excluded."""
+        # Arrange
+        messages = [
+            make_simple_message("msg1", "app1"),
+            make_simple_message("msg2", "app2"),
+            make_simple_message("msg3", "app3"),
+        ]
+        app_to_tenant = {"app1": "tenant1", "app2": "tenant2", "app3": "tenant3"}
+        tenant_plans = {
+            "tenant1": {"plan": CloudPlan.PROFESSIONAL, "expiration_date": -1},
+            "tenant2": {"plan": CloudPlan.TEAM, "expiration_date": -1},
+            "tenant3": {"plan": CloudPlan.SANDBOX, "expiration_date": -1},  # Only this one
+        }
+        plan_provider = make_plan_provider(tenant_plans)
+
+        policy = BillingSandboxPolicy(
+            plan_provider=plan_provider,
+            graceful_period_days=self.GRACEFUL_PERIOD_DAYS,
+            current_timestamp=self.CURRENT_TIMESTAMP,
+        )
+
+        # Act
+        result = policy.filter_message_ids(messages, app_to_tenant)
+
+        # Assert - only msg3 (sandbox tenant) should be included
+        assert set(result) == {"msg3"}
+
+    def test_whitelist_skip(self):
+        """Test that whitelisted tenants are excluded even if sandbox + expired."""
+        # Arrange
+        messages = [
+            make_simple_message("msg1", "app1"),  # Whitelisted - excluded
+            make_simple_message("msg2", "app2"),  # Not whitelisted - included
+            make_simple_message("msg3", "app3"),  # Whitelisted - excluded
+        ]
+        app_to_tenant = {"app1": "tenant1", "app2": "tenant2", "app3": "tenant3"}
+        tenant_plans = {
+            "tenant1": {"plan": CloudPlan.SANDBOX, "expiration_date": -1},
+            "tenant2": {"plan": CloudPlan.SANDBOX, "expiration_date": -1},
+            "tenant3": {"plan": CloudPlan.SANDBOX, "expiration_date": -1},
+        }
+        plan_provider = make_plan_provider(tenant_plans)
+        tenant_whitelist = ["tenant1", "tenant3"]
+
+        policy = BillingSandboxPolicy(
+            plan_provider=plan_provider,
+            graceful_period_days=self.GRACEFUL_PERIOD_DAYS,
+            tenant_whitelist=tenant_whitelist,
+            current_timestamp=self.CURRENT_TIMESTAMP,
+        )
+
+        # Act
+        result = policy.filter_message_ids(messages, app_to_tenant)
+
+        # Assert - only msg2 should be included
+        assert set(result) == {"msg2"}
+
+    def test_no_previous_subscription_included(self):
+        """Test that messages with expiration_date=-1 (no previous subscription) are included."""
+        # Arrange
+        messages = [
+            make_simple_message("msg1", "app1"),
+            make_simple_message("msg2", "app2"),
+        ]
+        app_to_tenant = {"app1": "tenant1", "app2": "tenant2"}
+        tenant_plans = {
+            "tenant1": {"plan": CloudPlan.SANDBOX, "expiration_date": -1},
+            "tenant2": {"plan": CloudPlan.SANDBOX, "expiration_date": -1},
+        }
+        plan_provider = make_plan_provider(tenant_plans)
+
+        policy = BillingSandboxPolicy(
+            plan_provider=plan_provider,
+            graceful_period_days=self.GRACEFUL_PERIOD_DAYS,
+            current_timestamp=self.CURRENT_TIMESTAMP,
+        )
+
+        # Act
+        result = policy.filter_message_ids(messages, app_to_tenant)
+
+        # Assert - all messages should be included
+        assert set(result) == {"msg1", "msg2"}
+
+    def test_within_grace_period_excluded(self):
+        """Test that messages within grace period are excluded."""
+        # Arrange
+        now = self.CURRENT_TIMESTAMP
+        expired_1_day_ago = now - (1 * 24 * 60 * 60)
+        expired_5_days_ago = now - (5 * 24 * 60 * 60)
+        expired_7_days_ago = now - (7 * 24 * 60 * 60)
+
+        messages = [
+            make_simple_message("msg1", "app1"),
+            make_simple_message("msg2", "app2"),
+            make_simple_message("msg3", "app3"),
+        ]
+        app_to_tenant = {"app1": "tenant1", "app2": "tenant2", "app3": "tenant3"}
+        tenant_plans = {
+            "tenant1": {"plan": CloudPlan.SANDBOX, "expiration_date": expired_1_day_ago},
+            "tenant2": {"plan": CloudPlan.SANDBOX, "expiration_date": expired_5_days_ago},
+            "tenant3": {"plan": CloudPlan.SANDBOX, "expiration_date": expired_7_days_ago},
+        }
+        plan_provider = make_plan_provider(tenant_plans)
+
+        policy = BillingSandboxPolicy(
+            plan_provider=plan_provider,
+            graceful_period_days=self.GRACEFUL_PERIOD_DAYS,  # 8 days
+            current_timestamp=now,
+        )
+
+        # Act
+        result = policy.filter_message_ids(messages, app_to_tenant)
+
+        # Assert - all within 8-day grace period, none should be included
+        assert list(result) == []
+
+    def test_exactly_at_boundary_excluded(self):
+        """Test that messages exactly at grace period boundary are excluded (code uses >)."""
+        # Arrange
+        now = self.CURRENT_TIMESTAMP
+        expired_exactly_8_days_ago = now - self.GRACEFUL_PERIOD_SECONDS  # Exactly at boundary
+
+        messages = [make_simple_message("msg1", "app1")]
+        app_to_tenant = {"app1": "tenant1"}
+        tenant_plans = {
+            "tenant1": {"plan": CloudPlan.SANDBOX, "expiration_date": expired_exactly_8_days_ago},
+        }
+        plan_provider = make_plan_provider(tenant_plans)
+
+        policy = BillingSandboxPolicy(
+            plan_provider=plan_provider,
+            graceful_period_days=self.GRACEFUL_PERIOD_DAYS,
+            current_timestamp=now,
+        )
+
+        # Act
+        result = policy.filter_message_ids(messages, app_to_tenant)
+
+        # Assert - exactly at boundary (==) should be excluded (code uses >)
+        assert list(result) == []
+
+    def test_beyond_grace_period_included(self):
+        """Test that messages beyond grace period are included."""
+        # Arrange
+        now = self.CURRENT_TIMESTAMP
+        expired_9_days_ago = now - (9 * 24 * 60 * 60)  # Just beyond 8-day grace
+        expired_30_days_ago = now - (30 * 24 * 60 * 60)  # Well beyond
+
+        messages = [
+            make_simple_message("msg1", "app1"),
+            make_simple_message("msg2", "app2"),
+        ]
+        app_to_tenant = {"app1": "tenant1", "app2": "tenant2"}
+        tenant_plans = {
+            "tenant1": {"plan": CloudPlan.SANDBOX, "expiration_date": expired_9_days_ago},
+            "tenant2": {"plan": CloudPlan.SANDBOX, "expiration_date": expired_30_days_ago},
+        }
+        plan_provider = make_plan_provider(tenant_plans)
+
+        policy = BillingSandboxPolicy(
+            plan_provider=plan_provider,
+            graceful_period_days=self.GRACEFUL_PERIOD_DAYS,
+            current_timestamp=now,
+        )
+
+        # Act
+        result = policy.filter_message_ids(messages, app_to_tenant)
+
+        # Assert - both beyond grace period, should be included
+        assert set(result) == {"msg1", "msg2"}
+
+    def test_empty_messages_returns_empty(self):
+        """Test that empty messages returns empty list."""
+        # Arrange
+        messages: list[SimpleMessage] = []
+        app_to_tenant = {"app1": "tenant1"}
+        plan_provider = make_plan_provider({"tenant1": {"plan": CloudPlan.SANDBOX, "expiration_date": -1}})
+
+        policy = BillingSandboxPolicy(
+            plan_provider=plan_provider,
+            graceful_period_days=self.GRACEFUL_PERIOD_DAYS,
+            current_timestamp=self.CURRENT_TIMESTAMP,
+        )
+
+        # Act
+        result = policy.filter_message_ids(messages, app_to_tenant)
+
+        # Assert
+        assert list(result) == []
+
+    def test_plan_provider_called_with_correct_tenant_ids(self):
+        """Test that plan_provider is called with correct tenant_ids."""
+        # Arrange
+        messages = [
+            make_simple_message("msg1", "app1"),
+            make_simple_message("msg2", "app2"),
+            make_simple_message("msg3", "app3"),
+        ]
+        app_to_tenant = {"app1": "tenant1", "app2": "tenant2", "app3": "tenant1"}  # tenant1 appears twice
+        plan_provider = make_plan_provider({})
+
+        policy = BillingSandboxPolicy(
+            plan_provider=plan_provider,
+            graceful_period_days=self.GRACEFUL_PERIOD_DAYS,
+            current_timestamp=self.CURRENT_TIMESTAMP,
+        )
+
+        # Act
+        policy.filter_message_ids(messages, app_to_tenant)
+
+        # Assert - plan_provider should be called once with unique tenant_ids
+        plan_provider.assert_called_once()
+        called_tenant_ids = set(plan_provider.call_args[0][0])
+        assert called_tenant_ids == {"tenant1", "tenant2"}
+
+    def test_complex_mixed_scenario(self):
+        """Test complex scenario with mixed plans, expirations, whitelist, and missing mappings."""
+        # Arrange
+        now = self.CURRENT_TIMESTAMP
+        sandbox_expired_old = now - (15 * 24 * 60 * 60)  # Beyond grace
+        sandbox_expired_recent = now - (3 * 24 * 60 * 60)  # Within grace
+        future_expiration = now + (30 * 24 * 60 * 60)
+
+        messages = [
+            make_simple_message("msg1", "app1"),  # Sandbox, no subscription - included
+            make_simple_message("msg2", "app2"),  # Sandbox, expired old - included
+            make_simple_message("msg3", "app3"),  # Sandbox, within grace - excluded
+            make_simple_message("msg4", "app4"),  # Team plan, active - excluded
+            make_simple_message("msg5", "app5"),  # No tenant mapping - excluded
+            make_simple_message("msg6", "app6"),  # No plan info - excluded
+            make_simple_message("msg7", "app7"),  # Sandbox, expired old, whitelisted - excluded
+        ]
+        app_to_tenant = {
+            "app1": "tenant1",
+            "app2": "tenant2",
+            "app3": "tenant3",
+            "app4": "tenant4",
+            "app6": "tenant6",  # Has mapping but no plan
+            "app7": "tenant7",
+            # app5 has no mapping
+        }
+        tenant_plans = {
+            "tenant1": {"plan": CloudPlan.SANDBOX, "expiration_date": -1},
+            "tenant2": {"plan": CloudPlan.SANDBOX, "expiration_date": sandbox_expired_old},
+            "tenant3": {"plan": CloudPlan.SANDBOX, "expiration_date": sandbox_expired_recent},
+            "tenant4": {"plan": CloudPlan.TEAM, "expiration_date": future_expiration},
+            "tenant7": {"plan": CloudPlan.SANDBOX, "expiration_date": sandbox_expired_old},
+            # tenant6 has no plan
+        }
+        plan_provider = make_plan_provider(tenant_plans)
+        tenant_whitelist = ["tenant7"]
+
+        policy = BillingSandboxPolicy(
+            plan_provider=plan_provider,
+            graceful_period_days=self.GRACEFUL_PERIOD_DAYS,
+            tenant_whitelist=tenant_whitelist,
+            current_timestamp=now,
+        )
+
+        # Act
+        result = policy.filter_message_ids(messages, app_to_tenant)
+
+        # Assert - only msg1 and msg2 should be included
+        assert set(result) == {"msg1", "msg2"}
+
+
+class TestBillingDisabledPolicyFilterMessageIds:
+    """Unit tests for BillingDisabledPolicy.filter_message_ids method."""
+
+    def test_returns_all_message_ids(self):
+        """Test that all message IDs are returned (order-preserving)."""
+        # Arrange
+        messages = [
+            make_simple_message("msg1", "app1"),
+            make_simple_message("msg2", "app2"),
+            make_simple_message("msg3", "app3"),
+        ]
+        app_to_tenant = {"app1": "tenant1", "app2": "tenant2"}
+
+        policy = BillingDisabledPolicy()
+
+        # Act
+        result = policy.filter_message_ids(messages, app_to_tenant)
+
+        # Assert - all message IDs returned in order
+        assert list(result) == ["msg1", "msg2", "msg3"]
+
+    def test_ignores_app_to_tenant(self):
+        """Test that app_to_tenant mapping is ignored."""
+        # Arrange
+        messages = [
+            make_simple_message("msg1", "app1"),
+            make_simple_message("msg2", "app2"),
+        ]
+        app_to_tenant: dict[str, str] = {}  # Empty - should be ignored
+
+        policy = BillingDisabledPolicy()
+
+        # Act
+        result = policy.filter_message_ids(messages, app_to_tenant)
+
+        # Assert - all message IDs still returned
+        assert list(result) == ["msg1", "msg2"]
+
+    def test_empty_messages_returns_empty(self):
+        """Test that empty messages returns empty list."""
+        # Arrange
+        messages: list[SimpleMessage] = []
+        app_to_tenant = {"app1": "tenant1"}
+
+        policy = BillingDisabledPolicy()
+
+        # Act
+        result = policy.filter_message_ids(messages, app_to_tenant)
+
+        # Assert
+        assert list(result) == []
+
+
+class TestCreateMessageCleanPolicy:
+    """Unit tests for create_message_clean_policy factory function."""
+
+    @patch("services.retention.conversation.messages_clean_policy.dify_config")
+    def test_billing_disabled_returns_billing_disabled_policy(self, mock_config):
+        """Test that BILLING_ENABLED=False returns BillingDisabledPolicy."""
+        # Arrange
+        mock_config.BILLING_ENABLED = False
+
+        # Act
+        policy = create_message_clean_policy(graceful_period_days=21)
+
+        # Assert
+        assert isinstance(policy, BillingDisabledPolicy)
+
+    @patch("services.retention.conversation.messages_clean_policy.BillingService")
+    @patch("services.retention.conversation.messages_clean_policy.dify_config")
+    def test_billing_enabled_policy_has_correct_internals(self, mock_config, mock_billing_service):
+        """Test that BillingSandboxPolicy is created with correct internal values."""
+        # Arrange
+        mock_config.BILLING_ENABLED = True
+        whitelist = ["tenant1", "tenant2"]
+        mock_billing_service.get_expired_subscription_cleanup_whitelist.return_value = whitelist
+        mock_plan_provider = MagicMock()
+        mock_billing_service.get_plan_bulk_with_cache = mock_plan_provider
+
+        # Act
+        policy = create_message_clean_policy(graceful_period_days=14, current_timestamp=1234567)
+
+        # Assert
+        mock_billing_service.get_expired_subscription_cleanup_whitelist.assert_called_once()
+        assert isinstance(policy, BillingSandboxPolicy)
+        assert policy._graceful_period_days == 14
+        assert list(policy._tenant_whitelist) == whitelist
+        assert policy._plan_provider == mock_plan_provider
+        assert policy._current_timestamp == 1234567
+
+
+class TestMessagesCleanServiceFromTimeRange:
+    """Unit tests for MessagesCleanService.from_time_range factory method."""
+
+    def test_start_from_end_before_raises_value_error(self):
+        """Test that start_from == end_before raises ValueError."""
+        policy = BillingDisabledPolicy()
+
+        # Arrange
+        same_time = datetime.datetime(2024, 1, 1, 12, 0, 0)
+
+        # Act & Assert
+        with pytest.raises(ValueError, match="start_from .* must be less than end_before"):
+            MessagesCleanService.from_time_range(
+                policy=policy,
+                start_from=same_time,
+                end_before=same_time,
+            )
+
+        # Arrange
+        start_from = datetime.datetime(2024, 12, 31)
+        end_before = datetime.datetime(2024, 1, 1)
+
+        # Act & Assert
+        with pytest.raises(ValueError, match="start_from .* must be less than end_before"):
+            MessagesCleanService.from_time_range(
+                policy=policy,
+                start_from=start_from,
+                end_before=end_before,
+            )
+
+    def test_batch_size_raises_value_error(self):
+        """Test that batch_size=0 raises ValueError."""
+        # Arrange
+        start_from = datetime.datetime(2024, 1, 1)
+        end_before = datetime.datetime(2024, 2, 1)
+        policy = BillingDisabledPolicy()
+
+        # Act & Assert
+        with pytest.raises(ValueError, match="batch_size .* must be greater than 0"):
+            MessagesCleanService.from_time_range(
+                policy=policy,
+                start_from=start_from,
+                end_before=end_before,
+                batch_size=0,
+            )
+
+        start_from = datetime.datetime(2024, 1, 1)
+        end_before = datetime.datetime(2024, 2, 1)
+        policy = BillingDisabledPolicy()
+
+        # Act & Assert
+        with pytest.raises(ValueError, match="batch_size .* must be greater than 0"):
+            MessagesCleanService.from_time_range(
+                policy=policy,
+                start_from=start_from,
+                end_before=end_before,
+                batch_size=-100,
+            )
+
+    def test_valid_params_creates_instance(self):
+        """Test that valid parameters create a correctly configured instance."""
+        # Arrange
+        start_from = datetime.datetime(2024, 1, 1, 0, 0, 0)
+        end_before = datetime.datetime(2024, 12, 31, 23, 59, 59)
+        policy = BillingDisabledPolicy()
+        batch_size = 500
+        dry_run = True
+
+        # Act
+        service = MessagesCleanService.from_time_range(
+            policy=policy,
+            start_from=start_from,
+            end_before=end_before,
+            batch_size=batch_size,
+            dry_run=dry_run,
+        )
+
+        # Assert
+        assert isinstance(service, MessagesCleanService)
+        assert service._policy is policy
+        assert service._start_from == start_from
+        assert service._end_before == end_before
+        assert service._batch_size == batch_size
+        assert service._dry_run == dry_run
+
+    def test_default_params(self):
+        """Test that default parameters are applied correctly."""
+        # Arrange
+        start_from = datetime.datetime(2024, 1, 1)
+        end_before = datetime.datetime(2024, 2, 1)
+        policy = BillingDisabledPolicy()
+
+        # Act
+        service = MessagesCleanService.from_time_range(
+            policy=policy,
+            start_from=start_from,
+            end_before=end_before,
+        )
+
+        # Assert
+        assert service._batch_size == 1000  # default
+        assert service._dry_run is False  # default
+
+
+class TestMessagesCleanServiceFromDays:
+    """Unit tests for MessagesCleanService.from_days factory method."""
+
+    def test_days_raises_value_error(self):
+        """Test that days < 0 raises ValueError."""
+        # Arrange
+        policy = BillingDisabledPolicy()
+
+        # Act & Assert
+        with pytest.raises(ValueError, match="days .* must be greater than or equal to 0"):
+            MessagesCleanService.from_days(policy=policy, days=-1)
+
+        # Act
+        with patch("services.retention.conversation.messages_clean_service.datetime") as mock_datetime:
+            fixed_now = datetime.datetime(2024, 6, 15, 14, 0, 0)
+            mock_datetime.datetime.now.return_value = fixed_now
+            mock_datetime.timedelta = datetime.timedelta
+
+            service = MessagesCleanService.from_days(policy=policy, days=0)
+
+        # Assert
+        assert service._end_before == fixed_now
+
+    def test_batch_size_raises_value_error(self):
+        """Test that batch_size=0 raises ValueError."""
+        # Arrange
+        policy = BillingDisabledPolicy()
+
+        # Act & Assert
+        with pytest.raises(ValueError, match="batch_size .* must be greater than 0"):
+            MessagesCleanService.from_days(policy=policy, days=30, batch_size=0)
+
+        # Act & Assert
+        with pytest.raises(ValueError, match="batch_size .* must be greater than 0"):
+            MessagesCleanService.from_days(policy=policy, days=30, batch_size=-500)
+
+    def test_valid_params_creates_instance(self):
+        """Test that valid parameters create a correctly configured instance."""
+        # Arrange
+        policy = BillingDisabledPolicy()
+        days = 90
+        batch_size = 500
+        dry_run = True
+
+        # Act
+        with patch("services.retention.conversation.messages_clean_service.datetime") as mock_datetime:
+            fixed_now = datetime.datetime(2024, 6, 15, 10, 30, 0)
+            mock_datetime.datetime.now.return_value = fixed_now
+            mock_datetime.timedelta = datetime.timedelta
+
+            service = MessagesCleanService.from_days(
+                policy=policy,
+                days=days,
+                batch_size=batch_size,
+                dry_run=dry_run,
+            )
+
+        # Assert
+        expected_end_before = fixed_now - datetime.timedelta(days=days)
+        assert isinstance(service, MessagesCleanService)
+        assert service._policy is policy
+        assert service._start_from is None
+        assert service._end_before == expected_end_before
+        assert service._batch_size == batch_size
+        assert service._dry_run == dry_run
+
+    def test_default_params(self):
+        """Test that default parameters are applied correctly."""
+        # Arrange
+        policy = BillingDisabledPolicy()
+
+        # Act
+        with patch("services.retention.conversation.messages_clean_service.datetime") as mock_datetime:
+            fixed_now = datetime.datetime(2024, 6, 15, 10, 30, 0)
+            mock_datetime.datetime.now.return_value = fixed_now
+            mock_datetime.timedelta = datetime.timedelta
+
+            service = MessagesCleanService.from_days(policy=policy)
+
+        # Assert
+        expected_end_before = fixed_now - datetime.timedelta(days=30)  # default days=30
+        assert service._end_before == expected_end_before
+        assert service._batch_size == 1000  # default
+        assert service._dry_run is False  # default

api/uv.lock

@@ -1368,7 +1368,7 @@ wheels = [
 
 [[package]]
 name = "dify-api"
-version = "1.11.3"
+version = "1.11.4"
 source = { virtual = "." }
 dependencies = [
     { name = "aliyun-log-python-sdk" },

docker/docker-compose-template.yaml

@@ -21,7 +21,7 @@ services:
 
   # API service
   api:
-    image: langgenius/dify-api:1.11.3
+    image: langgenius/dify-api:1.11.4
     restart: always
     environment:
       # Use the shared environment variables.
@@ -63,7 +63,7 @@ services:
   # worker service
   # The Celery worker for processing all queues (dataset, workflow, mail, etc.)
   worker:
-    image: langgenius/dify-api:1.11.3
+    image: langgenius/dify-api:1.11.4
     restart: always
     environment:
       # Use the shared environment variables.
@@ -102,7 +102,7 @@ services:
   # worker_beat service
   # Celery beat for scheduling periodic tasks.
   worker_beat:
-    image: langgenius/dify-api:1.11.3
+    image: langgenius/dify-api:1.11.4
     restart: always
     environment:
       # Use the shared environment variables.
@@ -132,7 +132,7 @@ services:
 
   # Frontend web application.
   web:
-    image: langgenius/dify-web:1.11.3
+    image: langgenius/dify-web:1.11.4
     restart: always
     environment:
       CONSOLE_API_URL: ${CONSOLE_API_URL:-}

docker/docker-compose.yaml

@@ -705,7 +705,7 @@ services:
 
   # API service
   api:
-    image: langgenius/dify-api:1.11.3
+    image: langgenius/dify-api:1.11.4
     restart: always
     environment:
       # Use the shared environment variables.
@@ -747,7 +747,7 @@ services:
   # worker service
   # The Celery worker for processing all queues (dataset, workflow, mail, etc.)
   worker:
-    image: langgenius/dify-api:1.11.3
+    image: langgenius/dify-api:1.11.4
     restart: always
     environment:
       # Use the shared environment variables.
@@ -786,7 +786,7 @@ services:
   # worker_beat service
   # Celery beat for scheduling periodic tasks.
   worker_beat:
-    image: langgenius/dify-api:1.11.3
+    image: langgenius/dify-api:1.11.4
     restart: always
     environment:
       # Use the shared environment variables.
@@ -816,7 +816,7 @@ services:
 
   # Frontend web application.
   web:
-    image: langgenius/dify-web:1.11.3
+    image: langgenius/dify-web:1.11.4
     restart: always
     environment:
       CONSOLE_API_URL: ${CONSOLE_API_URL:-}

web/app/components/app/configuration/config-var/config-modal/index.tsx

@@ -21,7 +21,6 @@ import CodeEditor from '@/app/components/workflow/nodes/_base/components/editor/
 import FileUploadSetting from '@/app/components/workflow/nodes/_base/components/file-upload-setting'
 import { CodeLanguage } from '@/app/components/workflow/nodes/code/types'
 import { ChangeType, InputVarType, SupportUploadFileTypes } from '@/app/components/workflow/types'
-import { DEFAULT_VALUE_MAX_LEN } from '@/config'
 import ConfigContext from '@/context/debug-configuration'
 import { AppModeEnum, TransferMethod } from '@/types/app'
 import { checkKeys, getNewVarInWorkflow, replaceSpaceWithUnderscoreInVarNameInput } from '@/utils/var'
@@ -198,8 +197,6 @@ const ConfigModal: FC<IConfigModalProps> = ({
         if (type === InputVarType.multiFiles)
           draft.max_length = DEFAULT_FILE_UPLOAD_SETTING.max_length
       }
-      if (type === InputVarType.paragraph)
-        draft.max_length = DEFAULT_VALUE_MAX_LEN
     })
     setTempPayload(newPayload)
   }, [tempPayload])

web/app/components/app/configuration/config-var/index.tsx

@@ -15,7 +15,6 @@ import Confirm from '@/app/components/base/confirm'
 import Toast from '@/app/components/base/toast'
 import Tooltip from '@/app/components/base/tooltip'
 import { InputVarType } from '@/app/components/workflow/types'
-import { DEFAULT_VALUE_MAX_LEN } from '@/config'
 import ConfigContext from '@/context/debug-configuration'
 import { useEventEmitterContextContext } from '@/context/event-emitter'
 import { useModalContext } from '@/context/modal-context'
@@ -58,8 +57,6 @@ const buildPromptVariableFromInput = (payload: InputVar): PromptVariable => {
     key: variable,
     name: label as string,
   }
-  if (payload.type === InputVarType.textInput)
-    nextItem.max_length = nextItem.max_length || DEFAULT_VALUE_MAX_LEN
 
   if (payload.type !== InputVarType.select)
     delete nextItem.options

web/app/components/app/configuration/debug/chat-user-input.tsx

@@ -7,7 +7,6 @@ import Input from '@/app/components/base/input'
 import Select from '@/app/components/base/select'
 import Textarea from '@/app/components/base/textarea'
 import BoolInput from '@/app/components/workflow/nodes/_base/components/before-run-form/bool-input'
-import { DEFAULT_VALUE_MAX_LEN } from '@/config'
 import ConfigContext from '@/context/debug-configuration'
 import { cn } from '@/utils/classnames'
 
@@ -88,7 +87,7 @@ const ChatUserInput = ({
                     onChange={(e) => { handleInputValueChange(key, e.target.value) }}
                     placeholder={name}
                     autoFocus={index === 0}
-                    maxLength={max_length || DEFAULT_VALUE_MAX_LEN}
+                    maxLength={max_length}
                   />
                 )}
                 {type === 'paragraph' && (
@@ -115,7 +114,7 @@ const ChatUserInput = ({
                     onChange={(e) => { handleInputValueChange(key, e.target.value) }}
                     placeholder={name}
                     autoFocus={index === 0}
-                    maxLength={max_length || DEFAULT_VALUE_MAX_LEN}
+                    maxLength={max_length}
                   />
                 )}
                 {type === 'checkbox' && (

web/app/components/app/configuration/prompt-value-panel/index.tsx

@@ -20,7 +20,6 @@ import Select from '@/app/components/base/select'
 import Textarea from '@/app/components/base/textarea'
 import Tooltip from '@/app/components/base/tooltip'
 import BoolInput from '@/app/components/workflow/nodes/_base/components/before-run-form/bool-input'
-import { DEFAULT_VALUE_MAX_LEN } from '@/config'
 import ConfigContext from '@/context/debug-configuration'
 import { AppModeEnum, ModelModeType } from '@/types/app'
 import { cn } from '@/utils/classnames'
@@ -142,7 +141,7 @@ const PromptValuePanel: FC<IPromptValuePanelProps> = ({
                         onChange={(e) => { handleInputValueChange(key, e.target.value) }}
                         placeholder={name}
                         autoFocus={index === 0}
-                        maxLength={max_length || DEFAULT_VALUE_MAX_LEN}
+                        maxLength={max_length}
                       />
                     )}
                     {type === 'paragraph' && (
@@ -170,7 +169,7 @@ const PromptValuePanel: FC<IPromptValuePanelProps> = ({
                         onChange={(e) => { handleInputValueChange(key, e.target.value) }}
                         placeholder={name}
                         autoFocus={index === 0}
-                        maxLength={max_length || DEFAULT_VALUE_MAX_LEN}
+                        maxLength={max_length}
                       />
                     )}
                     {type === 'checkbox' && (

web/app/components/apps/list.tsx

@@ -12,7 +12,6 @@ import { useDebounceFn } from 'ahooks'
 import dynamic from 'next/dynamic'
 import {
   useRouter,
-  useSearchParams,
 } from 'next/navigation'
 import { parseAsString, useQueryState } from 'nuqs'
 import { useCallback, useEffect, useRef, useState } from 'react'
@@ -29,7 +28,6 @@ import { CheckModal } from '@/hooks/use-pay'
 import { useInfiniteAppList } from '@/service/use-apps'
 import { AppModeEnum } from '@/types/app'
 import { cn } from '@/utils/classnames'
-import { isServer } from '@/utils/client'
 import AppCard from './app-card'
 import { AppCardSkeleton } from './app-card-skeleton'
 import Empty from './empty'
@@ -59,7 +57,6 @@ const List = () => {
   const { t } = useTranslation()
   const { systemFeatures } = useGlobalPublicStore()
   const router = useRouter()
-  const searchParams = useSearchParams()
   const { isCurrentWorkspaceEditor, isCurrentWorkspaceDatasetOperator, isLoadingCurrentWorkspace } = useAppContext()
   const showTagManagementModal = useTagStore(s => s.showTagManagementModal)
   const [activeTab, setActiveTab] = useQueryState(
@@ -67,33 +64,6 @@ const List = () => {
     parseAsString.withDefault('all').withOptions({ history: 'push' }),
   )
 
-  // valid tabs for apps list; anything else should fallback to 'all'
-
-  // 1) Normalize legacy/incorrect query params like ?mode=discover -> ?category=all
-  useEffect(() => {
-    // avoid running on server
-    if (isServer)
-      return
-    const mode = searchParams.get('mode')
-    if (!mode)
-      return
-    const url = new URL(window.location.href)
-    url.searchParams.delete('mode')
-    if (validTabs.has(mode)) {
-      // migrate to category key
-      url.searchParams.set('category', mode)
-    }
-    else {
-      url.searchParams.set('category', 'all')
-    }
-    router.replace(url.pathname + url.search)
-  }, [router, searchParams])
-
-  // 2) If category has an invalid value (e.g., 'discover'), reset to 'all'
-  useEffect(() => {
-    if (!validTabs.has(activeTab))
-      setActiveTab('all')
-  }, [activeTab, setActiveTab])
   const { query: { tagIDs = [], keywords = '', isCreatedByMe: queryIsCreatedByMe = false }, setQuery } = useAppsQueryState()
   const [isCreatedByMe, setIsCreatedByMe] = useState(queryIsCreatedByMe)
   const [tagFilterValue, setTagFilterValue] = useState<string[]>(tagIDs)

web/app/components/datasets/documents/detail/completed/index.tsx

@@ -442,6 +442,10 @@ const Completed: FC<ICompletedProps> = ({
     setFullScreen(!fullScreen)
   }, [fullScreen])
 
+  const toggleCollapsed = useCallback(() => {
+    setIsCollapsed(prev => !prev)
+  }, [])
+
   const viewNewlyAddedChunk = useCallback(async () => {
     const totalPages = segmentListData?.total_pages || 0
     const total = segmentListData?.total || 0
@@ -578,15 +582,16 @@ const Completed: FC<ICompletedProps> = ({
     return selectedStatus ? 1 : 0
   }, [selectedStatus])
 
+  const contextValue = useMemo<SegmentListContextValue>(() => ({
+    isCollapsed,
+    fullScreen,
+    toggleFullScreen,
+    currSegment,
+    currChildChunk,
+  }), [isCollapsed, fullScreen, toggleFullScreen, currSegment, currChildChunk])
+
   return (
-    <SegmentListContext.Provider value={{
-      isCollapsed,
-      fullScreen,
-      toggleFullScreen,
-      currSegment,
-      currChildChunk,
-    }}
-    >
+    <SegmentListContext.Provider value={contextValue}>
       {/* Menu Bar */}
       {!isFullDocMode && (
         <div className={s.docSearchWrapper}>
@@ -618,7 +623,7 @@ const Completed: FC<ICompletedProps> = ({
             onClear={() => handleInputChange('')}
           />
           <Divider type="vertical" className="mx-3 h-3.5" />
-          <DisplayToggle isCollapsed={isCollapsed} toggleCollapsed={() => setIsCollapsed(!isCollapsed)} />
+          <DisplayToggle isCollapsed={isCollapsed} toggleCollapsed={toggleCollapsed} />
         </div>
       )}
       {/* Segment list */}

web/app/components/datasets/documents/detail/completed/segment-card/chunk-content.tsx

@@ -1,4 +1,5 @@
 import type { FC } from 'react'
+import type { SegmentListContextValue } from '..'
 import * as React from 'react'
 import { Markdown } from '@/app/components/base/markdown'
 import { cn } from '@/utils/classnames'
@@ -14,13 +15,15 @@ type ChunkContentProps = {
   className?: string
 }
 
+const selectIsCollapsed = (s: SegmentListContextValue) => s.isCollapsed
+
 const ChunkContent: FC<ChunkContentProps> = ({
   detail,
   isFullDocMode,
   className,
 }) => {
   const { answer, content, sign_content } = detail
-  const isCollapsed = useSegmentListContext(s => s.isCollapsed)
+  const isCollapsed = useSegmentListContext(selectIsCollapsed)
 
   if (answer) {
     return (

web/app/components/header/nav/index.tsx

@@ -2,9 +2,9 @@
 
 import type { INavSelectorProps } from './nav-selector'
 import Link from 'next/link'
-import { usePathname, useSearchParams, useSelectedLayoutSegment } from 'next/navigation'
+import { useSelectedLayoutSegment } from 'next/navigation'
 import * as React from 'react'
-import { useEffect, useState } from 'react'
+import { useState } from 'react'
 import { useStore as useAppStore } from '@/app/components/app/store'
 import { ArrowNarrowLeft } from '@/app/components/base/icons/src/vender/line/arrows'
 import { cn } from '@/utils/classnames'
@@ -36,14 +36,6 @@ const Nav = ({
   const [hovered, setHovered] = useState(false)
   const segment = useSelectedLayoutSegment()
   const isActivated = Array.isArray(activeSegment) ? activeSegment.includes(segment!) : segment === activeSegment
-  const pathname = usePathname()
-  const searchParams = useSearchParams()
-  const [linkLastSearchParams, setLinkLastSearchParams] = useState('')
-
-  useEffect(() => {
-    if (pathname === link)
-      setLinkLastSearchParams(searchParams.toString())
-  }, [pathname, searchParams])
 
   return (
     <div className={`
@@ -52,7 +44,7 @@ const Nav = ({
       ${!curNav && !isActivated && 'hover:bg-components-main-nav-nav-button-bg-hover'}
     `}
     >
-      <Link href={link + (linkLastSearchParams && `?${linkLastSearchParams}`)}>
+      <Link href={link}>
         <div
           onClick={(e) => {
             // Don't clear state if opening in new tab/window

web/app/components/plugins/marketplace/atoms.ts

@@ -1,4 +1,3 @@
-import type { ActivePluginType } from './constants'
 import type { PluginsSort, SearchParamsFromCollection } from './types'
 import { atom, useAtom, useAtomValue, useSetAtom } from 'jotai'
 import { useQueryState } from 'nuqs'
@@ -17,32 +16,14 @@ export function useSetMarketplaceSort() {
   return useSetAtom(marketplaceSortAtom)
 }
 
-/**
- * Preserve the state for marketplace
- */
-export const preserveSearchStateInQueryAtom = atom<boolean>(false)
-
-const searchPluginTextAtom = atom<string>('')
-const activePluginTypeAtom = atom<ActivePluginType>('all')
-const filterPluginTagsAtom = atom<string[]>([])
-
 export function useSearchPluginText() {
-  const preserveSearchStateInQuery = useAtomValue(preserveSearchStateInQueryAtom)
-  const queryState = useQueryState('q', marketplaceSearchParamsParsers.q)
-  const atomState = useAtom(searchPluginTextAtom)
-  return preserveSearchStateInQuery ? queryState : atomState
+  return useQueryState('q', marketplaceSearchParamsParsers.q)
 }
 export function useActivePluginType() {
-  const preserveSearchStateInQuery = useAtomValue(preserveSearchStateInQueryAtom)
-  const queryState = useQueryState('category', marketplaceSearchParamsParsers.category)
-  const atomState = useAtom(activePluginTypeAtom)
-  return preserveSearchStateInQuery ? queryState : atomState
+  return useQueryState('category', marketplaceSearchParamsParsers.category)
 }
 export function useFilterPluginTags() {
-  const preserveSearchStateInQuery = useAtomValue(preserveSearchStateInQueryAtom)
-  const queryState = useQueryState('tags', marketplaceSearchParamsParsers.tags)
-  const atomState = useAtom(filterPluginTagsAtom)
-  return preserveSearchStateInQuery ? queryState : atomState
+  return useQueryState('tags', marketplaceSearchParamsParsers.tags)
 }
 
 /**

web/app/components/plugins/marketplace/hydration-client.tsx

@@ -1,15 +0,0 @@
-'use client'
-
-import { useHydrateAtoms } from 'jotai/utils'
-import { preserveSearchStateInQueryAtom } from './atoms'
-
-export function HydrateMarketplaceAtoms({
-  preserveSearchStateInQuery,
-  children,
-}: {
-  preserveSearchStateInQuery: boolean
-  children: React.ReactNode
-}) {
-  useHydrateAtoms([[preserveSearchStateInQueryAtom, preserveSearchStateInQuery]])
-  return <>{children}</>
-}

web/app/components/plugins/marketplace/index.tsx

@@ -1,7 +1,6 @@
 import type { SearchParams } from 'nuqs'
 import { TanstackQueryInitializer } from '@/context/query-client'
 import Description from './description'
-import { HydrateMarketplaceAtoms } from './hydration-client'
 import { HydrateQueryClient } from './hydration-server'
 import ListWrapper from './list/list-wrapper'
 import StickySearchAndSwitchWrapper from './sticky-search-and-switch-wrapper'
@@ -10,8 +9,7 @@ type MarketplaceProps = {
   showInstallButton?: boolean
   pluginTypeSwitchClassName?: string
   /**
-   * Pass the search params from the request to prefetch data on the server
-   * and preserve the search params in the URL.
+   * Pass the search params from the request to prefetch data on the server.
    */
   searchParams?: Promise<SearchParams>
 }
@@ -24,15 +22,13 @@ const Marketplace = async ({
   return (
     <TanstackQueryInitializer>
       <HydrateQueryClient searchParams={searchParams}>
-        <HydrateMarketplaceAtoms preserveSearchStateInQuery={!!searchParams}>
-          <Description />
-          <StickySearchAndSwitchWrapper
-            pluginTypeSwitchClassName={pluginTypeSwitchClassName}
-          />
-          <ListWrapper
-            showInstallButton={showInstallButton}
-          />
-        </HydrateMarketplaceAtoms>
+        <Description />
+        <StickySearchAndSwitchWrapper
+          pluginTypeSwitchClassName={pluginTypeSwitchClassName}
+        />
+        <ListWrapper
+          showInstallButton={showInstallButton}
+        />
       </HydrateQueryClient>
     </TanstackQueryInitializer>
   )

web/app/components/plugins/plugin-page/context.tsx

@@ -68,7 +68,7 @@ export const PluginPageContextProvider = ({
   const options = useMemo(() => {
     return enable_marketplace ? tabs : tabs.filter(tab => tab.value !== PLUGIN_PAGE_TABS_MAP.marketplace)
   }, [tabs, enable_marketplace])
-  const [activeTab, setActiveTab] = useQueryState('category', {
+  const [activeTab, setActiveTab] = useQueryState('tab', {
     defaultValue: options[0].value,
   })
 

web/app/components/rag-pipeline/components/panel/input-field/editor/form/hooks.ts

@@ -6,7 +6,6 @@ import { useTranslation } from 'react-i18next'
 import { useFileSizeLimit } from '@/app/components/base/file-uploader/hooks'
 import { InputFieldType } from '@/app/components/base/form/form-scenarios/input-field/types'
 import { DEFAULT_FILE_UPLOAD_SETTING } from '@/app/components/workflow/constants'
-import { DEFAULT_VALUE_MAX_LEN } from '@/config'
 import { PipelineInputVarType } from '@/models/pipeline'
 import { useFileUploadConfig } from '@/service/use-common'
 import { formatFileSize } from '@/utils/format'
@@ -87,8 +86,6 @@ export const useConfigurations = (props: {
       if (type === PipelineInputVarType.multiFiles)
         setFieldValue('maxLength', DEFAULT_FILE_UPLOAD_SETTING.max_length)
     }
-    if (type === PipelineInputVarType.paragraph)
-      setFieldValue('maxLength', DEFAULT_VALUE_MAX_LEN)
   }, [setFieldValue])
 
   const handleVariableNameBlur = useCallback((value: string) => {

web/app/components/rag-pipeline/components/panel/input-field/editor/form/index.spec.tsx

@@ -779,27 +779,6 @@ describe('useConfigurations', () => {
       expect(mockSetFieldValue).toHaveBeenCalledWith('maxLength', expect.any(Number))
     })
 
-    it('should call setFieldValue when type changes to paragraph', () => {
-      // Arrange
-      const mockGetFieldValue = vi.fn()
-      const mockSetFieldValue = vi.fn()
-
-      const { result } = renderHookWithProviders(() =>
-        useConfigurations({
-          getFieldValue: mockGetFieldValue,
-          setFieldValue: mockSetFieldValue,
-          supportFile: false,
-        }),
-      )
-
-      // Act
-      const typeConfig = result.current.find(config => config.variable === 'type')
-      typeConfig?.listeners?.onChange?.(createMockEvent(PipelineInputVarType.paragraph))
-
-      // Assert
-      expect(mockSetFieldValue).toHaveBeenCalledWith('maxLength', 48) // DEFAULT_VALUE_MAX_LEN
-    })
-
     it('should set label from variable name on blur when label is empty', () => {
       // Arrange
       const mockGetFieldValue = vi.fn().mockReturnValue('')

web/app/components/share/text-generation/index.tsx

@@ -26,7 +26,7 @@ import DifyLogo from '@/app/components/base/logo/dify-logo'
 import Toast from '@/app/components/base/toast'
 import Res from '@/app/components/share/text-generation/result'
 import RunOnce from '@/app/components/share/text-generation/run-once'
-import { appDefaultIconBackground, BATCH_CONCURRENCY, DEFAULT_VALUE_MAX_LEN } from '@/config'
+import { appDefaultIconBackground, BATCH_CONCURRENCY } from '@/config'
 import { useGlobalPublicStore } from '@/context/global-public-context'
 import { useWebAppStore } from '@/context/web-app-context'
 import { useAppFavicon } from '@/hooks/use-app-favicon'
@@ -256,11 +256,10 @@ const TextGeneration: FC<IMainProps> = ({
       promptConfig?.prompt_variables.forEach((varItem, varIndex) => {
         if (errorRowIndex !== 0)
           return
-        if (varItem.type === 'string') {
-          const maxLen = varItem.max_length || DEFAULT_VALUE_MAX_LEN
-          if (item[varIndex].length > maxLen) {
+        if (varItem.type === 'string' && varItem.max_length) {
+          if (item[varIndex].length > varItem.max_length) {
             moreThanMaxLengthVarName = varItem.name
-            maxLength = maxLen
+            maxLength = varItem.max_length
             errorRowIndex = index + 1
             return
           }

web/app/components/share/text-generation/run-once/index.spec.tsx

@@ -236,4 +236,46 @@ describe('RunOnce', () => {
     const stopButton = screen.getByTestId('stop-button')
     expect(stopButton).toBeDisabled()
   })
+
+  describe('maxLength behavior', () => {
+    it('should not have maxLength attribute when max_length is not set', async () => {
+      const promptConfig: PromptConfig = {
+        prompt_template: 'template',
+        prompt_variables: [
+          createPromptVariable({
+            key: 'textInput',
+            name: 'Text Input',
+            type: 'string',
+            // max_length is not set
+          }),
+        ],
+      }
+      const { onInputsChange } = setup({ promptConfig, visionConfig: { ...baseVisionConfig, enabled: false } })
+      await waitFor(() => {
+        expect(onInputsChange).toHaveBeenCalled()
+      })
+      const input = screen.getByPlaceholderText('Text Input')
+      expect(input).not.toHaveAttribute('maxLength')
+    })
+
+    it('should have maxLength attribute when max_length is set', async () => {
+      const promptConfig: PromptConfig = {
+        prompt_template: 'template',
+        prompt_variables: [
+          createPromptVariable({
+            key: 'textInput',
+            name: 'Text Input',
+            type: 'string',
+            max_length: 100,
+          }),
+        ],
+      }
+      const { onInputsChange } = setup({ promptConfig, visionConfig: { ...baseVisionConfig, enabled: false } })
+      await waitFor(() => {
+        expect(onInputsChange).toHaveBeenCalled()
+      })
+      const input = screen.getByPlaceholderText('Text Input')
+      expect(input).toHaveAttribute('maxLength', '100')
+    })
+  })
 })

web/app/components/share/text-generation/run-once/index.tsx

@@ -19,7 +19,6 @@ import Textarea from '@/app/components/base/textarea'
 import BoolInput from '@/app/components/workflow/nodes/_base/components/before-run-form/bool-input'
 import CodeEditor from '@/app/components/workflow/nodes/_base/components/editor/code-editor'
 import { CodeLanguage } from '@/app/components/workflow/nodes/code/types'
-import { DEFAULT_VALUE_MAX_LEN } from '@/config'
 import useBreakpoints, { MediaType } from '@/hooks/use-breakpoints'
 import { cn } from '@/utils/classnames'
 
@@ -140,7 +139,7 @@ const RunOnce: FC<IRunOnceProps> = ({
                         placeholder={item.name}
                         value={inputs[item.key]}
                         onChange={(e: ChangeEvent<HTMLInputElement>) => { handleInputsChange({ ...inputsRef.current, [item.key]: e.target.value }) }}
-                        maxLength={item.max_length || DEFAULT_VALUE_MAX_LEN}
+                        maxLength={item.max_length}
                       />
                     )}
                     {item.type === 'paragraph' && (

web/app/components/workflow/nodes/_base/components/retry/retry-on-panel.tsx

@@ -9,10 +9,6 @@ import Split from '@/app/components/workflow/nodes/_base/components/split'
 import { useRetryConfig } from './hooks'
 import s from './style.module.css'
 
-// Nodes that support first token timeout configuration
-// These nodes internally call LLM and have streaming response characteristics
-const LLM_RELATED_NODE_TYPES = ['llm', 'agent', 'parameter-extractor', 'question-classifier']
-
 type RetryOnPanelProps = Pick<Node, 'id' | 'data'>
 const RetryOnPanel = ({
   id,
@@ -20,14 +16,10 @@ const RetryOnPanel = ({
 }: RetryOnPanelProps) => {
   const { t } = useTranslation()
   const { handleRetryConfigChange } = useRetryConfig(id)
-  const { retry_config, type } = data
-
-  // Check if this is an LLM-related node that supports first token timeout
-  const isLLMRelatedNode = LLM_RELATED_NODE_TYPES.includes(type)
+  const { retry_config } = data
 
   const handleRetryEnabledChange = (value: boolean) => {
     handleRetryConfigChange({
-      ...retry_config,
       retry_enabled: value,
       max_retries: retry_config?.max_retries || 3,
       retry_interval: retry_config?.retry_interval || 1000,
@@ -40,7 +32,6 @@ const RetryOnPanel = ({
     else if (value < 1)
       value = 1
     handleRetryConfigChange({
-      ...retry_config,
       retry_enabled: true,
       max_retries: value,
       retry_interval: retry_config?.retry_interval || 1000,
@@ -53,27 +44,12 @@ const RetryOnPanel = ({
     else if (value < 100)
       value = 100
     handleRetryConfigChange({
-      ...retry_config,
       retry_enabled: true,
       max_retries: retry_config?.max_retries || 3,
       retry_interval: value,
     })
   }
 
-  const handleFirstTokenTimeoutChange = (value: number) => {
-    if (value > 60000)
-      value = 60000
-    else if (value < 0)
-      value = 0
-    handleRetryConfigChange({
-      ...retry_config,
-      retry_enabled: true,
-      max_retries: retry_config?.max_retries || 3,
-      retry_interval: retry_config?.retry_interval || 1000,
-      first_token_timeout: value,
-    })
-  }
-
   return (
     <>
       <div className="pt-2">
@@ -86,76 +62,54 @@ const RetryOnPanel = ({
             onChange={v => handleRetryEnabledChange(v)}
           />
         </div>
-        {retry_config?.retry_enabled && (
-          <div className="px-4 pb-2">
-            <div className="mb-1 flex w-full items-center">
-              <div className="system-xs-medium-uppercase mr-2 grow text-text-secondary">{t('nodes.common.retry.maxRetries', { ns: 'workflow' })}</div>
-              <Slider
-                className="mr-3 w-[108px]"
-                value={retry_config?.max_retries || 3}
-                onChange={handleMaxRetriesChange}
-                min={1}
-                max={10}
-              />
-              <Input
-                type="number"
-                wrapperClassName="w-[100px]"
-                value={retry_config?.max_retries || 3}
-                onChange={e =>
-                  handleMaxRetriesChange(Number.parseInt(e.currentTarget.value, 10) || 3)}
-                min={1}
-                max={10}
-                unit={t('nodes.common.retry.times', { ns: 'workflow' }) || ''}
-                className={s.input}
-              />
-            </div>
-            <div className="mb-1 flex w-full items-center">
-              <div className="system-xs-medium-uppercase mr-2 grow text-text-secondary">{t('nodes.common.retry.retryInterval', { ns: 'workflow' })}</div>
-              <Slider
-                className="mr-3 w-[108px]"
-                value={retry_config?.retry_interval || 1000}
-                onChange={handleRetryIntervalChange}
-                min={100}
-                max={5000}
-              />
-              <Input
-                type="number"
-                wrapperClassName="w-[100px]"
-                value={retry_config?.retry_interval || 1000}
-                onChange={e =>
-                  handleRetryIntervalChange(Number.parseInt(e.currentTarget.value, 10) || 1000)}
-                min={100}
-                max={5000}
-                unit={t('nodes.common.retry.ms', { ns: 'workflow' }) || ''}
-                className={s.input}
-              />
-            </div>
-            {/* First token timeout - only for LLM-related nodes */}
-            {isLLMRelatedNode && (
-              <div className="flex w-full items-center">
-                <div className="system-xs-medium-uppercase mr-2 grow text-text-secondary">{t('nodes.common.retry.firstTokenTimeout', { ns: 'workflow' })}</div>
+        {
+          retry_config?.retry_enabled && (
+            <div className="px-4 pb-2">
+              <div className="mb-1 flex w-full items-center">
+                <div className="system-xs-medium-uppercase mr-2 grow text-text-secondary">{t('nodes.common.retry.maxRetries', { ns: 'workflow' })}</div>
                 <Slider
                   className="mr-3 w-[108px]"
-                  value={retry_config?.first_token_timeout ?? 3000}
-                  onChange={handleFirstTokenTimeoutChange}
-                  min={0}
-                  max={60000}
+                  value={retry_config?.max_retries || 3}
+                  onChange={handleMaxRetriesChange}
+                  min={1}
+                  max={10}
                 />
                 <Input
                   type="number"
                   wrapperClassName="w-[100px]"
-                  value={retry_config?.first_token_timeout ?? 3000}
+                  value={retry_config?.max_retries || 3}
                   onChange={e =>
-                    handleFirstTokenTimeoutChange(Number.parseInt(e.currentTarget.value, 10) || 0)}
-                  min={0}
-                  max={60000}
+                    handleMaxRetriesChange(Number.parseInt(e.currentTarget.value, 10) || 3)}
+                  min={1}
+                  max={10}
+                  unit={t('nodes.common.retry.times', { ns: 'workflow' }) || ''}
+                  className={s.input}
+                />
+              </div>
+              <div className="flex items-center">
+                <div className="system-xs-medium-uppercase mr-2 grow text-text-secondary">{t('nodes.common.retry.retryInterval', { ns: 'workflow' })}</div>
+                <Slider
+                  className="mr-3 w-[108px]"
+                  value={retry_config?.retry_interval || 1000}
+                  onChange={handleRetryIntervalChange}
+                  min={100}
+                  max={5000}
+                />
+                <Input
+                  type="number"
+                  wrapperClassName="w-[100px]"
+                  value={retry_config?.retry_interval || 1000}
+                  onChange={e =>
+                    handleRetryIntervalChange(Number.parseInt(e.currentTarget.value, 10) || 1000)}
+                  min={100}
+                  max={5000}
                   unit={t('nodes.common.retry.ms', { ns: 'workflow' }) || ''}
                   className={s.input}
                 />
               </div>
-            )}
-          </div>
-        )}
+            </div>
+          )
+        }
       </div>
       <Split className="mx-4 mt-2" />
     </>

web/app/components/workflow/nodes/_base/components/retry/types.ts

@@ -2,6 +2,4 @@ export type WorkflowRetryConfig = {
   max_retries: number
   retry_interval: number
   retry_enabled: boolean
-  // First token timeout for LLM nodes (seconds), 0 means no timeout
-  first_token_timeout?: number
 }

web/config/index.ts

@@ -208,7 +208,6 @@ export const VAR_ITEM_TEMPLATE = {
   key: '',
   name: '',
   type: 'string',
-  max_length: DEFAULT_VALUE_MAX_LEN,
   required: true,
 }
 
@@ -216,7 +215,6 @@ export const VAR_ITEM_TEMPLATE_IN_WORKFLOW = {
   variable: '',
   label: '',
   type: InputVarType.textInput,
-  max_length: DEFAULT_VALUE_MAX_LEN,
   required: true,
   options: [],
 }
@@ -225,7 +223,6 @@ export const VAR_ITEM_TEMPLATE_IN_PIPELINE = {
   variable: '',
   label: '',
   type: PipelineInputVarType.textInput,
-  max_length: DEFAULT_VALUE_MAX_LEN,
   required: true,
   options: [],
 }

web/eslint.config.mjs

@@ -26,7 +26,8 @@ export default antfu(
         'react-hooks/preserve-manual-memoization': 'warn',
         'react-hooks/purity': 'warn',
         'react-hooks/refs': 'warn',
-        'react-hooks/set-state-in-effect': 'warn',
+        // prefer react-hooks-extra/no-direct-set-state-in-use-effect
+        'react-hooks/set-state-in-effect': 'off',
         'react-hooks/set-state-in-render': 'warn',
         'react-hooks/static-components': 'warn',
         'react-hooks/unsupported-syntax': 'warn',
@@ -53,6 +54,14 @@ export default antfu(
       },
     },
   },
+  {
+    files: ['**/*.ts', '**/*.tsx'],
+    settings: {
+      'react-x': {
+        additionalStateHooks: '/^use\\w*State(?:s)?|useAtom$/u',
+      },
+    },
+  },
   // downgrade some rules from error to warn for gradual adoption
   // we should fix these in following pull requests
   {

web/i18n/en-US/workflow.json

@@ -433,7 +433,6 @@
   "nodes.common.memory.windowSize": "Window Size",
   "nodes.common.outputVars": "Output Variables",
   "nodes.common.pluginNotInstalled": "Plugin is not installed",
-  "nodes.common.retry.firstTokenTimeout": "First token timeout",
   "nodes.common.retry.maxRetries": "max retries",
   "nodes.common.retry.ms": "ms",
   "nodes.common.retry.retries": "{{num}} Retries",
@@ -445,8 +444,6 @@
   "nodes.common.retry.retrySuccessful": "Retry successful",
   "nodes.common.retry.retryTimes": "Retry {{times}} times on failure",
   "nodes.common.retry.retrying": "Retrying...",
-  "nodes.common.retry.seconds": "s",
-  "nodes.common.retry.timeoutDuration": "Timeout duration",
   "nodes.common.retry.times": "times",
   "nodes.common.typeSwitch.input": "Input value",
   "nodes.common.typeSwitch.variable": "Use variable",

web/i18n/zh-Hans/workflow.json

@@ -433,7 +433,6 @@
   "nodes.common.memory.windowSize": "记忆窗口",
   "nodes.common.outputVars": "输出变量",
   "nodes.common.pluginNotInstalled": "插件未安装",
-  "nodes.common.retry.firstTokenTimeout": "首个 Token 超时",
   "nodes.common.retry.maxRetries": "最大重试次数",
   "nodes.common.retry.ms": "毫秒",
   "nodes.common.retry.retries": "{{num}} 重试次数",
@@ -445,8 +444,6 @@
   "nodes.common.retry.retrySuccessful": "重试成功",
   "nodes.common.retry.retryTimes": "失败时重试 {{times}} 次",
   "nodes.common.retry.retrying": "重试中...",
-  "nodes.common.retry.seconds": "秒",
-  "nodes.common.retry.timeoutDuration": "超时时长",
   "nodes.common.retry.times": "次",
   "nodes.common.typeSwitch.input": "输入值",
   "nodes.common.typeSwitch.variable": "使用变量",

web/package.json

@@ -1,7 +1,7 @@
 {
   "name": "dify-web",
   "type": "module",
-  "version": "1.11.3",
+  "version": "1.11.4",
   "private": true,
   "packageManager": "[email protected]+sha512.72d699da16b1179c14ba9e64dc71c9a40988cbdc65c264cb0e489db7de917f20dcf4d64d8723625f2969ba52d4b7e2a1170682d9ac2a5dcaeaab732b7e16f04a",
   "imports": {
@@ -28,9 +28,10 @@
     "build:docker": "next build && node scripts/optimize-standalone.js",
     "start": "node ./scripts/copy-and-start.mjs",
     "lint": "eslint --cache --cache-location node_modules/.cache/eslint/.eslint-cache",
-    "lint:fix": "eslint --cache --cache-location node_modules/.cache/eslint/.eslint-cache --fix",
-    "lint:quiet": "eslint --cache --cache-location node_modules/.cache/eslint/.eslint-cache --quiet",
-    "lint:complexity": "eslint --cache --cache-location node_modules/.cache/eslint/.eslint-cache --rule 'complexity: [error, {max: 15}]' --quiet",
+    "lint:fix": "pnpm lint --fix",
+    "lint:quiet": "pnpm lint --quiet",
+    "lint:complexity": "pnpm lint --rule 'complexity: [error, {max: 15}]' --quiet",
+    "lint:report": "pnpm lint --output-file eslint_report.json --format json",
     "type-check": "tsc --noEmit",
     "type-check:tsgo": "tsgo --noEmit",
     "prepare": "cd ../ && node -e \"if (process.env.NODE_ENV !== 'production'){process.exit(1)} \" || husky ./web/.husky",
@@ -152,9 +153,9 @@
     "zustand": "^5.0.9"
   },
   "devDependencies": {
-    "@antfu/eslint-config": "^6.7.3",
+    "@antfu/eslint-config": "^7.0.1",
     "@chromatic-com/storybook": "^4.1.1",
-    "@eslint-react/eslint-plugin": "^2.3.13",
+    "@eslint-react/eslint-plugin": "^2.7.0",
     "@mdx-js/loader": "^3.1.1",
     "@mdx-js/react": "^3.1.1",
     "@next/bundle-analyzer": "15.5.9",
@@ -189,7 +190,7 @@
     "@types/semver": "^7.7.1",
     "@types/sortablejs": "^1.15.8",
     "@types/uuid": "^10.0.0",
-    "@typescript-eslint/parser": "^8.50.0",
+    "@typescript-eslint/parser": "^8.53.0",
     "@typescript/native-preview": "^7.0.0-dev",
     "@vitejs/plugin-react": "^5.1.2",
     "@vitest/coverage-v8": "4.0.16",
@@ -201,7 +202,7 @@
     "eslint-plugin-react-hooks": "^7.0.1",
     "eslint-plugin-react-refresh": "^0.4.26",
     "eslint-plugin-sonarjs": "^3.0.5",
-    "eslint-plugin-storybook": "^10.1.10",
+    "eslint-plugin-storybook": "^10.1.11",
     "eslint-plugin-tailwindcss": "^3.18.2",
     "husky": "^9.1.7",
     "jsdom": "^27.3.0",
@@ -224,7 +225,6 @@
   },
   "pnpm": {
     "overrides": {
-      "@eslint/plugin-kit@<0.3.4": "0.3.4",
       "@monaco-editor/loader": "1.5.0",
       "@nolyfill/safe-buffer": "npm:safe-buffer@^5.2.1",
       "array-includes": "npm:@nolyfill/array-includes@^1",
@@ -275,7 +275,6 @@
     ]
   },
   "resolutions": {
-    "@eslint/plugin-kit": "~0.3",
     "@types/react": "~19.2.7",
     "@types/react-dom": "~19.2.3",
     "brace-expansion": "~2.0",

web/utils/var.ts

@@ -30,7 +30,7 @@ export const getNewVar = (key: string, type: string) => {
 }
 
 export const getNewVarInWorkflow = (key: string, type = InputVarType.textInput): InputVar => {
-  const { max_length: _maxLength, ...rest } = VAR_ITEM_TEMPLATE_IN_WORKFLOW
+  const { ...rest } = VAR_ITEM_TEMPLATE_IN_WORKFLOW
   if (type !== InputVarType.textInput) {
     return {
       ...rest,