2

tests/test_loop_q3.py

@@ -112,6 +112,38 @@ class LoopAcceptanceTests(unittest.TestCase):
         self.assertEqual(sorted(simulator.cleared), truth)
 
 
+class AbortFinalizationTests(unittest.TestCase):
+    """断连超过重试上限：仍须 /exit、产出未完成 summary，不得静默崩溃。"""
+
+    def test_transport_failure_aborts_with_summary_and_exit(self):
+        """首个检测动作三次断连：run() 正常返回，/exit 成功，stop_reason 留档。"""
+        summary, _, _ = _run_scene(
+            random_omni_scene(seed=5, count=2),
+            fault_script=[None, "drop_before", "drop_before", "drop_before"])
+        self.assertFalse(summary["completed"])
+        self.assertEqual(summary["stop_reason"], "transport_failure")
+        self.assertEqual(summary["exit_reason"], "user_exit",
+                         "接口仍可用时必须尝试退出")
+        self.assertEqual(summary["abort_detail"]["type"], "TransportError")
+        self.assertLessEqual(summary["measures"] + summary["clears"],
+                             summary["total_requests"])
+        self.assertEqual(len(summary["channel_status"]), 20,
+                         "未完成状态须有逐频道快照")
+
+    def test_exit_failure_still_produces_summary(self):
+        """检测与 /exit 均三次断连：产出 exit_unreachable 的未完成 summary。"""
+        summary, _, _ = _run_scene(
+            random_omni_scene(seed=7, count=2),
+            fault_script=[None,
+                          "drop_before", "drop_before", "drop_before",
+                          "drop_before", "drop_before", "drop_before"])
+        self.assertFalse(summary["completed"])
+        self.assertEqual(summary["stop_reason"], "transport_failure")
+        self.assertEqual(summary["exit_reason"], "exit_unreachable")
+        self.assertEqual(summary["exit_error"]["type"], "TransportError")
+        self.assertIn("channel_status", summary)
+
+
 class CoverageLedgerIntegrationTests(unittest.TestCase):
     def test_grid_matches_plan_derivation(self):
         points = coverage_points(CONFIG)