#!/usr/bin/env -S python3 -I
"""usage: tools/q  (one-line summary)
       tools/q submit [--kind check|test|emu|full|batch|long] [--prio high|normal] [--label TEXT] [-C DIR]
                [--progress 'GLOB|N']… -- CMD…  (prints ID)
                [--stall SECS]
       tools/q wait ID [--max SECONDS] [--tail N] | ls | top [--once] | stats [--since 24h] [--id ID] | cancel ID | kill ID
       tools/q selftest | legacy [-w SECS] -- CMD…  (exclusive QLOCK, wait/run logged as kind=legacy)
Job queue for heavy runs: each job is a systemd user unit that waits for capacity (weighted semaphore, 4 units:
check 2, test 1, emu 2 + one of 2 emulator slots, full 4, batch 2 (replay batches), long 2 (dirty-room wine / full-set
runs); prio high first, strict FIFO behind the head), holds QLOCK SHARED while it runs, logs to QDIR/log.jsonl. wait exits with the job's code, or 75 = not done yet.
Stall watchdog: a running job whose process tree used < QSTALL_CPU % of one core and whose out.log did not grow for
QSTALL s (or --stall) shows as `stalled` in ls / top with its blocked threads; wait on it exits 76. It is not killed:
`kill ID` stops it (cancel, TERM to the whole tree, KILL after QKILL_GRACE s) and waits until it is gone.
emu jobs: QEMU_SLOT, QEMU_PORT and, above slot 0, ANDROID_AVD_HOME select the slot's own AVD; the slot's emulator is saved to snapshot q_pre before the job and restored after it (also when cancelled), plus
user_rotation / wm size / wm density; job env QEMU_HYGIENE=0 skips that, QADB names adb. ls warns when a job's
scratch dir (/tmp/claude-UID/<mangled cwd>) passes QSCRATCH_GB.
Env: QDIR (default <repo>/target/queue), QLOCK (default QDIR/compat.lock), QPOLL (s), QSTALL (600), QSTALL_CPU (2),
QKILL_GRACE (10), QSCRATCH_GB (3), QSCRATCH_ROOT."""
import glob as globmod, fcntl, json, os, random, re, shlex, shutil, signal, subprocess, sys, tempfile, threading, time

_me = os.path.realpath(__file__)
MAIN = _me[:_me.find("/target/queue/")] if "/target/queue/" in _me else os.path.dirname(os.path.dirname(_me))
QDIR = os.path.realpath(os.environ.get("QDIR") or f"{MAIN}/target/queue")
QLOCK = os.path.realpath(os.environ.get("QLOCK") or f"{QDIR}/compat.lock")
ARCH_TIMES = f"{MAIN}/devdata/arch_ref/times.txt"
POLL = float(os.environ.get("QPOLL", "2"))
CAP = 4
WEIGHT = {"check": 2, "test": 1, "emu": 2, "full": 4, "batch": 2, "long": 2}
EMU_SLOTS = 2
WAIT_MAX = 540  # one Bash tool call never reaches the 600 s limit
EXIT_PENDING = 75  # EX_TEMPFAIL: job not finished, call `q wait` again
EXIT_CANCELLED = 125
EXIT_STALLED = 76  # job running but stalled (see the watchdog); not killed
STALL = float(os.environ.get("QSTALL", "600"))
STALL_CPU = float(os.environ.get("QSTALL_CPU", "2")) / 100
KILL_GRACE = float(os.environ.get("QKILL_GRACE", "10"))
SCRATCH_GB = float(os.environ.get("QSCRATCH_GB", "3"))
SCRATCH_ROOT = os.environ.get("QSCRATCH_ROOT") or f"/tmp/claude-{os.getuid()}"
# Unit prefix: the real queue gets q-ID, a scratch QDIR (selftest) qt-ID, so both can coexist.
PREFIX = "q" if QDIR == f"{MAIN}/target/queue" else "qt"


def jdir(i): return f"{QDIR}/jobs/{i}"
def now(): return time.time()


def rjson(p, d=None):
    try:
        with open(p) as f: return json.load(f)
    except (OSError, ValueError): return d


def wjson(p, o):
    t = f"{p}.tmp{os.getpid()}"
    with os.fdopen(os.open(t, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600), "w") as f: json.dump(o, f)
    os.replace(t, p)


# Variables never written to env.json or passed to jobs: credentials and desktop-session handles.
SECRET = re.compile(r"TOKEN|SECRET|PASS|AUTH|COOKIE|CREDENTIAL|PRIVATE|API_?KEY|SESSION")
KEEP_DAYS, KEEP_JOBS = 14, 300


def job_env(src):
    return {k: v for k, v in src.items() if k.startswith("TH08_") or not SECRET.search(k)}


def gc_jobs(s):
    """Removes finished job dirs older than KEEP_DAYS beyond the newest KEEP_JOBS; log.jsonl keeps their records."""
    try: ids = sorted(os.listdir(f"{QDIR}/jobs"), reverse=True)
    except OSError: return
    for i in ids[KEEP_JOBS:]:
        d = jdir(i)
        if i in s["jobs"] or now() - os.path.getmtime(d) < KEEP_DAYS * 86400: continue
        if rjson(f"{d}/status.json", {}).get("state") in ("queued", "running"): continue
        shutil.rmtree(d, ignore_errors=True)


class Locked:
    """The queue state (QDIR/state.json) under an exclusive flock of QDIR/state.lock."""
    def __enter__(self):
        os.makedirs(f"{QDIR}/jobs", exist_ok=True)
        self.f = open(f"{QDIR}/state.lock", "a")
        fcntl.flock(self.f, fcntl.LOCK_EX)
        self.s = rjson(f"{QDIR}/state.json", {"seq": 0, "jobs": {}})
        return self.s

    def __exit__(self, *a):
        wjson(f"{QDIR}/state.json", self.s)
        self.f.close()


def alive(pid):
    try: os.kill(pid, 0); return True
    except ProcessLookupError: return False
    except PermissionError: return True


def prune(s):
    """Drops entries whose runner died (no unit started within 60 s, or pid gone) and marks their status."""
    for i, e in list(s["jobs"].items()):
        dead = (e.get("pid") and not alive(e["pid"])) or (not e.get("pid") and now() - e["submit"] > 60)
        if dead:
            del s["jobs"][i]
            st = rjson(f"{jdir(i)}/status.json", {})
            if st.get("state") in ("queued", "running"):
                st.update(state="lost", end=now(), rc=EXIT_CANCELLED)
                wjson(f"{jdir(i)}/status.json", st)


def order(s):
    """Queued jobs in start order: prio high first, then submit sequence."""
    q = [(i, e) for i, e in s["jobs"].items() if e["state"] == "queued"]
    return sorted(q, key=lambda x: (x[1]["prio"] != "high", x[1]["seq"]))


def used(s):
    r = [e for e in s["jobs"].values() if e["state"] == "running"]
    return sum(e["weight"] for e in r), sum(1 for e in r if e["kind"] == "emu")


def append_log(rec):
    os.makedirs(QDIR, exist_ok=True)
    with open(f"{QDIR}/log.jsonl", "a") as f:
        fcntl.flock(f, fcntl.LOCK_EX)
        f.write(json.dumps(rec) + "\n")


def read_log():
    out = []
    try:
        with open(f"{QDIR}/log.jsonl") as f:
            for l in f:
                try: out.append(json.loads(l))
                except ValueError: pass
    except OSError: pass
    return out


def worktree(cwd):
    m = re.search(r"/worktrees/[^/]+/([^/]+)", cwd)
    return m.group(1) if m else ("main" if cwd.startswith(MAIN) else os.path.basename(cwd))


# ---- process tree, stall watchdog -------------------------------------------------------------------------------

def cgroup_dir():
    """The runner's own cgroup v2 dir (the job's systemd unit), or None."""
    try:
        for l in open("/proc/self/cgroup"):
            if l.startswith("0::"):
                d = "/sys/fs/cgroup" + l[3:].strip()
                return d if os.path.isfile(f"{d}/cgroup.procs") else None
    except OSError: pass
    return None


def tree_pids(cg, root):
    """The job's processes except the runner: its cgroup, or (no cgroup) every process in the child's session."""
    me = os.getpid()
    if cg:
        try: return [int(x) for x in open(f"{cg}/cgroup.procs").read().split() if int(x) != me]
        except (OSError, ValueError): pass
    out = []
    for p in os.listdir("/proc"):
        if not p.isdigit(): continue
        try:
            f = open(f"/proc/{p}/stat").read().rpartition(")")[2].split()
            if int(f[3]) == root: out.append(int(p))  # field 6 = session id
        except (OSError, ValueError, IndexError): pass
    return out


TICK = os.sysconf("SC_CLK_TCK")


def tree_cpu(cg, root):
    """CPU seconds used by the job's process tree so far (cgroup total minus the runner's own share)."""
    if cg:
        try:
            u = next(int(l.split()[1]) for l in open(f"{cg}/cpu.stat") if l.startswith("usage_usec "))
            t = os.times()
            return u / 1e6 - t.user - t.system
        except (OSError, ValueError, StopIteration): pass
    tot = 0.0
    for p in tree_pids(None, root):
        try:
            f = open(f"/proc/{p}/stat").read().rpartition(")")[2].split()
            tot += sum(int(x) for x in f[11:15]) / TICK  # utime stime cutime cstime
        except (OSError, ValueError, IndexError): pass
    return tot


def blocked_threads(pids, top=6):
    """Threads of pids grouped by (comm, wchan), busiest first: 'comm wchan ×n (cpu s)'."""
    g = {}
    for p in pids:
        try: tids = os.listdir(f"/proc/{p}/task")
        except OSError: continue
        for t in tids:
            b = f"/proc/{p}/task/{t}"
            try:
                raw = open(f"{b}/stat").read()
                comm = raw[raw.find("(") + 1:raw.rfind(")")]
                f = raw.rpartition(")")[2].split()
                state, cpu = f[0], (int(f[11]) + int(f[12])) / TICK
                wc = open(f"{b}/wchan").read().strip() or "-"
            except (OSError, ValueError, IndexError): continue
            if state == "R": wc = "running"
            k = (comm, wc)
            n, c = g.get(k, (0, 0.0)); g[k] = (n + 1, c + cpu)
    rows = sorted(g.items(), key=lambda x: (-x[1][1], -x[1][0]))[:top]
    return ", ".join(f"{c} {w}" + (f" ×{n}" if n > 1 else "") + f" ({cpu:.0f} s)" for (c, w), (n, cpu) in rows) or "no threads"


class Watchdog:
    """Stall check of a running job: CPU of the tree and out.log growth over the last `win` seconds."""
    def __init__(self, cg, root, out_path, win):
        self.cg, self.root, self.out, self.win = cg, root, out_path, win
        self.samples, self.size, self.grow, self.t0 = [], -1, now(), now()

    def check(self):
        """None when the job is active, else the stall record."""
        t = now(); c = tree_cpu(self.cg, self.root)
        self.samples.append((t, c))
        while len(self.samples) > 1 and self.samples[1][0] <= t - self.win: self.samples.pop(0)
        try: sz = os.path.getsize(self.out)
        except OSError: sz = 0
        if sz != self.size: self.size, self.grow = sz, t
        ot, oc = self.samples[0]
        if t - self.grow < self.win or ot > t - self.win: return None
        use = max(0.0, (c - oc) / (t - ot))
        if use >= STALL_CPU: return None
        return {"since": self.grow, "cpu_pct": round(100 * use, 2), "silent": round(t - self.grow),
                "reason": f"out.log silent {t - self.grow:.0f} s, {100 * use:.2f} % of one core over {t - ot:.0f} s",
                "threads": blocked_threads(tree_pids(self.cg, self.root))}


def kill_tree(cg, root, sig):
    try: os.killpg(root, sig)
    except (ProcessLookupError, PermissionError): pass
    for p in tree_pids(cg, root):
        try: os.kill(p, sig)
        except (ProcessLookupError, PermissionError): pass


# ---- emulator hygiene, scratch quota ---------------------------------------------------------------------------

def serial_slot(serial):
    """The emulator slot of a running emulator: its process's ANDROID_AVD_HOME, slot 0 when it has none."""
    port = serial.rsplit("-", 1)[-1]
    for p in os.listdir("/proc"):
        if not p.isdigit(): continue
        try:
            with open(f"/proc/{p}/cmdline", "rb") as f: argv = f.read().decode(errors="replace").split("\0")
            if "qemu-system" not in argv[0] or not any(a == "-port" and b == port for a, b in zip(argv, argv[1:])): continue
            with open(f"/proc/{p}/environ", "rb") as f: envs = f.read().decode(errors="replace").split("\0")
        except OSError: continue
        m = next((re.search(r"avd-slot(\d+)$", v) for v in envs if v.startswith("ANDROID_AVD_HOME=")), None)
        return int(m.group(1)) if m else 0
    return 0


class Emu:
    """Snapshot q_pre of the job slot's running emulators before an emu job, restored with rotation / wm size / density after."""
    def __init__(self, env, log_path, slot=0):
        self.env, self.log, self.saved, self.slot = env, log_path, {}, slot
        sdk = env.get("ANDROID_HOME") or env.get("ANDROID_SDK_ROOT") or ""
        self.adb = env.get("QADB") or shutil.which("adb", path=env.get("PATH")) or (
            f"{sdk}/platform-tools/adb" if sdk and os.path.exists(f"{sdk}/platform-tools/adb") else None)
        self.on = env.get("QEMU_HYGIENE", "1") != "0" and bool(self.adb)

    def run(self, *a, t=120):
        try:
            r = subprocess.run([self.adb, *a], env=self.env, capture_output=True, text=True, timeout=t, stdin=subprocess.DEVNULL)
            out = (r.stdout + r.stderr).strip()
            rc = r.returncode
        except (OSError, subprocess.TimeoutExpired) as x: out, rc = str(x), -1
        with open(self.log, "a") as f: f.write(f"$ adb {' '.join(a)} -> {rc}\n" + (out + "\n" if out else ""))
        return rc, out

    def serials(self):
        rc, out = self.run("devices")
        return [l.split()[0] for l in out.splitlines() if l.startswith("emulator-") and l.split()[1:2] == ["device"]
                and serial_slot(l.split()[0]) == self.slot]

    def save(self):
        if not self.on: return "emu hygiene off" if self.adb else "no adb"
        for s in self.serials():
            rot = self.run("-s", s, "shell", "settings", "get", "system", "user_rotation")[1].strip()
            ok = self.run("-s", s, "emu", "avd", "snapshot", "save", "q_pre")[0] == 0
            self.saved[s] = {"snap": ok, "rot": rot if rot.isdigit() else None}
        return "saved q_pre on " + (", ".join(self.saved) if self.saved else "no running emulator")

    def restore(self):
        if not self.on: return ""
        done = []
        for s in self.serials():
            sv = self.saved.get(s, {})
            if sv.get("snap"):
                self.run("-s", s, "emu", "avd", "snapshot", "load", "q_pre")
                self.run("-s", s, "wait-for-device", t=60)
            if sv.get("rot") is not None: self.run("-s", s, "shell", "settings", "put", "system", "user_rotation", sv["rot"])
            self.run("-s", s, "shell", "wm", "size", "reset"); self.run("-s", s, "shell", "wm", "density", "reset")
            done.append(s + (" (snapshot)" if sv.get("snap") else ""))
        return "restored " + (", ".join(done) if done else "no running emulator")


def du(p):
    tot = 0
    for r, ds, fs in os.walk(p):
        for f in fs:
            try: tot += os.lstat(os.path.join(r, f)).st_blocks * 512
            except OSError: pass
    return tot


def scratch_warn(job):
    """A warning when the job's scratch dir (else all of SCRATCH_ROOT) passes SCRATCH_GB, or ''; sizes cached 60 s."""
    env = rjson(f"{jdir(job['id'])}/env.json", {})
    cand = []
    tmp = env.get("TMPDIR") or ""
    if tmp.startswith(SCRATCH_ROOT + "/"): cand.append(SCRATCH_ROOT + "/" + tmp[len(SCRATCH_ROOT) + 1:].split("/")[0])
    cand.append(f"{SCRATCH_ROOT}/" + re.sub(r"[^A-Za-z0-9]", "-", job["cwd"]))
    p = next((c for c in cand if os.path.isdir(c)), None)
    whose = "job scratch" if p else "scratch total (job's dir not found)"
    p = p or SCRATCH_ROOT
    if not os.path.isdir(p): return ""
    cache = rjson(f"{QDIR}/scratch.json", {})
    t, n = cache.get(p, (0, 0))
    if now() - t > 60:
        n = du(p); cache[p] = (now(), n)
        try: wjson(f"{QDIR}/scratch.json", cache)
        except OSError: pass
    return f"{whose} {p} is {n / 2**30:.1f} GB (> {SCRATCH_GB:g} GB)" if n > SCRATCH_GB * 2**30 else ""


# ---- estimates -------------------------------------------------------------------------------------------------

def pct(v, p):
    if not v: return None
    v = sorted(v)
    return v[min(len(v) - 1, int(p / 100 * len(v)))]


def est_run(kind, hist):
    """Expected run seconds of a kind: median of the last 20 finished ok jobs; arch_ref times.txt as a fallback."""
    r = [x["run"] for x in hist if x.get("kind") == kind and x.get("rc") == 0 and x.get("run") is not None][-20:]
    if r: return pct(r, 50)
    if kind in ("check", "full"):
        try:
            with open(ARCH_TIMES) as f: tot = sum(float(l.split()[1]) for l in f if len(l.split()) >= 2)
            j = max(2, 16 * WEIGHT[kind] // CAP)
            return 150 + tot * (1.6 if kind == "full" else 1.4) / j  # fmt/build/wasm/android/test + items (+ emit-off)
        except (OSError, ValueError): pass
    return {"test": 120, "emu": 300, "batch": 1200, "long": 3600}.get(kind, 600)


def simulate(s, hist):
    """Estimated start time of each queued job: replays the strict-order admission on estimated run ends."""
    t = now()
    run = [(max(t, e["start"] + est_run(e["kind"], hist)), e["weight"], e["kind"] == "emu")
           for e in s["jobs"].values() if e["state"] == "running" and e.get("start")]
    res = {}
    for i, e in order(s):
        while True:
            w = sum(r[1] for r in run); emu = sum(r[2] for r in run)
            if w + e["weight"] <= CAP and not (e["kind"] == "emu" and emu >= EMU_SLOTS): break
            if not run: break
            run.sort(); t = max(t, run.pop(0)[0])
        res[i] = t
        run.append((t + est_run(e["kind"], hist), e["weight"], e["kind"] == "emu"))
    return res


def summary(s=None):
    if s is None:
        with Locked() as s: prune(s)
    w, emu = used(s)
    q = order(s)
    r = sum(1 for e in s["jobs"].values() if e["state"] == "running")
    oldest = max((now() - e["submit"] for _, e in q), default=0)
    return f"q: {r} running (capacity {w}/{CAP}, emu {emu}/{EMU_SLOTS}), {len(q)} queued, oldest waiting {oldest:.0f} s"


# ---- commands --------------------------------------------------------------------------------------------------

def cmd_submit(a):
    kind, prio, label, cwd, prog, stall = "check", "normal", "", os.getcwd(), [], None
    while a and a[0] != "--":
        o = a.pop(0)
        if o in ("--kind", "--prio", "--label", "-C", "--progress", "--stall") and a: v = a.pop(0)
        else: die(f"submit: unknown option {o}")
        if o == "--progress": prog.append(v)
        elif o == "--stall": stall = parse_dur(v)
        elif o == "--kind": kind = v
        elif o == "--prio": prio = v
        elif o == "--label": label = v
        else: cwd = os.path.realpath(v)
    a = a[1:] if a[:1] == ["--"] else a
    if kind not in WEIGHT or prio not in ("high", "normal") or not a: die(__doc__, 2)
    if not os.path.isdir(cwd): die(f"submit: no directory {cwd}")
    env = job_env(os.environ)
    try:  # local paths (gitignored, so missing in worktrees) from the main checkout
        for ln in open(f"{MAIN}/packaging/local/local.env"):
            k, eq, v = ln.strip().partition("=")
            if eq and k and not k.startswith("#"): env.setdefault(k, v)
    except OSError: pass
    env.setdefault("J", str(max(2, 16 * WEIGHT[kind] // CAP)))
    i = time.strftime("%Y%m%d-%H%M%S") + "-" + "".join(random.choices("0123456789abcdef", k=4))
    d = jdir(i)
    os.makedirs(d, mode=0o700)
    job = {"id": i, "cmd": a, "cwd": cwd, "label": label, "kind": kind, "prio": prio, "weight": WEIGHT[kind],
           "submit": now(), "worktree": worktree(cwd), "unit": f"{PREFIX}-{i}", "progress": prog,
           "stall": stall}
    wjson(f"{d}/job.json", job)
    wjson(f"{d}/env.json", env)
    wjson(f"{d}/status.json", {"state": "queued", "submit": job["submit"]})
    shutil.copy(os.path.abspath(__file__), f"{d}/runner.py")  # the unit survives removal of the caller's worktree
    with Locked() as s:
        gc_jobs(s)
        s["seq"] += 1
        s["jobs"][i] = {"state": "queued", "kind": kind, "prio": prio, "weight": WEIGHT[kind], "seq": s["seq"],
                        "submit": job["submit"], "label": label, "cwd": cwd, "pid": None}
    cpu = os.cpu_count() * 75  # 75 % of the machine + nice 10
    r = subprocess.run(["systemd-run", "--user", "-q", "--collect", "--unit", job["unit"], f"--working-directory={QDIR}",
                        "--description", f"th08 queue {kind} {label}", "-p", "Nice=10", "-p", f"CPUQuota={cpu}%", "-p", "LimitCORE=0",
                        "-E", f"QDIR={QDIR}", "-E", f"QLOCK={QLOCK}", "-E", f"QPOLL={POLL}",
                        "-E", f"QSTALL={STALL}", "-E", f"QSTALL_CPU={STALL_CPU * 100}", "-E", f"QKILL_GRACE={KILL_GRACE}",
                        "/usr/bin/python3", "-I", f"{d}/runner.py", "_run", i], capture_output=True, text=True)
    if r.returncode:
        with Locked() as s: s["jobs"].pop(i, None)
        wjson(f"{d}/status.json", {"state": "failed-to-start", "submit": job["submit"], "end": now(), "rc": 1})
        die(f"submit: systemd-run failed: {r.stderr.strip()}")
    print(i)


def finish(i, job, st, state, rc, out_path):
    st.update(state=state, end=now(), rc=rc)
    wjson(f"{jdir(i)}/status.json", st)
    with Locked() as s: s["jobs"].pop(i, None)
    start = st.get("start")
    append_log({"id": i, "kind": job["kind"], "label": job["label"], "prio": job["prio"], "weight": job["weight"],
                "worktree": job["worktree"], "cwd": job["cwd"], "cmd": job["cmd"], "status": state, "rc": rc,
                "submit": job["submit"], "start": start, "end": st["end"],
                "wait": round((start or st["end"]) - job["submit"], 1),
                "lockwait": st.get("lockwait"), "run": round(st["end"] - start, 1) if start else None,
                "J": st.get("J"), "load": st.get("load"), "load_max": st.get("load_max"), "out": out_path,
                "stalled": st.get("stalled", 0), "emu": st.get("emu")})


def cmd_run(a):
    """The unit body: wait for capacity in strict order, take the shared legacy lock, run, record."""
    i = a[0]; d = jdir(i)
    job = rjson(f"{d}/job.json"); env = rjson(f"{d}/env.json")
    st = rjson(f"{d}/status.json")
    out_path = f"{d}/out.log"
    child = [None]
    term = [False]
    term_t = [None]
    cg = cgroup_dir()

    def on_term(*_):
        term[0] = True
        if term_t[0] is None: term_t[0] = now()
        if child[0]: kill_tree(cg, child[0].pid, signal.SIGTERM)
    signal.signal(signal.SIGTERM, on_term)

    with Locked() as s:
        if i in s["jobs"]: s["jobs"][i]["pid"] = os.getpid()
    while True:
        with Locked() as s:
            prune(s)
            e = s["jobs"].get(i)
            if e is None or e.get("cancel") or term[0]:
                break
            if not os.path.isdir(job["cwd"]):
                e["cancel"] = "worktree removed"; break
            q = order(s)
            w, emu = used(s)
            if q and q[0][0] == i and w + e["weight"] <= CAP and not (e["kind"] == "emu" and emu >= EMU_SLOTS):
                if e["kind"] == "emu":
                    taken = {x.get("emu_slot", 0) for x in s["jobs"].values() if x["state"] == "running" and x["kind"] == "emu"}
                    e["emu_slot"] = min(set(range(EMU_SLOTS)) - taken)
                e["state"] = "running"; e["start"] = now(); break
        time.sleep(POLL)
    e = e or {}
    if e.get("state") != "running":
        st["reason"] = e.get("cancel") or "cancelled"
        finish(i, job, st, "cancelled", EXIT_CANCELLED, out_path); return
    # Shared .agents.lock: q jobs share it, legacy exclusive holders (flock target/.agents.lock) exclude them.
    os.makedirs(os.path.dirname(QLOCK), exist_ok=True)
    lk = open(QLOCK, "a")
    t0 = now()
    st.update(state="running", start=now(), lock="waiting"); wjson(f"{d}/status.json", st)
    # Blocking flock in a thread: the job joins the kernel's waiters (an LOCK_NB poll starves behind back-to-back
    # legacy holders); the main thread still sees cancel / a removed worktree.
    got = threading.Event()
    threading.Thread(target=lambda: (fcntl.flock(lk, fcntl.LOCK_SH), got.set()), daemon=True).start()
    while not term[0] and not got.wait(POLL):
        if not os.path.isdir(job["cwd"]): term[0] = True; st["reason"] = "worktree removed"
    st["lockwait"] = round(now() - t0, 1)
    if term[0]:
        st.update(start=None); st.setdefault("reason", "cancelled")
        finish(i, job, st, "cancelled", EXIT_CANCELLED, out_path); return
    st.update(start=now(), lock="shared"); wjson(f"{d}/status.json", st)
    slot = e.get("emu_slot", 0)
    if job["kind"] == "emu":
        env.update(QEMU_SLOT=str(slot), QEMU_PORT=str(5600 + 20 * slot))
        if slot: env["ANDROID_AVD_HOME"] = os.path.expanduser(f"~/.android/avd-slot{slot}")
    emu = Emu(env, f"{d}/emu.log", slot) if job["kind"] == "emu" else None
    if emu:
        st["emu"] = emu.save(); wjson(f"{d}/status.json", st)
    with os.fdopen(os.open(out_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600), "wb") as out:
        try:
            child[0] = subprocess.Popen(job["cmd"], cwd=job["cwd"], env=env, stdout=out, stderr=subprocess.STDOUT,
                                        stdin=subprocess.DEVNULL, start_new_session=True)
        except OSError as x:
            out.write(f"q: cannot start {job['cmd']}: {x}\n".encode())
            if emu: st["emu"] += "; " + emu.restore()
            finish(i, job, st, "failed", 127, out_path); return
        loads = []
        st["J"] = env.get("J")
        wd = Watchdog(cg, child[0].pid, out_path, job.get("stall") or STALL)
        killed = False
        last_wd = 0.0
        while child[0].poll() is None:
            time.sleep(POLL)
            try: loads.append(float(open("/proc/loadavg").read().split()[0]))
            except (OSError, ValueError): pass
            if loads: st.update(load=round(sum(loads) / len(loads), 1), load_max=round(max(loads), 1))
            if not os.path.isdir(job["cwd"]) and not term[0]:
                st["reason"] = "worktree removed"; on_term()
            if term[0] and not killed and now() - term_t[0] >= KILL_GRACE:
                kill_tree(cg, child[0].pid, signal.SIGKILL); killed = True
            stall = wd.check() if not term[0] else None
            if stall:
                if not st.get("stall") or now() - last_wd >= 60:
                    st["stall"] = stall; st["stalled"] = st.get("stalled", 0) + (0 if st.get("stall_on") else 1)
                    st["stall_on"] = True; last_wd = now(); wjson(f"{d}/status.json", st)
            elif st.get("stall"):
                st.pop("stall"); st["stall_on"] = False; wjson(f"{d}/status.json", st)
        st.pop("stall", None); st.pop("stall_on", None)
        if term[0]: kill_tree(cg, child[0].pid, signal.SIGKILL)  # leftovers of a cancelled job
    if emu:
        st["emu"] = (st.get("emu") or "") + "; " + emu.restore()
    rc = child[0].returncode
    if term[0]:
        with Locked() as s: why = (s["jobs"].get(i) or {}).get("cancel")
        st.setdefault("reason", why or "cancelled"); finish(i, job, st, "cancelled", EXIT_CANCELLED, out_path)
    else:
        finish(i, job, st, "done" if rc == 0 else "failed", rc if rc >= 0 else 128 - rc, out_path)


def resolve(i):
    if os.path.isdir(jdir(i)): return i
    m = [x for x in os.listdir(f"{QDIR}/jobs")] if os.path.isdir(f"{QDIR}/jobs") else []
    m = [x for x in m if x.endswith(i) or x.startswith(i)]
    if len(m) == 1: return m[0]
    die(f"no job {i}" if not m else f"ambiguous job {i}: {' '.join(m)}")


def job_line(i, st, job):
    w = (st.get("start") or st.get("end") or now()) - job["submit"]
    r = (st["end"] - st["start"]) if st.get("start") and st.get("end") else None
    return (f"q stats: job {i} kind={job['kind']} label={job['label'] or '-'} waited {w:.0f} s"
            + (f" (lock {st['lockwait']:.0f} s)" if st.get("lockwait") else "") + (f", ran {r:.0f} s" if r is not None else "")
            + (f", J={st['J']} load {st['load']} (max {st['load_max']})" if st.get("load") is not None else ""))


def cmd_wait(a):
    mx, tail, i = WAIT_MAX, 40, None
    while a:
        o = a.pop(0)
        if o == "--max": mx = float(a.pop(0))
        elif o == "--tail": tail = int(a.pop(0))
        elif i is None: i = o
        else: die(__doc__, 2)
    if i is None: die(__doc__, 2)
    i = resolve(i)
    job = rjson(f"{jdir(i)}/job.json")
    t0 = now()
    while True:
        st = rjson(f"{jdir(i)}/status.json", {})
        if st.get("state") not in ("queued", "running") or st.get("stall"): break
        if now() - t0 >= mx: break
        with Locked() as s: prune(s)
        time.sleep(min(POLL, max(0.05, mx - (now() - t0))))
    hist = read_log()
    with Locked() as s:
        prune(s); summ = summary(s)
        st = rjson(f"{jdir(i)}/status.json", {})
        if st.get("state") == "queued":
            q = [x for x, _ in order(s)]
            est = simulate(s, hist).get(i, now()) - now()
            pos = q.index(i) + 1 if i in q else 0
            print(f"still queued: position {pos} of {len(q)}, waited {now() - job['submit']:.0f} s, est. start in ~{max(0, est):.0f} s")
            print(summ); sys.exit(EXIT_PENDING)
        if st.get("state") == "running":
            ran = now() - st["start"]
            if st.get("stall"):
                print(stall_text(i, st["stall"]))
                print(summ); sys.exit(EXIT_STALLED)
            if st.get("lock") == "waiting":
                print(f"admitted, waiting {ran:.0f} s for a legacy exclusive .agents.lock holder (flock without q)")
            else:
                left = est_run(job["kind"], hist) - ran
                print(f"running for {ran:.0f} s, est. ~{max(0, left):.0f} s left{' (over the estimate)' if left < 0 else ''}{progress(job, st)}")
            print(summ); sys.exit(EXIT_PENDING)
    out = f"{jdir(i)}/out.log"
    try:
        with open(out, errors="replace") as f: lines = f.readlines()
    except OSError: lines = []
    for l in lines[-tail:] if tail > 0 else []: sys.stdout.write(l)
    rc = st.get("rc", 1)
    print(f"job {i} {st.get('state')}{' (' + st['reason'] + ')' if st.get('reason') else ''}, exit {rc}; full output: {out}")
    print(job_line(i, st, job))
    print(summ)
    sys.exit(rc)


def stall_text(i, sl):
    return (f"STALLED: job {i} {sl['reason']}\n  blocked threads: {sl['threads']}\n"
            f"  not killed; `tools/q wait {i}` again if it should resume, `tools/q kill {i}` to stop it")


def progress(job, st):
    """Files matching the job's --progress 'GLOB|N' specs (relative to its cwd, newer than its start); else arch_check items."""
    if job.get("progress") and st.get("start"):
        done = total = 0
        for spec in job["progress"]:
            g, _, n = spec.partition("|")
            try: done += sum(1 for f in globmod.glob(os.path.join(job["cwd"], g)) if os.path.getmtime(f) >= st["start"])
            except OSError: pass
            total += int(n) if n.isdigit() else 0
        return f", {done}/{total}" if total else f", {done} done"
    if job["kind"] not in ("check", "full"): return ""
    env = rjson(f"{jdir(job['id'])}/env.json", {})
    out = env.get("RUN_OUT") or f"{env.get('CARGO_TARGET_DIR') or job['cwd'] + '/target'}/arch_run"
    try: n = sum(1 for d in os.scandir(out) if os.path.isfile(f"{d.path}/digest") and os.path.getmtime(f"{d.path}/digest") >= st["start"])
    except OSError: return ""
    return f", {n} arch_check items done" if n else ""


def cmd_cancel(a):
    if len(a) != 1: die(__doc__, 2)
    i = resolve(a[0]); job = rjson(f"{jdir(i)}/job.json")
    with Locked() as s:
        e = s["jobs"].get(i)
        if e is None:
            print(f"job {i} is not queued or running"); return
        e["cancel"] = "cancelled"
        state = e["state"]
    if state == "running":
        subprocess.run(["systemctl", "--user", "kill", "-s", "TERM", "--kill-whom=main", job["unit"]], capture_output=True)
    print(f"job {i} cancel requested ({state})")


def cmd_kill(a):
    """Cancel and wait until the job is gone (runner: TERM, KILL after KILL_GRACE; else the whole unit is killed)."""
    if len(a) != 1: die(__doc__, 2)
    i = resolve(a[0]); job = rjson(f"{jdir(i)}/job.json")
    with Locked() as s:
        e = s["jobs"].get(i)
        if e is None:
            print(f"job {i} is not queued or running"); return
        e["cancel"] = "killed"
        state = e["state"]
    if state == "running":
        subprocess.run(["systemctl", "--user", "kill", "-s", "TERM", "--kill-whom=main", job["unit"]], capture_output=True)
    t0 = now()
    while now() - t0 < KILL_GRACE + 30 + 2 * POLL:
        st = rjson(f"{jdir(i)}/status.json", {})
        if st.get("state") not in ("queued", "running"):
            print(f"job {i} {st.get('state')} ({st.get('reason', '-')}) after {now() - t0:.0f} s"); return
        time.sleep(min(POLL, 0.5))
    subprocess.run(["systemctl", "--user", "kill", "-s", "KILL", job["unit"]], capture_output=True)
    subprocess.run(["systemctl", "--user", "stop", job["unit"]], capture_output=True)
    st = rjson(f"{jdir(i)}/status.json", {})
    if st.get("state") in ("queued", "running"):
        st["reason"] = "killed with its unit"
        finish(i, job, st, "cancelled", EXIT_CANCELLED, f"{jdir(i)}/out.log")
    print(f"job {i} cancelled (killed with its unit after {now() - t0:.0f} s)")


def cmd_ls(a):
    hist = read_log()
    with Locked() as s:
        prune(s); est = simulate(s, hist); summ = summary(s)
        rows = [(i, e) for i, e in s["jobs"].items() if e["state"] == "running"] + order(s)
    print(f"{'ID':22} {'state':8} {'kind':5} {'prio':6} {'worktree':14} {'label':20} {'waited':>7} {'running':>7} {'est.':>12}")
    for i, e in rows:
        if e["state"] == "running":
            ran = now() - e["start"]; waited = e["start"] - e["submit"]
            st = rjson(f"{jdir(i)}/status.json", {})
            et = "lock wait" if st.get("lock") == "waiting" else f"~{max(0, est_run(e['kind'], hist) - ran):.0f}s left"
            pg = progress(rjson(f"{jdir(i)}/job.json", {"kind": e["kind"], "cwd": e["cwd"], "id": i}), st).lstrip(", ")
            stl = st.get("stall")
            print(f"{i:22} {'stalled' if stl else 'running':8} {e['kind']:5} {e['prio']:6} {worktree(e['cwd'])[:14]:14} {e['label'][:20]:20} {waited:6.0f}s {ran:6.0f}s {et:>12}  {pg}")
            if stl: print(f"  ! stalled: {stl['reason']}; blocked: {stl['threads']}")
            sw = scratch_warn(rjson(f"{jdir(i)}/job.json", {"id": i, "cwd": e["cwd"]}))
            if sw: print(f"  ! {sw}")
        else:
            print(f"{i:22} {'queued':8} {e['kind']:5} {e['prio']:6} {worktree(e['cwd'])[:14]:14} {e['label'][:20]:20} {now() - e['submit']:6.0f}s {'-':>7} {'start ~' + str(int(max(0, est.get(i, now()) - now()))) + 's':>12}")
    print(summ)


def cmd_top(a):
    """Live view: queue, progress, output tails of running jobs, machine load; Ctrl-C quits (--once: one frame)."""
    once = "--once" in a
    while True:
        frame = subprocess.run([sys.executable, "-I", os.path.realpath(__file__), "ls"], capture_output=True, text=True).stdout
        with Locked() as s: run = [i for i, e in s["jobs"].items() if e["state"] == "running"]
        for i in run:
            try:
                with open(f"{jdir(i)}/out.log", errors="replace") as f: tail = f.readlines()[-3:]
            except OSError: tail = []
            frame += f"\n── {i}\n" + "".join("   " + l[:150].rstrip() + "\n" for l in tail)
        la = open("/proc/loadavg").read().split()[:3]
        mem = next((int(l.split()[1]) // 1024 for l in open("/proc/meminfo") if l.startswith("MemAvailable")), 0)
        frame += f"\nload {' '.join(la)} / {os.cpu_count()} threads, {mem} MB available  ({time.strftime('%H:%M:%S')})\n"
        sys.stdout.write(("" if once else "\x1b[H\x1b[2J") + frame); sys.stdout.flush()
        if once: return
        try: time.sleep(2)
        except KeyboardInterrupt: return


def parse_dur(v):
    m = re.fullmatch(r"(\d+(?:\.\d+)?)([smhd]?)", v)
    if not m: die(f"bad duration {v}")
    return float(m.group(1)) * {"": 1, "s": 1, "m": 60, "h": 3600, "d": 86400}[m.group(2)]


def cmd_stats(a):
    since, jid = 86400.0, None
    while a:
        o = a.pop(0)
        if o == "--since": since = parse_dur(a.pop(0))
        elif o == "--id": jid = resolve(a.pop(0))
        else: die(__doc__, 2)
    t1 = now(); t0 = t1 - since
    hist = [x for x in read_log() if x.get("submit", 0) >= t0 or (x.get("end") or 0) >= t0]
    if jid:
        x = [h for h in hist if h["id"] == jid]
        st = rjson(f"{jdir(jid)}/status.json", {}); job = rjson(f"{jdir(jid)}/job.json")
        print(job_line(jid, st, job)); return
    print(f"q stats since {time.strftime('%Y-%m-%d %H:%M', time.localtime(t0))} ({len(hist)} jobs; times in s)")
    print(f"{'kind':7} {'n':>4} {'ok':>4} {'wait p50':>8} {'p90':>6} {'max':>6} {'run p50':>8} {'p90':>6} {'>540s':>6} {'load p50':>8}")
    for k in ["check", "test", "emu", "full", "batch", "long", "legacy"]:
        h = [x for x in hist if x.get("kind") == k]
        if not h: continue
        w = [x["wait"] for x in h if x.get("wait") is not None]
        r = [x["run"] for x in h if x.get("run") is not None]
        f = lambda v: "-" if v is None else f"{v:.0f}"
        print(f"{k:7} {len(h):4} {sum(1 for x in h if x.get('rc') == 0):4} {f(pct(w, 50)):>8} {f(pct(w, 90)):>6} {f(max(w) if w else None):>6}"
              f" {f(pct(r, 50)):>8} {f(pct(r, 90)):>6} {sum(1 for v in w if v > WAIT_MAX):6}"
              f" {f(pct([x['load'] for x in h if x.get('load') is not None], 50)):>8}")
    qh = [x for x in hist if x.get("kind") != "legacy"]
    ev = []
    for x in qh:
        st = x.get("start") or x["end"]
        ev += [(x["submit"], 1), (st, -1)]
    ev.sort(); n = peak = 0
    for _, d in ev: n += d; peak = max(peak, n)
    lo = max(t0, min((x["submit"] for x in qh), default=t1))
    busy = sum(x["weight"] * max(0, x["end"] - max(x["start"], lo)) for x in qh if x.get("start"))
    span = max(1e-9, t1 - lo)
    print(f"peak queue length {peak}; capacity use {100 * busy / (CAP * span):.1f} % of {CAP} units over {span / 3600:.2f} h")
    print(summary())


def cmd_legacy(a):
    """q legacy: an exclusive hold of QLOCK, wait and run logged (kind=legacy)."""
    w = None
    if a[:1] == ["-w"] and len(a) > 1: w = float(a[1]); a = a[2:]
    a = a[1:] if a[:1] == ["--"] else a
    if not a: die("usage: tools/q legacy [-w SECS] -- CMD…", 2)
    os.makedirs(os.path.dirname(QLOCK), exist_ok=True)
    lk = open(QLOCK, "a"); t0 = now()
    while True:
        try: fcntl.flock(lk, fcntl.LOCK_EX | fcntl.LOCK_NB); break
        except BlockingIOError:
            if w is not None and now() - t0 >= w:
                append_log({"id": f"legacy-{int(t0)}-{os.getpid()}", "kind": "legacy", "label": " ".join(a)[:60],
                            "cwd": os.getcwd(), "worktree": worktree(os.getcwd()), "cmd": a, "status": "lock-timeout",
                            "rc": 1, "submit": t0, "start": None, "end": now(), "wait": round(now() - t0, 1), "run": None, "weight": 0})
                print(f"q legacy: lock not acquired in {w:.0f} s", file=sys.stderr); sys.exit(1)
            time.sleep(min(POLL, 1))
    t1 = now()
    rc = subprocess.call(a)
    append_log({"id": f"legacy-{int(t0)}-{os.getpid()}", "kind": "legacy", "label": " ".join(a)[:60], "cwd": os.getcwd(),
                "worktree": worktree(os.getcwd()), "cmd": a, "status": "done" if rc == 0 else "failed", "rc": rc,
                "submit": t0, "start": t1, "end": now(), "wait": round(t1 - t0, 1), "run": round(now() - t1, 1), "weight": 0})
    sys.exit(rc)


def die(m, rc=1):
    print(m, file=sys.stderr); sys.exit(rc)


# ---- selftest --------------------------------------------------------------------------------------------------

def cmd_selftest(a):
    tmp = tempfile.mkdtemp(prefix="qtest.", dir=os.environ.get("TMPDIR"))
    adb = f"{tmp}/fake_adb"
    with open(adb, "w") as f:
        f.write('#!/bin/sh\necho "$*" >> "$ADBLOG"\ncase "$*" in\n'
                "  devices) printf 'List of devices attached\\nemulator-5554\\tdevice\\nPHONE1\\tdevice\\n';;\n"
                '  *"settings get system user_rotation") echo 1;;\nesac\n')
    os.chmod(adb, 0o755)
    env = dict(os.environ, QDIR=f"{tmp}/queue", QLOCK=f"{tmp}/agents.lock", QPOLL="0.2", QSTALL="600", QKILL_GRACE="1",
               QADB=adb, ADBLOG=f"{tmp}/adb.log", QSCRATCH_ROOT=f"{tmp}/scratch", QSCRATCH_GB="0.001")
    env.pop("QEMU_HYGIENE", None); env.pop("J", None)
    me = os.path.abspath(__file__)
    fails = []

    def q(*args, check=True):
        r = subprocess.run([sys.executable, "-I", me, *args], env=env, capture_output=True, text=True)
        return r

    def sub(kind, *cmd, prio="normal", label="", C=None, extra=()):
        r = q("submit", "--kind", kind, "--prio", prio, "--label", label, *(["-C", C] if C else []), *extra, "--", *cmd)
        if r.returncode: fails.append(f"submit failed: {r.stderr}"); return None
        return r.stdout.strip()

    def st(i): return rjson(f"{tmp}/queue/jobs/{i}/status.json", {})
    def ok(c, m): print(("ok   " if c else "FAIL ") + m); (None if c else fails.append(m))

    def until(f, t=20):
        e = now() + t
        while now() < e:
            if f(): return True
            time.sleep(0.1)
        return False

    def done(*ids): return until(lambda: all(st(i).get("state") not in ("queued", "running") for i in ids), 60)

    try:
        # capacity: 3 checks of weight 2 into 4 units → 2 run, 1 waits
        c = [sub("check", "sleep", "2", label=f"cap{n}") for n in range(3)]
        until(lambda: sum(st(i).get("state") == "running" for i in c) == 2, 10); time.sleep(0.6)
        ok([st(i).get("state") for i in c].count("running") == 2 and st(c[2]).get("state") == "queued",
           "capacity: 2 of 3 weight-2 jobs run, third queued")
        done(*c)
        ok(st(c[2])["start"] >= min(st(c[0])["end"], st(c[1])["end"]) - 0.01, "capacity: third starts after one ends")
        # J export: weight 2 → J=8, full → 16, caller's J kept
        j1 = sub("check", "sh", "-c", 'echo "J=$J"'); j2 = sub("full", "sh", "-c", 'echo "J=$J"')
        env["J"] = "3"; j3 = sub("test", "sh", "-c", 'echo "J=$J"'); del env["J"]
        done(j1, j2, j3)
        rd = lambda i: open(f"{tmp}/queue/jobs/{i}/out.log").read().strip()
        ok((rd(j1), rd(j2), rd(j3)) == ("J=8", "J=16", "J=3"), f"J export: {rd(j1)} {rd(j2)} {rd(j3)}")
        # priority + strict order: full holds all 4; normal A, normal full B, high H → H, then A, then B
        f = sub("full", "sleep", "1.5", label="blocker")
        until(lambda: st(f).get("state") == "running")
        A = sub("test", "sleep", "0.5", label="A"); B = sub("full", "sleep", "0.5", label="B")
        T = sub("test", "sleep", "0.5", label="T-behind-B")
        H = sub("test", "sleep", "0.5", prio="high", label="H")
        done(f, A, B, T, H)
        s = {k: st(v)["start"] for k, v in dict(A=A, B=B, T=T, H=H).items()}
        ok(s["H"] <= s["A"] <= s["B"] <= s["T"], f"priority high first, then FIFO, no overtaking of waiting full B")
        # emulator slots: two emu jobs overlap on their own AVDs, a third waits for a slot
        sf = f"{tmp}/slot2"
        e1 = sub("emu", "sleep", "1"); e2 = sub("emu", "sh", "-c", f'echo "$QEMU_SLOT $QEMU_PORT $ANDROID_AVD_HOME" > {sf}; sleep 2'); e3 = sub("emu", "sleep", "0.2")
        done(e1, e2, e3)
        ok(st(e2)["start"] < st(e1)["end"] and st(e3)["start"] >= st(e1)["end"] - 0.01, "emulator slots: two overlap, the third waits")
        ok(open(sf).read().split()[:2] == ["1", "5620"] and open(sf).read().strip().endswith("avd-slot1"), "emulator slots: the second job gets slot 1's port and AVD")
        # wait exit codes
        w0 = sub("test", "true"); w1 = sub("test", "sh", "-c", "echo boom; exit 3"); w2 = sub("test", "sleep", "30")
        r0 = q("wait", w0, "--max", "20"); r1 = q("wait", w1, "--max", "20"); r2 = q("wait", w2, "--max", "1")
        ok(r0.returncode == 0 and "job " in r0.stdout, "wait: exit 0 for a successful job")
        ok(r1.returncode == 3 and "boom" in r1.stdout and "out.log" in r1.stdout, "wait: job exit code 3 + tail + log path")
        ok(r2.returncode == EXIT_PENDING and "running for" in r2.stdout and "q: " in r2.stdout, "wait: 75 while running + q summary")
        print("     " + r2.stdout.strip().replace("\n", "\n     "))
        # cancel: running w2; and a queued job behind a full blocker
        q("cancel", w2); rw = q("wait", w2, "--max", "10")
        ok(rw.returncode == EXIT_CANCELLED and st(w2)["state"] == "cancelled" and st(w2)["end"] - st(w2)["start"] < 15,
           f"cancel: running job stopped ({st(w2)})")
        f = sub("full", "sleep", "8"); until(lambda: st(f).get("state") == "running")
        cq = sub("test", "sleep", "1")
        rq = q("wait", cq, "--max", "0.5")
        ok(rq.returncode == EXIT_PENDING and "still queued: position 1 of 1" in rq.stdout, "wait: 75 while queued + position")
        print("     " + rq.stdout.strip().replace("\n", "\n     "))
        q("cancel", cq)
        rc = q("wait", cq, "--max", "10")
        ok(rc.returncode == EXIT_CANCELLED and st(cq)["state"] == "cancelled", "cancel: queued job cancelled")
        # removed worktree: queued job whose cwd disappears
        wt = tempfile.mkdtemp(dir=tmp)
        rmq = sub("test", "sleep", "1", C=wt)
        os.rmdir(wt); done(rmq); q("cancel", f); done(f)
        ok(st(rmq).get("state") == "cancelled" and st(rmq).get("reason") == "worktree removed", "removed worktree: job cancelled")
        # legacy interplay: exclusive holder blocks q jobs; q jobs share; running q jobs block q legacy
        hold = subprocess.Popen(["flock", env["QLOCK"], "sleep", "1.5"])
        time.sleep(0.3)
        l1 = sub("test", "sleep", "1"); l2 = sub("test", "sleep", "1")
        until(lambda: st(l1).get("lock") == "waiting", 5)
        ok(st(l1).get("lock") == "waiting", "legacy: q job waits while a legacy exclusive holder runs")
        hold.wait()
        until(lambda: st(l1).get("lock") == "shared" and st(l2).get("lock") == "shared", 5)
        ok(st(l1).get("lock") == "shared" and st(l2).get("lock") == "shared", "legacy: two q jobs hold the lock shared at once")
        ql = subprocess.run([me, "legacy", "-w", "0.3", "--", "true"], env=env, capture_output=True, text=True)
        ok(ql.returncode == 1, "legacy: qlock -w times out while q jobs run")
        done(l1, l2)
        ql = subprocess.run([me, "legacy", "--", "sh", "-c", "exit 4"], env=env, capture_output=True, text=True)
        ok(ql.returncode == 4, "legacy: qlock runs CMD and returns its exit code")
        # log.jsonl + stats
        log = read_log_at(f"{tmp}/queue/log.jsonl")
        keys = {"id", "kind", "label", "submit", "start", "end", "wait", "run", "rc", "status", "worktree", "cmd"}
        ok(len([x for x in log if x["kind"] != "legacy"]) == 22 and all(keys <= set(x) for x in log)
           and len([x for x in log if x["kind"] == "legacy"]) == 2, f"log.jsonl: {len(log)} lines with all fields")
        rs = q("stats", "--since", "1h")
        ok(rs.returncode == 0 and all(k in rs.stdout for k in ("check", "test", "emu", "full", "legacy", "peak queue length", "capacity use")),
           "stats: all kinds, peak queue, capacity use")
        print("     " + rs.stdout.strip().replace("\n", "\n     "))
        # stall watchdog: a silent sleeper (ignoring TERM) stalls; a busy and a chatty job do not; kill stops the tree
        sj = sub("test", "sh", "-c", "trap '' TERM; sleep 60; echo late", label="sleeper", extra=("--stall", "2"))
        bz = sub("test", sys.executable, "-c", "import time\ne = time.time() + 5\nwhile time.time() < e: pass",
                 label="busy", extra=("--stall", "2"))
        ch = sub("test", "sh", "-c", "for i in 1 2 3 4 5 6 7 8 9 10 11 12; do echo $i; sleep 0.5; done", label="chatty",
                 extra=("--stall", "2"))
        until(lambda: st(sj).get("stall"), 15)
        rl = q("ls")
        ok(re.search(rf"{sj}\s+stalled", rl.stdout) and "sleep" in rl.stdout, "stall: ls marks the sleeper stalled with its threads")
        print("     " + rl.stdout.strip().replace("\n", "\n     "))
        rs = q("wait", sj, "--max", "5")
        ok(rs.returncode == EXIT_STALLED and "STALLED" in rs.stdout and "sleep" in rs.stdout, f"stall: wait exits {EXIT_STALLED}")
        print("     " + rs.stdout.strip().replace("\n", "\n     "))
        done(bz, ch)
        ok(st(bz).get("rc") == 0 and st(ch).get("rc") == 0 and not st(bz).get("stalled") and not st(ch).get("stalled"),
           "stall: busy (no output) and chatty (low CPU) jobs never stalled")
        ok(st(sj).get("state") == "running", "stall: a stalled job is not killed")
        t0 = now(); rk = q("kill", sj)
        act = subprocess.run(["systemctl", "--user", "is-active", f"qt-{sj}"], capture_output=True, text=True).stdout.strip()
        ok(rk.returncode == 0 and st(sj).get("state") == "cancelled" and now() - t0 < 15 and act != "active",
           f"kill: TERM-ignoring tree gone in {now() - t0:.1f} s ({rk.stdout.strip()}, unit {act})")
        # emulator hygiene with a fake adb: snapshot + settings around emu jobs, also when cancelled; never a phone
        alog = lambda: open(env["ADBLOG"]).read() if os.path.exists(env["ADBLOG"]) else ""
        ok(alog().count("emu avd snapshot save q_pre") == 2 and alog().count("emu avd snapshot load q_pre") == 2
           and "settings put system user_rotation 1" in alog() and "wm size reset" in alog() and "wm density reset" in alog(),
           "emu: q_pre saved and restored around both emu jobs, rotation / size / density reset")
        ec = sub("emu", "sleep", "30", label="emu-cancel")
        until(lambda: "saved" in (st(ec).get("emu") or ""), 15); q("cancel", ec); done(ec)
        ok(alog().count("emu avd snapshot load q_pre") == 3 and st(ec).get("state") == "cancelled", "emu: restored after a cancel")
        ok("PHONE1" not in alog(), "emu: a non-emulator device is never touched")
        # scratch quota: a job whose scratch dir passes QSCRATCH_GB gets a warning line in ls
        wt = tempfile.mkdtemp(dir=tmp)
        sd = f"{env['QSCRATCH_ROOT']}/" + re.sub(r"[^A-Za-z0-9]", "-", os.path.realpath(wt)) + "/sess"
        os.makedirs(sd)
        with open(f"{sd}/big", "wb") as f: f.write(os.urandom(2 << 20))
        sq = sub("test", "sleep", "3", C=wt); until(lambda: st(sq).get("lock") == "shared")
        rl = q("ls"); ok("job scratch" in rl.stdout and "GB" in rl.stdout, "scratch: ls warns about a job scratch dir over quota")
        done(sq)
        rl = q("ls"); ok(rl.returncode == 0 and "0 queued" in rl.stdout, "ls: empty queue at the end")
    finally:
        with open(f"{tmp}/queue/state.json") as f: left = json.load(f)["jobs"]
        for i in left: subprocess.run(["systemctl", "--user", "stop", f"qt-{i}"], capture_output=True)
        shutil.rmtree(tmp, ignore_errors=True)
    print("selftest " + ("FAILED: " + "; ".join(fails) if fails else "ok"))
    sys.exit(1 if fails else 0)


def read_log_at(p):
    with open(p) as f: return [json.loads(l) for l in f]


def main():
    a = sys.argv[1:]
    if not a: print(summary()); return
    c = a.pop(0)
    f = {"submit": cmd_submit, "_run": cmd_run, "wait": cmd_wait, "ls": cmd_ls, "stats": cmd_stats, "cancel": cmd_cancel, "kill": cmd_kill,
         "legacy": cmd_legacy, "selftest": cmd_selftest, "top": cmd_top}.get(c)
    if not f: die(__doc__, 2)
    f(a)


if __name__ == "__main__":
    main()
