feat(cognitive): implement real-time cognitive sensing, menu navigation, and single-task lock

2026-10-10 10:27:24 -04:00
parent 3620b42297
commit bccdd29d53
5 changed files with 698 additions and 4 deletions
@@ -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()