[pull] main from openai:main

src/openai/_send_queue.py

@@ -37,26 +37,34 @@ def enqueue(self, data: str) -> None:
             self._queue.append((data, byte_length))
             self._bytes += byte_length
 
-    def flush_sync(self, send: typing.Callable[[str], object]) -> None:
+    def flush_sync(self, send: typing.Callable[[str], object], *, requeue_failed: bool = True) -> None:
         """Send every queued message via *send*.
 
         If *send* raises, the failing message and all subsequent messages
-        are re-queued and the error is re-raised.
+        are re-queued and the error is re-raised. When `requeue_failed` is
+        false, release the attempted message even on failure or interruption.
         """
         while isinstance(pending := self._begin_flush(), threading.Event):
             pending.wait()
 
         try:
             while pending:
                 data, byte_length = pending[0]
-                send(data)
-                with self._lock:
-                    pending.popleft()
-                    self._bytes -= byte_length
+                sent = False
+                try:
+                    send(data)
+                    sent = True
+                finally:
+                    if sent or not requeue_failed:
+                        with self._lock:
+                            pending.popleft()
+                            self._bytes -= byte_length
         finally:
             self._end_flush(pending)
 
-    async def flush_async(self, send: typing.Callable[[str], typing.Awaitable[object]]) -> None:
+    async def flush_async(
+        self, send: typing.Callable[[str], typing.Awaitable[object]], *, requeue_failed: bool = True
+    ) -> None:
         """Async variant of :meth:`flush_sync`."""
         while isinstance(pending := self._begin_flush(), threading.Event):
             # Waiting in a worker keeps the event loop responsive. Cancellation
@@ -66,10 +74,15 @@ async def flush_async(self, send: typing.Callable[[str], typing.Awaitable[object
         try:
             while pending:
                 data, byte_length = pending[0]
-                await send(data)
-                with self._lock:
-                    pending.popleft()
-                    self._bytes -= byte_length
+                sent = False
+                try:
+                    await send(data)
+                    sent = True
+                finally:
+                    if sent or not requeue_failed:
+                        with self._lock:
+                            pending.popleft()
+                            self._bytes -= byte_length
         finally:
             self._end_flush(pending)
 

src/openai/lib/live/README.md

@@ -0,0 +1,33 @@
+# Live WebSocket lifecycle
+
+`client.live.connect()` and `client.live.forks.connect(session_id=...)` leave
+startup to the caller: send `connection.session.start(...)` and wait for
+`session.started`. A sideband connection created by
+`client.live.sideband.connect(session_id=...)` attaches to an existing session.
+It does not send a start or require a new `session.started`; an attachment's
+short replay is not a complete session snapshot.
+
+Direct `recv()` and `recv_bytes()` calls report transport errors to their
+caller. Iterator reconnection is available only when an `on_reconnecting`
+callback was explicitly supplied. The callback controls retry and can update
+credentials or query parameters. A new socket is not proof that the previous
+Live session, recording, or application state was restored. Applications own
+their recovery decision and any necessary startup or state reconstruction.
+Don't use multiple physical readers: a dispatcher owns the read loop while it
+runs. Detaching or closing one transcript grouper doesn't cancel another
+observer or the connection.
+
+All modes preserve the manager's caller-configured `max_queue_size`, including
+if the queue was empty at connection time. Only messages that haven't been
+attempted on a socket remain eligible for flushing after a retry. A direct send
+exception is raised to its caller and cannot prove the server didn't receive
+that message. A failure while flushing an already queued message still logs a
+warning, as before; it cannot be raised at the original queueing call. Neither
+failed attempt is automatically retried. Unattempted pre-open messages and
+messages explicitly queued during recovery remain in order within their
+existing budget. No automatic session reopening or restoration is implied.
+
+Treat each wire error as an event to handle, not as a successful session result.
+In particular, preserve `session_storage_failed` if `session.closed` follows;
+the close doesn't mean that recording storage succeeded. Unknown events and
+fields are preserved for callers that need them.

src/openai/lib/responses_websocket/README.md

@@ -108,8 +108,47 @@ 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.
 
+`detailed_snapshot()` returns a separate mutable view when you need the fields
+beyond selected text and tool inputs. It contains `stream_id`, `response_id`,
+`terminal_type`, `response` and `output`. `response` is the last observed
+lifecycle response metadata (excluding `output`), or `None` if none arrived.
+`output` is a list of `{"output_index": index, "item": metadata, "content": rows}`.
+Each content row is `{"content_index": index, "part": observed_fields}`. The
+part's `annotations`, when present as a list, use
+`{"annotation_index": index, "annotation": observed_fields}` rows. Indices may
+be sparse; list position is not the API index.
+For a message that has no projected content, the row's `content` is omitted,
+null or empty according to what was actually received.
+
+For example, after collecting an item or terminal as above:
+
+```python
+details = accumulator.detailed_snapshot()
+for output in details["output"]:
+    for content in output.get("content") or []:
+        part = content["part"]
+        # Fields exist only if received: a WS delta may have no part/item type.
+        text = part.get("text")
+        citations = part.get("annotations")
+        token_scores = part.get("logprobs")
+```
+
+Logprobs accumulate with text deltas and a supplied `output_text.done.logprobs`
+replaces them, including empty or null values. Content/item/lifecycle replacements
+also replace their corresponding annotations and other metadata; text-only done
+events do not erase citations. Refusal, tool/MCP and unknown item/part fields are
+retained as observed, with unset and null distinct. Unknown standalone events still
+pass through unchanged and are not accumulated. Annotation events enrich matching
+known items on the accumulator's lane; annotations received before any matching
+item/text remain available in the original event and do not start or replace a turn.
+A partial field or unknown type
+is provisional, not a fabricated validated response or a successful tool result.
+You may mutate this returned view without changing the accumulator, events, or
+earlier snapshots. The original `snapshot()` remains immutable and hashable.
+
 `snapshot()` materializes the entire current projection and joins retained
-fragments. It is proportional to the accumulated output, so requesting it after
+fragments. `detailed_snapshot()` also materializes and copies its whole projection.
+Both are proportional to the accumulated output, so requesting either 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

src/openai/lib/responses_websocket/_accumulator.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
-from typing import cast
+from copy import deepcopy
+from typing import Any, cast
 from dataclasses import field, dataclass
 
 from ._session import ResponsesWebSocketError, _field
-from ..._compat import model_copy
+from ..._compat import PYDANTIC_V1, model_copy
+from ..._models import BaseModel
 from ...types.responses import Response
 from ...types.responses.responses_server_event import ResponsesServerEvent
 
@@ -47,6 +49,9 @@ class _Output:
     arguments: list[str] = field(default_factory=list[str])
     input: list[str] = field(default_factory=list[str])
     text: dict[str, list[str]] = field(default_factory=dict[str, list[str]])
+    data: dict[str, object] = field(default_factory=dict[str, object])
+    content: dict[str, dict[str, object]] = field(default_factory=dict[str, dict[str, object]])
+    annotations: dict[str, dict[str, object]] = field(default_factory=dict[str, dict[str, object]])
 
 
 class ResponsesWebSocketAccumulator:
@@ -67,6 +72,7 @@ def __init__(self) -> None:
         # String keys get Python's randomized hash. Hex preserves arbitrary-size
         # non-negative indices without decimal string conversion limits.
         self._output: dict[str, _Output] = {}
+        self._response: dict[str, object] | None = None
         self._final: Response | None = None
         self._error: Exception | None = None
 
@@ -75,6 +81,7 @@ def reset(self) -> None:
         self._stream_id = self._response_id = self._terminal_type = None
         self._bound = False
         self._output.clear()
+        self._response = None
         self._final = self._error = None
 
     def snapshot(self) -> ResponsesWebSocketSnapshot:
@@ -105,6 +112,61 @@ def snapshot(self) -> ResponsesWebSocketSnapshot:
             ),
         )
 
+    def detailed_snapshot(self) -> dict[str, Any]:
+        """Return an independent mutable projection of observed response, item and part data.
+
+        Output and content are lists of indexed rows, even when wire indices are
+        sparse. Part annotations use the same indexed-row form. Unset and null
+        fields stay distinct; missing scaffolding never invents a Response or
+        an item/part type. Cost is proportional to the full accumulated data.
+        """
+        output: list[dict[str, object]] = []
+        for index, item in sorted(self._output.items(), key=lambda pair: int(pair[0], 16)):
+            data = deepcopy(item.data)
+            if item.item_id:
+                data["id"] = item.item_id
+            if item.arguments:
+                data["arguments"] = "".join(item.arguments)
+            if item.input:
+                data["input"] = "".join(item.input)
+            content: list[dict[str, object]] = []
+            for pos in sorted(
+                item.content.keys() | item.text.keys() | item.annotations.keys(), key=lambda k: int(k, 16)
+            ):
+                part = deepcopy(item.content.get(pos, {}))
+                if pos in item.text:
+                    part["text"] = "".join(item.text[pos])
+                original = part.get("annotations")
+                if isinstance(original, list) or pos in item.annotations:
+                    annotations = (
+                        {hex(i): value for i, value in enumerate(cast("list[object]", original))}
+                        if isinstance(original, list)
+                        else {}
+                    )
+                    annotations.update(deepcopy(item.annotations.get(pos, {})))
+                    part["annotations"] = [
+                        {"annotation_index": int(i, 16), "annotation": annotation}
+                        for i, annotation in sorted(annotations.items(), key=lambda pair: int(pair[0], 16))
+                    ]
+                content.append({"content_index": int(pos, 16), "part": part})
+            row: dict[str, object] = {"output_index": int(index, 16), "item": data}
+            if item.type == "message":
+                # Message content is extracted into indexed rows. Retain the
+                # presence marker until a part/delta supplies projected content.
+                if "content" in data or content:
+                    original_content = data.pop("content", None)
+                    row["content"] = content if content or isinstance(original_content, list) else None
+            else:
+                row["content"] = content
+            output.append(row)
+        return {
+            "stream_id": self._stream_id,
+            "response_id": self._response_id,
+            "terminal_type": self._terminal_type,
+            "response": deepcopy(self._response),
+            "output": output,
+        }
+
     def get_final_response(self) -> Response:
         """Return an independent copy of the received terminal response, never a partial success."""
         error = self._error
@@ -134,6 +196,7 @@ def add_event(self, event: ResponsesServerEvent) -> None:
             "response.content_part.done",
             "response.output_text.delta",
             "response.output_text.done",
+            "response.output_text.annotation.added",
             "response.function_call_arguments.delta",
             "response.function_call_arguments.done",
             "response.mcp_call_arguments.delta",
@@ -144,6 +207,31 @@ def add_event(self, event: ResponsesServerEvent) -> None:
         }:
             return
         stream_id = _field(event, "stream_id")
+        # Annotations were historically ignored. They may enrich the matching
+        # known item, but never bind a lane/turn, replace or retire an item, or
+        # change the errors observed by existing accumulator callers.
+        if kind == "response.output_text.annotation.added":
+            output_pos = _field(event, "output_index")
+            content_pos = _field(event, "content_index")
+            annotation_pos = _field(event, "annotation_index")
+            if (
+                self._bound
+                and self._stream_id == stream_id
+                and self._terminal_type is None
+                and self._error is None
+                and all(
+                    isinstance(value, int) and not isinstance(value, bool) and value >= 0
+                    for value in (output_pos, content_pos, annotation_pos)
+                )
+            ):
+                existing = self._output.get(hex(output_pos))
+                if existing is not None and existing.item_id == _field(event, "item_id"):
+                    value_data = _data(event, include={"annotation"})
+                    if "annotation" in value_data:
+                        existing.annotations.setdefault(hex(content_pos), {})[hex(annotation_pos)] = value_data[
+                            "annotation"
+                        ]
+            return
         if stream_id is not None and not isinstance(stream_id, str):
             raise ValueError("WebSocket stream_id must be a string or null")
         if self._bound and self._stream_id != stream_id:
@@ -186,6 +274,7 @@ def add_event(self, event: ResponsesServerEvent) -> None:
                         self._error = error
                     raise
                 self._output = replacement
+            self._response = _data(response, exclude={"output"})
             self._response_id = response_id or self._response_id
             self._bound, self._stream_id = True, stream_id
             if terminal:
@@ -248,16 +337,28 @@ def add_event(self, event: ResponsesServerEvent) -> None:
             "response.output_text.delta",
             "response.output_text.done",
         }:
+            position = hex(pos)
             if kind == "response.output_text.delta":
-                item.text.setdefault(hex(pos), []).append(value)
+                item.text.setdefault(position, []).append(value)
+                data = item.content.setdefault(position, {})
+                prob_data = _data(event, include={"logprobs"})
+                previous = data.get("logprobs")
+                incoming = prob_data.get("logprobs")
+                if isinstance(previous, list) and isinstance(incoming, list):
+                    cast("list[object]", previous).extend(cast("list[object]", incoming))
+                else:
+                    data.update(prob_data)
             elif kind == "response.output_text.done":
-                item.text[hex(pos)] = [value]
+                item.text[position] = [value]
+                item.content.setdefault(position, {}).update(_data(event, include={"logprobs"}))
             else:
                 part = _field(event, "part")
+                item.content[position] = _data(part)
+                item.annotations.pop(position, None)
                 if _field(part, "type") == "output_text":
-                    item.text[hex(pos)] = [value]
+                    item.text[position] = [value]
                 else:
-                    item.text.pop(hex(pos), None)
+                    item.text.pop(position, None)
         elif kind in {"response.function_call_arguments.delta", "response.mcp_call_arguments.delta"}:
             item.arguments.append(value)
         elif kind in {"response.function_call_arguments.done", "response.mcp_call_arguments.done"}:
@@ -276,21 +377,32 @@ def _add_item(output: dict[str, _Output], index: int, source: object) -> None:
             type=_text_field(source, "type"),
             name=_optional_text_field(source, "name"),
             call_id=_optional_text_field(source, "call_id"),
+            data=_data(source, exclude={"content"} if _field(source, "type") == "message" else None),
         )
         if item.type == "message":
             content = _field(source, "content")
             if content is not None:
                 if not isinstance(content, list):
                     raise ValueError("WebSocket message content must be a list or null")
+                # Marker only; keep actual parts indexed once, never copied twice.
+                item.data["content"] = []
                 for pos, part in enumerate(cast("list[object]", content)):
+                    if part is not None:
+                        item.content[hex(pos)] = _data(part)
                     if part is not None and _text_field(part, "type") == "output_text":
                         text = _optional_text_field(part, "text")
                         if text is not None:
                             item.text[hex(pos)] = [text]
+            else:
+                item.data.update(_data(source, include={"content"}))
         elif item.type in {"function_call", "mcp_call", "mcp_approval_request"}:
-            item.arguments = [_optional_text_field(source, "arguments") or ""]
+            arguments = _optional_text_field(source, "arguments")
+            if arguments is not None:
+                item.arguments = [arguments]
         elif item.type == "custom_tool_call":
-            item.input = [_optional_text_field(source, "input") or ""]
+            input_text = _optional_text_field(source, "input")
+            if input_text is not None:
+                item.input = [input_text]
         key = hex(index)
         previous = output.get(key)
         if previous is not None:
@@ -302,6 +414,24 @@ def _add_item(output: dict[str, _Output], index: int, source: object) -> None:
         output[key] = item
 
 
+def _data(source: object, *, include: set[str] | None = None, exclude: set[str] | None = None) -> dict[str, object]:
+    if isinstance(source, BaseModel):
+        return deepcopy(
+            source.model_dump(
+                mode="python", by_alias=True, exclude_unset=True, include=include, exclude=exclude, warnings=PYDANTIC_V1
+            )
+        )
+    if isinstance(source, dict):
+        return deepcopy(
+            {
+                key: value
+                for key, value in cast("dict[str, object]", source).items()
+                if (include is None or key in include) and (exclude is None or key not in exclude)
+            }
+        )
+    return {}
+
+
 def _text_field(value: object, name: str) -> str:
     text = _field(value, name)
     if not isinstance(text, str):