feat(data-quality): 财务七规则可配置化最小实现（ASK-20260928-05）

任总 2026-09-28 13:38 放行「4 和 5」→ 落笔最小实现：
- 新表 quality_rule_config（39 项：37 规则级 + 2 全局）= 唯一配置源
- 三层解析：账套行 → 全账套模板(entity_id=0) → 代码内置默认（=清单 §7 建议值，零行为差异）
- 端点 /api/cma/data-quality/rule-config（GET/PUT）+ /history + /meta：写入仅 ceo/it，
  platform 级仅 it，演示账套(functional)写禁，变更落 operation_logs 同事务
- 数据质量页新增「规则配置」只读 Tab（懒加载）
- 唯一行为变更：污染判定去掉 sync —— 酣客 zxt-sync 系统迁入数据不再被误判（口径可配置）
- 出厂默认落库脚本 scripts/seed_dq_rule_config.py（--check/--apply/--rollback，幂等 39/0）
- 回归：新套件 35 例 + 存量 1508 passed（唯一 FAILED 为他人未完成
  migrate_subjects_entity_id_20260923.py 改动所致，HEAD 干净树同例通过）

生产写（DDL/seed）与 push 均走审批门：DDL=54434656 push=3bd00774

backend/tests/test_dq_rule_config_20260928.py

@@ -0,0 +1,529 @@
+# -*- coding: utf-8 -*-
+"""CMA 财务七规则可配置化（ASK-20260928-05）回归套件 —— 规格 / 三层解析 / 引擎接线 / 端点。
+
+任务：任总 2026-09-28 13:38「放行 4 和 5」→ 最小实现 = 新表 `quality_rule_config` + 端点 + 只读 UI，
+外加本单唯一的行为变更（去掉 `sync` 污染判定）与「口径用清单默认值」。
+
+判据分四组：
+  A 规格面：37 规则级参数 + 2 全局项；默认值 == 清单 §7；边界「拒绝 / 钳制」两类；锁死项 400。
+  B 引擎接线（**真函数** `_run_governance_checks` + 夹具库，非重写逻辑）：
+      · 默认配置 = 原硬编码口径（零行为差异，除 sync）；
+      · **sync A/B 三态**：默认(0) / 恢复旧关键词表(命中) / 白名单豁免(0) —— 证明配置真的接线；
+      · 阈值、条数下限、开关、扣分上限系数、比较方向、明细条数逐项可配；
+      · `/rule-detail` total 与新口径一致性（两处同源，不漂移）。
+  C 端点面：GET 只读四角色可见；PUT 仅 ceo/it（平台级仅 it）；Bot 通道禁写；演示账套拒写；
+      change_reason 必填；锁死/未知参数 400；越界拒绝或钳制；写行与 `operation_logs` 同事务。
+  D 韧性：配置表缺失（DDL 未落地）时读路径回落内置默认、不 500（部署顺序解耦）。
+"""
+import json
+from datetime import datetime
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app import dq_rule_config as dqcfg
+from app.api.data_quality import _run_governance_checks
+from app.models import (
+    KPIDefinition, KPIValue, KPIAlert, CashPlan, BudgetPlan, QualityRuleConfig,
+    OperationLog, Entity,
+)
+from tests.conftest import create_test_user, get_token_for_user, auth_header
+
+REASON = "回归用例：调整治理规则阈值以验证配置化接线（≥10 字）"
+
+
+# ══════════════════════════════════════════════════════════════════
+# 夹具
+# ══════════════════════════════════════════════════════════════════
+def _mk_kpi(db: Session, **kw) -> KPIDefinition:
+    defaults = {
+        "entity_id": 1, "kpi_code": "KPI_X", "kpi_name": "夹具KPI", "status": "active",
+        "target_value": 100.0, "target_yearly": 100.0, "formula": "x",
+        "data_source": "test", "data_owner": "财务部", "unit": "元",
+    }
+    defaults.update(kw)
+    k = KPIDefinition(**defaults)
+    db.add(k)
+    db.commit()
+    db.refresh(k)
+    return k
+
+
+def _mk_plan(db: Session, **kw) -> CashPlan:
+    defaults = {
+        "entity_id": 1, "plan_type": "receive", "amount": 10.0, "source": "manual",
+        "status": "pending", "plan_date": datetime(2026, 8, 1),
+    }
+    defaults.update(kw)
+    p = CashPlan(**defaults)
+    db.add(p)
+    db.commit()
+    db.refresh(p)
+    return p
+
+
+def _mk_row(db: Session, rule_key: str, param_key: str, value, **kw) -> QualityRuleConfig:
+    """直接落/更新一行配置（模拟「已存在配置」；端点写路径另有用例）。"""
+    entity_id = kw.pop("entity_id", 1)
+    exist = (db.query(QualityRuleConfig)
+               .filter_by(entity_id=entity_id, rule_key=rule_key, param_key=param_key).first())
+    if exist is not None:
+        exist.param_value = json.dumps(value, ensure_ascii=False)
+        exist.value_type = dqcfg.value_type_of(rule_key, param_key)
+        db.commit()
+        db.refresh(exist)
+        return exist
+    row = QualityRuleConfig(
+        entity_id=entity_id, rule_key=rule_key, param_key=param_key,
+        param_value=json.dumps(value, ensure_ascii=False),
+        value_type=dqcfg.value_type_of(rule_key, param_key),
+        open_level=dqcfg.open_level_of(rule_key, param_key),
+        is_default=kw.pop("is_default", False), updated_by=kw.pop("updated_by", "fixture"),
+        **kw,
+    )
+    db.add(row)
+    db.commit()
+    db.refresh(row)
+    return row
+
+
+def _counts(db: Session, eid: int = 0):
+    """跑一次真函数，返回 (按规则聚合的 issues, 完整结果)。"""
+    r = _run_governance_checks(db, eid)
+    return {i["rule"]: i for i in r["issues"]}, r
+
+
+def _vp_fixture(db: Session) -> None:
+    """规则4 夹具：5 条正常同步来源（酣客形态）+ 2 条测试来源 + 1 条 manual + 1 条博海来源。"""
+    for i in range(5):
+        _mk_plan(db, id=100 + i, source="zxt-sync", amount=10.0 + i)
+    for i in range(2):
+        _mk_plan(db, id=200 + i, source="test_import", amount=20.0 + i)
+    _mk_plan(db, id=300, source="manual", amount=30.0)
+    _mk_plan(db, id=301, source="bohai_ar_20260630", amount=40.0)
+
+
+# ══════════════════════════════════════════════════════════════════
+# A 规格面
+# ══════════════════════════════════════════════════════════════════
+class TestSpecSurface:
+    def test_param_count_is_37_rule_level_plus_2_global(self):
+        """清单 §10.2：37 个规则级参数 + 2 个全局项；且 (rule, param) 无重复。"""
+        assert dqcfg.count_rule_params() == 37
+        assert len(dqcfg.GLOBAL_SPEC) == 2
+        pairs = dqcfg.all_param_pairs()
+        assert len(pairs) == 39
+        assert len(set(pairs)) == 39
+        per_rule = {r: len(dqcfg.SPEC[r]) for r in dqcfg.RULE_KEYS}
+        assert per_rule == {"unit_check": 5, "dup_alert": 5, "orphan_check": 4,
+                            "virtual_pollution": 9, "entity_check": 4,
+                            "kpi_completeness": 4, "reconciliation": 6}
+
+    def test_defaults_match_inventory_table(self):
+        """清单 §7 建议默认值逐项（口径「用清单默认值」）；唯一行为变更 = 关键词表无 sync。"""
+        vp = dqcfg.SPEC["virtual_pollution"]
+        assert vp["suspect_keywords"]["default"] == ["test", "虚拟", "demo"]
+        assert "sync" not in vp["suspect_keywords"]["default"]
+        assert vp["source_whitelist"]["default"] == [
+            "zxt-sync", "bohai_ar", "bohai_ar_20260630", "bohai_prepay_20260630",
+            "bohai_ap_20260630", "receivables_migrate"]
+        assert vp["manual_source_value"]["default"] == "manual"
+        assert vp["manual_cap"]["default"] == 10 and vp["manual_factor"]["default"] == 1
+        assert dqcfg.SPEC["unit_check"]["amount_threshold"]["default"] == 10000
+        assert dqcfg.SPEC["dup_alert"]["min_count"]["default"] == 2
+        assert dqcfg.SPEC["reconciliation"]["diff_threshold_pct"]["default"] == 20
+        assert dqcfg.SPEC["reconciliation"]["compare_mode"]["default"] == "abs"
+        for rule in dqcfg.RULE_KEYS:
+            caps = dqcfg.SPEC[rule]
+            assert caps["enabled"]["default"] is True
+            assert caps["detail_limit"]["default"] == 10
+            assert caps["deduct_factor"]["default"] == (3 if rule in
+                   ("unit_check", "dup_alert", "orphan_check", "virtual_pollution", "entity_check") else 1)
+        assert dqcfg.GLOBAL_SPEC["detail_limit"]["default"] == 10
+        # F6 排除名单默认空集（不排除）→ 不为本单引入第二个行为变更
+        assert dqcfg.GLOBAL_SPEC["scope.excluded_entities"]["default"] == []
+
+    def test_locked_items_cover_inventory_ten(self):
+        """清单 §4 十类锁死面完整登记（少一类即视为私自扩面）。"""
+        assert len(dqcfg.LOCKED_ITEMS) == 10
+        keys = {it["key"] for it in dqcfg.LOCKED_ITEMS}
+        assert {"expression", "level", "score_fn", "entity_scope", "rule_set",
+                "match_mode", "completeness_caliber", "null_entity_policy",
+                "bot_write", "weight_range"} == keys
+
+    @pytest.mark.parametrize("rule,param,value,expect_ok", [
+        ("unit_check", "amount_threshold", 5, False),           # 越界 → 拒绝
+        ("unit_check", "amount_threshold", 1000, True),
+        ("dup_alert", "min_count", 1, False),                   # 1 无意义 → 拒绝
+        ("dup_alert", "min_count", 3, True),
+        ("reconciliation", "diff_threshold_pct", 500, False),   # 越界 → 拒绝
+        ("reconciliation", "compare_mode", "both", False),      # 非枚举 → 拒绝
+        ("unit_check", "enabled", "yes-please", False),         # 非布尔 → 拒绝
+        ("virtual_pollution", "suspect_keywords", ["test", ""], False),   # 空元素 → 拒绝
+        ("virtual_pollution", "suspect_keywords", ["te%st"], False),      # 通配符 → 拒绝
+        ("virtual_pollution", "suspect_keywords", ["a" * 33], False),     # 元素超长 → 拒绝
+        ("virtual_pollution", "manual_source_value", "", False),          # 空串 → 拒绝
+    ])
+    def test_validate_reject_cases(self, rule, param, value, expect_ok):
+        r = dqcfg.validate_param(rule, param, value)
+        assert r["ok"] is expect_ok
+        if not expect_ok:
+            assert r["error"]
+
+    def test_validate_clamp_cases(self):
+        """权重/明细条数越界 → 钳制到端点 + 告警（清单 §3 的「钳制 + 告警」）。"""
+        r = dqcfg.validate_param("kpi_completeness", "deduct_cap", 999)
+        assert r["ok"] and r["value"] == 50 and r["warning"]
+        r = dqcfg.validate_param("unit_check", "deduct_factor", -3)
+        assert r["ok"] and r["value"] == 0 and r["warning"]
+        r = dqcfg.validate_param("unit_check", "detail_limit", 0)
+        assert r["ok"] and r["value"] == 1 and r["warning"]
+
+    def test_locked_and_unknown_params_rejected(self):
+        for param in ("level", "expression", "score_fn", "regex", "exempt_kpi_codes"):
+            r = dqcfg.validate_param("virtual_pollution", param, "x")
+            assert not r["ok"]
+            assert "锁死" in r["error"]
+        r = dqcfg.validate_param("unit_check", "no_such_param", 1)
+        assert not r["ok"] and "未知参数" in r["error"]
+
+
+# ══════════════════════════════════════════════════════════════════
+# B 引擎接线（真函数 + 夹具库）
+# ══════════════════════════════════════════════════════════════════
+class TestEngineWiring:
+    def test_default_zero_behavior_change_except_sync(self, db):
+        """默认配置 = 原硬编码口径：阈值 10000 / 条数 2 / 差异 20%；sync 来源不再算污染。"""
+        _vp_fixture(db)
+        _mk_plan(db, id=400, amount=50000.0, source="bohai_ar")   # 单位校验命中（>10000）
+        k = _mk_kpi(db, kpi_code="K1", target_yearly=100.0)
+        db.add(KPIValue(kpi_id=k.id, entity_id=1, period="2026-06", actual_value=50.0))
+        for m in range(1, 13):
+            db.add(BudgetPlan(kpi_id=k.id, entity_id=1, period=f"2026-{m:02d}",
+                              budget_value=200.0 / 12, budget_year=2026, budget_month=m,
+                              version="v1.0", status="active"))
+        _mk_kpi(db, kpi_code="K2", target_yearly=10.0)   # 无值 → KPI完整性
+        db.commit()
+
+        issues, r = _counts(db, 1)
+        assert issues["unit_check"]["count"] == 1
+        assert issues["virtual_pollution"]["count"] == 2          # 只剩 test_import 两条
+        assert issues["virtual_pollution"]["manual_count"] == 1
+        assert issues["kpi_completeness"]["count"] == 1
+        assert issues["reconciliation"]["count"] == 1
+        assert r["config_version"] and len(r["config_version"]) == 12
+        assert r["disabled"] == []
+        assert r["config_source"]["table_ready"] is True
+        assert r["scope_excluded_entities"] == []
+        assert r["excluded_from_business_caliber"] is False
+
+    def test_sync_ab_three_states(self, db):
+        """A/B：默认(0 命中) → 恢复旧关键词表+空白名单(7 命中) → 只留白名单(2 命中)。
+
+        三态各跑一次**同一个** `_run_governance_checks`（不是重写逻辑），
+        证明「配置真的驱动引擎」+「去掉 sync 确实消除了 100% 假阳性」。
+        """
+        _vp_fixture(db)
+
+        issues, _ = _counts(db, 1)
+        assert issues["virtual_pollution"]["count"] == 2, "默认：zxt-sync 不再误判"
+
+        legacy = _mk_row(db, "virtual_pollution", "suspect_keywords",
+                         ["test", "sync", "虚拟", "demo"])          # 改前口径
+        _mk_row(db, "virtual_pollution", "source_whitelist", [])     # 改前：无白名单
+        issues, r = _counts(db, 1)
+        assert issues["virtual_pollution"]["count"] == 7, "恢复旧口径 → zxt-sync 5 条重新被误判"
+        # 扣分 = min(15, 7×3) + manual 附加 min(10, 1×1) = 16（夹具里有 1 条 manual）
+        assert issues["virtual_pollution"]["deducted"] == 16
+
+        db.query(QualityRuleConfig).filter(QualityRuleConfig.rule_key == "virtual_pollution").delete()
+        _mk_row(db, "virtual_pollution", "source_whitelist", ["zxt-sync"])
+        issues, _ = _counts(db, 1)
+        assert issues["virtual_pollution"]["count"] == 2, "白名单豁免优先于关键词（豁免即不算污染）"
+        db.delete(legacy)
+        db.commit()
+
+    def test_threshold_and_switch_and_caps(self, db):
+        _mk_plan(db, id=500, amount=150.0)
+        _mk_plan(db, id=501, amount=50000.0)
+
+        issues, _ = _counts(db, 1)
+        assert issues["unit_check"]["count"] == 1                    # 默认 10000
+
+        _mk_row(db, "unit_check", "amount_threshold", 100)           # 下调阈值 → 2 条
+        issues, _ = _counts(db, 1)
+        assert issues["unit_check"]["count"] == 2
+
+        _mk_row(db, "unit_check", "deduct_cap", 0)                   # 只报不计分
+        issues, r = _counts(db, 1)
+        assert issues["unit_check"]["count"] == 2 and issues["unit_check"]["deducted"] == 0
+
+        _mk_row(db, "unit_check", "enabled", False)                  # 关开关 → 零命中零扣分
+        issues, r = _counts(db, 1)
+        assert issues["unit_check"]["count"] == 0 and issues["unit_check"]["deducted"] == 0
+        assert issues["unit_check"]["enabled"] is False
+        assert "unit_check" in r["disabled"] and "unit_check" not in r["passed"]
+        assert r["total_rules"] == 7                                  # 规则集锁死 7 条
+
+    def test_dup_alert_min_count(self, db):
+        k = _mk_kpi(db, kpi_code="K9", target_yearly=10.0)
+        _mk_plan(db, id=600, source="manual")
+        for _ in range(2):                                            # plan#600 两条 pending
+            db.add(KPIAlert(kpi_id=k.id, alert_type="cash_plan", status="pending",
+                            alert_message="x", suggestion=json.dumps({"plan_id": 600})))
+        db.commit()
+        issues, _ = _counts(db, 1)
+        assert issues["dup_alert"]["count"] == 1                      # 默认 ≥2 即重复
+
+        _mk_row(db, "dup_alert", "min_count", 3)
+        issues, _ = _counts(db, 1)
+        assert issues["dup_alert"]["count"] == 0                      # 阈值抬到 3 → 不再重复
+
+    def test_reconciliation_threshold_and_direction(self, db):
+        k = _mk_kpi(db, kpi_code="K7", target_yearly=100.0)
+        for m in range(1, 13):
+            db.add(BudgetPlan(kpi_id=k.id, entity_id=1, period=f"2026-{m:02d}",
+                              budget_value=200.0 / 12, budget_year=2026, budget_month=m,
+                              version="v1.0", status="active"))    # 合计 200 → +100%
+        k2 = _mk_kpi(db, kpi_code="K8", target_yearly=100.0)
+        for m in range(1, 13):
+            db.add(BudgetPlan(kpi_id=k2.id, entity_id=1, period=f"2026-{m:02d}",
+                              budget_value=50.0 / 12, budget_year=2026, budget_month=m,
+                              version="v1.0", status="active"))    # 合计 50 → -50%
+        db.commit()
+
+        issues, _ = _counts(db, 1)
+        assert issues["reconciliation"]["count"] == 2                 # 默认 abs 双侧 >20%
+
+        _mk_row(db, "reconciliation", "diff_threshold_pct", 150)
+        issues, _ = _counts(db, 1)
+        assert issues["reconciliation"]["count"] == 0                 # 阈值抬高
+
+        _mk_row(db, "reconciliation", "diff_threshold_pct", 20)
+        _mk_row(db, "reconciliation", "compare_mode", "upper_only")
+        issues, _ = _counts(db, 1)
+        assert issues["reconciliation"]["count"] == 1                 # 只看「超目标」→ 低于目标的 K8 不算
+
+        _mk_row(db, "reconciliation", "compare_mode", "lower_only")
+        issues, _ = _counts(db, 1)
+        assert issues["reconciliation"]["count"] == 1                 # 只看「低于目标」→ K8
+
+    def test_detail_limit_configurable(self, db):
+        for i in range(4):
+            _mk_plan(db, id=700 + i, amount=50000.0)
+        issues, _ = _counts(db, 1)
+        assert len([d for d in issues["unit_check"]["detail"] if not d.startswith("…")]) == 4
+
+        _mk_row(db, "unit_check", "detail_limit", 2)
+        issues, _ = _counts(db, 1)
+        rows = [d for d in issues["unit_check"]["detail"] if not d.startswith("…")]
+        assert len(rows) == 2 and any(d.startswith("…") for d in issues["unit_check"]["detail"])
+
+    def test_entity_config_isolation_no_cross_entity_fallback(self, db):
+        """账套隔离：账套 2 的行不得被账套 1 读到（禁止跨账套回退）。"""
+        _mk_plan(db, id=800, entity_id=1, amount=50000.0)
+        _mk_plan(db, id=801, entity_id=2, amount=50000.0)
+        _mk_row(db, "unit_check", "amount_threshold", 1000000, entity_id=2)
+
+        issues1, _ = _counts(db, 1)
+        issues2, _ = _counts(db, 2)
+        assert issues1["unit_check"]["count"] == 1          # 账套1 用默认 10000
+        assert issues2["unit_check"]["count"] == 0          # 账套2 用自己那行 1000000
+
+
+# ══════════════════════════════════════════════════════════════════
+# C 端点面
+# ══════════════════════════════════════════════════════════════════
+def _ceo_token(db: Session, client: TestClient) -> str:
+    create_test_user(db, username="cfgceo", role="ceo")
+    return get_token_for_user(client, "cfgceo")
+
+
+class TestConfigEndpoints:
+    def test_get_rule_config_readonly_shape(self, db, client):
+        token = _ceo_token(db, client)
+        r = client.get("/api/cma/data-quality/rule-config", headers=auth_header(token))
+        assert r.status_code == 200, r.text
+        body = r.json()
+        assert body["entity_id"] == 1
+        assert body["entity_config_enabled"] is True
+        assert body["config_entities"] == [1, 2]
+        assert body["rule_param_count"] == 37
+        assert set(body["rules"].keys()) == set(dqcfg.RULE_KEYS)
+        vp = body["rules"]["virtual_pollution"]["params"]
+        assert vp["suspect_keywords"]["value"] == ["test", "虚拟", "demo"]
+        assert vp["suspect_keywords"]["source"] == "builtin"
+        assert vp["suspect_keywords"]["open_level"] == "entity"
+        assert body["global_params"]["detail_limit"]["value"] == 10
+        assert body["scope"]["excluded_entities"] == []
+        assert len(body["locked"]) == 10
+        assert body["readonly"] is True
+
+    def test_get_rule_config_finance_readonly_allowed(self, db, client):
+        create_test_user(db, username="cfgfin", role="finance")
+        token = get_token_for_user(client, "cfgfin")
+        assert client.get("/api/cma/data-quality/rule-config",
+                          headers=auth_header(token)).status_code == 200
+
+    def test_put_requires_ceo_or_it(self, db, client):
+        create_test_user(db, username="cfgfin2", role="finance")
+        token = get_token_for_user(client, "cfgfin2")
+        r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
+                       json={"rule_key": "unit_check", "param_key": "amount_threshold",
+                             "value": 5000, "change_reason": REASON})
+        assert r.status_code == 403, r.text
+
+    def test_put_writes_row_and_audit_same_txn(self, db, client):
+        token = _ceo_token(db, client)
+        before_version = client.get("/api/cma/data-quality/rule-config",
+                                    headers=auth_header(token)).json()["config_version"]
+        r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
+                       json={"rule_key": "unit_check", "param_key": "amount_threshold",
+                             "value": 5000, "change_reason": REASON})
+        assert r.status_code == 200, r.text
+        body = r.json()
+        assert body["success"] and body["before"] == 10000 and body["after"] == 5000
+        assert body["warning"] is None
+
+        row = db.query(QualityRuleConfig).filter_by(
+            entity_id=1, rule_key="unit_check", param_key="amount_threshold").one()
+        assert json.loads(row.param_value) == 5000 and row.updated_by == "cfgceo"
+
+        log = db.query(OperationLog).filter_by(action="update_dq_rule_config").one()
+        assert log.user_id is not None                       # 清单 §5.2：user_id 必写
+        assert log.target_type == "dq_rule_config" and log.target_id == row.id
+        assert log.detail["entity_id"] == 1 and log.detail["before"] == 10000
+        assert log.detail["after"] == 5000 and log.detail["reason"] == REASON
+
+        # 生效回读：GET 显示来源=账套覆盖；配置指纹随之变化
+        g = client.get("/api/cma/data-quality/rule-config", headers=auth_header(token)).json()
+        entry = g["rules"]["unit_check"]["params"]["amount_threshold"]
+        assert entry["value"] == 5000 and entry["source"] == "entity"
+        assert g["config_version"] == body["config_version"] != before_version
+
+    def test_put_locked_unknown_and_reason_rejected(self, db, client):
+        token = _ceo_token(db, client)
+        base = {"rule_key": "unit_check", "param_key": "level", "value": "warning",
+                "change_reason": REASON}
+        r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token), json=base)
+        assert r.status_code == 400 and "锁死" in r.json()["detail"]
+
+        base["param_key"] = "nope"
+        assert client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
+                          json=base).status_code == 400
+
+        r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
+                       json={"rule_key": "unit_check", "param_key": "amount_threshold",
+                             "value": 5000, "change_reason": "太短"})
+        assert r.status_code == 400 and "change_reason" in r.json()["detail"]
+
+    def test_put_boundary_reject_and_clamp(self, db, client):
+        token = _ceo_token(db, client)
+        r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
+                       json={"rule_key": "unit_check", "param_key": "amount_threshold",
+                             "value": 5, "change_reason": REASON})
+        assert r.status_code == 400 and "越界" in r.json()["detail"]
+
+        r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
+                       json={"rule_key": "unit_check", "param_key": "deduct_cap",
+                             "value": 999, "change_reason": REASON})
+        assert r.status_code == 200 and r.json()["after"] == 50 and r.json()["warning"]
+
+    def test_platform_param_requires_it(self, db, client):
+        ceo = _ceo_token(db, client)
+        payload = {"rule_key": "unit_check", "param_key": "detail_limit", "value": 25,
+                   "change_reason": REASON}
+        r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(ceo), json=payload)
+        assert r.status_code == 403, r.text
+        assert "平台级" in r.json()["detail"]
+
+        create_test_user(db, username="cma", role="it")
+        it = get_token_for_user(client, "cma")
+        r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(it), json=payload)
+        assert r.status_code == 200 and r.json()["after"] == 25
+
+    def test_bot_channel_write_forbidden(self, db, client):
+        token = _ceo_token(db, client)
+        headers = {**auth_header(token), "X-BOT-KEY": "dummy-bot-key"}
+        r = client.put("/api/cma/data-quality/rule-config", headers=headers,
+                       json={"rule_key": "unit_check", "param_key": "amount_threshold",
+                             "value": 5000, "change_reason": REASON})
+        assert r.status_code == 403 and "Bot" in r.json()["detail"]
+
+    def test_demo_entity_write_rejected(self, db, client):
+        """口径②：演示/测试账套只读、不开配置（写 400，读仍 200）。"""
+        db.add(Entity(id=6, name="演示企业", short_name="演示", status="demo"))
+        db.commit()
+        u = create_test_user(db, username="cfgdemo", role="ceo")
+        from app.models import UserEntity
+        db.add(UserEntity(user_id=u.id, entity_id=1, granted_by=1))
+        db.add(UserEntity(user_id=u.id, entity_id=6, granted_by=1))
+        db.commit()
+
+        # 登录 entity 6（该账号已显式授权 1/6）
+        resp = client.post("/api/cma/auth/login", json={
+            "username": "cfgdemo", "password": "test-fixture-pass", "entity_id": 6})
+        assert resp.status_code == 200, resp.text
+        tok6 = resp.json()["token"]
+
+        assert client.get("/api/cma/data-quality/rule-config",
+                          headers=auth_header(tok6)).status_code == 200
+        assert client.get("/api/cma/data-quality/rule-config",
+                          headers=auth_header(tok6)).json()["entity_config_enabled"] is False
+        r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(tok6),
+                       json={"rule_key": "unit_check", "param_key": "amount_threshold",
+                             "value": 5000, "change_reason": REASON})
+        assert r.status_code == 400 and "不开放规则配置" in r.json()["detail"]
+
+    def test_history_endpoint(self, db, client):
+        token = _ceo_token(db, client)
+        client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
+                   json={"rule_key": "dup_alert", "param_key": "min_count", "value": 3,
+                         "change_reason": REASON})
+        r = client.get("/api/cma/data-quality/rule-config/history", headers=auth_header(token))
+        assert r.status_code == 200
+        items = r.json()["items"]
+        assert items and items[0]["rule_key"] == "dup_alert"
+        assert items[0]["before"] == 2 and items[0]["after"] == 3
+        assert items[0]["reason"] == REASON and items[0]["user_id"] is not None
+
+    def test_rule_detail_total_matches_governance_check(self, db, client):
+        """两处口径同源：改了阈值后 `/rule-detail` total == `governance-check` count。"""
+        _mk_plan(db, entity_id=1, id=900, amount=150.0)
+        token = _ceo_token(db, client)
+        client.put("/api/cma/data-quality/rule-config", headers=auth_header(token),
+                   json={"rule_key": "unit_check", "param_key": "amount_threshold",
+                         "value": 100, "change_reason": REASON})
+        gov = client.get("/api/cma/data-quality/governance-check",
+                         headers=auth_header(token)).json()
+        count = {i["rule"]: i["count"] for i in gov["issues"]}["unit_check"]
+        detail = client.get("/api/cma/data-quality/rule-detail?rule=unit_check",
+                            headers=auth_header(token)).json()
+        assert count == 1 and detail["total"] == count
+        assert detail["items"][0]["key_fields"]["threshold"] == ">100（万元口径可疑）"
+
+
+# ══════════════════════════════════════════════════════════════════
+# D 韧性：DDL 未落地时不 500
+# ══════════════════════════════════════════════════════════════════
+class TestTableMissingTolerance:
+    def test_read_falls_back_to_builtin_when_table_absent(self, db, client):
+        from tests import conftest as cf
+
+        token = _ceo_token(db, client)
+        _mk_plan(db, entity_id=1, id=950, amount=50000.0)
+        QualityRuleConfig.__table__.drop(bind=cf.TEST_ENGINE)
+        try:
+            assert dqcfg.table_ready(db) is False
+            r = client.get("/api/cma/data-quality/rule-config", headers=auth_header(token))
+            assert r.status_code == 200, r.text
+            assert r.json()["table_ready"] is False
+            assert r.json()["rules"]["unit_check"]["params"]["amount_threshold"]["value"] == 10000
+
+            gov = client.get("/api/cma/data-quality/governance-check",
+                             headers=auth_header(token))
+            assert gov.status_code == 200, gov.text
+            assert {i["rule"]: i["count"] for i in gov.json()["issues"]}["unit_check"] == 1
+        finally:
+            QualityRuleConfig.__table__.create(bind=cf.TEST_ENGINE)