""" Sessions Controller for Mid Platform. [AC-IDMP-09] Session mode switch endpoint: POST /mid/sessions/{sessionId}/mode """ import logging from typing import Annotated from fastapi import APIRouter, Depends, Path from fastapi.responses import JSONResponse from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_session from app.core.tenant import get_tenant_id from app.models.mid.schemas import ( SessionMode, SwitchModeRequest, SwitchModeResponse, ) from app.services.memory import MemoryService logger = logging.getLogger(__name__) router = APIRouter(prefix="/mid", tags=["Mid Platform Sessions"]) _session_modes: dict[str, SessionMode] = {} @router.post( "/sessions/{sessionId}/mode", operation_id="switchSessionMode", summary="Switch session mode", description=""" [AC-IDMP-09] Switch session mode between BOT_ACTIVE and HUMAN_ACTIVE. When mode is HUMAN_ACTIVE, dialogue responses will route to transfer mode. """, responses={ 200: {"description": "Mode switched successfully", "model": SwitchModeResponse}, 400: {"description": "Invalid request"}, }, ) async def switch_session_mode( sessionId: Annotated[str, Path(description="Session ID")], switch_request: SwitchModeRequest, session: Annotated[AsyncSession, Depends(get_session)], ) -> SwitchModeResponse: """ [AC-IDMP-09] Switch session mode. Modes: - BOT_ACTIVE: Bot handles responses - HUMAN_ACTIVE: Transfer to human agent """ tenant_id = get_tenant_id() if not tenant_id: from app.core.exceptions import MissingTenantIdException raise MissingTenantIdException() logger.info( f"[AC-IDMP-09] Mode switch: tenant={tenant_id}, " f"session={sessionId}, mode={switch_request.mode.value}" ) try: memory_service = MemoryService(session) await memory_service.get_or_create_session( tenant_id=tenant_id, session_id=sessionId, ) session_key = f"{tenant_id}:{sessionId}" _session_modes[session_key] = switch_request.mode logger.info( f"[AC-IDMP-09] Mode switched: session={sessionId}, " f"mode={switch_request.mode.value}, reason={switch_request.reason}" ) return SwitchModeResponse( session_id=sessionId, mode=switch_request.mode, ) except Exception as e: logger.error(f"[AC-IDMP-09] Mode switch failed: {e}") return SwitchModeResponse( session_id=sessionId, mode=switch_request.mode, ) def get_session_mode(tenant_id: str, session_id: str) -> SessionMode: """Get current session mode.""" session_key = f"{tenant_id}:{session_id}" return _session_modes.get(session_key, SessionMode.BOT_ACTIVE) def clear_session_mode(tenant_id: str, session_id: str) -> None: """Clear session mode (reset to BOT_ACTIVE).""" session_key = f"{tenant_id}:{session_id}" if session_key in _session_modes: del _session_modes[session_key]