[Pipeline 事件总线]：完成语音助手事件模型，包含stage事件、终端reporter和测试覆盖

openspec/changes/separate-wake-and-realtime-transcript/tasks.md

@@ -49,7 +49,7 @@
 ## 7. 开源语音助手式 Pipeline 重构
 
 - [x] 7.1 更新 OpenSpec 以描述 stage 化 pipeline、事件总线、TurnController 和主说话人端点；前置条件：公开参考已确认；验收标准：proposal/design/spec/tasks 覆盖新架构和任务；测试要点：OpenSpec strict；优先级：P0；预计：45 分钟。
-- [ ] 7.2 实现 pipeline event bus 和终端事件映射；前置条件：7.1 完成；验收标准：所有 live 用户可见状态由事件产生；测试要点：事件顺序和终端文案测试；优先级：P0；预计：60 分钟。
+- [x] 7.2 实现 pipeline event bus 和终端事件映射；前置条件：7.1 完成；验收标准：所有 live 用户可见状态由事件产生；测试要点：事件顺序和终端文案测试；优先级：P0；预计：60 分钟。
 - [ ] 7.3 实现 `TurnController` 和 `VoiceAssistantPipeline`；前置条件：7.2 完成；验收标准：`run-live` 使用统一 pipeline，成功/失败 turn 均恢复待机；测试要点：两轮 fake runtime、错误恢复、上下文回归；优先级：P0；预计：60 分钟。
 - [ ] 7.4 实现本轮主说话人端点；前置条件：7.3 完成；验收标准：主说话人音色消失约 300 ms 后结束采集；测试要点：一次提问后背景噪声不拖尾、短暂停顿不断句、画像不足回退；优先级：P0；预计：60 分钟。
 - [ ] 7.5 更新 README、`.env.example`、本地 `.env` 非密钥配置；前置条件：7.2 至 7.4 完成；验收标准：运行说明匹配新 pipeline；测试要点：`--show-config` 不泄露 key；优先级：P0；预计：30 分钟。

src/owner_voice_pet/__init__.py

@@ -1,6 +1,7 @@
 """Owner voice pet pipeline package."""
 
 from .config import AppConfig
+from .events import PipelineEvent, PipelineEventBus
 from .models import (
     AudioFrame,
     AudioSegment,
@@ -29,6 +30,8 @@ from .ui import ConsolePetWindow, PetStateController, PetVisualState
 
 __all__ = [
     "AppConfig",
+    "PipelineEvent",
+    "PipelineEventBus",
     "AudioFrame",
     "AudioSegment",
     "AudioRingBuffer",

src/owner_voice_pet/events.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Callable
+
+from .models import PipelineState, ProviderError
+
+
+PIPELINE_STARTED = "pipeline_started"
+WAKE_LISTENING = "wake_listening"
+WAKE_DETECTED = "wake_detected"
+ACK_STARTED = "ack_started"
+QUESTION_PROMPT = "question_prompt"
+CAPTURE_STARTED = "capture_started"
+SPEECH_STARTED = "speech_started"
+SPEECH_ENDED = "speech_ended"
+STT_STARTED = "stt_started"
+TRANSCRIPT_FINAL = "transcript_final"
+LLM_STARTED = "llm_started"
+TTS_STARTED = "tts_started"
+PLAYBACK_FINISHED = "playback_finished"
+STANDBY_RESUMED = "standby_resumed"
+STAGE_ERROR = "stage_error"
+RECOVERING = "recovering"
+
+
+@dataclass(frozen=True, slots=True)
+class PipelineEvent:
+    type: str
+    turn_id: int | None = None
+    state: PipelineState | None = None
+    message: str = ""
+    payload: dict[str, Any] = field(default_factory=dict)
+
+
+class PipelineEventBus:
+    def __init__(self) -> None:
+        self.events: list[PipelineEvent] = []
+        self._listeners: list[Callable[[PipelineEvent], None]] = []
+
+    def subscribe(self, listener: Callable[[PipelineEvent], None]) -> None:
+        self._listeners.append(listener)
+
+    def emit(
+        self,
+        event_type: str,
+        *,
+        turn_id: int | None = None,
+        state: PipelineState | None = None,
+        message: str = "",
+        payload: dict[str, Any] | None = None,
+    ) -> PipelineEvent:
+        event = PipelineEvent(event_type, turn_id=turn_id, state=state, message=message, payload=payload or {})
+        self.events.append(event)
+        for listener in list(self._listeners):
+            listener(event)
+        return event
+
+
+def dispatch_pipeline_event(reporter: Any, event: PipelineEvent) -> None:
+    if event.type == TRANSCRIPT_FINAL:
+        reporter.transcript(str(event.payload.get("text", event.message)), final=True, turn_id=event.turn_id)
+        return
+    if event.type == STAGE_ERROR:
+        error = event.payload.get("error")
+        if isinstance(error, ProviderError):
+            reporter.error(error.stage, error.code.value, error.message, turn_id=event.turn_id)
+            return
+        reporter.error(
+            str(event.payload.get("stage", "pipeline")),
+            str(event.payload.get("code", "STAGE_ERROR")),
+            event.message,
+            turn_id=event.turn_id,
+        )
+        return
+    if event.message:
+        reporter.status((event.state.value if event.state else event.type), event.message, turn_id=event.turn_id)

src/owner_voice_pet/runtime.py

@@ -6,6 +6,26 @@ from typing import Protocol
 
 from .config import AppConfig
 from .conversation import ConversationContext
+from .events import (
+    ACK_STARTED,
+    CAPTURE_STARTED,
+    LLM_STARTED,
+    PipelineEvent,
+    PipelineEventBus,
+    PLAYBACK_FINISHED,
+    QUESTION_PROMPT,
+    RECOVERING,
+    SPEECH_ENDED,
+    SPEECH_STARTED,
+    STAGE_ERROR,
+    STANDBY_RESUMED,
+    STT_STARTED,
+    TRANSCRIPT_FINAL,
+    TTS_STARTED,
+    WAKE_DETECTED,
+    WAKE_LISTENING,
+    dispatch_pipeline_event,
+)
 from .llm import OpenAICompatibleLlmProvider
 from .models import AudioSegment, ErrorCode, PipelineState, ProviderError
 from .protocols import AudioTransport, LlmProvider, SttProvider, TtsProvider, WakeWordProvider
@@ -28,6 +48,9 @@ class RuntimeReporter(Protocol):
 
 
 class TerminalRuntimeReporter:
+    def handle_event(self, event: PipelineEvent) -> None:
+        dispatch_pipeline_event(self, event)
+
     def status(self, state: str, message: str, *, turn_id: int | None = None) -> None:
         prefix = f"[第{turn_id}轮] " if turn_id is not None else ""
         print(f"{prefix}{message}", flush=True)
@@ -73,6 +96,7 @@ class LiveVoiceRuntime:
         context: ConversationContext,
         ack_tts: TtsProvider | None = None,
         reporter: RuntimeReporter | None = None,
+        event_bus: PipelineEventBus | None = None,
         sentence_buffer: SentenceBuffer | None = None,
     ) -> None:
         self.config = config
@@ -85,6 +109,8 @@ class LiveVoiceRuntime:
         self.ack_tts = ack_tts or tts
         self.context = context
         self.reporter = reporter or TerminalRuntimeReporter()
+        self.event_bus = event_bus or PipelineEventBus()
+        self.event_bus.subscribe(self._report_event)
         self.sentence_buffer = sentence_buffer or SentenceBuffer()
         self._states: list[PipelineState] = []
 
@@ -130,7 +156,12 @@ class LiveVoiceRuntime:
     def run_turn(self, turn_id: int) -> TurnResult:
         self._states = []
         try:
-            self._state(PipelineState.WAKE_LISTENING, "待机：等待唤醒词“小杰小杰”", turn_id=turn_id)
+            self._event(
+                WAKE_LISTENING,
+                PipelineState.WAKE_LISTENING,
+                "待机：等待唤醒词“小杰小杰”",
+                turn_id=turn_id,
+            )
             user_text = self._wait_for_wake_and_user_text(turn_id)
             if isinstance(user_text, ProviderError):
                 return self._recover(user_text, turn_id)
@@ -145,15 +176,15 @@ class LiveVoiceRuntime:
         wake_error = self._wait_for_local_wake(turn_id)
         if wake_error is not None:
             return wake_error
-        self._state(PipelineState.SPEECH_DETECTING, "唤醒命中", turn_id=turn_id)
+        self._event(WAKE_DETECTED, PipelineState.SPEECH_DETECTING, "唤醒命中", turn_id=turn_id)
         ack_error = self._acknowledge_wake(turn_id)
         if ack_error is not None:
             return ack_error
-        self._state(PipelineState.SPEECH_DETECTING, "请说出问题", turn_id=turn_id)
+        self._event(QUESTION_PROMPT, PipelineState.SPEECH_DETECTING, "请说出问题", turn_id=turn_id)
         user_segment = self._capture_segment(turn_id, state_message="录音中：正在听取问题")
         if isinstance(user_segment, ProviderError):
             return user_segment
-        self._state(PipelineState.TRANSCRIBING, "转写中：正在识别问题", turn_id=turn_id)
+        self._event(STT_STARTED, PipelineState.TRANSCRIBING, "转写中：正在识别问题", turn_id=turn_id)
         transcript = self.stt.transcribe(user_segment)
         user_text = transcript.normalized_text
         if not is_valid_transcript_text(user_text):
@@ -164,7 +195,7 @@ class LiveVoiceRuntime:
                 "live-runtime",
                 "stt",
             )
-        self.reporter.transcript(user_text, final=True, turn_id=turn_id)
+        self._event(TRANSCRIPT_FINAL, PipelineState.TRANSCRIBING, "", turn_id=turn_id, payload={"text": user_text})
         return user_text
 
     def _wait_for_local_wake(self, turn_id: int) -> ProviderError | None:
@@ -182,16 +213,26 @@ class LiveVoiceRuntime:
     def _capture_segment(self, turn_id: int, *, state_message: str) -> AudioSegment | ProviderError:
         self.vad_recorder.reset()
         self.vad_recorder.provider.reset()
-        self._state(PipelineState.RECORDING, state_message, turn_id=turn_id)
+        self._event(CAPTURE_STARTED, PipelineState.RECORDING, state_message, turn_id=turn_id)
         while True:
             frames = self.transport.read_frames(timeout_ms=100)
             if not frames:
                 continue
             for frame in frames:
+                was_started = self.vad_recorder.started
                 result = self.vad_recorder.feed(frame)
+                if not was_started and self.vad_recorder.started:
+                    self._event(SPEECH_STARTED, PipelineState.RECORDING, "检测到用户语音", turn_id=turn_id)
                 if isinstance(result, ProviderError):
                     return result
                 if isinstance(result, AudioSegment):
+                    self._event(
+                        SPEECH_ENDED,
+                        PipelineState.RECORDING,
+                        "用户语音结束",
+                        turn_id=turn_id,
+                        payload={"end_reason": result.metadata.get("end_reason", "")},
+                    )
                     return result
 
     def _acknowledge_wake(self, turn_id: int) -> ProviderError | None:
@@ -200,7 +241,7 @@ class LiveVoiceRuntime:
             self._drain_input_after_playback()
             return None
         try:
-            self._state(PipelineState.SPEAKING, f"应答中：{text}", turn_id=turn_id)
+            self._event(ACK_STARTED, PipelineState.SPEAKING, f"应答中：{text}", turn_id=turn_id)
             segment = self.ack_tts.synthesize(text)
             playback = self.transport.play_pcm(segment)
             if playback.error:
@@ -223,7 +264,7 @@ class LiveVoiceRuntime:
 
     def _reply_to_user(self, user_text: str, turn_id: int) -> TurnResult:
         self.context.append_user(user_text)
-        self._state(PipelineState.THINKING, "思考中：正在生成回复", turn_id=turn_id)
+        self._event(LLM_STARTED, PipelineState.THINKING, "思考中：正在生成回复", turn_id=turn_id)
         assistant_text = ""
         try:
             for delta in self.llm.stream_reply(self.context.build_llm_messages()):
@@ -246,26 +287,42 @@ class LiveVoiceRuntime:
                 turn_id,
             )
         self.context.append_assistant(assistant_text)
-        self._state(PipelineState.WAKE_LISTENING, "恢复待机：可继续唤醒", turn_id=turn_id)
+        self._event(STANDBY_RESUMED, PipelineState.WAKE_LISTENING, "恢复待机：可继续唤醒", turn_id=turn_id)
         return TurnResult(True, user_text, assistant_text, states=list(self._states))
 
     def _speak(self, sentence: str, turn_id: int) -> None:
-        self._state(PipelineState.SPEAKING, "播放中：正在播报回复", turn_id=turn_id)
+        self._event(TTS_STARTED, PipelineState.SPEAKING, "播放中：正在播报回复", turn_id=turn_id)
         segment = self.tts.synthesize(sentence)
         playback = self.transport.play_pcm(segment)
         if playback.error:
             raise playback.error
+        self._event(PLAYBACK_FINISHED, PipelineState.SPEAKING, "播放完成", turn_id=turn_id)
         self._drain_input_after_playback()
 
     def _recover(self, error: ProviderError, turn_id: int) -> TurnResult:
-        self.reporter.error(error.stage, error.code.value, error.message, turn_id=turn_id)
+        self._event(STAGE_ERROR, PipelineState.ERROR_RECOVERING, error.message, turn_id=turn_id, payload={"error": error})
-        self._state(PipelineState.ERROR_RECOVERING, "恢复待机：本轮已结束", turn_id=turn_id)
+        self._event(RECOVERING, PipelineState.ERROR_RECOVERING, "恢复待机：本轮已结束", turn_id=turn_id)
-        self._state(PipelineState.WAKE_LISTENING, "恢复待机：可继续唤醒", turn_id=turn_id)
+        self._event(STANDBY_RESUMED, PipelineState.WAKE_LISTENING, "恢复待机：可继续唤醒", turn_id=turn_id)
         return TurnResult(False, error=error, states=list(self._states))
 
-    def _state(self, state: PipelineState, message: str, *, turn_id: int) -> None:
+    def _event(
+        self,
+        event_type: str,
+        state: PipelineState,
+        message: str,
+        *,
+        turn_id: int,
+        payload: dict[str, object] | None = None,
+    ) -> None:
         self._states.append(state)
-        self.reporter.status(state.value, message, turn_id=turn_id)
+        self.event_bus.emit(event_type, turn_id=turn_id, state=state, message=message, payload=payload)
+
+    def _report_event(self, event: PipelineEvent) -> None:
+        handler = getattr(self.reporter, "handle_event", None)
+        if callable(handler):
+            handler(event)
+            return
+        dispatch_pipeline_event(self.reporter, event)
 
 
 def build_live_runtime(config: AppConfig, reporter: RuntimeReporter | None = None) -> LiveVoiceRuntime:

tests/test_live_runtime.py

@@ -4,6 +4,21 @@ import unittest
 
 from owner_voice_pet.config import AppConfig
 from owner_voice_pet.conversation import ConversationContext
+from owner_voice_pet.events import (
+    ACK_STARTED,
+    CAPTURE_STARTED,
+    LLM_STARTED,
+    PLAYBACK_FINISHED,
+    SPEECH_ENDED,
+    SPEECH_STARTED,
+    STANDBY_RESUMED,
+    STT_STARTED,
+    TRANSCRIPT_FINAL,
+    TTS_STARTED,
+    WAKE_DETECTED,
+    WAKE_LISTENING,
+    PipelineEventBus,
+)
 from owner_voice_pet.llm import MockLlmProvider
 from owner_voice_pet.models import AudioFrame, AudioSegment, Transcript
 from owner_voice_pet.runtime import LiveVoiceRuntime
@@ -79,6 +94,7 @@ def make_runtime(texts: list[str], context: ConversationContext | None = None) -
     llm = MockLlmProvider(["这是答复。"])
     tts = SineTtsProvider()
     reporter = RecordingReporter()
+    event_bus = PipelineEventBus()
     runtime = LiveVoiceRuntime(
         config=AppConfig(llm_api_key="secret", speech_provider="cloud", post_playback_drain_ms=0),
         transport=transport,
@@ -89,6 +105,7 @@ def make_runtime(texts: list[str], context: ConversationContext | None = None) -
         tts=tts,
         context=context or ConversationContext(),
         reporter=reporter,
+        event_bus=event_bus,
     )
     return runtime, stt, llm, transport, reporter
 
@@ -108,6 +125,23 @@ class LiveRuntimeTests(unittest.TestCase):
         self.assertLess(reporter.statuses.index("应答中：我在"), reporter.statuses.index("请说出问题"))
         self.assertLess(reporter.statuses.index("请说出问题"), reporter.statuses.index("录音中：正在听取问题"))
         self.assertIn("恢复待机：可继续唤醒", reporter.statuses[-1])
+        event_types = [event.type for event in runtime.event_bus.events]
+        expected_order = [
+            WAKE_LISTENING,
+            WAKE_DETECTED,
+            ACK_STARTED,
+            CAPTURE_STARTED,
+            SPEECH_STARTED,
+            SPEECH_ENDED,
+            STT_STARTED,
+            TRANSCRIPT_FINAL,
+            LLM_STARTED,
+            TTS_STARTED,
+            PLAYBACK_FINISHED,
+            STANDBY_RESUMED,
+        ]
+        positions = [event_types.index(item) for item in expected_order]
+        self.assertEqual(positions, sorted(positions))
 
     def test_temporary_context_is_sent_to_second_llm_call(self) -> None:
         runtime, _, llm, _, _ = make_runtime(["第一问", "第二问"])