diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py index 0052db2ce8..29e7b814cc 100644 --- a/benchmark/edgebench/run.py +++ b/benchmark/edgebench/run.py @@ -71,22 +71,29 @@ def on_interrupt(signum, frame): def _result_status(*, interrupted, started, runtime_seconds, exit_code, timed_out=False, - execution_started=True): + execution_started=True, native_goal_outcome=None): if interrupted: return "cancelled" if not started: return "launch_failed" + if native_goal_outcome == "cancelled": + return "cancelled" + if native_goal_outcome in {"failed", "usage_limited", "unknown"}: + return "runner_failed" if not execution_started or runtime_seconds <= 0 or (not timed_out and exit_code != 0): return "runner_failed" return "terminal" -def _write_native_final_result(trial, result, *, status, agent, task, run_id, model, effort, feedback="native"): - """Match the native CLI's visualizer handoff after a completed run only.""" - if status != "terminal": +def _write_native_final_result(trial, result, *, status, agent, task, run_id, model, effort, + feedback="native", native_goal_outcome=None): + """Preserve native scores separately from the solver's terminal outcome.""" + if status != "terminal" and not (native_goal_outcome and result.runtime_seconds > 0): return final = dict(agent=agent, task=task, run_id=run_id, model=model, effort=effort, **result.to_dict()) + if native_goal_outcome is not None: + final.update(runtime_status=status, native_goal_outcome=native_goal_outcome) if feedback == "best-only": final["evaluation_coverage"] = "online_only" final["offline_scoring_complete"] = False @@ -413,19 +420,26 @@ def main(argv=None): # Native cancellation and swallowed Docker failures return runtime=0. # Observe the signal independently; never infer cancellation from output prose. execution = backend.execution_receipt or {} + native = backend.native_goal_receipt or {} + if args.worker == "native-goal" and not native.get("native_goal_outcome"): + native["native_goal_outcome"] = "unknown" status = _result_status(interrupted=was_interrupted, started=(trial / "started_at").is_file(), runtime_seconds=result.runtime_seconds, execution_started=execution.get("execution_started", False), exit_code=execution.get("exit_code"), + native_goal_outcome=native.get("native_goal_outcome"), timed_out=result.timed_out or execution.get("timed_out", False)) receipt.update(status=status, controller_elapsed_seconds=elapsed, timed_out=result.timed_out, runtime_seconds=result.runtime_seconds, best_score=result.best_score, total_rounds=result.total_rounds) receipt["execution"] = execution + if native: + receipt.update(native) _write_native_final_result(trial, result, status=status, agent=agent.name, task=task.task_id, run_id=args.run_id, - model=args.model, effort=args.effort, feedback=args.feedback) + model=args.model, effort=args.effort, feedback=args.feedback, + native_goal_outcome=native.get("native_goal_outcome")) receipt_path.write_text(json.dumps(receipt, indent=2)) print(json.dumps(receipt)) return 0 if status == "terminal" else 130 if status == "cancelled" else 1 diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 3b887e55be..978c9e4539 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -79,6 +79,29 @@ Native Goal continuation stays with Codex; blocked Goals are not automatically unblocked. Plain exec versus Goal app-server also changes transport; it does not isolate the continuation effect alone. +Native worker receipts retain both the latest Turn status and Goal status. +`native_goal_outcome` distinguishes `complete`, `blocked`, `paused`, +`budget_exhausted`, `usage_limited`, `cancelled`, `failed`, and `unknown`. +A failed Turn takes precedence over a blocked or complete Goal and exits nonzero; +intentional blocking/pausing and normal budget exhaustion do not become execution +failures. `ok` indicates a handled stop, never task acceptance. Only completion +may enter the explicitly enabled continuation option below. + +Terminal errors retain a digest, an allowlisted native error variant and an +optional structured HTTP status. Unrecognized errors remain unknown; message +text is not parsed into a category or copied into the receipt. The existing +`error_event_count` counts error notifications, not requests: a terminal event +that repeats an error does not increment it. A later native continuation clears +the terminal-error fields and may complete normally. + +The EdgeBench adapter carries this outcome through artifact collection into +`runtime-receipt.json`. If a native Goal fails after execution starts, its +`final_result.json` still retains native scores, with explicit `runtime_status` +and `native_goal_outcome`; the presence of that file alone is not successful +completion or result qualification. Ordinary non-native profile handling is +unchanged. Adopt this behavior only for newly pinned attempts; frozen histories +retain their original receipts and require explicit post-run classification. + The optional `native_goal_continue: true` loop retains the original task's requirements and evaluation rules after completion. Its continuation prompt no longer requires a candidate to improve on an accepted best before submission. diff --git a/benchmark/runtime/sforge_backend.py b/benchmark/runtime/sforge_backend.py index aa301db308..3a14a258dd 100644 --- a/benchmark/runtime/sforge_backend.py +++ b/benchmark/runtime/sforge_backend.py @@ -24,6 +24,7 @@ def __init__(self, *, log_dir: Path, logger, oauth_proxy: bool = False, self.execution_command = None self.execution_start_marker = None self.execution_receipt = None + self.native_goal_receipt = None self.auth_ips = [] if oauth_proxy else resolve_hostname("auth.openai.com", logger) if not oauth_proxy and not self.auth_ips: raise RuntimeError("Cannot resolve Codex OAuth endpoint") @@ -124,6 +125,23 @@ def cleanup_container(self, handle, logger=None): observations.append({"artifact": name, "collected": False, "error_kind": type(error).__name__}) (self.log_dir / "artifact-collection.json").write_text(json.dumps(observations)) + # A native-Goal worker owns exactly one wake (including opt-in + # continuation Goals). Read its retained receipt after collection; + # don't infer completion from a shell exit or solver output prose. + wakes = self.log_dir / "runtime" / "agent" / "wakes" + native = [] + for path in wakes.glob("*/receipt.json"): + try: + receipt = json.loads(path.read_text()) + if isinstance(receipt, dict) and receipt.get("mode") in {"native-goal", "loopx-goal"}: + native.append(receipt) + except (OSError, ValueError): + continue # Collection gaps retain the transport failure. + if len(native) == 1: + self.native_goal_receipt = {key: native[0][key] for key in + ("native_goal_outcome", "native_goal_rounds", "native_goal") if key in native[0]} + elif native: + self.native_goal_receipt = {"native_goal_outcome": "unknown"} super().cleanup_container(handle, logger) if self.feedback is not None: # Native cleanup logs some failures instead of raising. Verify the diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py index df67e685f3..33aa49869e 100644 --- a/benchmark/runtime/worker.py +++ b/benchmark/runtime/worker.py @@ -266,8 +266,8 @@ def native_goal_rounds(run_round, *, budget_seconds, continue_after_complete, cl """Drive one native Goal and, when opted in, continuation Goals until the budget ends. ``run_round(objective_suffix, timeout_sec)`` runs one Goal to a terminal status and - returns that status. Returns ``(rounds, outcome)`` where outcome is ``complete``, - ``budget_exhausted`` (a Goal was still active at the budget edge) or ``stopped``. + returns its typed outcome. Only ``complete`` may start an explicitly opted-in + continuation. Failure, blocking, cancellation and unknown outcomes stop. """ from loopx.capabilities.benchmark_toolkit.native_codex_goal import NativeGoalDeadlineExceeded @@ -284,7 +284,7 @@ def native_goal_rounds(run_round, *, budget_seconds, continue_after_complete, cl return rounds + 1, "budget_exhausted" rounds += 1 if status != "complete" or not continue_after_complete: - return rounds, ("complete" if status == "complete" else "stopped") + return rounds, status remaining = None if deadline is None else deadline - clock() if remaining is not None and remaining <= reserve_seconds: return rounds, "complete" @@ -304,6 +304,7 @@ def run_native_goal( NativeGoalConfig, StdioNativeGoalTransport, compact_native_goal_receipt, + native_goal_outcome, refresh_native_goal_status, run_native_goal_until_terminal, ) @@ -339,7 +340,9 @@ def run_round(suffix, timeout_sec): transport, NativeGoalConfig(objective=body + suffix, **base), timeout_sec=timeout_sec, on_turn_started=observed.append, ) - return turn.post_goal_status or refresh_native_goal_status(transport, turn) + if not turn.post_goal_status: + refresh_native_goal_status(transport, turn) + return native_goal_outcome(turn) try: rounds, outcome = native_goal_rounds( @@ -352,10 +355,10 @@ def run_round(suffix, timeout_sec): # Reaching the budget with an active Goal is normal exhaustion for # a native-Goal worker, not a runner failure. receipt["budget_exhausted"] = True - receipt["ok"] = True + receipt["ok"] = outcome in {"complete", "blocked", "paused", "budget_exhausted"} finally: if observed: - receipt["native_goal"] = compact_native_goal_receipt(observed[0]) + receipt["native_goal"] = compact_native_goal_receipt(observed[-1]) def native_command(env, execution, stage, wake, session_wake): @@ -656,6 +659,8 @@ def cancelled(signum, frame): resource.setrlimit(resource.RLIMIT_CORE, (0, hard)) receipt = run_once(dict(os.environ)) print(json.dumps(receipt)) + if receipt.get("native_goal_outcome") == "cancelled": + return 130 if receipt.get("budget_exhausted") and receipt["mode"] in {"heartbeat", "turn"}: from benchmark.runtime.scheduler import BUDGET_EXHAUSTED_EXIT_CODE diff --git a/benchmark/tests/test_native_goal_outcomes.py b/benchmark/tests/test_native_goal_outcomes.py new file mode 100644 index 0000000000..d10894de1f --- /dev/null +++ b/benchmark/tests/test_native_goal_outcomes.py @@ -0,0 +1,269 @@ +"""Native terminal evidence must survive the worker and scoring handoff.""" + +import json +import logging +import os +import subprocess +import sys +import uuid +from pathlib import Path, PurePosixPath +from types import SimpleNamespace + +import pytest + +from benchmark.tests.test_shared_codex_runtime import worker_env + + +def native_worker(tmp_path, goal_status, turn_status, *, http_status=None): + env = worker_env(tmp_path) + env.update(LOOPX_EXECUTION_MODE="native-goal", LOOPX_TURN_TIMEOUT_SECONDS="10") + error = ({"message": "PRIVATE-SENTINEL", "codexErrorInfo": { + "httpConnectionFailed": {"httpStatusCode": http_status}}} if http_status else None) + terminal = {"method": "turn/completed", "params": {"threadId": "t1", + "turn": {"id": "r1", "status": turn_status, "error": error}}} + Path(env["CODEX_BIN"]).write_text(f"#!{sys.executable}\n" + f''' +import json, sys +objective = "" +started = False +for line in sys.stdin: + request = json.loads(line) + method = request["method"] + params = request.get("params", {{}}) + if method == "initialized": + continue + if method == "initialize": + result = {{}} + elif method == "thread/start": + result = {{"thread": {{"id": "t1"}}}} + elif method == "thread/goal/set": + objective = params["objective"] + result = {{"goal": {{"threadId": "t1"}}}} + elif method == "thread/goal/get": + result = {{"goal": {{"threadId": "t1", "objective": objective, + "status": {goal_status!r} if started else "active"}}}} + elif method == "turn/start": + assert not started, "unexpected external continuation" + started = True + result = {{"turn": {{"id": "r1", "status": "inProgress"}}}} + else: + raise AssertionError(method) + print(json.dumps({{"id": request["id"], "result": result}}), flush=True) + if method == "turn/start": + print(json.dumps({{"method": "turn/started", "params": {{"threadId": "t1", + "turn": {{"id": "r1", "status": "inProgress"}}}}}}), flush=True) + if {error!r}: + print(json.dumps({{"method": "error", "params": {{"threadId": "t1", + "turnId": "r1", "error": {error!r}}}}}), flush=True) + print(json.dumps({terminal!r}), flush=True) +''') + result = subprocess.run([sys.executable, "-m", "benchmark.runtime.worker"], + env=env, capture_output=True, text=True, timeout=20) + assert result.stdout, result.stderr + receipt = json.loads(result.stdout) + persisted = list((tmp_path / "logs/wakes").glob("*/receipt.json")) + assert len(persisted) == 1 + assert json.loads(persisted[0].read_text()) == receipt + assert "PRIVATE-SENTINEL" not in json.dumps(receipt) + return result, receipt + + +@pytest.mark.parametrize("goal,turn,outcome,exit_code", [ + ("complete", "completed", "complete", 0), + ("blocked", "completed", "blocked", 0), + ("blocked", "failed", "failed", 1), + ("complete", "failed", "failed", 1), + ("paused", "interrupted", "cancelled", 130), + ("paused", "completed", "paused", 0), + ("budgetLimited", "completed", "budget_exhausted", 0), + ("usageLimited", "completed", "usage_limited", 1), + ("future-status", "completed", "unknown", 1), +]) +def test_native_terminal_outcome_survives_real_worker_process(tmp_path, goal, turn, outcome, exit_code): + result, receipt = native_worker(tmp_path, goal, turn) + assert receipt["native_goal_outcome"] == outcome + assert result.returncode == exit_code + assert receipt["ok"] is (exit_code == 0) + assert receipt["native_goal"]["post_goal_status"] == goal + + +@pytest.mark.parametrize("http_status", [401, 429, 503]) +def test_terminal_error_is_failed_without_counting_printed_error_twice(tmp_path, http_status): + result, receipt = native_worker(tmp_path, "blocked", "failed", http_status=http_status) + assert result.returncode == 1 + assert receipt["native_goal_outcome"] == "failed" + native = receipt["native_goal"] + assert native["turn_status"] == "failed" + assert native["error_event_count"] == 1 # terminal error is evidence, not a second request + assert native["terminal_error_http_status"] == http_status + assert native["terminal_error_signature"] + + +def test_jsonl_terminal_error_and_later_recovery_do_not_leak_or_poison_completion(): + from benchmark.tests.test_native_codex_goal import FakeTransport, _config + from loopx.capabilities.benchmark_toolkit.native_codex_goal import ( + compact_native_goal_receipt, native_goal_outcome, observe_native_goal_event, start_native_goal_turn, + ) + + turn = start_native_goal_turn(FakeTransport(), _config()) + observe_native_goal_event(turn, {"type": "event_msg", "payload": { + "type": "task_complete", "error": {"message": "PRIVATE-SENTINEL", + "codexErrorInfo": {"responseTooManyFailedAttempts": {"httpStatusCode": 429}}}}}) + turn.post_goal_status = "blocked" + assert native_goal_outcome(turn) == "failed" + assert turn.terminal_error_kind == "responseTooManyFailedAttempts" + assert turn.terminal_error_http_status == 429 + assert turn.error_event_count == 0 # no error notification in this stream + assert "PRIVATE-SENTINEL" not in json.dumps(compact_native_goal_receipt(turn)) + observe_native_goal_event(turn, {"method": "turn/started", "params": { + "threadId": turn.thread_id, "turn": {"id": "next", "status": "inProgress"}}}) + assert turn.terminal_error_signature is None + assert not turn.terminal_event_observed + observe_native_goal_event(turn, {"method": "turn/completed", "params": { + "threadId": turn.thread_id, "turn": {"id": "next", "status": "completed"}}}) + turn.post_goal_status = "complete" + assert native_goal_outcome(turn) == "complete" + + +@pytest.mark.parametrize("error", ["PRIVATE-SENTINEL", {"codexErrorInfo": "PRIVATE-SENTINEL"}, + {"codexErrorInfo": {"private-field": {"httpStatusCode": 401}}}]) +def test_unknown_terminal_error_stays_unknown(error): + from benchmark.tests.test_native_codex_goal import FakeTransport, _config + from loopx.capabilities.benchmark_toolkit.native_codex_goal import ( + compact_native_goal_receipt, observe_native_goal_event, start_native_goal_turn, + ) + turn = start_native_goal_turn(FakeTransport(), _config()) + observe_native_goal_event(turn, {"type": "event_msg", "payload": {"type": "task_complete", "error": error}}) + assert turn.turn_status == "failed" + assert turn.terminal_error_kind == "unknown" + assert turn.terminal_error_http_status is None + assert "PRIVATE-SENTINEL" not in json.dumps(compact_native_goal_receipt(turn)) + + +def test_edgebench_retains_failed_native_score_without_claiming_completion(tmp_path): + pytest.importorskip("sforge") + pytest.importorskip("harbor") + from benchmark.edgebench.run import _result_status, _write_native_final_result + from sforge.harness.run_agent import RunResult + + result, receipt = native_worker(tmp_path, "blocked", "failed", http_status=401) + status = _result_status(interrupted=False, started=True, execution_started=True, + runtime_seconds=2, exit_code=result.returncode, + native_goal_outcome=receipt["native_goal_outcome"]) + assert status == "runner_failed" + score = RunResult(runtime_seconds=2, best_score=12.5, total_rounds=2) + _write_native_final_result(tmp_path, score, status=status, agent="fixture", task="fixture", + run_id="fixture", model="fixture", effort="high", + native_goal_outcome=receipt["native_goal_outcome"]) + final = json.loads((tmp_path / "final_result.json").read_text()) + assert final["best_score"] == 12.5 + assert final["runtime_status"] == "runner_failed" + assert final["native_goal_outcome"] == "failed" + + +@pytest.mark.parametrize("outcome,expected_code", [("failed", 1), ("blocked", 0), (None, 1)]) +def test_edgebench_cli_persists_native_outcome_even_when_transport_exits_zero(tmp_path, monkeypatch, outcome, expected_code): + pytest.importorskip("sforge") + pytest.importorskip("harbor") + from benchmark.edgebench import run + from sforge.harness.run_agent import RunResult + + for name, value in {"LOOPX_SRC_DIR": str(tmp_path), "LOOPX_EXPECTED_COMMIT": "fixture", + "OPENAI_BASE_URL": "http://localhost:8123/v1", "OPENAI_API_KEY": "fixture-key"}.items(): + monkeypatch.setenv(name, value) + monkeypatch.setattr(run, "source_pins", lambda *a: ("fixture", "fixture")) + monkeypatch.setattr(run, "active_source_receipt", lambda: {}) + monkeypatch.setattr(run, "load_benchmark", lambda *a: None) + monkeypatch.setattr(run, "make_task_spec", lambda *a: SimpleNamespace( + task_id="fixture", cwd="/task", work_image_key="work", judge_image_key="judge", + internet=False, game_mode=False, submit_paths=["solver.py"], + work=SimpleNamespace(agent_query="Synthetic task"), + judge=SimpleNamespace(selection="score_first", score_direction="maximize"))) + backend = SimpleNamespace(image_exists=lambda image: True, + execution_receipt={"exit_code": 0, "execution_started": True}, + native_goal_receipt={"native_goal_outcome": outcome} if outcome else None) + monkeypatch.setattr(run, "RecordingDockerBackend", lambda **kwargs: backend) + trial = tmp_path / "runs" / "attempt" / "fixture" + def completed_scoring(**kwargs): + assert kwargs["disable_auto_resume"] is True + (trial / "started_at").touch() + return RunResult(runtime_seconds=2, best_score=4, total_rounds=1) + monkeypatch.setattr(run, "run_agent", completed_scoring) + (tmp_path / "fixture.json").write_text("{}") + code = run.main(["--task", "fixture", "--tasks-dir", str(tmp_path), "--log-dir", str(tmp_path), + "--run-id", "attempt", "--worker", "native-goal", "--model", "fixture", + "--effort", "high", "--judge-url", "http://localhost:9999"]) + assert code == expected_code + receipt = json.loads((trial / "runtime-receipt.json").read_text()) + assert receipt["native_goal_outcome"] == (outcome or "unknown") + assert receipt["score_countable"] is False + assert receipt["status"] == ("terminal" if expected_code == 0 else "runner_failed") + final = json.loads((trial / "final_result.json").read_text()) + assert final["runtime_status"] == receipt["status"] + assert final["best_score"] == 4 + + +@pytest.mark.skipif(os.environ.get("LOOPX_EDGEBENCH_DOCKER_SMOKE") != "1", + reason="Requires explicit isolated Docker qualification") +@pytest.mark.parametrize("goal,turn,expected_status", [ + ("complete", "completed", "terminal"), + ("blocked", "completed", "terminal"), + ("blocked", "failed", "runner_failed"), +]) +def test_docker_collection_preserves_real_worker_outcome(tmp_path, goal, turn, expected_status): + from benchmark.edgebench.run import _result_status, _write_native_final_result + from benchmark.runtime.sforge_backend import RecordingDockerBackend + from benchmark.runtime.harbor import NATIVE_GOAL_SOURCE_PATHS + from sforge.harness.run_agent import RunResult + + # The production worker really spawns a stdio app-server; only inference is + # replaced. No real task, credential, network service or leaderboard is used. + native_worker(tmp_path, goal, turn) + binary = tmp_path / "codex" + binary.write_text(binary.read_text().replace(f"#!{sys.executable}", "#!/usr/local/bin/python3", 1)) + logger = logging.getLogger("native-outcome-smoke") + backend = RecordingDockerBackend(log_dir=tmp_path / "collected", logger=logger, oauth_proxy=True) + handle = backend.create_container("python:3.12-slim", "native-outcome-" + uuid.uuid4().hex[:10], + cpu_limit=1, mem_limit="256m") + backend.start_container(handle) + cleaned = False + try: + root = Path(__file__).resolve().parents[2] + sources = [] + for name in NATIVE_GOAL_SOURCE_PATHS: + source = root / name + if source.is_dir(): + sources.extend(str(path.relative_to(root)) for path in source.rglob("*.py")) + else: + sources.append(name) + for name in sources: + target = PurePosixPath("/tmp/source") / name + backend.exec_run(handle, ["mkdir", "-p", str(target.parent)], user="root") + backend.copy_to_container(handle, root / name, target) + backend.copy_to_container(handle, binary, PurePosixPath("/tmp/codex")) + backend.copy_to_container(handle, tmp_path / "task.md", PurePosixPath("/tmp/task.md")) + env = {"LOOPX_EXECUTION_MODE": "native-goal", "LOOPX_PROJECT": "/tmp", + "LOOPX_TASK_DOC": "/tmp/task.md", "LOOPX_CODEX_HOME": "/tmp/home", + "LOOPX_WAKE_LOG_DIR": "/logs/agent/wakes", "CODEX_BIN": "/tmp/codex", + "MODEL_NAME": "fixture", "REASONING_EFFORT": "high", "PYTHONPATH": "/tmp/source", + "OPENAI_BASE_URL": "http://localhost:8123/v1", "OPENAI_API_KEY": "fixture-key"} + backend.execution_command = "python3 -m benchmark.runtime.worker" + observed = backend.exec_run_with_timeout(handle, ["/bin/bash", "-c", backend.execution_command], + environment=env, timeout=30, user="root") + assert '"native_goal_outcome"' in observed.output, observed.output + backend.cleanup_container(handle) + cleaned = True + native = backend.native_goal_receipt + assert native and native["native_goal"]["turn_status"] == turn + status = _result_status(interrupted=False, started=True, runtime_seconds=observed.elapsed_seconds, + exit_code=backend.execution_receipt["exit_code"], + native_goal_outcome=native["native_goal_outcome"]) + assert status == expected_status + _write_native_final_result(tmp_path, RunResult(runtime_seconds=observed.elapsed_seconds, best_score=7), + status=status, agent="fixture", task="fixture", run_id="fixture", + model="fixture", effort="high", native_goal_outcome=native["native_goal_outcome"]) + final = json.loads((tmp_path / "final_result.json").read_text()) + assert final["best_score"] == 7 and final["runtime_status"] == expected_status + finally: + if not cleaned: + backend.cleanup_container(handle) + backend.client.close() diff --git a/benchmark/tests/test_shared_codex_runtime.py b/benchmark/tests/test_shared_codex_runtime.py index 5239587250..7cf42f30db 100644 --- a/benchmark/tests/test_shared_codex_runtime.py +++ b/benchmark/tests/test_shared_codex_runtime.py @@ -213,7 +213,7 @@ def never_ends(suffix, timeout): assert native_goal_rounds(never_ends, budget_seconds=1000, continue_after_complete=True, clock=lambda: clock[0]) == (1, "budget_exhausted") # A non-complete terminal status stops without continuation. clock[0] = 0 - assert native_goal_rounds(lambda s, t: "cancelled", budget_seconds=1000, continue_after_complete=True, clock=lambda: clock[0]) == (1, "stopped") + assert native_goal_rounds(lambda s, t: "cancelled", budget_seconds=1000, continue_after_complete=True, clock=lambda: clock[0]) == (1, "cancelled") def test_fresh_wakes_share_environment_without_resuming_or_duplicate_session_copies( diff --git a/loopx/capabilities/benchmark_toolkit/native_codex_goal.py b/loopx/capabilities/benchmark_toolkit/native_codex_goal.py index 0fb0ec83ad..55eafb5108 100644 --- a/loopx/capabilities/benchmark_toolkit/native_codex_goal.py +++ b/loopx/capabilities/benchmark_toolkit/native_codex_goal.py @@ -110,6 +110,9 @@ class NativeGoalTurn: notifications: list[str] = field(default_factory=list) notification_counts: dict[str, int] = field(default_factory=dict) error_signatures: list[str] = field(default_factory=list) + terminal_error_signature: str | None = None + terminal_error_kind: str | None = None + terminal_error_http_status: int | None = None event_turn_id_observed: bool = False terminal_event_observed: bool = False turn_status: str = "not_started" @@ -320,6 +323,10 @@ def observe_native_goal_event( turn.turn_id = event_turn_id turn.event_turn_id_observed = True turn.turn_status = str(event_turn.get("status") or "inProgress") + turn.terminal_event_observed = False + turn.terminal_error_signature = None + turn.terminal_error_kind = None + turn.terminal_error_http_status = None turn.turn_started_count += 1 return False if event_turn_id and event_turn_id != turn.turn_id: @@ -337,11 +344,57 @@ def observe_native_goal_event( "event_msg:turn_completed", }: turn.terminal_event_observed = True - turn.turn_status = str(event_turn.get("status") or "completed") + error = event_turn.get("error") or params.get("error") + # JSONL task_complete has no turn.status; an attached error is a failed + # Turn, not a successful terminal event. Do not count a repeated error + # notification as another failed request or retain its message. + turn.turn_status = "failed" if error else str(event_turn.get("status") or "completed") + if error: + turn.terminal_error_signature = _digest(json.dumps(error, sort_keys=True, default=str))[:12] + turn.terminal_error_kind = "unknown" + if isinstance(error, Mapping): + info = error.get("codexErrorInfo", error.get("codex_error_info")) + # Native Codex variants only; unknown/free-text data stays out + # of this public receipt. Do not infer HTTP status from prose. + scalar_kinds = {"contextWindowExceeded", "sessionBudgetExceeded", "usageLimitExceeded", + "rateLimitExceeded", "serverOverloaded", "cyberPolicy", + "misalignmentPolicyViolation", "internalServerError", "unauthorized", + "badRequest", "threadRollbackFailed", "sandboxError", "other"} + http_kinds = {"httpConnectionFailed", "responseStreamConnectionFailed", + "responseStreamDisconnected", "responseTooManyFailedAttempts"} + statuses = [] + if isinstance(info, str) and info in scalar_kinds: + turn.terminal_error_kind = info + elif isinstance(info, Mapping) and len(info) == 1: + kind = next(iter(info)) + if kind in http_kinds and isinstance(info[kind], Mapping): + turn.terminal_error_kind = kind + statuses.append(info[kind].get("httpStatusCode")) + valid = {value for value in statuses if type(value) is int and 100 <= value <= 599} + turn.terminal_error_http_status = next(iter(valid)) if len(valid) == 1 else None turn.turn_completed_count += 1 return turn.terminal_event_observed +def native_goal_outcome(turn: NativeGoalTurn) -> str: + """Classify the latest terminal Turn separately from its Goal's status. + + This is native transport evidence, not task acceptance or retry authority. + Historical recoverable errors must not override a later completed Turn. + """ + if not turn.terminal_event_observed: + return "unknown" + if turn.turn_status in {"failed", "error"}: + return "failed" + if turn.turn_status in {"interrupted", "cancelled"}: + return "cancelled" + if turn.turn_status != "completed": + return "unknown" + return {"complete": "complete", "blocked": "blocked", "paused": "paused", + "budgetLimited": "budget_exhausted", "usageLimited": "usage_limited" + }.get(turn.post_goal_status, "unknown") + + def wait_native_goal_turn( transport: NativeGoalEventTransport, turn: NativeGoalTurn, @@ -567,6 +620,9 @@ def compact_native_goal_receipt(turn: NativeGoalTurn) -> dict[str, Any]: else None ), "error_signatures": list(turn.error_signatures), + "terminal_error_signature": turn.terminal_error_signature, + "terminal_error_kind": turn.terminal_error_kind, + "terminal_error_http_status": turn.terminal_error_http_status, "public_boundary": { "raw_objective_recorded": False, "raw_task_instruction_recorded": False, @@ -588,6 +644,7 @@ def compact_native_goal_receipt(turn: NativeGoalTurn) -> dict[str, Any]: "attach_native_goal", "compact_native_goal_receipt", "observe_native_goal_event", + "native_goal_outcome", "probe_native_goal_process", "refresh_native_goal_status", "run_native_goal_process",