synaps 0.12.0

#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
#
# Memory-footprint benchmark for terminal agent harnesses.
#
# Measures total PSS + SwapPss (Proportional Set Size, resident and swapped)
# across the full process tree of a tool running N concurrent sessions. SwapPss
# is included because with zram/swap the kernel swaps an idle TUI's pages out
# within seconds, and resident PSS alone then undercounts by pressure. PSS (from /proc/<pid>/smaps_rollup) splits
# shared pages proportionally, so it is the correct metric for multi-process /
# daemon architectures where RSS would overcount shared memory.
#
# For synaps (a daemon architecture: `synaps --attach` shares one daemon across
# sessions), the daemon + its workers + every attached client are summed. Other
# tools launch N independent sessions; their trees are summed the same way.
#
# Methodology and portions of this harness are adapted from jcode's
# `bench_memory_cli.py` by Jeremy Huang (MIT). See benchmarks/NOTICE.
#
# Isolation: every synaps invocation runs against an isolated base dir
# (SYNAPS_BASE_DIR if set, else /tmp/synaps-bench-home); the harness refuses to
# run against ~/.synaps-cli, so it can never stop or attach to a live daemon.
# It also aborts if any pre-existing synaps daemon pid disappears mid-run.
#
# Requires: python3, pyte  (pip install pyte); Linux (/proc/*/smaps_rollup)
# Escaped processes: every launched process carries a BENCH_RUN_TAG env marker;
# tagged processes are summed even if they left the launched tree/pgroup
# (jcode's TUI starts `jcode serve` with setsid, which outlives its parent).
# jcode runs with an isolated JCODE_HOME/JCODE_RUNTIME_DIR per run.
#
# Timing: with --hold N, the reported sample is taken N seconds after the
# last launch, with every PTY drained meanwhile: some tools trim their heap
# once idle (jcode's server drops ~60 MB about two minutes after start), so a
# sample right after launch depends on how long the launches took. The
# at-launch sample is recorded too.
#
# Usage:    python3 bench_memory.py --sessions 1 10 --reps 3 --tools synaps jcode
#           SYNAPS_BIN=/path/to/synaps python3 bench_memory.py --sessions 10
#           python3 bench_memory.py --sessions 1000 --launch-wait 0.1   # scaling

from __future__ import annotations

import argparse
import json
import os
import pty
import re
import resource
import select
import shutil
import signal
import socket
import statistics
import subprocess
import sys
import tempfile
import time
import uuid
from dataclasses import dataclass, field
from pathlib import Path

PROBE = "zx91q"
DEFAULT_TIMEOUT_S = 20.0
DEFAULT_SETTLE_S = 3.0
ANSI_RE = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~]|\][^\x1b\x07]*(?:\x07|\x1b\\))")

CAP_REPLIES = [
    (b"\x1b[6n", b"\x1b[1;1R"), (b"\x1b[c", b"\x1b[?62;c"),
    (b"\x1b[>c", b"\x1b[>0;0;0c"), (b"\x1b[?u", b"\x1b[?0u"),
    (b"\x1b]10;?\x1b\\", b"\x1b]10;rgb:ffff/ffff/ffff\x1b\\"),
    (b"\x1b]11;?\x1b\\", b"\x1b]11;rgb:0000/0000/0000\x1b\\"),
    (b"\x1b[?2026$p", b"\x1b[?2026;1$y"), (b"\x1b[?2027$p", b"\x1b[?2027;1$y"),
    (b"\x1b[?2004$p", b"\x1b[?2004;1$y"),
]


DEFAULT_BENCH_BASE = "/tmp/synaps-bench-home"


def bench_base_dir() -> Path:
    """Isolated Synaps base dir. Refuses anything that is, contains or lives in
    the real ~/.synaps-cli: the benchmark stops and restarts its daemon."""
    raw = os.environ.get("SYNAPS_BASE_DIR") or DEFAULT_BENCH_BASE
    base = Path(raw).expanduser().resolve()
    live = (Path.home() / ".synaps-cli").resolve()
    if base == live or live in base.parents or base in live.parents:
        raise SystemExit(f"refusing to benchmark in {base}: it overlaps the live base dir {live}. "
                         f"Unset SYNAPS_BASE_DIR or point it at a scratch dir (default {DEFAULT_BENCH_BASE}).")
    return base


def synaps_env(base: Path) -> dict[str, str]:
    """Environment for every synaps process: inherited SYNAPS_* (runtime dir,
    profile, session ids) is dropped so nothing can route to the live daemon."""
    env = {k: v for k, v in os.environ.items() if not k.startswith("SYNAPS_")}
    env["SYNAPS_BASE_DIR"] = str(base)
    return env


def synaps_daemon_pids() -> dict[int, str]:
    """Every running synaps daemon process: {pid: cmdline}."""
    out: dict[int, str] = {}
    for e in Path("/proc").iterdir():
        if not e.name.isdigit():
            continue
        try:
            argv = (e / "cmdline").read_bytes().split(b"\0")
        except OSError:
            continue
        if len(argv) > 1 and os.path.basename(argv[0]).startswith(b"synaps") and b"daemon" in argv[1:3]:
            out[int(e.name)] = b" ".join(argv).decode(errors="replace").strip()
    return out


def is_default_daemon(pid: int, argv: str) -> bool:
    """The user's own daemon: default base dir and no profile. Daemons of
    other profiles/base dirs (other sessions' demos) may come and go."""
    if "--profile" in argv.split():
        return False
    try:
        env = dict(kv.split(b"=", 1) for kv in Path(f"/proc/{pid}/environ").read_bytes().split(b"\0") if b"=" in kv)
    except OSError:
        return True  # unreadable: assume it matters
    if env.get(b"SYNAPS_PROFILE"):
        return False
    base = env.get(b"SYNAPS_BASE_DIR")
    return base is None or Path(base.decode()).expanduser().resolve() == (Path.home() / ".synaps-cli").resolve()


def check_live_daemons(before: dict[int, str]) -> None:
    """Abort if the user's default daemon that existed before the run is gone."""
    now = synaps_daemon_pids()
    gone = {p: c for p, c in before.items() if p not in now}
    vital = {p: c for p, c in gone.items() if p in DEFAULT_DAEMONS}
    if vital:
        print(f"ABORT: the live default synaps daemon disappeared during the run: {vital}", file=sys.stderr)
        raise SystemExit(2)
    for p, c in gone.items():
        if p not in NOTED_GONE:
            NOTED_GONE.add(p)
            print(f"note: another session's daemon exited (not ours, not the default): {p} {c}", flush=True)


DEFAULT_DAEMONS: set[int] = set()
NOTED_GONE: set[int] = set()


@dataclass
class ToolSpec:
    name: str
    argv: list[str]
    version_argv: list[str]
    env: dict[str, str] = field(default_factory=dict)
    synaps_daemon: bool = False
    jcode_home: bool = False  # isolated JCODE_HOME / JCODE_RUNTIME_DIR per run
    xdg_home: bool = False    # isolated XDG_{CONFIG,DATA,STATE,CACHE}_HOME per run


def build_specs() -> dict[str, ToolSpec]:
    synaps = os.environ.get("SYNAPS_BIN", "synaps")
    jcode = ["jcode", "--no-update", "--no-selfdev"]
    return {
        "synaps": ToolSpec("synaps", [synaps], [synaps, "--version"], synaps_daemon=True),
        # jcode's TUI spawns `jcode serve` in its own session (setsid); later
        # TUIs attach to it. Default config has features.memory = true.
        "jcode": ToolSpec("jcode", jcode, ["jcode", "version"],
                          {"JCODE_NO_TELEMETRY": "1"}, jcode_home=True),
        "jcode_memory_off": ToolSpec("jcode_memory_off", jcode, ["jcode", "version"],
                                     {"JCODE_NO_TELEMETRY": "1", "JCODE_MEMORY_ENABLED": "0"},
                                     jcode_home=True),
        "codex": ToolSpec("codex", ["codex"], ["codex", "--version"]),
        # opencode 2.x TUIs attach to a background `opencode serve --service`
        # found via $XDG_STATE_HOME/opencode/service.json (auto-started if
        # absent): isolated XDG dirs give it its own service, never the user's.
        "opencode": ToolSpec("opencode", ["opencode"], ["opencode", "--version"], xdg_home=True),
        "pi": ToolSpec("pi", ["pi"], ["pi", "--version"]),
        "claude_code": ToolSpec("claude_code", ["claude"], ["claude", "--version"]),
        "cursor": ToolSpec("cursor", ["cursor-agent"], ["cursor-agent", "--version"]),
    }


def tool_version(spec: ToolSpec) -> str:
    env = synaps_env(bench_base_dir()) if spec.synaps_daemon else os.environ.copy()
    try:
        p = subprocess.run(spec.version_argv, capture_output=True, text=True, timeout=30,
                           stdin=subprocess.DEVNULL, env=env)
    except (OSError, subprocess.TimeoutExpired) as e:
        return f"unavailable ({e.__class__.__name__})"
    lines = [l.strip() for l in (p.stdout + p.stderr).splitlines() if l.strip()]
    return " ".join(lines[0].split()) if lines else f"exit {p.returncode}"


def machine_info() -> dict:
    mem_kb = next(int(l.split()[1]) for l in Path("/proc/meminfo").read_text().splitlines()
                  if l.startswith("MemTotal:"))
    osr = dict(l.split("=", 1) for l in Path("/etc/os-release").read_text().splitlines() if "=" in l)
    model = next((l.split(":", 1)[1].strip() for l in Path("/proc/cpuinfo").read_text().splitlines()
                  if l.startswith("model name")), "?")
    return {"cpu": model, "cores": os.cpu_count(), "mem_gb": round(mem_kb / 1024 / 1024, 1),
            "os": osr.get("PRETTY_NAME", "?").strip('"'), "kernel": os.uname().release,
            "arch": os.uname().machine}


def answer_caps(fd: int, buf: bytes) -> bytes:
    changed = True
    while changed:
        changed = False
        for q, rep in CAP_REPLIES:
            if q in buf:
                os.write(fd, rep)
                buf = buf.replace(q, b"")
                changed = True
    return buf


@dataclass
class Launch:
    pid: int
    pgid: int
    fd: int
    ready: bool = False
    excerpt: str | None = None


def launch(argv: list[str], env: dict[str, str], timeout_s: float, settle_s: float) -> Launch:
    master, slave = pty.openpty()
    proc = subprocess.Popen(argv, stdin=slave, stdout=slave, stderr=slave,
                            env=env, preexec_fn=os.setsid)
    os.close(slave)
    os.set_blocking(master, False)
    start = time.perf_counter()
    buf = b""
    ready = probe_sent = False
    excerpt = None
    while time.perf_counter() - start < timeout_s:
        r, _, _ = select.select([master], [], [], 0.05)
        if r:
            try:
                chunk = os.read(master, 65536)
            except BlockingIOError:
                chunk = b""
            except OSError:  # EIO: the client exited and its side of the PTY is closed
                break
            if chunk:
                buf = answer_caps(master, buf + chunk)
                plain = ANSI_RE.sub("", buf.decode("utf-8", "replace"))
                lines = [" ".join(l.split()) for l in plain.splitlines()]
                hit = next((l for l in lines if sum(c.isalnum() for c in l) >= 3), None)
                if not ready and hit:
                    ready, excerpt = True, hit[:100]
                    if not probe_sent:
                        try:
                            os.write(master, PROBE.encode()); probe_sent = True
                        except OSError:
                            break
                if probe_sent and PROBE in plain:
                    break
        if proc.poll() is not None:
            break
    if ready:
        time.sleep(settle_s)
    # preexec_fn=os.setsid makes pgid == pid. os.getpgid() would raise
    # ProcessLookupError for a client that exited (and was reaped) during launch.
    return Launch(proc.pid, proc.pid, master, ready, excerpt)


def proc_map() -> dict[int, tuple[int, int]]:
    out: dict[int, tuple[int, int]] = {}
    for e in Path("/proc").iterdir():
        if not e.name.isdigit():
            continue
        try:
            stat = (e / "stat").read_text()
            rest = stat[stat.rfind(")") + 2:].split()
            out[int(e.name)] = (int(rest[1]), int(rest[2]))  # ppid, pgid
        except Exception:
            continue
    return out


def descendants(roots: list[int]) -> set[int]:
    m = proc_map()
    kids: dict[int, list[int]] = {}
    for pid, (ppid, _) in m.items():
        kids.setdefault(ppid, []).append(pid)
    seen: set[int] = set()
    stack = list(roots)
    while stack:
        p = stack.pop()
        if p in seen:
            continue
        seen.add(p)
        stack.extend(kids.get(p, []))
    return seen


def pgroup_pids(pgids: list[int]) -> set[int]:
    want = set(pgids)
    return {pid for pid, (_, pgid) in proc_map().items() if pgid in want}


def pss_mb(pid: int) -> dict[str, float] | None:
    """Pss, SwapPss, Pss_Anon, Pss_File in MB. SwapPss matters: with zram/swap,
    an idle TUI's anon pages are swapped out within seconds and vanish from Pss
    alone. Pss_File drops when the kernel evicts file pages (binary text)."""
    try:
        vals = {}
        for line in Path(f"/proc/{pid}/smaps_rollup").read_text().splitlines():
            k, _, rest = line.partition(":")
            if k in ("Pss", "SwapPss", "Pss_Anon", "Pss_File"):
                vals[k] = int(rest.split()[0]) / 1024.0
        return vals if "Pss" in vals else None
    except Exception:
        return None


def stale_bench_pids() -> set[int]:
    """Processes left by an earlier, crashed run of this harness (any tag)."""
    out: set[int] = set()
    for e in Path("/proc").iterdir():
        if not e.name.isdigit() or int(e.name) == os.getpid():
            continue
        try:
            if any(v.startswith(b"BENCH_RUN_TAG=") for v in (e / "environ").read_bytes().split(b"\0")):
                out.add(int(e.name))
        except OSError:
            continue
    return out


def mem_pressure() -> dict:
    """Memory PSI avg10 (% of time stalled on memory) and MemAvailable."""
    out = {}
    try:
        for line in Path("/proc/pressure/memory").read_text().splitlines():
            kind, *kv = line.split()
            out[f"psi_{kind}_avg10"] = float(dict(x.split("=") for x in kv)["avg10"])
    except OSError:
        pass
    avail = next(int(l.split()[1]) for l in Path("/proc/meminfo").read_text().splitlines()
                 if l.startswith("MemAvailable:"))
    out["mem_available_gb"] = round(avail / 1024 / 1024, 1)
    return out


def wait_quiet(max_psi: float, max_wait_s: float) -> dict:
    """Under memory pressure the kernel reclaims a TUI's pages while it is being
    sampled (file pages evicted, anon pages swapped), so wait for a quiet box."""
    end = time.time() + max_wait_s
    while True:
        mp = mem_pressure()
        if mp.get("psi_some_avg10", 0.0) <= max_psi or time.time() > end:
            return mp
        time.sleep(5)


def tagged_pids(tag: str) -> set[int]:
    """Processes carrying this run's BENCH_RUN_TAG in their environment. This
    catches anything a tool detached from its tree/pgroup (setsid, double
    fork, reparenting to init), e.g. a background server spawned by the TUI."""
    needle = f"BENCH_RUN_TAG={tag}".encode()
    out: set[int] = set()
    for e in Path("/proc").iterdir():
        if not e.name.isdigit():
            continue
        try:
            if needle in (e / "environ").read_bytes().split(b"\0"):
                out.add(int(e.name))
        except OSError:
            continue
    return out


def cmdline(pid: int) -> str:
    try:
        return Path(f"/proc/{pid}/cmdline").read_bytes().replace(b"\0", b" ").decode(errors="replace").strip()
    except OSError:
        return "?"


def sum_tree_pss(roots: list[int], pgids: list[int], tag: str) -> dict:
    """Total PSS over: descendants of the roots, their process groups, and
    every process tagged with this run's environment marker."""
    tree = descendants(roots) | pgroup_pids(pgids)
    tagged = tagged_pids(tag)
    procs = []
    for pid in sorted(tree | tagged):
        v = pss_mb(pid)
        if v is not None:
            swap = v.get("SwapPss", 0.0)
            procs.append({"pid": pid, "pss_mb": round(v["Pss"], 1), "swap_pss_mb": round(swap, 1),
                          "mem_mb": round(v["Pss"] + swap, 1),
                          "pss_anon_mb": round(v.get("Pss_Anon", 0.0), 1),
                          "pss_file_mb": round(v.get("Pss_File", 0.0), 1),
                          "in_tree": pid in tree, "cmd": cmdline(pid)[:120]})
    total = lambda key, pred=lambda p: True: round(sum(p[key] for p in procs if pred(p)), 1)
    return {"mem_mb": total("mem_mb"),            # Pss + SwapPss: the reported number
            "pss_mb": total("pss_mb"),            # resident only
            "swap_pss_mb": total("swap_pss_mb"),
            "pss_anon_mb": total("pss_anon_mb"),
            "pss_file_mb": total("pss_file_mb"),
            "process_count": len(procs),
            "outside_tree_mem_mb": total("mem_mb", lambda p: not p["in_tree"]),
            "processes": procs}


def kill_pgroup(pgid: int) -> None:
    for sig in (signal.SIGTERM, signal.SIGKILL):
        try:
            os.killpg(pgid, sig)
            time.sleep(0.2)
        except ProcessLookupError:
            return


def wait_socket(path: str, timeout_s: float) -> bool:
    end = time.time() + timeout_s
    while time.time() < end:
        if Path(path).exists():
            return True
        time.sleep(0.05)
    return False


def reap_tagged(tag: str) -> list[int]:
    """Kill whatever this run left behind; returns survivors (should be [])."""
    for sig in (signal.SIGTERM, signal.SIGKILL):
        left = tagged_pids(tag)
        if not left:
            return []
        for pid in left:
            try:
                os.kill(pid, sig)
            except ProcessLookupError:
                pass
        time.sleep(0.5)
    return sorted(tagged_pids(tag))


def fresh_bench_base() -> Path:
    """Empty isolated base dir: no plugins, no sessions, no pins carried over
    from a previous run (pinned sessions are rehydrated on daemon start)."""
    base = bench_base_dir()
    shutil.rmtree(base, ignore_errors=True)
    base.mkdir(parents=True)
    return base


def stop_bench_daemon(bin_: str, base: Path) -> None:
    env = synaps_env(base)
    assert Path(env["SYNAPS_BASE_DIR"]).resolve() == bench_base_dir() == base
    subprocess.run([bin_, "daemon", "stop", "--force"], env=env,
                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)


def drain(launches: list[Launch], seconds: float) -> None:
    """Keep every PTY read (and answer capability queries) for `seconds`: a TUI
    whose terminal is never read blocks on write and stops its own idle work
    (e.g. jcode's idle heap trim), which would freeze it in its launch state."""
    end = time.time() + seconds
    fds = {l.fd for l in launches}
    while fds and time.time() < end:
        r, _, _ = select.select(list(fds), [], [], 0.2)
        for fd in r:
            try:
                answer_caps(fd, os.read(fd, 65536))
            except OSError:
                fds.discard(fd)


def exited(pid: int) -> bool:
    """True once a launched process is gone or a zombie."""
    try:
        stat = Path(f"/proc/{pid}/stat").read_text()
    except OSError:
        return True
    return stat[stat.rfind(")") + 2:].startswith("Z")


def run_tool(spec: ToolSpec, sessions: int, timeout_s: float, settle_s: float, hold_s: float = 0.0,
             launch_wait: float = 4.0) -> dict:
    tag = f"{os.getpid()}-{uuid.uuid4().hex[:12]}"
    env = os.environ.copy()
    env.update(spec.env)
    launches: list[Launch] = []
    cleanup: list[int] = []
    base = fresh_bench_base() if spec.synaps_daemon else None
    jroot = None
    try:
        if spec.synaps_daemon:
            bin_ = spec.argv[0]
            env = synaps_env(base)
            env["BENCH_RUN_TAG"] = tag
            sock = str(base / "run" / "daemon.sock")
            stop_bench_daemon(bin_, base)
            time.sleep(0.5)
            server = subprocess.Popen([bin_, "daemon", "start"], env=env,
                                      stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
                                      stderr=subprocess.DEVNULL, preexec_fn=os.setsid)
            cleanup.append(os.getpgid(server.pid))
            if not wait_socket(sock, timeout_s):
                raise RuntimeError("synaps daemon did not become ready")
            # keep-warm attach clients (thin clients don't paint a detectable
            # banner; short launch timeout + explicit settle below)
            for _ in range(sessions):
                lc = launch([bin_, "--attach", "--new", "--keep-warm"], env, min(timeout_s, launch_wait), settle_s)
                launches.append(lc)
                cleanup.append(lc.pgid)
            time.sleep(max(settle_s, 3.0))
            roots = [server.pid] + [l.pid for l in launches]
        else:
            env["BENCH_RUN_TAG"] = tag
            if spec.jcode_home:
                # Isolated jcode state + sockets: never the user's ~/.jcode or
                # their running server. The TUI still spawns its own server as
                # it normally would; TEMP_SERVER/OWNER_PID make that server
                # exit with this harness as a backstop.
                jroot = Path(tempfile.mkdtemp(prefix="jcode-bench-"))
                (jroot / "home").mkdir()
                (jroot / "run").mkdir()
                env.update({"JCODE_HOME": str(jroot / "home"), "JCODE_RUNTIME_DIR": str(jroot / "run"),
                            "JCODE_TEMP_SERVER": "1", "JCODE_SERVER_OWNER_PID": str(os.getpid())})
            if spec.xdg_home:
                jroot = Path(tempfile.mkdtemp(prefix=f"{spec.name}-bench-"))
                for d in ("config", "data", "state", "cache"):
                    (jroot / d).mkdir()
                    env[f"XDG_{d.upper()}_HOME"] = str(jroot / d)
                # Pre-start the isolated service, as for the synaps daemon: on a
                # fresh state dir, TUIs launched before service.json exists each
                # start their own service. Its managed port is fixed (49374 by
                # default) and may be the user's own service: pick a free one.
                with socket.socket() as sk:
                    sk.bind(("127.0.0.1", 0))
                    port = sk.getsockname()[1]
                for cmd in (["service", "set", "port", str(port)], ["service", "start"]):
                    subprocess.run(["opencode", *cmd], env=env, timeout=60,
                                   stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
                                   stderr=subprocess.DEVNULL, check=False)
                if not wait_socket(str(jroot / "state" / "opencode" / "service.json"), timeout_s):
                    raise RuntimeError("opencode service did not become ready")
            for _ in range(sessions):
                lc = launch(spec.argv, env, timeout_s, settle_s)
                launches.append(lc)
                cleanup.append(lc.pgid)
            roots = [l.pid for l in launches]
        dead = [l for l in launches if exited(l.pid)]
        gone = len(dead)
        if dead:
            said = next((l.excerpt for l in dead if l.excerpt), None)
            print(f"  WARNING: {gone}/{sessions} launched session(s) exited before the sample; "
                  f"the total undercounts. First one said: {said}", flush=True)
        at_launch = sum_tree_pss(roots, cleanup, tag)
        res = at_launch
        if hold_s > 0:
            drain(launches, hold_s)
            res = sum_tree_pss(roots, cleanup, tag)
        res["launch"] = {k: at_launch[k] for k in ("mem_mb", "pss_mb", "swap_pss_mb", "pss_anon_mb",
                                                   "pss_file_mb", "process_count")}
        res["hold_s"] = hold_s
        res.update({"tool": spec.name, "sessions": sessions,
                    "ready": sum(l.ready for l in launches),
                    "exited_early": gone,
                    "excerpt": next((l.excerpt for l in launches if l.excerpt), None)})
        return res
    finally:
        for l in launches:
            try:
                os.close(l.fd)
            except Exception:
                pass
        for pgid in reversed(cleanup):
            kill_pgroup(pgid)
        if base is not None:
            stop_bench_daemon(spec.argv[0], base)
        if spec.xdg_home and jroot is not None:
            subprocess.run(["opencode", "service", "stop"], env=env, timeout=60,
                           stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
                           stderr=subprocess.DEVNULL, check=False)
        left = reap_tagged(tag)
        if left:
            print(f"WARNING: {spec.name}: processes survived cleanup: {left}", file=sys.stderr)
        if jroot is not None:
            shutil.rmtree(jroot, ignore_errors=True)
        if base is not None:
            shutil.rmtree(base, ignore_errors=True)


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--sessions", type=int, nargs="+", required=True,
                    help="one or more session counts, e.g. --sessions 1 10")
    ap.add_argument("--tools", nargs="*", default=["synaps"])
    ap.add_argument("--reps", type=int, default=1, help="repetitions per (tool, sessions); median reported")
    ap.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT_S)
    ap.add_argument("--settle", type=float, default=DEFAULT_SETTLE_S)
    ap.add_argument("--json-out", default=None, help="write raw results (every repetition) here")
    ap.add_argument("--hold", type=float, default=0.0,
                    help="after the last launch, keep every session idle (PTYs drained) this long "
                         "before the reported sample; a sample right after launch is kept too")
    ap.add_argument("--max-psi", type=float, default=1.0,
                    help="max memory PSI some-avg10 (%%) before/after a sample")
    ap.add_argument("--quiet-wait", type=float, default=900.0,
                    help="seconds to wait for memory pressure to drop before each sample")
    ap.add_argument("--launch-wait", type=float, default=4.0,
                    help="synaps only: seconds each attach client gets before the next launch "
                         "(thin clients paint no detectable frame, so this is a fixed wait). "
                         "0.1 makes a 1000-session run take minutes; the sample is still "
                         "taken --settle s after the last launch")
    ap.add_argument("--max-attempts", type=int, default=3,
                    help="retries of a repetition that saw memory pressure (last one is kept, flagged)")
    args = ap.parse_args()
    specs = build_specs()
    # a kill/hangup must still run every run's cleanup (finally blocks)
    for sig in (signal.SIGTERM, signal.SIGHUP):
        signal.signal(sig, lambda *_: sys.exit(130))
    stale = stale_bench_pids()
    if stale:
        print(f"reaping {len(stale)} process(es) left by an earlier run: {sorted(stale)}", flush=True)
        for pid in stale:
            try:
                os.kill(pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
        time.sleep(0.5)
    live = synaps_daemon_pids()
    DEFAULT_DAEMONS.update(p for p, c in live.items() if is_default_daemon(p, c))
    print(f"live default synaps daemon(s) (abort if gone): {sorted(DEFAULT_DAEMONS) or 'none'}", flush=True)
    print(f"pre-existing synaps daemons (must survive): {sorted(live) or 'none'}", flush=True)
    if "synaps" in args.tools:
        print(f"synaps base dir (isolated): {bench_base_dir()}", flush=True)
    # The synaps daemon (spawned from here, so it inherits this limit) holds
    # ~5 fds per session and this process one PTY per launch. At the usual
    # soft limit of 1024 the daemon refuses new sessions past ~170-200
    # ("Too many open files"): raise the soft limit for large runs, and say so.
    soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
    need = 6 * max(args.sessions) + 128
    if "synaps" in args.tools and soft != resource.RLIM_INFINITY and soft < need:
        new = need if hard == resource.RLIM_INFINITY else min(need, hard)
        if new > soft:
            resource.setrlimit(resource.RLIMIT_NOFILE, (new, hard))
            print(f"RLIMIT_NOFILE soft {soft} -> {new} for {max(args.sessions)} sessions "
                  f"(a daemon started from a default shell keeps {soft})", flush=True)
            soft = new
        if new < need:
            print(f"WARNING: RLIMIT_NOFILE hard limit {hard} is below the ~{need} fds "
                  f"{max(args.sessions)} sessions need; expect the daemon to refuse sessions "
                  f"past ~{max(0, (new - 16) // 6)}", flush=True)
    payload = {"date_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
               "machine": machine_info(), "args": vars(args), "versions": {},
               "live_daemons_before": live, "rlimit_nofile": [soft, hard], "summary": [], "runs": []}
    for name in args.tools:
        spec = specs.get(name)
        if not spec:
            print(f"unknown tool: {name}")
            continue
        payload["versions"][name] = tool_version(spec)
        print(f"=== {name} ({payload['versions'][name]}) ===", flush=True)
        for n in args.sessions:
            vals = []
            for rep in range(1, args.reps + 1):
                for attempt in range(1, args.max_attempts + 1):
                    before = wait_quiet(args.max_psi, args.quiet_wait)
                    check_live_daemons(live)
                    res = run_tool(spec, n, args.timeout, args.settle, args.hold, args.launch_wait)
                    check_live_daemons(live)
                    after = mem_pressure()
                    res.update({"pressure_before": before, "pressure_after": after, "attempt": attempt})
                    quiet = max(before.get("psi_some_avg10", 0), after.get("psi_some_avg10", 0)) <= args.max_psi
                    res["quiet"] = quiet
                    if quiet:
                        break
                    print(f"  {n:>2} session(s) rep {rep}: memory pressure during sample "
                          f"({before} -> {after}), retrying", flush=True)
                res["rep"] = rep
                payload["runs"].append(res)
                vals.append(res["mem_mb"])
                print(f"  {n:>2} session(s) rep {rep}: {res['mem_mb']:7.1f} MB settled, "
                      f"{res['launch']['mem_mb']:7.1f} MB at launch  "
                      f"(pss {res['pss_mb']} + swap {res['swap_pss_mb']}) "
                      f"procs={res['process_count']} outside-tree={res['outside_tree_mem_mb']} MB "
                      f"psi={res['pressure_after'].get('psi_some_avg10')}{'' if res['quiet'] else ' PRESSURED'} "
                      f"ready={res['ready']}/{n} :: {res['excerpt']}", flush=True)
                if args.json_out:  # incremental: an abort keeps what was measured
                    Path(args.json_out).write_text(json.dumps(payload, indent=2) + "\n")
            med = round(statistics.median(vals), 1)
            payload["summary"].append({"tool": name, "sessions": n, "median_mem_mb": med, "raw_mem_mb": vals})
            print(f"  {n:>2} session(s) median: {med} MB  raw={vals}", flush=True)
    payload["live_daemons_after"] = synaps_daemon_pids()
    if args.json_out:
        Path(args.json_out).write_text(json.dumps(payload, indent=2) + "\n")
        print(f"wrote {args.json_out}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())