diff --git a/.harness/engine.lock b/.harness/engine.lock index 104e996..9405fe1 100644 --- a/.harness/engine.lock +++ b/.harness/engine.lock @@ -1,6 +1,6 @@ { "engine": "delivery-harness", "version": "0.1.0", - "commit": "a1f73548fd6b61dcac7fe670233b152c3cd1c88a", - "tree": "sha256:6f39a109a94ae624c5c9cfbcc20f215bc8829069079d2f1237c768cb39eec8a6" + "commit": "0efb6c08d68915ba416c6f112d8225667cc703cc", + "tree": "sha256:516079f11a28697d0453de45cb28ba44e8163e2427610bec64b52f7af8e310c4" } diff --git a/.harness/engine/agents/review.py b/.harness/engine/agents/review.py index eb1d093..b20ea4c 100644 --- a/.harness/engine/agents/review.py +++ b/.harness/engine/agents/review.py @@ -40,6 +40,7 @@ from engine.agents import dispatch, dispatch_host, review_pack_io from engine.agents import dispatch_observation as observation +from engine.core import alerts from engine.core.common import ENGINE_DIR, ROOT, changed_files, commit_field, git, load_rules from engine.routing import run_check @@ -411,6 +412,7 @@ def review_pr(number: int, reviewer_name: str | None, root: Path = ROOT, github= designer=designer, model=verdict.audit_model, model_basis=verdict.model_basis, parsed=verdict.parsed, independent=independent, same_host=name in executors, failure=verdict.failure, error_kind=verdict.error_kind) + alerts.publish(trace, "review_error", pr=number, gh=github) # 评审方自身失败也复用发布（D067） print(f"PR #{number}：评审方失败（{verdict.failure}），没有评论，保留待评审标签") return 1 same_host = name in executors @@ -433,6 +435,8 @@ def review_pr(number: int, reviewer_name: str | None, root: Path = ROOT, github= github._run(["gh", "pr", "edit", str(number), "--remove-label", LABEL], agent=True) except RuntimeError: pass # 没有该标签 + if verdict.verdict == "不通过": # D067：评审否决在已有评论发布后补标签/告警，不重跑评审 + alerts.publish(trace, "review_rejected", pr=number, gh=github) print(f"PR #{number}：{verdict.verdict}（{name}），已评论") return 0 diff --git a/.harness/engine/cli.py b/.harness/engine/cli.py index 22cd376..8c6ac54 100644 --- a/.harness/engine/cli.py +++ b/.harness/engine/cli.py @@ -47,6 +47,10 @@ # 可观测性查询（同一模块两入口；不追加自身事件，见 QUIET_COMMANDS） "events": "engine.reports.trace:events_main", "trace": "engine.reports.trace:trace_main", + # 审计引用复原与哈希核对（不追加自身事件；观察事件属 T404） + "audit": "engine.reports.audit", + # 告警发布（B46 设计 5；不追加通用入口事件，见 QUIET_COMMANDS） + "alert": "engine.core.alerts", # 安装与升级（只能从引擎仓库的检出运行） "install": "engine.core.install", "upgrade": "engine.core.install:upgrade_main", diff --git a/.harness/engine/core/alerts.py b/.harness/engine/core/alerts.py index 451cabd..fe54a54 100644 --- a/.harness/engine/core/alerts.py +++ b/.harness/engine/core/alerts.py @@ -1,20 +1,31 @@ -"""共用告警 API（B46 C4，T205）：在 PR 评论或升级议题上发布带远端标记的告警。 +"""共用告警 API（B46 C4，T205 定义；T403 扩展 CLI 与工作流承载）：在 PR 评论或升级议题上发布带远端标记的告警。 渠道只有 PR/议题评论加 escalation 标签（设计 5，不新增通知渠道）。远端标记是 trace_id、NUL、 reason 的 sha256：同键先查分页评论/带标签议题再更新，不重发；无 PR 时查/建一条升级议题。 publish 只消费结构化既有状态，永不抛异常（C0）：发布失败返回 ok=False 并留本机 alert 观察事件， 不改调用方的判定、退出码与升级。details 只含安全阶段摘要、trace 查询方式与 PR/CI/账本链接。 +本机事件库只是记录，不是去重权威：删掉本机库后重发同键告警仍更新同一条远端评论（换机同理）。 reason 只用 C4 稳定键。派发侧两个预警由 rules.toml 的可选 [alerts] 节启用（删除整节即全部关闭）： 最后一轮 CI 在 wait 前预警（总预算 1 则首轮即预警）；守卫拒绝预警按 [alerts] guard_denials_threshold （每轮被守卫拒绝的工具调用数，一次调用多个拒绝理由算一次）触发，未配置不启用，非正或非法值禁用 并提示一次配置错误，不改执行预算。阈值/末轮条件与安全详情组装都在本模块，dispatch 只接线。 + +T403 补 CLI 与工作流承载（F3）：`alert` 子命令（cli.py 注册，QUIET_COMMANDS 不追加通用入口事件） +供评审后补之外的告警发布——CI 的 harness/auto-merge 告警 job 与本机手工调用。`--bundle` 读一个 +harness-events 事件包（C5 形状，只读不导入），包里出现该 reason 的触发证据才发布；未知失败不自动 +告警。工作流里 pull_request 触发的检查保持只读，push main 与 workflow_run 各由指定告警 job 承载 +（模板见 templates/.github/workflows/），本模块只提供发布与证据核对。 """ from __future__ import annotations +import argparse import hashlib +import json +import subprocess import sys +from pathlib import Path from engine.core import events from engine.core.common import load_rules @@ -128,8 +139,11 @@ def publish(trace_id: str, reason: str, *, pr: int | None = None, task: str | No """共用告警发布（C4）：有 PR 在 PR 上评论（带 escalation 标签），无 PR 查/建一条升级议题； 同键（trace_id+reason 的远端标记）查分页评论/议题后更新，不重发。永不抛异常：失败返回 {ok: False, target: None, updated: False, error_kind} 并留本机 alert 事件，不改调用方的 - 判定、退出码与升级。""" + 判定、退出码与升级。gh 适配器没有 C4 发布接口（list_comments 缺失，如只评论的最小桩）时 + 是空操作：发布不可能成功，不发起远端调用也不留必然失败的本机事件。""" result = {"ok": False, "target": None, "updated": False, "error_kind": None} + if gh is not None and getattr(gh, "list_comments", None) is None: + return result try: if reason not in REASONS: result["error_kind"] = "unknown_reason" @@ -170,12 +184,157 @@ def _marked(items, mark: str, key: str) -> int | None: return None -def _record(trace_id: str, reason: str, result: dict) -> None: +def _record(trace_id: str, reason: str, result: dict, status: str | None = None) -> None: """本机 alert 观察事件（C1：alert 自己记发布结果）；emit 永不抛，组装失败也只丢弃。""" try: - events.emit(stage="alert", step=reason, status="ok" if result["ok"] else "error", + events.emit(stage="alert", step=reason, status=status or ("ok" if result["ok"] else "error"), trace_id=trace_id, outputs={"target": result["target"], "updated": result["updated"]}, error={"kind": result["error_kind"]} if result["error_kind"] else None) except Exception: # noqa: BLE001 观察旁路 return + + +# ---- CLI（T403）：cli.py COMMANDS 注册 alert；无证据跳过留 skip 事件，其余经 publish ---- + +# --bundle 证据签名：(stage, step, status) 精确匹配；audit.finding 再按 outputs.rule 区分。 +# 只映射设计 5 里由 CI 事件包承载的四类；审计的链头/锚点一致性发现（chain_invalid、ledger_mismatch） +# 与环节缺失（missing_*）同属设计 5 第七行的两个告警族。其余 reason 没有包内证据形态，配 --bundle +# 按无证据跳过（不发布）。 +BUNDLE_TRIGGERS: dict[str, tuple[str, str, str, tuple[str, ...] | None]] = { + "integrity_failure": ("verify", "integrity", "fail", None), + "budget_exhausted": ("route", "预算", "fail", None), + "audit_missing_stage": ("ci", "audit.finding", "fail", ("missing_",)), + "audit_anchor_mismatch": ("ci", "audit.finding", "fail", + ("anchor_mismatch", "chain_invalid", "ledger_mismatch")), +} + + +def bundle_evidence(bundle: dict, reason: str) -> bool: + """事件包（C5 形状）里是否有该 reason 的触发证据；包形状不对或未知失败按无证据处理。""" + trigger = BUNDLE_TRIGGERS.get(reason) + if trigger is None: + return False + stage, step, status, rules = trigger + for event in bundle.get("events") or []: + if not isinstance(event, dict): + continue + if (event.get("stage"), event.get("step"), event.get("status")) != (stage, step, status): + continue + if rules is None or str((event.get("outputs") or {}).get("rule") or "").startswith(rules): + return True + return False + + +def _load_bundle(path: Path) -> dict: + """读一个事件包：目录（含 harness-events.json）或 JSON 文件；形状不对抛 TypeError。""" + file = path / "harness-events.json" if path.is_dir() else path + bundle = json.loads(file.read_text(encoding="utf-8")) + if not isinstance(bundle, dict) or not isinstance(bundle.get("events"), list): + raise TypeError(f"{file} 不是 harness-events 事件包") + return bundle + + +class GhPublisher: + """CLI/CI 侧的 C4 发布适配器：直接调用 gh（认证取 GH_TOKEN/GH_REPO 环境），不经 bin/as-agent。 + + 派发与评审在本机以 Agent 账号发布（dispatch.GitHub 走 as-agent）；工作流告警 job 持工作流自身的 + 最小写令牌，本机手工发布用操作者自己的 gh 登录。方法形状与 dispatch.GitHub 的发布接口一致， + publish 按同一接口调用两者。 + """ + + def __init__(self, cwd: Path | str | None = None): + self.cwd = str(cwd) if cwd is not None else None + + def _run(self, argv: list[str], stdin: str | None = None) -> str: + result = subprocess.run(["gh", *argv], cwd=self.cwd, input=stdin, capture_output=True, text=True, + check=False) + if result.returncode != 0: + raise RuntimeError(f"gh {' '.join(argv[:3])} 失败：{result.stderr.strip()[-500:]}") + return result.stdout.strip() + + def list_comments(self, pr: int) -> list[dict]: + raw = self._run(["api", f"repos/{{owner}}/{{repo}}/issues/{pr}/comments", "--paginate", "--slurp"]) + return [item for page in json.loads(raw or "[]") for item in page] + + def comment(self, pr: int, body: str, label: str | None = None) -> str: + url = self._run(["pr", "comment", str(pr), "--body-file", "-"], stdin=body) + if label: + self.add_label(pr, label) + return url + + def add_label(self, pr: int, label: str) -> None: + try: # 不存在才创建；已有标签不被改写（与 dispatch.GitHub 同口径） + self._run(["label", "create", label, "--color", "d93f0b"]) + except RuntimeError: + pass + self._run(["pr", "edit", str(pr), "--add-label", label]) + + def edit_comment(self, comment_id: int, body: str) -> None: + self._run(["api", "-X", "PATCH", f"repos/{{owner}}/{{repo}}/issues/comments/{comment_id}", + "--input", "-"], stdin=json.dumps({"body": body})) + + def list_issues(self, label: str) -> list[dict]: + raw = self._run(["issue", "list", "--state", "all", "--label", label, "--limit", "200", + "--json", "number,body"]) + return json.loads(raw or "[]") + + def edit_issue(self, number: int, body: str) -> None: + self._run(["issue", "edit", str(number), "--body-file", "-"], stdin=body) + + def create_issue(self, title: str, body: str, labels: list[str]) -> str: + argv = ["issue", "create", "--title", title, "--body-file", "-"] + for label in labels: + argv += ["--label", label] + return self._run(argv, stdin=body) + + +def main(argv: list[str] | None = None, gh=None) -> int: + """alert 子命令（设计 5）：发布一条告警；退出码只区分参数错误（2），发布失败退出 0。 + + 发布是观察旁路：gh 失败返回 ok=False 并留本机 alert 事件，不改调用方（工作流步骤/人工 shell） + 的结论；--pr 缺省查/建一条 escalation 升级议题；--trace 缺省取当前追踪 ID（CI 为 PR 分支， + main 上 main@<短提交>）；--bundle 给出 harness-events 事件包（目录或 JSON）时先核对触发证据， + 无证据不发布（留 skip 事件）；未知 reason 或事件包不可用属参数错误。gh 参数供测试注入。 + """ + parser = argparse.ArgumentParser(prog="alert", description="发布告警（B46 设计 5：PR/议题评论 + escalation 标签）") + parser.add_argument("reason", help="C4 稳定 reason 键：" + "、".join(sorted(REASONS))) + parser.add_argument("--pr", type=int, help="告警落在这个 PR；缺省查/建一条 escalation 升级议题") + parser.add_argument("--trace", default="", help="追踪 ID（缺省当前分支；main 上 main@<短提交>）") + parser.add_argument("--task", default=None, help="任务编号（标题用；缺省用 trace）") + parser.add_argument("--bundle", type=Path, default=None, + help="harness-events 事件包（目录或 JSON）：先核对触发证据，无证据不发布") + parser.add_argument("--json", action="store_true", help="打印结果 JSON") + args = parser.parse_args(argv) + if args.reason not in REASONS: + print(f"alert：未知 reason {args.reason!r}（可用：{'、'.join(sorted(REASONS))}）", file=sys.stderr) + return 2 + trace = args.trace or events.current_trace() + if args.bundle is not None: + try: + bundle = _load_bundle(args.bundle) + except (OSError, ValueError, TypeError) as error: + print(f"alert：事件包不可用（{error}）", file=sys.stderr) + return 2 + if not bundle_evidence(bundle, args.reason): + _record(trace, args.reason, {"ok": False, "target": None, "updated": False, + "error_kind": None}, status="skip") + _print_result({"ok": False, "target": None, "updated": False, "error_kind": "no_evidence"}, + args.json) + return 0 + result = publish(trace, args.reason, pr=args.pr, task=args.task, gh=gh if gh is not None else GhPublisher()) + _print_result(result, args.json) + return 0 + + +def _print_result(result: dict, as_json: bool) -> None: + if as_json: + print(json.dumps(result, ensure_ascii=False)) + return + if result["ok"]: + state = "已更新" if result["updated"] else "已发布" + elif result["error_kind"] == "no_evidence": + state = "包内无触发证据，未发布" + else: + state = f"发布失败（{result['error_kind']}）" + print(f"alert：{state}（target={result['target']}）") diff --git a/.harness/engine/reports/audit.py b/.harness/engine/reports/audit.py new file mode 100644 index 0000000..ca52838 --- /dev/null +++ b/.harness/engine/reports/audit.py @@ -0,0 +1,762 @@ +"""审计引用复原与哈希核对（B46 T401，设计 4.2，共用合同 C7）。 + + python3 cli.py audit [--json] + python3 cli.py audit --all-merged [--since 30d] [--json] + +复原一个（或 --all-merged 批量分页确定性时间窗口内的全部）已合并 PR 的安全环节视图：按 API 的 +trace/head、C6 合并账本与本机/CI 已导入事件列出各阶段的输入、输出与决定，来源分别标注 +local/ci/github。下载 CI 由 C5/T302 共享的 load_ci 完成，GitHub 事实更新由 T304 共享的 sync 完成， +本命令不写合并路由；cli.py QUIET_COMMANDS 不记 cli.audit 入口事件，每次运行结束由本模块写一条 +ci/audit.summary 与每条 finding 一条 ci/audit.finding 观察事件（T404，C7 供周报，失败隔离见 +_emit_observations）。 + +逐引用核对四类：git 文件按 path@rev 用 git show 取原始字节（结尾换行原样参与哈希，不去尾、不再 +规范化）；GitHub 评论按不可变正文原字节、可变 PR 字段按评审材料同一渲染规则复取（变动报 +snapshot_changed，不与固定绑定提交的事实混比）；本机产物按 sha256/size 并区分保留期到期；CI 事件 +包的规范化哈希核对由 load_ci 导入完成，发现按来源映射进报告。只访问当前仓库受支持的 GitHub 资源 +（评论、PR 字段、Actions 包），引用内容只解析与比对、绝不执行；path 不能逃出仓库。失败区分 +hash_mismatch / reference_unavailable / reference_expired，未核对在 coverage 里如实计数，不冒充通过。 +完整性规则、锚点防篡改与 [audit] 配置由 audit_completeness 在复原核对之后判定（B46 T402）。 + +inspect_pr(pr, *, gh=None, cwd=ROOT) 返回 C7 基础形状：pr/trace_id/head_sha/stages/references/ +findings/coverage/ok；finding 有 rule/severity/source/stage/ref/reason，reason 不回显私密内容。 +退出码：0 全部适用引用核对通过；1 有发现（哈希不符、引用不可得、到期或完整性/防篡改发现）； +2 参数错误、整体 API 故障、[audit] 配置错误或 PR 无法审计。 +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import sqlite3 +import subprocess +import sys +from collections import Counter +from contextlib import closing +from datetime import UTC, datetime, timedelta +from pathlib import Path + +from engine.core import events, events_db, events_io +from engine.core.common import ROOT, clean_git_env +from engine.reports import audit_completeness, github_events, ledger + +# 批量模式 --since 的相对时长单位（折算为秒）。 +_DURATION_RE = re.compile(r"(\d+)([dhms])\Z") +_UNITS = {"d": 86400, "h": 3600, "m": 60, "s": 1} + +# C7 稳定 rule：T401 产生引用核对四类；完整性/锚点/账本一致性等其余稳定 rule 由 T402 扩展。 +_RULE_FOR_STATUS = { + "hash_mismatch": "hash_mismatch", + "snapshot_changed": "snapshot_changed", + "unavailable": "reference_unavailable", + "expired": "reference_expired", +} +_STATUSES = ("verified", "hash_mismatch", "snapshot_changed", "unavailable", "expired", "unchecked") +_SOURCE_CLASSES = ("local", "ci", "github") +_HASH_RE = re.compile(r"[0-9a-f]{64}\Z") +_SHA_RE = re.compile(r"[0-9a-f]{40}\Z") +_GIT_REF_RE = re.compile(r"(?P.+)@(?P[0-9a-f]{7,40})\Z") +# 只接受 owner/repo#PR/comments/ID 形状的评论引用；其余（任意 URL、其他仓库）一律拒绝，不发起请求。 +_COMMENT_REF_RE = re.compile(r"(?P[A-Za-z0-9_.-]+)/(?P[A-Za-z0-9_.-]+)" + r"#(?P\d+)/comments/(?P\d+)\Z") +# PR 锚点评论（T305 ledger._anchor_body 的固定行）里记录的账本路径、提交与文件 sha256。 +_ANCHOR_FILE_RE = re.compile(r"- 文件：`(?P[^`]+)` @ (?P[0-9a-f]{40})") +_ANCHOR_SHA_RE = re.compile(r"- 文件 sha256：`(?P[0-9a-f]{64})`") + + +class AuditError(RuntimeError): + """PR 无法审计（整体 API 故障、PR 不存在或未合并）；CLI 转退出码 2。""" + + +def _parse_window(text: str) -> datetime: + """批量模式 --since 的窗口起点： 相对时长或 ISO 时间戳，坏值抛 ValueError。 + + 相对时长按审计自身的时间基准（events_db._now）折算而不取墙钟：批量范围与事件/锚点同一时钟， + 因此是确定性时间范围（同一时钟下两次运行覆盖同一集合）。 + """ + if match := _DURATION_RE.fullmatch(text.strip()): + now = _parse_utc(events_db._now()) + if now is None: + raise ValueError("--since 无法确定审计时间基准") + return now - timedelta(seconds=int(match[1]) * _UNITS[match[2]]) + parsed = _parse_utc(text) + if parsed is None: + raise ValueError(f"--since 无法解析为时长或 UTC 时间：{text}") + return parsed + + +# 缺省 gh 客户端：repo/pr/api/download 全都具备（load_ci 与 sync 只用这些）；测试可注入替换。 +GhClient = ledger.GhClient + + +def _finding(rule: str, *, source: str, stage: str, ref: str, reason: str, + severity: str = "error") -> dict: + """C7 finding：rule/severity/source/stage/ref/reason；reason 只写分类，不回显引用内容。""" + return {"rule": rule, "severity": severity, "source": source, "stage": stage, + "ref": ref, "reason": reason} + + +def _digest(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _parse_utc(value) -> datetime | None: + """UTC ISO 时间戳 → aware UTC；缺失或损坏返回 None（调用方按无法判定处理）。""" + if not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value.strip()) + except ValueError: + return None + return parsed.replace(tzinfo=UTC) if parsed.tzinfo is None else parsed.astimezone(UTC) + + +def _source_class(source: str) -> str: + """来源显示分类（C2/C7）：github:/ci 前缀各成一类，其余（含 local）归本机。""" + if source.startswith("github:"): + return "github" + if source.startswith("ci"): + return "ci" + return "local" + + +def _safe_repo_path(path: str) -> bool: + """仓库相对路径合法性：拒绝绝对路径、盘符、反斜杠、~ 与 . / .. 段（path 不能逃出仓库）。""" + if not path or path.startswith(("/", "~")) or ":" in path or "\\" in path or "\0" in path: + return False + return not any(part in ("", ".", "..") for part in path.split("/")) + + +def _git_run(cwd: Path, *args: str) -> tuple[bool, str]: + """行输出 git 调用（ls-remote/fetch/ls-tree）：返回（成功, stdout 去尾换行）。""" + done = subprocess.run(["git", *args], cwd=cwd, capture_output=True, env=clean_git_env(), check=False) + return done.returncode == 0, done.stdout.decode("utf-8", "replace").rstrip("\n") + + +def _git_bytes(cwd: Path, rev: str, path: str) -> bytes | None: + """git show : 的原始字节（结尾换行原样保留，不做任何规范化）；失败返回 None。""" + done = subprocess.run(["git", "show", f"{rev}:{path}"], cwd=cwd, capture_output=True, + env=clean_git_env(), check=False) + return done.stdout if done.returncode == 0 else None + + +def _artifact_created(digest: str) -> str | None: + """本机产物索引里的 created（只读查询；行不存在或读取失败返回 None）。""" + path = events_db.db_path() + if path is None or not path.exists(): + return None + try: + with closing(sqlite3.connect(path, timeout=events_db.BUSY_TIMEOUT_MS / 1000)) as conn: + row = conn.execute("SELECT created FROM artifacts WHERE sha256=?", (digest,)).fetchone() + except sqlite3.Error: + return None + return row[0] if row else None + + +class _Auditor: + """一个已合并 PR 的复原与逐引用核对：收集 → 合并账本期望 → 逐条核对 → 汇总报告。""" + + def __init__(self, *, client, repo: str, pr: int, trace: str, head: str, merge_sha: str, + merged_at: str, cwd: Path): + self.client, self.repo, self.pr = client, repo, pr + self.trace, self.head, self.merge_sha, self.merged_at = trace, head, merge_sha, merged_at + self.cwd = Path(cwd) + now = _parse_utc(events_db._now()) or datetime.now(UTC) + # 窗口与到期判定的时间基准一次取定（测试冻结时钟）；到期阈值沿用实际生效的产物保留期。 + self.cutoff = now - timedelta(days=events.artifact_days()) + self.findings: list[dict] = [] + self.refs: list[dict] = [] + self.stages: list[dict] = [] + self.anchors: list[dict] = [] + self.chains: list[dict] = [] + self.event_rows: list[dict] = [] + self.ledger_doc = None # 解析后的账本（T402 账本一致性核对用；None=无账本/不可解析） + self.db_error = None # 事件库不可读标记（T402：坏库如实报告，不当作无事件） + self.ledger_state = "absent" + self._expect: dict[tuple, dict] = {} # (kind, ref) → 账本记录的期望 sha256/size + + # ---- 收集 ---- + + def _add(self, *, check: str, kind: str, ref, sha256, size, source: str, stage: str, + expectation: str, event_ts: str | None = None, artifact: str | None = None, + note: str | None = None, ledger_bytes: bytes | None = None) -> dict: + entry = {"check": check, "kind": kind, "ref": ref, "sha256": sha256, "size": size, + "source": source, "stage": stage, "expectation": expectation, + "event_ts": event_ts, "artifact": artifact, "note": note, + "ledger_bytes": ledger_bytes, "status": "unchecked", "reason": note, + "observed_sha256": None, "observed_size": None} + self.refs.append(entry) + return entry + + def _collect_events(self): + """本机/CI/GitHub 已导入事件 → 阶段视图、链头与锚点；local 事件的引用逐条登记核对。""" + try: + rows = events_io.query(trace_id=self.trace) + anchors = events_db.read_anchors(trace_id=self.trace) + except Exception: # noqa: BLE001 坏库：如实交完整性层报告（T402），不崩溃、不当作无事件 + self.db_error = True + rows, anchors = [], [] + self.event_rows = sorted(rows, + key=lambda row: (row["ts"], row["source"], row["seq"])) + for row in self.event_rows: + self.stages.append({"evidence": "event", "source": row["source"], + "source_class": _source_class(row["source"]), "stage": row["stage"], + "step": row["step"], "status": row["status"], "ts": row["ts"], + "duration_ms": row["duration_ms"], "seq": row["seq"], "hash": row["hash"], + "inputs": row["inputs"], "outputs": row["outputs"], + "decision": row["decision"], "error": row["error"], "actor": row["actor"]}) + heads: dict[str, tuple[int, str]] = {} + for row in self.event_rows: + if row["source"] not in heads or row["seq"] > heads[row["source"]][0]: + heads[row["source"]] = (row["seq"], row["hash"]) + self.chains = [{"source": source, "trace_id": self.trace, "head_hash": heads[source][1]} + for source in sorted(heads)] + self.anchors.extend(anchors) + for row in self.event_rows: + if _source_class(row["source"]) != "local": + continue # CI 事件的产物在 CI 侧（C5 导入不恢复内容），到期由 load_ci 发现报告 + self._collect_input_refs(row) + self._collect_output_refs(row) + + def _collect_input_refs(self, row: dict): + """local 事件输入引用：artifact 引用与 path@rev 形状的 git 文件引用（须带记录的 sha256）。""" + for item in row["inputs"]: + if not isinstance(item, dict) or not isinstance(item.get("sha256"), str): + continue + if item.get("kind") == "artifact": + ref_value = item.get("ref") + self._add(check="artifact", kind="artifact", ref=ref_value or item["sha256"], + sha256=item["sha256"], size=item.get("size"), source="local", + stage=row["stage"], expectation="event", event_ts=row["ts"], + artifact=ref_value if isinstance(ref_value, str) + and _HASH_RE.fullmatch(ref_value) is not None else item["sha256"]) + elif _GIT_REF_RE.fullmatch(str(item.get("ref") or "")) is not None: + self._add(check="git", kind=str(item.get("kind") or "git"), ref=item["ref"], + sha256=item["sha256"], size=item.get("size"), source="local", + stage=row["stage"], expectation="event") + + def _collect_output_refs(self, row: dict): + """输出引用从扁平 outputs 三元字段（.sha256/.size/.ref）派生（C1），核对本机产物。""" + outputs = row["outputs"] if isinstance(row["outputs"], dict) else {} + prefixes = sorted({key.rsplit(".", 1)[0] for key in outputs if key.endswith(".sha256")}) + for prefix in prefixes: + recorded = outputs.get(f"{prefix}.sha256") + if not isinstance(recorded, str) or _HASH_RE.fullmatch(recorded) is None: + continue + ref_value = outputs.get(f"{prefix}.ref") + size_value = outputs.get(f"{prefix}.size") + artifact = ref_value if isinstance(ref_value, str) and _HASH_RE.fullmatch(ref_value) else recorded + self._add(check="artifact", kind="artifact", + ref=ref_value if isinstance(ref_value, str) else artifact, sha256=recorded, + size=size_value if isinstance(size_value, int) and not isinstance(size_value, bool) + else None, + source="local", stage=row["stage"], expectation="event", event_ts=row["ts"], + artifact=artifact) + + def _collect_comments(self): + """PR 评论：评审审计摘要（C6 标记）→ 评审阶段视图与材料核对；锚点评论 → 账本期望。""" + problems: list[dict] = [] + comments = github_events._collect_pages( + self.client, f"repos/{self.repo}/issues/{self.pr}/comments", "PR 评论", problems) + if comments is None: + self.findings.append(_finding( + "reference_unavailable", source="github", stage="review", + ref=f"{self.repo}#{self.pr}/comments", + reason=f"PR 评论列表无法读取（{problems[0]['code'] if problems else 'api'}）")) + return None + anchor = None + marker = f"" + for comment in comments: + body = comment.get("body") if isinstance(comment, dict) else None + if not isinstance(body, str): + continue + if marker in body: + file_match, sha_match = _ANCHOR_FILE_RE.search(body), _ANCHOR_SHA_RE.search(body) + if file_match and sha_match and anchor is None: + anchor = {"path": file_match["path"], "commit": file_match["commit"], + "sha256": sha_match["sha256"], "comment_id": comment.get("id")} + continue + parsed, _problem = ledger._parse_review_audit(body, self.head) + if parsed is None: + continue # 解析失败或非本 head 的评审摘要不作为证据（缺评审的完整性判断属 T402） + self.stages.append({ + "evidence": "review_comment_summary", "source": "github", "source_class": "github", + "stage": "review", "step": "review", "ts": comment.get("created_at"), "trace_id": parsed.get("trace_id"), + "head": parsed.get("head"), "reviewer": parsed.get("reviewer"), + "model": parsed.get("model"), "designer": parsed.get("designer"), + "implementers": parsed.get("implementers"), "independent": parsed.get("independent"), + "verdict": parsed.get("verdict"), "duration_ms": parsed.get("duration_ms"), + "severity_counts": parsed.get("severity_counts"), + "materials": parsed.get("materials"), + "comment": {"id": comment.get("id"), "url": comment.get("html_url")}}) + self._add(check="comment", kind="review_comment", + ref=f"{self.repo}#{self.pr}/comments/{comment.get('id')}", + sha256=None, size=len(body.encode("utf-8")), source="github", stage="review", + expectation=None) + self._collect_materials(parsed) + return anchor + + def _collect_materials(self, parsed: dict): + """评审审计材料：task@head 与 pack 产物按原字节核对；pr 材料按同一渲染规则复取比对； + diff/ci 是评审时按 recipe 渲染的字节（C6：不按将来复取对象重算），登记为不核对。""" + for material in parsed.get("materials") or []: + if not isinstance(material, dict): + continue + kind, ref = material.get("kind"), material.get("ref") + sha256, size = material.get("sha256"), material.get("size") + recipe = material.get("recipe") if isinstance(material.get("recipe"), dict) else {} + if kind == "task" and ref == "none": + continue # 无任务书的「无」哨兵：没有内容可核对 + if kind == "task": + self._add(check="git", kind="material_task", ref=ref, sha256=sha256, size=size, + source="local", stage="review", expectation="review_material") + elif kind == "pack" and isinstance(sha256, str): + digest = recipe.get("artifact_sha256") if isinstance(recipe.get("artifact_sha256"), str) else sha256 + self._add(check="artifact", kind="material_pack", ref=ref if isinstance(ref, str) else digest, + sha256=sha256, size=size, source="local", stage="review", + expectation="review_material", artifact=digest) + elif kind == "pr" and ref == f"pull/{self.pr}" and isinstance(sha256, str): + self._add(check="snapshot", kind="material_pr", ref=ref, sha256=sha256, size=size, + source="github", stage="review", expectation="review_material") + else: + self._add(check="skip", kind=f"material_{kind}", ref=ref, sha256=sha256, size=size, + source="github", stage="review", expectation="review_material", + note="评审时按 recipe 渲染，复取不保证同字节，不核对") + + def _collect_run_records(self): + """合并 head 上的运行记录（路径@merge_sha 原始字节）：摘要入阶段视图，记录本身登记核对。""" + ok, listing = _git_run(self.cwd, "ls-tree", "-r", "--name-only", self.merge_sha, "--", "docs/runs") + if not ok: + return + for path in [line.strip() for line in listing.splitlines() if line.strip().endswith(".json")]: + data = _git_bytes(self.cwd, self.merge_sha, path) + if data is None: + self.findings.append(_finding( + "reference_unavailable", source="local", stage="dispatch", + ref=f"{path}@{self.merge_sha}", reason="运行记录无法从合并 head 读取")) + continue + try: + record = json.loads(data) + except ValueError: + continue + if not isinstance(record, dict) or self.trace not in (record.get("trace_id"), record.get("branch")): + continue + for item in record.get("stages") or []: + if isinstance(item, dict): + self.stages.append({"evidence": "run_record_summary", "source": "local", + "source_class": "local", "record": f"{path}@{self.merge_sha}", **item}) + for anchor in record.get("anchors") or []: + if isinstance(anchor, dict): + self.anchors.append({"trace_id": self.trace, **anchor}) + entry = self._add(check="git", kind="run_record", ref=f"{path}@{self.merge_sha}", + sha256=None, size=None, source="local", stage="dispatch", expectation=None) + entry["observed_sha256"], entry["observed_size"] = _digest(data), len(data) + + def _collect_ledger(self, anchor): + """C6 账本：锚点评论记录账本字节 sha256；账本 references 是运行记录/评审评论的核对期望。""" + if anchor is None: + return + ref = f"{ledger.AUDIT_BRANCH}:{anchor['path']}@{anchor['commit']}" + ok, _ = _git_run(self.cwd, "ls-remote", "origin", f"refs/heads/{ledger.AUDIT_BRANCH}") + fetched = False + data = None + if ok: + ok, _ = _git_run(self.cwd, "fetch", "--quiet", "origin", ledger.AUDIT_BRANCH) + if ok: + data = _git_bytes(self.cwd, anchor["commit"], anchor["path"]) + fetched = data is not None + entry = self._add(check="ledger", kind="ledger", ref=ref, sha256=anchor["sha256"], size=None, + source="github", stage="merge", expectation="anchor_comment", ledger_bytes=data) + if not fetched: + entry["status"] = "unavailable" + entry["reason"] = "账本文件无法从 harness-audit 分支读取" + return + try: + parsed = json.loads(data) + except ValueError: + parsed = None + self.ledger_doc = parsed if isinstance(parsed, dict) else None + if not isinstance(parsed, dict): + entry["status"] = "hash_mismatch" + entry["reason"] = "账本文件不是 JSON 对象" + entry["observed_sha256"], entry["observed_size"] = _digest(data), len(data) + return + entry["observed_sha256"], entry["observed_size"] = _digest(data), len(data) + for item in parsed.get("references") or []: + if not isinstance(item, dict) or not isinstance(item.get("ref"), str) \ + or not isinstance(item.get("sha256"), str): + continue + self._expect[(item.get("kind"), item["ref"])] = {"sha256": item["sha256"], + "size": item.get("size")} + + def _apply_expectations(self): + """账本记录的期望补到已收集的引用上；账本里有、视图里没收集到的引用按账本记录补收核对。""" + for entry in self.refs: + hit = self._expect.pop((entry["kind"], entry["ref"]), None) + if hit and entry["sha256"] is None: + entry["sha256"] = hit["sha256"] + entry["size"] = entry["size"] if entry["size"] is not None else hit.get("size") + entry["expectation"] = "ledger" + for (kind, ref), hit in self._expect.items(): + source = "github" if kind == "review_comment" else "local" + check = "comment" if kind == "review_comment" else "git" + self._add(check=check, kind=str(kind), ref=ref, sha256=hit["sha256"], size=hit.get("size"), + source=source, stage="review" if kind == "review_comment" else "dispatch", + expectation="ledger") + + # ---- 核对 ---- + + def _compare(self, entry: dict, data: bytes) -> None: + entry["observed_sha256"], entry["observed_size"] = _digest(data), len(data) + if entry["sha256"] is not None and entry["observed_sha256"] != entry["sha256"]: + entry.update(status="hash_mismatch", reason="内容 sha256 与记录不符（原始字节比对）") + elif entry["size"] is not None and entry["observed_size"] != entry["size"]: + entry.update(status="hash_mismatch", reason="内容大小与记录不符") + else: + entry.update(status="verified", reason=None) + + def _verify(self, entry: dict) -> None: + if entry["check"] == "skip": + return # 保持 unchecked 与不核对的理由 + if not isinstance(entry["sha256"], str) or _HASH_RE.fullmatch(entry["sha256"]) is None: + entry["reason"] = "没有记录的 sha256，无从核对" + return + checker = {"git": self._check_git, "comment": self._check_comment, + "artifact": self._check_artifact, "snapshot": self._check_snapshot, + "ledger": self._check_ledger}[entry["check"]] + checker(entry) + + def _check_git(self, entry: dict) -> None: + match = _GIT_REF_RE.fullmatch(str(entry["ref"])) + if match is None or not _safe_repo_path(match["path"]): + entry.update(status="unavailable", reason="path 不是仓库相对引用或逃出仓库，已拒绝核对") + return + data = _git_bytes(self.cwd, match["rev"], match["path"]) + if data is None: + entry.update(status="unavailable", reason=f"{match['path']}@{match['rev']} 无法从 git 读取") + return + self._compare(entry, data) + + def _check_comment(self, entry: dict) -> None: + match = _COMMENT_REF_RE.fullmatch(str(entry["ref"])) + if match is None: + entry.update(status="unavailable", reason="评论引用形状不符，已拒绝核对") + return + if f"{match['owner']}/{match['name']}" != self.repo or int(match["pr"]) != self.pr: + entry.update(status="unavailable", reason="评论引用不属于当前仓库或本 PR，已拒绝核对") + return + try: + comment = self.client.api(f"repos/{self.repo}/issues/comments/{match['id']}") + except (RuntimeError, ValueError, AttributeError): + entry.update(status="unavailable", reason="评论无法从 API 读取") + return + body = comment.get("body") if isinstance(comment, dict) else None + if not isinstance(body, str): + entry.update(status="unavailable", reason="评论响应形状不符") + return + self._compare(entry, body.encode("utf-8")) + + def _check_artifact(self, entry: dict) -> None: + digest = entry["artifact"] or entry["sha256"] + if not isinstance(digest, str) or _HASH_RE.fullmatch(digest) is None: + entry.update(status="unavailable", reason="产物引用不是 sha256，已拒绝核对") + return + directory = events_db.artifacts_dir() + data = None + if directory is not None: + try: + data = (directory / digest).read_bytes() + except OSError: + data = None + if data is not None: + self._compare(entry, data) + return + # 到期判定：产物索引行还在看 created，行已被保留期清理则回退引用事件的 ts。 + moment = _parse_utc(_artifact_created(digest)) or _parse_utc(entry.get("event_ts")) + if moment is not None and moment < self.cutoff: + entry.update(status="expired", reason="本机产物已过保留期清理，无法核对") + else: + entry.update(status="unavailable", reason="本机产物缺失，无法核对") + + def _check_snapshot(self, entry: dict) -> None: + try: + pull = self.client.api(f"repos/{self.repo}/pulls/{self.pr}") + except (RuntimeError, ValueError, AttributeError): + entry.update(status="unavailable", reason="PR 字段无法从 API 读取") + return + rendered = f"# {pull.get('title')}\n\n{pull.get('body') or ''}".encode() \ + if isinstance(pull, dict) else b"" + # 可变 PR 字段按评审同一渲染规则复取：变动是快照漂移（C6 snapshot_changed），不冒充字节篡改。 + if _digest(rendered) != entry["sha256"] or len(rendered) != entry["size"]: + entry.update(status="snapshot_changed", reason="PR 标题/正文与评审材料快照不符（可变字段已变动）") + else: + entry.update(status="verified", reason=None) + + def _check_ledger(self, entry: dict) -> None: + data = entry.get("ledger_bytes") + if data is None: + entry.update(status="unavailable", reason="账本文件无法从 harness-audit 分支读取") + return + self._compare(entry, data) + + # ---- 共享能力与汇总 ---- + + def _import_ci(self): + """CI 事件包交给 C5/T302 load_ci：下载、来源核对与规范化哈希导入都在那里完成。""" + try: + result = events_io.load_ci(self.pr, head=self.head, gh=self.client) + except Exception as exc: # noqa: BLE001 下载适配故障转发现，不替代审计 + result = {"findings": [{"code": "api", "detail": str(exc)}]} + for item in result.get("findings") or []: + code = item.get("code") + if code == "artifact_expired": + rule, reason = "reference_expired", "CI 事件包 artifact 已过保留期，未下载" + elif code == "hash_mismatch": + rule, reason = "hash_mismatch", "CI 事件包内容与规范化哈希不符，导入被拒" + else: + rule, reason = "reference_unavailable", f"CI 事件包不可用（{code}）" + self.findings.append(_finding(rule, source="ci", stage="ci", + ref=f"{self.repo}#{self.pr}/actions/artifacts", reason=reason)) + + def _sync_github(self): + """GitHub 事实由 T304 共享入口同步（合并/抽审/逃逸快照幂等导入），审计不自行改写。""" + try: + result = github_events.sync(pr=self.pr, gh=self.client) + except Exception as exc: # noqa: BLE001 同步故障转发现，不替代审计 + result = {"findings": [{"code": "api", "detail": str(exc)}]} + for item in result.get("findings") or []: + self.findings.append(_finding( + "reference_unavailable", source="github", stage="merge", ref=f"{self.repo}#{self.pr}", + reason=f"GitHub 事实同步不完整（{item.get('code')}）")) + + def run(self) -> dict: + self._import_ci() + self._sync_github() + self._collect_events() + anchor = self._collect_comments() + self._collect_run_records() + self._collect_ledger(anchor) + self._apply_expectations() + for entry in self.refs: + self._verify(entry) + for entry in self.refs: + rule = _RULE_FOR_STATUS.get(entry["status"]) + if rule is not None: + self.findings.append(_finding(rule, source=entry["source"], stage=entry["stage"], + ref=str(entry["ref"]), reason=entry["reason"])) + self.findings.extend(audit_completeness.check(self)) # T402：完整性与防篡改（含 [audit] 配置校验） + self.findings.sort(key=lambda item: (item["rule"], item["source"], item["stage"], item["ref"])) + self.refs.sort(key=lambda item: (item["kind"], str(item["ref"]))) + for entry in self.refs: + entry.pop("ledger_bytes", None) # 内部核对输入（字节）不进报告，报告只留哈希与结论 + self.stages.sort(key=lambda item: (str(item.get("ts")), str(item.get("stage")), str(item.get("step")))) + self.anchors.sort(key=lambda item: (str(item.get("ts")), str(item.get("source")))) + entry = next((item for item in self.refs if item["kind"] == "ledger"), None) + self.ledger_state = entry["status"] if entry is not None else "absent" + classes = Counter(_source_class(row["source"]) for row in self.event_rows) + statuses = Counter(item["status"] for item in self.refs) + coverage = {"events": {name: classes.get(name, 0) for name in _SOURCE_CLASSES}, + "chains": len(self.chains), "anchors": len(self.anchors), + "references": {"total": len(self.refs), + **{name: statuses.get(name, 0) for name in _STATUSES}}, + "ledger": self.ledger_state} + return {"pr": self.pr, "repository": self.repo, "trace_id": self.trace, "head_sha": self.head, + "merge_sha": self.merge_sha, "merged_at": self.merged_at, "stages": self.stages, + "references": self.refs, "findings": self.findings, "chains": self.chains, + "anchors": self.anchors, "coverage": coverage, "ok": not self.findings} + + +def _exit_code(report: dict) -> int: + """0 全部通过；1 有发现；2 配置错误（configuration_error 优先，非法 [audit] 不静默放宽）。""" + if any(item.get("rule") == "configuration_error" for item in report["findings"]): + return 2 + return 1 if report["findings"] else 0 + + +def _flatten_outputs(prefix: str, value, out: dict) -> None: + """嵌套汇总（coverage、规则/severity 计数）摊平成 outputs 扁平键：emit 过滤器只收标量值。""" + if isinstance(value, dict): + for key, item in value.items(): + _flatten_outputs(f"{prefix}.{key}" if prefix else str(key), item, out) + else: + out[prefix] = value + + +def _emit_observations(report: dict) -> None: + """audit 观察事件（C1：audit 自己记汇总；C7：写安全 ci/audit.summary 与 audit.finding 供周报）。 + + 每次运行恰一条 ci/audit.summary（ok/fail 随报告结论），每条 finding 追加一条 ci/audit.finding； + stage 固定 ci，不新增阶段枚举。outputs 只放安全字段（pr/head_sha/规则与 severity 计数/coverage/ok， + finding 事件另含 rule/severity/source/stage/ref，供周报按同 PR/head 取最新、唯一规则计数去重）， + reason 原文不入事件；trace 沿用当前分支，不混入被审计 PR 的事件链。emit 永不抛，组装失败同样 + 只丢弃——退出码与 stdout/stderr 不受影响（C0 三态等价）。 + """ + try: + findings = report.get("findings") or [] + summary = {"pr": report.get("pr"), "head_sha": report.get("head_sha"), "ok": report.get("ok"), + "severity": dict(Counter(item.get("severity") for item in findings)), + "rule": dict(Counter(item.get("rule") for item in findings)), + "coverage": report.get("coverage")} + outputs: dict = {} + _flatten_outputs("", summary, outputs) + events.emit(stage="ci", step="audit.summary", status="ok" if report.get("ok") else "fail", + outputs=outputs) + for item in findings: + events.emit(stage="ci", step="audit.finding", status="fail", + outputs={"pr": report.get("pr"), "head_sha": report.get("head_sha"), + "rule": item.get("rule"), "severity": item.get("severity"), + "source": item.get("source"), "stage": item.get("stage"), + "ref": item.get("ref")}) + except Exception: # noqa: BLE001 观察旁路（C0）：任何组装失败都不影响审计本身 + return + + +def inspect_pr(pr: int, *, gh=None, cwd: Path = ROOT) -> dict: + """复原并核对一个已合并 PR（C7 基础形状）；PR 无法审计（未合并/整体 API 故障）抛 AuditError。 + + 报告返回前写本运行的观察事件（T404）；写入失败不影响返回值与调用方。 + """ + client = GhClient() if gh is None else gh + try: + repo = events_io._repo_identity(str(client.repo())) + pull = client.api(f"repos/{repo}/pulls/{pr}") + except (RuntimeError, ValueError, AttributeError) as exc: + raise AuditError(f"读取 PR {pr} 失败：{exc}") from exc + if not isinstance(pull, dict) or pull.get("merged") is not True: + raise AuditError(f"PR {pr} 未合并或不存在（审计只覆盖已合并 PR，被拦下 PR 属 B52）") + head = pull.get("head") if isinstance(pull.get("head"), dict) else {} + trace, head_sha = head.get("ref"), head.get("sha") + merge_sha, merged_at = pull.get("merge_commit_sha"), pull.get("merged_at") + if not (isinstance(trace, str) and trace and isinstance(head_sha, str) and _SHA_RE.fullmatch(head_sha) + and isinstance(merge_sha, str) and _SHA_RE.fullmatch(merge_sha) + and isinstance(merged_at, str) and _parse_utc(merged_at) is not None): + raise AuditError(f"PR {pr} 的合并事实缺失（headRefName/head/merge_commit_sha/merged_at）") + report = _Auditor(client=client, repo=repo, pr=pr, trace=trace, head=head_sha, merge_sha=merge_sha, + merged_at=merged_at, cwd=Path(cwd)).run() + _emit_observations(report) + return report + + +# ---- CLI（cli.py COMMANDS 注册 audit；QUIET_COMMANDS 已含 audit，不追加自身事件）---- + + +def _render(report: dict) -> str: + head = report.get("head_sha") or "" + coverage = report["coverage"] + refs = coverage["references"] + lines = [(f"audit PR {report['pr']}（{report.get('trace_id') or '—'}）：head {head[:12] or '—'}…，" + f"合并 {report.get('merged_at') or '—'}")] + lines.append(f" 覆盖：阶段 {len(report['stages'])}（local {coverage['events']['local']} / " + f"ci {coverage['events']['ci']} / github {coverage['events']['github']}，链 " + f"{coverage['chains']}、锚点 {coverage['anchors']}）；引用 {refs['total']}" + f"（核对通过 {refs['verified']}、未核对 {refs['unchecked']}、不符 {refs['hash_mismatch']}、" + f"不可得 {refs['unavailable']}、到期 {refs['expired']}）；账本 {coverage['ledger']}") + if report["findings"]: + lines.append(f" 发现 {len(report['findings'])} 项：") + lines += [f" - [{item['severity']}] {item['rule']} {item['source']}/{item['stage']} " + f"{item['ref']}：{item['reason']}" for item in report["findings"]] + else: + lines.append(" 发现：无") + return "\n".join(lines) + + +def _select_merged(client, repo: str, cutoff: datetime | None, + problems: list[dict]) -> list[tuple[datetime, int]] | None: + """分页取全部已关闭 PR，按一次性确定的 UTC 窗口过滤出已合并者；列表失败返回 None。""" + pulls = github_events._collect_pages( + client, f"repos/{repo}/pulls?state=closed&sort=created&direction=desc", "已关闭 PR 列表", problems) + if pulls is None: + return None + selected: list[tuple[datetime, int]] = [] + for pull in pulls: + if not isinstance(pull, dict) or pull.get("merged") is not True: + continue + moment = _parse_utc(pull.get("merged_at")) + if moment is None or (cutoff is not None and moment < cutoff): + continue + if isinstance(pull.get("number"), int): + selected.append((moment, pull["number"])) + selected.sort() + return selected + + +def _audit_or_failure(client, repo: str, number: int) -> dict: + """批量里单个 PR 的审计；无法审计（合并事实缺失等）转成一条发现，不中断整批。""" + try: + return inspect_pr(number, gh=client, cwd=ROOT) + except AuditError: + report = {"pr": number, "repository": repo, "trace_id": None, "head_sha": None, + "merge_sha": None, "merged_at": None, "stages": [], "references": [], + "chains": [], "anchors": [], + "findings": [_finding("reference_unavailable", source="github", stage="merge", + ref=f"{repo}#{number}", reason="PR 无法审计：合并事实缺失")], + "coverage": {"events": {name: 0 for name in _SOURCE_CLASSES}, "chains": 0, + "anchors": 0, "references": {"total": 0, **{name: 0 for name in _STATUSES}}, + "ledger": "absent"}, + "ok": False} + _emit_observations(report) # 无法审计的运行同样留一条汇总与发现事件（C7 供周报） + return report + + +def _all_merged(args, as_json: bool) -> int: + """批量模式：分页全部已关闭 PR，按一次性确定的窗口过滤已合并者逐个审计。""" + client = GhClient() + try: + repo = events_io._repo_identity(str(client.repo())) + except (RuntimeError, ValueError, AttributeError) as exc: + print(f"audit：确定仓库失败：{exc}", file=sys.stderr) + return 2 + try: + cutoff = _parse_window(args.since) if args.since else None + except ValueError as exc: + print(str(exc), file=sys.stderr) + return 2 + problems: list[dict] = [] + selected = _select_merged(client, repo, cutoff, problems) + if selected is None: + for item in problems: + print(f"audit：{item['detail']}", file=sys.stderr) + return 2 + until = _parse_utc(events_db._now()) or datetime.now(UTC) + reports = [_audit_or_failure(client, repo, number) for _moment, number in selected] + for report in reports: + if not as_json: + print("\n".join(f" {line}" for line in _render(report).splitlines())) + window = {"since": cutoff.astimezone(UTC).isoformat() if cutoff else None, + "until": until.astimezone(UTC).isoformat()} + if as_json: + print(json.dumps({"window": window, "prs": reports}, ensure_ascii=False, indent=2)) + else: + print(f"audit --all-merged：窗口 {window['since'] or '全部'} → {window['until']}，" + f"共 {len(reports)} 个已合并 PR") + return max((_exit_code(report) for report in reports), default=0) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="audit", description="审计引用复原与哈希核对（设计 4.2）") + parser.add_argument("pr", nargs="?", type=int, help="审计指定已合并 PR") + parser.add_argument("--all-merged", action="store_true", help="批量审计时间窗口内的全部已合并 PR") + parser.add_argument("--since", metavar="时长|UTC", help="批量模式的窗口起点（如 30d 或 ISO 时间）") + parser.add_argument("--json", action="store_true", help="输出 JSON") + args = parser.parse_args(argv) + if args.pr is not None and args.all_merged: + print("PR 号与 --all-merged 二选一（两种模式互斥）", file=sys.stderr) + return 2 + if args.since is not None and not args.all_merged: + print("--since 只与 --all-merged 同用", file=sys.stderr) + return 2 + if args.pr is None and not args.all_merged: + parser.print_usage(sys.stderr) + print("audit：缺少审计目标（PR 号或 --all-merged）", file=sys.stderr) + return 2 + if args.pr is not None: + try: + report = inspect_pr(args.pr, cwd=ROOT) # ROOT 在调用时解析（审计所在仓库） + except AuditError as exc: + print(f"audit：{exc}", file=sys.stderr) + return 2 + print(json.dumps(report, ensure_ascii=False, indent=2) if args.json else _render(report)) + return _exit_code(report) + return _all_merged(args, args.json) diff --git a/.harness/engine/reports/audit_completeness.py b/.harness/engine/reports/audit_completeness.py new file mode 100644 index 0000000..5253cd7 --- /dev/null +++ b/.harness/engine/reports/audit_completeness.py @@ -0,0 +1,313 @@ +"""审计完整性与防篡改判定（B46 T402，设计 4.2，共用合同 C7）。 + +在 T401 复原与逐引用核对之上补两层： + +完整性（应有环节）：按风险与 PR 种类检查 R2+ 有独立评审且身份不同、自动合并有受信任 CI 的 +route.result（outputs.auto_merge=true）、任务 PR 有运行记录与阶段锚点、app 模式 R0/R1 自动合并由 +批准 App 批准且绑定合并 head；none 模式明确无 App 适用，不凭空要求批准。风险取受信任 CI 链上最新 +route.facts 的机器判定；任务 PR＝route.facts 声明任务书类别 ∨ 评审摘要列有执行方 ∨ 已有运行记录 +（2026-10-02 任务书修订口径），设计方 PR 不要求派发记录。「自动合并」取合并事实的 merger_type=Bot +（两种模式的自动合并工作流都以机器人身份合并，人工合并是 User）；风险不可判定时对应规则不适用， +缺失的 CI 事实本身已由引用核对与 missing_route 报告。 + +防篡改：校验各 (source, trace) 原始链（events_db.verify）；每个固定锚点（运行记录/CI artifact/PR +评论）核对「对应前缀链上有该链头」而非最终链头——合法后续追加不误报，删尾即使内部链仍合法也能由 +固定锚点发现；账本链头与运行层一致（账本固定后的合法追加不误报）。链未知或不可得不当完整：事件库 +不可读产生如实发现并按链不可校验处理，不当「无事件」放过。 + +checks.toml 可选 [audit]：require_review_risk（0..3，缺省 2）、require_route_for_auto、 +require_run_record_for_task、verify_anchors（布尔，缺省 true）。缺省执行设计规则；未知键、类型或 +取值非法产生 configuration_error 发现（CLI 退出 2），不静默放宽——非法键按设计缺省执行并在报告中 +指明。配置只影响 audit 报告：本模块只在审计时被调用，policy/guard/verify 判定不经过这里。 +""" + +from __future__ import annotations + +import re + +from engine.core import events_db +from engine.core.common import load_checks +from engine.routing import policy + +# [audit] 已知键与设计缺省（C7）；缺省值即缺省执行的设计规则。 +_DEFAULTS = {"require_review_risk": 2, "require_route_for_auto": True, + "require_run_record_for_task": True, "verify_anchors": True} +_CONFIG_REF = ".harness/config/checks.toml [audit]" +_RISK_RE = re.compile(r"R([0-3])\Z") +# T201 运行记录固定的锚点；missing_anchor 只认它，CI/PR 评论锚点是额外固定点，不豁免记录锚点。 +_RECORD_FIXED_IN = "run_record" + + +def _finding(rule: str, *, source: str, stage: str, ref: str, reason: str) -> dict: + """C7 finding（severity 全部 error）：reason 只写分类与事实，不回显引用内容。""" + return {"rule": rule, "severity": "error", "source": source, "stage": stage, + "ref": ref, "reason": reason} + + +def load_config() -> tuple[dict, list[dict]]: + """读 [audit] 并校验（C7）：返回（生效配置, configuration_error 发现）；非法键按设计缺省执行。""" + config = dict(_DEFAULTS) + try: + section = load_checks().get("audit", {}) + except Exception as exc: # noqa: BLE001 配置损坏明确报告，不静默放宽 + return config, [_finding("configuration_error", source="local", stage="audit", ref=_CONFIG_REF, + reason=f"配置无法读取（{type(exc).__name__}），按设计缺省执行")] + if section is None: + section = {} + if not isinstance(section, dict): + return config, [_finding("configuration_error", source="local", stage="audit", ref=_CONFIG_REF, + reason="[audit] 不是配置表，按设计缺省执行")] + problems: list[dict] = [] + for key, value in section.items(): + if key not in _DEFAULTS: + problems.append(_finding("configuration_error", source="local", stage="audit", ref=_CONFIG_REF, + reason=f"未知配置键 {key}，按设计缺省执行")) + elif key == "require_review_risk": + if isinstance(value, bool) or not isinstance(value, int) or not 0 <= value <= 3: + problems.append(_finding("configuration_error", source="local", stage="audit", + ref=_CONFIG_REF, reason="require_review_risk 须是 0..3 的整数")) + else: + config[key] = value + elif not isinstance(value, bool): + problems.append(_finding("configuration_error", source="local", stage="audit", ref=_CONFIG_REF, + reason=f"{key} 须是布尔值")) + else: + config[key] = value + return config, problems + + +def _risk_level(label) -> int | None: + """route.facts 的机器判定 R0..R3 → 0..3；缺失或形状不符返回 None（无法判定）。""" + if not isinstance(label, str): + return None + match = _RISK_RE.fullmatch(label.strip().upper()) + return int(match[1]) if match else None + + +def _source_class(source: str) -> str: + if str(source).startswith("github:"): + return "github" + if str(source).startswith("ci"): + return "ci" + return "local" + + +def _trusted_ci(row: dict) -> bool: + """受信任 CI 链（C5 导入时已核对 origin head）：ci: 前缀。local 链不冒充 main 判定。""" + return str(row.get("source") or "").startswith("ci") + + +def _outputs(row: dict | None) -> dict: + value = row.get("outputs") if isinstance(row, dict) else None + return value if isinstance(value, dict) else {} + + +class _Facts: + """从复原数据提炼的适用性事实：规则只依赖这些判定，不各自翻原始行。""" + + def __init__(self, auditor): + rows = auditor.event_rows or [] + trusted_route = [row for row in rows if row.get("stage") == "route" and _trusted_ci(row)] + facts_rows = [row for row in trusted_route if row.get("step") == "facts"] + latest_facts = max(facts_rows, key=lambda row: str(row.get("ts")), default=None) + self.risk = _risk_level(_outputs(latest_facts).get("risk")) + declared = _outputs(latest_facts).get("declared_class") + self.reviews = [stage for stage in auditor.stages or [] + if stage.get("evidence") == "review_comment_summary"] + merge_rows = [row for row in rows + if row.get("stage") == "merge" and row.get("step") == "github.merge"] + latest_merge = max(merge_rows, key=lambda row: str(row.get("ts")), default=None) + merge_out = _outputs(latest_merge) + self.auto_merged = merge_out.get("merger_type") == "Bot" + self.approval_out = merge_out + self.route_result_ok = any(_outputs(row).get("auto_merge") is True + for row in trusted_route if row.get("step") == "result") + self.run_records = any(stage.get("evidence") == "run_record_summary" + for stage in auditor.stages or []) + # 任务 PR 三证据（任务书修订口径）：声明类别 ∨ 评审列有执行方 ∨ 已有运行记录，任一即成立。 + implemented = any(isinstance(stage.get("implementers"), list) and stage["implementers"] + for stage in self.reviews) + self.task_pr = declared is not None or implemented or self.run_records + self.record_anchors = [anchor for anchor in auditor.anchors or [] + if anchor.get("fixed_in") == _RECORD_FIXED_IN] + + +def _independent(review: dict) -> bool: + """评审身份独立：independent 不为 False，且评审者不同于设计方、不在执行方列表里（不区分大小写）。""" + if review.get("independent") is False: + return False + reviewer = str(review.get("reviewer") or "").strip().lower() + designer = str(review.get("designer") or "").strip().lower() + implementers = {str(item).strip().lower() for item in review.get("implementers") or [] + if isinstance(item, str)} + return bool(reviewer) and reviewer != designer and reviewer not in implementers + + +def _review(facts: _Facts, config: dict, ref: str) -> list[dict]: + """missing_review（无合格独立评审）与 reviewer_not_independent（逐条身份冲突）。""" + findings: list[dict] = [] + if facts.risk is not None and facts.risk >= config["require_review_risk"] \ + and not any(_independent(review) for review in facts.reviews): + reason = "没有评审摘要，应有独立评审" if not facts.reviews else "现有评审身份不独立，无合格独立评审" + findings.append(_finding("missing_review", source="github", stage="review", ref=ref, reason=reason)) + for review in facts.reviews: + if _independent(review): + continue + comment = review.get("comment") if isinstance(review.get("comment"), dict) else {} + where = f"{ref}/comments/{comment['id']}" if comment.get("id") is not None else ref + findings.append(_finding("reviewer_not_independent", source="github", stage="review", ref=where, + reason="评审者与设计方或执行方身份相同（或 independent 显式为 false）")) + return findings + + +def _route(facts: _Facts, config: dict, ref: str) -> list[dict]: + """missing_route：自动合并必须有受信任 CI 链上 auto_merge=true 的 route.result。""" + if config["require_route_for_auto"] and facts.auto_merged and not facts.route_result_ok: + return [_finding("missing_route", source="ci", stage="route", ref=ref, + reason="自动合并没有受信任 CI 来源（ci: 前缀）的 route.result（auto_merge=true）")] + return [] + + +def _record_anchors(facts: _Facts, config: dict, ref: str) -> list[dict]: + """missing_run_record / missing_anchor：任务 PR 必须有运行记录，且记录固定了阶段锚点。""" + if not config["require_run_record_for_task"] or not facts.task_pr: + return [] + if not facts.run_records: + return [_finding("missing_run_record", source="local", stage="dispatch", ref=ref, + reason="任务 PR 在合并 head 上没有运行记录")] + if not facts.record_anchors: + return [_finding("missing_anchor", source="local", stage="dispatch", ref=ref, + reason="运行记录没有固定的阶段锚点（fixed_in=run_record）")] + return [] + + +def _approval(facts: _Facts, config: dict, mode: str | None, ref: str) -> list[dict]: + """approval_actor / approval_head：app 模式 R0/R1 自动合并须批准者 App 且绑定合并 head。 + + none 模式无 App 适用（不凭空要求批准）；人工合并不是自动合并，同样不适用。 + """ + if mode != "app" or not facts.auto_merged: + return [] + if facts.risk is None or facts.risk > 1: + return [] + findings: list[dict] = [] + if facts.approval_out.get("approver_type") != "Bot": + findings.append(_finding("approval_actor", source="github", stage="merge", ref=ref, + reason=f"app 模式 R{facts.risk} 自动合并的批准者不是 App" + f"（approver_type={facts.approval_out.get('approver_type')}）")) + if facts.approval_out.get("approval_bound") is not True: + detail = "批准没有绑定提交" if facts.approval_out.get("approval_commit") is None \ + else "批准绑定的提交不是合并的 head" + findings.append(_finding("approval_head", source="github", stage="merge", ref=ref, reason=detail)) + return findings + + +def _chains(auditor, config: dict) -> list[dict]: + """chain_invalid（原始链校验失败）；锚点核对见 _anchors。""" + findings: list[dict] = [] + hashes: dict[str, set] = {} + last_stage: dict[str, str] = {} + last_seq: dict[str, int] = {} + for row in auditor.event_rows or []: + source = str(row.get("source") or "") + hashes.setdefault(source, set()).add(row.get("hash")) + if source not in last_seq or int(row.get("seq") or 0) > last_seq[source]: + last_seq[source] = int(row.get("seq") or 0) + last_stage[source] = str(row.get("stage") or "dispatch") + if getattr(auditor, "db_error", None): + findings.append(_finding("chain_invalid", source="local", stage="audit", ref="harness.db", + reason="事件库无法读取，链与锚点不可校验（不当完好）")) + for chain in auditor.chains or []: + source = str(chain.get("source") or "") + try: + problems = events_db.verify(trace_id=chain.get("trace_id"), source=source) + except Exception as exc: # noqa: BLE001 坏库：链不可校验要如实报告 + problems = [f"链校验无法执行（{type(exc).__name__}）"] + if problems: + findings.append(_finding("chain_invalid", source=_source_class(source), + stage=last_stage.get(source, "dispatch"), + ref=f"{source}/{chain.get('trace_id')}", + reason=f"原始链校验失败：{problems[0]}")) + return findings + _anchors(auditor, config, hashes) + + +def _anchors(auditor, config: dict, hashes: dict[str, set]) -> list[dict]: + """anchor_mismatch：固定锚点按前缀成员核对——锚点之后的合法追加不算篡改，链头消失即发现。""" + if not config["verify_anchors"]: + return [] + findings: list[dict] = [] + seen: set[tuple] = set() + for anchor in auditor.anchors or []: + key = (anchor.get("source"), anchor.get("stage"), anchor.get("head_hash"), anchor.get("fixed_in")) + if key in seen: + continue + seen.add(key) + known = hashes.get(str(anchor.get("source") or "")) + if known is not None and anchor.get("head_hash") in known: + continue # 前缀成员：链在锚点之后继续增长属合法追加 + findings.append(_finding("anchor_mismatch", source=_source_class(anchor.get("source") or ""), + stage=str(anchor.get("stage") or "dispatch"), + ref=f"{anchor.get('fixed_in')}:{anchor.get('stage')}" + f"@{str(anchor.get('head_hash') or '')[:12]}", + reason="固定锚点的链头不在对应链上（链缺失或锚点之后的历史被改动）")) + return findings + + +def _ledger(auditor) -> list[dict]: + """ledger_mismatch：账本链头与运行层一致；账本固定后的合法追加（链头仍是前缀成员）不误报。""" + doc = getattr(auditor, "ledger_doc", None) + if not isinstance(doc, dict): + return [] # 账本缺失/不可解析由 T401 的引用核对报告，不在这里重复 + hashes: dict[str, set] = {} + for row in auditor.event_rows or []: + hashes.setdefault(str(row.get("source") or ""), set()).add(row.get("hash")) + findings: list[dict] = [] + for chain in doc.get("chains") or []: + if not isinstance(chain, dict): + findings.append(_finding("ledger_mismatch", source="github", stage="merge", ref="ledger:chains", + reason="账本链条目形状不符")) + continue + source, head = str(chain.get("source") or ""), chain.get("head_hash") + known = hashes.get(source) + if not known: + continue # 运行层没有该链：链不可得由引用核对/missing_route 报告，不在这里猜 + if head == _chain_head(auditor, source, chain.get("trace_id")) or head in known: + continue + findings.append(_finding("ledger_mismatch", source="github", stage="merge", + ref=f"ledger:{source}/{chain.get('trace_id')}", + reason="账本链头不在运行层对应链上（账本与运行层不一致）")) + return findings + + +def _chain_head(auditor, source: str, trace_id) -> str | None: + """运行层该链当前链头；库里没有事件或读取失败返回 None（调用方再按前缀成员兜底）。""" + try: + return events_db.chain_head(source, trace_id or auditor.trace) + except Exception: # noqa: BLE001 坏库：与 None 同样交由前缀成员判定 + return None + + +def _approval_mode(findings: list[dict]) -> str | None: + """被审计仓库的批准方式（与自动合并工作流同一配置）；配置损坏记 configuration_error 并跳过审批核对。""" + try: + return policy.platform_outputs()["approval"] + except Exception as exc: # noqa: BLE001 平台配置非法是配置错误，不是「none」 + findings.append(_finding("configuration_error", source="local", stage="audit", + ref=".harness/config/checks.toml [platform]", + reason=f"批准方式配置非法（{type(exc).__name__}），审批核对未执行")) + return None + + +def check(auditor) -> list[dict]: + """完整性与防篡改 findings（在 T401 复原核对之后调用；只读，不改复原结果）。""" + config, findings = load_config() + mode = _approval_mode(findings) + facts = _Facts(auditor) + ref = f"{auditor.repo}#{auditor.pr}" + findings += _review(facts, config, ref) + findings += _route(facts, config, ref) + findings += _record_anchors(facts, config, f"docs/runs@{auditor.merge_sha}") + findings += _approval(facts, config, mode, ref) + findings += _chains(auditor, config) + findings += _ledger(auditor) + return findings diff --git a/tests/test_harness_dispatch.py b/tests/test_harness_dispatch.py index 3c4dd0a..b61bdef 100644 --- a/tests/test_harness_dispatch.py +++ b/tests/test_harness_dispatch.py @@ -21,9 +21,12 @@ from engine.agents import dispatch, dispatch_host GIT_ENV = { - "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@example.com", - "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@example.com", - "GIT_CONFIG_GLOBAL": os.devnull, "GIT_CONFIG_NOSYSTEM": "1", + "GIT_AUTHOR_NAME": "t", + "GIT_AUTHOR_EMAIL": "t@example.com", + "GIT_COMMITTER_NAME": "t", + "GIT_COMMITTER_EMAIL": "t@example.com", + "GIT_CONFIG_GLOBAL": os.devnull, + "GIT_CONFIG_NOSYSTEM": "1", } TASKBOOK = """--- @@ -58,7 +61,7 @@ SPEC = "| 编号 | 内容 | 证据类型 | 覆盖 |\n|---|---|---|---|\n| DR14 | 合并 | 单测 | `test_mod` |\n" # 假执行方：按模式在槽位里做事。 -FAKE = textwrap.dedent(''' +FAKE = textwrap.dedent(""" import subprocess, sys, time from pathlib import Path mode = sys.argv[1] @@ -81,15 +84,15 @@ def commit(name, text): elif mode == "note": Path("build/dispatch").mkdir(parents=True, exist_ok=True) Path("build/dispatch/escalation.md").write_text("- 可选方案与推荐：A\\n- 需要决定的问题：Q") -''') +""") # 假 verify：有 bad*.txt 即以同样的理由失败。 -VERIFY = textwrap.dedent(''' +VERIFY = textwrap.dedent(""" import sys from pathlib import Path if list(Path(".").glob("bad*.txt")): print("FAIL: test_mod.Case 断言失败 1 != 2"); sys.exit(1) -''') +""") class FakeHost: @@ -113,13 +116,24 @@ def parse(self, events): class FakeGitHub: def __init__(self, claimed=(), ci=(True,)): self.claimed, self.ci = set(claimed), list(ci) - self.pushes, self.prs, self.comments, self.labels, self.issues = [], [], [], [], [] + self.pushes, self.prs, self.comments, self.labels, self.issues = ( + [], + [], + [], + [], + [], + ) def remote_branch_exists(self, branch): return branch in self.claimed def push(self, slot, branch): - subprocess.run(["git", "push", "-q", "-u", "origin", branch], cwd=slot, check=True, capture_output=True) + subprocess.run( + ["git", "push", "-q", "-u", "origin", branch], + cwd=slot, + check=True, + capture_output=True, + ) self.pushes.append(branch) def open_pr(self, slot, branch, title, body): @@ -136,6 +150,7 @@ def create_issue(self, title, body, labels): self.issues.append((title, body, labels)) def wait_ci(self, branch, sha, timeout, detail=None): + # detail：引擎 B71 修复后 wait_ci 携带的升级通道附言（已推 head 等），夹具忽略其内容。 ok = self.ci.pop(0) if len(self.ci) > 1 else self.ci[0] return ok, "" if ok else "CI 未通过：test_mod.Case" @@ -148,9 +163,15 @@ def setUp(self): self.tmp = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, self.tmp, True) origin = self.tmp / "origin.git" - subprocess.run(["git", "init", "-q", "--bare", "-b", "main", str(origin)], check=True) + subprocess.run( + ["git", "init", "-q", "--bare", "-b", "main", str(origin)], check=True + ) self.root = self.tmp / "repo" - subprocess.run(["git", "clone", "-q", str(origin), str(self.root)], check=True, capture_output=True) + subprocess.run( + ["git", "clone", "-q", str(origin), str(self.root)], + check=True, + capture_output=True, + ) self.write_task(ci_rounds=3) (self.root / "docs/specs").mkdir(parents=True) (self.root / "docs/specs/daily-report.md").write_text(SPEC) @@ -161,23 +182,40 @@ def setUp(self): self.git("commit", "-q", "-m", "base") self.git("push", "-q", "origin", "HEAD:main") self.git("fetch", "-q", "origin") - self.config = dispatch.Config(slots=2, slot_root=self.tmp / "slots", stall_seconds=30, poll_seconds=0.1, - verify=[sys.executable, "verify.py"]) + self.config = dispatch.Config( + slots=2, + slot_root=self.tmp / "slots", + stall_seconds=30, + poll_seconds=0.1, + verify=[sys.executable, "verify.py"], + ) (self.tmp / "slots").mkdir() - guard = mock.patch.object(dispatch, "prepare_guard", return_value=(Path("guard.ts"), "abc123")) + guard = mock.patch.object( + dispatch, "prepare_guard", return_value=(Path("guard.ts"), "abc123") + ) guard.start() self.addCleanup(guard.stop) def git(self, *args, cwd=None): - return subprocess.run(["git", *args], cwd=cwd or self.root, check=True, capture_output=True, text=True).stdout + return subprocess.run( + ["git", *args], + cwd=cwd or self.root, + check=True, + capture_output=True, + text=True, + ).stdout def write_task(self, ci_rounds): (self.root / "docs/plans").mkdir(parents=True, exist_ok=True) - (self.root / "docs/plans/task-005-new-test.md").write_text(TASKBOOK.format(ci_rounds=ci_rounds)) + (self.root / "docs/plans/task-005-new-test.md").write_text( + TASKBOOK.format(ci_rounds=ci_rounds) + ) def dispatcher(self, modes, github=None): host = FakeHost(self.root / "fake.py", modes) - return dispatch.Dispatcher(self.root, self.config, github or FakeGitHub(), host), host + return dispatch.Dispatcher( + self.root, self.config, github or FakeGitHub(), host + ), host def run_task(self, modes, github=None): runner, host = self.dispatcher(modes, github) @@ -188,24 +226,35 @@ def slot(self): return dispatch.slot_path(self.root, self.config, 1) def record(self, number=1): - return json.loads((self.slot() / f"docs/runs/task-005-new-test/{number}.json").read_text()) + return json.loads( + (self.slot() / f"docs/runs/task-005-new-test/{number}.json").read_text() + ) def test_success_writes_record_opens_pr_and_leaves_main_untouched(self): github = FakeGitHub() code, _, host = self.run_task(["commit"], github) self.assertEqual(code, 0) record = self.record() - self.assertEqual((record["task"], record["exit"], record["gen_ai.request.model"]), ("T005", "ok", "fake-model")) + self.assertEqual( + (record["task"], record["exit"], record["gen_ai.request.model"]), + ("T005", "ok", "fake-model"), + ) self.assertEqual(record["gen_ai.usage.input_tokens"], 10) self.assertEqual(record["guard_ref"], "abc123") self.assertTrue((self.slot() / record["prompt_path"]).exists()) self.assertEqual(len(github.prs), 1) self.assertIn("docs/runs/task-005-new-test/1.json", github.prs[0][2]) - self.assertEqual(github.pushes, ["task/005-new-test", "task/005-new-test"]) # 认领 + 结果 - self.assertIn("Task: T005", self.git("log", "-1", "--format=%B", cwd=self.slot())) + self.assertEqual( + github.pushes, ["task/005-new-test", "task/005-new-test"] + ) # 认领 + 结果 + self.assertIn( + "Task: T005", self.git("log", "-1", "--format=%B", cwd=self.slot()) + ) self.assertEqual(self.git("status", "--porcelain"), "") # 主目录不变 self.assertIn("docs/plans/task-005-new-test.md", host.prompts[0]) - self.assertFalse(list((dispatch.state_dir(self.root) / "slots").glob("*.json"))) # 槽位已归还 + self.assertFalse( + list((dispatch.state_dir(self.root) / "slots").glob("*.json")) + ) # 槽位已归还 def test_executor_gets_no_github_token(self): seen = {} @@ -215,8 +264,12 @@ def spy(argv, cwd, env, *args, **kwargs): seen.update(env) return original(argv, cwd, env, *args, **kwargs) - with mock.patch.dict(os.environ, {"GH_TOKEN": "secret-token", "GITHUB_TOKEN": "x"}), \ - mock.patch.object(dispatch_host, "run_monitored", spy): + with ( + mock.patch.dict( + os.environ, {"GH_TOKEN": "secret-token", "GITHUB_TOKEN": "x"} + ), + mock.patch.object(dispatch_host, "run_monitored", spy), + ): self.run_task(["commit"]) self.assertNotIn("GH_TOKEN", seen) self.assertNotIn("GITHUB_TOKEN", seen) @@ -225,7 +278,9 @@ def spy(argv, cwd, env, *args, **kwargs): self.assertEqual(seen["GIT_AUTHOR_NAME"], dispatch.agent_identity()[0]) def test_claimed_branch_stops_before_running(self): - runner, host = self.dispatcher(["commit"], FakeGitHub(claimed={"task/005-new-test"})) + runner, host = self.dispatcher( + ["commit"], FakeGitHub(claimed={"task/005-new-test"}) + ) with self.assertRaisesRegex(dispatch.Stop, "已被认领"): runner.run("docs/plans/task-005-new-test.md") self.assertEqual(host.prompts, []) @@ -239,7 +294,9 @@ def test_taskbook_not_on_main_is_refused(self): def test_exempt_history_taskbook_is_refused(self): (self.root / ".harness/state").mkdir(parents=True, exist_ok=True) - (self.root / ".harness/state/taskbook-exempt.txt").write_text("docs/plans/task-005-new-test.md # 历史\n") + (self.root / ".harness/state/taskbook-exempt.txt").write_text( + "docs/plans/task-005-new-test.md # 历史\n" + ) with self.assertRaisesRegex(dispatch.Stop, "豁免"): dispatch.admit("docs/plans/task-005-new-test.md", self.root) @@ -266,7 +323,9 @@ def test_same_failure_twice_is_a_loop(self): record = self.record() self.assertEqual(record["exit"], "loop") self.assertEqual(len(record["failure_signatures"]), 2) - self.assertEqual(record["failure_signatures"][0], record["failure_signatures"][1]) + self.assertEqual( + record["failure_signatures"][0], record["failure_signatures"][1] + ) self.assertIn("断言失败", host.prompts[1]) # 第二轮带上了失败摘要 self.assertIn("打转", runner.github.issues[0][1]) @@ -337,26 +396,51 @@ def setUp(self): self.tmp = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, self.tmp, True) origin = self.tmp / "origin.git" - subprocess.run(["git", "init", "-q", "--bare", "-b", "main", str(origin)], check=True) + subprocess.run( + ["git", "init", "-q", "--bare", "-b", "main", str(origin)], check=True + ) self.root = self.tmp / "repo" - subprocess.run(["git", "clone", "-q", str(origin), str(self.root)], check=True, capture_output=True) + subprocess.run( + ["git", "clone", "-q", str(origin), str(self.root)], + check=True, + capture_output=True, + ) def publish(self, broken=False): - shutil.copytree(ROOT / ".harness", self.root / ".harness", ignore=shutil.ignore_patterns("__pycache__")) + shutil.copytree( + ROOT / ".harness", + self.root / ".harness", + ignore=shutil.ignore_patterns("__pycache__"), + ) shutil.copytree(ROOT / ".pi", self.root / ".pi") if broken: - (self.root / ".harness/engine/guards/command_guard.py").write_text("raise SystemExit(0)\n") - for args in (["add", "-A"], ["commit", "-q", "-m", "guard"], ["push", "-q", "origin", "HEAD:main"], - ["fetch", "-q", "origin"]): - subprocess.run(["git", *args], cwd=self.root, check=True, capture_output=True) + (self.root / ".harness/engine/guards/command_guard.py").write_text( + "raise SystemExit(0)\n" + ) + for args in ( + ["add", "-A"], + ["commit", "-q", "-m", "guard"], + ["push", "-q", "origin", "HEAD:main"], + ["fetch", "-q", "origin"], + ): + subprocess.run( + ["git", *args], cwd=self.root, check=True, capture_output=True + ) def test_guard_comes_from_origin_main(self): self.publish() - (self.root / ".harness/engine/guards/command_guard.py").write_text("raise SystemExit(0)\n") # 工作区被改：不影响 + (self.root / ".harness/engine/guards/command_guard.py").write_text( + "raise SystemExit(0)\n" + ) # 工作区被改：不影响 extension, ref = dispatch.prepare_guard(self.root) self.assertTrue(extension.exists()) self.assertIn(ref, str(extension)) - self.assertNotIn("SystemExit(0)", (extension.parents[2] / ".harness/engine/guards/command_guard.py").read_text()) + self.assertNotIn( + "SystemExit(0)", + ( + extension.parents[2] / ".harness/engine/guards/command_guard.py" + ).read_text(), + ) def test_broken_guard_on_main_stops_dispatch(self): self.publish(broken=True) @@ -373,19 +457,51 @@ def test_argv_ignores_project_files_and_loads_guard_explicitly(self): def test_parse_usage_model_and_guard_denials(self): events = [ - {"type": "message_end", "message": {"role": "assistant", "model": "glm", "usage": { - "input": 100, "cacheRead": 50, "output": 20, "cost": {"total": 0.01}}}}, - {"type": "tool_execution_end", "isError": True, "result": {"content": [{"type": "text", "text": - "harness 守卫拒绝了这次操作：\n - 执行者不能编辑判定器与护栏\n 如确需执行……"}]}}, - {"type": "message_end", "message": {"role": "assistant", "model": "glm", "usage": { - "input": 10, "output": 5, "cost": {"total": 0.002}}}}, + { + "type": "message_end", + "message": { + "role": "assistant", + "model": "glm", + "usage": { + "input": 100, + "cacheRead": 50, + "output": 20, + "cost": {"total": 0.01}, + }, + }, + }, + { + "type": "tool_execution_end", + "isError": True, + "result": { + "content": [ + { + "type": "text", + "text": "harness 守卫拒绝了这次操作：\n - 执行者不能编辑判定器与护栏\n 如确需执行……", + } + ] + }, + }, + { + "type": "message_end", + "message": { + "role": "assistant", + "model": "glm", + "usage": {"input": 10, "output": 5, "cost": {"total": 0.002}}, + }, + }, ] with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "events.jsonl" - path.write_text("\n".join(json.dumps(event, ensure_ascii=False) for event in events) + "\nnot json\n") + path.write_text( + "\n".join(json.dumps(event, ensure_ascii=False) for event in events) + + "\nnot json\n" + ) model, usage, denials = dispatch_host.PiHost().parse(path) self.assertEqual(model, "glm") - self.assertEqual(usage, {"input_tokens": 160, "output_tokens": 25, "cost": 0.012}) + self.assertEqual( + usage, {"input_tokens": 160, "output_tokens": 25, "cost": 0.012} + ) self.assertEqual(denials, {"执行者不能编辑判定器与护栏": 1})