codex-app-server/cloud/audit_wiring.py

"""Wiring between three already-landed tracks (no new semantics here):

- track 1 (``cloud/merge_gate_hard.py``, PR #240): the hard merge
  gate, five checks + branch protection, machine-readable ``blocks``
  codes on its 409 response;
- track 3 (``cloud/audit_pushback.py``, PR #244): the independent
  audit + auto-pushback decision, the pushback package, and the QA
  re-run rule;
- track 4 (``cloud/ledger.py``, PR #241): the append-only hash-chain
  ledger, the only durable store in this wiring.

What this module owns, and nothing else:

1.  Translation at the boundary. Track-3 field names and enums are
    authoritative for callers; track-4 storage field names are
    authoritative on the chain. This module translates, it does not
    invent a second ledger schema:

    ============================  ===================================
    track-3 event                 chain record (track 4 action)
    ============================  ===================================
    audit verdict                 ``audit_verdict``
    pushback round (pushed)       ``audit_pushback_round`` (pushed)
    fix (landed, re-audit open)   ``audit_pushback_round`` with
                                  ``fix_diff`` + package marker
                                  ``wiring_event="fix"`` (evidence
                                  row only; see the gate reader)
    fix + re-audit                ``audit_pushback_round``
                                  (re_reviewed, ``fix_diff`` +
                                  ``re_review_result``)
    rebuild decision / done       ``audit_rebuild``
    third-failure triage          ``audit_third_failure_assessment``
    test-change single audit      ``audit_test_change``
    audit-of-audit correction     ``audit_meta_review``
    rebuild notice / confirmation ``audit_rebuild_notice``
    escalation / blocked quota /  ``audit_wiring_event`` (the one
    audit sample                  compat action added to track 4
                                  by this wiring; see below)
    ============================  ===================================

2.  CONDITIONAL mapping, done here and only here. Track 4 stores
    verdicts as ``pass`` / ``fail`` only, and track 3 says
    CONDITIONAL is not a pass. Every verdict (audit verdict and
    re-audit verdict) that arrives as ``"conditional"`` -- or as
    anything that is not exactly ``"pass"`` -- is stored as
    ``"fail"``, with the original value preserved in
    ``extra.original_verdict`` (verdict rows) or in
    ``re_review_result.original_verdict`` (pushback rows). The
    original conclusion is never overwritten: the chain keeps the
    mapped value AND the original, and corrections are separate
    ``audit_meta_review`` rows, never edits.

3.  The gate reader (``pushback_block_for_task``). It replays a
    task's chain rows in id order and returns a ``gate_block()``-
    shaped dict (``merge_blocked`` / ``release_blocked`` /
    ``blocked_by="audit_pushback"`` / ``round`` / ``failed_checks``,
    plus a ``modules`` detail map). ``cloud/server.py`` ANDs this
    with the hard gate's own five checks: either side blocks and
    the merge returns 409 with the task status unchanged. The
    pushback block applies in every gate mode -- an open audit
    pushback is not something a ``record``-mode caller can confirm
    away, and ``merge_gate_confirm`` never touches it.

4.  The gate-blocks-to-package bridge
    (``pushback_package_from_gate``). A hard-gate 409's ``blocks``
    codes are machine input to the pushback package: each code maps
    to a track-3 gate check (``ci_*`` -> ``ci_green``,
    ``directory_*`` -> ``directory_isolation``,
    ``contract_*`` -> ``contract_version``,
    ``coordinator_*`` -> ``coordinator_approval``,
    ``review_*`` / ``unresolved_review_comments`` ->
    ``unresolved_review_comments``). Branch-protection codes have
    no track-3 gate-check member, so they travel as the extra check
    name ``"branch_protection"`` in the package's failed items --
    they are never dropped, and they are never disguised as one of
    the ten checks. Callers must take the codes from the 409
    ``merge_gate`` object, never by parsing audit detail text.

Known translation losses (stated, not hidden):

- Track 3 distinguishes root causes ``"contract"`` vs ``"design"``;
  track 4 stores only ``"implementation"`` / ``"contract_design"``.
  Both contract and design map to ``"contract_design"`` on rebuild
  rows (track 3's rebuild branch treats them identically: refreeze,
  no code rebuild). The finer distinction survives on escalation
  rows, which carry the original ``root_cause`` verbatim.
- Track 3's third-failure triage verdict ``"early_refreeze"`` has
  no track-4 assessment determination (only ``early_rebuild`` /
  ``continuation_round`` exist there). It is recorded as an
  ``audit_wiring_event`` of kind ``escalate_refreeze``, never
  forced into an assessment row with the wrong determination.
- Track 3's ``"unknown"`` root cause has no track-4 value and is
  refused (``ValueError``) rather than guessed: a rebuild row must
  say which side the root cause is on.
- A fix-only row is counted by track 4's timeline summary inside
  ``pushback_rounds`` (that summary counts rows, and a round
  already spans two rows there). The gate reader skips fix rows
  for state transitions, so the count wart cannot unblock or block
  a merge by itself.

What this module does NOT do:

- It does not change any track-1 / track-3 / track-4 semantic:
  hard stays fail-closed by default, protection unknown /
  unprotected still block in hard mode, confirmation still never
  covers a directory violation, the chain stays fail-closed and
  append-only, and the cost dual-column rule is untouched.
- It does not create pushback subtasks, send notices, or run
  fixes. ``record_*`` only appends chain rows; deriving a fix
  subtask from a package is the caller's job (track 3's
  ``build_pushback_package`` already describes its shape).
- It does not gate ``deploy.py``. There is no server-side release
  endpoint in this codebase: the merge endpoint is where
  ``merge_blocked`` is enforced, and ``release_blocked`` is
  returned alongside it so a future release endpoint can enforce
  it with this same reader. Deploy remains an operator action
  outside this wiring.
"""

from cloud.audit_pushback import GATE_CHECKS
from cloud.audit_pushback import build_pushback_package as _build_package
from cloud.audit_pushback import gate_block as _gate_block
from cloud.ledger import WIRING_EVENT_ACTION
from cloud.ledger import audit_timeline as _audit_timeline
from cloud.merge_gate_hard import block_descriptions as _block_descriptions
from cloud.qa_matrix import (
    check_reaudit_reran_links as _check_reran_links,
)

# Hard-gate block code -> track-3 gate check. Branch-protection
# codes have no track-3 member; see BRANCH_PROTECTION_CHECK.
_GATE_BLOCK_TO_CHECK = {}
for _code in (
        "ci_evidence_missing", "ci_not_green", "ci_check_failed",
        "ci_required_checks_unconfigured", "ci_required_check_missing"):
    _GATE_BLOCK_TO_CHECK[_code] = "ci_green"
for _code in (
        "directory_evidence_missing", "directory_owned_dir_unknown",
        "directory_isolation_violation"):
    _GATE_BLOCK_TO_CHECK[_code] = "directory_isolation"
for _code in (
        "contract_version_missing", "contract_not_frozen",
        "contract_version_mismatch"):
    _GATE_BLOCK_TO_CHECK[_code] = "contract_version"
for _code in (
        "coordinator_approval_missing", "coordinator_not_approved"):
    _GATE_BLOCK_TO_CHECK[_code] = "coordinator_approval"
for _code in ("review_evidence_missing", "unresolved_review_comments"):
    _GATE_BLOCK_TO_CHECK[_code] = "unresolved_review_comments"
# Branch-protection codes (hard-mode codes from track 1's #240
# layer plus record-mode codes from the #209 layer).
for _code in ("branch_protection_unknown",
              "branch_protection_unprotected",
              "branch_protection_enforce",
              "branch_protection_risk_unconfirmed"):
    _GATE_BLOCK_TO_CHECK[_code] = "branch_protection"

BRANCH_PROTECTION_CHECK = "branch_protection"

# Track-3 decision actions that block merge/release (mirrors
# track 3's gate_block() membership; kept here so the package
# builder does not have to re-derive it).
BLOCKING_ACTIONS = (
    "pushback", "rebuild", "escalate", "escalate_refreeze",
    "blocked_quota", "rebuild_pending_user_confirmation",
)

# Wiring-event kinds that block (compat action rows).
_BLOCKING_WIRING_KINDS = ("escalate", "escalate_refreeze",
                          "blocked_quota")

# Triage verdict -> track-4 assessment determination. "early_refreeze"
# is deliberately absent: it goes to record_escalation instead.
_TRIAGE_TO_DETERMINATION = {
    "early_rebuild": "early_rebuild",
    "targeted_fix": "continuation_round",
}

# Track-3 root cause -> track-4 root cause. "unknown" is absent
# on purpose (refused, not guessed).
_ROOT_CAUSE_MAP = {
    "implementation": "implementation",
    "contract": "contract_design",
    "design": "contract_design",
}


def normalize_verdict(verdict):
    """Track-3 verdict -> (track-4 verdict, original).

    Exactly ``"pass"`` stays a pass; anything else -- including
    ``"conditional"`` and missing/garbage values -- maps to
    ``"fail"`` (fail-closed), with the original preserved for the
    chain. The caller stores the original alongside; this function
    never drops it silently.
    """
    if verdict == "pass":
        return "pass", "pass"
    original = verdict if isinstance(verdict, str) else None
    return "fail", original


def normalize_re_review_result(result):
    """Copy of a re-review result with its verdict normalized.

    ``CONDITIONAL`` re-audits do not pass (track 3); the stored
    verdict becomes ``"fail"`` and the original is kept under
    ``original_verdict``. A result without a verdict dict is
    returned unchanged (the ledger validates the shape).
    """
    if not isinstance(result, dict):
        return result
    copied = dict(result)
    normalized, original = normalize_verdict(result.get("verdict"))
    copied["verdict"] = normalized
    if normalized == "fail" and original != "fail":
        copied["original_verdict"] = original
    return copied


def gate_failed_checks(gate):
    """Hard-gate 409 -> (track-3 checks, extra checks), in order.

    Track-3 checks come first in GATE_CHECKS order, then the
    branch-protection extra (if any). Unknown codes pass through
    as extras rather than being dropped: a code the wiring does
    not recognise must still show up in the package.
    """
    checks = []
    extras = []
    for code in (gate or {}).get("blocks") or []:
        mapped = _GATE_BLOCK_TO_CHECK.get(code)
        if mapped == BRANCH_PROTECTION_CHECK:
            if BRANCH_PROTECTION_CHECK not in extras:
                extras.append(BRANCH_PROTECTION_CHECK)
        elif mapped is not None:
            if mapped not in checks:
                checks.append(mapped)
        elif code not in extras:
            extras.append(code)
    order = {name: index for index, name in enumerate(GATE_CHECKS)}
    checks.sort(key=lambda name: order.get(name, len(order)))
    return checks, extras


def gate_block_items(gate):
    """One pushback-package failed item per hard-gate block code.

    The package builder also appends decision ``failed_checks``
    that are not already items, so the item ``check`` values here
    use the mapped track-3 names (branch protection keeps its
    extra name) to avoid duplicate generic entries.
    """
    descriptions = _block_descriptions((gate or {}).get("blocks") or [])
    items = []
    for code, description in zip((gate or {}).get("blocks") or [],
                                 descriptions):
        check = _GATE_BLOCK_TO_CHECK.get(code, code)
        items.append({
            "check": check,
            "detail": description,
            "evidence": {
                "test_name": None,
                "file_path": None,
                "line_no": None,
                "contract_clause": None,
                "diff_range": None,
            },
            "reproduction_steps": [
                "re-run the merge gate with full evidence; "
                "the failing code is %s" % code,
            ],
            "reaudit_criterion": "re-run the full re-audit checklist "
                                 "for this check",
        })
    return items


def pushback_package_from_gate(gate, task_id, round_no,
                               decision_action="pushback",
                               previous_artifacts=None,
                               qa_failed_links=None):
    """Build a track-3 pushback package from a hard-gate 409.

    ``gate`` is the ``merge_gate`` object from the 409 response
    (not audit text). ``round_no`` is the track-3 decision round
    (0 for the first audit); the ledger round number is derived
    by the caller (see ``record_pushback``'s docstring). The
    package carries the QA failed links verbatim when given, so
    the re-audit must re-run exactly those links (see
    ``record_reaudit``'s QA guard).
    """
    if decision_action not in BLOCKING_ACTIONS and \
            decision_action != "allow":
        raise ValueError("unknown decision action: %r" % (decision_action,))
    checks, extras = gate_failed_checks(gate)
    audit = {
        "task_id": task_id,
        "verdict": "fail",
        "verdict_text": "merge gate blocked: %s" % (
            ", ".join((gate or {}).get("blocks") or ["unknown"])),
        "failed_items": gate_block_items(gate),
    }
    decision = {
        "round": round_no,
        "next_round": round_no + 1,
        "action": decision_action,
        "failed_checks": checks + extras,
    }
    return _build_package(audit, decision, previous_artifacts,
                          qa_failed_links=qa_failed_links)


def _require_module_contract(module_id, contract_version):
    """Wiring strictness: pushback rows must be attributable.

    Track 4 allows module/contract to be None on pushback rows,
    but unattributed rows cannot be scoped by the gate reader, so
    this wiring refuses them instead of writing rows the gate
    cannot reason about. Verdict/rebuild rows already require
    both in track 4 itself.
    """
    if not isinstance(module_id, str) or not module_id:
        raise ValueError("module_id is required for wiring records")
    if not isinstance(contract_version, str) or not contract_version:
        raise ValueError("contract_version is required for wiring records")


def record_audit_verdict(store, task_id, project_id, module_id,
                         contract_version, verdict, source,
                         failures=None, user_id="system", extra=None):
    """Append one audit/acceptance verdict to the chain.

    ``verdict`` uses track-3 vocabulary (``pass`` / ``fail`` /
    ``conditional``); ``"conditional"`` is stored as ``"fail"``
    with the original in ``extra.original_verdict``.
    """
    normalized, original = normalize_verdict(verdict)
    merged_extra = dict(extra or {})
    if normalized == "fail" and original != "fail":
        merged_extra.setdefault("original_verdict", original)
    return store.append_audit_verdict(
        task_id, project_id, module_id, contract_version,
        normalized, source, failures=failures, user_id=user_id,
        extra=merged_extra)


def record_pushback(store, task_id, project_id, module_id,
                    contract_version, round_no, trigger_items,
                    package, failure_source=None, failure_counts=None,
                    user_id="system"):
    """Append one pushed pushback round to the chain.

    ``round_no`` is the LEDGER round number, starting at 1 for
    the first pushback. Track 3's ``decide_pushback`` counts
    rounds already used (0 for the first audit) and returns
    ``next_round``; for a pushback action that value IS the
    ledger round number (first failure: decision round 0,
    next_round 1, ledger round_no 1). Trigger items are the
    package's failed checks/items (machine codes, not prose).
    """
    _require_module_contract(module_id, contract_version)
    return store.append_pushback_round(
        task_id, project_id, round_no, trigger_items, package,
        user_id=user_id, module_id=module_id,
        contract_version=contract_version,
        failure_source=failure_source, failure_counts=failure_counts)


def record_fix(store, task_id, project_id, module_id,
               contract_version, round_no, fix_diff, package,
               user_id="system"):
    """Append fix evidence for a round whose re-audit is still open.

    The row carries ``fix_diff`` (summary + sha, never the full
    diff) and a package copy marked ``wiring_event="fix"``. The
    gate reader skips such rows for state transitions: landing a
    fix does not unblock anything -- only a passing re-audit does.
    When fix and re-audit arrive together, use ``record_reaudit``
    with ``fix_diff`` instead of writing both rows.
    """
    _require_module_contract(module_id, contract_version)
    marked = dict(package or {})
    marked["wiring_event"] = "fix"
    return store.append_pushback_round(
        task_id, project_id, round_no, _package_trigger_items(package),
        marked, fix_diff=fix_diff, user_id=user_id,
        module_id=module_id, contract_version=contract_version)


def _package_trigger_items(package):
    """Trigger items for a fix row: the package's failed checks."""
    if isinstance(package, dict):
        checks = package.get("failed_checks")
        if isinstance(checks, list) and checks:
            return list(checks)
    return ["audit_fix"]


def check_reaudit_qa_guard(package, re_review_result, rerun_results):
    """Fail-closed QA guard for a claimed passing re-audit.

    If the package lists ``reaudit_must_rerun_links`` and the
    re-audit claims a pass, every listed link must have a passing
    re-run result; a missing or failed re-run raises ValueError
    instead of letting a false pass reach the chain. A failing
    re-audit needs no guard (it stays blocked either way).
    Returns the normalized re-review result.
    """
    normalized_result = normalize_re_review_result(re_review_result)
    if normalized_result.get("verdict") != "pass":
        return normalized_result
    required = (package or {}).get("reaudit_must_rerun_links") or []
    if not required:
        return normalized_result
    checked = _check_reran_links(package, rerun_results or {})
    if not checked["reran_all_failed_links"]:
        raise ValueError(
            "re-audit claims pass but QA failed links were not "
            "all re-run passing: missing=%s failed=%s"
            % (checked["missing"], checked["failed"]))
    return normalized_result


def record_reaudit(store, task_id, project_id, module_id,
                   contract_version, round_no, re_review_result,
                   package=None, fix_diff=None, rerun_results=None,
                   failure_source=None, failure_counts=None,
                   user_id="system"):
    """Append one re-audited pushback round to the chain.

    This is the row that can unblock: a stored ``pass`` verdict
    here clears the module's block in the gate reader, a stored
    ``fail`` (including a mapped ``conditional``) keeps it
    blocked. The QA guard in ``check_reaudit_qa_guard`` runs
    before anything is written, so a claimed pass with un-rerun
    or failed QA links never reaches the chain.
    """
    _require_module_contract(module_id, contract_version)
    guarded = check_reaudit_qa_guard(package, re_review_result,
                                     rerun_results)
    return store.append_pushback_round(
        task_id, project_id, round_no,
        _package_trigger_items(package), package or {},
        fix_diff=fix_diff, re_review_result=guarded,
        user_id=user_id, module_id=module_id,
        contract_version=contract_version,
        failure_source=failure_source, failure_counts=failure_counts)


def record_rebuild(store, task_id, project_id, module_id,
                   contract_version, rebuild_no, old_impl_ref,
                   new_impl_ref=None, root_cause="implementation",
                   status="decided", prior_failure_total=None,
                   retained_assets=None, trigger_source=None,
                   assessment_entry_id=None, user_id="system"):
    """Append one rebuild decision / completion to the chain.

    ``root_cause`` uses track-3 vocabulary; ``"contract"`` and
    ``"design"`` both store as track-4 ``"contract_design"``
    (see the module docstring). ``"unknown"`` is refused.
    A ``rebuilt`` row does NOT unblock the gate on its own: the
    new implementation still needs a passing re-audit.
    """
    mapped = _ROOT_CAUSE_MAP.get(root_cause)
    if mapped is None:
        raise ValueError(
            "root_cause must be implementation|contract|design, "
            "not %r (unknown is refused, not guessed)" % (root_cause,))
    return store.append_rebuild_decision(
        task_id, project_id, module_id, contract_version,
        rebuild_no, old_impl_ref, new_impl_ref=new_impl_ref,
        root_cause=mapped, status=status,
        prior_failure_total=prior_failure_total,
        user_id=user_id, retained_assets=retained_assets,
        trigger_source=trigger_source,
        assessment_entry_id=assessment_entry_id)


def record_third_failure_assessment(store, task_id, project_id,
                                    module_id, contract_version,
                                    triage_verdict, basis,
                                    failure_total=3,
                                    max_extra_rounds=None,
                                    user_id="system"):
    """Append one third-failure triage determination to the chain.

    Track-3 triage verdicts map as ``early_rebuild`` ->
    ``early_rebuild`` and ``targeted_fix`` -> ``continuation_round``.
    ``early_refreeze`` has no assessment determination in track 4
    and is refused here: record it with ``record_escalation``
    (kind ``escalate_refreeze``) instead.
    """
    determination = _TRIAGE_TO_DETERMINATION.get(triage_verdict)
    if determination is None:
        raise ValueError(
            "triage_verdict %r has no assessment determination; "
            "use record_escalation(kind='escalate_refreeze') for "
            "early_refreeze" % (triage_verdict,))
    return store.append_third_failure_assessment(
        task_id, project_id, module_id, contract_version,
        determination, basis, failure_total=failure_total,
        max_extra_rounds=max_extra_rounds, user_id=user_id)


def record_test_change(store, task_id, project_id, module_id,
                       contract_version, test_changes, auditor,
                       contract_refrozen, refreeze_ref=None,
                       reason="", user_id="system"):
    """Append one test-change single audit (track-4 native)."""
    return store.append_test_change_audit(
        task_id, project_id, module_id, contract_version,
        test_changes, auditor, contract_refrozen,
        refreeze_ref=refreeze_ref, reason=reason, user_id=user_id)


def record_meta_review(store, task_id, project_id, module_id,
                       contract_version, original_entry_id,
                       original_verdict, corrected_verdict,
                       misjudgment_found, reviewer, findings=None,
                       triggered_process=None, user_id="system"):
    """Append one audit-of-audit correction (history is kept).

    The correction never rewrites the original row. How the gate
    reader treats it is a resolved semantic tension, stated here
    rather than buried (see ``pushback_block_for_task``): a
    correction to ``fail`` (a wrong pass) re-blocks immediately,
    but a correction to ``pass`` (a wrong block) does NOT unblock
    by itself -- track 3's restored action for that case is
    "schedule a re-audit", not "release the gate", and a
    meta-review is a review of an audit, not a full re-audit.
    Unblocking still needs a fresh passing verdict or a passing
    re-reviewed pushback row. Track 4's failure counts are
    unaffected by this choice: counts use effective (corrected)
    verdicts to decide rebuild triggers, while the gate asks a
    different question ("has a passing re-audit been recorded
    since the block?"), so the two may legitimately disagree
    while a scheduled re-audit is still pending.
    """
    return store.append_audit_of_audit(
        task_id, project_id, module_id, contract_version,
        original_entry_id, original_verdict, corrected_verdict,
        misjudgment_found, reviewer, findings=findings,
        triggered_process=triggered_process, user_id=user_id)


def record_rebuild_notice(store, task_id, project_id, module_id,
                          contract_version, rebuild_no,
                          notification_content, notified_party,
                          confirmation_status, confirmed_by=None,
                          user_id="system"):
    """Append one rebuild-notice row (pending -> confirmed are two rows)."""
    return store.append_rebuild_notice(
        task_id, project_id, module_id, contract_version,
        rebuild_no, notification_content, notified_party,
        confirmation_status, confirmed_by=confirmed_by,
        user_id=user_id)


def record_escalation(store, task_id, project_id, module_id,
                      contract_version, kind, target=None, reason="",
                      round_no=None, root_cause=None,
                      failed_checks=None, user_id="system"):
    """Append one escalation / quota / audit-sample wiring event.

    ``kind`` is track-3 vocabulary: ``escalate``,
    ``escalate_refreeze``, ``blocked_quota``, ``audit_sample``.
    Escalation kinds block the gate; ``audit_sample`` (a sampling
    selection, not a verdict) does not change gate state. The
    original ``root_cause`` (including the contract/design
    distinction) is stored verbatim here.
    """
    _require_module_contract(module_id, contract_version)
    return store.append_wiring_event(
        task_id, project_id, module_id, contract_version, kind,
        target=target, reason=reason, round_no=round_no,
        root_cause=root_cause, failed_checks=failed_checks,
        user_id=user_id)


def _trigger_check_names(trigger_items):
    """Trigger items (str or dict) -> check-name strings."""
    names = []
    for item in trigger_items or []:
        if isinstance(item, str):
            names.append(item)
        elif isinstance(item, dict) and item.get("check"):
            names.append(item["check"])
        else:
            names.append(str(item))
    return names


def _module_key(parsed):
    """(module_id, contract_version) or None for unattributed rows."""
    module_id = parsed.get("module_id") if isinstance(parsed, dict) else None
    contract_version = parsed.get("contract_version") \
        if isinstance(parsed, dict) else None
    if module_id and contract_version:
        return (module_id, contract_version)
    return None


def pushback_block_for_task(conn, task_id):
    """Replay a task's chain into a gate_block()-shaped verdict.

    State machine, in chain-id order, per (module, contract
    version) plus one global slot for unattributed rows:

    - verdict row: its stored verdict sets blocked (pass clears,
      fail blocks).
    - meta-review row: a correction to ``fail`` blocks at once
      (a wrong pass is re-blocked); a correction to ``pass``
      changes nothing by itself -- the wrong block stays blocked
      until a fresh passing verdict or re-reviewed row arrives
      (track 3's "schedule a re-audit", not "release the gate").
      This is deliberately NOT track 4's effective-verdict
      counting: counts and gate state answer different
      questions, and conflating them would let a review of an
      audit substitute for a re-audit.
    - pushback pushed: blocked. pushback re_reviewed: blocked
      unless the stored re-audit verdict is a pass (a stored
      ``conditional`` was already mapped to fail at write time;
      anything non-pass here still blocks, fail-closed).
    - pushback fix-evidence row (package ``wiring_event="fix"``):
      no state change.
    - rebuild row (decided or rebuilt): blocked -- a rebuilt
      implementation still needs a passing re-audit.
    - wiring event escalate / escalate_refreeze / blocked_quota:
      blocked. audit_sample: no change.
    - rebuild notice pending: blocked. confirmed / rejected: no
      change (confirmation alone never unblocks).
    - assessment / test-change / meta rows: no direct change
      (meta rows act only through corrections).

    A task is blocked if the global slot or any module slot is
    blocked. A missing ``audit_ledger`` table means no chain has
    ever been written for this database, which is unblocked, not
    an error; any other read failure propagates so the server can
    fail closed (see server.py).
    """
    try:
        timeline = _audit_timeline(conn, task_id=task_id, limit=10000)
    except Exception as exc:
        if "no such table" in str(exc):
            return _gate_block("allow", [], 0)
        raise
    entries = timeline.get("entries") or []
    modules = {}
    global_state = {"blocked": False, "round": None, "checks": []}
    last_blocked_round = None

    def slot_for(parsed):
        key = _module_key(parsed)
        if key is None:
            return None, global_state
        if key not in modules:
            modules[key] = {"blocked": False, "round": None,
                            "checks": []}
        return key, modules[key]

    for entry in entries:
        parsed = entry.get("parsed")
        if not isinstance(parsed, dict):
            continue
        action = entry.get("action")
        _key, slot = slot_for(parsed)
        if action == "audit_verdict":
            blocked = parsed.get("verdict") != "pass"
            slot["blocked"] = blocked
            slot["checks"] = _trigger_check_names(
                parsed.get("failures")) if blocked else []
        elif action == "audit_meta_review":
            # Re-block on a corrected fail; a corrected pass
            # schedules a re-audit and does not unblock (see the
            # module docstring: correction != re-audit).
            if parsed.get("misjudgment_found") and \
                    parsed.get("corrected_verdict") == "fail":
                slot["blocked"] = True
                if not slot["checks"]:
                    slot["checks"] = ["audit_meta_review"]
        elif action == "audit_pushback_round":
            package = parsed.get("pushback_package") or {}
            if package.get("wiring_event") == "fix":
                continue
            round_no = parsed.get("round_no")
            if parsed.get("round_status") == "pushed":
                slot["blocked"] = True
                slot["round"] = round_no
                slot["checks"] = _trigger_check_names(
                    parsed.get("trigger_items"))
                last_blocked_round = round_no
            elif parsed.get("round_status") == "re_reviewed":
                result = parsed.get("re_review_result") or {}
                blocked = result.get("verdict") != "pass"
                slot["blocked"] = blocked
                slot["round"] = round_no
                slot["checks"] = _trigger_check_names(
                    parsed.get("trigger_items")) if blocked else []
                if blocked:
                    last_blocked_round = round_no
        elif action == "audit_rebuild":
            slot["blocked"] = True
            if not slot["checks"]:
                slot["checks"] = ["audit_rebuild"]
            last_blocked_round = slot["round"]
        elif action == WIRING_EVENT_ACTION:
            if parsed.get("wiring_kind") in _BLOCKING_WIRING_KINDS:
                slot["blocked"] = True
                checks = parsed.get("failed_checks")
                if checks:
                    slot["checks"] = list(checks)
                if parsed.get("round_no") is not None:
                    slot["round"] = parsed.get("round_no")
                    last_blocked_round = parsed.get("round_no")
        elif action == "audit_rebuild_notice":
            if parsed.get("confirmation_status") == "pending":
                slot["blocked"] = True
    blocked = global_state["blocked"] or any(
        slot["blocked"] for slot in modules.values())
    checks = []
    if global_state["blocked"]:
        checks.extend(global_state["checks"])
    for slot in modules.values():
        if slot["blocked"]:
            for check in slot["checks"]:
                if check not in checks:
                    checks.append(check)
    block = _gate_block(
        "pushback" if blocked else "allow", checks,
        last_blocked_round if last_blocked_round is not None else 0)
    block["modules"] = {
        "%s@%s" % key: {
            "blocked": slot["blocked"],
            "round": slot["round"],
            "failed_checks": list(slot["checks"]),
        }
        for key, slot in modules.items()
    }
    if global_state["blocked"]:
        block["modules"]["<global>"] = {
            "blocked": True,
            "round": global_state["round"],
            "failed_checks": list(global_state["checks"]),
        }
    return block