lct-hack/backend/app/api/ws/control.py

"""Канал преподавателя: только передача.

**Ни одной команды, меняющей карточку курсанта.** Преподаватель управляет
ситуацией, а не работой обучаемого, иначе оценка перестаёт быть оценкой
курсанта (docs/arch/CONTRACT.md).

Ответы сюда не идут — канал односторонний. Всё, что сервер хочет сказать
преподавателю, уходит на его же сокет `observe`.
"""

import asyncio
import logging
import math
import secrets
from uuid import UUID

from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from pydantic import TypeAdapter, ValidationError

from app.api.ws.session import close_fenced, run_command, session_socket
from app.config import get_settings
from app.db.base import get_sessionmaker
from app.db.repo import SessionNodeConflict
from app.dialog.director import apply as apply_directive
from app.dialog.director import mood_of
from app.dialog.factory import build_caller
from app.dialog.persona import PersonaState
from app.dialog.runtime import get_embedder
from app.dialog.slots import SlotMachine
from app.domain.classifiers import Outcome
from app.domain.events import (
    CallEnded,
    CallEndReason,
    CallIncoming,
    CardBriefing,
    ErrorEvent,
    ErrorKind,
    Exercise,
    InstructorNoteShown,
    InstructorToServer,
    ModeSet,
    ReferenceStarted,
    SessionEnded,
    SessionPaused,
    StationState,
)
from app.domain.roles import Role
from app.domain.timers import TimerCode
from app.scenarios import store
from app.session.dds import prepare_queue
from app.session.access import can_access
from app.session.hub import hub
from app.session.finish import end_session, override_score
from app.session.state import SessionState, now_utc
from app.session.store import (
    LessonIdentity,
    LessonPaused,
    LessonRequest,
    LessonResumed,
    NoteAdded,
    ScoreOverridden,
)
from app.voice.models import get_voice_models
from app.voice.pipeline import FILLERS, prefetch

log = logging.getLogger(__name__)
router = APIRouter()

_adapter = TypeAdapter(InstructorToServer)

# Пульт может молчать; без входящих кадров потеря lease замечается этим опросом.
_FENCE_POLL_SECONDS = 1.0


def _dds_ineligible_scenarios(scenarios):
    """Консультация и передача региона не являются готовыми карточками ДДС."""
    return [scenario for scenario in scenarios if scenario.outcome is not Outcome.CARD]


def card_briefing(state: SessionState) -> CardBriefing:
    """Первую реплику показывает курсант; факты раскрываются только в ответах."""
    scenario = state.scenario
    return CardBriefing(
        scenario_id=scenario.id, mode=state.mode, text=scenario.first_line,
        required_fields=([field for field in state.required_fields if field != "dds"]
                         if scenario.ground_truth.incident_code else list(state.required_fields)),
        card=state.kio,
        handoff_to_dds=state.handoff_to_dds,
    )


def _build_state(session_id: UUID, event, who, scenario, scenarios,
                 identity: LessonIdentity) -> SessionState:
    """Занятие собирается целиком и только потом регистрируется: иначе
    наблюдатель, подключившийся в эту щель, увидит полусобранное состояние
    без слот-автомата и звонящего. Снимок пишется в той же транзакции, что
    и строка занятия."""
    state = SessionState(
        session_id=session_id,
        scenario_id=scenario.id,
        scenario_title=scenario.title,
        level=scenario.level.value,
        mode=event.mode,
        owner_login=who.login if who is not None else None,
        backend_fencing_epoch=identity.fencing_epoch,
        exercise=event.exercise,
        handoff_to_dds=event.handoff_to_dds,
        scenario=scenario.model_copy(deep=True),
        required_fields=scenario.required_fields,
        trainee_name=event.trainee,
        trainee_id=identity.trainee_id,
        dds_service=identity.service or event.dds_service,
        attempt=identity.attempt,
        criteria=event.criteria,
    )
    state.timers.limits[TimerCode.DDS_ACK] = event.criteria.decision_time_limit_seconds * 1000
    state.timers.limits[TimerCode.CARD_FILL] = event.criteria.card_fill_time_limit_seconds * 1000
    state.timers.limits[TimerCode.DDS_WORK] = event.criteria.dds_card_work_time_limit_seconds * 1000
    if event.exercise is Exercise.CALL:
        embedder = get_embedder()
        if embedder is not None:
            state.slots = SlotMachine(state.scenario, embedder)
        state.persona = PersonaState(state.scenario.persona)
        state.caller = build_caller(
            scenario.id,
            use_pregenerated=scenario.tree.pregenerated,
        )
    elif event.exercise is Exercise.DDS:
        prepare_queue(
            state,
            [item.model_copy(deep=True) for item in scenarios],
            arrival_interval_seconds=event.dds_arrival_interval_seconds,
            max_waiting=event.dds_max_waiting,
        )
        state.started_at = state.desk.active.dispatched_at if state.desk.active else None
    else:
        state.started_at = now_utc()
        if event.exercise is Exercise.CARD:
            embedder = get_embedder()
            if embedder is not None:
                state.slots = SlotMachine(state.scenario, embedder)
            state.persona = PersonaState(state.scenario.persona)
            state.caller = build_caller(
                scenario.id, use_pregenerated=scenario.tree.pregenerated,
            )
            state.on_event("card.start")
        if event.handoff_to_dds:
            state.pending_dds_scenarios = [item.model_copy(deep=True) for item in scenarios[1:]]
            state.desk.arrival_interval_seconds = event.dds_arrival_interval_seconds
            state.desk.max_waiting = event.dds_max_waiting
    if event.exercise is Exercise.CALL:
        state.on_event("call.incoming")
    return state


async def _start(session_id: UUID, event, who=None) -> None:
    if event.handoff_to_dds and event.exercise is not Exercise.CARD:
        hub.to_observers(session_id, ErrorEvent(
            code=ErrorKind.SCENARIO_INVALID,
            message="Передача в ДДС доступна только для текстовой карточки 112",
        ))
        return
    scenario_ids = list(dict.fromkeys([
        event.scenario_id, *(event.scenario_ids or []), *(event.random_scenario_ids or []),
    ]))
    if get_settings().demo_no_db or not hub.store.persistent:
        catalog = {
            scenario_id: scenario for scenario_id in scenario_ids
            if (scenario := store.get(scenario_id)) is not None
        }
        hidden_scenario_ids = (
            await store.scenario_ids_owned_by_other(None, who.login)
            if who is not None else set()
        )
    else:
        try:
            async with get_sessionmaker()() as db:
                catalog, hidden_scenario_ids = await store.published_catalog(
                    db, scenario_ids, who.login if who is not None else None,
                )
        except Exception as exc:
            # Avoid serializing scenario facts or SQL bind values into application logs.
            log.error("не удалось разрешить сценарий из общей библиотеки (%s)", type(exc).__name__)
            hub.to_observers(session_id, ErrorEvent(
                code=ErrorKind.INTERNAL,
                message="Не удалось проверить сценарий в общей библиотеке; запуск отменён.",
            ))
            return
    scenario = catalog.get(event.scenario_id)
    if scenario is None:
        hub.to_observers(
            session_id,
            ErrorEvent(code=ErrorKind.SCENARIO_INVALID, message=f"Нет сценария {event.scenario_id}"),
        )
        return
    if scenario.id in hidden_scenario_ids:
        hub.to_observers(session_id, ErrorEvent(
            code=ErrorKind.SCENARIO_INVALID,
            message="Сценарий не найден или недоступен этому преподавателю",
        ))
        return

    if event.random_scenario_ids:
        if len(event.random_scenario_ids) > 96:
            hub.to_observers(session_id, ErrorEvent(
                code=ErrorKind.SCENARIO_INVALID,
                message="Случайный отбор ограничен 96 карточками",
            ))
            return
        pool_ids = list(dict.fromkeys(event.random_scenario_ids))
        pool = [catalog.get(scenario_id) for scenario_id in pool_ids]
        if any(item is None for item in pool):
            hub.to_observers(session_id, ErrorEvent(
                code=ErrorKind.SCENARIO_INVALID,
                message="В случайном отборе есть неизвестный сценарий",
            ))
            return
        if hidden_scenario_ids.intersection(pool_ids):
            hub.to_observers(session_id, ErrorEvent(
                code=ErrorKind.SCENARIO_INVALID,
                message="Сценарий не найден или недоступен этому преподавателю",
            ))
            return
        ineligible_pool = _dds_ineligible_scenarios(pool)
        if ineligible_pool:
            hub.to_observers(session_id, ErrorEvent(
                code=ErrorKind.SCENARIO_INVALID,
                message="Случайный отбор должен содержать только готовые карточки ДДС",
            ))
            return
        scenario = secrets.choice(pool)

    scenario_ids = event.scenario_ids or [event.scenario_id]
    if event.exercise is Exercise.DDS or event.handoff_to_dds:
        if (not scenario_ids or scenario_ids[0] != event.scenario_id
                or len(scenario_ids) > 96):
            hub.to_observers(session_id, ErrorEvent(
                code=ErrorKind.SCENARIO_INVALID,
                message="Очередь ДДС должна начинаться с scenario_id и содержать не более 96 карточек",
            ))
            return
        if event.random_scenario_ids:
            extra_ids = list(dict.fromkeys(
                item for item in scenario_ids[1:] if item != scenario.id
            ))
            extras = [catalog.get(item) for item in extra_ids]
            remaining_random = [item for item in pool if item.id != scenario.id
                                and item.id not in extra_ids]
            randomized_tail = secrets.SystemRandom().sample(
                remaining_random, k=len(remaining_random)
            )
            scenarios = [scenario, *extras, *randomized_tail]
        else:
            scenarios = [catalog.get(scenario_id) for scenario_id in scenario_ids]
        if len(scenarios) > 96:
            hub.to_observers(session_id, ErrorEvent(
                code=ErrorKind.SCENARIO_INVALID,
                message="Очередь ДДС не может содержать более 96 карточек",
            ))
            return
        if any(item is None for item in scenarios):
            hub.to_observers(session_id, ErrorEvent(
                code=ErrorKind.SCENARIO_INVALID, message="В очереди ДДС есть неизвестный сценарий",
            ))
            return
        if any(item.id in hidden_scenario_ids for item in scenarios):
            hub.to_observers(session_id, ErrorEvent(
                code=ErrorKind.SCENARIO_INVALID,
                message="Сценарий не найден или недоступен этому преподавателю",
            ))
            return
        ineligible = _dds_ineligible_scenarios(scenarios)
        if ineligible:
            titles = ", ".join(item.title for item in ineligible)
            hub.to_observers(session_id, ErrorEvent(
                code=ErrorKind.SCENARIO_INVALID,
                message=("В очередь ДДС можно добавить только готовые карточки с исходом "
                         f"«карточка и передача в ДДС». Исключите: {titles}"),
            ))
            return
    else:
        scenarios = []

    try:
        state = await hub.store.open(LessonRequest(
            session_id=session_id, scenario_id=scenario.id, mode=event.mode.value,
            trainee_name=event.trainee, trainee_id=event.trainee_id,
            owner_login=who.login if who is not None else None,
            backend_node_id=get_settings().backend_node_id,
        ), lambda identity: _build_state(session_id, event, who, scenario, scenarios, identity))
    except SessionNodeConflict:
        hub.to_observers(session_id, ErrorEvent(
            code=ErrorKind.FORBIDDEN,
            message="Сессия закреплена за другим backend-узлом; проверьте маршрутизацию proxy",
        ))
        return
    except PermissionError:
        hub.to_observers(session_id, ErrorEvent(
            code=ErrorKind.FORBIDDEN,
            message="Занятие создано другим преподавателем",
        ))
        return
    except Exception as exc:  # noqa: BLE001 — без долговечной истории занятие не стартует
        log.error("занятие %s не записано (%s)", session_id, type(exc).__name__)
        hub.to_observers(session_id, ErrorEvent(
            code=ErrorKind.INTERNAL,
            message="Не удалось записать занятие и аудит; запуск отменён.",
        ))
        return
    hub.register(state)

    # Первая реплика и филлеры синтезируются, пока курсант не снял трубку:
    # «Алло! Помогите!» должно прозвучать мгновенно (docs/arch/BACKEND.md).
    if event.exercise is Exercise.CALL:
        models = get_voice_models()
        if models is not None:
            asyncio.create_task(prefetch(models, [scenario.first_line, *FILLERS.values()]))
    hub.start_ticker(session_id)
    if event.exercise is Exercise.CALL:
        hub.to_trainee(
            session_id,
            CallIncoming(
                scenario_id=scenario.id,
                caller_number="+7 (495) 000-00-00",
                level=scenario.level,
                mode=event.mode,
                required_fields=[field for field in scenario.required_fields
                                 if field != "dds" or not scenario.ground_truth.incident_code],
            ),
        )
    elif event.exercise is Exercise.DDS:
        hub.to_station(session_id, state.card_received_event())
        hub.to_station(session_id, StationState(snapshot=state.station_snapshot()))
    else:
        hub.to_trainee(session_id, card_briefing(state))
    hub.to_observers(session_id, ModeSet(mode=event.mode))
    hub.to_observers(session_id, state.snapshot())


async def _stop(session_id: UUID) -> None:
    state = hub.get(session_id)
    if state is None:
        return
    await end_session(session_id, state, CallEndReason.INSTRUCTOR)


async def _command(session_id: UUID, event, who) -> None:
    """Команда пульта. Всё, кроме запуска, идёт внутри `hub.operation`."""
    state = hub.get(session_id)
    match event.type:
        case "scenario.start":
            await _start(session_id, event, who)
        case "session.stop":
            await _stop(session_id)
        case "session.pause":
            if state is None or state.ended or state.paused:
                return
            state.pause()
            hub.record(session_id, LessonPaused(
                at=state.paused_at, author=who.login, role=who.role.value,
            ))
            hub.broadcast(session_id, SessionPaused(paused=True))
            hub.to_station(session_id, SessionPaused(paused=True))
        case "session.resume":
            if state is None or state.ended or not state.paused:
                return
            paused_ms_before = state.total_paused_ms
            state.resume()
            hub.record(session_id, LessonResumed(
                at=now_utc(), author=who.login, role=who.role.value,
                paused_ms=state.total_paused_ms - paused_ms_before,
            ))
            hub.broadcast(session_id, SessionPaused(paused=False))
            hub.to_station(session_id, SessionPaused(paused=False))
        case "instructor_note.add":
            if state is not None:
                state.notes.append({
                    "type": "instructor_note.shown",
                    "transcript_ref": event.transcript_ref,
                    "text": event.text,
                    "author": "преподаватель",
                })
                hub.record(session_id, NoteAdded(event.transcript_ref, event.text, "преподаватель"))
            hub.to_observers(
                session_id,
                InstructorNoteShown(
                    transcript_ref=event.transcript_ref,
                    text=event.text,
                    author="преподаватель",
                ),
            )
        case "reference.play":
            if state is not None:
                hub.to_observers(session_id, ReferenceStarted(scenario_id=state.scenario_id))
        case "score.override":
            if event.session_id != session_id:
                hub.to_observers(session_id, ErrorEvent(
                    code=ErrorKind.FORBIDDEN,
                    message="Оценка относится к другому занятию",
                ))
                return
            if state is None or state.score is None:
                return
            try:
                verdict = float(event.verdict)
            except ValueError:
                verdict = float("nan")
            if not math.isfinite(verdict) or not 0 <= verdict <= 100:
                hub.to_observers(session_id, ErrorEvent(
                    code=ErrorKind.UNSUPPORTED_EVENT,
                    message="Оценка должна быть числом от 0 до 100",
                ))
                return
            override_score(state, ScoreOverridden(
                score_final=verdict, author=who.login, role=who.role.value,
                comment=event.comment,
            ))
        case "director.inject":
            if state is None:
                return
            if state.exercise is not Exercise.CALL:
                hub.to_observers(session_id, ErrorEvent(
                    code=ErrorKind.UNSUPPORTED_EVENT,
                    message="Директивы звонящему доступны только в голосовом упражнении",
                ))
                return
            result = apply_directive(state, event.directive)
            if result.needs_network:
                hub.to_observers(session_id, ErrorEvent(
                    code=ErrorKind.DIRECTIVE_NEEDS_NETWORK,
                    message="Свободный текст требует локально запущенной модели; без неё доступны кнопки",
                ))
                return
            state.directives.append(event.directive)
            voice = state.voice
            if result.drop_line and voice is not None:
                # Обрыв рвёт звук на полуслове тем же механизмом, что
                # перебивание, и запускает норматив обратного дозвона.
                voice.barge_in()
            if result.drop_line:
                state.on_event("call.dropped")
                hub.to_trainee(session_id, CallEnded(reason=CallEndReason.DROPPED))
                hub.to_observers(session_id, SessionEnded(reason=CallEndReason.DROPPED))
            elif result.say and voice is not None:
                voice.speak(result.say, mood_of(state))
        case _:
            hub.to_observers(
                session_id,
                ErrorEvent(
                    code=ErrorKind.UNSUPPORTED_EVENT,
                    message=f"{event.type} ещё не реализовано",
                ),
            )


@router.websocket("/ws/control/{session_id}")
async def control(ws: WebSocket, session_id: UUID) -> None:
    # Пульт преподавателя: управление занятием доступно только ему. Занятия
    # до `scenario.start` ещё нет — доступ проверяется на каждой команде.
    entered = await session_socket(ws, session_id, (Role.INSTRUCTOR,), require_state=False)
    if entered is None:
        return
    who, _state = entered
    # Чтение живёт дольше тика опроса: wait_for отменял бы его, а отмена после
    # того, как receive уже забрал кадр, теряет команду или websocket.disconnect.
    # Команда обрабатывается здесь, одной операцией хранилища.
    read: asyncio.Future | None = None
    try:
        while True:
            if read is None:
                read = asyncio.ensure_future(ws.receive_json())
            done, _ = await asyncio.wait({read}, timeout=_FENCE_POLL_SECONDS)
            if not done:
                if hub.is_lease_fenced(session_id):
                    await close_fenced(ws)
                    return
                continue
            payload, read = read.result(), None
            try:
                event = _adapter.validate_python(payload)
            except ValidationError:
                hub.to_observers(
                    session_id,
                    ErrorEvent(code=ErrorKind.UNSUPPORTED_EVENT, message=str(payload)[:200]),
                )
                continue

            existing_state = hub.get(session_id)
            if event.type == "scenario.start":
                if existing_state is not None:
                    hub.to_observers(session_id, ErrorEvent(
                        code=ErrorKind.FORBIDDEN,
                        message="Для нового занятия создайте новый номер сессии",
                    ))
                    continue
                # Запуск пишется своей транзакцией в `store.open`: до неё занятия нет.
                await _command(session_id, event, who)
                continue
            if existing_state is None or not can_access(who, existing_state):
                hub.to_observers(session_id, ErrorEvent(
                    code=ErrorKind.FORBIDDEN,
                    message="Это занятие принадлежит другому преподавателю",
                ))
                continue
            if not await run_command(ws, session_id, lambda: _command(session_id, event, who)):
                return
    except WebSocketDisconnect:
        return
    finally:
        if read is not None:
            read.cancel()