admin/codex-app-server
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

audit: wire pushback gate to hard merge gate and hash-chain ledger (gap 1)

Browse Source 
Track 3 decisions (audit_pushback) now really block the server merge
endpoint and really land on the track 4 hash chain, via the new
cloud/audit_wiring.py translation layer:

- POST /tasks/{id}/merge ANDs the chain pushback block
  (merge_blocked/release_blocked/blocked_by=audit_pushback/round/
  failed_checks, replayed per module+contract version) with the
  hard gate's five checks in every gate mode; either side blocks
  -> 409, status unchanged, audit_pushback_blocked appended to
  merge_gate.blocks. Ledger reads that fail (other than a missing
  audit_ledger table) fail closed to 409.
- Every pushback/fix/re-audit/rebuild/escalation step is written
  to the ledger with track 3 fields authoritative and track 4
  storage authoritative; CONDITIONAL maps to fail at the boundary
  with the original preserved, history never rewritten.
- Hard-gate 409 blocks codes feed pushback packages machine-first
  (branch protection survives as its own extra check).
- Re-audit QA guard: a claimed pass with un-rerun or failed QA
  links never reaches the chain.
- Track 4 compat: one audit_wiring_event action for escalation /
  blocked-quota / audit-sample kinds, counted in the timeline.
- Resolved tension (documented in audit_wiring.py): a meta-review
  correction to fail re-blocks at once, but a correction to pass
  does NOT unblock by itself -- track 3 schedules a re-audit,
  it does not release the gate; track 4 failure counts still use
  effective verdicts, so the two views may disagree while a
  re-audit is pending.

Tests: cloud/tests/test_audit_wiring.py (18 items), incl. a real
HTTP end-to-end chain (fail -> 409 -> fix -> re-audit pass ->
200) and a separate QA-link chain test.
This commit is contained in:
host committed 2026-11-04 20:55:49 +00:00
1 parent 04a7a4280b
commit 4c38e3c452
5 files changed
+1512 -12
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

cloud/audit_wiring.py
+746
View File
@@ -0,0 +1,746 @@
"""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