fix(core): move callbacks to the caller - #5039
Conversation
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| for tool in tools: | ||
| info: RawFunctionToolInfo | FunctionToolInfo | None = None | ||
| if isinstance(tool, (llm.RawFunctionTool, llm.FunctionTool)): | ||
| if isinstance(tool, llm.RawFunctionTool | llm.FunctionTool): |
There was a problem hiding this comment.
🟡 isinstance(x, A | B) union syntax breaks Python 3.9 compatibility
AGENTS.md states "Python 3.9+ compatibility required" under Code Style. The change from isinstance(tool, (llm.RawFunctionTool, llm.FunctionTool)) to isinstance(tool, llm.RawFunctionTool | llm.FunctionTool) uses the PEP 604 union type syntax in a runtime expression, which requires Python 3.10+. On Python 3.9, type.__or__ is not defined, so llm.RawFunctionTool | llm.FunctionTool raises TypeError: unsupported operand type(s) for |: 'type' and 'type'. The old code using a tuple worked on all Python versions. Note: this applies only if the project still genuinely targets 3.9; if pyproject.toml has since raised the minimum, this is moot.
| if isinstance(tool, llm.RawFunctionTool | llm.FunctionTool): | |
| if isinstance(tool, (llm.RawFunctionTool, llm.FunctionTool)): |
Was this helpful? React with 👍 or 👎 to provide feedback.
| @@ -506,7 +509,7 @@ def _tool_completed(out: ToolExecutionOutput) -> None: | |||
| ) | |||
| continue | |||
|
|
|||
There was a problem hiding this comment.
🟡 isinstance(x, A | B) union syntax breaks Python 3.9 compatibility (second instance)
Same issue as BUG-0001 but in generation.py. isinstance(function_tool, llm.FunctionTool | llm.RawFunctionTool) uses Python 3.10+ union syntax in a runtime isinstance call, violating the "Python 3.9+ compatibility required" rule in AGENTS.md. The old code used isinstance(function_tool, (llm.FunctionTool, llm.RawFunctionTool)).
Was this helpful? React with 👍 or 👎 to provide feedback.
longcw
left a comment
longcw
left a comment
There was a problem hiding this comment.
the fix looks good to me, just not sure about the resolve and cancel, maybe we can call the audio_out.first_frame_fut.cancel() directly, and resolve is also not needed to be exposed?
| out = _AudioOutput(audio=[], first_frame_fut=asyncio.Future()) | ||
| # out.first_frame_fut should be cancelled in the caller after the playout is finished or interrupted | ||
| audio_output.on("playback_started", out._resolve_first_frame_fut) | ||
| out.first_frame_fut.add_done_callback( | ||
| lambda _: audio_output.off("playback_started", out._resolve_first_frame_fut) | ||
| ) |
There was a problem hiding this comment.
🟡 Event listener cleanup on audio_output is no longer guaranteed, causing potential stale listener leak and spurious state updates
Previously, the playback_started listener and first_frame_fut.cancel() were in _audio_forwarding_task's finally block (generation.py:408-411 in old code), guaranteeing cleanup regardless of how the task ended (including cancellation). The new code removes this finally cleanup and instead relies on callers to cancel first_frame_fut at the end of their methods (agent_activity.py:1877, agent_activity.py:2189, agent_activity.py:2680). However, none of these cancel calls are in try/finally blocks. If the caller's coroutine is cancelled (e.g. via speech interruption calling cancel_and_wait on the speech task) or an unexpected exception occurs after perform_audio_forwarding but before the cancel line, first_frame_fut stays pending and the playback_started listener remains registered on the shared audio_output object. When a subsequent speech starts playing and fires playback_started, the stale listener resolves the old future, triggering the _on_first_frame callback which calls self._session._update_agent_state("speaking", ...) with the stale speech handle's context.
Prompt for agents
The first_frame_fut.cancel() calls added to the three callers in agent_activity.py (_tts_task_impl at line 1877, _pipeline_reply_task_impl at line 2189, _realtime_reply_task_impl at line 2680) are not protected by try/finally blocks. If the caller task is cancelled or an exception occurs before reaching these lines, the playback_started listener registered at generation.py:368 will leak on the shared audio_output object.
To fix this, each caller should wrap the section between perform_audio_forwarding and the cancel call in a try/finally. For example, in _tts_task_impl (agent_activity.py), the code from the perform_audio_forwarding call (around line 1793) through to line 1878 should have the cancel in a finally block. Similarly for _pipeline_reply_task_impl (around lines 2081-2190) and _realtime_reply_task_impl (the section in _read_messages where perform_audio_forwarding is called at line 2517 through to line 2681).
Alternatively, you could keep the cleanup in _audio_forwarding_task's finally block as a safety net in addition to the caller-side cancel, ensuring both approaches co-exist for robustness.
Was this helpful? React with 👍 or 👎 to provide feedback.
…#1909) Use the bare livekit#1909 short form referenced once at the core fix points, matching existing comments (e.g. livekit#1662, livekit#1430, livekit#1124), instead of the verbose cross-repo 'livekit#1909 (port of livekit/agents#5039)' form and the per-call-site repetition. The port context lives in the commit/PR/changeset.
2 participants
Previously
first_frame_futis resolved by callbacks, but the callback might be unregistered when forwarding finishes before the playout starts.This makes sure the callback stays alive until the speech is finished or interrupted and is self-cleaned with a
done_callback. This closes #5038