""" visualizer.py — Visual blast radius renderer. Renders the blast radius as a colored, indented tree showing: - The changed symbol at the root - Direct callers (who calls this?) - Direct callees (what does this call?) - Transitive impact propagation - Proof chains: the actual code line creating each edge Designed for terminal output with ANSI colors. """ import ast import os import re from typing import Dict, List, Optional, Set, Tuple # ANSI color codes class _C: RED = "\033[91m" YELLOW = "\033[93m" GREEN = "\033[92m" CYAN = "\033[96m" MAGENTA = "\033[95m" BLUE = "\033[94m" DIM = "\033[2m" BOLD = "\033[1m" RESET = "\033[0m" WHITE = "\033[97m" def render_blast_radius( graph: Dict[str, List[str]], changed_symbols: List[str], symbols: dict, max_depth: int = 3, show_proof: bool = False, repo_path: str = "", ) -> str: """ Render a visual tree of the blast radius for changed symbols. Returns a formatted string ready for terminal output. """ # Build reverse graph (callers) reverse: Dict[str, Set[str]] = {} for caller, callees in graph.items(): for callee in callees: reverse.setdefault(callee, set()).add(caller) lines: List[str] = [] lines.append("") lines.append(f"{_C.BOLD}{_C.WHITE}{'═' * 70}{_C.RESET}") lines.append(f"{_C.BOLD}{_C.WHITE} BLAST RADIUS ANALYSIS{_C.RESET}") lines.append(f"{_C.BOLD}{_C.WHITE}{'═' * 70}{_C.RESET}") for sym_id in changed_symbols: lines.append("") lines.append(f" {_C.BOLD}{_C.RED}⚡ CHANGED:{_C.RESET} {_C.BOLD}{sym_id}{_C.RESET}") # Show symbol location info if sym_id in symbols: sym = symbols[sym_id] lines.append(f" {_C.DIM} File: {sym.file}{_C.RESET}") lines.append(f" {_C.DIM} Line: {sym.lineno}{_C.RESET}") lines.append("") # ---- CALLERS (who is affected by this change?) ---- direct_callers = sorted(reverse.get(sym_id, set())) lines.append(f" {_C.BOLD}{_C.YELLOW}▲ WHO CALLS THIS? (directly affected){_C.RESET}") if not direct_callers: lines.append(f" {_C.DIM} (no direct callers found){_C.RESET}") else: for i, caller in enumerate(direct_callers): is_last = i == len(direct_callers) - 1 prefix = "└──" if is_last else "├──" lines.append(f" {_C.YELLOW} {prefix} {caller}{_C.RESET}") if show_proof: proof = _find_proof(caller, sym_id, symbols, graph) if proof: pad = " " if is_last else "│ " lines.append(f" {_C.DIM} {pad} proof: {proof}{_C.RESET}") # 2nd-level callers (transitive) if max_depth >= 2: indirect_callers = sorted(reverse.get(caller, set())) for j, caller2 in enumerate(indirect_callers[:5]): is_last2 = j == len(indirect_callers[:5]) - 1 indent = " " if is_last else "│ " prefix2 = "└──" if is_last2 else "├──" lines.append( f" {_C.DIM} {indent}{prefix2} {caller2}{_C.RESET}" ) if len(indirect_callers) > 5: indent = " " if is_last else "│ " lines.append( f" {_C.DIM} {indent}... +{len(indirect_callers) - 5} more{_C.RESET}" ) lines.append("") # ---- CALLEES (what does this function depend on?) ---- direct_callees = sorted(graph.get(sym_id, [])) lines.append(f" {_C.BOLD}{_C.GREEN}▼ WHAT DOES THIS CALL? (dependencies){_C.RESET}") if not direct_callees: lines.append(f" {_C.DIM} (no outgoing calls resolved){_C.RESET}") else: for i, callee in enumerate(direct_callees): is_last = i == len(direct_callees) - 1 prefix = "└──" if is_last else "├──" lines.append(f" {_C.GREEN} {prefix} {callee}{_C.RESET}") if show_proof: proof = _find_proof(sym_id, callee, symbols, graph) if proof: pad = " " if is_last else "│ " lines.append(f" {_C.DIM} {pad} proof: {proof}{_C.RESET}") # 2nd-level callees if max_depth >= 2: indirect_callees = sorted(graph.get(callee, [])) for j, callee2 in enumerate(indirect_callees[:5]): is_last2 = j == len(indirect_callees[:5]) - 1 indent = " " if is_last else "│ " prefix2 = "└──" if is_last2 else "├──" lines.append( f" {_C.DIM} {indent}{prefix2} {callee2}{_C.RESET}" ) if len(indirect_callees) > 5: indent = " " if is_last else "│ " lines.append( f" {_C.DIM} {indent}... +{len(indirect_callees) - 5} more{_C.RESET}" ) lines.append("") # ---- FULL TRANSITIVE BLAST RADIUS ---- all_affected = _get_transitive_callers(reverse, sym_id, max_depth) lines.append(f" {_C.BOLD}{_C.CYAN}◉ FULL BLAST RADIUS{_C.RESET}") lines.append( f" {_C.CYAN} {len(all_affected)} symbols transitively affected{_C.RESET}" ) # Group by file by_file: Dict[str, List[str]] = {} for affected_sym in all_affected: parts = affected_sym.split(":", 1) filepath = parts[0] if len(parts) == 2 else "unknown" by_file.setdefault(filepath, []).append(affected_sym) for filepath, syms in sorted(by_file.items()): lines.append(f" {_C.BLUE} 📄 {filepath} ({len(syms)} symbols){_C.RESET}") for sym in sorted(syms)[:8]: name = sym.split(":", 1)[1] if ":" in sym else sym lines.append(f" {_C.DIM} · {name}{_C.RESET}") if len(syms) > 8: lines.append(f" {_C.DIM} ... +{len(syms) - 8} more{_C.RESET}") # ---- SUMMARY ---- lines.append("") lines.append(f"{_C.BOLD}{_C.WHITE}{'─' * 70}{_C.RESET}") total_blast = set() for sym_id in changed_symbols: total_blast.update(_get_transitive_callers(reverse, sym_id, max_depth)) total_deps = set() for sym_id in changed_symbols: total_deps.update(_get_transitive_callees(graph, sym_id, max_depth)) blast_files = set() for sym in total_blast: parts = sym.split(":", 1) if len(parts) == 2: blast_files.add(parts[0]) lines.append(f" {_C.BOLD}Summary:{_C.RESET}") lines.append(f" Changed symbols : {_C.RED}{len(changed_symbols)}{_C.RESET}") lines.append(f" Direct callers : {_C.YELLOW}{sum(len(reverse.get(s, set())) for s in changed_symbols)}{_C.RESET}") lines.append(f" Direct dependencies: {_C.GREEN}{sum(len(graph.get(s, [])) for s in changed_symbols)}{_C.RESET}") lines.append(f" Total blast radius : {_C.CYAN}{len(total_blast)} symbols across {len(blast_files)} files{_C.RESET}") lines.append(f" Total dependencies : {_C.GREEN}{len(total_deps)} symbols{_C.RESET}") lines.append(f"{_C.BOLD}{_C.WHITE}{'═' * 70}{_C.RESET}") lines.append("") return "\n".join(lines) def render_verification( graph: Dict[str, List[str]], changed_symbols: List[str], symbols: dict, ) -> str: """ Render verification proof: for each edge in the blast radius, show the actual code line that creates the dependency. """ reverse: Dict[str, Set[str]] = {} for caller, callees in graph.items(): for callee in callees: reverse.setdefault(callee, set()).add(caller) lines: List[str] = [] lines.append("") lines.append(f"{_C.BOLD}{_C.WHITE}{'═' * 70}{_C.RESET}") lines.append(f"{_C.BOLD}{_C.WHITE} VERIFICATION: Proof of each connection{_C.RESET}") lines.append(f"{_C.BOLD}{_C.WHITE}{'═' * 70}{_C.RESET}") for sym_id in changed_symbols: lines.append("") lines.append(f" {_C.BOLD}{_C.RED}⚡ {sym_id}{_C.RESET}") lines.append("") # Verify each caller callers = sorted(reverse.get(sym_id, set())) if callers: lines.append(f" {_C.BOLD}{_C.YELLOW} Callers (these functions call the changed code):{_C.RESET}") for caller in callers: lines.append(f" {_C.YELLOW} → {caller}{_C.RESET}") proof = _find_proof(caller, sym_id, symbols, graph) if proof: lines.append(f" {_C.DIM} evidence: {proof}{_C.RESET}") else: lines.append(f" {_C.DIM} evidence: (edge exists in call graph){_C.RESET}") lines.append("") # Verify each callee callees = sorted(graph.get(sym_id, [])) if callees: lines.append(f" {_C.BOLD}{_C.GREEN} Callees (the changed code calls these):{_C.RESET}") for callee in callees: lines.append(f" {_C.GREEN} → {callee}{_C.RESET}") proof = _find_proof(sym_id, callee, symbols, graph) if proof: lines.append(f" {_C.DIM} evidence: {proof}{_C.RESET}") else: lines.append(f" {_C.DIM} evidence: (edge exists in call graph){_C.RESET}") lines.append("") lines.append(f"{_C.BOLD}{_C.WHITE}{'═' * 70}{_C.RESET}") lines.append("") return "\n".join(lines) # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _get_transitive_callers( reverse: Dict[str, Set[str]], start: str, max_depth: int, ) -> List[str]: """BFS up the reverse graph to find all transitively affected symbols.""" visited: Set[str] = {start} result: List[str] = [] frontier = [start] depth = 0 while frontier and depth < max_depth: next_frontier = [] for node in frontier: for caller in reverse.get(node, set()): if caller not in visited: visited.add(caller) result.append(caller) next_frontier.append(caller) frontier = next_frontier depth += 1 return result def _get_transitive_callees( graph: Dict[str, List[str]], start: str, max_depth: int, ) -> List[str]: """BFS down the forward graph to find all dependencies.""" visited: Set[str] = {start} result: List[str] = [] frontier = [start] depth = 0 while frontier and depth < max_depth: next_frontier = [] for node in frontier: for callee in graph.get(node, []): if callee not in visited: visited.add(callee) result.append(callee) next_frontier.append(callee) frontier = next_frontier depth += 1 return result def _find_proof( caller_id: str, callee_id: str, symbols: dict, graph: Dict[str, List[str]], ) -> Optional[str]: """ Find the actual code line in `caller` that references `callee`. This is the "proof" that the edge is real — a grep through the caller's source code for the callee's function name. """ if caller_id not in symbols: return None caller_sym = symbols[caller_id] callee_name = callee_id.split(":")[-1] if ":" in callee_id else callee_id # Strip class prefix for method calls: "ClassName.method" -> "method" if "." in callee_name: bare_name = callee_name.split(".")[-1] else: bare_name = callee_name # Search caller's code for the call for i, line in enumerate(caller_sym.code.splitlines(), start=1): stripped = line.strip() # Look for function/method call pattern if bare_name + "(" in stripped or f".{bare_name}(" in stripped: # Truncate long lines if len(stripped) > 80: stripped = stripped[:77] + "..." return f"line {caller_sym.lineno + i - 1}: {stripped}" return None