diff --git a/backend/routes/conversations.py b/backend/routes/conversations.py index b5f41c4..0c15413 100644 --- a/backend/routes/conversations.py +++ b/backend/routes/conversations.py @@ -102,6 +102,7 @@ async def get_conversation( ).scalars().all() return ConversationDetailOut( **_session_out(conversation).model_dump(), + extra=conversation.extra or {}, messages=[ ConversationMessageOut( id=message.id, diff --git a/backend/schemas.py b/backend/schemas.py index 109bed1..b94c0a1 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -422,6 +422,7 @@ class ConversationOut(CamelModel): class ConversationDetailOut(ConversationOut): + extra: dict[str, Any] = Field(default_factory=dict) messages: list[ConversationMessageOut] = Field(default_factory=list) diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index 9bf360b..702195c 100644 --- a/backend/services/brains/base.py +++ b/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 diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index 25401c1..142312e 100644 --- a/backend/services/brains/workflow_brain.py +++ b/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: diff --git a/backend/services/conversation_history.py b/backend/services/conversation_history.py index 9139274..96fb755 100644 --- a/backend/services/conversation_history.py +++ b/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}") diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index b9878d5..a91e553 100644 --- a/backend/services/pipecat/pipeline.py +++ b/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( diff --git a/backend/services/workflow/models.py b/backend/services/workflow/models.py index 89d0fdf..7661fa8 100644 --- a/backend/services/workflow/models.py +++ b/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 - diff --git a/backend/services/workflow/output.py b/backend/services/workflow/output.py index 15e4361..b3d5994 100644 --- a/backend/services/workflow/output.py +++ b/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, *, diff --git a/backend/services/workflow_engine.py b/backend/services/workflow_engine.py index 208968d..fe33641 100644 --- a/backend/services/workflow_engine.py +++ b/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) diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index ef5d1f4..432eef7 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -29,7 +29,12 @@ from services.brains.dify_llm import ( ) from services.brains.workflow_brain import WorkflowBrain from services.runtime_variables import prepare_dynamic_config -from services.workflow.models import LLMRouteResult, RouteStatus, WorkflowStatus +from services.workflow.models import ( + ActionStatus, + LLMRouteResult, + RouteStatus, + WorkflowStatus, +) class FakeLLM: @@ -831,11 +836,18 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): set_system_prompt=lambda _prompt: None, set_tools=lambda _tools: None, call_end=FakeCallEnd(), + session_id="conv_action", ) brain._tools.execute = execute - await brain._enter_action("lookup_action") + outcome = await brain._enter_action("lookup_action") + self.assertEqual(outcome.status, ActionStatus.SUCCESS) + self.assertEqual(outcome.updated_variables, ("order_status",)) + self.assertEqual( + brain._store.values["system__last_action_invocation_id"], + outcome.invocation_id, + ) variable_events = [ frame.message for frame in queued @@ -845,6 +857,17 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(variable_events[-1]["reason"], "action") self.assertEqual(variable_events[-1]["changed"], ["order_status"]) self.assertEqual(variable_events[-1]["variables"], {"order_status": "paid"}) + completed_event = next( + frame.message + for frame in queued + if isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("event") == "action_completed" + ) + self.assertEqual(completed_event["outcome"]["status"], "success") + self.assertEqual(completed_event["outcome"]["updatedVariables"], ["order_status"]) + self.assertEqual(completed_event["sessionId"], "conv_action") + self.assertEqual(completed_event["workflowRevision"], brain._engine.revision) + self.assertNotIn("result", completed_event["outcome"]) async def test_action_result_assignment_modes_reach_tool_executor(self): tool = RuntimeTool( @@ -958,14 +981,87 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): ) brain._tools.execute = execute - await brain._enter_action("action") + outcome = await brain._enter_action("action") + self.assertEqual(outcome.status, ActionStatus.FAILURE) + self.assertEqual(outcome.error.code, "tool_error") self.assertEqual(brain._store.values["system__last_action_status"], "error") self.assertEqual( brain._store.values["system__last_action_error"], "用户关闭了确认弹窗", ) + async def test_cancelled_action_does_not_follow_failure_or_default_edge(self): + tool = RuntimeTool( + id="client_action", + name="客户端操作", + function_name="show_message", + type="client", + ) + brain = WorkflowBrain( + AssistantConfig( + type="workflow", + graph={ + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "action", + "type": "action", + "data": {"toolId": "client_action"}, + }, + {"id": "end", "type": "end", "data": {}}, + ], + "edges": [ + { + "id": "after_action", + "source": "action", + "target": "end", + "data": {"mode": "always"}, + } + ], + }, + tools=[tool], + ) + ) + queued = [] + + async def queue_frame(frame): + queued.append(frame) + + async def execute(_tool, _arguments, *, result_assignments=None): + return { + "status": "error", + "message": "会话已结束", + "updated_variables": [], + } + + brain._runtime = BrainRuntime( + context=LLMContext(messages=[]), + llm=FakeLLM(), + queue_frame=queue_frame, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=FakeCallEnd(), + ) + brain._tools.execute = execute + + config = await brain._resolve_path("action") + + self.assertEqual(config["name"], "action") + self.assertEqual( + brain._store.values["system__last_action_status"], + "cancelled", + ) + self.assertFalse( + any( + isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("event") == "edge_selected" + for frame in queued + ) + ) + async def test_action_block_policy_only_suppresses_input_while_running(self): tool = RuntimeTool( id="client_action", diff --git a/backend/tests/test_conversation_history.py b/backend/tests/test_conversation_history.py index fcba94c..ada7ccd 100644 --- a/backend/tests/test_conversation_history.py +++ b/backend/tests/test_conversation_history.py @@ -1,7 +1,8 @@ from __future__ import annotations import unittest -from unittest.mock import AsyncMock +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch from services.conversation_history import ConversationRecorder @@ -32,6 +33,53 @@ class ConversationRecorderTest(unittest.IsolatedAsyncioTestCase): }, ) + async def test_workflow_trace_is_persisted_without_counting_as_message(self): + conversation = SimpleNamespace( + extra={"workflow": {"revision": "sha256:test"}}, + message_count=3, + ) + + class FakeSession: + committed = False + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return None + + async def get(self, _model, _session_id): + return conversation + + async def commit(self): + self.committed = True + + session = FakeSession() + recorder = ConversationRecorder("conv_test") + event = { + "type": "workflow-event", + "eventId": "wfe_test", + "event": "node_entered", + "timestamp": "2026-08-01T10:00:00+08:00", + "workflowRevision": "sha256:test", + "transitionId": 0, + "nodeId": "start", + } + + with patch( + "services.conversation_history.SessionLocal", + return_value=session, + ): + await recorder.record_transport_message(event) + await recorder.record_transport_message(event) + + self.assertTrue(session.committed) + self.assertEqual(conversation.message_count, 3) + self.assertEqual(len(conversation.extra["workflowTrace"]), 1) + saved = conversation.extra["workflowTrace"][0] + self.assertEqual(saved["sessionId"], "conv_test") + self.assertEqual(saved["sequence"], 1) + if __name__ == "__main__": unittest.main() diff --git a/backend/tests/test_workflow_v3.py b/backend/tests/test_workflow_v3.py index 6d1e263..70936c3 100644 --- a/backend/tests/test_workflow_v3.py +++ b/backend/tests/test_workflow_v3.py @@ -1,6 +1,7 @@ from __future__ import annotations import unittest +from copy import deepcopy from types import SimpleNamespace from models import AssistantConfig, RuntimeModelResource @@ -54,6 +55,26 @@ def valid_graph(): class WorkflowGraphTests(unittest.TestCase): + def test_revision_pins_the_normalized_workflow_snapshot(self): + first = WorkflowEngine(valid_graph()) + same_graph = deepcopy(valid_graph()) + same_graph["settings"] = dict(reversed(list(same_graph["settings"].items()))) + second = WorkflowEngine(same_graph) + + self.assertEqual(first.revision, second.revision) + self.assertTrue(first.revision.startswith("sha256:")) + + metadata = first.session_metadata() + first.graph["settings"]["globalPrompt"] = "运行时外部修改" + self.assertEqual( + metadata["workflow"]["snapshot"]["settings"]["globalPrompt"], + "服务 {{customer}}", + ) + + changed = valid_graph() + changed["settings"]["globalPrompt"] = "另一份配置" + self.assertNotEqual(first.revision, WorkflowEngine(changed).revision) + def test_workflow_graph_owns_flat_vision_session_flag(self): graph = valid_graph() graph["settings"].update( diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 74c8ce2..a480268 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -278,6 +278,14 @@ export type ConversationMessage = { }; export type ConversationDetail = Conversation & { + extra: { + workflow?: { + revision: string; + specVersion?: number; + snapshot: Record; + }; + workflowTrace?: Array>; + }; messages: ConversationMessage[]; };