[模拟麦克风验收]：完成自动化语音闭环自测，包含模拟音频输入、两轮Pipeline验证和fixture回放

README.md

@@ -99,6 +99,21 @@ python3.11 scripts/download_speech_models.py --dir models
 
 ## 运行
 
+模拟麦克风自动验收：
+
+```bash
+.venv/bin/python -m owner_voice_pet simulate-live --turns 2
+```
+
+这个命令不打开真实麦克风，会把生成的模拟麦克风音频帧喂进当前 `VoiceAssistantPipeline`，默认跑两轮唤醒到播放闭环。输出 JSON 中 `success=true` 表示两轮 wake、录音、实时字幕、final STT、临时上下文、LLM、TTS、播放和恢复待机都通过。模拟帧里故意包含 `家`、`家确` 这类短噪声 partial 和背景说话帧，用来验证实时字幕过滤和主说话人端点。
+
+需要复现同一组模拟输入时：
+
+```bash
+.venv/bin/python -m owner_voice_pet simulate-live --turns 2 --write-fixture /tmp/owner-simulated-mic.jsonl
+.venv/bin/python -m owner_voice_pet simulate-live --turns 2 --fixture /tmp/owner-simulated-mic.jsonl
+```
+
 单轮验收：
 
 ```bash
@@ -118,6 +133,7 @@ python3.11 scripts/download_speech_models.py --dir models
 ```bash
 .venv/bin/python -m compileall src tests scripts
 .venv/bin/python -m unittest discover -s tests
+.venv/bin/python -m owner_voice_pet simulate-live --turns 2
 .venv/bin/python -m owner_voice_pet acceptance
 .venv/bin/python -m owner_voice_pet validate-assets
 .venv/bin/python -m owner_voice_pet security-check

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

@@ -207,6 +207,36 @@ models/
 16. Local speech provider test validates `OWNER_SPEECH_PROVIDER=local` does not construct cloud ASR/TTS providers.
 17. Partial filter test validates single-character and short transient results are not emitted.
 18. CTC STT loading test validates manifest type selects `from_zipformer2_ctc` and does not require transducer encoder/decoder/joiner files.
+19. Simulated microphone live acceptance validates the current `VoiceAssistantPipeline` with generated wake frames, formal question frames, noisy short partials, background speech, two repeated turns, session context, TTS playback, and standby recovery.
+20. Fixture replay test validates the generated simulated microphone JSONL can be written and replayed for deterministic debugging.
+
+## Simulated Microphone Acceptance
+
+`owner_voice_pet simulate-live` is a deterministic no-device acceptance runner. It does not replace `run-live` with a real microphone, but it closes the gap between small unit tests and human testing by feeding realistic ordered audio frames into the same live pipeline controller.
+
+```text
+simulate-live
+  -> generated AudioFrame sequence
+  -> BoundedMemoryAudioTransport(simulated microphone)
+  -> VoiceAssistantPipeline
+  -> KeywordWakeWordProvider(metadata wake)
+  -> SimulatedNoiseFilter(metadata + PCM passthrough)
+  -> PrimarySpeakerVadRecorder
+  -> MetadataSttProvider(partial + final)
+  -> MockLlmProvider
+  -> SineTtsProvider
+  -> JSON checks
+```
+
+The generated sequence intentionally contains:
+
+1. one wake frame per turn;
+2. six primary speaker frames per question, enough to establish the temporary profile;
+3. short noisy partials `家` and `家确`, followed by stable question partials;
+4. fifteen background speech frames after the primary speaker stops, enough to trigger the 300 ms primary-speaker absence endpoint;
+5. a second wake/question sequence to prove recovery to standby and temporary context continuity.
+
+The transport is bounded: if frames are exhausted unexpectedly, it raises a structured validation error instead of letting the pipeline loop forever. `--write-fixture` writes the exact simulated microphone frames as JSONL, and `--fixture` replays them for repeatable debugging.
 
 ## Migration
 

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

@@ -56,6 +56,7 @@
 14. 语音链路一致性问题：当前 `OWNER_SPEECH_PROVIDER=cloud` 时 final STT/TTS 走云端，而 partial 走本地模型；同一轮对话里 partial 和 final 来自不同模型，容易出现“实时字幕和最终转写明显冲突”的体验。
 15. 降噪缺失问题：正式问题录音直接把原始麦克风帧送入 VAD、partial STT 和 final STT，背景噪声会同时影响端点、实时字幕和最终识别。
 16. 本地模型落后问题：默认 STT 仍是 2023 年 14M 小模型，适合最小验收但不适合作为默认实时字幕质量基线；应升级为 sherpa-onnx 官方 2025 中文 CTC int8 模型。
+17. 自测闭环问题：现有 `acceptance` 只覆盖旧单轮 pipeline，不能证明当前 `VoiceAssistantPipeline` 在“模拟麦克风 -> wake -> capture -> partial -> final -> LLM -> TTS -> standby -> 第二轮”路径上完整正常；真人测试前缺少可重复的自动调试入口。
 
 ## 详细需求
 
@@ -91,6 +92,11 @@
 28. 降噪 provider 失败 SHALL 作为结构化可恢复错误进入 `stage_error -> recovering -> standby`，不得把未经标记的半处理音频写入对话上下文。
 29. partial transcript SHALL 增加稳定过滤：不得显示单个中文/英文有效字符，不得重复显示同一文本，不得把极短的瞬态跳变作为终端实时字幕输出。
 30. final transcript SHALL 是唯一进入 LLM 的用户文本；降噪帧、partial 文本、denoiser metadata 和 ASR raw metadata 均不得进入 `ConversationContext`。
+31. 系统 SHALL 提供 `owner_voice_pet simulate-live` 命令，用模拟麦克风帧驱动当前 `VoiceAssistantPipeline`，默认完成两轮 wake-to-playback turn。
+32. `simulate-live` SHALL 输出 JSON 检查项，至少包含完成轮数、失败轮数、wake/speech/STT/LLM/TTS/standby 事件计数、final transcripts、partial 噪声过滤、第二轮临时上下文和播放段数。
+33. 模拟麦克风输入 SHALL 包含 wake 帧、正式问题主说话人帧、短噪声 partial、背景噪声/非主说话人帧和第二轮重复唤醒帧。
+34. 模拟 transport SHALL 有边界保护：如果帧提前耗尽，命令必须结构化失败并退出，不能无限等待。
+35. `simulate-live` SHALL 支持写入和回放 JSONL fixture，便于后续持续复现同一组模拟麦克风输入。
 
 ### 非功能需求
 
@@ -332,6 +338,9 @@ standby
 | CTC 模型比旧 14M 模型更大导致下载慢 | 中 | 中 | 仍放 `models/` 并跳过已存在文件；README 明确首次下载较久；下载脚本保留重试 |
 | partial 过滤过严导致实时字幕少显示 | 中 | 低 | 过滤只影响用户可见 partial，不影响 final STT 和 LLM；后续可配置更细阈值 |
 | 降噪失败后用户无法继续本轮 | 低 | 中 | 当前 turn 结构化失败并恢复待机，避免错误音频进入 LLM；下一轮可继续唤醒 |
+| 模拟验收与真实麦克风仍有差异 | 中 | 中 | 明确模拟验收用于自动调试 live pipeline 状态机和事件顺序；仍保留 4.3 真人 `run-live` 验收，不在用户确认前归档 |
+| 模拟帧耗尽导致测试卡死 | 中 | 高 | 使用 bounded simulated transport，空读超过阈值直接抛结构化错误并让命令返回非 0 |
+| 模拟 provider 掩盖真实模型加载问题 | 中 | 中 | 模拟命令只验证 pipeline；模型加载仍由 `model-check` 真实加载 KWS/VAD/STT/denoiser 覆盖 |
 
 ## 任务分解
 
@@ -397,6 +406,14 @@ standby
 - [ ] 10.7 下载/校验新本地模型并执行门禁；前置条件：10.2 至 10.6 完成；验收标准：下载脚本、compileall、unittest、security-check、model-check、device-check、OpenSpec strict、git diff check 通过或记录真实设备失败原因；测试要点：输出不泄露 key，模型文件不进 Git；优先级：P0；预计：60 分钟。
 - [ ] 10.8 提交“本地语音降噪”模块；前置条件：10.7 门禁通过；验收标准：中文 commit 信息为 `[本地语音降噪]：完成本地语音链路和噪音过滤，包含降噪模型、本地ASR和实时字幕稳定策略`，提交后除本地 `.env` 非提交修改外无未提交源码/文档中间状态；优先级：P0；预计：10 分钟。
 
+### 11. 模拟麦克风自动验收与自调试
+
+- [ ] 11.1 更新 OpenSpec 描述模拟麦克风验收；前置条件：用户要求“自己一直测试到完整正常，自己弄模拟音频给麦克风”；验收标准：proposal/design/spec/tasks 明确新增可重复 `simulate-live`，覆盖两轮 live pipeline、实时字幕、降噪、上下文和恢复待机；测试要点：OpenSpec strict；优先级：P0；预计：35 分钟。
+- [ ] 11.2 实现模拟麦克风帧生成和 bounded transport；前置条件：live pipeline 已 stage 化；验收标准：模拟输入包含 wake、正式问题、短噪声 partial、背景噪声、两轮 turn，空队列不会无限等待；测试要点：失败时命令返回非 0；优先级：P0；预计：50 分钟。
+- [ ] 11.3 新增 `owner_voice_pet simulate-live` CLI；前置条件：11.2 完成；验收标准：默认两轮模拟，输出 JSON checks，可写入/回放 JSONL fixture；测试要点：CLI 单测和真实命令；优先级：P0；预计：40 分钟。
+- [ ] 11.4 补充 README 和自动化测试；前置条件：11.3 完成；验收标准：README 包含模拟验收命令，测试验证两轮闭环、partial 噪声过滤、第二轮临时上下文、fixture 写入/回放；测试要点：unittest；优先级：P0；预计：35 分钟。
+- [ ] 11.5 执行全量门禁并提交“模拟麦克风验收”模块；前置条件：11.1 至 11.4 完成；验收标准：`simulate-live --turns 2`、compileall、unittest、security-check、model-check、device-check、OpenSpec strict、git diff check 通过后中文 commit；优先级：P0；预计：45 分钟。
+
 ## Spec Deltas
 
 ### 新增能力
@@ -416,6 +433,7 @@ standby
 9. `Low latency capture and first utterance preservation`：新增 ACK 后不额外丢弃正式问题、批量读帧、独立画像就绪阈值和快速主说话人端点要求。
 10. `Realtime partial transcript output`：新增录音期间 partial transcript 事件、终端显示和上下文隔离要求。
 11. `Local voice chain and noise filtering`：新增本地 STT/TTS 默认、GTCRN 降噪、CTC 模型、partial 稳定过滤和音频不上传要求。
+12. `Simulated microphone live acceptance`：新增不依赖真人麦克风的 live pipeline 模拟验收，覆盖两轮重复对话、短噪声过滤、背景噪声端点、fixture 写入/回放和空帧防卡死。
 
 ### 删除项
 
@@ -436,6 +454,7 @@ standby
 7. M7：低延迟端点与首句保留修正完成后提交，保留真人 `run-live` 验收任务，不在用户确认前归档。
 8. M8：录音期间实时转写显示完成后提交，保留真人 `run-live` 验收任务，不在用户确认前归档。
 9. M9：本地语音链路、高质量实时字幕和正式问题降噪完成后提交，继续保留真人 `run-live` 验收任务，不在用户确认前归档。
+10. M10：模拟麦克风自动验收完成后提交，作为真人验收前的可重复自测入口；仍不归档变更，直到真实 `run-live` 行为由用户确认。
 
 估时：
 

openspec/changes/separate-wake-and-realtime-transcript/specs/voice-pet-pipeline/spec.md

@@ -105,6 +105,31 @@ The live runtime SHALL run wake, VAD, STT, realtime transcript, TTS, and capture
 - **WHEN** local noise filtering and local STT run during capture
 - **THEN** raw PCM, denoised PCM, VAD features, partial transcript metadata, and temporary speaker profile data SHALL remain process-local and SHALL NOT be sent to the cloud LLM
 
+### Requirement: Simulated microphone live acceptance
+The system SHALL provide a deterministic simulated microphone acceptance command that exercises the current live `VoiceAssistantPipeline` without requiring a human to speak into a physical microphone.
+
+#### Scenario: Simulated live command completes repeated turns
+- **WHEN** the developer runs `.venv/bin/python -m owner_voice_pet simulate-live --turns 2`
+- **THEN** the command SHALL feed simulated microphone audio frames into the live pipeline and complete two wake-to-playback turns
+- **AND** the JSON output SHALL report two completed turns, zero failed turns, two final transcripts, two LLM calls, at least two playback completions, and standby resumed after each turn
+
+#### Scenario: Simulated input includes noisy realtime partials
+- **WHEN** simulated formal question frames contain short noisy partial text such as `家` or `家确`
+- **THEN** realtime transcript output SHALL filter those short noise partials and SHALL only report stable question text
+
+#### Scenario: Simulated input includes background speech after the user stops
+- **WHEN** simulated background speech remains after the primary speaker frames end
+- **THEN** primary speaker endpointing SHALL close the formal utterance and SHALL NOT merge later simulated turns into the current question
+
+#### Scenario: Simulated command can write and replay fixtures
+- **WHEN** the developer passes `--write-fixture path.jsonl`
+- **THEN** the command SHALL write the generated simulated microphone frames as JSONL
+- **AND** a later run using `--fixture path.jsonl` SHALL replay the same frames and produce equivalent successful live acceptance results
+
+#### Scenario: Simulated microphone frames are exhausted unexpectedly
+- **WHEN** the simulated microphone transport runs out of frames before the expected turn completes
+- **THEN** the command SHALL fail with a structured error instead of waiting indefinitely
+
 ### Requirement: Wake acknowledgement before recording
 The live runtime SHALL provide an audible local acknowledgement after local wake detection and before it starts recording the user's formal question.
 

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

@@ -81,3 +81,11 @@
 - [x] 10.6 更新 README、`.env.example` 和本地 `.env` 非密钥配置；前置条件：10.5 完成；验收标准：中文运行说明包含本地语音默认值、降噪开关、模型下载、model-check、run-live 验收；测试要点：命令可复制，`.env` 不进入提交；优先级：P0；预计：35 分钟。
 - [x] 10.7 下载/校验新本地模型并执行门禁；前置条件：10.2 至 10.6 完成；验收标准：下载脚本、compileall、unittest、security-check、model-check、device-check、OpenSpec strict、git diff check 通过或记录真实设备失败原因；测试要点：输出不泄露 key，模型文件不进 Git；优先级：P0；预计：60 分钟。
 - [x] 10.8 提交“本地语音降噪”模块；前置条件：10.7 门禁通过；验收标准：中文 commit 信息为 `[本地语音降噪]：完成本地语音链路和噪音过滤，包含降噪模型、本地ASR和实时字幕稳定策略`，提交后除本地 `.env` 非提交修改外无未提交源码/文档中间状态；优先级：P0；预计：10 分钟。
+
+## 11. 模拟麦克风自动验收与自调试
+
+- [x] 11.1 更新 OpenSpec 描述模拟麦克风验收；前置条件：用户要求“自己一直测试到完整正常，自己弄模拟音频给麦克风”；验收标准：proposal/design/spec/tasks 明确新增可重复 `simulate-live`，覆盖两轮 live pipeline、实时字幕、降噪、上下文和恢复待机；测试要点：OpenSpec strict；优先级：P0；预计：35 分钟。
+- [x] 11.2 实现模拟麦克风帧生成和 bounded transport；前置条件：live pipeline 已 stage 化；验收标准：模拟输入包含 wake、正式问题、短噪声 partial、背景噪声、两轮 turn，空队列不会无限等待；测试要点：失败时命令返回非 0；优先级：P0；预计：50 分钟。
+- [x] 11.3 新增 `owner_voice_pet simulate-live` CLI；前置条件：11.2 完成；验收标准：默认两轮模拟，输出 JSON checks，可写入/回放 JSONL fixture；测试要点：CLI 单测和真实命令；优先级：P0；预计：40 分钟。
+- [x] 11.4 补充 README 和自动化测试；前置条件：11.3 完成；验收标准：README 包含模拟验收命令，测试验证两轮闭环、partial 噪声过滤、第二轮临时上下文、fixture 写入/回放；测试要点：unittest；优先级：P0；预计：35 分钟。
+- [x] 11.5 执行全量门禁并提交“模拟麦克风验收”模块；前置条件：11.1 至 11.4 完成；验收标准：`simulate-live --turns 2`、compileall、unittest、security-check、model-check、device-check、OpenSpec strict、git diff check 通过后中文 commit；优先级：P0；预计：45 分钟。

src/owner_voice_pet/__init__.py

@@ -26,6 +26,7 @@ from .conversation import ConversationContext
 from .llm import MockLlmProvider, OpenAICompatibleLlmProvider
 from .pipeline import PipelineResult, VoicePipeline
 from .runtime import LiveVoiceRuntime, RuntimeSummary, TerminalRuntimeReporter, TurnResult, build_live_runtime
+from .simulation import run_simulated_live
 from .tts import CloudTtsProvider, MacSayTtsProvider, SentenceBuffer, SineTtsProvider
 from .assets import validate_pet_assets
 from .ui import ConsolePetWindow, PetStateController, PetVisualState
@@ -64,6 +65,7 @@ __all__ = [
     "TerminalRuntimeReporter",
     "TurnResult",
     "build_live_runtime",
+    "run_simulated_live",
     "CloudTtsProvider",
     "MacSayTtsProvider",
     "SentenceBuffer",

src/owner_voice_pet/cli.py

@@ -14,6 +14,7 @@ from .llm import MockLlmProvider, OpenAICompatibleLlmProvider
 from .models import AudioFrame, ProviderError
 from .pipeline import VoicePipeline
 from .runtime import build_live_runtime
+from .simulation import run_simulated_live
 from .speech_models import check_speech_models, model_status_errors
 from .stt import MetadataSttProvider, SherpaOnnxSttProvider
 from .transport import MemoryAudioTransport, sounddevice_device_report
@@ -35,6 +36,10 @@ def main(argv: list[str] | None = None) -> int:
     subparsers.add_parser("device-check", help="Validate local microphone and speaker availability")
     live = subparsers.add_parser("run-live", help="Run real repeated live voice conversation")
     live.add_argument("--once", action="store_true", help="Run one completed live turn and exit")
+    simulate = subparsers.add_parser("simulate-live", help="Run live pipeline with simulated microphone frames")
+    simulate.add_argument("--turns", type=int, default=2, help="Number of simulated turns. Default: 2")
+    simulate.add_argument("--fixture", default=None, help="Replay simulated microphone frames from JSONL")
+    simulate.add_argument("--write-fixture", default=None, help="Write generated simulated microphone frames to JSONL")
     smoke = subparsers.add_parser("llm-smoke", help="Call configured OpenAI/NewAPI endpoint")
     smoke.add_argument("--message", default="用一句中文回复：小杰在线。")
     smoke.add_argument("--no-stream", action="store_true")
@@ -141,6 +146,19 @@ def main(argv: list[str] | None = None) -> int:
             return 1
         return 0 if summary.completed_turns > 0 or summary.interrupted else 1
 
+    if args.command == "simulate-live":
+        try:
+            data = run_simulated_live(
+                turns=args.turns,
+                fixture_path=args.fixture,
+                write_fixture=args.write_fixture,
+            )
+        except (ProviderError, ValueError) as exc:
+            print(json.dumps({"success": False, "error": str(exc)}, ensure_ascii=False, sort_keys=True))
+            return 1
+        print(json.dumps(data, ensure_ascii=False, sort_keys=True))
+        return 0 if data["success"] else 1
+
     if args.command == "acceptance":
         result = run_acceptance()
         print(json.dumps(result, ensure_ascii=False, sort_keys=True))

src/owner_voice_pet/simulation.py

@@ -0,0 +1,283 @@
+from __future__ import annotations
+
+import math
+import struct
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from .audio_preprocess import NoopAudioPreprocessor
+from .assistant_pipeline import VoiceAssistantPipeline
+from .config import AppConfig
+from .conversation import ConversationContext
+from .events import (
+    LLM_STARTED,
+    PLAYBACK_FINISHED,
+    SPEECH_ENDED,
+    SPEECH_STARTED,
+    STANDBY_RESUMED,
+    STT_STARTED,
+    TRANSCRIPT_FINAL,
+    TRANSCRIPT_PARTIAL,
+    TTS_STARTED,
+    WAKE_DETECTED,
+    WAKE_LISTENING,
+)
+from .llm import MockLlmProvider
+from .models import AudioFrame, AudioSegment, ErrorCode, ProviderError
+from .stt import MetadataSttProvider
+from .transport import FileReplayTransport, MemoryAudioTransport
+from .tts import SineTtsProvider
+from .vad import EnergyVadProvider, PrimarySpeakerVadRecorder
+from .wakeword import KeywordWakeWordProvider
+
+
+@dataclass(slots=True)
+class SimulationReporter:
+    statuses: list[str] = field(default_factory=list)
+    partials: list[str] = field(default_factory=list)
+    finals: list[str] = field(default_factory=list)
+    errors: list[str] = field(default_factory=list)
+
+    def status(self, state: str, message: str, *, turn_id: int | None = None) -> None:
+        prefix = f"第{turn_id}轮:" if turn_id is not None else ""
+        self.statuses.append(f"{prefix}{message}")
+
+    def transcript(self, text: str, *, final: bool, turn_id: int | None = None) -> None:
+        if final:
+            self.finals.append(text)
+        else:
+            self.partials.append(text)
+
+    def error(self, stage: str, code: str, message: str, *, turn_id: int | None = None) -> None:
+        self.errors.append(f"{stage}:{code}:{message}")
+
+
+class BoundedMemoryAudioTransport(MemoryAudioTransport):
+    def __init__(self, frames: list[AudioFrame], *, max_empty_reads: int = 5) -> None:
+        super().__init__(frames)
+        self.max_empty_reads = max_empty_reads
+        self.empty_reads = 0
+
+    def read_frames(self, timeout_ms: int) -> list[AudioFrame]:
+        frames = super().read_frames(timeout_ms)
+        if frames:
+            self.empty_reads = 0
+            return frames
+        self.empty_reads += 1
+        if self.empty_reads > self.max_empty_reads:
+            raise ProviderError(
+                ErrorCode.VALIDATION_FAILED,
+                "simulated microphone frames were exhausted",
+                False,
+                "simulated-microphone",
+                "transport",
+            )
+        return []
+
+
+class SimulatedNoiseFilter(NoopAudioPreprocessor):
+    def __init__(self) -> None:
+        self.loaded = False
+        self.processed_frames = 0
+
+    def load(self) -> None:
+        self.loaded = True
+
+    def reset(self) -> None:
+        return None
+
+    def process_frame(self, frame: AudioFrame) -> AudioFrame:
+        self.processed_frames += 1
+        metadata = dict(frame.metadata)
+        metadata["denoised"] = True
+        metadata["noise_filter_provider"] = "simulated"
+        return AudioFrame(
+            pcm=frame.pcm,
+            sample_rate=frame.sample_rate,
+            channels=frame.channels,
+            timestamp_ms=frame.timestamp_ms,
+            frame_id=frame.frame_id,
+            metadata=metadata,
+        )
+
+
+def run_simulated_live(
+    *,
+    turns: int = 2,
+    fixture_path: str | Path | None = None,
+    write_fixture: str | Path | None = None,
+) -> dict[str, Any]:
+    if turns <= 0:
+        raise ValueError("turns must be positive")
+    frames = FileReplayTransport.from_jsonl(fixture_path)._frames if fixture_path else _simulated_turn_frames(turns)
+    frame_list = list(frames)
+    if write_fixture:
+        FileReplayTransport.write_jsonl(write_fixture, frame_list)
+
+    transport = BoundedMemoryAudioTransport(frame_list)
+    reporter = SimulationReporter()
+    preprocessor = SimulatedNoiseFilter()
+    llm = MockLlmProvider(["这是模拟回复。"])
+    config = AppConfig(
+        llm_api_key="simulated",
+        speech_provider="local",
+        realtime_transcript_enabled=True,
+        noise_filter_enabled=True,
+        post_playback_drain_ms=0,
+        endpoint_mode="primary_speaker",
+        speaker_profile_ms=120,
+        speaker_profile_min_ms=120,
+        speaker_absent_ms=300,
+        vad_min_duration_ms=250,
+        vad_end_silence_ms=350,
+        vad_no_speech_timeout_ms=3000,
+        vad_max_recording_ms=6000,
+    )
+    pipeline = VoiceAssistantPipeline(
+        config=config,
+        transport=transport,
+        wakeword=KeywordWakeWordProvider(threshold=0.5),
+        vad_recorder=PrimarySpeakerVadRecorder(
+            EnergyVadProvider(),
+            min_duration_ms=config.vad_min_duration_ms,
+            end_silence_ms=config.vad_end_silence_ms,
+            no_speech_timeout_ms=config.vad_no_speech_timeout_ms,
+            max_recording_ms=config.vad_max_recording_ms,
+            speaker_profile_ms=config.speaker_profile_ms,
+            speaker_profile_min_ms=config.speaker_profile_min_ms,
+            speaker_absent_ms=config.speaker_absent_ms,
+            similarity_threshold=config.speaker_similarity_threshold,
+            min_rms=config.speaker_min_rms,
+        ),
+        audio_preprocessor=preprocessor,
+        stt=MetadataSttProvider(),
+        realtime_stt=MetadataSttProvider(),
+        llm=llm,
+        tts=SineTtsProvider(),
+        ack_tts=SineTtsProvider(),
+        context=ConversationContext(),
+        reporter=reporter,
+    )
+    completed_turns = 0
+    failed_turns = 0
+    pipeline.load()
+    transport.start_input(sample_rate=config.sample_rate, channels=config.channels)
+    try:
+        for turn_id in range(1, turns + 1):
+            result = pipeline.run_turn(turn_id)
+            if result.success:
+                completed_turns += 1
+                continue
+            failed_turns += 1
+            break
+    finally:
+        pipeline.shutdown()
+    event_types = [event.type for event in pipeline.event_bus.events]
+    expected_transcripts = [f"第{index}轮模拟问题" for index in range(1, turns + 1)]
+    checks = {
+        "completed_turns": completed_turns == turns,
+        "no_failed_turns": failed_turns == 0,
+        "wake_per_turn": event_types.count(WAKE_DETECTED) == turns,
+        "speech_per_turn": event_types.count(SPEECH_STARTED) == turns and event_types.count(SPEECH_ENDED) == turns,
+        "stt_per_turn": event_types.count(STT_STARTED) == turns and event_types.count(TRANSCRIPT_FINAL) == turns,
+        "llm_per_turn": event_types.count(LLM_STARTED) == turns and len(llm.calls) == turns,
+        "tts_per_turn": event_types.count(TTS_STARTED) == turns and event_types.count(PLAYBACK_FINISHED) >= turns,
+        "standby_per_turn": event_types.count(STANDBY_RESUMED) == turns,
+        "transcripts_match": reporter.finals == expected_transcripts,
+        "partial_noise_filtered": "家" not in reporter.partials and "家确" not in reporter.partials,
+        "denoised_capture": preprocessor.loaded and preprocessor.processed_frames > 0,
+        "context_in_second_turn": turns < 2 or _second_turn_has_first_history(llm.calls, expected_transcripts[0]),
+    }
+    return {
+        "success": all(checks.values()) and not reporter.errors,
+        "turns": turns,
+        "completed_turns": completed_turns,
+        "failed_turns": failed_turns,
+        "checks": checks,
+        "errors": reporter.errors,
+        "partials": reporter.partials,
+        "final_transcripts": reporter.finals,
+        "played_segments": len(transport.played_segments),
+        "llm_calls": len(llm.calls),
+        "event_types": event_types,
+    }
+
+
+def _second_turn_has_first_history(calls: list[list[Any]], first_user_text: str) -> bool:
+    if len(calls) < 2:
+        return False
+    return any(getattr(message, "content", "") == first_user_text for message in calls[1])
+
+
+def _simulated_turn_frames(turns: int) -> list[AudioFrame]:
+    frames: list[AudioFrame] = []
+    frame_id = 0
+    timestamp_ms = 0
+    for index in range(1, turns + 1):
+        frames.append(
+            _frame(
+                frame_id,
+                timestamp_ms,
+                frequency=880.0,
+                metadata={"duration_ms": 20, "wake_word": "小杰小杰", "wake_confidence": 0.99},
+            )
+        )
+        frame_id += 1
+        timestamp_ms += 20
+        question = f"第{index}轮模拟问题"
+        for speech_index in range(6):
+            partial = "家" if speech_index == 0 else "家确" if speech_index == 1 else question
+            metadata: dict[str, object] = {
+                "duration_ms": 20,
+                "speech": True,
+                "speaker_id": "owner",
+                "partial_transcript": partial,
+            }
+            if speech_index == 0:
+                metadata["transcript"] = question
+            frames.append(
+                _frame(
+                    frame_id,
+                    timestamp_ms,
+                    frequency=440.0,
+                    metadata=metadata,
+                )
+            )
+            frame_id += 1
+            timestamp_ms += 20
+        for _ in range(15):
+            frames.append(
+                _frame(
+                    frame_id,
+                    timestamp_ms,
+                    frequency=180.0,
+                    amplitude=500,
+                    metadata={
+                        "duration_ms": 20,
+                        "speech": True,
+                        "speaker_id": "background",
+                        "partial_transcript": "家确",
+                    },
+                )
+            )
+            frame_id += 1
+            timestamp_ms += 20
+    return frames
+
+
+def _frame(
+    frame_id: int,
+    timestamp_ms: int,
+    *,
+    frequency: float,
+    amplitude: int = 8000,
+    metadata: dict[str, object] | None = None,
+) -> AudioFrame:
+    sample_rate = 16000
+    samples = int(sample_rate * 0.02)
+    pcm = bytearray()
+    for index in range(samples):
+        value = int(math.sin(2 * math.pi * frequency * index / sample_rate) * amplitude)
+        pcm.extend(struct.pack("<h", value))
+    return AudioFrame(bytes(pcm), sample_rate, 1, timestamp_ms, frame_id, metadata or {"duration_ms": 20})

tests/test_cli_acceptance.py

@@ -103,6 +103,14 @@ class CliAcceptanceTests(unittest.TestCase):
         self.assertEqual(code, 0)
         self.assertTrue(fake_runtime.once)
 
+    def test_simulate_live_command_runs_two_turns(self) -> None:
+        code, data = self.call("simulate-live", "--turns", "2")
+        self.assertEqual(code, 0)
+        self.assertTrue(data["success"])
+        self.assertEqual(data["completed_turns"], 2)
+        self.assertEqual(data["final_transcripts"], ["第1轮模拟问题", "第2轮模拟问题"])
+        self.assertTrue(data["checks"]["partial_noise_filtered"])
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_simulation.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from owner_voice_pet.simulation import run_simulated_live
+
+
+class SimulationTests(unittest.TestCase):
+    def test_simulated_live_runs_two_turns_with_context_and_noise_filter(self) -> None:
+        result = run_simulated_live(turns=2)
+
+        self.assertTrue(result["success"], result)
+        self.assertEqual(result["completed_turns"], 2)
+        self.assertEqual(result["failed_turns"], 0)
+        self.assertEqual(result["final_transcripts"], ["第1轮模拟问题", "第2轮模拟问题"])
+        self.assertNotIn("家", result["partials"])
+        self.assertNotIn("家确", result["partials"])
+        self.assertTrue(result["checks"]["context_in_second_turn"])
+        self.assertTrue(result["checks"]["denoised_capture"])
+        self.assertEqual(result["checks"]["standby_per_turn"], True)
+
+    def test_simulated_live_can_write_and_replay_fixture(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            fixture = Path(tmp) / "simulated-mic.jsonl"
+            generated = run_simulated_live(turns=1, write_fixture=fixture)
+            replayed = run_simulated_live(turns=1, fixture_path=fixture)
+
+        self.assertTrue(generated["success"], generated)
+        self.assertTrue(replayed["success"], replayed)
+        self.assertEqual(replayed["final_transcripts"], ["第1轮模拟问题"])
+
+
+if __name__ == "__main__":
+    unittest.main()