#!/usr/bin/env python3 """ Semantic Impact Report Generator v3 — Three-Layer Architecture Layer 1 — Cargo.toml Dependency Graph (auto) Scans workspace Cargo.toml files to build crate dependency DAG. Assigns blast radius automatically: leaf crates = LOW, hub crates = HIGH. Layer 2 — git diff API Analysis Parses git diff output for public API changes (+pub fn, -pub struct, etc.). Classifies each change as ADD (safe), REMOVE (breaking), CHANGE (potentially breaking). Layer 3 — cargo-semver-checks Integration (optional, --semver) Runs cargo semver-checks on changed crates for definitive semver analysis. Usage: # Pre-commit (staged changes) python3 .githooks/generate_semantic_report.py --mode staged # Post-commit (last commit) python3 .githooks/generate_semantic_report.py --mode last # With semver check (slow) python3 .githooks/generate_semantic_report.py --mode staged --semver # Skip Layer 2 API analysis python3 .githooks/generate_semantic_report.py --mode staged --no-api Exit codes: 0 = safe (LOW/MEDIUM impact) 1 = HIGH risk found — review required Dependencies: Python 3.8+, tomllib (stdlib 3.11+) or tomli (fallback) """ import json import os import re import subprocess import sys import tempfile from collections import defaultdict from pathlib import Path # ───────────────────────────────────────────────────────────────────────────── # CONFIG # ───────────────────────────────────────────────────────────────────────────── REPO_ROOT = Path(__file__).resolve().parent.parent WORKSPACE_CARGO = REPO_ROOT / "Cargo.toml" MANIFEST_PATH = REPO_ROOT / "project_manifest.json" LOG_DIR = REPO_ROOT / "logs" # ───────────────────────────────────────────────────────────────────────────── # TOML parser — compatible with Python 3.8+ # ───────────────────────────────────────────────────────────────────────────── try: import tomllib except ImportError: import tomli as tomllib # ============================================================================ # LAYER 1: Cargo.toml Dependency Graph # ============================================================================ def parse_cargo_toml(path: Path) -> dict: """Parse a Cargo.toml file and return as dict.""" try: with open(path, "rb") as f: return tomllib.load(f) except Exception as e: print(f" [WARN] Failed to parse {path}: {e}", file=sys.stderr) return {} def build_dependency_graph() -> dict: """ Build the full workspace dependency graph from Cargo.toml files. Returns: { "crates": { "hcie-protocol": { "path": "crates/hcie-protocol", "deps": [], "dependents": ["hcie-color", "hcie-blend", ...] } }, "all_files": [ "crates/hcie-protocol/src/lib.rs", ... ], "file_to_crate": { "crates/hcie-protocol/src/lib.rs": "hcie-protocol", ... }, "blast_radius": { "hcie-protocol": "HIGH", "hcie-color": "LOW", ... }, "errors": [] } """ workspace = parse_cargo_toml(WORKSPACE_CARGO) members = workspace.get("workspace", {}).get("members", []) errors = [] # 1. Discover all crate directories crate_paths = {} for member in members: member_path = REPO_ROOT / member cargo_path = member_path / "Cargo.toml" if cargo_path.exists(): data = parse_cargo_toml(cargo_path) name = data.get("package", {}).get("name", member.rsplit("/", 1)[-1]) crate_paths[name] = str(member_path.relative_to(REPO_ROOT)) else: errors.append(f"Member path missing Cargo.toml: {member}") if not crate_paths: return {"crates": {}, "all_files": [], "file_to_crate": {}, "blast_radius": {}, "errors": ["No crates found in workspace"]} # 2. Parse each crate's dependencies crates = {} for name, rel_path in crate_paths.items(): cargo_path = REPO_ROOT / rel_path / "Cargo.toml" data = parse_cargo_toml(cargo_path) deps = data.get("dependencies", {}) internal_deps = [] for dep_name, dep_info in deps.items(): if isinstance(dep_info, dict) and "path" in dep_info: internal_deps.append(dep_name) crates[name] = { "path": rel_path, "deps": internal_deps, "dependents": [], } # 3. Build reverse dependency list (who depends on whom) for name, info in crates.items(): for dep_name in info["deps"]: if dep_name in crates: crates[dep_name]["dependents"].append(name) # 4. Discover all source files — .rs per crate, plus GUI files (src/ directory for Svelte/TS/CSS) all_files = [] file_to_crate = {} for name, info in crates.items(): src_dir = REPO_ROOT / info["path"] / "src" if src_dir.exists(): for rs_file in sorted(src_dir.rglob("*.rs")): rel = rs_file.relative_to(REPO_ROOT).as_posix() all_files.append(rel) file_to_crate[rel] = name # 4b. Add GUI source files (Svelte, TypeScript, CSS) as virtual "gui" crates gui_configs = [ ("gui-svelte", "src", ["*.svelte", "*.ts", "*.css"]), ("gui-tauri", "src-tauri/src", ["*.rs"]), # src-tauri .rs already covered but re-scan for completeness ] for virtual_crate, rel_dir, globs in gui_configs: scan_dir = REPO_ROOT / rel_dir if scan_dir.exists(): for g in globs: for f in sorted(scan_dir.rglob(g)): rel = f.relative_to(REPO_ROOT).as_posix() if rel not in file_to_crate: all_files.append(rel) file_to_crate[rel] = virtual_crate # Register virtual crate in graph for blast radius reporting if virtual_crate not in crates: crates[virtual_crate] = { "path": rel_dir, "deps": [], "dependents": [], } # 5. Auto-assign blast radius based on dependency hub weight blast_radius = {} for name, info in crates.items(): num_dependents = len(info["dependents"]) if num_dependents >= 5: blast_radius[name] = "HIGH" elif num_dependents >= 2: blast_radius[name] = "MEDIUM" elif num_dependents == 0 and any(name in c["deps"] for c in crates.values()): blast_radius[name] = "LOW" else: blast_radius[name] = "LOW" # 6. Set blast radius for GUI virtual crates (always HIGH — UI changes affect everything) # NOTE: Must be after step 5 which reinitializes blast_radius blast_radius["gui-svelte"] = "HIGH" blast_radius["gui-tauri"] = "HIGH" # 7. Validate existing manifest entries for file-level info manifest_annotations = {} if MANIFEST_PATH.exists(): try: with open(MANIFEST_PATH) as f: manifest = json.load(f) for entry in manifest.get("files", []): fp = entry["file_path"] manifest_annotations[fp] = { "description": entry.get("description", ""), "blast_radius": entry.get("blast_radius", "LOW"), "semantic_dependencies": entry.get("semantic_dependencies", []), } except Exception as e: errors.append(f"Failed to read manifest: {e}") return { "crates": crates, "all_files": all_files, "file_to_crate": file_to_crate, "blast_radius": blast_radius, "manifest_annotations": manifest_annotations, "errors": errors, } def get_file_blast_radius(file_path: str, graph: dict) -> str: """Determine blast radius for a file based on crate-level + manifest annotations.""" # Check manifest first (manual override) ann = graph.get("manifest_annotations", {}).get(file_path, {}) if ann and ann.get("blast_radius", "LOW") != "LOW": return ann["blast_radius"] # Auto-detect from crate-level crate = graph.get("file_to_crate", {}).get(file_path) if crate: return graph.get("blast_radius", {}).get(crate, "LOW") return "LOW" def get_file_crate(file_path: str, graph: dict) -> str: """Return crate name for a given file path.""" return graph.get("file_to_crate", {}).get(file_path, "") # ============================================================================ # LAYER 2: git diff API Analysis # ============================================================================ # Patterns for pub API changes in git diff output PUB_ADD_PATTERNS = [ re.compile(r'^\+pub\s+(fn|struct|enum|trait|type|const|static|mod|use|macro_rules!)\s+(\w+)'), re.compile(r'^\+pub\s+(unsafe\s+)?fn\s+(\w+)'), re.compile(r'^\+pub\s+(unsafe\s+)?trait\s+(\w+)'), ] PUB_REMOVE_PATTERNS = [ re.compile(r'^\-pub\s+(fn|struct|enum|trait|type|const|static|mod|use|macro_rules!)\s+(\w+)'), re.compile(r'^\-pub\s+(unsafe\s+)?fn\s+(\w+)'), re.compile(r'^\-pub\s+(unsafe\s+)?trait\s+(\w+)'), ] # Detect hunk context headers HUNK_CONTEXT_PATTERN = re.compile( r'@@\s+-(\d+),?\d*\s+\+(\d+),?\d*\s+@@\s*(.*)' ) PUB_CONTEXT_PATTERNS = [ re.compile(r'pub\s+(fn|struct|enum|trait|type|const|mod)\s+(\w+)'), re.compile(r'pub\s+(unsafe\s+)?(fn|trait)\s+(\w+)'), ] # Detect enum variant or struct field in diff lines ENUM_VARIANT_LINE = re.compile(r'^\s*(\w+)\s*(?:\(.*\))?\s*[,{]?\s*$') PUB_FIELD_LINE = re.compile(r'^\s+pub\s+(\w+)\s*:') # Items that are implementation detail, not API PRIV_PATTERNS = [ re.compile(r'^[+-]\s+(?!pub\b)'), ] # Lines to ignore in API analysis IGNORE_PATTERNS = [ re.compile(r'^[+-]\s*$'), # empty lines re.compile(r'^[+-]\s*(//|#\[)'), # comments / attributes re.compile(r'^[+-]\s*\}'), # closing braces re.compile(r'^[+-]\s*use\s+'), # use imports (unless pub use) ] class APIAnalysis: def __init__(self): self.added = [] # pub items added (enum variants, struct fields, top-level items) self.removed = [] # pub items removed self.changed = [] # pub items with signature changes self.field_changes = [] # pub struct field additions self.variant_additions = [] # pub enum variant additions self.priv_changes = 0 # count of private changes (safe) self.total_hunks = 0 @property def has_breaking_changes(self) -> bool: return bool(self.removed or self.changed) @property def has_safe_changes(self) -> bool: return bool(self.added or self.variant_additions or self.field_changes) @property def summary(self) -> str: parts = [] if self.added: parts.append(f"+{len(self.added)} pub item") if self.variant_additions: parts.append(f"+{len(self.variant_additions)} enum var") if self.field_changes: parts.append(f"+{len(self.field_changes)} struct field") if self.removed: parts.append(f"-{len(self.removed)} pub (BREAKING)") if self.changed: parts.append(f"~{len(self.changed)} sig change") if self.priv_changes: parts.append(f"{self.priv_changes} internal") return ", ".join(parts) if parts else "no API changes" def analyze_git_diff(file_path: str, mode: str = "staged") -> APIAnalysis: """ Analyze git diff for a .rs file and classify API changes. """ analysis = APIAnalysis() if not (file_path.endswith(".rs") or file_path.endswith(".svelte") or file_path.endswith(".ts")): return analysis try: if mode == "last": cmd = ["git", "diff", "HEAD~1", "HEAD", "--", file_path] elif mode == "staged": cmd = ["git", "diff", "--cached", "--", file_path] else: cmd = ["git", "diff", "--", file_path] result = subprocess.run( cmd, capture_output=True, text=True, cwd=str(REPO_ROOT), ) diff_text = result.stdout except Exception: return analysis if not diff_text.strip(): return analysis # Parse hunks current_hunk = [] in_hunk = False for line in diff_text.split("\n"): if line.startswith("@@"): if current_hunk: _classify_hunk(current_hunk, analysis) current_hunk = [line] in_hunk = True elif in_hunk: current_hunk.append(line) if current_hunk: _classify_hunk(current_hunk, analysis) return analysis def _classify_hunk(hunk_lines: list, analysis: APIAnalysis): """Classify a single diff hunk into API change types.""" analysis.total_hunks += 1 # Extract hunk context (the @@ -a,b +c,d @@ section header) header = hunk_lines[0] context = "" hunk_m = HUNK_CONTEXT_PATTERN.search(header) if hunk_m: context = hunk_m.group(3).strip() # Determine if hunk is inside a public type inside_pub_enum = False inside_pub_struct = False pub_context_name = "" for pat in PUB_CONTEXT_PATTERNS: cm = pat.search(context) if cm: groups = cm.groups() kind = groups[0] if groups[0] else groups[2] name = groups[-1] if kind == "enum": inside_pub_enum = True pub_context_name = name elif kind == "struct": inside_pub_struct = True pub_context_name = name break lines = [l for l in hunk_lines if not any(p.match(l) for p in IGNORE_PATTERNS)] added = [l[1:] for l in lines if l.startswith("+")] removed = [l[1:] for l in lines if l.startswith("-")] # Check for removed pub items (top-level) for line in removed: for pat in PUB_REMOVE_PATTERNS: m = pat.search(line) if m: kind = m.group(1) if m.lastindex >= 1 else "item" name = m.group(m.lastindex) analysis.removed.append(f"{kind} {name}") break # Check for added pub items (top-level) for line in added: for pat in PUB_ADD_PATTERNS: m = pat.search(line) if m: kind = m.group(1) if m.lastindex >= 1 else "item" name = m.group(m.lastindex) analysis.added.append(f"{kind} {name}") break # Detect additions inside pub enums (variant additions) if inside_pub_enum: for line in added: stripped = line.strip().rstrip(",") vm = ENUM_VARIANT_LINE.match(stripped) if vm and not stripped.startswith("///") and not stripped.startswith("#["): variant = vm.group(1) # Skip known non-variant words if variant not in ("{", "}", "pub", "enum"): analysis.variant_additions.append( f"{pub_context_name}::{variant}" ) # Detect field additions inside pub structs if inside_pub_struct: for line in added: fm = PUB_FIELD_LINE.match(line) if fm: analysis.field_changes.append( f"{pub_context_name}.{fm.group(1)}" ) # Count private changes for line in lines: if any(p.match(line) for p in PRIV_PATTERNS) and not any(p.match(line) for p in IGNORE_PATTERNS): analysis.priv_changes += 1 # Detect signature changes (same fn name in both + and -) added_fns = set() removed_fns = set() FN_SIG_PATTERN = re.compile(r'^[+-]\s+pub\s+(?:unsafe\s+)?fn\s+(\w+)\s*\(') for line in added: m = FN_SIG_PATTERN.search(line) if m: added_fns.add(m.group(1)) for line in removed: m = FN_SIG_PATTERN.search(line) if m: removed_fns.add(m.group(1)) for fn_name in removed_fns & added_fns: analysis.changed.append(fn_name) if fn_name in analysis.removed: analysis.removed = [x for x in analysis.removed if fn_name not in x] # ============================================================================ # LAYER 3: cargo-semver-checks Integration # ============================================================================ def _is_lib_crate(crate_rel: str) -> bool: """Check if a crate has a lib.rs (is a library crate).""" lib_rs = REPO_ROOT / crate_rel / "src" / "lib.rs" return lib_rs.exists() def run_semver_checks(changed_crates: list, graph: dict, mode: str = "last") -> dict: """ Run cargo-semver-checks on changed crates. Baseline = parent commit, Current = working tree / HEAD. Uses git checkout to temporarily swap crate file versions. Returns { crate_name: { "violations": [...], "error": str } } """ results = {} # Check if cargo-semver-checks is available try: subprocess.run(["cargo", "semver-checks", "--version"], capture_output=True, text=True, check=True) except (subprocess.CalledProcessError, FileNotFoundError): for c in changed_crates: results[c] = {"violations": [], "error": "cargo-semver-checks not installed. Run: cargo install cargo-semver-checks"} return results # Check if nightly is available for rustdoc JSON nightly = subprocess.run( ["rustup", "run", "nightly", "rustc", "--version"], capture_output=True, text=True, ) if nightly.returncode != 0: for c in changed_crates: results[c] = {"violations": [], "error": "Nightly Rust required for rustdoc JSON. Run: rustup toolchain install nightly"} return results # Determine baseline git ref if mode == "last": baseline_ref = "HEAD~1" else: baseline_ref = "HEAD" for crate_name in changed_crates: crate_info = graph.get("crates", {}).get(crate_name) if not crate_info: continue crate_rel = crate_info["path"] # Skip binary crates (no lib.rs → no rustdoc JSON output) if not _is_lib_crate(crate_rel): results[crate_name] = {"violations": [], "error": "skipped (binary crate, no lib.rs)"} continue results[crate_name] = _check_single_crate( crate_name, crate_rel, baseline_ref, ) return results def _gen_rustdoc(crate_name: str, crate_path: str): """Generate rustdoc JSON for a crate (lib target only).""" subprocess.run( ["cargo", "+nightly", "rustdoc", "--lib", "--manifest-path", os.path.join(crate_path, "Cargo.toml"), "--", "-Z", "unstable-options", "--output-format", "json"], capture_output=True, text=True, cwd=str(REPO_ROOT), timeout=300, ) def _check_single_crate(crate_name: str, crate_rel: str, baseline_ref: str) -> dict: """ Check a single crate: 1. git checkout baseline_ref -- crates/hcie-X/ (swap to baseline) 2. Generate baseline rustdoc JSON 3. git checkout HEAD -- crates/hcie-X/ (restore to current) 4. Generate current rustdoc JSON 5. Run cargo-semver-checks comparison """ result = {"violations": [], "error": None} rustdoc_json_name = crate_name.replace("-", "_") crate_path = os.path.join(str(REPO_ROOT), crate_rel) try: # Step 1: Checkout baseline version subprocess.run( ["git", "checkout", baseline_ref, "--", crate_rel], capture_output=True, text=True, cwd=str(REPO_ROOT), ) # Step 2: Generate baseline rustdoc JSON _gen_rustdoc(crate_name, crate_path) import shutil baseline_json = REPO_ROOT / "target" / "doc" / f"{rustdoc_json_name}.json" if not baseline_json.exists(): baseline_json = REPO_ROOT / "target" / "doc" / crate_name / "index.json" tmp_baseline = Path(tempfile.mktemp(suffix="_baseline.json")) if baseline_json.exists(): shutil.copy(str(baseline_json), str(tmp_baseline)) # Step 3: Restore to current version (HEAD) subprocess.run( ["git", "checkout", "HEAD", "--", crate_rel], capture_output=True, text=True, cwd=str(REPO_ROOT), ) # Step 4: Generate current rustdoc JSON _gen_rustdoc(crate_name, crate_path) current_json = REPO_ROOT / "target" / "doc" / f"{rustdoc_json_name}.json" if not current_json.exists(): current_json = REPO_ROOT / "target" / "doc" / crate_name / "index.json" tmp_current = Path(tempfile.mktemp(suffix="_current.json")) if current_json.exists(): shutil.copy(str(current_json), str(tmp_current)) # Step 5: Compare if tmp_baseline.exists() and tmp_current.exists(): proc = subprocess.run( ["cargo", "semver-checks", "--baseline-rustdoc", str(tmp_baseline), "--current-rustdoc", str(tmp_current)], capture_output=True, text=True, cwd=str(REPO_ROOT), timeout=180, ) output = proc.stdout + proc.stderr if proc.returncode != 0: result["violations"] = _parse_semver_output(output) else: parts = [] if not tmp_baseline.exists(): parts.append("baseline") if not tmp_current.exists(): parts.append("current") result["error"] = f"rustdoc JSON not found: {' + '.join(parts)}" # Cleanup for p in [tmp_baseline, tmp_current]: if p.exists(): p.unlink() except subprocess.TimeoutExpired: result["error"] = "cargo-semver-checks timed out (rustdoc compilation too slow)" subprocess.run(["git", "checkout", "HEAD", "--", crate_rel], capture_output=True, cwd=str(REPO_ROOT)) except Exception as e: result["error"] = str(e).split("\n")[0][:120] subprocess.run(["git", "checkout", "HEAD", "--", crate_rel], capture_output=True, cwd=str(REPO_ROOT)) return result def _parse_semver_output(output: str) -> list: """Parse cargo-semver-checks output into structured violations.""" violations = [] for line in output.split("\n"): line = line.strip() if not line: continue # Skip info/status lines if line.startswith("[") or line.startswith("Checking") or \ line.startswith("Comparing") or line.startswith("Detected") or \ line.startswith("Finished") or line.startswith("Running"): continue if "error" in line.lower() or "breaking" in line.lower() or \ "major" in line.lower() or "violation" in line.lower() or \ "removed" in line.lower() or "changed" in line.lower(): violations.append(line) return violations[:30] # ============================================================================ # MAIN REPORT GENERATOR # ============================================================================ def get_commit_info(mode: str) -> dict: """ Retrieve commit ID and message for the report context. Returns: { "commit_id": full SHA (e.g. "ebe469a6752842739a79202a24d924ca359d9c8c"), "commit_short": first 7 chars (e.g. "ebe469a"), "commit_message": first line of commit message, "previous_commit_id": parent commit SHA (for mode="last"), } """ result = { "commit_id": "", "commit_short": "", "commit_message": "", "previous_commit_id": "", } try: # Current HEAD (or latest commit) head = subprocess.check_output( ["git", "rev-parse", "HEAD"], cwd=str(REPO_ROOT), text=True ).strip() result["commit_id"] = head result["commit_short"] = head[:7] # Commit message (first line) msg = subprocess.check_output( ["git", "log", "-1", "--pretty=%s"], cwd=str(REPO_ROOT), text=True ).strip() result["commit_message"] = msg # For --mode last, also get parent commit if mode == "last": try: parent = subprocess.check_output( ["git", "rev-parse", "HEAD~1"], cwd=str(REPO_ROOT), text=True ).strip() result["previous_commit_id"] = parent except subprocess.CalledProcessError: result["previous_commit_id"] = "unknown" except subprocess.CalledProcessError: pass return result def get_changed_files(mode: str) -> list: """Get list of changed files from git.""" if mode == "staged": cmd = ["git", "diff", "--cached", "--name-only"] elif mode == "last": cmd = ["git", "diff", "HEAD~1", "HEAD", "--name-only"] elif mode == "working": # Combined: staged + unstaged staged = subprocess.check_output( ["git", "diff", "--cached", "--name-only"], cwd=str(REPO_ROOT), text=True ).strip().split("\n") unstaged = subprocess.check_output( ["git", "diff", "--name-only"], cwd=str(REPO_ROOT), text=True ).strip().split("\n") changed = set( s.strip() for s in staged + unstaged if s.strip() ) return sorted(changed) else: return [] try: out = subprocess.check_output(cmd, cwd=str(REPO_ROOT), text=True) return [l.strip() for l in out.strip().split("\n") if l.strip()] except subprocess.CalledProcessError: return [] def _reports_dir() -> Path: """Ensure reports directory exists.""" reports = LOG_DIR reports.mkdir(parents=True, exist_ok=True) return reports def _save_report(report_text: str, mode: str, has_high_risk: bool, commit_short: str = ""): """Save report to file with timestamp.""" import datetime ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") risk_tag = "HIGH" if has_high_risk else "LOW" suffix = f"_{commit_short}" if commit_short else "" filename = f"semantic_impact_{ts}_{risk_tag}_{mode}{suffix}.txt" report_path = _reports_dir() / filename with open(report_path, "w") as f: f.write(report_text) return report_path def generate_report(graph: dict, changed_files: list, mode: str, run_api_analysis: bool = True, run_semver: bool = False, commit_info: dict = None) -> tuple: """ Generate the full semantic impact report. Args: graph: Dependency graph from build_dependency_graph(). changed_files: List of changed file paths. mode: Analysis mode ("staged", "last", or "working"). run_api_analysis: Whether to run Layer 2 API analysis. run_semver: Whether to run Layer 3 semver checks. commit_info: Dict with commit_id, commit_short, commit_message, previous_commit_id. Returns: (report_text: str, highest_risk: str) """ lines = [] highest_risk = "LOW" # ── Header ── lines.append("=== SEMANTIC IMPACT REPORT v3 ===") lines.append(f"Mode : {mode}") if commit_info and commit_info.get("commit_id"): lines.append(f"Commit : {commit_info['commit_short']} — \"{commit_info['commit_message']}\"") if mode == "last" and commit_info.get("previous_commit_id"): lines.append(f"Range : {commit_info['previous_commit_id'][:7]} → {commit_info['commit_short']}") lines.append(f"Files changed : {len(changed_files)}") lines.append("") # ── Layer 1: Changed files grouped by crate ── changed_by_crate = defaultdict(list) for fp in changed_files: crate = get_file_crate(fp, graph) changed_by_crate[crate].append(fp) # Organize changed crates by blast radius high_crates = [] med_crates = [] low_crates = [] for crate in changed_by_crate: br = graph.get("blast_radius", {}).get(crate, "LOW") if br == "HIGH": high_crates.append(crate) elif br == "MEDIUM": med_crates.append(crate) else: low_crates.append(crate) num_high = sum(len(changed_by_crate[c]) for c in high_crates) if high_crates: highest_risk = "HIGH" elif med_crates: highest_risk = "MEDIUM" # ── Layer 1 output ── lines.append("[LAYER 1 — CRATE DEPENDENCY IMPACT]") lines.append(f"High-radius crates affected : {len(high_crates)} ({num_high} files)") lines.append(f"Medium-radius crates : {len(med_crates)}") lines.append(f"Low-radius crates : {len(low_crates)}") lines.append(f"Overall risk : {highest_risk}") lines.append("") if high_crates: lines.append("── HIGH-RISK CRATES WITH CHANGES ──") for crate in high_crates: crate_info = graph.get("crates", {}).get(crate, {}) files = changed_by_crate[crate] lines.append(f" {crate}/ ({len(crate_info.get('dependents', []))} dependents)") for fp in files: desc = graph.get("manifest_annotations", {}).get(fp, {}).get("description", "") if desc: lines.append(f" {fp}") lines.append(f" → {desc}") else: lines.append(f" {fp}") # Show dependents deps = crate_info.get("dependents", []) if deps: lines.append(f" ⤷ Impacts: {', '.join(deps)}") lines.append("") if med_crates: lines.append("── MEDIUM-RISK CRATES ──") for crate in med_crates: crate_info = graph.get("crates", {}).get(crate, {}) files = changed_by_crate[crate] lines.append(f" {crate}/ ({len(crate_info.get('dependents', []))} dependents)") for fp in files: desc = graph.get("manifest_annotations", {}).get(fp, {}).get("description", "") if desc: lines.append(f" {fp} — {desc}") else: lines.append(f" {fp}") deps = crate_info.get("dependents", []) if deps: lines.append(f" ⤷ Impacts: {', '.join(deps)}") lines.append("") if low_crates: lines.append("── LOW-RISK CRATES ──") for crate in low_crates: for fp in changed_by_crate[crate]: lines.append(f" {fp}") lines.append("") # ── Layer 2: API Analysis ── if run_api_analysis: lines.append("[LAYER 2 — API CHANGE ANALYSIS]") total_breaking = 0 total_safe = 0 for crate in sorted(changed_by_crate): for fp in changed_by_crate[crate]: analysis = analyze_git_diff(fp, mode) if analysis.total_hunks > 0: tag = "BREAKING" if analysis.has_breaking_changes else "safe" if analysis.has_breaking_changes: total_breaking += 1 else: total_safe += 1 lines.append(f" {fp}") lines.append(f" {tag}: {analysis.summary}") if analysis.removed: for item in analysis.removed: lines.append(f" ❌ REMOVED: {item}") if analysis.changed: for item in analysis.changed: lines.append(f" ⚠️ CHANGED SIG: {item}") if analysis.added: for item in analysis.added: lines.append(f" ✅ ADDED: {item}") if analysis.variant_additions: for v in analysis.variant_additions: lines.append(f" 📦 NEW VARIANT: {v}") if analysis.field_changes: for f in analysis.field_changes: lines.append(f" 🏛️ NEW FIELD: {f}") lines.append("") if total_breaking == 0 and total_safe == 0: lines.append(" No API-level changes detected (private/internal changes only).") lines.append("") # ── Layer 3: cargo-semver-checks ── if run_semver: lines.append("[LAYER 3 — SEMVER CHECK]") changed_crate_names = [ c for c in changed_by_crate if c and c in graph.get("crates", {}) ] semver_results = run_semver_checks(changed_crate_names, graph, mode) for crate, result in semver_results.items(): if result["error"]: lines.append(f" {crate}: ERROR — {result['error']}") elif result["violations"]: lines.append(f" {crate}: {len(result['violations'])} semver violation(s)") for v in result["violations"][:5]: lines.append(f" ⚠️ {v}") highest_risk = "HIGH" else: lines.append(f" {crate}: ✅ No semver violations detected") lines.append("") # ── Manifest File Suggestions ── unannotated = [ fp for fp in changed_files if fp not in graph.get("manifest_annotations", {}) and (fp.endswith(".rs") or fp.endswith(".svelte") or fp.endswith(".ts")) ] if unannotated: lines.append("[SUGGESTION — Missing Manifest Annotations]") lines.append(" These .rs files have no description in project_manifest.json.") lines.append(" Consider adding entries for better semantic tracking:") for fp in unannotated[:5]: crate = get_file_crate(fp, graph) br = get_file_blast_radius(fp, graph) lines.append(f" {fp} (crate: {crate}, auto-blast: {br})") if len(unannotated) > 5: lines.append(f" ... and {len(unannotated) - 5} more") lines.append("") # ── Errors ── if graph.get("errors"): lines.append("[WARNINGS]") for err in graph["errors"]: lines.append(f" ⚠️ {err}") lines.append("") # ── Blocking Warning ── if highest_risk == "HIGH": lines.append("[BLOCKING WARNING]") lines.append(" HIGH-risk crate(s) directly modified — review required.") if run_api_analysis: lines.append(" Check Layer 2 API changes above for breaking changes.") lines.append("") lines.append("─" * 60) return "\n".join(lines), highest_risk # ============================================================================ # CLI # ============================================================================ def main(): import argparse parser = argparse.ArgumentParser( description="Semantic Impact Report Generator v3", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # Pre-commit check (fast — layers 1+2) python3 .githooks/generate_semantic_report.py --mode staged # Post-commit audit python3 .githooks/generate_semantic_report.py --mode last # Full pre-commit with semver check (slow) python3 .githooks/generate_semantic_report.py --mode staged --semver # Quick mode — only dependency graph, no API analysis python3 .githooks/generate_semantic_report.py --mode staged --no-api # Check working tree changes (staged + unstaged) python3 .githooks/generate_semantic_report.py --mode working """, ) parser.add_argument( "--mode", choices=["staged", "last", "working"], default="staged", help="Which changes to analyze (default: staged)", ) parser.add_argument( "--no-api", action="store_true", help="Skip Layer 2 (git diff API analysis)", ) parser.add_argument( "--semver", action="store_true", help="Enable Layer 3 (cargo-semver-checks, SLOW)", ) parser.add_argument( "--save", action="store_true", help="Save report to logs/ directory with timestamp", ) parser.add_argument( "--json", action="store_true", help="Output JSON machine-readable report (for CI/tooling)", ) args = parser.parse_args() # ── Build dependency graph ── print(" [Layer 1] Scanning Cargo.toml dependency graph...", file=sys.stderr) graph = build_dependency_graph() if graph.get("errors"): for err in graph["errors"]: print(f" [WARN] {err}", file=sys.stderr) num_crates = len(graph.get("crates", {})) print(f" → {num_crates} crates, {len(graph['all_files'])} source files", file=sys.stderr) if args.mode == "last": print(f" [git] Comparing HEAD~1 → HEAD", file=sys.stderr) elif args.mode == "staged": print(f" [git] Checking staged changes", file=sys.stderr) else: print(f" [git] Checking working tree changes", file=sys.stderr) changed_files = get_changed_files(args.mode) if not changed_files: print(" → No changed files. Nothing to check.", file=sys.stderr) sys.exit(0) print(f" → {len(changed_files)} file(s) changed", file=sys.stderr) if not args.no_api: print(f" [Layer 2] Analyzing git diff for API changes...", file=sys.stderr) else: print(f" [Layer 2] Skipped (--no-api)", file=sys.stderr) if args.semver: print(f" [Layer 3] Running cargo-semver-checks...", file=sys.stderr) else: print(f" [Layer 3] Skipped (use --semver to enable)", file=sys.stderr) # ── Get commit info ── commit_info = get_commit_info(args.mode) # ── Generate report ── report_text, highest_risk = generate_report( graph=graph, changed_files=changed_files, mode=args.mode, run_api_analysis=not args.no_api, run_semver=args.semver, commit_info=commit_info, ) if args.json: json_out = { "version": 3, "mode": args.mode, "commit_id": commit_info.get("commit_id", ""), "commit_short": commit_info.get("commit_short", ""), "commit_message": commit_info.get("commit_message", ""), "previous_commit_id": commit_info.get("previous_commit_id", ""), "files_changed": len(changed_files), "highest_risk": highest_risk, "high_risk_files": [ fp for fp in changed_files if get_file_blast_radius(fp, graph) == "HIGH" ], "crates_affected": sorted(set( get_file_crate(fp, graph) for fp in changed_files if get_file_crate(fp, graph) )), } print(json.dumps(json_out, indent=2)) else: print(report_text) # ── Save ── if args.save or highest_risk == "HIGH": saved_path = _save_report( report_text, args.mode, highest_risk == "HIGH", commit_short=commit_info.get("commit_short", ""), ) print(f" Report saved: {saved_path}", file=sys.stderr) sys.exit(1 if highest_risk == "HIGH" else 0) if __name__ == "__main__": main()