diff --git a/cloud/audit_loop_driver.py b/cloud/audit_loop_driver.py new file mode 100644 index 0000000..799791e --- /dev/null +++ b/cloud/audit_loop_driver.py @@ -0,0 +1,1575 @@ +"""Project-level audit loop driver: pure decision logic, no execution. + +This is track 1 of "the project-level audit rework mechanism joins +the production built-in automatic loop". It owns ONLY the loop +decision. It deliberately does NOT: + +- serve HTTP or touch ``cloud/server.py`` (no merge/release endpoint + wiring here; that is the next track's job, using this interface + verbatim); +- create tasks, subtasks, or fix tasks (the caller does that from + the returned ``Action``); +- read or write any database, ledger store, quota table, or cost + table (the caller persists, via the #241 hash-chain layer and + the #247 wiring layer); +- tick user acceptance (there is no ``tick_acceptance`` action + anywhere in this module; G6 stays user-only). + +It reuses the semantics already fixed by #240 / #244 / #241 / #247 +and changes none of them. Where this module and an earlier track +use the same word, the earlier track's meaning wins: + +- #240 hard merge gate: five checks, hard fail-closed by default. +- #244 independent audit + auto-pushback + rebuild + three + tightenings: pushback limit 3, rebuild limit 1 per module per + contract version, targeted-fix limit 1 per module per contract + version, third-failure triage, rebuild-only-implementation, + acceptance-rebuild needs user confirmation first. +- #241 tamper-evident ledger: hash chain, field/enum table below. +- #247 wiring: CONDITIONAL maps to fail at the boundary (original + kept), a correction to fail re-blocks at once, a correction to + pass does NOT unblock by itself (it schedules a fresh full + re-audit), and ledger failure counts use effective verdicts + (post-correction) while gate blocking waits for a new pass -- + the two views may disagree while a re-audit is pending, and + that disagreement is intentional, not a bug to paper over. + +STABLE INTERFACE (the next track implements wiring against this, +and must not silently change it): + + LoopState(module_id, contract_version, pushback_round=0, + failure_count=0, audit_fail_count=0, + acceptance_fail_count=0, rebuild_no=0, + targeted_used=False, status="auditing", + pending_confirmation=False, retained_tests=None, + retained_passed_items=None, previous_artifacts=None, + history=None) + + new_loop_state(module_id, contract_version) -> LoopState + + decide_audit(state, verdict, *, quota_check=None, + cost_threshold=None, round_history=None, + cost_estimate=None) -> Action + + after_fix(state, fix_result, *, quota_check=None, + cost_threshold=None, round_history=None) -> Action + + apply_correction(state, correction) -> Action + + confirm_rebuild(state, *, user_confirmed, quota_check=None, + cost_threshold=None, cost_estimate=None, + round_history=None) -> Action + + emit(event_type, payload, sink=None) -> dict + build_ledger_event(event_type, payload) -> dict + validate_ledger_event(event_type, payload) -> list[str] + +``Action.kind`` is one of the progression actions +``pass`` / ``pushback`` / ``rebuild`` / ``refreeze`` / ``escalate`` +/ ``pending_user_confirmation``, plus one non-progress blocker +``quota_blocked``. ``quota_blocked`` is NOT a seventh loop step: +it means "the step was not taken at all" -- it consumes no +pushback round and no rebuild, and returns the state unchanged +(including any contract-version change, which is only applied by +a decision that actually progresses). The six progression kinds +are the task's list; ``quota_blocked`` exists only because the +quota rule separately requires it, and it is kept visibly distinct +from them for exactly that reason. + +``Action`` carries everything the caller needs to execute it, and +nothing else: ``next_state`` (the caller persists this, the driver +never does), ``pushback_package`` (see below), ``ledger_events`` +(event dicts ready for ``emit`` -- the driver never writes them), +``escalation_history`` (the FULL round history on escalate, never +just the last round), ``cost_estimate``, ``violations`` (test-change +violations, which never advance the loop), and +``escalation_grants_user_acceptance`` (always False). + +Counting rules, written identically here, in the tests, and in the +summary (thresholds are ``> 3`` everywhere, never ``>= 3`` for the +4th failure): + +- Counted per (module_id, contract_version). Audit failures and + acceptance failures are counted separately (``audit_fail_count`` + / ``acceptance_fail_count``) but trigger on their SUM + (``failure_count``). +- Pushbacks cover failures 1-3. A cumulative failure count > 3 + (i.e. the 4th failure) is no longer patched: it goes to the + rebuild branch. Exactly 3 failures still push back (unless the + third-failure triage says otherwise, see below). +- One pass clears the failure counts and the pushback rounds of + the current episode, but NOT ``rebuild_no`` or ``targeted_used`` + within the same contract version. A contract-version change + clears everything: failure counts, pushback rounds, rebuild + budget, and targeted budget. +- A rebuild clears the failure counts and pushback rounds (a + fresh episode) but keeps ``rebuild_no`` (now incremented) and + ``targeted_used``. +- Third-failure triage (only when the new total is exactly 3, + before the 3rd pushback): + * contract/design structural signal -> ``refreeze`` (G2 + re-freeze, no code rebuild, no rebuild budget consumed; + contract/design wins when both signals are present); + * implementation structural signal (architecture boundary, + module responsibility, tech stack, data model, or core + structure touched) -> early ``rebuild`` (counts against + the rebuild budget); + * nearly done (at most 2 failed items, no structural or + core-touching item, previously passed items unbroken -- + ``broken_passed_items`` must be explicitly ``[]``, an + absent list disqualifies -- and passed items outnumber + failed items) -> one ``pushback`` with scope ``targeted`` + (failed items plus directly related code only), which + IS the 3rd pushback round; a further failure is the 4th + and triggers the rebuild branch. The targeted budget is + 1 per module per contract version, is not replenished by + a pass or by a rebuild, and a nearly-done case with the + budget already used is an ordinary pushback. + * anything else -> ordinary 3rd ``pushback``. +- Rebuild branch (4th failure, or early rebuild at the 3rd): + contract/design root cause -> ``refreeze``; rebuild budget + exhausted (``rebuild_no >= 1`` per version) -> ``escalate`` to + the coordinator; acceptance-triggered rebuild without user + confirmation -> ``pending_user_confirmation`` (no rebuild + starts, no budget consumed); estimated cost above the + caller-injected ``cost_threshold`` -> ``escalate`` to the user + with the full history, costs, and token info the caller + supplied (the driver never invents costs); otherwise + ``rebuild``. +- Rebuild rebuilds ONLY implementation code. Tests, passed + acceptance items, the contract, the design, audit conclusions, + and the avoidance list are ``retained_assets`` in the rebuild + package. The new implementation's first gate is the old test + set plus the retained acceptance items, all passing; a missing + result counts as a failure, and a first-gate failure counts as + an audit failure of the new episode. Deleting a test or + changing an assertion requires BOTH a G2 re-freeze AND a + separate audit record for that change; otherwise the driver + returns an ``escalate`` action carrying ``violations`` and + leaves the state unchanged (violation, not progress). +- Acceptance-triggered rebuild: ``decide_audit`` returns + ``pending_user_confirmation`` first; the caller notifies the + user (ledger event ``audit_rebuild_notice`` with + ``confirmation_status="pending"``) and only then calls + ``confirm_rebuild``. A rejection escalates to the coordinator + and consumes no rebuild budget. Confirmation itself is also a + ledger event (``"confirmed"``, a second record, never an edit + of the pending one), and the user still holds final G6 + acceptance afterwards. +- Quota and cost, both caller-injected, never read by the driver: + ``quota_check`` is a callable taking the step's estimated cost + and returning True when quota allows, False when it blocks. + It is checked BEFORE any pushback/rebuild step is returned; + a block returns ``quota_blocked`` and consumes nothing. + ``cost_threshold`` is the user-set threshold supplied by the + caller (never hard-coded here); it is applied to rebuild + steps only, preserving #244's scope -- a fix pushback is + quota-checked but does not escalate on cost alone. +- Correction semantics (#247, unchanged): ``apply_correction`` + with corrected ``fail`` (original ``pass``) re-blocks at once + (a ``pushback`` action, counted as a failure, restoring the + block). With corrected ``pass`` (original ``fail``) it does + NOT unblock: the returned action keeps ``merge_blocked`` / + ``release_blocked`` True, sets ``status="blocked"`` and + ``requires_reaudit=True``, resets the EFFECTIVE failure count + to 0 (the #241 counting view), and only a fresh passing + verdict / full re-audit via ``decide_audit`` / ``after_fix`` + can move the state to ``passed``. + +``decide_audit`` verdict shape (all fields optional except the +verdict itself; unknown fields are ignored): + + { + "source": "audit" | "acceptance", # default "audit" + "verdict": "pass" | "fail" | "conditional" | anything else, + # only exactly "pass" passes; CONDITIONAL and anything + # else map to fail, original kept in ledger events + "verdict_text": str, # audit conclusion verbatim, carried + # into the pushback package unchanged + "failures": [str | {...}], # per-item evidence: check/test + # name, file, line, contract clause, diff range, detail, + # root_cause, structural_kind, touches_core_structure, + # evidence{contract_clause} + "passed_items": [...], + "broken_passed_items": [...] | None, # None disqualifies + # a nearly-done claim (unknown, not empty) + "failed_items_by_round": [[...], ...], # for root-cause + # classification; defaults to the current round only + "root_cause_hint": "implementation" | "contract" | "design" + | "implementation_structural" | "contract_structural" + | "unknown" | None, + "previous_artifacts": [{"path", "sha256", "bytes"}], + "qa_failed_links": [...], # QA failure chain, carried into + # the package; a pass claim with qa_links_rerun_passed + # False is a fail + "qa_links_rerun_passed": bool, + "reproduction_steps": [...], + "reaudit_criteria": [...], # defaults to the full criteria + "retained_tests": [...], # retained test ids for a rebuild + "retained_test_results": {id: bool} | [{id, passed}], + "retained_acceptance_results": {id: bool} | [{id, passed}], + "retained_tests_passed": bool, # shorthand for the gate + "test_changes": [{"action": "delete" | "modify_assertion" + | "add", "test_id", "separate_audit_id"}], + "refreeze_approved": bool, + "separate_audit_ids": [...], + "user_confirmed_rebuild": bool, # acceptance rebuilds only + "cost_estimate": number, + "contract_version": str, # a change resets all counters + } + +``after_fix`` fix_result shape: + + { + "reaudit_verdict": {...}, # REQUIRED: the full re-audit + # verdict, in the decide_audit shape above. A diff + # alone never passes: a missing reaudit verdict, or + # "diff_only": True, raises ValueError. + "retained_test_results": ..., # first-gate results for a + "retained_acceptance_results": ..., # rebuilding state + "new_artifacts": [...], # carried as previous_artifacts + # test_changes / refreeze_approved / separate_audit_ids / + # cost_estimate / contract_version are forwarded to the + # reaudit verdict when the verdict does not carry them. + } + +The pushback package (``Action.pushback_package``) always carries, +when there is a failure to fix: the audit conclusion verbatim +(``audit_verdict_text``), one entry per failed item +(``failures``), ``reproduction_steps``, the FULL re-audit +criteria (``reaudit_criteria`` -- never just the failed items), +the previous round's artifact manifest with hashes +(``previous_artifacts``), and the QA failure chain +(``qa_failed_links``). A rebuild package additionally carries +``retained_assets`` and ``scope`` (rebuild this module only). + +Ledger events (``emit`` / ``build_ledger_event``) use exactly the +#241 field table (see ``LEDGER_EVENT_FIELDS``). ``emit`` maps +CONDITIONAL (and any non-pass verdict) to ``fail`` at this +boundary and keeps the original in ``extra.original_verdict`` +(or ``re_review_result.original_verdict`` on pushback rows), +exactly as #247 does; the original conclusion and the correction +are both retained, never edited. A test-change event with +deleted/modified tests and ``contract_refrozen`` False is refused +by ``emit`` (the #241 hard gate), as is any event missing a +required field or carrying an unknown event type. +""" + +from dataclasses import dataclass, field + +MAX_PUSHBACK_ROUNDS = 3 +MAX_REBUILDS_PER_CONTRACT_VERSION = 1 +MAX_TARGETED_FIX_ROUNDS_PER_CONTRACT_VERSION = 1 +NEAR_COMPLETE_MAX_FAILED_ITEMS = 2 +STRUCTURAL_KINDS = ( + "architecture_boundary", + "module_responsibility", + "tech_stack", + "data_model", +) + +ACTION_KINDS = ( + "pass", + "pushback", + "rebuild", + "refreeze", + "escalate", + "pending_user_confirmation", + "quota_blocked", +) + +STATUS_VALUES = ( + "auditing", + "pushback", + "rebuilding", + "pending_user_confirmation", + "escalated", + "passed", + "blocked", + "refreeze", +) + +FULL_REAUDIT_CRITERIA = ( + "ci_green", + "directory_isolation", + "contract_version", + "coordinator_approval", + "unresolved_review_comments", + "security_unconditional", + "regression_exact_artifact", + "demo_runnable", + "known_deviation_due_todo", + "audit_signer_separation", + "rebuild_first_gate_old_tests_and_retained_acceptance", + "qa_failed_links_rerun", +) + + +@dataclass +class LoopState: + """Loop state for one (module_id, contract_version) slot. + + ``pushback_round`` counts pushback rounds already consumed in + the current episode (0 = no pushback yet). ``failure_count`` + is the sum of ``audit_fail_count`` and ``acceptance_fail_count`` + in the current episode. ``rebuild_no`` is both the number of + rebuilds consumed in this contract version and the number of + the last rebuild (0 = none yet). ``status`` is one of + ``STATUS_VALUES``. + """ + + module_id: str + contract_version: str + pushback_round: int = 0 + failure_count: int = 0 + audit_fail_count: int = 0 + acceptance_fail_count: int = 0 + rebuild_no: int = 0 + targeted_used: bool = False + status: str = "auditing" + pending_confirmation: bool = False + retained_tests: list = field(default_factory=list) + retained_passed_items: list = field(default_factory=list) + previous_artifacts: list = field(default_factory=list) + history: list = field(default_factory=list) + + @property + def rebuilds_used(self): + return self.rebuild_no + + +@dataclass +class Action: + """One driver decision. The caller executes it; the driver + never executes anything itself. + + ``kind`` is in ``ACTION_KINDS``. ``merge_blocked`` / + ``release_blocked`` are True for every kind except ``pass``. + ``escalation_grants_user_acceptance`` is always False: an + escalation never ticks G6 acceptance. + """ + + kind: str + reason: str + next_state: LoopState + merge_blocked: bool + release_blocked: bool + pushback_package: object = None + escalation_target: object = None + escalation_history: list = field(default_factory=list) + history: list = field(default_factory=list) + ledger_events: list = field(default_factory=list) + violations: list = field(default_factory=list) + targeted_items: object = None + scope: str = "normal" + rebuild_no: object = None + requires_reaudit: bool = False + correction_action: object = None + compat_hint: object = None + cost_estimate: object = None + escalation_grants_user_acceptance: bool = False + + +def new_loop_state(module_id, contract_version): + """Fresh loop state for one module under one contract version.""" + return LoopState(module_id=module_id, contract_version=contract_version) + + +# ------------------------------------------------------------------ +# Small pure helpers +# ------------------------------------------------------------------ + +def _copy_state(state, **changes): + values = { + "module_id": state.module_id, + "contract_version": state.contract_version, + "pushback_round": state.pushback_round, + "failure_count": state.failure_count, + "audit_fail_count": state.audit_fail_count, + "acceptance_fail_count": state.acceptance_fail_count, + "rebuild_no": state.rebuild_no, + "targeted_used": state.targeted_used, + "status": state.status, + "pending_confirmation": state.pending_confirmation, + "retained_tests": list(state.retained_tests or []), + "retained_passed_items": list(state.retained_passed_items or []), + "previous_artifacts": list(state.previous_artifacts or []), + "history": list(state.history or []), + } + values.update(changes) + return LoopState(**values) + + +def _reset_for_version(state, new_version): + """Contract-version change: every counter and budget resets.""" + return _copy_state( + state, + contract_version=new_version, + pushback_round=0, + failure_count=0, + audit_fail_count=0, + acceptance_fail_count=0, + rebuild_no=0, + targeted_used=False, + pending_confirmation=False, + history=[], + ) + + +def _clear_failures(state): + """One pass: failure counts and pushback rounds clear, budgets + (rebuild_no, targeted_used) do NOT.""" + return _copy_state( + state, + pushback_round=0, + failure_count=0, + audit_fail_count=0, + acceptance_fail_count=0, + pending_confirmation=False, + ) + + +def _normalize_failures(failures): + out = [] + for item in failures or []: + if isinstance(item, str): + out.append({"check": item}) + elif isinstance(item, dict): + out.append(dict(item)) + else: + out.append({"check": str(item)}) + return out + + +def _failure_checks(failures): + checks = [] + for item in failures: + checks.append(item.get("check") or item.get("test") or item.get("id") or "?") + return checks + + +def _as_result_map(values): + if isinstance(values, dict): + return dict(values) + out = {} + for entry in values or []: + if isinstance(entry, dict): + out[entry.get("id") or entry.get("check")] = bool(entry.get("passed")) + return out + + +def _check_first_gate(retained_tests, retained_passed_items, + test_results, acceptance_results): + """Mirror of #244 check_rebuild_first_gate: a retained test or + retained acceptance item with no result fails closed.""" + test_map = _as_result_map(test_results) + acceptance_map = _as_result_map(acceptance_results) + missing = [] + failed = [] + for test_id in retained_tests or []: + if test_id not in test_map: + missing.append("test:%s" % test_id) + elif not test_map[test_id]: + failed.append("test:%s" % test_id) + for item in retained_passed_items or []: + item_id = item.get("id") or item.get("check") if isinstance(item, dict) else str(item) + if item_id not in acceptance_map: + missing.append("acceptance:%s" % item_id) + elif not acceptance_map[item_id]: + failed.append("acceptance:%s" % item_id) + return {"gate_pass": not missing and not failed, + "missing": missing, "failed": failed} + + +def _validate_test_changes_local(changes, refreeze_approved, + separate_audit_ids): + """Mirror of #244 validate_test_changes: deleting a test or + changing an assertion needs BOTH a G2 re-freeze and a separate + audit record naming that change. Adding a test never needs + either. Returns a list of violations (empty means allowed).""" + audits = set(separate_audit_ids or []) + violations = [] + for change in changes or []: + action = change.get("action") if isinstance(change, dict) else None + test_id = change.get("test_id") if isinstance(change, dict) else None + if action in ("delete", "modify_assertion"): + audit_id = change.get("separate_audit_id") if isinstance(change, dict) else None + if not refreeze_approved: + violations.append("%s:%s without contract re-freeze" % (action, test_id)) + if audit_id is None: + violations.append("%s:%s without separate audit" % (action, test_id)) + elif audit_id not in audits: + violations.append("%s:%s separate audit not recorded" % (action, test_id)) + elif action == "add": + continue + else: + violations.append("unknown test change action:%s" % action) + return violations + + +def _validate_test_changes(changes, refreeze_approved, separate_audit_ids): + try: + from cloud.audit_pushback import validate_test_changes + return list(validate_test_changes( + changes, + refreeze_approved=refreeze_approved, + separate_audit_ids=separate_audit_ids)) + except Exception: + return _validate_test_changes_local( + changes, refreeze_approved, separate_audit_ids) + + +def _classify_root_cause_local(failed_items_by_round): + """Local mirror of #244 classify_root_cause, used only when + the #244 module cannot be imported. Contract/design need at + least two rounds (same clause recurring, or proof of + unimplementability); >= 2 implementation rounds give + implementation; anything weaker is unknown.""" + contract_rounds = set() + design_rounds = set() + implementation_rounds = set() + clause_rounds = {} + proof_rounds = set() + proof_cause = None + for round_index, items in enumerate(failed_items_by_round or []): + for item in items or []: + if not isinstance(item, dict): + continue + cause = item.get("root_cause") + evidence = item.get("evidence") or {} + clause = evidence.get("contract_clause") + if cause == "contract": + contract_rounds.add(round_index) + elif cause == "design": + design_rounds.add(round_index) + elif cause == "implementation": + implementation_rounds.add(round_index) + if clause and cause in ("contract", "design"): + clause_rounds.setdefault(clause, set()).add(round_index) + if item.get("proved_unimplementable") and cause in ("contract", "design"): + proof_rounds.add(round_index) + proof_cause = cause + if len(proof_rounds) >= 2 and proof_cause: + return proof_cause + if any(len(rounds) >= 2 for rounds in clause_rounds.values()): + return "contract" + if len(contract_rounds) >= 2: + return "contract" + if len(design_rounds) >= 2: + return "design" + if len(implementation_rounds) >= 2: + return "implementation" + return "unknown" + + +def _classify_root_cause(failed_items_by_round): + try: + from cloud.audit_pushback import classify_root_cause + cause, _evidence = classify_root_cause(failed_items_by_round) + return cause + except Exception: + return _classify_root_cause_local(failed_items_by_round) + + +def _root_cause_signals(verdict, failures, failed_items_by_round): + """Return (contract_signal, implementation_structural_signal). + + Contract/design wins when both signals are present (#244 + triage order). An explicit caller hint is honoured, and a + repeated contract/design clause across rounds (via #244 + classification) also counts as a contract signal. + """ + hint = verdict.get("root_cause_hint") + contract_signal = hint in ("contract", "design", "contract_structural") + impl_structural = hint == "implementation_structural" + if failed_items_by_round: + classified = _classify_root_cause(failed_items_by_round) + if classified in ("contract", "design"): + contract_signal = True + for item in failures: + kind = item.get("structural_kind") + if kind in STRUCTURAL_KINDS: + impl_structural = True + if item.get("touches_core_structure"): + impl_structural = True + if item.get("root_cause") in ("contract", "design"): + clause = (item.get("evidence") or {}).get("contract_clause") + if clause: + rounds_with_clause = 0 + for round_items in failed_items_by_round or []: + for round_item in round_items or []: + if isinstance(round_item, dict) and ( + round_item.get("evidence") or {}).get("contract_clause") == clause: + rounds_with_clause += 1 + break + if rounds_with_clause >= 2: + contract_signal = True + return contract_signal, impl_structural + + +def _triage_third_failure(state, verdict, failures, passed_items, + failed_items_by_round): + """Third-failure triage (#244 tightening 1). Returns a dict + with verdict early_rebuild / early_refreeze / targeted_fix / + normal_pushback, reasons, and the basis evidence.""" + contract_signal, impl_structural = _root_cause_signals( + verdict, failures, failed_items_by_round) + broken = verdict.get("broken_passed_items") + basis = { + "audit_verdict_text": verdict.get("verdict_text"), + "failed_items": failures, + "passed_items": list(passed_items or []), + "broken_passed_items": broken, + "failed_items_by_round": failed_items_by_round, + "root_cause_hint": verdict.get("root_cause_hint"), + } + if contract_signal: + return {"verdict": "early_refreeze", + "reasons": ["contract/design structural signal"], + "basis": basis} + if impl_structural: + return {"verdict": "early_rebuild", + "reasons": ["implementation structural signal"], + "basis": basis} + if (len(failures) <= NEAR_COMPLETE_MAX_FAILED_ITEMS + and broken == [] + and len(passed_items or []) > len(failures)): + if state.targeted_used: + return {"verdict": "normal_pushback", + "reasons": ["targeted budget already used (1 per version)"], + "basis": basis} + return {"verdict": "targeted_fix", + "reasons": ["failed_items=%d (<= %d), no structural item, " + "no broken passed item, passed=%d > failed=%d" + % (len(failures), NEAR_COMPLETE_MAX_FAILED_ITEMS, + len(passed_items or []), len(failures))], + "basis": basis} + reasons = [] + if len(failures) > NEAR_COMPLETE_MAX_FAILED_ITEMS: + reasons.append("failed_items=%d exceeds near-complete cap %d" + % (len(failures), NEAR_COMPLETE_MAX_FAILED_ITEMS)) + if broken != []: + reasons.append("broken passed items present or unknown") + if len(passed_items or []) <= len(failures): + reasons.append("passed items do not outnumber failed items") + return {"verdict": "normal_pushback", "reasons": reasons, "basis": basis} + + +def _quota_blocked(quota_check, estimate): + """Return (blocked, reason). No quota_check means no block. + A quota_check that raises fails closed to blocked.""" + if quota_check is None: + return False, "" + try: + allowed = quota_check(estimate) + except Exception as exc: # fail-closed, never crash the loop + return True, "quota check failed closed: %s" % exc + if allowed: + return False, "" + return True, ("step exceeds quota: a fix or rebuild cost is not " + "exempt from quota") + + +def _build_package(state, verdict, failures, scope="normal", + targeted_items=None, retained_assets=None): + package = { + "module_id": state.module_id, + "contract_version": state.contract_version, + "audit_verdict_text": verdict.get("verdict_text"), + "failures": failures, + "reproduction_steps": list(verdict.get("reproduction_steps") or []), + "reaudit_criteria": list(verdict.get("reaudit_criteria") + or FULL_REAUDIT_CRITERIA), + "previous_artifacts": list( + verdict.get("previous_artifacts") + or state.previous_artifacts or []), + "qa_failed_links": list(verdict.get("qa_failed_links") or []), + "scope": scope, + "merge_blocked_until_reaudit_pass": True, + "release_blocked_until_reaudit_pass": True, + } + if targeted_items is not None: + package["targeted_items"] = list(targeted_items) + if retained_assets is not None: + package["retained_assets"] = retained_assets + return package + + +def _retained_assets(state, verdict, passed_items): + tests = verdict.get("retained_tests") + if tests is None: + tests = list(state.retained_tests or []) + passed = list(passed_items or state.retained_passed_items or []) + return { + "tests": list(tests), + "retained_tests_missing": verdict.get("retained_tests") is None + and not state.retained_tests, + "passed_acceptance_items": passed, + "contract_unchanged": True, + "design_unchanged": True, + "audit_conclusions": [ + entry.get("verdict_text") + for entry in (state.history or []) + if isinstance(entry, dict)], + } + + +# ------------------------------------------------------------------ +# Ledger events: the #241 field table, enforced at emit time. +# The driver never writes these anywhere; it builds them and the +# caller persists them through the #241 / #247 layer. +# ------------------------------------------------------------------ + +LEDGER_EVENT_FIELDS = { + # (required fields, enum constraints as field -> allowed values) + "audit_verdict": ( + ("module_id", "contract_version", "verdict", "source", "failures"), + {"verdict": ("pass", "fail"), + "source": ("audit", "acceptance")}, + ), + "audit_pushback_round": ( + ("module_id", "contract_version", "round_no", "trigger_items", + "pushback_package", "round_status"), + {"round_status": ("pushed", "re_reviewed")}, + ), + "audit_rebuild": ( + ("module_id", "contract_version", "rebuild_no", "root_cause", + "status", "retained_assets"), + {"root_cause": ("implementation", "contract_design"), + "status": ("decided", "rebuilt")}, + ), + "audit_third_failure_assessment": ( + ("module_id", "contract_version", "determination", "basis", + "failure_total", "max_extra_rounds"), + {"determination": ("early_rebuild", "continuation_round")}, + ), + "audit_test_change": ( + ("module_id", "contract_version", "test_changes", "auditor", + "contract_refrozen", "reason"), + {}, + ), + "audit_meta_review": ( + ("original_entry_id", "original_verdict", "misjudgment_found", + "reviewer", "findings", "triggered_process"), + {"triggered_process": ("re-review", "rebuild", "none")}, + ), + "audit_rebuild_notice": ( + ("module_id", "contract_version", "rebuild_no", + "notification_content", "notified_party", "confirmation_status"), + {"confirmation_status": ("pending", "confirmed", "rejected")}, + ), +} + +LEDGER_EVENT_TYPES = tuple(LEDGER_EVENT_FIELDS) + + +def validate_ledger_event(event_type, payload): + """Validate one ledger event against the #241 field table. + Returns a list of error strings (empty means valid). This + checks shape only; hash chaining is the #241 layer's job.""" + errors = [] + if event_type not in LEDGER_EVENT_FIELDS: + return ["unknown ledger event type:%s" % event_type] + if not isinstance(payload, dict): + return ["ledger payload must be a dict"] + required, enums = LEDGER_EVENT_FIELDS[event_type] + for name in required: + if payload.get(name) is None or payload.get(name) == "": + errors.append("missing required field:%s" % name) + for name, allowed in enums.items(): + value = payload.get(name) + if value is not None and value not in allowed: + errors.append("invalid %s:%r (allowed %s)" % (name, value, allowed)) + if event_type == "audit_rebuild": + if not isinstance(payload.get("rebuild_no"), int) or payload.get("rebuild_no", 0) < 1: + errors.append("rebuild_no must be >= 1") + if event_type == "audit_test_change": + changes = payload.get("test_changes") or {} + changed = list(changes.get("modified") or []) + list(changes.get("deleted") or []) + if changed and not payload.get("contract_refrozen"): + errors.append("test deletions/modifications without contract re-freeze refused") + if changed and not (changes.get("added") or changed): + errors.append("test_changes must name at least one change") + if not (changes.get("added") or changed): + errors.append("test_changes must name at least one change") + if payload.get("contract_refrozen") and not payload.get("refreeze_ref"): + errors.append("refreeze_ref required when contract_refrozen") + if event_type == "audit_meta_review": + if payload.get("misjudgment_found") and payload.get("corrected_verdict") is None: + errors.append("corrected_verdict required when misjudgment_found") + if not payload.get("misjudgment_found") and payload.get("corrected_verdict") is not None: + errors.append("corrected_verdict must be None without a misjudgment") + if event_type == "audit_rebuild_notice": + if payload.get("confirmation_status") == "confirmed" and not payload.get("confirmed_by"): + errors.append("confirmed_by required when confirmation_status=confirmed") + content = payload.get("notification_content") + if content is None or content == "" or content == {}: + errors.append("notification_content must be non-empty") + return errors + + +def _map_conditional(payload): + """CONDITIONAL mapping at the boundary (#247): any verdict + that is not exactly 'pass' is stored as 'fail', with the + original value preserved. Returns a copied payload.""" + mapped = dict(payload) + verdict = mapped.get("verdict") + if verdict is not None and verdict != "pass": + extra = dict(mapped.get("extra") or {}) + # Never overwrite an original already recorded upstream + # (e.g. _verdict_event kept "conditional" here). + extra.setdefault("original_verdict", verdict) + mapped["extra"] = extra + mapped["verdict"] = "fail" + review = mapped.get("re_review_result") + if isinstance(review, dict): + review_verdict = review.get("verdict") + if review_verdict is not None and review_verdict != "pass": + review = dict(review) + review.setdefault("original_verdict", review_verdict) + review["verdict"] = "fail" + mapped["re_review_result"] = review + return mapped + + +def build_ledger_event(event_type, payload): + """Validate and (for verdicts) CONDITIONAL-map one event. + Returns {"event_type": ..., "payload": ...}. Raises + ValueError on any validation error. No side effects. The + CONDITIONAL mapping runs BEFORE enum validation, exactly as + #247: anything that is not exactly "pass" is stored as + "fail", so a conditional verdict validates as a fail row + with its original value kept in extra.""" + mapped = _map_conditional(payload) if isinstance(payload, dict) else payload + errors = validate_ledger_event(event_type, mapped) + if errors: + raise ValueError("invalid ledger event %s: %s" % (event_type, "; ".join(errors))) + return {"event_type": event_type, "payload": mapped} + + +def emit(event_type, payload, sink=None): + """Ledger event interface: emit(event_type, payload). + + Validates against the #241 table, maps CONDITIONAL to fail + (original kept), and -- only when the caller supplies a + ``sink`` callable -- forwards the event to it (the caller + persists it through the #241 chain; with no sink, the event + is returned without any side effect). The original conclusion + and any correction are both retained: corrections are + separate ``audit_meta_review`` events, never edits. + """ + event = build_ledger_event(event_type, payload) + if sink is not None: + sink(event) + return event + + +def _verdict_event(state, verdict, mapped_verdict, source, failures): + return build_ledger_event("audit_verdict", { + "module_id": state.module_id, + "contract_version": state.contract_version, + "verdict": mapped_verdict, + "source": source, + "failures": failures, + "extra": {"original_verdict": verdict.get("verdict")}, + }) + + +def _pushback_event(state, round_no, failures, package, source, counts): + return build_ledger_event("audit_pushback_round", { + "module_id": state.module_id, + "contract_version": state.contract_version, + "round_no": round_no, + "trigger_items": _failure_checks(failures), + "pushback_package": package, + "round_status": "pushed", + "failure_source": source, + "failure_counts": counts, + }) + + +def _rebuild_event(state, rebuild_no, trigger_source, retained_assets, + status="decided"): + return build_ledger_event("audit_rebuild", { + "module_id": state.module_id, + "contract_version": state.contract_version, + "rebuild_no": rebuild_no, + "root_cause": "implementation", + "status": status, + "retained_assets": retained_assets, + "trigger_source": trigger_source, + "old_impl_ref": None, + "new_impl_ref": None, + }) + + +def _assessment_event(state, determination, basis, failure_total, + max_extra_rounds): + return build_ledger_event("audit_third_failure_assessment", { + "module_id": state.module_id, + "contract_version": state.contract_version, + "determination": determination, + "basis": basis, + "failure_total": failure_total, + "max_extra_rounds": max_extra_rounds, + }) + + +def _notice_event(state, rebuild_no, content, party, status, confirmed_by=None): + payload = { + "module_id": state.module_id, + "contract_version": state.contract_version, + "rebuild_no": rebuild_no, + "notification_content": content, + "notified_party": party, + "confirmation_status": status, + } + if confirmed_by is not None: + payload["confirmed_by"] = confirmed_by + return build_ledger_event("audit_rebuild_notice", payload) + + +def _counts_snapshot(state): + return { + "audit_fail_count": state.audit_fail_count, + "acceptance_fail_count": state.acceptance_fail_count, + "total_fail_count": state.failure_count, + "pushback_rounds_used": state.pushback_round, + "rebuilds_used": state.rebuild_no, + } + + +# ------------------------------------------------------------------ +# The driver itself +# ------------------------------------------------------------------ + +def _rounds_with_current(failed_items_by_round, failures): + rounds = [list(r) for r in (failed_items_by_round or [])] + if not rounds or rounds[-1] != failures: + rounds.append(list(failures)) + return rounds + + +def decide_audit(state, verdict, *, quota_check=None, cost_threshold=None, + round_history=None, cost_estimate=None): + """One audit/acceptance verdict -> the next loop Action. + + Pure: returns an Action with ``next_state``; the caller + persists the state, creates any fix/rebuild subtask, blocks + merge/release, and emits the ledger events. See the module + docstring for the verdict shape and the counting rules. + """ + if not isinstance(state, LoopState): + raise TypeError("state must be a LoopState") + if not isinstance(verdict, dict): + raise TypeError("verdict must be a dict") + source = verdict.get("source", "audit") + if source not in ("audit", "acceptance"): + raise ValueError("unknown failure source: %r" % (source,)) + + # Test-change violations never advance the loop, and never + # consume anything: the state is returned exactly as passed. + test_changes = verdict.get("test_changes") + if test_changes is not None: + violations = _validate_test_changes( + test_changes, + refreeze_approved=bool(verdict.get("refreeze_approved")), + separate_audit_ids=verdict.get("separate_audit_ids")) + if violations: + return Action( + kind="escalate", + reason="test change violation: %s; not advancing" + % "; ".join(violations), + next_state=state, + merge_blocked=True, + release_blocked=True, + escalation_target="coordinator", + violations=violations, + escalation_grants_user_acceptance=False, + ) + + failures = _normalize_failures(verdict.get("failures")) + passed_items = list(verdict.get("passed_items") or []) + raw_verdict = verdict.get("verdict") + new_version = verdict.get("contract_version") + version_changed = new_version is not None and new_version != state.contract_version + + # Rebuild first gate: in a rebuilding episode, the old test + # set plus retained acceptance items must all pass, and a + # missing result counts as a failure (#244 tightening 2). + gate_fail_items = [] + if state.status == "rebuilding" and not version_changed and ( + state.retained_tests or state.retained_passed_items): + if (verdict.get("retained_test_results") is not None + or verdict.get("retained_acceptance_results") is not None): + gate = _check_first_gate( + state.retained_tests, state.retained_passed_items, + verdict.get("retained_test_results"), + verdict.get("retained_acceptance_results")) + elif isinstance(verdict.get("retained_tests_passed"), bool): + gate = _check_first_gate( + state.retained_tests, state.retained_passed_items, + {tid: verdict["retained_tests_passed"] + for tid in state.retained_tests}, + verdict.get("retained_acceptance_results")) + else: + gate = {"gate_pass": False, + "missing": (["test:%s" % t for t in state.retained_tests] + + ["acceptance:%s" % ( + item.get("id") or item.get("check") + if isinstance(item, dict) else str(item)) + for item in state.retained_passed_items]), + "failed": []} + if not gate["gate_pass"]: + gate_fail_items = [{ + "check": "rebuild_first_gate", + "detail": "missing: %s; failed: %s" + % (gate["missing"], gate["failed"]), + }] + failures = failures + gate_fail_items + if verdict.get("qa_failed_links") and verdict.get("qa_links_rerun_passed") is False: + failures = failures + [{ + "check": "qa_failed_links_rerun", + "detail": "QA failed links were not re-run passing", + }] + + is_pass = (raw_verdict == "pass" and not failures) + + if is_pass: + if version_changed: + base = _reset_for_version(state, new_version) + else: + base = _clear_failures(state) + history_basis = list(round_history) if round_history is not None else list(base.history) + summary = {"failure_total": 0, "source": source, "verdict": "pass", + "verdict_text": verdict.get("verdict_text"), + "failures": [], "action": "pass"} + next_state = _copy_state( + base, status="passed", + retained_tests=list(verdict.get("retained_tests") + or base.retained_tests or []), + retained_passed_items=list(passed_items or base.retained_passed_items or []), + previous_artifacts=list(verdict.get("previous_artifacts") + or base.previous_artifacts or []), + history=history_basis + [summary]) + return Action( + kind="pass", + reason="full re-audit passed; failure counts cleared, " + "rebuild/targeted budgets kept", + next_state=next_state, + merge_blocked=False, + release_blocked=False, + history=next_state.history, + ledger_events=[_verdict_event(next_state, verdict, "pass", source, [])], + escalation_grants_user_acceptance=False, + ) + + # Failure path: quota is checked BEFORE any pushback/rebuild + # step is returned, and a block consumes nothing at all. + estimate = cost_estimate if cost_estimate is not None else verdict.get("cost_estimate") + blocked, quota_reason = _quota_blocked(quota_check, estimate) + if blocked: + return Action( + kind="quota_blocked", + reason=quota_reason, + next_state=state, + merge_blocked=True, + release_blocked=True, + compat_hint="audit_wiring_event:blocked_quota", + cost_estimate=estimate, + escalation_grants_user_acceptance=False, + ) + + if version_changed: + base = _reset_for_version(state, new_version) + else: + base = state + counted = _copy_state( + base, + failure_count=base.failure_count + 1, + audit_fail_count=base.audit_fail_count + (1 if source == "audit" else 0), + acceptance_fail_count=base.acceptance_fail_count + (1 if source == "acceptance" else 0)) + total = counted.failure_count + history_basis = list(round_history) if round_history is not None else list(base.history) + failed_items_by_round = _rounds_with_current( + verdict.get("failed_items_by_round"), failures) + contract_signal, _impl_signal = _root_cause_signals( + verdict, failures, failed_items_by_round) + user_confirmed = bool(verdict.get("user_confirmed_rebuild")) + + def _summary(action, scope="normal", rebuild_no=None): + return {"failure_total": total, "source": source, "verdict": "fail", + "verdict_text": verdict.get("verdict_text"), + "failures": failures, "action": action, + "pushback_scope": scope, "rebuild_no": rebuild_no} + + def _refreeze_action(triage=None): + history_new = history_basis + [_summary("refreeze")] + next_state = _copy_state(counted, status="refreeze", history=history_new) + return Action( + kind="refreeze", + reason="root cause is in the contract/design itself; " + "no code rebuild is started, G2 re-freeze instead" + + ((" (triage: %s)" % "; ".join(triage["reasons"])) + if triage else ""), + next_state=next_state, + merge_blocked=True, + release_blocked=True, + escalation_target="coordinator", + history=history_new, + ledger_events=[_verdict_event(next_state, verdict, "fail", source, failures)], + compat_hint="audit_wiring_event:escalate_refreeze", + escalation_grants_user_acceptance=False, + ) + + def _escalate_action(target, reason, history_new): + next_state = _copy_state(counted, status="escalated", + pending_confirmation=False, history=history_new) + return Action( + kind="escalate", + reason=reason, + next_state=next_state, + merge_blocked=True, + release_blocked=True, + escalation_target=target, + escalation_history=history_new, + history=history_new, + ledger_events=[_verdict_event(next_state, verdict, "fail", source, failures)], + compat_hint="audit_wiring_event:escalate", + cost_estimate=estimate, + escalation_grants_user_acceptance=False, + ) + + def _rebuild_branch(early=False): + """Shared rebuild branch for the 4th failure and for an + early rebuild at the 3rd failure.""" + if contract_signal: + return _refreeze_action() + if counted.rebuild_no >= MAX_REBUILDS_PER_CONTRACT_VERSION: + return _escalate_action( + "coordinator", + "rebuild budget exhausted (%d rebuild(s) per module " + "per contract version) and failures still exceed 3" + % MAX_REBUILDS_PER_CONTRACT_VERSION, + history_basis + [_summary("escalate")]) + if source == "acceptance" and not user_confirmed: + proposed = counted.rebuild_no + 1 + retained = _retained_assets(counted, verdict, passed_items) + notice_content = { + "failure_reasons": _failure_checks(failures), + "rebuild_scope": {"rebuild_only": counted.module_id}, + "retained_assets": retained, + "first_gate": "old tests plus retained acceptance items " + "must all pass before any new full audit", + } + history_new = history_basis + [_summary("pending_user_confirmation", + rebuild_no=proposed)] + next_state = _copy_state( + counted, status="pending_user_confirmation", + pending_confirmation=True, history=history_new) + package = _build_package( + counted, verdict, failures, scope="rebuild_pending", + retained_assets=retained) + return Action( + kind="pending_user_confirmation", + reason="rebuild triggered by acceptance failures: user " + "notice and confirmation required before the " + "rebuild starts; nothing is rebuilt silently " + "and no rebuild budget is consumed yet", + next_state=next_state, + merge_blocked=True, + release_blocked=True, + pushback_package=package, + escalation_target="user", + history=history_new, + ledger_events=[ + _verdict_event(next_state, verdict, "fail", source, failures), + _notice_event(next_state, proposed, notice_content, + "user", "pending"), + ], + rebuild_no=proposed, + escalation_grants_user_acceptance=False, + ) + if (cost_threshold is not None and estimate is not None + and estimate > cost_threshold): + return _escalate_action( + "user", + "estimated rebuild cost exceeds the user-set " + "cost-escalation threshold; escalating before " + "burning it, with the full round history attached", + history_basis + [_summary("escalate")]) + rebuild_no = counted.rebuild_no + 1 + retained = _retained_assets(counted, verdict, passed_items) + history_new = history_basis + [_summary("rebuild", rebuild_no=rebuild_no)] + retained["audit_conclusions"] = [ + entry.get("verdict_text") for entry in history_new + if isinstance(entry, dict)] + next_state = _copy_state( + counted, + failure_count=0, audit_fail_count=0, acceptance_fail_count=0, + pushback_round=0, rebuild_no=rebuild_no, + status="rebuilding", pending_confirmation=False, + retained_tests=list(retained["tests"]), + retained_passed_items=list(retained["passed_acceptance_items"]), + previous_artifacts=list(verdict.get("previous_artifacts") + or counted.previous_artifacts or []), + history=history_new) + package = _build_package( + counted, verdict, failures, scope="rebuild", + retained_assets=retained) + return Action( + kind="rebuild", + reason=("structural triage at the 3rd failure: rebuild " + "early, without waiting for a 4th failure" + if early else + "cumulative failures exceed 3; patching stops, " + "the module is rebuilt from the frozen contract"), + next_state=next_state, + merge_blocked=True, + release_blocked=True, + pushback_package=package, + history=history_new, + ledger_events=[ + _verdict_event(next_state, verdict, "fail", source, failures), + _rebuild_event(next_state, rebuild_no, source, retained), + ], + rebuild_no=rebuild_no, + cost_estimate=estimate, + escalation_grants_user_acceptance=False, + ) + + def _pushback_action(scope, targeted_items, triage=None): + next_round = counted.pushback_round + 1 + history_new = history_basis + [_summary("pushback", scope=scope)] + next_state = _copy_state( + counted, pushback_round=next_round, status="pushback", + targeted_used=True if scope == "targeted" else counted.targeted_used, + previous_artifacts=list(verdict.get("previous_artifacts") + or counted.previous_artifacts or []), + history=history_new) + package = _build_package( + counted, verdict, failures, scope=scope, + targeted_items=targeted_items) + events = [ + _verdict_event(next_state, verdict, "fail", source, failures), + _pushback_event(next_state, next_round, failures, package, + source, _counts_snapshot(next_state)), + ] + if triage is not None: + events.append(_assessment_event( + next_state, "continuation_round", triage["basis"], + total, 1)) + return Action( + kind="pushback", + reason=("targeted fix round: scope limited to the failed " + "items and directly related code; no unrelated " + "refactor" if scope == "targeted" else + "audit failed: %s" % ", ".join(_failure_checks(failures)) + if failures else "audit verdict is not an unconditional pass"), + next_state=next_state, + merge_blocked=True, + release_blocked=True, + pushback_package=package, + history=history_new, + ledger_events=events, + targeted_items=targeted_items, + scope=scope, + escalation_grants_user_acceptance=False, + ) + + if total > MAX_PUSHBACK_ROUNDS: + # 4th failure (total > 3): no more patching. + return _rebuild_branch(early=False) + if total == MAX_PUSHBACK_ROUNDS: + triage = _triage_third_failure( + counted, verdict, failures, passed_items, failed_items_by_round) + if triage["verdict"] == "early_refreeze": + return _refreeze_action(triage=triage) + if triage["verdict"] == "early_rebuild": + action = _rebuild_branch(early=True) + if action.kind == "rebuild": + assessment = _assessment_event( + action.next_state, "early_rebuild", triage["basis"], + total, 0) + action = Action( + kind=action.kind, reason=action.reason, + next_state=action.next_state, + merge_blocked=action.merge_blocked, + release_blocked=action.release_blocked, + pushback_package=action.pushback_package, + escalation_target=action.escalation_target, + escalation_history=action.escalation_history, + history=action.history, + ledger_events=action.ledger_events + [assessment], + violations=action.violations, + targeted_items=action.targeted_items, + scope=action.scope, rebuild_no=action.rebuild_no, + requires_reaudit=action.requires_reaudit, + correction_action=action.correction_action, + compat_hint=action.compat_hint, + cost_estimate=action.cost_estimate, + escalation_grants_user_acceptance=False) + return action + if triage["verdict"] == "targeted_fix": + return _pushback_action("targeted", _failure_checks(failures), triage=triage) + return _pushback_action("normal", None, triage=triage) + # Failures 1-2: ordinary pushback, no triage yet. + return _pushback_action("normal", None) + + +def after_fix(state, fix_result, *, quota_check=None, cost_threshold=None, + round_history=None): + """A fix landing -> automatic FULL re-audit, never a diff-only + check. ``fix_result`` must carry ``reaudit_verdict`` (the full + re-audit verdict in the ``decide_audit`` shape); a missing + verdict, or ``diff_only`` True, raises ValueError fail-closed. + The fix itself is never counted: only the re-audit verdict + counts, via ``decide_audit``. A re-audit failure therefore + enters the next pushback/rebuild decision, and a re-audit pass + clears the episode's failure counts.""" + if not isinstance(fix_result, dict): + raise TypeError("fix_result must be a dict") + if fix_result.get("diff_only"): + raise ValueError("a diff alone never passes: a full re-audit " + "verdict is required after a fix") + reaudit = (fix_result.get("reaudit_verdict") + or fix_result.get("reaudit") + or fix_result.get("verdict")) + if reaudit is None: + raise ValueError("after_fix requires a full reaudit verdict; " + "a diff alone never passes") + if not isinstance(reaudit, dict): + raise TypeError("reaudit verdict must be a dict") + verdict = dict(reaudit) + for key in ("retained_test_results", "retained_acceptance_results", + "retained_tests_passed", "test_changes", + "refreeze_approved", "separate_audit_ids", "cost_estimate", + "contract_version"): + if key not in verdict and key in fix_result: + verdict[key] = fix_result[key] + if "previous_artifacts" not in verdict: + verdict["previous_artifacts"] = ( + fix_result.get("new_artifacts") + or fix_result.get("previous_artifacts") + or state.previous_artifacts) + return decide_audit(state, verdict, quota_check=quota_check, + cost_threshold=cost_threshold, + round_history=round_history) + + +def apply_correction(state, correction): + """Apply an audit-of-audit correction (#247 semantics, unchanged). + + ``correction``: {"original_verdict", "corrected_verdict", + "reviewer"|"corrector", "original_entry_id", "reason", + "findings", "source"}. The original record is never edited: + the returned action carries an ``audit_meta_review`` ledger + event, and the caller appends it as a new row. + + - corrected fail (original pass): re-blocks AT ONCE. The + action is ``pushback`` (retrigger pushback and re-block), + counted as one failure of the correction's source + (default audit), with the pushback round incremented to + match. ``requires_reaudit`` is True: a fix plus a full + re-audit must follow. + - corrected pass (original fail): does NOT unblock. The + action keeps ``merge_blocked``/``release_blocked`` True, + sets ``status="blocked"`` and ``requires_reaudit=True``, + and resets the EFFECTIVE failure count to 0 (the #241 + counting view: the fail no longer counts). Only a fresh + passing verdict / full re-audit via ``decide_audit`` / + ``after_fix`` can move the state to ``passed``. This is + the intentional divergence between the counting view + (count 0) and the gate view (still blocked), pinned by + test, not a contradiction to resolve by unblocking. + """ + if not isinstance(correction, dict): + raise TypeError("correction must be a dict") + original = correction.get("original_verdict") + corrected = correction.get("corrected_verdict") + original_mapped = "pass" if original == "pass" else "fail" + corrected_mapped = "pass" if corrected == "pass" else "fail" + if original_mapped not in ("pass", "fail") or corrected_mapped not in ("pass", "fail"): + raise ValueError("correction verdicts must be pass/fail") + if corrected_mapped == original_mapped: + raise ValueError("correction matches original verdict") + reviewer = correction.get("reviewer") or correction.get("corrector") + if not reviewer: + raise ValueError("correction requires a reviewer/corrector") + source = correction.get("source", "audit") + if source not in ("audit", "acceptance"): + raise ValueError("unknown failure source: %r" % (source,)) + meta = build_ledger_event("audit_meta_review", { + "original_entry_id": correction.get("original_entry_id"), + "original_verdict": original_mapped, + "corrected_verdict": corrected_mapped, + "misjudgment_found": True, + "reviewer": reviewer, + "findings": list(correction.get("findings") or []), + "triggered_process": "re-review", + }) + if corrected_mapped == "fail": + next_state = _copy_state( + state, + failure_count=state.failure_count + 1, + audit_fail_count=state.audit_fail_count + (1 if source == "audit" else 0), + acceptance_fail_count=state.acceptance_fail_count + (1 if source == "acceptance" else 0), + pushback_round=state.pushback_round + 1, + status="pushback") + return Action( + kind="pushback", + reason="meta-review corrected a pass to fail: re-block " + "at once, retrigger pushback", + next_state=next_state, + merge_blocked=True, + release_blocked=True, + requires_reaudit=True, + correction_action="retrigger_pushback_and_reblock", + ledger_events=[meta], + escalation_grants_user_acceptance=False, + ) + next_state = _copy_state( + state, + failure_count=0, audit_fail_count=0, acceptance_fail_count=0, + pushback_round=0, status="blocked") + return Action( + kind="pushback", + reason="meta-review corrected a fail to pass: this does NOT " + "unblock by itself; a fresh full re-audit is required " + "(schedule reaudit, not release)", + next_state=next_state, + merge_blocked=True, + release_blocked=True, + scope="reaudit_only", + requires_reaudit=True, + correction_action="schedule_reaudit", + ledger_events=[meta], + escalation_grants_user_acceptance=False, + ) + + +def confirm_rebuild(state, *, user_confirmed, quota_check=None, + cost_threshold=None, cost_estimate=None, + round_history=None): + """Resolve a pending acceptance-triggered rebuild. + + Only valid while ``state.pending_confirmation`` is True. + Rejection escalates to the coordinator and consumes no + rebuild budget. Confirmation re-runs the rebuild branch + checks (quota, budget, cost threshold) and, when they pass, + returns the ``rebuild`` action and consumes one rebuild. + The confirmation itself is a second ``audit_rebuild_notice`` + record (``"confirmed"``), never an edit of the pending one, + and the user still holds final G6 acceptance afterwards + (this function never ticks acceptance). + """ + if not isinstance(state, LoopState): + raise TypeError("state must be a LoopState") + if not state.pending_confirmation: + raise ValueError("no pending acceptance rebuild to confirm") + proposed = state.rebuild_no + 1 + history_basis = list(round_history) if round_history is not None else list(state.history) + if not user_confirmed: + next_state = _copy_state(state, status="escalated", + pending_confirmation=False, + history=history_basis) + return Action( + kind="escalate", + reason="user rejected the acceptance-triggered rebuild; " + "escalating to the coordinator without consuming " + "rebuild budget", + next_state=next_state, + merge_blocked=True, + release_blocked=True, + escalation_target="coordinator", + escalation_history=history_basis, + history=history_basis, + ledger_events=[_notice_event( + state, proposed, + {"decision": "user rejected rebuild"}, "user", "rejected")], + escalation_grants_user_acceptance=False, + ) + estimate = cost_estimate + blocked, quota_reason = _quota_blocked(quota_check, estimate) + if blocked: + # Quota block consumes nothing and leaves the pending + # confirmation pending, so the caller can retry. + return Action( + kind="quota_blocked", + reason=quota_reason, + next_state=state, + merge_blocked=True, + release_blocked=True, + compat_hint="audit_wiring_event:blocked_quota", + cost_estimate=estimate, + escalation_grants_user_acceptance=False, + ) + if state.rebuild_no >= MAX_REBUILDS_PER_CONTRACT_VERSION: + next_state = _copy_state(state, status="escalated", + pending_confirmation=False, + history=history_basis) + return Action( + kind="escalate", + reason="rebuild budget exhausted at confirmation time", + next_state=next_state, + merge_blocked=True, + release_blocked=True, + escalation_target="coordinator", + escalation_history=history_basis, + history=history_basis, + ledger_events=[_notice_event( + state, proposed, + {"decision": "confirmation blocked by exhausted budget"}, + "user", "rejected")], + escalation_grants_user_acceptance=False, + ) + if (cost_threshold is not None and estimate is not None + and estimate > cost_threshold): + next_state = _copy_state(state, status="escalated", + pending_confirmation=False, + history=history_basis) + return Action( + kind="escalate", + reason="estimated rebuild cost exceeds the user-set " + "cost-escalation threshold at confirmation time", + next_state=next_state, + merge_blocked=True, + release_blocked=True, + escalation_target="user", + escalation_history=history_basis, + history=history_basis, + cost_estimate=estimate, + escalation_grants_user_acceptance=False, + ) + retained = _retained_assets(state, {}, state.retained_passed_items) + next_state = _copy_state( + state, + failure_count=0, audit_fail_count=0, acceptance_fail_count=0, + pushback_round=0, rebuild_no=proposed, + status="rebuilding", pending_confirmation=False, + retained_tests=list(retained["tests"]), + retained_passed_items=list(retained["passed_acceptance_items"]), + history=history_basis) + package = _build_package( + state, {"verdict_text": "acceptance rebuild confirmed by user"}, + [], scope="rebuild", retained_assets=retained) + return Action( + kind="rebuild", + reason="user confirmed the acceptance-triggered rebuild; " + "rebuild starts now, user still holds final G6 " + "acceptance afterwards", + next_state=next_state, + merge_blocked=True, + release_blocked=True, + pushback_package=package, + history=history_basis, + ledger_events=[ + _notice_event(state, proposed, + {"decision": "user confirmed rebuild"}, + "user", "confirmed", confirmed_by="user"), + _rebuild_event(next_state, proposed, "acceptance", retained), + ], + rebuild_no=proposed, + cost_estimate=estimate, + escalation_grants_user_acceptance=False, + ) diff --git a/cloud/tests/test_audit_loop_driver.py b/cloud/tests/test_audit_loop_driver.py new file mode 100644 index 0000000..98e78d0 --- /dev/null +++ b/cloud/tests/test_audit_loop_driver.py @@ -0,0 +1,578 @@ +"""Audit loop driver: pure decision logic for the project-level +audit rework loop (track 1 of the production built-in loop). + +Pins the driver interface the next (wiring) track must use +verbatim, and the counting rules shared by #244 / #241 / #247: +pushbacks cover failures 1-3, the 4th failure (> 3, never >= 3) +rebuilds, rebuilds are 1 per module per contract version, the +3rd failure is triaged, a pass clears failures but not budgets, +a contract-version change clears everything, quota blocks consume +nothing, a correction to pass does not unblock, and ledger events +match the #241 field table. + +Style follows the other cloud tests: unittest, stdlib only, import +with a sys.path fallback so the file also runs standalone. +""" + +import sys +import unittest +from pathlib import Path + +try: + from cloud.audit_loop_driver import ( + ACTION_KINDS, + LEDGER_EVENT_TYPES, + Action, + LoopState, + apply_correction, + build_ledger_event, + confirm_rebuild, + decide_audit, + emit, + new_loop_state, + validate_ledger_event, + after_fix, + ) +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.audit_loop_driver import ( + ACTION_KINDS, + LEDGER_EVENT_TYPES, + Action, + LoopState, + apply_correction, + build_ledger_event, + confirm_rebuild, + decide_audit, + emit, + new_loop_state, + validate_ledger_event, + after_fix, + ) + + +def fail_verdict(**overrides): + verdict = { + "source": "audit", + "verdict": "fail", + "verdict_text": "audit fail: ci red", + "failures": [{"check": "ci_green", "detail": "ci red"}], + "passed_items": [], + "previous_artifacts": [{"path": "app.py", "sha256": "a" * 64}], + "reproduction_steps": ["run ci"], + } + verdict.update(overrides) + return verdict + + +def pass_verdict(**overrides): + verdict = { + "source": "audit", + "verdict": "pass", + "verdict_text": "audit pass", + "failures": [], + "passed_items": [], + } + verdict.update(overrides) + return verdict + + +def near_complete_verdict(**overrides): + verdict = fail_verdict( + failures=[{"check": "edge-case"}], + passed_items=[{"check": "p1"}, {"check": "p2"}, {"check": "p3"}], + broken_passed_items=[], + verdict_text="audit fail: one edge case left", + ) + verdict.update(overrides) + return verdict + + +def drive(state, verdict, **kwargs): + return decide_audit(state, verdict, **kwargs) + + +class PushbackCountingTestCase(unittest.TestCase): + def test_fail_pushbacks_and_blocks(self): + state = new_loop_state("m1", "v1") + action = drive(state, fail_verdict()) + self.assertEqual(action.kind, "pushback") + self.assertTrue(action.merge_blocked) + self.assertTrue(action.release_blocked) + self.assertEqual(action.next_state.failure_count, 1) + self.assertEqual(action.next_state.pushback_round, 1) + self.assertEqual(action.next_state.status, "pushback") + package = action.pushback_package + self.assertEqual(package["audit_verdict_text"], "audit fail: ci red") + self.assertEqual(package["failures"][0]["check"], "ci_green") + self.assertEqual(package["reproduction_steps"], ["run ci"]) + # Full re-audit criteria, never just the failed items. + self.assertIn("ci_green", package["reaudit_criteria"]) + self.assertIn("audit_signer_separation", package["reaudit_criteria"]) + self.assertEqual(package["previous_artifacts"][0]["path"], "app.py") + self.assertEqual( + [e["event_type"] for e in action.ledger_events], + ["audit_verdict", "audit_pushback_round"]) + + def test_exactly_three_still_pushes_back(self): + state = new_loop_state("m1", "v1") + for _ in range(3): + action = drive(state, fail_verdict()) + state = action.next_state + self.assertEqual(action.kind, "pushback") + self.assertEqual(action.next_state.failure_count, 3) + self.assertEqual(action.next_state.pushback_round, 3) + + def test_fourth_failure_rebuilds(self): + state = new_loop_state("m1", "v1") + for _ in range(3): + state = drive(state, fail_verdict()).next_state + action = drive(state, fail_verdict(retained_tests=["t1", "t2"])) + self.assertEqual(action.kind, "rebuild") + self.assertEqual(action.rebuild_no, 1) + self.assertEqual(action.next_state.rebuild_no, 1) + self.assertEqual(action.next_state.failure_count, 0) + self.assertEqual(action.next_state.pushback_round, 0) + self.assertEqual(action.next_state.status, "rebuilding") + self.assertEqual(action.next_state.retained_tests, ["t1", "t2"]) + retained = action.pushback_package["retained_assets"] + self.assertEqual(retained["tests"], ["t1", "t2"]) + self.assertTrue(retained["contract_unchanged"]) + self.assertTrue(retained["design_unchanged"]) + + def test_rebuild_budget_exhausted_escalates(self): + state = new_loop_state("m1", "v1") + for _ in range(3): + state = drive(state, fail_verdict()).next_state + state = drive(state, fail_verdict()).next_state # rebuild 1 + self.assertEqual(state.rebuild_no, 1) + for _ in range(3): + state = drive(state, fail_verdict()).next_state + action = drive(state, fail_verdict()) + self.assertEqual(action.kind, "escalate") + self.assertEqual(action.escalation_target, "coordinator") + self.assertEqual(action.next_state.rebuild_no, 1) + self.assertEqual(action.next_state.status, "escalated") + self.assertFalse(action.escalation_grants_user_acceptance) + self.assertFalse(hasattr(action, "tick_acceptance")) + # Full history, not just the last round. + texts = [entry.get("verdict_text") + for entry in action.escalation_history] + self.assertIn("audit fail: ci red", texts) + self.assertGreaterEqual(len(action.escalation_history), 4) + + def test_audit_and_acceptance_counts_combine_but_stay_separate(self): + state = new_loop_state("m1", "v1") + state = drive(state, fail_verdict()).next_state + state = drive(state, fail_verdict(source="acceptance")).next_state + action = drive(state, fail_verdict()) + self.assertEqual(action.next_state.failure_count, 3) + self.assertEqual(action.next_state.audit_fail_count, 2) + self.assertEqual(action.next_state.acceptance_fail_count, 1) + + def test_pass_clears_failures_keeps_rebuild_budget(self): + state = new_loop_state("m1", "v1") + for _ in range(3): + state = drive(state, fail_verdict()).next_state + state = drive(state, fail_verdict()).next_state # rebuild 1 + state = drive(state, fail_verdict()).next_state # failure 1, new episode + action = drive(state, pass_verdict()) + self.assertEqual(action.kind, "pass") + self.assertFalse(action.merge_blocked) + self.assertEqual(action.next_state.failure_count, 0) + self.assertEqual(action.next_state.pushback_round, 0) + self.assertEqual(action.next_state.rebuild_no, 1) + self.assertEqual(action.next_state.status, "passed") + + def test_contract_version_change_clears_everything(self): + state = new_loop_state("m1", "v1") + for _ in range(3): + state = drive(state, fail_verdict()).next_state + state = drive(state, fail_verdict()).next_state # rebuild 1 + state = drive(state, fail_verdict()).next_state # failure 1, v1 + action = drive(state, fail_verdict(contract_version="v2")) + self.assertEqual(action.kind, "pushback") + self.assertEqual(action.next_state.contract_version, "v2") + self.assertEqual(action.next_state.failure_count, 1) + self.assertEqual(action.next_state.pushback_round, 1) + self.assertEqual(action.next_state.rebuild_no, 0) + self.assertFalse(action.next_state.targeted_used) + + +class ThirdFailureTriageTestCase(unittest.TestCase): + def _two_failures(self): + state = new_loop_state("m1", "v1") + for _ in range(2): + state = drive(state, fail_verdict()).next_state + return state + + def test_structural_third_failure_early_rebuilds(self): + state = self._two_failures() + action = drive(state, fail_verdict( + failures=[{"check": "arch", + "structural_kind": "architecture_boundary"}], + verdict_text="audit fail: architecture boundary crossed")) + self.assertEqual(action.kind, "rebuild") + self.assertEqual(action.rebuild_no, 1) + assessments = [e for e in action.ledger_events + if e["event_type"] == "audit_third_failure_assessment"] + self.assertEqual(len(assessments), 1) + self.assertEqual(assessments[0]["payload"]["determination"], "early_rebuild") + self.assertEqual(assessments[0]["payload"]["failure_total"], 3) + self.assertEqual(assessments[0]["payload"]["max_extra_rounds"], 0) + + def test_contract_root_third_failure_goes_to_refreeze(self): + state = self._two_failures() + action = drive(state, fail_verdict(root_cause_hint="contract")) + self.assertEqual(action.kind, "refreeze") + self.assertEqual(action.next_state.status, "refreeze") + self.assertEqual(action.next_state.rebuild_no, 0) + self.assertEqual(action.compat_hint, "audit_wiring_event:escalate_refreeze") + # No rebuild event, and no assessment row forced into the + # wrong determination (see #247: early_refreeze has no + # assessment determination). + kinds = [e["event_type"] for e in action.ledger_events] + self.assertNotIn("audit_rebuild", kinds) + self.assertNotIn("audit_third_failure_assessment", kinds) + + def test_near_complete_gets_one_targeted_round_then_rebuilds(self): + state = self._two_failures() + action = drive(state, near_complete_verdict()) + self.assertEqual(action.kind, "pushback") + self.assertEqual(action.scope, "targeted") + self.assertEqual(action.targeted_items, ["edge-case"]) + self.assertTrue(action.next_state.targeted_used) + self.assertEqual(action.next_state.pushback_round, 3) + assessments = [e for e in action.ledger_events + if e["event_type"] == "audit_third_failure_assessment"] + self.assertEqual(assessments[0]["payload"]["determination"], + "continuation_round") + self.assertEqual(assessments[0]["payload"]["max_extra_rounds"], 1) + # One more failure is the 4th: rebuild, not a 4th pushback. + action = drive(action.next_state, fail_verdict()) + self.assertEqual(action.kind, "rebuild") + + def test_targeted_budget_not_replenished_by_pass(self): + state = self._two_failures() + state = drive(state, near_complete_verdict()).next_state + self.assertTrue(state.targeted_used) + state = drive(state, pass_verdict()).next_state + self.assertTrue(state.targeted_used) + for _ in range(2): + state = drive(state, fail_verdict()).next_state + action = drive(state, near_complete_verdict()) + self.assertEqual(action.kind, "pushback") + self.assertEqual(action.scope, "normal") + self.assertIsNone(action.targeted_items) + + def test_unknown_broken_passed_disqualifies_near_complete(self): + state = self._two_failures() + # broken_passed_items absent (unknown, not []) -> normal. + action = drive(state, fail_verdict( + failures=[{"check": "edge-case"}], + passed_items=[{"check": "p1"}, {"check": "p2"}, {"check": "p3"}])) + self.assertEqual(action.scope, "normal") + + +class RebuildGateAndAcceptanceTestCase(unittest.TestCase): + def _rebuilding_state(self): + state = new_loop_state("m1", "v1") + for _ in range(3): + state = drive(state, fail_verdict()).next_state + return drive(state, fail_verdict(retained_tests=["t1", "t2"])).next_state + + def test_old_test_gate_missing_result_counts_as_failure(self): + state = self._rebuilding_state() + action = after_fix(state, {"reaudit_verdict": pass_verdict()}) + self.assertEqual(action.kind, "pushback") + self.assertEqual(action.next_state.failure_count, 1) + self.assertEqual(action.next_state.status, "pushback") + checks = [item.get("check") for item in action.pushback_package["failures"]] + self.assertIn("rebuild_first_gate", checks) + + def test_old_test_gate_failing_result_counts_as_failure(self): + state = self._rebuilding_state() + action = after_fix(state, { + "reaudit_verdict": pass_verdict(), + "retained_test_results": {"t1": True, "t2": False}, + }) + self.assertEqual(action.kind, "pushback") + self.assertEqual(action.next_state.failure_count, 1) + + def test_old_test_gate_passing_allows_reaudit_pass(self): + state = self._rebuilding_state() + action = after_fix(state, { + "reaudit_verdict": pass_verdict(), + "retained_test_results": {"t1": True, "t2": True}, + }) + self.assertEqual(action.kind, "pass") + self.assertEqual(action.next_state.status, "passed") + + def _acceptance_pending(self): + state = new_loop_state("m1", "v1") + for _ in range(3): + state = drive(state, fail_verdict(source="acceptance")).next_state + return drive(state, fail_verdict(source="acceptance")) + + def test_acceptance_rebuild_unconfirmed_does_not_start(self): + action = self._acceptance_pending() + self.assertEqual(action.kind, "pending_user_confirmation") + self.assertTrue(action.next_state.pending_confirmation) + self.assertEqual(action.next_state.rebuild_no, 0) + self.assertEqual(action.next_state.status, "pending_user_confirmation") + notices = [e for e in action.ledger_events + if e["event_type"] == "audit_rebuild_notice"] + self.assertEqual(notices[0]["payload"]["confirmation_status"], "pending") + + def test_acceptance_rebuild_confirmed_then_starts(self): + pending = self._acceptance_pending().next_state + action = confirm_rebuild(pending, user_confirmed=True) + self.assertEqual(action.kind, "rebuild") + self.assertEqual(action.rebuild_no, 1) + self.assertEqual(action.next_state.status, "rebuilding") + self.assertFalse(action.next_state.pending_confirmation) + notices = [e for e in action.ledger_events + if e["event_type"] == "audit_rebuild_notice"] + self.assertEqual(notices[0]["payload"]["confirmation_status"], "confirmed") + self.assertEqual(notices[0]["payload"]["confirmed_by"], "user") + + def test_acceptance_rebuild_rejected_escalates_without_budget(self): + pending = self._acceptance_pending().next_state + action = confirm_rebuild(pending, user_confirmed=False) + self.assertEqual(action.kind, "escalate") + self.assertEqual(action.escalation_target, "coordinator") + self.assertEqual(action.next_state.rebuild_no, 0) + + def test_confirm_without_pending_raises(self): + state = new_loop_state("m1", "v1") + with self.assertRaises(ValueError): + confirm_rebuild(state, user_confirmed=True) + + +class AfterFixTestCase(unittest.TestCase): + def test_after_fix_pass_clears(self): + state = new_loop_state("m1", "v1") + state = drive(state, fail_verdict()).next_state + action = after_fix(state, {"reaudit_verdict": pass_verdict()}) + self.assertEqual(action.kind, "pass") + self.assertEqual(action.next_state.failure_count, 0) + + def test_after_fix_failing_reaudit_goes_to_next_pushback(self): + state = new_loop_state("m1", "v1") + state = drive(state, fail_verdict()).next_state + action = after_fix(state, {"reaudit_verdict": fail_verdict()}) + self.assertEqual(action.kind, "pushback") + self.assertEqual(action.next_state.failure_count, 2) + self.assertEqual(action.next_state.pushback_round, 2) + + def test_after_fix_diff_only_raises(self): + state = new_loop_state("m1", "v1") + state = drive(state, fail_verdict()).next_state + with self.assertRaises(ValueError): + after_fix(state, {"diff": "--- a/x\n+++ b/x\n"}) + with self.assertRaises(ValueError): + after_fix(state, {"diff_only": True, + "reaudit_verdict": pass_verdict()}) + + +class CorrectionTestCase(unittest.TestCase): + def test_correction_to_pass_does_not_unblock(self): + state = new_loop_state("m1", "v1") + for _ in range(2): + state = drive(state, fail_verdict()).next_state + action = apply_correction(state, { + "original_verdict": "fail", "corrected_verdict": "pass", + "reviewer": "auditor-b", "original_entry_id": "e1", + "reason": "second look: evidence was sufficient", + }) + # Counting view: effective count is 0 ... + self.assertEqual(action.next_state.failure_count, 0) + # ... but the gate view is still blocked until a fresh pass. + self.assertTrue(action.merge_blocked) + self.assertTrue(action.release_blocked) + self.assertEqual(action.next_state.status, "blocked") + self.assertTrue(action.requires_reaudit) + self.assertEqual(action.correction_action, "schedule_reaudit") + self.assertEqual(action.ledger_events[0]["event_type"], "audit_meta_review") + # Only a fresh passing verdict unblocks. + action = drive(action.next_state, pass_verdict()) + self.assertEqual(action.kind, "pass") + self.assertFalse(action.merge_blocked) + + def test_correction_to_fail_reblocks_at_once(self): + state = drive(new_loop_state("m1", "v1"), pass_verdict()).next_state + action = apply_correction(state, { + "original_verdict": "pass", "corrected_verdict": "fail", + "reviewer": "auditor-b", "original_entry_id": "e1", + "reason": "missed a failing check", + }) + self.assertEqual(action.kind, "pushback") + self.assertTrue(action.merge_blocked) + self.assertEqual(action.next_state.failure_count, 1) + self.assertEqual(action.correction_action, "retrigger_pushback_and_reblock") + + +class QuotaAndCostTestCase(unittest.TestCase): + def test_quota_block_consumes_nothing(self): + state = new_loop_state("m1", "v1") + for _ in range(2): + state = drive(state, fail_verdict()).next_state + action = drive(state, fail_verdict(), + quota_check=lambda estimate: False) + self.assertEqual(action.kind, "quota_blocked") + self.assertIs(action.next_state, state) + self.assertEqual(action.next_state.failure_count, 2) + self.assertEqual(action.next_state.pushback_round, 2) + + def test_quota_block_at_rebuild_consumes_no_rebuild(self): + state = new_loop_state("m1", "v1") + for _ in range(3): + state = drive(state, fail_verdict()).next_state + action = drive(state, fail_verdict(), + quota_check=lambda estimate: False) + self.assertEqual(action.kind, "quota_blocked") + self.assertEqual(action.next_state.rebuild_no, 0) + self.assertEqual(action.next_state.failure_count, 3) + + def test_quota_check_exception_fails_closed(self): + state = new_loop_state("m1", "v1") + def _boom(estimate): + raise RuntimeError("quota service down") + action = drive(state, fail_verdict(), quota_check=_boom) + self.assertEqual(action.kind, "quota_blocked") + self.assertEqual(action.next_state.failure_count, 0) + + def test_cost_threshold_escalates_rebuild_to_user(self): + state = new_loop_state("m1", "v1") + for _ in range(3): + state = drive(state, fail_verdict()).next_state + action = drive(state, fail_verdict(cost_estimate=500), + cost_threshold=100) + self.assertEqual(action.kind, "escalate") + self.assertEqual(action.escalation_target, "user") + self.assertEqual(action.next_state.rebuild_no, 0) + self.assertEqual(action.cost_estimate, 500) + self.assertGreaterEqual(len(action.escalation_history), 4) + self.assertFalse(action.escalation_grants_user_acceptance) + + def test_cost_threshold_not_applied_to_pushback(self): + # #244 scope preserved: the threshold gates rebuild steps, + # a fix pushback is quota-checked only. + state = new_loop_state("m1", "v1") + action = drive(state, fail_verdict(cost_estimate=500), + cost_threshold=100) + self.assertEqual(action.kind, "pushback") + + +class TestChangeTestCase(unittest.TestCase): + def test_test_deletion_violation_does_not_advance(self): + state = new_loop_state("m1", "v1") + state = drive(state, fail_verdict()).next_state + action = drive(state, fail_verdict(test_changes=[ + {"action": "delete", "test_id": "t1"}])) + self.assertEqual(action.kind, "escalate") + self.assertTrue(action.violations) + self.assertEqual(action.next_state.failure_count, 1) + self.assertEqual(action.next_state.pushback_round, 1) + + def test_test_deletion_allowed_with_refreeze_and_audit(self): + state = new_loop_state("m1", "v1") + action = drive(state, fail_verdict(test_changes=[ + {"action": "delete", "test_id": "t1", + "separate_audit_id": "audit-9"}], + refreeze_approved=True, separate_audit_ids=["audit-9"])) + self.assertEqual(action.kind, "pushback") + self.assertEqual(action.violations, []) + + +class LedgerEventTestCase(unittest.TestCase): + def test_all_seven_event_types_validate(self): + cases = { + "audit_verdict": {"module_id": "m1", "contract_version": "v1", + "verdict": "pass", "source": "audit", "failures": []}, + "audit_pushback_round": {"module_id": "m1", "contract_version": "v1", + "round_no": 1, "trigger_items": ["ci_green"], + "pushback_package": {}, "round_status": "pushed"}, + "audit_rebuild": {"module_id": "m1", "contract_version": "v1", + "rebuild_no": 1, "root_cause": "implementation", + "status": "decided", "retained_assets": {}}, + "audit_third_failure_assessment": { + "module_id": "m1", "contract_version": "v1", + "determination": "continuation_round", "basis": {}, + "failure_total": 3, "max_extra_rounds": 1}, + "audit_test_change": {"module_id": "m1", "contract_version": "v1", + "test_changes": {"added": ["t9"], "modified": [], + "deleted": []}, + "auditor": "auditor-a", + "contract_refrozen": False, "reason": "add only"}, + "audit_meta_review": {"original_entry_id": "e1", + "original_verdict": "fail", + "corrected_verdict": "pass", + "misjudgment_found": True, + "reviewer": "auditor-b", "findings": [], + "triggered_process": "re-review"}, + "audit_rebuild_notice": {"module_id": "m1", "contract_version": "v1", + "rebuild_no": 1, + "notification_content": "notice text", + "notified_party": "user", + "confirmation_status": "pending"}, + } + for event_type, payload in cases.items(): + self.assertEqual(validate_ledger_event(event_type, payload), [], + event_type) + event = emit(event_type, payload) + self.assertEqual(event["event_type"], event_type) + self.assertEqual(set(LEDGER_EVENT_TYPES), set(cases)) + + def test_unknown_event_type_and_missing_field_rejected(self): + self.assertTrue(validate_ledger_event("nope", {})) + with self.assertRaises(ValueError): + emit("audit_verdict", {"module_id": "m1"}) + with self.assertRaises(ValueError): + emit("audit_verdict", {"module_id": "m1", "contract_version": "v1", + "verdict": "pass", "source": "audit"}) + + def test_conditional_maps_to_fail_and_keeps_original(self): + event = emit("audit_verdict", { + "module_id": "m1", "contract_version": "v1", + "verdict": "conditional", "source": "audit", "failures": []}) + self.assertEqual(event["payload"]["verdict"], "fail") + self.assertEqual(event["payload"]["extra"]["original_verdict"], + "conditional") + # A driver verdict of conditional is a fail, not a pass. + action = drive(new_loop_state("m1", "v1"), + fail_verdict(verdict="conditional")) + self.assertEqual(action.kind, "pushback") + + def test_test_change_hard_gate_in_emit(self): + payload = {"module_id": "m1", "contract_version": "v1", + "test_changes": {"added": [], "modified": ["t1"], + "deleted": []}, + "auditor": "auditor-a", "contract_refrozen": False, + "reason": "drop a test"} + with self.assertRaises(ValueError): + emit("audit_test_change", payload) + payload["contract_refrozen"] = True + payload["refreeze_ref"] = "freeze-2" + emit("audit_test_change", payload) # no raise + + def test_rebuild_notice_confirmed_needs_confirmer(self): + payload = {"module_id": "m1", "contract_version": "v1", + "rebuild_no": 1, "notification_content": "notice", + "notified_party": "user", "confirmation_status": "confirmed"} + with self.assertRaises(ValueError): + emit("audit_rebuild_notice", payload) + payload["confirmed_by"] = "user" + emit("audit_rebuild_notice", payload) # no raise + + def test_action_kinds_are_bounded(self): + for kind in ACTION_KINDS: + self.assertIn(kind, ("pass", "pushback", "rebuild", "refreeze", + "escalate", "pending_user_confirmation", + "quota_blocked")) + action = drive(new_loop_state("m1", "v1"), fail_verdict()) + self.assertIn(action.kind, ACTION_KINDS) + + +if __name__ == "__main__": + unittest.main()