Expand intel coverage and refresh monitoring

scripts/lab/render.py

@@ -1263,17 +1263,23 @@ def render_run(run: Dict[str, Any]) -> Dict[str, str]:
     return {"bundle_dir": str(run_dir), "report_md": str(report_md), "report_html": str(report_html), "timeline": str(timeline_path)}
 
 
-def render_dashboard() -> Dict[str, str]:
+def render_dashboard(
+    *,
+    advisory_records: List[Dict[str, Any]] | None = None,
+    runs: List[Dict[str, Any]] | None = None,
+    source_map_data: Dict[str, Any] | None = None,
+    repro_map_data: Dict[str, Any] | None = None,
+) -> Dict[str, str]:
     ensure_dir(DASHBOARD_DIR)
-    advisory_records = load_json_dir(ADVISORIES_DIR)
-    runs = load_json_dir(RUNS_DIR)
+    advisory_records = advisory_records if advisory_records is not None else load_json_dir(ADVISORIES_DIR)
+    runs = runs if runs is not None else load_json_dir(RUNS_DIR)
     run_summary = read_json(ROOT / "08-threat-intel" / "generated" / "run-summary.json", default={}) or {}
     source_health = read_json(ROOT / "08-threat-intel" / "generated" / "source-health.json", default={}) or {}
     alerts = read_json(ROOT / "08-threat-intel" / "generated" / "alerts.json", default=[]) or []
     monitor_summary = read_json(ROOT / "08-threat-intel" / "generated" / "monitor-summary.json", default={}) or {}
     source_catalog_audit = read_json(ROOT / "08-threat-intel" / "generated" / "source-catalog-audit.json", default={}) or {}
-    source_map = read_yaml(SOURCE_MAP_PATH, default={}) or {}
-    repro_map = read_yaml(REPRO_MAP_PATH, default={}) or {}
+    source_map = source_map_data if source_map_data is not None else (read_yaml(SOURCE_MAP_PATH, default={}) or {})
+    repro_map = repro_map_data if repro_map_data is not None else (read_yaml(REPRO_MAP_PATH, default={}) or {})
     source_system_map = {item["system_id"]: item for item in source_map.get("systems", []) if item.get("system_id")}
     merged_advisories = _merge_latest_advisories(advisory_records, runs, source_system_map)
     advisory_map = {item["canonical_id"]: item for item in merged_advisories if item.get("canonical_id")}

scripts/lab/repro.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from functools import lru_cache
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -25,10 +26,12 @@ FAMILY_KEYWORDS = {
 }
 
 
+@lru_cache(maxsize=1)
 def load_repro_map() -> Dict[str, Any]:
     return read_yaml(REPRO_MAP_PATH, default={"systems": []}) or {"systems": []}
 
 
+@lru_cache(maxsize=1)
 def load_profiles() -> Dict[str, Dict[str, Any]]:
     profiles: Dict[str, Dict[str, Any]] = {}
     if not REPRO_PROFILES_DIR.exists():

scripts/lab/utils.py

@@ -33,7 +33,10 @@ def read_json(path: Path, default: Any = None) -> Any:
 
 def write_json(path: Path, data: Any) -> None:
     ensure_dir(path.parent)
-    path.write_text(json.dumps(data, indent=2, ensure_ascii=True, sort_keys=False) + "\n", encoding="utf-8")
+    content = json.dumps(data, indent=2, ensure_ascii=True, sort_keys=False) + "\n"
+    if path.exists() and path.read_text(encoding="utf-8") == content:
+        return
+    path.write_text(content, encoding="utf-8")
 
 
 def read_yaml(path: Path, default: Any = None) -> Any:
@@ -51,7 +54,10 @@ def write_yaml(path: Path, data: Any) -> None:
 
 def write_text(path: Path, content: str) -> None:
     ensure_dir(path.parent)
-    path.write_text(content.rstrip() + "\n", encoding="utf-8")
+    rendered = content.rstrip() + "\n"
+    if path.exists() and path.read_text(encoding="utf-8") == rendered:
+        return
+    path.write_text(rendered, encoding="utf-8")
 
 
 def load_json_dir(path: Path) -> List[Dict[str, Any]]: