import json import logging import time from collections.abc import Mapping, Sequence from dataclasses import dataclass from datetime import datetime from typing import Any, NewType, TypedDict, Union from sqlalchemy import select from sqlalchemy.orm import Session from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom, WorkflowAppGenerateEntity from core.app.entities.queue_entities import ( QueueAgentLogEvent, QueueHumanInputFormFilledEvent, QueueHumanInputFormTimeoutEvent, QueueIterationCompletedEvent, QueueIterationNextEvent, QueueIterationStartEvent, QueueLoopCompletedEvent, QueueLoopNextEvent, QueueLoopStartEvent, QueueNodeExceptionEvent, QueueNodeFailedEvent, QueueNodeRetryEvent, QueueNodeStartedEvent, QueueNodeSucceededEvent, QueueWorkflowPausedEvent, ) from core.app.entities.task_entities import ( AgentLogStreamResponse, HumanInputFormFilledResponse, HumanInputFormTimeoutResponse, HumanInputRequiredResponse, IterationNodeCompletedStreamResponse, IterationNodeNextStreamResponse, IterationNodeStartStreamResponse, LoopNodeCompletedStreamResponse, LoopNodeNextStreamResponse, LoopNodeStartStreamResponse, NodeFinishStreamResponse, NodeRetryStreamResponse, NodeStartStreamResponse, StreamResponse, WorkflowFinishStreamResponse, WorkflowPauseStreamResponse, WorkflowStartStreamResponse, ) from core.plugin.impl.datasource import PluginDatasourceManager from core.tools.entities.tool_entities import ToolProviderType from core.tools.tool_manager import ToolManager from core.trigger.constants import TRIGGER_PLUGIN_NODE_TYPE from core.trigger.trigger_manager import TriggerManager from core.workflow.human_input_forms import load_form_tokens_by_form_id from core.workflow.human_input_policy import HumanInputSurface, enrich_human_input_pause_reasons # Maps the entry surface a workflow was invoked from to the HITL surface that # its resume tokens must be filtered for. Surfaces not in this map fall back to # the general priority ordering (typically CONSOLE > BACKSTAGE). _INVOKE_FROM_TO_HITL_SURFACE: Mapping[InvokeFrom, HumanInputSurface] = { InvokeFrom.SERVICE_API: HumanInputSurface.SERVICE_API, InvokeFrom.OPENAPI: HumanInputSurface.OPENAPI, } from core.workflow.system_variables import SystemVariableKey, system_variables_to_mapping from core.workflow.workflow_entry import WorkflowEntry from extensions.ext_database import db from graphon.entities import WorkflowStartReason from graphon.entities.pause_reason import HumanInputRequired from graphon.enums import ( BuiltinNodeTypes, WorkflowExecutionStatus, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus, ) from graphon.file import FILE_MODEL_IDENTITY, File from graphon.runtime import GraphRuntimeState from graphon.variables.segments import ArrayFileSegment, FileSegment, Segment from graphon.variables.variables import Variable from graphon.workflow_type_encoder import WorkflowRuntimeTypeConverter from libs.datetime_utils import naive_utc_now from models import Account, EndUser from models.human_input import HumanInputForm from models.workflow import WorkflowRun from services.variable_truncator import BaseTruncator, DummyVariableTruncator, VariableTruncator NodeExecutionId = NewType("NodeExecutionId", str) logger = logging.getLogger(__name__) class AccountCreatedByDict(TypedDict): id: str name: str email: str class EndUserCreatedByDict(TypedDict): id: str user: str CreatedByDict = AccountCreatedByDict | EndUserCreatedByDict @dataclass(slots=True) class _NodeSnapshot: """In-memory cache for node metadata between start and completion events.""" title: str index: int start_at: datetime iteration_id: str = "" """Empty string means the node is not executing inside an iteration.""" loop_id: str = "" """Empty string means the node is not executing inside a loop.""" class WorkflowResponseConverter: _truncator: BaseTruncator def __init__( self, *, application_generate_entity: Union[AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity], user: Union[Account, EndUser], system_variables: Sequence[Variable], ): self._application_generate_entity = application_generate_entity self._user = user self._system_variables = system_variables_to_mapping(system_variables) self._workflow_inputs = self._prepare_workflow_inputs() # Disable truncation for SERVICE_API calls to keep backward compatibility. if application_generate_entity.invoke_from == InvokeFrom.SERVICE_API: self._truncator = DummyVariableTruncator() else: self._truncator = VariableTruncator.default() self._node_snapshots: dict[NodeExecutionId, _NodeSnapshot] = {} self._workflow_execution_id: str | None = None self._workflow_started_at: datetime | None = None # ------------------------------------------------------------------ # Workflow lifecycle helpers # ------------------------------------------------------------------ def _prepare_workflow_inputs(self) -> Mapping[str, Any]: inputs = dict(self._application_generate_entity.inputs) for field_name, value in self._system_variables.items(): # TODO(@future-refactor): store system variables separately from user inputs so we don't # need to flatten `sys.*` entries into the input payload just for rerun/export tooling. if field_name == SystemVariableKey.CONVERSATION_ID: # Conversation IDs are session-scoped; omitting them keeps workflow inputs # reusable without pinning new runs to a prior conversation. continue inputs[f"sys.{field_name}"] = value handled = WorkflowEntry.handle_special_values(inputs) return dict(handled or {}) def _ensure_workflow_run_id(self, workflow_run_id: str | None = None) -> str: """Return the memoized workflow run id, optionally seeding it during start events.""" if workflow_run_id is not None: self._workflow_execution_id = workflow_run_id if not self._workflow_execution_id: raise ValueError("workflow_run_id missing before streaming workflow events") return self._workflow_execution_id # ------------------------------------------------------------------ # Node snapshot helpers # ------------------------------------------------------------------ def _store_snapshot(self, event: QueueNodeStartedEvent) -> _NodeSnapshot: snapshot = _NodeSnapshot( title=event.node_title, index=event.node_run_index, start_at=event.start_at, iteration_id=event.in_iteration_id or "", loop_id=event.in_loop_id or "", ) node_execution_id = NodeExecutionId(event.node_execution_id) self._node_snapshots[node_execution_id] = snapshot return snapshot def _get_snapshot(self, node_execution_id: str) -> _NodeSnapshot | None: return self._node_snapshots.get(NodeExecutionId(node_execution_id)) def _pop_snapshot(self, node_execution_id: str) -> _NodeSnapshot | None: return self._node_snapshots.pop(NodeExecutionId(node_execution_id), None) @staticmethod def _merge_metadata( base_metadata: Mapping[WorkflowNodeExecutionMetadataKey, Any] | None, snapshot: _NodeSnapshot | None, ) -> Mapping[WorkflowNodeExecutionMetadataKey, Any] | None: if not base_metadata and not snapshot: return base_metadata merged: dict[WorkflowNodeExecutionMetadataKey, Any] = {} if base_metadata: merged.update(base_metadata) if snapshot: if snapshot.iteration_id: merged[WorkflowNodeExecutionMetadataKey.ITERATION_ID] = snapshot.iteration_id if snapshot.loop_id: merged[WorkflowNodeExecutionMetadataKey.LOOP_ID] = snapshot.loop_id return merged or None def _truncate_mapping( self, mapping: Mapping[str, Any] | None, ) -> tuple[Mapping[str, Any] | None, bool]: if mapping is None: return None, False if not mapping: return {}, False normalized = WorkflowEntry.handle_special_values(dict(mapping)) if normalized is None: return None, False truncated, is_truncated = self._truncator.truncate_variable_mapping(dict(normalized)) return truncated, is_truncated @staticmethod def _encode_outputs(outputs: Mapping[str, Any] | None) -> Mapping[str, Any] | None: if outputs is None: return None converter = WorkflowRuntimeTypeConverter() return converter.to_json_encodable(outputs) def workflow_start_to_stream_response( self, *, task_id: str, workflow_run_id: str, workflow_id: str, reason: WorkflowStartReason, ) -> WorkflowStartStreamResponse: run_id = self._ensure_workflow_run_id(workflow_run_id) started_at = naive_utc_now() self._workflow_started_at = started_at return WorkflowStartStreamResponse( task_id=task_id, workflow_run_id=run_id, data=WorkflowStartStreamResponse.Data( id=run_id, workflow_id=workflow_id, inputs=self._workflow_inputs, created_at=int(started_at.timestamp()), reason=reason, ), ) def workflow_finish_to_stream_response( self, *, task_id: str, workflow_id: str, status: WorkflowExecutionStatus, graph_runtime_state: GraphRuntimeState, error: str | None = None, exceptions_count: int = 0, ) -> WorkflowFinishStreamResponse: run_id = self._ensure_workflow_run_id() started_at = self._workflow_started_at if started_at is None: raise ValueError( "workflow_finish_to_stream_response called before workflow_start_to_stream_response", ) finished_at = naive_utc_now() elapsed_time = (finished_at - started_at).total_seconds() outputs_mapping = graph_runtime_state.outputs or {} encoded_outputs = WorkflowRuntimeTypeConverter().to_json_encodable(outputs_mapping) created_by: CreatedByDict | dict[str, object] = {} user = self._user if isinstance(user, Account): created_by = AccountCreatedByDict( id=user.id, name=user.name, email=user.email, ) elif isinstance(user, EndUser): created_by = EndUserCreatedByDict( id=user.id, user=user.session_id, ) return WorkflowFinishStreamResponse( task_id=task_id, workflow_run_id=run_id, data=WorkflowFinishStreamResponse.Data( id=run_id, workflow_id=workflow_id, status=status, outputs=encoded_outputs, error=error, elapsed_time=elapsed_time, total_tokens=graph_runtime_state.total_tokens, total_steps=graph_runtime_state.node_run_steps, created_by=created_by, created_at=int(started_at.timestamp()), finished_at=int(finished_at.timestamp()), files=self.fetch_files_from_node_outputs(outputs_mapping), exceptions_count=exceptions_count, ), ) def workflow_pause_to_stream_response( self, *, event: QueueWorkflowPausedEvent, task_id: str, graph_runtime_state: GraphRuntimeState, ) -> list[StreamResponse]: run_id = self._ensure_workflow_run_id() started_at = self._workflow_started_at if started_at is None: raise ValueError( "workflow_pause_to_stream_response called before workflow_start_to_stream_response", ) paused_at = naive_utc_now() elapsed_time = (paused_at - started_at).total_seconds() encoded_outputs = self._encode_outputs(event.outputs) or {} if self._application_generate_entity.invoke_from == InvokeFrom.SERVICE_API: encoded_outputs = {} pause_reasons = [reason.model_dump(mode="json") for reason in event.reasons] human_input_form_ids = [reason.form_id for reason in event.reasons if isinstance(reason, HumanInputRequired)] expiration_times_by_form_id: dict[str, datetime] = {} display_in_ui_by_form_id: dict[str, bool] = {} form_token_by_form_id: dict[str, str] = {} if human_input_form_ids: stmt = select( HumanInputForm.id, HumanInputForm.expiration_time, HumanInputForm.form_definition, ).where(HumanInputForm.id.in_(human_input_form_ids)) with Session(bind=db.engine) as session: for form_id, expiration_time, form_definition in session.execute(stmt): expiration_times_by_form_id[str(form_id)] = expiration_time try: definition_payload = json.loads(form_definition) if form_definition else {} except (TypeError, json.JSONDecodeError): definition_payload = {} display_in_ui_by_form_id[str(form_id)] = bool(definition_payload.get("display_in_ui")) form_token_by_form_id = load_form_tokens_by_form_id( human_input_form_ids, session=session, surface=_INVOKE_FROM_TO_HITL_SURFACE.get(self._application_generate_entity.invoke_from), ) # Reconnect paths must preserve the same pause-reason contract as live streams; # otherwise clients see schema drift after resume. pause_reasons = enrich_human_input_pause_reasons( pause_reasons, form_tokens_by_form_id=form_token_by_form_id, expiration_times_by_form_id={ form_id: int(expiration_time.timestamp()) for form_id, expiration_time in expiration_times_by_form_id.items() }, ) responses: list[StreamResponse] = [] for reason in event.reasons: if isinstance(reason, HumanInputRequired): expiration_time = expiration_times_by_form_id.get(reason.form_id) if expiration_time is None: raise ValueError(f"HumanInputForm not found for pause reason, form_id={reason.form_id}") responses.append( HumanInputRequiredResponse( task_id=task_id, workflow_run_id=run_id, data=HumanInputRequiredResponse.Data( form_id=reason.form_id, node_id=reason.node_id, node_title=reason.node_title, form_content=reason.form_content, inputs=reason.inputs, actions=reason.actions, display_in_ui=display_in_ui_by_form_id.get(reason.form_id, False), form_token=form_token_by_form_id.get(reason.form_id), resolved_default_values=reason.resolved_default_values, expiration_time=int(expiration_time.timestamp()), ), ) ) responses.append( WorkflowPauseStreamResponse( task_id=task_id, workflow_run_id=run_id, data=WorkflowPauseStreamResponse.Data( workflow_run_id=run_id, paused_nodes=list(event.paused_nodes), outputs=encoded_outputs, reasons=pause_reasons, status=WorkflowExecutionStatus.PAUSED, created_at=int(started_at.timestamp()), elapsed_time=elapsed_time, total_tokens=graph_runtime_state.total_tokens, total_steps=graph_runtime_state.node_run_steps, ), ) ) return responses def human_input_form_filled_to_stream_response( self, *, event: QueueHumanInputFormFilledEvent, task_id: str ) -> HumanInputFormFilledResponse: run_id = self._ensure_workflow_run_id() return HumanInputFormFilledResponse( task_id=task_id, workflow_run_id=run_id, data=HumanInputFormFilledResponse.Data( node_id=event.node_id, node_title=event.node_title, rendered_content=event.rendered_content, action_id=event.action_id, action_text=event.action_text, ), ) def human_input_form_timeout_to_stream_response( self, *, event: QueueHumanInputFormTimeoutEvent, task_id: str ) -> HumanInputFormTimeoutResponse: run_id = self._ensure_workflow_run_id() return HumanInputFormTimeoutResponse( task_id=task_id, workflow_run_id=run_id, data=HumanInputFormTimeoutResponse.Data( node_id=event.node_id, node_title=event.node_title, expiration_time=int(event.expiration_time.timestamp()), ), ) @classmethod def workflow_run_result_to_finish_response( cls, *, task_id: str, workflow_run: WorkflowRun, creator_user: Account | EndUser, ) -> WorkflowFinishStreamResponse: run_id = workflow_run.id elapsed_time = workflow_run.elapsed_time encoded_outputs = workflow_run.outputs_dict finished_at = workflow_run.finished_at assert finished_at is not None created_by: Mapping[str, object] user = creator_user if isinstance(user, Account): created_by = { "id": user.id, "name": user.name, "email": user.email, } else: created_by = { "id": user.id, "user": user.session_id, } return WorkflowFinishStreamResponse( task_id=task_id, workflow_run_id=run_id, data=WorkflowFinishStreamResponse.Data( id=run_id, workflow_id=workflow_run.workflow_id, status=workflow_run.status, outputs=encoded_outputs, error=workflow_run.error, elapsed_time=elapsed_time, total_tokens=workflow_run.total_tokens, total_steps=workflow_run.total_steps, created_by=created_by, created_at=int(workflow_run.created_at.timestamp()), finished_at=int(finished_at.timestamp()), files=cls.fetch_files_from_node_outputs(encoded_outputs), exceptions_count=workflow_run.exceptions_count, ), ) def workflow_node_start_to_stream_response( self, *, event: QueueNodeStartedEvent, task_id: str, ) -> NodeStartStreamResponse | None: if event.node_type in {BuiltinNodeTypes.ITERATION, BuiltinNodeTypes.LOOP}: return None run_id = self._ensure_workflow_run_id() snapshot = self._store_snapshot(event) response = NodeStartStreamResponse( task_id=task_id, workflow_run_id=run_id, data=NodeStartStreamResponse.Data( id=event.node_execution_id, node_id=event.node_id, node_type=event.node_type, title=snapshot.title, index=snapshot.index, created_at=int(snapshot.start_at.timestamp()), iteration_id=event.in_iteration_id, loop_id=event.in_loop_id, agent_strategy=event.agent_strategy, ), ) try: if event.node_type == BuiltinNodeTypes.TOOL: response.data.extras["icon"] = ToolManager.get_tool_icon( tenant_id=self._application_generate_entity.app_config.tenant_id, provider_type=ToolProviderType(event.provider_type), provider_id=event.provider_id, ) elif event.node_type == BuiltinNodeTypes.DATASOURCE: manager = PluginDatasourceManager() provider_entity = manager.fetch_datasource_provider( self._application_generate_entity.app_config.tenant_id, event.provider_id, ) response.data.extras["icon"] = provider_entity.declaration.identity.generate_datasource_icon_url( self._application_generate_entity.app_config.tenant_id ) elif event.node_type == TRIGGER_PLUGIN_NODE_TYPE: response.data.extras["icon"] = TriggerManager.get_trigger_plugin_icon( self._application_generate_entity.app_config.tenant_id, event.provider_id, ) except Exception: # metadata fetch may fail, for example, the plugin daemon is down or plugin is uninstalled. logger.warning("failed to fetch icon for %s", event.provider_id) return response def workflow_node_finish_to_stream_response( self, *, event: QueueNodeSucceededEvent | QueueNodeFailedEvent | QueueNodeExceptionEvent, task_id: str, ) -> NodeFinishStreamResponse | None: if event.node_type in {BuiltinNodeTypes.ITERATION, BuiltinNodeTypes.LOOP}: return None run_id = self._ensure_workflow_run_id() snapshot = self._pop_snapshot(event.node_execution_id) start_at = snapshot.start_at if snapshot else event.start_at finished_at = event.finished_at or naive_utc_now() elapsed_time = (finished_at - start_at).total_seconds() inputs, inputs_truncated = self._truncate_mapping(event.inputs) process_data, process_data_truncated = self._truncate_mapping(event.process_data) encoded_outputs = self._encode_outputs(event.outputs) outputs, outputs_truncated = self._truncate_mapping(encoded_outputs) metadata = self._merge_metadata(event.execution_metadata, snapshot) match event: case QueueNodeSucceededEvent(): status = WorkflowNodeExecutionStatus.SUCCEEDED error_message = event.error case QueueNodeFailedEvent(): status = WorkflowNodeExecutionStatus.FAILED error_message = event.error case _: status = WorkflowNodeExecutionStatus.EXCEPTION error_message = event.error return NodeFinishStreamResponse( task_id=task_id, workflow_run_id=run_id, data=NodeFinishStreamResponse.Data( id=event.node_execution_id, node_id=event.node_id, node_type=event.node_type, index=snapshot.index if snapshot else 0, title=snapshot.title if snapshot else "", inputs=inputs, inputs_truncated=inputs_truncated, process_data=process_data, process_data_truncated=process_data_truncated, outputs=outputs, outputs_truncated=outputs_truncated, status=status, error=error_message, elapsed_time=elapsed_time, execution_metadata=metadata, created_at=int(start_at.timestamp()), finished_at=int(finished_at.timestamp()), files=self.fetch_files_from_node_outputs(event.outputs or {}), iteration_id=event.in_iteration_id, loop_id=event.in_loop_id, ), ) def workflow_node_retry_to_stream_response( self, *, event: QueueNodeRetryEvent, task_id: str, ) -> NodeRetryStreamResponse | None: if event.node_type in {BuiltinNodeTypes.ITERATION, BuiltinNodeTypes.LOOP}: return None run_id = self._ensure_workflow_run_id() snapshot = self._get_snapshot(event.node_execution_id) if snapshot is None: raise AssertionError("node retry event arrived without a stored snapshot") finished_at = naive_utc_now() elapsed_time = (finished_at - event.start_at).total_seconds() inputs, inputs_truncated = self._truncate_mapping(event.inputs) process_data, process_data_truncated = self._truncate_mapping(event.process_data) encoded_outputs = self._encode_outputs(event.outputs) outputs, outputs_truncated = self._truncate_mapping(encoded_outputs) metadata = self._merge_metadata(event.execution_metadata, snapshot) return NodeRetryStreamResponse( task_id=task_id, workflow_run_id=run_id, data=NodeRetryStreamResponse.Data( id=event.node_execution_id, node_id=event.node_id, node_type=event.node_type, index=snapshot.index, title=snapshot.title, inputs=inputs, inputs_truncated=inputs_truncated, process_data=process_data, process_data_truncated=process_data_truncated, outputs=outputs, outputs_truncated=outputs_truncated, status=WorkflowNodeExecutionStatus.RETRY, error=event.error, elapsed_time=elapsed_time, execution_metadata=metadata, created_at=int(snapshot.start_at.timestamp()), finished_at=int(finished_at.timestamp()), files=self.fetch_files_from_node_outputs(event.outputs or {}), iteration_id=event.in_iteration_id, loop_id=event.in_loop_id, retry_index=event.retry_index, ), ) def workflow_iteration_start_to_stream_response( self, *, task_id: str, workflow_execution_id: str, event: QueueIterationStartEvent, ) -> IterationNodeStartStreamResponse: new_inputs, truncated = self._truncator.truncate_variable_mapping(event.inputs or {}) return IterationNodeStartStreamResponse( task_id=task_id, workflow_run_id=workflow_execution_id, data=IterationNodeStartStreamResponse.Data( id=event.node_id, node_id=event.node_id, node_type=event.node_type, title=event.node_title, created_at=int(time.time()), extras={}, inputs=new_inputs, inputs_truncated=truncated, metadata=event.metadata or {}, ), ) def workflow_iteration_next_to_stream_response( self, *, task_id: str, workflow_execution_id: str, event: QueueIterationNextEvent, ) -> IterationNodeNextStreamResponse: return IterationNodeNextStreamResponse( task_id=task_id, workflow_run_id=workflow_execution_id, data=IterationNodeNextStreamResponse.Data( id=event.node_id, node_id=event.node_id, node_type=event.node_type, title=event.node_title, index=event.index, created_at=int(time.time()), extras={}, ), ) def workflow_iteration_completed_to_stream_response( self, *, task_id: str, workflow_execution_id: str, event: QueueIterationCompletedEvent, ) -> IterationNodeCompletedStreamResponse: json_converter = WorkflowRuntimeTypeConverter() new_inputs, inputs_truncated = self._truncator.truncate_variable_mapping(event.inputs or {}) new_outputs, outputs_truncated = self._truncator.truncate_variable_mapping( json_converter.to_json_encodable(event.outputs) or {} ) return IterationNodeCompletedStreamResponse( task_id=task_id, workflow_run_id=workflow_execution_id, data=IterationNodeCompletedStreamResponse.Data( id=event.node_id, node_id=event.node_id, node_type=event.node_type, title=event.node_title, outputs=new_outputs, outputs_truncated=outputs_truncated, created_at=int(time.time()), extras={}, inputs=new_inputs, inputs_truncated=inputs_truncated, status=WorkflowNodeExecutionStatus.SUCCEEDED if event.error is None else WorkflowNodeExecutionStatus.FAILED, error=None, elapsed_time=(naive_utc_now() - event.start_at).total_seconds(), total_tokens=(lambda x: x if isinstance(x, int) else 0)(event.metadata.get("total_tokens", 0)), execution_metadata=event.metadata, finished_at=int(time.time()), steps=event.steps, ), ) def workflow_loop_start_to_stream_response( self, *, task_id: str, workflow_execution_id: str, event: QueueLoopStartEvent ) -> LoopNodeStartStreamResponse: new_inputs, truncated = self._truncator.truncate_variable_mapping(event.inputs or {}) return LoopNodeStartStreamResponse( task_id=task_id, workflow_run_id=workflow_execution_id, data=LoopNodeStartStreamResponse.Data( id=event.node_id, node_id=event.node_id, node_type=event.node_type, title=event.node_title, created_at=int(time.time()), extras={}, inputs=new_inputs, inputs_truncated=truncated, metadata=event.metadata or {}, ), ) def workflow_loop_next_to_stream_response( self, *, task_id: str, workflow_execution_id: str, event: QueueLoopNextEvent, ) -> LoopNodeNextStreamResponse: return LoopNodeNextStreamResponse( task_id=task_id, workflow_run_id=workflow_execution_id, data=LoopNodeNextStreamResponse.Data( id=event.node_id, node_id=event.node_id, node_type=event.node_type, title=event.node_title, index=event.index, # The `pre_loop_output` field is not utilized by the frontend. # Previously, it was assigned the value of `event.output`. pre_loop_output={}, created_at=int(time.time()), extras={}, ), ) def workflow_loop_completed_to_stream_response( self, *, task_id: str, workflow_execution_id: str, event: QueueLoopCompletedEvent, ) -> LoopNodeCompletedStreamResponse: json_converter = WorkflowRuntimeTypeConverter() new_inputs, inputs_truncated = self._truncator.truncate_variable_mapping(event.inputs or {}) new_outputs, outputs_truncated = self._truncator.truncate_variable_mapping( json_converter.to_json_encodable(event.outputs) or {} ) return LoopNodeCompletedStreamResponse( task_id=task_id, workflow_run_id=workflow_execution_id, data=LoopNodeCompletedStreamResponse.Data( id=event.node_id, node_id=event.node_id, node_type=event.node_type, title=event.node_title, outputs=new_outputs, outputs_truncated=outputs_truncated, created_at=int(time.time()), extras={}, inputs=new_inputs, inputs_truncated=inputs_truncated, status=WorkflowNodeExecutionStatus.SUCCEEDED if event.error is None else WorkflowNodeExecutionStatus.FAILED, error=None, elapsed_time=(naive_utc_now() - event.start_at).total_seconds(), total_tokens=(lambda x: x if isinstance(x, int) else 0)(event.metadata.get("total_tokens", 0)), execution_metadata=event.metadata, finished_at=int(time.time()), steps=event.steps, ), ) @classmethod def fetch_files_from_node_outputs(cls, outputs_dict: Mapping[str, Any] | None) -> Sequence[Mapping[str, Any]]: """ Fetch files from node outputs :param outputs_dict: node outputs dict :return: """ if not outputs_dict: return [] files = [cls._fetch_files_from_variable_value(output_value) for output_value in outputs_dict.values()] # Remove None files = [file for file in files if file] # Flatten list # Flatten the list of sequences into a single list of mappings flattened_files = [file for sublist in files if sublist for file in sublist] # Convert to tuple to match Sequence type return tuple(flattened_files) @classmethod def _fetch_files_from_variable_value(cls, value: Union[dict, list, Segment]) -> Sequence[Mapping[str, Any]]: """ Fetch files from variable value :param value: variable value :return: """ if not value: return [] files: list[Mapping[str, Any]] = [] match value: case FileSegment(): files.append(value.value.to_dict()) case ArrayFileSegment(): files.extend([i.to_dict() for i in value.value]) case File(): files.append(value.to_dict()) case list(): for item in value: file = cls._get_file_var_from_value(item) if file: files.append(file) case dict(): file = cls._get_file_var_from_value(value) if file: files.append(file) case _: pass return files @classmethod def _get_file_var_from_value(cls, value: Union[dict, list]) -> Mapping[str, Any] | None: """ Get file var from value :param value: variable value :return: """ if not value: return None if isinstance(value, dict) and value.get("dify_model_identity") == FILE_MODEL_IDENTITY: return value elif isinstance(value, File): return value.to_dict() return None def handle_agent_log(self, task_id: str, event: QueueAgentLogEvent) -> AgentLogStreamResponse: """ Handle agent log :param task_id: task id :param event: agent log event :return: """ return AgentLogStreamResponse( task_id=task_id, data=AgentLogStreamResponse.Data( node_execution_id=event.node_execution_id, id=event.id, parent_id=event.parent_id, label=event.label, error=event.error, status=event.status, data=event.data, metadata=event.metadata, node_id=event.node_id, ), )