From 2243f305b87294d072d24491f3fa378b07526c99 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Fri, 28 Aug 2026 16:22:32 +0500 Subject: [PATCH 1/2] feat: L1->L2 agent pool and routing engine for voice escalation Replaces the hardcoded single-extension redirect for AI->human call escalation with a real Agent Pool + Routing Engine: - agents/escalations/routing_rules tables (migration 0031), asterisk_call_links gains tenant_id/current_level/required_skills_json/priority. - services/routing_service/engine.py: level/tenant/skill filtered agent selection with atomic (CAS) reservation, no double-booking. - routing-service: /agents CRUD + /internal/routing/reserve-agent and /internal/routing/release-agent. - asterisk-bridge-service: voice_ai.request_handoff now uses the Routing Engine automatically for any queue_code configured in ASTERISK_QUEUE_LEVEL_MAP_JSON (all other queue_codes keep the existing static ASTERISK_TRANSFER_TARGET_MAP_JSON behavior unchanged); new POST /asterisk/live-calls/{call_id}/escalations entrypoint; agent is released back to AVAILABLE and the escalation closed when the call ends. Targets the Tele2 Kazgaz DID +77476456048 (from-tele2-kazgaz context) as the first queue wired to real L2 routing instead of AI-only. Known gap (documented in docs/architecture/l1-l2-routing-engine.md): automatic no-answer retry-to-next-agent needs a small, separately reviewed dialplan change and is left for a follow-up MR rather than guessed at blind. Tests: services/routing_service/engine.py covered by tests/test_routing_engine.py (selection filtering, atomic reservation, release); existing test_asterisk_bridge_service.py and test_routing_service_pg_counter.py suites still pass unmodified. --- docs/architecture/l1-l2-routing-engine.md | 75 ++++++++ .../sql/0031_agent_pool_routing_postgres.sql | 80 ++++++++ .../sql/0031_agent_pool_routing_sqlite.sql | 80 ++++++++ services/asterisk_bridge_service/app.py | 16 ++ .../bridge_processing.py | 16 ++ services/asterisk_bridge_service/config.py | 34 ++++ services/asterisk_bridge_service/routes.py | 7 + services/asterisk_bridge_service/voice_ai.py | 181 +++++++++++++++++- services/routing_service/app.py | 143 +++++++++++++- services/routing_service/engine.py | 131 +++++++++++++ services/shared/models.py | 78 ++++++++ services/shared/sql_init.py | 19 ++ services/shared/sql_models.py | 67 +++++++ tests/test_routing_engine.py | 136 +++++++++++++ 14 files changed, 1057 insertions(+), 6 deletions(-) create mode 100644 docs/architecture/l1-l2-routing-engine.md create mode 100644 migrations/sql/0031_agent_pool_routing_postgres.sql create mode 100644 migrations/sql/0031_agent_pool_routing_sqlite.sql create mode 100644 services/routing_service/engine.py create mode 100644 tests/test_routing_engine.py diff --git a/docs/architecture/l1-l2-routing-engine.md b/docs/architecture/l1-l2-routing-engine.md new file mode 100644 index 0000000..8d24b64 --- /dev/null +++ b/docs/architecture/l1-l2-routing-engine.md @@ -0,0 +1,75 @@ +# L1 -> L2 Routing Engine (Phase 1) + +Implements a real Agent Pool + Routing Engine for voice escalation from +Voice AI (L1) to live L2 operators, replacing the previous static +single-extension redirect target. + +## What changed + +- `agents` table (`services/routing_service/engine.py`): agents are rows with + `level`, `tenant_ids`, `skills`, `status`, `max_concurrent_calls`, `enabled` - + not a hardcoded list. Selection filters by level/tenant/skills and picks the + longest-idle, least-loaded match. Reservation is an atomic + `UPDATE ... WHERE status='AVAILABLE'` (rowcount-checked), so two concurrent + calls can never reserve the same agent. +- `escalations` table: one row per escalation attempt (`from_level`, + `to_level`, `reason_code`, `required_skills`, `priority`, `status`). +- `routing_rules` table: seeded data (not code) describing which + level-to-level transitions are allowed; `L3` and multi-tenant rules can be + added later without code changes. +- `asterisk_call_links` gained `tenant_id`, `current_level`, + `required_skills_json`, `priority`. +- `POST /asterisk/live-calls/{call_id}/escalations` (asterisk-bridge-service): + explicit escalation entrypoint. `voice_ai.request_handoff` (used by the AI + runtime) now also uses the Agent Pool automatically for any `queue_code` + configured in `ASTERISK_QUEUE_LEVEL_MAP_JSON`; every other queue_code keeps + the old static `ASTERISK_TRANSFER_TARGET_MAP_JSON` behavior unchanged. +- `routing-service` gained `/agents` CRUD + `/internal/routing/reserve-agent` + and `/internal/routing/release-agent`, called by asterisk-bridge-service + the same way it already calls interaction-service/ai-voice-runtime-service + (bridge-issued bearer token, `ROUTING_SERVICE_URL`). +- Agent is released back to `AVAILABLE` when the call ends + (`process_call_ended` in `bridge_processing.py`), which also closes the + matching open `escalations` row. + +## Config for a given DID/queue + +``` +ASTERISK_QUEUE_LEVEL_MAP_JSON={"":{"level":"L2","tenant_id":"konturai"}} +``` + +Any `queue_code` not listed here keeps behaving exactly as before (static +`ASTERISK_TRANSFER_TARGET_MAP_JSON` redirect) - this is additive, not a +replacement of the legacy mechanism. + +Seed `agents` rows via `POST /agents` (admin/supervisor). Level `L2`, +`extension` must match a real SIP endpoint (currently only `2001`/`2002` +exist on the telecom server). + +## Known gap - no-answer retry + +Today the Routing Engine picks and reserves one agent and issues a single AMI +`Redirect` to their extension; if that extension does not answer, the caller +currently depends on whatever the dialplan does at that extension (unchanged +from before this change). Automatic "agent didn't answer -> release and pick +next" requires either: + +1. A small dialplan addition at the target extension's context: `Dial` with a + fixed timeout and, on failure, a `UserEvent` back into the bridge (mirrors + the existing `MVPCCAIFallbackToHuman` pattern) so `asterisk_bridge_service` + can call `/internal/routing/reserve-agent` again with the failed agent in + `exclude_agent_ids`, or +2. Switching from `Redirect` to AMI `Originate` with `Timeout` and reacting to + `OriginateResponse`. + +Left out of Phase 1 deliberately: changing dialplan behavior without full +visibility into what a given extension's context currently does on no-answer +risks silently dropping a live customer call. Needs a follow-up MR with an +explicit, reviewed dialplan change. + +## Not in Phase 1 + +- L3 (technical specialists) - `agents.level='L3'` and `routing_rules` already + support it; enabling it is a data-only follow-up once real L3 agents exist. +- SLA config / callback (ТЗ sections 20-21). +- Metrics (ТЗ section 38). diff --git a/migrations/sql/0031_agent_pool_routing_postgres.sql b/migrations/sql/0031_agent_pool_routing_postgres.sql new file mode 100644 index 0000000..6521c4e --- /dev/null +++ b/migrations/sql/0031_agent_pool_routing_postgres.sql @@ -0,0 +1,80 @@ +ALTER TABLE asterisk_call_links ADD COLUMN IF NOT EXISTS tenant_id TEXT; +ALTER TABLE asterisk_call_links ADD COLUMN IF NOT EXISTS current_level TEXT; +ALTER TABLE asterisk_call_links ADD COLUMN IF NOT EXISTS required_skills_json TEXT NOT NULL DEFAULT '[]'; +ALTER TABLE asterisk_call_links ADD COLUMN IF NOT EXISTS priority INTEGER NOT NULL DEFAULT 3; +CREATE INDEX IF NOT EXISTS idx_asterisk_call_links_tenant_id ON asterisk_call_links(tenant_id); +CREATE INDEX IF NOT EXISTS idx_asterisk_call_links_current_level ON asterisk_call_links(current_level); + +CREATE TABLE IF NOT EXISTS agents ( + id BIGSERIAL PRIMARY KEY, + agent_id TEXT NOT NULL UNIQUE, + tenant_ids_json TEXT NOT NULL DEFAULT '[]', + extension TEXT NOT NULL, + endpoint TEXT, + display_name TEXT NOT NULL, + level TEXT NOT NULL, + skills_json TEXT NOT NULL DEFAULT '[]', + status TEXT NOT NULL DEFAULT 'OFFLINE', + current_call_id TEXT, + max_concurrent_calls INTEGER NOT NULL DEFAULT 1, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + calls_handled_count INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS ix_agents_agent_id ON agents(agent_id); +CREATE INDEX IF NOT EXISTS ix_agents_extension ON agents(extension); +CREATE INDEX IF NOT EXISTS ix_agents_level ON agents(level); +CREATE INDEX IF NOT EXISTS ix_agents_status ON agents(status); +CREATE INDEX IF NOT EXISTS ix_agents_current_call_id ON agents(current_call_id); +CREATE INDEX IF NOT EXISTS ix_agents_enabled ON agents(enabled); +CREATE INDEX IF NOT EXISTS idx_agents_level_status_enabled ON agents(level, status, enabled); + +CREATE TABLE IF NOT EXISTS escalations ( + id BIGSERIAL PRIMARY KEY, + escalation_id TEXT NOT NULL UNIQUE, + call_id TEXT NOT NULL, + tenant_id TEXT, + from_level TEXT NOT NULL, + to_level TEXT NOT NULL, + reason_code TEXT NOT NULL, + required_skills_json TEXT NOT NULL DEFAULT '[]', + priority INTEGER NOT NULL DEFAULT 3, + topic TEXT, + summary TEXT, + status TEXT NOT NULL DEFAULT 'requested', + assigned_agent_id TEXT, + requested_at TEXT NOT NULL, + connected_at TEXT, + completed_at TEXT +); + +CREATE INDEX IF NOT EXISTS ix_escalations_escalation_id ON escalations(escalation_id); +CREATE INDEX IF NOT EXISTS ix_escalations_call_id ON escalations(call_id); +CREATE INDEX IF NOT EXISTS ix_escalations_tenant_id ON escalations(tenant_id); +CREATE INDEX IF NOT EXISTS ix_escalations_from_level ON escalations(from_level); +CREATE INDEX IF NOT EXISTS ix_escalations_to_level ON escalations(to_level); +CREATE INDEX IF NOT EXISTS ix_escalations_reason_code ON escalations(reason_code); +CREATE INDEX IF NOT EXISTS ix_escalations_status ON escalations(status); +CREATE INDEX IF NOT EXISTS ix_escalations_assigned_agent_id ON escalations(assigned_agent_id); +CREATE INDEX IF NOT EXISTS ix_escalations_requested_at ON escalations(requested_at); +CREATE INDEX IF NOT EXISTS idx_escalations_call_id_status ON escalations(call_id, status); + +CREATE TABLE IF NOT EXISTS routing_rules ( + id BIGSERIAL PRIMARY KEY, + rule_id TEXT NOT NULL UNIQUE, + from_level TEXT NOT NULL, + to_level TEXT NOT NULL, + reason_code TEXT, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + priority INTEGER NOT NULL DEFAULT 3, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS ix_routing_rules_rule_id ON routing_rules(rule_id); +CREATE INDEX IF NOT EXISTS ix_routing_rules_from_level ON routing_rules(from_level); +CREATE INDEX IF NOT EXISTS ix_routing_rules_to_level ON routing_rules(to_level); +CREATE INDEX IF NOT EXISTS ix_routing_rules_reason_code ON routing_rules(reason_code); +CREATE INDEX IF NOT EXISTS ix_routing_rules_enabled ON routing_rules(enabled); diff --git a/migrations/sql/0031_agent_pool_routing_sqlite.sql b/migrations/sql/0031_agent_pool_routing_sqlite.sql new file mode 100644 index 0000000..b4c2f8b --- /dev/null +++ b/migrations/sql/0031_agent_pool_routing_sqlite.sql @@ -0,0 +1,80 @@ +ALTER TABLE asterisk_call_links ADD COLUMN tenant_id TEXT; +ALTER TABLE asterisk_call_links ADD COLUMN current_level TEXT; +ALTER TABLE asterisk_call_links ADD COLUMN required_skills_json TEXT NOT NULL DEFAULT '[]'; +ALTER TABLE asterisk_call_links ADD COLUMN priority INTEGER NOT NULL DEFAULT 3; +CREATE INDEX IF NOT EXISTS idx_asterisk_call_links_tenant_id ON asterisk_call_links(tenant_id); +CREATE INDEX IF NOT EXISTS idx_asterisk_call_links_current_level ON asterisk_call_links(current_level); + +CREATE TABLE IF NOT EXISTS agents ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + agent_id TEXT NOT NULL UNIQUE, + tenant_ids_json TEXT NOT NULL DEFAULT '[]', + extension TEXT NOT NULL, + endpoint TEXT, + display_name TEXT NOT NULL, + level TEXT NOT NULL, + skills_json TEXT NOT NULL DEFAULT '[]', + status TEXT NOT NULL DEFAULT 'OFFLINE', + current_call_id TEXT, + max_concurrent_calls INTEGER NOT NULL DEFAULT 1, + enabled BOOLEAN NOT NULL DEFAULT 1, + calls_handled_count INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS ix_agents_agent_id ON agents(agent_id); +CREATE INDEX IF NOT EXISTS ix_agents_extension ON agents(extension); +CREATE INDEX IF NOT EXISTS ix_agents_level ON agents(level); +CREATE INDEX IF NOT EXISTS ix_agents_status ON agents(status); +CREATE INDEX IF NOT EXISTS ix_agents_current_call_id ON agents(current_call_id); +CREATE INDEX IF NOT EXISTS ix_agents_enabled ON agents(enabled); +CREATE INDEX IF NOT EXISTS idx_agents_level_status_enabled ON agents(level, status, enabled); + +CREATE TABLE IF NOT EXISTS escalations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + escalation_id TEXT NOT NULL UNIQUE, + call_id TEXT NOT NULL, + tenant_id TEXT, + from_level TEXT NOT NULL, + to_level TEXT NOT NULL, + reason_code TEXT NOT NULL, + required_skills_json TEXT NOT NULL DEFAULT '[]', + priority INTEGER NOT NULL DEFAULT 3, + topic TEXT, + summary TEXT, + status TEXT NOT NULL DEFAULT 'requested', + assigned_agent_id TEXT, + requested_at TEXT NOT NULL, + connected_at TEXT, + completed_at TEXT +); + +CREATE INDEX IF NOT EXISTS ix_escalations_escalation_id ON escalations(escalation_id); +CREATE INDEX IF NOT EXISTS ix_escalations_call_id ON escalations(call_id); +CREATE INDEX IF NOT EXISTS ix_escalations_tenant_id ON escalations(tenant_id); +CREATE INDEX IF NOT EXISTS ix_escalations_from_level ON escalations(from_level); +CREATE INDEX IF NOT EXISTS ix_escalations_to_level ON escalations(to_level); +CREATE INDEX IF NOT EXISTS ix_escalations_reason_code ON escalations(reason_code); +CREATE INDEX IF NOT EXISTS ix_escalations_status ON escalations(status); +CREATE INDEX IF NOT EXISTS ix_escalations_assigned_agent_id ON escalations(assigned_agent_id); +CREATE INDEX IF NOT EXISTS ix_escalations_requested_at ON escalations(requested_at); +CREATE INDEX IF NOT EXISTS idx_escalations_call_id_status ON escalations(call_id, status); + +CREATE TABLE IF NOT EXISTS routing_rules ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + rule_id TEXT NOT NULL UNIQUE, + from_level TEXT NOT NULL, + to_level TEXT NOT NULL, + reason_code TEXT, + enabled BOOLEAN NOT NULL DEFAULT 1, + priority INTEGER NOT NULL DEFAULT 3, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS ix_routing_rules_rule_id ON routing_rules(rule_id); +CREATE INDEX IF NOT EXISTS ix_routing_rules_from_level ON routing_rules(from_level); +CREATE INDEX IF NOT EXISTS ix_routing_rules_to_level ON routing_rules(to_level); +CREATE INDEX IF NOT EXISTS ix_routing_rules_reason_code ON routing_rules(reason_code); +CREATE INDEX IF NOT EXISTS ix_routing_rules_enabled ON routing_rules(enabled); diff --git a/services/asterisk_bridge_service/app.py b/services/asterisk_bridge_service/app.py index 1738226..9b3fd67 100644 --- a/services/asterisk_bridge_service/app.py +++ b/services/asterisk_bridge_service/app.py @@ -16,6 +16,8 @@ from services.shared.models import ( AsteriskBridgeStatusOut, AsteriskEventOut, BrowserSoftphoneConfigOut, + EscalationOut, + EscalationRequestIn, HealthResponse, VoiceAICallStateUpdateIn, VoiceAIHandoffRequestIn, @@ -105,6 +107,9 @@ _bridge_auth_role = bridge_config.bridge_auth_role _bridge_auth_subject = bridge_config.bridge_auth_subject _bridge_auth_token_ttl_seconds = bridge_config.bridge_auth_token_ttl_seconds _ai_voice_runtime_trusted_subjects = bridge_config.ai_voice_runtime_trusted_subjects +_routing_service_url = bridge_config.routing_service_url +_routing_level_for_queue_code = bridge_config.routing_level_for_queue_code +_routing_tenant_for_queue_code = bridge_config.routing_tenant_for_queue_code def _legacy_headers() -> dict[str, str]: @@ -298,6 +303,8 @@ _queue_code_for_queue_id = bridge_voice_ai._queue_code_for_queue_id _request_voice_ai_handoff = bridge_voice_ai.request_handoff _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 _first_non_empty = bridge_ami.first_non_empty _extract_call_id = bridge_ami.extract_call_id @@ -422,6 +429,15 @@ def blind_transfer_live_call( return bridge_routes.blind_transfer_live_call(call_id=call_id, body=body, actor=actor) +@app.post("/asterisk/live-calls/{call_id}/escalations", response_model=EscalationOut) +def escalate_live_call( + call_id: str, + body: EscalationRequestIn, + actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)), +) -> EscalationOut: + return bridge_routes.escalate_live_call(call_id=call_id, body=body, actor=actor) + + @app.get("/asterisk/live-calls/{call_id}/actions", response_model=list[VoiceCallActionOut]) def list_live_call_actions( call_id: str, diff --git a/services/asterisk_bridge_service/bridge_processing.py b/services/asterisk_bridge_service/bridge_processing.py index d380a4d..1118f6e 100644 --- a/services/asterisk_bridge_service/bridge_processing.py +++ b/services/asterisk_bridge_service/bridge_processing.py @@ -18,6 +18,7 @@ from services.shared.sql_models import ( AsteriskCallActionLogRow, AsteriskCallLinkRow, AsteriskEventLogRow, + EscalationRow, Interaction, VoiceAISessionRow, ) @@ -880,6 +881,21 @@ def process_call_ended( payload.get("DurationSeconds"), link.ai_state, ) + open_escalation = session.execute( + select(EscalationRow).where( + EscalationRow.call_id == row.call_id, + EscalationRow.status.in_(["requested", "ringing"]), + ) + ).scalar_one_or_none() + 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 try: bridge._notify_voice_ai_telephony_event( voice_session_id=link.voice_session_id, diff --git a/services/asterisk_bridge_service/config.py b/services/asterisk_bridge_service/config.py index 6f44030..43bdc33 100644 --- a/services/asterisk_bridge_service/config.py +++ b/services/asterisk_bridge_service/config.py @@ -176,6 +176,40 @@ def interaction_service_url() -> str: return os.getenv("INTERACTION_SERVICE_URL", "http://localhost:8004").rstrip("/") +def routing_service_url() -> str: + return os.getenv("ROUTING_SERVICE_URL", "http://localhost:8003").rstrip("/") + + +def queue_level_map() -> dict[str, dict[str, str | None]]: + raw = os.getenv("ASTERISK_QUEUE_LEVEL_MAP_JSON", "{}").strip() + try: + payload = json.loads(raw or "{}") + except json.JSONDecodeError as exc: + raise RuntimeError("Invalid ASTERISK_QUEUE_LEVEL_MAP_JSON") from exc + if not isinstance(payload, dict): + raise RuntimeError("ASTERISK_QUEUE_LEVEL_MAP_JSON must be a JSON object") + result: dict[str, dict[str, str | None]] = {} + for queue_code, item in payload.items(): + key = str(queue_code or "").strip() + if not key or not isinstance(item, dict): + continue + level = str(item.get("level") or "").strip() + if level not in {"L2", "L3"}: + continue + result[key] = {"level": level, "tenant_id": str(item.get("tenant_id") or "").strip() or None} + return result + + +def routing_level_for_queue_code(queue_code: str | None) -> str | None: + entry = queue_level_map().get(str(queue_code or "").strip()) + return entry.get("level") if entry else None + + +def routing_tenant_for_queue_code(queue_code: str | None) -> str | None: + entry = queue_level_map().get(str(queue_code or "").strip()) + return entry.get("tenant_id") if entry else None + + def voice_adapter_service_url() -> str: return os.getenv("VOICE_ADAPTER_SERVICE_URL", "http://localhost:8006").rstrip("/") diff --git a/services/asterisk_bridge_service/routes.py b/services/asterisk_bridge_service/routes.py index 607fea5..b690987 100644 --- a/services/asterisk_bridge_service/routes.py +++ b/services/asterisk_bridge_service/routes.py @@ -6,6 +6,8 @@ from sqlalchemy import select from services.shared.core import utc_now_iso from services.shared.db import get_session from services.shared.models import ( + EscalationOut, + EscalationRequestIn, HealthResponse, VoiceAICallStateUpdateIn, VoiceAIHandoffRequestIn, @@ -95,6 +97,11 @@ def get_live_call_ai_summary(call_id: str) -> VoiceAISummaryOut | None: return bridge._voice_ai_summary_for_call(call_id) +def escalate_live_call(call_id: str, body: EscalationRequestIn, actor: dict) -> EscalationOut: + bridge = _bridge_app() + return bridge._create_escalation(call_id, body, actor) + + def voice_ai_handoff_call(call_id: str, body: VoiceAIHandoffRequestIn, actor: dict) -> VoiceLiveCallOut: bridge = _bridge_app() return bridge._request_voice_ai_handoff(call_id, body, actor) diff --git a/services/asterisk_bridge_service/voice_ai.py b/services/asterisk_bridge_service/voice_ai.py index 0b5fda3..615ff82 100644 --- a/services/asterisk_bridge_service/voice_ai.py +++ b/services/asterisk_bridge_service/voice_ai.py @@ -10,6 +10,8 @@ from sqlalchemy import select from services.shared.core import new_id, utc_now_iso from services.shared.db import get_session from services.shared.models import ( + EscalationOut, + EscalationRequestIn, VoiceAICallStateUpdateIn, VoiceAIHandoffRequestIn, VoiceAIMediaBridgeEventIn, @@ -17,7 +19,13 @@ from services.shared.models import ( VoiceAISummaryTranscriptSegmentOut, VoiceLiveCallOut, ) -from services.shared.sql_models import AISessionRow, AsteriskCallLinkRow, VoiceAISessionRow, VoiceTranscriptSegmentRow +from services.shared.sql_models import ( + AISessionRow, + AsteriskCallLinkRow, + EscalationRow, + VoiceAISessionRow, + VoiceTranscriptSegmentRow, +) from services.shared.voice_transcripts import add_transcript_segment, next_transcript_sequence @@ -174,18 +182,74 @@ def _queue_code_for_queue_id(queue_id: str | None) -> str | None: return None -def _resolve_handoff_extension(target_queue_id: str | None, fallback_queue_code: str | None) -> tuple[str, str]: +def _reserve_routing_agent( + *, + call_id: str, + level: str, + tenant_id: str | None, + required_skills: list[str] | None = None, +) -> str | None: + bridge = _bridge_app() + try: + response = bridge._post_json( + f"{bridge._routing_service_url()}/internal/routing/reserve-agent", + { + "call_id": call_id, + "level": level, + "tenant_id": tenant_id, + "required_skills": required_skills or [], + "exclude_agent_ids": [], + }, + timeout_seconds=bridge._callcontrol_side_effect_timeout_seconds(), + max_attempts=1, + retry_backoff_seconds=0.0, + ) + except Exception: + LOGGER.warning( + "bridge.routing_reserve_failed call_id=%s level=%s tenant_id=%s", + call_id, + level, + tenant_id, + ) + return None + extension = str((response or {}).get("extension") or "").strip() + return extension or None + + +def _resolve_handoff_extension( + target_queue_id: str | None, + fallback_queue_code: str | None, + *, + call_id: str | None = None, + target_level: str | None = None, + tenant_id: str | None = None, + required_skills: list[str] | None = None, +) -> tuple[str, str, 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: raise HTTPException(status_code=400, detail="Unable to resolve handoff queue_code") + + 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( + 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 + raise HTTPException(status_code=409, detail=f"No available {level} agent right now") + extension = bridge._transfer_target_map().get(queue_code) if not extension: raise HTTPException( status_code=400, detail=f"Unknown transfer queue_code: {queue_code}", ) - return queue_code, extension + return queue_code, extension, None, None def _resolve_handoff_channel(session, link: AsteriskCallLinkRow) -> str: @@ -395,9 +459,10 @@ 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 = _resolve_handoff_extension( + queue_code, target_extension, resolved_level, resolved_tenant_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, ) handoff_metadata = body.metadata or {} actor_user = str(actor.get("user") or actor.get("sub") or "ai-voice-runtime").strip() @@ -477,6 +542,9 @@ def request_handoff(call_id: str, body: VoiceAIHandoffRequestIn, actor: dict) -> link.claimed_by_user = None link.claimed_at = None link.operator_extension = None + if resolved_level: + link.current_level = resolved_level + link.tenant_id = resolved_tenant_id if reuse_queue_id: link.queue_code = queue_code link.queue_id = reuse_queue_id @@ -573,6 +641,111 @@ def request_handoff(call_id: str, body: VoiceAIHandoffRequestIn, actor: dict) -> session.close() +def release_routing_agent(call_id: str) -> None: + bridge = _bridge_app() + bridge._post_json( + f"{bridge._routing_service_url()}/internal/routing/release-agent", + {"call_id": call_id}, + timeout_seconds=bridge._callcontrol_side_effect_timeout_seconds(), + max_attempts=1, + retry_backoff_seconds=0.0, + ) + + +def _escalation_to_out(row: EscalationRow) -> EscalationOut: + return EscalationOut( + escalation_id=row.escalation_id, + call_id=row.call_id, + tenant_id=row.tenant_id, + from_level=row.from_level, + to_level=row.to_level, + reason_code=row.reason_code, + required_skills=json.loads(row.required_skills_json or "[]"), + priority=row.priority, + topic=row.topic, + summary=row.summary, + status=row.status, + assigned_agent_id=row.assigned_agent_id, + requested_at=row.requested_at, + connected_at=row.connected_at, + completed_at=row.completed_at, + ) + + +def create_escalation(call_id: str, body: EscalationRequestIn, actor: dict) -> EscalationOut: + bridge = _bridge_app() + session = get_session() + try: + link = bridge._load_live_call_or_404(session, call_id) + if bridge._call_is_ended(link): + raise HTTPException(status_code=409, detail="Call is already ended") + + existing = session.execute( + select(EscalationRow).where( + EscalationRow.call_id == call_id, + EscalationRow.status.in_(["requested", "ringing"]), + ) + ).scalar_one_or_none() + if existing is not None: + raise HTTPException(status_code=409, detail="An active escalation already exists for this call") + + from_level = str(link.current_level or "L1") + now = utc_now_iso() + escalation = EscalationRow( + escalation_id=new_id("esc"), + call_id=call_id, + tenant_id=link.tenant_id, + from_level=from_level, + to_level=body.target_level, + reason_code=body.reason_code, + required_skills_json=json.dumps(body.required_skills, ensure_ascii=False), + priority=body.priority, + topic=body.topic, + summary=body.summary, + status="requested", + requested_at=now, + ) + session.add(escalation) + session.flush() + + channel = _resolve_handoff_channel(session, link) + reserved_extension = _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: + escalation.status = "failed" + escalation.completed_at = now + session.commit() + raise HTTPException(status_code=409, detail=f"No available {body.target_level} agent right now") + + bridge._ami_action( + "Redirect", + { + "Channel": channel, + "Context": bridge._transfer_context(), + "Exten": reserved_extension, + "Priority": 1, + }, + ) + + escalation.status = "ringing" + escalation.assigned_agent_id = reserved_extension + link.current_level = body.target_level + link.required_skills_json = json.dumps(body.required_skills, ensure_ascii=False) + link.priority = body.priority + link.claimed_by_user = None + link.claimed_at = None + link.operator_extension = None + link.updated_at = now + session.commit() + return _escalation_to_out(escalation) + finally: + session.close() + + def update_call_ai_state(call_id: str, body: VoiceAICallStateUpdateIn, actor: dict) -> VoiceLiveCallOut: bridge = _bridge_app() assert_trusted_voice_runtime_actor(actor) diff --git a/services/routing_service/app.py b/services/routing_service/app.py index e63321d..f49bc29 100644 --- a/services/routing_service/app.py +++ b/services/routing_service/app.py @@ -5,12 +5,22 @@ import json from fastapi import Depends, FastAPI, HTTPException from sqlalchemy import select, text +from services.routing_service import engine as routing_engine from services.shared.core import Role, new_id, utc_now_iso from services.shared.db import engine, get_session -from services.shared.models import HealthResponse, QueueCreate, QueueOut +from services.shared.models import ( + AgentCreate, + AgentPoolOut, + AgentStatusUpdateIn, + HealthResponse, + QueueCreate, + QueueOut, + RoutingAgentReserveIn, + RoutingAgentReserveOut, +) from services.shared.security import require_roles from services.shared.sql_init import init_sql_schema -from services.shared.sql_models import IvrSessionRow, Queue, RoutingCounter +from services.shared.sql_models import AgentRow, IvrSessionRow, Queue, RoutingCounter app = FastAPI(title="routing-service", version="1.1.0") @@ -230,3 +240,132 @@ def delete_queue( return {"queue_id": queue_id, "deleted": True} finally: session.close() + + +def _agent_to_out(row: AgentRow) -> AgentPoolOut: + return AgentPoolOut( + agent_id=row.agent_id, + tenant_ids=routing_engine.parse_list_json(row.tenant_ids_json), + extension=row.extension, + endpoint=row.endpoint, + display_name=row.display_name, + level=row.level, + skills=routing_engine.parse_list_json(row.skills_json), + status=row.status, + current_call_id=row.current_call_id, + max_concurrent_calls=row.max_concurrent_calls, + enabled=row.enabled, + calls_handled_count=row.calls_handled_count, + created_at=row.created_at, + updated_at=row.updated_at, + ) + + +@app.post("/agents", response_model=AgentPoolOut) +def create_agent( + payload: AgentCreate, + _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)), +) -> AgentPoolOut: + session = get_session() + try: + now = utc_now_iso() + row = AgentRow( + agent_id=new_id("agt"), + tenant_ids_json=json.dumps(payload.tenant_ids, ensure_ascii=False), + extension=payload.extension, + endpoint=payload.endpoint, + display_name=payload.display_name, + level=payload.level, + skills_json=json.dumps(payload.skills, ensure_ascii=False), + status="OFFLINE", + max_concurrent_calls=payload.max_concurrent_calls, + enabled=payload.enabled, + created_at=now, + updated_at=now, + ) + session.add(row) + session.commit() + session.refresh(row) + return _agent_to_out(row) + finally: + session.close() + + +@app.get("/agents", response_model=list[AgentPoolOut]) +def list_agents( + level: str | None = None, + _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), +) -> list[AgentPoolOut]: + session = get_session() + try: + stmt = select(AgentRow).order_by(AgentRow.id.desc()) + if level: + stmt = stmt.where(AgentRow.level == level) + rows = session.execute(stmt).scalars().all() + return [_agent_to_out(r) for r in rows] + finally: + session.close() + + +@app.patch("/agents/{agent_id}/status", response_model=AgentPoolOut) +def update_agent_status( + agent_id: str, + payload: AgentStatusUpdateIn, + _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), +) -> AgentPoolOut: + session = get_session() + try: + row = session.execute(select(AgentRow).where(AgentRow.agent_id == agent_id)).scalar_one_or_none() + if not row: + raise HTTPException(status_code=404, detail="Agent not found") + row.status = payload.status + if payload.status == "AVAILABLE": + row.current_call_id = None + row.updated_at = utc_now_iso() + session.commit() + return _agent_to_out(row) + finally: + session.close() + + +@app.post("/internal/routing/reserve-agent", response_model=RoutingAgentReserveOut) +def reserve_agent_endpoint( + payload: RoutingAgentReserveIn, + _: dict = Depends(require_roles(Role.ADMIN)), +) -> RoutingAgentReserveOut: + session = get_session() + try: + agent = routing_engine.reserve_agent( + session, + call_id=payload.call_id, + level=payload.level, + tenant_id=payload.tenant_id, + required_skills=payload.required_skills, + exclude_agent_ids=payload.exclude_agent_ids, + ) + if agent is None: + raise HTTPException(status_code=409, detail=f"No available {payload.level} agent") + return RoutingAgentReserveOut( + agent_id=agent.agent_id, + extension=agent.extension, + endpoint=agent.endpoint, + display_name=agent.display_name, + ) + finally: + session.close() + + +@app.post("/internal/routing/release-agent") +def release_agent_endpoint( + payload: dict, + _: 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") + 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} + finally: + session.close() diff --git a/services/routing_service/engine.py b/services/routing_service/engine.py new file mode 100644 index 0000000..09b5f96 --- /dev/null +++ b/services/routing_service/engine.py @@ -0,0 +1,131 @@ +from __future__ import annotations + +import json + +from sqlalchemy import select, text + +from services.shared.core import utc_now_iso +from services.shared.sql_models import AgentRow + + +def parse_list_json(raw: str | None) -> list[str]: + if not raw: + return [] + try: + data = json.loads(raw) + except (TypeError, ValueError): + return [] + if isinstance(data, list): + return [str(item) for item in data] + return [] + + +def _agent_matches( + agent: AgentRow, + *, + level: str, + tenant_id: str | None, + required_skills: list[str], + exclude_agent_ids: set[str], +) -> bool: + if agent.agent_id in exclude_agent_ids: + return False + if not agent.enabled: + return False + if agent.level != level: + return False + if agent.status != "AVAILABLE": + return False + tenant_ids = parse_list_json(agent.tenant_ids_json) + if tenant_id and tenant_ids and tenant_id not in tenant_ids: + return False + skills = set(parse_list_json(agent.skills_json)) + if required_skills and not set(required_skills).issubset(skills): + return False + return True + + +def select_candidate_agents( + session, + *, + level: str, + tenant_id: str | None, + required_skills: list[str] | None = None, + exclude_agent_ids: list[str] | None = None, +) -> list[AgentRow]: + exclude = set(exclude_agent_ids or []) + skills = required_skills or [] + rows = session.execute( + select(AgentRow).where(AgentRow.level == level, AgentRow.enabled.is_(True)) + ).scalars().all() + candidates = [ + row + for row in rows + if _agent_matches(row, level=level, tenant_id=tenant_id, required_skills=skills, exclude_agent_ids=exclude) + ] + candidates.sort(key=lambda a: (a.updated_at, a.calls_handled_count)) + return candidates + + +def reserve_agent( + session, + *, + call_id: str, + level: str, + tenant_id: str | None, + required_skills: list[str] | None = None, + exclude_agent_ids: list[str] | None = None, +) -> AgentRow | None: + candidates = select_candidate_agents( + session, + level=level, + tenant_id=tenant_id, + required_skills=required_skills, + exclude_agent_ids=exclude_agent_ids, + ) + now = utc_now_iso() + for candidate in candidates: + result = session.execute( + text( + """ + UPDATE agents SET status='RESERVED', current_call_id=:call_id, updated_at=:now + WHERE agent_id=:agent_id AND status='AVAILABLE' + """ + ), + {"call_id": call_id, "now": now, "agent_id": candidate.agent_id}, + ) + if result.rowcount == 1: + session.commit() + session.refresh(candidate) + return candidate + session.rollback() + return None + + +def release_agent_by_call_id(session, *, call_id: str) -> AgentRow | None: + agent = session.execute( + select(AgentRow).where(AgentRow.current_call_id == call_id) + ).scalar_one_or_none() + if agent is None: + return None + now = utc_now_iso() + agent.status = "AVAILABLE" + agent.current_call_id = None + agent.calls_handled_count = (agent.calls_handled_count or 0) + 1 + agent.updated_at = now + session.commit() + return agent + + +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) + ).scalar_one_or_none() + if agent is None: + return None + now = utc_now_iso() + agent.status = next_status + agent.current_call_id = None + agent.updated_at = now + session.commit() + return agent diff --git a/services/shared/models.py b/services/shared/models.py index 2e7be51..ff83c45 100644 --- a/services/shared/models.py +++ b/services/shared/models.py @@ -1253,6 +1253,84 @@ class AgentStateOut(AgentStateIn): updated_at: str +AgentLevel = Literal["L2", "L3"] +AgentStatus = Literal["OFFLINE", "AVAILABLE", "RESERVED", "RINGING", "TALKING", "AFTER_CALL_WORK", "PAUSED"] + + +class AgentCreate(BaseModel): + tenant_ids: list[str] = Field(default_factory=list) + extension: str = Field(min_length=1) + endpoint: str | None = None + display_name: str = Field(min_length=1) + level: AgentLevel + skills: list[str] = Field(default_factory=list) + max_concurrent_calls: int = Field(default=1, ge=1) + enabled: bool = True + + +class AgentStatusUpdateIn(BaseModel): + status: AgentStatus + + +class AgentPoolOut(BaseModel): + agent_id: str + tenant_ids: list[str] = Field(default_factory=list) + extension: str + endpoint: str | None = None + display_name: str + level: AgentLevel + skills: list[str] = Field(default_factory=list) + status: AgentStatus + current_call_id: str | None = None + max_concurrent_calls: int + enabled: bool + calls_handled_count: int = 0 + created_at: str + updated_at: str + + +class EscalationRequestIn(BaseModel): + target_level: AgentLevel + reason_code: str = Field(min_length=1) + topic: str | None = None + required_skills: list[str] = Field(default_factory=list) + priority: int = Field(default=3, ge=1, le=5) + summary: str | None = None + + +class EscalationOut(BaseModel): + escalation_id: str + call_id: str + tenant_id: str | None = None + from_level: str + to_level: str + reason_code: str + required_skills: list[str] = Field(default_factory=list) + priority: int + topic: str | None = None + summary: str | None = None + status: str + assigned_agent_id: str | None = None + requested_at: str + connected_at: str | None = None + completed_at: str | None = None + + +class RoutingAgentReserveIn(BaseModel): + call_id: str = Field(min_length=1) + level: AgentLevel + tenant_id: str | None = None + required_skills: list[str] = Field(default_factory=list) + exclude_agent_ids: list[str] = Field(default_factory=list) + + +class RoutingAgentReserveOut(BaseModel): + agent_id: str + extension: str + endpoint: str | None = None + display_name: str + + class AIAnalyticsWindowOut(BaseModel): from_ts: str to_ts: str diff --git a/services/shared/sql_init.py b/services/shared/sql_init.py index f090062..4db4ec6 100644 --- a/services/shared/sql_init.py +++ b/services/shared/sql_init.py @@ -471,6 +471,25 @@ def _apply_runtime_schema_compatibility() -> None: "ON asterisk_call_links(customer_name_resolved_at)" ) ) + _add_column_if_missing(conn, columns, "asterisk_call_links", "tenant_id", "VARCHAR(64)") + _add_column_if_missing(conn, columns, "asterisk_call_links", "current_level", "VARCHAR(16)") + _add_column_if_missing(conn, columns, "asterisk_call_links", "required_skills_json", "TEXT DEFAULT '[]'") + _add_column_if_missing(conn, columns, "asterisk_call_links", "priority", "INTEGER DEFAULT 3") + indexes = _table_indexes(inspector, "asterisk_call_links") + if "idx_asterisk_call_links_tenant_id" not in indexes: + conn.execute( + text( + "CREATE INDEX IF NOT EXISTS idx_asterisk_call_links_tenant_id " + "ON asterisk_call_links(tenant_id)" + ) + ) + if "idx_asterisk_call_links_current_level" not in indexes: + conn.execute( + text( + "CREATE INDEX IF NOT EXISTS idx_asterisk_call_links_current_level " + "ON asterisk_call_links(current_level)" + ) + ) if "voice_ai_sessions" in table_names: columns = _table_columns(inspector, "voice_ai_sessions") diff --git a/services/shared/sql_models.py b/services/shared/sql_models.py index b7d8477..1b9d6a1 100644 --- a/services/shared/sql_models.py +++ b/services/shared/sql_models.py @@ -280,6 +280,10 @@ class AsteriskCallLinkRow(Base): connected_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) ended_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) updated_at: Mapped[str] = mapped_column(String(64), index=True) + tenant_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) + current_level: Mapped[str | None] = mapped_column(String(16), nullable=True, index=True) + required_skills_json: Mapped[str] = mapped_column(Text, default="[]") + priority: Mapped[int] = mapped_column(Integer, default=3) class AsteriskCallActionLogRow(Base): @@ -804,3 +808,66 @@ class SupervisorQueueSnapshotRow(Base): in_queue: Mapped[int] = mapped_column(Integer, default=0) avg_wait_seconds: Mapped[int] = mapped_column(Integer, default=0) updated_at: Mapped[str] = mapped_column(String(64), index=True) + + +class AgentRow(Base): + __tablename__ = "agents" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + agent_id: Mapped[str] = mapped_column(String(64), unique=True, index=True) + tenant_ids_json: Mapped[str] = mapped_column(Text, default="[]") + extension: Mapped[str] = mapped_column(String(64), index=True) + endpoint: Mapped[str | None] = mapped_column(String(128), nullable=True) + display_name: Mapped[str] = mapped_column(String(256)) + level: Mapped[str] = mapped_column(String(16), index=True) + skills_json: Mapped[str] = mapped_column(Text, default="[]") + status: Mapped[str] = mapped_column(String(32), index=True, default="OFFLINE") + current_call_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True) + max_concurrent_calls: Mapped[int] = mapped_column(Integer, default=1) + enabled: Mapped[bool] = mapped_column(Boolean, default=True, index=True) + calls_handled_count: Mapped[int] = mapped_column(Integer, default=0) + created_at: Mapped[str] = mapped_column(String(64)) + updated_at: Mapped[str] = mapped_column(String(64), index=True) + + +class EscalationRow(Base): + __tablename__ = "escalations" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + escalation_id: Mapped[str] = mapped_column(String(64), unique=True, index=True) + call_id: Mapped[str] = mapped_column(String(128), index=True) + tenant_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) + from_level: Mapped[str] = mapped_column(String(16), index=True) + to_level: Mapped[str] = mapped_column(String(16), index=True) + reason_code: Mapped[str] = mapped_column(String(64), index=True) + required_skills_json: Mapped[str] = mapped_column(Text, default="[]") + priority: Mapped[int] = mapped_column(Integer, default=3) + topic: Mapped[str | None] = mapped_column(String(256), nullable=True) + 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) + 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) + + __table_args__ = ( + Index( + "idx_escalations_call_id_status", + "call_id", + "status", + ), + ) + + +class RoutingRuleRow(Base): + __tablename__ = "routing_rules" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + rule_id: Mapped[str] = mapped_column(String(64), unique=True, index=True) + from_level: Mapped[str] = mapped_column(String(16), index=True) + to_level: Mapped[str] = mapped_column(String(16), index=True) + reason_code: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) + enabled: Mapped[bool] = mapped_column(Boolean, default=True, index=True) + priority: Mapped[int] = mapped_column(Integer, default=3) + created_at: Mapped[str] = mapped_column(String(64)) + updated_at: Mapped[str] = mapped_column(String(64)) diff --git a/tests/test_routing_engine.py b/tests/test_routing_engine.py new file mode 100644 index 0000000..d0aab69 --- /dev/null +++ b/tests/test_routing_engine.py @@ -0,0 +1,136 @@ +import json + +from sqlalchemy import select + +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 + + +def _make_agent(session, *, level="L2", tenant_ids=None, skills=None, status="AVAILABLE", enabled=True): + now = utc_now_iso() + row = AgentRow( + agent_id=new_id("agt"), + tenant_ids_json=json.dumps(tenant_ids or []), + extension="2001", + endpoint=None, + display_name="Test Agent", + level=level, + skills_json=json.dumps(skills or []), + status=status, + max_concurrent_calls=1, + enabled=enabled, + created_at=now, + updated_at=now, + ) + session.add(row) + session.commit() + session.refresh(row) + return row + + +def test_reserve_agent_picks_available_agent_of_requested_level(): + init_sql_schema() + session = get_session() + try: + agent_l2 = _make_agent(session, level="L2") + _make_agent(session, level="L3") + + reserved = routing_engine.reserve_agent( + session, + call_id="call-1", + level="L2", + tenant_id=None, + required_skills=[], + ) + assert reserved is not None + assert reserved.agent_id == agent_l2.agent_id + assert reserved.status == "RESERVED" + assert reserved.current_call_id == "call-1" + finally: + session.close() + + +def test_reserve_agent_does_not_double_reserve(): + init_sql_schema() + session = get_session() + try: + _make_agent(session, level="L2") + + first = routing_engine.reserve_agent(session, call_id="call-a", level="L2", tenant_id=None, required_skills=[]) + second = routing_engine.reserve_agent(session, call_id="call-b", level="L2", tenant_id=None, required_skills=[]) + assert first is not None + assert second is None + finally: + session.close() + + +def test_reserve_agent_filters_by_tenant_and_skills(): + init_sql_schema() + session = get_session() + try: + wrong_tenant = _make_agent(session, level="L2", tenant_ids=["other"]) + no_skill = _make_agent(session, level="L2", tenant_ids=["konturai"], skills=[]) + matching = _make_agent(session, level="L2", tenant_ids=["konturai"], skills=["billing"]) + + reserved = routing_engine.reserve_agent( + session, + call_id="call-skill", + level="L2", + tenant_id="konturai", + required_skills=["billing"], + ) + assert reserved is not None + assert reserved.agent_id == matching.agent_id + assert reserved.agent_id != wrong_tenant.agent_id + assert reserved.agent_id != no_skill.agent_id + finally: + session.close() + + +def test_release_agent_by_call_id_makes_agent_available_again(): + init_sql_schema() + session = get_session() + try: + agent = _make_agent(session, level="L2") + reserved = routing_engine.reserve_agent(session, call_id="call-z", level="L2", tenant_id=None, required_skills=[]) + assert reserved is not None + + released = routing_engine.release_agent_by_call_id(session, call_id="call-z") + assert released is not None + assert released.agent_id == agent.agent_id + assert released.status == "AVAILABLE" + assert released.current_call_id is None + assert released.calls_handled_count == 1 + + again = routing_engine.reserve_agent(session, call_id="call-again", level="L2", tenant_id=None, required_skills=[]) + assert again is not None + assert again.agent_id == agent.agent_id + finally: + session.close() + + +def test_reserve_agent_excludes_disabled_and_excluded_ids(): + init_sql_schema() + session = get_session() + try: + disabled = _make_agent(session, level="L2", enabled=False) + excluded = _make_agent(session, level="L2") + available = _make_agent(session, level="L2") + + reserved = routing_engine.reserve_agent( + session, + call_id="call-exc", + level="L2", + tenant_id=None, + required_skills=[], + exclude_agent_ids=[excluded.agent_id], + ) + assert reserved is not None + assert reserved.agent_id == available.agent_id + assert reserved.agent_id != disabled.agent_id + assert reserved.agent_id != excluded.agent_id + finally: + session.close() From 1ad4e1ec6e2eab39d6de8b1bce886eedbfe8b19b Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sat, 29 Aug 2026 13:28:14 +0500 Subject: [PATCH 2/2] feat: surface L2 agent pool and escalations on supervisor screen - routing-service: GET /escalations (list recent escalation attempts) - supervisor UI: new panel showing the real agent pool (status, level, tenant, skills, calls handled) fed by GET /agents, with a form to add operators to the pool - supervisor UI: new live escalation feed (AI->L2 handoffs, status, assigned operator), auto-refreshed every 5s alongside live calls --- services/routing_service/app.py | 40 ++++++- ui/operator/styles.css | 63 +++++++++++ ui/supervisor/app.js | 190 +++++++++++++++++++++++++++++++- ui/supervisor/index.html | 41 +++++++ 4 files changed, 332 insertions(+), 2 deletions(-) diff --git a/services/routing_service/app.py b/services/routing_service/app.py index f49bc29..dc16848 100644 --- a/services/routing_service/app.py +++ b/services/routing_service/app.py @@ -12,6 +12,7 @@ from services.shared.models import ( AgentCreate, AgentPoolOut, AgentStatusUpdateIn, + EscalationOut, HealthResponse, QueueCreate, QueueOut, @@ -20,7 +21,7 @@ from services.shared.models import ( ) from services.shared.security import require_roles from services.shared.sql_init import init_sql_schema -from services.shared.sql_models import AgentRow, IvrSessionRow, Queue, RoutingCounter +from services.shared.sql_models import AgentRow, EscalationRow, IvrSessionRow, Queue, RoutingCounter app = FastAPI(title="routing-service", version="1.1.0") @@ -328,6 +329,43 @@ def update_agent_status( session.close() +def _escalation_to_out(row: EscalationRow) -> EscalationOut: + return EscalationOut( + escalation_id=row.escalation_id, + call_id=row.call_id, + tenant_id=row.tenant_id, + from_level=row.from_level, + to_level=row.to_level, + reason_code=row.reason_code, + required_skills=routing_engine.parse_list_json(row.required_skills_json), + priority=row.priority, + topic=row.topic, + summary=row.summary, + status=row.status, + assigned_agent_id=row.assigned_agent_id, + requested_at=row.requested_at, + connected_at=row.connected_at, + completed_at=row.completed_at, + ) + + +@app.get("/escalations", response_model=list[EscalationOut]) +def list_escalations( + status: str | None = None, + limit: int = 50, + _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)), +) -> list[EscalationOut]: + session = get_session() + try: + stmt = select(EscalationRow).order_by(EscalationRow.id.desc()).limit(max(1, min(limit, 200))) + if status: + stmt = stmt.where(EscalationRow.status == status) + rows = session.execute(stmt).scalars().all() + return [_escalation_to_out(r) for r in rows] + finally: + session.close() + + @app.post("/internal/routing/reserve-agent", response_model=RoutingAgentReserveOut) def reserve_agent_endpoint( payload: RoutingAgentReserveIn, diff --git a/ui/operator/styles.css b/ui/operator/styles.css index c7c83e7..607d72d 100644 --- a/ui/operator/styles.css +++ b/ui/operator/styles.css @@ -5987,3 +5987,66 @@ body.analytics-drilldown-open { padding: 10px 16px 0; } } + +.agent-pool-item, +.escalation-item { + display: flex; + flex-wrap: wrap; + align-items: center; + gap: 6px 10px; + padding: 12px 14px; + border: 1px solid var(--line); + border-radius: 12px; + margin-bottom: 8px; + background: #fff; +} + +.agent-pool-item strong, +.escalation-item strong { + flex-basis: 100%; +} + +.status-badge { + display: inline-flex; + align-items: center; + gap: 6px; + padding: 2px 10px; + border-radius: 999px; + font-size: 12px; + font-weight: 700; + white-space: nowrap; +} + +.status-badge::before { + content: ""; + width: 6px; + height: 6px; + border-radius: 50%; + background: currentColor; +} + +.status-badge.is-success { + background: var(--success-soft); + color: var(--success); +} + +.status-badge.is-warning { + background: var(--warning-soft); + color: var(--warning); +} + +.status-badge.is-danger { + background: var(--danger-soft); + color: var(--danger); +} + +.status-badge.is-muted { + background: #eef1f6; + color: #6b7280; +} + +.agent-pool-meta, +.escalation-meta { + font-size: 12px; + color: #6b7280; +} diff --git a/ui/supervisor/app.js b/ui/supervisor/app.js index a654d49..159a84c 100644 --- a/ui/supervisor/app.js +++ b/ui/supervisor/app.js @@ -698,6 +698,178 @@ async function setAgentState() { } } +const AGENT_STATUS_LABELS = { + AVAILABLE: 'Свободен', + RESERVED: 'Зарезервирован', + RINGING: 'Звонок идёт', + TALKING: 'На линии', + AFTER_CALL_WORK: 'Постобработка', + PAUSED: 'Пауза', + OFFLINE: 'Offline', +}; + +const AGENT_STATUS_TONE = { + AVAILABLE: 'is-success', + RESERVED: 'is-warning', + RINGING: 'is-warning', + TALKING: 'is-success', + AFTER_CALL_WORK: 'is-warning', + PAUSED: 'is-muted', + OFFLINE: 'is-danger', +}; + +const ESCALATION_STATUS_LABELS = { + requested: 'Запрошена', + ringing: 'Дозвон оператора', + completed: 'Завершена', + failed: 'Не удалась', +}; + +const ESCALATION_STATUS_TONE = { + requested: 'is-warning', + ringing: 'is-warning', + completed: 'is-success', + failed: 'is-danger', +}; + +function statusBadge(labels, tones, value) { + const label = labels[value] || value || 'неизвестно'; + const tone = tones[value] || 'is-muted'; + return `${escapeHtml(label)}`; +} + +function renderCallAgents(items) { + const byStatus = {}; + items.forEach((item) => { + byStatus[item.status] = (byStatus[item.status] || 0) + 1; + }); + $('callAgentsSummary').innerHTML = [ + renderSummaryCard('Всего в пуле', items.length, 'Операторы L2/L3'), + renderSummaryCard('Свободны', byStatus.AVAILABLE || 0, 'Могут принять звонок сейчас'), + renderSummaryCard('На линии', (byStatus.TALKING || 0) + (byStatus.RINGING || 0) + (byStatus.RESERVED || 0), 'Заняты звонком или дозвоном'), + renderSummaryCard('Offline', byStatus.OFFLINE || 0, 'Недоступны'), + ].join(''); + + if (!items.length) { + $('callAgentsList').innerHTML = '
  • Пул пуст. Добавьте оператора ниже.
    • '; + return; + } + + $('callAgentsList').innerHTML = items + .map( + (item) => ` +
  • + ${escapeHtml(item.display_name)} · вн. ${escapeHtml(item.extension)} + ${statusBadge(AGENT_STATUS_LABELS, AGENT_STATUS_TONE, item.status)} + ${escapeHtml(item.level)} + ${item.enabled ? '' : 'отключён'} + tenant: ${escapeHtml((item.tenant_ids || []).join(', ') || '-')} + навыки: ${escapeHtml((item.skills || []).join(', ') || '-')} + принято звонков: ${Number(item.calls_handled_count || 0)} + ${item.current_call_id ? `текущий звонок: ${escapeHtml(item.current_call_id)}` : ''} +
    • + `, + ) + .join(''); +} + +async function loadCallAgents(silent = false) { + try { + const data = await api('routing', 'agents'); + renderCallAgents(Array.isArray(data) ? data : []); + if (!silent) { + log('Пул L2-операторов обновлён', { count: data.length }); + } + } catch (err) { + $('callAgentsList').innerHTML = '
  • Не удалось загрузить пул операторов.
    • '; + if (!silent) { + log('Не удалось загрузить пул операторов', { error: err.message }); + } + } +} + +async function createCallAgent() { + try { + const extension = $('newAgentExtension').value.trim(); + if (!extension) { + log('Укажите внутренний номер оператора'); + return; + } + const skills = $('newAgentSkills').value + .split(',') + .map((item) => item.trim()) + .filter(Boolean); + const tenantId = $('newAgentTenant').value.trim(); + const data = await api('routing', 'agents', { + method: 'POST', + body: JSON.stringify({ + extension, + display_name: $('newAgentDisplayName').value.trim() || `Оператор ${extension}`, + level: $('newAgentLevel').value, + tenant_ids: tenantId ? [tenantId] : [], + skills, + max_concurrent_calls: 1, + enabled: true, + }), + }); + log('Оператор добавлен в пул', { agent_id: data.agent_id, extension: data.extension }); + $('newAgentExtension').value = ''; + $('newAgentDisplayName').value = ''; + $('newAgentSkills').value = ''; + await loadCallAgents(); + } catch (err) { + log('Не удалось добавить оператора', { error: err.message }); + } +} + +function renderEscalations(items) { + const byStatus = {}; + items.forEach((item) => { + byStatus[item.status] = (byStatus[item.status] || 0) + 1; + }); + $('escalationsSummary').innerHTML = [ + renderSummaryCard('Активных', (byStatus.requested || 0) + (byStatus.ringing || 0), 'Ждут/дозваниваются до оператора'), + renderSummaryCard('Завершено', byStatus.completed || 0, 'Успешно приняты оператором'), + renderSummaryCard('Не удалось', byStatus.failed || 0, 'Свободного оператора не нашлось'), + ].join(''); + + if (!items.length) { + $('escalationsList').innerHTML = '
  • Эскалаций пока не было.
    • '; + return; + } + + $('escalationsList').innerHTML = items + .map( + (item) => ` +
  • + звонок ${escapeHtml(item.call_id)} + ${statusBadge(ESCALATION_STATUS_LABELS, ESCALATION_STATUS_TONE, item.status)} + ${escapeHtml(item.from_level)} → ${escapeHtml(item.to_level)} + причина: ${escapeHtml(item.reason_code || '-')} + оператор: ${escapeHtml(item.assigned_agent_id || 'не назначен')} + запрошена: ${formatTime(item.requested_at)} + ${item.summary ? `тема: ${escapeHtml(item.summary)}` : ''} +
    • + `, + ) + .join(''); +} + +async function loadEscalations(silent = false) { + try { + const data = await api('routing', 'escalations?limit=30'); + renderEscalations(Array.isArray(data) ? data : []); + if (!silent) { + log('Лента эскалаций обновлена', { count: data.length }); + } + } catch (err) { + $('escalationsList').innerHTML = '
  • Не удалось загрузить эскалации.
    • '; + if (!silent) { + log('Не удалось загрузить эскалации', { error: err.message }); + } + } +} + async function saveQueueMetrics() { try { const queueId = encodeURIComponent($('queueMetricId').value.trim() || 'line2'); @@ -973,7 +1145,16 @@ function wire() { $('loginBtn').addEventListener('click', login); $('corporateLoginBtn').addEventListener('click', startCorporateLogin); $('refreshBtn').addEventListener('click', async () => { - await Promise.all([loadRealtime(), loadKpi(), loadLiveCalls(), loadAgents(), loadRecordings(), refreshKnowledgeBase(true)]); + await Promise.all([ + loadRealtime(), + loadKpi(), + loadLiveCalls(), + loadAgents(), + loadCallAgents(), + loadEscalations(), + loadRecordings(), + refreshKnowledgeBase(true), + ]); log('Данные супервизора обновлены'); }); $('loadRealtimeBtn').addEventListener('click', loadRealtime); @@ -981,6 +1162,9 @@ function wire() { $('loadLiveCallsBtn').addEventListener('click', () => loadLiveCalls()); $('loadAgentsBtn').addEventListener('click', loadAgents); $('setAgentBtn').addEventListener('click', setAgentState); + $('loadCallAgentsBtn').addEventListener('click', () => loadCallAgents()); + $('createCallAgentBtn').addEventListener('click', createCallAgent); + $('loadEscalationsBtn').addEventListener('click', () => loadEscalations()); $('saveQueueMetricsBtn').addEventListener('click', saveQueueMetrics); $('createKbCategoryBtn').addEventListener('click', createKnowledgeCategory); $('loadKbBtn').addEventListener('click', () => refreshKnowledgeBase()); @@ -1015,11 +1199,15 @@ async function init() { loadKpi(), loadLiveCalls(true), loadAgents(), + loadCallAgents(true), + loadEscalations(true), loadRecordings(), refreshKnowledgeBase(true), ]); window.setInterval(() => { void loadLiveCalls(true); + void loadCallAgents(true); + void loadEscalations(true); }, 5000); log('Пульт супервизора готов'); } diff --git a/ui/supervisor/index.html b/ui/supervisor/index.html index 7531572..94b117c 100644 --- a/ui/supervisor/index.html +++ b/ui/supervisor/index.html @@ -35,6 +35,8 @@ Оперативный срез Живые звонки Агенты + Пул L2 (звонки) + Эскалации Очереди База знаний @@ -153,6 +155,45 @@
      +
      +

      Пул операторов L2 (звонки)

      +

      Реальный пул для эскалации звонков с Voice AI (L1) на живых операторов +7 747 645 6048. Отдельно от демо-статусов в блоке «Агенты» выше — здесь показан фактический пул Routing Engine.

      +
      + +
      +
      +
        + +
        + Добавить оператора в пул +
        + + +
        +
        + + +
        +
        + + +
        +
        +
        + +
        +

        Эскалации AI → оператор

        +

        Живая лента: какие звонки Voice AI передал(-ает) на L2 и кто из операторов их принял.

        +
        + +
        +
        +
          +
          +

          Очередь