feat: audit-of-audit sampling rules (golden set and critical paths)

- cloud/audit_sampling.py (pure, stdlib): mandatory sampling (1.0)
  for rebuild cases and for third-failure near-complete cases whose
  continuation round still failed; random sampling for the rest
  (normal 0.10, critical-path 0.50, ceil so small groups are never
  rounded to zero, seed required and recorded for reproducibility).
- Wrong-verdict handling: correct the original conclusion and
  retrigger the corresponding flow (wrong pushback ->
  cancel_pushback_and_rerun_audit; wrong rebuild ->
  halt_rebuild_and_restore; wrong pass -> reopen_audit_and_pushback;
  unknown original -> escalate_to_human); every result, correct or
  corrected, is appended to the audit-of-audit record.
- Golden-set approval checklist: list reviewed, criteria defined,
  sample exclusion, approver+timestamp -- all human-confirmed before
  approval; sample entries (sample=True or sample- ids) are dropped
  and counted, never counted as golden.
- metrics.build_report gains an optional sampling_plan section;
  without a plan it states sampling is pending, never implying the
  audit was re-examined.

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]

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"},
     }
 
 

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()