github-featured/omnigent-ai--omnigent
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix(codex-native): surface standalone turn errors (#5060)

Browse Source 
* fix(codex-native): surface standalone turn errors

Signed-off-by: Pat Sukprasert <pattara.sk127@gmail.com>

* fix(codex-native): dedupe terminal error edges

Signed-off-by: Pat Sukprasert <pattara.sk127@gmail.com>

---------

Signed-off-by: Pat Sukprasert <pattara.sk127@gmail.com>
This commit is contained in:
Pat Sukprasert
2026-08-20 14:07:36 +08:00
committed by GitHub
parent 33ec51372b
commit 1ed6b49671
5 changed files with 355 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
omnigent/codex_native_forwarder.py
+81 -13
View File
@@ -192,14 +192,10 @@ _CODEX_ELICITATION_REQUEST_METHODS = frozenset(
}
)
# Turn-error surfacing. A failed Codex turn arrives as ``turn/completed``
# (or ``turn/failed``) with ``turn.status == "failed"`` and a ``turn.error``
# object ``{message, codexErrorInfo?, additionalDetails?}``; keying status off
# the method alone mapped such turns to ``idle`` — a "silent success". The
# forwarder inspects ``turn.status``/``turn.error``, forces ``failed``, and
# surfaces the reason. As a fallback it also catches an ``error`` ThreadItem in
# ``turn.items``: both shapes exist in the app-server type system and the wire
# shape varies by version, so detecting either keeps the fix robust.
# Turn-error surfacing. Codex reports failures through a standalone ``error``
# notification and on terminal turn boundaries via ``turn.error`` / failed
# status. The forwarder handles both, plus the older ``error`` ThreadItem
# fallback, so every non-retrying failure reaches the session UI.
#
# ``codexErrorInfo`` is the app-server's structured classification (e.g.
# ``unauthorized``, ``usage_limit_exceeded``); auth-class values get a re-auth
@@ -359,6 +355,9 @@ class _CodexForwarderState:
:param synced_item_keys: Stable item keys already posted to Omnigent this
connection, e.g. ``{"thread_c:turn_c:item-1"}``. In-memory only;
guards replay-vs-live overlap within one forwarder lifetime.
:param surfaced_terminal_error_turns: Turn ids whose standalone terminal
``error`` notification was already surfaced. Used to suppress a later
terminal boundary for the same turn.
:param posted_user_turns: Turn ids whose ``userMessage`` has been
posted to Omnigent this connection, e.g. ``{"turn_123"}``. Used to
enforce user-before-assistant ordering: before posting a turn's
@@ -406,6 +405,7 @@ class _CodexForwarderState:
pending_child_threads: dict[str, str | None] = field(default_factory=dict)
subscribed_child_threads: set[str] = field(default_factory=set)
synced_item_keys: set[str] = field(default_factory=set)
surfaced_terminal_error_turns: set[str] = field(default_factory=set)
posted_user_turns: set[str] = field(default_factory=set)
posted_tool_calls: set[str] = field(default_factory=set)
partial_text_by_turn: dict[str, list[_PartialTextBuffer]] = field(default_factory=dict)
@@ -1006,6 +1006,15 @@ def _terminal_error_from_turn(params: _JsonObject) -> _CodexTerminalError | None
return _CodexTerminalError(message=message, kind=_classify_codex_error(payload, message))
def _terminal_error_from_notification(params: _JsonObject) -> _CodexTerminalError | None:
"""Return the failure carried by Codex's standalone ``error`` notification."""
payload = params.get("error")
if not isinstance(payload, dict):
return None
message = _error_payload_message(payload)
return _CodexTerminalError(message=message, kind=_classify_codex_error(payload, message))
@dataclass(frozen=True)
class _CodexTurnStatusEdge:
"""
@@ -2985,6 +2994,41 @@ async def _maybe_handle_turn_event(
:param forwarder_state: Optional forwarder state.
:returns: ``True`` when this event was handled.
"""
if method == "error":
if params.get("willRetry") is True:
_logger.info(
"Codex forwarder observed retryable turn error: turn_id=%s",
_turn_id_from_payload(params),
)
return True
if delta_coalescer is not None:
await delta_coalescer.flush()
error = _terminal_error_from_notification(params)
if error is None:
_logger.warning("Codex forwarder ignored malformed error notification")
return True
turn_id = _turn_id_from_payload(params)
if forwarder_state is not None and turn_id is not None:
if turn_id in forwarder_state.surfaced_terminal_error_turns:
_logger.info(
"Codex forwarder ignored duplicate terminal error: turn_id=%s",
turn_id,
)
return True
forwarder_state.surfaced_terminal_error_turns.add(turn_id)
clear_active_turn_id_if_matches(bridge_dir, turn_id)
await _post_turn_status_edge(
client,
session_id,
_CodexTurnStatusEdge(
status="failed",
turn_id=turn_id,
source="error",
error=error,
),
)
await usage_coalescer.flush()
return True
if method == "turn/started":
if delta_coalescer is not None:
await delta_coalescer.flush()
@@ -3230,7 +3274,14 @@ async def _handle_terminal_turn_boundary(
params=params,
forwarder_state=forwarder_state,
)
handled = await _handle_terminal_turn_event(client, session_id, bridge_dir, method, params)
handled = await _handle_terminal_turn_event(
client,
session_id,
bridge_dir,
method,
params,
forwarder_state=forwarder_state,
)
if handled:
await elicitation_tracker.resolve_by_terminal_turn_event(
client,
@@ -4124,21 +4175,38 @@ async def _handle_terminal_turn_event(
bridge_dir: Path,
method: str,
params: _JsonObject,
*,
forwarder_state: _CodexForwarderState | None = None,
) -> bool:
"""
Forward a terminal-observed Codex turn completion/failure event.
Handle a terminal-observed Codex turn completion/failure event.
:param client: HTTP client for Omnigent event posts.
:param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``.
:param bridge_dir: Native Codex bridge directory.
:param method: Codex method, e.g. ``"turn/completed"``.
:param params: Codex turn event params.
:returns: ``True`` when the terminal event belonged to the active
turn and was forwarded, ``False`` when it was stale.
:param forwarder_state: Optional connection state used to suppress a
terminal boundary whose standalone error was already surfaced.
:returns: ``True`` when the terminal event belonged to the active turn
and its lifecycle was handled, ``False`` when it was stale.
"""
terminal_turn_id = _terminal_turn_id_from_params(params)
if (
forwarder_state is not None
and terminal_turn_id is not None
and terminal_turn_id in forwarder_state.surfaced_terminal_error_turns
):
clear_active_turn_id_if_matches(bridge_dir, terminal_turn_id)
_logger.info(
"Codex forwarder suppressed terminal boundary after standalone error: "
"method=%s turn_id=%s",
method,
terminal_turn_id,
)
return True
edge = _terminal_turn_status_edge(bridge_dir, method, params)
if edge is None:
terminal_turn_id = _terminal_turn_id_from_params(params)
_logger.info(
"Codex forwarder ignored stale terminal turn event: method=%s turn_id=%s",
method,
tests/e2e_ui/chat/test_failure_error_card.py
+58
View File
@@ -17,12 +17,33 @@ from __future__ import annotations
import json
import httpx
import pytest
from playwright.sync_api import Page, Route, expect
from tests.e2e_ui.conftest import _server_state, seed_committed_turn
def _publish_native_status(
base_url: str,
session_id: str,
status: str,
*,
response_id: str,
output: str | None = None,
) -> None:
"""Publish the status payload used by native harness forwarders."""
data: dict[str, object] = {"status": status, "response_id": response_id}
if output is not None:
data["output"] = output
response = httpx.post(
f"{base_url}/v1/sessions/{session_id}/events",
json={"type": "external_session_status", "data": data},
timeout=10.0,
)
response.raise_for_status()
def _seed_error_item(session_id: str, *, code: str, message: str) -> None:
"""Append a committed ``error`` transcript item to the session's store.
@@ -84,6 +105,43 @@ def test_unclassified_failure_renders_english_headline_not_raw_code(
expect(pill).not_to_contain_text("Error · required_terminal_exited", timeout=15_000)
def test_live_native_failure_status_surfaces_each_turn(
page: Page,
seeded_session: tuple[str, str],
) -> None:
"""Each failed native response renders its status-carried error message."""
base_url, session_id = seeded_session
message = "You've hit your usage limit."
page.goto(f"{base_url}/c/{session_id}")
expect(page.get_by_role("textbox", name="Message the agent")).to_be_visible(timeout=15_000)
_publish_native_status(base_url, session_id, "running", response_id="codex_turn_1")
_publish_native_status(
base_url,
session_id,
"failed",
response_id="codex_turn_1",
output=message,
)
pills = page.get_by_test_id("error-pill")
expect(pills).to_have_count(1, timeout=15_000)
_publish_native_status(base_url, session_id, "running", response_id="codex_turn_2")
_publish_native_status(
base_url,
session_id,
"failed",
response_id="codex_turn_2",
output=message,
)
expect(pills).to_have_count(2, timeout=15_000)
second_pill = pills.nth(1)
second_pill.locator('button[aria-expanded="false"]').click()
expect(second_pill.get_by_test_id("error-message-content")).to_contain_text(message)
def test_persisted_failure_expands_retries_and_dismisses_locally(
page: Page,
seeded_session: tuple[str, str],
tests/test_codex_native_forwarder.py
+157
View File
@@ -1026,6 +1026,163 @@ def test_terminal_error_from_turn_prefers_turn_error_over_item() -> None:
assert error.message == "from turn.error"
def test_terminal_error_from_notification_reads_usage_limit() -> None:
"""The standalone Codex ``error`` notification carries the visible reason."""
error = fwd._terminal_error_from_notification(
{
"threadId": "thread_123",
"turnId": "turn_123",
"willRetry": False,
"error": {
"message": "You've hit your usage limit.",
"codexErrorInfo": "usageLimitExceeded",
},
}
)
assert error is not None
assert error.message == "You've hit your usage limit."
assert error.kind == fwd._CODEX_ERROR_KIND_GENERIC
@pytest.mark.asyncio
async def test_handle_event_surfaces_non_retrying_error_notification(tmp_path: Path) -> None:
"""A terminal standalone ``error`` notification reaches the session UI."""
client = _RecordingClient()
await fwd._handle_event(
client, # type: ignore[arg-type]
session_id="conv_x",
bridge_dir=tmp_path,
event={
"method": "error",
"params": {
"threadId": "thread_123",
"turnId": "turn_123",
"willRetry": False,
"error": {
"message": "You've hit your usage limit.",
"codexErrorInfo": "usageLimitExceeded",
},
},
},
usage_coalescer=fwd._SessionUsageCoalescer(client, "conv_x"), # type: ignore[arg-type]
elicitation_tracker=fwd._CodexElicitationTaskTracker(),
expected_thread_id="thread_123",
)
assert client.posts == [
(
"/v1/sessions/conv_x/events",
{
"type": "external_session_status",
"data": {
"status": "failed",
"response_id": "codex_turn_123",
"output": "You've hit your usage limit.",
},
},
)
]
@pytest.mark.asyncio
async def test_handle_event_ignores_retrying_error_notification(tmp_path: Path) -> None:
"""Retryable Codex errors remain internal while Codex retries the turn."""
client = _RecordingClient()
await fwd._handle_event(
client, # type: ignore[arg-type]
session_id="conv_x",
bridge_dir=tmp_path,
event={
"method": "error",
"params": {
"threadId": "thread_123",
"turnId": "turn_123",
"willRetry": True,
"error": {"message": "connection dropped"},
},
},
usage_coalescer=fwd._SessionUsageCoalescer(client, "conv_x"), # type: ignore[arg-type]
elicitation_tracker=fwd._CodexElicitationTaskTracker(),
expected_thread_id="thread_123",
)
assert client.posts == []
@pytest.mark.asyncio
async def test_handle_event_deduplicates_error_then_terminal_boundary(tmp_path: Path) -> None:
"""A standalone error owns the terminal status for its turn."""
write_bridge_state(
tmp_path,
CodexNativeBridgeState(
session_id="conv_x",
socket_path=str(tmp_path / "app-server.sock"),
thread_id="thread_123",
codex_home=str(tmp_path / "codex-home"),
active_turn_id="turn_123",
),
)
client = _RecordingClient()
usage_coalescer = fwd._SessionUsageCoalescer(client, "conv_x") # type: ignore[arg-type]
elicitation_tracker = fwd._CodexElicitationTaskTracker()
forwarder_state = fwd._CodexForwarderState()
error_event = {
"method": "error",
"params": {
"threadId": "thread_123",
"turnId": "turn_123",
"willRetry": False,
"error": {"message": "You've hit your usage limit."},
},
}
for event in (
error_event,
error_event,
{
"method": "turn/completed",
"params": {
"threadId": "thread_123",
"turn": {
"id": "turn_123",
"status": "completed",
"items": [{"type": "agentMessage", "text": ""}],
},
},
},
):
await fwd._handle_event(
client, # type: ignore[arg-type]
session_id="conv_x",
bridge_dir=tmp_path,
event=event,
usage_coalescer=usage_coalescer,
elicitation_tracker=elicitation_tracker,
expected_thread_id="thread_123",
forwarder_state=forwarder_state,
)
assert client.posts == [
(
"/v1/sessions/conv_x/events",
{
"type": "external_session_status",
"data": {
"status": "failed",
"response_id": "codex_turn_123",
"output": "You've hit your usage limit.",
},
},
)
]
state = read_bridge_state(tmp_path)
assert state is not None
assert state.active_turn_id is None
def test_terminal_error_from_turn_none_for_clean_turn() -> None:
"""A turn with no ``error`` object or item yields ``None`` (no false positives)."""
params = {
web/src/store/chatStore.test.ts
+34
View File
@@ -3701,6 +3701,40 @@ describe("chatStore — handleSessionEvent (session.* events)", () => {
expect(useChatStore.getState().sessionStatus).toBe("waiting");
});
it("surfaces one terminal error per native response", () => {
useChatStore.setState({ blocks: [] });
const error = {
code: "codex_turn_error",
message: "You've hit your usage limit.",
};
handleSessionEvent({
type: "session_status",
conversationId: "conv_abc",
status: "failed",
responseId: "codex_turn_1",
error,
});
handleSessionEvent({
type: "session_status",
conversationId: "conv_abc",
status: "failed",
responseId: "codex_turn_1",
error,
});
handleSessionEvent({
type: "session_status",
conversationId: "conv_abc",
status: "failed",
responseId: "codex_turn_2",
error,
});
const errors = useChatStore.getState().blocks.filter((block) => block.type === "error");
expect(errors).toHaveLength(2);
expect(errors.map((block) => block.ctx.responseId)).toEqual(["codex_turn_1", "codex_turn_2"]);
});
it("idle clears local streaming when no active response will send response_end", () => {
useChatStore.setState({
status: "streaming",
web/src/store/chatStore.ts
+25 -15
View File
@@ -5400,26 +5400,36 @@ export function handleSessionEvent(event: StreamEvent, streamConversationId?: st
patch.pendingUserMessages = [];
}
}
// Surface the error inline when the harness reports a terminal failure
// with a structured error payload (e.g. token expiration on startup).
// `response.failed` / `response.error` handle mid-turn failures, but
// startup failures only emit `session.status: failed` — nothing
// converts that into a visible ErrorBlock. Synthesize one here so the
// user sees the message without having to reload.
if (
event.status === "failed" &&
event.error != null &&
!s.blocks.some((b) => b.type === "error")
) {
// Surface terminal-native failures carried only by session status.
// Deduplicate repeated status edges for one response, but preserve the
// same failure on later turns so each rejected prompt has a visible error.
const statusError = event.error;
const hasMatchingStatusError =
statusError != null &&
s.blocks.some(
(block) =>
block.type === "error" &&
block.ctx.responseId === (event.responseId ?? "") &&
block.code === statusError.code &&
block.message === statusError.message,
);
if (event.status === "failed" && statusError != null && !hasMatchingStatusError) {
patch.blocks = [
...s.blocks,
{
type: "error",
ctx: { agent: null, depth: 0, turn: 0, timestamp: 0, responseId: "", itemId: null },
message: event.error.message,
ctx: {
agent: null,
depth: 0,
turn: 0,
timestamp: 0,
responseId: event.responseId ?? "",
itemId: null,
},
message: statusError.message,
source: "",
code: event.error.code,
...structuredErrorFields(event.error),
code: statusError.code,
...structuredErrorFields(statusError),
} satisfies ErrorBlock,
];
}