[pull] main from openai:main

src/openai/lib/live/_transcript_grouping.py

@@ -96,10 +96,11 @@ def process(self, fragments: list[TranscriptFragment]) -> list[GroupingUpdate]:
         return events
 
     def advance(self, time_ms: float, has_incoming_user: bool = False) -> list[GroupingUpdate]:
+        # Use the same sums as deadline() so fractional deadlines remain reachable.
         if (
             self._current is not None
             and self._buffered is not None
-            and time_ms - self._current.end_ms >= self._options.min_turn_separation_ms
+            and time_ms >= self._current.end_ms + self._options.min_turn_separation_ms
             and not self._keep_backchannel(time_ms)
         ):
             self._buffered = self._maybe_drop_backchannel(time_ms)
@@ -109,7 +110,7 @@ def advance(self, time_ms: float, has_incoming_user: bool = False) -> list[Group
             self._current is not None
             and self._current.speaker == "assistant"
             and not has_incoming_user
-            and time_ms - self._current.end_ms >= self._options.assistant_silence_ms
+            and time_ms >= self._current.end_ms + self._options.assistant_silence_ms
         ):
             return self._finish_current("inactivity")
         return []
@@ -274,7 +275,7 @@ def _keep_backchannel(self, time_ms: float) -> bool:
             and self._buffered.can_drop_as_backchannel
             and not self._user_continued()
             and not self._recent_assistant()
-            and time_ms - self._buffered.end_ms < self._options.backchannel_isolation_ms
+            and time_ms < self._buffered.end_ms + self._options.backchannel_isolation_ms
         )
 
     def _maybe_drop_backchannel(
@@ -292,7 +293,7 @@ def _maybe_drop_backchannel(
         ):
             return self._buffered
         if next_fragment is None and (
-            time_ms is None or time_ms - self._buffered.end_ms < self._options.backchannel_isolation_ms
+            time_ms is None or time_ms < self._buffered.end_ms + self._options.backchannel_isolation_ms
         ):
             return self._buffered
         return None if self._buffered.can_drop_as_backchannel else self._buffered

src/openai/lib/responses_websocket/README.md

@@ -68,6 +68,62 @@ the next response. It does not wait for a hypothetical future successor. Use
 `recv()` to observe that boundary when coordinating steering; accepted steering
 alone does not prove that a successor has started.
 
+For incremental text and tool-input snapshots, opt in with a separate helper
+fed by the events you already receive:
+
+```python
+from openai.lib.responses_websocket import ResponsesWebSocketAccumulator
+
+accumulator = ResponsesWebSocketAccumulator()
+expected_stream_id = "conversation"  # Same ID passed to session.lane(); None for session.default.
+while True:
+    event = await lane.recv()  # Use lane.recv(timeout=...) in a sync session.
+    # Original event fields, including unknown variants/fields, remain available.
+    # Inspect events in memory before filtering. Never log raw events: they can
+    # contain customer data, tool arguments, credentials, or error details.
+    if getattr(event, "stream_id", None) != expected_stream_id:
+        continue
+    accumulator.add_event(event)
+    event_type = event.type
+    if event.type == "response.output_text.delta":
+        pass  # Update the application's UI with event.delta (provisional).
+    elif event.type == "response.output_item.done":
+        pass  # Replace the provisional item at event.output_index with event.item.
+    elif isinstance(event_type, str) and event_type in {"response.completed", "response.failed", "response.incomplete"}:
+        snapshot = accumulator.snapshot()  # One full projection of all items.
+        response = accumulator.get_final_response()
+        break
+accumulator.reset()  # Does not close the lane or connection.
+```
+
+Use one helper per lane and reset it before the next turn. The default lane
+also receives raw events for unregistered or detached lanes. Observe those
+events before filtering, then feed only the response you are collecting.
+The same check applies when consuming directly from a connection.
+
+Snapshots are immutable and contain selected text, function arguments, and custom tool input,
+grouped by output/content index and item ID. They never execute tools. Done
+events replace provisional fields. Full item replacements discard old fields;
+a changed nonempty item ID starts fresh at its index. A supplied response
+output list overrides the projected items, including an explicit empty list;
+omitted or null output retains only the helper's earlier projections.
+
+`snapshot()` materializes the entire current projection and joins retained
+fragments. It is proportional to the accumulated output, so requesting it after
+every delta or completed item repeatedly rebuilds growing prefixes. Use the
+original event for progress, including the final item itself on
+`response.output_item.done`. Read the full snapshot at a terminal or on explicit
+user demand. A per-delta progress display is provisional; done events can shorten or
+correct earlier text. A UI can use the received item to replace the item at
+`event.output_index`, and use the terminal snapshot for the whole projection.
+
+The helper's `get_final_response()` returns a copy of the exact received server
+response, including its original missing/null/empty output, on completed,
+failed or incomplete. It never substitutes the projection for that response.
+Before a valid terminal it raises; protocol errors retain their original event.
+This caller-fed helper owns no socket, reader or timers and can also observe
+events from a connection when there is no session.
+
 Cancel an async `recv` or `get_final_response` wait with normal asyncio
 cancellation. It leaves queued events and accumulated state available for a later
 wait, and does not close the lane or connection. Synchronous waits accept

src/openai/lib/responses_websocket/__init__.py

@@ -7,3 +7,8 @@
     ResponsesWebSocketBufferError as ResponsesWebSocketBufferError,
     AsyncResponsesWebSocketSession as AsyncResponsesWebSocketSession,
 )
+from ._accumulator import (
+    ResponsesWebSocketOutput as ResponsesWebSocketOutput,
+    ResponsesWebSocketSnapshot as ResponsesWebSocketSnapshot,
+    ResponsesWebSocketAccumulator as ResponsesWebSocketAccumulator,
+)