12from __future__ import annotations34import argparse5import json6import os7import queue8import re9import signal10import subprocess11import sys12import threading13import time14from collections.abc import Callable, Sequence15from dataclasses import dataclass16from datetime import datetime, timezone17from pathlib import Path18from typing import TextIO192021@dataclass(frozen=True)22class Scenario:23 id: str24 name: str252627SCENARIOS = (28 Scenario("1", "equal versions, both unlocked, both manifest-v5"),29 Scenario("2", "unequal data versions, both unlocked, both manifest-v5"),30 Scenario("3", "equal versions, one manifest-v5 and one manifest-v3"),31 Scenario("4", "unequal data versions, one manifest-v5 and one manifest-v3"),32 Scenario("9", "manifest negotiation: no peer sends SYNC_DEVICE_MANIFEST unilaterally"),33 Scenario("13", "vault id mismatch stops cleanly"),34 Scenario("5", "browser locked (spool active), android unlocked, unequal versions"),35 Scenario("6", "android locked (listener running), browser unlocked"),36 Scenario("8", "both sides locked"),37 Scenario("10", "process death mid-session, next round completes cleanly"),38 Scenario("12", "manifest with a device vouched too deeply is rejected"),39 Scenario("14", "manifest issued outside the freshness window is rejected"),40 Scenario("15", "10,000-entry vault with padded frames"),41 Scenario("16.3.1", "stale issuedAt → rejected"),42 Scenario("16.3.2", "future issuedAt → rejected"),43 Scenario("16.3.3", "vouchDepth above the absolute bound → rejected"),44 Scenario("16.3.4", "device count above the bound → rejected"),45 Scenario("16.3.5", "malformed device key → rejected"),46 Scenario("16.3.6", "oversized issuer id → rejected"),47 Scenario("16.3.7", "signature from an untrusted key → rejected"),48 Scenario("16.3.8", "valid manifest (control) → accepted"),49)5051COVERAGE_EXCLUSIONS = (52 {53 "id": "7",54 "disposition": "external_evidence_required",55 "reason": "cold-locked Android requires unit and real-device evidence",56 },57 {58 "id": "11",59 "disposition": "external_evidence_required",60 "reason": "revocation then re-vouch requires storage-adapter evidence",61 },62)6364RESULT_PATTERN = re.compile(r"^\s*✓\s+(\S+)\s+(.+?)\s+\(([0-9.]+)s\)\s*$")656667def utc_now() -> str:68 return datetime.now(timezone.utc).isoformat()697071def emit_json(event: dict[str, object]) -> None:72 print(json.dumps(event, separators=(",", ":")), flush=True)737475def terminate_process_group(process: subprocess.Popen[str]) -> None:76 if process.poll() is not None:77 return78 try:79 os.killpg(process.pid, signal.SIGTERM)80 process.wait(timeout=0.2)81 except ProcessLookupError:82 return83 except subprocess.TimeoutExpired:84 try:85 os.killpg(process.pid, signal.SIGKILL)86 except ProcessLookupError:87 pass88 process.wait()899091def write_summary(path: Path, summary: dict[str, object]) -> None:92 path.parent.mkdir(parents=True, exist_ok=True)93 temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")94 temporary.write_text(f"{json.dumps(summary, indent=2)}\n", encoding="utf-8")95 os.replace(temporary, path)969798def run_gate(99 command: Sequence[str],100 summary_path: Path,101 *,102 scenario_timeout_seconds: float,103 total_timeout_seconds: float,104 emit: Callable[[dict[str, object]], None] = emit_json,105 raw_output: TextIO = sys.stderr,106) -> int:107 started_at = utc_now()108 gate_started = time.monotonic()109 records: list[dict[str, object]] = [110 {111 "id": scenario.id,112 "name": scenario.name,113 "status": "pending",114 "startedAt": None,115 "finishedAt": None,116 }117 for scenario in SCENARIOS118 ]119 current_index = 0120 current_started = gate_started121 runner_status = "start_failed"122 runner_exit_code: int | None = None123124 def start_scenario(index: int) -> None:125 nonlocal current_started126 current_started = time.monotonic()127 records[index]["startedAt"] = utc_now()128 emit(129 {130 "event": "scenario_start",131 "id": SCENARIOS[index].id,132 "name": SCENARIOS[index].name,133 "timeoutSeconds": scenario_timeout_seconds,134 }135 )136137 def finish_scenario(138 index: int,139 status: str,140 reason: str | None = None,141 runner_seconds: float | None = None,142 ) -> None:143 record = records[index]144 record["status"] = status145 record["finishedAt"] = utc_now()146 if record["startedAt"] is not None:147 record["elapsedSeconds"] = round(time.monotonic() - current_started, 3)148 if reason is not None:149 record["reason"] = reason150 if runner_seconds is not None:151 record["runnerReportedSeconds"] = runner_seconds152 event: dict[str, object] = {153 "event": "scenario_finish",154 "id": record["id"],155 "name": record["name"],156 "status": status,157 "started": record["startedAt"] is not None,158 }159 if reason is not None:160 event["reason"] = reason161 emit(event)162163 def fail_unreported(reason: str) -> None:164 for index in range(current_index, len(records)):165 if records[index]["status"] == "pending":166 finish_scenario(index, "fail", reason)167168 process: subprocess.Popen[str] | None = None169 reader_thread: threading.Thread | None = None170 try:171 process = subprocess.Popen(172 list(command),173 stdout=subprocess.PIPE,174 stderr=subprocess.STDOUT,175 text=True,176 bufsize=1,177 start_new_session=True,178 )179 if process.stdout is None:180 raise RuntimeError("runner stdout pipe was not created")181182 output_queue: queue.Queue[str | None] = queue.Queue()183184 def read_output() -> None:185 for line in process.stdout:186 output_queue.put(line)187 output_queue.put(None)188189 reader_thread = threading.Thread(target=read_output, daemon=True)190 reader_thread.start()191 runner_status = "running"192 start_scenario(current_index)193194 while True:195 now = time.monotonic()196 total_remaining = total_timeout_seconds - (now - gate_started)197 scenario_remaining = (198 scenario_timeout_seconds - (now - current_started)199 if current_index < len(records)200 else total_remaining201 )202 wait_seconds = min(total_remaining, scenario_remaining)203 if wait_seconds <= 0:204 if current_index < len(records):205 reason = "total_timeout" if total_remaining <= 0 else "scenario_timeout"206 finish_scenario(current_index, "timeout", reason)207 current_index += 1208 fail_unreported(f"not_started_after_{reason}")209 runner_status = "timeout"210 terminate_process_group(process)211 break212213 try:214 line = output_queue.get(timeout=wait_seconds)215 except queue.Empty:216 continue217218 if line is None:219 runner_exit_code = process.wait()220 runner_status = "pass" if runner_exit_code == 0 else f"exit_{runner_exit_code}"221 if current_index < len(records):222 fail_unreported(223 "runner_exited_before_result"224 if runner_exit_code == 0225 else f"runner_exit_{runner_exit_code}"226 )227 break228229 raw_output.write(line)230 raw_output.flush()231 match = RESULT_PATTERN.match(line)232 if match is None:233 continue234235 reported_id, reported_name, reported_seconds = match.groups()236 if current_index >= len(SCENARIOS):237 runner_status = "unexpected_result"238 terminate_process_group(process)239 break240241 expected = SCENARIOS[current_index]242 if reported_id != expected.id or reported_name != expected.name:243 finish_scenario(244 current_index,245 "fail",246 f"unexpected_result:{reported_id}:{reported_name}",247 )248 current_index += 1249 fail_unreported("not_started_after_unexpected_result")250 runner_status = "unexpected_result"251 terminate_process_group(process)252 break253254 finish_scenario(current_index, "pass", runner_seconds=float(reported_seconds))255 current_index += 1256 if current_index < len(SCENARIOS):257 start_scenario(current_index)258 except OSError as error:259 fail_unreported(f"runner_start_error:{error.__class__.__name__}")260 runner_status = "start_failed"261 finally:262 if process is not None and process.poll() is None:263 terminate_process_group(process)264 if reader_thread is not None:265 reader_thread.join(timeout=0.2)266 if process is not None and process.stdout is not None:267 process.stdout.close()268269 counts = {270 status: sum(record["status"] == status for record in records)271 for status in ("pass", "fail", "timeout")272 }273 overall_status = (274 "pass"275 if runner_status == "pass" and counts == {"pass": len(records), "fail": 0, "timeout": 0}276 else "fail"277 )278 summary: dict[str, object] = {279 "schemaVersion": 1,280 "startedAt": started_at,281 "finishedAt": utc_now(),282 "overallStatus": overall_status,283 "runnerStatus": runner_status,284 "runnerExitCode": runner_exit_code,285 "scenarioTimeoutSeconds": scenario_timeout_seconds,286 "totalTimeoutSeconds": total_timeout_seconds,287 "counts": counts,288 "scenarios": records,289 "coverageExclusions": list(COVERAGE_EXCLUSIONS),290 }291 write_summary(summary_path, summary)292 emit(293 {294 "event": "summary",295 "overallStatus": overall_status,296 "runnerStatus": runner_status,297 "counts": counts,298 "path": str(summary_path),299 }300 )301 return 0 if overall_status == "pass" else 1302303304def positive_seconds(value: str) -> float:305 seconds = float(value)306 if seconds <= 0:307 raise argparse.ArgumentTypeError("must be greater than zero")308 return seconds309310311def main() -> int:312 script_dir = Path(__file__).resolve().parent313 parser = argparse.ArgumentParser(description="Run the bounded cross-client sync matrix")314 parser.add_argument(315 "--scenario-timeout-seconds",316 type=positive_seconds,317 default=300.0,318 help="deadline for each process scenario (default: 300)",319 )320 parser.add_argument(321 "--total-timeout-seconds",322 type=positive_seconds,323 default=1200.0,324 help="deadline for the complete process matrix (default: 1200)",325 )326 parser.add_argument(327 "--summary",328 type=Path,329 default=script_dir.parent / "build/reports/cross-client-sync-summary.json",330 help="machine-readable result path",331 )332 arguments = parser.parse_args()333 runner = script_dir.parent.parent / "peach-browser/scripts/check-cross-client-sync.mjs"334 return run_gate(335 ["node", str(runner)],336 arguments.summary,337 scenario_timeout_seconds=arguments.scenario_timeout_seconds,338 total_timeout_seconds=arguments.total_timeout_seconds,339 )340341342if __name__ == "__main__":343 raise SystemExit(main())