Add box-ctl.py: allowlisted bl helper for Box API mutations\n\n- 14 actions: timer-list/status/create/delete/start/stop/enable/disable,\n job-list/get/put/delete/trigger, notify\n- Name regex ^[a-z0-9-]{1,64}$, full job schema validation per design §6\n- Cron→OnCalendar conversion + systemd-analyze verification\n- Fixed unit templates (only validated name interpolated)\n- Git commits on job put/delete; audit log to box-ctl.jsonl\n- No shell=True, no string interpolation into commands\n- Tested: full lifecycle on bl (boxtest job)

bin/box-ctl.py

@@ -0,0 +1,695 @@
+#!/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
+from datetime import datetime, timezone
+from pathlib import Path
+
+NETVM_ROOT = Path("/home/super/Projects/NetVM")
+BIN = NETVM_ROOT / "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"
+
+NAME_RE = re.compile(r"^[a-z0-9-]{1,64}$")
+VALID_AGENTS = {"muse", "pip", "646", "opm"}
+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 H * * * with stepped hour: M */n * * * -> 0/n:MM
+    if m.isdigit() and h.startswith("*/") and h[2:].isdigit() and all(is_star(f) for f in (dom, mon, dow)):
+        return f"0/{h[2:]}:{int(m):02d}"
+    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"
+    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")
+
+    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)
+
+
+def act_notify(agent, message):
+    if agent not in VALID_AGENTS:
+        fail("BAD_NAME", f"agent must be one of {sorted(VALID_AGENTS)}")
+    if not message or len(message) > 1000:
+        fail("INVALID_JOB", "message must be 1–1000 chars")
+    # dm.py send --agent opm --to <agent> --target main "<message>"
+    r = run([sys.executable, str(DM_PY), "send",
+             "--agent", "opm", "--to", agent, "--target", "main", 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, sent=True)
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+USAGE = """usage: box-ctl.py <action> [args]
+
+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>
+
+notify:
+  notify <agent> <message>"""
+
+
+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 == "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 == "notify":
+        if len(rest) != 2:
+            fail("BAD_NAME", "usage: notify <agent> <message>")
+        act_notify(rest[0], rest[1])
+    else:
+        print(USAGE, file=sys.stderr)
+        fail("BAD_NAME", f"unknown action: {action}")
+
+
+if __name__ == "__main__":
+    main(sys.argv)