v7 mech-7: scorecard user_side_items + three-table rule (code unchanged, production forced not_pass on user-side pending/unknown/invalid)

(cherry picked from commit 59b4c49a896972501aeeb8472a9fc2918ec226fa)
This commit is contained in:
host committed 2026-11-03 02:49:27 +00:00
1 parent f8a6a9d258
commit f406398ce2
2 files changed
+495 -8

No files matched your search

+231 -8
View File
@@ -59,11 +59,13 @@
- 文档与技能并入架构与契约一致性
(技能文本、附录溯源、移植规则最终都是契约一致性的一部分)。
输入格式:build_scorecard(evidence, thresholds=None, rubric_version=None)
输入格式:build_scorecard(evidence, thresholds=None, rubric_version=None,
baseline_head=None, evidence_rows=None, user_side_items=None,
product_evidence=None)
rubric_version 只是卡片上的版本锚点(版本化持久存储见
cloud/scorecard_rubric.py:标尺改动写新版本行、旧行只读,
本函数不查库、不改旧分)。
evidence: dict,键为上面 7 个维 id,值为该维的证据 dict。
输入格式:build_scorecard(evidence, thresholds=None, user_side_items=None) evidence: dict,键为上面 7 个维 id,值为该维的证据 dict。
每个维的证据字段见各 _score_* 函数的 docstring。
调用方必须带来源、时间、检查名;本模块不自动从测试文件、
摘要文档、PR 描述里拼凑分数。
@@ -76,8 +78,26 @@
产品轮证据时恒 not_pass;这只是评分卡输入结构拆分,
不代表任何产品功能已落地。build_scorecard 的返回 dict
同时带 code/production/product 三个子对象,GET /audit-scorecard
端点返回的示例卡因此也带这三个子对象(端点已接线)。
"""
端点返回的示例卡因此也带这三个子对象(端点已接线)。v7 机制第 7 项 user_side_items(本模块内实现,不依赖第 2 项的三张表函数):
- evidence 可带可选键 "user_side_items",或由 build_scorecard 的
user_side_items 关键字参数显式传入(关键字参数优先于 evidence 键)。
每项至少含 id、status(pending/done/blocked)、blocked_by
(pending/blocked 时必填,例如 user/host_env/contract_decision)。
- 非法项绝不静默丢弃、绝不编造状态:记入
user_side_items["invalid_items"](含下标与原因),不进 items。
- 输出增加 user_side_items 字段:原样透传有效项(多余字段保留),
并附汇总 pending_ids、done_ids、blocked_ids、blocked_by_summary。
- 三表规则见 apply_user_side_three_table_rule():代码表结论
(code_overall)绝不因用户侧 pending 被拉低;用户侧有未完成项
(pending/blocked)或非法项时,生产表结论强制为 not_pass,
原因字段明确指向用户侧(如 "user_side_pending: B1,B2,...");
user_side_items 为空/缺失时,输出 user_side_unknown=True,
生产表原因写 user_side_unknown,绝不把“无用户侧项”写成
“用户侧已完成”。
- 本项不做 B/C 拍板、不改阈值:阈值仍为 test_default_unapproved;
B/C 清单只是 id 与标题常量(见 USER_SIDE_ITEM_IDS),未知 id
不因此判非法(清单可能扩展),非法只按缺字段/未知状态/缺
blocked_by/重复 id 判定。"""
from __future__ import annotations
@@ -124,6 +144,190 @@ DEFAULT_THRESHOLDS = {
UNTESTED_KINDS = {"fixture", "viewport_substitute", "code_review_only",
"fixture_simulated", "dry_run", "estimated"}
# v7 审计 B/C 用户侧清单(id 与标题;仅作常量与测试证据,不作拍板)。
USER_SIDE_ITEM_IDS = [
"B1", "B2", "B3", "B4", "B5", "B6", "B7", "B8", "B9",
"C1", "C2", "C3", "C4",
]
USER_SIDE_ITEM_TITLES = {
"B1": "分支保护",
"B2": "合并部署窗口",
"B3": "阈值拍板",
"B4": "金任务人工审定",
"B5": "真实 PG 环境",
"B6": "真实 App 安装",
"B7": "第二台主机与账单",
"B8": "v1.1 确认",
"B9": "回滚演练窗口",
"C1": "证据绑定强制化与签名",
"C2": "live 验证接线",
"C3": "锚点策略",
"C4": "actual-only 语义",
}
VALID_USER_SIDE_STATUSES = {"pending", "done", "blocked"}
def normalize_user_side_items(raw):
"""归一化用户侧项:有效项原样透传,非法项记入 invalid_items。
返回 dict,键:items、pending_ids、done_ids、blocked_ids、
unfinished_ids、blocked_by_summary、invalid_items、count、
all_pending、all_done、user_side_unknown。
user_side_unknown 仅在 raw 为 None 或空列表时为 True;
非列表、非法项等情形不算 unknown,而是 invalid(生产表
同样不能因此通过,原因指向 invalid,见三表规则)。
本函数不修改调用方传入的原始项(逐项浅拷贝)。
"""
summary = {
"items": [],
"pending_ids": [],
"done_ids": [],
"blocked_ids": [],
"unfinished_ids": [],
"blocked_by_summary": {},
"invalid_items": [],
"count": 0,
"all_pending": False,
"all_done": False,
"user_side_unknown": False,
}
if raw is None:
summary["user_side_unknown"] = True
return summary
if isinstance(raw, (str, bytes)) or not isinstance(raw, (list, tuple)):
summary["invalid_items"].append({
"index": None,
"reason": "user_side_items_not_a_list",
"raw_type": type(raw).__name__,
})
return summary
if len(raw) == 0:
summary["user_side_unknown"] = True
return summary
seen_ids = set()
for index, item in enumerate(raw):
if not isinstance(item, dict):
summary["invalid_items"].append({
"index": index,
"reason": "item_not_a_dict",
"raw_type": type(item).__name__,
})
continue
item_id = item.get("id")
if not isinstance(item_id, str) or not item_id:
summary["invalid_items"].append({
"index": index,
"reason": "id_missing",
})
continue
if item_id in seen_ids:
summary["invalid_items"].append({
"index": index,
"id": item_id,
"reason": "duplicate_id",
})
continue
status = item.get("status")
if status not in VALID_USER_SIDE_STATUSES:
summary["invalid_items"].append({
"index": index,
"id": item_id,
"reason": "status_missing_or_unknown",
"status": status,
})
continue
blocked_by = item.get("blocked_by")
if status in ("pending", "blocked"):
if not isinstance(blocked_by, str) or not blocked_by:
summary["invalid_items"].append({
"index": index,
"id": item_id,
"reason": "blocked_by_missing",
})
continue
seen_ids.add(item_id)
copied = dict(item)
summary["items"].append(copied)
if status == "pending":
summary["pending_ids"].append(item_id)
summary["unfinished_ids"].append(item_id)
elif status == "blocked":
summary["blocked_ids"].append(item_id)
summary["unfinished_ids"].append(item_id)
else:
summary["done_ids"].append(item_id)
if isinstance(blocked_by, str) and blocked_by:
summary["blocked_by_summary"].setdefault(blocked_by, []).append(item_id)
summary["count"] = len(summary["items"])
if summary["items"]:
summary["all_pending"] = all(
it["status"] == "pending" for it in summary["items"])
summary["all_done"] = all(
it["status"] == "done" for it in summary["items"])
return summary
def _extract_overall(value):
"""从字符串或评分卡 dict 里取出 overall(字符串原样返回)。"""
if isinstance(value, dict):
for key in ("overall", "code_overall", "production_overall"):
if isinstance(value.get(key), str):
return value[key]
return None
if isinstance(value, str):
return value
return None
def apply_user_side_three_table_rule(code_overall, production_overall,
user_side_items):
"""三表规则(v7 机制第 7 项;位置:本模块)。
输入:代码表结论、生产表结论(各为 "pass"/"not_pass" 字符串,
或含 overall 键的评分卡 dict)、user_side_items 原始列表。
输出:代码表结论原样不动;生产表结论在用户侧未完成/非法/
未知时强制为 not_pass,原因明确指向用户侧;用户侧全部 done
且无非法项时,生产表结论跟随其自身输入结论,不被本规则改写。
本函数是纯函数,不读库、不改阈值、不做 B/C 拍板。
"""
normalized = normalize_user_side_items(user_side_items)
code = _extract_overall(code_overall)
production = _extract_overall(production_overall)
forced = False
reason = None
if normalized["user_side_unknown"]:
production = "not_pass"
reason = ("user_side_unknown: no user_side_items supplied; "
"not equivalent to user_side_done")
forced = True
elif normalized["invalid_items"]:
production = "not_pass"
reasons = "; ".join(
"%s(%s)" % (inv.get("reason", "invalid"),
inv.get("id", inv.get("index")))
for inv in normalized["invalid_items"])
reason = "user_side_invalid: %s" % reasons
if normalized["unfinished_ids"]:
reason += "; user_side_pending: %s" % ",".join(
normalized["unfinished_ids"])
forced = True
elif normalized["unfinished_ids"]:
production = "not_pass"
reason = "user_side_pending: %s" % ",".join(
normalized["unfinished_ids"])
forced = True
return {
"code_overall": code,
"production_overall": production,
"production_reason": reason,
"user_side_forced_production_not_pass": forced,
"code_unchanged_by_user_side": True,
"user_side_unknown": normalized["user_side_unknown"],
"user_side_items": normalized,
}
def thresholds_source(thresholds):
"""阈值来源:只有调用方显式传入 user_approved 才算已拍板。"""
@@ -690,18 +894,18 @@ def _build_table(dim_ids, evidence, th):
# 钉死:平均分绝不能作为达标依据(各表各自独立判断)。
overall_pass = all(d["score"] is not None and d["score"] >= PASS_SCORE
for d in dimensions)
overall = "pass" if overall_pass else "not_pass"
return {
"dimensions": dimensions,
"average_score": average,
"average_is_not_pass_criterion": True,
"overall": "pass" if overall_pass else "not_pass",
"overall": overall,
"overall_pass": overall_pass,
"pass_score": PASS_SCORE,
"missing_items": missing_all,
"untested_items": untested_all,
"thresholds_source": thresholds_source(th),
"thresholds": _effective_thresholds(th),
}
"thresholds": _effective_thresholds(th), }
def check_baseline_stale(scorecard, current_main_head):
@@ -881,10 +1085,29 @@ def build_scorecard(evidence, thresholds=None, rubric_version=None,
report["evidence_row_ids"] = ledger["row_ids"]
report["thresholds"] = _effective_thresholds(th)
report["code"] = build_code_scorecard(evidence, thresholds)
raw_user_side = (user_side_items if user_side_items is not None
else evidence.get("user_side_items"))
report["production"] = build_production_scorecard(
evidence, thresholds, user_side_items)
evidence, thresholds, raw_user_side)
report["product"] = build_product_scorecard(
evidence, thresholds, product_evidence)
# v7 机制第 7 项三表规则(合一接线):代码表结论绝不因用户侧
# pending 被拉低;用户侧未完成/非法/未知时顶层
# production_overall 强制 not_pass 且原因指向用户侧。注意:
# 顶层 production_overall 是规则结论,report["production"]
# 子对象仍是第 2 项的证据表(维分数照常展示),两者分工不同,
# 顶层字段不回写子对象 overall,以免第 2 项的证据表语义被改。
rule = apply_user_side_three_table_rule(
report["code"]["overall"], report["production"]["overall"],
raw_user_side)
report["user_side_items"] = rule["user_side_items"]
report["code_overall"] = rule["code_overall"]
report["production_overall"] = rule["production_overall"]
report["production_reason"] = rule["production_reason"]
report["user_side_unknown"] = rule["user_side_unknown"]
report["code_unchanged_by_user_side"] = True
report["user_side_forced_production_not_pass"] = (
rule["user_side_forced_production_not_pass"])
return report
+264
View File
@@ -0,0 +1,264 @@
"""v7 机制第 7 项 user_side_items 定向测试。
钉住的验收(只测本项,不碰其它 7 项):
- B/C 全 pending 时:7 维 overall 与代码表仍可 pass,
生产表强制 not_pass,原因明确指向用户侧。
- user_side_items 为空/缺失时:标 user_side_unknown=True,
绝不写成用户侧已完成。
- 非法项不静默丢弃、不编造状态:记入 invalid_items。
- 阈值仍为 test_default_unapproved;本项不做 B/C 拍板。
"""
import sys
import unittest
from pathlib import Path
try:
from cloud.audit_scorecard import (
DEFAULT_THRESHOLDS, USER_SIDE_ITEM_IDS,
apply_user_side_three_table_rule, build_scorecard,
normalize_user_side_items)
except ImportError:
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.audit_scorecard import (
DEFAULT_THRESHOLDS, USER_SIDE_ITEM_IDS,
apply_user_side_three_table_rule, build_scorecard,
normalize_user_side_items)
def _full_evidence():
return {
"delivery_quality": {
"first_pass_rate": 0.96, "defect_rate": 0.01,
"rework_rate": 0.04, "first_pass_known": True,
"defect_known": True, "rework_known": True,
"attempts_present": True, "evidence_kind": "real",
},
"gate_audit": {
"hard_gate_enforced": True, "gate_items_passed": 5,
"gate_items_total": 5, "merge_blocked_verified": True,
"loop_end_to_end_verified": True,
"sampling_misjudgment_rate": 0.0,
},
"arch_contract": {
"contract_frozen": True, "arch_violations": 0,
"directory_violations": 0, "import_violations": 0,
"stack_changed_without_approval": False,
},
"test_coverage": {
"prd_coverage_rate": 1.0, "cross_matrix_coverage_rate": 1.0,
"coverage_pct": 0.92, "cross_matrix_kind": "real",
"real_pg": True, "real_service": True,
"agent_six_complete": True,
"lint_items": 0, "mypy_items": 0, "bandit_high": 0,
},
"security_ledger": {
"ledger_verify_ok": True, "ledger_coverage_rate": 1.0,
"branch_protection": "protected",
"shell_whitelist_pass_rate": 1.0, "anchor_check_recent": True,
},
"cost_quota": {
"cost_records_complete": True, "unpriced_as_zero": False,
"per_creator_present": True, "quota_cutoff_verified": True,
},
"ops_scale": {
"ci_green_rate": 1.0, "deploy_success_rate": 1.0,
"rollback_drill_done": True, "ha_status": "second_host",
"parallel_scale_proven": 40,
},
}
# v7 审计 B/C 清单全 pending 证据(blocked_by 按阻塞方示例分三组)。
def _all_bc_pending():
blocked_by = {
"B1": "user", "B2": "user", "B3": "contract_decision",
"B4": "user", "B5": "host_env", "B6": "host_env",
"B7": "user", "B8": "contract_decision", "B9": "user",
"C1": "contract_decision", "C2": "host_env",
"C3": "contract_decision", "C4": "contract_decision",
}
return [{"id": item_id, "status": "pending",
"blocked_by": blocked_by[item_id]}
for item_id in USER_SIDE_ITEM_IDS]
class UserSideItemsTestCase(unittest.TestCase):
def test_bc_list_is_the_v7_audit_set(self):
self.assertEqual(USER_SIDE_ITEM_IDS,
["B1", "B2", "B3", "B4", "B5", "B6", "B7", "B8",
"B9", "C1", "C2", "C3", "C4"])
self.assertEqual(len(_all_bc_pending()), 13)
def test_bc_all_pending_code_passes_production_not_pass(self):
ev = _full_evidence()
ev["user_side_items"] = _all_bc_pending()
report = build_scorecard(ev)
# 验收核心:代码表仍可 pass,生产表 not_pass 且原因指向用户侧。
self.assertEqual(report["overall"], "pass")
self.assertEqual(report["code_overall"], "pass")
self.assertTrue(report["code_unchanged_by_user_side"])
self.assertEqual(report["production_overall"], "not_pass")
self.assertEqual(
report["production_reason"],
"user_side_pending: " + ",".join(USER_SIDE_ITEM_IDS))
self.assertTrue(report["user_side_forced_production_not_pass"])
us = report["user_side_items"]
self.assertEqual(us["pending_ids"], USER_SIDE_ITEM_IDS)
self.assertEqual(us["done_ids"], [])
self.assertEqual(us["blocked_ids"], [])
self.assertTrue(us["all_pending"])
self.assertFalse(us["user_side_unknown"])
self.assertEqual(us["invalid_items"], [])
self.assertEqual(us["count"], 13)
# blocked_by 汇总:三组阻塞方覆盖全部 13 个 id,不多不少。
grouped = sorted(i for ids in us["blocked_by_summary"].values()
for i in ids)
self.assertEqual(grouped, USER_SIDE_ITEM_IDS)
self.assertEqual(us["blocked_by_summary"]["user"],
["B1", "B2", "B4", "B7", "B9"])
def test_kwarg_user_side_items_overrides_evidence_key(self):
report = build_scorecard(_full_evidence(),
user_side_items=_all_bc_pending())
self.assertEqual(report["code_overall"], "pass")
self.assertEqual(report["production_overall"], "not_pass")
self.assertEqual(report["user_side_items"]["count"], 13)
def test_passthrough_preserves_extra_fields_verbatim(self):
items = [{"id": "B1", "status": "pending", "blocked_by": "user",
"note": "branch protection not yet enabled"},
{"id": "B3", "status": "done", "blocked_by": None}]
report = build_scorecard(_full_evidence(), user_side_items=items)
self.assertEqual(report["user_side_items"]["items"], items)
self.assertEqual(report["user_side_items"]["pending_ids"], ["B1"])
self.assertEqual(report["user_side_items"]["done_ids"], ["B3"])
def test_empty_list_is_unknown_not_done(self):
report = build_scorecard(_full_evidence(), user_side_items=[])
self.assertTrue(report["user_side_unknown"])
self.assertTrue(report["user_side_items"]["user_side_unknown"])
self.assertEqual(report["code_overall"], "pass")
self.assertEqual(report["production_overall"], "not_pass")
self.assertIn("user_side_unknown", report["production_reason"])
# 否定表述本身含 user_side_done 一词,故断言否定语义而非子串缺席。
self.assertTrue(report["production_reason"].startswith(
"user_side_unknown"))
self.assertIn("not equivalent to user_side_done",
report["production_reason"])
self.assertEqual(report["user_side_items"]["items"], [])
self.assertEqual(report["user_side_items"]["pending_ids"], [])
def test_missing_key_is_unknown_not_done(self):
report = build_scorecard(_full_evidence())
self.assertTrue(report["user_side_unknown"])
self.assertEqual(report["production_overall"], "not_pass")
self.assertIn("user_side_unknown", report["production_reason"])
self.assertNotIn("done", report["production_reason"].split(":")[0])
def test_invalid_items_are_recorded_not_silently_dropped(self):
items = [
{"id": "B1", "status": "pending"}, # 缺 blocked_by
{"status": "pending", "blocked_by": "user"}, # 缺 id
{"id": "B2", "status": "whatever", "blocked_by": "user"},
{"id": "B3", "status": "pending", "blocked_by": "user"},
{"id": "B3", "status": "pending", "blocked_by": "user"},
"not-a-dict",
]
norm = normalize_user_side_items(items)
self.assertEqual(norm["items"],
[{"id": "B3", "status": "pending",
"blocked_by": "user"}])
reasons = [inv["reason"] for inv in norm["invalid_items"]]
self.assertEqual(reasons,
["blocked_by_missing", "id_missing",
"status_missing_or_unknown", "duplicate_id",
"item_not_a_dict"])
self.assertEqual(len(norm["invalid_items"]), 5)
# 非法项不编造状态:B1/B2 绝不在 items/pending_ids 里。
self.assertNotIn("B1", norm["pending_ids"])
self.assertNotIn("B2", norm["pending_ids"])
report = build_scorecard(_full_evidence(), user_side_items=items)
self.assertEqual(report["code_overall"], "pass")
self.assertEqual(report["production_overall"], "not_pass")
self.assertIn("user_side_invalid", report["production_reason"])
self.assertIn("user_side_pending: B3", report["production_reason"])
def test_non_list_user_side_is_invalid_not_crash(self):
norm = normalize_user_side_items("B1")
self.assertEqual(norm["items"], [])
self.assertEqual(norm["invalid_items"][0]["reason"],
"user_side_items_not_a_list")
self.assertFalse(norm["user_side_unknown"])
report = build_scorecard(_full_evidence(), user_side_items="B1")
self.assertEqual(report["production_overall"], "not_pass")
self.assertIn("user_side_invalid", report["production_reason"])
def test_mixed_pending_and_done_keeps_code_pass(self):
items = [{"id": "B1", "status": "done", "blocked_by": None},
{"id": "B2", "status": "pending", "blocked_by": "user"},
{"id": "C1", "status": "blocked",
"blocked_by": "contract_decision"}]
report = build_scorecard(_full_evidence(), user_side_items=items)
self.assertEqual(report["code_overall"], "pass")
self.assertEqual(report["production_overall"], "not_pass")
# 原因只列未完成项,不把已 done 的 B1 写进 pending。
self.assertEqual(report["production_reason"],
"user_side_pending: B2,C1")
self.assertEqual(report["user_side_items"]["blocked_ids"], ["C1"])
self.assertFalse(report["user_side_items"]["all_pending"])
self.assertFalse(report["user_side_items"]["all_done"])
def test_all_done_production_follows_its_own_overall(self):
items = [{"id": item_id, "status": "done", "blocked_by": None}
for item_id in USER_SIDE_ITEM_IDS]
report = build_scorecard(_full_evidence(), user_side_items=items)
self.assertTrue(report["user_side_items"]["all_done"])
self.assertEqual(report["code_overall"], "pass")
self.assertEqual(report["production_overall"], "pass")
self.assertIsNone(report["production_reason"])
self.assertFalse(report["user_side_forced_production_not_pass"])
def test_standalone_rule_never_changes_code_overall(self):
rule = apply_user_side_three_table_rule(
"pass", "pass", _all_bc_pending())
self.assertEqual(rule["code_overall"], "pass")
self.assertEqual(rule["production_overall"], "not_pass")
self.assertEqual(rule["production_reason"],
"user_side_pending: " + ",".join(USER_SIDE_ITEM_IDS))
# 代码表本身不达标时,用户侧全 done 也不把它抬高。
rule2 = apply_user_side_three_table_rule(
"not_pass", "not_pass",
[{"id": "B1", "status": "done", "blocked_by": None}])
self.assertEqual(rule2["code_overall"], "not_pass")
self.assertEqual(rule2["production_overall"], "not_pass")
self.assertFalse(rule2["user_side_forced_production_not_pass"])
# dict 输入同样只取 overall,不改代码表结论。
rule3 = apply_user_side_three_table_rule(
{"overall": "pass"}, {"overall": "pass"}, [])
self.assertEqual(rule3["code_overall"], "pass")
self.assertTrue(rule3["user_side_unknown"])
def test_thresholds_still_unapproved_no_decision_fabricated(self):
for entry in DEFAULT_THRESHOLDS.values():
self.assertEqual(entry["source"], "test_default_unapproved")
report = build_scorecard(_full_evidence(),
user_side_items=_all_bc_pending())
self.assertEqual(report["thresholds_source"],
"test_default_unapproved")
# 7 维 overall 本身仍只由 7 维决定:用户侧项不进 dimensions。
self.assertEqual(len(report["dimensions"]), 7)
self.assertEqual([d["id"] for d in report["dimensions"]],
["delivery_quality", "gate_audit", "arch_contract",
"test_coverage", "security_ledger", "cost_quota",
"ops_scale"])
def test_input_items_not_mutated(self):
items = _all_bc_pending()
snapshot = [dict(it) for it in items]
build_scorecard(_full_evidence(), user_side_items=items)
self.assertEqual(items, snapshot)
if __name__ == "__main__":
unittest.main()