ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

fix(codex): settle pending Responses tool calls when output_item.done is omitted

Browse Source 
Backends that omit per-item done events on a successful completion
(anomalyco/opencode#37159) caused an announced function call to be
silently dropped: the turn ended with output == [] and the tool never
executed. Track calls announced via output_item.added, accumulate
argument deltas, and settle still-pending calls from accumulated state
at a successful terminal event. output_item.done stays authoritative.
Mirrors anomalyco/opencode#43575.
This commit is contained in:
cxxCoolStar
2026-08-23 14:49:30 +08:00
committed by Teknium
parent 081cdd9911
commit 720344cfba
2 changed files with 164 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/codex_runtime.py
+51 -1
View File
@@ -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
tests/agent/test_codex_responses_settle_pending_tool_calls.py
+113
View File
@@ -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"