From 410d35dc6412e0be6b0b3f9ff465e2bfe0d1d62c Mon Sep 17 00:00:00 2001 From: lili Date: Wed, 22 Jul 2026 23:06:57 -0700 Subject: [PATCH] fix(dogfood-ops): bind waves to durable runtime evidence --- deploy/dogfood-ops/README.md | 31 +- deploy/dogfood-ops/dogfood_ops.py | 597 +++++++++++++++++-- deploy/dogfood-ops/tests/test_dogfood_ops.py | 518 +++++++++++++--- 3 files changed, 999 insertions(+), 147 deletions(-) diff --git a/deploy/dogfood-ops/README.md b/deploy/dogfood-ops/README.md index b0f4bae8..5a6ecaea 100644 --- a/deploy/dogfood-ops/README.md +++ b/deploy/dogfood-ops/README.md @@ -6,10 +6,10 @@ | 波次 | 创作者 | 玩家 | 连续值班覆盖 | |---|---:|---:|---:| -| wave0 | 3 | 5 | 至少 24 小时 | -| wave1 | 10 | 30 | 至少 72 小时 | +| wave0 | 3 | 5 | 从工具固化的 `startedAt` 起真实经过至少 24 小时 | +| wave1 | 10 | 30 | Wave0 收口后开始，真实经过至少 72 小时 | -初始化只接受这两个名称。重复初始化会校验既有 `manifest.json` 后返回，不覆盖任何已填写证据。 +初始化只接受这两个名称。`startedAt` 由工具读取 UTC 当前时间写入，CLI 不提供回填参数。重复初始化会校验既有 `manifest.json` 后返回，不覆盖任何已填写证据。Wave1 初始化前会重跑 Wave0 收口门，并把 Wave0 的 `close-result.json` 哈希、最后一份冻结清单哈希和 `closedAt` 固化进 Wave1 manifest；任一绑定变化、P0、确认发布假阳或超过 4 小时待审积压都会拒绝 Wave1。 ```bash deploy/dogfood-ops/init-wave.sh --root /受控证据根 --wave wave0 @@ -33,8 +33,12 @@ deploy/dogfood-ops/freeze-evidence.sh \ --wave-dir /受控证据根/wave0 \ --date 2026-07-22 -# 4. 收口门除全部开波条件外，还会重算最近冻结清单。 +# 4. 收口门除全部开波条件外，还会校验真实运行时长和全部实际 UTC 运行日； +# 通过后由工具生成 close-result.json。 deploy/dogfood-ops/check-close.sh --wave-dir /受控证据根/wave0 + +# 5. 只有 Wave0 收口回执仍可复核时，才允许初始化 Wave1。 +deploy/dogfood-ops/init-wave.sh --root /受控证据根 --wave wave1 ``` `GATE_PASS` 只表示本地证据在当前规则下可对账，不替代人工判断，不代表 staging 已部署、告警已送达或波次已完成。真实结论必须回到原始截图、网络记录、日志和数据库 ID 核验。 @@ -43,17 +47,17 @@ deploy/dogfood-ops/check-close.sh --wave-dir /受控证据根/wave0 | 文件 | 必填内容 | |---|---| -| `manifest.json` | 固定波次、人数和持续时间，不得修改 | -| `rc.json` | 完整 40 位 commit、tracked 干净标志、制品路径与实际 SHA-256 | +| `manifest.json` | 固定波次、人数、持续时间和工具生成的 `startedAt`；Wave1 另绑定 Wave0 收口回执、最终冻结哈希和 `closedAt` | +| `rc.json` | 完整 40 位 commit、tracked 干净标志、`rcArtifactRef` 与实际 `rcArtifactSha256`，只证明部署候选 | | `config-identity.json` | 同一 RC commit、配置包路径、实际 SHA-256、采集时间 | | `snapshot.json` | MySQL 快照路径/哈希，以及非 staging/production 目标的隔离恢复 PASS、时间和报告 | | `storage-audit.json` | 对象存储审计报告、实际哈希、采集时间 | | `accounts.json` | batchId、精确创作者/玩家计数、FREE 额度；wave0 至少 4，wave1 至少 12 | | `role-permissions.json` | 创作者本账号允许、跨账号拒绝、审核员只审、审核员改配置拒绝、只读写拒绝；每项带截图 | -| `main-chain.json` | 同一 project/version/artifact/game 的 create→generate→preview→submit-review→approve→feed→play→telemetry；每段同时带截图、网络、日志、数据库证据 | +| `main-chain.json` | 同一 project/version/game 的 `gameArtifactSha256`，分别绑定 APPROVED 审核记录、实际 runtime package 和主链阶段真实使用的数据库证据；完整覆盖 create→generate→preview→submit-review→approve→feed→play→telemetry | | `release-negatives.json` | DRAFT、REVIEWING、REJECTED、直调绕过、批准后产物漂移、下架后 feed/详情/深链八类拒绝证据 | | `manual-play.json` | 每个发布游戏在进入 feed 前由 `/gstack@mini-desktop` 真玩，带测试人、时间和截图 | -| `performance.json` | 固定环境下至少 30 次禁用缓存的独立导航，只含 cold 样本 | +| `performance.json` | 固定环境下至少 30 次独立 cold 导航；每条绑定 `/gstack` 原始 CDP trace/HAR 路径、实际 SHA-256、采集时间与采集器身份 | | `p0.tsv` | 单一 P0 账本；任一 `severity=P0` 行都会拒绝开波和收口 | | `telemetry.json` | 期望完整事件数、完整收到数、原始证据；完整率至少 95% | | `alerts.json` | 至少一条接收时间、接收人、外部消息 id 和原始证据，不接受仅写 PASS | @@ -62,15 +66,16 @@ deploy/dogfood-ops/check-close.sh --wave-dir /受控证据根/wave0 | `manual-acceptance.tsv` | 角色权限、主链、发布负例、入 feed 前真玩四项人工签认 | | `drills.json` | stop/delete/rollback 三类演练的精确 ACK、操作人、时间和目录内证据 | | `safety-status.json` | 固定停止条件逐项评估，任一缺失或触发即拒绝 | -| `freezes/*.sha256.json` | 当日全部证据和附件的 SHA-256 清单；wave0 至少 1 份、wave1 至少 3 份，按日连续并以前序清单哈希串联 | +| `freezes/*.sha256.json` | 工具生成的 `createdAt`、当日全部证据和附件 SHA-256；覆盖 `startedAt` 到收口时的每个实际 UTC 日期，并以前序清单哈希串联 | +| `close-result.json` | 仅在收口门通过后生成；绑定 manifest、最终冻结清单、真实 `startedAt/closedAt` 和三类 Wave1 阻断计数 | 所有时间使用带时区 ISO 8601，例如 `2026-07-22T08:00:00Z`。哈希必须是小写 64 位 SHA-256，并与附件内容一致。 ## 4. 首屏 P75 口径 -采样环境固定为 mini-desktop、Chrome 146、`390x844@1`、Tailscale staging。每次禁用缓存后独立导航，从 navigation commit 计到首个可交互游戏帧；每条记录使用唯一 `navigationId`，并标记 `cache=cold`、`source=/gstack@mini-desktop`。热缓存另行报告，禁止写入 `performance.json`。 +采样环境固定为 mini-desktop、Chrome 146、`390x844@1`、Tailscale staging。每次禁用缓存后独立导航，从 navigation commit 计到首个可交互游戏帧；每条记录使用唯一 `navigationId` 和唯一原始文件，并标记 `cache=cold`、`source=/gstack@mini-desktop`。`collector` 必须包含 `tool=/gstack`、`host=mini-desktop` 和非空 `sessionId`，且与原始文件内身份一致；`collectedAt` 必须带时区、不早于波次开始且不在未来。热缓存另行报告，禁止写入 `performance.json`。 -算法使用 nearest-rank：将 `n` 个 `durationMs` 升序排列，排名为 `ceil(0.75 × n)`，按 1 起算取值。30 个样本取第 23 个值。门槛是严格 `P75 < 3000ms`，所以 2999ms 通过，3000ms 失败；少于 30 个样本直接失败。 +门不接受样本自填 `durationMs`。CDP trace 必须各有一个同 `navigationId` 的 `dogfood.navigationCommit` 和 `dogfood.firstInteractiveGameFrame` 微秒事件；HAR 必须在对应 page 的 `_gstack` 中记录这两个原始时间点。门从受 SHA-256 保护的原始文件重算每条时长，再按 nearest-rank 排序，排名为 `ceil(0.75 × n)`。30 个样本取第 23 个值；门槛严格为 `P75 < 3000ms`，2999ms 通过，3000ms 失败，少于 30 个样本直接失败。 ## 5. 停止条件 @@ -119,7 +124,7 @@ deploy/dogfood-ops/record-drill.sh \ ## 8. 每日冻结 -每天完成采集后运行 `freeze-evidence.sh`。清单覆盖该波次目录下除 `freezes/` 自身外的全部普通文件，并记录上一日清单哈希。工具拒绝符号链接、日期空档和同日覆盖；wave0 收口至少需要 1 份，wave1 至少需要连续 3 份。若当天冻结后又补证，保留原清单并在下一日期重新冻结，不要修改历史清单。 +每天完成采集后运行 `freeze-evidence.sh`。`--date` 必须等于工具当前 UTC 日期，不能填未来日期，也不能回填历史日期；清单中的 `createdAt` 由工具同时生成。清单覆盖该波次目录下除 `freezes/` 和工具生成的 `close-result.json` 外的全部普通文件，并记录上一日清单哈希。工具拒绝符号链接、日期空档和同日覆盖；收口时按 `startedAt` 到当前时间的全部 UTC 日期逐日对账，因此跨日运行的 Wave0 通常需要两份，运行 72 小时的 Wave1 通常需要四份，不能在同一分钟伪造三天。若当天冻结后又补证，当日清单会拒绝覆盖，波次保持红灯。 ## 9. 真实采集待办 @@ -129,7 +134,7 @@ deploy/dogfood-ops/record-drill.sh \ - MySQL 本轮表快照、无网络隔离恢复报告、对象存储前缀审计； - wave0/wave1 精确账号与 FREE 额度盘点； - `/gstack` 在 mini-desktop 产生的角色权限、完整主链、八类发布负例、逐游戏入 feed 前真玩截图与网络证据； -- 30 次独立冷导航原始计时和单列的热缓存报告； +- 30 次独立冷导航的 `/gstack` 原始 CDP trace/HAR、逐文件哈希、采集时间和采集器身份，以及单列的热缓存报告； - P0、埋点完整率、告警接收回执、值班覆盖和反馈账本； - 停服、限定 batchId 删档、第二个 RC 回滚的人工演练证据； - 每个采集日的 SHA-256 冻结清单。 diff --git a/deploy/dogfood-ops/dogfood_ops.py b/deploy/dogfood-ops/dogfood_ops.py index c3f1fc41..99d9a0d6 100755 --- a/deploy/dogfood-ops/dogfood_ops.py +++ b/deploy/dogfood-ops/dogfood_ops.py @@ -10,7 +10,7 @@ import json import math import re import sys -from datetime import date, datetime +from datetime import date, datetime, timedelta, timezone from pathlib import Path from typing import Any @@ -20,6 +20,14 @@ WAVE_SCALES = { "wave1": {"creatorCount": 10, "playerCount": 30, "durationHours": 72}, } +MANIFEST_SCHEMA = "dogfood-wave-manifest/2" +CLOSE_RESULT_SCHEMA = "dogfood-wave-close/1" +CLOSE_RESULT_NAME = "close-result.json" +RC_SCHEMA = "dogfood-rc/2" +MAIN_CHAIN_SCHEMA = "dogfood-main-chain/2" +PERFORMANCE_SCHEMA = "dogfood-first-frame-performance/2" +FREEZE_SCHEMA = "dogfood-evidence-freeze/2" + TSV_TEMPLATES = { "oncall.tsv": "start_at\tend_at\tprimary\tbackup\tcontact\n", "feedback.tsv": "id\tcreated_at\treporter_id\trole\tcategory\tseverity\tsummary\tstatus\towner\tevidence_ref\n", @@ -28,16 +36,16 @@ TSV_TEMPLATES = { } JSON_TEMPLATES: dict[str, dict[str, Any]] = { - "rc.json": {"schema": "dogfood-rc/1", "commit": None, "repoClean": None, "artifactRef": None, "artifactSha256": None}, + "rc.json": {"schema": RC_SCHEMA, "commit": None, "repoClean": None, "rcArtifactRef": None, "rcArtifactSha256": None}, "config-identity.json": {"schema": "dogfood-config-identity/1", "capturedAt": None, "configSha256": None, "configBundleRef": None, "rcCommit": None}, "snapshot.json": {"schema": "dogfood-snapshot/1", "mysql": {}, "isolatedRestore": {}}, "storage-audit.json": {"schema": "dogfood-storage-audit/1", "result": None, "capturedAt": None, "reportRef": None, "reportSha256": None}, "accounts.json": {"schema": "dogfood-accounts/1", "batchId": None, "creatorCount": None, "playerCount": None, "freeQuotaCount": None}, "role-permissions.json": {"schema": "dogfood-role-permissions/1", "checks": []}, - "main-chain.json": {"schema": "dogfood-main-chain/1", "identity": {}, "stages": []}, + "main-chain.json": {"schema": MAIN_CHAIN_SCHEMA, "identity": {}, "stages": []}, "release-negatives.json": {"schema": "dogfood-release-negatives/1", "checks": []}, "manual-play.json": {"schema": "dogfood-manual-play/1", "releasedGameCount": 0, "games": []}, - "performance.json": {"schema": "dogfood-first-frame-performance/1", "context": {}, "samples": []}, + "performance.json": {"schema": PERFORMANCE_SCHEMA, "context": {}, "samples": []}, "telemetry.json": {"schema": "dogfood-telemetry-completeness/1", "expectedCount": None, "completeCount": None, "evidenceRef": None}, "alerts.json": {"schema": "dogfood-alert-delivery/1", "deliveries": []}, "drills.json": {"schema": "dogfood-drills/1", "drills": []}, @@ -100,15 +108,37 @@ def parse_timestamp(value: Any) -> datetime | None: return parsed if parsed.tzinfo is not None else None +def normalized_utc(value: datetime | None = None) -> datetime: + """返回秒精度 UTC 时间；测试可注入时钟，CLI 始终使用系统当前时间。""" + current = value or datetime.now(timezone.utc) + if current.tzinfo is None: + raise ValueError("时间必须包含时区") + return current.astimezone(timezone.utc).replace(microsecond=0) + + +def format_timestamp(value: datetime) -> str: + """按统一 UTC ISO 8601 格式输出证据时间。""" + return normalized_utc(value).isoformat().replace("+00:00", "Z") + + +def utc_dates(start: date, end: date) -> list[date]: + """生成包含首尾的连续 UTC 日期，供实际运行日覆盖校验。""" + if end < start: + return [] + return [start + timedelta(days=offset) for offset in range((end - start).days + 1)] + + class EvidenceGate: """对同一波次的结构化证据执行默认拒绝校验。""" - def __init__(self, wave_dir: Path, phase: str) -> None: + def __init__(self, wave_dir: Path, phase: str, now: datetime | None = None) -> None: self.wave_dir = wave_dir.resolve() self.phase = phase + self.now = normalized_utc(now) self.errors: list[tuple[str, str]] = [] self.manifest: dict[str, Any] | None = None - self.rc_artifact_sha256: str | None = None + self.started_at: datetime | None = None + self.final_freeze_path: Path | None = None def fail(self, code: str, message: str) -> None: """记录一条可追踪失败，不因首个错误隐藏其余缺口。""" @@ -173,6 +203,19 @@ class EvidenceGate: if path is not None and sha256_file(path) != expected_hash: self.fail(f"{context}.hash-mismatch", f"{context} 的 SHA-256 与附件不一致") + def checked_hash_ref(self, ref: Any, expected_hash: Any, context: str) -> Path | None: + """返回通过路径与内容哈希校验的附件，失败时保持默认拒绝。""" + path = self.checked_ref(ref, context) + if not isinstance(expected_hash, str) or not SHA256_PATTERN.fullmatch(expected_hash): + self.fail(f"{context}.sha256", f"{context} 缺少合法 SHA-256") + return None + if path is None: + return None + if sha256_file(path) != expected_hash: + self.fail(f"{context}.hash-mismatch", f"{context} 的 SHA-256 与附件不一致") + return None + return path + def read_tsv(self, name: str, fields: list[str]) -> list[dict[str, str]] | None: """用标准库 TSV 解析器校验表头，禁止文本猜字段。""" path = self.wave_dir / name @@ -191,32 +234,59 @@ class EvidenceGate: return None def validate_manifest(self) -> None: - """确认波次身份和规模没有被改写。""" - manifest = self.load_json("manifest.json", "dogfood-wave-manifest/1") + """确认波次身份、可信开始时间和前序绑定没有被改写。""" + manifest = self.load_json("manifest.json", MANIFEST_SCHEMA) if manifest is None: return wave = manifest.get("wave") if wave not in WAVE_SCALES: self.fail("manifest.wave", "manifest.wave 只能是 wave0 或 wave1") return - expected = {"schema": "dogfood-wave-manifest/1", "wave": wave, **WAVE_SCALES[wave]} - if manifest != expected or self.wave_dir.name != wave: + started_at = parse_timestamp(manifest.get("startedAt")) + if started_at is None: + self.fail("manifest.started-at", "manifest.startedAt 必须是工具生成的带时区时间") + else: + started_at = started_at.astimezone(timezone.utc) + if started_at > self.now: + self.fail("manifest.started-at", "manifest.startedAt 不得晚于当前时间") + self.started_at = started_at + expected_keys = {"schema", "wave", "creatorCount", "playerCount", "durationHours", "startedAt"} + if wave == "wave1": + expected_keys.update({"wave0CloseResultSha256", "wave0FinalFreezeSha256", "wave0ClosedAt"}) + fixed_values_match = all(manifest.get(key) == value for key, value in WAVE_SCALES[wave].items()) + if set(manifest) != expected_keys or not fixed_values_match or self.wave_dir.name != wave: self.fail("manifest.identity", "波次目录名、身份或固定规模不匹配") return self.manifest = manifest + if wave == "wave1": + self.validate_wave0_prerequisite() + + def validate_wave0_prerequisite(self) -> None: + """持续核验 Wave1 对 Wave0 收口回执和最终冻结清单的固定绑定。""" + if self.manifest is None: + return + binding, errors = inspect_wave0_close(self.wave_dir.parent, self.now) + for code, message in errors: + self.fail(code, message) + if binding is None: + return + for key in ("wave0CloseResultSha256", "wave0FinalFreezeSha256", "wave0ClosedAt"): + if self.manifest.get(key) != binding.get(key): + self.fail("wave1.wave0-binding", f"Wave1 manifest 未绑定当前 Wave0 收口字段：{key}") + closed_at = parse_timestamp(binding.get("wave0ClosedAt")) + if self.started_at is not None and closed_at is not None and self.started_at < closed_at: + self.fail("wave1.started-before-wave0-close", "Wave1 startedAt 不得早于 Wave0 closedAt") def validate_rc_and_config(self) -> None: """校验完整提交、干净构建、制品与配置身份。""" - rc = self.load_json("rc.json", "dogfood-rc/1") + rc = self.load_json("rc.json", RC_SCHEMA) config = self.load_json("config-identity.json", "dogfood-config-identity/1") if rc is not None: if not isinstance(rc.get("commit"), str) or not COMMIT_PATTERN.fullmatch(rc["commit"]): self.fail("rc.commit", "RC commit 必须是完整 40 位小写哈希") if rc.get("repoClean") is not True: self.fail("rc.clean", "RC 必须来自 tracked 工作区干净构建") - self.check_hash_ref(rc.get("artifactRef"), rc.get("artifactSha256"), "rc.artifact") - if isinstance(rc.get("artifactSha256"), str) and SHA256_PATTERN.fullmatch(rc["artifactSha256"]): - self.rc_artifact_sha256 = rc["artifactSha256"] + self.check_hash_ref(rc.get("rcArtifactRef"), rc.get("rcArtifactSha256"), "rc.artifact") if config is not None: if parse_timestamp(config.get("capturedAt")) is None: self.fail("config.captured-at", "配置身份缺少带时区采集时间") @@ -283,19 +353,31 @@ class EvidenceGate: def validate_main_chain(self) -> None: """校验同一项目、版本、产物和游戏贯穿完整主链。""" - payload = self.load_json("main-chain.json", "dogfood-main-chain/1") + payload = self.load_json("main-chain.json", MAIN_CHAIN_SCHEMA) if payload is None: return identity = payload.get("identity") - if not isinstance(identity, dict) or any(not identity.get(key) for key in ("projectId", "versionId", "artifactSha256", "gameId")): - self.fail("main-chain.identity", "主链身份缺少 project/version/artifact/game") - elif identity.get("artifactSha256") != self.rc_artifact_sha256: - self.fail("main-chain.rc-artifact", "主链产物哈希未绑定当前 RC 制品") + required_identity = ( + "projectId", "versionId", "gameId", "gameArtifactSha256", + "reviewRecordRef", "runtimePackageRef", "mainChainDatabaseRef", + ) + if not isinstance(identity, dict) or any(not identity.get(key) for key in required_identity): + self.fail("main-chain.identity", "主链身份缺少 project/version/gameArtifact/game 或三类绑定证据") + else: + self.validate_game_artifact_bindings(identity) stages = payload.get("stages") if not isinstance(stages, list): self.fail("main-chain.stages", "主链 stages 必须是数组") return by_stage = {item.get("stage"): item for item in stages if isinstance(item, dict)} + if isinstance(identity, dict): + stage_database_refs = { + item.get("databaseRef") + for item in by_stage.values() + if isinstance(item, dict) and isinstance(item.get("databaseRef"), str) + } + if identity.get("mainChainDatabaseRef") not in stage_database_refs: + self.fail("main-chain.game-artifact", "游戏数据库绑定必须是主链阶段实际使用的 databaseRef") evidence_types = { "screenshotRef": {".png", ".jpg", ".jpeg", ".webp"}, "networkRef": {".har", ".json"}, @@ -313,6 +395,59 @@ class EvidenceGate: for field, suffixes in evidence_types.items(): self.checked_typed_ref(item.get(field), f"main-chain.{stage}.{field}", suffixes) + def load_binding_json(self, ref: Any, context: str) -> dict[str, Any] | None: + """读取波次附件内的结构化绑定证据，拒绝非法路径和非对象 JSON。""" + path = self.checked_ref(ref, context) + if path is None: + return None + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + self.fail("main-chain.game-artifact", f"{context} 不是合法 JSON：{exc}") + return None + if not isinstance(payload, dict): + self.fail("main-chain.game-artifact", f"{context} 顶层必须是对象") + return None + return payload + + def validate_game_artifact_bindings(self, identity: dict[str, Any]) -> None: + """将游戏产物哈希独立绑定到审核记录、运行包和主链数据库证据。""" + game_hash = identity.get("gameArtifactSha256") + if not isinstance(game_hash, str) or not SHA256_PATTERN.fullmatch(game_hash): + self.fail("main-chain.game-artifact", "gameArtifactSha256 必须是小写 64 位 SHA-256") + return + refs = [identity.get(key) for key in ("reviewRecordRef", "runtimePackageRef", "mainChainDatabaseRef")] + if len(set(ref for ref in refs if isinstance(ref, str))) != 3: + self.fail("main-chain.game-artifact", "审核记录、运行包和数据库绑定必须使用三份独立证据") + + runtime_path = self.checked_ref(identity.get("runtimePackageRef"), "main-chain.runtime-package") + if runtime_path is not None and sha256_file(runtime_path) != game_hash: + self.fail("main-chain.game-artifact", "runtime package 实际 SHA-256 与 gameArtifactSha256 不一致") + + expected_identity = {key: identity.get(key) for key in ("projectId", "versionId", "gameId", "gameArtifactSha256")} + review = self.load_binding_json(identity.get("reviewRecordRef"), "main-chain.review-record") + if review is not None: + review_matches = all(review.get(key) == value for key, value in expected_identity.items()) + if ( + review.get("schema") != "dogfood-review-artifact-binding/1" + or not review_matches + or review.get("status") != "APPROVED" + or not review.get("reviewId") + or parse_timestamp(review.get("approvedAt")) is None + ): + self.fail("main-chain.game-artifact", "审核记录未批准或未绑定同一游戏产物身份") + + database = self.load_binding_json(identity.get("mainChainDatabaseRef"), "main-chain.database-binding") + if database is not None: + database_matches = all(database.get(key) == value for key, value in expected_identity.items()) + if ( + database.get("schema") != "dogfood-main-chain-db-binding/1" + or not database_matches + or not database.get("queryId") + or parse_timestamp(database.get("capturedAt")) is None + ): + self.fail("main-chain.game-artifact", "主链数据库证据未绑定同一游戏产物身份") + def validate_release_and_manual_play(self) -> None: """校验发布负例全部被拦截，并确保入 feed 前完成真玩。""" negatives = self.load_json("release-negatives.json", "dogfood-release-negatives/1") @@ -346,8 +481,8 @@ class EvidenceGate: self.checked_ref(item.get("screenshotRef"), "manual-play.screenshot") def validate_performance(self) -> None: - """按 nearest-rank 算法校验 30 个独立冷导航样本的 P75。""" - payload = self.load_json("performance.json", "dogfood-first-frame-performance/1") + """从逐条绑定的 /gstack 原始 CDP trace/HAR 重算冷导航 P75。""" + payload = self.load_json("performance.json", PERFORMANCE_SCHEMA) if payload is None: return expected_context = { @@ -363,6 +498,7 @@ class EvidenceGate: if len(samples) < 30: self.fail("performance.sample-count", "冷缓存独立导航样本不得少于 30") navigation_ids: list[str] = [] + raw_refs: list[str] = [] durations: list[float] = [] for sample in samples: if not isinstance(sample, dict): @@ -372,23 +508,48 @@ class EvidenceGate: self.fail("performance.cold-only", "热缓存样本不得混入冷样本") if sample.get("source") != "/gstack@mini-desktop": self.fail("performance.source", "性能证据只能来自 /gstack@mini-desktop") + collector = sample.get("collector") + if ( + not isinstance(collector, dict) + or collector.get("tool") != "/gstack" + or collector.get("host") != "mini-desktop" + or not isinstance(collector.get("sessionId"), str) + or not collector.get("sessionId") + ): + self.fail("performance.collector", "每个样本必须绑定 /gstack、mini-desktop 和非空 sessionId") navigation_id = sample.get("navigationId") if not isinstance(navigation_id, str) or not navigation_id: self.fail("performance.navigation-id", "每个样本必须有独立 navigationId") else: navigation_ids.append(navigation_id) - duration = sample.get("durationMs") + if "durationMs" in sample: + self.fail("performance.self-reported-duration", "不接受自填 durationMs，必须从原始记录重算") + collected_at = parse_timestamp(sample.get("collectedAt")) + if collected_at is None or collected_at > self.now: + self.fail("performance.collected-at", "每个样本必须包含不晚于当前时间的带时区采集时间") + elif self.started_at is not None and collected_at < self.started_at: + self.fail("performance.collected-at", "性能样本采集时间不得早于波次 startedAt") + raw_ref = sample.get("rawRef") + if isinstance(raw_ref, str): + raw_refs.append(raw_ref) + raw_path = self.checked_hash_ref(raw_ref, sample.get("rawSha256"), "performance.raw-record") + if raw_path is not None and raw_path.suffix.lower() not in {".json", ".har"}: + self.fail("performance.raw-record", "原始性能证据只能是 CDP trace JSON 或 HAR") + raw_path = None if ( - not isinstance(duration, (int, float)) - or isinstance(duration, bool) - or not math.isfinite(duration) - or duration < 0 + raw_path is not None + and isinstance(navigation_id, str) + and navigation_id + and isinstance(collector, dict) + and collected_at is not None ): - self.fail("performance.duration", "durationMs 必须是非负数") - else: - durations.append(float(duration)) + duration = self.duration_from_raw_record(raw_path, navigation_id, collector, sample.get("collectedAt")) + if duration is not None: + durations.append(duration) if len(set(navigation_ids)) != len(navigation_ids): self.fail("performance.independent", "navigationId 重复，样本不是独立导航") + if len(set(raw_refs)) != len(raw_refs): + self.fail("performance.independent", "每个冷导航样本必须绑定独立原始记录文件") if len(durations) == len(samples) and len(durations) >= 30: ordered = sorted(durations) rank = math.ceil(0.75 * len(ordered)) @@ -396,6 +557,90 @@ class EvidenceGate: if p75 >= 3000: self.fail("performance.p75", f"nearest-rank P75={p75:g}ms，必须严格小于 3000ms") + def duration_from_raw_record( + self, + path: Path, + navigation_id: str, + collector: dict[str, Any], + collected_at: Any, + ) -> float | None: + """从受哈希保护的 trace/HAR 时间戳计算导航提交到首个可交互帧的时长。""" + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + self.fail("performance.raw-record", f"原始性能证据不可读：{path.name}：{exc}") + return None + if not isinstance(payload, dict): + self.fail("performance.raw-record", f"原始性能证据顶层必须是对象：{path.name}") + return None + if path.suffix.lower() == ".har": + return self.duration_from_har(payload, navigation_id, collector, collected_at, path.name) + return self.duration_from_cdp_trace(payload, navigation_id, collector, collected_at, path.name) + + def duration_from_cdp_trace( + self, + payload: dict[str, Any], + navigation_id: str, + collector: dict[str, Any], + collected_at: Any, + filename: str, + ) -> float | None: + """读取 /gstack 标记过的 CDP trace 微秒时间戳。""" + metadata = payload.get("metadata") + events = payload.get("traceEvents") + if not isinstance(metadata, dict) or metadata.get("collector") != collector or metadata.get("collectedAt") != collected_at: + self.fail("performance.raw-record", f"CDP trace 采集器身份或采集时间不匹配：{filename}") + return None + if not isinstance(events, list): + self.fail("performance.raw-record", f"CDP trace 缺少 traceEvents：{filename}") + return None + timestamps: dict[str, list[float]] = { + "dogfood.navigationCommit": [], + "dogfood.firstInteractiveGameFrame": [], + } + for event in events: + if not isinstance(event, dict) or event.get("name") not in timestamps: + continue + args = event.get("args") + event_navigation = args.get("navigationId") if isinstance(args, dict) else None + timestamp = event.get("ts") + if event_navigation == navigation_id and isinstance(timestamp, (int, float)) and not isinstance(timestamp, bool) and math.isfinite(timestamp): + timestamps[event["name"]].append(float(timestamp)) + if any(len(values) != 1 for values in timestamps.values()): + self.fail("performance.raw-record", f"CDP trace 必须各含一个同 navigationId 的起止事件：{filename}") + return None + duration = (timestamps["dogfood.firstInteractiveGameFrame"][0] - timestamps["dogfood.navigationCommit"][0]) / 1000 + if not math.isfinite(duration) or duration < 0: + self.fail("performance.duration", f"CDP trace 起止时间非法：{filename}") + return None + return duration + + def duration_from_har( + self, + payload: dict[str, Any], + navigation_id: str, + collector: dict[str, Any], + collected_at: Any, + filename: str, + ) -> float | None: + """读取 /gstack HAR 页面扩展中的原始起止时间戳。""" + log = payload.get("log") + pages = log.get("pages") if isinstance(log, dict) else None + matches = [page for page in pages or [] if isinstance(page, dict) and page.get("id") == navigation_id] + if len(matches) != 1: + self.fail("performance.raw-record", f"HAR 必须恰有一个同 navigationId 页面：{filename}") + return None + metadata = matches[0].get("_gstack") + if not isinstance(metadata, dict) or metadata.get("collector") != collector or metadata.get("collectedAt") != collected_at: + self.fail("performance.raw-record", f"HAR 采集器身份或采集时间不匹配：{filename}") + return None + start = parse_timestamp(metadata.get("navigationCommitAt")) + end = parse_timestamp(metadata.get("firstInteractiveGameFrameAt")) + if start is None or end is None or end < start: + self.fail("performance.duration", f"HAR 起止时间非法：{filename}") + return None + return (end - start).total_seconds() * 1000 + def validate_p0_telemetry_alerts(self) -> None: """校验 P0 为零、埋点完整率和告警真送达字段。""" p0_rows = self.read_tsv("p0.tsv", ["id", "severity", "status", "summary", "evidence_ref"]) @@ -499,22 +744,37 @@ class EvidenceGate: for condition in REQUIRED_STOP_CONDITIONS: item = by_condition.get(condition) if item is None or item.get("triggered") is not False: - self.fail("safety-status.stop-condition", f"停止条件缺失或已触发：{condition}") + code = ( + f"safety-status.{condition}" + if condition in {"confirmed-false-positive", "review-backlog-over-4h"} + else "safety-status.stop-condition" + ) + self.fail(code, f"停止条件缺失或已触发：{condition}") + + def validate_elapsed_runtime(self) -> None: + """收口只认 startedAt 到当前可信时钟的真实经过时长。""" + if self.manifest is None or self.started_at is None: + return + required = timedelta(hours=self.manifest["durationHours"]) + if self.now - self.started_at < required: + elapsed_hours = max(0.0, (self.now - self.started_at).total_seconds() / 3600) + self.fail( + "wave.duration", + f"实际运行仅 {elapsed_hours:.2f} 小时，必须达到 {self.manifest['durationHours']} 小时", + ) def validate_freeze(self) -> None: - """收口时校验每日冻结哈希链，并重算最新证据。""" + """收口时校验每日冻结时间、哈希链和所有实际 UTC 运行日覆盖。""" freeze_dir = self.wave_dir / "freezes" freeze_files = sorted(freeze_dir.glob("*.sha256.json")) if freeze_dir.is_dir() else [] if not freeze_files: self.fail("freeze.missing", "收口前必须生成至少一次每日 SHA-256 冻结") return - required_count = math.ceil((self.manifest or {}).get("durationHours", 0) / 24) - if len(freeze_files) < required_count: - self.fail("freeze.daily-count", f"每日冻结不足：需要 {required_count} 份，实际 {len(freeze_files)} 份") - payloads: list[dict[str, Any]] = [] previous_file: Path | None = None previous_date: date | None = None + previous_created_at: datetime | None = None + freeze_dates: list[date] = [] for freeze_file in freeze_files: expected_date = freeze_file.name.removesuffix(".sha256.json") try: @@ -527,7 +787,7 @@ class EvidenceGate: self.fail("freeze.schema", "冻结清单 schema 或 files 非法") return if ( - payload.get("schema") != "dogfood-evidence-freeze/1" + payload.get("schema") != FREEZE_SCHEMA or payload.get("algorithm") != "SHA-256" or payload.get("wave") != (self.manifest or {}).get("wave") or payload.get("date") != expected_date @@ -535,6 +795,18 @@ class EvidenceGate: ): self.fail("freeze.schema", "冻结清单身份、日期、算法或 files 非法") return + created_at = parse_timestamp(payload.get("createdAt")) + if created_at is None: + self.fail("freeze.created-at", f"冻结清单缺少带时区 createdAt：{freeze_file.name}") + else: + created_at = created_at.astimezone(timezone.utc) + if created_at.date() != parsed_date or created_at > self.now: + self.fail("freeze.created-at", f"冻结日期必须等于真实创建 UTC 日期且不得在未来：{freeze_file.name}") + if self.started_at is not None and created_at < self.started_at: + self.fail("freeze.created-at", f"冻结不得早于波次 startedAt：{freeze_file.name}") + if previous_created_at is not None and created_at <= previous_created_at: + self.fail("freeze.created-at", "冻结 createdAt 必须随日期严格递增") + previous_created_at = created_at expected_previous = sha256_file(previous_file) if previous_file is not None else None if payload.get("previousFreezeSha256") != expected_previous: self.fail("freeze.chain", f"冻结哈希链断裂：{freeze_file.name}") @@ -543,6 +815,13 @@ class EvidenceGate: payloads.append(payload) previous_file = freeze_file previous_date = parsed_date + freeze_dates.append(parsed_date) + + if self.started_at is not None: + expected_dates = utc_dates(self.started_at.date(), self.now.date()) + if freeze_dates != expected_dates: + expected_text = ",".join(day.isoformat() for day in expected_dates) + self.fail("freeze.actual-days", f"冻结必须覆盖全部实际 UTC 运行日：{expected_text}") latest = freeze_files[-1] entries = payloads[-1]["files"] @@ -566,25 +845,93 @@ class EvidenceGate: self.fail("freeze.hash-mismatch", f"冻结后证据发生变化：{relative}") if recorded_paths != expected_paths: self.fail("freeze.file-set", "当前证据文件集合与冻结清单不一致") + self.final_freeze_path = latest + + def validate(self) -> None: + """执行全部规则但不打印或写收口回执，供跨波次复核复用。""" + if not self.wave_dir.is_dir(): + self.fail("wave-dir.missing", "波次目录不存在") + return + self.validate_manifest() + self.validate_rc_and_config() + self.validate_snapshot_and_storage() + self.validate_accounts() + self.validate_permissions() + self.validate_main_chain() + self.validate_release_and_manual_play() + self.validate_performance() + self.validate_p0_telemetry_alerts() + self.validate_oncall_feedback_checklist() + self.validate_drills_and_safety() + if self.phase == "close": + self.validate_elapsed_runtime() + self.validate_freeze() + + def write_close_result(self) -> bool: + """原子写入只由通过收口门产生的回执，并固定最终冻结清单哈希。""" + if self.manifest is None or self.started_at is None or self.final_freeze_path is None: + self.fail("close-result.internal", "缺少生成收口回执所需的已验证身份") + return False + payload = { + "schema": CLOSE_RESULT_SCHEMA, + "wave": self.manifest["wave"], + "result": "PASS", + "startedAt": format_timestamp(self.started_at), + "closedAt": format_timestamp(self.now), + "durationHours": (self.now - self.started_at).total_seconds() / 3600, + "manifestSha256": sha256_file(self.wave_dir / "manifest.json"), + "finalFreezeRef": self.final_freeze_path.relative_to(self.wave_dir).as_posix(), + "finalFreezeSha256": sha256_file(self.final_freeze_path), + "blockingSummary": { + "p0Count": 0, + "confirmedFalsePositiveCount": 0, + "reviewBacklogOver4hCount": 0, + }, + } + target = self.wave_dir / CLOSE_RESULT_NAME + if target.exists(): + if target.is_symlink(): + self.fail("close-result.invalid", "收口回执不得是符号链接") + return False + try: + existing = json.loads(target.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + self.fail("close-result.invalid", f"既有收口回执不可读：{exc}") + return False + stable_keys = {key: value for key, value in payload.items() if key not in {"closedAt", "durationHours"}} + existing_stable = {key: value for key, value in existing.items() if key not in {"closedAt", "durationHours"}} if isinstance(existing, dict) else None + existing_closed_at = parse_timestamp(existing.get("closedAt")) if isinstance(existing, dict) else None + existing_hours = existing.get("durationHours") if isinstance(existing, dict) else None + recomputed_hours = ( + (existing_closed_at - self.started_at).total_seconds() / 3600 + if existing_closed_at is not None + else -1 + ) + if ( + existing_stable != stable_keys + or existing_closed_at is None + or existing_closed_at > self.now + or not isinstance(existing_hours, (int, float)) + or isinstance(existing_hours, bool) + or not math.isclose(float(existing_hours), recomputed_hours) + ): + self.fail("close-result.stale", "既有收口回执与当前 manifest 或最终冻结不一致") + return False + return True + temporary = target.with_suffix(".json.tmp") + try: + write_json(temporary, payload) + temporary.replace(target) + except OSError as exc: + self.fail("close-result.write", f"收口回执写入失败：{exc}") + return False + return True def run(self) -> int: """执行完整闸门并输出机器可读错误代码。""" - if not self.wave_dir.is_dir(): - self.fail("wave-dir.missing", "波次目录不存在") - else: - self.validate_manifest() - self.validate_rc_and_config() - self.validate_snapshot_and_storage() - self.validate_accounts() - self.validate_permissions() - self.validate_main_chain() - self.validate_release_and_manual_play() - self.validate_performance() - self.validate_p0_telemetry_alerts() - self.validate_oncall_feedback_checklist() - self.validate_drills_and_safety() - if self.phase == "close": - self.validate_freeze() + self.validate() + if not self.errors and self.phase == "close": + self.write_close_result() if self.errors: print(f"GATE_FAIL phase={self.phase} errors={len(self.errors)}") for code, message in self.errors: @@ -594,6 +941,76 @@ class EvidenceGate: return 0 +def inspect_wave0_close(root: Path, now: datetime) -> tuple[dict[str, str] | None, list[tuple[str, str]]]: + """从固定兄弟目录复核 Wave0 收口，返回可写入 Wave1 manifest 的哈希绑定。""" + errors: list[tuple[str, str]] = [] + wave0_dir = root.resolve() / "wave0" + receipt_path = wave0_dir / CLOSE_RESULT_NAME + if wave0_dir.is_symlink() or not wave0_dir.is_dir() or receipt_path.is_symlink() or not receipt_path.is_file(): + return None, [("wave1.wave0-close-missing", "Wave1 需要同一证据根下 Wave0 的工具生成收口回执")] + try: + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + return None, [("wave1.wave0-close-invalid", f"Wave0 收口回执不可读：{exc}")] + if not isinstance(receipt, dict): + return None, [("wave1.wave0-close-invalid", "Wave0 收口回执顶层必须是对象")] + closed_at = parse_timestamp(receipt.get("closedAt")) + if ( + receipt.get("schema") != CLOSE_RESULT_SCHEMA + or receipt.get("wave") != "wave0" + or receipt.get("result") != "PASS" + or closed_at is None + or closed_at > now + or receipt.get("blockingSummary") != { + "p0Count": 0, + "confirmedFalsePositiveCount": 0, + "reviewBacklogOver4hCount": 0, + } + ): + return None, [("wave1.wave0-close-invalid", "Wave0 收口回执身份、时间或阻断汇总非法")] + manifest_path = wave0_dir / "manifest.json" + final_ref = receipt.get("finalFreezeRef") + relative = Path(final_ref) if isinstance(final_ref, str) else Path() + if relative.is_absolute() or len(relative.parts) != 2 or relative.parts[0] != "freezes": + return None, [("wave1.wave0-close-invalid", "Wave0 最终冻结引用必须位于固定 freezes/ 路径")] + final_path = wave0_dir / relative + if ( + manifest_path.is_symlink() + or not manifest_path.is_file() + or final_path.is_symlink() + or not final_path.is_file() + or receipt.get("manifestSha256") != sha256_file(manifest_path) + or receipt.get("finalFreezeSha256") != sha256_file(final_path) + ): + return None, [("wave1.wave0-close-invalid", "Wave0 manifest 或最终冻结哈希与收口回执不一致")] + + # 用回执中的 closedAt 重跑 Wave0 全门，避免 Wave1 启动时间延长历史波次的运行日区间。 + wave0_gate = EvidenceGate(wave0_dir, "close", now=closed_at) + wave0_gate.validate() + if wave0_gate.errors: + codes = {code for code, _ in wave0_gate.errors} + if "p0.nonzero" in codes: + errors.append(("wave1.wave0-p0", "Wave0 存在 P0，禁止进入 Wave1")) + if "safety-status.confirmed-false-positive" in codes: + errors.append(("wave1.wave0-false-positive", "Wave0 存在确认发布假阳，禁止进入 Wave1")) + if "safety-status.review-backlog-over-4h" in codes: + errors.append(("wave1.wave0-review-backlog", "Wave0 存在超过 4 小时待审积压，禁止进入 Wave1")) + errors.append(("wave1.wave0-close-invalid", "Wave0 当前证据已不能通过其收口门")) + return None, errors + if wave0_gate.final_freeze_path != final_path: + return None, [("wave1.wave0-close-invalid", "Wave0 回执绑定的不是当前最后冻结清单")] + started_at = wave0_gate.started_at + actual_hours = (closed_at - started_at).total_seconds() / 3600 if started_at is not None else -1 + recorded_hours = receipt.get("durationHours") + if not isinstance(recorded_hours, (int, float)) or isinstance(recorded_hours, bool) or not math.isclose(float(recorded_hours), actual_hours): + return None, [("wave1.wave0-close-invalid", "Wave0 回执持续时长无法由 startedAt/closedAt 重算")] + return { + "wave0CloseResultSha256": sha256_file(receipt_path), + "wave0FinalFreezeSha256": sha256_file(final_path), + "wave0ClosedAt": format_timestamp(closed_at), + }, [] + + def collect_evidence_files(wave_dir: Path) -> list[Path]: """收集冻结范围，排除冻结清单自身并拒绝符号链接。""" files: list[Path] = [] @@ -601,6 +1018,8 @@ def collect_evidence_files(wave_dir: Path) -> list[Path]: relative = path.relative_to(wave_dir) if relative.parts and relative.parts[0] == "freezes": continue + if relative.as_posix() == CLOSE_RESULT_NAME: + continue if path.is_symlink(): raise ValueError(f"冻结范围不允许符号链接：{relative.as_posix()}") if path.is_file(): @@ -608,14 +1027,18 @@ def collect_evidence_files(wave_dir: Path) -> list[Path]: return files -def freeze_evidence(wave_dir: Path, freeze_date: str) -> int: +def freeze_evidence(wave_dir: Path, freeze_date: str, now: datetime | None = None) -> int: """为指定日期原子写入不可覆盖的 SHA-256 文件清单。""" wave_dir = wave_dir.resolve() + current_time = normalized_utc(now) try: parsed_date = date.fromisoformat(freeze_date) except ValueError: print("FREEZE_REFUSED：--date 必须是 YYYY-MM-DD。", file=sys.stderr) return 2 + if parsed_date != current_time.date(): + print("FREEZE_REFUSED：--date 必须等于工具当前 UTC 日期，禁止未来或回填冻结。", file=sys.stderr) + return 2 manifest_path = wave_dir / "manifest.json" if not manifest_path.is_file(): print("FREEZE_REFUSED：波次目录缺少 manifest.json。", file=sys.stderr) @@ -626,6 +1049,14 @@ def freeze_evidence(wave_dir: Path, freeze_date: str) -> int: except (OSError, json.JSONDecodeError, ValueError) as exc: print(f"FREEZE_REFUSED：{exc}", file=sys.stderr) return 2 + started_at = parse_timestamp(manifest.get("startedAt")) if isinstance(manifest, dict) else None + if not isinstance(manifest, dict) or manifest.get("schema") != MANIFEST_SCHEMA or manifest.get("wave") not in WAVE_SCALES or started_at is None: + print("FREEZE_REFUSED：manifest 身份或 startedAt 非法。", file=sys.stderr) + return 2 + started_at = started_at.astimezone(timezone.utc) + if started_at > current_time or parsed_date < started_at.date(): + print("FREEZE_REFUSED：冻结不得早于 startedAt 或使用未来 startedAt。", file=sys.stderr) + return 2 freeze_dir = wave_dir / "freezes" freeze_dir.mkdir(mode=0o700, exist_ok=True) target = freeze_dir / f"{freeze_date}.sha256.json" @@ -642,20 +1073,27 @@ def freeze_evidence(wave_dir: Path, freeze_date: str) -> int: return 2 previous = earlier[-1] if earlier else None payload = { - "schema": "dogfood-evidence-freeze/1", + "schema": FREEZE_SCHEMA, "date": freeze_date, + "createdAt": format_timestamp(current_time), "wave": manifest.get("wave"), "algorithm": "SHA-256", "previousFreezeSha256": sha256_file(previous) if previous is not None else None, "files": [{"path": path.relative_to(wave_dir).as_posix(), "sha256": sha256_file(path)} for path in files], } - serialized = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n" if target.exists(): - if target.read_text(encoding="utf-8") == serialized: + try: + existing = json.loads(target.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + existing = None + comparable = {key: value for key, value in payload.items() if key != "createdAt"} + existing_comparable = {key: value for key, value in existing.items() if key != "createdAt"} if isinstance(existing, dict) else None + if existing_comparable == comparable: print(f"FREEZE_ALREADY_EXISTS path={target}") return 0 print("FREEZE_REFUSED：同日冻结已存在且当前证据不同。", file=sys.stderr) return 2 + serialized = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n" temporary = target.with_suffix(target.suffix + ".tmp") temporary.write_text(serialized, encoding="utf-8") temporary.replace(target) @@ -753,24 +1191,47 @@ def ensure_wave_scaffold(wave_dir: Path) -> list[str]: return created -def initialize_wave(root: Path, wave: str) -> int: +def initialize_wave(root: Path, wave: str, now: datetime | None = None) -> int: """初始化固定波次目录；已存在时只校验身份，不覆盖证据。""" - wave_dir = root.resolve() / wave + if wave not in WAVE_SCALES: + print("INIT_REFUSED：仅允许 wave0 或 wave1。", file=sys.stderr) + return 2 + current_time = normalized_utc(now) + root = root.resolve() + wave_dir = root / wave manifest_path = wave_dir / "manifest.json" - expected = {"schema": "dogfood-wave-manifest/1", "wave": wave, **WAVE_SCALES[wave]} if wave_dir.exists(): if not manifest_path.is_file(): print("INIT_REFUSED：既有目录缺少 manifest.json。", file=sys.stderr) return 2 try: - current = json.loads(manifest_path.read_text(encoding="utf-8")) + current_manifest = json.loads(manifest_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: print(f"INIT_REFUSED：既有 manifest 不可读：{exc}", file=sys.stderr) return 2 - if current != expected: + expected_keys = {"schema", "wave", "creatorCount", "playerCount", "durationHours", "startedAt"} + if wave == "wave1": + expected_keys.update({"wave0CloseResultSha256", "wave0FinalFreezeSha256", "wave0ClosedAt"}) + started_at = parse_timestamp(current_manifest.get("startedAt")) if isinstance(current_manifest, dict) else None + fixed_values_match = isinstance(current_manifest, dict) and all(current_manifest.get(key) == value for key, value in WAVE_SCALES[wave].items()) + if ( + not isinstance(current_manifest, dict) + or set(current_manifest) != expected_keys + or current_manifest.get("schema") != MANIFEST_SCHEMA + or current_manifest.get("wave") != wave + or started_at is None + or started_at > current_time + or not fixed_values_match + ): print("INIT_REFUSED：既有目录的波次身份或规模不匹配。", file=sys.stderr) return 2 + if wave == "wave1": + binding, errors = inspect_wave0_close(root, current_time) + if binding is None or any(current_manifest.get(key) != value for key, value in (binding or {}).items()): + detail = errors[0][1] if errors else "Wave0 收口绑定已变化" + print(f"WAVE0_CLOSE_REQUIRED：{detail}", file=sys.stderr) + return 2 repaired = ensure_wave_scaffold(wave_dir) if repaired: print(f"INIT_REPAIRED wave={wave} path={wave_dir} created={','.join(repaired)}") @@ -778,6 +1239,20 @@ def initialize_wave(root: Path, wave: str) -> int: print(f"INIT_ALREADY_EXISTS wave={wave} path={wave_dir}") return 0 + expected: dict[str, Any] = { + "schema": MANIFEST_SCHEMA, + "wave": wave, + **WAVE_SCALES[wave], + "startedAt": format_timestamp(current_time), + } + if wave == "wave1": + binding, errors = inspect_wave0_close(root, current_time) + if binding is None: + detail = errors[0][1] if errors else "Wave0 尚未通过收口门" + print(f"WAVE0_CLOSE_REQUIRED：{detail}", file=sys.stderr) + return 2 + expected.update(binding) + wave_dir.mkdir(parents=True, mode=0o700) write_json(manifest_path, expected) ensure_wave_scaffold(wave_dir) diff --git a/deploy/dogfood-ops/tests/test_dogfood_ops.py b/deploy/dogfood-ops/tests/test_dogfood_ops.py index 9be7402f..105906a5 100644 --- a/deploy/dogfood-ops/tests/test_dogfood_ops.py +++ b/deploy/dogfood-ops/tests/test_dogfood_ops.py @@ -4,6 +4,7 @@ from __future__ import annotations import hashlib +import importlib.util import json import subprocess import tempfile @@ -16,6 +17,11 @@ from typing import Any TOOL = Path(__file__).resolve().parents[1] / "dogfood_ops.py" OPS_DIR = TOOL.parent +MODULE_SPEC = importlib.util.spec_from_file_location("dogfood_ops", TOOL) +assert MODULE_SPEC is not None and MODULE_SPEC.loader is not None +DOGFOOD_OPS = importlib.util.module_from_spec(MODULE_SPEC) +MODULE_SPEC.loader.exec_module(DOGFOOD_OPS) + class DogfoodOpsTest(unittest.TestCase): """通过命令行入口验证运维人员实际使用的行为。""" @@ -55,12 +61,148 @@ class DogfoodOpsTest(unittest.TestCase): attachment.write_text(content, encoding="utf-8") return attachment.relative_to(wave_dir).as_posix(), hashlib.sha256(attachment.read_bytes()).hexdigest() + def close_valid_wave0(self) -> tuple[Path, datetime]: + """按两个实际 UTC 运行日构造已通过收口门的 Wave0。""" + started_at = datetime(2026, 7, 20, 9, 0, tzinfo=timezone.utc) + self.assertEqual(DOGFOOD_OPS.initialize_wave(self.root, "wave0", now=started_at), 0) + wave_dir = self.root / "wave0" + self.populate_valid_evidence(wave_dir) + self.assertEqual( + DOGFOOD_OPS.freeze_evidence( + wave_dir, "2026-07-20", now=datetime(2026, 7, 20, 23, 0, tzinfo=timezone.utc) + ), + 0, + ) + closed_at = datetime(2026, 7, 21, 9, 1, tzinfo=timezone.utc) + self.assertEqual(DOGFOOD_OPS.freeze_evidence(wave_dir, "2026-07-21", now=closed_at), 0) + gate = DOGFOOD_OPS.EvidenceGate(wave_dir, "close", now=closed_at) + self.assertEqual(gate.run(), 0, gate.errors) + return wave_dir, closed_at + + def set_trace_duration(self, wave_dir: Path, sample: dict[str, Any], duration_ms: float) -> None: + """修改测试 trace 的结束时间并同步样本哈希。""" + raw_path = wave_dir / sample["rawRef"] + payload = json.loads(raw_path.read_text(encoding="utf-8")) + payload["traceEvents"][1]["ts"] = payload["traceEvents"][0]["ts"] + duration_ms * 1000 + self.write_json(raw_path, payload) + sample["rawSha256"] = hashlib.sha256(raw_path.read_bytes()).hexdigest() + + def test_wave1_init_requires_closed_wave0(self) -> None: + """没有 Wave0 收口回执时不得初始化 Wave1。""" + result = self.run_tool("init", "--root", str(self.root), "--wave", "wave1") + + self.assertNotEqual(result.returncode, 0) + self.assertIn("WAVE0_CLOSE_REQUIRED", result.stderr) + + def test_manifest_and_freeze_use_tool_generated_trusted_times(self) -> None: + """开始与冻结时间必须由工具生成，并拒绝未来或回填日期。""" + started_at = datetime(2026, 7, 20, 9, 0, tzinfo=timezone.utc) + result = DOGFOOD_OPS.initialize_wave(self.root, "wave0", now=started_at) + self.assertEqual(result, 0) + wave_dir = self.root / "wave0" + manifest = json.loads((wave_dir / "manifest.json").read_text(encoding="utf-8")) + self.assertEqual(manifest["startedAt"], "2026-07-20T09:00:00Z") + + backfill = DOGFOOD_OPS.freeze_evidence(wave_dir, "2026-07-19", now=started_at) + future = DOGFOOD_OPS.freeze_evidence(wave_dir, "2026-07-21", now=started_at) + current = DOGFOOD_OPS.freeze_evidence(wave_dir, "2026-07-20", now=started_at) + + self.assertNotEqual(backfill, 0) + self.assertNotEqual(future, 0) + self.assertEqual(current, 0) + freeze = json.loads((wave_dir / "freezes/2026-07-20.sha256.json").read_text(encoding="utf-8")) + self.assertEqual(freeze["createdAt"], "2026-07-20T09:00:00Z") + + def test_close_requires_elapsed_runtime_and_every_actual_utc_day(self) -> None: + """收口必须达到真实时长，并覆盖从 startedAt 到 close 的每个 UTC 运行日。""" + started_at = datetime(2026, 7, 20, 9, 0, tzinfo=timezone.utc) + DOGFOOD_OPS.initialize_wave(self.root, "wave0", now=started_at) + wave_dir = self.root / "wave0" + self.populate_valid_evidence(wave_dir) + DOGFOOD_OPS.freeze_evidence(wave_dir, "2026-07-20", now=started_at.replace(hour=23)) + + too_early = DOGFOOD_OPS.EvidenceGate( + wave_dir, "close", now=started_at + timedelta(hours=23, minutes=59) + ) + self.assertNotEqual(too_early.run(), 0) + self.assertTrue(any(code == "wave.duration" for code, _ in too_early.errors)) + + close_time = started_at + timedelta(hours=24, minutes=1) + missing_day = DOGFOOD_OPS.EvidenceGate(wave_dir, "close", now=close_time) + self.assertNotEqual(missing_day.run(), 0) + self.assertTrue(any(code == "freeze.actual-days" for code, _ in missing_day.errors)) + + def test_same_minute_cannot_forge_three_daily_freezes(self) -> None: + """同一时刻只能冻结当天，不能用三个日期伪造 Wave1 三天运行。""" + now = datetime(2026, 7, 23, 5, 30, tzinfo=timezone.utc) + DOGFOOD_OPS.initialize_wave(self.root, "wave0", now=now) + wave_dir = self.root / "wave0" + + results = [ + DOGFOOD_OPS.freeze_evidence(wave_dir, day, now=now) + for day in ("2026-07-21", "2026-07-22", "2026-07-23") + ] + + self.assertEqual(results, [2, 2, 0]) + + def test_close_rejects_forged_freeze_created_at(self) -> None: + """手工把 createdAt 改成未来或与文件日期不一致时不得收口。""" + started_at = datetime(2026, 7, 20, 9, 0, tzinfo=timezone.utc) + DOGFOOD_OPS.initialize_wave(self.root, "wave0", now=started_at) + wave_dir = self.root / "wave0" + self.populate_valid_evidence(wave_dir) + freeze_now = datetime(2026, 7, 20, 23, 0, tzinfo=timezone.utc) + self.assertEqual(DOGFOOD_OPS.freeze_evidence(wave_dir, "2026-07-20", now=freeze_now), 0) + freeze_path = wave_dir / "freezes/2026-07-20.sha256.json" + freeze = json.loads(freeze_path.read_text(encoding="utf-8")) + freeze["createdAt"] = "2026-07-21T23:00:00Z" + self.write_json(freeze_path, freeze) + + gate = DOGFOOD_OPS.EvidenceGate(wave_dir, "close", now=started_at + timedelta(hours=24)) + + self.assertNotEqual(gate.run(), 0) + self.assertTrue(any(code == "freeze.created-at" for code, _ in gate.errors)) + + def test_performance_rejects_self_reported_duration_without_raw_gstack_record(self) -> None: + """自填 durationMs 且未绑定原始 /gstack 记录时必须失败。""" + wave_dir = self.init_wave("wave0") + self.populate_valid_evidence(wave_dir) + path = wave_dir / "performance.json" + payload = json.loads(path.read_text(encoding="utf-8")) + for sample in payload["samples"]: + sample["durationMs"] = 2000 + sample.pop("rawRef") + sample.pop("rawSha256") + self.write_json(path, payload) + + result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") + + self.assertNotEqual(result.returncode, 0) + self.assertIn("performance.raw-record", result.stdout) + + def test_rc_and_game_artifact_hashes_are_independent_and_game_is_bound_three_ways(self) -> None: + """RC 制品可不同于游戏包，游戏哈希必须同时绑定审核、运行包和数据库证据。""" + wave_dir = self.init_wave("wave0") + self.populate_valid_evidence(wave_dir) + rc = json.loads((wave_dir / "rc.json").read_text(encoding="utf-8")) + chain = json.loads((wave_dir / "main-chain.json").read_text(encoding="utf-8")) + self.assertNotEqual(rc["rcArtifactSha256"], chain["identity"]["gameArtifactSha256"]) + + result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") + + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertNotIn("main-chain.rc-artifact", result.stdout) + self.assertNotIn("main-chain.game-artifact", result.stdout) + def populate_valid_evidence(self, wave_dir: Path) -> None: """构造不依赖 staging 的完整本地通过 fixture。""" manifest = json.loads((wave_dir / "manifest.json").read_text(encoding="utf-8")) wave = manifest["wave"] + started_at = datetime.fromisoformat(manifest["startedAt"].replace("Z", "+00:00")) + captured_at = started_at.isoformat().replace("+00:00", "Z") commit = "1" * 40 - artifact_ref, artifact_hash = self.add_attachment(wave_dir, "rc-artifact.bin") + rc_artifact_ref, rc_artifact_hash = self.add_attachment(wave_dir, "rc-artifact.bin", "rc-package") + runtime_ref, game_artifact_hash = self.add_attachment(wave_dir, "game-runtime.zip", "game-package") config_ref, config_hash = self.add_attachment(wave_dir, "config-bundle.txt") dump_ref, dump_hash = self.add_attachment(wave_dir, "mysql.sql.gz") restore_ref, _ = self.add_attachment(wave_dir, "restore-report.json") @@ -68,11 +210,32 @@ class DogfoodOpsTest(unittest.TestCase): common_ref, _ = self.add_attachment(wave_dir, "browser-proof.png") network_ref, _ = self.add_attachment(wave_dir, "main-chain.har") log_ref, _ = self.add_attachment(wave_dir, "application.log") + review_ref, _ = self.add_attachment(wave_dir, "review-binding.json") database_ref, _ = self.add_attachment(wave_dir, "database.json") + game_identity = { + "projectId": "p-1", + "versionId": "v-1", + "gameId": "g-1", + "gameArtifactSha256": game_artifact_hash, + } + self.write_json(wave_dir / review_ref, { + "schema": "dogfood-review-artifact-binding/1", + **game_identity, + "reviewId": "review-1", + "status": "APPROVED", + "approvedAt": captured_at, + }) + self.write_json(wave_dir / database_ref, { + "schema": "dogfood-main-chain-db-binding/1", + **game_identity, + "queryId": "query-1", + "capturedAt": captured_at, + }) + self.write_json(wave_dir / "rc.json", { - "schema": "dogfood-rc/1", "commit": commit, "repoClean": True, - "artifactRef": artifact_ref, "artifactSha256": artifact_hash, + "schema": "dogfood-rc/2", "commit": commit, "repoClean": True, + "rcArtifactRef": rc_artifact_ref, "rcArtifactSha256": rc_artifact_hash, }) self.write_json(wave_dir / "config-identity.json", { "schema": "dogfood-config-identity/1", "capturedAt": "2026-07-22T00:00:00Z", @@ -108,8 +271,13 @@ class DogfoodOpsTest(unittest.TestCase): evidence_bundle = {"screenshotRef": common_ref, "networkRef": network_ref, "logRef": log_ref, "databaseRef": database_ref} stages = ["create", "generate", "preview", "submit-review", "approve", "feed", "play", "telemetry"] self.write_json(wave_dir / "main-chain.json", { - "schema": "dogfood-main-chain/1", - "identity": {"projectId": "p-1", "versionId": "v-1", "artifactSha256": artifact_hash, "gameId": "g-1"}, + "schema": "dogfood-main-chain/2", + "identity": { + **game_identity, + "reviewRecordRef": review_ref, + "runtimePackageRef": runtime_ref, + "mainChainDatabaseRef": database_ref, + }, "stages": [{"stage": stage, "result": "PASS", **evidence_bundle} for stage in stages], }) @@ -126,12 +294,30 @@ class DogfoodOpsTest(unittest.TestCase): "games": [{"gameId": "g-1", "tester": "tester-1", "playedAt": "2026-07-22T02:00:00Z", "feedEnteredAt": "2026-07-22T02:10:00Z", "screenshotRef": common_ref, "source": "/gstack@mini-desktop"}], }) - samples = [ - {"navigationId": f"nav-{index:02d}", "cache": "cold", "durationMs": 2000, "source": "/gstack@mini-desktop"} - for index in range(30) - ] + samples = [] + for index in range(30): + navigation_id = f"nav-{index:02d}" + collector = {"tool": "/gstack", "host": "mini-desktop", "sessionId": f"session-{index:02d}"} + raw_ref, raw_hash = self.add_attachment(wave_dir, f"perf-{index:02d}.trace.json") + self.write_json(wave_dir / raw_ref, { + "metadata": {"collector": collector, "collectedAt": captured_at}, + "traceEvents": [ + {"name": "dogfood.navigationCommit", "ts": 1_000_000, "args": {"navigationId": navigation_id}}, + {"name": "dogfood.firstInteractiveGameFrame", "ts": 3_000_000, "args": {"navigationId": navigation_id}}, + ], + }) + raw_hash = hashlib.sha256((wave_dir / raw_ref).read_bytes()).hexdigest() + samples.append({ + "navigationId": navigation_id, + "cache": "cold", + "source": "/gstack@mini-desktop", + "rawRef": raw_ref, + "rawSha256": raw_hash, + "collectedAt": captured_at, + "collector": collector, + }) self.write_json(wave_dir / "performance.json", { - "schema": "dogfood-first-frame-performance/1", + "schema": "dogfood-first-frame-performance/2", "context": {"host": "mini-desktop", "browser": "Chrome", "browserVersion": "146", "viewport": "390x844@1", "network": "Tailscale staging", "timing": "navigation-commit-to-first-interactive-game-frame"}, "samples": samples, }) @@ -144,7 +330,7 @@ class DogfoodOpsTest(unittest.TestCase): "deliveries": [{"alertId": "alert-1", "receivedAt": "2026-07-22T03:00:00Z", "recipient": "oncall-1", "externalMessageId": "msg-1", "evidenceRef": common_ref}], }) - start = datetime(2026, 7, 22, tzinfo=timezone.utc) + start = started_at end = start + timedelta(hours=manifest["durationHours"]) (wave_dir / "oncall.tsv").write_text( "start_at\tend_at\tprimary\tbackup\tcontact\n" @@ -202,10 +388,76 @@ class DogfoodOpsTest(unittest.TestCase): def test_init_wave1_uses_approved_scale(self) -> None: """wave1 必须固定为累计 10 名创作者和 30 名玩家。""" - wave_dir = self.init_wave("wave1") + wave0_dir, closed_at = self.close_valid_wave0() + self.assertTrue((wave0_dir / "close-result.json").is_file()) + self.assertEqual(DOGFOOD_OPS.initialize_wave(self.root, "wave1", now=closed_at + timedelta(minutes=1)), 0) + wave_dir = self.root / "wave1" manifest = json.loads((wave_dir / "manifest.json").read_text(encoding="utf-8")) self.assertEqual((manifest["creatorCount"], manifest["playerCount"]), (10, 30)) + self.assertEqual(manifest["wave0CloseResultSha256"], hashlib.sha256((wave0_dir / "close-result.json").read_bytes()).hexdigest()) + final_freeze = wave0_dir / "freezes/2026-07-21.sha256.json" + self.assertEqual(manifest["wave0FinalFreezeSha256"], hashlib.sha256(final_freeze.read_bytes()).hexdigest()) + + def test_wave1_revalidates_bound_wave0_close_and_final_freeze(self) -> None: + """Wave1 开波门必须持续复核 Wave0 回执和最终冻结哈希。""" + wave0_dir, closed_at = self.close_valid_wave0() + wave1_started_at = closed_at + timedelta(minutes=1) + self.assertEqual(DOGFOOD_OPS.initialize_wave(self.root, "wave1", now=wave1_started_at), 0) + wave1_dir = self.root / "wave1" + self.populate_valid_evidence(wave1_dir) + (wave0_dir / "close-result.json").write_text( + (wave0_dir / "close-result.json").read_text(encoding="utf-8") + " ", + encoding="utf-8", + ) + + gate = DOGFOOD_OPS.EvidenceGate(wave1_dir, "readiness", now=wave1_started_at) + + self.assertNotEqual(gate.run(), 0) + self.assertTrue(any(code == "wave1.wave0-binding" for code, _ in gate.errors)) + + def test_wave1_init_rejects_wave0_with_p0(self) -> None: + """Wave0 收口后出现 P0 时不得初始化 Wave1。""" + wave0_dir, closed_at = self.close_valid_wave0() + (wave0_dir / "p0.tsv").write_text( + "id\tseverity\tstatus\tsummary\tevidence_ref\n" + "P0-1\tP0\tOPEN\tsecurity\tattachments/browser-proof.png\n", + encoding="utf-8", + ) + + result = DOGFOOD_OPS.initialize_wave(self.root, "wave1", now=closed_at + timedelta(minutes=1)) + + self.assertNotEqual(result, 0) + _, errors = DOGFOOD_OPS.inspect_wave0_close(self.root, closed_at + timedelta(minutes=1)) + self.assertTrue(any(code == "wave1.wave0-p0" for code, _ in errors)) + + def test_wave1_init_rejects_wave0_with_false_positive(self) -> None: + """Wave0 存在确认发布假阳时不得初始化 Wave1。""" + wave0_dir, closed_at = self.close_valid_wave0() + path = wave0_dir / "safety-status.json" + payload = json.loads(path.read_text(encoding="utf-8")) + next(item for item in payload["conditions"] if item["condition"] == "confirmed-false-positive")["triggered"] = True + self.write_json(path, payload) + + result = DOGFOOD_OPS.initialize_wave(self.root, "wave1", now=closed_at + timedelta(minutes=1)) + + self.assertNotEqual(result, 0) + _, errors = DOGFOOD_OPS.inspect_wave0_close(self.root, closed_at + timedelta(minutes=1)) + self.assertTrue(any(code == "wave1.wave0-false-positive" for code, _ in errors)) + + def test_wave1_init_rejects_wave0_with_review_backlog(self) -> None: + """Wave0 存在超过四小时待审积压时不得初始化 Wave1。""" + wave0_dir, closed_at = self.close_valid_wave0() + path = wave0_dir / "safety-status.json" + payload = json.loads(path.read_text(encoding="utf-8")) + next(item for item in payload["conditions"] if item["condition"] == "review-backlog-over-4h")["triggered"] = True + self.write_json(path, payload) + + result = DOGFOOD_OPS.initialize_wave(self.root, "wave1", now=closed_at + timedelta(minutes=1)) + + self.assertNotEqual(result, 0) + _, errors = DOGFOOD_OPS.inspect_wave0_close(self.root, closed_at + timedelta(minutes=1)) + self.assertTrue(any(code == "wave1.wave0-review-backlog" for code, _ in errors)) def test_repeated_init_is_idempotent_and_does_not_overwrite(self) -> None: """重复初始化应成功返回，但不得覆盖已经填写的证据。""" @@ -285,7 +537,7 @@ class DogfoodOpsTest(unittest.TestCase): path = wave_dir / "performance.json" payload = json.loads(path.read_text(encoding="utf-8")) for index, sample in enumerate(payload["samples"]): - sample["durationMs"] = 1000 if index < 22 else 3000 + self.set_trace_duration(wave_dir, sample, 1000 if index < 22 else 3000) self.write_json(path, payload) result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") @@ -300,7 +552,7 @@ class DogfoodOpsTest(unittest.TestCase): path = wave_dir / "performance.json" payload = json.loads(path.read_text(encoding="utf-8")) for index, sample in enumerate(payload["samples"]): - sample["durationMs"] = 1000 if index < 22 else 2999 + self.set_trace_duration(wave_dir, sample, 1000 if index < 22 else 2999) self.write_json(path, payload) result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") @@ -321,20 +573,97 @@ class DogfoodOpsTest(unittest.TestCase): self.assertNotEqual(result.returncode, 0) self.assertIn("performance.cold-only", result.stdout) - def test_performance_rejects_non_finite_duration(self) -> None: - """JSON NaN 不得绕过非负数和 P75 阈值。""" + def test_performance_rejects_non_finite_raw_trace_timestamp(self) -> None: + """原始 trace 的 NaN 时间戳不得绕过 P75 阈值。""" wave_dir = self.init_wave("wave0") self.populate_valid_evidence(wave_dir) path = wave_dir / "performance.json" payload = json.loads(path.read_text(encoding="utf-8")) for sample in payload["samples"]: - sample["durationMs"] = float("nan") + self.set_trace_duration(wave_dir, sample, float("nan")) self.write_json(path, payload) result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") self.assertNotEqual(result.returncode, 0) - self.assertIn("performance.duration", result.stdout) + self.assertIn("performance.raw-record", result.stdout) + + def test_performance_rejects_tampered_raw_record_hash(self) -> None: + """原始 trace 内容变化但样本哈希未同步时必须失败。""" + wave_dir = self.init_wave("wave0") + self.populate_valid_evidence(wave_dir) + performance = json.loads((wave_dir / "performance.json").read_text(encoding="utf-8")) + raw_path = wave_dir / performance["samples"][0]["rawRef"] + raw_path.write_text(raw_path.read_text(encoding="utf-8") + " ", encoding="utf-8") + + result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") + + self.assertNotEqual(result.returncode, 0) + self.assertIn("performance.raw-record.hash-mismatch", result.stdout) + + def test_performance_recomputes_har_timestamps(self) -> None: + """HAR 样本也必须从原始起止时间重算，不读取 durationMs。""" + wave_dir = self.init_wave("wave0") + self.populate_valid_evidence(wave_dir) + path = wave_dir / "performance.json" + payload = json.loads(path.read_text(encoding="utf-8")) + sample = payload["samples"][0] + old_path = wave_dir / sample["rawRef"] + old_path.unlink() + har_ref = "attachments/perf-00.har" + har_path = wave_dir / har_ref + self.write_json(har_path, { + "log": { + "pages": [{ + "id": sample["navigationId"], + "_gstack": { + "collector": sample["collector"], + "collectedAt": sample["collectedAt"], + "navigationCommitAt": "2026-07-23T00:00:00Z", + "firstInteractiveGameFrameAt": "2026-07-23T00:00:02.999Z", + }, + }], + }, + }) + sample["rawRef"] = har_ref + sample["rawSha256"] = hashlib.sha256(har_path.read_bytes()).hexdigest() + self.write_json(path, payload) + + result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") + + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_performance_rejects_collector_identity_mismatch_with_raw_record(self) -> None: + """样本声明的采集器身份必须与原始记录内身份完全一致。""" + wave_dir = self.init_wave("wave0") + self.populate_valid_evidence(wave_dir) + path = wave_dir / "performance.json" + payload = json.loads(path.read_text(encoding="utf-8")) + payload["samples"][0]["collector"]["sessionId"] = "forged-session" + self.write_json(path, payload) + + result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") + + self.assertNotEqual(result.returncode, 0) + self.assertIn("performance.raw-record", result.stdout) + + def test_performance_rejects_missing_or_future_collection_time(self) -> None: + """每条原始记录都必须绑定合法且不在未来的采集时间。""" + for collected_at in (None, "2999-01-01T00:00:00Z"): + with self.subTest(collected_at=collected_at): + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) / "evidence" + self.assertEqual(DOGFOOD_OPS.initialize_wave(root, "wave0"), 0) + wave_dir = root / "wave0" + self.populate_valid_evidence(wave_dir) + path = wave_dir / "performance.json" + payload = json.loads(path.read_text(encoding="utf-8")) + payload["samples"][0]["collectedAt"] = collected_at + self.write_json(path, payload) + + gate = DOGFOOD_OPS.EvidenceGate(wave_dir, "readiness") + self.assertNotEqual(gate.run(), 0) + self.assertTrue(any(code == "performance.collected-at" for code, _ in gate.errors)) def test_main_chain_requires_distinct_evidence_modalities(self) -> None: """同一个占位附件不得同时冒充截图、网络、日志和数据库证据。""" @@ -392,89 +721,135 @@ class DogfoodOpsTest(unittest.TestCase): self.assertNotEqual(result.returncode, 0) self.assertIn("reference.outside-wave", result.stdout) - def test_main_chain_artifact_must_match_the_rc_artifact(self) -> None: - """主链不能用另一份产物哈希冒充当前 RC 的验收结果。""" + def test_main_chain_game_artifact_must_match_runtime_package(self) -> None: + """游戏产物哈希必须与实际 runtime package 内容一致。""" wave_dir = self.init_wave("wave0") self.populate_valid_evidence(wave_dir) path = wave_dir / "main-chain.json" payload = json.loads(path.read_text(encoding="utf-8")) - payload["identity"]["artifactSha256"] = "f" * 64 + payload["identity"]["gameArtifactSha256"] = "f" * 64 self.write_json(path, payload) result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") self.assertNotEqual(result.returncode, 0) - self.assertIn("main-chain.rc-artifact", result.stdout) + self.assertIn("main-chain.game-artifact", result.stdout) + + def test_main_chain_game_artifact_must_match_review_and_database_records(self) -> None: + """审核记录或主链数据库任一身份漂移都必须拒绝。""" + for ref_key in ("reviewRecordRef", "mainChainDatabaseRef"): + with self.subTest(ref_key=ref_key): + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) / "evidence" + self.assertEqual(DOGFOOD_OPS.initialize_wave(root, "wave0"), 0) + wave_dir = root / "wave0" + self.populate_valid_evidence(wave_dir) + chain = json.loads((wave_dir / "main-chain.json").read_text(encoding="utf-8")) + binding_path = wave_dir / chain["identity"][ref_key] + binding = json.loads(binding_path.read_text(encoding="utf-8")) + binding["gameArtifactSha256"] = "f" * 64 + self.write_json(binding_path, binding) + + gate = DOGFOOD_OPS.EvidenceGate(wave_dir, "readiness") + self.assertNotEqual(gate.run(), 0) + self.assertTrue(any(code == "main-chain.game-artifact" for code, _ in gate.errors)) + + def test_game_database_binding_must_be_used_by_main_chain_stage(self) -> None: + """独立伪造的数据库绑定不能替代主链阶段实际采集的数据库证据。""" + wave_dir = self.init_wave("wave0") + self.populate_valid_evidence(wave_dir) + chain_path = wave_dir / "main-chain.json" + chain = json.loads(chain_path.read_text(encoding="utf-8")) + original = wave_dir / chain["identity"]["mainChainDatabaseRef"] + detached_ref = "attachments/detached-database-binding.json" + (wave_dir / detached_ref).write_bytes(original.read_bytes()) + chain["identity"]["mainChainDatabaseRef"] = detached_ref + self.write_json(chain_path, chain) + + result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") + + self.assertNotEqual(result.returncode, 0) + self.assertIn("main-chain.game-artifact", result.stdout) def test_close_requires_freeze_and_detects_post_freeze_tampering(self) -> None: """收口门必须验证最近一次 SHA-256 冻结，冻结后改动立即失效。""" - wave_dir = self.init_wave("wave0") + started_at = datetime(2026, 7, 20, 9, 0, tzinfo=timezone.utc) + DOGFOOD_OPS.initialize_wave(self.root, "wave0", now=started_at) + wave_dir = self.root / "wave0" self.populate_valid_evidence(wave_dir) + closed_at = started_at + timedelta(hours=24, minutes=1) - missing = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close") - self.assertNotEqual(missing.returncode, 0) - self.assertIn("freeze.missing", missing.stdout) + missing = DOGFOOD_OPS.EvidenceGate(wave_dir, "close", now=closed_at) + self.assertNotEqual(missing.run(), 0) + self.assertTrue(any(code == "freeze.missing" for code, _ in missing.errors)) - freeze = self.run_tool("freeze", "--wave-dir", str(wave_dir), "--date", "2026-07-22") - self.assertEqual(freeze.returncode, 0, freeze.stdout + freeze.stderr) - passed = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close") - self.assertEqual(passed.returncode, 0, passed.stdout + passed.stderr) + self.assertEqual(DOGFOOD_OPS.freeze_evidence(wave_dir, "2026-07-20", now=started_at.replace(hour=23)), 0) + self.assertEqual(DOGFOOD_OPS.freeze_evidence(wave_dir, "2026-07-21", now=closed_at), 0) + passed = DOGFOOD_OPS.EvidenceGate(wave_dir, "close", now=closed_at) + self.assertEqual(passed.run(), 0, passed.errors) + self.assertTrue((wave_dir / "close-result.json").is_file()) with (wave_dir / "feedback.tsv").open("a", encoding="utf-8") as stream: stream.write("F-2\t2026-07-22T08:00:00Z\tu-2\tplayer\tbug\tP1\ttampered\tOPEN\tops\tattachments/browser-proof.png\n") - tampered = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close") - self.assertNotEqual(tampered.returncode, 0) - self.assertIn("freeze.hash-mismatch", tampered.stdout) + tampered = DOGFOOD_OPS.EvidenceGate(wave_dir, "close", now=closed_at) + self.assertNotEqual(tampered.run(), 0) + self.assertTrue(any(code == "freeze.hash-mismatch" for code, _ in tampered.errors)) def test_close_rejects_malformed_freeze_without_traceback(self) -> None: """畸形冻结清单应形成可追踪失败，而不是让闸门异常退出。""" - wave_dir = self.init_wave("wave0") + started_at = datetime(2026, 7, 20, 9, 0, tzinfo=timezone.utc) + DOGFOOD_OPS.initialize_wave(self.root, "wave0", now=started_at) + wave_dir = self.root / "wave0" self.populate_valid_evidence(wave_dir) freeze_dir = wave_dir / "freezes" - (freeze_dir / "2026-07-22.sha256.json").write_text("[]\n", encoding="utf-8") + (freeze_dir / "2026-07-20.sha256.json").write_text("[]\n", encoding="utf-8") - result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close") + gate = DOGFOOD_OPS.EvidenceGate(wave_dir, "close", now=started_at + timedelta(hours=24)) + result = gate.run() - self.assertEqual(result.returncode, 1, result.stdout + result.stderr) - self.assertIn("freeze.schema", result.stdout) - self.assertNotIn("Traceback", result.stderr) + self.assertEqual(result, 1) + self.assertTrue(any(code == "freeze.schema" for code, _ in gate.errors)) def test_close_rejects_symlink_added_after_freeze_without_traceback(self) -> None: """冻结后加入符号链接时应 fail-closed，并保持稳定退出码。""" - wave_dir = self.init_wave("wave0") + started_at = datetime(2026, 7, 20, 9, 0, tzinfo=timezone.utc) + DOGFOOD_OPS.initialize_wave(self.root, "wave0", now=started_at) + wave_dir = self.root / "wave0" self.populate_valid_evidence(wave_dir) - freeze = self.run_tool("freeze", "--wave-dir", str(wave_dir), "--date", "2026-07-22") - self.assertEqual(freeze.returncode, 0, freeze.stderr) + self.assertEqual(DOGFOOD_OPS.freeze_evidence(wave_dir, "2026-07-20", now=started_at.replace(hour=23)), 0) (wave_dir / "attachments" / "late-link").symlink_to(wave_dir / "attachments" / "browser-proof.png") - result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close") + gate = DOGFOOD_OPS.EvidenceGate(wave_dir, "close", now=started_at + timedelta(hours=24)) + result = gate.run() - self.assertEqual(result.returncode, 1, result.stdout + result.stderr) - self.assertIn("freeze.collection", result.stdout) - self.assertNotIn("Traceback", result.stderr) + self.assertEqual(result, 1) + self.assertTrue(any(code == "freeze.collection" for code, _ in gate.errors)) def test_wave1_close_requires_three_chained_daily_freezes(self) -> None: - """三天 wave1 不得用单日快照代替每日冻结，且历史清单必须受哈希链保护。""" - wave_dir = self.init_wave("wave1") + """Wave1 必须真实运行 72 小时并覆盖四个实际 UTC 日期，且哈希链不可篡改。""" + _, wave0_closed_at = self.close_valid_wave0() + started_at = wave0_closed_at + timedelta(minutes=1) + self.assertEqual(DOGFOOD_OPS.initialize_wave(self.root, "wave1", now=started_at), 0) + wave_dir = self.root / "wave1" self.populate_valid_evidence(wave_dir) - first = self.run_tool("freeze", "--wave-dir", str(wave_dir), "--date", "2026-07-22") - self.assertEqual(first.returncode, 0, first.stderr) + close_time = started_at + timedelta(hours=72, minutes=1) + for offset in range(4): + freeze_now = datetime.combine( + started_at.date() + timedelta(days=offset), + datetime.min.time(), + tzinfo=timezone.utc, + ) + timedelta(hours=23) + if freeze_now > close_time: + freeze_now = close_time + self.assertEqual(DOGFOOD_OPS.freeze_evidence(wave_dir, freeze_now.date().isoformat(), now=freeze_now), 0) + passed = DOGFOOD_OPS.EvidenceGate(wave_dir, "close", now=close_time) + self.assertEqual(passed.run(), 0, passed.errors) - too_few = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close") - self.assertNotEqual(too_few.returncode, 0) - self.assertIn("freeze.daily-count", too_few.stdout) - - for day in ("2026-07-23", "2026-07-24"): - frozen = self.run_tool("freeze", "--wave-dir", str(wave_dir), "--date", day) - self.assertEqual(frozen.returncode, 0, frozen.stderr) - passed = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close") - self.assertEqual(passed.returncode, 0, passed.stdout + passed.stderr) - - first_path = wave_dir / "freezes" / "2026-07-22.sha256.json" + first_path = sorted((wave_dir / "freezes").glob("*.sha256.json"))[0] first_path.write_text(first_path.read_text(encoding="utf-8") + " ", encoding="utf-8") - tampered = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close") - self.assertNotEqual(tampered.returncode, 0) - self.assertIn("freeze.chain", tampered.stdout) + tampered = DOGFOOD_OPS.EvidenceGate(wave_dir, "close", now=close_time) + self.assertNotEqual(tampered.run(), 0) + self.assertTrue(any(code == "freeze.chain" for code, _ in tampered.errors)) def test_record_drill_requires_exact_ack_and_is_idempotent(self) -> None: """演练记录必须使用精确 ACK，重复签收同一证据不得制造重复行。""" @@ -500,17 +875,14 @@ class DogfoodOpsTest(unittest.TestCase): def test_freeze_is_idempotent_but_refuses_same_day_rewrite(self) -> None: """同一证据可重复冻结，内容变化后不得覆盖同日清单。""" - wave_dir = self.init_wave("wave0") - first = self.run_tool("freeze", "--wave-dir", str(wave_dir), "--date", "2026-07-22") - self.assertEqual(first.returncode, 0, first.stderr) - repeated = self.run_tool("freeze", "--wave-dir", str(wave_dir), "--date", "2026-07-22") - self.assertEqual(repeated.returncode, 0, repeated.stderr) - self.assertIn("FREEZE_ALREADY_EXISTS", repeated.stdout) + now = datetime(2026, 7, 22, 8, 0, tzinfo=timezone.utc) + DOGFOOD_OPS.initialize_wave(self.root, "wave0", now=now) + wave_dir = self.root / "wave0" + self.assertEqual(DOGFOOD_OPS.freeze_evidence(wave_dir, "2026-07-22", now=now), 0) + self.assertEqual(DOGFOOD_OPS.freeze_evidence(wave_dir, "2026-07-22", now=now + timedelta(minutes=1)), 0) (wave_dir / "feedback.tsv").write_text("changed\n", encoding="utf-8") - refused = self.run_tool("freeze", "--wave-dir", str(wave_dir), "--date", "2026-07-22") - self.assertNotEqual(refused.returncode, 0) - self.assertIn("FREEZE_REFUSED", refused.stderr) + self.assertNotEqual(DOGFOOD_OPS.freeze_evidence(wave_dir, "2026-07-22", now=now + timedelta(minutes=2)), 0) def test_shell_entrypoints_delegate_to_the_single_cli(self) -> None: """值班入口应直接调用同一套实现，避免脚本间口径漂移。"""