"""Tools 4, 12, 16: review context, affected flows, detect changes.""" from __future__ import annotations import logging from pathlib import Path from typing import Any from ..changes import analyze_changes, parse_diff_ranges, parse_git_diff_ranges # noqa: F401 from ..context_savings import attach_context_savings, estimate_file_tokens from ..flows import get_affected_flows as _get_affected_flows from ..graph import edge_to_dict, node_to_dict from ..hints import generate_hints, get_session from ..incremental import get_changed_files, get_staged_and_unstaged from ._common import _get_store, _resolve_graph_file_paths logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Tool 4: get_review_context # --------------------------------------------------------------------------- def get_review_context( changed_files: list[str] | None = None, max_depth: int = 2, include_source: bool = True, max_lines_per_file: int = 200, repo_root: str | None = None, base: str = "HEAD~1", detail_level: str = "standard", ) -> dict[str, Any]: """Generate a focused review context from changed files. Builds a token-optimized subgraph + source snippets for code review. Args: changed_files: Files to review (auto-detected from git diff if omitted). max_depth: Impact radius depth (default: 2). include_source: Whether to include source code snippets (default: True). max_lines_per_file: Max source lines per file in output (default: 200). repo_root: Repository root path. Auto-detected if omitted. base: Git ref for change detection (default: HEAD~1). detail_level: Output detail level. "standard" returns full context; "minimal" returns summary, risk level, changed/impacted file counts, top 5 key entity names, test gap count, and next tool suggestions. Default: "standard". Returns: Structured review context with subgraph, source snippets, and review guidance. """ store, root = _get_store(repo_root) try: # Get impact radius first if changed_files is None: changed_files = get_changed_files(root, base) if not changed_files: changed_files = get_staged_and_unstaged(root) if not changed_files: return { "status": "ok", "summary": "No changes detected. Nothing to review.", "context": {}, } graph_files = _resolve_graph_file_paths(store, root, changed_files) original_tokens = estimate_file_tokens(root, changed_files) impact = store.get_impact_radius(graph_files, max_depth=max_depth) if detail_level == "minimal": impacted_count = len(impact["impacted_nodes"]) if impacted_count > 20: risk = "high" elif impacted_count > 5: risk = "medium" else: risk = "low" key_entities = [ n.name for n in impact["changed_nodes"][:5] ] # Count test gaps among changed functions. changed_funcs = [ n for n in impact["changed_nodes"] if n.kind == "Function" and not n.is_test ] test_edges = [ e for e in impact["edges"] if e.kind == "TESTED_BY" ] tested_qualified = {e.source_qualified for e in test_edges} test_gap_count = sum( 1 for f in changed_funcs if f.qualified_name not in tested_qualified ) summary_parts = [ f"Review context for {len(changed_files)} changed file(s):", f" - Risk: {risk}", f" - {len(impact['impacted_nodes'])} impacted nodes" f" in {len(impact['impacted_files'])} files", ] result = { "status": "ok", "summary": "\n".join(summary_parts), "risk": risk, "changed_file_count": len(changed_files), "impacted_file_count": len(impact["impacted_files"]), "key_entities": key_entities, "test_gaps": test_gap_count, "next_tool_suggestions": [ "detect_changes", "get_affected_flows", "get_impact_radius", ], } attach_context_savings(result, original_tokens=original_tokens) return result # Build review context context: dict[str, Any] = { "changed_files": changed_files, "impacted_files": impact["impacted_files"], "graph": { "changed_nodes": [ node_to_dict(n) for n in impact["changed_nodes"] ], "impacted_nodes": [ node_to_dict(n) for n in impact["impacted_nodes"] ], "edges": [edge_to_dict(e) for e in impact["edges"]], }, } # Add source snippets for changed files if include_source: snippets = {} for rel_path in changed_files: full_path = root / rel_path if full_path.is_file(): try: lines = full_path.read_text( errors="replace" ).splitlines() if len(lines) > max_lines_per_file: # Include only the relevant functions/classes relevant_lines = _extract_relevant_lines( lines, impact["changed_nodes"], str(full_path), ) snippets[rel_path] = relevant_lines else: snippets[rel_path] = "\n".join( f"{i+1}: {line}" for i, line in enumerate(lines) ) except (OSError, UnicodeDecodeError): snippets[rel_path] = "(could not read file)" context["source_snippets"] = snippets # Generate review guidance guidance = _generate_review_guidance(impact, changed_files) context["review_guidance"] = guidance summary_parts = [ f"Review context for {len(changed_files)} changed file(s):", f" - {len(impact['changed_nodes'])} directly changed nodes", f" - {len(impact['impacted_nodes'])} impacted nodes" f" in {len(impact['impacted_files'])} files", "", "Review guidance:", guidance, ] result = { "status": "ok", "summary": "\n".join(summary_parts), "context": context, } attach_context_savings(result, original_tokens=original_tokens) return result finally: store.close() def _extract_relevant_lines( lines: list[str], nodes: list, file_path: str ) -> str: """Extract only the lines relevant to changed nodes.""" ranges = [] for n in nodes: if n.file_path == file_path: start = max(0, n.line_start - 3) # 2 lines context before end = min(len(lines), n.line_end + 2) # 1 line context after ranges.append((start, end)) if not ranges: # Show first N lines as fallback return "\n".join( f"{i+1}: {line}" for i, line in enumerate(lines[:50]) ) # Merge overlapping ranges ranges.sort() merged = [ranges[0]] for start, end in ranges[1:]: if start <= merged[-1][1] + 1: merged[-1] = (merged[-1][0], max(merged[-1][1], end)) else: merged.append((start, end)) parts: list[str] = [] for start, end in merged: if parts: parts.append("...") for i in range(start, end): parts.append(f"{i+1}: {lines[i]}") return "\n".join(parts) def _generate_review_guidance( impact: dict, changed_files: list[str] ) -> str: """Generate review guidance based on the impact analysis.""" guidance_parts = [] # Check for test coverage changed_funcs = [ n for n in impact["changed_nodes"] if n.kind == "Function" ] test_edges = [e for e in impact["edges"] if e.kind == "TESTED_BY"] tested_funcs = {e.source_qualified for e in test_edges} untested = [ f for f in changed_funcs if f.qualified_name not in tested_funcs and not f.is_test ] if untested: guidance_parts.append( f"- {len(untested)} changed function(s) lack test coverage: " + ", ".join(n.name for n in untested[:5]) ) # Check for wide blast radius if len(impact["impacted_nodes"]) > 20: guidance_parts.append( f"- Wide blast radius: {len(impact['impacted_nodes'])} " "nodes impacted. " "Review callers and dependents carefully." ) # Check for inheritance changes inheritance_edges = [ e for e in impact["edges"] if e.kind in ("INHERITS", "IMPLEMENTS") ] if inheritance_edges: guidance_parts.append( f"- {len(inheritance_edges)} inheritance/implementation " "relationship(s) affected. " "Check for Liskov substitution violations." ) # Check for cross-file impact impacted_file_count = len(impact["impacted_files"]) if impacted_file_count > 3: guidance_parts.append( f"- Changes impact {impacted_file_count} other files." " Consider splitting into smaller PRs." ) if not guidance_parts: guidance_parts.append( "- Changes appear well-contained with minimal blast radius." ) return "\n".join(guidance_parts) # --------------------------------------------------------------------------- # Tool 12: get_affected_flows [REVIEW] # --------------------------------------------------------------------------- def get_affected_flows_func( changed_files: list[str] | None = None, base: str = "HEAD~1", repo_root: str | None = None, ) -> dict[str, Any]: """Find execution flows affected by changed files. [REVIEW] Identifies which execution flows pass through nodes in the changed files. Useful during code review to understand which user-facing or critical paths are affected by a change. Args: changed_files: List of changed file paths (relative to repo root). Auto-detected from git diff if omitted. base: Git ref for auto-detecting changes (default: HEAD~1). repo_root: Repository root path. Auto-detected if omitted. Returns: Affected flows sorted by criticality, with step details. """ store, root = _get_store(repo_root) try: if changed_files is None: changed_files = get_changed_files(root, base) if not changed_files: changed_files = get_staged_and_unstaged(root) if not changed_files: return { "status": "ok", "summary": "No changed files detected.", "affected_flows": [], "total": 0, } # Convert to absolute paths for graph lookup abs_files = [str(root / f) for f in changed_files] result = _get_affected_flows(store, abs_files) total = result["total"] out = { "status": "ok", "summary": ( f"{total} flow(s) affected by changes " f"in {len(changed_files)} file(s)" ), "changed_files": changed_files, "affected_flows": result["affected_flows"], "total": total, } out["_hints"] = generate_hints( "get_affected_flows", out, get_session() ) return out except Exception as exc: return {"status": "error", "error": str(exc)} finally: store.close() # --------------------------------------------------------------------------- # Tool 16: detect_changes [REVIEW] # --------------------------------------------------------------------------- def detect_changes_func( base: str = "HEAD~1", changed_files: list[str] | None = None, include_source: bool = False, max_depth: int = 2, repo_root: str | None = None, detail_level: str = "standard", ) -> dict[str, Any]: """Detect changes and produce risk-scored review guidance. [REVIEW] Primary tool for code review. Maps git diffs to affected functions, flows, communities, and test coverage gaps. Returns priority-ordered review guidance with risk scores. Args: base: Git ref to diff against (default: HEAD~1). changed_files: Explicit list of changed file paths (relative to repo root). Auto-detected from git diff if omitted. include_source: If True, include source code snippets for changed functions. Default: False. max_depth: Impact radius depth for BFS traversal. Default: 2. repo_root: Repository root path. Auto-detected if omitted. detail_level: Output detail level. "standard" returns full analysis; "minimal" returns only summary, risk_score, changed_file_count, test_gap_count, and top 3 review priorities (text only). Default: "standard". Returns: Risk-scored analysis with changed functions, affected flows, test gaps, and review priorities. """ store, root = _get_store(repo_root) try: # Detect changed files if not provided. if changed_files is None: changed_files = get_changed_files(root, base) if not changed_files: changed_files = get_staged_and_unstaged(root) if not changed_files: return { "status": "ok", "summary": "No changed files detected.", "risk_score": 0.0, "changed_functions": [], "affected_flows": [], "test_gaps": [], "review_priorities": [], } original_tokens = estimate_file_tokens(root, changed_files) # Convert to absolute paths for graph lookup. abs_files = [str(root / f) for f in changed_files] # Parse diff ranges for line-level mapping. diff_ranges = parse_diff_ranges(str(root), base) # Remap to absolute paths so they match graph file_paths. abs_ranges: dict[str, list[tuple[int, int]]] = {} for rel_path, ranges in diff_ranges.items(): abs_path = str(root / rel_path) abs_ranges[abs_path] = ranges analysis = analyze_changes( store, changed_files=abs_files, changed_ranges=abs_ranges if abs_ranges else None, repo_root=str(root), base=base, ) # Optionally include source snippets for changed functions. if include_source: for func in analysis.get("changed_functions", []): fp = func.get("file_path") ls = func.get("line_start") le = func.get("line_end") if fp and ls and le: file_path = Path(fp) if file_path.is_file(): try: lines = file_path.read_text( errors="replace" ).splitlines() start = max(0, ls - 1) end = min(len(lines), le) func["source"] = "\n".join( f"{i + 1}: {lines[i]}" for i in range(start, end) ) except (OSError, UnicodeDecodeError): func["source"] = "(could not read file)" if detail_level == "minimal": priorities = analysis.get("review_priorities", []) top_priorities = [ p.get("name", p.get("qualified_name", "")) for p in priorities[:3] ] result: dict[str, Any] = { "status": "ok", "summary": analysis.get("summary", ""), "risk_score": analysis.get("risk_score", 0.0), "changed_file_count": len(changed_files), "test_gap_count": len(analysis.get("test_gaps", [])), "review_priorities": top_priorities, } else: result = { "status": "ok", "changed_files": changed_files, **analysis, } result["_hints"] = generate_hints( "detect_changes", result, get_session() ) attach_context_savings(result, original_tokens=original_tokens) return result except Exception as exc: return {"status": "error", "error": str(exc)} finally: store.close()