zizi-al/games-development-ai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(dogfood-ops): bind waves to durable runtime evidence

Browse Source
This commit is contained in:
lili 2026-07-22 23:06:57 -07:00
parent 34da791f26
commit 410d35dc64
3 changed files with 999 additions and 147 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
deploy/dogfood-ops/README.md
View File

@ -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 冻结清单。

597
deploy/dogfood-ops/dogfood_ops.py
View File

@ -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)

518
deploy/dogfood-ops/tests/test_dogfood_ops.py
View File

@ -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:
"""值班入口应直接调用同一套实现，避免脚本间口径漂移。"""