"""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, )