forge: step 10 — add CI safety checks

scripts/parse-size-report.py

@@ -98,6 +98,102 @@ def add_delta(entry, baseline_by_env):
     entry["delta"] = delta
 
 
+def metric_limit(config, metric, level):
+    value = config.get(metric, {})
+    if isinstance(value, dict):
+        return value.get(f"{level}_delta", value.get(level))
+    return None
+
+
+def env_threshold_config(thresholds, env):
+    config = dict(thresholds.get("defaults", {}))
+    config.update(thresholds.get("environments", {}).get(env, {}))
+    return config
+
+
+def evaluate_thresholds(summary, thresholds):
+    checks = []
+    status = "ok"
+    for entry in summary["environments"]:
+        delta = entry.get("delta")
+        if not delta:
+            checks.append({"env": entry["env"], "status": "not_evaluated", "reason": "no baseline delta"})
+            continue
+
+        config = env_threshold_config(thresholds, entry["env"])
+        for metric in ("flash_used", "ram_used"):
+            value = delta.get(metric)
+            if value is None:
+                checks.append({"env": entry["env"], "metric": metric, "status": "not_evaluated", "reason": "missing delta"})
+                continue
+            warn_limit = metric_limit(config, metric, "warn")
+            fail_limit = metric_limit(config, metric, "fail")
+            check_status = "ok"
+            if fail_limit is not None and value > fail_limit:
+                check_status = "fail"
+                status = "fail"
+            elif warn_limit is not None and value > warn_limit:
+                check_status = "warn"
+                if status == "ok":
+                    status = "warn"
+            checks.append({
+                "env": entry["env"],
+                "metric": metric,
+                "delta": value,
+                "warn_delta": warn_limit,
+                "fail_delta": fail_limit,
+                "status": check_status,
+            })
+
+        artifact_config = config.get("artifact_bytes_by_kind", {})
+        for kind, value in delta.get("artifact_bytes_by_kind", {}).items():
+            limits = artifact_config.get(kind, {})
+            warn_limit = limits.get("warn_delta", limits.get("warn"))
+            fail_limit = limits.get("fail_delta", limits.get("fail"))
+            check_status = "ok"
+            if fail_limit is not None and value > fail_limit:
+                check_status = "fail"
+                status = "fail"
+            elif warn_limit is not None and value > warn_limit:
+                check_status = "warn"
+                if status == "ok":
+                    status = "warn"
+            checks.append({
+                "env": entry["env"],
+                "metric": f"artifact_bytes_by_kind.{kind}",
+                "delta": value,
+                "warn_delta": warn_limit,
+                "fail_delta": fail_limit,
+                "status": check_status,
+            })
+
+    return {"status": status, "checks": checks}
+
+
+def report_lines(summary):
+    lines = [f"Size summary ({summary['mode']})"]
+    for entry in summary["environments"]:
+        lines.append(f"- {entry['env']}:")
+        lines.append(f"  RAM: {entry['ram_used']} / {entry['ram_total']} bytes")
+        lines.append(f"  Flash: {entry['flash_used']} / {entry['flash_total']} bytes")
+        if entry.get("delta"):
+            delta = entry["delta"]
+            lines.append(f"  Delta: RAM {delta.get('ram_used')} bytes, flash {delta.get('flash_used')} bytes")
+        for name, size in sorted(entry.get("artifact_bytes", {}).items()):
+            lines.append(f"  Artifact: {name} = {size} bytes")
+    thresholds = summary.get("thresholds")
+    if thresholds:
+        lines.append(f"Threshold status: {thresholds['status']}")
+        for check in thresholds["checks"]:
+            if check["status"] == "ok":
+                continue
+            reason = check.get("reason", "")
+            metric = check.get("metric", "all")
+            delta = check.get("delta", "n/a")
+            lines.append(f"- {check['env']} {metric}: {check['status']} delta={delta} {reason}".rstrip())
+    return lines
+
+
 def build_summary(args):
     log_dir = args.logs
     artifact_dir = args.artifacts
@@ -136,16 +232,29 @@ def main():
     parser.add_argument("--env", action="append", help="Environment name to include; may be passed more than once")
     parser.add_argument("--baseline", action="store_true", help="Mark output as a baseline summary")
     parser.add_argument("--compare", type=Path, help="Baseline JSON to compare against")
+    parser.add_argument("--thresholds", type=Path, help="JSON file with warn/fail size delta thresholds")
+    parser.add_argument("--enforce-thresholds", action="store_true", help="Exit nonzero when threshold status is fail")
+    parser.add_argument("--report", type=Path, help="Write a human-readable size report to this path")
     parser.add_argument("--output", type=Path, help="Write JSON summary to this path")
     args = parser.parse_args()
 
     summary = build_summary(args)
+    if args.thresholds:
+        thresholds = json.loads(args.thresholds.read_text())
+        summary["thresholds"] = evaluate_thresholds(summary, thresholds)
+
     text = json.dumps(summary, indent=2, sort_keys=True)
     if args.output:
         args.output.parent.mkdir(parents=True, exist_ok=True)
         args.output.write_text(text + "\n")
+    if args.report:
+        args.report.parent.mkdir(parents=True, exist_ok=True)
+        args.report.write_text("\n".join(report_lines(summary)) + "\n")
     print(text)
 
+    if args.enforce_thresholds and summary.get("thresholds", {}).get("status") == "fail":
+        raise SystemExit(1)
+
 
 if __name__ == "__main__":
     main()