Add LiveKit and ElevenLabs streaming voice output - #1
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
📝 WalkthroughWalkthroughThis PR adds a complete browser voice-session pipeline. The backend handles WebSocket admission, speech input, streamed text, LiveKit audio, interruptions, limits, errors, and cleanup. The frontend manages microphone capture, validated events, LiveKit playback, session state, and controls. ChangesBackend voice pipeline
Frontend voice session
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Browser
participant VoiceSession
participant WebSocket
participant Backend
participant LiveKit
Browser->>VoiceSession: Start voice session
VoiceSession->>LiveKit: Unlock and connect audio
VoiceSession->>WebSocket: Connect and send client.ready
Browser->>WebSocket: Send WebM/Opus microphone chunks
WebSocket->>Backend: Forward media and control events
Backend-->>WebSocket: Send transcript, text, state, and error events
Backend->>LiveKit: Publish streamed assistant PCM audio
LiveKit-->>Browser: Deliver assistant audio
Browser->>WebSocket: Send playback.hard_cut
Backend->>LiveKit: Stop current generation audio
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 13
Note
Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.
🟡 Minor comments (16)
backend/tests/unit/test_turn_manager.py-12-12 (1)
12-12: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winAwait the scheduled task instead of a fixed delay.
The 20 ms delay can expire before the event loop runs the debounce task under CI load. Capture
manager.pending_taskafter the updates and await it before the assertions.Proposed fix
- await asyncio.sleep(.02) + task = manager.pending_task + assert task is not None + await task🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/tests/unit/test_turn_manager.py` at line 12, Replace the fixed asyncio.sleep delay in the test with awaiting the scheduled debounce task: after applying the updates, capture manager.pending_task and await it before performing assertions, preserving the existing test flow.frontend/app/hooks/useLiveKitAudio.ts-187-187 (1)
187-187: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winFilter
TrackUnsubscribedbefore detaching audio.An unrelated unsubscribe currently detaches the assistant audio and sets
subscriberReadytofalse. Match the event participant, publication, and track against the expected and attached references before detaching. Add a regression test.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@frontend/app/hooks/useLiveKitAudio.ts` at line 187, Update onTrackUnsubscribed in useLiveKitAudio to ignore unrelated TrackUnsubscribed events by validating the event participant, publication, and track against the expected and currently attached references before calling detachAudioRef.current(). Preserve subscriberReady changes only for the matching assistant audio, and add a regression test covering an unrelated unsubscribe.backend/app/providers/deepgram.py-27-27 (1)
27-27: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winSplit the compound statements.
Ruff reports E701 on both lines. Put the
raiseand cancellation statements on separate lines.Proposed fix
- if error is not None: raise RuntimeError("Deepgram listening failed") from error + if error is not None: + raise RuntimeError("Deepgram listening failed") from error ... - if not self.listening_task.done(): self.listening_task.cancel() + if not self.listening_task.done(): + self.listening_task.cancel()Also applies to: 44-44
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/providers/deepgram.py` at line 27, Split the compound statements in the Deepgram listening error handling and cancellation branches into separate lines, keeping the existing RuntimeError chaining and cancellation behavior unchanged. Update both affected branches in the relevant Deepgram provider function.Source: Linters/SAST tools
backend/app/audio/text_buffer.py-41-46 (1)
41-46: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winEmit a sentence that ends at the current buffer boundary.
Line 41 rejects a terminator when it is the last buffered character. A value such as
"Hello."then waits until a later delta orflush(). Accept an end-of-buffer terminator unless the abbreviation check rejects it.Proposed fix
- if index + 1 >= len(self._buffer) or not self._buffer[index + 1].isspace(): + if index + 1 < len(self._buffer) and not self._buffer[index + 1].isspace(): continue🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/audio/text_buffer.py` around lines 41 - 46, Update the sentence-boundary logic around the buffer scan to accept a terminator at the current buffer boundary, rather than continuing solely because index + 1 is outside _buffer. Preserve the abbreviation and numeric-decimal exclusions in the existing prefix check, so a final character such as “Hello.” emits immediately while rejected abbreviations still wait.backend/app/providers/groq.py-21-24 (1)
21-24: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winCall
response.close()in thefinallyblock.groq==1.0.0returnsAsyncStreamforstream=True, andAsyncStreamexposes asyncclose(), notaclose(). The current code skips cleanup after early cancellation or return.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/providers/groq.py` around lines 21 - 24, Update the cleanup in the finally block to invoke the async close() method exposed by the response returned from the streaming Groq request, rather than looking up aclose. Preserve cleanup for early cancellation and return, and await the close operation when available.backend/app/api/audio_ws.py-275-296 (1)
275-296: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winRuff reports E701/E702 errors on these lines.
Ruff flags multiple statements on one line at lines 275, 279, 281, 285, 287, 290, 291, 293, 296, 331, 333, 335, 341, 343, 344, and 345, and also at line 218. If Ruff runs in CI with these rules enabled, the lint job fails. Split the compound statements, or confirm that
E701/E702are excluded in the Ruff configuration.Also applies to: 330-345
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/api/audio_ws.py` around lines 275 - 296, Resolve Ruff E701/E702 violations in the websocket handling flow around the message-processing block and the related lines near 218 and 330-345 by splitting compound statements and same-line control-flow statements into separate lines. Alternatively, update the Ruff configuration to explicitly exclude E701/E702 if that is the intended project policy, ensuring CI linting passes.Source: Linters/SAST tools
backend/tests/integration/test_audio_websocket.py-118-131 (1)
118-131: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winThe media assertion depends on task scheduling and can be flaky.
send_bytesonly enqueues the chunk intomedia_queue.media_senderforwards it in a separate task. Leaving thewebsocket_connectblock sends a disconnect, and the endpoint'sfinallyblock cancelsmedia_taskimmediately. Ifmedia_senderhas not been scheduled yet,connection.mediastays empty and the assertion at line 129 fails.Make the test wait for observable progress before closing the socket. For example, have
FakeDeepgramConnection.send_mediapush into a queue that the endpoint acknowledges, or read one more server message aftersend_bytesso the server loop advances.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/tests/integration/test_audio_websocket.py` around lines 118 - 131, Update the websocket integration test around send_bytes and the connection.media assertion to wait for observable server-side progress before leaving the websocket_connect block. Use an existing server acknowledgment/message path or add a test-only synchronization mechanism through FakeDeepgramConnection.send_media, ensuring the media chunk has been forwarded before the socket closes and the assertions run.backend/app/api/audio_ws.py-250-259 (1)
250-259: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winKeep a reference to the close task created in
observe.
asyncio.create_task(websocket.close(...))at line 256 is not stored. The event loop keeps only a weak reference, so the task can be garbage-collected before the close completes. The receive loop then stays blocked onwebsocket.receive()until the browser acts, which defeats the purpose of the fast close.Store the task in the existing
observer_taskslist, which thefinallyblock already cancels and gathers.🛡️ Proposed fix
def observe(task: asyncio.Task[object]) -> None: if task.cancelled() or worker_failure.done(): return error = task.exception() if error is not None: worker_failure.set_result(error) - asyncio.create_task(websocket.close(code=1011, reason="session_worker_failed")) + observer_tasks.append(asyncio.create_task(websocket.close(code=1011, reason="session_worker_failed")))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/api/audio_ws.py` around lines 250 - 259, Store the task created by websocket.close in the existing observer_tasks list within observe, rather than discarding the create_task result. Preserve the existing worker_failure handling and ensure the finally block can cancel and gather this close task.Source: Linters/SAST tools
backend/livekit.dev.yaml-3-11 (1)
3-11: 🔒 Security & Privacy | 🟡 Minor | ⚡ Quick winBind the development server to the loopback address.
The file combines
bind_addresses: 0.0.0.0with a well-known key pair (devkey/secret). Withnetwork_mode: hostindocker-compose.livekit.yml, any host on the same network can mint tokens and create rooms on this server. For a local-only server, bind to127.0.0.1.🛡️ Proposed change
bind_addresses: - - 0.0.0.0 + - 127.0.0.1If a phone or a second machine must reach the server during testing, keep
0.0.0.0and document that the key pair must be changed.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/livekit.dev.yaml` around lines 3 - 11, Update the bind_addresses setting in the development LiveKit configuration to use 127.0.0.1 instead of 0.0.0.0, keeping the local-only default while leaving the RTC and keys settings unchanged.backend/README.md-7-20 (1)
7-20: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winThe setup block cannot run as one sequence.
docker compose -f docker-compose.livekit.yml upruns in the foreground and blocks. Theuvicorncommand on line 14 and the frontend commands never run in that shell. Add-d, or state that each server needs its own terminal.📝 Proposed change
# Start a local LiveKit server (Linux host networking): -docker compose -f docker-compose.livekit.yml up -uvicorn app.main:app --host 0.0.0.0 --port 3300 +docker compose -f docker-compose.livekit.yml up -d +# Run the backend in this terminal, then use a new terminal for the frontend: +uvicorn app.main:app --host 0.0.0.0 --port 3300🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/README.md` around lines 7 - 20, Update the README setup sequence so the LiveKit Docker Compose command runs detached with the existing configuration, allowing the subsequent uvicorn and frontend commands to execute in the same shell.backend/app/providers/elevenlabs.py-36-40 (1)
36-40: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winClose the socket if the initial handshake message fails.
__aenter__connects and then sends the voice-settings message. If that send raises (ConnectionClosed, timeout, cancellation),__aenter__propagates the error and__aexit__never runs, because the context was never entered.self._socketstays open until garbage collection, which consumes an ElevenLabs concurrent stream slot.🛡️ Proposed fix
async def __aenter__(self) -> "ElevenLabsStream": self._socket = await connect(self.url, additional_headers={"xi-api-key": self.options.api_key}, max_queue=self.options.max_queue, max_size=self.options.max_size) - await self._send({"text": " ", "voice_settings": {"stability": .5, "similarity_boost": .75}}) - return self + try: + await self._send({"text": " ", "voice_settings": {"stability": .5, "similarity_boost": .75}}) + except BaseException: + await self.close() + raise + return self🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/providers/elevenlabs.py` around lines 36 - 40, Update ElevenLabsStream.__aenter__ to close self._socket when the initial _send handshake fails, then re-raise the original exception. Ensure cleanup covers connection, timeout, and cancellation failures while preserving normal entry behavior and returning self after a successful handshake.backend/app/main.py-52-55 (1)
52-55: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winDisable
voice_agentpropagation.
app.core.loggingsets thevoice_agentlogger toINFObeforecreate_app()runs, soINFOevents are not dropped. However,create_app()adds a handler and leaves propagation enabled, which can duplicate records when the root logger has a handler. Setvoice_logger.propagate = False.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/main.py` around lines 52 - 55, Update create_app so the voice_agent logger sets propagate to False after configuring its handler, preventing records from being duplicated through the root logger.backend/app/models/events.py-61-64 (1)
61-64: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winBuild
_CLIENT_ADAPTERfrom theClientEventalias.Line 63 restates the union
ClientReady | PlaybackHardCutinstead of reusingClientEventfrom line 62. The two definitions can diverge. If a third client event is added toClientEvent,parse_client_eventcontinues to reject it, and the failure appears as a protocol validation error far from this file.♻️ Proposed fix to remove the duplicated union
ServerEvent: TypeAlias = SessionReady | TurnStarted | AssistantTextDelta | AssistantState | AssistantInterrupted | ErrorEvent ClientEvent: TypeAlias = ClientReady | PlaybackHardCut -_CLIENT_ADAPTER = TypeAdapter(ClientReady | PlaybackHardCut) +_CLIENT_ADAPTER: TypeAdapter[ClientEvent] = TypeAdapter(ClientEvent) _SERVER_ADAPTER = TypeAdapter(ServerEvent)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/models/events.py` around lines 61 - 64, Update _CLIENT_ADAPTER to construct its TypeAdapter from the ClientEvent alias instead of repeating ClientReady | PlaybackHardCut, ensuring parse_client_event automatically supports every event included in ClientEvent.backend/app/audio/livekit_publisher.py-103-104 (1)
103-104: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winDerive the error message from
FRAME_BYTES.The check uses
FRAME_BYTES, but the message hardcodes480 bytes. IfFRAME_BYTESchanges, the message reports the wrong size to whoever reads the traceback.🐛 Proposed fix
if len(frame) != FRAME_BYTES: - raise ValueError("PCM frame must be exactly 480 bytes") + raise ValueError(f"PCM frame must be exactly {FRAME_BYTES} bytes, got {len(frame)}")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/audio/livekit_publisher.py` around lines 103 - 104, Update the validation error in the frame-length check within the publisher method to derive the reported byte count from FRAME_BYTES instead of hardcoding 480, keeping the existing ValueError behavior and condition unchanged.backend/tests/integration/test_livekit_publisher.py-82-87 (1)
82-87: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winReplace the scheduling yield with a deterministic wait.
Line 84 relies on one
await asyncio.sleep(0)to let several steps complete: the worker resumes fromget(), callscapture_frame, raises, the task finishes, and the done-callback records_worker_error. A single yield does not guarantee all of them.The test likely passes today because
writeitself starts withawait asyncio.sleep(0), which adds another yield. The outcome still depends on scheduling order and can flake under a different event loop policy. A flaky test on the error-propagation path is worse than no test, because it trains readers to re-run CI.Await the worker task instead.
💚 Proposed fix
await publisher.start_generation(1) + worker=publisher._worker await publisher.write(1, b'x' * FRAME_BYTES) - await asyncio.sleep(0) + await asyncio.wait({worker}) with pytest.raises(RuntimeError, match='LiveKit audio publishing failed'): await publisher.write(1, b'x' * FRAME_BYTES)
asyncio.waitdoes not re-raise the worker's exception, so the assertion below still exercises the deferred surfacing path.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/tests/integration/test_livekit_publisher.py` around lines 82 - 87, Replace the scheduling-only asyncio.sleep(0) in the publisher error-path test with a deterministic wait for the worker task, such as awaiting it through asyncio.wait without propagating its exception. Ensure the worker has completed and its done-callback has recorded _worker_error before the subsequent publisher.write assertion, while preserving the deferred RuntimeError behavior being tested.backend/app/services/admission.py-82-84 (1)
82-84: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winDo not rely on the pre-accept close code
With Uvicorn 0.46.0,
websocket.close()beforeaccept()returns an HTTP 403 handshake rejection. The browser does not receive4403or4429, and the frontend does not inspect close codes. Accept the socket before closing when the frontend must distinguish errors; otherwise document all handshake failures as generic.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/services/admission.py` around lines 82 - 84, Update reject_before_accept so it accepts the WebSocket before closing when the frontend must distinguish AdmissionRejected errors, preserving error.code and error.error_code in the close call; otherwise document that pre-accept handshake failures are generic and do not expose the configured close code.
🧹 Nitpick comments (23)
backend/tests/unit/test_admission.py (1)
9-10: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winResolve the reported Ruff E701 and E702 violations.
Split each executable statement onto its own physical line.
backend/tests/unit/test_admission.py#L9-L10: Split the conditional statements from the assignments.backend/tests/unit/test_admission.py#L20-L20: Split the context manager from the awaited call.backend/tests/unit/test_admission.py#L23-L23: Split the context manager from the awaited call.backend/tests/unit/test_audio_ws_buffers.py#L23-L23: Split the queue retrieval from the completion call.backend/tests/unit/test_events_config_logging.py#L11-L11: Splitbase.update(more)fromreturn base.backend/tests/unit/test_events_config_logging.py#L22-L22: Put theSettingscall on a separate line from the context manager.backend/tests/unit/test_events_config_logging.py#L24-L24: Put the parser call on a separate line from the context manager.backend/tests/unit/test_generation_runner.py#L11-L11: Expand the double initializer.backend/tests/unit/test_generation_runner.py#L19-L19: Expand the double initializer.backend/tests/unit/test_generation_runner.py#L23-L23: Split the append operation from the event set.backend/tests/unit/test_generation_runner.py#L31-L31: Expand the test setup assignments.backend/tests/unit/test_generation_runner.py#L42-L42: Expand the test setup assignments.backend/tests/unit/test_generation_runner.py#L60-L60: Expand the test setup assignments.backend/tests/unit/test_generation_runner.py#L82-L82: Expand the test setup assignments.backend/tests/unit/test_turn_manager.py#L21-L22: Put each cancellation and close call on its own line.backend/tests/unit/test_voice_session.py#L31-L31: Split the sink and clock initialization.backend/tests/unit/test_voice_session.py#L35-L35: Split the turn signal from the clock update.backend/tests/unit/test_voice_session.py#L37-L37: Put each close call on its own line.backend/tests/unit/test_voice_session.py#L42-L42: Expand the test setup assignments.backend/tests/unit/test_voice_session.py#L44-L44: Split generation start from the turn signal.backend/tests/unit/test_voice_session.py#L59-L59: Expand the test setup assignments.backend/tests/unit/test_voice_session.py#L62-L62: Split the turn signal from the clock update.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/tests/unit/test_admission.py` around lines 9 - 10, Resolve Ruff E701/E702 across the listed test files by placing every executable statement on its own physical line: in backend/tests/unit/test_admission.py (9-10, 20, 23), split conditional assignments and context managers from awaited calls; backend/tests/unit/test_audio_ws_buffers.py (23), separate queue retrieval from completion; backend/tests/unit/test_events_config_logging.py (11, 22, 24), separate update/return, Settings/context-manager, and parser/context-manager calls; backend/tests/unit/test_generation_runner.py (11, 19, 23, 31, 42, 60, 82), expand double initializers and setup assignments, and separate append/event-set operations; backend/tests/unit/test_turn_manager.py (21-22), put cancellation and close calls on separate lines; backend/tests/unit/test_voice_session.py (31, 35, 37, 42, 44, 59, 62), split initialization, signal/clock, close, setup, and generation/signal statements. Preserve the existing test behavior.Source: Linters/SAST tools
frontend/app/hooks/useVoiceSession.ts (1)
140-140: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the dead start guard and the device log.
Two small items:
- Line 279 and line 287 set and clear
startingRefwith noawaitbetween them. The guard at line 271 can never observetrue. Remove the ref or drop the guard.- Line 140 logs the full microphone track settings at
infolevel on every session start. Useconsole.debug, or remove the log.♻️ Proposed refactor
const track = stream.getAudioTracks()[0]; if (track) { - console.info("Voice microphone processing", track.getSettings()); + console.debug("Voice microphone processing", track.getSettings()); }Also applies to: 279-287
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@frontend/app/hooks/useVoiceSession.ts` at line 140, Remove the ineffective startingRef guard and its associated set/clear operations in the voice-session start flow, since no await allows it to prevent concurrent starts. Change the microphone track settings log near the session-start handling from console.info to console.debug, or remove it.frontend/tests/config.test.ts (1)
20-37: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueCover the remaining validation branches.
getAudioWebSocketUrlthrows for three distinct reasons. The suite covers only the path check. Add cases for an unparsable value and for a non-wsscheme.♻️ Proposed additional tests
it("rejects a configured URL outside the audio route", () => { process.env.NEXT_PUBLIC_WS_URL = "ws://127.0.0.1:3300/other"; expect(() => getAudioWebSocketUrl(location)).toThrow( "NEXT_PUBLIC_WS_URL must target /ws/audio", ); }); + + it("rejects a value that is not a URL", () => { + process.env.NEXT_PUBLIC_WS_URL = "not a url"; + expect(() => getAudioWebSocketUrl(location)).toThrow( + "NEXT_PUBLIC_WS_URL must be a valid ws(s) URL", + ); + }); + + it("rejects a non-websocket scheme", () => { + process.env.NEXT_PUBLIC_WS_URL = "http://127.0.0.1:3300/ws/audio"; + expect(() => getAudioWebSocketUrl(location)).toThrow( + "NEXT_PUBLIC_WS_URL must use ws or wss", + ); + }); });🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@frontend/tests/config.test.ts` around lines 20 - 37, Extend the getAudioWebSocketUrl test suite with cases for an unparsable NEXT_PUBLIC_WS_URL value and a URL using a non-ws scheme. Assert that each invocation throws the corresponding validation error, while preserving the existing path-validation coverage.frontend/app/hooks/useSocket.ts (1)
59-71: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueLog dropped server events.
parseServerEventreturnsnullfor a payload that has valid JSON but an unknowntypeor a failed field check. The handler then drops the payload with no signal. If the backend event schema inbackend/app/models/events.pychanges ahead offrontend/app/lib/protocol.ts, the session goes silent with no diagnostic trace.Add a debug-level log for the dropped payload type.
♻️ Proposed refactor
try { const event = parseServerEvent(JSON.parse(message.data)); if (event) { handlersRef.current.onEvent(event, sendControl); + } else { + console.debug("Ignored unrecognized voice-session event."); } } catch {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@frontend/app/hooks/useSocket.ts` around lines 59 - 71, In the socket onmessage handler, update the branch where parseServerEvent returns null to emit a debug-level log containing the dropped payload’s type. Preserve the existing event dispatch and invalid-JSON error handling, and use the project’s established logging mechanism.frontend/tests/useVoiceSession.test.ts (2)
171-180: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a case for a failed assistant generation.
No test sends
assistant.statewithstate: "failed". That branch infrontend/app/hooks/useVoiceSession.tslines 227-230 currently sets theerrorstate without running cleanup, so the microphone and the socket stay active. A test that asserts teardown on that event would catch the leak and lock in the corrected behavior.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@frontend/tests/useVoiceSession.test.ts` around lines 171 - 180, Add a test alongside the existing useVoiceSession lifecycle tests that sends an assistant.state event with state "failed" after starting and reaching session readiness, then assert recorder stop, socket disconnect, and LiveKit disconnect are called. Use the existing socketOptions.onEvent, sessionReady, socket, stop, liveKit, and renderHook setup to verify failed assistant generation performs the same cleanup as end.
16-25: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMake the doubles match the real contracts.
Two fidelity gaps:
- Line 20:
unlockAudio: vi.fn()returnsundefined.useVoiceSessioncallsunlockAudio().then(...)inenableSound, so any test that callsenableSoundwithoutmockResolvedValueOncethrows aTypeError. Give the mock a default async implementation.MockRecorderdiscards the constructor options. The suite therefore cannot assert the requestedmimeType, which is the value that fails on browsers without WebM recording support.♻️ Proposed refactor
- unlockAudio: vi.fn(), + unlockAudio: vi.fn(async () => true),class MockRecorder { static instances: MockRecorder[] = []; + static isTypeSupported = vi.fn(() => true); state = "inactive"; + mimeType: string; ondataavailable: ((event: BlobEvent) => void) | null = null; start = vi.fn(() => { this.state = "recording"; }); stop = vi.fn(() => { this.state = "inactive"; }); - constructor() { + constructor(_stream: MediaStream, options?: MediaRecorderOptions) { + this.mimeType = options?.mimeType ?? ""; MockRecorder.instances.push(this); } }Also applies to: 50-60
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@frontend/tests/useVoiceSession.test.ts` around lines 16 - 25, Update the liveKit test double’s unlockAudio mock to resolve by default so enableSound can safely call its promise methods without per-test setup. Update MockRecorder to retain the constructor options, including mimeType, so tests can inspect and assert the requested recording configuration.backend/app/api/audio_ws.py (1)
316-327: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueCompare against the
WebSocketStateenum instead of the literal1.
websocket.application_state.value == 1depends on the numeric order of Starlette'sWebSocketStatemembers. Use the enum member so the intent is explicit and the check survives an upstream reordering.♻️ Proposed refactor
+from starlette.websockets import WebSocketState + - if websocket.application_state.value == 1: + if websocket.application_state is WebSocketState.CONNECTED:🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/api/audio_ws.py` around lines 316 - 327, Update the application-state check in the websocket error-sending block to compare websocket.application_state against the appropriate WebSocketState enum member instead of the literal value 1, preserving the existing send and suppression behavior.backend/app/core/config.py (1)
27-37: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueIgnore environment variables for explicit
Settingsconstruction.
_env_file=Nonedisables dotenv loading, but process environment variables still populate omitted fields. Use an init-only settings source when explicit construction must be deterministic.
split_originsalso raisesTypeErrorfor non-iterable non-string values. Pydantic v2 does not convert this exception toValidationError, so it bypasses the sanitized error mapping. RaiseValueErrorfor unsupported values if all startup errors must use the same format.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/core/config.py` around lines 27 - 37, Update Settings.__init__ to use an init-only settings source when explicit values are supplied, preventing process environment variables from filling omitted fields while preserving normal environment loading otherwise. Update split_origins to raise ValueError for non-iterable, non-string inputs so Settings.__init__ consistently sanitizes startup failures through its existing error mapping.backend/app/core/logging.py (1)
19-20: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDocument that the value is monotonic, not epoch-based.
monotonic_msreturns an arbitrary origin. The value is valid for durations only. The protocol sends a related monotonic value to the browser inAssistantInterrupted.qualified_interrupt_at_ms. A future reader can compare that field toDate.now()and get a meaningless result. Add a one-line docstring to state the constraint.📝 Proposed docstring
def monotonic_ms() -> int: + """Return monotonic milliseconds. The origin is arbitrary; use for durations only.""" return int(time.monotonic() * 1000)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/core/logging.py` around lines 19 - 20, Update the monotonic_ms function with a one-line docstring stating that its result uses an arbitrary monotonic origin and is suitable only for measuring durations, not comparing with epoch-based timestamps such as Date.now().backend/app/audio/livekit_publisher.py (3)
97-99: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExplain the
asyncio.sleep(0)yield.Line 98 yields control so a pending done-callback from
_observe_workercan record_worker_errorbefore line 99 reads it. The integration test atbackend/tests/integration/test_livekit_publisher.pylines 83-86 depends on this ordering.The call looks removable to a future reader. Add a comment to protect it.
📝 Proposed comment
async def write(self, generation_id: int, frame: bytes) -> None: - await asyncio.sleep(0) + # Yield so a completed worker's done-callback records _worker_error before the check below. + await asyncio.sleep(0) self._raise_worker_error()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/audio/livekit_publisher.py` around lines 97 - 99, Add a concise comment immediately above asyncio.sleep(0) in the write method explaining that the yield lets _observe_worker’s pending done-callback record _worker_error before _raise_worker_error() checks it; preserve the existing ordering and behavior.
69-78: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueLog the worker error message, not only its type.
log_event("livekit.publisher_failed", ...)recordserror_typeonly. ARuntimeErrorfrom a capture failure and aRuntimeErrorfrom a different cause produce identical log lines. The stored exception is also re-raised later from a different call site, so this callback is the only place the original context exists.Add a truncated message to keep the record diagnosable.
📝 Proposed fix
- log_event("livekit.publisher_failed", generation_id=generation_id, error_type=type(error).__name__) + log_event("livekit.publisher_failed", level=logging.ERROR, generation_id=generation_id, + error_type=type(error).__name__, error=str(error)[:200])Add
import loggingat the top of the file.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/audio/livekit_publisher.py` around lines 69 - 78, Update _observe_worker to include a truncated representation of the captured error message in the livekit.publisher_failed log event alongside error_type. Add the necessary logging import if needed to apply the project’s established truncation or formatting approach, while preserving the existing exception storage and hard_cut behavior.
20-27: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConstruct
LiveKitMetadatawith keyword arguments.The dataclass has six fields and five are
str. Lines 50-51 construct it positionally. A field reorder or an inserted field silently mis-assigns values with no type error.The consequence is concrete.
build_session_readyinbackend/app/api/audio_ws.pymapsmetadata.assistant_identitytoparticipant_identity, which the browser uses to find the track to subscribe to. Ifbrowser_identityandassistant_identitywere ever swapped, the browser would subscribe to the wrong identity and audio playback would fail with no error anywhere in the pipeline.♻️ Proposed fix
- self.metadata = LiveKitMetadata(room_name, browser_identity, assistant_identity, ASSISTANT_TRACK_NAME, - browser_token, self.settings.LIVEKIT_URL) + self.metadata = LiveKitMetadata(room_name=room_name, browser_identity=browser_identity, + assistant_identity=assistant_identity, track_name=ASSISTANT_TRACK_NAME, + browser_token=browser_token, livekit_url=self.settings.LIVEKIT_URL)To enforce this, add
kw_only=Trueto the decorator:-@dataclass(frozen=True) +@dataclass(frozen=True, kw_only=True) class LiveKitMetadata:🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/audio/livekit_publisher.py` around lines 20 - 27, Update the LiveKitMetadata dataclass decorator to use kw_only=True, and change its construction near the affected lines to pass all fields by keyword rather than position. Preserve the existing field values and names so build_session_ready continues mapping assistant_identity correctly.backend/tests/integration/test_livekit_publisher.py (3)
50-60: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueClose the publisher at the end of this test.
The sibling tests call
await publisher.close(). This test does not.start_generation(8)created a_consumeworker, andhard_cut(8)cancelled it without awaiting it. The cancelled task is still pending when the event loop tears down, which can produce a pending-task warning that appears to come from an unrelated test.💚 Proposed fix
assert publisher._generation_id is None assert publisher._queue.empty() and source.clears == 1 + await publisher.close()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/tests/integration/test_livekit_publisher.py` around lines 50 - 60, Add cleanup to test_hard_cut_is_synchronous_invalidates_and_clears by awaiting publisher.close() after the existing assertions. Ensure the cancelled _consume worker created by start_generation is awaited before the test finishes, matching the cleanup used by sibling publisher tests.
9-28: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd coverage for the
create_and_publishfailure path.The fakes never fail.
FakeRoom.connectandFakeParticipant.publish_trackalways succeed, so no test reaches theexcept BaseException: await self.close(); raisebranch on lines 58-60 ofbackend/app/audio/livekit_publisher.py.That branch carries the room-deletion guarantee. A test that makes
connectraise, and then assertsdeleted == [room_name], would pin the behavior. A second test that cancels the task duringconnectwould cover the cancellation case discussed on that file.Regarding Ruff
S106on line 28: the credentials are dummy test values, so the rule is a false positive here. If Ruff runs theSrules over tests in CI, add a per-file ignore for the test directory rather than changing the fixture.💚 Proposed test for the failure path
`@pytest.mark.asyncio` async def test_failed_connect_deletes_the_room(monkeypatch): import app.audio.livekit_publisher as module class BrokenRoom(FakeRoom): async def connect(self, url, token): raise RuntimeError('connect failed') api_client = FakeAPI() monkeypatch.setattr(module.rtc, 'AudioSource', lambda *a, **kw: FakeSource()) monkeypatch.setattr(module.rtc, 'Room', BrokenRoom) publisher = LiveKitPublisher(settings(), api_client) with pytest.raises(RuntimeError, match='connect failed'): await publisher.create_and_publish() assert api_client.room.deleted == api_client.room.createdDo you want me to add this test and the cancellation variant?
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/tests/integration/test_livekit_publisher.py` around lines 9 - 28, Add tests for the create_and_publish failure path using a BrokenRoom whose connect method raises, then assert the exception propagates and FakeRoomService.deleted matches the created room. Add a cancellation-during-connect test to cover cleanup on task cancellation, and configure Ruff with a per-file S-rule ignore for the test directory rather than changing settings() credentials.Source: Linters/SAST tools
62-72: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert that the replaced worker has terminated.
This test confirms the new generation id and a single
clear_queuecall. It does not confirm that the previous_consumeworker finished. The test therefore passes while the old worker is still running, which is the race described on lines 84-95 ofbackend/app/audio/livekit_publisher.py.Capture the first worker and assert it is done after the replacement. That assertion fails today and passes once
start_generationawaits the cancelled worker.💚 Proposed fix
await publisher.start_generation(1) + first_worker=publisher._worker await publisher.start_generation(2) assert publisher._generation_id == 2 assert source.clears == 1 + assert first_worker is not publisher._worker + assert first_worker.done() await publisher.close()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/tests/integration/test_livekit_publisher.py` around lines 62 - 72, Update test_starting_replacement_generation_hard_cuts_the_old_generation to capture the first generation’s _consume worker before starting generation 2, then assert that captured worker has terminated after the replacement. Ensure start_generation awaits the cancelled previous worker so this assertion verifies the race is fixed.backend/app/core/turn_manager.py (2)
45-47: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueConsider emitting sooner on an explicit end-of-turn signal.
on_end_of_turnstarts a full debounce cycle. The provider's end-of-turn signal is stronger evidence that the user stopped speaking than the silence timer is. The current path adds aboutdebounce_secondsof latency, which defaults to 2 seconds, before the assistant starts generating.If the signal is trustworthy, use a shorter debounce for this path so the first token arrives sooner. Keep the debounce for the timer path.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/core/turn_manager.py` around lines 45 - 47, Update on_end_of_turn to schedule with a shorter explicit-end debounce instead of the full debounce_seconds delay, while preserving the existing full debounce behavior for the silence-timer path and the current closed/transcript guards.
78-82: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winAwait the cancelled task in
closefor deterministic shutdown.
closeisasyncbut performs no await. It cancels the pending task and returns before that task finishes unwinding. A caller that awaitsclosecan therefore still have an in-flight callback running afterward.This matters during shutdown. The lifespan handler gathers
session.close()and then tears down providers and the LiveKit API client. A still-unwinding callback can touch a provider that is already closing. Await the task socloseis a real barrier.♻️ Proposed fix to make `close` a barrier
async def close(self) -> None: if self._closed: return self._closed = True - self.cancel_pending() + task, self._pending_task = self._pending_task, None + if task and not task.done(): + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await taskAdd
import contextlibat the top of the file.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/core/turn_manager.py` around lines 78 - 82, Update TurnManager.close to await the task cancelled by cancel_pending, making the method a shutdown barrier before returning. Use contextlib to suppress the expected cancellation exception while preserving the existing idempotent _closed guard and cancellation flow.backend/app/services/voice_session.py (3)
40-45: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExpand the compressed statements that Ruff flags.
Ruff reports
E701on lines 48, 61, 64, 89, 94, 101, 106, and 107, andE702on lines 82, 90, and 104. If these rules run in CI, the build fails.The density also hurts the code that matters most. Line 82 packs task creation, set insertion, and callback registration onto one line, and lines 104 and 107 compress shutdown steps. This is the barge-in and teardown path, where ordering is the correctness argument.
Run
ruff formaton the file, or expand the flagged lines.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/services/voice_session.py` around lines 40 - 45, Expand all Ruff E701 and E702 violations in the voice-session implementation, including the generation_id, active_generation_id, and closed properties and the barge-in/teardown paths. Replace semicolon- and inline-statement chains with separate indented statements, preserving task creation, set insertion, callback registration, and shutdown ordering; run Ruff formatting on the file to verify the result.Source: Linters/SAST tools
33-38: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winBound the per-generation dictionaries.
_turn_idsgrows on everystart_generationand is never pruned._interrupt_atis pruned only inplayback_hard_cut, which runs only when the browser sendsplayback.hard_cut. If the browser disconnects after an interruption, or a generation completes without an interruption, entries remain.Both dicts are per-session, so growth is bounded by the session lifetime rather than by process lifetime. A long session still accumulates one entry per turn indefinitely.
Note that
turn_idraisesKeyErrorfor a missing generation, so entries cannot be dropped without care. A bounded mapping is the safer fix.♻️ Proposed direction using a bounded mapping
+from collections import OrderedDict- self._turn_ids: dict[int, str] = {} + self._turn_ids: OrderedDict[int, str] = OrderedDict()def start_generation(self, turn_id: str | None = None) -> int: if self._closed: raise RuntimeError("session is closed") self._generation_id += 1 self._active_generation = self._generation_id self._turn_ids[self._generation_id] = turn_id or str(uuid.uuid4()) + while len(self._turn_ids) > _MAX_TRACKED_GENERATIONS: + self._turn_ids.popitem(last=False) return self._generation_idApply the same trimming to
_interrupt_at, and define_MAX_TRACKED_GENERATIONSas a module constant.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/services/voice_session.py` around lines 33 - 38, Introduce a module-level _MAX_TRACKED_GENERATIONS limit and use bounded mappings for both _turn_ids and _interrupt_at. Preserve the existing turn_id missing-generation behavior while trimming the oldest entries when new generations exceed the limit, and apply the same bounded cleanup regardless of whether playback_hard_cut occurs.
100-109: 🩺 Stability & Availability | 🔵 TrivialConsider a timeout around session shutdown.
The ordering here is correct, and the sink-ownership comment matches how
lifespaninbackend/app/main.pygatherssession.close()beforelivekit_api.aclose().One operational risk remains. Neither
closenor the lifespan gather applies a timeout.audio_sink.close()callsdelete_roomagainst the LiveKit server. If that server is unreachable or slow,closeblocks, and the lifespan gather blocks with it. Process shutdown then hangs until an external supervisor kills it, which turns a deploy into an outage.Apply a bounded timeout at the shutdown boundary and log the sessions that did not finish.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/services/voice_session.py` around lines 100 - 109, Add a bounded timeout around the lifespan shutdown gather that awaits each session.close(), and log the sessions that remain unfinished when the timeout expires. Preserve the existing shutdown ordering so session cleanup completes before livekit_api.aclose(), while ensuring a slow audio_sink.close() cannot block process shutdown indefinitely.backend/app/audio/protocols.py (1)
1-4: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winGive the sentinel a distinct type so the queue union can discriminate.
PCM_COMPLETED = object()types asobject. Inbackend/app/audio/livekit_publisher.pythe queue is annotatedasyncio.Queue[tuple[int, bytes | object]], andbytes | objectcollapses toobject. The type checker then loses the frame type, which is why line 154 of that file needs# type: ignore[arg-type]when it passesitemtortc.AudioFrame.A dedicated singleton type fixes both. After the change, a
item is PCM_COMPLETEDcheck narrows the union tobytes, and the ignore comment can be removed.Add a module docstring as well, to match the sibling modules in this package.
♻️ Proposed typed sentinel
+"""Audio output contracts shared by the generation runner and the LiveKit publisher.""" from __future__ import annotations -from typing import Protocol, runtime_checkable +from enum import Enum +from typing import Protocol, runtime_checkable -PCM_COMPLETED = object() + +class PcmControl(Enum): + """Non-audio control markers carried on the PCM queue.""" + + COMPLETED = "completed" + + +PCM_COMPLETED = PcmControl.COMPLETEDThen in
backend/app/audio/livekit_publisher.py:- self._queue: asyncio.Queue[tuple[int, bytes | object]] = asyncio.Queue(maxsize=settings.PCM_QUEUE_FRAMES) + self._queue: asyncio.Queue[tuple[int, bytes | PcmControl]] = asyncio.Queue(maxsize=settings.PCM_QUEUE_FRAMES)- frame = rtc.AudioFrame(item, 24_000, 1, SAMPLES_PER_FRAME) # type: ignore[arg-type] + frame = rtc.AudioFrame(item, 24_000, 1, SAMPLES_PER_FRAME)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/audio/protocols.py` around lines 1 - 4, Replace the untyped PCM_COMPLETED object in the protocols module with a dedicated singleton sentinel type and annotate the sentinel accordingly, adding the requested module docstring. Update the queue and item handling in livekit_publisher around the PCM_COMPLETED identity check so the union narrows to bytes, then remove the unnecessary type: ignore[arg-type] on the rtc.AudioFrame call.backend/app/services/admission.py (1)
45-51: 🔒 Security & Privacy | 🔵 Trivial | 💤 Low valueDocument the proxy-trust requirement for
TRUST_PROXY_HEADERS.The leftmost
x-forwarded-forentry is the correct choice when the header is trusted, and the fallback to the socket peer is the right default. One deployment hazard is worth recording next to the code: ifTRUST_PROXY_HEADERSis enabled without a proxy that overwrites the header, a client sets its ownx-forwarded-forand bypassesMAX_SESSIONS_PER_IPcompletely.Add a comment so the coupling stays visible to operators.
📝 Proposed comment
def client_ip(self, websocket: WebSocket) -> str: headers: Mapping[str, str] = websocket.headers + # Enable TRUST_PROXY_HEADERS only behind a proxy that overwrites x-forwarded-for. + # Otherwise a client sets the header itself and bypasses MAX_SESSIONS_PER_IP. if self.settings.TRUST_PROXY_HEADERS:🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/services/admission.py` around lines 45 - 51, Add an inline comment in client_ip next to the TRUST_PROXY_HEADERS check documenting that it must only be enabled behind a proxy that overwrites x-forwarded-for; otherwise clients can spoof the header and bypass MAX_SESSIONS_PER_IP. Preserve the existing leftmost-entry parsing and socket-peer fallback.backend/app/models/events.py (1)
61-67: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUse
typeas the discriminator for both event unions.Each member has a unique
Literalvalue, soField(discriminator="type")enables direct model selection and focused validation errors.parse_client_eventconverts validation errors to a generic WebSocket close reason, so the error-detail benefit applies only to server-side diagnostics.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/app/models/events.py` around lines 61 - 67, Define the client and server event unions with `type` as their discriminator by applying `Field(discriminator="type")` to both `ClientEvent` and `ServerEvent`. Update the corresponding `TypeAdapter` definitions, including `_CLIENT_ADAPTER` and `_SERVER_ADAPTER`, to use the discriminated unions while preserving `parse_client_event` behavior.
| try: | ||
| await self._room.connect(self.settings.LIVEKIT_URL, assistant_token) | ||
| track = rtc.LocalAudioTrack.create_audio_track(ASSISTANT_TRACK_NAME, self._source) | ||
| await self._room.local_participant.publish_track(track) | ||
| except BaseException: | ||
| await self.close() | ||
| raise | ||
| return self.metadata |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Cleanup can be interrupted, and the LiveKit room leaks.
except BaseException catches asyncio.CancelledError, then awaits self.close() inside the handler. close awaits room.disconnect() and delete_room. When the original exception is a CancelledError, an await inside the handler can raise a second CancelledError and abort close partway. The room is then never deleted.
This path is routine, not exceptional. The WebSocket task is cancelled whenever the browser disconnects. If the client disconnects while _room.connect or publish_track is in flight, cleanup runs under cancellation and can leave an orphaned room on the LiveKit server for every such session.
Shield the cleanup so it completes.
🛡️ Proposed fix to protect cleanup from cancellation
try:
await self._room.connect(self.settings.LIVEKIT_URL, assistant_token)
track = rtc.LocalAudioTrack.create_audio_track(ASSISTANT_TRACK_NAME, self._source)
await self._room.local_participant.publish_track(track)
except BaseException:
- await self.close()
+ # Shield so a cancellation of this task cannot abort room deletion.
+ with contextlib.suppress(Exception):
+ await asyncio.shield(asyncio.ensure_future(self.close()))
raise
return self.metadataConsider adding a timeout around the shielded cleanup so a hung LiveKit API call cannot delay the unwind indefinitely.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/app/audio/livekit_publisher.py` around lines 54 - 61, Update the
exception handler in the publisher method containing _room.connect and
publish_track so self.close() runs under cancellation protection and completes
even when the original error is asyncio.CancelledError. Wrap the shielded
cleanup in a bounded timeout so a hung LiveKit disconnect or delete operation
cannot block unwinding indefinitely, then re-raise the original exception.
| async def start_generation(self, generation_id: int) -> None: | ||
| if self._closed or self._source is None: | ||
| raise RuntimeError("LiveKit publisher is unavailable") | ||
| self._raise_worker_error() | ||
| old_generation = self._generation_id | ||
| if old_generation is not None: | ||
| # Never clear a new generation when replacing an old one. | ||
| self.hard_cut(old_generation) | ||
| self._generation_id = generation_id | ||
| self._worker_error = None | ||
| self._worker = asyncio.create_task(self._consume(generation_id)) | ||
| self._worker.add_done_callback(self._observe_worker) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Await the cancelled worker before starting the replacement.
hard_cut(old_generation) on line 91 calls self._worker.cancel() but does not await the task. Line 94 then overwrites self._worker with a new _consume task. Cancellation is delivered on a later loop iteration, so the old worker and the new worker can both run for a short window, and both loop on the same shared self._queue.
The old worker captured the old id in its generation parameter. If it wins a get() race during that window, it takes a frame belonging to the new generation and drops it on line 146 as worker_generation_invalid. The frame is consumed and never published.
The user-visible result is clipped assistant audio at the start of a replacement turn. Replacement is the barge-in path, so this window is reached whenever a user interrupts.
🐛 Proposed fix to serialize worker replacement
old_generation = self._generation_id
if old_generation is not None:
# Never clear a new generation when replacing an old one.
self.hard_cut(old_generation)
+ # Await the cancelled worker so it cannot consume frames of the new generation.
+ previous = self._worker
+ if previous and not previous.done():
+ with contextlib.suppress(asyncio.CancelledError, Exception):
+ await previous
self._generation_id = generation_id
self._worker_error = None
self._worker = asyncio.create_task(self._consume(generation_id))
self._worker.add_done_callback(self._observe_worker)Note that _observe_worker runs for the cancelled task first. It returns early on task.cancelled(), so it will not overwrite _worker_error after line 93 clears it.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async def start_generation(self, generation_id: int) -> None: | |
| if self._closed or self._source is None: | |
| raise RuntimeError("LiveKit publisher is unavailable") | |
| self._raise_worker_error() | |
| old_generation = self._generation_id | |
| if old_generation is not None: | |
| # Never clear a new generation when replacing an old one. | |
| self.hard_cut(old_generation) | |
| self._generation_id = generation_id | |
| self._worker_error = None | |
| self._worker = asyncio.create_task(self._consume(generation_id)) | |
| self._worker.add_done_callback(self._observe_worker) | |
| async def start_generation(self, generation_id: int) -> None: | |
| if self._closed or self._source is None: | |
| raise RuntimeError("LiveKit publisher is unavailable") | |
| self._raise_worker_error() | |
| old_generation = self._generation_id | |
| if old_generation is not None: | |
| # Never clear a new generation when replacing an old one. | |
| self.hard_cut(old_generation) | |
| # Await the cancelled worker so it cannot consume frames of the new generation. | |
| previous = self._worker | |
| if previous and not previous.done(): | |
| with contextlib.suppress(asyncio.CancelledError, Exception): | |
| await previous | |
| self._generation_id = generation_id | |
| self._worker_error = None | |
| self._worker = asyncio.create_task(self._consume(generation_id)) | |
| self._worker.add_done_callback(self._observe_worker) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/app/audio/livekit_publisher.py` around lines 84 - 95, Make
start_generation asynchronous replacement serialize the old worker before
creating the new one: after hard_cut(old_generation), await the cancelled
existing worker task and handle its expected cancellation, then assign the new
generation and create _consume. Preserve _observe_worker behavior and ensure
_worker is not overwritten until the prior task has fully stopped.
| worker = self._worker | ||
| if worker: | ||
| try: | ||
| await worker | ||
| except asyncio.CancelledError: | ||
| raise | ||
| except Exception: | ||
| # The callback recorded the original failure and hard-cut synchronously. | ||
| pass | ||
| self._raise_worker_error() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
A concurrent hard cut turns finish into a spurious cancellation.
hard_cut cancels self._worker. If a barge-in occurs while finish awaits that worker on line 118, await worker raises CancelledError, and lines 119-120 re-raise it.
The caller was not cancelled. backend/app/services/generation.py line 194 awaits audio_sink.finish(request.generation_id), so the runner task receives a CancelledError that did not originate from its own cancellation. That unwinds the runner as though the request were cancelled and corrupts asyncio's cancellation bookkeeping for that task.
Distinguish worker cancellation from caller cancellation. A cancelled worker after a hard cut is a normal barge-in outcome and should return quietly.
Regarding the Ruff S110 and BLE001 hints on lines 121-123: the suppression is intentional, because _observe_worker records the failure and line 124 re-raises it. Keep the behavior and silence the rules explicitly.
🐛 Proposed fix
worker = self._worker
if worker:
- try:
- await worker
- except asyncio.CancelledError:
- raise
- except Exception:
- # The callback recorded the original failure and hard-cut synchronously.
- pass
+ # Wait without absorbing the worker's own cancellation into this task.
+ await asyncio.wait({worker})
+ if worker.cancelled():
+ # A concurrent hard cut cancelled the worker. This is a normal barge-in.
+ return
+ if worker.exception() is not None:
+ # _observe_worker recorded the failure; _raise_worker_error below reports it.
+ pass
self._raise_worker_error()asyncio.wait does not propagate the awaited task's exception or cancellation, so a genuine cancellation of the caller still interrupts the await correctly.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| worker = self._worker | |
| if worker: | |
| try: | |
| await worker | |
| except asyncio.CancelledError: | |
| raise | |
| except Exception: | |
| # The callback recorded the original failure and hard-cut synchronously. | |
| pass | |
| self._raise_worker_error() | |
| worker = self._worker | |
| if worker: | |
| # Wait without absorbing the worker's own cancellation into this task. | |
| await asyncio.wait({worker}) | |
| if worker.cancelled(): | |
| # A concurrent hard cut cancelled the worker. This is a normal barge-in. | |
| return | |
| if worker.exception() is not None: | |
| # _observe_worker recorded the failure; _raise_worker_error below reports it. | |
| pass | |
| self._raise_worker_error() |
🧰 Tools
🪛 Ruff (0.16.1)
[error] 121-123: try-except-pass detected, consider logging the exception
(S110)
[warning] 121-121: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/app/audio/livekit_publisher.py` around lines 115 - 124, Update finish
around self._worker to wait via asyncio.wait so worker cancellation from
hard_cut is observed without re-raising it as caller cancellation, while genuine
cancellation of the caller still interrupts finish. Preserve _observe_worker’s
recorded-error handling and _raise_worker_error propagation, and explicitly
suppress the intentional Ruff S110 and BLE001 diagnostics around the
worker-await exception handling.
| if self._created_room and self.metadata: | ||
| try: | ||
| await self.api_client.room.delete_room(room_proto.DeleteRoomRequest(room=self.metadata.room_name)) | ||
| except Exception as exc: | ||
| if "not found" not in str(exc).lower(): | ||
| raise | ||
| finally: | ||
| self._created_room = False |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
livekit-api Python TwirpError code attribute delete_room room not found error type
💡 Result:
A TwirpError with code "not_found" when calling delete_room in LiveKit indicates that the LiveKit Server does not recognize the specified room name or ID as an active room [1][2][3]. ### Understanding the Error When you call delete_room, you are instructing the LiveKit SFU (Selective Forwarding Unit) to terminate a specific room session. A "not_found" error is the authoritative signal from the server that the room does not exist in its current, active state [4]. This can occur for several reasons: 1. Room Already Deleted: The room has already been closed, either by a previous delete_room call or because it naturally expired due to inactivity (e.g., all participants left and the empty timeout was reached) [4][3]. 2. Stale Observability Data: You may be attempting to delete a room based on information from an observability dashboard or historical session record [4]. These dashboards sometimes show sessions as "Active" due to delays in synchronization or long-running tasks (like Egress jobs) even after the actual room SFU process has terminated [4][5]. 3. Race Conditions: In rare cases, intermittent 404 errors can occur due to server-side timing issues or propagation delays, especially under high load or in clustered environments [1][6]. ### Recommended Actions * Verify Active Status: Use the LiveKit CLI command lk room list to check if the room is truly active on the server [4][5]. If the room does not appear in this list, it is not active, and a 404 on delete_room is expected behavior [4]. * Check for Stuck Egress/Ingress Jobs: If you believe a room should be gone but it remains visible in a dashboard, check if there are lingering Egress or Ingress jobs associated with that session that are keeping the record "Active" [4]. * Handle the Exception: Your application code should be prepared to catch and handle this error gracefully. Because delete_room is intended to ensure a room is gone, receiving a 404 effectively confirms the desired state (that the room no longer exists) [4][3]. Example error handling in Python: from livekit import api try: await lkapi.room.delete_room(api.DeleteRoomRequest(room="myroom")) except api.ServerError as e: if e.code == api.TwirpErrorCode.NOT_FOUND: print("Room already closed or does not exist.") else: print(f"Failed to delete room: {e.code} - {e.message}") [7][8][9]
Citations:
- 1: Get TwirpError(code=not_found, message=requested room does not exist, status=404) livekit/python-sdks#446
- 2: lk room delete returns "could not find object" even though room is listed livekit/livekit#3766
- 3: URGENT (Room Delete Permanently) livekit/server-sdk-go#284
- 4: https://community.livekit.io/t/room-session-not-finishing-despite-no-participants/1139
- 5: https://community.livekit.io/t/livekit-cloud-session-remains-active-in-sessions-dashboard-even-after-room-is-gone-from-roomservice-api/617
- 6: Bug: Room appears in list, but updateRoomMetadata and updateParticipantMetadata return 404 intermittently livekit/livekit#3763
- 7: https://docs.livekit.io/python/livekit/api/
- 8: https://docs.livekit.io/intro/basics/rooms-participants-tracks/rooms/
- 9: https://github.com/livekit/client-sdk-python
🏁 Script executed:
sed -n '1,230p' backend/app/audio/livekit_publisher.py
printf '\n--- dependency references ---\n'
rg -n "livekit-api|livekit|TwirpError|DeleteRoomRequest|delete_room" -S pyproject.toml poetry.lock requirements*.txt backend 2>/dev/null | head -200Repository: muhammad-hammad-sarwar/realtime-voice-agent
Length of output: 11751
🌐 Web query:
livekit-api 1.2.0 Python source TwirpError ServerError code not_found
💡 Result:
In the LiveKit Python SDK (including version 1.2.0 and later), a TwirpError with the code not_found and status 404 indicates that the requested resource—such as a room or participant—could not be located on the LiveKit server [1][2]. Key details regarding this error: 1. Nature of the Error: It is a server-side error, specifically a standard Twirp (RPC) error code representing an HTTP 404 condition [3][2]. 2. Common Causes: - Timing or Race Conditions: It often occurs when an API call (e.g., attempting to start Egress or modify a participant) is made just as a room is being created or closed, leading to a temporary "not found" state [1]. - Invalid Resource Name: The specific room, participant, or identity provided in the request does not exist in the current server state [1]. 3. Recommended Handling: - Error Class: The SDK uses ServerError as the primary base class for these API exceptions, with TwirpError typically kept as an alias for backward compatibility [4][5]. - Implementation: In application logic, you should catch ServerError and inspect the code attribute [4]. - Mitigation: If the error is due to a race condition (e.g., checking for a room that is in the process of being initialized), implementing a retry strategy with backoff is often effective [1]. Example: try: await lkapi.egress.start_room_composite_egress(req) except api.ServerError as e: if e.code == "not_found": # Handle or retry the request print("Resource not found:", e.message)
Citations:
- 1: Get TwirpError(code=not_found, message=requested room does not exist, status=404) livekit/python-sdks#446
- 2: https://docs.livekit.io/python/livekit/api/
- 3: https://docs.livekit.io/python/livekit/api/twirp_client.html
- 4: https://github.com/livekit/python-sdks
- 5: consistent error class ServerError, test suite livekit/python-sdks#744
🏁 Script executed:
python3 - <<'PY'
import json
import urllib.request
urls = [
"https://raw.githubusercontent.com/livekit/python-sdks/v1.2.0/livekit-api/livekit/api/twirp_client.py",
"https://raw.githubusercontent.com/livekit/python-sdks/v1.2.0/livekit-api/livekit/api/__init__.py",
"https://raw.githubusercontent.com/livekit/python-sdks/main/livekit-api/livekit/api/twirp_client.py",
]
for url in urls:
print(f"\n--- {url} ---")
try:
with urllib.request.urlopen(url, timeout=10) as response:
text = response.read().decode()
for i, line in enumerate(text.splitlines(), 1):
if "class ServerError" in line or "class TwirpError" in line or "code" in line:
print(f"{i}: {line}")
except Exception as exc:
print(type(exc).__name__, exc)
PYRepository: muhammad-hammad-sarwar/realtime-voice-agent
Length of output: 937
🏁 Script executed:
for url in \
'https://raw.githubusercontent.com/livekit/python-sdks/v1.2.0/livekit-api/livekit/api/twirp_client.py' \
'https://raw.githubusercontent.com/livekit/python-sdks/v1.2.0/livekit-api/livekit/api/__init__.py' \
'https://raw.githubusercontent.com/livekit/python-sdks/main/livekit-api/livekit/api/twirp_client.py'
do
echo "--- $url ---"
curl -ksSfL "$url" | nl -ba | grep -E 'class (ServerError|TwirpError)|code|__all__' | head -80 || true
doneRepository: muhammad-hammad-sarwar/realtime-voice-agent
Length of output: 910
🏁 Script executed:
python3 - <<'PY'
import json
import ssl
import tarfile
import urllib.request
from io import BytesIO
ctx = ssl._create_unverified_context()
meta_url = "https://pypi.org/pypi/livekit-api/1.2.0/json"
with urllib.request.urlopen(meta_url, context=ctx, timeout=15) as response:
meta = json.load(response)
for item in meta["urls"]:
if item["packagetype"] == "sdist":
source_url = item["url"]
break
else:
raise SystemExit("No source distribution found")
print("source:", source_url)
with urllib.request.urlopen(source_url, context=ctx, timeout=15) as response:
data = response.read()
with tarfile.open(fileobj=BytesIO(data), mode="r:*") as archive:
for member in archive.getmembers():
if member.name.endswith(("twirp_client.py", "api/__init__.py")):
print(f"\n--- {member.name} ---")
text = archive.extractfile(member).read().decode()
for number, line in enumerate(text.splitlines(), 1):
if any(token in line for token in ("class ServerError", "class TwirpError", "code", "ServerError")):
print(f"{number}: {line}")
PYRepository: muhammad-hammad-sarwar/realtime-voice-agent
Length of output: 2135
🏁 Script executed:
python3 - <<'PY'
import ssl, tarfile, urllib.request
from io import BytesIO
url = "https://files.pythonhosted.org/packages/f3/19/36ff6712ec638a4b7dad4d8f03795952e401dc31db0b04cddec7892650da/livekit_api-1.2.0.tar.gz"
with urllib.request.urlopen(url, context=ssl._create_unverified_context(), timeout=15) as response:
data = response.read()
with tarfile.open(fileobj=BytesIO(data), mode="r:*") as archive:
for name in (
"livekit_api-1.2.0/livekit/api/twirp_client.py",
"livekit_api-1.2.0/livekit/api/__init__.py",
):
text = archive.extractfile(name).read().decode().splitlines()
print(f"\n--- {name} ---")
for start, end in ((40, 160), (270, 290)):
if start <= len(text):
for number in range(start, min(end, len(text)) + 1):
print(f"{number}: {text[number - 1]}")
PY
printf '\n--- teardown call sites ---\n'
sed -n '130,190p' backend/app/api/audio_ws.py
sed -n '1,120p' backend/tests/integration/test_livekit_publisher.pyRepository: muhammad-hammad-sarwar/realtime-voice-agent
Length of output: 14005
🏁 Script executed:
rg -n -C 5 "publisher\.close|\.close\(\)|create_and_publish|except BaseException|lifespan|sessions" backend/app backend/tests -g '*.py' | head -240Repository: muhammad-hammad-sarwar/realtime-voice-agent
Length of output: 18717
Handle room-deletion errors by code, not message text.
Catch api.ServerError and compare exc.code with api.ServerErrorCode.NOT_FOUND. Treat only NOT_FOUND as idempotent. Log other deletion failures and continue teardown so cleanup cannot mask the original failure or abort lifespan shutdown.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/app/audio/livekit_publisher.py` around lines 171 - 178, Update the
room cleanup block around the publisher teardown to catch api.ServerError and
compare exc.code with api.ServerErrorCode.NOT_FOUND instead of inspecting
exception text. Treat only NOT_FOUND as idempotent; log other deletion failures
and continue teardown without re-raising, while preserving the finally reset of
_created_room.
| def log_event(event: str, *, level: int = logging.INFO, **fields: Any) -> None: | ||
| """Write a one-line JSON record with a stable event name and safe fields.""" | ||
| safe = {key: value for key, value in fields.items() if "key" not in key.lower() and "secret" not in key.lower()} | ||
| LOGGER.log(level, json.dumps({"event": event, **safe}, default=str, separators=(",", ":"))) |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy lift
Extend the redaction deny list beyond key and secret.
The filter matches only field names that contain key or secret. Sensitive names used in this cohort do not match. token is one example: SessionReady in backend/app/models/events.py carries a LiveKit JWT in a field named token, and LiveKitPublisher._token mints both browser and assistant tokens. Any future log_event(..., token=...) call writes the JWT to logs. User speech is a second example: transcript fields flow through TurnManager and VoiceSession, and logging one records speech content.
Add the common sensitive substrings, and prefer redaction markers over silent removal so the record still shows that a field was present.
🔒 Proposed fix to broaden redaction
+_REDACTED_SUBSTRINGS = ("key", "secret", "token", "password", "authorization", "credential")
+
+
def log_event(event: str, *, level: int = logging.INFO, **fields: Any) -> None:
"""Write a one-line JSON record with a stable event name and safe fields."""
- safe = {key: value for key, value in fields.items() if "key" not in key.lower() and "secret" not in key.lower()}
+ safe = {
+ name: ("[redacted]" if any(part in name.lower() for part in _REDACTED_SUBSTRINGS) else value)
+ for name, value in fields.items()
+ }
LOGGER.log(level, json.dumps({"event": event, **safe}, default=str, separators=(",", ":")))🧰 Tools
🪛 ast-grep (0.45.1)
[info] 15-15: use jsonify instead of json.dumps for JSON output
Context: json.dumps({"event": event, **safe}, default=str, separators=(",", ":"))
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/app/core/logging.py` around lines 13 - 16, Update log_event’s
safe-field filtering to redact fields whose names contain token or transcript in
addition to key or secret. Replace matched values with a clear redaction marker
rather than removing those fields, while preserving the existing JSON event
structure and serialization behavior.
| try: | ||
| emitted = emit_interrupt(interrupt.generation_id, interrupt.qualified_at_ms, interrupt.turn_id) | ||
| except TypeError: | ||
| # Compatibility for internal two-argument callbacks; route callbacks receive turn_id. | ||
| emitted = emit_interrupt(interrupt.generation_id, interrupt.qualified_at_ms) # type: ignore[call-arg] |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
The TypeError fallback can call the callback twice and hide real errors.
except TypeError wraps the call to emit_interrupt, so it catches every TypeError raised anywhere inside the callback body, not only an arity mismatch. When a three-argument callback raises TypeError internally, this handler treats it as a signature mismatch and invokes the callback a second time with two arguments. The result is either a duplicate interrupt sent to the browser or a second confusing TypeError that masks the original fault.
The fallback also contradicts the declared contract. InterruptEmitter on line 15 is Callable[[int, int, str], Awaitable[None] | None], and the two-argument path needs # type: ignore[call-arg] to type-check.
The only two-argument caller is a test. backend/tests/unit/test_voice_session.py line 21 passes lambda gid, _:, while line 63 already uses the three-argument form. Update the test to the declared signature and remove the fallback.
🐛 Proposed fix
interrupt = Interrupt(generation_id, self.turn_id(generation_id), int(qualified_at * 1000))
- try:
- emitted = emit_interrupt(interrupt.generation_id, interrupt.qualified_at_ms, interrupt.turn_id)
- except TypeError:
- # Compatibility for internal two-argument callbacks; route callbacks receive turn_id.
- emitted = emit_interrupt(interrupt.generation_id, interrupt.qualified_at_ms) # type: ignore[call-arg]
+ emitted = emit_interrupt(interrupt.generation_id, interrupt.qualified_at_ms, interrupt.turn_id)In backend/tests/unit/test_voice_session.py, change the two-argument lambda to accept turn_id:
- session.on_update("interrupt", teardown=lambda gid: calls.append(("teardown", gid)),
- emit_interrupt=lambda gid, _: calls.append(("priority", gid)))
+ session.on_update("interrupt", teardown=lambda gid: calls.append(("teardown", gid)),
+ emit_interrupt=lambda gid, _at, _turn: calls.append(("priority", gid)))📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| try: | |
| emitted = emit_interrupt(interrupt.generation_id, interrupt.qualified_at_ms, interrupt.turn_id) | |
| except TypeError: | |
| # Compatibility for internal two-argument callbacks; route callbacks receive turn_id. | |
| emitted = emit_interrupt(interrupt.generation_id, interrupt.qualified_at_ms) # type: ignore[call-arg] | |
| interrupt = Interrupt(generation_id, self.turn_id(generation_id), int(qualified_at * 1000)) | |
| emitted = emit_interrupt(interrupt.generation_id, interrupt.qualified_at_ms, interrupt.turn_id) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/app/services/voice_session.py` around lines 76 - 80, Remove the
TypeError compatibility fallback around emit_interrupt in the voice-session
interrupt flow and invoke it only with the declared three-argument
InterruptEmitter signature. Update the two-argument lambda in
test_voice_session.py to accept turn_id, preserving the existing three-argument
test behavior and allowing callback TypeErrors to propagate without retrying.
| if inspect.isawaitable(emitted): | ||
| task = asyncio.create_task(emitted); self._interrupt_tasks.add(task); task.add_done_callback(self._interrupt_tasks.discard) | ||
| self._schedule_teardown(generation_id, teardown) | ||
| return interrupt | ||
|
|
||
| def _schedule_teardown(self, generation_id: int, teardown: Teardown) -> None: | ||
| async def run() -> None: | ||
| result = teardown(generation_id) | ||
| if inspect.isawaitable(result): await result | ||
| task = asyncio.create_task(run()); self._teardown_tasks.add(task); task.add_done_callback(self._teardown_tasks.discard) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Fire-and-forget tasks drop their exceptions across the pipeline. Three call sites create an asyncio.Task and then discard the reference without ever calling task.exception() or wrapping the body in a handler. When the coroutine raises, asyncio reports "Task exception was never retrieved" at garbage-collection time, and the originating turn or barge-in fails with no log and no ErrorEvent to the browser.
backend/app/services/voice_session.py#L81-L90: replaceadd_done_callback(self._interrupt_tasks.discard)andadd_done_callback(self._teardown_tasks.discard)with a callback that discards the task and logstask.exception(). The teardown path matters most, becauseaudio_ws.pyline 221 passesteardown=lambda _: cancel_generation(); a silent failure there leaves the assistant speaking after the user interrupted.backend/app/core/turn_manager.py#L59-L71: wrap theself._callback(transcript)invocation in_wait_and_emitwithexcept asyncio.CancelledError: raisefollowed byexcept Exceptionthat logs the failure. Line 62 already cleared_transcript, so an unlogged failure discards the user's turn with no trace.
Add one shared helper for the done-callback pattern so the third site cannot regress.
🧰 Tools
🪛 Ruff (0.16.1)
[error] 82-82: Multiple statements on one line (semicolon)
(E702)
[error] 82-82: Multiple statements on one line (semicolon)
(E702)
[error] 89-89: Multiple statements on one line (colon)
(E701)
[error] 90-90: Multiple statements on one line (semicolon)
(E702)
[error] 90-90: Multiple statements on one line (semicolon)
(E702)
📍 Affects 2 files
backend/app/services/voice_session.py#L81-L90(this comment)backend/app/core/turn_manager.py#L59-L71
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/app/services/voice_session.py` around lines 81 - 90, The
fire-and-forget task paths lose exceptions without logging or propagating
failure. In backend/app/services/voice_session.py:81-90, add one shared
done-callback helper that removes completed tasks from their tracking set and
retrieves/logs task.exception(), then use it for both interrupt and teardown
tasks; ensure cancellation is handled without treating it as an error. In
backend/app/core/turn_manager.py:59-71, update _wait_and_emit to re-raise
asyncio.CancelledError and log other exceptions from self._callback(transcript)
after _transcript is cleared.
| source=FakeSource(24000, 1, queue_size_ms=100); room=FakeRoom() | ||
| monkeypatch.setattr(module.rtc, 'AudioSource', lambda *a,**kw: source) | ||
| monkeypatch.setattr(module.rtc, 'Room', lambda: room) | ||
| monkeypatch.setattr(module.rtc.LocalAudioTrack, 'create_audio_track', lambda name, _: (name, 'track')) | ||
| publisher=LiveKitPublisher(settings(), FakeAPI()) | ||
| metadata=await publisher.create_and_publish() | ||
| assert metadata.room_name.startswith('voice-') and metadata.browser_identity != metadata.assistant_identity | ||
| assert source.args == (24000, 1) and source.kwargs['queue_size_ms'] == 100 | ||
| assert publisher._queue.maxsize == 2500 |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
The bounded-source assertion is vacuous.
Line 34 patches AudioSource with lambda *a, **kw: source, which discards the arguments the publisher passes. The source object was built by the test itself on line 33 with FakeSource(24000, 1, queue_size_ms=100).
Line 40 therefore asserts the values the test hardcoded, not the values create_and_publish supplied. The publisher could pass a 48000 Hz rate, or omit queue_size_ms entirely, and this test would still pass. The bounded-source guarantee named in the test title is not verified.
Capture the real arguments in the patch.
🐛 Proposed fix
- source=FakeSource(24000, 1, queue_size_ms=100); room=FakeRoom()
- monkeypatch.setattr(module.rtc, 'AudioSource', lambda *a,**kw: source)
+ room=FakeRoom()
+ created=[]
+ def make_source(*a, **kw):
+ created.append(FakeSource(*a, **kw))
+ return created[-1]
+ monkeypatch.setattr(module.rtc, 'AudioSource', make_source)
monkeypatch.setattr(module.rtc, 'Room', lambda: room)
monkeypatch.setattr(module.rtc.LocalAudioTrack, 'create_audio_track', lambda name, _: (name, 'track'))
publisher=LiveKitPublisher(settings(), FakeAPI())
metadata=await publisher.create_and_publish()
+ source=created[0]
assert metadata.room_name.startswith('voice-') and metadata.browser_identity != metadata.assistant_identity
assert source.args == (24000, 1) and source.kwargs['queue_size_ms'] == 100
- assert publisher._queue.maxsize == 2500
+ assert publisher._queue.maxsize == settings().PCM_QUEUE_FRAMESDeriving maxsize from the settings value also removes the magic 2500, so a changed default does not fail this test without explanation.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| source=FakeSource(24000, 1, queue_size_ms=100); room=FakeRoom() | |
| monkeypatch.setattr(module.rtc, 'AudioSource', lambda *a,**kw: source) | |
| monkeypatch.setattr(module.rtc, 'Room', lambda: room) | |
| monkeypatch.setattr(module.rtc.LocalAudioTrack, 'create_audio_track', lambda name, _: (name, 'track')) | |
| publisher=LiveKitPublisher(settings(), FakeAPI()) | |
| metadata=await publisher.create_and_publish() | |
| assert metadata.room_name.startswith('voice-') and metadata.browser_identity != metadata.assistant_identity | |
| assert source.args == (24000, 1) and source.kwargs['queue_size_ms'] == 100 | |
| assert publisher._queue.maxsize == 2500 | |
| room=FakeRoom() | |
| created=[] | |
| def make_source(*a, **kw): | |
| created.append(FakeSource(*a, **kw)) | |
| return created[-1] | |
| monkeypatch.setattr(module.rtc, 'AudioSource', make_source) | |
| monkeypatch.setattr(module.rtc, 'Room', lambda: room) | |
| monkeypatch.setattr(module.rtc.LocalAudioTrack, 'create_audio_track', lambda name, _: (name, 'track')) | |
| publisher=LiveKitPublisher(settings(), FakeAPI()) | |
| metadata=await publisher.create_and_publish() | |
| source=created[0] | |
| assert metadata.room_name.startswith('voice-') and metadata.browser_identity != metadata.assistant_identity | |
| assert source.args == (24000, 1) and source.kwargs['queue_size_ms'] == 100 | |
| assert publisher._queue.maxsize == settings().PCM_QUEUE_FRAMES |
🧰 Tools
🪛 Ruff (0.16.1)
[error] 33-33: Multiple statements on one line (semicolon)
(E702)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/tests/integration/test_livekit_publisher.py` around lines 33 - 41,
Update the AudioSource monkeypatch in the test around
LiveKitPublisher.create_and_publish to capture the actual positional and keyword
arguments passed by the publisher, then construct or configure the FakeSource
from those captured values. Assert against the captured arguments rather than
the preconfigured source, and derive the expected publisher._queue.maxsize from
the relevant settings value instead of hardcoding 2500.
| if (track) { | ||
| console.info("Voice microphone processing", track.getSettings()); | ||
| } | ||
| const recorder = new MediaRecorder(stream, { mimeType: "audio/webm;codecs=opus" }); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Find the encoding/container the backend expects for microphone frames.
fd -t f -e py . backend | xargs rg -n -C4 'encoding|webm|opus|mimetype|mime_type|container|sample_rate'Repository: muhammad-hammad-sarwar/realtime-voice-agent
Length of output: 1079
🏁 Script executed:
#!/bin/bash
set -eu
echo '--- backend Python files ---'
git ls-files 'backend/**/*.py' | sed -n '1,200p'
echo '--- audio websocket implementation ---'
rg -n -C8 'send_bytes|receive_bytes|Deepgram|deepgram|audio|encoding|webm|opus|mp4|sample_rate|mime' backend -g '*.py' | sed -n '1,260p'
echo '--- integration test context ---'
sed -n '1,180p' backend/tests/integration/test_audio_websocket.pyRepository: muhammad-hammad-sarwar/realtime-voice-agent
Length of output: 30508
🏁 Script executed:
#!/bin/bash
set -eu
echo '--- Deepgram provider ---'
sed -n '1,260p' backend/app/providers/deepgram.py
echo '--- WebSocket route ---'
sed -n '1,300p' backend/app/api/audio_ws.py
echo '--- frontend recorder and socket code ---'
sed -n '115,165p' frontend/app/hooks/useVoiceSession.ts
sed -n '1,110p' frontend/app/hooks/useSocket.ts 2>/dev/null || true
echo '--- repository references to recorder MIME types and Deepgram options ---'
rg -n -C5 'MediaRecorder|audio/webm|audio/mp4|LiveKit|DeepgramOptions|encoding|sample_rate|channels' frontend backend -g '*.{ts,tsx,py}'Repository: muhammad-hammad-sarwar/realtime-voice-agent
Length of output: 50399
🏁 Script executed:
#!/bin/bash
set -eu
echo '--- Deepgram-related configuration and provider call graph ---'
rg -n -C10 'Deepgram|deepgram|send_media|encoding|sample_rate|channels|AudioOptions|ListenOptions' backend/app backend/tests -g '*.py'
echo '--- frontend hook imports and error path ---'
sed -n '1,190p' frontend/app/hooks/useVoiceSession.ts
sed -n '190,330p' frontend/app/hooks/useVoiceSession.tsRepository: muhammad-hammad-sarwar/realtime-voice-agent
Length of output: 42836
🌐 Web query:
Deepgram Flux v2 live streaming WebSocket supported audio containers WebM Opus audio/mp4 encoding
💡 Result:
Deepgram's Flux v2 live streaming via WebSocket supports specific containerized and raw audio formats [1][2]. For containerized audio, Deepgram Flux v2 supports: - WebM containers with Opus encoding [1][3][2] - Ogg containers with Opus encoding [4][1][2] - WAV containers with linear16 encoding [4][1][2] When sending containerized audio, you must omit the encoding and sample_rate parameters in your WebSocket URL, as the Flux model automatically detects these from the container metadata [4][1][3]. For raw audio (non-containerized), Flux v2 supports: - linear16, linear32, mulaw, alaw, opus, and ogg-opus [5][1][2] - When sending raw audio, both the encoding and sample_rate parameters are required [1][2]. Regarding audio/mp4: Deepgram Flux does not natively support MP4 containers for live streaming [1][2]. If your audio source is in an MP4 container, it must be demuxed or transcoded into a supported format (such as raw linear16 or a supported containerized format like WebM/Opus) before being sent to the WebSocket endpoint [1][2]. For all live streaming, ensure you are connecting to the /v2/listen endpoint (e.g., wss://api.deepgram.com/v2/listen) [1][2]. Using an 80ms audio chunk size is strongly recommended for optimal performance and latency [1][2][6].
Citations:
- 1: https://developers.deepgram.com/docs/flux/quickstart.mdx
- 2: https://developers.deepgram.com/docs/flux/nova-3-migration
- 3: https://developers.deepgram.com/changelog/2026/1/13.md
- 4: https://developers.deepgram.com/changelog/2025/10/16.md
- 5: https://developers.deepgram.com/reference/speech-to-text/listen-flux.mdx
- 6: https://github.com/deepgram-devs/dg-sagemaker/blob/main/python-flux/README.md
Handle browsers without WebM recording support
MediaRecorder throws NotSupportedError when audio/webm;codecs=opus is unavailable. The Deepgram Flux connection accepts WebM/Opus and Ogg/Opus containers, but not audio/mp4.
Use MediaRecorder.isTypeSupported to select a supported WebM/Opus or Ogg/Opus type. If none is supported, show an unsupported-browser error instead of passing audio/mp4 to Deepgram.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@frontend/app/hooks/useVoiceSession.ts` at line 142, Update the MediaRecorder
initialization in the voice-session recording flow to select a supported
WebM/Opus or Ogg/Opus MIME type via MediaRecorder.isTypeSupported. If neither
format is supported, show the existing unsupported-browser error and do not
create or connect a recorder; never fall back to audio/mp4.
| } else { | ||
| setError("The assistant audio could not be completed."); | ||
| setSessionState("error"); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Release resources when a generation fails.
This branch sets the error state but skips cleanup(). The MediaRecorder, the microphone tracks, the WebSocket, and the LiveKit room all stay active. frontend/app/page.tsx disables End for the error state, so the user cannot stop the microphone. Pressing Start then calls connect(), which returns early in frontend/app/hooks/useSocket.ts line 50 because the socket is still OPEN; no new session.ready arrives and the UI stays in connecting.
Call failSession so teardown and state change stay consistent with every other failure path.
🐛 Proposed fix
} else if (event.state === "completed") {
setSessionState("listening");
} else {
- setError("The assistant audio could not be completed.");
- setSessionState("error");
+ await failSession("The assistant audio could not be completed.");
}The backend reports assistant output failures as recoverable (assistant_output_failed). If a failed generation must keep the session alive, keep the socket open and return to listening instead of entering error. Confirm the intended behavior.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| } else { | |
| setError("The assistant audio could not be completed."); | |
| setSessionState("error"); | |
| } | |
| } else { | |
| await failSession("The assistant audio could not be completed."); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@frontend/app/hooks/useVoiceSession.ts` around lines 227 - 230, Update the
assistant-output failure branch in useVoiceSession to call failSession instead
of directly setting the error state, ensuring cleanup and consistent failure
teardown for MediaRecorder, microphone tracks, WebSocket, and LiveKit resources.
Preserve the existing user-facing error message unless the intended
recoverable-session behavior is confirmed; if recovery is required, keep the
socket open and transition back to listening instead.
1 participant
Adds a hybrid real-time voice pipeline that keeps browser microphone/Deepgram input on the existing WebSocket while publishing ElevenLabs-generated assistant speech as a native LiveKit audio track.
Changes
Testing
cd backend && python3 -m pytest tests/unit -q— passed, 29 tests.cd backend && python3 -m pytest tests/integration -m 'not external' -q— passed, 10 tests.cd frontend && npm run lint— passed.cd frontend && npm run test -- --run— passed, 14 tests.cd frontend && npx tsc --noEmit— passed.cd frontend && npm run build— passed.client.ready→ Listening → End → Ready./v1/voicescheck to return HTTP 401; a replacement key has been requested.Evidence:
Attached Images and Videos
🎥 View recording: final-review-fake-mic-lifecycle.webm
Session Details
(aside)to your comment to have me ignore it.Summary by CodeRabbit
New Features
Documentation
Tests