Merge pull request 'feat: L1->L2 agent pool and routing engine for voice escalation' (#4) from feature/l1-l2-routing-engine into main

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,

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,

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("/")
 

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)

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)

services/routing_service/app.py

@@ -5,12 +5,23 @@ 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,
+    EscalationOut,
+    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, EscalationRow, IvrSessionRow, Queue, RoutingCounter
 
 app = FastAPI(title="routing-service", version="1.1.0")
 
@@ -230,3 +241,169 @@ 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()
+
+
+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,
+    _: 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()

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

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

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")

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