fix(codex): avoid persisting app-server user echo

agent/codex_runtime.py

@@ -678,7 +678,7 @@ def on_event(note: dict) -> None:
 def run_codex_app_server_turn(
     agent,
     *,
-    user_message: str,
+    user_message: Any,
     original_user_message: Any,
     messages: List[Dict[str, Any]],
     effective_task_id: str,
@@ -827,7 +827,29 @@ def run_codex_app_server_turn(
     if turn.projected_messages:
         from agent.message_metadata import append_message
 
-        for projected_message in turn.projected_messages:
+        projected_messages = turn.projected_messages
+        # ``turn/start`` materializes the submitted input as the leading
+        # ``userMessage`` item.  That item is a transport echo, not a second
+        # user action: build_turn_context already appended and persisted the
+        # same input before this early-return runtime was entered. ``run_turn``
+        # records the exact text it serialized into the wire input after rich
+        # content coercion; use that value rather than the original Hermes
+        # shape. Drop only an exact match — normalization here could erase a
+        # distinct user event such as a future turn/steer projection.
+        first_projected = projected_messages[0]
+        submitted_user_text = getattr(turn, "submitted_user_text", None)
+        if submitted_user_text is None and isinstance(user_message, str):
+            # Compatibility for older/mocked TurnResult shapes. Live sessions
+            # always return submitted_user_text.
+            submitted_user_text = user_message
+        if (
+            isinstance(first_projected, dict)
+            and first_projected.get("role") == "user"
+            and first_projected.get("content") == submitted_user_text
+        ):
+            projected_messages = projected_messages[1:]
+
+        for projected_message in projected_messages:
             append_message(messages, projected_message)
 
         # Persist the newly-projected assistant/tool messages ourselves.
@@ -938,8 +960,8 @@ def run_codex_app_server_turn(
         # The codex app-server runtime IS an early-return path that bypasses
         # conversation_loop, but we flush the projected assistant/tool messages
         # ourselves above (see the _flush_messages_to_session_db call after
-        # messages.extend). The inbound user turn was already flushed at turn
-        # start (turn_context._persist_session) and the flush dedups via
+        # the projection splice). The inbound user turn was already flushed at
+        # turn start (turn_context._persist_session) and the flush dedups via
         # _DB_PERSISTED_MARKER, so state.db ends up with each real message
         # exactly once and session_search / conversation-distill see the full
         # gateway conversation. Report agent_persisted=True so the gateway

agent/transports/codex_app_server_session.py

@@ -72,6 +72,9 @@ class TurnResult:
     error: Optional[str] = None  # Set if turn ended in a non-recoverable error
     turn_id: Optional[str] = None
     thread_id: Optional[str] = None
+    # Exact text serialized into the turn/start input item. The runtime uses
+    # this to distinguish Codex's transport echo from a separate user event.
+    submitted_user_text: Optional[str] = None
     token_usage_last: Optional[dict[str, Any]] = None
     token_usage_total: Optional[dict[str, Any]] = None
     model_context_window: Optional[int] = None
@@ -515,6 +518,7 @@ def run_turn(
         projector = CodexEventProjector()
 
         user_input_text = _coerce_turn_input_text(user_input)
+        result.submitted_user_text = user_input_text
 
         # Send turn/start with the user input. Text-only for now (codex
         # supports rich content but Hermes' text path is the common case).

tests/agent/test_codex_app_server_persist.py

@@ -14,13 +14,20 @@
 skips its own ``append_to_transcript`` DB write. This is critical: the inbound
 user turn is already flushed at turn start (``turn_context._persist_session``),
 and ``append_message`` is a raw INSERT with no dedup — a gateway re-write would
-duplicate the user turn (#860 / #42039). This test locks in:
+duplicate the user turn (#860 / #42039).
+
+The projection-splice comment in ``run_codex_app_server_turn`` documents why
+Codex's leading ``userMessage`` is a transport echo rather than a second user
+action. These tests lock in:
 
 1. ``run_codex_app_server_turn`` flushes projected messages and returns
    ``agent_persisted=True``.
 2. Exactly-once persistence: the already-flushed user turn is NOT re-written,
-   and the new projected assistant message lands once.
-3. The gateway resolution expression preserves standard-runtime behaviour.
+   the projected input echo is NOT inserted, and the assistant lands once.
+3. Assistant-only and non-matching leading projections are preserved.
+4. A later distinct user projection is preserved.
+5. Rich input is matched against the exact text submitted on the wire.
+6. The gateway resolution expression preserves standard-runtime behaviour.
 """
 
 import tempfile
@@ -33,13 +40,18 @@
 from run_agent import AIAgent
 
 
-def _make_turn():
+def _make_turn(*, user_echo=None, submitted_user_text=None):
+    projected_messages = []
+    if user_echo is not None:
+        projected_messages.append({"role": "user", "content": user_echo})
+    projected_messages.append({"role": "assistant", "content": "CODEX_ASSISTANT"})
     return SimpleNamespace(
         interrupted=False,
         error=None,
         thread_id="thread-1",
         turn_id="turn-1",
-        projected_messages=[{"role": "assistant", "content": "CODEX_ASSISTANT"}],
+        submitted_user_text=submitted_user_text,
+        projected_messages=projected_messages,
         tool_iterations=0,
         final_text="CODEX_ASSISTANT",
         should_retire=False,
@@ -72,7 +84,11 @@ def test_codex_success_flushes_and_reports_persisted():
         effective_task_id="task-1",
     )
     assert result["completed"] is True
-    assert isinstance(result["messages"][-1]["timestamp"], float)
+    assert [(message["role"], message.get("content")) for message in result["messages"]] == [
+        ("user", "hello"),
+        ("assistant", "CODEX_ASSISTANT"),
+    ]
+    assert isinstance(result["messages"][1]["timestamp"], float)
     # With the agent as sole persister, the gateway must SKIP its DB write.
     assert result["agent_persisted"] is True
 
@@ -105,12 +121,88 @@ def clear_interrupt():
     assert agent._interrupt_requested is False
 
 
+def test_codex_drops_only_the_leading_matching_user_echo():
+    """A later distinct user projection must survive the input-echo filter."""
+    agent = _make_agent(session_db=None)
+    turn = _make_turn()
+    turn.projected_messages = [
+        {"role": "user", "content": "hello"},
+        {"role": "assistant", "content": "INTERIM"},
+        {"role": "user", "content": "STEER"},
+        {"role": "assistant", "content": "CODEX_ASSISTANT"},
+    ]
+    turn.submitted_user_text = "hello"
+    agent._codex_session.run_turn.return_value = turn
+
+    result = run_codex_app_server_turn(
+        agent,
+        user_message="hello",
+        original_user_message="hello",
+        messages=[{"role": "user", "content": "hello"}],
+        effective_task_id="task-1",
+    )
+
+    assert [(message["role"], message.get("content")) for message in result["messages"]] == [
+        ("user", "hello"),
+        ("assistant", "INTERIM"),
+        ("user", "STEER"),
+        ("assistant", "CODEX_ASSISTANT"),
+    ]
+
+
+def test_codex_preserves_nonmatching_leading_user_projection():
+    agent = _make_agent(session_db=None)
+    turn = _make_turn(user_echo="DIFFERENT", submitted_user_text="hello")
+    agent._codex_session.run_turn.return_value = turn
+
+    result = run_codex_app_server_turn(
+        agent,
+        user_message="hello",
+        original_user_message="hello",
+        messages=[{"role": "user", "content": "hello"}],
+        effective_task_id="task-1",
+    )
+
+    assert [(message["role"], message.get("content")) for message in result["messages"]] == [
+        ("user", "hello"),
+        ("user", "DIFFERENT"),
+        ("assistant", "CODEX_ASSISTANT"),
+    ]
+
+
+def test_codex_drops_echo_of_coerced_rich_wire_text():
+    rich_input = [
+        {"type": "text", "text": "caption"},
+        {"type": "image_url", "image_url": {"url": "data:image/png;base64,abc"}},
+    ]
+    submitted_text = "caption\n\n[image attached]"
+    agent = _make_agent(session_db=None)
+    agent._codex_session.run_turn.return_value = _make_turn(
+        user_echo=submitted_text,
+        submitted_user_text=submitted_text,
+    )
+
+    result = run_codex_app_server_turn(
+        agent,
+        user_message=rich_input,
+        original_user_message=rich_input,
+        messages=[{"role": "user", "content": rich_input}],
+        effective_task_id="task-1",
+    )
+
+    assert [(message["role"], message.get("content")) for message in result["messages"]] == [
+        ("user", rich_input),
+        ("assistant", "CODEX_ASSISTANT"),
+    ]
+
+
 def test_codex_turn_persists_each_message_exactly_once():
     """The user turn (flushed at turn start) must not be duplicated; the
     projected assistant message must land once.  Uses a real SessionDB and the
     real AIAgent._flush_messages_to_session_db to prove no #860/#42039
     duplicate-write regression on the codex path."""
     tmp = tempfile.mkdtemp(prefix="codex_persist_")
+    db = None
     try:
         db = SessionDB(Path(tmp) / "state.db")
         sid = "sess-codex-once"
@@ -128,7 +220,14 @@ def test_codex_turn_persists_each_message_exactly_once():
         )
         agent._session_db_created = True
         agent._codex_session = MagicMock()
-        agent._codex_session.run_turn.return_value = _make_turn()
+        # A real app-server turn projects the submitted input back as a
+        # leading userMessage before the assistant response.  Hermes already
+        # owns and flushed that input at turn start, so the transport echo
+        # must not become a second durable user row.
+        agent._codex_session.run_turn.return_value = _make_turn(
+            user_echo="USER_TURN",
+            submitted_user_text="USER_TURN",
+        )
         agent.tool_progress_callback = None
 
         # Model the real flow: the inbound user turn is flushed at turn start
@@ -163,6 +262,8 @@ def test_codex_turn_persists_each_message_exactly_once():
     finally:
         import shutil
 
+        if db is not None:
+            db.close()
         shutil.rmtree(tmp)
 
 

tests/agent/transports/test_codex_app_server_session.py

@@ -209,6 +209,32 @@ def test_simple_text_turn_returns_final_message(self):
         # turn_id propagated for downstream session-DB linkage
         assert r.turn_id == "turn-fake-001"
 
+    def test_rich_input_records_exact_submitted_wire_text(self):
+        client = FakeClient()
+        client.queue_notification(
+            "item/completed",
+            item={"type": "agentMessage", "id": "m1", "text": "done"},
+            threadId="t",
+            turnId="tu1",
+        )
+        client.queue_notification(
+            "turn/completed",
+            threadId="t",
+            turn={"id": "tu1", "status": "completed", "error": None},
+        )
+        rich_input = [
+            {"type": "text", "text": "caption"},
+            {"type": "image_url", "image_url": {"url": "data:image/png;base64,abc"}},
+        ]
+
+        result = make_session(client).run_turn(rich_input, turn_timeout=2.0)
+
+        _, params = next(request for request in client.requests if request[0] == "turn/start")
+        assert params["input"] == [
+            {"type": "text", "text": "caption\n\n[image attached]"}
+        ]
+        assert result.submitted_user_text == "caption\n\n[image attached]"
+
 
 
     def test_foreign_completion_in_server_request_drain_is_ignored(self):
@@ -895,4 +921,3 @@ def test_empty_inputs(self):
         assert _classify_oauth_failure() is None
         assert _classify_oauth_failure("") is None
         assert _classify_oauth_failure("", None) is None  # type: ignore[arg-type]
-