diff --git a/cloud/audit_scorecard.py b/cloud/audit_scorecard.py new file mode 100644 index 0000000..11c9c54 --- /dev/null +++ b/cloud/audit_scorecard.py @@ -0,0 +1,516 @@ +"""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_current_state(): + """用当前真实状态证据跑出来的系统评分卡(不是上一轮判断分)。 + + 证据口径:生产仍跑 v0.2、main 仍 aebc623、CI 账单模式、 + #238/#240/#244/#241/#247/#249/#253/#255/#256/#258/#259/#260/#261 + 全部只到 PR OPEN。凡是只到 PR、未合并部署、未真实验证的, + 在这里都记为未知或已知未强制,不估算、不写成判断分。 + 总体结论必然是不达标。 + """ + 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, + }, + } + return build_scorecard(evidence) diff --git a/cloud/server.py b/cloud/server.py index 017f3b1..402b012 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -3133,6 +3133,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._metrics(user_id) elif path == "/metrics-report" and method == "GET": self._metrics_report(user_id) + elif path == "/audit-scorecard" and method == "GET": + self._audit_scorecard(user_id) elif path == "/models" and method == "GET": self._models_overview(user_id) elif path == "/relay/models" and method == "GET": @@ -10713,6 +10715,32 @@ pid 须为整数、address 须为字符串,否则 400(注册表里的 conn.close() metrics = _compute_metrics(records, costs) self._send_json(200, _build_metrics_report(metrics)) + def _audit_scorecard(self, user_id): + """只读 7 维审计评分卡报告端点。 + + 权限:与 /metrics 同款,任意项目成员即可读(不要求 owner), + 但本端点不统计任何项目数据、不写库、不写账本、不改链头、 + 不触发推回或重审——它只是把评分卡模块的当前状态报出来。 + 主基底(main)上没有存放结构化评分证据的表(metrics-report、 + 台账、账本、门禁模块都在未合并的 PR 分支上),所以本端点 + 在主基底上返回的只能是“全部未知/已知未强制”的诚实状态, + 总体必然是不达标;调用方要拿到真实分数,必须显式把结构化 + 证据传给 cloud/audit_scorecard.py 的 build_scorecard(), + 端点本身绝不从测试文件、摘要文档、PR 描述里拼凑分数。 + """ + try: + from cloud.audit_scorecard import example_scorecard_current_state + report = example_scorecard_current_state() + except ImportError: # pragma: no cover - 导入失败时仍要诚实返回 + report = { + "dimensions": [], "average_score": None, + "average_is_not_pass_criterion": True, + "overall": "not_pass", "overall_pass": False, + "missing_items": ["audit_scorecard_module_unavailable"], + "untested_items": [], + "thresholds_source": "test_default_unapproved", + } + self._send_json(200, report) # -- OAuth(pre-auth,靠 state + secret,不走 X-API-Key)----------- def _oauth_provider(self, path): diff --git a/cloud/tests/test_audit_scorecard.py b/cloud/tests/test_audit_scorecard.py new file mode 100644 index 0000000..32908a7 --- /dev/null +++ b/cloud/tests/test_audit_scorecard.py @@ -0,0 +1,232 @@ +"""7 维审计评分卡定向测试。 + +钉住的规则(任何一处改动都必须先改文档再改测试): +- 7 维全部 >=9 才达标;任一维 <9 总体不达标。 +- 平均分绝不能作为达标依据:6 维 10 分、1 维 4 分,平均 9.1 分, + 总体仍是不达标。 +- 未知维记 None 且总体不达标;未知不记 0、不记 10、不估算。 +- 夹具/视口替代/代码评审证据记未实测,不算通过。 +- 阈值默认值标注 test_default_unapproved;只有调用方显式传入 + user_approved 才算已拍板。 +- 端点只读:GET /audit-scorecard 不写库、不写账本、不改链头。 +""" + +import http.client +import json +import os +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +try: + from cloud.audit_scorecard import ( + DEFAULT_THRESHOLDS, build_scorecard, example_scorecard_current_state) + from cloud.auth import AuthStore + from cloud.server import create_server +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.audit_scorecard import ( + DEFAULT_THRESHOLDS, build_scorecard, example_scorecard_current_state) + from cloud.auth import AuthStore + from cloud.server import create_server + + +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, + }, + } + + +class ScorecardTestCase(unittest.TestCase): + def test_all_nine_passes(self): + report = build_scorecard(_full_evidence()) + self.assertEqual(report["overall"], "pass") + self.assertTrue(report["overall_pass"]) + self.assertEqual(len(report["dimensions"]), 7) + for dim in report["dimensions"]: + self.assertIsNotNone(dim["score"]) + self.assertGreaterEqual(dim["score"], 9.0) + self.assertEqual(dim["status"], "pass") + + def test_one_dimension_below_nine_fails_overall(self): + ev = _full_evidence() + ev["gate_audit"]["hard_gate_enforced"] = False + report = build_scorecard(ev) + self.assertEqual(report["overall"], "not_pass") + self.assertFalse(report["overall_pass"]) + gate = [d for d in report["dimensions"] if d["id"] == "gate_audit"][0] + self.assertEqual(gate["score"], 4.0) + + def test_average_91_with_one_four_still_fails(self): + # 6 维 10 分、1 维 4 分:平均 9.1 分,总体仍是不达标。 + # 构造方式:只让门禁维已知未强制(4.0),其余维满分证据。 + ev = _full_evidence() + ev["gate_audit"]["hard_gate_enforced"] = False + report = build_scorecard(ev) + self.assertEqual(report["average_score"], 9.1) + self.assertTrue(report["average_is_not_pass_criterion"]) + self.assertEqual(report["overall"], "not_pass") + self.assertFalse(report["overall_pass"]) + + def test_unknown_dimension_is_none_and_fails_overall(self): + report = build_scorecard({}) + self.assertEqual(report["overall"], "not_pass") + for dim in report["dimensions"]: + self.assertIsNone(dim["score"]) + self.assertEqual(dim["status"], "unknown") + self.assertTrue(report["missing_items"]) + self.assertIsNone(report["average_score"]) + + def test_unknown_is_not_zero_and_not_ten(self): + ev = _full_evidence() + ev["delivery_quality"] = {} + report = build_scorecard(ev) + dim = [d for d in report["dimensions"] + if d["id"] == "delivery_quality"][0] + self.assertIsNone(dim["score"]) + self.assertNotEqual(dim["score"], 0) + self.assertNotEqual(dim["score"], 10) + + def test_fixture_evidence_is_untested_not_pass(self): + ev = _full_evidence() + ev["test_coverage"]["cross_matrix_kind"] = "viewport_substitute" + report = build_scorecard(ev) + dim = [d for d in report["dimensions"] + if d["id"] == "test_coverage"][0] + self.assertIsNone(dim["score"]) + self.assertEqual(dim["status"], "unknown") + self.assertTrue(dim["untested_items"]) + self.assertEqual(report["overall"], "not_pass") + + def test_conditional_verdict_is_not_pass(self): + ev = _full_evidence() + ev["gate_audit"]["audit_verdict"] = "conditional" + report = build_scorecard(ev) + dim = [d for d in report["dimensions"] if d["id"] == "gate_audit"][0] + self.assertIsNone(dim["score"]) + self.assertTrue(dim["untested_items"]) + + def test_threshold_defaults_are_unapproved(self): + for entry in DEFAULT_THRESHOLDS.values(): + self.assertEqual(entry["source"], "test_default_unapproved") + report = build_scorecard(_full_evidence()) + self.assertEqual(report["thresholds_source"], + "test_default_unapproved") + + def test_user_approved_thresholds_change_source_only(self): + th = {k: {"value": v["value"], "source": "user_approved"} + for k, v in DEFAULT_THRESHOLDS.items()} + report = build_scorecard(_full_evidence(), thresholds=th) + self.assertEqual(report["thresholds_source"], "user_approved") + self.assertEqual(report["overall"], "pass") + + def test_example_current_state_is_not_pass_and_not_estimated(self): + # 示例是系统用当前真实状态证据算出来的,不是上一轮判断分。 + report = example_scorecard_current_state() + self.assertEqual(report["overall"], "not_pass") + self.assertFalse(report["overall_pass"]) + # 至少有一维是未知(None),且总体不达标不靠平均分。 + self.assertTrue(any(d["score"] is None + for d in report["dimensions"])) + + +class ScorecardEndpointTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def _req(self, method, path, key=None): + conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) + hdrs = {} + if key is not None: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, headers=hdrs) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + try: + parsed = json.loads(data.decode("utf-8")) if data else None + except ValueError: + parsed = None + return resp.status, parsed + + def test_endpoint_requires_auth(self): + status, _ = self._req("GET", "/audit-scorecard") + self.assertEqual(status, 401) + + def test_endpoint_is_readonly_and_not_pass_on_main_base(self): + status, report = self._req("GET", "/audit-scorecard", + key=self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(len(report["dimensions"]), 7) + self.assertEqual(report["overall"], "not_pass") + self.assertTrue(report["average_is_not_pass_criterion"]) + self.assertEqual(report["thresholds_source"], + "test_default_unapproved") + # 只读:端点调用前后,库里不产生评分卡表(本端点不写库)。 + import sqlite3 + conn = sqlite3.connect(self.db_path) + try: + tables = {r[0] for r in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'")} + finally: + conn.close() + self.assertNotIn("audit_scorecard", tables) diff --git a/docs/audit-scorecard.md b/docs/audit-scorecard.md new file mode 100644 index 0000000..9af447c --- /dev/null +++ b/docs/audit-scorecard.md @@ -0,0 +1,165 @@ +# 7 维项目级审计评分卡规格 + +本文件是评分卡的规格文档,也是唯一权威的计分说明。 +代码在 `cloud/audit_scorecard.py`,只读端点在 `server.py` 的 +`GET /audit-scorecard`,测试在 `cloud/tests/test_audit_scorecard.py`。 +三处不一致时,以“总体规则钉死”那一节为准,先改文档再改代码。 + +## 总体规则(钉死) + +- 7 维全部 >= 9.0 才算达标(overall="pass")。 +- 任一维 < 9.0,总体就不达标(overall="not_pass")。 +- 平均分(average_score)只作参考信息,绝不能作为达标依据。 + 反例已钉进测试:6 维 10 分、1 维 4 分,平均 9.1 分,总体仍是不达标。 +- 未知维记 None:数据源缺失、未知、未验证时,该维分数记为 None, + 不记 0、不记 10、不估算,总体直接不达标,并列出缺失项。 + 平均分只对已知维求平均(未知维不计入分母),所以平均分 + 绝不能被读成“7 维总分”。 +- 未实测维记未实测:调用方传入“夹具模拟”“视口替代” + “未编译端代码评审”“干跑”“估计”这类证据时,该项记为 + untested_items,而不是“通过”;若某维的关键证据全部是未实测, + 该维记为 None。 +- 阈值默认值一律标注 test_default_unapproved(测试默认值、未拍板)。 + 生产阈值须用户拍板后,由调用方显式传入 + thresholds(含 source="user_approved")才能替换默认值。 + 本模块绝不把测试初值写成生产值。 + +## 7 维定义与数据源 + +1. 交付质量(delivery_quality) + 数据源:#238 metrics + #249/#255 台账。 + 输入:first_pass_rate、defect_rate、rework_rate, + first_pass_known、defect_known、rework_known、attempts_present。 + 计算:三个率各自按“floor -> 5 分、9 分线 -> 9 分、10 分线 -> 10 分” + 分段线性算子分,维分数取三者最小值(任一率不达标,维就不达标)。 + +2. 门禁与审计有效性(gate_audit) + 数据源:#240 硬门禁、#244 推回、#247 接线、#260/#261 自动循环。 + 输入:hard_gate_enforced、gate_items_passed/total、 + merge_blocked_verified、loop_end_to_end_verified、 + sampling_misjudgment_rate、audit_verdict。 + 计算:门禁项通过率(99% -> 9 分、100% -> 10 分)与抽查判错率 + (越低越好)取最小值;hard_gate_enforced=False 已知未强制时 + 分数封顶 4.0;合并阻塞或自动循环端到端未真实验证时封顶 4.0; + audit_verdict="conditional" 记无效证据,不算通过。 + +3. 架构与契约一致性(arch_contract) + 数据源:#238 arch_conformance。 + 输入:contract_frozen、arch_violations、directory_violations、 + import_violations、stack_changed_without_approval。 + 计算:违规总数为 0 且合同已冻结且技术栈未擅自变更时 10 分; + 任一违规时封顶 5.0;未冻结或擅自换栈时封顶 4.0; + 任一违规数未知时记 None。 + +4. 测试与验证覆盖(test_coverage) + 数据源:G5 QA 矩阵、agent 专属六项、#253/#251 覆盖率与 + lint/mypy/bandit。 + 输入:prd_coverage_rate、cross_matrix_coverage_rate、coverage_pct、 + cross_matrix_kind、real_pg、real_service、agent_six_complete、 + lint_items、mypy_items、bandit_high。 + 计算:PRD 覆盖率、跨端矩阵覆盖率、代码覆盖率取最小值; + agent 六项缺项时封顶 4.0;静态检查总数 >0 时封顶 8.0; + 跨端矩阵不是 real、或真实 PG、真实服务未验证时,该维记 None。 + +5. 安全与账本可信(security_ledger) + 数据源:#241 哈希链、#250 旁路行上链、#256 shell 白名单、 + 分支保护状态。 + 输入:ledger_verify_ok、ledger_coverage_rate、 + shell_whitelist_pass_rate、branch_protection、anchor_check_recent。 + 计算:账本覆盖率取分;分支保护 unknown/缺失时记 None, + unprotected 已知未保护时封顶 4.0,账本校验未通过时封顶 4.0, + 锚点定期校验未落地时封顶 8.0。 + +6. 成本与配额可观测(cost_quota) + 数据源:#212 双列成本、#249 台账、#255 钩子。 + 输入:cost_records_complete、unpriced_as_zero、per_creator_present、 + quota_cutoff_verified。 + 计算:成本记录不完整(turns/llm_requests/usd_micros 缺失)时 + 记 None;无价记成 0/免费时 0.0 分;配额硬切断未真实验证时 + 封顶 4.0;按创建者成本未落地时封顶 8.0;四项全齐时 10 分。 + +7. 生产运维与规模(ops_scale) + 数据源:主机健康、4 个 unit、CI 状态、部署与回滚记录、 + 并行规模测试。 + 输入:ci_green_rate、deploy_success_rate、rollback_drill_done、 + ha_status、parallel_scale_proven、parallel_scale_accepted_alternative。 + 计算:CI 变绿率、部署成功率取最小值;回滚演练未真实跑过时 + 封顶 4.0;单主机且未接受风险时封顶 5.0;并行规模按 + proven/target * 10 算分(40 人满分,18 人约 4.5 分); + CI 变绿率未知或并行规模未知(且无已接受替代线)时记 None。 + +## 阈值表(默认值,全部未拍板) + +| 阈值 | 默认值 | 来源 | +|---|---|---| +| defect_rate 9 分线 / 10 分线 / floor | 0.05 / 0.02 / 0.20 | 测试默认值、未拍板 | +| first_pass 9 分线 / 10 分线 / floor | 0.90 / 0.95 / 0.50 | 测试默认值、未拍板 | +| rework_rate 9 分线 / 10 分线 / floor | 0.10 / 0.05 / 0.50 | 测试默认值、未拍板 | +| overturn_rate floor | 0.30 | 测试默认值、未拍板 | +| coverage 9 分线 / 10 分线 / floor | 0.85 / 0.90 / 0.75 | 测试默认值、未拍板 | +| sampling 普通 / 关键 | 0.10 / 0.50 | 测试默认值、未拍板 | +| parallel_scale_target | 40 | 测试默认值、未拍板 | + +注意:floor 不是达标线。floor 是分数进入 0-5 区间的参考点, +达标线永远是 9 分线。把 floor 读成达标线是错的。 + +## 与已有机制的关系(本评分卡不改它们) + +- #238 metrics-report:评分卡的交付质量维复用它的指标口径, + 但 metrics-report 本身不是评分卡;它的 evaluate_feedback + auto_apply 保持 False,评分卡不自动改标准。 +- #244 抽查:评分卡的门禁维复用它的抽查口径(必抽、随机比例、 + 纠错),但评分卡不重新做抽查,只接受抽查结果作为证据。 +- #260/#261 自动循环:评分卡的门禁维复用它的端到端验证 + (审计 fail -> 修复子任务 -> 完整重审 -> 解封),但评分卡 + 不触发推回或重审;端点本身只读。 +- #247 纠正语义:纠正为 pass 本身不解封,评分卡不设特例。 +- #241 账本:评分卡只接受账本校验结果作为证据,不写链、不改链头。 + +## 10 维到 7 维的合并映射 + +- 流程完整性并入门禁与审计有效性:G0-G8 最终靠门禁和重审证明。 +- 自动化程度并入门禁与审计有效性 + 生产运维与规模: + 自动化的证据就是自动循环是否真实跑通、真实 App 是否安装。 +- 文档与技能并入架构与契约一致性: + 技能文本、附录溯源、移植规则最终都是契约一致性的一部分。 +- 其余 7 维与上一轮 10 维一一对应,不再展开。 + +## 只读端点 + +`GET /audit-scorecard`,与 /metrics 同款认证与成员范围 +(任意项目成员即可读,不要求 owner),但本端点不统计任何 +项目数据、不写库、不写账本、不改链头、不触发推回或重审。 +主基底(main)上没有存放结构化评分证据的表(metrics-report、 +台账、账本、门禁模块都在未合并的 PR 分支上),所以本端点 +在主基底上返回的只能是“全部未知/已知未强制”的诚实状态, +总体必然是不达标;调用方要拿到真实分数,必须显式把结构化 +证据传给 `cloud/audit_scorecard.py` 的 `build_scorecard()`, +端点本身绝不从测试文件、摘要文档、PR 描述里拼凑分数。 + +## 示例评分卡(系统算出来的,不是上一轮判断分) + +下面是 `example_scorecard_current_state()` 用当前真实状态证据 +跑出来的系统输出,证据口径:生产仍跑 v0.2、main 仍 aebc623、 +CI 账单模式、#238/#240/#244/#241/#247/#249/#253/#255/#256/ +#258/#259/#260/#261 全部只到 PR OPEN。凡是只到 PR、未合并部署、 +未真实验证的,在这里都记为未知或已知未强制,不估算。 + +- 交付质量:None(未知,台账无行、三个率都未知) +- 门禁与审计有效性:4.0(已知未强制,record 模式) +- 架构与契约一致性:None(未知,违规数与冻结状态都未知) +- 测试与验证覆盖:None(未知,PRD 与跨端覆盖率缺失, + 跨端矩阵是视口替代、真实 PG 与真实服务未验证) +- 安全与账本可信:None(未知,账本校验与覆盖率未知, + 分支保护 unknown) +- 成本与配额可观测:None(未知,成本记录不完整) +- 生产运维与规模:None(未知,CI 变绿率未知) +- 平均分(仅已知维平均,参考信息):4.0 +- 总体结论:不达标(not_pass) + +上一轮十维估计分(流程 8、自动化 5、门禁审计代码 8/生产实效 4、 +代码架构 6.5、测试 6、生产运维 5、安全 5.5、成本 6、规模 4、 +文档技能 7.5)是上一轮的判断分,不是系统算出来的分数, +不能写进示例评分卡的分数栏,只能作为对照表放在这里, +且示例总体结论必须是不达标——因为门禁审计生产实效、测试、 +运维规模等维都远低于 9 分。