dify/api/services/workflow_runtime/response_converter.py

import json
import logging
import time
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime
from typing import TYPE_CHECKING, Any, NewType, TypedDict

if TYPE_CHECKING:
    from services.dify_builder.native_testing.host_execution import NativeHostLaunch

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 (
    NodeExecutionSnapshot,
    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_dispositions_by_form_id,
)
from core.workflow.human_input_policy import (
    HumanInputSurface,
    enrich_human_input_pause_reasons,
    resolve_variable_select_input_options,
)
from core.workflow.nodes.human_input.pause_reason import HumanInputRequired
from core.workflow.nodes.human_input_v2.presentation import build_human_input_v2_pause_reason
from core.workflow.nodes.human_input_v2.runtime import PreparedForm
from core.workflow.system_variables import SystemVariableKey, system_variables_to_mapping
from extensions.ext_database import db
from graphon.entities import WorkflowStartReason
from graphon.enums import (
    BuiltinNodeTypes,
    WorkflowExecutionStatus,
    WorkflowNodeExecutionMetadataKey,
    WorkflowNodeExecutionStatus,
)
from graphon.file import FILE_MODEL_IDENTITY, File
from graphon.runtime import RuntimeState
from graphon.runtime.runtime_state_protocol import ReadOnlyVariablePool
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, to_utc_timestamp
from models import Account, EndUser
from models.human_input import HumanInputForm
from models.workflow import WorkflowRun
from services.dify_builder.native_testing.observation_adapters import NativeObservationGuard
from services.dify_builder.native_testing.observation_stream import (
    observed_runtime_outputs,
    observed_values,
    project_native_queue_event,
    require_native_observation,
)
from services.variable_truncator import BaseTruncator, DummyVariableTruncator, VariableTruncator
from services.workflow_runtime.entry import WorkflowEntry

# 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,
}

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(frozen=True, slots=True)
class _HumanInputPauseOutput:
    """Keep a form event paired with its entry in the workflow pause response."""

    response: HumanInputRequiredResponse
    pause_reason: dict[str, Any]


class WorkflowResponseConverter:
    @staticmethod
    def human_input_v2_required_data(prepared: PreparedForm, *, node_id: str) -> HumanInputRequiredResponse.Data:
        """Project Dify-owned form data into the existing HITL wire contract.

        Live execution supplies the node's snapshot; reconnect loads the form
        from its run-owned v2 records. Neither path resolves a v1 token.
        """
        reason = build_human_input_v2_pause_reason(
            prepared.form.resolved_form, form_id=prepared.form.id, node_id=node_id
        )
        return HumanInputRequiredResponse.Data(
            form_version=reason.form_version,
            form_id=reason.form_id,
            node_id=node_id,
            node_title=reason.node_title,
            form_content=reason.form_content,
            inputs=reason.inputs,
            actions=reason.actions,
            resolved_default_values=reason.resolved_default_values,
            expiration_time=to_utc_timestamp(prepared.form.expiration_time),
            display_in_ui=prepared.form_token is not None,
            form_token=prepared.form_token.get_secret_value() if prepared.form_token is not None else None,
        )

    _truncator: BaseTruncator

    def __init__(
        self,
        *,
        application_generate_entity: AdvancedChatAppGenerateEntity | WorkflowAppGenerateEntity,
        user: Account | EndUser,
        system_variables: Sequence[Variable],
        native_observation: NativeObservationGuard | None = None,
        native_host: "NativeHostLaunch | None" = None,
        native_queue: Any = None,
    ):
        require_native_observation(guard=native_observation, entity=application_generate_entity)
        if native_host is not None:
            from services.dify_builder.native_testing.runtime import validate_projection_host

            validate_projection_host(
                host=native_host, guard=native_observation, entity=application_generate_entity, queue=native_queue
            )
        self._native_host = native_host
        self._native_queue = native_queue
        self._native_observation = native_observation
        if native_observation is not None:
            with native_observation.persistence_scope():
                if not isinstance(user, Account) or application_generate_entity.files:
                    native_observation.fail()
        self._application_generate_entity = application_generate_entity
        self._user = user
        if native_observation is None:
            self._system_variables = system_variables_to_mapping(system_variables)
            self._workflow_inputs = self._prepare_workflow_inputs()
        else:
            with native_observation.persistence_scope():
                self._workflow_inputs = native_observation.workflow_inputs(
                    inputs=application_generate_entity.inputs,
                    system_variables=system_variables_to_mapping(system_variables),
                )
            self._system_variables = {}

        # Disable truncation for SERVICE_API calls to keep backward compatibility.
        if native_observation is not None or application_generate_entity.invoke_from == InvokeFrom.SERVICE_API:
            self._truncator = DummyVariableTruncator()
        else:
            self._truncator = VariableTruncator.default()

        self._node_snapshots: dict[NodeExecutionId, NodeExecutionSnapshot] = {}
        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) -> NodeExecutionSnapshot:
        # The engine may have persisted this start before resumption metadata was loaded.
        snapshot = self._get_snapshot(event.node_execution_id)
        if snapshot is not None:
            return snapshot

        snapshot = NodeExecutionSnapshot(
            execution_id=event.node_execution_id,
            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) -> NodeExecutionSnapshot | None:
        return self._node_snapshots.get(NodeExecutionId(node_execution_id))

    def _pop_snapshot(self, node_execution_id: str) -> NodeExecutionSnapshot | None:
        return self._node_snapshots.pop(NodeExecutionId(node_execution_id), None)

    @staticmethod
    def _merge_metadata(
        base_metadata: Mapping[WorkflowNodeExecutionMetadataKey, Any] | None,
        snapshot: NodeExecutionSnapshot | 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 self._native_observation is not None:
            return observed_values(guard=self._native_observation, values=mapping), False
        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

    def _encode_outputs(self, outputs: Mapping[str, Any] | None) -> Mapping[str, Any] | None:
        if self._native_observation is not None:
            return observed_values(guard=self._native_observation, values=outputs)
        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,
        node_execution_snapshots: Sequence[NodeExecutionSnapshot] = (),
    ) -> WorkflowStartStreamResponse:
        if self._native_observation is not None:
            self._native_observation.require_healthy()
            if reason is WorkflowStartReason.RESUMPTION:
                from services.dify_builder.native_testing.runtime import require_resume_projection

                require_resume_projection(
                    host=self._native_host,
                    guard=self._native_observation,
                    queue=self._native_queue,
                    entity=self._application_generate_entity,
                    workflow_run_id=workflow_run_id,
                )
            elif reason is not WorkflowStartReason.INITIAL:
                self._native_observation.fail()
        run_id = self._ensure_workflow_run_id(workflow_run_id)
        started_at = naive_utc_now()
        self._workflow_started_at = started_at
        self._application_generate_entity.workflow_execution_timing.start()
        if reason == WorkflowStartReason.RESUMPTION:
            self._node_snapshots.update(
                (NodeExecutionId(snapshot.execution_id), snapshot) for snapshot in node_execution_snapshots
            )

        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: RuntimeState,
        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 = self._application_generate_entity.workflow_execution_timing.finish()

        outputs_mapping = (
            observed_runtime_outputs(guard=self._native_observation, state=graph_runtime_state)
            if self._native_observation is not None
            else graph_runtime_state.outputs or {}
        )
        encoded_outputs = self._encode_outputs(outputs_mapping)
        if self._native_observation is not None:
            error = self._native_observation.error(error)

        created_by: CreatedByDict | dict[str, object] = {}
        user = self._user
        match user:
            case Account() if self._native_observation is not None:
                created_by = {"id": user.id}
            case Account():
                created_by = AccountCreatedByDict(
                    id=user.id,
                    name=user.name,
                    email=user.email,
                )
            case 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=[]
                if self._native_observation is not None
                else 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: RuntimeState,
    ) -> 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",
            )
        elapsed_time = self._application_generate_entity.workflow_execution_timing.finish()
        if self._native_observation is not None:
            event = project_native_queue_event(
                guard=self._native_observation, event=event, native_host=self._native_host, queue=self._native_queue
            )
            outputs = observed_runtime_outputs(guard=self._native_observation, state=graph_runtime_state)
            event_outputs = observed_values(guard=self._native_observation, values=event.outputs)
            if event_outputs is not None and event_outputs["shape"]["fields"] > 0:
                outputs = event_outputs
            return [
                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=outputs or {},
                        reasons=[reason.model_dump(mode="json") for reason in event.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,
                    ),
                )
            ]
        encoded_outputs = self._encode_outputs(event.outputs) or {}
        if self._application_generate_entity.invoke_from == InvokeFrom.SERVICE_API:
            encoded_outputs = {}
        v1_outputs = iter(
            self._human_input_v1_pause_responses(
                reasons=[
                    reason
                    for reason in event.reasons
                    if isinstance(reason, HumanInputRequired) and reason.form_version == "1"
                ],
                task_id=task_id,
                variable_pool=graph_runtime_state.variable_pool,
            )
        )
        v2_outputs = iter(
            self._human_input_v2_pause_responses(
                reasons=[
                    reason
                    for reason in event.reasons
                    if isinstance(reason, HumanInputRequired) and reason.form_version == "2"
                ],
                forms=event.human_input_v2_forms,
                task_id=task_id,
            )
        )
        responses: list[StreamResponse] = []
        pause_reasons: list[dict[str, Any]] = []
        # Each version owns its complete conversion; merge in the original event
        # order, including scheduling reasons interleaved with approval forms.
        for reason in event.reasons:
            if not isinstance(reason, HumanInputRequired):
                pause_reasons.append(reason.model_dump(mode="json"))
                continue
            match reason.form_version:
                case "1":
                    converted = next(v1_outputs)
                case "2":
                    converted = next(v2_outputs)
                case _:
                    raise ValueError(f"Unsupported Human Input form version: {reason.form_version}")
            responses.append(converted.response)
            pause_reasons.append(converted.pause_reason)

        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:
        if self._native_observation is not None:
            event = project_native_queue_event(
                guard=self._native_observation, event=event, native_host=self._native_host, queue=self._native_queue
            )
        run_id = self._ensure_workflow_run_id()
        data = HumanInputFormFilledResponse.Data(
            form_id=event.form_id,
            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,
        )
        if event.submitted_data is not None:
            runtime_type_converter = WorkflowRuntimeTypeConverter()

            data.submitted_data = runtime_type_converter.value_to_json_encodable_recursive(event.submitted_data)

        return HumanInputFormFilledResponse(task_id=task_id, workflow_run_id=run_id, data=data)

    def human_input_form_timeout_to_stream_response(
        self, *, event: QueueHumanInputFormTimeoutEvent, task_id: str
    ) -> HumanInputFormTimeoutResponse:
        if self._native_observation is not None:
            event = project_native_queue_event(
                guard=self._native_observation, event=event, native_host=self._native_host, queue=self._native_queue
            )
        run_id = self._ensure_workflow_run_id()
        return HumanInputFormTimeoutResponse(
            task_id=task_id,
            workflow_run_id=run_id,
            data=HumanInputFormTimeoutResponse.Data(
                form_id=event.form_id,
                node_id=event.node_id,
                node_title=event.node_title,
                expiration_time=to_utc_timestamp(event.expiration_time),
            ),
        )

    @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
        match user:
            case Account():
                created_by = {
                    "id": user.id,
                    "name": user.name,
                    "email": user.email,
                }
            case _:
                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 self._native_observation is not None:
            event = project_native_queue_event(
                guard=self._native_observation, event=event, native_host=self._native_host, queue=self._native_queue
            )
        snapshot = self._store_snapshot(event)
        if event.node_type in {BuiltinNodeTypes.ITERATION, BuiltinNodeTypes.LOOP}:
            return None
        run_id = self._ensure_workflow_run_id()

        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:
                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),
                    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 self._native_observation is not None:
            event = project_native_queue_event(
                guard=self._native_observation, event=event, native_host=self._native_host, queue=self._native_queue
            )
        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=[]
                if self._native_observation is not None
                else 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 self._native_observation is not None:
            event = project_native_queue_event(
                guard=self._native_observation, event=event, native_host=self._native_host, queue=self._native_queue
            )
        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=[]
                if self._native_observation is not None
                else 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:
        if self._native_observation is not None:
            event = project_native_queue_event(
                guard=self._native_observation, event=event, native_host=self._native_host, queue=self._native_queue
            )
        new_inputs, truncated = (
            self._truncate_mapping(event.inputs or {})
            if self._native_observation is not None
            else 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 or {},
                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:
        if self._native_observation is not None:
            event = project_native_queue_event(
                guard=self._native_observation, event=event, native_host=self._native_host, queue=self._native_queue
            )
        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:
        if self._native_observation is not None:
            event = project_native_queue_event(
                guard=self._native_observation, event=event, native_host=self._native_host, queue=self._native_queue
            )

        new_inputs, inputs_truncated = (
            self._truncate_mapping(event.inputs or {})
            if self._native_observation is not None
            else self._truncator.truncate_variable_mapping(event.inputs or {})
        )
        new_outputs, outputs_truncated = (
            self._truncate_mapping(self._encode_outputs(event.outputs) or {})
            if self._native_observation is not None
            else self._truncator.truncate_variable_mapping(
                WorkflowRuntimeTypeConverter().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 or {},
                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:
        if self._native_observation is not None:
            event = project_native_queue_event(
                guard=self._native_observation, event=event, native_host=self._native_host, queue=self._native_queue
            )
        new_inputs, truncated = (
            self._truncate_mapping(event.inputs or {})
            if self._native_observation is not None
            else 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 or {},
                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:
        if self._native_observation is not None:
            event = project_native_queue_event(
                guard=self._native_observation, event=event, native_host=self._native_host, queue=self._native_queue
            )
        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:
        if self._native_observation is not None:
            event = project_native_queue_event(
                guard=self._native_observation, event=event, native_host=self._native_host, queue=self._native_queue
            )
        new_inputs, inputs_truncated = (
            self._truncate_mapping(event.inputs or {})
            if self._native_observation is not None
            else self._truncator.truncate_variable_mapping(event.inputs or {})
        )
        new_outputs, outputs_truncated = (
            self._truncate_mapping(self._encode_outputs(event.outputs) or {})
            if self._native_observation is not None
            else self._truncator.truncate_variable_mapping(
                WorkflowRuntimeTypeConverter().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 or {},
                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: dict | list | Segment | File | None
    ) -> 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: object) -> Mapping[str, Any] | None:
        """
        Get file var from value
        :param value: variable value
        :return:
        """
        if not value:
            return None

        match value:
            case dict() if value.get("dify_model_identity") == FILE_MODEL_IDENTITY:
                return value
            case File():
                return value.to_dict()

        return None

    def handle_agent_log(self, task_id: str, event: QueueAgentLogEvent) -> AgentLogStreamResponse:
        if self._native_observation is not None:
            event = project_native_queue_event(
                guard=self._native_observation, event=event, native_host=self._native_host, queue=self._native_queue
            )
        """
        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,
            ),
        )

    def _human_input_v1_pause_responses(
        self,
        *,
        reasons: Sequence[HumanInputRequired],
        task_id: str,
        variable_pool: ReadOnlyVariablePool,
    ) -> list[_HumanInputPauseOutput]:
        """Resolve legacy inputs, access channels, and both V1 response payloads."""
        if not reasons:
            return []
        run_id = self._ensure_workflow_run_id()
        resolved_reasons = [
            reason.model_copy(
                update={
                    "inputs": resolve_variable_select_input_options(reason.inputs, variable_pool=variable_pool),
                }
            )
            for reason in reasons
        ]
        form_ids = [reason.form_id for reason in resolved_reasons]
        expiration_times: dict[str, datetime] = {}
        display_in_ui: dict[str, bool] = {}
        stmt = select(HumanInputForm.id, HumanInputForm.expiration_time, HumanInputForm.form_definition).where(
            HumanInputForm.id.in_(form_ids)
        )
        surface = _INVOKE_FROM_TO_HITL_SURFACE.get(self._application_generate_entity.invoke_from)
        with Session(bind=db.engine) as session:
            for form_id, expiration_time, form_definition in session.execute(stmt):
                expiration_times[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[str(form_id)] = bool(definition_payload.get("display_in_ui"))
            dispositions = load_form_dispositions_by_form_id(form_ids, session=session, surface=surface)
        # Preserve the legacy pause-reason fields, which differ from the
        # human_input_required data payload (notably display_in_ui).
        pause_reasons = enrich_human_input_pause_reasons(
            [reason.model_dump(mode="json") for reason in resolved_reasons],
            dispositions_by_form_id=dispositions,
            expiration_times_by_form_id={
                form_id: to_utc_timestamp(value) for form_id, value in expiration_times.items()
            },
        )
        outputs: list[_HumanInputPauseOutput] = []
        for reason, payload in zip(resolved_reasons, pause_reasons, strict=True):
            expiration_time = expiration_times.get(reason.form_id)
            if expiration_time is None:
                raise ValueError(f"HumanInputForm not found for pause reason, form_id={reason.form_id}")
            disposition = dispositions.get(reason.form_id)
            response = HumanInputRequiredResponse(
                task_id=task_id,
                workflow_run_id=run_id,
                data=HumanInputRequiredResponse.Data(
                    form_version=reason.form_version,
                    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.get(reason.form_id, False),
                    form_token=disposition.form_token if disposition else None,
                    approval_channels=list(disposition.approval_channels) if disposition else [],
                    resolved_default_values=reason.resolved_default_values,
                    expiration_time=to_utc_timestamp(expiration_time),
                ),
            )
            outputs.append(_HumanInputPauseOutput(response=response, pause_reason=payload))
        return outputs

    def _human_input_v2_pause_responses(
        self,
        *,
        reasons: Sequence[HumanInputRequired],
        forms: Mapping[str, PreparedForm],
        task_id: str,
    ) -> list[_HumanInputPauseOutput]:
        """Build both V2 response payloads from the snapshots emitted by nodes."""
        if not reasons:
            return []
        run_id = self._ensure_workflow_run_id()
        outputs: list[_HumanInputPauseOutput] = []
        for reason in reasons:
            prepared = forms.get(reason.form_id)
            if prepared is None:
                raise ValueError(f"Human Input v2 node did not supply its form snapshot: {reason.form_id}")
            data = self.human_input_v2_required_data(prepared, node_id=reason.node_id)
            response = HumanInputRequiredResponse(task_id=task_id, workflow_run_id=run_id, data=data)
            payload = reason.model_dump(mode="json")
            payload.update(data.model_dump(mode="json"))
            outputs.append(_HumanInputPauseOutput(response=response, pause_reason=payload))
        return outputs