diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py
index eb6114b809..0d3e9c477d 100644
--- a/benchmark/edgebench/run.py
+++ b/benchmark/edgebench/run.py
@@ -124,6 +124,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("--provider-request-admission", action="store_true",
                         help="Opt-in canonical admission before each heartbeat custom Responses request")
     parser.add_argument("--native-goal-continue", action="store_true",
@@ -204,6 +206,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")
@@ -374,6 +378,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,
@@ -444,6 +450,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 ccf8e9cfae..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
@@ -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`，
diff --git a/benchmark/runtime/harbor.py b/benchmark/runtime/harbor.py
index 6ecd140fee..2d1da47048 100644
--- a/benchmark/runtime/harbor.py
+++ b/benchmark/runtime/harbor.py
@@ -366,11 +366,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/research_probe.py b/benchmark/runtime/research_probe.py
index 9b4f87025c..316be73ef0 100644
--- a/benchmark/runtime/research_probe.py
+++ b/benchmark/runtime/research_probe.py
@@ -6,6 +6,7 @@
 """
 from __future__ import annotations
 
+from _thread import LockType
 import argparse
 import hashlib
 import json
@@ -14,6 +15,8 @@
 from pathlib import Path
 import signal
 import subprocess
+import sys
+import threading
 import time
 
 
@@ -45,6 +48,9 @@ def execution_observation_guidance() -> str:
         "use `loopx-research-probe --help` to wrap the relevant local execution within its existing "
         "budget. Do not repeat an experiment merely to produce an attachment. Retain actual host "
         "elapsed time, exit or timeout, and selected-file hashes; keep unmeasured values unknown. "
+        "When aggregate memory could change the choice, reuse or estimate the complete workload with "
+        "the applicable reference using --memory-limit-bytes. If the estimate is decision-critical, "
+        "verify real overlap and compare a lower-footprint method on the same workload and accepted outcome. "
         "Method ids are producer declarations and optional solver counters are reports, not "
         "independently verified execution. Attach available execution_observation through ordinary "
         "Explore writeback with matching candidate, input revision and evidence refs. This tool "
@@ -69,14 +75,81 @@ def candidate_snapshot(paths: list[Path], workspace: Path) -> tuple[str, list[di
     return hashlib.sha256(encoded.encode()).hexdigest(), files
 
 
+def _group_members(group_id: int) -> dict[int, bytes]:
+    members = {}
+    for entry in Path("/proc").iterdir():
+        if not entry.name.isdecimal():
+            continue
+        try:
+            stat = (entry / "stat").read_bytes().rsplit(b")", 1)[1].split()
+        except (FileNotFoundError, ProcessLookupError):
+            continue
+        # Fields after comm start with state, ppid and pgrp. Use starttime to
+        # distinguish PID reuse, and skip zombies which no longer own memory.
+        if int(stat[2]) == group_id and int(stat[3]) == group_id and stat[0] != b"Z":
+            members[int(entry.name)] = stat[19]
+    return members
+
+
+def _group_pss_sample(group_id: int) -> tuple[int, int] | None:
+    try:
+        members = _group_members(group_id)
+        if not members:
+            return None
+        footprint = 0
+        for pid in members:
+            rollup = Path(f"/proc/{pid}/smaps_rollup").read_bytes()
+            pss = next(line.split() for line in rollup.splitlines() if line.startswith(b"Pss:"))
+            if len(pss) != 3 or pss[2] != b"kB" or not pss[1].isdigit():
+                return None
+            footprint += int(pss[1]) * 1024
+        # This checks membership only. PSS reads happen one PID at a time,
+        # so a stable group can still move memory between those reads.
+        if members == _group_members(group_id):
+            return footprint, len(members)
+    except (OSError, IndexError, ValueError, StopIteration):
+        pass  # Racing/denied reads are unknown, not a smaller valid sample.
+    return None
+
+
+def _observe_group_pss(group_id: int, stopped: threading.Event, samples: dict, lock: LockType) -> None:
+    while not stopped.is_set():
+        sample = _group_pss_sample(group_id)
+        with lock:
+            if stopped.is_set():
+                return
+            if sample is None:
+                samples["incomplete_samples"] += 1
+            else:
+                peak, members = sample
+                samples["complete_samples"] += 1
+                if samples["peak_sampled_pss_bytes"] is None or peak > samples["peak_sampled_pss_bytes"]:
+                    samples.update(peak_sampled_pss_bytes=peak, members_at_peak=members)
+        stopped.wait(0.25)
+
+
+def _memory_observation(samples: dict, source: dict, limit_bytes: int, evidence_ref: str) -> dict:
+    return {
+        "measurement_basis": "linux-sequential-process-group-pss-v1",
+        "candidate_sha256": source["candidate_sha256"],
+        "input_revision": source["input_revision"],
+        "executed_method_id": source["executed_method_id"],
+        "probe_evidence_ref": evidence_ref.strip(),
+        "reference_memory_bytes": limit_bytes,
+        **samples,
+    }
+
+
 def run_probe(*, command: list[str], candidate_files: list[Path], output: Path,
               input_revision: str, planned_method_id: str, executed_method_id: str,
               evidence_ref: str, limit_seconds: float, change_reason: str | None = None,
-              diagnostics: Path | None = None) -> dict:
+              diagnostics: Path | None = None, memory_limit_bytes: int | None = None) -> dict:
     from loopx.capabilities.explore.execution_observation import normalize_execution_observation
 
     if not command or not math.isfinite(limit_seconds) or limit_seconds <= 0:
         raise ValueError("a command and finite positive local probe limit are required")
+    if memory_limit_bytes is not None and (type(memory_limit_bytes) is not int or memory_limit_bytes <= 0):
+        raise ValueError("memory_limit_bytes requires a positive integer")
     workspace = Path.cwd().resolve()
     before, files = candidate_snapshot(candidate_files, workspace)
     observation = dict(schema_version="explore_execution_observation_v0",
@@ -87,7 +160,11 @@ def run_probe(*, command: list[str], candidate_files: list[Path], output: Path,
     normalize_execution_observation(observation)  # Reject bad attribution before execution.
     if diagnostics is not None and diagnostics.exists():
         raise FileExistsError("use a fresh diagnostic path; previous observations cannot describe this probe")
-    # Reserve before any process effect. Existing evidence is never overwritten.
+    # Reserve the primary receipt before any process effect. A stale sidecar
+    # cannot be adopted as a new probe, including a dangling symbolic link.
+    memory_output = output.with_suffix(output.suffix + ".memory.json")
+    if memory_limit_bytes is not None and os.path.lexists(memory_output):
+        raise FileExistsError("use a fresh memory observation path")
     with output.open("x", encoding="utf-8") as record:
         with output.with_suffix(output.suffix + ".candidate.json").open("x", encoding="utf-8") as source:
             source.write(json.dumps({"candidate_sha256": before, "files": files}, sort_keys=True) + "\n")
@@ -96,15 +173,33 @@ def run_probe(*, command: list[str], candidate_files: list[Path], output: Path,
             env["LOOPX_RESEARCH_DIAGNOSTICS"] = str(diagnostics.resolve())
         started = time.monotonic()
         timed_out = False
+        samples = (dict(complete_samples=0, incomplete_samples=0, peak_sampled_pss_bytes=None, members_at_peak=None)
+                   if memory_limit_bytes is not None else None)
         with output.with_suffix(output.suffix + ".stdout").open("xb") as stdout, \
              output.with_suffix(output.suffix + ".stderr").open("xb") as stderr:
             process = subprocess.Popen(command, cwd=workspace, env=env, stdout=stdout, stderr=stderr,
                                        start_new_session=True)
+            stopped = None
+            sampler = None
+            wait_seconds = limit_seconds
             try:
-                process.wait(timeout=limit_seconds)
+                if samples is not None and sys.platform == "linux":
+                    sampling_start = time.monotonic()
+                    stopped, lock = threading.Event(), threading.Lock()
+                    sampler = threading.Thread(target=_observe_group_pss,
+                        args=(process.pid, stopped, samples, lock), daemon=True)
+                    try:
+                        sampler.start()
+                    except RuntimeError:
+                        sampler = None  # Unavailable diagnostics must not cancel the probe.
+                    wait_seconds = max(0, limit_seconds - (time.monotonic() - sampling_start))
+                process.wait(timeout=wait_seconds)
             except subprocess.TimeoutExpired:
                 timed_out = True
             finally:
+                if stopped is not None:
+                    with lock:
+                        stopped.set()
                 # Stop the owned group, including descendants that outlive a
                 # successful leader; they must not continue changing a candidate.
                 try:
@@ -112,6 +207,8 @@ def run_probe(*, command: list[str], candidate_files: list[Path], output: Path,
                 except ProcessLookupError:
                     pass
                 process.wait()
+                if sampler is not None:
+                    sampler.join(timeout=0.25)
         observation["elapsed_seconds"] = time.monotonic() - started
         observation["exit_code"] = process.returncode
         observation["stop_reason"] = "host_timeout" if timed_out else f"process_exit:{process.returncode}"
@@ -125,6 +222,19 @@ def run_probe(*, command: list[str], candidate_files: list[Path], output: Path,
         # completed execution or change its real exit/timeout status.
         result = normalize_execution_observation(observation)
         try:
+            if samples is not None and memory_limit_bytes is not None:
+                measurement = _memory_observation(samples, result, memory_limit_bytes, evidence_ref)
+                encoded = json.dumps(measurement, sort_keys=True, separators=(",", ":")).encode() + b"\n"
+                try:
+                    with memory_output.open("xb", buffering=0) as memory_record:
+                        if memory_record.write(encoded) != len(encoded):
+                            raise OSError("incomplete optional memory observation")
+                        os.fsync(memory_record.fileno())
+                except OSError:
+                    print("Optional process-group memory observation could not be retained.", file=sys.stderr)
+                else:
+                    result["evidence_refs"].append("resource:" + hashlib.sha256(encoded).hexdigest())
+                    result = normalize_execution_observation(result)
             if diagnostics is not None:
                 try:
                     if diagnostics.exists():
@@ -178,6 +288,8 @@ def main() -> int:
     parser.add_argument("--change-reason")
     parser.add_argument("--evidence-ref", required=True)
     parser.add_argument("--limit-seconds", type=float, required=True)
+    parser.add_argument("--memory-limit-bytes", type=int,
+        help="Write a .memory.json diagnostic with sequential Linux PSS samples and this reference (bytes)")
     parser.add_argument("--diagnostics", type=Path)
     parser.add_argument("command", nargs=argparse.REMAINDER)
     args = vars(parser.parse_args())
diff --git a/benchmark/runtime/sforge.py b/benchmark/runtime/sforge.py
index c742e69b26..a446e9dd51 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,
                  provider_request_admission: bool = False,
                  native_goal_continue: bool = False,
                  replan_after_turns: int | None = None,
@@ -114,6 +116,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-"):
@@ -336,6 +341,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 {}),
                 **({"provider_request_admission": True} if self.provider_request_admission else {}),
                 native_goal_continue=self.native_goal_continue,
                 replan_after_turns=self.replan_after_turns,
@@ -435,6 +442,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 571e83e78f..6b58f0c3a8 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
diff --git a/tests/control_plane/test_research_execution_loop.py b/tests/control_plane/test_research_execution_loop.py
index cee7f893ca..50e0570ef3 100644
--- a/tests/control_plane/test_research_execution_loop.py
+++ b/tests/control_plane/test_research_execution_loop.py
@@ -141,6 +141,180 @@ def test_probe_cli_preserves_a_failed_candidate_exit_and_its_observation(tmp_pat
     observed = json.loads(output.read_text())
     assert observed["exit_code"] == 3 and observed["search_iterations"] is None
     assert json.loads(output.with_suffix(".json.candidate.json").read_text())["candidate_sha256"] == observed["candidate_sha256"]
+    assert not output.with_suffix(".json.memory.json").exists()
+
+
+def test_memory_observation_without_samples_leaves_timeout_and_next_method_unknown(tmp_path, monkeypatch):
+    import benchmark.runtime.research_probe as producer
+
+    monkeypatch.setattr(producer, "_group_pss_sample", lambda _group: None)
+    observed = probe(tmp_path, monkeypatch, source="import time; time.sleep(1)",
+                     limit_seconds=0.08, memory_limit_bytes=1024)
+    sidecar = tmp_path / "observation.json.memory.json"
+    evidence = json.loads(sidecar.read_bytes())
+    assert observed["stop_reason"] == "host_timeout" and observed["exit_code"] < 0
+    assert evidence["candidate_sha256"] == observed["candidate_sha256"]
+    assert evidence["complete_samples"] == 0 and evidence["peak_sampled_pss_bytes"] is None
+    assert "observed_limit_exceeded" not in evidence and "next_action" not in evidence
+    assert "resource:" + hashlib.sha256(sidecar.read_bytes()).hexdigest() in observed["evidence_refs"]
+
+
+def test_changed_candidate_cannot_publish_memory_evidence(tmp_path, monkeypatch):
+    with pytest.raises(ValueError, match="candidate changed"):
+        probe(tmp_path, monkeypatch,
+              source="from pathlib import Path; Path(__file__).write_text('print(0)')", memory_limit_bytes=1024)
+    assert json.loads((tmp_path / "observation.json").read_text())["ok"] is False
+    assert not (tmp_path / "observation.json.memory.json").exists()
+
+
+@pytest.mark.skipif(sys.platform != "linux", reason="requires a real Linux file size limit")
+def test_optional_memory_file_short_write_preserves_cli_result(tmp_path):
+    candidate = tmp_path / "candidate.py"
+    candidate.write_text("print('candidate finished'); raise SystemExit(7)")
+    # Change only the kernel limit for the optional file. The memory read
+    # and resulting real FileIO.write still use their production paths.
+    (tmp_path / "sitecustomize.py").write_text("""
+from contextlib import contextmanager
+from pathlib import Path
+import resource
+
+original_open = Path.open
+
+@contextmanager
+def limit_size(stream):
+    soft, hard = resource.getrlimit(resource.RLIMIT_FSIZE)
+    try:
+        resource.setrlimit(resource.RLIMIT_FSIZE, (64, hard))
+        with stream as opened:
+            yield opened
+    finally:
+        resource.setrlimit(resource.RLIMIT_FSIZE, (soft, hard))
+
+def open_with_limit(path, *args, **kwargs):
+    stream = original_open(path, *args, **kwargs)
+    return limit_size(stream) if path.name.endswith(".memory.json") else stream
+
+Path.open = open_with_limit
+""")
+    output = tmp_path / "observed.json"
+    process = subprocess.run([sys.executable, "-m", "benchmark.runtime.research_probe",
+        "--candidate-file", str(candidate), "--output", str(output), "--input-revision", "fixture-v1",
+        "--planned-method-id", "method:work", "--executed-method-id", "method:work",
+        "--evidence-ref", "probe:failed", "--limit-seconds", "5", "--memory-limit-bytes", "1024",
+        "--", sys.executable, str(candidate)], cwd=tmp_path, capture_output=True, text=True, timeout=10,
+        env={**os.environ, "PYTHONPATH": os.pathsep.join((str(tmp_path), str(fixture.REPO_ROOT)))})
+    assert process.returncode == 7, process.stderr
+    assert "Optional process-group memory observation could not be retained." in process.stderr
+    assert output.with_suffix(".json.memory.json").stat().st_size == 64
+    assert output.with_suffix(".json.stdout").read_text().strip() == "candidate finished"
+    observation = json.loads(output.read_bytes())
+    assert json.loads(process.stdout) == observation and observation["exit_code"] == 7
+    assert not any(ref.startswith("resource:") for ref in observation["evidence_refs"])
+
+
+@pytest.mark.skipif(sys.platform != "linux" or not Path("/proc/self/smaps_rollup").exists(),
+                    reason="requires live Linux process-group PSS")
+def test_sampled_memory_compares_complete_workloads_with_equal_outputs(tmp_path):
+    limit_bytes = 38 * 1024 * 1024
+    revision = "two-complete-shards-v1"
+    shard_size = 22 * 1024 * 1024
+    # All three mechanisms compute the same independent workload. One expands
+    # both buffers at once. A streaming contender bounds retained state.
+    template = """
+import json
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+if len(sys.argv) == 2:
+    marker = int(sys.argv[1])
+    if {streaming}:
+        total = 0
+        chunk_size = 1024 * 1024
+        for offset in range(0, {shard_size}, chunk_size):
+            scratch = bytearray([marker]) * min(chunk_size, {shard_size} - offset)
+            total += sum(scratch)
+    else:
+        scratch = bytearray([marker]) * {shard_size}
+        total = sum(scratch)
+    # The parent releases this worker only after reading the amount of live
+    # payload. In the concurrent case both buffers are held at that barrier.
+    print(total, len(scratch), flush=True)
+    sys.stdin.readline()
+else:
+    def shard(marker):
+        return subprocess.Popen(
+            [sys.executable, __file__, str(marker)],
+            stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True,
+        )
+    def read(process):
+        total, retained = (int(item) for item in process.stdout.readline().split())
+        assert process.poll() is None
+        return total, retained
+    def finish(process):
+        process.stdin.close()
+        assert process.wait(timeout=3) == 0
+    first = shard(1)
+    a, retained_a = read(first)
+    if {parallel}:
+        second = shard(2)
+        b, retained_b = read(second)
+        simultaneous_payload = retained_a + retained_b
+        time.sleep(0.60)
+        finish(first)
+        finish(second)
+    else:
+        time.sleep(0.60)
+        finish(first)
+        second = shard(2)
+        b, retained_b = read(second)
+        simultaneous_payload = max(retained_a, retained_b)
+        time.sleep(0.60)
+        finish(second)
+    receipt = dict(values=[a, b], simultaneous_payload=simultaneous_payload)
+    Path("last-workload.json").write_text(json.dumps(receipt))
+    print(json.dumps(receipt))
+"""
+
+    def execute(name, *, streaming=False, parallel, method_id):
+        candidate = tmp_path / f"{name}.py"
+        candidate.write_text(template.format(shard_size=shard_size, streaming=streaming, parallel=parallel))
+        output = tmp_path / f"{name}.observation.json"
+        command = [sys.executable, "-m", "benchmark.runtime.research_probe",
+            "--candidate-file", str(candidate), "--output", str(output), "--input-revision", revision,
+            "--planned-method-id", method_id, "--executed-method-id", method_id,
+            "--evidence-ref", f"probe:{name}", "--limit-seconds", "6",
+            "--memory-limit-bytes", str(limit_bytes), "--", sys.executable, str(candidate)]
+        process = subprocess.run(command, cwd=tmp_path, env={**os.environ, "PYTHONPATH": str(fixture.REPO_ROOT)},
+                                 text=True, capture_output=True, timeout=10)
+        assert process.returncode == 0, process.stderr
+        execution = json.loads(output.with_suffix(".json.stdout").read_text())
+        assert execution["values"] == [shard_size, 2 * shard_size]
+        observation = json.loads(process.stdout)
+        memory_bytes = output.with_suffix(".json.memory.json").read_bytes()
+        memory = json.loads(memory_bytes)
+        ref = "resource:" + hashlib.sha256(memory_bytes).hexdigest()
+        assert ref in observation["evidence_refs"]
+        assert memory["candidate_sha256"] == observation["candidate_sha256"]
+        assert memory["input_revision"] == revision and memory["complete_samples"] > 0
+        assert memory["reference_memory_bytes"] == limit_bytes
+        assert "observed_limit_exceeded" not in memory and "next_action" not in memory
+        return observation, memory, execution
+
+    baseline, incumbent_memory, incumbent_run = execute(
+        "incumbent", parallel=False, method_id="method:sequential")
+    challenger, expanded_memory, expanded_run = execute(
+        "expanded", parallel=True, method_id="method:expanded")
+    contender, streamed_memory, streamed_run = execute(
+        "streamed", streaming=True, parallel=True, method_id="method:streaming")
+    assert incumbent_run["simultaneous_payload"] == shard_size < limit_bytes
+    assert expanded_run["simultaneous_payload"] == 2 * shard_size > limit_bytes
+    assert streamed_run["simultaneous_payload"] == 2 * 1024 * 1024 < limit_bytes
+    assert expanded_memory["members_at_peak"] == 3
+    assert (streamed_memory["peak_sampled_pss_bytes"] < incumbent_memory["peak_sampled_pss_bytes"]
+            < expanded_memory["peak_sampled_pss_bytes"])
+    assert len({baseline["candidate_sha256"], challenger["candidate_sha256"], contender["candidate_sha256"]}) == 3
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])