EPM: expert paging CPU inference for MiniCPM-V-4.7-35B-A3B Q4_K_M

12eb3a5 verified 15 days ago

12.6 kB

#!/usr/bin/env python3
"""EPM strict-mode runner: enforce an 8 GiB / no-swap ceiling on one process.
Why this exists
---------------
The repository's headline deployment claim is "MiniCPM-V-4.7-35B-A3B decodes inside
8 GiB of RAM with swap disabled". Past runs only ever set ``RLIMIT_DATA`` (which
caps anonymous data/heap, ignores file-backed mappings, and does not touch swap)
and then *observed* ``VmSwap: 0``. That is a pager-sizing number, not a
guarantee.
cgroups would be the clean way to express this — ``memory.max=8G`` plus
``memory.swap.max=0`` — but a writable cgroup2 mount needs ``CAP_SYS_ADMIN``.
Where that is unavailable this harness reaches the same two guarantees:
no-swap
    Every anonymous mapping the process creates is ``mlock``ed by
    ``tools/epq_lock.so``. Locked pages can never be written to a swap device, so
    no-swap is enforced by the kernel for the whole run rather than inferred from
    a single reading afterwards. File-backed mappings are deliberately *not*
    locked: clean page cache is reclaimable, so a real 8 GiB machine evicts it
    rather than swapping, and a cgroup would reclaim it too. ``VmLck`` in the
    record shows how much was actually locked.
8 GiB
    ``RLIMIT_DATA`` on the child is a hard kernel cap on anonymous memory: past
    it, ``brk``/``mmap(MAP_ANONYMOUS)`` fail instead of quietly growing. A
    supervisor samples ``/proc/<pid>/status`` every few milliseconds and
    ``SIGKILL``s on a breach, so transient spikes cannot slip through. Enforcement
    and evidence come from kernel counters, not from the sampler:
    ``RssAnon``/``VmHWM``/``VmSwap``/``VmLck`` are exact, so the sampler only has
    to be fast, not lucky.
``--metric anon`` (default) caps anonymous RSS, which is the quantity that binds
on real hardware: 8 GiB of RAM must hold the arena, the KV cache, the compute
buffers and the runtime, and the kernel can always reclaim the rest.
``--metric total`` caps total RSS instead, which is strictly harder than a cgroup
because it forbids even reclaimable cache. Use it when you want the pessimistic
number; use ``anon`` for the number that matches a cgroup.
Usage
-----
    python3 tools/strict8g.py --label decode -- epq-chat -m model.gguf ...
    python3 tools/strict8g.py --limit-mb 8192 --metric total --record r.jsonl -- cmd
Exit status is the child's own, except 124 when the ceiling was breached or the
run timed out (matching ``timeout``), and 125 for harness errors.
"""

from __future__ import annotations

import argparse
import json
import os
import resource
import signal
import subprocess
import sys
import tempfile
import threading
import time
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
LOCK_SRC = REPO / "tools" / "epq_lock.so.c"
LOCK_SO = REPO / "tools" / "epq_lock.so"

MIB = 1024 * 1024


def build_lock_helper() -> Path:
    """Compile the anonymous-memory locker once; it is one small C file."""
    if not LOCK_SRC.exists():
        raise SystemExit(f"[strict8g] 缺少 {LOCK_SRC}")
    cc = os.environ.get("CC", "cc")
    if (not LOCK_SO.exists()) or LOCK_SO.stat().st_mtime < LOCK_SRC.stat().st_mtime:
        subprocess.run([cc, "-O2", "-fPIC", "-shared", "-o", str(LOCK_SO), str(LOCK_SRC)],
                       check=True)
    return LOCK_SO


_WANTED = ("RssAnon", "RssFile", "VmRSS", "VmHWM", "VmSwap", "VmLck", "VmSize", "VmData")


def read_status(pid: int) -> dict:
    """Read one process's exact memory counters, in bytes."""
    out = dict.fromkeys(_WANTED, 0)
    try:
        with open(f"/proc/{pid}/status", "r") as fh:
            for line in fh:
                name = line.split(":", 1)[0]
                if name in out:
                    out[name] = int(line.split()[1]) * 1024
    except (FileNotFoundError, ProcessLookupError, ValueError, IndexError):
        pass
    return out


class Sampler(threading.Thread):
    """Polls the child's memory counters and kills it on a ceiling breach."""

    def __init__(self, pid: int, limit: int, metric: str, interval: float):
        super().__init__(daemon=True)
        self.pid = pid
        self.limit = limit
        self.metric = metric
        self.interval = interval
        self.peak = dict.fromkeys(_WANTED, 0)
        self.samples = 0
        self.swap_samples = 0
        self.breach = 0
        self.stop = threading.Event()

    def run(self) -> None:
        while not self.stop.is_set():
            st = read_status(self.pid)
            self.samples += 1
            for k in _WANTED:
                self.peak[k] = max(self.peak[k], st[k])
            if st["VmSwap"] > 0:
                self.swap_samples += 1
            cur = st[self.metric]
            if cur > self.limit:
                self.breach = cur
                try:
                    os.kill(self.pid, signal.SIGKILL)
                except ProcessLookupError:
                    pass
                return
            time.sleep(self.interval)


def _mib(x: int) -> float:
    return round(x / MIB, 1)


def main() -> int:
    ap = argparse.ArgumentParser(description="Run one command under a strict 8 GiB / no-swap ceiling")
    ap.add_argument("--limit-mb", type=int, default=8192, help="hard ceiling in MiB")
    ap.add_argument("--data-mb", type=int, default=0,
                    help="RLIMIT_DATA in MiB; default is --limit-mb plus VA slack "
                         "(RLIMIT_DATA bounds anonymous *virtual* memory, which is "
                         "stricter than physical RAM, so it is used as a runaway guard)")
    ap.add_argument("--no-rlimit", action="store_true",
                    help="do not set RLIMIT_DATA at all")
    ap.add_argument("--metric", choices=["anon", "total"], default="anon",
                    help="cap anonymous RSS (default, cgroup-equivalent) or total RSS")
    ap.add_argument("--interval-ms", type=float, default=4.0, help="sampling interval")
    ap.add_argument("--label", default="strict")
    ap.add_argument("--record", default="", help="append a JSON record to this file")
    ap.add_argument("--timeout", type=float, default=0, help="wall-clock limit in seconds (0 = none)")
    ap.add_argument("--no-lock", action="store_true",
                    help="skip mlock (no-swap then rests on VmSwap sampling alone)")
    ap.add_argument("--quiet", action="store_true")
    ap.add_argument("cmd", nargs=argparse.REMAINDER)
    args = ap.parse_args()

    cmd = args.cmd[1:] if args.cmd and args.cmd[0] == "--" else args.cmd
    if not cmd:
        ap.error("missing command; put it after `--`")

    limit = args.limit_mb * MIB
    # RLIMIT_DATA caps anonymous *virtual* memory; llama.cpp reserves graph and
    # compute buffers that are mapped but never touched, so charging them against
    # a physical 8 GiB budget costs ~1 GiB of expert arena for no physical
    # benefit. Physical residency is bounded by the sampler instead, exactly as a
    # cgroup memory.max bounds it. RLIMIT_DATA stays as an independent kernel
    # guard against a runaway, sized with slack for untouched mappings.
    VA_SLACK_MB = 1536
    data_mb = args.data_mb or (args.limit_mb + VA_SLACK_MB)
    metric = "RssAnon" if args.metric == "anon" else "VmRSS"

    try:
        helper = build_lock_helper() if not args.no_lock else None
    except (subprocess.CalledProcessError, OSError) as exc:
        raise SystemExit(f"[strict8g] 無法編譯 mlock helper：{exc}")

    fd, status_path = tempfile.mkstemp(prefix="epq-mlock-", suffix=".txt")
    os.close(fd)

    try:
        env = dict(os.environ)
        if helper:
            env["LD_PRELOAD"] = str(helper) + (":" + env["LD_PRELOAD"] if env.get("LD_PRELOAD") else "")
            env["EPM_MLOCK_STATUS"] = status_path

        def child_setup() -> None:
            # mlock() is bounded by RLIMIT_MEMLOCK, so raise it before main runs.
            try:
                resource.setrlimit(resource.RLIMIT_MEMLOCK,
                                   (resource.RLIM_INFINITY, resource.RLIM_INFINITY))
            except (ValueError, OSError):
                pass
            # Kernel guard on anonymous virtual memory (see VA_SLACK_MB above).
            if not args.no_rlimit:
                cap = data_mb * MIB
                try:
                    resource.setrlimit(resource.RLIMIT_DATA, (cap, cap))
                except (ValueError, OSError):
                    pass
            # A multi-GiB core dump would fill the disk and mask the failure we
            # are trying to measure.
            try:
                resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
            except (ValueError, OSError):
                pass
            os.setpgrp()

        t0 = time.time()
        try:
            proc = subprocess.Popen(cmd, env=env, preexec_fn=child_setup,
                                    stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                    text=True)
        except OSError as exc:
            print(f"[strict8g] 無法執行 {cmd[0]}: {exc}", file=sys.stderr)
            return 125

        sampler = Sampler(proc.pid, limit, metric, args.interval_ms / 1000.0)
        sampler.start()

        timed_out = False
        try:
            stdout, stderr = proc.communicate(timeout=args.timeout or None)
        except subprocess.TimeoutExpired:
            timed_out = True
            proc.kill()
            stdout, stderr = proc.communicate()
        finally:
            sampler.stop.set()
            sampler.join(timeout=2.0)
        wall = time.time() - t0

        lock_note = ""
        if helper:
            try:
                lock_note = Path(status_path).read_text().strip().replace("\n", " ")
            except OSError:
                pass
    finally:
        try:
            os.unlink(status_path)
        except OSError:
            pass

    rc = proc.returncode
    overflow = sampler.breach > 0
    swapped = sampler.peak["VmSwap"] > 0
    locked = sampler.peak["VmLck"]
    # VmHWM is a kernel high-water mark: the exact peak even if it lasted well
    # under one sampling interval.
    ok = (rc == 0 and not overflow and not swapped and not timed_out
          and (args.no_lock or locked > 0))

    rec = {
        "label": args.label,
        "cmd": cmd,
        "returncode": rc,
        "wall_s": round(wall, 3),
        "limit_mb": args.limit_mb,
        "metric": args.metric,
        "rlimit_data_mb": data_mb,
        "interval_ms": args.interval_ms,
        "samples": sampler.samples,
        "peak_anon_mb": _mib(sampler.peak["RssAnon"]),
        "peak_file_mb": _mib(sampler.peak["RssFile"]),
        "peak_rss_mb": _mib(sampler.peak["VmRSS"]),
        "vm_hwm_mb": _mib(sampler.peak["VmHWM"]),
        "peak_swap_mb": _mib(sampler.peak["VmSwap"]),
        "swap_samples_over_zero": sampler.swap_samples,
        "vm_lck_mb": _mib(locked),
        "peak_vmsize_mb": _mib(sampler.peak["VmSize"]),
        "peak_vmdata_mb": _mib(sampler.peak["VmData"]),
        "overflow_killed": overflow,
        "breach_mb": _mib(sampler.breach) if overflow else None,
        "timed_out": timed_out,
        "lock": lock_note,
        "strict_ok": ok,
    }

    if args.record:
        p = Path(args.record)
        p.parent.mkdir(parents=True, exist_ok=True)
        with p.open("a", encoding="utf-8") as fh:
            fh.write(json.dumps(rec, ensure_ascii=False) + "\n")

    if not args.quiet:
        print(f"[strict8g] {'PASS' if ok else 'FAIL'} {args.label}: rc={rc} wall={wall:.1f}s "
              f"| anon={rec['peak_anon_mb']} file={rec['peak_file_mb']} "
              f"total={rec['peak_rss_mb']} hwm={rec['vm_hwm_mb']} MiB "
              f"vmdata={rec['peak_vmdata_mb']} "
              f"| cap {args.limit_mb} MiB ({args.metric}) swap={rec['peak_swap_mb']} "
              f"lck={rec['vm_lck_mb']} samples={sampler.samples}", file=sys.stderr)
        if overflow:
            print(f"[strict8g] 超標：{metric} 到 {rec['breach_mb']} MiB 時 SIGKILL", file=sys.stderr)
        if swapped:
            print(f"[strict8g] 偵測到 swap：{rec['peak_swap_mb']} MiB", file=sys.stderr)
        if locked <= 0 and not args.no_lock:
            print(f"[strict8g] 沒有任何匿名記憶體被鎖定，no-swap 無法成立"
                  f"（{lock_note or 'helper 未回報'}）", file=sys.stderr)
        elif not ok and not overflow and not swapped and not timed_out:
            print(f"[strict8g] 子行程自行失敗 rc={rc}", file=sys.stderr)

    sys.stdout.write(stdout or "")
    sys.stderr.write(stderr or "")
    if overflow or timed_out:
        return 124
    return rc


if __name__ == "__main__":
    sys.exit(main())

	#!/usr/bin/env python3
	"""EPM strict-mode runner: enforce an 8 GiB / no-swap ceiling on one process.

	Why this exists
	---------------
	The repository's headline deployment claim is "MiniCPM-V-4.7-35B-A3B decodes inside
	8 GiB of RAM with swap disabled". Past runs only ever set ``RLIMIT_DATA`` (which
	caps anonymous data/heap, ignores file-backed mappings, and does not touch swap)
	and then observed ``VmSwap: 0``. That is a pager-sizing number, not a
	guarantee.

	cgroups would be the clean way to express this — ``memory.max=8G`` plus
	``memory.swap.max=0`` — but a writable cgroup2 mount needs ``CAP_SYS_ADMIN``.
	Where that is unavailable this harness reaches the same two guarantees:

	no-swap
	Every anonymous mapping the process creates is ``mlock``ed by
	``tools/epq_lock.so``. Locked pages can never be written to a swap device, so
	no-swap is enforced by the kernel for the whole run rather than inferred from
	a single reading afterwards. File-backed mappings are deliberately not
	locked: clean page cache is reclaimable, so a real 8 GiB machine evicts it
	rather than swapping, and a cgroup would reclaim it too. ``VmLck`` in the
	record shows how much was actually locked.

	8 GiB
	``RLIMIT_DATA`` on the child is a hard kernel cap on anonymous memory: past
	it, ``brk``/``mmap(MAP_ANONYMOUS)`` fail instead of quietly growing. A
	supervisor samples ``/proc/<pid>/status`` every few milliseconds and
	``SIGKILL``s on a breach, so transient spikes cannot slip through. Enforcement
	and evidence come from kernel counters, not from the sampler:
	``RssAnon``/``VmHWM``/``VmSwap``/``VmLck`` are exact, so the sampler only has
	to be fast, not lucky.

	``--metric anon`` (default) caps anonymous RSS, which is the quantity that binds
	on real hardware: 8 GiB of RAM must hold the arena, the KV cache, the compute
	buffers and the runtime, and the kernel can always reclaim the rest.
	``--metric total`` caps total RSS instead, which is strictly harder than a cgroup
	because it forbids even reclaimable cache. Use it when you want the pessimistic
	number; use ``anon`` for the number that matches a cgroup.

	Usage
	-----
	python3 tools/strict8g.py --label decode -- epq-chat -m model.gguf ...
	python3 tools/strict8g.py --limit-mb 8192 --metric total --record r.jsonl -- cmd

	Exit status is the child's own, except 124 when the ceiling was breached or the
	run timed out (matching ``timeout``), and 125 for harness errors.
	"""

	from __future__ import annotations

	import argparse
	import json
	import os
	import resource
	import signal
	import subprocess
	import sys
	import tempfile
	import threading
	import time
	from pathlib import Path

	REPO = Path(__file__).resolve().parent.parent
	LOCK_SRC = REPO / "tools" / "epq_lock.so.c"
	LOCK_SO = REPO / "tools" / "epq_lock.so"

	MIB = 1024 * 1024


	def build_lock_helper() -> Path:
	"""Compile the anonymous-memory locker once; it is one small C file."""
	if not LOCK_SRC.exists():
	raise SystemExit(f"[strict8g] 缺少 {LOCK_SRC}")
	cc = os.environ.get("CC", "cc")
	if (not LOCK_SO.exists()) or LOCK_SO.stat().st_mtime < LOCK_SRC.stat().st_mtime:
	subprocess.run([cc, "-O2", "-fPIC", "-shared", "-o", str(LOCK_SO), str(LOCK_SRC)],
	check=True)
	return LOCK_SO


	_WANTED = ("RssAnon", "RssFile", "VmRSS", "VmHWM", "VmSwap", "VmLck", "VmSize", "VmData")


	def read_status(pid: int) -> dict:
	"""Read one process's exact memory counters, in bytes."""
	out = dict.fromkeys(_WANTED, 0)
	try:
	with open(f"/proc/{pid}/status", "r") as fh:
	for line in fh:
	name = line.split(":", 1)[0]
	if name in out:
	out[name] = int(line.split()[1]) * 1024
	except (FileNotFoundError, ProcessLookupError, ValueError, IndexError):
	pass
	return out


	class Sampler(threading.Thread):
	"""Polls the child's memory counters and kills it on a ceiling breach."""

	def __init__(self, pid: int, limit: int, metric: str, interval: float):
	super().__init__(daemon=True)
	self.pid = pid
	self.limit = limit
	self.metric = metric
	self.interval = interval
	self.peak = dict.fromkeys(_WANTED, 0)
	self.samples = 0
	self.swap_samples = 0
	self.breach = 0
	self.stop = threading.Event()

	def run(self) -> None:
	while not self.stop.is_set():
	st = read_status(self.pid)
	self.samples += 1
	for k in _WANTED:
	self.peak[k] = max(self.peak[k], st[k])
	if st["VmSwap"] > 0:
	self.swap_samples += 1
	cur = st[self.metric]
	if cur > self.limit:
	self.breach = cur
	try:
	os.kill(self.pid, signal.SIGKILL)
	except ProcessLookupError:
	pass
	return
	time.sleep(self.interval)


	def _mib(x: int) -> float:
	return round(x / MIB, 1)


	def main() -> int:
	ap = argparse.ArgumentParser(description="Run one command under a strict 8 GiB / no-swap ceiling")
	ap.add_argument("--limit-mb", type=int, default=8192, help="hard ceiling in MiB")
	ap.add_argument("--data-mb", type=int, default=0,
	help="RLIMIT_DATA in MiB; default is --limit-mb plus VA slack "
	"(RLIMIT_DATA bounds anonymous virtual memory, which is "
	"stricter than physical RAM, so it is used as a runaway guard)")
	ap.add_argument("--no-rlimit", action="store_true",
	help="do not set RLIMIT_DATA at all")
	ap.add_argument("--metric", choices=["anon", "total"], default="anon",
	help="cap anonymous RSS (default, cgroup-equivalent) or total RSS")
	ap.add_argument("--interval-ms", type=float, default=4.0, help="sampling interval")
	ap.add_argument("--label", default="strict")
	ap.add_argument("--record", default="", help="append a JSON record to this file")
	ap.add_argument("--timeout", type=float, default=0, help="wall-clock limit in seconds (0 = none)")
	ap.add_argument("--no-lock", action="store_true",
	help="skip mlock (no-swap then rests on VmSwap sampling alone)")
	ap.add_argument("--quiet", action="store_true")
	ap.add_argument("cmd", nargs=argparse.REMAINDER)
	args = ap.parse_args()

	cmd = args.cmd[1:] if args.cmd and args.cmd[0] == "--" else args.cmd
	if not cmd:
	ap.error("missing command; put it after `--`")

	limit = args.limit_mb * MIB
	# RLIMIT_DATA caps anonymous virtual memory; llama.cpp reserves graph and
	# compute buffers that are mapped but never touched, so charging them against
	# a physical 8 GiB budget costs ~1 GiB of expert arena for no physical
	# benefit. Physical residency is bounded by the sampler instead, exactly as a
	# cgroup memory.max bounds it. RLIMIT_DATA stays as an independent kernel
	# guard against a runaway, sized with slack for untouched mappings.
	VA_SLACK_MB = 1536
	data_mb = args.data_mb or (args.limit_mb + VA_SLACK_MB)
	metric = "RssAnon" if args.metric == "anon" else "VmRSS"

	try:
	helper = build_lock_helper() if not args.no_lock else None
	except (subprocess.CalledProcessError, OSError) as exc:
	raise SystemExit(f"[strict8g] 無法編譯 mlock helper：{exc}")

	fd, status_path = tempfile.mkstemp(prefix="epq-mlock-", suffix=".txt")
	os.close(fd)

	try:
	env = dict(os.environ)
	if helper:
	env["LD_PRELOAD"] = str(helper) + (":" + env["LD_PRELOAD"] if env.get("LD_PRELOAD") else "")
	env["EPM_MLOCK_STATUS"] = status_path

	def child_setup() -> None:
	# mlock() is bounded by RLIMIT_MEMLOCK, so raise it before main runs.
	try:
	resource.setrlimit(resource.RLIMIT_MEMLOCK,
	(resource.RLIM_INFINITY, resource.RLIM_INFINITY))
	except (ValueError, OSError):
	pass
	# Kernel guard on anonymous virtual memory (see VA_SLACK_MB above).
	if not args.no_rlimit:
	cap = data_mb * MIB
	try:
	resource.setrlimit(resource.RLIMIT_DATA, (cap, cap))
	except (ValueError, OSError):
	pass
	# A multi-GiB core dump would fill the disk and mask the failure we
	# are trying to measure.
	try:
	resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
	except (ValueError, OSError):
	pass
	os.setpgrp()

	t0 = time.time()
	try:
	proc = subprocess.Popen(cmd, env=env, preexec_fn=child_setup,
	stdout=subprocess.PIPE, stderr=subprocess.PIPE,
	text=True)
	except OSError as exc:
	print(f"[strict8g] 無法執行 {cmd[0]}: {exc}", file=sys.stderr)
	return 125

	sampler = Sampler(proc.pid, limit, metric, args.interval_ms / 1000.0)
	sampler.start()

	timed_out = False
	try:
	stdout, stderr = proc.communicate(timeout=args.timeout or None)
	except subprocess.TimeoutExpired:
	timed_out = True
	proc.kill()
	stdout, stderr = proc.communicate()
	finally:
	sampler.stop.set()
	sampler.join(timeout=2.0)
	wall = time.time() - t0

	lock_note = ""
	if helper:
	try:
	lock_note = Path(status_path).read_text().strip().replace("\n", " ")
	except OSError:
	pass
	finally:
	try:
	os.unlink(status_path)
	except OSError:
	pass

	rc = proc.returncode
	overflow = sampler.breach > 0
	swapped = sampler.peak["VmSwap"] > 0
	locked = sampler.peak["VmLck"]
	# VmHWM is a kernel high-water mark: the exact peak even if it lasted well
	# under one sampling interval.
	ok = (rc == 0 and not overflow and not swapped and not timed_out
	and (args.no_lock or locked > 0))

	rec = {
	"label": args.label,
	"cmd": cmd,
	"returncode": rc,
	"wall_s": round(wall, 3),
	"limit_mb": args.limit_mb,
	"metric": args.metric,
	"rlimit_data_mb": data_mb,
	"interval_ms": args.interval_ms,
	"samples": sampler.samples,
	"peak_anon_mb": _mib(sampler.peak["RssAnon"]),
	"peak_file_mb": _mib(sampler.peak["RssFile"]),
	"peak_rss_mb": _mib(sampler.peak["VmRSS"]),
	"vm_hwm_mb": _mib(sampler.peak["VmHWM"]),
	"peak_swap_mb": _mib(sampler.peak["VmSwap"]),
	"swap_samples_over_zero": sampler.swap_samples,
	"vm_lck_mb": _mib(locked),
	"peak_vmsize_mb": _mib(sampler.peak["VmSize"]),
	"peak_vmdata_mb": _mib(sampler.peak["VmData"]),
	"overflow_killed": overflow,
	"breach_mb": _mib(sampler.breach) if overflow else None,
	"timed_out": timed_out,
	"lock": lock_note,
	"strict_ok": ok,
	}

	if args.record:
	p = Path(args.record)
	p.parent.mkdir(parents=True, exist_ok=True)
	with p.open("a", encoding="utf-8") as fh:
	fh.write(json.dumps(rec, ensure_ascii=False) + "\n")

	if not args.quiet:
	print(f"[strict8g] {'PASS' if ok else 'FAIL'} {args.label}: rc={rc} wall={wall:.1f}s "
	f"\| anon={rec['peak_anon_mb']} file={rec['peak_file_mb']} "
	f"total={rec['peak_rss_mb']} hwm={rec['vm_hwm_mb']} MiB "
	f"vmdata={rec['peak_vmdata_mb']} "
	f"\| cap {args.limit_mb} MiB ({args.metric}) swap={rec['peak_swap_mb']} "
	f"lck={rec['vm_lck_mb']} samples={sampler.samples}", file=sys.stderr)
	if overflow:
	print(f"[strict8g] 超標：{metric} 到 {rec['breach_mb']} MiB 時 SIGKILL", file=sys.stderr)
	if swapped:
	print(f"[strict8g] 偵測到 swap：{rec['peak_swap_mb']} MiB", file=sys.stderr)
	if locked <= 0 and not args.no_lock:
	print(f"[strict8g] 沒有任何匿名記憶體被鎖定，no-swap 無法成立"
	f"（{lock_note or 'helper 未回報'}）", file=sys.stderr)
	elif not ok and not overflow and not swapped and not timed_out:
	print(f"[strict8g] 子行程自行失敗 rc={rc}", file=sys.stderr)

	sys.stdout.write(stdout or "")
	sys.stderr.write(stderr or "")
	if overflow or timed_out:
	return 124
	return rc


	if __name__ == "__main__":
	sys.exit(main())