diff --git a/backend/services/brains/prompt_brain.py b/backend/services/brains/prompt_brain.py index 659fbe0..a84a64d 100644 --- a/backend/services/brains/prompt_brain.py +++ b/backend/services/brains/prompt_brain.py @@ -231,6 +231,22 @@ class PromptBrain(BaseBrain): speech, opening_message, ) + await self._emit_trace( + "message_started", + phase="opening", + hasSpeech=bool(speech), + showsMessage=True, + completionPolicy=MESSAGE_CONFIRMATION, + requiresConfirmation=True, + messageTitle=message_spec.display.title, + messageContent=message_spec.display.message, + messageActions=[ + { + "id": "confirmed", + "label": message_spec.display.confirm_label, + } + ], + ) message_result = await self._message_stages.run( message_spec, speak=self._speak_opening, @@ -242,10 +258,20 @@ class PromptBrain(BaseBrain): release_input_on_failure=False, ) if not message_result.succeeded: + await self._emit_trace( + "message_failed", + phase="opening", + error=message_result.error or "开场消息显示失败", + ) await self._fail_opening( message_result.error or "开场消息显示失败" ) return + await self._emit_trace( + "message_completed", + phase="opening", + action=message_result.action, + ) context_message = confirmation_context_message( message_result, source="prompt-opening", diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index 262c1ee..92506e8 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -1557,6 +1557,20 @@ class WorkflowBrain(BaseBrain): showsMessage=require_confirmation, completionPolicy=completion_policy, requiresConfirmation=require_confirmation, + **( + { + "messageTitle": display.title, + "messageContent": display.message, + "messageActions": [ + { + "id": "confirmed", + "label": display.confirm_label, + } + ], + } + if display is not None + else {} + ), ), ) if result.succeeded: diff --git a/backend/services/client_tools.py b/backend/services/client_tools.py index 1fe1cf7..d570126 100644 --- a/backend/services/client_tools.py +++ b/backend/services/client_tools.py @@ -194,6 +194,7 @@ class ClientToolBroker(FrameProcessor): toolCallId=tool_call_id, functionName=pending.function_name, status=status, + internalStage=pending.function_name == "show_message", **({"userAction": user_action} if user_action else {}), **( {"error": str(message.get("message") or "客户端工具执行失败")} diff --git a/backend/services/conversation_history.py b/backend/services/conversation_history.py index 3b2ae83..02f5383 100644 --- a/backend/services/conversation_history.py +++ b/backend/services/conversation_history.py @@ -120,6 +120,8 @@ class ConversationRecorder: async def _append_workflow_event(self, message: dict) -> None: """Persist bounded control-plane history without creating chat rows.""" + if message.get("internalStage") is True: + return event_id = str(message.get("eventId") or "") event_name = str(message.get("event") or "") event_key = f"workflow:{event_id}" diff --git a/backend/services/pipecat/call_lifecycle.py b/backend/services/pipecat/call_lifecycle.py index fc39e01..61b17a1 100644 --- a/backend/services/pipecat/call_lifecycle.py +++ b/backend/services/pipecat/call_lifecycle.py @@ -5,6 +5,8 @@ from __future__ import annotations import asyncio from collections.abc import Awaitable, Callable from dataclasses import dataclass +from typing import Any +from uuid import uuid4 from loguru import logger from pipecat.frames.frames import ( @@ -14,6 +16,38 @@ from pipecat.frames.frames import ( InterruptionFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.utils.time import time_now_iso8601 + + +CALL_END_EVENT_BY_STATE = { + "pending": "conversation_end_requested", + "speaking": "conversation_end_speaking", + "cancelled": "conversation_end_cancelled", + "ended": "conversation_ended", +} + + +def call_end_trace_event( + state: str, + *, + reason: str = "", + interruptible: bool = False, + voice_interruptible: bool = False, +) -> dict[str, Any]: + """Build one shared record used by live calls and headless tests.""" + + return { + "type": "workflow-event", + "eventId": f"wfe_{uuid4().hex[:20]}", + "event": CALL_END_EVENT_BY_STATE.get( + state, + "conversation_end_state_changed", + ), + "timestamp": time_now_iso8601(), + **({"reason": reason} if reason else {}), + "interruptible": interruptible, + "voiceInterruptible": voice_interruptible, + } class SpeechPlaybackCompletion: @@ -76,6 +110,7 @@ class CallEndCoordinator: ): self._queue_end = queue_end self._on_state_change = on_state_change + self._state_tasks: set[asyncio.Task[None]] = set() self._ending = False self._interruptible = False self._armed = False @@ -86,7 +121,7 @@ class CallEndCoordinator: self._tracked_speech_completions: set[SpeechPlaybackCompletion] = set() self._finish_after_tracked_speech = False self._finished = False - self._reason = "completed" + self._reason = "" @property def ending(self) -> bool: @@ -106,6 +141,12 @@ class CallEndCoordinator: """Whether transport output is currently producing bot speech.""" return self._speaking + @property + def reason(self) -> str: + """Current requested ending reason for client feedback and records.""" + + return self._reason + async def wait_until_silent(self) -> None: """Wait for the transport-owned bot speech boundary.""" await self._speech_stopped.wait() @@ -124,6 +165,7 @@ class CallEndCoordinator: """Cancel one pending ending when its tool explicitly allows it.""" if not self.interruptible: return False + was_interruptible = self._interruptible self._ending = False self._interruptible = False self._armed = False @@ -132,12 +174,26 @@ class CallEndCoordinator: # Transport speech state remains authoritative: if audio is currently # playing, the interruption path will still emit BotStoppedSpeakingFrame. logger.info("用户打断结束语,已撤销待挂断状态") - self._schedule_state("cancelled") + self._schedule_state("cancelled", interruptible=was_interruptible) return True - def _schedule_state(self, state: str) -> None: + def _schedule_state( + self, + state: str, + *, + interruptible: bool | None = None, + ) -> None: if self._on_state_change is not None: - asyncio.create_task(self._on_state_change(state, self.interruptible)) + task = asyncio.create_task( + self._on_state_change( + state, + self.interruptible + if interruptible is None + else interruptible, + ) + ) + self._state_tasks.add(task) + task.add_done_callback(self._state_tasks.discard) def begin_response(self) -> None: """Start tracking speech produced by one LLM response.""" @@ -196,6 +252,9 @@ class CallEndCoordinator: return self._finished = True self._interruptible = False + pending_states = tuple(self._state_tasks) + if pending_states: + await asyncio.gather(*pending_states, return_exceptions=True) await self._queue_end(self._reason) async def observe(self, frame) -> None: diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 988254b..ad1da7a 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -17,6 +17,7 @@ from services.conversation_history import ConversationRecorder from services.pipecat.call_lifecycle import ( CallEndCoordinator, EndCallAfterSpeechProcessor, + call_end_trace_event, ) from services.pipecat.service_factory import ( config_with_resource, @@ -208,6 +209,7 @@ def _call_end_state_frame( *, interruptible: bool, voice_interruptible: bool, + reason: str = "", ) -> OutputTransportMessageUrgentFrame: """Build the client event used to render the pending call-end state.""" return OutputTransportMessageUrgentFrame( @@ -216,10 +218,30 @@ def _call_end_state_frame( "state": state, "interruptible": interruptible, "voice_interruptible": voice_interruptible, + **({"reason": reason} if reason else {}), } ) +def _call_lifecycle_trace_frame( + state: str, + *, + reason: str = "", + interruptible: bool = False, + voice_interruptible: bool = False, +) -> OutputTransportMessageUrgentFrame: + """Build one persistent runtime record for the call-ending lifecycle.""" + + return OutputTransportMessageUrgentFrame( + message=call_end_trace_event( + state, + reason=reason, + interruptible=interruptible, + voice_interruptible=voice_interruptible, + ) + ) + + async def run_pipeline( transport, cfg: AssistantConfig, @@ -307,6 +329,12 @@ async def run_pipeline( if worker is None: return logger.info(f"结束通话: reason={reason}") + await worker.queue_frame( + _call_lifecycle_trace_frame( + "ended", + reason=reason, + ) + ) await worker.queue_frame( OutputTransportMessageUrgentFrame( message={"type": "call-ended", "reason": reason} @@ -318,13 +346,21 @@ async def run_pipeline( worker = worker_holder.get("worker") if worker is None: return + voice_interruptible = interruptible and current_enable_interrupt await worker.queue_frame( _call_end_state_frame( state, interruptible=interruptible, - voice_interruptible=( - interruptible and current_enable_interrupt - ), + voice_interruptible=voice_interruptible, + reason=call_end.reason, + ) + ) + await worker.queue_frame( + _call_lifecycle_trace_frame( + state, + reason=call_end.reason, + interruptible=interruptible, + voice_interruptible=voice_interruptible, ) ) @@ -1043,6 +1079,12 @@ async def run_realtime_pipeline( if worker is None: return logger.info(f"结束 Realtime 通话: reason={reason}") + await worker.queue_frame( + _call_lifecycle_trace_frame( + "ended", + reason=reason, + ) + ) await worker.queue_frame( OutputTransportMessageUrgentFrame( message={"type": "call-ended", "reason": reason} @@ -1054,13 +1096,21 @@ async def run_realtime_pipeline( worker = worker_holder.get("worker") if worker is None: return + voice_interruptible = interruptible and cfg.enableInterrupt await worker.queue_frame( _call_end_state_frame( state, interruptible=interruptible, - voice_interruptible=( - interruptible and cfg.enableInterrupt - ), + voice_interruptible=voice_interruptible, + reason=call_end.reason, + ) + ) + await worker.queue_frame( + _call_lifecycle_trace_frame( + state, + reason=call_end.reason, + interruptible=interruptible, + voice_interruptible=voice_interruptible, ) ) diff --git a/backend/services/test_runs/evaluator.py b/backend/services/test_runs/evaluator.py index 08498c1..55e0d15 100644 --- a/backend/services/test_runs/evaluator.py +++ b/backend/services/test_runs/evaluator.py @@ -163,6 +163,7 @@ class EvaluationEngine: user_input=expected_turn.user_input, assistant_reply="", tool_calls=[], + runtime_records=[], ) evaluations = [ await self._evaluate_behavior(behavior, raw) @@ -175,6 +176,7 @@ class EvaluationEngine: user_input=raw.user_input, assistant_reply=raw.assistant_reply, tool_calls=[item.as_result() for item in raw.tool_calls], + runtime_records=list(raw.runtime_records), evaluations=evaluations, ) ) diff --git a/backend/services/test_runs/orchestrator.py b/backend/services/test_runs/orchestrator.py index 292b466..ee78b1d 100644 --- a/backend/services/test_runs/orchestrator.py +++ b/backend/services/test_runs/orchestrator.py @@ -302,6 +302,7 @@ class TestRunOrchestrator: "greeting": "", "assistantOutputs": [], "autoConfirmedMessages": [], + "runtimeRecords": [], }, "overallCriteria": [], } @@ -324,6 +325,7 @@ class TestRunOrchestrator: "autoConfirmedMessages": list( startup.auto_confirmed_messages ), + "runtimeRecords": list(startup.runtime_records), } row.result = result await session.commit() @@ -346,6 +348,7 @@ class TestRunOrchestrator: "userInput": turn.user_input, "assistantReply": turn.assistant_reply, "toolCalls": [item.as_result() for item in turn.tool_calls], + "runtimeRecords": list(turn.runtime_records), "evaluations": [], } turns = [item for item in turns if item.get("id") != turn.id] @@ -382,6 +385,7 @@ class TestRunOrchestrator: "autoConfirmedMessages": list( startup.auto_confirmed_messages ), + "runtimeRecords": list(startup.runtime_records), }, "overallCriteria": [ item.model_dump(mode="json", by_alias=True) diff --git a/backend/services/test_runs/text_runner.py b/backend/services/test_runs/text_runner.py index d47ed0f..5555692 100644 --- a/backend/services/test_runs/text_runner.py +++ b/backend/services/test_runs/text_runner.py @@ -26,6 +26,7 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMMessagesAppendFrame, ManuallySwitchServiceFrame, + OutputTransportMessageUrgentFrame, TTSSpeakFrame, ) from pipecat.pipeline.pipeline import Pipeline @@ -44,6 +45,7 @@ from services.knowledge import search as search_knowledge from services.pipecat.call_lifecycle import ( CallEndCoordinator, FixedSpeechPlaybackMarkerFrame, + call_end_trace_event, ) from services.pipecat.processors import ( KnowledgeRetrievalProcessor, @@ -102,6 +104,7 @@ class RawTurnResult: user_input: str assistant_reply: str tool_calls: list[RawToolCall] + runtime_records: list[dict[str, Any]] = field(default_factory=list) @dataclass @@ -111,6 +114,7 @@ class TextStartupResult: greeting: str = "" assistant_outputs: list[str] = field(default_factory=list) auto_confirmed_messages: list[dict[str, Any]] = field(default_factory=list) + runtime_records: list[dict[str, Any]] = field(default_factory=list) @dataclass @@ -148,6 +152,30 @@ class _ResponseWindow: return self.active_llm == 0 and self.active_assistant == 0 and not self.pending_tools +def _opening_speech_from_message(message: Any) -> str: + """Extract only the transcript emitted by Prompt's opening Message stage.""" + + if ( + not isinstance(message, dict) + or message.get("type") != "transcript" + or message.get("source") != "prompt-opening-speech" + ): + return "" + return str(message.get("content") or "").strip() + + +def _runtime_record_from_message(message: Any) -> dict[str, Any] | None: + """Keep the same runtime-record envelope used by live conversation history.""" + + if ( + not isinstance(message, dict) + or message.get("type") != "workflow-event" + or message.get("internalStage") is True + ): + return None + return dict(message) + + class _TextCaptureProcessor(FrameProcessor): """Capture assistant/tool output and expose a reliable response boundary.""" @@ -156,6 +184,8 @@ class _TextCaptureProcessor(FrameProcessor): self.context = context self.window = _ResponseWindow() self.outputs: list[str] = [] + self.opening_speech_outputs: list[str] = [] + self.runtime_records: list[dict[str, Any]] = [] self.tool_calls: list[RawToolCall] = [] self._tool_by_id: dict[str, RawToolCall] = {} self.error: TestExecutionError | None = None @@ -195,6 +225,14 @@ class _TextCaptureProcessor(FrameProcessor): await self.on_fixed_speech_complete() return + if isinstance(frame, OutputTransportMessageUrgentFrame): + opening_speech = _opening_speech_from_message(frame.message) + if opening_speech: + self.opening_speech_outputs.append(opening_speech) + runtime_record = _runtime_record_from_message(frame.message) + if runtime_record is not None: + self.runtime_records.append(runtime_record) + if isinstance(frame, FixedSpeechPlaybackMarkerFrame): await frame.completion.mark_played() self.window.touch() @@ -248,6 +286,27 @@ class _TextCaptureProcessor(FrameProcessor): await self.push_frame(frame, direction) +def _resolve_startup_greeting( + shared_greeting: str, + opening_speech_outputs: list[str], +) -> str: + """Return the opening speech shown by the headless run. + + Prompt confirmation mode deliberately suppresses the shared greeting + because its atomic Message stage owns both speech and the dialog. In text + tests the fixed speech transcript carries a dedicated source marker, so + preserve that exact output instead of guessing from unrelated fixed speech. + """ + + greeting = shared_greeting.strip() + if greeting: + return greeting + return next( + (text.strip() for text in opening_speech_outputs if text.strip()), + "", + ) + + def _context_messages(context_turns: list[ContextTurn]) -> list[dict[str, Any]]: messages: list[dict[str, Any]] = [] latest_call_by_name: dict[str, str] = {} @@ -494,9 +553,34 @@ class TextPipelineRunner: async def queue_call_end(reason: str) -> None: logger.debug(f"文本测试收到会话结束请求: {reason}") + await worker.queue_frame( + OutputTransportMessageUrgentFrame( + message=call_end_trace_event( + "ended", + reason=reason, + ) + ) + ) await worker.queue_frame(EndFrame(reason=reason)) - call_end = CallEndCoordinator(queue_call_end) + async def record_call_end_state( + state: str, + interruptible: bool, + ) -> None: + await worker.queue_frame( + OutputTransportMessageUrgentFrame( + message=call_end_trace_event( + state, + reason=call_end.reason, + interruptible=interruptible, + ) + ) + ) + + call_end = CallEndCoordinator( + queue_call_end, + on_state_change=record_call_end_state, + ) async def finish_fixed_speech_if_ending() -> None: if call_end.ending: @@ -667,6 +751,7 @@ class TextPipelineRunner: runner_task: asyncio.Task[None] | None = None try: + startup_runtime_record_marker = len(capture.runtime_records) await brain.run_preflight() if executor.unmocked_calls: raise TestExecutionError( @@ -693,12 +778,19 @@ class TextPipelineRunner: runner_task, marker, ) + startup_greeting = _resolve_startup_greeting( + greeting, + capture.opening_speech_outputs, + ) startup = TextStartupResult( - greeting=greeting, + greeting=startup_greeting, assistant_outputs=list(capture.outputs), auto_confirmed_messages=list( text_client_tools.auto_confirmed_messages ), + runtime_records=list( + capture.runtime_records[startup_runtime_record_marker:] + ), ) if on_startup_complete is not None: await on_startup_complete(startup) @@ -775,6 +867,7 @@ class TextPipelineRunner: activity_marker = capture.window.activity_count output_marker = len(capture.outputs) tool_marker = len(capture.tool_calls) + runtime_record_marker = len(capture.runtime_records) await worker.queue_frame( LLMMessagesAppendFrame( messages=[{"role": "user", "content": turn.user_input}], @@ -795,4 +888,7 @@ class TextPipelineRunner: user_input=turn.user_input, assistant_reply="\n\n".join(capture.outputs[output_marker:]).strip(), tool_calls=list(capture.tool_calls[tool_marker:]), + runtime_records=list( + capture.runtime_records[runtime_record_marker:] + ), ) diff --git a/backend/test_schemas.py b/backend/test_schemas.py index b97c958..60a7880 100644 --- a/backend/test_schemas.py +++ b/backend/test_schemas.py @@ -364,6 +364,7 @@ class BatchTurnResult(CamelModel): user_input: str assistant_reply: str tool_calls: list[BatchToolCallRecord] = Field(default_factory=list) + runtime_records: list[dict[str, Any]] = Field(default_factory=list) evaluations: list[BatchEvaluationResult] = Field(default_factory=list) @@ -372,6 +373,7 @@ class BatchStartupResult(CamelModel): greeting: str = "" assistant_outputs: list[str] = Field(default_factory=list) auto_confirmed_messages: list[dict[str, Any]] = Field(default_factory=list) + runtime_records: list[dict[str, Any]] = Field(default_factory=list) class BatchRunCaseOut(CamelModel): diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index ea70351..7dfc435 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -521,6 +521,24 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): self.assertIn("点击“确认”", confirmation_messages[0]["content"]) self.assertIn("弹窗标题：重要提示", confirmation_messages[0]["content"]) self.assertIn("弹窗内容：请确认已阅读。", confirmation_messages[0]["content"]) + message_traces = [ + frame.message + for frame in queued + if isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("type") == "workflow-event" + and frame.message.get("event") + in {"message_started", "message_completed"} + ] + self.assertEqual( + [event["event"] for event in message_traces], + ["message_started", "message_completed"], + ) + self.assertEqual(message_traces[0]["messageTitle"], "重要提示") + self.assertEqual( + message_traces[0]["messageContent"], + "请确认已阅读。", + ) + self.assertEqual(message_traces[1]["action"], "confirmed") self.assertEqual( sum(isinstance(frame, LLMRunFrame) for frame in queued), 1, diff --git a/backend/tests/test_call_lifecycle.py b/backend/tests/test_call_lifecycle.py index 89b4ba9..9962f25 100644 --- a/backend/tests/test_call_lifecycle.py +++ b/backend/tests/test_call_lifecycle.py @@ -11,12 +11,27 @@ from pipecat.frames.frames import ( ) from services.pipecat.call_lifecycle import ( CallEndCoordinator, + call_end_trace_event, playback_marker_for, ) from services.pipecat.processors import CallEndingUserMuteStrategy class CallEndCoordinatorTest(unittest.IsolatedAsyncioTestCase): + def test_shared_trace_event_uses_runtime_record_contract(self): + event = call_end_trace_event( + "pending", + reason="用户已完成咨询", + interruptible=True, + voice_interruptible=True, + ) + + self.assertEqual(event["type"], "workflow-event") + self.assertEqual(event["event"], "conversation_end_requested") + self.assertEqual(event["reason"], "用户已完成咨询") + self.assertTrue(event["interruptible"]) + self.assertTrue(event["voiceInterruptible"]) + async def asyncSetUp(self): self.reasons: list[str] = [] self.states: list[tuple[str, bool]] = [] @@ -87,7 +102,7 @@ class CallEndCoordinatorTest(unittest.IsolatedAsyncioTestCase): self.assertTrue(self.coordinator.accepts_input) self.assertTrue(self.coordinator.cancel_pending_end()) await asyncio.sleep(0) - self.assertEqual(self.states[-1], ("cancelled", False)) + self.assertEqual(self.states[-1], ("cancelled", True)) self.assertFalse(self.coordinator.ending) self.assertTrue(self.coordinator.accepts_input) self.assertTrue(self.coordinator.speaking) @@ -95,6 +110,7 @@ class CallEndCoordinatorTest(unittest.IsolatedAsyncioTestCase): await self.coordinator.observe(BotStoppedSpeakingFrame()) self.assertFalse(self.coordinator.speaking) self.assertEqual(self.reasons, []) + self.assertEqual(self.coordinator.reason, "user_goodbye") async def test_committed_end_rejects_cancellation(self): self.coordinator.begin("required_shutdown") diff --git a/backend/tests/test_conversation_history.py b/backend/tests/test_conversation_history.py index e0b90f6..2b85974 100644 --- a/backend/tests/test_conversation_history.py +++ b/backend/tests/test_conversation_history.py @@ -144,6 +144,22 @@ class ConversationRecorderTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(saved["sessionId"], "conv_test") self.assertEqual(saved["sequence"], 1) + async def test_internal_message_stage_transport_trace_is_not_persisted_twice(self): + recorder = ConversationRecorder("conv_test") + event = { + "type": "workflow-event", + "eventId": "wfe_internal", + "event": "client_tool_completed", + "functionName": "show_message", + "internalStage": True, + } + + with patch("services.conversation_history.SessionLocal") as session_local: + await recorder.record_transport_message(event) + + session_local.assert_not_called() + self.assertEqual(recorder._trace_sequence, 0) + async def test_image_is_persisted_as_message_with_artifact(self): conversation = SimpleNamespace(message_count=0) diff --git a/backend/tests/test_debug_auto_test.py b/backend/tests/test_debug_auto_test.py index 6e03129..6fc3b2b 100644 --- a/backend/tests/test_debug_auto_test.py +++ b/backend/tests/test_debug_auto_test.py @@ -10,10 +10,13 @@ from services.runtime_variables import DynamicVariableStore, prepare_dynamic_con from services.test_runs.mock_tools import MockToolExecutor, TextClientToolPort from services.test_runs.text_runner import ( _TextCaptureProcessor, + _opening_speech_from_message, + _resolve_startup_greeting, + _runtime_record_from_message, _wait_for_startup, _wait_for_quiescence, ) -from test_schemas import BatchRunCreate +from test_schemas import BatchRunCreate, BatchStartupResult class FakeLLM: @@ -126,6 +129,25 @@ class DebugAutoTestContractTests(unittest.TestCase): "你好，测试用户", ) + def test_startup_result_exposes_real_runtime_records(self): + result = BatchStartupResult.model_validate( + { + "completed": True, + "runtimeRecords": [ + { + "type": "workflow-event", + "eventId": "wfe_opening", + "event": "message_started", + } + ], + } + ) + + self.assertEqual( + result.runtime_records[0]["event"], + "message_started", + ) + def test_text_client_port_records_automatic_message_confirmation(self): port = TextClientToolPort() @@ -142,6 +164,51 @@ class DebugAutoTestContractTests(unittest.TestCase): [{"title": "重要提示", "message": "请确认"}], ) + def test_confirmation_owned_speech_is_kept_as_startup_greeting(self): + self.assertEqual( + _resolve_startup_greeting( + "", + ["您好，请先阅读弹窗。"], + ), + "您好，请先阅读弹窗。", + ) + self.assertEqual( + _resolve_startup_greeting( + "普通开场白", + ["其他固定播报"], + ), + "普通开场白", + ) + self.assertEqual( + _resolve_startup_greeting( + "", + [], + ), + "", + ) + + def test_only_prompt_opening_transcript_is_identified_as_greeting(self): + self.assertEqual( + _opening_speech_from_message( + { + "type": "transcript", + "content": "您好，请先阅读弹窗。", + "source": "prompt-opening-speech", + } + ), + "您好，请先阅读弹窗。", + ) + self.assertEqual( + _opening_speech_from_message( + { + "type": "transcript", + "content": "普通 Workflow 播报", + "source": "workflow-message-speech", + } + ), + "", + ) + class DebugAutoTestRuntimeTests(unittest.IsolatedAsyncioTestCase): async def test_system_end_tool_keeps_real_lifecycle_in_text_tests(self): @@ -180,6 +247,29 @@ class DebugAutoTestRuntimeTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(params.result["action"], "ending_call") self.assertEqual(executor_holder["executor"].unmocked_calls, []) + async def test_text_capture_keeps_runtime_records_for_auto_test_ui(self): + event = { + "type": "workflow-event", + "eventId": "wfe_end", + "event": "conversation_end_requested", + "reason": "测试完成", + } + + self.assertEqual(_runtime_record_from_message(event), event) + self.assertIsNone( + _runtime_record_from_message({"type": "call-end-state"}) + ) + self.assertIsNone( + _runtime_record_from_message( + { + "type": "workflow-event", + "eventId": "wfe_internal", + "event": "client_tool_completed", + "internalStage": True, + } + ) + ) + async def test_startup_waits_for_queued_entry_reply(self): capture = _TextCaptureProcessor(LLMContext(messages=[])) diff --git a/frontend/src/components/assistant-editor/debug-auto-test.tsx b/frontend/src/components/assistant-editor/debug-auto-test.tsx index 8aece17..9f22354 100644 --- a/frontend/src/components/assistant-editor/debug-auto-test.tsx +++ b/frontend/src/components/assistant-editor/debug-auto-test.tsx @@ -22,6 +22,10 @@ import { X, } from "lucide-react"; +import { + DebugTimeline, + type DebugTimelineItem, +} from "@/components/debug-timeline"; import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; import { @@ -252,17 +256,32 @@ function CasePicker({ export function AutoTestLivePanel({ testCase, state, + onStart, }: { testCase: TestCase | null; state: AutoTestRunState; + onStart: () => void; }) { if (!testCase) { return ( - } - title="选择一个测试用例" - description="运行真实 Pipeline，并在这里查看逐轮回复、工具调用和判断结果。" - /> +
+ + + +
+ 选择测试用例后开始 +
+ +
); } @@ -270,20 +289,20 @@ export function AutoTestLivePanel({ const validationMessage = runnableMessage(testCase); return (
-
-
+
+
{testCase.name}
-
- - {TEST_CASE_INPUT_MODE_SHORT_LABEL[testCase.inputMode]} - - - {testCase.turns.length} 轮 - +
+ {TEST_CASE_INPUT_MODE_SHORT_LABEL[testCase.inputMode]} + · + {testCase.turns.length} 轮 + {state.evaluatorName && ( + <> + · + {state.evaluatorName} + + )}
{testCase.description && (

@@ -294,11 +313,22 @@ export function AutoTestLivePanel({

{validationMessage}
- ) : state.evaluatorName ? ( -

- 评估模型 · {state.evaluatorName} -

) : null} + + {state.blockedMessage && !validationMessage && ( +

+ {state.blockedMessage} +

+ )}
); @@ -327,8 +357,11 @@ export function AutoTestLivePanel({ if (startup?.completed) { return (
- - +
正在执行第 1 轮固定输入… @@ -351,62 +384,11 @@ export function AutoTestLivePanel({ return (
- {startup?.completed && } - {startup?.completed && } - {turns.map((turn) => ( -
-
- - 固定输入 - - {turn.index + 1} / {state.maxTurns} - - -
- {turn.userInput} -
-
- -
- 助手 -
- {turn.assistantReply || "（没有生成文字回复）"} -
-
- - {turn.toolCalls.length > 0 && ( -
-
- - 工具调用 -
- {turn.toolCalls.map((toolCall) => ( -
- {toolCall.functionName} - - {toolCall.outcome === "success" ? "成功" : "错误"} ·{" "} - {toolCall.durationMs} ms - - 
-                      {toolCall.argumentsJson}
-
-
- ))} -
- )} - - {turn.evaluations.length > 0 && ( -
- {turn.evaluations.map((evaluation) => ( - - ))} -
- )} -
- ))} + {state.phase === "running" && (
@@ -446,78 +428,15 @@ export function AutoTestLivePanel({ ); } -function StartupDetails({ startup }: { startup: BatchStartupResult }) { +function startupAssistantMessages(startup: BatchStartupResult): string[] { const greeting = startup.greeting.trim(); - - return ( -
-
-
-
开场记录
-

- 已按真实开场流程模拟，不计入脚本轮次断言 -

-
- - 已完成 - -
- - {greeting && ( - - )} - - {startup.autoConfirmedMessages.map((message, index) => { - const confirmedAction = - message.actions?.find((action) => action.id === "confirmed") ?? - message.actions?.[0]; - return ( -
-
- 消息弹窗{startup.autoConfirmedMessages.length > 1 ? ` ${index + 1}` : ""} -
-
- {message.title?.trim() || "未命名消息"} -
-

- {message.message?.trim() || "（没有消息正文）"} -

-
- - 自动点击“{confirmedAction?.label?.trim() || "确认"}” -
-
- ); - })} -
- ); -} - -function StartupAssistantMessages({ - startup, -}: { - startup: BatchStartupResult; -}) { - const openingTexts = new Set( - [ - startup.greeting, - ...startup.autoConfirmedMessages.map((message) => message.message ?? ""), - ] - .map(normalizeDisplayText) - .filter(Boolean), - ); + const greetingText = normalizeDisplayText(greeting); const seen = new Set(); const assistantMessages = startup.assistantOutputs.filter((output) => { const normalizedOutput = normalizeDisplayText(output); if ( !normalizedOutput || - openingTexts.has(normalizedOutput) || + normalizedOutput === greetingText || seen.has(normalizedOutput) ) { return false; @@ -525,43 +444,235 @@ function StartupAssistantMessages({ seen.add(normalizedOutput); return true; }); + return [greeting, ...assistantMessages].filter(Boolean); +} - if (assistantMessages.length === 0) return null; +function legacyStartupRuntimeRecords( + startup: BatchStartupResult, +): Array> { + return startup.autoConfirmedMessages.flatMap((message, index) => { + const confirmedAction = + message.actions?.find((action) => action.id === "confirmed") ?? + message.actions?.[0]; + const legacyId = `${index}-${message.title ?? message.message ?? "message"}`; + return [ + { + eventId: `legacy-message-started-${legacyId}`, + event: "message_started", + phase: "opening", + requiresConfirmation: true, + messageTitle: message.title?.trim() || "未命名消息", + messageContent: message.message?.trim() || "（没有消息正文）", + messageActions: message.actions ?? [], + }, + { + eventId: `legacy-message-completed-${legacyId}`, + event: "message_completed", + phase: "opening", + action: confirmedAction?.id?.trim() || "confirmed", + }, + ]; + }); +} - const label = - startup.autoConfirmedMessages.length > 0 - ? "助手 · 确认后自动输出" - : "助手 · 开场自动输出"; - - return ( -
- {assistantMessages.map((message, index) => ( -
- {label} -
- {message.trim()} -
-
- ))} -
- ); +function runtimeRecordKey( + event: Record, + prefix: string, + index: number, +): string { + const id = typeof event.eventId === "string" ? event.eventId : ""; + const name = typeof event.event === "string" ? event.event : "record"; + return id || `${prefix}-${name}-${index}`; } function normalizeDisplayText(value: string): string { return value.trim().replace(/\s+/g, " "); } -function StartupTextBlock({ label, value }: { label: string; value: string }) { +function AutoTestTimeline({ + startup, + turns, + maxTurns, +}: { + startup?: BatchStartupResult; + turns: BatchRunCase["turns"]; + maxTurns: number; +}) { + const items: DebugTimelineItem[] = []; + + if (startup?.completed) { + items.push({ + id: "startup-boundary", + kind: "content", + content: ( +
+ + 开场阶段 · 自动执行，不计入脚本断言 + +
+ ), + }); + + const openingMessages = startupAssistantMessages(startup); + const records = + (startup.runtimeRecords?.length ?? 0) > 0 + ? startup.runtimeRecords ?? [] + : legacyStartupRuntimeRecords(startup); + const firstMessageStartedIndex = records.findIndex( + (event) => event.event === "message_started", + ); + const firstMessageCompletedIndex = records.findIndex( + (event) => event.event === "message_completed", + ); + const appendOpeningMessage = (message: string, index: number) => { + items.push({ + id: `startup-message-${index}`, + kind: "message", + message: { + id: `startup-message-${index}`, + role: "assistant", + content: message, + }, + }); + }; + + const appendRuntimeRecord = ( + event: Record, + index: number, + ) => { + const id = runtimeRecordKey(event, "startup-record", index); + items.push({ + id, + kind: "runtime-record", + record: { + id, + event, + timestamp: + typeof event.timestamp === "string" ? event.timestamp : undefined, + }, + }); + }; + + if (firstMessageStartedIndex < 0) { + records.forEach(appendRuntimeRecord); + openingMessages.forEach(appendOpeningMessage); + } else { + records.forEach((event, index) => { + appendRuntimeRecord(event, index); + if (index === firstMessageStartedIndex && openingMessages[0]) { + appendOpeningMessage(openingMessages[0], 0); + } + if (index === firstMessageCompletedIndex) { + openingMessages.slice(1).forEach((message, outputIndex) => { + appendOpeningMessage(message, outputIndex + 1); + }); + } + }); + if (firstMessageCompletedIndex < 0) { + openingMessages.slice(1).forEach((message, outputIndex) => { + appendOpeningMessage(message, outputIndex + 1); + }); + } + } + } + + turns.forEach((turn) => { + items.push({ + id: `turn-user-${turn.id}`, + kind: "message", + message: { + id: `turn-user-${turn.id}`, + role: "user", + label: `固定输入 · ${turn.index + 1} / ${maxTurns}`, + content: turn.userInput, + }, + }); + items.push({ + id: `turn-assistant-${turn.id}`, + kind: "message", + message: { + id: `turn-assistant-${turn.id}`, + role: "assistant", + content: turn.assistantReply || "（没有生成文字回复）", + }, + }); + turn.runtimeRecords?.forEach((event, index) => { + const id = runtimeRecordKey(event, `turn-record-${turn.id}`, index); + items.push({ + id, + kind: "runtime-record", + record: { + id, + event, + timestamp: + typeof event.timestamp === "string" ? event.timestamp : undefined, + }, + }); + }); + if (turn.toolCalls.length > 0) { + items.push({ + id: `turn-tools-${turn.id}`, + kind: "content", + content: , + }); + } + if (turn.evaluations.length > 0) { + items.push({ + id: `turn-evaluations-${turn.id}`, + kind: "content", + content: ( +
+ {turn.evaluations.map((evaluation) => ( + + ))} +
+ ), + }); + } + }); + + return ; +} + +function AutoTestToolDetails({ + toolCalls, +}: { + toolCalls: BatchRunCase["turns"][number]["toolCalls"]; +}) { return ( -
-
{label}
-

- {value} -

-
+
+ + + 工具调用详情 + + {toolCalls.length} 次 + + +
+ {toolCalls.map((toolCall) => ( +
+
+ {toolCall.functionName} + + {toolCall.outcome === "success" ? "成功" : "错误"} ·{" "} + {toolCall.durationMs} ms + +
+
+ 
+                {toolCall.argumentsJson}
+
+ 
+                {toolCall.resultJson}
+
+
+
+ ))} +
+
); } @@ -671,7 +782,6 @@ export function AutoTestRunBar({ loading, onSelectCase, onClearCase, - onStart, onStop, onRerun, }: { @@ -682,7 +792,6 @@ export function AutoTestRunBar({ loading: boolean; onSelectCase: (id: string) => void; onClearCase: () => void; - onStart: () => void; onStop: () => void; onRerun: () => void; }) { @@ -818,17 +927,6 @@ export function AutoTestRunBar({ )}
-
{state.blockedMessage && testCase && (

diff --git a/frontend/src/components/assistant-editor/debug-preview.tsx b/frontend/src/components/assistant-editor/debug-preview.tsx index 0e68790..a58168f 100644 --- a/frontend/src/components/assistant-editor/debug-preview.tsx +++ b/frontend/src/components/assistant-editor/debug-preview.tsx @@ -29,6 +29,10 @@ import { DebugModeTabs, useAutoTestRunner, } from "@/components/assistant-editor/debug-auto-test"; +import { + DebugTimeline, + type DebugTimelineItem, +} from "@/components/debug-timeline"; import { NetworkQualityIndicator } from "@/components/network-quality-indicator"; import { ClientMessageDialog } from "@/components/client-message-dialog"; import { AuraVisualizer } from "@/components/ui/aura-visualizer"; @@ -60,6 +64,7 @@ import { type ChatMessage, type CallEndState, type ClientToolDefinition, + type LiveRuntimeRecord, type UserInputPart, type VoicePreview, type VoicePreviewStatus, @@ -383,7 +388,11 @@ function AutoTestPanel({ return (

- +
@@ -895,6 +903,7 @@ function DebugVoicePanel({ localStream, remoteStream, messages, + runtimeRecords, sendText, sendUserInput, appendUserImage, @@ -924,7 +933,8 @@ function DebugVoicePanel({ const idleOrFailed = status === "idle" || status === "failed"; const showIdleHub = idleOrFailed && - (view === "video" || (messages.length === 0 && inChatView)); + (view === "video" || + (messages.length === 0 && runtimeRecords.length === 0 && inChatView)); const startBlockedMessage = !assistantId ? "请先保存助手，再开始对话。" : hasUnsavedChanges @@ -1087,6 +1097,7 @@ function DebugVoicePanel({ view={view} onViewChange={onViewChange} messages={messages} + runtimeRecords={runtimeRecords} recording={recording} camera={camera} videoStream={preview.videoStream} @@ -1115,7 +1126,11 @@ function DebugVoicePanel({ connect={startConversation} /> ) : ( - + ) ) : (
@@ -1496,6 +1511,7 @@ function DebugVisionWorkspace({ view, onViewChange, messages, + runtimeRecords, recording, camera, videoStream, @@ -1507,6 +1523,7 @@ function DebugVisionWorkspace({ view: DebugView; onViewChange: (view: DebugView) => void; messages: ChatMessage[]; + runtimeRecords: LiveRuntimeRecord[]; recording: boolean; camera: CameraPreview; videoStream: MediaStream | null; @@ -1599,7 +1616,11 @@ function DebugVisionWorkspace({ return (
- +