From 3d928c4f8cda48e1870316812b2b790eb6a5e471 Mon Sep 17 00:00:00 2001 From: lili Date: Thu, 23 Jul 2026 00:42:58 -0700 Subject: [PATCH] fix(dogfood-ops): bind evidence to wave windows --- deploy/dogfood-ops/dogfood_ops.py | 80 ++++++++---- deploy/dogfood-ops/tests/test_dogfood_ops.py | 124 +++++++++++++++++-- 2 files changed, 165 insertions(+), 39 deletions(-) diff --git a/deploy/dogfood-ops/dogfood_ops.py b/deploy/dogfood-ops/dogfood_ops.py index 99d9a0d6..5ca7c1da 100755 --- a/deploy/dogfood-ops/dogfood_ops.py +++ b/deploy/dogfood-ops/dogfood_ops.py @@ -144,6 +144,21 @@ class EvidenceGate: """记录一条可追踪失败，不因首个错误隐藏其余缺口。""" self.errors.append((code, message)) + def checked_evidence_time(self, value: Any, context: str) -> datetime | None: + """统一校验证据时间必须落在波次开始与当前可信时钟之间。""" + parsed = parse_timestamp(value) + if parsed is None: + self.fail("evidence-time.invalid", f"{context} 必须是带时区 ISO 8601 时间") + return None + parsed = parsed.astimezone(timezone.utc) + if self.started_at is not None and (parsed < self.started_at or parsed > self.now): + self.fail( + "evidence-time.window", + f"{context} 必须满足 startedAt <= evidenceTime <= now：{value}", + ) + return None + return parsed + def load_json(self, name: str, schema: str) -> dict[str, Any] | None: """读取固定 JSON 文件并校验对象类型与 schema。""" path = self.wave_dir / name @@ -288,7 +303,7 @@ class EvidenceGate: self.fail("rc.clean", "RC 必须来自 tracked 工作区干净构建") 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: + if self.checked_evidence_time(config.get("capturedAt"), "config.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"): @@ -303,7 +318,7 @@ class EvidenceGate: 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: + if self.checked_evidence_time(mysql.get("capturedAt"), "snapshot.mysql.capturedAt") is None: self.fail("snapshot.mysql", "MySQL 快照缺少带时区采集时间") restore = snapshot.get("isolatedRestore") if not isinstance(restore, dict): @@ -312,13 +327,14 @@ class EvidenceGate: 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: + if self.checked_evidence_time(restore.get("completedAt"), "snapshot.isolatedRestore.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: + if storage.get("result") != "PASS" or self.checked_evidence_time( + storage.get("capturedAt"), "storage-audit.capturedAt") is None: self.fail("storage-audit.result", "对象存储审计必须含真实 PASS 与采集时间") self.check_hash_ref(storage.get("reportRef"), storage.get("reportSha256"), "storage-audit.report") @@ -433,7 +449,7 @@ class EvidenceGate: or not review_matches or review.get("status") != "APPROVED" or not review.get("reviewId") - or parse_timestamp(review.get("approvedAt")) is None + or self.checked_evidence_time(review.get("approvedAt"), "main-chain.review.approvedAt") is None ): self.fail("main-chain.game-artifact", "审核记录未批准或未绑定同一游戏产物身份") @@ -444,7 +460,7 @@ class EvidenceGate: 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 + or self.checked_evidence_time(database.get("capturedAt"), "main-chain.database.capturedAt") is None ): self.fail("main-chain.game-artifact", "主链数据库证据未绑定同一游戏产物身份") @@ -472,8 +488,8 @@ class EvidenceGate: 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")) + played_at = self.checked_evidence_time(item.get("playedAt"), "manual-play.playedAt") + feed_at = self.checked_evidence_time(item.get("feedEnteredAt"), "manual-play.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": @@ -524,11 +540,9 @@ class EvidenceGate: navigation_ids.append(navigation_id) 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: + collected_at = self.checked_evidence_time(sample.get("collectedAt"), "performance.collectedAt") + if collected_at is None: 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) @@ -634,8 +648,10 @@ class EvidenceGate: 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")) + start = self.checked_evidence_time( + metadata.get("navigationCommitAt"), f"performance.{filename}.navigationCommitAt") + end = self.checked_evidence_time( + metadata.get("firstInteractiveGameFrameAt"), f"performance.{filename}.firstInteractiveGameFrameAt") if start is None or end is None or end < start: self.fail("performance.duration", f"HAR 起止时间非法：{filename}") return None @@ -667,7 +683,9 @@ class EvidenceGate: 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"): + received_at = self.checked_evidence_time( + item.get("receivedAt") if isinstance(item, dict) else None, "alerts.receivedAt") + if not isinstance(item, dict) or received_at 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") @@ -678,8 +696,8 @@ class EvidenceGate: 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")) + start = self.checked_evidence_time(row.get("start_at"), "oncall.start_at") + end = self.checked_evidence_time(row.get("end_at"), "oncall.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: @@ -701,7 +719,8 @@ class EvidenceGate: 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: + created_at = self.checked_evidence_time(row.get("created_at"), f"feedback.{row.get('id', 'unknown')}.created_at") + if any(not row.get(field) for field in row) or created_at is None: self.fail("feedback.row", "反馈账本存在字段不完整的行") continue if row["id"] in seen_ids: @@ -714,7 +733,9 @@ class EvidenceGate: 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: + checked_at = self.checked_evidence_time( + row.get("checked_at") if row else None, f"manual-acceptance.{item_name}.checked_at") + if row is None or row.get("result") != "PASS" or not row.get("operator") or 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") @@ -730,14 +751,18 @@ class EvidenceGate: 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: + performed_at = self.checked_evidence_time( + item.get("performedAt") if item else None, f"drills.{action}.performedAt") + if item is None or item.get("result") != "PASS" or item.get("ack") != expected_ack or not item.get("operator") or performed_at 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"): + assessed_at = self.checked_evidence_time( + safety.get("assessedAt") if safety else None, "safety-status.assessedAt") + if not isinstance(conditions, list) or assessed_at 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)} @@ -1128,14 +1153,17 @@ def record_drill( 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) + if not operator: + print("DRILL_REFUSED：operator 必填。", file=sys.stderr) return 2 - # 复用闸门的可信路径边界，但不运行其它证据校验。 + # 复用闸门的 manifest、统一证据时间窗和可信路径边界，但不运行其它业务证据校验。 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) + ref_gate.validate_manifest() + performed_time = ref_gate.checked_evidence_time(performed_at, f"drills.{action}.performedAt") + evidence_path = ref_gate.checked_ref(evidence_ref, f"drills.{action}.evidence") + if ref_gate.errors or performed_time is None or evidence_path is None: + print("DRILL_REFUSED：performed-at 越过波次时间窗，或 evidence-ref 非法。", file=sys.stderr) return 2 record = { "action": action, diff --git a/deploy/dogfood-ops/tests/test_dogfood_ops.py b/deploy/dogfood-ops/tests/test_dogfood_ops.py index 105906a5..2c1f777d 100644 --- a/deploy/dogfood-ops/tests/test_dogfood_ops.py +++ b/deploy/dogfood-ops/tests/test_dogfood_ops.py @@ -26,6 +26,8 @@ MODULE_SPEC.loader.exec_module(DOGFOOD_OPS) class DogfoodOpsTest(unittest.TestCase): """通过命令行入口验证运维人员实际使用的行为。""" + FIXTURE_STARTED_AT = datetime(2026, 7, 20, 9, 0, tzinfo=timezone.utc) + def setUp(self) -> None: """每个用例使用独立临时证据根，避免测试相互污染。""" self.temp_dir = tempfile.TemporaryDirectory() @@ -46,8 +48,9 @@ class DogfoodOpsTest(unittest.TestCase): 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) + # 公共通过夹具固定在已过去的可信时间，避免用未来证据伪造刚初始化波次的 readiness。 + result = DOGFOOD_OPS.initialize_wave(self.root, wave, now=self.FIXTURE_STARTED_AT) + self.assertEqual(result, 0) return self.root / wave def write_json(self, path: Path, value: Any) -> None: @@ -87,6 +90,92 @@ class DogfoodOpsTest(unittest.TestCase): self.write_json(raw_path, payload) sample["rawSha256"] = hashlib.sha256(raw_path.read_bytes()).hexdigest() + def set_json_time(self, wave_dir: Path, filename: str, keys: tuple[Any, ...], value: str) -> None: + """修改一处 JSON 证据时间，供统一时间窗接线测试复用。""" + path = wave_dir / filename + payload = json.loads(path.read_text(encoding="utf-8")) + target: Any = payload + for key in keys[:-1]: + target = target[key] + target[keys[-1]] = value + self.write_json(path, payload) + + def test_checked_evidence_time_rejects_both_sides_of_wave_window(self) -> None: + """统一时间窗必须同时拒绝波次开始前和可信当前时钟之后的证据。""" + started_at = self.FIXTURE_STARTED_AT + now = started_at + timedelta(hours=24) + wave_dir = self.init_wave("wave0") + gate = DOGFOOD_OPS.EvidenceGate(wave_dir, "readiness", now=now) + gate.started_at = started_at + + self.assertIsNone(gate.checked_evidence_time( + DOGFOOD_OPS.format_timestamp(started_at - timedelta(seconds=1)), "fixture.before" + )) + self.assertIsNone(gate.checked_evidence_time( + DOGFOOD_OPS.format_timestamp(now + timedelta(seconds=1)), "fixture.future" + )) + self.assertEqual( + [code for code, _ in gate.errors], + ["evidence-time.window", "evidence-time.window"], + ) + + def test_all_operational_evidence_times_use_the_same_wave_window(self) -> None: + """值班、告警、反馈、人工验收、性能及其余结构化证据时间必须全部接入统一时间窗。""" + started_at = self.FIXTURE_STARTED_AT + now = started_at + timedelta(hours=25) + before = DOGFOOD_OPS.format_timestamp(started_at - timedelta(seconds=1)) + future = DOGFOOD_OPS.format_timestamp(now + timedelta(seconds=1)) + + def mutate_performance(wave_dir: Path, value: str) -> None: + path = wave_dir / "performance.json" + payload = json.loads(path.read_text(encoding="utf-8")) + sample = payload["samples"][0] + sample["collectedAt"] = value + raw_path = wave_dir / sample["rawRef"] + raw = json.loads(raw_path.read_text(encoding="utf-8")) + raw["metadata"]["collectedAt"] = value + self.write_json(raw_path, raw) + sample["rawSha256"] = hashlib.sha256(raw_path.read_bytes()).hexdigest() + self.write_json(path, payload) + + def mutate_tsv(wave_dir: Path, filename: str, old: str, value: str) -> None: + path = wave_dir / filename + path.write_text(path.read_text(encoding="utf-8").replace(old, value), encoding="utf-8") + + scenarios = [ + ("config", before, lambda wave_dir, value: self.set_json_time(wave_dir, "config-identity.json", ("capturedAt",), value)), + ("snapshot", future, lambda wave_dir, value: self.set_json_time(wave_dir, "snapshot.json", ("mysql", "capturedAt"), value)), + ("restore", before, lambda wave_dir, value: self.set_json_time(wave_dir, "snapshot.json", ("isolatedRestore", "completedAt"), value)), + ("storage", future, lambda wave_dir, value: self.set_json_time(wave_dir, "storage-audit.json", ("capturedAt",), value)), + ("review", before, lambda wave_dir, value: self.set_json_time(wave_dir, "attachments/review-binding.json", ("approvedAt",), value)), + ("database", future, lambda wave_dir, value: self.set_json_time(wave_dir, "attachments/database.json", ("capturedAt",), value)), + ("manual-play", before, lambda wave_dir, value: self.set_json_time(wave_dir, "manual-play.json", ("games", 0, "playedAt"), value)), + ("performance", future, mutate_performance), + ("alerts", before, lambda wave_dir, value: self.set_json_time(wave_dir, "alerts.json", ("deliveries", 0, "receivedAt"), value)), + ("oncall-start", before, lambda wave_dir, value: mutate_tsv(wave_dir, "oncall.tsv", DOGFOOD_OPS.format_timestamp(started_at), value)), + ("oncall-end", future, lambda wave_dir, value: mutate_tsv(wave_dir, "oncall.tsv", DOGFOOD_OPS.format_timestamp(started_at + timedelta(hours=24)), value)), + ("feedback", before, lambda wave_dir, value: mutate_tsv(wave_dir, "feedback.tsv", DOGFOOD_OPS.format_timestamp(started_at), value)), + ("manual-acceptance", future, lambda wave_dir, value: mutate_tsv(wave_dir, "manual-acceptance.tsv", DOGFOOD_OPS.format_timestamp(started_at), value)), + ("drills", before, lambda wave_dir, value: self.set_json_time(wave_dir, "drills.json", ("drills", 0, "performedAt"), value)), + ("safety", future, lambda wave_dir, value: self.set_json_time(wave_dir, "safety-status.json", ("assessedAt",), value)), + ] + + for name, invalid_time, mutate in scenarios: + with self.subTest(evidence=name): + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) / "evidence" + self.assertEqual(DOGFOOD_OPS.initialize_wave(root, "wave0", now=started_at), 0) + wave_dir = root / "wave0" + self.populate_valid_evidence(wave_dir) + mutate(wave_dir, invalid_time) + + gate = DOGFOOD_OPS.EvidenceGate(wave_dir, "readiness", now=now) + self.assertNotEqual(gate.run(), 0) + self.assertTrue( + any(code == "evidence-time.window" for code, _ in gate.errors), + f"{name} 未接入统一时间窗：{gate.errors}", + ) + def test_wave1_init_requires_closed_wave0(self) -> None: """没有 Wave0 收口回执时不得初始化 Wave1。""" result = self.run_tool("init", "--root", str(self.root), "--wave", "wave1") @@ -199,7 +288,9 @@ class DogfoodOpsTest(unittest.TestCase): 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") + captured_at = DOGFOOD_OPS.format_timestamp(started_at) + played_at = DOGFOOD_OPS.format_timestamp(started_at + timedelta(seconds=1)) + feed_entered_at = DOGFOOD_OPS.format_timestamp(started_at + timedelta(seconds=2)) commit = "1" * 40 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") @@ -238,16 +329,16 @@ class DogfoodOpsTest(unittest.TestCase): "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", + "schema": "dogfood-config-identity/1", "capturedAt": captured_at, "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}, + "mysql": {"snapshotRef": dump_ref, "sha256": dump_hash, "capturedAt": captured_at}, + "isolatedRestore": {"result": "PASS", "target": "isolated-dogfood-restore-001", "completedAt": captured_at, "evidenceRef": restore_ref}, }) self.write_json(wave_dir / "storage-audit.json", { - "schema": "dogfood-storage-audit/1", "result": "PASS", "capturedAt": "2026-07-22T00:30:00Z", + "schema": "dogfood-storage-audit/1", "result": "PASS", "capturedAt": captured_at, "reportRef": storage_ref, "reportSha256": storage_hash, }) self.write_json(wave_dir / "accounts.json", { @@ -291,7 +382,7 @@ class DogfoodOpsTest(unittest.TestCase): }) 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"}], + "games": [{"gameId": "g-1", "tester": "tester-1", "playedAt": played_at, "feedEnteredAt": feed_entered_at, "screenshotRef": common_ref, "source": "/gstack@mini-desktop"}], }) samples = [] @@ -327,7 +418,7 @@ class DogfoodOpsTest(unittest.TestCase): }) 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}], + "deliveries": [{"alertId": "alert-1", "receivedAt": captured_at, "recipient": "oncall-1", "externalMessageId": "msg-1", "evidenceRef": common_ref}], }) start = started_at @@ -339,19 +430,19 @@ class DogfoodOpsTest(unittest.TestCase): ) (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", + f"F-1\t{captured_at}\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 + f"{item}\tPASS\toperator-1\t{captured_at}\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"]], + "drills": [{"action": action, "result": "PASS", "ack": f"ACK-DOGFOOD-{wave}-{action}", "performedAt": captured_at, "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", @@ -360,7 +451,7 @@ class DogfoodOpsTest(unittest.TestCase): "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", + "schema": "dogfood-safety-status/1", "assessedAt": captured_at, "assessor": "ops-1", "conditions": [{"condition": condition, "triggered": False} for condition in stop_conditions], }) @@ -865,6 +956,13 @@ class DogfoodOpsTest(unittest.TestCase): self.assertNotEqual(refused.returncode, 0) self.assertIn("ACK_REFUSED", refused.stderr) + for invalid_time in ("2026-07-20T08:59:59Z", "2999-01-01T00:00:00Z"): + invalid_args = list(common_args) + invalid_args[invalid_args.index("--performed-at") + 1] = invalid_time + refused_time = self.run_tool(*invalid_args, "--ack", "ACK-DOGFOOD-wave0-stop") + self.assertNotEqual(refused_time.returncode, 0) + self.assertIn("DRILL_REFUSED", refused_time.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")