lct-39: пауза останавливает действия и голос

backend/app/api/http/sessions.py

@@ -228,7 +228,10 @@ async def active(
                 states[state.session_id] = state
 
     for state in states.values():
-        elapsed = (max(0, int((now - state.started_at).total_seconds()))
+        paused_ms = state.total_paused_ms
+        if state.paused and state.paused_at is not None:
+            paused_ms += max(0, int((now - state.paused_at).total_seconds() * 1000))
+        elapsed = (max(0, int((now - state.started_at).total_seconds() - paused_ms / 1000))
                    if state.started_at else 0)
         station = state.station_snapshot() if state.exercise is Exercise.DDS else None
         queue = station.queue_cards if station else []

backend/app/api/ws/call.py

@@ -235,6 +235,9 @@ async def _handle(session_id: UUID, state, event) -> None:
                     code=ErrorKind.INTERNAL, message="Не удалось получить ответ заявителя. Попробуйте ещё раз.",
                 ))
                 return
+            if state.paused:
+                # Ответ модели мог закончиться уже после команды преподавателя.
+                return
             caller_entry = state.append(Speaker.CALLER, line.text, line.mood)
             hub.to_trainee(session_id, CallerUtterance(
                 utterance_id=uuid4(), text=line.text,
@@ -399,10 +402,9 @@ async def call(ws: WebSocket, session_id: UUID) -> None:
     _who, state = entered
 
     with hub.trainee(session_id) as queue:
-        if state.paused:
-            # Переподключение не ждёт следующего `session.pause`: баннер должен
-            # быть на экране сразу, раз сервер уже отклоняет команды (lct-39).
-            hub.to_trainee(session_id, SessionPaused(paused=True))
+        # Каждый новый канал получает серверное состояние, включая снятую паузу:
+        # клиент мог потерять событие resume во время переподключения.
+        queue.put_nowait(SessionPaused(paused=state.paused))
         if state.exercise is Exercise.CARD:
             from app.api.ws.control import card_briefing
 

backend/app/session/finish.py

@@ -12,7 +12,7 @@ import logging
 import time
 from uuid import UUID
 
-from app.domain.events import CallEnded, CallEndReason, Exercise, ScoreReady, SessionEnded
+from app.domain.events import CallEnded, CallEndReason, Exercise, ScoreReady, SessionEnded, SessionPaused
 from app.domain.statuses import ServiceStatus, current
 from app.domain.timers import TimerCode
 from app.scoring.ai_coach import coach
@@ -27,6 +27,7 @@ from app.session.score import ScoringSnapshot, score_session, scoring_scenario
 from app.session.state import DdsCardRecord, DdsLiveCard, now_utc
 from app.session.store import (
     LessonEnded,
+    LessonResumed,
     ScoreArchived,
     ScoreCalculated,
     ScoreOverridden,
@@ -181,7 +182,16 @@ async def end_session(session_id: UUID, state, reason: CallEndReason) -> None:
     """
     if state.ended:
         return
+    was_paused = state.paused
+    paused_ms_before = state.total_paused_ms
     ended_at = state.end(reason)
+    if was_paused:
+        hub.record(session_id, LessonResumed(
+            at=ended_at, author="система", role="system",
+            paused_ms=state.total_paused_ms - paused_ms_before,
+        ))
+        hub.broadcast(session_id, SessionPaused(paused=False))
+        hub.to_station(session_id, SessionPaused(paused=False))
     if state.exercise is Exercise.CARD and state.dispatched_card is None:
         state.on_event("card.end")
     # Норматив отработки фиксируется событием, а не текущим значением часов.

backend/app/session/state.py

@@ -197,6 +197,8 @@ class SessionState(PersistedSession):
             card.timers.pause(now)
         self.paused = True
         self.paused_at = now_utc()
+        if self.voice is not None:
+            self.voice.pause()
 
     def resume(self) -> None:
         """Снять паузу: таймеры продолжают с прежней отметки, очередь сдвигается на простой."""

backend/app/voice/pipeline.py

@@ -86,6 +86,8 @@ class VoiceSession:
     # ── вход: кадры микрофона ──
 
     def feed(self, frame: bytes) -> None:
+        if self.state.paused:
+            return
         for event in self._vad.push(frame):
             if isinstance(event, SpeechStarted) and self.speaking:
                 self.barge_in()
@@ -104,6 +106,8 @@ class VoiceSession:
         первая фраза «Алло! Помогите!» — ровно та, которую оператор перебивает
         чаще всего, — не гасилась вовсе.
         """
+        if self.state.paused:
+            return asyncio.create_task(asyncio.sleep(0))
         if self.speaking:
             self._reply.cancel()
         self._reply = asyncio.create_task(self.say(text, mood))
@@ -119,6 +123,17 @@ class VoiceSession:
             self.send_event(TtsCancel(utterance_id=self._utterance_id, reason="barge_in"))
         log.info("сессия %s: перебивание", self.session_id)
 
+    def pause(self) -> None:
+        """Прервать текущую реплику и убрать речь, накопленную до паузы."""
+        if self.speaking:
+            self._reply.cancel()
+        if self._utterance_id is not None:
+            self.send_event(TtsCancel(utterance_id=self._utterance_id, reason="director"))
+            self._utterance_id = None
+        while not self._queue.empty():
+            self._queue.get_nowait()
+        self._vad.reset()
+
     async def close(self) -> None:
         for task in (self._reply, self._worker):
             if task is not None:
@@ -129,6 +144,8 @@ class VoiceSession:
     async def _work(self) -> None:
         while True:
             audio, ended_at = await self._queue.get()
+            if self.state.paused:
+                continue
             # Новая фраза оператора, пока звонящий ещё говорит, — тоже перебивание.
             if self.speaking:
                 self.barge_in()
@@ -143,6 +160,8 @@ class VoiceSession:
         timing = TurnTiming()
         started = time.monotonic()
         text = await self.models.transcribe(audio)
+        if self.state.paused:
+            return
         timing.stt_ms = (time.monotonic() - started) * 1000
         if not text:
             return
@@ -155,6 +174,8 @@ class VoiceSession:
 
         started = time.monotonic()
         line = await self._caller_line(text)
+        if self.state.paused:
+            return
         timing.caller_ms = (time.monotonic() - started) * 1000
         await self.say(line.text, line.mood, ended_at=ended_at, timing=timing)
 
@@ -171,6 +192,8 @@ class VoiceSession:
         self, text: str, mood: Mood, *, ended_at: float | None = None, timing: TurnTiming | None = None
     ) -> None:
         """Произнести реплику: событие с текстом, звук по предложениям, ожидание конца."""
+        if self.state.paused:
+            return
         self._utterance_id = utterance_id = uuid4()
         entry = self.state.append(Speaker.CALLER, text, mood)
         self.send_event(CallerUtterance(utterance_id=utterance_id, text=text, at=entry.at, mood=mood))
@@ -186,6 +209,8 @@ class VoiceSession:
                 pcm = await self._first_sentence(sentence, mood, ended_at, timing)
             else:
                 pcm = await self.synthesize(sentence)
+            if self.state.paused:
+                return
             if index == 0 and timing is not None:
                 timing.tts_first_ms = (time.monotonic() - synth_started) * 1000
             if not pcm:
@@ -205,6 +230,7 @@ class VoiceSession:
         # Ждём, пока курсант дослушает: перебивание в это время отменит задачу.
         await asyncio.sleep(max(0.0, playback_ends - time.monotonic()))
         self.send_event(TtsEnd(utterance_id=utterance_id))
+        self._utterance_id = None
 
     async def _first_sentence(
         self, sentence: str, mood: Mood, ended_at: float, timing: TurnTiming | None
@@ -212,17 +238,25 @@ class VoiceSession:
         """Первое предложение с филлером: если к секунде после конца фразы звука
         ещё нет, звонящий «переспрашивает», а ответ встанет в очередь за ним."""
         synthesis = asyncio.ensure_future(self.synthesize(sentence))
-        remaining = FILLER_AFTER_S - (time.monotonic() - ended_at)
-        if remaining > 0:
-            done, _ = await asyncio.wait({synthesis}, timeout=remaining)
-            if done:
-                return synthesis.result()
-        filler = await self.synthesize(FILLERS.get(mood, FILLERS[Mood.PANIC]))
-        if filler:
-            self.send_audio(filler)
-            if timing is not None:
-                timing.filler = True
-        return await synthesis
+        try:
+            remaining = FILLER_AFTER_S - (time.monotonic() - ended_at)
+            if remaining > 0:
+                done, _ = await asyncio.wait({synthesis}, timeout=remaining)
+                if done:
+                    return synthesis.result()
+            filler = await self.synthesize(FILLERS.get(mood, FILLERS[Mood.PANIC]))
+            if filler and not self.state.paused:
+                self.send_audio(filler)
+                if timing is not None:
+                    timing.filler = True
+            return await synthesis
+        finally:
+            if not synthesis.done():
+                synthesis.cancel()
+                try:
+                    await synthesis
+                except asyncio.CancelledError:
+                    pass
 
     async def synthesize(self, text: str) -> bytes:
         return await cached_synthesize(self.models, text)

backend/app/voice/vad.py

@@ -71,6 +71,17 @@ class StreamingVad:
     def in_speech(self) -> bool:
         return self._in_speech
 
+    def reset(self) -> None:
+        """Не склеивать фрагменты речи до и после паузы занятия."""
+        self._state.fill(0)
+        self._context.fill(0)
+        self._pending = np.zeros(0, dtype=np.float32)
+        self._preroll.clear()
+        self._speech.clear()
+        self._voiced_ms = 0
+        self._silence_ms = 0
+        self._in_speech = False
+
     def _probability(self, window: np.ndarray) -> float:
         frame = np.concatenate([self._context, window])[None, :]
         output, self._state = self._session.run(

backend/tests/test_session_pause.py

@@ -9,6 +9,7 @@ import asyncio
 import time
 from datetime import timedelta
 from pathlib import Path
+from types import SimpleNamespace
 from uuid import uuid4
 
 import pytest
@@ -16,7 +17,8 @@ from fastapi.testclient import TestClient
 
 from app.api.http import sessions as sessions_http
 from app.api.ws import control as control_ws
-from app.domain.events import CallEndReason, Exercise, LessonCriteria, SessionMode
+from app.api.ws import call as call_ws
+from app.domain.events import CallEndReason, Exercise, LessonCriteria, SessionMode, SessionPaused, TtsCancel, TtsEnd
 from app.domain.statuses import ServiceStatus
 from app.domain.timers import TimerCode
 from app.main import app
@@ -163,6 +165,12 @@ def test_checkpoint_round_trip_keeps_pause_without_adding_downtime(clock):
     assert restored_timer.elapsed_ms == card.timers.timers[TimerCode.DDS_ACK].elapsed_ms
     assert restored_timer.started_at is None
 
+    # После takeover пауза всё ещё действует; лишь явное возобновление
+    # продолжает норматив с пяти секунд до прерывания.
+    restored.resume()
+    clock.advance(10)
+    assert restored_timer.current_ms(clock.now) == 15_000
+
 
 def test_pause_and_resume_are_recorded_for_audit(monkeypatch):
     scenario = load_file(LIBRARY / "fire-apartment-l2.yaml", LIBRARY)
@@ -250,15 +258,30 @@ def test_ending_session_while_paused_finalizes_pause_and_unblocks_report(clock,
     idle = timedelta(seconds=45)
     monkeypatch.setattr("app.session.state.now_utc", lambda: base + idle)
 
+    events = []
+
     async def run():
-        async with hub.operation(state.session_id):
-            await end_session(state.session_id, state, CallEndReason.INSTRUCTOR)
+        with hub.trainee(state.session_id) as queue:
+            async with hub.operation(state.session_id):
+                await end_session(state.session_id, state, CallEndReason.INSTRUCTOR)
+            async with hub.operation(state.session_id):
+                await call_ws._handle(
+                    state.session_id, state,
+                    SimpleNamespace(type="self_assessment.submit", missed=[], comment="Проверил"),
+                )
+            while not queue.empty():
+                events.append(queue.get_nowait())
 
     asyncio.run(run())
 
     assert state.ended
     assert state.paused is False
     assert state.total_paused_ms >= 30_000
+    assert state.self_assessed
+    assert state.score is not None
+    assert state.score["full_report"]["total_paused_ms"] >= 30_000
+    assert any(isinstance(event, SessionPaused) and not event.paused for event in events)
+    assert any(isinstance(record, LessonResumed) for _sid, batch in store.commits for record in batch)
 
 
 def test_dds_work_overdue_survives_pause():
@@ -280,6 +303,27 @@ def test_dds_work_overdue_survives_pause():
     assert overdue is True
 
 
+def test_active_registry_keeps_existing_overdue_during_pause(http_client, clock):
+    app.dependency_overrides[sessions_http.optional_session] = lambda: None
+    state, card = _dds_state_with_active_card(clock)
+    state.owner_login = "dev"
+    timer = card.timers.timers.setdefault(TimerCode.DDS_WORK, Timer(code=TimerCode.DDS_WORK))
+    card.timers.limits[TimerCode.DDS_WORK] = 1_000
+    timer.start(clock.now)
+    clock.advance(5)
+    state.pause()
+    hub._sessions[state.session_id] = state
+    try:
+        response = http_client.get("/api/sessions/active")
+        assert response.status_code == 200
+        row = next(item for item in response.json() if item["session_id"] == str(state.session_id))
+        assert row["dds_work_overdue_cards"] == 1
+        assert row["paused"] is True
+    finally:
+        hub._sessions.pop(state.session_id, None)
+        app.dependency_overrides.pop(sessions_http.optional_session, None)
+
+
 @pytest.fixture
 def http_client(monkeypatch):
     async def audit_in_memory(*_args, **_kwargs):
@@ -347,3 +391,62 @@ def test_reconnect_to_call_replays_paused_banner(http_client):
                 message["type"] == "session.paused" and message["paused"]
                 for message in messages
             )
+
+        control.send_json({"type": "session.resume"})
+        _wait(lambda: not hub.get(session_id).paused)
+        with http_client.websocket_connect(f"/ws/call/{session_id}") as trainee:
+            message = trainee.receive_json()
+            assert message == {"type": "session.paused", "paused": False}
+
+
+@pytest.mark.asyncio
+async def test_pause_cancels_inflight_voice_and_clears_buffer(monkeypatch):
+    from app.voice import pipeline
+
+    class FakeVad:
+        def __init__(self, *_args, **_kwargs):
+            self.reset_count = 0
+
+        def reset(self):
+            self.reset_count += 1
+
+        def push(self, _frame):
+            return []
+
+    monkeypatch.setattr(pipeline, "StreamingVad", FakeVad)
+    state, _card = _dds_state_with_active_card(lambda: 1_000.0)
+    events = []
+    audio = []
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    class Models:
+        vad_path = None
+
+    voice = pipeline.VoiceSession(
+        state.session_id, state, Models(), events.append, lambda _event: None, audio.append,
+    )
+    state.voice = voice
+
+    async def slow_synthesis(_text):
+        started.set()
+        await release.wait()
+        return b"\x00\x00" * 2_400
+
+    monkeypatch.setattr(voice, "synthesize", slow_synthesis)
+    try:
+        speaking = voice.speak("Алло, помогите!", state.persona.mood if state.persona else pipeline.Mood.PANIC)
+        await asyncio.wait_for(started.wait(), timeout=1)
+        voice._queue.put_nowait((b"old", time.monotonic()))
+        state.pause()
+        release.set()
+        with pytest.raises(asyncio.CancelledError):
+            await speaking
+        await asyncio.sleep(0)
+        assert any(isinstance(event, TtsCancel) and event.reason == "director" for event in events)
+        assert not any(isinstance(event, TtsEnd) for event in events)
+        assert audio == []
+        assert voice._queue.empty()
+        assert voice._vad.reset_count == 1
+    finally:
+        await voice.close()