Start GPT-Live replies where the model takes the turn, and settle results its backend gave up on

docs/realtime/history.md

@@ -208,13 +208,13 @@ Input transcription defaults to `'auto'`; see [Input transcription](audio.md#inp
 and each provider page for configuration. Transcripts are recorded with the user turn they describe,
 even when they arrive after that turn's response or overlap the following turn.
 
-On OpenAI, Azure OpenAI, xAI, and Gemini Live, history follows the order of the provider's own conversation:
+History follows the order of the provider's own conversation:
 
 - A spoken turn sits where the provider added it, and text, images, and tool results where they reached
   it. A turn the user started while the model was still answering, but which the provider only committed
-  after that answer ended, is recorded after the answer. Gemini Live doesn't say where an input joined its
-  conversation, nor when a reply starts: a spoken turn is recorded ahead of the reply to it, and something
-  sent while a reply is owed (requested, or already being spoken) is recorded after that reply.
+  after that answer ended, is recorded after the answer. Gemini Live and GPT-Live don't say where an input
+  joined their conversation, nor when a reply starts: a spoken turn is recorded ahead of the reply to it, and
+  something sent while a reply is owed (requested, or already being spoken) is recorded after that reply.
 - A message appears in `all_messages()` once everything before it is final. A reply waits for the
   transcript of the spoken turn before it, which can arrive after the reply itself is done, for up to 30
   seconds after that; past that, the turn is recorded with the transcript it has so far.
@@ -226,13 +226,6 @@ On OpenAI, Azure OpenAI, xAI, and Gemini Live, history follows the order of the
   call, or `'final_answer'`) as `'phase'` in the part's `provider_details`, the way a standard OpenAI run
   records [text phases](../models/openai.md#text-phases).
 
-On GPT-Live, a turn the user starts while the model is still answering, whether they
-[barge in](turns.md#barge-in) or push to talk over it, is recorded after that answer. Such a turn joins
-history once the provider ends the answer it cut off, or after a few seconds if the provider never does.
-In that fallback the turn is recorded where history stands, so it lands before the answer it interrupted,
-and ahead of anything sent with [`send()`][pydantic_ai.realtime.RealtimeSession.send] while that answer was
-still in flight.
-
 If a reported speech segment never receives a transcript, the session still records its retained audio or
 a content-less `SpeechPart`, at the latest when the session closes.
 

docs/realtime/openai.md

@@ -229,7 +229,7 @@ history nor waited for by [`wait_for_reply()`][pydantic_ai.realtime.RealtimeSess
 split it, or give long material to the backend as a tool result instead.
 [`enqueue()`](tools.md#enqueuing-prompts) delivers text the same way once the model is idle. The
 [`EnqueuedMessagesEvent`][pydantic_ai.messages.EnqueuedMessagesEvent] it produces marks when the text
-was sent and recorded in history, not when the model took it in. Since Live can take text as context
+was sent, not when the model took it in. Since Live can take text as context
 without answering it, bound [`wait_for_reply()`][pydantic_ai.realtime.RealtimeSession.wait_for_reply]
 with a timeout rather than relying on a reply always following.
 

pydantic_ai_slim/pydantic_ai/realtime/_inferred_lifecycle.py

@@ -67,7 +67,9 @@ class _Turn:
 class InferredLifecycle:
     """Turns a connection's sends and messages into lifecycle events, for a protocol that identifies nothing."""
 
-    def __init__(self, *, transcribes: bool, transcripts_lag_replies: bool = False) -> None:
+    def __init__(
+        self, *, transcribes: bool, transcripts_lag_replies: bool = False, audio_starts_response: bool = True
+    ) -> None:
         """Track a connection's lifecycle.
 
         Args:
@@ -75,7 +77,11 @@ class InferredLifecycle:
             transcripts_lag_replies: Whether the provider can transcribe the user's words after the model started
                 answering them (Gemini does): a transcript that starts while a reply nobody asked for is under way
                 is then the speech that reply answers, rather than speech over it.
+            audio_starts_response: Whether output audio is the model speaking. GPT-Live's output is a continuous
+                track, silent between replies: its connection says where the model takes the turn instead
+                (`message(takes_turn=True)`), and audio outside a reply is no response.
         """
+        self._audio_starts_response = audio_starts_response
         self._transcribes = transcribes
         self._transcripts_lag_replies = transcripts_lag_replies
         self._pending: list[LifecycleEvent] = []
@@ -126,6 +132,12 @@ class InferredLifecycle:
         self._held = [held for held in self._held if held != input_id]
         self._pending = [event for event in self._pending if event != InputAdded(input_id=input_id)]
 
+    def input_unanswerable(self, input_id: InputId) -> None:
+        """An input went (or was meant to go) out, but the provider will never reply to it."""
+        if input_id in self._unanswered:
+            self._unanswered.remove(input_id)
+            self._pending.append(InputLost(input_ids=(input_id,)))
+
     def take_pending(self) -> list[LifecycleEvent]:
         pending, self._pending = self._pending, []
         return pending
@@ -157,8 +169,10 @@ class InferredLifecycle:
     def _event(self, event: RealtimeCodecEvent) -> tuple[list[LifecycleEvent], list[LifecycleEvent]]:
         before: list[LifecycleEvent] = []
         after: list[LifecycleEvent] = []
-        output = isinstance(event, (AudioDelta, PartStartEvent, ToolCall)) or (
-            isinstance(event, OutputTranscript) and bool(event.text)
+        output = (
+            isinstance(event, (PartStartEvent, ToolCall))
+            or (isinstance(event, AudioDelta) and self._audio_starts_response)
+            or (isinstance(event, OutputTranscript) and bool(event.text))
         )
         if self._deferred is not None:
             if isinstance(event, (ToolCall, ResponseDone)):
@@ -179,7 +193,7 @@ class InferredLifecycle:
                 self._calls_awaiting_usage = True
         elif isinstance(event, SessionUsage):
             # Usage reported between responses doesn't start one: it is the next response's (see `_response_done`).
-            if self._calls_awaiting_usage:
+            if event.response_scoped and self._calls_awaiting_usage:
                 # A tool-call frame's report: the calls' response is over, and the answer comes once every result
                 # is in.
                 self._end(after, status='completed', finish_reason='tool_call')

pydantic_ai_slim/pydantic_ai/realtime/openai_live.py

@@ -690,10 +690,10 @@ class OpenAILiveConnection(RealtimeConnection):
         self._pending_reasoning: dict[str | None, list[ResponseReasoningItem]] = {}
         # Which response, user turn, and input each frame is about (see `_tagged_frames()`). Live always
         # transcribes, and closes the user's turn as the model's reply starts.
-        self._lifecycle = InferredLifecycle(transcribes=True)
+        self._lifecycle = InferredLifecycle(transcribes=True, audio_starts_response=False)
         # Every `send()` call is numbered (see `InputRejected.input_index`).
         self._inputs_sent = 0
-        # Whether the frame just mapped had the model take the turn without saying anything (delegating work).
+        # Whether the frame just mapped had the model take the turn (see `_open_response()`).
         self._took_turn = False
 
     @property
@@ -734,7 +734,12 @@ class OpenAILiveConnection(RealtimeConnection):
         self._inputs_sent += 1
         self._lifecycle.input_sent(input_index, content)
         try:
-            await self._send_input(content)
+            if isinstance(content, ToolResult):
+                if not await self._send_tool_result(content):
+                    # Its backend gave up on the call: nothing will reply to the result.
+                    self._lifecycle.input_unanswerable(input_index)
+            else:
+                await self._send_input(content)
         except BaseException:
             self._lifecycle.input_failed(input_index)
             raise
@@ -752,9 +757,6 @@ class OpenAILiveConnection(RealtimeConnection):
             await _check_context_length(content.text)
             await self._send_event({'type': 'session.thinking.append', 'delegation_id': None, 'content': content.text})
             return
-        if isinstance(content, ToolResult):
-            await self._send_tool_result(content)
-            return
         if isinstance(content, CreateResponse):
             # Reaches the connection only right after an image: the profile's
             # `image_input_requires_response` is what lets the session send one, and `create_response()`
@@ -839,8 +841,8 @@ class OpenAILiveConnection(RealtimeConnection):
     def _idle_frame_due(self) -> float:
         return max(self._next_idle_frame, self._input_audio_end + _IDLE_AUDIO_GAP)
 
-    async def _send_tool_result(self, result: ToolResult) -> None:
-        """Return a tool result to the delegated Responses backend and let it continue."""
+    async def _send_tool_result(self, result: ToolResult) -> bool:
+        """Return a tool result to the delegated Responses backend and let it continue; `False` if it goes nowhere."""
         # Mapped first: downloading media can wait, and the backend can give up on the call meanwhile,
         # which only the checks below, made after it, can see.
         follow_up = await _tool_result_follow_up(result, provider_name=self._provider_name)
@@ -850,7 +852,7 @@ class OpenAILiveConnection(RealtimeConnection):
             # The backend that asked for this call gave up before it was answered. Sending the output
             # would attach it to nothing, and `response.create` would start a turn nobody asked for.
             self._abandoned_calls.discard(result.tool_call_id)
-            return
+            return False
         delegation = self._delegations.get(delegation_id) if delegation_id is not None else None
         if delegation is not None:
             delegation.pending_tool_calls.discard(result.tool_call_id)
@@ -865,15 +867,15 @@ class OpenAILiveConnection(RealtimeConnection):
         if delegation is None:
             # A call we can't correlate to a delegation has no response we can wait on, so continue now.
             await self._send_event({'type': 'response.create'})
-            return
-        if delegation.pending_tool_calls:
+        elif delegation.pending_tool_calls:
             # With `parallel_tool_calls` the backend resumes from all of its calls' outputs together.
-            return
-        if delegation.response_in_flight:
+            pass
+        elif delegation.response_in_flight:
             # The response that asked for these calls may still ask for more: continue at its terminal.
             delegation.continuation_due = True
-            return
-        await self._continue(delegation)
+        else:
+            await self._continue(delegation)
+        return True
 
     async def _send_due_continuations(self) -> None:
         """Continue each delegation whose asking response just ended with every result already in."""
@@ -1164,6 +1166,9 @@ class OpenAILiveConnection(RealtimeConnection):
         self._heard_voice()
         events = self._close_input_turn()
         self._response_open = True
+        # Its reply starts here, even with nothing of it to report: work it delegated, or speech a WebRTC
+        # sideband doesn't forward.
+        self._took_turn = True
         return events
 
     def _map_frame(self, raw: str | bytes) -> list[RealtimeCodecEvent]:
@@ -1381,8 +1386,6 @@ class OpenAILiveConnection(RealtimeConnection):
                 )
             ]
         self._delegations[delegation.id] = _Delegation(id=delegation.id)
-        # The model takes the turn to delegate, before it says anything.
-        self._took_turn = True
         return self._open_response()
 
     def _map_response_event(self, nested: dict[str, Any], *, delegation_id: str | None) -> list[RealtimeCodecEvent]:

tests/realtime/test_session_core.py

@@ -49,7 +49,7 @@ from pydantic_ai.realtime._core import (
     ToolReturned,
     TranscriptOverdue,
 )
-from pydantic_ai.realtime._google_lifecycle import GeminiLifecycle
+from pydantic_ai.realtime._inferred_lifecycle import InferredLifecycle
 from pydantic_ai.realtime._lifecycle import (
     InputAdded,
     InputLost,
@@ -72,6 +72,7 @@ from pydantic_ai.realtime.codec import (
     SessionUsage,
     ToolCall,
     ToolCallCancelled,
+    ToolResult,
 )
 from pydantic_ai.usage import RequestUsage
 
@@ -1132,7 +1133,7 @@ def test_a_turn_recorded_after_the_reply_to_it_is_evicted_before_that_reply() ->
 
 def test_a_held_filler_keeps_the_wait_through_a_transcript_until_the_model_carries_on() -> None:
     """Gemini's extended-thinking model holds its filler open (`IN_PROGRESS`): the user's words then don't end the wait."""
-    tracker = GeminiLifecycle(transcribes=True)
+    tracker = InferredLifecycle(transcribes=True, transcripts_lag_replies=True)
     session_core = core()
 
     def message(*codec: Any) -> None:
@@ -1159,3 +1160,12 @@ def test_a_held_filler_keeps_the_wait_through_a_transcript_until_the_model_carri
             'None [assistant:Still checking.] complete stop',
         ]
     )
+
+
+def test_a_reply_the_provider_will_never_give_is_lost_once() -> None:
+    """GPT-Live drops a tool result whose backend gave up: its reply is lost, unless a response already took it."""
+    tracker = InferredLifecycle(transcribes=True)
+    tracker.input_sent(0, ToolResult('call_1', output='sunny'))
+    tracker.input_unanswerable(0)
+    tracker.input_unanswerable(0)
+    assert tracker.take_pending() == snapshot([InputAdded(input_id=0), InputLost(input_ids=(0,))])