Bind each turn to the question it was spoken against

services/voice-gateway/src/voice_gateway/contracts/models.py

@@ -36,6 +36,14 @@ class TurnOutcome(StrEnum):
     REPROMPT = "reprompt"
     RECONNECT_REASK = "reconnect_reask"
 
+    LATE_UTTERANCE = "late_utterance"
+    """Spoken against a question the session has already moved past.
+
+    Kept rather than dropped: the patient said it, so it is part of the record.
+    Distinguished rather than merged into ANSWERED, because a clinician reading
+    the transcript needs to know these words did not decide the next question.
+    """
+
 
 class QuestionNode(BaseModel):
     model_config = ConfigDict(frozen=True)

services/voice-gateway/src/voice_gateway/graph/mock_graph.py

@@ -100,6 +100,19 @@ def all_prerenderable_nodes() -> tuple[QuestionNode, ...]:
     return (SESSION_GREETING, *MOCK_GRAPH.nodes, SESSION_CLOSING, RECONNECT_NOTICE)
 
 
+def node_by_id(question_id: str) -> QuestionNode | None:
+    """Resolve a question id back to its node, or None if it is not a question.
+
+    Unknown ids are a miss rather than an error: a turn can carry an id from a
+    graph version that has since been swapped, and losing one utterance is a
+    better failure than dropping the session.
+    """
+    try:
+        return MOCK_GRAPH.node(question_id)
+    except KeyError:
+        return None
+
+
 def next_node(current_id: str) -> QuestionNode | None:
     """Deterministic transition. No LLM, no network, no I/O."""
     node = MOCK_GRAPH.node(current_id)

services/voice-gateway/src/voice_gateway/turn/detector.py

@@ -49,6 +49,18 @@ class CommittedTurn:
     start_seconds: float
     end_seconds: float
 
+    question_id: str | None = None
+    """Which question was on the floor when these words were spoken.
+
+    Stamped here, at commit time, and not read from session state later. A turn
+    is handled on a worker task behind a persistence round trip and a stretch of
+    audio playback, so by the time it is handled the session has often moved on
+    — and a patient who keeps talking through that window would otherwise have
+    their words filed under a question they had not yet been asked. `None` means
+    nothing was on the floor, which happens if the patient talks over the
+    greeting.
+    """
+
 
 @dataclass(slots=True)
 class _Accumulator:
@@ -78,13 +90,16 @@ def is_empty(self) -> bool:
         # whitespace-only buffer is not a turn here either.
         return not self.text.strip()
 
-    def drain(self, reason: CommitReason, time_offset: float) -> CommittedTurn:
+    def drain(
+        self, reason: CommitReason, time_offset: float, question_id: str | None
+    ) -> CommittedTurn:
         turn = CommittedTurn(
             transcript=self.text.strip(),
             reason=reason,
             segment_count=self.segment_count,
             start_seconds=(self.start_seconds or 0.0) + time_offset,
             end_seconds=self.end_seconds + time_offset,
+            question_id=question_id,
         )
         self.reset()
         return turn
@@ -114,6 +129,11 @@ def __init__(self, on_commit: Callable[[CommittedTurn], None]) -> None:
         # provenance stays continuous across a reconnect.
         self._time_offset = 0.0
 
+        # Additive: the question currently on the floor. Set by the session at
+        # the moment it starts listening, read at the moment a turn commits.
+        # See CommittedTurn.question_id for why it is captured this early.
+        self._question_id: str | None = None
+
     # --- 05 §4, verbatim ---------------------------------------------------
 
     def on_results(self, msg: Results) -> None:
@@ -135,10 +155,11 @@ def on_utterance_end(self, msg: UtteranceEnd) -> None:
     # --- additive ----------------------------------------------------------
 
     def _commit(self, reason: CommitReason) -> None:
-        turn = self._buffer.drain(reason, self._time_offset)
+        turn = self._buffer.drain(reason, self._time_offset, self._question_id)
         logger.debug(
-            "turn committed via %s: %d segments, %.2fs-%.2fs",
+            "turn committed via %s for %s: %d segments, %.2fs-%.2fs",
             reason,
+            turn.question_id,
             turn.segment_count,
             turn.start_seconds,
             turn.end_seconds,
@@ -153,6 +174,14 @@ def time_offset(self) -> float:
     def time_offset(self, value: float) -> None:
         self._time_offset = value
 
+    @property
+    def question_id(self) -> str | None:
+        return self._question_id
+
+    @question_id.setter
+    def question_id(self, value: str | None) -> None:
+        self._question_id = value
+
     @property
     def has_pending_audio(self) -> bool:
         return not self._buffer.is_empty()

services/voice-gateway/src/voice_gateway/turn/session.py

@@ -20,6 +20,7 @@
 import logging
 import time
 import uuid
+from collections.abc import Coroutine
 from typing import Protocol
 
 from voice_gateway.audio.cache import CacheKey, PrerenderedAudioCache
@@ -42,6 +43,7 @@
     SESSION_CLOSING,
     SESSION_GREETING,
     next_node,
+    node_by_id,
 )
 from voice_gateway.persistence.client import TurnPersistenceClient
 from voice_gateway.telemetry import traced
@@ -53,6 +55,9 @@
 _PCM_CHUNK_BYTES = 8192
 MAX_REPROMPTS = 2
 
+MAX_QUEUED_TURNS = 64
+"""Committed turns awaiting the worker. See `IntakeSession._turns`."""
+
 
 @traced("intake.turn")
 def _trace_turn(
@@ -116,13 +121,25 @@ def __init__(
         self._connection_index = 0
         self._started_at = time.monotonic()
 
-        self._turns: asyncio.Queue[CommittedTurn] = asyncio.Queue()
-        self._detector = TurnDetector(self._turns.put_nowait)
+        # Bounded. A turn is handled behind a persistence round trip and a
+        # stretch of playback, so a patient who never stops talking commits
+        # faster than the worker drains — and the queue is fed from the socket
+        # reader, which cannot block. The cap is far above any real exchange;
+        # reaching it means something is wrong, and shedding is better than
+        # growing without limit.
+        self._turns: asyncio.Queue[CommittedTurn] = asyncio.Queue(maxsize=MAX_QUEUED_TURNS)
+        self._detector = TurnDetector(self._enqueue_turn)
         self._stt: ListenConnection | None = None
         self._tts: SpeakConnection | None = None
 
         self._worker: asyncio.Task[None] | None = None
-        self._supervisor: asyncio.Task[None] | None = None
+        # Every task the session spawns, not just the latest one. `_recover`
+        # runs *inside* a supervisor task and starts its replacement, so a
+        # single `_supervisor` slot loses the old task while it is still
+        # running — and `aclose()` would then leave it retrying against a
+        # channel that is already gone. A strong reference also keeps
+        # fire-and-forget tasks from being collected mid-flight.
+        self._tasks: set[asyncio.Task[None]] = set()
         self._stopping = asyncio.Event()
         self._finished = asyncio.Event()
         self._speaking = False
@@ -176,13 +193,32 @@ async def _connect_stt(self) -> None:
         # clinical record needs per-answer timing for provenance. Rebase.
         self._detector.time_offset = time.monotonic() - self._started_at
 
-        self._supervisor = asyncio.create_task(self._supervise(connection))
+        self._spawn(self._supervise(connection))
+
+    def _spawn(self, coro: Coroutine[None, None, None]) -> asyncio.Task[None]:
+        task = asyncio.create_task(coro)
+        self._tasks.add(task)
+        task.add_done_callback(self._tasks.discard)
+        return task
+
+    def _enqueue_turn(self, turn: CommittedTurn) -> None:
+        """Called from the socket reader, so it must not block and must not raise."""
+        try:
+            self._turns.put_nowait(turn)
+        except asyncio.QueueFull:
+            logger.error(
+                "committed-turn queue is full (%d); shedding a turn for %s",
+                MAX_QUEUED_TURNS,
+                turn.question_id,
+            )
 
     async def aclose(self) -> None:
         self._stopping.set()
-        for task in (self._supervisor, self._worker):
+        for task in (*self._tasks, self._worker):
             if task is not None:
                 task.cancel()
+        for task in (*self._tasks, self._worker):
+            if task is not None:
                 with contextlib.suppress(asyncio.CancelledError):
                     await task
         if self._stt is not None:
@@ -242,7 +278,7 @@ def _on_speech_started(self, msg: SpeechStarted) -> None:
         # The primitives, wired: SpeechStarted from STT pairs with Aura's
         # `Clear` to drop queued audio (05 §7). Off by default — see
         # Settings.barge_in_enabled for why the policy is not ours to pick.
-        asyncio.create_task(self._barge_in())
+        self._spawn(self._barge_in())
 
     async def _barge_in(self) -> None:
         if self._tts is not None:
@@ -323,8 +359,27 @@ async def _turn_worker(self) -> None:
                 logger.exception("turn handling failed")
 
     async def _handle_turn(self, turn: CommittedTurn) -> None:
-        node = self._current
+        node = node_by_id(turn.question_id) if turn.question_id else None
         if node is None:
+            # Nothing was on the floor when these words were spoken — the
+            # patient talked over the greeting or the closing.
+            logger.info("discarding a turn committed with no question on the floor")
+            return
+
+        if self._current is None or node.id != self._current.id:
+            # The patient kept talking while this turn was still behind a
+            # persistence round trip and a stretch of playback, so we have
+            # already moved on. The words belong to the question they were
+            # spoken against — record them there, truthfully, but do not let
+            # them drive a transition for a question the patient has not heard.
+            logger.info("late utterance for %s; already on %s", node.id, self._current)
+            await self._record_turn(
+                node,
+                outcome=TurnOutcome.LATE_UTTERANCE,
+                answer=self._answer_for(node, turn),
+                commit_reason=turn.reason.value,
+                tts_cache_hit=self._last_tts_cache_hit,
+            )
             return
 
         await self._channel.send_event(
@@ -354,7 +409,14 @@ async def _handle_turn(self, turn: CommittedTurn) -> None:
             await self._advance(node)
             return
 
-        answer = IntakeAnswer(
+        await self._record_turn(
+            node, outcome=TurnOutcome.ANSWERED, answer=self._answer_for(node, turn),
+            commit_reason=turn.reason.value, tts_cache_hit=self._last_tts_cache_hit,
+        )
+        await self._advance(node)
+
+    def _answer_for(self, node: QuestionNode, turn: CommittedTurn) -> IntakeAnswer:
+        return IntakeAnswer(
             question_id=node.id,
             prompt_version=node.prompt_version,
             answer_type=node.answer_type,
@@ -366,17 +428,15 @@ async def _handle_turn(self, turn: CommittedTurn) -> None:
                 connection_index=self._connection_index,
             ),
         )
-        await self._record_turn(
-            node, outcome=TurnOutcome.ANSWERED, answer=answer,
-            commit_reason=turn.reason.value, tts_cache_hit=self._last_tts_cache_hit,
-        )
-        await self._advance(node)
 
     async def _advance(self, node: QuestionNode) -> None:
         self._reprompts = 0
         nxt = next_node(node.id)
         self._current = nxt
         if nxt is None:
+            # Nothing is on the floor any more; anything said over the closing
+            # is not an answer to the last question.
+            self._detector.question_id = None
             await self._speak(SESSION_CLOSING)
             await self._channel.send_event({"type": "session_complete"})
             self._finished.set()
@@ -385,6 +445,9 @@ async def _advance(self, node: QuestionNode) -> None:
 
     async def _ask(self, node: QuestionNode) -> None:
         await self._speak(node)
+        # This is the moment the question is on the floor, so it is the moment
+        # anything committed from here on belongs to.
+        self._detector.question_id = node.id
         await self._channel.send_event({"type": "listening", "question_id": node.id})
 
     # --- output -----------------------------------------------------------

services/voice-gateway/tests/test_intake_session.py

@@ -273,3 +273,97 @@ async def test_questions_are_served_from_the_pre_rendered_cache(
     assert channel.audio_bytes > 0
     assert all(t.tts_cache_hit for t in persistence.received)
     assert len(channel.of_type("audio_begin")) >= 2  # greeting + first question
+
+
+# --- a patient who keeps talking ------------------------------------------
+
+
+async def test_words_are_filed_under_the_question_they_were_spoken_against(
+    session: IntakeSession, channel: FakeChannel, persistence: RecordingPersistence
+) -> None:
+    """The failure this guards against put a wrong answer in a clinical record.
+
+    A turn is handled on a worker task, behind a persistence round trip and a
+    stretch of audio playback. A patient who carries on talking through that
+    window commits a second turn while the session is still on the first — and
+    if the question is read from session state at *handling* time rather than
+    stamped at *commit* time, those words are filed under a question the
+    patient has not yet been asked.
+    """
+    await session.start()
+    await channel.wait_for_listening_on("knee.onset")
+
+    # Both spoken while only knee.onset has been asked. The second is a
+    # well-formed severity answer, which is what makes the mis-filing silent:
+    # it passes the answer gate and reads as a real reply.
+    await session.push_text("it started about three weeks ago")
+    await session.push_text("i would say a seven")
+
+    await channel.wait_for_listening_on("knee.severity")
+    await channel.wait_for(
+        lambda events: any(
+            e.get("type") == "listening" and e.get("question_id") == "knee.severity"
+            for e in events
+        )
+    )
+
+    severity = [t for t in persistence.received if t.question_id == "knee.severity"]
+    assert not [t for t in severity if t.outcome is TurnOutcome.ANSWERED], (
+        "a question the patient had not heard yet was recorded as answered"
+    )
+
+    late = [t for t in persistence.received if t.outcome is TurnOutcome.LATE_UTTERANCE]
+    assert [t.question_id for t in late] == ["knee.onset"]
+    assert late[0].answer is not None
+    assert late[0].answer.transcript == "i would say a seven"
+
+
+async def test_a_late_utterance_does_not_advance_the_graph(
+    session: IntakeSession, channel: FakeChannel, persistence: RecordingPersistence
+) -> None:
+    """Keeping the words is right; letting them drive the next question is not.
+    The patient never heard severity, so severity must still be asked."""
+    await session.start()
+    await channel.wait_for_listening_on("knee.onset")
+
+    await session.push_text("it started about three weeks ago")
+    await session.push_text("i would say a seven")
+    await channel.wait_for_listening_on("knee.severity")
+
+    asked = [e["question_id"] for e in channel.of_type("listening")]
+    assert asked == ["knee.onset", "knee.severity"]
+
+
+# --- shutdown --------------------------------------------------------------
+
+
+async def test_closing_leaves_nothing_running(
+    settings: Settings,
+    channel: FakeChannel,
+    cache: PrerenderedAudioCache,
+    persistence: RecordingPersistence,
+) -> None:
+    """Reaching into `_tasks` because task lifetime has no public surface, and
+    this is the property that matters: reconnect supervision spawns its own
+    replacement from *inside* a supervisor task, so tracking only the latest
+    one leaves the previous still retrying against a channel that is gone.
+    """
+    import asyncio
+
+    s = IntakeSession(
+        settings=settings,
+        channel=channel,
+        tokens=EphemeralTokenProvider(settings),
+        cache=cache,
+        persistence=persistence,
+        session_id="closing",
+    )
+    await s.start()
+    await channel.wait_for_listening_on("knee.onset")
+    await s.aclose()
+
+    assert not [t for t in s._tasks if not t.done()]
+    assert s._worker is not None and s._worker.done()
+    assert not [
+        t for t in asyncio.all_tasks() if t is not asyncio.current_task() and not t.done()
+    ]