diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 420d12670e62..6c5a171aaa84 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -4496,21 +4496,41 @@ def run_conversation( # drifts per continuation even when the visible output # is identical, so including it in the comparison defeats # dedup and causes message storms (#52711). + last_interim_visible = ( + agent._interim_assistant_visible_text(last_msg) + if isinstance(last_msg, dict) + else "" + ) + current_interim_visible = agent._interim_assistant_visible_text(interim_msg) + if last_interim_visible or current_interim_visible: + same_visible_output = last_interim_visible == current_interim_visible + else: + # Preserve the existing reasoning-only behavior when + # neither response has text eligible for interim delivery. + same_visible_output = ( + (last_msg.get("content") or "") == (interim_msg.get("content") or "") + and (last_msg.get("reasoning") or "") == (interim_msg.get("reasoning") or "") + ) if isinstance(last_msg, dict) else False visible_duplicate = ( isinstance(last_msg, dict) and last_msg.get("role") == "assistant" and last_msg.get("finish_reason") == "incomplete" - and (last_msg.get("content") or "") == (interim_msg.get("content") or "") - and (last_msg.get("reasoning") or "") == (interim_msg.get("reasoning") or "") + and same_visible_output ) if visible_duplicate: - # Update opaque state in-place so the latest - # provider payload is preserved without emitting - # a duplicate visible message. - for _key in ("codex_reasoning_items", "codex_message_items"): - _new_val = interim_msg.get(_key) - if _new_val is not None: - last_msg[_key] = _new_val + # Update replay state in-place so the latest provider + # payload is preserved without re-emitting identical + # user-visible commentary. + for _key in ( + "content", + "reasoning", + "reasoning_content", + "reasoning_details", + "codex_reasoning_items", + "codex_message_items", + ): + if _key in interim_msg: + last_msg[_key] = interim_msg[_key] else: messages.append(interim_msg) agent._emit_interim_assistant_message(interim_msg) @@ -4844,8 +4864,23 @@ def run_conversation( # a LATER tool round. agent._post_tool_empty_retried = False + previous_msg = messages[-1] if messages else None + current_interim_visible = agent._interim_assistant_visible_text(assistant_msg) + previous_interim_visible = ( + agent._interim_assistant_visible_text(previous_msg) + if isinstance(previous_msg, dict) + else "" + ) + duplicate_previous_interim = ( + bool(current_interim_visible) + and isinstance(previous_msg, dict) + and previous_msg.get("role") == "assistant" + and previous_msg.get("finish_reason") == "incomplete" + and previous_interim_visible == current_interim_visible + ) messages.append(assistant_msg) - agent._emit_interim_assistant_message(assistant_msg) + if not duplicate_previous_interim: + agent._emit_interim_assistant_message(assistant_msg) try: # Persist the assistant tool-call turn before any tool # side effects run. If a destructive tool restarts or diff --git a/run_agent.py b/run_agent.py index 1b2a3b9cf140..6cc646603cc2 100644 --- a/run_agent.py +++ b/run_agent.py @@ -4794,15 +4794,26 @@ class AIAgent: visible = redact_sensitive_text(visible) return visible + def _interim_assistant_visible_text(self, assistant_msg: Dict[str, Any]) -> str: + """Return the exact assistant text eligible for interim delivery. + + Prefer structured Codex commentary over top-level content. A Codex + response can contain both commentary and a partial/final-answer message + while tools are still pending; treating top-level content as progress + in that shape leaks the answer before the tool call runs. + """ + visible = self._extract_codex_interim_visible_text(assistant_msg) + if visible: + return visible + content = assistant_msg.get("content") + return self._strip_think_blocks(content or "").strip() + def _emit_interim_assistant_message(self, assistant_msg: Dict[str, Any]) -> None: """Surface a real mid-turn assistant commentary message to the UI layer.""" cb = getattr(self, "interim_assistant_callback", None) if cb is None or not isinstance(assistant_msg, dict): return - content = assistant_msg.get("content") - visible = self._strip_think_blocks(content or "").strip() - if not visible or visible == "(empty)": - visible = self._extract_codex_interim_visible_text(assistant_msg) + visible = self._interim_assistant_visible_text(assistant_msg) if not visible or visible == "(empty)": return already_streamed = self._interim_content_was_streamed(visible) diff --git a/tests/run_agent/test_run_agent_codex_responses.py b/tests/run_agent/test_run_agent_codex_responses.py index ea6398142219..b963e50eef30 100644 --- a/tests/run_agent/test_run_agent_codex_responses.py +++ b/tests/run_agent/test_run_agent_codex_responses.py @@ -158,6 +158,35 @@ def _codex_commentary_message_response(text: str): ) +def _codex_commentary_final_tool_response(commentary: str, final_answer: str = "Done."): + return SimpleNamespace( + output=[ + SimpleNamespace( + type="message", + phase="commentary", + status="completed", + content=[SimpleNamespace(type="output_text", text=commentary)], + ), + SimpleNamespace( + type="message", + phase="final_answer", + status="completed", + content=[SimpleNamespace(type="output_text", text=final_answer)], + ), + SimpleNamespace( + type="function_call", + id="fc_1", + call_id="call_1", + name="terminal", + arguments="{}", + ), + ], + usage=SimpleNamespace(input_tokens=8, output_tokens=5, total_tokens=13), + status="completed", + model="gpt-5-codex", + ) + + def _codex_ack_message_response(text: str): return SimpleNamespace( output=[ @@ -2100,44 +2129,24 @@ def test_interim_commentary_preserves_assistant_content(monkeypatch): assert "I'll inspect the repo structure first." in observed["text"] -def test_interim_commentary_uses_codex_commentary_items_when_content_is_empty(monkeypatch): - """Codex Responses stores commentary phase text outside content. - - The gateway still needs that visible mid-turn narration to flow through the - interim assistant callback; otherwise Discord with streaming disabled gets - no natural progress commentary before tools run. Analysis/final-answer items - stay hidden from interim delivery. - """ +def test_interim_commentary_precedes_content_from_real_codex_normalization(monkeypatch): + """Structured commentary wins over final-answer content on tool turns.""" agent = _build_agent(monkeypatch) + from agent.codex_responses_adapter import _normalize_codex_response + observed = {} agent.interim_assistant_callback = lambda text, *, already_streamed=False: observed.update( {"text": text, "already_streamed": already_streamed} ) - agent._emit_interim_assistant_message({ - "role": "assistant", - "content": "", - "codex_message_items": [ - { - "type": "message", - "role": "assistant", - "phase": "analysis", - "content": [{"type": "output_text", "text": "Need inspect files."}], - }, - { - "type": "message", - "role": "assistant", - "phase": "commentary", - "content": [{"type": "output_text", "text": "I'll inspect the repo first."}], - }, - { - "type": "message", - "role": "assistant", - "phase": "final_answer", - "content": [{"type": "output_text", "text": "Done."}], - }, - ], - }) + normalized, finish_reason = _normalize_codex_response( + _codex_commentary_final_tool_response("I'll inspect the repo first.") + ) + assert finish_reason == "tool_calls" + assert normalized.content == "Done." + agent._emit_interim_assistant_message( + agent._build_assistant_message(normalized, finish_reason) + ) assert observed == { "text": "I'll inspect the repo first.", @@ -2286,6 +2295,12 @@ def test_stream_delta_preserves_code_fence_newlines(monkeypatch): def test_run_conversation_codex_continues_after_commentary_phase_message(monkeypatch): agent = _build_agent(monkeypatch) + emitted = [] + stream_events = [] + agent.interim_assistant_callback = ( + lambda text, *, already_streamed=False: emitted.append((text, already_streamed)) + ) + agent.stream_delta_callback = stream_events.append responses = [ _codex_commentary_message_response("I'll inspect the repo structure first."), _codex_tool_call_response(), @@ -2309,6 +2324,7 @@ def test_run_conversation_codex_continues_after_commentary_phase_message(monkeyp assert result["completed"] is True assert result["final_response"] == "Architecture summary complete." + assert emitted == [("I'll inspect the repo structure first.", False)] commentary_messages = [ msg for msg in result["messages"] if msg.get("role") == "assistant" and msg.get("finish_reason") == "incomplete" @@ -2324,6 +2340,39 @@ def test_run_conversation_codex_continues_after_commentary_phase_message(monkeyp assert any(msg.get("role") == "tool" and msg.get("tool_call_id") == "call_1" for msg in result["messages"]) +def test_codex_commentary_emits_before_tool_and_withholds_final_answer(monkeypatch): + agent = _build_agent(monkeypatch) + events = [] + agent.interim_assistant_callback = ( + lambda text, *, already_streamed=False: events.append(("interim", text)) + ) + responses = [ + _codex_commentary_message_response("I'll inspect the repo first."), + _codex_commentary_final_tool_response("I'll inspect the repo first."), + _codex_message_response("Verified final answer."), + ] + monkeypatch.setattr(agent, "_interruptible_api_call", lambda api_kwargs: responses.pop(0)) + + def _fake_execute_tool_calls(assistant_message, messages, effective_task_id, *_args): + events.append(("tool", assistant_message.tool_calls[0].function.name)) + messages.append({ + "role": "tool", + "tool_call_id": assistant_message.tool_calls[0].id, + "content": '{"ok":true}', + }) + + monkeypatch.setattr(agent, "_execute_tool_calls", _fake_execute_tool_calls) + + result = agent.run_conversation("analyze repo") + + assert result["completed"] is True + assert events == [ + ("interim", "I'll inspect the repo first."), + ("tool", "terminal"), + ] + assert all(text != "Done." for kind, text in events if kind == "interim") + + def test_run_conversation_codex_continues_after_ack_stop_message(monkeypatch): agent = _build_agent(monkeypatch) responses = [ @@ -2814,13 +2863,23 @@ def test_duplicate_detection_distinguishes_different_codex_reasoning(monkeypatch assert items[0].get("encrypted_content") == "enc_second" -def test_duplicate_detection_distinguishes_different_codex_message_items(monkeypatch): - """Incomplete turns with same visible content but different message ids - are deduped — only one interim kept, opaque state updated in-place (#52711).""" +def test_duplicate_detection_uses_commentary_when_hidden_reasoning_changes(monkeypatch): + """Identical commentary is emitted once while newer replay state wins.""" agent = _build_agent(monkeypatch) + emitted = [] + agent.interim_assistant_callback = ( + lambda text, *, already_streamed=False: emitted.append(text) + ) responses = [ SimpleNamespace( output=[ + SimpleNamespace( + type="reasoning", + id="rs_first", + encrypted_content="enc_first", + summary=[SimpleNamespace(text="hidden first")], + status="completed", + ), SimpleNamespace( type="message", id="msg_first", @@ -2835,6 +2894,13 @@ def test_duplicate_detection_distinguishes_different_codex_message_items(monkeyp ), SimpleNamespace( output=[ + SimpleNamespace( + type="reasoning", + id="rs_second", + encrypted_content="enc_second", + summary=[SimpleNamespace(text="hidden second")], + status="completed", + ), SimpleNamespace( type="message", id="msg_second", @@ -2854,6 +2920,7 @@ def test_duplicate_detection_distinguishes_different_codex_message_items(monkeyp result = agent.run_conversation("keep going") assert result["completed"] is True + assert emitted == ["Still working..."] interim_msgs = [ msg for msg in result["messages"] if msg.get("role") == "assistant" @@ -2865,6 +2932,10 @@ def test_duplicate_detection_distinguishes_different_codex_message_items(monkeyp items = interim_msgs[0].get("codex_message_items") if items: assert items[0].get("id") == "msg_second" + assert "hidden second" in (interim_msgs[0].get("reasoning") or "") + reasoning_items = interim_msgs[0].get("codex_reasoning_items") + if reasoning_items: + assert reasoning_items[0].get("id") == "rs_second" def test_chat_messages_to_responses_input_deduplicates_reasoning_ids(monkeypatch):