[实时字幕端点]：完成无新文字快速结束录音，包含1.5秒停滞配置、Capture端点和回归测试

tests/test_live_runtime.py

@@ -47,6 +47,28 @@ def segment_frames(start_id: int, start_ms: int, partials: list[str] | None = No
     ]
 
 
+def long_speech_frames_with_stale_partial(start_id: int, start_ms: int, duration_ms: int) -> list[AudioFrame]:
+    frames: list[AudioFrame] = []
+    for index in range(duration_ms // 20):
+        metadata: dict[str, object] = {
+            "duration_ms": 20,
+            "speech": True,
+            "partial_transcript": "你知道",
+            "transcript": "你知道我在说什么吗",
+        }
+        frames.append(
+            AudioFrame(
+                b"\xff\x7f",
+                16000,
+                1,
+                start_ms + index * 20,
+                start_id + index,
+                metadata,
+            )
+        )
+    return frames
+
+
 def wake_frame(frame_id: int, timestamp_ms: int) -> AudioFrame:
     return AudioFrame(
         b"\xff\x7f",
@@ -266,6 +288,47 @@ class LiveRuntimeTests(unittest.TestCase):
         self.assertLess(event_types.index(TRANSCRIPT_PARTIAL), event_types.index(SPEECH_ENDED))
         self.assertLess(event_types.index(TRANSCRIPT_PARTIAL), event_types.index(TRANSCRIPT_FINAL))
 
+    def test_realtime_transcript_idle_ends_current_utterance(self) -> None:
+        frames = [wake_frame(0, 0)]
+        frames.extend(long_speech_frames_with_stale_partial(1, 20, 2200))
+        transport = MemoryAudioTransport(frames, flush_clears_input=False)
+        stt = QueueSttProvider(["你知道我在说什么吗"])
+        llm = MockLlmProvider(["这是答复。"])
+        reporter = RecordingReporter()
+        runtime = VoiceAssistantPipeline(
+            config=AppConfig(
+                llm_api_key="secret",
+                speech_provider="cloud",
+                wake_ack_text="",
+                realtime_transcript_idle_timeout_ms=1500,
+            ),
+            transport=transport,
+            wakeword=KeywordWakeWordProvider(),
+            vad_recorder=VadRecorder(
+                EnergyVadProvider(),
+                min_duration_ms=40,
+                end_silence_ms=5000,
+                max_recording_ms=10000,
+            ),
+            stt=stt,
+            audio_preprocessor=MarkerAudioPreprocessor(partial_text="你知道"),
+            realtime_stt=MetadataSttProvider(),
+            llm=llm,
+            tts=SineTtsProvider(),
+            context=ConversationContext(),
+            reporter=reporter,
+            event_bus=PipelineEventBus(),
+        )
+
+        summary = runtime.run(max_turns=1)
+
+        self.assertEqual(summary.completed_turns, 1)
+        self.assertEqual(reporter.partials, ["你知道"])
+        self.assertEqual(reporter.transcripts, ["你知道我在说什么吗"])
+        self.assertEqual(len(stt.calls), 1)
+        self.assertEqual(stt.calls[0].metadata["end_reason"], "partial_transcript_idle")
+        self.assertLess(stt.calls[0].duration_ms, 1700)
+
     def test_capture_uses_denoised_frames_for_partial_and_final_stt(self) -> None:
         preprocessor = MarkerAudioPreprocessor(partial_text="降噪后问题")
         runtime, stt, _, _, reporter = make_runtime(