Skip to content

fix(core): move callbacks to the caller #5039

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chenghao-mou merged 2 commits into from
Mar 9, 2026
Merged

fix(core): move callbacks to the caller #5039

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
41f9a3c
remove callbacks to the caller
chenghao-mou Mar 8, 2026
1247b1c
remove abstraction and rename private function
chenghao-mou Mar 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion livekit-agents/livekit/agents/voice/agent_activity.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -906,7 +906,7 @@ def _generate_reply(
filtered_tools: list[llm.Tool | llm.Toolset] = []
for tool in tools:
info: RawFunctionToolInfo | FunctionToolInfo | None = None
if isinstance(tool, (llm.RawFunctionTool, llm.FunctionTool)):
if isinstance(tool, llm.RawFunctionTool | llm.FunctionTool):

@devin-ai-integration devin-ai-integration Bot Mar 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 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.

Suggested change
if isinstance(tool, llm.RawFunctionTool | llm.FunctionTool):
if isinstance(tool, (llm.RawFunctionTool, llm.FunctionTool)):
Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

info = tool.info

if info and (info.flags & ToolFlag.IGNORE_ON_ENTER):
Expand Down Expand Up @@ -1874,6 +1874,9 @@ def _on_first_frame(fut: asyncio.Future[float] | asyncio.Future[None]) -> None:
if self._session.agent_state == "speaking":
self._session._update_agent_state("listening")

if audio_out is not None and not audio_out.first_frame_fut.done():
audio_out.first_frame_fut.cancel()

@utils.log_exceptions(logger=logger)
async def _pipeline_reply_task(
self,
Expand Down Expand Up @@ -2183,6 +2186,9 @@ def _tool_execution_completed_cb(out: ToolExecutionOutput) -> None:
elif self._session.agent_state == "speaking":
self._session._update_agent_state("listening")

if audio_out is not None and not audio_out.first_frame_fut.done():
audio_out.first_frame_fut.cancel()

await text_tee.aclose()

speech_handle._mark_generation_done() # mark the playout done before waiting for the tool execution # noqa: E501
Expand Down Expand Up @@ -2671,6 +2677,9 @@ def _create_assistant_message(
self._session._conversation_item_added(msg)
current_span.set_attribute(trace_types.ATTR_RESPONSE_TEXT, forwarded_text)

if audio_out is not None and not audio_out.first_frame_fut.done():
audio_out.first_frame_fut.cancel()

for tee in tees:
await tee.aclose()
speech_handle._mark_generation_done()
Expand Down
21 changes: 10 additions & 11 deletions livekit-agents/livekit/agents/voice/generation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -353,13 +353,22 @@ class _AudioOutput:
first_frame_fut: asyncio.Future[float]
"""Future that will be set with the timestamp of the first frame's capture"""

def _resolve_first_frame_fut(self, ev: io.PlaybackStartedEvent) -> None:
if not self.first_frame_fut.done():
self.first_frame_fut.set_result(ev.created_at)


def perform_audio_forwarding(
*,
audio_output: io.AudioOutput,
tts_output: AsyncIterable[rtc.AudioFrame],
) -> tuple[asyncio.Task[None], _AudioOutput]:
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)
)
Comment on lines 366 to +371

@devin-ai-integration devin-ai-integration Bot Mar 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 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.
Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

task = asyncio.create_task(_audio_forwarding_task(audio_output, tts_output, out))
return task, out

Expand All @@ -372,12 +381,7 @@ async def _audio_forwarding_task(
) -> None:
resampler: rtc.AudioResampler | None = None

def _on_playback_started(ev: io.PlaybackStartedEvent) -> None:
if not out.first_frame_fut.done():
out.first_frame_fut.set_result(ev.created_at)

try:
audio_output.on("playback_started", _on_playback_started)
audio_output.resume()

async for frame in tts_output:
Expand Down Expand Up @@ -406,11 +410,6 @@ def _on_playback_started(ev: io.PlaybackStartedEvent) -> None:
await audio_output.capture_frame(frame)

finally:
audio_output.off("playback_started", _on_playback_started)

if not out.first_frame_fut.done():
out.first_frame_fut.cancel()

if isinstance(tts_output, _ACloseable):
try:
await tts_output.aclose()
Expand Down Expand Up @@ -506,7 +505,7 @@ def _tool_completed(out: ToolExecutionOutput) -> None:
)
continue

@devin-ai-integration devin-ai-integration Bot Mar 8, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 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)).

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

if not isinstance(function_tool, (llm.FunctionTool, llm.RawFunctionTool)):
if not isinstance(function_tool, llm.FunctionTool | llm.RawFunctionTool):
logger.error(
f"unknown tool type: {type(function_tool)}",
extra={
Expand Down
Loading
Loading