Author SHA1 Message Date
arys 8382dfa9ba fix: treat AST_CAUSE no-route/unallocated as a no-answer retry outcome
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.
2026-08-31 00:43:20 +05:00
didar d2438b6954 feat: canonical intent taxonomy for AI operator (kb_answer -> intent_code)
deploy / deploy (push) Successful in 30s
Centralizes fixed control intents and adds a data-driven intent_code
field on kb_articles so many phrasings of the same FAQ question
resolve to one stable code (e.g. VOUCHER_ACTIVATION) instead of a
free-form, unvalidated string the LLM invented on the fly.

- services/shared/intents.py: CONTROL_INTENTS + normalize_intent()
- kb_articles.intent_code column (ORM + dev/sqlite runtime compat +
  migrations/sql/0034_* for postgres/sqlite)
- kb_service CRUD exposes intent_code
- orchestrator surfaces intent_code to the LLM and validates its
  intent output against control intents + the KB codes shown that turn
- voice.py: _voice_early_intent_bucket renamed to _voice_ack_topic_bucket
  to stop it being conflated with the canonical FAQ intent
2026-08-31 00:17:51 +05:00
arystanbek 3826f5704a Merge pull request 'fix: track escalation for legacy AI voice handoff path' (#9) from fix/handoff-escalation-tracking into main
deploy / deploy (push) Successful in 31s
2026-08-30 19:08:20 +00:00
arys 2cb358e80d fix: track escalation for legacy AI voice handoff path
request_handoff() (called by ai_voice_runtime_service for every real
call handoff, the only handoff path production calls actually use)
reserved an agent from the same routing pool as create_escalation()
but never created an EscalationRow, so the Phase 2 no-answer-retry
listener (DialEnd/Hangup) could never find it. A failed transfer
(no SIP registration, no answer, redirect error) left the agent
stuck in RESERVED forever with no retry to the next agent.

Now creates an EscalationRow (status=ringing) alongside the agent
reservation, releases the agent + marks the escalation failed if the
AMI Redirect itself errors immediately, and lets the existing
DialEnd/Hangup handler drive no-answer retry / release exactly like
the /escalations endpoint already does.

Reproduced live: call handed off to extension 2002 with no SIP
contact registered -> immediate hangup, cause=3, both pool agents
stuck in RESERVED indefinitely (had to release manually via psql).
2026-08-31 00:07:57 +05:00
didar 2f4a9795b5 fix: small fixes on filler phrases
deploy / deploy (push) Successful in 34s
2026-08-30 23:52:06 +05:00
didar 48d1fabba1 fix: update ElevenLabs Russian and Kazakh voice IDs for TTS
deploy / deploy (push) Successful in 32s
2026-08-30 23:44:05 +05:00
arystanbek 66652845d8 Merge pull request 'feat: no-answer retry, agent status machine, escalation events (ТЗ Phase 2)' (#8) from feature/escalation-no-answer-retry-and-agent-state into main
deploy / deploy (push) Successful in 32s
2026-08-30 09:19:59 +00:00
arys 1dcfaf46cf feat: no-answer retry, agent status machine, escalation events/timeline (ТЗ §13-15,22,25-27,37, AC-08,16-19)
Phase 2 of the L1->L2 routing engine (Phase 1: MR!4).

- ami_loop() now also captures native AMI DialEnd/Hangup frames (not
  only UserEvent), needed to detect that an escalated agent did not
  answer. No dialplan change required - Redirect already routes the
  client channel into an existing Dial()-based transfer context, so
  Asterisk emits these events on its own; the listener just wasn't
  reading them before.
- retry_escalation_no_answer(): on NOANSWER/BUSY/CANCEL/CHANUNAVAIL/
  CONGESTION, releases the non-answering agent, excludes it, and
  reserves+redirects to the next available agent via the routing
  engine's existing exclude_agent_ids support. Exhausted pool marks
  the escalation failed and leaves the call with the AI instead of
  dropping the client (ТЗ §32).
- Agent status now actually moves through
  RESERVED -> RINGING -> TALKING -> AFTER_CALL_WORK -> AVAILABLE
  instead of staying stuck on RESERVED for the whole call; a new
  acw_sweep_loop background thread (same pattern as the existing
  failed_retry_loop) times out AFTER_CALL_WORK back to AVAILABLE.
- escalations gains attempt_count/real_agent_id/attempted_agent_ids_json
  (migration 0033); fixes a latent bug where assigned_agent_id stored
  the SIP extension instead of the real agent_id despite routing-service
  already returning it in RoutingAgentReserveOut.
- Every transition now records an interaction timeline entry and
  publishes the ТЗ §25 event catalog (AgentReserved/AgentRinging/
  AgentNoAnswer/AgentConnected/TransferCompleted/TransferFailed)
  through the existing emit_voice_event/EventOutboxRow idempotent path.

Not in this MR (see plan): SLA config, Callback, L3 (needs real
technical agents from the business), metrics.
2026-08-30 14:19:23 +05:00
didar 010a8dcab6 fix: ensure emotive ack rotation and prebaked ack caching apply outside v2 queue eligibility
deploy / deploy (push) Successful in 30s
2026-08-30 12:49:26 +05:00
Codex 13ba8f8b56 merge: bring aimaq host-override files back into git (fix/aimaq-remove-host-override)
deploy / deploy (push) Successful in 32s
2026-08-30 07:46:21 +00:00
arystanbek 9d918af12a fix: bring aimaq host-override files (voice.py, persona env) back into git
Both services/ai_orchestrator_service/voice.py and the aimaq persona/DOMAIN
SCOPE prompt were bind-mounted straight from the host on the aimaq stack,
bypassing git and CI entirely since they were first hand-edited in prod.

voice.py: merged the host's live business logic (gas/aimaq domain keyword
list, off-domain Kazakh/Russian replies, disabled re-correction of an already
obtained name) with the timeout_seconds fix from 09bcf74 that never reached
aimaq because the bind mount blocked it.

aimaq.env.production: replaced the AI_OPERATOR_* env values (which were never
interpolated -- {agent_name}/{company_name} would have been read literally)
with the final resolved Zhanna/Kazakgaz Aimaq text including the DOMAIN SCOPE
clause, matching what was actually live on the host.
2026-08-30 07:45:35 +00:00
arystanbek ff85fa27fb Merge pull request 'feat: aimaq voice AI persona Zhanna + language-choice greeting' (#7) from feature/aimaq-zhanna-persona-language-choice into main
deploy / deploy (push) Successful in 30s
2026-08-30 07:22:16 +00:00
35 changed files with 1211 additions and 112 deletions
+13
View File
@@ -0,0 +1,13 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "AgentConnected",
"type": "object",
"required": ["event", "call_id", "escalation_id", "agent_id", "created_at"],
"properties": {
"event": { "const": "AgentConnected" },
"call_id": { "type": "string" },
"escalation_id": { "type": "string" },
"agent_id": { "type": "string" },
"created_at": { "type": "string", "format": "date-time" }
}
}
+15
View File
@@ -0,0 +1,15 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "AgentNoAnswer",
"type": "object",
"required": ["event", "call_id", "escalation_id", "agent_id", "dial_outcome", "created_at"],
"properties": {
"event": { "const": "AgentNoAnswer" },
"call_id": { "type": "string" },
"escalation_id": { "type": "string" },
"agent_id": { "type": ["string", "null"] },
"dial_outcome": { "type": "string" },
"attempt_count": { "type": "integer" },
"created_at": { "type": "string", "format": "date-time" }
}
}
+18
View File
@@ -0,0 +1,18 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "AgentReserved",
"type": "object",
"required": ["event", "call_id", "escalation_id", "agent_id", "created_at"],
"properties": {
"event": { "const": "AgentReserved" },
"call_id": { "type": "string" },
"escalation_id": { "type": "string" },
"agent_id": { "type": "string" },
"tenant_id": { "type": ["string", "null"] },
"from_level": { "type": "string" },
"to_level": { "type": "string" },
"reason_code": { "type": "string" },
"attempt_count": { "type": "integer" },
"created_at": { "type": "string", "format": "date-time" }
}
}
+13
View File
@@ -0,0 +1,13 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "AgentRinging",
"type": "object",
"required": ["event", "call_id", "escalation_id", "agent_id", "created_at"],
"properties": {
"event": { "const": "AgentRinging" },
"call_id": { "type": "string" },
"escalation_id": { "type": "string" },
"agent_id": { "type": "string" },
"created_at": { "type": "string", "format": "date-time" }
}
}
+13
View File
@@ -0,0 +1,13 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "TransferCompleted",
"type": "object",
"required": ["event", "call_id", "escalation_id", "agent_id", "created_at"],
"properties": {
"event": { "const": "TransferCompleted" },
"call_id": { "type": "string" },
"escalation_id": { "type": "string" },
"agent_id": { "type": "string" },
"created_at": { "type": "string", "format": "date-time" }
}
}
+15
View File
@@ -0,0 +1,15 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "TransferFailed",
"type": "object",
"required": ["event", "call_id", "created_at"],
"properties": {
"event": { "const": "TransferFailed" },
"call_id": { "type": "string" },
"escalation_id": { "type": ["string", "null"] },
"agent_id": { "type": ["string", "null"] },
"reason": { "type": ["string", "null"] },
"error": { "type": ["string", "null"] },
"created_at": { "type": "string", "format": "date-time" }
}
}
+7 -7
View File
@@ -84,9 +84,9 @@ AI_VOICE_TTS_CACHE_DIR=/app/.data_local/ai_voice_tts_cache
AI_VOICE_TTS_ELEVENLABS_API_KEY=sk_1c50faab05df13fd0ecde9aeca4abadcc7269ffdc15a03ad
AI_VOICE_TTS_ELEVENLABS_API_BASE=https://api.elevenlabs.io
AI_VOICE_TTS_ELEVENLABS_MODEL_ID=eleven_turbo_v2_5
AI_VOICE_TTS_ELEVENLABS_RU_VOICE_ID=0ArNnoIAWKlT4WweaVMY
AI_VOICE_TTS_ELEVENLABS_RU_VOICE_ID=ut2XM2wJyIZLTtW6lFzZ
AI_VOICE_TTS_ELEVENLABS_RU_LANGUAGE_CODE=ru
AI_VOICE_TTS_ELEVENLABS_KK_VOICE_ID=0ArNnoIAWKlT4WweaVMY
AI_VOICE_TTS_ELEVENLABS_KK_VOICE_ID=ut2XM2wJyIZLTtW6lFzZ
AI_VOICE_TTS_ELEVENLABS_KK_LANGUAGE_CODE=kk
AI_VOICE_TTS_ELEVENLABS_OUTPUT_FORMAT=pcm_16000
AI_VOICE_TTS_YANDEX_API_KEY=AQWJUMiUaXmbegxN4kgvM2XIlNqAPoBR5Wtq-40
@@ -187,8 +187,8 @@ ERROR_TELEGRAM_ALERT_BOT_TOKEN=8734026216:AAEK-xdgVR5al6rfKp_DIb-HSJZXGUG2Tvg
ERROR_TELEGRAM_ALERT_CHANNEL_ID=t.me/konturaitelecom
AI_OPERATOR_AGENT_NAME=Жанна
AI_OPERATOR_COMPANY_NAME=Казакгаз Аймак
AI_OPERATOR_VOICE_GREETING_RU=Добрый день! Меня зовут {agent_name}, я ии-оператор контакт-центра {company_name}. Вам удобнее получить консультацию на казахском или на русском языке?
AI_OPERATOR_VOICE_GREETING_KZ=Қайырлы күн! Менің атым {agent_name}, мен «{company_name}» байланыс орталығында жасанды интеллект операторымын. Кеңесті қазақ немесе орыс тілінде алғыңыз келе ме?
AI_OPERATOR_IDENTITY_REPLY_RU=Я {agent_name}, ии-оператор контакт-центра {company_name}. Чем могу помочь?
AI_OPERATOR_IDENTITY_REPLY_KZ=Мен {agent_name}мын, «{company_name}» байланыс орталығының жасанды интеллект операторымын. Қалай көмектесе аламын?
AI_OPERATOR_BASE_SYSTEM_PROMPT=Ты {agent_name}, ии-оператор контакт-центра {company_name} для звонков, Telegram и других каналов. Всегда сохраняй одну и ту же личность: тебя зовут {agent_name}. Если клиент спрашивает, кто ты или как тебя зовут, отвечай, что ты {agent_name}. В начале разговора, сразу после приветствия, ты уже спросила клиента, на каком языке ему удобнее — на казахском или на русском. Как только клиент ответит, полностью веди остаток разговора на выбранном им языке и больше не спрашивай про язык повторно. После того как язык определён, уточни у клиента, как к нему обращаться, и только затем переходи к сути вопроса. Прежде чем завершить разговор, обязательно спроси, нужна ли клиенту ещё какая-то помощь, и попрощайся («До свидания» или «Қош болыңыз») только после того, как клиент подтвердит, что вопросов больше нет — не завершай диалог самостоятельно. Отвечай естественно, кратко и по делу. Когда говоришь о себе, используй женский род: могла, смогла, сделала, готова, проверила, нашла. Не используй Markdown, URL или таблицы.
AI_OPERATOR_VOICE_GREETING_RU=Добрый день! Меня зовут Жанна, я ИИ-оператор контакт-центра Казакгаз Аймак. Вам удобнее получить консультацию на казахском или на русском языке?
AI_OPERATOR_VOICE_GREETING_KZ=Қайырлы күн! Менің атым Жанна, мен «Қазақгаз Аймақ» байланыс орталығында жасанды интеллект операторымын. Кеңесті қазақ немесе орыс тілінде алғыңыз келе ме?
AI_OPERATOR_IDENTITY_REPLY_RU=Я Жанна, ИИ-оператор контакт-центра Казакгаз Аймак. Чем могу помочь?
AI_OPERATOR_IDENTITY_REPLY_KZ=Мен Жаннамын, «Қазақгаз Аймақ» байланыс орталығының жасанды интеллект операторымын. Қалай көмектесе аламын?
AI_OPERATOR_BASE_SYSTEM_PROMPT=Ты Жанна, единый ИИ-оператор контакт-центра Казакгаз Аймак для звонков, Telegram и других каналов. Всегда сохраняй одну и ту же личность: тебя зовут Жанна. Если клиент спрашивает, кто ты или как тебя зовут, отвечай, что ты Жанна. В начале разговора, сразу после приветствия, ты уже спросила клиента, на каком языке ему удобнее — на казахском или на русском. Как только клиент ответит, полностью веди остаток разговора на выбранном им языке и больше не спрашивай про язык повторно. После того как язык определён, уточни у клиента, как к нему обращаться, и только затем переходи к сути вопроса. Прежде чем завершить разговор, обязательно спроси, нужна ли клиенту ещё какая-то помощь, и попрощайся («До свидания» или «Қош болыңыз») только после того, как клиент подтвердит, что вопросов больше нет — не завершай диалог самостоятельно. Отвечай естественно, кратко и по делу. Когда говоришь о себе, используй женский род: могла, смогла, сделала, готова, проверила, нашла. Телефонные номера читай по цифрам. Не используй Markdown, URL или таблицы. DOMAIN SCOPE: ты отвечаешь ТОЛЬКО на вопросы, связанные с услугами газоснабжения: оплата газа, счётчики, ваучеры, технические условия, отключение газа, мобильное приложение, квитанции, передача показаний, контакты филиалов, безопасность газа. Если вопрос НЕ связан с газом (погода, спорт, политика, другие компании, личные советы, кулинария, рецепты, анекдоты, новости, фильмы), вежливо откажи БЕЗ поиска в базе знаний и БЕЗ попытки ответить: RU «Я могу ответить только на вопросы, связанные с услугами газоснабжения. Пожалуйста, задайте вопрос по теме газа.» KK «Мен тек газмен қамтамасыз ету қызметтеріне байланысты сұрақтарға жауап беремін. Газ тақырыбында сұрақ қойыңызшы.» Не пытайся ответить на вопрос не по теме, даже если знаешь ответ.
@@ -0,0 +1,4 @@
ALTER TABLE escalations ADD COLUMN IF NOT EXISTS attempt_count INTEGER NOT NULL DEFAULT 0;
ALTER TABLE escalations ADD COLUMN IF NOT EXISTS real_agent_id TEXT;
ALTER TABLE escalations ADD COLUMN IF NOT EXISTS attempted_agent_ids_json TEXT NOT NULL DEFAULT '[]';
CREATE INDEX IF NOT EXISTS ix_escalations_real_agent_id ON escalations(real_agent_id);
@@ -0,0 +1,4 @@
ALTER TABLE escalations ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 0;
ALTER TABLE escalations ADD COLUMN real_agent_id TEXT;
ALTER TABLE escalations ADD COLUMN attempted_agent_ids_json TEXT NOT NULL DEFAULT '[]';
CREATE INDEX IF NOT EXISTS ix_escalations_real_agent_id ON escalations(real_agent_id);
@@ -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);
@@ -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);
+17 -4
View File
@@ -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(
@@ -179,4 +179,9 @@ def operator_system_prompt(*, language: str, channel_label: str, is_voice: bool,
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."
)
+95 -36
View File
@@ -604,16 +604,6 @@ def _voice_has_name_correction(text: str | None) -> bool:
def _voice_reply_already_names_customer(reply_text: str, normalized_short: str) -> bool:
"""Whether `reply_text` already addresses the customer by name.
Russian and Kazakh names decline grammatically (e.g. "Данияр" -> "Данияре",
"Данияру", "Данияра"), so the model's own reply often names the customer in
an inflected form that never exact-matches `normalized_short`. An exact
token match alone misses that and this function then goes on to prepend
the name a second time, producing a literal "Данияр, Здравствуйте, Данияре!"
duplicate. Matching on a stem prefix instead of the full token catches any
declined form of the same name.
"""
tokens = _voice_text_key(reply_text).split(" ")
if normalized_short in tokens:
return True
@@ -635,14 +625,6 @@ def _voice_reply_with_name(
*,
greet: bool = False,
) -> str:
"""Prefix `reply_text` with the customer's name.
The prefix is always one of exactly two deterministic forms "{name}, ..."
or "Здравствуйте, {name}, ..." (`greet=True`, used the turn their name is
first learned) never left to the model's own free-form phrasing. That is
what keeps this from colliding with a self-introduced name inside
`reply_text` in the first place (see `_voice_reply_already_names_customer`).
"""
short_name = _voice_short_name(name)
if not short_name:
return reply_text
@@ -786,13 +768,7 @@ def _voice_downstream_name_update(
action = "provide" if not name_value else "confirm"
candidate = candidate or name_value
elif status == "name_obtained":
if (
candidate
and candidate_key
and candidate_key != current_key
and _voice_has_name_correction(transcript_text)
):
action = "correct"
pass
else:
if explicit_candidate:
action = "provide"
@@ -1079,7 +1055,6 @@ def _voice_has_service_topic(text: str | None) -> bool:
"ошибк",
"проблем",
"сбой",
"интернет",
"связь",
"оператор",
"менеджер",
@@ -1087,6 +1062,40 @@ def _voice_has_service_topic(text: str | None) -> bool:
"компан",
"подключ",
"доставк",
"газ",
"счётчик",
"счетчик",
"ваучер",
"отключ",
"квитанц",
"показан",
"приложен",
"безопасн",
"техническ",
"плит",
"котл",
"труб",
"утечк",
"запах",
"абонент",
"договор",
"поверк",
"монтаж",
"счёт",
"счет",
"долг",
"задолжен",
"перерасчёт",
"перерасчет",
"регион",
"аимак",
"aimaq",
"qazaqgaz",
"казахгаз",
"есептегіш",
"төлеу",
"өтінім",
"шарт",
)
return any(marker in normalized for marker in service_markers)
@@ -1120,6 +1129,37 @@ def _voice_is_off_domain_request(text: str | None) -> bool:
"java",
"javascript",
"погод",
"плов",
"приготов",
"кулинар",
"блюдо",
"фильм",
"кино",
"спорт",
"футбол",
"хоккей",
"теннис",
"баскетбол",
"президент",
"политик",
"выборы",
"правительств",
"курс валют",
"криптовалют",
"биткоин",
"песн",
"музык",
"танц",
"шутк",
"загадк",
"сериал",
"книга",
"стихотвор",
"чемпионат",
"ауа райы",
"аспаздық",
"кітап",
"ән айт",
)
if any(marker in normalized for marker in broad_markers):
return True
@@ -1129,7 +1169,14 @@ def _voice_is_off_domain_request(text: str | None) -> bool:
"что такое",
"объясни",
"расскажи про",
"почему",
"посоветуй",
"как приготовить",
"какая погода",
"кто президент",
"кто выиграл",
"какая команда",
"ауа райы қалай",
"әнді айт",
)
return any(normalized.startswith(prefix) for prefix in broad_openers)
@@ -1137,13 +1184,12 @@ def _voice_is_off_domain_request(text: str | None) -> bool:
def _voice_off_domain_reply(language: str) -> tuple[str, str]:
if language == "kz":
return (
"Men kompaniyamyzdyn qyzmetteri men otinishteri boiynsha komek bere alamyn. "
"Eger suraq bizdin qyzmetke qatysty bolsa, qysqasha naqtylaңыз. Qalasaңыз, operatorga qosamyn.",
"AI qongyraudyn taqyrybyn kompaniya qyzmetteri sheginde naqtylaudy usyndy.",
"Кешіріңіз, мен тек газ қызметтері бойынша көмектесемін: төлеу, есептегіш, ваучер. Қалай көмектесе аламын?",
"AI off-topic сұрауды газ қызметтері тақырыбына шектеді.",
)
return (
"Я помогу по вопросам наших услуг и обращений. Если вопрос связан с нашей компанией, скажите коротко, что именно нужно. Если хотите, сразу соединю с оператором.",
"AI мягко вернул разговор к вопросам компании и предложил перевод на оператора.",
"Извините, я помогаю только по вопросам газа: оплата, счётчики, ваучеры. Чем могу помочь?",
"AI отклонил off-topic вопрос и ограничил тему газоснабжением.",
)
@@ -1313,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"
@@ -1463,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,
@@ -1622,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]
]
@@ -1721,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"):
@@ -1856,7 +1915,7 @@ def _voice_decision(
"latency_ms": 1,
}
if not kb_results and _voice_is_off_domain_request(normalized):
if _voice_is_off_domain_request(normalized):
reply_text, summary_text = _voice_off_domain_reply(language)
decision = {
"language": language,
+8 -5
View File
@@ -822,11 +822,14 @@ def _media_registration_from_row(row: VoiceAISessionRow, *, queue_code: str | No
voice_v2_partial_asr=bool(voice_v2_for_session and _voice_v2_partial_asr_enabled()),
voice_v2_duplex=bool(voice_v2_for_session and _voice_v2_duplex_enabled()),
voice_v2_streaming_asr_backend=streaming_backend,
voice_v2_prebaked_ack=bool(voice_v2_for_session and _voice_v2_prebaked_ack_enabled()),
# Deliberately NOT gated on voice_v2_for_session: phrase-variant rotation only
# needs a text pool + live TTS, not the v2 duplex/partial-ASR pipeline, so calls
# outside the v2 queue allowlist still get varied fillers instead of always the
# single fixed "Секунду." fallback string.
# voice_v2_prebaked_ack and voice_v2_emotive_ack are deliberately NOT gated on
# voice_v2_for_session: filler-ack synthesis/caching and phrase-variant rotation
# only need a text pool + the ack bank, not the v2 duplex/partial-ASR pipeline.
# Without this, calls outside the v2 queue allowlist would synthesize every
# filler live via ElevenLabs before it could play — adding real TTS round-trip
# time to the one phrase whose whole job is to hide that latency — and would
# always get the single fixed "Секунду." fallback string instead of rotating.
voice_v2_prebaked_ack=bool(_voice_v2_prebaked_ack_enabled()),
voice_v2_emotive_ack=bool(_voice_v2_emotive_ack_enabled()),
voice_v2_emotive_ack_ru_only=bool(_voice_v2_emotive_ack_ru_only()),
)
@@ -482,7 +482,7 @@ class AudioSocketMediaRuntime:
if ack_kind == "handoff":
return (
"Угу, секунду.",
"Понял вас, соединяю.",
"Понялa вас, соединяю.",
"Мхм, соединяю.",
"Хорошо, сейчас соединю.",
"Да, сейчас соединю.",
@@ -492,7 +492,7 @@ class AudioSocketMediaRuntime:
return (
"Угу, сейчас подскажу.",
"Мхм, сориентирую.",
"Ага, понял вас.",
"Ага, понялa вас.",
"Хм, сейчас уточню.",
"Хорошо, сейчас подскажу.",
"Ясно, сейчас разберусь.",
@@ -509,14 +509,12 @@ class AudioSocketMediaRuntime:
"Дайте уточню.",
)
return (
"Ага.",
"Ясно, минутку.",
"Угу.",
"Хорошо, сейчас.",
"Так, слушаю.",
"Момент.",
"Понял вас.",
"Хм, сейчас гляну.",
"Так, слушаю вас.",
"Один момент.",
"Поняла вас.",
"Хмм, сейчас гляну.",
)
def _select_ack_payload(
+7 -2
View File
@@ -412,13 +412,18 @@ def ami_loop(stop_event=None) -> None:
raise RuntimeError("AMI connection closed")
if not frame:
continue
if frame.get("Event") != "UserEvent":
continue
event_name = frame.get("Event")
if event_name == "UserEvent":
user_event = str(frame.get("UserEvent") or "").strip()
if not user_event.startswith(bridge._ami_prefix()):
continue
bridge._STATE.set_last_event()
bridge._record_ami_payload(frame)
elif event_name in {"DialEnd", "Hangup"}:
bridge._STATE.set_last_event()
bridge._record_ami_payload(frame, event_name=event_name)
else:
continue
except Exception as exc:
bridge._STATE.set_connected(False)
bridge._STATE.set_error(str(exc))
+6
View File
@@ -262,6 +262,7 @@ def _start_background_threads() -> None:
extra_loops: list[tuple[str, Callable[..., Any]]] = []
if _ivr_fastagi_enabled():
extra_loops.append(("asterisk-ivr-fastagi-loop", _ivr_fastagi_loop))
extra_loops.append(("agent-acw-sweep-loop", _acw_sweep_loop))
bridge_runtime.start_background_threads(
ami_loop=_ami_loop,
failed_retry_loop=_failed_retry_loop,
@@ -305,6 +306,9 @@ _update_voice_ai_call_state = bridge_voice_ai.update_call_ai_state
_voice_ai_summary_for_call = bridge_voice_ai.voice_ai_summary_for_call
_create_escalation = bridge_voice_ai.create_escalation
_release_routing_agent = bridge_voice_ai.release_routing_agent
_retry_escalation_no_answer = bridge_voice_ai.retry_escalation_no_answer
_routing_release_by_agent_id = bridge_voice_ai.routing_release_by_agent_id
_set_routing_agent_status = bridge_voice_ai.set_routing_agent_status
_first_non_empty = bridge_ami.first_non_empty
_extract_call_id = bridge_ami.extract_call_id
@@ -373,6 +377,8 @@ _process_audio_bridge_ended = bridge_processing.process_audio_bridge_ended
_process_call_ended = bridge_processing.process_call_ended
_process_operator_connected = bridge_processing.process_operator_connected
_process_recording_ready = bridge_processing.process_recording_ready
_process_agent_dial_outcome = bridge_processing.process_agent_dial_outcome
_acw_sweep_loop = bridge_processing.acw_sweep_loop
_process_bridge_row = bridge_processing.process_bridge_row
_record_ami_payload = bridge_processing.record_ami_payload
_retry_failed_events_once = bridge_processing.retry_failed_events_once
@@ -4,6 +4,7 @@ from datetime import datetime, timezone
import hashlib
import json
import logging
import os
from pathlib import Path
import threading
from typing import Any
@@ -884,18 +885,34 @@ def process_call_ended(
open_escalation = session.execute(
select(EscalationRow).where(
EscalationRow.call_id == row.call_id,
EscalationRow.status.in_(["requested", "ringing"]),
EscalationRow.status.in_(["requested", "ringing", "connected"]),
)
).scalar_one_or_none()
was_talking = open_escalation is not None and open_escalation.status == "connected"
if open_escalation is not None:
open_escalation.status = "completed" if answered else "failed"
open_escalation.completed_at = now
if answered and not open_escalation.connected_at:
open_escalation.connected_at = now
if was_talking and open_escalation and open_escalation.real_agent_id:
try:
bridge._set_routing_agent_status(open_escalation.real_agent_id, "AFTER_CALL_WORK")
except Exception:
logger.warning("bridge.agent_acw_status_failed call_id=%s agent_id=%s", row.call_id, open_escalation.real_agent_id)
else:
try:
bridge._release_routing_agent(row.call_id)
except Exception:
pass
if open_escalation is not None:
try:
bridge._append_interaction_timeline(
interaction_id=link.interaction_id,
action="escalation.completed" if answered else "escalation.failed_client_disconnected",
metadata={"call_id": row.call_id, "escalation_id": open_escalation.escalation_id, "agent_id": open_escalation.real_agent_id},
)
except Exception:
pass
try:
bridge._notify_voice_ai_telephony_event(
voice_session_id=link.voice_session_id,
@@ -969,6 +986,48 @@ def process_operator_connected(
link.telephony_status = "connected"
link.connected_at = now
link.updated_at = now
open_escalation = session.execute(
select(EscalationRow).where(
EscalationRow.call_id == row.call_id,
EscalationRow.status == "ringing",
)
).scalar_one_or_none()
if open_escalation is not None:
open_escalation.status = "connected"
open_escalation.connected_at = now
if open_escalation.real_agent_id:
try:
bridge._set_routing_agent_status(open_escalation.real_agent_id, "TALKING")
except Exception:
logger.warning("bridge.agent_talking_status_failed call_id=%s agent_id=%s", row.call_id, open_escalation.real_agent_id)
try:
bridge._append_interaction_timeline(
interaction_id=link.interaction_id,
action="escalation.agent_connected",
metadata={"call_id": row.call_id, "agent_id": open_escalation.real_agent_id, "escalation_id": open_escalation.escalation_id},
)
except Exception:
pass
try:
bridge._emit_voice_event(
event_type="AgentConnected",
call_id=row.call_id,
interaction_id=link.interaction_id,
payload={"escalation_id": open_escalation.escalation_id, "agent_id": open_escalation.real_agent_id},
)
except Exception:
pass
try:
bridge._emit_voice_event(
event_type="TransferCompleted",
call_id=row.call_id,
interaction_id=link.interaction_id,
payload={"escalation_id": open_escalation.escalation_id, "agent_id": open_escalation.real_agent_id},
)
except Exception:
pass
_upsert_voice_reporting_fact(
session,
link,
@@ -1173,6 +1232,55 @@ def process_recording_ready(
local_path.unlink(missing_ok=True)
_NO_ANSWER_DIAL_STATUSES = {"NOANSWER", "BUSY", "CANCEL", "CHANUNAVAIL", "CONGESTION"}
_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:
bridge = _bridge_app()
call_id = row.call_id
if not call_id:
bridge._mark_log_forwarded(session, row, interaction_id=None)
return
dial_status = str(payload.get("DialStatus") or "").strip().upper()
hangup_cause = str(payload.get("Cause") or "").strip()
is_no_answer_outcome = (
(row.ami_event_name == "DialEnd" and dial_status in _NO_ANSWER_DIAL_STATUSES)
or (row.ami_event_name == "Hangup" and hangup_cause in _NO_ANSWER_HANGUP_CAUSES)
)
if not is_no_answer_outcome:
bridge._mark_log_forwarded(session, row, interaction_id=None)
return
escalation = session.execute(
select(EscalationRow).where(
EscalationRow.call_id == call_id,
EscalationRow.status == "ringing",
)
).scalar_one_or_none()
if escalation is None:
bridge._mark_log_forwarded(session, row, interaction_id=None)
return
dialed_channel = payload.get("DestChannel") if row.ami_event_name == "DialEnd" else payload.get("Channel")
dialed_extension = bridge._extract_extension_from_channel(dialed_channel)
if dialed_extension and escalation.assigned_agent_id and dialed_extension != escalation.assigned_agent_id:
bridge._mark_log_forwarded(session, row, interaction_id=None)
return
try:
bridge._retry_escalation_no_answer(
session,
call_id=call_id,
dial_outcome=dial_status or f"hangup_cause_{hangup_cause}",
)
except Exception:
logger.exception("bridge.agent_dial_outcome_retry_failed call_id=%s", call_id)
bridge._mark_log_forwarded(session, row, interaction_id=None)
def process_bridge_row(session, row: AsteriskEventLogRow) -> AsteriskEventLogRow:
bridge = _bridge_app()
payload = json.loads(row.payload_json or "{}")
@@ -1209,6 +1317,8 @@ def process_bridge_row(session, row: AsteriskEventLogRow) -> AsteriskEventLogRow
bridge._process_call_ended(session, row, payload)
elif row.ami_event_name == f"{bridge._ami_prefix()}RecordingReady":
bridge._process_recording_ready(session, row, payload)
elif row.ami_event_name in {"DialEnd", "Hangup"}:
bridge._process_agent_dial_outcome(session, row, payload)
else:
row.forward_status = "received"
row.updated_at = utc_now_iso()
@@ -1216,9 +1326,9 @@ def process_bridge_row(session, row: AsteriskEventLogRow) -> AsteriskEventLogRow
return row
def record_ami_payload(payload: dict[str, Any]) -> None:
def record_ami_payload(payload: dict[str, Any], *, event_name: str | None = None) -> None:
bridge = _bridge_app()
event_name = str(payload.get("UserEvent") or "").strip()
event_name = str(event_name or payload.get("UserEvent") or "").strip()
call_id = bridge._extract_call_id(payload)
linked_id = bridge._extract_linked_id(payload, call_id)
if not event_name or not call_id:
@@ -1262,6 +1372,42 @@ def retry_failed_events_once() -> None:
bridge._process_claimed_bridge_event(bridge_event_id)
def acw_duration_seconds() -> int:
raw = str(os.environ.get("ACW_DURATION_SECONDS", "30")).strip()
try:
return max(int(raw), 1)
except ValueError:
return 30
def acw_sweep_interval_seconds() -> float:
return max(min(float(acw_duration_seconds()) / 2, 15.0), 5.0)
def acw_sweep_once() -> None:
from services.routing_service import engine as routing_engine
session = get_session()
try:
swept = routing_engine.sweep_after_call_work(session, older_than_seconds=acw_duration_seconds())
for agent in swept:
logger.warning("bridge.agent_acw_swept agent_id=%s", agent.agent_id)
finally:
session.close()
def acw_sweep_loop(stop_event: threading.Event | None = None) -> None:
bridge = _bridge_app()
active_stop_event = stop_event or bridge._background_stop_event()
while not active_stop_event.is_set():
try:
acw_sweep_once()
except Exception:
logger.exception("bridge.acw_sweep_failed")
if active_stop_event.wait(acw_sweep_interval_seconds()):
break
def failed_retry_loop(stop_event: threading.Event | None = None) -> None:
bridge = _bridge_app()
active_stop_event = stop_event or bridge._background_stop_event()
+259 -21
View File
@@ -63,6 +63,57 @@ def append_interaction_timeline(
)
def _append_escalation_timeline(
escalation: EscalationRow,
link: AsteriskCallLinkRow,
*,
action: str,
extra: dict[str, Any] | None = None,
) -> None:
try:
append_interaction_timeline(
interaction_id=link.interaction_id,
action=action,
metadata={
"call_id": escalation.call_id,
"escalation_id": escalation.escalation_id,
"from_level": escalation.from_level,
"to_level": escalation.to_level,
"attempt_count": escalation.attempt_count,
**(extra or {}),
},
)
except Exception:
pass
def _emit_escalation_event(
*,
event_type: str,
escalation: EscalationRow,
link: AsteriskCallLinkRow,
extra: dict[str, Any] | None = None,
) -> None:
bridge = _bridge_app()
try:
bridge._emit_voice_event(
event_type=event_type,
call_id=escalation.call_id,
interaction_id=link.interaction_id,
payload={
"escalation_id": escalation.escalation_id,
"tenant_id": escalation.tenant_id,
"from_level": escalation.from_level,
"to_level": escalation.to_level,
"reason_code": escalation.reason_code,
"attempt_count": escalation.attempt_count,
**(extra or {}),
},
)
except Exception:
LOGGER.warning("bridge.escalation_event_failed event_type=%s call_id=%s", event_type, escalation.call_id)
def start_voice_ai_session(
*,
call_id: str,
@@ -188,7 +239,8 @@ def _reserve_routing_agent(
level: str,
tenant_id: str | None,
required_skills: list[str] | None = None,
) -> str | None:
exclude_agent_ids: list[str] | None = None,
) -> dict[str, Any] | None:
bridge = _bridge_app()
try:
response = bridge._post_json(
@@ -198,7 +250,7 @@ def _reserve_routing_agent(
"level": level,
"tenant_id": tenant_id,
"required_skills": required_skills or [],
"exclude_agent_ids": [],
"exclude_agent_ids": exclude_agent_ids or [],
},
timeout_seconds=bridge._callcontrol_side_effect_timeout_seconds(),
max_attempts=1,
@@ -212,8 +264,43 @@ def _reserve_routing_agent(
tenant_id,
)
return None
agent_id = str((response or {}).get("agent_id") or "").strip()
extension = str((response or {}).get("extension") or "").strip()
return extension or None
if not agent_id or not extension:
return None
return {
"agent_id": agent_id,
"extension": extension,
"endpoint": (response or {}).get("endpoint"),
"display_name": (response or {}).get("display_name"),
}
def set_routing_agent_status(agent_id: str, status: str) -> None:
bridge = _bridge_app()
try:
bridge._patch_json(
f"{bridge._routing_service_url()}/internal/routing/agents/{agent_id}/status",
{"status": status},
timeout_seconds=bridge._callcontrol_side_effect_timeout_seconds(),
max_attempts=1,
retry_backoff_seconds=0.0,
)
except Exception:
LOGGER.warning("bridge.routing_set_status_failed agent_id=%s status=%s", agent_id, status)
def _redirect_channel_to_agent(*, channel: str, extension: str) -> None:
bridge = _bridge_app()
bridge._ami_action(
"Redirect",
{
"Channel": channel,
"Context": bridge._transfer_context(),
"Exten": extension,
"Priority": 1,
},
)
def _resolve_handoff_extension(
@@ -224,7 +311,7 @@ def _resolve_handoff_extension(
target_level: str | None = None,
tenant_id: str | None = None,
required_skills: list[str] | None = None,
) -> tuple[str, str, str | None, str | None]:
) -> tuple[str, str, str | None, str | None, str | None]:
bridge = _bridge_app()
queue_code = _queue_code_for_queue_id(target_queue_id) or str(fallback_queue_code or "").strip()
if not queue_code:
@@ -233,14 +320,14 @@ def _resolve_handoff_extension(
level = target_level or bridge._routing_level_for_queue_code(queue_code)
if level and call_id:
resolved_tenant_id = tenant_id if tenant_id is not None else bridge._routing_tenant_for_queue_code(queue_code)
reserved_extension = _reserve_routing_agent(
reserved_agent = _reserve_routing_agent(
call_id=call_id,
level=level,
tenant_id=resolved_tenant_id,
required_skills=required_skills,
)
if reserved_extension:
return queue_code, reserved_extension, level, resolved_tenant_id
if reserved_agent:
return queue_code, reserved_agent["extension"], level, resolved_tenant_id, reserved_agent["agent_id"]
raise HTTPException(status_code=409, detail=f"No available {level} agent right now")
extension = bridge._transfer_target_map().get(queue_code)
@@ -249,7 +336,7 @@ def _resolve_handoff_extension(
status_code=400,
detail=f"Unknown transfer queue_code: {queue_code}",
)
return queue_code, extension, None, None
return queue_code, extension, None, None, None
def _resolve_handoff_channel(session, link: AsteriskCallLinkRow) -> str:
@@ -459,11 +546,33 @@ def request_handoff(call_id: str, body: VoiceAIHandoffRequestIn, actor: dict) ->
if voice_session is None:
raise HTTPException(status_code=404, detail="Voice AI session not found")
queue_code, target_extension, resolved_level, resolved_tenant_id = _resolve_handoff_extension(
queue_code, target_extension, resolved_level, resolved_tenant_id, reserved_agent_id = _resolve_handoff_extension(
body.target_queue_id or voice_session.handoff_target_queue_id or link.queue_id,
fallback_queue_code=_queue_code_for_queue_id(link.queue_id),
call_id=call_id,
)
escalation: EscalationRow | None = None
if reserved_agent_id and target_extension != "7100":
now_reserved = utc_now_iso()
escalation = EscalationRow(
escalation_id=new_id("esc"),
call_id=call_id,
tenant_id=resolved_tenant_id,
from_level=str(link.current_level or "L1"),
to_level=resolved_level,
reason_code="AI_HANDOFF",
required_skills_json="[]",
priority=3,
status="ringing",
real_agent_id=reserved_agent_id,
assigned_agent_id=target_extension,
attempted_agent_ids_json=json.dumps([reserved_agent_id], ensure_ascii=False),
requested_at=now_reserved,
)
session.add(escalation)
session.flush()
_append_escalation_timeline(escalation, link, action="escalation.agent_reserved", extra={"agent_id": reserved_agent_id})
_emit_escalation_event(event_type="AgentReserved", escalation=escalation, link=link, extra={"agent_id": reserved_agent_id})
handoff_metadata = body.metadata or {}
actor_user = str(actor.get("user") or actor.get("sub") or "ai-voice-runtime").strip()
actor_role = str(actor.get("role") or "admin").strip() or "admin"
@@ -510,6 +619,7 @@ def request_handoff(call_id: str, body: VoiceAIHandoffRequestIn, actor: dict) ->
target_extension,
body.target_queue_id or voice_session.handoff_target_queue_id or link.queue_id,
)
try:
ami_result = bridge._ami_action(
"Redirect",
{
@@ -519,6 +629,23 @@ def request_handoff(call_id: str, body: VoiceAIHandoffRequestIn, actor: dict) ->
"Priority": 1,
},
)
except Exception as exc:
if escalation is not None:
try:
routing_release_by_agent_id(reserved_agent_id)
except Exception:
LOGGER.warning("bridge.handoff_release_agent_failed call_id=%s agent_id=%s", call_id, reserved_agent_id)
escalation.status = "failed"
escalation.completed_at = utc_now_iso()
session.commit()
_append_escalation_timeline(escalation, link, action="escalation.transfer_failed", extra={"agent_id": reserved_agent_id, "error": str(exc)})
_emit_escalation_event(event_type="TransferFailed", escalation=escalation, link=link, extra={"agent_id": reserved_agent_id, "error": str(exc)})
raise HTTPException(status_code=502, detail=f"Failed to redirect call to agent: {exc}") from exc
if escalation is not None:
set_routing_agent_status(reserved_agent_id, "RINGING")
_append_escalation_timeline(escalation, link, action="escalation.agent_ringing", extra={"agent_id": reserved_agent_id})
_emit_escalation_event(event_type="AgentRinging", escalation=escalation, link=link, extra={"agent_id": reserved_agent_id})
now = utc_now_iso()
link.voice_session_id = (
@@ -666,6 +793,7 @@ def _escalation_to_out(row: EscalationRow) -> EscalationOut:
summary=row.summary,
status=row.status,
assigned_agent_id=row.assigned_agent_id,
attempt_count=row.attempt_count or 0,
requested_at=row.requested_at,
connected_at=row.connected_at,
completed_at=row.completed_at,
@@ -709,28 +837,32 @@ def create_escalation(call_id: str, body: EscalationRequestIn, actor: dict) -> E
session.flush()
channel = _resolve_handoff_channel(session, link)
reserved_extension = _reserve_routing_agent(
agent = _reserve_routing_agent(
call_id=call_id,
level=body.target_level,
tenant_id=link.tenant_id,
required_skills=body.required_skills,
)
if not reserved_extension:
if not agent:
escalation.status = "failed"
escalation.completed_at = now
session.commit()
_emit_escalation_event(
event_type="TransferFailed",
escalation=escalation,
link=link,
extra={"reason": "no_available_agent"},
)
raise HTTPException(status_code=409, detail=f"No available {body.target_level} agent right now")
escalation.real_agent_id = agent["agent_id"]
escalation.attempted_agent_ids_json = json.dumps([agent["agent_id"]], ensure_ascii=False)
session.flush()
_append_escalation_timeline(escalation, link, action="escalation.agent_reserved", extra={"agent_id": agent["agent_id"]})
_emit_escalation_event(event_type="AgentReserved", escalation=escalation, link=link, extra={"agent_id": agent["agent_id"]})
try:
bridge._ami_action(
"Redirect",
{
"Channel": channel,
"Context": bridge._transfer_context(),
"Exten": reserved_extension,
"Priority": 1,
},
)
_redirect_channel_to_agent(channel=channel, extension=agent["extension"])
except Exception as exc:
try:
bridge._release_routing_agent(call_id)
@@ -739,10 +871,13 @@ def create_escalation(call_id: str, body: EscalationRequestIn, actor: dict) -> E
escalation.status = "failed"
escalation.completed_at = utc_now_iso()
session.commit()
_append_escalation_timeline(escalation, link, action="escalation.transfer_failed", extra={"agent_id": agent["agent_id"], "error": str(exc)})
_emit_escalation_event(event_type="TransferFailed", escalation=escalation, link=link, extra={"agent_id": agent["agent_id"], "error": str(exc)})
raise HTTPException(status_code=502, detail=f"Failed to redirect call to agent: {exc}") from exc
set_routing_agent_status(agent["agent_id"], "RINGING")
escalation.status = "ringing"
escalation.assigned_agent_id = reserved_extension
escalation.assigned_agent_id = agent["extension"]
link.current_level = body.target_level
link.required_skills_json = json.dumps(body.required_skills, ensure_ascii=False)
link.priority = body.priority
@@ -751,11 +886,114 @@ def create_escalation(call_id: str, body: EscalationRequestIn, actor: dict) -> E
link.operator_extension = None
link.updated_at = now
session.commit()
_append_escalation_timeline(escalation, link, action="escalation.agent_ringing", extra={"agent_id": agent["agent_id"]})
_emit_escalation_event(event_type="AgentRinging", escalation=escalation, link=link, extra={"agent_id": agent["agent_id"]})
return _escalation_to_out(escalation)
finally:
session.close()
def retry_escalation_no_answer(session, *, call_id: str, dial_outcome: str) -> None:
"""AC-08 / ТЗ §14: агент не ответил — освободить его и попробовать следующего.
Вызывается из bridge_processing при нативном AMI DialEnd/Hangup с исходом
NOANSWER/BUSY/CANCEL/CHANUNAVAIL/CONGESTION на канале агента, зарезервированного
под активную (status='ringing') эскалацию этого call_id.
"""
bridge = _bridge_app()
escalation = session.execute(
select(EscalationRow).where(
EscalationRow.call_id == call_id,
EscalationRow.status == "ringing",
)
).scalar_one_or_none()
if escalation is None:
return
link = bridge._find_call_link(session, call_id)
if link is None:
return
now = utc_now_iso()
no_answer_agent_id = escalation.real_agent_id
if no_answer_agent_id:
try:
routing_release_by_agent_id(no_answer_agent_id)
except Exception:
LOGGER.warning("bridge.escalation_retry_release_failed call_id=%s agent_id=%s", call_id, no_answer_agent_id)
escalation.attempt_count = (escalation.attempt_count or 0) + 1
session.flush()
_append_escalation_timeline(
escalation,
link,
action="escalation.agent_no_answer",
extra={"agent_id": no_answer_agent_id, "dial_outcome": dial_outcome},
)
_emit_escalation_event(
event_type="AgentNoAnswer",
escalation=escalation,
link=link,
extra={"agent_id": no_answer_agent_id, "dial_outcome": dial_outcome},
)
attempted_ids = json.loads(escalation.attempted_agent_ids_json or "[]")
channel = _resolve_handoff_channel(session, link)
required_skills = json.loads(escalation.required_skills_json or "[]")
next_agent = _reserve_routing_agent(
call_id=call_id,
level=escalation.to_level,
tenant_id=escalation.tenant_id,
required_skills=required_skills,
exclude_agent_ids=attempted_ids,
)
if not next_agent:
escalation.status = "failed"
escalation.completed_at = now
session.commit()
_append_escalation_timeline(escalation, link, action="escalation.failed_no_agents")
_emit_escalation_event(event_type="TransferFailed", escalation=escalation, link=link, extra={"reason": "no_more_agents"})
LOGGER.warning("bridge.escalation_retry_exhausted call_id=%s attempts=%s", call_id, escalation.attempt_count)
return
escalation.real_agent_id = next_agent["agent_id"]
escalation.attempted_agent_ids_json = json.dumps(attempted_ids + [next_agent["agent_id"]], ensure_ascii=False)
session.flush()
_append_escalation_timeline(escalation, link, action="escalation.agent_reserved", extra={"agent_id": next_agent["agent_id"]})
_emit_escalation_event(event_type="AgentReserved", escalation=escalation, link=link, extra={"agent_id": next_agent["agent_id"]})
try:
_redirect_channel_to_agent(channel=channel, extension=next_agent["extension"])
except Exception as exc:
try:
routing_release_by_agent_id(next_agent["agent_id"])
except Exception:
pass
escalation.status = "failed"
escalation.completed_at = utc_now_iso()
session.commit()
_append_escalation_timeline(escalation, link, action="escalation.transfer_failed", extra={"agent_id": next_agent["agent_id"], "error": str(exc)})
_emit_escalation_event(event_type="TransferFailed", escalation=escalation, link=link, extra={"agent_id": next_agent["agent_id"], "error": str(exc)})
return
set_routing_agent_status(next_agent["agent_id"], "RINGING")
escalation.assigned_agent_id = next_agent["extension"]
session.commit()
_append_escalation_timeline(escalation, link, action="escalation.agent_ringing", extra={"agent_id": next_agent["agent_id"]})
_emit_escalation_event(event_type="AgentRinging", escalation=escalation, link=link, extra={"agent_id": next_agent["agent_id"]})
def routing_release_by_agent_id(agent_id: str, *, next_status: str = "AVAILABLE") -> None:
bridge = _bridge_app()
bridge._post_json(
f"{bridge._routing_service_url()}/internal/routing/release-agent",
{"agent_id": agent_id, "next_status": next_status},
timeout_seconds=bridge._callcontrol_side_effect_timeout_seconds(),
max_attempts=1,
retry_backoff_seconds=0.0,
)
def update_call_ai_state(call_id: str, body: VoiceAICallStateUpdateIn, actor: dict) -> VoiceLiveCallOut:
bridge = _bridge_app()
assert_trusted_voice_runtime_actor(actor)
+8
View File
@@ -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:
+26 -3
View File
@@ -18,6 +18,7 @@ from services.shared.models import (
QueueOut,
RoutingAgentReserveIn,
RoutingAgentReserveOut,
RoutingAgentStatusIn,
)
from services.shared.security import require_roles
from services.shared.sql_init import init_sql_schema
@@ -343,6 +344,7 @@ def _escalation_to_out(row: EscalationRow) -> EscalationOut:
summary=row.summary,
status=row.status,
assigned_agent_id=row.assigned_agent_id,
attempt_count=row.attempt_count or 0,
requested_at=row.requested_at,
connected_at=row.connected_at,
completed_at=row.completed_at,
@@ -399,11 +401,32 @@ def release_agent_endpoint(
_: dict = Depends(require_roles(Role.ADMIN)),
) -> dict:
call_id = str(payload.get("call_id") or "").strip()
if not call_id:
raise HTTPException(status_code=400, detail="call_id is required")
agent_id = str(payload.get("agent_id") or "").strip()
next_status = str(payload.get("next_status") or "AVAILABLE").strip() or "AVAILABLE"
if not call_id and not agent_id:
raise HTTPException(status_code=400, detail="call_id or agent_id is required")
session = get_session()
try:
if agent_id:
agent = routing_engine.release_agent_by_id(session, agent_id=agent_id, next_status=next_status)
else:
agent = routing_engine.release_agent_by_call_id(session, call_id=call_id)
return {"call_id": call_id, "released_agent_id": agent.agent_id if agent else None}
return {"call_id": call_id or None, "released_agent_id": agent.agent_id if agent else None}
finally:
session.close()
@app.patch("/internal/routing/agents/{agent_id}/status")
def set_agent_status_endpoint(
agent_id: str,
payload: RoutingAgentStatusIn,
_: dict = Depends(require_roles(Role.ADMIN)),
) -> dict:
session = get_session()
try:
agent = routing_engine.set_agent_status(session, agent_id=agent_id, status=payload.status)
if agent is None:
raise HTTPException(status_code=404, detail="Agent not found")
return {"agent_id": agent.agent_id, "status": agent.status}
finally:
session.close()
+36
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import json
from datetime import datetime
from sqlalchemy import select, text
@@ -117,6 +118,41 @@ def release_agent_by_call_id(session, *, call_id: str) -> AgentRow | None:
return agent
def set_agent_status(session, *, agent_id: str, status: str) -> AgentRow | None:
agent = session.execute(
select(AgentRow).where(AgentRow.agent_id == agent_id)
).scalar_one_or_none()
if agent is None:
return None
agent.status = status
agent.updated_at = utc_now_iso()
session.commit()
return agent
def sweep_after_call_work(session, *, older_than_seconds: int) -> list[AgentRow]:
cutoff = utc_now_iso()
rows = session.execute(
select(AgentRow).where(AgentRow.status == "AFTER_CALL_WORK")
).scalars().all()
swept: list[AgentRow] = []
for agent in rows:
try:
age_seconds = (
datetime.fromisoformat(cutoff) - datetime.fromisoformat(str(agent.updated_at))
).total_seconds()
except (TypeError, ValueError):
continue
if age_seconds >= older_than_seconds:
agent.status = "AVAILABLE"
agent.current_call_id = None
agent.updated_at = cutoff
swept.append(agent)
if swept:
session.commit()
return swept
def release_agent_by_id(session, *, agent_id: str, next_status: str = "AVAILABLE") -> AgentRow | None:
agent = session.execute(
select(AgentRow).where(AgentRow.agent_id == agent_id)
+39
View File
@@ -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
+1
View File
@@ -25,6 +25,7 @@ class KBSearchRow(Protocol):
title: str
body: str
tags_json: str
intent_code: str | None
T = TypeVar("T", bound=KBSearchRow)
+7
View File
@@ -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
@@ -1311,11 +1313,16 @@ class EscalationOut(BaseModel):
summary: str | None = None
status: str
assigned_agent_id: str | None = None
attempt_count: int = 0
requested_at: str
connected_at: str | None = None
completed_at: str | None = None
class RoutingAgentStatusIn(BaseModel):
status: str = Field(min_length=1)
class RoutingAgentReserveIn(BaseModel):
call_id: str = Field(min_length=1)
level: AgentLevel
+8
View File
@@ -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")
+4
View File
@@ -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)
@@ -846,6 +847,9 @@ class EscalationRow(Base):
summary: Mapped[str | None] = mapped_column(Text, nullable=True)
status: Mapped[str] = mapped_column(String(32), index=True, default="requested")
assigned_agent_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
real_agent_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
attempted_agent_ids_json: Mapped[str] = mapped_column(Text, default="[]")
attempt_count: Mapped[int] = mapped_column(Integer, default=0)
requested_at: Mapped[str] = mapped_column(String(64), index=True)
connected_at: Mapped[str | None] = mapped_column(String(64), nullable=True)
completed_at: Mapped[str | None] = mapped_column(String(64), nullable=True)
+84
View File
@@ -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")
+7 -3
View File
@@ -859,11 +859,14 @@ def test_push_voice_ai_telephony_event_call_ended_returns_detached_safe_payload(
def test_media_registration_emotive_ack_rotation_is_not_gated_on_v2_queue_eligibility(monkeypatch):
# AI_VOICE_V2_QUEUE_CODES defaults to "voice_lab_ai" only, so a queue outside
# that allowlist runs the plain v1 ack path. Emotive-ack rotation must still
# apply there — otherwise every filler collapses to the single fixed
# "Секунду." fallback string instead of rotating through phrase variants.
# that allowlist runs the plain v1 ack path. Emotive-ack rotation and prebaked-ack
# caching must still apply there — otherwise every filler collapses to the single
# fixed "Секунду." fallback string instead of rotating, and gets synthesized live
# via ElevenLabs each time instead of served from the pre-cached ack bank, adding
# real TTS round-trip time to the one phrase whose job is to hide that latency.
monkeypatch.delenv("AI_VOICE_V2_QUEUE_CODES", raising=False)
monkeypatch.delenv("AI_VOICE_V2_EMOTIVE_ACK_ENABLED", raising=False)
monkeypatch.delenv("AI_VOICE_V2_PREBAKED_ACK_ENABLED", raising=False)
row = VoiceAISessionRow(
session_id="avs_runtime_emotive_ack_v1",
call_id="call_runtime_emotive_ack_v1",
@@ -879,3 +882,4 @@ def test_media_registration_emotive_ack_rotation_is_not_gated_on_v2_queue_eligib
assert registration.voice_v2_enabled is False
assert registration.voice_v2_emotive_ack is True
assert registration.voice_v2_prebaked_ack is True
+2 -1
View File
@@ -2195,6 +2195,7 @@ def test_disabled_bridge_startup_does_not_require_singleton_guard(monkeypatch):
monkeypatch.delenv("ASTERISK_BRIDGE_ENABLED", raising=False)
monkeypatch.setattr(bridge_module, "_ami_loop", _wait_until_stopped)
monkeypatch.setattr(bridge_module, "_failed_retry_loop", _wait_until_stopped)
monkeypatch.setattr(bridge_module, "_acw_sweep_loop", _wait_until_stopped)
monkeypatch.setattr(
bridge_module,
"_try_acquire_bridge_singleton_guard",
@@ -2202,7 +2203,7 @@ def test_disabled_bridge_startup_does_not_require_singleton_guard(monkeypatch):
)
bridge_module._startup()
assert len(bridge_module._background_threads_alive()) == 2
assert len(bridge_module._background_threads_alive()) == 3
def test_shutdown_releases_singleton_guard(monkeypatch):
+190
View File
@@ -0,0 +1,190 @@
import json
import pytest
import services.asterisk_bridge_service.voice_ai as voice_ai
from services.routing_service import engine as routing_engine
from services.shared.core import new_id, utc_now_iso
from services.shared.db import get_session
from services.shared.sql_init import init_sql_schema
from services.shared.sql_models import AgentRow, AsteriskCallLinkRow, EscalationRow
def _make_agent(session, *, extension: str, status: str = "AVAILABLE"):
now = utc_now_iso()
row = AgentRow(
agent_id=new_id("agt"),
tenant_ids_json="[]",
extension=extension,
endpoint=None,
display_name=f"Agent {extension}",
level="L2",
skills_json="[]",
status=status,
max_concurrent_calls=1,
enabled=True,
created_at=now,
updated_at=now,
)
session.add(row)
session.commit()
session.refresh(row)
return row
def _make_call_link(session, *, call_id: str):
now = utc_now_iso()
link = AsteriskCallLinkRow(
call_id=call_id,
linked_id=call_id,
queue_code="voice_lab_ai",
queue_id="que_test",
interaction_id="int_test",
status="active",
telephony_status="ringing",
channel_name=f"PJSIP/tele2-kazgaz-{call_id}",
current_level="L2",
started_at=now,
updated_at=now,
)
session.add(link)
session.commit()
session.refresh(link)
return link
def _make_ringing_escalation(session, *, call_id: str, agent: AgentRow):
now = utc_now_iso()
escalation = EscalationRow(
escalation_id=new_id("esc"),
call_id=call_id,
tenant_id=None,
from_level="L1",
to_level="L2",
reason_code="AI_UNABLE_TO_RESOLVE",
required_skills_json="[]",
priority=3,
status="ringing",
assigned_agent_id=agent.extension,
real_agent_id=agent.agent_id,
attempted_agent_ids_json=json.dumps([agent.agent_id]),
requested_at=now,
)
session.add(escalation)
session.commit()
session.refresh(escalation)
return escalation
def _patch_routing_over_http(monkeypatch, session):
"""Make voice_ai's HTTP-facing routing helpers operate on the same test session directly."""
def fake_reserve(*, call_id, level, tenant_id, required_skills=None, exclude_agent_ids=None):
agent = routing_engine.reserve_agent(
session,
call_id=call_id,
level=level,
tenant_id=tenant_id,
required_skills=required_skills,
exclude_agent_ids=exclude_agent_ids,
)
if agent is None:
return None
return {
"agent_id": agent.agent_id,
"extension": agent.extension,
"endpoint": agent.endpoint,
"display_name": agent.display_name,
}
def fake_release_by_agent_id(agent_id, *, next_status="AVAILABLE"):
routing_engine.release_agent_by_id(session, agent_id=agent_id, next_status=next_status)
def fake_set_status(agent_id, status):
routing_engine.set_agent_status(session, agent_id=agent_id, status=status)
monkeypatch.setattr(voice_ai, "_reserve_routing_agent", fake_reserve)
monkeypatch.setattr(voice_ai, "routing_release_by_agent_id", fake_release_by_agent_id)
monkeypatch.setattr(voice_ai, "set_routing_agent_status", fake_set_status)
monkeypatch.setattr(voice_ai, "_append_escalation_timeline", lambda *a, **k: None)
monkeypatch.setattr(voice_ai, "_emit_escalation_event", lambda *a, **k: None)
monkeypatch.setattr(voice_ai, "_resolve_handoff_channel", lambda session, link: link.channel_name)
def test_retry_escalation_no_answer_moves_to_next_available_agent(monkeypatch):
init_sql_schema()
redirected_to: list[str] = []
session = get_session()
try:
_patch_routing_over_http(monkeypatch, session)
monkeypatch.setattr(
voice_ai,
"_redirect_channel_to_agent",
lambda *, channel, extension: redirected_to.append(extension),
)
agent_a = _make_agent(session, extension="2001")
agent_b = _make_agent(session, extension="2002")
routing_engine.reserve_agent(session, call_id="call-retry-1", level="L2", tenant_id=None, required_skills=[])
link = _make_call_link(session, call_id="call-retry-1")
escalation = _make_ringing_escalation(session, call_id="call-retry-1", agent=agent_a)
voice_ai.retry_escalation_no_answer(session, call_id="call-retry-1", dial_outcome="NOANSWER")
session.refresh(escalation)
session.refresh(agent_a)
session.refresh(agent_b)
assert escalation.attempt_count == 1
assert escalation.status == "ringing"
assert escalation.real_agent_id == agent_b.agent_id
assert escalation.assigned_agent_id == agent_b.extension
assert json.loads(escalation.attempted_agent_ids_json) == [agent_a.agent_id, agent_b.agent_id]
assert agent_a.status == "AVAILABLE"
assert agent_a.current_call_id is None
assert agent_b.status == "RINGING"
assert agent_b.current_call_id == "call-retry-1"
assert redirected_to == [agent_b.extension]
finally:
session.close()
def test_retry_escalation_no_answer_exhausts_pool_marks_failed(monkeypatch):
init_sql_schema()
session = get_session()
try:
_patch_routing_over_http(monkeypatch, session)
monkeypatch.setattr(voice_ai, "_redirect_channel_to_agent", lambda **kwargs: None)
agent_a = _make_agent(session, extension="3001")
routing_engine.reserve_agent(session, call_id="call-retry-2", level="L2", tenant_id=None, required_skills=[])
link = _make_call_link(session, call_id="call-retry-2")
escalation = _make_ringing_escalation(session, call_id="call-retry-2", agent=agent_a)
voice_ai.retry_escalation_no_answer(session, call_id="call-retry-2", dial_outcome="NOANSWER")
session.refresh(escalation)
session.refresh(agent_a)
assert escalation.attempt_count == 1
assert escalation.status == "failed"
assert agent_a.status == "AVAILABLE"
finally:
session.close()
def test_retry_escalation_no_answer_ignores_calls_without_ringing_escalation(monkeypatch):
init_sql_schema()
called = []
session = get_session()
try:
_patch_routing_over_http(monkeypatch, session)
monkeypatch.setattr(voice_ai, "_redirect_channel_to_agent", lambda **kwargs: called.append(kwargs))
_make_call_link(session, call_id="call-no-escalation")
voice_ai.retry_escalation_no_answer(session, call_id="call-no-escalation", dial_outcome="NOANSWER")
assert called == []
finally:
session.close()
+34
View File
@@ -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",
}
+48
View File
@@ -1,3 +1,4 @@
from datetime import datetime, timedelta, timezone
import json
from sqlalchemy import select
@@ -134,3 +135,50 @@ def test_reserve_agent_excludes_disabled_and_excluded_ids():
assert reserved.agent_id != excluded.agent_id
finally:
session.close()
def test_set_agent_status_transitions_ringing_to_talking():
init_sql_schema()
session = get_session()
try:
agent = _make_agent(session, level="L2")
routing_engine.reserve_agent(session, call_id="call-status", level="L2", tenant_id=None, required_skills=[])
ringing = routing_engine.set_agent_status(session, agent_id=agent.agent_id, status="RINGING")
assert ringing is not None
assert ringing.status == "RINGING"
talking = routing_engine.set_agent_status(session, agent_id=agent.agent_id, status="TALKING")
assert talking.status == "TALKING"
missing = routing_engine.set_agent_status(session, agent_id="unknown-agent", status="AVAILABLE")
assert missing is None
finally:
session.close()
def test_sweep_after_call_work_releases_only_expired_agents():
init_sql_schema()
session = get_session()
try:
stale = _make_agent(session, level="L2", status="AFTER_CALL_WORK")
fresh = _make_agent(session, level="L2", status="AFTER_CALL_WORK")
stale_ts = (datetime.now(timezone.utc) - timedelta(seconds=120)).replace(microsecond=0).isoformat()
stale.updated_at = stale_ts
stale.current_call_id = "call-stale"
session.commit()
swept = routing_engine.sweep_after_call_work(session, older_than_seconds=30)
swept_ids = {a.agent_id for a in swept}
assert stale.agent_id in swept_ids
assert fresh.agent_id not in swept_ids
session.refresh(stale)
session.refresh(fresh)
assert stale.status == "AVAILABLE"
assert stale.current_call_id is None
assert fresh.status == "AFTER_CALL_WORK"
finally:
session.close()
+40
View File
@@ -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"}