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
35 changed files with 1127 additions and 74 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" }
}
}
+2 -2
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
@@ -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(
@@ -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."
)
+16 -3
View File
@@ -1359,7 +1359,13 @@ def _voice_v2_enabled(metadata: dict[str, Any] | None = None) -> bool:
return _voice_policy_mode() in {"v2_fast_conversational", "v2_streaming_duplex"}
def _voice_early_intent_bucket(text: str) -> str:
def _voice_ack_topic_bucket(text: str) -> str:
"""Coarse keyword heuristic used only to pick an ack phrase / early clarifying
question (see _voice_ack_kind_for_intent and _voice_early_plan) while the real,
KB-grounded decision is still in flight. This is NOT the canonical FAQ intent
(see services.shared.intents) and must never be echoed back as the final
decision's `intent` value.
"""
normalized = " ".join(str(text or "").strip().lower().split())
if not normalized:
return "unknown"
@@ -1509,7 +1515,7 @@ def _voice_v2_metadata(
if not _voice_v2_enabled(request_metadata):
return {}
payload = request_metadata if isinstance(request_metadata, dict) else {}
early_intent = _voice_early_intent_bucket(transcript_text)
early_intent = _voice_ack_topic_bucket(transcript_text)
metadata: dict[str, Any] = {
"voice_v2_enabled": True,
"early_intent": early_intent,
@@ -1668,6 +1674,7 @@ def _voice_llm_prompt_messages(
"article_id": article.article_id,
"title": article.title,
"snippet": app._article_snippet(article, limit=240),
"intent_code": getattr(article, "intent_code", None),
}
for article in kb_results[:3]
]
@@ -1767,7 +1774,13 @@ def _voice_llm_decision(
)
except Exception:
return None
decision = app._sanitize_decision(raw, fallback_language=language)
decision = app._sanitize_decision(
raw,
fallback_language=language,
known_topic_codes=[
code for article in kb_results if (code := getattr(article, "intent_code", None))
],
)
if not str(decision.get("reply_text") or "").strip():
decision["reply_text"] = _voice_generic_prompt(language)
if decision.get("needs_handoff") and not decision.get("handoff_reason"):
+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(
+11 -6
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":
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
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)
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
try:
bridge._release_routing_agent(row.call_id)
except Exception:
pass
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()
+268 -30
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,15 +619,33 @@ 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,
)
ami_result = bridge._ami_action(
"Redirect",
{
"Channel": channel,
"Context": bridge._transfer_context(),
"Exten": target_extension,
"Priority": 1,
},
)
try:
ami_result = bridge._ami_action(
"Redirect",
{
"Channel": channel,
"Context": bridge._transfer_context(),
"Exten": target_extension,
"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:
+27 -4
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:
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}
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 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"}