Observe aggregate memory in local work probes

benchmark/runtime/EXECUTION_OBSERVATIONS.md

@@ -126,6 +126,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 +208,18 @@ Stop、原预算和真实阻塞继续有效，不为填满时长制造任务。
 变化拒绝归因；文件清单哈希不是原生归档哈希，只覆盖明确选择的文件及探针
 边界，不覆盖传递依赖或独立写入者，结果仍须证明对应实际评测快照和输入版本。
 
+当并行内存占用会影响候选选择时，用同一入口执行完整代表性工作集并传入
+`--memory-limit-bytes <参考上限>`。Linux 逐 PID 读取 PSS，组成员变化或数据
+不完整时跳过该次采样。组成员稳定时，内存仍可能在两次读取之间转移，累加值
+可能高于真实同时占用。相邻 `.memory.json` 将估计和参考值绑定到候选及输入
+版本。写入成功才把其摘要记入原观测，写入失败仍保留原进程结果。组外进程和
+采样间峰值未覆盖，PSS 与容器内存限额的统计口径有差异。
+
+这项估计会影响选择时，独立核对真实同时占用，再通过原 Yardstick 写入有证据
+的下一比较。保留最好的权威结果，复用同口径 incumbent 测量；确有并行保留
+影响时，把下一次有界试验用于同一完整工作集上的较低占用机制。估计本身不产生
+超限判定或自动选择。只有超时回执时，内存保持未知。
+
 当前授权工作实际产生或取得新的决策证据后，才进入相应工作段的检查点。旧反馈读回、
 记账和没有新增信息的确定性重复验证只是上下文，应先用于推进原任务。使用当前
 settlement plan 的 `durable_writeback.semantic_closeout_command_template`，

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,22 +173,42 @@ 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:
                     os.killpg(process.pid, signal.SIGKILL)
                 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())