audit: independent audit role, auto-pushback, rebuild branch, three tightenings, QA full-chain

Independent audit (no code, independent re-run, separate signature;
CONDITIONAL never passes; deviation needs due todo) and automatic
pushback (fail -> derived fix subtask, merge/release blocked,
package with per-item evidence, full re-audit, 3-round limit with
a separate CI auto-fix counter, ledger hash-chain interface for
track 4, gate-block interface for track 1). Over 3 failures per
module per contract version: rebuild only that module from the
frozen contract (old implementation archived, avoidance list),
contract/design root cause -> G2 re-freeze escalation instead.
Tightenings: 3rd-failure triage (structural early rebuild /
early refreeze, near-complete one targeted round), rebuild keeps
tests and passed assets with an old-test first gate, audit-of-
audit sampling/correction and G6 user-only acceptance plus
acceptance-rebuild user confirmation. QA full chain: self-test
three parts, PRD coverage, agent six, cross-client matrix with
honest uncompiled verdicts, G5 gates, impact-scoped regression,
dual QA/audit signatures, failed-link re-runs in re-audit.
Skill text is purely additive (+459/-0) against main v0.2 text;
port this section into v0.3/v0.4/v0.5 texts on merge, never
overwrite them with this file.
This commit is contained in:
host committed 2026-11-03 02:49:25 +00:00
1 parent 41e8c98512
commit e921db3472
10 files changed
+3358 -10

No files matched your search

File diff suppressed because it is too large. Load diff
+333
View File
@@ -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}
@@ -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 处引用原样保留。
+352
View File
@@ -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 验收,不替协调员给合并决定。
@@ -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 验收,不替协调员给合并决定。
@@ -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/<id>.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-]+)"
+883
View File
@@ -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()
+247
View File
@@ -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()
@@ -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__":
+2 -1
View File
@@ -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 "同样写进了附录的协调员、客户端、验收章节".