box/bin/box-ctl.py

#!/usr/bin/env python3
"""box-ctl.py — allowlisted bl helper for Box API mutations.

The board server (VM) never runs raw systemctl or shell over SSH.
All bl mutations go through this helper, invoked as:

    /home/super/Projects/NetVM/bin/box-ctl.py <action> [args...]

Security properties:
- Fixed verb set; every argument validated before acting.
- <name> must match ^[a-z0-9-]{1,64}$ (kills path traversal).
- Unit files generated from a fixed template; only the validated
  name is interpolated. User-controlled strings never appear in
  unit files.
- No shell=True anywhere. No string interpolation into commands.
- Job JSON schema-validated before writing; changes git-committed.
- Every action audit-logged to box-ctl.jsonl with caller identity.

Output: JSON to stdout ({"ok": true, ...} or {"ok": false, ...}),
exit 0 on success, nonzero on failure.
"""

import json
import os
import re
import subprocess
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

NETVM_ROOT = Path("/home/super/Projects/NetVM")
BIN = NETVM_ROOT / "bin"
if str(BIN) not in sys.path:
    sys.path.insert(0, str(BIN))
JOBS_DIR = NETVM_ROOT / "jobs"
SYSTEMD_USER = Path.home() / ".config" / "systemd" / "user"
CTL_LOG = NETVM_ROOT / "box-ctl.jsonl"
JOB_LOG = NETVM_ROOT / "job-log.jsonl"
DISPATCHER = BIN / "job-dispatch.py"
DM_PY = BIN / "dm.py"
POLICY_FILE = NETVM_ROOT / "CHAT_POLICY.md"
DM_LOG = NETVM_ROOT / "dm-log.jsonl"
WATCHDOG_STATE = NETVM_ROOT / "main-chat-watchdog.state"

NAME_RE = re.compile(r"^[a-z0-9-]{1,64}$")
VALID_AGENTS = {"muse", "pip", "646", "opm", "dev", "def"}
THREAD_RE = re.compile(r"^[A-Za-z0-9-]{1,64}$")
THREAD_OPS = ("list", "pin", "unpin", "archive", "unarchive")

# Default sidechat per agent for `box notify`. Mirrors super-cli.py
# DEFAULT_AGENT_SIDECHATS (kept in sync manually; super-cli is the
# canonical copy). Sidechat-first policy: notify never targets main
# chat unless --allow-main-chat is passed explicitly.
NOTIFY_SIDECHATS = {
    "646": "646 tasks",
    "opm": "main-loop brain",
    "pip": "pip tasks",
    "muse": "muse tasks",
    "dev": "onboarding-dev",
}
VALID_ON_FAILURE = {"retry", "alert", "ignore"}
KNOWN_PLACEHOLDERS = {"job_id", "job_name", "datetime", "date", "last_run"}
PROTOCOL_LITERALS = ("[REQ", "[CONFIRM", "[JOB", "[RESULT")
DOW_NAMES = ["Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"]


def utcnow():
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def out(ok, **kw):
    payload = {"ok": ok}
    payload.update(kw)
    print(json.dumps(payload))


def fail(code, error, detail=None, exit_code=1):
    payload = {"ok": False, "code": code, "error": error}
    if detail is not None:
        payload["detail"] = detail
    print(json.dumps(payload))
    sys.exit(exit_code)


def audit(action, name=None):
    """Append {ts, action, name, caller} to the bl-side log."""
    try:
        entry = {
            "ts": utcnow(),
            "action": action,
            "name": name,
            "caller": os.environ.get("BOX_CALLER", "unknown"),
        }
        with open(CTL_LOG, "a") as f:
            f.write(json.dumps(entry) + "\n")
    except Exception:
        pass  # audit failure must not break the action


def check_name(name):
    if not name or not NAME_RE.match(name):
        fail("BAD_NAME",
             "name must match ^[a-z0-9-]{1,64}$",
             {"field": "name", "value": name})
    return name


def run(cmd, timeout=30):
    """Run a command with no shell. Returns CompletedProcess."""
    return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)


# ---------------------------------------------------------------------------
# Cron validation and conversion
# ---------------------------------------------------------------------------

CRON_RANGES = {
    0: (0, 59),   # minute
    1: (0, 23),   # hour
    2: (1, 31),   # day of month
    3: (1, 12),   # month
    4: (0, 7),    # day of week (0 and 7 = Sunday)
}


def validate_cron_field(field, idx):
    """Validate one cron field. Returns True/False."""
    lo, hi = CRON_RANGES[idx]
    if field == "*":
        return True
    for part in field.split(","):
        # step: base/n
        if "/" in part:
            base, step = part.split("/", 1)
            if not step.isdigit() or int(step) < 1:
                return False
            if base != "*" and not validate_cron_field(base, idx):
                return False
            continue
        # range: a-b
        if "-" in part:
            a, b = part.split("-", 1)
            if not (a.isdigit() and b.isdigit()):
                return False
            if not (lo <= int(a) <= hi and lo <= int(b) <= hi):
                return False
            if int(a) > int(b):
                return False
            continue
        # single value
        if not part.isdigit():
            return False
        if not (lo <= int(part) <= hi):
            return False
    return True


def validate_cron(schedule):
    """Validate a 5-field cron string. Returns (ok, reason)."""
    parts = schedule.split()
    if len(parts) != 5:
        return False, "schedule must have exactly 5 fields (M H dom mon dow)"
    for i, p in enumerate(parts):
        if not validate_cron_field(p, i):
            return False, f"schedule field {i + 1} ({p!r}) is invalid"
    return True, ""


def cron_to_oncalendar(schedule):
    """Convert common cron patterns to systemd OnCalendar.

    Returns the OnCalendar string, or None if the pattern is not
    supported (caller rejects with INVALID_SCHEDULE).
    """
    m, h, dom, mon, dow = schedule.split()

    def is_star(f):
        return f == "*"

    # */n * * * * -> *:0/n  (every n minutes)
    if m.startswith("*/") and m[2:].isdigit() and all(is_star(f) for f in (h, dom, mon, dow)):
        return f"*:0/{m[2:]}"
    # M * * * * -> hourly or *:M
    if m.isdigit() and all(is_star(f) for f in (h, dom, mon, dow)):
        if m == "0":
            return "hourly"
        return f"*:{int(m):02d}"
    # M H * * * -> HH:MM (daily)
    if m.isdigit() and h.isdigit() and all(is_star(f) for f in (dom, mon, dow)):
        return f"{int(h):02d}:{int(m):02d}"
    # M H * * DOW -> Dow HH:MM (weekly)
    if m.isdigit() and h.isdigit() and dow != "*" and is_star(dom) and is_star(mon):
        # take first dow value for the weekly form
        d = dow.split(",")[0].split("-")[0].split("/")[0]
        if d.isdigit() and 0 <= int(d) <= 7:
            return f"{DOW_NAMES[int(d) % 7]} {int(h):02d}:{int(m):02d}"
        return None
    # M H DOM * * -> *-*-DOM HH:MM:00 (monthly)
    if m.isdigit() and h.isdigit() and dom.isdigit() and is_star(mon) and is_star(dow):
        return f"*-*-{int(dom):02d} {int(h):02d}:{int(m):02d}:00"
    # M,M... * * * * -> *:M1,M2...
    if all(part.isdigit() for part in m.split(",")) and all(is_star(f) for f in (h, dom, mon, dow)):
        formatted_m = ",".join(f"{int(part):02d}" for part in m.split(","))
        return f"*:{formatted_m}"
    return None


def validate_oncalendar(expr):
    """Check an OnCalendar expression with systemd-analyze."""
    r = run(["systemd-analyze", "calendar", expr, "--iterations=2"], timeout=15)
    return r.returncode == 0


# ---------------------------------------------------------------------------
# Job schema validation (§6 of the design doc)
# ---------------------------------------------------------------------------

def validate_job(data):
    """Validate a job definition dict. Returns (ok, field, reason)."""
    if not isinstance(data, dict):
        return False, None, "job must be a JSON object"

    # name
    name = data.get("name")
    if not isinstance(name, str) or not NAME_RE.match(name):
        return False, "name", "must match ^[a-z0-9-]{1,64}$"

    # description
    desc = data.get("description")
    if desc is not None:
        if not isinstance(desc, str) or len(desc) > 280:
            return False, "description", "must be a string ≤ 280 chars"

    # schedule
    sched = data.get("schedule")
    if not isinstance(sched, str):
        return False, "schedule", "is required and must be a string"
    if sched != "manual":
        ok, reason = validate_cron(sched)
        if not ok:
            return False, "schedule", reason
        oc = cron_to_oncalendar(sched)
        if oc is None:
            return False, "schedule", (
                "cron pattern not supported for OnCalendar conversion; "
                "supported: */n * * * *, M * * * *, M H * * *, M H * * DOW, M H DOM * *"
            )
        if not validate_oncalendar(oc):
            return False, "schedule", f"converted OnCalendar {oc!r} rejected by systemd-analyze"

    # scheduler
    scheduler = data.get("scheduler", "systemd")
    if scheduler != "systemd":
        return False, "scheduler", 'v1 accepts only "systemd"'

    # agent
    agent = data.get("agent")
    if agent not in VALID_AGENTS:
        return False, "agent", f"must be one of {sorted(VALID_AGENTS)}"

    # prompt_template
    pt = data.get("prompt_template")
    if not isinstance(pt, str) or not (1 <= len(pt) <= 4000):
        return False, "prompt_template", "must be a string of 1–4000 chars"
    for lit in PROTOCOL_LITERALS:
        if lit in pt:
            return False, "prompt_template", (
                f"must not contain {lit!r} literal (protocol framing is added by the dispatcher)"
            )
    for ph in re.findall(r"\{([a-zA-Z_][a-zA-Z0-9_]*)\}", pt):
        if ph not in KNOWN_PLACEHOLDERS:
            return False, "prompt_template", f"unknown placeholder {{{ph}}}"

    # timeout
    timeout = data.get("timeout", 300)
    if not isinstance(timeout, int) or isinstance(timeout, bool) or not (60 <= timeout <= 3600):
        return False, "timeout", "must be an integer 60–3600"

    # on_failure
    onf = data.get("on_failure", "alert")
    if onf not in VALID_ON_FAILURE:
        return False, "on_failure", f"must be one of {sorted(VALID_ON_FAILURE)}"

    # chain_next
    cn = data.get("chain_next")
    if cn is not None:
        if not isinstance(cn, str) or not NAME_RE.match(cn):
            return False, "chain_next", "must be null or a valid job name"
        if not (JOBS_DIR / f"{cn}.json").exists():
            return False, "chain_next", f"job {cn!r} does not exist (dangling chain)"

    # sidechat
    sc = data.get("sidechat")
    if sc is not None:
        if not isinstance(sc, dict):
            return False, "sidechat", "must be an object"
        create = sc.get("create", False)
        if not isinstance(create, bool):
            return False, "sidechat.create", "must be a boolean"
        nt = sc.get("name_template")
        if nt is not None:
            if not isinstance(nt, str) or len(nt) > 120:
                return False, "sidechat.name_template", "must be a string ≤ 120 chars"
            for ph in re.findall(r"\{([a-zA-Z_][a-zA-Z0-9_]*)\}", nt):
                if ph not in {"job_name", "date", "job_id"}:
                    return False, "sidechat.name_template", f"unknown placeholder {{{ph}}}"
        rk = sc.get("reuse_key")
        if rk is not None:
            if not isinstance(rk, str) or not NAME_RE.match(rk):
                return False, "sidechat.reuse_key", "must match ^[a-z0-9-]{1,64}$"

    return True, None, ""


# ---------------------------------------------------------------------------
# Unit file templates (fixed; only <name> interpolated, regex-validated)
# ---------------------------------------------------------------------------

SERVICE_TEMPLATE = """[Unit]
Description=Dispatch {name} job
After=network.target

[Service]
Type=oneshot
ExecStart=/home/super/Projects/NetVM/bin/job-dispatch.py {name}
User=super
WorkingDirectory=/home/super/Projects/NetVM
"""

TIMER_TEMPLATE = """[Unit]
Description=Run {name} job on schedule

[Timer]
OnCalendar={oncalendar}
Persistent=true
AccuracySec=1min

[Install]
WantedBy=timers.target
"""


def unit_path(name, kind):
    return SYSTEMD_USER / f"job-{name}.{kind}"


def systemctl(*args, timeout=30):
    return run(["systemctl", "--user"] + list(args), timeout=timeout)


# ---------------------------------------------------------------------------
# Actions
# ---------------------------------------------------------------------------

def act_timer_list():
    timers = []
    try:
        entries = sorted(SYSTEMD_USER.glob("job-*.timer"))
    except Exception:
        entries = []
    for p in entries:
        unit = p.name
        name = unit[len("job-"):-len(".timer")]
        active = systemctl("is-active", unit).stdout.strip() == "active"
        enabled_out = systemctl("is-enabled", unit).stdout.strip()
        enabled = enabled_out == "enabled"
        job_path = JOBS_DIR / f"{name}.json"
        timers.append({
            "name": name,
            "unit": unit,
            "active": active,
            "enabled": enabled,
            "orphan": not job_path.exists(),
        })
    audit("timer-list")
    out(True, timers=timers)


def _timer_show_property(unit, prop):
    r = systemctl("show", unit, f"--property={prop}", "--value")
    if r.returncode != 0:
        return None
    return r.stdout.strip() or None


def act_timer_status(name):
    check_name(name)
    unit = f"job-{name}.timer"
    if not unit_path(name, "timer").exists():
        fail("NOT_FOUND", f"no such timer: {unit}")
    active = systemctl("is-active", unit).stdout.strip() == "active"
    enabled = systemctl("is-enabled", unit).stdout.strip() == "enabled"
    last_usec = _timer_show_property(unit, "LastTriggerUSec")
    next_usec = _timer_show_property(unit, "NextElapseUSec")
    oncalendar_raw = _timer_show_property(unit, "TimersCalendar")
    # TimersCalendar returns "{ OnCalendar=... ; next_elapse=... }" — extract just the expr
    oncalendar = oncalendar_raw
    if oncalendar_raw:
        m = re.search(r"OnCalendar=([^;}]+)", oncalendar_raw)
        if m:
            oncalendar = m.group(1).strip()
    result = _timer_show_property(unit, "Result")

    def usec_to_iso(v):
        if not v or v in ("0", "n/a"):
            return None
        try:
            ts = int(v) / 1_000_000
            return datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
        except Exception:
            return None

    # last_result from job-log.jsonl
    last_result = "unknown"
    try:
        if JOB_LOG.exists():
            for line in reversed(JOB_LOG.read_text().splitlines()[-500:]):
                try:
                    e = json.loads(line)
                except Exception:
                    continue
                if e.get("job_name") != name and not str(e.get("job_id", "")).startswith(name + "-"):
                    continue
                t = e.get("type")
                if t in ("job_dispatched",):
                    last_result = "success"
                    break
                if t in ("job_failed", "job_timeout"):
                    last_result = "failed"
                    break
    except Exception:
        pass

    job_def = None
    job_path = JOBS_DIR / f"{name}.json"
    if job_path.exists():
        try:
            job_def = json.loads(job_path.read_text())
        except Exception:
            job_def = None

    audit("timer-status", name)
    out(True, name=name, unit=unit, active=active, enabled=enabled,
        last_run=usec_to_iso(last_usec), next_run=usec_to_iso(next_usec),
        last_result=last_result, oncalendar=oncalendar,
        job_definition=job_def)


def act_timer_create(name):
    check_name(name)
    job_path = JOBS_DIR / f"{name}.json"
    if not job_path.exists():
        fail("NOT_FOUND", f"job definition missing: jobs/{name}.json — create the job first")
    try:
        job = json.loads(job_path.read_text())
    except Exception as e:
        fail("INVALID_JOB", f"job JSON unreadable: {e}")
    ok, field, reason = validate_job(job)
    if not ok:
        fail("INVALID_JOB", f"job validation failed: {field}: {reason}",
             {"field": field, "reason": reason})
    if unit_path(name, "timer").exists():
        fail("ALREADY_EXISTS", f"timer job-{name}.timer already exists — delete it first")

    if job.get("schedule") == "manual":
        fail("INVALID_SCHEDULE", "cannot create a systemd timer for a job with schedule 'manual'")

    oncalendar = cron_to_oncalendar(job["schedule"])
    if oncalendar is None or not validate_oncalendar(oncalendar):
        fail("INVALID_SCHEDULE", "schedule did not convert to a valid OnCalendar")

    SYSTEMD_USER.mkdir(parents=True, exist_ok=True)
    unit_path(name, "service").write_text(SERVICE_TEMPLATE.format(name=name))
    unit_path(name, "timer").write_text(
        TIMER_TEMPLATE.format(name=name, oncalendar=oncalendar))

    r = systemctl("daemon-reload")
    if r.returncode != 0:
        fail("TIMER_CREATE_FAILED", f"daemon-reload failed: {r.stderr.strip()}")
    r = systemctl("enable", "--now", f"job-{name}.timer")
    if r.returncode != 0:
        fail("TIMER_CREATE_FAILED", f"enable --now failed: {r.stderr.strip()}")

    audit("timer-create", name)
    out(True, name=name, unit=f"job-{name}.timer", created=True,
        oncalendar=oncalendar)


def act_timer_delete(name, keep_job=False):
    check_name(name)
    unit = f"job-{name}.timer"
    if not unit_path(name, "timer").exists():
        fail("NOT_FOUND", f"no such timer: {unit}")
    systemctl("stop", unit)
    systemctl("disable", unit)
    for kind in ("timer", "service"):
        p = unit_path(name, kind)
        try:
            p.unlink()
        except FileNotFoundError:
            pass
    r = systemctl("daemon-reload")
    if r.returncode != 0:
        fail("TIMER_CREATE_FAILED", f"daemon-reload failed: {r.stderr.strip()}")
    audit("timer-delete", name)
    out(True, name=name, unit=unit, deleted=True, job_kept=keep_job or (JOBS_DIR / f"{name}.json").exists())


def act_timer_control(name, op):
    check_name(name)
    unit = f"job-{name}.timer"
    if not unit_path(name, "timer").exists():
        fail("NOT_FOUND", f"no such timer: {unit}")
    valid = {"start", "stop", "enable", "disable"}
    if op not in valid:
        fail("BAD_NAME", f"unknown timer op: {op}")
    args = ["--now", unit] if op == "enable" else [unit]
    r = systemctl(op, *args)
    if r.returncode != 0:
        fail("TIMER_CREATE_FAILED", f"systemctl {op} failed: {r.stderr.strip()}")
    audit(f"timer-{op}", name)
    out(True, name=name, unit=unit, op=op)


def _job_summary(name, path):
    try:
        d = json.loads(path.read_text())
    except Exception:
        return {"name": name, "error": "unreadable"}
    return {
        "name": name,
        "description": d.get("description"),
        "schedule": d.get("schedule"),
        "agent": d.get("agent"),
        "timeout": d.get("timeout", 300),
    }


def act_job_list():
    jobs = []
    if JOBS_DIR.exists():
        for p in sorted(JOBS_DIR.glob("*.json")):
            jobs.append(_job_summary(p.stem, p))
    audit("job-list")
    out(True, jobs=jobs)


def act_job_get(name):
    check_name(name)
    p = JOBS_DIR / f"{name}.json"
    if not p.exists():
        fail("NOT_FOUND", f"no such job: {name}")
    try:
        data = json.loads(p.read_text())
    except Exception as e:
        fail("INVALID_JOB", f"job JSON unreadable: {e}")
    audit("job-get", name)
    out(True, job=data)


def _git(*args):
    return run(["git", "-C", str(NETVM_ROOT)] + list(args), timeout=30)


def act_job_put(name):
    check_name(name)
    raw = sys.stdin.read()
    try:
        data = json.loads(raw)
    except Exception as e:
        fail("INVALID_JOB", f"stdin is not valid JSON: {e}")
    if data.get("name") != name:
        fail("NAME_MISMATCH", "path name != body name",
             {"path": name, "body": data.get("name")})
    ok, field, reason = validate_job(data)
    if not ok:
        fail("INVALID_JOB", f"schema validation failed: {field}: {reason}",
             {"field": field, "reason": reason})
    existed = (JOBS_DIR / f"{name}.json").exists()
    JOBS_DIR.mkdir(parents=True, exist_ok=True)
    (JOBS_DIR / f"{name}.json").write_text(json.dumps(data, indent=2) + "\n")
    r = _git("add", f"jobs/{name}.json")
    if r.returncode != 0:
        fail("DISPATCH_FAILED", f"git add failed: {r.stderr.strip()}")
    msg = f"{'Update' if existed else 'Add'} job {name} via box-ctl"
    r = _git("-c", "user.name=box-ctl", "-c", "user.email=box-ctl@bl.local",
             "commit", "-m", msg)
    if r.returncode != 0 and "nothing to commit" not in (r.stdout + r.stderr):
        fail("DISPATCH_FAILED", f"git commit failed: {r.stderr.strip()}")
    audit("job-put", name)
    out(True, name=name, created=not existed, updated=existed)


def act_job_delete(name, force=False):
    check_name(name)
    p = JOBS_DIR / f"{name}.json"
    if not p.exists():
        fail("NOT_FOUND", f"no such job: {name}")
    if unit_path(name, "timer").exists() and not force:
        fail("TIMER_STILL_ACTIVE",
             f"timer job-{name}.timer still exists — delete it first or use --force")
    r = _git("rm", "-q", f"jobs/{name}.json")
    if r.returncode != 0:
        fail("DISPATCH_FAILED", f"git rm failed: {r.stderr.strip()}")
    r = _git("-c", "user.name=box-ctl", "-c", "user.email=box-ctl@bl.local",
             "commit", "-m", f"Delete job {name} via box-ctl")
    if r.returncode != 0:
        fail("DISPATCH_FAILED", f"git commit failed: {r.stderr.strip()}")
    audit("job-delete", name)
    out(True, name=name, deleted=True)


def act_job_trigger(name):
    check_name(name)
    p = JOBS_DIR / f"{name}.json"
    if not p.exists():
        fail("NOT_FOUND", f"no such job: {name}")
    r = run([sys.executable, str(DISPATCHER), name], timeout=300)
    if r.returncode != 0:
        fail("DISPATCH_FAILED", f"job-dispatch.py failed: {(r.stderr or r.stdout).strip()[-500:]}")
    audit("job-trigger", name)
    out(True, name=name, triggered=True)


# ---------------------------------------------------------------------------
# Recursive workflow: box -> agent -> box -> agent ...
# The box stays in the loop: every RESULT recorded here advances the chain
# through the box (shared dedupe state with response-harvester.py).
# ---------------------------------------------------------------------------

CHAINED_JOBS_FILE = BIN / "chained-jobs.json"


def _chain_has(job_id):
    try:
        if CHAINED_JOBS_FILE.exists():
            return job_id in json.loads(CHAINED_JOBS_FILE.read_text())
    except Exception:
        pass
    return False


def _chain_mark(job_id):
    try:
        c = []
        if CHAINED_JOBS_FILE.exists():
            c = json.loads(CHAINED_JOBS_FILE.read_text())
        if job_id not in c:
            c.append(job_id)
            c = c[-1000:]
            CHAINED_JOBS_FILE.write_text(json.dumps(c))
    except Exception:
        pass


def _job_name_from_id(job_id):
    m = re.match(r"^(.*)-(\d{8}-\d{6}-[a-f0-9]{8})$", str(job_id))
    if m:
        return m.group(1)
    parts = str(job_id).split("-")
    if len(parts) < 3:
        return None
    return "-".join(parts[:-2])


def _chain_target(job_name, success):
    """Return (next_job, via_field) for a completed job, or (None, None)."""
    p = JOBS_DIR / f"{job_name}.json"
    if not p.exists():
        return None, None
    try:
        cfg = json.loads(p.read_text())
    except Exception:
        return None, None
    if success:
        nxt = cfg.get("on_success") or cfg.get("chain_next")
        via = "on_success" if cfg.get("on_success") else "chain_next"
    else:
        nxt = cfg.get("on_failure")
        via = "on_failure"
    if nxt and isinstance(nxt, str) and (JOBS_DIR / f"{nxt}.json").exists():
        return nxt, via
    return None, None


def _dispatch_chained(next_job, prev_job_id, prev_result):
    """Dispatch the next job in a chain. Returns (ok, err)."""
    env = os.environ.copy()
    env["CHAIN_PREV_JOB_ID"] = prev_job_id
    env["CHAIN_PREV_RESULT"] = (prev_result or "")[:1000]
    try:
        r = subprocess.run([sys.executable, str(DISPATCHER), next_job],
                           capture_output=True, text=True, timeout=300, env=env)
    except Exception as e:
        return False, f"dispatch exception: {e}"
    if r.returncode != 0:
        return False, (r.stderr or r.stdout).strip()[-500:]
    return True, None


def act_job_result(job_id):
    """Agent -> box: record a RESULT and advance the chain (the recursion step)."""
    raw = sys.stdin.read()
    try:
        data = json.loads(raw)
    except Exception as e:
        fail("INVALID_RESULT", f"stdin is not valid JSON: {e}")
    agent = data.get("agent")
    if agent not in VALID_AGENTS:
        fail("BAD_NAME", f"agent must be one of {sorted(VALID_AGENTS)}")
    success = data.get("success", True)
    if not isinstance(success, bool):
        fail("INVALID_RESULT", "success must be a boolean")
    summary = data.get("summary", "")
    if not isinstance(summary, str) or len(summary) > 2000:
        fail("INVALID_RESULT", "summary must be a string of 0-2000 chars")
    rec = {
        "ts": utcnow(),
        "type": "job_result",
        "job_id": job_id,
        "agent": agent,
        "success": success,
        "result_snippet": summary[:300],
        "thread_id": None,
        "msg_id": None,
        "source": "box-ctl",
    }
    try:
        with open(JOB_LOG, "a") as f:
            f.write(json.dumps(rec) + "\n")
    except Exception as e:
        fail("LOG_FAILED", f"could not append job-log.jsonl: {e}")
    # recursion: the box dispatches the next step
    job_name = _job_name_from_id(job_id)
    chained_to = None
    chain_via = None
    chain_dispatched = False
    chain_error = None
    if job_name and not _chain_has(job_id):
        _chain_mark(job_id)
        nxt, via = _chain_target(job_name, success)
        if nxt:
            try:
                ccfg = json.loads((JOBS_DIR / f"{job_name}.json").read_text())
            except Exception:
                ccfg = {}
            try:
                step_delay = int(ccfg.get("step_delay", 0) or 0)
            except Exception:
                step_delay = 0
            if step_delay > 0:
                time.sleep(min(step_delay, 30))
            ok, err = _dispatch_chained(nxt, job_id, summary)
            chain_dispatched = ok
            chain_error = err
            if ok:
                chained_to = nxt
                chain_via = via
    audit("job-result", job_id)
    out(True, job_id=job_id, agent=agent, success=success, recorded=True,
        chained_to=chained_to, chain_via=chain_via,
        chain_dispatched=chain_dispatched, chain_error=chain_error)


def act_job_status(name):
    """Box-side view of a job's run state: dispatches, results, chain wiring."""
    events = []
    try:
        if JOB_LOG.exists():
            for line in JOB_LOG.read_text().splitlines()[-2000:]:
                try:
                    e = json.loads(line)
                except Exception:
                    continue
                jn = e.get("job_name")
                jid = str(e.get("job_id", ""))
                if jn == name or jid == name or jid.startswith(name + "-"):
                    events.append(e)
    except Exception as e:
        fail("LOG_FAILED", f"could not read job-log.jsonl: {e}")
    dispatches = [e for e in events if e.get("type") == "job_dispatched"]
    results = [e for e in events if e.get("type") == "job_result"]
    failures = [e for e in events if e.get("type") in ("job_failed", "job_timeout")]
    last_result = results[-1] if results else None
    state = "pending"
    if last_result:
        state = "resulted"
    elif dispatches:
        state = "dispatched"
    chain = {}
    p = JOBS_DIR / f"{name}.json"
    if p.exists():
        try:
            cfg = json.loads(p.read_text())
            for k in ("chain_next", "on_success", "on_failure"):
                if cfg.get(k):
                    chain[k] = cfg[k]
        except Exception:
            pass
    audit("job-status", name)
    out(True, name=name, state=state,
        dispatches=len(dispatches), results=len(results), failures=len(failures),
        last_dispatch=dispatches[-1].get("ts") if dispatches else None,
        last_result=({
            "ts": last_result.get("ts"),
            "job_id": last_result.get("job_id"),
            "agent": last_result.get("agent"),
            "success": last_result.get("success"),
            "snippet": (last_result.get("result_snippet") or "")[:200],
        } if last_result else None),
        chain=chain)


def act_job_next(job_id, success=None):
    """Dry-run: what WOULD the box dispatch next for this job_id? No dispatch."""
    job_name = _job_name_from_id(job_id)
    if not job_name:
        fail("BAD_NAME", f"cannot parse job name from job_id: {job_id}")
    if success is None:
        success = True
        try:
            if JOB_LOG.exists():
                for line in reversed(JOB_LOG.read_text().splitlines()[-2000:]):
                    try:
                        e = json.loads(line)
                    except Exception:
                        continue
                    if e.get("type") == "job_result" and str(e.get("job_id")) == job_id:
                        success = bool(e.get("success", True))
                        break
        except Exception:
            pass
    nxt, via = _chain_target(job_name, success)
    already = _chain_has(job_id)
    audit("job-next", job_id)
    out(True, job_id=job_id, job_name=job_name, success=success,
        would_dispatch=bool(nxt) and not already,
        next_job=nxt, via=via, already_chained=already)


def act_job_chain(frm, to, on_failure=False):
    """Wire recursion: set chain_next (or on_failure) from one job to another."""
    check_name(frm)
    check_name(to)
    if frm == to:
        fail("INVALID_JOB", "a job cannot chain to itself")
    fp = JOBS_DIR / f"{frm}.json"
    if not fp.exists():
        fail("NOT_FOUND", f"no such job: {frm}")
    if not (JOBS_DIR / f"{to}.json").exists():
        fail("NOT_FOUND", f"no such job: {to}")
    try:
        cfg = json.loads(fp.read_text())
    except Exception as e:
        fail("INVALID_JOB", f"job JSON unreadable: {e}")
    field = "on_failure" if on_failure else "chain_next"
    cfg[field] = to
    # cycle guard
    seen = {frm}
    cur = to
    while cur:
        if cur in seen:
            fail("INVALID_JOB", f"chain would create a cycle at {cur!r}")
        seen.add(cur)
        try:
            c = json.loads((JOBS_DIR / f"{cur}.json").read_text())
        except Exception:
            break
        cur = c.get("chain_next") or c.get("on_success")
    fp.write_text(json.dumps(cfg, indent=2) + "\n")
    r = _git("add", f"jobs/{frm}.json")
    if r.returncode != 0:
        fail("DISPATCH_FAILED", f"git add failed: {r.stderr.strip()}")
    r = _git("-c", "user.name=box-ctl", "-c", "user.email=box-ctl@bl.local",
             "commit", "-m", f"Chain job {frm} -> {to} via box-ctl")
    if r.returncode != 0 and "nothing to commit" not in (r.stdout + r.stderr):
        fail("DISPATCH_FAILED", f"git commit failed: {r.stderr.strip()}")
    audit("job-chain", f"{frm}->{to}")
    out(True, **{"from": frm, "to": to, "field": field, "chained": True})


def _valid_sidechat_name(name):
    # Sidechat names may contain spaces ("646 tasks"); reject only
    # control characters and enforce a sane length. dm.py resolves
    # the name (alias -> job-sidechats.json -> autoprovision).
    return bool(name) and len(name) <= 64 and not any(
        ord(c) < 32 or ord(c) == 127 for c in name)


def act_notify(agent, message, sidechat=None, allow_main_chat=False, sender=None):
    if agent not in VALID_AGENTS:
        fail("BAD_NAME", f"agent must be one of {sorted(VALID_AGENTS)}")
    if sender is not None and sender not in VALID_AGENTS:
        fail("BAD_NAME", f"sender must be one of {sorted(VALID_AGENTS)}")
    sender_agent = sender if sender else "opm"
    if not message or len(message) > 1000:
        fail("INVALID_JOB", "message must be 1\u20131000 chars")
    # Sidechat-first policy: default to the agent's sidechat, never main.
    if allow_main_chat:
        target = "main"
        extra = ["--allow-main-chat"]
    else:
        target = sidechat if sidechat else NOTIFY_SIDECHATS.get(agent)
        if not target:
            fail("BAD_NAME", f"no default sidechat for agent {agent}; pass --sidechat <name>")
        if sidechat and not _valid_sidechat_name(sidechat):
            fail("BAD_NAME", "sidechat name must be 1-64 chars, no control characters")
        extra = []
    r = run([sys.executable, str(DM_PY), "send",
             "--agent", sender_agent, "--to", agent, "--target", target,
             *extra, message],
            timeout=120)
    if r.returncode != 0:
        fail("DISPATCH_FAILED", f"dm.py send failed: {(r.stderr or r.stdout).strip()[-500:]}")
    audit("notify", agent)
    out(True, agent=agent, target=target, sent=True)


def act_main_loop(sub, agent=None):
    """Run the main-chat self-monitor loop (self_main_loop.py).

    check:   one loop iteration -> JSON {ok, prompted, errors, new_total, agents}
    status:  watermark / last run / prompt-sidechat config, no chat reads.
    enable:  enable the loop (per-agent or all); check skips disabled agents.
    disable: disable the loop (per-agent or all).
    """
    if sub not in ("check", "status", "enable", "disable"):
        fail("BAD_NAME", "usage: main-loop check|status|enable|disable [--agent <name>]")
    if agent is not None and agent not in VALID_AGENTS:
        fail("BAD_NAME", f"agent must be one of {sorted(VALID_AGENTS)}")
    audit("main-loop-" + sub + (f"-{agent}" if agent else ""))
    script = BIN / "self_main_loop.py"
    cmd = [sys.executable, str(script), sub]
    if agent:
        cmd += ["--agent", agent]
    try:
        r = subprocess.run(cmd,
                           capture_output=True, text=True, timeout=600)
    except subprocess.TimeoutExpired:
        fail("LOOP_TIMEOUT", "self_main_loop.py timed out")
    try:
        data = json.loads(r.stdout.strip())
    except Exception:
        fail("LOOP_ERROR", "self_main_loop.py returned non-JSON",
             {"stdout": (r.stdout or "")[-300:], "stderr": (r.stderr or "")[-300:]})
    ok = bool(data.pop("ok", False))
    out(ok, **data)


def act_watchdog_alerts():
    """Check for new failed browser relaunches since the watermark.

    Runs bin/watchdog-alert-check.sh, which exits 0 (no new failures)
    or 1 (new failures printed to stdout) and maintains its own
    watermark at NETVM_ROOT/watchdog-alert-watermark.txt.
    """
    audit("watchdog-alerts")
    script = BIN / "watchdog-alert-check.sh"
    r = subprocess.run([str(script)], capture_output=True, text=True, timeout=30)
    failures = [l for l in (r.stdout or "").splitlines() if l.strip()]
    # Exit 0 = no new failures, exit 1 = new failures found. Any other
    # exit code is a script error.
    if r.returncode not in (0, 1):
        fail("WATCHDOG_CHECK_ERROR", "watchdog-alert-check.sh failed",
             {"stderr": (r.stderr or "").strip()[-500:]})
    out(True, new_failures=len(failures), failures=failures)


def act_fleet_status():
    audit("fleet-status")
    cmd = [sys.executable, str(BIN / "super-cli.py"), "fleet", "status", "--json"]
    r = subprocess.run(cmd, capture_output=True, text=True)
    if r.returncode == 0:
        try:
            data = json.loads(r.stdout)
            print(json.dumps(data))
            return
        except Exception:
            pass
    fail("FLEET_ERROR", "failed to collect fleet data", {"stderr": r.stderr})


def act_relay_health():
    """Run relay-health-check.sh and return JSON results."""
    audit("relay-health")
    script = BIN / "relay-health-check.sh"
    r = subprocess.run([str(script)], capture_output=True, text=True, timeout=60)
    results = {}
    for line in r.stdout.strip().split("\n"):
        if ":" in line:
            name, code = line.split(":", 1)
            results[name.strip()] = code.strip()
    all_healthy = r.returncode == 0
    out(all_healthy, relays=results, healthy=all_healthy)


def act_identity_audit():
    """Run identity-audit-check.sh and return drift as JSON."""
    audit("identity-audit")
    script = BIN / "identity-audit-check.sh"
    r = subprocess.run([str(script)], capture_output=True, text=True, timeout=60)
    if r.returncode == 2:
        fail("AUDIT_UNAVAILABLE", (r.stderr or r.stdout).strip()[:500])
    drift = []
    for line in r.stdout.strip().split("\n"):
        if line.startswith("DRIFT: "):
            drift.append(line[len("DRIFT: "):])
    clean = r.returncode == 0
    out(clean, drift=drift, clean=clean)



def act_cdp_latency():
    """Run cdp-latency-check.sh and return per-node latency JSON."""
    audit("cdp-latency")
    script = BIN / "cdp-latency-check.sh"
    try:
        r = subprocess.run([str(script)], capture_output=True, text=True, timeout=120)
    except subprocess.TimeoutExpired:
        fail("LATENCY_TIMEOUT", "cdp-latency-check.sh timed out")
    nodes = []
    for line in r.stdout.strip().split("\n"):
        parts = line.strip().split(":")
        if len(parts) != 3:
            continue
        name, lat, code = parts
        entry = {"node": name.strip(), "http_code": code.strip()}
        if lat.strip() == "FAIL":
            entry["ok"] = False
            entry["latency_ms"] = None
        else:
            try:
                entry["latency_ms"] = int(lat.strip())
            except ValueError:
                entry["latency_ms"] = None
            entry["ok"] = code.strip() == "200"
        nodes.append(entry)
    all_ok = bool(nodes) and all(n["ok"] for n in nodes)
    out(all_ok, nodes=nodes)

def act_chrome_errors():
    """Run chrome-error-scan.sh and return per-profile error counts as JSON."""
    audit("chrome-errors")
    cmd = ["/bin/bash", str(BIN / "chrome-error-scan.sh"), "--json"]
    r = subprocess.run(cmd, capture_output=True, text=True)
    if r.returncode == 0:
        try:
            data = json.loads(r.stdout)
            out(True, profiles=data)
            return
        except Exception:
            pass
    fail("SCAN_ERROR", "chrome-error-scan.sh failed", {"stderr": r.stderr})


def act_dm_log(limit=50):
    audit("dm-log", str(limit))
    cmd = [sys.executable, str(BIN / "super-cli.py"), "dm", "log", "--json", "-n", str(limit)]
    r = subprocess.run(cmd, capture_output=True, text=True)
    if r.returncode == 0:
        try:
            data = json.loads(r.stdout)
            data["dms"] = data.get("entries", [])
            print(json.dumps(data))
            return
        except Exception:
            pass
    fail("DM_LOG_ERROR", "failed to read dm log", {"stderr": r.stderr})



def _policy_meta():
    """Parse Version:/Date: from CHAT_POLICY.md."""
    version, date = "unknown", "unknown"
    try:
        for line in POLICY_FILE.read_text(encoding="utf-8", errors="replace").splitlines():
            s = line.strip()
            if s.startswith("**Version:**"):
                version = s.split("**Version:**", 1)[1].strip().rstrip("\\")
            elif s.startswith("**Date:**"):
                date = s.split("**Date:**", 1)[1].strip().rstrip("\\")
    except OSError:
        pass
    return version, date


def _watchdog_state():
    """Read the main-chat watchdog watermark (None if missing)."""
    try:
        return json.loads(WATCHDOG_STATE.read_text(encoding="utf-8"))
    except Exception:
        return None


def _policy_scan():
    """Classify dm-log.jsonl events for the sidechat-first policy.

    Mirrors main-chat-watchdog.py classification:
      VIOLATION : type=sent, target=main, no tags.allow_main_chat
      AUTHORIZED: type=sent, target=main, tags.allow_main_chat set
      BLOCKED   : type=main_chat_blocked (gate worked)
    Events before the allow_main_chat audit marker was adopted cannot be
    classified, so the compliance window starts at marker adoption.
    """
    per_agent = {}
    legacy_untagged = 0
    malformed = 0
    adoption_ts = None
    total_scanned = 0

    def bucket(agent):
        return per_agent.setdefault(agent, {
            "blocked": 0,
            "authorized_main": 0,
            "violations": 0,
            "total_sends": 0,
        })

    try:
        with open(DM_LOG, "r", encoding="utf-8", errors="replace") as f:
            lines = f.readlines()
    except OSError:
        return None, {"error": f"cannot read {DM_LOG}"}

    for line in lines:
        line = line.strip()
        if not line:
            continue
        try:
            ev = json.loads(line)
        except json.JSONDecodeError:
            continue
        tags = ev.get("tags")
        if isinstance(tags, dict) and "allow_main_chat" in tags:
            ts = ev.get("ts") or ""
            if adoption_ts is None or ts < adoption_ts:
                adoption_ts = ts

    for line in lines:
        line = line.strip()
        if not line:
            continue
        total_scanned += 1
        try:
            ev = json.loads(line)
        except json.JSONDecodeError:
            malformed += 1
            continue
        ts = ev.get("ts") or ""
        if adoption_ts is not None and ts < adoption_ts:
            if ev.get("type") == "sent" and ev.get("target") == "main":
                legacy_untagged += 1
            continue
        etype = ev.get("type")
        agent = ev.get("agent") or "unknown"
        if etype == "main_chat_blocked":
            bucket(agent)["blocked"] += 1
        elif etype == "sent":
            bucket(agent)["total_sends"] += 1
            if ev.get("target") == "main":
                tags = ev.get("tags") or {}
                if tags.get("allow_main_chat"):
                    bucket(agent)["authorized_main"] += 1
                else:
                    bucket(agent)["violations"] += 1

    return per_agent, {
        "window_start": adoption_ts,
        "events_scanned": total_scanned,
        "malformed": malformed,
        "legacy_untagged_main_sends": legacy_untagged,
    }


def act_policy():
    audit("policy")
    version, date = _policy_meta()
    per_agent, meta = _policy_scan()
    if per_agent is None:
        fail("POLICY_ERROR", "failed to read dm log", meta)
    totals = {"blocked": 0, "authorized_main": 0, "violations": 0, "total_sends": 0}
    for counts in per_agent.values():
        for k in totals:
            totals[k] += counts[k]
    out(True,
        version=version,
        date=date,
        rule=("No DM naturally lands in main chat. Main requires explicit "
              "opt-in (--allow-main-chat / allow_main_chat)."),
        compliance_window=meta,
        watchdog_state=_watchdog_state(),
        agents=per_agent,
        totals=totals,
        status="clean" if totals["violations"] == 0 else "violations found")


def act_policy_check(agent):
    if agent not in VALID_AGENTS:
        fail("BAD_NAME", f"agent must be one of {sorted(VALID_AGENTS)}")
    audit("policy-check", agent)
    per_agent, meta = _policy_scan()
    if per_agent is None:
        fail("POLICY_ERROR", "failed to read dm log", meta)
    counts = per_agent.get(agent, {
        "blocked": 0, "authorized_main": 0, "violations": 0, "total_sends": 0})
    out(True, agent=agent,
        compliance_window=meta, **counts,
        status="clean" if counts["violations"] == 0 else "violations found")


def act_policy_show():
    audit("policy-show")
    try:
        text = POLICY_FILE.read_text(encoding="utf-8", errors="replace")
    except OSError as e:
        fail("POLICY_ERROR", f"cannot read {POLICY_FILE}: {e}")
    version, date = _policy_meta()
    out(True, version=version, date=date, path=str(POLICY_FILE), policy=text)


def act_vars_list():
    try:
        from variables import Variables
        v = Variables()
        audit("vars-list")
        out(True, variables=v.all(), schemas={n: v.schema(n) for n in v.names()})
    except Exception as e:
        fail("VARS_ERROR", str(e))


def act_vars_get(name):
    try:
        from variables import Variables
        v = Variables()
        audit("vars-get", name)
        out(True, name=name, value=v.get(name), schema=v.schema(name))
    except Exception as e:
        fail("VARS_ERROR", str(e))


def act_vars_set(name, val_str):
    try:
        from variables import Variables
        v = Variables()
        spec = v.schema(name)
        vtype = spec.get("type", "str")
        if vtype == "int":
            val = int(val_str)
        elif vtype == "float":
            val = float(val_str)
        elif vtype == "bool":
            val = val_str.lower() in ("true", "1", "yes")
        else:
            val = val_str
        caller = os.environ.get("BOX_CALLER", "box-ctl")
        new_val = v.set(name, val, by=caller)
        audit("vars-set", f"{name}={new_val}")
        out(True, name=name, value=new_val)
    except Exception as e:
        fail("VARS_ERROR", str(e))


def act_vars_reset(name):
    try:
        from variables import Variables
        v = Variables()
        caller = os.environ.get("BOX_CALLER", "box-ctl")
        new_val = v.reset(name, by=caller)
        audit("vars-reset", name)
        out(True, name=name, value=new_val)
    except Exception as e:
        fail("VARS_ERROR", str(e))


def act_vars_history(name=None, limit=20):
    try:
        from variables import Variables
        v = Variables()
        entries = v.history(name=name, limit=limit)
        audit("vars-history", name or "*")
        out(True, history=entries, count=len(entries))
    except Exception as e:
        fail("VARS_ERROR", str(e))


def act_vars_rollback(name, revision=None):
    try:
        from variables import Variables
        v = Variables()
        caller = os.environ.get("BOX_CALLER", "box-ctl")
        new_val = v.rollback(name, revision=revision, by=caller)
        audit("vars-rollback", f"{name}={new_val}")
        out(True, name=name, value=new_val)
    except Exception as e:
        fail("VARS_ERROR", str(e))


def act_strat_list():
    try:
        from modulate import get_all_strategies
        audit("strat-list")
        out(True, strategies=get_all_strategies())
    except Exception as e:
        fail("STRAT_ERROR", str(e))


def act_strat_get(itype, subtype=None, agent=None):
    try:
        from modulate import get_strategy_row, InputType, Priority
        it = InputType(itype.lower())
        row = get_strategy_row(it, subtype.upper() if subtype else None, agent=agent)
        audit("strat-get", f"{itype}:{subtype or '*'}:{agent or '*'}")
        out(True, type=it.value, subtype=subtype, agent=agent, track=row[0],
            priority=row[1].value if isinstance(row[1], Priority) else str(row[1]),
            timeout_s=row[2], nudges=row[3], escalate=row[4])
    except Exception as e:
        fail("STRAT_ERROR", str(e))


def act_strat_set(itype, payload_str=None):
    try:
        from modulate import set_strategy_override
        if not payload_str:
            payload_str = sys.stdin.read()
        data = json.loads(payload_str)
        subtype = data.get("subtype")
        agent = data.get("agent")
        caller = os.environ.get("BOX_CALLER", "box-ctl")
        res = set_strategy_override(
            itype, subtype, agent,
            track=data.get("track"),
            priority=data.get("priority"),
            timeout_s=data.get("timeout_s") or data.get("timeout"),
            nudges=data.get("nudges"),
            escalate=data.get("escalate"),
            by=caller
        )
        audit("strat-set", f"{itype}:{subtype or '*'}:{agent or '*'}")
        out(True, type=itype, subtype=subtype, agent=agent, strategy=res)
    except Exception as e:
        fail("STRAT_ERROR", str(e))


def act_strat_reset(itype, subtype=None, agent=None):
    try:
        from modulate import reset_strategy_override
        caller = os.environ.get("BOX_CALLER", "box-ctl")
        ok = reset_strategy_override(itype, subtype, agent, by=caller)
        audit("strat-reset", f"{itype}:{subtype or '*'}:{agent or '*'}")
        out(True, type=itype, subtype=subtype, agent=agent, reset=ok)
    except Exception as e:
        fail("STRAT_ERROR", str(e))


def act_loop_status(agent=None, limit=20, status=None):
    try:
        from gravity import reconstruct_loops
        loops = reconstruct_loops(limit=limit, agent=agent, status_filter=status)
        audit("loop-status", f"agent={agent or '*'}")
        out(True, loops=loops, count=len(loops))
    except Exception as e:
        fail("LOOP_ERROR", str(e))


def act_loop_health(threshold=None):
    try:
        from gravity import get_fleet_loop_health
        t_val = float(threshold) if threshold is not None else None
        h = get_fleet_loop_health(threshold=t_val)
        audit("loop-health")
        out(True, **h)
    except Exception as e:
        fail("LOOP_ERROR", str(e))


def act_loop_breaks():
    try:
        from gravity import diagnose_breaks
        breaks = diagnose_breaks()
        audit("loop-breaks")
        out(True, breaks=breaks, count=len(breaks))
    except Exception as e:
        fail("LOOP_ERROR", str(e))



# ---------------------------------------------------------------------------
# SSH key minting & management
# ---------------------------------------------------------------------------

KEY_DIR = Path("/home/super/.ssh")
ALLOWED_SIGNERS_PATHS = [
    NETVM_ROOT / "dm-signers" / "allowed_signers",
    Path("/home/super/.exec-signers")
]

def act_ssh_mint(name, force=False):
    """Mint a new ed25519 SSH keypair and register it in allowed_signers."""
    key_name = f"id_{name}"
    priv_path = KEY_DIR / key_name
    pub_path = KEY_DIR / f"{key_name}.pub"

    if priv_path.exists() and not force:
        fail("KEY_EXISTS", f"SSH key {priv_path} already exists -- pass --force to overwrite",
             {"key_path": str(priv_path)})

    if priv_path.exists() and force:
        try:
            priv_path.unlink()
            if pub_path.exists():
                pub_path.unlink()
        except Exception as e:
            fail("IO_ERROR", f"Failed removing old key: {e}")

    cmd = ["ssh-keygen", "-t", "ed25519", "-N", "", "-C", f"{name}@netvm", "-f", str(priv_path)]
    r = subprocess.run(cmd, capture_output=True, text=True)
    if r.returncode != 0:
        fail("KEYGEN_FAILED", f"ssh-keygen failed: {r.stderr.strip()}")

    pub_line = pub_path.read_text(encoding="utf-8").strip()
    pub_key_body = pub_line.split()[1] if len(pub_line.split()) > 1 else pub_line

    # Save to dm-signers/{name}.pub
    ds_pub = NETVM_ROOT / "dm-signers" / f"{name}.pub"
    try:
        ds_pub.write_text(pub_line + "\n", encoding="utf-8")
    except Exception:
        pass

    # Register into signers files
    registered_files = []
    for sp in ALLOWED_SIGNERS_PATHS:
        try:
            lines = sp.read_text(encoding="utf-8").splitlines() if sp.exists() else []
            new_lines = []
            found = False
            for line in lines:
                if not line.strip():
                    continue
                parts = line.split()
                if parts[0] in (name, f"operator-{name}"):
                    new_lines.append(f"{parts[0]} ssh-ed25519 {pub_key_body} {name}@netvm")
                    found = True
                else:
                    new_lines.append(line)
            if not found:
                new_lines.append(f"{name} ssh-ed25519 {pub_key_body} {name}@netvm")
                new_lines.append(f"operator-{name} ssh-ed25519 {pub_key_body} {name}@netvm")

            sp.write_text("\n".join(new_lines) + "\n", encoding="utf-8")
            registered_files.append(str(sp))
        except Exception as e:
            sys.stderr.write(f"Warning: failed updating {sp}: {e}\n")

    audit("ssh-mint", f"name={name} priv={priv_path}")
    out(True, name=name, private_key=str(priv_path), public_key=str(pub_path),
        public_key_content=pub_line, registered_in=registered_files)


def act_ssh_list():
    """List managed SSH keypairs and registrations."""
    keys = []
    if KEY_DIR.exists():
        for p in sorted(KEY_DIR.glob("id_*")):
            if not p.name.endswith(".pub"):
                pub_p = p.with_name(p.name + ".pub")
                pub_content = pub_p.read_text(encoding="utf-8").strip() if pub_p.exists() else None
                keys.append({
                    "name": p.name[3:],
                    "private_key": str(p),
                    "has_public": pub_p.exists(),
                    "public_key": pub_content,
                })
    out(True, keys=keys, count=len(keys))


def act_ssh_show(name):
    """Show public key details and fingerprints for an identity."""
    priv_path = KEY_DIR / f"id_{name}"
    pub_path = KEY_DIR / f"id_{name}.pub"
    if not pub_path.exists():
        fail("NOT_FOUND", f"No public key found for {name} at {pub_path}")

    pub_line = pub_path.read_text(encoding="utf-8").strip()
    cmd = ["ssh-keygen", "-lf", str(pub_path)]
    r = subprocess.run(cmd, capture_output=True, text=True)
    fp = r.stdout.strip() if r.returncode == 0 else ""

    out(True, name=name, private_key=str(priv_path), public_key=str(pub_path),
        public_key_content=pub_line, fingerprint=fp)


def act_ssh_ports():
    """List container reverse tunnel port mappings."""
    import agent_md
    out(True, **agent_md.get_ssh_info())


def act_ssh_info(name):
    """Show SSH dial-in command and reverse tunnel coordinates for a specific agent."""
    import agent_md
    out(True, **agent_md.get_ssh_info(name))


def act_md_audit(accounts=None):
    """Audit markdown files and operational drive across fleet agents."""
    import agent_md
    audit("md-audit", ",".join(accounts or []))
    res = agent_md.audit_agents(accounts=accounts)
    out(True, agents=res)


def act_md_list(account, path=""):
    """List container files via Hatch."""
    import agent_md
    audit("md-list", f"{account}:{path}")
    entries = agent_md.list_files(account, path=path)
    out(True, account=account, path=path, entries=entries)


def act_md_read(account, filename):
    """Read a markdown file from the agent container via Hatch."""
    import agent_md
    audit("md-read", f"{account}:{filename}")
    res = agent_md.read_md(account, filename)
    out(res.pop("ok", True), **res)


def act_md_write(account, filename, content, overwrite=True, append=False):
    """Write content to an agent's container file via Hatch."""
    import agent_md
    audit("md-write", f"{account}:{filename}")
    res = agent_md.write_md(account, filename, content, overwrite=overwrite, append=append)
    out(res.pop("ok", True), **res)


def act_md_diff(account, filename):
    """Diff remote container file against local shared/operators template."""
    import agent_md
    res = agent_md.diff_md(account, filename)
    out(res.pop("ok", True), **res)


def act_md_inject_drive(account, force=False):
    """Inject high-drive operational templates into an agent container via Hatch."""
    import agent_md
    audit("md-inject-drive", account)
    res = agent_md.inject_drive(account, force=force)
    out(res.pop("ok", True), **res)


def act_md_sync_all(force=False):
    """Inject high-drive operational templates across all active agents."""
    import agent_md
    audit("md-sync-all")
    results = {}
    for acct in agent_md.VALID_ACCOUNTS:
        try:
            results[acct] = agent_md.inject_drive(acct, force=force)
        except Exception as e:
            results[acct] = {"ok": False, "error": str(e)}
    out(True, results=results)


def act_loop_resolve(dm_id, note=None):
    f_path = NETVM_ROOT / "followups.json"
    resolved = False
    if f_path.exists():
        try:
            with open(f_path, "r") as f:
                data = json.load(f)
            if dm_id in data:
                data[dm_id]["status"] = "resolved"
                data[dm_id]["resolved_at"] = datetime.now(timezone.utc).isoformat()
                if note:
                    data[dm_id]["note"] = note
                tmp = f"{f_path}.tmp.{os.getpid()}"
                with open(tmp, "w") as f:
                    json.dump(data, f, indent=2)
                os.replace(tmp, f_path)
                resolved = True
        except Exception as e:
            fail("LOOP_ERROR", f"Failed updating followups.json: {e}")

    # Append to job-log.jsonl
    try:
        with open(JOB_LOG, "a") as f:
            f.write(json.dumps({
                "ts": datetime.now(timezone.utc).isoformat(),
                "type": "loop_resolved",
                "dm_id": dm_id,
                "note": note or "manually resolved via box-ctl",
                "caller": os.environ.get("BOX_CALLER", "box-ctl"),
            }) + "\n")
    except Exception:
        pass

    audit("loop-resolve", dm_id)
    out(True, dm_id=dm_id, resolved=resolved)


def act_loop_remediate(dry_run=False):
    try:
        from gravity import remediate_breaks
        res = remediate_breaks(dry_run=dry_run)
        audit("loop-remediate", f"dry_run={dry_run}")
        res.pop("ok", None)
        out(True, **res)
    except Exception as e:
        fail("LOOP_ERROR", str(e))


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

USAGE = """usage: box-ctl.py <action> [args]

fleet:
  fleet-status
  chrome-errors         per-profile chrome FATAL/crash counts (new since watermark)
  watchdog-alerts
  relay-health
  cdp-latency
  identity-audit        VM identity audit drift check
  main-loop check|status|enable|disable [--agent <name>]
                        main-chat self-monitor; enable/disable per agent or all

timer actions:
  timer-list
  timer-status <name>
  timer-create <name>
  timer-delete <name> [--keep-job]
  timer-start|timer-stop|timer-enable|timer-disable <name>

job actions:
  job-list
  job-get <name>
  job-put <name>            (job JSON on stdin)
  job-delete <name> [--force]
  job-trigger <name>
  job-result <job-id>       (result JSON on stdin; records RESULT + advances chain)
  job-status <name-or-id>   (dispatch/result/chain state)
  job-next <job-id> [--success|--fail]  (dry-run: what would box dispatch next)
  job-chain <from> <to> [--on-failure]  (wire chain_next between jobs)

variable actions:
  vars-list
  vars-get <name>
  vars-set <name> <value>
  vars-reset <name>
  vars-history [name] [limit]
  vars-rollback <name> [revision]

strategy actions:
  strat-list
  strat-get <type> [subtype] [--agent AGENT]
  strat-set <type>          (JSON on stdin or as arg)
  strat-reset <type> [subtype] [--agent AGENT]

loop actions:
  loop-status [--agent AGENT] [--limit N] [--status STATUS]
  loop-health [--threshold T]
  loop-breaks
  loop-resolve <dm_id> [note]
  loop-remediate [--dry-run]

thread actions (muse-cli gateway; hyphenated aliases for `thread <sub>`):
  thread-list --agent <agent> [--limit N]
  thread-pin --agent <agent> --thread <id>
  thread-unpin --agent <agent> --thread <id>
  thread-archive --agent <agent> --thread <id> --confirm
  thread-unarchive --agent <agent> --thread <id>
  thread-rename --agent <agent> --thread <id> --title <title>

notify:
  notify <agent> <message> [--sidechat <name>] [--allow-main-chat]
    Send a DM to an agent's default sidechat (never main chat unless
    --allow-main-chat is passed explicitly).

policy:
  policy                  chat policy version, rule, and compliance summary
  policy check <agent>    per-agent compliance detail
  policy show             print the full CHAT_POLICY.md

thread actions (muse-cli gateway):
  thread list <agent> [--limit N]
  thread pin <agent> <thread-id>
  thread unpin <agent> <thread-id>
  thread archive <agent> <thread-id> --confirm
  thread unarchive <agent> <thread-id>

swarm actions (agent swarms; bl owns state, operator agents do the spawning):
  swarm-spawn <count> <task...> [--label L]
  swarm-list
  swarm-status <swarm-id>
  swarm-attach <swarm-id> <slot> <agent-id>
  swarm-report <swarm-id> <slot>      (result JSON on stdin)
  swarm-results <swarm-id>
  swarm-kill <swarm-id> --confirm
  (space-separated aliases: swarm spawn|list|status|attach|report|results|kill)
quality:
  quality-check               run box-ctl self-diagnostics (validators, output
                              contract, audit path, atomic writes)
  quality-validate <action> [args...]
                              dry-run: validate args without executing
  (space-separated alias: quality check|validate)

dm-log:
  dm-log [limit]              recent DM send log"""


def act_thread(op, agent, thread=None, title=None, limit=None, confirm=False):
    """Thread bookkeeping via bin/muse-threads.py (hybrid gateway).

    ops: list | pin | unpin | archive | unarchive | rename
    archive requires confirm=True (the --confirm flag; BatchMode SSH
    cannot prompt, so the flag is the explicit confirmation).
    """
    if op not in ("list", "pin", "unpin", "archive", "unarchive", "rename"):
        fail("BAD_NAME", "usage: thread list|pin|unpin|archive|unarchive|rename <agent> [args]")
    if agent not in VALID_AGENTS:
        fail("BAD_NAME", f"agent must be one of {sorted(VALID_AGENTS)}")
    if op == "list":
        audit("thread-list-" + agent)
    else:
        if not thread or not THREAD_RE.match(thread):
            fail("BAD_NAME", "thread id must match ^[A-Za-z0-9-]{1,64}$",
                 {"field": "thread", "value": thread})
        if op == "archive" and not confirm:
            fail("CONFIRM_REQUIRED",
                 "archive is a visible side effect -- pass --confirm to proceed",
                 {"would_archive": {"agent": agent, "thread": thread}})
        if op == "rename" and not title:
            fail("BAD_NAME", "rename requires a title")
        audit("thread-" + op, f"{agent}/{thread}")
    helper = os.path.join(os.path.dirname(os.path.abspath(__file__)), "muse-threads.py")
    cmd = [sys.executable, helper, op, "--agent", agent]
    if thread:
        cmd += ["--thread", thread]
    if title:
        cmd += ["--title", title]
    try:
        p = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
    except subprocess.TimeoutExpired:
        fail("GATEWAY_UNAVAILABLE", f"muse-threads.py timed out (op={op})")
    except OSError as e:
        fail("GATEWAY_UNAVAILABLE", f"cannot run muse-threads.py: {e}")
    # Helper prints one JSON object; relay it, applying --limit for list
    if p.returncode == 0:
        raw = p.stdout.strip()
        if limit is not None and op == "list" and raw:
            try:
                data = json.loads(raw)
                if isinstance(data.get("threads"), list):
                    data["threads"] = data["threads"][:limit]
                print(json.dumps(data))
                return
            except ValueError:
                pass
        print(raw if raw else '{"ok": true}')
        return
    # Failure: try to extract the helper's JSON error, else wrap
    try:
        err = json.loads(p.stdout.strip() or "{}")
        fail(err.get("code", "GATEWAY_ERROR"), err.get("error", "thread op failed"),
             err.get("detail"))
    except (ValueError, AttributeError):
        fail("GATEWAY_ERROR", f"thread {op} failed",
             (p.stderr.strip() or p.stdout.strip())[:300])


# ---------------------------------------------------------------------------
# Agent swarms — parallel subagent groups tracked on bl.
#
# box-ctl.py owns the swarm STATE (bl is the source of truth). Actual agent
# spawning is done by an operator agent in the container via subagent.spawn;
# the spawner attaches agent IDs with swarm-attach and records outcomes with
# swarm-report. A `swarm-spawner` cron agent can poll `swarm-list` for pending
# swarms and drive them without human involvement.
#
# Lifecycle: pending -> running -> completed|partial|killed
# Slot lifecycle: pending -> running -> done|failed|killed
# ---------------------------------------------------------------------------

SWARM_FILE = NETVM_ROOT / "swarms.json"
SWARM_ID_RE = re.compile(r"^[a-z0-9-]{1,64}$")
SWARM_AGENT_RE = re.compile(r"^[A-Za-z0-9-]{1,64}$")
SWARM_OPS = ("spawn", "list", "status", "attach", "report", "results", "kill", "prune")
SWARM_MAX_COUNT = 50
SWARM_MAX_TASK_LEN = 2000


def _swarm_load():
    try:
        with open(SWARM_FILE) as f:
            data = json.load(f)
            return data if isinstance(data, dict) else {}
    except (OSError, ValueError):
        return {}


def _swarm_save(swarms):
    tmp = str(SWARM_FILE) + ".tmp"
    with open(tmp, "w") as f:
        json.dump(swarms, f, indent=2)
    os.replace(tmp, SWARM_FILE)


def _swarm_new_id():
    return ("sw-" + datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
            + "-" + os.urandom(2).hex())


def _swarm_get(swarms, sid):
    if not sid or not SWARM_ID_RE.match(sid):
        fail("BAD_NAME", "swarm id must match ^[a-z0-9-]{1,64}$",
             {"field": "swarm_id", "value": sid})
    swarm = swarms.get(sid)
    if swarm is None:
        fail("SWARM_NOT_FOUND", "no such swarm: %s" % sid,
             {"swarm_id": sid})
    return swarm


def _swarm_counts(swarm):
    counts = {"pending": 0, "running": 0, "done": 0, "failed": 0,
              "killed": 0}
    for slot in swarm.get("slots", []):
        st = slot.get("status")
        if st in counts:
            counts[st] += 1
    return counts


def _swarm_rollup(swarm):
    """Recompute swarm-level status from slot statuses."""
    counts = _swarm_counts(swarm)
    total = len(swarm.get("slots", []))
    if total and counts["killed"] == total:
        return "killed"
    if total and counts["pending"] + counts["running"] == 0:
        return "completed" if counts["failed"] == 0 else "partial"
    if counts["running"] or counts["done"] or counts["failed"]:
        return "running"
    return "pending"


def _swarm_slot(swarm, slot_str):
    try:
        slot = int(slot_str)
    except (TypeError, ValueError):
        fail("BAD_SLOT", "slot must be an integer", {"value": slot_str})
    slots = swarm.get("slots", [])
    if not 0 <= slot < len(slots):
        fail("BAD_SLOT", "slot must be 0-%d" % (len(slots) - 1),
             {"value": slot_str})
    return slots[slot]


def act_swarm_spawn(count_str, task, label=None):
    try:
        count = int(count_str)
    except (TypeError, ValueError):
        fail("BAD_COUNT", "count must be an integer 1-50",
             {"value": count_str})
    if not 1 <= count <= SWARM_MAX_COUNT:
        fail("BAD_COUNT", "count must be 1-%d" % SWARM_MAX_COUNT,
             {"value": count})
    if not task or not task.strip():
        fail("BAD_TASK", "task must be a non-empty string")
    if len(task) > SWARM_MAX_TASK_LEN:
        fail("BAD_TASK", "task must be <= %d chars" % SWARM_MAX_TASK_LEN,
             {"length": len(task)})
    if label is not None and not SWARM_ID_RE.match(label):
        fail("BAD_NAME", "label must match ^[a-z0-9-]{1,64}$",
             {"field": "label", "value": label})
    swarms = _swarm_load()
    sid = _swarm_new_id()
    while sid in swarms:  # vanishingly unlikely; be safe
        sid = _swarm_new_id()
    now = utcnow()
    swarm = {
        "swarm_id": sid,
        "task": task,
        "count": count,
        "label": label,
        "status": "pending",
        "created_ts": now,
        "created_by": os.environ.get("BOX_CALLER", "unknown"),
        "updated_ts": now,
        "slots": [
            {"slot": i, "agent_id": None, "status": "pending",
             "result": None, "updated_ts": now}
            for i in range(count)
        ],
    }
    swarms[sid] = swarm
    _swarm_save(swarms)
    audit("swarm-spawn", sid)
    out(True, swarm_id=sid, count=count, status="pending", label=label)


def act_swarm_list():
    swarms = _swarm_load()
    items = []
    for sid, swarm in sorted(
            swarms.items(), key=lambda kv: kv[1].get("created_ts", ""),
            reverse=True):
        counts = _swarm_counts(swarm)
        item = {
            "swarm_id": sid,
            "label": swarm.get("label"),
            "status": swarm.get("status"),
            "count": swarm.get("count"),
            "task_preview": (swarm.get("task") or "")[:80],
            "created_ts": swarm.get("created_ts"),
        }
        item.update(counts)
        items.append(item)
    out(True, swarms=items)


def act_swarm_status(sid):
    swarms = _swarm_load()
    swarm = _swarm_get(swarms, sid)
    out(True, swarm=swarm, counts=_swarm_counts(swarm))


def act_swarm_attach(sid, slot_str, agent_id):
    if not agent_id or not SWARM_AGENT_RE.match(agent_id):
        fail("BAD_NAME", "agent id must match ^[A-Za-z0-9-]{1,64}$",
             {"field": "agent_id", "value": agent_id})
    swarms = _swarm_load()
    swarm = _swarm_get(swarms, sid)
    if swarm.get("status") in ("killed", "completed", "partial"):
        fail("SWARM_CLOSED", "swarm is %s; cannot attach" % swarm["status"],
             {"swarm_id": sid})
    slot_rec = _swarm_slot(swarm, slot_str)
    if slot_rec["status"] != "pending":
        fail("SWARM_SLOT_BUSY",
             "slot %d is %s, not pending" % (slot_rec["slot"],
                                             slot_rec["status"]),
             {"swarm_id": sid, "slot": slot_rec["slot"]})
    slot_rec["agent_id"] = agent_id
    slot_rec["status"] = "running"
    slot_rec["updated_ts"] = utcnow()
    swarm["status"] = _swarm_rollup(swarm)
    swarm["updated_ts"] = utcnow()
    _swarm_save(swarms)
    audit("swarm-attach", "%s/%d" % (sid, slot_rec["slot"]))
    out(True, swarm_id=sid, slot=slot_rec["slot"], agent_id=agent_id,
        status="running")


def act_swarm_report(sid, slot_str):
    swarms = _swarm_load()
    swarm = _swarm_get(swarms, sid)
    slot_rec = _swarm_slot(swarm, slot_str)
    if slot_rec["status"] in ("done", "failed", "killed"):
        fail("SWARM_SLOT_CLOSED",
             "slot %d already %s" % (slot_rec["slot"], slot_rec["status"]),
             {"swarm_id": sid, "slot": slot_rec["slot"]})
    raw = sys.stdin.read()
    ok = True
    result = None
    if raw.strip():
        try:
            payload = json.loads(raw)
            if isinstance(payload, dict):
                ok = bool(payload.get("ok", True))
                result = payload.get("result", payload)
            else:
                result = payload
        except ValueError:
            result = raw[:4000]  # plain-text result
    slot_rec["status"] = "done" if ok else "failed"
    slot_rec["result"] = result
    slot_rec["updated_ts"] = utcnow()
    swarm["status"] = _swarm_rollup(swarm)
    swarm["updated_ts"] = utcnow()
    _swarm_save(swarms)
    audit("swarm-report", "%s/%d" % (sid, slot_rec["slot"]))
    out(True, swarm_id=sid, slot=slot_rec["slot"],
        status=slot_rec["status"], swarm_status=swarm["status"])


def act_swarm_results(sid):
    swarms = _swarm_load()
    swarm = _swarm_get(swarms, sid)
    results = [
        {"slot": s["slot"], "agent_id": s.get("agent_id"),
         "status": s["status"], "result": s.get("result")}
        for s in swarm.get("slots", [])
    ]
    out(True, swarm_id=sid, status=swarm.get("status"),
        counts=_swarm_counts(swarm), results=results)


def act_swarm_prune(stale_hours=6, confirm=False):
    import datetime
    swarms = _swarm_load()
    cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(hours=stale_hours)
    pruned = []
    
    for sid, swarm in swarms.items():
        st = swarm.get("status")
        up_ts = swarm.get("updated_ts") or swarm.get("created_ts")
        if not up_ts:
            continue
        try:
            dt = datetime.datetime.fromisoformat(up_ts.replace("Z", "+00:00"))
        except Exception:
            continue
            
        if dt < cutoff and st in ("partial", "completed", "killed"):
            pruned.append(sid)
            
    if not confirm:
        fail("CONFIRM_REQUIRED",
             f"prune transitions {len(pruned)} stale swarms to 'archived' -- pass --confirm to proceed",
             {"stale_hours": stale_hours, "matching_count": len(pruned), "sample": pruned[:10]})

    now = utcnow()
    archived_count = 0
    for sid in pruned:
        swarms[sid]["status"] = "archived"
        swarms[sid]["updated_ts"] = now
        archived_count += 1
        
    _swarm_save(swarms)
    audit("swarm-prune", f"stale_hours={stale_hours} count={archived_count}")
    out(True, archived_count=archived_count, stale_hours=stale_hours)


def act_swarm_kill(sid, confirm=False):
    swarms = _swarm_load()
    swarm = _swarm_get(swarms, sid)
    if not confirm:
        fail("CONFIRM_REQUIRED",
             "kill terminates a swarm -- pass --confirm to proceed",
             {"would_kill": {"swarm_id": sid,
                             "active_slots": _swarm_counts(swarm)}})
    now = utcnow()
    killed_slots = 0
    for s in swarm.get("slots", []):
        if s["status"] in ("pending", "running"):
            s["status"] = "killed"
            s["updated_ts"] = now
            killed_slots += 1
    counts = _swarm_counts(swarm)
    # All slots are terminal now: killed if nothing ever completed/failed,
    # otherwise partial (some work finished before the kill).
    if counts["done"] + counts["failed"] > 0:
        swarm["status"] = "partial"
    else:
        swarm["status"] = "killed"
    swarm["updated_ts"] = now
    _swarm_save(swarms)
    audit("swarm-kill", sid)
    out(True, swarm_id=sid, status=swarm["status"],
        killed_slots=killed_slots)


# ---------------------------------------------------------------------------
# Quality functions — input validation, output contract, idempotency helpers,
# and self-diagnostics.
#
#   quality check                    run self-diagnostics (validators, output
#                                    contract, audit path, atomic writes)
#   quality validate <action> [...]  dry-run: validate args for <action>
#                                    without executing it
#   (hyphenated aliases: quality-check, quality-validate)
#
# Conventions:
# - qv_*() validators are NON-FATAL: they return check dicts, never exit.
#   (The fatal check_*() helpers above are for live actions; qv_*() are for
#   pre-flight validation and diagnostics.)
# - A check dict is {"check": name, "ok": bool, "code": str|None,
#   "reason": str|None}.
# - Output contract: every response is {"ok": bool, ...}; ok=false responses
#   MUST carry "code" (a member of KNOWN_ERROR_CODES) and "error".
# - Idempotency: mutating actions append state and are NOT implicitly
#   retry-safe. q_atomic_write() is the primitive for retry-safe writes
#   (tmp file + os.replace); IDEMPOTENT_ACTIONS lists the read-only/dry-run
#   actions that are safe to retry.
# ---------------------------------------------------------------------------

QUALITY_VERSION = "1"

KNOWN_ERROR_CODES = frozenset([
    "ALREADY_EXISTS", "AUDIT_UNAVAILABLE", "BAD_ARGS", "BAD_COUNT",
    "BAD_LIMIT", "BAD_NAME", "BAD_SLOT", "BAD_TASK", "CONFIRM_REQUIRED",
    "DISPATCH_FAILED", "DM_LOG_ERROR", "FLEET_ERROR", "GATEWAY_ERROR",
    "GATEWAY_UNAVAILABLE", "INVALID_JOB", "INVALID_RESULT",
    "INVALID_SCHEDULE", "LATENCY_TIMEOUT", "LOG_FAILED", "LOOP_ERROR",
    "LOOP_TIMEOUT", "NAME_MISMATCH", "NOT_FOUND", "POLICY_ERROR",
    "SCAN_ERROR", "STRAT_ERROR", "SWARM_CLOSED", "SWARM_NOT_FOUND",
    "SWARM_SLOT_BUSY", "SWARM_SLOT_CLOSED", "TIMER_CREATE_FAILED",
    "TIMER_STILL_ACTIVE", "VARS_ERROR", "WATCHDOG_CHECK_ERROR",
])

# Read-only / dry-run actions: safe to retry.
IDEMPOTENT_ACTIONS = frozenset([
    "fleet-status", "watchdog-alerts", "relay-health", "cdp-latency",
    "chrome-errors", "identity-audit", "timer-list", "timer-status",
    "job-list", "job-get", "job-status", "job-next", "vars-list",
    "vars-get", "vars-history", "strat-list", "strat-get", "loop-status",
    "loop-health", "loop-breaks", "policy", "policy-check", "policy-show",
    "thread-list", "dm-log", "swarm-list", "swarm-status", "swarm-results",
    "quality-check", "quality-validate", "main-loop",
])

JOB_ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,127}$")
SWARM_ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$")
AGENT_REF_RE = re.compile(r"^[A-Za-z0-9_.-]{1,128}$")
DM_ID_RE = re.compile(r"^[0-9a-fA-F]{6,64}$")


def _qcheck(name, ok, code=None, reason=None):
    d = {"check": name, "ok": bool(ok)}
    if code:
        d["code"] = code
    if reason:
        d["reason"] = reason
    return d


def qv_name(value, field="name"):
    if not value or not NAME_RE.match(value):
        return _qcheck(field, False, "BAD_NAME",
                       "must match ^[a-z0-9-]{1,64}$")
    return _qcheck(field, True)


def qv_agent(value):
    if value not in VALID_AGENTS:
        return _qcheck("agent", False, "BAD_NAME",
                       "unknown agent %r; valid: %s" % (value, sorted(VALID_AGENTS)))
    return _qcheck("agent", True)


def qv_thread_id(value):
    if not value or not THREAD_RE.match(value):
        return _qcheck("thread_id", False, "BAD_NAME",
                       "must match ^[A-Za-z0-9-]{1,64}$")
    return _qcheck("thread_id", True)


def qv_job_id(value):
    if not value or not JOB_ID_RE.match(value):
        return _qcheck("job_id", False, "BAD_NAME",
                       "must match ^[a-z0-9][a-z0-9-]{0,127}$")
    return _qcheck("job_id", True)


def qv_swarm_id(value):
    if not value or not SWARM_ID_RE.match(value):
        return _qcheck("swarm_id", False, "BAD_NAME",
                       "must match ^[a-z0-9][a-z0-9-]{0,63}$")
    return _qcheck("swarm_id", True)


def qv_agent_ref(value, field="agent_id"):
    # Looser than qv_agent: accepts subagent UUIDs / external refs.
    if not value or not AGENT_REF_RE.match(value):
        return _qcheck(field, False, "BAD_NAME",
                       "must match ^[A-Za-z0-9_.-]{1,128}$")
    return _qcheck(field, True)


def qv_dm_id(value):
    if not value or not DM_ID_RE.match(value):
        return _qcheck("dm_id", False, "BAD_NAME",
                       "must be 6-64 hex chars")
    return _qcheck("dm_id", True)


def qv_int_range(value, lo, hi, field, code):
    try:
        n = int(str(value))
    except (TypeError, ValueError):
        return _qcheck(field, False, code, "not an integer: %r" % (value,))
    if not (lo <= n <= hi):
        return _qcheck(field, False, code, "must be %d..%d" % (lo, hi))
    return _qcheck(field, True)


def qv_count(value, lo=1, hi=50):
    return qv_int_range(value, lo, hi, "count", "BAD_COUNT")


def qv_limit(value):
    return qv_int_range(value, 1, 10000, "limit", "BAD_LIMIT")


def qv_slot(value):
    return qv_int_range(value, 0, 9999, "slot", "BAD_SLOT")


def qv_nonempty(value, field):
    if value is None or not str(value).strip():
        return _qcheck(field, False, "BAD_ARGS", "%s must be non-empty" % field)
    return _qcheck(field, True)


def qv_title(value):
    c = qv_nonempty(value, "title")
    if not c["ok"]:
        return c
    if len(str(value)) > 200:
        return _qcheck("title", False, "BAD_ARGS", "title must be <= 200 chars")
    return _qcheck("title", True)


def qv_name_or_job_id(value, field="name_or_job_id"):
    c1 = qv_name(value, field)
    if c1["ok"]:
        return c1
    c2 = qv_job_id(value)
    if c2["ok"]:
        c2["check"] = field
        return c2
    return _qcheck(field, False, "BAD_NAME",
                   "must be a job name ^[a-z0-9-]{1,64}$ or job id "
                   "^[a-z0-9][a-z0-9-]{0,127}$")


def q_output_contract(payload):
    """Validate a box-ctl response dict against the JSON output contract.
    Returns (ok_bool, issues_list)."""
    issues = []
    if not isinstance(payload, dict):
        return False, ["payload is not a JSON object"]
    if "ok" not in payload:
        issues.append("missing 'ok'")
    elif not isinstance(payload["ok"], bool):
        issues.append("'ok' is not a boolean")
    if payload.get("ok") is False:
        if "code" not in payload:
            issues.append("ok:false missing 'code'")
        elif payload["code"] not in KNOWN_ERROR_CODES:
            issues.append("unknown error code %r" % (payload["code"],))
        if "error" not in payload:
            issues.append("ok:false missing 'error'")
    return (not issues), issues


def q_atomic_write(path, text):
    """Idempotency-safe file write: write temp + os.replace (atomic).
    Returns (ok_bool, reason). Cleans up the temp file on failure."""
    p = Path(path)
    tmp = p.parent / (p.name + ".tmp")
    try:
        tmp.write_text(text)
        os.replace(tmp, p)
        return True, ""
    except OSError as e:
        try:
            if tmp.exists():
                tmp.unlink()
        except OSError:
            pass
        return False, str(e)


def _qsrc():
    return Path(__file__).read_text()


def act_quality_check():
    """Run box-ctl self-diagnostics. Read-only (no state mutations)."""
    results = []

    def rec(name, ok, detail=None):
        r = {"check": name, "ok": bool(ok)}
        if detail is not None:
            r["detail"] = detail
        results.append(r)
        return ok

    src = _qsrc()

    # 1. validator functions present and callable
    for fn in ("qv_name", "qv_agent", "qv_thread_id", "qv_job_id",
               "qv_swarm_id", "qv_agent_ref", "qv_dm_id", "qv_count",
               "qv_limit", "qv_slot", "qv_nonempty", "qv_title",
               "qv_name_or_job_id", "q_output_contract", "q_atomic_write"):
        rec("validator:%s" % fn, callable(globals().get(fn)))

    # 2. every fail() code is a member of KNOWN_ERROR_CODES (typo guard)
    codes = set(re.findall(r'fail\(\s*"([A-Z0-9_]+)"', src))
    unknown = sorted(codes - KNOWN_ERROR_CODES)
    rec("error_codes_known", not unknown,
        {"codes_found": len(codes), "unknown": unknown})

    # 3. every dispatched action is documented in USAGE
    actions = set(re.findall(r'(?:el)?if action == "([a-z0-9-]+)"', src))
    for tup in re.findall(r'elif action in \((.*?)\)', src, re.DOTALL):
        actions.update(re.findall(r'"([a-z0-9-]+)"', tup))
    missing = sorted(a for a in actions if a not in USAGE)
    rec("usage_coverage", not missing,
        {"actions": len(actions), "undocumented": missing})

    # 4. output contract on the failure path (live subprocess probe)
    try:
        r = run([sys.executable, str(Path(__file__)),
                 "quality-probe-bogus-action"], timeout=30)
        payload = json.loads(r.stdout)
        okc, issues = q_output_contract(payload)
        rec("output_contract:fail-path",
            okc and payload.get("code") == "BAD_NAME", {"issues": issues})
    except Exception as e:
        rec("output_contract:fail-path", False, {"error": str(e)})

    # 5. output contract on the success path (live subprocess probe)
    try:
        r = run([sys.executable, str(Path(__file__)), "swarm-list"], timeout=30)
        payload = json.loads(r.stdout)
        okc, issues = q_output_contract(payload)
        rec("output_contract:ok-path",
            okc and payload.get("ok") is True, {"issues": issues})
    except Exception as e:
        rec("output_contract:ok-path", False, {"error": str(e)})

    # 6. audit log path writable
    try:
        rec("audit_path_writable", os.access(CTL_LOG.parent, os.W_OK),
            {"log": str(CTL_LOG)})
    except Exception as e:
        rec("audit_path_writable", False, {"error": str(e)})

    # 7. atomic-write round trip (idempotency primitive)
    import tempfile
    tmpd = tempfile.mkdtemp(prefix="qcheck-")
    tp = os.path.join(tmpd, "probe.json")
    okw, why = q_atomic_write(tp, '{"probe": true}')
    leftover = os.path.exists(tp + ".tmp")
    try:
        content_ok = okw and open(tp).read() == '{"probe": true}'
    except OSError:
        content_ok = False
    try:
        if os.path.exists(tp):
            os.unlink(tp)
        os.rmdir(tmpd)
    except OSError:
        pass
    rec("atomic_write_roundtrip", content_ok and not leftover,
        {"write_error": why or None})

    # 8. state directories exist and are writable
    for label, p in (("netvm_root", NETVM_ROOT), ("jobs_dir", JOBS_DIR)):
        rec("state_dir:%s" % label, p.is_dir() and os.access(p, os.W_OK),
            {"path": str(p)})

    passed = sum(1 for r_ in results if r_["ok"])
    failed = len(results) - passed
    out(True, quality_version=QUALITY_VERSION, passed=passed,
        failed=failed, checks=results,
        idempotent_actions=sorted(IDEMPOTENT_ACTIONS))


def _qv_flag(args, flag, takes_value=False):
    """Split flag out of an arg list copy.
    Returns (present, value_or_None, remaining_args)."""
    a = list(args)
    if flag not in a:
        return False, None, a
    i = a.index(flag)
    if takes_value:
        if i + 1 < len(a) and not a[i + 1].startswith("--"):
            return True, a[i + 1], a[:i] + a[i + 2:]
        return True, None, a[:i] + a[i + 1:]
    return True, None, a[:i] + a[i + 1:]


def _qv_no_unknown(args, usage):
    """All remaining args must be consumed; anything left is unknown."""
    if args:
        return [_qcheck("argv", False, "BAD_ARGS",
                        "unknown arguments %r; usage: %s" % (args, usage))]
    return []


def _qv_args(action, rest):
    """Return a list of check dicts for (action, rest), or None if the
    action is unknown to quality-validate."""
    a = list(rest)

    # -- timer -----------------------------------------------------------
    if action in ("timer-status", "timer-create", "timer-start",
                  "timer-stop", "timer-enable", "timer-disable"):
        if len(a) != 1:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: %s <name>" % action)]
        return [qv_name(a[0])]
    if action == "timer-delete":
        _, _, b = _qv_flag(a, "--keep-job")
        if len(b) != 1:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: timer-delete <name> [--keep-job]")]
        return [qv_name(b[0])]
    if action == "timer-list":
        if a:
            return [_qcheck("argv", False, "BAD_ARGS", "usage: timer-list")]
        return [_qcheck("argv", True)]

    # -- job -------------------------------------------------------------
    if action in ("job-get", "job-trigger"):
        if len(a) != 1:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: %s <name>" % action)]
        return [qv_name(a[0])]
    if action == "job-list":
        if a:
            return [_qcheck("argv", False, "BAD_ARGS", "usage: job-list")]
        return [_qcheck("argv", True)]
    if action == "job-put":
        if len(a) != 1:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: job-put <name>  (job JSON on stdin)")]
        return [qv_name(a[0]),
                _qcheck("stdin", True, reason="job JSON payload read from "
                        "stdin at execution; not consumed by dry-run")]
    if action == "job-delete":
        _, _, b = _qv_flag(a, "--force")
        if len(b) != 1:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: job-delete <name> [--force]")]
        return [qv_name(b[0])]
    if action == "job-result":
        if len(a) != 1:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: job-result <job-id>  (result JSON on stdin)")]
        return [qv_job_id(a[0]),
                _qcheck("stdin", True, reason="result JSON payload read from "
                        "stdin at execution; not consumed by dry-run")]
    if action == "job-status":
        if len(a) != 1:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: job-status <name-or-job-id>")]
        return [qv_name_or_job_id(a[0])]
    if action == "job-next":
        succ, _, b = _qv_flag(a, "--success")
        failf, _, b = _qv_flag(b, "--fail")
        if len(b) != 1:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: job-next <job-id> [--success|--fail]")]
        return [qv_job_id(b[0])]
    if action == "job-chain":
        _, _, b = _qv_flag(a, "--on-failure")
        if len(b) != 2:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: job-chain <from-job> <to-job> [--on-failure]")]
        return [qv_name(b[0], "from_job"), qv_name(b[1], "to_job")]

    # -- notify ----------------------------------------------------------
    if action == "notify":
        allow, _, b = _qv_flag(a, "--allow-main-chat")
        has_sc, sc_val, b = _qv_flag(b, "--sidechat", takes_value=True)
        if has_sc and sc_val is None:
            return [_qcheck("sidechat", False, "BAD_ARGS",
                            "--sidechat requires a value")]
        if len(b) != 2:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: notify <agent> <message> [--sidechat <name>] "
                            "[--allow-main-chat]")]
        checks = [qv_agent(b[0]), qv_nonempty(b[1], "message")]
        if has_sc:
            checks.append(qv_nonempty(sc_val, "sidechat"))
        return checks

    # -- thread (hyphenated) ---------------------------------------------
    if action in ("thread-list", "thread-pin", "thread-unpin",
                  "thread-archive", "thread-unarchive", "thread-rename"):
        op = action[len("thread-"):]
        has_agent, agent, b = _qv_flag(a, "--agent", takes_value=True)
        has_thread, thread, b = _qv_flag(b, "--thread", takes_value=True)
        has_title, title, b = _qv_flag(b, "--title", takes_value=True)
        has_limit, limit, b = _qv_flag(b, "--limit", takes_value=True)
        has_confirm, _, b = _qv_flag(b, "--confirm")
        checks = _qv_no_unknown(b, "%s --agent <agent> [...]" % action)
        if has_agent and agent is None:
            checks.append(_qcheck("agent", False, "BAD_ARGS",
                                  "--agent requires a value"))
        elif not has_agent:
            checks.append(_qcheck("agent", False, "BAD_NAME",
                                  "--agent is required"))
        else:
            checks.append(qv_agent(agent))
        if op == "list":
            if has_thread:
                checks.append(_qcheck("thread", False, "BAD_ARGS",
                                      "--thread does not apply to thread-list"))
            if has_limit:
                checks.append(qv_limit(limit) if limit is not None
                              else _qcheck("limit", False, "BAD_ARGS",
                                           "--limit requires a value"))
            if has_confirm:
                checks.append(_qcheck("confirm", False, "BAD_ARGS",
                                      "--confirm only applies to thread-archive"))
        else:
            if has_limit:
                checks.append(_qcheck("limit", False, "BAD_ARGS",
                                      "--limit only applies to thread-list"))
            if has_confirm and op != "archive":
                checks.append(_qcheck("confirm", False, "BAD_ARGS",
                                      "--confirm only applies to thread-archive"))
            if not has_thread or thread is None:
                checks.append(_qcheck("thread_id", False, "BAD_NAME",
                                      "--thread <id> is required"))
            else:
                checks.append(qv_thread_id(thread))
            if op == "archive" and not has_confirm:
                checks.append(_qcheck("confirm", False, "CONFIRM_REQUIRED",
                                      "archive requires --confirm"))
            if op == "rename":
                if not has_title or title is None:
                    checks.append(_qcheck("title", False, "BAD_ARGS",
                                          "--title is required for rename"))
                else:
                    checks.append(qv_title(title))
        return checks

    # -- thread (space-separated) ----------------------------------------
    if action == "thread":
        if not a or a[0] not in THREAD_OPS:
            return [_qcheck("sub", False, "BAD_NAME",
                            "usage: thread list|pin|unpin|archive|unarchive "
                            "<agent> [...]")]
        sub, b = a[0], a[1:]
        has_confirm, _, b = _qv_flag(b, "--confirm")
        has_limit, limit, b = _qv_flag(b, "--limit", takes_value=True)
        checks = []
        if not b:
            return [_qcheck("argv", False, "BAD_NAME",
                            "usage: thread %s <agent> [...]" % sub)]
        checks.append(qv_agent(b[0]))
        pos = b[1:]
        if sub == "list":
            if pos:
                checks.append(_qcheck("argv", False, "BAD_ARGS",
                                      "usage: thread list <agent> [--limit N]"))
            if has_limit:
                checks.append(qv_limit(limit) if limit is not None
                              else _qcheck("limit", False, "BAD_ARGS",
                                           "--limit requires a value"))
            if has_confirm:
                checks.append(_qcheck("confirm", False, "BAD_ARGS",
                                      "--confirm only applies to thread archive"))
        else:
            if has_limit:
                checks.append(_qcheck("limit", False, "BAD_ARGS",
                                      "--limit only applies to thread list"))
            if has_confirm and sub != "archive":
                checks.append(_qcheck("confirm", False, "BAD_ARGS",
                                      "--confirm only applies to thread archive"))
            if len(pos) != 1:
                checks.append(_qcheck("argv", False, "BAD_NAME",
                                      "usage: thread %s <agent> <thread-id> "
                                      "[--confirm]" % sub))
            else:
                checks.append(qv_thread_id(pos[0]))
            if sub == "archive" and not has_confirm:
                checks.append(_qcheck("confirm", False, "CONFIRM_REQUIRED",
                                      "archive requires --confirm"))
        checks.extend(_qv_no_unknown(
            [x for x in b if x.startswith("--")],
            "thread %s <agent> [...]" % sub))
        return checks

    # -- swarm (hyphenated) ----------------------------------------------
    if action == "swarm-spawn":
        has_label, label, b = _qv_flag(a, "--label", takes_value=True)
        checks = []
        if has_label and label is None:
            checks.append(_qcheck("label", False, "BAD_ARGS",
                                  "--label requires a value"))
        for x in b:
            if x.startswith("--"):
                checks.append(_qcheck("argv", False, "BAD_ARGS",
                                      "usage: swarm-spawn <count> <task...> "
                                      "[--label L]"))
                return checks
        if len(b) < 2:
            checks.append(_qcheck("argv", False, "BAD_ARGS",
                                  "usage: swarm-spawn <count> <task...> "
                                  "[--label L]"))
            return checks
        checks.append(qv_count(b[0]))
        checks.append(qv_nonempty(" ".join(b[1:]), "task"))
        return checks
    if action == "swarm-list":
        if a:
            return [_qcheck("argv", False, "BAD_ARGS", "usage: swarm-list")]
        return [_qcheck("argv", True)]
    if action in ("swarm-status", "swarm-results"):
        if len(a) != 1:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: %s <swarm-id>" % action)]
        return [qv_swarm_id(a[0])]
    if action == "swarm-attach":
        if len(a) != 3:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: swarm-attach <swarm-id> <slot> <agent-id>")]
        return [qv_swarm_id(a[0]), qv_slot(a[1]), qv_agent_ref(a[2])]
    if action == "swarm-report":
        if len(a) != 2:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: swarm-report <swarm-id> <slot> "
                            "(result JSON on stdin)")]
        return [qv_swarm_id(a[0]), qv_slot(a[1]),
                _qcheck("stdin", True, reason="result JSON payload read from "
                        "stdin at execution; not consumed by dry-run")]
    if action == "swarm-kill":
        has_confirm, _, b = _qv_flag(a, "--confirm")
        if len(b) != 1:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: swarm-kill <swarm-id> --confirm")]
        checks = [qv_swarm_id(b[0])]
        if not has_confirm:
            checks.append(_qcheck("confirm", False, "CONFIRM_REQUIRED",
                                  "kill requires --confirm"))
        return checks

    # -- swarm (space-separated) -----------------------------------------
    if action == "swarm":
        if not a or a[0] not in SWARM_OPS:
            return [_qcheck("sub", False, "BAD_NAME",
                            "usage: swarm spawn|list|status|attach|report|"
                            "results|kill [...]")]
        return _qv_args("swarm-" + a[0], a[1:])

    # -- job (space-separated) -------------------------------------------
    if action == "job":
        if not a or a[0] not in ("result", "status", "next", "chain"):
            return [_qcheck("sub", False, "BAD_NAME",
                            "usage: job result|status|next|chain [...]")]
        return _qv_args("job-" + a[0], a[1:])

    # -- vars ------------------------------------------------------------
    if action in ("vars-get", "vars-reset"):
        if len(a) != 1:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: %s <name>" % action)]
        return [qv_name(a[0])]
    if action == "vars-list":
        if a:
            return [_qcheck("argv", False, "BAD_ARGS", "usage: vars-list")]
        return [_qcheck("argv", True)]
    if action == "vars-set":
        if len(a) != 2:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: vars-set <name> <value>")]
        return [qv_name(a[0]), _qcheck("value", True)]
    if action == "vars-history":
        checks = []
        b = list(a)
        if b and not b[0].isdigit():
            checks.append(qv_name(b.pop(0), "name"))
        if b:
            if len(b) != 1:
                return [_qcheck("argv", False, "BAD_ARGS",
                                "usage: vars-history [name] [limit]")]
            checks.append(qv_limit(b[0]))
        return checks or [_qcheck("argv", True)]
    if action == "vars-rollback":
        if not (1 <= len(a) <= 2):
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: vars-rollback <name> [revision]")]
        checks = [qv_name(a[0])]
        if len(a) == 2:
            checks.append(qv_nonempty(a[1], "revision"))
        return checks

    # -- strategy --------------------------------------------------------
    if action == "strat-list":
        if a:
            return [_qcheck("argv", False, "BAD_ARGS", "usage: strat-list")]
        return [_qcheck("argv", True)]
    if action in ("strat-get", "strat-reset"):
        has_agent, agent, b = _qv_flag(a, "--agent", takes_value=True)
        checks = []
        if has_agent:
            if agent is None:
                checks.append(_qcheck("agent", False, "BAD_ARGS",
                                      "--agent requires a value"))
            else:
                checks.append(qv_agent(agent))
        if not b:
            checks.append(_qcheck("argv", False, "BAD_ARGS",
                                  "usage: %s <type> [subtype] [--agent AGENT]"
                                  % action))
        else:
            checks.append(qv_nonempty(b[0], "type"))
            if len(b) > 1:
                checks.append(qv_nonempty(b[1], "subtype"))
            if len(b) > 2:
                checks.append(_qcheck("argv", False, "BAD_ARGS",
                                      "usage: %s <type> [subtype] [--agent AGENT]"
                                      % action))
        return checks
    if action == "strat-set":
        if not a:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: strat-set <type> [JSON]")]
        return [qv_nonempty(a[0], "type"),
                _qcheck("stdin", True, reason="JSON payload may come from "
                        "stdin at execution; not consumed by dry-run")]

    # -- loop ------------------------------------------------------------
    if action == "loop-status":
        has_agent, agent, b = _qv_flag(a, "--agent", takes_value=True)
        has_limit, limit, b = _qv_flag(b, "--limit", takes_value=True)
        has_status, status, b = _qv_flag(b, "--status", takes_value=True)
        checks = _qv_no_unknown(b, "loop-status [--agent A] [--limit N] "
                                   "[--status S]")
        if has_agent:
            checks.append(qv_agent(agent) if agent is not None
                          else _qcheck("agent", False, "BAD_ARGS",
                                       "--agent requires a value"))
        if has_limit:
            checks.append(qv_limit(limit) if limit is not None
                          else _qcheck("limit", False, "BAD_ARGS",
                                       "--limit requires a value"))
        if has_status:
            checks.append(qv_nonempty(status, "status") if status is not None
                          else _qcheck("status", False, "BAD_ARGS",
                                       "--status requires a value"))
        return checks
    if action == "loop-health":
        if len(a) > 1:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: loop-health [--threshold T]")]
        if not a:
            return [_qcheck("argv", True)]
        try:
            float(a[0])
            return [_qcheck("threshold", True)]
        except (TypeError, ValueError):
            return [_qcheck("threshold", False, "BAD_ARGS",
                            "threshold must be numeric")]
    if action == "loop-breaks":
        if a:
            return [_qcheck("argv", False, "BAD_ARGS", "usage: loop-breaks")]
        return [_qcheck("argv", True)]
    if action == "loop-resolve":
        if not (1 <= len(a) <= 2):
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: loop-resolve <dm_id> [note]")]
        checks = [qv_dm_id(a[0])]
        if len(a) == 2:
            checks.append(_qcheck("note", True))
        return checks
    if action == "loop-remediate":
        _, _, b = _qv_flag(a, "--dry-run")
        if b:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: loop-remediate [--dry-run]")]
        return [_qcheck("argv", True)]

    # -- policy / fleet / misc -------------------------------------------
    if action == "policy":
        if not a:
            return [_qcheck("argv", True)]
        if a[0] == "check":
            if len(a) != 2:
                return [_qcheck("argv", False, "BAD_ARGS",
                                "usage: policy check <agent>")]
            return [qv_agent(a[1])]
        if a[0] == "show":
            if len(a) != 1:
                return [_qcheck("argv", False, "BAD_ARGS",
                                "usage: policy show")]
            return [_qcheck("argv", True)]
        return [_qcheck("argv", False, "BAD_NAME",
                        "usage: policy [check <agent>|show]")]
    if action in ("fleet-status", "watchdog-alerts", "relay-health",
                  "cdp-latency", "chrome-errors", "identity-audit"):
        if a:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: %s (no arguments)" % action)]
        return [_qcheck("argv", True)]
    if action == "main-loop":
        if not a or a[0] not in ("check", "status", "enable", "disable"):
            return [_qcheck("sub", False, "BAD_NAME",
                            "usage: main-loop check|status|enable|disable "
                            "[--agent <name>]")]
        has_agent, agent, b = _qv_flag(a[1:], "--agent", takes_value=True)
        checks = _qv_no_unknown(b, "main-loop %s [--agent <name>]" % a[0])
        if has_agent:
            checks.append(qv_agent(agent) if agent is not None
                          else _qcheck("agent", False, "BAD_ARGS",
                                       "--agent requires a value"))
        return checks
    if action == "dm-log":
        if len(a) > 1:
            return [_qcheck("argv", False, "BAD_ARGS", "usage: dm-log [limit]")]
        if not a:
            return [_qcheck("argv", True)]
        return [qv_limit(a[0])]

    # -- quality (self) ----------------------------------------------------
    if action == "quality-check":
        if a:
            return [_qcheck("argv", False, "BAD_ARGS", "usage: quality-check")]
        return [_qcheck("argv", True)]
    if action == "quality-validate":
        if not a:
            return [_qcheck("argv", False, "BAD_ARGS",
                            "usage: quality-validate <action> [args...]")]
        inner = _qv_args(a[0], a[1:])
        if inner is None:
            return [_qcheck("action", False, "BAD_NAME",
                            "unknown action %r" % (a[0],))]
        bad = [c for c in inner if not c["ok"]]
        code = bad[0].get("code") if bad else None
        return [_qcheck("nested:%s" % a[0], not bad, code=code,
                        reason="nested validation of %s" % a[0])]
    if action == "quality":
        if not a or a[0] not in ("check", "validate"):
            return [_qcheck("sub", False, "BAD_NAME",
                            "usage: quality check|validate [...]")]
        return _qv_args("quality-" + a[0], a[1:])

    return None


def act_quality_validate(action, rest):
    """Dry-run: validate argv for <action> without executing it."""
    checks = _qv_args(action, rest)
    if checks is None:
        fail("BAD_NAME", "quality validate: unknown action %r" % (action,))
    valid = all(c["ok"] for c in checks)
    out(True, action=action, valid=valid, checks=checks,
        note="stdin payloads are never consumed by dry-run validation")

def main(argv):
    if len(argv) < 2:
        print(USAGE, file=sys.stderr)
        sys.exit(2)
    action = argv[1]
    rest = argv[2:]

    if action == "timer-list":
        act_timer_list()
    elif action == "timer-status":
        if len(rest) != 1:
            fail("BAD_NAME", "usage: timer-status <name>")
        act_timer_status(rest[0])
    elif action == "timer-create":
        if len(rest) != 1:
            fail("BAD_NAME", "usage: timer-create <name>")
        act_timer_create(rest[0])
    elif action == "timer-delete":
        if not rest or len(rest) > 2:
            fail("BAD_NAME", "usage: timer-delete <name> [--keep-job]")
        keep = "--keep-job" in rest[1:]
        act_timer_delete(rest[0], keep_job=keep)
    elif action in ("timer-start", "timer-stop", "timer-enable", "timer-disable"):
        if len(rest) != 1:
            fail("BAD_NAME", f"usage: {action} <name>")
        act_timer_control(rest[0], action[len("timer-"):])
    elif action == "timer":
        if not rest:
            act_timer_list()
        else:
            sub = rest[0]
            if sub in ("list",):
                act_timer_list()
            elif sub in ("status",):
                if len(rest) != 2:
                    fail("BAD_NAME", "usage: timer status <name>")
                act_timer_status(rest[1])
            elif sub in ("create",):
                if len(rest) != 2:
                    fail("BAD_NAME", "usage: timer create <name>")
                act_timer_create(rest[1])
            elif sub in ("delete",):
                keep = "--keep-job" in rest[2:]
                act_timer_delete(rest[1], keep_job=keep)
            elif sub in ("start", "stop", "enable", "disable"):
                if len(rest) != 2:
                    fail("BAD_NAME", f"usage: timer {sub} <name>")
                act_timer_control(rest[1], sub)
            else:
                fail("BAD_NAME", "usage: timer list|status|create|delete|start|stop|enable|disable [...]")
    elif action == "job-list":
        act_job_list()
    elif action == "job-get":
        if len(rest) != 1:
            fail("BAD_NAME", "usage: job-get <name>")
        act_job_get(rest[0])
    elif action == "job-put":
        if len(rest) != 1:
            fail("BAD_NAME", "usage: job-put <name>  (JSON on stdin)")
        act_job_put(rest[0])
    elif action == "job-delete":
        if not rest or len(rest) > 2:
            fail("BAD_NAME", "usage: job-delete <name> [--force]")
        force = "--force" in rest[1:]
        act_job_delete(rest[0], force=force)
    elif action == "job-trigger":
        if len(rest) != 1:
            fail("BAD_NAME", "usage: job-trigger <name>")
        act_job_trigger(rest[0])
    elif action == "job-result":
        if len(rest) != 1:
            fail("BAD_NAME", "usage: job-result <job-id>  (result JSON on stdin)")
        act_job_result(rest[0])
    elif action == "job-status":
        if len(rest) != 1:
            fail("BAD_NAME", "usage: job-status <name-or-job-id>")
        act_job_status(rest[0])
    elif action == "job-next":
        if not rest or len(rest) > 2:
            fail("BAD_NAME", "usage: job-next <job-id> [--success|--fail]")
        s = None
        if "--success" in rest[1:]:
            s = True
        if "--fail" in rest[1:]:
            s = False
        act_job_next(rest[0], s)
    elif action == "job-chain":
        if len(rest) < 2 or len(rest) > 3:
            fail("BAD_NAME", "usage: job-chain <from-job> <to-job> [--on-failure]")
        act_job_chain(rest[0], rest[1], on_failure=("--on-failure" in rest[2:]))
    elif action == "vars-list":
        act_vars_list()
    elif action == "vars-get":
        if len(rest) != 1:
            fail("BAD_NAME", "usage: vars-get <name>")
        act_vars_get(rest[0])
    elif action == "vars-set":
        if len(rest) != 2:
            fail("BAD_NAME", "usage: vars-set <name> <value>")
        act_vars_set(rest[0], rest[1])
    elif action == "vars-reset":
        if len(rest) != 1:
            fail("BAD_NAME", "usage: vars-reset <name>")
        act_vars_reset(rest[0])
    elif action == "vars-history":
        name = None
        limit = 20
        if rest:
            name = rest[0]
        if len(rest) > 1:
            try:
                limit = int(rest[1])
            except ValueError:
                limit = 20
        act_vars_history(name=name, limit=limit)
    elif action == "vars-rollback":
        if len(rest) < 1 or len(rest) > 2:
            fail("BAD_NAME", "usage: vars-rollback <name> [revision]")
        rev = rest[1] if len(rest) > 1 else None
        act_vars_rollback(rest[0], revision=rev)
    elif action == "strat-list":
        act_strat_list()
    elif action == "strat-get":
        if not rest:
            fail("BAD_NAME", "usage: strat-get <type> [subtype] [--agent AGENT]")
        itype = rest[0]
        subtype = None
        agent = None
        idx = 1
        while idx < len(rest):
            if rest[idx] == "--agent" and idx + 1 < len(rest):
                agent = rest[idx + 1]
                idx += 2
            elif subtype is None and not rest[idx].startswith("--"):
                subtype = rest[idx]
                idx += 1
            else:
                idx += 1
        act_strat_get(itype, subtype=subtype, agent=agent)
    elif action == "strat-set":
        if len(rest) < 1:
            fail("BAD_NAME", "usage: strat-set <type> [JSON]")
        payload = rest[1] if len(rest) > 1 else None
        act_strat_set(rest[0], payload)
    elif action == "strat-reset":
        if not rest:
            fail("BAD_NAME", "usage: strat-reset <type> [subtype] [--agent AGENT]")
        itype = rest[0]
        subtype = None
        agent = None
        idx = 1
        while idx < len(rest):
            if rest[idx] == "--agent" and idx + 1 < len(rest):
                agent = rest[idx + 1]
                idx += 2
            elif subtype is None and not rest[idx].startswith("--"):
                subtype = rest[idx]
                idx += 1
            else:
                idx += 1
        act_strat_reset(itype, subtype=subtype, agent=agent)
    elif action == "loop-status":
        agent = None
        limit = 20
        status = None
        idx = 0
        while idx < len(rest):
            if rest[idx] == "--agent" and idx + 1 < len(rest):
                agent = rest[idx + 1]
                idx += 2
            elif rest[idx] == "--limit" and idx + 1 < len(rest):
                limit = int(rest[idx + 1])
                idx += 2
            elif rest[idx] == "--status" and idx + 1 < len(rest):
                status = rest[idx + 1]
                idx += 2
            else:
                idx += 1
        act_loop_status(agent=agent, limit=limit, status=status)
    elif action == "loop-health":
        thresh = rest[0] if rest else None
        act_loop_health(threshold=thresh)
    elif action == "loop-breaks":
        act_loop_breaks()
    elif action == "loop-resolve":
        if len(rest) < 1:
            fail("BAD_NAME", "usage: loop-resolve <dm_id> [note]")
        note = rest[1] if len(rest) > 1 else None
        act_loop_resolve(rest[0], note=note)
    elif action in ("ssh", "ssh-mint", "ssh-list", "ssh-show", "ssh-ports", "ssh-info"):
        if action == "ssh-mint":
            if not rest:
                fail("BAD_ARGS", "usage: ssh-mint <name> [--force]")
            act_ssh_mint(rest[0], force=("--force" in rest[1:]))
        elif action == "ssh-list":
            act_ssh_list()
        elif action == "ssh-show":
            if not rest:
                fail("BAD_ARGS", "usage: ssh-show <name>")
            act_ssh_show(rest[0])
        elif action == "ssh-ports":
            act_ssh_ports()
        elif action == "ssh-info":
            act_ssh_info(rest[0] if rest else None)
        elif action == "ssh":
            if not rest or rest[0] not in ("mint", "list", "show", "ports", "info"):
                fail("BAD_NAME", "usage: ssh mint|list|show|ports|info [...]")
            sub = rest[0]
            args = rest[1:]
            if sub == "mint":
                if not args:
                    fail("BAD_ARGS", "usage: ssh mint <name> [--force]")
                act_ssh_mint(args[0], force=("--force" in args[1:]))
            elif sub == "list":
                act_ssh_list()
            elif sub == "show":
                if not args:
                    fail("BAD_ARGS", "usage: ssh show <name>")
                act_ssh_show(args[0])
            elif sub == "ports":
                act_ssh_ports()
            elif sub == "info":
                act_ssh_info(args[0] if args else None)
    elif action in ("md", "md-audit", "md-list", "md-read", "md-write", "md-diff", "md-inject-drive", "md-sync-all"):
        if action == "md-audit":
            act_md_audit(accounts=rest or None)
        elif action == "md-list":
            if not rest:
                fail("BAD_ARGS", "usage: md-list <account> [path]")
            act_md_list(rest[0], path=rest[1] if len(rest) > 1 else "")
        elif action == "md-read":
            if len(rest) < 2:
                fail("BAD_ARGS", "usage: md-read <account> <filename>")
            act_md_read(rest[0], rest[1])
        elif action == "md-write":
            if len(rest) < 3:
                fail("BAD_ARGS", "usage: md-write <account> <filename> <content>")
            act_md_write(rest[0], rest[1], rest[2])
        elif action == "md-diff":
            if len(rest) < 2:
                fail("BAD_ARGS", "usage: md-diff <account> <filename>")
            act_md_diff(rest[0], rest[1])
        elif action == "md-inject-drive":
            if not rest:
                fail("BAD_ARGS", "usage: md-inject-drive <account> [--force]")
            act_md_inject_drive(rest[0], force=("--force" in rest[1:]))
        elif action == "md-sync-all":
            act_md_sync_all(force=("--force" in rest))
        elif action == "md":
            if not rest:
                fail("BAD_NAME", "usage: md audit|list|read|write|diff|inject-drive|sync-all [...]")
            sub = rest[0]
            args = rest[1:]
            if sub == "audit":
                act_md_audit(accounts=args or None)
            elif sub == "list":
                if not args:
                    fail("BAD_ARGS", "usage: md list <account> [path]")
                act_md_list(args[0], path=args[1] if len(args) > 1 else "")
            elif sub == "read":
                if len(args) < 2:
                    fail("BAD_ARGS", "usage: md read <account> <filename>")
                act_md_read(args[0], args[1])
            elif sub == "write":
                if len(args) < 2:
                    fail("BAD_ARGS", "usage: md write <account> <filename> [--content text | --file path]")
                content = None
                if "--file" in args:
                    fidx = args.index("--file")
                    if fidx + 1 < len(args):
                        content = Path(args[fidx + 1]).read_text(encoding="utf-8")
                elif "--content" in args:
                    cidx = args.index("--content")
                    if cidx + 1 < len(args):
                        content = args[cidx + 1]
                elif len(args) >= 3:
                    content = args[2]
                if content is None:
                    fail("BAD_ARGS", "usage: md write <account> <filename> <content>")
                act_md_write(args[0], args[1], content)
            elif sub == "diff":
                if len(args) < 2:
                    fail("BAD_ARGS", "usage: md diff <account> <filename>")
                act_md_diff(args[0], args[1])
            elif sub == "inject-drive":
                if not args:
                    fail("BAD_ARGS", "usage: md inject-drive <account> [--force]")
                act_md_inject_drive(args[0], force=("--force" in args[1:]))
            elif sub == "sync-all":
                act_md_sync_all(force=("--force" in args))
            else:
                fail("BAD_NAME", f"unknown md subcommand: {sub}")
    elif action == "loop-remediate":
        dry = "--dry-run" in rest
        act_loop_remediate(dry_run=dry)
    elif action == "notify":
        # notify <agent> <message> [--sidechat <name>] [--allow-main-chat] [--sender <agent>]
        args = list(rest)
        allow_main = False
        sidechat = None
        sender = None
        if "--allow-main-chat" in args:
            allow_main = True
            args.remove("--allow-main-chat")
        if "--sidechat" in args:
            i = args.index("--sidechat")
            if i + 1 >= len(args):
                fail("BAD_NAME", "usage: notify <agent> <message> [--sidechat <name>] [--allow-main-chat] [--sender <agent>]")
            sidechat = args[i + 1]
            del args[i:i + 2]
        if "--sender" in args:
            i = args.index("--sender")
            if i + 1 >= len(args):
                fail("BAD_NAME", "usage: notify <agent> <message> [--sidechat <name>] [--allow-main-chat] [--sender <agent>]")
            sender = args[i + 1]
            del args[i:i + 2]
        if len(args) != 2:
            fail("BAD_NAME", "usage: notify <agent> <message> [--sidechat <name>] [--allow-main-chat] [--sender <agent>]")
        act_notify(args[0], args[1], sidechat=sidechat, allow_main_chat=allow_main, sender=sender)
    elif action == "policy":
        if not rest:
            act_policy()
        elif rest[0] == "check":
            if len(rest) != 2:
                fail("BAD_NAME", "usage: policy check <agent>")
            act_policy_check(rest[1])
        elif rest[0] == "show":
            if len(rest) != 1:
                fail("BAD_NAME", "usage: policy show")
            act_policy_show()
        else:
            fail("BAD_NAME", "usage: policy [check <agent>|show]")
    elif action == "fleet-status":
        act_fleet_status()
    elif action == "watchdog-alerts":
        if rest:
            fail("BAD_NAME", "usage: watchdog-alerts")
        act_watchdog_alerts()
    elif action == "relay-health":
        act_relay_health()
    elif action == "identity-audit":
        if rest:
            fail("BAD_ARGS", "usage: identity-audit")
        act_identity_audit()
    elif action == "cdp-latency":
        act_cdp_latency()
    elif action == "chrome-errors":
        if rest:
            fail("BAD_ARGS", "usage: chrome-errors")
        act_chrome_errors()
    elif action == "main-loop":
        if not rest or rest[0] not in ("check", "status", "enable", "disable"):
            fail("BAD_NAME", "usage: main-loop check|status|enable|disable [--agent <name>]")
        sub = rest[0]
        agent = None
        args = rest[1:]
        if "--agent" in args:
            i = args.index("--agent")
            if i + 1 >= len(args):
                fail("BAD_ARGS", "usage: main-loop %s [--agent <name>]" % sub)
            agent = args[i + 1]
            args = args[:i] + args[i + 2:]
        if args:
            fail("BAD_ARGS", "usage: main-loop %s [--agent <name>]" % sub)
        act_main_loop(sub, agent)
    elif action == "thread":
        if not rest or rest[0] not in THREAD_OPS:
            fail("BAD_NAME", "usage: thread list|pin|unpin|archive|unarchive <agent> [args]")
        sub = rest[0]
        args = rest[1:]
        if not args:
            fail("BAD_NAME", f"usage: thread {sub} <agent> [...]")
        agent = args[0]
        args = args[1:]
        thread_id = None
        limit = None
        confirm = "--confirm" in args
        args = [a for a in args if a != "--confirm"]
        if confirm and sub != "archive":
            fail("BAD_ARGS", "--confirm only applies to thread archive")
        idx = 0
        while idx < len(args):
            if args[idx] == "--limit" and idx + 1 < len(args):
                try:
                    limit = int(args[idx + 1])
                except ValueError:
                    fail("BAD_LIMIT", "usage: thread list <agent> [--limit N]")
                idx += 2
            elif thread_id is None and not args[idx].startswith("--"):
                thread_id = args[idx]
                idx += 1
            else:
                fail("BAD_ARGS", f"usage: thread {sub} <agent> [...]")
        if limit is not None and sub != "list":
            fail("BAD_ARGS", "--limit only applies to thread list")
        if sub in ("pin", "unpin", "archive", "unarchive") and not thread_id:
            fail("BAD_NAME", f"usage: thread {sub} <agent> <thread-id> [--confirm]")
        if sub == "list" and thread_id:
            fail("BAD_ARGS", "usage: thread list <agent> [--limit N]")
        act_thread(sub, agent, thread=thread_id, limit=limit, confirm=confirm)
    elif action in ("thread-list", "thread-pin", "thread-unpin",
                    "thread-archive", "thread-unarchive", "thread-rename"):
        op = action[len("thread-"):]
        agent = None
        thread_id = None
        title = None
        limit = None
        confirm = "--confirm" in rest
        args = [a for a in rest if a != "--confirm"]
        if confirm and op != "archive":
            fail("BAD_ARGS", "--confirm only applies to thread-archive")
        idx = 0
        while idx < len(args):
            if args[idx] == "--agent" and idx + 1 < len(args):
                agent = args[idx + 1]
                idx += 2
            elif args[idx] == "--thread" and idx + 1 < len(args):
                thread_id = args[idx + 1]
                idx += 2
            elif args[idx] == "--title" and idx + 1 < len(args):
                title = args[idx + 1]
                idx += 2
            elif args[idx] == "--limit" and idx + 1 < len(args):
                try:
                    limit = int(args[idx + 1])
                except ValueError:
                    fail("BAD_LIMIT", "usage: thread-list --agent <agent> [--limit N]")
                idx += 2
            else:
                fail("BAD_ARGS", f"usage: {action} --agent <agent> [--thread <id>] [--title <t>] [--confirm] [--limit N]")
        if not agent:
            fail("BAD_NAME", f"usage: {action} --agent <agent> [...]")
        if limit is not None and op != "list":
            fail("BAD_ARGS", "--limit only applies to thread-list")
        if op in ("pin", "unpin", "archive", "unarchive", "rename") and not thread_id:
            fail("BAD_NAME", f"usage: {action} --agent <agent> --thread <id> [...]")
        act_thread(op, agent, thread=thread_id, title=title, limit=limit, confirm=confirm)
    elif action in ("swarm-spawn", "swarm-list", "swarm-status",
                     "swarm-attach", "swarm-report", "swarm-results",
                     "swarm-kill", "swarm-prune"):
        op = action[len("swarm-"):]
        if op == "spawn":
            if len(rest) < 2:
                fail("BAD_ARGS", "usage: swarm-spawn <count> <task...> [--label L]")
            label = None
            pargs = list(rest[1:])
            if "--label" in pargs:
                i = pargs.index("--label")
                if i + 1 >= len(pargs):
                    fail("BAD_ARGS", "usage: swarm-spawn <count> <task...> [--label L]")
                label = pargs[i + 1]
                del pargs[i:i + 2]
            for a in pargs:
                if a.startswith("--"):
                    fail("BAD_ARGS", "usage: swarm-spawn <count> <task...> [--label L]")
            task = " ".join(pargs)
            if not task.strip():
                fail("BAD_TASK", "task must be a non-empty string")
            act_swarm_spawn(rest[0], task, label)
        elif op == "list":
            if rest:
                fail("BAD_ARGS", "usage: swarm-list")
            act_swarm_list()
        elif op == "status":
            if len(rest) != 1:
                fail("BAD_ARGS", "usage: swarm-status <swarm-id>")
            act_swarm_status(rest[0])
        elif op == "attach":
            if len(rest) != 3:
                fail("BAD_ARGS", "usage: swarm-attach <swarm-id> <slot> <agent-id>")
            act_swarm_attach(rest[0], rest[1], rest[2])
        elif op == "report":
            if len(rest) != 2:
                fail("BAD_ARGS", "usage: swarm-report <swarm-id> <slot>  (result JSON on stdin)")
            act_swarm_report(rest[0], rest[1])
        elif op == "results":
            if len(rest) != 1:
                fail("BAD_ARGS", "usage: swarm-results <swarm-id>")
            act_swarm_results(rest[0])
        elif op == "kill":
            if not rest or len(rest) > 2:
                fail("BAD_ARGS", "usage: swarm-kill <swarm-id> --confirm")
            act_swarm_kill(rest[0], confirm=("--confirm" in rest[1:]))
    elif action == "swarm":
        if not rest or rest[0] not in SWARM_OPS:
            fail("BAD_NAME", "usage: swarm spawn|list|status|attach|report|results|kill [...]")
        sub = rest[0]
        args = rest[1:]
        if sub == "spawn":
            if len(args) < 2:
                fail("BAD_ARGS", "usage: swarm spawn <count> <task...> [--label L]")
            label = None
            pargs = list(args[1:])
            if "--label" in pargs:
                i = pargs.index("--label")
                if i + 1 >= len(pargs):
                    fail("BAD_ARGS", "usage: swarm spawn <count> <task...> [--label L]")
                label = pargs[i + 1]
                del pargs[i:i + 2]
            for a in pargs:
                if a.startswith("--"):
                    fail("BAD_ARGS", "usage: swarm spawn <count> <task...> [--label L]")
            task = " ".join(pargs)
            if not task.strip():
                fail("BAD_TASK", "task must be a non-empty string")
            act_swarm_spawn(args[0], task, label)
        elif sub == "list":
            if args:
                fail("BAD_ARGS", "usage: swarm list")
            act_swarm_list()
        elif sub == "status":
            if len(args) != 1:
                fail("BAD_ARGS", "usage: swarm status <swarm-id>")
            act_swarm_status(args[0])
        elif sub == "attach":
            if len(args) != 3:
                fail("BAD_ARGS", "usage: swarm attach <swarm-id> <slot> <agent-id>")
            act_swarm_attach(args[0], args[1], args[2])
        elif sub == "report":
            if len(args) != 2:
                fail("BAD_ARGS", "usage: swarm report <swarm-id> <slot>  (result JSON on stdin)")
            act_swarm_report(args[0], args[1])
        elif sub == "results":
            if len(args) != 1:
                fail("BAD_ARGS", "usage: swarm results <swarm-id>")
            act_swarm_results(args[0])
        elif sub == "kill":
            if not args or len(args) > 2:
                fail("BAD_ARGS", "usage: swarm kill <swarm-id> --confirm")
            act_swarm_kill(args[0], confirm=("--confirm" in args[1:]))
        elif sub == "prune":
            stale_h = 6
            confirm = "--confirm" in args
            if "--stale-hours" in args:
                i = args.index("--stale-hours")
                if i + 1 < len(args):
                    try:
                        stale_h = float(args[i+1])
                    except ValueError:
                        pass
            act_swarm_prune(stale_hours=stale_h, confirm=confirm)
    elif action == "job":
        if not rest or rest[0] not in ("result", "status", "next", "chain"):
            fail("BAD_NAME", "usage: job result|status|next|chain [...]")
        sub = rest[0]
        args = rest[1:]
        if sub == "result":
            if len(args) != 1:
                fail("BAD_ARGS", "usage: job result <job-id>  (result JSON on stdin)")
            act_job_result(args[0])
        elif sub == "status":
            if len(args) != 1:
                fail("BAD_ARGS", "usage: job status <name-or-job-id>")
            act_job_status(args[0])
        elif sub == "next":
            if not args or len(args) > 2:
                fail("BAD_ARGS", "usage: job next <job-id> [--success|--fail]")
            s = None
            if "--success" in args[1:]:
                s = True
            if "--fail" in args[1:]:
                s = False
            act_job_next(args[0], s)
        elif sub == "chain":
            if len(args) < 2 or len(args) > 3:
                fail("BAD_ARGS", "usage: job chain <from-job> <to-job> [--on-failure]")
            act_job_chain(args[0], args[1], on_failure=("--on-failure" in args[2:]))
    elif action in ("quality-check", "quality-validate"):
        op = action[len("quality-"):]
        if op == "check":
            if rest:
                fail("BAD_ARGS", "usage: quality-check")
            audit("quality-check")
            act_quality_check()
        else:
            if not rest:
                fail("BAD_ARGS", "usage: quality-validate <action> [args...]")
            audit("quality-validate", rest[0])
            act_quality_validate(rest[0], rest[1:])
    elif action == "quality":
        if not rest or rest[0] not in ("check", "validate"):
            fail("BAD_NAME", "usage: quality check|validate [...]")
        if rest[0] == "check":
            if len(rest) != 1:
                fail("BAD_ARGS", "usage: quality check")
            audit("quality-check")
            act_quality_check()
        else:
            if len(rest) < 2:
                fail("BAD_ARGS", "usage: quality validate <action> [args...]")
            audit("quality-validate", rest[1])
            act_quality_validate(rest[1], rest[2:])
    elif action == "dm-log":
        limit = 50
        if rest:
            try:
                limit = int(rest[0])
            except ValueError:
                fail("BAD_LIMIT", "usage: dm-log [limit]")
        act_dm_log(limit=limit)
    else:
        print(USAGE, file=sys.stderr)
        fail("BAD_NAME", f"unknown action: {action}")


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