codex-app-server/cloud/audit_sampling.py

"""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.

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). 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.
   - ``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.
   - ``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). 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`` 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
   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_triggers(case, golden_task_ids=None):
    """Return ALL mandatory trigger names for ``case`` (possibly []).

    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).
    """
    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")):
        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,
                  golden_task_ids=None):
    """Select the audit sample. Returns a plan dict.

    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 "
                         "sample cannot be reproduced on review")
    mandatory = []
    critical = []
    normal = []
    for case in cases:
        triggers = mandatory_triggers(case, golden_task_ids)
        if triggers:
            mandatory.append((case, triggers))
        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},
        "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],
        "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,
                           corrector=None, original_signer=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``. 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",
                "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]