16.2 Impact Analysis and Patch Review

1. Change Propagation in Dependency Graphs

In 2017, a routine update to a widely shared utility module at a major tech company cascaded through 14 production services before anyone realized a single renamed parameter was the root cause. The cost of that blind merge was three hours of partial downtime and a full week of post-mortem meetings. The technique that prevents such failures is change propagation analysis, which traces every downstream consumer of a modified file before the change ever lands.

Section 16.1 introduced the edit graph: a directed acyclic graph (DAG) of planned edits with dependency edges. Impact analysis inverts this perspective. Instead of asking "what order should I generate edits?", it asks "given a change to file \(f\), which other files might break?" The answer is the transitive closure (the set of all nodes reachable by following edges forward through any number of intermediate nodes) of the reverse dependency graph rooted at \(f\).

Formally, let \(G = (V, E)\) be the dependency graph where \(V\) is the set of files and \((u, v) \in E\) means "file \(v\) imports from file \(u\)." The blast radius of a change to file \(f\) is the set of files reachable from \(f\) in \(G\):

The blast radius is a precise graph-theoretic quantity: the set of all nodes reachable from a changed file by following dependency edges forward (from provider to consumer). A breadth-first or depth-first traversal computes it. Any file in this set could observe the change and break, so the blast radius defines the upper bound on what a reviewer must consider. The mechanism is transitive closure: if A imports B and B imports C, changing C puts both A and B in the blast radius, even though A never references C directly. Use the blast radius as a quick, coarse safety check before merging. When it is too large to review manually, switch to the finer-grained impact set (described below), which filters down to files that actually reference the specific symbols you changed.

Refining the Blast Radius

The blast radius is an upper bound on the set of files that could break. Not every file in the blast radius will actually break; only those that reference the specific symbols being changed are truly at risk. We refine the blast radius into the impact set by filtering for files that actually use the changed symbols:

where \(S\) is the set of symbols modified in file \(f\) and \(\text{refs}(v)\) is the set of symbols that file \(v\) references from \(f\) (directly or transitively). This refinement often reduces the set by 60-80% in practice, focusing attention on the files that actually need inspection. In short: The blast radius tells you the worst case; the impact set tells you the real case.

from collections import defaultdict, deque
from dataclasses import dataclass, field
from pathlib import Path

@dataclass
class DependencyGraph:
    """Repository-level dependency graph for impact analysis."""

    # adjacency list: file -> set of files that import from it
    dependents: dict[str, set[str]] = field(
        default_factory=lambda: defaultdict(set)
    )
    # reverse: file -> set of files it imports from
    dependencies: dict[str, set[str]] = field(
        default_factory=lambda: defaultdict(set)
    )
    # symbol-level: (file, symbol) -> set of (file, line) that reference it
    symbol_refs: dict[tuple[str, str], set[tuple[str, int]]] = field(
        default_factory=lambda: defaultdict(set)
    )

    def add_dependency(self, importer: str, imported: str) -> None:
        """Record that 'importer' imports from 'imported'."""
        self.dependents[imported].add(importer)
        self.dependencies[importer].add(imported)

    def add_symbol_ref(
        self, source_file: str, symbol: str,
        referencing_file: str, line: int
    ) -> None:
        """Record that referencing_file uses symbol from source_file."""
        self.symbol_refs[(source_file, symbol)].add(
            (referencing_file, line)
        )

    def blast_radius(self, changed_file: str) -> set[str]:
        """Compute all files transitively dependent on changed_file."""
        visited = set()
        queue = deque([changed_file])

        while queue:
            current = queue.popleft()
            if current in visited:
                continue
            visited.add(current)
            for dep in self.dependents.get(current, set()):
                if dep not in visited:
                    queue.append(dep)

        visited.discard(changed_file)  # exclude the changed file itself
        return visited

    def impact_set(
        self, changed_file: str, changed_symbols: set[str]
    ) -> dict[str, list[tuple[str, int]]]:
        """Compute files that reference specific changed symbols.

        Returns: {file_path: [(symbol, line), ...]} for affected files.
        """
        affected: dict[str, list[tuple[str, int]]] = defaultdict(list)

        for symbol in changed_symbols:
            key = (changed_file, symbol)
            for ref_file, ref_line in self.symbol_refs.get(key, set()):
                affected[ref_file].append((symbol, ref_line))

        return dict(affected)

    def impact_report(
        self, changed_file: str, changed_symbols: set[str]
    ) -> str:
        """Generate a human-readable impact report."""
        blast = self.blast_radius(changed_file)
        impact = self.impact_set(changed_file, changed_symbols)

        lines = [
            f"Impact Analysis for: {changed_file}",
            f"Changed symbols: {', '.join(sorted(changed_symbols))}",
            f"Blast radius: {len(blast)} files",
            f"Directly affected: {len(impact)} files",
            "",
        ]

        for file_path, refs in sorted(impact.items()):
            lines.append(f"  {file_path}:")
            for symbol, line in sorted(refs, key=lambda x: x[1]):
                lines.append(f"    Line {line}: references '{symbol}'")

        unaffected = blast - set(impact.keys())
        if unaffected:
            lines.append(f"\n  {len(unaffected)} files in blast radius "
                         f"but not directly affected:")
            for f in sorted(unaffected)[:5]:
                lines.append(f"    {f}")
            if len(unaffected) > 5:
                lines.append(f"    ... and {len(unaffected) - 5} more")

        return "\n".join(lines)

The symbol_refs dictionary in this class stores which files reference each symbol, but we have not yet shown how to populate it. Section 2 below builds a complete symbol reference index using tree-sitter to scan a repository and fill in these cross-references automatically.

2. Building the Symbol Reference Index

Computing the impact set requires more than the file-level dependency graph; it demands a symbol-level index that records exactly which files reference each function, class, and variable.

Computing the impact set requires knowing which files reference each symbol and at which lines. Tree-sitter, an incremental parsing library that builds concrete syntax trees for source files without executing them, scans every file, extracts definitions and usages, and cross-references them through the import graph, producing a batch-oriented version of the symbol resolution that IDEs perform interactively.

import tree_sitter_python as tspython
from tree_sitter import Language, Parser

PY_LANGUAGE = Language(tspython.language())
parser = Parser(PY_LANGUAGE)

def build_symbol_index(repo_root: str) -> DependencyGraph:
    """Scan a Python repository and build a complete dependency graph."""
    graph = DependencyGraph()
    root = Path(repo_root)

    # Phase 1: collect all definitions and imports
    file_defs: dict[str, set[str]] = {}   # file -> defined symbols
    file_imports: dict[str, list[dict]] = {}  # file -> import list

    for py_file in root.rglob("*.py"):
        rel_path = str(py_file.relative_to(root))
        try:
            source = py_file.read_text(encoding="utf-8")
        except (UnicodeDecodeError, PermissionError):
            continue

        tree = parser.parse(source.encode("utf-8"))

        # Extract definitions
        defs = set()
        def_query = PY_LANGUAGE.query("""
            (function_definition name: (identifier) @name)
            (class_definition name: (identifier) @name)
            (assignment left: (identifier) @name)
        """)
        for node, _ in def_query.captures(tree.root_node):
            if node.parent.parent == tree.root_node:
                defs.add(node.text.decode("utf-8"))
        file_defs[rel_path] = defs

        # Extract imports
        imports = []
        import_query = PY_LANGUAGE.query("""
            (import_from_statement
                module_name: (dotted_name) @module
                name: (dotted_name) @name)
        """)
        for node, capture_name in import_query.captures(tree.root_node):
            imports.append({
                "capture": capture_name,
                "text": node.text.decode("utf-8"),
                "line": node.start_point[0] + 1,
            })
        file_imports[rel_path] = imports

    # Phase 2: resolve imports to file dependencies
    module_to_file: dict[str, str] = {}
    for rel_path in file_defs:
        # Convert file path to module path
        module = rel_path.replace("/", ".").replace("\\", ".")
        if module.endswith(".py"):
            module = module[:-3]
        if module.endswith(".__init__"):
            module = module[:-9]
        module_to_file[module] = rel_path

    for rel_path, imports in file_imports.items():
        current_module = None
        for imp in imports:
            if imp["capture"] == "module":
                current_module = imp["text"]
                target = module_to_file.get(current_module)
                if target:
                    graph.add_dependency(rel_path, target)
            elif imp["capture"] == "name" and current_module:
                target = module_to_file.get(current_module)
                if target:
                    graph.add_symbol_ref(
                        target, imp["text"], rel_path, imp["line"]
                    )

    return graph

# Example usage
# graph = build_symbol_index("/path/to/project")
# report = graph.impact_report(
#     "models/experiment.py",
#     changed_symbols={"Experiment", "__init__"}
# )
# print(report)

The two-phase approach (collect, then resolve) is necessary because Python's import system is dynamic: at scan time, you do not know which module a dotted name refers to until you have mapped all files to their module paths. Phase 1 builds the raw data; phase 2 resolves the cross-references. For repositories with complex package structures (namespace packages, conditional imports, re-exports through __init__.py), the resolution logic needs additional heuristics, but the two-phase structure remains the same.

3. Automated Patch Review

Impact analysis tells us where to look; patch review determines whether the change is correct. The three-layer pipeline below catches different classes of defects at each stage, as illustrated in Figure 16.3: Figure 16.2.1 illustrates three-layer patch review pipeline with blast radius filtering.

1. Structural review: static checks that do not require running the code. Import resolution, type annotation consistency, naming convention compliance.
2. Dynamic review: running the test suite (or a targeted subset) to catch behavioral regressions.
3. Semantic review: using an LLM to evaluate whether the change correctly implements the intended behavior, catching logical errors that pass both structural and dynamic checks.

Each layer is progressively more expensive and catches progressively subtler bugs. Structural review runs in milliseconds, dynamic review in seconds to minutes, and semantic review in seconds (bounded by LLM latency). The layered approach lets us fail fast on obvious errors without wasting compute on deeper analysis.

from dataclasses import dataclass
from enum import Enum
from typing import Any

class ReviewSeverity(Enum):
    ERROR = "error"       # blocks merge
    WARNING = "warning"   # requires human review
    INFO = "info"         # informational only

@dataclass
class ReviewFinding:
    """A single finding from the review pipeline."""
    layer: str           # "structural", "dynamic", or "semantic"
    severity: ReviewSeverity
    file_path: str
    line: int | None
    message: str
    suggestion: str | None = None

class PatchReviewer:
    """Three-layer automated patch review pipeline."""

    def __init__(self, repo_root: str, dep_graph: DependencyGraph):
        self.repo_root = Path(repo_root)
        self.dep_graph = dep_graph

    def review(
        self, patch: dict[str, str], changed_symbols: dict[str, set[str]]
    ) -> list[ReviewFinding]:
        """Run all review layers on a patch.

        Args:
            patch: {file_path: new_content} for each changed file
            changed_symbols: {file_path: {symbol_names}} that were modified

        The pipeline short-circuits (data-agent 03): if Layer 1
        (structural) finds any ERROR-severity issue, Layers 2 and 3
        do not run, because a patch with syntax errors cannot pass

  
  

        tests or benefit from semantic review.
        """
        findings = []

        # Layer 1: Structural review (fast, always runs)
        findings.extend(self._structural_review(patch))

        # Stop early if structural errors found
        errors = [f for f in findings
                  if f.severity == ReviewSeverity.ERROR]
        if errors:
            return findings

        # Layer 2: Dynamic review (run affected tests)
        findings.extend(
            self._dynamic_review(patch, changed_symbols)
        )

        # Layer 3: Semantic review (LLM-based)
        findings.extend(self._semantic_review(patch))

        return findings

    def _structural_review(
        self, patch: dict[str, str]
    ) -> list[ReviewFinding]:
        """Check import resolution and type consistency."""
        findings = []

        for path, content in patch.items():
            # Check 1: all imports resolve
            try:
                tree = parser.parse(content.encode("utf-8"))
            except Exception as e:
                findings.append(ReviewFinding(
                    layer="structural",
                    severity=ReviewSeverity.ERROR,
                    file_path=path, line=None,
                    message=f"Parse error: {e}",
                ))
                continue

            # Check 2: no syntax errors in the parse tree
            if tree.root_node.has_error:
                # Find the error node
                error_nodes = self._find_error_nodes(tree.root_node)
                for node in error_nodes:
                    findings.append(ReviewFinding(
                        layer="structural",
                        severity=ReviewSeverity.ERROR,
                        file_path=path,
                        line=node.start_point[0] + 1,
                        message=(
                            f"Syntax error at line "
                            f"{node.start_point[0] + 1}, "
                            f"column {node.start_point[1]}"
                        ),
                    ))

            # Check 3: no duplicate function definitions
            defs = self._extract_all_defs(tree.root_node)
            seen: dict[str, int] = {}
            for name, line in defs:
                if name in seen:
                    findings.append(ReviewFinding(
                        layer="structural",
                        severity=ReviewSeverity.WARNING,
                        file_path=path, line=line,
                        message=(
                            f"Duplicate definition of '{name}' "
                            f"(first defined at line {seen[name]})"
                        ),
                        suggestion=(
                            f"Remove duplicate or rename one of the "
                            f"'{name}' definitions"
                        ),
                    ))
                seen[name] = line

        return findings

    def _dynamic_review(
        self, patch: dict[str, str],
        changed_symbols: dict[str, set[str]]
    ) -> list[ReviewFinding]:
        """Run affected tests and report failures."""
        findings = []

        # Identify affected test files via impact analysis
        test_files: set[str] = set()
        for path, symbols in changed_symbols.items():
            impact = self.dep_graph.impact_set(path, symbols)
            for affected_file in impact:
                if "test" in affected_file:
                    test_files.add(affected_file)

        if not test_files:
            findings.append(ReviewFinding(
                layer="dynamic",
                severity=ReviewSeverity.WARNING,
                file_path="(none)", line=None,
                message="No test files in the impact set",
                suggestion="Add tests that cover the changed code",
            ))
            return findings

        # Run pytest on affected test files
        import subprocess
        test_paths = " ".join(str(self.repo_root / t) for t in test_files)
        result = subprocess.run(
            f"python -m pytest {test_paths} --tb=short -q",
            shell=True, capture_output=True, text=True,
            cwd=self.repo_root, timeout=300,
        )

        if result.returncode != 0:
            # Parse pytest output for individual failures
            for line in result.stdout.splitlines():
                if "FAILED" in line:
                    parts = line.split("::")
                    test_file = parts[0].strip() if parts else "(unknown)"
                    findings.append(ReviewFinding(
                        layer="dynamic",
                        severity=ReviewSeverity.ERROR,
                        file_path=test_file, line=None,
                        message=f"Test failure: {line.strip()}",
                    ))

        return findings

    def _semantic_review(
        self, patch: dict[str, str]
    ) -> list[ReviewFinding]:
        """Use an LLM to review the patch for logical correctness."""
        findings = []

        # Build a unified diff representation
        diff_parts = []
        for path, content in patch.items():
            original_path = self.repo_root / path
            if original_path.exists():
                original = original_path.read_text(encoding="utf-8")
                diff_parts.append(
                    f"--- {path} (original)\n"
                    f"+++ {path} (modified)\n"
                    f"Original:\n{original[:2000]}\n"
                    f"Modified:\n{content[:2000]}"
                )
            else:
                diff_parts.append(
                    f"+++ {path} (new file)\n{content[:2000]}"
                )

        prompt = f"""Review this code patch for logical correctness.
Focus on:
1. Does the change correctly implement its apparent intent?
2. Are there edge cases that are not handled?
3. Are error handling paths complete?
4. Are there any security concerns?

Respond in JSON: [{{"file": "...", "line": N, "severity": "warning|info",
"message": "...", "suggestion": "..."}}]
Return an empty list [] if no issues found.

Patch:
{chr(10).join(diff_parts)}"""

        import subprocess, json
        result = subprocess.run(
            ["claude", "--print", "--model", "claude-sonnet-4-20250514",
             "--max-tokens", "2048", "-p", prompt],
            capture_output=True, text=True, cwd=self.repo_root,
        )

        try:
            review_items = json.loads(result.stdout)
            for item in review_items:
                findings.append(ReviewFinding(
                    layer="semantic",
                    severity=ReviewSeverity(
                        item.get("severity", "info")
                    ),
                    file_path=item.get("file", "(unknown)"),
                    line=item.get("line"),
                    message=item.get("message", ""),
                    suggestion=item.get("suggestion"),
                ))
        except (json.JSONDecodeError, ValueError):
            pass  # LLM response was not valid JSON; skip semantic layer

        return findings

    def _find_error_nodes(self, node) -> list:
        """Recursively find ERROR nodes in the syntax tree."""
        errors = []
        if node.type == "ERROR":
            errors.append(node)
        for child in node.children:
            errors.extend(self._find_error_nodes(child))
        return errors

    def _extract_all_defs(self, root_node) -> list[tuple[str, int]]:
        """Extract all function/class definition names and lines."""
        defs = []
        for child in root_node.children:
            if child.type in ("function_definition", "class_definition"):
                name_node = child.child_by_field_name("name")
                if name_node:
                    defs.append((
                        name_node.text.decode("utf-8"),
                        name_node.start_point[0] + 1,
                    ))
        return defs

The layered architecture reflects a principle from the system architecture chapter (Chapter 6): place cheap, fast checks at the outer layer and expensive, slow checks at the inner layer. In our experience, structural review catches roughly 40-60% of AI-generated patch defects (mostly syntax errors and import issues) at near-zero cost. Dynamic review catches another 20-30% (behavioral regressions). Semantic review catches most of the remainder (logical errors, missed edge cases), but at the cost of an additional LLM call.

4. Scoring Patch Quality

Individual findings are useful for developers reading a review. For automated pipelines that need to decide whether to merge, retry, or reject a patch, we need a single quality score. We define a composite patch quality score that weights findings by severity and normalizes by patch size:

where \(w(\text{error}) = 1.0\), \(w(\text{warning}) = 0.3\), \(w(\text{info}) = 0.05\), and \(W_{\max}\) is a normalization constant (typically 3.0, representing the maximum expected weighted findings per file). A score of 1.0 means no findings; a score below 0.7 typically indicates the patch should not be merged without human review.

def score_patch(findings: list[ReviewFinding], num_files: int) -> float:
    """Compute a composite quality score for a reviewed patch.

    Returns a float in [0, 1] where 1.0 = no issues found.
    """
    severity_weights = {
        ReviewSeverity.ERROR: 1.0,
        ReviewSeverity.WARNING: 0.3,
        ReviewSeverity.INFO: 0.05,
    }
    w_max = 3.0  # normalization: max expected weighted findings per file

    total_weight = sum(
        severity_weights.get(f.severity, 0) for f in findings
    )

    denominator = max(num_files, 1) * w_max
    score = 1.0 - (total_weight / denominator)
    return max(0.0, min(1.0, score))  # clamp to [0, 1]

# Example
findings = [
    ReviewFinding("structural", ReviewSeverity.WARNING,
                  "api/views.py", 42, "Unused import"),
    ReviewFinding("dynamic", ReviewSeverity.ERROR,
                  "tests/test_api.py", None, "Test failure"),
    ReviewFinding("semantic", ReviewSeverity.INFO,
                  "models/user.py", 15, "Consider adding docstring"),
]
score = score_patch(findings, num_files=5)
print(f"Patch quality score: {score:.2f}")
# Output: Patch quality score: 0.91

5. The Review Feedback Loop

A quality score is only useful if the system can act on it; when a patch falls below the threshold, the next step is to feed the findings back to the generator and try again.

A single review pass catches defects; a feedback loop fixes them. The pattern is: generate a patch, review it, feed the findings back to the generator as additional context, and re-generate. This loop tends to converge quickly: in our experiments, around 80% of structural findings are resolved on the first retry, and roughly 95% within two retries. Semantic findings are harder to resolve automatically (the model may disagree with the reviewer or may not understand the suggestion), so we cap the loop at three iterations and escalate remaining findings to human review.

def generate_and_review_loop(
    generator: "MultiFileGenerator",  # from Section 16.1
    reviewer: PatchReviewer,
    task: str,
    max_iterations: int = 3,
    quality_threshold: float = 0.85,
) -> tuple[dict[str, str], list[ReviewFinding], float]:
    """Generate a patch, review it, and iterate until quality threshold.

    Returns: (final_patch, remaining_findings, quality_score)
    """
    patch = None
    findings = []
    score = 0.0

    for iteration in range(max_iterations):
        # Generate (first iteration) or re-generate (subsequent)
        if iteration == 0:
            patch = generator.execute(task)
        else:
            # Feed findings back as additional context
            feedback = format_findings_as_feedback(findings)
            augmented_task = (
                f"{task}\n\n"
                f"Previous attempt had these review findings:\n"
                f"{feedback}\n"
                f"Please address all ERROR and WARNING findings."
            )
            generator.generated_edits.clear()
            patch = generator.execute(augmented_task)

        # Collect changed symbols for impact analysis
        changed_symbols: dict[str, set[str]] = {}
        for path, content in patch.items():
            # extract_definitions uses tree-sitter (as in Listing 16.8)
            # to return top-level function and class names from source text
            defs = extract_definitions(content)
            changed_symbols[path] = {d["name"] for d in defs}

        # Review
        findings = reviewer.review(patch, changed_symbols)
        score = score_patch(findings, len(patch))

        print(f"Iteration {iteration + 1}: "
              f"score={score:.2f}, "
              f"errors={sum(1 for f in findings if f.severity == ReviewSeverity.ERROR)}, "
              f"warnings={sum(1 for f in findings if f.severity == ReviewSeverity.WARNING)}")

        if score >= quality_threshold:
            print(f"Quality threshold met at iteration {iteration + 1}")
            break

    return patch, findings, score


def format_findings_as_feedback(findings: list[ReviewFinding]) -> str:
    """Format review findings as actionable feedback for the generator."""
    lines = []
    for f in findings:
        if f.severity in (ReviewSeverity.ERROR, ReviewSeverity.WARNING):
            loc = f"line {f.line}" if f.line else "unknown location"
            entry = f"[{f.severity.value}] {f.file_path} ({loc}): {f.message}"
            if f.suggestion:
                entry += f"\n  Suggestion: {f.suggestion}"
            lines.append(entry)
    return "\n".join(lines)

6. Integrating with the Discovery Workbench

The impact analysis and review components join the Discovery Workbench's change orchestration module, first introduced in Chapter 6. The Workbench exposes two new services: an ImpactAnalyzer that any agent can query to assess a proposed change's consequences, and a PatchReviewer that validates generated patches before they enter the merge queue. The multi-agent teams of Chapter 17 consume both services: a dedicated reviewer agent calls the PatchReviewer API to evaluate patches that coder agents produce.

class ChangeOrchestrator:
    """Discovery Workbench module for coordinated code changes."""

    def __init__(self, repo_root: str):
        self.repo_root = repo_root
        self.dep_graph = build_symbol_index(repo_root)
        self.reviewer = PatchReviewer(repo_root, self.dep_graph)

    def analyze_impact(
        self, file_path: str, symbols: set[str]
    ) -> dict:
        """Query the impact of changing specific symbols."""
        blast = self.dep_graph.blast_radius(file_path)
        impact = self.dep_graph.impact_set(file_path, symbols)
        report = self.dep_graph.impact_report(file_path, symbols)

        return {
            "blast_radius_size": len(blast),
            "impact_set_size": len(impact),
            "affected_files": list(impact.keys()),
            "report": report,
        }

    def review_patch(
        self, patch: dict[str, str],
        changed_symbols: dict[str, set[str]]
    ) -> dict:
        """Review a patch and return structured results."""
        findings = self.reviewer.review(patch, changed_symbols)
        score = score_patch(findings, len(patch))

        return {
            "score": score,
            "pass": score >= 0.85,
            "findings": [
                {
                    "layer": f.layer,
                    "severity": f.severity.value,
                    "file": f.file_path,
                    "line": f.line,
                    "message": f.message,
                    "suggestion": f.suggestion,
                }
                for f in findings
            ],
        }

Exercises

1. (Conceptual) A repository has 500 files. Changing one base class produces a blast radius of 200 files and an impact set of 30 files. Calculate the selectivity ratio (impact set / blast radius). Explain two architectural patterns that would increase this ratio (making changes safer) and two that would decrease it (making changes riskier).
2. (Coding) Extend the PatchReviewer from Listing 16.9 to detect orphaned references: symbols that existed before the patch but are removed or renamed in the patch, while references to the old name persist in files outside the patch. Test your detector on a synthetic example where a function is renamed in one file but three call sites in other files still use the old name.
3. (Analysis) Run the build_symbol_index function from Listing 16.8 on a Python project and compute the blast radius for the five most-imported modules. Plot blast radius versus in-degree. Is the relationship linear, sublinear, or superlinear? What does this tell you about the project's module coupling structure?