给 trigger_rule_codes 加 JSON 多值索引；登记 P3 处理结果（docs/25 P3 #21）

实测确认 #21 描述准确：fin_risk_alert 只有 11 个普通 BTREE 索引 + 主键 + alert_no
唯一键，规则命中的 JSON_CONTAINS 查询 EXPLAIN 为 type=ALL、possible_keys=NULL，
即全表扫。MySQL 8.0.27 支持多值索引，故新增迁移 20260911_risk_rule_index：

    ADD INDEX idx_fin_risk_alert_trigger_rule_codes
      ((CAST(`trigger_rule_codes` AS CHAR(16) ARRAY)))

迁移幂等（先查 information_schema.STATISTICS），upgrade/downgrade 往返已验证。
生效后 EXPLAIN 变为 access_type=range 且 key 命中该索引，原始证据留档在
docs/evidence/risk-index-probe.json（由 tools/probe_risk_index.py 生成，只读探查）。

只解决一半，另一半如实记为限制：若干 like(f"%{keyword}%") 全表扫无法用 B-tree 索引，
根治需全文索引 + 中文分词组件（部署依赖），本轮不做。

revision 名刻意压到 32 字符以内 —— alembic_version.version_num 是 VARCHAR(32)，
超长会在写版本号时报 1406，而 DDL 是非事务的，那时索引已经建好了。

docs/25 追加"P3 处理结果"表，逐条登记 17-25 的状态：#19 是协议级重做（keyset 分页）
不单方面改，#21 部分修复，#25 前半段不成立，其余已修。

tools/probe_risk_index.py

@@ -0,0 +1,138 @@
+"""只读探查：`fin_risk_alert.trigger_rule_codes` 能否用 JSON 多值索引（docs/25 P3 #21）。
+
+只做 SELECT / EXPLAIN，不做任何写入。结果写成 JSON 便于在 GBK 控制台下查看：
+
+    python tools/probe_risk_index.py
+
+排查点：
+1. MySQL 版本是否 ≥ 8.0.17（多值索引的下限）；
+2. 列是否确实是 JSON、是否已存在同类索引；
+3. 是否存在**非数组**取值 —— 有的话 `CAST(... AS CHAR ARRAY)` 建索引会直接失败；
+4. 当前 `JSON_CONTAINS` 查询走的是什么访问路径（有没有可用的 key）。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import text
+
+from app.infrastructure.db import SessionFactory
+
+OUTPUT = Path("docs/evidence/risk-index-probe.json")
+
+
+async def collect() -> dict[str, Any]:
+    report: dict[str, Any] = {}
+    async with SessionFactory() as session:
+        report["mysql_version"] = (
+            await session.execute(text("SELECT VERSION()"))
+        ).scalar_one()
+
+        report["column"] = [
+            dict(row)
+            for row in (
+                (
+                    await session.execute(
+                        text(
+                            """
+                            SELECT COLUMN_NAME, COLUMN_TYPE, DATA_TYPE, IS_NULLABLE
+                            FROM information_schema.COLUMNS
+                            WHERE TABLE_SCHEMA = DATABASE()
+                              AND TABLE_NAME = 'fin_risk_alert'
+                              AND COLUMN_NAME = 'trigger_rule_codes'
+                            """
+                        )
+                    )
+                )
+                .mappings()
+                .all()
+            )
+        ]
+
+        report["indexes"] = [
+            dict(row)
+            for row in (
+                (
+                    await session.execute(
+                        text(
+                            """
+                            SELECT INDEX_NAME, COLUMN_NAME, INDEX_TYPE, EXPRESSION
+                            FROM information_schema.STATISTICS
+                            WHERE TABLE_SCHEMA = DATABASE()
+                              AND TABLE_NAME = 'fin_risk_alert'
+                            ORDER BY INDEX_NAME, SEQ_IN_INDEX
+                            """
+                        )
+                    )
+                )
+                .mappings()
+                .all()
+            )
+        ]
+
+        report["row_count"] = (
+            await session.execute(text("SELECT COUNT(*) FROM fin_risk_alert"))
+        ).scalar_one()
+        report["non_array_rows"] = (
+            await session.execute(
+                text(
+                    "SELECT COUNT(*) FROM fin_risk_alert "
+                    "WHERE trigger_rule_codes IS NULL "
+                    "   OR JSON_TYPE(trigger_rule_codes) <> 'ARRAY'"
+                )
+            )
+        ).scalar_one()
+        report["distinct_rule_codes"] = [
+            dict(row)
+            for row in (
+                (
+                    await session.execute(
+                        text(
+                            """
+                            SELECT CAST(trigger_rule_codes AS CHAR) AS raw_codes,
+                                   COUNT(*) AS rows_count
+                            FROM fin_risk_alert
+                            GROUP BY 1
+                            """
+                        )
+                    )
+                )
+                .mappings()
+                .all()
+            )
+        ]
+
+        # 用 FORMAT=JSON：传统 EXPLAIN 会把"该查询无法被缓存"写成 warning 打到 stderr，
+        # 让脚本在 CI 里看起来是失败的，而它其实成功了。
+        raw_explain = (
+            await session.execute(
+                text(
+                    """
+                    EXPLAIN FORMAT=JSON
+                    SELECT id FROM fin_risk_alert
+                    WHERE JSON_CONTAINS(trigger_rule_codes, '"RW-018"')
+                    """
+                )
+            )
+        ).scalar_one()
+        report["explain_json_contains"] = json.loads(raw_explain)
+
+    return report
+
+
+async def main() -> None:
+    report = await collect()
+    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
+    OUTPUT.write_text(
+        json.dumps(report, ensure_ascii=False, indent=2, default=str),
+        encoding="utf-8",
+    )
+    print(f"wrote {OUTPUT}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())