fix(benchmark): retain partial results until native grading completes

benchmark/edgebench/README.md

@@ -308,9 +308,18 @@ do not publish `final_result.json`. Keep their captures and failure evidence;
 do not count an initial artifact's score as a solver outcome.
 For native Goals with positive runtime, `partial_result.json` preserves the full
 scoring result and its `runtime_status` / `native_goal_outcome` without making
-SForge's visualizer count the attempt as done. Handled terminal stops retain
-`final_result.json`; its presence does not independently qualify a score. Use
-new source pins for new attempts and leave frozen histories unchanged; see the
+SForge's visualizer count the attempt as done. A handled terminal stop records
+completed execution, not necessarily completed grading. When the native result
+includes `grading_status`, only `complete` permits `final_result.json`; explicit
+`pending`, `unknown` or other values remain in `partial_result.json`, preserving
+the native grading reason, scores and `runtime_status: terminal`. The report
+exporter also rejects explicit incomplete grading, including older attempts that
+already wrote a final marker. Sources predating `grading_status` retain their
+existing completion behavior; absence does not add a new grading attestation.
+Result publication does not rescore submissions, replace the current final
+archive with a historical best, or rewrite native history. A final marker still
+does not independently qualify a score. Use new source pins for new attempts and
+leave frozen histories unchanged; see the
 [native outcome contract](../runtime/RUNTIME.md) for rollback and classification.
 
 New native archive-submission trials default to **no periodic auto-evaluation**

benchmark/edgebench/export_report.py

@@ -19,6 +19,7 @@
     normalize_benchmark_experiment_board_row,
 )
 from .feedback_hook import FEEDBACK_PAYLOAD
+from .result_publication import native_grading_complete
 
 
 def _validate_feedback_payload(runner, profile):
@@ -234,6 +235,8 @@ def export(runs_root: Path, selection: list, output: Path, *, observed_at: str):
             "completed",
         }:
             raise ValueError("Only terminal, scored runs can be exported")
+        if not native_grading_complete(final):
+            raise ValueError("Native grading is not complete; terminal execution alone cannot be exported")
         for source in (receipt, final):
             if (
                 source["run_id"] != run

benchmark/edgebench/result_publication.py

@@ -0,0 +1,30 @@
+"""Publish native results without changing grading, ranking or archives."""
+
+import json
+
+
+def native_grading_complete(result):
+    """Require explicit native completion; retain sources predating this field."""
+    return "grading_status" not in result or result["grading_status"] == "complete"
+
+
+def write_native_result(trial, result, *, status, agent, task, run_id, model, effort,
+                        feedback="native", native_goal_outcome=None):
+    """Retain partial native scores without publishing a completion marker."""
+    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())
+    complete = status == "terminal" and native_grading_complete(final)
+    if not complete or native_goal_outcome is not None:
+        final["runtime_status"] = status
+    if native_goal_outcome is not None:
+        final["native_goal_outcome"] = native_goal_outcome
+    if feedback == "best-only":
+        final["evaluation_coverage"] = "online_only"
+        final["offline_scoring_complete"] = False
+    # SForge consumers treat final_result.json itself as a completed run.
+    destination = trial / ("final_result.json" if complete else "partial_result.json")
+    pending = destination.with_suffix(".json.tmp")
+    pending.write_text(json.dumps(final, indent=2, ensure_ascii=False))
+    pending.replace(destination)

benchmark/edgebench/run.py

@@ -44,6 +44,7 @@
 from benchmark.edgebench.online_sampling import OnlineSampler
 from benchmark.edgebench.feedback import BestOnlyFeedback, FEEDBACK_MODES, validate_best_only
 from benchmark.edgebench.feedback_hook import FEEDBACK_PAYLOAD
+from benchmark.edgebench.result_publication import write_native_result as _write_native_final_result
 
 
 def _observe_run(call):
@@ -85,25 +86,6 @@ def _result_status(*, interrupted, started, runtime_seconds, exit_code, timed_ou
     return "terminal"
 
 
-def _write_native_final_result(trial, result, *, status, agent, task, run_id, model, effort,
-                               feedback="native", native_goal_outcome=None):
-    """Retain partial native scores without publishing a completion marker."""
-    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
-    # SForge consumers treat final_result.json itself as a completed run.
-    destination = trial / ("final_result.json" if status == "terminal" else "partial_result.json")
-    pending = destination.with_suffix(".json.tmp")
-    pending.write_text(json.dumps(final, indent=2, ensure_ascii=False))
-    pending.replace(destination)
-
-
 def _task_default(task_id: str, key: str, explicit: int | None, fallback: int) -> int:
     """Resolve EdgeBench task settings; explicit values retain runtime semantics."""
     if explicit is not None:

benchmark/runtime/RUNTIME.md

@@ -206,9 +206,13 @@ The EdgeBench adapter carries this outcome through artifact collection into
 `runtime_status` and `native_goal_outcome`. Failed, cancelled, usage-limited and
 unknown native outcomes do not publish `final_result.json`: existing SForge
 consumers use that file as a completion marker without interpreting the outcome
-labels. Handled terminal stops retain the existing final-result behavior;
-neither artifact grants result qualification. Ordinary non-native profile
-handling is unchanged. Adopt this behavior only for newly pinned attempts;
+labels. A handled terminal stop completes execution; explicit native grading
+must also be `complete` to publish a final result. Pending or unknown grading
+retains its scores and reason in a partial result, including for ordinary
+non-native profiles. Sources that omit `grading_status` retain the legacy
+publication behavior. Neither artifact grants result qualification; see the
+[EdgeBench result publication rules](../edgebench/README.md) for export behavior.
+Adopt this behavior only for newly pinned attempts;
 frozen histories retain their original receipts and require explicit post-run
 classification. Roll back by pinning the previous source for another new attempt,
 not by renaming artifacts in a frozen run.

benchmark/tests/test_edgebench_feedback_docker.py

@@ -325,7 +325,10 @@ async def installed(self, environment):
                 assert backend.exec_run(handle, ["test", "-f", "/tmp/solver-entered"]).exit_code != 0
             trial = tmp_path / name
             trial.mkdir()
-            _write_native_final_result(trial, RunResult(runtime_seconds=observed.elapsed_seconds),
+            result = RunResult(runtime_seconds=observed.elapsed_seconds)
+            if hasattr(result, "grading_status"):
+                result.grading_status = "complete"
+            _write_native_final_result(trial, result,
                 status=status, agent="fixture", task="fixture", run_id=name, model="fixture", effort="xhigh")
             assert (trial / "final_result.json").exists() is (expected_status == "terminal")
     finally:

benchmark/tests/test_edgebench_report.py

@@ -7,10 +7,12 @@
 import subprocess
 import sys
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
 from benchmark.edgebench.export_report import digest, export, verify, write
+from benchmark.edgebench.result_publication import write_native_result
 
 
 ROOT = Path(__file__).resolve().parents[2]
@@ -173,6 +175,34 @@ def test_terminal_report_retains_unqualified_status_and_unknowns(trial):
         build(trial)
 
 
+@pytest.mark.parametrize("grading_status", ["pending", "unknown", None, "future-status"])
+def test_report_rejects_explicit_incomplete_grading_before_publication(trial, grading_status):
+    final_path = trial[3] / "final_result.json"
+    final = json.loads(final_path.read_text())
+    final.update(grading_status=grading_status, grading_reason="synthetic_unfinished_grading")
+    write(final_path, final)
+    with pytest.raises(ValueError, match="grading is not complete"):
+        build(trial)
+    assert not trial[2].exists()
+    assert json.loads(final_path.read_text()) == final
+
+
+@pytest.mark.parametrize("legacy", [False, True])
+def test_report_accepts_published_complete_or_legacy_native_grading(trial, legacy):
+    final_path = trial[3] / "final_result.json"
+    payload = json.loads(final_path.read_text())
+    for key in ("run_id", "task", "model"):
+        payload.pop(key)
+    if not legacy:
+        payload.update(grading_status="complete", grading_reason="all_submissions_terminal")
+    final_path.unlink()
+    result = SimpleNamespace(runtime_seconds=payload["runtime_seconds"],
+                             to_dict=lambda: dict(payload))
+    write_native_result(trial[3], result, status="terminal", agent="codex", task="case",
+                        run_id="run", model="model", effort="xhigh")
+    assert verify(build(trial))["runs"] == 1
+
+
 @pytest.mark.parametrize("fault", [None, "profile", "mode", "unknown"])
 def test_official_feedback_payload_is_preserved_without_relabeling_history(trial, fault):
     for name in ("runtime-receipt", "worker-profile"):

benchmark/tests/test_edgebench_result_publication.py

@@ -0,0 +1,126 @@
+"""Completion markers follow native grading; partial evidence remains native."""
+
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from benchmark.edgebench.result_publication import write_native_result
+
+
+@pytest.fixture
+def native_result():
+    payload = dict(
+        best_score=0.0, best_pass_rate=1.0, best_round="auto-1", total_rounds=2,
+        agent_submissions=1, auto_submissions=1, runtime_seconds=12.0,
+        timed_out=True, resume_count=1, archive_size_bytes=17,
+        grading_status="complete", grading_reason="all_submissions_terminal",
+        solver_execution={"status": "timed_out"},
+    )
+    result = SimpleNamespace(runtime_seconds=12.0, to_dict=lambda: dict(payload))
+    return result, payload
+
+
+def publish(trial, result, **kwargs):
+    write_native_result(trial, result, agent="codex", task="case", run_id="run",
+                        model="model", effort="high", **kwargs)
+
+
+@pytest.mark.parametrize("legacy", [False, True])
+@pytest.mark.parametrize("native_goal_outcome", [None, "completed"])
+def test_complete_or_legacy_result_retains_final_shape(tmp_path, native_result, legacy,
+                                                       native_goal_outcome):
+    result, payload = native_result
+    if legacy:
+        payload.pop("grading_status")
+        payload.pop("grading_reason")
+    publish(tmp_path, result, status="terminal", native_goal_outcome=native_goal_outcome)
+    expected = dict(agent="codex", task="case", run_id="run", model="model", effort="high",
+                    **payload)
+    if native_goal_outcome is not None:
+        expected.update(runtime_status="terminal", native_goal_outcome=native_goal_outcome)
+    assert json.loads((tmp_path / "final_result.json").read_text()) == expected
+    assert not (tmp_path / "partial_result.json").exists()
+    assert not list(tmp_path.glob("*.tmp"))
+
+
+@pytest.mark.parametrize("grading_status", ["pending", "unknown", None, "future-status"])
+def test_terminal_incomplete_grading_retains_partial_evidence(tmp_path, native_result,
+                                                              grading_status):
+    result, payload = native_result
+    payload.update(grading_status=grading_status, grading_reason="synthetic_unfinished_grading")
+    history = {"run_id": "run", "best_score": 0.0, "best_round": "auto-1", "entries": [
+        {"type": "submission", "status": "completed", "round": "auto-1", "score": 0.0},
+        {"type": "submission", "status": "pending", "round": "agent-1"},
+    ]}
+    retained = {
+        "run_history.json": json.dumps(history).encode(),
+        "final_archive.tar.gz": b"synthetic current workspace",
+        "submissions/auto-1/submission.tar.gz": b"synthetic earlier incumbent",
+    }
+    for name, content in retained.items():
+        path = tmp_path / name
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(content)
+    publish(tmp_path, result, status="terminal")
+    assert not (tmp_path / "final_result.json").exists()
+    partial = json.loads((tmp_path / "partial_result.json").read_text())
+    assert partial == dict(agent="codex", task="case", run_id="run", model="model",
+                           effort="high", runtime_status="terminal", **payload)
+    assert {name: (tmp_path / name).read_bytes() for name in retained} == retained
+    assert not list(tmp_path.glob("*.tmp"))
+
+
+@pytest.mark.parametrize("grading_status", ["pending", "unknown"])
+def test_completed_native_goal_does_not_override_unfinished_grading(tmp_path, native_result,
+                                                                   grading_status):
+    result, payload = native_result
+    payload.update(grading_status=grading_status, grading_reason="synthetic_unfinished_grading")
+    publish(tmp_path, result, status="terminal", native_goal_outcome="completed")
+    assert not (tmp_path / "final_result.json").exists()
+    partial = json.loads((tmp_path / "partial_result.json").read_text())
+    assert partial["runtime_status"] == "terminal"
+    assert partial["native_goal_outcome"] == "completed"
+    assert all(partial[key] == value for key, value in payload.items())
+
+
+@pytest.mark.parametrize("status,outcome", [
+    ("cancelled", "cancelled"), ("runner_failed", "failed"),
+    ("runner_failed", "usage_limited"), ("runner_failed", "unknown"),
+])
+def test_completed_grading_preserves_partial_failed_native_goal(tmp_path, native_result,
+                                                                status, outcome):
+    result, payload = native_result
+    publish(tmp_path, result, status=status, native_goal_outcome=outcome)
+    assert not (tmp_path / "final_result.json").exists()
+    partial = json.loads((tmp_path / "partial_result.json").read_text())
+    assert partial["runtime_status"] == status
+    assert partial["native_goal_outcome"] == outcome
+    assert all(partial[key] == value for key, value in payload.items())
+
+
+@pytest.mark.parametrize("status,outcome,runtime", [
+    ("launch_failed", None, 0.0), ("runner_failed", None, 12.0),
+    ("cancelled", None, 12.0), ("runner_failed", "failed", 0.0),
+])
+def test_unqualified_execution_keeps_existing_publication_behavior(tmp_path, native_result,
+                                                                  status, outcome, runtime):
+    result, payload = native_result
+    result.runtime_seconds = runtime
+    payload["runtime_seconds"] = runtime
+    publish(tmp_path, result, status=status, native_goal_outcome=outcome)
+    assert not list(tmp_path.iterdir())
+
+
+@pytest.mark.parametrize("grading_status,destination", [
+    ("complete", "final_result.json"), ("pending", "partial_result.json"),
+])
+def test_best_only_retains_coverage_without_claiming_offline_scoring(tmp_path, native_result,
+                                                                   grading_status, destination):
+    result, payload = native_result
+    payload["grading_status"] = grading_status
+    publish(tmp_path, result, status="terminal", feedback="best-only")
+    observed = json.loads((tmp_path / destination).read_text())
+    assert observed["evaluation_coverage"] == "online_only"
+    assert observed["offline_scoring_complete"] is False
+    assert all(observed[key] == value for key, value in payload.items())

benchmark/tests/test_native_goal_outcomes.py

@@ -197,7 +197,10 @@ def test_edgebench_cli_persists_native_outcome_even_when_transport_exits_zero(tm
     def completed_scoring(**kwargs):
         assert kwargs["disable_auto_resume"] is True
         (trial / "started_at").write_text("2024-01-01T00:00:00Z\n1704067200\n")
-        return RunResult(runtime_seconds=2, best_score=4, total_rounds=1)
+        result = RunResult(runtime_seconds=2, best_score=4, total_rounds=1)
+        if hasattr(result, "grading_status"):
+            result.grading_status = "complete"
+        return result
     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),
@@ -295,7 +298,10 @@ def test_docker_collection_preserves_real_worker_outcome(tmp_path, goal, turn, e
                                 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),
+        result = RunResult(runtime_seconds=observed.elapsed_seconds, best_score=7)
+        if hasattr(result, "grading_status"):
+            result.grading_status = "complete"
+        _write_native_final_result(tmp_path, result,
                                   status=status, agent="fixture", task="fixture", run_id="fixture",
                                   model="fixture", effort="high", native_goal_outcome=native["native_goal_outcome"])
         artifact = tmp_path / ("final_result.json" if expected_status == "terminal" else "partial_result.json")

benchmark/tests/test_sforge_runtime.py

@@ -617,6 +617,8 @@ def test_native_terminal_handoff_is_readable_by_visualizer(tmp_path, timed_out):
     result = RunResult(best_score=0.0, best_pass_rate=1.0, best_round="auto-1",
                        total_rounds=1, auto_submissions=1, runtime_seconds=12.0,
                        timed_out=timed_out)
+    if hasattr(result, "grading_status"):
+        result.grading_status = "complete"
     _write_native_final_result(trial, result, status="terminal", agent="codex",
                               task="case", run_id="run", model="model", effort="xhigh")
     final = json.loads((trial / "final_result.json").read_text())