diff --git a/cloud/audit_sampling.py b/cloud/audit_sampling.py new file mode 100644 index 0000000..10e32af --- /dev/null +++ b/cloud/audit_sampling.py @@ -0,0 +1,243 @@ +"""Audit-of-audit sampling: human review of audit verdicts. + +Zero third-party dependencies; standard library only, no ``cloud.*`` +imports. All functions are pure: they return plans, verdicts and +records, never auto-apply a correction -- applying a correction is a +separate, recorded step (see ``handle_sampling_result``). + +This module implements the sampling half of the "audit itself must +be audited" rule. It does NOT change the audit verdicts, the +pushback/rebuild counters, or the golden-set approval flow; it only +defines which audit cases a human must re-examine, and what happens +when that re-examination finds the audit wrong. + +Sampling rules (all testable, none left to "regular review"): + +1. Mandatory sampling (ratio 1.0, never skipped): + - ``rebuild``: any case whose audit decision was rebuild-from- + scratch (推倒重做). A rebuild destroys the old implementation + in the workspace, so its verdict must always be re-examined. + - ``third_failure_near_complete_still_failed``: a case whose + third failure was root-caused as near-complete (接近完成), + was given one targeted continuation round, and still failed. + This is the exact boundary where the "one more round vs + rebuild" judgement is most error-prone, so it is always sampled. + A near-complete third failure whose continuation round SUCCEEDED + is not mandatory (the judgement proved itself), and a + near-complete first/second failure is not mandatory either (the + threshold is the third failure, per the rebuild rule). + +2. Random sampling of the rest, with the seed recorded: + - normal audit cases: default ratio 0.10; + - critical-path cases (``critical_path=True``): default ratio + 0.50 -- critical paths get a higher rate, not a free pass. + The seed is part of the plan: a sample selected without a + recorded seed is not reproducible and therefore not auditable. + Selection is deterministic given (cases, ratios, seed). + +3. Golden-set approval is separate from sampling and is NOT + satisfied by sample files: ``filter_golden_tasks`` drops any task + flagged as a sample (``sample=True`` or an id starting with + ``sample-``) and reports how many were dropped; a golden set + counts only its remaining, human-approved tasks. The approval + checklist (``golden_approval_checklist``) lists every item that + must be human-completed before a golden set is approved; until + then every item reads ``pending_human``. + +Handling a sampling result (``handle_sampling_result``): + +- Sampling says the audit was CORRECT: no correction, only a record + entry (the audit stands, with evidence it was re-examined). +- Sampling says the audit was WRONG: the original conclusion is + corrected in the record, and the corresponding flow is + re-triggered: + - wrongly issued pushback -> ``cancel_pushback_and_rerun_audit`` + (cancel the pushback, restore the pre-pushback state, rerun); + - wrongly issued rebuild -> ``halt_rebuild_and_restore`` + (halt the rebuild, restore the original implementation from + git history/tag, rerun -- the old implementation is preserved + by the rebuild rule itself, so restoration is possible); + - wrongly issued pass (a failure the audit missed) -> + ``reopen_audit_and_pushback`` (reopen with the corrected + verdict and run the pushback flow the audit should have run); + - unknown original decision -> ``escalate_to_human`` (never + invent a retrigger). + Every outcome, correct or corrected, is appended to the + audit-of-audit record (``append_audit_record``); a sampling result + that is not recorded is treated as not having happened. +""" + +import math +import random + +DEFAULT_NORMAL_RATIO = 0.10 +DEFAULT_CRITICAL_PATH_RATIO = 0.50 + +# Original audit decisions -> retrigger when sampling proves them wrong. +_RETRIGGER = { + "fail_pushback": "cancel_pushback_and_rerun_audit", + "rebuild": "halt_rebuild_and_restore", + "pass": "reopen_audit_and_pushback", +} + + +def mandatory_trigger(case): + """Return the mandatory trigger name for ``case``, or None. + + ``case`` fields used: ``decision``, ``failure_count``, + ``root_cause`` (``"structural"`` / ``"near_complete"`` / other), + ``continuation_failed`` (bool). + """ + if case.get("decision") == "rebuild": + return "rebuild" + if (case.get("failure_count", 0) >= 3 + and case.get("root_cause") == "near_complete" + and case.get("continuation_failed")): + return "third_failure_near_complete_still_failed" + return None + + +def select_sample(cases, seed, normal_ratio=DEFAULT_NORMAL_RATIO, + critical_ratio=DEFAULT_CRITICAL_PATH_RATIO): + """Select the audit sample. Returns a plan dict. + + Mandatory cases are always included. The remaining cases are + grouped into critical-path and normal, each shuffled with + ``random.Random(seed + ":" + group)`` and truncated to + ``ceil(ratio * group_size)`` (a group of 1 with any positive + ratio therefore contributes 1 -- a small group is never rounded + down to zero and silently unsampled). ``seed`` is required: an + unseeded selection cannot be reproduced on review. + """ + if seed is None or seed == "": + raise ValueError("sampling seed is required: an unseeded " + "sample cannot be reproduced on review") + mandatory = [] + critical = [] + normal = [] + for case in cases: + trigger = mandatory_trigger(case) + if trigger is not None: + mandatory.append((case, trigger)) + elif case.get("critical_path"): + critical.append(case) + else: + normal.append(case) + + def _draw(group, ratio, label): + shuffled = list(group) + random.Random("%s:%s" % (seed, label)).shuffle(shuffled) + if not shuffled or ratio <= 0: + return [] + return shuffled[:math.ceil(ratio * len(shuffled))] + + sampled_critical = _draw(critical, critical_ratio, "critical") + sampled_normal = _draw(normal, normal_ratio, "normal") + return { + "seed": seed, + "ratios": {"normal": normal_ratio, + "critical_path": critical_ratio, + "mandatory": 1.0}, + "mandatory": [{"case_id": c.get("id"), "trigger": t} + for c, t in mandatory], + "random_critical_path": [c.get("id") for c in sampled_critical], + "random_normal": [c.get("id") for c in sampled_normal], + "sampled_ids": ([c.get("id") for c, _ in mandatory] + + [c.get("id") for c in sampled_critical] + + [c.get("id") for c in sampled_normal]), + } + + +def filter_golden_tasks(tasks): + """Drop sample entries from a golden-set candidate list. + + A task is a sample (and never counts as golden) when it is + flagged ``sample=True`` or its id starts with ``sample-``. + Returns ``{"golden": [...], "dropped_samples": int}``. + """ + golden = [] + dropped = 0 + for task in tasks: + task_id = task.get("id", "") if isinstance(task, dict) else "" + is_sample = (isinstance(task, dict) and task.get("sample")) \ + or str(task_id).startswith("sample-") + if is_sample: + dropped += 1 + else: + golden.append(task) + return {"golden": golden, "dropped_samples": dropped} + + +def golden_approval_checklist(tasks, approver=None, approved_at=None, + list_reviewed=False, + criteria_defined=False): + """Approval checklist for a golden set. Every item is + ``pending_human`` until a human completes it (via the + ``list_reviewed`` / ``criteria_defined`` confirmations, plus + approver + timestamp); the set is approved only when every item + is done AND at least one non-sample golden task remains. Sample + files never satisfy an item (see ``filter_golden_tasks``).""" + filtered = filter_golden_tasks(tasks) + items = [ + {"item": "task_list_reviewed", + "status": "done" if list_reviewed else "pending_human", + "note": "a human reviewed the golden task list"}, + {"item": "acceptance_criteria_defined", + "status": "done" if criteria_defined else "pending_human", + "note": "each golden task has human-defined acceptance criteria"}, + {"item": "sample_files_excluded", + "status": "done", + "note": "%d sample entries excluded from the golden set " + "(exclusion is applied by filter_golden_tasks, never " + "by trusting the candidate list)" + % filtered["dropped_samples"]}, + {"item": "approver_and_timestamp_recorded", + "status": "done" if approver and approved_at else "pending_human", + "note": "approver and approval timestamp recorded"}, + ] + approved = all(i["status"] == "done" for i in items) \ + and bool(filtered["golden"]) and bool(approver) \ + and bool(approved_at) + return {"approved": approved, "items": items, + "golden_tasks": filtered["golden"], + "dropped_samples": filtered["dropped_samples"]} + + +def handle_sampling_result(case, sampling_verdict, corrected_verdict=None): + """Turn one human sampling result into a record entry + actions. + + ``sampling_verdict`` is ``"correct"`` or ``"incorrect"``. + When incorrect, ``corrected_verdict`` (the audit decision that + should have been issued) must be supplied; the retrigger follows + the ORIGINAL decision via ``_RETRIGGER``, or escalates when the + original decision is unknown. Returns the entry dict; the caller + appends it with ``append_audit_record``. + """ + case_id = case.get("id") + original = case.get("decision") + if sampling_verdict not in ("correct", "incorrect"): + raise ValueError("sampling_verdict must be 'correct' or " + "'incorrect', got %r" % (sampling_verdict,)) + if sampling_verdict == "correct": + return {"case_id": case_id, "original_decision": original, + "sampling_verdict": "correct", + "corrected_decision": None, "retrigger": None, + "recorded_in": "audit_of_audit"} + if corrected_verdict is None: + raise ValueError("an incorrect sampling verdict requires a " + "corrected_verdict") + retrigger = _RETRIGGER.get(original, "escalate_to_human") + return {"case_id": case_id, "original_decision": original, + "sampling_verdict": "incorrect", + "corrected_decision": corrected_verdict, + "retrigger": retrigger, + "recorded_in": "audit_of_audit"} + + +def append_audit_record(log, entry): + """Append one sampling entry to the audit-of-audit record. + + Returns a new list; the caller persists it. A sampling result + that is not appended here is treated as not having happened. + """ + return list(log) + [entry] diff --git a/cloud/metrics.py b/cloud/metrics.py index dfbf09e..2ff2cdf 100644 --- a/cloud/metrics.py +++ b/cloud/metrics.py @@ -206,7 +206,7 @@ def evaluate_feedback(metrics, thresholds=None, check_stats=None): def build_report(metrics, golden=None, thresholds=None, thresholds_source="default_unapproved", - check_stats=None): + check_stats=None, sampling_plan=None): """Assemble the JSON report document. ``golden``: None, or a dict with ``tasks`` (list of task ids), @@ -245,6 +245,15 @@ def build_report(metrics, golden=None, thresholds=None, # (arch_conformance.check_contract_version) with coordinator # approval, never by auto-applying an action record. "feedback_consumers": ["routing", "audit_standards"], + # Audit-of-audit sampling (cloud/audit_sampling.py): when a + # plan is supplied it is carried verbatim; otherwise the + # report states sampling is pending rather than implying + # the audit was re-examined. + "sampling": sampling_plan or { + "status": "pending", + "note": "no audit-of-audit sample plan supplied; audit " + "verdicts in this report have not been " + "re-examined"}, } diff --git a/cloud/tests/test_audit_sampling.py b/cloud/tests/test_audit_sampling.py new file mode 100644 index 0000000..36a8a92 --- /dev/null +++ b/cloud/tests/test_audit_sampling.py @@ -0,0 +1,173 @@ +"""Tests for cloud.audit_sampling: audit-of-audit sampling rules. + +Every rule has a positive and a negative case. Only the standard +library; no real model, no network, no secrets. +""" + +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname( + os.path.dirname(os.path.abspath(__file__))))) + +from cloud.audit_sampling import ( # noqa: E402 + append_audit_record, filter_golden_tasks, + golden_approval_checklist, handle_sampling_result, + mandatory_trigger, select_sample) + + +def _case(case_id, **kw): + case = {"id": case_id} + case.update(kw) + return case + + +class TestMandatoryTriggers(unittest.TestCase): + def test_rebuild_is_always_mandatory(self): + self.assertEqual( + mandatory_trigger(_case("r", decision="rebuild")), "rebuild") + + def test_third_failure_near_complete_still_failed_is_mandatory(self): + case = _case("n", decision="fail_pushback", failure_count=3, + root_cause="near_complete", continuation_failed=True) + self.assertEqual(mandatory_trigger(case), + "third_failure_near_complete_still_failed") + + def test_near_complete_continuation_success_is_not_mandatory(self): + case = _case("n", decision="pass", failure_count=3, + root_cause="near_complete", continuation_failed=False) + self.assertIsNone(mandatory_trigger(case)) + + def test_second_failure_near_complete_is_not_mandatory(self): + case = _case("n", decision="fail_pushback", failure_count=2, + root_cause="near_complete", continuation_failed=True) + self.assertIsNone(mandatory_trigger(case)) + + def test_structural_third_failure_is_not_this_trigger(self): + case = _case("n", decision="rebuild", failure_count=3, + root_cause="structural", continuation_failed=True) + # It IS mandatory, but via the rebuild trigger, not the + # near-complete trigger -- the trigger name must be exact. + self.assertEqual(mandatory_trigger(case), "rebuild") + + def test_ordinary_case_is_not_mandatory(self): + self.assertIsNone(mandatory_trigger( + _case("o", decision="pass", failure_count=1))) + + +class TestRandomSample(unittest.TestCase): + def _cases(self): + normal = [_case("n%d" % i, decision="pass") for i in range(20)] + critical = [_case("c%d" % i, decision="pass", critical_path=True) + for i in range(10)] + return normal + critical + + def test_ratios_and_reproducibility(self): + plan = select_sample(self._cases(), seed="seed-1") + # ceil(0.10 * 20) normal, ceil(0.50 * 10) critical. + self.assertEqual(len(plan["random_normal"]), 2) + self.assertEqual(len(plan["random_critical_path"]), 5) + self.assertEqual(plan["seed"], "seed-1") + again = select_sample(self._cases(), seed="seed-1") + self.assertEqual(plan["sampled_ids"], again["sampled_ids"]) + + def test_mandatory_cases_are_included_not_double_counted(self): + cases = [_case("r", decision="rebuild"), + _case("n0", decision="pass")] + plan = select_sample(cases, seed="s") + self.assertIn("r", plan["sampled_ids"]) + self.assertEqual(plan["mandatory"][0]["trigger"], "rebuild") + self.assertEqual(len(plan["sampled_ids"]), + len(set(plan["sampled_ids"]))) + + def test_small_group_is_never_rounded_to_zero(self): + plan = select_sample([_case("only", decision="pass")], seed="s") + self.assertEqual(plan["random_normal"], ["only"]) + + def test_missing_seed_is_rejected(self): + with self.assertRaises(ValueError): + select_sample(self._cases(), seed="") + + +class TestGoldenApproval(unittest.TestCase): + def test_sample_entries_never_count_as_golden(self): + tasks = [{"id": "sample-1"}, {"id": "real-1", "sample": True}, + {"id": "real-2"}] + filtered = filter_golden_tasks(tasks) + self.assertEqual([t["id"] for t in filtered["golden"]], + ["real-2"]) + self.assertEqual(filtered["dropped_samples"], 2) + + def test_unapproved_checklist_is_pending(self): + checklist = golden_approval_checklist([{"id": "real-2"}]) + self.assertFalse(checklist["approved"]) + self.assertTrue(any(i["status"] == "pending_human" + for i in checklist["items"])) + + def test_fully_confirmed_checklist_is_approved(self): + checklist = golden_approval_checklist( + [{"id": "real-2"}], approver="human", + approved_at="2026-10-31", list_reviewed=True, + criteria_defined=True) + self.assertTrue(checklist["approved"]) + + def test_sample_only_set_cannot_be_approved(self): + checklist = golden_approval_checklist( + [{"id": "sample-1"}], approver="human", + approved_at="2026-10-31", list_reviewed=True, + criteria_defined=True) + self.assertFalse(checklist["approved"]) + self.assertEqual(checklist["golden_tasks"], []) + + +class TestSamplingResultHandling(unittest.TestCase): + def test_correct_verdict_only_records(self): + entry = handle_sampling_result( + _case("a", decision="pass"), "correct") + self.assertIsNone(entry["corrected_decision"]) + self.assertIsNone(entry["retrigger"]) + log = append_audit_record([], entry) + self.assertEqual(len(log), 1) + self.assertEqual(log[0]["recorded_in"], "audit_of_audit") + + def test_wrong_pushback_is_cancelled_and_rerun(self): + entry = handle_sampling_result( + _case("a", decision="fail_pushback"), "incorrect", + corrected_verdict="pass") + self.assertEqual(entry["retrigger"], + "cancel_pushback_and_rerun_audit") + self.assertEqual(entry["corrected_decision"], "pass") + + def test_wrong_rebuild_is_halted_and_restored(self): + entry = handle_sampling_result( + _case("a", decision="rebuild"), "incorrect", + corrected_verdict="fail_pushback") + self.assertEqual(entry["retrigger"], "halt_rebuild_and_restore") + + def test_wrong_pass_reopens_and_pushes_back(self): + entry = handle_sampling_result( + _case("a", decision="pass"), "incorrect", + corrected_verdict="fail_pushback") + self.assertEqual(entry["retrigger"], + "reopen_audit_and_pushback") + + def test_unknown_original_decision_escalates(self): + entry = handle_sampling_result( + _case("a", decision="mystery"), "incorrect", + corrected_verdict="pass") + self.assertEqual(entry["retrigger"], "escalate_to_human") + + def test_incorrect_without_correction_is_rejected(self): + with self.assertRaises(ValueError): + handle_sampling_result(_case("a", decision="pass"), + "incorrect") + + def test_invalid_verdict_is_rejected(self): + with self.assertRaises(ValueError): + handle_sampling_result(_case("a", decision="pass"), + "maybe") + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/arch-conformance-metrics.md b/docs/arch-conformance-metrics.md index d37951c..426f829 100644 --- a/docs/arch-conformance-metrics.md +++ b/docs/arch-conformance-metrics.md @@ -64,3 +64,32 @@ workflow——第 2 轨拥有 lint/类型/安扫，本轨不碰。 推翻率 0.30）只是测试用初值，报告标注 `thresholds_source: "default_unapproved"`；生产阈值须人工拍板后 以 `thresholds_source: "human_approved"` 传入，见待审定清单。 + +## 审计的审计：抽查口径（`cloud/audit_sampling.py`，纯函数） + +审计结论本身要被人工抽查，抽查规则可测试、可核查，不留 +“定期复核”口号： + +- **必抽（比例 1.0）**：推倒重做案例必抽；第三次失败根因判为 + 接近完成、续轮仍 fail 的案例必抽。接近完成但续轮成功的、 + 第 1/2 次失败接近完成的，均不属必抽（阈值就是第三次）。 +- **随机抽查**：其余案例按组抽取，普通案例默认 0.10，关键路径 + 案例默认 0.50；小数组 ceil 取整，绝不因四舍五入抽到 0。 + 抽取须带 seed（`select_sample` 无 seed 即报错），seed 记入 + 抽查计划，抽取结果可复现、可事后核查。 +- **抽查结果处理**：抽查判审计正确 → 只记一条审计的审计记录， + 原结论不动；判错 → 纠正原结论并重新触发相应流程：错发推回 + → `cancel_pushback_and_rerun_audit`；错发推倒 → + `halt_rebuild_and_restore`（旧实现按推倒规则保存在 git 历史， + 可恢复）；错放行 → `reopen_audit_and_pushback`；原结论未知 + → `escalate_to_human`，绝不编造触发动作。每条结果（无论对错） + 都须记入审计的审计记录，未记入即视为未发生。 +- **金任务审定清单**（`golden_approval_checklist`）：任务清单人工 + 复核、每任务验收标准人工定义、样本排除、审定人+时间记录， + 四项全 done 且至少留一个非样本任务才算批准；样本任务 + （`sample=True` 或 id 以 `sample-` 开头）一律从金任务集中剔除 + 并计数，示例文件永远不算金任务。报告的 `sampling` 段在无抽查 + 计划时明写 pending，不暗示审计已被复核。 + +待人工项（单独列）：金任务集审定清单的四项人工确认；抽查 seed +的记录与保管；生产抽查比例拍板（默认值 0.10/0.50 只是测试初值）。