Retry Codex messages across turn boundaries

omnigent/runtime/pending_inputs.py

@@ -121,6 +121,19 @@ class DrainedInput:
     created_by: str | None = None
 
 
+@dataclass
+class RetryClaim:
+    """A pending input claimed for one automatic delivery retry.
+
+    :param input: Copy of the pending input to forward again.
+    :param should_forward: ``False`` when the same terminal status already
+        triggered the retry and this is only a duplicate status delivery.
+    """
+
+    input: DrainedInput
+    should_forward: bool
+
+
 @dataclass
 class MatchedDrain:
     """Result from draining pending inputs up to a text-matched entry."""
@@ -149,11 +162,17 @@ class _Entry:
         carries the correct author on all clients.
     :param created_at: ``time.monotonic()`` timestamp at record time,
         used only for TTL eviction.
+    :param delivery_attempts: Number of runner forwards attempted for this
+        input, including the original delivery.
+    :param retry_response_id: Terminal turn id that triggered the automatic
+        retry, used to make duplicate status delivery idempotent.
     """
 
     pending_id: str
     content: list[dict[str, Any]]
     created_by: str | None = None
+    delivery_attempts: int = 1
+    retry_response_id: str | None = None
     # Lambda (not ``_now`` directly) so a monkeypatched ``_now`` is
     # resolved at construction time rather than bound at class def.
     created_at: float = field(default_factory=lambda: _now())
@@ -243,6 +262,39 @@ def resolve(conversation_id: str, pending_id: str) -> None:
             _pending.pop(conversation_id, None)
 
 
+def claim_oldest_retry(
+    conversation_id: str,
+    response_id: str | None,
+) -> RetryClaim | None:
+    """Claim the oldest pending input for one automatic retry.
+
+    The original forward can race the previous native turn's completion and
+    be accepted by the runner without reaching Codex. A terminal status while
+    the prompt is still pending proves the transcript did not accept it, so AP
+    may safely forward it once more after the forwarder clears the active turn.
+    The entry remains pending until Codex mirrors the accepted user message.
+
+    :param conversation_id: Conversation/session id, e.g. ``"conv_abc123"``.
+    :param response_id: Terminal native turn id. Used as an idempotency key
+        when the forwarder retries delivery of the same status POST.
+    :returns: A retry claim, or ``None`` when no input is pending or its one
+        automatic retry was already consumed by a different terminal turn.
+    """
+    with _lock:
+        _evict_stale_locked(conversation_id, _now())
+        entries = _pending.get(conversation_id)
+        if entries is None:
+            return None
+        entry = next(iter(entries.values()))
+        if response_id is not None and entry.retry_response_id == response_id:
+            return RetryClaim(input=_drained_input(entry), should_forward=False)
+        if entry.delivery_attempts >= 2:
+            return None
+        entry.delivery_attempts += 1
+        entry.retry_response_id = response_id
+        return RetryClaim(input=_drained_input(entry), should_forward=True)
+
+
 def resolve_oldest(conversation_id: str) -> DrainedInput | None:
     """
     Drain the oldest pending entry (FIFO) and return it.

omnigent/server/routes/sessions.py

@@ -20730,12 +20730,11 @@ async def post_event(
             # forwarders mark that boundary ``input_missing`` after a targeted
             # thread/resume recovery also finds no input. Older, already-live
             # sandboxes cannot emit that hint, so a terminal Codex status with
-            # an AP-side pending web-composer message is the compatibility
-            # signal: successful Codex turns synchronously persist their user
-            # item before this status edge. A remaining pending input is thus
-            # the exact prompt Codex failed to accept.
-            # Persist the prompt plus a durable error and consume its pending
-            # bubble so reload/rebind can never make the user's message vanish.
+            # a pending prompt is the compatibility signal. The common case is
+            # a follow-up racing the previous turn's final boundary: retry it
+            # once now that the forwarder has cleared the active turn. If that
+            # retry also ends without a mirrored user item, persist the prompt
+            # plus a durable error so it can never disappear.
             if (
                 status in {"idle", "failed"}
                 and (
@@ -20744,15 +20743,59 @@ async def post_event(
                 )
                 and _is_native_terminal_session(conv)
             ):
+                raw_missing_input_output = body.data.get("output")
+                missing_input_output = (
+                    raw_missing_input_output.strip()
+                    if isinstance(raw_missing_input_output, str)
+                    and raw_missing_input_output.strip()
+                    else None
+                )
+                retry_claim = (
+                    pending_inputs.claim_oldest_retry(session_id, response_id)
+                    if status == "idle"
+                    and missing_input_output is None
+                    and body.data.get("reauth_required") is not True
+                    else None
+                )
+                if retry_claim is not None:
+                    if retry_claim.should_forward:
+                        retry_runner = await _get_runner_client(session_id, runner_router)
+                        if retry_runner is not None:
+                            try:
+                                await _forward_native_terminal_message(
+                                    retry_runner,
+                                    session_id,
+                                    conv,
+                                    SessionEventInput(
+                                        type="message",
+                                        data={
+                                            "role": "user",
+                                            "content": retry_claim.input.content,
+                                        },
+                                    ),
+                                    file_store=file_store,
+                                    artifact_store=artifact_store,
+                                )
+                            except HTTPException:
+                                _logger.warning(
+                                    "Codex pending-input automatic retry failed: session=%s",
+                                    session_id,
+                                    exc_info=True,
+                                )
+                            else:
+                                return {
+                                    "queued": False,
+                                    "retried_pending_id": retry_claim.input.pending_id,
+                                }
+                    else:
+                        # The forwarder retried an already-processed terminal
+                        # status POST. Acknowledge without delivering twice.
+                        return {
+                            "queued": False,
+                            "retried_pending_id": retry_claim.input.pending_id,
+                        }
                 drained_input = pending_inputs.resolve_oldest(session_id)
                 if drained_input is not None:
-                    raw_missing_input_output = body.data.get("output")
-                    missing_input_output = (
-                        raw_missing_input_output.strip()
-                        if isinstance(raw_missing_input_output, str)
-                        and raw_missing_input_output.strip()
-                        else None
-                    )
                     missing_input_error = ErrorData(
                         source="execution",
                         code=(
@@ -20767,8 +20810,8 @@ async def post_event(
                         message=(
                             missing_input_output
                             or (
-                                "Codex completed the turn without accepting this message. "
-                                "Please retry."
+                                "Codex could not accept this message after an automatic retry. "
+                                "Please send it again."
                             )
                         ),
                     )

tests/runtime/test_pending_inputs.py

@@ -130,6 +130,38 @@ def test_resolve_oldest_returns_none_when_empty() -> None:
     assert pending_inputs.resolve_oldest("conv_a") is None
 
 
+def test_claim_oldest_retry_is_bounded_and_keeps_input_pending() -> None:
+    """One automatic retry preserves the bubble until transcript acceptance."""
+    first = pending_inputs.record("conv_a", [_text_block("first")])
+    second = pending_inputs.record("conv_a", [_text_block("second")])
+
+    claim = pending_inputs.claim_oldest_retry("conv_a", "turn_previous")
+    assert claim is not None
+    assert claim.should_forward is True
+    assert claim.input.pending_id == first
+    assert [entry["pending_id"] for entry in pending_inputs.snapshot_for("conv_a")] == [
+        first,
+        second,
+    ]
+    assert pending_inputs.claim_oldest_retry("conv_a", "turn_retry_failed") is None
+
+
+def test_duplicate_terminal_status_does_not_retry_twice() -> None:
+    """A repeated terminal status acknowledges the existing retry."""
+    first = pending_inputs.record("conv_a", [_text_block("first")])
+    second = pending_inputs.record("conv_a", [_text_block("second")])
+
+    first_claim = pending_inputs.claim_oldest_retry("conv_a", "turn_first")
+    duplicate = pending_inputs.claim_oldest_retry("conv_a", "turn_first")
+    assert first_claim is not None and first_claim.should_forward is True
+    assert duplicate is not None and duplicate.should_forward is False
+    assert duplicate.input.pending_id == first
+    assert [entry["pending_id"] for entry in pending_inputs.snapshot_for("conv_a")] == [
+        first,
+        second,
+    ]
+
+
 def test_resolve_oldest_drains_regardless_of_reformatted_text() -> None:
     """
     Regression: a queued message drains even when the transcript

tests/server/integration/test_sessions_endpoints.py

@@ -3197,46 +3197,83 @@ async def test_post_external_session_status_failed_surfaces_output_and_reauth(
     assert "401 Unauthorized" in error["message"]
 
 
-async def test_post_external_session_status_missing_input_persists_failed_turn(
+async def test_post_external_session_status_missing_input_retries_once_then_persists_failure(
     client: httpx.AsyncClient,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """A Codex empty turn cannot silently discard a pending web prompt.
+    """A Codex empty turn retries once, then preserves a durable failure.
 
     Regression for the production failure where ``turn/start`` returned
     accepted, Codex immediately completed with zero observed items, and the
-    optimistic user bubble disappeared on rebind. This deliberately omits the
-    newer forwarder ``input_missing`` hint to prove that already-live Codex
-    sandboxes are protected too: AP must promote its pending input into durable
-    history, attach an actionable error, and clear the exact pending bubble id.
+    optimistic user bubble disappeared on rebind. AP first redelivers the
+    pending input after Codex clears the completed turn. If the retry also
+    completes without accepting it, AP promotes the prompt into durable
+    history, attaches an actionable error, and clears the exact pending id.
     """
     from omnigent.runtime import pending_inputs
 
     published: list[tuple[str, dict[str, Any]]] = []
+    forwarded: list[list[dict[str, Any]]] = []
     monkeypatch.setattr(
         "omnigent.server.routes.sessions.session_stream.publish",
         lambda session_id, event: published.append((session_id, event)),
     )
+
+    async def _runner_client(*_args: Any, **_kwargs: Any) -> object:
+        return object()
+
+    async def _forward_retry(
+        _runner: object,
+        _session_id: str,
+        _conv: Any,
+        event: Any,
+        **_kwargs: Any,
+    ) -> None:
+        forwarded.append(event.data["content"])
+
     pending_inputs.reset_for_tests()
     agent = await create_test_agent(client)
     session = await _create_session(
         client,
         agent["id"],
         labels={"omnigent.wrapper": "codex-native-ui"},
     )
+    monkeypatch.setattr("omnigent.server.routes.sessions._get_runner_client", _runner_client)
+    monkeypatch.setattr(
+        "omnigent.server.routes.sessions._forward_native_terminal_message",
+        _forward_retry,
+    )
     pending_id = pending_inputs.record(
         session["id"],
         [{"type": "input_text", "text": "Still working?"}],
         created_by="[email protected]",
     )
     try:
+        retry_resp = await client.post(
+            f"/v1/sessions/{session['id']}/events",
+            json={
+                "type": "external_session_status",
+                "data": {
+                    "status": "idle",
+                    "response_id": "codex_turn_initial",
+                },
+            },
+        )
+        assert retry_resp.status_code == 202, retry_resp.text
+        assert retry_resp.json()["retried_pending_id"] == pending_id
+        assert forwarded == [[{"type": "input_text", "text": "Still working?"}]]
+        assert [entry["pending_id"] for entry in pending_inputs.snapshot_for(session["id"])] == [
+            pending_id
+        ]
+
         resp = await client.post(
             f"/v1/sessions/{session['id']}/events",
             json={
                 "type": "external_session_status",
                 "data": {
                     "status": "idle",
-                    "response_id": "codex_turn_empty",
+                    "response_id": "codex_turn_retry",
+                    "input_missing": True,
                 },
             },
         )
@@ -3270,6 +3307,79 @@ async def test_post_external_session_status_missing_input_persists_failed_turn(
         pending_inputs.reset_for_tests()
 
 
+async def test_post_external_session_status_old_turn_does_not_reject_queued_followup(
+    client: httpx.AsyncClient,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A completion racing a follow-up automatically redelivers it.
+
+    Native messages are buffered one turn at a time. The old turn's terminal
+    status can therefore reach AP after the next composer message is recorded,
+    but before the runner starts its continuation. That status must not consume
+    or fail the newer message.
+    """
+    from omnigent.runtime import pending_inputs
+
+    forwarded: list[list[dict[str, Any]]] = []
+
+    async def _runner_client(*_args: Any, **_kwargs: Any) -> object:
+        return object()
+
+    async def _forward_retry(
+        _runner: object,
+        _session_id: str,
+        _conv: Any,
+        event: Any,
+        **_kwargs: Any,
+    ) -> None:
+        forwarded.append(event.data["content"])
+
+    pending_inputs.reset_for_tests()
+    agent = await create_test_agent(client)
+    session = await _create_session(
+        client,
+        agent["id"],
+        labels={"omnigent.wrapper": "codex-native-ui"},
+    )
+    try:
+        prior_running = await client.post(
+            f"/v1/sessions/{session['id']}/events",
+            json={
+                "type": "external_session_status",
+                "data": {"status": "running", "response_id": "codex_turn_prior"},
+            },
+        )
+        assert prior_running.status_code == 202, prior_running.text
+        monkeypatch.setattr("omnigent.server.routes.sessions._get_runner_client", _runner_client)
+        monkeypatch.setattr(
+            "omnigent.server.routes.sessions._forward_native_terminal_message",
+            _forward_retry,
+        )
+
+        pending_id = pending_inputs.record(
+            session["id"],
+            [{"type": "input_text", "text": "Can you see this message?"}],
+        )
+        terminal = await client.post(
+            f"/v1/sessions/{session['id']}/events",
+            json={
+                "type": "external_session_status",
+                "data": {"status": "idle", "response_id": "codex_turn_prior"},
+            },
+        )
+
+        assert terminal.status_code == 202, terminal.text
+        assert terminal.json()["retried_pending_id"] == pending_id
+        assert forwarded == [[{"type": "input_text", "text": "Can you see this message?"}]]
+        assert [entry["pending_id"] for entry in pending_inputs.snapshot_for(session["id"])] == [
+            pending_id
+        ]
+        items = (await client.get(f"/v1/sessions/{session['id']}/items")).json()["data"]
+        assert not any(item["type"] == "error" for item in items)
+    finally:
+        pending_inputs.reset_for_tests()
+
+
 async def test_post_external_session_status_carries_response_id(
     client: httpx.AsyncClient,
     monkeypatch: pytest.MonkeyPatch,