path: root/wf_res.py

"""wf res — shared memory/CPU ledger for agent jobs (files, /proc, systemd, printing).
Pure logic: wflib/res.py. Spec: docs/resource-ledger.md.

  run --mem 10G [--cpus N] --for 40m --title T [--queue|--force] [--by NAME] -- CMD…   start a job or get `busy` (exit 3)
  status [ID] [--json] · wait ID · release ID [--stop] · note --mem 3G --for 20m [--force] [--by NAME] TITLE
  game on [--for 4h] | off · clean [--yes] · timer on|off · tick · shell-init
"""
from __future__ import annotations

import re
import argparse
import contextlib
import datetime as dt
import fcntl
import json
import os
import shlex
import shutil
import stat
import subprocess
import sys
import time
import tomllib
from dataclasses import dataclass
from pathlib import Path
from typing import Callable

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))

from wflib import res  # noqa: E402

LEDGER, LOCK = "resources.json", "resources.lock"
HISTORY = "resources-history.jsonl"      # finished runs pruned from the ledger: reservation sizes (wf res hist)
ACTIVE = {"active", "activating", "deactivating", "reloading", "refreshing"}


@dataclass
class Env:
    """Everything wf res touches outside its own state; tests swap the callables."""
    state: Path
    config: Path
    units: Path
    scratch: Path
    run: Callable[[list[str]], subprocess.CompletedProcess]
    meminfo: Callable[[], str]
    nproc: int
    now: Callable[[], dt.datetime]
    pid_alive: Callable[[int], bool]
    owner: Callable[[], int]
    root: Callable[[], Path]
    cwd: Callable[[], str]
    procs: Callable[[], list[tuple[int, int, str, str]]]   # (pid, ppid, comm, cgroup)
    sleep: Callable[[float], None]
    tmp_used_gb: Callable[[], float]
    stdin: Callable[[], str] = lambda: sys.stdin.read()
    live_sessions: Callable[[], set[str]] = lambda: _live_sessions()
    tmp_dirs: tuple = (Path("/tmp"), Path("/var/tmp"))      # swept for litter (res.litter_victims)
    cgread: Callable[[str, str], str] = lambda cg, name: _read(f"/sys/fs/cgroup{cg}/{name}")   # ControlGroup, file
    environ: Callable[[], dict] = lambda: dict(os.environ)


# ---------------------------------------------------------------- real machine

def _read(path: str) -> str:
    try:
        return Path(path).read_text()
    except OSError:
        return ""


def _live_sessions() -> set[str]:
    """Session ids of running Claude processes (Claude Code's ~/.claude/sessions/<pid>.json); unreadable → empty."""
    out = set()
    for f in (Path.home() / ".claude" / "sessions").glob("*.json"):
        try:
            stat = Path(f"/proc/{f.stem}/stat").read_text()
            sid = res.live_session_id(f.read_text(), stat.rsplit(")", 1)[1].split()[19])
        except (OSError, IndexError):
            continue
        if sid:
            out.add(sid)
    return out


def _run(argv):
    return subprocess.run(argv, capture_output=True, text=True)


def _xdg(var: str, default: str) -> Path:
    return Path(os.environ.get(var) or Path.home() / default)


def _argv0(d: Path) -> str:
    try:
        return (d / "cmdline").read_bytes().split(b"\0", 1)[0].decode(errors="replace")
    except OSError:
        return ""


def _comm(pid: int) -> str:
    """Process name; `claude` for any Claude Code process (res.proc_name)."""
    d = Path(f"/proc/{pid}")
    try:
        comm = (d / "comm").read_text().strip()
    except OSError:
        return ""
    return res.proc_name(comm, _argv0(d))


def _ppid(pid: int) -> int:
    return int(Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()[1])


def _owner() -> int:
    """Nearest ancestor named `claude` (the session outlives its shells), else wf's parent."""
    parent = p = os.getppid()
    while p > 1:
        if _comm(p) == "claude":
            return p
        try:
            p = _ppid(p)
        except (OSError, ValueError, IndexError):
            break
    return parent


def _root() -> Path:
    r = _run(["git", "rev-parse", "--show-toplevel"])
    return Path(r.stdout.strip()) if r.returncode == 0 and r.stdout.strip() else Path.cwd()


def _procs() -> list[tuple[int, int, str, str]]:
    out = []
    for d in Path("/proc").iterdir():
        if not d.name.isdigit():
            continue
        try:
            stat = (d / "stat").read_text()
            comm = stat[stat.index("(") + 1:stat.rindex(")")]
            ppid = int(stat.rsplit(")", 1)[1].split()[1])
            cgroup = (d / "cgroup").read_text().strip().split(":", 2)[2]
        except (OSError, ValueError, IndexError):
            continue
        out.append((int(d.name), ppid, res.proc_name(comm, _argv0(d)), cgroup))
    return out


def _tmp_used_gb() -> float:
    st = os.statvfs("/tmp")
    return (st.f_blocks - st.f_bfree) * st.f_frsize / res.GB


def real_env() -> Env:
    runtime = Path(os.environ.get("XDG_RUNTIME_DIR") or f"/run/user/{os.getuid()}")
    if not (runtime / "systemd").is_dir():
        raise res.ResError("wf res needs a systemd user session")
    return Env(state=_xdg("XDG_STATE_HOME", ".local/state") / "wf",
               config=_xdg("XDG_CONFIG_HOME", ".config") / "wf" / "resources.toml",
               units=_xdg("XDG_CONFIG_HOME", ".config") / "systemd" / "user",
               scratch=Path(f"/tmp/claude-{os.getuid()}"), run=_run,
               meminfo=lambda: Path("/proc/meminfo").read_text(), nproc=os.cpu_count() or 1,
               now=lambda: dt.datetime.now().astimezone(), pid_alive=lambda p: Path(f"/proc/{p}").exists(),
               owner=_owner, root=_root, cwd=os.getcwd, procs=_procs, sleep=time.sleep,
               tmp_used_gb=_tmp_used_gb)


# ---------------------------------------------------------------- ledger file

def warn(msg: str) -> None:
    print(f"wf: {msg}", file=sys.stderr)


def write_atomic(path: Path, text: str) -> None:
    tmp = path.with_name(path.name + ".tmp")
    tmp.write_text(text)
    os.replace(tmp, path)


@contextlib.contextmanager
def locked(env: Env):
    """Exclusive flock; yields the ledger; writes it back atomically when the block ends without error."""
    env.state.mkdir(parents=True, exist_ok=True)
    with open(env.state / LOCK, "w") as lock:
        fcntl.flock(lock, fcntl.LOCK_EX)
        path = env.state / LEDGER
        text = path.read_text() if path.exists() else ""
        try:
            led = res.loads(text)
        except res.ResError as e:
            bad = path.with_name(f"{LEDGER}.bad-{env.now():%Y%m%d-%H%M%S}")
            path.rename(bad)
            warn(f"{e}; moved to {bad.name}, starting empty")
            led = res.Ledger(next=res.salvage_next(text))
        yield led
        write_atomic(path, res.dumps(led))


# ---------------------------------------------------------------- systemd

def systemctl(env: Env, args: list[str]) -> str:
    r = env.run(["systemctl", "--user", *args])
    if r.returncode != 0:
        first = (r.stderr or r.stdout).strip().splitlines()
        raise res.ResError(f"systemctl {args[0]}: " + (first[0] if first else f"exit {r.returncode}"))
    return r.stdout


def ensure_slice(env: Env) -> None:
    f = env.units / res.SLICE
    if not f.exists():
        env.units.mkdir(parents=True, exist_ok=True)
        f.write_text(res.slice_unit_text())
        systemctl(env, ["daemon-reload"])


def _result(env: Env, e: res.Entry) -> tuple:
    """(rc, peak_gb, why); no .rc (wrapper killed with the job) → why from the journal."""
    def read(name):
        try:
            return (env.state / "logs" / name).read_text().strip()
        except OSError:
            return ""
    rc, peak = read(f"{e.id}.rc"), read(f"{e.id}.peak")
    rc, peak = int(rc) if rc.lstrip("-").isdigit() else None, int(peak) / res.GB if peak.isdigit() else None
    if rc is not None:
        return rc, peak, ""
    argv = ["journalctl", "--user", "-u", e.unit, "-o", "cat", "--no-pager"]
    if e.started:
        argv += ["--since", f"@{int(e.started.timestamp())}"]
    why, jpeak = res.journal_reason(env.run(argv).stdout or "", e.mem_gb)
    return rc, peak if peak is not None else jpeak, why or f"no exit code; see journalctl --user -u {e.unit}"


def bare_units(env: Env) -> list:
    names = [l.split()[0] for l in systemctl(env, ["list-units", "--type=service", "--state=running", "--no-legend",
                                                   "--plain"]).splitlines() if l.strip()]
    if not names:
        return []
    return res.bare_units(res.show_units(systemctl(env, ["show", "-p", "Id,Transient,Slice,MemoryCurrent", *names])))


def session_scopes(env: Env) -> dict[str, dict[str, str]]:
    """Running scopes in agents.slice (Claude sessions) → show props."""
    names = [l.split()[0] for l in systemctl(env, ["list-units", "--type=scope", "--state=running", "--no-legend",
                                                   "--plain"]).splitlines() if l.strip()]
    if not names:
        return {}
    blocks = res.show_units(systemctl(env, ["show", "-p", "Id,Slice,MemoryHigh,MemoryCurrent,ControlGroup", *names]))
    return {n: b for n, b in blocks.items() if b.get("Slice") == res.SLICE}


def cap_sessions(env: Env, cfg: res.Config) -> None:
    for unit, value in res.session_caps(session_scopes(env), cfg.session_mem_gb):
        systemctl(env, ["set-property", "--runtime", unit, f"MemoryHigh={value}"])


def gather(env: Env, led: res.Ledger) -> res.Facts:
    total, available = res.meminfo(env.meminfo())
    names = sorted({e.unit for e in led.entries if e.state == "running"})
    blocks = res.show_units(systemctl(env, ["show", "-p", "Id,ActiveState,MemoryCurrent,ControlGroup",
                                            res.SLICE, *names]))
    units = {}
    for n in names:
        b = blocks.get(n, {})
        active, cg = b.get("ActiveState") in ACTIVE, b.get("ControlGroup", "")
        units[n] = res.Unit(active, res.gb_or_none(b.get("MemoryCurrent", "")),
                            res.psi_some_avg60(env.cgread(cg, "memory.pressure")) if active and cg else None)
    scg = blocks.get(res.SLICE, {}).get("ControlGroup", "")
    return res.Facts(
        now=env.now(), total_gb=total, available_gb=available, nproc=env.nproc, units=units,
        live={e.owner for e in led.entries if e.state == "note" and env.pid_alive(e.owner)},
        results={e.id: _result(env, e) for e in led.entries if e.state == "running" and not units[e.unit].active},
        slice_gb=res.gb_or_none(blocks.get(res.SLICE, {}).get("MemoryCurrent", "")) or 0.0,
        slice_cache_gb=res.cache_gb(env.cgread(scg, "memory.stat")) if scg else 0.0)


def start(env: Env, e: res.Entry, now: dt.datetime) -> str | None:
    """Launch e as a unit; on success e is running. Returns an error line or None."""
    logs = env.state / "logs"
    logs.mkdir(parents=True, exist_ok=True)
    e.unit, e.log = f"wf-{e.id}.service", str(logs / f"{e.id}.log")
    for stale in (f"{e.id}.rc", f"{e.id}.peak"):    # a reused id must not inherit an old exit code
        (logs / stale).unlink(missing_ok=True)
    ensure_slice(env)
    r = env.run(res.run_argv(e, str(logs)))
    if r.returncode != 0:
        first = (r.stderr or r.stdout).strip().splitlines()
        return "systemd-run: " + (first[0] if first else f"exit {r.returncode}")
    e.state, e.started = "running", now
    e.expires = now + dt.timedelta(minutes=2 * e.est_min)
    return None


def housekeep(env: Env, cfg: res.Config, led: res.Ledger, facts: res.Facts) -> list[str]:
    """Every command, under the lock: prune, start queued work."""
    done = [e for e in led.entries if e.state == "done"]
    lines = res.prune(led, facts)
    kept = {id(e) for e in led.entries}
    save_history(env, [r for r in (res.history_record(e) for e in done if id(e) not in kept) if r])
    for e in res.to_start(cfg, led, facts):
        err = start(env, e, facts.now)
        if err:
            e.state, e.ended, e.why = "done", facts.now, f"start failed: {err}"
            lines.append(f"{e.id} {e.why}")
        else:
            lines.append(f"{e.id} started (queued)")
    if res.gaming(led, facts.now) or "game off (expired)" in lines:
        apply_slice(env, cfg, led, facts)
    if not led.last_clean or facts.now - led.last_clean >= res.CLEAN_EVERY:
        lines.extend(clean_auto(env, cfg, led, facts.now))
    return lines


def read_history(env: Env) -> list[dict]:
    path = env.state / HISTORY
    out = []
    for line in (path.read_text().splitlines() if path.exists() else []):
        try:
            r = json.loads(line)
        except ValueError:
            continue
        if isinstance(r, dict):
            out.append(r)
    return out


def save_history(env: Env, recs: list[dict]) -> None:
    """Append runs leaving the ledger; keep the newest res.HIST_KEEP (call under the lock)."""
    if not recs:
        return
    path = env.state / HISTORY
    old = path.read_text().splitlines() if path.exists() else []
    lines = old + [json.dumps(r) for r in recs]
    if len(lines) > res.HIST_KEEP:
        write_atomic(path, "".join(x + "\n" for x in lines[-res.HIST_KEEP:]))
    else:
        with open(path, "a") as f:
            f.write("".join(json.dumps(r) + "\n" for r in recs))


def suggestion(env: Env, led: res.Ledger, title: str) -> tuple[float, int, int] | None:
    """History's (mem GB, minutes, n) for this project's kind of title (res.suggest), else None."""
    return res.suggest(res.all_runs(read_history(env), led), main_tree(env).name, res.title_kind(title))


def hint(env: Env, led: res.Ledger, title: str, mem: float, est: int) -> None:
    line = res.hint_line(mem, est, suggestion(env, led, title))
    if line:
        print(line, file=sys.stderr)


def session(env: Env, cfg: res.Config, led: res.Ledger) -> res.Facts:
    facts = gather(env, led)
    housekeep(env, cfg, led, facts)
    return facts


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

def _git(env: Env, *args) -> str:
    r = env.run(["git", "-C", env.cwd(), *args])
    return r.stdout.strip() if r.returncode == 0 else ""


def main_tree(env: Env) -> Path:
    """Project's main tree: git common dir's parent (same for its lane worktrees), else env.root()."""
    common = _git(env, "rev-parse", "--path-format=absolute", "--git-common-dir")
    return Path(common).parent if common and Path(common).name == ".git" else env.root()


def _session_records(env: Env) -> list[dict]:
    """wf lane session records (.wf/sessions/*.json) of this project: main tree + root."""
    out = []
    for top in dict.fromkeys([main_tree(env), env.root()]):
        for f in sorted((top / ".wf" / "sessions").glob("*.json")):
            try:
                out.append(json.loads(f.read_text()))
            except (OSError, ValueError):
                pass
    return out


def owner_by(env: Env, name: str = "") -> dict:
    return res.owner_by(env.environ(), _git(env, "branch", "--show-current"), _session_records(env), name)


def _new_entry(env, led, facts, args, mem, est, state) -> res.Entry:
    return res.Entry(id=led.new_id(), project=main_tree(env).name, owner=env.owner(), title=args.title, mem_gb=mem,
                     cpus=args.cpus, est_min=est, state=state, cwd=env.cwd(), queued=facts.now,
                     by=owner_by(env, getattr(args, "by", "") or ""))


def force_busy(cfg, led, facts, mem, cpus, room) -> str:
    """--force refused: not even the really free memory (beyond the user reserve) holds it."""
    free = f"{res.fmt_gb(max(0.0, room[0]))}" if mem > room[0] + res.EPS else f"{max(0, room[1])} cpus"
    return f"busy even with --force: only {free} really free beyond the reserve; retry later or work on something else"


def cmd_run(env, cfg, args) -> int:
    mem, est = res.parse_size(args.mem), res.parse_duration(args.for_)
    cmd = args.cmd[1:] if args.cmd[:1] == ["--"] else args.cmd
    if not cmd:
        raise res.ResError("no command after --")
    fail = busy = out = None
    shown = env.run(["systemctl", "--user", "show-environment"])
    caller_env = res.env_diff(env.environ(), res.parse_show_environment(shown.stdout if shown.returncode == 0 else ""))
    lock = res.lock_for(args.title, getattr(args, "lock", ""), str(main_tree(env)))
    with locked(env) as led:
        facts = session(env, cfg, led)
        hint(env, led, args.title, mem, est)
        fail = res.never_fits(cfg, facts, mem, args.cpus)
        holder = res.lock_holder(led, lock)
        if not fail and holder and not args.queue:
            busy = res.lock_busy_line(holder, facts)
        elif not fail and holder:
            e = _new_entry(env, led, facts, args, mem, est, "queued")
            e.cmd, e.env, e.lock = cmd, caller_env, lock
            led.entries.append(e)
            when = res.queue_estimate(cfg, led, facts, e)
            out = (f"{e.id} queued, position {res.position(led, e)} (lock held by {holder.id}); est. start "
                   + (f"~{res.hhmm(when)}" if when else "unknown") + f"; cancel: wf res release {e.id}")
        elif not fail:
            q_gb, q_cpus = res.queued_total(led)
            room = res.force_room(cfg, led, facts) if args.force else res.budget(cfg, led, facts)
            if args.force:
                q_gb, q_cpus = 0.0, 0
            if res.fits(mem + q_gb, args.cpus + q_cpus, room):
                e = _new_entry(env, led, facts, args, mem, est, "queued")
                e.cmd, e.env, e.lock = cmd, caller_env, lock
                fail = start(env, e, facts.now)
                if not fail:
                    led.entries.append(e)
                    out = f"{e.id} started; log {e.log}; ETA ~{res.hhmm(e.eta())} (estimate: not killed when over)"
            elif args.queue:
                e = _new_entry(env, led, facts, args, mem, est, "queued")
                e.cmd, e.env, e.lock = cmd, caller_env, lock
                led.entries.append(e)
                when = res.queue_estimate(cfg, led, facts, e)
                out = (f"{e.id} queued, position {res.position(led, e)}; est. start "
                       + (f"~{res.hhmm(when)}" if when else "unknown") + f"; cancel: wf res release {e.id}")
            elif args.force:
                busy = force_busy(cfg, led, facts, mem, args.cpus, room)
            else:
                busy = res.busy_line(cfg, led, facts, mem + q_gb, args.cpus + q_cpus,
                                     res.force_hint(cfg, led, facts, mem, args.cpus))
    if fail:
        raise res.ResError(fail)
    if busy:
        raise res.Busy(busy)
    print(out)
    return 0


def cmd_status(env, cfg, args) -> int:
    with locked(env) as led:
        facts = session(env, cfg, led)
        if args.json:
            text = res.status_json(cfg, led, facts)
        elif args.id:
            text = "\n".join(res.detail_lines(led.get(args.id)))
        else:
            outside = len(res.adopt_groups(env.procs()))
            scopes = session_scopes(env)
            stall = {n: res.psi_some_avg60(env.cgread(b.get("ControlGroup", ""), "memory.pressure")) or 0.0
                     for n, b in scopes.items() if b.get("ControlGroup")}
            text = "\n".join(res.status_lines(cfg, led, facts) + res.throttle_lines(led, facts)
                             + res.session_cap_lines(scopes, stall, cfg.session_mem_gb)
                             + res.warning_lines(facts.total_gb, env.tmp_used_gb(), outside, bare_units(env)))
    print(text)
    return 0


def cmd_hist(env, cfg, args) -> int:
    with locked(env) as led:
        runs = res.all_runs(read_history(env), led)
    print("\n".join(res.hist_lines(runs, args.project)))
    return 0


def cmd_release(env, cfg, args) -> int:
    fail = out = None
    with locked(env) as led:
        facts = session(env, cfg, led)
        e = led.get(args.id)
        if e.state == "running" and not args.stop:
            fail = f"{e.id} is running; --stop to kill it"
        elif e.state == "running":
            systemctl(env, ["stop", e.unit])
            e.rc, e.peak_gb, _ = _result(env, e)
            e.state, e.ended, e.why = "done", facts.now, "released"
            out = f"{e.id} stopped"
        else:
            led.entries.remove(e)
            out = f"{e.id} released"
    if fail:
        raise res.ResError(fail)
    print(out)
    return 0


POLL_SECONDS = 15


def cmd_wait(env, cfg, args) -> int:
    limit = res.parse_duration(args.timeout)
    deadline = env.now() + dt.timedelta(minutes=limit)
    warned = False
    while True:
        with locked(env) as led:
            facts = session(env, cfg, led)
            e = led.get(args.id)
            line, state = (res.done_line(e) if e.state == "done" else None), e.state
            hot = res.throttle_lines(res.Ledger(entries=[e]), facts)
        if hot and not warned:      # once per wait
            warn(hot[0])
            warned = True
        if line:
            print(line)
            return 0
        if env.now() >= deadline:
            raise res.ResError(f"{args.id} not done after {res.fmt_dur(limit)} ({state})")
        env.sleep(POLL_SECONDS)
def cmd_note(env, cfg, args) -> int:
    mem, est = res.parse_size(args.mem), res.parse_duration(args.for_)
    fail = busy = out = None
    shown = env.run(["systemctl", "--user", "show-environment"])
    caller_env = res.env_diff(env.environ(), res.parse_show_environment(shown.stdout if shown.returncode == 0 else ""))
    with locked(env) as led:
        facts = session(env, cfg, led)
        hint(env, led, args.title, mem, est)
        fail = res.never_fits(cfg, facts, mem, args.cpus)
        q_gb, q_cpus = (0.0, 0) if args.force else res.queued_total(led)
        room = res.force_room(cfg, led, facts) if args.force else res.budget(cfg, led, facts)
        if not fail and res.fits(mem + q_gb, args.cpus + q_cpus, room):
            e = _new_entry(env, led, facts, args, mem, est, "note")
            e.started, e.expires = facts.now, facts.now + dt.timedelta(minutes=est)
            led.entries.append(e)
            out = f"{e.id} noted {res.fmt_gb(mem)} until ~{res.hhmm(e.expires)} (then freed; nothing is killed)"
        elif not fail and args.force:
            busy = force_busy(cfg, led, facts, mem, args.cpus, room)
        elif not fail:
            busy = res.busy_line(cfg, led, facts, mem + q_gb, args.cpus + q_cpus,
                                 res.force_hint(cfg, led, facts, mem, args.cpus))
    if fail:
        raise res.ResError(fail)
    if busy:
        raise res.Busy(busy)
    print(out)
    return 0
def apply_slice(env: Env, cfg: res.Config, led: res.Ledger, facts: res.Facts) -> None:
    ensure_slice(env)
    props = res.slice_props(cfg, led, facts)
    systemctl(env, ["set-property", "--runtime", res.SLICE, *(f"{k}={v}" for k, v in props.items())])


def cmd_game(env, cfg, args) -> int:
    with locked(env) as led:
        facts = session(env, cfg, led)
        if args.state == "on":
            minutes = res.parse_duration(args.for_) if args.for_ else int(cfg.game_hours * 60)
            led.game_until = max(led.game_until or facts.now, facts.now + dt.timedelta(minutes=minutes))
            lines = res.game_on_lines(cfg, led, facts, minutes)
        else:
            led.game_until = None
            lines = ["game off"]
        apply_slice(env, cfg, led, facts)
    print("\n".join(lines))
    return 0
def _lstat(path) -> os.stat_result | None:
    """None when the path vanished meanwhile (scratch is shared with running sessions)."""
    try:
        return os.lstat(path)
    except OSError:
        return None


def _newest(p: Path) -> float:
    st = _lstat(p)
    newest = st.st_mtime if st else 0.0
    if p.is_dir() and not p.is_symlink():
        for dirpath, dirnames, filenames in os.walk(p, followlinks=False):
            for name in dirnames + filenames:
                if st := _lstat(os.path.join(dirpath, name)):
                    newest = max(newest, st.st_mtime)
    return newest


def _size(p: Path) -> int:
    if p.is_symlink() or not p.is_dir():
        st = _lstat(p)
        return st.st_size if st else 0
    return sum(st.st_size for d, _, fs in os.walk(p, followlinks=False) for f in fs
               if (st := _lstat(os.path.join(d, f))))


def _remove(p: Path) -> None:
    if p.is_dir() and not p.is_symlink():
        shutil.rmtree(p, ignore_errors=True)
    else:
        p.unlink(missing_ok=True)


def scan_scratch(env: Env) -> dict:
    tree = {}
    if not env.scratch.is_dir():
        return tree
    for top in env.scratch.iterdir():
        kids = {}
        if top.is_dir() and not top.is_symlink():
            kids = {str(c): _newest(c) for c in top.iterdir()}
        tree[str(top)] = (_newest(top), kids)
    return tree


def scan_litter(env: Env) -> list[tuple[str, str, float]]:
    uid, out = os.getuid(), []
    for d in env.tmp_dirs:
        try:
            names = os.listdir(d)
        except OSError:
            continue
        for n in names:
            p = Path(d) / n
            st = _lstat(p)
            if st is None or st.st_uid != uid:
                continue
            if stat.S_ISDIR(st.st_mode):
                try:
                    kind = "dir" if os.listdir(p) else "emptydir"
                except OSError:
                    continue
            else:
                kind = "file" if stat.S_ISREG(st.st_mode) and st.st_size else "other"
            out.append((str(p), kind, st.st_mtime))
    return out


def sweep_litter(env: Env, cfg: res.Config, now: dt.datetime) -> int:
    n = 0
    for path in res.litter_victims(scan_litter(env), now.timestamp(), cfg.scratch_hours, env.pid_alive):
        try:
            (os.rmdir if os.path.isdir(path) and not os.path.islink(path) else os.unlink)(path)
            n += 1
        except OSError:
            pass            # refilled or gone meanwhile
    return n


def clean_auto(env: Env, cfg: res.Config, led: res.Ledger, now: dt.datetime) -> list[str]:
    """Scratch untouched for scratch_hours (D4) and failed wf units; no confirmation needed."""
    lines = []
    for path in res.scratch_victims(scan_scratch(env), now.timestamp(), cfg.scratch_hours,
                                     env.live_sessions()):
        p = Path(path)
        size = _size(p)
        _remove(p)
        lines.append(f"freed {res.fmt_bytes(size)}: {p}")
    swept = sweep_litter(env, cfg, now)
    if swept:
        lines.append(f"swept {swept} temp-dir litter entries (empty dirs, dead .NET pipes)")
    env.run(["systemctl", "--user", "reset-failed", "wf-r-*"])
    led.last_clean = now
    return lines


def project_rules(env: Env) -> tuple[Path, list[tuple[str, int | None]]]:
    root = env.root()
    toml = root / "workflow.toml"
    if not toml.exists():
        return root, []
    try:
        patterns = tomllib.loads(toml.read_text()).get("cleanup", [])
    except tomllib.TOMLDecodeError as e:
        raise res.ResError(f"workflow.toml: {e}") from None
    return root, [res.cleanup_rule(p) for p in patterns]


def project_matches(env: Env, now: dt.datetime) -> list[Path]:
    root, rules = project_rules(env)
    out = []
    for pattern, days in rules:
        for p in sorted(root.glob(pattern)):
            if p.is_symlink() or not p.resolve().is_relative_to(root.resolve()):
                continue
            if days is not None and now.timestamp() - _newest(p) < days * 86400:
                continue
            out.append(p)
    return out


def cmd_clean(env, cfg, args) -> int:
    with locked(env) as led:
        facts = gather(env, led)
        lines = [l for l in housekeep(env, cfg, led, facts) if l.startswith("freed ")]
        lines += clean_auto(env, cfg, led, facts.now)
    root = env.root()
    for p in project_matches(env, facts.now):
        size, rel = res.fmt_bytes(_size(p)), p.relative_to(root)
        if args.yes:
            _remove(p)
            lines.append(f"deleted {size}: {rel}")
        else:
            lines.append(f"would delete {size}: {rel} (wf res clean --yes)")
    print("\n".join(lines) if lines else "nothing to clean")
    return 0


def cmd_timer(env, cfg, args) -> int:
    service, timer = env.units / "wf-res.service", env.units / "wf-res.timer"
    if args.state == "on":
        env.units.mkdir(parents=True, exist_ok=True)
        service.write_text(res.service_text(sys.executable, str(HERE / "wf.py")))
        timer.write_text(res.timer_text())
        systemctl(env, ["daemon-reload"])
        systemctl(env, ["enable", "--now", "wf-res.timer"])
        print("timer on: wf-res.timer every 1 min")
    else:
        if timer.exists():
            systemctl(env, ["disable", "--now", "wf-res.timer"])
        service.unlink(missing_ok=True)
        timer.unlink(missing_ok=True)
        systemctl(env, ["daemon-reload"])
        print("timer off")
    return 0


def adopt(env: Env) -> list[str]:
    """Move running Claude sessions (and their descendants) into agents.slice; one line per session."""
    lines = []
    groups = res.adopt_groups(env.procs())
    if groups:
        ensure_slice(env)
    for root, pids in groups.items():
        r = env.run(res.adopt_argv(root, pids))
        if r.returncode == 0:
            lines.append(f"adopted claude {root} ({len(pids)} processes)")
        else:
            first = (r.stderr or r.stdout).strip().splitlines()
            lines.append(f"claude {root}: " + (first[0] if first else f"exit {r.returncode}"))
    return lines


def cmd_adopt(env, cfg, args) -> int:
    print("\n".join(adopt(env)) or "all claude sessions already in agents.slice")
    return 0


def cmd_tick(env, cfg, args) -> int:
    with locked(env) as led:
        session(env, cfg, led)
    adopt(env)          # silent; a session that vanished mid-move is retried next minute
    try:
        cap_sessions(env, cfg)
    except res.ResError:
        pass            # a session that ended between list and set-property; retried next minute
    return 0


def cmd_hook(env, cfg, args) -> int:
    """PreToolUse hook; read-only (no lock, no systemctl) and silent on any error: it runs before every command."""
    try:
        payload = json.loads(env.stdin())
        led = res.loads((env.state / LEDGER).read_text())
        out = res.hook_output(payload, led.game_until, env.now())
    except (OSError, ValueError, AttributeError, res.ResError):
        return 0
    if out:
        print(json.dumps(out))
    return 0


def cmd_shell_init(env, cfg, args) -> int:
    print(res.shell_init_line())
    return 0
# ---------------------------------------------------------------- main

def parser() -> argparse.ArgumentParser:
    ap = argparse.ArgumentParser(prog="wf res", description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = ap.add_subparsers(dest="command", metavar="COMMAND", required=True)

    def cmd(name, func, help):
        sp = sub.add_parser(name, help=help, description=help)
        sp.set_defaults(func=func)
        return sp

    sp = cmd("run", cmd_run, "reserve and start a job as a systemd user unit; busy → exit 3")
    sp.add_argument("--mem", required=True, metavar="SIZE", help="e.g. 10G, 512M")
    sp.add_argument("--cpus", type=int, default=1, choices=range(1, 1025), metavar="N")
    sp.add_argument("--for", dest="for_", required=True, metavar="DURATION", help="estimate, e.g. 40m, 2h (ETA and queue; the job is never killed at it)")
    sp.add_argument("--title", required=True)
    sp.add_argument("--queue", action="store_true", help="wait in the FIFO instead of exit 3")
    sp.add_argument("--force", action="store_true", help=FORCE_HELP)
    sp.add_argument("--by", default="", metavar="NAME", help="your session name (ListAgents), shown as the owner; default: WF_SESSION_NAME, else the wf lane session record")
    sp.add_argument("--lock", default="", metavar="KEY", help="one queued/running job per KEY in this project (main tree "
                    "and its lane worktrees) at a time, e.g. a shared checkout; a title starting with 'gate' locks "
                    "'gate' unasked; held → exit 3, --queue waits for it, --force does not override it")
    sp.add_argument("cmd", nargs=argparse.REMAINDER, metavar="-- CMD …")

    sp = cmd("status", cmd_status, "reservations, budget, game mode, warnings")
    sp.add_argument("id", nargs="?")
    sp.add_argument("--json", action="store_true")

    sp = cmd("wait", cmd_wait, "poll every 15 s until the entry is done; prints rc and peak")
    sp.add_argument("id")
    sp.add_argument("--timeout", default="2h", metavar="DURATION")
    sp = cmd("hist", cmd_hist, "reservation sizes from history, per project and title kind (title minus trailing "
             "sha/hex words and (…)): runs n, request, peak p50/p95, estimate, duration p50/p90, suggestion "
             "(≥ 3 runs: mem = p95 peak ×1.15 ≥ 0.2 GB, for = p90 duration ×1.5 ≥ 5 min); run/note print a hint "
             "when asking > 2× it")
    sp.add_argument("--project", metavar="P", help="only this project (dir name of its main tree)")
    sp = cmd("note", cmd_note, "reserve for foreground work; freed when this Claude session exits or at --for")
    sp.add_argument("--mem", required=True, metavar="SIZE")
    sp.add_argument("--cpus", type=int, default=1, choices=range(1, 1025), metavar="N")
    sp.add_argument("--for", dest="for_", required=True, metavar="DURATION", help="reservation freed after this (nothing is killed)")
    sp.add_argument("--force", action="store_true", help=FORCE_HELP)
    sp.add_argument("--by", default="", metavar="NAME", help="your session name (ListAgents), shown as the owner; default: WF_SESSION_NAME, else the wf lane session record")
    sp.add_argument("title")
    sp = cmd("game", cmd_game, "gaming reserve on (auto-off, default 4h) or off; jobs slow, never stop")
    sp.add_argument("state", choices=["on", "off"])
    sp.add_argument("--for", dest="for_", metavar="DURATION")
    sp = cmd("clean", cmd_clean, "delete stale /tmp/claude-* scratch (auto) and list project cleanup patterns")
    sp.add_argument("--yes", action="store_true", help="also delete the project's workflow.toml cleanup matches")

    cmd("adopt", cmd_adopt, "move running Claude sessions into agents.slice (no restart); the timer does this every minute")
    sp = cmd("timer", cmd_timer, "install/remove the 1-minute wf-res.timer (queue, game expiry, clean)")
    sp.add_argument("state", choices=["on", "off"])
    cmd("tick", cmd_tick, "housekeeping once (run by the timer); silent")
    cmd("hook", cmd_hook, "Claude Code PreToolUse hook (Bash): no display for agent commands while game is on")
    cmd("shell-init", cmd_shell_init, "print the alias that starts Claude inside agents.slice (add to ~/.bashrc)")
    sp = cmd("release", cmd_release, "drop an entry (cancels a queued job, frees a note); --stop kills a running job")
    sp.add_argument("id")
    sp.add_argument("--stop", action="store_true")
    return ap


# ---------------------------------------------------------------- wf batch

FORCE_HELP = ("memory is really free though the ledger says busy (unused reservations, stale notes): fit against "
              "MemAvailable minus the user reserve and headroom only, ignoring claims and the queue; still recorded; "
              "the busy line names --force when it would fit")
CEILING_VAR = "CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS"
BATCH_TITLE = "wf-batch"
BATCH_FOR = "6h"          # default --for and ceiling (no history)
PREP_MEM = "1G"
STOP_FILE = "out/wf-batch.stop"


def cmd_batch(env, cfg, args) -> int:
    if args.stop:
        f = env.root() / STOP_FILE
        f.parent.mkdir(exist_ok=True)
        f.touch()
        print(f"stop requested: {STOP_FILE} (a batch or loop stops before its next spawn; running workers finish)")
        return 0
    if args.status:
        return batch_status(env, cfg)
    if args.time_left:
        return batch_time_left(env, args.time_left)
    if args.n is None:
        batch_parser().print_usage(sys.stderr)
        print("wf batch: error: N is required (or --status)", file=sys.stderr)
        return 2
    if args.left:
        if args.prep:
            raise res.ResError("--left is for work batches, not --prep (use --for)")
        left = res.parse_duration(args.left)
        p90, runs = _task_p90(env)
        k = res.batch_fit(args.n, left, p90)
        line = f"fit: {k} of {args.n} tasks (left {res.fmt_dur(left)}, task p90 {_p90_text(p90, runs)})"
        if not k:
            print(f"{line}; nothing started")
            return 0
        print(line)
        args.n = k
        args.for_ = args.for_ or args.left
    sug = None
    if not args.prep and (args.mem is None or args.for_ is None):     # prep jobs share the title, not the size
        with locked(env) as led:
            sug = suggestion(env, led, BATCH_TITLE)
    minutes = res.parse_duration(args.for_ or BATCH_FOR)      # the ceiling: never shortened by history
    if args.for_ is None:
        args.for_ = f"{sug[1]}m" if sug else BATCH_FOR
    if args.mem is None:      # prep: one claude -p, subagents in-process, no builds → small
        args.mem = PREP_MEM if args.prep else f"{sug[0]:g}G" if sug else f"{min(8.0, max(4.0, 1.5 * args.n)):g}G"
    claude = shutil.which("claude", path=env.environ().get("PATH"))
    if not claude:
        raise res.ResError("claude not found on PATH")
    out = f"out/wf-batch-{env.now().strftime('%Y-%m-%d-%H%M')}.md"
    names = [x.strip() for x in (args.lanes or "").split(",") if x.strip()]
    stamp = env.now().strftime('%Y-%m-%d %H:%M')
    if args.prep:
        ids, tasks_rel = prep_ids(env.root(), set(names) if names else None, args.n)
        if not ids:
            print(f"prep: no pending task without Done (lanes: {', '.join(names) or 'all'}); nothing started")
            return 0
        prompt = (HERE / "templates" / "prep-prompt.md").read_text().strip().format(
            here=HERE, root=env.root(), ids=" ".join(ids), out=out, tasks=tasks_rel)
        header = f"prep K={args.n}: {' '.join(ids)}"
    else:
        lanes = ", ".join(names) if names else \
            "every lane with ready tasks, see wf lanes"
        prompt = (HERE / "templates" / "batch-prompt.md").read_text().strip().format(
            here=HERE, n=args.n, lanes=lanes, out=out, stop=env.root() / STOP_FILE,
            deadline=(env.now() + dt.timedelta(minutes=minutes)).isoformat(timespec="minutes"))
        header = f"N={args.n}, lanes: {lanes}"
    summary = env.root() / out
    if not args.dry_run:      # dry-run touches nothing; an existing summary (same minute) is appended to, never clobbered
        summary.parent.mkdir(exist_ok=True)
        with summary.open("a") as f:
            f.write(f"# wf-batch {env.root().name} {stamp} ({header})\n")
    cmd = [claude, "-p", "--model", args.model, "--permission-mode", "auto", "--permission-prompts", "none", prompt]
    if not args.prep and _cloud_project(env.root()):
        cmd = [sys.executable, str(HERE / "wf_res.py"), SIDECAR, "--root", str(env.root()), "--summary", out,
               "--every", str(SIDECAR_EVERY), "--", *cmd]
    ceiling = str(minutes * 60_000)
    if args.dry_run:
        print(f"{CEILING_VAR}={ceiling} wf res run --mem {args.mem} --for {args.for_} --title {BATCH_TITLE} -- "
              + " ".join(shlex.quote(a) for a in cmd))
        return 0
    stop = env.root() / STOP_FILE
    if stop.exists():
        stop.unlink()
        print(f"warning: stale {STOP_FILE} cleared; starting batch", file=sys.stderr)
    base = env.environ
    env.environ = lambda: {**base(), CEILING_VAR: ceiling}
    try:
        code = cmd_run(env, cfg, argparse.Namespace(mem=args.mem, cpus=1, for_=args.for_, title=BATCH_TITLE,
                                                     queue=False, force=args.force, cmd=cmd))
    finally:
        env.environ = base
        if stop.exists():
            stop.unlink()
    print(f"summary: {out} · status: wf batch --status")
    return code


SIDECAR = "batch-sidecar"
SIDECAR_EVERY = 600        # s between the sidecar's wf cloud pull --all runs
PULLED = re.compile(r"^(\S+): (\w+) \(")
TAIL_CAP = 24 * 3600     # s: the tail after the orchestrator exits pulls at most this long
TAIL_MEM_GB, TAIL_CPUS = 0.2, 1   # the job's reservation during the tail


def _cloud_project(root: Path) -> bool:
    from wflib import config
    try:
        return config.load(config.find_root(root)).cloud
    except (config.ConfigError, OSError):
        return False


def sidecar(child: list[str], root: Path, summary: Path, every: float, wf_cmd: list[str],
            cap: float = TAIL_CAP, shrink: Callable[[], str] | None = None) -> int:
    """Run the batch orchestrator `child`; while it lives, every `every` s (until the stop file appears):
    wf cloud pull --all, each ended '<id>: <state> (…' line → wf orch post <id> cloud --result <state>
    [--commit <sha>] --no-pick + one summary line. After it exits with .wf/cloud records left: the tail —
    shrink() the job's reservation, keep pulling every `every` s (no picks, no sends) until no record is left,
    the stop file appears or `cap` s passed; one summary line says why it ended. Returns the child's exit code."""
    proc = subprocess.Popen(child)
    while True:
        try:
            rc = proc.wait(timeout=every)
            break
        except subprocess.TimeoutExpired:
            pass
        if (root / STOP_FILE).exists():
            print("sidecar: stop file, no more pulls", flush=True)
            return proc.wait()
        _pull_post(root, summary, wf_cmd)
    if not _cloud_records(root):
        return rc
    print(f"sidecar: orchestrator exited ({rc}), cloud records left: {' '.join(_cloud_records(root))}; "
          f"pulling every {every:g}s (no picks) until none, stop file or {cap / 3600:g}h", flush=True)
    if shrink:
        print(shrink(), flush=True)
    t0, pulls = time.monotonic(), 0
    while True:
        time.sleep(every)
        if (root / STOP_FILE).exists():
            why = "stop file"
        elif time.monotonic() - t0 >= cap:
            why = f"{cap / 3600:g}h cap"
        else:
            _pull_post(root, summary, wf_cmd)
            pulls += 1
            if _cloud_records(root):
                continue
            why = "no cloud records left"
        break
    left = " ".join(_cloud_records(root)) or "none"
    line = f"cloud tail ended: {why}; {pulls} pulls; left: {left}"
    print(f"sidecar: {line}", flush=True)
    with summary.open("a") as f:
        f.write(f"{time.strftime('%H:%M')} {line} (sidecar)\n")
    return rc


def _cloud_records(root: Path) -> list[str]:
    return sorted(p.stem for p in (root / ".wf" / "cloud").glob("*.json"))


def _pull_post(root: Path, summary: Path, wf_cmd: list[str]) -> None:
    pull = subprocess.run([*wf_cmd, "cloud", "pull", "--all", "--project", str(root)], cwd=root,
                          capture_output=True, text=True)
    print(pull.stdout + pull.stderr, end="", flush=True)
    for line in pull.stdout.splitlines():
        m = PULLED.match(line)
        if not m or m[2] == "running" or line.startswith("archive "):
            continue
        sha = re.search(r"merged ([0-9a-f]{7,40})\b", line)
        post = [*wf_cmd, "orch", "post", m[1], "cloud", "--result", m[2], "--no-pick"]
        post += ["--commit", sha[1]] if sha else []
        r = subprocess.run(post, cwd=root, capture_output=True, text=True)
        print(r.stdout + r.stderr, end="", flush=True)
        with summary.open("a") as f:
            f.write(f"{time.strftime('%H:%M')} cloud {m[1]} {m[2]} {sha[1] if sha else '-'} (sidecar pull)\n")


def shrink_reservation(env: Env, res_id: str, mem_gb: float = TAIL_MEM_GB, cpus: int = TAIL_CPUS) -> str:
    """The batch job's ledger entry down to the tail's size (ledger only: the unit's MemoryMax stays, a pull
    must never be OOM-killed)."""
    with locked(env) as led:
        e = next((x for x in led.entries if x.id == res_id), None)
        if not e or e.state != "running":
            return f"sidecar: {res_id} not running in the ledger; reservation kept"
        old = e.mem_gb
        e.mem_gb, e.cpus = min(e.mem_gb, mem_gb), min(e.cpus, cpus)
        return (f"sidecar: {res_id} reservation {res.fmt_gb(old)} -> {res.fmt_gb(e.mem_gb)}, {e.cpus} cpu "
                "for the cloud tail")


def sidecar_main(argv: list[str]) -> int:
    ap = argparse.ArgumentParser(prog=f"wf_res.py {SIDECAR}", description="wf batch job wrapper of cloud projects "
                                 "(started by wf batch, not by hand): runs the orchestrator and pulls cloud tasks")
    ap.add_argument("--root", type=Path, required=True)
    ap.add_argument("--summary", required=True, help="summary file, relative to --root")
    ap.add_argument("--every", type=float, default=SIDECAR_EVERY)
    ap.add_argument("--wf", default=f"{sys.executable} {HERE / 'wf.py'}", help="wf command (tests: a fake)")
    ap.add_argument("--cap", type=float, default=TAIL_CAP, help="s the tail (cloud records left after the "
                    "orchestrator exits) pulls at most")
    ap.add_argument("cmd", nargs=argparse.REMAINDER)
    a = ap.parse_args(argv)
    cmd = a.cmd[1:] if a.cmd[:1] == ["--"] else a.cmd
    if not cmd:
        ap.error("no command after --")
    rid = os.environ.get(res.RES_ID_VAR, "")

    def shrink() -> str:
        try:
            return shrink_reservation(real_env(), rid)
        except (res.ResError, OSError) as e:
            return f"sidecar: reservation of {rid} kept: {e}"
    return sidecar(cmd, a.root, a.root / a.summary, a.every, shlex.split(a.wf), a.cap, shrink if rid else None)


def prep_ids(root: Path, only: set[str] | None, k: int) -> tuple[list[str], str]:
    """Up to k ids a prep batch works (wflib.lanes.prep_targets) + the task file path relative to root."""
    from wflib import config, lanes, tasks
    try:
        cfg = config.load(config.find_root(root))
        doc = tasks.parse(cfg.tasks.read_text(encoding="utf-8"))
        archived = tasks.archive_ids(cfg.archive.read_text(encoding="utf-8"))
    except (config.ConfigError, OSError) as e:
        raise res.ResError(f"batch --prep: {e}") from None
    items = lanes.prep_targets(doc, archived, cfg.lanes, cfg.slice_above, only)
    return [i.id for i in items[:k]], cfg.rel(cfg.tasks)


def _orch_progress(root: Path, stem: str) -> list[str]:
    """out/wf-orch.log lines (start 'YYYY-MM-DD HH:MM' or 'T') at/after the batch start (from the summary name)."""
    m = re.fullmatch(r"wf-batch-(\d{4}-\d\d-\d\d)-(\d\d)(\d\d)", stem)
    log = root / "out" / "wf-orch.log"
    if not m or not log.is_file():
        return []
    start = f"{m[1]} {m[2]}:{m[3]}"
    done = [ln for ln in log.read_text().splitlines()
            if re.match(r"\d{4}-\d\d-\d\d[ T]\d\d:\d\d", ln) and ln[:16].replace("T", " ") >= start]
    return ["finished since batch start (out/wf-orch.log):"] + done[-30:] if done else []


def _task_p90(env) -> tuple[int, int]:
    log = env.root() / "out" / "wf-orch.log"
    return res.task_p90(log.read_text(errors="replace") if log.exists() else "")


def _p90_text(p90: int, runs: int) -> str:
    return f"{res.fmt_dur(p90)} from {runs} runs" if runs >= res.TASK_P90_MIN_RUNS else \
        f"{res.fmt_dur(p90)} default ({runs} runs)"


def batch_time_left(env, deadline: str) -> int:
    """Per-round check of a running batch: spawn while the time to the deadline is ≥ the task p90."""
    try:
        end = dt.datetime.fromisoformat(deadline)
    except ValueError:
        raise res.ResError(f"--time-left '{deadline}' (want an ISO time, e.g. 2026-10-01T20:00+02:00)")
    now = env.now()
    if end.tzinfo is None:
        end = end.replace(tzinfo=now.tzinfo)
    left = max(0, int((end - now).total_seconds() // 60))
    p90, runs = _task_p90(env)
    word = "spawn" if left >= p90 else "stop"
    print(f"time left {res.fmt_dur(left)}, task p90 {res.fmt_dur(p90)} ({runs} runs): {word}")
    return 0


def batch_status(env, cfg) -> int:
    root = env.root()
    with locked(env) as led:
        facts = session(env, cfg, led)
        mine = [e for e in led.entries if e.title == BATCH_TITLE and e.project == main_tree(env).name]
        lines = [res.entry_line(mine[-1], led, facts), f"  log {mine[-1].log}"] if mine else []
    summaries = sorted((root / "out").glob("wf-batch-*.md"))
    if summaries:
        rel = summaries[-1].relative_to(root)
        lines += [f"{rel}:"] + summaries[-1].read_text().rstrip("\n").splitlines()[-30:]
        lines += _orch_progress(root, summaries[-1].stem)
    if (root / STOP_FILE).exists():
        lines.append(f"stop requested: {STOP_FILE}")
    print("\n".join(lines) or "no batch in this project")
    return 0


def _count(text: str) -> int:
    if not text.isdigit() or int(text) < 1:
        raise argparse.ArgumentTypeError("N must be ≥ 1")
    return int(text)


def batch_parser() -> argparse.ArgumentParser:
    ap = argparse.ArgumentParser(
        prog="wf batch", description="Start an unattended batch orchestrator: claude -p (auto mode, no prompts) "
        "as a wf res job, prompt templates/batch-prompt.md; it works up to N tasks and writes out/wf-batch-*.md. "
        "Owner allow rule (once): Bash(python3 /projects/public/workflow/wf.py batch:*).")
    ap.add_argument("n", nargs="?", type=_count, metavar="N", help="at most N tasks")
    ap.add_argument("--lanes", metavar="L,…", help="lanes to work (default: every lane with ready tasks)")
    ap.add_argument("--model", default="opus", help="the batch orchestrator's model (default opus)")
    ap.add_argument("--for", dest="for_", default=None, metavar="DURATION",
                    help=f"estimate and {CEILING_VAR} (default: estimate from wf res hist for wf-batch in this "
                    f"project when ≥ 3 runs (not --prep), else {BATCH_FOR}; the ceiling stays {BATCH_FOR} unless given)")
    ap.add_argument("--mem", default=None, metavar="SIZE",
                    help="reservation (default: wf res hist suggestion for wf-batch in this project when ≥ 3 runs, "
                    "else max(4G, 1.5G x N), cap 8G; --prep: 1G; explicit wins; a request > 2x history prints a hint)")
    ap.add_argument("--dry-run", action="store_true", help="print the command, start nothing")
    ap.add_argument("--force", action="store_true", help=FORCE_HELP)
    ap.add_argument("--prep", action="store_true",
                    help="prep batch: for up to N pending tasks without Done (pickable first, then priority) sonnet "
                    "workers write Done (+Model) from the task text, unclear ones → awaiting; prompt "
                    "templates/prep-prompt.md; summary lines 'id → Done: …' | 'id → awaiting a-…'. "
                    "None to prep → prints so, starts nothing")
    ap.add_argument("--left", metavar="DURATION",
                    help="time left to the pilot's deadline: N becomes min(N, floor(left / task p90)) (p90 of done/"
                    f"handback durations in out/wf-orch.log, < {res.TASK_P90_MIN_RUNS} runs → {res.TASK_P90_DEFAULT}m), "
                    "prints 'fit: K of N tasks (…)'; K = 0 → nothing started (rc 0); --for defaults to DURATION")
    ap.add_argument("--time-left", metavar="DEADLINE",
                    help="(run by the batch orchestrator before each round) print 'time left X, task p90 Y (n runs): "
                    "spawn|stop' for the ISO DEADLINE its prompt names (= start + ceiling); stop = left < p90")
    ap.add_argument("--status", action="store_true", help="newest batch job of this project + its summary file + tasks finished since start (out/wf-orch.log)")
    ap.add_argument("--stop", action="store_true",
                    help=f"graceful stop: create {STOP_FILE}; batch and loop check it before every spawn (latency ≤ the running "
                    "tasks), stop with 'stopped: stop file' and removes it")
    ap.set_defaults(func=cmd_batch)
    return ap


def batch_main(argv: list[str], env: Env | None = None) -> int:
    return main(argv, env, batch_parser())


PREP_ALL = 1000     # "wf prep all": more than any task list


def prep_main(argv: list[str], env: Env | None = None) -> int:
    """wf prep [N|all] [batch opts] = wf batch --prep N (all = every not-ready task)."""
    if argv and argv[0] == "all":
        argv = [str(PREP_ALL), *argv[1:]]
    return batch_main([*argv, "--prep"], env)


def load_cfg(env: Env) -> res.Config:
    return res.load_config(env.config.read_text() if env.config.exists() else "")


def main(argv: list[str], env: Env | None = None, ap: argparse.ArgumentParser | None = None) -> int:
    try:
        args = (ap or parser()).parse_args(argv)
    except SystemExit as e:          # usage error → 2, -h → 0; callers (and tests) get a code, not an exception
        return e.code if isinstance(e.code, int) else 2
    try:
        env = env or real_env()
        return args.func(env, load_cfg(env), args)
    except res.Busy as e:
        print(e)
        return 3
    except res.ResError as e:
        warn(str(e))
        return 1
    except BrokenPipeError:
        return 0
    except OSError as e:
        warn(str(e))
        return 1


if __name__ == "__main__":
    sys.exit(sidecar_main(sys.argv[2:]) if sys.argv[1:2] == [SIDECAR] else main(sys.argv[1:]))