fix(voice): stabilize v2 streaming topic handling

services/ai_orchestrator_service/voice.py

@@ -787,7 +787,7 @@ def _voice_text_key(text: str | None) -> str:
 def _voice_recent_caller_texts(
     transcript_window: list[VoiceTranscriptSegmentRow],
     *,
-    limit: int = 4,
+    limit: int = 8,
 ) -> list[str]:
     caller_texts = [str(segment.text or "").strip() for segment in transcript_window if segment.speaker == "caller"]
     return caller_texts[-max(limit, 1) :]
@@ -1256,12 +1256,24 @@ def _voice_caller_context_before_current(caller_texts: list[str], transcript_tex
     return caller_texts
 
 
+def _voice_service_context_texts(caller_texts: list[str]) -> list[str]:
+    return [text for text in caller_texts if _voice_has_service_topic(text)]
+
+
 def _voice_is_midcall_greeting_reply(text: str | None) -> bool:
     normalized = _voice_text_key(text)
     if not normalized:
         return False
     greeting_markers = ("здравствуйте", "сәлеметсіз", "сәлем")
-    followup_markers = ("чем помочь", "как я могу помочь", "коротко расскажите", "қалай көмектесе")
+    followup_markers = (
+        "чем помочь",
+        "как я могу помочь",
+        "коротко расскажите",
+        "о чем именно",
+        "что именно",
+        "что вас интересует",
+        "қалай көмектесе",
+    )
     return any(marker in normalized for marker in greeting_markers) and any(
         marker in normalized for marker in followup_markers
     )
@@ -1308,15 +1320,27 @@ def _voice_postprocess_reply_text(
         return normalized_reply
     caller_texts = _voice_recent_caller_texts(transcript_window)
     prior_caller_texts = _voice_caller_context_before_current(caller_texts, transcript_text)
+    active_topic_texts = _voice_service_context_texts(prior_caller_texts or caller_texts)
+    active_topic_prompt = _voice_topic_prompt(language, active_topic_texts) if active_topic_texts else None
     if _voice_is_hearing_check_caller_text(transcript_text):
         return _voice_hearing_check_reply(language, prior_caller_texts or caller_texts)
     if _voice_contains_false_lookup_promise(normalized_reply) and not kb_results:
-        return _voice_topic_prompt(language, caller_texts) or _voice_generic_prompt(language)
+        return active_topic_prompt or _voice_topic_prompt(language, caller_texts) or _voice_generic_prompt(language)
     if _voice_is_midcall_greeting_reply(normalized_reply) and any(
         segment.speaker == "assistant" for segment in transcript_window
     ):
         context_texts = prior_caller_texts if _voice_is_low_signal_caller_text(transcript_text) else caller_texts
         return _voice_topic_prompt(language, context_texts) or _voice_generic_prompt(language)
+    if (
+        active_topic_prompt
+        and not _voice_has_service_topic(transcript_text)
+        and (
+            "тариф" in _voice_text_key(normalized_reply)
+            or "услуг" in _voice_text_key(normalized_reply)
+            or "как я могу помочь" in _voice_text_key(normalized_reply)
+        )
+    ):
+        return active_topic_prompt
     return normalized_reply
 
 

services/ai_voice_runtime_service/media_runtime.py

@@ -107,6 +107,7 @@ class MediaActor:
     last_ack_completed_monotonic: float = 0.0
     last_ack_variant: str | None = None
     current_reply_phase: str | None = None
+    finalized_caller_turn_count: int = 0
 
 
 class AudioSocketMediaRuntime:
@@ -266,6 +267,20 @@ class AudioSocketMediaRuntime:
         normalized = str(language or registration.language or "").strip().lower()
         return normalized.startswith("ru")
 
+    def _should_emit_partial_ack(self, actor: MediaActor, partial_transcript: str, partial_intent: str) -> bool:
+        transcript_text = str(partial_transcript or "").strip()
+        if not transcript_text or self._is_low_signal_transcript(transcript_text):
+            return False
+        normalized_intent = str(partial_intent or "").strip() or "unknown"
+        if actor.finalized_caller_turn_count > 0:
+            return True
+        return normalized_intent != "unknown"
+
+    def _should_emit_blind_ack(self, actor: MediaActor, pcm_bytes: bytes) -> bool:
+        if len(pcm_bytes) < self._immediate_ack_min_bytes:
+            return False
+        return actor.finalized_caller_turn_count > 0
+
     @staticmethod
     def _base_ack_variants(language: str | None, ack_kind: str) -> tuple[str, ...]:
         normalized = str(language or "").strip().lower()
@@ -929,7 +944,6 @@ class AudioSocketMediaRuntime:
                     actor.asr_stream_id,
                     pcm_frame,
                 )
-                await self._poll_streaming_partial(actor)
             except StreamingASRUnavailable as exc:
                 logger.warning(
                     "audiosocket.streaming_asr_push_failed session_id=%s error=%s",
@@ -937,6 +951,19 @@ class AudioSocketMediaRuntime:
                     str(exc)[:500],
                 )
                 await self._close_streaming_asr(actor)
+            else:
+                try:
+                    await self._poll_streaming_partial(actor)
+                except StreamingASRUnavailable as exc:
+                    actor.asr_poll_due_monotonic = time.monotonic() + max(
+                        self._partial_poll_interval_seconds,
+                        0.60,
+                    )
+                    logger.warning(
+                        "audiosocket.streaming_asr_partial_poll_failed session_id=%s error=%s",
+                        actor.registration.voice_session_id,
+                        str(exc)[:500],
+                    )
         elif actor.registration.voice_v2_partial_asr:
             self._maybe_schedule_partial_asr(actor)
         if vad_result.utterance_pcm:
@@ -1002,14 +1029,14 @@ class AudioSocketMediaRuntime:
                         partial_intent = str(actor.partial_intent or "").strip() or self._detect_early_intent(partial_transcript)
                         base_metadata["partial_transcript"] = partial_transcript
                         base_metadata["early_intent"] = partial_intent
-            if partial_transcript:
+            if self._should_emit_partial_ack(actor, partial_transcript, partial_intent):
                 await self._emit_early_ack(
                     actor,
                     language=actor.registration.language,
                     metadata=base_metadata,
                     ack_source="streaming_partial" if actor.asr_streaming_enabled else "precomputed_partial_asr",
                 )
-            elif len(pcm_bytes) >= self._immediate_ack_min_bytes:
+            elif self._should_emit_blind_ack(actor, pcm_bytes):
                 await self._emit_early_ack(
                     actor,
                     language=actor.registration.language,
@@ -1041,6 +1068,7 @@ class AudioSocketMediaRuntime:
             return
 
         actor.finalized_utterance_generation = max(actor.finalized_utterance_generation, utterance_generation)
+        actor.finalized_caller_turn_count += 1
         actor.partial_transcript = transcript_text if actor.registration.voice_v2_partial_asr else None
         actor.partial_intent = self._detect_early_intent(transcript_text)
         actor.stable_partial_intent = actor.partial_intent

services/streaming_asr_sidecar_service/app.py

@@ -51,13 +51,17 @@ def _partial_min_audio_ms() -> int:
 
 
 def _partial_recompute_interval_ms() -> int:
-    return max(_int_env("AI_VOICE_V2_STREAMING_ASR_PARTIAL_RECOMPUTE_INTERVAL_MS", 200), 80)
+    return max(_int_env("AI_VOICE_V2_STREAMING_ASR_PARTIAL_RECOMPUTE_INTERVAL_MS", 450), 80)
 
 
 def _partial_stability_hold_ms() -> int:
     return max(_int_env("AI_VOICE_V2_STREAMING_ASR_PARTIAL_STABILITY_HOLD_MS", 400), 120)
 
 
+def _partial_max_audio_ms() -> int:
+    return max(_int_env("AI_VOICE_V2_STREAMING_ASR_PARTIAL_MAX_AUDIO_MS", 1800), 400)
+
+
 def _model_name() -> str:
     return str(os.getenv("AI_VOICE_V2_STREAMING_ASR_MODEL", "base") or "base").strip() or "base"
 
@@ -430,6 +434,15 @@ def _copy_stream_snapshot(stream: StreamState) -> tuple[bytes, int, int, str]:
     )
 
 
+def _trim_partial_snapshot(pcm_bytes: bytes, *, sample_rate_hz: int) -> bytes:
+    if not pcm_bytes or sample_rate_hz <= 0:
+        return pcm_bytes
+    max_bytes = int((_partial_max_audio_ms() / 1000.0) * sample_rate_hz * 2)
+    if max_bytes <= 0 or len(pcm_bytes) <= max_bytes:
+        return pcm_bytes
+    return pcm_bytes[-max_bytes:]
+
+
 def _transcribe_stream_snapshot(pcm_bytes: bytes, *, sample_rate_hz: int, language_hint: str) -> SidecarTranscript:
     return _get_engine().transcribe_pcm(
         pcm_bytes,
@@ -540,7 +553,7 @@ def poll_partial(stream_id: str, _: None = Depends(_require_internal_actor)) ->
 
     pcm_bytes, buffer_version, sample_rate_hz, language_hint = _copy_stream_snapshot(stream)
     transcript = _transcribe_stream_snapshot(
-        pcm_bytes,
+        _trim_partial_snapshot(pcm_bytes, sample_rate_hz=sample_rate_hz),
         sample_rate_hz=sample_rate_hz,
         language_hint=language_hint,
     )

tests/test_ai_orchestrator_service.py

@@ -1497,6 +1497,45 @@ def test_voice_postprocess_reply_rewrites_false_lookup_promise():
     assert "филиал" in reply_text.lower() or "адрес" in reply_text.lower()
 
 
+def test_voice_postprocess_reply_rewrites_midcall_greeting_with_followup_question():
+    reply_text = voice_module._voice_postprocess_reply_text(
+        language="ru",
+        transcript_text="Здравствуйте, мне надо узнать...",
+        transcript_window=[
+            SimpleNamespace(speaker="assistant", text=voice_module._voice_greeting("ru"), sequence_no=1, source_type="voice_policy", barge_in_interrupted=False, created_at=utc_now_iso()),
+            SimpleNamespace(speaker="caller", text="Здравствуйте, мне надо узнать...", sequence_no=2, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
+        ],
+        reply_text="Здравствуйте, Ернур! О чем именно вы хотите узнать?",
+        kb_results=[],
+        needs_handoff=False,
+    )
+
+    normalized = reply_text.lower()
+    assert "здравствуйте" not in normalized
+    assert "о чем именно" not in normalized
+
+
+def test_voice_postprocess_reply_reuses_active_topic_after_frustration_turn():
+    reply_text = voice_module._voice_postprocess_reply_text(
+        language="ru",
+        transcript_text="Сколько раз повторять тебе?",
+        transcript_window=[
+            SimpleNamespace(speaker="caller", text="Мне надо узнать график работы.", sequence_no=1, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
+            SimpleNamespace(speaker="caller", text="В городе Алма-Ата.", sequence_no=2, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
+            SimpleNamespace(speaker="caller", text="Меня интересует филиал в Ауэзовском районе.", sequence_no=3, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
+            SimpleNamespace(speaker="caller", text="Сколько раз повторять тебе?", sequence_no=4, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
+        ],
+        reply_text="Пожалуйста, уточните, какая услуга вас интересует, чтобы я мог подсказать тариф.",
+        kb_results=[],
+        needs_handoff=False,
+    )
+
+    normalized = reply_text.lower()
+    assert "тариф" not in normalized
+    assert "услуг" not in normalized
+    assert "филиал" in normalized or "адрес" in normalized
+
+
 def test_turn_voice_session_early_plan_does_not_persist_partial_turns(monkeypatch):
     monkeypatch.setenv("AI_VOICE_POLICY_MODE", "v2_streaming_duplex")
 

tests/test_ai_voice_media_runtime.py

@@ -624,6 +624,7 @@ def test_media_runtime_voice_v2_plays_short_ack_before_main_reply():
             frame_ms=20,
             frame_bytes=320,
         )
+        actor.finalized_caller_turn_count = 1
         await runtime._process_utterance(actor, pcm_frame, False)
 
     asyncio.run(_scenario())
@@ -842,6 +843,7 @@ def test_media_runtime_voice_v2_emits_generic_ack_before_full_asr_without_partia
             frame_ms=20,
             frame_bytes=320,
         )
+        actor.finalized_caller_turn_count = 1
         await runtime._process_utterance(actor, pcm_frame * 45, False)
 
     asyncio.run(_scenario())
@@ -940,6 +942,7 @@ def test_media_runtime_voice_v2_emits_ack_for_short_utterance_after_reduced_thre
             frame_ms=20,
             frame_bytes=320,
         )
+        actor.finalized_caller_turn_count = 1
         await runtime._process_utterance(actor, pcm_frame * 40, False)
 
     asyncio.run(_scenario())
@@ -949,6 +952,100 @@ def test_media_runtime_voice_v2_emits_ack_for_short_utterance_after_reduced_thre
     assert speak_events[0][1] < timings["full_finished"]
 
 
+def test_media_runtime_voice_v2_skips_blind_ack_on_first_turn_without_partial_signal():
+    speak_events: list[tuple[str, float]] = []
+
+    class _FastASRProvider(ASRProvider):
+        name = "fast-asr"
+
+        def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
+            assert audio_bytes
+            time.sleep(0.08)
+            return ASRTranscription(text="hello there", language=language_hint or "ru", confidence=0.9)
+
+    runtime = AudioSocketMediaRuntime(
+        enabled=True,
+        host="127.0.0.1",
+        port=0,
+        frame_ms=20,
+        idle_timeout_seconds=2.0,
+        registration_wait_timeout_seconds=0.5,
+        min_speech_ms=40,
+        trailing_silence_ms=40,
+        max_turn_ms=2000,
+        asr_provider=_FastASRProvider(),
+        tts_provider=_StubTTSProvider(),
+        load_registration_by_media_uuid=lambda value: None,
+        mark_media_connected=lambda session_id, value: None,
+        mark_media_ended=lambda session_id, reason: None,
+        touch_media_frame=lambda session_id: None,
+        set_state=lambda session_id, state, handoff_reason, metadata: None,
+        get_pending_greeting=lambda session_id: None,
+        mark_reply_delivered=lambda session_id, text, is_greeting: None,
+        plan_reply=lambda session_id, text, metadata, kind: None,
+        process_turn=lambda session_id, transcript_text, language, barge_in, metadata: VoiceAITurnDecisionOut(
+            language=language or "ru",
+            intent="clarification",
+            reply_text="Подскажите подробнее, пожалуйста.",
+            confidence=0.9,
+            needs_handoff=False,
+            handoff_reason=None,
+            case_action="keep_open",
+            kb_refs=[],
+            summary_text="reply ready",
+            model="stub-voice",
+            latency_ms=1,
+            status="active",
+        ),
+        request_handoff=lambda session_id, customer_request_text, decision: None,
+        handle_media_error=lambda session_id, message, metadata: None,
+    )
+
+    async def _fake_speak_text(
+        current_actor,
+        text: str,
+        *,
+        is_greeting: bool,
+        style_hints: dict[str, object] | None = None,
+    ) -> None:
+        del current_actor, is_greeting, style_hints
+        speak_events.append((text, time.monotonic()))
+        await asyncio.sleep(0)
+
+    runtime._speak_text = _fake_speak_text  # type: ignore[method-assign]
+    pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160
+
+    async def _scenario() -> None:
+        actor = MediaActor(
+            registration=MediaRegistration(
+                voice_session_id="avs_media_runtime_v2_first_turn",
+                call_id="call_media_runtime_v2_first_turn",
+                interaction_id="int_media_runtime_v2_first_turn",
+                ai_session_id="ais_media_runtime_v2_first_turn",
+                language="ru",
+                media_uuid=str(uuid.uuid4()),
+                queue_code="voice_lab_ai",
+                queue_id="que_voice_lab_ai",
+                agent_profile="voice_support",
+                voice_v2_enabled=True,
+                voice_v2_ack_mode="immediate_short",
+                voice_v2_streaming_tts=True,
+                voice_v2_partial_asr=False,
+            ),
+            reader=asyncio.StreamReader(),
+            writer=None,  # type: ignore[arg-type]
+            vad=EnergyVAD(frame_ms=20, min_speech_ms=40, trailing_silence_ms=40, max_turn_ms=2000),
+            frame_ms=20,
+            frame_bytes=320,
+        )
+        await runtime._process_utterance(actor, pcm_frame * 40, False)
+
+    asyncio.run(_scenario())
+
+    assert len(speak_events) == 1
+    assert speak_events[0][0] == "Подскажите подробнее, пожалуйста."
+
+
 def test_media_runtime_ignores_low_signal_utterance_without_ack_or_turn():
     planned: list[tuple[str, str, str, dict | None]] = []
     delivered: list[tuple[str, str, bool]] = []
@@ -1120,6 +1217,7 @@ def test_media_runtime_voice_v2_inserts_small_gap_between_ack_and_main_reply():
             frame_ms=20,
             frame_bytes=320,
         )
+        actor.finalized_caller_turn_count = 1
         await runtime._process_utterance(actor, pcm_frame * 40, False)
 
     asyncio.run(_scenario())
@@ -1226,6 +1324,7 @@ def test_media_runtime_voice_v2_emotive_ack_uses_ru_variants_and_style_hints_onl
             frame_ms=20,
             frame_bytes=320,
         )
+        actor.finalized_caller_turn_count = 1
         pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160
         await runtime._process_utterance(actor, pcm_frame, False)
 

tests/test_streaming_asr_sidecar_service.py

@@ -133,6 +133,49 @@ def test_streaming_asr_sidecar_stream_lifecycle(monkeypatch):
     assert len(engine.calls) >= 3
 
 
+def test_streaming_asr_sidecar_partial_uses_recent_audio_tail(monkeypatch):
+    engine = _reset_sidecar(monkeypatch)
+    monkeypatch.setattr(sidecar_module, "_partial_max_audio_ms", lambda: 1800)
+    client = TestClient(sidecar_module.app)
+
+    open_response = client.post(
+        "/internal/asr/streams",
+        headers=_admin_headers(),
+        json={
+            "session_id": "avs_sidecar_tail",
+            "language_hint": "ru",
+            "sample_rate_hz": 8000,
+            "encoding": "pcm_s16le",
+        },
+    )
+    assert open_response.status_code == 200
+    stream_id = open_response.json()["stream_id"]
+
+    long_chunk = _pcm_chunk(3200)
+    push_response = client.post(
+        f"/internal/asr/streams/{stream_id}/chunks",
+        headers=_admin_headers(),
+        json={
+            "pcm_b64": base64.b64encode(long_chunk).decode("ascii"),
+            "sample_rate_hz": 8000,
+            "encoding": "pcm_s16le",
+        },
+    )
+    assert push_response.status_code == 200
+
+    partial_response = client.get(
+        f"/internal/asr/streams/{stream_id}/partial",
+        headers=_admin_headers(),
+    )
+    assert partial_response.status_code == 200
+    assert engine.calls
+    partial_len, sample_rate_hz, language_hint = engine.calls[-1]
+    assert sample_rate_hz == 8000
+    assert language_hint == "ru"
+    assert partial_len < len(long_chunk)
+    assert partial_len <= int(1.8 * 8000 * 2)
+
+
 def test_streaming_asr_sidecar_rejects_unsupported_language(monkeypatch):
     _reset_sidecar(monkeypatch, supported_languages={"ru"})
     client = TestClient(sidecar_module.app)