Merge pull request 'fix: treat AST_CAUSE no-route/unallocated as a no-answer retry outcome' (#10) from fix/no-route-retry-cause into main

process_agent_dial_outcome only recognized hangup causes 17/18/19/21/34/38.
When the reserved agent's AOR has zero registered contacts (e.g. the
softphone dropped, or nobody ever registered), Asterisk immediately
hangs up with cause 3 (no route to destination) instead of running a
Dial() long enough to produce a DialEnd/NOANSWER at all - so the retry
listener silently ignored it and the escalation was left dangling in
'ringing' status (the agent itself still got released via the
call-ended fallback path, but no retry to the next agent was ever
attempted and the escalation record never reflects the failure).

Added causes 1 (unallocated number), 3 (no route), 20 (subscriber
absent), 22 (number changed) alongside the existing set.

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
     ]
@@ -2410,10 +2412,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 +2618,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

@@ -178,5 +178,10 @@ def operator_system_prompt(*, language: str, channel_label: str, is_voice: bool,
         "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."
+        "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

@@ -19,6 +19,7 @@ from services.shared.ai_context_summary import (
 )
 from services.shared.core import new_id, utc_now_iso
 from services.shared.db import get_session
+from services.shared.intents import normalize_intent
 from services.shared.models import VoiceAIStartIn, VoiceAIStartOut, VoiceAITurnDecisionOut, VoiceAITurnIn, VoiceStartResult
 from services.shared.sql_models import (
     AISessionRow,
@@ -1359,7 +1360,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 +1516,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,
@@ -1528,6 +1535,7 @@ def _voice_early_plan(
     transcript_text: str,
     context_summary: str | dict[str, Any] | None = None,
     request_metadata: dict[str, Any] | None = None,
+    kb_results: list[Any] = (),
 ) -> dict[str, Any]:
     v2_metadata = _voice_v2_metadata(transcript_text, request_metadata)
     payload = request_metadata if isinstance(request_metadata, dict) else {}
@@ -1570,10 +1578,56 @@ def _voice_early_plan(
                 "reply_phase": "early_plan",
             },
         }
-    if early_intent in {"schedule", "address", "price", "status", "problem", "operator_request"}:
+    if early_intent == "operator_request":
+        return {
+            "language": language,
+            "intent": "handoff_request",
+            "reply_text": _voice_compact_reply_text(_voice_handoff_reply(language), language=language),
+            "confidence": 0.62,
+            "needs_handoff": True,
+            "handoff_reason": "Запрос требует участия живого оператора.",
+            "case_action": "keep_open",
+            "kb_refs": [],
+            "summary_text": "Early domain plan is prepared.",
+            "model": "voice_early_plan_domain",
+            "latency_ms": 1,
+            "metadata": {
+                **v2_metadata,
+                "reply_phase": "early_plan",
+                "early_intent": early_intent,
+            },
+        }
+    if early_intent in {"schedule", "address", "price", "status", "problem"} and kb_results:
+        # A cheap lexical KB lookup fits the early-plan latency budget (no LLM
+        # round-trip), unlike the properly grounded/paraphrased "final" decision.
+        # Answering from the KB here beats guessing a generic clarifying question
+        # when the FAQ already has the answer — see the plan doc for why this
+        # branch exists at all (the early reply can win the race and get spoken
+        # before the final, LLM-grounded decision is ready).
+        article = kb_results[0]
+        snippet = _app()._article_snippet(article, limit=220)
+        reply_text = f"Қысқаша айтайын: {snippet}" if language == "kz" else f"Коротко подскажу: {snippet}"
+        topic_code = getattr(article, "intent_code", None)
+        return {
+            "language": language,
+            "intent": normalize_intent(topic_code or "kb_answer", known_topic_codes=[topic_code] if topic_code else []),
+            "reply_text": _voice_compact_reply_text(reply_text, language=language),
+            "confidence": 0.7,
+            "needs_handoff": False,
+            "handoff_reason": None,
+            "case_action": "keep_open",
+            "kb_refs": [article.article_id],
+            "summary_text": "AI ответил по базе ЧЗВ на предварительной стадии.",
+            "model": "voice_early_plan_kb",
+            "latency_ms": 1,
+            "metadata": {
+                **v2_metadata,
+                "reply_phase": "early_plan",
+                "early_intent": early_intent,
+            },
+        }
+    if early_intent in {"schedule", "address", "price", "status", "problem"}:
         reply_text = _voice_summary_slot_prompt(language, context_summary) or _voice_topic_prompt(language, [transcript_text])
-        if early_intent == "operator_request":
-            reply_text = _voice_handoff_reply(language)
         if not reply_text and language == "kz":
             if early_intent == "schedule":
                 reply_text = "Qai filialdyn, mekenjaidyn nemese qalanyng jumys uaqyty qyzyqtyratynyn aitnyz."
@@ -1599,15 +1653,11 @@ def _voice_early_plan(
         if reply_text:
             return {
                 "language": language,
-                "intent": "handoff_request" if early_intent == "operator_request" else "clarification",
+                "intent": "clarification",
                 "reply_text": _voice_compact_reply_text(reply_text, language=language),
                 "confidence": 0.62,
-                "needs_handoff": early_intent == "operator_request",
-                "handoff_reason": (
-                    "Запрос требует участия живого оператора."
-                    if early_intent == "operator_request"
-                    else None
-                ),
+                "needs_handoff": False,
+                "handoff_reason": None,
                 "case_action": "keep_open",
                 "kb_refs": [],
                 "summary_text": "Early domain plan is prepared.",
@@ -1668,6 +1718,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 +1818,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"):
@@ -1823,6 +1880,7 @@ def _voice_decision(
             transcript_text=transcript_text,
             context_summary=context_summary,
             request_metadata=request_metadata,
+            kb_results=kb_results,
         )
 
     if persona.is_identity_request(normalized):
@@ -2524,13 +2582,17 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe
             )
             ai_session.context_summary_json = dump_context_summary(user_context_summary)
             ai_session.context_summary_updated_at = now
-        kb_results = []
-        if not early_plan_only:
-            kb_results = app._kb_search(
-                session,
-                payload.transcript_text,
-                language=ai_session.language,
-            )
+        # Unlike name extraction/context-summary writes above (skipped for
+        # early_plan_only since they're stateful and heavier), KB search is a
+        # cheap in-memory lexical scan over a DB-cached row set (see
+        # _load_kb_rows_cached) and comfortably fits the early-plan latency
+        # budget, so it always runs — this lets _voice_early_plan answer from
+        # the FAQ instead of guessing a generic clarifying question.
+        kb_results = app._kb_search(
+            session,
+            payload.transcript_text,
+            language=ai_session.language,
+        )
         disclosure_required = voice_session.disclosure_played_at is None
         decision = _voice_decision(
             language=ai_session.language or "ru",

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/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

@@ -25,6 +25,7 @@ class KBSearchRow(Protocol):
     title: str
     body: str
     tags_json: str
+    intent_code: str | None
 
 
 T = TypeVar("T", bound=KBSearchRow)

services/shared/models.py

@@ -983,6 +983,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 +992,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")
@@ -1505,6 +1589,79 @@ def test_voice_v2_streaming_duplex_early_plan_returns_domain_followup_without_ll
     assert decision["metadata"]["early_intent"] == "schedule"
 
 
+def test_voice_v2_streaming_duplex_early_plan_answers_from_kb_without_llm(monkeypatch):
+    monkeypatch.setenv("AI_VOICE_POLICY_MODE", "v2_streaming_duplex")
+
+    def _unexpected_llm(messages, **kwargs):
+        raise AssertionError(f"LLM should not be called for early plan: {messages!r}")
+
+    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _unexpected_llm)
+
+    kb_article = SimpleNamespace(
+        article_id="kba_early_schedule",
+        title="График работы филиалов",
+        body="Филиалы работают с понедельника по пятницу с 9:00 до 18:00.",
+        intent_code="BRANCH_SCHEDULE",
+    )
+    decision = voice_module._voice_decision(
+        language="ru",
+        customer=None,
+        interaction=SimpleNamespace(interaction_id="int_voice_early_schedule_kb", status="new", queue_id="que_voice", subject="unknown"),
+        transcript_text="Мне надо узнать график работы",
+        transcript_window=[],
+        kb_results=[kb_article],
+        disclosure_required=False,
+        request_metadata={
+            "voice_v2_enabled": True,
+            "reply_phase": "early_plan",
+            "response_plan_id": "rsp_early_schedule_kb",
+            "early_intent": "schedule",
+        },
+    )
+
+    assert decision["model"] == "voice_early_plan_kb"
+    assert decision["intent"] == "BRANCH_SCHEDULE"
+    assert decision["kb_refs"] == ["kba_early_schedule"]
+    assert decision["needs_handoff"] is False
+    assert "9:00" in decision["reply_text"] or "9" in decision["reply_text"]
+    assert decision["metadata"]["reply_phase"] == "early_plan"
+
+
+def test_voice_v2_streaming_duplex_early_plan_operator_request_skips_kb(monkeypatch):
+    monkeypatch.setenv("AI_VOICE_POLICY_MODE", "v2_streaming_duplex")
+
+    def _unexpected_llm(messages, **kwargs):
+        raise AssertionError(f"LLM should not be called for early plan: {messages!r}")
+
+    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _unexpected_llm)
+
+    kb_article = SimpleNamespace(
+        article_id="kba_should_not_be_used",
+        title="Unrelated article",
+        body="Should not be referenced for an operator handoff.",
+        intent_code="SOMETHING_ELSE",
+    )
+    decision = voice_module._voice_decision(
+        language="ru",
+        customer=None,
+        interaction=SimpleNamespace(interaction_id="int_voice_early_operator", status="new", queue_id="que_voice", subject="unknown"),
+        transcript_text="Соедините меня с оператором",
+        transcript_window=[],
+        kb_results=[kb_article],
+        disclosure_required=False,
+        request_metadata={
+            "voice_v2_enabled": True,
+            "reply_phase": "early_plan",
+            "response_plan_id": "rsp_early_operator",
+            "early_intent": "operator_request",
+        },
+    )
+
+    assert decision["intent"] == "handoff_request"
+    assert decision["needs_handoff"] is True
+    assert decision["kb_refs"] == []
+
+
 def test_voice_decision_hearing_check_keeps_active_topic_without_llm(monkeypatch):
     def _unexpected_llm(messages, **kwargs):
         raise AssertionError(f"LLM should not be called for hearing check: {messages!r}")

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_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"}