From b5feee69821f3160ad44aa5c37792e6c4cf04d53 Mon Sep 17 00:00:00 2001 From: qiaoqichao Date: Tue, 29 Sep 2026 20:28:45 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20T07-a=20=E9=A2=84=E5=AE=A1=E7=BB=93?= =?UTF-8?q?=E6=9E=9C=E7=BA=A6=E6=9D=9F=E4=B8=8E=E7=BB=93=E8=AE=BA=E6=A0=A1?= =?UTF-8?q?=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit T07 是"预审执行"大卡，按纪律拆子卡。本批为第一块：纯确定性的结论校验， 不调用任何模型、不计算概率。覆盖 AC-05/06/07/09/12/45 与 BR-05/06/07/12。 新增 services/prescreen_policy.py： - decide()：四级优先（hard_block → critical_incomplete → open_findings → clean）， 优先级顺序即业务口径，改动等于改结论，已显式标注 - **硬性阻断不可覆盖**：命中 hard 规则即 stop，并指出规则版本、保留模型原始 结构化提案供审计；专项模型的正面单题判断不能翻盘（有专门反例） - **未知不等于风险也不等于通过**：关键核查不可完成给 insufficient_information 并列出未完成项，findings 的 result 原样保留、**不得被改写成 issue** - **没有 second path 给概率**：probability() 不接受任何入参，模型分数无法顺手传入； native_metrics 里名为 probability 的指标会被 assert_no_probability_leak 拦下， self_reported 置信度可入档但不得变概率 - stop 明确不触碰正式审批（AC-45），预算耗尽按 E-08 列未完成项而非伪造完成 实现中修正了两处自己的测试写法：断言"不含高风险三字"会被免责说明本身命中； 断言 open_findings 为空与"unknown 属待处理"的定义矛盾。两处都改成检验真正的 不变量（结论非风险判定 + findings 不被改写）。 测试：新增 test_prescreen_policy.py 19 项。全量 368 项 OK（349 → 368）； verify.py 10/10 PASS。 Co-Authored-By: Claude Code --- backend/services/prescreen_policy.py | 126 ++++++++++++++++++++++ backend/test_prescreen_policy.py | 156 +++++++++++++++++++++++++++ 2 files changed, 282 insertions(+) create mode 100644 backend/services/prescreen_policy.py create mode 100644 backend/test_prescreen_policy.py diff --git a/backend/services/prescreen_policy.py b/backend/services/prescreen_policy.py new file mode 100644 index 0000000..8b2394a --- /dev/null +++ b/backend/services/prescreen_policy.py @@ -0,0 +1,126 @@ +"""T07-a 预审结果约束：把核查结果收敛成受约束的建议。 + +依据：docs/v2/01-requirements.md §5、03-detailed-design.md §5； +BR-05/06/07/12、E-02/06/07/08、AC-05/06/07/09/12/45。 + +本模块只做**确定性结论校验**，不调用任何模型、不算概率。四条不可让渡的规则： + +1. **硬性阻断不可被覆盖**（BR-06）：已确认的 hard 规则命中即 `stop`，模型的高分 + 或"建议推进"都不能翻盘。同时**保留模型原始结构化提案**供审计，并指出规则版本。 +2. **未知不等于风险，也不等于通过**（BR-07）：关键核查不可完成时给 + `insufficient_information` 并列补件，**不得**据此评判企业高风险。 +3. **单题不能代表综合意见**（BR-05）：专项判断只进 findings 与被引用，不能直接 + 写入 recommendation。 +4. **未校准就没有概率**（BR-12）：`probability` 恒为 None + `not_calibrated`， + 任何模型自报置信度都不改写它。 +""" + +from __future__ import annotations + +RECOMMENDATIONS = ("advance", "supplement", "stop", "insufficient_information") + +NOT_CALIBRATED = "not_calibrated" +NO_PROBABILITY_REASON = "尚未完成真实历史校准；不展示审批通过概率" + +# 判定优先级（从上到下，先命中先返回）。顺序本身是业务口径，改动等于改结论。 +PRECEDENCE = ("hard_block", "critical_incomplete", "open_findings", "clean") + +INCOMPLETE_RESULTS = ("unknown", "error", "not_applicable") + + +def _codes(findings) -> set[str]: + return {str(item.get("code")) for item in findings or () if item.get("code")} + + +def hard_blocks(findings, hard_rule_codes) -> list[dict]: + """命中的硬性阻断项。`hard_rule_codes` 来自冻结方案里 effect=hard 且 on_match=block 的规则。""" + wanted = set(hard_rule_codes or ()) + return [dict(item) for item in findings or () if item.get("code") in wanted] + + +def critical_incomplete(findings, critical_check_ids) -> list[str]: + """关键核查项中未完成的那些编码（未知/执行失败/不适用）。""" + critical = set(critical_check_ids or ()) + return sorted({ + str(item.get("code")) for item in findings or () + if item.get("code") in critical and item.get("result") in INCOMPLETE_RESULTS + }) + + +def open_findings(findings) -> list[dict]: + """需要业务处理但未阻断的问题（issue / unknown）。""" + return [dict(item) for item in findings or () + if item.get("result") in ("issue", "unknown")] + + +def probability() -> dict: + """P0 恒为空。**不接受任何入参**——不给"顺手传个模型分数"留口子。""" + return {"value": None, "reason": NOT_CALIBRATED, "message": NO_PROBABILITY_REASON} + + +def decide(findings, missing_inputs=(), degradations=(), judgments=(), + critical_check_ids=(), hard_rule_codes=(), rule_versions=()) -> dict: + """收敛结论。返回 recommendation 与支撑它的理由、降级与审计信息。""" + blocks = hard_blocks(findings, hard_rule_codes) + incomplete = critical_incomplete(findings, critical_check_ids) + open_items = open_findings(findings) + + degradations = [dict(item) for item in degradations or ()] + missing = sorted({str(item) for item in missing_inputs or ()}) + + if blocks: + # 命中硬阻断：即便专项模型给出正面单题判断，也不改变结论（BR-05 / BR-06） + stage = "hard_block" + recommendation = "stop" + reason = "存在已确认的硬性阻断，暂不建议推进；" + "、".join( + f'{item.get("code")}@{item.get("basis_version", "未标注版本")}' for item in blocks) + elif incomplete: + stage = "critical_incomplete" + recommendation = "insufficient_information" + reason = ("关键核查未能完成，无法形成结论，请补充材料后重新发起；" + "不得据此判定企业高风险。未完成项：" + "、".join(incomplete)) + missing = sorted(set(missing) | set(incomplete)) + elif open_items: + stage = "open_findings" + recommendation = "supplement" + reason = f"存在 {len(open_items)} 项待处理事项，建议补充材料或说明后重新核查" + else: + stage = "clean" + recommendation = "advance" + reason = "已完成方案内的确定性核查，未发现阻断项；正式审批仍由业务流程决定" + + return { + "recommendation": recommendation, + "recommendation_reason": reason, + "stage": stage, + "findings": [dict(item) for item in findings or ()], + "open_findings": open_items, + "hard_blocks": blocks, + "critical_incomplete": incomplete, + "missing_inputs": missing, + "degradations": degradations, + # 专项判断只作为依据入档，**不参与** recommendation 的判定 + "judgments": [dict(item) for item in judgments or ()], + "judgment_can_override": False, + "rule_versions": [dict(item) for item in rule_versions or ()], + "probability": probability(), + # AC-45：stop 只是"暂不建议推进"，系统不得自动关闭项目或执行正式审批 + "formal_approval_touched": False, + } + + +def assert_no_probability_leak(result: dict) -> None: + """任何路径都不得产出未校准概率（BR-12 / AC-12）。""" + value = (result.get("probability") or {}).get("value") + if value is not None: + raise ValueError("未校准的数据集不得产出审批通过概率") + for item in result.get("judgments", ()): + for metric in item.get("native_metrics", ()) or (): + if str(metric.get("name", "")).lower() in ("probability", "approval_probability"): + raise ValueError("专项模型的原生指标不得被当作审批概率") + + +def budget_exhausted(reason: str = "RUN_BUDGET_EXCEEDED") -> dict: + """预算耗尽（E-08）：停止并列出未完成项，不伪造完成。""" + return {"code": reason, "message": "本次核查已达到执行上限，请复核未完成项", + "retryable": False} diff --git a/backend/test_prescreen_policy.py b/backend/test_prescreen_policy.py new file mode 100644 index 0000000..89c2b08 --- /dev/null +++ b/backend/test_prescreen_policy.py @@ -0,0 +1,156 @@ +"""T07-a 预审结果约束测试：AC-05/06/07/09/12/45。 + +每条规则都配反例——只测"正常给出 advance"证明不了约束存在。 +本模块纯确定性，不调用模型、不算概率。 +""" + +import unittest + +from services import prescreen_policy as policy + + +def finding(code, result, basis_version=None, **over): + item = {"code": code, "title": f"检查项 {code}", "result": result} + if basis_version: + item["basis_version"] = basis_version + item.update(over) + return item + + +class HardBlockTests(unittest.TestCase): + def test_hard_block_forces_stop(self): + result = policy.decide([finding("lease.coverage.below", "issue")], + hard_rule_codes=["lease.coverage.below"], + critical_check_ids=["lease.coverage.below"]) + self.assertEqual(result["recommendation"], "stop") + self.assertEqual(result["stage"], "hard_block") + + def test_hard_block_cites_the_rule_version(self): + result = policy.decide([finding("r.hard", "issue", basis_version="rule.credit@v3")], + hard_rule_codes=["r.hard"]) + self.assertIn("rule.credit@v3", result["recommendation_reason"]) + + def test_positive_single_judgment_cannot_override_a_hard_block(self): + # AC-05 / BR-05：单题正面判断不能覆盖综合意见 + judgments = [{"question_id": "q1", "status": "completed", + "answer": {"kind": "choice", "value": "supported"}}] + result = policy.decide([finding("r.hard", "issue")], hard_rule_codes=["r.hard"], + judgments=judgments) + self.assertEqual(result["recommendation"], "stop") + self.assertFalse(result["judgment_can_override"]) + self.assertEqual(result["judgments"], judgments, "判断仍应入档供审计") + + def test_hard_block_beats_insufficient_information(self): + result = policy.decide( + [finding("r.hard", "issue"), finding("k1", "unknown")], + hard_rule_codes=["r.hard"], critical_check_ids=["k1"]) + self.assertEqual(result["recommendation"], "stop") + + def test_block_does_not_touch_formal_approval(self): + # AC-45：stop 只表示暂不建议推进，系统不得自动关闭项目或审批 + result = policy.decide([finding("r.hard", "issue")], hard_rule_codes=["r.hard"]) + self.assertFalse(result["formal_approval_touched"]) + self.assertIn("暂不建议推进", result["recommendation_reason"]) + + def test_advisory_code_is_not_a_hard_block(self): + result = policy.decide([finding("r.advisory", "issue")], + hard_rule_codes=["r.hard"]) + self.assertNotEqual(result["recommendation"], "stop") + + +class InsufficientInformationTests(unittest.TestCase): + def test_incomplete_critical_check_never_becomes_high_risk(self): + # AC-07 / BR-07：全部关键资料缺失时给信息不足，**不判高风险** + result = policy.decide([finding("k1", "unknown"), finding("k2", "unknown")], + critical_check_ids=["k1", "k2"]) + self.assertEqual(result["recommendation"], "insufficient_information") + # 要求不是"别出现高风险三个字"（免责说明本身就会写到它），而是： + # 结论不能变成风险判定，且不得凭空生成"高风险"这类 finding + self.assertIn("不得据此判定企业高风险", result["recommendation_reason"]) + self.assertNotEqual(result["recommendation"], "stop") + # 缺证可以（也应该）出现在待处理列表里，但**不得被改写成风险问题**： + # findings 数量与 result 取值必须原样保留，不能凭空多出 issue + self.assertEqual(len(result["findings"]), 2) + self.assertTrue(all(item["result"] == "unknown" for item in result["findings"])) + self.assertFalse(any(item["result"] == "issue" for item in result["open_findings"])) + + def test_missing_inputs_include_the_incomplete_critical_checks(self): + result = policy.decide([finding("k1", "unknown")], critical_check_ids=["k1"], + missing_inputs=["现金流材料"]) + self.assertEqual(result["missing_inputs"], ["k1", "现金流材料"]) + + def test_execution_failure_counts_as_incomplete(self): + result = policy.decide([finding("k1", "error")], critical_check_ids=["k1"]) + self.assertEqual(result["recommendation"], "insufficient_information") + + def test_non_critical_unknown_does_not_force_insufficient(self): + # 非关键项的未知只要求补件，不阻断成信息不足 + result = policy.decide([finding("other", "unknown")], critical_check_ids=["k1"]) + self.assertEqual(result["recommendation"], "supplement") + + +class SupplementAndAdvanceTests(unittest.TestCase): + def test_open_issue_leads_to_supplement(self): + result = policy.decide([finding("c1", "issue")]) + self.assertEqual(result["recommendation"], "supplement") + + def test_clean_run_advances_but_does_not_approve(self): + result = policy.decide([finding("c1", "pass")]) + self.assertEqual(result["recommendation"], "advance") + self.assertIn("正式审批仍由业务流程决定", result["recommendation_reason"]) + + def test_no_findings_at_all_is_treated_as_clean(self): + result = policy.decide([]) + self.assertEqual(result["stage"], "clean") + self.assertEqual(result["open_findings"], []) + + +class DegradationTests(unittest.TestCase): + def test_degradations_are_preserved_not_swallowed(self): + result = policy.decide([finding("c1", "pass")], + degradations=[{"code": "JUDGMENT_UNAVAILABLE", + "message": "专项核查未完成"}]) + self.assertEqual(len(result["degradations"]), 1) + + def test_budget_exhaustion_reports_unfinished_not_success(self): + # E-08：耗尽即停止并列未完成，不伪造完成 + item = policy.budget_exhausted() + self.assertEqual(item["code"], "RUN_BUDGET_EXCEEDED") + self.assertFalse(item["retryable"]) + self.assertIn("未完成项", item["message"]) + + +class ProbabilityGateTests(unittest.TestCase): + def test_probability_is_always_null_with_a_reason(self): + for findings in ([], [finding("c1", "pass")], [finding("c1", "issue")]): + with self.subTest(findings=len(findings)): + result = policy.decide(findings) + self.assertIsNone(result["probability"]["value"]) + self.assertEqual(result["probability"]["reason"], "not_calibrated") + + def test_probability_takes_no_input_so_a_model_score_cannot_leak_in(self): + with self.assertRaises(TypeError): + policy.probability(0.93) + + def test_native_metric_named_probability_is_rejected(self): + # BR-12 / AC-18：厂商原生指标不得被当作审批概率 + result = policy.decide( + [finding("c1", "pass")], + judgments=[{"question_id": "q1", "status": "completed", + "native_metrics": [{"name": "probability", "value": 0.91}]}]) + with self.assertRaises(ValueError) as ctx: + policy.assert_no_probability_leak(result) + self.assertIn("不得被当作审批概率", str(ctx.exception)) + + def test_self_reported_confidence_does_not_become_probability(self): + result = policy.decide( + [finding("c1", "pass")], + judgments=[{"question_id": "q1", "status": "completed", + "native_metrics": [{"name": "confidence", "value": 0.99, + "source": "self_reported"}]}]) + policy.assert_no_probability_leak(result) # 可以入档，但不得变概率 + self.assertIsNone(result["probability"]["value"]) + + +if __name__ == "__main__": + unittest.main()