feat(responses): preserve detailed WebSocket accumulator snapshots (openai#3981)

The opt-in Responses WebSocket accumulator currently retains selected
text/tool fields but loses response metadata, streamed logprobs,
citations and non-text parts. This adds `detailed_snapshot()`, an
independent mutable view of observed fields and sparse indexed
output/content/annotation rows. Existing immutable, hashable
`snapshot()`, raw events, and the exact server terminal response keep
their contracts.

Done events replace supplied logprobs and whole part/item/response
events replace the relevant projections. Missing and null are preserved,
and unknown parts or tool fields remain provisional data instead of
fabricated validated final models. Annotations only enrich a matching
item on the bound lane; they cannot bind a turn or retire an unrelated
item. The helper still never reads, sends, runs tools or changes
sockets. Handwritten helper, docs and tests only.

Validation:
- Before change: 12 new real-wire tests failed. Review regressions
reproduced: 8 failures (foreign or omitted annotation, null/missing
content).
- Current accumulator/session suite: 438 passed on each of Pydantic 1
and 2, sync and async local WebSockets.
- SDK mypy: 1,865 files passed; Ruff and formatting pass.
- Base's verified custom-code check: unchanged 7,826 / 10,000; no schema
or generator change.

Author: markstuart-oai

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):