diff --git a/backend/app/api/http/sessions.py b/backend/app/api/http/sessions.py index 640bf29..02ff444 100644 --- a/backend/app/api/http/sessions.py +++ b/backend/app/api/http/sessions.py @@ -20,7 +20,7 @@ from app.api.auth import add_audit_entry, audit, audit_required, require from app.config import get_settings from app.db import repo from app.db.base import get_session -from app.db.models import AuditLog, Group, Score, Session, Trainee +from app.db.models import Group, Score, Session, Trainee from app.domain.events import Exercise, SessionMode, SessionReport from app.domain.roles import Role from app.domain.statuses import SERVICE_STATUS_LABELS, StationSnapshot, current @@ -28,8 +28,9 @@ from app.domain.timers import TimerCode from app.scoring.export import to_csv, to_pdf from app.scoring.report import build as build_report from app.session.checkpoint import load_state -from app.session.finish import scoring_scenario +from app.session.finish import override_score, scoring_scenario from app.session.hub import hub +from app.session.store import ScoreOverridden, apply_score_override from app.voice.recording import recording_path router = APIRouter(prefix="/api/sessions", tags=["sessions"]) @@ -424,14 +425,14 @@ async def _report_data( raise HTTPException(status_code=409, detail="self_assessment_required") if state.score is None: raise HTTPException(status_code=409, detail="score_not_ready") - if hub.journal is not None and isinstance(db, AsyncSession): + if hub.store.persistent and isinstance(db, AsyncSession): persisted_session = await db.scalar( select(Session.id).where(Session.id == session_id) ) if (persisted_session is not None and await db.scalar( select(Score.session_id).where(Score.session_id == session_id) ) is None): - # Live state is populated just before the journal transaction commits. + # Live state is populated just before the store commit finishes. # Do not expose a report that looks ready but cannot yet be corrected # or retrieved after restart. raise HTTPException(status_code=409, detail="score_not_ready") @@ -547,85 +548,47 @@ async def override( session_id: UUID, body: ScoreOverride, request: Request, - db: AsyncSession = Depends(get_session), + db: AsyncSession | None = Depends(optional_session), ) -> SessionReport: """Тренажёр готовит материал, преподаватель имеет последнее слово. Администратору сюда нельзя: ТЗ запрещает ему вмешиваться в оценки прямо. + Правка идёт тем же путём, что команда `score.override` с пульта. """ who = require(request, Role.INSTRUCTOR) - session = await repo.get_session(db, session_id) - if session is None: - raise HTTPException(status_code=404, detail="session_not_found") - if session.owner_login != who.login: - raise HTTPException(status_code=404, detail="session_not_found") + correction = ScoreOverridden( + score_final=body.score_final, author=who.login, role=who.role.value, + comment=body.comment, + ) state = hub.get(session_id) - live_ready = state is not None and state.score is not None - scenario = scoring_scenario(state) if live_ready else None - if live_ready and scenario is None: - raise HTTPException(status_code=409, detail="scenario_not_found") + if state is not None: + if state.owner_login != who.login: + raise HTTPException(status_code=404, detail="session_not_found") + if state.score is None: + raise HTTPException(status_code=409, detail="score_not_ready") + scenario = scoring_scenario(state) + if scenario is None: + raise HTTPException(status_code=409, detail="scenario_not_found") + async with hub.operation(session_id): + override_score(state, correction) + return build_report(session_id, state, scenario) + + # Занятия нет в памяти узла: правка из истории, по архивному разбору. + if db is None: + raise HTTPException(status_code=404, detail="session_not_found") + session = await repo.get_session(db, session_id) + if session is None or session.owner_login != who.login: + raise HTTPException(status_code=404, detail="session_not_found") score = await db.scalar(select(Score).where(Score.session_id == session_id)) if score is None: raise HTTPException(status_code=409, detail="score_not_ready") archived = (score.report or {}).get("full_report") - if not live_ready and archived is None: + if archived is None: raise HTTPException(status_code=409, detail="report_not_archived") - - # Persist score and audit entry in the same request transaction: the API - # must not report success if either durable record failed to commit. - score.score_final = body.score_final - score.overridden_by = who.login - score.override_comment = body.comment - report_payload = dict(score.report or {}) - if archived is not None: - archived_payload = dict(archived) - archived_payload.update({ - "score_auto": score.score_auto, - "score_final": body.score_final, - "overridden_by": who.login, - "override_comment": body.comment, - }) - report_payload["full_report"] = archived_payload - score.report = report_payload - db.add(AuditLog( - actor=who.login, - role=who.role.value, - action="score.override", - object_id=str(session_id), - # The actual reason remains attached to the instructor-facing score - # report. The durable security audit needs the change and actor, not - # a second indefinite copy of free-text that may contain personal data. - detail=(f"{score.score_auto} → {body.score_final}; " - f"comment_chars={len(body.comment)}"), - )) - await db.commit() - - if live_ready: - assert state is not None and state.score is not None and scenario is not None - state.score = { - **state.score, - "score_final": body.score_final, - "overridden_by": who.login, - "override_comment": body.comment, - } - state.score["full_report"] = { - **state.score.get("full_report", {}), - "score_auto": score.score_auto, - "score_final": body.score_final, - "overridden_by": who.login, - "override_comment": body.comment, - } - result = build_report(session_id, state, scenario) - else: - # Historical/archived sessions no longer have a live object in the hub. - # Keep score correction available from the instructor's report history. - result = SessionReport.model_validate(archived).model_copy(update={ - "score_auto": score.score_auto, - "score_final": body.score_final, - "overridden_by": who.login, - "override_comment": body.comment, - }) - return result + report = apply_score_override(dict(score.report), correction) + await db.rollback() # чтение закончено; запись — одной транзакцией хранилища + await hub.store.commit_archived(session_id, [correction]) + return SessionReport.model_validate(report["full_report"]) @router.get("", response_model=list[SessionOut]) diff --git a/backend/app/api/ws/call.py b/backend/app/api/ws/call.py index f3e3fbf..cf375f2 100644 --- a/backend/app/api/ws/call.py +++ b/backend/app/api/ws/call.py @@ -47,6 +47,13 @@ from app.session.dds import prepare_handoff_queue from app.session.finish import finish, refresh_archived_report, release_score from app.session.hub import LEASE_FENCED_MESSAGE, hub from app.session.state import now_utc +from app.session.store import ( + HintRecorded, + LessonEnded, + LessonStarted, + SelfAssessed, + UtteranceAppended, +) from app.voice.models import TTS_RATE, get_voice_models from app.voice.pipeline import VoiceSession from app.voice.recording import start_recording @@ -213,8 +220,7 @@ async def _handle(session_id: UUID, state, event) -> None: accepted = TextTurnAccepted(text=event.text, at=operator_entry.at) hub.to_trainee(session_id, accepted) hub.to_observers(session_id, TranscriptAppend(entry=operator_entry)) - if hub.journal: - await hub.journal.utterance(session_id, operator_entry) + hub.record(session_id, UtteranceAppended(operator_entry)) try: slots = state.slots if state.slots is not None else _TextSlotView(state) line = await state.caller.reply(turn, state.persona, slots) @@ -232,8 +238,7 @@ async def _handle(session_id: UUID, state, event) -> None: at=caller_entry.at, mood=line.mood, source=line.source, )) hub.to_observers(session_id, TranscriptAppend(entry=caller_entry)) - if hub.journal: - await hub.journal.utterance(session_id, caller_entry) + hub.record(session_id, UtteranceAppended(caller_entry)) case "card.submit": state.on_event("card.submit") state.kio.registered_at = state.started_at or now_utc() @@ -260,18 +265,14 @@ async def _handle(session_id: UUID, state, event) -> None: hub.to_observers(session_id, state.snapshot()) else: hub.to_observers(session_id, SessionEnded(reason=CallEndReason.COMPLETE)) - if hub.journal: - await hub.journal.session_ended( - session_id, state.ended_at, CallEndReason.COMPLETE.value - ) + hub.record(session_id, LessonEnded(state.ended_at, CallEndReason.COMPLETE.value)) await finish(session_id, state) case "call.answer": first_answer = state.started_at is None if first_answer: state.on_event("call.answer") state.started_at = now_utc() - if hub.journal: - await hub.journal.session_started(session_id, state.started_at) + hub.record(session_id, LessonStarted(state.started_at)) hub.to_trainee(session_id, CallStarted(started_at=state.started_at)) hub.to_observers(session_id, state.snapshot()) if state.recorder is None: @@ -313,8 +314,7 @@ async def _handle(session_id: UUID, state, event) -> None: state.hints_log.append((checklist_id, now_utc())) shown = HintShown(checklist_id=checklist_id, question=question) hub.broadcast(session_id, shown) - if hub.journal: - await hub.journal.hint(session_id, checklist_id, question, now_utc()) + hub.record(session_id, HintRecorded(checklist_id, question, now_utc())) case "dds.dispatch": if event.service is None and not (state.kio.incident_code or state.kio.notify): @@ -354,14 +354,7 @@ async def _handle(session_id: UUID, state, event) -> None: state.on_event("callback.dial") case "self_assessment.submit": - if hub.journal and not await hub.journal.self_assessment( - session_id, event.missed, event.comment, now_utc() - ): - hub.to_trainee(session_id, ErrorEvent( - code=ErrorKind.INTERNAL, - message="Не удалось сохранить самооценку и аудит; итог пока не выдан.", - )) - return + hub.record(session_id, SelfAssessed(list(event.missed), event.comment, now_utc())) state.self_assessed = True state.self_assessment = {"missed": event.missed, "comment": event.comment} await refresh_archived_report(session_id, state) @@ -376,14 +369,9 @@ async def _handle(session_id: UUID, state, event) -> None: hub.stop_ticker(session_id) hub.to_trainee(session_id, CallEnded(reason=CallEndReason.HANGUP)) hub.to_observers(session_id, SessionEnded(reason=CallEndReason.HANGUP)) - if hub.journal: - await hub.journal.session_ended( - session_id, state.ended_at, CallEndReason.HANGUP.value - ) + hub.record(session_id, LessonEnded(state.ended_at, CallEndReason.HANGUP.value)) await finish(session_id, state) - await hub.checkpoint(session_id) - def _start_voice(session_id: UUID, state, *, initial_statement: bool = True) -> None: """Голос включается, когда курсант снял трубку: звонящий сразу кричит первую реплику.""" @@ -403,7 +391,7 @@ def _start_voice(session_id: UUID, state, *, initial_statement: bool = True) -> send_event=lambda event: hub.to_trainee(session_id, event), send_observer=lambda event: hub.to_observers(session_id, event), send_audio=send_audio, - journal=hub.journal, + persist=lambda entry: hub.commit(session_id, UtteranceAppended(entry)), ) if scenario.background: event = BgStart(loop=scenario.background.loop, gain_db=scenario.background.gain_db) @@ -540,7 +528,7 @@ async def call(ws: WebSocket, session_id: UUID) -> None: ErrorEvent(code=ErrorKind.UNSUPPORTED_EVENT, message=str(payload)[:200]), ) continue - async with hub.durable_transition(session_id): + async with hub.operation(session_id): await _handle(session_id, state, event) except WebSocketDisconnect: return diff --git a/backend/app/api/ws/control.py b/backend/app/api/ws/control.py index 9cad775..bf07bb4 100644 --- a/backend/app/api/ws/control.py +++ b/backend/app/api/ws/control.py @@ -17,7 +17,7 @@ from uuid import UUID from fastapi import APIRouter, WebSocket, WebSocketDisconnect from pydantic import TypeAdapter, ValidationError -from app.api.auth import audit, principal_of, websocket_origin_allowed +from app.api.auth import principal_of, websocket_origin_allowed from app.config import get_settings from app.db.base import get_sessionmaker from app.db.repo import SessionNodeConflict @@ -49,7 +49,9 @@ from app.domain.timers import TimerCode from app.scenarios import store from app.session.dds import prepare_queue from app.session.hub import LEASE_FENCED_MESSAGE, hub +from app.session.finish import finish, override_score from app.session.state import SessionState, now_utc +from app.session.store import LessonEnded, LessonIdentity, LessonRequest, NoteAdded, ScoreOverridden from app.voice.models import get_voice_models from app.voice.pipeline import FILLERS, prefetch @@ -79,6 +81,70 @@ def card_briefing(state: SessionState) -> CardBriefing: ) +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( @@ -89,7 +155,7 @@ async def _start(session_id: UUID, event, who=None) -> None: scenario_ids = list(dict.fromkeys([ event.scenario_id, *(event.scenario_ids or []), *(event.random_scenario_ids or []), ])) - if get_settings().demo_no_db or hub.journal is None: + 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 @@ -207,96 +273,33 @@ async def _start(session_id: UUID, event, who=None) -> None: else: scenarios = [] - attempt = 1 - recorded_trainee_id = event.trainee_id - recorded_service = None - fencing_epoch = 0 - if hub.journal: - try: - persisted = await hub.journal.start_lesson( - session_id, scenario.id, event.mode.value, event.trainee, event.trainee_id, - owner_login=who.login if who is not None else None, - backend_node_id=get_settings().backend_node_id, - ) - if persisted is None: - hub.to_observers(session_id, ErrorEvent( - code=ErrorKind.INTERNAL, - message="Не удалось записать занятие и аудит; запуск отменён.", - )) - return - attempt, recorded_trainee_id, recorded_service, fencing_epoch = persisted - 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 - - # Занятие собирается целиком и только потом регистрируется: иначе - # наблюдатель, подключившийся в эту щель, увидит полусобранное состояние - # без слот-автомата и звонящего. - 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=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=recorded_trainee_id, - dds_service=recorded_service or event.dds_service, - attempt=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 + 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) - if event.exercise is not Exercise.CALL and hub.journal and state.started_at is not None: - await hub.journal.session_started(session_id, state.started_at) # Первая реплика и филлеры синтезируются, пока курсант не снял трубку: # «Алло! Помогите!» должно прозвучать мгновенно (docs/arch/BACKEND.md). @@ -304,8 +307,6 @@ async def _start(session_id: UUID, event, who=None) -> None: models = get_voice_models() if models is not None: asyncio.create_task(prefetch(models, [scenario.first_line, *FILLERS.values()])) - state.on_event("call.incoming") - await hub.checkpoint(session_id) hub.start_ticker(session_id) if event.exercise is Exercise.CALL: hub.to_trainee( @@ -340,7 +341,6 @@ async def _stop(session_id: UUID) -> None: await state.voice.close() hub.stop_ticker(session_id) hub.to_observers(session_id, SessionEnded(reason=CallEndReason.INSTRUCTOR)) - from app.session.finish import finish if state.exercise in {Exercise.DDS, Exercise.CARD}: if state.exercise is Exercise.DDS or state.handoff_to_dds and state.dispatched_card is not None: @@ -351,9 +351,97 @@ async def _stop(session_id: UUID) -> None: hub.to_station(session_id, ScoreReady(session_id=session_id)) else: hub.to_trainee(session_id, CallEnded(reason=CallEndReason.INSTRUCTOR)) + hub.record(session_id, LessonEnded(state.ended_at, CallEndReason.INSTRUCTOR.value)) await finish(session_id, state) - if hub.journal: - await hub.journal.session_ended(session_id, state.ended_at, CallEndReason.INSTRUCTOR.value) + + +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 "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} ещё не реализовано", + ), + ) async def _reject(ws: WebSocket, message: str) -> None: @@ -379,10 +467,9 @@ async def control(ws: WebSocket, session_id: UUID) -> None: if who is None or who.role not in (Role.INSTRUCTOR,): await _reject(ws, "Недостаточно прав для этого экрана") return - event_stream = hub.begin_event_stream(session_id) # Чтение живёт дольше тика опроса: wait_for отменял бы его, а отмена после # того, как receive уже забрал кадр, теряет команду или websocket.disconnect. - # Команда обрабатывается здесь, в задаче, открывшей event stream. + # Команда обрабатывается здесь, одной операцией хранилища. read: asyncio.Future | None = None try: while True: @@ -422,119 +509,13 @@ async def control(ws: WebSocket, session_id: UUID) -> None: )) continue - match event.type: - case "scenario.start": - await _start(session_id, event, who) - case "session.stop": - await _stop(session_id) - case "instructor_note.add": - state = hub.get(session_id) - if state is not None: - state.notes.append({ - "type": "instructor_note.shown", - "transcript_ref": event.transcript_ref, - "text": event.text, - "author": "преподаватель", - }) - hub.to_observers( - session_id, - InstructorNoteShown( - transcript_ref=event.transcript_ref, - text=event.text, - author="преподаватель", - ), - ) - if hub.journal: - await hub.journal.note( - session_id, event.transcript_ref, event.text, "преподаватель" - ) - case "reference.play": - state = hub.get(session_id) - 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="Оценка относится к другому занятию", - )) - continue - state = hub.get(session_id) - if state is not None and state.score is not None: - 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", - )) - continue - if hub.journal is not None: - saved = await hub.journal.score_override( - session_id, verdict, who.login, event.comment - ) - if not saved: - hub.to_observers(session_id, ErrorEvent( - code=ErrorKind.INTERNAL, - message="Не удалось сохранить оценку и запись аудита; изменение отменено", - )) - continue - else: - # Explicit in-memory demo mode has no Score table. - await audit( - who.login, who.role.value, "score.override", str(session_id), - f"{state.score.get('score_auto')} → {verdict}; " - f"comment_chars={len(event.comment)}", - ) - # Автооценка остаётся рядом: видно, что скорректировано и кем. - state.score = { - **state.score, - "score_final": verdict, - "overridden_by": who.login, - "override_comment": event.comment, - } - hub.to_observers(session_id, ScoreReady(session_id=session_id)) - case "director.inject": - state = hub.get(session_id) - if state is None: - continue - if state.exercise is not Exercise.CALL: - hub.to_observers(session_id, ErrorEvent( - code=ErrorKind.UNSUPPORTED_EVENT, - message="Директивы звонящему доступны только в голосовом упражнении", - )) - continue - result = apply_directive(state, event.directive) - if result.needs_network: - hub.to_observers(session_id, ErrorEvent( - code=ErrorKind.DIRECTIVE_NEEDS_NETWORK, - message="Свободный текст требует локально запущенной модели; без неё доступны кнопки", - )) - continue - 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} ещё не реализовано", - ), - ) + if event.type == "scenario.start": + # Запуск пишется своей транзакцией в `store.open`: до неё занятия нет. + await _command(session_id, event, who) + continue try: - await hub.checkpoint(session_id) + async with hub.operation(session_id): + await _command(session_id, event, who) except Exception: if hub.is_lease_fenced(session_id): await ws.send_text(ErrorEvent( @@ -551,4 +532,3 @@ async def control(ws: WebSocket, session_id: UUID) -> None: finally: if read is not None: read.cancel() - await hub.end_event_stream(event_stream) diff --git a/backend/app/api/ws/station.py b/backend/app/api/ws/station.py index 3c2b584..737c7ec 100644 --- a/backend/app/api/ws/station.py +++ b/backend/app/api/ws/station.py @@ -8,7 +8,7 @@ в отчёте и становится сорванным выездом с конкретной причиной. Правила пульта — в `DdsDesk.apply`; здесь разбор команды, отсев повторов, -durable transition и рассылка итога. +операция хранилища и рассылка итога. """ import asyncio @@ -32,6 +32,7 @@ from app.domain.events import ( from app.domain.roles import Role from app.session.finish import finish from app.session.hub import LEASE_FENCED_MESSAGE, hub +from app.session.store import LessonEnded log = logging.getLogger(__name__) router = APIRouter() @@ -48,8 +49,7 @@ async def _finish_dds(session_id: UUID, state) -> None: hub.stop_ticker(session_id) hub.to_station(session_id, SessionEnded(reason=CallEndReason.COMPLETE)) hub.to_observers(session_id, SessionEnded(reason=CallEndReason.COMPLETE)) - if hub.journal: - await hub.journal.session_ended(session_id, ended_at, CallEndReason.COMPLETE.value) + hub.record(session_id, LessonEnded(ended_at, CallEndReason.COMPLETE.value)) await finish(session_id, state) hub.to_station(session_id, ScoreReady(session_id=session_id)) @@ -65,7 +65,6 @@ async def _handle(session_id: UUID, state, event) -> None: await _finish_dds(session_id, state) if not outcome.changed: return - await hub.checkpoint(session_id) hub.to_station(session_id, StationState(snapshot=state.station_snapshot())) hub.to_observers(session_id, state.snapshot()) @@ -156,18 +155,13 @@ async def station(ws: WebSocket, session_id: UUID, role: str = "dds") -> None: # Re-ack it without rerunning its business transition. hub.to_station(session_id, CommandAck(command_id=command_id)) continue - async with hub.durable_transition(session_id): + async with hub.operation(session_id): await _handle(session_id, state, event) if command_id is not None: + # ID команды входит в тот же снимок, что и её переход: + # подтверждение уходит только после их общего commit. state.processed_station_commands.append(str(command_id)) del state.processed_station_commands[:-512] - # Commit the state+dedupe ID before acknowledging. The - # transition context can have already flushed other - # events; an explicit checkpoint here makes the - # command/ACK boundary independent of that batch state. - await hub.checkpoint(session_id) - # The hub stages non-error events until the checkpoint - # transaction has committed, including this ack. hub.to_station(session_id, CommandAck(command_id=command_id)) except WebSocketDisconnect: return @@ -178,7 +172,7 @@ async def station(ws: WebSocket, session_id: UUID, role: str = "dds") -> None: "закрытие станционного WebSocket после fencing занятия %s", session_id, ) - # `hub.checkpoint` broadcasts a structured fence event before + # `hub.operation` broadcasts a structured fence event before # propagating the failed database write. Let the sender deliver # that event and close with 1012 instead of an opaque 1006. await asyncio.gather(sender, return_exceptions=True) diff --git a/backend/app/db/repo.py b/backend/app/db/repo.py index aa5742d..79a9698 100644 --- a/backend/app/db/repo.py +++ b/backend/app/db/repo.py @@ -40,6 +40,7 @@ async def create_session( owner_login: str | None = None, backend_node_id: str | None = None, before_commit: BeforeSessionCommit | None = None, + commit: bool = True, ) -> Session: session = Session( scenario_id=scenario_id, @@ -56,7 +57,10 @@ async def create_session( if before_commit is not None: await db.flush() before_commit(db, session) - await db.commit() + if commit: + await db.commit() + else: + await db.flush() return session @@ -72,6 +76,7 @@ async def ensure_session( owner_login: str | None = None, backend_node_id: str | None = None, before_commit: BeforeSessionCommit | None = None, + commit: bool = True, ) -> Session: """Занятие, запущенное с пульта, должно иметь строку в журнале. @@ -98,8 +103,7 @@ async def ensure_session( changed = True if before_commit is not None: before_commit(db, existing) - await db.commit() - elif changed: + if commit and (before_commit is not None or changed): await db.commit() return existing @@ -123,6 +127,7 @@ async def ensure_session( owner_login=owner_login, backend_node_id=backend_node_id, before_commit=before_commit, + commit=commit, ) diff --git a/backend/app/main.py b/backend/app/main.py index 402ab53..d63e4d6 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -28,7 +28,8 @@ from app.monitoring import install_diagnostics from app.scenarios import store from app.scenarios.loader import ScenarioError from app.session.hub import hub -from app.session.journal import DbJournal, SessionLeaseLost +from app.session.pg_store import PostgresSessionStore +from app.session.store import MemorySessionStore, SessionLeaseLost from app.voice.models import get_voice_models LIBRARY = Path(__file__).resolve().parents[2] / "scenarios" @@ -77,7 +78,7 @@ async def lifespan(app: FastAPI): ) else: logging.getLogger(__name__).warning( - "DEMO_NO_DB: занятия и оценки живут только до перезапуска; журнал БД выключен" + "DEMO_NO_DB: занятия и оценки живут только до перезапуска; хранилище в памяти" ) if not settings.demo_no_db: @@ -89,16 +90,17 @@ async def lifespan(app: FastAPI): "не удалось загрузить версии полномочий: %s", exc ) - # Журнал: всё, что не записано, для оценки не существует. - hub.journal = ( - None if settings.demo_no_db - else DbJournal(get_sessionmaker(), node_id=settings.backend_node_id) + # Хранилище: всё, что не записано, для оценки не существует. Без БД — + # in-memory adapter, занятие живёт до перезапуска. + hub.store = ( + MemorySessionStore() if settings.demo_no_db + else PostgresSessionStore(get_sessionmaker(), node_id=settings.backend_node_id) ) app.state.sessions_restored = 0 - if hub.journal is not None: + if hub.store.persistent: try: async with asyncio.timeout(3): - restored = await hub.journal.restore_active() + restored = await hub.store.restore_active() for state in restored: hub.register(state) hub.start_ticker(state.session_id) @@ -113,18 +115,16 @@ async def lifespan(app: FastAPI): ) lease_task = None - if hub.journal is not None and settings.backend_node_id: + if hub.store.persistent and settings.backend_node_id: async def supervise_session_ownership() -> None: while True: await asyncio.sleep(5) - journal = hub.journal - if journal is None: - return + store = hub.store for state in list(hub._sessions.values()): if state.ended or state.lease_fenced: continue try: - await journal.renew(state.session_id) + await store.renew(state.session_id) except SessionLeaseLost: await hub.fence(state) except (SQLAlchemyError, OSError, TimeoutError): @@ -139,7 +139,7 @@ async def lifespan(app: FastAPI): ) await hub.fence(state) try: - restored = await journal.claim_expired() + restored = await store.claim_expired() for state in restored: current = hub._sessions.get(state.session_id) if current is not None and not current.lease_fenced: @@ -178,11 +178,11 @@ async def lifespan(app: FastAPI): with suppress(asyncio.CancelledError): await lease_task - if hub.journal is not None: + if hub.store.persistent: for state in list(hub._sessions.values()): - if not state.ended: + if not state.ended and not state.lease_fenced: try: - await hub.journal.checkpoint(state) + await hub.store.commit(state) except Exception: # noqa: BLE001 — shutdown must release the process logging.getLogger(__name__).exception( "не удалось сохранить checkpoint %s при shutdown", state.session_id diff --git a/backend/app/session/finish.py b/backend/app/session/finish.py index 719977f..952c630 100644 --- a/backend/app/session/finish.py +++ b/backend/app/session/finish.py @@ -12,7 +12,7 @@ import logging import time from uuid import UUID -from app.domain.events import ErrorKind, ErrorEvent, Exercise, Metric, ScoreReady +from app.domain.events import Exercise, Metric, ScoreReady from app.domain.statuses import ServiceStatus, current from app.domain.taxonomy import Competency, ErrorCode, Finding, FindingSource from app.domain.timers import TimerCode @@ -29,6 +29,7 @@ from app.scoring.timing import time_credit from app.scoring.weights import apply_weights from app.session.hub import hub from app.session.state import DdsCardRecord, DdsLiveCard, now_utc +from app.session.store import ScoreArchived, ScoreCalculated, ScoreOverridden, apply_score_override log = logging.getLogger(__name__) @@ -351,13 +352,9 @@ async def finish(session_id: UUID, state) -> None: state.score["full_report"] = build_report(session_id, state, scenario).model_dump(mode="json") log.info("сессия %s: оценка %.1f, отметок %d", session_id, result.score, len(result.findings)) - if hub.journal and not await hub.journal.score(session_id, result.score, state.score): - hub.to_observers(session_id, ErrorEvent( - code=ErrorKind.INTERNAL, - message="Не удалось сохранить оценку и аудит; итог не выдан. Обратитесь к преподавателю.", - )) - return - + # Оценка и аудит уходят в commit операции, завершившей занятие; ScoreReady + # ждёт того же commit — без записи итог не выдаётся. + hub.record(session_id, ScoreCalculated(result.score, state.score)) hub.to_observers(session_id, ScoreReady(session_id=session_id)) await release_score(session_id, state) @@ -370,11 +367,21 @@ async def refresh_archived_report(session_id: UUID, state) -> None: if scenario is None: return state.score["full_report"] = build_report(session_id, state, scenario).model_dump(mode="json") - if hub.journal: - await hub.journal.score_snapshot(session_id, state.score) + hub.record(session_id, ScoreArchived(state.score)) async def release_score(session_id: UUID, state) -> None: """Отдать оценку курсанту, когда самооценка сдана.""" if state.score is not None and (state.self_assessed or state.exercise is Exercise.CARD): hub.to_trainee(session_id, ScoreReady(session_id=session_id)) + + +def override_score(state, override: ScoreOverridden) -> None: + """Правка балла преподавателем — одна операция для HTTP и WS. + + Вызывается внутри `hub.operation`: строка `Score`, архивный разбор и аудит + пишутся тем же commit, что и снимок с исправленной оценкой. + """ + hub.record(state.session_id, override) + state.score = apply_score_override(state.score, override) + hub.to_observers(state.session_id, ScoreReady(session_id=state.session_id)) diff --git a/backend/app/session/hub.py b/backend/app/session/hub.py index 9e2b8b4..4b29112 100644 --- a/backend/app/session/hub.py +++ b/backend/app/session/hub.py @@ -11,7 +11,6 @@ import logging from contextvars import ContextVar, Token from collections.abc import AsyncIterator, Iterator from datetime import UTC, datetime -from typing import Protocol from uuid import UUID from pydantic import BaseModel @@ -24,6 +23,7 @@ from app.domain.events import ( TimerTick, ) from app.session.state import SessionState +from app.session.store import MemorySessionStore, Record, SessionStore #: Очередь одного подписчика. Медленный наблюдатель не тормозит занятие: #: очередь ограничена, переполнение роняет соединение, а не сессию. @@ -34,51 +34,30 @@ LEASE_FENCED_MESSAGE = "Занятие передано другому backend- log = logging.getLogger(__name__) -def _current_task(): - try: - return asyncio.current_task() - except RuntimeError: # synchronous tests and tooling have no running loop - return None +class _Operation: + """Отложенная публикация одной операции: события ждут её commit.""" - -class Journal(Protocol): - """Запись в БД. Вынесена за хаб: без базы занятие должно идти, - но молчать об ошибке записи нельзя.""" - - async def start_lesson( - self, session_id: UUID, scenario_id: str, mode: str, trainee_name: str | None, - trainee_id: UUID | None = None, owner_login: str | None = None, - backend_node_id: str | None = None, - ) -> tuple[int, UUID | None, str | None, int] | None: ... - async def utterance(self, session_id: UUID, entry) -> None: ... - async def hint(self, session_id: UUID, checklist_id: str, question: str, at) -> None: ... - async def note(self, session_id: UUID, ref: str, text: str, author: str) -> None: ... - async def self_assessment( - self, session_id: UUID, missed: list[str], comment: str, at - ) -> bool: ... - async def score(self, session_id: UUID, score_auto: float, report: dict) -> bool: ... - async def score_snapshot(self, session_id: UUID, report: dict) -> None: ... - async def score_override( - self, session_id: UUID, score_final: float, author: str, comment: str, - ) -> bool: ... - async def session_started(self, session_id: UUID, at) -> None: ... - async def session_ended(self, session_id: UUID, at, reason: str) -> None: ... - async def checkpoint(self, state: SessionState) -> None: ... - async def restore_active(self) -> list[SessionState]: ... - async def renew(self, session_id: UUID) -> None: ... - async def claim_expired(self, session_id: UUID | None = None) -> list[SessionState]: ... + def __init__(self, session_id: UUID) -> None: + self.session_id = session_id + self.records: list[Record] = [] + self.events: list[tuple[dict[UUID, set[asyncio.Queue]], BaseModel]] = [] + self.open = True + #: Такт, которому нечего сохранять, не пишет снимок каждую секунду. + self.persist = True class SessionHub: - def __init__(self, journal: Journal | None = None) -> None: - self.journal = journal + def __init__(self, store: SessionStore | None = None) -> None: + self.store: SessionStore = store if store is not None else MemorySessionStore() self._sessions: dict[UUID, SessionState] = {} self._observers: dict[UUID, set[asyncio.Queue]] = {} self._trainees: dict[UUID, set[asyncio.Queue]] = {} self._stations: dict[UUID, set[asyncio.Queue]] = {} self._tickers: dict[UUID, asyncio.Task] = {} - self._event_batch: ContextVar[dict | None] = ContextVar( - f"session-event-batch-{id(self)}", default=None + # Задачи, порождённые внутри операции, наследуют контекст; после + # закрытия операции их события идут напрямую (`_Operation.open`). + self._operation: ContextVar[_Operation | None] = ContextVar( + f"session-operation-{id(self)}", default=None ) # ── реестр ── @@ -139,84 +118,67 @@ class SessionHub: self._sessions.pop(session_id, None) self.stop_ticker(session_id) - async def checkpoint(self, session_id: UUID) -> None: - """Зафиксировать подтверждённое состояние, если журнал доступен.""" - state = self._sessions.get(session_id) - if state is not None and state.lease_fenced: - raise RuntimeError(LEASE_FENCED_MESSAGE) - if state is not None and self.journal is not None: - try: - await self.journal.checkpoint(state) - except Exception: - self._discard_event_batch(session_id) - await self.fence(state) - raise - self._flush_event_batch(session_id) - - def begin_event_stream(self, session_id: UUID) -> Token: - """Stage controller output until each explicit checkpoint in its loop.""" - return self._event_batch.set({ - "session_id": session_id, "events": [], "committed": False, - "persistent": True, "owner_task": _current_task(), - }) - - async def end_event_stream(self, token: Token) -> None: - batch = self._event_batch.get() - try: - if batch is not None and batch["events"]: - session_id = batch["session_id"] - batch["events"].clear() - state = self._sessions.get(session_id) - if self.journal is not None and state is not None and not state.lease_fenced: - await self.fence(state) - finally: - self._event_batch.reset(token) + # ── операция: один commit, потом события ── @contextlib.asynccontextmanager - async def durable_transition(self, session_id: UUID): - """Do not publish state-changing events until its checkpoint commits.""" - batch = { - "session_id": session_id, "events": [], "committed": False, - "persistent": False, "owner_task": _current_task(), - } - token: Token = self._event_batch.set(batch) + async def operation(self, session_id: UUID) -> AsyncIterator[_Operation]: + """Изменение занятия фиксируется одним `store.commit` на выходе. + + События копятся до коммита. Исключение внутри операции или сбой + коммита отбрасывают их и закрывают занятие на узле (fail closed). + """ + op = _Operation(session_id) + token: Token = self._operation.set(op) try: - yield - if not batch["committed"]: - await self.checkpoint(session_id) - else: - self._flush_event_batch(session_id) - except Exception: - self._discard_event_batch(session_id) + yield op state = self._sessions.get(session_id) - if self.journal is not None and state is not None and not state.lease_fenced: + if state is not None and op.persist: + if state.lease_fenced: + raise RuntimeError(LEASE_FENCED_MESSAGE) + op.persist = False # второй commit из обработчика отмены не нужен + await asyncio.shield(self.store.commit(state, op.records)) + except asyncio.CancelledError: + # Задачу сокета отменили посреди операции — это не сбой хранилища. + # Сделанное фиксируется, события слать уже некому. + op.events.clear() + state = self._sessions.get(session_id) + if state is not None and op.persist and not state.lease_fenced: + try: + await asyncio.shield(self.store.commit(state, op.records)) + except Exception: + await self.fence(state) + raise + except Exception: + op.events.clear() + state = self._sessions.get(session_id) + if state is not None and not state.lease_fenced: await self.fence(state) raise finally: - self._event_batch.reset(token) + op.open = False + self._operation.reset(token) + for registry, event in op.events: + self._put(registry.get(session_id, set()), event) - def _discard_event_batch(self, session_id: UUID) -> None: - batch = self._event_batch.get() - if batch is not None and batch["session_id"] == session_id: - batch["events"].clear() + def record(self, session_id: UUID, record: Record) -> None: + """Строка журнала уходит в commit текущей операции, не отдельной транзакцией.""" + op = self._operation.get() + if op is None or not op.open or op.session_id != session_id: + raise RuntimeError(f"запись занятия {session_id} вне операции") + op.records.append(record) - def _flush_event_batch(self, session_id: UUID) -> None: - batch = self._event_batch.get() - if batch is None or batch["session_id"] != session_id: - return - pending, batch["events"] = batch["events"], [] - batch["committed"] = not batch.get("persistent", False) - for registry, target_session_id, event in pending: - self._put(registry.get(target_session_id, set()), event) + async def commit(self, session_id: UUID, *records: Record) -> None: + """Операция из одних строк — реплика голосового контура.""" + async with self.operation(session_id): + for record in records: + self.record(session_id, record) def _send(self, registry: dict[UUID, set[asyncio.Queue]], session_id: UUID, event: BaseModel) -> None: - batch = self._event_batch.get() - if isinstance(event, ErrorEvent): - self._put(registry.get(session_id, set()), event) - elif (batch is not None and batch["session_id"] == session_id - and batch["owner_task"] is _current_task()): - batch["events"].append((registry, session_id, event)) + op = self._operation.get() + if (op is not None and op.open and op.session_id == session_id + and not isinstance(event, ErrorEvent)): + op.events.append((registry, event)) else: self._put(registry.get(session_id, set()), event) @@ -316,12 +278,11 @@ class SessionHub: if state is None or state.ended: return if state.dds_phase: - active_before = state.desk.active_id - delivered = state.desk.deliver_due() - if state.desk.active_id != active_before and state.desk.active_id: - self.to_station(session_id, state.card_received_event()) - if delivered: - await self.checkpoint(session_id) + async with self.operation(session_id) as op: + active_before = state.desk.active_id + op.persist = bool(state.desk.deliver_due()) + if state.desk.active_id != active_before and state.desk.active_id: + self.to_station(session_id, state.card_received_event()) # Keep the pending count and countdown live even while # the active dispatcher card is being handled. self.to_station(session_id, StationState(snapshot=state.station_snapshot())) diff --git a/backend/app/session/journal.py b/backend/app/session/journal.py deleted file mode 100644 index 6e06156..0000000 --- a/backend/app/session/journal.py +++ /dev/null @@ -1,423 +0,0 @@ -"""Запись хода занятия в БД. - -Профиль курсанта, дельта попыток и аналитика группы строятся по журналу, -а не по памяти процесса: всё, что здесь не записано, для оценки не существует. -""" - -import logging -from datetime import datetime, timedelta -from uuid import UUID - -from sqlalchemy import select, update -from sqlalchemy.ext.asyncio import async_sessionmaker - -from app.db import repo -from app.db.models import AuditLog, Score, SelfAssessment, Session, User, Utterance -from app.domain.events import Mood, Speaker, TranscriptEntry -from app.session.checkpoint import dump_state, load_state -from app.session.state import SessionState, now_utc - -log = logging.getLogger(__name__) -LEASE_SECONDS = 15 - - -class SessionLeaseLost(RuntimeError): - """This process no longer owns the durable session generation.""" - - -class DbJournal: - def __init__(self, sessionmaker: async_sessionmaker, node_id: str | None = None) -> None: - self._sessionmaker = sessionmaker - self._node_id = node_id - self._epochs: dict[UUID, int] = {} - - async def _fence(self, db, session_id: UUID, expected_epoch: int | None = None) -> None: - """Renew and fence this write in the same transaction as its mutation.""" - if self._node_id is None: - return - epoch = expected_epoch if expected_epoch is not None else self._epochs.get(session_id) - if epoch is None: - raise SessionLeaseLost(f"session {session_id} has no local fencing epoch") - now = now_utc() - result = await db.execute( - update(Session) - .where( - Session.id == session_id, - Session.backend_node_id == self._node_id, - Session.backend_fencing_epoch == epoch, - ) - .values(backend_lease_until=now + timedelta(seconds=LEASE_SECONDS)) - .returning(Session.id) - ) - if result.scalar_one_or_none() is None: - raise SessionLeaseLost(f"session {session_id} owner epoch {epoch} was fenced") - - async def _write( - self, action, *args, _fence_session_id: UUID | None = None, - _fence_epoch: int | None = None, _raise_errors: bool = False, **kwargs - ) -> None: - """Ошибка записи не роняет занятие, но и не проглатывается молча: - занятие идёт дальше, в логе остаётся след.""" - try: - async with self._sessionmaker() as db: - if _fence_session_id is not None: - await self._fence(db, _fence_session_id, _fence_epoch) - await action(db, *args, **kwargs) - except SessionLeaseLost: - raise - except Exception as exc: # noqa: BLE001 — журнал не должен ронять живую сессию - session_id = _fence_session_id or kwargs.get("session_id") - log.error("журнал: запись не удалась для сессии %s (%s)", - session_id, type(exc).__name__) - if _raise_errors: - raise - - async def start_lesson( - self, session_id: UUID, scenario_id: str, mode: str, trainee_name: str | None, - trainee_id: UUID | None = None, owner_login: str | None = None, - backend_node_id: str | None = None, - ) -> tuple[int, UUID | None, str | None, int] | None: - """Завести сессию в журнале и вернуть номер попытки и ID курсанта. - - Строка сессии и событие аудита фиксируются вместе. При сбое транзакции - занятие не запускается без долговечной истории. - """ - try: - async with self._sessionmaker() as db: - node_id = backend_node_id or self._node_id - - def audit_start(transaction, row): - if row.backend_fencing_epoch <= 0: - row.backend_fencing_epoch = 1 - row.backend_lease_until = now_utc() + timedelta(seconds=LEASE_SECONDS) - transaction.add(AuditLog( - actor=owner_login or "system", - role="instructor" if owner_login else "system", - action="lesson.start", - object_id=str(row.id), - detail=f"{scenario_id}, mode {mode}", - )) - - row = await repo.ensure_session( - db, - session_id=session_id, - scenario_id=scenario_id, - mode=mode, - trainee_name=trainee_name, - trainee_id=trainee_id, - owner_login=owner_login, - backend_node_id=node_id, - before_commit=audit_start, - ) - epoch = getattr(row, "backend_fencing_epoch", 0) or 1 - self._epochs[session_id] = epoch - service = None - if row.trainee_id is not None: - service = await db.scalar( - select(User.service) - .where(User.trainee_id == row.trainee_id, User.blocked.is_(False)) - .limit(1) - ) - return row.attempt, row.trainee_id, service, epoch - except PermissionError: - raise - except Exception as exc: # noqa: BLE001 — журнал не должен ронять живую сессию - log.error("журнал: не удалось завести сессию %s (%s)", - session_id, type(exc).__name__) - return None - - async def checkpoint(self, state: SessionState) -> None: - """Сохранить снимок после подтверждённого действия пользователя.""" - async def action(db): - values = ( - {"live_state": None, "checkpoint_at": None} - if state.ended - else {"live_state": dump_state(state), "checkpoint_at": now_utc()} - ) - await db.execute(update(Session).where(Session.id == state.session_id).values(**values)) - await db.commit() - - if state.backend_fencing_epoch > 0: - self._epochs.setdefault(state.session_id, state.backend_fencing_epoch) - await self._write( - lambda db: action(db), _fence_session_id=state.session_id, - _fence_epoch=state.backend_fencing_epoch or None, - _raise_errors=True, - ) - - async def renew(self, session_id: UUID) -> None: - """Refresh an owned session lease; concurrent takeover is row-serialized.""" - async with self._sessionmaker() as db: - await self._fence(db, session_id) - await db.commit() - - async def claim_expired(self, session_id: UUID | None = None) -> list[SessionState]: - """Atomically fence and restore expired owners on this backend node.""" - if self._node_id is None: - return [] - now = now_utc() - conditions = [ - Session.ended_at.is_(None), - Session.live_state.is_not(None), - Session.checkpoint_at.is_not(None), - (Session.backend_node_id.is_(None) | (Session.backend_node_id != self._node_id)), - (Session.backend_lease_until.is_(None) | (Session.backend_lease_until <= now)), - ] - if session_id is not None: - conditions.append(Session.id == session_id) - async with self._sessionmaker() as db: - rows = (await db.scalars( - select(Session).where(*conditions).with_for_update(skip_locked=True).limit(100) - )).all() - for row in rows: - row.backend_node_id = self._node_id - row.backend_fencing_epoch = max(1, row.backend_fencing_epoch + 1) - row.backend_lease_until = now + timedelta(seconds=LEASE_SECONDS) - if rows: - await db.commit() - if not rows: - return [] - return await self.restore_active(bump_owned_epoch=False) - - async def restore_active(self, *, bump_owned_epoch: bool = True) -> list[SessionState]: - """Восстановить только незавершённые сессии с валидным снимком.""" - restored: list[SessionState] = [] - async with self._sessionmaker() as db: - active_with_snapshot = ( - Session.ended_at.is_(None), - Session.live_state.is_not(None), - Session.checkpoint_at.is_not(None), - ) - if self._node_id is not None: - # Adopt legacy unassigned snapshots exactly once. Concurrent - # nodes lock disjoint rows; subsequent restores are owner-only. - unassigned = (await db.scalars( - select(Session) - .where(*active_with_snapshot, Session.backend_node_id.is_(None)) - .with_for_update(skip_locked=True) - )).all() - for row in unassigned: - row.backend_node_id = self._node_id - row.backend_fencing_epoch = max(1, row.backend_fencing_epoch + 1) - row.backend_lease_until = now_utc() + timedelta(seconds=LEASE_SECONDS) - if unassigned: - await db.commit() - # A restarted process with the same stable node ID is a new - # owner generation. Bump before exposing any restored state. - owned = (await db.scalars( - select(Session) - .where(*active_with_snapshot, Session.backend_node_id == self._node_id) - .with_for_update(skip_locked=True) - )).all() - if bump_owned_epoch: - for row in owned: - row.backend_fencing_epoch = max(1, row.backend_fencing_epoch + 1) - row.backend_lease_until = now_utc() + timedelta(seconds=LEASE_SECONDS) - if owned: - await db.commit() - rows = (await db.scalars( - select(Session).where( - *active_with_snapshot, - Session.backend_node_id == self._node_id, - ) - )).all() - else: - rows = (await db.scalars( - select(Session).where(*active_with_snapshot) - )).all() - for row in rows: - try: - state = load_state(row.live_state, row.checkpoint_at) - if state.session_id != row.id: - raise ValueError("ID снимка не совпадает с записью занятия") - state.owner_login = row.owner_login - state.backend_fencing_epoch = row.backend_fencing_epoch - self._epochs[row.id] = row.backend_fencing_epoch - # Реплики пишутся отдельно сразу после появления. Если - # процесс умер между репликой и общим снимком, отдельный - # журнал не даёт потерять последний фрагмент диалога. - utterances = (await db.scalars( - select(Utterance) - .where(Utterance.session_id == row.id) - .order_by(Utterance.at, Utterance.ref) - )).all() - if utterances: - state.transcript = [ - TranscriptEntry( - ref=item.ref, - speaker=Speaker(item.speaker), - text=item.text, - at=item.at, - mood=Mood(item.mood) if item.mood else None, - ) - for item in utterances - ] - restored.append(state) - except Exception as exc: # noqa: BLE001 — один снимок не блокирует весь стенд - log.error("журнал: снимок занятия %s повреждён (%s)", - row.id, type(exc).__name__) - return restored - - async def utterance(self, session_id: UUID, entry) -> None: - await self._write( - repo.append_utterance, - session_id=session_id, - ref=entry.ref, - speaker=entry.speaker.value, - text=entry.text, - at=entry.at, - mood=entry.mood.value if entry.mood else None, - _fence_session_id=session_id, - ) - - async def hint(self, session_id: UUID, checklist_id: str, question: str, at: datetime) -> None: - await self._write( - repo.record_hint, _fence_session_id=session_id, session_id=session_id, - checklist_id=checklist_id, question=question, at=at - ) - - async def note(self, session_id: UUID, ref: str, text: str, author: str) -> None: - await self._write( - repo.add_note, _fence_session_id=session_id, session_id=session_id, - transcript_ref=ref, text=text, author=author - ) - - async def self_assessment( - self, session_id: UUID, missed: list[str], comment: str, at: datetime - ) -> bool: - """Persist trainee reflection and its security audit together.""" - try: - async with self._sessionmaker() as db: - session = await db.get(Session, session_id) - if session is None: - return False - actor = "system" - role = "system" - if session.trainee_id is not None: - login = await db.scalar( - select(User.login).where(User.trainee_id == session.trainee_id) - ) - if login: - actor, role = login, "trainee" - else: - actor, role = f"trainee:{session.trainee_id}", "trainee" - db.add(SelfAssessment( - session_id=session_id, missed=missed, comment=comment, submitted_at=at - )) - db.add(AuditLog( - actor=actor, - role=role, - action="self_assessment.submit", - object_id=str(session_id), - detail=f"missed_count={len(missed)}; comment_chars={len(comment)}", - )) - await self._fence(db, session_id) - await db.commit() - return True - except SessionLeaseLost: - raise - except Exception as exc: # noqa: BLE001 — do not accept an unaudited reflection - log.error("самооценка и аудит сессии %s не сохранены (%s)", - session_id, type(exc).__name__) - return False - - async def score(self, session_id: UUID, score_auto: float, report: dict) -> bool: - """Persist the initial result and its audit event atomically.""" - try: - async with self._sessionmaker() as db: - await self._fence(db, session_id) - db.add(Score( - session_id=session_id, score_auto=score_auto, - score_final=score_auto, report=report, - )) - db.add(AuditLog( - actor="system", role="system", action="score.calculate", - object_id=str(session_id), detail=f"score_auto={score_auto}", - )) - await db.commit() - return True - except SessionLeaseLost: - raise - except Exception as exc: # noqa: BLE001 — result is not complete until durable - log.error("итоговая оценка и аудит сессии %s не сохранены (%s)", - session_id, type(exc).__name__) - return False - - async def score_override( - self, session_id: UUID, score_final: float, author: str, comment: str - ) -> bool: - """Persist a live correction and its security audit as one transaction.""" - try: - async with self._sessionmaker() as db: - await self._fence(db, session_id) - score = await db.scalar( - select(Score) - .where(Score.session_id == session_id) - .with_for_update() - ) - if score is None: - return False - score.score_final = score_final - score.overridden_by = author - score.override_comment = comment - report = dict(score.report or {}) - archived = report.get("full_report") - if isinstance(archived, dict): - archived = dict(archived) - archived.update({ - "score_auto": score.score_auto, - "score_final": score_final, - "overridden_by": author, - "override_comment": comment, - }) - report["full_report"] = archived - score.report = report - db.add(AuditLog( - actor=author, - role="instructor", - action="score.override", - object_id=str(session_id), - detail=(f"{score.score_auto} → {score_final}; " - f"comment_chars={len(comment)}"), - )) - await db.commit() - return True - except SessionLeaseLost: - raise - except Exception as exc: # noqa: BLE001 — do not confirm a correction without its audit - log.error("корректировка оценки и аудит сессии %s не сохранены (%s)", - session_id, type(exc).__name__) - return False - - async def score_snapshot(self, session_id: UUID, report: dict) -> None: - """Обновить полный архивный разбор после самооценки курсанта.""" - async def action(db): - await db.execute( - update(Score).where(Score.session_id == session_id).values(report=report) - ) - await db.commit() - - await self._write(lambda db: action(db), _fence_session_id=session_id) - - async def session_started(self, session_id: UUID, at: datetime) -> None: - async def action(db): - await db.execute(update(Session).where(Session.id == session_id).values(started_at=at)) - await db.commit() - - await self._write(lambda db: action(db), _fence_session_id=session_id) - - async def session_ended(self, session_id: UUID, at: datetime, reason: str) -> None: - async def action(db): - await db.execute( - update(Session) - .where(Session.id == session_id) - .values( - ended_at=at, - end_reason=reason, - live_state=None, - checkpoint_at=None, - ) - ) - await db.commit() - - await self._write(lambda db: action(db), _fence_session_id=session_id) diff --git a/backend/app/session/pg_store.py b/backend/app/session/pg_store.py new file mode 100644 index 0000000..b35e33c --- /dev/null +++ b/backend/app/session/pg_store.py @@ -0,0 +1,362 @@ +"""PostgreSQL adapter хранилища занятия. + +Профиль курсанта, дельта попыток и аналитика группы строятся по этим строкам, +а не по памяти процесса: всё, что здесь не записано, для оценки не существует. +Каждый `commit` — одна транзакция: продление lease с проверкой `(node_id, epoch)`, +append-only строки операции и снимок `live_state`. +""" + +import logging +from collections.abc import Callable, Sequence +from datetime import timedelta +from uuid import UUID + +from sqlalchemy import select, update +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from app.db import repo +from app.db.models import ( + AuditLog, + HintUse, + InstructorNote, + Score, + SelfAssessment, + Session, + User, + Utterance, +) +from app.domain.events import Mood, Speaker, TranscriptEntry +from app.session.checkpoint import dump_state, load_state +from app.session.state import SessionState, now_utc +from app.session.store import ( + HintRecorded, + LessonEnded, + LessonIdentity, + LessonRequest, + LessonStarted, + NoteAdded, + Record, + ScoreArchived, + ScoreCalculated, + ScoreOverridden, + SelfAssessed, + SessionLeaseLost, + UtteranceAppended, + apply_score_override, +) + +log = logging.getLogger(__name__) +LEASE_SECONDS = 15 + +__all__ = ["LEASE_SECONDS", "PostgresSessionStore", "SessionLeaseLost"] + + +class PostgresSessionStore: + persistent = True + + def __init__(self, sessionmaker: async_sessionmaker, node_id: str | None = None) -> None: + self._sessionmaker = sessionmaker + self._node_id = node_id + self._epochs: dict[UUID, int] = {} + + async def _fence(self, db, session_id: UUID, expected_epoch: int | None = None) -> None: + """Renew and fence this write in the same transaction as its mutation.""" + if self._node_id is None: + return + epoch = expected_epoch if expected_epoch is not None else self._epochs.get(session_id) + if epoch is None: + raise SessionLeaseLost(f"session {session_id} has no local fencing epoch") + now = now_utc() + result = await db.execute( + update(Session) + .where( + Session.id == session_id, + Session.backend_node_id == self._node_id, + Session.backend_fencing_epoch == epoch, + ) + .values(backend_lease_until=now + timedelta(seconds=LEASE_SECONDS)) + .returning(Session.id) + ) + if result.scalar_one_or_none() is None: + raise SessionLeaseLost(f"session {session_id} owner epoch {epoch} was fenced") + + async def open( + self, request: LessonRequest, build: Callable[[LessonIdentity], SessionState], + ) -> SessionState: + """Строка занятия, аудит запуска и первый снимок — одна транзакция. + + При сбое занятие не запускается без долговечной истории. + """ + async with self._sessionmaker() as db: + row = await repo.ensure_session( + db, + session_id=request.session_id, + scenario_id=request.scenario_id, + mode=request.mode, + trainee_name=request.trainee_name, + trainee_id=request.trainee_id, + owner_login=request.owner_login, + backend_node_id=request.backend_node_id or self._node_id, + commit=False, + ) + if row.backend_fencing_epoch <= 0: + row.backend_fencing_epoch = 1 + row.backend_lease_until = now_utc() + timedelta(seconds=LEASE_SECONDS) + db.add(AuditLog( + actor=request.owner_login or "system", + role="instructor" if request.owner_login else "system", + action="lesson.start", + object_id=str(row.id), + detail=f"{request.scenario_id}, mode {request.mode}", + )) + service = None + if row.trainee_id is not None: + service = await db.scalar( + select(User.service) + .where(User.trainee_id == row.trainee_id, User.blocked.is_(False)) + .limit(1) + ) + epoch = row.backend_fencing_epoch + state = build(LessonIdentity( + attempt=row.attempt, trainee_id=row.trainee_id, + service=service, fencing_epoch=epoch, + )) + if state.started_at is not None: + row.started_at = state.started_at + row.live_state = dump_state(state) + row.checkpoint_at = now_utc() + await db.commit() + self._epochs[request.session_id] = epoch + return state + + async def commit(self, state: SessionState, records: Sequence[Record] = ()) -> None: + session_id = state.session_id + if state.backend_fencing_epoch > 0: + self._epochs.setdefault(session_id, state.backend_fencing_epoch) + try: + async with self._sessionmaker() as db: + await self._fence(db, session_id, state.backend_fencing_epoch or None) + for record in records: + await self._apply(db, session_id, record) + values = ( + {"live_state": None, "checkpoint_at": None} + if state.ended + else {"live_state": dump_state(state), "checkpoint_at": now_utc()} + ) + await db.execute( + update(Session).where(Session.id == session_id).values(**values) + ) + await db.commit() + except SessionLeaseLost: + raise + except Exception as exc: + # SQL-трасса несёт реплики и комментарии курсанта — в лог только тип. + log.error("хранилище: commit занятия %s не удался (%s)", + session_id, type(exc).__name__) + raise + + async def commit_archived(self, session_id: UUID, records: Sequence[Record]) -> None: + """Занятия нет в памяти узла, владеть нечем: без lease и снимка.""" + try: + async with self._sessionmaker() as db: + for record in records: + await self._apply(db, session_id, record) + await db.commit() + except Exception as exc: + log.error("хранилище: запись архивного занятия %s не удалась (%s)", + session_id, type(exc).__name__) + raise + + async def _apply(self, db: AsyncSession, session_id: UUID, record: Record) -> None: + match record: + case UtteranceAppended(entry=entry): + db.add(Utterance( + session_id=session_id, ref=entry.ref, speaker=entry.speaker.value, + text=entry.text, at=entry.at, + mood=entry.mood.value if entry.mood else None, + )) + case HintRecorded(): + db.add(HintUse( + session_id=session_id, checklist_id=record.checklist_id, + question=record.question, at=record.at, + )) + case NoteAdded(): + db.add(InstructorNote( + session_id=session_id, transcript_ref=record.transcript_ref, + text=record.text, author=record.author, + )) + case SelfAssessed(): + actor, role = await self._trainee_actor(db, session_id) + db.add(SelfAssessment( + session_id=session_id, missed=record.missed, + comment=record.comment, submitted_at=record.at, + )) + db.add(AuditLog( + actor=actor, role=role, action="self_assessment.submit", + object_id=str(session_id), + detail=f"missed_count={len(record.missed)}; comment_chars={len(record.comment)}", + )) + case LessonStarted(at=at): + await db.execute( + update(Session).where(Session.id == session_id).values(started_at=at) + ) + case LessonEnded(at=at, reason=reason): + await db.execute( + update(Session).where(Session.id == session_id) + .values(ended_at=at, end_reason=reason) + ) + case ScoreCalculated(): + db.add(Score( + session_id=session_id, score_auto=record.score_auto, + score_final=record.score_auto, report=record.report, + )) + db.add(AuditLog( + actor="system", role="system", action="score.calculate", + object_id=str(session_id), detail=f"score_auto={record.score_auto}", + )) + case ScoreArchived(): + await db.flush() + await db.execute( + update(Score).where(Score.session_id == session_id) + .values(report=record.report) + ) + case ScoreOverridden(): + await db.flush() + score = await db.scalar( + select(Score).where(Score.session_id == session_id).with_for_update() + ) + if score is None: + raise LookupError(f"нет оценки занятия {session_id}") + score.score_final = record.score_final + score.overridden_by = record.author + score.override_comment = record.comment + score.report = apply_score_override(dict(score.report or {}), record) + db.add(AuditLog( + actor=record.author, role=record.role, action="score.override", + object_id=str(session_id), + # Обоснование остаётся в разборе; аудиту нужны изменение и автор, + # а не вторая бессрочная копия свободного текста. + detail=(f"{score.score_auto} → {record.score_final}; " + f"comment_chars={len(record.comment)}"), + )) + await db.flush() + + @staticmethod + async def _trainee_actor(db: AsyncSession, session_id: UUID) -> tuple[str, str]: + session = await db.get(Session, session_id) + if session is None or session.trainee_id is None: + return "system", "system" + login = await db.scalar(select(User.login).where(User.trainee_id == session.trainee_id)) + return (login, "trainee") if login else (f"trainee:{session.trainee_id}", "trainee") + + async def renew(self, session_id: UUID) -> None: + """Refresh an owned session lease; concurrent takeover is row-serialized.""" + async with self._sessionmaker() as db: + await self._fence(db, session_id) + await db.commit() + + async def claim_expired(self, session_id: UUID | None = None) -> list[SessionState]: + """Atomically fence and restore expired owners on this backend node.""" + if self._node_id is None: + return [] + now = now_utc() + conditions = [ + Session.ended_at.is_(None), + Session.live_state.is_not(None), + Session.checkpoint_at.is_not(None), + (Session.backend_node_id.is_(None) | (Session.backend_node_id != self._node_id)), + (Session.backend_lease_until.is_(None) | (Session.backend_lease_until <= now)), + ] + if session_id is not None: + conditions.append(Session.id == session_id) + async with self._sessionmaker() as db: + rows = (await db.scalars( + select(Session).where(*conditions).with_for_update(skip_locked=True).limit(100) + )).all() + for row in rows: + row.backend_node_id = self._node_id + row.backend_fencing_epoch = max(1, row.backend_fencing_epoch + 1) + row.backend_lease_until = now + timedelta(seconds=LEASE_SECONDS) + if rows: + await db.commit() + if not rows: + return [] + return await self.restore_active(bump_owned_epoch=False) + + async def restore_active(self, *, bump_owned_epoch: bool = True) -> list[SessionState]: + """Восстановить только незавершённые сессии с валидным снимком.""" + restored: list[SessionState] = [] + async with self._sessionmaker() as db: + active_with_snapshot = ( + Session.ended_at.is_(None), + Session.live_state.is_not(None), + Session.checkpoint_at.is_not(None), + ) + if self._node_id is not None: + # Adopt legacy unassigned snapshots exactly once. Concurrent + # nodes lock disjoint rows; subsequent restores are owner-only. + unassigned = (await db.scalars( + select(Session) + .where(*active_with_snapshot, Session.backend_node_id.is_(None)) + .with_for_update(skip_locked=True) + )).all() + for row in unassigned: + row.backend_node_id = self._node_id + row.backend_fencing_epoch = max(1, row.backend_fencing_epoch + 1) + row.backend_lease_until = now_utc() + timedelta(seconds=LEASE_SECONDS) + if unassigned: + await db.commit() + # A restarted process with the same stable node ID is a new + # owner generation. Bump before exposing any restored state. + owned = (await db.scalars( + select(Session) + .where(*active_with_snapshot, Session.backend_node_id == self._node_id) + .with_for_update(skip_locked=True) + )).all() + if bump_owned_epoch: + for row in owned: + row.backend_fencing_epoch = max(1, row.backend_fencing_epoch + 1) + row.backend_lease_until = now_utc() + timedelta(seconds=LEASE_SECONDS) + if owned: + await db.commit() + rows = (await db.scalars( + select(Session).where( + *active_with_snapshot, + Session.backend_node_id == self._node_id, + ) + )).all() + else: + rows = (await db.scalars( + select(Session).where(*active_with_snapshot) + )).all() + for row in rows: + try: + state = load_state(row.live_state, row.checkpoint_at) + if state.session_id != row.id: + raise ValueError("ID снимка не совпадает с записью занятия") + state.owner_login = row.owner_login + state.backend_fencing_epoch = row.backend_fencing_epoch + self._epochs[row.id] = row.backend_fencing_epoch + # Стенограмма — по строкам реплик: снимок, записанный до + # перехода на единый commit, мог отстать от них. + utterances = (await db.scalars( + select(Utterance) + .where(Utterance.session_id == row.id) + .order_by(Utterance.at, Utterance.ref) + )).all() + if utterances: + state.transcript = [ + TranscriptEntry( + ref=item.ref, + speaker=Speaker(item.speaker), + text=item.text, + at=item.at, + mood=Mood(item.mood) if item.mood else None, + ) + for item in utterances + ] + restored.append(state) + except Exception as exc: # noqa: BLE001 — один снимок не блокирует весь стенд + log.error("журнал: снимок занятия %s повреждён (%s)", + row.id, type(exc).__name__) + return restored diff --git a/backend/app/session/store.py b/backend/app/session/store.py new file mode 100644 index 0000000..ea053b1 --- /dev/null +++ b/backend/app/session/store.py @@ -0,0 +1,215 @@ +"""Port хранилища живого занятия. + +Операция курсанта или преподавателя фиксируется одним `commit`: проверка +владения `(node_id, epoch)`, снимок состояния и append-only строки — в одной +транзакции. Хаб публикует события только после успешного `commit`; сбой +отбрасывает их и закрывает занятие на этом узле (docs/arch/SCALE-OUT.md). + +Adapter-ы: PostgreSQL (`app.session.pg_store`) и in-memory — для `DEMO_NO_DB` +и юнит-тестов. +""" + +from collections.abc import Callable, Sequence +from dataclasses import dataclass +from datetime import datetime +from typing import Protocol +from uuid import UUID + +from app.domain.events import TranscriptEntry +from app.session.checkpoint import dump_state +from app.session.state import SessionState + + +class SessionLeaseLost(RuntimeError): + """This process no longer owns the durable session generation.""" + + +# ── append-only строки операции ── + +@dataclass(frozen=True) +class UtteranceAppended: + entry: TranscriptEntry + + +@dataclass(frozen=True) +class HintRecorded: + checklist_id: str + question: str + at: datetime + + +@dataclass(frozen=True) +class NoteAdded: + transcript_ref: str + text: str + author: str + + +@dataclass(frozen=True) +class SelfAssessed: + missed: list[str] + comment: str + at: datetime + + +@dataclass(frozen=True) +class LessonStarted: + at: datetime + + +@dataclass(frozen=True) +class LessonEnded: + at: datetime + reason: str + + +@dataclass(frozen=True) +class ScoreCalculated: + score_auto: float + report: dict + + +@dataclass(frozen=True) +class ScoreArchived: + """Полный разбор дополнен после оценки — например, самооценкой курсанта.""" + report: dict + + +@dataclass(frozen=True) +class ScoreOverridden: + score_final: float + author: str + role: str + comment: str + + +Record = (UtteranceAppended | HintRecorded | NoteAdded | SelfAssessed | LessonStarted + | LessonEnded | ScoreCalculated | ScoreArchived | ScoreOverridden) + + +def apply_score_override(report: dict, override: ScoreOverridden) -> dict: + """Правка балла одинакова для живой сессии, строки `Score` и архива. + + Автооценка остаётся рядом: видно, что скорректировано и кем. Архивный + `full_report` правится тем же образом, иначе PDF и история покажут + старую оценку. + """ + fields = { + "score_final": override.score_final, + "overridden_by": override.author, + "override_comment": override.comment, + } + updated = {**report, **fields} + archived = report.get("full_report") + if isinstance(archived, dict): + updated["full_report"] = { + **archived, + "score_auto": report.get("score_auto", archived.get("score_auto")), + **fields, + } + return updated + + +@dataclass(frozen=True) +class LessonIdentity: + """То, что о занятии знает только хранилище: номер попытки, курсант, эпоха.""" + attempt: int + trainee_id: UUID | None + service: str | None + fencing_epoch: int + + +@dataclass(frozen=True) +class LessonRequest: + session_id: UUID + scenario_id: str + mode: str + trainee_name: str | None + trainee_id: UUID | None + owner_login: str | None + backend_node_id: str | None + + +class SessionStore(Protocol): + #: Переживает ли занятие перезапуск процесса. Без этого каталог сценариев + #: и отчёты берутся только из памяти. + persistent: bool + + async def open( + self, request: LessonRequest, build: Callable[[LessonIdentity], SessionState], + ) -> SessionState: + """Завести занятие, аудит запуска и первый снимок одной транзакцией.""" + ... + + async def commit(self, state: SessionState, records: Sequence[Record] = ()) -> None: + """Проверка владения, снимок и строки — одна транзакция. Ошибка — исключение.""" + ... + + async def commit_archived(self, session_id: UUID, records: Sequence[Record]) -> None: + """Строки к занятию, которого уже нет в памяти узла (правка балла из истории).""" + ... + + async def restore_active(self) -> list[SessionState]: ... + async def renew(self, session_id: UUID) -> None: ... + async def claim_expired(self, session_id: UUID | None = None) -> list[SessionState]: ... + + +class MemorySessionStore: + """Занятие живёт до перезапуска процесса: `DEMO_NO_DB` и юнит-тесты. + + Хранит то же, что PostgreSQL, в словарях — тесты читают снимок, строки + и оценку тем же путём, что и сквозной сценарий. + """ + + persistent = False + + def __init__(self) -> None: + self.commits: list[tuple[UUID, list[Record]]] = [] + self.records: dict[UUID, list[Record]] = {} + self.scores: dict[UUID, dict] = {} + self._snapshots: dict[UUID, dict | None] = {} + + async def open( + self, request: LessonRequest, build: Callable[[LessonIdentity], SessionState], + ) -> SessionState: + state = build(LessonIdentity( + attempt=1, trainee_id=request.trainee_id, service=None, fencing_epoch=0, + )) + self.commits.append((request.session_id, [])) + self._snapshots[request.session_id] = dump_state(state) + return state + + async def commit(self, state: SessionState, records: Sequence[Record] = ()) -> None: + records = list(records) + self.commits.append((state.session_id, records)) + self._apply(state.session_id, records) + self._snapshots[state.session_id] = None if state.ended else dump_state(state) + + async def commit_archived(self, session_id: UUID, records: Sequence[Record]) -> None: + records = list(records) + self.commits.append((session_id, records)) + self._apply(session_id, records) + + def _apply(self, session_id: UUID, records: list[Record]) -> None: + for record in records: + if isinstance(record, ScoreCalculated): + self.scores[session_id] = dict(record.report) + elif isinstance(record, ScoreArchived): + self.scores[session_id] = dict(record.report) + elif isinstance(record, ScoreOverridden): + if session_id not in self.scores: + raise LookupError(f"нет оценки занятия {session_id}") + self.scores[session_id] = apply_score_override(self.scores[session_id], record) + self.records.setdefault(session_id, []).extend(records) + + def snapshot(self, session_id: UUID) -> dict | None: + return self._snapshots.get(session_id) + + async def restore_active(self) -> list[SessionState]: + return [] + + async def renew(self, session_id: UUID) -> None: + return None + + async def claim_expired(self, session_id: UUID | None = None) -> list[SessionState]: + return [] diff --git a/backend/app/voice/pipeline.py b/backend/app/voice/pipeline.py index 56447bc..b8a060c 100644 --- a/backend/app/voice/pipeline.py +++ b/backend/app/voice/pipeline.py @@ -11,7 +11,7 @@ import asyncio import hashlib import logging import time -from collections.abc import Callable +from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from pathlib import Path from uuid import UUID, uuid4 @@ -68,7 +68,8 @@ class VoiceSession: send_event: Callable[[object], None] send_observer: Callable[[object], None] send_audio: Callable[[bytes], None] - journal: object | None = None + #: Сохранить реплику одной операцией хранилища; без него реплика живёт в памяти. + persist: Callable[[object], Awaitable[None]] | None = None timings: list[TurnTiming] = field(default_factory=list) _vad: StreamingVad = field(init=False) @@ -149,8 +150,8 @@ class VoiceSession: entry = self.state.append(Speaker.OPERATOR, text) self.send_event(SttFinal(text=text, at=entry.at)) self.send_observer(TranscriptAppend(entry=entry)) - if self.journal: - await self.journal.utterance(self.session_id, entry) + if self.persist is not None: + await self.persist(entry) started = time.monotonic() line = await self._caller_line(text) @@ -174,8 +175,8 @@ class VoiceSession: entry = self.state.append(Speaker.CALLER, text, mood) self.send_event(CallerUtterance(utterance_id=utterance_id, text=text, at=entry.at, mood=mood)) self.send_observer(TranscriptAppend(entry=entry)) - if self.journal: - await self.journal.utterance(self.session_id, entry) + if self.persist is not None: + await self.persist(entry) self.send_event(TtsBegin(utterance_id=utterance_id)) playback_ends = time.monotonic() diff --git a/backend/tests/test_admin.py b/backend/tests/test_admin.py index 86fcb45..6cbabae 100644 --- a/backend/tests/test_admin.py +++ b/backend/tests/test_admin.py @@ -15,12 +15,13 @@ from app.api.auth import Principal from app.domain.roles import Role from app.main import app from app.session.hub import hub +from app.session.store import MemorySessionStore @pytest.fixture def client(): with TestClient(app) as test_client: - hub.journal = None + hub.store = MemorySessionStore() yield test_client diff --git a/backend/tests/test_auth.py b/backend/tests/test_auth.py index 5f6abe6..e97f839 100644 --- a/backend/tests/test_auth.py +++ b/backend/tests/test_auth.py @@ -19,12 +19,13 @@ from app.api.auth import hash_password, verify_password from app.domain.roles import Role from app.main import app from app.session.hub import hub +from app.session.store import MemorySessionStore @pytest.fixture def client(): with TestClient(app) as test_client: - hub.journal = None + hub.store = MemorySessionStore() yield test_client diff --git a/backend/tests/test_auth_hardening.py b/backend/tests/test_auth_hardening.py index b35b247..05f22cc 100644 --- a/backend/tests/test_auth_hardening.py +++ b/backend/tests/test_auth_hardening.py @@ -18,6 +18,7 @@ from app.api.http import admin from app.domain.roles import Role from app.main import app from app.session.hub import hub +from app.session.store import MemorySessionStore @pytest.mark.parametrize( @@ -92,7 +93,7 @@ def client(): # process' in-memory cache. auth.prime_generations({}) with TestClient(app) as test_client: - hub.journal = None + hub.store = MemorySessionStore() yield test_client diff --git a/backend/tests/test_call_privacy.py b/backend/tests/test_call_privacy.py index e1ffae4..4729d0e 100644 --- a/backend/tests/test_call_privacy.py +++ b/backend/tests/test_call_privacy.py @@ -7,6 +7,7 @@ import pytest from app.api.ws import call from app.dialog.slots import TurnResult from app.domain.events import Exercise, TranscriptEntry +from app.session.store import MemorySessionStore @pytest.mark.asyncio @@ -24,7 +25,7 @@ async def test_text_dialogue_provider_error_does_not_log_prompt_or_provider_body def revealed_facts(self): return [] - monkeypatch.setattr(call.hub, "journal", None) + monkeypatch.setattr(call.hub, "store", MemorySessionStore()) monkeypatch.setattr(call.hub, "to_trainee", lambda *_args: None) monkeypatch.setattr(call.hub, "to_observers", lambda *_args: None) state = SimpleNamespace( @@ -41,7 +42,9 @@ async def test_text_dialogue_provider_error_does_not_log_prompt_or_provider_body ), ) - await call._handle(uuid4(), state, SimpleNamespace(type="text.turn", text="where is the incident")) + session_id = uuid4() + async with call.hub.operation(session_id): + await call._handle(session_id, state, SimpleNamespace(type="text.turn", text="where is the incident")) assert secret not in caplog.text assert "RuntimeError" in caplog.text diff --git a/backend/tests/test_card_exercise.py b/backend/tests/test_card_exercise.py index 16e3bbd..0a79aee 100644 --- a/backend/tests/test_card_exercise.py +++ b/backend/tests/test_card_exercise.py @@ -13,6 +13,7 @@ from app.api.ws.call import _text_turn from app.scenarios import store from app.scoring.grammar import basic_check from app.session.hub import hub +from app.session.store import MemorySessionStore @pytest.fixture @@ -23,7 +24,7 @@ def client(monkeypatch): monkeypatch.setattr(sessions_http, "audit_required", audit_in_memory) with TestClient(app) as test_client: test_client.post("/api/auth/dev-token") - hub.journal = None + hub.store = MemorySessionStore() yield test_client @@ -450,41 +451,25 @@ def test_handoff_report_and_checklist_follow_operator_scenario(client, monkeypat state.self_assessed = True state.self_assessment = {"missed": [checklist[0]["id"]], "comment": ""} - asyncio.run(refresh_archived_report(session_id, state)) + async def refresh(): + async with hub.operation(session_id): + await refresh_archived_report(session_id, state) + + asyncio.run(refresh()) assert state.score["full_report"]["scenario_id"] == "fire-apartment-l2" assert state.score["full_report"]["metrics"] == state.score["metrics"] - session_row = SimpleNamespace(id=session_id, owner_login="teacher") - score_row = SimpleNamespace( - score_auto=state.score["score_auto"], score_final=state.score["score_auto"], - overridden_by=None, override_comment=None, - report={"full_report": state.score["full_report"]}, - ) - - class FakeDb: - async def scalar(self, statement): - return score_row - - def add(self, row): - pass - - async def commit(self): - pass - - async def fake_session(db, key): - return session_row - - monkeypatch.setattr(sessions_http.repo, "get_session", fake_session) monkeypatch.setattr( sessions_http, "require", - lambda request, *roles: Principal(login="teacher", full_name="Преподаватель", + lambda request, *roles: Principal(login=state.owner_login, full_name="Преподаватель", role=Role.INSTRUCTOR), ) corrected = asyncio.run(sessions_http.override( session_id, sessions_http.ScoreOverride(score_final=50, comment="проверено"), - object(), FakeDb(), + object(), None, )) assert corrected.scenario_id == "fire-apartment-l2" + assert state.score["full_report"]["score_final"] == 50 finally: control.__exit__(None, None, None) diff --git a/backend/tests/test_db.py b/backend/tests/test_db.py index 8a48aa6..04f442a 100644 --- a/backend/tests/test_db.py +++ b/backend/tests/test_db.py @@ -27,18 +27,59 @@ from app.db.models import ( Score, SelfAssessment, Session, Trainee, User, Utterance, ) from app.db.repo import SessionNodeConflict, ensure_session -from app.domain.events import Exercise, SessionMode +from app.domain.events import Exercise, SessionMode, Speaker from app.domain.roles import Role from app.main import LIBRARY, app from app.scenarios import store from app.session.checkpoint import load_state from app.session.dds import build_card from app.session.hub import hub -from app.session.journal import DbJournal, SessionLeaseLost +from app.session.pg_store import PostgresSessionStore +from app.session.store import ( + LessonRequest, + ScoreCalculated, + ScoreOverridden, + SelfAssessed, + SessionLeaseLost, + UtteranceAppended, +) from app.session.state import SessionState from app.session.timers import SessionTimers +def _state(session_id, scenario, **fields) -> SessionState: + return SessionState( + session_id=session_id, scenario_id=scenario.id, scenario_title=scenario.title, + level=scenario.level, mode=SessionMode.TRAINING, **fields, + ) + + +async def _open(store, scenario, session_id, trainee_name, trainee_id=None, owner_login=None): + """Открыть занятие и вернуть его идентичность — как делает пульт.""" + seen = {} + + def build(identity): + seen["identity"] = identity + return _state( + session_id, scenario, owner_login=owner_login, trainee_id=identity.trainee_id, + attempt=identity.attempt, backend_fencing_epoch=identity.fencing_epoch, + ) + + state = await store.open(LessonRequest( + session_id=session_id, scenario_id=scenario.id, mode="training", + trainee_name=trainee_name, trainee_id=trainee_id, owner_login=owner_login, + backend_node_id=None, + ), build) + return state, seen["identity"] + + +def _entry(text): + from app.domain.events import Speaker, TranscriptEntry + + return TranscriptEntry(ref="u1", speaker=Speaker.CALLER, text=text, + at=datetime.now(timezone.utc)) + + @pytest.fixture async def db(): """Свой движок на каждый тест: глобальный в app.db.base кэшируется и @@ -340,16 +381,9 @@ async def test_station_command_ids_survive_a_real_postgres_checkpoint(db, scenar db, scenario_id=scenario.id, mode="training", session_id=uuid4(), ) command_id = str(uuid4()) - state = SessionState( - session_id=session.id, - scenario_id=scenario.id, - scenario_title=scenario.title, - level=scenario.level, - mode=SessionMode.TRAINING, - processed_station_commands=[command_id], - ) - journal = DbJournal(async_sessionmaker(db.bind, expire_on_commit=False)) - await journal.checkpoint(state) + state = _state(session.id, scenario, processed_station_commands=[command_id]) + pg = PostgresSessionStore(async_sessionmaker(db.bind, expire_on_commit=False)) + await pg.commit(state) observer_engine = create_async_engine(get_settings().database_url) try: @@ -406,24 +440,22 @@ async def test_security_audit_older_than_six_months_remains_queryable(db, monkey async def test_websocket_score_override_and_audit_commit_atomically(db, scenario): - from app.session.journal import DbJournal - session = await repo.create_session( db, scenario_id=scenario.id, mode="training", owner_login="score-teacher" ) score = Score( session_id=session.id, score_auto=70, score_final=70, - report={"full_report": {"score_auto": 70, "score_final": 70}}, + report={"score_auto": 70, "full_report": {"score_auto": 70, "score_final": 70}}, ) db.add(score) await db.commit() - maker = async_sessionmaker(db.bind, expire_on_commit=False) - journal = DbJournal(maker) + pg = PostgresSessionStore(async_sessionmaker(db.bind, expire_on_commit=False)) + state = _state(session.id, scenario, owner_login="score-teacher") # Force the audit insert to fail after the score row is staged. PostgreSQL - # must roll back both, and the websocket handler can then refuse the change. - saved = await journal.score_override(session.id, 85, "x" * 81, "manual review") - assert not saved + # must roll back both, and the commit must fail the whole operation. + with pytest.raises(Exception): + await pg.commit(state, [ScoreOverridden(85, "x" * 81, "instructor", "manual review")]) await db.refresh(score) assert score.score_final == 70 assert score.overridden_by is None @@ -433,14 +465,12 @@ async def test_websocket_score_override_and_audit_commit_atomically(db, scenario ) ) is None - saved = await journal.score_override( - session.id, 85, "score-teacher", "manual review" - ) - assert saved + await pg.commit(state, [ScoreOverridden(85, "score-teacher", "instructor", "manual review")]) await db.refresh(score) assert score.score_auto == 70 assert score.score_final == 85 assert score.overridden_by == "score-teacher" + assert score.report["score_final"] == 85 assert score.report["full_report"]["score_final"] == 85 audit_row = await db.scalar( select(AuditLog).where( @@ -455,16 +485,16 @@ async def test_websocket_score_override_and_audit_commit_atomically(db, scenario async def test_initial_result_and_calculation_audit_commit_atomically(db, scenario): - from app.session.journal import DbJournal - session = await repo.create_session( db, scenario_id=scenario.id, mode="training", owner_login="score-teacher" ) - journal = DbJournal(async_sessionmaker(db.bind, expire_on_commit=False)) + pg = PostgresSessionStore(async_sessionmaker(db.bind, expire_on_commit=False)) + state = _state(session.id, scenario, owner_login="score-teacher") # Invalid JSON makes PostgreSQL reject the score transaction; no orphaned # calculation-audit row may remain (or vice versa). - assert not await journal.score(session.id, 71, {"bad": object()}) + with pytest.raises(Exception): + await pg.commit(state, [ScoreCalculated(71, {"bad": object()})]) assert await db.scalar( select(Score.id).where(Score.session_id == session.id) ) is None @@ -472,7 +502,7 @@ async def test_initial_result_and_calculation_audit_commit_atomically(db, scenar AuditLog.action == "score.calculate", AuditLog.object_id == str(session.id) )) is None - assert await journal.score(session.id, 71, {"full_report": {"score_auto": 71}}) + await pg.commit(state, [ScoreCalculated(71, {"full_report": {"score_auto": 71}})]) score = await db.scalar(select(Score).where(Score.session_id == session.id)) audit_row = await db.scalar(select(AuditLog).where( AuditLog.action == "score.calculate", AuditLog.object_id == str(session.id) @@ -574,12 +604,9 @@ async def test_websocket_lesson_start_cannot_claim_foreign_group_trainee(db, sce await db.commit() session_id = uuid4() - journal = DbJournal(async_sessionmaker(db.bind, expire_on_commit=False)) + pg = PostgresSessionStore(async_sessionmaker(db.bind, expire_on_commit=False)) with pytest.raises(PermissionError, match="другой учебной групп"): - await journal.start_lesson( - session_id, scenario.id, "training", trainee.name, trainee.id, - owner_login=teacher, - ) + await _open(pg, scenario, session_id, trainee.name, trainee.id, owner_login=teacher) assert await db.get(Session, session_id) is None assert await db.scalar(select(AuditLog.id).where( AuditLog.action == "lesson.start", AuditLog.object_id == str(session_id) @@ -592,14 +619,13 @@ async def test_websocket_lesson_start_cannot_claim_foreign_group_trainee(db, sce async def test_websocket_lesson_start_audit_is_committed_with_session(db, scenario): session_id = uuid4() - journal = DbJournal(async_sessionmaker(db.bind, expire_on_commit=False)) - result = await journal.start_lesson( - session_id, scenario.id, "training", "Курсант", owner_login="lesson-teacher" - ) - assert result is not None and result[0] == 1 and result[1] is not None - assert result[2] is None - assert result[3] == 1 + pg = PostgresSessionStore(async_sessionmaker(db.bind, expire_on_commit=False)) + _, identity = await _open(pg, scenario, session_id, "Курсант", owner_login="lesson-teacher") + assert identity.attempt == 1 and identity.trainee_id is not None + assert identity.service is None + assert identity.fencing_epoch == 1 persisted = await db.get(Session, session_id) + assert persisted.live_state is not None, "первый снимок пишется вместе со строкой занятия" audit_row = await db.scalar(select(AuditLog).where( AuditLog.action == "lesson.start", AuditLog.object_id == str(session_id) )) @@ -609,25 +635,67 @@ async def test_websocket_lesson_start_audit_is_committed_with_session(db, scenar await db.commit() +async def test_store_commit_is_one_transaction_with_one_fencing_check(db, scenario): + """Реплика, финиш и оценка — одна транзакция и один `(node_id, epoch)`-чек.""" + from sqlalchemy import event + + from app.session.store import LessonEnded + + session_id = uuid4() + maker = async_sessionmaker(db.bind, expire_on_commit=False) + opened = [] + + def counting_maker(): + opened.append(True) + return maker() + + pg = PostgresSessionStore(counting_maker, node_id="node-a") + state, _ = await _open(pg, scenario, session_id, "Курсант", owner_login="lesson-teacher") + statements = [] + + def seen(_conn, _cursor, statement, *_args): + statements.append(statement) + + event.listen(db.bind.sync_engine, "before_cursor_execute", seen) + opened.clear() + try: + entry = state.append(Speaker.CALLER, "адрес: Ленина, 14") + state.ended_at = datetime.now(timezone.utc) + await pg.commit(state, [ + UtteranceAppended(entry), + LessonEnded(state.ended_at, "complete"), + ScoreCalculated(71, {"score_auto": 71}), + ]) + finally: + event.remove(db.bind.sync_engine, "before_cursor_execute", seen) + + assert len(opened) == 1, "одна операция — одна транзакция" + fence_checks = [sql for sql in statements + if sql.startswith("UPDATE sessions SET backend_lease_until")] + assert len(fence_checks) == 1 + row = await db.get(Session, session_id) + await db.refresh(row) + assert row.end_reason == "complete" and row.live_state is None + assert await db.scalar(select(Utterance.text).where(Utterance.session_id == session_id)) == ( + "адрес: Ленина, 14" + ) + await db.execute(delete(AuditLog).where(AuditLog.object_id == str(session_id))) + await db.execute(delete(Session).where(Session.id == session_id)) + await db.commit() + + async def test_old_backend_fencing_epoch_cannot_append_transcript(db, scenario): session_id = uuid4() - journal = DbJournal(async_sessionmaker(db.bind, expire_on_commit=False), node_id="node-a") - started = await journal.start_lesson( - session_id, scenario.id, "training", "Курсант", owner_login="lesson-teacher" - ) - assert started is not None + pg = PostgresSessionStore(async_sessionmaker(db.bind, expire_on_commit=False), node_id="node-a") + state, identity = await _open(pg, scenario, session_id, "Курсант", owner_login="lesson-teacher") await db.execute( update(Session) .where(Session.id == session_id) - .values(backend_node_id="node-b", backend_fencing_epoch=started[3] + 1) + .values(backend_node_id="node-b", backend_fencing_epoch=identity.fencing_epoch + 1) ) await db.commit() - entry = SimpleNamespace( - ref="u1", speaker=SimpleNamespace(value="caller"), text="проверка fencing", - at=datetime.now(timezone.utc), mood=None, - ) with pytest.raises(SessionLeaseLost): - await journal.utterance(session_id, entry) + await pg.commit(state, [UtteranceAppended(_entry("проверка fencing"))]) assert await db.scalar(select(Utterance.id).where(Utterance.session_id == session_id)) is None await db.execute(delete(AuditLog).where(AuditLog.object_id == str(session_id))) await db.execute(delete(Session).where(Session.id == session_id)) @@ -636,44 +704,28 @@ async def test_old_backend_fencing_epoch_cannot_append_transcript(db, scenario): async def test_expired_backend_lease_is_atomically_claimed_and_restored(db, scenario): session_id = uuid4() - old = DbJournal(async_sessionmaker(db.bind, expire_on_commit=False), node_id="node-a") - initial = await old.start_lesson( - session_id, scenario.id, "training", "Курсант", owner_login="lease-teacher" - ) - assert initial is not None - snapshot = SessionState( - session_id=session_id, - scenario_id=scenario.id, - scenario_title=scenario.title, - level=scenario.level, - mode=SessionMode.TRAINING, - owner_login="lease-teacher", - exercise=Exercise.DDS, - backend_fencing_epoch=initial[3], - ) - await old.checkpoint(snapshot) + old = PostgresSessionStore(async_sessionmaker(db.bind, expire_on_commit=False), node_id="node-a") + snapshot, identity = await _open(old, scenario, session_id, "Курсант", owner_login="lease-teacher") + snapshot.exercise = Exercise.DDS + await old.commit(snapshot) row = await db.get(Session, session_id) assert row is not None row.backend_lease_until = datetime.now(timezone.utc) - timedelta(seconds=1) await db.commit() - new = DbJournal(async_sessionmaker(db.bind, expire_on_commit=False), node_id="node-b") + new = PostgresSessionStore(async_sessionmaker(db.bind, expire_on_commit=False), node_id="node-b") restored = await new.claim_expired(session_id) assert len(restored) == 1 assert restored[0].session_id == session_id - assert restored[0].backend_fencing_epoch == initial[3] + 1 + assert restored[0].backend_fencing_epoch == identity.fencing_epoch + 1 persisted = await db.get(Session, session_id) assert persisted is not None await db.refresh(persisted) assert persisted.backend_node_id == "node-b" - assert persisted.backend_fencing_epoch == initial[3] + 1 + assert persisted.backend_fencing_epoch == identity.fencing_epoch + 1 - entry = SimpleNamespace( - ref="u1", speaker=SimpleNamespace(value="caller"), text="stale owner", - at=datetime.now(timezone.utc), mood=None, - ) with pytest.raises(SessionLeaseLost): - await old.utterance(session_id, entry) + await old.commit(snapshot, [UtteranceAppended(_entry("stale owner"))]) assert await db.scalar(select(Utterance.id).where(Utterance.session_id == session_id)) is None await db.execute(delete(AuditLog).where(AuditLog.object_id == str(session_id))) await db.execute(delete(Session).where(Session.id == session_id)) @@ -681,23 +733,21 @@ async def test_expired_backend_lease_is_atomically_claimed_and_restored(db, scen async def test_self_assessment_and_audit_commit_atomically(db, scenario): - from app.session.journal import DbJournal - trainee = await repo.ensure_trainee(db, f"reflection-{uuid4().hex[:12]}") session = await repo.create_session( db, scenario_id=scenario.id, mode="training", trainee_id=trainee.id, owner_login="reflection-teacher", ) - journal = DbJournal(async_sessionmaker(db.bind, expire_on_commit=False)) + pg = PostgresSessionStore(async_sessionmaker(db.bind, expire_on_commit=False)) + state = _state(session.id, scenario, trainee_id=trainee.id) class InvalidText: def __len__(self): return 12 # Let the DB reject the staged assessment, and verify its audit rolls back too. - assert not await journal.self_assessment( - session.id, ["q_address"], InvalidText(), datetime.now(timezone.utc) - ) + with pytest.raises(Exception): + await pg.commit(state, [SelfAssessed(["q_address"], InvalidText(), datetime.now(timezone.utc))]) assert await db.scalar(select(SelfAssessment.id).where( SelfAssessment.session_id == session.id )) is None @@ -705,9 +755,9 @@ async def test_self_assessment_and_audit_commit_atomically(db, scenario): AuditLog.action == "self_assessment.submit", AuditLog.object_id == str(session.id) )) is None - assert await journal.self_assessment( - session.id, ["q_address"], "адрес уточнил поздно", datetime.now(timezone.utc) - ) + await pg.commit(state, [SelfAssessed( + ["q_address"], "адрес уточнил поздно", datetime.now(timezone.utc), + )]) assessment = await db.scalar(select(SelfAssessment).where( SelfAssessment.session_id == session.id )) diff --git a/backend/tests/test_dds_desk_checkpoint.py b/backend/tests/test_dds_desk_checkpoint.py index d70ac9f..b0e551b 100644 --- a/backend/tests/test_dds_desk_checkpoint.py +++ b/backend/tests/test_dds_desk_checkpoint.py @@ -23,6 +23,7 @@ from app.session import timers as timers_module from app.session.checkpoint import dump_state, load_state from app.session.dds import prepare_handoff_queue, prepare_queue from app.session.hub import hub +from app.session.store import MemorySessionStore from app.session.state import SessionState from app.session.timers import now_utc @@ -30,6 +31,12 @@ LIBRARY = Path(__file__).resolve().parents[2] / "scenarios" _events = TypeAdapter(StationToServer) +async def finish_in_operation(state): + """finish() пишет оценку в commit операции, завершившей занятие.""" + async with hub.operation(state.session_id): + await finish_module.finish(state.session_id, state) + + @pytest.fixture(autouse=True) def offline_finish(monkeypatch): class NoCoaching: @@ -40,7 +47,7 @@ def offline_finish(monkeypatch): return NoCoaching() monkeypatch.setattr(finish_module, "coach", no_coach) - monkeypatch.setattr(hub, "journal", None) + monkeypatch.setattr(hub, "store", MemorySessionStore()) def fire(): @@ -150,7 +157,7 @@ def test_call_checkpoint_keeps_dispatcher_work_on_shared_timers(): assert snapshot.queue_cards == [] state.ended_at = now_utc() - asyncio.run(finish_module.finish(state.session_id, state)) + asyncio.run(finish_in_operation(state)) crew = next(item for item in state.score["metrics"] if item["key"] == "dds_crew") assert crew["passed"] and crew["fact"] == card.crew_selected @@ -189,7 +196,7 @@ def test_finish_scores_every_card_without_switching_active(name): unscored = {card.card_id for card in state.desk.cards.values()} state.ended_at = now_utc() - asyncio.run(finish_module.finish(state.session_id, state)) + asyncio.run(finish_in_operation(state)) assert state.desk.active is active assert {record.card_id for record in state.desk.completed} == unscored diff --git a/backend/tests/test_dds_desk_scores.py b/backend/tests/test_dds_desk_scores.py index af7c05b..f6d1412 100644 --- a/backend/tests/test_dds_desk_scores.py +++ b/backend/tests/test_dds_desk_scores.py @@ -24,6 +24,7 @@ from app.scenarios.loader import load_file from app.session import finish as finish_module from app.session.dds import prepare_handoff_queue, prepare_queue from app.session.hub import hub +from app.session.store import MemorySessionStore from app.session.state import SessionState LIBRARY = Path(__file__).resolve().parents[2] / "scenarios" @@ -32,6 +33,12 @@ COMMENT = "Основание: доклад старшего.\nСведения: _events = TypeAdapter(StationToServer) +async def finish_in_operation(state): + """finish() пишет оценку в commit операции, завершившей занятие.""" + async with hub.operation(state.session_id): + await finish_module.finish(state.session_id, state) + + @pytest.fixture(autouse=True) def offline_finish(monkeypatch): class NoCoaching: @@ -42,7 +49,7 @@ def offline_finish(monkeypatch): return NoCoaching() monkeypatch.setattr(finish_module, "coach", no_coach) - monkeypatch.setattr(hub, "journal", None) + monkeypatch.setattr(hub, "store", MemorySessionStore()) def scenario(name: str): @@ -51,7 +58,13 @@ def scenario(name: str): def station(state: SessionState, **payload) -> None: hub.register(state) - asyncio.run(station_ws._handle(state.session_id, state, _events.validate_python(payload))) + event = _events.validate_python(payload) + + async def command(): + async with hub.operation(state.session_id): + await station_ws._handle(state.session_id, state, event) + + asyncio.run(command()) def active_id(state: SessionState): @@ -167,7 +180,7 @@ def call_with_live_dispatcher() -> SessionState: state.dispatch() work_card(state) state.ended_at = state.dispatched_at - asyncio.run(finish_module.finish(state.session_id, state)) + asyncio.run(finish_in_operation(state)) return state diff --git a/backend/tests/test_dds_exercise.py b/backend/tests/test_dds_exercise.py index 4b42f1a..519c2f2 100644 --- a/backend/tests/test_dds_exercise.py +++ b/backend/tests/test_dds_exercise.py @@ -17,6 +17,7 @@ from app.domain.timers import TimerCode from app.main import app from app.scenarios import store from app.session.hub import hub +from app.session.store import MemorySessionStore @pytest.fixture @@ -42,11 +43,10 @@ def client(monkeypatch): optional_session_override, ) monkeypatch.setattr(sessions_http, "audit_required", audit_override) - monkeypatch.setattr("app.api.ws.control.audit", audit_override) try: with TestClient(app) as test_client: test_client.post("/api/auth/dev-token") - hub.journal = None + hub.store = MemorySessionStore() yield test_client finally: get_settings.cache_clear() @@ -995,3 +995,32 @@ def test_each_dds_card_uses_its_own_scenario_weights(): next(item.weight for item in second_record.metrics if item.key == "dds_primary") == 2.0 ) + + +def test_dds_station_command_and_finish_are_one_store_commit_each(client): + """Команда пульта ДДС и её подтверждение фиксируются одним commit.""" + from app.session.store import LessonEnded, ScoreCalculated + + session_id, control = start(client) + store = hub.store + try: + with client.websocket_connect(f"/ws/station/{session_id}") as station: + read_until(station, "card.received") + service = read_until(station, "station.state")["snapshot"]["services"][0] + before = len(store.commits) + command_id = str(uuid4()) + station.send_json({ + "type": "card.status", "service": service, "status": "accepted", + "comment": "Основание: профиль полномочий ДДС.\nСведения: карточка принята.", + "_command_id": command_id, + }) + read_until(station, "command.ack") + assert len(store.commits) == before + 1 + assert command_id in store.snapshot(session_id)["processed_station_commands"] + + station.send_json({"type": "station.finish"}) + read_until(station, "score.ready") + assert len(store.commits) == before + 2 + assert [type(item) for item in store.commits[-1][1]] == [LessonEnded, ScoreCalculated] + finally: + control.__exit__(None, None, None) diff --git a/backend/tests/test_demo_no_db.py b/backend/tests/test_demo_no_db.py index 7da1c20..097c8cc 100644 --- a/backend/tests/test_demo_no_db.py +++ b/backend/tests/test_demo_no_db.py @@ -9,6 +9,7 @@ from app.api.auth import DEMO_TRAINEE_ID from app.config import get_settings from app.main import app from app.session.hub import hub +from app.session.store import MemorySessionStore def _wait_for(predicate, timeout=3): @@ -39,7 +40,7 @@ def test_demo_without_db_starts_dds_and_issues_owned_trainee_cookie(monkeypatch) health = client.get("/api/health").json() assert health["status"] == "ok" and health["demo_no_db"] is True assert health["scenarios_loaded"] > 0 - assert hub.journal is None + assert isinstance(hub.store, MemorySessionStore) assert client.post("/api/auth/login", json={ "login": "demo-instructor", "password": "demo" diff --git a/backend/tests/test_finish_card_status.py b/backend/tests/test_finish_card_status.py index 18568fb..f5f2370 100644 --- a/backend/tests/test_finish_card_status.py +++ b/backend/tests/test_finish_card_status.py @@ -12,11 +12,18 @@ from app.scenarios.loader import load_file from app.session import finish as finish_module from app.session.dds import prepare_handoff_queue, prepare_queue from app.session.hub import hub +from app.session.store import MemorySessionStore from app.session.state import SessionState LIBRARY = Path(__file__).resolve().parents[2] / "scenarios" +async def finish_in_operation(state): + """finish() пишет оценку в commit операции, завершившей занятие.""" + async with hub.operation(state.session_id): + await finish_module.finish(state.session_id, state) + + @pytest.fixture(autouse=True) def offline_finish(monkeypatch): class NoCoaching: @@ -27,7 +34,7 @@ def offline_finish(monkeypatch): return NoCoaching() monkeypatch.setattr(finish_module, "coach", no_coach) - monkeypatch.setattr(hub, "journal", None) + monkeypatch.setattr(hub, "store", MemorySessionStore()) def accept_active_card(state: SessionState) -> None: @@ -59,7 +66,7 @@ def test_dds_finish_reports_status_of_active_card_not_last_scored(): accept_active_card(state) active_status = state.station_snapshot().card - asyncio.run(finish_module.finish(state.session_id, state)) + asyncio.run(finish_in_operation(state)) assert_active_card_kept(state, active_id, active_status) @@ -83,6 +90,6 @@ def test_handoff_finish_reports_status_of_active_card_not_last_scored(): accept_active_card(state) active_status = state.station_snapshot().card - asyncio.run(finish_module.finish(state.session_id, state)) + asyncio.run(finish_in_operation(state)) assert_active_card_kept(state, active_id, active_status) diff --git a/backend/tests/test_report_export.py b/backend/tests/test_report_export.py index 25031a1..9a87484 100644 --- a/backend/tests/test_report_export.py +++ b/backend/tests/test_report_export.py @@ -8,6 +8,8 @@ from types import SimpleNamespace from uuid import uuid4 import pytest +from app.session.pg_store import PostgresSessionStore +from app.session.store import MemorySessionStore from fastapi import HTTPException from fastapi.testclient import TestClient @@ -129,7 +131,7 @@ def client(monkeypatch): with TestClient(app) as test_client: # These endpoint tests exercise the in-memory live-report path. Durable # report readiness is covered by the isolated PostgreSQL integration suite. - monkeypatch.setattr(sessions.hub, "journal", None) + monkeypatch.setattr(sessions.hub, "store", MemorySessionStore()) test_client.post("/api/auth/dev-token") yield test_client, state, report @@ -243,10 +245,23 @@ def test_archived_score_override_updates_json_exports_and_audit(monkeypatch): def add(self, row): self.added.append(row) + async def flush(self): + pass + + async def rollback(self): + pass + async def commit(self): self.commits += 1 + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return None + db = FakeDb() + monkeypatch.setattr(sessions.hub, "store", PostgresSessionStore(lambda: db)) monkeypatch.setattr(sessions.hub, "get", lambda session_id: None) monkeypatch.setattr(sessions, "_live", lambda session_id: (_ for _ in ()).throw( HTTPException(status_code=404, detail="session_not_found") diff --git a/backend/tests/test_session_access.py b/backend/tests/test_session_access.py index d0e6e16..b8d4fbf 100644 --- a/backend/tests/test_session_access.py +++ b/backend/tests/test_session_access.py @@ -15,12 +15,13 @@ from app.api.http import sessions from app.api.ws import call as call_ws from app.api.ws import observe as observe_ws from app.api.ws import station as station_ws -from app.domain.events import Exercise, SessionMode +from app.domain.events import Exercise, SessionMode, Speaker from app.domain.roles import Role from app.session.checkpoint import dump_state from app.session.hub import SessionHub from app.session.state import SessionState -from app.session.journal import DbJournal +from app.session.pg_store import PostgresSessionStore +from app.session.store import LessonRequest, UtteranceAppended def request() -> Request: @@ -28,7 +29,7 @@ def request() -> Request: @pytest.mark.asyncio -async def test_journal_write_failure_does_not_log_user_text(caplog): +async def test_store_commit_failure_does_not_log_user_text(caplog): private_text = "private caller address and medical detail" class FakeDb: @@ -38,11 +39,16 @@ async def test_journal_write_failure_does_not_log_user_text(caplog): async def __aexit__(self, *_args): return None - async def fail_write(_db, text): - raise RuntimeError(text) + def add(self, _row): + raise RuntimeError(private_text) - journal = DbJournal(lambda: FakeDb()) - await journal._write(fail_write, private_text) + state = SessionState( + session_id=uuid4(), scenario_id="case", scenario_title="Сессия", level="L1", + mode=SessionMode.TRAINING, + ) + entry = state.append(Speaker.CALLER, private_text) + with pytest.raises(RuntimeError): + await PostgresSessionStore(lambda: FakeDb()).commit(state, [UtteranceAppended(entry)]) assert private_text not in caplog.text assert "RuntimeError" in caplog.text @@ -480,10 +486,13 @@ async def test_unassigned_existing_session_is_claimed_once(): @pytest.mark.asyncio -async def test_journal_assigns_new_lesson_to_its_backend_node(monkeypatch): +async def test_store_opens_new_lesson_on_its_backend_node(monkeypatch): seen = {} trainee_id = uuid4() - row = SimpleNamespace(attempt=3, trainee_id=trainee_id) + row = SimpleNamespace( + id=uuid4(), attempt=3, trainee_id=trainee_id, backend_fencing_epoch=0, + backend_lease_until=None, started_at=None, live_state=None, checkpoint_at=None, + ) async def ensure(_db, **kwargs): seen.update(kwargs) @@ -501,18 +510,38 @@ async def test_journal_assigns_new_lesson_to_its_backend_node(monkeypatch): async def scalar(self, _query): return "01" - journal = DbJournal(lambda: FakeDb(), node_id="node-a") - result = await journal.start_lesson( - uuid4(), "case", "training", "Курсант", trainee_id, - owner_login="teacher-a", + def add(self, item): + added.append(item) + + async def commit(self): + committed.append(True) + + added, committed, identities = [], [], [] + + def build(identity): + identities.append(identity) + return SessionState( + session_id=row.id, scenario_id="case", scenario_title="Сессия", level="L1", + mode=SessionMode.TRAINING, + ) + + store = PostgresSessionStore(lambda: FakeDb(), node_id="node-a") + await store.open(LessonRequest( + session_id=row.id, scenario_id="case", mode="training", trainee_name="Курсант", + trainee_id=trainee_id, owner_login="teacher-a", backend_node_id=None, + ), build) + identity = identities[0] + assert (identity.attempt, identity.trainee_id, identity.service, identity.fencing_epoch) == ( + 3, trainee_id, "01", 1, ) - assert result == (3, trainee_id, "01", 1) assert seen["backend_node_id"] == "node-a" - assert callable(seen["before_commit"]), "session creation must carry its audit into commit" + assert seen["commit"] is False, "строка, аудит и снимок коммитятся вместе" + assert [item.action for item in added] == ["lesson.start"] + assert row.live_state is not None and committed == [True] @pytest.mark.asyncio -async def test_journal_restores_checkpoint_owned_by_this_node(): +async def test_store_restores_checkpoint_owned_by_this_node(): owner = "teacher-a" state = SessionState( session_id=uuid4(), @@ -562,8 +591,8 @@ async def test_journal_restores_checkpoint_owned_by_this_node(): return Rows([]) return Rows([row]) - journal = DbJournal(lambda: FakeDb(), node_id="node-a") - restored = await journal.restore_active() + store = PostgresSessionStore(lambda: FakeDb(), node_id="node-a") + restored = await store.restore_active() assert len(restored) == 1 assert restored[0].session_id == state.session_id assert restored[0].owner_login == owner diff --git a/backend/tests/test_session_checkpoint.py b/backend/tests/test_session_checkpoint.py index cf20b75..d5842e8 100644 --- a/backend/tests/test_session_checkpoint.py +++ b/backend/tests/test_session_checkpoint.py @@ -1,14 +1,11 @@ """Промежуточное состояние занятия переживает смену backend-процесса.""" -import asyncio from datetime import UTC, datetime, timedelta from pathlib import Path from uuid import uuid4 import pytest from app.domain.events import ( - CallStarted, - CommandAck, Exercise, LessonCriteria, Metric, @@ -21,7 +18,6 @@ from app.domain.timers import TimerCode from app.scenarios.loader import load_file from app.session.checkpoint import dump_state, load_state from app.session.dds import prepare_handoff_queue, prepare_queue -from app.session.hub import LEASE_FENCED_MESSAGE, SessionHub from app.session.state import DdsCardRecord, SessionState, now_utc LIBRARY = Path(__file__).resolve().parents[2] / "scenarios" @@ -228,69 +224,3 @@ def test_mixed_handoff_checkpoint_preserves_operator_card_and_generated_queue(): assert restored.desk.scenarios[1].id == second.id assert len(restored.desk.cards) == 2 assert restored.desk.active_id == state.desk.active_id - - -def test_checkpoint_storage_failure_fences_and_notifies_all_data_channels(): - class BrokenJournal: - async def checkpoint(self, _state): - raise OSError("simulated database partition") - - local_hub = SessionHub(journal=BrokenJournal()) - state = dds_state() - local_hub.register(state) - - with local_hub.observer(state.session_id) as observers, \ - local_hub.trainee(state.session_id) as trainee, \ - local_hub.station(state.session_id) as station: - async def failing_transition(): - async with local_hub.durable_transition(state.session_id): - local_hub.to_trainee( - state.session_id, CallStarted(started_at=now_utc()) - ) - assert trainee.empty(), "success event escaped before durable checkpoint" - - with pytest.raises(OSError, match="partition"): - asyncio.run(failing_transition()) - - assert state.lease_fenced - assert local_hub.get(state.session_id) is None - for queue in (observers, trainee, station): - event = queue.get_nowait() - assert event.message == LEASE_FENCED_MESSAGE - assert queue.empty(), "uncommitted success event leaked during fencing" - - -def test_durable_transition_publishes_event_only_after_checkpoint_commit(): - class CommitJournal: - committed = False - - async def checkpoint(self, _state): - await asyncio.sleep(0) - self.committed = True - - journal = CommitJournal() - local_hub = SessionHub(journal=journal) - state = dds_state() - local_hub.register(state) - - command_id = uuid4() - with local_hub.trainee(state.session_id) as trainee, \ - local_hub.station(state.session_id) as station: - async def transition(): - async with local_hub.durable_transition(state.session_id): - local_hub.to_trainee( - state.session_id, CallStarted(started_at=now_utc()) - ) - local_hub.to_station( - state.session_id, CommandAck(command_id=command_id) - ) - assert trainee.empty() - assert station.empty() - assert journal.committed - - asyncio.run(transition()) - event = trainee.get_nowait() - assert isinstance(event, CallStarted) - ack = station.get_nowait() - assert isinstance(ack, CommandAck) - assert ack.command_id == command_id diff --git a/backend/tests/test_session_store.py b/backend/tests/test_session_store.py new file mode 100644 index 0000000..3684034 --- /dev/null +++ b/backend/tests/test_session_store.py @@ -0,0 +1,70 @@ +"""Одна операция занятия — один commit хранилища, события только после него.""" + +import asyncio + +import pytest + +from app.domain.events import CallStarted, Speaker +from app.session.hub import LEASE_FENCED_MESSAGE, SessionHub +from app.session.state import now_utc +from app.session.store import MemorySessionStore, UtteranceAppended +from tests.test_session_checkpoint import dds_state + + +class FailingStore(MemorySessionStore): + async def commit(self, state, records=()): + raise OSError("simulated database partition") + + +def test_operation_commits_snapshot_and_records_once_then_publishes(): + store = MemorySessionStore() + local_hub = SessionHub(store=store) + state = local_hub.register(dds_state()) + sid = state.session_id + + with local_hub.trainee(sid) as trainee: + async def operation(): + async with local_hub.operation(sid): + entry = state.append(Speaker.OPERATOR, "Адрес?") + local_hub.record(sid, UtteranceAppended(entry)) + local_hub.to_trainee(sid, CallStarted(started_at=now_utc())) + assert trainee.empty(), "событие ушло до коммита" + assert store.commits == [] + + asyncio.run(operation()) + assert isinstance(trainee.get_nowait(), CallStarted) + + assert len(store.commits) == 1 + committed_id, records = store.commits[0] + assert committed_id == sid + assert [type(item) for item in records] == [UtteranceAppended] + assert store.snapshot(sid)["transcript"][0]["text"] == "Адрес?" + + +def test_failed_commit_drops_events_and_fences_session(): + local_hub = SessionHub(store=FailingStore()) + state = local_hub.register(dds_state()) + sid = state.session_id + + with local_hub.observer(sid) as observers, \ + local_hub.trainee(sid) as trainee, \ + local_hub.station(sid) as station: + async def operation(): + async with local_hub.operation(sid): + local_hub.to_trainee(sid, CallStarted(started_at=now_utc())) + + with pytest.raises(OSError, match="partition"): + asyncio.run(operation()) + + assert state.lease_fenced + assert local_hub.get(sid) is None + for queue in (observers, trainee, station): + assert queue.get_nowait().message == LEASE_FENCED_MESSAGE + assert queue.empty(), "событие несостоявшейся операции утекло" + + +def test_record_outside_operation_is_rejected(): + local_hub = SessionHub(store=MemorySessionStore()) + state = local_hub.register(dds_state()) + with pytest.raises(RuntimeError): + local_hub.record(state.session_id, UtteranceAppended(state.append(Speaker.OPERATOR, "x"))) diff --git a/backend/tests/test_station.py b/backend/tests/test_station.py index 70c6010..e029f76 100644 --- a/backend/tests/test_station.py +++ b/backend/tests/test_station.py @@ -9,6 +9,7 @@ from fastapi.testclient import TestClient from app.api.http import sessions as sessions_http from app.main import app from app.session.hub import hub +from app.session.store import MemorySessionStore @pytest.fixture @@ -21,7 +22,7 @@ def client(monkeypatch): # Сокеты закрыты ролями (lct-23): тесты входят так же, # как `make lesson`, — через dev-token за флагом. test_client.post("/api/auth/dev-token") - hub.journal = None + hub.store = MemorySessionStore() yield test_client diff --git a/backend/tests/test_ws.py b/backend/tests/test_ws.py index 18fae38..fc1a767 100644 --- a/backend/tests/test_ws.py +++ b/backend/tests/test_ws.py @@ -18,6 +18,7 @@ from app.api.http import sessions as sessions_http from app.main import app from app.scenarios import store from app.session.hub import hub +from app.session.store import MemorySessionStore @pytest.fixture @@ -30,7 +31,7 @@ def client(monkeypatch): # Сокеты закрыты ролями (lct-23): тесты входят так же, # как `make lesson`, — через dev-token за флагом. test_client.post("/api/auth/dev-token") - hub.journal = None # тесты не пишут в БД: проверяется поведение каналов + hub.store = MemorySessionStore() # тесты не пишут в БД: in-memory adapter хранилища yield test_client @@ -469,46 +470,93 @@ def test_report_shows_missed_questions_and_self_assessment_gap(client): assert report["hints_used"], "использованные подсказки попадают в разбор" -def test_instructor_correction_keeps_the_automatic_score(client, postgres_access): - from app.db.base import get_sessionmaker - from app.session.journal import DbJournal +@pytest.mark.parametrize("path", ["http", "ws"]) +def test_score_override_is_one_operation_for_http_and_ws(client, postgres_access, path): + """Правка с пульта и из отчёта даёт одну и ту же строку Score и разбор. - # Unlike the websocket-only cases above, this regression exercises the - # durable HTTP correction endpoint against a real PostgreSQL score row. - journal = DbJournal(get_sessionmaker()) + Раньше WS-путь не трогал `full_report` живой сессии: отчёт показывал + старую оценку до перезагрузки. + """ + from sqlalchemy import select + + from app.db.base import get_sessionmaker + from app.db.models import Score + from app.session.pg_store import PostgresSessionStore + + comment = "связь рвалась не по вине курсанта" + hub.store = PostgresSessionStore(get_sessionmaker()) try: - with lesson(client) as (session_id, _): + with lesson(client) as (session_id, control), \ + client.websocket_connect(f"/ws/observe/{session_id}") as observer: + state = hub.get(session_id) with client.websocket_connect(f"/ws/call/{session_id}") as trainee: trainee.send_json({"type": "call.answer"}) trainee.send_json({"type": "call.hangup"}) - wait_for(lambda: hub.get(session_id).score is not None) - # The websocket-only fixture deliberately disables journalling; - # persist its computed result before testing the HTTP override. - state = hub.get(session_id) - client.portal.call( - journal.start_lesson, - session_id, - state.scenario_id, - state.mode.value, - state.trainee_name, - state.trainee_id, - "dev", - ) - client.portal.call(journal.score, session_id, state.score["score_auto"], state.score) - + # score.ready уходит только после commit оценки. + read_until(observer, "score.ready", limit=50) auto = client.get(f"/api/sessions/{session_id}/report").json()["score_auto"] - response = client.patch( - f"/api/sessions/{session_id}/report", - json={"score_final": 80.0, "comment": "связь рвалась не по вине курсанта"}, - ) - assert response.status_code == 200, response.text - corrected = response.json() - finally: - hub.journal = None - assert corrected["score_final"] == 80.0 - assert corrected["score_auto"] == auto, "автооценка должна сохраниться рядом" - assert corrected["overridden_by"] == "dev" + if path == "http": + response = client.patch( + f"/api/sessions/{session_id}/report", + json={"score_final": 80.0, "comment": comment}, + ) + assert response.status_code == 200, response.text + else: + control.send_json({ + "type": "score.override", "session_id": str(session_id), + "verdict": "80", "comment": comment, + }) + read_until(observer, "score.ready", limit=50) + report = client.get(f"/api/sessions/{session_id}/report").json() + + async def score_row(): + async with get_sessionmaker()() as db: + return await db.scalar(select(Score).where(Score.session_id == session_id)) + + row = client.portal.call(score_row) + finally: + hub.store = MemorySessionStore() + + corrected = {"score_final": 80.0, "overridden_by": "dev", "override_comment": comment} + assert {key: report[key] for key in corrected} == corrected + assert report["score_auto"] == auto, "автооценка должна сохраниться рядом" + assert (row.score_auto, row.score_final, row.overridden_by, row.override_comment) == ( + auto, 80.0, "dev", comment, + ) + assert {key: row.report[key] for key in corrected} == corrected + assert {key: row.report["full_report"][key] for key in corrected} == corrected + assert row.report["full_report"]["score_auto"] == auto + assert {key: state.score["full_report"][key] for key in corrected} == corrected + + +def test_each_lesson_operation_is_exactly_one_store_commit(client): + """Раньше реплика, подсказка и финиш шли отдельной транзакцией журнала + и ещё одним checkpoint — между ними возможен частичный коммит.""" + from app.session.store import HintRecorded, LessonEnded, LessonStarted, NoteAdded, ScoreCalculated + + store = hub.store + with lesson(client) as (session_id, control): + assert len(store.commits) == 1, "запуск: строка занятия и снимок вместе" + + def commits_after(action) -> list: + before = len(store.commits) + action() + wait_for(lambda: len(store.commits) > before) + time.sleep(0.05) + assert len(store.commits) == before + 1 + return [type(item) for item in store.commits[-1][1]] + + with client.websocket_connect(f"/ws/call/{session_id}") as trainee: + assert commits_after(lambda: trainee.send_json({"type": "call.answer"})) == [LessonStarted] + assert commits_after(lambda: trainee.send_json({"type": "hint.request"})) == [HintRecorded] + assert commits_after(lambda: trainee.send_json({"type": "call.hangup"})) == [ + LessonEnded, ScoreCalculated, + ] + assert commits_after(lambda: control.send_json({ + "type": "instructor_note.add", "transcript_ref": "u1", "text": "поздно спросил адрес", + })) == [NoteAdded] + assert all(committed_id == session_id for committed_id, _ in store.commits) def test_ws_score_override_rejects_other_session_and_invalid_value(client): @@ -538,13 +586,10 @@ def test_ws_score_override_rejects_other_session_and_invalid_value(client): assert state.score["overridden_by"] == "dev" -def test_ws_score_override_is_not_applied_when_atomic_persistence_fails(client): - class FailedJournal: - async def score_override(self, *_args): - return False - - async def checkpoint(self, *_args): - return None +def test_ws_score_override_is_not_confirmed_when_commit_fails(client): + class FailedStore(MemorySessionStore): + async def commit(self, state, records=()): + raise OSError("simulated database partition") with lesson(client) as (session_id, control): state = hub.get(session_id) @@ -553,15 +598,18 @@ def test_ws_score_override_is_not_applied_when_atomic_persistence_fails(client): trainee.send_json({"type": "call.hangup"}) wait_for(lambda: state.score is not None) - auto = state.score["score_auto"] - hub.journal = FailedJournal() - control.send_json({ - "type": "score.override", "session_id": str(session_id), - "verdict": "85", "comment": "manual review", - }) - time.sleep(0.1) - assert state.score["score_auto"] == auto - assert "score_final" not in state.score + with client.websocket_connect(f"/ws/observe/{session_id}") as observer: + observer.receive_json() + hub.store = FailedStore() + control.send_json({ + "type": "score.override", "session_id": str(session_id), + "verdict": "85", "comment": "manual review", + }) + message = read_until(observer, "error") + + assert "переподключитесь" in message["message"], "без записи правка не подтверждается" + assert state.lease_fenced + assert hub.get(session_id) is None def test_soft_directive_changes_how_the_caller_sounds(client): diff --git a/backend/tests/test_ws_ownership.py b/backend/tests/test_ws_ownership.py index abf955b..1c00a24 100644 --- a/backend/tests/test_ws_ownership.py +++ b/backend/tests/test_ws_ownership.py @@ -14,6 +14,7 @@ from app.domain.roles import Role from app.main import app from app.scenarios import store from app.session.hub import hub +from app.session.store import MemorySessionStore from app.session.state import SessionState @@ -21,7 +22,7 @@ from app.session.state import SessionState def client(): with TestClient(app) as test_client: test_client.post("/api/auth/dev-token") - hub.journal = None + hub.store = MemorySessionStore() yield test_client @@ -153,8 +154,8 @@ def test_control_command_checkpoint_failure_returns_fencing_error(monkeypatch): mode="training", owner_login="lease-owner", ) - class BrokenJournal: - async def checkpoint(self, _state): + class BrokenStore(MemorySessionStore): + async def commit(self, _state, _records=()): raise OSError("simulated database partition") class OneCommandSocket: @@ -176,8 +177,8 @@ def test_control_command_checkpoint_failure_returns_fencing_error(monkeypatch): who = Principal(login="lease-owner", full_name="Преподаватель", role=Role.INSTRUCTOR) socket = OneCommandSocket() - old_journal = hub.journal - hub.journal = BrokenJournal() + old_store = hub.store + hub.store = BrokenStore() hub.register(state) monkeypatch.setattr(control_module, "websocket_origin_allowed", lambda _ws: True) monkeypatch.setattr(control_module, "principal_of", lambda _ws: who) @@ -193,7 +194,7 @@ def test_control_command_checkpoint_failure_returns_fencing_error(monkeypatch): assert "переподключитесь" in event.message assert observer_queue.empty(), "uncommitted controller event leaked to observers" finally: - hub.journal = old_journal + hub.store = old_store hub.drop(session_id)