feat(workflow): expose authorized nested tool traces

Query child executions by their structured caller identity and require source-app permissions before exposing internal execution data.

api/controllers/console/app/workflow_run.py

@@ -4,10 +4,11 @@ from uuid import UUID
 from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
 from controllers.common.errors import NotFoundError
-from controllers.common.rbac import PlainApp, RBACCheck
+from controllers.common.rbac import PlainApp, RBACCheck, enforce_rbac_checks
 from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
 from controllers.common.session import with_session
 from controllers.console import console_ns
@@ -26,6 +27,7 @@ from fields.workflow_run_fields import (
     WorkflowRunNodeExecutionListResponse,
     WorkflowRunNodeExecutionResponse,
     WorkflowRunPaginationResponse,
+    WorkflowToolNodeExecutionListResponse,
     workflow_run_pagination_response_source,
     workflow_run_response_source,
 )
@@ -127,6 +129,7 @@ register_response_schema_models(
     WorkflowRunPaginationResponse,
     WorkflowRunCountResponse,
     WorkflowRunDetailResponse,
+    WorkflowToolNodeExecutionListResponse,
     WorkflowRunNodeExecutionResponse,
     WorkflowRunNodeExecutionListResponse,
     HumanInputPauseTypeResponse,
@@ -323,6 +326,46 @@ class WorkflowRunNodeExecutionListApi(Resource):
         return dump_response(WorkflowRunNodeExecutionListResponse, {"data": node_executions})
 
 
+@console_ns.route("/apps/<uuid:app_id>/workflow-runs/<uuid:run_id>/node-executions/<string:node_execution_id>/children")
+class WorkflowToolNodeExecutionListApi(Resource):
+    @console_ns.doc("get_workflow_tool_node_executions")
+    @console_ns.doc(description="Get the internal node executions of one workflow tool invocation")
+    @console_ns.doc(
+        params={"app_id": "Root application ID", "run_id": "Workflow run ID", "node_execution_id": "Tool execution ID"}
+    )
+    @console_ns.response(
+        200,
+        "Internal node executions retrieved successfully",
+        console_ns.models[WorkflowToolNodeExecutionListResponse.__name__],
+    )
+    @console_ns.response(404, "Workflow run not found")
+    @console_account_admission(
+        rbac_checks=[RBACCheck(RBACPermission.APP_CREATE_AND_MANAGEMENT, PlainApp())],
+    )
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def get(self, request_context: RequestContext, app_model: App, run_id: UUID, node_execution_id: str):
+        executions = application_services().workflow_runs.get_workflow_tool_node_executions(
+            request_context,
+            app_id=app_model.id,
+            run_id=str(run_id),
+            node_execution_id=node_execution_id,
+        )
+        if executions is None:
+            raise NotFoundError("Workflow run not found")
+        try:
+            # Check every source separately because RBAC check bundles use OR semantics.
+            for source_app_id in {execution.app_id for execution in executions}:
+                enforce_rbac_checks(
+                    tenant_id=request_context.active_workspace_id,
+                    account_id=request_context.account_id,
+                    checks=[RBACCheck(RBACPermission.APP_CREATE_AND_MANAGEMENT, PlainApp())],
+                    path_args={"app_id": source_app_id},
+                )
+        except Forbidden as error:
+            raise Forbidden("You do not have permission to view this tool's internal execution details.") from error
+        return dump_response(WorkflowToolNodeExecutionListResponse, {"data": executions})
+
+
 @console_ns.route("/workflow/<string:workflow_run_id>/pause-details")
 class ConsoleWorkflowPauseDetailsApi(Resource):
     """Console API for getting workflow pause details."""

api/controllers/console/human_input_form.py

@@ -7,15 +7,17 @@ import logging
 from collections.abc import Generator, Sequence
 from typing import Any
 
-from flask import Response, jsonify, request
+from flask import Response, jsonify, request, stream_with_context
 from flask_restx import Resource
 from pydantic import RootModel
 from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
+from werkzeug.exceptions import Forbidden, NotFound
 
 from controllers.common.errors import InvalidArgumentError, NotFoundError
 from controllers.common.fields import EventStreamResponse
 from controllers.common.human_input import HumanInputFormSubmitPayload
+from controllers.common.rbac import PlainApp, RBACCheck, RBACPermission, enforce_rbac_checks
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import (
@@ -228,6 +230,20 @@ class ConsoleWorkflowEventsApi(Resource):
 
             include_state_snapshot = request.args.get("include_state_snapshot", "false").lower() == "true"
             continue_on_pause = request.args.get("continue_on_pause", "false").lower() == "true"
+            include_node_details = False
+            if include_state_snapshot:
+                # Owning a run permits replay; internal details require the same admission as run history.
+                try:
+                    enforce_rbac_checks(
+                        tenant_id=tenant_id,
+                        account_id=user.id,
+                        checks=[RBACCheck(RBACPermission.APP_CREATE_AND_MANAGEMENT, PlainApp())],
+                        path_args={"app_id": app.id},
+                    )
+                except (Forbidden, NotFound):
+                    pass
+                else:
+                    include_node_details = True
 
             def _generate_stream_events():
                 if include_state_snapshot:
@@ -239,6 +255,7 @@ class ConsoleWorkflowEventsApi(Resource):
                             app_id=workflow_run.app_id,
                             session_maker=session_maker,
                             human_input_surface=HumanInputSurface.CONSOLE,
+                            include_node_details=include_node_details,
                             close_on_pause=not continue_on_pause,
                         )
                     )
@@ -249,7 +266,7 @@ class ConsoleWorkflowEventsApi(Resource):
             event_generator = _generate_stream_events
 
         return Response(
-            event_generator(),
+            stream_with_context(event_generator()),  # pyrefly: ignore[no-matching-overload]
             mimetype="text/event-stream",
             headers={
                 "Cache-Control": "no-cache",

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

@@ -535,6 +535,8 @@ class WorkflowResponseConverter:
 
         try:
             if event.node_type == BuiltinNodeTypes.TOOL:
+                if event.provider_type == ToolProviderType.WORKFLOW:
+                    response.data.extras["workflow_tool"] = True
                 response.data.extras["icon"] = ToolManager.get_tool_icon(
                     tenant_id=self._application_generate_entity.app_config.tenant_id,
                     provider_type=ToolProviderType(event.provider_type),

api/extensions/logstore/repositories/logstore_api_workflow_node_execution_repository.py

@@ -16,7 +16,7 @@ from sqlalchemy.orm import sessionmaker
 from extensions.logstore.aliyun_logstore import AliyunLogStore
 from extensions.logstore.repositories import safe_float, safe_int
 from extensions.logstore.sql_escape import escape_identifier, escape_logstore_query_value
-from graphon.enums import WorkflowNodeExecutionStatus
+from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionStatus
 from libs.datetime_utils import ensure_naive_utc, naive_utc_now
 from models.enums import CreatorUserRole
 from models.workflow import WorkflowNodeExecutionModel, WorkflowNodeExecutionTriggeredFrom
@@ -276,6 +276,63 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep
                 return executions
             offset += len(page)
 
+    @override
+    def get_workflow_tool_executions(
+        self,
+        tenant_id: str,
+        workflow_run_id: str,
+        parent_node_execution_id: str,
+    ) -> Sequence[WorkflowNodeExecutionModel]:
+        sql_filters = (
+            f"tenant_id = '{escape_identifier(tenant_id)}' "
+            f"AND workflow_run_id = '{escape_identifier(workflow_run_id)}' AND __time__ > 0"
+        )
+        search_query = (
+            f"tenant_id: {escape_logstore_query_value(tenant_id)} "
+            f"and workflow_run_id: {escape_logstore_query_value(workflow_run_id)}"
+        )
+        requested_id = escape_identifier(parent_node_execution_id)
+        to_time = int(time.time())
+        parents = self.logstore_client.execute_sql(
+            sql=f"""
+                SELECT id, node_execution_id, workflow_id FROM "{AliyunLogStore.workflow_node_execution_logstore}"
+                WHERE {sql_filters} AND node_type = '{BuiltinNodeTypes.TOOL}'
+                  AND (id = '{requested_id}' OR node_execution_id = '{requested_id}')
+                ORDER BY log_version DESC LIMIT 1
+            """,
+            logstore=AliyunLogStore.workflow_node_execution_logstore,
+            query=search_query,
+            to_time=to_time,
+        )
+        if not parents:
+            return []
+        parent_execution_id = escape_identifier(parents[0].get("node_execution_id") or parents[0]["id"])
+        parent_workflow_id = escape_identifier(parents[0]["workflow_id"])
+        children: list[WorkflowNodeExecutionModel] = []
+        offset = 0
+        while True:
+            page = self.logstore_client.execute_sql(
+                sql=f"""
+                    SELECT * FROM (
+                        SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY log_version DESC) AS rn
+                        FROM "{AliyunLogStore.workflow_node_execution_logstore}"
+                        WHERE {sql_filters}
+                          AND triggered_from = '{WorkflowNodeExecutionTriggeredFrom.WORKFLOW_TOOL.value}'
+                          AND triggered_from_node_execution_id = '{parent_execution_id}'
+                          AND triggered_from_workflow_id = '{parent_workflow_id}'
+                    ) AS executions WHERE rn = 1
+                    ORDER BY created_at, "index", id LIMIT 1000 OFFSET {offset}
+                """,
+                logstore=AliyunLogStore.workflow_node_execution_logstore,
+                query=search_query,
+                to_time=to_time,
+            )
+            children.extend(_dict_to_workflow_node_execution_model(row) for row in page)
+            if len(page) < 1000:
+                self._apply_stopped_run(children, tenant_id, workflow_run_id)
+                return children
+            offset += len(page)
+
     @override
     def get_execution_by_id(
         self,
@@ -302,6 +359,6 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep
         if not results:
             return None
         execution = _dict_to_workflow_node_execution_model(results[0])
-        if execution.workflow_run_id is not None:
+        if execution.workflow_run_id:
             self._apply_stopped_run([execution], execution.tenant_id, execution.workflow_run_id)
         return execution

api/extensions/logstore/repositories/logstore_workflow_node_execution_repository.py

@@ -301,6 +301,24 @@ class LogstoreWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
                 logger.exception("Failed to dual-write node execution to SQL database: id=%s", execution.id)
                 # Don't raise - LogStore write succeeded, SQL is just a backup
 
+    @override
+    def save_many(self, executions: Sequence[WorkflowNodeExecution]) -> None:
+        """Append each record through the existing writer and batch the SQL backup."""
+        if not executions:
+            return
+        controlled_executions = []
+        for execution in executions:
+            execution, version = self._prepare_controlled_write(execution)
+            controlled_executions.append(execution)
+            self.logstore_client.put_log(
+                AliyunLogStore.workflow_node_execution_logstore, self._to_logstore_model(execution, log_version=version)
+            )
+        if self._enable_dual_write:
+            try:
+                self.sql_repository.save_many(controlled_executions)
+            except Exception:
+                logger.exception("Failed to dual-write workflow node execution batch: %s", [e.id for e in executions])
+
     def _prepare_controlled_write(self, execution: WorkflowNodeExecution) -> tuple[WorkflowNodeExecution, int]:
         # Assign ordering before checking SQL so an in-flight pause cannot sort
         # after a stop that commits while the LogStore upload is pending.
@@ -320,24 +338,6 @@ class LogstoreWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
                 )
         return execution, version
 
-    @override
-    def save_many(self, executions: Sequence[WorkflowNodeExecution]) -> None:
-        """Append each record through the existing writer and batch the SQL backup."""
-        if not executions:
-            return
-        controlled_executions = []
-        for execution in executions:
-            execution, version = self._prepare_controlled_write(execution)
-            controlled_executions.append(execution)
-            self.logstore_client.put_log(
-                AliyunLogStore.workflow_node_execution_logstore, self._to_logstore_model(execution, log_version=version)
-            )
-        if self._enable_dual_write:
-            try:
-                self.sql_repository.save_many(controlled_executions)
-            except Exception:
-                logger.exception("Failed to dual-write workflow node execution batch: %s", [e.id for e in executions])
-
     @override
     def save_synchronously(self, execution: WorkflowNodeExecution) -> None:
         """Create the SQL caller row required by Agent v2 participant ownership."""

api/fields/workflow_run_fields.py

@@ -227,3 +227,13 @@ def node_execution_response_source(node_execution: Any, *, session: Session) ->
 
 class WorkflowRunNodeExecutionListResponse(ResponseModel):
     data: list[WorkflowRunNodeExecutionResponse]
+
+
+class WorkflowToolNodeExecutionResponse(WorkflowRunNodeExecutionResponse):
+    app_id: str
+    workflow_id: str
+    node_execution_id: str | None = None
+
+
+class WorkflowToolNodeExecutionListResponse(ResponseModel):
+    data: list[WorkflowToolNodeExecutionResponse]

api/models/workflow.py

@@ -1108,6 +1108,7 @@ class WorkflowNodeExecutionModel(Base):  # This model is expected to have `offlo
 
     @property
     def extras(self) -> dict[str, Any]:
+        from core.tools.entities.tool_entities import ToolProviderType
         from core.tools.tool_manager import ToolManager
         from core.trigger.trigger_manager import TriggerManager
 
@@ -1116,9 +1117,12 @@ class WorkflowNodeExecutionModel(Base):  # This model is expected to have `offlo
         if execution_metadata:
             if self.node_type == BuiltinNodeTypes.TOOL and "tool_info" in execution_metadata:
                 tool_info: dict[str, Any] = execution_metadata["tool_info"]
+                provider_type = ToolProviderType(tool_info["provider_type"])
+                if provider_type == ToolProviderType.WORKFLOW:
+                    extras["workflow_tool"] = True
                 extras["icon"] = ToolManager.get_tool_icon(
                     tenant_id=self.tenant_id,
-                    provider_type=tool_info["provider_type"],
+                    provider_type=provider_type,
                     provider_id=tool_info["provider_id"],
                 )
             elif self.node_type == BuiltinNodeTypes.DATASOURCE and "datasource_info" in execution_metadata:

api/openapi/markdown/console-openapi.md

@@ -3828,6 +3828,24 @@ Stop running workflow task
 | 404 | Workflow run not found | **application/json**: [AppNotFoundError](#appnotfounderror)<br>[ErrorResponse](#errorresponse)<br> |
 | default | Error | **application/json**: [ErrorResponse](#errorresponse)<br> |
 
+### [GET] /apps/{app_id}/workflow-runs/{run_id}/node-executions/{node_execution_id}/children
+Get the internal node executions of one workflow tool invocation
+
+#### Parameters
+
+| Name | Located in | Description | Required | Schema |
+| ---- | ---------- | ----------- | -------- | ------ |
+| app_id | path | Root application ID | Yes | string (uuid) |
+| node_execution_id | path | Tool execution ID | Yes | string |
+| run_id | path | Workflow run ID | Yes | string (uuid) |
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 200 | Internal node executions retrieved successfully | **application/json**: [WorkflowToolNodeExecutionListResponse](#workflowtoolnodeexecutionlistresponse)<br> |
+| 404 | Workflow run not found |  |
+
 ### [GET] /apps/{app_id}/workflow-runs/{workflow_run_id}/agent-nodes/{node_id}/sandbox/files
 List a directory in a workflow Agent node sandbox
 
@@ -28968,6 +28986,43 @@ Query parameters for workflow runs.
 | ---- | ---- | ----------- | -------- |
 | workflow_tool_id | string |  | Yes |
 
+#### WorkflowToolNodeExecutionListResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| data | [ [WorkflowToolNodeExecutionResponse](#workflowtoolnodeexecutionresponse) ] |  | Yes |
+
+#### WorkflowToolNodeExecutionResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| app_id | string |  | Yes |
+| created_at | integer |  | No |
+| created_by_account | [SimpleAccountResponse](#simpleaccountresponse) |  | No |
+| created_by_end_user | [SimpleEndUser](#simpleenduser) |  | No |
+| created_by_role | string |  | No |
+| elapsed_time | number |  | No |
+| error | string |  | No |
+| execution_metadata |  |  | No |
+| extras |  |  | No |
+| finished_at | integer |  | No |
+| id | string |  | Yes |
+| index | integer |  | No |
+| inputs |  |  | No |
+| inputs_truncated | boolean |  | No |
+| node_execution_id | string |  | No |
+| node_id | string |  | No |
+| node_type | string |  | No |
+| outputs |  |  | No |
+| outputs_truncated | boolean |  | No |
+| predecessor_node_id | string |  | No |
+| process_data |  |  | No |
+| process_data_truncated | boolean |  | No |
+| retry_index | integer |  | No |
+| status | string |  | No |
+| title | string |  | No |
+| workflow_id | string |  | Yes |
+
 #### WorkflowToolParameterConfiguration
 
 Workflow tool configuration

api/repositories/api_workflow_node_execution_repository.py

@@ -184,6 +184,19 @@ class DifyAPIWorkflowNodeExecutionRepository(Protocol):
         """
         ...
 
+    def get_workflow_tool_executions(
+        self,
+        tenant_id: str,
+        workflow_run_id: str,
+        parent_node_execution_id: str,
+    ) -> Sequence[WorkflowNodeExecutionModel]:
+        """Get one Tool execution's children after the service admits the owning app/run.
+
+        The parent may belong to a nested source app. Both parent and children must
+        belong to the admitted tenant and run; preserve each child's source app.
+        """
+        ...
+
     def get_execution_by_id(
         self,
         execution_id: str,

api/repositories/sqlalchemy_api_workflow_node_execution_repository.py

@@ -9,12 +9,12 @@ from collections.abc import Mapping, Sequence
 from datetime import datetime
 from typing import Any, cast, override
 
-from sqlalchemy import asc, delete, desc, func, or_, select
+from sqlalchemy import String, asc, delete, desc, func, or_, select
 from sqlalchemy.engine import CursorResult
 from sqlalchemy.orm import Session, sessionmaker
 
 from extensions.ext_storage import storage
-from graphon.enums import WorkflowNodeExecutionStatus
+from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionStatus
 from models.workflow import WorkflowNodeExecutionModel, WorkflowNodeExecutionOffload, WorkflowNodeExecutionTriggeredFrom
 from repositories.api_workflow_node_execution_repository import (
     DifyAPIWorkflowNodeExecutionRepository,
@@ -133,6 +133,49 @@ class DifyAPISQLAlchemyWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecut
         with self._session_maker() as session:
             return session.execute(stmt).scalars().all()
 
+    @override
+    def get_workflow_tool_executions(
+        self,
+        tenant_id: str,
+        workflow_run_id: str,
+        parent_node_execution_id: str,
+    ) -> Sequence[WorkflowNodeExecutionModel]:
+        with self._session_maker() as session:
+            parent = session.execute(
+                select(
+                    func.coalesce(
+                        func.nullif(WorkflowNodeExecutionModel.node_execution_id, ""),
+                        WorkflowNodeExecutionModel.id.cast(String),
+                    ),
+                    WorkflowNodeExecutionModel.workflow_id,
+                )
+                .where(
+                    WorkflowNodeExecutionModel.tenant_id == tenant_id,
+                    WorkflowNodeExecutionModel.workflow_run_id == workflow_run_id,
+                    WorkflowNodeExecutionModel.node_type == BuiltinNodeTypes.TOOL,
+                    or_(
+                        WorkflowNodeExecutionModel.id.cast(String) == parent_node_execution_id,
+                        WorkflowNodeExecutionModel.node_execution_id == parent_node_execution_id,
+                    ),
+                )
+                .limit(1)
+            ).one_or_none()
+            if parent is None:
+                return []
+            parent_execution_id, parent_workflow_id = parent
+            stmt = (
+                WorkflowNodeExecutionModel.preload_offload_data(select(WorkflowNodeExecutionModel))
+                .where(
+                    WorkflowNodeExecutionModel.tenant_id == tenant_id,
+                    WorkflowNodeExecutionModel.workflow_run_id == workflow_run_id,
+                    WorkflowNodeExecutionModel.triggered_from == WorkflowNodeExecutionTriggeredFrom.WORKFLOW_TOOL,
+                    WorkflowNodeExecutionModel.triggered_from_node_execution_id == parent_execution_id,
+                    WorkflowNodeExecutionModel.triggered_from_workflow_id == parent_workflow_id,
+                )
+                .order_by(WorkflowNodeExecutionModel.created_at, WorkflowNodeExecutionModel.index)
+            )
+            return session.scalars(stmt).all()
+
     @override
     def get_execution_snapshots_by_workflow_run(
         self,

api/services/app_task_service.py

@@ -326,8 +326,7 @@ class AppTaskService:
                 app_id=app_id,
                 triggered_from=triggered_from,
             )
-            for execution in stopped_nodes:
-                target.save(execution)
+            target.save_many(stopped_nodes)
 
     @staticmethod
     def stop_task(

api/services/workflow_event_snapshot_service.py

@@ -40,7 +40,7 @@ from core.workflow.nodes.human_input.pause_reason import (
     HumanInputRequired,
 )
 from graphon.entities import WorkflowStartReason
-from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionStatus
+from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from graphon.runtime import RuntimeState
 from graphon.runtime.runtime_state_protocol import ReadOnlyVariablePool
 from graphon.workflow_type_encoder import WorkflowRuntimeTypeConverter
@@ -48,7 +48,7 @@ from libs.broadcast_channel.exc import SubscriptionClosedError
 from libs.datetime_utils import to_utc_timestamp
 from models.human_input import HumanInputForm
 from models.model import AppMode, Message
-from models.workflow import WorkflowNodeExecutionTriggeredFrom, WorkflowRun
+from models.workflow import WorkflowNodeExecutionModel, WorkflowNodeExecutionTriggeredFrom, WorkflowRun
 from repositories.api_workflow_node_execution_repository import WorkflowNodeExecutionSnapshot
 from repositories.entities.workflow_pause import WorkflowPauseEntity
 from repositories.factory import DifyAPIRepositoryFactory
@@ -85,6 +85,7 @@ def build_workflow_event_stream(
     idle_timeout: float = 300,
     ping_interval: float = 10.0,
     close_on_pause: bool = True,
+    include_node_details: bool = False,
 ) -> Generator[Mapping[str, Any] | str]:
     topic = MessageGenerator.get_response_topic(app_mode, workflow_run.id)
     workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
@@ -157,6 +158,18 @@ def build_workflow_event_stream(
         triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
         workflow_run_id=workflow_run.id,
     )
+    # Console replay includes the same inline previews as run history. Other surfaces
+    # retain the detail-free snapshot and avoid loading internal node data.
+    node_executions = (
+        {
+            execution.node_execution_id or execution.id: execution
+            for execution in node_execution_repo.get_executions_by_workflow_run(
+                tenant_id=tenant_id, app_id=app_id, workflow_run_id=workflow_run.id
+            )
+        }
+        if include_node_details
+        else {}
+    )
 
     def _generate() -> Generator[Mapping[str, Any] | str]:
         # send a PING event immediately to prevent the connection staying in pending state for a long time.
@@ -182,6 +195,7 @@ def build_workflow_event_stream(
                     resumption_context=resumption_context,
                     session_maker=session_maker,
                     human_input_surface=human_input_surface,
+                    node_executions=node_executions,
                 )
 
                 for event in snapshot_events:
@@ -325,6 +339,7 @@ def _build_snapshot_events(
     resumption_context: WorkflowResumptionContext | None,
     session_maker: sessionmaker[Session] | None = None,
     human_input_surface: HumanInputSurface | None = None,
+    node_executions: Mapping[str, WorkflowNodeExecutionModel] | None = None,
 ) -> list[Mapping[str, Any]]:
     events: list[Mapping[str, Any]] = []
     variable_pool = _load_variable_pool_from_resumption_context(resumption_context)
@@ -342,10 +357,12 @@ def _build_snapshot_events(
         events.append(message_replace)
 
     for snapshot in node_snapshots:
+        execution = node_executions.get(snapshot.execution_id) if node_executions else None
         node_started = _build_node_started_event(
             workflow_run_id=workflow_run.id,
             task_id=task_id,
             snapshot=snapshot,
+            execution=execution,
         )
         _apply_message_context(node_started, message_context)
         events.append(node_started)
@@ -355,6 +372,7 @@ def _build_snapshot_events(
                 workflow_run_id=workflow_run.id,
                 task_id=task_id,
                 snapshot=snapshot,
+                execution=execution,
             )
             _apply_message_context(node_finished, message_context)
             events.append(node_finished)
@@ -424,6 +442,7 @@ def _build_node_started_event(
     workflow_run_id: str,
     task_id: str,
     snapshot: WorkflowNodeExecutionSnapshot,
+    execution: WorkflowNodeExecutionModel | None = None,
 ) -> dict[str, Any]:
     created_at = int(snapshot.created_at.timestamp()) if snapshot.created_at else 0
     response = NodeStartStreamResponse(
@@ -435,15 +454,16 @@ def _build_node_started_event(
             node_type=snapshot.node_type,
             title=snapshot.title,
             index=snapshot.index,
-            predecessor_node_id=None,
-            inputs=None,
+            predecessor_node_id=execution.predecessor_node_id if execution else None,
+            inputs=execution.inputs_dict if execution else None,
+            inputs_truncated=execution.inputs_truncated if execution else False,
             created_at=created_at,
-            extras={},
+            extras=execution.extras if execution else {},
             iteration_id=snapshot.iteration_id,
             loop_id=snapshot.loop_id,
         ),
     )
-    return response.to_ignore_detail_dict()
+    return response.model_dump(mode="json") if execution else response.to_ignore_detail_dict()
 
 
 def _build_human_input_required_events(
@@ -490,10 +510,7 @@ def _build_human_input_required_events(
         if expiration_time is None:
             continue
 
-        resolved_inputs = resolve_variable_select_input_options(
-            reason.inputs,
-            variable_pool=variable_pool,
-        )
+        resolved_inputs = resolve_variable_select_input_options(reason.inputs, variable_pool=variable_pool)
         disposition = dispositions_by_form_id.get(form_id)
 
         response = HumanInputRequiredResponse(
@@ -535,6 +552,7 @@ def _build_node_finished_event(
     workflow_run_id: str,
     task_id: str,
     snapshot: WorkflowNodeExecutionSnapshot,
+    execution: WorkflowNodeExecutionModel | None = None,
 ) -> dict[str, Any]:
     created_at = int(snapshot.created_at.timestamp()) if snapshot.created_at else 0
     finished_at = int(snapshot.finished_at.timestamp()) if snapshot.finished_at else created_at
@@ -547,14 +565,24 @@ def _build_node_finished_event(
             node_type=snapshot.node_type,
             title=snapshot.title,
             index=snapshot.index,
-            predecessor_node_id=None,
-            inputs=None,
-            process_data=None,
-            outputs=None,
+            predecessor_node_id=execution.predecessor_node_id if execution else None,
+            inputs=execution.inputs_dict if execution else None,
+            inputs_truncated=execution.inputs_truncated if execution else False,
+            process_data=execution.process_data_dict if execution else None,
+            process_data_truncated=execution.process_data_truncated if execution else False,
+            outputs=execution.outputs_dict if execution else None,
+            outputs_truncated=execution.outputs_truncated if execution else False,
             status=WorkflowNodeExecutionStatus(snapshot.status),
-            error=None,
+            error=execution.error if execution else None,
             elapsed_time=snapshot.elapsed_time,
-            execution_metadata=None,
+            execution_metadata=(
+                {
+                    WorkflowNodeExecutionMetadataKey(key): value
+                    for key, value in execution.execution_metadata_dict.items()
+                }
+                if execution
+                else None
+            ),
             created_at=created_at,
             finished_at=finished_at,
             files=[],
@@ -562,7 +590,7 @@ def _build_node_finished_event(
             loop_id=snapshot.loop_id,
         ),
     )
-    return response.to_ignore_detail_dict()
+    return response.model_dump(mode="json") if execution else response.to_ignore_detail_dict()
 
 
 def _build_pause_event(

api/services/workflow_run_service.py

@@ -4,6 +4,10 @@ from datetime import datetime
 from typing import TypedDict
 
 import contexts
+from core.workflow.node_execution_process_data import (
+    WORKFLOW_TOOL_INVOCATION_ID_KEY,
+    WORKFLOW_TOOL_ROOT_APP_ID_KEY,
+)
 from core.workflow.nodes.human_input.pause_reason import HumanInputRequired
 from extensions.ext_database import db
 from graphon.enums import WorkflowExecutionStatus
@@ -205,6 +209,38 @@ class WorkflowRunService:
         )
         return assemble_workflow_node_execution_traces(node_executions, self._node_executions, session=db.session())
 
+    def get_workflow_tool_node_executions(
+        self,
+        context: RequestContext,
+        *,
+        app_id: str,
+        run_id: str,
+        node_execution_id: str,
+    ) -> list[WorkflowNodeExecutionTrace] | None:
+        """Read nested Tool traces through the admitted root app and workflow run."""
+        if self.get_workflow_run(context, app_id=app_id, run_id=run_id) is None:
+            return None
+        contexts.plugin_tool_providers.set({})
+        contexts.plugin_tool_providers_lock.set(threading.Lock())
+        executions = self._node_executions.get_workflow_tool_executions(
+            tenant_id=context.active_workspace_id,
+            workflow_run_id=run_id,
+            parent_node_execution_id=node_execution_id,
+        )
+        traces = assemble_workflow_node_execution_traces(executions, self._node_executions, session=db.session())
+        for trace in traces:
+            if trace.process_data is not None:
+                trace.process_data = {
+                    key: value
+                    for key, value in trace.process_data.items()
+                    if key
+                    not in (
+                        WORKFLOW_TOOL_INVOCATION_ID_KEY,
+                        WORKFLOW_TOOL_ROOT_APP_ID_KEY,
+                    )
+                }
+        return traces
+
     def get_pause_details(
         self,
         context: RequestContext,

api/tests/unit_tests/controllers/console/app/test_workflow_run_api.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+from collections.abc import Callable
 from datetime import UTC, datetime
 from inspect import unwrap
 from types import SimpleNamespace
@@ -8,12 +9,15 @@ from typing import Any
 from unittest.mock import Mock
 
 import pytest
-from flask import Flask
+from flask import Flask, request
 from flask_restx import marshal
 from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
 
 from controllers.common.errors import NotFoundError
+from controllers.common.rbac import locators as rbac_locators
 from controllers.console.app import workflow_run as workflow_run_module
+from core.rbac import RBACPermission, RBACResourceScope
 from extensions.ext_database import db
 from fields.workflow_run_fields import node_execution_response_source
 from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionStatus
@@ -26,7 +30,10 @@ from models.workflow import (
     WorkflowRun,
     WorkflowType,
 )
+from services.enterprise.rbac_service import RBACService
+from services.workflow_node_execution_trace_service import WorkflowNodeExecutionTrace
 from tests.unit_tests.model_factories import make_account, make_app
+from tests.unit_tests.rbac_fakes import RBACDomain
 
 
 def _serialize_200_response(handler, payload: Any) -> Any:
@@ -385,3 +392,108 @@ def test_workflow_run_node_executions_return_frontend_trace_contract(
         app_id="app-1",
         run_id="run-1",
     )
+
+
+@pytest.mark.parametrize("source_access", ["granted", "maintainer", "rbac-disabled"])
+def test_workflow_tool_children_preserve_source_identity(
+    app: Flask,
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_session: Session,
+    rbac_domain: RBACDomain,
+    config_overrides: Callable[..., None],
+    source_access: str,
+) -> None:
+    config_overrides(RBAC_ENABLED=source_access != "rbac-disabled")
+    monkeypatch.setattr(rbac_locators, "application_services", lambda: rbac_domain)
+    sqlite_session.add(make_app(app_id="source-app", maintainer="account-1" if source_access == "maintainer" else None))
+    sqlite_session.commit()
+    monkeypatch.setattr(db, "session", sqlite_session)
+    check_access = Mock(return_value=source_access == "granted")
+    monkeypatch.setattr(RBACService.CheckAccess, "check", check_access)
+    workflow_runs = Mock()
+    workflow_runs.get_workflow_tool_node_executions.return_value = [
+        WorkflowNodeExecutionTrace(
+            id="child-row",
+            tenant_id="tenant-1",
+            app_id="source-app",
+            workflow_id="source-workflow",
+            triggered_from="workflow-tool",
+            node_execution_id="source-execution",
+            created_by="account-1",
+            outputs={"answer": "world"},
+        )
+    ]
+    _mock_application_services(monkeypatch, workflow_runs)
+    context = _request_context()
+    api = workflow_run_module.WorkflowToolNodeExecutionListApi()
+    handler = unwrap(api.get)
+    with app.test_request_context("/apps/app-1/workflow-runs/run-1/node-executions/parent/children"):
+        response = handler(api, context, app_model=_app(), run_id="run-1", node_execution_id="parent")
+        workflow_runs.get_workflow_tool_node_executions.return_value = None
+        with pytest.raises(NotFoundError, match="Workflow run not found"):
+            handler(api, context, app_model=_app(), run_id="run-1", node_execution_id="parent")
+    assert response["data"][0]["app_id"] == "source-app"
+    assert response["data"][0]["workflow_id"] == "source-workflow"
+    assert response["data"][0]["node_execution_id"] == "source-execution"
+    assert response["data"][0]["outputs"] == {"answer": "world"}
+    if source_access == "granted":
+        check_access.assert_called_once_with(
+            "tenant-1",
+            "account-1",
+            scene=RBACPermission.APP_CREATE_AND_MANAGEMENT,
+            resource_type=RBACResourceScope.APP,
+            resource_id="source-app",
+        )
+    else:
+        check_access.assert_not_called()
+    workflow_runs.get_workflow_tool_node_executions.assert_called_with(
+        context, app_id="app-1", run_id="run-1", node_execution_id="parent"
+    )
+
+
+@pytest.mark.parametrize("source_app_ids", [("source-app",), ("allowed-source", "source-app")])
+def test_workflow_tool_children_deny_any_unauthorized_source(
+    app: Flask,
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_session: Session,
+    rbac_domain: RBACDomain,
+    config_overrides: Callable[..., None],
+    source_app_ids: tuple[str, ...],
+) -> None:
+    config_overrides(RBAC_ENABLED=True)
+    monkeypatch.setattr(rbac_locators, "application_services", lambda: rbac_domain)
+    sqlite_session.add_all(make_app(app_id=app_id) for app_id in ("app-1", "allowed-source", "source-app"))
+    sqlite_session.commit()
+    monkeypatch.setattr(db, "session", sqlite_session)
+    check_access = Mock(side_effect=lambda *_args, resource_id, **_kwargs: resource_id != "source-app")
+    monkeypatch.setattr(RBACService.CheckAccess, "check", check_access)
+    workflow_runs = Mock()
+    workflow_runs.get_workflow_tool_node_executions.return_value = [
+        WorkflowNodeExecutionTrace(
+            id=f"{app_id}-row",
+            tenant_id="tenant-1",
+            app_id=app_id,
+            workflow_id=f"{app_id}-workflow",
+            triggered_from="workflow-tool",
+            created_by="account-1",
+            inputs={"secret": "source input"},
+            process_data={"secret": "source internals"},
+            outputs={"secret": "source output"},
+        )
+        for app_id in source_app_ids
+    ]
+    _mock_application_services(monkeypatch, workflow_runs)
+    context = _request_context()
+    api = workflow_run_module.WorkflowToolNodeExecutionListApi()
+    handler = unwrap(api.get)
+
+    with app.test_request_context("/apps/app-1/workflow-runs/run-1/node-executions/parent/children"):
+        request.view_args = {"app_id": "app-1"}
+        with pytest.raises(Forbidden, match="permission to view this tool's internal execution details") as error:
+            handler(api, context, app_model=_app(), run_id="run-1", node_execution_id="parent")
+
+    assert error.value.code == 403
+    assert check_access.call_args.kwargs["resource_id"] == "source-app"
+    workflow_runs.get_workflow_tool_node_executions.assert_called_once_with(
+        context, app_id="app-1", run_id="run-1", node_execution_id="parent"
+    )

api/tests/unit_tests/controllers/console/test_human_input_form.py

@@ -1,17 +1,19 @@
 from __future__ import annotations
 
 import json
+from collections.abc import Iterator
 from datetime import UTC, datetime
 from inspect import unwrap
 from types import SimpleNamespace
 from unittest.mock import ANY, Mock
 
 import pytest
-from flask import Flask, Response
+from flask import Flask, Response, has_request_context
 from sqlalchemy import Engine
 
 from controllers.common.errors import NotFoundError
 from controllers.common.human_input import HumanInputFormSubmitPayload
+from controllers.common.rbac import locators as rbac_locators
 from controllers.console import human_input_form as human_input_form_module
 from controllers.console.human_input_form import (
     ConsoleHumanInputFormApi,
@@ -20,6 +22,7 @@ from controllers.console.human_input_form import (
     WorkflowResponseConverter,
     _jsonify_form_definition,
 )
+from core.rbac import RBACPermission, RBACResourceScope
 from core.workflow.human_input_policy import HumanInputSurface
 from core.workflow.nodes.human_input.entities import SelectInputConfig, StringListSource
 from core.workflow.nodes.human_input.enums import ValueSourceType
@@ -29,8 +32,11 @@ from models.enums import CreatorUserRole, WorkflowRunTriggeredFrom
 from models.human_input import RecipientType
 from models.model import App, AppMode
 from models.workflow import WorkflowRun
+from services.enterprise.rbac_service import RBACService
+from services.entities.app_entities import AppAgentBinding
 from tests.unit_tests.config_override import apply_config_overrides
 from tests.unit_tests.model_factories import make_account, make_app
+from tests.unit_tests.rbac_fakes import RBACDomain
 
 
 @pytest.fixture(autouse=True)
@@ -381,9 +387,42 @@ def test_workflow_events_finished(app: Flask, monkeypatch: pytest.MonkeyPatch) -
     assert "data" in response.get_data(as_text=True)
 
 
-def test_workflow_events_snapshot_can_continue_across_pauses(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
+@pytest.mark.parametrize(
+    ("rbac_enabled", "has_permission", "is_maintainer", "agent_backed", "include_node_details"),
+    [
+        (True, False, False, False, False),
+        (True, True, False, False, True),
+        (True, False, True, False, True),
+        (True, True, False, True, False),
+        (False, False, False, False, True),
+    ],
+    ids=["run-only", "app-manager", "maintainer", "agent-app", "rbac-disabled"],
+)
+def test_workflow_events_snapshot_can_continue_across_pauses(
+    app: Flask,
+    monkeypatch: pytest.MonkeyPatch,
+    rbac_domain: RBACDomain,
+    rbac_enabled: bool,
+    has_permission: bool,
+    is_maintainer: bool,
+    agent_backed: bool,
+    include_node_details: bool,
+) -> None:
     workflow_run = _workflow_run()
+    workflow_run.triggered_from = WorkflowRunTriggeredFrom.APP_RUN
     app_model = _app()
+    apply_config_overrides(monkeypatch, RBAC_ENABLED=rbac_enabled)
+    check_access = Mock(return_value=has_permission)
+    monkeypatch.setattr(RBACService.CheckAccess, "check", check_access)
+    monkeypatch.setattr(rbac_locators, "application_services", lambda: rbac_domain)
+    monkeypatch.setattr(
+        rbac_domain.rbac.queries,
+        "get_app_agent_binding",
+        Mock(return_value=AppAgentBinding(id="agent-1", workflow_only=False) if agent_backed else None),
+    )
+    monkeypatch.setattr(
+        rbac_domain.rbac.queries, "get_app_maintainer", Mock(return_value="user-1" if is_maintainer else None)
+    )
 
     class _RepoStub:
         def get_workflow_run_by_id_and_tenant_id(self, **_kwargs):
@@ -391,9 +430,13 @@ def test_workflow_events_snapshot_can_continue_across_pauses(app: Flask, monkeyp
 
     snapshot_calls: list[dict[str, object]] = []
 
-    def snapshot_builder(**kwargs: object) -> list[dict[str, str]]:
+    def snapshot_events() -> Iterator[dict[str, str]]:
+        assert has_request_context()
+        yield {"event": "snapshot"}
+
+    def snapshot_builder(**kwargs: object) -> Iterator[dict[str, str]]:
         snapshot_calls.append(kwargs)
-        return [{"event": "snapshot"}]
+        return snapshot_events()
 
     monkeypatch.setattr(
         DifyAPIRepositoryFactory,
@@ -427,6 +470,17 @@ def test_workflow_events_snapshot_can_continue_across_pauses(app: Flask, monkeyp
             "app_id": "app-1",
             "session_maker": ANY,
             "human_input_surface": HumanInputSurface.CONSOLE,
+            "include_node_details": include_node_details,
             "close_on_pause": False,
         }
     ]
+    if rbac_enabled and not is_maintainer and not agent_backed:
+        check_access.assert_called_once_with(
+            "t1",
+            "user-1",
+            scene=RBACPermission.APP_CREATE_AND_MANAGEMENT,
+            resource_type=RBACResourceScope.APP,
+            resource_id="app-1",
+        )
+    else:
+        check_access.assert_not_called()

api/tests/unit_tests/core/app/apps/common/test_workflow_response_converter_resumption.py

@@ -1,8 +1,10 @@
+import json
 from datetime import datetime, timedelta
 
 import pytest
 
 from core.app.app_config.entities import WorkflowUIBasedAppConfig
+from core.app.apps.common import workflow_response_converter
 from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
 from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
 from core.app.entities.queue_entities import (
@@ -16,6 +18,7 @@ from graphon.entities import WorkflowStartReason
 from graphon.enums import BuiltinNodeTypes
 from models import AppMode
 from models.account import Account
+from models.workflow import WorkflowNodeExecutionModel
 
 
 def _build_converter() -> WorkflowResponseConverter:
@@ -94,6 +97,34 @@ def test_node_start_preserves_the_allocated_index_when_events_arrive_out_of_orde
         assert response.data.index == index
 
 
+@pytest.mark.parametrize("provider_type", ["workflow", "builtin"])
+def test_workflow_tool_detail_hint_in_live_and_saved_traces(monkeypatch, provider_type):
+    monkeypatch.setattr(workflow_response_converter.ToolManager, "get_tool_icon", lambda **_kwargs: "icon")
+    converter = _build_converter()
+    converter.workflow_start_to_stream_response(
+        task_id="task-1", workflow_run_id="run-1", workflow_id="wf-1", reason=WorkflowStartReason.INITIAL
+    )
+    response = converter.workflow_node_start_to_stream_response(
+        task_id="task-1",
+        event=QueueNodeStartedEvent(
+            node_execution_id="tool-execution",
+            node_id="tool",
+            node_title="Tool",
+            node_type=BuiltinNodeTypes.TOOL,
+            start_at=datetime(2026, 9, 7),
+            provider_type=provider_type,
+            provider_id="provider-1",
+        ),
+    )
+    saved = WorkflowNodeExecutionModel(
+        node_type=BuiltinNodeTypes.TOOL,
+        execution_metadata=json.dumps({"tool_info": {"provider_type": provider_type, "provider_id": "provider-1"}}),
+    )
+    assert response is not None
+    assert response.data.extras == saved.extras
+    assert response.data.extras.get("workflow_tool", False) == (provider_type == "workflow")
+
+
 @pytest.mark.parametrize("next_node_already_persisted", [False, True])
 def test_resumed_node_metadata_and_sequence(next_node_already_persisted):
     started_at = datetime(2026, 9, 7)

api/tests/unit_tests/core/app/workflow/test_persistence_layer.py

@@ -62,6 +62,9 @@ from graphon.runtime.execution import ROOT_FRAME_ID
 from models import Account, WorkflowRun
 from models.enums import WorkflowRunTriggeredFrom
 from models.workflow import WorkflowNodeExecutionModel, WorkflowNodeExecutionTriggeredFrom
+from repositories.sqlalchemy_api_workflow_node_execution_repository import (
+    DifyAPISQLAlchemyWorkflowNodeExecutionRepository,
+)
 from services.workflow_persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from tests.workflow_test_utils import build_test_graph_init_params
 
@@ -286,6 +289,43 @@ def test_workflow_tool_agent_nodes_persist_in_source_app_under_only_root_run(sql
         assert row.process_data_dict[WORKFLOW_TOOL_ROOT_APP_ID_KEY] == "caller-app"
 
 
+def test_nested_tool_callers_keep_distinct_invocations_in_shared_source_layers(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    ids = ["a-call-1", "a-call-2", "b-call-1", "b-call-2", "c-child"]
+    layer = _make_sql_layer(sqlite_session_factory, node_run_indices=dict.fromkeys(ids, 1))
+    source = _tool_source()
+    nested = replace(source, app_id="nested-app", workflow_id="nested-workflow")
+    first_call = layer.create_workflow_tool_event_listener(source, "workflow-id", "a-call-1")
+    second_call = layer.create_workflow_tool_event_listener(source, "workflow-id", "a-call-2")
+    nested_call = layer.create_workflow_tool_event_listener(nested, source.workflow_id, "b-call-1")
+    layer.on_event(GraphRunStartedEvent())
+    for execution_id, listener in zip(
+        ids, [layer.on_event, layer.on_event, first_call, second_call, nested_call], strict=True
+    ):
+        listener(
+            NodeRunStartedEvent(
+                id=execution_id,
+                node_id="same-canvas-node",
+                node_type=BuiltinNodeTypes.TOOL,
+                node_title="Tool",
+                start_at=_naive_utc_now(),
+            )
+        )
+    repository = DifyAPISQLAlchemyWorkflowNodeExecutionRepository(sqlite_session_factory)
+    for caller, expected, source_app, source_workflow in (
+        ("a-call-1", "b-call-1", "source-app", "source-workflow"),
+        ("a-call-2", "b-call-2", "source-app", "source-workflow"),
+        ("b-call-1", "c-child", "nested-app", "nested-workflow"),
+    ):
+        children = repository.get_workflow_tool_executions("tenant", "run-id", caller)
+        assert [child.id for child in children] == [expected]
+        child = children[0]
+        assert (child.app_id, child.workflow_id, child.workflow_run_id) == (source_app, source_workflow, "run-id")
+        assert child.triggered_from_workflow_id == ("source-workflow" if caller == "b-call-1" else "workflow-id")
+        assert child.triggered_from_node_execution_id == caller
+
+
 def test_workflow_tool_resume_updates_existing_running_container(sqlite_session_factory):
     layer = _make_sql_layer(sqlite_session_factory, node_run_indices={"loop-exec": 1})
     listen = layer.create_workflow_tool_event_listener(_tool_source(), "workflow-id", "caller-exec")

api/tests/unit_tests/core/repositories/test_workflow_control_persistence.py

@@ -228,14 +228,16 @@ def test_logstore_stop_orders_before_delayed_upload_and_overrides_stale_indexes(
                     )
                 )
             else:
-                node_writer.save(
-                    node.model_copy(
-                        update={
-                            "status": WorkflowNodeExecutionStatus.FAILED,
-                            "finished_at": stopped.finished_at,
-                            "error": stopped.error,
-                        }
-                    )
+                node_writer.save_many(
+                    [
+                        node.model_copy(
+                            update={
+                                "status": WorkflowNodeExecutionStatus.FAILED,
+                                "finished_at": stopped.finished_at,
+                                "error": stopped.error,
+                            }
+                        )
+                    ]
                 )
         records.append(dict(fields))
 
@@ -271,6 +273,12 @@ def test_logstore_stop_orders_before_delayed_upload_and_overrides_stale_indexes(
         by_id = api.get_execution_by_id("node-1", "tenant-1")
         assert by_id is not None
         assert by_id.status == expected_status
+        assert (
+            api.get_execution_snapshots_by_workflow_run("tenant-1", "app-1", "workflow-1", "workflow-run", "run-1")[
+                0
+            ].status
+            == expected_status
+        )
         assert node_writer.get_by_workflow_execution("run-1")[0].status == expected_status
         node_writer.save(node)
     assert records[-1]["status"] == expected_status

api/tests/unit_tests/extensions/logstore/repositories/test_logstore_api_workflow_node_execution_repository.py

@@ -206,6 +206,126 @@ def test_resumption_snapshots_read_latest_nodes_with_full_owner_scope() -> None:
     )
 
 
+@pytest.mark.parametrize("requested_id", ["parent-row", "parent-engine"])
+def test_workflow_tool_children_only_fetch_matching_latest_payloads(requested_id: str) -> None:
+    with patch("extensions.logstore.repositories.logstore_api_workflow_node_execution_repository.AliyunLogStore"):
+        repository = LogstoreAPIWorkflowNodeExecutionRepository(session_maker=None)
+    repository.logstore_client = MagicMock()
+    rows = [
+        {
+            "id": "parent-row",
+            "node_execution_id": "parent-engine",
+            "workflow_id": "caller-workflow",
+            "node_type": "tool",
+            "log_version": 1,
+            "created_at": 0,
+        },
+        *[
+            {
+                "id": "child",
+                "node_type": "human-input",
+                "triggered_from": "workflow-tool",
+                "triggered_from_node_execution_id": "parent-engine",
+                "log_version": version,
+                "status": status,
+                "created_at": 1,
+                "index": 2,
+                "outputs": '{"answer":"approved"}',
+            }
+            for version, status in ((1, "running"), (2, "paused"))
+        ],
+        {
+            "id": "other-call",
+            "triggered_from": "workflow-tool",
+            "triggered_from_node_execution_id": "other-engine",
+            "created_at": 2,
+        },
+        {
+            "id": "earlier-child",
+            "triggered_from": "workflow-tool",
+            "triggered_from_node_execution_id": "parent-engine",
+            "created_at": 1,
+            "index": 1,
+        },
+        {
+            "id": "wrong-origin",
+            "triggered_from": "workflow-run",
+            "triggered_from_node_execution_id": "parent-engine",
+        },
+        {
+            "id": "other-tenant",
+            "tenant_id": "other-tenant",
+            "triggered_from": "workflow-tool",
+            "triggered_from_node_execution_id": "parent-engine",
+        },
+        {
+            "id": "other-run",
+            "workflow_run_id": "other-run",
+            "triggered_from": "workflow-tool",
+            "triggered_from_node_execution_id": "parent-engine",
+        },
+        {
+            "id": "other-caller",
+            "triggered_from": "workflow-tool",
+            "triggered_from_workflow_id": "other-workflow",
+            "triggered_from_node_execution_id": "parent-engine",
+        },
+        *[{"id": f"unrelated-{index}"} for index in range(1001)],
+    ]
+    defaults = {
+        "id": "",
+        "tenant_id": "tenant",
+        "app_id": "source-app",
+        "workflow_run_id": "run",
+        "node_execution_id": None,
+        "workflow_id": "source-workflow",
+        "triggered_from_workflow_id": "caller-workflow",
+        "triggered_from_node_execution_id": None,
+        "node_type": "start",
+        "triggered_from": "workflow-run",
+        "process_data": None,
+        "outputs": None,
+        "status": "succeeded",
+        "index": 0,
+        "created_at": 0,
+        "log_version": 1,
+        "__time__": 1,
+    }
+    rows = [{**defaults, **row} for row in rows]
+    fetched_payload_ids: list[str] = []
+    with closing(sqlite3.connect(":memory:")) as database:
+        database.row_factory = sqlite3.Row
+        database.execute(
+            "CREATE TABLE workflow_node_execution (id TEXT, tenant_id TEXT, app_id TEXT, workflow_run_id TEXT, "
+            "node_execution_id TEXT, workflow_id TEXT, triggered_from_workflow_id TEXT, "
+            "triggered_from_node_execution_id TEXT, node_type TEXT, triggered_from TEXT, "
+            "process_data TEXT, outputs TEXT, "
+            'status TEXT, "index" INTEGER, created_at INTEGER, log_version INTEGER, __time__ INTEGER)'
+        )
+        database.executemany(
+            "INSERT INTO workflow_node_execution VALUES "
+            "(:id, :tenant_id, :app_id, :workflow_run_id, :node_execution_id, :workflow_id, "
+            ":triggered_from_workflow_id, :triggered_from_node_execution_id, :node_type, :triggered_from, "
+            ":process_data, :outputs, :status, :index, :created_at, :log_version, :__time__)",
+            rows,
+        )
+
+        def execute_query(*, sql: str, **_kwargs: object) -> list[dict[str, object]]:
+            result = [dict(row) for row in database.execute(sql)]
+            fetched_payload_ids.extend(str(row["id"]) for row in result if "outputs" in row)
+            return result
+
+        repository.logstore_client.execute_sql.side_effect = execute_query
+        children = repository.get_workflow_tool_executions("tenant", "run", requested_id)
+        assert repository.get_workflow_tool_executions("tenant", "run", "unknown") == []
+        assert repository.get_workflow_tool_executions("tenant", "run", "child") == []
+
+    assert [child.id for child in children] == ["earlier-child", "child"]
+    assert children[1].status.value == "paused"
+    assert children[1].outputs_dict == {"answer": "approved"}
+    assert fetched_payload_ids == ["earlier-child", "child"]
+
+
 def test_load_full_process_data_returns_logstore_mapping() -> None:
     with patch("extensions.logstore.repositories.logstore_api_workflow_node_execution_repository.AliyunLogStore"):
         repository = LogstoreAPIWorkflowNodeExecutionRepository(session_maker=None)

api/tests/unit_tests/repositories/test_sqlalchemy_api_workflow_node_execution_repository.py

@@ -4,7 +4,7 @@ from unittest.mock import patch
 from uuid import uuid4
 
 import pytest
-from sqlalchemy import func, select
+from sqlalchemy import event, func, select
 from sqlalchemy.orm import Session, sessionmaker
 
 from core.workflow.node_execution_process_data import (
@@ -131,3 +131,88 @@ def test_workflow_history_and_last_execution_exclude_recursive_tool_nodes(
     assert [execution.id for execution in history] == [root_id]
     assert latest is not None
     assert latest.id == root_id
+
+
+def test_workflow_tool_children_are_scoped_to_exact_parent_and_run(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    parent_id = str(uuid4())
+    records = [
+        (parent_id, "tenant", "run", "caller-app", "tool", None, "parent-engine"),
+        ("direct-child", "tenant", "run", "source-app", "tool", "parent-engine", "child-engine"),
+        ("nested-child", "tenant", "run", "nested-app", "start", "child-engine", None),
+        ("other-call", "tenant", "run", "source-app", "start", "other-engine", None),
+        ("other-run", "tenant", "other-run", "source-app", "start", "parent-engine", None),
+        ("other-tenant", "other-tenant", "run", "source-app", "start", "parent-engine", None),
+        ("other-caller", "tenant", "run", "source-app", "start", "parent-engine", None),
+        ("legacy-child", "tenant", "run", "source-app", "start", None, None),
+    ]
+    with sqlite_session_factory.begin() as session:
+        session.add_all(
+            WorkflowNodeExecutionModel(
+                id=execution_id,
+                tenant_id=tenant,
+                app_id=app,
+                workflow_id=f"{app}-workflow",
+                workflow_run_id=run,
+                node_execution_id=engine_id,
+                node_id="same-node",
+                node_type=node_type,
+                title="Node",
+                index=index,
+                triggered_from_workflow_id=(
+                    "source-app-workflow"
+                    if execution_id in {"nested-child", "other-caller"}
+                    else "caller-app-workflow"
+                    if owner
+                    else None
+                ),
+                triggered_from_node_execution_id=owner,
+                process_data=None,
+                outputs='{"answer":"approved"}',
+                status=WorkflowNodeExecutionStatus.PAUSED
+                if execution_id == "direct-child"
+                else WorkflowNodeExecutionStatus.SUCCEEDED,
+                triggered_from=(
+                    WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN
+                    if execution_id == parent_id
+                    else WorkflowNodeExecutionTriggeredFrom.WORKFLOW_TOOL
+                ),
+                created_by_role=CreatorUserRole.ACCOUNT,
+                created_by="account",
+                created_at=datetime(2026, 1, 1) + timedelta(seconds=index),
+            )
+            for index, (execution_id, tenant, run, app, node_type, owner, engine_id) in enumerate(records)
+        )
+        session.add(
+            WorkflowNodeExecutionOffload(
+                tenant_id="tenant",
+                app_id="source-app",
+                node_execution_id="direct-child",
+                type_=ExecutionOffLoadType.PROCESS_DATA,
+                file_id=str(uuid4()),
+            )
+        )
+    repository = DifyAPISQLAlchemyWorkflowNodeExecutionRepository(sqlite_session_factory)
+    loaded_execution_ids: list[str] = []
+
+    def record_loaded_execution(execution: WorkflowNodeExecutionModel, _context: object) -> None:
+        loaded_execution_ids.append(execution.id)
+
+    for requested_id in (parent_id, "parent-engine"):
+        loaded_execution_ids.clear()
+        event.listen(WorkflowNodeExecutionModel, "load", record_loaded_execution)
+        try:
+            children = repository.get_workflow_tool_executions("tenant", "run", requested_id)
+        finally:
+            event.remove(WorkflowNodeExecutionModel, "load", record_loaded_execution)
+        assert [child.id for child in children] == ["direct-child"]
+        assert loaded_execution_ids == ["direct-child"]
+        assert children[0].app_id == "source-app"
+        assert children[0].outputs_dict == {"answer": "approved"}
+        assert children[0].status == WorkflowNodeExecutionStatus.PAUSED
+        assert children[0].process_data_truncated
+    assert [child.id for child in repository.get_workflow_tool_executions("tenant", "run", "direct-child")] == [
+        "nested-child"
+    ]
+    assert repository.get_workflow_tool_executions("tenant", "run", "unknown") == []

api/tests/unit_tests/services/test_paused_workflow_stop.py

@@ -76,7 +76,7 @@ def test_paused_stop_publishes_terminal_event_after_commit_and_retires_run_resou
     queued_node_saves: list[object] = []
     queued_saves: list[object] = []
     monkeypatch.setattr(
-        "core.repositories.celery_workflow_node_execution_repository.save_workflow_node_execution_task.delay",
+        "core.repositories.celery_workflow_node_execution_repository.save_workflow_node_executions_task.delay",
         lambda *args, **kwargs: queued_node_saves.append((args, kwargs)),
     )
     monkeypatch.setattr(
@@ -104,7 +104,10 @@ def test_paused_stop_publishes_terminal_event_after_commit_and_retires_run_resou
     pause = WorkflowPause(workflow_id=run.workflow_id, workflow_run_id=run.id, state_object_key="state")
     sqlite_session.add_all([run, pause])
     sqlite_session.commit()
-    for node_app, origin in (("app-1", WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN),):
+    for node_app, origin in (
+        ("app-1", WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN),
+        ("source-app", WorkflowNodeExecutionTriggeredFrom.WORKFLOW_TOOL),
+    ):
         SQLAlchemyWorkflowNodeExecutionRepository(
             session_factory=sqlite_session_factory,
             tenant_id="tenant-1",
@@ -199,6 +202,7 @@ def test_paused_stop_publishes_terminal_event_after_commit_and_retires_run_resou
         )
         assert {(record["app_id"], record["triggered_from"], record["status"]) for record in node_history} == {
             ("app-1", "workflow-run", "failed"),
+            ("source-app", "workflow-tool", "failed"),
         }
 
 

api/tests/unit_tests/services/test_workflow_run_service.py

@@ -1,10 +1,14 @@
 """Unit tests for the Console workflow-run application service."""
 
 from types import SimpleNamespace
-from unittest.mock import ANY, MagicMock
+from unittest.mock import ANY, MagicMock, create_autospec
 
 import pytest
 
+from core.workflow.node_execution_process_data import (
+    WORKFLOW_TOOL_INVOCATION_ID_KEY,
+    WORKFLOW_TOOL_ROOT_APP_ID_KEY,
+)
 from graphon.enums import WorkflowExecutionStatus
 from machinery.context import RequestContext
 from models import WorkflowRun, WorkflowRunTriggeredFrom, WorkflowType
@@ -56,6 +60,46 @@ def _workflow_run(
 
 
 class TestWorkflowRunServiceQueries:
+    def test_workflow_tool_details_require_the_root_app_run_and_hide_internal_correlation(
+        self, service_dependencies: tuple[MagicMock, MagicMock], monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        node_executions, workflow_runs = service_dependencies
+        service = _service(service_dependencies)
+        workflow_runs.get_workflow_run_by_id.return_value = None
+        admitted_run = MagicMock(return_value=None)
+        monkeypatch.setattr(service, "get_workflow_run", admitted_run)
+        context = _request_context()
+        assert (
+            service.get_workflow_tool_node_executions(
+                context, app_id="app-1", run_id="run-1", node_execution_id="parent-1"
+            )
+            is None
+        )
+        node_executions.get_workflow_tool_executions.assert_not_called()
+
+        admitted_run.return_value = _workflow_run()
+        trace = SimpleNamespace(
+            process_data={
+                "answer": "approved",
+                WORKFLOW_TOOL_INVOCATION_ID_KEY: "invocation-1",
+                WORKFLOW_TOOL_ROOT_APP_ID_KEY: "app-1",
+            }
+        )
+        assemble = create_autospec(service_module.assemble_workflow_node_execution_traces, return_value=[trace])
+        monkeypatch.setattr(service_module, "assemble_workflow_node_execution_traces", assemble)
+        result = service.get_workflow_tool_node_executions(
+            context, app_id="app-1", run_id="run-1", node_execution_id="parent-1"
+        )
+        admitted_run.assert_called_with(context, app_id="app-1", run_id="run-1")
+        node_executions.get_workflow_tool_executions.assert_called_once_with(
+            tenant_id="tenant-1", workflow_run_id="run-1", parent_node_execution_id="parent-1"
+        )
+        assemble.assert_called_once_with(
+            node_executions.get_workflow_tool_executions.return_value, node_executions, session=ANY
+        )
+        assert result == [trace]
+        assert trace.process_data == {"answer": "approved"}
+
     def test_get_paginate_workflow_runs_should_forward_filters_and_parse_limit(
         self,
         service_dependencies: tuple[MagicMock, MagicMock],

api/tests/unit_tests/services/workflow/test_workflow_event_snapshot_service.py

@@ -32,7 +32,7 @@ from libs.datetime_utils import to_utc_timestamp
 from models.enums import ConversationFromSource, CreatorUserRole
 from models.human_input import HumanInputForm, HumanInputFormRecipient, RecipientType
 from models.model import AppMode, Message
-from models.workflow import WorkflowPause, WorkflowRun
+from models.workflow import WorkflowNodeExecutionModel, WorkflowPause, WorkflowRun
 from repositories.api_workflow_node_execution_repository import WorkflowNodeExecutionSnapshot
 from repositories.entities.workflow_pause import WorkflowPauseEntity
 from repositories.sqlalchemy_api_workflow_node_execution_repository import (
@@ -717,6 +717,71 @@ def test_start_buffering_should_treat_closed_subscription_as_done(caplog: pytest
     assert "Failed while buffering workflow events" not in caplog.text
 
 
+@pytest.mark.parametrize("include_node_details", [False, True])
+def test_console_snapshot_replays_node_details_without_exposing_them_by_default(
+    monkeypatch: pytest.MonkeyPatch,
+    unbound_session_factory: sessionmaker[Session],
+    include_node_details: bool,
+) -> None:
+    snapshot = _build_snapshot(WorkflowNodeExecutionStatus.SUCCEEDED)
+    execution = WorkflowNodeExecutionModel(
+        id="row-1",
+        node_execution_id=snapshot.execution_id,
+        node_type=snapshot.node_type,
+        predecessor_node_id="start",
+        inputs=json.dumps({"request": "review"}),
+        process_data=json.dumps({"action": "approve"}),
+        outputs=json.dumps({"approved_text": "approved"}),
+        execution_metadata=json.dumps({"total_tokens": 7}),
+        offload_data=[],
+    )
+    node_repo = SimpleNamespace(
+        get_execution_snapshots_by_workflow_run=MagicMock(return_value=[snapshot]),
+        get_executions_by_workflow_run=MagicMock(return_value=[execution]),
+    )
+    factory = SimpleNamespace(
+        create_api_workflow_run_repository=MagicMock(),
+        create_api_workflow_node_execution_repository=MagicMock(return_value=node_repo),
+    )
+    monkeypatch.setattr(service_module, "DifyAPIRepositoryFactory", factory)
+    monkeypatch.setattr(
+        service_module.MessageGenerator, "get_response_topic", lambda *_args: _Topic(_StaticSubscription())
+    )
+    buffer_state = BufferState(queue.Queue(), Event(), Event(), Event(), task_id_hint="task-1")
+    buffer_state.done_event.set()
+    monkeypatch.setattr(service_module, "_start_buffering", lambda _sub: buffer_state)
+
+    events = list(
+        build_workflow_event_stream(
+            app_mode=AppMode.WORKFLOW,
+            workflow_run=_build_workflow_run(WorkflowExecutionStatus.RUNNING),
+            tenant_id="tenant-1",
+            app_id="app-1",
+            session_maker=unbound_session_factory,
+            **({"include_node_details": True} if include_node_details else {}),
+        )
+    )
+    started, finished = [
+        event["data"]
+        for event in events
+        if isinstance(event, Mapping) and event["event"] in (StreamEvent.NODE_STARTED, StreamEvent.NODE_FINISHED)
+    ]
+    assert started["id"] == finished["id"] == snapshot.execution_id
+    assert started["index"] == finished["index"] == 1
+    if include_node_details:
+        node_repo.get_executions_by_workflow_run.assert_called_once_with(
+            tenant_id="tenant-1", app_id="app-1", workflow_run_id="run-1"
+        )
+        assert started["inputs"] == finished["inputs"] == {"request": "review"}
+        assert finished["process_data"] == {"action": "approve"}
+        assert finished["outputs"] == {"approved_text": "approved"}
+        assert finished["execution_metadata"] == {"total_tokens": 7}
+    else:
+        node_repo.get_executions_by_workflow_run.assert_not_called()
+        assert started["inputs"] is None
+        assert all(finished[key] is None for key in ("inputs", "process_data", "outputs", "execution_metadata"))
+
+
 def test_build_workflow_event_stream_should_emit_ping_and_terminal_snapshot_event(
     monkeypatch: pytest.MonkeyPatch,
     unbound_session_factory: sessionmaker[Session],

api/tests/unit_tests/services/workflow/test_workflow_run_agg.py

@@ -588,6 +588,18 @@ def test_active_execution_snapshot_failure_is_not_published_as_a_pause(sqlite_en
         assert session.scalar(select(WorkflowPause)) is None
 
 
+def test_legacy_caller_rejects_human_input_before_creating_run_or_form(sqlite_engine: Engine) -> None:
+    from models.human_input import HumanInputForm
+
+    runner, workflow_repository, node_repository = make_workflow_runner(sqlite_engine, human_input=True)
+    runner.application_generate_entity.allow_human_input = False
+    with pytest.raises(ValueError, match="require Engine-managed container execution"):
+        WorkflowRunAgg.run(runner, None, workflow_repository, node_repository)
+    with Session(sqlite_engine) as session:
+        assert session.scalar(select(WorkflowRun)) is None
+        assert session.scalar(select(HumanInputForm)) is None
+
+
 @pytest.mark.usefixtures("_local_storage")
 @pytest.mark.parametrize("stop_timing", ["before_summary", "during_upload"])
 def test_stop_wins_while_engine_is_persisting_a_pause(
@@ -651,15 +663,3 @@ def test_stop_wins_while_engine_is_persisting_a_pause(
         forms = session.scalars(select(HumanInputForm).where(HumanInputForm.workflow_run_id == run.id)).all()
         assert forms
         assert all(form.status == HumanInputFormStatus.EXPIRED for form in forms)
-
-
-def test_legacy_caller_rejects_human_input_before_creating_run_or_form(sqlite_engine: Engine) -> None:
-    from models.human_input import HumanInputForm
-
-    runner, workflow_repository, node_repository = make_workflow_runner(sqlite_engine, human_input=True)
-    runner.application_generate_entity.allow_human_input = False
-    with pytest.raises(ValueError, match="require Engine-managed container execution"):
-        WorkflowRunAgg.run(runner, None, workflow_repository, node_repository)
-    with Session(sqlite_engine) as session:
-        assert session.scalar(select(WorkflowRun)) is None
-        assert session.scalar(select(HumanInputForm)) is None

api/tests/unit_tests/tasks/test_workflow_node_execution_tasks.py

@@ -22,6 +22,7 @@ from tasks.workflow_node_execution_tasks import (
     _create_node_execution_from_domain,
     _update_node_execution_from_domain,
     save_workflow_node_execution_task,
+    save_workflow_node_executions_task,
 )
 
 TENANT_ID = "00000000-0000-0000-0000-000000000010"
@@ -343,7 +344,7 @@ def test_task_repeated_delivery_updates_one_row(
     assert persisted.created_at == created_at.replace(tzinfo=None)
 
 
-@pytest.mark.parametrize("writer", ["celery", "sql", "sql-data"])
+@pytest.mark.parametrize("writer", ["celery", "celery-batch", "sql", "sql-batch", "sql-data"])
 @pytest.mark.parametrize("existing", [False, True])
 @pytest.mark.parametrize(
     "status",
@@ -378,7 +379,7 @@ def test_delayed_nonterminal_node_write_cannot_revive_or_insert_a_live_node_afte
             node.error = "User requested stop"
             session.add(node)
     execution = _execution(status=status)
-    if writer == "celery":
+    if writer.startswith("celery"):
         kwargs: _ExecutionTaskKwargs = {
             "tenant_id": TENANT_ID,
             "app_id": APP_ID,
@@ -386,7 +387,10 @@ def test_delayed_nonterminal_node_write_cannot_revive_or_insert_a_live_node_afte
             "creator_user_id": ACCOUNT_ID,
             "creator_user_role": CreatorUserRole.ACCOUNT.value,
         }
-        assert save_workflow_node_execution_task.run(execution_data=execution.model_dump(), **kwargs)
+        if writer == "celery":
+            assert save_workflow_node_execution_task.run(execution_data=execution.model_dump(), **kwargs)
+        else:
+            assert save_workflow_node_executions_task.run(executions_data=[execution.model_dump()], **kwargs)
     else:
         user = Account(name="Test", email="test@example.com")
         user.id = ACCOUNT_ID
@@ -399,6 +403,8 @@ def test_delayed_nonterminal_node_write_cannot_revive_or_insert_a_live_node_afte
         )
         if writer == "sql":
             repository.save(execution)
+        elif writer == "sql-batch":
+            repository.save_many([execution])
         else:
             repository.save_execution_data(execution)
     with sqlite_session_factory() as session:

packages/contracts/generated/api/console/apps/orpc.gen.ts

@@ -213,6 +213,8 @@ import {
   zGetAppsByAppIdWorkflowCommentsMentionUsersResponse,
   zGetAppsByAppIdWorkflowCommentsPath,
   zGetAppsByAppIdWorkflowCommentsResponse,
+  zGetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsByNodeExecutionIdChildrenPath,
+  zGetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsByNodeExecutionIdChildrenResponse,
   zGetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsPath,
   zGetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsResponse,
   zGetAppsByAppIdWorkflowRunsByRunIdPath,
@@ -2881,12 +2883,39 @@ export const tasks = {
   byTaskId: byTaskId3,
 }
 
+/**
+ * Get the internal node executions of one workflow tool invocation
+ */
+export const get48 = oc
+  .route({
+    description: 'Get the internal node executions of one workflow tool invocation',
+    inputStructure: 'detailed',
+    method: 'GET',
+    operationId: 'getAppsByAppIdWorkflowRunsByRunIdNodeExecutionsByNodeExecutionIdChildren',
+    path: '/apps/{app_id}/workflow-runs/{run_id}/node-executions/{node_execution_id}/children',
+    tags: ['console'],
+  })
+  .input(
+    z.object({
+      params: zGetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsByNodeExecutionIdChildrenPath,
+    }),
+  )
+  .output(zGetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsByNodeExecutionIdChildrenResponse)
+
+export const children = {
+  get: get48,
+}
+
+export const byNodeExecutionId = {
+  children,
+}
+
 /**
  * Get workflow run node execution list
  *
  * Get workflow run node execution list
  */
-export const get48 = oc
+export const get49 = oc
   .route({
     description: 'Get workflow run node execution list',
     inputStructure: 'detailed',
@@ -2901,7 +2930,8 @@ export const get48 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const nodeExecutions = {
-  get: get48,
+  get: get49,
+  byNodeExecutionId,
 }
 
 /**
@@ -2909,7 +2939,7 @@ export const nodeExecutions = {
  *
  * Get workflow run detail
  */
-export const get49 = oc
+export const get50 = oc
   .route({
     description: 'Get workflow run detail',
     inputStructure: 'detailed',
@@ -2924,7 +2954,7 @@ export const get49 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const byRunId = {
-  get: get49,
+  get: get50,
   nodeExecutions,
 }
 
@@ -2963,7 +2993,7 @@ export const download4 = {
 /**
  * Read a text/binary preview file in a workflow Agent node sandbox
  */
-export const get50 = oc
+export const get51 = oc
   .route({
     description: 'Read a text/binary preview file in a workflow Agent node sandbox',
     inputStructure: 'detailed',
@@ -2987,13 +3017,13 @@ export const get50 = oc
   })
 
 export const read = {
-  get: get50,
+  get: get51,
 }
 
 /**
  * List a directory in a workflow Agent node sandbox
  */
-export const get51 = oc
+export const get52 = oc
   .route({
     description: 'List a directory in a workflow Agent node sandbox',
     inputStructure: 'detailed',
@@ -3017,7 +3047,7 @@ export const get51 = oc
   })
 
 export const files3 = {
-  get: get51,
+  get: get52,
   download: download4,
   read,
 }
@@ -3043,7 +3073,7 @@ export const byWorkflowRunId = {
  *
  * Get workflow run list
  */
-export const get52 = oc
+export const get53 = oc
   .route({
     description: 'Get workflow run list',
     inputStructure: 'detailed',
@@ -3063,7 +3093,7 @@ export const get52 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const workflowRuns2 = {
-  get: get52,
+  get: get53,
   count: count3,
   tasks,
   byRunId,
@@ -3075,7 +3105,7 @@ export const workflowRuns2 = {
  *
  * Get all users in current tenant for mentions
  */
-export const get53 = oc
+export const get54 = oc
   .route({
     description: 'Get all users in current tenant for mentions',
     inputStructure: 'detailed',
@@ -3093,7 +3123,7 @@ export const get53 = oc
   })
 
 export const mentionUsers = {
-  get: get53,
+  get: get54,
 }
 
 /**
@@ -3238,7 +3268,7 @@ export const delete10 = oc
  *
  * Get a specific workflow comment
  */
-export const get54 = oc
+export const get55 = oc
   .route({
     description: 'Get a specific workflow comment',
     inputStructure: 'detailed',
@@ -3284,7 +3314,7 @@ export const put3 = oc
 
 export const byCommentId = {
   delete: delete10,
-  get: get54,
+  get: get55,
   put: put3,
   replies,
   resolve,
@@ -3295,7 +3325,7 @@ export const byCommentId = {
  *
  * Get all comments for a workflow
  */
-export const get55 = oc
+export const get56 = oc
   .route({
     description: 'Get all comments for a workflow',
     inputStructure: 'detailed',
@@ -3341,7 +3371,7 @@ export const post42 = oc
   })
 
 export const comments = {
-  get: get55,
+  get: get56,
   post: post42,
   mentionUsers,
   byCommentId,
@@ -3350,7 +3380,7 @@ export const comments = {
 /**
  * Get workflow average app interaction statistics
  */
-export const get56 = oc
+export const get57 = oc
   .route({
     description: 'Get workflow average app interaction statistics',
     inputStructure: 'detailed',
@@ -3369,13 +3399,13 @@ export const get56 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const averageAppInteractions = {
-  get: get56,
+  get: get57,
 }
 
 /**
  * Get workflow daily runs statistics
  */
-export const get57 = oc
+export const get58 = oc
   .route({
     description: 'Get workflow daily runs statistics',
     inputStructure: 'detailed',
@@ -3394,13 +3424,13 @@ export const get57 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const dailyConversations2 = {
-  get: get57,
+  get: get58,
 }
 
 /**
  * Get workflow daily terminals statistics
  */
-export const get58 = oc
+export const get59 = oc
   .route({
     description: 'Get workflow daily terminals statistics',
     inputStructure: 'detailed',
@@ -3419,13 +3449,13 @@ export const get58 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const dailyTerminals = {
-  get: get58,
+  get: get59,
 }
 
 /**
  * Get workflow daily token cost statistics
  */
-export const get59 = oc
+export const get60 = oc
   .route({
     description: 'Get workflow daily token cost statistics',
     inputStructure: 'detailed',
@@ -3444,7 +3474,7 @@ export const get59 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const tokenCosts2 = {
-  get: get59,
+  get: get60,
 }
 
 export const statistics2 = {
@@ -3464,7 +3494,7 @@ export const workflow = {
  *
  * Get default block configuration by type
  */
-export const get60 = oc
+export const get61 = oc
   .route({
     description: 'Get default block configuration by type',
     inputStructure: 'detailed',
@@ -3484,7 +3514,7 @@ export const get60 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const byBlockType = {
-  get: get60,
+  get: get61,
 }
 
 /**
@@ -3492,7 +3522,7 @@ export const byBlockType = {
  *
  * Get default block configurations for workflow
  */
-export const get61 = oc
+export const get62 = oc
   .route({
     description: 'Get default block configurations for workflow',
     inputStructure: 'detailed',
@@ -3507,14 +3537,14 @@ export const get61 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const defaultWorkflowBlockConfigs = {
-  get: get61,
+  get: get62,
   byBlockType,
 }
 
 /**
  * Get conversation variables for workflow
  */
-export const get62 = oc
+export const get63 = oc
   .route({
     description: 'Get conversation variables for workflow',
     inputStructure: 'detailed',
@@ -3549,7 +3579,7 @@ export const post43 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const conversationVariables2 = {
-  get: get62,
+  get: get63,
   post: post43,
 }
 
@@ -3558,7 +3588,7 @@ export const conversationVariables2 = {
  *
  * Get environment variables for workflow
  */
-export const get63 = oc
+export const get64 = oc
   .route({
     description: 'Get environment variables for workflow',
     inputStructure: 'detailed',
@@ -3597,7 +3627,7 @@ export const post44 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const environmentVariables = {
-  get: get63,
+  get: get64,
   post: post44,
 }
 
@@ -3808,7 +3838,7 @@ export const loop2 = {
   nodes: nodes6,
 }
 
-export const get64 = oc
+export const get65 = oc
   .route({
     inputStructure: 'detailed',
     method: 'GET',
@@ -3823,7 +3853,7 @@ export const get64 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const candidates = {
-  get: get64,
+  get: get65,
 }
 
 export const post51 = oc
@@ -3929,7 +3959,7 @@ export const validate = {
   post: post54,
 }
 
-export const get65 = oc
+export const get66 = oc
   .route({
     inputStructure: 'detailed',
     method: 'GET',
@@ -3976,7 +4006,7 @@ export const put4 = oc
   })
 
 export const agentComposer = {
-  get: get65,
+  get: get66,
   put: put4,
   candidates,
   copyFromRoster,
@@ -3988,7 +4018,7 @@ export const agentComposer = {
 /**
  * Get last run result for draft workflow node
  */
-export const get66 = oc
+export const get67 = oc
   .route({
     description: 'Get last run result for draft workflow node',
     inputStructure: 'detailed',
@@ -4002,7 +4032,7 @@ export const get66 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const lastRun = {
-  get: get66,
+  get: get67,
 }
 
 /**
@@ -4080,7 +4110,7 @@ export const delete11 = oc
 /**
  * Get variables for a specific node
  */
-export const get67 = oc
+export const get68 = oc
   .route({
     description: 'Get variables for a specific node',
     inputStructure: 'detailed',
@@ -4095,7 +4125,7 @@ export const get67 = oc
 
 export const variables = {
   delete: delete11,
-  get: get67,
+  get: get68,
 }
 
 export const byNodeId8 = {
@@ -4144,7 +4174,7 @@ export const run10 = {
 /**
  * Server-Sent Events stream of inspector deltas for a draft workflow run.
  */
-export const get68 = oc
+export const get69 = oc
   .route({
     description: 'Server-Sent Events stream of inspector deltas for a draft workflow run.',
     inputStructure: 'detailed',
@@ -4161,13 +4191,13 @@ export const get68 = oc
   })
 
 export const events = {
-  get: get68,
+  get: get69,
 }
 
 /**
  * Full value for one declared output, including signed download URL for files.
  */
-export const get69 = oc
+export const get70 = oc
   .route({
     description: 'Full value for one declared output, including signed download URL for files.',
     inputStructure: 'detailed',
@@ -4190,7 +4220,7 @@ export const get69 = oc
   })
 
 export const preview5 = {
-  get: get69,
+  get: get70,
 }
 
 export const byOutputName = {
@@ -4200,7 +4230,7 @@ export const byOutputName = {
 /**
  * One node's declared outputs for a draft workflow run.
  */
-export const get70 = oc
+export const get71 = oc
   .route({
     description: "One node's declared outputs for a draft workflow run.",
     inputStructure: 'detailed',
@@ -4218,14 +4248,14 @@ export const get70 = oc
   })
 
 export const byNodeId9 = {
-  get: get70,
+  get: get71,
   byOutputName,
 }
 
 /**
  * Snapshot of every node's declared outputs for a draft workflow run.
  */
-export const get71 = oc
+export const get72 = oc
   .route({
     description: "Snapshot of every node's declared outputs for a draft workflow run.",
     inputStructure: 'detailed',
@@ -4242,7 +4272,7 @@ export const get71 = oc
   })
 
 export const nodeOutputs = {
-  get: get71,
+  get: get72,
   events,
   byNodeId: byNodeId9,
 }
@@ -4258,7 +4288,7 @@ export const runs = {
 /**
  * Get system variables for workflow
  */
-export const get72 = oc
+export const get73 = oc
   .route({
     description: 'Get system variables for workflow',
     inputStructure: 'detailed',
@@ -4272,7 +4302,7 @@ export const get72 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const systemVariables = {
-  get: get72,
+  get: get73,
 }
 
 /**
@@ -4382,7 +4412,7 @@ export const delete12 = oc
 /**
  * Get a specific workflow variable
  */
-export const get73 = oc
+export const get74 = oc
   .route({
     description: 'Get a specific workflow variable',
     inputStructure: 'detailed',
@@ -4418,7 +4448,7 @@ export const patch2 = oc
 
 export const byVariableId = {
   delete: delete12,
-  get: get73,
+  get: get74,
   patch: patch2,
   reset,
 }
@@ -4445,7 +4475,7 @@ export const delete13 = oc
  *
  * Get draft workflow variables
  */
-export const get74 = oc
+export const get75 = oc
   .route({
     description: 'Get draft workflow variables',
     inputStructure: 'detailed',
@@ -4469,7 +4499,7 @@ export const get74 = oc
 
 export const variables2 = {
   delete: delete13,
-  get: get74,
+  get: get75,
   byVariableId,
 }
 
@@ -4478,7 +4508,7 @@ export const variables2 = {
  *
  * Get draft workflow for an application
  */
-export const get75 = oc
+export const get76 = oc
   .route({
     description: 'Get draft workflow for an application',
     inputStructure: 'detailed',
@@ -4523,7 +4553,7 @@ export const post60 = oc
   })
 
 export const draft2 = {
-  get: get75,
+  get: get76,
   post: post60,
   conversationVariables: conversationVariables2,
   environmentVariables,
@@ -4544,7 +4574,7 @@ export const draft2 = {
  *
  * Get published workflow for an application
  */
-export const get76 = oc
+export const get77 = oc
   .route({
     description: 'Get published workflow for an application',
     inputStructure: 'detailed',
@@ -4580,14 +4610,14 @@ export const post61 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const publish = {
-  get: get76,
+  get: get77,
   post: post61,
 }
 
 /**
  * Server-Sent Events stream of inspector deltas for a published workflow run.
  */
-export const get77 = oc
+export const get78 = oc
   .route({
     description: 'Server-Sent Events stream of inspector deltas for a published workflow run.',
     inputStructure: 'detailed',
@@ -4604,13 +4634,13 @@ export const get77 = oc
   })
 
 export const events2 = {
-  get: get77,
+  get: get78,
 }
 
 /**
  * Full value for one declared output of a published run.
  */
-export const get78 = oc
+export const get79 = oc
   .route({
     description: 'Full value for one declared output of a published run.',
     inputStructure: 'detailed',
@@ -4637,7 +4667,7 @@ export const get78 = oc
   })
 
 export const preview6 = {
-  get: get78,
+  get: get79,
 }
 
 export const byOutputName2 = {
@@ -4647,7 +4677,7 @@ export const byOutputName2 = {
 /**
  * One node's declared outputs for a published workflow run.
  */
-export const get79 = oc
+export const get80 = oc
   .route({
     description: "One node's declared outputs for a published workflow run.",
     inputStructure: 'detailed',
@@ -4665,14 +4695,14 @@ export const get79 = oc
   })
 
 export const byNodeId10 = {
-  get: get79,
+  get: get80,
   byOutputName: byOutputName2,
 }
 
 /**
  * Snapshot of every node's declared outputs for a published workflow run.
  */
-export const get80 = oc
+export const get81 = oc
   .route({
     description: "Snapshot of every node's declared outputs for a published workflow run.",
     inputStructure: 'detailed',
@@ -4689,7 +4719,7 @@ export const get80 = oc
   })
 
 export const nodeOutputs2 = {
-  get: get80,
+  get: get81,
   events: events2,
   byNodeId: byNodeId10,
 }
@@ -4709,7 +4739,7 @@ export const published = {
 /**
  * Get webhook trigger for a node
  */
-export const get81 = oc
+export const get82 = oc
   .route({
     inputStructure: 'detailed',
     method: 'GET',
@@ -4728,7 +4758,7 @@ export const get81 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const webhook = {
-  get: get81,
+  get: get82,
 }
 
 export const triggers2 = {
@@ -4807,7 +4837,7 @@ export const byWorkflowId = {
  *
  * Get all published workflows for an application
  */
-export const get82 = oc
+export const get83 = oc
   .route({
     description: 'Get all published workflows for an application',
     inputStructure: 'detailed',
@@ -4827,7 +4857,7 @@ export const get82 = oc
   .errors({ app_not_found: { status: 404, data: zAppNotFoundError } })
 
 export const workflows3 = {
-  get: get82,
+  get: get83,
   defaultWorkflowBlockConfigs,
   draft: draft2,
   publish,
@@ -4855,7 +4885,7 @@ export const delete15 = oc
 /**
  * Get application details
  */
-export const get83 = oc
+export const get84 = oc
   .route({
     description: 'Get application details',
     inputStructure: 'detailed',
@@ -4884,7 +4914,7 @@ export const put6 = oc
 
 export const byAppId2 = {
   delete: delete15,
-  get: get83,
+  get: get84,
   put: put6,
   advancedChat,
   agent,
@@ -4947,7 +4977,7 @@ export const byApiKeyId = {
 /**
  * Get all API keys for an app
  */
-export const get84 = oc
+export const get85 = oc
   .route({
     description: 'Get all API keys for an app',
     inputStructure: 'detailed',
@@ -4980,7 +5010,7 @@ export const post63 = oc
   })
 
 export const apiKeys = {
-  get: get84,
+  get: get85,
   post: post63,
   byApiKeyId,
 }
@@ -4992,7 +5022,7 @@ export const byResourceId = {
 /**
  * Get list of applications with pagination and filtering
  */
-export const get85 = oc
+export const get86 = oc
   .route({
     description: 'Get list of applications with pagination and filtering',
     inputStructure: 'detailed',
@@ -5021,7 +5051,7 @@ export const post64 = oc
   .output(zPostAppsResponse)
 
 export const apps = {
-  get: get85,
+  get: get86,
   post: post64,
   imports,
   recent,

packages/contracts/generated/api/console/apps/types.gen.ts

@@ -949,6 +949,10 @@ export type WorkflowRunNodeExecutionListResponse = {
   data: Array<WorkflowRunNodeExecutionResponse>
 }
 
+export type WorkflowToolNodeExecutionListResponse = {
+  data: Array<WorkflowToolNodeExecutionResponse>
+}
+
 export type SandboxListResponse = {
   entries?: Array<SandboxFileEntryResponse>
   path: string
@@ -2166,6 +2170,35 @@ export type SimpleEndUser = {
   type: string
 }
 
+export type WorkflowToolNodeExecutionResponse = {
+  app_id: string
+  created_at?: number | null
+  created_by_account?: SimpleAccountResponse | null
+  created_by_end_user?: SimpleEndUser | null
+  created_by_role?: string | null
+  elapsed_time?: number | null
+  error?: string | null
+  execution_metadata?: unknown
+  extras?: unknown
+  finished_at?: number | null
+  id: string
+  index?: number | null
+  inputs?: unknown
+  inputs_truncated?: boolean | null
+  node_execution_id?: string | null
+  node_id?: string | null
+  node_type?: string | null
+  outputs?: unknown
+  outputs_truncated?: boolean | null
+  predecessor_node_id?: string | null
+  process_data?: unknown
+  process_data_truncated?: boolean | null
+  retry_index?: number | null
+  status?: string | null
+  title?: string | null
+  workflow_id: string
+}
+
 export type SandboxFileEntryResponse = {
   mtime?: number | null
   name: string
@@ -7215,6 +7248,28 @@ export type GetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsResponses = {
 export type GetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsResponse =
   GetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsResponses[keyof GetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsResponses]
 
+export type GetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsByNodeExecutionIdChildrenData = {
+  body?: never
+  path: {
+    app_id: string
+    node_execution_id: string
+    run_id: string
+  }
+  query?: never
+  url: '/apps/{app_id}/workflow-runs/{run_id}/node-executions/{node_execution_id}/children'
+}
+
+export type GetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsByNodeExecutionIdChildrenErrors = {
+  404: unknown
+}
+
+export type GetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsByNodeExecutionIdChildrenResponses = {
+  200: WorkflowToolNodeExecutionListResponse
+}
+
+export type GetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsByNodeExecutionIdChildrenResponse =
+  GetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsByNodeExecutionIdChildrenResponses[keyof GetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsByNodeExecutionIdChildrenResponses]
+
 export type GetAppsByAppIdWorkflowRunsByWorkflowRunIdAgentNodesByNodeIdSandboxFilesData = {
   body?: never
   path: {

packages/contracts/generated/api/console/apps/zod.gen.ts

@@ -2079,6 +2079,45 @@ export const zWorkflowRunNodeExecutionListResponse = z.object({
   data: z.array(zWorkflowRunNodeExecutionResponse),
 })
 
+/**
+ * WorkflowToolNodeExecutionResponse
+ */
+export const zWorkflowToolNodeExecutionResponse = z.object({
+  app_id: z.string(),
+  created_at: z.int().nullish(),
+  created_by_account: zSimpleAccountResponse.nullish(),
+  created_by_end_user: zSimpleEndUser.nullish(),
+  created_by_role: z.string().nullish(),
+  elapsed_time: z.number().nullish(),
+  error: z.string().nullish(),
+  execution_metadata: z.unknown().optional(),
+  extras: z.unknown().optional(),
+  finished_at: z.int().nullish(),
+  id: z.string(),
+  index: z.int().nullish(),
+  inputs: z.unknown().optional(),
+  inputs_truncated: z.boolean().nullish(),
+  node_execution_id: z.string().nullish(),
+  node_id: z.string().nullish(),
+  node_type: z.string().nullish(),
+  outputs: z.unknown().optional(),
+  outputs_truncated: z.boolean().nullish(),
+  predecessor_node_id: z.string().nullish(),
+  process_data: z.unknown().optional(),
+  process_data_truncated: z.boolean().nullish(),
+  retry_index: z.int().nullish(),
+  status: z.string().nullish(),
+  title: z.string().nullish(),
+  workflow_id: z.string(),
+})
+
+/**
+ * WorkflowToolNodeExecutionListResponse
+ */
+export const zWorkflowToolNodeExecutionListResponse = z.object({
+  data: z.array(zWorkflowToolNodeExecutionResponse),
+})
+
 /**
  * SandboxFileEntryResponse
  */
@@ -7300,6 +7339,19 @@ export const zGetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsPath = z.object({
 export const zGetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsResponse =
   zWorkflowRunNodeExecutionListResponse
 
+export const zGetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsByNodeExecutionIdChildrenPath =
+  z.object({
+    app_id: z.uuid(),
+    node_execution_id: z.string(),
+    run_id: z.uuid(),
+  })
+
+/**
+ * Internal node executions retrieved successfully
+ */
+export const zGetAppsByAppIdWorkflowRunsByRunIdNodeExecutionsByNodeExecutionIdChildrenResponse =
+  zWorkflowToolNodeExecutionListResponse
+
 export const zGetAppsByAppIdWorkflowRunsByWorkflowRunIdAgentNodesByNodeIdSandboxFilesPath =
   z.object({
     app_id: z.uuid(),