Opt in to local work probes from SForge heartbeat

benchmark/edgebench/run.py

@@ -126,6 +126,8 @@ def main(argv=None):
                         help="Opt-in short heartbeat context with same-invocation full captures")
     parser.add_argument("--context-delivery", choices=("off", "on"),
                         help="Native Codex context delivery: on by default for heartbeat workers; pass off to disable")
+    parser.add_argument("--execution-observation", choices=("local_probe",),
+                        help="Optional local execution probe for heartbeat workers")
     parser.add_argument("--native-child-limit", type=int,
                         help="Explicit native child capacity and optional task delegation authorization, shared across study arms; omitted preserves existing behavior")
     parser.add_argument("--provider-request-admission", action="store_true",
@@ -227,6 +229,8 @@ def main(argv=None):
         args.context_delivery = Execution(mode="heartbeat" if args.worker.startswith("heartbeat-") else "plain").context_delivery
     if args.context_delivery != "off" and not args.worker.startswith("heartbeat-"):
         parser.error("--context-delivery requires a heartbeat worker")
+    if args.execution_observation is not None and not args.worker.startswith("heartbeat-"):
+        parser.error("--execution-observation requires a heartbeat worker")
     if args.provider_request_admission:
         if not args.worker.startswith("heartbeat-"):
             parser.error("--provider-request-admission requires a heartbeat worker")
@@ -409,6 +413,8 @@ def main(argv=None):
                          task_entry=args.task_entry,
                          turn_envelope=args.turn_envelope,
                          context_delivery=args.context_delivery,
+                         **({"execution_observation": args.execution_observation}
+                            if args.execution_observation is not None else {}),
                          **({"provider_request_admission": True} if args.provider_request_admission else {}),
                          native_goal_continue=args.native_goal_continue,
                          replan_after_turns=args.replan_after_turns,
@@ -485,6 +491,8 @@ def main(argv=None):
         "sforge_source": active_source_receipt(),
         **turn_envelope_receipt(explicit=args.turn_envelope),
         **({"context_delivery": "on"} if args.context_delivery != "off" else {}),
+        **({"execution_observation": args.execution_observation}
+           if args.execution_observation is not None else {}),
         **({"provider_request_admission": "on"} if args.provider_request_admission else {}),
         **({"native_goal_continue": True} if args.native_goal_continue else {}),
         "task_sha256": hashlib.sha256(task_file.read_bytes()).hexdigest(),

benchmark/runtime/EXECUTION_OBSERVATIONS.md

@@ -39,6 +39,15 @@ merely to produce an attachment. Configuration and readback invoke neither a
 probe nor a model. Runtime receipts record the selected producer, which does
 not attest execution, adoption or improvement.
 
+The native SForge runner accepts the same producer with
+`--execution-observation local_probe` on a heartbeat worker. It forwards the
+selection to the shared installer and records it in both runner and worker
+receipts. The wrapper uses isolated Python to execute the staged probe source
+while keeping the task working directory for candidate paths. The installer
+checks the wrapper's `--help` once as the worker user before model entry.
+An unavailable wrapper fails setup for an explicitly selected producer.
+The observation runs only when a worker invokes it.
+
 Choose the settlement plan's ordinary `command_template` for in-flight progress
 without new decision evidence. Use `semantic_closeout_command_template` for new
 evidence from this admitted work, a route change or Todo completion; bind the
@@ -208,6 +217,12 @@ Stop、原预算和真实阻塞继续有效，不为填满时长制造任务。
 变化拒绝归因；文件清单哈希不是原生归档哈希，只覆盖明确选择的文件及探针
 边界，不覆盖传递依赖或独立写入者，结果仍须证明对应实际评测快照和输入版本。
 
+SForge heartbeat worker 可显式传入 `--execution-observation local_probe`，
+沿用共享安装路径，并在 runner 和 worker 回执中记录选择。入口以隔离 Python
+执行已部署的探针源码，候选路径仍相对于任务工作目录。安装时用 worker 身份
+核对一次 `--help`。该入口不可用时在模型进入前报告安装失败。测量只在
+worker 自己调用探针时发生。
+
 当并行内存占用会影响候选选择时，用同一入口执行完整代表性工作集并传入
 `--memory-limit-bytes <参考上限>`。Linux 逐 PID 读取 PSS，组成员变化或数据
 不完整时跳过该次采样。组成员稳定时，内存仍可能在两次读取之间转移，累加值

benchmark/runtime/harbor.py

@@ -370,11 +370,22 @@ async def install(self, environment: BaseEnvironment) -> None:
             if doctor_payload.get("ok") is not True:
                 raise RuntimeError(f"LoopX doctor failed: {doctor_payload}")
             if self.execution_observation is not None:
-                wrapper = (f"#!/bin/sh\nexec env PYTHONPATH={shlex.quote(_SRC)} "
-                           f"{_PYTHON}/bin/python3 -m benchmark.runtime.research_probe \"$@\"\n")
-                await self.exec_as_agent(environment,
-                    command=f"printf %s {shlex.quote(wrapper)} > {_PROFILE}/bin/loopx-research-probe; "
-                            f"chmod 0755 {_PROFILE}/bin/loopx-research-probe", env=self._profile_env())
+                entrypoint = f"{_SRC}/benchmark/runtime/research_probe.py"
+                # Keep the task cwd for candidate paths without importing the
+                # probe or its dependencies from that cwd or user site.
+                bootstrap = (
+                    "import runpy\nimport sys\n"
+                    f"sys.path.insert(0, {_SRC!r})\n"
+                    f"runpy.run_path({entrypoint!r}, run_name='__main__')\n"
+                )
+                launcher = shlex.join([f"{_PYTHON}/bin/python3", "-I", "-c", bootstrap])
+                wrapper = f"#!/bin/sh\nexec {launcher} \"$@\"\n"
+                probe = shlex.quote(f"{_PROFILE}/bin/loopx-research-probe")
+                installed = await self.exec_as_agent(environment, env=self._profile_env(), timeout_sec=30,
+                    command=f"test ! -L {probe} && printf %s {shlex.quote(wrapper)} > {probe} && "
+                            f"chmod 0755 {probe} && {probe} --help")
+                if installed.return_code or "--candidate-file" not in (installed.stdout or ""):
+                    raise RuntimeError("Local execution probe is unavailable to the benchmark worker")
 
         receipt = {
             "loopx_commit": actual_commit,

benchmark/runtime/sforge.py

@@ -25,6 +25,7 @@
 from .harbor import (
     BenchmarkCodex, _CONTROL, _CODEX_HOME, _PYTHON, _SCHEDULER_STATE, _SRC,
 )
+from .research_probe import resolve_execution_observation
 from .scheduler import worker_command
 
 
@@ -93,6 +94,7 @@ def __init__(self, config, *, profile: str, cwd: str,
                  task_entry: str | None = None,
                  turn_envelope: bool = False,
                  context_delivery: str | None = None,
+                 execution_observation: str | None = None,
                  native_child_limit: int | None = None,
                  provider_request_admission: bool = False,
                  native_goal_continue: bool = False,
@@ -122,6 +124,9 @@ def __init__(self, config, *, profile: str, cwd: str,
             mode="heartbeat" if profile.startswith("heartbeat-") else "plain",
             context_delivery=context_delivery,
         ).context_delivery
+        self.execution_observation = resolve_execution_observation(execution_observation)
+        if self.execution_observation is not None and not profile.startswith("heartbeat-"):
+            raise ValueError("execution_observation requires a heartbeat worker")
         if type(provider_request_admission) is not bool:
             raise ValueError("provider_request_admission must be a boolean opt-in")
         if provider_request_admission and not profile.startswith("heartbeat-"):
@@ -344,6 +349,8 @@ def _install_worker(self, backend, handle, log_dir, logger):
                 task_entry=self.task_entry,
                 turn_envelope=self.turn_envelope,
                 context_delivery=self.context_delivery,
+                **({"execution_observation": self.execution_observation}
+                   if self.execution_observation is not None else {}),
                 native_child_limit=self.native_child_limit,
                 **({"provider_request_admission": True} if self.provider_request_admission else {}),
                 native_goal_continue=self.native_goal_continue,
@@ -446,6 +453,8 @@ def _install_worker(self, backend, handle, log_dir, logger):
             "explore_graph": self.profile == "heartbeat-explore",
             "explore_harness": self.profile == "heartbeat-explore",
             "feedback": self.feedback,
+            **({"execution_observation": self.execution_observation}
+               if self.execution_observation is not None else {}),
             **({"feedback_delivery": "codex_hooks", "feedback_payload": FEEDBACK_PAYLOAD}
                if self.feedback == "best-only" else {}),
             **(self.runtime._replan_receipt() if self.runtime and

benchmark/tests/test_sforge_runtime.py

@@ -1470,3 +1470,62 @@ def before_solver(**kwargs):
         run.main(args)
     receipt = json.loads((tmp_path / "runs/storage/fixture/runtime-receipt.json").read_text())
     assert receipt.get("context_delivery") == (expected_delivery if expected_delivery != "off" else None)
+
+
+@pytest.mark.parametrize(("profile", "selected"), [
+    ("heartbeat-explore", False), ("heartbeat-explore", True), ("native-goal", True),
+])
+def test_edgebench_cli_reuses_local_probe_and_reports_selection_before_model(tmp_path, monkeypatch, capsys,
+                                                                              profile, selected):
+    pytest.importorskip("sforge")
+    pytest.importorskip("harbor")
+    from types import SimpleNamespace
+    from benchmark.edgebench import run
+    from benchmark.runtime.harbor import BenchmarkCodex
+
+    producer = "local_probe" if selected else None
+    monkeypatch.delenv("OPENAI_BASE_URL", raising=False)
+    monkeypatch.setenv("LOOPX_SRC_DIR", str(tmp_path))
+    monkeypatch.setenv("LOOPX_EXPECTED_COMMIT", "fixture")
+    monkeypatch.setenv("CODEX_AUTH_JSON_PATH", "/synthetic-unused-credential")
+    (tmp_path / "fixture.json").write_text("{}")
+    monkeypatch.setattr(run, "source_pins", lambda *a: ("fixture", "fixture"))
+    monkeypatch.setattr(run, "load_benchmark", lambda *a: None)
+    monkeypatch.setattr(run, "make_task_spec", lambda *a: SimpleNamespace(
+        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")))
+    backend = SimpleNamespace(image_exists=lambda image: True,
+                              exec_run=lambda *a, **k: SimpleNamespace(exit_code=0))
+    monkeypatch.setattr(run, "RecordingDockerBackend", lambda **k: backend)
+
+    async def install_at_transport_boundary(self, environment):
+        assert environment.default_user == "agent"
+        assert self.execution_observation == producer
+        assert self._worker_env(cwd="/task").get("LOOPX_EXECUTION_OBSERVATION") == producer
+
+    monkeypatch.setattr(BenchmarkCodex, "install", install_at_transport_boundary)
+    trial = tmp_path / "runs" / "probe" / "fixture"
+
+    def stop_before_model(**kwargs):
+        worker = kwargs["agent"]
+        worker.install_stop_hook(backend, None, trial, None)
+        assert json.loads((trial / "worker-profile.json").read_text()).get("execution_observation") == producer
+        raise RuntimeError("isolated install boundary")
+
+    monkeypatch.setattr(run, "run_agent", stop_before_model)
+    args = ["--task", "fixture", "--tasks-dir", str(tmp_path), "--log-dir", str(tmp_path),
+            "--run-id", "probe", "--worker", profile, "--model", "fixture",
+            "--effort", "xhigh", "--judge-url", "http://127.0.0.1:9999"]
+    if selected:
+        args += ["--execution-observation", producer]
+    if profile == "native-goal":
+        with pytest.raises(SystemExit, match="2"):
+            run.main(args)
+        assert "--execution-observation requires a heartbeat worker" in capsys.readouterr().err
+        assert not trial.exists()
+        return
+    with pytest.raises(RuntimeError, match="isolated install boundary"):
+        run.main(args)
+    receipt = json.loads((trial / "runtime-receipt.json").read_text())
+    assert receipt.get("execution_observation") == producer

benchmark/tests/test_task_entry.py

@@ -125,13 +125,42 @@ def test_raw_worker_rejects_invalid_observation_before_creating_a_wake(tmp_path,
     assert not wakes.exists()
 
 
-@pytest.mark.parametrize("settings", [{}, {"execution_observation": "local_probe"}])
-def test_install_records_and_exposes_only_the_selected_observation_producer(tmp_path, monkeypatch, settings):
+@pytest.mark.parametrize("entry", ["omitted", "ready", "workspace-shadow", "symlink", "missing-interpreter"])
+def test_install_records_and_exposes_only_the_selected_observation_producer(tmp_path, monkeypatch, entry):
     pytest.importorskip("harbor")
-    from benchmark.runtime.harbor import BenchmarkCodex, CodexOffline
+    if not Path("/bin/sh").is_file():
+        pytest.skip("requires a POSIX shell for the actual worker entrypoint")
+    from benchmark.runtime import harbor
 
-    agent = BenchmarkCodex(logs_dir=tmp_path, model_name="fixture", **settings)
+    selected = entry != "omitted"
+    profile = tmp_path / "profile"
+    bin_dir = profile / "bin"
+    bin_dir.mkdir(parents=True)
+    interpreter_dir = tmp_path / "python" / "bin"
+    interpreter_dir.mkdir(parents=True)
+    if entry != "missing-interpreter":
+        (interpreter_dir / "python3").symlink_to(sys.executable)
+    wrapper = bin_dir / "loopx-research-probe"
+    sentinel = tmp_path / "untouched"
+    if entry == "symlink":
+        sentinel.write_text("unchanged")
+        wrapper.symlink_to(sentinel)
+    if entry == "workspace-shadow":
+        shadow = tmp_path / "benchmark"
+        (shadow / "runtime").mkdir(parents=True)
+        (shadow / "__init__.py").write_text("")
+        (shadow / "runtime" / "research_probe.py").write_text(
+            "from pathlib import Path\nPath('shadow-called').touch()\nprint('--candidate-file --memory-limit-bytes')\n")
+        shadow_loopx = tmp_path / "loopx"
+        shadow_loopx.mkdir()
+        (shadow_loopx / "__init__.py").write_text("from pathlib import Path\nPath('shadow-called').touch()\n")
+    monkeypatch.setattr(harbor, "_PROFILE", str(profile))
+    monkeypatch.setattr(harbor, "_PYTHON", str(interpreter_dir.parent))
+    monkeypatch.setattr(harbor, "_SRC", str(Path(__file__).resolve().parents[2]))
+    agent = harbor.BenchmarkCodex(logs_dir=tmp_path, model_name="fixture",
+                                 execution_observation="local_probe" if selected else None)
     commands = []
+    wrapper_reads = []
 
     async def no_op(*args, **kwargs):
         pass
@@ -141,28 +170,71 @@ async def stage(*args, **kwargs):
         return "fixture"
 
     async def execute(*args, **kwargs):
-        commands.append(kwargs["command"])
+        command = kwargs["command"]
+        commands.append(command)
+        if "benchmark/runtime/research_probe.py" in command:
+            result = subprocess.run(["/bin/sh", "-c", command], cwd=tmp_path,
+                                    env={"PATH": "/usr/bin:/bin", "PYTHONPATH": str(tmp_path),
+                                         "PYTHONDONTWRITEBYTECODE": "1"},
+                                    text=True, capture_output=True, timeout=10)
+            wrapper_reads.append(result)
+            return SimpleNamespace(stdout=result.stdout, stderr=result.stderr, return_code=result.returncode)
         return SimpleNamespace(stdout='{"ok":true}', stderr="", return_code=0)
 
     async def task_exec(**kwargs):
         return SimpleNamespace(stdout="/usr/local/bin:/usr/bin:/bin", return_code=0)
 
+    async def execute_root(*args, **kwargs):
+        assert "benchmark/runtime/research_probe.py" not in kwargs["command"]
+        return await execute(*args, **kwargs)
+
     for key in ("LOOPX_SRC_DIR", "LOOPX_PORTABLE_PYTHON", "LOOPX_NODE_DIR"):
         monkeypatch.setenv(key, str(tmp_path))
-    monkeypatch.setattr(CodexOffline, "install", no_op)
+    monkeypatch.setattr(harbor.CodexOffline, "install", no_op)
     monkeypatch.setattr(agent, "_stage_source", stage)
-    monkeypatch.setattr(agent, "exec_as_root", execute)
+    monkeypatch.setattr(agent, "exec_as_root", execute_root)
     monkeypatch.setattr(agent, "exec_as_agent", execute)
     monkeypatch.setattr(agent, "_get_env", lambda key: None)
-    asyncio.run(agent.install(SimpleNamespace(default_user="fixture", upload_dir=no_op, exec=task_exec)))
-    selected = bool(settings)
-    assert any("-m benchmark.runtime.research_probe" in command for command in commands) is selected
+    environment = SimpleNamespace(default_user="fixture", upload_dir=no_op, exec=task_exec)
+    if entry in {"symlink", "missing-interpreter"}:
+        with pytest.raises(RuntimeError, match="Local execution probe is unavailable"):
+            asyncio.run(agent.install(environment))
+        assert len(wrapper_reads) == 1 and wrapper_reads[0].returncode != 0
+        assert not any("> /logs/agent/loopx-install.json" in command for command in commands)
+        if entry == "symlink":
+            assert wrapper.is_symlink() and sentinel.read_text() == "unchanged"
+        return
+
+    asyncio.run(agent.install(environment))
+    assert any("benchmark/runtime/research_probe.py" in command for command in commands) is selected
+    assert len(wrapper_reads) == int(selected)
+    if selected:
+        assert os.access(wrapper, os.X_OK) and "--memory-limit-bytes" in wrapper_reads[0].stdout
+    else:
+        assert not wrapper.exists()
     command = next(command for command in commands if "> /logs/agent/loopx-install.json" in command)
     receipt = json.loads(shlex.split(command)[2])
     assert receipt.get("execution_observation") == ("local_probe" if selected else None)
     assert "research_loop" not in receipt
     assert receipt["task_entry"] == "loopx-planned"
     assert receipt.get("stride_strategy") is None
+    if entry == "workspace-shadow":
+        candidate = tmp_path / "candidate.py"
+        candidate.write_text("print('real probe kept the task cwd')")
+        invocation = subprocess.run([str(wrapper),
+            "--candidate-file", candidate.name, "--output", "probe.json", "--input-revision", "inputs-v1",
+            "--planned-method-id", "method:local", "--executed-method-id", "method:local",
+            "--evidence-ref", "probe:trusted", "--limit-seconds", "5", "--memory-limit-bytes", "8388608",
+            "--", sys.executable, candidate.name],
+            cwd=tmp_path, env={**os.environ, "PYTHONPATH": str(tmp_path)},
+            capture_output=True, text=True, timeout=12)
+        assert invocation.returncode == 0, invocation.stderr
+        observation = json.loads(invocation.stdout)
+        assert observation == json.loads((tmp_path / "probe.json").read_text())
+        memory = json.loads((tmp_path / "probe.json.memory.json").read_text())
+        assert memory["reference_memory_bytes"] == 8388608 and memory["candidate_sha256"] == observation["candidate_sha256"]
+        assert (tmp_path / "probe.json.stdout").read_text().strip() == "real probe kept the task cwd"
+        assert not (tmp_path / "shadow-called").exists()
 
 
 @pytest.fixture