Conversation
|
Important Review skippedReview was skipped due to path filters ⛔ Files ignored due to path filters (1)
CodeRabbit blocks several paths by default. You can override this behavior by explicitly including those paths in the path filters. For example, including ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughAdds a new workspace plugin Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant Agent
participant LocalEdge
participant AudioInput as "Audio Input Device"
participant AudioOutput as "Audio Output Device"
participant LLM
User->>Agent: start / join call
Agent->>LocalEdge: join(agent)
LocalEdge->>AudioInput: start()
LocalEdge->>LocalEdge: spawn mic polling loop
User->>AudioInput: speak (mic)
AudioInput->>LocalEdge: PCM frames
LocalEdge->>Agent: emit AudioReceivedEvent (PCM)
Agent->>LLM: request/process
LLM-->>Agent: text response
Agent->>LocalEdge: publish_tracks(audio_track)
LocalEdge->>LocalEdge: start playback task
LocalEdge->>AudioOutput: write PCM samples
AudioOutput->>User: speaker playback
User->>Agent: finish / close
Agent->>LocalEdge: finish()
LocalEdge->>LocalEdge: stop mic loop, cancel tasks
LocalEdge->>AudioInput: stop()
LocalEdge->>AudioOutput: stop()
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
📝 Coding Plan
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 |
github-actions
bot
added
dependencies
agents-core
plugins
examples
config
docs
project-info
labels
Contributor
|
You have run out of free Bugbot PR reviews for this billing cycle. This will reset on April 7. To receive reviews on all of your PRs, visit the Cursor dashboard to activate Pro and start your 14-day free trial. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 11
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/agents.py (1)
633-696:⚠️ Potential issue | 🟠 MajorWiden the
EdgeTransportcontract to match the agent'sOptional[Call]signature.
Agent.join()acceptsOptional[Call]and passescalldirectly toedge.join()andedge.create_conversation(). However, the abstractEdgeTransportmethods require non-optionalT_CallandCall, forcingLocalTransportto suppress the override mismatch with# type: ignore[override]on both methods. Update the abstract signatures inEdgeTransportto match:
join(self, agent: "Agent", call: T_Call | None, **kwargs)create_conversation(self, call: Call | None, user: User, instructions: str)This eliminates the type suppressions and keeps the contracts aligned.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/agents/agents.py` around lines 633 - 696, Agent.join accepts Optional[Call] but EdgeTransport currently requires non-optional T_Call/Call causing override ignores; update the EdgeTransport abstract method signatures so join(self, agent: "Agent", call: T_Call | None, **kwargs) and create_conversation(self, call: Call | None, user: User, instructions: str) accept None (and adjust any related type variable bounds if necessary), then remove the # type: ignore[override] from LocalTransport.join and LocalTransport.create_conversation and fix any callers to handle Optional[Call] accordingly (e.g., Agent.join and edge.create_conversation usage remains valid).
🧹 Nitpick comments (4)
plugins/getstream/vision_agents/plugins/getstream/stream_edge_transport.py (2)
130-131: Guideline violation:except Exception as eshould catch specific exceptions.This pre-existing code uses the broad
except Exception as epattern, which the coding guidelines explicitly forbid. Consider catching more specific exceptions here (e.g.,asyncio.CancelledError, connection-related errors).♻️ Proposed refactor to catch specific exceptions
- except Exception as e: - logger.error(f"Error during connection close: {e}") + except OSError as e: + logger.error(f"OS error during connection close: {e}")As per coding guidelines: "Never write
except Exception as e. Catch specific exceptions instead."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/getstream/vision_agents/plugins/getstream/stream_edge_transport.py` around lines 130 - 131, The broad `except Exception as e` in the connection close block should be replaced with specific exception handlers: catch asyncio.CancelledError (re-raise immediately), then catch connection-related exceptions such as ConnectionError and OSError to log the error via logger.error, and optionally handle any protocol/library-specific errors used in this module; for any truly unexpected errors, let them propagate instead of silently catching them. Locate the try/except around the connection close (the block that currently logs with logger.error(f"Error during connection close: {e}")) and replace it with multiple specific except clauses (e.g., except asyncio.CancelledError: raise, except ConnectionError/ OSError: logger.error(...)) so the code follows the guideline to not use `except Exception as e`.
317-317: Guideline violation: Avoidhasattr; prefer direct attribute access.The use of
hasattr(event.payload, "type")violates the coding guideline to avoidhasattr. Consider using a type check or catchingAttributeErrorinstead, or restructuring with explicit type narrowing.♻️ Proposed refactor to avoid hasattr
- if hasattr(event.payload, "type") and event.payload is not None: + if isinstance(event, sfu_events.TrackUnpublishedEvent) and event.payload is not None: # TrackUnpublishedEvent - single track tracks_to_remove = [event.payload.type] event_desc = "Track unpublished"As per coding guidelines: "Avoid
getattr,hasattr,delattr,setattr; prefer normal attribute access."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/getstream/vision_agents/plugins/getstream/stream_edge_transport.py` at line 317, The code uses hasattr(event.payload, "type"); replace it by first checking event.payload is not None and then access the attribute directly inside a try/except AttributeError (or perform an explicit isinstance check if you have a concrete payload class) — e.g. ensure you test "if event.payload is not None:" then do "try: payload_type = event.payload.type; except AttributeError: payload_type = None" (or use isinstance(payload, YourPayloadClass) to narrow the type) and use payload_type instead of calling hasattr; this change should be applied where event.payload and its "type" attribute are referenced in stream_edge_transport.py.examples/10_local_transport_example/README.md (2)
35-65: Prerequisites and setup instructions are clear.The organization of API keys by example type is helpful, and the .env template provides good guidance. The setup steps are straightforward.
Optional enhancement: Consider adding system requirements (Python version, supported OS platforms) to help users quickly determine compatibility.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@examples/10_local_transport_example/README.md` around lines 35 - 65, Add a short "System requirements" subsection under the existing "Prerequisites" heading that lists the minimum runtime and platform requirements (e.g., required Python/Node major version like "Python 3.10+ or Node 18+", supported OS: Windows/macOS/Linux, and RAM/CPU guidance if relevant) and mention any required package managers or tooling; also include a one-line note in the "Setup" section advising which runtime/version the provided .env template and commands assume so users can verify compatibility.
102-118: Troubleshooting section is helpful; minor clarity improvement suggested.The troubleshooting guidance provides practical solutions for common issues. Line 108 references
list_audio_devices()but doesn't remind users that it needs to be imported first (as shown in lines 82-84).Optional improvement: For consistency, consider rephrasing line 108 to reference the "Listing Audio Devices" section or add a brief import reminder.
📝 Suggested clarification
-2. Run `list_audio_devices()` to see available devices +2. Run `list_audio_devices()` to see available devices (see "Listing Audio Devices" section above for import)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@examples/10_local_transport_example/README.md` around lines 102 - 118, Update the Troubleshooting "No audio input/output" step that mentions list_audio_devices() to either reference the existing "Listing Audio Devices" section or add a brief import reminder; specifically mention the need to import list_audio_devices (as shown earlier in the README) before calling it and keep the LocalTransport and blocksize tips unchanged for context.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@agents-core/vision_agents/core/edge/local_devices.py`:
- Around line 72-75: The code assumes sd.query_devices() returns devices and
sd.default.device contains valid indices, causing invalid default dereferences
and bad prompts; update the selection flow around sd.query_devices(),
sd.default.device, and get_device_sample_rate to first validate that devices is
non-empty and that sd.default.device exists and its indices are within the
filtered devices list, fallback to None or a safe sentinel when defaults are
missing, avoid printing invalid ranges like [0--1], and ensure
get_device_sample_rate(None, ...) and any prompt logic handle a
None/default-missing case gracefully (skip prompting or present only valid
choices) so no invalid index or name is dereferenced.
- Around line 157-159: Remove the unnecessary PyAV pre-check in list_cameras:
delete the call to _check_pyav() from list_cameras() so camera enumeration (and
select_video_device()) can run without PyAV installed; leave PyAV checks to
create_video_track() (which already handles missing PyAV with a warning). Locate
the _check_pyav() invocation inside list_cameras and remove it, ensuring the
function continues to use subprocess/glob for enumeration and its docstring
remains accurate.
In `@agents-core/vision_agents/core/edge/local_transport.py`:
- Around line 388-391: Constructor saves self._sample_rate and
self._output_channels but create_audio_track() is still hardcoding 48000 Hz and
stereo; update create_audio_track (and the duplicate block around the second
occurrence) to default to self._sample_rate and self._output_channels when no
sample_rate/output_channels args are provided so the transport honors configured
output settings (refer to symbols: create_audio_track, self._sample_rate,
self._output_channels, and the duplicate block around lines 526-537).
- Around line 404-406: The unbounded self._input_queue allows the PortAudio
thread's _microphone_callback to enqueue audio forever; change the queue to a
bounded asyncio.Queue by initializing self._input_queue = asyncio.Queue(maxsize=
N) (choose a sensible N), and update _microphone_callback to handle a full queue
(use put_nowait inside try/except asyncio.QueueFull or drop the oldest frame by
doing a non-blocking get_nowait before put_nowait) so backpressure results in
controlled audio drops rather than unbounded memory growth; apply the same
bounding + full-queue handling for the other queue(s) created around the 443-446
region.
- Around line 318-325: The code sets frame.time_base using av.Fraction which is
not part of PyAV's stable public API; replace av.Fraction with
fractions.Fraction to avoid relying on av internals—import fractions.Fraction at
top (or reference fractions) and set frame.time_base = fractions.Fraction(1,
self._fps) in the blank-frame fallback where frame is created (references:
frame, frame.time_base, self._fps, creation block using self._width/self._height
and self._frame_count).
In `@examples/10_local_transport_example/local_transport_example.py`:
- Around line 53-67: The code passes output_sample_rate into
LocalTransport(sample_rate=...), which is then used by
LocalTransport._start_audio() for the microphone InputStream, causing wrong
capture rate when input/output differ; update the call to pass input_sample_rate
as the sample_rate argument (or modify LocalTransport to accept separate
input_sample_rate/output_sample_rate and ensure LocalTransport._start_audio()
uses the input_sample_rate when opening the InputStream) so that
input_sample_rate is actually used for microphone capture (referencing
input_sample_rate, output_sample_rate, LocalTransport, and _start_audio).
- Around line 107-109: The signal registration using asyncio.get_event_loop()
and loop.add_signal_handler(sig, signal_handler) is Unix-only and will raise
NotImplementedError on Windows; update the startup to detect the platform or the
loop's support for add_signal_handler (e.g., check if getattr(loop,
"add_signal_handler", None) is callable or use sys.platform != "win32") before
calling loop.add_signal_handler for SIGINT and SIGTERM, leaving the existing
signal_handler function intact and only registering signals when supported.
In `@examples/10_local_transport_example/README.md`:
- Around line 77-85: The README imports list_audio_devices from the wrong
module; replace the incorrect import path that uses
vision_agents.core.edge.local_transport with the correct module
vision_agents.core.edge.local_devices so that list_audio_devices is imported
from local_devices (update the import statement referencing list_audio_devices
and remove the old local_transport reference).
- Around line 86-101: Update the README example to include the LocalTransport
video parameters by adding video_device, video_width, video_height, and
video_fps to the LocalTransport constructor call; explicitly show video_device:
str | None (e.g., "/dev/video0" or index) to enable camera, and include the
integer defaults video_width=640, video_height=480, video_fps=30 alongside the
existing audio params so the example demonstrates how to enable and configure
video when instantiating LocalTransport.
In `@plugins/getstream/vision_agents/plugins/getstream/stream_edge_transport.py`:
- Around line 466-470: The docstring update introduced formatting that fails
ruff checks; run `ruff format` on the file to reformat the docstring and other
style issues, then re-run checks; specifically locate the triple-quoted
docstring containing "Publish the agent's media tracks to participants." and the
line with "Audio Direction: OUTPUT" and ensure paragraph/line breaks follow
ruff/Google-style expectations (or move the "Audio Direction" note into a
dedicated Note: section) so the file (stream_edge_transport.py) passes
lint/formatting.
In `@tests/test_local_transport.py`:
- Around line 3-64: The tests currently patch sys.modules with fake
sounddevice/av but never reload the target module, causing order-dependent
caching; update tests that patch sys.modules (affecting LocalTransport,
LocalOutputAudioTrack, list_audio_devices) to either stop mocking sys.modules
(use fixture-based fakes) or, if you must patch sys.modules, call
importlib.reload on the module under test (the module that defines
LocalTransport/LocalOutputAudioTrack/list_audio_devices) immediately after the
patch.dict context so module-level imports/availability flags are re-evaluated
and the tests become deterministic.
---
Outside diff comments:
In `@agents-core/vision_agents/core/agents/agents.py`:
- Around line 633-696: Agent.join accepts Optional[Call] but EdgeTransport
currently requires non-optional T_Call/Call causing override ignores; update the
EdgeTransport abstract method signatures so join(self, agent: "Agent", call:
T_Call | None, **kwargs) and create_conversation(self, call: Call | None, user:
User, instructions: str) accept None (and adjust any related type variable
bounds if necessary), then remove the # type: ignore[override] from
LocalTransport.join and LocalTransport.create_conversation and fix any callers
to handle Optional[Call] accordingly (e.g., Agent.join and
edge.create_conversation usage remains valid).
---
Nitpick comments:
In `@examples/10_local_transport_example/README.md`:
- Around line 35-65: Add a short "System requirements" subsection under the
existing "Prerequisites" heading that lists the minimum runtime and platform
requirements (e.g., required Python/Node major version like "Python 3.10+ or
Node 18+", supported OS: Windows/macOS/Linux, and RAM/CPU guidance if relevant)
and mention any required package managers or tooling; also include a one-line
note in the "Setup" section advising which runtime/version the provided .env
template and commands assume so users can verify compatibility.
- Around line 102-118: Update the Troubleshooting "No audio input/output" step
that mentions list_audio_devices() to either reference the existing "Listing
Audio Devices" section or add a brief import reminder; specifically mention the
need to import list_audio_devices (as shown earlier in the README) before
calling it and keep the LocalTransport and blocksize tips unchanged for context.
In `@plugins/getstream/vision_agents/plugins/getstream/stream_edge_transport.py`:
- Around line 130-131: The broad `except Exception as e` in the connection close
block should be replaced with specific exception handlers: catch
asyncio.CancelledError (re-raise immediately), then catch connection-related
exceptions such as ConnectionError and OSError to log the error via
logger.error, and optionally handle any protocol/library-specific errors used in
this module; for any truly unexpected errors, let them propagate instead of
silently catching them. Locate the try/except around the connection close (the
block that currently logs with logger.error(f"Error during connection close:
{e}")) and replace it with multiple specific except clauses (e.g., except
asyncio.CancelledError: raise, except ConnectionError/ OSError:
logger.error(...)) so the code follows the guideline to not use `except
Exception as e`.
- Line 317: The code uses hasattr(event.payload, "type"); replace it by first
checking event.payload is not None and then access the attribute directly inside
a try/except AttributeError (or perform an explicit isinstance check if you have
a concrete payload class) — e.g. ensure you test "if event.payload is not None:"
then do "try: payload_type = event.payload.type; except AttributeError:
payload_type = None" (or use isinstance(payload, YourPayloadClass) to narrow the
type) and use payload_type instead of calling hasattr; this change should be
applied where event.payload and its "type" attribute are referenced in
stream_edge_transport.py.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 9c0cfb27-dc03-492a-bd4a-2a45bcd5fd5d
📒 Files selected for processing (10)
agents-core/pyproject.tomlagents-core/vision_agents/core/agents/agents.pyagents-core/vision_agents/core/edge/local_devices.pyagents-core/vision_agents/core/edge/local_transport.pyexamples/10_local_transport_example/README.mdexamples/10_local_transport_example/__init__.pyexamples/10_local_transport_example/local_transport_example.pyexamples/10_local_transport_example/pyproject.tomlplugins/getstream/vision_agents/plugins/getstream/stream_edge_transport.pytests/test_local_transport.py
plugins/getstream/vision_agents/plugins/getstream/stream_edge_transport.py
Outdated
Show resolved
Hide resolved
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (4)
agents-core/vision_agents/core/edge/edge_transport.py (1)
130-134: Add docstring for the abstract method.The
create_conversationabstract method lacks a docstring explaining the contract, parameters, and expected behavior whencallisNone. This would help implementers understand when they can expect aNonecall and how to handle it.`@abc.abstractmethod` async def create_conversation( self, call: Optional[Call], user: User, instructions: str ): + """Create a conversation for the call session. + + Args: + call: The call to create a conversation for, or None for local + transports that don't require a call context. + user: The user creating the conversation. + instructions: System instructions for the conversation. + + Returns: + A Conversation instance, or None if conversations are not supported. + """ pass🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/edge_transport.py` around lines 130 - 134, Add a clear docstring to the abstractmethod create_conversation(self, call: Optional[Call], user: User, instructions: str) describing the method contract: explain what the method should do, document each parameter (Call, User, instructions), state that call may be None and describe the expected behavior/handling when call is None (e.g., start a new call context or operate in offline mode), specify the return type or raised exceptions implementers should use, and note any side effects or async expectations so subclasses implementing create_conversation know how to behave.agents-core/vision_agents/core/agents/agents.py (1)
814-820: Redundant None check inside_start_tracing.The method is only called when
call is not None(line 661-662), yet line 816 checkscall is not Noneagain. This redundancy suggests either the guard at the call-site or the internal check is unnecessary.Since
_start_tracingreceives aCalltype (notOptional[Call]), the internal check is inconsistent with the type signature.def _start_tracing(self, call: Call) -> None: self._root_span = self.tracer.start_span("join").__enter__() - call_id = call.id if call is not None else "" + call_id = call.id self._root_span.set_attribute("call_id", call_id)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/agents/agents.py` around lines 814 - 820, The _start_tracing method currently performs a redundant None check on call even though its signature requires a Call; remove the unnecessary conditional and directly assign call_id = call.id, update the call_id attribute setting to use that value, and keep the rest (starting span, setting agent_id on self._root_span, and setting _root_ctx) unchanged; update references inside _start_tracing (function name _start_tracing, parameter call, attribute _root_span, and _root_ctx) so the method body assumes a non-None Call per its type signature.tests/test_local_transport.py (1)
240-258: Test creates a MagicMock agent instead of a real or minimal fake.mock_agent = MagicMock() mock_agent.agent_user = MagicMock()This couples tests to internal implementation details. Consider creating a minimal test fixture or dataclass that satisfies the
Agentinterface requirements.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/test_local_transport.py` around lines 240 - 258, The test test_join_starts_microphone uses a generic MagicMock for the agent which couples the test to internals; replace the MagicMock with a minimal concrete test double that implements only the attributes/methods LocalTransport.join needs (e.g., a simple dataclass or lightweight class with an agent_user attribute and any callables expected by LocalTransport.join), instantiate that minimal Agent substitute in place of mock_agent, and update assertions accordingly so the test exercises LocalTransport.join and microphone start without relying on MagicMock internals.agents-core/vision_agents/core/edge/local_transport.py (1)
556-564: Multiple# type: ignore[override]comments indicate interface mismatch.Lines 510, 556, and 602 use
# type: ignore[override]. These suppressions suggestLocalTransportdoesn't fully conform to theEdgeTransportinterface.Consider either:
- Updating
EdgeTransportto better accommodate local transport scenarios- Using proper type signatures that match the base class
As per coding guidelines: "Avoid
# type: ignorecomments."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_transport.py` around lines 556 - 564, The methods in LocalTransport (notably add_track_subscriber and join) use "# type: ignore[override]" because their signatures don't match the EdgeTransport interface; fix this by aligning their type signatures with the base interface (or update EdgeTransport if LocalTransport truly needs different types): ensure add_track_subscriber has the exact return and parameter types EdgeTransport declares (e.g., Optional[LocalVideoTrack] vs LocalVideoTrack | None), make join match the declared return type (LocalConnection) and parameter types, and remove the type-ignore comments; optionally add the typing.override decorator (from typing or typing_extensions) to each overridden method to make the intent explicit and let the typechecker verify conformity.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@agents-core/vision_agents/core/agents/agents.py`:
- Around line 656-667: The join method can call create_conversation with
call=None which crashes for stream transports
(StreamEdgeTransport.create_conversation raises ValueError when call is None);
update join to guard against call being None before invoking create_conversation
(or detect stream-style transports and raise a clear, descriptive error) —
locate the join function and its use of create_conversation and modify the flow
so that when call is None you either skip calling create_conversation (preserve
local-transport behavior) or raise an explicit error mentioning
StreamEdgeTransport/create_conversation so callers know a non-local transport
requires a call.
In `@agents-core/vision_agents/core/edge/local_transport.py`:
- Around line 602-605: The override create_conversation in LocalTransport
currently uses call: Any with a type-ignore; change its signature to match the
base class (call: Optional[Call]) and remove the "# type: ignore[override]" so
it satisfies the Liskov substitution principle, and explicitly ignore the
parameter inside the body (e.g., name it _call or use a no-op reference) before
returning None; update the function declaration for create_conversation to use
Optional[Call] and keep the implementation behavior unchanged.
---
Nitpick comments:
In `@agents-core/vision_agents/core/agents/agents.py`:
- Around line 814-820: The _start_tracing method currently performs a redundant
None check on call even though its signature requires a Call; remove the
unnecessary conditional and directly assign call_id = call.id, update the
call_id attribute setting to use that value, and keep the rest (starting span,
setting agent_id on self._root_span, and setting _root_ctx) unchanged; update
references inside _start_tracing (function name _start_tracing, parameter call,
attribute _root_span, and _root_ctx) so the method body assumes a non-None Call
per its type signature.
In `@agents-core/vision_agents/core/edge/edge_transport.py`:
- Around line 130-134: Add a clear docstring to the abstractmethod
create_conversation(self, call: Optional[Call], user: User, instructions: str)
describing the method contract: explain what the method should do, document each
parameter (Call, User, instructions), state that call may be None and describe
the expected behavior/handling when call is None (e.g., start a new call context
or operate in offline mode), specify the return type or raised exceptions
implementers should use, and note any side effects or async expectations so
subclasses implementing create_conversation know how to behave.
In `@agents-core/vision_agents/core/edge/local_transport.py`:
- Around line 556-564: The methods in LocalTransport (notably
add_track_subscriber and join) use "# type: ignore[override]" because their
signatures don't match the EdgeTransport interface; fix this by aligning their
type signatures with the base interface (or update EdgeTransport if
LocalTransport truly needs different types): ensure add_track_subscriber has the
exact return and parameter types EdgeTransport declares (e.g.,
Optional[LocalVideoTrack] vs LocalVideoTrack | None), make join match the
declared return type (LocalConnection) and parameter types, and remove the
type-ignore comments; optionally add the typing.override decorator (from typing
or typing_extensions) to each overridden method to make the intent explicit and
let the typechecker verify conformity.
In `@tests/test_local_transport.py`:
- Around line 240-258: The test test_join_starts_microphone uses a generic
MagicMock for the agent which couples the test to internals; replace the
MagicMock with a minimal concrete test double that implements only the
attributes/methods LocalTransport.join needs (e.g., a simple dataclass or
lightweight class with an agent_user attribute and any callables expected by
LocalTransport.join), instantiate that minimal Agent substitute in place of
mock_agent, and update assertions accordingly so the test exercises
LocalTransport.join and microphone start without relying on MagicMock internals.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 22cff578-9c46-4213-a46e-8c5e2381d7f4
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (6)
agents-core/vision_agents/core/agents/agents.pyagents-core/vision_agents/core/edge/edge_transport.pyagents-core/vision_agents/core/edge/local_devices.pyagents-core/vision_agents/core/edge/local_transport.pyplugins/getstream/vision_agents/plugins/getstream/stream_edge_transport.pytests/test_local_transport.py
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (5)
plugins/getstream/vision_agents/plugins/getstream/stream_edge_transport.py (1)
357-359: Add an explicit return type tocreate_conversation.This changed method should declare its return type to keep the edge transport API strongly typed.
♻️ Proposed update
async def create_conversation( self, call: Optional[Call], user: User, instructions: str -): +) -> StreamConversation:As per coding guidelines, "Use type annotations everywhere."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/getstream/vision_agents/plugins/getstream/stream_edge_transport.py` around lines 357 - 359, The create_conversation method currently lacks an explicit return type; add a precise type annotation to its signature (e.g., -> Conversation | Awaitable[Conversation] or the appropriate Optional/Union/Coroutine type used across the transport API) so the edge transport remains strongly typed. Update the signature of async def create_conversation(self, call: Optional[Call], user: User, instructions: str) to include the correct return type used by other transport methods (reference existing return types for similar methods in this module) and run type checks to ensure compatibility with Call, User and downstream callers.tests/test_local_transport.py (2)
398-416: Deeply nested context managers reduce readability.The triple-nested
with patch...blocks are verbose and could benefit from a helper or combined context manager.♻️ Example helper approach
from contextlib import ExitStack `@pytest.fixture` def mock_video_deps(mock_av): """Fixture that patches all video dependencies together.""" with ExitStack() as stack: stack.enter_context(patch.dict("sys.modules", {"av": mock_av})) stack.enter_context(patch("vision_agents.core.edge.local_transport.PYAV_AVAILABLE", True)) stack.enter_context(patch("vision_agents.core.edge.local_transport.AIORTC_AVAILABLE", True)) yield mock_avThen tests become:
async def test_add_track_subscriber_returns_video_track( self, mock_sounddevice, mock_video_deps ): from vision_agents.core.edge.local_transport import LocalTransport # ...🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/test_local_transport.py` around lines 398 - 416, The test test_add_track_subscriber_returns_video_track uses three nested with patch(...) contexts which hurts readability; refactor by collapsing those patches into a single reusable fixture or context manager (e.g., a pytest fixture or helper that patches sys.modules["av"], local_transport.PYAV_AVAILABLE, and local_transport.AIORTC_AVAILABLE together) and update the test to accept that fixture instead of the nested with blocks; locate the patches around vision_agents.core.edge.local_transport (and the test function and LocalTransport.create_video_track / add_track_subscriber calls) to ensure the helper applies the same mocks before importing/using LocalTransport.
228-248: Test relies on timing withasyncio.sleep(0.2).The test
test_audio_track_playback_thread_processes_queueusesasyncio.sleep(0.2)to wait for the playback thread to process data. This can be flaky in CI environments under load. Consider using an event or polling with retry instead.♻️ More robust waiting
async def test_audio_track_playback_thread_processes_queue(self, mock_sounddevice): """Test that the playback thread processes queued data.""" from vision_agents.core.edge.local_transport import LocalOutputAudioTrack output = _FakeAudioOutput(sample_rate=48000, channels=2) track = LocalOutputAudioTrack(audio_output=output) track.start() # Put data in the queue test_data = np.array([100, 200, 300, 400, 500, 600, 700, 800], dtype=np.int16) track._queue.put(test_data) - # Give the playback thread time to process - await asyncio.sleep(0.2) + # Wait for playback thread to process (with timeout) + for _ in range(20): + if track._queue.empty() and len(output.written) > 0: + break + await asyncio.sleep(0.05) # Queue should be empty after processing assert track._queue.empty() assert len(output.written) == 1 track.stop()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/test_local_transport.py` around lines 228 - 248, The test test_audio_track_playback_thread_processes_queue is flaky because it uses a fixed asyncio.sleep(0.2); replace that timing-based wait with a bounded polling loop or an awaitable condition that repeatedly checks track._queue.empty() (and/or len(output.written) > 0) until success or a timeout (e.g., 1s) to avoid CI races; keep the existing assertions (track._queue.empty() and len(output.written) == 1) and still call track.stop() at the end, but remove the fixed sleep and use the retry/polling approach to wait deterministically for LocalOutputAudioTrack to process the queued data.agents-core/vision_agents/core/edge/local_transport.py (2)
656-663:create_audio_trackignoressample_rateandstereoparameters.The method signature accepts
sample_rateandstereoparameters, but the implementation creates the track usingself._audio_outputdirectly, ignoring these arguments. This could confuse callers expecting their parameters to be honored.Past review comment noted this was addressed, but the parameters are still unused in the current implementation.
♻️ Option: Remove unused parameters or document they're ignored
def create_audio_track( - self, sample_rate: int = 48000, stereo: bool = True + self, ) -> "LocalOutputAudioTrack": - """Create an audio track that plays through the audio output backend.""" + """Create an audio track that plays through the audio output backend. + + Note: Sample rate and channel config are determined by the audio_output + backend configured at LocalTransport construction time. + """ self._audio_track = LocalOutputAudioTrack( audio_output=self._audio_output, ) return self._audio_track🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_transport.py` around lines 656 - 663, create_audio_track currently ignores its sample_rate and stereo parameters; update the method so the provided sample_rate and stereo are forwarded into the LocalOutputAudioTrack creation (or removed/marked as ignored). Specifically, modify create_audio_track to pass sample_rate and stereo into the LocalOutputAudioTrack constructor (or, if the class does not accept them, extend LocalOutputAudioTrack to accept and use sample_rate and stereo) instead of only using self._audio_output; reference the create_audio_track method, the LocalOutputAudioTrack class, and self._audio_output when locating the change.
218-224: Consider validating sample alignment.The frame calculation
frames = len(samples) // self._channelsuses integer division. Iflen(samples)isn't evenly divisible bychannels, the reshape will silently truncate samples. While audio samples should be properly aligned, a validation or assertion could prevent subtle bugs.♻️ Optional validation
def write(self, samples: np.ndarray) -> None: """Write flat int16 samples to the device.""" if self._stream is None: return + if len(samples) % self._channels != 0: + logger.warning("Sample count not aligned to channel count, truncating") frames = len(samples) // self._channels audio = samples.reshape(frames, self._channels) self._stream.write(audio)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_transport.py` around lines 218 - 224, The write method may silently truncate misaligned samples because frames = len(samples) // self._channels uses integer division; add a validation in write (method name: write, attributes: self._channels, self._stream) to check that len(samples) % self._channels == 0 and raise a clear ValueError (including len(samples) and self._channels) if not aligned, so the caller is notified instead of silently reshaping/truncating; keep the rest of the logic (compute frames, reshape, write) unchanged once validation passes.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@agents-core/vision_agents/core/edge/local_transport.py`:
- Around line 728-729: The create_call method on LocalTransport currently always
raises NotImplementedError which can surprise runtime callers; update
LocalTransport.create_call to either return a simple stub Call-like object
(e.g., a dict or small dataclass with id/status/timestamps) so codepaths that
expect a call object can proceed, or replace the unconditional raise with a
clearer exception and docs: change the message to include guidance (e.g.,
"LocalTransport does not support create_call — use X or check Y") and add a
module/class-level docstring comment that documents this limitation; locate the
create_call function in class LocalTransport and implement one of these two
options consistently with surrounding error handling and logging conventions.
---
Nitpick comments:
In `@agents-core/vision_agents/core/edge/local_transport.py`:
- Around line 656-663: create_audio_track currently ignores its sample_rate and
stereo parameters; update the method so the provided sample_rate and stereo are
forwarded into the LocalOutputAudioTrack creation (or removed/marked as
ignored). Specifically, modify create_audio_track to pass sample_rate and stereo
into the LocalOutputAudioTrack constructor (or, if the class does not accept
them, extend LocalOutputAudioTrack to accept and use sample_rate and stereo)
instead of only using self._audio_output; reference the create_audio_track
method, the LocalOutputAudioTrack class, and self._audio_output when locating
the change.
- Around line 218-224: The write method may silently truncate misaligned samples
because frames = len(samples) // self._channels uses integer division; add a
validation in write (method name: write, attributes: self._channels,
self._stream) to check that len(samples) % self._channels == 0 and raise a clear
ValueError (including len(samples) and self._channels) if not aligned, so the
caller is notified instead of silently reshaping/truncating; keep the rest of
the logic (compute frames, reshape, write) unchanged once validation passes.
In `@plugins/getstream/vision_agents/plugins/getstream/stream_edge_transport.py`:
- Around line 357-359: The create_conversation method currently lacks an
explicit return type; add a precise type annotation to its signature (e.g., ->
Conversation | Awaitable[Conversation] or the appropriate
Optional/Union/Coroutine type used across the transport API) so the edge
transport remains strongly typed. Update the signature of async def
create_conversation(self, call: Optional[Call], user: User, instructions: str)
to include the correct return type used by other transport methods (reference
existing return types for similar methods in this module) and run type checks to
ensure compatibility with Call, User and downstream callers.
In `@tests/test_local_transport.py`:
- Around line 398-416: The test test_add_track_subscriber_returns_video_track
uses three nested with patch(...) contexts which hurts readability; refactor by
collapsing those patches into a single reusable fixture or context manager
(e.g., a pytest fixture or helper that patches sys.modules["av"],
local_transport.PYAV_AVAILABLE, and local_transport.AIORTC_AVAILABLE together)
and update the test to accept that fixture instead of the nested with blocks;
locate the patches around vision_agents.core.edge.local_transport (and the test
function and LocalTransport.create_video_track / add_track_subscriber calls) to
ensure the helper applies the same mocks before importing/using LocalTransport.
- Around line 228-248: The test test_audio_track_playback_thread_processes_queue
is flaky because it uses a fixed asyncio.sleep(0.2); replace that timing-based
wait with a bounded polling loop or an awaitable condition that repeatedly
checks track._queue.empty() (and/or len(output.written) > 0) until success or a
timeout (e.g., 1s) to avoid CI races; keep the existing assertions
(track._queue.empty() and len(output.written) == 1) and still call track.stop()
at the end, but remove the fixed sleep and use the retry/polling approach to
wait deterministically for LocalOutputAudioTrack to process the queued data.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 92a00aa3-713b-4209-abeb-dd651fc6b324
📒 Files selected for processing (5)
agents-core/pyproject.tomlagents-core/vision_agents/core/agents/agents.pyagents-core/vision_agents/core/edge/local_transport.pyplugins/getstream/vision_agents/plugins/getstream/stream_edge_transport.pytests/test_local_transport.py
🚧 Files skipped from review as they are similar to previous changes (1)
- agents-core/vision_agents/core/agents/agents.py
There was a problem hiding this comment.
Actionable comments posted: 3
♻️ Duplicate comments (6)
examples/10_local_transport_example/local_transport_example.py (1)
105-108:⚠️ Potential issue | 🟠 MajorGuard signal registration on Windows.
asynciodocuments that Windows event loops do not supportloop.add_signal_handler(). Because this example registers it unconditionally insiderun_agent(), startup fails on Windows before the agent ever joins. (docs.python.org)♻️ Minimal fix
- loop = asyncio.get_event_loop() + loop = asyncio.get_running_loop() for sig in (signal.SIGINT, signal.SIGTERM): - loop.add_signal_handler(sig, signal_handler) + if sys.platform != "win32": + loop.add_signal_handler(sig, signal_handler)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@examples/10_local_transport_example/local_transport_example.py` around lines 105 - 108, The signal registration in run_agent() uses loop.add_signal_handler unconditionally which fails on Windows; update run_agent() to guard that call by checking platform support before registering (e.g., use sys.platform != "win32" or hasattr(loop, "add_signal_handler")) and only call loop.add_signal_handler(sig, signal_handler) for signal.SIGINT and signal.SIGTERM when supported; keep using the same signal_handler and loop variables so behavior is unchanged on POSIX but avoids startup failure on Windows.agents-core/vision_agents/core/edge/local_transport.py (2)
466-474:⚠️ Potential issue | 🟠 MajorUse
fractions.Fractionforframe.time_base.PyAV documents
Frame.time_baseasfractions.Fraction | None, and its changelog says settingtime_baseonly accepts aFraction. Usingav.Fractionhere doesn't match the documented type. (pyav.basswood-io.com)♻️ Minimal fix
+from fractions import Fraction ... - frame.time_base = av.Fraction(1, self._fps) # type: ignore[attr-defined] + frame.time_base = Fraction(1, self._fps)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_transport.py` around lines 466 - 474, The assignment to frame.time_base uses av.Fraction but PyAV expects a fractions.Fraction; update the code in local_transport.py so that frame.time_base = fractions.Fraction(1, self._fps) (import fractions at top), keeping the surrounding logic that creates av.VideoFrame and sets frame.pts/frame_count (refer to av.VideoFrame, frame.pts, self._frame_count, and self._fps) so the time_base type matches the documented API.
522-552:⚠️ Potential issue | 🟠 MajorGive
LocalTransportseparate input/output sample-rate settings.
SoundDeviceInputandSoundDeviceOutputalready support different rates, but this constructor feeds them the samesample_rate. That means the microphone gets opened at the speaker's rate whenever the selected devices differ, even though the example computes those rates separately. Please split this into distinct input/output sample-rate parameters, or infer each side from the injected backend.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_transport.py` around lines 522 - 552, The constructor of LocalTransport currently uses a single sample_rate for both SoundDeviceInput and SoundDeviceOutput, causing mismatched-device scenarios; change the signature of __init__ to accept separate input_sample_rate and output_sample_rate (or derive them from provided audio_input/audio_output when non-None), keep the existing sample_rate as a convenience default if you want backward compatibility, and pass input_sample_rate into SoundDeviceInput and output_sample_rate into SoundDeviceOutput when constructing them; ensure constructors for SoundDeviceInput and SoundDeviceOutput are only called when audio_input/audio_output are None so injected backends retain their own rates.agents-core/vision_agents/core/edge/local_devices.py (2)
162-165:⚠️ Potential issue | 🟠 MajorDon't hard-fail camera selection when PyAV isn't installed.
list_cameras()only shells out toffmpeg/ inspects/dev/video*, andselect_video_device()just wraps that result. The_check_pyav()guard turns a missing camera dependency into a startup failure even for audio-only runs, because the example callsselect_video_device()unconditionally. Keep the hard PyAV check inLocalTransport.create_video_track(), where capture actually starts.Also applies to: 258-266
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_devices.py` around lines 162 - 165, Remove the unconditional _check_pyav() call from list_cameras and from the video-selection helper (the block around select_video_device / lines ~258-266) so that missing PyAV does not hard-fail when merely listing or selecting device names; keep the existing PyAV guard only in LocalTransport.create_video_track() where actual capture starts. Specifically, delete the _check_pyav() invocation in list_cameras() and any similar early-exit guard in select_video_device(), ensure those functions continue to enumerate devices via ffmpeg or /dev/video* without raising, and leave _check_pyav() intact in LocalTransport.create_video_track() to enforce dependency presence at capture time.
77-79:⚠️ Potential issue | 🟠 MajorValidate missing defaults before prompting or querying sample rates.
This flow assumes there is at least one filtered input/output device and that the default indices are usable. On machines with no mic/speaker or no default configured, the prompts print invalid ranges, Enter dereferences a bad default on Line 117 and Line 136, and
get_device_sample_rate(None, ...)on Line 156 can fail before the example even starts. Add an explicit "no devices / no valid default" path and resolve defaults against the filtered lists first.Also applies to: 109-147, 151-159
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_devices.py` around lines 77 - 79, The code assumes sd.default.device indices and queried devices exist; fix by validating filtered device lists and resolving default indices against those filtered lists before using them: after devices = sd.query_devices() and before using default_in/default_out, check for empty input/output device lists and return or raise a clear "no devices" path; map sd.default.device indices to the filtered devices (e.g., find matching device by index or name) and only set default_in/default_out if a valid match exists, otherwise treat as no-default and ensure prompts show valid ranges; update the code paths that call get_device_sample_rate(None, ...) (referenced by get_device_sample_rate) so they skip or select a safe fallback sample rate when defaults are absent, and replace any direct dereferences (used later in the prompt logic around the functions that display valid ranges and prompt for selection) with guarded logic that prevents using invalid indices.agents-core/vision_agents/core/agents/agents.py (1)
635-647:⚠️ Potential issue | 🟠 MajorFail fast when
call=Noneis unsupported by the active transport.
join()now advertisescall=None, but Line 697 still unconditionally callsself.edge.create_conversation(call, ...). That keeps the local path working, yet transports that still require a real call will only fail afterstart(), MCP setup, auth, andedge.join()have already run. Please rejectcall=Noneup front unless the transport explicitly supports it.Also applies to: 657-699
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@agents-core/vision_agents/core/edge/local_transport.py`:
- Around line 452-462: In recv(), do not set self._started before attempting to
open the camera; instead attempt the camera open first (call self._open_camera
via the event loop) and only set self._started and self._start_time and
self._loop after the open succeeds, or if you prefer to keep the current order,
catch exceptions from await self._loop.run_in_executor(None, self._open_camera)
and clear/reset self._started on failure so subsequent recv() calls will retry;
update the logic around recv, self._started, self._open_camera,
self._start_time, and self._loop accordingly.
In `@tests/test_local_transport.py`:
- Around line 126-140: The test races and leaks the playback thread by asserting
on track._queue and never stopping the track; update test_audio_track_write (and
the other similar tests) to assert on the observable sink (use output.written)
instead of inspecting LocalOutputAudioTrack._queue, and ensure the track is
stopped in a finally block (call track.stop()) after awaiting track.write(pcm)
so the playback thread is cleaned up; locate usages around track.start(),
track.write(...), and replace the queue assertions with checks against the
_FakeAudioOutput.written buffer and add a try/finally to stop the track.
- Around line 3-5: Remove the unittest.mock import and replace all uses of
MagicMock/patch in this file with pytest-native alternatives: create small
pytest fixtures or simple stub classes that mimic the behavior you need (or
reuse the existing fake backends in this test module) and use pytest's
monkeypatch fixture where you need to swap implementations. Update tests that
currently call MagicMock/patch (the tests around the ranges shown) to accept the
new fixtures or call monkeypatch.setattr(...) to inject stubs; keep the same
assertions but instantiate and configure the stub objects (or the fake backend)
instead of using MagicMock. Ensure any helper names you add are obvious (e.g.,
fake_transport_stub, fake_backend_fixture) and replace references to
MagicMock/patch throughout the file.
---
Duplicate comments:
In `@agents-core/vision_agents/core/edge/local_devices.py`:
- Around line 162-165: Remove the unconditional _check_pyav() call from
list_cameras and from the video-selection helper (the block around
select_video_device / lines ~258-266) so that missing PyAV does not hard-fail
when merely listing or selecting device names; keep the existing PyAV guard only
in LocalTransport.create_video_track() where actual capture starts.
Specifically, delete the _check_pyav() invocation in list_cameras() and any
similar early-exit guard in select_video_device(), ensure those functions
continue to enumerate devices via ffmpeg or /dev/video* without raising, and
leave _check_pyav() intact in LocalTransport.create_video_track() to enforce
dependency presence at capture time.
- Around line 77-79: The code assumes sd.default.device indices and queried
devices exist; fix by validating filtered device lists and resolving default
indices against those filtered lists before using them: after devices =
sd.query_devices() and before using default_in/default_out, check for empty
input/output device lists and return or raise a clear "no devices" path; map
sd.default.device indices to the filtered devices (e.g., find matching device by
index or name) and only set default_in/default_out if a valid match exists,
otherwise treat as no-default and ensure prompts show valid ranges; update the
code paths that call get_device_sample_rate(None, ...) (referenced by
get_device_sample_rate) so they skip or select a safe fallback sample rate when
defaults are absent, and replace any direct dereferences (used later in the
prompt logic around the functions that display valid ranges and prompt for
selection) with guarded logic that prevents using invalid indices.
In `@agents-core/vision_agents/core/edge/local_transport.py`:
- Around line 466-474: The assignment to frame.time_base uses av.Fraction but
PyAV expects a fractions.Fraction; update the code in local_transport.py so that
frame.time_base = fractions.Fraction(1, self._fps) (import fractions at top),
keeping the surrounding logic that creates av.VideoFrame and sets
frame.pts/frame_count (refer to av.VideoFrame, frame.pts, self._frame_count, and
self._fps) so the time_base type matches the documented API.
- Around line 522-552: The constructor of LocalTransport currently uses a single
sample_rate for both SoundDeviceInput and SoundDeviceOutput, causing
mismatched-device scenarios; change the signature of __init__ to accept separate
input_sample_rate and output_sample_rate (or derive them from provided
audio_input/audio_output when non-None), keep the existing sample_rate as a
convenience default if you want backward compatibility, and pass
input_sample_rate into SoundDeviceInput and output_sample_rate into
SoundDeviceOutput when constructing them; ensure constructors for
SoundDeviceInput and SoundDeviceOutput are only called when
audio_input/audio_output are None so injected backends retain their own rates.
In `@examples/10_local_transport_example/local_transport_example.py`:
- Around line 105-108: The signal registration in run_agent() uses
loop.add_signal_handler unconditionally which fails on Windows; update
run_agent() to guard that call by checking platform support before registering
(e.g., use sys.platform != "win32" or hasattr(loop, "add_signal_handler")) and
only call loop.add_signal_handler(sig, signal_handler) for signal.SIGINT and
signal.SIGTERM when supported; keep using the same signal_handler and loop
variables so behavior is unchanged on POSIX but avoids startup failure on
Windows.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 7da64848-17cc-4149-afc2-38169c3f71f2
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (5)
agents-core/vision_agents/core/agents/agents.pyagents-core/vision_agents/core/edge/local_devices.pyagents-core/vision_agents/core/edge/local_transport.pyexamples/10_local_transport_example/local_transport_example.pytests/test_local_transport.py
There was a problem hiding this comment.
♻️ Duplicate comments (6)
agents-core/vision_agents/core/edge/local_transport.py (3)
474-474:⚠️ Potential issue | 🟠 MajorReplace
av.Fractionwithfractions.Fraction.
av.Fractionis not part of PyAV's public API. The documented type forframe.time_baseisfractions.Fractionfrom stdlib.Fix
Add import at top:
from fractions import FractionThen:
- frame.time_base = av.Fraction(1, self._fps) # type: ignore[attr-defined] + frame.time_base = Fraction(1, self._fps)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_transport.py` at line 474, The code sets frame.time_base using av.Fraction which is not part of PyAV's public API; replace the usage with the stdlib fractions.Fraction by importing Fraction at the top and changing the assignment in the code that sets frame.time_base (the line currently reading frame.time_base = av.Fraction(1, self._fps)) to use Fraction(1, self._fps) instead so the documented type is used.
457-461:⚠️ Potential issue | 🟠 MajorSet
_startedonly after camera opens successfully.The flag is set before
_open_camera()runs. If opening fails, subsequentrecv()calls skip the initialization path entirely, leaving the track permanently broken.Fix
if not self._started: - self._started = True - self._start_time = time.time() self._loop = asyncio.get_running_loop() await self._loop.run_in_executor(None, self._open_camera) + self._start_time = time.time() + self._started = True🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_transport.py` around lines 457 - 461, The code sets self._started before actually opening the camera, so if self._open_camera() fails recv() will treat the transport as started and skip initialization; fix by deferring setting self._started (and self._start_time) until after self._open_camera() completes successfully: obtain the event loop (self._loop = asyncio.get_running_loop()) then run self._open_camera() in the executor inside a try/except, only on success set self._started = True and self._start_time = time.time(); ensure exceptions from _open_camera are propagated or logged so callers can retry and that recv() will still perform initialization on next attempt. Include references to _open_camera, recv, self._started, self._start_time, and self._loop when locating the change.
654-661:⚠️ Potential issue | 🟡 Minor
create_audio_trackignoressample_rateandstereoparameters.The method signature accepts
sample_rateandstereoarguments but completely ignores them, always using the audio output backend's configuration. This could surprise callers who expect the parameters to take effect.Consider either:
- Documenting that parameters are ignored (if intentional)
- Using the parameters to configure the track
Option 1: Document the behavior
def create_audio_track( self, sample_rate: int = 48000, stereo: bool = True ) -> "LocalOutputAudioTrack": - """Create an audio track that plays through the audio output backend.""" + """Create an audio track that plays through the audio output backend. + + Note: sample_rate and stereo are ignored; the track uses the + audio output backend's configured sample rate and channels. + """ self._audio_track = LocalOutputAudioTrack(🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_transport.py` around lines 654 - 661, The create_audio_track method currently ignores its sample_rate and stereo parameters; update it to pass these through to the LocalOutputAudioTrack so caller settings are honored: modify create_audio_track(sample_rate: int = 48000, stereo: bool = True) to instantiate LocalOutputAudioTrack(audio_output=self._audio_output, sample_rate=sample_rate, stereo=stereo) and ensure the LocalOutputAudioTrack constructor and any downstream handling (e.g., buffering/resampling code) accept and apply sample_rate and stereo, or alternatively add clear docstring comments on create_audio_track indicating the parameters are intentionally ignored if you choose not to support them. Ensure references to create_audio_track, LocalOutputAudioTrack, and self._audio_output are updated accordingly.agents-core/vision_agents/core/edge/local_devices.py (3)
77-79:⚠️ Potential issue | 🟠 MajorHandle missing devices and invalid defaults before prompting.
When no audio devices are detected or the system has no configured default (
sd.default.devicereturns-1), the prompt displays[0--1]and accessingdevices[default_in]on Enter will raise an IndexError.Suggested validation
devices = sd.query_devices() default_in = sd.default.device[0] default_out = sd.default.device[1] + + if not devices: + raise RuntimeError("No audio devices detected on this system")Then validate
default_in/default_outare within bounds before using them:if choice == "": - input_device = None - print(f" -> Using default: {devices[default_in]['name']}") + if default_in >= 0 and default_in < len(devices): + input_device = None + print(f" -> Using default: {devices[default_in]['name']}") + else: + print(" No default input device configured, please select one") + continue🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_devices.py` around lines 77 - 79, The code reads devices = sd.query_devices() and default_in/default_out = sd.default.device[...] without validating results; update the logic in local_devices.py to handle empty device lists and invalid defaults by checking devices is non-empty and that default_in/default_out are within range (0 <= index < len(devices)) before using them; if sd.query_devices() returns empty, show a clear message and skip prompting, and if sd.default.device returns -1 or out-of-range, treat defaults as None (or use a safe fallback index like 0 only if devices exist) and avoid indexing devices[default_in]/devices[default_out] until validated.
162-164:⚠️ Potential issue | 🟠 MajorRemove unnecessary PyAV gate from camera enumeration.
The
_check_pyav()call prevents camera listing even though this function uses onlysubprocess(ffmpeg) andglob—not PyAV. This blocks audio-only configurations from even seeing available cameras. PyAV is only needed when actually capturing frames inLocalVideoTrack.Fix
def list_cameras() -> list[CameraDevice]: """List available cameras on the system.""" - _check_pyav() - cameras: list[CameraDevice] = []🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_devices.py` around lines 162 - 164, Remove the unnecessary PyAV gating in list_cameras by deleting the _check_pyav() call at the start of the function so camera enumeration (list_cameras) relies only on subprocess/ffmpeg and glob; leave LocalVideoTrack and any places that actually capture frames to continue calling _check_pyav() if needed, and run tests to ensure audio-only configurations can now enumerate cameras without invoking PyAV.
258-264:⚠️ Potential issue | 🟠 MajorSame unnecessary PyAV gate applies here.
Remove
_check_pyav()fromselect_video_device()as well—it callslist_cameras()which doesn't need PyAV.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_devices.py` around lines 258 - 264, Remove the unnecessary PyAV gate from select_video_device(): delete the call to _check_pyav() inside the select_video_device() function so it no longer invokes _check_pyav() before calling list_cameras(); list_cameras() does not require PyAV, so ensure select_video_device() continues to call list_cameras() and perform its interactive logic without the _check_pyav() guard.
🧹 Nitpick comments (2)
agents-core/vision_agents/core/edge/local_transport.py (1)
138-141: Silent drop on queue full may hide issues.When the buffer fills, audio chunks are silently discarded. Consider logging at
debuglevel to aid troubleshooting without spamming logs during normal backpressure.Suggestion
try: self._buffer.put_nowait(indata.copy()) except queue.Full: - pass + pass # Expected during backpressure; enable debug logging if neededOr add occasional logging:
+ self._drop_count = 0 ... except queue.Full: + self._drop_count += 1 + if self._drop_count % 100 == 1: + logger.debug("Audio input buffer full, dropping chunks") pass🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_transport.py` around lines 138 - 141, The current silent drop in the try/except around self._buffer.put_nowait(indata.copy()) swallows queue.Full without any visibility; modify the except queue.Full handler in local_transport.py to emit a debug-level log via the agent's logger (or self.logger) including a short message and context (e.g., buffer size/name) and implement simple throttling (e.g., only log once per N drops or use a timestamp check) to avoid spamming; reference the put_nowait call and the queue.Full exception so you update that specific handler.agents-core/vision_agents/core/edge/local_devices.py (1)
14-28: Type ignore comments for optional dependencies.The
# type: ignore[assignment]on lines 19 and 27 are necessary here for the optional import pattern, but coding guidelines discourage them. Consider using a different pattern such as conditional type definitions or a sentinel module object.That said, this is a common pattern for optional dependencies and the trade-off is reasonable given the runtime flexibility required.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@agents-core/vision_agents/core/edge/local_devices.py` around lines 14 - 28, Optional dependency imports use "# type: ignore[assignment]" which the guidelines discourage; replace those ignores by adding explicit type declarations and/or TYPE_CHECKING guards: import typing (TYPE_CHECKING, Any, Optional), declare sd: Optional[Any] = None and av: Optional[Any] = None before the try blocks (or use a sentinel module object) and keep setting SOUNDDEVICE_AVAILABLE and PYAV_AVAILABLE inside the try/except blocks; ensure references to sd and av remain typed without using "# type: ignore[assignment]" by moving the type info outside the exception handlers (refer to the sd, SOUNDDEVICE_AVAILABLE, av, and PYAV_AVAILABLE symbols).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@agents-core/vision_agents/core/edge/local_devices.py`:
- Around line 77-79: The code reads devices = sd.query_devices() and
default_in/default_out = sd.default.device[...] without validating results;
update the logic in local_devices.py to handle empty device lists and invalid
defaults by checking devices is non-empty and that default_in/default_out are
within range (0 <= index < len(devices)) before using them; if
sd.query_devices() returns empty, show a clear message and skip prompting, and
if sd.default.device returns -1 or out-of-range, treat defaults as None (or use
a safe fallback index like 0 only if devices exist) and avoid indexing
devices[default_in]/devices[default_out] until validated.
- Around line 162-164: Remove the unnecessary PyAV gating in list_cameras by
deleting the _check_pyav() call at the start of the function so camera
enumeration (list_cameras) relies only on subprocess/ffmpeg and glob; leave
LocalVideoTrack and any places that actually capture frames to continue calling
_check_pyav() if needed, and run tests to ensure audio-only configurations can
now enumerate cameras without invoking PyAV.
- Around line 258-264: Remove the unnecessary PyAV gate from
select_video_device(): delete the call to _check_pyav() inside the
select_video_device() function so it no longer invokes _check_pyav() before
calling list_cameras(); list_cameras() does not require PyAV, so ensure
select_video_device() continues to call list_cameras() and perform its
interactive logic without the _check_pyav() guard.
In `@agents-core/vision_agents/core/edge/local_transport.py`:
- Line 474: The code sets frame.time_base using av.Fraction which is not part of
PyAV's public API; replace the usage with the stdlib fractions.Fraction by
importing Fraction at the top and changing the assignment in the code that sets
frame.time_base (the line currently reading frame.time_base = av.Fraction(1,
self._fps)) to use Fraction(1, self._fps) instead so the documented type is
used.
- Around line 457-461: The code sets self._started before actually opening the
camera, so if self._open_camera() fails recv() will treat the transport as
started and skip initialization; fix by deferring setting self._started (and
self._start_time) until after self._open_camera() completes successfully: obtain
the event loop (self._loop = asyncio.get_running_loop()) then run
self._open_camera() in the executor inside a try/except, only on success set
self._started = True and self._start_time = time.time(); ensure exceptions from
_open_camera are propagated or logged so callers can retry and that recv() will
still perform initialization on next attempt. Include references to
_open_camera, recv, self._started, self._start_time, and self._loop when
locating the change.
- Around line 654-661: The create_audio_track method currently ignores its
sample_rate and stereo parameters; update it to pass these through to the
LocalOutputAudioTrack so caller settings are honored: modify
create_audio_track(sample_rate: int = 48000, stereo: bool = True) to instantiate
LocalOutputAudioTrack(audio_output=self._audio_output, sample_rate=sample_rate,
stereo=stereo) and ensure the LocalOutputAudioTrack constructor and any
downstream handling (e.g., buffering/resampling code) accept and apply
sample_rate and stereo, or alternatively add clear docstring comments on
create_audio_track indicating the parameters are intentionally ignored if you
choose not to support them. Ensure references to create_audio_track,
LocalOutputAudioTrack, and self._audio_output are updated accordingly.
---
Nitpick comments:
In `@agents-core/vision_agents/core/edge/local_devices.py`:
- Around line 14-28: Optional dependency imports use "# type:
ignore[assignment]" which the guidelines discourage; replace those ignores by
adding explicit type declarations and/or TYPE_CHECKING guards: import typing
(TYPE_CHECKING, Any, Optional), declare sd: Optional[Any] = None and av:
Optional[Any] = None before the try blocks (or use a sentinel module object) and
keep setting SOUNDDEVICE_AVAILABLE and PYAV_AVAILABLE inside the try/except
blocks; ensure references to sd and av remain typed without using "# type:
ignore[assignment]" by moving the type info outside the exception handlers
(refer to the sd, SOUNDDEVICE_AVAILABLE, av, and PYAV_AVAILABLE symbols).
In `@agents-core/vision_agents/core/edge/local_transport.py`:
- Around line 138-141: The current silent drop in the try/except around
self._buffer.put_nowait(indata.copy()) swallows queue.Full without any
visibility; modify the except queue.Full handler in local_transport.py to emit a
debug-level log via the agent's logger (or self.logger) including a short
message and context (e.g., buffer size/name) and implement simple throttling
(e.g., only log once per N drops or use a timestamp check) to avoid spamming;
reference the put_nowait call and the queue.Full exception so you update that
specific handler.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: fae8341f-707d-409b-895b-b7e9706adb88
📒 Files selected for processing (2)
agents-core/vision_agents/core/edge/local_devices.pyagents-core/vision_agents/core/edge/local_transport.py
turbopuffer plugin fails the pre-commit hooks because of breaking changes in the lib
- AudioInputDevice: calculate blocksize automatically - Add VideoDisplay to render the agent video output - Make audio output blocking to prevent TTS audio overflow
dangusev
approved these changes
Mar 20, 2026
There was a problem hiding this comment.
Actionable comments posted: 9
🧹 Nitpick comments (6)
plugins/local/tests/test_edge.py (1)
216-235: Avoidhasattrand# type: ignorein this test.Line 218 can use direct attribute access, and Line 235 can use a typed stub/cast instead of suppressing the checker. That keeps the test aligned with the repo’s Python rules.
As per coding guidelines, "Avoid using getattr, hasattr, delattr and setattr; prefer normal attribute access in Python" and "Avoid
# type: ignorecomments".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/local/tests/test_edge.py` around lines 216 - 235, The test uses hasattr and a type: ignore; replace hasattr(transport, "open_demo_for_agent") with a direct attribute access/assertion on transport.open_demo_for_agent (e.g., assert callable(transport.open_demo_for_agent)) in test_open_demo_for_agent_exists (using _make_transport to get transport), and replace the type: ignore on the _video_display assignment in test_close_cleans_up_video_display by casting the stub to the proper display type or declaring a typed stub class that matches the transport's expected VideoDisplay interface (e.g., import/cast to the concrete VideoDisplay protocol/class before assigning _FakeDisplay instance to transport._video_display) so the static checker no longer needs suppression while preserving the same runtime behavior with _FakeAgent and _FakeDisplay.plugins/local/vision_agents/plugins/local/tracks.py (2)
145-169: Consider using specific PyAV types instead ofAny.Lines 163-164 use
Anyfor_containerand_stream. Usingav.container.InputContainer | Noneandav.stream.Stream | Nonewould improve type safety and IDE support.♻️ Suggested type annotations
- self._container: Any = None - self._stream: Any = None + self._container: av.container.InputContainer | None = None + self._stream: av.stream.Stream | None = None🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/local/vision_agents/plugins/local/tracks.py` around lines 145 - 169, Replace the broad Any annotations for LocalVideoTrack._container and _stream with the concrete PyAV types: use av.container.InputContainer | None for _container and av.stream.Stream | None for _stream, and add the corresponding imports from the av package at the top of the file; update the __init__ attribute declarations in class LocalVideoTrack accordingly to improve static typing and IDE support while preserving the existing None initialization logic.
67-74: Synchronousstart()creates an asyncio task—requires a running event loop.The method is sync but calls
asyncio.create_task(), which will raiseRuntimeErrorif no event loop is running. Currently safe becausepublish_tracks()(async) calls it, but consider makingstart()async or documenting the precondition.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/local/vision_agents/plugins/local/tracks.py` around lines 67 - 74, The synchronous start() method creates an asyncio task and can raise RuntimeError if no event loop is running; change start to an async def start(self) and update callers (e.g., publish_tracks) to await self.start(), so asyncio.create_task(self._playback_loop()) is invoked from an active event loop; ensure _playback_loop and assignment to self._playback_task remain unchanged and update any callers/tests to await start() accordingly.examples/10_local_transport_example/local_transport_example.py (1)
42-49: Consider narrowing the return type fromdict[str, Any].The
get_weathertool returnsdict[str, Any], but ifget_weather_by_locationhas a more specific return type, propagating it would improve type safety. This is minor given it's a tool callback.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@examples/10_local_transport_example/local_transport_example.py` around lines 42 - 49, The get_weather wrapper in setup_llm currently types its return as dict[str, Any]; change it to the actual return type of get_weather_by_location (e.g., WeatherInfo or whatever specific dataclass/TypedDict/Dict shape it returns) so the type is propagated; update the async def get_weather signature to return that specific type and import or define the corresponding type alias used by get_weather_by_location (reference: setup_llm, get_weather, get_weather_by_location).plugins/local/vision_agents/plugins/local/edge.py (2)
204-214: Remove the stray comment on line 213.Line 213 has an empty
#comment that serves no purpose.♻️ Suggested fix
async def create_conversation( self, call: Any, user: User, instructions: str ) -> None: - # return None🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/local/vision_agents/plugins/local/edge.py` around lines 204 - 214, Remove the stray lone comment character in the create_conversation method of LocalEdge: locate the async def create_conversation(self, call: Any, user: User, instructions: str) -> None function and delete the single '#' line so the method body cleanly returns None without an extraneous comment.
179-179: Empty stubopen_demosatisfies the abstract contract but is a no-op.The base class declares
open_demoas abstract. This ellipsis implementation technically satisfies Python's ABC requirements, but callers might expect it to do something. The actual functionality lives inopen_demo_for_agent. Consider adding a brief docstring or delegating toopen_demo_for_agentif the signatures are compatible.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/local/vision_agents/plugins/local/edge.py` at line 179, The empty stub open_demo currently satisfies the abstract method but does nothing; replace it with a real implementation that either calls open_demo_for_agent with the same args/kwargs (delegating behavior) or, if signatures differ, implement a thin adapter that maps parameters and invokes open_demo_for_agent; alternatively add a short docstring on open_demo explaining it's intentionally a no-op and directing callers to open_demo_for_agent. Ensure you update the function body of open_demo (not just the signature) and reference open_demo_for_agent for delegation so callers get the expected behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@plugins/deepgram/vision_agents/plugins/deepgram/deepgram_stt.py`:
- Around line 304-305: In close(), stop using a broad except Exception and
instead catch the concrete exceptions thrown by the Deepgram async client and IO
operations (e.g., the SDK’s specific websocket/connection exception classes
returned by AsyncV2SocketClient.send_close_stream and the listen.v2.connect
context manager __aexit__ plus asyncio.CancelledError and OSError/IOError as
appropriate); replace logger.warning(...) with logger.exception(...) in those
handlers to preserve tracebacks, handle each exception type separately (log and
suppress only expected teardown errors, re-raise unexpected ones), and reference
AsyncV2SocketClient.send_close_stream and the listen.v2.connect context manager
in your changes so the error-handling is specific and traceable.
In `@plugins/local/tests/conftest.py`:
- Around line 18-29: The fake implementation of _FakeAudioInput.read currently
returns immediately when _data is empty, causing a hot spin; update
_FakeAudioInput.read to match the real AudioInputDevice.read contract by
blocking up to ~100ms when no data is available before returning None (e.g.,
wait with a timeout or sleep loop), so tests that exercise Mic poller timing
(using methods like enqueue, start, read and the started flag) behave like
production; ensure the blocking respects interrupts/stop semantics used by the
test harness.
In `@plugins/local/tests/test_devices.py`:
- Around line 3-162: Rewrite tests to remove unittest.mock usage: replace
patch/MagicMock in TestListAudioDevices and TestCameraEnumeration with pytest
fixtures that provide concrete stubs and temporary files to exercise parsing
logic; for audio tests, create a fixture that monkeypatches the sd.query_devices
and sd.default.device via a real small stub object instead of patch(), and call
list_audio_input_devices and list_audio_output_devices to assert results; for
camera tests, use tmp_path to create fake /dev/video* files and real temp files
containing camera names (and a subprocess stub invoked via a pytest fixture that
simulates stderr output or raises FileNotFoundError) instead of patching
subprocess.run or glob; keep test names and assertions targeting list_cameras
and _get_camera_input_format so behavior is unchanged.
In `@plugins/local/tests/test_edge.py`:
- Around line 131-173: The tests test_emit_audio_event and
test_mic_polling_with_custom_input rely on fixed asyncio.sleep delays and are
flaky; replace the sleeps by using an asyncio.Event (or a shared wait-until
helper) that the on_audio subscriber sets when it receives an
AudioReceivedEvent, then await that event with a reasonable timeout (e.g., 1-2s)
instead of sleeping; for test_emit_audio_event trigger
transport._emit_audio_event(data) and await the event flag, and for
test_mic_polling_with_custom_input await the event after calling
transport.join(_FakeAgent()) and _FakeAudioInput.enqueue(...), then assert
against received_events as before and still close the transport. Ensure you
reference the existing on_audio subscriber, received_events list,
transport._emit_audio_event, transport.join, and _FakeAudioInput.enqueue when
adding the event-wait logic.
In `@plugins/local/tests/test_tracks.py`:
- Around line 33-49: The tests race the playback worker by asserting on
track._queue after track.start() and using fixed sleeps; update the tests (e.g.,
test_audio_track_write and the other cases) to wait for a deterministic signal
instead: modify _FakeAudioOutput to expose an asyncio.Event (or use an
asyncio.Event provided to it) that is set when the fake consumer
receives/consumes samples, then in the tests await that Event (or a specific
consumer callback) before asserting on LocalOutputAudioTrack._queue or stopping
the track; reference _FakeAudioOutput, LocalOutputAudioTrack.write/start/stop,
and the internal _queue to locate where to replace timing-based checks with
awaiting the Event.
In `@plugins/local/vision_agents/plugins/local/display.py`:
- Around line 69-81: The start(self, video_track: MediaStreamTrack) method
currently warns and returns when _TKINTER_AVAILABLE is false, breaking the
documented RuntimeError contract and leaving callers like
LocalEdge.open_demo_for_agent() assuming success; replace the warnings.warn +
return with raising RuntimeError("tkinter is not available; install python3-tk
or equivalent for your platform to use the video display.") so callers receive
the expected exception and _video_display isn't treated as started.
- Around line 116-163: The Tk initialization and setup in _tk_loop should be
moved inside the try/finally block so cleanup always runs if creation fails:
declare root/prev_sigint/_photo/_label vars before the try, then create
tkinter.Tk(), set signal.signal, root.title, geometry, protocol, assign
self._root and widgets inside the try; ensure the finally block destroys root
(if set) and restores the previous SIGINT handler; this prevents _tk_task from
exiting before cleanup while _recv_task continues and avoids stop() re-raising
startup exceptions.
- Around line 51-61: The constructor __init__ currently computes
self._frame_interval = 1.0 / fps without validating inputs, which causes a
ZeroDivisionError for fps==0 and later failures for non-positive dimensions;
update __init__ to validate that fps > 0 and that width and height are positive
integers (and optionally that title is non-empty) and raise descriptive
ValueError messages (e.g., "fps must be > 0", "width must be > 0", "height must
be > 0") before computing self._frame_interval or assigning size fields so
invalid arguments fail fast.
In `@plugins/local/vision_agents/plugins/local/edge.py`:
- Around line 109-116: The create_audio_track method currently ignores the
sample_rate and stereo parameters; update it to pass these through to the
LocalOutputAudioTrack (or otherwise apply them to the AudioOutputDevice) so
callers' requested format is honored: modify create_audio_track to construct
LocalOutputAudioTrack(audio_output=self._audio_output, sample_rate=sample_rate,
stereo=stereo) (and if LocalOutputAudioTrack doesn't yet accept those args, add
corresponding parameters/handling there to configure output format or fall back
to self._audio_output), or if the local transport truly cannot support custom
rates/stereo, remove the unused parameters from create_audio_track signature
and/or emit a clear warning; refer to create_audio_track, LocalOutputAudioTrack,
and self._audio_output when making the change.
---
Nitpick comments:
In `@examples/10_local_transport_example/local_transport_example.py`:
- Around line 42-49: The get_weather wrapper in setup_llm currently types its
return as dict[str, Any]; change it to the actual return type of
get_weather_by_location (e.g., WeatherInfo or whatever specific
dataclass/TypedDict/Dict shape it returns) so the type is propagated; update the
async def get_weather signature to return that specific type and import or
define the corresponding type alias used by get_weather_by_location (reference:
setup_llm, get_weather, get_weather_by_location).
In `@plugins/local/tests/test_edge.py`:
- Around line 216-235: The test uses hasattr and a type: ignore; replace
hasattr(transport, "open_demo_for_agent") with a direct attribute
access/assertion on transport.open_demo_for_agent (e.g., assert
callable(transport.open_demo_for_agent)) in test_open_demo_for_agent_exists
(using _make_transport to get transport), and replace the type: ignore on the
_video_display assignment in test_close_cleans_up_video_display by casting the
stub to the proper display type or declaring a typed stub class that matches the
transport's expected VideoDisplay interface (e.g., import/cast to the concrete
VideoDisplay protocol/class before assigning _FakeDisplay instance to
transport._video_display) so the static checker no longer needs suppression
while preserving the same runtime behavior with _FakeAgent and _FakeDisplay.
In `@plugins/local/vision_agents/plugins/local/edge.py`:
- Around line 204-214: Remove the stray lone comment character in the
create_conversation method of LocalEdge: locate the async def
create_conversation(self, call: Any, user: User, instructions: str) -> None
function and delete the single '#' line so the method body cleanly returns None
without an extraneous comment.
- Line 179: The empty stub open_demo currently satisfies the abstract method but
does nothing; replace it with a real implementation that either calls
open_demo_for_agent with the same args/kwargs (delegating behavior) or, if
signatures differ, implement a thin adapter that maps parameters and invokes
open_demo_for_agent; alternatively add a short docstring on open_demo explaining
it's intentionally a no-op and directing callers to open_demo_for_agent. Ensure
you update the function body of open_demo (not just the signature) and reference
open_demo_for_agent for delegation so callers get the expected behavior.
In `@plugins/local/vision_agents/plugins/local/tracks.py`:
- Around line 145-169: Replace the broad Any annotations for
LocalVideoTrack._container and _stream with the concrete PyAV types: use
av.container.InputContainer | None for _container and av.stream.Stream | None
for _stream, and add the corresponding imports from the av package at the top of
the file; update the __init__ attribute declarations in class LocalVideoTrack
accordingly to improve static typing and IDE support while preserving the
existing None initialization logic.
- Around line 67-74: The synchronous start() method creates an asyncio task and
can raise RuntimeError if no event loop is running; change start to an async def
start(self) and update callers (e.g., publish_tracks) to await self.start(), so
asyncio.create_task(self._playback_loop()) is invoked from an active event loop;
ensure _playback_loop and assignment to self._playback_task remain unchanged and
update any callers/tests to await start() accordingly.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 40742ea6-3222-4503-a6db-fc6410dc0c4f
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (23)
.github/workflows/run_tests.ymlagents-core/pyproject.tomlexamples/10_local_transport_example/README.mdexamples/10_local_transport_example/__init__.pyexamples/10_local_transport_example/local_transport_example.pyexamples/10_local_transport_example/pyproject.tomlplugins/deepgram/vision_agents/plugins/deepgram/deepgram_stt.pyplugins/local/README.mdplugins/local/py.typedplugins/local/pyproject.tomlplugins/local/tests/__init__.pyplugins/local/tests/conftest.pyplugins/local/tests/test_devices.pyplugins/local/tests/test_edge.pyplugins/local/tests/test_tracks.pyplugins/local/vision_agents/plugins/local/__init__.pyplugins/local/vision_agents/plugins/local/devices.pyplugins/local/vision_agents/plugins/local/display.pyplugins/local/vision_agents/plugins/local/edge.pyplugins/local/vision_agents/plugins/local/tracks.pyplugins/local/vision_agents/plugins/local/utils.pyplugins/turbopuffer/pyproject.tomlpyproject.toml
✅ Files skipped from review due to trivial changes (7)
- plugins/turbopuffer/pyproject.toml
- agents-core/pyproject.toml
- examples/10_local_transport_example/README.md
- plugins/local/vision_agents/plugins/local/init.py
- plugins/local/pyproject.toml
- examples/10_local_transport_example/pyproject.toml
- examples/10_local_transport_example/init.py
There was a problem hiding this comment.
🧹 Nitpick comments (4)
plugins/local/tests/test_edge.py (2)
218-221: Avoidhasattr; prefer direct attribute access.The coding guidelines discourage
hasattr. Consider calling the method directly or checking it's callable, since the method's existence is enforced by the base class anyway.🔧 Proposed fix
async def test_open_demo_for_agent_exists(self) -> None: transport = _make_transport() - assert hasattr(transport, "open_demo_for_agent") + # Method existence enforced by EdgeTransport base class; verify it's callable + assert callable(transport.open_demo_for_agent)As per coding guidelines: "Avoid
getattr,hasattr,delattr,setattr; prefer normal attribute access."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/local/tests/test_edge.py` around lines 218 - 221, The test test_open_demo_for_agent_exists uses hasattr which is discouraged; instead directly access the attribute on the transport and assert it's callable or can be accessed: replace the hasattr check with a direct assertion like calling or checking callable(transport.open_demo_for_agent) to verify the method exists and is usable (reference: test_open_demo_for_agent_exists, transport, open_demo_for_agent).
232-241: Consider eliminating the# type: ignorecomment.The guidelines discourage
# type: ignore. You could define_FakeDisplaywith a proper protocol or usecastif this is truly intentional.🔧 Alternative using a minimal protocol match
+ from typing import Protocol + + class _DisplayProtocol(Protocol): + async def stop(self) -> None: ... + class _FakeDisplay: async def stop(self) -> None: nonlocal stopped stopped = True - transport._video_display = _FakeDisplay() # type: ignore[assignment] + transport._video_display = _FakeDisplay() # type: ignore[assignment]Alternatively, since this is test code and the assignment is intentional, you may consider this an acceptable pragmatic exception. The test verifies behavior, not types.
As per coding guidelines: "Avoid
# type: ignorecomments."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/local/tests/test_edge.py` around lines 232 - 241, Remove the "# type: ignore[assignment]" by giving _FakeDisplay a proper static type that matches transport._video_display: either declare a minimal Protocol (e.g., with async def stop(self) -> None) and annotate _FakeDisplay accordingly, or keep the simple class and use typing.cast to cast the instance to the expected display type when assigning to transport._video_display; ensure the test still sets stopped and calls await transport.close() and asserts stopped and transport._video_display is None.plugins/local/vision_agents/plugins/local/edge.py (2)
208-212: Remove the empty comment or add meaningful documentation.Line 211 contains a solitary
#with no text—a ghost of an intention, perhaps, but it serves no purpose now.🧹 Proposed fix
async def create_conversation( self, call: Any, user: User, instructions: str ) -> None: - # + # Local transport does not require conversation setup return None🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/local/vision_agents/plugins/local/edge.py` around lines 208 - 212, Remove the stray solitary comment in the create_conversation method: either delete the lone "#" on line inside async def create_conversation(self, call: Any, user: User, instructions: str) -> None so the method contains only the intended return, or replace it with a short meaningful docstring or comment explaining the method's purpose/why it returns None (e.g., describe expected behavior, inputs, and that it's a no-op stub) to keep the code clean and documented.
46-64: Consider validating video parameters.The constructor accepts
video_width,video_height, andvideo_fpswithout validation, yetVideoDisplayvalidates these same parameters (as evidenced by the tests intest_display.py). Invalid values (zero or negative) passed here would create aQueuedVideoTrackwith bad dimensions beforeVideoDisplayever gets a chance to reject them.🛡️ Proposed validation
""" super().__init__() + if video_width <= 0: + raise ValueError("video_width must be > 0") + if video_height <= 0: + raise ValueError("video_height must be > 0") + if video_fps <= 0: + raise ValueError("video_fps must be > 0") + self._audio_input = audio_input self._audio_output = audio_output🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/local/vision_agents/plugins/local/edge.py` around lines 46 - 64, The constructor (__init__) currently accepts video_width, video_height, and video_fps without checks, which can create a QueuedVideoTrack with invalid dimensions; add validation at the top of __init__ to ensure video_width and video_height are positive integers (>0) and video_fps is a positive number (>0), and raise a ValueError with a clear message (e.g., "video_width must be > 0") when any check fails so invalid values are rejected early (this mirrors VideoDisplay's expectations and prevents constructing a bad QueuedVideoTrack).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@plugins/local/tests/test_edge.py`:
- Around line 218-221: The test test_open_demo_for_agent_exists uses hasattr
which is discouraged; instead directly access the attribute on the transport and
assert it's callable or can be accessed: replace the hasattr check with a direct
assertion like calling or checking callable(transport.open_demo_for_agent) to
verify the method exists and is usable (reference:
test_open_demo_for_agent_exists, transport, open_demo_for_agent).
- Around line 232-241: Remove the "# type: ignore[assignment]" by giving
_FakeDisplay a proper static type that matches transport._video_display: either
declare a minimal Protocol (e.g., with async def stop(self) -> None) and
annotate _FakeDisplay accordingly, or keep the simple class and use typing.cast
to cast the instance to the expected display type when assigning to
transport._video_display; ensure the test still sets stopped and calls await
transport.close() and asserts stopped and transport._video_display is None.
In `@plugins/local/vision_agents/plugins/local/edge.py`:
- Around line 208-212: Remove the stray solitary comment in the
create_conversation method: either delete the lone "#" on line inside async def
create_conversation(self, call: Any, user: User, instructions: str) -> None so
the method contains only the intended return, or replace it with a short
meaningful docstring or comment explaining the method's purpose/why it returns
None (e.g., describe expected behavior, inputs, and that it's a no-op stub) to
keep the code clean and documented.
- Around line 46-64: The constructor (__init__) currently accepts video_width,
video_height, and video_fps without checks, which can create a QueuedVideoTrack
with invalid dimensions; add validation at the top of __init__ to ensure
video_width and video_height are positive integers (>0) and video_fps is a
positive number (>0), and raise a ValueError with a clear message (e.g.,
"video_width must be > 0") when any check fails so invalid values are rejected
early (this mirrors VideoDisplay's expectations and prevents constructing a bad
QueuedVideoTrack).
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 8d6bbcc6-3588-40cf-a6e7-b687704113b6
📒 Files selected for processing (7)
plugins/local/tests/conftest.pyplugins/local/tests/test_devices.pyplugins/local/tests/test_display.pyplugins/local/tests/test_edge.pyplugins/local/tests/test_tracks.pyplugins/local/vision_agents/plugins/local/display.pyplugins/local/vision_agents/plugins/local/edge.py
✅ Files skipped from review due to trivial changes (2)
- plugins/local/tests/test_devices.py
- plugins/local/tests/test_tracks.py
🚧 Files skipped from review as they are similar to previous changes (2)
- plugins/local/tests/conftest.py
- plugins/local/vision_agents/plugins/local/display.py
Having no conversation breaks many things
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
plugins/local/tests/test_edge.py (2)
181-182: Please consolidateLocalEdgetests into a single test class
TestLocalEdgeVideois still unit-testingLocalEdge, but those tests are split fromTestLocalEdge. Please merge them into one class for consistency and discoverability. As per coding guidelines, "Keep unit-tests for the class under the same test class. Do not spread them around different test classes."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/local/tests/test_edge.py` around lines 181 - 182, Merge the tests currently in class TestLocalEdgeVideo into the existing TestLocalEdge test class: move all test methods and the docstring from TestLocalEdgeVideo into TestLocalEdge (preserving method names and any fixtures or setup/teardown methods they rely on), remove the empty/duplicate TestLocalEdgeVideo class, and run tests to ensure no name collisions or missing imports; ensure the combined TestLocalEdge contains the consolidated docstring "Tests for LocalEdge video functionality." where appropriate.
240-240: Use a Protocol instead of# type: ignore[assignment]The test stub only requires a
stop()method, but assigning_FakeDisplay()toVideoDisplay | Nonetriggers a type mismatch. Define a Protocol with just thestopmethod that_FakeDisplayimplements, eliminating the need for suppression:Example approach
from typing import Protocol class _VideoDisplayStub(Protocol): async def stop(self) -> None: ... transport._video_display: _VideoDisplayStub = _FakeDisplay()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@plugins/local/tests/test_edge.py` at line 240, The assignment to transport._video_display currently uses a type ignore because _FakeDisplay doesn't match VideoDisplay | None; define a lightweight Protocol (e.g., _VideoDisplayStub) that declares the async stop(self) -> None method and annotate transport._video_display with that Protocol type before assigning _FakeDisplay(), then ensure _FakeDisplay implements stop(); this removes the "# type: ignore[assignment]" and keeps the test stub type-safe while referencing transport._video_display, _FakeDisplay, and the stop() method.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@plugins/local/tests/test_edge.py`:
- Around line 221-224: The test test_open_demo_for_agent_exists uses hasattr
which is discouraged; update the assertion to access the attribute directly by
calling or checking it on the transport returned by _make_transport, for example
assert callable(transport.open_demo_for_agent) or assert
transport.open_demo_for_agent to ensure the attribute exists and is callable
rather than using hasattr.
---
Nitpick comments:
In `@plugins/local/tests/test_edge.py`:
- Around line 181-182: Merge the tests currently in class TestLocalEdgeVideo
into the existing TestLocalEdge test class: move all test methods and the
docstring from TestLocalEdgeVideo into TestLocalEdge (preserving method names
and any fixtures or setup/teardown methods they rely on), remove the
empty/duplicate TestLocalEdgeVideo class, and run tests to ensure no name
collisions or missing imports; ensure the combined TestLocalEdge contains the
consolidated docstring "Tests for LocalEdge video functionality." where
appropriate.
- Line 240: The assignment to transport._video_display currently uses a type
ignore because _FakeDisplay doesn't match VideoDisplay | None; define a
lightweight Protocol (e.g., _VideoDisplayStub) that declares the async
stop(self) -> None method and annotate transport._video_display with that
Protocol type before assigning _FakeDisplay(), then ensure _FakeDisplay
implements stop(); this removes the "# type: ignore[assignment]" and keeps the
test stub type-safe while referencing transport._video_display, _FakeDisplay,
and the stop() method.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 096f6464-0960-486a-95d0-aa0bd94a50e6
📒 Files selected for processing (6)
agents-core/pyproject.tomlexamples/10_local_transport_example/README.mdexamples/10_local_transport_example/local_transport_example.pyexamples/10_local_transport_example/pyproject.tomlplugins/local/tests/test_edge.pyplugins/local/vision_agents/plugins/local/edge.py
✅ Files skipped from review due to trivial changes (4)
- agents-core/pyproject.toml
- examples/10_local_transport_example/pyproject.toml
- examples/10_local_transport_example/README.md
- examples/10_local_transport_example/local_transport_example.py
🚧 Files skipped from review as they are similar to previous changes (1)
- plugins/local/vision_agents/plugins/local/edge.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Work-in-progress branch for adding support for local devices.
cc: @dangusev
Summary by CodeRabbit
New Features
Tests
Chores