Bind each turn to the question it was spoken against - #23
Conversation
`_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 <noreply@anthropic.com>
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
There was a problem hiding this comment.
Pull request overview
Fixes a race where turns could be persisted under the wrong question by stamping each committed turn with the question id at commit time, and introducing a late_utterance outcome when handling lags behind graph progression. It also hardens the session loop by tracking all spawned tasks for cancellation and bounding the committed-turn queue to prevent unbounded growth under sustained input.
Changes:
- Add
CommittedTurn.question_idcaptured byTurnDetectorat commit time, and use it in_handle_turn()to bind turns to the question they were spoken against (includingLATE_UTTERANCEhandling). - Track all session-spawned tasks in a set and cancel them on shutdown; bound the committed-turn queue with logging on shedding.
- Add tests covering turn/question binding, “late utterance does not advance,” and “no tasks left running after close.”
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| services/voice-gateway/src/voice_gateway/turn/detector.py | Stamp committed turns with question_id at commit time and expose detector question_id property. |
| services/voice-gateway/src/voice_gateway/turn/session.py | Use stamped question_id during handling, add LATE_UTTERANCE path, bound turn queue, and track/cancel spawned tasks. |
| services/voice-gateway/src/voice_gateway/contracts/models.py | Introduce TurnOutcome.LATE_UTTERANCE to preserve but distinguish late speech. |
| services/voice-gateway/src/voice_gateway/graph/mock_graph.py | Add node_by_id() helper for resolving question ids to nodes. |
| services/voice-gateway/tests/test_intake_session.py | Add tests for correct binding, non-advancing late utterances, and shutdown task cleanup. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| 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}) |
| node, | ||
| outcome=TurnOutcome.LATE_UTTERANCE, | ||
| answer=self._answer_for(node, turn), | ||
| commit_reason=turn.reason.value, | ||
| tts_cache_hit=self._last_tts_cache_hit, |
| 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 |
| 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: |
|
| Filename | Overview |
|---|---|
| services/voice-gateway/src/voice_gateway/turn/session.py | Core of the fix: bounded queue, _tasks set replacing single _supervisor slot, _spawn helper, _enqueue_turn, and _handle_turn now using the stamped question_id for routing. aclose() cancel/await loops each re-evaluate *self._tasks, which is functionally identical since no await separates them — but tasks spawned during the second loop's awaits (reconnect mid-flight) could slip past cancellation. |
| services/voice-gateway/src/voice_gateway/turn/detector.py | Adds _question_id field to TurnDetector with a getter/setter, threads it through _Accumulator.drain() into CommittedTurn. Clean and minimal. |
| services/voice-gateway/src/voice_gateway/contracts/models.py | Adds LATE_UTTERANCE to TurnOutcome with clear docstring explaining its clinical semantics. No issues. |
| services/voice-gateway/src/voice_gateway/graph/mock_graph.py | Adds node_by_id() with graceful KeyError handling for stale graph versions. Consistent with the existing next_node() pattern; the MOCK_GRAPH singleton usage is a pre-existing constraint. |
| services/voice-gateway/tests/test_intake_session.py | Three new tests added. The late-utterance binding tests directly cover the bug. The shutdown test checks _tasks is empty after aclose(), but only for a clean shutdown — not for shutdown during an active reconnect, which is the scenario the _tasks-set refactor was designed to fix. |
Sequence Diagram
sequenceDiagram
participant P as Patient
participant Det as TurnDetector
participant Q as CommittedTurn Queue
participant W as Worker Task
participant S as Session State
participant DB as Persistence
Note over S: _ask(knee.onset)
S->>Det: "question_id = knee.onset"
P->>Det: speaks three weeks ago
Det->>Q: "CommittedTurn(question_id=knee.onset)"
P->>Det: speaks i would say a seven
Det->>Q: "CommittedTurn(question_id=knee.onset)"
W->>Q: "dequeue turn 1 (question_id=knee.onset)"
W->>S: "_handle_turn node=onset current=onset ANSWERED"
W->>DB: record ANSWERED for knee.onset
S->>S: "_advance current = knee.severity"
S->>Det: "question_id = knee.severity"
W->>Q: "dequeue turn 2 (question_id=knee.onset)"
W->>S: "_handle_turn node=onset current=severity LATE"
W->>DB: record LATE_UTTERANCE for knee.onset
Note over W: no graph transition
Note over S: _ask(knee.severity) - asks patient again
Comments Outside Diff (1)
-
services/voice-gateway/src/voice_gateway/turn/session.py, line 215-229 (link)Task spawned during second loop's awaits won't be cancelled or awaited
The two
(*self._tasks, self._worker)expressions are evaluated at the same event-loop turn (noawaitbetween them), so their snapshots are identical — that part is fine. The gap is in the second loop: eachawait taskyields to the event loop, and a supervisor spawned by_recover()during that window (i.e.,connection.open()returned and_spawn(new_supervisor)ran synchronously before theCancelledErrorpropagated to the nextawait) is added toself._tasksafter the snapshot was taken.aclose()never cancels or awaits it.In practice the new supervisor self-terminates quickly:
_stoppingis already set whenconnection.wait_closed()returns, and it returns immediately afteraclose()closesself._stt. So no harm actually occurs — but the invariant "nothing running afteraclose()returns" is only upheld by timing luck, andtest_closing_leaves_nothing_runningdoesn't exercise the reconnect path where this can fire. Adding a test that callsaclose()while_recover()is mid-flight (e.g. aftersimulate_connection_drop()and before the reconnect completes) would pin this down explicitly.Prompt To Fix With AI
This is a comment left during a code review. Path: services/voice-gateway/src/voice_gateway/turn/session.py Line: 215-229 Comment: **Task spawned during second loop's awaits won't be cancelled or awaited** The two `(*self._tasks, self._worker)` expressions are evaluated at the same event-loop turn (no `await` between them), so their snapshots are identical — that part is fine. The gap is in the second loop: each `await task` yields to the event loop, and a supervisor spawned by `_recover()` during that window (i.e., `connection.open()` returned and `_spawn(new_supervisor)` ran synchronously before the `CancelledError` propagated to the next `await`) is added to `self._tasks` after the snapshot was taken. `aclose()` never cancels or awaits it. In practice the new supervisor self-terminates quickly: `_stopping` is already set when `connection.wait_closed()` returns, and it returns immediately after `aclose()` closes `self._stt`. So no harm actually occurs — but the invariant "nothing running after `aclose()` returns" is only upheld by timing luck, and `test_closing_leaves_nothing_running` doesn't exercise the reconnect path where this can fire. Adding a test that calls `aclose()` while `_recover()` is mid-flight (e.g. after `simulate_connection_drop()` and before the reconnect completes) would pin this down explicitly. --- For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.
Prompt To Fix All With AI
### Issue 1
services/voice-gateway/src/voice_gateway/turn/session.py:215-229
**Task spawned during second loop's awaits won't be cancelled or awaited**
The two `(*self._tasks, self._worker)` expressions are evaluated at the same event-loop turn (no `await` between them), so their snapshots are identical — that part is fine. The gap is in the second loop: each `await task` yields to the event loop, and a supervisor spawned by `_recover()` during that window (i.e., `connection.open()` returned and `_spawn(new_supervisor)` ran synchronously before the `CancelledError` propagated to the next `await`) is added to `self._tasks` after the snapshot was taken. `aclose()` never cancels or awaits it.
In practice the new supervisor self-terminates quickly: `_stopping` is already set when `connection.wait_closed()` returns, and it returns immediately after `aclose()` closes `self._stt`. So no harm actually occurs — but the invariant "nothing running after `aclose()` returns" is only upheld by timing luck, and `test_closing_leaves_nothing_running` doesn't exercise the reconnect path where this can fire. Adding a test that calls `aclose()` while `_recover()` is mid-flight (e.g. after `simulate_connection_drop()` and before the reconnect completes) would pin this down explicitly.
### Issue 2
services/voice-gateway/tests/test_intake_session.py:302-308
The `wait_for_listening_on` call above already waits for this exact event, so the second `wait_for` resolves immediately and can be removed.
```suggestion
await channel.wait_for_listening_on("knee.severity")
```
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.Reviews (1): Last reviewed commit: "fix(voice-gateway): bind each turn to th..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| 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 | ||
| ) | ||
| ) |
There was a problem hiding this comment.
wait_for_listening_on call above already waits for this exact event, so the second wait_for resolves immediately and can be removed.
| 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 | |
| ) | |
| ) | |
| await channel.wait_for_listening_on("knee.severity") |
Prompt To Fix With AI
This is a comment left during a code review.
Path: services/voice-gateway/tests/test_intake_session.py
Line: 302-308
Comment:
The `wait_for_listening_on` call above already waits for this exact event, so the second `wait_for` resolves immediately and can be removed.
```suggestion
await channel.wait_for_listening_on("knee.severity")
```
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
Stack — merge bottom to top
slice/voice→prod)The bug
_handle_turnread 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 keeps talking through that window commits a second turn while the session is still on the first — and by the time it is handled, the graph has moved on.Reproduced against the stub. Two utterances, both spoken while only
knee.onsethad 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. A clinician reading that record sees a severity rating the patient was never asked for.
The fix
CommittedTurncarries the question id, stamped by the detector at commit time rather than read from mutable state later. A turn whose question the session has already left is recorded asLATE_UTTERANCEagainst the question it was actually spoken against.Kept rather than dropped — the patient said it, so it belongs in the record. Distinguished rather than merged into
ANSWERED— those words must not drive a transition for a question the patient has not heard.Same run, after:
Also in the turn loop
Both found in the same review, both in
session.py: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, able to emitfatalafter close. 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.Tests
Three added; 129 pass. Two cover the binding directly — that a late utterance is filed under the right question, and that it does not advance the graph. The third asserts nothing is left running after
aclose().Verified live in the browser as well as in tests; the before/after above is from the persistence stub.
🤖 Generated with Claude Code