feat: add workflow action outcomes and tracing

backend/services/brains/base.py

@@ -83,6 +83,7 @@ class BrainRuntime:
     set_system_prompt: Callable[[str], None]
     set_tools: Callable[[list[FunctionSchema] | None], None]
     call_end: CallEndPort
+    session_id: str = ""
     client_tools: ClientToolPort | None = None
     worker: Any = None
     context_aggregator: Any = None

backend/services/brains/workflow_brain.py

@@ -5,7 +5,9 @@ from __future__ import annotations
 import asyncio
 from copy import deepcopy
 from dataclasses import replace
+from time import monotonic
 from typing import Any
+from uuid import uuid4
 
 from loguru import logger
 from models import AssistantConfig, RuntimeTool
@@ -40,6 +42,9 @@ 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 (
+    ActionError,
+    ActionOutcome,
+    ActionStatus,
     RouteStatus,
     WorkflowRuntimeState,
     WorkflowStatus,
@@ -53,6 +58,23 @@ from services.workflow_router import WorkflowLLMRouter
 MAX_AUTOMATIC_HOPS = 50
 
 
+class _ActionFailure(RuntimeError):
+    """Internal adapter from heterogeneous tool failures to ActionOutcome."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        code: str,
+        retryable: bool = False,
+        result: dict[str, Any] | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.code = code
+        self.retryable = retryable
+        self.result = result
+
+
 class ConfiguredFlowManager(FlowManager):
     """Preserve Flow transitions while suppressing late async-tool replies."""
 
@@ -257,7 +279,7 @@ class WorkflowBrain(BaseBrain):
         )
         if current_node != self._state.current_node_id:
             self._state.current_node_id = current_node
-        await self._emit_node_active(current_node)
+        await self._emit_node_active(current_node, trace=False)
         await self._emit_variables(
             reason="client_ready",
             node_id=current_node,
@@ -643,7 +665,7 @@ class WorkflowBrain(BaseBrain):
         *,
         triggering_user_text: str = "",
     ) -> NodeConfig:
-        self._state.begin_transition()
+        await self._begin_edge_transition(edge)
         leading_messages: list[dict[str, str]] = []
         speech = self._engine.edge_transition_speech(edge)
         if speech:
@@ -690,7 +712,9 @@ class WorkflowBrain(BaseBrain):
                 await self._enter_end(node_id)
                 return self._passive_node_config(node_id, context_messages)
             if node_type == "action":
-                await self._enter_action(node_id)
+                outcome = await self._enter_action(node_id)
+                if not outcome.should_route:
+                    return self._passive_node_config(node_id, context_messages)
             elif node_type == "handoff":
                 await self._enter_handoff(node_id)
             elif node_type == "start":
@@ -703,7 +727,7 @@ class WorkflowBrain(BaseBrain):
             edge = await self._select_edge(node_id)
             if not edge:
                 return self._passive_node_config(node_id, context_messages)
-            self._state.begin_transition()
+            await self._begin_edge_transition(edge)
             speech = self._engine.edge_transition_speech(edge)
             if speech:
                 content = self._store.render(speech).strip()
@@ -720,11 +744,13 @@ class WorkflowBrain(BaseBrain):
             node_id = str(edge.get("target") or "")
         raise RuntimeError("工作流连续自动跳转超过安全上限")
 
-    async def _enter_action(self, node_id: str) -> None:
+    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 = f"act_{uuid4().hex[:20]}"
+        started_at = monotonic()
         block_user_input = data.get("userInputPolicy") == "block"
         if block_user_input and runtime.set_input_enabled:
             # Blocking only suppresses new audio/text input while the Action
@@ -734,9 +760,20 @@ class WorkflowBrain(BaseBrain):
             runtime.set_input_enabled(False)
         tool_id = str(data.get("toolId") or "")
         tool = self._tool_by_id.get(tool_id)
+        result: dict[str, Any] | None = None
         try:
+            await self._emit_trace(
+                "action_started",
+                nodeId=node_id,
+                invocationId=invocation_id,
+                toolId=tool_id,
+                toolType=tool.type if tool else None,
+            )
             if not tool:
-                raise ToolExecutionError(f"工具不存在:{tool_id}")
+                raise _ActionFailure(
+                    f"工具不存在:{tool_id}",
+                    code="tool_not_found",
+                )
             arguments = self._store.render_data(data.get("arguments") or {})
             result = await self._tools.execute(
                 tool,
@@ -744,8 +781,12 @@ class WorkflowBrain(BaseBrain):
                 result_assignments=self._action_result_assignments(data),
             )
             if result.get("status") != "ok":
-                raise ToolExecutionError(
-                    str(result.get("message") or "工具返回执行失败状态")
+                returned_status = str(result.get("status") or "error")
+                raise _ActionFailure(
+                    str(result.get("message") or "工具返回执行失败状态"),
+                    code=str(result.get("code") or f"tool_{returned_status}"),
+                    retryable=bool(result.get("retryable", False)),
+                    result=result,
                 )
             updated_variables = list(result.get("updated_variables") or [])
             if updated_variables:
@@ -754,14 +795,114 @@ class WorkflowBrain(BaseBrain):
                     node_id=node_id,
                     changed=updated_variables,
                 )
-            self._store.values["system__last_action_status"] = "ok"
-            self._store.values["system__last_action_error"] = ""
-        except (ToolExecutionError, ValueError) as exc:
-            self._store.values["system__last_action_status"] = "error"
-            self._store.values["system__last_action_error"] = str(exc)[:2048]
+            outcome = ActionOutcome(
+                invocation_id=invocation_id,
+                status=ActionStatus.SUCCESS,
+                duration_ms=self._elapsed_ms(started_at),
+                result=result,
+                updated_variables=tuple(str(name) for name in updated_variables),
+            )
+        except asyncio.CancelledError:
+            outcome = ActionOutcome(
+                invocation_id=invocation_id,
+                status=ActionStatus.CANCELLED,
+                duration_ms=self._elapsed_ms(started_at),
+                result=result,
+                error=ActionError(
+                    code="action_cancelled",
+                    message="Action 随当前任务取消",
+                ),
+            )
+            self._set_last_action(outcome)
+            await self._emit_action_outcome(node_id, outcome)
+            raise
+        except (_ActionFailure, ToolExecutionError, ValueError) as exc:
+            cancelled = self._action_was_cancelled(exc, runtime)
+            outcome = ActionOutcome(
+                invocation_id=invocation_id,
+                status=(
+                    ActionStatus.CANCELLED if cancelled else ActionStatus.FAILURE
+                ),
+                duration_ms=self._elapsed_ms(started_at),
+                result=(exc.result if isinstance(exc, _ActionFailure) else result),
+                error=ActionError(
+                    code=(
+                        "session_ended"
+                        if cancelled
+                        else (
+                            exc.code
+                            if isinstance(exc, _ActionFailure)
+                            else (
+                                "invalid_action_configuration"
+                                if isinstance(exc, ValueError)
+                                else "tool_execution_error"
+                            )
+                        )
+                    ),
+                    message=str(exc)[:2048],
+                    retryable=(
+                        exc.retryable if isinstance(exc, _ActionFailure) else False
+                    ),
+                ),
+            )
         finally:
             if block_user_input and runtime.set_input_enabled:
                 runtime.set_input_enabled(True)
+        self._set_last_action(outcome)
+        await self._emit_action_outcome(node_id, outcome)
+        return outcome
+
+    @staticmethod
+    def _elapsed_ms(started_at: float) -> int:
+        return max(0, round((monotonic() - started_at) * 1000))
+
+    @staticmethod
+    def _action_was_cancelled(exc: Exception, runtime: BrainRuntime) -> bool:
+        if getattr(runtime.call_end, "ending", False):
+            return True
+        message = str(exc)
+        return any(
+            marker in message
+            for marker in (
+                "会话已结束",
+                "会话已取消",
+                "管线已停止",
+                "通道已关闭",
+                "连接已断开",
+            )
+        )
+
+    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(
@@ -815,6 +956,12 @@ class WorkflowBrain(BaseBrain):
         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(
@@ -835,8 +982,46 @@ class WorkflowBrain(BaseBrain):
         else:
             await runtime.call_end.finish()
 
-    async def _emit_node_active(self, node_id: str | None) -> None:
+    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,
@@ -851,6 +1036,19 @@ class WorkflowBrain(BaseBrain):
             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:

backend/services/conversation_history.py

@@ -5,6 +5,7 @@ conversation_artifacts，并把对象存储地址关联到会话或消息。
 """
 
 import asyncio
+from copy import deepcopy
 from datetime import UTC, datetime
 from uuid import uuid4
 
@@ -13,6 +14,9 @@ from db.session import SessionLocal
 from loguru import logger
 
 
+MAX_WORKFLOW_TRACE_EVENTS = 1000
+
+
 def _parse_timestamp(value: object) -> datetime:
     if not isinstance(value, str) or not value:
         return datetime.now(UTC)
@@ -29,6 +33,7 @@ class ConversationRecorder:
     def __init__(self, session_id: str):
         self.session_id = session_id
         self._sequence = 0
+        self._trace_sequence = 0
         self._lock = asyncio.Lock()
         self._seen_events: set[str] = set()
 
@@ -41,6 +46,7 @@ class ConversationRecorder:
         channel: str,
         runtime_mode: str,
         session_id: str | None = None,
+        extra: dict | None = None,
     ) -> "ConversationRecorder | None":
         session_id = session_id or f"conv_{uuid4().hex[:20]}"
         try:
@@ -54,7 +60,7 @@ class ConversationRecorder:
                         runtime_mode=runtime_mode,
                         status="active",
                         message_count=0,
-                        extra={},
+                        extra=deepcopy(extra or {}),
                     )
                 )
                 await db.commit()
@@ -67,6 +73,9 @@ class ConversationRecorder:
         if not isinstance(message, dict):
             return
         event_type = message.get("type")
+        if event_type == "workflow-event":
+            await self._append_workflow_event(message)
+            return
         role = ""
         content = ""
         extra: dict = {}
@@ -98,6 +107,36 @@ class ConversationRecorder:
         self._seen_events.add(event_key)
         await self._append(role, content, timestamp, extra)
 
+    async def _append_workflow_event(self, message: dict) -> None:
+        """Persist bounded control-plane history without creating chat rows."""
+
+        event_id = str(message.get("eventId") or "")
+        event_name = str(message.get("event") or "")
+        event_key = f"workflow:{event_id}"
+        if not event_id or not event_name or event_key in self._seen_events:
+            return
+        self._seen_events.add(event_key)
+
+        async with self._lock:
+            next_sequence = self._trace_sequence + 1
+            payload = deepcopy(message)
+            payload["sessionId"] = self.session_id
+            payload["sequence"] = next_sequence
+            try:
+                async with SessionLocal() as db:
+                    conversation = await db.get(ConversationSession, self.session_id)
+                    if not conversation:
+                        return
+                    extra = dict(conversation.extra or {})
+                    trace = list(extra.get("workflowTrace") or [])
+                    trace.append(payload)
+                    extra["workflowTrace"] = trace[-MAX_WORKFLOW_TRACE_EVENTS:]
+                    conversation.extra = extra
+                    await db.commit()
+                self._trace_sequence = next_sequence
+            except Exception as exc:
+                logger.error(f"保存 Workflow 轨迹失败，不影响本次通话: {exc}")
+
     async def _append(
         self,
         role: str,
@@ -130,13 +169,14 @@ class ConversationRecorder:
                 logger.error(f"保存对话文本失败，不影响本次通话: {exc}")
 
     async def finish(self, *, status: str = "completed") -> None:
-        try:
-            async with SessionLocal() as db:
-                conversation = await db.get(ConversationSession, self.session_id)
-                if conversation:
-                    conversation.status = status
-                    conversation.ended_at = datetime.now(UTC)
-                    conversation.message_count = self._sequence
-                    await db.commit()
-        except Exception as exc:
-            logger.error(f"结束对话历史会话失败: {exc}")
+        async with self._lock:
+            try:
+                async with SessionLocal() as db:
+                    conversation = await db.get(ConversationSession, self.session_id)
+                    if conversation:
+                        conversation.status = status
+                        conversation.ended_at = datetime.now(UTC)
+                        conversation.message_count = self._sequence
+                        await db.commit()
+            except Exception as exc:
+                logger.error(f"结束对话历史会话失败: {exc}")

backend/services/pipecat/pipeline.py

@@ -208,8 +208,9 @@ async def run_pipeline(
                    只要有 .input() / .output() / event_handler 即可。
         cfg:       助手配置(随请求内联传入)。
     """
-    if cfg.type == "workflow":
-        vision_enabled = WorkflowEngine(cfg.graph).uses_vision()
+    workflow_engine = WorkflowEngine(cfg.graph) if cfg.type == "workflow" else None
+    if workflow_engine:
+        vision_enabled = workflow_engine.uses_vision()
 
     logger.info(
         f"启动管线: assistant={cfg.name} type={cfg.type} "
@@ -578,6 +579,7 @@ async def run_pipeline(
         channel=channel,
         runtime_mode=cfg.runtimeMode,
         session_id=cfg.conversation_id or None,
+        extra=(workflow_engine.session_metadata() if workflow_engine else None),
     )
     pipeline = Pipeline(
         [
@@ -664,6 +666,7 @@ async def run_pipeline(
             set_system_prompt=set_system_prompt,
             set_tools=set_visible_tools,
             call_end=call_end,
+            session_id=recorder.session_id if recorder else "",
             client_tools=client_tools,
             worker=worker,
             context_aggregator=WorkflowAggregatorPair(

backend/services/workflow/models.py

@@ -27,6 +27,65 @@ class RouteStatus(StrEnum):
     ERROR = "error"
 
 
+class ActionStatus(StrEnum):
+    """Stable Action outcomes used by routing and future debug tooling."""
+
+    SUCCESS = "success"
+    FAILURE = "failure"
+    CANCELLED = "cancelled"
+
+
+@dataclass(frozen=True)
+class ActionError:
+    """Machine-readable failure details without losing the operator message."""
+
+    code: str
+    message: str
+    retryable: bool = False
+
+
+@dataclass(frozen=True)
+class ActionOutcome:
+    """One completed Action invocation.
+
+    ``result`` remains an in-memory value because a tool response may contain
+    private business data. Trace events publish only its shape and variable
+    names, never the raw response.
+    """
+
+    invocation_id: str
+    status: ActionStatus
+    duration_ms: int
+    result: dict[str, Any] | None = None
+    updated_variables: tuple[str, ...] = ()
+    error: ActionError | None = None
+
+    @property
+    def should_route(self) -> bool:
+        """Cancellation is a lifecycle outcome, not a failure branch."""
+
+        return self.status != ActionStatus.CANCELLED
+
+    def trace_payload(self) -> dict[str, Any]:
+        """Return a persistence-safe summary of the execution result."""
+
+        payload: dict[str, Any] = {
+            "invocationId": self.invocation_id,
+            "status": self.status.value,
+            "durationMs": self.duration_ms,
+            "updatedVariables": list(self.updated_variables),
+        }
+        if self.result is not None:
+            payload["resultKeys"] = sorted(str(key) for key in self.result)
+        if self.error is not None:
+            payload["error"] = {
+                "code": self.error.code,
+                "message": self.error.message,
+                "retryable": self.error.retryable,
+            }
+        return payload
+
+
 @dataclass(frozen=True)
 class UserTurn:
     """One committed user turn that may cross automatic Workflow nodes."""
@@ -89,4 +148,3 @@ class EdgeEvaluation:
     status: RouteStatus
     edge: dict[str, Any] | None = None
     error: str | None = None
-

backend/services/workflow/output.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from typing import Any
+from uuid import uuid4
 
 from pipecat.frames.frames import OutputTransportMessageUrgentFrame, TTSSpeakFrame
 from pipecat.utils.time import time_now_iso8601
@@ -67,6 +68,29 @@ class WorkflowOutput:
         if node_id:
             await self.emit({"type": "node-active", "nodeId": node_id})
 
+    async def emit_trace(
+        self,
+        event: str,
+        *,
+        revision: str,
+        transition_id: int,
+        **details: Any,
+    ) -> None:
+        """Publish one ordered, machine-readable Workflow runtime event."""
+
+        await self.emit(
+            {
+                "type": "workflow-event",
+                "eventId": f"wfe_{uuid4().hex[:20]}",
+                "event": event,
+                "timestamp": time_now_iso8601(),
+                "sessionId": self._runtime.session_id,
+                "workflowRevision": revision,
+                "transitionId": transition_id,
+                **details,
+            }
+        )
+
     async def emit_variables(
         self,
         *,

backend/services/workflow_engine.py

@@ -2,8 +2,11 @@
 
 from __future__ import annotations
 
+import json
 import re
+from copy import deepcopy
 from dataclasses import dataclass
+from hashlib import sha256
 from typing import Any
 
 from services.node_specs import normalize_graph
@@ -32,6 +35,7 @@ class AgentStageConfig:
 class WorkflowEngine:
     def __init__(self, graph: dict[str, Any]):
         self.graph = normalize_graph(graph)
+        self.revision = self._revision_for(self.graph)
         self.settings = self.graph.get("settings") or {}
         self.nodes: dict[str, dict] = {
             str(node["id"]): node
@@ -48,6 +52,29 @@ class WorkflowEngine:
             None,
         )
 
+    @staticmethod
+    def _revision_for(normalized_graph: dict[str, Any]) -> str:
+        """Hash the exact normalized graph used by this runtime."""
+
+        canonical = json.dumps(
+            normalized_graph,
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+        ).encode("utf-8")
+        return f"sha256:{sha256(canonical).hexdigest()}"
+
+    def session_metadata(self) -> dict[str, Any]:
+        """Pin a conversation to an immutable, inspectable graph snapshot."""
+
+        return {
+            "workflow": {
+                "revision": self.revision,
+                "specVersion": self.graph.get("specVersion"),
+                "snapshot": deepcopy(self.graph),
+            }
+        }
+
     def has_graph(self) -> bool:
         return bool(self.start_id)