fix(codex): integrate pending call-id aliases with stream correlation

Port PR #94708 to _CodexResponseAssembler, preserving rotating argument IDs and output ordering. Coalesce all matching call_id aliases, inherit earliest announcement metadata, and protect unrelated calls from conflicting item/index identities.

Adapted from commit 70b311ee50a0350b79c6c2b6123754d6aa6505da (PR #94708).

Verification: 130 focused tests passed; independent review passed.

Co-authored-by: DragonLi-Mi <4321063+DragonLi-Mi@users.noreply.github.com>
This commit is contained in:
VictorYXL
2026-09-17 11:01:51 +08:00
committed by Teknium
parent 5e24de476d
commit c3302dd4fb
3 changed files with 114 additions and 11 deletions

View File

@@ -971,16 +971,33 @@ class _CodexResponseAssembler:
# Reuse the announced position when known (fresh tail sequence for unannounced items); the .done
# event's own output_index wins over the announced one.
done_id = str(_event_field(done_item, "id", ""))
pending_key = None
announced_sequence, announced_index = self.announced_output_order.get(done_id, (None, None))
pending_keys = []
if "function_call" in str(_event_field(done_item, "type", "")):
pending_key = self._pending_function_key(done_id, _event_field(event, "output_index"))
announced_sequence, announced_index = self.announced_output_order.get(pending_key or done_id, (None, None))
done_call_id = _event_field(done_item, "call_id", "")
if isinstance(done_call_id, str) and done_call_id.strip():
# A stable call_id is exclusive: conflicting item/index identities
# must not evict unrelated calls, even when no alias matches.
pending_keys = [key for key, pending in self.pending_function_calls.items()
if _event_field(pending["item"], "call_id") == done_call_id]
announced_sequence, announced_index = None, None
else:
# Without a stable call_id, retain item-id then index association,
# as used by argument events (which never carry a call_id).
pending_key = self._pending_function_key(done_id, _event_field(event, "output_index"))
if pending_key is not None:
pending_keys = [pending_key]
if pending_keys:
announced_alias = min((self.pending_function_calls[key] for key in pending_keys),
key=lambda pending: pending["sequence"])
announced_sequence, announced_index = announced_alias["sequence"], announced_alias["output_index"]
if announced_sequence is None:
announced_sequence, self.next_output_sequence = self.next_output_sequence, self.next_output_sequence + 1
self.output_indexes.append(_event_field(event, "output_index", announced_index))
self.output_sequences.append(announced_sequence)
# Confirmed by the authoritative done event; never settle it twice.
self.pending_function_calls.pop(pending_key or done_id, None)
# The done payload is authoritative for every pending alias of this call.
for pending_key in pending_keys:
self.pending_function_calls.pop(pending_key, None)
if _message_phase(done_item) == "commentary" and self.on_commentary_message is not None:
commentary_text = "".join(self.commentary_text_deltas).strip() or _output_text_of(done_item)
if commentary_text:

View File

@@ -120,6 +120,83 @@ def test_control_stream_with_done_still_authoritative():
assert final.status == "completed"
@pytest.mark.parametrize('done_id', ['fc_a_done', 'fc_a_alias_2'])
def test_done_call_coalesces_all_pending_aliases_in_announced_order(done_id):
"""PR #94708: done aliases inherit the earliest call_id announcement."""
events = []
for item_id, call_id in [
('fc_a', 'call_a'), ('fc_b', 'call_b'),
('fc_a_alias_1', 'call_a'), ('fc_a_alias_2', 'call_a'),
('fc_zero', 'call_zero'),
]:
events.append(SimpleNamespace(
type='response.output_item.added',
item=SimpleNamespace(type='function_call', id=item_id, call_id=call_id,
name='same_tool', arguments=''),
))
done_items = [
SimpleNamespace(type='function_call', id='fc_b_done', call_id='call_b',
name='same_tool', arguments='{"step": 2}'),
SimpleNamespace(type='function_call', id=done_id, call_id='call_a',
name='same_tool', arguments='{"step": 1}'),
]
events.extend(SimpleNamespace(type='response.output_item.done', item=item)
for item in done_items)
events.append(SimpleNamespace(type='response.completed',
response=SimpleNamespace(status='completed', output=None)))
final = _consume_codex_event_stream(events, model='gpt-test')
assert [(item.call_id, item.arguments) for item in final.output] == [
('call_a', '{"step": 1}'), ('call_b', '{"step": 2}'), ('call_zero', '{}'),
]
assert final.output[0] is done_items[1]
assert final.output[1] is done_items[0]
@pytest.mark.parametrize(('pending_fields', 'done_fields', 'done_index', 'confirms_a'), [
pytest.param({'call_id': 'call_a'}, {'id': 'fc_b', 'call_id': 'call_a'}, 1, True,
id='call_id_over_conflicting_item_and_index'),
pytest.param({'call_id': 'call_a'}, {'id': 'fc_a', 'call_id': 'call_new'}, 0, False,
id='unmatched_call_id_cannot_evict_by_item_or_index'),
*[pytest.param({'call_id': 'call_a'}, {**fields, 'call_id': 'call_a'}, 1, True,
id=f'{label}_item_id_uses_call_id')
for label, fields in [('missing', {}), ('null', {'id': None}),
('empty', {'id': ''}), ('blank', {'id': ' '})]],
*[pytest.param({'call_id': 'call_a'}, {'id': 'fc_a', **fields}, 1, True,
id=f'{label}_call_id_uses_item_before_index')
for label, fields in [('missing', {}), ('null', {'call_id': None}),
('empty', {'call_id': ''}), ('blank', {'call_id': ' '})]],
*[pytest.param(fields, {'id': 'fc_done', **fields}, None, False,
id=f'{label}_call_ids_are_not_shared_identity')
for label, fields in [('missing', {}), ('null', {'call_id': None}),
('empty', {'call_id': ''}), ('blank', {'call_id': ' '})]],
])
def test_done_identity_precedence_does_not_remove_unrelated_calls(
pending_fields, done_fields, done_index, confirms_a,
):
"""Nonblank call_id is exclusive; without one, item id precedes index."""
events = [SimpleNamespace(
type='response.output_item.added', output_index=index,
item=SimpleNamespace(type='function_call', id=f'fc_{name}', name=name,
arguments='', **fields),
) for index, (name, fields) in enumerate([
('a', pending_fields), ('b', {'call_id': 'call_b'}),
('zero', {'call_id': 'call_zero'}),
])]
done_item = SimpleNamespace(type='function_call', name='confirmed',
arguments='{"authoritative": true}', **done_fields)
events.extend([
SimpleNamespace(type='response.output_item.done', output_index=done_index, item=done_item),
SimpleNamespace(type='response.completed', response=SimpleNamespace(status='completed')),
])
final = _consume_codex_event_stream(events, model='gpt-test')
expected = [('b', '{}'), ('zero', '{}'), ('confirmed', done_item.arguments)]
if not confirms_a:
expected.append(('a', '{}'))
assert sorted((item.name, item.arguments) for item in final.output) == sorted(expected)
assert next(item for item in final.output if item.name == 'confirmed') is done_item
def test_no_terminal_frame_does_not_settle_pending_function_call():
"""EOF/interruption before any terminal frame must NOT settle the pending
call: unconfirmed stream state must not become executable authority.

View File

@@ -10,13 +10,15 @@ import pytest
from agent.codex_runtime import _consume_codex_event_stream
@pytest.mark.parametrize(('completion', 'done_order'), [
('item_done', (0, 1)),
pytest.param('item_done', (1, 0), id='item_done_reversed'),
('arguments_done', (0, 1)),
('deltas', (0, 1)),
@pytest.mark.parametrize(('completion', 'done_order', 'indexed'), [
('item_done', (0, 1), True),
pytest.param('item_done', (1, 0), True, id='item_done_reversed'),
('arguments_done', (0, 1), True),
('deltas', (0, 1), True),
pytest.param('item_done', (0, 1), False, id='call_id_without_indexes'),
pytest.param('item_done', (1, 0), False, id='call_id_without_indexes_reversed'),
])
def test_rotating_item_ids_preserve_distinct_calls_and_arguments(completion, done_order):
def test_rotating_item_ids_preserve_distinct_calls_and_arguments(completion, done_order, indexed):
events = []
for index in range(2):
events.append({
@@ -49,6 +51,13 @@ def test_rotating_item_ids_preserve_distinct_calls_and_arguments(completion, don
'call_id': f'call_{index}', 'name': 'diagnostic_echo',
'arguments': json.dumps({'text': f'authoritative-{index}'})},
})
if not indexed:
for event in events:
index = event.pop('output_index')
# Argument events have no call_id: without an index, keep their
# item_id stable. Only the completed item rotates its identity.
if 'item_id' in event:
event['item_id'] = f'announced_{index}'
events.append({'type': 'response.completed', 'response': {'status': 'completed'}})
response = _consume_codex_event_stream(events, model='diagnostic')
def field(item, name):