#!/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 [args...] Security properties: - Fixed verb set; every argument validated before acting. - 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" 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 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 ") 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 ]") 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_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 [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 ] main-chat self-monitor; enable/disable per agent or all timer actions: timer-list timer-status timer-create timer-delete [--keep-job] timer-start|timer-stop|timer-enable|timer-disable job actions: job-list job-get job-put (job JSON on stdin) job-delete [--force] job-trigger job-result (result JSON on stdin; records RESULT + advances chain) job-status (dispatch/result/chain state) job-next [--success|--fail] (dry-run: what would box dispatch next) job-chain [--on-failure] (wire chain_next between jobs) variable actions: vars-list vars-get vars-set vars-reset vars-history [name] [limit] vars-rollback [revision] strategy actions: strat-list strat-get [subtype] [--agent AGENT] strat-set (JSON on stdin or as arg) strat-reset [subtype] [--agent AGENT] loop actions: loop-status [--agent AGENT] [--limit N] [--status STATUS] loop-health [--threshold T] loop-breaks loop-resolve [note] loop-remediate [--dry-run] thread actions (muse-cli gateway; hyphenated aliases for `thread `): thread-list --agent [--limit N] thread-pin --agent --thread thread-unpin --agent --thread thread-archive --agent --thread --confirm thread-unarchive --agent --thread thread-rename --agent --thread --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"): 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": if not rest or rest[0] not in ("mint", "list", "show"): fail("BAD_NAME", "usage: ssh mint|list|show [...]") 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 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)