ai-video-fullstack/backend/services/brains/workflow_brain.py

"""Pipecat Flows-backed Workflow v3 brain."""

from __future__ import annotations

import asyncio
from collections.abc import Awaitable
from copy import deepcopy
from dataclasses import dataclass, replace
from typing import Any

from loguru import logger
from models import AssistantConfig, RuntimeTool
from db.session import SessionLocal
from pipecat.flows import (
    ContextStrategy,
    ContextStrategyConfig,
    FlowManager,
    FlowsFunctionSchema,
    NodeConfig,
)
from pipecat.frames.frames import (
    LLMRunFrame,
    LLMUpdateSettingsFrame,
    OutputTransportMessageUrgentFrame,
)
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.frame_processor import FrameProcessor
from pipecat.services.llm_service import (
    FunctionCallParams,
    FunctionCallResultProperties,
)
from services.brains.base import (
    BaseBrain,
    BrainRuntime,
    BrainSpec,
    SessionVariableUpdate,
)
from services.action_runtime import (
    ActionInvocationCancelled,
    ActionOutcome,
    ActionRunner,
    ActionStatus,
)
from services.action_stage import ActionStageRunner, ActionStageSpec, StageAction
from services.fixed_speech import fixed_speech_context_message
from services.knowledge import search as search_knowledge
from services.message_policy import (
    MESSAGE_COMPLETION_POLICIES,
    MESSAGE_CONFIRMATION,
    MESSAGE_INTERRUPTIBLE,
    MESSAGE_PLAYBACK,
)
from services.message_stage import (
    MessageDisplaySpec,
    MessageStageResult,
    MessageStageRunner,
    MessageStageSpec,
)
from services.runtime_variables import DynamicVariableStore
from services.tool_executor import ToolExecutionError, ToolExecutor
from services.tool_policy import policy_for_tool
from services.workflow.agent import WorkflowAgentStage
from services.workflow.models import RouteStatus, WorkflowRuntimeState, WorkflowStatus
from services.workflow.output import WorkflowOutput
from services.workflow.routing import WorkflowEdgeEvaluator
from services.workflow_engine import WorkflowEngine
from services.workflow_router import WorkflowLLMRouter


MAX_AUTOMATIC_HOPS = 50


@dataclass
class _MessageContinuation:
    """Resume one Workflow path after its visible Message gate completes."""

    token: int
    node_id: str
    context_messages: list[dict[str, Any]]
    triggering_user_text: str
    triggering_user_message: dict[str, Any] | None
    task: asyncio.Task[None] | None = None


class ConfiguredFlowManager(FlowManager):
    """Preserve Flow transitions while suppressing late async-tool replies."""

    ENTRY_ACTION_TYPE = "workflow_function_transition_entry"

    async def _set_node(self, node_id: str, node_config: NodeConfig) -> None:
        """Notify Workflow only after FlowManager committed the active node."""
        await super()._set_node(node_id, node_config)
        after_activation = node_config.get("workflow_after_activation")
        if callable(after_activation):
            await after_activation(node_id)

    async def _create_transition_func(self, name, handler):
        transition = await super()._create_transition_func(name, handler)
        native_vision_handler = getattr(
            handler,
            "_workflow_native_vision_handler",
            None,
        )
        native_vision_enabled = getattr(
            handler,
            "_workflow_native_vision_enabled",
            None,
        )
        if callable(native_vision_handler) and callable(native_vision_enabled):
            fallback_transition = transition

            async def vision_transition(params: FunctionCallParams) -> None:
                if native_vision_enabled():
                    await native_vision_handler(params)
                    return
                await fallback_transition(params)

            transition = vision_transition
        if not getattr(handler, "_suppress_followup_llm", False):
            return transition

        async def configured_transition(params: FunctionCallParams) -> None:
            original_callback = params.result_callback

            async def result_callback(result, *, properties=None):
                if properties and properties.on_context_updated:
                    # Deterministic Workflow transitions already use run_llm=False
                    # and must retain their context-updated callback.
                    configured_properties = properties
                elif properties:
                    configured_properties = replace(properties, run_llm=False)
                else:
                    configured_properties = FunctionCallResultProperties(
                        run_llm=False
                    )
                await original_callback(
                    result,
                    properties=configured_properties,
                )

            await transition(replace(params, result_callback=result_callback))

        return configured_transition


class WorkflowBrain(BaseBrain):
    spec = BrainSpec(
        type="workflow",
        supported_runtime_modes=frozenset({"pipeline"}),
        owns_context=True,
    )

    def __init__(self, cfg_or_graph: AssistantConfig | dict[str, Any]):
        cfg = cfg_or_graph if isinstance(cfg_or_graph, AssistantConfig) else None
        graph = deepcopy(cfg.graph if cfg is not None else cfg_or_graph)
        if cfg is not None:
            # Graph v3 owns Workflow defaults. Keep older saved graphs compatible
            # by filling the new interaction settings from the assistant row.
            settings = graph.setdefault("settings", {})
            settings.setdefault("enableInterrupt", cfg.enableInterrupt)
            settings.setdefault("turnConfig", deepcopy(cfg.turnConfig))
        self._engine = WorkflowEngine(graph or {})
        if not self._engine.has_graph() or not self._engine.start_id:
            raise ValueError("WorkflowBrain 缺少有效的 Start 节点")
        self._cfg = cfg
        self._store = DynamicVariableStore.from_config(cfg or AssistantConfig(type="workflow"))
        self._tools = ToolExecutor(self._store)
        self._actions = ActionRunner(self._tools)
        self._action_stages = ActionStageRunner(self._actions)
        self._message_stages = MessageStageRunner()
        self._tool_by_id: dict[str, RuntimeTool] = {
            tool.id: tool for tool in (cfg.tools if cfg else [])
        }
        self._runtime: BrainRuntime | None = None
        self._manager: FlowManager | None = None
        self._router = WorkflowLLMRouter(cfg or AssistantConfig(type="workflow"))
        self._edge_evaluator = WorkflowEdgeEvaluator(
            self._engine,
            self._store,
            self._router_for_node,
        )
        self._state = WorkflowRuntimeState(current_node_id=self._engine.start_id)
        self._turn_lock = asyncio.Lock()
        self._output: WorkflowOutput | None = None
        self._agent_stage: WorkflowAgentStage | None = None
        self._ended = False
        self._next_message_token = 1
        self._pending_message: _MessageContinuation | None = None

    async def greeting(self, _cfg: AssistantConfig) -> str:
        """Workflow opening speech belongs to an explicit Message or Agent."""
        return ""

    def system_prompt(self, cfg: AssistantConfig) -> str:
        return self._store.render(self._engine.global_prompt())

    def build_llm(self, cfg: AssistantConfig, context: LLMContext) -> FrameProcessor:
        from services.pipecat.service_factory import create_llm

        return create_llm(cfg)

    async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None:
        if runtime.worker is None or runtime.context_aggregator is None:
            raise RuntimeError("WorkflowBrain 需要 PipelineWorker 和 context aggregator pair")
        self._cfg = cfg
        self._runtime = runtime
        self._store = DynamicVariableStore.from_config(cfg)
        self._tools = ToolExecutor(self._store, client_tools=runtime.client_tools)
        self._actions = ActionRunner(
            self._tools,
            is_session_ending=lambda: runtime.call_end.ending,
        )
        self._action_stages = ActionStageRunner(self._actions)
        self._message_stages = MessageStageRunner(runtime.client_tools)
        self._tool_by_id = {tool.id: tool for tool in cfg.tools}
        self._router = WorkflowLLMRouter(cfg)
        self._edge_evaluator = WorkflowEdgeEvaluator(
            self._engine,
            self._store,
            self._router_for_node,
        )
        self._state = WorkflowRuntimeState(current_node_id=self._engine.start_id)
        self._turn_lock = asyncio.Lock()
        self._output = WorkflowOutput(self._store, runtime)
        self._agent_stage = WorkflowAgentStage(
            cfg=cfg,
            engine=self._engine,
            store=self._store,
            runtime=runtime,
        )
        self._ended = False
        self._next_message_token = 1
        self._pending_message = None
        self._manager = ConfiguredFlowManager(
            worker=runtime.worker,
            llm=runtime.llm,
            context_aggregator=runtime.context_aggregator,
            transport=runtime.transport,
            global_functions=runtime.flow_global_functions,
        )
        self._manager.state["variables"] = self._store.values

    async def on_connected(self, *, greeting_pending: bool = False) -> None:
        self._state.enter(self._engine.start_id, WorkflowStatus.STARTING)
        await self._emit_node_active(self._engine.start_id)
        await self._emit_variables(
            reason="initialized",
            node_id=self._engine.start_id,
        )
        if self._manager is None:
            raise RuntimeError("Workflow FlowManager 尚未初始化")

        node_config = await self._initial_node_config()
        await self._activate_node_config(node_config, initialize=True)
        logger.info(f"工作流模式启用: 当前节点={self._manager.current_node}")

    async def _initial_node_config(self) -> NodeConfig:
        """Only a default-only Start advances before the first user turn."""
        outgoing = self._engine.outgoing(self._engine.start_id)
        has_condition = any(
            self._engine.edge_mode(edge) != "always" for edge in outgoing
        )
        if has_condition:
            self._state.enter(self._engine.start_id, WorkflowStatus.WAITING_USER)
            return self._passive_node_config(self._engine.start_id)
        edge = next(
            (
                candidate
                for candidate in outgoing
                if self._engine.edge_mode(candidate) == "always"
            ),
            None,
        )
        return (
            await self._follow_edge(edge)
            if edge
            else self._passive_node_config(self._engine.start_id)
        )

    async def on_client_ready(self) -> None:
        """Replay state that may have been emitted before WebRTC data was ready."""
        await self._require_output().mark_client_ready()
        current_node = (
            str(self._manager.current_node)
            if self._manager and self._manager.current_node
            else self._state.current_node_id
        )
        if current_node != self._state.current_node_id:
            self._state.current_node_id = current_node
        await self._emit_node_active(current_node, trace=False)
        await self._emit_variables(
            reason="client_ready",
            node_id=current_node,
        )

    async def on_session_update(
        self,
        dynamic_variables: dict[str, Any],
    ) -> SessionVariableUpdate:
        if self._ended:
            raise ValueError("工作流会话已经结束")
        changed = self._store.assign_declared_many(dynamic_variables)
        current = self._state.current_node_id
        if changed and current and self._engine.node_type(current) == "agent":
            await self._refresh_agent_prompt(current)
        if changed:
            await self._emit_variables(
                reason="session_update",
                node_id=current or None,
                changed=changed,
            )
        return SessionVariableUpdate(
            changed=changed,
            dynamic_variables=self._store.public_values(),
        )

    def record_user_message(self, content: str) -> None:
        if content and not self._ended:
            self._store.record("user", content)

    async def on_user_turn_end(
        self,
        content: str,
        user_message: dict[str, Any] | None = None,
    ) -> bool:
        """Route a complete user turn before the active stage may reply."""
        if not content or self._ended:
            return True
        async with self._turn_lock:
            return await self._handle_user_turn_end(
                content,
                user_message=user_message,
            )

    async def _handle_user_turn_end(
        self,
        content: str,
        *,
        user_message: dict[str, Any] | None = None,
    ) -> bool:
        """Serialized implementation so one user turn cannot transition twice."""
        manager = self._require_manager()
        current = self._state.current_node_id
        if not current:
            return True

        continuation = self._pending_message
        if (
            self._engine.node_type(current) == "message"
            and continuation is not None
            and continuation.node_id == current
        ):
            if self._message_completion_policy(current) != MESSAGE_INTERRUPTIBLE:
                # Protected Message stages keep their playback/confirmation gate.
                # Normal transports reject this input before it reaches the brain;
                # this guard also covers programmatic context injections.
                return True
            self.record_user_message(content)
            self._state.begin_user_turn(content)
            return await self._interrupt_message_continuation(
                continuation,
                content=content,
                user_message=user_message,
            )

        self.record_user_message(content)
        self._state.begin_user_turn(content)

        self._state.status = WorkflowStatus.ROUTING
        decision = await self._edge_evaluator.evaluate(
            current,
            current_user_message=user_message,
        )
        if decision.status == RouteStatus.ERROR:
            await self._require_output().emit_error(
                decision.error or "工作流路由失败",
                node_id=current,
                code="workflow_routing_error",
            )
            return await self._continue_current_node_after_no_transition(current)

        if decision.edge and manager.current_node == current:
            next_config = await self._follow_edge(
                decision.edge,
                triggering_user_text=content,
                triggering_user_message=user_message,
            )
            await self._activate_node_config(
                next_config,
                triggering_user_text=content,
            )
            return True

        return await self._continue_current_node_after_no_transition(current)

    async def _interrupt_message_continuation(
        self,
        continuation: _MessageContinuation,
        *,
        content: str,
        user_message: dict[str, Any] | None,
    ) -> bool:
        """Finish an interruptible Message once and forward the user turn."""
        if self._pending_message is not continuation:
            return True

        self._pending_message = None
        task = continuation.task
        if task is not None and task is not asyncio.current_task():
            task.cancel()

        runtime = self._require_runtime()
        if runtime.set_input_enabled is not None:
            runtime.set_input_enabled(True)
        await self._emit_trace(
            "message_interrupted",
            nodeId=continuation.node_id,
            reason="user_input",
        )

        context_messages = [
            dict(message) for message in continuation.context_messages
        ]

        if not self._engine.has_outgoing(continuation.node_id):
            self._state.enter(
                continuation.node_id,
                WorkflowStatus.WAITING_USER,
            )
            return True

        self._state.status = WorkflowStatus.ROUTING
        decision = await self._edge_evaluator.evaluate(
            continuation.node_id,
            current_user_message=user_message,
        )
        if decision.status == RouteStatus.ERROR:
            await self._require_output().emit_error(
                decision.error or "工作流路由失败",
                node_id=continuation.node_id,
                code="workflow_routing_error",
            )
            self._state.enter(
                continuation.node_id,
                WorkflowStatus.WAITING_USER,
            )
            return True

        manager = self._require_manager()
        if not decision.edge or manager.current_node != continuation.node_id:
            self._state.enter(
                continuation.node_id,
                WorkflowStatus.WAITING_USER,
            )
            return True

        next_config = await self._follow_edge(
            decision.edge,
            leading_messages=context_messages,
            triggering_user_text=content,
            triggering_user_message=user_message,
        )
        await self._activate_node_config(
            next_config,
            triggering_user_text=content,
        )
        return True

    async def _continue_current_node_after_no_transition(
        self,
        node_id: str,
    ) -> bool:
        if self._engine.node_type(node_id) != "agent":
            self._state.enter(node_id, WorkflowStatus.WAITING_USER)
            return True

        await self._refresh_agent_prompt(node_id)
        self._state.enter(node_id, WorkflowStatus.RUNNING_AGENT)
        await self._require_runtime().queue_frame(LLMRunFrame())
        return True

    async def _select_edge(
        self,
        node_id: str,
    ) -> dict | None:
        """Compatibility helper used by automatic-node traversal and tests."""
        decision = await self._edge_evaluator.evaluate(node_id)
        if decision.status == RouteStatus.ERROR:
            await self._require_output().emit_error(
                decision.error or "工作流路由失败",
                node_id=node_id,
                code="workflow_routing_error",
            )
            return None
        return decision.edge

    async def on_assistant_text_end(
        self,
        _turn_id: str,
        content: str,
        interrupted: bool,
    ) -> None:
        if not content or interrupted or self._ended:
            return
        self._store.record("agent", content, completed_agent_turn=True)
        self._state.consume_user_turn()
        if self._engine.node_type(self._state.current_node_id) == "agent":
            self._state.status = WorkflowStatus.WAITING_USER

    async def _refresh_agent_prompt(self, node_id: str) -> None:
        await self._require_agent_stage().refresh_prompt(node_id)

    def _agent_role_message(self, node_id: str) -> str:
        return self._require_agent_stage().role_message(node_id)

    def _router_for_node(self, node_id: str) -> WorkflowLLMRouter:
        if self._agent_stage is None:
            return self._router
        return self._agent_stage.router_for_node(node_id, self._router)

    async def _apply_agent_stage(self, node_id: str) -> None:
        await self._require_agent_stage().apply(node_id)

    def _agent_config(
        self,
        node_id: str,
        leading_messages: list[dict[str, Any]] | None = None,
    ) -> NodeConfig:
        stage = self._engine.agent_stage_config(node_id)
        functions: list[FlowsFunctionSchema] = []
        for tool_id in stage.tool_ids:
            tool = self._tool_by_id.get(str(tool_id))
            if tool and tool.type in {"http", "mcp", "client"}:
                functions.append(self._flow_tool(tool, node_id))
        knowledge_function = self._knowledge_function(node_id)
        if knowledge_function:
            functions.append(knowledge_function)
        if stage.vision_enabled and self._require_runtime().vision_function:
            functions.append(self._require_runtime().vision_function)
        return self._require_agent_stage().node_config(
            node_id,
            functions=functions,
            leading_messages=leading_messages,
        )

    async def _after_node_activated(
        self,
        node_config: NodeConfig,
        *,
        triggering_user_text: str = "",
    ) -> None:
        """Run the entry behavior owned by the newly active node."""
        node_id = str(node_config.get("name") or "")
        node_type = self._engine.node_type(node_id)
        if node_type == "message":
            await self._activate_message_continuation(node_id)
            return
        if node_type != "agent":
            if node_type == "start":
                self._state.enter(node_id, WorkflowStatus.WAITING_USER)
            return

        await self._emit_node_active(node_id)
        if self._agent_runs_on_entry(node_id):
            self._state.enter(node_id, WorkflowStatus.RUNNING_AGENT)
            await self._require_runtime().queue_frame(LLMRunFrame())
            return

        # The turn that selected this node belongs to the previous stage.
        # A strict wait must not leave it pending until the next user message.
        if triggering_user_text:
            self._state.consume_user_turn()
        self._state.enter(node_id, WorkflowStatus.WAITING_USER)

    def _agent_runs_on_entry(self, node_id: str) -> bool:
        """Only explicit generate mode may start a reply on node entry."""
        entry_mode = str(
            self._engine.data(node_id).get("entryMode") or "wait_user"
        )
        return entry_mode == "generate"

    async def _activate_node_config(
        self,
        node_config: NodeConfig,
        *,
        triggering_user_text: str = "",
        initialize: bool = False,
    ) -> None:
        """Install one node and dispatch its entry behavior exactly once."""
        manager = self._require_manager()
        if initialize:
            await manager.initialize(node_config)
        else:
            await manager.set_node_from_config(node_config)
        await self._after_node_activated(
            node_config,
            triggering_user_text=triggering_user_text,
        )

    async def _queue_visible_speech(
        self,
        text: str,
        *,
        source: str = "workflow-speech",
        node_id: str | None = None,
    ) -> Awaitable[None] | None:
        return await self._require_output().speak(
            text,
            source=source,
            node_id=node_id,
        )

    def _passive_node_config(
        self,
        node_id: str,
        task_messages: list[dict[str, Any]] | None = None,
    ) -> NodeConfig:
        """Keep a non-conversational terminal node active without ending the call."""
        return {
            "name": node_id,
            "role_message": self._store.render(self._engine.global_prompt()),
            "task_messages": list(task_messages or []),
            "functions": [],
            "context_strategy": ContextStrategyConfig(strategy=ContextStrategy.APPEND),
            "respond_immediately": False,
        }

    def _flow_tool(self, tool: RuntimeTool, node_id: str) -> FlowsFunctionSchema:
        properties, required = self._tools.schema_parts(tool)
        self._tools.register_secrets(tool)
        policy = policy_for_tool(tool)

        async def handler(args, _flow_manager):
            transition_id = self._state.transition_id
            try:
                result = await self._tools.execute(tool, dict(args or {}))
            except ToolExecutionError as exc:
                return {"status": "error", "message": str(exc)}
            if (
                self._state.current_node_id != node_id
                or self._state.transition_id != transition_id
            ):
                return {
                    "status": "stale",
                    "message": "工具完成时当前 Agent 已经切换，结果不再触发路由。",
                }
            updated_variables = list(result.get("updated_variables") or [])
            if updated_variables:
                await self._emit_variables(
                    reason="tool",
                    node_id=node_id,
                    changed=updated_variables,
                )
                await self._refresh_agent_prompt(node_id)
                edge = self._engine.deterministic_edge(
                    node_id,
                    self._store,
                    include_default=False,
                )
                if edge:
                    next_config = await self._follow_edge(
                        edge,
                        triggering_user_text=(
                            self._state.pending_user_turn.text
                            if self._state.pending_user_turn
                            else ""
                        ),
                    )
                    return result, self._flow_managed_transition_config(
                        next_config,
                        triggering_user_text=(
                            self._state.pending_user_turn.text
                            if self._state.pending_user_turn
                            else ""
                        ),
                    )
            return result

        if not policy.runs_llm_after_result:
            setattr(handler, "_suppress_followup_llm", True)

        return FlowsFunctionSchema(
            name=tool.function_name,
            description=tool.description or f"调用 {tool.name}",
            properties=properties,
            required=required,
            handler=handler,
            cancel_on_interruption=policy.cancel_on_interruption,
            timeout_secs=(
                float(((tool.definition or {}).get("config") or {}).get("timeout_seconds") or 3)
                if tool.type == "client" and policy.response_wait_mode == "timeout"
                else None
            ),
        )

    def _flow_managed_transition_config(
        self,
        node_config: NodeConfig,
        *,
        triggering_user_text: str,
    ) -> NodeConfig:
        """Let FlowManager finish a tool result before activating its target.

        Function-returned transitions are the one place where FlowManager must
        schedule the LLM run. Its pre-action only updates our explicit runtime
        state; it never performs a second LLM run.
        """
        node_id = str(node_config.get("name") or "")
        node_type = self._engine.node_type(node_id)
        if node_type == "message":
            configured = dict(node_config)
            configured["workflow_after_activation"] = (
                self._activate_message_continuation
            )
            return configured
        if node_type != "agent":
            return node_config
        should_run = self._agent_runs_on_entry(node_id)
        configured = dict(node_config)
        configured["respond_immediately"] = should_run
        configured["pre_actions"] = [
            {
                "type": ConfiguredFlowManager.ENTRY_ACTION_TYPE,
                "node_id": node_id,
                "should_run": should_run,
                "consume_triggering_turn": bool(triggering_user_text)
                and not should_run,
                "handler": self._activate_from_flow_transition,
            }
        ]
        return configured

    async def _activate_from_flow_transition(
        self,
        action: dict,
        _flow_manager: FlowManager,
    ) -> None:
        """Apply visible entry state without manually queueing an LLM run."""
        node_id = str(action.get("node_id") or "")
        await self._emit_node_active(node_id)
        if action.get("should_run"):
            self._state.enter(node_id, WorkflowStatus.RUNNING_AGENT)
        else:
            if action.get("consume_triggering_turn"):
                self._state.consume_user_turn()
            self._state.enter(node_id, WorkflowStatus.WAITING_USER)

    def _knowledge_function(self, node_id: str) -> FlowsFunctionSchema | None:
        stage = self._engine.agent_stage_config(node_id)
        knowledge_id = str(stage.knowledge_base_id or "")
        if not knowledge_id or stage.knowledge_mode != "on_demand":
            return None
        cfg = self._cfg or AssistantConfig(type="workflow")
        knowledge = cfg.workflow_knowledge_bases.get(knowledge_id)
        description = "在当前 Agent 绑定的知识库中检索资料。"
        if knowledge:
            description += f"知识库：{knowledge.name}。{knowledge.description}"

        async def handler(args, _flow_manager):
            query = str((args or {}).get("query") or "").strip()
            if not query:
                return {"status": "error", "message": "检索问题为空"}
            try:
                async with SessionLocal() as session:
                    results = await search_knowledge(
                        session,
                        knowledge_id,
                        query,
                        top_k=stage.knowledge_top_n,
                        score_threshold=stage.knowledge_score_threshold,
                    )
                return {"status": "ok", "results": results}
            except Exception as exc:  # noqa: BLE001 - tool errors are returned to the LLM
                logger.warning(f"Workflow 知识库检索失败:{exc}")
                return {"status": "error", "message": "知识库检索暂时不可用"}

        return FlowsFunctionSchema(
            name="search_knowledge_base",
            description=description,
            properties={
                "query": {"type": "string", "description": "完整问题或检索关键词"}
            },
            required=["query"],
            handler=handler,
            cancel_on_interruption=True,
        )

    async def _follow_edge(
        self,
        edge: dict,
        *,
        leading_messages: list[dict[str, Any]] | None = None,
        triggering_user_text: str = "",
        triggering_user_message: dict[str, Any] | None = None,
    ) -> NodeConfig:
        await self._begin_edge_transition(edge)
        return await self._resolve_path(
            str(edge.get("target") or ""),
            leading_messages=list(leading_messages or []),
            triggering_user_text=triggering_user_text,
            triggering_user_message=triggering_user_message,
        )

    async def _resolve_path(
        self,
        node_id: str,
        *,
        leading_messages: list[dict[str, Any]] | None = None,
        triggering_user_text: str = "",
        triggering_user_message: dict[str, Any] | None = None,
    ) -> NodeConfig:
        context_messages = list(leading_messages or [])
        for hop in range(MAX_AUTOMATIC_HOPS):
            self._state.automatic_hops = hop
            node_type = self._engine.node_type(node_id)
            if node_type == "agent":
                await self._apply_agent_stage(node_id)
                agent_messages = context_messages
                if (
                    triggering_user_text
                    and self._engine.data(node_id).get("contextPolicy") == "fresh"
                ):
                    current_user_message = (
                        deepcopy(triggering_user_message)
                        if triggering_user_message
                        else {"role": "user", "content": triggering_user_text}
                    )
                    # Fixed speech is represented by system facts. Put those
                    # facts before the triggering user turn so RESET contexts
                    # still end with the real user input rather than a control
                    # message.
                    agent_messages = [*context_messages, current_user_message]
                return self._agent_config(node_id, agent_messages)
            if node_type == "end":
                await self._enter_end(node_id)
                return self._passive_node_config(node_id, context_messages)
            if node_type == "action":
                outcome = await self._enter_action(node_id)
                if not outcome.should_route:
                    return self._passive_node_config(node_id, context_messages)
            elif node_type == "message":
                self._prepare_message_continuation(
                    node_id,
                    context_messages=context_messages,
                    triggering_user_text=triggering_user_text,
                    triggering_user_message=triggering_user_message,
                )
                return self._passive_node_config(node_id, context_messages)
            elif node_type == "handoff":
                await self._enter_handoff(node_id)
            elif node_type == "start":
                self._state.enter(node_id, WorkflowStatus.WAITING_USER)
                await self._emit_node_active(node_id)
            else:
                raise RuntimeError(f"工作流指向未知节点:{node_id}")
            if not self._engine.has_outgoing(node_id):
                return self._passive_node_config(node_id, context_messages)
            edge = await self._select_edge(node_id)
            if not edge:
                return self._passive_node_config(node_id, context_messages)
            await self._begin_edge_transition(edge)
            node_id = str(edge.get("target") or "")
        raise RuntimeError("工作流连续自动跳转超过安全上限")

    def _prepare_message_continuation(
        self,
        node_id: str,
        *,
        context_messages: list[dict[str, Any]],
        triggering_user_text: str,
        triggering_user_message: dict[str, Any] | None,
    ) -> None:
        """Save the path state without waiting inside the pipeline call stack."""
        token = self._next_message_token
        self._next_message_token += 1
        self._pending_message = _MessageContinuation(
            token=token,
            node_id=node_id,
            context_messages=[dict(message) for message in context_messages],
            triggering_user_text=triggering_user_text,
            triggering_user_message=deepcopy(triggering_user_message),
        )
        self._state.enter(node_id, WorkflowStatus.RUNNING_MESSAGE)

    async def _activate_message_continuation(self, node_id: str) -> None:
        """Start a prepared Message only after FlowManager made it active."""
        continuation = self._pending_message
        if (
            continuation is None
            or continuation.node_id != node_id
            or continuation.task is not None
        ):
            return
        self._state.enter(node_id, WorkflowStatus.RUNNING_MESSAGE)
        await self._emit_node_active(node_id)
        runtime = self._require_runtime()
        if runtime.set_input_enabled is not None:
            runtime.set_input_enabled(
                self._message_completion_policy(node_id) == MESSAGE_INTERRUPTIBLE
            )
        continuation.task = asyncio.create_task(
            self._complete_message_continuation(continuation),
            name=f"workflow-message-{node_id}-{continuation.token}",
        )

    async def _complete_message_continuation(
        self,
        continuation: _MessageContinuation,
    ) -> None:
        """Wait for playback/confirmation outside routing, then resume once."""
        try:
            result = await self._enter_message(
                continuation.node_id,
                already_active=True,
                input_already_blocked=True,
            )
            if not result.succeeded:
                if self._pending_message is continuation:
                    self._pending_message = None
                return
            await self._resume_message_continuation(continuation, result)
        except asyncio.CancelledError:
            raise
        except Exception as exc:  # noqa: BLE001 - keep callback failures visible
            logger.exception(f"Message 节点续跑失败:{exc}")
            if self._pending_message is continuation:
                self._pending_message = None
                self._state.enter(
                    continuation.node_id,
                    WorkflowStatus.WAITING_USER,
                )
            runtime = self._require_runtime()
            if runtime.set_input_enabled is not None:
                runtime.set_input_enabled(True)
            await self._require_output().emit_error(
                "Message 节点完成后无法继续工作流",
                node_id=continuation.node_id,
                code="workflow_message_resume_error",
            )

    async def _resume_message_continuation(
        self,
        continuation: _MessageContinuation,
        result: MessageStageResult,
    ) -> None:
        """Advance from the completed Message without blocking media frames."""
        async with self._turn_lock:
            manager = self._require_manager()
            if (
                self._ended
                or self._pending_message is not continuation
                or self._state.current_node_id != continuation.node_id
                or str(manager.current_node or "") != continuation.node_id
            ):
                return

            self._pending_message = None
            context_messages = [
                dict(message) for message in continuation.context_messages
            ]
            if result.speech:
                context_message = fixed_speech_context_message(result.speech)
                if context_message is not None:
                    context_messages.append(context_message)

            if not self._engine.has_outgoing(continuation.node_id):
                self._state.enter(
                    continuation.node_id,
                    WorkflowStatus.WAITING_USER,
                )
                return
            edge = await self._select_edge(continuation.node_id)
            if not edge:
                self._state.enter(
                    continuation.node_id,
                    WorkflowStatus.WAITING_USER,
                )
                return

            next_config = await self._follow_edge(
                edge,
                leading_messages=context_messages,
                triggering_user_text=continuation.triggering_user_text,
                triggering_user_message=continuation.triggering_user_message,
            )
            await self._activate_node_config(
                next_config,
                triggering_user_text=continuation.triggering_user_text,
            )

    async def _enter_action(self, node_id: str) -> ActionOutcome:
        self._state.enter(node_id, WorkflowStatus.RUNNING_ACTION)
        await self._emit_node_active(node_id)
        data = self._engine.data(node_id)
        runtime = self._require_runtime()
        invocation_id = self._actions.new_invocation_id()
        tool_id = str(data.get("toolId") or "")
        tool = self._tool_by_id.get(tool_id)
        try:
            stage_result = await self._action_stages.run(
                ActionStageSpec(
                    actions=(
                        StageAction(
                            id=node_id,
                            tool=tool,
                            arguments=data.get("arguments") or {},
                            result_assignments=self._action_result_assignments(data),
                            invocation_id=invocation_id,
                        ),
                    ),
                    input_policy=(
                        "block"
                        if data.get("userInputPolicy") == "block"
                        else "queue"
                    ),
                ),
                set_input_enabled=runtime.set_input_enabled,
                on_started=lambda: self._emit_trace(
                    "action_started",
                    nodeId=node_id,
                    invocationId=invocation_id,
                    toolId=tool_id,
                    toolType=tool.type if tool else None,
                ),
            )
            outcome = stage_result.outcomes[0]
            updated_variables = list(outcome.updated_variables)
            if updated_variables:
                await self._emit_variables(
                    reason="action",
                    node_id=node_id,
                    changed=updated_variables,
                )
        except ActionInvocationCancelled as exc:
            outcome = exc.outcome
            self._set_last_action(outcome)
            await self._emit_action_outcome(node_id, outcome)
            raise
        self._set_last_action(outcome)
        await self._emit_action_outcome(node_id, outcome)
        return outcome

    async def _enter_message(
        self,
        node_id: str,
        *,
        already_active: bool = False,
        input_already_blocked: bool = False,
    ) -> MessageStageResult:
        if not already_active:
            self._state.enter(node_id, WorkflowStatus.RUNNING_MESSAGE)
            await self._emit_node_active(node_id)
        data = self._engine.data(node_id)
        runtime = self._require_runtime()
        speech = self._store.render(str(data.get("speech") or "")).strip()
        completion_policy = self._message_completion_policy(node_id)
        require_confirmation = completion_policy == MESSAGE_CONFIRMATION
        display = (
            MessageDisplaySpec(
                title=self._store.render(
                    str(data.get("title") or "重要提示")
                ).strip(),
                message=self._store.render(
                    str(data.get("message") or "")
                ).strip(),
                confirm_label=self._store.render(
                    str(data.get("confirmLabel") or "确认")
                ).strip(),
            )
            if require_confirmation
            else None
        )
        result = await self._message_stages.run(
            MessageStageSpec(
                speech=speech,
                display=display,
                completion_policy=completion_policy,
            ),
            speak=lambda content: self._queue_visible_speech(
                content,
                source="workflow-message-speech",
                node_id=node_id,
            ),
            set_input_enabled=runtime.set_input_enabled,
            input_already_blocked=input_already_blocked,
            on_started=lambda: self._emit_trace(
                "message_started",
                nodeId=node_id,
                hasSpeech=bool(speech),
                showsMessage=require_confirmation,
                completionPolicy=completion_policy,
                requiresConfirmation=require_confirmation,
            ),
        )
        if result.succeeded:
            if (
                completion_policy == MESSAGE_CONFIRMATION
                and result.action == "confirmed"
            ):
                if runtime.wait_for_output_stopped is None:
                    raise RuntimeError("当前管线不支持等待 Message 确认语音停止")
                # ClientToolBroker has already broadcast the interruption
                # before resolving the confirmation. Keep the next node behind
                # the pipeline and transport playback boundaries.
                await runtime.wait_for_output_stopped()
            await self._emit_trace(
                "message_completed",
                nodeId=node_id,
                action=result.action,
            )
            return result

        self._state.enter(node_id, WorkflowStatus.WAITING_USER)
        await self._emit_trace(
            "message_failed",
            nodeId=node_id,
            error=result.error or "Message 节点执行失败",
        )
        await self._require_output().emit_error(
            result.error or "Message 节点执行失败",
            node_id=node_id,
            code="workflow_message_error",
        )
        return result

    def _message_completion_policy(self, node_id: str) -> str:
        value = str(
            self._engine.data(node_id).get("completionPolicy")
            or MESSAGE_PLAYBACK
        )
        return value if value in MESSAGE_COMPLETION_POLICIES else MESSAGE_PLAYBACK

    def _set_last_action(self, outcome: ActionOutcome) -> None:
        legacy_status = {
            ActionStatus.SUCCESS: "ok",
            ActionStatus.FAILURE: "error",
            ActionStatus.CANCELLED: "cancelled",
        }[outcome.status]
        self._store.values.update(
            {
                "system__last_action_status": legacy_status,
                "system__last_action_invocation_id": outcome.invocation_id,
                "system__last_action_duration_ms": outcome.duration_ms,
                "system__last_action_error_code": (
                    outcome.error.code if outcome.error else ""
                ),
                "system__last_action_error": (
                    outcome.error.message if outcome.error else ""
                ),
            }
        )

    async def _emit_action_outcome(
        self,
        node_id: str,
        outcome: ActionOutcome,
    ) -> None:
        event = {
            ActionStatus.SUCCESS: "action_completed",
            ActionStatus.FAILURE: "action_failed",
            ActionStatus.CANCELLED: "action_cancelled",
        }[outcome.status]
        await self._emit_trace(event, nodeId=node_id, outcome=outcome.trace_payload())

    @staticmethod
    def _action_result_assignments(
        data: dict[str, Any],
    ) -> dict[str, str] | None:
        """Resolve node mapping semantics for ToolExecutor.

        ``None`` means inherit the reusable tool's mapping, while an empty
        dictionary explicitly disables all result assignments.
        """
        mode = str(data.get("resultAssignmentMode") or "none")
        if mode == "inherit":
            return None
        if mode == "override":
            assignments = data.get("resultAssignments")
            return dict(assignments) if isinstance(assignments, dict) else {}
        return {}

    async def _enter_handoff(self, node_id: str) -> None:
        self._state.enter(node_id, WorkflowStatus.HANDOFF)
        await self._emit_node_active(node_id)
        data = self._engine.data(node_id)
        message = self._store.render(str(data.get("message") or ""))
        await self._require_runtime().queue_frame(
            OutputTransportMessageUrgentFrame(
                message={
                    "type": "handoff-requested",
                    "nodeId": node_id,
                    "targetType": data.get("targetType", "human"),
                    "target": data.get("target", ""),
                    "message": message,
                }
            )
        )
        if message:
            await self._queue_visible_speech(message)
        self._store.values["system__handoff_status"] = "requested"

    async def _enter_end(self, node_id: str) -> None:
        self._ended = True
        self._state.enter(node_id, WorkflowStatus.ENDED)
        self._state.finish()
        await self._emit_node_active(node_id)
        runtime = self._require_runtime()
        if runtime.set_knowledge_scope:
            runtime.set_knowledge_scope({"mode": "disabled"})
        if runtime.set_vision_scope:
            runtime.set_vision_scope({"enabled": False})
        if runtime.set_input_enabled:
            runtime.set_input_enabled(False)
        data = self._engine.data(node_id)
        message = self._store.render(str(data.get("message") or ""))
        scope = str(data.get("scope") or "session")
        await self._emit_trace(
            "workflow_ended",
            nodeId=node_id,
            scope=scope,
            outcome="success",
        )
        if scope == "flow":
            await runtime.queue_frame(
                OutputTransportMessageUrgentFrame(
                    message={"type": "flow-ended", "nodeId": node_id}
                )
            )
            if message:
                await self._queue_visible_speech(message)
            return
        runtime.call_end.begin("workflow_completed")
        if message:
            await self._queue_visible_speech(message)
        arm_tracked = getattr(runtime.call_end, "arm_after_tracked_speech", None)
        if callable(arm_tracked):
            await arm_tracked()
        elif message:
            runtime.call_end.arm_after_speech()
        else:
            await runtime.call_end.finish()

    async def _emit_node_active(
        self,
        node_id: str | None,
        *,
        trace: bool = True,
    ) -> None:
        await self._require_output().emit_node_active(node_id)
        if trace and node_id:
            await self._emit_trace(
                "node_entered",
                nodeId=node_id,
                nodeType=self._engine.node_type(node_id),
            )

    async def _begin_edge_transition(self, edge: dict) -> None:
        transition_id = self._state.begin_transition()
        source_id = str(edge.get("source") or self._state.current_node_id or "")
        target_id = str(edge.get("target") or "")
        if source_id:
            await self._emit_trace("node_exited", nodeId=source_id)
        await self._emit_trace(
            "edge_selected",
            edgeId=str(edge.get("id") or ""),
            sourceNodeId=source_id,
            targetNodeId=target_id,
            edgeMode=self._engine.edge_mode(edge),
            transitionId=transition_id,
        )

    async def _emit_trace(self, event: str, **details: Any) -> None:
        transition_id = int(details.pop("transitionId", self._state.transition_id))
        try:
            await self._require_output().emit_trace(
                event,
                revision=self._engine.revision,
                transition_id=transition_id,
                **details,
            )
        except Exception as exc:  # noqa: BLE001 - trace must not alter execution
            logger.warning(f"发送 Workflow 轨迹失败，不影响当前流程: {exc}")

    async def _emit_variables(
        self,
        *,
        reason: str,
        node_id: str | None,
        changed: list[str] | None = None,
    ) -> None:
        """Publish a safe snapshot so Workflow debug mirrors runtime state."""
        await self._require_output().emit_variables(
            reason=reason,
            node_id=node_id,
            changed=changed,
        )
        public_variables = self._require_output().public_variables()
        public_names = [
            name
            for name in (changed or public_variables.keys())
            if not name.startswith(("system__", "secret__"))
        ]
        await self._emit_trace(
            "variables_updated" if changed else "variables_snapshot",
            nodeId=node_id,
            reason=reason,
            variableNames=public_names,
            variables=public_variables,
        )

    def _require_runtime(self) -> BrainRuntime:
        if self._runtime is None:
            raise RuntimeError("WorkflowBrain 尚未绑定 pipeline runtime")
        return self._runtime

    def _require_manager(self) -> FlowManager:
        if self._manager is None:
            raise RuntimeError("Workflow FlowManager 尚未初始化")
        return self._manager

    def _require_output(self) -> WorkflowOutput:
        if self._output is None:
            # A few focused unit tests bind BrainRuntime directly. Lazily create
            # the output adapter so those tests exercise the same production path.
            self._output = WorkflowOutput(self._store, self._require_runtime())
        return self._output

    def _require_agent_stage(self) -> WorkflowAgentStage:
        if self._agent_stage is None:
            self._agent_stage = WorkflowAgentStage(
                cfg=self._cfg or AssistantConfig(type="workflow"),
                engine=self._engine,
                store=self._store,
                runtime=self._require_runtime(),
            )
        return self._agent_stage