From 1c051a4b6578fdfe61f19abe18c692dc636d3901 Mon Sep 17 00:00:00 2001 From: Sean Knowles Date: Thu, 23 Jul 2026 00:40:46 +0100 Subject: [PATCH] Retry Codex messages across turn boundaries --- omnigent/runtime/pending_inputs.py | 52 ++++++++ omnigent/server/routes/sessions.py | 73 ++++++++--- tests/runtime/test_pending_inputs.py | 32 +++++ .../integration/test_sessions_endpoints.py | 124 +++++++++++++++++- 4 files changed, 259 insertions(+), 22 deletions(-) diff --git a/omnigent/runtime/pending_inputs.py b/omnigent/runtime/pending_inputs.py index 7142a69140..2bb4ce45bb 100644 --- a/omnigent/runtime/pending_inputs.py +++ b/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. diff --git a/omnigent/server/routes/sessions.py b/omnigent/server/routes/sessions.py index 4b639e681a..af5d08c48b 100644 --- a/omnigent/server/routes/sessions.py +++ b/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." ) ), ) diff --git a/tests/runtime/test_pending_inputs.py b/tests/runtime/test_pending_inputs.py index f263686826..262e19dff5 100644 --- a/tests/runtime/test_pending_inputs.py +++ b/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 diff --git a/tests/server/integration/test_sessions_endpoints.py b/tests/server/integration/test_sessions_endpoints.py index c6e7a1c56e..198739d507 100644 --- a/tests/server/integration/test_sessions_endpoints.py +++ b/tests/server/integration/test_sessions_endpoints.py @@ -3197,26 +3197,40 @@ 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( @@ -3224,19 +3238,42 @@ async def test_post_external_session_status_missing_input_persists_failed_turn( 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="owner@example.com", ) 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,