feat: require complete KB-grounded answers (no dropped facts) + ask did-that-answer-your-question after each answer, per ai-operator spec

search_kb_rows had no relevance floor: any exact-token hit, however
generic, scored above zero and could win as the top/only result. A
caller utterance as thin as a bare "да" (confirming the language) could
exact-match that same common word inside an unrelated FAQ article's
body and get returned as "the" answer, which then got read back
almost verbatim — this is what surfaced live as the AI unprompted
launching into a voucher-activation explanation right after the
customer confirmed Russian, having said nothing else.

tokenize_kb_text now drops a curated set of RU/KZ greetings,
confirmations, pronouns, and particles. A stopword-only query naturally
falls through to the existing "no query tokens -> no results" path
instead of returning a coincidental match; genuine single-content-word
queries (e.g. "ваучер") are unaffected. Applies to every channel that
calls _kb_search (voice, Telegram, WhatsApp), not just voice.

deployment/aimaq.env.production

@@ -10,10 +10,10 @@ ALLOW_LEGACY_HEADER_AUTH=0
 
 AI_PROVIDER=openai_compatible
 AI_API_BASE=https://api.openai.com/v1
-AI_API_KEY=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
+AI_API_KEY=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
-AI_MODEL=gpt-4o-mini
+AI_MODEL=gpt-5-mini
 AI_TIMEOUT_SECONDS=30
-AI_VOICE_AI_TIMEOUT_SECONDS=10
+AI_VOICE_AI_TIMEOUT_SECONDS=15
 AI_WEB_SEARCH_ENABLED=1
 AI_WEB_SEARCH_MAX_RESULTS=5
 AI_WEB_SEARCH_GL=kz

migrations/sql/0034_kb_intent_code_postgres.sql

@@ -0,0 +1,2 @@
+ALTER TABLE kb_articles ADD COLUMN IF NOT EXISTS intent_code TEXT;
+CREATE INDEX IF NOT EXISTS idx_kb_articles_intent_code ON kb_articles(intent_code);

migrations/sql/0034_kb_intent_code_sqlite.sql

@@ -0,0 +1,2 @@
+ALTER TABLE kb_articles ADD COLUMN intent_code TEXT;
+CREATE INDEX IF NOT EXISTS idx_kb_articles_intent_code ON kb_articles(intent_code);

services/ai_orchestrator_service/app.py

@@ -9,7 +9,7 @@ import os
 import re
 import time
 from threading import Lock
-from typing import Any
+from typing import Any, Iterable
 
 import httpx
 from fastapi import Depends, FastAPI, HTTPException, Query
@@ -23,6 +23,7 @@ from services.shared.ai_context_summary import (
     update_context_summary_from_assistant_turn,
     update_context_summary_from_user_turn,
 )
+from services.shared.intents import normalize_intent
 from services.shared.kb_localization import normalize_kb_language
 from services.shared.kb_search import search_kb_rows
 from services.shared.models import (
@@ -2315,6 +2316,7 @@ def _openai_prompt(
             "article_id": article.article_id,
             "title": article.title,
             "snippet": _article_snippet(article),
+            "intent_code": getattr(article, "intent_code", None),
         }
         for article in kb_results
     ]
@@ -2355,13 +2357,18 @@ def _request_structured_model_decision(
     if not _ai_api_base() or not _ai_api_key():
         raise RuntimeError("AI_API_BASE / AI_API_KEY are required for openai_compatible provider")
     started = time.perf_counter()
-    payload = {
+    model = _ai_model()
-        "model": _ai_model(),
+    payload: dict[str, Any] = {
-        "temperature": 0.2,
+        "model": model,
-        "max_tokens": _ai_decision_max_tokens(),
         "response_format": {"type": "json_object"},
         "messages": messages,
     }
+    if model.startswith("gpt-5"):
+        payload["max_completion_tokens"] = _ai_decision_max_tokens()
+        payload["reasoning_effort"] = "minimal"
+    else:
+        payload["temperature"] = 0.2
+        payload["max_tokens"] = _ai_decision_max_tokens()
     effective_timeout = timeout_seconds if timeout_seconds is not None else _ai_timeout_seconds()
     with httpx.Client(timeout=effective_timeout) as client:
         response = client.post(
@@ -2410,10 +2417,15 @@ def _openai_compatible_decision(
     )
 
 
-def _sanitize_decision(raw: dict[str, Any], *, fallback_language: str) -> dict[str, Any]:
+def _sanitize_decision(
+    raw: dict[str, Any],
+    *,
+    fallback_language: str,
+    known_topic_codes: Iterable[str] = (),
+) -> dict[str, Any]:
     decision = {
         "language": str(raw.get("language") or fallback_language or "ru"),
-        "intent": str(raw.get("intent") or "unknown"),
+        "intent": normalize_intent(raw.get("intent"), known_topic_codes=known_topic_codes),
         "reply_text": str(raw.get("reply_text") or "").strip(),
         "extracted_name": str(raw.get("extracted_name") or "").strip() or None,
         "confidence": float(raw.get("confidence") or 0.0),
@@ -2611,7 +2623,13 @@ def _decide_reply(
         raw["_model"] = _ai_model()
         raw["_latency_ms"] = 1
         raw["_finish_reason"] = "stop"
-    return _sanitize_decision(raw, fallback_language=language)
+    return _sanitize_decision(
+        raw,
+        fallback_language=language,
+        known_topic_codes=[
+            code for article in kb_results if (code := getattr(article, "intent_code", None))
+        ],
+    )
 
 
 def _update_ai_session_context_summary_from_user_turn(

services/ai_orchestrator_service/operator_persona.py

@@ -135,8 +135,12 @@ def human_fallback_reply(language: str, *, is_greeting: bool = False) -> str:
 def operator_system_prompt(*, language: str, channel_label: str, is_voice: bool, config: Any | None = None) -> str:
     preferred_language = "Kazakh" if str(language or "").strip().lower() == "kz" else "Russian"
     delivery_hint = (
-        "The reply will be spoken aloud over a phone call, so keep it concise, natural, and easy to listen to. "
+        "The reply will be spoken aloud over a phone call, so keep it concise, natural, and easy to listen to, "
-        "Prefer one or two short sentences and at most one clarifying question. "
+        "but never at the cost of dropping a required fact. "
+        "Use as many short sentences as needed to cover every material fact from the grounding kb_results "
+        "snippet completely - required steps, codes, commands, deadlines, amounts, and conditions - typically "
+        "two to four short sentences; never silently omit or shorten out a required step just to sound brief. "
+        "At most one clarifying question per turn. "
         "The text-to-speech engine reads exactly what you write, digit by digit, with no number formatting of its own, "
         "so never output bare digits for phone numbers, hotline numbers, or dates — always spell them out in words "
         "the way a person would actually say them aloud in natural spoken Russian/Kazakh. "
@@ -174,9 +178,16 @@ def operator_system_prompt(*, language: str, channel_label: str, is_voice: bool,
         "Paraphrase them naturally instead of quoting them verbatim. "
         "Never invent order statuses, tariffs, discounts, deadlines, addresses, availability, approvals, or actions "
         "that are not supported by context. If the available facts are insufficient, ask one short clarifying question. "
+        "After you deliver a complete answer grounded in kb_results (not when you are asking a clarifying question, "
+        "handling an identity/off-topic reply, or closing the call), end reply_text with a brief natural check such as «Ответила ли я на ваш вопрос?» in Russian, or its natural Kazakh equivalent, phrased differently each time so it does not sound scripted. "
         "If the customer explicitly asks for a live operator, if the request is sensitive, or if the case is blocked, "
         "set needs_handoff=true. "
         f"{delivery_hint} "
         "Return only a JSON object with keys: language, intent, reply_text, extracted_name, confidence, needs_handoff, "
         "handoff_reason, case_action, kb_refs. case_action must be one of none, close, escalate, keep_open. "
+        "For `intent`: if your reply is grounded in one of the provided kb_results, set intent to that snippet's "
+        "intent_code exactly as given (do not translate, reformat, or invent your own code). If no kb_results were "
+        "used, use one of these fixed values as appropriate: identity_question, handoff_request, sensitive_request, "
+        "resolution_confirmed, clarification, kb_answer, unknown. Never invent a new intent value outside of these "
+        "two sources — the platform discards anything else."
     )

services/ai_orchestrator_service/voice.py

@@ -1359,7 +1359,13 @@ def _voice_v2_enabled(metadata: dict[str, Any] | None = None) -> bool:
     return _voice_policy_mode() in {"v2_fast_conversational", "v2_streaming_duplex"}
 
 
-def _voice_early_intent_bucket(text: str) -> str:
+def _voice_ack_topic_bucket(text: str) -> str:
+    """Coarse keyword heuristic used only to pick an ack phrase / early clarifying
+    question (see _voice_ack_kind_for_intent and _voice_early_plan) while the real,
+    KB-grounded decision is still in flight. This is NOT the canonical FAQ intent
+    (see services.shared.intents) and must never be echoed back as the final
+    decision's `intent` value.
+    """
     normalized = " ".join(str(text or "").strip().lower().split())
     if not normalized:
         return "unknown"
@@ -1509,7 +1515,7 @@ def _voice_v2_metadata(
     if not _voice_v2_enabled(request_metadata):
         return {}
     payload = request_metadata if isinstance(request_metadata, dict) else {}
-    early_intent = _voice_early_intent_bucket(transcript_text)
+    early_intent = _voice_ack_topic_bucket(transcript_text)
     metadata: dict[str, Any] = {
         "voice_v2_enabled": True,
         "early_intent": early_intent,
@@ -1668,6 +1674,7 @@ def _voice_llm_prompt_messages(
             "article_id": article.article_id,
             "title": article.title,
             "snippet": app._article_snippet(article, limit=240),
+            "intent_code": getattr(article, "intent_code", None),
         }
         for article in kb_results[:3]
     ]
@@ -1767,7 +1774,13 @@ def _voice_llm_decision(
         )
     except Exception:
         return None
-    decision = app._sanitize_decision(raw, fallback_language=language)
+    decision = app._sanitize_decision(
+        raw,
+        fallback_language=language,
+        known_topic_codes=[
+            code for article in kb_results if (code := getattr(article, "intent_code", None))
+        ],
+    )
     if not str(decision.get("reply_text") or "").strip():
         decision["reply_text"] = _voice_generic_prompt(language)
     if decision.get("needs_handoff") and not decision.get("handoff_reason"):

services/ai_voice_runtime_service/app.py

@@ -1335,6 +1335,27 @@ def _process_voice_ai_turn_sync(
     payload: VoiceAITurnIn,
     *,
     auto_handoff: bool,
+) -> VoiceAITurnDecisionOut:
+    last_exc: Exception | None = None
+    for attempt in range(2):
+        try:
+            return _process_voice_ai_turn_sync_once(session_id, payload, auto_handoff=auto_handoff)
+        except HTTPException as exc:
+            if attempt == 0 and exc.status_code == 502 and "deadlock detected" in str(exc.detail).lower():
+                logging.getLogger(__name__).warning("voice_turn_deadlock_retry session_id=%s", session_id)
+                last_exc = exc
+                continue
+            raise
+    if last_exc is not None:
+        raise last_exc
+    raise RuntimeError("unreachable")
+
+
+def _process_voice_ai_turn_sync_once(
+    session_id: str,
+    payload: VoiceAITurnIn,
+    *,
+    auto_handoff: bool,
 ) -> VoiceAITurnDecisionOut:
     session = get_session()
     voice_session = None

services/ai_voice_runtime_service/media_runtime.py

@@ -956,6 +956,11 @@ class AudioSocketMediaRuntime:
                 bool(partial.is_final),
                 transcript_text[:160],
             )
+            # Fresh speech content is still arriving, so push the
+            # no-speech silence-timeout deadline forward instead of
+            # interrupting a caller who is actively mid-utterance.
+            if actor.state == "listening":
+                actor.listening_since_monotonic = time.monotonic()
         actor.partial_transcript = transcript_text
         self._update_stable_partial_transcript(actor, transcript_text, provider_stable=bool(partial.is_stable or partial.is_final))
         intent = self._detect_early_intent(transcript_text)

services/asterisk_bridge_service/bridge_processing.py

@@ -1233,7 +1233,7 @@ def process_recording_ready(
 
 
 _NO_ANSWER_DIAL_STATUSES = {"NOANSWER", "BUSY", "CANCEL", "CHANUNAVAIL", "CONGESTION"}
-_NO_ANSWER_HANGUP_CAUSES = {"17", "18", "19", "21", "34", "38"}
+_NO_ANSWER_HANGUP_CAUSES = {"1", "3", "17", "18", "19", "20", "21", "22", "34", "38"}
 
 
 def process_agent_dial_outcome(session, row: AsteriskEventLogRow, payload: dict[str, Any]) -> None:

services/asterisk_bridge_service/voice_ai.py

@@ -938,7 +938,11 @@ def retry_escalation_no_answer(session, *, call_id: str, dial_outcome: str) -> N
     )
 
     attempted_ids = json.loads(escalation.attempted_agent_ids_json or "[]")
-    channel = _resolve_handoff_channel(session, link)
+    # Use the already-known, actively-maintained channel name directly instead of
+    # _resolve_handoff_channel()'s live AMI CoreShowChannels re-discovery: that
+    # round-trip can take ~10s, which races (and loses) against the dialplan's
+    # own short MusicOnHold-then-hangup wait window for this exact retry path.
+    channel = str(link.channel_name or "").strip() or _resolve_handoff_channel(session, link)
     required_skills = json.loads(escalation.required_skills_json or "[]")
     next_agent = _reserve_routing_agent(
         call_id=call_id,

services/kb_service/app.py

@@ -40,6 +40,7 @@ def _article_out(row: KBArticleRow) -> KBArticleOut:
         article_id=row.article_id,
         category_id=row.category_id,
         article_group_id=resolve_article_group_id(row.article_id, row.article_group_id),
+        intent_code=row.intent_code,
         language=normalize_kb_language(row.language),
         title=row.title,
         body=row.body,
@@ -49,6 +50,10 @@ def _article_out(row: KBArticleRow) -> KBArticleOut:
     )
 
 
+def _normalize_intent_code(value: str | None) -> str | None:
+    return str(value or "").strip().upper() or None
+
+
 def _article_group_expr():
     return func.coalesce(KBArticleRow.article_group_id, KBArticleRow.article_id)
 
@@ -133,6 +138,7 @@ def create_article(
             article_id=article_id,
             category_id=payload.category_id,
             article_group_id=article_group_id,
+            intent_code=_normalize_intent_code(payload.intent_code),
             language=language,
             title=payload.title,
             body=payload.body,
@@ -196,6 +202,8 @@ def update_article(
 
         if "article_group_id" in data:
             row.article_group_id = target_group_id
+        if "intent_code" in data:
+            row.intent_code = _normalize_intent_code(data["intent_code"])
         if "language" in data:
             row.language = target_language
         if "title" in data:

services/shared/intents.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from typing import Iterable
+
+# Fixed, code-level conversational signals — not FAQ topics. These strings are
+# already used as intent literals across ai_orchestrator_service/app.py,
+# voice.py, and asserted directly in tests; centralized here rather than
+# renamed so every call site validates against the same set.
+CONTROL_INTENTS: frozenset[str] = frozenset(
+    {
+        "identity_question",
+        "handoff_request",
+        "sensitive_request",
+        "resolution_confirmed",
+        "clarification",
+        "kb_answer",
+        "unknown",
+    }
+)
+
+UNKNOWN_INTENT = "unknown"
+
+
+def normalize_intent(raw: str | None, *, known_topic_codes: Iterable[str] = ()) -> str:
+    """Validate a model-produced intent against control intents and KB topic codes.
+
+    `known_topic_codes` are the `intent_code` values of the KB articles actually
+    shown to the model for this turn — anything else the model invents collapses
+    to UNKNOWN_INTENT rather than being trusted verbatim.
+    """
+    candidate = str(raw or "").strip()
+    if not candidate:
+        return UNKNOWN_INTENT
+    if candidate in CONTROL_INTENTS:
+        return candidate
+    normalized_topic_codes = {str(code or "").strip().upper() for code in known_topic_codes if str(code or "").strip()}
+    if candidate.upper() in normalized_topic_codes:
+        return candidate.upper()
+    return UNKNOWN_INTENT

services/shared/kb_search.py

@@ -19,12 +19,41 @@ _MIN_SOFT_MATCH_LENGTH = 4
 _NON_WORD_RE = re.compile(r"[^\w]+", re.UNICODE)
 _SPACE_RE = re.compile(r"\s+")
 
+# Common RU/KZ greetings, confirmations, pronouns, and particles that carry no
+# topical signal on their own. Without this, a caller utterance as thin as
+# "да" or "хорошо" could still exact-token-match some unrelated KB article
+# that happens to contain that word in its body, and get returned as the
+# top/only search result — read back to the caller as if it were the answer
+# to their question. Filtering these keeps _score_row's exact-token match
+# meaningful: a match now requires an actual content word.
+_STOPWORDS = frozenset(
+    {
+        # RU: greetings / confirmations / fillers
+        "алло", "ага", "да", "неа", "нет", "ой", "ок", "окей", "угу", "ясно",
+        "ладно", "хорошо", "понял", "поняла", "привет", "здравствуйте",
+        "добрый", "день", "вечер", "утро", "слышу", "слышно", "спасибо",
+        "пожалуйста", "извините", "простите", "алло",
+        # RU: pronouns / conjunctions / particles with no topical content
+        "я", "ты", "вы", "мы", "он", "она", "они", "это", "то", "и", "а",
+        "но", "или", "что", "как", "где", "когда", "если", "чтобы", "для",
+        "из", "по", "на", "в", "с", "у", "о", "же", "ли", "бы", "не", "ну",
+        "вот", "просто", "есть", "быть", "можно", "нужно", "надо", "уже",
+        # KZ: greetings / confirmations / fillers
+        "иә", "ия", "жоқ", "жарайды", "түсінікті", "рахмет", "сәлем",
+        "сәлеметсіз", "бе", "кешіріңіз",
+        # KZ: pronouns / conjunctions / particles
+        "мен", "сен", "сіз", "біз", "ол", "олар", "және", "бірақ", "немесе",
+        "не", "қалай", "қайда", "қашан", "үшін", "туралы",
+    }
+)
+
 
 class KBSearchRow(Protocol):
     id: int
     title: str
     body: str
     tags_json: str
+    intent_code: str | None
 
 
 T = TypeVar("T", bound=KBSearchRow)
@@ -37,7 +66,11 @@ def normalize_kb_text(value: str | None) -> str:
 
 
 def tokenize_kb_text(value: str | None) -> list[str]:
-    return [token for token in normalize_kb_text(value).split(" ") if len(token) >= _MIN_TOKEN_LENGTH]
+    return [
+        token
+        for token in normalize_kb_text(value).split(" ")
+        if len(token) >= _MIN_TOKEN_LENGTH and token not in _STOPWORDS
+    ]
 
 
 def search_kb_rows(

services/shared/models.py

@@ -363,6 +363,12 @@ class VoiceEventIn(BaseModel):
         "recording.ready",
         "call.connected",
         "call.transferred",
+        "AgentReserved",
+        "AgentRinging",
+        "AgentNoAnswer",
+        "AgentConnected",
+        "TransferCompleted",
+        "TransferFailed",
     ]
     call_id: str
     interaction_id: str | None = None
@@ -983,6 +989,7 @@ class KBCategoryOut(KBCategoryCreate):
 class KBArticleCreate(BaseModel):
     category_id: str
     article_group_id: str | None = None
+    intent_code: str | None = None
     language: str = "ru"
     title: str
     body: str
@@ -991,6 +998,7 @@ class KBArticleCreate(BaseModel):
 
 class KBArticleUpdate(BaseModel):
     article_group_id: str | None = None
+    intent_code: str | None = None
     language: str | None = None
     title: str | None = None
     body: str | None = None

services/shared/sql_init.py

@@ -583,6 +583,7 @@ def _apply_runtime_schema_compatibility() -> None:
         if "kb_articles" in table_names:
             columns = _table_columns(inspector, "kb_articles")
             _add_column_if_missing(conn, columns, "kb_articles", "article_group_id", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "kb_articles", "intent_code", "VARCHAR(64)")
             _add_column_if_missing(conn, columns, "kb_articles", "language", "VARCHAR(8) DEFAULT 'ru'")
             if "language" in columns:
                 conn.execute(
@@ -619,6 +620,13 @@ def _apply_runtime_schema_compatibility() -> None:
                         "ON kb_articles(language)"
                     )
                 )
+            if "idx_kb_articles_intent_code" not in indexes:
+                conn.execute(
+                    text(
+                        "CREATE INDEX IF NOT EXISTS idx_kb_articles_intent_code "
+                        "ON kb_articles(intent_code)"
+                    )
+                )
 
         if "sales_automation_tasks" in table_names:
             columns = _table_columns(inspector, "sales_automation_tasks")

services/shared/sql_models.py

@@ -700,6 +700,7 @@ class KBArticleRow(Base):
     article_id: Mapped[str] = mapped_column(String(64), unique=True, index=True)
     category_id: Mapped[str] = mapped_column(String(64), index=True)
     article_group_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    intent_code: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
     language: Mapped[str] = mapped_column(String(8), index=True, default="ru")
     title: Mapped[str] = mapped_column(String(512), index=True)
     body: Mapped[str] = mapped_column(Text)

tests/test_ai_orchestrator_service.py

@@ -414,6 +414,7 @@ def seed_kb_article(
     *,
     language: str = "ru",
     article_group_id: str | None = None,
+    intent_code: str | None = None,
 ) -> dict[str, str]:
     session = get_session()
     try:
@@ -434,6 +435,7 @@ def seed_kb_article(
                 article_id=article_id,
                 category_id=category_id,
                 article_group_id=resolved_group_id,
+                intent_code=intent_code,
                 language=language,
                 title=title,
                 body=body,
@@ -1347,6 +1349,88 @@ def test_voice_llm_guarded_decision_uses_operator_style_without_ai_or_kb(monkeyp
     assert "Do not say or imply that you are an AI" in system_prompt
 
 
+def test_voice_llm_decision_echoes_kb_article_intent_code(monkeypatch):
+    monkeypatch.setenv("AI_VOICE_POLICY_MODE", "llm_guarded")
+    monkeypatch.setattr(ai_module, "_ai_provider", lambda: "openai_compatible")
+
+    def _fake_structured(messages, **kwargs):
+        del messages
+        return {
+            "language": "ru",
+            "intent": "voucher_activation",
+            "reply_text": "Подтвердите СМС с номера 1414 командой 21*1, затем завершите активацию в eGov.",
+            "confidence": 0.9,
+            "needs_handoff": False,
+            "handoff_reason": None,
+            "case_action": "keep_open",
+            "kb_refs": ["kba_voucher_1"],
+            "_model": "gpt-test",
+            "_latency_ms": 30,
+            "_finish_reason": "stop",
+        }
+
+    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _fake_structured)
+
+    kb_article = SimpleNamespace(
+        article_id="kba_voucher_1",
+        title="Активация ваучера",
+        body="Подтвердите СМС 1414 командой 21*1, затем перейдите по ссылке и завершите в eGov Mobile.",
+        intent_code="VOUCHER_ACTIVATION",
+    )
+    decision = voice_module._voice_decision(
+        language="ru",
+        customer=None,
+        interaction=SimpleNamespace(interaction_id="int_voice_voucher", customer_id=None, status="new", queue_id="que_voice", subject="voucher"),
+        transcript_text="Что делать с СМС от 1414?",
+        transcript_window=[SimpleNamespace(speaker="caller", text="Что делать с СМС от 1414?", sequence_no=1, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso())],
+        kb_results=[kb_article],
+        disclosure_required=False,
+    )
+
+    assert decision["intent"] == "VOUCHER_ACTIVATION"
+
+
+def test_voice_llm_decision_rejects_invented_intent_not_in_kb_results(monkeypatch):
+    monkeypatch.setenv("AI_VOICE_POLICY_MODE", "llm_guarded")
+    monkeypatch.setattr(ai_module, "_ai_provider", lambda: "openai_compatible")
+
+    def _fake_structured(messages, **kwargs):
+        del messages
+        return {
+            "language": "ru",
+            "intent": "totally_made_up_intent",
+            "reply_text": "Подтвердите СМС с номера 1414 командой 21*1.",
+            "confidence": 0.9,
+            "needs_handoff": False,
+            "handoff_reason": None,
+            "case_action": "keep_open",
+            "kb_refs": ["kba_voucher_2"],
+            "_model": "gpt-test",
+            "_latency_ms": 30,
+            "_finish_reason": "stop",
+        }
+
+    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _fake_structured)
+
+    kb_article = SimpleNamespace(
+        article_id="kba_voucher_2",
+        title="Активация ваучера",
+        body="Подтвердите СМС 1414 командой 21*1.",
+        intent_code="VOUCHER_ACTIVATION",
+    )
+    decision = voice_module._voice_decision(
+        language="ru",
+        customer=None,
+        interaction=SimpleNamespace(interaction_id="int_voice_voucher_2", customer_id=None, status="new", queue_id="que_voice", subject="voucher"),
+        transcript_text="Куда отправлять 21*1?",
+        transcript_window=[SimpleNamespace(speaker="caller", text="Куда отправлять 21*1?", sequence_no=1, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso())],
+        kb_results=[kb_article],
+        disclosure_required=False,
+    )
+
+    assert decision["intent"] == "unknown"
+
+
 def test_voice_v2_fast_conversational_adds_ack_metadata_and_compacts_reply(monkeypatch):
     monkeypatch.setenv("AI_VOICE_POLICY_MODE", "v2_fast_conversational")
     monkeypatch.setattr(ai_module, "_ai_provider", lambda: "openai_compatible")

tests/test_intents.py

@@ -0,0 +1,34 @@
+from services.shared.intents import CONTROL_INTENTS, UNKNOWN_INTENT, normalize_intent
+
+
+def test_control_intent_passes_through_unchanged():
+    assert normalize_intent("handoff_request") == "handoff_request"
+    assert normalize_intent("kb_answer", known_topic_codes=["VOUCHER_ACTIVATION"]) == "kb_answer"
+
+
+def test_matching_topic_code_passes_through_case_insensitively():
+    assert normalize_intent("voucher_activation", known_topic_codes=["VOUCHER_ACTIVATION"]) == "VOUCHER_ACTIVATION"
+    assert normalize_intent("  Voucher_Activation  ", known_topic_codes=["voucher_activation"]) == "VOUCHER_ACTIVATION"
+
+
+def test_unknown_topic_code_falls_back_to_unknown():
+    assert normalize_intent("made_up_intent", known_topic_codes=["VOUCHER_ACTIVATION"]) == UNKNOWN_INTENT
+    assert normalize_intent("voucher_activation", known_topic_codes=[]) == UNKNOWN_INTENT
+
+
+def test_empty_or_missing_intent_falls_back_to_unknown():
+    assert normalize_intent(None) == UNKNOWN_INTENT
+    assert normalize_intent("") == UNKNOWN_INTENT
+    assert normalize_intent("   ") == UNKNOWN_INTENT
+
+
+def test_control_intents_frozenset_matches_documented_values():
+    assert CONTROL_INTENTS == {
+        "identity_question",
+        "handoff_request",
+        "sensitive_request",
+        "resolution_confirmed",
+        "clarification",
+        "kb_answer",
+        "unknown",
+    }

tests/test_kb_search.py

@@ -114,6 +114,61 @@ def test_search_kb_rows_prefers_title_and_tags_over_body_only_mentions():
     assert results[0].title == rows[0].title
 
 
+def test_search_kb_rows_ignores_stopword_only_query():
+    # A caller confirming the language ("да") should never surface an
+    # unrelated FAQ article just because that article's body happens to
+    # contain the word "да" somewhere in ordinary prose.
+    rows = [
+        _row(
+            row_id=1,
+            title="Активация ваучера",
+            body="Да, подтвердите СМС с номера 1414 командой 21*1.",
+            tags=["ваучер"],
+        ),
+    ]
+
+    assert search_kb_rows(rows, "да", limit=5) == []
+    assert search_kb_rows(rows, "Хорошо", limit=5) == []
+
+
+def test_search_kb_rows_still_matches_real_content_word_amid_fillers():
+    rows = [
+        _row(
+            row_id=1,
+            title="Активация ваучера",
+            body="Подтвердите СМС с номера 1414 командой 21*1.",
+            tags=["ваучер"],
+        ),
+        _row(
+            row_id=2,
+            title="График работы",
+            body="Филиалы работают с 9 до 18.",
+            tags=["график"],
+        ),
+    ]
+
+    results = search_kb_rows(rows, "да, у меня вопрос про ваучер", limit=5)
+
+    assert results
+    assert results[0].title == "Активация ваучера"
+
+
+def test_search_kb_rows_single_real_word_query_still_matches():
+    rows = [
+        _row(
+            row_id=1,
+            title="Активация ваучера",
+            body="Подтвердите СМС с номера 1414 командой 21*1.",
+            tags=["ваучер"],
+        ),
+    ]
+
+    results = search_kb_rows(rows, "ваучер", limit=5)
+
+    assert results
+    assert results[0].title == "Активация ваучера"
+
+
 def test_search_kb_rows_breaks_ties_by_newer_id():
     older = _row(
         row_id=10,

tests/test_stage3_services.py

@@ -40,6 +40,46 @@ def test_kb_lite_search():
     assert len(search.json()) >= 1
 
 
+def test_kb_article_intent_code_round_trips_through_create_get_update():
+    client = TestClient(kb_app)
+    headers = {"X-User": "analyst", "X-Role": "analyst"}
+
+    cat = client.post(
+        "/knowledge/categories",
+        json={"name": "Vouchers", "description": "Voucher help"},
+        headers=headers,
+    )
+    assert cat.status_code == 200
+    category_id = cat.json()["category_id"]
+
+    created = client.post(
+        "/knowledge/articles",
+        json={
+            "category_id": category_id,
+            "title": "Активация ваучера",
+            "body": "Подтвердите СМС 1414 командой 21*1.",
+            "tags": ["voucher"],
+            "intent_code": "voucher_activation",
+        },
+        headers=headers,
+    )
+    assert created.status_code == 200
+    assert created.json()["intent_code"] == "VOUCHER_ACTIVATION"
+    article_id = created.json()["article_id"]
+
+    fetched = client.get(f"/knowledge/articles/{article_id}")
+    assert fetched.status_code == 200
+    assert fetched.json()["intent_code"] == "VOUCHER_ACTIVATION"
+
+    updated = client.patch(
+        f"/knowledge/articles/{article_id}",
+        json={"intent_code": "voucher_activation_v2"},
+        headers=headers,
+    )
+    assert updated.status_code == 200
+    assert updated.json()["intent_code"] == "VOUCHER_ACTIVATION_V2"
+
+
 def test_kb_lite_search_ranks_title_over_body_only_matches():
     client = TestClient(kb_app)
     headers = {"X-User": "analyst", "X-Role": "analyst"}