diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index cce81119d2..2dd1ce4773 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -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: diff --git a/tests/agent/test_codex_responses_settle_pending_tool_calls.py b/tests/agent/test_codex_responses_settle_pending_tool_calls.py index 205a76429e..c1f152b674 100644 --- a/tests/agent/test_codex_responses_settle_pending_tool_calls.py +++ b/tests/agent/test_codex_responses_settle_pending_tool_calls.py @@ -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. diff --git a/tests/agent/test_codex_stream_rotating_item_ids.py b/tests/agent/test_codex_stream_rotating_item_ids.py index fb9cdc2d30..605a954b69 100644 --- a/tests/agent/test_codex_stream_rotating_item_ids.py +++ b/tests/agent/test_codex_stream_rotating_item_ids.py @@ -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):