Integrate C1 and evidence-driven benchmark candidate selection

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
@@ -126,6 +135,25 @@ Malformed optional diagnostics preserve reliable host measurements and retain
 a bounded diagnostic-error receipt. Evidence paths must be fresh; raw output
 stays local. The probe neither submits nor accesses hidden evaluation.
 
+When simultaneous memory could change a candidate choice, run the complete
+representative workload with the same probe and pass
+`--memory-limit-bytes <applicable-reference-limit>`. On Linux the producer
+estimates group proportional set size by reading one PID at a time. It skips
+samples with unstable membership. A stable group can move retained memory
+between PID reads, so the sampled sum can overstate simultaneous occupancy.
+The adjacent `.memory.json` binds that estimate and reference to the exact
+candidate and input revision. The main observation cites its digest only after
+a successful write. Missing samples stay unknown. A sidecar failure preserves
+the original process result. Processes in other groups and peaks between samples
+are outside coverage, and PSS accounting differs from a container limit.
+
+When this estimate could change a choice, check actual overlap independently
+and record the supported next comparison in the existing Yardstick decision.
+Keep the best authoritative evidence, reuse the matching incumbent observation,
+and compare a lower-footprint mechanism on the same complete workload where
+overlap matters. The estimate alone supplies no limit verdict or selected
+action. A native timeout alone supplies no memory observation.
+
 The adjacent `.candidate.json` preserves selected relative paths and hashes.
 `candidate_sha256` hashes the sorted JSON file manifest, not native archive
 container bytes. Pins cover only declared files at the probe boundaries, not
@@ -189,6 +217,24 @@ Stop、原预算和真实阻塞继续有效，不为填满时长制造任务。
 变化拒绝归因；文件清单哈希不是原生归档哈希，只覆盖明确选择的文件及探针
 边界，不覆盖传递依赖或独立写入者，结果仍须证明对应实际评测快照和输入版本。
 
+SForge heartbeat worker 可显式传入 `--execution-observation local_probe`，
+沿用共享安装路径，并在 runner 和 worker 回执中记录选择。入口以隔离 Python
+执行已部署的探针源码，候选路径仍相对于任务工作目录。安装时用 worker 身份
+核对一次 `--help`。该入口不可用时在模型进入前报告安装失败。测量只在
+worker 自己调用探针时发生。
+
+当并行内存占用会影响候选选择时，用同一入口执行完整代表性工作集并传入
+`--memory-limit-bytes <参考上限>`。Linux 逐 PID 读取 PSS，组成员变化或数据
+不完整时跳过该次采样。组成员稳定时，内存仍可能在两次读取之间转移，累加值
+可能高于真实同时占用。相邻 `.memory.json` 将估计和参考值绑定到候选及输入
+版本。写入成功才把其摘要记入原观测，写入失败仍保留原进程结果。组外进程和
+采样间峰值未覆盖，PSS 与容器内存限额的统计口径有差异。
+
+这项估计会影响选择时，独立核对真实同时占用，再通过原 Yardstick 写入有证据
+的下一比较。保留最好的权威结果，复用同口径 incumbent 测量；确有并行保留
+影响时，把下一次有界试验用于同一完整工作集上的较低占用机制。估计本身不产生
+超限判定或自动选择。只有超时回执时，内存保持未知。
+
 当前授权工作实际产生或取得新的决策证据后，才进入相应工作段的检查点。旧反馈读回、
 记账和没有新增信息的确定性重复验证只是上下文，应先用于推进原任务。使用当前
 settlement plan 的 `durable_writeback.semantic_closeout_command_template`，

benchmark/runtime/checkpoint.py

@@ -229,26 +229,32 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
                "--runtime-root", env["LOOPX_RUNTIME_ROOT"], "quota", "should-run",
                "--goal-id", env["LOOPX_GOAL_ID"], "--agent-id", env["LOOPX_AGENT_ID"],
                "--runtime-profile", "generic_cli", "--turn-instance-id", receipt["turn_id"],
-               "--available-capability", "shell", "--available-capability", "filesystem_write",
-               "--decision-output-root", str(capture_root)]
+               "--available-capability", "shell", "--available-capability", "filesystem_write"]
     native_context = env.get("LOOPX_CONTEXT_DELIVERY") == "on" and bool(env.get("LOOPX_CODEX_CONTEXT_DIR"))
     if native_context:
-        # Capture the complete Goal now, but let the actual native tool select
-        # full/delta after SessionStart. Do not consume or invalidate continuity
-        # while the Host is preparing a checkpoint outside the model session.
-        command += ["--context-delivery", "on", "--codex-context-dir", env["LOOPX_CODEX_CONTEXT_DIR"],
+        # The native reader selects full/delta after SessionStart. Ask the
+        # existing preparation route for the committed capture reference now.
+        capture_directory = capture_root / "decision-original"
+        command += ["--decision-output-dir", str(capture_directory), "--prepared-capture-reference",
+                    "--context-delivery", "on", "--codex-context-dir", env["LOOPX_CODEX_CONTEXT_DIR"],
                     "--defer-context-presentation"]
+    else:
+        command += ["--decision-output-root", str(capture_root)]
     guard_run = subprocess.run(command, cwd=env["LOOPX_PROJECT"], env=env, capture_output=True,
                                text=True, timeout=120, check=True)
     projected = json.loads(guard_run.stdout)
     captures = list(capture_root.glob("decision-*/decision.json"))
     if len(captures) != 1:
         raise ValueError("checkpoint requires exactly one captured original quota decision; do not replay the guard")
-    raw_guard = captures[0].read_bytes()
+    source_path = captures[0]
+    raw_guard = source_path.read_bytes()
+    guard_digest = hashlib.sha256(raw_guard).hexdigest() if native_context else None
     guard = json.loads(raw_guard)
     if (not isinstance(guard, dict) or guard.get("ok") is not True or
             not isinstance(projected, dict) or projected.get("ok") is not True):
         raise ValueError("checkpoint requires a successful original quota guard")
+    if native_context and projected.get("capture_sha256") != guard_digest:
+        raise ValueError("checkpoint prepared capture differs from the original guard; do not replay the guard")
     # A previous preflight on this wake may already have allowed model work,
     # even if its optional attempt marker (or final receipt) could not be saved.
     # Such reentry can verify an existing point, but cannot capture a later one.
@@ -258,23 +264,15 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
     if native_context:
         from .context_delivery import preflight_context_instruction
 
-        # A repeated preflight can replace wake-level paths. Bind both native
-        # read commands to the exact CLI view of this particular guard.
-        view_bytes = guard_run.stdout.encode("utf-8")
-        source_path = captures[0]
-        if view_bytes != raw_guard:
-            source_path = capture_root / "checkpoint-start-view.json"
-            with source_path.open("xb") as stream:
-                os.fchmod(stream.fileno(), 0o600)
-                stream.write(view_bytes)
-        reading = preflight_context_instruction(env, source_path,
-            hashlib.sha256(view_bytes).hexdigest(), receipt["turn_id"], guide_in_host=guide_in_host)
+        assert guard_digest is not None
+        reading = preflight_context_instruction(env, source_path, guard_digest,
+            receipt["turn_id"], guide_in_host=guide_in_host)
     else:
         view_path = guard_path.with_name("checkpoint-start-view.json")
         view_path.write_text(guard_run.stdout, encoding="utf-8")
         view_path.chmod(0o600)
         reading = (f"Read the complete CLI view at {view_path} before work, then follow its interaction_contract "
-            "and the next step in the replan review when present. ")
+            "and replan review/writeback reads when present. ")
     preloaded_body = ("Host checkpoint preflight has already executed this wake's original quota guard once. "
         + reading +
         "Do not repeat the initial guard below; use its selection or recovery command when required.\n\n" + body)

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,