"""Independent-audit auto-pushback: decision, package, ledger. Pins the six required behaviours: (1) audit fail auto-pushes back, (2) merge/release stay blocked until a full re-audit passes, (3) the pushback package carries per-item evidence, (4) the round budget escalates instead of looping forever, (5) a fix gets a full re-audit, not a failed-items-only re-check, (6) every round stays visible in the audit ledger. Plus the fail-closed edges: CONDITIONAL is not a pass, missing checks fail, signer separation, exact- artifact regression, runnable demos, deviation expiry, and quota for fix tasks. Style follows the other cloud tests: unittest, stdlib only, import with a sys.path fallback so the file also runs standalone. """ import sys import unittest from pathlib import Path try: from cloud.audit_pushback import ( MAX_PUSHBACK_ROUNDS, MAX_REBUILDS_PER_CONTRACT_VERSION, build_acceptance_rebuild_notice, build_audit_correction, build_ledger_entry, build_pushback_package, build_rebuild_package, check_g6_guard, check_rebuild_first_gate, classify_root_cause, collect_failed_checks, confirm_acceptance_rebuild, decide_pushback, deviation_is_blocking, evaluate_regression, full_reaudit_criteria, gate_block, new_module_failure_state, quota_allows, record_module_failure, record_module_pass, record_module_rebuild, record_module_targeted_fix, select_audit_sample, triage_third_failure, validate_golden_task_set, validate_ledger_chain, validate_signer_separation, validate_test_changes, verify_previous_artifacts, ) except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.audit_pushback import ( MAX_PUSHBACK_ROUNDS, MAX_REBUILDS_PER_CONTRACT_VERSION, build_acceptance_rebuild_notice, build_audit_correction, build_ledger_entry, build_pushback_package, build_rebuild_package, check_g6_guard, check_rebuild_first_gate, classify_root_cause, collect_failed_checks, confirm_acceptance_rebuild, decide_pushback, deviation_is_blocking, evaluate_regression, full_reaudit_criteria, gate_block, new_module_failure_state, quota_allows, record_module_failure, record_module_pass, record_module_rebuild, record_module_targeted_fix, select_audit_sample, triage_third_failure, validate_golden_task_set, validate_ledger_chain, validate_signer_separation, validate_test_changes, verify_previous_artifacts, ) ALL_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", ) def passing_audit(**overrides): audit = { "task_id": "task-1", "verdict": "pass", "verdict_text": "audit pass: all checks verified independently", "signer": "auditor-a", "executor": "executor-e", "coordinator_approver": "coordinator-c", "qa_tester": "qa-q", "security_verdict": "pass", "checks": {name: True for name in ALL_CHECKS}, "failed_items": [], "deviations": [], "regression": {"items": [ {"path": "apps/web/app.py", "sha256_manifest": "a" * 64, "sha256_recomputed": "a" * 64, "source": "exact"}, ]}, "demo": {"verified_runnable": True, "runs_app_entry": True, "command": "python3 apps/web/app.py"}, } audit.update(overrides) return audit class DecideTestCase(unittest.TestCase): def test_fail_verdict_auto_pushbacks_and_blocks(self): decision = decide_pushback(passing_audit(verdict="fail"), 0) self.assertEqual(decision["action"], "pushback") self.assertTrue(decision["merge_blocked"]) self.assertTrue(decision["release_blocked"]) self.assertEqual(decision["pushback_form"], "derived_fix_subtask") self.assertEqual(decision["next_round"], 1) def test_conditional_verdict_is_not_a_pass(self): for verdict in ("conditional", "fail"): with self.subTest(verdict=verdict): decision = decide_pushback(passing_audit(verdict=verdict), 0) self.assertEqual(decision["action"], "pushback") def test_conditional_security_blocks_even_with_pass_verdict(self): audit = passing_audit(security_verdict="conditional") decision = decide_pushback(audit, 0) self.assertEqual(decision["action"], "pushback") self.assertIn("security_unconditional", decision["failed_checks"]) def test_each_gate_check_failing_triggers_pushback(self): for check in ALL_CHECKS: with self.subTest(check=check): checks = {name: True for name in ALL_CHECKS} checks[check] = False decision = decide_pushback(passing_audit(checks=checks), 0) self.assertEqual(decision["action"], "pushback") self.assertIn(check, decision["failed_checks"]) def test_missing_check_fails_closed(self): checks = {name: True for name in ALL_CHECKS} del checks["ci_green"] decision = decide_pushback(passing_audit(checks=checks), 0) self.assertEqual(decision["action"], "pushback") self.assertIn("ci_green", decision["failed_checks"]) def test_pass_allows_merge_and_release(self): decision = decide_pushback(passing_audit(), 0) self.assertEqual(decision["action"], "allow") self.assertFalse(decision["merge_blocked"]) self.assertFalse(decision["release_blocked"]) self.assertEqual(decision["gate"]["blocked_by"], None) def test_gate_block_interface(self): blocked = gate_block("pushback", ["ci_green"], 1) self.assertTrue(blocked["merge_blocked"]) self.assertTrue(blocked["release_blocked"]) self.assertEqual(blocked["blocked_by"], "audit_pushback") self.assertEqual(blocked["failed_checks"], ["ci_green"]) class SeparationTestCase(unittest.TestCase): def test_signer_equal_to_executor_coordinator_or_qa_invalid(self): for field, value in (("executor", "auditor-a"), ("coordinator_approver", "auditor-a"), ("qa_tester", "auditor-a")): with self.subTest(field=field): audit = passing_audit(**{field: "auditor-a"}) valid, _reasons = validate_signer_separation(audit) self.assertFalse(valid) decision = decide_pushback(audit, 0) self.assertEqual(decision["action"], "pushback") self.assertIn("audit_signer_separation", decision["failed_checks"]) def test_distinct_signer_is_valid(self): valid, reasons = validate_signer_separation(passing_audit()) self.assertTrue(valid) self.assertEqual(reasons, []) class DeviationTestCase(unittest.TestCase): def test_deviation_without_due_or_todo_blocks(self): self.assertTrue(deviation_is_blocking({"id": "d1"})) self.assertTrue(deviation_is_blocking( {"id": "d1", "todo": "follow up", "due_date": None})) self.assertTrue(deviation_is_blocking( {"id": "d1", "due_date": "2099-01-01"})) def test_deviation_with_future_due_and_todo_does_not_block(self): self.assertFalse(deviation_is_blocking( {"id": "d1", "todo": "follow up", "due_date": "2099-01-01"}, today="2026-10-31")) def test_expired_deviation_blocks_until_followed_up(self): expired = {"id": "d1", "todo": "follow up", "due_date": "2026-01-01"} self.assertTrue(deviation_is_blocking(expired, today="2026-10-31")) self.assertFalse(deviation_is_blocking( dict(expired, followed_up=True), today="2026-10-31")) def test_expired_deviation_triggers_pushback(self): audit = passing_audit(deviations=[ {"id": "d1", "todo": "follow up", "due_date": "2026-01-01"}]) decision = decide_pushback(audit, 0, today="2026-10-31") self.assertEqual(decision["action"], "pushback") self.assertIn("known_deviation_due_todo", decision["failed_checks"]) class EvidenceTestCase(unittest.TestCase): def test_regression_hash_mismatch_and_fixture_block(self): mismatch = [{"path": "x.py", "sha256_manifest": "a" * 64, "sha256_recomputed": "b" * 64, "source": "exact"}] fixture = [{"path": "x.py", "sha256_manifest": "a" * 64, "sha256_recomputed": "a" * 64, "source": "fixture"}] self.assertTrue(evaluate_regression(mismatch)) self.assertTrue(evaluate_regression(fixture)) self.assertEqual(evaluate_regression([ {"path": "x.py", "sha256_manifest": "a" * 64, "sha256_recomputed": "a" * 64, "source": "exact"}]), []) def test_demo_must_run_the_app_entry(self): audit = passing_audit(demo={"verified_runnable": True, "runs_app_entry": False, "command": "python3 -m http.server"}) decision = decide_pushback(audit, 0) self.assertEqual(decision["action"], "pushback") self.assertIn("demo_runnable", decision["failed_checks"]) def test_pushback_package_carries_evidence_and_full_reaudit(self): audit = passing_audit( verdict="fail", verdict_text="AUDIT-FAIL verbatim: ci red, demo not runnable", failed_items=[{ "check": "ci_green", "detail": "checks red on current head", "evidence": {"test_name": "test_x", "file_path": "a/b.py", "line_no": 42, "contract_clause": "G2 v1", "diff_range": "a/b.py@10-20"}, "reproduction_steps": ["run CI on the PR head"], "reaudit_criterion": "CI green on the fixed head", }], checks={**{name: True for name in ALL_CHECKS}, "ci_green": False}, ) decision = decide_pushback(audit, 1) package = build_pushback_package( audit, decision, [{"path": "apps/web/app.py", "sha256": "a" * 64, "bytes": 123}]) self.assertEqual(package["audit_verdict_text"], "AUDIT-FAIL verbatim: ci red, demo not runnable") item = package["failed_items"][0] self.assertEqual(item["evidence"]["test_name"], "test_x") self.assertEqual(item["evidence"]["line_no"], 42) self.assertEqual(item["evidence"]["contract_clause"], "G2 v1") self.assertEqual(item["reproduction_steps"], ["run CI on the PR head"]) # Full re-audit, not failed-items-only. self.assertEqual(package["reaudit_criteria"], full_reaudit_criteria()) self.assertEqual(len(package["reaudit_criteria"]), len(ALL_CHECKS)) self.assertEqual(package["previous_artifacts"][0]["sha256"], "a" * 64) self.assertTrue(package["derived_subtask"]["merge_blocked_until_reaudit_pass"]) class PreviousArtifactTestCase(unittest.TestCase): PREVIOUS = [{"path": "a.py", "sha256": "a" * 64}, {"path": "b.py", "sha256": "b" * 64}] def test_dropped_or_silently_changed_artifact_is_a_violation(self): dropped = verify_previous_artifacts( self.PREVIOUS, [{"path": "a.py", "sha256": "a" * 64}]) self.assertFalse(dropped["ok"]) self.assertIn("dropped:b.py", dropped["violations"]) silent = verify_previous_artifacts( self.PREVIOUS, [{"path": "a.py", "sha256": "a" * 64}, {"path": "b.py", "sha256": "c" * 64}]) self.assertFalse(silent["ok"]) self.assertIn("silently_changed:b.py", silent["violations"]) def test_changed_in_fix_diff_is_reauditable_not_a_violation(self): result = verify_previous_artifacts( self.PREVIOUS, [{"path": "a.py", "sha256": "a" * 64}, {"path": "b.py", "sha256": "c" * 64}], fix_diff_paths=["b.py"]) self.assertTrue(result["ok"]) self.assertEqual(result["changed_and_reauditable"], ["b.py"]) class LimitAndQuotaTestCase(unittest.TestCase): def test_pushback_budget_exhausted_rebuilds_first_with_full_history(self): # Over-3 rule (follow-up, superseding the earlier # "exhausted -> escalate" wording): the 4th failure rebuilds # first; escalation comes only after the rebuild budget too. history = [{"round": 0, "action": "pushback"}, {"round": 1, "action": "pushback"}, {"round": 2, "action": "pushback"}] decision = decide_pushback(passing_audit(verdict="fail"), 3, history=history) self.assertEqual(decision["action"], "rebuild") self.assertEqual(decision["rebuild_no"], 1) self.assertTrue(decision["merge_blocked"]) # Full history, not just the last round. self.assertEqual(len(decision["history"]), 4) self.assertEqual(decision["history"][0]["round"], 0) self.assertEqual(decision["history"][-1]["round"], 3) def test_rebuild_budget_exhausted_escalates_to_coordinator(self): decision = decide_pushback(passing_audit(verdict="fail"), 3, rebuilds_used=1) self.assertEqual(decision["action"], "escalate") self.assertEqual(decision["escalation_target"], "coordinator") self.assertFalse(decision["escalation_grants_user_acceptance"]) self.assertTrue(decision["merge_blocked"]) def test_max_rounds_constant_is_three(self): self.assertEqual(MAX_PUSHBACK_ROUNDS, 3) def test_ci_auto_fix_rounds_do_not_consume_pushback_budget(self): # Same limit value, separate counter: 3 CI fixes must not # turn a first-round audit failure into an escalation. decision = decide_pushback(passing_audit(verdict="fail"), 0, ci_auto_fix_rounds=3) self.assertEqual(decision["action"], "pushback") self.assertFalse(decision["counters"]["ci_auto_fix_rounds_shared"]) def test_fix_task_quota_block_consumes_no_pushback_round(self): decision = decide_pushback(passing_audit(verdict="fail"), 1, quota_used=90, quota_limit=100, fix_cost_estimate=50) self.assertEqual(decision["action"], "blocked_quota") self.assertTrue(decision["merge_blocked"]) self.assertEqual(decision["next_round"], 1) self.assertTrue(quota_allows(90, 10, 100)) self.assertFalse(quota_allows(90, 50, 100)) class FullReauditTestCase(unittest.TestCase): def test_previously_passing_check_broken_by_fix_still_fails(self): # A fix that repairs ci_green but breaks directory isolation # must not pass on a failed-items-only re-check. checks = {name: True for name in ALL_CHECKS} checks["directory_isolation"] = False decision = decide_pushback(passing_audit(checks=checks), 1) self.assertEqual(decision["action"], "pushback") self.assertIn("directory_isolation", decision["failed_checks"]) self.assertEqual(decision["next_round"], 2) class LedgerTestCase(unittest.TestCase): def _chain(self): first = build_ledger_entry("pushback", "task-1", 0, trigger_checks=["ci_green"], package_ref="pkg-0") second = build_ledger_entry("fix", "task-1", 1, fix_diff=["a.py"], prev_hash=first["entry_hash"]) third = build_ledger_entry("reaudit", "task-1", 1, reaudit_result="fail", prev_hash=second["entry_hash"]) return [first, second, third] def test_chain_validates_and_keeps_every_round(self): entries = self._chain() valid, errors = validate_ledger_chain(entries) self.assertTrue(valid, errors) self.assertEqual([entry["kind"] for entry in entries], ["pushback", "fix", "reaudit"]) def test_tampered_entry_breaks_validation(self): entries = self._chain() entries[1]["fix_diff"] = ["tampered.py"] valid, errors = validate_ledger_chain(entries) self.assertFalse(valid) self.assertTrue(errors) def test_broken_prev_link_breaks_validation(self): entries = self._chain() entries[2]["prev_hash"] = "0" * 64 valid, _errors = validate_ledger_chain(entries) self.assertFalse(valid) def test_unknown_ledger_kind_rejected(self): with self.assertRaises(ValueError): build_ledger_entry("bogus", "task-1", 0) class ModuleCountingTestCase(unittest.TestCase): def test_audit_and_acceptance_failures_combine_but_stay_separate(self): state = new_module_failure_state("mod-web", "v1") state = record_module_failure(state, "audit", "v1") state = record_module_failure(state, "acceptance", "v1") state = record_module_failure(state, "audit", "v1") self.assertEqual(state["audit_fail_count"], 2) self.assertEqual(state["acceptance_fail_count"], 1) self.assertEqual(state["total_fail_count"], 3) def test_pass_resets_failures_but_not_rebuilds(self): state = new_module_failure_state("mod-web", "v1") state = record_module_failure(state, "audit", "v1") state = record_module_rebuild(state, 1) state = record_module_pass(state, "v1") self.assertEqual(state["total_fail_count"], 0) self.assertEqual(state["rebuilds_used"], 1) def test_contract_version_change_resets_failures_and_rebuilds(self): state = new_module_failure_state("mod-web", "v1") state = record_module_failure(state, "audit", "v1") state = record_module_rebuild(state, 1) state = record_module_failure(state, "audit", "v2") self.assertEqual(state["contract_version"], "v2") self.assertEqual(state["total_fail_count"], 1) self.assertEqual(state["rebuilds_used"], 0) def test_boundary_exactly_three_pushbacks_fourth_rebuilds(self): # Failures 1-3 (round_no 0,1,2) push back for a fix; the # 4th failure (round_no 3, total > 3) rebuilds instead. for round_no in (0, 1, 2): with self.subTest(round_no=round_no): decision = decide_pushback( passing_audit(verdict="fail"), round_no) self.assertEqual(decision["action"], "pushback") decision = decide_pushback(passing_audit(verdict="fail"), 3) self.assertEqual(decision["action"], "rebuild") self.assertEqual(MAX_PUSHBACK_ROUNDS, 3) self.assertEqual(MAX_REBUILDS_PER_CONTRACT_VERSION, 1) def test_module_tracker_boundary_via_counts(self): state = new_module_failure_state("mod-web", "v1") for _ in range(3): state = record_module_failure(state, "audit", "v1") self.assertEqual(state["total_fail_count"], 3) state = record_module_failure(state, "audit", "v1") self.assertEqual(state["total_fail_count"], 4) self.assertTrue(state["total_fail_count"] > 3) class RebuildPackageTestCase(unittest.TestCase): def _rebuild_decision(self): return decide_pushback(passing_audit(verdict="fail"), 3) def test_rebuild_scopes_to_one_module_only(self): package = build_rebuild_package( "mod-web", self._rebuild_decision(), [], [], old_implementation_ref="abc1234", other_module_ids=["mod-ios", "mod-android"], contract_version="v1") self.assertEqual(package["scope"]["rebuild_only"], "mod-web") self.assertEqual(package["scope"]["untouched_modules"], ["mod-ios", "mod-android"]) self.assertTrue(package["scope"]["frozen_contract_unchanged"]) self.assertTrue(package["scope"]["design_unchanged"]) self.assertTrue(package["scope"]["other_modules_passed_artifacts_untouched"]) self.assertTrue(package["scope"]["directory_isolation_unchanged"]) def test_old_implementation_archived_not_deleted(self): package = build_rebuild_package( "mod-web", self._rebuild_decision(), [], [], old_implementation_ref="abc1234") old = package["old_implementation"] self.assertEqual(old["ref"], "abc1234") self.assertFalse(old["deleted_from_history"]) self.assertEqual(old["workspace_action"], "replace_with_new_implementation") def test_rebuild_carries_avoidance_list_and_passed_items(self): history = [passing_audit( verdict="fail", verdict_text="round 0: ci red", failed_items=[{"check": "ci_green", "detail": "checks red on current head"}])] package = build_rebuild_package( "mod-web", self._rebuild_decision(), history, [], old_implementation_ref="abc1234", passed_items=[{"check": "contract_version", "detail": "contract v1 matched"}]) kinds = [entry["kind"] for entry in package["avoidance_list"]] self.assertIn("prior_failure", kinds) self.assertIn("previously_passed_must_not_regress", kinds) self.assertTrue(any("ci red" in (entry.get("detail") or "") or "checks red" in (entry.get("detail") or "") for entry in package["avoidance_list"])) def test_rebuild_package_is_labelled_a_rebuild_not_a_new_task(self): package = build_rebuild_package( "mod-web", self._rebuild_decision(), [], [], old_implementation_ref="abc1234") self.assertEqual(package["kind"], "rebuild") self.assertIn("推倒重做第 1 次", package["task_label"]) self.assertIn("mod-web", package["task_label"]) class RootCauseTestCase(unittest.TestCase): def _contract_round(self): return [{"check": "contract_version", "root_cause": "contract", "evidence": {"contract_clause": "G2 §3 timeout"}}] def test_repeated_same_clause_classifies_contract(self): cause, _evidence = classify_root_cause( [self._contract_round(), self._contract_round()]) self.assertEqual(cause, "contract") def test_single_contract_round_is_unknown_not_refreeze(self): cause, _evidence = classify_root_cause([self._contract_round()]) self.assertEqual(cause, "unknown") def test_implementation_rounds_classify_implementation(self): rounds = [[{"check": "ci_green", "root_cause": "implementation"}], [{"check": "ci_green", "root_cause": "implementation"}]] cause, _evidence = classify_root_cause(rounds) self.assertEqual(cause, "implementation") def test_contract_root_cause_at_over_limit_goes_to_refreeze(self): decision = decide_pushback(passing_audit(verdict="fail"), 3, root_cause="contract") self.assertEqual(decision["action"], "escalate_refreeze") self.assertEqual(decision["escalation_target"], "coordinator") self.assertEqual(decision["refreeze_flow"], "G2") self.assertTrue(decision["merge_blocked"]) self.assertIsNone(decision["rebuild_no"]) def test_proved_unimplementable_twice_classifies_design(self): rounds = [[{"check": "contract_version", "root_cause": "design", "proved_unimplementable": True}], [{"check": "contract_version", "root_cause": "design", "proved_unimplementable": True}]] cause, _evidence = classify_root_cause(rounds) self.assertEqual(cause, "design") class RebuildLimitAndCostTestCase(unittest.TestCase): def test_rebuilt_module_failing_over_three_again_escalates(self): # New implementation, fresh count, fails > 3 again, rebuild # budget already used -> escalate, no second automatic rebuild. decision = decide_pushback(passing_audit(verdict="fail"), 3, rebuilds_used=1) self.assertEqual(decision["action"], "escalate") self.assertEqual(decision["escalation_target"], "coordinator") def test_rebuild_exceeding_quota_blocks_without_consuming_rebuild(self): decision = decide_pushback( passing_audit(verdict="fail"), 3, quota_used=90, quota_limit=100, rebuild_cost_estimate=50) self.assertEqual(decision["action"], "blocked_quota") self.assertTrue(decision["merge_blocked"]) def test_rebuild_cost_above_user_threshold_escalates_to_user(self): decision = decide_pushback( passing_audit(verdict="fail"), 3, rebuild_cost_estimate=500, cost_escalation_threshold=100) self.assertEqual(decision["action"], "escalate") self.assertEqual(decision["escalation_target"], "user") self.assertFalse(decision["escalation_grants_user_acceptance"]) self.assertIsNone(decision["rebuild_no"]) class RebuildLedgerTimelineTestCase(unittest.TestCase): def test_timeline_shows_pushbacks_then_rebuild_then_reaudit(self): entries = [] prev = "" for round_no in (0, 1, 2): entry = build_ledger_entry( "pushback", "task-1", round_no, trigger_checks=["ci_green"], module_id="mod-web", contract_version="v1", fail_counts={"audit": round_no + 1, "acceptance": 0, "total": round_no + 1}, prev_hash=prev) entries.append(entry) prev = entry["entry_hash"] rebuild = build_ledger_entry( "rebuild", "task-1", 3, module_id="mod-web", contract_version="v1", rebuild_no=1, old_implementation_ref="abc1234", prev_hash=prev) entries.append(rebuild) reaudit = build_ledger_entry( "reaudit", "task-1", 3, reaudit_result="pass", module_id="mod-web", contract_version="v1", prev_hash=rebuild["entry_hash"]) entries.append(reaudit) valid, errors = validate_ledger_chain(entries) self.assertTrue(valid, errors) self.assertEqual([entry["kind"] for entry in entries], ["pushback", "pushback", "pushback", "rebuild", "reaudit"]) self.assertEqual(entries[3]["rebuild_no"], 1) self.assertEqual(entries[3]["old_implementation_ref"], "abc1234") class TriageTestCase(unittest.TestCase): def _audit(self, failed_items): return {"verdict": "fail", "verdict_text": "round 3 audit", "failed_items": failed_items} def test_structural_kind_triggers_early_rebuild(self): triage = triage_third_failure( self._audit([{"check": "ci_green", "structural_kind": "architecture_boundary"}]), [[{"check": "ci_green", "structural_kind": "architecture_boundary"}]], passed_items=[{"check": "x"}, {"check": "y"}], broken_passed_items=[]) self.assertEqual(triage["verdict"], "early_rebuild") self.assertIn("round 3 audit", triage["basis"]["audit_verdict_text"]) def test_repeated_contract_clause_triggers_early_refreeze(self): item = {"check": "contract_version", "root_cause": "contract", "evidence": {"contract_clause": "G2 §3 timeout"}} triage = triage_third_failure( self._audit([item]), [[item], [item]], passed_items=[{"check": "x"}], broken_passed_items=[]) self.assertEqual(triage["verdict"], "early_refreeze") decision = decide_pushback(passing_audit(verdict="fail"), 2, triage_verdict="early_refreeze") self.assertEqual(decision["action"], "escalate_refreeze") self.assertEqual(decision["refreeze_flow"], "G2") def test_near_complete_gets_one_targeted_round_only(self): audit = self._audit([{"check": "ci_green"}]) triage = triage_third_failure( audit, [[{"check": "ci_green"}]], passed_items=[{"check": "a"}, {"check": "b"}, {"check": "c"}], broken_passed_items=[]) self.assertEqual(triage["verdict"], "targeted_fix") self.assertEqual(triage["targeted_scope"], ["ci_green"]) decision = decide_pushback(passing_audit(verdict="fail"), 2, triage_verdict="targeted_fix", triage_scope=["ci_green"]) self.assertEqual(decision["action"], "pushback") self.assertEqual(decision["pushback_scope"], "targeted") self.assertEqual(decision["targeted_items"], ["ci_green"]) # Targeted budget used: a later near-complete triage is # an ordinary pushback, not a second targeted round. triage2 = triage_third_failure( audit, [[{"check": "ci_green"}]], passed_items=[{"check": "a"}, {"check": "b"}, {"check": "c"}], broken_passed_items=[], targeted_already_used=True) self.assertEqual(triage2["verdict"], "normal_pushback") def test_targeted_round_failure_counts_to_four_then_rebuilds(self): state = new_module_failure_state("mod-web", "v1") for _ in range(3): state = record_module_failure(state, "audit", "v1") state = record_module_targeted_fix(state) # Targeted round fails: that failure is failure 4. state = record_module_failure(state, "audit", "v1") self.assertEqual(state["total_fail_count"], 4) decision = decide_pushback(passing_audit(verdict="fail"), 3) self.assertEqual(decision["action"], "rebuild") def test_too_many_failures_or_broken_passed_is_not_near_complete(self): many = self._audit([{"check": "a"}, {"check": "b"}, {"check": "c"}]) triage = triage_third_failure( many, [[{"check": "a"}]], passed_items=[{"check": "x"}] * 5, broken_passed_items=[]) self.assertEqual(triage["verdict"], "normal_pushback") broken = triage_third_failure( self._audit([{"check": "ci_green"}]), [[{"check": "ci_green"}]], passed_items=[{"check": "a"}, {"check": "b"}], broken_passed_items=[{"check": "a"}]) self.assertEqual(broken["verdict"], "normal_pushback") unknown = triage_third_failure( self._audit([{"check": "ci_green"}]), [[{"check": "ci_green"}]], passed_items=[{"check": "a"}, {"check": "b"}], broken_passed_items=None) self.assertEqual(unknown["verdict"], "normal_pushback") def test_early_rebuild_consumes_rebuild_budget(self): decision = decide_pushback(passing_audit(verdict="fail"), 2, triage_verdict="early_rebuild") self.assertEqual(decision["action"], "rebuild") self.assertEqual(decision["rebuild_no"], 1) state = new_module_failure_state("mod-web", "v1") state = record_module_rebuild(state, 1) # Rebuild resets failure counts for the new implementation. self.assertEqual(state["total_fail_count"], 0) self.assertEqual(state["rebuilds_used"], 1) # ...but no second rebuild after fresh failures over 3. decision2 = decide_pushback(passing_audit(verdict="fail"), 3, rebuilds_used=1) self.assertEqual(decision2["action"], "escalate") class RetainedAssetsTestCase(unittest.TestCase): def _package(self): decision = decide_pushback(passing_audit(verdict="fail"), 3) return build_rebuild_package( "mod-web", decision, [], [], old_implementation_ref="abc1234", passed_items=[{"id": "acc-1", "check": "contract_version"}], retained_tests=["test_web_smoke", "test_contract"]) def test_rebuild_package_retains_tests_and_passed_items(self): package = self._package() self.assertEqual(package["retained_assets"]["tests"], ["test_web_smoke", "test_contract"]) self.assertFalse(package["retained_assets"]["retained_tests_missing"]) self.assertTrue(package["retained_assets"]["contract_unchanged"]) self.assertTrue(package["retained_assets"]["design_unchanged"]) def test_package_without_retained_tests_is_flagged(self): decision = decide_pushback(passing_audit(verdict="fail"), 3) package = build_rebuild_package("mod-web", decision, [], []) self.assertTrue(package["retained_assets"]["retained_tests_missing"]) def test_first_gate_requires_all_old_tests_and_acceptance(self): gate = check_rebuild_first_gate( ["test_web_smoke", "test_contract"], [{"id": "acc-1"}], {"test_web_smoke": True, "test_contract": True}, {"acc-1": True}) self.assertTrue(gate["gate_pass"]) missing = check_rebuild_first_gate( ["test_web_smoke", "test_contract"], [], {"test_web_smoke": True}, {}) self.assertFalse(missing["gate_pass"]) self.assertIn("test:test_contract", missing["missing"]) failed = check_rebuild_first_gate( ["test_web_smoke"], [{"id": "acc-1"}], {"test_web_smoke": False}, {"acc-1": True}) self.assertFalse(failed["gate_pass"]) self.assertIn("test:test_web_smoke", failed["failed"]) # First-gate failure counts as an audit failure. state = new_module_failure_state("mod-web", "v1") state = record_module_failure(state, "audit", "v1") self.assertEqual(state["total_fail_count"], 1) def test_test_deletion_needs_refreeze_and_separate_audit(self): delete = [{"action": "delete", "test_id": "test_old"}] self.assertTrue(validate_test_changes(delete)) self.assertTrue(validate_test_changes( delete, refreeze_approved=True)) # The change itself must name its separate audit; a bare # set of audit ids without a reference on the change proves # nothing about this change. self.assertTrue(validate_test_changes( delete, refreeze_approved=True, separate_audit_ids={"audit-9"})) self.assertEqual(validate_test_changes( [{"action": "delete", "test_id": "test_old", "separate_audit_id": "audit-9"}], refreeze_approved=True, separate_audit_ids={"audit-9"}), []) change = [{"action": "modify_assertion", "test_id": "t", "separate_audit_id": "audit-9"}] self.assertTrue(validate_test_changes( change, refreeze_approved=True)) self.assertEqual(validate_test_changes( [{"action": "add", "test_id": "test_new"}]), []) class AuditOfAuditTestCase(unittest.TestCase): def test_golden_set_rejects_examples_and_requires_human(self): valid, _reasons = validate_golden_task_set( [{"id": "gold-1", "human_approved_by": "user"}]) self.assertTrue(valid) valid2, reasons2 = validate_golden_task_set( [{"id": "sample-1", "is_example": True, "human_approved_by": "user"}]) self.assertFalse(valid2) self.assertTrue(reasons2) valid3, _r3 = validate_golden_task_set([{"id": "gold-2"}]) self.assertFalse(valid3) def test_sampling_covers_golden_mandatory_and_random(self): # Unified rule (gap 2): mandatory kinds carry their own # trigger names (no "critical_path" label for kinds, no # "periodic_sample"); ordinary records are drawn by seeded # random sampling with ceil, so 9 ordinary records still # contribute exactly ceil(0.10 * 9) == 1 -- the old # deterministic every-10th rule drew zero from this batch. records = [{"task_id": "gold-1", "kind": "pushback"}] records += [{"task_id": "t-%d" % i, "kind": "pushback"} for i in range(9)] records.append({"task_id": "t-re", "kind": "rebuild"}) selected = select_audit_sample(records, {"gold-1"}, seed="s") reasons = {entry["record"]["task_id"]: entry["reason"] for entry in selected} self.assertEqual(reasons["gold-1"], "golden_task") self.assertEqual(reasons["t-re"], "rebuild") ordinary = [t for t in reasons if t.startswith("t-") and t != "t-re"] self.assertEqual(len(ordinary), 1) self.assertNotIn("periodic_sample", set(reasons.values())) self.assertNotIn("critical_path", set(reasons.values())) with self.assertRaises(ValueError): select_audit_sample(records, {"gold-1"}) def test_correction_preserves_original_and_restores_flow(self): original = {"task_id": "t-1", "round": 0, "verdict": "pass", "signer": "auditor-a", "entry_hash": "h" * 64} correction, errors = build_audit_correction( original, "fail", "auditor-b", "sample re-run found ci red") self.assertEqual(errors, []) self.assertEqual(original["verdict"], "pass") self.assertEqual(correction["original_entry_hash"], "h" * 64) self.assertEqual(correction["restored_action"], "retrigger_pushback_and_reblock") wrong_block, _e = build_audit_correction( dict(original, verdict="fail"), "pass", "auditor-b", "sample shows the failure was a fixture artefact") self.assertEqual(wrong_block["restored_action"], "schedule_reaudit") _bad, bad_errors = build_audit_correction( original, "fail", "auditor-a", "same signer") self.assertTrue(bad_errors) def test_g6_guard_blocks_non_user_acceptance(self): self.assertTrue(check_g6_guard( {"g6_acceptance": True, "g6_accepted_by": "coordinator"})) self.assertEqual(check_g6_guard( {"g6_acceptance": True, "g6_accepted_by": "user"}), []) self.assertEqual(check_g6_guard({}), []) class AcceptanceRebuildNoticeTestCase(unittest.TestCase): def test_unconfirmed_acceptance_rebuild_does_not_start(self): decision = decide_pushback( passing_audit(verdict="fail"), 3, acceptance_triggered=True, user_confirmed_rebuild=False) self.assertEqual(decision["action"], "rebuild_pending_user_confirmation") self.assertTrue(decision["merge_blocked"]) self.assertIsNone(decision["rebuild_no"]) package = build_rebuild_package( "mod-web", dict(decision, rebuild_no=1), [], [], retained_tests=["test_web_smoke"]) notice = build_acceptance_rebuild_notice( "mod-web", dict(decision, rebuild_no=1), package, ["acceptance failed 4 times"]) self.assertFalse(notice["confirmed"]) self.assertTrue(notice["requires_user_confirmation"]) self.assertIn("test_web_smoke", notice["retained_assets"]["tests"]) def test_confirmed_acceptance_rebuild_proceeds(self): decision = decide_pushback( passing_audit(verdict="fail"), 3, acceptance_triggered=True, user_confirmed_rebuild=True) self.assertEqual(decision["action"], "rebuild") notice = build_acceptance_rebuild_notice( "mod-web", decision, {}, ["acceptance failed"]) confirmed = confirm_acceptance_rebuild(notice, True) self.assertTrue(confirmed["confirmed"]) self.assertTrue(confirmed["g6_acceptance_retained_by_user"]) class SkillTextTestCase(unittest.TestCase): """The skill mechanism must actually be in the skill text.""" def test_audit_section_present_in_skill(self): skill = Path(__file__).resolve().parents[1] / "skills" \ / "software-engineering" / "SKILL.md" text = skill.read_text(encoding="utf-8") for phrase in ( "独立审计", "自动推回", "CONDITIONAL", "三轮", "完整重审", "审计记录", "推倒重做", "重新冻结", "根因分流", "旧测试", "审计的审计", ): with self.subTest(phrase=phrase): self.assertIn(phrase, text) if __name__ == "__main__": unittest.main()