From c103345d7e3684d692c92ca158d4e6661f1057a3 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 13:57:17 +0800
Subject: [PATCH 1/2] feat(benchmark): observe aggregate memory in local work
 probes

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/runtime/EXECUTION_OBSERVATIONS.md   |  32 ++++
 benchmark/runtime/research_probe.py           | 127 ++++++++++++++-
 .../test_research_execution_loop.py           | 153 ++++++++++++++++++
 3 files changed, 309 insertions(+), 3 deletions(-)

diff --git a/benchmark/runtime/EXECUTION_OBSERVATIONS.md b/benchmark/runtime/EXECUTION_OBSERVATIONS.md
index ccf8e9cfae..90d06b8592 100644
--- a/benchmark/runtime/EXECUTION_OBSERVATIONS.md
+++ b/benchmark/runtime/EXECUTION_OBSERVATIONS.md
@@ -126,6 +126,26 @@ 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
+samples proportional set size across the command's process group, including
+workers and the process holding their results. An unstable group is skipped.
+The adjacent `.memory.json` binds the sampled peak, limit, candidate and input
+revision. Its digest is included in the execution observation's evidence refs.
+A missing sample stays unknown. Use the same input, limits and measurement
+method when comparing candidates. Sampled PSS has different accounting from a
+container memory limit, and a sample below the reference cannot qualify
+headroom. Processes in other groups and peaks between samples are outside
+coverage. The option leaves the command's execution and time limit in place.
+
+Observed excess provides an optional next action for the current Yardstick
+decision. Keep the best authoritative evidence, reuse the matching incumbent
+measurement and spend the next bounded comparison on a different lower-footprint
+mechanism over the full workload. Check its contribution to the accepted outcome
+before promotion. The current Todo, execution authority and evaluator stay with
+their owners. 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 +209,18 @@ Stop、原预算和真实阻塞继续有效，不为填满时长制造任务。
 变化拒绝归因；文件清单哈希不是原生归档哈希，只覆盖明确选择的文件及探针
 边界，不覆盖传递依赖或独立写入者，结果仍须证明对应实际评测快照和输入版本。
 
+当并行内存占用会影响候选选择时，用同一入口执行完整代表性工作集并传入
+`--memory-limit-bytes <参考上限>`。Linux 采样命令进程组内各进程的 PSS，
+组成员变化或读不到完整数据时跳过该采样。相邻 `.memory.json` 绑定采样峰值、
+参考上限、候选和输入版本，其摘要进入原执行观测引用。组外进程和采样间峰值
+未覆盖。PSS 与容器内存限额的统计口径有差异，低于上限的采样不能确认余量。
+此选项沿用命令原执行及超时行为。
+
+实际观测超限时，现有 Yardstick 决策可采纳证据中给出的下一动作。保留最好的
+权威结果，复用同口径 incumbent 测量，把下一次有界试验用在较低占用的不同
+机制上，并在同一完整工作集验证对目标的贡献。Todo、授权和评测仍由原 owner
+管理。只有超时回执时，内存保持未知。
+
 当前授权工作实际产生或取得新的决策证据后，才进入相应工作段的检查点。旧反馈读回、
 记账和没有新增信息的确定性重复验证只是上下文，应先用于推进原任务。使用当前
 settlement plan 的 `durable_writeback.semantic_closeout_command_template`，
diff --git a/benchmark/runtime/research_probe.py b/benchmark/runtime/research_probe.py
index 9b4f87025c..9aa27133ec 100644
--- a/benchmark/runtime/research_probe.py
+++ b/benchmark/runtime/research_probe.py
@@ -6,7 +6,9 @@
 """
 from __future__ import annotations
 
+from _thread import LockType
 import argparse
+from contextlib import nullcontext
 import hashlib
 import json
 import math
@@ -14,6 +16,8 @@
 from pathlib import Path
 import signal
 import subprocess
+import sys
+import threading
 import time
 
 
@@ -45,6 +49,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 a candidate's aggregate memory could change the choice, reuse or measure the complete "
+        "workload with the applicable limit using --memory-limit-bytes. Use observed excess to compare "
+        "a lower-footprint method against the scored incumbent before scaling the same design. "
         "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 +76,91 @@ 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
+        # Do not silently sum a partial group when a worker exited or forked.
+        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_pss_bytes"] is None or peak > samples["peak_pss_bytes"]:
+                    samples.update(peak_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:
+    peak = samples["peak_pss_bytes"]
+    exceeded = None if peak is None else peak > limit_bytes
+    observation = {
+        "measurement_basis": "linux-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,
+        "memory_limit_bytes": limit_bytes,
+        "observed_limit_exceeded": exceeded,
+        **samples,
+    }
+    if exceeded:
+        observation["next_action"] = {
+            "method_id": "method:lower-footprint-challenge",
+            "reason": "Keep the best authoritative result. Compare a lower-footprint mechanism on the same complete "
+                      "workload before scaling this candidate, and measure its contribution to the accepted outcome.",
+            "cost_bound": "Reuse applicable measurements and bound one contender check by the existing workload and budget.",
+        }
+    return observation
+
+
 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",
@@ -88,7 +172,10 @@ def run_probe(*, command: list[str], candidate_files: list[Path], output: Path,
     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.
-    with output.open("x", encoding="utf-8") as record:
+    memory_output = output.with_suffix(output.suffix + ".memory.json")
+    with output.open("x", encoding="utf-8") as record, (
+        memory_output.open("xb") if memory_limit_bytes is not None else nullcontext()
+    ) as memory_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")
         env = dict(os.environ)
@@ -96,15 +183,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_pss_bytes=None, members_at_peak=None)
+                   if memory_record 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 +217,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 +232,18 @@ 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 memory_record is not None and samples is not None and memory_limit_bytes is not None:
+                try:
+                    measurement = _memory_observation(samples, result, memory_limit_bytes, evidence_ref)
+                    encoded = json.dumps(measurement, sort_keys=True, separators=(",", ":")).encode() + b"\n"
+                    memory_record.write(encoded)
+                    memory_record.flush()
+                    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 +297,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="Optionally compare sampled Linux process-group PSS to this positive diagnostic reference limit")
     parser.add_argument("--diagnostics", type=Path)
     parser.add_argument("command", nargs=argparse.REMAINDER)
     args = vars(parser.parse_args())
diff --git a/tests/control_plane/test_research_execution_loop.py b/tests/control_plane/test_research_execution_loop.py
index cee7f893ca..8ed371c1bd 100644
--- a/tests/control_plane/test_research_execution_loop.py
+++ b/tests/control_plane/test_research_execution_loop.py
@@ -141,6 +141,159 @@ 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_pss_bytes"] is None
+    assert evidence["observed_limit_exceeded"] is None 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 (tmp_path / "observation.json.memory.json").read_bytes() == b""
+
+
+@pytest.mark.skipif(sys.platform != "linux" or not Path("/proc/self/smaps_rollup").exists(),
+                    reason="requires live Linux process-group PSS")
+def test_full_workload_memory_evidence_changes_next_ordinary_work(tmp_path):
+    from tests.control_plane.test_work_yardstick_cli import attachment, binding, context, refresh, run, setup
+
+    project, runtime, registry = setup(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)
+    time.sleep(0.75)
+    print(total)
+else:
+    def shard(marker):
+        return subprocess.Popen(
+            [sys.executable, __file__, str(marker)],
+            stdout=subprocess.PIPE, text=True,
+        )
+    def read(process):
+        output, _ = process.communicate(timeout=3)
+        assert process.returncode == 0
+        return int(output)
+    first = shard(1)
+    if {parallel}:
+        second = shard(2)
+        values = [read(first), read(second)]
+    else:
+        values = [read(first), read(shard(2))]
+    Path("last-workload.json").write_text(json.dumps(values))
+    print(json.dumps(values))
+"""
+
+    def execute(name, *, streaming=False, parallel, method_id):
+        candidate = project / f"{name}.py"
+        candidate.write_text(template.format(shard_size=shard_size, streaming=streaming, parallel=parallel))
+        output = project / 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=project, env={**os.environ, "PYTHONPATH": str(fixture.REPO_ROOT)},
+                                 text=True, capture_output=True, timeout=10)
+        assert process.returncode == 0, process.stderr
+        assert json.loads(output.with_suffix(".json.stdout").read_text()) == [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
+        return observation, memory, ref
+
+    baseline, incumbent_memory, incumbent_ref = execute("incumbent", parallel=False, method_id="method:sequential")
+    challenger, expanded_memory, expanded_ref = execute("expanded", parallel=True, method_id="method:expanded")
+    assert incumbent_memory["observed_limit_exceeded"] is False
+    assert expanded_memory["observed_limit_exceeded"] is True and expanded_memory["members_at_peak"] == 3
+    assert "next_action" not in incumbent_memory
+
+    def result(observation, memory, evidence_ref, prior_finding_id, next_action, counterevidence):
+        packet = attachment()
+        yardstick = packet["work_yardstick_decision"]
+        yardstick["authoritative"]["input_revision"] = revision
+        yardstick["proxy"].update(metric="sampled-pipeline-memory", unit="bytes", higher_is_better=False,
+                                  input_revision=revision, evaluation_ref="evaluator:linux-process-group-pss-v1")
+        yardstick.update(prior_finding_id=prior_finding_id, next_action=next_action, counterevidence_refs=counterevidence)
+        packet.update(input_revision=revision, evidence_refs=observation["evidence_refs"],
+                      execution_observation=observation,
+                      measurements=[{**yardstick["proxy"], "value": memory["peak_pss_bytes"],
+                          "candidate_ref": "sha256:" + observation["candidate_sha256"], "evidence_ref": evidence_ref}])
+        return packet
+
+    # Use the real commit/next ordinary work surface, without manufacturing a
+    # replan obligation or running another evaluation in the control plane.
+    exceeded_attachment = project / "expanded.json"
+    exceeded_attachment.write_text(json.dumps(result(challenger, expanded_memory, expanded_ref, None,
+                                                     expanded_memory["next_action"], [expanded_ref])))
+    first = refresh(project, runtime, registry, "memory-study-1", exceeded_attachment)
+    assert first["explore_result_delivery"]["ok"]
+    run(project, runtime, registry, "quota", "spend-slot", *binding("memory-study-1"),
+        "--slots", "1", "--source", "heartbeat", "--execute")
+    guard = run(project, runtime, registry, "quota", "should-run", "--codex-app", *binding("memory-study-2"))
+    cue = context(runtime, registry)["graph"]["work_yardsticks"][0]
+    assert cue["source_finding_id"] in json.dumps(guard)
+    assert cue["counterevidence_refs"] == [expanded_ref]
+    assert cue["next_action"] == expanded_memory["next_action"]
+    chosen = cue["next_action"]["method_id"]
+    assert chosen == "method:lower-footprint-challenge"
+    contender, streamed_memory, streamed_ref = execute("streamed", streaming=True, parallel=True, method_id=chosen)
+    assert streamed_memory["observed_limit_exceeded"] is False
+    assert streamed_memory["peak_pss_bytes"] < incumbent_memory["peak_pss_bytes"] < expanded_memory["peak_pss_bytes"]
+    assert len({baseline["candidate_sha256"], challenger["candidate_sha256"], contender["candidate_sha256"]}) == 3
+    assert incumbent_ref in baseline["evidence_refs"]
+
+    adopted_attachment = project / "streamed.json"
+    adopted_attachment.write_text(json.dumps(result(contender, streamed_memory, streamed_ref, cue["source_finding_id"],
+        {"method_id": "method:confirm-outcome", "reason": "Check accepted outcome with the new mechanism",
+         "cost_bound": "Use the existing evaluation allowance"}, [])))
+    second = run(project, runtime, registry, "refresh-state", *binding("memory-study-2"), "--classification", "validated_change",
+        "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress",
+        "--explore-result-json", str(adopted_attachment), "--no-global-sync", "--suppress-external-sinks")
+    assert second["explore_result_delivery"]["ok"]
+    followthrough = context(runtime, registry)["graph"]["work_yardsticks"][0]["previous_decision_followthrough"]
+    assert followthrough["executed_method_id"] == chosen and followthrough["recorded_planned_method_matches"]
+    assert followthrough["candidate_sha256"] == contender["candidate_sha256"]
+    print(json.dumps({name: evidence["peak_pss_bytes"] for name, evidence in (
+        ("incumbent", incumbent_memory), ("expanded", expanded_memory), ("streamed", streamed_memory))}))
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])

From 3216cfe7ce629d89cea571efb098ec0d1103339f Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 14:20:48 +0800
Subject: [PATCH 2/2] fix(benchmark): preserve command results and scope
 sampled PSS

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/runtime/EXECUTION_OBSERVATIONS.md   |  51 ++++---
 benchmark/runtime/research_probe.py           |  61 ++++----
 .../test_research_execution_loop.py           | 134 ++++++++++++++----
 3 files changed, 156 insertions(+), 90 deletions(-)

diff --git a/benchmark/runtime/EXECUTION_OBSERVATIONS.md b/benchmark/runtime/EXECUTION_OBSERVATIONS.md
index 90d06b8592..f4858774b7 100644
--- a/benchmark/runtime/EXECUTION_OBSERVATIONS.md
+++ b/benchmark/runtime/EXECUTION_OBSERVATIONS.md
@@ -129,22 +129,21 @@ 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
-samples proportional set size across the command's process group, including
-workers and the process holding their results. An unstable group is skipped.
-The adjacent `.memory.json` binds the sampled peak, limit, candidate and input
-revision. Its digest is included in the execution observation's evidence refs.
-A missing sample stays unknown. Use the same input, limits and measurement
-method when comparing candidates. Sampled PSS has different accounting from a
-container memory limit, and a sample below the reference cannot qualify
-headroom. Processes in other groups and peaks between samples are outside
-coverage. The option leaves the command's execution and time limit in place.
-
-Observed excess provides an optional next action for the current Yardstick
-decision. Keep the best authoritative evidence, reuse the matching incumbent
-measurement and spend the next bounded comparison on a different lower-footprint
-mechanism over the full workload. Check its contribution to the accepted outcome
-before promotion. The current Todo, execution authority and evaluator stay with
-their owners. A native timeout alone supplies no memory observation.
+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
@@ -210,16 +209,16 @@ Stop、原预算和真实阻塞继续有效，不为填满时长制造任务。
 边界，不覆盖传递依赖或独立写入者，结果仍须证明对应实际评测快照和输入版本。
 
 当并行内存占用会影响候选选择时，用同一入口执行完整代表性工作集并传入
-`--memory-limit-bytes <参考上限>`。Linux 采样命令进程组内各进程的 PSS，
-组成员变化或读不到完整数据时跳过该采样。相邻 `.memory.json` 绑定采样峰值、
-参考上限、候选和输入版本，其摘要进入原执行观测引用。组外进程和采样间峰值
-未覆盖。PSS 与容器内存限额的统计口径有差异，低于上限的采样不能确认余量。
-此选项沿用命令原执行及超时行为。
-
-实际观测超限时，现有 Yardstick 决策可采纳证据中给出的下一动作。保留最好的
-权威结果，复用同口径 incumbent 测量，把下一次有界试验用在较低占用的不同
-机制上，并在同一完整工作集验证对目标的贡献。Todo、授权和评测仍由原 owner
-管理。只有超时回执时，内存保持未知。
+`--memory-limit-bytes <参考上限>`。Linux 逐 PID 读取 PSS，组成员变化或数据
+不完整时跳过该次采样。组成员稳定时，内存仍可能在两次读取之间转移，累加值
+可能高于真实同时占用。相邻 `.memory.json` 将估计和参考值绑定到候选及输入
+版本。写入成功才把其摘要记入原观测，写入失败仍保留原进程结果。组外进程和
+采样间峰值未覆盖，PSS 与容器内存限额的统计口径有差异。
+
+这项估计会影响选择时，独立核对真实同时占用，再通过原 Yardstick 写入有证据
+的下一比较。保留最好的权威结果，复用同口径 incumbent 测量；确有并行保留
+影响时，把下一次有界试验用于同一完整工作集上的较低占用机制。估计本身不产生
+超限判定或自动选择。只有超时回执时，内存保持未知。
 
 当前授权工作实际产生或取得新的决策证据后，才进入相应工作段的检查点。旧反馈读回、
 记账和没有新增信息的确定性重复验证只是上下文，应先用于推进原任务。使用当前
diff --git a/benchmark/runtime/research_probe.py b/benchmark/runtime/research_probe.py
index 9aa27133ec..316be73ef0 100644
--- a/benchmark/runtime/research_probe.py
+++ b/benchmark/runtime/research_probe.py
@@ -8,7 +8,6 @@
 
 from _thread import LockType
 import argparse
-from contextlib import nullcontext
 import hashlib
 import json
 import math
@@ -49,9 +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 a candidate's aggregate memory could change the choice, reuse or measure the complete "
-        "workload with the applicable limit using --memory-limit-bytes. Use observed excess to compare "
-        "a lower-footprint method against the scored incumbent before scaling the same design. "
+        "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 "
@@ -104,7 +103,8 @@ def _group_pss_sample(group_id: int) -> tuple[int, int] | None:
             if len(pss) != 3 or pss[2] != b"kB" or not pss[1].isdigit():
                 return None
             footprint += int(pss[1]) * 1024
-        # Do not silently sum a partial group when a worker exited or forked.
+        # 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):
@@ -123,32 +123,21 @@ def _observe_group_pss(group_id: int, stopped: threading.Event, samples: dict, l
             else:
                 peak, members = sample
                 samples["complete_samples"] += 1
-                if samples["peak_pss_bytes"] is None or peak > samples["peak_pss_bytes"]:
-                    samples.update(peak_pss_bytes=peak, members_at_peak=members)
+                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:
-    peak = samples["peak_pss_bytes"]
-    exceeded = None if peak is None else peak > limit_bytes
-    observation = {
-        "measurement_basis": "linux-process-group-pss-v1",
+    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,
-        "memory_limit_bytes": limit_bytes,
-        "observed_limit_exceeded": exceeded,
+        "probe_evidence_ref": evidence_ref.strip(),
+        "reference_memory_bytes": limit_bytes,
         **samples,
     }
-    if exceeded:
-        observation["next_action"] = {
-            "method_id": "method:lower-footprint-challenge",
-            "reason": "Keep the best authoritative result. Compare a lower-footprint mechanism on the same complete "
-                      "workload before scaling this candidate, and measure its contribution to the accepted outcome.",
-            "cost_bound": "Reuse applicable measurements and bound one contender check by the existing workload and budget.",
-        }
-    return observation
 
 
 def run_probe(*, command: list[str], candidate_files: list[Path], output: Path,
@@ -171,11 +160,12 @@ 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")
-    with output.open("x", encoding="utf-8") as record, (
-        memory_output.open("xb") if memory_limit_bytes is not None else nullcontext()
-    ) as memory_record:
+    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")
         env = dict(os.environ)
@@ -183,8 +173,8 @@ 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_pss_bytes=None, members_at_peak=None)
-                   if memory_record is not None else None)
+        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,
@@ -232,13 +222,14 @@ 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 memory_record is not None and samples is not None and memory_limit_bytes is not None:
+            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:
-                    measurement = _memory_observation(samples, result, memory_limit_bytes, evidence_ref)
-                    encoded = json.dumps(measurement, sort_keys=True, separators=(",", ":")).encode() + b"\n"
-                    memory_record.write(encoded)
-                    memory_record.flush()
-                    os.fsync(memory_record.fileno())
+                    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:
@@ -298,7 +289,7 @@ def main() -> int:
     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="Optionally compare sampled Linux process-group PSS to this positive diagnostic reference limit")
+        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/tests/control_plane/test_research_execution_loop.py b/tests/control_plane/test_research_execution_loop.py
index 8ed371c1bd..e40e85809d 100644
--- a/tests/control_plane/test_research_execution_loop.py
+++ b/tests/control_plane/test_research_execution_loop.py
@@ -154,8 +154,8 @@ def test_memory_observation_without_samples_leaves_timeout_and_next_method_unkno
     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_pss_bytes"] is None
-    assert evidence["observed_limit_exceeded"] is None and "next_action" not in evidence
+    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"]
 
 
@@ -164,7 +164,52 @@ def test_changed_candidate_cannot_publish_memory_evidence(tmp_path, monkeypatch)
         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 (tmp_path / "observation.json.memory.json").read_bytes() == b""
+    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(),
@@ -196,26 +241,43 @@ def test_full_workload_memory_evidence_changes_next_ordinary_work(tmp_path):
     else:
         scratch = bytearray([marker]) * {shard_size}
         total = sum(scratch)
-    time.sleep(0.75)
-    print(total)
+    # 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)],
-            stdout=subprocess.PIPE, text=True,
+            stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True,
         )
     def read(process):
-        output, _ = process.communicate(timeout=3)
-        assert process.returncode == 0
-        return int(output)
+        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)
-        values = [read(first), read(second)]
+        b, retained_b = read(second)
+        simultaneous_payload = retained_a + retained_b
+        time.sleep(0.60)
+        finish(first)
+        finish(second)
     else:
-        values = [read(first), read(shard(2))]
-    Path("last-workload.json").write_text(json.dumps(values))
-    print(json.dumps(values))
+        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):
@@ -230,7 +292,8 @@ def execute(name, *, streaming=False, parallel, method_id):
         process = subprocess.run(command, cwd=project, env={**os.environ, "PYTHONPATH": str(fixture.REPO_ROOT)},
                                  text=True, capture_output=True, timeout=10)
         assert process.returncode == 0, process.stderr
-        assert json.loads(output.with_suffix(".json.stdout").read_text()) == [shard_size, 2 * shard_size]
+        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)
@@ -238,32 +301,43 @@ def execute(name, *, streaming=False, parallel, method_id):
         assert ref in observation["evidence_refs"]
         assert memory["candidate_sha256"] == observation["candidate_sha256"]
         assert memory["input_revision"] == revision and memory["complete_samples"] > 0
-        return observation, memory, ref
-
-    baseline, incumbent_memory, incumbent_ref = execute("incumbent", parallel=False, method_id="method:sequential")
-    challenger, expanded_memory, expanded_ref = execute("expanded", parallel=True, method_id="method:expanded")
-    assert incumbent_memory["observed_limit_exceeded"] is False
-    assert expanded_memory["observed_limit_exceeded"] is True and expanded_memory["members_at_peak"] == 3
-    assert "next_action" not in incumbent_memory
+        assert memory["reference_memory_bytes"] == limit_bytes
+        assert "observed_limit_exceeded" not in memory and "next_action" not in memory
+        return observation, memory, ref, execution
+
+    baseline, incumbent_memory, incumbent_ref, incumbent_run = execute(
+        "incumbent", parallel=False, method_id="method:sequential")
+    challenger, expanded_memory, expanded_ref, expanded_run = execute(
+        "expanded", parallel=True, method_id="method:expanded")
+    assert incumbent_run["simultaneous_payload"] == shard_size < limit_bytes
+    assert expanded_run["simultaneous_payload"] == 2 * shard_size > limit_bytes
+    assert expanded_memory["members_at_peak"] == 3
 
     def result(observation, memory, evidence_ref, prior_finding_id, next_action, counterevidence):
         packet = attachment()
         yardstick = packet["work_yardstick_decision"]
         yardstick["authoritative"]["input_revision"] = revision
         yardstick["proxy"].update(metric="sampled-pipeline-memory", unit="bytes", higher_is_better=False,
-                                  input_revision=revision, evaluation_ref="evaluator:linux-process-group-pss-v1")
+                                  input_revision=revision, evaluation_ref="evaluator:linux-sequential-process-group-pss-v1")
         yardstick.update(prior_finding_id=prior_finding_id, next_action=next_action, counterevidence_refs=counterevidence)
         packet.update(input_revision=revision, evidence_refs=observation["evidence_refs"],
                       execution_observation=observation,
-                      measurements=[{**yardstick["proxy"], "value": memory["peak_pss_bytes"],
+                      measurements=[{**yardstick["proxy"], "value": memory["peak_sampled_pss_bytes"],
                           "candidate_ref": "sha256:" + observation["candidate_sha256"], "evidence_ref": evidence_ref}])
         return packet
 
     # Use the real commit/next ordinary work surface, without manufacturing a
     # replan obligation or running another evaluation in the control plane.
+    # The actor's choice follows the independent overlap witness, not a
+    # threshold verdict from the non-atomic memory estimate.
+    next_action = {
+        "method_id": "method:lower-footprint-challenge",
+        "reason": "Both full shards are retained at the same barrier. Compare streaming on the same workload.",
+        "cost_bound": "Reuse the incumbent and run one complete contender comparison",
+    }
     exceeded_attachment = project / "expanded.json"
     exceeded_attachment.write_text(json.dumps(result(challenger, expanded_memory, expanded_ref, None,
-                                                     expanded_memory["next_action"], [expanded_ref])))
+                                                     next_action, [expanded_ref])))
     first = refresh(project, runtime, registry, "memory-study-1", exceeded_attachment)
     assert first["explore_result_delivery"]["ok"]
     run(project, runtime, registry, "quota", "spend-slot", *binding("memory-study-1"),
@@ -272,12 +346,14 @@ def result(observation, memory, evidence_ref, prior_finding_id, next_action, cou
     cue = context(runtime, registry)["graph"]["work_yardsticks"][0]
     assert cue["source_finding_id"] in json.dumps(guard)
     assert cue["counterevidence_refs"] == [expanded_ref]
-    assert cue["next_action"] == expanded_memory["next_action"]
+    assert cue["next_action"] == next_action
     chosen = cue["next_action"]["method_id"]
     assert chosen == "method:lower-footprint-challenge"
-    contender, streamed_memory, streamed_ref = execute("streamed", streaming=True, parallel=True, method_id=chosen)
-    assert streamed_memory["observed_limit_exceeded"] is False
-    assert streamed_memory["peak_pss_bytes"] < incumbent_memory["peak_pss_bytes"] < expanded_memory["peak_pss_bytes"]
+    contender, streamed_memory, streamed_ref, streamed_run = execute(
+        "streamed", streaming=True, parallel=True, method_id=chosen)
+    assert streamed_run["simultaneous_payload"] == 2 * 1024 * 1024 < limit_bytes
+    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
     assert incumbent_ref in baseline["evidence_refs"]
 
@@ -292,7 +368,7 @@ def result(observation, memory, evidence_ref, prior_finding_id, next_action, cou
     followthrough = context(runtime, registry)["graph"]["work_yardsticks"][0]["previous_decision_followthrough"]
     assert followthrough["executed_method_id"] == chosen and followthrough["recorded_planned_method_matches"]
     assert followthrough["candidate_sha256"] == contender["candidate_sha256"]
-    print(json.dumps({name: evidence["peak_pss_bytes"] for name, evidence in (
+    print(json.dumps({name: evidence["peak_sampled_pss_bytes"] for name, evidence in (
         ("incumbent", incumbent_memory), ("expanded", expanded_memory), ("streamed", streamed_memory))}))