box-ctl: expand actions (vars, strat, loop) + harden input validation

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 sched != "manual":
-    if not ok:
+        ok, reason = validate_cron(sched)
-        return False, "schedule", reason
+        if not ok:
-    oc = cron_to_oncalendar(sched)
+            return False, "schedule", reason
-    if oc is None:
+        oc = cron_to_oncalendar(sched)
-        return False, "schedule", (
+        if oc is None:
-            "cron pattern not supported for OnCalendar conversion; "
+            return False, "schedule", (
-            "supported: */n * * * *, M * * * *, M H * * *, M H * * DOW, M H DOM * *"
+                "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 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 <action> [args]
 
+fleet:
+  fleet-status
+
 timer actions:
   timer-list
   timer-status <name>
@@ -633,6 +881,24 @@ job actions:
   job-delete <name> [--force]
   job-trigger <name>
 
+variable actions:
+  vars-list
+  vars-get <name>
+  vars-set <name> <value>
+  vars-reset <name>
+
+strategy actions:
+  strat-list
+  strat-get <type> [subtype]
+  strat-set <type>          (JSON on stdin or as arg)
+  strat-reset <type> [subtype]
+
+loop actions:
+  loop-status [--agent AGENT] [--limit N] [--status STATUS]
+  loop-health [--threshold T]
+  loop-breaks
+  loop-resolve <dm_id> [note]
+
 notify:
   notify <agent> <message>"""
 
@@ -682,10 +948,79 @@ def main(argv):
         if len(rest) != 1:
             fail("BAD_NAME", "usage: job-trigger <name>")
         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 <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 == "strat-list":
+        act_strat_list()
+    elif action == "strat-get":
+        if len(rest) < 1 or len(rest) > 2:
+            fail("BAD_NAME", "usage: strat-get <type> [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 <type> [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 <type> [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 <dm_id> [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 <agent> <message>")
         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}")