#!/usr/bin/env python3 """Validate and summarize one Feature #149 six-run cohort.""" from __future__ import annotations import argparse import json import statistics from datetime import datetime, timezone from pathlib import Path from typing import Any REQUIRED_ARTIFACTS = { "run-meta.json", "run-summary.json", "input.cir", "result.ccr", "report.json", "vrps.csv", "vaps.csv", "stage-timing.json", "process-time.txt", } def load_jsonl(path: Path) -> list[dict[str, Any]]: records: list[dict[str, Any]] = [] for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): if line.strip(): value = json.loads(line) if not isinstance(value, dict): raise ValueError(f"{path}:{line_number} is not an object") records.append(value) return records def require_success(records: list[dict[str, Any]], package_root: Path) -> list[dict[str, Any]]: if len(records) != 6: raise ValueError(f"expected 6 records (snapshot + 5 delta), found {len(records)}") for index, record in enumerate(records): expected_mode = "snapshot" if index == 0 else "delta" if record.get("status") != "success": raise ValueError(f"run {index + 1} status is {record.get('status')!r}") if record.get("syncMode") != expected_mode: raise ValueError( f"run {index + 1} syncMode is {record.get('syncMode')!r}, expected {expected_mode!r}" ) run_dir = package_root / "runs" / f"run_{int(record['maxRunIndexAfter']):04d}" missing = sorted(name for name in REQUIRED_ARTIFACTS if not (run_dir / name).is_file()) record["runDir"] = str(run_dir) record["artifactsComplete"] = not missing record["missingArtifacts"] = missing if missing: raise ValueError(f"run {index + 1} missing required artifacts: {', '.join(missing)}") return records def percentile(values: list[float], fraction: float) -> float | None: if not values: return None ordered = sorted(values) position = (len(ordered) - 1) * fraction lower = int(position) upper = min(lower + 1, len(ordered) - 1) return ordered[lower] + (ordered[upper] - ordered[lower]) * (position - lower) def stats(values: list[int | float | None]) -> dict[str, float | None]: nums = [float(value) for value in values if isinstance(value, (int, float))] if not nums: return {"count": 0, "min": None, "median": None, "p90": None, "max": None, "mean": None} return { "count": len(nums), "min": min(nums), "median": statistics.median(nums), "p90": percentile(nums, 0.9), "max": max(nums), "mean": statistics.fmean(nums), } def iso(epoch: Any) -> str | None: if not isinstance(epoch, (int, float)): return None return datetime.fromtimestamp(epoch, timezone.utc).isoformat().replace("+00:00", "Z") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--cohort-id", required=True) parser.add_argument("--cohort-disk", required=True, choices=("system", "data")) parser.add_argument("--cohort-profile", required=True, choices=("prefetch-pp-object", "pp-object")) parser.add_argument("--package-root", required=True, type=Path) parser.add_argument("--summary-jsonl", required=True, type=Path) parser.add_argument("--out-dir", required=True, type=Path) args = parser.parse_args() records = require_success(load_jsonl(args.summary_jsonl), args.package_root) delta = records[1:] elapsed_seconds = records[-1]["completedEpoch"] - records[0]["actualStartEpoch"] report = { "feature": "149", "createdAtUtc": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), "cohortId": args.cohort_id, "disk": args.cohort_disk, "profile": args.cohort_profile, "packageRoot": str(args.package_root), "runCount": len(records), "successCount": len(records), "artifactStatus": "complete", "elapsedSeconds": elapsed_seconds, "actualStartedAtUtc": iso(records[0].get("actualStartEpoch")), "completedAtUtc": iso(records[-1].get("completedEpoch")), "maxRssKb": max(int(record["maxRssKb"]) for record in records if isinstance(record.get("maxRssKb"), (int, float))), "snapshot": {"wallMs": records[0].get("wallMs"), "maxRssKb": records[0].get("maxRssKb")}, "deltaWallMs": stats([record.get("wallMs") for record in delta]), "stableDeltaWallMs": stats([record.get("wallMs") for record in delta[1:]]), "latestCounts": { key: records[-1].get(key) for key in ("vrps", "vaps", "publicationPoints", "warnings") }, "runs": records, } args.out_dir.mkdir(parents=True, exist_ok=True) json_path = args.out_dir / "cohort-summary.json" markdown_path = args.out_dir / "cohort-summary.md" json_path.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8") latest = report["latestCounts"] markdown_path.write_text( "\n".join( [ f"# Feature #149 cohort: {args.cohort_id}", "", f"- 磁盘：`{args.cohort_disk}`", f"- profile：`{args.cohort_profile}`", f"- 状态：`success`，6/6 run 成功，必要产物完整", f"- 总执行耗时：`{elapsed_seconds}s`", f"- 最高 RSS：`{report['maxRssKb']} KiB`", f"- snapshot wall：`{records[0].get('wallMs')} ms`", f"- delta median / p90：`{report['deltaWallMs']['median']} ms` / `{report['deltaWallMs']['p90']} ms`", f"- 最新产物：VRP `{latest['vrps']}`、VAP `{latest['vaps']}`、PP `{latest['publicationPoints']}`、warnings `{latest['warnings']}`", "", "| run | mode | wall ms | max RSS KiB | VRP | VAP | PP | warnings | artifacts |", "| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | --- |", *[ "| {run} | {mode} | {wall} | {rss} | {vrps} | {vaps} | {pps} | {warnings} | {artifacts} |".format( run=index + 1, mode=record.get("syncMode"), wall=record.get("wallMs"), rss=record.get("maxRssKb"), vrps=record.get("vrps"), vaps=record.get("vaps"), pps=record.get("publicationPoints"), warnings=record.get("warnings"), artifacts="complete" if record.get("artifactsComplete") else "missing", ) for index, record in enumerate(records) ], "", ] ), encoding="utf-8", ) print(json.dumps({"cohortSummaryJson": str(json_path), "cohortSummaryMarkdown": str(markdown_path), "elapsedSeconds": elapsed_seconds, "maxRssKb": report["maxRssKb"]}, ensure_ascii=False)) if __name__ == "__main__": main()