Preserve scoring when terminal worker capture fails

benchmark/edgebench/README.md

@@ -896,6 +896,16 @@ requires a running container; latest files remain in the stopped source and, on
 successful export, the terminal point. Native graded history/results remain
 independent and are not discarded with that optional archive.
 
+If the joint export failed, a confirmed stopped worker with a completed grading
+drain gets one independent native scorer capture before the original scorer
+exits. The receipt links this separate checkpoint, including the released results,
+incumbent and consumed rounds. It carries no worker clock or resume qualification;
+a receiver still needs to prove the original workspace, session and remaining
+allowance together. An unknown grading outcome or unconfirmed stop does not
+trigger this capture.
+If the native runner raises after writing its drain receipt, the runner
+checks the matching run history before attempting the same capture.
+
 If the optional preservation receipt cannot be written, a redacted warning reports
 that gap; observer shutdown, one capture attempt and verified source stopping
 still proceed. The known solver exit and native grading remain authoritative.

benchmark/edgebench/checkpoint_supervisor.py

@@ -94,6 +94,24 @@ def finish_solver(self, execution):
         else:
             self.preservation.finish(execution)
 
+    def finish_grading(self, status):
+        if self.preservation is not None:
+            self.preservation.finish_grading(status)
+
+    def finish_grading_from_receipt(self, trial, *, run_id, task_id):
+        """Recover the drain observation when the native runner raises."""
+        if self.preservation is None or self.preservation.receipt is None:
+            return
+        from .managed_run import native_grading_drained
+
+        try:
+            drained = native_grading_drained(
+                trial, history_file="run_history.json", run_id=run_id, task_id=task_id)
+        except (OSError, ValueError):
+            return  # Bad optional evidence must not mask the original exception.
+        if drained:
+            self.finish_grading("complete")
+
     def close(self):
         self.stop.set()
         if self.thread is not None:

benchmark/edgebench/managed_run.py

@@ -200,7 +200,7 @@ def _terminal_history_entry(entry, *, task_id: str) -> bool:
     )
 
 
-def _grading_drained(trial: Path, *, history_file: str, run_id: str, task_id: str) -> bool:
+def native_grading_drained(trial: Path, *, history_file: str, run_id: str, task_id: str) -> bool:
     try:
         grading, _ = _read_record(trial / "grading_status.json")
     except FileNotFoundError:
@@ -272,7 +272,7 @@ def publish_managed_terminal(trial: Path, *, execute: bool = False) -> bool:
     if not solver_matches or not _solver_ended(execution, admitted_at=admitted.timestamp(),
                                                observed_at=datetime.now(UTC).timestamp()):
         raise ValueError("Retained solver termination evidence differs")
-    if not _grading_drained(trial, history_file=history_file, run_id=run_id, task_id=task_id):
+    if not native_grading_drained(trial, history_file=history_file, run_id=run_id, task_id=task_id):
         return False
     continuity = build_benchmark_runtime_continuity(
         launch_runtime_digest=retained.get("launch_runtime_sha256"),

benchmark/edgebench/run.py

@@ -609,10 +609,14 @@ def main(argv=None):
         if provider_failure := getattr(backend, "provider_failure_receipt", None):
             receipt["provider_failure"] = provider_failure
         receipt_path.write_text(json.dumps(receipt, indent=2))
+        if checkpoint is not None:
+            checkpoint.finish_grading_from_receipt(trial, run_id=args.run_id, task_id=task.task_id)
         raise
     finally:
         if feedback is not None:
             feedback.close()
+    if checkpoint is not None:
+        checkpoint.finish_grading(getattr(result, "grading_status", 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 {}

benchmark/edgebench/terminal_capture.py

@@ -168,6 +168,38 @@ def stop_source(self, container):
         finally:
             self._persist_receipt()
 
+    def finish_grading(self, grading_status):
+        """Seal the drained native registration if the joint export failed."""
+        receipt = self.receipt
+        if (grading_status != "complete" or receipt is None
+                or not receipt["capture_attempted"] or receipt["capture_succeeded"]
+                or receipt.get("source_stopped") is not True
+                or "native_scoring_checkpoint" in receipt):
+            return
+
+        # This point is captured after grading, with no worker phase clock.
+        # It is deliberately not the earlier joint attempt or a resume grant.
+        capture_id = uuid.uuid4().hex
+        retained = {"capture_id": capture_id, "capture_succeeded": False}
+        receipt["native_scoring_checkpoint"] = retained
+        self._persist_receipt()
+        try:
+            from .checkpoint_controller import capture
+
+            retained["capture"] = capture(
+                self.supervisor.controller_file,
+                token=self.supervisor.token,
+                capture_id=capture_id,
+                destination=self.receipt_path.parent / f"terminal-native-{capture_id}",
+            )
+            retained["capture_succeeded"] = True
+        except Exception as error:
+            # A failed or uncertain export remains evidence, never a retry
+            # under a different capture identity or a changed solver outcome.
+            retained["error_kind"] = type(error).__name__
+        finally:
+            self._persist_receipt()
+
 
 def _stop_retained_worker(container, receipt):
     if container.id != receipt["container_id"]:

benchmark/tests/test_terminal_preservation.py

@@ -78,6 +78,27 @@ def test_success_retains_normal_cleanup_unless_writer_barrier_is_uncertain(tmp_p
         assert "capture" not in calls
 
 
+def test_native_fallback_failure_keeps_original_result_and_does_not_retry(tmp_path, monkeypatch):
+    preservation, calls = fixture_preservation(tmp_path, monkeypatch, "capture")
+    preservation.finish({"exit_code": 2, "execution_started": True})
+
+    def failed_export(*args, **kwargs):
+        calls.append("native")
+        raise OSError("private scorer transport details")
+
+    monkeypatch.setattr("benchmark.edgebench.checkpoint_controller.capture", failed_export)
+    preservation.finish_grading("unknown")
+    assert "native_scoring_checkpoint" not in preservation.receipt
+    preservation.finish_grading("complete")
+    preservation.finish_grading("complete")
+    receipt = json.loads((tmp_path / "terminal-preservation.json").read_text())
+    assert calls == ["close", "capture", "native"]
+    assert receipt["error_kind"] == "TimeoutError" and receipt["source_stopped"] is True
+    assert not receipt["capture_succeeded"] and not receipt["restore_qualified"]
+    assert receipt["native_scoring_checkpoint"]["error_kind"] == "OSError"
+    assert "private scorer" not in json.dumps(receipt)
+
+
 def test_component_declaration_captures_full_workspace_and_no_auth():
     original = {"components": [{"id": "submitted", "source_path": "/task/solution.py"},
                                 {"id": "notes", "source_path": "/private-notes"}]}
@@ -314,6 +335,13 @@ def registered(token):
     assert result.solver_execution["status"] == "nonzero_exit"
     assert result.best_score == 7 and result.agent_submissions == 1
     assert drained == ["complete"] and result.archive == b""
+    if capture_failure:
+        supervisor.finish_grading_from_receipt(trial, run_id="other-run", task_id="fixture")
+        assert "native_scoring_checkpoint" not in supervisor.preservation.receipt
+        supervisor.finish_grading_from_receipt(trial, run_id="source", task_id="fixture")
+    else:
+        supervisor.finish_grading(result.grading_status)
+    supervisor.finish_grading(result.grading_status)  # A second notification cannot reseal.
     assert json.loads((trial / "run_history.json").read_text())["best_score"] == 7
     container.reload()
     assert not container.attrs["State"]["Running"] and container.attrs["State"]["Pid"] == 0
@@ -327,6 +355,20 @@ def registered(token):
     assert not receipt["restore_qualified"]
     assert receipt["capture_succeeded"] is not capture_failure
     assert len(grades) == 1  # Saving the tail did not create an evaluation opportunity.
+    if capture_failure:
+        native_capture = receipt["native_scoring_checkpoint"]
+        assert native_capture["capture_succeeded"]
+        assert native_capture["capture_id"] != receipt["capture_id"]
+        point, = tmp_path.glob("terminal-native-*")
+        manifest = json.loads((point / "components/native_judge/manifest.json").read_text())
+        assert [entry["score"] for entry in manifest["history"]] == [7]
+        assert manifest["opportunity"]["next_agent"] == 2
+        assert 0 < manifest["opportunity"]["remaining_cooldown_seconds"] <= 3600
+        assert (point / "components/native_judge/incumbent.tar.gz").read_bytes() == b"7"
+        assert not (point / "components/host_phase_deadline").exists()
+        assert not native_capture["capture"]["research_point_qualified"]
+    else:
+        assert "native_scoring_checkpoint" not in receipt
     if not capture_failure:
         point = supervisor.destination / receipt["capture_id"]
         manifest = json.loads((point / "checkpoint.json").read_text())