Feat/lemonslice plugin stream - #628
Conversation
📝 WalkthroughWalkthroughLemonSlice now uses GetStream for RTC calls instead of LiveKit. The avatar accepts Stream credentials and a configurable call type. The RTC manager creates Stream calls, publishes avatar tracks, handles media, sends custom call events, and cleans up resources. 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: 5
🧹 Nitpick comments (4)
plugins/lemonslice/tests/test_lemonslice_plugin.py (1)
33-39: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winCover each missing Stream credential.
This test removes both credentials at once. It does not verify failure when only
STREAM_API_KEYor onlySTREAM_API_SECRETis missing. Add parameterized cases with the other credential present. Rename the test to reflect credentials plural.Based on the supplied
plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.pycontract, both fields are validated independently. As per coding guidelines, assert the validation behavior and exception message.Proposed test coverage
- async def test_init_missing_stream_secret_raises( + `@pytest.mark.parametrize`( + ("stream_api_key", "stream_api_secret"), + [(None, None), (None, "secret"), ("key", None)], + ) + async def test_init_missing_stream_credentials_raises( self, monkeypatch: pytest.MonkeyPatch ): ... - _make_avatar(stream_api_key=None, stream_api_secret=None) + _make_avatar( + stream_api_key=stream_api_key, + stream_api_secret=stream_api_secret, + )Source: Coding guidelines
plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py (2)
116-118: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the unused
FrameResampler.
self._resampleris assigned here and never read.send_audiowrites PCM straight toAvatarInputTrack, and the track keeps its own resampler from the base class. TheFrameResamplerimport becomes unnecessary too.♻️ Proposed cleanup
self._call: Call | None = None self._connection: ConnectionManager | None = None self._input_track: AvatarInputTrack | None = None - self._resampler = FrameResampler( - rate=_AVATAR_AUDIO_SAMPLE_RATE, layout="mono", format="s16", frame_size=0 - ) self._connected = FalseAnd at line 19:
-from getstream.video.rtc.track_util import FrameResampler, PcmData +from getstream.video.rtc.track_util import PcmData
190-190: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReplace
Anywith concrete types in the event handlers.
userinon_track_addedandeventinon_call_endedare typed asAny, which violates the Python guidelines. Use the participant type already used byon_participant_left, and the concrete call-ended event payload type forcall_ended.Applies to lines 190 and 216.
Source: Coding guidelines
plugins/lemonslice/tests/test_track.py (1)
23-26: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRename the test class to
TestAvatarInputTrack.The class under test is
AvatarInputTrack.TestStampedAudioTracklooks like a leftover from a rename. The guidelines require unit tests for a class to live in the matching test class.♻️ Proposed rename
-class TestStampedAudioTrack: +class TestAvatarInputTrack:As per coding guidelines: "Keep unit tests for a class under the same test class. Do not spread tests around different test classes (e.g., tests for
Agentmust be insideTestAgent)".Source: Coding guidelines
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 8b653374-6092-4ff8-84bb-1af3d512ef86
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (9)
plugins/lemonslice/README.mdplugins/lemonslice/example/lemonslice_avatar_example.pyplugins/lemonslice/pyproject.tomlplugins/lemonslice/tests/test_lemonslice_plugin.pyplugins/lemonslice/tests/test_track.pyplugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.pyplugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.pyplugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.pyplugins/lemonslice/vision_agents/plugins/lemonslice/track.py
💤 Files with no reviewable changes (2)
- plugins/lemonslice/example/lemonslice_avatar_example.py
- plugins/lemonslice/pyproject.toml
| connection = await rtc.join( | ||
| call, | ||
| self._plugin_user_id, | ||
| subscription_config=subscription_config, | ||
| ) | ||
| self._connection = connection | ||
|
|
||
| input_track = AvatarInputTrack( | ||
| sample_rate=_AVATAR_AUDIO_SAMPLE_RATE, | ||
| channels=_AVATAR_AUDIO_CHANNELS, | ||
| ) | ||
| self._input_track = input_track | ||
|
|
||
| @connection.on("track_added") | ||
| async def on_track_added(track_id: str, kind: str, user: Any) -> None: | ||
| if user is None or user.user_id != self._avatar_user_id: | ||
| return | ||
|
|
||
| if kind == "video": | ||
| logger.info("Received video track from LemonSlice avatar") | ||
| track = connection.subscriber_pc.add_track_subscriber(track_id) | ||
| if track is not None: | ||
| self._create_task(self._consume_video(track)) | ||
|
|
||
| @connection.on("audio") | ||
| async def on_audio(pcm: PcmData) -> None: | ||
| participant = pcm.participant | ||
| if participant is None or participant.user_id != self._avatar_user_id: | ||
| return | ||
| await self._on_audio(pcm) | ||
|
|
||
| @connection.on("participant_left") | ||
| async def on_participant_left(event: events_pb2.ParticipantLeft) -> None: | ||
| if event.participant.user_id != self._avatar_user_id: | ||
| return | ||
| logger.info("LemonSlice avatar left the call") | ||
| self._connected = False | ||
| self._create_task(self._on_disconnect()) | ||
| if self._room is not None: | ||
| self._create_task(self._room.disconnect()) | ||
|
|
||
| @room.on("disconnected") | ||
| def on_disconnected(reason: str) -> None: | ||
| # The "disconnected" callback may be triggered multiple times | ||
| # because we disconnect ourselves when the avatar leaves the call. | ||
| if self._connected: | ||
| logger.info(f"Room disconnected; reason: {reason}") | ||
| self._connected = False | ||
| self._create_task(self._on_disconnect()) | ||
|
|
||
| logger.info(f"Connecting to LiveKit room {credentials.room_name}") | ||
| await room.connect(self._livekit_url, credentials.agent_token) | ||
| logger.info(f"Connected to LiveKit room {credentials.room_name}") | ||
|
|
||
| room.local_participant.register_rpc_method( | ||
| "lk.playback_finished", self._rpc_on_playback_finished | ||
| ) | ||
|
|
||
| self._room = room | ||
| @connection.on("call_ended") | ||
| async def on_call_ended(event: Any) -> None: | ||
| if not self._connected: | ||
| return | ||
| logger.info("Stream call ended") | ||
| self._connected = False | ||
| self._create_task(self._on_disconnect()) | ||
|
|
||
| logger.info( | ||
| f"Joining Stream call {credentials.call_type}:{credentials.call_id}" | ||
| ) | ||
| await connection.__aenter__() | ||
| await connection.add_tracks(audio=input_track) | ||
| await connection.republish_tracks() | ||
| self._connected = True |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Clean up partial state if connect() fails after rtc.join.
rtc.join and connection.__aenter__() succeed before add_tracks and republish_tracks run. If either later call raises, connect() propagates and the joined connection stays open. LemonSliceAvatar._connect only calls self._rtc_manager.close() when create_session fails, so nothing releases the RTC connection or the AsyncStream client on this path.
Wrap the post-join steps and call close() on failure.
🛡️ Proposed fix
- await connection.__aenter__()
- await connection.add_tracks(audio=input_track)
- await connection.republish_tracks()
- self._connected = True
+ try:
+ await connection.__aenter__()
+ await connection.add_tracks(audio=input_track)
+ await connection.republish_tracks()
+ except Exception:
+ await self.close()
+ raise
+ self._connected = True📝 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.
| connection = await rtc.join( | |
| call, | |
| self._plugin_user_id, | |
| subscription_config=subscription_config, | |
| ) | |
| self._connection = connection | |
| input_track = AvatarInputTrack( | |
| sample_rate=_AVATAR_AUDIO_SAMPLE_RATE, | |
| channels=_AVATAR_AUDIO_CHANNELS, | |
| ) | |
| self._input_track = input_track | |
| @connection.on("track_added") | |
| async def on_track_added(track_id: str, kind: str, user: Any) -> None: | |
| if user is None or user.user_id != self._avatar_user_id: | |
| return | |
| if kind == "video": | |
| logger.info("Received video track from LemonSlice avatar") | |
| track = connection.subscriber_pc.add_track_subscriber(track_id) | |
| if track is not None: | |
| self._create_task(self._consume_video(track)) | |
| @connection.on("audio") | |
| async def on_audio(pcm: PcmData) -> None: | |
| participant = pcm.participant | |
| if participant is None or participant.user_id != self._avatar_user_id: | |
| return | |
| await self._on_audio(pcm) | |
| @connection.on("participant_left") | |
| async def on_participant_left(event: events_pb2.ParticipantLeft) -> None: | |
| if event.participant.user_id != self._avatar_user_id: | |
| return | |
| logger.info("LemonSlice avatar left the call") | |
| self._connected = False | |
| self._create_task(self._on_disconnect()) | |
| if self._room is not None: | |
| self._create_task(self._room.disconnect()) | |
| @room.on("disconnected") | |
| def on_disconnected(reason: str) -> None: | |
| # The "disconnected" callback may be triggered multiple times | |
| # because we disconnect ourselves when the avatar leaves the call. | |
| if self._connected: | |
| logger.info(f"Room disconnected; reason: {reason}") | |
| self._connected = False | |
| self._create_task(self._on_disconnect()) | |
| logger.info(f"Connecting to LiveKit room {credentials.room_name}") | |
| await room.connect(self._livekit_url, credentials.agent_token) | |
| logger.info(f"Connected to LiveKit room {credentials.room_name}") | |
| room.local_participant.register_rpc_method( | |
| "lk.playback_finished", self._rpc_on_playback_finished | |
| ) | |
| self._room = room | |
| @connection.on("call_ended") | |
| async def on_call_ended(event: Any) -> None: | |
| if not self._connected: | |
| return | |
| logger.info("Stream call ended") | |
| self._connected = False | |
| self._create_task(self._on_disconnect()) | |
| logger.info( | |
| f"Joining Stream call {credentials.call_type}:{credentials.call_id}" | |
| ) | |
| await connection.__aenter__() | |
| await connection.add_tracks(audio=input_track) | |
| await connection.republish_tracks() | |
| self._connected = True | |
| connection = await rtc.join( | |
| call, | |
| self._plugin_user_id, | |
| subscription_config=subscription_config, | |
| ) | |
| self._connection = connection | |
| input_track = AvatarInputTrack( | |
| sample_rate=_AVATAR_AUDIO_SAMPLE_RATE, | |
| channels=_AVATAR_AUDIO_CHANNELS, | |
| ) | |
| self._input_track = input_track | |
| `@connection.on`("track_added") | |
| async def on_track_added(track_id: str, kind: str, user: Any) -> None: | |
| if user is None or user.user_id != self._avatar_user_id: | |
| return | |
| if kind == "video": | |
| logger.info("Received video track from LemonSlice avatar") | |
| track = connection.subscriber_pc.add_track_subscriber(track_id) | |
| if track is not None: | |
| self._create_task(self._consume_video(track)) | |
| `@connection.on`("audio") | |
| async def on_audio(pcm: PcmData) -> None: | |
| participant = pcm.participant | |
| if participant is None or participant.user_id != self._avatar_user_id: | |
| return | |
| await self._on_audio(pcm) | |
| `@connection.on`("participant_left") | |
| async def on_participant_left(event: events_pb2.ParticipantLeft) -> None: | |
| if event.participant.user_id != self._avatar_user_id: | |
| return | |
| logger.info("LemonSlice avatar left the call") | |
| self._connected = False | |
| self._create_task(self._on_disconnect()) | |
| `@connection.on`("call_ended") | |
| async def on_call_ended(event: Any) -> None: | |
| if not self._connected: | |
| return | |
| logger.info("Stream call ended") | |
| self._connected = False | |
| self._create_task(self._on_disconnect()) | |
| logger.info( | |
| f"Joining Stream call {credentials.call_type}:{credentials.call_id}" | |
| ) | |
| try: | |
| await connection.__aenter__() | |
| await connection.add_tracks(audio=input_track) | |
| await connection.republish_tracks() | |
| except Exception: | |
| await self.close() | |
| raise | |
| self._connected = True |
| async def close(self) -> None: | ||
| """Disconnect from the LiveKit room and clean up resources.""" | ||
| """Leave the Stream call and clean up resources.""" | ||
| try: | ||
| if self._stream_writer is not None: | ||
| await self._stream_writer.aclose() | ||
|
|
||
| await cancel_and_wait(*self._tasks) | ||
| self._tasks.clear() | ||
|
|
||
| if self._room is not None: | ||
| await self._room.disconnect() | ||
| if self._connection is not None: | ||
| await self._connection.leave() | ||
|
|
||
| if self._call is not None: | ||
| await self._call.end() | ||
| await self._client.aclose() | ||
| finally: | ||
| self._room = None | ||
| self._stream_writer = None | ||
| self._connection = None | ||
| self._call = None | ||
| self._input_track = None | ||
| self._connected = False | ||
| logger.debug("LemonSlice RTC manager closed") | ||
|
|
||
| async def _consume_video(self, video_stream: rtc.VideoStream) -> None: | ||
| async for event in video_stream: | ||
| lk_frame = event.frame.convert(rtc.VideoBufferType.RGBA) | ||
| img = Image.frombuffer( | ||
| "RGBA", (lk_frame.width, lk_frame.height), lk_frame.data | ||
| ) | ||
| frame = av.VideoFrame.from_image(img) | ||
| await self._on_video(frame) | ||
|
|
||
| async def _consume_audio(self, audio_stream: rtc.AudioStream) -> None: | ||
| async for event in audio_stream: | ||
| frame = event.frame | ||
| pcm = PcmData.from_bytes( | ||
| frame.data, # type: ignore[arg-type] | ||
| sample_rate=frame.sample_rate, | ||
| format=AudioFormat.S16, | ||
| channels=frame.num_channels, | ||
| ) | ||
| await self._on_audio(pcm) | ||
| logger.debug("LemonSlice Stream RTC manager closed") |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Isolate each cleanup step so one failure does not skip the rest.
If self._connection.leave() raises, self._call.end() and self._client.aclose() never run. The Stream call stays active and the HTTP client leaks. LemonSliceAvatar.close() only logs a warning, so the leak is silent.
🛡️ Proposed fix
async def close(self) -> None:
"""Leave the Stream call and clean up resources."""
try:
await cancel_and_wait(*self._tasks)
self._tasks.clear()
if self._connection is not None:
- await self._connection.leave()
-
- if self._call is not None:
- await self._call.end()
- await self._client.aclose()
+ try:
+ await self._connection.leave()
+ except Exception:
+ logger.exception("Failed to leave the Stream call")
+
+ if self._call is not None:
+ try:
+ await self._call.end()
+ except Exception:
+ logger.exception("Failed to end the Stream call")
+ await self._client.aclose()
finally:As per coding guidelines: "Clean up resources in finally blocks".
📝 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 close(self) -> None: | |
| """Disconnect from the LiveKit room and clean up resources.""" | |
| """Leave the Stream call and clean up resources.""" | |
| try: | |
| if self._stream_writer is not None: | |
| await self._stream_writer.aclose() | |
| await cancel_and_wait(*self._tasks) | |
| self._tasks.clear() | |
| if self._room is not None: | |
| await self._room.disconnect() | |
| if self._connection is not None: | |
| await self._connection.leave() | |
| if self._call is not None: | |
| await self._call.end() | |
| await self._client.aclose() | |
| finally: | |
| self._room = None | |
| self._stream_writer = None | |
| self._connection = None | |
| self._call = None | |
| self._input_track = None | |
| self._connected = False | |
| logger.debug("LemonSlice RTC manager closed") | |
| async def _consume_video(self, video_stream: rtc.VideoStream) -> None: | |
| async for event in video_stream: | |
| lk_frame = event.frame.convert(rtc.VideoBufferType.RGBA) | |
| img = Image.frombuffer( | |
| "RGBA", (lk_frame.width, lk_frame.height), lk_frame.data | |
| ) | |
| frame = av.VideoFrame.from_image(img) | |
| await self._on_video(frame) | |
| async def _consume_audio(self, audio_stream: rtc.AudioStream) -> None: | |
| async for event in audio_stream: | |
| frame = event.frame | |
| pcm = PcmData.from_bytes( | |
| frame.data, # type: ignore[arg-type] | |
| sample_rate=frame.sample_rate, | |
| format=AudioFormat.S16, | |
| channels=frame.num_channels, | |
| ) | |
| await self._on_audio(pcm) | |
| logger.debug("LemonSlice Stream RTC manager closed") | |
| async def close(self) -> None: | |
| """Leave the Stream call and clean up resources.""" | |
| try: | |
| await cancel_and_wait(*self._tasks) | |
| self._tasks.clear() | |
| if self._connection is not None: | |
| try: | |
| await self._connection.leave() | |
| except Exception: | |
| logger.exception("Failed to leave the Stream call") | |
| if self._call is not None: | |
| try: | |
| await self._call.end() | |
| except Exception: | |
| logger.exception("Failed to end the Stream call") | |
| await self._client.aclose() | |
| finally: | |
| self._connection = None | |
| self._call = None | |
| self._input_track = None | |
| self._connected = False | |
| logger.debug("LemonSlice Stream RTC manager closed") |
Source: Coding guidelines
| async def _consume_video(self, track: aiortc.mediastreams.MediaStreamTrack) -> None: | ||
| while True: | ||
| frame = await track.recv() | ||
| if isinstance(frame, av.VideoFrame): | ||
| await self._on_video(frame) | ||
|
|
||
| def _create_task(self, coro: Coroutine[None, None, None]) -> None: | ||
| task: asyncio.Task[None] = asyncio.create_task(coro) | ||
| self._tasks.add(task) | ||
| task.add_done_callback(self._tasks.discard) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Handle track termination and task exceptions.
Two gaps:
_consume_videoloops forever. When the avatar track ends,track.recv()raisesaiortc.mediastreams.MediaStreamErrorand the loop exits through the exception._create_taskinstalls a done callback that only discards the task. No code retrieves the exception, so failures in_consume_videoand inself._on_disconnect()disappear except for an asyncio "Task exception was never retrieved" warning.
lemonslice_avatar.py already uses _task_done_callback for this purpose. Follow the same pattern here.
🛡️ Proposed fix
async def _consume_video(self, track: aiortc.mediastreams.MediaStreamTrack) -> None:
- while True:
- frame = await track.recv()
- if isinstance(frame, av.VideoFrame):
- await self._on_video(frame)
+ try:
+ while True:
+ frame = await track.recv()
+ if isinstance(frame, av.VideoFrame):
+ await self._on_video(frame)
+ except aiortc.mediastreams.MediaStreamError:
+ logger.debug("Avatar video track ended")
def _create_task(self, coro: Coroutine[None, None, None]) -> None:
task: asyncio.Task[None] = asyncio.create_task(coro)
self._tasks.add(task)
- task.add_done_callback(self._tasks.discard)
+ task.add_done_callback(self._on_task_done)
+
+ def _on_task_done(self, task: asyncio.Task[None]) -> None:
+ self._tasks.discard(task)
+ if not task.cancelled() and task.exception() is not None:
+ logger.error("LemonSlice RTC task failed", exc_info=task.exception())📝 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 _consume_video(self, track: aiortc.mediastreams.MediaStreamTrack) -> None: | |
| while True: | |
| frame = await track.recv() | |
| if isinstance(frame, av.VideoFrame): | |
| await self._on_video(frame) | |
| def _create_task(self, coro: Coroutine[None, None, None]) -> None: | |
| task: asyncio.Task[None] = asyncio.create_task(coro) | |
| self._tasks.add(task) | |
| task.add_done_callback(self._tasks.discard) | |
| async def _consume_video(self, track: aiortc.mediastreams.MediaStreamTrack) -> None: | |
| try: | |
| while True: | |
| frame = await track.recv() | |
| if isinstance(frame, av.VideoFrame): | |
| await self._on_video(frame) | |
| except aiortc.mediastreams.MediaStreamError: | |
| logger.debug("Avatar video track ended") | |
| def _create_task(self, coro: Coroutine[None, None, None]) -> None: | |
| task: asyncio.Task[None] = asyncio.create_task(coro) | |
| self._tasks.add(task) | |
| task.add_done_callback(self._on_task_done) | |
| def _on_task_done(self, task: asyncio.Task[None]) -> None: | |
| self._tasks.discard(task) | |
| if not task.cancelled() and task.exception() is not None: | |
| logger.error("LemonSlice RTC task failed", exc_info=task.exception()) |
| async def pts(self) -> int: | ||
| async with self._frame_lock: | ||
| # last emitted PTS + samples still queued = wire PTS of the last buffered sample | ||
| ts = self._timestamp or 0 | ||
| return ts + self._buffered_samples | ||
|
|
||
| async def recv(self) -> av.AudioFrame: | ||
| """Drain buffered audio without pacing; pace only when emitting silence.""" | ||
| if self.readyState != "live": | ||
| raise aiortc.mediastreams.MediaStreamError | ||
|
|
||
| if self._timestamp is None: | ||
| self._timestamp = 0 | ||
| else: | ||
| self._timestamp += self._samples_per_frame | ||
|
|
||
| async with self._frame_lock: | ||
| if not self._frame_buffer: | ||
| # Starved: emit the resampler's partial tail instead of waiting for a full frame. | ||
| for tail in self._resampler.flush(): | ||
| self._frame_buffer.append(tail) | ||
| self._buffered_samples += tail.samples | ||
| frame = self._frame_buffer.popleft() if self._frame_buffer else None | ||
| if frame is not None: | ||
| self._buffered_samples -= frame.samples |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Move the _timestamp update inside _frame_lock.
recv() advances self._timestamp at lines 35-38, outside the lock. pts() reads it under the lock. The lock therefore does not protect _timestamp.
StreamRTCManager.flush() calls pts() from the audio-input task while aiortc calls recv() from the sender task. If pts() runs after the increment but before the frame is popped, the queued frame is counted twice and the reported PTS is one frame too high. The avatar then receives a wrong end-of-utterance position.
🔒️ Proposed fix
async def recv(self) -> av.AudioFrame:
"""Drain buffered audio without pacing; pace only when emitting silence."""
if self.readyState != "live":
raise aiortc.mediastreams.MediaStreamError
- if self._timestamp is None:
- self._timestamp = 0
- else:
- self._timestamp += self._samples_per_frame
-
async with self._frame_lock:
+ if self._timestamp is None:
+ self._timestamp = 0
+ else:
+ self._timestamp += self._samples_per_frame
+
if not self._frame_buffer:📝 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 pts(self) -> int: | |
| async with self._frame_lock: | |
| # last emitted PTS + samples still queued = wire PTS of the last buffered sample | |
| ts = self._timestamp or 0 | |
| return ts + self._buffered_samples | |
| async def recv(self) -> av.AudioFrame: | |
| """Drain buffered audio without pacing; pace only when emitting silence.""" | |
| if self.readyState != "live": | |
| raise aiortc.mediastreams.MediaStreamError | |
| if self._timestamp is None: | |
| self._timestamp = 0 | |
| else: | |
| self._timestamp += self._samples_per_frame | |
| async with self._frame_lock: | |
| if not self._frame_buffer: | |
| # Starved: emit the resampler's partial tail instead of waiting for a full frame. | |
| for tail in self._resampler.flush(): | |
| self._frame_buffer.append(tail) | |
| self._buffered_samples += tail.samples | |
| frame = self._frame_buffer.popleft() if self._frame_buffer else None | |
| if frame is not None: | |
| self._buffered_samples -= frame.samples | |
| async def pts(self) -> int: | |
| async with self._frame_lock: | |
| # last emitted PTS + samples still queued = wire PTS of the last buffered sample | |
| ts = self._timestamp or 0 | |
| return ts + self._buffered_samples | |
| async def recv(self) -> av.AudioFrame: | |
| """Drain buffered audio without pacing; pace only when emitting silence.""" | |
| if self.readyState != "live": | |
| raise aiortc.mediastreams.MediaStreamError | |
| async with self._frame_lock: | |
| if self._timestamp is None: | |
| self._timestamp = 0 | |
| else: | |
| self._timestamp += self._samples_per_frame | |
| if not self._frame_buffer: | |
| # Starved: emit the resampler's partial tail instead of waiting for a full frame. | |
| for tail in self._resampler.flush(): | |
| self._frame_buffer.append(tail) | |
| self._buffered_samples += tail.samples | |
| frame = self._frame_buffer.popleft() if self._frame_buffer else None | |
| if frame is not None: | |
| self._buffered_samples -= frame.samples |
There was a problem hiding this comment.
🧹 Nitpick comments (1)
plugins/lemonslice/vision_agents/plugins/lemonslice/track.py (1)
25-31: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd the constructor return annotation.
Add
-> NonetoAvatarInputTrack.__init__.As per coding guidelines, “Use type annotations everywhere.”
Source: Coding guidelines
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 00dfe4cd-b33b-4a69-8f3f-b018a21c905d
📒 Files selected for processing (2)
plugins/lemonslice/tests/test_track.pyplugins/lemonslice/vision_agents/plugins/lemonslice/track.py
…nding audio (default: 30s)
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py (1)
63-100: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winReject non-positive
avatar_join_timeoutvalues.A zero or negative timeout cannot wait for the avatar participant. Raise
ValueErrorbefore storing the value.Proposed fix
if not stream_api_key or not stream_api_secret: raise ValueError( "Stream API key and secret required. Set STREAM_API_KEY and " "STREAM_API_SECRET environment variables or pass them as parameters." ) + if avatar_join_timeout <= 0: + raise ValueError("avatar_join_timeout must be greater than zero") self._stream_api_key = stream_api_keyAs per coding guidelines: “Raise
ValueErrorwith a descriptive message for invalid args.”Source: Coding guidelines
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 50d5f068-3015-404a-876d-34fe95726ef2
📒 Files selected for processing (4)
plugins/lemonslice/README.mdplugins/lemonslice/tests/test_lemonslice_plugin.pyplugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.pyplugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py
🚧 Files skipped from review as they are similar to previous changes (3)
- plugins/lemonslice/tests/test_lemonslice_plugin.py
- plugins/lemonslice/README.md
- plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (3)
plugins/lemonslice/tests/test_lemonslice_plugin.py (3)
51-57: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winTest each missing Stream credential.
This case covers only both credentials missing. Add cases for a missing
stream_api_key, a missingstream_api_secret, and both missing so incomplete credential pairs cannot pass validation.
86-86: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winClose each injected
httpx.AsyncClient.Both tests create a client and never close it. Use an async fixture or
try/finally, then callaclose()after the request.As per coding guidelines, clean up resources in
finallyblocks.Also applies to: 104-104
Sources: Coding guidelines, MCP tools
96-112: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winAssert every protected Stream field.
The test checks only
transport_typeandproperties["call_id"]. Add conflicting values forcall_type,token, andapi_key, then assert thatLemonSliceClient.create_sessionpreserves the call-generated values.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 11babaa7-52c2-4a9a-8757-fd734bf5c0ff
📒 Files selected for processing (4)
plugins/lemonslice/README.mdplugins/lemonslice/tests/test_lemonslice_plugin.pyplugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.pyplugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py
🚧 Files skipped from review as they are similar to previous changes (2)
- plugins/lemonslice/README.md
- plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py
| api_url: str = DEFAULT_API_URL, | ||
| lemonslice_properties: dict[str, Any] | None = None, |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Expected: callers use api_url, or an explicit compatibility alias exists.
rg -n -C 4 --glob '*.py' 'base_url\s*=|api_url\s*=|LemonSliceClient\(' .Repository: GetStream/Vision-Agents
Length of output: 32284
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect LemonSliceClient, LemonSliceAvatar, and package exports relevant to the constructor signature.
for f in \
plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py \
plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py \
plugins/lemonslice/vision_agents/plugins/lemonslice/__init__.py \
vision_agents/plugins/lemonslice/__init__.py
do
if [ -f "$f" ]; then
echo "===== $f ====="
wc -l "$f"
sed -n '1,180p' "$f"
else
echo "missing: $f"
fi
done
# Structured scan for direct LemonSliceClient calls/overrides outside the client itself.
python3 - <<'PY'
from pathlib import Path
import re
roots = [Path('plugins/lemonslice'), Path('plugins'), Path('vision_agents'), Path('examples'), Path('tests')]
patterns = [re.compile(r'^\s*[^#\s].*\bLemonSliceClient\s*\('), re.compile(r'base_url\s*(=|:)')]
for root in roots:
if not root.exists():
continue
for path in root.rglob('*.py'):
text = path.read_text()
if 'vision_agents.plugins.lemonslice' in text or 'LemonSliceClient' in text or 'base_url' in text:
hits = []
for i, line in enumerate(text.splitlines(), 1):
if any(p.search(line) for p in patterns):
hits.append((i, line.strip()))
if hits:
print(f"\n{path}")
for i, line in hits:
print(f"{i}: {line}")
PYRepository: GetStream/Vision-Agents
Length of output: 17240
Handle the renamed LemonSlice constructor parameter for consumers.
LemonSliceAvatar passes api_url, but LemonSliceClient is exposed by import from this package and only accepts api_url now. If external consumers construct LemonSliceClient with the previous base_url, they will get a constructor keyword error. Keep base_url as a compatibility alias, or document the breaking change.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Expected: api_url comes only from trusted deployment configuration.
rg -n -C 4 'api_url|LEMONSLICE_API_KEY|LemonSliceClient\(' plugins/lemonsliceRepository: GetStream/Vision-Agents
Length of output: 12802
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files with API key or LemonSlice references =="
rg -n -C 3 'LemonSliceClient|lemonslice_api_url|api_url|LEMONSLICE_API_KEY|lemonslice.api_url|api_url' -S .
echo
echo "== relevant plugin and client source =="
sed -n '1,150p' plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py
echo
sed -n '1,130p' plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py
echo
echo "== repository configuration references to lemon slice/env variables =="
rg -n -C 3 -S 'LemonSlice|lemonslice|LEMONSLICE|API_KEY' . | sed -n '1,240p'Repository: GetStream/Vision-Agents
Length of output: 45276
Enforce the LemonSlice host before sending the API credentials.
LemonSliceAvatar(..., api_url=...) can pass an arbitrary URL to LemonSliceClient, which then posts the LemonSlice X-API-Key, Stream call token, and Stream API key to that URL. Default to DEFAULT_API_URL and remove api_url from the public avatar options, or validate it as an HTTPS allowlisted LemonSlice session URL before creating the client.
| response = await self._http_client.post(self._api_url, json=payload) | ||
|
|
||
| if response.status_code != 201: | ||
| if response.status_code >= 400: |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Treat only 2xx responses as successful.
Line 107 allows 1xx and 3xx responses to reach response.json(). A redirect or empty response can then raise a JSON decoding error instead of LemonSliceSessionError.
- if response.status_code >= 400:
+ if not 200 <= response.status_code < 300:📝 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.
| response = await self._http_client.post(self._api_url, json=payload) | |
| if response.status_code != 201: | |
| if response.status_code >= 400: | |
| response = await self._http_client.post(self._api_url, json=payload) | |
| if not 200 <= response.status_code < 300: |
There was a problem hiding this comment.
Actionable comments posted: 1
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 668e59e5-ccff-47a4-9d92-e14cc828a452
📒 Files selected for processing (2)
plugins/lemonslice/tests/test_lemonslice_plugin.pyplugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py
🚧 Files skipped from review as they are similar to previous changes (1)
- plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py
| task = asyncio.create_task(avatar._process_audio_input()) | ||
| stream.send_nowait(AudioOutputFlush()) | ||
| await asyncio.sleep(0.05) | ||
| await cancel_and_wait(task) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Wait for the interrupt event instead of sleeping.
asyncio.sleep(0.05) does not guarantee that _process_audio_input() processed AudioOutputFlush. On a busy CI worker, cancel_and_wait(task) can cancel the task before it emits lemonslice.interrupt.
Make the event capture signal completion. Await that signal with a bounded timeout before cancelling the task.
TranscribingInferenceFlow was emitting final chunks for every TTSOutputChunk with final=True, which may mean the end of a single sentence rather than the whole synthesis.
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
agents-core/vision_agents/core/agents/inference/transcribing_flow.py (1)
541-555: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winTerminate TTS output after a suppressed synthesis failure. If
send_iter()yields audio and then raises,process_tts()emits noTTSOutputEnd.write_audio_output()keepsspeaking=True, so later replies merge with the failed reply. SendTTSOutputEnd(interrupted=True)when synthesis fails.
🧹 Nitpick comments (1)
agents-core/vision_agents/core/agents/inference/transcribing_flow.py (1)
557-561: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd the return annotation.
Add
-> Nonetowrite_audio_output. This method has no return value.As per coding guidelines: “Use type annotations everywhere.”
Source: Coding guidelines
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: cd3abb33-8c98-42d3-9806-2e44a56e7cfd
📒 Files selected for processing (2)
agents-core/vision_agents/core/agents/inference/transcribing_flow.pytests/test_agents/test_inference/test_transcribing_flow.py
1 participant
Updated LemonSlice avatar plugin to use Stream as a video transport