diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 6c8a811dd4..7f366fda2d 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -1086,6 +1086,13 @@ def _consume_codex_event_stream( collected_output_items: List[Any] = [] collected_text_deltas: List[str] = [] has_tool_calls = False + # Function calls announced via output_item.added but not yet confirmed by + # output_item.done, keyed by item id. Some OpenAI-compatible backends omit + # per-item done events on a successful completion (upstream evidence: + # anomalyco/opencode#37159); these are settled from accumulated stream + # state at the terminal event so the tool call executes instead of being + # silently dropped. + pending_function_calls: Dict[str, Dict[str, Any]] = {} first_delta_fired = False active_message_phase: str | None = None commentary_text_deltas: List[str] = [] @@ -1143,6 +1150,12 @@ def _consume_codex_event_stream( active_message_phase = None if "function_call" in str(item_type): has_tool_calls = True + item_id = str(_item_field(item, "id", "")) + if item_id: + pending_function_calls[item_id] = { + "item": item, + "arguments": "", + } continue if "output_text.delta" in event_type or event_type == "response.output_text.delta": @@ -1181,7 +1194,24 @@ def _consume_codex_event_stream( if "function_call" in event_type: has_tool_calls = True - # fall through — function_call items still get added on output_item.done + # Accumulate streamed argument deltas for calls announced via + # output_item.added, so a stream that completes without per-item + # done events can still be settled from accumulated state. + if "delta" in event_type: + delta_args = _event_field(event, "delta", "") + pending = pending_function_calls.get(str(_event_field(event, "item_id", ""))) + if pending is not None and delta_args: + pending["arguments"] += delta_args + continue + if event_type.endswith("function_call_arguments.done"): + done_args = _event_field(event, "arguments", "") + pending = pending_function_calls.get(str(_event_field(event, "item_id", ""))) + if pending is not None and done_args: + # Per-item arguments.done is authoritative for the + # accumulated string when the item itself never lands. + pending["arguments"] = done_args + continue + # other function_call frames fall through — function_call items still get added on output_item.done if "reasoning" in event_type and "delta" in event_type: reasoning_text = _event_field(event, "delta", "") @@ -1207,6 +1237,9 @@ def _consume_codex_event_stream( done_item = _event_field(event, "item") if done_item is not None: collected_output_items.append(done_item) + # Confirmed by the authoritative per-item done event; remove + # from pending so it is not settled twice. + pending_function_calls.pop(str(_item_field(done_item, "id", "")), None) done_phase = _item_field(done_item, "phase", None) done_phase = done_phase.strip().lower() if isinstance(done_phase, str) else None if done_phase == "commentary" and on_commentary_message is not None: @@ -1279,6 +1312,23 @@ def _consume_codex_event_stream( else: output = [] + # Settle function calls that were announced via output_item.added and + # streamed argument deltas but never confirmed by output_item.done: some + # OpenAI-compatible backends omit per-item done events on a successful + # completion (anomalyco/opencode#37159). Done items stay authoritative; + # this only fills the gap so the call executes instead of vanishing. + if pending_function_calls and terminal_status == "completed": + for pending in pending_function_calls.values(): + item = pending["item"] + output.append(SimpleNamespace( + type="function_call", + id=_item_field(item, "id", None), + call_id=_item_field(item, "call_id", None), + name=_item_field(item, "name", None), + arguments=pending["arguments"], + status="completed", + )) + # If the stream ended without any terminal event AND produced no usable # content (no items, no text deltas), surface that as a RuntimeError so # callers can distinguish "stream truncated mid-flight / provider rejected diff --git a/tests/agent/test_codex_responses_settle_pending_tool_calls.py b/tests/agent/test_codex_responses_settle_pending_tool_calls.py new file mode 100644 index 0000000000..72616048cd --- /dev/null +++ b/tests/agent/test_codex_responses_settle_pending_tool_calls.py @@ -0,0 +1,113 @@ +"""Regression: settle pending function calls when a Responses stream +completes successfully without ``response.output_item.done``. + +Some OpenAI-compatible backends (see anomalyco/opencode#37159, fixed in +anomalyco/opencode#43575) omit per-item ``response.output_item.done`` +events on a successful completion. ``_consume_codex_event_stream`` assembles +its final ``output`` purely from ``.done`` items, so a function call that +was announced via ``response.output_item.added`` and streamed argument +deltas is silently dropped: the turn ends with an empty output and the tool +never executes. + +These tests pin the desired behavior (settle the pending call from the +accumulated stream state at the terminal event, mirroring the opencode fix +semantics). They fail against current ``main`` — that red state is the +reproduction for the linked issue. +""" + +from types import SimpleNamespace + +from agent.codex_runtime import _consume_codex_event_stream + + +def _stream_completed_without_done(): + """Successful Responses stream whose only function call never receives + ``response.output_item.done`` (backend omits it; terminal response + carries ``output=None`` so there is nothing to reconstruct from).""" + return [ + SimpleNamespace( + type="response.created", + response=SimpleNamespace(id="resp_1"), + ), + SimpleNamespace( + type="response.output_item.added", + output_index=0, + item=SimpleNamespace( + type="function_call", + id="fc_1", + call_id="call_1", + name="get_weather", + arguments="", + ), + ), + SimpleNamespace( + type="response.function_call_arguments.delta", + item_id="fc_1", + output_index=0, + delta='{"city"', + ), + SimpleNamespace( + type="response.function_call_arguments.delta", + item_id="fc_1", + output_index=0, + delta=': "SF"}', + ), + # NOTE: no response.output_item.done for fc_1. + SimpleNamespace( + type="response.completed", + response=SimpleNamespace( + id="resp_1", + status="completed", + usage=SimpleNamespace( + input_tokens=10, output_tokens=5, total_tokens=15 + ), + output=None, + ), + ), + ] + + +def test_completed_without_done_settles_pending_function_call(): + final = _consume_codex_event_stream(_stream_completed_without_done(), model="gpt-test") + + calls = [ + item + for item in final.output + if getattr(item, "type", "") == "function_call" + ] + assert calls, ( + "function_call announced via output_item.added (+ argument deltas) " + "was silently dropped on successful completion without " + "output_item.done; the tool never executes" + ) + settled = calls[0] + assert getattr(settled, "name", None) == "get_weather" + assert getattr(settled, "arguments", None) == '{"city": "SF"}' + assert final.status == "completed" + + +def test_control_stream_with_done_still_authoritative(): + """Sanity: the same stream WITH output_item.done must keep working + (the fix must not regress the normal path or override .done data).""" + events = _stream_completed_without_done() + done = SimpleNamespace( + type="response.output_item.done", + output_index=0, + item=SimpleNamespace( + type="function_call", + id="fc_1", + call_id="call_1", + name="get_weather", + arguments='{"city": "SF"}', + ), + ) + events.insert(-1, done) + final = _consume_codex_event_stream(events, model="gpt-test") + + calls = [ + item + for item in final.output + if getattr(item, "type", "") == "function_call" + ] + assert calls and calls[0].arguments == '{"city": "SF"}' + assert final.status == "completed"