v7 mech 8: MCDT acceptance/key-fact nodes as scorecard input model only

(cherry picked from commit 93e491b6a1b83fcdbbacf3493cdd2fd76418d0b6)

cloud/mcdt_nodes.py

@@ -0,0 +1,269 @@
+"""MCDT 结果指标：验收节点与关键事实节点的评分卡输入模型。
+
+起点事实（照实写清，不得淡化）：
+- 此前 v6 批头 b7ac9d7 全仓 grep MCDT / 关键事实节点 / 验收节点为
+  0 命中：此前代码里根本没有 MCDT。
+- 本模块只是评分卡输入模型，不是产品功能。不得声称 MCDT 已是
+  产品功能、已接入生产评分、已跑通真实审计循环。
+- 本模块不改任何生产行为：不碰 #238 auto_apply（保持 False）、
+  不碰 #247“纠正为 pass 本身不解封”、不改 v1.1 契约、不接线到
+  build_scorecard / 服务器端点 / 门禁。本模块的输出只是“每维的
+  得分输入”，调用方是否采用、如何采用，须另行拍板。
+
+设计（v7 机制第 8 项，只做这一项）：
+- acceptance_node（验收节点）：一项可验收的结果主张，至少含
+  id、title、acceptance_criteria（列表）、status、evidence_refs。
+  另含 dimension：该节点映射到哪一维（7 维 id 之一）；无 dimension
+  或 dimension 非法时该节点不映射到任何维（记入未映射清单），
+  绝不靠猜测、按 id 前缀、按其它维借分。
+- key_fact_node（关键事实节点）：一项支撑/证伪验收的关键事实，
+  至少含 id、fact、status、source。另含 dimension：与验收节点同
+  规则；无 dimension 的关键事实不影响任何维。
+- status 词汇（两类节点共用）：pending / pass / fail / unknown。
+  pending = 尚未判定；unknown = 未知/未验证；fail = 已证伪或未达标；
+  pass = 已验收通过。其它任何值一律按 unknown 处理（fail-closed，
+  不把拼写错误读成通过）。
+
+映射到哪几维（写清，不留含糊）：
+- delivery_quality（交付质量）：验收节点证明交付结果达到验收标准。
+- gate_audit（门禁与审计有效性）：验收节点证明门禁真实拦截、
+  审计结论可复核。
+- arch_contract（架构与契约一致性）：验收节点证明契约冻结、
+  架构 / 目录 / 导入一致。
+- 其余四维 test_coverage、security_ledger、cost_quota、ops_scale：
+  本项不预设节点。无对应验收节点时一律记 unknown，并在
+  missing_items 里说明；不得用过程指标（测试数、覆盖率）填补，
+  不得从其它维借分。注意 test_coverage 尤其如此：覆盖率数字
+  本身是过程指标，即使调用方手头有覆盖率，也不能因此把
+  test_coverage 从 unknown 改成 pass/fail——只有映射到该维的
+  验收节点状态变动才能改分。
+
+映射规则（nodes_to_dimension_inputs）：
+(a) 某维无对应验收节点时，该维输入为 unknown（不得用过程指标填补）。
+(b) 过程指标（test_count、coverage）只作为证据字段透传，绝不直接
+    进入得分输入：它们只出现在返回值的顶层 process_evidence 里，
+    绝不出现在任何一维的 score_input / status / basis 里。只改
+    test_count / coverage 而节点状态不变时，每维的得分输入不变；
+    只改节点状态（pending -> pass 或 pass -> fail）时，映射输出变化。
+    节点 dict 上附带的 test_count / coverage 等额外键同样被忽略，
+    只读 id / title / acceptance_criteria / status / evidence_refs /
+    dimension（验收节点）与 id / fact / status / source / dimension
+    （关键事实节点）。
+(c) 节点状态为 fail 时对应维不得为 pass；节点状态 unknown 时对应维
+    为 unknown。具体聚合（fail-closed，优先级 fail > unknown >
+    pending > pass 的反向表述见代码）：
+    - 任一映射验收节点 fail -> 该维 fail；
+    - 任一映射关键事实 fail -> 该维 fail（已证伪的关键事实不得
+      让该维通过，即使验收节点全 pass）；
+    - 否则任一映射验收节点 unknown 或 pending -> 该维 unknown；
+    - 否则任一映射关键事实 unknown 或 pending -> 该维 unknown
+      （支撑事实未验证，不得宣称通过）；
+    - 否则（映射验收节点全 pass，且映射关键事实全 pass 或无映射
+      关键事实）-> 该维 pass。
+    关键事实单独存在时不能让维通过：无验收节点时该维仍 unknown。
+
+阈值口径（沿用 audit_scorecard.py）：
+- thresholds 参数只用于报告 thresholds_source，绝不改变任何一维
+  的状态：传阈值不能把 unknown 变成 pass。
+- 默认 thresholds_source = "test_default_unapproved"（测试默认值、
+  未拍板）；只有调用方显式传入且全部条目 source 均为
+  "user_approved" 时才记 "user_approved"。本模块绝不把测试默认
+  值写成生产已拍板；B / C 维与阈值本身不在本项决定范围。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+# 维顺序与 id 与 cloud/audit_scorecard.py 的 DIMENSIONS 保持一致；
+# 若那边增维、减维、改名，本模块须同步改（测试钉住两者相等）。
+DIMENSION_IDS = [
+    "delivery_quality",
+    "gate_audit",
+    "arch_contract",
+    "test_coverage",
+    "security_ledger",
+    "cost_quota",
+    "ops_scale",
+]
+
+# 本项有映射语义说明的三维（其余四维无预设节点，见模块 docstring）。
+MAPPED_DIMENSIONS_DOC = {
+    "delivery_quality": "验收节点证明交付结果达到验收标准。",
+    "gate_audit": "验收节点证明门禁真实拦截、审计结论可复核。",
+    "arch_contract": "验收节点证明契约冻结、架构 / 目录 / 导入一致。",
+}
+
+VALID_STATUSES = ("pending", "pass", "fail", "unknown")
+
+MODEL_NOTE = (
+    "MCDT input model only, not a product feature: "
+    "v6 head b7ac9d7 had 0 hits for MCDT / acceptance nodes / "
+    "key fact nodes; this module only maps node statuses to "
+    "per-dimension score inputs and does not wire into scoring, "
+    "gates, or production."
+)
+
+
+@dataclass
+class AcceptanceNode:
+    """验收节点。status 非法值在映射时按 unknown 处理。"""
+
+    id: str
+    title: str
+    acceptance_criteria: list = field(default_factory=list)
+    status: str = "unknown"
+    evidence_refs: list = field(default_factory=list)
+    dimension: str | None = None
+
+
+@dataclass
+class KeyFactNode:
+    """关键事实节点。status 非法值在映射时按 unknown 处理。"""
+
+    id: str
+    fact: str
+    status: str = "unknown"
+    source: str = ""
+    dimension: str | None = None
+
+
+def _as_dict(node):
+    """接受 dataclass 实例或 dict，一律转成 dict 只读视图。"""
+    if isinstance(node, dict):
+        return node
+    if hasattr(node, "__dataclass_fields__"):
+        return {
+            f: getattr(node, f) for f in node.__dataclass_fields__  # noqa: B905
+        }
+    raise TypeError("node must be a dict or dataclass, got %r" % type(node))
+
+
+def _norm_status(value):
+    """非法状态 -> unknown（fail-closed，不把拼写错误读成通过）。"""
+    if value in VALID_STATUSES:
+        return value
+    return "unknown"
+
+
+def _thresholds_source(thresholds):
+    """与 audit_scorecard.py 同口径：只有全 user_approved 才算已拍板。"""
+    if not thresholds:
+        return "test_default_unapproved"
+    sources = set()
+    for entry in thresholds.values():
+        if isinstance(entry, dict):
+            sources.add(entry.get("source", "test_default_unapproved"))
+        else:
+            sources.add("test_default_unapproved")
+    if sources == {"user_approved"}:
+        return "user_approved"
+    return "test_default_unapproved"
+
+
+def nodes_to_dimension_inputs(acceptance_nodes, key_fact_nodes,
+                              thresholds=None, process_metrics=None):
+    """把节点状态映射为每维的得分输入。
+
+    acceptance_nodes / key_fact_nodes：AcceptanceNode / KeyFactNode
+    实例或同字段 dict 的列表。process_metrics：过程指标 dict
+    （如 {"test_count": ..., "coverage": ...}），只透传到顶层
+    process_evidence，绝不进入任何一维的得分输入。
+    返回：{"dimensions": {维 id: 维输入}, "dimension_order": [...],
+    "process_evidence": {...}, "unmapped_acceptance_node_ids": [...],
+    "unmapped_key_fact_node_ids": [...],
+    "thresholds_source": ..., "model_note": ...}。
+    每维输入含 dimension / status / score_input / acceptance_node_ids /
+    key_fact_node_ids / missing_items / basis；score_input 里只有
+    节点状态计数，绝无 test_count / coverage 键。
+    """
+    acceptance_nodes = list(acceptance_nodes or [])
+    key_fact_nodes = list(key_fact_nodes or [])
+
+    per_dim_acc = {dim: [] for dim in DIMENSION_IDS}
+    per_dim_facts = {dim: [] for dim in DIMENSION_IDS}
+    unmapped_acc, unmapped_facts = [], []
+
+    for raw in acceptance_nodes:
+        node = _as_dict(raw)
+        dim = node.get("dimension")
+        if dim in DIMENSION_IDS:
+            per_dim_acc[dim].append(node)
+        else:
+            unmapped_acc.append(node.get("id"))
+
+    for raw in key_fact_nodes:
+        node = _as_dict(raw)
+        dim = node.get("dimension")
+        if dim in DIMENSION_IDS:
+            per_dim_facts[dim].append(node)
+        else:
+            unmapped_facts.append(node.get("id"))
+
+    dimensions = {}
+    for dim in DIMENSION_IDS:
+        acc = per_dim_acc[dim]
+        facts = per_dim_facts[dim]
+        acc_statuses = [_norm_status(n.get("status")) for n in acc]
+        fact_statuses = [_norm_status(n.get("status")) for n in facts]
+        missing, basis = [], []
+        if not acc:
+            missing.append("no_acceptance_node_mapped")
+            if dim not in MAPPED_DIMENSIONS_DOC:
+                missing.append(
+                    "no_preset_node_for_dimension:"
+                    " process metrics must not fill this dimension")
+            basis.append("no mapped acceptance node: dimension unknown")
+            status = "unknown"
+        elif "fail" in acc_statuses:
+            status = "fail"
+            basis.append("mapped acceptance node fail: dimension not pass")
+        elif "fail" in fact_statuses:
+            status = "fail"
+            basis.append("mapped key fact fail: dimension not pass")
+        elif "unknown" in acc_statuses or "pending" in acc_statuses:
+            status = "unknown"
+            basis.append(
+                "mapped acceptance node not yet pass"
+                " (unknown/pending): dimension unknown")
+        elif "unknown" in fact_statuses or "pending" in fact_statuses:
+            status = "unknown"
+            basis.append(
+                "mapped key fact not yet verified"
+                " (unknown/pending): dimension unknown")
+        else:
+            status = "pass"
+            basis.append(
+                "all mapped acceptance nodes pass and mapped key facts"
+                " pass or absent")
+        score_input = {
+            "node_status": status,
+            "acceptance_pass_count": acc_statuses.count("pass"),
+            "acceptance_fail_count": acc_statuses.count("fail"),
+            "acceptance_pending_count": acc_statuses.count("pending"),
+            "acceptance_unknown_count": acc_statuses.count("unknown"),
+            "key_fact_pass_count": fact_statuses.count("pass"),
+            "key_fact_fail_count": fact_statuses.count("fail"),
+            "key_fact_pending_count": fact_statuses.count("pending"),
+            "key_fact_unknown_count": fact_statuses.count("unknown"),
+        }
+        dimensions[dim] = {
+            "dimension": dim,
+            "status": status,
+            "score_input": score_input,
+            "acceptance_node_ids": [n.get("id") for n in acc],
+            "key_fact_node_ids": [n.get("id") for n in facts],
+            "missing_items": missing,
+            "basis": basis,
+        }
+
+    return {
+        "dimensions": dimensions,
+        "dimension_order": list(DIMENSION_IDS),
+        # 过程指标只在这里透传：顶层证据字段，不是得分输入。
+        "process_evidence": dict(process_metrics) if process_metrics else {},
+        "unmapped_acceptance_node_ids": unmapped_acc,
+        "unmapped_key_fact_node_ids": unmapped_facts,
+        "thresholds_source": _thresholds_source(thresholds),
+        "model_note": MODEL_NOTE,
+    }

cloud/tests/test_scorecard_mech_mcdt.py

@@ -0,0 +1,235 @@
+"""v7 机制第 8 项：MCDT 结果指标（评分卡输入模型）定向测试。
+
+钉住的规则（对应任务 (a)(b)(c)）：
+(a) 某维无对应验收节点时该维输入为 unknown，不得用过程指标填补。
+(b) 过程指标（test_count、coverage）只作为证据字段透传，绝不直接
+    进入得分输入：只改 test_count/coverage 而节点状态不变时，每维
+    的得分输入不变；只改节点状态（pending->pass 或 pass->fail）时，
+    映射输出变化。
+(c) 节点状态为 fail 时对应维不得为 pass；节点状态 unknown 时对应维
+    为 unknown。
+
+诚实边界：此前 v6 头全仓 grep MCDT / 关键事实节点 / 验收节点为
+0 命中，本项只是评分卡输入模型，不是产品功能；测试不声称 MCDT
+已接入生产评分、门禁或真实审计循环。阈值一律仍是
+test_default_unapproved，除非调用方显式传入全 user_approved。
+"""
+
+import sys
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.audit_scorecard import DIMENSION_IDS as SCORECARD_DIMS
+    from cloud.mcdt_nodes import (
+        AcceptanceNode,
+        KeyFactNode,
+        DIMENSION_IDS,
+        nodes_to_dimension_inputs,
+    )
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.audit_scorecard import DIMENSION_IDS as SCORECARD_DIMS
+    from cloud.mcdt_nodes import (
+        AcceptanceNode,
+        KeyFactNode,
+        DIMENSION_IDS,
+        nodes_to_dimension_inputs,
+    )
+
+
+def _acc(node_id, dim, status, **extra):
+    node = {
+        "id": node_id,
+        "title": "t-%s" % node_id,
+        "acceptance_criteria": ["criterion 1"],
+        "status": status,
+        "evidence_refs": ["ev-1"],
+        "dimension": dim,
+    }
+    node.update(extra)
+    return node
+
+
+def _fact(node_id, dim, status):
+    return {
+        "id": node_id,
+        "fact": "fact-%s" % node_id,
+        "status": status,
+        "source": "src-1",
+        "dimension": dim,
+    }
+
+
+def _walk(obj):
+    if isinstance(obj, dict):
+        for key, value in obj.items():
+            yield key
+            yield from _walk(value)
+    elif isinstance(obj, (list, tuple)):
+        for value in obj:
+            yield from _walk(value)
+
+
+class McdtNodesTestCase(unittest.TestCase):
+    def test_dimension_ids_match_scorecard(self):
+        self.assertEqual(DIMENSION_IDS, SCORECARD_DIMS)
+        self.assertEqual(len(DIMENSION_IDS), 7)
+
+    def test_no_nodes_all_dimensions_unknown(self):
+        out = nodes_to_dimension_inputs([], [])
+        for dim in DIMENSION_IDS:
+            entry = out["dimensions"][dim]
+            self.assertEqual(entry["status"], "unknown")
+            self.assertIn("no_acceptance_node_mapped",
+                          entry["missing_items"])
+            self.assertEqual(entry["score_input"]["node_status"], "unknown")
+
+    def test_only_one_dimension_mapped_others_unknown(self):
+        out = nodes_to_dimension_inputs(
+            [_acc("a1", "delivery_quality", "pass")], [])
+        self.assertEqual(out["dimensions"]["delivery_quality"]["status"],
+                         "pass")
+        for dim in DIMENSION_IDS:
+            if dim != "delivery_quality":
+                self.assertEqual(out["dimensions"][dim]["status"], "unknown")
+
+    def test_process_metrics_do_not_change_dimension_inputs(self):
+        nodes = [_acc("a1", "delivery_quality", "pass"),
+                 _acc("a2", "gate_audit", "pending")]
+        base = nodes_to_dimension_inputs(
+            nodes, [], process_metrics={"test_count": 100,
+                                        "coverage": 0.5})
+        changed = nodes_to_dimension_inputs(
+            nodes, [], process_metrics={"test_count": 10000,
+                                        "coverage": 0.99})
+        self.assertEqual(base["dimensions"], changed["dimensions"])
+        # 透传只在顶层，不在任何一维里。
+        self.assertEqual(changed["process_evidence"]["test_count"], 10000)
+        self.assertEqual(changed["process_evidence"]["coverage"], 0.99)
+        for entry in changed["dimensions"].values():
+            keys = set(_walk(entry))
+            self.assertNotIn("test_count", keys)
+            self.assertNotIn("coverage", keys)
+
+    def test_node_carried_process_keys_are_ignored(self):
+        plain = nodes_to_dimension_inputs(
+            [_acc("a1", "delivery_quality", "pass")], [])
+        noisy = nodes_to_dimension_inputs(
+            [_acc("a1", "delivery_quality", "pass",
+                  test_count=99999, coverage=0.01)], [])
+        self.assertEqual(plain["dimensions"], noisy["dimensions"])
+
+    def test_pending_to_pass_changes_output(self):
+        before = nodes_to_dimension_inputs(
+            [_acc("a1", "delivery_quality", "pending")], [])
+        after = nodes_to_dimension_inputs(
+            [_acc("a1", "delivery_quality", "pass")], [])
+        self.assertEqual(
+            before["dimensions"]["delivery_quality"]["status"], "unknown")
+        self.assertEqual(
+            after["dimensions"]["delivery_quality"]["status"], "pass")
+        self.assertNotEqual(before["dimensions"], after["dimensions"])
+
+    def test_pass_to_fail_changes_output_and_is_not_pass(self):
+        before = nodes_to_dimension_inputs(
+            [_acc("a1", "arch_contract", "pass")], [])
+        after = nodes_to_dimension_inputs(
+            [_acc("a1", "arch_contract", "fail")], [])
+        self.assertEqual(before["dimensions"]["arch_contract"]["status"],
+                         "pass")
+        self.assertEqual(after["dimensions"]["arch_contract"]["status"],
+                         "fail")
+        self.assertNotEqual(before["dimensions"], after["dimensions"])
+
+    def test_fail_node_blocks_pass_even_with_other_pass_nodes(self):
+        out = nodes_to_dimension_inputs(
+            [_acc("a1", "gate_audit", "pass"),
+             _acc("a2", "gate_audit", "fail")], [])
+        self.assertEqual(out["dimensions"]["gate_audit"]["status"], "fail")
+
+    def test_unknown_node_means_dimension_unknown(self):
+        out = nodes_to_dimension_inputs(
+            [_acc("a1", "delivery_quality", "unknown")], [])
+        self.assertEqual(out["dimensions"]["delivery_quality"]["status"],
+                         "unknown")
+        out2 = nodes_to_dimension_inputs(
+            [_acc("a1", "delivery_quality", "pass"),
+             _acc("a2", "delivery_quality", "unknown")], [])
+        self.assertEqual(out2["dimensions"]["delivery_quality"]["status"],
+                         "unknown")
+
+    def test_key_fact_fail_blocks_pass(self):
+        out = nodes_to_dimension_inputs(
+            [_acc("a1", "delivery_quality", "pass")],
+            [_fact("f1", "delivery_quality", "fail")])
+        self.assertEqual(out["dimensions"]["delivery_quality"]["status"],
+                         "fail")
+
+    def test_key_fact_unknown_blocks_pass(self):
+        out = nodes_to_dimension_inputs(
+            [_acc("a1", "delivery_quality", "pass")],
+            [_fact("f1", "delivery_quality", "unknown")])
+        self.assertEqual(out["dimensions"]["delivery_quality"]["status"],
+                         "unknown")
+
+    def test_key_fact_alone_cannot_pass_dimension(self):
+        out = nodes_to_dimension_inputs(
+            [], [_fact("f1", "delivery_quality", "pass")])
+        self.assertEqual(out["dimensions"]["delivery_quality"]["status"],
+                         "unknown")
+
+    def test_unmapped_nodes_do_not_affect_any_dimension(self):
+        out = nodes_to_dimension_inputs(
+            [_acc("a1", None, "pass")],
+            [_fact("f1", "bogus_dim", "pass")])
+        for dim in DIMENSION_IDS:
+            self.assertEqual(out["dimensions"][dim]["status"], "unknown")
+        self.assertIn("a1", out["unmapped_acceptance_node_ids"])
+        self.assertIn("f1", out["unmapped_key_fact_node_ids"])
+
+    def test_invalid_status_is_unknown_not_pass(self):
+        out = nodes_to_dimension_inputs(
+            [_acc("a1", "delivery_quality", "PASS")], [])
+        self.assertEqual(out["dimensions"]["delivery_quality"]["status"],
+                         "unknown")
+
+    def test_dataclass_nodes_work(self):
+        out = nodes_to_dimension_inputs(
+            [AcceptanceNode(id="a1", title="t", acceptance_criteria=["c"],
+                            status="pass", evidence_refs=["e"],
+                            dimension="arch_contract")],
+            [KeyFactNode(id="f1", fact="f", status="pass", source="s",
+                         dimension="arch_contract")])
+        self.assertEqual(out["dimensions"]["arch_contract"]["status"],
+                         "pass")
+
+    def test_thresholds_default_unapproved_and_do_not_change_status(self):
+        nodes = [_acc("a1", "delivery_quality", "pass")]
+        default = nodes_to_dimension_inputs(nodes, [])
+        self.assertEqual(default["thresholds_source"],
+                         "test_default_unapproved")
+        approved = nodes_to_dimension_inputs(
+            nodes, [],
+            thresholds={"x": {"value": 1, "source": "user_approved"}})
+        self.assertEqual(approved["thresholds_source"], "user_approved")
+        self.assertEqual(default["dimensions"], approved["dimensions"])
+        # 未知维传了已拍板阈值也不能变 pass。
+        unknown_approved = nodes_to_dimension_inputs(
+            [], [],
+            thresholds={"x": {"value": 1, "source": "user_approved"}})
+        self.assertEqual(
+            unknown_approved["dimensions"]["delivery_quality"]["status"],
+            "unknown")
+
+    def test_test_coverage_not_filled_by_coverage_metric(self):
+        out = nodes_to_dimension_inputs(
+            [], [], process_metrics={"test_count": 5000, "coverage": 0.99})
+        self.assertEqual(out["dimensions"]["test_coverage"]["status"],
+                         "unknown")
+        self.assertIn("no_acceptance_node_mapped",
+                      out["dimensions"]["test_coverage"]["missing_items"])
+
+
+if __name__ == "__main__":
+    unittest.main()