livekit · longcw · Jul 29, 2026 · u9g · Jul 29, 2026 · u9g
@@ -3983,6 +3983,27 @@ def _create_assistant_message(
 
         if speech_handle.interrupted:
             await utils.aio.cancel_and_wait(exe_task)
+
+            # commit results of tools that finished despite the interruption, similar to the pipeline task
+            interrupted_fnc_outputs = [
+                sanitized_out.fnc_call_out
+                for sanitized_out in tool_output.output
+                if sanitized_out.fnc_call_out is not None and sanitized_out.agent_task is None
+            ]
+            if interrupted_fnc_outputs:
+                self._agent._chat_ctx.insert(interrupted_fnc_outputs)
+                self._session._tool_items_added(interrupted_fnc_outputs)
+
+                # unlike the pipeline, a realtime model holds the call open server-side
+                chat_ctx = self._rt_session.chat_ctx.copy()
+                chat_ctx.items.extend(interrupted_fnc_outputs)
+                try:
+                    await self._rt_session.update_chat_ctx(chat_ctx)
+                except llm.RealtimeError as e:
+                    logger.warning(
+                        "failed to sync the tool results of an interrupted generation",
+                        extra={"error": str(e)},
+                    )
             return
 
         # wait for the tool execution to complete

@@ -90,6 +90,13 @@ def _validate_model_api_match(model: str, use_vertexai: bool) -> None:
         )
 
 
+def _warn_vertex_scheduling_unsupported() -> None:
+    logger.warning(
+        "tool_response_scheduling is not supported by Vertex AI and will be ignored; "
+        "tool responses use the default scheduling there."
+    )
+
+
 def _get_1008_error_hint(error_message: str) -> str | None:
     """
     Generate a hint for WebSocket 1008 policy violation errors.
@@ -296,6 +303,8 @@ def __init__(
             if is_given(vertexai)
             else os.environ.get("GOOGLE_GENAI_USE_VERTEXAI", "0").lower() in ["true", "1"]
         )
+        if use_vertexai and is_given(tool_response_scheduling):
+            _warn_vertex_scheduling_unsupported()
         if not is_given(model):
             model = (
                 "gemini-live-2.5-flash-native-audio"
@@ -506,6 +515,8 @@ def __init__(self, realtime_model: RealtimeModel) -> None:
         # means we're draining that turn's trailing events (which have no generation to attach
         # to). reset when the next generation starts.
         self._rejected_tool_calls = 0
+        # whether playout of the current turn was cut short
+        self._playout_interrupted = False
 
         self._session_resumption_handle: str | None = (
             self._opts.session_resumption.handle
@@ -572,6 +583,8 @@ def update_options(
             and self._opts.tool_response_scheduling != tool_response_scheduling
         ):
             self._opts.tool_response_scheduling = tool_response_scheduling
+            if self._opts.vertexai:
+                _warn_vertex_scheduling_unsupported()
             # no need to restart
 
         if is_given(tool_choice):
@@ -657,10 +670,20 @@ async def update_chat_ctx(self, chat_ctx: llm.ChatContext) -> None:
                 append_ctx.items.append(item)
 
         if append_ctx.items:
+            scheduling = self._opts.tool_response_scheduling
+            if (
+                not self._opts.vertexai  # scheduling is not supported by vertex
+                and self._playout_interrupted
+                # only honoured on NON_BLOCKING declarations
+                and self._opts.tool_behavior == types.Behavior.NON_BLOCKING
+            ):
+                # the turn is over, so record the result without prompting more speech
+                scheduling = types.FunctionResponseScheduling.SILENT
+
             tool_results = get_tool_results_for_realtime(
                 append_ctx,
                 vertexai=self._opts.vertexai,
-                tool_response_scheduling=self._opts.tool_response_scheduling,
+                tool_response_scheduling=scheduling,
             )
             if self._realtime_model.capabilities.mutable_chat_context:
                 turns_dict, _ = append_ctx.copy(exclude_function_call=True).to_provider_format(
@@ -817,6 +840,9 @@ def start_user_activity(self) -> None:
             )
 
     def interrupt(self) -> None:
+        # recorded locally since this cannot reach Gemini under automatic activity detection
+        self._playout_interrupted = True
+
         # Gemini Live treats activity start as interruption, so we rely on start_user_activity
         # notifications to handle it
         if (
@@ -1201,6 +1227,7 @@ def _build_connect_config(self) -> types.LiveConnectConfig:
 
     def _start_new_generation(self) -> None:
         self._rejected_tool_calls = 0
+        self._playout_interrupted = False
         if self._current_generation and not self._current_generation._done:
             logger.warning("starting new generation while another is active. Finalizing previous.")
             self._mark_current_generation_done()

@@ -69,13 +69,12 @@ def create_function_response(
         name=output.name,
         response={"error": output.output} if output.is_error else {"output": output.output},
     )
-    if is_given(tool_response_scheduling):
-        # vertexai currently doesn't support the scheduling parameter, gemini api defaults to idle
-        # it's the user's responsibility to avoid this parameter when using vertexai
-        res.scheduling = tool_response_scheduling
     if not vertexai:
-        # vertexai does not support id in FunctionResponse
+        # vertexai supports neither scheduling nor id in FunctionResponse; the gemini api
+        # defaults scheduling to WHEN_IDLE
         # see: https://github.com/googleapis/python-genai/blob/85e00bc/google/genai/_live_converters.py#L1435
+        if is_given(tool_response_scheduling):
+            res.scheduling = tool_response_scheduling
         res.id = output.call_id
     return res
 

@@ -5,8 +5,10 @@
 import pytest
 from google.genai import types
 
-from livekit.agents import utils
+from livekit.agents import llm, utils
+from livekit.plugins.google.realtime.api_proto import ClientEvents
 from livekit.plugins.google.realtime.realtime_api import RealtimeModel, RealtimeSession
+from livekit.plugins.google.utils import create_function_response
 
 pytestmark = pytest.mark.unit
 
@@ -95,3 +97,182 @@ async def test_late_content_after_generation_complete_is_dropped(
     assert gen.output_text == ""
     assert not gen._done
     assert any("after generation completed" in r.message for r in caplog.records)
+
+
+def _tool_call(call_id: str = "fc_1", name: str = "lookup") -> types.LiveServerToolCall:
+    return types.LiveServerToolCall(
+        function_calls=[types.FunctionCall(id=call_id, name=name, args={})]
+    )
+
+
+def _tool_output(call_id: str = "fc_1", name: str = "lookup") -> llm.FunctionCallOutput:
+    return llm.FunctionCallOutput(call_id=call_id, name=name, output="42", is_error=False)
+
+
+async def _make_connected_session(
+    monkeypatch: pytest.MonkeyPatch, *, non_blocking_tools: bool = False
+) -> RealtimeSession:
+    """A session that believes it is connected, so update_chat_ctx actually emits.
+
+    The placeholder is never called: the send task is not running, so client events just
+    queue up in `_msg_ch` for the test to inspect. `_make_session` closes that channel to
+    stop the connect loop, so it is replaced with an open one first.
+    """
+    session = await _make_session(monkeypatch)
+    if non_blocking_tools:
+        session._opts.tool_behavior = types.Behavior.NON_BLOCKING
+    session._msg_ch = utils.aio.Chan[ClientEvents]()
+    session._active_session = object()  # type: ignore[assignment]
+    return session
+
+
+async def _drain_sent(session: RealtimeSession) -> list[object]:
+    sent: list[object] = []
+    while not session._msg_ch.empty():
+        sent.append(session._msg_ch.recv_nowait())
+    return sent
+
+
+async def test_tool_response_sent_after_client_interruption(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A tool result owed by an interrupted turn still reaches Gemini, silently.
+
+    Gemini blocks the turn until every tool call is answered and offers no cancel, so dropping
+    the result strands the session: it stops responding and `generate_reply` never gets a
+    generation (issue #6569). SILENT records the result without prompting speech the user never
+    asked for.
+    """
+    session = await _make_connected_session(monkeypatch, non_blocking_tools=True)
+    session._start_new_generation()
+    session._handle_tool_calls(_tool_call())
+
+    # the user was interrupted locally; Gemini was never told
+    session.interrupt()
+    await _drain_sent(session)
+
+    chat_ctx = session.chat_ctx.copy()
+    chat_ctx.items.append(_tool_output())
+    await session.update_chat_ctx(chat_ctx)
+
+    sent = await _drain_sent(session)
+    responses = [m for m in sent if isinstance(m, types.LiveClientToolResponse)]
+    assert len(responses) == 1, f"expected the tool response to be sent, got {sent}"
+    assert responses[0].function_responses is not None
+    assert responses[0].function_responses[0].id == "fc_1"
+    assert responses[0].function_responses[0].scheduling == types.FunctionResponseScheduling.SILENT
+
+
+async def test_tool_response_sent_after_server_interruption(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Gemini drops the call when it interrupts the turn itself, but the result is still sent.
+
+    It costs nothing — the API can use it on the next turn — and withholding it would leave the
+    local context claiming a result the server never saw.
+    """
+    session = await _make_connected_session(monkeypatch)
+    session._start_new_generation()
+    session._handle_tool_calls(_tool_call())
+    session._handle_server_content(types.LiveServerContent(interrupted=True))
+    await _drain_sent(session)
+
+    chat_ctx = session.chat_ctx.copy()
+    chat_ctx.items.append(_tool_output())
+    await session.update_chat_ctx(chat_ctx)
+
+    responses = [
+        m for m in await _drain_sent(session) if isinstance(m, types.LiveClientToolResponse)
+    ]
+    assert len(responses) == 1
+
+
+async def test_tool_response_uses_configured_scheduling_when_not_interrupted(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The normal path keeps the configured scheduling; SILENT is only for interrupted turns."""
+    session = await _make_connected_session(monkeypatch)
+    session._start_new_generation()
+    session._handle_tool_calls(_tool_call())
+    await _drain_sent(session)
+
+    chat_ctx = session.chat_ctx.copy()
+    chat_ctx.items.append(_tool_output())
+    await session.update_chat_ctx(chat_ctx)
+
+    responses = [
+        m for m in await _drain_sent(session) if isinstance(m, types.LiveClientToolResponse)
+    ]
+    assert len(responses) == 1
+    assert responses[0].function_responses is not None
+    assert responses[0].function_responses[0].scheduling is None
+
+
+async def test_interrupted_tool_response_keeps_default_scheduling_for_blocking_tools(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Gemini ignores scheduling on BLOCKING declarations, so none is claimed.
+
+    The response is still sent — unblocking the turn matters more than the reply it prompts.
+    """
+    session = await _make_connected_session(monkeypatch)
+    session._start_new_generation()
+    session._handle_tool_calls(_tool_call())
+    session.interrupt()
+    await _drain_sent(session)
+
+    chat_ctx = session.chat_ctx.copy()
+    chat_ctx.items.append(_tool_output())
+    await session.update_chat_ctx(chat_ctx)
+
+    responses = [
+        m for m in await _drain_sent(session) if isinstance(m, types.LiveClientToolResponse)
+    ]
+    assert len(responses) == 1
+    assert responses[0].function_responses is not None
+    assert responses[0].function_responses[0].scheduling is None
+
+
+@pytest.mark.parametrize("vertexai", [False, True])
+def test_function_response_scheduling_only_for_gemini_api(vertexai: bool) -> None:
+    """Vertex AI rejects `scheduling` (and `id`), so neither is set for it."""
+    res = create_function_response(
+        _tool_output(),
+        vertexai=vertexai,
+        tool_response_scheduling=types.FunctionResponseScheduling.SILENT,
+    )
+
+    if vertexai:
+        assert res.scheduling is None
+        assert res.id is None
+    else:
+        assert res.scheduling == types.FunctionResponseScheduling.SILENT
+        assert res.id == "fc_1"
+
+
+def test_vertex_scheduling_warns(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An explicitly set scheduling is dropped on Vertex AI, so say so instead of ignoring it."""
+    monkeypatch.setenv("GOOGLE_API_KEY", "fake-key")
+
+    with caplog.at_level(logging.WARNING):
+        RealtimeModel(
+            vertexai=True,
+            project="p",
+            location="us-central1",
+            tool_response_scheduling=types.FunctionResponseScheduling.SILENT,
+        )
+
+    assert any("tool_response_scheduling is not supported" in r.message for r in caplog.records)
+
+
+def test_gemini_api_scheduling_does_not_warn(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    monkeypatch.setenv("GOOGLE_API_KEY", "fake-key")
+
+    with caplog.at_level(logging.WARNING):
+        RealtimeModel(tool_response_scheduling=types.FunctionResponseScheduling.SILENT)
+
+    assert not any("tool_response_scheduling is not supported" in r.message for r in caplog.records)