super/box
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
07c67e5da3a55d9a9c8ee8a0b80cca8f41a78adc
box/bin/super-cli.py
T

3833 lines
159 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""super-cli.py — Unified Off-Board Orchestrator CLI for NetVM & Box.
Empowers 'off-board super' to govern, task, and observe the fleet of
autonomous browser agents (muse, pip, 646, opm) on bl and interface
with the Google Cloud VM web surfaces.
Domains:
super fleet — Node health, CDP status, active tabs, watch, restart
super dm — Inter-agent DMs, work orders ([WO]), acks, live log tail
super thread — Inspect agent main chats, sidechats, and scrollbacks
super job — Manage scheduled jobs, systemd timers, triggers, logs
super web — Probe VM web surfaces (box.muse-dev.online), auth, sync
super loop — Intrinsic loop strategy, fleet loop health, break taxonomy, vars
super strat — Shortcut to inspect & configure loop modulation strategies
super vars — Shortcut to inspect & adjust runtime control variables
Usage:
super fleet [status|watch|restart|cdp]
super dm [send|wo|ack|log|tail]
super thread [list|view]
super job [list|status|run|log|create|enable|disable|delete]
super web [health|test-auth|sync]
super loop [status|health|breaks|strat|vars|close|nudge|sweep|harvest|remediate]
super strat [show|set|reset|eval]
super vars [list|get|set|reset|history|rollback]
"""
import argparse
import hashlib
import json
import os
import re
import shutil
import subprocess
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
# Paths
NETVM_ROOT = Path("/home/super/Projects/NetVM")
BIN_DIR = NETVM_ROOT / "bin"
JOBS_DIR = NETVM_ROOT / "jobs"
TRANSFERS_DIR = NETVM_ROOT / "transfers"
DM_LOG = NETVM_ROOT / "dm-log.jsonl"
JOB_LOG = NETVM_ROOT / "job-log.jsonl"
CTL_LOG = NETVM_ROOT / "box-ctl.jsonl"
REGISTRY_FILE = NETVM_ROOT / "NODES.md"
CHAT_HISTORY_LOG = NETVM_ROOT / "logs" / "chat-history.jsonl"
WATERMARKS_FILE = NETVM_ROOT / "siphon-watermarks.json"
FOLLOWUPS_FILE = NETVM_ROOT / "followups.json"
JOB_SIDECHATS_FILE = NETVM_ROOT / "job-sidechats.json"
RESPONSE_HARVESTER_PY = BIN_DIR / "response-harvester.py"
FOLLOWUP_SWEEPER_PY = BIN_DIR / "followup-sweeper.py"
PIPELINES_FILE = NETVM_ROOT / "pipelines.json"
JOB_DISPATCH_PY = BIN_DIR / "job-dispatch.py"
# Agent Constants
VALID_NODES = ["muse", "pip", "646", "opm", "def", "dev"]
DEFAULT_SENDER = "super"
DEFAULT_AGENT_SIDECHATS = {
"646": "646 tasks",
"opm": "heartbeat",
"pip": "646-pip-coord",
"muse": "muse tasks",
}
# ---------------------------------------------------------------------------
# ANSI Color & Formatting Engine
# ---------------------------------------------------------------------------
USE_COLOR = sys.stdout.isatty() and os.environ.get("NO_COLOR") is None
def _c(code: str, text: str) -> str:
return f"\033[{code}m{text}\033[0m" if USE_COLOR else str(text)
def c_bold(s: str) -> str: return _c("1", s)
def c_dim(s: str) -> str: return _c("2", s)
def c_green(s: str) -> str: return _c("32", s)
def c_red(s: str) -> str: return _c("31", s)
def c_yellow(s: str) -> str: return _c("33", s)
def c_blue(s: str) -> str: return _c("34", s)
def c_magenta(s: str) -> str: return _c("35", s)
def c_cyan(s: str) -> str: return _c("36", s)
def badge_ok(s: str = "UP") -> str: return c_green(f"● {s}")
def badge_warn(s: str = "WARN") -> str: return c_yellow(f"● {s}")
def badge_err(s: str = "DOWN") -> str: return c_red(f"● {s}")
def badge_dim(s: str) -> str: return c_dim(f"○ {s}")
def print_table(headers: list, rows: list, alignments: list = None):
"""Print an aligned tabular display with clean ANSI formatting."""
if not rows:
print(c_dim(" (no records found)"))
return
num_cols = len(headers)
alignments = alignments or ["left"] * num_cols
# Calculate max visual length per column (stripping ANSI escapes)
strip_ansi = re.compile(r"\033\[[0-9;]*m")
def vlen(s):
return len(strip_ansi.sub("", str(s)))
col_widths = [vlen(h) for h in headers]
for row in rows:
for i in range(num_cols):
val = str(row[i]) if i < len(row) else ""
col_widths[i] = max(col_widths[i], vlen(val))
# Header
hdr_parts = []
sep_parts = []
for i, h in enumerate(headers):
w = col_widths[i]
align = alignments[i]
pad = w - vlen(h)
if align == "right":
hdr_parts.append(" " * pad + c_bold(h))
else:
hdr_parts.append(c_bold(h) + " " * pad)
sep_parts.append("─" * w)
print(" " + " ".join(hdr_parts))
print(" " + c_dim(" ".join(sep_parts)))
# Rows
for row in rows:
row_parts = []
for i in range(num_cols):
val = str(row[i]) if i < len(row) else ""
w = col_widths[i]
align = alignments[i]
pad = max(0, w - vlen(val))
if align == "right":
row_parts.append(" " * pad + val)
else:
row_parts.append(val + " " * pad)
print(" " + " ".join(row_parts))
def parse_relative_time(ts_str: str) -> str:
"""Convert ISO timestamp to short relative time (e.g. 5m ago, 2h ago)."""
if not ts_str:
return "-"
try:
dt = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
now = datetime.now(timezone.utc)
diff = int((now - dt).total_seconds())
if diff < 0:
return "just now"
if diff < 60:
return f"{diff}s ago"
if diff < 3600:
return f"{diff // 60}m ago"
if diff < 86400:
return f"{diff // 3600}h ago"
return f"{diff // 86400}d ago"
except Exception:
return ts_str[:16].replace("T", " ")
# ---------------------------------------------------------------------------
# Registry & Network Helpers
# ---------------------------------------------------------------------------
def get_node_network_info(node: str) -> dict:
"""Compute deterministic veth/peer IP and load pinned CDP port."""
tag = hashlib.sha256(node.encode()).hexdigest()[:8]
idx = int(tag[:3], 16) % 200 + 10
gw_ip = f"10.201.{idx}.1"
peer_ip = f"10.201.{idx}.2"
# Pinned ports from NODES.md registry
pinned_ports = {
"muse": 9410,
"pip": 9420,
"646": 9430,
"opm": 9440,
"def": 9450,
"dev": 9460,
}
cdp_port = pinned_ports.get(node, 9222 + int(tag[4:7], 16) % 2000)
return {
"node": node,
"netns": f"warp-{node}",
"tag": tag,
"gw_ip": gw_ip,
"peer_ip": peer_ip,
"cdp_port": cdp_port,
}
def probe_cdp_status(peer_ip: str, cdp_port: int, timeout: float = 1.5) -> dict:
"""Probe Chromium DevTools protocol on peer IP."""
url = f"http://{peer_ip}:{cdp_port}/json/version"
t0 = time.time()
try:
req = urllib.request.Request(url, headers={"User-Agent": "super-cli/1.0"})
with urllib.request.urlopen(req, timeout=timeout) as resp:
elapsed_ms = int((time.time() - t0) * 1000)
if resp.status == 200:
data = json.load(resp)
return {
"ok": True,
"latency_ms": elapsed_ms,
"browser": data.get("Browser", "Chromium"),
}
except Exception as e:
return {"ok": False, "error": str(e), "latency_ms": None}
return {"ok": False, "error": "unknown error", "latency_ms": None}
def fetch_active_tab(peer_ip: str, cdp_port: int, timeout: float = 1.5) -> dict:
"""Retrieve title and URL of the active browser page."""
url = f"http://{peer_ip}:{cdp_port}/json/list"
try:
req = urllib.request.Request(url, headers={"User-Agent": "super-cli/1.0"})
with urllib.request.urlopen(req, timeout=timeout) as resp:
tabs = json.load(resp)
pages = [t for t in tabs if t.get("type") == "page"]
if pages:
return {
"title": pages[0].get("title", ""),
"url": pages[0].get("url", ""),
}
except Exception:
pass
return {"title": "-", "url": "-"}
def check_process_alive(node: str) -> bool:
"""Check if Chromium process exists for the node's profile."""
try:
res = subprocess.run(
["pgrep", "-f", f"chrome-box/profiles/{node}"],
capture_output=True, text=True, timeout=2
)
return res.returncode == 0
except Exception:
return False
def get_queue_depth(node: str) -> int:
"""Check active flock tickets in /tmp/cdp-queue/<node>/."""
q_dir = Path(f"/tmp/cdp-queue/{node}")
if not q_dir.exists():
return 0
try:
tickets = [f for f in q_dir.iterdir() if f.name.startswith("ticket-")]
return len(tickets)
except Exception:
return 0
# ---------------------------------------------------------------------------
# Domain: FLEET
# ---------------------------------------------------------------------------
def collect_fleet_data() -> list:
results = []
for node in VALID_NODES:
info = get_node_network_info(node)
proc_alive = check_process_alive(node)
cdp = probe_cdp_status(info["peer_ip"], info["cdp_port"])
tab = fetch_active_tab(info["peer_ip"], info["cdp_port"]) if cdp["ok"] else {"title": "-", "url": "-"}
q_depth = get_queue_depth(node)
results.append({
"node": node,
"netns": info["netns"],
"peer_ip": info["peer_ip"],
"cdp_port": info["cdp_port"],
"proc_alive": proc_alive,
"cdp_ok": cdp["ok"],
"latency_ms": cdp["latency_ms"],
"title": tab["title"],
"url": tab["url"],
"queue_depth": q_depth,
})
return results
def cmd_fleet_status(args):
data = collect_fleet_data()
if args.json:
print(json.dumps({"ok": True, "fleet": data}, indent=2))
return
print("\n" + c_bold("=== NETVM FLEET STATUS ===") + c_dim(f" ({datetime.now().strftime('%H:%M:%S')} local)\n"))
headers = ["NODE", "STATUS", "PEER IP:PORT", "LATENCY", "QUEUE", "ACTIVE PAGE / THREAD"]
rows = []
for item in data:
# Status calculation
if item["proc_alive"] and item["cdp_ok"]:
status = badge_ok("ACTIVE")
elif item["proc_alive"] and not item["cdp_ok"]:
status = badge_warn("CDP_DOWN")
else:
status = badge_err("STOPPED")
lat = f"{item['latency_ms']}ms" if item["latency_ms"] is not None else "-"
q = str(item["queue_depth"]) if item["queue_depth"] > 0 else c_dim("idle")
# Format title/URL nicely
title = item["title"]
if "thread/" in item["url"]:
m = re.search(r"thread/([0-9a-fA-F-]+)", item["url"])
if m:
uuid_short = m.group(1)[:8]
title = f"{title} [{c_cyan(uuid_short)}]"
elif item["url"] == "https://muse.ai/":
title = f"{title} [{c_dim('home')}]"
rows.append([
c_bold(item["node"]),
status,
f"{item['peer_ip']}:{item['cdp_port']}",
lat,
q,
title[:45]
])
print_table(headers, rows)
print("\n" + c_dim(" Commands: super fleet watch | super fleet restart <node> | super fleet cdp <node>") + "\n")
def cmd_fleet_watch(args):
interval = getattr(args, "interval", 2)
try:
while True:
# Clear terminal
sys.stdout.write("\033[2J\033[H")
sys.stdout.flush()
cmd_fleet_status(args)
print(c_dim(f" [Auto-refreshing every {interval}s. Press Ctrl+C to exit]"))
time.sleep(interval)
except KeyboardInterrupt:
print("\n" + c_dim("Exited watch mode."))
def cmd_fleet_restart(args):
node = args.node
if node not in VALID_NODES:
print(f"Error: Unknown node '{node}'. Valid: {', '.join(VALID_NODES)}", file=sys.stderr)
sys.exit(1)
print(f"Restarting node '{c_bold(node)}' via netvm-node-up.sh...")
cmd = ["sudo", str(BIN_DIR / "netvm-node-up.sh"), node]
res = subprocess.run(cmd, text=True)
if res.returncode == 0:
print(c_green(f"Node '{node}' brought up successfully."))
else:
print(c_red(f"Node restart exited with code {res.returncode}."))
sys.exit(res.returncode)
def cmd_fleet_cdp(args):
node = args.node
if node not in VALID_NODES:
print(f"Error: Unknown node '{node}'", file=sys.stderr)
sys.exit(1)
info = get_node_network_info(node)
peer = info["peer_ip"]
port = info["cdp_port"]
if args.json:
print(json.dumps({
"node": node,
"peer_ip": peer,
"port": port,
"url": f"http://{peer}:{port}",
"ssh_forward": f"ssh -L {port}:{peer}:{port} super@100.123.153.75"
}, indent=2))
return
print(f"\n{c_bold('CDP Access for ' + node)}:")
print(f" Host Endpoint : {c_cyan(f'http://{peer}:{port}/json/version')}")
print(f" SSH Forward : {c_yellow(f'ssh -L {port}:{peer}:{port} super@100.123.153.75')}")
print(f" Local Connect : {c_dim(f'http://127.0.0.1:{port}')} (after forwarding)\n")
def resolve_sender(args) -> str:
explicit = getattr(args, "from_agent", None)
if explicit and explicit != DEFAULT_SENDER:
return explicit
caller_env = os.environ.get("BOX_CALLER")
if caller_env and caller_env in ("muse", "pip", "646", "opm", "super"):
return caller_env
return explicit or DEFAULT_SENDER
def sign_message(sender: str, message: str, key_path: Path = None, msg_id: str = None) -> tuple:
"""Signs message with SSH key (namespace 'dm') using ssh-keygen -Y sign.
Returns (signed_raw_wire_format, msg_id)."""
import tempfile
key_path = key_path or Path.home() / ".ssh" / "id_ed25519"
if not key_path.exists():
raise FileNotFoundError(f"Signing key not found: {key_path}")
mid = msg_id or hashlib.sha256(f"{sender}-{time.time()}-{os.urandom(8).hex()}".encode()).hexdigest()[:8]
payload = f"[from:{sender}] [id:{mid}]\n\n{message}"
with tempfile.TemporaryDirectory() as td:
payf = Path(td) / "payload"
with open(payf, "w") as f:
f.write(payload)
res = subprocess.run(
["ssh-keygen", "-Y", "sign", "-f", str(key_path), "-n", "dm", str(payf)],
capture_output=True, text=True, timeout=10
)
if res.returncode != 0:
raise RuntimeError(f"ssh-keygen sign failed: {res.stderr}")
sigf = Path(td) / "payload.sig"
with open(sigf, "r") as f:
sigblock = f.read().strip()
signed_wire = f"[from:{sender}] [id:{mid}]\n\n{message}\n\n{sigblock}"
return signed_wire, mid
# ---------------------------------------------------------------------------
# Domain: DM (Direct Messaging & Work Orders)
# ---------------------------------------------------------------------------
def wait_for_reply(recipient: str, target: str, msg_id: str = None, timeout: int = 60) -> str:
"""Wait for recipient agent to reply, displaying an animated spinner."""
spinner_chars = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]
t0 = time.time()
idx = 0
dm_pos = DM_LOG.stat().st_size if DM_LOG.exists() else 0
found_reply = None
chat_hist_file = NETVM_ROOT / "logs" / "chat-history.jsonl"
hist_pos = chat_hist_file.stat().st_size if chat_hist_file.exists() else 0
sys.stdout.write("\n")
try:
while time.time() - t0 < timeout:
spin = spinner_chars[idx % len(spinner_chars)]
elapsed = int(time.time() - t0)
sys.stdout.write(f"\r {c_cyan(spin)} Awaiting response from {c_bold(recipient)}/{target}... ({elapsed}s/{timeout}s) [Ctrl+C to stop waiting]")
sys.stdout.flush()
idx += 1
# 1. Check DM_LOG for return message from recipient
if DM_LOG.exists() and DM_LOG.stat().st_size > dm_pos:
with open(DM_LOG, "r") as f:
f.seek(dm_pos)
for line in f:
line = line.strip()
if not line:
continue
try:
d = json.loads(line)
ev_type = d.get("type", "")
sender = d.get("agent", "")
if sender == recipient and ev_type in ("sent", "send_done", "verified"):
m = d.get("msg") or d.get("body")
if m and (not msg_id or msg_id not in m):
found_reply = m
break
except Exception:
pass
dm_pos = f.tell()
if found_reply:
break
# 2. Check chat-history.jsonl (from harvester if active)
if chat_hist_file.exists() and chat_hist_file.stat().st_size > hist_pos:
with open(chat_hist_file, "r") as f:
f.seek(hist_pos)
for line in f:
line = line.strip()
if not line:
continue
try:
d = json.loads(line)
if d.get("agent") == recipient:
m = d.get("text") or d.get("msg")
if m and (not msg_id or msg_id not in m) and not m.startswith("[from:super]"):
found_reply = m
break
except Exception:
pass
hist_pos = f.tell()
if found_reply:
break
# 3. Periodically poll thread messages via dm.py read every ~2.5s
if idx % 12 == 0:
try:
read_cmd = ["python3", str(BIN_DIR / "dm.py"), "read", "--agent", recipient, "--target", target, "--n", "3"]
r_read = subprocess.run(read_cmd, capture_output=True, text=True, timeout=8)
if r_read.returncode == 0 and r_read.stdout:
for block in r_read.stdout.split("\n---\n"):
block_s = block.strip()
if block_s and (not msg_id or f"[id:{msg_id}]" not in block_s) and not block_s.startswith("[from:super]"):
found_reply = block_s
break
except Exception:
pass
if found_reply:
break
time.sleep(0.2)
sys.stdout.write("\r" + " " * 85 + "\r")
sys.stdout.flush()
if found_reply:
print(f"\n{c_bold(c_cyan('[' + recipient + ']:'))}")
for l in found_reply.splitlines():
print(f" {l}")
print()
return found_reply
else:
print(c_yellow(f"\n (No response received from {recipient} within {timeout}s)\n"))
return None
except KeyboardInterrupt:
sys.stdout.write("\r" + " " * 85 + "\r")
sys.stdout.flush()
print(c_dim("\n (Stopped waiting for response)\n"))
return None
def get_available_targets(agent: str) -> list:
"""Find known sidechats and targets for an agent."""
targets = []
# 1. From job-sidechats.json
if (NETVM_ROOT / "job-sidechats.json").exists():
try:
with open(NETVM_ROOT / "job-sidechats.json") as f:
sc_data = json.load(f)
for k, v in sc_data.items():
if isinstance(v, dict):
if v.get("agent") == agent:
targets.append({"name": k, "type": "sidechat", "desc": f"registered coordination ({k})"})
elif agent in k:
targets.append({"name": k, "type": "sidechat", "desc": f"registered sidechat ({k})"})
except Exception:
pass
# 2. Add well-known agent sidechats
if agent == "646":
targets.append({"name": "646 tasks", "type": "sidechat", "desc": "primary task & check-in sidechat"})
targets.append({"name": "646-opm-work", "type": "sidechat", "desc": "opm work sidechat"})
elif agent == "opm":
targets.append({"name": "heartbeat", "type": "sidechat", "desc": "heartbeat & health sidechat"})
elif agent == "pip":
targets.append({"name": "646-pip-coord", "type": "sidechat", "desc": "pip & 646 coordination sidechat"})
# Deduplicate by name
seen = set()
unique_targets = []
for t in targets:
if t["name"] not in seen:
seen.add(t["name"])
unique_targets.append(t)
# Add Main Chat last with caution label
unique_targets.append({
"name": "main",
"type": "main",
"desc": "Main Chat (CAUTION: avoid per CHAT_POLICY.md)"
})
return unique_targets
def _render_chat_history(agent: str, target: str, limit: int = 5):
"""Fetch and display recent chat messages from thread."""
try:
read_cmd = ["python3", str(BIN_DIR / "dm.py"), "read", "--agent", agent, "--target", target, "--n", str(limit)]
r = subprocess.run(read_cmd, capture_output=True, text=True, timeout=12)
if r.returncode == 0 and r.stdout:
blocks = [b.strip() for b in r.stdout.split("\n---\n") if b.strip()]
if blocks:
print(c_dim("--- Recent Messages ---"))
for b in blocks[-limit:]:
if b.startswith("[from:super]"):
print(f"{c_green('[super]:')}")
elif f"[from:{agent}]" in b or f"from {agent}" in b:
print(f"{c_cyan('[' + agent + ']:')}")
else:
print(f"{c_dim('[message]:')}")
for line in b.splitlines()[:5]:
print(f" {line}")
print(c_dim("-----------------------\n"))
except Exception:
pass
def cmd_dm_chat(args):
agent = args.agent
target = getattr(args, "target", None)
timeout = getattr(args, "timeout", 60) or 60
print("\n" + c_bold(f"=== INTERACTIVE CONVERSATION SESSION: {c_cyan(agent.upper())} ===") + "\n")
# Target selection if not provided
if not target:
available = get_available_targets(agent)
print(c_bold("Select chat target:") + c_dim(" (Prefer sidechats to keep Main Chat clean)\n"))
for idx, t in enumerate(available, 1):
tag = c_green("[Sidechat - RECOMMENDED]") if t["type"] == "sidechat" else c_yellow("[Main Chat - USE WITH CAUTION]")
print(f" [{idx}] {c_bold(t['name']):<18} {tag} {c_dim(t['desc'])}")
print()
try:
choice = input(f"Choose target [1-{len(available)}] (default: 1): ").strip()
if not choice:
target = available[0]["name"]
else:
choice_idx = int(choice) - 1
if 0 <= choice_idx < len(available):
target = available[choice_idx]["name"]
else:
target = available[0]["name"]
except (ValueError, KeyboardInterrupt):
target = available[0]["name"]
print(f"\nEntering chat session with {c_bold(agent)} on {c_cyan(target)}...")
if target == "main":
print(c_yellow(" ⚠ WARNING: Main Chat targeted. Per CHAT_POLICY.md, keep interactions minimal."))
print(c_dim(" Commands: /exit (or /quit), /refresh (reload history), /switch (change target), /wo <title> <body>\n"))
# Initial history render
_render_chat_history(agent, target, limit=5)
# REPL loop
while True:
try:
prompt_str = f"{c_green('super')} ({c_cyan(target)}) > "
line = input(prompt_str).strip()
if not line:
continue
# Handle slash commands
if line in ("/exit", "/quit", ":q"):
print(c_dim("\nExited chat session.\n"))
break
elif line in ("/refresh", "/history"):
_render_chat_history(agent, target, limit=10)
continue
elif line == "/switch":
available = get_available_targets(agent)
print(c_bold("\nSelect new target:"))
for idx, t in enumerate(available, 1):
print(f" [{idx}] {t['name']} ({t['desc']})")
try:
c = input(f"Choose target [1-{len(available)}]: ").strip()
if c and int(c) - 1 in range(len(available)):
target = available[int(c) - 1]["name"]
print(f"Switched target to {c_cyan(target)}.\n")
_render_chat_history(agent, target, limit=5)
except Exception:
pass
continue
elif line.startswith("/wo "):
parts = line[4:].split(" ", 1)
wo_title = parts[0]
wo_body = parts[1] if len(parts) > 1 else wo_title
wo_id = hashlib.sha256(f"super-{agent}-{wo_title}-{time.time()}".encode()).hexdigest()[:8]
wo_content = f"[WO:{wo_id}] {wo_title} — {wo_body}"
signed_wire, mid = sign_message("super", wo_content, msg_id=wo_id)
print(c_dim(f"Issuing Work Order [WO:{wo_id}]..."))
r = subprocess.run([
"python3", str(BIN_DIR / "dm.py"), "send",
"--agent", "super", "--to", agent, "--target", target,
"--raw", signed_wire
], capture_output=True, text=True)
if r.returncode == 0:
print(c_green(f"✔ Work Order [WO:{wo_id}] delivered."))
wait_for_reply(agent, target, msg_id=wo_id, timeout=timeout)
else:
print(c_red(f"Delivery failed: {r.stderr or r.stdout}"))
continue
# Standard message
signed_wire, mid = sign_message("super", line)
r = subprocess.run([
"python3", str(BIN_DIR / "dm.py"), "send",
"--agent", "super", "--to", agent, "--target", target,
"--raw", signed_wire
], capture_output=True, text=True)
if r.returncode == 0:
print(c_green(f"✔ Delivered & verified [{mid}]."))
wait_for_reply(agent, target, msg_id=mid, timeout=timeout)
else:
print(c_red(f"Delivery failed: {r.stderr or r.stdout}"))
except (KeyboardInterrupt, EOFError):
print(c_dim("\nExited chat session.\n"))
break
def cmd_dm_send(args):
sender = resolve_sender(args)
recipient = args.to
target = getattr(args, "target", "main")
message = args.message
wait = getattr(args, "wait", False)
timeout = getattr(args, "timeout", 60) or 60
allow_main = getattr(args, "allow_main_chat", False)
if target == "main":
if not allow_main:
print(c_red("ERROR: Targeting Main Chat is blocked by CHAT_POLICY.md to prevent chat degradation."), file=sys.stderr)
print(c_dim(" To send routine task/payloads, use a sidechat: --target '<sidechat>'.\n To override intentionally, pass --allow-main-chat."), file=sys.stderr)
sys.exit(1)
print(c_yellow(" ⚠ [CHAT POLICY OVERRIDE] Main Chat targeted with --allow-main-chat. Keep interaction minimal."))
should_sign = (sender == "super") and not getattr(args, "no_sign", False)
mid = None
if should_sign:
try:
signed_wire, mid = sign_message(sender, message)
cmd = [
"python3", str(BIN_DIR / "dm.py"), "send",
"--agent", sender,
"--to", recipient,
"--target", target,
"--raw",
signed_wire
]
print(f"Dispatching Cryptographically Signed DM [{c_green('verified from:' + sender)}] -> [{c_bold(recipient)}/{target}]...")
res = subprocess.run(cmd)
if res.returncode != 0:
sys.exit(res.returncode)
if wait:
wait_for_reply(recipient, target, msg_id=mid, timeout=timeout)
sys.exit(0)
except Exception as e:
print(f"Signing notice: {e}, sending standard DM...", file=sys.stderr)
cmd = [
"python3", str(BIN_DIR / "dm.py"), "send",
"--agent", sender,
"--to", recipient,
"--target", target,
]
if getattr(args, "expect_reply", False):
cmd.append("--expect-reply")
if getattr(args, "reply_timeout", None):
cmd.extend(["--reply-timeout", str(args.reply_timeout)])
cmd.append(message)
print(f"Dispatching DM [{c_cyan('from:' + sender)}] -> [{c_bold(recipient)}/{target}]...")
res = subprocess.run(cmd)
if res.returncode != 0:
sys.exit(res.returncode)
if wait:
wait_for_reply(recipient, target, msg_id=mid, timeout=timeout)
sys.exit(0)
def cmd_dm_wo(args):
sender = resolve_sender(args)
recipient = args.to
target = getattr(args, "target", None)
if not target:
target = DEFAULT_AGENT_SIDECHATS.get(recipient, "main")
title = args.title
body = args.body
priority = getattr(args, "priority", "routine")
wait = getattr(args, "wait", True)
timeout = getattr(args, "timeout", 60) or 60
allow_main = getattr(args, "allow_main_chat", False)
if target == "main":
if not allow_main:
print(c_red("ERROR: Work Orders must target sidechats per CHAT_POLICY.md."), file=sys.stderr)
print(c_dim(" Defaulting to sidechat recommended. To override, pass --allow-main-chat."), file=sys.stderr)
sys.exit(1)
print(c_yellow(" ⚠ [CHAT POLICY OVERRIDE] Work Order targeted to Main Chat with --allow-main-chat."))
wo_id = hashlib.sha256(f"{sender}-{recipient}-{title}-{time.time()}".encode()).hexdigest()[:8]
prefix = "[URGENT] " if priority == "urgent" else ""
wo_content = f"{prefix}[WO:{wo_id}] {title} — {body}"
should_sign = (sender == "super") and not getattr(args, "no_sign", False)
if should_sign:
try:
signed_wire, mid = sign_message(sender, wo_content, msg_id=wo_id)
cmd = [
"python3", str(BIN_DIR / "dm.py"), "send",
"--agent", sender,
"--to", recipient,
"--target", target,
"--raw",
signed_wire
]
print(f"Issuing Cryptographically Signed Work Order {c_green('[WO:' + wo_id + ']')} [{c_green('verified from:' + sender)}] -> [{c_bold(recipient)}/{c_cyan(target)}]...")
res = subprocess.run(cmd)
if res.returncode != 0:
sys.exit(res.returncode)
if wait:
wait_for_reply(recipient, target, msg_id=wo_id, timeout=timeout)
sys.exit(0)
except Exception as e:
print(f"Signing notice: {e}, sending standard work order...", file=sys.stderr)
cmd = [
"python3", str(BIN_DIR / "dm.py"), "send",
"--agent", sender,
"--to", recipient,
"--target", target,
"--expect-reply",
wo_content
]
print(f"Issuing Work Order {c_cyan('[WO:' + wo_id + ']')} [{c_cyan('from:' + sender)}] -> [{c_bold(recipient)}/{c_cyan(target)}]...")
res = subprocess.run(cmd)
if res.returncode != 0:
sys.exit(res.returncode)
if wait:
wait_for_reply(recipient, target, msg_id=wo_id, timeout=timeout)
sys.exit(0)
def cmd_dm_ack(args):
sender = resolve_sender(args)
recipient = args.to
target = getattr(args, "target", "main")
ref_id = args.id
ack_message = f"[ACK:{ref_id}]"
cmd = [
"python3", str(BIN_DIR / "dm.py"), "send",
"--agent", sender,
"--to", recipient,
"--target", target,
ack_message
]
print(f"Sending Acknowledgment {c_green('[ACK:' + ref_id + ']')} -> [{c_bold(recipient)}/{target}]...")
res = subprocess.run(cmd)
sys.exit(res.returncode)
def cmd_dm_verify(args):
cmd = ["python3", str(BIN_DIR / "dm.py"), "verify-sig"]
if getattr(args, "agent", None):
cmd.extend(["--agent", args.agent])
if getattr(args, "target", None):
cmd.extend(["--target", args.target])
if getattr(args, "message", None):
cmd.append(args.message)
res = subprocess.run(cmd)
sys.exit(res.returncode)
def cmd_dm_log(args):
n = getattr(args, "n", 25)
filter_agent = getattr(args, "agent", None)
filter_text = getattr(args, "filter", None)
if not DM_LOG.exists():
print(c_dim("dm-log.jsonl not found."))
return
entries = []
with open(DM_LOG, "r") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
data = json.loads(line)
if filter_agent and (data.get("agent") != filter_agent and data.get("to") != filter_agent):
continue
if filter_text:
if filter_text.lower() not in json.dumps(data).lower():
continue
entries.append(data)
except Exception:
continue
entries = entries[-n:]
if args.json:
print(json.dumps({"ok": True, "entries": entries}, indent=2))
return
print("\n" + c_bold(f"=== RECENT INTER-AGENT DMs (last {len(entries)}) ===") + "\n")
headers = ["TIME", "TYPE", "FROM -> TO", "TARGET", "STATUS / DETAIL"]
rows = []
for item in entries:
t_rel = parse_relative_time(item.get("ts", ""))
ev_type = item.get("type", "unknown")
# Colorize type badges
if ev_type == "sent" or ev_type == "send_done":
badge = c_green("SENT")
elif ev_type == "verified":
badge = c_green("✔ VERIFIED")
elif "fail" in ev_type:
badge = c_red("FAIL")
elif ev_type == "retry":
badge = c_yellow("RETRY")
elif ev_type == "send_start":
badge = c_blue("START")
else:
badge = c_dim(ev_type[:8].upper())
from_to = f"{item.get('agent', '-')} -> {item.get('to', '-')}"
target = item.get("target", "-")
msg = item.get("msg") or item.get("reason") or item.get("error") or ""
if not msg:
if ev_type == "retry":
msg = f"attempt {item.get('attempt', '?')} (id: {item.get('id', '-')})"
elif ev_type == "failed":
msg = f"delivery failed (id: {item.get('id', '-')}, verified: {item.get('verified', False)})"
elif ev_type == "sent":
msg = f"delivered (id: {item.get('id', '-')}, verified: {item.get('verified', True)})"
elif ev_type == "verified":
msg = f"confirmed in DOM (id: {item.get('id', '-')})"
elif ev_type == "send_done":
msg = f"completed send (id: {item.get('id', '-')})"
elif ev_type == "alias_resolved":
msg = f"resolved {item.get('target')} -> {item.get('thread_uuid', '')[:8]}"
elif ev_type == "nav_ok":
msg = f"navigated to thread {item.get('thread_uuid', '')[:8]}"
elif item.get("id"):
msg = f"id: {item.get('id')}"
msg_clean = " ".join(msg.split())
# Highlight protocol tokens
if "[WO:" in msg_clean:
msg_clean = re.sub(r"\[WO:([a-f0-9]+)\]", lambda m: c_magenta(m.group(0)), msg_clean)
if "[ACK:" in msg_clean:
msg_clean = re.sub(r"\[ACK:([a-f0-9]+)\]", lambda m: c_green(m.group(0)), msg_clean)
if "[JOB" in msg_clean:
msg_clean = re.sub(r"\[JOB ([^\]]+)\]", lambda m: c_cyan(m.group(0)), msg_clean)
if "[RESULT" in msg_clean:
msg_clean = re.sub(r"\[RESULT ([^\]]+)\]", lambda m: c_yellow(m.group(0)), msg_clean)
rows.append([
t_rel,
badge,
from_to,
target[:16],
msg_clean[:60] if msg_clean else c_dim("-")
])
print_table(headers, rows)
print("\n" + c_dim(" Commands: super dm tail | super dm send --to <agent> \"msg\" | super dm wo --to <agent> --title \"T\" \"body\"") + "\n")
def cmd_dm_tail(args):
filter_text = getattr(args, "filter", None)
print(c_bold("=== STREAMING DM LOG (Ctrl+C to stop) ===") + "\n")
if not DM_LOG.exists():
print(c_red("Error: dm-log.jsonl does not exist."))
return
with open(DM_LOG, "r") as f:
# Seek to end
f.seek(0, os.SEEK_END)
try:
while True:
line = f.readline()
if not line:
time.sleep(0.5)
continue
try:
data = json.loads(line.strip())
if filter_text and filter_text.lower() not in line.lower():
continue
ts = parse_relative_time(data.get("ts", ""))
ev_type = data.get("type", "")
sender = data.get("agent", "")
to = data.get("to", "")
msg = data.get("msg") or data.get("reason") or data.get("error") or ""
print(f"[{ts}] {c_cyan(ev_type.upper()):<12} {sender} -> {to}: {msg[:90]}")
except Exception:
pass
except KeyboardInterrupt:
print("\n" + c_dim("Tail stopped."))
TRANSFERS_REGISTRY_FILE = TRANSFERS_DIR / ".transfers-registry.json"
def _load_transfers_registry() -> dict:
if TRANSFERS_REGISTRY_FILE.exists():
try:
with open(TRANSFERS_REGISTRY_FILE, "r") as f:
return json.load(f)
except Exception:
return {}
return {}
def _save_transfers_registry(reg: dict):
TRANSFERS_DIR.mkdir(parents=True, exist_ok=True)
with open(TRANSFERS_REGISTRY_FILE, "w") as f:
json.dump(reg, f, indent=2)
def cmd_dm_send_file(args):
file_path = Path(args.file).resolve()
if not file_path.exists() or not file_path.is_file():
print(c_red(f"Error: File not found: {file_path}"), file=sys.stderr)
sys.exit(1)
recipient = args.to
target = getattr(args, "target", None)
if not target:
target = DEFAULT_AGENT_SIDECHATS.get(recipient, "main")
if target == "main":
print(c_yellow(" ⚠ [CHAT POLICY NOTE] Targeting Main Chat. Per CHAT_POLICY.md, avoid using Main Chat for routine payloads."))
note = getattr(args, "note", None)
wait = getattr(args, "wait", False)
timeout = getattr(args, "timeout", 60) or 60
# 1. Compute checksum and file stats
file_stat = file_path.stat()
size_bytes = file_stat.st_size
size_kb = size_bytes / 1024.0
hasher = hashlib.sha256()
with open(file_path, "rb") as f:
while chunk := f.read(65536):
hasher.update(chunk)
sha256_full = hasher.hexdigest()
sha256_short = sha256_full[:12]
# 2. Stage file in transfers directory
agent_transfers_dir = TRANSFERS_DIR / recipient
agent_transfers_dir.mkdir(parents=True, exist_ok=True)
ts_prefix = datetime.now().strftime("%Y%m%d_%H%M%S")
dest_filename = f"{ts_prefix}_{file_path.name}"
dest_path = agent_transfers_dir / dest_filename
shutil.copy2(file_path, dest_path)
# 3. Construct protocol payload pointer
file_id = hashlib.sha256(f"{recipient}-{dest_filename}-{time.time()}".encode()).hexdigest()[:8]
# Save to transfers registry
reg = _load_transfers_registry()
reg[file_id] = {
"file_id": file_id,
"recipient": recipient,
"original_name": file_path.name,
"original_path": str(file_path),
"staged_path": str(dest_path),
"sha256": sha256_full,
"size_bytes": size_bytes,
"size_kb": f"{size_kb:.1f} KB",
"staged_at": datetime.now(timezone.utc).isoformat(),
"note": note,
"sender": "super",
"target": target
}
_save_transfers_registry(reg)
lines = [
f"[FILE:{file_id}] {file_path.name} ({size_kb:.1f} KB, sha256:{sha256_short})",
f"Path: {dest_path}",
]
if note:
lines.append(f"Note: {note}")
payload_message = "\n".join(lines)
# 4. Dispatch cryptographically signed DM to recipient sidechat
signed_wire, mid = sign_message("super", payload_message, msg_id=file_id)
print(f"Staged file {c_bold(file_path.name)} -> {c_dim(str(dest_path))}")
print(f"Dispatching File Pointer {c_green('[FILE:' + file_id + ']')} [{c_green('verified from:super')}] -> [{c_bold(recipient)}/{c_cyan(target)}]...")
cmd = [
"python3", str(BIN_DIR / "dm.py"), "send",
"--agent", "super",
"--to", recipient,
"--target", target,
"--raw",
signed_wire
]
res = subprocess.run(cmd)
if res.returncode != 0:
print(c_red(f"File notification delivery failed with code {res.returncode}"), file=sys.stderr)
sys.exit(res.returncode)
print(c_green(f"✔ File notification delivered to {recipient}/{target}."))
if wait:
wait_for_reply(recipient, target, msg_id=file_id, timeout=timeout)
def cmd_dm_files(args):
subaction = getattr(args, "files_action", "list")
filter_agent = getattr(args, "agent", None)
if subaction == "clean":
older_than_days = getattr(args, "older_than", 7)
cutoff_sec = time.time() - (older_than_days * 86400)
reg = _load_transfers_registry()
removed_count = 0
retained_reg = {}
# Scan files on disk
for ad in (TRANSFERS_DIR.iterdir() if TRANSFERS_DIR.exists() else []):
if not ad.is_dir() or ad.name.startswith("."):
continue
if filter_agent and ad.name != filter_agent:
continue
for fp in list(ad.iterdir()):
if fp.is_file() and fp.stat().st_mtime < cutoff_sec:
try:
fp.unlink()
removed_count += 1
except Exception as e:
print(c_red(f"Error removing {fp}: {e}"))
for fid, rec in reg.items():
staged = Path(rec.get("staged_path", ""))
if staged.exists():
retained_reg[fid] = rec
_save_transfers_registry(retained_reg)
print(c_green(f"✔ Cleaned {removed_count} payload file(s) older than {older_than_days} day(s)."))
return
if not TRANSFERS_DIR.exists():
print(c_dim("No transferred files found (transfers/ directory is empty)."))
return
reg = _load_transfers_registry()
records = []
agent_dirs = [TRANSFERS_DIR / filter_agent] if filter_agent else sorted(TRANSFERS_DIR.iterdir())
for ad in agent_dirs:
if not ad.is_dir() or ad.name.startswith("."):
continue
agent_name = ad.name
for fp in sorted(ad.iterdir(), key=lambda p: p.stat().st_mtime, reverse=True):
if not fp.is_file() or fp.name.startswith("."):
continue
st = fp.stat()
size_kb = f"{st.st_size / 1024:.1f} KB"
mtime = datetime.fromtimestamp(st.st_mtime, tz=timezone.utc).isoformat()
t_rel = parse_relative_time(mtime)
clean_name = fp.name
m = re.match(r"^\d{8}_\d{6}_(.+)$", fp.name)
if m:
clean_name = m.group(1)
# Match in registry if available
reg_entry = next((r for r in reg.values() if r.get("staged_path") == str(fp)), {})
file_id = reg_entry.get("file_id", "-")
note = reg_entry.get("note", "")
records.append({
"agent": agent_name,
"file_id": file_id,
"filename": clean_name,
"staged_name": fp.name,
"path": str(fp),
"size_kb": size_kb,
"mtime": mtime,
"time_rel": t_rel,
"note": note
})
if args.json:
print(json.dumps({"ok": True, "files": records}, indent=2))
return
print("\n" + c_bold(f"=== TRANSFERRED PAYLOADS & STAGED FILES ({len(records)}) ===") + "\n")
headers = ["AGENT", "FILE ID", "FILENAME", "SIZE", "TRANSFERRED", "NOTE", "STAGED PATH"]
rows = []
for r in records:
rows.append([
c_cyan(r["agent"]),
c_green(r["file_id"]) if r["file_id"] != "-" else c_dim("-"),
c_bold(r["filename"]),
r["size_kb"],
r["time_rel"],
r["note"][:20] if r["note"] else c_dim("-"),
c_dim(r["path"])
])
print_table(headers, rows)
print("\n" + c_dim(" Commands: super dm send-file --to <agent> <path> | super dm files clean [--older-than N]") + "\n")
# ---------------------------------------------------------------------------
# Domain: THREAD (Chat Oversight via box-chat.py)
# ---------------------------------------------------------------------------
def cmd_thread_list(args):
agent = args.agent
threads = []
def is_noise(title):
t = (title or "").lower().strip()
return bool(re.search(r"generate.*(chat|session)?.*title", t))
# Fast path: try fast headless gateway via muse_hybrid (isolated per-node WARP egress)
try:
import muse_hybrid
gw_threads, gw_err = muse_hybrid.get_threads(agent)
if gw_threads and not gw_err:
for t in gw_threads:
if is_noise(t.get("title")):
continue
threads.append({
"id": t.get("session_id", ""),
"kind": "thread" if t.get("thread") else "chat",
"title": t.get("title") or "(no title)",
"participants": [agent],
"last_message_at": t.get("updated", "")
})
except Exception:
pass
# Fallback to box-chat.py / CDP if gateway returned no threads
if not threads:
cmd = ["python3", str(BIN_DIR / "box-chat.py"), "thread-list", agent]
res = subprocess.run(cmd, capture_output=True, text=True)
try:
data = json.loads(res.stdout)
raw_threads = data.get("threads", []) if data.get("ok") else []
threads = [t for t in raw_threads if not is_noise(t.get("title"))]
except Exception:
threads = []
if args.json:
print(json.dumps({"ok": True, "threads": threads}, indent=2))
return
print("\n" + c_bold(f"=== ACTIVE THREADS FOR {agent.upper()} ({len(threads)}) ===") + "\n")
headers = ["ID / UUID", "KIND", "TITLE", "PARTICIPANTS", "LAST ACTIVE"]
rows = []
for t in threads:
tid = t.get("id", "")
tid_disp = c_cyan(tid[:12]) if tid != "main" else c_bold("main")
kind = t.get("kind", "")
title = t.get("title", "")
parts = ", ".join(t.get("participants", []))
last_act = parse_relative_time(t.get("last_message_at", ""))
rows.append([tid_disp, kind, title[:35], parts, last_act])
print_table(headers, rows)
print("\n" + c_dim(f" View thread: super thread view {agent} <id>") + "\n")
def cmd_thread_view(args):
agent = args.agent
thread_id = args.thread_id
limit = getattr(args, "limit", 15)
messages = []
# Fast path: try fast headless gateway via muse_hybrid (isolated per-node WARP egress)
try:
import muse_hybrid
gw_msgs, gw_err = muse_hybrid.get_history(agent, thread_id=thread_id, limit=limit)
if gw_msgs and not gw_err:
for m in gw_msgs:
role = m.get("role", "unknown")
messages.append({
"from": {"name": role, "role": role},
"text": m.get("text", "")
})
except Exception:
pass
# Fallback to box-chat.py / CDP if gateway returned no messages
if not messages:
cmd = ["python3", str(BIN_DIR / "box-chat.py"), "thread-messages", agent, thread_id, "--limit", str(limit)]
try:
res = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
data = json.loads(res.stdout)
messages = data.get("messages", []) if data.get("ok") else []
except Exception:
messages = []
if not messages:
print(c_red(f"Error viewing thread {thread_id}: no messages returned"))
return
if args.json:
print(json.dumps({"ok": True, "messages": messages}, indent=2))
return
print("\n" + c_bold(f"=== THREAD {thread_id} ({agent}) — {len(messages)} messages ===") + "\n")
for msg in messages:
sender_info = msg.get("from", {})
sender_name = sender_info.get("name") or sender_info.get("role") or "unknown"
text = msg.get("text", "")
# Colorize sender
if sender_name in ("human", "super", "operator-main"):
sender_tag = c_yellow(f"[{sender_name}]")
else:
sender_tag = c_cyan(f"[{sender_name}]")
print(f"{sender_tag}:")
for line in text.split("\n"):
print(f" {line}")
print()
# ---------------------------------------------------------------------------
# Domain: JOB (Systemd Timers & Job Definitions)
# ---------------------------------------------------------------------------
def cmd_job_list(args):
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "job-list"]
res = subprocess.run(cmd, capture_output=True, text=True)
try:
data = json.loads(res.stdout)
except Exception:
print(c_red("Failed to load jobs:") + f"\n{res.stdout}")
return
if args.json:
print(json.dumps(data, indent=2))
return
jobs = data.get("jobs", [])
print("\n" + c_bold(f"=== DEFINED JOBS ({len(jobs)}) ===") + "\n")
headers = ["NAME", "AGENT", "SCHEDULE", "TIMEOUT", "DESCRIPTION"]
rows = []
for j in jobs:
sched = j.get("schedule", "-")
sched_disp = c_green(sched) if sched != "manual" else c_dim("manual")
rows.append([
c_bold(j.get("name", "")),
j.get("agent", "-"),
sched_disp,
f"{j.get('timeout', '-')}s",
(j.get("description") or "-")[:50]
])
print_table(headers, rows)
print("\n" + c_dim(" Commands: super job show <name> | super job status <name> | super job run <name> | super job create <name>") + "\n")
def cmd_job_show(args):
name = args.name
# 1. Fetch job definition via box-ctl
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "job-get", name]
res = subprocess.run(cmd, capture_output=True, text=True)
try:
jdata = json.loads(res.stdout)
except Exception:
print(c_red(f"Failed to query job '{name}': {res.stdout or res.stderr}"))
return
if not jdata.get("ok"):
print(c_red(f"Error: {jdata.get('error', 'Job not found')}"))
return
job = jdata.get("job", {})
# 2. Fetch timer status
tcmd = ["python3", str(BIN_DIR / "box-ctl.py"), "timer-status", name]
tres = subprocess.run(tcmd, capture_output=True, text=True)
tdata = None
try:
parsed_t = json.loads(tres.stdout)
if parsed_t.get("ok"):
tdata = parsed_t
except Exception:
pass
if args.json:
print(json.dumps({"ok": True, "job": job, "timer": tdata}, indent=2))
return
print("\n" + c_bold(f"=== JOB DEFINITION: {name} ===") + "\n")
print(f" Description : {job.get('description', '-')}")
print(f" Target Agent: {c_cyan(job.get('agent', '-'))}")
sched = job.get('schedule', '-')
print(f" Schedule : {c_green(sched) if sched != 'manual' else c_dim('manual')}")
print(f" Timeout : {job.get('timeout', 300)}s")
print(f" On Failure : {job.get('on_failure', 'alert')}")
if job.get("chain_next"):
print(f" Chain Next : {job.get('chain_next')}")
if job.get("dm_target"):
print(f" DM Target : {job.get('dm_target')}")
sidechat = job.get("sidechat")
if sidechat and any(sidechat.values()):
print(f"\n {c_bold('Sidechat Config:')}")
print(f" Create : {sidechat.get('create', False)}")
if sidechat.get("name_template"):
print(f" Name Template : {sidechat.get('name_template')}")
if sidechat.get("reuse_key"):
print(f" Reuse Key : {sidechat.get('reuse_key')}")
followup = job.get("followup")
if followup:
print(f"\n {c_bold('Follow-up Policy:')}")
print(f" Expect Reply : {followup.get('expect_reply', False)}")
print(f" Timeout : {followup.get('timeout', '-')}")
print(f" Nudges : {followup.get('nudges', 0)}")
if followup.get("escalate"):
print(f" Escalate To : {followup.get('escalate')}")
if followup.get("route"):
print(f" Route : {followup.get('route')}")
print(f"\n {c_bold('Prompt Template:')}")
prompt = job.get("prompt_template", "")
for line in prompt.splitlines():
print(f" {c_dim(line)}")
print(f"\n {c_bold('Systemd Timer Status:')}")
if tdata:
unit = tdata.get("unit", f"job-{name}.timer")
act = badge_ok() if tdata.get("active") else badge_err("INACTIVE")
en = str(tdata.get("enabled"))
cal = tdata.get("oncalendar") or "-"
last_run = tdata.get("last_run") or "-"
next_run = tdata.get("next_run") or "-"
res_str = tdata.get("last_result") or "-"
print(f" Unit File : {unit}")
print(f" Active : {act}")
print(f" Enabled : {en}")
print(f" Schedule : {c_cyan(cal)}")
print(f" Last Run : {last_run}")
print(f" Next Run : {next_run}")
print(f" Result : {res_str}")
else:
print(f" {badge_dim('NOT CONFIGURED')} (Run 'super job enable {name}' to activate)")
print()
def cmd_job_status(args):
name = getattr(args, "name", None)
if name:
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "timer-status", name]
else:
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "timer-list"]
res = subprocess.run(cmd, capture_output=True, text=True)
try:
data = json.loads(res.stdout)
except Exception:
print(c_red("Failed to query timers:") + f"\n{res.stdout}")
return
if args.json:
print(json.dumps(data, indent=2))
return
if name:
if not data.get("ok"):
print(c_red(f"Error querying timer '{name}': {data.get('error', 'Unknown error')}"))
return
print(f"\n{c_bold('Timer Status: ' + name)}")
print(f" Unit File : {data.get('unit')}")
print(f" Active : {badge_ok() if data.get('active') else badge_err('INACTIVE')}")
print(f" Enabled : {data.get('enabled')}")
print(f" Schedule : {c_cyan(data.get('oncalendar', '-'))}")
print(f" Last Run : {data.get('last_run') or '-'}")
print(f" Next Run : {data.get('next_run') or '-'}")
print(f" Result : {data.get('last_result') or '-'}\n")
else:
timers = data.get("timers", [])
print("\n" + c_bold(f"=== ACTIVE TIMERS ({len(timers)}) ===") + "\n")
headers = ["NAME", "UNIT", "ACTIVE", "ENABLED"]
rows = []
for t in timers:
act = badge_ok() if t.get("active") else badge_err("INACTIVE")
rows.append([c_bold(t.get("name", "")), t.get("unit", ""), act, str(t.get("enabled"))])
print_table(headers, rows)
print()
def cmd_job_create(args):
name = args.name
if not re.match(r"^[a-z0-9-]{1,64}$", name):
print(c_red("Error: Job name must match ^[a-z0-9-]{1,64}$"), file=sys.stderr)
sys.exit(1)
job_file = JOBS_DIR / f"{name}.json"
existed = job_file.exists()
update_mode = getattr(args, "update", False)
if existed and not update_mode:
print(c_red(f"Error: Job '{name}' already exists. Use --update to modify existing job."), file=sys.stderr)
sys.exit(1)
file_arg = getattr(args, "file", None)
if file_arg:
try:
if file_arg == "-":
raw_data = json.load(sys.stdin)
else:
with open(file_arg, "r") as f:
raw_data = json.load(f)
job_dict = raw_data
job_dict["name"] = name
except Exception as e:
print(c_red(f"Error loading JSON from '{file_arg}': {e}"), file=sys.stderr)
sys.exit(1)
else:
# Check required fields
if not args.agent:
print(c_red("Error: --agent is required when not using --file."), file=sys.stderr)
sys.exit(1)
if not args.prompt:
print(c_red("Error: --prompt is required when not using --file."), file=sys.stderr)
sys.exit(1)
job_dict = {
"name": name,
"agent": args.agent,
"schedule": args.schedule or "manual",
"description": args.description or f"Job {name}",
"prompt_template": args.prompt,
"timeout": args.timeout or 300,
"on_failure": getattr(args, "on_failure", "alert") or "alert",
"chain_next": getattr(args, "chain_next", None),
}
if getattr(args, "sidechat_create", False) or getattr(args, "sidechat_name", None) or getattr(args, "reuse_key", None):
job_dict["sidechat"] = {
"create": bool(getattr(args, "sidechat_create", False) or getattr(args, "sidechat_name", None) or getattr(args, "reuse_key", None)),
}
if getattr(args, "sidechat_name", None):
job_dict["sidechat"]["name_template"] = args.sidechat_name
if getattr(args, "reuse_key", None):
job_dict["sidechat"]["reuse_key"] = args.reuse_key
if getattr(args, "dm_target", None):
job_dict["dm_target"] = args.dm_target
# Call box-ctl.py job-put
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "job-put", name]
res = subprocess.run(cmd, input=json.dumps(job_dict), capture_output=True, text=True)
try:
resp = json.loads(res.stdout)
except Exception:
print(c_red(f"Failed to save job via box-ctl: {res.stdout or res.stderr}"), file=sys.stderr)
sys.exit(1)
if not resp.get("ok"):
print(c_red(f"Error saving job: {resp.get('error', 'unknown error')}"), file=sys.stderr)
if "detail" in resp:
print(c_dim(f"Detail: {json.dumps(resp['detail'])}"), file=sys.stderr)
sys.exit(1)
action_label = "Updated" if existed else "Created"
print(c_green(f"✔ {action_label} job '{name}' (saved to jobs/{name}.json and committed to git)"))
# Manage timer
schedule = job_dict.get("schedule", "manual")
no_enable = getattr(args, "no_enable", False)
timer_unit = Path.home() / ".config" / "systemd" / "user" / f"job-{name}.timer"
timer_existed = timer_unit.exists()
if schedule != "manual" and not no_enable:
if timer_existed:
subprocess.run(["python3", str(BIN_DIR / "box-ctl.py"), "timer-delete", name, "--keep-job"],
capture_output=True, text=True)
r_timer = subprocess.run(["python3", str(BIN_DIR / "box-ctl.py"), "timer-create", name],
capture_output=True, text=True)
try:
tresp = json.loads(r_timer.stdout)
if tresp.get("ok"):
print(c_green(f"✔ Systemd timer active: job-{name}.timer ({tresp.get('oncalendar')})"))
else:
print(c_yellow(f"⚠ Timer could not be activated: {tresp.get('error')}"))
except Exception:
print(c_yellow(f"⚠ Timer creation output: {r_timer.stdout or r_timer.stderr}"))
elif timer_existed and schedule == "manual":
subprocess.run(["python3", str(BIN_DIR / "box-ctl.py"), "timer-delete", name, "--keep-job"],
capture_output=True, text=True)
print(c_dim("Systemd timer removed because schedule is set to 'manual'."))
def cmd_job_enable(args):
name = args.name
job_file = JOBS_DIR / f"{name}.json"
if not job_file.exists():
print(c_red(f"Error: Job definition jobs/{name}.json not found."), file=sys.stderr)
sys.exit(1)
try:
jdata = json.loads(job_file.read_text())
if jdata.get("schedule") == "manual":
print(c_red(f"Error: Cannot enable timer for job '{name}' with schedule 'manual'."), file=sys.stderr)
print(c_dim("Update schedule to a valid cron pattern first using 'super job create --update'."), file=sys.stderr)
sys.exit(1)
except Exception as e:
print(c_red(f"Error reading job definition: {e}"), file=sys.stderr)
sys.exit(1)
timer_unit = Path.home() / ".config" / "systemd" / "user" / f"job-{name}.timer"
if not timer_unit.exists():
print(f"Timer unit not found for '{name}'. Creating systemd timer...")
res = subprocess.run(["python3", str(BIN_DIR / "box-ctl.py"), "timer-create", name],
capture_output=True, text=True)
else:
print(f"Activating existing timer for '{name}'...")
res = subprocess.run(["python3", str(BIN_DIR / "box-ctl.py"), "timer-enable", name],
capture_output=True, text=True)
try:
data = json.loads(res.stdout)
if data.get("ok"):
print(c_green(f"✔ Timer job-{name}.timer enabled and active."))
else:
print(c_red(f"Error enabling timer: {data.get('error', 'unknown error')}"))
sys.exit(1)
except Exception:
print(res.stdout or res.stderr)
def cmd_job_disable(args):
name = args.name
timer_unit = Path.home() / ".config" / "systemd" / "user" / f"job-{name}.timer"
if not timer_unit.exists():
print(c_yellow(f"Timer job-{name}.timer is not currently installed."))
return
print(f"Disabling timer for '{name}'...")
res = subprocess.run(["python3", str(BIN_DIR / "box-ctl.py"), "timer-disable", name],
capture_output=True, text=True)
subprocess.run(["python3", str(BIN_DIR / "box-ctl.py"), "timer-stop", name],
capture_output=True, text=True)
try:
data = json.loads(res.stdout)
if data.get("ok"):
print(c_green(f"✔ Timer job-{name}.timer disabled and stopped."))
else:
print(c_red(f"Error disabling timer: {data.get('error', 'unknown error')}"))
sys.exit(1)
except Exception:
print(res.stdout or res.stderr)
def cmd_job_delete(args):
name = args.name
job_file = JOBS_DIR / f"{name}.json"
timer_unit = Path.home() / ".config" / "systemd" / "user" / f"job-{name}.timer"
if not job_file.exists() and not timer_unit.exists():
print(c_red(f"Error: Job '{name}' not found."), file=sys.stderr)
sys.exit(1)
force = getattr(args, "force", False)
yes = getattr(args, "yes", False)
if not (yes or force):
prompt_msg = f"Are you sure you want to delete job '{c_bold(name)}' and its timer? [y/N]: "
try:
choice = input(prompt_msg).strip().lower()
if choice not in ("y", "yes"):
print(c_dim("Deletion aborted."))
return
except KeyboardInterrupt:
print("\n" + c_dim("Deletion aborted."))
return
# 1. Delete timer if exists
if timer_unit.exists():
print(f"Removing systemd timer job-{name}.timer...")
r_timer = subprocess.run(["python3", str(BIN_DIR / "box-ctl.py"), "timer-delete", name, "--keep-job"],
capture_output=True, text=True)
try:
tdata = json.loads(r_timer.stdout)
if not tdata.get("ok") and not force:
print(c_red(f"Error deleting timer: {tdata.get('error')}"), file=sys.stderr)
sys.exit(1)
except Exception:
pass
# 2. Delete job file
if job_file.exists():
print(f"Removing job definition jobs/{name}.json...")
r_job = subprocess.run(["python3", str(BIN_DIR / "box-ctl.py"), "job-delete", name, "--force"],
capture_output=True, text=True)
try:
jdata = json.loads(r_job.stdout)
if jdata.get("ok"):
print(c_green(f"✔ Job '{name}' and its systemd units deleted successfully."))
else:
print(c_red(f"Error deleting job: {jdata.get('error')}"), file=sys.stderr)
sys.exit(1)
except Exception:
print(r_job.stdout or r_job.stderr)
def cmd_job_run(args):
name = args.name
follow = getattr(args, "follow", False)
job_file = JOBS_DIR / f"{name}.json"
if not job_file.exists():
print(c_red(f"Error: Job definition jobs/{name}.json not found."), file=sys.stderr)
sys.exit(1)
print(f"Triggering job dispatch for '{c_bold(name)}'...")
if not follow:
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "job-trigger", name]
res = subprocess.run(cmd)
sys.exit(res.returncode)
# Follow mode: open log files and track from current end position
dm_pos = DM_LOG.stat().st_size if DM_LOG.exists() else 0
job_pos = JOB_LOG.stat().st_size if JOB_LOG.exists() else 0
# Trigger job in background
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "job-trigger", name]
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
print(c_dim(f"Following execution logs for '{name}' (Press Ctrl+C to exit)..."))
start_time = time.time()
dispatched = False
completed = False
try:
while time.time() - start_time < 90: # 90s max follow
if proc.poll() is not None and not dispatched:
out_txt, err_txt = proc.communicate()
try:
pdata = json.loads(out_txt)
if not pdata.get("ok"):
print(c_red(f"Trigger failed: {pdata.get('error')}"))
return
else:
dispatched = True
except Exception:
pass
if JOB_LOG.exists() and JOB_LOG.stat().st_size > job_pos:
with open(JOB_LOG, "r") as f:
f.seek(job_pos)
for line in f:
line = line.strip()
if not line:
continue
try:
d = json.loads(line)
if d.get("job_name") == name or name in str(d.get("job_id", "")):
ev = d.get("type") or d.get("event") or "event"
badge = c_green("DISPATCH") if "dispatched" in ev else (
c_red("FAILED") if "fail" in ev else c_cyan(ev.upper())
)
print(f" {badge} [{c_bold(name)}] {d.get('target', '')} {d.get('error', '')}")
if "failed" in ev or "timeout" in ev:
completed = True
except Exception:
pass
job_pos = f.tell()
if DM_LOG.exists() and DM_LOG.stat().st_size > dm_pos:
with open(DM_LOG, "r") as f:
f.seek(dm_pos)
for line in f:
line = line.strip()
if not line:
continue
try:
d = json.loads(line)
msg = d.get("msg") or d.get("error") or ""
if name in msg or "[RESULT" in msg or "[WO" in msg:
ev_type = d.get("type", "")
sender = d.get("agent", "")
to = d.get("to", "")
badge = c_green("VERIFIED") if ev_type == "verified" else c_cyan(ev_type.upper())
print(f" {badge} DM {sender} -> {to}: {msg[:100]}")
if "[RESULT" in msg:
completed = True
except Exception:
pass
dm_pos = f.tell()
if completed:
print(c_green(f"✔ Job execution completed."))
break
time.sleep(0.5)
except KeyboardInterrupt:
print("\n" + c_dim("Stopped following."))
def cmd_job_log(args):
n = getattr(args, "n", 20)
filter_name = getattr(args, "name", None)
if not JOB_LOG.exists():
print(c_dim("job-log.jsonl not found."))
return
entries = []
with open(JOB_LOG, "r") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
d = json.loads(line)
if filter_name and d.get("job_name") != filter_name and filter_name not in d.get("job_id", ""):
continue
entries.append(d)
except Exception:
continue
entries = entries[-n:]
if args.json:
print(json.dumps({"ok": True, "entries": entries}, indent=2))
return
print("\n" + c_bold(f"=== RECENT JOB ACTIVITY (last {len(entries)}) ===") + "\n")
headers = ["TIME", "EVENT", "JOB ID", "AGENT", "DETAIL"]
rows = []
for item in entries:
t_rel = parse_relative_time(item.get("ts", ""))
ev = item.get("event", "-")
badge = c_green("DISPATCH") if "dispatched" in ev else c_dim(ev[:12])
job_id = item.get("job_id", "-")
agent = item.get("agent", "-")
detail = item.get("target") or item.get("error") or ""
rows.append([t_rel, badge, job_id[:25], agent, detail])
print_table(headers, rows)
print()
# ---------------------------------------------------------------------------
# Domain: WEB (Google Cloud VM Web Surfaces Probe)
# ---------------------------------------------------------------------------
def cmd_web_health(args):
endpoints = [
{"name": "Box Console (Front-Door)", "url": "https://box.muse-dev.online/"},
{"name": "Board Service", "url": "https://board.muse-dev.online/"},
{"name": "VM Reverse Proxy (sslip)", "url": "https://34-139-37-135.sslip.io/"},
{"name": "Box Timers API", "url": "https://box.muse-dev.online/api/box/timers"},
]
print("\n" + c_bold("=== GOOGLE CLOUD VM WEB SURFACES HEALTH (34.139.37.135) ===") + "\n")
results = []
headers = ["SURFACE", "ENDPOINT", "STATUS", "LATENCY", "SERVER / VIA"]
rows = []
for ep in endpoints:
t0 = time.time()
try:
req = urllib.request.Request(
ep["url"],
headers={"User-Agent": "super-cli/1.0 (bl; health-probe)"}
)
with urllib.request.urlopen(req, timeout=4) as resp:
elapsed_ms = int((time.time() - t0) * 1000)
code = resp.status
via = resp.headers.get("Via") or resp.headers.get("Server") or "ok"
badge = badge_ok(f"{code}")
except urllib.error.HTTPError as e:
elapsed_ms = int((time.time() - t0) * 1000)
code = e.code
via = e.headers.get("Via") or e.headers.get("Server") or "-"
# 401/403 means auth works as intended
if code in (401, 403):
badge = badge_ok(f"{code} AUTH_GATED")
else:
badge = badge_warn(f"{code}")
except Exception as e:
elapsed_ms = None
code = "ERR"
via = str(e)[:25]
badge = badge_err("UNREACHABLE")
lat_disp = f"{elapsed_ms}ms" if elapsed_ms is not None else "-"
rows.append([c_bold(ep["name"]), ep["url"], badge, lat_disp, via[:25]])
results.append({"name": ep["name"], "url": ep["url"], "code": code, "latency_ms": elapsed_ms})
print_table(headers, rows)
print("\n" + c_dim(" Operator note: box.muse-dev.online requires PIN or Bearer session cookie.") + "\n")
if args.json:
print(json.dumps({"ok": True, "results": results}, indent=2))
def cmd_web_test_auth(args):
print("\n" + c_bold("Testing Box API Authentication Gate:") + "\n")
url = "https://box.muse-dev.online/api/box/timers"
try:
req = urllib.request.Request(url, headers={"User-Agent": "super-cli/1.0"})
with urllib.request.urlopen(req, timeout=4) as resp:
print(c_yellow(f"Unexpected: Endpoint responded without auth: HTTP {resp.status}"))
except urllib.error.HTTPError as e:
body = e.read().decode("utf-8", errors="ignore")
if e.code in (401, 403):
print(f" Endpoint: {url}")
print(f" Response: {badge_ok(f'HTTP {e.code}')} (Protection active)")
print(f" Body : {c_dim(body.strip()[:100])}\n")
else:
print(c_warn(f"HTTP {e.code}: {body}"))
except Exception as e:
print(c_red(f"Error testing auth: {e}"))
def cmd_web_sync(args):
print("\n" + c_bold("=== DATA SYNC INTEGRITY (bl -> VM) ===") + "\n")
files = [
("DM Log", DM_LOG),
("Job Log", JOB_LOG),
("Control Audit", CTL_LOG),
]
headers = ["DATA STORE", "PATH", "SIZE", "RECORDS", "LAST MODIFIED"]
rows = []
for label, path in files:
if path.exists():
stat = path.stat()
size_kb = f"{stat.st_size / 1024:.1f} KB"
# Count lines
try:
with open(path, "r") as f:
lines = sum(1 for _ in f)
except Exception:
lines = 0
mtime = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat()
mtime_rel = parse_relative_time(mtime)
rows.append([c_bold(label), str(path), size_kb, str(lines), mtime_rel])
else:
rows.append([c_bold(label), str(path), "-", "0", c_dim("missing")])
print_table(headers, rows)
print()
# ---------------------------------------------------------------------------
# Domain: CRED (Credentials & Onboarding Client)
# ---------------------------------------------------------------------------
def cmd_cred_initiate(args):
import cred_client
client = cred_client.CredClient()
res = client.initiate(args.node, args.email, service=getattr(args, "service", "muse"), account_name=getattr(args, "account_name", None))
if getattr(args, "json", False):
print(json.dumps(res, indent=2))
else:
st = res.get("status")
if st == "awaiting_otp":
print("\n" + c_yellow(f"[APPROVAL_NEEDED] Code sent to {args.email} for node '{args.node}'.") + "\n")
print(f" Submit OTP with: super cred submit-otp --node {args.node} --otp <code>\n")
sys.exit(2)
elif st == "active":
print("\n" + c_green(f"[SUCCESS] Node '{args.node}' is already authenticated and active.") + "\n")
else:
print(f"[{st.upper()}] {res.get('message') or res.get('detail')}")
sys.exit(res.get("code", 1))
def cmd_cred_submit_otp(args):
import cred_client
client = cred_client.CredClient()
res = client.submit_otp(args.node, args.otp, email=getattr(args, "email", None))
if getattr(args, "json", False):
print(json.dumps(res, indent=2))
else:
st = res.get("status")
if st == "active":
print("\n" + c_green(f"[SUCCESS] Node '{args.node}' successfully signed in!") + "\n")
else:
print(f"[{st.upper()}] {res.get('message') or res.get('detail')}")
sys.exit(res.get("code", 1))
def cmd_cred_status(args):
import cred_client
client = cred_client.CredClient()
res = client.status(args.node)
if getattr(args, "json", False):
print(json.dumps(res, indent=2))
else:
print("\n" + c_bold(f"=== CRED STATUS: {args.node} ===") + "\n")
print(f" Node : {c_bold(res['node'])}")
print(f" Email : {res['email']}")
print(f" Status : {res['status']}")
print(f" CDP Port : {res['cdp_port'] or '-'}")
alive_b = badge_ok("YES") if res['session_alive'] else badge_warn("NO")
print(f" Session Alive: {alive_b}")
if res["detail"]:
print(f" Detail : {c_dim(res['detail'])}")
print()
def cmd_cred_link_instagram(args):
import cred_client
client = cred_client.CredClient()
res = client.link_instagram(args.node, notify=getattr(args, "notify", False))
if getattr(args, "json", False):
print(json.dumps(res, indent=2))
else:
print("\n" + c_bold(f"=== INSTAGRAM LINKING: {args.node} ===") + "\n")
if res.get("error"):
print(c_err(f" Error: {res['error']}"))
sys.exit(1)
print(f" Tailscale Portal: {c_cyan(res['portal_url'])}")
print(f" Direct IP Portal: {c_cyan(res['portal_ip_url'])}")
oauth_preview = res['direct_oauth_url'][:75] + "..." if res.get('direct_oauth_url') else "-"
print(f" OAuth URL : {c_dim(oauth_preview)}")
if getattr(args, "notify", False):
n_badge = badge_ok("SENT") if res['email_notified'] else badge_err("FAILED")
print(f" Email Alert : {n_badge}")
print("\n" + c_yellow(" → Tap either Tailscale link from your phone/browser to complete Meta age verification.") + "\n")
def cmd_cred_meta_audit(args):
import cred_client
client = cred_client.CredClient()
res = client.audit_meta(args.node)
if getattr(args, "json", False):
print(json.dumps(res, indent=2))
else:
print("\n" + c_bold(f"=== META ACCOUNTS CENTER AUDIT: {args.node} ===") + "\n")
if res.get("error"):
print(c_red(f" Error: {res['error']}"))
sys.exit(1)
email_str = res.get('email') or c_dim('(none / phone-only)')
print(f" Meta Account Email: {c_cyan(email_str)}")
profiles = res.get("profiles", [])
print(f" Linked Profiles ({len(profiles)}):")
for p_info in profiles:
p_type = p_info.get("type", "unknown")
p_name = p_info.get("name", "unknown")
badge = c_green(f"[{p_type}]") if p_type == "instagram" else c_magenta(f"[{p_type}]")
print(f" - {badge} {c_bold(p_name)}")
print()
def cmd_cred_list(args):
import cred_client
client = cred_client.CredClient()
items = client.list_all()
if getattr(args, "json", False):
print(json.dumps(items, indent=2))
else:
print("\n" + c_bold("=== FLEET CREDENTIAL & ONBOARDING STATUS ===") + "\n")
headers = ["NODE", "STATUS", "CDP", "ALIVE", "EMAIL"]
rows = []
for it in items:
alive_b = badge_ok("YES") if it["session_alive"] else badge_warn("NO")
rows.append([c_bold(it['node']), it['status'], str(it['cdp_port'] or '-'), alive_b, it['email']])
print_table(headers, rows)
print()
# ---------------------------------------------------------------------------
# Domain: HARVEST (Response & Readback Harvester)
# ---------------------------------------------------------------------------
def cmd_harvest_run(args):
cmd = [sys.executable, str(RESPONSE_HARVESTER_PY), "--once"]
if getattr(args, "agent", None):
cmd.extend(["--agent", args.agent])
if getattr(args, "dry_run", False):
cmd.append("--dry-run")
if getattr(args, "json", False):
cmd.append("--json")
try:
res = subprocess.run(cmd, capture_output=True, text=True)
if getattr(args, "json", False):
print(res.stdout.strip())
else:
if res.stdout.strip():
print(res.stdout.strip())
if res.stderr.strip():
print(c_yellow(res.stderr.strip()), file=sys.stderr)
except Exception as e:
print(c_red(f"Error running harvest: {e}"), file=sys.stderr)
sys.exit(1)
def cmd_harvest_status(args):
watermarks = {}
if WATERMARKS_FILE.exists():
try:
with open(WATERMARKS_FILE, "r") as f:
watermarks = json.load(f)
except Exception:
pass
sidechats = {}
if JOB_SIDECHATS_FILE.exists():
try:
with open(JOB_SIDECHATS_FILE, "r") as f:
sidechats = json.load(f)
except Exception:
pass
latest_seen = {}
if CHAT_HISTORY_LOG.exists():
try:
with open(CHAT_HISTORY_LOG, "r") as f:
for line in f:
if not line.strip(): continue
try:
rec = json.loads(line)
k = f"{rec.get('agent')}:{rec.get('thread_id')}"
latest_seen[k] = rec.get("ts")
except Exception:
continue
except Exception:
pass
if getattr(args, "json", False):
print(json.dumps({
"watermarks": watermarks,
"sidechats": sidechats,
"latest_events": latest_seen
}, indent=2))
return
now_str = datetime.now().strftime("%H:%M:%S")
print(f"\n=== RESPONSE HARVESTER STATUS === ({now_str} local)\n")
headers = ["AGENT", "THREAD / ALIAS", "ID", "WATERMARK", "LAST HARVESTED", "STATUS"]
rows = []
for agent in VALID_NODES:
wm_main = watermarks.get(f"{agent}:main", "-")
last_ts = latest_seen.get(f"{agent}:main")
status_b = badge_ok("ACTIVE") if wm_main != "-" else badge_dim("IDLE")
rows.append([
c_bold(agent),
"Main Chat",
c_dim("main"),
wm_main[:16] if wm_main != "-" else c_dim("-"),
parse_relative_time(last_ts) if last_ts else c_dim("none"),
status_b
])
for key, val in sidechats.items():
if key.startswith("_"): continue
if isinstance(val, dict):
tid = val.get("thread_uuid") or val.get("uuid")
a = val.get("agent", "opm")
elif isinstance(val, str):
tid = val
a = "opm"
else:
continue
if a == agent and tid:
wm_sc = watermarks.get(f"{agent}:{tid}", "-")
last_sc_ts = latest_seen.get(f"{agent}:{tid}")
status_sc = badge_ok("ACTIVE") if wm_sc != "-" else badge_dim("IDLE")
rows.append([
c_dim(f" └─ {agent}"),
key,
c_dim(tid[:8] + "…"),
wm_sc[:16] if wm_sc != "-" else c_dim("-"),
parse_relative_time(last_sc_ts) if last_sc_ts else c_dim("none"),
status_sc
])
print_table(headers, rows)
print(c_dim("\n Commands: super harvest run | super harvest tail\n"))
def _print_harvest_line(line: str, filter_text: str = None):
try:
e = json.loads(line)
text = e.get("text", "")
if filter_text and filter_text.lower() not in text.lower():
return
ts = e.get("ts", "")[11:19]
agent = e.get("agent", "unknown")
author = e.get("author", "unknown")
thread = e.get("thread_name") or e.get("thread_id", "main")
auth_color = c_green if author == "assistant" else c_cyan
snippet = text.replace("\n", " ")[:120]
print(f" {c_dim(ts)} {c_bold(f'[{agent}:{thread}]')} {auth_color(author)}: {snippet}")
except Exception:
pass
def cmd_harvest_tail(args):
if not CHAT_HISTORY_LOG.exists():
print(c_dim(f"Chat history log empty ({CHAT_HISTORY_LOG})"))
return
print(c_bold(f"=== TAILING CHAT HARVEST STREAM ({CHAT_HISTORY_LOG.name}) === (Ctrl+C to stop)") + "\n")
try:
with open(CHAT_HISTORY_LOG, "r") as f:
lines = f.readlines()
for line in lines[-15:]:
_print_harvest_line(line, getattr(args, "filter", None))
f.seek(0, os.SEEK_END)
while True:
line = f.readline()
if line:
_print_harvest_line(line, getattr(args, "filter", None))
else:
time.sleep(0.5)
except KeyboardInterrupt:
print()
# ---------------------------------------------------------------------------
# Domain: FOLLOWUP (Deadline Tracking & Nudges)
# ---------------------------------------------------------------------------
def cmd_followup_list(args):
followups = {}
if FOLLOWUPS_FILE.exists():
try:
with open(FOLLOWUPS_FILE, "r") as f:
followups = json.load(f)
except Exception:
pass
if getattr(args, "json", False):
print(json.dumps(followups, indent=2))
return
now_str = datetime.now().strftime("%H:%M:%S")
print(f"\n=== FOLLOW-UP TRACKING === ({now_str} local)\n")
headers = ["DM ID", "ROUTE", "TARGET", "NUDGES", "DEADLINE", "STATUS", "SNIPPET"]
rows = []
now = datetime.now(timezone.utc)
for dm_id, rec in sorted(followups.items(), key=lambda x: x[1].get("sent_at", ""), reverse=True):
st = rec.get("status", "pending")
if not getattr(args, "all", False) and st not in ("pending", "escalated"):
continue
route = f"{rec.get('sender', '?')} → {c_bold(rec.get('recipient', '?'))}"
target = rec.get("target", "main")
if len(target) > 18:
target = target[:17] + "…"
nudges = f"{rec.get('nudges_sent', 0)}/{rec.get('nudges_allowed', 2)}"
deadline_str = rec.get("deadline")
deadline_display = c_dim("-")
if deadline_str and st == "pending":
try:
dl_dt = datetime.fromisoformat(deadline_str.replace("Z", "+00:00"))
diff = int((dl_dt - now).total_seconds())
if diff > 0:
deadline_display = c_green(f"in {diff // 60}m")
else:
deadline_display = c_red(f"{abs(diff) // 60}m overdue")
except Exception:
deadline_display = deadline_str[:16]
if st == "pending":
status_badge = badge_warn("PENDING")
elif st == "escalated":
status_badge = badge_err("ESCALATED")
elif st == "resolved":
status_badge = badge_ok("RESOLVED")
else:
status_badge = badge_dim(st.upper())
snippet = rec.get("prompt_snippet") or rec.get("resolved_snippet") or ""
snippet = snippet.replace("\n", " ")[:35]
rows.append([
c_bold(dm_id[:8]),
route,
target,
nudges,
deadline_display,
status_badge,
c_dim(snippet)
])
print_table(headers, rows)
if not getattr(args, "all", False):
print(c_dim("\n (Showing pending & escalated only; use --all to view resolved records)"))
print(c_dim(" Commands: super followup sweep | super followup cancel <dm_id>\n"))
def cmd_followup_sweep(args):
cmd = [sys.executable, str(FOLLOWUP_SWEEPER_PY), "--once"]
if getattr(args, "dry_run", False):
cmd.append("--dry-run")
try:
res = subprocess.run(cmd, capture_output=True, text=True)
if res.stdout.strip():
print(res.stdout.strip())
if res.stderr.strip():
print(c_yellow(res.stderr.strip()), file=sys.stderr)
except Exception as e:
print(c_red(f"Error running sweeper: {e}"), file=sys.stderr)
sys.exit(1)
def cmd_followup_cancel(args):
if not FOLLOWUPS_FILE.exists():
print(c_red("No follow-ups found."), file=sys.stderr)
sys.exit(1)
try:
with open(FOLLOWUPS_FILE, "r") as f:
data = json.load(f)
except Exception as e:
print(c_red(f"Error reading follow-ups: {e}"), file=sys.stderr)
sys.exit(1)
target_id = args.dm_id.strip()
match_key = None
for k in data.keys():
if k == target_id or k.startswith(target_id):
match_key = k
break
if not match_key:
print(c_red(f"Follow-up ID '{target_id}' not found."), file=sys.stderr)
sys.exit(1)
data[match_key]["status"] = "canceled"
data[match_key]["canceled_at"] = datetime.now(timezone.utc).isoformat()
tmp_path = f"{FOLLOWUPS_FILE}.tmp.{os.getpid()}"
with open(tmp_path, "w") as f:
json.dump(data, f, indent=2)
os.replace(tmp_path, FOLLOWUPS_FILE)
print(c_green(f"● Cancelled follow-up {match_key}."))
# Also stand down the VM-side nudge controller (request-sweeper.py),
# which tracks dm_followup records independently of bl's local file.
# Best-effort: the local cancel already succeeded, so a VM failure
# (unreachable, no record, bad response) must not fail the command.
_vm_followup_cancel(match_key)
def _vm_followup_cancel(dm_id):
"""POST /api/box/followups/cancel so the VM-side sweeper stops nudging.
Best-effort: never raises. Prints a one-line outcome.
"""
import json as _json
import os as _os
import subprocess as _sp
import time as _time
import urllib.parse as _up
import urllib.request as _ur
import urllib.error as _ue
box_api = _os.environ.get("BOX_API_BASE", "https://box.muse-dev.online")
key = _os.environ.get("BOX_SIGN_KEY",
_os.path.expanduser("~/.ssh/id_ed25519"))
endpoint = "followups/cancel"
ts_now = str(int(_time.time()))
payload = ("%s\n%s" % (ts_now, endpoint)).encode()
try:
pr = _sp.run(["ssh-keygen", "-Y", "sign", "-f", key, "-n", "box"],
input=payload, capture_output=True, timeout=15)
sig = pr.stdout.decode().strip()
if not sig:
raise RuntimeError("empty signature")
except Exception as e:
print(c_warn(" VM follow-up cancel skipped (signing failed: %s)" % e))
return
query = _up.urlencode({"identity": "bl", "ts": ts_now, "sig": sig})
url = "%s/api/box/followups/cancel?%s" % (box_api, query)
try:
req = _ur.Request(url,
data=_json.dumps({"dm_id": dm_id}).encode(),
headers={"Content-Type": "application/json",
"User-Agent": "super-cli/1.0 (bl)"},
method="POST")
with _ur.urlopen(req, timeout=30) as r:
resp = _json.load(r)
except _ue.HTTPError as e:
if e.code == 404:
print(c_dim(" VM: no dm_followup record (bl-only follow-up)."))
else:
try:
detail = e.read().decode("utf-8", errors="ignore")[:120]
except Exception:
detail = ""
print(c_warn(" VM cancel failed: HTTP %s %s" % (e.code, detail)))
return
except Exception as e:
print(c_warn(" VM cancel failed: %s" % e))
return
if resp.get("canceled"):
print(c_green(" VM: follow-up canceled (%s)."
% resp.get("request_id")))
else:
print(c_dim(" VM: %s." % resp.get("reason", "not canceled")))
# ---------------------------------------------------------------------------
# Domain: PIPELINE (Multi-Agent Workflow Pipelines)
# ---------------------------------------------------------------------------
def cmd_pipeline_run(args):
name = args.name.strip()
job_file = JOBS_DIR / f"{name}.json"
if not job_file.exists():
print(c_red(f"Error: Pipeline root job '{name}.json' not found in jobs/"), file=sys.stderr)
sys.exit(1)
import pipeline_engine
run_id = pipeline_engine.generate_pipeline_run_id(name)
entry = pipeline_engine.create_pipeline(name, run_id=run_id)
target = entry.get("target")
print(c_bold(f"\n=== LAUNCHING MULTI-AGENT PIPELINE [{name}] ==="))
print(f" Run ID: {c_cyan(run_id)}")
print(f" Target: {c_yellow(target)}")
print(f" Step 1: {c_bold(name)}\n")
cmd = [sys.executable, str(JOB_DISPATCH_PY), name, "--pipeline-run", run_id, "--step-n", "1"]
if getattr(args, "dry_run", False):
cmd.append("--dry-run")
res = subprocess.run(cmd)
if res.returncode == 0:
print(c_green(f"\n✔ Pipeline '{run_id}' dispatched step 1 successfully."))
print(c_dim(" Track with: super pipeline status | super pipeline history\n"))
else:
print(c_red(f"\n✖ Step 1 dispatch failed with code {res.returncode}"), file=sys.stderr)
sys.exit(res.returncode)
def cmd_pipeline_status(args):
import pipeline_engine
active = pipeline_engine.list_active_pipelines()
if getattr(args, "json", False):
print(json.dumps(active, indent=2))
return
now_str = datetime.now().strftime("%H:%M:%S")
print(f"\n=== ACTIVE MULTI-AGENT PIPELINES === ({now_str} local)\n")
if not active:
print(c_dim(" (no active pipeline runs currently executing)"))
print(c_dim(" Launch one with: super pipeline run <job_name>\n"))
return
headers = ["RUN ID", "PIPELINE", "TARGET", "STEP", "CURRENT AGENT", "ACTIVE JOB ID", "STARTED", "STATUS"]
rows = []
for p in active:
run_id = p.get("run_id", "-")
p_name = p.get("pipeline_name", "-")
target = p.get("target", "-")
cur_step_n = p.get("current_step", 1)
steps = p.get("steps", [])
last_step = steps[-1] if steps else {}
agent = last_step.get("agent", "-")
job_id = last_step.get("job_id", "-")
created_at = p.get("created_at")
rows.append([
c_bold(run_id),
p_name,
c_yellow(target),
f"Step {cur_step_n}",
c_cyan(agent),
c_dim(job_id[:16] + "…") if len(job_id) > 16 else job_id,
parse_relative_time(created_at) if created_at else c_dim("-"),
badge_ok("RUNNING"),
])
print_table(headers, rows)
# Details on steps for active runs
for p in active:
print(c_bold(f"\n Active Pipeline Details [{p.get('run_id')}]:"))
for s in p.get("steps", []):
st = s.get("status", "unknown")
st_badge = badge_ok("COMPLETED") if st == "completed" else badge_warn("RUNNING") if st == "dispatched" else badge_err(st.upper())
res_snippet = s.get("result", "")
res_disp = f" → {c_dim(res_snippet[:70])}" if res_snippet else ""
print(f" • Step {s.get('step_n')}: {c_bold(s.get('job_name'))} ({s.get('agent')}) [{st_badge}]{res_disp}")
print(c_dim("\n Commands: super pipeline run <name> | super pipeline history\n"))
def cmd_pipeline_history(args):
import pipeline_engine
limit = getattr(args, "limit", 20) or 20
history = pipeline_engine.list_pipeline_history(limit=limit)
if getattr(args, "json", False):
print(json.dumps(history, indent=2))
return
now_str = datetime.now().strftime("%H:%M:%S")
print(f"\n=== PIPELINE RUN HISTORY (Recent {len(history)}) === ({now_str} local)\n")
if not history:
print(c_dim(" (no pipeline history found)"))
return
headers = ["RUN ID", "PIPELINE", "TARGET", "STEPS", "STARTED", "COMPLETED", "STATUS"]
rows = []
for p in history:
run_id = p.get("run_id", "-")
p_name = p.get("pipeline_name", "-")
target = p.get("target", "-")
steps = p.get("steps", [])
step_summary = f"{len(steps)} step{'s' if len(steps) != 1 else ''}"
created_at = p.get("created_at")
completed_at = p.get("completed_at")
st = p.get("status", "unknown")
if st == "completed":
badge = badge_ok("COMPLETED")
elif st == "running":
badge = badge_warn("RUNNING")
else:
badge = badge_err("FAILED")
rows.append([
c_bold(run_id),
p_name,
c_yellow(target),
step_summary,
parse_relative_time(created_at) if created_at else c_dim("-"),
parse_relative_time(completed_at) if completed_at else c_dim("active"),
badge,
])
print_table(headers, rows)
print(c_dim("\n Commands: super pipeline status | super pipeline run <name>\n"))
def cmd_pipeline_stop(args):
import pipeline_engine
run_id = args.run_id.strip()
reason = getattr(args, "reason", "cancelled_by_operator") or "cancelled_by_operator"
entry = pipeline_engine.stop_pipeline(run_id, reason=reason)
if not entry:
print(c_red(f"Error: Pipeline run '{run_id}' not found."), file=sys.stderr)
sys.exit(1)
print(c_green(f"✔ Pipeline run '{entry.get('run_id')}' stopped/canceled."))
if getattr(args, "json", False):
print(json.dumps(entry, indent=2))
def cmd_pipeline_prune(args):
import pipeline_engine
max_age = getattr(args, "max_age", 1) or 1
pruned = pipeline_engine.prune_pipelines(max_age_hours=max_age)
if pruned:
print(c_green(f"✔ Pruned {len(pruned)} stale pipeline runs (> {max_age}h): {', '.join(pruned)}"))
else:
print(c_dim(f"✔ No stale running pipelines found (> {max_age}h)."))
if getattr(args, "json", False):
print(json.dumps({"pruned": pruned, "count": len(pruned)}, indent=2))
# ---------------------------------------------------------------------------
# Domain: DEPLOY (Multi-Agent Pipelines, Subagent Spawning, and Oversight)
# ---------------------------------------------------------------------------
def cmd_deploy(args):
action = getattr(args, "action", None)
if action == "subagent":
agent = args.agent
title = args.title or "subagent-task"
prompt = args.prompt
wait_val = getattr(args, "wait", 30)
wait = 30 if wait_val is None else int(wait_val)
print(c_bold(f"\n=== DEPLOYING SUBAGENT ON [{agent}] ==="))
print(f" Agent: {c_cyan(agent)}")
print(f" Title: {c_yellow(title)}")
print(f" Prompt: {c_dim(prompt[:90])}...\n")
import muse_hybrid
res, err = muse_hybrid.start_session(agent, title=title)
if err or not res:
print(c_err(f"✖ Failed to spawn subagent session: {err}"), file=sys.stderr)
sys.exit(1)
session_id = res.get("session_id")
print(c_green(f"✔ Subagent session spawned: {session_id}"))
try:
import subagent_tracker
subagent_tracker.register_session(agent, session_id, title=title, prompt=prompt)
except Exception:
pass
if prompt:
print(f" Dispatching task prompt to subagent (waiting up to {wait}s)...")
send_res, send_err = muse_hybrid.send_message(agent, prompt, thread_id=session_id, wait=wait)
if send_err:
print(c_warn(f"Notice: {send_err}"))
else:
print(c_green("✔ Task prompt delivered."))
# Fetch fresh history to show assistant's initial work
msgs, _ = muse_hybrid.get_history(agent, thread_id=session_id, limit=5)
if msgs:
for m in msgs:
role = m.get("role")
if role == "assistant":
print(f"\n{c_bold('--- Subagent Response ---')}\n{m.get('text')}\n")
print(f" Oversee anytime with: {c_cyan(f'box thread view {agent} {session_id}')}\n")
return
# Pipeline deployment
pipe_name = getattr(args, "name", None)
if not pipe_name:
print(c_err("Error: Specify pipeline name (e.g. box deploy pipeline pipe-demo-step1)"), file=sys.stderr)
sys.exit(1)
args.name = pipe_name
cmd_pipeline_run(args)
# ---------------------------------------------------------------------------
# Domain: LOOP (Intrinsic Loop Strategy, Health, Break Taxonomy, and Control)
# ---------------------------------------------------------------------------
def cmd_loop_status(args):
agent = getattr(args, "agent", None)
status_filter = getattr(args, "status", None)
limit = getattr(args, "n", 30)
from gravity import reconstruct_loops
loops = reconstruct_loops(limit=limit, agent=agent, status_filter=status_filter)
if getattr(args, "json", False):
print(json.dumps({"ok": True, "loops": loops, "count": len(loops)}, indent=2))
return
prog = Path(sys.argv[0]).name if sys.argv and sys.argv[0] else "super"
if prog.endswith(".py"):
prog = "super"
print("\n" + c_bold(f"=== INTRINSIC LOOPS ({len(loops)} tracked) ===") + "\n")
headers = ["LOOP ID", "AGENT", "TARGET", "PURPOSE", "STATE", "AGE", "DEADLINE", "NUDGES", "DETAIL"]
rows = []
now = datetime.now(timezone.utc)
for l in loops:
lid = l.get("loop_id", "-")[:8]
ag = l.get("agent", "-")
tgt = l.get("target", "main")
if len(tgt) > 16:
tgt = tgt[:14] + ".."
purp = l.get("purpose", "-")
st = l.get("state", "LANDED")
if st == "FIRING":
st_badge = c_cyan("○ FIRING")
elif st == "LANDED":
st_badge = c_blue("● LANDED")
elif st == "SEEN":
st_badge = c_yellow("● SEEN")
elif st in ("ANSWERED", "CLOSED"):
st_badge = badge_ok(st)
elif st == "NUDGED":
st_badge = badge_warn("NUDGED")
elif st == "ESCALATED":
st_badge = badge_err("ESCALATED")
elif st == "BROKEN":
st_badge = badge_err("BROKEN")
else:
st_badge = badge_dim(st)
sent_rel = parse_relative_time(l.get("sent_at", ""))
dl_str = l.get("deadline", "")
if dl_str:
try:
dl_dt = datetime.fromisoformat(dl_str.replace("Z", "+00:00"))
diff_s = int((dl_dt - now).total_seconds())
if diff_s > 0:
dl_disp = c_green(f"in {diff_s // 60}m")
else:
dl_disp = c_red(f"{abs(diff_s) // 60}m ago")
except Exception:
dl_disp = dl_str[:10]
else:
dl_disp = c_dim("-")
nudges = f"{l.get('nudges_sent', 0)}/{l.get('nudges_allowed', 2)}"
detail = l.get("summary", "")[:30]
rows.append([c_bold(lid), ag, tgt, purp, st_badge, sent_rel, dl_disp, nudges, c_dim(detail)])
print_table(headers, rows)
print(c_dim(f"\n Commands: {prog} loop health | {prog} loop breaks | {prog} loop strat | {prog} loop vars | {prog} loop close <id>\n"))
def cmd_loop_health(args):
threshold = getattr(args, "threshold", None)
from gravity import get_fleet_loop_health
data = get_fleet_loop_health(threshold=threshold)
if getattr(args, "json", False):
print(json.dumps({"ok": True, **data}, indent=2))
return
summary = data.get("summary", {})
t_val = summary.get("threshold", 0.5)
print("\n" + c_bold(f"=== INTRINSIC FLEET LOOP HEALTH (Threshold: {t_val:.2f}) ===") + "\n")
headers = ["AGENT", "STATUS", "HEALTH RATIO", "LANDED", "ANSWERED", "VERDICT"]
rows = []
for ag, info in sorted(data.get("agents", {}).items()):
st = info.get("status", "IDLE")
if st == "HEALTHY":
badge = badge_ok("HEALTHY")
verd = c_green("PASS")
elif st == "DEGRADED":
badge = badge_err("DEGRADED")
verd = c_red("FAIL")
else:
badge = badge_dim("IDLE")
verd = c_dim("NO TRAFFIC")
hpct = info.get("health_pct", "-")
landed = str(info.get("landed", 0))
answered = str(info.get("answered", 0))
rows.append([c_bold(ag), badge, hpct, landed, answered, verd])
print_table(headers, rows)
ov = summary.get("overall_health_pct", "-")
tot_l = summary.get("total_landed", 0)
tot_a = summary.get("total_answered", 0)
fleet_st = c_green("HEALTHY") if summary.get("healthy") else c_red("DEGRADED")
print(f"\n {c_bold('Overall Fleet Health')}: {ov} ({tot_a}/{tot_l} loops closed/answered) — {fleet_st}\n")
def cmd_loop_breaks(args):
from gravity import diagnose_breaks
breaks = diagnose_breaks()
if getattr(args, "json", False):
print(json.dumps({"ok": True, "breaks": breaks, "count": len(breaks)}, indent=2))
return
if not breaks:
print("\n" + c_green("✔ All intrinsic loops and tracking mechanisms healthy. No breaks detected.") + "\n")
return
print("\n" + c_bold(f"=== INTRINSIC LOOP BREAK DIAGNOSTICS ({len(breaks)} issues) ===") + "\n")
headers = ["SEVERITY", "BREAK TYPE", "COMPONENT / AGENT", "DIAGNOSIS", "REMEDIATION"]
rows = []
for b in breaks:
sev = badge_err(b.get("severity", "CRITICAL")) if b.get("severity") == "CRITICAL" else badge_warn("WARNING")
btype = b.get("type", "-")
target = b.get("component") or b.get("agent") or "-"
diag = b.get("detail", "")
rem = b.get("remedy", "")
rows.append([sev, c_bold(btype), target, diag, c_dim(rem)])
print_table(headers, rows)
print()
def cmd_loop_strat(args):
subact = getattr(args, "strat_action", None) or "show"
from modulate import get_all_strategies, set_strategy_override, reset_strategy_override, modulate, render_tags, InputType
prog = Path(sys.argv[0]).name if sys.argv and sys.argv[0] else "super"
if prog.endswith(".py"):
prog = "super"
if subact == "show":
strats = get_all_strategies()
filter_type = getattr(args, "type", None)
if filter_type:
strats = [s for s in strats if s.get("input_type") == filter_type.lower()]
if getattr(args, "json", False):
print(json.dumps({"ok": True, "strategies": strats}, indent=2))
return
print("\n" + c_bold("=== INTRINSIC LOOP STRATEGY MATRIX (MODULATION) ===") + "\n")
headers = ["INPUT TYPE", "SUBTYPE", "AGENT", "TRACK", "PRIORITY", "TIMEOUT", "NUDGES", "ESCALATE", "SOURCE"]
rows = []
for s in strats:
itype = s.get("input_type", "")
sub = s.get("subtype") or "-"
ag = s.get("agent") or "-"
tr = str(s.get("track", "-"))
prio = s.get("priority", "normal")
if prio == "critical":
prio_disp = c_red("CRITICAL")
elif prio == "important":
prio_disp = c_yellow("IMPORTANT")
elif prio == "routine":
prio_disp = c_dim("ROUTINE")
else:
prio_disp = c_cyan(prio.upper())
timeout = f"{s.get('timeout_s', 0)}s"
nudges = str(s.get("nudges", 0))
esc = s.get("escalate") or "-"
src = c_yellow("OVERRIDE") if s.get("is_override") else c_dim("BUILTIN")
rows.append([c_bold(itype), sub, ag, tr, prio_disp, timeout, nudges, esc, src])
print_table(headers, rows)
print(c_dim(f"\n Modify: {prog} loop strat set <type> [--subtype S] [--agent A] [--priority P] [--timeout S] [--nudges N] [--escalate AGENT]\n"))
elif subact == "set":
itype = args.type
sub = getattr(args, "subtype", None)
agent = getattr(args, "agent", None)
res = set_strategy_override(
itype, sub, agent,
track=getattr(args, "track", None),
priority=getattr(args, "priority", None),
timeout_s=getattr(args, "timeout", None),
nudges=getattr(args, "nudges", None),
escalate=getattr(args, "escalate", None),
by="super"
)
target_desc = f"{itype}:{sub or '*'}:{agent or '*'}" if agent else f"{itype}:{sub or '*'}"
print(c_green(f"✔ Updated loop strategy override for '{target_desc}'."))
if getattr(args, "json", False):
print(json.dumps({"ok": True, "strategy": res}, indent=2))
elif subact == "reset":
itype = args.type
sub = getattr(args, "subtype", None)
agent = getattr(args, "agent", None)
ok = reset_strategy_override(itype, sub, agent, by="super")
target_desc = f"{itype}:{sub or '*'}:{agent or '*'}" if agent else f"{itype}:{sub or '*'}"
if ok:
print(c_green(f"✔ Reset loop strategy override for '{target_desc}' to default."))
else:
print(c_yellow(f"No override was active for '{target_desc}'."))
elif subact == "eval":
itype_str = args.type
sub = getattr(args, "subtype", None)
agent = getattr(args, "agent", None)
actionable = getattr(args, "actionable", False)
try:
itype = InputType(itype_str.lower())
except ValueError:
itype = InputType.MANUAL
pol = modulate(itype, sub.upper() if sub else None, agent=agent, actionable=actionable)
if getattr(args, "json", False):
print(json.dumps({"ok": True, "policy": pol.__dict__ if pol else None}, indent=2))
return
if pol:
print(f"\n{c_bold('Resolved Policy:')}")
print(f" Input Type : {pol.input_type.value}")
print(f" Subtype : {pol.subtype or '-'}")
print(f" Agent Scope: {agent or '-'}")
print(f" Priority : {pol.priority.value}")
print(f" Timeout : {pol.timeout_s}s")
print(f" Max Nudges : {pol.nudges}")
print(f" Escalate To: {pol.escalate or '-'}")
print(f" Rendered Tags: {c_cyan(render_tags(pol))}\n")
else:
print(f"\n{c_dim('Policy: Non-tracked input (no follow-up record created).')}\n")
def cmd_loop_vars(args):
subact = getattr(args, "vars_action", None) or "list"
from variables import Variables
v = Variables()
prog = Path(sys.argv[0]).name if sys.argv and sys.argv[0] else "super"
if prog.endswith(".py"):
prog = "super"
if subact == "list":
if getattr(args, "json", False):
print(json.dumps({"ok": True, "variables": v.all(), "schemas": {n: v.schema(n) for n in v.names()}}, indent=2))
return
names = v.names()
print("\n" + c_bold(f"=== RUNTIME CONTROL VARIABLES ({len(names)} registered) ===") + "\n")
headers = ["VARIABLE NAME", "VALUE", "DEFAULT", "UNIT", "RANGE", "DESCRIPTION"]
rows = []
for name in names:
val = v.get(name)
spec = v.schema(name)
d_val = spec.get("default")
unit = spec.get("unit", "-")
min_v = spec.get("min")
max_v = spec.get("max")
rng = f"{min_v}..{max_v}" if min_v is not None and max_v is not None else "-"
desc = spec.get("description", "")[:40]
val_str = str(val)
if val != d_val:
val_disp = c_cyan(f"{val_str} *")
else:
val_disp = val_str
rows.append([c_bold(name), val_disp, str(d_val), unit, rng, c_dim(desc)])
print_table(headers, rows)
print(c_dim(f"\n Adjust: {prog} loop vars set <name> <value> | Reset: {prog} loop vars reset <name> (* = modified)\n"))
elif subact == "get":
name = args.name
try:
val = v.get(name)
spec = v.schema(name)
if getattr(args, "json", False):
print(json.dumps({"ok": True, "name": name, "value": val, "schema": spec}, indent=2))
return
print(f"\n{c_bold('Variable: ' + name)}")
print(f" Current Value : {c_cyan(str(val))}")
print(f" Default Value : {spec.get('default')}")
print(f" Type : {spec.get('type')}")
print(f" Unit : {spec.get('unit', '-')}")
if "min" in spec and "max" in spec:
print(f" Allowed Range : {spec.get('min')} .. {spec.get('max')}")
print(f" Description : {spec.get('description')}\n")
except Exception as e:
print(c_red(f"Error: {e}"), file=sys.stderr)
sys.exit(1)
elif subact == "set":
name = args.name
val_raw = args.value
try:
spec = v.schema(name)
vtype = spec.get("type", "str")
if vtype == "int":
val = int(val_raw)
elif vtype == "float":
val = float(val_raw)
elif vtype == "bool":
val = val_raw.lower() in ("true", "1", "yes")
else:
val = val_raw
new_val = v.set(name, val, by="super")
print(c_green(f"✔ Variable '{name}' updated to {new_val}."))
if getattr(args, "json", False):
print(json.dumps({"ok": True, "name": name, "value": new_val}, indent=2))
except Exception as e:
print(c_red(f"Error setting variable: {e}"), file=sys.stderr)
sys.exit(1)
elif subact == "reset":
name = args.name
try:
def_val = v.reset(name, by="super")
print(c_green(f"✔ Variable '{name}' reset to default ({def_val})."))
if getattr(args, "json", False):
print(json.dumps({"ok": True, "name": name, "value": def_val}, indent=2))
except Exception as e:
print(c_red(f"Error resetting variable: {e}"), file=sys.stderr)
sys.exit(1)
elif subact == "history":
name = getattr(args, "name", None)
limit = getattr(args, "limit", 20)
entries = v.history(name=name, limit=limit)
if getattr(args, "json", False):
print(json.dumps({"ok": True, "history": entries, "count": len(entries)}, indent=2))
return
if not entries:
print("\n" + c_dim(" (no variable change history found)") + "\n")
return
print("\n" + c_bold(f"=== VARIABLE MODIFICATION AUDIT HISTORY ({len(entries)} entries) ===") + "\n")
headers = ["TIMESTAMP", "VARIABLE", "ACTION", "OLD VALUE", "NEW VALUE", "CHANGED BY"]
rows = []
for e in entries:
act = e.get("action", "")
if act == "set":
act_badge = c_cyan("SET")
elif act == "reset":
act_badge = c_yellow("RESET")
elif act == "rollback":
act_badge = c_magenta("ROLLBACK")
else:
act_badge = act
rows.append([
e.get("ts", "")[:19].replace("T", " "),
c_bold(e.get("name", "")),
act_badge,
str(e.get("old_value", "-")),
c_bold(str(e.get("new_value", "-"))),
e.get("by", "-")
])
print_table(headers, rows)
print(c_dim(f"\n Rollback: {prog} loop vars rollback <name> [--revision REV]\n"))
elif subact == "rollback":
name = args.name
rev = getattr(args, "revision", None)
try:
new_val = v.rollback(name, revision=rev, by="super")
print(c_green(f"✔ Variable '{name}' rolled back to {new_val}."))
if getattr(args, "json", False):
print(json.dumps({"ok": True, "name": name, "value": new_val}, indent=2))
except Exception as e:
print(c_red(f"Error rolling back variable: {e}"), file=sys.stderr)
sys.exit(1)
def cmd_loop_close(args):
dm_id = args.dm_id.strip()
cmd = [sys.executable, str(BIN_DIR / "box-ctl.py"), "loop-resolve", dm_id]
if getattr(args, "note", None):
cmd.append(args.note)
res = subprocess.run(cmd, capture_output=True, text=True)
try:
data = json.loads(res.stdout)
if data.get("ok"):
print(c_green(f"✔ Loop {dm_id} marked resolved/closed."))
else:
print(c_red(f"Error closing loop: {data.get('error')}"), file=sys.stderr)
except Exception:
print(res.stdout or res.stderr)
def cmd_loop_nudge(args):
dm_id = args.dm_id.strip()
if not FOLLOWUPS_FILE.exists():
print(c_red("followups.json not found."), file=sys.stderr)
return
try:
with open(FOLLOWUPS_FILE, "r") as f:
fdata = json.load(f)
rec = fdata.get(dm_id)
if not rec:
print(c_red(f"No pending follow-up found for ID '{dm_id}'."), file=sys.stderr)
return
rec["deadline"] = datetime.now(timezone.utc).isoformat()
with open(FOLLOWUPS_FILE, "w") as f:
json.dump(fdata, f, indent=2)
print(c_green(f"✔ Armed immediate nudge for loop {dm_id}. Running sweeper..."))
subprocess.run([sys.executable, str(FOLLOWUP_SWEEPER_PY), "--once"])
except Exception as e:
print(c_red(f"Error nudging loop: {e}"), file=sys.stderr)
def cmd_loop_sweep(args):
cmd_followup_sweep(args)
def cmd_loop_harvest(args):
cmd_harvest_run(args)
def cmd_loop_remediate(args):
dry_run = getattr(args, "dry_run", False)
from gravity import remediate_breaks
res = remediate_breaks(dry_run=dry_run)
if getattr(args, "json", False):
print(json.dumps({"ok": True, **res}, indent=2))
return
mode_str = c_yellow("DRY RUN (Simulated)") if dry_run else c_green("ACTIVE RECOVERY")
print("\n" + c_bold(f"=== PROGRESSIVE INTRINSIC LOOP REMEDIATION [{mode_str}] ===") + "\n")
remediated = res.get("remediated", [])
escalated = res.get("escalated", [])
if remediated:
print(c_bold(f"Auto-Remediated Soft Breakages ({len(remediated)}):"))
headers = ["ACTION", "LOOP ID", "AGENT", "REMEDIATION DETAIL"]
rows = []
for r in remediated:
act_badge = c_green("✔ " + r.get("action", ""))
rows.append([act_badge, c_bold(r.get("loop_id", "-")[:8]), r.get("agent", "-"), r.get("detail", "")])
print_table(headers, rows)
print()
else:
print(c_dim(" ✔ No soft breakages require remediation at this time.\n"))
if escalated:
print(c_red(c_bold(f"⚠ Escalated Hard Failures ({len(escalated)} - Manual Intervention Required):")))
headers = ["SEVERITY", "BREAK TYPE", "TARGET", "ISSUE DETAIL", "REMEDY"]
rows = []
for esc in escalated:
sev = badge_err("CRITICAL") if esc.get("severity") == "CRITICAL" else badge_warn("WARNING")
rows.append([sev, c_bold(esc.get("type", "-")), esc.get("agent") or esc.get("component") or "-", esc.get("detail", ""), c_dim(esc.get("remedy", ""))])
print_table(headers, rows)
print()
# ---------------------------------------------------------------------------
# CLI Argument Parser Setup
# ---------------------------------------------------------------------------
def cmd_ssh_mint(args):
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "ssh-mint", args.name]
if args.force:
cmd.append("--force")
res = subprocess.run(cmd, capture_output=True, text=True)
if args.json:
sys.stdout.write(res.stdout)
return
try:
d = json.loads(res.stdout)
if d.get("ok"):
print(c_green(f"✓ Minted ed25519 SSH keypair for '{args.name}'"))
print(f" Private key: {d.get('private_key')}")
print(f" Public key: {d.get('public_key')}")
print(f" Public key text: {d.get('public_key_content')}")
print(c_dim(" Registered into: ") + ", ".join(d.get("registered_in", [])))
else:
print(c_red(f"✗ Failed: {d.get('error')} ({d.get('code')})"))
except Exception:
sys.stdout.write(res.stdout)
def cmd_ssh_list(args):
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "ssh-list"]
res = subprocess.run(cmd, capture_output=True, text=True)
if args.json:
sys.stdout.write(res.stdout)
return
try:
d = json.loads(res.stdout)
if d.get("ok"):
keys = d.get("keys", [])
print("\n" + c_bold(f"=== MANAGED SSH KEYS ({len(keys)}) ===") + "\n")
headers = ["NAME", "PRIVATE KEY", "PUB STATUS", "PUBLIC KEY PREVIEW"]
rows = []
for k in keys:
pub_txt = k.get("public_key") or ""
preview = (pub_txt[:32] + "..." + pub_txt[-16:]) if len(pub_txt) > 48 else pub_txt
rows.append([c_cyan(k.get("name")), k.get("private_key"), c_green("✓ present") if k.get("has_public") else c_red("missing"), preview])
print_table(headers, rows)
print("\n" + c_dim(" Mint new key: box ssh mint <name> [--force]") + "\n")
else:
print(c_red(f"✗ Failed: {d.get('error')}"))
except Exception:
sys.stdout.write(res.stdout)
def cmd_ssh_show(args):
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "ssh-show", args.name]
res = subprocess.run(cmd, capture_output=True, text=True)
if args.json:
sys.stdout.write(res.stdout)
return
try:
d = json.loads(res.stdout)
if d.get("ok"):
print(c_bold(f"=== SSH KEY DETAILS: {args.name} ==="))
print(f" Private key: {d.get('private_key')}")
print(f" Public key: {d.get('public_key')}")
print(f" Fingerprint: {d.get('fingerprint')}")
print(f"\n Public Key:\n {c_cyan(d.get('public_key_content'))}\n")
else:
print(c_red(f"✗ Failed: {d.get('error')}"))
except Exception:
sys.stdout.write(res.stdout)
def cmd_swarm_list(args):
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "swarm-list"]
res = subprocess.run(cmd, capture_output=True, text=True)
if args.json:
sys.stdout.write(res.stdout)
return
try:
d = json.loads(res.stdout)
if d.get("ok"):
swarms = d.get("swarms", [])
print("\n" + c_bold(f"=== SWARMS ({len(swarms)}) ===") + "\n")
headers = ["ID", "STATUS", "LABEL", "TOTAL", "DONE", "RUNNING", "FAILED", "CREATED"]
rows = []
for sw in swarms:
st = sw.get("status")
st_color = c_green(st) if st == "completed" else (c_yellow(st) if st in ("running", "partial") else c_dim(st))
rows.append([
c_cyan(sw.get("swarm_id", "")[:18]),
st_color,
(sw.get("label") or "-")[:20],
str(sw.get("count", 0)),
str(sw.get("done", 0)),
str(sw.get("running", 0)),
str(sw.get("failed", 0)),
(sw.get("created_ts") or "")[:16]
])
print_table(headers, rows)
print("\n" + c_dim(" Spawn swarm: box swarm spawn <count> <task> [--label L]") + "\n")
else:
print(c_red(f"✗ Failed: {d.get('error')}"))
except Exception:
sys.stdout.write(res.stdout)
def cmd_swarm_status(args):
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "swarm-status", args.swarm_id]
res = subprocess.run(cmd, capture_output=True, text=True)
if args.json:
sys.stdout.write(res.stdout)
return
try:
d = json.loads(res.stdout)
if d.get("ok"):
print(c_bold(f"\n=== SWARM STATUS: {args.swarm_id} ==="))
sw = d.get("swarm", {})
print(f" Status: {sw.get('status')}")
print(f" Counts: {d.get('counts')}")
slots = sw.get("slots", [])
print(f"\n Slots ({len(slots)}):")
for s in slots:
res_txt = s.get("result")
res_prev = (str(res_txt)[:50] + "...") if res_txt and len(str(res_txt)) > 50 else str(res_txt)
print(f" Slot {s.get('slot')}: [{s.get('status')}] agent={s.get('agent_id')} subagent={s.get('subagent_session_id')} res={res_prev}")
print()
else:
print(c_red(f"✗ Failed: {d.get('error')}"))
except Exception:
sys.stdout.write(res.stdout)
def cmd_swarm_spawn(args):
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "swarm-spawn", str(args.count), args.task]
if args.label:
cmd.extend(["--label", args.label])
res = subprocess.run(cmd, capture_output=True, text=True)
if args.json:
sys.stdout.write(res.stdout)
return
try:
d = json.loads(res.stdout)
if d.get("ok"):
print(c_green(f"✓ Spawned swarm: {d.get('swarm_id')} ({d.get('count')} slots)"))
else:
print(c_red(f"✗ Failed: {d.get('error')} ({d.get('code')})"))
except Exception:
sys.stdout.write(res.stdout)
def cmd_swarm_prune(args):
cmd = ["python3", str(BIN_DIR / "box-ctl.py"), "swarm-prune", "--stale-hours", str(args.stale_hours)]
if args.confirm:
cmd.append("--confirm")
res = subprocess.run(cmd, capture_output=True, text=True)
if args.json:
sys.stdout.write(res.stdout)
return
try:
d = json.loads(res.stdout)
if d.get("ok"):
print(c_green(f"✓ Pruned {d.get('archived_count')} swarms older than {d.get('stale_hours')}h to 'archived'"))
else:
print(c_yellow(f"! {d.get('error')} (use --confirm)"))
except Exception:
sys.stdout.write(res.stdout)
def build_parser():
common = argparse.ArgumentParser(add_help=False)
common.add_argument("--json", action="store_true", help="Output machine-readable JSON")
prog_name = Path(sys.argv[0]).name if sys.argv and sys.argv[0] else "super"
if prog_name.endswith(".py"):
prog_name = "super"
parser = argparse.ArgumentParser(
prog=prog_name,
description=f"{prog_name} — Unified Orchestrator CLI for NetVM & Box",
formatter_class=argparse.RawDescriptionHelpFormatter,
parents=[common]
)
subparsers = parser.add_subparsers(dest="domain", help="Orchestration Domain")
# Domain: FLEET
p_fleet = subparsers.add_parser("fleet", parents=[common], help="Node health, CDP status, active tabs, watch, restart")
p_fleet.add_argument("action", nargs="?", default="status", choices=["status", "watch", "restart", "cdp"])
p_fleet.add_argument("node", nargs="?", default=None, help="Target node (for restart / cdp)")
p_fleet.add_argument("--interval", type=int, default=2, help="Watch refresh interval in seconds")
# Domain: DM
p_dm = subparsers.add_parser("dm", parents=[common], help="Inter-agent DMs, work orders ([WO]), acks, live log tail")
dm_sub = p_dm.add_subparsers(dest="action")
# super dm send
p_dm_send = dm_sub.add_parser("send", parents=[common], help="Send a direct message")
p_dm_send.add_argument("--to", required=True, choices=VALID_NODES, help="Recipient node")
p_dm_send.add_argument("--from", dest="from_agent", default=DEFAULT_SENDER, help="Sender identity (default: super)")
p_dm_send.add_argument("--target", default="main", help="Target chat/sidechat (default: main)")
p_dm_send.add_argument("--allow-main-chat", action="store_true", help="Explicit override allowing send directly to Main Chat")
p_dm_send.add_argument("-w", "--wait", action="store_true", help="Wait for recipient agent to reply")
p_dm_send.add_argument("-t", "--timeout", type=int, default=60, help="Wait timeout in seconds (default: 60)")
p_dm_send.add_argument("--expect-reply", action="store_true", help="Arm follow-up tracking")
p_dm_send.add_argument("--reply-timeout", type=int, default=None, help="Timeout in seconds")
p_dm_send.add_argument("--no-sign", action="store_true", help="Send without cryptographic SSH signature")
p_dm_send.add_argument("message", help="Message body")
# super dm chat
p_dm_chat = dm_sub.add_parser("chat", parents=[common], help="Interactive conversational chat session with an agent")
p_dm_chat.add_argument("agent", choices=VALID_NODES, help="Target agent to converse with")
p_dm_chat.add_argument("--target", default=None, help="Target sidechat or 'main'")
p_dm_chat.add_argument("-t", "--timeout", type=int, default=60, help="Reply timeout in seconds (default: 60)")
# super dm send-file
p_dm_sf = dm_sub.add_parser("send-file", parents=[common], help="Transfer a file to an agent and send a pointer in chat")
p_dm_sf.add_argument("--to", required=True, choices=VALID_NODES, help="Recipient node")
p_dm_sf.add_argument("--target", default=None, help="Target sidechat (default: agent primary task sidechat)")
p_dm_sf.add_argument("--note", default=None, help="Optional operator note or instructions")
p_dm_sf.add_argument("-w", "--wait", action="store_true", help="Wait for recipient agent to reply/acknowledge")
p_dm_sf.add_argument("-t", "--timeout", type=int, default=60, help="Reply timeout in seconds (default: 60)")
p_dm_sf.add_argument("file", help="Path to file to transfer")
# super dm files
p_dm_files = dm_sub.add_parser("files", parents=[common], help="List transferred files and staged payloads")
p_dm_files.add_argument("files_action", nargs="?", default="list", choices=["list", "clean"], help="Action: list (default) or clean")
p_dm_files.add_argument("--agent", choices=VALID_NODES, default=None, help="Filter by recipient agent")
p_dm_files.add_argument("--older-than", type=int, default=7, help="Retention cutoff in days for clean (default: 7)")
# super dm wo
p_dm_wo = dm_sub.add_parser("wo", parents=[common], help="Dispatch a structured Work Order ([WO:...])")
p_dm_wo.add_argument("--to", required=True, choices=VALID_NODES, help="Recipient node")
p_dm_wo.add_argument("--title", required=True, help="Work Order title")
p_dm_wo.add_argument("--priority", choices=["routine", "urgent"], default="routine", help="Priority level")
p_dm_wo.add_argument("--from", dest="from_agent", default=DEFAULT_SENDER, help="Sender identity (default: super)")
p_dm_wo.add_argument("--target", default=None, help="Target sidechat (default: agent primary task sidechat)")
p_dm_wo.add_argument("--allow-main-chat", action="store_true", help="Explicit override allowing work order in Main Chat")
p_dm_wo.add_argument("--no-wait", dest="wait", action="store_false", help="Do not wait for reply")
p_dm_wo.add_argument("-t", "--timeout", type=int, default=60, help="Wait timeout in seconds (default: 60)")
p_dm_wo.add_argument("--no-sign", action="store_true", help="Issue without cryptographic SSH signature")
p_dm_wo.add_argument("body", help="Work Order detailed description")
# super dm ack
p_dm_ack = dm_sub.add_parser("ack", parents=[common], help="Acknowledge a DM or Work Order")
p_dm_ack.add_argument("id", help="DM ID or Work Order ID")
p_dm_ack.add_argument("--to", required=True, choices=VALID_NODES, help="Recipient node")
p_dm_ack.add_argument("--from", dest="from_agent", default=DEFAULT_SENDER, help="Sender identity (default: super)")
p_dm_ack.add_argument("--target", default="main", help="Target chat (default: main)")
# super dm log
p_dm_log = dm_sub.add_parser("log", parents=[common], help="Display recent DM activity")
p_dm_log.add_argument("-n", type=int, default=25, help="Number of records to show")
p_dm_log.add_argument("--agent", choices=VALID_NODES, default=None, help="Filter by agent")
p_dm_log.add_argument("--filter", default=None, help="Search filter in text")
# super dm tail
p_dm_tail = dm_sub.add_parser("tail", parents=[common], help="Live stream incoming DMs")
p_dm_tail.add_argument("--filter", default=None, help="Search filter in text")
# super dm verify
p_dm_verify = dm_sub.add_parser("verify", parents=[common], help="Verify cryptographic signature on a DM")
p_dm_verify.add_argument("message", nargs="?", default=None, help="Message text containing signature block")
p_dm_verify.add_argument("--agent", choices=VALID_NODES, default=None, help="Scan agent's recent reads")
p_dm_verify.add_argument("--target", default=None, help="Scan target chat's recent reads")
# Domain: THREAD
p_thread = subparsers.add_parser("thread", parents=[common], help="Inspect agent main chats, sidechats, and scrollbacks")
thread_sub = p_thread.add_subparsers(dest="action")
p_thread_list = thread_sub.add_parser("list", parents=[common], help="List active threads for an agent")
p_thread_list.add_argument("agent", choices=VALID_NODES, help="Agent node to inspect")
p_thread_view = thread_sub.add_parser("view", parents=[common], help="View recent messages from an agent's thread")
p_thread_view.add_argument("agent", choices=VALID_NODES, help="Agent node to inspect")
p_thread_view.add_argument("thread_id", help="Thread UUID or 'main'")
p_thread_view.add_argument("--limit", type=int, default=15, help="Number of messages to retrieve")
# Domain: JOB
p_job = subparsers.add_parser("job", parents=[common], help="Manage scheduled jobs, systemd timers, triggers, logs")
job_sub = p_job.add_subparsers(dest="action")
p_job_list = job_sub.add_parser("list", parents=[common], help="List defined jobs")
p_job_show = job_sub.add_parser("show", parents=[common], help="Inspect full job definition and timer details")
p_job_show.add_argument("name", help="Job name")
p_job_status = job_sub.add_parser("status", parents=[common], help="Check timer status")
p_job_status.add_argument("name", nargs="?", default=None, help="Job/timer name")
p_job_create = job_sub.add_parser("create", parents=[common], help="Create or update a scheduled job")
p_job_create.add_argument("name", help="Job name (^[a-z0-9-]{1,64}$)")
p_job_create.add_argument("--agent", choices=VALID_NODES, default=None, help="Target agent")
p_job_create.add_argument("--schedule", default=None, help="Cron schedule (e.g. '*/15 * * * *' or 'manual')")
p_job_create.add_argument("--prompt", default=None, help="Prompt template")
p_job_create.add_argument("--description", default=None, help="Job description")
p_job_create.add_argument("--timeout", type=int, default=300, help="Timeout in seconds (60-3600, default: 300)")
p_job_create.add_argument("--on-failure", choices=["alert", "retry", "ignore"], default="alert", help="Failure policy")
p_job_create.add_argument("--chain-next", default=None, help="Next job name to chain on success")
p_job_create.add_argument("--sidechat-create", action="store_true", help="Dispatch in a sidechat")
p_job_create.add_argument("--sidechat-name", default=None, help="Sidechat name template")
p_job_create.add_argument("--reuse-key", default=None, help="Sidechat reuse key")
p_job_create.add_argument("--dm-target", default=None, help="Target chat or thread name")
p_job_create.add_argument("--file", default=None, help="Path to JSON file with job definition (or '-' for stdin)")
p_job_create.add_argument("--no-enable", action="store_true", help="Do not enable systemd timer immediately")
p_job_create.add_argument("--update", action="store_true", help="Allow updating an existing job")
p_job_enable = job_sub.add_parser("enable", parents=[common], help="Enable and activate a job's systemd timer")
p_job_enable.add_argument("name", help="Job name")
p_job_disable = job_sub.add_parser("disable", parents=[common], help="Stop and disable a job's systemd timer")
p_job_disable.add_argument("name", help="Job name")
p_job_delete = job_sub.add_parser("delete", parents=[common], help="Delete a job and its timer")
p_job_delete.add_argument("name", help="Job name")
p_job_delete.add_argument("-y", "--yes", action="store_true", help="Bypass confirmation prompt")
p_job_delete.add_argument("--force", action="store_true", help="Force deletion even if timer error")
p_job_run = job_sub.add_parser("run", parents=[common], help="Trigger a job immediately")
p_job_run.add_argument("name", help="Job name to trigger")
p_job_run.add_argument("-f", "--follow", action="store_true", help="Stream live dispatch and execution events")
p_job_log = job_sub.add_parser("log", parents=[common], help="View recent job execution events")
p_job_log.add_argument("-n", type=int, default=20, help="Number of events")
p_job_log.add_argument("name", nargs="?", default=None, help="Filter by job name")
# Domain: WEB
p_web = subparsers.add_parser("web", parents=[common], help="Probe VM web surfaces (box.muse-dev.online), auth, sync")
web_sub = p_web.add_subparsers(dest="action")
web_sub.add_parser("health", parents=[common], help="Check Google Cloud VM web surface health")
web_sub.add_parser("test-auth", parents=[common], help="Verify operator auth gating on the VM")
web_sub.add_parser("sync", parents=[common], help="Check local log size and sync integrity")
# Domain: CRED
p_cred = subparsers.add_parser("cred", parents=[common], help="Credential console API & client onboarding")
cred_sub = p_cred.add_subparsers(dest="action")
p_c_init = cred_sub.add_parser("initiate", parents=[common], help="Initiate onboarding flow for a node")
p_c_init.add_argument("--node", required=True, help="Node label (e.g. opm, pip, 646)")
p_c_init.add_argument("--email", required=True, help="Client login email")
p_c_init.add_argument("--service", default="muse", help="Service name (default: muse)")
p_c_init.add_argument("--account-name", default=None, help="Display name hint for multi-account selector")
p_c_otp = cred_sub.add_parser("submit-otp", parents=[common], help="Submit transient OTP verification code")
p_c_otp.add_argument("--node", required=True, help="Node label")
p_c_otp.add_argument("--otp", required=True, help="6-digit verification code")
p_c_otp.add_argument("--email", default=None, help="Client email (optional)")
p_c_stat = cred_sub.add_parser("status", parents=[common], help="Query onboarding & vitality status for a node")
p_c_stat.add_argument("--node", required=True, help="Node label")
p_c_link = cred_sub.add_parser("link-instagram", parents=[common], help="Generate Tailscale one-tap portal & OAuth link for age verification")
p_c_link.add_argument("--node", required=True, help="Node label")
p_c_link.add_argument("--notify", action="store_true", help="Send email alert to operator via local MTA")
p_c_audit = cred_sub.add_parser("meta-audit", parents=[common], help="Audit Meta Accounts Center for linked profiles and email")
p_c_audit.add_argument("--node", required=True, help="Node label")
cred_sub.add_parser("list", parents=[common], help="List all registered nodes and vitality statuses")
# Domain: SSH
p_ssh = subparsers.add_parser("ssh", parents=[common], help="Mint and manage fleet SSH signing/access keys")
ssh_sub = p_ssh.add_subparsers(dest="action")
p_s_mint = ssh_sub.add_parser("mint", parents=[common], help="Mint a new ed25519 SSH keypair and register in allowed_signers")
p_s_mint.add_argument("name", help="Identity/agent name (e.g. 646, pip, dev, canary)")
p_s_mint.add_argument("--force", action="store_true", help="Overwrite existing keypair")
p_s_list = ssh_sub.add_parser("list", parents=[common], help="List managed SSH keypairs and registration status")
p_s_show = ssh_sub.add_parser("show", parents=[common], help="Show public key content and fingerprint")
p_s_show.add_argument("name", help="Identity/agent name")
# Domain: HARVEST
p_harvest = subparsers.add_parser("harvest", parents=[common], help="Readback & response harvesting engine")
harvest_sub = p_harvest.add_subparsers(dest="action")
p_h_run = harvest_sub.add_parser("run", parents=[common], help="Run a harvest cycle immediately")
p_h_run.add_argument("--agent", choices=VALID_NODES, default=None, help="Target specific agent")
p_h_run.add_argument("--dry-run", action="store_true", help="Scrape without persisting watermarks")
p_h_status = harvest_sub.add_parser("status", parents=[common], help="Display active watermarks and harvest metrics")
p_h_tail = harvest_sub.add_parser("tail", parents=[common], help="Live stream ingested chat messages")
p_h_tail.add_argument("--filter", default=None, help="Search filter in text")
# Domain: FOLLOWUP
p_followup = subparsers.add_parser("followup", parents=[common], help="Scheduled follow-up deadline tracking and nudges")
followup_sub = p_followup.add_subparsers(dest="action")
p_f_list = followup_sub.add_parser("list", parents=[common], help="List tracked follow-ups and deadlines")
p_f_list.add_argument("--all", action="store_true", help="Include resolved and historical records")
p_f_sweep = followup_sub.add_parser("sweep", parents=[common], help="Trigger deadline check and nudge dispatch immediately")
p_f_sweep.add_argument("--dry-run", action="store_true", help="Simulate sweep without sending DMs")
p_f_cancel = followup_sub.add_parser("cancel", parents=[common], help="Cancel a pending follow-up")
p_f_cancel.add_argument("dm_id", help="DM ID to cancel")
# Domain: LOOP (Intrinsic Loop Management)
p_loop = subparsers.add_parser("loop", parents=[common], help="Intrinsic loop strategy, health, break taxonomy, and variables")
loop_sub = p_loop.add_subparsers(dest="action")
p_l_status = loop_sub.add_parser("status", parents=[common], help="Show active and recent intrinsic loops")
p_l_status.add_argument("--agent", choices=VALID_NODES, default=None, help="Filter by agent")
p_l_status.add_argument("--status", choices=["pending", "active", "nudged", "escalated", "closed", "resolved", "all"], default=None, help="Filter by state")
p_l_status.add_argument("-n", type=int, default=30, help="Number of loops to display")
p_l_health = loop_sub.add_parser("health", parents=[common], help="Evaluate per-agent and fleet loop health metrics")
p_l_health.add_argument("--threshold", type=float, default=None, help="Health ratio threshold (default: 0.5)")
p_l_breaks = loop_sub.add_parser("breaks", parents=[common], help="Run break detection diagnostics across loops")
# loop strat
p_l_strat = loop_sub.add_parser("strat", parents=[common], help="Inspect and configure loop modulation strategies")
strat_sub = p_l_strat.add_subparsers(dest="strat_action")
p_ls_show = strat_sub.add_parser("show", parents=[common], help="Display active modulation strategy matrix")
p_ls_show.add_argument("type", nargs="?", default=None, help="Filter by input type (wake, job, siphon, manual, health, heartbeat)")
p_ls_set = strat_sub.add_parser("set", parents=[common], help="Override loop modulation strategy")
p_ls_set.add_argument("type", help="Input type (wake, job, siphon, manual, health, heartbeat)")
p_ls_set.add_argument("--subtype", default=None, help="Subtype (e.g. overdue, canary, alert, blocker)")
p_ls_set.add_argument("--agent", choices=VALID_NODES, default=None, help="Target agent scope")
p_ls_set.add_argument("--track", choices=["true", "false", "actionable", "always", "never"], default=None, help="Track mode")
p_ls_set.add_argument("--priority", choices=["routine", "normal", "important", "critical"], default=None, help="Priority rank")
p_ls_set.add_argument("--timeout", type=int, default=None, help="Timeout in seconds before nudge")
p_ls_set.add_argument("--nudges", type=int, default=None, help="Max automatic nudges")
p_ls_set.add_argument("--escalate", default=None, help="Escalation target identity (e.g. opm, user, none)")
p_ls_reset = strat_sub.add_parser("reset", parents=[common], help="Reset loop strategy override to builtin default")
p_ls_reset.add_argument("type", help="Input type")
p_ls_reset.add_argument("--subtype", default=None, help="Subtype")
p_ls_reset.add_argument("--agent", choices=VALID_NODES, default=None, help="Target agent scope")
p_ls_eval = strat_sub.add_parser("eval", parents=[common], help="Evaluate/simulate strategy resolution for input")
p_ls_eval.add_argument("type", help="Input type")
p_ls_eval.add_argument("--subtype", default=None, help="Subtype")
p_ls_eval.add_argument("--agent", choices=VALID_NODES, default=None, help="Target agent scope")
p_ls_eval.add_argument("--actionable", action="store_true", help="Actionable flag (for wake)")
# loop vars
p_l_vars = loop_sub.add_parser("vars", parents=[common], help="Inspect and adjust runtime control variables")
vars_sub = p_l_vars.add_subparsers(dest="vars_action")
vars_sub.add_parser("list", parents=[common], help="List all registered control variables")
p_lv_get = vars_sub.add_parser("get", parents=[common], help="Inspect a specific control variable")
p_lv_get.add_argument("name", help="Variable name")
p_lv_set = vars_sub.add_parser("set", parents=[common], help="Set a control variable value")
p_lv_set.add_argument("name", help="Variable name")
p_lv_set.add_argument("value", help="New value")
p_lv_reset = vars_sub.add_parser("reset", parents=[common], help="Reset a control variable to default")
p_lv_reset.add_argument("name", help="Variable name")
p_lv_hist = vars_sub.add_parser("history", parents=[common], help="Show variable modification audit history")
p_lv_hist.add_argument("name", nargs="?", default=None, help="Variable name")
p_lv_hist.add_argument("-n", "--limit", type=int, default=20, help="Number of records")
p_lv_rb = vars_sub.add_parser("rollback", parents=[common], help="Roll back a variable to a previous value")
p_lv_rb.add_argument("name", help="Variable name")
p_lv_rb.add_argument("--revision", default=None, help="Revision step (int) or timestamp")
p_l_remed = loop_sub.add_parser("remediate", parents=[common], help="Run progressive auto-remediation for soft breaks")
p_l_remed.add_argument("--dry-run", action="store_true", help="Simulate remediation without applying fixes")
p_l_close = loop_sub.add_parser("close", parents=[common], help="Manually close/resolve a pending loop")
p_l_close.add_argument("dm_id", help="DM ID or Loop ID")
p_l_close.add_argument("--note", default=None, help="Resolution note")
p_l_nudge = loop_sub.add_parser("nudge", parents=[common], help="Trigger immediate nudge for a pending loop")
p_l_nudge.add_argument("dm_id", help="DM ID or Loop ID")
p_l_sweep = loop_sub.add_parser("sweep", parents=[common], help="Run immediate sweep cycle")
p_l_sweep.add_argument("--dry-run", action="store_true", help="Simulate sweep without sending DMs")
p_l_harvest = loop_sub.add_parser("harvest", parents=[common], help="Run immediate harvest cycle")
p_l_harvest.add_argument("--dry-run", action="store_true", help="Scrape without persisting watermarks")
# Domain: PIPELINE
p_pipe = subparsers.add_parser("pipeline", parents=[common], help="Multi-agent workflow pipelines and chained handoffs")
pipe_sub = p_pipe.add_subparsers(dest="action")
p_pipe_run = pipe_sub.add_parser("run", parents=[common], help="Start a new multi-agent pipeline run")
p_pipe_run.add_argument("name", help="Root pipeline job name (e.g. pipe-demo-step1)")
p_pipe_run.add_argument("--dry-run", action="store_true", help="Simulate pipeline without sending DMs")
p_pipe_status = pipe_sub.add_parser("status", parents=[common], help="Display active pipeline runs and step states")
p_pipe_hist = pipe_sub.add_parser("history", parents=[common], help="Show historical pipeline runs and outcomes")
p_pipe_hist.add_argument("-n", "--limit", type=int, default=20, help="Number of records to display")
p_pipe_stop = pipe_sub.add_parser("stop", parents=[common], help="Cancel/stop an active pipeline run")
p_pipe_stop.add_argument("run_id", help="Pipeline run ID (prefix match supported)")
p_pipe_stop.add_argument("--reason", default="cancelled_by_operator", help="Cancellation reason")
p_pipe_prune = pipe_sub.add_parser("prune", parents=[common], help="Prune/timeout stale running pipelines")
p_pipe_prune.add_argument("--max-age", type=float, default=1.0, help="Max age in hours (default 1.0)")
# Top-level domain aliases: strat and vars
p_strat_alias = subparsers.add_parser("strat", parents=[common], help="Shortcut for 'loop strat'")
strat_alias_sub = p_strat_alias.add_subparsers(dest="strat_action")
p_sa_show = strat_alias_sub.add_parser("show", parents=[common], help="Display active modulation strategy matrix")
p_sa_show.add_argument("type", nargs="?", default=None, help="Filter by input type")
p_sa_set = strat_alias_sub.add_parser("set", parents=[common], help="Override loop modulation strategy")
p_sa_set.add_argument("type", help="Input type")
p_sa_set.add_argument("--subtype", default=None, help="Subtype")
p_sa_set.add_argument("--agent", choices=VALID_NODES, default=None, help="Target agent scope")
p_sa_set.add_argument("--track", choices=["true", "false", "actionable", "always", "never"], default=None, help="Track mode")
p_sa_set.add_argument("--priority", choices=["routine", "normal", "important", "critical"], default=None, help="Priority rank")
p_sa_set.add_argument("--timeout", type=int, default=None, help="Timeout in seconds before nudge")
p_sa_set.add_argument("--nudges", type=int, default=None, help="Max automatic nudges")
p_sa_set.add_argument("--escalate", default=None, help="Escalation target identity")
p_sa_reset = strat_alias_sub.add_parser("reset", parents=[common], help="Reset loop strategy override")
p_sa_reset.add_argument("type", help="Input type")
p_sa_reset.add_argument("--subtype", default=None, help="Subtype")
p_sa_reset.add_argument("--agent", choices=VALID_NODES, default=None, help="Target agent scope")
p_sa_eval = strat_alias_sub.add_parser("eval", parents=[common], help="Evaluate/simulate strategy resolution")
p_sa_eval.add_argument("type", help="Input type")
p_sa_eval.add_argument("--subtype", default=None, help="Subtype")
p_sa_eval.add_argument("--agent", choices=VALID_NODES, default=None, help="Target agent scope")
p_sa_eval.add_argument("--actionable", action="store_true", help="Actionable flag")
p_vars_alias = subparsers.add_parser("vars", parents=[common], help="Shortcut for 'loop vars'")
vars_alias_sub = p_vars_alias.add_subparsers(dest="vars_action")
vars_alias_sub.add_parser("list", parents=[common], help="List all control variables")
p_va_get = vars_alias_sub.add_parser("get", parents=[common], help="Inspect a specific control variable")
p_va_get.add_argument("name", help="Variable name")
p_va_set = vars_alias_sub.add_parser("set", parents=[common], help="Set a control variable value")
p_va_set.add_argument("name", help="Variable name")
p_va_set.add_argument("value", help="New value")
p_va_reset = vars_alias_sub.add_parser("reset", parents=[common], help="Reset a control variable to default")
p_va_reset.add_argument("name", help="Variable name")
p_va_hist = vars_alias_sub.add_parser("history", parents=[common], help="Show variable modification audit history")
p_va_hist.add_argument("name", nargs="?", default=None, help="Variable name")
p_va_hist.add_argument("-n", "--limit", type=int, default=20, help="Number of records")
p_va_rb = vars_alias_sub.add_parser("rollback", parents=[common], help="Roll back a variable to a previous value")
p_va_rb.add_argument("name", help="Variable name")
p_va_rb.add_argument("--revision", default=None, help="Revision step (int) or timestamp")
# Domain: deploy (Deploy multi-agent pipelines or subagent tasks with oversight)
p_deploy = subparsers.add_parser("deploy", parents=[common], help="Deploy multi-agent task or pipeline across nodes with live oversight")
deploy_sub = p_deploy.add_subparsers(dest="action")
p_dep_pipe = deploy_sub.add_parser("pipeline", parents=[common], help="Deploy multi-agent pipeline")
p_dep_pipe.add_argument("name", help="Pipeline root job name (e.g. pipe-demo-step1)")
p_dep_pipe.add_argument("--dry-run", action="store_true", help="Simulate pipeline without sending DMs")
p_dep_sub = deploy_sub.add_parser("subagent", parents=[common], help="Spawn sub-agent session and dispatch task")
p_dep_sub.add_argument("--agent", required=True, choices=VALID_NODES, help="Agent node to spawn subagent on")
p_dep_sub.add_argument("--title", default="subagent-task", help="Title for the subagent session")
p_dep_sub.add_argument("prompt", help="Task prompt for the subagent")
p_dep_sub.add_argument("--wait", type=int, default=30, help="Seconds to wait for subagent response")
# Domain: SWARM
p_swarm = subparsers.add_parser("swarm", parents=[common], help="Manage multi-agent swarm tasks and worker allocations")
swarm_sub = p_swarm.add_subparsers(dest="action")
p_sw_list = swarm_sub.add_parser("list", parents=[common], help="List all swarms and slot rollups")
p_sw_status = swarm_sub.add_parser("status", parents=[common], help="Show status and slot details for a swarm")
p_sw_status.add_argument("swarm_id", help="Swarm identifier (sw-...)")
p_sw_spawn = swarm_sub.add_parser("spawn", parents=[common], help="Spawn a new multi-slot worker swarm")
p_sw_spawn.add_argument("count", type=int, help="Number of worker slots (1-50)")
p_sw_spawn.add_argument("task", help="Task prompt for the swarm workers")
p_sw_spawn.add_argument("--label", default=None, help="Optional tracking label")
p_sw_prune = swarm_sub.add_parser("prune", parents=[common], help="Archive stale partial/terminal swarms")
p_sw_prune.add_argument("--stale-hours", type=float, default=6.0, help="Hours of inactivity before pruning (default: 6)")
p_sw_prune.add_argument("--confirm", action="store_true", help="Confirm execution of pruning")
# Domain: subagent (Direct alias for spawning & managing subagents)
p_subagent = subparsers.add_parser("subagent", parents=[common], help="Spawn sub-agent session and dispatch task")
subagent_sub = p_subagent.add_subparsers(dest="action")
p_sub_spawn = subagent_sub.add_parser("spawn", parents=[common], help="Spawn sub-agent session and dispatch task")
p_sub_spawn.add_argument("--agent", required=True, choices=VALID_NODES, help="Agent node to spawn subagent on")
p_sub_spawn.add_argument("--title", default="subagent-task", help="Title for the subagent session")
p_sub_spawn.add_argument("prompt", help="Task prompt for the subagent")
p_sub_spawn.add_argument("--wait", type=int, default=30, help="Seconds to wait for subagent response")
# Domain: muse (fast headless gateway via muse-cli-node with isolated per-node Cloudflare WARP egress)
p_muse = subparsers.add_parser("muse", parents=[common], help="Direct headless gateway client (muse-cli-node)")
p_muse.add_argument("node", choices=VALID_NODES, help="Target agent node")
p_muse.add_argument("muse_args", nargs=argparse.REMAINDER, help="Arguments passed directly to muse-cli-node")
return parser
def main():
parser = build_parser()
if len(sys.argv) == 1:
# Default behavior with no arguments: show fleet status
sys.argv.append("fleet")
sys.argv.append("status")
args = parser.parse_args()
# Route commands
if args.domain == "fleet":
act = getattr(args, "action", "status") or "status"
if act == "status":
cmd_fleet_status(args)
elif act == "watch":
cmd_fleet_watch(args)
elif act == "restart":
if not getattr(args, "node", None):
print("Error: Specify node to restart (e.g. super fleet restart muse)", file=sys.stderr)
sys.exit(1)
cmd_fleet_restart(args)
elif act == "cdp":
if not getattr(args, "node", None):
print("Error: Specify node (e.g. super fleet cdp muse)", file=sys.stderr)
sys.exit(1)
cmd_fleet_cdp(args)
elif args.domain == "dm":
act = getattr(args, "action", None)
if not act or act == "log":
cmd_dm_log(args)
elif act == "send":
cmd_dm_send(args)
elif act == "wo":
cmd_dm_wo(args)
elif act == "ack":
cmd_dm_ack(args)
elif act == "tail":
cmd_dm_tail(args)
elif act == "chat":
cmd_dm_chat(args)
elif act == "send-file":
cmd_dm_send_file(args)
elif act == "files":
cmd_dm_files(args)
elif act == "verify":
cmd_dm_verify(args)
else:
parser.print_help()
elif args.domain == "thread":
act = getattr(args, "action", None)
if act == "list":
cmd_thread_list(args)
elif act == "view":
cmd_thread_view(args)
else:
parser.print_help()
elif args.domain == "job":
act = getattr(args, "action", None)
if not act or act == "list":
cmd_job_list(args)
elif act == "show":
cmd_job_show(args)
elif act == "create":
cmd_job_create(args)
elif act == "enable":
cmd_job_enable(args)
elif act == "disable":
cmd_job_disable(args)
elif act == "delete":
cmd_job_delete(args)
elif act == "status":
cmd_job_status(args)
elif act == "run":
cmd_job_run(args)
elif act == "log":
cmd_job_log(args)
else:
parser.print_help()
elif args.domain == "web":
act = getattr(args, "action", None)
if not act or act == "health":
cmd_web_health(args)
elif act == "test-auth":
cmd_web_test_auth(args)
elif act == "sync":
cmd_web_sync(args)
else:
parser.print_help()
elif args.domain == "cred":
act = getattr(args, "action", None)
if not act or act == "list":
cmd_cred_list(args)
elif act == "initiate":
cmd_cred_initiate(args)
elif act == "submit-otp":
cmd_cred_submit_otp(args)
elif act == "status":
cmd_cred_status(args)
elif act == "link-instagram":
cmd_cred_link_instagram(args)
elif act == "meta-audit":
cmd_cred_meta_audit(args)
else:
parser.print_help()
elif args.domain == "ssh":
act = getattr(args, "action", None)
if not act or act == "list":
cmd_ssh_list(args)
elif act == "mint":
cmd_ssh_mint(args)
elif act == "show":
cmd_ssh_show(args)
else:
p_ssh.print_help()
elif args.domain == "harvest":
act = getattr(args, "action", None)
if not act or act == "status":
cmd_harvest_status(args)
elif act == "run":
cmd_harvest_run(args)
elif act == "tail":
cmd_harvest_tail(args)
else:
parser.print_help()
elif args.domain == "followup":
act = getattr(args, "action", None)
if not act or act == "list":
cmd_followup_list(args)
elif act == "sweep":
cmd_followup_sweep(args)
elif act == "cancel":
cmd_followup_cancel(args)
else:
parser.print_help()
elif args.domain == "pipeline":
act = getattr(args, "action", None)
if not act or act == "status":
cmd_pipeline_status(args)
elif act == "run":
cmd_pipeline_run(args)
elif act == "history":
cmd_pipeline_history(args)
elif act == "stop":
cmd_pipeline_stop(args)
elif act == "prune":
cmd_pipeline_prune(args)
else:
parser.print_help()
elif args.domain == "loop":
act = getattr(args, "action", None)
if not act or act == "status":
cmd_loop_status(args)
elif act == "health":
cmd_loop_health(args)
elif act == "breaks":
cmd_loop_breaks(args)
elif act == "strat":
cmd_loop_strat(args)
elif act == "vars":
cmd_loop_vars(args)
elif act == "close":
cmd_loop_close(args)
elif act == "nudge":
cmd_loop_nudge(args)
elif act == "sweep":
cmd_loop_sweep(args)
elif act == "harvest":
cmd_loop_harvest(args)
elif act == "remediate":
cmd_loop_remediate(args)
else:
parser.print_help()
elif args.domain == "strat":
cmd_loop_strat(args)
elif args.domain == "vars":
cmd_loop_vars(args)
elif args.domain == "swarm":
act = getattr(args, "action", None)
if not act or act == "list":
cmd_swarm_list(args)
elif act == "status":
cmd_swarm_status(args)
elif act == "spawn":
cmd_swarm_spawn(args)
elif act == "prune":
cmd_swarm_prune(args)
else:
p_swarm.print_help()
elif args.domain in ("deploy", "subagent"):
if args.domain == "subagent":
args.action = "subagent"
cmd_deploy(args)
elif args.domain == "muse":
cmd = [str(BIN_DIR / "muse-cli-node"), args.node] + (args.muse_args or [])
res = subprocess.run(cmd)
sys.exit(res.returncode)
else:
parser.print_help()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink