diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py index 59a839480b..e624d2e171 100644 --- a/benchmark/edgebench/run.py +++ b/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(), diff --git a/benchmark/runtime/EXECUTION_OBSERVATIONS.md b/benchmark/runtime/EXECUTION_OBSERVATIONS.md index f4858774b7..844c44aee1 100644 --- a/benchmark/runtime/EXECUTION_OBSERVATIONS.md +++ b/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，组成员变化或数据 不完整时跳过该次采样。组成员稳定时，内存仍可能在两次读取之间转移，累加值 diff --git a/benchmark/runtime/harbor.py b/benchmark/runtime/harbor.py index 8d2e353984..7ec96e56a9 100644 --- a/benchmark/runtime/harbor.py +++ b/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, diff --git a/benchmark/runtime/sforge.py b/benchmark/runtime/sforge.py index 66b94a75f3..f4e200e334 100644 --- a/benchmark/runtime/sforge.py +++ b/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 diff --git a/benchmark/tests/test_sforge_runtime.py b/benchmark/tests/test_sforge_runtime.py index f1ad1b25cc..336cecac39 100644 --- a/benchmark/tests/test_sforge_runtime.py +++ b/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 diff --git a/benchmark/tests/test_task_entry.py b/benchmark/tests/test_task_entry.py index 42f11fa6d5..cb2ac46142 100644 --- a/benchmark/tests/test_task_entry.py +++ b/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