"""Independent-audit auto-pushback: decision, package, ledger interface. Pure functions, standard library only. This module does NOT wire itself into ``cloud/server.py`` or ``cloud/worker.py`` on purpose: the hard merge gate (track 1) owns the server-side blocking check, and the tamper-evident ledger (track 4) owns durable storage. What this module does own, and what its tests pin down, is the decision itself, so the rule "audit fail -> automatic pushback, merge/release blocked" cannot drift into "audit fail -> wait for someone to reopen it manually". Decision vocabulary (``decide_pushback`` returns one of these): - ``allow``: every gate check passes, the audit signature is valid, the security verdict is unconditional, and no blocking deviation exists. Only then may merge/release proceed. - ``pushback``: automatic, no human nod required. The original task stays ``completed`` (its record and its passed artifacts are NOT rewound or overwritten); a fix/optimisation subtask is derived automatically and merge/release stay blocked until a full re-audit passes. This is pushback form (2) in the skill text; form (1) (rewind the original task to a re-runnable state) was rejected there because rewinding destroys the completed record, invites re-running side effects, and lets a fix silently drop or overwrite artifacts that already passed. - ``rebuild``: the pushback budget is exhausted (see "Rebuild" below) and the root cause is in the implementation, not in the contract or design. The module is rebuilt from the frozen contract; this is the FIRST escalation branch, not a parallel loop. - ``escalate``: no further automatic step applies -- either the rebuild budget is also exhausted, or the rebuild cost estimate exceeds the user-set cost-escalation threshold. The escalation target is the coordinator (or the user, for a cost-threshold escalation), with the full round history attached -- never just the last round. Escalation does NOT grant user acceptance: G6 acceptance can only be ticked by the user, and an escalated audit never substitutes for it. - ``escalate_refreeze``: the root cause is in the contract or design itself (see "Root-cause gate" below). No code rebuild is started automatically; the G2 re-freeze flow is triggered and the case is escalated to the coordinator for human review, who must bring any new freeze to the user. A refreeze never ticks G6 acceptance either. - ``blocked_quota``: the fix itself would exceed quota. The fix is interrupted under the normal quota rules, not exempted because it is a fix. This consumes no pushback round. ``round_no`` counts pushback rounds already used for this task (0 = the first audit). A failed audit with ``round_no < 3`` pushes back; with ``round_no >= 3`` it escalates. The limit value (3) matches the full-auto CI auto-fix limit, but the counters are NOT shared: CI auto-fix rounds happen before audit, audit pushback rounds happen after audit fails, and one must never consume the other's budget. ``decide_pushback`` accepts ``ci_auto_fix_rounds`` only as informational input and ignores it for the limit -- that is tested. Fail-closed rules, all tested: - A missing gate check is a failure, never a pass. - ``CONDITIONAL`` is not a pass: a ``conditional`` audit verdict, or a ``conditional`` security verdict, always pushes back. - A known deviation only excuses a failure while it has a follow-up todo AND a due date that has not expired without follow-up. A deviation with no due date, no todo, or an expired due date with no follow-up blocks. - Regression must run the exact artifact, verified the #211 way: manifest hash vs recomputed hash must match, and a fixture substitute is reported as "unverified", never as a pass. - A demo must be verifiably runnable and must run the app's own entry point; a generic file server standing in for the app (``runs_app_entry=False``) is not a runnable demo. - The audit signer must differ from the executor, the coordinator approver, and the QA tester. If the signer equals any of them, the audit conclusion itself is invalid and counts as a failure. Auditor / QA / coordinator / executor / merge-executor are five separate functions; one person (or thread) never holds two of them on the same task. Gate interface for track 1 (proposed contract, to be confirmed when track 1 lands -- see the skill section): the hard gate reads ``gate_block()`` output, i.e. ``merge_blocked`` / ``release_blocked`` plus ``blocked_by="audit_pushback"``, ``round`` and ``failed_checks``. The gate must AND this with its own checks: an audit pass does not unblock a failing track-1 check, and a track-1 pass does not unblock an open audit pushback. Until track 1 wires this in, this module alone does not block any server endpoint -- that wiring gap is stated in the PR summary, not hidden. Ledger interface for track 4 (proposed contract, to be confirmed when track 4 lands): ``build_ledger_entry`` produces one JSON-able record per pushback / fix / re-audit / escalation event, chained by ``prev_hash`` / ``entry_hash`` (sha256 over the canonical JSON of the entry without ``entry_hash``). ``validate_ledger_chain`` recomputes every hash and link. If track 4 lands a different hash format, an adapter is required -- the field names here are the proposal, not a claim that track 4 already speaks them. The timeline rule itself does not wait for track 4: every round's entry is appended and never rewritten, so the timeline always shows "audit fail -> auto pushback -> fix -> re-audit", never only the final passing record. Rebuild (the over-3 escalation branch -- this supersedes the earlier, vaguer "exhausted budget -> escalate" wording; the escalation after a rebuild is a second, distinct step): Counting is per module AND per contract version. Audit failures (independent audit / hard gate) and acceptance failures (G6 user acceptance / QA acceptance) are counted separately by source but trigger on their SUM for the module. One pass resets the module's failure count to zero (the next failure starts counting again); a contract-version change (re-freeze) resets the failure count AND the rebuild count, because the module is then judged against a different contract. A pass does NOT reset the rebuild count within the same contract version: otherwise a pass-fail cycle could rebuild the same module forever, one rebuild per episode. Threshold, written identically here and in the skill text (tests pin the boundary both ways): fix pushbacks cover failures 1-3 (``round_no`` 0, 1, 2 at decision time). A cumulative failure count > 3 -- i.e. the 4th failure, ``round_no >= 3`` at decision time -- is no longer patched: it triggers the rebuild branch below. "Over three" is ``> 3``, never ``>= 3``; exactly 3 failures still push back for a fix. Rebuild meaning: ONLY the failing module is rebuilt. Other modules, the frozen contract, the design, and other modules' passed artifacts are untouched, and directory isolation is unchanged. The module is re-implemented from the frozen contract. The old implementation is NOT deleted from history: it is kept via its commit / tag / archive reference in git history and in the audit record for review, and only the working tree is replaced by the new implementation. The previous rounds' audit conclusions, failure reasons, and passed-item list travel into the new implementation's task package as an avoidance list, so the rebuild cannot repeat the same error and still claim a fresh start. Root-cause gate (checked BEFORE deciding rebuild vs refreeze): if the continuous failures are caused by the contract or design itself -- e.g. a self-contradictory contract, an unimplementable design, a wrong technology-stack choice -- rebuilding the code treats the symptom, not the cause, so no code rebuild is started automatically. Instead the G2 re-freeze flow is triggered and the case escalates to the coordinator. ``classify_root_cause`` implements the testable standard: at least two distinct rounds point to the same contract/design clause with a contract/design root cause, or at least two distinct rounds prove the contract cannot be implemented as written. Fewer than two such rounds leaves the cause "unknown", and an unknown cause defaults to the rebuild branch: refreeze needs positive evidence, and the rebuild branch is the reversible first step. A rebuild package always carries the avoidance list, so even an unknown-cause rebuild does not start blind. Rebuild budget: at most ``MAX_REBUILDS_PER_CONTRACT_VERSION`` (1) rebuild per module per contract version. Rationale for 1 rather than 2: a rebuild is the most expensive step (a full re-implementation); one rebuild is a genuine second chance, and if the new implementation also fails more than 3 times, the problem is almost certainly in the contract/design or in capability, not in patchable code, so a second automatic rebuild would mostly burn cost while masking that. After a rebuild, a fresh failure count starts for the new implementation, and if it again exceeds 3, the case escalates (coordinator) with no further automatic rebuild. Rebuild cost is costed and quota-checked like any task: exceeding quota blocks the rebuild (``blocked_quota``, consuming no rebuild), and a rebuild cost estimate above the user-set cost-escalation threshold escalates to the USER before anything is burned -- never silently. Three tightenings (added after the pushback/rebuild rules; they do NOT change the counting above, they refine what happens at its edges -- the single counting story is restated at the end of this section so the two cannot drift apart): Tightening 1 -- third-failure triage (``triage_third_failure``). 3 is a default, not a truth. At the 3rd failure (cumulative fail count == 3, before the rebuild threshold), a root-cause triage runs first and its result is written into the audit record: (a) structural error -- an item labelled with a structural kind (architecture boundary, module responsibility, tech stack, data model) or the repeated-clause / contract-design signal from ``classify_root_cause`` -- rebuilds EARLY, without waiting for the 4th failure, and that early rebuild consumes the rebuild budget; (b) nearly done -- at most ``NEAR_COMPLETE_MAX_FAILED_ITEMS`` failed items, none touching core structure, no previously passed item broken, and a pass rate over the known items -- gets ONE targeted fix round whose scope is exactly the failed items plus directly related code, never an unrelated refactor; (c) anything else is an ordinary 3rd pushback. A targeted round's failure counts in the cumulative total (it is the 3rd pushback, not a free round): if it fails, the total becomes 4 and the ordinary rebuild branch fires. The targeted budget is 1 per module per contract version, independent of the rebuild budget; a rebuild does not replenish it. Tightening 2 -- rebuild keeps tests and passed assets (``build_rebuild_package`` / ``check_rebuild_first_gate`` / ``validate_test_changes``). A rebuild replaces ONLY the module's implementation code. Tests, passed acceptance items, the contract, the design, audit conclusions, and the avoidance list are all retained and listed in the rebuild package; a package built without a retained test list is flagged ``retained_tests_missing``. The new implementation's FIRST gate is the old test set plus the retained acceptance items, all passing; missing results fail closed, and a first-gate failure counts as an audit failure for the new implementation's episode. Deleting a test or changing an assertion requires a contract change via the G2 re-freeze flow AND a separate audit record; without both it is a violation, not a quiet edit. Tightening 3 -- the audit itself is audited. (i) Independent re-runs use exact artifacts (#211 triple-hash), never reports alone. (ii) Golden tasks (human-approved only; example/sample files are rejected by ``validate_golden_task_set``) and rebuild/refreeze/escalation records are always sampled, and ordinary records are sampled every ``AUDIT_SAMPLE_EVERY_NTH`` records, deterministically (``select_audit_sample``). (iii) G6 user acceptance is the final human gate and is never ticked by automation (``check_g6_guard``). A sampled audit found wrong is corrected by appending a correction record that preserves the original (``build_audit_correction``): a wrong pass re-triggers pushback and re-blocks, a wrong block schedules a re-audit, and the corrector must differ from the original signer. (iv) A rebuild triggered by acceptance failures needs a user notice first (``build_acceptance_rebuild_notice``): reasons, rebuild scope, retained assets, and the old-test first gate, with the user's confirmation recorded; until confirmed, the rebuild does not start (``rebuild_pending_user_confirmation``), and after confirmation the user still holds final G6 acceptance. Single counting story (all rules together, in one place): failures count per module per contract version, audit and acceptance separately recorded but summed for thresholds; a pass resets failure counts and pushback rounds but NOT rebuild count or targeted budget within the same version; a version change resets failures, pushback rounds, rebuild count, and targeted budget. Failures 1-3 push back (the 3rd is triaged: structural -> early rebuild consuming rebuild budget; nearly done -> one targeted 3rd pushback; otherwise ordinary 3rd pushback). A targeted round's failure is failure 4. Failure 4 (or an early rebuild) rebuilds only if the rebuild budget remains, the root cause is not contract/design, the cost threshold is not exceeded, and (for acceptance-triggered rebuilds) the user has confirmed. A rebuild resets failure counts and pushback rounds for the new implementation but keeps rebuild count and targeted budget; a first-gate failure counts as an audit failure for the new episode. After a rebuild, failures over 3 again escalate with no second automatic rebuild. Relation to the three existing loops (also spelled out in the skill text, and encoded in ``EXISTING_LOOP_RELATIONS``): - G5 defect loop (defect back to G4, re-regress in G5): REUSED as the fix-execution channel. An audit pushback's derived subtask is executed the G5 way; no second defect state machine is built. The difference from plain G5 is only the automatic trigger and the merge/release block, which G5 alone does not impose. - G7 hotfix loop: PARALLEL, not reused. Pushback is pre-merge; hotfix is post-release emergency repair. A pushback can never un-release something, and a hotfix never satisfies a pushback. - Full-auto CI auto-fix (limit 3): PARALLEL with a separate counter (see above). CI auto-fix runs before audit; audit pushback runs after audit fails. A CI fix success does not skip audit, and an audit pass does not retroactively excuse a CI failure. """ import hashlib import json from datetime import date MAX_PUSHBACK_ROUNDS = 3 # Rebuilds per module per contract version. See module docstring # ("Rebuild budget") for why 1, not 2. MAX_REBUILDS_PER_CONTRACT_VERSION = 1 # Targeted fix rounds per module per contract version (tightening 1). # A targeted round is the 3rd pushback with a restricted scope, not # an extra round outside the failure count (see "Third-failure # triage" in the module docstring). MAX_TARGETED_FIX_ROUNDS_PER_CONTRACT_VERSION = 1 # Near-complete threshold (tightening 1): at most this many failed # items, with no core-structure item and no broken passed item, # may qualify as "nearly done". More than this is not near-complete. NEAR_COMPLETE_MAX_FAILED_ITEMS = 2 # Audit-of-audit sampling (tightening 3): golden tasks and # rebuild/refreeze/escalation records are always sampled; ordinary # records are sampled deterministically every Nth record. AUDIT_SAMPLE_EVERY_NTH = 10 # Gate checks, in the order they are reported. The first nine are the # hard trigger list from the audit-hardening brief; the tenth is the # signer-separation validity check from the independent-audit role. GATE_CHECKS = ( "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", ) LEDGER_KINDS = ("pushback", "fix", "reaudit", "rebuild", "refreeze", "escalate", "blocked_quota", "audit_sample", "audit_correction", "rebuild_notice", "user_confirmation") EXISTING_LOOP_RELATIONS = { # G5 defect loop: reused as the fix-execution channel. "g5_defect_loop": "reuse", # G7 hotfix loop: pre-merge pushback vs post-release hotfix. "g7_hotfix_loop": "parallel", # Full-auto CI auto-fix: separate counter, same limit value. "ci_auto_fix": "parallel_separate_counter", } def full_reaudit_criteria(): """Full re-audit checklist: every gate check, every time. A re-audit after a fix never re-checks only the items that failed last round: previously passing items must be re-verified too, because a fix can break them. """ return list(GATE_CHECKS) def _is_pass(value): return value is True def validate_signer_separation(audit): """Check auditor independence. Returns (valid, [reasons]). The signer must differ from the executor, the coordinator approver, and the QA tester. A missing identity field is also invalid: an audit record that does not say who signed, who executed, who approved, or who tested cannot be verified, so it fails closed instead of passing on an incomplete record. """ reasons = [] signer = audit.get("signer") for field in ("executor", "coordinator_approver", "qa_tester"): other = audit.get(field) if not signer or not other: reasons.append("missing_%s" % field if not other else "missing_signer") elif signer == other: reasons.append("signer_equals_%s" % field) return (not reasons, reasons) def deviation_is_blocking(deviation, today=None): """True when a known deviation must block (i.e. push back). Non-blocking only when all three hold: a follow-up todo is recorded, a due date is recorded and parses, and the due date has not expired without follow-up. ``followed_up=True`` clears the block even past the due date, because the follow-up actually happened; an expired, un-followed deviation blocks. """ if today is None: today = date.today().isoformat() if not deviation.get("todo"): return True due = deviation.get("due_date") if not due: return True try: due_date = date.fromisoformat(str(due)) today_date = date.fromisoformat(str(today)) except ValueError: return True if deviation.get("followed_up"): return False return due_date < today_date def evaluate_regression(items): """Check exact-artifact regression evidence. Returns [reasons]. Each item: ``path``, ``sha256_manifest``, ``sha256_recomputed``, ``source`` (``"exact"`` or ``"fixture"``). A fixture item is never a pass: it is reported as unverified. A hash mismatch stops the regression, the #211 way, instead of running a fixture stand-in. """ reasons = [] for item in items or []: path = item.get("path", "?") if item.get("source") == "fixture": reasons.append("fixture_not_exact_artifact:%s" % path) continue if item.get("source") not in (None, "exact"): reasons.append("unknown_regression_source:%s" % path) continue manifest = item.get("sha256_manifest") recomputed = item.get("sha256_recomputed") if not manifest or not recomputed: reasons.append("missing_regression_hash:%s" % path) elif manifest != recomputed: reasons.append("regression_hash_mismatch:%s" % path) return reasons def evaluate_demo(demo): """True only for a verifiably runnable demo of the app itself.""" if not demo: return False return bool(demo.get("verified_runnable")) and bool(demo.get("runs_app_entry")) def collect_failed_checks(audit, today=None): """All failing gate checks for an audit, in GATE_CHECKS order. Fail-closed: a check absent from ``audit["checks"]`` counts as failed. Rule overlays (security, separation, regression, demo, deviations, verdict) can only add failures, never remove one that the checks dict already reports. """ checks = audit.get("checks") or {} failed = [] for name in GATE_CHECKS: if not _is_pass(checks.get(name)): failed.append(name) # Note: a non-"pass" verdict is a trigger even if every individual # check was (wrongly) marked pass; decide_pushback reports it via # verdict_failed, so it needs no pseudo-check here. security = audit.get("security_verdict") if security in ("conditional", "fail") or not _is_pass( checks.get("security_unconditional") ): if "security_unconditional" not in failed: failed.append("security_unconditional") valid, _reasons = validate_signer_separation(audit) if not valid and "audit_signer_separation" not in failed: failed.append("audit_signer_separation") if evaluate_regression((audit.get("regression") or {}).get("items")): if "regression_exact_artifact" not in failed: failed.append("regression_exact_artifact") if not evaluate_demo(audit.get("demo")): if "demo_runnable" not in failed: failed.append("demo_runnable") for deviation in audit.get("deviations") or []: if deviation_is_blocking(deviation, today=today): if "known_deviation_due_todo" not in failed: failed.append("known_deviation_due_todo") break order = {name: i for i, name in enumerate(GATE_CHECKS)} return sorted(set(failed), key=lambda name: order[name]) def quota_allows(quota_used, fix_cost_estimate, quota_limit): """True when a fix fits in quota. None limit means unmetered.""" if quota_limit is None: return True if quota_used is None: quota_used = 0 if fix_cost_estimate is None: return quota_used < quota_limit return quota_used + fix_cost_estimate <= quota_limit def gate_block(action, failed_checks, round_no): """Track-1 gate interface: what the hard gate must read.""" blocked = action in ("pushback", "rebuild", "escalate", "escalate_refreeze", "blocked_quota", "rebuild_pending_user_confirmation") return { "merge_blocked": blocked, "release_blocked": blocked, "blocked_by": "audit_pushback" if blocked else None, "round": round_no, "failed_checks": list(failed_checks), } def _rebuild_branch(round_no, root_cause, rebuilds_used, max_rebuilds, rebuild_cost_estimate, cost_escalation_threshold, acceptance_triggered, user_confirmed, early): """Shared rebuild decision: over-limit AND early (triage) rebuilds. Gate order: contract/design root cause -> refreeze; rebuild budget exhausted -> escalate (coordinator); acceptance-triggered rebuild without user confirmation -> pending (no rebuild starts); rebuild cost above the user threshold -> escalate (user); else rebuild. ``early`` only changes the reason text. """ if root_cause in ("contract", "design"): return ("escalate_refreeze", "root cause is in the %s itself; rebuilding code " "would treat the symptom, so no code rebuild is started" % root_cause, "coordinator", None, "G2", round_no) if rebuilds_used >= max_rebuilds: return ("escalate", "rebuild budget exhausted (%d rebuild(s) per module " "per contract version) and failures still exceed 3" % max_rebuilds, "coordinator", None, None, round_no) if acceptance_triggered and not user_confirmed: return ("rebuild_pending_user_confirmation", "rebuild triggered by acceptance failures: user notice " "and confirmation required before the rebuild starts", "user", None, None, round_no) if cost_escalation_threshold is not None \ and rebuild_cost_estimate is not None \ and rebuild_cost_estimate > cost_escalation_threshold: return ("escalate", "estimated rebuild cost exceeds the user-set " "cost-escalation threshold; escalating before burning it", "user", None, None, round_no) 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") return ("rebuild", reason, None, rebuilds_used + 1, None, round_no) def decide_pushback(audit, round_no, max_rounds=MAX_PUSHBACK_ROUNDS, today=None, quota_used=None, quota_limit=None, fix_cost_estimate=None, ci_auto_fix_rounds=None, history=None, root_cause=None, rebuilds_used=0, max_rebuilds=MAX_REBUILDS_PER_CONTRACT_VERSION, rebuild_cost_estimate=None, cost_escalation_threshold=None, triage_verdict=None, triage_scope=None, acceptance_triggered=False, user_confirmed_rebuild=False): """Audit conclusion -> allow / pushback / rebuild / escalate. ``ci_auto_fix_rounds`` is informational only and is never counted against the pushback budget (separate counter, same limit value -- see module docstring). ``history`` is the list of previous round summaries; on rebuild / escalation the full history (previous rounds plus this one) is attached, never just the last round. Over-limit order (``round_no >= max_rounds``, i.e. the 4th failure -- cumulative failures > 3): see ``_rebuild_branch``. ``root_cause`` should come from ``classify_root_cause`` over the round history; when it is None (unclassified) the branch defaults to rebuild, because refreeze needs positive contract/design evidence. A rebuild cost estimate above ``cost_escalation_threshold`` escalates to the USER before anything is burned; quota is checked before that and blocks outright (``blocked_quota``). Third-failure triage (tightening 1): when ``round_no`` is the last pushback round (``max_rounds - 1``, the 3rd failure) and ``triage_verdict`` is ``"early_rebuild"``, the rebuild branch runs early instead of a 3rd patch; when it is ``"early_refreeze"``, the refreeze branch runs early (the effective root cause is forced to contract/design for that branch, since triage already established it); when it is ``"targeted_fix"``, the pushback is scope-limited to ``triage_scope`` (the failed items plus directly related code). Any other triage verdict is an ordinary pushback. """ failed = collect_failed_checks(audit, today=today) verdict = audit.get("verdict") verdict_failed = verdict != "pass" at_rebuild = round_no >= max_rounds cost_for_this_step = rebuild_cost_estimate \ if (at_rebuild or triage_verdict == "early_rebuild") \ else fix_cost_estimate rebuild_no = None refreeze_flow = None pushback_scope = "normal" targeted_items = None if quota_limit is not None and not quota_allows( quota_used, cost_for_this_step, quota_limit ): action = "blocked_quota" reason = "fix/rebuild exceeds quota: a fix or rebuild cost " \ "is not exempt from quota" target = None next_round = round_no elif not failed and not verdict_failed: action = "allow" reason = "all gate checks pass with an unconditional audit" target = None next_round = round_no elif at_rebuild or (round_no == max_rounds - 1 and triage_verdict in ("early_rebuild", "early_refreeze")): effective_root = root_cause if triage_verdict == "early_refreeze" \ and effective_root not in ("contract", "design"): effective_root = "contract" action, reason, target, rebuild_no, refreeze_flow, next_round = \ _rebuild_branch(round_no, effective_root, rebuilds_used, max_rebuilds, rebuild_cost_estimate, cost_escalation_threshold, acceptance_triggered, user_confirmed_rebuild, early=not at_rebuild) elif round_no < max_rounds: action = "pushback" reason = "audit failed: %s" % ", ".join(failed) if failed else \ "audit verdict is %s, not an unconditional pass" % verdict target = None next_round = round_no + 1 if triage_verdict == "targeted_fix": pushback_scope = "targeted" targeted_items = list(triage_scope) if triage_scope is not None \ else list(failed) reason = "targeted fix round: scope limited to the failed " \ "items and directly related code; no unrelated refactor" else: # pragma: no cover - round_no cannot be both >= max and < max action = "escalate" reason = "unreachable pushback state" target = "coordinator" next_round = round_no current_summary = { "round": round_no, "action": action, "failed_checks": list(failed), "verdict": verdict, "root_cause": root_cause, "rebuild_no": rebuild_no, "pushback_scope": pushback_scope, } full_history = list(history or []) + [current_summary] attach_history = action in ("rebuild", "escalate", "escalate_refreeze", "rebuild_pending_user_confirmation") return { "action": action, "merge_blocked": action != "allow", "release_blocked": action != "allow", "round": round_no, "next_round": next_round, "failed_checks": failed, "verdict_failed": verdict_failed, "reason": reason, "pushback_form": "derived_fix_subtask" if action == "pushback" else None, "pushback_scope": pushback_scope, "targeted_items": targeted_items, "rebuild_no": rebuild_no, "refreeze_flow": refreeze_flow, "escalation_target": target, # Escalation never grants user acceptance; that stays G6-only. "escalation_grants_user_acceptance": False, "history": full_history if attach_history else list(history or []), "counters": { "pushback_rounds_used": round_no, "pushback_limit": max_rounds, "rebuilds_used": rebuilds_used, "rebuild_limit": max_rebuilds, "ci_auto_fix_rounds_shared": False, "ci_auto_fix_rounds_seen": ci_auto_fix_rounds, }, "gate": gate_block(action, failed, round_no), } # ------------------------------------------------------------------ # Per-module failure counting (audit + acceptance, per contract # version) and the root-cause gate for the rebuild branch. # ------------------------------------------------------------------ def new_module_failure_state(module_id, contract_version): """Fresh per-module counting state (JSON-able dict).""" return { "module_id": module_id, "contract_version": contract_version, "audit_fail_count": 0, "acceptance_fail_count": 0, "total_fail_count": 0, "pushback_rounds_used": 0, "rebuilds_used": 0, "rebuild_history": [], # Tightening 1: one targeted fix round per module per # contract version; a pass does not replenish it. "targeted_fix_used": False, "last_outcome": None, } def record_module_failure(state, source, contract_version=None): """Record one failure for a module; returns a NEW state dict. ``source`` is ``"audit"`` (independent audit / hard gate) or ``"acceptance"`` (G6 user acceptance / QA acceptance). Sources are counted separately but trigger on their sum. A contract version different from the state's resets failure AND rebuild counts before this failure is counted (the module is judged against a different contract, so the old count must not carry over); the failure itself then counts as 1 under the new version. """ if source not in ("audit", "acceptance"): raise ValueError("unknown failure source: %r" % (source,)) new = dict(state) new["rebuild_history"] = list(state.get("rebuild_history") or []) if contract_version is not None \ and contract_version != state.get("contract_version"): # Version change: reset first, then count this failure as 1 # under the new version (see docstring). new["contract_version"] = contract_version new["audit_fail_count"] = 0 new["acceptance_fail_count"] = 0 new["total_fail_count"] = 0 new["pushback_rounds_used"] = 0 new["rebuilds_used"] = 0 new["rebuild_history"] = [] new["targeted_fix_used"] = False if source == "audit": new["audit_fail_count"] = new.get("audit_fail_count", 0) + 1 else: new["acceptance_fail_count"] = new.get("acceptance_fail_count", 0) + 1 new["total_fail_count"] = new["audit_fail_count"] \ + new["acceptance_fail_count"] new["last_outcome"] = "fail:%s" % source return new def record_module_pass(state, contract_version=None): """Record a pass: failure counts reset, rebuild count kept. A pass clears the failure count (and the pushback rounds for the new episode) so the next failure starts counting again, but it does NOT reset ``rebuilds_used`` within the same contract version -- otherwise pass-fail cycles could rebuild forever. A version change resets both (see ``record_module_failure``). """ new = dict(state) new["rebuild_history"] = list(state.get("rebuild_history") or []) if contract_version is not None \ and contract_version != state.get("contract_version"): new["contract_version"] = contract_version new["rebuilds_used"] = 0 new["rebuild_history"] = [] new["targeted_fix_used"] = False new["audit_fail_count"] = 0 new["acceptance_fail_count"] = 0 new["total_fail_count"] = 0 new["pushback_rounds_used"] = 0 # A pass does NOT replenish rebuild count or targeted budget # within the same contract version (see module docstring). new["last_outcome"] = "pass" return new def record_module_rebuild(state, rebuild_no): """Record that a rebuild happened; returns a NEW state dict. The new implementation starts a fresh episode: failure counts and pushback rounds reset to 0. Rebuild count and targeted budget are NOT reset within the same contract version, so a rebuilt module gets fresh failures but no second rebuild and no replenished targeted round. """ new = dict(state) new["rebuild_history"] = list(state.get("rebuild_history") or []) new["rebuild_history"].append(rebuild_no) new["rebuilds_used"] = len(new["rebuild_history"]) new["audit_fail_count"] = 0 new["acceptance_fail_count"] = 0 new["total_fail_count"] = 0 new["pushback_rounds_used"] = 0 new["last_outcome"] = "rebuild:%d" % rebuild_no return new def record_module_targeted_fix(state): """Record that the targeted fix round was used (tightening 1).""" new = dict(state) new["rebuild_history"] = list(state.get("rebuild_history") or []) new["targeted_fix_used"] = True new["last_outcome"] = "targeted_fix" return new def classify_root_cause(failed_items_by_round): """Classify a continuous failure: implementation/contract/design. Input: one list per round, each a list of failed-item dicts with optional ``root_cause`` (``"implementation"`` / ``"contract"`` / ``"design"``), ``evidence.contract_clause``, and ``proved_unimplementable`` (the fix side proved the contract cannot be implemented as written). Returns ``(cause, evidence)``. ``"contract"`` / ``"design"`` require at least two distinct rounds: either the same contract clause recurs across >= 2 rounds with a contract/design root cause, or >= 2 rounds prove unimplementability. Anything weaker is ``"unknown"`` (the caller defaults an unknown cause to the rebuild branch), and >= 2 rounds labelled implementation with no contract/design signal is ``"implementation"``. """ 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 []: 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 evidence = { "contract_rounds": sorted(contract_rounds), "design_rounds": sorted(design_rounds), "repeated_clauses": {clause: sorted(rounds) for clause, rounds in clause_rounds.items() if len(rounds) >= 2}, "proof_rounds": sorted(proof_rounds), } repeated = any(len(rounds) >= 2 for rounds in clause_rounds.values()) if len(proof_rounds) >= 2 and proof_cause: return proof_cause, evidence if repeated: # The repeated clause's cause wins; contract beats design # only on a tie of first appearance, which the evidence # dict makes auditable either way. for clause, rounds in clause_rounds.items(): if len(rounds) >= 2: for items in failed_items_by_round or []: for item in items or []: if (item.get("evidence") or {}).get("contract_clause") == clause \ and item.get("root_cause") in ("contract", "design"): return item["root_cause"], evidence if len(contract_rounds) >= 2: return "contract", evidence if len(design_rounds) >= 2: return "design", evidence if len(implementation_rounds) >= 2: return "implementation", evidence return "unknown", evidence STRUCTURAL_KINDS = ( "architecture_boundary", "module_responsibility", "tech_stack", "data_model", ) def triage_third_failure(audit, failed_items_by_round, passed_items, broken_passed_items, targeted_already_used=False): """Third-failure triage (tightening 1). Returns a verdict dict. Call this when the cumulative failure count reaches 3, before deciding the 3rd pushback. Inputs are evidence, not opinions: - ``audit``: the current (3rd-failure) audit; its ``failed_items`` carry optional ``structural_kind``, ``touches_core_structure``, ``root_cause`` and ``evidence`` (test name, file path, line, contract clause, diff range). - ``failed_items_by_round``: per-round failed items, for the repeated-clause check via ``classify_root_cause``. - ``passed_items``: items that passed before this round. - ``broken_passed_items``: previously passed items that this round broke; must be supplied (an absent list is treated as unknown, which disqualifies "nearly done" -- a near-complete claim needs positive evidence that nothing passed broke). Verdicts: ``"early_rebuild"`` (implementation-structural), ``"early_refreeze"`` (contract/design-structural: the existing root-cause gate, applied early instead of waiting for the 4th failure -- contract/design priority wins when both signals are present, because a contract fix may change the architecture), ``"targeted_fix"`` (nearly done: at most NEAR_COMPLETE_MAX_FAILED_ITEMS failed items, none structural or core-touching, no broken passed item, and passed items outnumber failed items), or ``"normal_pushback"`` (everything else, including a nearly-done case whose targeted budget is already used). This split keeps tightening 1 consistent with the earlier root-cause rule: a repeated contract/design clause is structural, but it is NOT a code rebuild -- rebuilding code against a broken contract treats the symptom. The returned ``basis`` copies the deciding evidence verbatim so a later sample can re-check the triage. """ failed = list(audit.get("failed_items") or []) passed = list(passed_items or []) broken = broken_passed_items # None means unknown, not empty. root_cause, root_evidence = classify_root_cause( failed_items_by_round) impl_structural = [] contract_structural = [] for item in failed: kind = item.get("structural_kind") if kind in STRUCTURAL_KINDS: impl_structural.append("structural_kind:%s" % kind) if item.get("touches_core_structure"): impl_structural.append("core_structure_touched") if item.get("root_cause") in ("contract", "design") \ and (item.get("evidence") or {}).get("contract_clause") \ in root_evidence.get("repeated_clauses", {}): contract_structural.append("repeated_contract_design_clause") if root_cause in ("contract", "design") or contract_structural: verdict = "early_refreeze" reasons = contract_structural or ["root_cause:%s" % root_cause] elif impl_structural: verdict = "early_rebuild" reasons = impl_structural elif len(failed) <= NEAR_COMPLETE_MAX_FAILED_ITEMS \ and broken == [] \ and len(passed) > len(failed): if targeted_already_used: verdict = "normal_pushback" reasons = ["targeted budget already used (1 per version)"] else: verdict = "targeted_fix" reasons = ["failed_items=%d (<= %d), no structural item, " "no broken passed item, passed=%d > failed=%d" % (len(failed), NEAR_COMPLETE_MAX_FAILED_ITEMS, len(passed), len(failed))] else: verdict = "normal_pushback" reasons = [] if len(failed) > NEAR_COMPLETE_MAX_FAILED_ITEMS: reasons.append("failed_items=%d exceeds near-complete cap %d" % (len(failed), NEAR_COMPLETE_MAX_FAILED_ITEMS)) if broken != []: reasons.append("broken passed items present or unknown") if len(passed) <= len(failed): reasons.append("passed items do not outnumber failed items") return { "verdict": verdict, "reasons": reasons, "targeted_scope": [item.get("check") for item in failed] if verdict == "targeted_fix" else None, "root_cause": root_cause, "basis": { "audit_verdict_text": audit.get("verdict_text"), "failed_items": failed, "passed_items": passed, "broken_passed_items": broken, "failed_items_by_round": failed_items_by_round, "root_cause_evidence": root_evidence, }, } def build_rebuild_package(module_id, decision, audit_history, previous_artifacts, old_implementation_ref=None, other_module_ids=None, contract_version=None, passed_items=None, rebuild_cost_estimate=None, actual_cost_note=None, retained_tests=None): """Build the rebuild task package for a ``rebuild`` decision. The package is explicitly labelled a rebuild (never disguised as an ordinary new task), scopes the rebuild to this module only, keeps the old implementation by reference in git history and in the audit record (never deleted from history), and carries the avoidance list: every previous audit conclusion, failure reason, and passed item, so the new implementation does not repeat the same error and claim a fresh start. Tightening 2: only the implementation code is rebuilt. The package's ``retained_assets`` list the old test set, the retained passed acceptance items, the unchanged contract and design, and the audit conclusions -- none of these is reset with the implementation. ``retained_tests=None`` (not supplied) is flagged ``retained_tests_missing``; an explicitly empty list means the module genuinely has no tests, which the first gate then treats as a vacuous pass only if the caller also supplies no retained tests to check (see ``check_rebuild_first_gate``). """ avoidance = [] for round_audit in audit_history or []: for item in round_audit.get("failed_items") or []: avoidance.append({ "kind": "prior_failure", "check": item.get("check"), "detail": item.get("detail"), "audit_verdict_text": round_audit.get("verdict_text"), }) for item in passed_items or []: avoidance.append({ "kind": "previously_passed_must_not_regress", "check": item.get("check"), "detail": item.get("detail"), }) return { "kind": "rebuild", "task_label": "推倒重做第 %s 次：模块 %s，上一轮失败原因见避坑清单" % (decision.get("rebuild_no"), module_id), "module_id": module_id, "contract_version": contract_version, "rebuild_no": decision.get("rebuild_no"), "scope": { "rebuild_only": module_id, "untouched_modules": list(other_module_ids or []), "frozen_contract_unchanged": True, "design_unchanged": True, "other_modules_passed_artifacts_untouched": True, "directory_isolation_unchanged": True, }, "old_implementation": { "retention": "git_history_and_audit_record", "ref": old_implementation_ref, "archive_ref_required_before_replace": old_implementation_ref is None, "workspace_action": "replace_with_new_implementation", "deleted_from_history": False, }, "implementation_basis": "frozen_contract", "avoidance_list": avoidance, # Tightening 2: retained assets. Only implementation code # is replaced; everything listed here survives the rebuild. "retained_assets": { "tests": list(retained_tests) if retained_tests is not None else [], "retained_tests_missing": retained_tests is None, "passed_acceptance_items": list(passed_items or []), "contract_unchanged": True, "design_unchanged": True, "audit_conclusions": [round_audit.get("verdict_text") for round_audit in audit_history or []], }, "first_gate": "old tests plus retained acceptance items must " "all pass before any new full audit (see " "check_rebuild_first_gate); a first-gate failure counts " "as an audit failure for the new implementation", "reaudit_criteria": full_reaudit_criteria(), "previous_artifacts": [ {"path": artifact.get("path"), "sha256": artifact.get("sha256"), "bytes": artifact.get("bytes")} for artifact in previous_artifacts or [] ], "merge_blocked_until_reaudit_pass": True, "release_blocked_until_reaudit_pass": True, "cost": { "estimated_rebuild_cost": rebuild_cost_estimate, "actual_cost_note": actual_cost_note or "record estimated cost here before starting, and " "actual cost plus token usage here when the rebuild " "finishes; never report an unpriced rebuild as 0", "quota_note": "rebuild is costed and quota-checked like any " "other task; being a rebuild grants no exemption", }, } def check_rebuild_first_gate(retained_tests, retained_passed_items, test_results, acceptance_results): """First gate of a rebuild (tightening 2): old tests must pass. ``test_results`` / ``acceptance_results`` map id -> bool (or a list of ``{"id": ..., "passed": bool}`` dicts). A retained test or retained acceptance item with no result fails closed (missing is not a pass), and any failed item fails the gate. A gate failure counts as an audit failure for the new implementation's episode (record it with ``record_module_failure(state, "audit")``); it is never waved through because the new implementation "has new features". Returns ``{"gate_pass": bool, "missing": [...], "failed": [...]}``. """ def _as_map(values): if isinstance(values, dict): return values out = {} for entry in values or []: out[entry.get("id")] = bool(entry.get("passed")) return out test_map = _as_map(test_results) acceptance_map = _as_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 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(changes, refreeze_approved=False, separate_audit_ids=None): """Validate test deletions / assertion changes (tightening 2). Each change: ``{"action": "delete" | "modify_assertion" | "add", "test_id": ..., "separate_audit_id": ...}``. Deleting a test or changing an assertion is allowed only when the contract changed via the G2 re-freeze flow AND the change has its own separate audit record (its id in ``separate_audit_ids`` or on the change itself). Adding a test never needs refreeze, but an addition must not silently replace a retained test -- that is a delete, not an add. Returns a list of violations (empty means allowed). """ audits = set(separate_audit_ids or []) violations = [] for change in changes or []: action = change.get("action") test_id = change.get("test_id") if action in ("delete", "modify_assertion"): audit_id = change.get("separate_audit_id") if not refreeze_approved: violations.append( "%s:%s without contract re-freeze" % (action, test_id)) if audit_id not in audits and audit_id is None: violations.append( "%s:%s without separate audit" % (action, test_id)) elif audit_id is not None and 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_golden_task_set(tasks): """Validate a golden-task set (tightening 3). Each task needs a human approver (``human_approved_by``) and the set must contain no example/sample files (``is_example``): a sample file is not a human-approved golden task, however realistic it looks. Returns (valid, [reasons]). """ reasons = [] for task in tasks or []: task_id = task.get("id", "?") if task.get("is_example"): reasons.append("example file is not a golden task:%s" % task_id) if not task.get("human_approved_by"): reasons.append("golden task lacks human approver:%s" % task_id) return (not reasons, reasons) def select_audit_sample(records, golden_task_ids): """Select audit records for audit-of-audit sampling (tightening 3). Always sampled: golden tasks, and rebuild / refreeze / escalation records (the critical paths). Ordinary records are sampled deterministically every AUDIT_SAMPLE_EVERY_NTH record (by position, so the sample is reproducible and checkable). Returns a list of ``{"record": ..., "reason": ...}``. """ golden = set(golden_task_ids or []) selected = [] for index, record in enumerate(records or []): reason = None if record.get("task_id") in golden: reason = "golden_task" elif record.get("kind") in ("rebuild", "refreeze", "escalate"): reason = "critical_path" elif (index + 1) % AUDIT_SAMPLE_EVERY_NTH == 0: reason = "periodic_sample" if reason: selected.append({"record": record, "reason": reason}) return selected def build_audit_correction(original_entry, corrected_verdict, corrector, reason): """Correct a wrong audit conclusion without rewriting it. Returns ``(correction, errors)``. The original entry is never mutated: the correction appends a new record that references the original by its ``entry_hash`` (or task/round when no hash exists). The corrector must differ from the original signer -- the audit of the audit cannot be signed by the same person. Restored action is derived, not chosen freely: a wrong pass (should have blocked) re-triggers pushback and re-blocks; a wrong block (should have passed) schedules a re-audit -- it does not jump straight to merge, because the correction itself is a review, not a full audit. """ errors = [] signer = original_entry.get("signer") if not corrector or (signer and corrector == signer): errors.append("corrector_equals_original_signer") original_verdict = original_entry.get("verdict") if corrected_verdict == original_verdict: errors.append("correction_matches_original_verdict") if corrected_verdict == "fail" and original_verdict == "pass": restored = "retrigger_pushback_and_reblock" elif corrected_verdict == "pass" and original_verdict == "fail": restored = "schedule_reaudit" else: restored = "review_required" correction = { "kind": "audit_correction", "original_entry_hash": original_entry.get("entry_hash"), "original_task_id": original_entry.get("task_id"), "original_round": original_entry.get("round"), "original_verdict": original_verdict, "corrected_verdict": corrected_verdict, "corrector": corrector, "reason": reason, "restored_action": restored, "valid": not errors, "errors": errors, } return correction, errors def check_g6_guard(record): """G6 guard (tightening 3): only the user ticks acceptance. ``record`` may carry ``g6_acceptance`` and ``g6_accepted_by``. Acceptance ticked by anyone other than the user is a violation; an absent acceptance field is not a violation (it simply means acceptance has not happened). Returns violations. """ if record.get("g6_acceptance") and record.get("g6_accepted_by") != "user": return ["g6 acceptance ticked by non-user:%s" % record.get("g6_accepted_by")] return [] def build_acceptance_rebuild_notice(module_id, decision, package, failed_reasons): """User notice required before an acceptance-triggered rebuild. Contents: failure reasons, rebuild scope, retained assets, and the old-test first gate. The notice itself does not start the rebuild: ``confirmed`` starts False and only a recorded user confirmation (see ``confirm_acceptance_rebuild``) unblocks it. """ return { "kind": "rebuild_notice", "to": "user", "module_id": module_id, "failure_reasons": list(failed_reasons or []), "rebuild_scope": (package or {}).get("scope"), "retained_assets": (package or {}).get("retained_assets"), "first_gate": (package or {}).get("first_gate"), "rebuild_no": decision.get("rebuild_no"), "requires_user_confirmation": True, "confirmed": False, } def confirm_acceptance_rebuild(notice, user_confirmed): """Apply the user's confirmation to a rebuild notice. Unconfirmed: the rebuild must not start silently -- the returned notice stays unconfirmed and the caller must keep the ``rebuild_pending_user_confirmation`` decision. Confirmed: the notice records the confirmation and the rebuild may proceed; the user still holds final G6 acceptance afterwards. """ updated = dict(notice) updated["confirmed"] = bool(user_confirmed) updated["g6_acceptance_retained_by_user"] = True return updated def build_pushback_package(audit, decision, previous_artifacts, fix_diff_paths=None, qa_failed_links=None): """Build the executable pushback package for a pushback/escalate. Carries the audit conclusion verbatim, one entry per failed item with its evidence (test name, file path, line number, contract clause, diff range), reproduction steps, the full re-audit criteria (not just the failed items), and the previous round's artifact manifest with hashes, so a fix can be diffed against it and cannot silently drop or overwrite a passed artifact (see ``verify_previous_artifacts``). QA full-chain supplement: ``qa_failed_links`` (from the QA matrix, see ``cloud/qa_matrix.py``) lists the failed business links and cases (link name, case name, failure, evidence, reproduction). The re-audit must re-run exactly those links, not just unit tests; they are carried separately in ``qa_failed_links`` and ``reaudit_must_rerun_links`` so the re-audit checklist cannot drop them. """ failed_items = [] for item in audit.get("failed_items") or []: evidence = item.get("evidence") or {} failed_items.append({ "check": item.get("check"), "detail": item.get("detail"), "evidence": { "test_name": evidence.get("test_name"), "file_path": evidence.get("file_path"), "line_no": evidence.get("line_no"), "contract_clause": evidence.get("contract_clause"), "diff_range": evidence.get("diff_range"), }, "reproduction_steps": item.get("reproduction_steps") or [], "reaudit_criterion": item.get("reaudit_criterion") or "re-run the full re-audit checklist for this check", }) for check in decision.get("failed_checks") or []: if not any(entry["check"] == check for entry in failed_items): failed_items.append({ "check": check, "detail": "gate check failed (no per-item audit detail supplied)", "evidence": { "test_name": None, "file_path": None, "line_no": None, "contract_clause": None, "diff_range": None, }, "reproduction_steps": ["re-run the audit checks"], "reaudit_criterion": "re-run the full re-audit checklist", }) reproduction = [] for entry in failed_items: reproduction.extend(entry["reproduction_steps"]) if not reproduction: reproduction = ["re-run the audit checks listed in failed_checks"] return { "task_id": audit.get("task_id"), "round": decision.get("round"), "next_round": decision.get("next_round"), "decision_action": decision.get("action"), "audit_verdict_text": audit.get("verdict_text"), "failed_items": failed_items, "failed_checks": list(decision.get("failed_checks") or []), "reproduction_steps": reproduction, # Full re-audit, not failed-items-only. "reaudit_criteria": full_reaudit_criteria(), "previous_artifacts": [ { "path": artifact.get("path"), "sha256": artifact.get("sha256"), "bytes": artifact.get("bytes"), } for artifact in previous_artifacts or [] ], "derived_subtask": { "kind": "audit_fix", "parent_task_id": audit.get("task_id"), "pushback_round": decision.get("round"), "merge_blocked_until_reaudit_pass": True, "release_blocked_until_reaudit_pass": True, "cost_and_quota_note": "the fix task is costed and quota-checked " "like any other task; being a fix grants no exemption", }, "fix_diff_paths": list(fix_diff_paths or []), "qa_failed_links": list( qa_failed_links if qa_failed_links is not None else audit.get("qa_failed_links") or []), "reaudit_must_rerun_links": list( qa_failed_links if qa_failed_links is not None else audit.get("qa_failed_links") or []), } def verify_previous_artifacts(previous_artifacts, new_manifest, fix_diff_paths=None): """Check a fix did not silently drop/overwrite passed artifacts. Returns ``{"ok": bool, "violations": [...], "changed_and_reauditable": [...]}``. A previously passed artifact missing from the new manifest is a violation ("dropped"); one whose hash changed without appearing in the fix diff is a violation ("silently_changed"); one whose hash changed AND is listed in the fix diff is allowed but must go through full re-audit, so it is reported separately, not as a pass. """ fix_diff = set(fix_diff_paths or []) new_by_path = {item.get("path"): item for item in new_manifest or []} violations = [] changed = [] for artifact in previous_artifacts or []: path = artifact.get("path") new = new_by_path.get(path) if new is None: violations.append("dropped:%s" % path) elif new.get("sha256") != artifact.get("sha256"): if path in fix_diff: changed.append(path) else: violations.append("silently_changed:%s" % path) return { "ok": not violations, "violations": violations, "changed_and_reauditable": changed, } def _canonical_hash(entry_without_hash): payload = json.dumps(entry_without_hash, sort_keys=True, separators=(",", ":"), ensure_ascii=False) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def build_ledger_entry(kind, task_id, round_no, trigger_checks=None, package_ref=None, fix_diff=None, reaudit_result=None, cost=None, tokens=None, prev_hash="", module_id=None, contract_version=None, fail_counts=None, rebuild_no=None, root_cause=None, old_implementation_ref=None): """One append-only audit-ledger record (track-4 interface proposal). Every pushback, fix, re-audit, rebuild, refreeze escalation, and escalation appends one entry; entries are never rewritten, so the timeline keeps failed rounds. Module fields record the per-module failure counts, the rebuild number, the root-cause verdict, and the old implementation's archive reference, so a rebuild's old/new correspondence stays auditable. """ if kind not in LEDGER_KINDS: raise ValueError("unknown ledger kind: %r" % (kind,)) entry = { "kind": kind, "task_id": task_id, "round": round_no, "trigger_checks": list(trigger_checks or []), "package_ref": package_ref, "fix_diff": list(fix_diff or []), "reaudit_result": reaudit_result, "cost": cost, "tokens": tokens, "prev_hash": prev_hash, "module_id": module_id, "contract_version": contract_version, "fail_counts": fail_counts, "rebuild_no": rebuild_no, "root_cause": root_cause, "old_implementation_ref": old_implementation_ref, } entry["entry_hash"] = _canonical_hash(entry) return entry def validate_ledger_chain(entries): """Validate a ledger chain. Returns (valid, [errors]). Checks every entry hash, every prev_hash link, the kind vocabulary, and non-decreasing rounds per task. It does NOT claim a chain is complete merely because it validates: a chain that kept only the final passing entry would validate structurally while violating the timeline rule, so callers must also retain (and show) every round's entry. """ errors = [] prev_hash = "" last_round = {} for index, entry in enumerate(entries or []): claimed = entry.get("entry_hash") body = {key: value for key, value in entry.items() if key != "entry_hash"} if claimed != _canonical_hash(body): errors.append("entry %d: hash mismatch" % index) if entry.get("prev_hash", "") != prev_hash: errors.append("entry %d: prev_hash link broken" % index) if entry.get("kind") not in LEDGER_KINDS: errors.append("entry %d: unknown kind" % index) task_id = entry.get("task_id") round_no = entry.get("round") if task_id in last_round and round_no < last_round[task_id]: errors.append("entry %d: round goes backwards" % index) if isinstance(round_no, int): last_round[task_id] = round_no prev_hash = claimed or "" return (not errors, errors)