diff --git a/src/openai/lib/responses_websocket/README.md b/src/openai/lib/responses_websocket/README.md index 52833cca43..98c62bdeb9 100644 --- a/src/openai/lib/responses_websocket/README.md +++ b/src/openai/lib/responses_websocket/README.md @@ -115,6 +115,7 @@ omitted or null output retains only the helper's earlier projections. 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. +This includes `response.queued` and `response.in_progress` refreshes. `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 @@ -151,6 +152,26 @@ 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. +The detailed view also tracks reasoning summary/text, shell commands and output, +interpreter code, and search, image and MCP call status updates. Reasoning text stays +separate from the compact snapshot's `output_text`. Full item and part replacements +override their provisional fields without discarding another part's progress. +Reasoning `item.content` and its indexed content rows show the same observed text. + +Shell output rows also contain `shell_output`, a list of +`{"command_index": index, "data": observed_fields}` entries. Commands can produce +output out of order. These indexed rows retain later commands even before earlier +commands have produced output. A delta adds observed stdout/stderr. A done event +replaces that command's data with `{"output": [all_received_contents]}`, including +an empty array. The rows preserve each content's order, outcome and metadata. +`item.output` retains the received whole-item array: it does not identify command +boundaries, so the helper does not guess how command progress fits into that array. +A text delta does not invent an outcome, exit code, or missing stdout/stderr. +Full-item and terminal replacements clear provisional command facts. A matching +shell call supplies a command bound; without it, the bound remains unknown. +Queue and response limits +are unchanged; the helper never allocates missing slots up to a sparse wire index. + `snapshot()` materializes the entire current projection and joins retained fragments. `detailed_snapshot()` also materializes and copies its whole projection. Both are proportional to the accumulated output, so requesting either after diff --git a/src/openai/lib/responses_websocket/_accumulator.py b/src/openai/lib/responses_websocket/_accumulator.py index 38e7f2e569..5fcc674d78 100644 --- a/src/openai/lib/responses_websocket/_accumulator.py +++ b/src/openai/lib/responses_websocket/_accumulator.py @@ -1,7 +1,7 @@ from __future__ import annotations from copy import deepcopy -from typing import Any, cast +from typing import Any, Literal, cast from dataclasses import field, dataclass from ._session import ResponsesWebSocketError, _field @@ -10,6 +10,38 @@ from ...types.responses import Response from ...types.responses.responses_server_event import ResponsesServerEvent +_RichField = Literal["summary", "reasoning", "code", "command", "stdout", "stderr"] + +# Fixed wire fields only: these are not paths supplied by events or callers. +_RICH_TEXT_EVENTS: dict[str, tuple[str, Literal["summary", "reasoning", "code", "command"], str, str, str | None]] = { + "response.reasoning_summary_text.delta": ("reasoning", "summary", "delta", "delta", "summary_index"), + "response.reasoning_summary_text.done": ("reasoning", "summary", "done", "text", "summary_index"), + "response.reasoning_text.delta": ("reasoning", "reasoning", "delta", "delta", "content_index"), + "response.reasoning_text.done": ("reasoning", "reasoning", "done", "text", "content_index"), + "response.code_interpreter_call_code.delta": ("code_interpreter_call", "code", "delta", "delta", None), + "response.code_interpreter_call_code.done": ("code_interpreter_call", "code", "done", "code", None), + "response.shell_call_command.delta": ("shell_call", "command", "delta", "delta", "command_index"), + "response.shell_call_command.done": ("shell_call", "command", "done", "command", "command_index"), +} +_RICH_STATUS_EVENTS = { + f"response.{kind}.{status}": (kind, status) + for kind, statuses in ( + ("code_interpreter_call", ("in_progress", "interpreting", "completed")), + ("file_search_call", ("in_progress", "searching", "completed")), + ("web_search_call", ("in_progress", "searching", "completed")), + ("image_generation_call", ("in_progress", "generating", "completed")), + ("mcp_call", ("in_progress", "completed", "failed")), + ) + for status in statuses +} +_RICH_PART_EVENTS = { + "response.reasoning_summary_part.added": "reasoning", + "response.reasoning_summary_part.done": "reasoning", + "response.shell_call_command.added": "shell_call", + "response.shell_call_output_content.delta": "shell_call_output", + "response.shell_call_output_content.done": "shell_call_output", +} + @dataclass(frozen=True) class ResponsesWebSocketOutput: @@ -53,10 +85,12 @@ class _Output: 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]]) + rich_text: dict[tuple[_RichField, str], list[str]] = field(default_factory=dict[tuple[_RichField, str], list[str]]) + shell_output: dict[str, dict[str, object]] = field(default_factory=dict[str, dict[str, object]]) class ResponsesWebSocketAccumulator: - """Opt-in, caller-fed collection of text, function/MCP arguments and custom tool input. + """Caller-fed text/tool inputs and detailed reasoning, code and shell progress. Feed typed events from a connection or lane's recv(). Use one accumulator per lane and call reset() before another turn. The helper never reads, sends, @@ -76,6 +110,7 @@ def __init__(self) -> None: self._response: dict[str, object] | None = None self._final: Response | None = None self._error: Exception | None = None + self._shell_command_counts: dict[str, int] = {} def reset(self) -> None: """Release this helper's state only; prior snapshots and the socket remain usable.""" @@ -84,6 +119,7 @@ def reset(self) -> None: self._output.clear() self._response = None self._final = self._error = None + self._shell_command_counts.clear() def snapshot(self) -> ResponsesWebSocketSnapshot: """Materialize the entire current immutable projection. @@ -124,13 +160,18 @@ def detailed_snapshot(self) -> dict[str, Any]: 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: + for (field, pos), fragments in item.rich_text.items(): + if field not in {"reasoning", "stdout", "stderr"}: + container, key = _rich_text_location(data, field, int(pos, 16)) + container[key] = "".join(fragments) + if item.item_id and not item.data: 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]] = [] + reasoning_parts: dict[int, dict[str, object]] = {} for pos in sorted( item.content.keys() | item.text.keys() | item.refusals.keys() | item.annotations.keys(), key=lambda k: int(k, 16), @@ -140,6 +181,13 @@ def detailed_snapshot(self) -> dict[str, Any]: part["text"] = "".join(item.text[pos]) if pos in item.refusals: part["refusal"] = "".join(item.refusals[pos]) + reasoning_text = item.rich_text.get(("reasoning", pos)) + if reasoning_text is not None: + part["text"] = "".join(reasoning_text) + if item.type == "reasoning": + # Item.content keeps wire-shaped metadata; indexed content + # below may wrap annotations for its public detailed view. + reasoning_parts[int(pos, 16)] = deepcopy(part) original = part.get("annotations") if isinstance(original, list) or pos in item.annotations: annotations = ( @@ -153,6 +201,16 @@ def detailed_snapshot(self) -> dict[str, Any]: for i, annotation in sorted(annotations.items(), key=lambda pair: int(pair[0], 16)) ] content.append({"content_index": int(pos, 16), "part": part}) + if item.type == "reasoning" and reasoning_parts: + _project_rich_parts(data, "content", reasoning_parts) + shell_parts: dict[int, dict[str, object]] = {} + for pos, source in sorted(item.shell_output.items(), key=lambda pair: int(pair[0], 16)): + part = deepcopy(source) + for field in ("stdout", "stderr"): + fragments = item.rich_text.get((field, pos)) + if fragments is not None: + part[field] = "".join(fragments) + shell_parts[int(pos, 16)] = 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 @@ -162,6 +220,8 @@ def detailed_snapshot(self) -> dict[str, Any]: row["content"] = content if content or isinstance(original_content, list) else None else: row["content"] = content + if item.type == "shell_call_output": + row["shell_output"] = [{"command_index": pos, "data": part} for pos, part in shell_parts.items()] output.append(row) return { "stream_id": self._stream_id, @@ -190,6 +250,7 @@ def add_event(self, event: ResponsesServerEvent) -> None: kind = _field(event, "type") if not isinstance(kind, str) or kind not in { "response.created", + "response.queued", "response.in_progress", "response.completed", "response.failed", @@ -209,6 +270,9 @@ def add_event(self, event: ResponsesServerEvent) -> None: "response.mcp_call_arguments.done", "response.custom_tool_call_input.delta", "response.custom_tool_call_input.done", + *_RICH_TEXT_EVENTS, + *_RICH_STATUS_EVENTS, + *_RICH_PART_EVENTS, "error", }: return @@ -249,7 +313,7 @@ def add_event(self, event: ResponsesServerEvent) -> None: self._error = protocol_error raise protocol_error terminal = kind in {"response.completed", "response.failed", "response.incomplete"} - if kind in {"response.created", "response.in_progress"} or terminal: + if kind in {"response.created", "response.queued", "response.in_progress"} or terminal: response = _field(event, "response") if not isinstance(response, Response): error = ValueError("WebSocket event is missing a valid response") @@ -280,6 +344,9 @@ def add_event(self, event: ResponsesServerEvent) -> None: self._error = error raise self._output = replacement + self._shell_command_counts.clear() + for item in replacement.values(): + self._index_shell_call(item) self._response = _data(response, exclude={"output"}) self._response_id = response_id or self._response_id self._bound, self._stream_id = True, stream_id @@ -287,11 +354,21 @@ def add_event(self, event: ResponsesServerEvent) -> None: self._final = model_copy(response, deep=True) self._terminal_type = kind return + if kind in _RICH_TEXT_EVENTS or kind in _RICH_STATUS_EVENTS or kind in _RICH_PART_EVENTS: + self._add_rich_event(event, kind) + return index = _field(event, "output_index") if not isinstance(index, int) or isinstance(index, bool) or index < 0: raise ValueError("WebSocket output_index must be a non-negative integer") if kind in {"response.output_item.added", "response.output_item.done"}: + previous = self._output.get(hex(index)) self._add_item(self._output, index, _field(event, "item")) + current = self._output.get(hex(index)) + if previous is not current: + if previous is not None and previous.type == "shell_call" and previous.call_id is not None: + self._shell_command_counts.pop(previous.call_id, None) + if current is not None: + self._index_shell_call(current) self._bound, self._stream_id = True, stream_id return # Validate consumed values before replacing an item's retained state. @@ -345,10 +422,21 @@ def add_event(self, event: ResponsesServerEvent) -> None: item = self._output.get(key) if item is not None and item_id in item.retired_ids: return + if item is not None and item.type in {"function_call", "custom_tool_call"} and "id" not in item.data: + # Full records that omit id bind only to their own progress family. + matching_family = ( + kind in {"response.function_call_arguments.delta", "response.function_call_arguments.done"} + if item.type == "function_call" + else kind in {"response.custom_tool_call_input.delta", "response.custom_tool_call_input.done"} + ) + if not matching_family or (item.item_id and item_id != item.item_id): + return if item is None: item = _Output(item_id=item_id) self._output[key] = item elif item_id and item.item_id and item_id != item.item_id: + if item.type == "shell_call" and item.call_id is not None: + self._shell_command_counts.pop(item.call_id, None) item.retired_ids.add(item.item_id) item = _Output(item_id=item_id, retired_ids=item.retired_ids) self._output[key] = item @@ -383,6 +471,7 @@ def add_event(self, event: ResponsesServerEvent) -> None: else: part = _field(event, "part") item.content[position] = _data(part) + item.rich_text.pop(("reasoning", position), None) item.annotations.pop(position, None) if _field(part, "type") == "output_text": item.text[position] = [value] @@ -398,6 +487,111 @@ def add_event(self, event: ResponsesServerEvent) -> None: elif kind == "response.custom_tool_call_input.done": item.input = [value] + def _add_rich_event(self, event: ResponsesServerEvent, kind: str) -> None: + index = _index_field(event, "output_index") + # Shell-command events identify their call by output_index only. An + # unmodeled item_id extra must stay raw, not become a routing guard. + item_id = None if kind.startswith("response.shell_call_command.") else _text_field(event, "item_id") + item = self._output.get(hex(index)) + text_event = _RICH_TEXT_EVENTS.get(kind) + status_event = _RICH_STATUS_EVENTS.get(kind) + expected_kind = text_event[0] if text_event else status_event[0] if status_event else _RICH_PART_EVENTS[kind] + # Progress cannot create or retire an item, including when an old ID + # arrives after a full-item replacement at this output index. + if item is None or item.type != expected_kind or (item_id is not None and item.item_id != item_id): + return + if status_event is not None: + item.data["status"] = status_event[1] + return + if text_event is not None: + _, field, operation, value_field, index_field = text_event + pos = 0 if index_field is None else _index_field(event, index_field) + text = _text_field(event, value_field) + if field == "reasoning": + part = item.content.get(hex(pos)) + if part is None or part.get("type") != "reasoning_text": + raise ValueError("WebSocket reasoning text is missing its content part") + container, key = part, "text" + else: + container, key = _rich_text_location(item.data, field, pos) + buffer_key = (field, hex(pos)) + if operation == "done": + item.rich_text[buffer_key] = [text] + else: + seed = container.get(key) if isinstance(container, dict) else container[key] + if field == "code" and seed is None: + seed = "" + if not isinstance(seed, str): + raise ValueError("WebSocket rich text seed must be a string") + item.rich_text.setdefault(buffer_key, [seed]).append(text) + return + if kind.startswith("response.reasoning_summary_part."): + pos = _index_field(event, "summary_index") + parts = _rich_array(item.data.get("summary"), "summary") + added = kind == "response.reasoning_summary_part.added" + _check_rich_position(parts, pos, append=added) + part = _field(event, "part") + if _text_field(part, "type") != "summary_text": + raise ValueError("WebSocket summary part must be summary_text") + _text_field(part, "text") + replacement = _data(part) + if pos == len(parts): + parts.append(replacement) + else: + parts[pos] = replacement + item.rich_text.pop(("summary", hex(pos)), None) + return + pos = _index_field(event, "command_index") + if kind == "response.shell_call_command.added": + action = item.data.get("action") + if not isinstance(action, dict): + raise ValueError("WebSocket shell call is missing its action") + commands = _rich_array(action.get("commands"), "commands") + _check_rich_position(commands, pos, append=True) + command = _text_field(event, "command") + if pos == len(commands): + commands.append(command) + self._index_shell_call(item) + else: + commands[pos] = command + item.rich_text.pop(("command", hex(pos)), None) + return + bound = self._shell_command_counts.get(item.call_id) if item.call_id is not None else None + if bound is not None and pos >= bound: + raise ValueError("WebSocket shell output is missing its command") + if kind == "response.shell_call_output_content.done": + replacement = _rich_array(_field(event, "output"), "completed shell output") + completed = [_data(part) for part in replacement] + for part in completed: + _text_field(part, "stdout") + _text_field(part, "stderr") + item.shell_output[hex(pos)] = {"output": completed} + item.rich_text.pop(("stdout", hex(pos)), None) + item.rich_text.pop(("stderr", hex(pos)), None) + else: + delta = _field(event, "delta") + part = item.shell_output.get(hex(pos), {}) + changes: list[tuple[_RichField, str, str]] = [] + for field in ("stdout", "stderr"): + incoming = _field(delta, field) + if incoming is not None: + text = _text_field(delta, field) + seed = _text_field(part, field) if field in part else "" + changes.append((field, text, seed)) + # Validate both streams before extending either one. + item.shell_output.setdefault(hex(pos), part) + for field, text, seed in changes: + item.rich_text.setdefault((field, hex(pos)), [seed]).append(text) + + def _index_shell_call(self, item: _Output) -> None: + if item.type != "shell_call" or item.call_id is None: + return + action = item.data.get("action") + commands = action.get("commands") if isinstance(action, dict) else None + # Index only observed counts; absent or malformed raw fields stay unknown. + if isinstance(commands, list): + self._shell_command_counts[item.call_id] = len(commands) + @staticmethod def _add_item(output: dict[str, _Output], index: int, source: object) -> None: if source is None: @@ -433,6 +627,12 @@ def _add_item(output: dict[str, _Output], index: int, source: object) -> None: input_text = _optional_text_field(source, "input") if input_text is not None: item.input = [input_text] + elif item.type == "reasoning": + content = _field(source, "content") + if isinstance(content, list): + for pos, part in enumerate(content): + if isinstance(part, (BaseModel, dict)): + item.content[hex(pos)] = _data(part) key = hex(index) previous = output.get(key) if previous is not None: @@ -469,6 +669,60 @@ def _text_field(value: object, name: str) -> str: return text +def _index_field(value: object, name: str) -> int: + index = _field(value, name) + if not isinstance(index, int) or isinstance(index, bool) or index < 0: + raise ValueError(f"WebSocket {name} must be a non-negative integer") + return index + + +def _rich_array(value: object, name: str) -> list[Any]: + if not isinstance(value, list): + raise ValueError(f"WebSocket rich item is missing its {name}") + return value + + +def _check_rich_position(values: list[Any], index: int, *, append: bool = False) -> None: + if index > len(values) or (index == len(values) and not append): + raise ValueError("WebSocket rich item is missing its indexed field") + + +def _rich_text_location( + data: dict[str, object], field: Literal["summary", "code", "command"], index: int +) -> tuple[Any, str | int]: + """Locate only the finite rich wire fields; no caller-supplied paths.""" + if field == "code": + return data, "code" + if field == "command": + action = data.get("action") + if not isinstance(action, dict): + raise ValueError("WebSocket shell call is missing its action") + commands = _rich_array(action.get("commands"), "commands") + _check_rich_position(commands, index) + return commands, index + parts = _rich_array(data.get("summary"), "summary") + _check_rich_position(parts, index) + part = parts[index] + if not isinstance(part, dict): + raise ValueError("WebSocket rich item is missing its part") + return part, "text" + + +def _project_rich_parts( + data: dict[str, object], field: Literal["content", "output"], parts: dict[int, dict[str, object]] +) -> None: + original = data.get(field) + projected = original if isinstance(original, list) else [] + for pos, part in parts.items(): + if pos < len(projected): + projected[pos] = part + elif pos == len(projected): + projected.append(part) + # Keep holes in indexed rows, never as invented null/outcome values. + if projected: + data[field] = projected + + def _optional_text_field(value: object, name: str) -> str | None: if _field(value, name) is None: return None diff --git a/tests/lib/responses/test_websocket_accumulator_rich.py b/tests/lib/responses/test_websocket_accumulator_rich.py new file mode 100644 index 0000000000..b5017ac169 --- /dev/null +++ b/tests/lib/responses/test_websocket_accumulator_rich.py @@ -0,0 +1,1364 @@ +from __future__ import annotations + +from copy import deepcopy +from typing import Any, cast +from itertools import count + +import pytest + +from openai._compat import model_copy +from openai._models import BaseModel +from openai.types.responses import responses_server_event as events +from openai.lib.responses_websocket import ResponsesWebSocketAccumulator +from openai.types.responses.responses_server_event import ResponsesServerEvent + +INDEX = 7 +ITEM_ID = "rich_item" +LANE = "rich_lane" +_SEQUENCE = count() + + +def response(status: str = "in_progress", **fields: Any) -> dict[str, Any]: + return { + "id": "rich_response", + "object": "response", + "created_at": 1, + "model": "synthetic-model", + "output": [], + "parallel_tool_calls": False, + "tool_choice": "auto", + "tools": [], + "status": status, + "metadata": {"stage": "created"}, + **fields, + } + + +def typed(model: type[BaseModel], tag: str, **fields: Any) -> ResponsesServerEvent: + # Construct real generated WS models with validation, not model_construct. + return cast( + ResponsesServerEvent, + model(type=tag, sequence_number=next(_SEQUENCE), stream_id=LANE, **fields), + ) + + +def item(kind: str, **fields: Any) -> dict[str, Any]: + return {"id": ITEM_ID, "type": kind, "future_item": {"nested": [1], "optional": None}, **fields} + + +def reasoning(**fields: Any) -> dict[str, Any]: + return item("reasoning", summary=[{"type": "summary_text", "text": "seed"}], **fields) + + +def interpreter() -> dict[str, Any]: + return item("code_interpreter_call", container_id="synthetic_container", status="in_progress", code="seed") + + +def shell(commands: list[str]) -> dict[str, Any]: + return item( + "shell_call", + call_id="synthetic_call", + status="in_progress", + action={"commands": commands, "timeout_ms": 100, "future_action": None}, + ) + + +def shell_output() -> dict[str, Any]: + return item( + "shell_call_output", + call_id="synthetic_call", + status="in_progress", + output=[ + { + "stdout": "seed-out", + "stderr": "seed-err", + "outcome": {"type": "exit", "exit_code": 0}, + "future_output": {"keep": True}, + } + ], + ) + + +def seeded(source: dict[str, Any]) -> ResponsesWebSocketAccumulator: + acc = ResponsesWebSocketAccumulator() + acc.add_event(typed(events.ResponseWsCreated, "response.created", response=response())) + acc.add_event( + typed(events.ResponseOutputItemWsAdded, "response.output_item.added", output_index=INDEX, item=source) + ) + return acc + + +def row(acc: ResponsesWebSocketAccumulator, output_index: int = INDEX) -> dict[str, Any]: + return next( + cast("dict[str, Any]", item) + for item in acc.detailed_snapshot()["output"] + if item["output_index"] == output_index + ) + + +def value(data: Any, path: tuple[str | int, ...]) -> Any: + try: + for key in path: + data = data[key] + except (KeyError, IndexError, TypeError): + return ("absent", path) + return data + + +def rich_cases() -> list[Any]: + cases = [ + ( + events.ResponseReasoningSummaryPartWsAdded, + "response.reasoning_summary_part.added", + item("reasoning", summary=[]), + {"summary_index": 0, "part": {"type": "summary_text", "text": "added", "future_part": None}}, + ("item", "summary", 0, "text"), + "added", + ), + ( + events.ResponseReasoningSummaryPartWsDone, + "response.reasoning_summary_part.done", + reasoning(), + {"summary_index": 0, "part": {"type": "summary_text", "text": "done", "future_part": "done"}}, + ("item", "summary", 0, "text"), + "done", + ), + ( + events.ResponseReasoningSummaryTextWsDelta, + "response.reasoning_summary_text.delta", + reasoning(), + {"summary_index": 0, "delta": "+delta"}, + ("item", "summary", 0, "text"), + "seed+delta", + ), + ( + events.ResponseReasoningSummaryTextWsDone, + "response.reasoning_summary_text.done", + reasoning(), + {"summary_index": 0, "text": "done"}, + ("item", "summary", 0, "text"), + "done", + ), + ( + events.ResponseReasoningTextWsDelta, + "response.reasoning_text.delta", + reasoning(content=[{"type": "reasoning_text", "text": "seed"}]), + {"content_index": 0, "delta": "+delta"}, + ("content", 0, "part", "text"), + "seed+delta", + ), + ( + events.ResponseReasoningTextWsDone, + "response.reasoning_text.done", + reasoning(content=[{"type": "reasoning_text", "text": "seed"}]), + {"content_index": 0, "text": "done"}, + ("content", 0, "part", "text"), + "done", + ), + ( + events.ResponseShellCallCommandWsAdded, + "response.shell_call_command.added", + shell([]), + {"command_index": 0, "command": "added"}, + ("item", "action", "commands"), + ["added"], + ), + ( + events.ResponseShellCallCommandWsDelta, + "response.shell_call_command.delta", + shell(["seed"]), + {"command_index": 0, "delta": "+delta"}, + ("item", "action", "commands"), + ["seed+delta"], + ), + ( + events.ResponseShellCallCommandWsDone, + "response.shell_call_command.done", + shell(["seed"]), + {"command_index": 0, "command": "done"}, + ("item", "action", "commands"), + ["done"], + ), + ( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + shell_output(), + {"command_index": 0, "delta": {"stdout": "+out", "stderr": "+err"}}, + ("shell_output", 0, "data"), + {"stdout": "+out", "stderr": "+err"}, + ), + ( + events.ResponseShellCallOutputContentWsDone, + "response.shell_call_output_content.done", + shell_output(), + { + "command_index": 0, + "output": [ + { + "stdout": "done-out", + "stderr": "done-err", + "outcome": {"type": "timeout"}, + "future_output": {"done": True}, + } + ], + }, + ("shell_output", 0, "data", "output"), + [ + { + "stdout": "done-out", + "stderr": "done-err", + "outcome": {"type": "timeout"}, + "future_output": {"done": True}, + } + ], + ), + ( + events.ResponseCodeInterpreterCallCodeWsDelta, + "response.code_interpreter_call_code.delta", + interpreter(), + {"delta": "+delta"}, + ("item", "code"), + "seed+delta", + ), + ( + events.ResponseCodeInterpreterCallCodeWsDone, + "response.code_interpreter_call_code.done", + interpreter(), + {"code": "done"}, + ("item", "code"), + "done", + ), + ] + status_families = [ + ( + interpreter(), + [ + (events.ResponseCodeInterpreterCallInWsProgress, "in_progress"), + (events.ResponseCodeInterpreterCallWsInterpreting, "interpreting"), + (events.ResponseCodeInterpreterCallWsCompleted, "completed"), + ], + "code_interpreter_call", + ), + ( + item("file_search_call", queries=["synthetic"], status="in_progress"), + [ + (events.ResponseFileSearchCallInWsProgress, "in_progress"), + (events.ResponseFileSearchCallWsSearching, "searching"), + (events.ResponseFileSearchCallWsCompleted, "completed"), + ], + "file_search_call", + ), + ( + item("web_search_call", action={"type": "search", "query": "synthetic"}, status="in_progress"), + [ + (events.ResponseWebSearchCallInWsProgress, "in_progress"), + (events.ResponseWebSearchCallWsSearching, "searching"), + (events.ResponseWebSearchCallWsCompleted, "completed"), + ], + "web_search_call", + ), + ( + item("image_generation_call", status="in_progress"), + [ + (events.ResponseImageGenCallInWsProgress, "in_progress"), + (events.ResponseImageGenCallWsGenerating, "generating"), + (events.ResponseImageGenCallWsCompleted, "completed"), + ], + "image_generation_call", + ), + ( + item("mcp_call", name="synthetic", server_label="fixture", arguments="{}"), + [ + (events.ResponseMcpCallInWsProgress, "in_progress"), + (events.ResponseMcpCallWsCompleted, "completed"), + (events.ResponseMcpCallWsFailed, "failed"), + ], + "mcp_call", + ), + ] + for source, statuses, family in status_families: + for model, status in statuses: + cases.append((model, f"response.{family}.{status}", source, {}, ("item", "status"), status)) + return [pytest.param(*case, id=case[1]) for case in cases] + + +@pytest.mark.parametrize("model,tag,source,fields,path,expected", rich_cases()) +def test_rich_progress_through_public_accumulator( + model: type[BaseModel], + tag: str, + source: dict[str, Any], + fields: dict[str, Any], + path: tuple[str | int, ...], + expected: Any, +) -> None: + acc = seeded(source) + old_details = acc.detailed_snapshot() + old_compact = acc.snapshot() + routing = {} if tag.startswith("response.shell_call_command.") else {"item_id": ITEM_ID} + event = typed(model, tag, output_index=INDEX, **routing, **fields) + original = event.to_dict() + if tag.startswith("response.shell_call_command."): + assert "item_id" not in original + acc.add_event(event) + current = row(acc) + assert value(current, path) == expected + assert current["output_index"] == INDEX + assert current["item"]["future_item"] == {"nested": [1], "optional": None} + assert acc.snapshot() == old_compact + assert acc.snapshot().output_text == "" + assert event.to_dict() == original + current["item"]["future_item"]["nested"].append(2) + assert row(acc)["item"]["future_item"]["nested"] == [1] + assert old_details["output"][0]["item"]["future_item"]["nested"] == [1] + + +@pytest.mark.parametrize("phase", ["added", "done"]) +def test_reasoning_content_parts_retain_other_indexed_parts(phase: str) -> None: + parts = [{"type": "reasoning_text", "text": "keep", "future_part": {"keep": True}}] + if phase == "done": + parts.append({"type": "reasoning_text", "text": "draft"}) + acc = seeded(reasoning(content=parts)) + model = events.ResponseContentPartWsAdded if phase == "added" else events.ResponseContentPartWsDone + event = typed( + model, + f"response.content_part.{phase}", + output_index=INDEX, + content_index=1, + item_id=ITEM_ID, + part={"type": "reasoning_text", "text": phase}, + ) + original = event.to_dict() + acc.add_event(event) + assert row(acc)["content"] == [ + {"content_index": 0, "part": parts[0]}, + {"content_index": 1, "part": {"type": "reasoning_text", "text": phase}}, + ] + assert acc.snapshot().output_text == "" + assert event.to_dict() == original + + +@pytest.mark.parametrize("initial_created", [False, True]) +def test_queued_lifecycle_refreshes_observed_metadata(initial_created: bool) -> None: + acc = ResponsesWebSocketAccumulator() + if initial_created: + acc.add_event(typed(events.ResponseWsCreated, "response.created", response=response("queued"))) + event = typed( + events.ResponseWsQueued, + "response.queued", + response=response("queued", metadata={"stage": "queued"}, future_response={"observed": None}), + ) + original = event.to_dict() + acc.add_event(event) + details = acc.detailed_snapshot() + assert details["response"] is not None + assert details["response"]["metadata"] == {"stage": "queued"} + assert details["response"]["future_response"] == {"observed": None} + assert acc.snapshot().response_id == "rich_response" + assert event.to_dict() == original + with pytest.raises(RuntimeError, match="No terminal"): + acc.get_final_response() + + +def test_normal_reasoning_parts_and_text_share_one_detached_projection() -> None: + acc = seeded(reasoning(content=[])) + summary_delta = typed( + events.ResponseReasoningSummaryTextWsDelta, + "response.reasoning_summary_text.delta", + output_index=INDEX, + item_id=ITEM_ID, + summary_index=0, + delta="+keep", + ) + added = typed( + events.ResponseContentPartWsAdded, + "response.content_part.added", + output_index=INDEX, + item_id=ITEM_ID, + content_index=0, + part={"type": "reasoning_text", "text": "part-seed", "future_part": {"keep": True}}, + ) + delta = typed( + events.ResponseReasoningTextWsDelta, + "response.reasoning_text.delta", + output_index=INDEX, + item_id=ITEM_ID, + content_index=0, + delta="+delta", + ) + done = typed( + events.ResponseContentPartWsDone, + "response.content_part.done", + output_index=INDEX, + item_id=ITEM_ID, + content_index=0, + part={"type": "reasoning_text", "text": "part-done", "future_part": {"done": True}}, + ) + originals = [event.to_dict() for event in (summary_delta, added, delta, done)] + for event in (summary_delta, added, delta): + acc.add_event(event) + before_done = acc.detailed_snapshot() + assert before_done["output"][0]["content"][0]["part"]["text"] == "part-seed+delta" + assert before_done["output"][0]["item"]["content"][0]["text"] == "part-seed+delta" + acc.add_event(done) + current = row(acc) + expected_part = {"type": "reasoning_text", "text": "part-done", "future_part": {"done": True}} + assert current["content"] == [{"content_index": 0, "part": expected_part}] + assert current["item"]["content"] == [expected_part] + assert current["item"]["summary"][0]["text"] == "seed+keep" + assert before_done["output"][0]["content"][0]["part"]["text"] == "part-seed+delta" + assert acc.snapshot().output_text == "" + assert [event.to_dict() for event in (summary_delta, added, delta, done)] == originals + + +def test_shell_output_can_start_empty_without_fabricating_an_outcome() -> None: + source = shell_output() + source["output"] = [] + acc = seeded(source) + event = typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=INDEX, + item_id=ITEM_ID, + command_index=0, + delta={"stdout": "first"}, + ) + original = event.to_dict() + acc.add_event(event) + current = row(acc) + assert current["item"]["output"] == [] + assert current["shell_output"] == [{"command_index": 0, "data": {"stdout": "first"}}] + assert event.to_dict() == original + + +def test_shell_output_retains_a_later_command_first_as_sparse_progress() -> None: + acc = seeded(shell(["first", "second", "third"])) + source = shell_output() + source["id"] = "shell_output" + source["output"] = [] + output_index = INDEX + 1 + acc.add_event( + typed(events.ResponseOutputItemWsAdded, "response.output_item.added", output_index=output_index, item=source) + ) + for command_index, delta in [(2, {"stdout": "late"}), (0, {"stdout": "first"})]: + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=output_index, + item_id="shell_output", + command_index=command_index, + delta=delta, + ) + ) + before_done = acc.detailed_snapshot() + current = row(acc, output_index) + assert current["item"]["output"] == [] + assert current["shell_output"] == [ + {"command_index": 0, "data": {"stdout": "first"}}, + {"command_index": 2, "data": {"stdout": "late"}}, + ] + final_output = { + "stdout": "done", + "stderr": "done-error", + "outcome": {"type": "exit", "exit_code": 7}, + "future_output": None, + } + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDone, + "response.shell_call_output_content.done", + output_index=output_index, + item_id="shell_output", + command_index=2, + output=[final_output], + ) + ) + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=output_index, + item_id="shell_output", + command_index=1, + delta={"stderr": "middle"}, + ) + ) + assert row(acc, output_index)["item"]["output"] == [] + assert row(acc, output_index)["shell_output"] == [ + {"command_index": 0, "data": {"stdout": "first"}}, + {"command_index": 1, "data": {"stderr": "middle"}}, + {"command_index": 2, "data": {"output": [final_output]}}, + ] + assert before_done["output"][1]["shell_output"][1]["data"] == {"stdout": "late"} + + +def test_summary_done_full_item_and_terminal_have_separate_authority() -> None: + acc = ResponsesWebSocketAccumulator() + acc.add_event(typed(events.ResponseWsCreated, "response.created", response=response())) + scaffold = typed( + events.ResponseOutputItemWsAdded, "response.output_item.added", output_index=INDEX, item=reasoning() + ) + original_scaffold = scaffold.to_dict() + acc.add_event(scaffold) + delta = typed( + events.ResponseReasoningSummaryTextWsDelta, + "response.reasoning_summary_text.delta", + output_index=INDEX, + item_id=ITEM_ID, + summary_index=0, + delta="+delta", + ) + acc.add_event(delta) + after_delta = acc.detailed_snapshot() + assert row(acc)["item"]["summary"][0]["text"] == "seed+delta" + acc.add_event( + typed( + events.ResponseReasoningSummaryTextWsDone, + "response.reasoning_summary_text.done", + output_index=INDEX, + item_id=ITEM_ID, + summary_index=0, + text="done", + ) + ) + after_done = acc.detailed_snapshot() + assert row(acc)["item"]["summary"][0]["text"] == "done" + full = reasoning() + full["summary"][0]["text"] = "full-item" + full["future_item"] = {"authority": "full-item"} + acc.add_event(typed(events.ResponseOutputItemWsDone, "response.output_item.done", output_index=INDEX, item=full)) + assert row(acc)["item"]["summary"][0]["text"] == "full-item" + terminal_item = deepcopy(full) + terminal_item["summary"][0]["text"] = "terminal" + terminal = typed( + events.ResponseWsCompleted, "response.completed", response=response("completed", output=[terminal_item]) + ) + original_terminal = terminal.to_dict() + acc.add_event(terminal) + final_item = acc.get_final_response().output[0] + assert final_item.type == "reasoning" + assert final_item.summary[0].text == "terminal" + assert after_delta["output"][0]["item"]["summary"][0]["text"] == "seed+delta" + assert after_done["output"][0]["item"]["summary"][0]["text"] == "done" + assert terminal.to_dict() == original_terminal + assert scaffold.to_dict() == original_scaffold + returned = acc.get_final_response() + returned.output.clear() + assert len(acc.get_final_response().output) == 1 + + +def test_status_and_other_part_replacements_preserve_text_fragments() -> None: + acc = seeded(interpreter()) + acc.add_event( + typed( + events.ResponseCodeInterpreterCallCodeWsDelta, + "response.code_interpreter_call_code.delta", + output_index=INDEX, + item_id=ITEM_ID, + delta="+before", + ) + ) + acc.add_event( + typed( + events.ResponseCodeInterpreterCallWsInterpreting, + "response.code_interpreter_call.interpreting", + output_index=INDEX, + item_id=ITEM_ID, + ) + ) + acc.add_event( + typed( + events.ResponseCodeInterpreterCallCodeWsDelta, + "response.code_interpreter_call_code.delta", + output_index=INDEX, + item_id=ITEM_ID, + delta="+after", + ) + ) + assert row(acc)["item"]["code"] == "seed+before+after" + assert row(acc)["item"]["status"] == "interpreting" + + source = reasoning() + source["summary"].append({"type": "summary_text", "text": "other"}) + acc = seeded(source) + acc.add_event( + typed( + events.ResponseReasoningSummaryTextWsDelta, + "response.reasoning_summary_text.delta", + output_index=INDEX, + item_id=ITEM_ID, + summary_index=0, + delta="+keep", + ) + ) + acc.add_event( + typed( + events.ResponseReasoningSummaryPartWsDone, + "response.reasoning_summary_part.done", + output_index=INDEX, + item_id=ITEM_ID, + summary_index=1, + part={"type": "summary_text", "text": "replacement"}, + ) + ) + assert [part["text"] for part in row(acc)["item"]["summary"]] == ["seed+keep", "replacement"] + + +def test_shell_command_replacement_preserves_another_commands_fragments() -> None: + acc = seeded(shell(["seed"])) + acc.add_event( + typed( + events.ResponseShellCallCommandWsDelta, + "response.shell_call_command.delta", + output_index=INDEX, + command_index=0, + delta="+keep", + ) + ) + acc.add_event( + typed( + events.ResponseShellCallCommandWsAdded, + "response.shell_call_command.added", + output_index=INDEX, + command_index=1, + command="second", + ) + ) + before_done = acc.detailed_snapshot() + acc.add_event( + typed( + events.ResponseShellCallCommandWsDelta, + "response.shell_call_command.delta", + output_index=INDEX, + command_index=1, + delta="+delta", + ) + ) + acc.add_event( + typed( + events.ResponseShellCallCommandWsDone, + "response.shell_call_command.done", + output_index=INDEX, + command_index=1, + command="done", + ) + ) + assert row(acc)["item"]["action"]["commands"] == ["seed+keep", "done"] + assert before_done["output"][0]["item"]["action"]["commands"] == ["seed+keep", "second"] + + +@pytest.mark.parametrize("extra_item_id", [42, "unrelated_item"]) +@pytest.mark.parametrize( + "model,tag,fields,expected", + [ + ( + events.ResponseShellCallCommandWsAdded, + "response.shell_call_command.added", + {"command_index": 1, "command": "added"}, + ["seed", "added"], + ), + ( + events.ResponseShellCallCommandWsDelta, + "response.shell_call_command.delta", + {"command_index": 0, "delta": "+delta"}, + ["seed+delta"], + ), + ( + events.ResponseShellCallCommandWsDone, + "response.shell_call_command.done", + {"command_index": 0, "command": "done"}, + ["done"], + ), + ], +) +def test_shell_commands_preserve_but_do_not_route_by_unmodeled_item_id( + model: type[BaseModel], tag: str, fields: dict[str, Any], expected: list[str], extra_item_id: Any +) -> None: + acc = seeded(shell(["seed"])) + event = typed(model, tag, output_index=INDEX, item_id=extra_item_id, future_event={"keep": [1]}, **fields) + original = event.to_dict() + acc.add_event(event) + assert row(acc)["item"]["action"]["commands"] == expected + assert event.to_dict() == original + assert original["item_id"] == extra_item_id + + +@pytest.mark.parametrize("target", ["absent", "wrong_kind"]) +def test_shell_commands_require_an_observed_shell_call_at_their_index(target: str) -> None: + acc = seeded(item("message", role="assistant", status="completed", content=[], action={"commands": ["owned"]})) + before = acc.detailed_snapshot() + event = typed( + events.ResponseShellCallCommandWsDelta, + "response.shell_call_command.delta", + output_index=INDEX + 1 if target == "absent" else INDEX, + command_index=0, + delta="must-not-apply", + ) + acc.add_event(event) + assert acc.detailed_snapshot() == before + + +def test_shell_bounds_refresh_after_late_call_growth_and_replacement() -> None: + source = shell_output() + source["output"] = [] + acc = seeded(source) + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=INDEX, + item_id=ITEM_ID, + command_index=0, + delta={"stdout": "first"}, + ) + ) + call_index = INDEX + 1 + call = shell(["first", "second"]) + call["id"] = "call_item" + acc.add_event( + typed(events.ResponseOutputItemWsAdded, "response.output_item.added", output_index=call_index, item=call) + ) + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=INDEX, + item_id=ITEM_ID, + command_index=1, + delta={"stdout": "second"}, + ) + ) + acc.add_event( + typed( + events.ResponseShellCallCommandWsAdded, + "response.shell_call_command.added", + output_index=call_index, + command_index=2, + command="third", + ) + ) + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=INDEX, + item_id=ITEM_ID, + command_index=2, + delta={"stdout": "third"}, + ) + ) + assert [part["data"]["stdout"] for part in row(acc)["shell_output"]] == ["first", "second", "third"] + + call["action"]["commands"] = ["only"] + acc.add_event( + typed(events.ResponseOutputItemWsDone, "response.output_item.done", output_index=call_index, item=call) + ) + before = acc.detailed_snapshot() + with pytest.raises(ValueError): + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=INDEX, + item_id=ITEM_ID, + command_index=1, + delta={"stdout": "must-not-apply"}, + ) + ) + assert acc.detailed_snapshot() == before + + replacement = item("message", role="assistant", status="completed", content=[]) + replacement["id"] = "message_item" + acc.add_event( + typed(events.ResponseOutputItemWsDone, "response.output_item.done", output_index=call_index, item=replacement) + ) + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=INDEX, + item_id=ITEM_ID, + command_index=1, + delta={"stdout": "+after-removal"}, + ) + ) + assert row(acc)["shell_output"][1]["data"]["stdout"] == "second+after-removal" + + +@pytest.mark.parametrize("with_call", [False, True]) +def test_many_shell_fragments_keep_snapshots_and_new_lifecycle_bounds(with_call: bool) -> None: + acc = ResponsesWebSocketAccumulator() + acc.add_event(typed(events.ResponseWsCreated, "response.created", response=response())) + source = shell_output() + source["id"] = "output_item" + source["output"] = [] + output_index = INDEX + 1 + for index in range(64): + unrelated = item("message", role="assistant", status="completed", content=[]) + unrelated["id"] = f"unrelated_{index}" + acc.add_event( + typed( + events.ResponseOutputItemWsAdded, + "response.output_item.added", + output_index=INDEX + 2 + index, + item=unrelated, + ) + ) + if with_call: + acc.add_event( + typed( + events.ResponseOutputItemWsAdded, + "response.output_item.added", + output_index=INDEX, + item=shell(["first", "second"]), + ) + ) + acc.add_event( + typed(events.ResponseOutputItemWsAdded, "response.output_item.added", output_index=output_index, item=source) + ) + command_index = 1 if with_call else 0 + for _ in range(2048): + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=output_index, + item_id="output_item", + command_index=command_index, + delta={"stdout": "x"}, + ) + ) + prior = acc.detailed_snapshot() + assert row(acc, output_index)["shell_output"] == [{"command_index": command_index, "data": {"stdout": "x" * 2048}}] + new_call = shell(["only"]) + new_call["id"] = "new_call" + acc.add_event( + typed(events.ResponseInWsProgress, "response.in_progress", response=response(output=[new_call, source])) + ) + before = acc.detailed_snapshot() + with pytest.raises(ValueError): + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=1, + item_id="output_item", + command_index=1, + delta={"stdout": "must-not-apply"}, + ) + ) + assert acc.detailed_snapshot() == before + assert prior["output"][1 if with_call else 0]["shell_output"][0]["data"]["stdout"] == "x" * 2048 + + +def test_many_distinct_shell_outputs_keep_their_bounds_across_unrelated_call_insertions() -> None: + count = 64 + outputs: list[dict[str, Any]] = [] + calls: list[dict[str, Any]] = [] + for index in range(count): + output = shell_output() + output.update(id=f"output_{index}", call_id=f"call_{index}", output=[]) + outputs.append(output) + call = shell(["command"] * (index % 3 + 1)) + call.update(id=f"call_item_{index}", call_id=f"call_{index}") + calls.append(call) + acc = ResponsesWebSocketAccumulator() + acc.add_event(typed(events.ResponseWsCreated, "response.created", response=response(parallel_tool_calls=True))) + lifecycle = typed( + events.ResponseInWsProgress, + "response.in_progress", + response=response(parallel_tool_calls=True, output=outputs + calls), + ) + original = lifecycle.to_dict() + acc.add_event(lifecycle) + for index in range(count): + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=index, + item_id=f"output_{index}", + command_index=index % 3, + delta={"stdout": f"first_{index}"}, + ) + ) + prior = acc.detailed_snapshot() + for index in range(count): + unrelated = shell(["unrelated"]) + unrelated.update(id=f"unrelated_item_{index}", call_id=f"unrelated_call_{index}") + acc.add_event( + typed( + events.ResponseOutputItemWsAdded, + "response.output_item.added", + output_index=2 * count + index, + item=unrelated, + ) + ) + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=index, + item_id=f"output_{index}", + command_index=index % 3, + delta={"stdout": "+after-unrelated-call"}, + ) + ) + details = acc.detailed_snapshot() + for index, output in enumerate(details["output"][:count]): + assert output["item"]["output"] == [] + assert output["shell_output"] == [ + {"command_index": index % 3, "data": {"stdout": f"first_{index}+after-unrelated-call"}} + ] + assert prior["output"][index]["shell_output"][0]["data"] == {"stdout": f"first_{index}"} + for index in range(3): + with pytest.raises(ValueError, match="missing its command"): + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=index, + item_id=f"output_{index}", + command_index=index % 3 + 1, + delta={"stdout": "must-not-apply"}, + ) + ) + assert acc.detailed_snapshot() == details + assert lifecycle.to_dict() == original + + +@pytest.mark.parametrize("clear", ["lifecycle", "reset"]) +def test_shell_bounds_do_not_survive_removal_from_lifecycle_or_reset(clear: str) -> None: + output = shell_output() + output["output"] = [] + acc = seeded(output) + call = shell([]) + call["id"] = "empty_call" + acc.add_event( + typed(events.ResponseOutputItemWsAdded, "response.output_item.added", output_index=INDEX + 1, item=call) + ) + with pytest.raises(ValueError, match="missing its command"): + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=INDEX, + item_id=ITEM_ID, + command_index=0, + delta={"stdout": "must-not-apply"}, + ) + ) + if clear == "reset": + acc.reset() + acc.add_event(typed(events.ResponseWsCreated, "response.created", response=response(output=[output]))) + else: + acc.add_event(typed(events.ResponseInWsProgress, "response.in_progress", response=response(output=[output]))) + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=0, + item_id=ITEM_ID, + command_index=5, + delta={"stdout": "unknown-bound"}, + ) + ) + assert row(acc, 0)["shell_output"] == [{"command_index": 5, "data": {"stdout": "unknown-bound"}}] + + +@pytest.mark.parametrize("presence", ["absent", "null", "present"]) +@pytest.mark.parametrize("family", ["function_call", "custom_tool_call"]) +def test_full_tool_item_id_presence_is_separate_from_observed_delta_identity(family: str, presence: str) -> None: + field = "arguments" if family == "function_call" else "input" + seed = "{" if family == "function_call" else "seed" + fragment = '"observed":true}' if family == "function_call" else "+delta" + completed = '{"replacement":true}' if family == "function_call" else "replacement" + source = item(family, name="synthetic", call_id="synthetic_call", status="in_progress", **{field: seed}) + if presence == "absent": + del source["id"] + elif presence == "null": + source["id"] = None + acc = ResponsesWebSocketAccumulator() + acc.add_event(typed(events.ResponseWsCreated, "response.created", response=response())) + scaffold = typed(events.ResponseOutputItemWsAdded, "response.output_item.added", output_index=INDEX, item=source) + original_scaffold = scaffold.to_dict() + acc.add_event(scaffold) + before_delta = acc.detailed_snapshot() + model = ( + events.ResponseFunctionCallArgumentsWsDelta + if family == "function_call" + else events.ResponseCustomToolCallInputWsDelta + ) + tag = ( + "response.function_call_arguments.delta" + if family == "function_call" + else "response.custom_tool_call_input.delta" + ) + delta = typed(model, tag, output_index=INDEX, item_id=ITEM_ID, delta=fragment) + original_delta = delta.to_dict() + acc.add_event(delta) + current = row(acc) + assert ("id" in current["item"]) == ("id" in source) + if "id" in source: + assert current["item"]["id"] == source["id"] + assert current["item"][field] == seed + fragment + assert acc.snapshot().output[0].item_id == ITEM_ID + assert before_delta["output"][0]["item"] == source + assert scaffold.to_dict() == original_scaffold + assert delta.to_dict() == original_delta + + before_replacement = acc.detailed_snapshot() + replacement = deepcopy(source) + replacement.update(status="completed", future_item={"authority": "full-item"}, **{field: completed}) + full = typed(events.ResponseOutputItemWsDone, "response.output_item.done", output_index=INDEX, item=replacement) + original_full = full.to_dict() + acc.add_event(full) + assert row(acc)["item"] == replacement + assert before_replacement["output"][0]["item"] == current["item"] + assert before_delta["output"][0]["item"] == source + assert full.to_dict() == original_full + assert scaffold.to_dict() == original_scaffold + assert delta.to_dict() == original_delta + + +@pytest.mark.parametrize("scaffold", ["absent", "null", "present", "bootstrap"]) +@pytest.mark.parametrize("phase", ["delta", "done"]) +@pytest.mark.parametrize("family", ["function_call", "custom_tool_call"]) +def test_omitted_id_full_tools_keep_bound_identity_and_legacy_replacement_cases( + family: str, phase: str, scaffold: str +) -> None: + field = "arguments" if family == "function_call" else "input" + prefix = "response.function_call_arguments" if family == "function_call" else "response.custom_tool_call_input" + delta_model = ( + events.ResponseFunctionCallArgumentsWsDelta + if family == "function_call" + else events.ResponseCustomToolCallInputWsDelta + ) + done_model = ( + events.ResponseFunctionCallArgumentsWsDone + if family == "function_call" + else events.ResponseCustomToolCallInputWsDone + ) + acc = ResponsesWebSocketAccumulator() + acc.add_event(typed(events.ResponseWsCreated, "response.created", response=response())) + if scaffold != "bootstrap": + source = item(family, name="synthetic", call_id="synthetic_call", status="in_progress", **{field: ""}) + if scaffold == "absent": + del source["id"] + elif scaffold == "null": + source["id"] = None + acc.add_event( + typed(events.ResponseOutputItemWsAdded, "response.output_item.added", output_index=INDEX, item=source) + ) + first_text = '{"first":true}' if family == "function_call" else "first" + first = typed(delta_model, prefix + ".delta", output_index=INDEX, item_id=ITEM_ID, delta=first_text) + original_first = first.to_dict() + acc.add_event(first) + before = acc.detailed_snapshot() + compact_before = acc.snapshot() + assert compact_before.output[0].item_id == ITEM_ID + second_text = '{"second":true}' if family == "function_call" else "second" + fields = {"delta" if phase == "delta" else field: second_text} + conflict = typed( + delta_model if phase == "delta" else done_model, + prefix + "." + phase, + output_index=INDEX, + item_id="other_item", + **fields, + ) + original_conflict = conflict.to_dict() + acc.add_event(conflict) + assert first.to_dict() == original_first + assert conflict.to_dict() == original_conflict + if scaffold == "absent": + assert acc.detailed_snapshot() == before + assert acc.snapshot() == compact_before + else: + # Explicit-null/present IDs and unscaffolded inputs retain replacement policy. + assert acc.snapshot().output[0].item_id == "other_item" + assert acc.snapshot().output[0].type is None + assert getattr(acc.snapshot().output[0], field) == second_text + assert before["output"][0]["item"][field] == first_text + + +@pytest.mark.parametrize("phase", ["delta", "done"]) +@pytest.mark.parametrize("family", ["function_call", "custom_tool_call"]) +def test_omitted_id_full_tool_ignores_other_progress_family_before_binding(family: str, phase: str) -> None: + field = "arguments" if family == "function_call" else "input" + source = item(family, name="synthetic", call_id="synthetic_call", status="in_progress", **{field: ""}) + del source["id"] + acc = seeded(source) + before = acc.detailed_snapshot() + compact_before = acc.snapshot() + if family == "function_call": + wrong_prefix = "response.custom_tool_call_input" + wrong_field = "input" + wrong_model = ( + events.ResponseCustomToolCallInputWsDelta if phase == "delta" else events.ResponseCustomToolCallInputWsDone + ) + matching_prefix = "response.function_call_arguments" + matching_model = events.ResponseFunctionCallArgumentsWsDelta + else: + wrong_prefix = "response.function_call_arguments" + wrong_field = "arguments" + wrong_model = ( + events.ResponseFunctionCallArgumentsWsDelta + if phase == "delta" + else events.ResponseFunctionCallArgumentsWsDone + ) + matching_prefix = "response.custom_tool_call_input" + matching_model = events.ResponseCustomToolCallInputWsDelta + wrong = typed( + wrong_model, + wrong_prefix + "." + phase, + output_index=INDEX, + item_id="wrong_family_item", + **{"delta" if phase == "delta" else wrong_field: '{"wrong":true}'}, + ) + original_wrong = wrong.to_dict() + acc.add_event(wrong) + assert acc.snapshot() == compact_before + assert acc.detailed_snapshot() == before + correct_text = '{"correct":true}' if family == "function_call" else "correct" + matching = typed( + matching_model, + matching_prefix + ".delta", + output_index=INDEX, + item_id=ITEM_ID, + delta=correct_text, + ) + original_matching = matching.to_dict() + acc.add_event(matching) + assert acc.snapshot().output[0].item_id == ITEM_ID + assert row(acc)["item"][field] == correct_text + assert "id" not in row(acc)["item"] + assert before["output"][0]["item"] == source + assert wrong.to_dict() == original_wrong + assert matching.to_dict() == original_matching + + +@pytest.mark.parametrize("code_presence", ["missing", "null"]) +def test_optional_code_seed_stays_observed_until_progress(code_presence: str) -> None: + source = interpreter() + if code_presence == "missing": + del source["code"] + else: + source["code"] = None + acc = seeded(source) + before = row(acc) + assert ("code" in before["item"]) is (code_presence == "null") + acc.add_event( + typed( + events.ResponseCodeInterpreterCallCodeWsDelta, + "response.code_interpreter_call_code.delta", + output_index=INDEX, + item_id=ITEM_ID, + delta="observed", + ) + ) + assert row(acc)["item"]["code"] == "observed" + acc.add_event(typed(events.ResponseOutputItemWsDone, "response.output_item.done", output_index=INDEX, item=source)) + assert row(acc)["item"].get("code") is None + assert ("code" in row(acc)["item"]) is (code_presence == "null") + + +def test_invalid_shell_stream_cannot_partially_extend_the_other_stream() -> None: + acc = seeded(shell_output()) + event = typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=INDEX, + item_id=ITEM_ID, + command_index=0, + delta={"stdout": "+valid", "stderr": "+valid"}, + ) + # The wire parser can retain known events without runtime validation. + malformed = model_copy(event, deep=True) + object.__setattr__(malformed, "delta", {"stdout": "+valid", "stderr": 42}) + before = acc.detailed_snapshot() + with pytest.raises(ValueError, match="stderr must be a string"): + acc.add_event(malformed) + assert acc.detailed_snapshot() == before + + +def test_negative_shell_command_index_cannot_mutate() -> None: + command_index = -1 + source = shell_output() + source["output"] = [] + acc = seeded(source) + before = acc.detailed_snapshot() + event = typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=INDEX, + item_id=ITEM_ID, + command_index=command_index, + delta={"stdout": "must-not-apply"}, + ) + with pytest.raises(ValueError): + acc.add_event(event) + assert acc.detailed_snapshot() == before + + +@pytest.mark.parametrize("empty", [False, True], ids=["two-chunks", "empty"]) +def test_shell_done_array_replaces_only_its_command_and_preserves_full_authority(empty: bool) -> None: + acc = seeded(shell(["first", "second", "third"])) + source = shell_output() + source["id"] = "output_item" + output_index = INDEX + 1 + acc.add_event( + typed(events.ResponseOutputItemWsAdded, "response.output_item.added", output_index=output_index, item=source) + ) + for command_index, text in [(2, "neighbor"), (0, "draft")]: + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=output_index, + item_id="output_item", + command_index=command_index, + delta={"stdout": text}, + ) + ) + before_done = acc.detailed_snapshot() + chunks = ( + [] + if empty + else [ + { + "stdout": "first-output", + "stderr": "first-error", + "outcome": {"type": "exit", "exit_code": 3}, + "created_by": "first-actor", + "future_output": {"keep": [False, None]}, + }, + { + "stdout": "second-output", + "stderr": "second-error", + "outcome": {"type": "timeout"}, + "created_by": "second-actor", + "future_output": None, + }, + ] + ) + done = typed( + events.ResponseShellCallOutputContentWsDone, + "response.shell_call_output_content.done", + output_index=output_index, + item_id="output_item", + command_index=0, + output=chunks, + ) + original = done.to_dict() + acc.add_event(done) + current = row(acc, output_index) + assert current["shell_output"] == [ + {"command_index": 0, "data": {"output": chunks}}, + {"command_index": 2, "data": {"stdout": "neighbor"}}, + ] + assert current["item"]["output"] == source["output"] + assert before_done["output"][1]["shell_output"][0]["data"] == {"stdout": "draft"} + assert done.to_dict() == original + current["shell_output"][0]["data"]["output"].append({"caller": "mutation"}) + assert row(acc, output_index)["shell_output"][0]["data"]["output"] == chunks + + full = deepcopy(source) + full["output"][0]["stdout"] = "whole-item-output" + acc.add_event( + typed(events.ResponseOutputItemWsDone, "response.output_item.done", output_index=output_index, item=full) + ) + assert row(acc, output_index)["item"]["output"] == full["output"] + assert row(acc, output_index)["shell_output"] == [] + terminal_item = deepcopy(full) + terminal_item["output"][0]["stdout"] = "terminal-output" + terminal = typed( + events.ResponseWsCompleted, "response.completed", response=response("completed", output=[terminal_item]) + ) + original_terminal = terminal.to_dict() + acc.add_event(terminal) + assert row(acc, 0)["item"]["output"] == terminal_item["output"] + assert row(acc, 0)["shell_output"] == [] + final_item = acc.get_final_response().output[0] + assert final_item.type == "shell_call_output" + assert final_item.output[0].stdout == "terminal-output" + assert terminal.to_dict() == original_terminal + assert done.to_dict() == original + assert before_done["output"][1]["shell_output"][0]["data"] == {"stdout": "draft"} + + +@pytest.mark.parametrize("command_index", [5, 2**65]) +def test_shell_without_companion_retains_sparse_observed_command_without_holes(command_index: int) -> None: + source = shell_output() + source["output"] = [] + acc = seeded(source) + event = typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=INDEX, + item_id=ITEM_ID, + command_index=command_index, + delta={"stdout": "sparse"}, + ) + original = event.to_dict() + acc.add_event(event) + assert row(acc)["item"]["output"] == [] + assert row(acc)["shell_output"] == [{"command_index": command_index, "data": {"stdout": "sparse"}}] + assert event.to_dict() == original + + +def test_shell_flat_array_does_not_establish_command_grouping_or_seed_progress() -> None: + source = shell_output() + source["output"].append(deepcopy(source["output"][0])) + source["output"][1]["stdout"] = "another-whole-item-row" + acc = seeded(source) + acc.add_event( + typed( + events.ResponseShellCallOutputContentWsDelta, + "response.shell_call_output_content.delta", + output_index=INDEX, + item_id=ITEM_ID, + command_index=1, + delta={"stdout": "observed"}, + ) + ) + assert row(acc)["item"]["output"] == source["output"] + assert row(acc)["shell_output"] == [{"command_index": 1, "data": {"stdout": "observed"}}] + + +@pytest.mark.parametrize("target", ["absent", "wrong_id", "wrong_kind", "retired_id"]) +def test_rich_progress_cannot_create_or_replace_an_unrelated_item(target: str) -> None: + acc = seeded(reasoning()) + item_id = ITEM_ID + if target == "absent": + output_index = INDEX + 1 + else: + output_index = INDEX + if target == "wrong_id": + item_id = "other_item" + elif target == "wrong_kind": + acc.add_event( + typed( + events.ResponseOutputItemWsDone, + "response.output_item.done", + output_index=INDEX, + item=item("message", role="assistant", status="completed", content=[]), + ) + ) + elif target == "retired_id": + replacement = reasoning() + replacement["id"] = "replacement_item" + acc.add_event( + typed(events.ResponseOutputItemWsDone, "response.output_item.done", output_index=INDEX, item=replacement) + ) + before = acc.detailed_snapshot() + acc.add_event( + typed( + events.ResponseReasoningSummaryTextWsDelta, + "response.reasoning_summary_text.delta", + output_index=output_index, + item_id=item_id, + summary_index=0, + delta="must-not-apply", + ) + ) + assert acc.detailed_snapshot() == before