audit: unify audit-of-audit sampling rule (gap 2)

Single sampling rule; track 3 (#244) is authoritative for WHO is
mandatory, track 5 (#238) for HOW the rest are drawn:

- Mandatory (1.0): golden-task records + rebuild / refreeze /
  escalation records + third-failure near-complete still-failed
  cases (mandatory_triggers, full list per case). Track 3's old
  deterministic every-10th ordinary rule is removed (it drew zero
  from batches under 10 ordinary records and was predictable);
  track 5's old narrow mandatory list is expanded with golden /
  refreeze / escalate. blocked_quota stays non-mandatory.
- Random tiers: normal 0.10, critical-path 0.50, seed required
  and recorded, ceil so a non-empty group is never rounded to
  zero. 0.10/0.50 are TEST DEFAULTS, not approved production
  values; production rates remain a user decision.
- One pass, no double counting: mandatory records leave the
  random pools before drawing; a multi-trigger record (e.g.
  golden + rebuild) is sampled once with all triggers listed,
  and its correction follows the (original, corrected) verdict
  pair, never the trigger list.
- cloud/audit_pushback.py::select_audit_sample is now only a
  record-shaped adapter over cloud/audit_sampling.py::select_sample
  (seed required, retired reason names critical_path-for-kinds
  and periodic_sample removed); AUDIT_SAMPLE_EVERY_NTH deleted.
- Correction: original never mutated, correction appended,
  corrector != signer enforced in audit_sampling when both names
  are supplied; gate effect pinned (correction to pass never
  unblocks by itself -- schedule/cancel-and-rerun, blocked until
  a new passing re-audit; correction to fail re-blocks at once).
  Track 3 / track 5 retrigger vocabularies documented as synonyms.
- Skill text (tightening 3 item 2 + correction paragraph)
  rewritten to the unified rule; new test file pins the
  boundaries (mandatory union, dedup, ceil incl. the 9-record
  case, seed reproducibility, golden-set construction under both
  sample vocabularies, adapter order/non-mutation, correction
  preservation on both tracks).

cloud/audit_pushback.py

@@ -212,9 +212,10 @@ Tightening 3 -- the audit itself is audited. (i) Independent
 re-runs use exact artifacts (#211 triple-hash), never reports
 alone. (ii) Golden tasks (human-approved only; example/sample
 files are rejected by ``validate_golden_task_set``) and
-rebuild/refreeze/escalation records are always sampled, and
-ordinary records are sampled every ``AUDIT_SAMPLE_EVERY_NTH``
-records, deterministically (``select_audit_sample``). (iii) G6
+rebuild/refreeze/escalation records are always sampled, and the
+rest are drawn by seeded random sampling (``select_audit_sample``,
+which only adapts ``cloud/audit_sampling.py::select_sample`` --
+the unified sampling rule lives there, not here). (iii) G6
 user acceptance is the final human gate and is never ticked by
 automation (``check_g6_guard``). A sampled audit found wrong is
 corrected by appending a correction record that preserves the
@@ -282,10 +283,15 @@ MAX_TARGETED_FIX_ROUNDS_PER_CONTRACT_VERSION = 1
 # items, with no core-structure item and no broken passed item,
 # may qualify as "nearly done". More than this is not near-complete.
 NEAR_COMPLETE_MAX_FAILED_ITEMS = 2
-# Audit-of-audit sampling (tightening 3): golden tasks and
-# rebuild/refreeze/escalation records are always sampled; ordinary
-# records are sampled deterministically every Nth record.
-AUDIT_SAMPLE_EVERY_NTH = 10
+# Audit-of-audit sampling (tightening 3): the unified rule lives in
+# cloud/audit_sampling.py (mandatory = golden tasks + rebuild /
+# refreeze / escalation + third-failure near-complete still failed,
+# all at 1.0; the rest by seeded random draw, normal 0.10 /
+# critical-path 0.50, ceil so a small group is never rounded to
+# zero). The old deterministic "every 10th record" rule is
+# superseded and removed: it drew zero from batches under 10
+# ordinary records. The 0.10 / 0.50 ratios are TEST DEFAULTS, not
+# approved production values; production rates are a user decision.
 
 # Gate checks, in the order they are reported. The first nine are the
 # hard trigger list from the audit-hardening brief; the tenth is the
@@ -1151,27 +1157,66 @@ def validate_golden_task_set(tasks):
     return (not reasons, reasons)
 
 
-def select_audit_sample(records, golden_task_ids):
+def select_audit_sample(records, golden_task_ids, seed=None,
+                        normal_ratio=None, critical_ratio=None):
     """Select audit records for audit-of-audit sampling (tightening 3).
 
-    Always sampled: golden tasks, and rebuild / refreeze /
-    escalation records (the critical paths). Ordinary records are
-    sampled deterministically every AUDIT_SAMPLE_EVERY_NTH record
-    (by position, so the sample is reproducible and checkable).
-    Returns a list of ``{"record": ..., "reason": ...}``.
+    ADAPTER, not a second rule: this function translates records
+    into ``cloud/audit_sampling.py`` cases, runs the unified
+    ``select_sample`` there, and translates the plan back into
+    record order. The unified rule is: mandatory (1.0) = golden
+    tasks + rebuild / refreeze / escalation records + third
+    failure root-caused near-complete whose continuation round
+    still failed; the rest by seeded random draw (normal default
+    0.10, critical-path default 0.50, ceil so a non-empty group
+    is never rounded to zero). ``seed`` is REQUIRED -- a missing
+    seed raises ValueError, because an unseeded draw cannot be
+    reproduced on review. Selection is a single pass: a mandatory
+    record is removed from the random pools before drawing, so it
+    consumes no random quota, and a multi-trigger record is
+    returned ONCE with all its triggers in ``reasons`` (``reason``
+    is the first trigger, in golden / rebuild / third-failure /
+    refreeze / escalate order).
+
+    Retired reason names: track 3's old ``critical_path`` label
+    for mandatory kinds and ``periodic_sample`` label for every
+    10th record no longer exist -- mandatory kinds carry their own
+    trigger names, and random draws carry ``random_critical_path``
+    / ``random_normal``. Input records are never mutated.
+
+    Returns a list of ``{"record": ..., "reason": ...,
+    "reasons": [...]}`` in original record order.
     """
-    golden = set(golden_task_ids or [])
+    from cloud.audit_sampling import select_sample as _select_sample
+
+    records = list(records or [])
+    cases = []
+    for index, record in enumerate(records):
+        case = dict(record)
+        # Unique per-record id for the plan; the golden check uses
+        # task_id, so sharing a task_id across records is fine.
+        case["id"] = "record-%d" % index
+        cases.append(case)
+    kwargs = {}
+    if normal_ratio is not None:
+        kwargs["normal_ratio"] = normal_ratio
+    if critical_ratio is not None:
+        kwargs["critical_ratio"] = critical_ratio
+    plan = _select_sample(cases, seed, golden_task_ids=golden_task_ids,
+                          **kwargs)
+    reasons_by_id = {}
+    for entry in plan["mandatory"]:
+        reasons_by_id[entry["case_id"]] = list(entry["triggers"])
+    for case_id in plan["random_critical_path"]:
+        reasons_by_id[case_id] = ["random_critical_path"]
+    for case_id in plan["random_normal"]:
+        reasons_by_id[case_id] = ["random_normal"]
     selected = []
-    for index, record in enumerate(records or []):
-        reason = None
-        if record.get("task_id") in golden:
-            reason = "golden_task"
-        elif record.get("kind") in ("rebuild", "refreeze", "escalate"):
-            reason = "critical_path"
-        elif (index + 1) % AUDIT_SAMPLE_EVERY_NTH == 0:
-            reason = "periodic_sample"
-        if reason:
-            selected.append({"record": record, "reason": reason})
+    for index, record in enumerate(records):
+        reasons = reasons_by_id.get("record-%d" % index)
+        if reasons:
+            selected.append({"record": record, "reason": reasons[0],
+                             "reasons": list(reasons)})
     return selected
 
 
@@ -1188,7 +1233,14 @@ def build_audit_correction(original_entry, corrected_verdict, corrector,
     (should have blocked) re-triggers pushback and re-blocks;
     a wrong block (should have passed) schedules a re-audit --
     it does not jump straight to merge, because the correction
-    itself is a review, not a full audit.
+    itself is a review, not a full audit. This is the SAME pair
+    rule as ``cloud/audit_sampling.py::handle_sampling_result``,
+    only the vocabulary differs (there: wrong pass ->
+    ``reopen_audit_and_pushback``, wrong block ->
+    ``cancel_pushback_and_rerun_audit`` where "cancel" means
+    cancel-and-rerun, never cancel-and-merge); the gate effect is
+    identical and is pinned by the wiring track (#247): a
+    correction to ``pass`` never unblocks by itself.
     """
     errors = []
     signer = original_entry.get("signer")

cloud/audit_sampling.py

@@ -11,9 +11,31 @@ 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.
 
+UNIFIED SAMPLING RULE (single rule; this module is canonical for
+selection, ``cloud/audit_pushback.py::select_audit_sample`` is only
+its record-shaped adapter -- there is no second rule):
+
+The scope of "mandatory" comes from track 3 (#244): golden-task
+records and rebuild / refreeze / escalation records. The drawing
+mechanism comes from track 5 (#238): seeded random draws with ceil
+rounding. Track 3's old deterministic "every 10th record" rule is
+superseded and removed: a batch of fewer than 10 ordinary records
+drew zero under it, and a position-known draw is predictable. The
+rates below are TEST DEFAULTS (``DEFAULT_*``), not approved
+production values; the production rates are still a user decision.
+Mandatory 1.0 is a rule, not a tunable rate.
+
 Sampling rules (all testable, none left to "regular review"):
 
-1. Mandatory sampling (ratio 1.0, never skipped):
+1. Mandatory sampling (ratio 1.0, never skipped). A case is
+   mandatory when ANY of these triggers holds (trigger order for
+   display and for the legacy single-trigger helper is golden,
+   rebuild, third-failure, refreeze, escalate):
+   - ``golden_task``: the case's task is in the human-approved
+     golden set (``task_id`` when present, else ``id``). A golden
+     task's records are mandatory whatever their kind -- an
+     ordinary pass verdict on a golden task is sampled, not just
+     its failures.
    - ``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.
@@ -22,18 +44,63 @@ Sampling rules (all testable, none left to "regular review"):
      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.
+   - ``refreeze``: a contract/design re-freeze record.
+   - ``escalate``: an escalation record.
    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).
+   threshold is the third failure, per the rebuild rule). A
+   ``blocked_quota`` record is NOT mandatory under this rule (it
+   is not in track 3's mandatory kind list); it is sampled only
+   through the random tiers below.
 
 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).
+   - critical-path cases (``critical_path=True`` and NOT already
+     mandatory): default ratio 0.50 -- critical paths get a higher
+     rate, not a free pass. (Track 3's old ``critical_path``
+     reason label for mandatory kinds is retired: mandatory kinds
+     now carry their own trigger names, and ``critical_path``
+     means only this 0.50 random tier.)
+   Mandatory cases are removed from the random pools BEFORE
+   drawing, so they consume no random quota. 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).
+
+4. One pass, no double counting. Selection partitions every case
+   exactly once: mandatory (all its triggers listed, one sample,
+   one sampling entry) or random-critical or random-normal. A
+   record that is both a golden task AND a rebuild is sampled
+   ONCE, with ``triggers == ["golden_task", "rebuild"]``; its
+   correction, if any, is a single record whose retrigger follows
+   the (original, corrected) verdict pair, not the trigger list --
+   triggers decide selection only, never the correction action.
+   Two different records (e.g. a triage record and the later
+   rebuild record for the same task) are two samples, not a
+   double count: deduplication is per record, not per task.
+
+5. Golden-set construction. The caller-supplied golden id set
+   must be built from human-approved, non-sample tasks only:
+   ``build_golden_task_ids`` applies both vocabularies -- a task
+   counts only with ``human_approved_by`` set AND none of
+   ``is_example``, ``sample=True``, or an id starting with
+   ``sample-``. Track 3's per-task approver rule and track 5's
+   sample-exclusion rule are both preserved; neither alone is
+   sufficient.
+
+6. Correction gate effect (shared with track 3 / wiring #247).
+   A sampling correction NEVER unblocks a merge by itself: a
+   correction to ``pass`` (a wrong block) schedules a re-audit
+   and the gate stays blocked until a new passing re-audit
+   conclusion exists; a correction to ``fail`` (a wrong pass)
+   re-triggers pushback and re-blocks at once. Vocabulary
+   synonyms for the same pair actions: track 3
+   ``retrigger_pushback_and_reblock`` == track 5
+   ``reopen_audit_and_pushback`` (wrong pass), and track 3
+   ``schedule_reaudit`` == track 5
+   ``cancel_pushback_and_rerun_audit`` (wrong block, where
+   "cancel" means cancel-and-rerun, never cancel-and-merge).
 
 3. Golden-set approval is separate from sampling and is NOT
    satisfied by sample files: ``filter_golden_tasks`` drops any task
@@ -81,33 +148,88 @@ _RETRIGGER = {
 }
 
 
-def mandatory_trigger(case):
-    """Return the mandatory trigger name for ``case``, or None.
+def mandatory_triggers(case, golden_task_ids=None):
+    """Return ALL mandatory trigger names for ``case`` (possibly []).
 
-    ``case`` fields used: ``decision``, ``failure_count``,
-    ``root_cause`` (``"structural"`` / ``"near_complete"`` / other),
+    Trigger order: golden_task, rebuild,
+    third_failure_near_complete_still_failed, refreeze, escalate.
+    ``case`` fields used: ``task_id`` (else ``id``) for the golden
+    check, ``decision`` / ``kind`` for rebuild / refreeze /
+    escalate, ``failure_count``, ``root_cause``
+    (``"structural"`` / ``"near_complete"`` / other),
     ``continuation_failed`` (bool).
     """
-    if case.get("decision") == "rebuild":
-        return "rebuild"
+    triggers = []
+    golden = set(golden_task_ids or [])
+    task_id = case.get("task_id") if case.get("task_id") is not None \
+        else case.get("id")
+    if task_id is not None and task_id in golden:
+        triggers.append("golden_task")
+    if case.get("decision") == "rebuild" or case.get("kind") == "rebuild":
+        triggers.append("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
+        triggers.append("third_failure_near_complete_still_failed")
+    if case.get("decision") == "refreeze" or case.get("kind") == "refreeze":
+        triggers.append("refreeze")
+    if case.get("decision") == "escalate" or case.get("kind") == "escalate":
+        triggers.append("escalate")
+    return triggers
+
+
+def mandatory_trigger(case, golden_task_ids=None):
+    """Return the first mandatory trigger name for ``case``, or None.
+
+    Legacy single-trigger view of ``mandatory_triggers``; new code
+    must use ``mandatory_triggers`` so a multi-trigger case is not
+    silently reported under one reason only.
+    """
+    triggers = mandatory_triggers(case, golden_task_ids)
+    return triggers[0] if triggers else None
+
+
+def build_golden_task_ids(tasks):
+    """Build the golden task id list from candidate tasks.
+
+    A task counts only when it has a human approver
+    (``human_approved_by``) AND is not a sample/example under any
+    vocabulary: ``is_example``, ``sample=True``, or an id starting
+    with ``sample-``. Returns ids in input order, samples dropped
+    silently here (count them with ``filter_golden_tasks`` when a
+    dropped count must be reported).
+    """
+    ids = []
+    for task in tasks or []:
+        if not isinstance(task, dict):
+            continue
+        task_id = task.get("id", "")
+        if task.get("is_example") or task.get("sample"):
+            continue
+        if str(task_id).startswith("sample-"):
+            continue
+        if not task.get("human_approved_by"):
+            continue
+        ids.append(task_id)
+    return ids
 
 
 def select_sample(cases, seed, normal_ratio=DEFAULT_NORMAL_RATIO,
-                  critical_ratio=DEFAULT_CRITICAL_PATH_RATIO):
+                  critical_ratio=DEFAULT_CRITICAL_PATH_RATIO,
+                  golden_task_ids=None):
     """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
+    Single pass, no double counting: mandatory cases (all their
+    triggers listed) are always included and removed from the
+    random pools before drawing. 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.
+    ``golden_task_ids`` is the human-approved golden set (see
+    ``build_golden_task_ids``); without it no case is golden.
     """
     if seed is None or seed == "":
         raise ValueError("sampling seed is required: an unseeded "
@@ -116,9 +238,9 @@ def select_sample(cases, seed, normal_ratio=DEFAULT_NORMAL_RATIO,
     critical = []
     normal = []
     for case in cases:
-        trigger = mandatory_trigger(case)
-        if trigger is not None:
-            mandatory.append((case, trigger))
+        triggers = mandatory_triggers(case, golden_task_ids)
+        if triggers:
+            mandatory.append((case, triggers))
         elif case.get("critical_path"):
             critical.append(case)
         else:
@@ -138,7 +260,9 @@ def select_sample(cases, seed, normal_ratio=DEFAULT_NORMAL_RATIO,
         "ratios": {"normal": normal_ratio,
                    "critical_path": critical_ratio,
                    "mandatory": 1.0},
-        "mandatory": [{"case_id": c.get("id"), "trigger": t}
+        "golden_task_ids": sorted(golden_task_ids or []),
+        "mandatory": [{"case_id": c.get("id"), "trigger": t[0],
+                       "triggers": list(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],
@@ -203,7 +327,8 @@ def golden_approval_checklist(tasks, approver=None, approved_at=None,
             "dropped_samples": filtered["dropped_samples"]}
 
 
-def handle_sampling_result(case, sampling_verdict, corrected_verdict=None):
+def handle_sampling_result(case, sampling_verdict, corrected_verdict=None,
+                           corrector=None, original_signer=None):
     """Turn one human sampling result into a record entry + actions.
 
     ``sampling_verdict`` is ``"correct"`` or ``"incorrect"``.
@@ -211,13 +336,30 @@ def handle_sampling_result(case, sampling_verdict, corrected_verdict=None):
     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``.
+    appends it with ``append_audit_record``. This function never
+    mutates ``case``: the original conclusion is preserved and the
+    correction lives only in the returned entry.
+
+    Corrector separation (track 3 rule, enforced here when both
+    names are supplied): ``corrector`` must differ from
+    ``original_signer``. Gate effect (shared with track 3 / wiring
+    #247, stated here so the retrigger names cannot be misread):
+    a correction never unblocks a merge by itself -- a correction
+    to ``pass`` schedules a re-audit (``cancel_pushback_and_rerun_audit``
+    == track 3 ``schedule_reaudit``) and the gate stays blocked
+    until a new passing re-audit exists; a correction to ``fail``
+    re-triggers pushback and re-blocks at once
+    (``reopen_audit_and_pushback`` == track 3
+    ``retrigger_pushback_and_reblock``).
     """
     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 (corrector is not None and original_signer is not None
+            and corrector == original_signer):
+        raise ValueError("corrector must differ from original signer")
     if sampling_verdict == "correct":
         return {"case_id": case_id, "original_decision": original,
                 "sampling_verdict": "correct",

cloud/skills/software-engineering/SKILL.md

@@ -923,11 +923,29 @@ tests/                 # QA 线程专属，含跨端一致性用例
 - 审计兜底三件套，缺一不可，全部可测试或可核查：
   ① 独立重跑的确切产物测试，不是只看报告：复用三方哈希模式
      （核心报告、产物清单、重算 hash 三方一致），产物不一致即停。
-  ② 金任务集与关键路径的人工审定样本，定期抽查审计结论本身：
-     金任务集必须人工审定（每项有 `human_approved_by`），示例文件
-     不算金任务；抽查比例写死：金任务与推倒/重新冻结/升级记录
-     100% 抽查，普通记录每 10 条确定性抽 1 条（按序号、可复现，
-     不是随机抽）；抽查结果处理：抽查发现判错即走纠错路径。
+  ② 审计结论本身的抽查，唯一口径（第 2 项补缺口对齐后，
+     代码 `cloud/audit_sampling.py`、本段、测试三处一致，
+     不再有两套口径）：必抽范围以第 3 轨为准、抽取机制以
+     第 5 轨为准。第 3 轨原文“普通记录每 10 条确定性抽 1 条
+     （不是随机抽）”已废止——不足 10 条的普通批次按它抽到
+     0 条，且序号抽取可被预测；第 5 轨原文必抽窄清单已
+     扩充——金任务、重新冻结、升级记录补进必抽。
+     必抽（比例 1.0，规则不是可调比例）：金任务记录（任务在
+     人工审定金任务集内，不论该记录本身是何种结论）、推倒、
+     重新冻结、升级记录，以及第三次失败根因判为接近完成、
+     续轮仍 fail 的案例。金任务集必须人工审定（每项有
+     `human_approved_by`），且样本/示例一律排除（`is_example`
+     或 `sample=True` 或 id 以 `sample-` 开头，任一即排除）。
+     随机抽查（其余记录，须带 seed，seed 记入抽查计划，
+     同 seed 同批同结果、可事后复现）：普通记录默认 0.10、
+     关键路径记录（`critical_path=True` 的非必抽记录）默认
+     0.50；每组 ceil 取整，非空组至少抽 1 条，绝不因四舍五入
+     抽到 0。0.10 / 0.50 只是测试默认值，不是已拍板的生产值，
+     生产抽查比例仍待用户拍板。单次抽取、不双重计入：必抽
+     记录先划出，不进随机池、不占随机名额；一记录多触发
+     （如既是金任务又是推倒）只抽一次、理由全列、纠错只记
+     一条；纠错动作按（原结论，纠正结论）对查表，与触发理由
+     无关。抽查发现判错即走纠错路径（见下）。
   ③ G6 用户验收这个最后人工关口不能被自动化绕过去：审计升级、
      推回、推倒、抽查纠错都不能替用户勾验收，验收清单最后只能用户勾。
 - 审计判错的纠错路径：事后抽查发现审计结论判错（该拦没拦、或
@@ -936,7 +954,17 @@ tests/                 # QA 线程专属，含跨端一致性用例
   原记录不动，追加一条纠错记录（引用原记录 hash、写清原结论、
   纠正后结论、纠错人与原因）；该拦没拦的纠正 → 重新触发推回并
   恢复阻塞；不该拦拦了的纠正 → 安排重新审计（不是直接放行合并，
-  因为纠错本身只是复核、不是完整审计）；纠错人必须与原签字人不同。
+  因为纠错本身只是复核、不是完整审计），门禁在新通过重审结论
+  出现前继续阻塞；纠错人必须与原签字人不同。动作名对照（同
+  一对动作、两轨词汇）：错放行 → 第 3 轨
+  `retrigger_pushback_and_reblock` = 第 5 轨
+  `reopen_audit_and_pushback`；错拦截 → 第 3 轨
+  `schedule_reaudit` = 第 5 轨 `cancel_pushback_and_rerun_audit`
+  （其中 cancel 指取消并重审，绝不是取消即放行）；错推倒 →
+  停推倒并从 git 历史恢复旧实现；原结论未知 → 升级人工。
+  已废止的理由名：`critical_path`（曾指必抽种类，现只指
+  0.50 随机档）、`periodic_sample`（曾指每 10 条抽取，已随
+  确定性抽取一并废止）。
 
 ## 诚实边界：原生客户端（v0.5 后置占位版）
 

cloud/tests/test_audit_pushback.py

@@ -784,18 +784,29 @@ class AuditOfAuditTestCase(unittest.TestCase):
         valid3, _r3 = validate_golden_task_set([{"id": "gold-2"}])
         self.assertFalse(valid3)
 
-    def test_sampling_covers_golden_critical_and_periodic(self):
+    def test_sampling_covers_golden_mandatory_and_random(self):
+        # Unified rule (gap 2): mandatory kinds carry their own
+        # trigger names (no "critical_path" label for kinds, no
+        # "periodic_sample"); ordinary records are drawn by seeded
+        # random sampling with ceil, so 9 ordinary records still
+        # contribute exactly ceil(0.10 * 9) == 1 -- the old
+        # deterministic every-10th rule drew zero from this batch.
         records = [{"task_id": "gold-1", "kind": "pushback"}]
         records += [{"task_id": "t-%d" % i, "kind": "pushback"}
                     for i in range(9)]
         records.append({"task_id": "t-re", "kind": "rebuild"})
-        selected = select_audit_sample(records, {"gold-1"})
+        selected = select_audit_sample(records, {"gold-1"}, seed="s")
         reasons = {entry["record"]["task_id"]: entry["reason"]
                    for entry in selected}
         self.assertEqual(reasons["gold-1"], "golden_task")
-        self.assertEqual(reasons["t-re"], "critical_path")
-        # 10th record (index 9) is the periodic sample.
-        self.assertIn("t-8", reasons)
+        self.assertEqual(reasons["t-re"], "rebuild")
+        ordinary = [t for t in reasons if t.startswith("t-")
+                    and t != "t-re"]
+        self.assertEqual(len(ordinary), 1)
+        self.assertNotIn("periodic_sample", set(reasons.values()))
+        self.assertNotIn("critical_path", set(reasons.values()))
+        with self.assertRaises(ValueError):
+            select_audit_sample(records, {"gold-1"})
 
     def test_correction_preserves_original_and_restores_flow(self):
         original = {"task_id": "t-1", "round": 0, "verdict": "pass",

cloud/tests/test_audit_sampling_alignment.py

@@ -0,0 +1,253 @@
+"""Tests for the UNIFIED audit-of-audit sampling rule (gap 2).
+
+This file pins the alignment of track 3 (#244) and track 5 (#238)
+only; it does not re-test either track's other semantics.
+
+Unified rule under test (see ``cloud/audit_sampling.py``):
+
+- Mandatory (1.0) = golden-task records + rebuild / refreeze /
+  escalation records + third-failure near-complete still-failed
+  cases. Scope from #244, expanded with #238's third-failure
+  trigger; #244's deterministic every-10th ordinary rule is gone.
+- The rest by seeded random draw: normal 0.10, critical-path
+  0.50, ceil so a non-empty group is never rounded to zero.
+  The 0.10 / 0.50 values are TEST DEFAULTS, not approved
+  production rates.
+- One pass, no double counting: mandatory records leave the
+  random pools before drawing; a multi-trigger record is sampled
+  once with all its triggers listed; correction follows the
+  (original, corrected) verdict pair, not the trigger list.
+- Correction preserves the original: the original record is never
+  mutated, the correction is a separate appended record, and the
+  corrector must differ from the original signer.
+"""
+
+import copy
+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_pushback import (  # noqa: E402
+    build_audit_correction, select_audit_sample)
+from cloud.audit_sampling import (  # noqa: E402
+    append_audit_record, build_golden_task_ids, handle_sampling_result,
+    mandatory_trigger, mandatory_triggers, select_sample)
+
+
+def _case(case_id, **kw):
+    case = {"id": case_id}
+    case.update(kw)
+    return case
+
+
+class TestMandatoryUnion(unittest.TestCase):
+    def test_all_mandatory_triggers_sampled(self):
+        cases = [
+            _case("g-push", task_id="gold-1", kind="pushback"),
+            _case("r", decision="rebuild"),
+            _case("rf", kind="refreeze"),
+            _case("esc", kind="escalate"),
+            _case("third", decision="fail_pushback", failure_count=3,
+                  root_cause="near_complete", continuation_failed=True),
+            _case("n0", decision="pass"),
+        ]
+        plan = select_sample(cases, seed="s", golden_task_ids={"gold-1"})
+        by_id = {e["case_id"]: e["triggers"] for e in plan["mandatory"]}
+        self.assertEqual(by_id["g-push"], ["golden_task"])
+        self.assertEqual(by_id["r"], ["rebuild"])
+        self.assertEqual(by_id["rf"], ["refreeze"])
+        self.assertEqual(by_id["esc"], ["escalate"])
+        self.assertEqual(by_id["third"],
+                         ["third_failure_near_complete_still_failed"])
+        self.assertIn("n0", plan["random_normal"])
+        self.assertEqual(len(plan["sampled_ids"]),
+                         len(set(plan["sampled_ids"])))
+
+    def test_blocked_quota_is_not_mandatory(self):
+        # Track 3's mandatory kind list is rebuild/refreeze/escalate
+        # only; a quota-blocked record reaches the random tiers.
+        case = _case("q", kind="blocked_quota")
+        self.assertEqual(mandatory_triggers(case), [])
+        plan = select_sample([case], seed="s")
+        self.assertEqual(plan["mandatory"], [])
+        self.assertEqual(plan["random_normal"], ["q"])
+
+    def test_golden_task_ordinary_record_is_mandatory(self):
+        # Golden membership is per task, not per record kind: even
+        # a plain pass record of a golden task is mandatory.
+        case = _case("g-pass", task_id="gold-1", kind="pushback",
+                     decision="pass")
+        plan = select_sample([case], seed="s",
+                             golden_task_ids={"gold-1"})
+        self.assertEqual(plan["mandatory"][0]["triggers"],
+                         ["golden_task"])
+
+
+class TestNoDoubleCounting(unittest.TestCase):
+    def test_golden_rebuild_sampled_once_with_all_triggers(self):
+        cases = [_case("g-re", task_id="gold-1", kind="rebuild")]
+        cases += [_case("n%d" % i, decision="pass") for i in range(20)]
+        plan = select_sample(cases, seed="s", golden_task_ids={"gold-1"})
+        self.assertEqual(plan["sampled_ids"].count("g-re"), 1)
+        self.assertEqual(plan["mandatory"][0]["triggers"],
+                         ["golden_task", "rebuild"])
+        # The mandatory record consumed no random quota: the draw
+        # is still ceil(0.10 * 20) == 2 over the ordinary pool.
+        self.assertEqual(len(plan["random_normal"]), 2)
+        self.assertNotIn("g-re", plan["random_normal"])
+        # Legacy single-trigger view reports the first trigger.
+        self.assertEqual(
+            mandatory_trigger(cases[0], {"gold-1"}), "golden_task")
+
+    def test_triage_and_later_rebuild_are_two_records_not_double(self):
+        # Deduplication is per record: a triage record and the
+        # later rebuild record for the same task are two samples.
+        cases = [
+            _case("triage", decision="fail_pushback", failure_count=3,
+                  root_cause="near_complete", continuation_failed=True),
+            _case("rebuild", decision="rebuild"),
+        ]
+        plan = select_sample(cases, seed="s")
+        self.assertEqual(sorted(plan["sampled_ids"]),
+                         ["rebuild", "triage"])
+
+
+class TestRandomTiersAndCeil(unittest.TestCase):
+    def test_ratios_and_seed_reproducibility(self):
+        cases = [_case("n%d" % i, decision="pass") for i in range(20)]
+        cases += [_case("c%d" % i, decision="pass", critical_path=True)
+                  for i in range(10)]
+        plan = select_sample(cases, seed="seed-1")
+        self.assertEqual(len(plan["random_normal"]), 2)
+        self.assertEqual(len(plan["random_critical_path"]), 5)
+        again = select_sample(cases, seed="seed-1")
+        self.assertEqual(plan["sampled_ids"], again["sampled_ids"])
+
+    def test_small_groups_never_round_to_zero(self):
+        self.assertEqual(len(select_sample(
+            [_case("only", decision="pass")], seed="s")["random_normal"]),
+            1)
+        # 9 ordinary records: the old every-10th rule drew ZERO
+        # here; ceil(0.10 * 9) == 1 under the unified rule.
+        self.assertEqual(len(select_sample(
+            [_case("n%d" % i, decision="pass") for i in range(9)],
+            seed="s")["random_normal"]), 1)
+        self.assertEqual(len(select_sample(
+            [_case("n%d" % i, decision="pass") for i in range(19)],
+            seed="s")["random_normal"]), 2)
+        self.assertEqual(len(select_sample(
+            [_case("c0", decision="pass", critical_path=True)],
+            seed="s")["random_critical_path"]), 1)
+        self.assertEqual(len(select_sample(
+            [_case("c%d" % i, decision="pass", critical_path=True)
+             for i in range(3)], seed="s")["random_critical_path"]), 2)
+
+    def test_missing_seed_rejected(self):
+        with self.assertRaises(ValueError):
+            select_sample([_case("n", decision="pass")], seed="")
+
+
+class TestGoldenSetConstruction(unittest.TestCase):
+    def test_build_golden_ids_applies_both_vocabularies(self):
+        tasks = [
+            {"id": "gold-1", "human_approved_by": "user"},
+            {"id": "sample-2", "human_approved_by": "user"},
+            {"id": "gold-2", "sample": True, "human_approved_by": "user"},
+            {"id": "gold-3", "is_example": True,
+             "human_approved_by": "user"},
+            {"id": "gold-4"},
+        ]
+        self.assertEqual(build_golden_task_ids(tasks), ["gold-1"])
+
+
+class TestAdapter(unittest.TestCase):
+    def test_adapter_returns_unified_selection_in_record_order(self):
+        records = [
+            {"task_id": "gold-1", "kind": "pushback"},
+            {"task_id": "gold-1", "kind": "fix"},
+            {"task_id": "gold-1", "kind": "rebuild"},
+            {"task_id": "t-rf", "kind": "refreeze"},
+            {"task_id": "t-esc", "kind": "escalate"},
+        ]
+        records += [{"task_id": "t-%d" % i, "kind": "pushback"}
+                    for i in range(9)]
+        before = copy.deepcopy(records)
+        golden = set(build_golden_task_ids(
+            [{"id": "gold-1", "human_approved_by": "user"}]))
+        selected = select_audit_sample(records, golden, seed="s")
+        by_record = {id(e["record"]): e for e in selected}
+        # Every golden record is mandatory, whatever its kind; the
+        # golden rebuild record carries both triggers, once.
+        self.assertEqual(by_record[id(records[0])]["reasons"],
+                         ["golden_task"])
+        self.assertEqual(by_record[id(records[1])]["reasons"],
+                         ["golden_task"])
+        self.assertEqual(by_record[id(records[2])]["reasons"],
+                         ["golden_task", "rebuild"])
+        self.assertEqual(by_record[id(records[3])]["reason"], "refreeze")
+        self.assertEqual(by_record[id(records[4])]["reason"], "escalate")
+        # 9 ordinary records contribute ceil(0.10 * 9) == 1.
+        ordinary = [e for e in selected
+                    if e["record"]["task_id"].startswith("t-")
+                    and e["record"]["task_id"] not in ("t-rf", "t-esc")]
+        self.assertEqual(len(ordinary), 1)
+        self.assertEqual(ordinary[0]["reason"], "random_normal")
+        # Selected entries keep original record order.
+        order = [records.index(e["record"]) for e in selected]
+        self.assertEqual(order, sorted(order))
+        # Input records are never mutated by selection.
+        self.assertEqual(records, before)
+        with self.assertRaises(ValueError):
+            select_audit_sample(records, golden)
+
+
+class TestCorrectionPreservesOriginal(unittest.TestCase):
+    def test_track3_correction_appends_without_mutating(self):
+        original = {"task_id": "t-1", "round": 0, "verdict": "pass",
+                    "signer": "auditor-a", "entry_hash": "h" * 64}
+        before = dict(original)
+        correction, errors = build_audit_correction(
+            original, "fail", "auditor-b", "sample re-run found ci red")
+        self.assertEqual(errors, [])
+        self.assertEqual(original, before)
+        self.assertEqual(correction["original_entry_hash"], "h" * 64)
+        self.assertEqual(correction["restored_action"],
+                         "retrigger_pushback_and_reblock")
+        log = [original, correction]
+        self.assertEqual(log[0]["verdict"], "pass")
+        _bad, bad_errors = build_audit_correction(
+            original, "fail", "auditor-a", "same signer")
+        self.assertTrue(bad_errors)
+
+    def test_track5_correction_appends_without_mutating(self):
+        case = {"id": "a", "decision": "pass"}
+        before = dict(case)
+        entry = handle_sampling_result(
+            case, "incorrect", corrected_verdict="fail_pushback",
+            corrector="auditor-b", original_signer="auditor-a")
+        # Track 5 vocabulary for the same wrong-pass pair as track
+        # 3's retrigger_pushback_and_reblock (synonyms, see the
+        # unified rule in cloud/audit_sampling.py).
+        self.assertEqual(entry["retrigger"], "reopen_audit_and_pushback")
+        self.assertEqual(case, before)
+        log = []
+        log2 = append_audit_record(log, entry)
+        self.assertEqual(log, [])
+        self.assertEqual(len(log2), 1)
+        with self.assertRaises(ValueError):
+            handle_sampling_result(
+                case, "incorrect", corrected_verdict="fail_pushback",
+                corrector="auditor-a", original_signer="auditor-a")
+
+    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"])
+
+
+if __name__ == "__main__":
+    unittest.main()