diff --git a/cloud/audit_pushback.py b/cloud/audit_pushback.py new file mode 100644 index 0000000..aed9e56 --- /dev/null +++ b/cloud/audit_pushback.py @@ -0,0 +1,1474 @@ +"""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) diff --git a/cloud/qa_matrix.py b/cloud/qa_matrix.py new file mode 100644 index 0000000..442c6fb --- /dev/null +++ b/cloud/qa_matrix.py @@ -0,0 +1,333 @@ +"""QA full-chain checks: self-test, PRD coverage, agent six, +cross-client matrix, G5 gate, regression scope, dual signatures. + +Pure functions, standard library only. This is the QA half of the +audit-hardening track; the audit half lives in +``cloud/audit_pushback.py``. The two are different checks and +neither report substitutes for the other: the QA matrix proves +the business links run on every selected client, the independent +audit re-runs the exact artifacts. The checklist therefore has +two separate signature seats (QA matrix signer, independent +audit signer) and both must be present and different, or the +round does not pass. + +Honest exceptions, written as rules rather than judgement calls: + +- On a Linux host, iOS / Android / macOS cannot be compiled. + Their honest verdict is ``"code_review_plus_web_core"`` -- never + ``"pass"`` / ``"build_pass"``. An untested client is never a + pass, even with a reason: the reason makes the matrix + *accounted for*, it does not make QA pass. +- QA passing is not user acceptance. G6 acceptance is ticked + only by the user; QA, audit, pushback, and rebuild can never + tick it (see ``check_g6_guard`` in ``audit_pushback``). + +Counting relation to pushback / rebuild (single story, no second +counter): one failed round counts once for its module, however +many QA links failed in that round -- links are failed items +inside the round, not extra failures. A standalone G5 QA failure +(without an audit round) runs the G5 defect loop (back to G4, +re-regress in G5) and does not by itself increment the module's +audit/acceptance failure count; it increments the count only +when the round reaches an audit failure or a QA-acceptance +sign-off failure, recorded with source ``"audit"`` or +``"acceptance"`` respectively. A targeted fix round and a +rebuild both re-run the QA matrix: the re-audit must re-run the +failed links listed in the pushback package (see +``check_reaudit_reran_links``), not just unit tests. +""" + +# Per-client self-test, from the v0.2 skill text (production skill). +SELF_TEST_PARTS = ("unit", "contract_integration", "manual_smoke") +MANUAL_SMOKE_STEPS = ("startup", "login", "core_agent_task", + "error_prompt") + +# Agent-specific six: one missing means the test round is incomplete. +AGENT_SIX = ( + "sandbox_escape", + "dangerous_action_approval", + "long_task_timeout_cancel", + "offline_resume", + "model_503_retry", + "token_quota_exceeded", +) +AGENT_SIX_LABELS = { + "sandbox_escape": "沙箱越界", + "dangerous_action_approval": "危险动作审批", + "long_task_timeout_cancel": "长任务超时和取消", + "offline_resume": "断网续跑", + "model_503_retry": "模型 503 重试", + "token_quota_exceeded": "token / 配额超限", +} + +# G5 pass gates, from the v0.2 skill text. +G5_GATES = ( + "unit", + "integration", + "cross_client_consistency", + "agent_security", + "long_task", + "performance_and_security", +) + +# Clients that cannot be compiled on a Linux host (v0.2 skill text). +UNCOMPILED_CLIENTS = ("ios", "android", "macos") + +# Fields compared across clients for the same task and data. +CONSISTENCY_FIELDS = ("result", "status_word", "sync_state") + + +def _as_bool_map(values): + if isinstance(values, dict): + return values + out = {} + for entry in values or []: + key = entry.get("id") or entry.get("key") or entry.get("name") + out[key] = bool(entry.get("tested", entry.get("passed", False))) + return out + + +def check_self_test(evidence): + """Per-client self-test completeness. Returns (complete, missing). + + ``evidence``: ``{"unit": bool, "contract_integration": bool, + "manual_smoke": {"startup": bool, "login": bool, + "core_agent_task": bool, "error_prompt": bool}}``. All three + parts are required; a missing smoke step is listed by name. + """ + missing = [] + if not evidence.get("unit"): + missing.append("unit") + if not evidence.get("contract_integration"): + missing.append("contract_integration") + smoke = evidence.get("manual_smoke") or {} + for step in MANUAL_SMOKE_STEPS: + if not smoke.get(step): + missing.append("manual_smoke:%s" % step) + return (not missing, missing) + + +def check_prd_coverage(prd_criteria, cases): + """PRD coverage: every criterion needs a valid case. + + A case is valid only when it carries a precondition, steps, + and an expected result, all non-empty. A criterion covered + only by an invalid case is still missing. Returns + ``{"covered": [...], "missing_criteria": [...], + "invalid_cases": [...]}``. + """ + valid_cover = {} + invalid = [] + for case in cases or []: + case_id = case.get("id", "?") + if case.get("precondition") and case.get("steps") \ + and case.get("expected"): + for criterion in case.get("covers") or []: + valid_cover.setdefault(criterion, []).append(case_id) + else: + invalid.append(case_id) + missing = [criterion for criterion in prd_criteria or [] + if criterion not in valid_cover] + return {"covered": sorted(valid_cover), + "missing_criteria": missing, + "invalid_cases": invalid} + + +def check_agent_six(agent_items): + """Agent-specific six completeness. Returns missing keys.""" + covered = _as_bool_map(agent_items) + return [key for key in AGENT_SIX if not covered.get(key)] + + +def check_cross_client_matrix(selected_clients, rows): + """Cross-client matrix: accounted-for vs QA-pass, kept apart. + + ``rows``: one per client, ``{"client": ..., "verdict": + "pass" | "untested" | "code_review_plus_web_core", + "untested_reason": ...}``. Returns ``{"accounted": bool, + "qa_pass": bool, "missing_clients": [...], "violations": + [...]}`` where ``accounted`` means every selected client has + a row (and a reason when untested), while ``qa_pass`` additionally + requires no untested row and no uncompiled client claiming a + build pass. An untested client with a reason is accounted but + does not pass; an untested client without a reason is neither. + """ + by_client = {row.get("client"): row for row in rows or []} + missing_clients = [client for client in selected_clients or [] + if client not in by_client] + violations = [] + has_untested = False + for client in selected_clients or []: + row = by_client.get(client) + if row is None: + continue + verdict = row.get("verdict") + if verdict == "untested": + has_untested = True + if not row.get("untested_reason"): + violations.append( + "untested without reason:%s" % client) + elif verdict == "code_review_plus_web_core": + if client not in UNCOMPILED_CLIENTS: + violations.append( + "code_review verdict on compilable client:%s" % client) + elif verdict == "pass": + if client in UNCOMPILED_CLIENTS: + violations.append( + "uncompiled client claims build pass:%s" % client) + else: + violations.append("unknown verdict:%s:%s" % (client, verdict)) + accounted = not missing_clients and not any( + "untested without reason" in violation for violation in violations) + qa_pass = accounted and not has_untested and not violations + return {"accounted": accounted, "qa_pass": qa_pass, + "missing_clients": missing_clients, + "violations": violations} + + +def check_consistency(rows): + """Same task, same data, across clients: consistent or classified. + + ``rows``: ``{"client": ..., "result": ..., "status_word": ..., + "sync_state": ..., "classification": "defect" | "intentional" + | None, "basis": ...}``. A field that differs across clients + must be classified: ``"defect"`` fails QA (it must be fixed), + ``"intentional"`` passes only with a non-empty basis -- + "intentional" without a basis is a defect hiding behind a word. + Returns ``{"consistent_or_explained": bool, "mismatches": [...], + "blocking_defects": [...]}``. + """ + mismatches = [] + blocking = [] + for field in CONSISTENCY_FIELDS: + values = {} + for row in rows or []: + values.setdefault(row.get(field), []).append(row.get("client")) + if len(values) <= 1: + continue + # The first row's value is the reference; only deviating + # rows need a classification. The reference row itself + # carries no classification and must not be flagged. + baseline = (rows or [{}])[0].get(field) + for row in rows or []: + if row.get(field) == baseline: + continue + classification = row.get("classification") + basis = row.get("basis") + if classification == "defect": + blocking.append("%s defect on %s" + % (field, row.get("client"))) + elif classification == "intentional" and not basis: + blocking.append("%s intentional without basis on %s" + % (field, row.get("client"))) + elif classification not in ("defect", "intentional"): + blocking.append("%s unclassified mismatch on %s" + % (field, row.get("client"))) + mismatches.append({"field": field, "values": values}) + return {"consistent_or_explained": not blocking, + "mismatches": mismatches, "blocking_defects": blocking} + + +def check_g5(g5_evidence): + """G5 gate: all gates pass and no severe defect unfixed. + + ``g5_evidence``: ``{"gates": {gate: bool}, "failures": + [{"severity": "severe" | ..., "fixed": bool, "owner": ..., + "reproduction": ...}]}``. A severe defect that is not fixed + blocks the recommendation to enter acceptance, however green + the gates look. Returns ``{"g5_pass": bool, "missing_gates": + [...], "blocking_defects": [...]}``. + """ + gates = g5_evidence.get("gates") or {} + missing = [gate for gate in G5_GATES if not gates.get(gate)] + blocking = [] + for failure in g5_evidence.get("failures") or []: + if failure.get("severity") == "severe" and not failure.get("fixed"): + blocking.append(failure) + return {"g5_pass": not missing and not blocking, + "missing_gates": missing, "blocking_defects": blocking} + + +def check_qa_fix_role(fix_records): + """QA must not fix code itself. Returns violations. + + ``fix_records``: ``{"defect": ..., "fixed_by": role}``. A fix + recorded as fixed by QA is a violation: QA reproduces and + verifies, the owning client/architecture thread fixes, and + the fix returns to G5 for regression. + """ + return ["qa fixed code directly:%s" % record.get("defect") + for record in fix_records or [] + if record.get("fixed_by") == "qa"] + + +def plan_regression(impact_areas, cases): + """Regression scope from the contract-change impact surface. + + Returns ``{"regression_cases": [...], "uncovered_areas": [...], + "is_full_retest": False}``. Only cases covering at least one + impact area are selected -- never a full retest by default. + Every impact area must have at least one regression case, or + it is listed uncovered. The impact list must include defect + areas even when the contract did not change; an empty impact + list selects nothing, it does not silently mean "everything". + """ + selected = [] + covered = set() + for case in cases or []: + covers = set(case.get("covers") or []) + hit = covers & set(impact_areas or []) + if hit: + selected.append(case.get("id")) + covered |= hit + uncovered = [area for area in impact_areas or [] + if area not in covered] + return {"regression_cases": selected, + "uncovered_areas": uncovered, + "is_full_retest": False} + + +def check_signatures(qa_signer, audit_signer): + """Dual signature seats: QA matrix + independent audit. + + Both seats must be filled and by different people. A missing + seat, or one person holding both seats, fails the round: the + QA report can never substitute for the audit, nor the reverse. + Returns violations (empty means both seats valid). + """ + violations = [] + if not qa_signer: + violations.append("missing qa matrix signature") + if not audit_signer: + violations.append("missing independent audit signature") + if qa_signer and audit_signer and qa_signer == audit_signer: + violations.append("qa and audit signatures by the same person") + return violations + + +def check_reaudit_reran_links(pushback_package, rerun_results): + """Re-audit must re-run the failed links, not just unit tests. + + ``rerun_results`` maps link name -> bool (or a list of + ``{"link": ..., "passed": bool}``). A failed link with no + re-run result fails closed; a failed re-run fails the + re-audit. Returns ``{"reran_all_failed_links": bool, + "missing": [...], "failed": [...]}``. + """ + results = _as_bool_map([ + {"id": key, "passed": value} + for key, value in rerun_results.items()] if isinstance( + rerun_results, dict) else [ + {"id": entry.get("link"), "passed": entry.get("passed")} + for entry in rerun_results or []]) + required = [] + for link in pushback_package.get("reaudit_must_rerun_links") or []: + if isinstance(link, dict): + required.append(link.get("link") or link.get("name")) + else: + required.append(link) + missing = [link for link in required if link not in results] + failed = [link for link in required + if link in results and not results[link]] + return {"reran_all_failed_links": not missing and not failed, + "missing": missing, "failed": failed} diff --git a/cloud/skills/software-engineering/CHANGELOG.md b/cloud/skills/software-engineering/CHANGELOG.md index dbb8876..ea44486 100644 --- a/cloud/skills/software-engineering/CHANGELOG.md +++ b/cloud/skills/software-engineering/CHANGELOG.md @@ -65,3 +65,21 @@ v0.1 → v0.2 的 5 个 P0 缺口明细无法从本文件追溯，本文件不 01/02/03/05/06/09 章后有 `---`。 - 校验脚本与两份附录测试的 EXPECTED_IDS 同步扩为 11 章；#214 的 “仅复制 SKILL.md”结论不受影响（播种行为未改）。 + +## #244 审计段落向 v0.5 的移植（集成时做，移植准备轨 #258 的 §1–§3 为准） + +- A 矩阵行：插入 QA 行之后、验收行之前（v0.5 矩阵行本身不动）。 +- B 主节：插在“失败、续跑与成本”节末、“诚实边界”节首之前； + 引言移植警告按 R1 改写为移植注；“与已有回路的关系”按 v0.5 + 实际回路对齐（G7、全自动在本文件已有定义，注意句相应改写， + 全自动下执行人字段口径见集成摘要：full-auto-executor + + installation id，签字人与执行人仍须不同）；B2、B3 全并， + 唯收紧三末段“验收失败导致的推倒要先通知用户”整段停下 + （全自动四类回头问用户未含此情形，须用户拍板 (a)/(b)，移植轨 + 不擅自选）；B4 QA 全链路整段停下（跨端版与 v0.5 跨视口口径正面 + 冲突，须视口版重写后再移植）；C 只插审计项，QA 项随 B4 停下。 +- D 附录章：落为 11-auditor（v0.5 已有 09-ops / 10-iteration， + 编号顺延），标题用严格格式，源文件为 + `templates/roles/11-auditor.md`（正文与 #258 的 09-auditor 同源）； + 校验表同步扩为 12 章。§4 的 templates/artifacts 删引用决策 + 未在集成基底执行（仍待用户拍板），SKILL.md 内 6 处引用原样保留。 diff --git a/cloud/skills/software-engineering/SKILL.md b/cloud/skills/software-engineering/SKILL.md index 23067fe..e061966 100644 --- a/cloud/skills/software-engineering/SKILL.md +++ b/cloud/skills/software-engineering/SKILL.md @@ -357,6 +357,7 @@ G0 的六问缺一问，G0 就不算过。用户没回复时，协调员等， | 共享核心 | 1 | `packages/shared/`（G3） | 共享类型、SDK、mock 服务、契约测试 | | 客户端开发 | 每客户端 1 | `apps/web`、`apps/macos`、`apps/windows`、`apps/ios`、`apps/android` 之一 | 该客户端可运行产物 + 自测证据；v0.5 默认只开 web，iOS / Android 原生后置占位，不在 MVP 内 | | QA / 测试 | 1（跨视口 E2E 可再分 1 个） | `tests/` | 测试用例、跨端一致性矩阵（v0.5 行是跨视口）、缺陷清单 | +| 独立审计 | 1，与 QA、协调员、执行人分开 | 只写审计记录与推回包，不写业务代码 | 独立重跑证据 + 单独签字的审计结论（见“独立审计与自动推回”） | | 验收 / 发布 | 协调员兼任，用户拍板 | `docs/release/` | 验收清单结果、web 单端发布说明、已知问题清单 | | 运维 | 1（G7） | `docs/ops/` | 运维手册、问题清单、热修复记录、回滚触发结论 | | 迭代 | 1（G8） | `docs/` | v1.1 交接清单终版、下一轮 G0-G6 输入冻结清单 | @@ -626,6 +627,311 @@ tests/ # QA 线程专属，含跨端一致性用例 较经济的模型，具体以项目实际可用模型和配额为准； 配额将满时先停新线程，不先透支再解释。 +## 独立审计与自动推回 + +本节是审计加固第 3 步的技能机制，与代码 `cloud/audit_pushback.py` +（推回决策纯函数）双轨落地，只写文档不算落地。只补审计机制， +不改上面已有规则的语义：G5 缺陷回路、合并拆分、G6 用户验收 +的原意都不变。移植注：本节源自审计加固轨，移植时已按本文件对应章节的回路口径对齐。 + +### 独立审计角色：不写代码，只独立重跑、抽查证据、单独签字 + +- 审计线程不写业务代码、不改待办、不给合并决定、不执行合并。 + 它只做三件事：独立重跑测试（不只复读 QA 报告）、抽查证据、 + 出一份单独签字的审计结论。谁写、谁测、谁审、谁执行、谁签字 + 五职分开：写是开发线程，测是 QA 线程，审是审计线程， + 执行合并/发布是人或宿主操作，签字是审计签字人。 +- 同一人不兼两职，可核查：审计记录的签字人字段必须与任务执行人 + 字段、协调员批准人字段、QA 测试人字段都不同，有一项相同， + 审计结论本身无效，按审计 fail 处理。代码里这就是 + `validate_signer_separation`，缺字段也按无效处理，不补猜。 +- 安全签字只有两种结果：无条件通过，或拦住。CONDITIONAL 通过 + 不算通过，仍触发推回。安全签字是 CONDITIONAL 时，哪怕审计 + 结论正文写了“通过”，也按 fail 处理。 +- 已知偏差必须有到期与跟进 todo，不能永久接受：偏差清单里 + 每一项都要有跟进 todo 和到期日；无到期跟进 todo 的偏差、 + 或到期未跟进的偏差，都触发推回。“已知偏差已接受”只有在 + 偏差有到期跟进 todo 且未到期时才不触发。 +- 证据标准升级三条，缺一即 fail： + 1. 回归必须跑确切产物，复用跨任务产物传递的三方哈希核对模式 + （核心报告 hash、产物清单 hash、协调员重算 hash 三方一致 + 才算数，清单 hash 不一致即停）；不用 fixture 替身跑回归， + fixture 模拟的项必须单独列为“未实测”，绝不写成通过。 + 2. 演示必须可运行验证：要跑应用自己的入口、跑出真实流程； + 不能用通用 `http.server` 冒充演示（只起文件服务器、不跑 + 应用入口的，一律算演示未通过）。 + 3. 缺失的门禁检查按 fail 处理，不按通过补猜：CI、目录隔离、 + 合同版本、协调员批准、未解决评审意见，缺一项证据即 fail。 + +### 自动推回：审计不过关，系统自动推回，不等人工点头 + +触发条件表（任一成立即触发，CONDITIONAL 不算通过）： + +| 触发项 | 含义 | +|--------|------| +| 独立审计签字为 fail | 审计结论本身不通过 | +| CI 未绿 | 含 CI 账单故障秒级失败的情形：没绿就是没绿，不假装账单已恢复、不拿旧 CI 顶替当前头 | +| 目录隔离违规 | 分支改了非拥有目录且未经批准的契约变更 | +| 合同版本不符 | 不是冻结版合同 | +| 协调员批准缺失 | 无协调员合并决定 | +| 未解决评审意见 | PR 里有未处理/未回复的评审意见 | +| 安全签字非无条件通过 | CONDITIONAL 或 fail | +| 回归未跑确切产物 | hash 不一致、或用了 fixture 替身 | +| 演示未可运行验证 | 演示跑不起来、或用通用文件服务器冒充 | +| 已知偏差无到期跟进 todo | 无 todo、无到期日、或到期未跟进 | +| 审计签字人不独立 | 签字人等于执行人、协调员批准人或 QA 测试人任一 | + +- 推回动作是自动的：审计 fail 后系统自动把任务推回重新优化/ + 修复，不停在“等用户/协调员手动重开”上。推回形式二选一， + 本技能选第 ② 种，并写清取舍： + - 选 ②：原任务保持 completed，但自动派生一个修复/优化子任务， + 并阻塞合并与发布，直到重新审计通过。 + - 否 ①：原任务状态回退到可重跑状态并附审计结论重跑。否的原因： + 回退会抹掉 completed 记录、可能重跑有副作用的步骤、还可能 + 让修复悄悄丢弃或覆盖上一轮已通过的产物；② 保留 completed + 记录与已通过产物原样可查，修复是可追踪的新任务，时间线不断。 + - 无论哪种，合并与发布在审计重新通过前一律阻塞，绝不能出现 + “审计 fail 但照样合并/发布”的路径。与第 1 轨硬门禁的接口约定： + 硬门禁读推回决策的阻塞字段（`merge_blocked` / + `release_blocked`、`blocked_by="audit_pushback"`、轮次、 + 失败检查清单），硬门禁自己的检查与推回阻塞是“与”关系： + 审计通过不解硬门禁的锁，硬门禁通过也不解推回的锁，两边都 + 通过才能合并/发布。代码侧该接口目前是提案（第 1 轨落地时 + 确认接线，在此之前本模块本身不拦任何服务端点，这一点要 + 如实写进 PR 摘要，不冒充已接线）。 +- 推回包必须可执行，缺一项就不算推回完成：审计结论原文； + 失败项逐条（含测试名、文件路径、行号、合同条款、diff 范围）； + 复现步骤；修复验收标准（即重新审计时要过的同一批检查， + 见下“完整重审”）；上一轮产物清单与哈希（复用三方哈希模式， + 修复不得悄悄丢弃或覆盖上一轮已通过的产物，改动处要可 diff， + 改动不在修复 diff 里的已通过产物 hash 变了即违规）。 +- 推回上限与升级：推回修复最多三轮（与全自动 CI 自动修复限 + 3 次同口径，但计数不共用：CI 修复轮次发生在审计前，审计推回 + 轮次发生在审计后，一方不得消耗另一方的预算，代码里 + `ci_auto_fix_rounds` 只作参考输入、不计入推回上限）。 + 超过三轮不再自动推回修复，改走下一节“推倒重做”升级分支； + 但第 3 次失败先按再下一节“三条收紧”的根因分流判定， + 结构性错误可提前推倒、接近完成可 targeted 续轮，分流不改变 + “超过 3 次才推倒”的阈值口径（见收紧一的计数说明）； + 推倒本身也超限后，才升级给协调员人工复核（G6 验收仍只能 + 用户勾，升级绝不能替用户勾验收）。升级时附带全部轮次的 + 审计结论、改动 diff、成本与 token 消耗，不得只报最后一轮。 +- 完整重审：修复后重新跑全部硬门禁与独立审计（CI、目录、 + 合同、回归确切产物、演示可运行、安全），上一轮通过项在修复后 + 也要复核是否被改坏；只复核失败项的重审不算通过。 +- 成本与配额：修复任务本身计成本与配额，超配额即按配额规则中断， + 不因“是修复”就绕过配额。 +- 状态与账本：每轮推回、修复、重审都要写审计记录（轮次、触发项、 + 推回包、修复 diff、重审结果），账本走第 4 轨的防篡改机制； + 本轨先按“审计记录字段 + 校验接口”定义（代码里 + `build_ledger_entry` / `validate_ledger_chain` 的 + `prev_hash` / `entry_hash` 链式格式是提案，第 4 轨落地后接线， + 接口不一致处在 PR 摘要写清）。任务时间线要能看到 + “审计 fail → 自动推回 → 修复 → 重审”的完整链路，不能只有 + 最终通过一条记录把失败轮次抹掉。 +- 与已有回路的关系，不重复造轮子：G5 缺陷回路（缺陷回 G4、 + 修完回 G5 回归）复用为修复执行通道，推回派生的修复子任务 + 就按 G5 方式执行，只补“自动触发 + 合并发布阻塞”这两块缺失； + G7 热修复回路并行（推回是合入前，热修复是发布后应急， + 推回不能替代热修复、热修复也不能抵推回）；全自动 CI 自动 + 修复并行且计数独立（见上限一段）。注意：G7 与全自动 CI 回路在本文件 v0.5 文本里已有定义 + （G7 发布后运维、全自动模式），本段按本文件的实际回路定义对齐； + 推回是合入前、热修复是发布后应急，计数独立。 + +## 推倒重做：超过三次后的升级分支（不是并行回路） + +本节是用户追加的升级分支，与上一节自动推回是同一条链路的两段： +自动推回是第 1–3 轮的修复回路，推倒重做是超过 3 次后的升级分支， +不是并行两套回路。不要把“推回修复”和“推倒重做”混成一回事： +前者是修补同一实现，后者是废弃同一实现、从冻结合同重新实现。 + +### 计数口径：按模块、按合同版本，审核与验收合计 + +- 按模块计数：每个模块（例如 web 客户端、iOS 客户端各算一个模块） + 各自计数，一个模块超限只推倒该模块，不连坐其他模块。 +- 审核 fail 与验收 fail 合并计入同一模块的失败计数：审核指独立审计 + 与硬门禁，验收指 G6 用户验收与 QA 验收；计数时来源分别记录 + （`audit_fail_count` / `acceptance_fail_count` 各记各的）， + 但触发阈值看合计（`total_fail_count`）。 +- 一次通过即清零该模块的失败计数，下一轮重新计；但同一合同版本 + 下的推倒次数不清零（否则“通过—失败”循环可以无限推倒， + 见“重做上限”一段）。 +- 按合同版本计：合同版本变了（重新冻结）计数重新起算——失败计数 + 与推倒次数都清零，因为模块改按另一份合同评判，旧计数不能带过去。 + 版本变化时的处理写死：先按新版本把计数清零，再把触发这次 + 重新计数的失败（如有）记为新版本下的第 1 次。 + +### 阈值语义：恰好 3 次仍推回修复，第 4 次才推倒 + +- 推回修复最多三轮：第 1 到第 3 次没通过走自动推回修复。 +- 累计失败超过 3 次（即第 4 次仍没通过，或同一模块在同一合同版本下 + 累计 fail 次数 > 3）不再修补，直接触发推倒重做。 +- “超过三次”在代码和本文档里一律是 `> 3`，不是 `>= 3`： + 恰好 3 次仍是推回修复，第 4 次才是推倒。代码与测试用 + `round_no` 0、1、2 对应第 1–3 次失败、`round_no >= 3` + 对应第 4 次失败同一口径钉住边界，两处不一致就是不合格。 + +### 推倒重做的含义：只推倒该模块，旧实现归档、新实现带避坑清单 + +- 只推倒该模块：不动其他模块、不动已冻结合同与设计、不动其他 + 已通过模块的产物；目录隔离不变，新实现仍只写该模块的拥有目录。 + 只推倒实现代码：测试、已通过验收项、合同、设计、审计结论、 + 避坑清单一律保留（细则见再下一节“三条收紧”之收紧二， + 新实现第一道门就是旧测试全过）。 +- 该模块的工作代码从冻结合同出发重新实现。旧代码不直接删除历史： + 旧实现以提交 / tag / 归档形式保留在 git 历史与审计记录里供复盘， + 工作区以新实现替换旧实现；审计记录里必须写清新旧实现的对应关系 + （旧实现 ref、新实现分支、替换点），删历史、改写历史冒充 + “从来没失败过”的一律不合格。 +- 上一轮的审计结论、失败原因、已通过项清单要作为新一轮的 + “避坑清单”带进新实现任务包：新实现任务包里逐条列出此前 + 失败项与此前已通过项，绝不能让新实现把同一错误再犯一遍后 + 还声称是全新开始；重审时避坑清单上的项要逐项复核。 + +### 先判根因再决定是重做还是重新冻结 + +- 连续失败先判根因，再决定是重做代码还是重新冻结，不要一超限 + 就无脑重做代码。 +- 根因判定标准（代码 `classify_root_cause` 的可测试版）： + 合同/设计根因成立当且仅当以下任一成立：① 至少两轮审计结论 + 指向同一合同/设计条款且根因为合同/设计；② 修复方至少两轮 + 证明按合同无法实现（`proved_unimplementable`）。只一轮指向 + 合同/设计、或指向条款各不相同，都算根因未定，不触发重新冻结。 +- 如果根因在合同或设计本身（例如合同自相矛盾、设计不可实现、 + 技术栈选错），推倒代码重做是治标不治本：此时不自动重做代码， + 改为触发 G2 重新冻结流程，并升级给协调员人工复核（协调员须把 + 新冻结方案交用户定，升级本身不替用户冻结合同、更不替用户勾 G6）。 +- 根因未定（无两轮以上的合同/设计证据）时默认走推倒重做分支： + 重新冻结需要肯定证据，不能凭猜测冻结；推倒重做是可逆的第一步， + 且新实现包里有避坑清单，不会盲目重来。 +- 测试/检查要能区分“代码实现问题”与“合同/设计问题”： + 两轮同条款合同根因 → 重新冻结；两轮实现根因 → 推倒重做； + 一轮合同指向 → 根因未定 → 推倒重做（见代码测试）。 + +### 重做也有上限，绝不无限重开 + +- 同一模块在同一合同版本下最多推倒 1 次（不是 2 次）：取舍写清—— + 推倒是全量重做、成本最高，1 次已是真正的第二次机会；若新实现 + 再累计失败超过 3 次，问题几乎不在可修补代码里，而在合同/设计 + 或能力本身，第二自动次推倒只会烧成本、掩盖根因，所以不再自动重开。 +- 重做后的新实现重新计失败次数（从第 1 次起算），再累计失败超过 + 3 次即升级，不再自动重开。升级去向：升级给协调员人工复核， + 附带全部轮次审计结论、历次推倒记录、改动 diff、成本与 token + 消耗；G6 验收仍只能用户勾，升级不能替用户勾验收。 +- 推倒次数与失败计数、推回轮次一样写审计记录（见下）。 + +### 成本与配额：重做是全量重做，成本最高，照常计 + +- 推倒重做仍计成本与配额，超配额即按配额规则中断；超配额的推倒 + 不消耗推倒次数（根本没开始重做，不能算用掉一次机会）。 +- 如预估重做成本超过用户定的成本超阈值 escalation 线，先升级给用户， + 而不是悄悄烧掉：升级时附预估成本与阈值，待用户决定是否重做、 + 缩范围还是停掉；用户没点头前不启动重做。 +- 预估与实际的记录方式：重做任务包里先写预估成本（启动前）， + 完成后补实际成本与 token 消耗，未定价的项绝不记 0（沿用成本 + 双列口径：预估一列、实际一列）。 +- 重做任务包里要写清“这是推倒重做第 N 次、上一轮失败原因”， + 任务标题与任务包 kind 都标明 rebuild，不能伪装成普通新任务。 + +### 状态与账本：计数、推回、推倒、根因都要留痕 + +- 模块失败计数（审核/验收分来源、合计）、推回轮次、推倒次数、 + 重做新旧实现对应关系、根因判定结果都要写审计记录，走第 4 轨 + 的防篡改机制（本轨先按字段 + 校验接口定义：账本条目含 + `module_id`、`contract_version`、`fail_counts`、`rebuild_no`、 + `root_cause`、`old_implementation_ref`，链式校验见代码 + `build_ledger_entry` / `validate_ledger_chain`，第 4 轨落地后接线）。 +- 任务时间线要能看到完整链路：“审计 fail → 自动推回修复 + （1–3 轮）→ 超过 3 次 → 推倒重做 → 重新审计”，不能只有最终 + 通过一条记录；推倒记录、重新冻结升级记录同样留痕，不得抹掉。 +- 与上一节自动推回的关系再写一遍以免混淆：自动推回（1–3 轮修补） + 在前、推倒重做（超过 3 次）在后，是同一链路的前后两段； + 与 G5 缺陷回路的关系是复用修复执行通道（重做任务本身仍按 + 阶段派生、仍走 G5 回归复核），与 G7 热修复回路并行（已发布 + 模块的应急修复走 G7，不走推倒），与全自动 CI 自动修复并行 + 且计数独立（CI 修复轮次不计入模块失败计数、也不计入推倒次数）。 + +## 三条收紧：分流、保留资产、审计的审计 + +本节与自动推回、推倒重做并列，不覆盖已有计数与推倒规则： +阈值仍是 `> 3` 才推倒、推倒每模块每合同版本最多 1 次、G6 验收 +仍只能用户勾。下面三条只把原来可能留主观口子的地方写成 +可测试机制。代码在 `cloud/audit_pushback.py`，判定函数、 +阈值与输入输出见该文件 docstring 与测试。 + +### 收紧一：阈值不能死板，第三次失败先判根因 + +- 固定 3 次只当默认值，不能当真理。第三次失败（累计 fail = 3、 + 尚未触发推倒的那一刻）必须先做一次根因判定，再决定下一条路； + 判定结果与判定依据要写进审计记录，事后可抽查。 +- 判定分两类，标准写死，不留“协调员觉得差不多”这种主观句： + ① 结构性错误 → 提前推倒重做，不必等到第 4 次。结构性指： + 架构边界错、模块职责错、技术栈或数据模型根本性错误——代码里 + 是失败项带 `structural_kind`（四类之一）或触及核心结构； + 但连续多轮指向同一设计/合同条款属于合同/设计结构性， + 按原根因规则转 G2 重新冻结升级，不直接推倒代码（合同优先于 + 代码推倒，因为合同修了架构可能跟着变；两者信号同时出现时 + 也先走重新冻结）。提前推倒计入推倒次数（每模块每合同版本 + 最多 1 次的预算照扣），不是免费的一次。 + ② 接近完成、只差收尾 → 再给一轮 targeted 修复。接近完成的 + 判定条件四条全满足才算：失败项 ≤ 2 项；失败项与核心结构无关 + （无结构性种类、无核心触及、无合同/设计根因）；上一轮已通过项 + 未被破坏（破坏清单为空，且须实测提供、不能缺省当作空）； + 已通过项数量多于失败项数量。targeted 修复范围只限失败项与 + 直接相关代码，不得借机重构无关部分；续轮最多 1 次 + （每模块每合同版本 1 次，代码 `targeted_fix_used`）。 +- 计数方式（两处一致，不得出现两套计数）：续轮算入累计失败计数， + 不单独计数——targeted 轮本身就是第 3 轮推回；该轮再 fail 即 + 累计第 4 次失败，按原规则触发推倒（不得以“接近完成”为由 + 再续一轮）；targeted 预算不计入推倒预算，推倒也不补回 + targeted 预算；推倒后新实现失败计数清零，但 targeted 预算 + 与推倒次数在同一合同版本下都不清零（合同版本变了才全清零）。 +- 判定本身可审计：判定依据（审计结论原文、失败项清单、diff 范围、 + 已通过项清单、已破坏项清单）附在记录里；判定为“接近完成” + 却在续轮后仍 fail 的案例，事后抽查时能从记录里复核当初四条 + 条件是否真的全满足，条件不满足却给了续轮的，记为判定错误 + 并走收紧三的纠错路径。 + +### 收紧二：推倒只推倒实现代码，测试与已通过资产全部保留 + +- 推倒重做的范围写死：只替换该模块的实现代码；测试用例、 + 已通过的验收项、合同、设计、审计结论、避坑清单一律保留， + 不得随实现代码一起清零；新实现任务包里必须包含旧测试集 + 与已通过验收项清单（代码里是重做包的 `retained_assets`， + 没给旧测试集的包标 `retained_tests_missing`，不算合格包）。 +- 新实现的第一道门就是旧测试全过：重做后的实现先跑旧测试集 + 与已通过验收项复核，全过才进入新一轮完整审计；过不了旧测试 + 就不算重做成功，直接记为重做失败并计入失败计数（新实现 + episode 的审核失败计数），不得以“新实现有新功能”为由 + 跳过旧测试；旧测试缺结果按 fail 处理，不按通过补猜。 +- 旧测试本身如因合同变更而失效，须走合同重新冻结流程改测试， + 不得在重做时悄悄删测试或改断言：删测试或改断言必须同时满足 + 两项——合同已走 G2 重新冻结、且该改动有单独审计记录；缺一项 + 即违规（代码 `validate_test_changes`）。新增测试不需要重新冻结， + 但新增不能冒充删除（把旧测试删了再以新名加回来，仍按删除算）。 +- 旧实现归档与新实现对应关系不变：旧实现以提交 / tag / 归档保留， + 新旧实现 diff、测试通过情况、已通过项保留情况都要写进审计记录。 + +### 收紧三：审计本身要被审计，G6 人工验收不能被自动化绕过去 + +- 审计兜底三件套，缺一不可，全部可测试或可核查： + ① 独立重跑的确切产物测试，不是只看报告：复用三方哈希模式 + （核心报告、产物清单、重算 hash 三方一致），产物不一致即停。 + ② 金任务集与关键路径的人工审定样本，定期抽查审计结论本身： + 金任务集必须人工审定（每项有 `human_approved_by`），示例文件 + 不算金任务；抽查比例写死：金任务与推倒/重新冻结/升级记录 + 100% 抽查，普通记录每 10 条确定性抽 1 条（按序号、可复现， + 不是随机抽）；抽查结果处理：抽查发现判错即走纠错路径。 + ③ G6 用户验收这个最后人工关口不能被自动化绕过去：审计升级、 + 推回、推倒、抽查纠错都不能替用户勾验收，验收清单最后只能用户勾。 +- 审计判错的纠错路径：事后抽查发现审计结论判错（该拦没拦、或 + 不该拦拦了），要能纠正该轮审计结论、恢复或重新触发相应流程， + 并把判错记入审计的审计记录，不得只改当轮结果不留痕。实现方式： + 原记录不动，追加一条纠错记录（引用原记录 hash、写清原结论、 + 纠正后结论、纠错人与原因）；该拦没拦的纠正 → 重新触发推回并 + 恢复阻塞；不该拦拦了的纠正 → 安排重新审计（不是直接放行合并， + 因为纠错本身只是复核、不是完整审计）；纠错人必须与原签字人不同。 + ## 诚实边界：原生客户端（v0.5 后置占位版） - 在只有 Linux 主机的环境里，PWA、“添加到主屏幕”、WebView 壳是能做能测的； @@ -771,6 +1077,11 @@ tests/ # QA 线程专属，含跨端一致性用例 回归提示词里产物清单与文件原文双通道齐了吗？回归结尾有 ARTIFACT-VERIFY 一节与 ARTIFACT-VERIFY-OK 吗？回归是跑同一份 核心代码，还是自己写 fixture 猜契约？ +- 审计与推回走了吗？审计签字人是否独立于执行人、协调员批准人、 + QA 测试人？安全签字是不是无条件通过？审计 fail 后有没有自动推回 + 并阻塞合并/发布？模块失败超过 3 次后有没有按推倒重做分支处理 + （先判根因、只推倒该模块、旧实现归档、避坑清单带进新包）， + 而不是无限修补或悄悄放行？ 有一项为否，就停在当前门禁，不进入下一阶段。 ## 附录：角色模板（播种随附版） @@ -1477,3 +1788,44 @@ iOS / Android 原生后置占位不进 QA 通过表，标注“未编译 / 未 Vitest、OpenAPI 或共享类型），G2 冻结后不准中途换，只有 用户特别要求时才换语言；codex-app-server 本身保持 Python， 不为统一语言重写。 + +## 附录角色：11-auditor（来源 templates/roles/11-auditor.md） + +# 角色模板：独立审计（1 个线程，与 QA、协调员、执行人分开） + +你是本项目的独立审计线程。你不写业务代码、不改待办、不给合并决定、 +不执行合并。你只做：独立重跑测试、抽查证据、单独签字。 + +## 你必须先做的事 + +1. 拿到交付物后自己重跑测试，不要只复读 QA 的报告；QA 说通过 + 不等于你通过，你的重跑命令和结果要写进审计记录。 +2. 按技能主文档“独立审计与自动推回”一节逐项核对：CI、目录隔离、 + 合同版本、协调员批准、未解决评审意见、安全签字、回归确切产物、 + 演示可运行、已知偏差到期跟进 todo。缺一项证据即 fail，不补猜。 +3. 签字人就是你，且必须与任务执行人、协调员批准人、QA 测试人 + 都不同；相同即审计结论无效，按 fail 处理。 + +## 你的签字规则 + +- 只有无条件通过，或拦住。CONDITIONAL 不算通过；安全签字 + CONDITIONAL 时结论就是 fail，并触发自动推回。 +- 已知偏差无到期跟进 todo、或到期未跟进时，结论就是 fail。 +- 回归用了 fixture 替身、或 hash 对不上时，结论就是 fail， + fixture 项单独列“未实测”，不写成通过。 +- 演示没跑应用入口（例如只起了通用文件服务器）时，结论就是 fail。 + +## 你的交付物 + +- 审计结论原文、失败项逐条（测试名、文件路径、行号、合同条款、 + diff 范围）、复现步骤、上一轮产物清单与哈希。 +- 审计 fail 时你不等人点头：系统按“自动推回”一节自动派生修复 + 子任务并阻塞合并/发布；你只负责把推回包写完整。 +- 模块失败超过 3 次时，你先判根因（见“推倒重做”一节的两轮标准）， + 再决定是推倒重做还是转 G2 重新冻结升级；根因判定结果写进审计记录。 + +## 禁止 + +- 不替开发线程改代码“顺手修掉”，修了就审不出原缺陷。 +- 不把“没测”写成“通过”，不把 CONDITIONAL 写成通过。 +- 不替用户勾 G6 验收，不替协调员给合并决定。 diff --git a/cloud/skills/software-engineering/templates/roles/11-auditor.md b/cloud/skills/software-engineering/templates/roles/11-auditor.md new file mode 100644 index 0000000..6089126 --- /dev/null +++ b/cloud/skills/software-engineering/templates/roles/11-auditor.md @@ -0,0 +1,38 @@ +# 角色模板：独立审计（1 个线程，与 QA、协调员、执行人分开） + +你是本项目的独立审计线程。你不写业务代码、不改待办、不给合并决定、 +不执行合并。你只做：独立重跑测试、抽查证据、单独签字。 + +## 你必须先做的事 + +1. 拿到交付物后自己重跑测试，不要只复读 QA 的报告；QA 说通过 + 不等于你通过，你的重跑命令和结果要写进审计记录。 +2. 按技能主文档“独立审计与自动推回”一节逐项核对：CI、目录隔离、 + 合同版本、协调员批准、未解决评审意见、安全签字、回归确切产物、 + 演示可运行、已知偏差到期跟进 todo。缺一项证据即 fail，不补猜。 +3. 签字人就是你，且必须与任务执行人、协调员批准人、QA 测试人 + 都不同；相同即审计结论无效，按 fail 处理。 + +## 你的签字规则 + +- 只有无条件通过，或拦住。CONDITIONAL 不算通过；安全签字 + CONDITIONAL 时结论就是 fail，并触发自动推回。 +- 已知偏差无到期跟进 todo、或到期未跟进时，结论就是 fail。 +- 回归用了 fixture 替身、或 hash 对不上时，结论就是 fail， + fixture 项单独列“未实测”，不写成通过。 +- 演示没跑应用入口（例如只起了通用文件服务器）时，结论就是 fail。 + +## 你的交付物 + +- 审计结论原文、失败项逐条（测试名、文件路径、行号、合同条款、 + diff 范围）、复现步骤、上一轮产物清单与哈希。 +- 审计 fail 时你不等人点头：系统按“自动推回”一节自动派生修复 + 子任务并阻塞合并/发布；你只负责把推回包写完整。 +- 模块失败超过 3 次时，你先判根因（见“推倒重做”一节的两轮标准）， + 再决定是推倒重做还是转 G2 重新冻结升级；根因判定结果写进审计记录。 + +## 禁止 + +- 不替开发线程改代码“顺手修掉”，修了就审不出原缺陷。 +- 不把“没测”写成“通过”，不把 CONDITIONAL 写成通过。 +- 不替用户勾 G6 验收，不替协调员给合并决定。 diff --git a/cloud/skills/software-engineering/tools/check_appendix.py b/cloud/skills/software-engineering/tools/check_appendix.py index 7a6df1f..805079b 100644 --- a/cloud/skills/software-engineering/tools/check_appendix.py +++ b/cloud/skills/software-engineering/tools/check_appendix.py @@ -10,10 +10,10 @@ source, then re-issue the appendix through the skill's own track), never a silent rewrite by this script. What it checks, in order: - 1. SKILL.md contains the appendix intro and exactly the eleven - chapters 00..10, in order, each with a source note naming its own + 1. SKILL.md contains the appendix intro and exactly the twelve + chapters 00..11, in order, each with a source note naming its own templates/roles/.md file. - 2. templates/roles/ contains exactly those eleven source files: + 2. templates/roles/ contains exactly those twelve source files: a missing source fails, an unexpected extra file fails. 3. Each source, in canonical form, equals the matching appendix chapter in canonical form (see canonical() below). @@ -29,7 +29,7 @@ and its appendix chapter -- anything else is a failure): b. One inter-chapter separator line "---": in the current SKILL.md it appears after chapters 01/02/03/05/06/09 only. Chapters 00/04/07 end with their GitHub subsection and run straight into - the next heading with no separator, and 10-iteration is the end + the next heading with no separator, and 11-auditor is the end of file. The check strips at most one trailing separator; it does NOT normalise SKILL.md to a uniform separator, so the current @@ -64,7 +64,8 @@ from pathlib import Path EXPECTED_IDS = ["00-coordinator", "01-product-manager", "02-architect", "03-designer", "04-client-developer", "05-backend", "06-qa", "07-acceptance", "08-shared-core", - "09-ops", "10-iteration"] + "09-ops", "10-iteration", + "11-auditor"] CHAPTER_RE = re.compile( r"^## 附录角色：(\d{2}-[a-z0-9-]+)" diff --git a/cloud/tests/test_audit_pushback.py b/cloud/tests/test_audit_pushback.py new file mode 100644 index 0000000..fec07c0 --- /dev/null +++ b/cloud/tests/test_audit_pushback.py @@ -0,0 +1,883 @@ +"""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_critical_and_periodic(self): + 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"}) + reasons = {entry["record"]["task_id"]: entry["reason"] + for entry in selected} + self.assertEqual(reasons["gold-1"], "golden_task") + self.assertEqual(reasons["t-re"], "critical_path") + # 10th record (index 9) is the periodic sample. + self.assertIn("t-8", reasons) + + 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() diff --git a/cloud/tests/test_qa_matrix.py b/cloud/tests/test_qa_matrix.py new file mode 100644 index 0000000..2a984a8 --- /dev/null +++ b/cloud/tests/test_qa_matrix.py @@ -0,0 +1,247 @@ +"""QA full-chain checks: self-test, PRD coverage, agent six, +cross-client matrix, G5 gate, regression scope, dual signatures. + +Pins the seven required behaviours: (1) PRD full coverage, +(2) agent-specific six complete, (3) cross-client matrix covers +every selected client, (4) a missing client states its reason, +(5) untested is never written as passed, (6) the uncompiled +platform verdict is code-review-plus-web-core, never build-pass, +(7) defect regression follows the impact surface, not a full +retest by default. Plus: self-test three parts, consistency +classification, QA never fixing code, and the dual-signature +rule (QA matrix seat vs independent audit seat). +""" + +import sys +import unittest +from pathlib import Path + +try: + from cloud.qa_matrix import ( + check_agent_six, + check_consistency, + check_cross_client_matrix, + check_g5, + check_prd_coverage, + check_qa_fix_role, + check_reaudit_reran_links, + check_self_test, + check_signatures, + plan_regression, + ) +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.qa_matrix import ( + check_agent_six, + check_consistency, + check_cross_client_matrix, + check_g5, + check_prd_coverage, + check_qa_fix_role, + check_reaudit_reran_links, + check_self_test, + check_signatures, + plan_regression, + ) + + +class SelfTestTestCase(unittest.TestCase): + def test_three_parts_required_including_smoke_steps(self): + complete, missing = check_self_test({ + "unit": True, "contract_integration": True, + "manual_smoke": {"startup": True, "login": True, + "core_agent_task": True, + "error_prompt": True}}) + self.assertTrue(complete) + self.assertEqual(missing, []) + _complete, missing2 = check_self_test({ + "unit": True, "contract_integration": True, + "manual_smoke": {"startup": True, "login": True, + "core_agent_task": True}}) + self.assertIn("manual_smoke:error_prompt", missing2) + _c3, missing3 = check_self_test({"unit": True}) + self.assertIn("contract_integration", missing3) + + +class PrdCoverageTestCase(unittest.TestCase): + def test_every_criterion_needs_a_valid_case(self): + result = check_prd_coverage( + ["prd-1", "prd-2"], + [{"id": "case-1", "covers": ["prd-1"], + "precondition": "logged in", "steps": "run task", + "expected": "result returned"}]) + self.assertEqual(result["missing_criteria"], ["prd-2"]) + self.assertEqual(result["invalid_cases"], []) + + def test_case_without_expected_result_does_not_cover(self): + result = check_prd_coverage( + ["prd-1"], + [{"id": "case-1", "covers": ["prd-1"], + "precondition": "logged in", "steps": "run task", + "expected": ""}]) + self.assertEqual(result["missing_criteria"], ["prd-1"]) + self.assertIn("case-1", result["invalid_cases"]) + + +class AgentSixTestCase(unittest.TestCase): + def test_all_six_required(self): + self.assertEqual(check_agent_six({ + "sandbox_escape": True, + "dangerous_action_approval": True, + "long_task_timeout_cancel": True, + "offline_resume": True, + "model_503_retry": True, + "token_quota_exceeded": True}), []) + missing = check_agent_six({"sandbox_escape": True}) + self.assertEqual(len(missing), 5) + self.assertIn("token_quota_exceeded", missing) + + +class MatrixTestCase(unittest.TestCase): + def test_matrix_covers_every_selected_client(self): + result = check_cross_client_matrix( + ["web", "ios"], + [{"client": "web", "verdict": "pass"}, + {"client": "ios", + "verdict": "code_review_plus_web_core"}]) + self.assertTrue(result["accounted"]) + self.assertTrue(result["qa_pass"]) + self.assertEqual(result["missing_clients"], []) + + def test_missing_client_is_listed(self): + result = check_cross_client_matrix( + ["web", "ios"], + [{"client": "web", "verdict": "pass"}]) + self.assertEqual(result["missing_clients"], ["ios"]) + self.assertFalse(result["accounted"]) + + def test_untested_with_reason_is_accounted_but_not_pass(self): + result = check_cross_client_matrix( + ["web"], + [{"client": "web", "verdict": "untested", + "untested_reason": "device lab unavailable"}]) + self.assertTrue(result["accounted"]) + self.assertFalse(result["qa_pass"]) + + def test_untested_without_reason_is_neither(self): + result = check_cross_client_matrix( + ["web"], [{"client": "web", "verdict": "untested"}]) + self.assertFalse(result["accounted"]) + self.assertFalse(result["qa_pass"]) + self.assertTrue(result["violations"]) + + def test_uncompiled_client_cannot_claim_build_pass(self): + for client in ("ios", "android", "macos"): + with self.subTest(client=client): + result = check_cross_client_matrix( + [client], [{"client": client, "verdict": "pass"}]) + self.assertFalse(result["qa_pass"]) + self.assertTrue(result["violations"]) + + +class ConsistencyTestCase(unittest.TestCase): + def test_matching_rows_are_consistent(self): + result = check_consistency([ + {"client": "web", "result": "ok", "status_word": "done", + "sync_state": "synced"}, + {"client": "ios", "result": "ok", "status_word": "done", + "sync_state": "synced"}]) + self.assertTrue(result["consistent_or_explained"]) + + def test_intentional_difference_needs_a_basis(self): + result = check_consistency([ + {"client": "web", "result": "ok", "status_word": "done", + "sync_state": "synced"}, + {"client": "ios", "result": "ok", "status_word": "完成", + "sync_state": "synced", "classification": "intentional", + "basis": ""}]) + self.assertFalse(result["consistent_or_explained"]) + ok = check_consistency([ + {"client": "web", "result": "ok", "status_word": "done", + "sync_state": "synced"}, + {"client": "ios", "result": "ok", "status_word": "完成", + "sync_state": "synced", "classification": "intentional", + "basis": "iOS uses localised status words per design §4"}]) + self.assertTrue(ok["consistent_or_explained"]) + defect = check_consistency([ + {"client": "web", "result": "ok", "status_word": "done", + "sync_state": "synced"}, + {"client": "ios", "result": "error", "status_word": "done", + "sync_state": "synced", "classification": "defect"}]) + self.assertFalse(defect["consistent_or_explained"]) + + +class G5AndRegressionTestCase(unittest.TestCase): + def test_g5_requires_all_gates_and_no_severe_unfixed(self): + gates = {"unit": True, "integration": True, + "cross_client_consistency": True, + "agent_security": True, "long_task": True, + "performance_and_security": True} + self.assertTrue(check_g5({"gates": gates, "failures": []})["g5_pass"]) + missing = check_g5({"gates": dict(gates, long_task=False), + "failures": []}) + self.assertIn("long_task", missing["missing_gates"]) + blocked = check_g5({ + "gates": gates, + "failures": [{"severity": "severe", "fixed": False, + "owner": "web", "reproduction": "run X"}]}) + self.assertFalse(blocked["g5_pass"]) + self.assertEqual(len(blocked["blocking_defects"]), 1) + + def test_regression_follows_impact_surface_not_full_retest(self): + cases = [{"id": "c-a", "covers": ["contract-A"]}, + {"id": "c-b", "covers": ["contract-B"]}] + plan = plan_regression(["contract-A"], cases) + self.assertEqual(plan["regression_cases"], ["c-a"]) + self.assertFalse(plan["is_full_retest"]) + self.assertEqual(plan["uncovered_areas"], []) + uncovered = plan_regression(["contract-C"], cases) + self.assertEqual(uncovered["uncovered_areas"], ["contract-C"]) + + def test_qa_must_not_fix_code_directly(self): + self.assertTrue(check_qa_fix_role( + [{"defect": "d-1", "fixed_by": "qa"}])) + self.assertEqual(check_qa_fix_role( + [{"defect": "d-1", "fixed_by": "client-web"}]), []) + + +class SignaturesAndReauditTestCase(unittest.TestCase): + def test_dual_signature_seats_both_required_and_distinct(self): + self.assertEqual(check_signatures("qa-q", "auditor-a"), []) + self.assertTrue(check_signatures("qa-q", None)) + self.assertTrue(check_signatures(None, "auditor-a")) + self.assertTrue(check_signatures("same", "same")) + + def test_reaudit_must_rerun_failed_links(self): + package = {"reaudit_must_rerun_links": ["link-create-task", + "link-cancel-task"]} + ok = check_reaudit_reran_links( + package, {"link-create-task": True, "link-cancel-task": True}) + self.assertTrue(ok["reran_all_failed_links"]) + missing = check_reaudit_reran_links( + package, {"link-create-task": True}) + self.assertFalse(missing["reran_all_failed_links"]) + self.assertIn("link-cancel-task", missing["missing"]) + failed = check_reaudit_reran_links( + package, {"link-create-task": True, + "link-cancel-task": False}) + self.assertIn("link-cancel-task", failed["failed"]) + + +class QaSkillTextTestCase(unittest.TestCase): + def test_qa_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 ( + "QA 全链路", + "agent 专属六项", + "跨端矩阵", + "代码评审 + Web 核心验证", + ): + with self.subTest(phrase=phrase): + self.assertIn(phrase, text) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_skill_appendix_provenance.py b/cloud/tests/test_skill_appendix_provenance.py index 5048252..616cafc 100644 --- a/cloud/tests/test_skill_appendix_provenance.py +++ b/cloud/tests/test_skill_appendix_provenance.py @@ -8,13 +8,13 @@ script. The appendix was therefore unverifiable -- a drifted chapter would pass CI silently. This file pins the repair, and nothing else: -- tools/check_appendix.py passes on the real pack (all eleven +- tools/check_appendix.py passes on the real pack (all twelve chapters match their templates/roles/ sources, CHANGELOG present and mentioning templates/roles/); - the same script FAILS (nonzero exit) on a temp copy with a source deleted, a source tampered, or an extra source added -- so a future "source missing" or "appendix drifted" state cannot slip through; -- --emit output is exactly the eleven headings plus the eleven +- --emit output is exactly the eleven headings plus the twelve source bodies, so the spliced form is itself derived from the sources. @@ -40,7 +40,8 @@ SCRIPT = SKILL_DIR / "tools" / "check_appendix.py" EXPECTED_IDS = ["00-coordinator", "01-product-manager", "02-architect", "03-designer", "04-client-developer", "05-backend", "06-qa", "07-acceptance", "08-shared-core", - "09-ops", "10-iteration"] + "09-ops", "10-iteration", + "11-auditor"] def _run(skill_dir=None): @@ -112,7 +113,7 @@ class AppendixProvenanceTestCase(unittest.TestCase): / (cid + ".md")).read_text(encoding="utf-8") expected_parts.append(heading + "\n\n" + source.rstrip("\n")) self.assertEqual(result.stdout, "\n\n".join(expected_parts) + "\n", - "--emit 输出须恰为 11 章标题 + 11 份源正文") + "--emit 输出须恰为 12 章标题 + 12 份源正文") if __name__ == "__main__": diff --git a/cloud/tests/test_skill_appendix_sync.py b/cloud/tests/test_skill_appendix_sync.py index 1189e55..a2d32a7 100644 --- a/cloud/tests/test_skill_appendix_sync.py +++ b/cloud/tests/test_skill_appendix_sync.py @@ -57,7 +57,8 @@ SKILL_PATH = (REPO_ROOT / "cloud" / "skills" / "software-engineering" EXPECTED_IDS = ["00-coordinator", "01-product-manager", "02-architect", "03-designer", "04-client-developer", "05-backend", "06-qa", "07-acceptance", "08-shared-core", - "09-ops", "10-iteration"] + "09-ops", "10-iteration", + "11-auditor"] # Chapter -> keywords that chapter must carry for the GitHub guarantee # the body claims is "同样写进了附录的协调员、客户端、验收章节".