From d7473332e18d9040aacc9561ab20ce7460f786c6 Mon Sep 17 00:00:00 2001 From: Giulio Lovisotto Date: Wed, 21 Oct 2026 12:22:33 +0200 Subject: [PATCH 1/7] fix(codex): restore sandbox evidence and retries Signed-off-by: Giulio Lovisotto --- responses_api_agents/codex_agent/README.md | 6 +- responses_api_agents/codex_agent/app.py | 134 +++++++--- .../codex_agent/observability.py | 141 ++++++---- responses_api_agents/codex_agent/sandbox.py | 19 ++ .../codex_agent/sandbox_runner.py | 26 +- .../codex_agent/sandbox_transport.py | 80 ++++++ .../codex_agent/tests/test_observability.py | 58 +++- .../codex_agent/tests/test_sandbox_runner.py | 15 ++ .../tests/test_sandbox_sessions.py | 253 ++++++++++++++++++ .../tests/test_sandbox_transport.py | 148 ++++++++++ 10 files changed, 797 insertions(+), 83 deletions(-) create mode 100644 responses_api_agents/codex_agent/sandbox_transport.py create mode 100644 responses_api_agents/codex_agent/tests/test_sandbox_transport.py diff --git a/responses_api_agents/codex_agent/README.md b/responses_api_agents/codex_agent/README.md index 590e2b1cb8c..795b0f3b92b 100644 --- a/responses_api_agents/codex_agent/README.md +++ b/responses_api_agents/codex_agent/README.md @@ -157,11 +157,9 @@ Cleanup coordinates lifecycle; it is not a security boundary against hostile tas Responses and close observations preserve text, reasoning, tool events, partial output, CLI aggregate usage, and sandbox hostname/PID/version. Wall-time and the pinned CLI's explicit -model-output-limit event yield incomplete, potentially gradable work. Provider/API/runtime -failures raise execution errors and bypass verification. Only Resources determines reward. +model-output-limit event yield incomplete, potentially gradable work. Reported provider failures retain partial output and classified cause; the Environment Server decides grading and episode retries. Worker and cleanup failures remain execution errors. Only Resources determines reward. -CLI events omit exact model-call join IDs and can omit failed-call usage even after a recovered -retry. Missing/default-zero cache and reasoning details remain unknown, with observation gaps. +Native session transcripts supply tool-call IDs and outcomes. The sandbox forwarder retains request statuses and invocation correlation across retries; native cumulative usage supplements CLI totals when a failed turn omits them. Missing/default-zero cache and reasoning details remain unknown, with observation gaps. Compare CLI totals with captured model calls; do not interpret unavailable usage as zero. Known model-metadata/compaction warnings remain visible as gaps after clean completion or a gradable model/wall-time limit. diff --git a/responses_api_agents/codex_agent/app.py b/responses_api_agents/codex_agent/app.py index 151efc5c4eb..863373f0eec 100644 --- a/responses_api_agents/codex_agent/app.py +++ b/responses_api_agents/codex_agent/app.py @@ -69,7 +69,11 @@ from nemo_gym.sandbox.config import resolve_provider_config from nemo_gym.server_utils import get_global_config_dict, get_response_json, raise_for_status from nemo_gym.skills import stage_skills -from responses_api_agents.codex_agent.observability import read_codex_observations +from responses_api_agents.codex_agent.observability import ( + parse_codex_observations, + parse_codex_usage, + read_codex_observations, +) from responses_api_agents.codex_agent.sandbox import CodexSandboxSession from responses_api_agents.codex_agent.setup_codex import ensure_codex from responses_api_agents.codex_agent.transport import rate_limit_proxy @@ -83,6 +87,40 @@ ) +def _terminal_model_error( + events: list[tuple[float, dict]], + cleanup: dict | None, + failure: BaseException | None, + errors: list[str], +) -> str | None: + """Recognize a reported provider failure. Tested on Codex 0.144.4. + + Codex reports an unrecovered HTTP error with turn.failed and exit code 1. + Require clean supervision; a separate worker, cleanup or CLI error must not be mistaken for that provider outcome. + """ + if ( + failure is not None + or cleanup is None + or cleanup["return_code"] != 1 + or cleanup["error"] + or cleanup["timed_out"] + ): + return None + terminal = next( + (event for _, event in reversed(events) if event.get("type") in ("turn.completed", "turn.failed")), + {}, + ) + message = (terminal.get("error") or {}).get("message") + if ( + terminal.get("type") != "turn.failed" + or not isinstance(message, str) + or not message.strip() + or any(error != message for error in errors) + ): + return None + return message + + def _sandbox_prepare_command(workdir: str, directory: str, runtime: str) -> str: validate = """ from pathlib import Path @@ -460,11 +498,17 @@ class CodexAgent(SimpleResponsesAPIAgent): def classify_outcome(self, native_result: Mapping[str, Any]) -> AgentOutcome: """Read Codex 0.144.4's turn.failed error; tested on that pinned version. - Some errors retain only the provider body, without an HTTP status. Those + Prefer the sandbox forwarder's retained status. Without it, body-only errors remain unknown; a provider's arbitrary error.code is not an HTTP status. """ if native_result.get("type") != "turn.failed": return AgentOutcome(failure_kind="unknown") + if "upstream_status" in native_result: + status = native_result["upstream_status"] + if status is None: + return AgentOutcome(failure_kind=MODEL_TRANSPORT_ERROR) + if type(status) is int and 400 <= status <= 599: + return AgentOutcome(failure_kind=MODEL_REQUEST_ERROR, status_code=status) message = str((native_result.get("error") or {}).get("message") or "") match = re.search(r"\bunexpected status ([45]\d{2})\b", message) if match: @@ -598,7 +642,8 @@ async def _seed_agent_session_state(self, body: AgentSeedSessionRequest) -> Code f"Codex installer exited {installed.return_code}, command={install_command}, " f"error_type={installed.error_type}; stdout={installed.stdout}; stderr={installed.stderr}" ) - await sandbox.upload(Path(__file__).with_name("sandbox_runner.py"), f"{directory}/sandbox_runner.py") + for filename in ("sandbox_runner.py", "sandbox_transport.py"): + await sandbox.upload(Path(__file__).with_name(filename), f"{directory}/{filename}") except BaseException as error: try: await state.close(self.config.session_close_timeout_seconds) @@ -701,12 +746,14 @@ async def _sandbox_response( prompt: str, system: str, ) -> NeMoGymResponse: - config = self._build_config( - self._resolve_call_base_url(state.request.episode_id.capture_key), developer_instructions=system - ) + invocation_id = state.request.episode_id.capture_key + base_url = self._resolve_call_base_url(invocation_id) + config = self._build_config(base_url, developer_instructions=system) + config["model_providers"]["gym"]["http_headers"] = {"x-session-id": invocation_id} await state.upload_text("home/.codex/config.toml", toml_dumps(config)) timeout = resolve_agent_timeout(body.metadata, ceiling=self.config.timeout) command = self._build_command("-", state.session.workdir) + command.remove("--ephemeral") command[0:1] = [ f"{state.runtime}/node/bin/node", f"{state.runtime}/codex/node_modules/@openai/codex/bin/codex.js", @@ -714,6 +761,9 @@ async def _sandbox_response( payload = { "directory": state.session.session_dir, "command": command, + "model_base_url": base_url, + "invocation_id": invocation_id, + "timeout": timeout, "prompt": prompt, "cwd": state.session.workdir, "env": { @@ -808,7 +858,9 @@ async def _sandbox_response( and failure is None and (output_limited or (cleanup["timed_out"] and not errors)) ) - if not ((successful_exit and not errors) or gradable_limit): + terminal_error = None if output_limited else _terminal_model_error(events, cleanup, failure, errors) + reported_model_error = terminal_error is not None + if not ((successful_exit and not errors) or gradable_limit or reported_model_error): errors = [*startup_warnings, *compaction_warnings, *errors] startup_warnings = [] compaction_warnings = [] @@ -820,6 +872,20 @@ async def _sandbox_response( if cleanup and not completed and not cleanup["timed_out"] and not output_limited: error = error or "Codex ended without turn.completed" status = "failed" if error else "incomplete" if cleanup["timed_out"] or output_limited else "completed" + response_usage = NeMoGymResponseUsage( + input_tokens=usage["input_tokens"], + output_tokens=usage["output_tokens"], + total_tokens=usage["input_tokens"] + usage["output_tokens"], + input_tokens_details=NeMoGymResponseInputTokensDetails(cached_tokens=usage["cached_input_tokens"]), + output_tokens_details=NeMoGymResponseOutputTokensDetails(reasoning_tokens=usage["reasoning_tokens"]), + ) + if reported_model_error and not completed: + native_usage = parse_codex_usage(state.native_transcript) + if native_usage is not None: + response_usage = native_usage + elif output or not state.model_attempt_statuses or 200 in state.model_attempt_statuses: + # Prior work without a usage artifact is unknown, not a zero-token run. + response_usage = None conversation = [NeMoGymEasyInputMessage(role="user", content=prompt)] if system: conversation.insert(0, NeMoGymEasyInputMessage(role="system", content=system)) @@ -830,11 +896,17 @@ async def _sandbox_response( *(ObservationGap(code="model_metadata_fallback", detail=warning) for warning in startup_warnings), *(ObservationGap(code="compaction_accuracy_advisory", detail=warning) for warning in compaction_warnings), ] - for field, code in ( - ("cached_input_tokens", "cached_token_usage_unavailable"), - ("reasoning_tokens", "reasoning_token_usage_unavailable"), + for count, code in ( + ( + response_usage.input_tokens_details.cached_tokens if response_usage else None, + "cached_token_usage_unavailable", + ), + ( + response_usage.output_tokens_details.reasoning_tokens if response_usage else None, + "reasoning_token_usage_unavailable", + ), ): - if usage[field] is None: + if count is None: gaps.append(ObservationGap(code=code, detail="CLI detail counters are absent or defaulted to zero")) stream_errors = [event for _, event in events if event.get("type") == "error"] if not completed or stream_errors: @@ -906,27 +978,28 @@ async def _sandbox_response( AgentInvocation( invocation_id=state.request.episode_id.capture_key, status="incomplete" if failure else status, + error_type="provider_error" if reported_model_error else None, conversation=[*conversation, *output], ), *tools, ], gaps=gaps, ) + if state.native_transcript is not None: + native = parse_codex_observations( + state.native_transcript, + invocation_id, + returncode=cleanup["return_code"] if cleanup else None, + timed_out=bool(cleanup and cleanup["timed_out"]), + ) + native.records[0].status = "incomplete" if failure else status + if reported_model_error: + native.records[0].error_type = "provider_error" + native.gaps.extend(gap for gap in gaps if gap.code != "model_call_join_key_unavailable") + state.observations = native if failure is not None: raise failure - terminal = next( - (event for _, event in reversed(events) if event.get("type") in {"turn.completed", "turn.failed"}), {} - ) - reported_error = ( - cleanup is not None - and cleanup["return_code"] == 1 - and not cleanup["timed_out"] - and not cleanup["error"] - and terminal.get("type") == "turn.failed" - # Ignore already-recognized advisories when checking for unrelated execution errors. - and all(message == (terminal.get("error") or {}).get("message") for message in usage.get("errors") or []) - ) - if error and not reported_error: + if error and not reported_model_error: raise HTTPException(502, f"Codex execution failed: {error}") response = NeMoGymResponse( id=f"resp_{uuid4().hex}", @@ -936,17 +1009,11 @@ async def _sandbox_response( output=output, status=status, incomplete_details={"reason": "max_output_tokens"} if output_limited else None, - error={"code": "server_error", "message": error} if error else None, + error={"code": "server_error", "message": terminal_error} if reported_model_error else None, tool_choice=body.tool_choice, tools=body.tools, parallel_tool_calls=body.parallel_tool_calls, - usage=NeMoGymResponseUsage( - input_tokens=usage["input_tokens"], - output_tokens=usage["output_tokens"], - total_tokens=usage["input_tokens"] + usage["output_tokens"], - input_tokens_details=NeMoGymResponseInputTokensDetails(cached_tokens=usage["cached_input_tokens"]), - output_tokens_details=NeMoGymResponseOutputTokensDetails(reasoning_tokens=usage["reasoning_tokens"]), - ), + usage=response_usage, metadata={ "harness_execution": "sandbox", **({"harness_hostname": runtime.hostname, "harness_pid": str(runtime.pid)} if runtime else {}), @@ -955,6 +1022,9 @@ async def _sandbox_response( }, ) + terminal = next((event for _, event in reversed(events) if event.get("type") == "turn.failed"), {}) + if state.model_attempt_statuses: + terminal = {**terminal, "upstream_status": state.model_attempt_statuses[-1]} return self.report_outcome(response, terminal) def _resolve_call_base_url(self, rollout_id: Optional[str]) -> str: diff --git a/responses_api_agents/codex_agent/observability.py b/responses_api_agents/codex_agent/observability.py index 4d625a0f5de..45eb0b22fea 100644 --- a/responses_api_agents/codex_agent/observability.py +++ b/responses_api_agents/codex_agent/observability.py @@ -10,66 +10,117 @@ from pydantic import TypeAdapter, ValidationError -from nemo_gym.openai_utils import NeMoGymResponseInputItem +from nemo_gym.openai_utils import ( + NeMoGymResponseInputItem, + NeMoGymResponseInputTokensDetails, + NeMoGymResponseOutputTokensDetails, + NeMoGymResponseUsage, +) from nemo_gym.rollout_observability import AgentInvocation, AgentObservationBundle, ObservationGap, ToolCallObservation _ITEM = TypeAdapter(NeMoGymResponseInputItem) +def parse_codex_usage(transcript: str | None) -> NeMoGymResponseUsage | None: + """Read the latest native cumulative totals when a failed turn omits CLI usage. + + These totals already include cached input and reasoning output. As in the CLI + parser, zero detail counters cannot distinguish absent provider measurements. + """ + for line in reversed((transcript or "").splitlines()): + try: + event = json.loads(line) + if event["type"] != "event_msg" or event["payload"]["type"] != "token_count": + continue + usage = event["payload"]["info"]["total_token_usage"] + if any( + type(usage[key]) is not int or usage[key] < 0 + for key in ("input_tokens", "output_tokens", "total_tokens") + ): + continue + cached = usage.get("cached_input_tokens") + reasoning = usage.get("reasoning_output_tokens") + return NeMoGymResponseUsage( + input_tokens=usage["input_tokens"], + output_tokens=usage["output_tokens"], + total_tokens=usage["total_tokens"], + input_tokens_details=NeMoGymResponseInputTokensDetails( + cached_tokens=cached if type(cached) is int and cached > 0 else None + ), + output_tokens_details=NeMoGymResponseOutputTokensDetails( + reasoning_tokens=reasoning if type(reasoning) is int and reasoning > 0 else None + ), + ) + except (ValueError, TypeError, KeyError): + continue + return None + + def read_codex_observations( home: Path, invocation_id: str, *, returncode: int | None, timed_out: bool ) -> AgentObservationBundle: """Use native call IDs; never join CLI display item IDs to calls by ordering.""" + files = sorted(home.glob("sessions/**/*.jsonl")) + return parse_codex_observations( + "\n".join(line for path in files for line in path.read_text(errors="replace").splitlines()) if files else None, + invocation_id, + returncode=returncode, + timed_out=timed_out, + ) + + +def parse_codex_observations( + transcript: str | None, invocation_id: str, *, returncode: int | None, timed_out: bool +) -> AgentObservationBundle: + """Parse a retained native transcript from either local or sandbox execution.""" conversation = [] tools = {} gaps = [] - files = list(home.glob("sessions/**/*.jsonl")) - for path in files: - for line in path.read_text(errors="replace").splitlines(): - try: - event = json.loads(line) - if not isinstance(event, dict): - raise ValueError("expected a Codex CLI session record") - if event.get("type") != "response_item": - continue - item = dict(event["payload"]) - item.pop("internal_chat_message_metadata_passthrough", None) - parsed = _ITEM.validate_python(item) - conversation.append(parsed) - timestamp = datetime.fromisoformat(event["timestamp"].replace("Z", "+00:00")).timestamp() - except (ValueError, TypeError, KeyError, ValidationError): - gaps.append(ObservationGap(code="agent_artifact_record_unparseable", invocation_id=invocation_id)) + for line in (transcript or "").splitlines(): + try: + event = json.loads(line) + if not isinstance(event, dict): + raise ValueError("expected a Codex CLI session record") + if event.get("type") != "response_item": continue - if item.get("type") == "function_call": - tools[item["call_id"]] = ToolCallObservation( - invocation_id=invocation_id, - tool_call_id=item["call_id"], - tool_name=item["name"], - started_at=timestamp, - timing_source="artifact", - status="incomplete", - ) - elif item.get("type") == "function_call_output" and item.get("call_id") in tools: - tool = tools[item["call_id"]] - tool.completed_at = max(timestamp, tool.started_at) - tool.duration_ms = (tool.completed_at - tool.started_at) * 1000 - # Only the native execution envelope determines the exit code; - # command stdout can itself contain arbitrary exit-code text. - output = item.get("output") - envelope = output.split("\nOutput:\n", 1)[0] if isinstance(output, str) else "" - exit_code = re.search(r"^Process exited with code (-?\d+)$", envelope, re.MULTILINE) - if tool.tool_name == "exec_command" and exit_code: - tool.status = "completed" if int(exit_code[1]) == 0 else "failed" - tool.error_type = "nonzero_exit" if tool.status == "failed" else None - else: - tool.status = "unknown" - gaps.append( - ObservationGap( - code="tool_outcome_unavailable", invocation_id=invocation_id, detail=tool.tool_call_id - ) + item = dict(event["payload"]) + item.pop("internal_chat_message_metadata_passthrough", None) + parsed = _ITEM.validate_python(item) + conversation.append(parsed) + timestamp = datetime.fromisoformat(event["timestamp"].replace("Z", "+00:00")).timestamp() + except (ValueError, TypeError, KeyError, ValidationError): + gaps.append(ObservationGap(code="agent_artifact_record_unparseable", invocation_id=invocation_id)) + continue + if item.get("type") == "function_call": + tools[item["call_id"]] = ToolCallObservation( + invocation_id=invocation_id, + tool_call_id=item["call_id"], + tool_name=item["name"], + started_at=timestamp, + timing_source="artifact", + status="incomplete", + ) + elif item.get("type") == "function_call_output" and item.get("call_id") in tools: + tool = tools[item["call_id"]] + tool.completed_at = max(timestamp, tool.started_at) + tool.duration_ms = (tool.completed_at - tool.started_at) * 1000 + # Only the native execution envelope determines the exit code; + # command stdout can itself contain arbitrary exit-code text. + output = item.get("output") + envelope = output.split("\nOutput:\n", 1)[0] if isinstance(output, str) else "" + exit_code = re.search(r"^Process exited with code (-?\d+)$", envelope, re.MULTILINE) + if tool.tool_name == "exec_command" and exit_code: + tool.status = "completed" if int(exit_code[1]) == 0 else "failed" + tool.error_type = "nonzero_exit" if tool.status == "failed" else None + else: + tool.status = "unknown" + gaps.append( + ObservationGap( + code="tool_outcome_unavailable", invocation_id=invocation_id, detail=tool.tool_call_id ) - if not files: + ) + if transcript is None: gaps.append(ObservationGap(code="agent_artifact_unavailable", invocation_id=invocation_id)) return AgentObservationBundle( source="codex", diff --git a/responses_api_agents/codex_agent/sandbox.py b/responses_api_agents/codex_agent/sandbox.py index bdbab7f10e5..10c74eff46b 100644 --- a/responses_api_agents/codex_agent/sandbox.py +++ b/responses_api_agents/codex_agent/sandbox.py @@ -5,6 +5,7 @@ import asyncio import json import logging +import shlex from dataclasses import dataclass from pydantic import BaseModel, ConfigDict, JsonValue, ValidationError @@ -46,6 +47,8 @@ class CodexSandboxSession(AgentSessionState): task: asyncio.Task[NeMoGymResponse] | None = None runtime_info: HarnessProcessInfo | None = None observations: AgentObservationBundle | None = None + native_transcript: str | None = None + model_attempt_statuses: list[int | None] | None = None activation_request: NeMoGymResponseCreateParamsNonStreaming | None = None async def upload_json(self, name: str, payload: JsonValue) -> None: @@ -114,4 +117,20 @@ async def collect_artifacts(self) -> str: except ValueError: runtime_payload = None self.runtime_info = parse_runtime_info(runtime_payload) + try: + snapshot = await self.session.sandbox.exec( + f"python3 -I {shlex.quote(self.session.session_dir + '/sandbox_runner.py')} " + f"--snapshot {shlex.quote(self.session.session_dir)}", + timeout_s=10, + ) + if snapshot.return_code != 0 or snapshot.error_type: + raise RuntimeError("Codex native session snapshot failed") + self.native_transcript = await self.read_text("native_session.jsonl") + except Exception: + LOG.warning("Codex native session transcript unavailable", exc_info=True) + try: + attempts = await self.read_text("model_attempts.jsonl") + self.model_attempt_statuses = [json.loads(line)["status"] for line in attempts.splitlines()] + except Exception: + LOG.warning("Codex model attempt statuses unavailable", exc_info=True) return events diff --git a/responses_api_agents/codex_agent/sandbox_runner.py b/responses_api_agents/codex_agent/sandbox_runner.py index 4a93bd3f152..66338b42c60 100644 --- a/responses_api_agents/codex_agent/sandbox_runner.py +++ b/responses_api_agents/codex_agent/sandbox_runner.py @@ -26,11 +26,27 @@ def interrupt(*_): signal.signal(signal.SIGTERM, interrupt) with ExitStack() as stack: + command = list(params["command"]) + if params.get("model_base_url"): + # -I intentionally excludes the session directory from sys.path. + # Load only the adapter file staged beside this worker. + import runpy + + proxy = runpy.run_path(str(Path(__file__).with_name("sandbox_transport.py")))["rate_limit_proxy"] + base_url = stack.enter_context( + proxy( + params["model_base_url"], + params["invocation_id"], + directory / "model_attempts.jsonl", + timeout=params["timeout"], + ) + ) + command[2:2] = ["-c", "model_providers.gym.base_url=" + json.dumps(base_url)] stderr = stack.enter_context((directory / "stderr.log").open("wb")) stdin = stack.enter_context((directory / "prompt.txt").open("rb")) events = stack.enter_context((directory / "events.jsonl").open("w")) process = subprocess.Popen( - params["command"], + command, cwd=params["cwd"], env={ **{ @@ -97,6 +113,14 @@ def capture(line: bytes) -> None: def main() -> None: + if sys.argv[1] == "--snapshot": + directory = Path(sys.argv[2]) + files = sorted((directory / "home/.codex/sessions").glob("**/*.jsonl")) + if files: + (directory / "native_session.jsonl").write_text( + "\n".join(line for path in files for line in path.read_text(errors="replace").splitlines()) + ) + return params = json.loads(Path(sys.argv[1]).read_text()) raise SystemExit(run(params)) diff --git a/responses_api_agents/codex_agent/sandbox_transport.py b/responses_api_agents/codex_agent/sandbox_transport.py new file mode 100644 index 00000000000..fde56c222c0 --- /dev/null +++ b/responses_api_agents/codex_agent/sandbox_transport.py @@ -0,0 +1,80 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Sandbox-local forwarding for Codex's disabled native HTTP 429 retries. + +The task image only promises Python's standard library, not Gym or aiohttp. +Synchronous requests run in the forwarding thread and preserve streamed bodies. +""" + +import json +import math +from contextlib import contextmanager +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from threading import Thread +from time import sleep +from typing import Iterator +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + + +@contextmanager +def rate_limit_proxy(base_url: str, invocation_id: str, attempts_path: Path, *, timeout: float) -> Iterator[str]: + """Apply the local adapter's two-retry policy inside the task sandbox.""" + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *_: object) -> None: + pass + + def do_POST(self) -> None: + body = self.rfile.read(int(self.headers["Content-Length"])) + headers = { + key: value + for key, value in self.headers.items() + if key.lower() not in {"host", "content-length", "connection", "x-session-id"} + } + headers["x-session-id"] = invocation_id + for attempt in range(3): + try: + upstream = urlopen(Request(f"{base_url}/responses", data=body, headers=headers), timeout=timeout) + except HTTPError as error: + upstream = error + except (URLError, OSError) as error: + with attempts_path.open("a") as attempts: + attempts.write(json.dumps({"status": None, "error": str(error)}) + "\n") + self.send_error(502, "Upstream model connection failed") + return + with upstream: + with attempts_path.open("a") as attempts: + attempts.write(json.dumps({"status": upstream.status}) + "\n") + if upstream.status == 429 and attempt < 2: + upstream.read() + try: + delay = float(upstream.headers.get("Retry-After", 2**attempt)) + except ValueError: + delay = 2**attempt + if not math.isfinite(delay): + delay = 2**attempt + sleep(min(max(delay, 0), 30)) + continue + self.send_response(upstream.status) + for key, value in upstream.headers.items(): + if key.lower() not in {"content-length", "transfer-encoding", "connection"}: + self.send_header(key, value) + self.send_header("Connection", "close") + self.end_headers() + while chunk := upstream.read1(64 * 1024): + self.wfile.write(chunk) + self.wfile.flush() + self.close_connection = True + return + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}" + finally: + server.shutdown() + server.server_close() + thread.join() diff --git a/responses_api_agents/codex_agent/tests/test_observability.py b/responses_api_agents/codex_agent/tests/test_observability.py index 9fc01c7fd98..58f1366bb89 100644 --- a/responses_api_agents/codex_agent/tests/test_observability.py +++ b/responses_api_agents/codex_agent/tests/test_observability.py @@ -7,7 +7,7 @@ import pytest from nemo_gym.rollout_observability import AgentInvocation, ToolCallObservation -from responses_api_agents.codex_agent.observability import read_codex_observations +from responses_api_agents.codex_agent.observability import parse_codex_usage, read_codex_observations @pytest.mark.parametrize("exit_code", [0, 7]) @@ -58,3 +58,59 @@ def test_malformed_native_record_reports_gap_without_losing_invocation(tmp_path: assert bundle.records[0].invocation_id == "invocation" assert bundle.records[0].status == "failed" assert [gap.code for gap in bundle.gaps] == ["agent_artifact_record_unparseable"] + + +def test_multiple_native_session_files_preserve_record_boundaries(tmp_path: Path) -> None: + folder = tmp_path / "sessions" + folder.mkdir() + for index in range(2): + record = { + "type": "response_item", + "timestamp": f"2026-10-19T12:00:0{index}Z", + "payload": { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": f"message {index}"}], + }, + } + (folder / f"episode-{index}.jsonl").write_text(json.dumps(record) + "\n") + bundle = read_codex_observations(tmp_path, "invocation", returncode=0, timed_out=False) + assert len(bundle.records[0].conversation) == 2 + assert not bundle.gaps + + +def test_native_usage_uses_latest_cumulative_total_without_adding_details() -> None: + records = [] + for count in (20, 41): + records.append( + json.dumps( + { + "type": "event_msg", + "payload": { + "type": "token_count", + "info": { + "total_token_usage": { + "input_tokens": count, + "output_tokens": 8, + "total_tokens": count + 8, + "cached_input_tokens": 0, + "reasoning_output_tokens": 0, + } + }, + }, + } + ) + ) + records.extend(["malformed", json.dumps({"type": "event_msg", "payload": {"type": "token_count", "info": None}})]) + usage = parse_codex_usage("\n".join(records)) + assert usage.input_tokens == 41 + assert usage.output_tokens == 8 + assert usage.total_tokens == 49 + assert usage.input_tokens_details.cached_tokens is None + assert usage.output_tokens_details.reasoning_tokens is None + + +@pytest.mark.parametrize("usage", [None, {}, {"input_tokens": True, "output_tokens": 8, "total_tokens": 9}]) +def test_missing_or_invalid_native_usage_is_unknown(usage: object) -> None: + record = {"type": "event_msg", "payload": {"type": "token_count", "info": {"total_token_usage": usage}}} + assert parse_codex_usage(json.dumps(record)) is None diff --git a/responses_api_agents/codex_agent/tests/test_sandbox_runner.py b/responses_api_agents/codex_agent/tests/test_sandbox_runner.py index d6224fb2616..958268f7837 100644 --- a/responses_api_agents/codex_agent/tests/test_sandbox_runner.py +++ b/responses_api_agents/codex_agent/tests/test_sandbox_runner.py @@ -247,3 +247,18 @@ def test_python38_worker_captures_output_and_preserves_task_children(tmp_path): os.kill(int((tmp_path / "child.pid").read_text()), signal.SIGKILL) except ProcessLookupError: pass + + +def test_snapshot_retains_native_session_before_cleanup(tmp_path): + sessions = tmp_path / "home/.codex/sessions/2026/10/19" + sessions.mkdir(parents=True) + records = '{"type":"response_item","payload":{"call_id":"native-call"}}\n' + (sessions / "session.jsonl").write_text(records) + (sessions / "second.jsonl").write_text(records) + subprocess.run( + [sys.executable, "-I", sandbox_runner.__file__, "--snapshot", str(tmp_path)], + check=True, + capture_output=True, + timeout=10, + ) + assert (tmp_path / "native_session.jsonl").read_text().splitlines() == records.splitlines() * 2 diff --git a/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py b/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py index 16e011e2d60..287fec9cede 100644 --- a/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py +++ b/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py @@ -2027,3 +2027,256 @@ async def test_native_failure_without_http_status_reports_unknown(setup): assert response.error.message == message assert json.loads(response.metadata["ng_agent_outcome"]) == {"failure_kind": "unknown"} await agent.close_agent_session(request, AgentCloseSessionRequest(**close_body(session_id))) + + +def native_transcript(items): + return "\n".join( + json.dumps({"type": "response_item", "timestamp": f"2026-10-19T12:00:0{index}Z", "payload": item}) + for index, item in enumerate(items) + ) + + +def test_session_uses_native_call_identity_arguments_and_results(setup): + agent, sandbox = setup + native_output = "Process exited with code 7\nOutput:\noriginal result" + transcript = native_transcript( + [ + {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "task"}]}, + { + "type": "function_call", + "call_id": "provider-call", + "name": "exec_command", + "arguments": '{"cmd":"exit 7"}', + }, + {"type": "function_call_output", "call_id": "provider-call", "output": native_output}, + ] + ) + original_publish = sandbox.publish + + def publish(): + original_publish() + sandbox.files[f"{sandbox.directory}/native_session.jsonl"] = transcript + sandbox.files[f"{sandbox.directory}/model_attempts.jsonl"] = '{"status":200}\n' + + sandbox.publish = publish + with TestClient(agent.setup_webserver()) as client: + session_id = client.post("/v1/agent_sessions", json=seed().model_dump(mode="json")).json()["agent_session_id"] + response = client.post("/ng-rollout/codex-smoke-a2/v1/responses", json={"input": "task"}) + assert response.status_code == 200, response.text + payload = json.loads(sandbox.files[f"{sandbox.directory}/input.json"]) + assert "--ephemeral" not in payload["command"] + assert payload["invocation_id"] == "codex-smoke-a2" + assert payload["model_base_url"] == "http://model.example:9000/ng-rollout/codex-smoke-a2/v1" + config = tomllib.loads(sandbox.files[f"{sandbox.directory}/home/.codex/config.toml"]) + assert config["model_providers"]["gym"]["http_headers"] == {"x-session-id": "codex-smoke-a2"} + closed = client.post("/v1/agent_sessions/close", json=close_body(session_id)).json() + bundle = closed["agent_observations"] + invocation, tool = bundle["records"] + assert invocation["conversation"][1]["call_id"] == tool["tool_call_id"] == "provider-call" + assert invocation["conversation"][1]["arguments"] == '{"cmd":"exit 7"}' + assert invocation["conversation"][2]["output"] == native_output + assert tool["status"] == "failed" and tool["error_type"] == "nonzero_exit" + assert not any(gap["code"] == "model_call_join_key_unavailable" for gap in bundle["gaps"]) + + +@pytest.mark.parametrize( + "statuses,timed_out,returncode,worker_error,extra_error,accepted", + [ + ([400], False, 1, None, None, True), + ([200, 400], False, 1, None, None, True), + ([429, 400], False, 1, None, None, True), + ([400, 200], False, 1, None, None, True), + ([401], False, 1, None, None, True), + ([403], False, 1, None, None, True), + ([429, 429, 429], False, 1, None, None, True), + ([500], False, 1, None, None, True), + ([], False, 1, None, None, True), + ([429, None], False, 1, None, None, True), + ([400], True, 1, None, None, False), + ([400], False, None, None, None, False), + ([400], False, 0, None, None, False), + ([400], False, 7, None, None, False), + ([400], False, 1, "worker failed", None, False), + ([400], False, 1, None, "sandbox state lost", False), + ], +) +def test_terminal_model_error_preserves_provider_and_worker_error_boundary( + setup, statuses, timed_out, returncode, worker_error, extra_error, accepted +): + agent, sandbox = setup + sandbox.result.update(return_code=returncode, timed_out=timed_out, error=worker_error) + records = [ + {"type": "turn.started"}, + {"type": "turn.failed", "error": {"message": "model rejected request"}}, + ] + if extra_error: + records.insert(1, {"type": "item.completed", "item": {"type": "error", "message": extra_error}}) + sandbox.events = "\n".join(json.dumps([float(i), event]) for i, event in enumerate(records)) + original_publish = sandbox.publish + + def publish(): + original_publish() + sandbox.files[f"{sandbox.directory}/native_session.jsonl"] = native_transcript( + [{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "task"}]}] + ) + sandbox.files[f"{sandbox.directory}/model_attempts.jsonl"] = "\n".join( + json.dumps({"status": status}) for status in statuses + ) + + sandbox.publish = publish + with TestClient(agent.setup_webserver()) as client: + session_id = client.post("/v1/agent_sessions", json=seed().model_dump(mode="json")).json()["agent_session_id"] + response = client.post("/ng-rollout/codex-smoke-a2/v1/responses", json={"input": "task"}) + assert response.status_code == (200 if accepted else 502), response.text + if accepted: + body = response.json() + assert body["status"] == "failed" + assert body["error"] == {"code": "server_error", "message": "model rejected request"} + assert body["output"] == [] + outcome = json.loads(body["metadata"]["ng_agent_outcome"]) + if statuses and statuses[-1] is None: + assert outcome == {"failure_kind": "model_transport_error"} + elif statuses and statuses[-1] >= 400: + assert outcome == {"failure_kind": "model_request_error", "status_code": statuses[-1]} + else: + assert outcome == {"failure_kind": "unknown"} + if 200 in statuses or not statuses: + assert body["usage"] is None + else: + assert body["usage"]["total_tokens"] == 0 + closed = client.post("/v1/agent_sessions/close", json=close_body(session_id)).json() + invocation = closed["agent_observations"]["records"][0] + assert invocation["status"] == "failed" + if accepted: + assert invocation["error_type"] == "provider_error" + assert all(item.get("role") == "user" for item in invocation["conversation"]) + + +@pytest.mark.parametrize("usage_source", ["native", "cli", "missing"]) +def test_later_rejection_preserves_real_tool_output_and_available_usage(setup, usage_source): + agent, sandbox = setup + sandbox.result["return_code"] = 1 + # Pinned 0.144.4 emits this shape after a successful tool call then HTTP 400: + # no turn.completed, but the native token_count retains the completed call's usage. + rejection = '{"error":{"message":"controlled model failure","type":"conformance_error","code":"400"}}' + tool = {"id": "item_1", "type": "command_execution", "command": "printf retained", "status": "in_progress"} + records = [ + {"type": "turn.started"}, + {"type": "item.started", "item": tool}, + { + "type": "item.completed", + "item": tool | {"status": "completed", "exit_code": 0, "aggregated_output": "retained"}, + }, + {"type": "error", "message": rejection}, + {"type": "turn.failed", "error": {"message": rejection}}, + ] + if usage_source == "cli": + records.insert(-2, {"type": "turn.completed", "usage": {"input_tokens": 17, "output_tokens": 5}}) + sandbox.events = "\n".join(json.dumps([float(i), event]) for i, event in enumerate(records)) + native_output = "Process exited with code 0\nOutput:\nretained" + transcript = native_transcript( + [ + {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "task"}]}, + { + "type": "function_call", + "call_id": "provider-call", + "name": "exec_command", + "arguments": '{"cmd":"printf retained"}', + }, + {"type": "function_call_output", "call_id": "provider-call", "output": native_output}, + ] + ) + if usage_source != "missing": + transcript += "\n" + json.dumps( + { + "type": "event_msg", + "payload": { + "type": "token_count", + "info": { + "total_token_usage": { + "input_tokens": 20, + "output_tokens": 8, + "total_tokens": 28, + "cached_input_tokens": 3, + "reasoning_output_tokens": 2, + } + }, + }, + } + ) + original_publish = sandbox.publish + + def publish(): + original_publish() + sandbox.files[f"{sandbox.directory}/native_session.jsonl"] = transcript + sandbox.files[f"{sandbox.directory}/model_attempts.jsonl"] = '{"status":200}\n{"status":400}\n' + + sandbox.publish = publish + with TestClient(agent.setup_webserver()) as client: + session_id = client.post("/v1/agent_sessions", json=seed().model_dump(mode="json")).json()["agent_session_id"] + response = client.post("/ng-rollout/codex-smoke-a2/v1/responses", json={"input": "task"}) + assert response.status_code == 200, response.text + body = response.json() + assert body["status"] == "failed" + assert body["error"] == {"code": "server_error", "message": rejection} + assert [item["type"] for item in body["output"]] == ["function_call", "function_call_output"] + assert body["output"][1]["output"] == "retained" + if usage_source == "native": + assert body["usage"] == { + "input_tokens": 20, + "output_tokens": 8, + "total_tokens": 28, + "input_tokens_details": {"cached_tokens": 3}, + "output_tokens_details": {"reasoning_tokens": 2}, + } + elif usage_source == "cli": + assert body["usage"]["input_tokens"] == 17 + assert body["usage"]["output_tokens"] == 5 + assert body["usage"]["total_tokens"] == 22 + else: + assert body["usage"] is None + closed = client.post("/v1/agent_sessions/close", json=close_body(session_id)).json() + invocation, observed_tool = closed["agent_observations"]["records"] + assert invocation["status"] == "failed" + assert observed_tool["status"] == "completed" + assert invocation["conversation"][1]["call_id"] == observed_tool["tool_call_id"] == "provider-call" + assert invocation["conversation"][2]["output"] == native_output + + +async def test_activation_failure_takes_precedence_over_recorded_rejection(setup): + agent, sandbox = setup + state = await agent._seed_agent_session_state(seed()) + state.session.cleanup = sandbox.result | {"return_code": 1} + state.session.artifacts = json.dumps( + [1.0, {"type": "turn.failed", "error": {"message": "model rejected request"}}] + ) + state.model_attempt_statuses = [200, 400] + with patch.object(state, "execute", AsyncMock(side_effect=RuntimeError("sandbox connection lost"))): + with pytest.raises(RuntimeError, match="sandbox connection lost"): + await agent._sandbox_response( + state, NeMoGymResponseCreateParamsNonStreaming(input="task"), prompt="task", system="" + ) + assert state.observations.records[0].status == "incomplete" + + +@pytest.mark.asyncio +async def test_failed_native_snapshot_preserves_collected_cli_events(setup): + agent, sandbox = setup + state = await agent._seed_agent_session_state(seed()) + original_execute = sandbox.execute + + async def execute(command, **kwargs): + if "--snapshot" in command: + raise OSError("snapshot transport failed") + return await original_execute(command, **kwargs) + + sandbox.exec.side_effect = execute + try: + response = await agent._sandbox_response( + state, NeMoGymResponseCreateParamsNonStreaming(input="task"), prompt="task", system="" + ) + assert response.output[-1].content[0].text == "Fixed" + assert state.observations.records[0].conversation[-1].content[0].text == "Fixed" + assert any(gap.code == "model_call_join_key_unavailable" for gap in state.observations.gaps) + finally: + await state.close(1) diff --git a/responses_api_agents/codex_agent/tests/test_sandbox_transport.py b/responses_api_agents/codex_agent/tests/test_sandbox_transport.py new file mode 100644 index 00000000000..4df78f91783 --- /dev/null +++ b/responses_api_agents/codex_agent/tests/test_sandbox_transport.py @@ -0,0 +1,148 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import io +import json +import subprocess +import sys +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from threading import Thread +from unittest.mock import patch +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + +import pytest + +from responses_api_agents.codex_agent.sandbox_transport import rate_limit_proxy + + +@pytest.mark.parametrize("statuses", [[429, 429, 200], [429, 429, 429], [400], [500]]) +def test_sandbox_forwarder_retries_only_429_and_retains_every_attempt(tmp_path, statuses): + requests = [] + + class Upstream(BaseHTTPRequestHandler): + def log_message(self, *_): + pass + + def do_POST(self): + requests.append( + (self.path, self.headers["x-session-id"], self.rfile.read(int(self.headers["Content-Length"]))) + ) + self.send_response(statuses[len(requests) - 1]) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Retry-After", "0") + self.end_headers() + self.wfile.write(b"data: first\n\ndata: last\n\n") + + upstream = ThreadingHTTPServer(("127.0.0.1", 0), Upstream) + thread = Thread(target=upstream.serve_forever, daemon=True) + thread.start() + attempts_path = tmp_path / "attempts.jsonl" + try: + with rate_limit_proxy( + f"http://127.0.0.1:{upstream.server_port}/v1", "invocation", attempts_path, timeout=10 + ) as proxy: + try: + response = urlopen( + Request(f"{proxy}/responses", data=b'{"input":"task"}', headers={"x-session-id": "native-session"}) + ) + except HTTPError as error: + response = error + with response: + assert response.status == statuses[-1] + assert response.headers["Content-Type"] == "text/event-stream" + assert response.read() == b"data: first\n\ndata: last\n\n" + finally: + upstream.shutdown() + upstream.server_close() + thread.join() + assert requests == [("/v1/responses", "invocation", b'{"input":"task"}')] * len(statuses) + assert [json.loads(line)["status"] for line in attempts_path.read_text().splitlines()] == statuses + + +@pytest.mark.parametrize("retry_after", ["nan", "inf", "-inf", "invalid"]) +def test_invalid_retry_after_uses_finite_backoff(tmp_path, retry_after): + errors = [ + HTTPError("http://model/responses", 429, "retry", {"Retry-After": retry_after}, io.BytesIO(b"retry")) + for _ in range(3) + ] + with ( + patch("responses_api_agents.codex_agent.sandbox_transport.urlopen", side_effect=errors), + patch("responses_api_agents.codex_agent.sandbox_transport.sleep") as sleep, + ): + with rate_limit_proxy("http://model", "invocation", tmp_path / "attempts.jsonl", timeout=1) as proxy: + with pytest.raises(HTTPError) as reported: + urlopen(Request(f"{proxy}/responses", data=b"{}")) + assert reported.value.code == 429 + assert [call.args[0] for call in sleep.call_args_list] == [1, 2] + + +def test_connection_error_replaces_prior_http_status(tmp_path): + attempts = tmp_path / "attempts.jsonl" + attempts.write_text('{"status":429}\n') + with patch( + "responses_api_agents.codex_agent.sandbox_transport.urlopen", side_effect=URLError("connection refused") + ): + with rate_limit_proxy("http://model", "invocation", attempts, timeout=1) as proxy: + with pytest.raises(HTTPError) as reported: + urlopen(Request(f"{proxy}/responses", data=b"{}")) + assert reported.value.code == 502 + records = [json.loads(line) for line in attempts.read_text().splitlines()] + assert records[-1]["status"] is None + assert "connection refused" in records[-1]["error"] + + +def test_runner_installs_proxy_and_retains_retry(tmp_path): + from responses_api_agents.codex_agent import sandbox_runner + + requests = [] + + class Upstream(BaseHTTPRequestHandler): + def log_message(self, *_): + pass + + def do_POST(self): + requests.append((self.headers.get("x-session-id"), self.rfile.read(int(self.headers["Content-Length"])))) + self.send_response(429 if len(requests) == 1 else 200) + self.send_header("Retry-After", "0") + self.end_headers() + self.wfile.write(b"done") + + fake = tmp_path / "fake_codex.py" + fake.write_text("""import json, sys +from urllib.request import Request, urlopen +assert sys.argv[1] == '-c' +url = json.loads(sys.argv[2].split('=', 1)[1]) +with urlopen(Request(url + '/responses', data=b'{}')) as response: + print(json.dumps({'type': 'turn.completed', 'result': response.read().decode()})) +""") + upstream = ThreadingHTTPServer(("127.0.0.1", 0), Upstream) + thread = Thread(target=upstream.serve_forever, daemon=True) + thread.start() + params = { + "directory": str(tmp_path), + "cwd": str(tmp_path), + "prompt": "task", + "env": {}, + "command": [sys.executable, str(fake)], + "timeout": 5, + "model_base_url": f"http://127.0.0.1:{upstream.server_port}", + "invocation_id": "runner-invocation", + } + request_file = tmp_path / "input.json" + request_file.write_text(json.dumps(params)) + try: + result = subprocess.run( + [sys.executable, "-I", sandbox_runner.__file__, str(request_file)], capture_output=True, timeout=15 + ) + assert result.returncode == 0, result.stderr.decode() + finally: + upstream.shutdown() + upstream.server_close() + thread.join() + assert requests == [("runner-invocation", b"{}")] * 2 + assert [json.loads(line)["status"] for line in (tmp_path / "model_attempts.jsonl").read_text().splitlines()] == [ + 429, + 200, + ] + assert json.loads((tmp_path / "events.jsonl").read_text())[1]["result"] == "done" From 7980ee2f8690e39efb9d478a1f6964bdba841828 Mon Sep 17 00:00:00 2001 From: Giulio Lovisotto Date: Wed, 21 Oct 2026 15:10:00 +0200 Subject: [PATCH 2/7] fix(codex): retain response body read failures Signed-off-by: Giulio Lovisotto --- responses_api_agents/codex_agent/app.py | 15 +++-- responses_api_agents/codex_agent/sandbox.py | 4 +- .../codex_agent/sandbox_transport.py | 60 ++++++++++++------- .../tests/test_sandbox_sessions.py | 39 +++++++++++- .../tests/test_sandbox_transport.py | 56 +++++++++++++++++ 5 files changed, 142 insertions(+), 32 deletions(-) diff --git a/responses_api_agents/codex_agent/app.py b/responses_api_agents/codex_agent/app.py index 863373f0eec..589833957f4 100644 --- a/responses_api_agents/codex_agent/app.py +++ b/responses_api_agents/codex_agent/app.py @@ -498,13 +498,16 @@ class CodexAgent(SimpleResponsesAPIAgent): def classify_outcome(self, native_result: Mapping[str, Any]) -> AgentOutcome: """Read Codex 0.144.4's turn.failed error; tested on that pinned version. - Prefer the sandbox forwarder's retained status. Without it, body-only errors + Prefer the sandbox forwarder's retained status and body-read failures. Without them, errors remain unknown; a provider's arbitrary error.code is not an HTTP status. """ if native_result.get("type") != "turn.failed": return AgentOutcome(failure_kind="unknown") - if "upstream_status" in native_result: - status = native_result["upstream_status"] + if "upstream_attempt" in native_result: + attempt = native_result["upstream_attempt"] + if "transport_error" in attempt: + return AgentOutcome(failure_kind=MODEL_TRANSPORT_ERROR) + status = attempt["status"] if status is None: return AgentOutcome(failure_kind=MODEL_TRANSPORT_ERROR) if type(status) is int and 400 <= status <= 599: @@ -883,7 +886,7 @@ async def _sandbox_response( native_usage = parse_codex_usage(state.native_transcript) if native_usage is not None: response_usage = native_usage - elif output or not state.model_attempt_statuses or 200 in state.model_attempt_statuses: + elif output or not state.model_attempts or any(a["status"] == 200 for a in state.model_attempts): # Prior work without a usage artifact is unknown, not a zero-token run. response_usage = None conversation = [NeMoGymEasyInputMessage(role="user", content=prompt)] @@ -1023,8 +1026,8 @@ async def _sandbox_response( ) terminal = next((event for _, event in reversed(events) if event.get("type") == "turn.failed"), {}) - if state.model_attempt_statuses: - terminal = {**terminal, "upstream_status": state.model_attempt_statuses[-1]} + if state.model_attempts: + terminal = {**terminal, "upstream_attempt": state.model_attempts[-1]} return self.report_outcome(response, terminal) def _resolve_call_base_url(self, rollout_id: Optional[str]) -> str: diff --git a/responses_api_agents/codex_agent/sandbox.py b/responses_api_agents/codex_agent/sandbox.py index 10c74eff46b..d667fe374b6 100644 --- a/responses_api_agents/codex_agent/sandbox.py +++ b/responses_api_agents/codex_agent/sandbox.py @@ -48,7 +48,7 @@ class CodexSandboxSession(AgentSessionState): runtime_info: HarnessProcessInfo | None = None observations: AgentObservationBundle | None = None native_transcript: str | None = None - model_attempt_statuses: list[int | None] | None = None + model_attempts: list[dict[str, JsonValue]] | None = None activation_request: NeMoGymResponseCreateParamsNonStreaming | None = None async def upload_json(self, name: str, payload: JsonValue) -> None: @@ -130,7 +130,7 @@ async def collect_artifacts(self) -> str: LOG.warning("Codex native session transcript unavailable", exc_info=True) try: attempts = await self.read_text("model_attempts.jsonl") - self.model_attempt_statuses = [json.loads(line)["status"] for line in attempts.splitlines()] + self.model_attempts = [json.loads(line) for line in attempts.splitlines()] except Exception: LOG.warning("Codex model attempt statuses unavailable", exc_info=True) return events diff --git a/responses_api_agents/codex_agent/sandbox_transport.py b/responses_api_agents/codex_agent/sandbox_transport.py index fde56c222c0..821849f78c6 100644 --- a/responses_api_agents/codex_agent/sandbox_transport.py +++ b/responses_api_agents/codex_agent/sandbox_transport.py @@ -9,6 +9,7 @@ import json import math from contextlib import contextmanager +from http.client import HTTPException, IncompleteRead from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from threading import Thread @@ -45,29 +46,42 @@ def do_POST(self) -> None: self.send_error(502, "Upstream model connection failed") return with upstream: - with attempts_path.open("a") as attempts: - attempts.write(json.dumps({"status": upstream.status}) + "\n") - if upstream.status == 429 and attempt < 2: - upstream.read() - try: - delay = float(upstream.headers.get("Retry-After", 2**attempt)) - except ValueError: - delay = 2**attempt - if not math.isfinite(delay): - delay = 2**attempt - sleep(min(max(delay, 0), 30)) - continue - self.send_response(upstream.status) - for key, value in upstream.headers.items(): - if key.lower() not in {"content-length", "transfer-encoding", "connection"}: - self.send_header(key, value) - self.send_header("Connection", "close") - self.end_headers() - while chunk := upstream.read1(64 * 1024): - self.wfile.write(chunk) - self.wfile.flush() - self.close_connection = True - return + record = {"status": upstream.status} + try: + if upstream.status == 429 and attempt < 2: + upstream.read() + try: + delay = float(upstream.headers.get("Retry-After", 2**attempt)) + except ValueError: + delay = 2**attempt + if not math.isfinite(delay): + delay = 2**attempt + sleep(min(max(delay, 0), 30)) + continue + self.send_response(upstream.status) + for key, value in upstream.headers.items(): + if key.lower() not in {"content-length", "transfer-encoding", "connection"}: + self.send_header(key, value) + self.send_header("Connection", "close") + self.end_headers() + while True: + try: + chunk = upstream.read1(64 * 1024) + if not chunk and upstream.length: + raise IncompleteRead(b"", upstream.length) + except (OSError, HTTPException) as error: + record["transport_error"] = str(error) + break + if not chunk: + break + # A downstream write failure can be Codex cancellation, not a model failure. + self.wfile.write(chunk) + self.wfile.flush() + self.close_connection = True + return + finally: + with attempts_path.open("a") as attempts: + attempts.write(json.dumps(record) + "\n") server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) thread = Thread(target=server.serve_forever, daemon=True) diff --git a/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py b/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py index 287fec9cede..2d1540089b1 100644 --- a/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py +++ b/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py @@ -2250,7 +2250,7 @@ async def test_activation_failure_takes_precedence_over_recorded_rejection(setup state.session.artifacts = json.dumps( [1.0, {"type": "turn.failed", "error": {"message": "model rejected request"}}] ) - state.model_attempt_statuses = [200, 400] + state.model_attempts = [{"status": 200}, {"status": 400}] with patch.object(state, "execute", AsyncMock(side_effect=RuntimeError("sandbox connection lost"))): with pytest.raises(RuntimeError, match="sandbox connection lost"): await agent._sandbox_response( @@ -2280,3 +2280,40 @@ async def execute(command, **kwargs): assert any(gap.code == "model_call_join_key_unavailable" for gap in state.observations.gaps) finally: await state.close(1) + + +@pytest.mark.parametrize("transport_error", ["timed out", "IncompleteRead(0 bytes read, 20 more expected)", None]) +def test_post_header_failure_reaches_outcome_policy(setup, transport_error): + from environment_servers.single_agent_turn.app import agent_outcome_policy + from nemo_gym.openai_utils import NeMoGymResponse + + agent, sandbox = setup + sandbox.result["return_code"] = 1 + records = [ + {"type": "turn.started"}, + {"type": "turn.failed", "error": {"message": "stream disconnected before completion: server error"}}, + ] + sandbox.events = "\n".join(json.dumps([float(i), event]) for i, event in enumerate(records)) + original_publish = sandbox.publish + + def publish(): + original_publish() + attempt = {"status": 200} + if transport_error is not None: + attempt["transport_error"] = transport_error + sandbox.files[f"{sandbox.directory}/model_attempts.jsonl"] = json.dumps(attempt) + "\n" + + sandbox.publish = publish + with TestClient(agent.setup_webserver()) as client: + session_id = client.post("/v1/agent_sessions", json=seed().model_dump(mode="json")).json()["agent_session_id"] + response = client.post("/ng-rollout/codex-smoke-a2/v1/responses", json={"input": "task"}) + assert response.status_code == 200, response.text + body = response.json() + assert body["status"] == "failed" + assert body["usage"] is None + outcome = json.loads(body["metadata"]["ng_agent_outcome"]) + assert outcome == {"failure_kind": "model_transport_error" if transport_error is not None else "unknown"} + policy = agent_outcome_policy(NeMoGymResponse.model_validate(body)) + assert not policy.verifiable + assert policy.retryable is (transport_error is not None) + client.post("/v1/agent_sessions/close", json=close_body(session_id)).raise_for_status() diff --git a/responses_api_agents/codex_agent/tests/test_sandbox_transport.py b/responses_api_agents/codex_agent/tests/test_sandbox_transport.py index 4df78f91783..64d2ef78c6b 100644 --- a/responses_api_agents/codex_agent/tests/test_sandbox_transport.py +++ b/responses_api_agents/codex_agent/tests/test_sandbox_transport.py @@ -146,3 +146,59 @@ def do_POST(self): 200, ] assert json.loads((tmp_path / "events.jsonl").read_text())[1]["result"] == "done" + + +@pytest.mark.parametrize("fault", ["timeout", "truncated_chunk", "truncated_length", "server_error"]) +def test_body_failure_retains_http_status_and_transport_error(tmp_path, fault): + from threading import Event + + release = Event() + + class Upstream(BaseHTTPRequestHandler): + def log_message(self, *_): + pass + + def do_POST(self): + self.rfile.read(int(self.headers["Content-Length"])) + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + if fault == "truncated_chunk": + self.send_header("Transfer-Encoding", "chunked") + elif fault == "truncated_length": + self.send_header("Content-Length", "1000") + self.end_headers() + if fault == "timeout": + release.wait(2) + elif fault == "truncated_chunk": + self.wfile.write(b"20\r\ndata: partial") + elif fault == "truncated_length": + self.wfile.write(b"data: partial") + else: + self.wfile.write( + b'data: {"type":"response.failed","response":{"error":{"message":"invalid input"}}}\n\n' + ) + + upstream = ThreadingHTTPServer(("127.0.0.1", 0), Upstream) + thread = Thread(target=upstream.serve_forever, daemon=True) + thread.start() + attempts_path = tmp_path / "attempts.jsonl" + try: + with rate_limit_proxy( + f"http://127.0.0.1:{upstream.server_port}", "invocation", attempts_path, timeout=0.1 + ) as proxy: + with urlopen(Request(f"{proxy}/responses", data=b"{}"), timeout=5) as response: + assert response.status == 200 + body = response.read() + records = [json.loads(line) for line in attempts_path.read_text().splitlines()] + assert len(records) == 1 + assert records[0]["status"] == 200 + if fault == "server_error": + assert b"invalid input" in body + assert "transport_error" not in records[0] + else: + assert records[0]["transport_error"] + finally: + release.set() + upstream.shutdown() + upstream.server_close() + thread.join() From 615e3dbfb02689530028dde13847309de2a1dafb Mon Sep 17 00:00:00 2001 From: Giulio Lovisotto Date: Wed, 21 Oct 2026 16:48:05 +0200 Subject: [PATCH 3/7] fix(codex): retain native custom tool evidence Signed-off-by: Giulio Lovisotto --- nemo_gym/rollout_collection.py | 2 +- .../codex_agent/observability.py | 35 ++++++++--- .../codex_agent/tests/test_observability.py | 62 ++++++++++++++++++- .../tests/test_sandbox_sessions.py | 27 +++++--- tests/unit_tests/test_rollout_collection.py | 7 ++- 5 files changed, 113 insertions(+), 20 deletions(-) diff --git a/nemo_gym/rollout_collection.py b/nemo_gym/rollout_collection.py index 13ed72ec6ac..051938d0870 100644 --- a/nemo_gym/rollout_collection.py +++ b/nemo_gym/rollout_collection.py @@ -670,7 +670,7 @@ def _build_trajectory_record(row: dict[str, Any], result: dict[str, Any]) -> Tra (invocation.invocation_id, item.call_id): item.output for invocation in invocations for item in invocation.conversation - if getattr(item, "type", None) == "function_call_output" + if getattr(item, "type", None) in {"function_call_output", "custom_tool_call_output"} } positions = {(tool.invocation_id, tool.tool_call_id): index for index, tool in enumerate(tools)} for observed in observed_tools: diff --git a/responses_api_agents/codex_agent/observability.py b/responses_api_agents/codex_agent/observability.py index 45eb0b22fea..693fe62bc1d 100644 --- a/responses_api_agents/codex_agent/observability.py +++ b/responses_api_agents/codex_agent/observability.py @@ -76,12 +76,20 @@ def parse_codex_observations( """Parse a retained native transcript from either local or sandbox execution.""" conversation = [] tools = {} + patch_outcomes = {} gaps = [] for line in (transcript or "").splitlines(): try: event = json.loads(line) if not isinstance(event, dict): raise ValueError("expected a Codex CLI session record") + if event.get("type") == "event_msg": + payload = event.get("payload") or {} + if not isinstance(payload, dict): + raise ValueError("expected a native event payload") + if payload.get("type") == "patch_apply_end" and type(payload.get("success")) is bool: + patch_outcomes[payload["call_id"]] = payload["success"] + continue if event.get("type") != "response_item": continue item = dict(event["payload"]) @@ -92,7 +100,7 @@ def parse_codex_observations( except (ValueError, TypeError, KeyError, ValidationError): gaps.append(ObservationGap(code="agent_artifact_record_unparseable", invocation_id=invocation_id)) continue - if item.get("type") == "function_call": + if item.get("type") in {"function_call", "custom_tool_call"}: tools[item["call_id"]] = ToolCallObservation( invocation_id=invocation_id, tool_call_id=item["call_id"], @@ -101,7 +109,16 @@ def parse_codex_observations( timing_source="artifact", status="incomplete", ) - elif item.get("type") == "function_call_output" and item.get("call_id") in tools: + elif item.get("type") == "web_search_call": + tools[item["id"]] = ToolCallObservation( + invocation_id=invocation_id, + tool_call_id=item["id"], + tool_name="web_search", + status=item["status"] if item["status"] in {"completed", "failed"} else "incomplete", + completed_at=timestamp if item["status"] in {"completed", "failed"} else None, + timing_source="artifact", + ) + elif item.get("type") in {"function_call_output", "custom_tool_call_output"} and item.get("call_id") in tools: tool = tools[item["call_id"]] tool.completed_at = max(timestamp, tool.started_at) tool.duration_ms = (tool.completed_at - tool.started_at) * 1000 @@ -115,11 +132,15 @@ def parse_codex_observations( tool.error_type = "nonzero_exit" if tool.status == "failed" else None else: tool.status = "unknown" - gaps.append( - ObservationGap( - code="tool_outcome_unavailable", invocation_id=invocation_id, detail=tool.tool_call_id - ) - ) + # Patch success is in native execution events, not in the model-visible output text. + for tool in tools.values(): + if tool.tool_name == "apply_patch" and tool.tool_call_id in patch_outcomes: + tool.status = "completed" if patch_outcomes[tool.tool_call_id] else "failed" + tool.error_type = "tool_error" if tool.status == "failed" else None + if tool.status == "unknown": + gaps.append( + ObservationGap(code="tool_outcome_unavailable", invocation_id=invocation_id, detail=tool.tool_call_id) + ) if transcript is None: gaps.append(ObservationGap(code="agent_artifact_unavailable", invocation_id=invocation_id)) return AgentObservationBundle( diff --git a/responses_api_agents/codex_agent/tests/test_observability.py b/responses_api_agents/codex_agent/tests/test_observability.py index 58f1366bb89..725a38da1f5 100644 --- a/responses_api_agents/codex_agent/tests/test_observability.py +++ b/responses_api_agents/codex_agent/tests/test_observability.py @@ -7,7 +7,11 @@ import pytest from nemo_gym.rollout_observability import AgentInvocation, ToolCallObservation -from responses_api_agents.codex_agent.observability import parse_codex_usage, read_codex_observations +from responses_api_agents.codex_agent.observability import ( + parse_codex_observations, + parse_codex_usage, + read_codex_observations, +) @pytest.mark.parametrize("exit_code", [0, 7]) @@ -49,7 +53,10 @@ def test_missing_codex_session_record_is_explicit_and_preserves_failed_invocatio assert bundle.gaps[0].code == "agent_artifact_unavailable" -@pytest.mark.parametrize("record", [None, [], "unexpected", {"type": "response_item", "payload": {}}]) +@pytest.mark.parametrize( + "record", + [None, [], "unexpected", {"type": "response_item", "payload": {}}, {"type": "event_msg", "payload": ["invalid"]}], +) def test_malformed_native_record_reports_gap_without_losing_invocation(tmp_path: Path, record: object) -> None: folder = tmp_path / "sessions" folder.mkdir() @@ -114,3 +121,54 @@ def test_native_usage_uses_latest_cumulative_total_without_adding_details() -> N def test_missing_or_invalid_native_usage_is_unknown(usage: object) -> None: record = {"type": "event_msg", "payload": {"type": "token_count", "info": {"total_token_usage": usage}}} assert parse_codex_usage(json.dumps(record)) is None + + +@pytest.mark.parametrize("success", [True, False, None]) +def test_native_patch_retains_call_and_output_with_execution_outcome(success: bool | None) -> None: + items = [ + { + "type": "custom_tool_call", + "call_id": "patch-1", + "name": "apply_patch", + "input": "*** Begin Patch\n*** End Patch", + }, + {"type": "custom_tool_call_output", "call_id": "patch-1", "output": "Untrusted output claiming success"}, + ] + events = [ + {"type": "response_item", "timestamp": f"2026-10-21T10:00:0{index}Z", "payload": item} + for index, item in enumerate(items) + ] + if success is not None: + events.insert( + 1, {"type": "event_msg", "payload": {"type": "patch_apply_end", "call_id": "patch-1", "success": success}} + ) + bundle = parse_codex_observations("\n".join(map(json.dumps, events)), "invocation", returncode=0, timed_out=False) + invocation, tool = bundle.records + assert tool.tool_call_id == invocation.conversation[0].call_id == "patch-1" + assert invocation.conversation[0].input == items[0]["input"] + assert invocation.conversation[1].output == items[1]["output"] + assert tool.tool_name == "apply_patch" + assert tool.status == ("unknown" if success is None else "completed" if success else "failed") + assert tool.error_type == ("tool_error" if success is False else None) + assert tool.duration_ms == 1000 + assert [gap.code for gap in bundle.gaps] == (["tool_outcome_unavailable"] if success is None else []) + + +@pytest.mark.parametrize("status", ["completed", "failed", "searching"]) +def test_native_web_search_retains_identity_action_and_status(status: str) -> None: + item = { + "type": "web_search_call", + "id": "search-1", + "status": status, + "action": {"type": "search", "query": "Gym"}, + } + event = {"type": "response_item", "timestamp": "2026-10-21T10:00:00Z", "payload": item} + bundle = parse_codex_observations(json.dumps(event), "invocation", returncode=0, timed_out=False) + invocation, tool = bundle.records + assert tool.tool_call_id == invocation.conversation[0].id == "search-1" + assert invocation.conversation[0].action.query == "Gym" + assert tool.tool_name == "web_search" + assert tool.status == ("incomplete" if status == "searching" else status) + assert tool.started_at is None + assert (tool.completed_at is None) == (status == "searching") + assert not bundle.gaps diff --git a/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py b/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py index 2d1540089b1..3949fe046e7 100644 --- a/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py +++ b/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py @@ -2036,21 +2036,32 @@ def native_transcript(items): ) -def test_session_uses_native_call_identity_arguments_and_results(setup): +@pytest.mark.parametrize("custom", [False, True]) +def test_session_uses_native_call_identity_arguments_and_results(setup, custom): agent, sandbox = setup native_output = "Process exited with code 7\nOutput:\noriginal result" transcript = native_transcript( [ {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "task"}]}, { - "type": "function_call", + "type": "custom_tool_call" if custom else "function_call", "call_id": "provider-call", - "name": "exec_command", - "arguments": '{"cmd":"exit 7"}', + "name": "apply_patch" if custom else "exec_command", + ("input" if custom else "arguments"): "*** Begin Patch\n*** End Patch" + if custom + else '{"cmd":"exit 7"}', + }, + { + "type": "custom_tool_call_output" if custom else "function_call_output", + "call_id": "provider-call", + "output": native_output, }, - {"type": "function_call_output", "call_id": "provider-call", "output": native_output}, ] ) + if custom: + transcript += "\n" + json.dumps( + {"type": "event_msg", "payload": {"type": "patch_apply_end", "call_id": "provider-call", "success": False}} + ) original_publish = sandbox.publish def publish(): @@ -2073,9 +2084,11 @@ def publish(): bundle = closed["agent_observations"] invocation, tool = bundle["records"] assert invocation["conversation"][1]["call_id"] == tool["tool_call_id"] == "provider-call" - assert invocation["conversation"][1]["arguments"] == '{"cmd":"exit 7"}' + assert invocation["conversation"][1]["input" if custom else "arguments"] == ( + "*** Begin Patch\n*** End Patch" if custom else '{"cmd":"exit 7"}' + ) assert invocation["conversation"][2]["output"] == native_output - assert tool["status"] == "failed" and tool["error_type"] == "nonzero_exit" + assert tool["status"] == "failed" and tool["error_type"] == ("tool_error" if custom else "nonzero_exit") assert not any(gap["code"] == "model_call_join_key_unavailable" for gap in bundle["gaps"]) diff --git a/tests/unit_tests/test_rollout_collection.py b/tests/unit_tests/test_rollout_collection.py index 90c571cb0e8..760a52104aa 100644 --- a/tests/unit_tests/test_rollout_collection.py +++ b/tests/unit_tests/test_rollout_collection.py @@ -947,7 +947,8 @@ def test_rollout_request_debug_summary_compact(self) -> None: ROLLOUT_ID_KEY_NAME: "caller-capture-id", } - def test_build_trajectory_record_merges_all_evidence_sources(self) -> None: + @pytest.mark.parametrize("output_type", ["function_call_output", "custom_tool_call_output"]) + def test_build_trajectory_record_merges_all_evidence_sources(self, output_type: str) -> None: row = {TASK_INDEX_KEY_NAME: 2, ROLLOUT_INDEX_KEY_NAME: 3} result = { "ng_trajectory": { @@ -958,8 +959,8 @@ def test_build_trajectory_record_merges_all_evidence_sources(self) -> None: "invocation_id": "root", "status": "completed", "conversation": [ - {"type": "function_call_output", "call_id": "tool-1", "output": "result"}, - {"type": "function_call_output", "call_id": "observed-only", "output": "new"}, + {"type": output_type, "call_id": "tool-1", "output": "result"}, + {"type": output_type, "call_id": "observed-only", "output": "new"}, ], } ], From 7427cee4abfe0a6d79cdf0109d163008a2880635 Mon Sep 17 00:00:00 2001 From: Giulio Lovisotto Date: Wed, 21 Oct 2026 19:19:12 +0200 Subject: [PATCH 4/7] test(codex): align policy enum assertions Signed-off-by: Giulio Lovisotto --- .../codex_agent/tests/test_sandbox_sessions.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py b/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py index 3949fe046e7..47c22850e06 100644 --- a/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py +++ b/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py @@ -2297,7 +2297,7 @@ async def execute(command, **kwargs): @pytest.mark.parametrize("transport_error", ["timed out", "IncompleteRead(0 bytes read, 20 more expected)", None]) def test_post_header_failure_reaches_outcome_policy(setup, transport_error): - from environment_servers.single_agent_turn.app import agent_outcome_policy + from environment_servers.single_agent_turn.app import OutcomeAction, agent_outcome_policy from nemo_gym.openai_utils import NeMoGymResponse agent, sandbox = setup @@ -2327,6 +2327,5 @@ def publish(): outcome = json.loads(body["metadata"]["ng_agent_outcome"]) assert outcome == {"failure_kind": "model_transport_error" if transport_error is not None else "unknown"} policy = agent_outcome_policy(NeMoGymResponse.model_validate(body)) - assert not policy.verifiable - assert policy.retryable is (transport_error is not None) + assert policy == OutcomeAction.RETRY client.post("/v1/agent_sessions/close", json=close_body(session_id)).raise_for_status() From 55e9e92a3cb2ebc76a8a658dd216cb588ef84db9 Mon Sep 17 00:00:00 2001 From: Giulio Lovisotto Date: Wed, 21 Oct 2026 19:41:35 +0200 Subject: [PATCH 5/7] test(codex): cover native errors and proxy upload Signed-off-by: Giulio Lovisotto --- .../tests/test_sandbox_sessions.py | 62 +++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py b/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py index 47c22850e06..5b11f8916eb 100644 --- a/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py +++ b/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py @@ -183,6 +183,8 @@ def test_http_agent_session_flow_runs_codex_in_borrowed_sandbox(setup): assert f"{directory}/process_supervisor.py" not in sandbox.files installer = f"{directory}/install_codex_runtime.sh" assert installer in sandbox.files + for filename in ("sandbox_runner.py", "sandbox_transport.py"): + assert sandbox.files[f"{directory}/{filename}"] == (Path(__file__).parents[1] / filename).read_text() assert agent.config.resources_server is None assert not sandbox.launch.called assert not any(path.startswith("/app/") for path in sandbox.files) @@ -2329,3 +2331,63 @@ def publish(): policy = agent_outcome_policy(NeMoGymResponse.model_validate(body)) assert policy == OutcomeAction.RETRY client.post("/v1/agent_sessions/close", json=close_body(session_id)).raise_for_status() + + +# Native Display strings from codex-rs/protocol/src/error.rs at rust-v0.144.4. +@pytest.mark.parametrize( + "attempt,message,expected", + [ + ( + {"status": 400}, + '{"error":{"message":"invalid input","code":"bad_request"}}', + {"failure_kind": "model_request_error", "status_code": 400}, + ), + ( + {"status": 429}, + "exceeded retry limit, last status: 429 Too Many Requests", + {"failure_kind": "model_request_error", "status_code": 429}, + ), + ( + {"status": 500}, + "We're currently experiencing high demand, which may cause temporary errors.", + {"failure_kind": "model_request_error", "status_code": 500}, + ), + ( + {"status": 503}, + "Selected model is at capacity. Please try a different model.", + {"failure_kind": "model_request_error", "status_code": 503}, + ), + ({"status": None, "error": "timed out"}, "request timed out", {"failure_kind": "model_transport_error"}), + ( + {"status": 200, "transport_error": "timed out"}, + "request timed out", + {"failure_kind": "model_transport_error"}, + ), + ], +) +def test_native_errors_use_captured_attempt_facts(setup, attempt, message, expected): + from environment_servers.single_agent_turn.app import OutcomeAction, agent_outcome_policy + from nemo_gym.openai_utils import NeMoGymResponse + + agent, sandbox = setup + sandbox.result["return_code"] = 1 + sandbox.events = json.dumps([1.0, {"type": "turn.failed", "error": {"message": message}}]) + original_publish = sandbox.publish + + def publish(): + original_publish() + sandbox.files[f"{sandbox.directory}/model_attempts.jsonl"] = json.dumps(attempt) + "\n" + + sandbox.publish = publish + with TestClient(agent.setup_webserver()) as client: + session_id = client.post("/v1/agent_sessions", json=seed().model_dump(mode="json")).json()["agent_session_id"] + response = client.post("/ng-rollout/codex-smoke-a2/v1/responses", json={"input": "task"}) + assert response.status_code == 200 + result = NeMoGymResponse.model_validate(response.json()) + assert result.status == "failed" + assert result.error.message == message + assert json.loads(result.metadata["ng_agent_outcome"]) == expected + assert agent_outcome_policy(result) == ( + OutcomeAction.VERIFY if attempt["status"] == 400 else OutcomeAction.RETRY + ) + client.post("/v1/agent_sessions/close", json=close_body(session_id)).raise_for_status() From 4b492e4a31bf5a4fda56f367f9d82a1075882ea9 Mon Sep 17 00:00:00 2001 From: Giulio Lovisotto Date: Wed, 21 Oct 2026 19:47:01 +0200 Subject: [PATCH 6/7] fix(codex): report rate-limit response errors Signed-off-by: Giulio Lovisotto --- responses_api_agents/codex_agent/app.py | 15 +++++++++++---- .../codex_agent/tests/test_sandbox_sessions.py | 11 ++++++++++- 2 files changed, 21 insertions(+), 5 deletions(-) diff --git a/responses_api_agents/codex_agent/app.py b/responses_api_agents/codex_agent/app.py index 589833957f4..85c0595972a 100644 --- a/responses_api_agents/codex_agent/app.py +++ b/responses_api_agents/codex_agent/app.py @@ -1004,6 +1004,9 @@ async def _sandbox_response( raise failure if error and not reported_model_error: raise HTTPException(502, f"Codex execution failed: {error}") + terminal = next((event for _, event in reversed(events) if event.get("type") == "turn.failed"), {}) + if state.model_attempts: + terminal = {**terminal, "upstream_attempt": state.model_attempts[-1]} response = NeMoGymResponse( id=f"resp_{uuid4().hex}", created_at=int(time()), @@ -1012,7 +1015,14 @@ async def _sandbox_response( output=output, status=status, incomplete_details={"reason": "max_output_tokens"} if output_limited else None, - error={"code": "server_error", "message": terminal_error} if reported_model_error else None, + error={ + "code": "rate_limit_exceeded" + if self.classify_outcome(terminal).status_code == 429 + else "server_error", + "message": terminal_error, + } + if reported_model_error + else None, tool_choice=body.tool_choice, tools=body.tools, parallel_tool_calls=body.parallel_tool_calls, @@ -1025,9 +1035,6 @@ async def _sandbox_response( }, ) - terminal = next((event for _, event in reversed(events) if event.get("type") == "turn.failed"), {}) - if state.model_attempts: - terminal = {**terminal, "upstream_attempt": state.model_attempts[-1]} return self.report_outcome(response, terminal) def _resolve_call_base_url(self, rollout_id: Optional[str]) -> str: diff --git a/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py b/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py index 5b11f8916eb..d46f0706b45 100644 --- a/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py +++ b/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py @@ -2146,7 +2146,10 @@ def publish(): if accepted: body = response.json() assert body["status"] == "failed" - assert body["error"] == {"code": "server_error", "message": "model rejected request"} + assert body["error"] == { + "code": "rate_limit_exceeded" if statuses and statuses[-1] == 429 else "server_error", + "message": "model rejected request", + } assert body["output"] == [] outcome = json.loads(body["metadata"]["ng_agent_outcome"]) if statuses and statuses[-1] is None: @@ -2358,6 +2361,11 @@ def publish(): {"failure_kind": "model_request_error", "status_code": 503}, ), ({"status": None, "error": "timed out"}, "request timed out", {"failure_kind": "model_transport_error"}), + ( + {"status": 429, "transport_error": "timed out"}, + "request timed out", + {"failure_kind": "model_transport_error"}, + ), ( {"status": 200, "transport_error": "timed out"}, "request timed out", @@ -2386,6 +2394,7 @@ def publish(): result = NeMoGymResponse.model_validate(response.json()) assert result.status == "failed" assert result.error.message == message + assert result.error.code == ("rate_limit_exceeded" if expected.get("status_code") == 429 else "server_error") assert json.loads(result.metadata["ng_agent_outcome"]) == expected assert agent_outcome_policy(result) == ( OutcomeAction.VERIFY if attempt["status"] == 400 else OutcomeAction.RETRY From 41b114c5d61e7754986ec617da3537b0bfc73f8a Mon Sep 17 00:00:00 2001 From: Giulio Lovisotto Date: Thu, 22 Oct 2026 10:35:20 +0200 Subject: [PATCH 7/7] fix(codex): finalize and guard session evidence Signed-off-by: Giulio Lovisotto --- responses_api_agents/codex_agent/app.py | 139 +++++++++--------- .../codex_agent/observability.py | 76 +++++++--- .../codex_agent/sandbox_transport.py | 7 +- .../codex_agent/tests/test_observability.py | 23 +++ .../tests/test_sandbox_sessions.py | 11 ++ .../codex_agent/tests/test_streaming_cli.py | 111 +++++++++++++- 6 files changed, 274 insertions(+), 93 deletions(-) diff --git a/responses_api_agents/codex_agent/app.py b/responses_api_agents/codex_agent/app.py index 85c0595972a..748b165736a 100644 --- a/responses_api_agents/codex_agent/app.py +++ b/responses_api_agents/codex_agent/app.py @@ -70,6 +70,7 @@ from nemo_gym.server_utils import get_global_config_dict, get_response_json, raise_for_status from nemo_gym.skills import stage_skills from responses_api_agents.codex_agent.observability import ( + TESTED_CODEX_VERSION, parse_codex_observations, parse_codex_usage, read_codex_observations, @@ -81,6 +82,9 @@ LOG = logging.getLogger(__name__) _SANDBOX_SESSION_KEY = "nemo_gym_codex_sandbox_session" +_MODEL_CALL_JOIN_GAP = "model_call_join_key_unavailable" +_PROVIDER_ERROR = "provider_error" +_OUTPUT_LIMIT_ERROR = "stream disconnected before completion: Incomplete response returned, reason: max_output_tokens" _COMPACTION_ADVISORY = ( "Heads up: Long threads and multiple compactions can cause the model to be less accurate. " "Start a new thread when possible to keep threads small and targeted." @@ -581,6 +585,12 @@ async def _seed_agent_session_state(self, body: AgentSeedSessionRequest) -> Code raise HTTPException(422, "Codex in the task sandbox requires a sandbox-reachable Gym model_server") if not self.config.codex_version or not re.fullmatch(r"\d+\.\d+\.\d+", self.config.codex_version): raise HTTPException(422, "Codex in the task sandbox requires an exact codex_version, for example 0.144.4") + if self.config.codex_version != TESTED_CODEX_VERSION: + LOG.warning( + "Codex %s is untested: error classification and transcript parsing are validated on %s", + self.config.codex_version, + TESTED_CODEX_VERSION, + ) if self.config.cwd is not None or self.config.extra_config or self.config.openai_base_url is not None: raise HTTPException( 422, @@ -846,11 +856,7 @@ async def _sandbox_response( errors = usage.get("errors") or [] output_limited = ( bool(errors) - and all( - message - == "stream disconnected before completion: Incomplete response returned, reason: max_output_tokens" - for message in errors - ) + and all(message == _OUTPUT_LIMIT_ERROR for message in errors) and cleanup is not None and not cleanup["error"] and failure is None @@ -893,7 +899,6 @@ async def _sandbox_response( if system: conversation.insert(0, NeMoGymEasyInputMessage(role="system", content=system)) gaps = [ - ObservationGap(code="model_call_join_key_unavailable", detail="CLI events omit model response IDs"), ObservationGap(code="subagent_hierarchy_unavailable"), ObservationGap(code="compaction_observations_unavailable"), *(ObservationGap(code="model_metadata_fallback", detail=warning) for warning in startup_warnings), @@ -931,75 +936,77 @@ async def _sandbox_response( gaps.append(ObservationGap(code="agent_activation_interrupted", detail=type(failure).__name__)) if not events: gaps.append(ObservationGap(code="agent_transcript_unavailable")) - tools = [] - starts = {} - for observed_at, event in events: - item = event.get("item") or {} - if item.get("type") not in ( - "command_execution", - "mcp_tool_call", - "file_change", - "web_search", - "todo_list", - ): - continue - item_id = item.get("id") - if not item_id: - continue - if event.get("type") == "item.started": - starts[item_id] = observed_at - if event.get("type") == "item.completed": - start = starts.pop(item_id, None) + invocation_status = "incomplete" if failure else status + invocation_error = _PROVIDER_ERROR if reported_model_error else None + if state.native_transcript is not None: + state.observations = parse_codex_observations( + state.native_transcript, + invocation_id, + returncode=cleanup["return_code"] if cleanup else None, + timed_out=bool(cleanup and cleanup["timed_out"]), + status=invocation_status, + error_type=invocation_error, + gaps=gaps, + ) + else: + gaps.append(ObservationGap(code=_MODEL_CALL_JOIN_GAP, detail="CLI events omit model response IDs")) + tools = [] + starts = {} + for observed_at, event in events: + item = event.get("item") or {} + if item.get("type") not in ( + "command_execution", + "mcp_tool_call", + "file_change", + "web_search", + "todo_list", + ): + continue + item_id = item.get("id") + if not item_id: + continue + if event.get("type") == "item.started": + starts[item_id] = observed_at + if event.get("type") == "item.completed": + start = starts.pop(item_id, None) + tools.append( + ToolCallObservation( + invocation_id=state.request.episode_id.capture_key, + tool_call_id=item_id, + tool_name={"command_execution": "exec_command", "file_change": "apply_patch"}.get( + item["type"], item["type"] + ), + started_at=start, + completed_at=observed_at, + status="failed" if item.get("status") == "failed" else "completed", + timing_source="artifact", + ) + ) + if start is None: + gaps.append(ObservationGap(code="tool_start_unavailable", detail=item_id)) + for item_id, start in starts.items(): tools.append( ToolCallObservation( invocation_id=state.request.episode_id.capture_key, tool_call_id=item_id, - tool_name={"command_execution": "exec_command", "file_change": "apply_patch"}.get( - item["type"], item["type"] - ), started_at=start, - completed_at=observed_at, - status="failed" if item.get("status") == "failed" else "completed", + status="incomplete", timing_source="artifact", ) ) - if start is None: - gaps.append(ObservationGap(code="tool_start_unavailable", detail=item_id)) - for item_id, start in starts.items(): - tools.append( - ToolCallObservation( - invocation_id=state.request.episode_id.capture_key, - tool_call_id=item_id, - started_at=start, - status="incomplete", - timing_source="artifact", - ) - ) - state.observations = AgentObservationBundle( - source="codex", - records=[ - AgentInvocation( - invocation_id=state.request.episode_id.capture_key, - status="incomplete" if failure else status, - error_type="provider_error" if reported_model_error else None, - conversation=[*conversation, *output], - ), - *tools, - ], - gaps=gaps, - ) - if state.native_transcript is not None: - native = parse_codex_observations( - state.native_transcript, - invocation_id, - returncode=cleanup["return_code"] if cleanup else None, - timed_out=bool(cleanup and cleanup["timed_out"]), + state.observations = AgentObservationBundle( + source="codex", + records=[ + AgentInvocation( + invocation_id=state.request.episode_id.capture_key, + status=invocation_status, + error_type=invocation_error, + conversation=[*conversation, *output], + ), + *tools, + ], + gaps=gaps, ) - native.records[0].status = "incomplete" if failure else status - if reported_model_error: - native.records[0].error_type = "provider_error" - native.gaps.extend(gap for gap in gaps if gap.code != "model_call_join_key_unavailable") - state.observations = native if failure is not None: raise failure if error and not reported_model_error: diff --git a/responses_api_agents/codex_agent/observability.py b/responses_api_agents/codex_agent/observability.py index 693fe62bc1d..01b56b8269a 100644 --- a/responses_api_agents/codex_agent/observability.py +++ b/responses_api_agents/codex_agent/observability.py @@ -4,9 +4,11 @@ """Read exact tool identities and model-visible results from a Codex CLI session record.""" import json +import logging import re from datetime import datetime from pathlib import Path +from typing import Literal from pydantic import TypeAdapter, ValidationError @@ -20,6 +22,31 @@ _ITEM = TypeAdapter(NeMoGymResponseInputItem) +LOG = logging.getLogger(__name__) +TESTED_CODEX_VERSION = "0.144.4" + + +def _token_usage(payload: dict) -> NeMoGymResponseUsage | None: + # A token_count with info=null is a rate-limit update, not a new usage snapshot. + info = payload["info"] + if info is None: + return None + usage = info["total_token_usage"] + if any(type(usage[key]) is not int or usage[key] < 0 for key in ("input_tokens", "output_tokens", "total_tokens")): + raise ValueError("invalid Codex token totals") + cached = usage.get("cached_input_tokens") + reasoning = usage.get("reasoning_output_tokens") + return NeMoGymResponseUsage( + input_tokens=usage["input_tokens"], + output_tokens=usage["output_tokens"], + total_tokens=usage["total_tokens"], + input_tokens_details=NeMoGymResponseInputTokensDetails( + cached_tokens=cached if type(cached) is int and cached > 0 else None + ), + output_tokens_details=NeMoGymResponseOutputTokensDetails( + reasoning_tokens=reasoning if type(reasoning) is int and reasoning > 0 else None + ), + ) def parse_codex_usage(transcript: str | None) -> NeMoGymResponseUsage | None: @@ -29,30 +56,19 @@ def parse_codex_usage(transcript: str | None) -> NeMoGymResponseUsage | None: parser, zero detail counters cannot distinguish absent provider measurements. """ for line in reversed((transcript or "").splitlines()): + usage_record = False try: event = json.loads(line) if event["type"] != "event_msg" or event["payload"]["type"] != "token_count": continue - usage = event["payload"]["info"]["total_token_usage"] - if any( - type(usage[key]) is not int or usage[key] < 0 - for key in ("input_tokens", "output_tokens", "total_tokens") - ): - continue - cached = usage.get("cached_input_tokens") - reasoning = usage.get("reasoning_output_tokens") - return NeMoGymResponseUsage( - input_tokens=usage["input_tokens"], - output_tokens=usage["output_tokens"], - total_tokens=usage["total_tokens"], - input_tokens_details=NeMoGymResponseInputTokensDetails( - cached_tokens=cached if type(cached) is int and cached > 0 else None - ), - output_tokens_details=NeMoGymResponseOutputTokensDetails( - reasoning_tokens=reasoning if type(reasoning) is int and reasoning > 0 else None - ), - ) + usage_record = True + usage = _token_usage(event["payload"]) + if usage is not None: + return usage except (ValueError, TypeError, KeyError): + if usage_record: + LOG.warning("Codex token usage unavailable: unsupported token_count record") + return None # An older total would silently undercount the failed turn. continue return None @@ -71,13 +87,20 @@ def read_codex_observations( def parse_codex_observations( - transcript: str | None, invocation_id: str, *, returncode: int | None, timed_out: bool + transcript: str | None, + invocation_id: str, + *, + returncode: int | None, + timed_out: bool, + status: Literal["completed", "failed", "incomplete", "unknown"] | None = None, + error_type: str | None = None, + gaps: list[ObservationGap] | None = None, ) -> AgentObservationBundle: - """Parse a retained native transcript from either local or sandbox execution.""" + """Parse a transcript, using the caller's final outcome and gaps when supplied.""" conversation = [] tools = {} patch_outcomes = {} - gaps = [] + gaps = list(gaps or []) for line in (transcript or "").splitlines(): try: event = json.loads(line) @@ -87,10 +110,14 @@ def parse_codex_observations( payload = event.get("payload") or {} if not isinstance(payload, dict): raise ValueError("expected a native event payload") + if payload.get("type") == "token_count": + _token_usage(payload) if payload.get("type") == "patch_apply_end" and type(payload.get("success")) is bool: patch_outcomes[payload["call_id"]] = payload["success"] continue if event.get("type") != "response_item": + if event.get("type") not in {"session_meta", "turn_context", "compacted", "world_state"}: + raise ValueError("unsupported Codex session record type") continue item = dict(event["payload"]) item.pop("internal_chat_message_metadata_passthrough", None) @@ -143,13 +170,16 @@ def parse_codex_observations( ) if transcript is None: gaps.append(ObservationGap(code="agent_artifact_unavailable", invocation_id=invocation_id)) + if status is None: + status = "incomplete" if timed_out else "completed" if returncode == 0 else "failed" return AgentObservationBundle( source="codex", records=[ AgentInvocation( invocation_id=invocation_id, conversation=conversation, - status="incomplete" if timed_out else "completed" if returncode == 0 else "failed", + status=status, + error_type=error_type, ), *tools.values(), ], diff --git a/responses_api_agents/codex_agent/sandbox_transport.py b/responses_api_agents/codex_agent/sandbox_transport.py index 821849f78c6..743ca769390 100644 --- a/responses_api_agents/codex_agent/sandbox_transport.py +++ b/responses_api_agents/codex_agent/sandbox_transport.py @@ -19,6 +19,9 @@ from urllib.request import Request, urlopen +MAX_RATE_LIMIT_RETRIES = 2 + + @contextmanager def rate_limit_proxy(base_url: str, invocation_id: str, attempts_path: Path, *, timeout: float) -> Iterator[str]: """Apply the local adapter's two-retry policy inside the task sandbox.""" @@ -35,7 +38,7 @@ def do_POST(self) -> None: if key.lower() not in {"host", "content-length", "connection", "x-session-id"} } headers["x-session-id"] = invocation_id - for attempt in range(3): + for attempt in range(MAX_RATE_LIMIT_RETRIES + 1): try: upstream = urlopen(Request(f"{base_url}/responses", data=body, headers=headers), timeout=timeout) except HTTPError as error: @@ -48,7 +51,7 @@ def do_POST(self) -> None: with upstream: record = {"status": upstream.status} try: - if upstream.status == 429 and attempt < 2: + if upstream.status == 429 and attempt < MAX_RATE_LIMIT_RETRIES: upstream.read() try: delay = float(upstream.headers.get("Retry-After", 2**attempt)) diff --git a/responses_api_agents/codex_agent/tests/test_observability.py b/responses_api_agents/codex_agent/tests/test_observability.py index 725a38da1f5..8f08acc31bc 100644 --- a/responses_api_agents/codex_agent/tests/test_observability.py +++ b/responses_api_agents/codex_agent/tests/test_observability.py @@ -123,6 +123,29 @@ def test_missing_or_invalid_native_usage_is_unknown(usage: object) -> None: assert parse_codex_usage(json.dumps(record)) is None +@pytest.mark.parametrize("info", [{"renamed_totals": {}}, {"total_token_usage": {"input_tokens": "7"}}]) +def test_changed_usage_schema_does_not_fall_back_to_stale_totals(info, caplog) -> None: + previous = {"input_tokens": 7, "output_tokens": 3, "total_tokens": 10} + transcript = "\n".join( + json.dumps({"type": "event_msg", "payload": {"type": "token_count", "info": snapshot}}) + for snapshot in ({"total_token_usage": previous}, info) + ) + assert parse_codex_usage(transcript) is None + assert "unsupported token_count record" in caplog.text + bundle = parse_codex_observations(transcript, "invocation", returncode=1, timed_out=False) + assert [gap.code for gap in bundle.gaps] == ["agent_artifact_record_unparseable"] + + +def test_unrecognized_transcript_record_reports_gap() -> None: + bundle = parse_codex_observations( + json.dumps({"type": "renamed_response_item", "payload": {}}), + "invocation", + returncode=1, + timed_out=False, + ) + assert [gap.code for gap in bundle.gaps] == ["agent_artifact_record_unparseable"] + + @pytest.mark.parametrize("success", [True, False, None]) def test_native_patch_retains_call_and_output_with_execution_outcome(success: bool | None) -> None: items = [ diff --git a/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py b/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py index d46f0706b45..6ae5fbe9b74 100644 --- a/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py +++ b/responses_api_agents/codex_agent/tests/test_sandbox_sessions.py @@ -46,6 +46,17 @@ def seed() -> AgentSeedSessionRequest: ) +@pytest.mark.parametrize("version,warn", [("0.144.4", False), ("0.145.0", True)]) +def test_session_warns_about_untested_codex_version(setup, caplog, version, warn): + agent, sandbox = setup + agent.config.codex_version = version + with TestClient(agent.setup_webserver()) as client: + created = client.post("/v1/agent_sessions", json=seed().model_dump(mode="json")) + created.raise_for_status() + client.post("/v1/agent_sessions/close", json=close_body(created.json()["agent_session_id"])).raise_for_status() + assert ("error classification and transcript parsing are validated on 0.144.4" in caplog.text) == warn + + def events(*, stop_reason="stop") -> str: items = [ {"type": "item.completed", "item": {"id": "think-1", "type": "reasoning", "text": "Inspect the repository"}}, diff --git a/responses_api_agents/codex_agent/tests/test_streaming_cli.py b/responses_api_agents/codex_agent/tests/test_streaming_cli.py index 9e04058ade1..0a3a60e473b 100644 --- a/responses_api_agents/codex_agent/tests/test_streaming_cli.py +++ b/responses_api_agents/codex_agent/tests/test_streaming_cli.py @@ -18,14 +18,19 @@ from responses_api_agents.codex_agent.tests.test_sandbox_sessions import setup as setup -@pytest.mark.parametrize("recover", [False, True]) -def test_pinned_cli_incomplete_stream_retries_preserve_usage_gaps(tmp_path: Path, setup, recover: bool) -> None: +def pinned_cli() -> str: binary = os.environ.get("CODEX_TEST_BIN") if not binary or not Path(binary).is_file(): pytest.skip("Set CODEX_TEST_BIN to an existing Codex 0.144.4 executable; never auto-install for this test") version = subprocess.run([binary, "--version"], capture_output=True, text=True, errors="replace", timeout=10) if version.returncode or version.stdout.strip() != "codex-cli 0.144.4": pytest.skip("This boundary regression targets Codex 0.144.4") + return binary + + +@pytest.mark.parametrize("recover", [False, True]) +def test_pinned_cli_incomplete_stream_retries_preserve_usage_gaps(tmp_path: Path, setup, recover: bool) -> None: + binary = pinned_cli() reasoning = "Inspect arithmetic before editing." response = { "id": "resp_incomplete", @@ -145,3 +150,105 @@ def log_message(self, format: str, *args: object) -> None: assert native["usage"]["output_tokens_details"]["reasoning_tokens"] is None closed = client.post("/v1/agent_sessions/close", json=close_body(session_id)).json() assert "partial_model_usage_unavailable" in {gap["code"] for gap in closed["agent_observations"]["gaps"]} + + +def test_pinned_cli_failed_turn_retains_transcript_usage(tmp_path: Path, setup) -> None: + binary = pinned_cli() + agent, sandbox = setup + requests = [] + expected_usage = {"input_tokens": 41, "output_tokens": 8, "total_tokens": 49} + + class Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: + requests.append(json.loads(self.rfile.read(int(self.headers["Content-Length"])))) + if len(requests) > 1: + wire = b'{"error":{"message":"controlled rejection","type":"invalid_request_error"}}' + self.send_response(400) + self.send_header("Content-Type", "application/json") + else: + response = { + "id": "resp_tool", + "object": "response", + "created_at": 1, + "model": "gym-policy-model", + "status": "completed", + "output": [ + { + "type": "function_call", + "id": "fc_tool", + "call_id": "call_tool", + "name": "exec_command", + "arguments": json.dumps({"cmd": "printf usage_probe"}), + "status": "completed", + } + ], + "usage": expected_usage, + } + wire = "".join(synthesize_responses_sse(response)).encode() + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Content-Length", str(len(wire))) + self.end_headers() + self.wfile.write(wire) + + def log_message(self, format: str, *args: object) -> None: + pass + + server = HTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + home = tmp_path / "home" + home.mkdir() + config = agent._build_config(f"http://127.0.0.1:{server.server_port}/v1") + (home / "config.toml").write_text(toml_dumps(config)) + try: + result = subprocess.run( + [binary, "exec", "--json", "--skip-git-repo-check", "--cd", str(tmp_path), "--", "Run the shell check."], + env={ + "PATH": os.environ["PATH"], + "HOME": str(home), + "CODEX_HOME": str(home), + "OPENAI_API_KEY": "synthetic", + }, + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + errors="replace", + timeout=60, + ) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + assert result.returncode == 1, result.stdout + result.stderr + assert len(requests) == 2 + events = [json.loads(line) for line in result.stdout.splitlines()] + assert any(event["type"] == "turn.failed" for event in events) + assert not any(event["type"] == "turn.completed" for event in events) + transcripts = sorted((home / "sessions").rglob("*.jsonl")) + assert transcripts, "The real CLI must persist its transcript for the usage fallback" + transcript = "\n".join(path.read_text() for path in transcripts) + sandbox.events = "\n".join(json.dumps([float(i), event]) for i, event in enumerate(events)) + sandbox.result["return_code"] = result.returncode + original_publish = sandbox.publish + + def publish(): + original_publish() + sandbox.files[f"{sandbox.directory}/native_session.jsonl"] = transcript + sandbox.files[f"{sandbox.directory}/model_attempts.jsonl"] = '{"status":200}\n{"status":400}\n' + + sandbox.publish = publish + with TestClient(agent.setup_webserver()) as client: + session_id = client.post("/v1/agent_sessions", json=seed().model_dump(mode="json")).json()["agent_session_id"] + response = client.post("/ng-rollout/codex-smoke-a2/v1/responses", json={"input": "Run the shell check."}) + response.raise_for_status() + body = response.json() + assert body["status"] == "failed" + assert {key: body["usage"][key] for key in expected_usage} == expected_usage + assert any(item["type"] == "function_call_output" for item in body["output"]) + closed = client.post("/v1/agent_sessions/close", json=close_body(session_id)) + closed.raise_for_status() + observations = closed.json()["agent_observations"] + assert observations["records"][0]["status"] == "failed" + assert observations["records"][0]["error_type"] == "provider_error" + assert "agent_artifact_record_unparseable" not in {gap["code"] for gap in observations["gaps"]}