Files
codex-app-server/cloud/audit_scorecard.py
T
host e32391fd85 code-to10 v6 A12: scorecard example renamed to 2026-10-31 snapshot with timeliness fields
(cherry picked from commit 633cdbc94e7a7b0c5d3b0b135189faf3de6673f4)

Reconcile-port: handler_admin.py hunks redirected into cloud/server.py _audit_scorecard (target keeps 6-mixin split; scorecard module itself applied cleanly).
2026-11-03 02:49:26 +00:00

524 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""7 维项目级审计评分卡(纯函数、标准库 only)。
定位:这是一个评分卡,不是门禁、不是审计、不是自动循环。
它不重写 metrics、arch_conformance、ledger、audit_loop 的逻辑,
只接受调用方显式传入的结构化证据,算出 7 维各自的分数。
计分规则(钉死,不可改为平均分达标):
- 每维分数 0-10,保留一位小数;>=9 通过,<9 不通过。
- 总体结论:7 维全部 >=9 才算达标(overall="pass")。
任一维 <9,总体就不达标(overall="not_pass")。
- 平均分只作参考信息(average_score),绝不能作为达标依据。
代码里把“平均分达标”写成明确的不通过路径并加测试钉住:
例如 6 维 10 分、1 维 4 分,平均 9.1 分,总体仍是不达标。
- 未知处理:任何一维的数据源缺失、未知、未验证,
该维分数记为 None(未知),不记 0、不记 10、不估算,
总体直接不达标,并列出缺失项(missing_items)。
“没数据”既不算通过,也不算 0 分去拉低平均分后又用平均分糊弄。
- 未实测处理:调用方传入“夹具模拟”“视口替代”“未编译端代码评审”
这类证据时,该项记为“未实测”(untested_items),而不是“通过”。
若某维的关键证据全部是未实测,该维记为 None。
- 阈值:DEFAULT_THRESHOLDS 全部标注 source="test_default_unapproved"
(测试默认值、未拍板)。生产阈值须用户拍板后,由调用方显式
传入 thresholds(含 source="user_approved")才能替换默认值。
本模块绝不把测试初值写成生产值。
- 与已定语义的关系(本模块不改它们):
#238 evaluate_feedback 的 auto_apply 保持 False;
#247 纠正为 pass 本身不解封;
#244 CONDITIONAL 先映射为 fail 再记(本模块只接受已映射的结论,
若调用方传入 verdict="conditional",记为未实测/无效证据,不算通过)。
7 维定义(不要擅自增维、减维、改名换义):
1. delivery_quality 交付质量
2. gate_audit 门禁与审计有效性
3. arch_contract 架构与契约一致性
4. test_coverage 测试与验证覆盖
5. security_ledger 安全与账本可信
6. cost_quota 成本与配额可观测
7. ops_scale 生产运维与规模
10 维到 7 维的合并映射(写进文档,也写在这里):
- 流程完整性并入门禁与审计有效性(G0-G8 最终靠门禁和重审证明);
- 自动化程度并入门禁与审计有效性 + 生产运维与规模
(自动化的证据就是自动循环是否真实跑通、真实 App 是否安装);
- 文档与技能并入架构与契约一致性
(技能文本、附录溯源、移植规则最终都是契约一致性的一部分)。
输入格式:build_scorecard(evidence, thresholds=None)
evidence: dict,键为上面 7 个维 id,值为该维的证据 dict。
每个维的证据字段见各 _score_* 函数的 docstring。
调用方必须带来源、时间、检查名;本模块不自动从测试文件、
摘要文档、PR 描述里拼凑分数。
"""
from __future__ import annotations
PASS_SCORE = 9.0
DIMENSIONS = [
("delivery_quality", "交付质量"),
("gate_audit", "门禁与审计有效性"),
("arch_contract", "架构与契约一致性"),
("test_coverage", "测试与验证覆盖"),
("security_ledger", "安全与账本可信"),
("cost_quota", "成本与配额可观测"),
("ops_scale", "生产运维与规模"),
]
DIMENSION_IDS = [d[0] for d in DIMENSIONS]
# 阈值表:每一项都带 source,默认值一律是测试初值、未拍板。
# “9 分线”是达到 9 分的下限,“10 分线”是达到 10 分的下限,
# “floor”是低于该值后分数进入 0-5 区间的参考点(不是达标线)。
# 这些数值本身不是生产标准;生产阈值须用户拍板后显式传入。
DEFAULT_THRESHOLDS = {
"defect_rate_nine": {"value": 0.05, "source": "test_default_unapproved"},
"defect_rate_ten": {"value": 0.02, "source": "test_default_unapproved"},
"defect_rate_floor": {"value": 0.20, "source": "test_default_unapproved"},
"first_pass_nine": {"value": 0.90, "source": "test_default_unapproved"},
"first_pass_ten": {"value": 0.95, "source": "test_default_unapproved"},
"first_pass_floor": {"value": 0.50, "source": "test_default_unapproved"},
"rework_rate_nine": {"value": 0.10, "source": "test_default_unapproved"},
"rework_rate_ten": {"value": 0.05, "source": "test_default_unapproved"},
"rework_rate_floor": {"value": 0.50, "source": "test_default_unapproved"},
"overturn_rate_floor": {"value": 0.30, "source": "test_default_unapproved"},
"coverage_nine": {"value": 0.85, "source": "test_default_unapproved"},
"coverage_ten": {"value": 0.90, "source": "test_default_unapproved"},
"coverage_floor": {"value": 0.75, "source": "test_default_unapproved"},
"sampling_normal": {"value": 0.10, "source": "test_default_unapproved"},
"sampling_critical": {"value": 0.50, "source": "test_default_unapproved"},
"parallel_scale_target": {"value": 40, "source": "test_default_unapproved"},
}
UNTESTED_KINDS = {"fixture", "viewport_substitute", "code_review_only",
"fixture_simulated", "dry_run", "estimated"}
def _thresholds_source(thresholds):
"""阈值来源:只有调用方显式传入 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 _t(thresholds, key):
if thresholds and key in thresholds:
entry = thresholds[key]
if isinstance(entry, dict):
return entry["value"]
return entry
return DEFAULT_THRESHOLDS[key]["value"]
def _round1(value):
return round(float(value) + 1e-9, 1)
def _score_higher(value, nine, ten, floor):
"""越高越好:floor->5 分,nine->9 分,ten->10 分,分段线性,低于 floor 按比例。"""
if value is None:
return None
v = float(value)
if v >= ten:
return 10.0
if v >= nine:
return 9.0 + (v - nine) / (ten - nine) if ten > nine else 10.0
if v >= floor:
return 5.0 + 4.0 * (v - floor) / (nine - floor) if nine > floor else 5.0
if floor > 0:
return max(0.0, 5.0 * v / floor)
return 0.0
def _score_lower(value, ten, nine, floor):
"""越低越好:ten->10 分,nine->9 分,floor->5 分,低于 ten 满分,高于 floor 递减。"""
if value is None:
return None
v = float(value)
if v <= ten:
return 10.0
if v <= nine:
return 9.0 + (nine - v) / (nine - ten) if nine > ten else 10.0
if v <= floor:
return 5.0 + 4.0 * (floor - v) / (floor - nine) if floor > nine else 5.0
return max(0.0, 5.0 * floor / v) if v > 0 else 0.0
def _result(dim_id, score, basis, missing, untested):
if score is None:
status = "unknown"
elif score >= PASS_SCORE:
status = "pass"
else:
status = "fail"
return {
"id": dim_id,
"score": _round1(score) if score is not None else None,
"status": status,
"basis": basis,
"missing_items": missing,
"untested_items": untested,
}
def _score_delivery(ev, th):
"""证据:first_pass_rate, defect_rate, rework_rate,
first_pass_known, defect_known, rework_known, attempts_present,
evidence_kind("real" 以外若为 estimated/fixture 则记未实测)。"""
missing, untested, basis = [], [], []
for key in ("first_pass_rate", "defect_rate", "rework_rate"):
if ev.get(key) is None:
missing.append(key)
for key in ("first_pass_known", "defect_known", "rework_known"):
if not ev.get(key):
missing.append(key + "_not_known")
if not ev.get("attempts_present"):
missing.append("attempts_present_false_or_missing")
kind = ev.get("evidence_kind")
if kind in UNTESTED_KINDS:
untested.append("metrics_evidence_kind=%s" % kind)
if missing or untested:
return _result("delivery_quality", None, basis, missing, untested)
subs = [
("first_pass_rate", _score_higher(ev["first_pass_rate"],
_t(th, "first_pass_nine"),
_t(th, "first_pass_ten"),
_t(th, "first_pass_floor"))),
("defect_rate", _score_lower(ev["defect_rate"],
_t(th, "defect_rate_ten"),
_t(th, "defect_rate_nine"),
_t(th, "defect_rate_floor"))),
("rework_rate", _score_lower(ev["rework_rate"],
_t(th, "rework_rate_ten"),
_t(th, "rework_rate_nine"),
_t(th, "rework_rate_floor"))),
]
for name, sub in subs:
basis.append("%s=%s -> %.1f" % (name, ev[name], sub))
return _result("delivery_quality", min(s for _, s in subs), basis, missing, untested)
def _score_gate(ev, th):
"""证据:hard_gate_enforced(bool/None)、gate_items_passed、
gate_items_total、merge_blocked_verified、loop_end_to_end_verified、
sampling_misjudgment_rate(None 表示未抽查/未知)、
audit_verdict("conditional" 记无效证据,不算通过)。"""
missing, untested, basis = [], [], []
if ev.get("hard_gate_enforced") is None:
missing.append("hard_gate_enforced_unknown")
if ev.get("gate_items_passed") is None or ev.get("gate_items_total") is None:
missing.append("gate_items_passed_or_total_missing")
if ev.get("audit_verdict") == "conditional":
untested.append("audit_verdict=conditional_not_a_pass")
if ev.get("evidence_kind") in UNTESTED_KINDS:
untested.append("gate_evidence_kind=%s" % ev["evidence_kind"])
if missing or untested:
return _result("gate_audit", None, basis, missing, untested)
if not ev.get("hard_gate_enforced"):
basis.append("hard_gate_enforced=False(record 模式):已知未强制,分数封顶 4.0")
return _result("gate_audit", 4.0, basis, missing, untested)
total = ev["gate_items_total"]
passed = ev["gate_items_passed"]
rate = passed / total if total else 0.0
sub_gate = _score_higher(rate, 0.99, 1.0, 0.80)
basis.append("gate_items %s/%s=%.3f -> %.1f" % (passed, total, rate, sub_gate))
subs = [sub_gate]
if not ev.get("merge_blocked_verified"):
basis.append("merge_blocked_verified=False:合并阻塞未真实验证,分数封顶 4.0")
return _result("gate_audit", 4.0, basis, missing, untested)
if not ev.get("loop_end_to_end_verified"):
basis.append("loop_end_to_end_verified=False:自动循环端到端未真实验证,分数封顶 4.0")
return _result("gate_audit", 4.0, basis, missing, untested)
mis = ev.get("sampling_misjudgment_rate")
if mis is None:
missing.append("sampling_misjudgment_rate_unknown")
return _result("gate_audit", None, basis, missing, untested)
sub_mis = _score_lower(mis, 0.0, 0.02, _t(th, "overturn_rate_floor"))
basis.append("sampling_misjudgment_rate=%s -> %.1f" % (mis, sub_mis))
subs.append(sub_mis)
return _result("gate_audit", min(subs), basis, missing, untested)
def _score_arch(ev, th):
"""证据:contract_frozen、arch_violations、directory_violations、
import_violations、stack_changed_without_approval(bool/None)。"""
missing, untested, basis = [], [], []
for key in ("arch_violations", "directory_violations", "import_violations"):
if ev.get(key) is None:
missing.append(key + "_unknown")
if ev.get("contract_frozen") is None:
missing.append("contract_frozen_unknown")
if missing:
return _result("arch_contract", None, basis, missing, untested)
total_violations = (ev["arch_violations"] + ev["directory_violations"]
+ ev["import_violations"])
basis.append("violations total=%s (arch=%s dir=%s import=%s)"
% (total_violations, ev["arch_violations"],
ev["directory_violations"], ev["import_violations"]))
if not ev.get("contract_frozen"):
basis.append("contract_frozen=False:已知未冻结,分数封顶 4.0")
return _result("arch_contract", 4.0, basis, missing, untested)
if ev.get("stack_changed_without_approval"):
basis.append("stack_changed_without_approval=True:分数封顶 4.0")
return _result("arch_contract", 4.0, basis, missing, untested)
if total_violations > 0:
basis.append("violations>0:分数封顶 5.0")
return _result("arch_contract", 5.0, basis, missing, untested)
return _result("arch_contract", 10.0, basis, missing, untested)
def _score_test(ev, th):
"""证据:prd_coverage_rate、agent_six_complete、cross_matrix_coverage_rate、
coverage_pct、lint_items、mypy_items、bandit_high、
cross_matrix_kind("real" 以外记未实测)、real_pg(bool)、
real_service(bool)。"""
missing, untested, basis = [], [], []
for key in ("prd_coverage_rate", "cross_matrix_coverage_rate", "coverage_pct"):
if ev.get(key) is None:
missing.append(key + "_missing")
kind = ev.get("cross_matrix_kind")
if kind in UNTESTED_KINDS or kind in ("viewport_substitute", "code_review_only"):
untested.append("cross_matrix_kind=%s" % kind)
if kind is not None and kind != "real":
untested.append("cross_matrix_not_real")
if not ev.get("real_pg"):
untested.append("real_pg_not_verified")
if not ev.get("real_service"):
untested.append("real_service_not_verified")
if missing or untested:
return _result("test_coverage", None, basis, missing, untested)
if not ev.get("agent_six_complete"):
basis.append("agent_six_complete=False:agent 专属六项缺项,分数封顶 4.0")
return _result("test_coverage", 4.0, basis, missing, untested)
subs = [
("prd_coverage_rate", _score_higher(ev["prd_coverage_rate"], 0.99, 1.0, 0.80)),
("cross_matrix_coverage_rate", _score_higher(ev["cross_matrix_coverage_rate"], 0.99, 1.0, 0.80)),
("coverage_pct", _score_higher(ev["coverage_pct"],
_t(th, "coverage_nine"),
_t(th, "coverage_ten"),
_t(th, "coverage_floor"))),
]
for name, sub in subs:
basis.append("%s=%s -> %.1f" % (name, ev[name], sub))
# 静态检查:数值缺失记缺失,不估算;有值时,>0 即封顶(9 分线要求接近清零)。
for key in ("lint_items", "mypy_items", "bandit_high"):
if ev.get(key) is None:
missing.append(key + "_missing")
if missing:
return _result("test_coverage", None, basis, missing, untested)
static_total = ev["lint_items"] + ev["mypy_items"] + ev["bandit_high"]
basis.append("static total lint=%s mypy=%s bandit_high=%s"
% (ev["lint_items"], ev["mypy_items"], ev["bandit_high"]))
if static_total > 0:
basis.append("static_total>0:分数封顶 8.0(静态检查未接近清零)")
return _result("test_coverage", min(8.0, min(s for _, s in subs)), basis, missing, untested)
return _result("test_coverage", min(s for _, s in subs), basis, missing, untested)
def _score_security(ev, th):
"""证据:ledger_verify_ok(bool/None)、ledger_coverage_rate、
shell_whitelist_pass_rate、branch_protection
("protected"/"unprotected"/"unknown"/None)、anchor_check_recent。"""
missing, untested, basis = [], [], []
if ev.get("ledger_verify_ok") is None:
missing.append("ledger_verify_ok_unknown")
if ev.get("ledger_coverage_rate") is None:
missing.append("ledger_coverage_rate_missing")
bp = ev.get("branch_protection")
if bp is None or bp == "unknown":
missing.append("branch_protection_unknown")
if missing:
return _result("security_ledger", None, basis, missing, untested)
if bp == "unprotected":
basis.append("branch_protection=unprotected:已知未保护,分数封顶 4.0")
return _result("security_ledger", 4.0, basis, missing, untested)
if not ev.get("ledger_verify_ok"):
basis.append("ledger_verify_ok=False:账本校验未通过,分数封顶 4.0")
return _result("security_ledger", 4.0, basis, missing, untested)
subs = [_score_higher(ev["ledger_coverage_rate"], 0.99, 1.0, 0.80)]
basis.append("ledger_coverage_rate=%s -> %.1f" % (ev["ledger_coverage_rate"], subs[0]))
rate = ev.get("shell_whitelist_pass_rate")
if rate is not None:
sub = _score_higher(rate, 0.99, 1.0, 0.80)
subs.append(sub)
basis.append("shell_whitelist_pass_rate=%s -> %.1f" % (rate, sub))
if not ev.get("anchor_check_recent"):
basis.append("anchor_check_recent=False:锚点定期校验未落地,分数封顶 8.0")
return _result("security_ledger", min(8.0, min(subs)), basis, missing, untested)
return _result("security_ledger", min(subs), basis, missing, untested)
def _score_cost(ev, th):
"""证据:cost_records_complete(bool)、unpriced_as_zero(bool,
True 即违规)、per_creator_present、quota_cutoff_verified、
thresholds_approved(bool,仅影响阈值来源标注,不封顶分数)。"""
missing, untested, basis = [], [], []
if not ev.get("cost_records_complete"):
missing.append("cost_records_complete_false_or_missing"
"(turns/llm_requests/usd_micros 缺失)")
if ev.get("unpriced_as_zero") is None:
missing.append("unpriced_as_zero_unknown")
if missing:
return _result("cost_quota", None, basis, missing, untested)
if ev.get("unpriced_as_zero"):
basis.append("unpriced_as_zero=True:无价记成 0/免费,分数封顶 0.0")
return _result("cost_quota", 0.0, basis, missing, untested)
if not ev.get("quota_cutoff_verified"):
basis.append("quota_cutoff_verified=False:配额硬切断未真实验证,分数封顶 4.0")
return _result("cost_quota", 4.0, basis, missing, untested)
if not ev.get("per_creator_present"):
basis.append("per_creator_present=False:按创建者成本未落地,分数封顶 8.0")
return _result("cost_quota", 8.0, basis, missing, untested)
basis.append("cost_records_complete=True, quota_cutoff_verified=True, "
"per_creator_present=True, unpriced_as_zero=False")
return _result("cost_quota", 10.0, basis, missing, untested)
def _score_ops(ev, th):
"""证据:ci_green_rate(None 表示 CI 未绿/未知)、deploy_success_rate、
rollback_drill_done、ha_status
("second_host"/"single_host_accepted"/"single_host_unaccepted"/None)、
parallel_scale_proven(int/None)、parallel_scale_accepted_alternative(bool)。"""
missing, untested, basis = [], [], []
if ev.get("ci_green_rate") is None:
missing.append("ci_green_rate_unknown(CI 未绿或未统计)")
if ev.get("rollback_drill_done") is None:
missing.append("rollback_drill_done_unknown")
ha = ev.get("ha_status")
if ha is None:
missing.append("ha_status_unknown")
if missing:
return _result("ops_scale", None, basis, missing, untested)
if not ev.get("rollback_drill_done"):
basis.append("rollback_drill_done=False:回滚演练未真实跑过,分数封顶 4.0")
return _result("ops_scale", 4.0, basis, missing, untested)
subs = [_score_higher(ev["ci_green_rate"], 0.99, 1.0, 0.80)]
basis.append("ci_green_rate=%s -> %.1f" % (ev["ci_green_rate"], subs[0]))
if ev.get("deploy_success_rate") is not None:
sub = _score_higher(ev["deploy_success_rate"], 0.99, 1.0, 0.80)
subs.append(sub)
basis.append("deploy_success_rate=%s -> %.1f" % (ev["deploy_success_rate"], sub))
if ha == "single_host_unaccepted":
basis.append("ha_status=single_host_unaccepted:分数封顶 5.0")
return _result("ops_scale", 5.0, basis, missing, untested)
proven = ev.get("parallel_scale_proven")
target = _t(th, "parallel_scale_target")
if proven is None and not ev.get("parallel_scale_accepted_alternative"):
missing.append("parallel_scale_proven_unknown")
return _result("ops_scale", None, basis, missing, untested)
if proven is not None:
sub = _score_higher(min(proven, target) / target if target else 0,
0.99, 1.0, 0.5)
# 规模是离散证据:40 人满分,18 人约 4.5 分,上面线性公式在
# 0.5-1.0 区间给出 5-10 分,与“18 人约 4.5”不一致,故直接用比例*10。
sub = _round1(10.0 * min(proven, target) / target) if target else 0.0
subs.append(sub)
basis.append("parallel_scale_proven=%s target=%s -> %.1f"
% (proven, target, sub))
return _result("ops_scale", min(subs), basis, missing, untested)
_SCORERS = {
"delivery_quality": _score_delivery,
"gate_audit": _score_gate,
"arch_contract": _score_arch,
"test_coverage": _score_test,
"security_ledger": _score_security,
"cost_quota": _score_cost,
"ops_scale": _score_ops,
}
def build_scorecard(evidence, thresholds=None):
"""算出 7 维评分卡。evidence 缺键按空证据处理(该维记 None)。"""
evidence = evidence or {}
th = thresholds or {}
dimensions = []
missing_all, untested_all = [], []
for dim_id, _name in DIMENSIONS:
result = _SCORERS[dim_id](evidence.get(dim_id) or {}, th)
dimensions.append(result)
missing_all.extend("%s: %s" % (dim_id, m) for m in result["missing_items"])
untested_all.extend("%s: %s" % (dim_id, u) for u in result["untested_items"])
scores = [d["score"] for d in dimensions if d["score"] is not None]
average = _round1(sum(scores) / len(scores)) if scores else None
# 总体规则钉死:平均分绝不能作为达标依据。
overall_pass = all(d["score"] is not None and d["score"] >= PASS_SCORE
for d in dimensions)
return {
"dimensions": dimensions,
"average_score": average,
"average_is_not_pass_criterion": True,
"overall": "pass" if overall_pass else "not_pass",
"overall_pass": overall_pass,
"pass_score": PASS_SCORE,
"missing_items": missing_all,
"untested_items": untested_all,
"thresholds_source": _thresholds_source(th),
"thresholds": {k: v for k, v in DEFAULT_THRESHOLDS.items()},
}
def example_scorecard_as_of_2026_10_31():
"""2026-10-31 快照示例卡(历史示例,不是当前系统得分)。
证据是本函数 2026-10-31 落笔时记录的快照值(生产 v0.2、
main aebc623、覆盖率 0.7787 等),此后未重新实测;调用方
绝不能把它读成当前系统得分。要当前分数,须显式把新证据
传给 build_scorecard()。凡是只到 PR、未合并部署、未真实
验证的,在这里都记为未知或已知未强制,不估算、不写成
判断分。阈值仍为测试默认值
(source="test_default_unapproved"),未拍板。
总体结论必然是不达标。
"""
evidence = {
"delivery_quality": {
"first_pass_rate": None, "defect_rate": None, "rework_rate": None,
"first_pass_known": False, "defect_known": False,
"rework_known": False, "attempts_present": False,
},
"gate_audit": {
"hard_gate_enforced": False, "gate_items_passed": 0,
"gate_items_total": 5, "merge_blocked_verified": False,
"loop_end_to_end_verified": False,
"sampling_misjudgment_rate": None,
},
"arch_contract": {
"contract_frozen": None, "arch_violations": None,
"directory_violations": None, "import_violations": None,
},
"test_coverage": {
"prd_coverage_rate": None, "cross_matrix_coverage_rate": None,
"coverage_pct": 0.7787, "cross_matrix_kind": "viewport_substitute",
"real_pg": False, "real_service": False,
"agent_six_complete": True,
"lint_items": 109, "mypy_items": 0, "bandit_high": 4,
},
"security_ledger": {
"ledger_verify_ok": None, "ledger_coverage_rate": None,
"branch_protection": "unknown",
},
"cost_quota": {
"cost_records_complete": False, "unpriced_as_zero": False,
"per_creator_present": False, "quota_cutoff_verified": False,
},
"ops_scale": {
"ci_green_rate": None, "rollback_drill_done": False,
"ha_status": "single_host_unaccepted",
"parallel_scale_proven": 18,
},
}
report = build_scorecard(evidence)
report["evidence_as_of"] = "2026-10-31"
report["evidence_kind"] = "historical_example_snapshot"
report["not_current_system_score"] = True
return report