feat(work): implement synchronous readback gate and cognitive true loopback engine

tests/test_box_readback_loopback.py

@@ -0,0 +1,121 @@
+"""test_box_readback_loopback.py — Unit tests for Readback Gate and True Loopback Engine."""
+
+import unittest
+from unittest.mock import patch, MagicMock
+from datetime import datetime, timezone, timedelta
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "bin"))
+import box_readback_loopback as brl
+
+
+class TestReadbackValidation(unittest.TestCase):
+
+    def test_strict_tag_match(self):
+        text = "[READBACK] Ticket #219 | Branch: dev/pip/219-work | Plan: run pytest"
+        valid, excerpt = brl.validate_readback(text, 219, "pip")
+        self.assertTrue(valid)
+        self.assertIn("Ticket #219", excerpt)
+
+    def test_semantic_fallback_match(self):
+        text = "Understood. I am working on ticket 219 on dev/pip/219-work."
+        valid, excerpt = brl.validate_readback(text, 219, "pip")
+        self.assertTrue(valid)
+        self.assertIn("219", excerpt)
+
+    def test_irrelevant_message_rejected(self):
+        text = "Heartbeat check-in completed, all systems green."
+        valid, excerpt = brl.validate_readback(text, 219, "pip")
+        self.assertFalse(valid)
+        self.assertEqual(excerpt, "")
+
+    def test_wrong_ticket_rejected(self):
+        text = "[READBACK] Ticket #218 | Branch: dev/pip/218-work"
+        valid, excerpt = brl.validate_readback(text, 219, "pip")
+        self.assertFalse(valid)
+
+
+class TestLoopbackEscalation(unittest.TestCase):
+
+    @patch("box_readback_loopback.acp.get_passive_cognitive_state")
+    @patch("box_readback_loopback.gitea_api")
+    def test_loopback_silence_when_active(self, mock_gitea, mock_cog):
+        # 5 minutes inactive -> silence
+        now = datetime.now(timezone.utc)
+        recent = (now - timedelta(minutes=5)).isoformat()
+        mock_gitea.return_value = [
+            {"number": 219, "title": "Test", "created_at": recent, "assignee": {"username": "dev"}}
+        ]
+        mock_cog.return_value = {"status": "IDLE", "cognitive_lock": False}
+
+        actions = brl.run_loopback_sweep(dry_run=True, verbose=False)
+        self.assertEqual(len(actions), 0)
+
+    @patch("box_readback_loopback.acp.get_passive_cognitive_state")
+    @patch("box_readback_loopback.gitea_api")
+    def test_loopback_tier1_after_15m_when_idle(self, mock_gitea, mock_cog):
+        # 20 minutes inactive -> Tier 1
+        now = datetime.now(timezone.utc)
+        old = (now - timedelta(minutes=20)).isoformat()
+        mock_gitea.side_effect = lambda ep, **kwargs: (
+            [{"number": 999, "title": "Stalled", "created_at": old, "assignee": {"username": "dev"}}]
+            if ep == "/repos/super/box/issues?state=open"
+            else []
+        )
+        mock_cog.return_value = {"status": "IDLE", "cognitive_lock": False}
+
+        actions = brl.run_loopback_sweep(dry_run=True, verbose=False)
+        self.assertTrue(any("Tier 1" in a for a in actions))
+
+    @patch("box_readback_loopback.acp.get_passive_cognitive_state")
+    @patch("box_readback_loopback.gitea_api")
+    def test_loopback_defers_when_thinking(self, mock_gitea, mock_cog):
+        # 25 minutes inactive, but agent is THINKING -> defer
+        now = datetime.now(timezone.utc)
+        old = (now - timedelta(minutes=25)).isoformat()
+        mock_gitea.side_effect = lambda ep, **kwargs: (
+            [{"number": 999, "title": "Stalled", "created_at": old, "assignee": {"username": "dev"}}]
+            if ep == "/repos/super/box/issues?state=open"
+            else []
+        )
+        mock_cog.return_value = {"status": "THINKING", "cognitive_lock": True}
+
+        actions = brl.run_loopback_sweep(dry_run=True, verbose=False)
+        self.assertEqual(len(actions), 0)
+
+    @patch("box_readback_loopback.acp.get_passive_cognitive_state")
+    @patch("box_readback_loopback.gitea_api")
+    def test_loopback_tier2_after_45m(self, mock_gitea, mock_cog):
+        # 50 minutes inactive -> Tier 2
+        now = datetime.now(timezone.utc)
+        old = (now - timedelta(minutes=50)).isoformat()
+        mock_gitea.side_effect = lambda ep, **kwargs: (
+            [{"number": 999, "title": "Stalled", "created_at": old, "assignee": {"username": "dev"}}]
+            if ep == "/repos/super/box/issues?state=open"
+            else []
+        )
+        mock_cog.return_value = {"status": "IDLE", "cognitive_lock": False}
+
+        actions = brl.run_loopback_sweep(dry_run=True, verbose=False)
+        self.assertTrue(any("Tier 2" in a for a in actions))
+
+    @patch("box_readback_loopback.acp.get_passive_cognitive_state")
+    @patch("box_readback_loopback.gitea_api")
+    def test_loopback_tier3_after_90m(self, mock_gitea, mock_cog):
+        # 95 minutes inactive -> Tier 3
+        now = datetime.now(timezone.utc)
+        old = (now - timedelta(minutes=95)).isoformat()
+        mock_gitea.side_effect = lambda ep, **kwargs: (
+            [{"number": 999, "title": "Stalled", "created_at": old, "assignee": {"username": "dev"}}]
+            if ep == "/repos/super/box/issues?state=open"
+            else []
+        )
+        mock_cog.return_value = {"status": "IDLE", "cognitive_lock": False}
+
+        actions = brl.run_loopback_sweep(dry_run=True, verbose=False)
+        self.assertTrue(any("Tier 3" in a for a in actions))
+
+
+if __name__ == "__main__":
+    unittest.main()