Play a filler ack during slow voice decisions and cache KB search rows

Two independent latency fixes for the voice-assistant reply pipeline,
both scoped to the parts of the flow that run regardless of whether
voice_v2 is enabled for a queue:

1. media_runtime._process_utterance: the v1/fallback turn path (used by
   any queue not covered by AI_VOICE_V2_QUEUE_CODES) silently awaited the
   full LLM decision with no audio playing at all, unlike the v2 path
   which already has a decision-timeout ack. Give v1 the same behavior:
   wait up to 600ms (_v1_ack_wait_seconds) for the decision, and if it's
   still not ready, play a short "Секунду." filler via the existing
   _emit_early_ack before the real reply, instead of leaving the caller
   in silence for the full LLM+TTS round trip. Reuses the same ack
   selection/playback code path v2 already exercises, so no new failure
   modes - just an added timeout branch mirroring the existing v2 one.

2. ai_orchestrator_service._kb_search: every voice/chat turn re-ran a
   full-table scan of kb_articles (all columns, including body text) and
   rescored every row in Python, even though the KB rarely changes
   mid-conversation. Added an in-process cache keyed by language, gated
   on a cheap content fingerprint (row count + max id + max updated_at +
   summed title/body/tags length, all computed server-side without
   transferring the text columns). A fingerprint mismatch always
   triggers a fresh fetch, so this can never serve stale results after
   an insert/update/delete - unlike a naive TTL cache, which would have
   been be wrong the moment a test (or a real KB edit) changed the table
   within the cache window.

   Note the first fingerprint design (count + max id + max updated_at
   only) was insufficient: utc_now_iso() truncates to whole seconds and
   SQLite reuses primary keys after a full-table delete, so two
   different row sets written in the same wall-clock second could share
   a fingerprint. Caught this via a real test failure
   (test_ai_whatsapp_relaxed_kb_search_answers_phrase_query breaking
   only when run after test_ai_orchestrator_service.py in the same
   process) before it could reach production; the summed content-length
   term closes the gap.

Added test_media_runtime_plays_filler_ack_when_v1_decision_is_slow
(asserts greeting -> ack -> reply delivery order when process_turn is
slow) and verified the KB cache against the full
test_ai_orchestrator_service.py + test_ai_whatsapp_orchestrator_service.py
suite plus a wider kb/orchestrator/whatsapp/telegram/voice-filtered run:
only the same pre-existing, already-documented failures remain (unrelated
sales_service test-isolation ordering, one known persona-prompt
assertion) - no new failures from either change.

Streaming the LLM decision itself (start speaking reply_text before the
full structured JSON response finishes generating) was scoped but
deliberately deferred: it needs incremental JSON parsing on top of SSE
streaming to detect when just the reply_text field is complete, shared
across both voice and text-channel decision paths - a separate,
higher-risk change that deserves its own PR and testing pass rather than
being bundled here.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

services/ai_orchestrator_service/app.py

@@ -8,11 +8,12 @@ import math
 import os
 import re
 import time
+from threading import Lock
 from typing import Any
 
 import httpx
 from fastapi import Depends, FastAPI, HTTPException, Query
-from sqlalchemy import select
+from sqlalchemy import func, select
 
 from services.shared.core import Role, new_id, utc_now_iso
 from services.shared.db import get_session
@@ -1909,13 +1910,61 @@ def _article_snippet(article: KBArticleRow, limit: int = 240) -> str:
     return f"{raw[: limit - 3]}..."
 
 
-def _kb_search(session, text: str, *, language: str | None = None) -> list[KBArticleRow]:
-    if not str(text or "").strip():
-        return []
+_KB_ROWS_CACHE: dict[str, tuple[tuple[int, int, str, int], list[KBArticleRow]]] = {}
+_KB_ROWS_CACHE_LOCK = Lock()
+
+
+def _kb_rows_fingerprint(session, language: str | None) -> tuple[int, int, str, int]:
+    # A cheap aggregate (row content lengths, computed server-side - no
+    # title/body/tags_json actually transferred) that changes on any
+    # insert, update, or delete, so the cache below can never serve stale
+    # results. count/max_id/max_updated_at alone aren't sufficient: ids can
+    # be reused after a row is deleted and updated_at has 1-second
+    # resolution, so two different row sets can otherwise share a
+    # fingerprint if they happen to be written within the same second.
+    content_length = func.length(KBArticleRow.title) + func.length(KBArticleRow.body) + func.length(
+        KBArticleRow.tags_json
+    )
+    stmt = select(
+        func.count(KBArticleRow.id),
+        func.max(KBArticleRow.id),
+        func.max(KBArticleRow.updated_at),
+        func.sum(content_length),
+    )
+    if language is not None:
+        stmt = stmt.where(KBArticleRow.language == normalize_kb_language(language))
+    count, max_id, max_updated_at, total_length = session.execute(stmt).one()
+    return (int(count or 0), int(max_id or 0), str(max_updated_at or ""), int(total_length or 0))
+
+
+def _load_kb_rows_cached(session, language: str | None) -> list[KBArticleRow]:
+    # Every voice/chat turn re-runs KB search, and this table rarely changes
+    # mid-call, so avoid re-fetching + re-transferring the whole table from
+    # the DB (title/body/tags_json for every article) when nothing changed.
+    cache_key = normalize_kb_language(language) if language is not None else "__all__"
+    fingerprint = _kb_rows_fingerprint(session, language)
+
+    with _KB_ROWS_CACHE_LOCK:
+        cached = _KB_ROWS_CACHE.get(cache_key)
+    if cached is not None and cached[0] == fingerprint:
+        return cached[1]
+
     stmt = select(KBArticleRow).order_by(KBArticleRow.id.desc())
     if language is not None:
         stmt = stmt.where(KBArticleRow.language == normalize_kb_language(language))
-    rows = session.execute(stmt).scalars().all()
+    rows = list(session.execute(stmt).scalars().all())
+    for row in rows:
+        session.expunge(row)
+
+    with _KB_ROWS_CACHE_LOCK:
+        _KB_ROWS_CACHE[cache_key] = (fingerprint, rows)
+    return rows
+
+
+def _kb_search(session, text: str, *, language: str | None = None) -> list[KBArticleRow]:
+    if not str(text or "").strip():
+        return []
+    rows = _load_kb_rows_cached(session, language)
     return search_kb_rows(rows, text, limit=_ai_max_kb_results())
 
 

services/ai_voice_runtime_service/media_runtime.py

@@ -193,6 +193,7 @@ class AudioSocketMediaRuntime:
         self._partial_asr_min_ms = 320
         self._immediate_ack_min_ms = 700
         self._v2_ack_post_gap_seconds = 0.10
+        self._v1_ack_wait_seconds = 0.6
         self._partial_poll_interval_seconds = 0.20
         self._streaming_asr_partial_poll_enabled = (
             str(os.getenv("AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED", "0")).strip().lower()
@@ -1619,6 +1620,17 @@ class AudioSocketMediaRuntime:
                     early_plan_decision = await self._take_early_plan_decision(actor, timeout_seconds=0.12)
                     decision = early_plan_decision if early_plan_decision is not None else await decision_task
         else:
+            try:
+                decision = await asyncio.wait_for(asyncio.shield(decision_task), timeout=self._v1_ack_wait_seconds)
+            except asyncio.TimeoutError:
+                if not actor.early_ack_started:
+                    await self._emit_early_ack(
+                        actor,
+                        language=transcription.language or actor.registration.language,
+                        metadata=metadata,
+                        ack_source="v1_decision_timeout",
+                        ack_kind="unknown",
+                    )
                 decision = await decision_task
         handoff_task: asyncio.Task | None = None
         if decision.needs_handoff:

tests/test_ai_voice_media_runtime.py

@@ -187,6 +187,103 @@ def test_media_runtime_streams_greeting_and_turn():
     assert errors == []
 
 
+def test_media_runtime_plays_filler_ack_when_v1_decision_is_slow():
+    registrations: dict[str, MediaRegistration] = {}
+    delivered: list[tuple[str, str, bool]] = []
+
+    media_uuid = str(uuid.uuid4())
+    registrations[media_uuid] = MediaRegistration(
+        voice_session_id="avs_media_runtime_slow_v1",
+        call_id="call_media_runtime_slow_v1",
+        interaction_id="int_media_runtime_slow_v1",
+        ai_session_id="ais_media_runtime_slow_v1",
+        language="ru",
+        media_uuid=media_uuid,
+    )
+
+    def _slow_process_turn(session_id, transcript_text, language, barge_in, metadata):
+        del transcript_text, barge_in, metadata
+        time.sleep(0.9)
+        return VoiceAITurnDecisionOut(
+            language=language or "ru",
+            intent="answer",
+            reply_text="reply",
+            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",
+        )
+
+    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=400,
+        asr_provider=_StubASRProvider(),
+        tts_provider=_StubTTSProvider(),
+        load_registration_by_media_uuid=lambda value: registrations.get(value),
+        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: "greeting" if session_id == "avs_media_runtime_slow_v1" else None,
+        mark_reply_delivered=lambda session_id, text, is_greeting: delivered.append((session_id, text, is_greeting)),
+        plan_reply=lambda session_id, text, metadata, kind: None,
+        process_turn=_slow_process_turn,
+        request_handoff=lambda session_id, customer_request_text, decision: None,
+        handle_media_error=lambda session_id, message, metadata: None,
+    )
+
+    async def _scenario() -> None:
+        await runtime.start()
+        port = runtime._server.sockets[0].getsockname()[1]
+
+        reader, writer = await asyncio.open_connection("127.0.0.1", port)
+        writer.write(encode_packet(AUDIO_SOCKET_PACKET_UUID, uuid.UUID(media_uuid).bytes))
+        await writer.drain()
+
+        packet_type, _ = await read_packet(reader, timeout=2.0)
+        assert packet_type == AUDIO_SOCKET_PACKET_PCM16
+        await asyncio.sleep(0.15)
+
+        speech_frame = (1000).to_bytes(2, "little", signed=True) * 160
+        silence_frame = b"\x00\x00" * 160
+        for _ in range(2):
+            writer.write(encode_audio_packet(speech_frame))
+        for _ in range(2):
+            writer.write(encode_audio_packet(silence_frame))
+        await writer.drain()
+
+        packet_type, _ = await read_packet(reader, timeout=2.0)
+        assert packet_type == AUDIO_SOCKET_PACKET_PCM16
+
+        for _ in range(60):
+            if len(delivered) >= 3:
+                break
+            await asyncio.sleep(0.05)
+
+        writer.close()
+        await writer.wait_closed()
+        await asyncio.sleep(0.2)
+        await runtime.stop()
+
+    asyncio.run(_scenario())
+
+    assert delivered[0] == ("avs_media_runtime_slow_v1", "greeting", True)
+    assert delivered[1] == ("avs_media_runtime_slow_v1", "Секунду.", False)
+    assert delivered[2] == ("avs_media_runtime_slow_v1", "reply", False)
+
+
 def test_media_runtime_speaks_technical_fallback_when_asr_transcribe_fails():
     registrations: dict[str, MediaRegistration] = {}
     delivered: list[tuple[str, str, bool]] = []