"""P4b 贝叶斯闭环端到端验收：全清、PF 缺席证明、预算与故障。 对齐 test_loop_q3.py 的验收口径（docs/algo-3-bayes.md）： - 10/16 源场景全部清除且守恒性零异常（凸证书 + 网格证书均不得失效）； - PF 缺席证明（幸存集为空）与覆盖账本口径一致，不误判有源频道； - 无源场景全部频道证明缺席后正常退出（persistent 关闭时 all_resolved）； - 近圆域边界源（r_eff=1000）不漏搜； - 持续复扫（persistent 默认开）：找全后复扫到虚拟安全余量， accounted_virtual_time_s 记录首次证明找全时刻； - 网络故障幂等重试不污染闭环结果； - 虚拟时刻逐步复算一致。 """ import math import unittest from dataclasses import replace from mcm26b.config import ProblemConfig from mcm26b.evaluation import recompute_virtual_time from mcm26b.planner.bayes_loop import BayesLoop from mcm26b.protocol.client import SimulatorClient from mcm26b.sim import (LocalSimTransport, LocalSimulator, ScriptedFaultTransport, Source) from mcm26b.sim.scenes import random_omni_scene CONFIG = replace(ProblemConfig(), search=replace(ProblemConfig().search, bearing_rounding_margin_deg=0.005)) SMALL_CAP = replace(CONFIG, field=replace(CONFIG.field, max_virtual_duration_s=60_000.0)) NO_PERSIST = replace(CONFIG, search=replace(CONFIG.search, persistent_search=False)) def _run_scene(sources, fault_script=None, config=SMALL_CAP): simulator = LocalSimulator(sources, robot_id="T001", config=config) transport = LocalSimTransport(simulator) if fault_script is not None: transport = ScriptedFaultTransport(transport, fault_script) client = SimulatorClient("T001", config=config, transport=transport) loop = BayesLoop(client, config) summary = loop.run() report = recompute_virtual_time(client.log.action_events()) return summary, report, simulator class BayesLoopAcceptanceTests(unittest.TestCase): def test_ten_sources_all_cleared(self): sources = random_omni_scene(seed=5, count=10, config=CONFIG) summary, report, simulator = _run_scene(sources) truth = sorted(s.channel for s in sources) self.assertTrue(summary["completed"]) self.assertEqual(sorted(summary["cleared_channels"]), truth) self.assertEqual(summary["unresolved_channels"], []) self.assertIsNotNone(summary["accounted_virtual_time_s"]) self.assertEqual(summary["anomalies"], [], "守恒性或证书不应被打破") self.assertEqual(report["mismatches"], []) self.assertEqual(sorted(simulator.cleared), truth) self.assertEqual(summary["exit_reason"], "user_exit") self.assertEqual(summary["strategy"], "bayes") def test_sixteen_sources_stop_by_count_bound(self): sources = random_omni_scene(seed=6, count=16, config=CONFIG) summary, report, simulator = _run_scene(sources) truth = sorted(s.channel for s in sources) self.assertTrue(summary["completed"]) self.assertEqual(summary["stop_reason"], "sixteen_cleared") self.assertEqual(sorted(summary["cleared_channels"]), truth) self.assertEqual(report["mismatches"], []) self.assertEqual(summary["anomalies"], []) def test_empty_arena_all_absent(self): """0 源：每个频道都必须由 PF 排除盘证明缺席后正常退出。""" summary, report, simulator = _run_scene([], config=NO_PERSIST) self.assertTrue(summary["completed"]) self.assertEqual(summary["stop_reason"], "all_resolved") self.assertEqual(sorted(summary["absent_channels"]), list(range(1, 21))) self.assertEqual(summary["pf_absent_channels"], list(range(1, 21))) self.assertEqual(summary["clears"], 0) self.assertEqual(summary["anomalies"], []) self.assertEqual(report["mismatches"], []) def test_boundary_sources_detected_and_cleared(self): """近圆域边界源（r_eff=1000）：七站覆盖半径 968.9m 保证发现。""" sources = [Source(c + 1, 1800.0 * math.cos(math.pi / 6 + 2 * math.pi * c / 4), 1800.0 * math.sin(math.pi / 6 + 2 * math.pi * c / 4), 1000.0) for c in range(4)] summary, report, simulator = _run_scene(sources, config=NO_PERSIST) truth = sorted(s.channel for s in sources) self.assertTrue(summary["completed"]) self.assertEqual(sorted(summary["cleared_channels"]), truth) self.assertEqual(summary["anomalies"], []) self.assertEqual(report["mismatches"], []) def test_persistent_phase_resweeps_until_cap(self): """复扫保险模式（--persistent）：找全后复扫到虚拟安全余量。""" config = replace(SMALL_CAP, search=replace(SMALL_CAP.search, persistent_search=True)) sources = random_omni_scene(seed=5, count=10, config=CONFIG) summary, _report, _simulator = _run_scene(sources, config=config) truth = sorted(s.channel for s in sources) self.assertTrue(summary["completed"]) self.assertEqual(summary["stop_reason"], "virtual_time_cap") self.assertEqual(sorted(summary["cleared_channels"]), truth) self.assertGreater(summary["measures"], 7 * 20, "找全后应继续复扫未清除频道") def test_transport_fault_retries_keep_session_clean(self): sources = random_omni_scene(seed=5, count=10, config=CONFIG) script = [("drop_before", 1)] # 首个请求丢弃一次：幂等重试路径 summary, report, simulator = _run_scene(sources, fault_script=script) truth = sorted(s.channel for s in sources) self.assertTrue(summary["completed"]) self.assertEqual(sorted(summary["cleared_channels"]), truth) self.assertEqual(report["mismatches"], []) def test_skip_rule_never_skips_detectable_channel(self): """跳测规则只允许跳过"必 no_signal"的 (站,频道) 对。""" sources = random_omni_scene(seed=5, count=10, config=CONFIG) simulator = LocalSimulator(sources, robot_id="T001", config=CONFIG) client = SimulatorClient("T001", config=CONFIG, transport=LocalSimTransport(simulator)) loop = BayesLoop(client, CONFIG) loop.run() from mcm26b.belief import ChannelStatus for index, point in enumerate(loop.ledger.points): for channel, belief in loop.belief.channels.items(): if belief.status is not ChannelStatus.UNKNOWN: continue skipped = loop._skip_station_channel(channel, index, point) if skipped and len(loop.particles[channel]): nearest = loop.particles[channel].nearest_alive_distance( point, loop.particles[channel].r_eff_max_m + loop.particles[channel].margin_m) self.assertIsNone( nearest, f"频道 {channel} 站点 {index} 被跳过但可能检出") def test_cli_strategy_registry(self): """CLI --strategy 注册表：bayes/seven 用七站，grid 用 49 点网格。""" import mcm26b.cli as cli from mcm26b.coverage import coverage_points from mcm26b.planner.seven import SevenBatch, seven_points self.assertIs(cli.Q3_STRATEGIES["bayes"], BayesLoop) for name, loop_cls in cli.Q3_STRATEGIES.items(): simulator = LocalSimulator([], robot_id="T001", config=CONFIG) client = SimulatorClient("T001", config=CONFIG, transport=LocalSimTransport(simulator)) loop = loop_cls(client, CONFIG) expected = seven_points() if name in ("bayes", "seven") \ else coverage_points(CONFIG) self.assertEqual(loop.ledger.points, expected) if __name__ == "__main__": unittest.main()