"""Offline audit of an official Q3 run; never connects to the simulator. Usage: python scripts/verify_q3_practice.py output/q3/practice/20260911-port2026 Preserves session.jsonl and summary.json; writes derived audit artifacts only. """ from __future__ import annotations import argparse import csv from dataclasses import asdict from datetime import datetime import hashlib import json import math from pathlib import Path from urllib.parse import urlsplit from benchmark_q3_search import SevenBatch, seven_points from mcm26b.config import ProblemConfig from mcm26b.coverage import coverage_points from mcm26b.evaluation import recompute_virtual_time from mcm26b.planner import OmnidirectionalLoop from mcm26b.protocol.client import SimulatorClient from mcm26b.sim.playback import PlaybackTransport class StrictPlayback(PlaybackTransport): """Check complete action content, not merely the sequence of API paths.""" def post(self, url, body, headers, timeout): if self.index >= len(self.events): raise AssertionError("Replay attempted an extra action") expected = self.events[self.index] actual_request = json.loads(body) without_id = lambda value: {k: v for k, v in value.items() if k != "request_id"} if (urlsplit(url).path != expected["path"] or without_id(actual_request) != without_id(expected["request"])): raise AssertionError(f"Replay action differs at index {self.index}") return super().post(url, body, headers, timeout) class AuditMixin: def __init__(self, client, config): super().__init__(client, config) self.certificates = [] def _attempt_clear(self, channel, target, cert): vertices = self.belief.candidate_set(channel).polygon self.certificates.append({**asdict(cert), "submitted_position": target, "submitted_max_vertex_distance_m": max(math.dist(target, v) for v in vertices)}) return super()._attempt_clear(channel, target, cert) class AuditedLoop(AuditMixin, OmnidirectionalLoop): pass class AuditedSevenBatch(AuditMixin, SevenBatch): pass def save_json(path, value): path.write_text(json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8") def audit(folder: Path) -> dict: records = [json.loads(line) for line in (folder / "session.jsonl").read_text(encoding="utf-8").splitlines()] events = [e for e in records if e.get("event") == "accepted"] original = json.loads((folder / "summary.json").read_text(encoding="utf-8-sig")) strategy = original.get("strategy", "grid49_immediate") if strategy not in ("grid49_immediate", "seven_batch"): raise ValueError(f"Unsupported strategy: {strategy}") official_paths = list(folder.glob("*.result.json")) if len(official_paths) != 1 or not events: raise ValueError("Expected one official result and nonempty session evidence") official_path = official_paths[0] official = json.loads(official_path.read_text(encoding="utf-8")) package = official_path.with_name(official_path.name.replace(".result.json", ".jlog")) checks = {} checks["official_problem_3_all_omnidirectional"] = ( official["problem_no"] == 3 and official["directional_jammer_count"] == 0 and 10 <= official["jammer_count"] <= 16 and official["omnidirectional_jammer_count"] == official["jammer_count"]) checks["official_package_sha256_matches"] = ( hashlib.sha256(package.read_bytes()).hexdigest() == official["package_sha256"]) checks["normal_enter_and_exit"] = ( events[0]["path"] == "/enter" and events[-1]["path"] == "/exit" and events[-1]["response"].get("exit_reason") == "user_exit") start_ms = datetime.fromisoformat(official["window_started_at_utc"]).timestamp() * 1000 end_ms = datetime.fromisoformat(official["ended_at_utc"]).timestamp() * 1000 checks["responses_within_official_run_window"] = all( start_ms <= e["response"]["real_timestamp_ms"] <= end_ms + 1 for e in events) ids = [e["request"]["request_id"] for e in events] checks["unique_accepted_request_ids"] = len(ids) == len(set(ids)) checks["no_rejections_or_retries"] = ( len(records) == 2 * len(events) and all(e.get("attempts") == 1 for e in events) and all(records[2*i].get("event") == "attempt" and records[2*i]["request"] == e["request"] and records[2*i+1] == e for i, e in enumerate(events))) clears = [e for e in events if e["path"] == "/clear"] successes = [e for e in clears if e["response"].get("clear_result") == "success"] cleared = sorted({e["request"]["channel"] for e in successes}) checks["all_official_sources_cleared"] = len(cleared) == official["jammer_count"] checks["every_clear_succeeded_once"] = len(clears) == len(successes) == len(cleared) checks["original_summary_completed"] = ( original["completed"] and original["cleared_channels"] == cleared and original["unresolved_channels"] == [] and original["anomalies"] == []) grid = set(seven_points() if strategy == "seven_batch" else coverage_points()) absent_coverage = {} for channel in sorted(set(range(1, 21)) - set(cleared)): channel_events = [e for e in events if e["path"] == "/measure" and e["request"]["channel"] == channel] points = {(e["request"]["position"]["x"], e["request"]["position"]["y"]) for e in channel_events if e["response"]["measure_result"] == "no_signal"} absent_coverage[str(channel)] = { "grid_points_measured": len(points & grid), "required_grid_points": len(grid), "all_no_signal": all(e["response"]["measure_result"] == "no_signal" for e in channel_events), "complete": grid <= points, } checks["absent_channels_fully_covered"] = all( item["complete"] and item["all_no_signal"] for item in absent_coverage.values()) # Official attachment 2 section 4.1: accumulate virtual time in microseconds. # Round each move to the nearest microsecond, then add integer action costs. position, channel, virtual_us = (0.0, 0.0), 1, 0 distance_total, switch_total, move_total_us = 0.0, 0, 0 rows = [] previous_recorded = 0.0 max_local_gap = 0.0 for i, event in enumerate(events): request, response, path = event["request"], event["response"], event["path"] distance, switches, action_s = 0.0, 0, 0 if path in ("/measure", "/clear"): target = (request["position"]["x"], request["position"]["y"]) distance = math.dist(position, target) position = target if path == "/measure": switches = int(channel != request["channel"]) channel = request["channel"] action_s = 5 else: action_s = 5 if response["clear_result"] == "success" else 3 move_us = round(distance / 5 * 1_000_000) virtual_us += move_us + (switches + action_s) * 1_000_000 distance_total += distance switch_total += switches move_total_us += move_us recorded = response["virtual_time_s"] gap_us = round(recorded * 1_000_000) - virtual_us max_local_gap = max(max_local_gap, abs(recorded - previous_recorded - distance/5 - switches - action_s)) previous_recorded = recorded rows.append({"index": i, "path": path, "request_id": request["request_id"], "channel": request.get("channel", ""), "x": position[0], "y": position[1], "result": response.get("measure_result", response.get("clear_result", "")), "svd_deg": response.get("svd_deg", ""), "distance_m": distance, "move_us": move_us, "switch_s": switches, "action_s": action_s, "recomputed_us": virtual_us, "recorded_s": recorded, "gap_us": gap_us}) checks["all_timestamps_match_microsecond_recomputation"] = all(r["gap_us"] == 0 for r in rows) real_s = (events[-1]["response"]["real_timestamp_ms"] - events[0]["response"]["real_timestamp_ms"]) / 1000 checks["within_real_and_virtual_limits"] = ( real_s < events[0]["response"]["remaining_real_duration_s"] and virtual_us / 1_000_000 < events[0]["response"]["max_virtual_duration_s"]) # Exact action replay independently checks decisions using only recorded responses. transport = StrictPlayback(events) config = ProblemConfig() client = SimulatorClient(original["robot_id"], transport=transport, config=config, clock=lambda: 0.0) loop = (AuditedSevenBatch if strategy == "seven_batch" else AuditedLoop)(client, config) replay_summary = loop.run() checks["deterministic_action_replay"] = ( transport.index == len(events) and replay_summary["completed"] and all(replay_summary[k] == original[k] for k in ("stop_reason", "cleared_channels", "absent_channels", "measures", "clears"))) checks["all_clear_certificates_valid"] = ( len(loop.certificates) == len(clears) and all(c["certified"] and c["radius_m"] <= c["limit_m"] < 20 and c["submitted_max_vertex_distance_m"] <= c["limit_m"] for c in loop.certificates)) raw_timing = recompute_virtual_time(events) report = { "passed": all(checks.values()), "source": "official_practice", "strategy": strategy, "case_code": official["case_code"], "checks": checks, "official_jammer_count": official["jammer_count"], "cleared_channels": cleared, "absent_channel_coverage": absent_coverage, "accepted_requests_including_enter_exit": len(events), "measures": len([e for e in events if e["path"] == "/measure"]), "clear_attempts": len(clears), "clear_successes": len(successes), "real_elapsed_server_s": real_s, "real_elapsed_local_s": events[-1]["logged_at_unix_s"] - events[0]["logged_at_unix_s"], "virtual_time_s": virtual_us / 1_000_000, "last_clear_virtual_time_s": successes[-1]["response"]["virtual_time_s"], "move_distance_m": distance_total, "move_time_s": move_total_us / 1_000_000, "switch_time_s": switch_total, "measurement_time_s": sum(r["action_s"] for r in rows if r["path"] == "/measure"), "clear_time_s": sum(r["action_s"] for r in rows if r["path"] == "/clear"), "timing_audit": { "original_unquantized_mismatch_count": len(raw_timing["mismatches"]), "original_unquantized_final_gap_s": previous_recorded - raw_timing["final_virtual_time_s"], "max_single_action_unquantized_gap_s": max_local_gap, "microsecond_mismatch_count": sum(r["gap_us"] != 0 for r in rows), "explanation": "原复算累计未量化浮点耗时；按动作移动耗时取最近整数微秒后，所有官方时刻完全一致。", }, "clear_certificates": loop.certificates, "limits": "仅验证当前官方演练案例；不证明所有案例或时间最优性。官方加密日志保持原样，未解密。", } with (folder / "actions.csv").open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) save_json(folder / "verification.json", report) save_json(folder / "replay_summary.json", replay_summary) manifest = {p.name: {"bytes": p.stat().st_size, "sha256": hashlib.sha256(p.read_bytes()).hexdigest()} for p in sorted(folder.iterdir()) if p.is_file() and p.name != "manifest.json"} save_json(folder / "manifest.json", manifest) return report if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("folder", type=Path) args = parser.parse_args() result = audit(args.folder) print(json.dumps({"passed": result["passed"], "case_code": result["case_code"], "checks": result["checks"]}, ensure_ascii=False, indent=2)) raise SystemExit(0 if result["passed"] else 1)