feat(cognitive): unlock settled sidechats as SIDECHAT_IDLE and add Main Chat refocus

This commit is contained in:
operator
2026-10-10 11:14:34 -04:00
parent 073ef556ba
commit 5f7e1e3a76
3 changed files with 131 additions and 4 deletions
+63 -4
View File
@@ -225,14 +225,14 @@ def get_passive_cognitive_state(node):
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}'"
elif not is_main:
state = "SIDECHAT_IDLE"
locked = False
reason = None
else:
state = "IDLE"
locked = False
@@ -398,6 +398,58 @@ def get_agent_live_screen(node: str) -> dict:
}
JS_NAVIGATE_MAIN_CHAT = """(() => {
try {
const url = window.location.href;
if (url === 'https://muse.ai/' || url === 'https://muse.ai/thread/new') {
return { ok: true, already_main: true };
}
const candidates = Array.from(document.querySelectorAll('a, button, [role="button"], [data-testid]'));
const mainBtn = candidates.find(b => {
const t = (b.innerText || '').trim().toLowerCase();
return t === 'main chat' || b.getAttribute('aria-label') === 'Main chat' || b.getAttribute('data-testid') === 'hatch-sidebar-main-chat';
});
if (mainBtn) {
mainBtn.click();
return { ok: true, method: 'click' };
}
window.location.href = 'https://muse.ai/';
return { ok: true, method: 'navigate' };
} catch (e) {
return { error: String(e) };
}
})()"""
def navigate_to_main_chat(node: str) -> dict:
"""Navigates the agent browser session back to Main Chat (https://muse.ai/)."""
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 nav-main {node} --json"]
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
if proc.returncode == 0 and proc.stdout.strip():
return json.loads(proc.stdout.strip())
except Exception as e:
return {"error": str(e)}
return run_cdp_eval_inside_netns(node, JS_NAVIGATE_MAIN_CHAT)
def cmd_nav_main(args):
node = getattr(args, "agent", None) or getattr(args, "node", None)
res = navigate_to_main_chat(node)
if getattr(args, "json", False):
print(json.dumps(res, indent=2))
else:
if isinstance(res, dict) and res.get("ok"):
m = res.get("method") or ("already in main chat" if res.get("already_main") else "default")
print(f"✓ Refocused agent '{node}' to Main Chat ({m})")
else:
err = res.get("error") if isinstance(res, dict) else str(res)
print(f"❌ Failed to refocus '{node}' to Main Chat: {err}")
def cmd_read(args):
node = getattr(args, "agent", None) or getattr(args, "node", None)
data = get_agent_live_screen(node)
@@ -451,6 +503,7 @@ def cmd_read(args):
def format_cognitive_badge(status):
badges = {
"IDLE": "🟢 IDLE",
"SIDECHAT_IDLE": "💬 SIDE_IDLE",
"THINKING": "🧠 THINKING",
"INPUT_WAIT": "⏸️ INPUT_WAIT",
"BUSY_SIDECHAT": "💬 SIDECHAT",
@@ -573,6 +626,10 @@ def main():
p_read.add_argument("node", help="Agent name")
p_read.add_argument("--json", action="store_true", help="Output JSON")
p_nav = sub.add_parser("nav-main", help="Navigate agent browser session back to Main Chat")
p_nav.add_argument("node", help="Agent name")
p_nav.add_argument("--json", action="store_true", help="Output JSON")
args = parser.parse_args()
if not args.command:
# Default to status
@@ -589,6 +646,8 @@ def main():
cmd_read(args)
elif args.command == "lock":
cmd_lock(args)
elif args.command == "nav-main":
cmd_nav_main(args)
if __name__ == "__main__":
+10
View File
@@ -746,6 +746,16 @@ def cmd_start(args):
pass
s.close()
# 3b. Refocus agent to Main Chat if resting in a sidechat
if acp:
try:
nav_res = acp.navigate_to_main_chat(agent)
if isinstance(nav_res, dict) and nav_res.get("ok") and not nav_res.get("already_main"):
print(c_dim(f"✓ Refocused @{agent} session to Main Chat"))
time.sleep(1.0)
except Exception:
pass
# 4. Notify agent via direct chat with Readback requirement
prompt_msg = (
f"New build ticket #{issue_num} assigned to you: {title}.\n"
+58
View File
@@ -14,11 +14,69 @@ class TestAgentCognitiveProbe(unittest.TestCase):
def test_badge_formatting(self):
self.assertIn("IDLE", acp.format_cognitive_badge("IDLE"))
self.assertIn("SIDE_IDLE", acp.format_cognitive_badge("SIDECHAT_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_sidechat_idle_state(self, mock_cdp, mock_bl):
# Simulate settled sidechat — generation complete, no parked modals
mock_cdp.return_value = {
"is_generating": False,
"is_typing": False,
"avatar_status": "Connected",
"is_main_chat": False,
"title": "Chat — test sidechat",
"is_input_wait": False
}
res = acp.get_passive_cognitive_state("def")
self.assertEqual(res["status"], "SIDECHAT_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_passive_sidechat_thinking_state(self, mock_cdp, mock_bl):
# In a sidechat but actively generating
mock_cdp.return_value = {
"is_generating": True,
"is_typing": False,
"avatar_status": "Connected",
"is_main_chat": False,
"title": "Chat — test sidechat",
"is_input_wait": False
}
res = acp.get_passive_cognitive_state("pip")
self.assertEqual(res["status"], "THINKING")
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_sidechat_input_wait_state(self, mock_cdp, mock_bl):
# In a sidechat with parked approval modal
mock_cdp.return_value = {
"is_generating": False,
"is_typing": False,
"avatar_status": "Connected",
"is_main_chat": False,
"title": "Chat — test sidechat",
"is_input_wait": True
}
res = acp.get_passive_cognitive_state("pip")
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_navigate_main_chat(self, mock_cdp, mock_bl):
mock_cdp.return_value = {"ok": True, "method": "click"}
res = acp.navigate_to_main_chat("def")
self.assertTrue(res["ok"])
self.assertEqual(res["method"], "click")
@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):