From ca2d15b3ee965bd7ab27204cdf4b96b117d148c1 Mon Sep 17 00:00:00 2001 From: Ysopwnable <161784452+N0zoM1z0@users.noreply.github.com> Date: Tue, 13 Oct 2026 09:44:41 +0800 Subject: [PATCH] test(perf): observe owned counters at payload completion Add opt-in, owned payload-completion cgroup witnesses without changing default receipts, terminal-counter semantics, resource limits, or REA public APIs. Preserve unknown and nonzero outcomes; verify ownership, observer capacity, cleanup, and actual SDK/model workflows. Validation: eleven Python semantic tests, supported full-project typecheck and lint, controlled Linux failure/cancellation/capacity probes, real managed/WASM/Android SDK and Codex runs, and all 39 exact-head CI checks. Curated evidence retains mixed timings and whole-scope uncertainty. Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- experiments/analysis-efficiency/README.md | 4 + .../analysis-efficiency/resource-counters.md | 103 ++++++++++++++++ .../analysis-efficiency/resource-run.py | 62 +++++++++- .../analysis-efficiency/resource_payload.py | 114 ++++++++++++++++++ .../test_resource_payload.py | 103 ++++++++++++++++ package.json | 2 +- 6 files changed, 384 insertions(+), 4 deletions(-) create mode 100644 experiments/analysis-efficiency/resource-counters.md create mode 100644 experiments/analysis-efficiency/resource_payload.py create mode 100644 experiments/analysis-efficiency/test_resource_payload.py diff --git a/experiments/analysis-efficiency/README.md b/experiments/analysis-efficiency/README.md index fb645b1e0..79bac97e0 100644 --- a/experiments/analysis-efficiency/README.md +++ b/experiments/analysis-efficiency/README.md @@ -37,6 +37,10 @@ raw MCP replies from observed client execution/discovery and usage. Optional trial capture retains tool/usage items while excluding account and ambient state; existing ephemeral trials keep their default behavior. +Follow-up: [resource counters at payload completion](resource-counters.md) adds +optional owned counter observations while preserving unknown whole-scope final +counters, default receipt shape and the unchanged CPU/memory safeguards. + The 12 detailed checkpoint receipts were archived byte-for-byte before removal from the final PR tree. The archive `review-receipts-00d51e0fc-20261012` is in the private experiment lab outside the checkout; its `archive-index.json` records diff --git a/experiments/analysis-efficiency/resource-counters.md b/experiments/analysis-efficiency/resource-counters.md new file mode 100644 index 000000000..8b21f54df --- /dev/null +++ b/experiments/analysis-efficiency/resource-counters.md @@ -0,0 +1,103 @@ +# Resource counters at payload completion + +Continues [#2016](https://github.com/morluto/rea/issues/2016) from merged main +`4f4f6cd7b5d7a400b90772a138d7b9f942c2a556`. A transient cgroup can disappear +before the existing supervisor reads terminal counters. Earlier reports keep +those counters unknown; sampled zero events are not final no-OOM proof. + +An optional `resource-run.py --observe-payload-exit -- COMMAND...` adds one small +observer inside the same owned scope. It waits for the direct payload, records +kernel `memory.peak`, CPU and memory/task events, and reports remaining processes +before acknowledging completion to the outer supervisor. The observer remains +alive until receipt acknowledgement, bounded by the existing ten-second cleanup +interval. Payload stdio and process exit status remain observable. + +The receipt phase is **payload-exited-observer-alive**. The supervisor validates +the owned-unit token, numeric fields, cgroup identity, live observer membership, +payload exit status and absence of additional processes before marking that +observation verified. Wrong-owner, malformed, missing or inconsistent receipts +stay unknown; lingering actors stay explicit and cannot certify cleanup. +Unexpected metadata is omitted rather than copying an inherited environment. + +This is not a final whole-scope counter: the observer still has to exit. Existing +`resource_events_final` and `resource_counters_verified` retain their original +meaning and unknown state. The grader's existing full resource gate is unchanged. +Earlier runs and source/runtime hashes are not relabeled. Observer CPU/memory and +acknowledgement time are included in this new measurement mode; comparable runs +must use the same mode. Default commands do not start an observer or add fields. + +One CPU, the unchanged 3 GiB early-stop/4 GiB hard ceiling, no swap, task limits +and host-available-memory safeguards remain in force. No resource limit is raised. +The opt-in mode requires at least two task slots for observer plus payload; +smaller caller-selected ceilings fail before process acquisition. Default mode +continues to accept a one-task ceiling. +Cleanup uses the uniquely owned systemd scope, without signaling a process merely +because a recorded PID exists. Failed scope cleanup retains the named owner. + +## Focused verification + +Five semantic Python cases cover acquisition-free capacity rejection, nonzero exit/kernel facts, missing or malformed +receipts, wrong ownership, integer types, contradictory cleanup, lingering actors, +foreign symlinks and exclusion of unrelated metadata. They join the existing +Python gate; simulated receipts do not establish actual cgroup behavior. + +Actual Linux cgroup/systemd probes additionally verify: + +- Default mode keeps its receipt shape and success status. +- Success and exit-code 3 preserve the payload outcome and verify a sole observer. +- A bounded 16 MiB Node allocation yields a kernel peak, including work that the + 100 ms sampler can miss. +- A payload leaving a child reports both actors and refuses verification; the + supervisor retires the owned scope. +- A caller-selected task ceiling of 4 triggers the existing task-limit early stop + with exit 125 and an unverified observation. +- Interrupting a confirmed-live owned controller preserves failure and retires + its scope; missing completion counters are not replaced with zeros. + +These are metering/lifecycle probes, not a new model/provider performance claim. +Actual REA workflows and repeated observer-overhead measurements remain necessary +before relying on the new mode for domain comparisons. Raw receipts remain in +the private experiment lab outside Git. + +## Actual REA and model observations + +Checkpoint `d1f2cfe76df28e6effe973123f2e19313f6d0d4b` repeats real SDK workflows +against the unchanged frozen candidate `8bc1601ca51e417ca472d90ea0367e9418a1a0f2` +(runtime file-table SHA-256 +`9e7380f537301bf0cfe2926f5619011af848af7c220750e75189498e65d825e1`). +The resource supervisor changes; the compiled REA/provider algorithms do not. +Managed pairs use default/observer/observer/default order with the same question, +settings, two public calls and identical selected facts. Raw receipts remain +private and new measurements do not replace historical resource unknowns. + +| Workflow | SDK milliseconds | Scope seconds | Sampled peak bytes | Kernel peak at payload exit | Payload CPU microseconds | +| -------------------- | ---------------: | ------------: | -----------------: | --------------------------: | -----------------------: | +| Managed default 1 | 5,168.485 | 11.814 | 525,873,152 | unknown | unknown | +| Managed observer 1 | 4,156.708 | 5.358 | 527,777,792 | 535,724,032 | 5,195,457 | +| Managed observer 2 | 3,810.362 | 5.056 | 538,742,784 | 539,529,216 | 4,818,497 | +| Managed default 2 | 4,263.993 | 5.360 | 526,032,896 | unknown | unknown | +| SQLite WASM observer | 6,733.216 | 8.087 | 777,199,616 | 777,199,616 | 7,830,130 | +| NewPipe APK observer | 23,162.195 | 24.600 | 1,286,586,368 | 1,287,548,928 | 23,965,627 | + +The WASM and APK runs use actual configured WABT and JADX/Java. Every observer +receipt verifies a sole remaining observer, successful payload exit and the +expected owned group. No guard abort occurs. The initial default scope timing is +an outlier relative to its SDK workflow and later scopes; its cause is not +established. Two managed pairs are insufficient to characterize observer overhead +or establish a speed/memory improvement. Kernel peak and CPU include the observer. + +One actual Codex CLI 0.161.0 trial uses the original managed question, exact-name +guidance, context capture and requested `gpt-5.6-terra` high. It passes selected +facts, preserves source bytes, removes its profile and makes three successful MCP +calls (including an additional artifact inspection). Reported usage is 61,173 +input, 48,128 cached, 13,045 uncached and 2,678 output tokens; wall time is 125.798 +seconds. The payload-exit witness records 812,896,256 kernel peak bytes and +8,566,029 CPU microseconds with only the observer alive and zero recorded +memory/task limit events at that phase. This confirms the completed model/MCP +payload boundary, not a token/time benefit or final whole-scope no-OOM proof. +Its full existing grader remains false for unknown terminal physical counters. + +The subsequent review adds supervisor/helper source SHA-256 to opt-in receipts +before process acquisition, without adding those fields to default receipts. +This distinguishes measurement implementations independently of frozen runtime +identity and keeps the observer's phase explicit. diff --git a/experiments/analysis-efficiency/resource-run.py b/experiments/analysis-efficiency/resource-run.py index 4417b390a..1790cdcca 100644 --- a/experiments/analysis-efficiency/resource-run.py +++ b/experiments/analysis-efficiency/resource-run.py @@ -1,14 +1,18 @@ """Run one owned experiment in a bounded, transient Linux systemd scope.""" import argparse +import hashlib import json import os from pathlib import Path import subprocess import sys +import tempfile import time import uuid +from resource_payload import read_observation + def available_bytes(): for line in Path('/proc/meminfo').read_text().splitlines(): @@ -64,10 +68,14 @@ def stop_owned_scope(unit, process, group): def main(): parser = argparse.ArgumentParser() parser.add_argument('--task-ceiling', type=int, default=192) + parser.add_argument('--observe-payload-exit', action='store_true', + help='Observe payload-exit counters before the cgroup observer exits') parser.add_argument('command', nargs=argparse.REMAINDER) options = parser.parse_args() if options.task_ceiling <= 0: parser.error('--task-ceiling must be positive') + if options.observe_payload_exit and options.task_ceiling < 2: + parser.error('--observe-payload-exit requires capacity for the observer and payload (at least 2 tasks)') args = options.command if args and args[0] == '--': args = args[1:] @@ -78,22 +86,41 @@ def main(): raise SystemExit('Refusing experiment: less than 8 GiB available') unit = 'rea-efficiency-' + uuid.uuid4().hex + '.scope' cpu = max(os.sched_getaffinity(0)) + receipt_root = tempfile.TemporaryDirectory(prefix='rea-payload-counters-') if options.observe_payload_exit else None + receipt_path = Path(receipt_root.name) / 'observation.json' if receipt_root else None + release_path = Path(receipt_root.name) / 'release' if receipt_root else None + payload = args if receipt_root is None else [ + sys.executable, str(Path(__file__).with_name('resource_payload.py')), + str(receipt_path), str(release_path), unit, *args, + ] command = [ 'systemd-run', '--user', '--scope', '--unit=' + unit, '-p', 'MemoryMax=4G', '-p', 'MemoryHigh=3G', '-p', 'MemorySwapMax=0', '-p', 'CPUQuota=100%', '-p', 'TasksMax=' + str(options.task_ceiling), - 'taskset', '-c', str(cpu), *args, + 'taskset', '-c', str(cpu), *payload, ] - process = subprocess.Popen(command) + observer_sources = { + name: hashlib.sha256(Path(__file__).with_name(name).read_bytes()).hexdigest() + for name in ['resource-run.py', 'resource_payload.py'] + } if options.observe_payload_exit else None + try: + process = subprocess.Popen(command) + except BaseException: + if receipt_root is not None: + receipt_root.cleanup() + raise peak = 0 peak_tasks = 0 events = {} cpu_stats = {} abort = None final_events = None + payload_observation = None + payload_group = None + code = None + group = None started = time.monotonic() try: - group = None while process.poll() is None: if group is None: reply = subprocess.run( @@ -104,6 +131,18 @@ def main(): path = reply.stdout.strip() if path: group = Path('/sys/fs/cgroup') / path.lstrip('/') + if receipt_path is not None and payload_observation is None and group is not None: + candidate = read_observation(receipt_path, unit) + if candidate is not None: + payload_observation = candidate + if candidate['control_group'] == str(group.relative_to('/sys/fs/cgroup')): + try: + actors = [int(x) for x in (group / 'cgroup.procs').read_text().splitlines()] + if candidate['observer_pid'] in actors: + payload_group = group + except (OSError, ValueError): + pass + release_path.touch(exist_ok=False) current = group / 'memory.current' if group else None if current and current.exists(): try: @@ -132,6 +171,8 @@ def main(): # Read while the owned group still exists, before wait/cleanup can remove it. final_events = final_resource_events(group) code = process.wait() + if receipt_path is not None and payload_observation is None: + payload_observation = read_observation(receipt_path, unit) except BaseException: stop_owned_scope(unit, process, group) raise @@ -150,7 +191,22 @@ def main(): 'resource_events_final': final_events, 'resource_counters_verified': final_events is not None, 'cpu_stat_last_observed': cpu_stats, + **({ + 'payload_resource_observation': payload_observation, + 'payload_resource_observation_verified': ( + payload_observation is not None + and payload_observation['payload_cleanup_verified'] + and payload_group is not None and abort is None + and code is not None + and (payload_observation['payload_exit_code'] if payload_observation['payload_exit_code'] >= 0 + else 128 - payload_observation['payload_exit_code']) == code + ), + 'payload_observation_scope': 'Counters at payload exit while observer remains alive; not final whole-scope counters.', + 'payload_observer_sources_sha256': observer_sources, + } if options.observe_payload_exit else {}), }), file=sys.stderr) + if receipt_root is not None: + receipt_root.cleanup() raise SystemExit(125 if abort else code) diff --git a/experiments/analysis-efficiency/resource_payload.py b/experiments/analysis-efficiency/resource_payload.py new file mode 100644 index 000000000..700598cbf --- /dev/null +++ b/experiments/analysis-efficiency/resource_payload.py @@ -0,0 +1,114 @@ +"""Observe a finished payload while its owned cgroup observer is still alive.""" + +import json +import os +from pathlib import Path +import subprocess +import sys +import time + + +def pairs(path): + values = {} + for line in path.read_text().splitlines(): + name, value = line.split() + number = int(value) + if number < 0 or name in values: + raise ValueError('Invalid kernel counter') + values[name] = number + return values + + +def snapshot(group, token, code): + actors = [int(line) for line in (group / 'cgroup.procs').read_text().splitlines()] + observer = os.getpid() + return { + 'phase': 'payload-exited-observer-alive', 'ownership_token': token, + 'payload_exit_code': code, 'observer_pid': observer, + 'remaining_processes': actors, 'payload_cleanup_verified': actors == [observer], + 'resource_events': {name: pairs(group / name) + for name in ['memory.events', 'pids.events']}, + 'cpu_stat': pairs(group / 'cpu.stat'), + 'memory_peak_bytes': int((group / 'memory.peak').read_text()), + 'control_group': str(group.relative_to('/sys/fs/cgroup')), + } + + +def read_observation(path, token): + """Validate the observer receipt; missing/malformed witnesses stay unknown.""" + try: + if path.is_symlink(): + return None + value = json.loads(path.read_text()) + if not isinstance(value, dict) or value.get('ownership_token') != token: + return None + if value.get('phase') != 'payload-exited-observer-alive': + return None + if not isinstance(value.get('control_group'), str) or not value['control_group']: + return None + for name in ['payload_exit_code', 'observer_pid', 'memory_peak_bytes']: + if type(value.get(name)) is not int: + return None + if value['observer_pid'] <= 0 or value['memory_peak_bytes'] < 0: + return None + actors = value.get('remaining_processes') + if not isinstance(actors, list) or any(type(x) is not int or x <= 0 for x in actors): + return None + if type(value.get('payload_cleanup_verified')) is not bool: + return None + if value['payload_cleanup_verified'] != (actors == [value['observer_pid']]): + return None + required = {'memory.events': ['high', 'max', 'oom', 'oom_kill'], + 'pids.events': ['max']} + events = value.get('resource_events') + if not isinstance(events, dict): + return None + for name, keys in required.items(): + record = events.get(name) + if not isinstance(record, dict) or any(key not in record for key in keys): + return None + if any(type(x) is not int or x < 0 for x in record.values()): + return None + cpu = value.get('cpu_stat') + if not isinstance(cpu, dict) or 'usage_usec' not in cpu: + return None + if any(type(x) is not int or x < 0 for x in cpu.values()): + return None + names = ['phase', 'ownership_token', 'payload_exit_code', 'observer_pid', + 'remaining_processes', 'payload_cleanup_verified', + 'cpu_stat', 'memory_peak_bytes', 'control_group'] + return {**{name: value[name] for name in names}, + 'resource_events': {name: events[name] for name in required}} + except (OSError, ValueError, TypeError): + return None + + +def main(): + output, release, token = Path(sys.argv[1]), Path(sys.argv[2]), sys.argv[3] + command = sys.argv[4:] + if not command: + raise SystemExit('Missing explicit payload command') + child = subprocess.Popen(command) + code = child.wait() + paths = [line[3:] for line in Path('/proc/self/cgroup').read_text().splitlines() + if line.startswith('0::')] + try: + if len(paths) != 1: + raise ValueError('Expected one unified cgroup') + group = Path('/sys/fs/cgroup') / paths[0].lstrip('/') + observation = snapshot(group, token, code) + with os.fdopen(os.open(output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), + 'w') as file: + json.dump(observation, file) + # Keep the owner alive until its supervisor acknowledges the receipt. + # Bound a broken supervisor by the existing ten-second cleanup interval. + deadline = time.monotonic() + 10 + while not release.exists() and time.monotonic() < deadline: + time.sleep(0.02) + except (OSError, ValueError) as error: + print('Payload counter observation unavailable: ' + type(error).__name__, file=sys.stderr) + raise SystemExit(code if code >= 0 else 128 - code) + + +if __name__ == '__main__': + main() diff --git a/experiments/analysis-efficiency/test_resource_payload.py b/experiments/analysis-efficiency/test_resource_payload.py new file mode 100644 index 000000000..3f091efb9 --- /dev/null +++ b/experiments/analysis-efficiency/test_resource_payload.py @@ -0,0 +1,103 @@ +"""Receipt semantics; real systemd/cgroup workflows are verified separately.""" + +import json +import importlib.util +import io +from pathlib import Path +import sys +import tempfile +import unittest +from unittest.mock import patch + +from resource_payload import read_observation + + +def receipt(): + return { + 'phase': 'payload-exited-observer-alive', 'ownership_token': 'own-unit', + 'payload_exit_code': 3, 'observer_pid': 100, 'remaining_processes': [100], + 'payload_cleanup_verified': True, 'memory_peak_bytes': 123, + 'control_group': 'user.slice/own-unit.scope', + 'resource_events': {'memory.events': {'high': 0, 'max': 0, 'oom': 0, 'oom_kill': 0}, + 'pids.events': {'max': 0}}, + 'cpu_stat': {'usage_usec': 1000}, + } + + +class PayloadObservationTests(unittest.TestCase): + def test_observer_capacity_failure_precedes_any_process_acquisition(self): + spec = importlib.util.spec_from_file_location('resource_run', Path(__file__).with_name('resource-run.py')) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + arguments = ['resource-run.py', '--observe-payload-exit', '--task-ceiling', '1', '--', 'payload'] + with patch.object(sys, 'argv', arguments), patch.object(sys, 'stderr', io.StringIO()), \ + patch.object(module.subprocess, 'Popen') as spawn: + with self.assertRaises(SystemExit) as failure: + module.main() + self.assertEqual(failure.exception.code, 2) + spawn.assert_not_called() + + def test_preserves_nonzero_status_kernel_peak_and_nonzero_events(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / 'receipt.json' + value = receipt() + value['resource_events']['memory.events']['oom'] = 1 + path.write_text(json.dumps(value)) + observed = read_observation(path, 'own-unit') + self.assertEqual(observed['payload_exit_code'], 3) + self.assertEqual(observed['memory_peak_bytes'], 123) + self.assertEqual(observed['resource_events']['memory.events']['oom'], 1) + + def test_rejects_missing_malformed_wrong_owner_and_inconsistent_cleanup(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / 'receipt.json' + self.assertIsNone(read_observation(path, 'own-unit')) + path.write_text('incomplete-json') + self.assertIsNone(read_observation(path, 'own-unit')) + cases = [] + for key, invalid in [('ownership_token', 'other-unit'), ('observer_pid', True), + ('memory_peak_bytes', -1), ('remaining_processes', [100, 101]), + ('payload_cleanup_verified', 'true'), ('phase', 'scope-exited')]: + value = receipt() + value[key] = invalid + cases.append(value) + value = receipt() + del value['resource_events']['memory.events']['oom'] + cases.append(value) + value = receipt() + value['cpu_stat']['usage_usec'] = False + cases.append(value) + for value in cases: + with self.subTest(value=value): + path.write_text(json.dumps(value)) + self.assertIsNone(read_observation(path, 'own-unit')) + + def test_retains_lingering_actor_facts_without_certifying_cleanup(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / 'receipt.json' + value = receipt() + value['remaining_processes'].append(101) + value['payload_cleanup_verified'] = False + path.write_text(json.dumps(value)) + observed = read_observation(path, 'own-unit') + self.assertEqual(observed['remaining_processes'], [100, 101]) + self.assertFalse(observed['payload_cleanup_verified']) + + def test_excludes_unrelated_metadata_and_foreign_symlinks(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / 'receipt.json' + value = receipt() + value['environment'] = {'ambient': 'excluded'} + value['resource_events']['unrelated'] = {'ambient': 'excluded'} + path.write_text(json.dumps(value)) + observed = read_observation(path, 'own-unit') + self.assertNotIn('environment', observed) + self.assertNotIn('unrelated', observed['resource_events']) + link = Path(directory) / 'linked.json' + link.symlink_to(path) + self.assertIsNone(read_observation(link, 'own-unit')) + self.assertEqual(json.loads(path.read_text()), value) + + +if __name__ == '__main__': + unittest.main() diff --git a/package.json b/package.json index 327c14315..f803cc00c 100644 --- a/package.json +++ b/package.json @@ -130,7 +130,7 @@ "test": "npm run test:with-build", "test:with-build": "npm run test:prepare && vitest run", "test:run": "vitest run", - "test:python": "python3 -m unittest bridge/native/test_rea_lldb_tracer.py -v && python3 -m unittest discover -s experiments/analysis-efficiency -p test_codex_context.py -v", + "test:python": "python3 -m unittest bridge/native/test_rea_lldb_tracer.py -v && python3 -m unittest discover -s experiments/analysis-efficiency -p 'test_*.py' -v", "test:local": "node scripts/run-development-tests.mjs local", "test:fast": "vitest run --project domain --project services --project adapters --project composition --project conformance --project evaluation", "pretest:boundary": "npm run test:prepare",