diff --git a/benchmark/runtime/EXECUTION_OBSERVATIONS.md b/benchmark/runtime/EXECUTION_OBSERVATIONS.md index ccf8e9cfae..f4858774b7 100644 --- a/benchmark/runtime/EXECUTION_OBSERVATIONS.md +++ b/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 `. 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`, 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/tests/control_plane/test_research_execution_loop.py b/tests/control_plane/test_research_execution_loop.py index cee7f893ca..e40e85809d 100644 --- a/tests/control_plane/test_research_execution_loop.py +++ b/tests/control_plane/test_research_execution_loop.py @@ -141,6 +141,235 @@ 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_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) + # 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 = 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 + 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, 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-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_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, + 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"] == next_action + chosen = cue["next_action"]["method_id"] + assert chosen == "method:lower-footprint-challenge" + 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"] + + 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_sampled_pss_bytes"] for name, evidence in ( + ("incumbent", incumbent_memory), ("expanded", expanded_memory), ("streamed", streamed_memory))})) @pytest.mark.parametrize("provider", ["file", "sqlite"])