diff --git a/bin/agent-cognitive-probe.py b/bin/agent-cognitive-probe.py new file mode 100755 index 0000000..c9ce637 --- /dev/null +++ b/bin/agent-cognitive-probe.py @@ -0,0 +1,484 @@ +#!/usr/bin/env python3 +"""agent-cognitive-probe.py β€” Real-time Cognitive Sensing and Agent Menu Navigation. + +Directly probes Cloud Muse / Hatch browser runtime via CDP: +1. Passive Cognitive Sensing (zero-click): + - Token streaming / generation state (stop button presence) + - Typing / thinking indicators + - Status text displayed under/beside avatar (Connected, Thinking, Working) + - Active context (Main chat vs Side chats with thread titles & snippets) + - Input wait / parked approval detection +2. Active Profile Menu Navigation: + - Status panel sliding surface navigation + - Tabs: Activity (tasks & processes), Upcoming (timers & recurring cron loops), + Approvals, and Identity. +3. Cognitive Lock Gate: + - Protects single-threaded thought process from interruptions. +""" + +import argparse +import json +import os +import subprocess +import sys +import time +import urllib.request + +# Node to pinned CDP port mapping +NODE_CDP_PORTS = { + "muse": 9410, + "pip": 9420, + "646": 9430, + "opm": 9440, + "def": 9450, + "dev": 9455, + "muse-main": 9410, +} + +REMOTE_HOST = "100.123.153.75" # bl control node + + +def is_running_on_bl(): + """Detect if we are running locally on bl or on tp/remote.""" + try: + import socket + hn = socket.gethostname().lower() + if "bl" in hn: + return True + except Exception: + pass + # Check if network namespaces exist locally + return os.path.exists("/var/run/netns/warp-muse") or os.path.exists("/run/netns/warp-muse") + + +def run_cdp_eval_inside_netns(node, js_code, timeout=8): + """Executes a JS snippet against the node's browser via CDP inside its netns.""" + port = NODE_CDP_PORTS.get(node) + if not port: + return {"error": f"Unknown node '{node}'"} + + # Python runner script to execute inside the target environment + runner_code = f''' +import json, sys, urllib.request +try: + import websocket +except ImportError: + print(json.dumps({{"error": "websocket package missing"}})) + sys.exit(1) + +try: + with urllib.request.urlopen("http://127.0.0.1:{port}/json/list", timeout=3) as r: + targets = json.load(r) + pages = [t for t in targets if t.get("type") == "page"] + if not pages: + print(json.dumps({{"error": "No active page target"}})) + sys.exit(0) + + ws_url = pages[0]["webSocketDebuggerUrl"] + ws = websocket.create_connection(ws_url, timeout={timeout}) + + ws.send(json.dumps({{ + "id": 1, + "method": "Runtime.evaluate", + "params": {{ + "expression": {json.dumps(js_code)}, + "returnByValue": True, + "awaitPromise": True + }} + }})) + + res = None + for _ in range(30): + msg = json.loads(ws.recv()) + if msg.get("id") == 1: + res = msg.get("result", {{}}).get("result", {{}}).get("value") + break + + print(json.dumps({{"ok": True, "value": res}})) +except Exception as e: + print(json.dumps({{"error": str(e)}})) +''' + + if is_running_on_bl(): + cmd = ["sudo", "ip", "netns", "exec", f"warp-{node}", "python3", "-c", runner_code] + else: + # Wrap via ssh to bl + # Use python3 on bl directly executing inside netns + remote_cmd = f"sudo ip netns exec warp-{node} python3 -c {subprocess.list2cmdline([runner_code])}" + cmd = ["ssh", "-q", f"super@{REMOTE_HOST}", remote_cmd] + + try: + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 5) + if proc.returncode != 0 and not proc.stdout: + return {"error": proc.stderr.strip() or f"Process exited with {proc.returncode}"} + + # Parse output line that contains valid json + for line in proc.stdout.strip().splitlines(): + line = line.strip() + if line.startswith("{") and line.endswith("}"): + try: + data = json.loads(line) + if "ok" in data: + return data["value"] + if "error" in data: + return {"error": data["error"]} + except Exception: + continue + return {"error": proc.stdout.strip() or proc.stderr.strip()} + except subprocess.TimeoutExpired: + return {"error": "CDP probe timed out"} + except Exception as e: + return {"error": str(e)} + + +JS_PASSIVE_COGNITIVE = """(() => { + // 1. Generation & Thinking signals + const stopBtn = document.querySelector('[data-testid="hatch-composer-stop-button"]'); + const isGenerating = !!stopBtn; + + const typingEl = document.querySelector('[data-testid="hatch-chat-typing-indicator"]'); + const isTyping = !!typingEl && typingEl.textContent.trim().length > 0; + + // 2. Avatar / Status text + const statusTextEl = document.querySelector('.group\\\\/status-avatar span, [class*="status-avatar"] span, span[class*="text-body-status"]'); + const avatarStatus = statusTextEl ? (statusTextEl.innerText || '').trim() : ''; + + // 3. Thread context & active URL + const url = window.location.href; + const isMainChat = url === 'https://muse.ai/' || url.endsWith('/thread/new'); + const pageTitle = document.title; + + // 4. Side chats overview + const sideRows = Array.from(document.querySelectorAll('[data-testid="hatch-thread-row"]')).map(r => { + const text = (r.innerText || '').trim().replace(/\\\\n+/g, ' β€” '); + return text; + }).slice(0, 5); + + // 5. Input wait / Parked prompt cards + // Detect buttons asking for approval / input in the message flow + const actionBtns = Array.from(document.querySelectorAll('div[data-message-id] button')).map(b => (b.innerText || '').trim()).filter(t => /approve|confirm|proceed|resume|start|allow/i.test(t)); + const isInputWait = actionBtns.length > 0 || /asking for input|pending approval/i.test(document.body.innerText.slice(-600)); + + // 6. Status panel state + const panel = document.querySelector('[data-testid="hatch-status-panel-sliding-surface"]'); + const panelOpen = !!panel && panel.getBoundingClientRect().width > 0; + + return { + url: url, + title: pageTitle, + is_main_chat: isMainChat, + is_generating: isGenerating, + is_typing: isTyping, + avatar_status: avatarStatus, + is_input_wait: isInputWait, + pending_actions: actionBtns, + panel_open: panelOpen, + side_chats: sideRows + }; +})()""" + + +def get_passive_cognitive_state(node): + """Gathers passive cognitive signals without altering UI state.""" + if not is_running_on_bl(): + try: + cmd = ["ssh", "-q", "-o", "ConnectTimeout=5", f"super@{REMOTE_HOST}", + f"python3 /home/super/Projects/NetVM/bin/agent-cognitive-probe.py status {node} --json"] + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=10) + if proc.returncode == 0 and proc.stdout.strip(): + data = json.loads(proc.stdout.strip()) + if isinstance(data, list) and len(data) > 0: + return data[0] + elif isinstance(data, dict): + return data + except Exception as e: + return { + "node": node, + "status": "DARK", + "error": f"Remote delegation failed: {e}", + "cognitive_lock": False, + "lock_reason": None, + } + + raw = run_cdp_eval_inside_netns(node, JS_PASSIVE_COGNITIVE) + if not isinstance(raw, dict) or "error" in raw: + return { + "node": node, + "status": "DARK", + "error": raw.get("error", "Unknown error") if isinstance(raw, dict) else str(raw), + "cognitive_lock": False, + "lock_reason": None, + } + + is_generating = raw.get("is_generating", False) + is_typing = raw.get("is_typing", False) + is_input_wait = raw.get("is_input_wait", False) + avatar_status = raw.get("avatar_status", "") + is_main = raw.get("is_main_chat", True) + + # Determine synthesized cognitive state + if is_generating or is_typing or "thinking" in avatar_status.lower(): + state = "THINKING" + locked = True + reason = "Agent is actively generating tokens / thinking (stop button active)" + elif is_input_wait: + state = "INPUT_WAIT" + locked = True + reason = "Agent is waiting for operator or system input on a parked prompt" + elif not is_main: + state = "BUSY_SIDECHAT" + locked = True + reason = f"Agent is active in sidechat: {raw.get('title', 'Side Chat')}" + elif "working" in avatar_status.lower() or "making" in avatar_status.lower(): + state = "WORKING" + locked = True + reason = f"Avatar status indicates work in progress: '{avatar_status}'" + else: + state = "IDLE" + locked = False + reason = None + + return { + "node": node, + "status": state, + "cognitive_lock": locked, + "lock_reason": reason, + "details": raw, + } + + +JS_NAVIGATE_MENU_TEMPLATE = """(async () => { + // 1. Ensure panel is open + let panel = document.querySelector('[data-testid="hatch-status-panel-sliding-surface"]'); + if (!panel) { + // Try clicking avatar or trigger + const avatarBtn = document.querySelector('[role="img"][aria-label*="avatar"], button[aria-label*="avatar"], .group\\\\/status-avatar'); + if (avatarBtn) avatarBtn.click(); + await new Promise(r => setTimeout(r, 350)); + } + + // 2. Click requested tab if specified + const targetTab = "%(tab)s"; + if (targetTab && targetTab !== "all") { + const btn = document.querySelector('button[aria-label="' + targetTab + '"]'); + if (btn) { + btn.click(); + await new Promise(r => setTimeout(r, 400)); + } + } + + panel = document.querySelector('[data-testid="hatch-status-panel-sliding-surface"]'); + if (!panel) return {error: "Status panel not rendered"}; + + // Read full text and structural items + const rawText = panel.innerText || ''; + const lines = rawText.split('\\n').map(s => s.trim()).filter(Boolean); + + // Extract activity / timer items + const items = []; + const buttons = Array.from(panel.querySelectorAll('button')).filter(b => !b.getAttribute('aria-label') && (b.innerText || '').length > 0); + buttons.forEach(b => { + const text = (b.innerText || '').trim(); + const parts = text.split('\\n').map(s => s.trim()).filter(Boolean); + if (parts.length >= 2) { + items.push({ + title: parts[0], + detail: parts[1], + time: parts.length > 2 ? parts[2] : null + }); + } + }); + + return { + raw_text: rawText, + lines: lines, + items: items + }; +})()""" + + +def get_agent_menu(node, tab="all"): + """Navigates and extracts data from the agent profile / status panel.""" + if not is_running_on_bl(): + try: + cmd = ["ssh", "-q", "-o", "ConnectTimeout=5", f"super@{REMOTE_HOST}", + f"python3 /home/super/Projects/NetVM/bin/agent-cognitive-probe.py menu {node} {tab} --json"] + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=15) + if proc.returncode == 0 and proc.stdout.strip(): + return json.loads(proc.stdout.strip()) + except Exception as e: + return { + "node": node, + "tab": tab, + "data": {"error": f"Remote delegation failed: {e}"} + } + + tab_map = { + "activity": "Activity", + "upcoming": "Upcoming", + "approvals": "Approvals", + "identity": "Identity", + "all": "all", + } + target_tab = tab_map.get(tab.lower(), "Activity") + + # If all is requested, gather activity, upcoming, and approvals + if target_tab == "all": + result = {} + for sub_tab in ["Activity", "Upcoming", "Approvals"]: + js = JS_NAVIGATE_MENU_TEMPLATE % {"tab": sub_tab} + tab_res = run_cdp_eval_inside_netns(node, js) + result[sub_tab.lower()] = tab_res + return { + "node": node, + "menu": result + } + + js = JS_NAVIGATE_MENU_TEMPLATE % {"tab": target_tab} + res = run_cdp_eval_inside_netns(node, js) + return { + "node": node, + "tab": target_tab, + "data": res + } + + +def format_cognitive_badge(status): + badges = { + "IDLE": "🟒 IDLE", + "THINKING": "🧠 THINKING", + "INPUT_WAIT": "⏸️ INPUT_WAIT", + "BUSY_SIDECHAT": "πŸ’¬ SIDECHAT", + "WORKING": "βš™οΈ WORKING", + "DARK": "⚫ DARK", + } + return badges.get(status, f"❓ {status}") + + +def cmd_status(args): + nodes = getattr(args, "agents", None) or getattr(args, "nodes", None) or ["muse", "pip", "646", "opm", "dev", "def"] + results = [] + for n in nodes: + results.append(get_passive_cognitive_state(n)) + + if getattr(args, "json", False): + print(json.dumps(results, indent=2)) + return + + print("\n=== AGENT COGNITIVE SENSOR (LIVE DOM PROBE) ===") + print(f"{'AGENT':<10} {'COGNITIVE STATE':<18} {'LOCK':<8} {'UNDER-AVATAR':<15} {'DETAILS / CURRENT THOUGHT':<40}") + print("-" * 95) + for r in results: + node = r["node"] + status = r["status"] + badge = format_cognitive_badge(status) + locked = "LOCKED" if r.get("cognitive_lock") else "OPEN" + det = r.get("details", {}) + avatar_status = det.get("avatar_status", "-") + reason = r.get("lock_reason") or det.get("title", "Settled") + if status == "DARK": + reason = r.get("error", "CDP unreachable") + avatar_status = "DARK" + print(f"{node:<10} {badge:<18} {locked:<8} {avatar_status:<15} {reason[:40]:<40}") + print() + + +def cmd_menu(args): + node = getattr(args, "agent", None) or getattr(args, "node", None) + tab = getattr(args, "tab", "activity") or "activity" + data = get_agent_menu(node, tab) + + if getattr(args, "json", False): + print(json.dumps(data, indent=2)) + return + + print(f"\n=== AGENT MENU: {node.upper()} (TAB: {tab.upper()}) ===") + if tab.lower() == "all": + menu = data.get("menu", {}) + for tname, tdata in menu.items(): + print(f"\n--- {tname.upper()} ---") + if isinstance(tdata, dict) and "error" in tdata: + print(f" Error: {tdata['error']}") + elif isinstance(tdata, dict): + items = tdata.get("items", []) + if items: + for it in items: + t_str = f" [{it['time']}]" if it.get("time") else "" + print(f" β€’ {it['title']}: {it['detail']}{t_str}") + else: + raw = tdata.get("raw_text", "") + for line in raw.split("\n"): + if line.strip(): + print(f" {line.strip()}") + print() + return + + tdata = data.get("data", {}) + if isinstance(tdata, dict) and "error" in tdata: + print(f" Error: {tdata['error']}") + elif isinstance(tdata, dict): + items = tdata.get("items", []) + if items: + for it in items: + t_str = f" [{it['time']}]" if it.get("time") else "" + print(f" β€’ {it['title']}: {it['detail']}{t_str}") + else: + raw = tdata.get("raw_text", "") + for line in raw.split("\n"): + if line.strip(): + print(f" {line.strip()}") + print() + + +def cmd_lock(args): + node = getattr(args, "agent", None) or getattr(args, "node", None) + state = get_passive_cognitive_state(node) + if args.json: + print(json.dumps(state, indent=2)) + else: + if state.get("cognitive_lock"): + print(f"πŸ”΄ COGNITIVE LOCK ENGAGED on '{node}' ({state['status']}): {state.get('lock_reason')}") + else: + print(f"🟒 COGNITIVELY IDLE: Agent '{node}' is free to receive new work without interruption.") + + if state.get("cognitive_lock") and not getattr(args, "force", False): + sys.exit(1) + sys.exit(0) + + +def main(): + parser = argparse.ArgumentParser(description="Real-time Cognitive Sensing and Agent Menu Navigation") + sub = parser.add_subparsers(dest="command") + + p_status = sub.add_parser("status", help="Show cognitive state for all or selected agents") + p_status.add_argument("nodes", nargs="*", help="Optional agent names") + p_status.add_argument("--json", action="store_true", help="Output JSON") + + p_menu = sub.add_parser("menu", help="Navigate agent profile menu (tasks, timers, approvals, identity)") + p_menu.add_argument("node", help="Agent name (muse, pip, 646, opm, dev, def)") + p_menu.add_argument("tab", nargs="?", default="activity", choices=["activity", "upcoming", "approvals", "identity", "all"], help="Menu tab to view") + p_menu.add_argument("--json", action="store_true", help="Output JSON") + + p_lock = sub.add_parser("lock", help="Check cognitive lock before dispatching work") + p_lock.add_argument("node", help="Agent name") + p_lock.add_argument("--force", action="store_true", help="Bypass lock check") + p_lock.add_argument("--json", action="store_true", help="Output JSON") + + args = parser.parse_args() + if not args.command: + # Default to status + args.nodes = [] + args.json = False + cmd_status(args) + return + + if args.command == "status": + cmd_status(args) + elif args.command == "menu": + cmd_menu(args) + elif args.command == "lock": + cmd_lock(args) + + +if __name__ == "__main__": + main() diff --git a/bin/agent_cognitive_probe.py b/bin/agent_cognitive_probe.py new file mode 120000 index 0000000..30a6aa9 --- /dev/null +++ b/bin/agent_cognitive_probe.py @@ -0,0 +1 @@ +agent-cognitive-probe.py \ No newline at end of file diff --git a/bin/box-work.py b/bin/box-work.py index dc36de6..dbdd81f 100755 --- a/bin/box-work.py +++ b/bin/box-work.py @@ -26,6 +26,11 @@ import hashlib from datetime import datetime, timezone from pathlib import Path +try: + import agent_cognitive_probe as acp +except ImportError: + acp = None + # Color helpers USE_COLOR = sys.stdout.isatty() or os.environ.get("CLICOLOR_FORCE") == "1" @@ -495,7 +500,7 @@ def cmd_status(args): uname = assignee.get("username") agent_active_issues[uname] = iss - headers = f"{'AGENT':<12} {'ROLE':<13} {'PORT':<6} {'TUNNEL':<8} {'SIGNAL':<10} {'ACTIVE WORK / ASSIGNMENT':<38} {'LAST CHAT'}" + headers = f"{'AGENT':<10} {'ROLE':<12} {'PORT':<6} {'TUNNEL':<7} {'COGNITIVE':<15} {'ACTIVE WORK / ASSIGNMENT':<38} {'LAST CHAT'}" print(c_dim(headers)) print(c_dim("-" * len(headers))) @@ -512,6 +517,15 @@ def cmd_status(args): active_task = claimed_tasks.get(name) active_issue = agent_active_issues.get(name) + # Live cognitive probe + cog_badge = "-" + if acp and name != "muse-main": + cog = acp.get_passive_cognitive_state(name) + cog_status = cog.get("status", "DARK") + cog_badge = acp.format_cognitive_badge(cog_status) + elif name == "muse-main": + cog_badge = c_dim("HOST") + if active_issue: num = active_issue.get("number") title = active_issue.get("title", "")[:32] @@ -542,7 +556,7 @@ def cmd_status(args): else: chat_str = c_dim("-") - print(f"{c_bold(name):<21} {role:<13} {port:<6} {tunnel_str:<17} {signal:<19} {work_desc:<38} {chat_str}") + print(f"{c_bold(name):<19} {role:<12} {port:<6} {tunnel_str:<16} {cog_badge:<24} {work_desc:<38} {chat_str}") print() @@ -644,7 +658,17 @@ def cmd_start(args): agent = args.agent body = args.goal or f"Work task for {agent}: {title}" - # 0. Pre-flight health gate: Hatch, Restore, Git Config with Auto-Heal + # 0a. Real-time Cognitive Sensing & Single-Task Lock Gate + if acp and not getattr(args, "force", False): + cog = acp.get_passive_cognitive_state(agent) + if cog.get("cognitive_lock"): + print(c_red(f"\n[COGNITIVE LOCK BLOCKED] Agent '{agent}' cannot receive new work right now ({cog['status']}):")) + print(f" β€’ Reason: {cog.get('lock_reason')}") + print(c_yellow("\nCloud agents have single-threaded thought processes and get interrupted by concurrent tasks.")) + print(c_dim(f"To inspect active thoughts/menu: box work menu {agent}\nTo bypass lock: box work start '{title}' --to {agent} --force\n")) + sys.exit(1) + + # 0b. Pre-flight health gate: Hatch, Restore, Git Config with Auto-Heal preflight = check_agent_preflight(agent) if not preflight["ready"] and not getattr(args, "force", False): if getattr(args, "no_heal", False): @@ -734,7 +758,17 @@ def cmd_assign(args): issue_num = args.issue agent = args.agent - # 0. Pre-flight health gate: Hatch, Restore, Git Config with Auto-Heal + # 0a. Real-time Cognitive Sensing & Single-Task Lock Gate + if acp and not getattr(args, "force", False): + cog = acp.get_passive_cognitive_state(agent) + if cog.get("cognitive_lock"): + print(c_red(f"\n[COGNITIVE LOCK BLOCKED] Agent '{agent}' cannot receive new work right now ({cog['status']}):")) + print(f" β€’ Reason: {cog.get('lock_reason')}") + print(c_yellow("\nCloud agents have single-threaded thought processes and get interrupted by concurrent tasks.")) + print(c_dim(f"To inspect active thoughts/menu: box work menu {agent}\nTo bypass lock: box work assign {issue_num} --to {agent} --force\n")) + sys.exit(1) + + # 0b. Pre-flight health gate: Hatch, Restore, Git Config with Auto-Heal preflight = check_agent_preflight(agent) if not preflight["ready"] and not getattr(args, "force", False): if getattr(args, "no_heal", False): @@ -817,9 +851,23 @@ def cmd_chats(args): print(f"[{c_cyan(ag)} : {c_dim(tname)}] {c_dim(ts)} {c_bold(author)}:\n{text}\n" + c_dim("-" * 60)) print() +def cmd_menu(args): + if not acp: + print(c_red("Error: agent_cognitive_probe module not found.")) + sys.exit(1) + acp.cmd_menu(args) + +def cmd_cognitive(args): + if not acp: + print(c_red("Error: agent_cognitive_probe module not found.")) + sys.exit(1) + acp.cmd_status(args) + WORK_COMMAND_EXAMPLES = { "box work": [ "box work # View fleet workspace dashboard & signals", + "box work cognitive [agent...] # Live zero-click cognitive sensor probe across fleet", + "box work menu [tab] # Inspect agent profile menu (tasks, timers, approvals)", "box work check [agent] # Audit pre-flight health gates", "box work heal # Automated remediation & chat nudge", "box work start \"\" --to <agent> # Start & dispatch new build ticket", @@ -827,6 +875,15 @@ WORK_COMMAND_EXAMPLES = { "box work merge <pr#> # Verify tests and merge PR to master", "box work chats --agent <name> # View live multi-agent chat feed", ], + "box work menu": [ + "box work menu muse upcoming # Inspect timers & recurring cron loops", + "box work menu 646 activity # Inspect recent tasks & active processes", + "box work menu pip all # Dump all tabs (activity, upcoming, approvals)", + ], + "box work cognitive": [ + "box work cognitive # Live cognitive sensor probe for all agents", + "box work cognitive pip # Check if pip is generating / thinking", + ], "box work start": [ "box work start \"Fix SSH perms\" --to 646", "box work start \"Build integration tests\" --to pip --goal \"Run pytest on endpoints\"", @@ -960,6 +1017,15 @@ def main(): p_chats.add_argument("--agent", help="Filter by agent name") p_chats.add_argument("--limit", type=int, default=10, help="Number of messages to show") + p_menu = sub.add_parser("menu", help="Navigate agent profile menu (tasks, timers, approvals, identity)") + p_menu.add_argument("agent", help="Agent username (muse, pip, 646, opm, dev, def)") + p_menu.add_argument("tab", nargs="?", default="activity", choices=["activity", "upcoming", "approvals", "identity", "all"], help="Menu tab to view") + p_menu.add_argument("--json", action="store_true", help="Output JSON") + + p_cog = sub.add_parser("cognitive", help="Probe real-time cognitive sensor (thinking, generating, sidechats)") + p_cog.add_argument("agents", nargs="*", help="Optional agent usernames") + p_cog.add_argument("--json", action="store_true", help="Output JSON") + args = parser.parse_args() action = args.work_action @@ -977,6 +1043,10 @@ def main(): cmd_merge(args) elif action == "chats": cmd_chats(args) + elif action == "menu": + cmd_menu(args) + elif action == "cognitive": + cmd_cognitive(args) else: parser.print_help() diff --git a/bin/super-cli.py b/bin/super-cli.py index 7882eb7..720b232 100755 --- a/bin/super-cli.py +++ b/bin/super-cli.py @@ -1489,6 +1489,10 @@ def cmd_work(args): box_work.cmd_heal(args) elif action == "chats": box_work.cmd_chats(args) + elif action == "menu": + box_work.cmd_menu(args) + elif action == "cognitive": + box_work.cmd_cognitive(args) else: box_work.cmd_status(args) @@ -7004,6 +7008,11 @@ def build_parser(): p_w_chats = work_sub.add_parser("chats", parents=[common], help="View recent live chat activity") p_w_chats.add_argument("--agent", help="Filter by agent name") p_w_chats.add_argument("--limit", type=int, default=10, help="Number of messages to show") + p_w_menu = work_sub.add_parser("menu", parents=[common], help="Navigate agent profile menu (tasks, timers, approvals, identity)") + p_w_menu.add_argument("agent", help="Agent username (muse, pip, 646, opm, dev, def)") + p_w_menu.add_argument("tab", nargs="?", default="activity", choices=["activity", "upcoming", "approvals", "identity", "all"], help="Menu tab to view") + p_w_cog = work_sub.add_parser("cognitive", parents=[common], help="Probe real-time cognitive sensor (thinking, generating, sidechats)") + p_w_cog.add_argument("agents", nargs="*", help="Optional agent usernames") p_tasks = subparsers.add_parser("tasks", parents=[common], help="Agent work queue: pending/claimed/done files (distinct from scheduled jobs)") p_tasks.add_argument("--dir", default=None, help="Task queue dir (default: fleet/tasks)") diff --git a/tests/test_box_cognitive.py b/tests/test_box_cognitive.py new file mode 100644 index 0000000..314ef93 --- /dev/null +++ b/tests/test_box_cognitive.py @@ -0,0 +1,130 @@ +"""test_box_cognitive.py β€” Unit tests for agent cognitive sensing and menu navigation.""" + +import unittest +from unittest.mock import patch, MagicMock +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "bin")) +import agent_cognitive_probe as acp +import box_work + + +class TestAgentCognitiveProbe(unittest.TestCase): + + def test_badge_formatting(self): + self.assertIn("IDLE", acp.format_cognitive_badge("IDLE")) + self.assertIn("THINKING", acp.format_cognitive_badge("THINKING")) + self.assertIn("INPUT_WAIT", acp.format_cognitive_badge("INPUT_WAIT")) + self.assertIn("SIDECHAT", acp.format_cognitive_badge("BUSY_SIDECHAT")) + self.assertIn("DARK", acp.format_cognitive_badge("DARK")) + + @patch("agent_cognitive_probe.is_running_on_bl", return_value=True) + @patch("agent_cognitive_probe.run_cdp_eval_inside_netns") + def test_passive_thinking_state(self, mock_cdp, mock_bl): + # Simulate active token generation (stop button present) + mock_cdp.return_value = { + "is_generating": True, + "is_typing": False, + "avatar_status": "Connected", + "is_main_chat": True, + "title": "Chat β€” test", + "is_input_wait": False + } + res = acp.get_passive_cognitive_state("pip") + self.assertEqual(res["status"], "THINKING") + self.assertTrue(res["cognitive_lock"]) + self.assertIn("stop button active", res["lock_reason"]) + + @patch("agent_cognitive_probe.is_running_on_bl", return_value=True) + @patch("agent_cognitive_probe.run_cdp_eval_inside_netns") + def test_passive_input_wait_state(self, mock_cdp, mock_bl): + # Simulate parked approval card + mock_cdp.return_value = { + "is_generating": False, + "is_typing": False, + "avatar_status": "Connected", + "is_main_chat": True, + "title": "Chat β€” test", + "is_input_wait": True + } + res = acp.get_passive_cognitive_state("646") + self.assertEqual(res["status"], "INPUT_WAIT") + self.assertTrue(res["cognitive_lock"]) + + @patch("agent_cognitive_probe.is_running_on_bl", return_value=True) + @patch("agent_cognitive_probe.run_cdp_eval_inside_netns") + def test_passive_idle_state(self, mock_cdp, mock_bl): + # Simulate fully settled idle state + mock_cdp.return_value = { + "is_generating": False, + "is_typing": False, + "avatar_status": "Connected", + "is_main_chat": True, + "title": "Chat β€” test", + "is_input_wait": False + } + res = acp.get_passive_cognitive_state("dev") + self.assertEqual(res["status"], "IDLE") + self.assertFalse(res["cognitive_lock"]) + self.assertIsNone(res["lock_reason"]) + + @patch("agent_cognitive_probe.is_running_on_bl", return_value=True) + @patch("agent_cognitive_probe.run_cdp_eval_inside_netns") + def test_menu_upcoming_timers(self, mock_cdp, mock_bl): + # Simulate Upcoming tab content + mock_cdp.return_value = { + "raw_text": "Fleet sync loop\nEvery 15 minutes\nHeartbeat\nEvery 30 minutes", + "lines": ["Fleet sync loop", "Every 15 minutes", "Heartbeat", "Every 30 minutes"], + "items": [ + {"title": "Fleet sync loop", "detail": "Every 15 minutes", "time": None}, + {"title": "Heartbeat", "detail": "Every 30 minutes", "time": None} + ] + } + res = acp.get_agent_menu("muse", tab="upcoming") + self.assertEqual(res["node"], "muse") + self.assertEqual(res["tab"], "Upcoming") + self.assertEqual(len(res["data"]["items"]), 2) + self.assertEqual(res["data"]["items"][0]["title"], "Fleet sync loop") + + +class TestBoxWorkCognitiveIntegration(unittest.TestCase): + + @patch("box_work.acp.get_passive_cognitive_state") + def test_cognitive_lock_blocks_start(self, mock_cog): + mock_cog.return_value = { + "node": "pip", + "status": "THINKING", + "cognitive_lock": True, + "lock_reason": "Stop button active" + } + args = MagicMock() + args.title = "New build" + args.agent = "pip" + args.goal = None + args.force = False + + with self.assertRaises(SystemExit) as cm: + box_work.cmd_start(args) + self.assertEqual(cm.exception.code, 1) + + @patch("box_work.acp.get_passive_cognitive_state") + def test_cognitive_lock_blocks_assign(self, mock_cog): + mock_cog.return_value = { + "node": "646", + "status": "INPUT_WAIT", + "cognitive_lock": True, + "lock_reason": "Parked input card" + } + args = MagicMock() + args.issue = 218 + args.agent = "646" + args.force = False + + with self.assertRaises(SystemExit) as cm: + box_work.cmd_assign(args) + self.assertEqual(cm.exception.code, 1) + + +if __name__ == "__main__": + unittest.main()