dify/api/services/workflow_runtime/agent_node.py

from __future__ import annotations

import logging
from collections.abc import Generator, Mapping, Sequence
from contextlib import contextmanager, suppress
from time import perf_counter
from typing import TYPE_CHECKING, Any, Literal, override

from dify_agent.protocol import CancelRunRequest
from dify_agent.protocol.snapshot import SessionSnapshot

from clients.agent_backend import (
    AgentBackendAgentMessageDeltaInternalEvent,
    AgentBackendError,
    AgentBackendHTTPError,
    AgentBackendInternalEventType,
    AgentBackendRunCancelledInternalEvent,
    AgentBackendRunClient,
    AgentBackendRunEventAdapter,
    AgentBackendRunFailedInternalEvent,
    AgentBackendRunSucceededInternalEvent,
    AgentBackendRunWaitingInternalEvent,
    AgentBackendStreamError,
    AgentBackendStreamInternalEvent,
    AgentBackendTransportError,
    AgentBackendValidationError,
)
from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
from core.workflow.node_execution_process_data import WORKFLOW_TOOL_INVOCATION_ID_KEY
from core.workflow.nodes.agent_v2.entities import DifyAgentNodeData
from core.workflow.nodes.agent_v2.output_adapter import WorkflowAgentOutputAdapter
from core.workflow.nodes.agent_v2.output_failure_orchestrator import (
    FailedOutput,
    OutputFailureDecision,
    OutputFailureKind,
    OutputFailureOrchestrator,
)
from core.workflow.nodes.agent_v2.output_type_checker import OutputTypeCheckOutcome, PerOutputTypeChecker
from core.workflow.nodes.agent_v2.runtime_request_builder import (
    WorkflowAgentRuntimeBuildContext,
    WorkflowAgentRuntimeRequestBuilder,
    WorkflowAgentRuntimeRequestBuildError,
)
from core.workflow.nodes.agent_v2.session_store import WorkflowAgentSessionScope, WorkflowAgentWorkspaceStore
from core.workflow.system_variables import SystemVariableKey, get_system_text
from graphon.engine_events import NodeRunPauseRequestedEvent
from graphon.entities.pause_reason import SchedulingPause
from graphon.enums import (
    BuiltinNodeTypes,
    ErrorStrategy,
    NodeExecutionType,
    WorkflowNodeExecutionMetadataKey,
    WorkflowNodeExecutionStatus,
)
from graphon.node_events import (
    NodeEventPayload,
    NodeRunResult,
    RunRetrieverResourceEvent,
    StreamChunkEvent,
    StreamCompletedEvent,
)
from graphon.nodes.base.node import Node
from graphon.nodes.base.variable_template_parser import VariableTemplateParser
from models.agent_config_entities import WorkflowNodeJobConfig, WorkflowOutputRoutes
from services.agent.knowledge_citations import knowledge_sources_from_stream_event
from services.agent.prompt_mentions import extract_workflow_node_output_selectors
from services.agent.workspace_service import AgentWorkspaceNotFoundError
from services.workflow_runtime.agent_binding import WorkflowAgentBindingError, WorkflowAgentBindingResolver

if TYPE_CHECKING:
    from graphon.runtime import InitParams, RuntimeState
    from runtime_contracts.builder.native.agent_context import BuilderAgentExecutionContext
    from services.agent.workflow_run_recovery import WorkflowAgentRunRecovery
    from services.dify_builder.native_testing.agent_preparation import AgentPreparationService
    from services.dify_builder.native_testing.agent_recovery import ConfirmedSnapshotIdentity
    from services.dify_builder.native_testing.confirmed_agent_commands import ConfirmedAgentCommands

logger = logging.getLogger(__name__)


@contextmanager
def _confirmed_node_phase(
    context: BuilderAgentExecutionContext,
    phase: Literal["parent_restore", "recover_existing", "load_wait", "continue_wait"],
) -> Generator[None]:
    """Disjoint node recovery phases; their native reads and transport spans are nested."""
    started, error_type = perf_counter(), "none"
    try:
        yield
    except BaseException as error:
        error_type = type(error).__name__
        raise
    finally:
        with suppress(Exception):  # Observability cannot replace a native control exception.
            logger.info(
                "Confirmed node timing request_id=%s run_id=%s phase=%s duration_seconds=%.6f error_type=%s",
                context.ref.request_id,
                context.native_run_id,
                phase,
                perf_counter() - started,
                error_type,
            )


# Terminal and waiting controls that `_consume_event_stream` may return.
# Stream + started events are filtered out (plain-run text deltas are yielded
# as stream chunks instead); transport errors
# are surfaced as a separate StreamCompletedEvent in the second tuple slot.
type _TerminalAgentBackendEvent = (
    AgentBackendRunSucceededInternalEvent
    | AgentBackendRunFailedInternalEvent
    | AgentBackendRunCancelledInternalEvent
    | AgentBackendRunWaitingInternalEvent
)


class DifyAgentNode(Node[DifyAgentNodeData]):
    """Execute a frozen node job and select one workflow exit on success.

    Enabled routing selects a stable route ID through the required ``switch``
    output. Disabled routing remains executable so
    graph-owned default values can continue without a selected route; Graphon
    still promotes nodes that opt into its failure branch.
    """

    node_type = BuiltinNodeTypes.AGENT

    def __init__(
        self,
        node_id: str,
        data: DifyAgentNodeData,
        *,
        init_params: InitParams,
        runtime_state: RuntimeState,
        binding_resolver: WorkflowAgentBindingResolver,
        runtime_request_builder: WorkflowAgentRuntimeRequestBuilder,
        agent_backend_client: AgentBackendRunClient,
        event_adapter: AgentBackendRunEventAdapter,
        output_adapter: WorkflowAgentOutputAdapter,
        type_checker: PerOutputTypeChecker,
        failure_orchestrator: OutputFailureOrchestrator,
        session_store: WorkflowAgentWorkspaceStore,
        agent_preparation: AgentPreparationService | None = None,
        agent_commands: ConfirmedAgentCommands | None = None,
        agent_run_recovery: WorkflowAgentRunRecovery | None = None,
    ) -> None:
        super().__init__(
            node_id=node_id,
            data=data,
            init_params=init_params,
            runtime_state=runtime_state,
        )
        if self.node_data.agent_output_routes.enabled and self.error_strategy == ErrorStrategy.DEFAULT_VALUE:
            raise ValueError("Output routes do not support the node-level default-value error strategy.")
        if self.node_data.agent_output_routes.enabled:
            self.execution_type = NodeExecutionType.BRANCH
        self._binding_resolver = binding_resolver
        self._runtime_request_builder = runtime_request_builder
        self._agent_backend_client = agent_backend_client
        self._event_adapter = event_adapter
        self._output_adapter = output_adapter
        self._type_checker = type_checker
        self._failure_orchestrator = failure_orchestrator
        self._session_store = session_store
        self._agent_preparation = agent_preparation
        self._agent_commands = agent_commands
        self._agent_run_recovery = agent_run_recovery

    @override
    def _get_error_strategy(self) -> ErrorStrategy | None:
        context = DifyRunContext.model_validate(self.require_run_context_value(DIFY_RUN_CONTEXT_KEY))
        # Confirmed control failures cannot become default output or an alternate branch.
        if context.builder_agent_execution is not None:
            return None
        return super()._get_error_strategy()

    @classmethod
    @override
    def version(cls) -> str:
        return "2"

    @override
    def populate_start_event(self, event) -> None:
        event.extras["agent_node"] = {"version": "2", "agent_node_kind": self.node_data.agent_node_kind}

    @override
    def _run(self) -> Generator[NodeEventPayload | NodeRunPauseRequestedEvent]:
        inputs: dict[str, Any] = {}
        process_data: dict[str, Any] = {}
        metadata: dict[str, Any] = {
            "agent_backend": {
                "status": "not_started",
            }
        }
        try:
            yield from self._run_inner(inputs=inputs, process_data=process_data, metadata=metadata)
        except Exception as error:
            if not process_data:
                raise
            yield self._failure_event(
                inputs=inputs,
                process_data=process_data,
                metadata=metadata,
                error=str(error),
                error_type="agent_workflow_node_runtime_error",
            )

    def _run_inner(
        self,
        *,
        inputs: dict[str, Any],
        process_data: dict[str, Any],
        metadata: dict[str, Any],
    ) -> Generator[NodeEventPayload | NodeRunPauseRequestedEvent]:
        from runtime_contracts.builder.execution import BuilderExecutionPolicyError
        from services.dify_builder.native_testing.agent_binding import AgentSnapshotChanged

        dify_ctx = DifyRunContext.model_validate(self.require_run_context_value(DIFY_RUN_CONTEXT_KEY))
        confirmed = dify_ctx.builder_agent_execution
        workflow_id = self.init_params.workflow_id
        workflow_run_id = get_system_text(
            self.runtime_state.variable_pool,
            SystemVariableKey.WORKFLOW_EXECUTION_ID,
        )
        # Chatflow sessions also carry their owning conversation.
        conversation_id = get_system_text(
            self.runtime_state.variable_pool,
            SystemVariableKey.CONVERSATION_ID,
        )
        if confirmed is not None and (
            workflow_run_id != confirmed.native_run_id
            or conversation_id is not None
            or dify_ctx.invoke_from.value != "debugger"
            or dify_ctx.user_from.value != "account"
            or dify_ctx.builder_execution is not None
            or dify_ctx.native_test_required
            or dify_ctx.native_test_ref is not None
            or self._agent_preparation is None
            or self._agent_commands is None
            or self._agent_run_recovery is None
        ):
            raise BuilderExecutionPolicyError("native_agent_owner_mismatch")

        # ──── Setup: resolve binding once + extract declared outputs for stage 4 checks ────
        try:
            builder_resolution: dict[str, Any] = {}
            # The ordinary Builder Test Run always supplies this server-owned
            # map, including {} for Agent-only graphs. Ordinary runs leave it
            # None. Recheck its graph selection at the actual Agent boundary.
            if dify_ctx.builder_tool_credential_bindings is not None:
                from core.dify_builder.models import Actor
                from services.dify_builder.native_testing.agent_binding import authorize_graph_agent_selection

                nodes = [
                    node
                    for node in self.init_params.graph_config.get("nodes", [])
                    if isinstance(node, Mapping) and node.get("id") == self._node_id
                ]
                if len(nodes) != 1 or not isinstance(nodes[0].get("data"), Mapping):
                    raise BuilderExecutionPolicyError("agent_binding_not_prepared")
                actor = Actor(tenant_id=dify_ctx.tenant_id, account_id=dify_ctx.user_id)
                selection = authorize_graph_agent_selection(
                    actor=actor, app_id=dify_ctx.app_id, node_data=nodes[0]["data"]
                )
                builder_resolution = {
                    "builder_actor": actor,
                    "builder_authorization": selection.authorization,
                    "selected_agent_id": selection.agent_id,
                    "selected_snapshot_id": selection.snapshot_id,
                }
            existing_scope = self._session_store.load_existing_node_execution_scope(
                tenant_id=dify_ctx.tenant_id,
                app_id=dify_ctx.app_id,
                workflow_id=workflow_id,
                workflow_run_id=workflow_run_id,
                node_id=self._node_id,
                node_execution_id=self.execution_id,
                conversation_id=conversation_id,
                workflow_tool_invocation_id=dify_ctx.workflow_tool_invocation_id,
            )
            bundle = self._binding_resolver.resolve(
                tenant_id=dify_ctx.tenant_id,
                app_id=dify_ctx.app_id,
                workflow_id=workflow_id,
                node_id=self._node_id,
                binding_id=existing_scope.workflow_agent_binding_id if existing_scope is not None else None,
                snapshot_id=existing_scope.agent_config_snapshot_id if existing_scope is not None else None,
                conversation_id=conversation_id if dify_ctx.workflow_tool_invocation_id is None else None,
                **builder_resolution,
            )
        except (AgentSnapshotChanged, BuilderExecutionPolicyError) as error:
            yield self._failure_event(
                inputs=inputs,
                process_data=process_data,
                metadata=metadata,
                error=str(error),
                error_type="agent_snapshot_changed" if isinstance(error, AgentSnapshotChanged) else error.reason_code,
            )
            return
        except WorkflowAgentBindingError as error:
            yield self._failure_event(
                inputs=inputs,
                process_data=process_data,
                metadata=metadata,
                error=str(error),
                error_type=error.error_code,
            )
            return
        except AgentWorkspaceNotFoundError as error:
            yield self._failure_event(
                inputs=inputs,
                process_data=process_data,
                metadata=metadata,
                error=str(error),
                error_type="agent_workflow_node_runtime_error",
            )
            return

        process_data.update(
            {
                "agent_id": bundle.agent.id,
                "agent_config_snapshot_id": bundle.snapshot.id,
                "workflow_agent_binding_id": bundle.binding.id,
            }
        )
        if dify_ctx.workflow_tool_invocation_id is not None:
            process_data[WORKFLOW_TOOL_INVOCATION_ID_KEY] = dify_ctx.workflow_tool_invocation_id
        session_scope = existing_scope or WorkflowAgentSessionScope(
            tenant_id=dify_ctx.tenant_id,
            app_id=dify_ctx.app_id,
            workflow_id=workflow_id,
            workflow_run_id=workflow_run_id,
            node_id=self._node_id,
            node_execution_id=self.execution_id,
            workflow_agent_binding_id=bundle.binding.id,
            agent_id=bundle.agent.id,
            agent_config_snapshot_id=bundle.snapshot.id,
            conversation_id=conversation_id,
            workflow_tool_invocation_id=dify_ctx.workflow_tool_invocation_id,
        )

        node_job = WorkflowNodeJobConfig.model_validate(bundle.binding.node_job_config_dict)
        node_job.output_routes.validate_for_execution()
        custom_outputs = list(node_job.declared_outputs)
        outputs_by_name = {output.name: output for output in custom_outputs}

        stored_session = self._session_store.load_or_create_node_execution_session(
            session_scope,
            home_snapshot_id=bundle.snapshot.home_snapshot_id,
        )
        # ──── Retry loop (Stage 4 §7) ────
        attempt = 0
        while True:
            parent_node = None
            snapshot_identity = None
            after = None
            runtime_request = None
            create_response = None
            try:
                if confirmed is not None:
                    from services.agent.workflow_run_recovery import AgentCreateAcknowledgementUnknown

                    try:
                        create_response, parent_node, after, snapshot_identity = self._start_or_resume_confirmed(
                            dify_context=dify_ctx,
                            session_scope=session_scope,
                            stored_session=stored_session,
                            process_data=process_data,
                        )
                    except AgentCreateAcknowledgementUnknown:
                        yield self._control_pause(inputs, process_data, metadata, "agent_create_ack_unknown")
                        return
                    except BuilderExecutionPolicyError as error:
                        if error.reason_code == "native_agent_continuation_unknown":
                            yield self._control_pause(inputs, process_data, metadata, error.reason_code)
                            return
                        raise
                    inputs["agent_backend_request"] = {"execution_mode": "builder_test", "profile": "confirmed"}
                    metadata["attempt"] = 0
                    metadata["agent_backend"] = {"run_id": create_response.run_id, "status": create_response.status}
                else:
                    runtime_request = self._runtime_request_builder.build(
                        WorkflowAgentRuntimeBuildContext(
                            dify_context=dify_ctx,
                            workflow_id=workflow_id,
                            workflow_run_id=workflow_run_id,
                            node_id=self._node_id,
                            node_execution_id=self.execution_id,
                            variable_pool=self.runtime_state.variable_pool,
                            binding=bundle.binding,
                            agent=bundle.agent,
                            snapshot=bundle.snapshot,
                            binding_id=stored_session.binding_id,
                            backend_binding_ref=stored_session.backend_binding_ref,
                            attempt=attempt,
                            session_snapshot=stored_session.session_snapshot,
                        )
                    )
            except WorkflowAgentRuntimeRequestBuildError as error:
                yield self._failure_event(
                    inputs=inputs,
                    process_data=process_data,
                    metadata=metadata,
                    error=str(error),
                    error_type=error.error_code,
                )
                return
            except Exception as error:
                yield self._failure_event(
                    inputs=inputs,
                    process_data=process_data,
                    metadata=metadata,
                    error=str(error),
                    error_type="agent_workflow_node_runtime_error",
                )
                return

            # Capture inputs only from the first attempt so retry doesn't churn the
            # node's "inputs" payload that ends up in the workflow detail view.
            if confirmed is None:
                assert runtime_request is not None
                if attempt == 0:
                    inputs["agent_backend_request"] = runtime_request.redacted_request
                metadata.clear()
                metadata.update(runtime_request.metadata)
                metadata["attempt"] = attempt
                try:
                    create_response = self._agent_backend_client.create_run(runtime_request.request)
                except AgentBackendError as error:
                    yield self._failure_event(
                        inputs=inputs,
                        process_data=process_data,
                        metadata=metadata,
                        error=str(error),
                        error_type=self._agent_backend_error_type(error),
                    )
                    return

            assert create_response is not None
            metadata["agent_backend"] = {
                **dict(metadata.get("agent_backend") or {}),
                "run_id": create_response.run_id,
                "status": create_response.status,
            }

            terminal_event, exhausted = yield from self._consume_event_stream(
                create_response.run_id,
                inputs=inputs,
                process_data=process_data,
                metadata=metadata,
                stream_text=not custom_outputs
                and not node_job.output_routes.enabled
                and self.error_strategy != ErrorStrategy.DEFAULT_VALUE,
                after=after,
            )
            if isinstance(terminal_event, AgentBackendRunWaitingInternalEvent):
                if confirmed is None or parent_node is None:
                    yield self._failure_event(
                        inputs=inputs,
                        process_data=process_data,
                        metadata=metadata,
                        error="Agent wait has no confirmed owner",
                        error_type="agent_wait_unsupported",
                    )
                    return
                try:
                    if terminal_event.run_id != create_response.run_id or terminal_event.source_event_id is None:
                        raise BuilderExecutionPolicyError("native_agent_checkpoint_unavailable")
                    assert self._agent_commands is not None
                    wait = self._agent_commands.prepare_wait(
                        ref=confirmed.ref,
                        parent_node=parent_node,
                        agent_generation=terminal_event.generation,
                        backend_wait_event_id=terminal_event.source_event_id,
                    )
                    process_data["agent_wait"] = wait.model_dump(mode="json")
                except BuilderExecutionPolicyError:
                    yield self._control_pause(inputs, process_data, metadata, "native_agent_wait_unverified")
                    return
                yield self._control_pause(inputs, process_data, metadata, "agent_tool_approval")
                return
            if (
                confirmed is not None
                and not self._is_graph_aborted()
                and (exhausted is not None or terminal_event is None)
            ):
                yield self._control_pause(inputs, process_data, metadata, "agent_stream_ack_unknown")
                return
            if metadata.get("knowledge_fs_sources"):
                yield RunRetrieverResourceEvent(retriever_resources=metadata["knowledge_fs_sources"], context="")
            # None means no post-exit snapshot was produced; leave the previously stored session snapshot untouched.
            if (
                isinstance(terminal_event, AgentBackendRunFailedInternalEvent | AgentBackendRunCancelledInternalEvent)
                and terminal_event.session_snapshot is not None
            ):
                self._save_session_snapshot(
                    session_scope=session_scope,
                    binding_id=stored_session.binding_id,
                    snapshot=terminal_event.session_snapshot,
                    metadata=metadata,
                    confirmed_identity=snapshot_identity,
                )
            if exhausted is not None:
                # Streaming error / unexpected end — surface immediately without
                # retrying because the failure is transport-level.
                yield exhausted
                return
            if terminal_event is None:
                yield StreamCompletedEvent(
                    node_run_result=self._output_adapter.build_stream_exhausted_result(
                        inputs=inputs,
                        process_data=process_data,
                        metadata=metadata,
                    )
                )
                return

            # A failed attempt does not retire the product-owned Binding. The
            # Workflow Run terminal lifecycle event owns that transition.
            if not isinstance(terminal_event, AgentBackendRunSucceededInternalEvent):
                yield StreamCompletedEvent(
                    node_run_result=self._output_adapter.build_failure_result(
                        event=terminal_event,
                        inputs=inputs,
                        process_data=process_data,
                        metadata=metadata,
                    )
                )
                return

            self._save_session_snapshot(
                session_scope=session_scope,
                binding_id=stored_session.binding_id,
                snapshot=terminal_event.session_snapshot,
                metadata=metadata,
                confirmed_identity=snapshot_identity,
            )

            success_event = terminal_event
            if custom_outputs:
                # ──── Stage 4: per-output type check ────
                type_check = self._type_checker.check(
                    declared_outputs=custom_outputs,
                    raw_output=terminal_event.output,
                    tenant_id=dify_ctx.tenant_id,
                )
                self._record_type_check_metadata(metadata, type_check)

                if type_check.has_failures:
                    # ──── Stage 4: orchestrate retry / default / fail ────
                    failures = [
                        FailedOutput(
                            declared=outputs_by_name[result.name],
                            failure_kind=OutputFailureKind.TYPE_CHECK,
                            reason=result.reason,
                        )
                        for result in type_check.failures
                        if result.name in outputs_by_name
                    ]
                    outcome = self._failure_orchestrator.decide(failures=failures, current_attempt=attempt)
                    metadata["output_failure_decision"] = outcome.decision.value
                    metadata["output_failure_reason"] = outcome.primary_reason

                    if outcome.decision == OutputFailureDecision.RETRY:
                        if confirmed is not None:
                            yield self._failure_event(
                                inputs=inputs,
                                process_data=process_data,
                                metadata=metadata,
                                error="Confirmed Agent output validation cannot create another child automatically.",
                                error_type="native_agent_output_retry_unavailable",
                            )
                            return
                        attempt = outcome.next_attempt
                        continue

                    if outcome.decision == OutputFailureDecision.USE_DEFAULT:
                        success_event = self._patch_event_with_defaults(terminal_event, outcome.per_output_actions)
                    else:
                        error_type = (
                            "output_type_check_failed_fail_branch"
                            if outcome.decision == OutputFailureDecision.TAKE_FAIL_BRANCH
                            else "output_type_check_failed"
                        )
                        yield self._failure_event(
                            inputs=inputs,
                            process_data=process_data,
                            metadata=metadata,
                            error=outcome.primary_reason,
                            error_type=error_type,
                        )
                        return

            # Select only after this attempt's custom outputs have passed. Failed
            # route output uses the node's normal retry / failure-branch handling.
            routes = node_job.output_routes
            edge_source_handle = "source"
            if routes.enabled:
                selected = success_event.output.get("switch") if isinstance(success_event.output, dict) else None
                if not isinstance(selected, str) or selected not in {route.id for route in routes.routes}:
                    yield self._failure_event(
                        inputs=inputs,
                        process_data=process_data,
                        metadata=metadata,
                        error="Agent output switch must select one configured output route ID.",
                        error_type="output_route_selection_failed",
                    )
                    return
                edge_source_handle = selected

            yield StreamCompletedEvent(
                node_run_result=self._output_adapter.build_success_result(
                    event=success_event,
                    edge_source_handle=edge_source_handle,
                    inputs=inputs,
                    process_data=process_data,
                    metadata=metadata,
                    declared_outputs=custom_outputs,
                )
            )
            return

    def _start_or_resume_confirmed(self, *, dify_context, session_scope, stored_session, process_data):
        from runtime_contracts.builder.execution import BuilderExecutionPolicyError, canonical_digest
        from runtime_contracts.builder.native.agent_contracts import AgentCreateInserted, AgentParentNodeRef
        from services.agent.workflow_run_recovery import (
            AgentCreateAcknowledgementUnknown,
            AgentCreateRecoveryClient,
            AgentCreateScope,
        )
        from services.dify_builder.native_testing.agent_confirmed_workspace import (
            load_confirmed_agent_parent,
            prepare_confirmed_agent_request,
        )

        marker = dify_context.builder_agent_execution
        assert marker is not None
        assert self._agent_preparation is not None
        assert self._agent_commands is not None
        assert self._agent_run_recovery is not None
        scope = AgentCreateScope(
            tenant_id=session_scope.tenant_id,
            app_id=session_scope.app_id,
            workflow_id=session_scope.workflow_id,
            workflow_run_id=session_scope.workflow_run_id,
            node_id=self._node_id,
            execution_id=self.execution_id,
            workflow_binding_id=session_scope.workflow_agent_binding_id,
        )
        client = self._agent_backend_client
        assert isinstance(client, AgentCreateRecoveryClient)
        with _confirmed_node_phase(marker, "parent_restore"):
            parent = load_confirmed_agent_parent(
                state_store=self._agent_preparation.state_store, ref=marker.ref, execution_id=self.execution_id
            )
        with _confirmed_node_phase(marker, "recover_existing"):
            response = self._agent_run_recovery.recover_existing(
                scope=scope,
                client=client,
                on_persisted=process_data.update,
            )
        if parent is not None:
            if response is None:
                # A previously prepared occurrence cannot acquire first-POST
                # authority merely because another process stopped before ACK.
                raise AgentCreateAcknowledgementUnknown("Agent create acknowledgement is unknown")
            with _confirmed_node_phase(marker, "load_wait"):
                protected = self._agent_commands.load_wait(ref=marker.ref, parent_node=parent)
            if protected is not None:
                if protected.association.agent_run_id != response.run_id:
                    raise BuilderExecutionPolicyError("native_agent_owner_mismatch")
                with _confirmed_node_phase(marker, "continue_wait"):
                    continued = self._agent_commands.continue_wait(ref=marker.ref, parent_node=parent)
                if continued.association.agent_run_id != response.run_id or continued.wait is None:
                    raise BuilderExecutionPolicyError("native_agent_owner_mismatch")
                identity = self._capture_confirmed_snapshot(
                    parent=parent,
                    response=response,
                    stored_session=stored_session,
                    agent_generation=continued.wait.agent_generation + 1,
                )
                return response, parent, continued.wait.backend_wait_event_id, identity
            identity = self._capture_confirmed_snapshot(
                parent=parent,
                response=response,
                stored_session=stored_session,
                agent_generation=0,
            )
            return response, parent, None, identity
        if response is not None:
            raise BuilderExecutionPolicyError("native_agent_input_unavailable")
        context = self._agent_preparation.state_store.load_context(ref=marker.ref)
        builder = self._agent_preparation.config_builder
        if builder is None:
            raise BuilderExecutionPolicyError("native_agent_preparation_unavailable")
        definition = builder.load_definition(context=context, node_id=self._node_id)
        typed_inputs = builder.typed_inputs(
            runtime_state=self.runtime_state,
            definition=definition,
        )
        input_ref = self._agent_preparation.state_store.store_node_inputs(
            ref=marker.ref,
            generation=context.generation,
            native_execution_id=self.execution_id,
            inputs=typed_inputs,
        )
        parent = AgentParentNodeRef(
            ref=marker.ref,
            execution_id=self.execution_id,
            input_result_ref=input_ref,
            input_digest=canonical_digest(typed_inputs),
        )
        prepared = prepare_confirmed_agent_request(
            service=self._agent_preparation,
            ref=marker.ref,
            parent_node=parent,
            node_id=self._node_id,
            stored_session=stored_session,
            inputs=typed_inputs,
        )
        if not isinstance(prepared.claim, AgentCreateInserted):
            raise AgentCreateAcknowledgementUnknown("Agent create acknowledgement is unknown")
        response = self._agent_run_recovery.start_or_recover(
            scope=scope,
            attempt=0,
            request=prepared.request,
            client=client,
            on_persisted=process_data.update,
        )
        identity = self._capture_confirmed_snapshot(
            parent=parent,
            response=response,
            stored_session=stored_session,
            agent_generation=0,
        )
        return response, parent, None, identity

    def _capture_confirmed_snapshot(self, *, parent, response, stored_session, agent_generation):
        """Obtain acknowledged association and capture this stream's original tuple.

        Initial bind uses the existing runtime identity owner, never an invented
        association ref. Failure refuses persistence while stream/cancellation
        retain their existing independent behavior.
        """
        from dify_agent.protocol.confirmed_recovery import AgentRecoveryBinding
        from dify_agent.protocol.native_testing import AgentParentNodeRef as WireParent
        from dify_agent.protocol.native_testing import AgentPrepareRunPayload

        from runtime_contracts.builder.execution import BuilderExecutionPolicyError

        assert self._agent_preparation is not None
        assert self._agent_commands is not None
        try:
            store = self._agent_preparation.state_store
            context = store.load_context(ref=parent.ref)
            config = store.load_prepared_config(ref=parent.ref, parent_node=parent, config_ref=None)
            try:
                protected = store.load_agent_association(ref=parent.ref, parent_node=parent)
            except BuilderExecutionPolicyError as error:
                if error.reason_code != "native_agent_object_unavailable" or agent_generation != 0:
                    raise
                self._agent_preparation.bind_created_run(
                    payload=AgentPrepareRunPayload(
                        parent_node=WireParent.model_validate(parent.model_dump(mode="json")),
                        agent_run_id=response.run_id,
                        request_digest=config.config_digest,
                    )
                )
                protected = store.load_agent_association(ref=parent.ref, parent_node=parent)
            workspace = protected.association.workspace
            if (stored_session.binding_id, stored_session.workspace_id, stored_session.backend_binding_ref) != (
                workspace.binding_id,
                workspace.workspace_id,
                workspace.backend_binding_ref,
            ):
                raise BuilderExecutionPolicyError("native_agent_snapshot_identity_mismatch")
            return self._agent_commands.capture_session_snapshot_identity(
                parent_node=parent,
                parent_generation=context.generation,
                agent_run_id=response.run_id,
                agent_generation=agent_generation,
                binding=AgentRecoveryBinding(
                    binding_id=workspace.binding_id,
                    workspace_id=workspace.workspace_id,
                    binding_ref=workspace.backend_binding_ref,
                    workspace_ref=workspace.backend_workspace_ref,
                    workspace_generation=workspace.workspace_generation,
                    association_object_ref=protected.association_object_ref,
                    config_ref=config.config_ref,
                    config_digest=config.config_digest,
                ),
            )
        except Exception as error:
            logger.warning(
                "Failed to capture confirmed Agent session snapshot identity: "
                "request_id=%s node_execution_id=%s agent_run_id=%s error_type=%s",
                parent.ref.request_id,
                parent.execution_id,
                response.run_id,
                type(error).__name__,
            )
            return None

    def _control_pause(self, inputs, process_data, metadata, reason):
        metadata["agent_backend"] = {**dict(metadata.get("agent_backend") or {}), "status": reason}
        return NodeRunPauseRequestedEvent(
            id=self.execution_id,
            node_id=self._node_id,
            node_type=self.node_type,
            node_version=self.version(),
            reason=SchedulingPause(message=reason),
            node_run_result=NodeRunResult(
                status=WorkflowNodeExecutionStatus.PAUSED,
                inputs=inputs,
                process_data=process_data,
                outputs={},
                metadata={WorkflowNodeExecutionMetadataKey.AGENT_LOG: metadata},
            ),
        )

    def _consume_event_stream(
        self,
        run_id: str,
        *,
        inputs: dict[str, Any],
        process_data: dict[str, Any],
        metadata: dict[str, Any],
        stream_text: bool,
        after: str | None = None,
    ) -> Generator[
        StreamChunkEvent,
        None,
        tuple[_TerminalAgentBackendEvent | None, StreamCompletedEvent | None],
    ]:
        """Consume the SSE stream for one Agent backend run.

        When ``stream_text`` is set (plain runs whose final string becomes the
        ``text`` output), agent-message deltas are yielded as ``text`` stream
        chunks so Answer nodes can stream them. The stream is closed only after
        at least one chunk was sent; otherwise the response coordinator falls
        back to the final ``text`` variable. Structured runs keep deltas as
        metadata because their ``text`` comes from the structured output.
        Plain runs with the default-value error strategy stay buffered too:
        once a selector has stream chunks, the response filter no longer reads
        the variable pool, so a configured fallback ``text`` would be dropped
        after a partial stream.

        Returns a 2-tuple ``(terminal_event, transport_failure)``:
        - ``terminal_event``: the first non-stream/non-started internal event,
          or ``None`` if the stream ended without one.
        - ``transport_failure``: a populated ``StreamCompletedEvent`` when the
          stream itself errored (backend/HTTP/protocol fault). A cancellation
          terminal may accompany it so the caller can persist the final session
          snapshot while preserving the original transport failure.
        """
        result = yield from self._read_event_stream(
            run_id,
            inputs=inputs,
            process_data=process_data,
            metadata=metadata,
            stream_text=stream_text,
            after=after,
        )
        if (
            not isinstance(result[0], AgentBackendRunWaitingInternalEvent)
            and stream_text
            and (metadata.get("agent_backend") or {}).get("agent_message_delta_count")
        ):
            yield StreamChunkEvent(selector=[self._node_id, "text"], chunk="", is_final=True)
        return result

    def _read_event_stream(
        self,
        run_id: str,
        *,
        inputs: dict[str, Any],
        process_data: dict[str, Any],
        metadata: dict[str, Any],
        stream_text: bool,
        after: str | None = None,
    ) -> Generator[
        StreamChunkEvent,
        None,
        tuple[_TerminalAgentBackendEvent | None, StreamCompletedEvent | None],
    ]:
        stream_event_count = 0
        last_event_id: str | None = None
        confirmed = (
            DifyRunContext.model_validate(self.require_run_context_value(DIFY_RUN_CONTEXT_KEY)).builder_agent_execution
            is not None
        )
        try:
            for public_event in self._agent_backend_client.stream_events(
                run_id,
                should_stop=self._is_graph_aborted,
                **({"after": after} if after is not None else {}),
            ):
                stream_event_count += 1
                if public_event.id is not None:
                    last_event_id = public_event.id
                for internal_event in self._event_adapter.adapt(public_event):
                    if internal_event.type in (
                        AgentBackendInternalEventType.RUN_STARTED,
                        AgentBackendInternalEventType.RUN_RESUMED,
                    ):
                        continue
                    if internal_event.type == AgentBackendInternalEventType.STREAM_EVENT:
                        if isinstance(internal_event, AgentBackendStreamInternalEvent):
                            self._record_stream_metadata(metadata, internal_event)
                        continue
                    if internal_event.type == AgentBackendInternalEventType.AGENT_MESSAGE_DELTA:
                        if isinstance(internal_event, AgentBackendAgentMessageDeltaInternalEvent):
                            self._record_agent_message_delta_metadata(metadata, internal_event)
                            if stream_text:
                                yield StreamChunkEvent(
                                    selector=[self._node_id, "text"],
                                    chunk=internal_event.delta,
                                    is_final=False,
                                )
                        continue
                    metadata["agent_backend"] = {
                        **dict(metadata.get("agent_backend") or {}),
                        "stream_event_count": stream_event_count,
                    }
                    # Known waiting is control only on the explicit confirmed path.
                    if isinstance(internal_event, AgentBackendRunWaitingInternalEvent) and confirmed:
                        return internal_event, None
                    if isinstance(
                        internal_event,
                        AgentBackendRunSucceededInternalEvent
                        | AgentBackendRunFailedInternalEvent
                        | AgentBackendRunCancelledInternalEvent,
                    ):
                        return internal_event, None
                    cancellation = (
                        None
                        if confirmed
                        else self._cancel_backend_run(
                            run_id,
                            reason="unexpected_event",
                            after=last_event_id,
                        )
                    )
                    return cancellation, self._failure_event(
                        inputs=inputs,
                        process_data=process_data,
                        metadata=metadata,
                        error=f"Unexpected internal event type {internal_event.type!r}",
                        error_type="agent_backend_stream_error",
                    )
        except AgentBackendError as error:
            cancellation = (
                None
                if confirmed and not self._is_graph_aborted()
                else self._cancel_backend_run(
                    run_id,
                    reason=self._stream_stop_reason(),
                    after=last_event_id,
                )
            )
            return cancellation, self._failure_event(
                inputs=inputs,
                process_data=process_data,
                metadata=metadata,
                error=str(error),
                error_type=self._agent_backend_error_type(error),
            )
        except Exception as error:
            cancellation = (
                None
                if confirmed and not self._is_graph_aborted()
                else self._cancel_backend_run(
                    run_id,
                    reason=self._stream_stop_reason(),
                    after=last_event_id,
                )
            )
            return cancellation, self._failure_event(
                inputs=inputs,
                process_data=process_data,
                metadata=metadata,
                error=str(error),
                error_type="agent_backend_stream_error",
            )

        cancellation = (
            None
            if confirmed and not self._is_graph_aborted()
            else self._cancel_backend_run(
                run_id,
                reason=self._stream_stop_reason()
                if self._is_graph_aborted()
                else "stream_ended_without_terminal_event",
                after=last_event_id,
            )
        )
        return cancellation, None

    def _is_graph_aborted(self) -> bool:
        """Let Agent SSE consumption observe Engine's cooperative abort state."""
        return self.runtime_state.graph_execution.aborted

    def _stream_stop_reason(self) -> str:
        return "workflow_graph_aborted" if self._is_graph_aborted() else "event_stream_failed"

    def _cancel_backend_run(
        self,
        run_id: str,
        *,
        reason: str,
        after: str | None,
    ) -> AgentBackendRunCancelledInternalEvent | None:
        try:
            public_event = self._agent_backend_client.cancel_run_and_wait(
                run_id,
                CancelRunRequest(reason=reason, message="Workflow Agent event consumption stopped"),
                after=after,
            )
            for internal_event in self._event_adapter.adapt(public_event):
                if isinstance(internal_event, AgentBackendRunCancelledInternalEvent):
                    return internal_event
        except Exception:
            logger.warning("Failed to finish cancelling Workflow Agent backend run: run_id=%s", run_id, exc_info=True)
        return None

    @staticmethod
    def _record_type_check_metadata(metadata: dict[str, Any], outcome: OutputTypeCheckOutcome) -> None:
        # Surface enough detail in metadata for Inspector / debug logs without
        # leaking the raw failing values (which may be sensitive).
        metadata["output_type_check"] = {
            "passed": not outcome.has_failures,
            "results": [
                {
                    "name": r.name,
                    "type": r.declared_type.value,
                    "status": r.status.value,
                    "reason": r.reason,
                }
                for r in outcome.results
            ],
        }

    def _save_session_snapshot(
        self,
        *,
        session_scope: WorkflowAgentSessionScope,
        binding_id: str,
        snapshot: SessionSnapshot | None,
        metadata: dict[str, Any],
        confirmed_identity: ConfirmedSnapshotIdentity | None = None,
    ) -> None:
        try:
            context = DifyRunContext.model_validate(self.require_run_context_value(DIFY_RUN_CONTEXT_KEY))
            if context.builder_agent_execution is not None:
                from runtime_contracts.builder.execution import BuilderExecutionPolicyError

                if self._agent_commands is None or confirmed_identity is None:
                    raise BuilderExecutionPolicyError("native_agent_snapshot_identity_unavailable")
                if (confirmed_identity.parent_node.ref, confirmed_identity.binding.binding_id) != (
                    context.builder_agent_execution.ref,
                    binding_id,
                ):
                    raise BuilderExecutionPolicyError("native_agent_snapshot_identity_mismatch")
                self._agent_commands.save_session_snapshot(
                    identity=confirmed_identity,
                    session_scope=session_scope,
                    snapshot=snapshot,
                )
            else:
                self._session_store.save_active_snapshot(
                    scope=session_scope,
                    binding_id=binding_id,
                    snapshot=snapshot,
                )
            agent_backend = dict(metadata.get("agent_backend") or {})
            agent_backend["session_snapshot_persisted"] = snapshot is not None
            metadata["agent_backend"] = agent_backend
        except Exception:
            logger.warning(
                "Failed to persist workflow Agent Binding session snapshot: "
                "tenant_id=%s workflow_run_id=%s node_id=%s binding_id=%s agent_id=%s",
                session_scope.tenant_id,
                session_scope.workflow_run_id,
                session_scope.node_id,
                session_scope.workflow_agent_binding_id,
                session_scope.agent_id,
                exc_info=True,
            )
            agent_backend = dict(metadata.get("agent_backend") or {})
            agent_backend["session_snapshot_persisted"] = False
            agent_backend["session_snapshot_persist_error"] = "workflow_agent_workspace_store_error"
            metadata["agent_backend"] = agent_backend

    @staticmethod
    def _patch_event_with_defaults(
        event: AgentBackendRunSucceededInternalEvent,
        per_output_actions: Mapping[str, Any],
    ) -> AgentBackendRunSucceededInternalEvent:
        """Merge USE_DEFAULT replacements into the success event's output dict.

        The event is a frozen dataclass / Pydantic model; we copy with the
        replacements applied so downstream code (output_adapter normalize) sees
        the patched payload.
        """
        if not per_output_actions:
            return event
        original = event.output if isinstance(event.output, Mapping) else {}
        patched_output: dict[str, Any] = dict(original)
        patched_output.update(per_output_actions)
        return event.model_copy(update={"output": patched_output})

    @staticmethod
    def _failure_event(
        *,
        inputs: dict[str, Any],
        process_data: dict[str, Any],
        metadata: dict[str, Any],
        error: str,
        error_type: str,
    ) -> StreamCompletedEvent:
        return StreamCompletedEvent(
            node_run_result=NodeRunResult(
                status=WorkflowNodeExecutionStatus.FAILED,
                inputs=inputs,
                process_data=process_data,
                metadata={WorkflowNodeExecutionMetadataKey.AGENT_LOG: metadata},
                outputs={},
                error=error,
                error_type=error_type,
            )
        )

    @staticmethod
    def _agent_backend_error_type(error: AgentBackendError) -> str:
        if isinstance(error, AgentBackendValidationError):
            return "agent_backend_validation_error"
        if isinstance(error, AgentBackendHTTPError):
            return "agent_backend_http_error"
        if isinstance(error, AgentBackendStreamError):
            return "agent_backend_stream_error"
        if isinstance(error, AgentBackendTransportError):
            return "agent_backend_transport_error"
        return "agent_backend_error"

    @staticmethod
    def _record_stream_metadata(metadata: dict[str, Any], event: AgentBackendStreamInternalEvent) -> None:
        agent_backend = dict(metadata.get("agent_backend") or {})
        agent_backend["last_stream_event_id"] = event.source_event_id
        if event.event_kind:
            agent_backend["last_stream_event_kind"] = event.event_kind
        if isinstance(event.data, Mapping):
            sources = knowledge_sources_from_stream_event(event.data)
            if sources:
                current = metadata.setdefault("knowledge_fs_sources", [])
                seen = {source["knowledge_fs_citation"]["id"] for source in current}
                for source in sources:
                    if (
                        source.knowledge_fs_citation
                        and source.knowledge_fs_citation.id not in seen
                        and len(current) < 200
                    ):
                        current.append(source.model_dump(mode="json"))
                        seen.add(source.knowledge_fs_citation.id)
            usage = event.data.get("usage") or event.data.get("model_usage")
            if isinstance(usage, Mapping):
                agent_backend["usage"] = dict(usage)
        metadata["agent_backend"] = agent_backend

    @staticmethod
    def _record_agent_message_delta_metadata(
        metadata: dict[str, Any], event: AgentBackendAgentMessageDeltaInternalEvent
    ) -> None:
        agent_backend = dict(metadata.get("agent_backend") or {})
        agent_backend["agent_message_delta_count"] = int(agent_backend.get("agent_message_delta_count") or 0) + 1
        agent_backend["agent_message_delta_length"] = int(agent_backend.get("agent_message_delta_length") or 0) + len(
            event.delta
        )
        metadata["agent_backend"] = agent_backend

    @classmethod
    @override
    def _extract_variable_selector_to_variable_mapping(
        cls,
        *,
        graph_config: Mapping[str, Any],
        node_id: str,
        node_data: DifyAgentNodeData,
    ) -> Mapping[str, Sequence[str]]:
        """Reuse frontend workflow-marker parsing for graph variable loading.

        Task mentions use the publish parser. Enabled route conditions use
        Graphon template selectors, including environment and system variables;
        disabled jobs do not load their saved conditions.
        """
        del graph_config

        if isinstance(node_data, Mapping):
            task = node_data.get("agent_task", "")
            routes = WorkflowOutputRoutes.model_validate(node_data.get("agent_output_routes", {}))
        else:
            task = node_data.agent_task
            routes = node_data.agent_output_routes
        selectors = list(extract_workflow_node_output_selectors(task))
        if routes.enabled:
            for route in routes.routes:
                selectors.extend(
                    tuple(selector.value_selector)
                    for selector in VariableTemplateParser(template=route.name).extract_variable_selectors()
                )
        return {f"{node_id}.{'.'.join(selector)}": list(selector) for selector in selectors}