feat(workflow): execute workflow tools as nested containers

api/controllers/console/workspace/tool_providers.py

@@ -61,12 +61,10 @@
     ToolProviderType,
     WorkflowToolParameterConfiguration,
 )
-from core.tools.errors import WorkflowToolHumanInputNotSupportedError
 from enums import DeploymentEdition
 from extensions.ext_application_services import application_services
 from extensions.ext_database import db
 from fields.base import ResponseModel
-from libs.external_api import error_responses
 from libs.helper import alphanumeric, dump_response, uuid_value
 from libs.login import login_required
 from models import Account
@@ -902,7 +900,6 @@ def post(self, req_data: ApiToolTestPayload, current_tenant_id: str):
 
 @console_ns.route("/workspaces/current/tool-provider/workflow/create")
 class ToolWorkflowProviderCreateApi(Resource):
-    @error_responses(WorkflowToolHumanInputNotSupportedError)
     @console_ns.expect(console_ns.models[WorkflowToolCreatePayload.__name__])
     @console_ns.response(200, "Workflow tool created successfully", console_ns.models[SimpleResultResponse.__name__])
     @setup_required
@@ -933,7 +930,6 @@ def post(self, req_data: WorkflowToolCreatePayload, tenant_id: str, user: Accoun
 
 @console_ns.route("/workspaces/current/tool-provider/workflow/update")
 class ToolWorkflowProviderUpdateApi(Resource):
-    @error_responses(WorkflowToolHumanInputNotSupportedError)
     @console_ns.expect(console_ns.models[WorkflowToolUpdatePayload.__name__])
     @console_ns.response(200, "Workflow tool updated successfully", console_ns.models[SimpleResultResponse.__name__])
     @setup_required

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

@@ -108,6 +108,7 @@ def generate(
         triggered_from: WorkflowRunTriggeredFrom | None = None,
         root_node_id: str | None = None,
         graph_engine_layers: Sequence[Layer] = (),
+        allow_human_input: bool = True,
         pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Generator[Mapping[str, Any] | str]: ...
 
@@ -126,6 +127,7 @@ def generate(
         triggered_from: WorkflowRunTriggeredFrom | None = None,
         root_node_id: str | None = None,
         graph_engine_layers: Sequence[Layer] = (),
+        allow_human_input: bool = True,
         pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Mapping[str, Any]: ...
 
@@ -144,6 +146,7 @@ def generate(
         triggered_from: WorkflowRunTriggeredFrom | None = None,
         root_node_id: str | None = None,
         graph_engine_layers: Sequence[Layer] = (),
+        allow_human_input: bool = True,
         pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Mapping[str, Any] | Generator[Mapping[str, Any] | str]: ...
 
@@ -161,6 +164,7 @@ def generate(
         triggered_from: WorkflowRunTriggeredFrom | None = None,
         root_node_id: str | None = None,
         graph_engine_layers: Sequence[Layer] = (),
+        allow_human_input: bool = True,
         pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Mapping[str, Any] | Generator[Mapping[str, Any] | str]:
         with self._bind_file_access_scope(tenant_id=app_model.tenant_id, user=user, invoke_from=invoke_from):
@@ -222,6 +226,7 @@ def generate(
                 user_id=user.id,
                 stream=streaming,
                 invoke_from=invoke_from,
+                allow_human_input=allow_human_input,
                 call_depth=call_depth,
                 trace_manager=trace_manager,
                 workflow_execution_id=workflow_run_id,

api/core/app/apps/workflow_app_runner.py

@@ -54,6 +54,7 @@
     resolve_workflow_node_class,
 )
 from core.workflow.nodes.agent.events import NodeRunAgentLogEvent
+from core.workflow.nodes.human_input.boundary import resolve_human_input_node_id
 from core.workflow.nodes.human_input.callback import DifyHITLCallback
 from core.workflow.nodes.human_input.enums import HumanInputFormStatus
 from core.workflow.nodes.human_input.pause_reason import HumanInputRequired, PauseReason
@@ -536,9 +537,10 @@ def publish_human_input_results(
         published_form_ids: set[str],
     ) -> None:
         """Publish materialized completions supplied by run orchestration."""
+        variable_pool = workflow_entry.graph_engine.runtime_state.variable_pool
         for form_id, reason in pending_forms.items():
             form = forms[form_id]
-            node_id = reason.node_id
+            node_id = resolve_human_input_node_id(node_id=reason.node_id, form_id=form_id, variable_pool=variable_pool)
             node_title = reason.node_title or form.definition.node_title or form.node_id
             if form.status == HumanInputFormStatus.TIMEOUT:
                 self._publish_event(
@@ -603,7 +605,11 @@ def handle_event(
                 self._publish_event(
                     QueueHumanInputFormFilledEvent(
                         form_id=event.id,
-                        node_id=event.node_id,
+                        node_id=resolve_human_input_node_id(
+                            node_id=event.node_id,
+                            form_id=event.id,
+                            variable_pool=workflow_entry.graph_engine.runtime_state.variable_pool,
+                        ),
                         node_type=event.node_type,
                         node_title=event.node_title,
                         rendered_content=event.rendered_content,
@@ -616,7 +622,11 @@ def handle_event(
                 self._publish_event(
                     QueueHumanInputFormTimeoutEvent(
                         form_id=event.id,
-                        node_id=event.node_id,
+                        node_id=resolve_human_input_node_id(
+                            node_id=event.node_id,
+                            form_id=event.id,
+                            variable_pool=workflow_entry.graph_engine.runtime_state.variable_pool,
+                        ),
                         node_type=event.node_type,
                         node_title=event.node_title,
                         expiration_time=event.expiration_time,

api/core/app/entities/app_invoke_entities.py

@@ -104,6 +104,7 @@ class DifyRunContext(BaseModel):
     app_type: CreditUsageAppType | None = None
     created_by: CreditUsageCreatedBy | None = None
     trace_session_id: str | None = None
+    workflow_tool_invocation_id: str | None = None
 
     @field_validator("created_by", mode="before")
     @classmethod
@@ -345,6 +346,8 @@ class WorkflowAppGenerateEntity(AppGenerateEntity):
     workflow_execution_timing: WorkflowExecutionTiming = Field(
         default_factory=WorkflowExecutionTiming, exclude=True, repr=False
     )
+    # Legacy synchronous Tool callers cannot resume a paused execution.
+    allow_human_input: bool = True
 
     class SingleIterationRunEntity(BaseModel):
         """

api/core/repositories/celery_workflow_node_execution_repository.py

@@ -7,7 +7,7 @@
 
 import logging
 from collections.abc import Sequence
-from typing import override
+from typing import Any, override
 
 from sqlalchemy.engine import Engine
 from sqlalchemy.orm import sessionmaker
@@ -20,10 +20,12 @@
     SQLAlchemyWorkflowNodeExecutionRepository,
 )
 from graphon.entities import WorkflowNodeExecution
+from graphon.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from models import Account, CreatorUserRole, EndUser
 from models.workflow import WorkflowNodeExecutionTriggeredFrom
 from tasks.workflow_node_execution_tasks import (
     save_workflow_node_execution_task,
+    save_workflow_node_executions_task,
 )
 
 logger = logging.getLogger(__name__)
@@ -94,6 +96,7 @@ def __init__(
         # Extract user context
         self._triggered_from = triggered_from
         self._creator_user_id = user.id
+        self._user = user
 
         # Determine user role based on user type
         self._creator_user_role = CreatorUserRole.ACCOUNT if isinstance(user, Account) else CreatorUserRole.END_USER
@@ -119,6 +122,16 @@ def __init__(
             self._triggered_from,
         )
 
+    @override
+    def for_workflow_tool(self, app_id: str) -> "CeleryWorkflowNodeExecutionRepository":
+        return CeleryWorkflowNodeExecutionRepository(
+            session_factory=self._session_factory,
+            tenant_id=self._tenant_id,
+            user=self._user,
+            app_id=app_id,
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_TOOL,
+        )
+
     @override
     def save(self, execution: WorkflowNodeExecution):
         """
@@ -142,7 +155,7 @@ def save(self, execution: WorkflowNodeExecution):
                     self._workflow_execution_mapping[execution.workflow_execution_id].append(execution.id)
 
             # Serialize execution for Celery task
-            execution_data = execution.model_dump()
+            execution_data = self._serialize_execution(execution)
 
             # Queue the save operation as a Celery task (fire and forget)
             save_workflow_node_execution_task.delay(
@@ -162,6 +175,37 @@ def save(self, execution: WorkflowNodeExecution):
             # For now, we'll re-raise the exception
             raise
 
+    @override
+    def save_many(self, executions: Sequence[WorkflowNodeExecution]) -> None:
+        """Cache exact records and queue one storage task for the whole batch."""
+        if not executions:
+            return
+        for execution in executions:
+            self._execution_cache[execution.id] = execution
+            if execution.workflow_execution_id:
+                execution_ids = self._workflow_execution_mapping.setdefault(execution.workflow_execution_id, [])
+                if execution.id not in execution_ids:
+                    execution_ids.append(execution.id)
+        save_workflow_node_executions_task.delay(
+            executions_data=[self._serialize_execution(execution) for execution in executions],
+            tenant_id=self._tenant_id,
+            app_id=self._app_id or "",
+            triggered_from=self._triggered_from.value if self._triggered_from else "",
+            creator_user_id=self._creator_user_id,
+            creator_user_role=self._creator_user_role.value,
+        )
+
+    @staticmethod
+    def _serialize_execution(execution: WorkflowNodeExecution) -> dict[str, Any]:
+        """Project runtime values before model serialization erases their segment types."""
+        converter = WorkflowRuntimeTypeConverter()
+        return {
+            **execution.model_dump(exclude={"inputs", "process_data", "outputs"}),
+            "inputs": converter.to_json_encodable(execution.inputs),
+            "process_data": converter.to_json_encodable(execution.process_data),
+            "outputs": converter.to_json_encodable(execution.outputs),
+        }
+
     @override
     def save_execution_data(self, execution: WorkflowNodeExecution) -> None:
         """`save` already queues the complete inputs, process data, and outputs."""

api/core/repositories/factory.py

@@ -37,8 +37,16 @@ def save_synchronously(self, execution: WorkflowExecution) -> None:
 
 
 class WorkflowNodeExecutionRepository(Protocol):
+    def for_workflow_tool(self, app_id: str) -> "WorkflowNodeExecutionRepository":
+        """Scope hidden Workflow Tool executions to their source app in the same tenant."""
+        ...
+
     def save(self, execution: WorkflowNodeExecution): ...
 
+    def save_many(self, executions: Sequence[WorkflowNodeExecution]) -> None:
+        """Save exact records in this owner scope without uploading execution data."""
+        ...
+
     def save_synchronously(self, execution: WorkflowNodeExecution) -> None: ...
 
     def save_execution_data(self, execution: WorkflowNodeExecution): ...
@@ -49,7 +57,9 @@ def get_by_workflow_execution(
         order_config: OrderConfig | None = None,
         *,
         include_paused: bool = False,
-    ) -> Sequence[WorkflowNodeExecution]: ...
+    ) -> Sequence[WorkflowNodeExecution]:
+        """Read executions in this repository's app/origin scope, optionally including paused nodes."""
+        ...
 
 
 class RepositoryImportError(Exception):