super/box
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
c305df1f723e4c0fdd38f60ef3e4fe3d94c76d1b
box/bin/onboard_pipeline.py
T

651 lines
26 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""onboard_pipeline.py — End-to-end agent-driven onboarding & invite salvage pipeline.
Orchestrates the complete lifecycle:
1. Identifies the most urgent beneficiary agent in need of tokens (blocked first, then lowest balance).
2. Provisions infrastructure (WireGuard, dedicated netns, CDP port, chrome-box profile).
3. Launches authentication via cred_client / onboard-driver.
4. Waits for / accepts OTP code.
5. Checks age verification gate (Tailscale portal / Instagram link).
6. Automatically redeems the urgent agent's invite code on the newly onboarded node.
7. Injects operational DRIVE and marks the node active.
8. Can dispatch cryptographically signed work orders ([WO]) to prompting sidechats.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import subprocess
import sys
import time
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
REPO_ROOT = Path(__file__).resolve().parent.parent
BIN_DIR = REPO_ROOT / "bin"
sys.path.insert(0, str(BIN_DIR))
try:
import invite
from cred_client import CredClient
from settings_rpa import SettingsRPA
except ImportError:
pass
STAGE_INFRA = "infra_provisioned"
STAGE_INITIATE = "auth_initiated"
STAGE_AWAIT_OTP = "awaiting_otp"
STAGE_AUTH_ACTIVE = "auth_active"
STAGE_REDEEMED = "invite_redeemed"
STAGE_DRIVE_INJECTED = "drive_injected"
STAGE_COMPLETED = "completed"
STATE_DIR = Path(os.environ.get("NETVM_ONBOARD_STATE", "/tmp/netvm-onboard"))
@dataclass
class OnboardState:
node: str
email: str
beneficiary_node: Optional[str]
invite_code: Optional[str]
stage: str
cdp_port: Optional[int] = None
created_at: float = 0.0
updated_at: float = 0.0
detail: Optional[str] = None
redemption_result: Optional[Dict[str, Any]] = None
def save(self) -> None:
STATE_DIR.mkdir(parents=True, exist_ok=True)
path = STATE_DIR / f"{self.node}.json"
self.updated_at = time.time()
with open(path, "w") as f:
json.dump(asdict(self), f, indent=2)
@classmethod
def load(cls, node: str) -> Optional[OnboardState]:
path = STATE_DIR / f"{node}.json"
if not path.exists():
return None
try:
with open(path) as f:
data = json.load(f)
return cls(**data)
except Exception:
return None
# Canonical agent operational roles in NetVM
AGENT_ROLES: Dict[str, str] = {
"646": "Production Lead / Swarm Execution",
"opm": "Fleet Coordinator / Loop Orchestrator",
"pip": "Production Agent / Pipeline Worker",
"muse": "Core Engine / Auditor & Dev",
"dev": "Development / Test & Staging",
"def": "Defense / Security & Standby",
}
# Base role priority multiplier
ROLE_WEIGHT_MULTIPLIER: Dict[str, float] = {
"646": 1.5, # Critical production executor
"opm": 1.4, # Central coordinator & orchestrator
"pip": 1.2, # High volume production worker
"muse": 1.1, # Auditor & core engine
"dev": 0.8, # Test & staging
"def": 0.7, # Defense & standby
}
def get_agent_metrics() -> Tuple[Dict[str, int], Dict[str, int]]:
"""Count job assignments and cumulative chat activity (work done over time) per agent."""
from collections import Counter
job_counts = Counter()
jobs_dir = REPO_ROOT / "jobs"
if jobs_dir.exists():
for p in jobs_dir.glob("*.json"):
try:
d = json.loads(p.read_text(encoding="utf-8"))
a = d.get("target_agent") or d.get("agent")
if a:
job_counts[a] += 1
except Exception:
pass
msg_counts = Counter()
chat_log = REPO_ROOT / "logs" / "chat-history.jsonl"
if chat_log.exists():
try:
with open(chat_log, encoding="utf-8") as f:
for line in f:
try:
row = json.loads(line)
s = row.get("sender") or row.get("from") or row.get("agent")
if s:
msg_counts[s] += 1
except Exception:
pass
except Exception:
pass
return dict(job_counts), dict(msg_counts)
def calculate_feeding_weights(nodes: Optional[List[str]] = None) -> List[Dict[str, Any]]:
"""Compute feeding weights for all agents considering:
- Urgency: Blocked (highest), weekly used %, token depletion
- Work done over time: Messages processed / chat history volume (strongest operational weight)
- Job amount: Active and assigned jobs in registry
- Role importance: Multiplier based on operational criticality
"""
target_nodes = nodes or ["646", "opm", "pip", "muse", "dev", "def"]
job_counts, msg_counts = get_agent_metrics()
usage_map = invite.fleet_usage(target_nodes)
invites_map = invite.fleet_invite_status(target_nodes)
rows = []
for n in target_nodes:
u = usage_map.get(n, {})
inv = invites_map.get(n, {})
code = inv.get("code") or "-"
wu = u.get("weekly_used_pct") or 0
extra_left = u.get("additional_left") or "-"
extra_used = u.get("additional_used_pct") or 0
is_blocked = (wu >= 100 and "0 tokens left" in str(extra_left)) or (wu >= 100 and extra_used >= 100)
jobs = job_counts.get(n, 0)
work_done = msg_counts.get(n, 0)
role_desc = AGENT_ROLES.get(n, "Agent Worker")
role_mult = ROLE_WEIGHT_MULTIPLIER.get(n, 1.0)
# Feeding score calculation:
# Base: Blocked = 1000 pts; Weekly limit 100% = 300 pts; proportional to weekly used %
urgency_score = 1000.0 if is_blocked else (300.0 if wu >= 100 else float(wu))
# Work done weight: 1 point per 10 messages (strongest historical work indicator)
work_score = float(work_done) * 0.1
# Job weight: 2 points per assigned job
job_score = float(jobs) * 2.0
# Composite feeding weight
feeding_weight = round((urgency_score + work_score + job_score) * role_mult, 1)
status_label = "BLOCKED" if is_blocked else ("LIMIT_REACHED" if wu >= 100 else "ACTIVE")
rows.append({
"node": n,
"role": role_desc,
"code": code,
"status": status_label,
"weekly_used_pct": wu,
"additional_left": extra_left,
"jobs_count": jobs,
"work_done_msgs": work_done,
"feeding_weight": feeding_weight,
"is_blocked": is_blocked,
"invites_left": inv.get("uses_remaining", 30),
})
# Sort descending by feeding weight
rows.sort(key=lambda r: r["feeding_weight"], reverse=True)
return rows
def select_urgent_beneficiary(explicit_node: Optional[str] = None, explicit_code: Optional[str] = None) -> Tuple[Optional[str], Optional[str], Optional[str]]:
"""Determine the optimal beneficiary agent to feed with Muse tokens and return (beneficiary_node, invite_code, reason).
Priority:
1. Explicit code / node requested by caller.
2. Highest feeding weight (combining urgency, work done over time, job volume, and role criticality).
"""
if explicit_code:
return explicit_node, explicit_code.strip().upper(), "Explicit invite code provided"
if explicit_node:
try:
inv = invite.get_invite(explicit_node)
if inv.get("ok") and inv.get("code"):
return explicit_node, inv["code"], f"Explicit beneficiary agent @{explicit_node}"
except Exception as e:
return explicit_node, None, f"Failed to fetch invite code for @{explicit_node}: {e}"
# Calculate weighted rankings across fleet
rankings = calculate_feeding_weights()
for row in rankings:
if row["code"] and row["code"] != "-":
n = row["node"]
code = row["code"]
weight = row["feeding_weight"]
st = row["status"]
role = row["role"]
work = row["work_done_msgs"]
reason = f"Top feeding weight: {weight} pts (@{n} [{role}] - status:{st}, work_done:{work} msgs, jobs:{row['jobs_count']})"
return n, code, reason
# Fallback to 646 or muse if available
for fallback in ["646", "pip", "muse", "opm"]:
try:
inv = invite.get_invite(fallback)
if inv.get("ok") and inv.get("code"):
return fallback, inv["code"], f"Default fleet agent @{fallback}"
except Exception:
pass
return None, None, "No active agent invite code found"
def provision_node_infra(node: str) -> Dict[str, Any]:
"""Execute ./bin/netvm-provision-node.sh <node> to setup netns, wireguard, chrome-box."""
script = BIN_DIR / "netvm-provision-node.sh"
cmd = [str(script), node]
res = subprocess.run(cmd, capture_output=True, text=True)
if res.returncode != 0:
return {
"ok": False,
"node": node,
"error": res.stderr.strip() or res.stdout.strip() or f"Exit {res.returncode}",
}
return {"ok": True, "node": node, "output": res.stdout.strip()}
def _registry_port(node: str) -> Optional[int]:
"""CDP port for a node via netvm-registry.py, or None if unregistered."""
import importlib.util
spec = importlib.util.spec_from_file_location(
"netvm_registry", str(BIN_DIR / "netvm-registry.py"))
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod.port_for(node)
def _cdp_dry_run(node: str) -> bool:
"""True when the node netns + CDP + page chain is healthy."""
cmd = [str(BIN_DIR / "netvm-exec.sh"), node, "--", sys.executable,
str(BIN_DIR / "onboard-driver.py"),
"--node", node, "--service", "muse", "--id-type", "email",
"--step", "initiate", "--dry-run"]
res = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
return res.returncode == 0
def ensure_node_browser(node: str, timeout: float = 90.0, poll_interval: float = 5.0) -> Dict[str, Any]:
"""Launch the node headless browser inside its netns if CDP is down.
start_onboarding() must call this after infra provisioning: provision
never starts a browser, so without this step auth initiation always
dies with CDP connection refused on fresh nodes.
"""
port = _registry_port(node)
if not port:
return {"ok": False, "node": node,
"error": "unknown node %s (not in NODES.md registry)" % node}
if _cdp_dry_run(node):
return {"ok": True, "node": node, "cdp_port": port, "already": True}
STATE_DIR.mkdir(parents=True, exist_ok=True)
log_path = STATE_DIR / ("%s-chrome.log" % node)
cmd = [str(BIN_DIR / "netvm-chrome.sh"), "--headless",
"--cdp-port", str(port), node, "https://muse.ai"]
with open(log_path, "ab") as log:
subprocess.Popen(cmd, start_new_session=True,
stdout=log, stderr=subprocess.STDOUT,
stdin=subprocess.DEVNULL)
deadline = time.time() + timeout
while time.time() < deadline:
time.sleep(poll_interval)
if _cdp_dry_run(node):
return {"ok": True, "node": node, "cdp_port": port, "already": False}
return {"ok": False, "node": node, "cdp_port": port,
"error": "browser launched but CDP stayed unreachable on port %s (log: %s)" % (port, log_path)}
def start_onboarding(node: str, email: str, beneficiary_node: Optional[str] = None, invite_code: Optional[str] = None, account_name: Optional[str] = None) -> Dict[str, Any]:
"""Phase 1 & 2: Provision infra, choose beneficiary invite code, and initiate authentication."""
b_node, code, reason = select_urgent_beneficiary(beneficiary_node, invite_code)
state = OnboardState(
node=node,
email=email,
beneficiary_node=b_node,
invite_code=code,
stage="starting",
created_at=time.time(),
updated_at=time.time(),
detail=reason,
)
state.save()
# 1. Provision infra
infra_res = provision_node_infra(node)
if not infra_res.get("ok"):
state.stage = "infra_failed"
state.detail = infra_res.get("error")
state.save()
return {"ok": False, "state": asdict(state), "error": f"Infra provisioning failed: {state.detail}"}
state.stage = STAGE_INFRA
state.save()
# 1b. Ensure the headless browser is up (provision never starts one).
browser_res = ensure_node_browser(node)
if not browser_res.get("ok"):
state.stage = "browser_failed"
state.detail = browser_res.get("error")
state.save()
return {"ok": False, "state": asdict(state), "error": state.detail}
# 2. Initiate authentication
client = CredClient()
cred_res = client.initiate(node, email, service="muse", account_name=account_name)
st = cred_res.get("status")
if st == "awaiting_otp":
state.stage = STAGE_AWAIT_OTP
state.detail = f"OTP verification code sent to {email}"
state.save()
return {
"ok": True,
"status": "awaiting_otp",
"state": asdict(state),
"beneficiary_node": b_node,
"invite_code_queued": code,
"message": f"Verification code sent to {email}. Submit with: box onboard submit-otp --node {node} --otp <code>",
}
elif st == "active":
state.stage = STAGE_AUTH_ACTIVE
state.save()
# Immediately redeem queued code
return finish_onboarding_redemption(state)
else:
state.stage = "auth_error"
state.detail = cred_res.get("detail") or cred_res.get("message")
state.save()
return {"ok": False, "status": st, "state": asdict(state), "error": state.detail}
def submit_onboarding_otp(node: str, otp: str, email: Optional[str] = None) -> Dict[str, Any]:
"""Phase 3: Submit transient OTP and proceed to redemption upon success."""
state = OnboardState.load(node)
if not state:
state = OnboardState(
node=node,
email=email or "unknown",
beneficiary_node=None,
invite_code=None,
stage=STAGE_AWAIT_OTP,
created_at=time.time(),
)
client = CredClient()
res = client.submit_otp(node, otp, email=state.email or email)
st = res.get("status")
if st == "active":
state.stage = STAGE_AUTH_ACTIVE
state.detail = "Session authenticated successfully"
state.save()
return finish_onboarding_redemption(state)
else:
state.detail = res.get("detail") or res.get("message")
state.save()
return {"ok": False, "status": st, "state": asdict(state), "error": state.detail}
def finish_onboarding_redemption(state: OnboardState) -> Dict[str, Any]:
"""Phase 4 & 5: Redeem queued invite code and inject DRIVE."""
# Ensure beneficiary code is present
if not state.invite_code:
b_node, code, _ = select_urgent_beneficiary(state.beneficiary_node, None)
state.beneficiary_node = b_node
state.invite_code = code
redemption_info = None
if state.invite_code:
try:
# Redeem queued code on the fresh node
redemption = invite.redeem_invite(state.node, state.invite_code, timeout=20.0)
redemption_info = redemption
state.redemption_result = redemption
if redemption.get("ok"):
state.stage = STAGE_REDEEMED
state.detail = f"Successfully redeemed code {state.invite_code}! 1B tokens credited to @{state.beneficiary_node} and @{state.node}."
else:
reason = redemption.get("reason", "unknown")
detail = redemption.get("detail", "")
state.detail = f"Redemption failed: {reason} - {detail}"
try:
from invite_handler import send_loopback_notice
target_chat = f"{state.beneficiary_node} tasks" if state.beneficiary_node else "heartbeat-opm"
send_loopback_notice(
recipient=state.beneficiary_node or "opm",
target=target_chat,
message=f"[ONBOARD-SALVAGE-LOOPBACK] Node @{state.node} could not redeem code {state.invite_code} for @{state.beneficiary_node}: {reason}. Stage remains safe.",
)
except Exception:
pass
except Exception as e:
redemption_info = {"ok": False, "error": str(e)}
state.detail = f"Redemption exception: {e}"
# Phase 5: Inject DRIVE
try:
drive_script = BIN_DIR / "agent_md.py"
if drive_script.exists():
subprocess.run([sys.executable, str(drive_script), "inject-drive", state.node], capture_output=True, text=True, timeout=15)
state.stage = STAGE_COMPLETED
except Exception:
pass
state.save()
return {
"ok": True,
"status": "completed" if (redemption_info and redemption_info.get("ok")) else "auth_active_redemption_warning",
"state": asdict(state),
"beneficiary_node": state.beneficiary_node,
"invite_code": state.invite_code,
"redemption": redemption_info,
"message": f"Node @{state.node} is fully active! 1B tokens granted.",
}
def issue_salvage_work_order(blocked_node: str = "646", to_sidechat: str = "646 tasks") -> Dict[str, Any]:
"""Issue a cryptographically signed Work Order prompting fleet operators or agents to initiate onboarding."""
inv = invite.get_invite(blocked_node)
code = inv.get("code") or "UNKNOWN"
title = f"Salvage Blocked Agent @{blocked_node}"
body = (
f"Agent @{blocked_node} is out of tokens (code: {code}). "
f"Initiate client onboarding to grant 1 Billion tokens: box onboard <new_node> --email <client_email> --for {blocked_node}"
)
cmd = [
"python3", str(BIN_DIR / "super-cli.py"), "dm", "wo",
"--to", "opm",
"--target", to_sidechat,
"--title", title,
"--priority", "urgent",
"--allow-main-chat",
body,
]
res = subprocess.run(cmd, capture_output=True, text=True)
out = res.stdout.strip()
result: Dict[str, Any] = {"ok": res.returncode == 0, "output": out}
if not result["ok"]:
err = res.stderr.strip()
result["error"] = err or out or "dm wo exited %d" % res.returncode
return result
def get_all_connects(fast: bool = True) -> List[Dict[str, Any]]:
"""Return consolidated inventory of all fleet and onboarded connects."""
connects = []
seen = set()
# Load recorded pipeline states
if STATE_DIR.exists():
for p in STATE_DIR.glob("*.json"):
try:
d = json.loads(p.read_text(encoding="utf-8"))
n = d.get("node")
if n:
seen.add(n)
connects.append({
"node": n,
"type": "onboard_pipeline",
"email": d.get("email"),
"stage": d.get("stage"),
"beneficiary": d.get("beneficiary_node"),
"invite_code": d.get("invite_code") or "-",
"cdp_port": d.get("cdp_port"),
"detail": d.get("detail"),
"updated_at": d.get("updated_at"),
})
except Exception:
pass
# Fleet nodes
w_map = {}
if not fast:
try:
weights = calculate_feeding_weights()
w_map = {r["node"]: r for r in weights}
except Exception:
w_map = {}
ports = {"muse": 9222, "pip": 9322, "646": 9430, "opm": 9440, "def": 9450, "dev": 9455}
for agent in ("muse", "pip", "646", "opm", "dev", "def"):
if agent not in seen:
w_info = w_map.get(agent, {})
connects.append({
"node": agent,
"type": "fleet_agent",
"email": f"{agent}@muse-dev.online",
"stage": "active_fleet",
"beneficiary": None,
"invite_code": w_info.get("code") or "-",
"cdp_port": ports.get(agent),
"role": AGENT_ROLES.get(agent, ""),
"status": w_info.get("status", "ACTIVE"),
"feeding_weight": w_info.get("feeding_weight", 1.0),
"updated_at": time.time(),
})
return connects
def main() -> None:
parser = argparse.ArgumentParser(description="End-to-end agent-driven onboarding & invite salvage pipeline")
sub = parser.add_subparsers(dest="action")
p_conn = sub.add_parser("connects", help="Inventory of all active fleet nodes & client onboard connects")
p_conn.add_argument("--json", action="store_true", help="Emit JSON output")
p_start = sub.add_parser("start", help="Start full onboarding pipeline for a client node")
p_start.add_argument("node", help="New node label (e.g. dev2, client1)")
p_start.add_argument("--email", required=True, help="Client login email")
p_start.add_argument("--for", dest="for_agent", help="Beneficiary agent to unblock (defaults to most urgent)")
p_start.add_argument("--code", help="Explicit 6-character invite code to redeem")
p_start.add_argument("--account-name", help="Display name hint")
p_start.add_argument("--json", action="store_true", help="Emit JSON output")
p_otp = sub.add_parser("submit-otp", help="Submit transient OTP verification code")
p_otp.add_argument("node", help="Node label")
p_otp.add_argument("otp", help="6-digit verification code")
p_otp.add_argument("--email", help="Client email (optional)")
p_otp.add_argument("--json", action="store_true", help="Emit JSON output")
p_status = sub.add_parser("status", help="Check onboarding pipeline state for a node")
p_status.add_argument("node", help="Node label")
p_status.add_argument("--json", action="store_true", help="Emit JSON output")
p_wo = sub.add_parser("salvage-wo", help="Dispatch salvage work order to sidechat")
p_wo.add_argument("node", nargs="?", default="646", help="Blocked agent node (default: 646)")
p_wo.add_argument("--target", default="646 tasks", help="Target sidechat")
p_feed = sub.add_parser("feed-matrix", help="Display all agents ranked by feeding weight, job volume, role, and work done over time")
p_feed.add_argument("--json", action="store_true", help="Emit JSON output")
args = parser.parse_args()
if args.action == "connects":
conn_list = get_all_connects()
if args.json:
print(json.dumps(conn_list, indent=2))
else:
print("\n=== ACTIVE FLEET & CLIENT ONBOARD CONNECTS ===")
print(f" {'NODE':6} {'TYPE':16} {'STAGE / STATUS':16} {'CDP':6} {'INVITE':8} {'ROLE / DETAIL'}")
print(f" {'─'*6} {'─'*16} {'─'*16} {'─'*6} {'─'*8} {'─'*32}")
for c in conn_list:
node = c.get("node", "")
t_str = c.get("type", "")
st_str = c.get("stage", c.get("status", ""))
cdp = str(c.get("cdp_port") or "-")
code = c.get("invite_code") or "-"
role = c.get("role") or c.get("detail") or c.get("email") or ""
print(f" {node:<6} {t_str:<16} {st_str:<16} {cdp:<6} {code:<8} {role}")
print()
elif args.action == "start":
res = start_onboarding(args.node, args.email, beneficiary_node=getattr(args, "for_agent", None), invite_code=args.code, account_name=args.account_name)
if args.json:
print(json.dumps(res, indent=2))
elif res.get("ok"):
print(f"\n[ONBOARDING STARTED] Node: {args.node}")
print(f" Beneficiary: @{res.get('beneficiary_node')} (Invite Code: {res.get('invite_code_queued')})")
print(f" Message: {res.get('message')}\n")
else:
print(f"\n[ERROR] {res.get('error')}\n", file=sys.stderr)
sys.exit(1)
elif args.action == "submit-otp":
res = submit_onboarding_otp(args.node, args.otp, email=args.email)
if args.json:
print(json.dumps(res, indent=2))
elif res.get("ok"):
print(f"\n[ONBOARDING COMPLETE] {res.get('message')}")
print(f" Beneficiary Credited: @{res.get('beneficiary_node')} (+1,000,000,000 tokens)\n")
else:
print(f"\n[ERROR] {res.get('error')}\n", file=sys.stderr)
sys.exit(1)
elif args.action == "status":
state = OnboardState.load(args.node)
if not state:
print(f"No onboarding pipeline record found for {args.node}", file=sys.stderr)
sys.exit(1)
if args.json:
print(json.dumps(asdict(state), indent=2))
else:
print(f"\n=== ONBOARDING STATUS: @{state.node} ===")
print(f" Email: {state.email}")
print(f" Stage: {state.stage}")
print(f" Beneficiary: @{state.beneficiary_node} (Code: {state.invite_code})")
if state.detail:
print(f" Detail: {state.detail}")
print()
elif args.action == "salvage-wo":
res = issue_salvage_work_order(args.node, to_sidechat=args.target)
print(f"Salvage work order dispatched for @{args.node}: ok={res.get('ok')}")
elif args.action == "feed-matrix":
matrix = calculate_feeding_weights()
if args.json:
print(json.dumps(matrix, indent=2))
else:
print("\n=== FLEET FEEDING & TOKEN ALLOCATION MATRIX ===")
print(" (Weighted by: Urgency + Work Done Over Time + Job Amount + Role Criticality)\n")
print(f" {'RANK':4} {'AGENT':6} {'ROLE':38} {'FEED SCORE':11} {'STATUS':10} {'JOBS':6} {'WORK DONE':10} {'CODE':8}")
print(f" {'─'*4} {'─'*6} {'─'*38} {'─'*11} {'─'*10} {'─'*6} {'─'*10} {'─'*8}")
for idx, r in enumerate(matrix, 1):
print(f" #{idx:<3} {r['node']:<6} {r['role']:<38} {r['feeding_weight']:<11} {r['status']:<10} {r['jobs_count']:<6} {str(r['work_done_msgs']) + ' msgs':<10} {r['code']:<8}")
print()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink