diff --git a/bin/box-ctl.py b/bin/box-ctl.py index 344c740..10ef80e 100755 --- a/bin/box-ctl.py +++ b/bin/box-ctl.py @@ -215,17 +215,18 @@ def validate_job(data): sched = data.get("schedule") if not isinstance(sched, str): return False, "schedule", "is required and must be a string" - ok, reason = validate_cron(sched) - if not ok: - return False, "schedule", reason - oc = cron_to_oncalendar(sched) - if oc is None: - return False, "schedule", ( - "cron pattern not supported for OnCalendar conversion; " - "supported: */n * * * *, M * * * *, M H * * *, M H * * DOW, M H DOM * *" - ) - if not validate_oncalendar(oc): - return False, "schedule", f"converted OnCalendar {oc!r} rejected by systemd-analyze" + 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") @@ -441,6 +442,9 @@ def act_timer_create(name): 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") @@ -613,12 +617,256 @@ def act_notify(agent, message): out(True, agent=agent, sent=True) +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_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 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)) + + +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}") + out(True, **res) + except Exception as e: + fail("LOOP_ERROR", str(e)) + + # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- USAGE = """usage: box-ctl.py [args] +fleet: + fleet-status + timer actions: timer-list timer-status @@ -633,6 +881,24 @@ job actions: job-delete [--force] job-trigger +variable actions: + vars-list + vars-get + vars-set + vars-reset + +strategy actions: + strat-list + strat-get [subtype] + strat-set (JSON on stdin or as arg) + strat-reset [subtype] + +loop actions: + loop-status [--agent AGENT] [--limit N] [--status STATUS] + loop-health [--threshold T] + loop-breaks + loop-resolve [note] + notify: notify """ @@ -682,10 +948,79 @@ def main(argv): if len(rest) != 1: fail("BAD_NAME", "usage: job-trigger ") act_job_trigger(rest[0]) + elif action == "vars-list": + act_vars_list() + elif action == "vars-get": + if len(rest) != 1: + fail("BAD_NAME", "usage: vars-get ") + act_vars_get(rest[0]) + elif action == "vars-set": + if len(rest) != 2: + fail("BAD_NAME", "usage: vars-set ") + act_vars_set(rest[0], rest[1]) + elif action == "vars-reset": + if len(rest) != 1: + fail("BAD_NAME", "usage: vars-reset ") + act_vars_reset(rest[0]) + elif action == "strat-list": + act_strat_list() + elif action == "strat-get": + if len(rest) < 1 or len(rest) > 2: + fail("BAD_NAME", "usage: strat-get [subtype]") + sub = rest[1] if len(rest) > 1 else None + act_strat_get(rest[0], sub) + elif action == "strat-set": + if len(rest) < 1: + fail("BAD_NAME", "usage: strat-set [JSON]") + payload = rest[1] if len(rest) > 1 else None + act_strat_set(rest[0], payload) + elif action == "strat-reset": + if len(rest) < 1 or len(rest) > 2: + fail("BAD_NAME", "usage: strat-reset [subtype]") + sub = rest[1] if len(rest) > 1 else None + act_strat_reset(rest[0], sub) + 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 [note]") + note = rest[1] if len(rest) > 1 else None + act_loop_resolve(rest[0], note=note) elif action == "notify": if len(rest) != 2: fail("BAD_NAME", "usage: notify ") act_notify(rest[0], rest[1]) + elif action == "fleet-status": + act_fleet_status() + 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}")