fix(benchmark): prepare native control registrations through backend

benchmark/README.md

@@ -727,9 +727,11 @@ An owner-only mode is a storage precaution, not proof of solver exclusion.
 
 When the selected native runner supports `host_registration_callback`, the
 adapter binds enabled checkpoint/feedback observers through that host callback
-before container creation. It does not inject the registration token into the
-solver environment. Frozen older providers retain their original environment
-binding path; runs without either observer add no callback. Blind modes also
+before container creation. The CLI consults the backend's registration options
+for every worker profile, including native controls. The default backend still
+returns no callback when both observers are absent. It does not inject the
+registration token into the solver environment. Frozen older providers retain
+their original environment binding path. Blind modes also
 remove the newer `SFORGE_FEEDBACK_URL` from the solver environment. A restored
 registration may include the native provider's nonempty `feedback_relay_id`;
 all original activation identities must still match, and unknown fields fail

benchmark/edgebench/run.py

@@ -543,8 +543,7 @@ def main(argv=None):
             **({"restored_registration": restored_registration} if restored_registration is not None else {}),
             disable_stop_hook=False, disable_auto_resume=agent.resume_cmd is None,
             max_submissions=0 if args.feedback != "native" else None,
-            **(backend.registration_options(run_agent)
-               if feedback is not None or checkpoint is not None else {}),
+            **backend.registration_options(run_agent),
         ))
     except Exception as error:
         receipt.update(status="runner_failed", error_kind=type(error).__name__)

benchmark/tests/test_edgebench_feedback.py

@@ -362,7 +362,8 @@ def qualify(self):
     def backend(**kwargs):
         constructed.update(kwargs)
         return SimpleNamespace(image_exists=lambda image: True,
-            registration_options=lambda runner: {"host_registration_callback": kwargs["feedback"].bind})
+            registration_options=lambda runner: {"host_registration_callback": kwargs["feedback"].bind}
+            if kwargs["feedback"] is not None else {})
     monkeypatch.setattr(run, "RecordingDockerBackend", backend)
     def handoff(**kwargs):
         selected = mode or "native"

benchmark/tests/test_edgebench_registration.py

@@ -0,0 +1,117 @@
+"""Backend registration preparation applies to native controls as well."""
+import json
+from types import SimpleNamespace
+
+import pytest
+
+pytest.importorskip("sforge")
+pytest.importorskip("harbor")
+
+
+@pytest.fixture
+def native_cli(tmp_path, monkeypatch):
+    from benchmark.edgebench import run
+
+    monkeypatch.setenv("LOOPX_SRC_DIR", str(tmp_path))
+    monkeypatch.setenv("LOOPX_EXPECTED_COMMIT", "fixture")
+    monkeypatch.setenv("CODEX_AUTH_JSON_PATH", "/synthetic-login")
+    monkeypatch.setenv("LOOPX_USAGE_PING", "0")
+    monkeypatch.setattr(run, "source_pins", lambda *args: ("fixture", "fixture"))
+    monkeypatch.setattr(run, "load_benchmark", lambda *args: None)
+    monkeypatch.setattr(run, "make_task_spec", lambda *args: SimpleNamespace(
+        cwd="/task", work_image_key="work", judge_image_key="judge", internet=False,
+        game_mode=False, submit_paths=["solver.py"], work=SimpleNamespace(agent_query="fixture task"),
+        judge=SimpleNamespace(selection="score_first", score_direction="maximize")))
+    (tmp_path / "fixture.json").write_text("{}")
+    # Reaching any of these is outside this CLI registration boundary.
+    monkeypatch.setattr(run, "archive_source", lambda *args, **kwargs:
+                        pytest.fail("registration preparation must precede source staging"))
+    args = ["--task", "fixture", "--tasks-dir", str(tmp_path), "--log-dir", str(tmp_path),
+            "--run-id", "run", "--model", "fixture", "--effort", "xhigh",
+            "--feedback", "native", "--judge-url", "http://192.0.2.1:8080"]
+    return run, args, tmp_path / "runs/run/fixture"
+
+
+@pytest.mark.parametrize("profile", ["official", "single", "native-goal", "heartbeat-resume", "heartbeat-explore"])
+@pytest.mark.parametrize("custom_registration", [False, True])
+def test_native_cli_prepares_registration_for_every_profile(native_cli, monkeypatch,
+                                                           profile, custom_registration):
+    from benchmark.runtime.sforge_backend import RecordingDockerBackend
+
+    run, args, trial = native_cli
+    events = []
+    constructed = {}
+
+    def bind(token):
+        assert token == "synthetic-native-registration"
+        events.append("bind")
+
+    def backend(**kwargs):
+        assert kwargs.get("feedback") is None and kwargs.get("checkpoint") is None
+        constructed.update(kwargs)
+        result = SimpleNamespace(image_exists=lambda image: True, feedback=None, checkpoint=None)
+
+        def options(provider):
+            assert provider is native_provider
+            assert events == [], "the backend prepares before native registration or worker effects"
+            events.append("prepare")
+            if custom_registration:
+                return {"host_registration_callback": bind}
+            # Exercise the shipped default policy, rather than duplicating it.
+            return RecordingDockerBackend.registration_options(result, provider)
+
+        result.registration_options = options
+        return result
+
+    def native_provider(*, host_registration_callback=None, **kwargs):
+        if custom_registration:
+            assert events == ["prepare"]
+        events.append("register")
+        assert kwargs["agent"].profile == profile
+        assert kwargs["eval_interval"] == 0
+        assert kwargs["submission_cooldown"] == 3600
+        assert kwargs["max_submissions"] is None
+        assert kwargs["disable_auto_eval"] is False
+        assert "restored_registration" not in kwargs
+        if custom_registration:
+            assert host_registration_callback is bind
+            host_registration_callback("synthetic-native-registration")
+        else:
+            assert host_registration_callback is None, "ordinary runs gain no callback or observer"
+        # A real provider may now create/stage its worker and call its model.
+        # Stop at that boundary: this test grants none of those operations.
+        raise RuntimeError("synthetic worker boundary")
+
+    monkeypatch.setattr(run, "RecordingDockerBackend", backend)
+    monkeypatch.setattr(run, "run_agent", native_provider)
+    with pytest.raises(RuntimeError, match="synthetic worker boundary"):
+        run.main([*args, "--worker", profile])
+    if custom_registration:
+        assert events == ["prepare", "register", "bind"]
+    else:
+        assert [event for event in events if event != "prepare"] == ["register"]
+    assert constructed["research_sidecar"] is True
+    receipt = json.loads((trial / "runtime-receipt.json").read_text())
+    assert receipt["status"] == "runner_failed"
+    assert receipt["feedback"] == "native"
+    assert "checkpoint_export_directory" not in receipt
+
+
+@pytest.mark.parametrize("profile", ["official", "native-goal"])
+def test_registration_preparation_failure_prevents_native_admission(native_cli, monkeypatch, profile):
+    run, args, trial = native_cli
+    events = []
+
+    def prepare(provider):
+        events.append("prepare")
+        raise RuntimeError("synthetic registration unavailable")
+
+    monkeypatch.setattr(run, "RecordingDockerBackend", lambda **kwargs: SimpleNamespace(
+        image_exists=lambda image: True, registration_options=prepare))
+    monkeypatch.setattr(run, "run_agent", lambda **kwargs:
+                        pytest.fail("failed preparation must prevent registration and all worker effects"))
+    with pytest.raises(RuntimeError, match="synthetic registration unavailable"):
+        run.main([*args, "--worker", profile])
+    assert events == ["prepare"]
+    receipt = json.loads((trial / "runtime-receipt.json").read_text())
+    assert receipt["status"] == "runner_failed" and receipt["error_kind"] == "RuntimeError"

benchmark/tests/test_native_goal_outcomes.py

@@ -184,7 +184,7 @@ def test_edgebench_cli_persists_native_outcome_even_when_transport_exits_zero(tm
         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,
+    backend = SimpleNamespace(image_exists=lambda image: True, registration_options=lambda provider: {},
                               execution_receipt={"exit_code": 0, "execution_started": True},
                               native_goal_receipt={"native_goal_outcome": outcome} if outcome else None,
                               provider_failure_receipt=({

benchmark/tests/test_research_artifacts.py

@@ -88,7 +88,7 @@ def test_native_entry_installs_identical_research_scope_and_declares_checkpoint_
         judge=SimpleNamespace(selection="score_first", score_direction="maximize")))
     def backend(**kwargs):
         assert kwargs["research_sidecar"] is True
-        return SimpleNamespace(image_exists=lambda image: True)
+        return SimpleNamespace(image_exists=lambda image: True, registration_options=lambda provider: {})
     monkeypatch.setattr(run, "RecordingDockerBackend", backend)
     capture = profile.startswith("heartbeat-")
     request = {"destination": "/logs/agent/checkpoints", "components": [

benchmark/tests/test_sforge_runtime.py

@@ -934,7 +934,8 @@ def test_edgebench_native_goal_preserves_task_with_resolved_entry_and_treatment(
         work=SimpleNamespace(agent_query="fixture task"),
         judge=SimpleNamespace(selection="score_first", score_direction="maximize")))
     monkeypatch.setenv("CODEX_AUTH_JSON_PATH", "/synthetic-credential")
-    monkeypatch.setattr(run, "RecordingDockerBackend", lambda **k: SimpleNamespace(image_exists=lambda image: True))
+    monkeypatch.setattr(run, "RecordingDockerBackend", lambda **k: SimpleNamespace(
+        image_exists=lambda image: True, registration_options=lambda provider: {}))
     def stop_before_solver(**kwargs):
         # Exercise the real adapter at the CLI handoff, including both context
         # modes. SForge must not resume an exited LoopX scheduler.
@@ -1394,7 +1395,7 @@ def test_edgebench_cli_merges_sforge_env_with_explicit_runner_config(
     def backend(**kwargs):
         assert kwargs['oauth_proxy'] is custom
         assert kwargs.get('docker_host_bridge', False) is bridge
-        return SimpleNamespace(image_exists=lambda image: True)
+        return SimpleNamespace(image_exists=lambda image: True, registration_options=lambda provider: {})
     monkeypatch.setattr(run, 'RecordingDockerBackend', backend)
     def stop_before_solver(**kwargs):
         config = kwargs['config']
@@ -1440,7 +1441,8 @@ def test_native_checkpoint_storage_guidance_reaches_both_prompt_surfaces(tmp_pat
         cwd="/task", work_image_key="work", judge_image_key="judge", internet=False,
         game_mode=False, submit_paths=["solver.py"], work=SimpleNamespace(agent_query="Original acceptance."),
         judge=SimpleNamespace(selection="score_first", score_direction="maximize")))
-    monkeypatch.setattr(run, "RecordingDockerBackend", lambda **k: SimpleNamespace(image_exists=lambda image: True))
+    monkeypatch.setattr(run, "RecordingDockerBackend", lambda **k: SimpleNamespace(
+        image_exists=lambda image: True, registration_options=lambda provider: {}))
     expected_delivery = "on" if context_delivery is None else context_delivery
     def before_solver(**kwargs):
         worker = kwargs["agent"]
@@ -1496,6 +1498,7 @@ def test_edgebench_cli_reuses_local_probe_and_reports_selection_before_model(tmp
         game_mode=False, submit_paths=["solver.py"], work=SimpleNamespace(agent_query="Original acceptance."),
         judge=SimpleNamespace(selection="score_first", score_direction="maximize")))
     backend = SimpleNamespace(image_exists=lambda image: True,
+                              registration_options=lambda provider: {},
                               exec_run=lambda *a, **k: SimpleNamespace(exit_code=0))
     monkeypatch.setattr(run, "RecordingDockerBackend", lambda **k: backend)
 

benchmark/tests/test_sforge_warm_start.py

@@ -347,7 +347,8 @@ def test_warm_cli_checks_sealed_native_child_limit_before_activation(point, tmp_
         work=SimpleNamespace(agent_query="Synthetic task"),
         judge=SimpleNamespace(selection="score_first", score_direction="maximize")))
     monkeypatch.setattr(checkpoint_supervisor, "CheckpointSupervisor", lambda *args, **kwargs: object())
-    monkeypatch.setattr(entry, "RecordingDockerBackend", lambda **kwargs: SimpleNamespace(image_exists=lambda _: True))
+    monkeypatch.setattr(entry, "RecordingDockerBackend", lambda **kwargs: SimpleNamespace(
+        image_exists=lambda _: True, registration_options=lambda provider: {}))
 
     class ReachedNativeActivation(Exception):
         pass