diff --git a/deploy/dogfood-ops/README.md b/deploy/dogfood-ops/README.md new file mode 100644 index 00000000..b0f4bae8 --- /dev/null +++ b/deploy/dogfood-ops/README.md @@ -0,0 +1,144 @@ +# 内部狗粮运行保障 + +本目录把 Wave0/Wave1 的开波和收口条件落实为默认拒绝的本地证据门。工具只读取本地 JSON、TSV 和附件并重算结论，不连接 staging，不发告警，不执行停服、删档或回滚，也不包含 Playwright/Puppeteer。浏览器采样、真玩和截图只能由 mini-desktop 上的项目统一入口 `/gstack` 产生。 + +## 1. 固定波次 + +| 波次 | 创作者 | 玩家 | 连续值班覆盖 | +|---|---:|---:|---:| +| wave0 | 3 | 5 | 至少 24 小时 | +| wave1 | 10 | 30 | 至少 72 小时 | + +初始化只接受这两个名称。重复初始化会校验既有 `manifest.json` 后返回，不覆盖任何已填写证据。 + +```bash +deploy/dogfood-ops/init-wave.sh --root /受控证据根 --wave wave0 +``` + +生成目录中的 JSON 带 `_template: true`，表示它还不是证据。填写完成后删除该字段。附件统一放在波次目录内的 `attachments/`；JSON/TSV 只写相对路径。绝对路径、`..` 路径、符号链接和不存在的附件都会被拒绝。 + +## 2. 执行顺序 + +开波前按以下顺序执行，任一步非零都保持发布冻结： + +```bash +# 1. 初始化并填写同一波次的全部结构化证据。 +deploy/dogfood-ops/init-wave.sh --root /受控证据根 --wave wave0 + +# 2. 先跑开波门；模板、缺字段、阈值不达标都返回非零。 +deploy/dogfood-ops/check-readiness.sh --wave-dir /受控证据根/wave0 + +# 3. 每日证据采集结束后冻结。当日清单已存在且内容变化时拒绝覆盖。 +deploy/dogfood-ops/freeze-evidence.sh \ + --wave-dir /受控证据根/wave0 \ + --date 2026-07-22 + +# 4. 收口门除全部开波条件外，还会重算最近冻结清单。 +deploy/dogfood-ops/check-close.sh --wave-dir /受控证据根/wave0 +``` + +`GATE_PASS` 只表示本地证据在当前规则下可对账，不替代人工判断，不代表 staging 已部署、告警已送达或波次已完成。真实结论必须回到原始截图、网络记录、日志和数据库 ID 核验。 + +## 3. 证据清单 + +| 文件 | 必填内容 | +|---|---| +| `manifest.json` | 固定波次、人数和持续时间，不得修改 | +| `rc.json` | 完整 40 位 commit、tracked 干净标志、制品路径与实际 SHA-256 | +| `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；每段同时带截图、网络、日志、数据库证据 | +| `release-negatives.json` | DRAFT、REVIEWING、REJECTED、直调绕过、批准后产物漂移、下架后 feed/详情/深链八类拒绝证据 | +| `manual-play.json` | 每个发布游戏在进入 feed 前由 `/gstack@mini-desktop` 真玩，带测试人、时间和截图 | +| `performance.json` | 固定环境下至少 30 次禁用缓存的独立导航，只含 cold 样本 | +| `p0.tsv` | 单一 P0 账本；任一 `severity=P0` 行都会拒绝开波和收口 | +| `telemetry.json` | 期望完整事件数、完整收到数、原始证据；完整率至少 95% | +| `alerts.json` | 至少一条接收时间、接收人、外部消息 id 和原始证据，不接受仅写 PASS | +| `oncall.tsv` | 无空档覆盖本波次完整时长的主备值班与联系方式 | +| `feedback.tsv` | 唯一反馈账本；每行 id、时间、报告人、角色、分类、级别、摘要、状态、owner、证据都完整 | +| `manual-acceptance.tsv` | 角色权限、主链、发布负例、入 feed 前真玩四项人工签认 | +| `drills.json` | stop/delete/rollback 三类演练的精确 ACK、操作人、时间和目录内证据 | +| `safety-status.json` | 固定停止条件逐项评估，任一缺失或触发即拒绝 | +| `freezes/*.sha256.json` | 当日全部证据和附件的 SHA-256 清单；wave0 至少 1 份、wave1 至少 3 份，按日连续并以前序清单哈希串联 | + +所有时间使用带时区 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`。 + +算法使用 nearest-rank：将 `n` 个 `durationMs` 升序排列，排名为 `ceil(0.75 × n)`，按 1 起算取值。30 个样本取第 23 个值。门槛是严格 `P75 < 3000ms`，所以 2999ms 通过，3000ms 失败；少于 30 个样本直接失败。 + +## 5. 停止条件 + +出现以下任一情况，立即停止扩容；前六项同时要求全局停止： + +1. 安全事件、跨账号访问、数据丢失、真实扣款、真实广告曝光、未审内容进入 feed。 +2. 生成成功率低于 80%。 +3. 出现任一确认假阳。 +4. 同一阻断根因影响两人。 +5. 埋点缺失超过 5%。 +6. 待审超过 4 小时。 +7. 验收成本突破生成 v3 上限。 + +`safety-status.json` 必须逐项使用固定 condition 名称评估。不要删除已触发项来让闸门变绿。 + +## 6. 回滚顺序 + +真实操作只由批准的部署窗口人工执行。本目录不提供环境命令，固定顺序不可调整： + +1. 后端关闭新账号与提交审核。 +2. 后端 feed 总闸切为 fail-closed。 +3. 生成验收从 v3 退回 `v3_shadow`。 +4. 下架受影响游戏。 +5. 回退到上一 commit-pinned RC。 +6. 保留全部证据，创建或关联根因工单。 + +首个 RC 没有上一版 commit-pinned RC 时，第 5 步不可伪造为已演练；保持 rollback 红灯，等第二个 RC 窗口补证。 + +## 7. 停服、删档和回滚演练记录 + +先由获批人员在隔离目标完成真实演练并保存证据，再用本工具签收。精确 ACK 只证明操作者明确选择了对应波次和动作，不会触发环境操作。 + +```bash +deploy/dogfood-ops/record-drill.sh \ + --wave-dir /受控证据根/wave0 \ + --action stop \ + --ack ACK-DOGFOOD-wave0-stop \ + --operator 值班人ID \ + --performed-at 2026-07-22T08:00:00Z \ + --evidence-ref attachments/stop-drill-report.json +``` + +`delete` 和 `rollback` 分别使用 `ACK-DOGFOOD-wave0-delete`、`ACK-DOGFOOD-wave0-rollback`。同一动作、同一记录重复执行幂等返回；已有不同记录时拒绝覆盖。 + +删档范围只能是当前 `batchId` 关联的账号、项目、版本、feed 和遥测业务数据。审核记录、安全日志、发布证据、根因工单至少保留 180 天，不得随业务删档删除。删档报告必须列出实际删除对象和保留对象，不能只写总数。 + +## 8. 每日冻结 + +每天完成采集后运行 `freeze-evidence.sh`。清单覆盖该波次目录下除 `freezes/` 自身外的全部普通文件，并记录上一日清单哈希。工具拒绝符号链接、日期空档和同日覆盖；wave0 收口至少需要 1 份，wave1 至少需要连续 3 份。若当天冻结后又补证，保留原清单并在下一日期重新冻结，不要修改历史清单。 + +## 9. 真实采集待办 + +本地测试 fixture 不可用于开波。真实执行仍需从批准窗口采集： + +- commit-pinned RC、制品和配置快照的原始文件及哈希； +- MySQL 本轮表快照、无网络隔离恢复报告、对象存储前缀审计； +- wave0/wave1 精确账号与 FREE 额度盘点； +- `/gstack` 在 mini-desktop 产生的角色权限、完整主链、八类发布负例、逐游戏入 feed 前真玩截图与网络证据； +- 30 次独立冷导航原始计时和单列的热缓存报告； +- P0、埋点完整率、告警接收回执、值班覆盖和反馈账本； +- 停服、限定 batchId 删档、第二个 RC 回滚的人工演练证据； +- 每个采集日的 SHA-256 冻结清单。 + +## 10. 本地验证 + +```bash +python3 -m unittest discover -s deploy/dogfood-ops/tests -v +find deploy/dogfood-ops -name '*.sh' -exec bash -n {} + +python3 -m py_compile deploy/dogfood-ops/dogfood_ops.py deploy/dogfood-ops/tests/test_dogfood_ops.py +git diff --check -- deploy/dogfood-ops +``` diff --git a/deploy/dogfood-ops/check-close.sh b/deploy/dogfood-ops/check-close.sh new file mode 100755 index 00000000..fe0d4c48 --- /dev/null +++ b/deploy/dogfood-ops/check-close.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +# 执行收口证据门，并重算最近一次每日冻结清单。 +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd -P)" +exec python3 "$SCRIPT_DIR/dogfood_ops.py" check --phase close "$@" diff --git a/deploy/dogfood-ops/check-readiness.sh b/deploy/dogfood-ops/check-readiness.sh new file mode 100755 index 00000000..d9db0a67 --- /dev/null +++ b/deploy/dogfood-ops/check-readiness.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +# 执行开波证据门；任一证据缺失、失配或越界即返回非零。 +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd -P)" +exec python3 "$SCRIPT_DIR/dogfood_ops.py" check --phase readiness "$@" diff --git a/deploy/dogfood-ops/dogfood_ops.py b/deploy/dogfood-ops/dogfood_ops.py new file mode 100755 index 00000000..c3f1fc41 --- /dev/null +++ b/deploy/dogfood-ops/dogfood_ops.py @@ -0,0 +1,826 @@ +#!/usr/bin/env python3 +"""内部狗粮波次证据初始化与校验入口。""" + +from __future__ import annotations + +import argparse +import csv +import hashlib +import json +import math +import re +import sys +from datetime import date, datetime +from pathlib import Path +from typing import Any + + +WAVE_SCALES = { + "wave0": {"creatorCount": 3, "playerCount": 5, "durationHours": 24}, + "wave1": {"creatorCount": 10, "playerCount": 30, "durationHours": 72}, +} + +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", + "p0.tsv": "id\tseverity\tstatus\tsummary\tevidence_ref\n", + "manual-acceptance.tsv": "item\tresult\toperator\tchecked_at\tevidence_ref\n", +} + +JSON_TEMPLATES: dict[str, dict[str, Any]] = { + "rc.json": {"schema": "dogfood-rc/1", "commit": None, "repoClean": None, "artifactRef": None, "artifactSha256": 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": []}, + "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": []}, + "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": []}, + "safety-status.json": {"schema": "dogfood-safety-status/1", "assessedAt": None, "assessor": None, "conditions": []}, +} + +REQUIRED_PERMISSION_CASES = { + "creator-own-project": "ALLOW", + "creator-cross-account-denied": "DENY", + "reviewer-review-only": "ALLOW", + "reviewer-config-write-denied": "DENY", + "readonly-write-denied": "DENY", +} +REQUIRED_CHAIN_STAGES = { + "create", "generate", "preview", "submit-review", "approve", "feed", "play", "telemetry", +} +REQUIRED_NEGATIVE_CASES = { + "draft-publish", "reviewing-publish", "rejected-publish", "direct-publish-bypass", + "artifact-drift", "delisted-feed", "delisted-detail", "delisted-deep-link", +} +REQUIRED_CHECKLIST_ITEMS = { + "role-permissions", "main-chain", "release-negatives", "manual-play-before-feed", +} +REQUIRED_DRILLS = {"stop", "delete", "rollback"} +REQUIRED_STOP_CONDITIONS = { + "security-incident", "cross-account-access", "data-loss", "real-charge", "real-ad-impression", + "unreviewed-content-in-feed", "generation-success-below-80", "confirmed-false-positive", + "same-root-cause-blocked-two-users", "telemetry-missing-above-5", "review-backlog-over-4h", + "acceptance-cost-exceeded", +} +SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") +COMMIT_PATTERN = re.compile(r"^[0-9a-f]{40}$") + + +def write_json(path: Path, value: Any) -> None: + """以稳定格式写入 JSON，便于冻结哈希和人工审阅。""" + path.write_text( + json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + +def sha256_file(path: Path) -> str: + """流式计算文件 SHA-256，避免大证据附件一次性进入内存。""" + digest = hashlib.sha256() + with path.open("rb") as stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def parse_timestamp(value: Any) -> datetime | None: + """解析带时区 ISO 8601 时间；无时区时间不作为可信证据。""" + if not isinstance(value, str) or not value: + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + return parsed if parsed.tzinfo is not None else None + + +class EvidenceGate: + """对同一波次的结构化证据执行默认拒绝校验。""" + + def __init__(self, wave_dir: Path, phase: str) -> None: + self.wave_dir = wave_dir.resolve() + self.phase = phase + self.errors: list[tuple[str, str]] = [] + self.manifest: dict[str, Any] | None = None + self.rc_artifact_sha256: str | None = None + + def fail(self, code: str, message: str) -> None: + """记录一条可追踪失败，不因首个错误隐藏其余缺口。""" + self.errors.append((code, message)) + + def load_json(self, name: str, schema: str) -> dict[str, Any] | None: + """读取固定 JSON 文件并校验对象类型与 schema。""" + path = self.wave_dir / name + if not path.is_file(): + self.fail(f"{name.removesuffix('.json')}.missing", f"缺少 {name}") + return None + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + self.fail(f"{name.removesuffix('.json')}.invalid-json", f"{name} 不是合法 JSON：{exc}") + return None + if not isinstance(value, dict): + self.fail(f"{name.removesuffix('.json')}.invalid-type", f"{name} 顶层必须是对象") + return None + if value.get("schema") != schema: + self.fail(f"{name.removesuffix('.json')}.schema", f"{name} schema 必须为 {schema}") + if value.get("_template") is True: + self.fail(f"{name.removesuffix('.json')}.missing", f"{name} 仍是未填写模板") + return None + return value + + def checked_ref(self, value: Any, context: str, require_attachment: bool = True) -> Path | None: + """只接受波次目录内现存普通文件，拒绝绝对路径、穿越和符号链接。""" + if not isinstance(value, str) or not value or Path(value).is_absolute(): + self.fail("reference.outside-wave", f"{context} 必须是波次目录内相对路径") + return None + relative = Path(value) + if require_attachment and (not relative.parts or relative.parts[0] != "attachments"): + self.fail("reference.outside-wave", f"{context} 必须位于 attachments/ 内：{value}") + return None + candidate = self.wave_dir / value + try: + resolved = candidate.resolve(strict=True) + resolved.relative_to(self.wave_dir) + except (FileNotFoundError, OSError, ValueError): + self.fail("reference.outside-wave", f"{context} 不存在或越过波次目录：{value}") + return None + if candidate.is_symlink() or not resolved.is_file(): + self.fail("reference.outside-wave", f"{context} 必须是目录内普通文件：{value}") + return None + return resolved + + def checked_typed_ref(self, value: Any, context: str, suffixes: set[str]) -> Path | None: + """在可信附件边界上继续校验证据文件类型。""" + path = self.checked_ref(value, context) + if path is not None and path.suffix.lower() not in suffixes: + self.fail("main-chain.evidence-type", f"{context} 文件类型不匹配：{value}") + return None + return path + + def check_hash_ref(self, ref: Any, expected_hash: Any, context: str) -> 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 + if path is not None and sha256_file(path) != expected_hash: + self.fail(f"{context}.hash-mismatch", f"{context} 的 SHA-256 与附件不一致") + + def read_tsv(self, name: str, fields: list[str]) -> list[dict[str, str]] | None: + """用标准库 TSV 解析器校验表头，禁止文本猜字段。""" + path = self.wave_dir / name + if not path.is_file(): + self.fail(f"{name.removesuffix('.tsv')}.missing", f"缺少 {name}") + return None + try: + with path.open("r", encoding="utf-8", newline="") as stream: + reader = csv.DictReader(stream, delimiter="\t") + if reader.fieldnames != fields: + self.fail(f"{name.removesuffix('.tsv')}.header", f"{name} 表头不匹配") + return None + return list(reader) + except (OSError, csv.Error) as exc: + self.fail(f"{name.removesuffix('.tsv')}.parse", f"{name} 解析失败：{exc}") + return None + + def validate_manifest(self) -> None: + """确认波次身份和规模没有被改写。""" + manifest = self.load_json("manifest.json", "dogfood-wave-manifest/1") + 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: + self.fail("manifest.identity", "波次目录名、身份或固定规模不匹配") + return + self.manifest = manifest + + def validate_rc_and_config(self) -> None: + """校验完整提交、干净构建、制品与配置身份。""" + rc = self.load_json("rc.json", "dogfood-rc/1") + 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"] + if config is not None: + if parse_timestamp(config.get("capturedAt")) is None: + self.fail("config.captured-at", "配置身份缺少带时区采集时间") + self.check_hash_ref(config.get("configBundleRef"), config.get("configSha256"), "config.bundle") + if rc is not None and config.get("rcCommit") != rc.get("commit"): + self.fail("config.rc-commit", "配置身份未绑定同一个 RC commit") + + def validate_snapshot_and_storage(self) -> None: + """校验 MySQL 快照、隔离恢复和对象存储审计的原子证据。""" + snapshot = self.load_json("snapshot.json", "dogfood-snapshot/1") + if snapshot is not None: + mysql = snapshot.get("mysql") + if not isinstance(mysql, dict): + self.fail("snapshot.mysql", "缺少 MySQL 快照对象") + else: + self.check_hash_ref(mysql.get("snapshotRef"), mysql.get("sha256"), "snapshot.mysql") + if parse_timestamp(mysql.get("capturedAt")) is None: + self.fail("snapshot.mysql", "MySQL 快照缺少带时区采集时间") + restore = snapshot.get("isolatedRestore") + if not isinstance(restore, dict): + self.fail("snapshot.isolated-restore", "缺少隔离恢复对象") + else: + target = str(restore.get("target", "")).lower() + if restore.get("result") != "PASS" or not target or any(word in target for word in ("staging", "prod", "production")): + self.fail("snapshot.isolated-restore", "隔离恢复必须 PASS 且目标不得是 staging/production") + if parse_timestamp(restore.get("completedAt")) is None: + self.fail("snapshot.isolated-restore", "隔离恢复缺少带时区完成时间") + self.checked_ref(restore.get("evidenceRef"), "snapshot.isolated-restore") + + storage = self.load_json("storage-audit.json", "dogfood-storage-audit/1") + if storage is not None: + if storage.get("result") != "PASS" or parse_timestamp(storage.get("capturedAt")) is None: + self.fail("storage-audit.result", "对象存储审计必须含真实 PASS 与采集时间") + self.check_hash_ref(storage.get("reportRef"), storage.get("reportSha256"), "storage-audit.report") + + def validate_accounts(self) -> None: + """校验波次账号规模、批次身份和额度余量。""" + accounts = self.load_json("accounts.json", "dogfood-accounts/1") + if accounts is None or self.manifest is None: + return + for key in ("creatorCount", "playerCount"): + if accounts.get(key) != self.manifest[key]: + self.fail(f"accounts.{key}", f"{key} 不符合批准波次规模") + required_quota = math.ceil(self.manifest["creatorCount"] * 1.2) + if not isinstance(accounts.get("freeQuotaCount"), int) or accounts["freeQuotaCount"] < required_quota: + self.fail("accounts.free-quota", f"FREE 额度必须至少为 {required_quota}") + if not isinstance(accounts.get("batchId"), str) or not accounts["batchId"]: + self.fail("accounts.batch-id", "账号必须绑定非空 batchId") + + def validate_permissions(self) -> None: + """校验角色权限用例和截图清单。""" + payload = self.load_json("role-permissions.json", "dogfood-role-permissions/1") + checks = payload.get("checks") if payload else None + if not isinstance(checks, list): + self.fail("role-permissions.checks", "角色权限 checks 必须是数组") + return + by_case = {item.get("case"): item for item in checks if isinstance(item, dict)} + for case, expected in REQUIRED_PERMISSION_CASES.items(): + item = by_case.get(case) + if item is None or item.get("result") != expected: + self.fail("role-permissions.case", f"角色权限缺少或结果错误：{case}") + continue + self.checked_ref(item.get("screenshotRef"), f"role-permissions.{case}.screenshot") + + def validate_main_chain(self) -> None: + """校验同一项目、版本、产物和游戏贯穿完整主链。""" + payload = self.load_json("main-chain.json", "dogfood-main-chain/1") + 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 制品") + 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)} + evidence_types = { + "screenshotRef": {".png", ".jpg", ".jpeg", ".webp"}, + "networkRef": {".har", ".json"}, + "logRef": {".log", ".txt"}, + "databaseRef": {".json", ".tsv"}, + } + for stage in REQUIRED_CHAIN_STAGES: + item = by_stage.get(stage) + if item is None or item.get("result") != "PASS": + self.fail("main-chain.stage", f"主链缺少真实通过阶段：{stage}") + continue + refs = [item.get(field) for field in evidence_types] + if len(set(ref for ref in refs if isinstance(ref, str))) != len(evidence_types): + self.fail("main-chain.evidence-modalities", f"主链四类证据路径必须互异：{stage}") + for field, suffixes in evidence_types.items(): + self.checked_typed_ref(item.get(field), f"main-chain.{stage}.{field}", suffixes) + + def validate_release_and_manual_play(self) -> None: + """校验发布负例全部被拦截，并确保入 feed 前完成真玩。""" + negatives = self.load_json("release-negatives.json", "dogfood-release-negatives/1") + checks = negatives.get("checks") if negatives else None + if not isinstance(checks, list): + self.fail("release-negatives.checks", "发布负例 checks 必须是数组") + else: + by_case = {item.get("case"): item for item in checks if isinstance(item, dict)} + for case in REQUIRED_NEGATIVE_CASES: + item = by_case.get(case) + if item is None or item.get("blocked") is not True: + self.fail("release-negatives.case", f"发布负例未证实被拦截：{case}") + continue + self.checked_ref(item.get("evidenceRef"), f"release-negatives.{case}") + + manual = self.load_json("manual-play.json", "dogfood-manual-play/1") + games = manual.get("games") if manual else None + if not isinstance(games, list) or not games or manual.get("releasedGameCount") != len(games): + self.fail("manual-play.coverage", "每个发布游戏都必须有人工真玩记录") + return + for item in games: + if not isinstance(item, dict): + self.fail("manual-play.item", "人工真玩记录必须是对象") + continue + played_at = parse_timestamp(item.get("playedAt")) + feed_at = parse_timestamp(item.get("feedEnteredAt")) + if not item.get("gameId") or not item.get("tester") or played_at is None or feed_at is None or played_at >= feed_at: + self.fail("manual-play.before-feed", "人工真玩必须发生在进入 feed 前") + if item.get("source") != "/gstack@mini-desktop": + self.fail("manual-play.source", "浏览器真玩证据只能来自 /gstack@mini-desktop") + 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") + if payload is None: + return + expected_context = { + "host": "mini-desktop", "browser": "Chrome", "browserVersion": "146", "viewport": "390x844@1", + "network": "Tailscale staging", "timing": "navigation-commit-to-first-interactive-game-frame", + } + if payload.get("context") != expected_context: + self.fail("performance.context", "首屏采样环境或计时边界不匹配") + samples = payload.get("samples") + if not isinstance(samples, list): + self.fail("performance.samples", "performance.samples 必须是数组") + return + if len(samples) < 30: + self.fail("performance.sample-count", "冷缓存独立导航样本不得少于 30") + navigation_ids: list[str] = [] + durations: list[float] = [] + for sample in samples: + if not isinstance(sample, dict): + self.fail("performance.sample", "性能样本必须是对象") + continue + if sample.get("cache") != "cold": + self.fail("performance.cold-only", "热缓存样本不得混入冷样本") + if sample.get("source") != "/gstack@mini-desktop": + self.fail("performance.source", "性能证据只能来自 /gstack@mini-desktop") + 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 ( + not isinstance(duration, (int, float)) + or isinstance(duration, bool) + or not math.isfinite(duration) + or duration < 0 + ): + self.fail("performance.duration", "durationMs 必须是非负数") + else: + durations.append(float(duration)) + if len(set(navigation_ids)) != len(navigation_ids): + self.fail("performance.independent", "navigationId 重复，样本不是独立导航") + if len(durations) == len(samples) and len(durations) >= 30: + ordered = sorted(durations) + rank = math.ceil(0.75 * len(ordered)) + p75 = ordered[rank - 1] + if p75 >= 3000: + self.fail("performance.p75", f"nearest-rank P75={p75:g}ms，必须严格小于 3000ms") + + def validate_p0_telemetry_alerts(self) -> None: + """校验 P0 为零、埋点完整率和告警真送达字段。""" + p0_rows = self.read_tsv("p0.tsv", ["id", "severity", "status", "summary", "evidence_ref"]) + if p0_rows is not None: + for row in p0_rows: + if any(not row.get(field) for field in row): + self.fail("p0.row", "P0 账本存在字段不完整的行") + if row.get("severity", "").upper() == "P0": + self.fail("p0.nonzero", "本波次出现 P0，不能开波或收口") + + telemetry = self.load_json("telemetry.json", "dogfood-telemetry-completeness/1") + if telemetry is not None: + expected = telemetry.get("expectedCount") + complete = telemetry.get("completeCount") + if not isinstance(expected, int) or isinstance(expected, bool) or expected <= 0 or not isinstance(complete, int) or isinstance(complete, bool) or complete < 0 or complete > expected: + self.fail("telemetry.counts", "埋点计数必须是有效非负整数且 complete<=expected") + elif complete * 100 < expected * 95: + self.fail("telemetry.completeness", f"埋点完整率 {complete}/{expected} 低于 95%") + self.checked_ref(telemetry.get("evidenceRef"), "telemetry.evidence") + + alerts = self.load_json("alerts.json", "dogfood-alert-delivery/1") + deliveries = alerts.get("deliveries") if alerts else None + if not isinstance(deliveries, list) or not deliveries: + self.fail("alerts.deliveries", "至少需要一条告警真实送达记录") + else: + for item in deliveries: + if not isinstance(item, dict) or parse_timestamp(item.get("receivedAt")) is None or not item.get("recipient") or not item.get("externalMessageId"): + self.fail("alerts.delivery", "告警缺少接收时间、接收人或外部消息 id") + continue + self.checked_ref(item.get("evidenceRef"), "alerts.evidence") + + def validate_oncall_feedback_checklist(self) -> None: + """校验连续值班覆盖、单一反馈账本和人工验收清单。""" + oncall = self.read_tsv("oncall.tsv", ["start_at", "end_at", "primary", "backup", "contact"]) + shifts: list[tuple[datetime, datetime]] = [] + if oncall is not None: + for row in oncall: + start = parse_timestamp(row.get("start_at")) + end = parse_timestamp(row.get("end_at")) + if start is None or end is None or start >= end or not row.get("primary") or not row.get("backup") or not row.get("contact"): + self.fail("oncall.row", "值班记录字段不完整或时间无效") + else: + shifts.append((start, end)) + if shifts and self.manifest is not None: + shifts.sort() + cursor = shifts[0][1] + for start, end in shifts[1:]: + if start > cursor: + self.fail("oncall.gap", "值班覆盖存在空档") + cursor = max(cursor, end) + covered_hours = (cursor - shifts[0][0]).total_seconds() / 3600 + if covered_hours < self.manifest["durationHours"]: + self.fail("oncall.duration", "值班覆盖不足批准波次持续时间") + elif oncall is not None: + self.fail("oncall.empty", "值班表不得为空") + + feedback = self.read_tsv("feedback.tsv", ["id", "created_at", "reporter_id", "role", "category", "severity", "summary", "status", "owner", "evidence_ref"]) + if feedback is not None: + seen_ids: set[str] = set() + for row in feedback: + if any(not row.get(field) for field in row) or parse_timestamp(row.get("created_at")) is None: + self.fail("feedback.row", "反馈账本存在字段不完整的行") + continue + if row["id"] in seen_ids: + self.fail("feedback.duplicate", f"反馈 id 重复：{row['id']}") + seen_ids.add(row["id"]) + self.checked_ref(row.get("evidence_ref"), f"feedback.{row.get('id', 'unknown')}.evidence") + + checklist = self.read_tsv("manual-acceptance.tsv", ["item", "result", "operator", "checked_at", "evidence_ref"]) + if checklist is not None: + by_item = {row.get("item"): row for row in checklist} + for item_name in REQUIRED_CHECKLIST_ITEMS: + row = by_item.get(item_name) + if row is None or row.get("result") != "PASS" or not row.get("operator") or parse_timestamp(row.get("checked_at")) is None: + self.fail("manual-acceptance.item", f"人工验收项缺失或未通过：{item_name}") + continue + self.checked_ref(row.get("evidence_ref"), f"manual-acceptance.{item_name}.evidence") + + def validate_drills_and_safety(self) -> None: + """校验三类演练 ACK 及固定停止条件均未触发。""" + drills = self.load_json("drills.json", "dogfood-drills/1") + records = drills.get("drills") if drills else None + if not isinstance(records, list): + self.fail("drills.records", "drills 必须是数组") + elif self.manifest is not None: + by_action = {item.get("action"): item for item in records if isinstance(item, dict)} + for action in REQUIRED_DRILLS: + item = by_action.get(action) + expected_ack = f"ACK-DOGFOOD-{self.manifest['wave']}-{action}" + if item is None or item.get("result") != "PASS" or item.get("ack") != expected_ack or not item.get("operator") or parse_timestamp(item.get("performedAt")) is None: + self.fail("drills.action", f"演练缺失、未通过或 ACK 错误：{action}") + continue + self.checked_ref(item.get("evidenceRef"), f"drills.{action}.evidence") + + safety = self.load_json("safety-status.json", "dogfood-safety-status/1") + conditions = safety.get("conditions") if safety else None + if not isinstance(conditions, list) or parse_timestamp(safety.get("assessedAt") if safety else None) is None or not (safety or {}).get("assessor"): + self.fail("safety-status.assessment", "安全状态缺少评估人、时间或条件数组") + return + by_condition = {item.get("condition"): item for item in conditions if isinstance(item, dict)} + 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}") + + def validate_freeze(self) -> None: + """收口时校验每日冻结哈希链，并重算最新证据。""" + 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 + for freeze_file in freeze_files: + expected_date = freeze_file.name.removesuffix(".sha256.json") + try: + parsed_date = date.fromisoformat(expected_date) + payload = json.loads(freeze_file.read_text(encoding="utf-8")) + except (OSError, ValueError, json.JSONDecodeError) as exc: + self.fail("freeze.invalid", f"冻结清单不可读：{freeze_file.name}：{exc}") + return + if not isinstance(payload, dict): + self.fail("freeze.schema", "冻结清单 schema 或 files 非法") + return + if ( + payload.get("schema") != "dogfood-evidence-freeze/1" + or payload.get("algorithm") != "SHA-256" + or payload.get("wave") != (self.manifest or {}).get("wave") + or payload.get("date") != expected_date + or not isinstance(payload.get("files"), list) + ): + self.fail("freeze.schema", "冻结清单身份、日期、算法或 files 非法") + return + 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}") + if previous_date is not None and (parsed_date - previous_date).days != 1: + self.fail("freeze.daily-sequence", "每日冻结日期必须连续") + payloads.append(payload) + previous_file = freeze_file + previous_date = parsed_date + + latest = freeze_files[-1] + entries = payloads[-1]["files"] + try: + expected_paths = {path.relative_to(self.wave_dir).as_posix() for path in collect_evidence_files(self.wave_dir)} + except (OSError, ValueError) as exc: + self.fail("freeze.collection", f"当前证据集合不可冻结：{exc}") + return + recorded_paths: set[str] = set() + for entry in entries: + if not isinstance(entry, dict): + self.fail("freeze.entry", "冻结条目必须是对象") + continue + relative = entry.get("path") + if isinstance(relative, str) and relative in recorded_paths: + self.fail("freeze.duplicate", f"冻结路径重复：{relative}") + if isinstance(relative, str): + recorded_paths.add(relative) + path = self.checked_ref(relative, "freeze.entry", require_attachment=False) + if path is not None and entry.get("sha256") != sha256_file(path): + self.fail("freeze.hash-mismatch", f"冻结后证据发生变化：{relative}") + if recorded_paths != expected_paths: + self.fail("freeze.file-set", "当前证据文件集合与冻结清单不一致") + + 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() + if self.errors: + print(f"GATE_FAIL phase={self.phase} errors={len(self.errors)}") + for code, message in self.errors: + print(f"[{code}] {message}") + return 1 + print(f"GATE_PASS phase={self.phase} wave={self.manifest['wave']}") + return 0 + + +def collect_evidence_files(wave_dir: Path) -> list[Path]: + """收集冻结范围，排除冻结清单自身并拒绝符号链接。""" + files: list[Path] = [] + for path in sorted(wave_dir.rglob("*")): + relative = path.relative_to(wave_dir) + if relative.parts and relative.parts[0] == "freezes": + continue + if path.is_symlink(): + raise ValueError(f"冻结范围不允许符号链接：{relative.as_posix()}") + if path.is_file(): + files.append(path) + return files + + +def freeze_evidence(wave_dir: Path, freeze_date: str) -> int: + """为指定日期原子写入不可覆盖的 SHA-256 文件清单。""" + wave_dir = wave_dir.resolve() + try: + parsed_date = date.fromisoformat(freeze_date) + except ValueError: + print("FREEZE_REFUSED：--date 必须是 YYYY-MM-DD。", 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) + return 2 + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + files = collect_evidence_files(wave_dir) + except (OSError, json.JSONDecodeError, ValueError) as exc: + print(f"FREEZE_REFUSED：{exc}", 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" + earlier = sorted(path for path in freeze_dir.glob("*.sha256.json") if path != target) + if earlier: + latest_date_text = earlier[-1].name.removesuffix(".sha256.json") + try: + latest_date = date.fromisoformat(latest_date_text) + except ValueError: + print("FREEZE_REFUSED：既有冻结文件名不是 YYYY-MM-DD。", file=sys.stderr) + return 2 + if latest_date >= parsed_date: + print("FREEZE_REFUSED：冻结日期必须晚于既有最近日期。", file=sys.stderr) + return 2 + previous = earlier[-1] if earlier else None + payload = { + "schema": "dogfood-evidence-freeze/1", + "date": freeze_date, + "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: + print(f"FREEZE_ALREADY_EXISTS path={target}") + return 0 + print("FREEZE_REFUSED：同日冻结已存在且当前证据不同。", file=sys.stderr) + return 2 + temporary = target.with_suffix(target.suffix + ".tmp") + temporary.write_text(serialized, encoding="utf-8") + temporary.replace(target) + print(f"FREEZE_CREATED path={target} files={len(files)}") + return 0 + + +def record_drill( + wave_dir: Path, + action: str, + ack: str, + operator: str, + performed_at: str, + evidence_ref: str, +) -> int: + """签收人工演练证据；本函数不包含任何真实破坏性操作。""" + wave_dir = wave_dir.resolve() + manifest_path = wave_dir / "manifest.json" + if not manifest_path.is_file(): + print("DRILL_REFUSED：波次目录缺少 manifest.json。", file=sys.stderr) + return 2 + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + print(f"DRILL_REFUSED：manifest 不可读：{exc}", file=sys.stderr) + return 2 + wave = manifest.get("wave") + if wave not in WAVE_SCALES: + print("DRILL_REFUSED：manifest 波次身份非法。", file=sys.stderr) + return 2 + expected_ack = f"ACK-DOGFOOD-{wave}-{action}" + if ack != expected_ack: + print(f"ACK_REFUSED：必须显式输入 {expected_ack}", file=sys.stderr) + return 2 + if not operator or parse_timestamp(performed_at) is None: + print("DRILL_REFUSED：operator 和带时区 performed-at 必填。", file=sys.stderr) + return 2 + + # 复用闸门的可信路径边界，但不运行其它证据校验。 + ref_gate = EvidenceGate(wave_dir, "record-drill") + if ref_gate.checked_ref(evidence_ref, f"drills.{action}.evidence") is None: + print("DRILL_REFUSED：evidence-ref 不在波次目录内或不存在。", file=sys.stderr) + return 2 + record = { + "action": action, + "result": "PASS", + "ack": ack, + "performedAt": performed_at, + "operator": operator, + "evidenceRef": evidence_ref, + } + drills_path = wave_dir / "drills.json" + try: + payload = json.loads(drills_path.read_text(encoding="utf-8")) if drills_path.is_file() else {"schema": "dogfood-drills/1", "drills": []} + except (OSError, json.JSONDecodeError) as exc: + print(f"DRILL_REFUSED：drills.json 不可读：{exc}", file=sys.stderr) + return 2 + if not isinstance(payload, dict) or payload.get("schema") != "dogfood-drills/1" or not isinstance(payload.get("drills"), list): + print("DRILL_REFUSED：drills.json 结构非法。", file=sys.stderr) + return 2 + existing = next((item for item in payload["drills"] if isinstance(item, dict) and item.get("action") == action), None) + if existing == record: + print(f"DRILL_ALREADY_RECORDED wave={wave} action={action}") + return 0 + if existing is not None: + print("DRILL_REFUSED：同一动作已有不同记录，不允许覆盖。", file=sys.stderr) + return 2 + payload.pop("_template", None) + payload["drills"].append(record) + temporary = drills_path.with_suffix(".json.tmp") + write_json(temporary, payload) + temporary.replace(drills_path) + print(f"DRILL_RECORDED wave={wave} action={action}（仅记录证据，未执行环境操作）") + return 0 + + +def ensure_wave_scaffold(wave_dir: Path) -> list[str]: + """只补齐缺失模板和目录，绝不覆盖已有证据。""" + created: list[str] = [] + for name, template in JSON_TEMPLATES.items(): + path = wave_dir / name + if not path.exists(): + write_json(path, {**template, "_template": True}) + created.append(name) + for name, header in TSV_TEMPLATES.items(): + path = wave_dir / name + if not path.exists(): + path.write_text(header, encoding="utf-8") + created.append(name) + for name in ("attachments", "freezes"): + path = wave_dir / name + if not path.exists(): + path.mkdir(mode=0o700) + created.append(name + "/") + return created + + +def initialize_wave(root: Path, wave: str) -> int: + """初始化固定波次目录；已存在时只校验身份，不覆盖证据。""" + wave_dir = root.resolve() / 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")) + except (OSError, json.JSONDecodeError) as exc: + print(f"INIT_REFUSED：既有 manifest 不可读：{exc}", file=sys.stderr) + return 2 + if current != expected: + print("INIT_REFUSED：既有目录的波次身份或规模不匹配。", 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)}") + return 0 + print(f"INIT_ALREADY_EXISTS wave={wave} path={wave_dir}") + return 0 + + wave_dir.mkdir(parents=True, mode=0o700) + write_json(manifest_path, expected) + ensure_wave_scaffold(wave_dir) + print(f"INIT_CREATED wave={wave} path={wave_dir}") + return 0 + + +def build_parser() -> argparse.ArgumentParser: + """构造严格参数解析器，波次名不接受任意路径。""" + parser = argparse.ArgumentParser(description="内部狗粮运行保障证据工具") + subparsers = parser.add_subparsers(dest="command", required=True) + init_parser = subparsers.add_parser("init", help="初始化 wave0 或 wave1 证据目录") + init_parser.add_argument("--root", required=True, type=Path, help="证据根目录") + init_parser.add_argument("--wave", required=True, choices=sorted(WAVE_SCALES), help="固定波次") + check_parser = subparsers.add_parser("check", help="执行开波或收口证据门") + check_parser.add_argument("--wave-dir", required=True, type=Path, help="单个波次证据目录") + check_parser.add_argument("--phase", required=True, choices=("readiness", "close"), help="闸门阶段") + freeze_parser = subparsers.add_parser("freeze", help="生成每日 SHA-256 冻结清单") + freeze_parser.add_argument("--wave-dir", required=True, type=Path, help="单个波次证据目录") + freeze_parser.add_argument("--date", required=True, help="冻结日期 YYYY-MM-DD") + drill_parser = subparsers.add_parser("record-drill", help="记录已完成的人工演练证据，不执行环境操作") + drill_parser.add_argument("--wave-dir", required=True, type=Path, help="单个波次证据目录") + drill_parser.add_argument("--action", required=True, choices=sorted(REQUIRED_DRILLS), help="演练动作") + drill_parser.add_argument("--ack", required=True, help="精确 ACK 文本") + drill_parser.add_argument("--operator", required=True, help="演练操作人") + drill_parser.add_argument("--performed-at", required=True, help="带时区 ISO 8601 时间") + drill_parser.add_argument("--evidence-ref", required=True, help="波次目录内证据相对路径") + return parser + + +def main() -> int: + """解析命令并返回稳定退出码。""" + args = build_parser().parse_args() + if args.command == "init": + return initialize_wave(args.root, args.wave) + if args.command == "check": + return EvidenceGate(args.wave_dir, args.phase).run() + if args.command == "freeze": + return freeze_evidence(args.wave_dir, args.date) + if args.command == "record-drill": + return record_drill(args.wave_dir, args.action, args.ack, args.operator, args.performed_at, args.evidence_ref) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/deploy/dogfood-ops/freeze-evidence.sh b/deploy/dogfood-ops/freeze-evidence.sh new file mode 100755 index 00000000..a73854a1 --- /dev/null +++ b/deploy/dogfood-ops/freeze-evidence.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +# 为单个波次生成不可覆盖的每日 SHA-256 证据清单。 +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd -P)" +exec python3 "$SCRIPT_DIR/dogfood_ops.py" freeze "$@" diff --git a/deploy/dogfood-ops/init-wave.sh b/deploy/dogfood-ops/init-wave.sh new file mode 100755 index 00000000..7a041b02 --- /dev/null +++ b/deploy/dogfood-ops/init-wave.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +# 初始化固定规模的内部狗粮证据目录，不连接任何远端环境。 +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd -P)" +exec python3 "$SCRIPT_DIR/dogfood_ops.py" init "$@" diff --git a/deploy/dogfood-ops/record-drill.sh b/deploy/dogfood-ops/record-drill.sh new file mode 100755 index 00000000..724d9727 --- /dev/null +++ b/deploy/dogfood-ops/record-drill.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +# 仅记录已经人工完成的停服、删档或回滚演练；本脚本不执行环境操作。 +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd -P)" +exec python3 "$SCRIPT_DIR/dogfood_ops.py" record-drill "$@" diff --git a/deploy/dogfood-ops/tests/test_dogfood_ops.py b/deploy/dogfood-ops/tests/test_dogfood_ops.py new file mode 100644 index 00000000..9be7402f --- /dev/null +++ b/deploy/dogfood-ops/tests/test_dogfood_ops.py @@ -0,0 +1,532 @@ +#!/usr/bin/env python3 +"""内部狗粮运行保障工具的行为测试。""" + +from __future__ import annotations + +import hashlib +import json +import subprocess +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + + +TOOL = Path(__file__).resolve().parents[1] / "dogfood_ops.py" +OPS_DIR = TOOL.parent + + +class DogfoodOpsTest(unittest.TestCase): + """通过命令行入口验证运维人员实际使用的行为。""" + + def setUp(self) -> None: + """每个用例使用独立临时证据根，避免测试相互污染。""" + self.temp_dir = tempfile.TemporaryDirectory() + self.root = Path(self.temp_dir.name) / "evidence" + + def tearDown(self) -> None: + """清理测试创建的本地证据。""" + self.temp_dir.cleanup() + + def run_tool(self, *args: str) -> subprocess.CompletedProcess[str]: + """调用真实 CLI，并保留标准输出供失败断言。""" + return subprocess.run( + ["python3", str(TOOL), *args], + check=False, + text=True, + capture_output=True, + ) + + def init_wave(self, wave: str) -> Path: + """初始化指定波次并返回证据目录。""" + result = self.run_tool("init", "--root", str(self.root), "--wave", wave) + self.assertEqual(result.returncode, 0, result.stderr) + return self.root / wave + + def write_json(self, path: Path, value: Any) -> None: + """写入稳定 JSON fixture。""" + path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + + def add_attachment(self, wave_dir: Path, name: str, content: str = "evidence") -> tuple[str, str]: + """创建目录内证据附件并返回相对路径与哈希。""" + attachment = wave_dir / "attachments" / name + attachment.parent.mkdir(exist_ok=True) + attachment.write_text(content, encoding="utf-8") + return attachment.relative_to(wave_dir).as_posix(), hashlib.sha256(attachment.read_bytes()).hexdigest() + + 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"] + commit = "1" * 40 + artifact_ref, artifact_hash = self.add_attachment(wave_dir, "rc-artifact.bin") + 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") + storage_ref, storage_hash = self.add_attachment(wave_dir, "storage-audit.json") + 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") + database_ref, _ = self.add_attachment(wave_dir, "database.json") + + self.write_json(wave_dir / "rc.json", { + "schema": "dogfood-rc/1", "commit": commit, "repoClean": True, + "artifactRef": artifact_ref, "artifactSha256": artifact_hash, + }) + self.write_json(wave_dir / "config-identity.json", { + "schema": "dogfood-config-identity/1", "capturedAt": "2026-07-22T00:00:00Z", + "rcCommit": commit, "configBundleRef": config_ref, "configSha256": config_hash, + }) + self.write_json(wave_dir / "snapshot.json", { + "schema": "dogfood-snapshot/1", + "mysql": {"snapshotRef": dump_ref, "sha256": dump_hash, "capturedAt": "2026-07-22T00:00:00Z"}, + "isolatedRestore": {"result": "PASS", "target": "isolated-dogfood-restore-001", "completedAt": "2026-07-22T01:00:00Z", "evidenceRef": restore_ref}, + }) + self.write_json(wave_dir / "storage-audit.json", { + "schema": "dogfood-storage-audit/1", "result": "PASS", "capturedAt": "2026-07-22T00:30:00Z", + "reportRef": storage_ref, "reportSha256": storage_hash, + }) + self.write_json(wave_dir / "accounts.json", { + "schema": "dogfood-accounts/1", "batchId": f"dogfood-{wave}", + "creatorCount": manifest["creatorCount"], "playerCount": manifest["playerCount"], + "freeQuotaCount": 4 if wave == "wave0" else 12, + }) + + permission_cases = { + "creator-own-project": "ALLOW", + "creator-cross-account-denied": "DENY", + "reviewer-review-only": "ALLOW", + "reviewer-config-write-denied": "DENY", + "readonly-write-denied": "DENY", + } + self.write_json(wave_dir / "role-permissions.json", { + "schema": "dogfood-role-permissions/1", + "checks": [{"case": case, "result": result, "screenshotRef": common_ref} for case, result in permission_cases.items()], + }) + + 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"}, + "stages": [{"stage": stage, "result": "PASS", **evidence_bundle} for stage in stages], + }) + + negative_cases = [ + "draft-publish", "reviewing-publish", "rejected-publish", "direct-publish-bypass", + "artifact-drift", "delisted-feed", "delisted-detail", "delisted-deep-link", + ] + self.write_json(wave_dir / "release-negatives.json", { + "schema": "dogfood-release-negatives/1", + "checks": [{"case": case, "blocked": True, "evidenceRef": common_ref} for case in negative_cases], + }) + self.write_json(wave_dir / "manual-play.json", { + "schema": "dogfood-manual-play/1", "releasedGameCount": 1, + "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) + ] + self.write_json(wave_dir / "performance.json", { + "schema": "dogfood-first-frame-performance/1", + "context": {"host": "mini-desktop", "browser": "Chrome", "browserVersion": "146", "viewport": "390x844@1", "network": "Tailscale staging", "timing": "navigation-commit-to-first-interactive-game-frame"}, + "samples": samples, + }) + self.write_json(wave_dir / "telemetry.json", { + "schema": "dogfood-telemetry-completeness/1", "expectedCount": 20, "completeCount": 19, + "evidenceRef": common_ref, + }) + self.write_json(wave_dir / "alerts.json", { + "schema": "dogfood-alert-delivery/1", + "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) + end = start + timedelta(hours=manifest["durationHours"]) + (wave_dir / "oncall.tsv").write_text( + "start_at\tend_at\tprimary\tbackup\tcontact\n" + f"{start.isoformat().replace('+00:00', 'Z')}\t{end.isoformat().replace('+00:00', 'Z')}\toncall-1\toncall-2\tinternal-roster\n", + encoding="utf-8", + ) + (wave_dir / "feedback.tsv").write_text( + "id\tcreated_at\treporter_id\trole\tcategory\tseverity\tsummary\tstatus\towner\tevidence_ref\n" + f"F-1\t2026-07-22T04:00:00Z\tu-1\tplayer\tux\tP2\tfixture feedback\tCLOSED\tops\t{common_ref}\n", + encoding="utf-8", + ) + checklist_items = ["role-permissions", "main-chain", "release-negatives", "manual-play-before-feed"] + (wave_dir / "manual-acceptance.tsv").write_text( + "item\tresult\toperator\tchecked_at\tevidence_ref\n" + "".join( + f"{item}\tPASS\toperator-1\t2026-07-22T05:00:00Z\t{common_ref}\n" for item in checklist_items + ), + encoding="utf-8", + ) + self.write_json(wave_dir / "drills.json", { + "schema": "dogfood-drills/1", + "drills": [{"action": action, "result": "PASS", "ack": f"ACK-DOGFOOD-{wave}-{action}", "performedAt": "2026-07-22T06:00:00Z", "operator": "ops-1", "evidenceRef": common_ref} for action in ["stop", "delete", "rollback"]], + }) + stop_conditions = [ + "security-incident", "cross-account-access", "data-loss", "real-charge", "real-ad-impression", + "unreviewed-content-in-feed", "generation-success-below-80", "confirmed-false-positive", + "same-root-cause-blocked-two-users", "telemetry-missing-above-5", "review-backlog-over-4h", + "acceptance-cost-exceeded", + ] + self.write_json(wave_dir / "safety-status.json", { + "schema": "dogfood-safety-status/1", "assessedAt": "2026-07-22T07:00:00Z", "assessor": "ops-1", + "conditions": [{"condition": condition, "triggered": False} for condition in stop_conditions], + }) + + def test_init_wave0_creates_fixed_ledgers_and_scale(self) -> None: + """wave0 必须固定为 3 名创作者和 5 名玩家。""" + wave_dir = self.init_wave("wave0") + + manifest = json.loads((wave_dir / "manifest.json").read_text(encoding="utf-8")) + self.assertEqual(manifest["wave"], "wave0") + self.assertEqual(manifest["creatorCount"], 3) + self.assertEqual(manifest["playerCount"], 5) + self.assertTrue((wave_dir / "oncall.tsv").is_file()) + self.assertTrue((wave_dir / "feedback.tsv").is_file()) + self.assertTrue((wave_dir / "p0.tsv").is_file()) + self.assertTrue((wave_dir / "config-identity.json").is_file()) + self.assertTrue((wave_dir / "manual-acceptance.tsv").is_file()) + for name in ( + "rc.json", "snapshot.json", "storage-audit.json", "accounts.json", "role-permissions.json", + "main-chain.json", "release-negatives.json", "manual-play.json", "performance.json", + "telemetry.json", "alerts.json", "drills.json", "safety-status.json", + ): + self.assertTrue((wave_dir / name).is_file(), name) + self.assertTrue((wave_dir / "attachments").is_dir()) + self.assertTrue((wave_dir / "freezes").is_dir()) + + def test_init_wave1_uses_approved_scale(self) -> None: + """wave1 必须固定为累计 10 名创作者和 30 名玩家。""" + wave_dir = self.init_wave("wave1") + + manifest = json.loads((wave_dir / "manifest.json").read_text(encoding="utf-8")) + self.assertEqual((manifest["creatorCount"], manifest["playerCount"]), (10, 30)) + + def test_repeated_init_is_idempotent_and_does_not_overwrite(self) -> None: + """重复初始化应成功返回，但不得覆盖已经填写的证据。""" + wave_dir = self.init_wave("wave0") + feedback = wave_dir / "feedback.tsv" + feedback.write_text(feedback.read_text(encoding="utf-8") + "F-1\t2026-07-22T00:00:00Z\tu1\tplayer\tbug\tP1\t卡住\tOPEN\tops\tattachments/f.txt\n", encoding="utf-8") + before = hashlib.sha256(feedback.read_bytes()).hexdigest() + + result = self.run_tool("init", "--root", str(self.root), "--wave", "wave0") + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(hashlib.sha256(feedback.read_bytes()).hexdigest(), before) + self.assertIn("INIT_ALREADY_EXISTS", result.stdout) + + def test_repeated_init_repairs_missing_templates_without_overwrite(self) -> None: + """初始化中断后重试应补回缺失模板，同时保留已经填写的账本。""" + wave_dir = self.init_wave("wave0") + feedback = wave_dir / "feedback.tsv" + feedback.write_text(feedback.read_text(encoding="utf-8") + "kept\n", encoding="utf-8") + (wave_dir / "rc.json").unlink() + (wave_dir / "attachments").rmdir() + + result = self.run_tool("init", "--root", str(self.root), "--wave", "wave0") + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("INIT_REPAIRED", result.stdout) + self.assertTrue((wave_dir / "rc.json").is_file()) + self.assertTrue((wave_dir / "attachments").is_dir()) + self.assertTrue(feedback.read_text(encoding="utf-8").endswith("kept\n")) + + def test_init_rejects_unknown_wave_instead_of_creating_a_path(self) -> None: + """波次名不得被当成任意相对路径使用。""" + result = self.run_tool("init", "--root", str(self.root), "--wave", "../wave0") + + self.assertNotEqual(result.returncode, 0) + self.assertFalse((Path(self.temp_dir.name) / "wave0").exists()) + self.assertIn("invalid choice", result.stderr) + + def test_readiness_fails_closed_when_evidence_is_missing(self) -> None: + """仅初始化模板不得被误判为可开波。""" + wave_dir = self.init_wave("wave0") + + result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") + + self.assertNotEqual(result.returncode, 0) + self.assertIn("rc.missing", result.stdout) + self.assertNotIn("GATE_PASS", result.stdout) + + def test_readiness_accepts_complete_structured_evidence(self) -> None: + """完整且互相对账的结构化证据应通过开波门。""" + wave_dir = self.init_wave("wave0") + self.populate_valid_evidence(wave_dir) + + result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") + + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertIn("GATE_PASS phase=readiness", result.stdout) + + def test_performance_requires_at_least_30_cold_independent_samples(self) -> None: + """冷样本少于 30 个时必须直接失败。""" + 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"] = payload["samples"][:29] + 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.sample-count", result.stdout) + + def test_performance_p75_uses_nearest_rank_and_rejects_3000ms(self) -> None: + """30 样本的第 23 个值是 P75，恰好 3000ms 不满足严格小于。""" + 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 index, sample in enumerate(payload["samples"]): + sample["durationMs"] = 1000 if index < 22 else 3000 + 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.p75", result.stdout) + + def test_performance_accepts_p75_of_2999ms(self) -> None: + """严格阈值内的第 23 个值应通过。""" + 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 index, sample in enumerate(payload["samples"]): + sample["durationMs"] = 1000 if index < 22 else 2999 + 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_hot_cache_mixed_into_cold_samples(self) -> None: + """冷样本文件中混入热缓存记录时不得计算 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")) + payload["samples"][0]["cache"] = "hot" + 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.cold-only", result.stdout) + + def test_performance_rejects_non_finite_duration(self) -> None: + """JSON 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.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) + + def test_main_chain_requires_distinct_evidence_modalities(self) -> None: + """同一个占位附件不得同时冒充截图、网络、日志和数据库证据。""" + 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")) + for field in ("networkRef", "logRef", "databaseRef"): + payload["stages"][0][field] = payload["stages"][0]["screenshotRef"] + 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.evidence-modalities", result.stdout) + + def test_telemetry_accepts_exactly_95_percent_and_rejects_below(self) -> None: + """完整率 95% 可过，低于 95% 必须失败。""" + wave_dir = self.init_wave("wave0") + self.populate_valid_evidence(wave_dir) + telemetry = wave_dir / "telemetry.json" + + pass_result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") + self.assertEqual(pass_result.returncode, 0, pass_result.stdout + pass_result.stderr) + + self.write_json(telemetry, {"schema": "dogfood-telemetry-completeness/1", "expectedCount": 100, "completeCount": 94, "evidenceRef": "attachments/browser-proof.png"}) + fail_result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") + self.assertNotEqual(fail_result.returncode, 0) + self.assertIn("telemetry.completeness", fail_result.stdout) + + def test_top_level_pass_cannot_forge_restore_evidence(self) -> None: + """写一个 PASS 字符串不能替代隔离恢复目标和证据。""" + wave_dir = self.init_wave("wave0") + self.populate_valid_evidence(wave_dir) + self.write_json(wave_dir / "snapshot.json", {"schema": "dogfood-snapshot/1", "result": "PASS"}) + + result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") + + self.assertNotEqual(result.returncode, 0) + self.assertIn("snapshot.mysql", result.stdout) + self.assertIn("snapshot.isolated-restore", result.stdout) + + def test_evidence_reference_cannot_escape_wave_directory(self) -> None: + """即使目录外文件存在，证据引用也不得越过波次根。""" + wave_dir = self.init_wave("wave0") + self.populate_valid_evidence(wave_dir) + outside = wave_dir.parent / "outside.png" + outside.write_text("outside", encoding="utf-8") + payload = json.loads((wave_dir / "release-negatives.json").read_text(encoding="utf-8")) + payload["checks"][0]["evidenceRef"] = "../outside.png" + self.write_json(wave_dir / "release-negatives.json", payload) + + result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness") + + 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 的验收结果。""" + 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 + 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) + + def test_close_requires_freeze_and_detects_post_freeze_tampering(self) -> None: + """收口门必须验证最近一次 SHA-256 冻结，冻结后改动立即失效。""" + wave_dir = self.init_wave("wave0") + self.populate_valid_evidence(wave_dir) + + missing = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close") + self.assertNotEqual(missing.returncode, 0) + self.assertIn("freeze.missing", missing.stdout) + + 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) + + 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) + + def test_close_rejects_malformed_freeze_without_traceback(self) -> None: + """畸形冻结清单应形成可追踪失败，而不是让闸门异常退出。""" + wave_dir = self.init_wave("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") + + result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close") + + self.assertEqual(result.returncode, 1, result.stdout + result.stderr) + self.assertIn("freeze.schema", result.stdout) + self.assertNotIn("Traceback", result.stderr) + + def test_close_rejects_symlink_added_after_freeze_without_traceback(self) -> None: + """冻结后加入符号链接时应 fail-closed，并保持稳定退出码。""" + wave_dir = self.init_wave("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) + (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") + + self.assertEqual(result.returncode, 1, result.stdout + result.stderr) + self.assertIn("freeze.collection", result.stdout) + self.assertNotIn("Traceback", result.stderr) + + def test_wave1_close_requires_three_chained_daily_freezes(self) -> None: + """三天 wave1 不得用单日快照代替每日冻结，且历史清单必须受哈希链保护。""" + wave_dir = self.init_wave("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) + + 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.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) + + def test_record_drill_requires_exact_ack_and_is_idempotent(self) -> None: + """演练记录必须使用精确 ACK，重复签收同一证据不得制造重复行。""" + wave_dir = self.init_wave("wave0") + evidence_ref, _ = self.add_attachment(wave_dir, "stop-drill.txt") + common_args = ( + "record-drill", "--wave-dir", str(wave_dir), "--action", "stop", + "--operator", "ops-1", "--performed-at", "2026-07-22T06:00:00Z", + "--evidence-ref", evidence_ref, + ) + + refused = self.run_tool(*common_args, "--ack", "yes") + self.assertNotEqual(refused.returncode, 0) + self.assertIn("ACK_REFUSED", refused.stderr) + + accepted = self.run_tool(*common_args, "--ack", "ACK-DOGFOOD-wave0-stop") + self.assertEqual(accepted.returncode, 0, accepted.stdout + accepted.stderr) + repeated = self.run_tool(*common_args, "--ack", "ACK-DOGFOOD-wave0-stop") + self.assertEqual(repeated.returncode, 0, repeated.stdout + repeated.stderr) + self.assertIn("DRILL_ALREADY_RECORDED", repeated.stdout) + payload = json.loads((wave_dir / "drills.json").read_text(encoding="utf-8")) + self.assertEqual(len(payload["drills"]), 1) + + 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) + + (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) + + def test_shell_entrypoints_delegate_to_the_single_cli(self) -> None: + """值班入口应直接调用同一套实现，避免脚本间口径漂移。""" + scripts = [ + "init-wave.sh", "check-readiness.sh", "check-close.sh", + "freeze-evidence.sh", "record-drill.sh", + ] + for name in scripts: + result = subprocess.run( + ["bash", str(OPS_DIR / name), "--help"], + check=False, + text=True, + capture_output=True, + ) + self.assertEqual(result.returncode, 0, f"{name}: {result.stdout}{result.stderr}") + + +if __name__ == "__main__": + unittest.main()