feat(kpi): expand KPI runtime monitoring, prompt advisory envelopes, and missing field resiliency

bin/onboard_pipeline.py

@@ -355,7 +355,19 @@ def finish_onboarding_redemption(state: OnboardState) -> Dict[str, Any]:
                 state.stage = STAGE_REDEEMED
                 state.detail = f"Successfully redeemed code {state.invite_code}! 1B tokens credited to @{state.beneficiary_node} and @{state.node}."
             else:
-                state.detail = f"Redemption failed: {redemption.get('reason')} - {redemption.get('detail')}"
+                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}"
@@ -404,10 +416,71 @@ def issue_salvage_work_order(blocked_node: str = "646", to_sidechat: str = "646
     return {"ok": res.returncode == 0, "output": res.stdout.strip()}
 
 
+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")
@@ -435,7 +508,25 @@ def main() -> None:
 
     args = parser.parse_args()
 
-    if args.action == "start":
+    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))