fix(agent): dedup codex incomplete interims on visible content

Two consecutive incomplete assistant interims with identical visible
content (content + reasoning) are collapsed even when opaque provider
state (encrypted reasoning item ids, message item phases) drifts per
continuation — previously that drift defeated dedup and caused message
storms (#52711). The latest opaque payload is written onto the existing
message in place, so continuation replay still uses fresh provider
state.

Salvage note: the original PR also made the 3-retry continuation cap
cumulative per turn (no reset on progress). That half is intentionally
dropped — a legitimate long turn alternating incomplete/progress would
hard-fail at 3 cumulative, changing semantics beyond the reported bug.

agent/conversation_loop.py

@@ -4465,27 +4465,28 @@ def run_conversation(
                     or interim_has_codex_message_items
                 ):
                     last_msg = messages[-1] if messages else None
-                    # Duplicate detection: two consecutive incomplete assistant
+                    # Duplicate detection: compare only visible content
-                    # messages with identical content AND reasoning are collapsed.
+                    # (content + reasoning).  Opaque provider state
-                    # For provider-state-only changes (encrypted reasoning
+                    # (encrypted reasoning items, message item ids/phases)
-                    # items or replayable message ids/phases/statuses differ
+                    # drifts per continuation even when the visible output
-                    # while visible content/reasoning are unchanged), compare
+                    # is identical, so including it in the comparison defeats
-                    # those opaque payloads too so we don't silently drop the
+                    # dedup and causes message storms (#52711).
-                    # newer continuation state.
+                    visible_duplicate = (
-                    last_codex_items = last_msg.get("codex_reasoning_items") if isinstance(last_msg, dict) else None
-                    interim_codex_items = interim_msg.get("codex_reasoning_items")
-                    last_codex_message_items = last_msg.get("codex_message_items") if isinstance(last_msg, dict) else None
-                    interim_codex_message_items = interim_msg.get("codex_message_items")
-                    duplicate_interim = (
                         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 last_codex_items == interim_codex_items
-                        and last_codex_message_items == interim_codex_message_items
                     )
-                    if not duplicate_interim:
+                    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
+                    else:
                         messages.append(interim_msg)
                         agent._emit_interim_assistant_message(interim_msg)
 

tests/run_agent/test_run_agent_codex_responses.py

@@ -1704,6 +1704,90 @@ def test_mid_turn_compaction_does_not_double_persist_in_place_rows(monkeypatch,
     )
 
 
+def _codex_incomplete_with_reasoning(text: str, reasoning_id: str = "rs_default"):
+    """Incomplete response with a reasoning item whose id/encrypted_content
+    can vary independently of the visible message text."""
+    return SimpleNamespace(
+        output=[
+            SimpleNamespace(
+                type="reasoning",
+                id=reasoning_id,
+                encrypted_content=f"opaque_{reasoning_id}",
+                summary=[SimpleNamespace(text="thinking...")],
+            ),
+            SimpleNamespace(
+                type="message",
+                status="in_progress",
+                content=[SimpleNamespace(type="output_text", text=text)],
+            ),
+        ],
+        usage=SimpleNamespace(input_tokens=4, output_tokens=2, total_tokens=6),
+        status="in_progress",
+        model="gpt-5-codex",
+    )
+
+
+def test_codex_incomplete_visible_dedup_suppresses_duplicate_interims(monkeypatch):
+    """Two consecutive incomplete responses with identical visible content
+    but different opaque reasoning items should be collapsed — only the first
+    interim is emitted to the user (#52711)."""
+    agent = _build_agent(monkeypatch)
+    # 2 incompletes with same text but different reasoning ids, then a final.
+    # (Only 2 to avoid hitting the cap of 3.)
+    responses = [
+        _codex_incomplete_with_reasoning("Working on it...", "rs_1"),
+        _codex_incomplete_with_reasoning("Working on it...", "rs_2"),
+        _codex_message_response("Done."),
+    ]
+    monkeypatch.setattr(agent, "_interruptible_api_call", lambda api_kwargs: responses.pop(0))
+
+    emitted: list = []
+    original_emit = agent._emit_interim_assistant_message
+    def _capture_emit(msg):
+        emitted.append(msg.get("content"))
+        original_emit(msg)
+    monkeypatch.setattr(agent, "_emit_interim_assistant_message", _capture_emit)
+
+    result = agent.run_conversation("test dedup")
+
+    assert result["completed"] is True
+    # Only ONE interim should have been emitted (the first), not two.
+    assert len(emitted) == 1
+    assert emitted[0] == "Working on it..."
+
+
+def test_codex_incomplete_opaque_state_updated_in_place(monkeypatch):
+    """When visible content is a duplicate, the last message's opaque state
+    (codex_reasoning_items) should be updated in-place without emitting a new
+    interim (#52711)."""
+    agent = _build_agent(monkeypatch)
+    responses = [
+        _codex_incomplete_with_reasoning("Partial output...", "rs_1"),
+        _codex_incomplete_with_reasoning("Partial output...", "rs_2"),
+        _codex_message_response("Final."),
+    ]
+    monkeypatch.setattr(agent, "_interruptible_api_call", lambda api_kwargs: responses.pop(0))
+
+    result = agent.run_conversation("test opaque update")
+
+    assert result["completed"] is True
+    # Find the incomplete interim message in the result.
+    incompletes = [
+        m for m in result["messages"]
+        if m.get("role") == "assistant" and m.get("finish_reason") == "incomplete"
+    ]
+    # Only one incomplete message should exist (the second was deduped).
+    assert len(incompletes) == 1
+    # The opaque state should reflect the LATEST reasoning item (rs_2),
+    # updated in-place on the single message.
+    items = incompletes[0].get("codex_reasoning_items")
+    if items:
+        assert any(
+            (i.get("id") if isinstance(i, dict) else getattr(i, "id", None)) == "rs_2"
+            for i in items
+        )
+
+
 def test_normalize_codex_response_marks_commentary_only_message_as_incomplete(monkeypatch):
     agent = _build_agent(monkeypatch)
     from agent.codex_responses_adapter import _normalize_codex_response
@@ -2608,7 +2692,8 @@ def test_codex_message_item_status_survives_conversion_and_preflight(monkeypatch
 
 def test_duplicate_detection_distinguishes_different_codex_reasoning(monkeypatch):
     """Two consecutive reasoning-only responses with different encrypted content
-    must NOT be treated as duplicates."""
+    are deduped on visible content — only one interim is kept, but opaque state
+    is updated in-place (#52711)."""
     agent = _build_agent(monkeypatch)
     responses = [
         # First reasoning-only response
@@ -2641,23 +2726,23 @@ def test_duplicate_detection_distinguishes_different_codex_reasoning(monkeypatch
 
     assert result["completed"] is True
     assert result["final_response"] == "Final answer after thinking."
-    # Both reasoning-only interim messages should be in history (not collapsed)
+    # Only one reasoning-only interim should be in history (deduped on
+    # visible content — both have empty visible output).
     interim_msgs = [
         msg for msg in result["messages"]
         if msg.get("role") == "assistant"
         and msg.get("finish_reason") == "incomplete"
     ]
-    assert len(interim_msgs) == 2
+    assert len(interim_msgs) == 1
-    encrypted_contents = [
+    # But the opaque state should reflect the LATEST reasoning item.
-        msg["codex_reasoning_items"][0]["encrypted_content"]
+    items = interim_msgs[0].get("codex_reasoning_items")
-        for msg in interim_msgs
+    if items:
-    ]
+        assert items[0].get("encrypted_content") == "enc_second"
-    assert "enc_first" in encrypted_contents
-    assert "enc_second" in encrypted_contents
 
 
 def test_duplicate_detection_distinguishes_different_codex_message_items(monkeypatch):
-    """Incomplete turns with new message ids/phases/statuses must not be collapsed."""
+    """Incomplete turns with same visible content but different message ids
+    are deduped — only one interim kept, opaque state updated in-place (#52711)."""
     agent = _build_agent(monkeypatch)
     responses = [
         SimpleNamespace(
@@ -2700,12 +2785,12 @@ def test_duplicate_detection_distinguishes_different_codex_message_items(monkeyp
         if msg.get("role") == "assistant"
         and msg.get("finish_reason") == "incomplete"
     ]
-    assert len(interim_msgs) == 2
+    # Only one interim — deduped on visible content ("Still working..." == "Still working...").
-    assert [msg["codex_message_items"][0]["id"] for msg in interim_msgs] == [
+    assert len(interim_msgs) == 1
-        "msg_first",
+    # Opaque state should reflect the latest message item.
-        "msg_second",
+    items = interim_msgs[0].get("codex_message_items")
-    ]
+    if items:
-    assert all(msg["codex_message_items"][0]["status"] == "in_progress" for msg in interim_msgs)
+        assert items[0].get("id") == "msg_second"
 
 
 def test_chat_messages_to_responses_input_deduplicates_reasoning_ids(monkeypatch):