From 10bb5109b6d5e4cb0bd3cd5fd154da0eeba1c11f Mon Sep 17 00:00:00 2001 From: owenfisher Date: Sun, 2 Aug 2026 19:53:45 -0700 Subject: [PATCH] fix(voice-gateway): bind each turn to the question it was spoken against MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_handle_turn` read the current question from session state at handling time, but a turn is handled on a worker task behind a persistence round trip and a stretch of audio playback. A patient who kept talking through that window committed a second turn while the session was still on the first, and the graph had moved on by the time it was handled — so the words were filed under a question the patient had not yet been asked. Reproduced against the stub: two utterances spoken while only `knee.onset` was on the floor were persisted as `knee.onset: answered` and `knee.severity: answered`. The second is a well-formed severity answer, which is what made the mis-filing silent — it passes the answer gate and reads as a real reply. `CommittedTurn` now carries the question id, stamped by the detector at commit time. A turn whose question the session has already left is recorded as LATE_UTTERANCE against the question it was actually spoken against: the patient said it, so it stays in the record, but it does not drive a transition for a question they have not heard. Also in the turn loop, both found in the same review: * `aclose()` tracked only the latest supervisor task, but `_recover()` runs *inside* a supervisor and spawns its replacement — so closing during a reconnect left the old one retrying against a channel that was gone. All session-spawned tasks are now tracked and cancelled together, which also keeps the fire-and-forget barge-in task from being collected mid-flight. * The committed-turn queue was unbounded and fed from the socket reader, which cannot block. Bounded, with shedding logged rather than silent. Co-Authored-By: Claude --- .../src/voice_gateway/contracts/models.py | 8 ++ .../src/voice_gateway/graph/mock_graph.py | 13 +++ .../src/voice_gateway/turn/detector.py | 35 ++++++- .../src/voice_gateway/turn/session.py | 89 +++++++++++++++--- .../tests/test_intake_session.py | 94 +++++++++++++++++++ 5 files changed, 223 insertions(+), 16 deletions(-) diff --git a/services/voice-gateway/src/voice_gateway/contracts/models.py b/services/voice-gateway/src/voice_gateway/contracts/models.py index afe4d65..986b7a9 100644 --- a/services/voice-gateway/src/voice_gateway/contracts/models.py +++ b/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) diff --git a/services/voice-gateway/src/voice_gateway/graph/mock_graph.py b/services/voice-gateway/src/voice_gateway/graph/mock_graph.py index 0b7a32c..ba6afa8 100644 --- a/services/voice-gateway/src/voice_gateway/graph/mock_graph.py +++ b/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) diff --git a/services/voice-gateway/src/voice_gateway/turn/detector.py b/services/voice-gateway/src/voice_gateway/turn/detector.py index 909c8fd..104abcc 100644 --- a/services/voice-gateway/src/voice_gateway/turn/detector.py +++ b/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() diff --git a/services/voice-gateway/src/voice_gateway/turn/session.py b/services/voice-gateway/src/voice_gateway/turn/session.py index e8cc68b..3d35c61 100644 --- a/services/voice-gateway/src/voice_gateway/turn/session.py +++ b/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 ----------------------------------------------------------- diff --git a/services/voice-gateway/tests/test_intake_session.py b/services/voice-gateway/tests/test_intake_session.py index 494de86..1f8aa8f 100644 --- a/services/voice-gateway/tests/test_intake_session.py +++ b/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() + ]