feat: B9b 演示走查——核查单收口: ①预警类 API 固定 disclaimer(alerts/handle/aml-scan) ②aml/scan 幂等防护(find_recent_aml_alert 同客户当日已有 aml 单不出单+skipped_existing) ⑥集成测试前置断言(A-1/A-3/A-4 清洁检查 fail+重灌提示, A-5 按 trade_id 定位); 演示 SOP 落地(脚本化重灌 MYSQL_PWD 方式); 真服务走查 A-1~A-5/A-7/A-9+核查单①②④⑤ 11/11 PASS(核查单⑥防护现场复现), 208 绿; M2 达成

tests/test_aml_service.py

@@ -171,7 +171,50 @@ def test_scan_all_no_hit_creates_nothing(env):
     with core._engine.begin() as conn:
         conn.execute(text("UPDATE risk_aml_list SET is_active = 0"))
     summary = scan_all(core_ro=core, risk_repo=repo)
-    assert summary == {"scanned": 4, "hit_customers": 0, "alerts": []}
+    assert summary == {"scanned": 4, "hit_customers": 0, "alerts": [], "skipped_existing": []}
     with core._engine.connect() as conn:
         assert conn.execute(text("SELECT COUNT(*) FROM risk_alert")).scalar_one() == 0
     assert pub.messages == []
+
+
+def test_scan_all_idempotent_same_day(env):
+    """B9b 核查单②（B6 评审 P3-6）：同日重扫不重复出单，重复点击防护。"""
+    core, repo, pub = env
+    first = scan_all(core_ro=core, risk_repo=repo)
+    assert len(first["alerts"]) == 2 and first["skipped_existing"] == []
+    second = scan_all(core_ro=core, risk_repo=repo)
+    assert second["alerts"] == []
+    assert sorted(second["skipped_existing"]) == sorted(first["alerts"])
+    assert len(pub.messages) == 2  # 重扫无新推送（出单时的推送各一次）
+
+
+def test_scan_skips_customer_already_alerted_by_trade(env):
+    """当日交易触发已出 aml 单的客户，scan 不再重复出单（同日命中留痕一次）。"""
+    from decimal import Decimal
+
+    from app.service.risk.engine import process_trade_event
+
+    core, repo, pub = env
+    with core._engine.begin() as conn:
+        conn.execute(
+            text(
+                "INSERT INTO core_trade (trade_id, customer_id, product_id, trade_type,"
+                " amount, trade_status, traded_at)"
+                " VALUES ('T-SCAN-1', 'C2', 'P1', 'subscribe', 1000, 'confirmed', :at)"
+            ),
+            {"at": datetime(2026, 9, 7, 10, 0, 0)},
+        )
+    result = process_trade_event(
+        {
+            "trade_id": "T-SCAN-1", "customer_id": "C2", "product_id": "P1",
+            "trade_type": "subscribe", "amount": Decimal("1000"),
+            "trade_status": "confirmed", "traded_at": datetime(2026, 9, 7, 10, 0, 0),
+        },
+        core_ro=core, risk_repo=repo,
+    )
+    assert result["aml_hit"] is True  # 交易触发已出 C2 的 aml 单
+    summary = scan_all(core_ro=core, risk_repo=repo)
+    assert len(summary["alerts"]) == 1  # 仅 C1（无当日 aml 单）
+    assert summary["skipped_existing"] == [
+        aid for aid in result["alert_ids"] if repo.get_alert(aid)["alert_type"] == "aml"
+    ]