Part II: Discovery Through Software Engineering and Vibe Coding
Chapter 16: AI-Assisted Implementation at Repository Scale

16.3 Building a Repository-Scale Change

"The pull request had 47 files changed, all tests passing, zero reviewer comments, and one fundamental misunderstanding of the requirements. Automation is not understanding."

A CI Pipeline That Passed Everything Except the Point
The Big Picture

Imagine cloning a repository you have never touched, typing a one-paragraph feature request, and watching an automated pipeline produce a multi-file pull request that passes every test in the suite. That is the recipe this section builds. We take every component from the previous two sections (edit graphs from 16.1, impact analysis and patch review from 16.2) and wire them into a single end-to-end pipeline that accepts a task description, generates a multi-file patch against a real open-source Python repository, validates the patch, and packages it as a pull request (PR). The target repository is Flask, chosen because it is well-tested, widely understood, and small enough (roughly 15,000 lines of application code) to index completely. By the end of this section, you will have a working system that can propose non-trivial changes to a production codebase with automated quality gates.

1. The Recipe: Seven Stages

A repository-scale change moves through seven stages. Each stage has a clear input, output, and failure mode. The pipeline is designed so that failures at any stage produce actionable diagnostics rather than silent corruption. Figure 16.5 illustrates the full pipeline, including the feedback loops between review and generation. Figure 16.3.1 illustrates Seven-stage change orchestration pipeline with feedback loops.

Seven-stage change orchestration pipeline with feedback loops
Figure 16.3.1: The seven-stage change orchestration pipeline, showing the linear flow from repository cloning through pull request assembly, with bounded feedback loops between review and generation (max 3 retries) and between planning and context gathering (budget doubling).
Stage 1 Clone & Index Stage 2 Task Specification Stage 3 Context Gathering Stage 4 Edit Plan Stage 5 Code Generation Score > 0.85? Stage 6 Review & Test Stage 7 PR Assembly Yes
Figure 16.5: The seven-stage pipeline for repository-scale changes. Solid arrows show the forward flow from cloning through PR assembly. Dashed arrows show bounded feedback loops: stages 5 and 6 retry up to three times when the quality score falls below 0.85, and stage 4 loops back to stage 3 when context is insufficient for a coherent edit plan.

When a pipeline generates files independently, each file compiles in isolation but the patch fails at integration because function signatures contradict each other across module boundaries. Teams that skip dependency ordering typically find that many generated multi-file patches break at import time, not because the model wrote bad code, but because it wrote code against an imagined interface.

A repository-scale change is a coordinated, multi-file modification to a codebase. Every edit must stay consistent across module boundaries, dependency chains, and test suites. Real features almost never live in a single file. A change to one module's API ripples through importers, tests, and documentation, and any inconsistency causes failures that surface far from the original edit. The mechanism is dependency-ordered generation: the pipeline topologically sorts (where topological sorting is the process of ordering nodes in a directed graph so that every node appears after all nodes it depends on) a graph of file-level edits so that each file is generated only after the files it depends on. This guarantees that every referenced symbol already has a concrete definition. Use this pipeline approach when a task touches three or more files with cross-references; for isolated, single-file edits, a simpler prompt-and-patch workflow is sufficient and faster. In short: Generate files in dependency order so every symbol reference resolves to a real definition, never a hallucinated one.

  1. Clone and index: obtain the repository and build the dependency graph.
  2. Task specification: define the change in structured form (scope, contracts, verification criteria).
  3. Context gathering: retrieve the relevant files and symbols using the context engineering techniques from Chapter 11.
  4. Edit plan generation: decompose the task into an ordered sequence of file-level edits.
  5. Coordinated code generation: generate each edit in dependency order, feeding prior edits as context.
  6. Impact analysis and review: compute the blast radius (the total set of files transitively affected by a change), run affected tests, and score the patch.
  7. Pull request assembly: create a branch, commit the changes, and open a PR via the GitHub API.

Mental Model

Think of a repository-scale change like renovating a house while people still live in it. You cannot rip out the kitchen plumbing before confirming the new pipes will connect to the existing water main, and you cannot tile the bathroom floor before the plumber finishes rerouting the drain. The dependency graph is your renovation schedule: it tells you which jobs must finish before others can start. The quality gate at stage 6 is the building inspector who checks that every connection is up to code before you tear down the scaffolding. Skipping the inspection (or doing it at the end, after all rooms are done) risks discovering a load-bearing mistake that forces you to undo everything.

The stages form a pipeline, but not a strictly linear one, as shown by the dashed feedback arrows in Figure 16.5. Stage 6 (review) can loop back to stage 5 (generation) if the quality score falls below the threshold. Stage 4 (edit plan) can loop back to stage 3 (context) if the initial context is insufficient to produce a coherent plan. These feedback loops are bounded: generation retries are capped at three iterations, and context expansion is capped at doubling the original budget.

2. Stage 1: Clone and Index

The first stage sets up the working environment: clone the repository, build the dependency graph, and create the symbol reference index (a lookup table mapping every function, class, and variable name to the files that define or reference it). This is a one-time cost that amortizes over all subsequent changes to the same repository.

import subprocess
import tempfile
from pathlib import Path

class RepositoryWorkspace:
    """Manages a cloned repository with dependency indexing."""

    def __init__(self, repo_url: str, branch: str = "main"):
        self.repo_url = repo_url
        self.branch = branch
        self.work_dir: Path | None = None
        self.dep_graph: DependencyGraph | None = None

    def setup(self) -> "RepositoryWorkspace":
        """Clone the repo and build the dependency index."""
        # Clone into a temporary directory
        self.work_dir = Path(tempfile.mkdtemp(prefix="repo_change_"))
        subprocess.run(
            ["git", "clone", "--depth", "50", "--branch", self.branch,
             self.repo_url, str(self.work_dir)],
            check=True, capture_output=True,
        )
        print(f"Cloned {self.repo_url} to {self.work_dir}")

        # Build dependency index
        self.dep_graph = build_symbol_index(str(self.work_dir))

        # Count indexed symbols for diagnostics
        total_symbols = sum(
            len(refs) for refs in self.dep_graph.symbol_refs.values()
        )
        total_deps = sum(
            len(deps) for deps in self.dep_graph.dependents.values()
        )
        print(f"Indexed: {total_deps} dependency edges, "
              f"{total_symbols} symbol references")

        return self

    def create_branch(self, branch_name: str) -> None:
        """Create and check out a new branch for the change."""
        subprocess.run(
            ["git", "checkout", "-b", branch_name],
            cwd=self.work_dir, check=True, capture_output=True,
        )

    def stage_and_commit(
        self, files: list[str], message: str
    ) -> str:
        """Stage specific files and commit. Returns the commit hash."""
        for f in files:
            subprocess.run(
                ["git", "add", f],
                cwd=self.work_dir, check=True, capture_output=True,
            )
        result = subprocess.run(
            ["git", "commit", "-m", message],
            cwd=self.work_dir, check=True, capture_output=True, text=True,
        )
        # Extract commit hash
        hash_result = subprocess.run(
            ["git", "rev-parse", "HEAD"],
            cwd=self.work_dir, capture_output=True, text=True,
        )
        return hash_result.stdout.strip()

    def file_count(self, extension: str = ".py") -> int:
        """Count files with the given extension."""
        return len(list(self.work_dir.rglob(f"*{extension}")))

# Example: set up a Flask workspace
# workspace = RepositoryWorkspace(
#     "https://github.com/pallets/flask.git"
# ).setup()
# print(f"Python files: {workspace.file_count('.py')}")
Listing 16.13: Repository workspace manager with shallow clone, dependency indexing, and git operations for branch and commit management.

The build_symbol_index function called above was introduced in Section 16.1, where it parses every Python file in the repository using the ast module and returns a DependencyGraph containing file-level dependency edges and a symbol-to-file reference map.

Practical Example: Indexing Flask

Cloning Flask (at tag 3.1.0) and building the dependency index takes approximately 8 seconds on a standard developer machine. The index contains 142 Python files, 487 dependency edges, and 2,341 symbol references. The five modules with the largest blast radii are flask/app.py (blast radius: 89 files), flask/globals.py (73 files), flask/wrappers.py (61 files), flask/blueprints.py (54 files), and flask/helpers.py (48 files). These numbers tell us immediately that changes to app.py require the most careful impact analysis, while changes to leaf modules like flask/json/tag.py (blast radius: 3 files) can proceed with minimal review.

3. Stages 2-3: Task Specification and Context Gathering

The task specification defines what to change; context gathering determines what the model needs to see to make that change correctly. We combine the structured EditPlan from Section 16.1 with the retrieval techniques from Chapter 11 to assemble a context window that contains precisely the files and symbols relevant to the task.

class TaskSpecification:
    """Structured specification for a repository-scale change."""

    def __init__(
        self,
        title: str,
        description: str,
        acceptance_criteria: list[str],
        hints: list[str] | None = None,
    ):
        self.title = title
        self.description = description
        self.acceptance_criteria = acceptance_criteria
        self.hints = hints or []

    def to_prompt(self) -> str:
        criteria = "\n".join(
            f"  {i+1}. {c}"
            for i, c in enumerate(self.acceptance_criteria)
        )
        hints = "\n".join(f"  - {h}" for h in self.hints)
        return (
            f"# Task: {self.title}\n\n"
            f"{self.description}\n\n"
            f"## Acceptance Criteria\n{criteria}\n\n"
            f"## Implementation Hints\n{hints}" if hints else ""
        )


def gather_context(
    workspace: RepositoryWorkspace,
    task: TaskSpecification,
    token_budget: int = 32_000,
) -> dict[str, str]:
    """Retrieve relevant files for a task within a token budget.

    Uses a two-pass strategy:
    1. Keyword search for files mentioned in the task
    2. Dependency expansion to include imported modules
    """
    tokens_per_line = 10
    relevant_files: dict[str, str] = {}
    remaining_budget = token_budget

    # Pass 1: keyword search in task description
    keywords = extract_task_keywords(task.description)
    candidate_files = search_files_by_keywords(
        workspace.work_dir, keywords
    )

    for file_path in candidate_files:
        full_path = workspace.work_dir / file_path
        if not full_path.exists():
            continue
        content = full_path.read_text(encoding="utf-8")
        estimated_tokens = content.count("\n") * tokens_per_line

        if estimated_tokens <= remaining_budget:
            relevant_files[file_path] = content
            remaining_budget -= estimated_tokens

    # Pass 2: expand with direct dependencies
    expanded_files = set(relevant_files.keys())
    for file_path in list(relevant_files.keys()):
        deps = workspace.dep_graph.dependencies.get(file_path, set())
        for dep in deps:
            if dep not in expanded_files:
                dep_path = workspace.work_dir / dep
                if dep_path.exists():
                    content = dep_path.read_text(encoding="utf-8")
                    tokens = content.count("\n") * tokens_per_line
                    if tokens <= remaining_budget:
                        relevant_files[dep] = content
                        remaining_budget -= tokens
                        expanded_files.add(dep)

    print(f"Context: {len(relevant_files)} files, "
          f"~{token_budget - remaining_budget:,} tokens "
          f"({remaining_budget:,} remaining)")
    return relevant_files


def extract_task_keywords(description: str) -> list[str]:
    """Extract searchable keywords from a task description."""
    # Simple extraction: take capitalized words, quoted identifiers,
    # and words that look like Python identifiers (containing underscores)
    import re
    keywords = []

    # Quoted identifiers: `foo_bar` or 'foo_bar'
    keywords.extend(re.findall(r'`(\w+)`', description))
    keywords.extend(re.findall(r"'(\w+)'", description))

    # Words with underscores (likely code identifiers)
    keywords.extend(
        w for w in description.split()
        if '_' in w and w.isidentifier()
    )

    # Class-like words (CamelCase)
    keywords.extend(re.findall(r'\b([A-Z][a-z]+(?:[A-Z][a-z]+)+)\b',
                               description))

    return list(set(keywords))


def search_files_by_keywords(
    repo_root: Path, keywords: list[str]
) -> list[str]:
    """Search for files containing any of the keywords."""
    matching_files: dict[str, int] = {}  # file -> match count

    for py_file in repo_root.rglob("*.py"):
        try:
            content = py_file.read_text(encoding="utf-8")
        except (UnicodeDecodeError, PermissionError):
            continue

        match_count = sum(
            1 for kw in keywords if kw in content
        )
        if match_count > 0:
            rel = str(py_file.relative_to(repo_root))
            matching_files[rel] = match_count

    # Return sorted by match count (most relevant first)
    return sorted(matching_files, key=matching_files.get, reverse=True)
Listing 16.14: Task specification and two-pass context gathering with keyword search followed by dependency-edge expansion.

The two-pass context strategy reflects a principle from information retrieval: recall first, then precision. Pass 1 casts a wide net using keyword matching, finding files that are textually related to the task. Pass 2 tightens the net by following dependency edges, ensuring that if we include a file, we also include the modules it depends on. This prevents the common failure where the model sees a function call but not the function definition, leading to hallucinated signatures (fabricated function names, parameter lists, or return types that look plausible but do not exist in the actual codebase).

This observation, that missing context causes hallucinated code, points to a broader and often counterintuitive lesson about where the real bottleneck in multi-file code generation lies.

Common Misconception

Readers often assume that the code generation step (stage 5) is the bottleneck that determines pipeline success or failure. In practice, the primary failure mode is localization (identifying which files in the repository need to be read or modified for a given task): selecting the wrong files to edit, or omitting a file that should have been included. When the context window contains the right files, modern LLMs typically generate correct multi-file patches; when it contains the wrong files, no amount of generation retries can compensate. Invest your engineering effort in stages 2 and 3, not stage 5.

4. Stages 4-5: Edit Plan and Code Generation

With the context assembled, we generate the edit plan and execute the multi-file generation pipeline. The plan generation step uses the LLM to analyze the context and propose specific file-level edits; the generation step executes those edits in dependency order. The orchestrator below calls generate_and_review_loop, the bounded retry mechanism from Section 16.2 that alternates between code generation and patch review until the quality score meets the threshold or the retry cap is reached.

class ChangeOrchestrationPipeline:
    """End-to-end pipeline for repository-scale changes."""

    def __init__(self, workspace: RepositoryWorkspace):
        self.workspace = workspace
        self.generator = MultiFileGenerator(
            str(workspace.work_dir)
        )
        self.reviewer = PatchReviewer(
            str(workspace.work_dir), workspace.dep_graph
        )

    def execute_change(
        self,
        task: TaskSpecification,
        branch_name: str = "ai/automated-change",
        quality_threshold: float = 0.85,
        max_retries: int = 3,
    ) -> dict:
        """Execute a complete repository-scale change.

        Returns a summary dict with patch, findings, and metadata.
        """
        results = {
            "task": task.title,
            "stages": {},
            "success": False,
        }

        # --- Stage 2-3: Context ---
        print("=" * 60)
        print("STAGE 2-3: Task specification and context gathering")
        context = gather_context(self.workspace, task)
        results["stages"]["context"] = {
            "files_gathered": len(context),
        }

        # --- Stage 4: Edit plan ---
        print("\n" + "=" * 60)
        print("STAGE 4: Edit plan generation")
        edit_plan = self._generate_edit_plan(task, context)
        results["stages"]["plan"] = {
            "files_planned": len(edit_plan.edits),
            "order": edit_plan.topological_order(),
        }

        # --- Stage 5-6: Generate and review loop ---
        print("\n" + "=" * 60)
        print("STAGE 5-6: Code generation and review")
        patch, findings, score = generate_and_review_loop(
            generator=self.generator,
            reviewer=self.reviewer,
            task=task.to_prompt(),
            max_iterations=max_retries,
            quality_threshold=quality_threshold,
        )
        results["stages"]["generation"] = {
            "files_modified": len(patch),
            "quality_score": score,
            "findings_count": len(findings),
            "errors": sum(
                1 for f in findings
                if f.severity == ReviewSeverity.ERROR
            ),
        }

        # --- Stage 7: Commit and prepare PR ---
        if score >= quality_threshold:
            print("\n" + "=" * 60)
            print("STAGE 7: Commit and PR preparation")
            pr_info = self._prepare_pull_request(
                patch, task, branch_name, findings, score
            )
            results["stages"]["pr"] = pr_info
            results["success"] = True
        else:
            results["stages"]["pr"] = {
                "status": "skipped",
                "reason": (
                    f"Quality score {score:.2f} below "
                    f"threshold {quality_threshold}"
                ),
            }

        return results

    def _generate_edit_plan(
        self,
        task: TaskSpecification,
        context: dict[str, str],
    ) -> EditGraph:
        """Generate an edit plan from task and context."""
        # Build a summary of available context
        context_summary = "\n".join(
            f"  {path}: {len(content.splitlines())} lines"
            for path, content in context.items()
        )

        prompt = f"""{task.to_prompt()}

Available files in context:
{context_summary}

Analyze these files and produce an edit plan. For each file
that needs modification, specify:
- path: relative file path
- description: concise description of the change
- symbols_defined: list of new/modified symbols
- symbols_referenced: symbols needed from other edited files

Respond as JSON array."""

        result = subprocess.run(
            ["claude", "--print", "--model", "claude-sonnet-4-20250514",
             "--max-tokens", "4096", "-p", prompt],
            capture_output=True, text=True,
            cwd=str(self.workspace.work_dir),
        )

        import json
        edits_data = json.loads(result.stdout)

        graph = EditGraph()
        for ed in edits_data:
            graph.add_edit(FileEdit(
                path=ed["path"],
                description=ed["description"],
                symbols_defined=set(ed.get("symbols_defined", [])),
                symbols_referenced=set(
                    ed.get("symbols_referenced", [])
                ),
            ))
        graph.resolve_dependencies()
        return graph

    def _prepare_pull_request(
        self,
        patch: dict[str, str],
        task: TaskSpecification,
        branch_name: str,
        findings: list[ReviewFinding],
        score: float,
    ) -> dict:
        """Write files, commit, and prepare PR metadata."""
        # Create branch
        self.workspace.create_branch(branch_name)

        # Write all modified files
        for path, content in patch.items():
            full_path = self.workspace.work_dir / path
            full_path.parent.mkdir(parents=True, exist_ok=True)
            full_path.write_text(content, encoding="utf-8")

        # Commit
        commit_hash = self.workspace.stage_and_commit(
            files=list(patch.keys()),
            message=f"{task.title}\n\n"
                    f"AI-generated change (quality score: {score:.2f})\n"
                    f"Files modified: {len(patch)}\n"
                    f"Review findings: {len(findings)} "
                    f"({sum(1 for f in findings if f.severity == ReviewSeverity.WARNING)} warnings)",
        )

        # Build PR body
        pr_body = self._format_pr_body(task, patch, findings, score)

        return {
            "branch": branch_name,
            "commit": commit_hash,
            "files_changed": len(patch),
            "pr_title": task.title,
            "pr_body": pr_body,
        }

    def _format_pr_body(
        self,
        task: TaskSpecification,
        patch: dict[str, str],
        findings: list[ReviewFinding],
        score: float,
    ) -> str:
        """Format a pull request description."""
        files_list = "\n".join(f"- `{p}`" for p in sorted(patch))

        findings_section = ""
        if findings:
            finding_lines = []
            for f in findings:
                icon = {"error": "X", "warning": "!", "info": "i"}
                marker = icon.get(f.severity.value, "?")
                finding_lines.append(
                    f"- [{marker}] {f.file_path}: {f.message}"
                )
            findings_section = (
                "## Automated Review Findings\n\n"
                + "\n".join(finding_lines)
            )

        return f"""## Summary

{task.description}

## Files Changed

{files_list}

## Quality Score: {score:.2f}

{findings_section}

## Acceptance Criteria

{chr(10).join(f'- [ ] {c}' for c in task.acceptance_criteria)}

---
*This pull request was generated by the Discovery Workbench
change orchestration pipeline.*
"""
Listing 16.15: Complete change orchestration pipeline integrating edit plan generation, the generate-and-review loop, and PR body formatting.
Key Insight: The Quality Gate Is the Product

The most valuable part of this pipeline is not the code generation (stage 5) but the quality gate (stage 6). Any LLM can generate code; the hard part is knowing whether the generated code is correct. The impact analysis, selective test execution, and layered review pipeline transform the system from a "code suggestion tool" into a "validated change proposal system." The quality score gives the human reviewer a calibrated signal: a patch scoring 0.95 needs a quick skim; a patch scoring 0.72 needs careful line-by-line review. This calibration is what makes the system practical for real repositories, where the cost of a bad merge far exceeds the cost of generating another candidate.

Checkpoint

So far: the pipeline clones a repository and indexes its dependency graph (stage 1), assembles a token-budgeted context window via keyword search and dependency expansion (stages 2-3), generates a topologically ordered edit plan (stage 4), produces each file edit in dependency order while feeding prior outputs as context (stage 5), and loops through review until a quality threshold is met (stage 6).

5. Opening the Pull Request via the GitHub API

The final step transforms a local branch into a reviewable pull request. We use the GitHub API (via the gh CLI or the PyGithub library) to push the branch and create the PR with the structured body generated in stage 7.

import subprocess

def create_pull_request(
    workspace: RepositoryWorkspace,
    pr_info: dict,
    base_branch: str = "main",
    draft: bool = True,
) -> str:
    """Push the branch and create a GitHub pull request.

    Returns the PR URL.
    """
    # Push the branch to the remote
    subprocess.run(
        ["git", "push", "-u", "origin", pr_info["branch"]],
        cwd=workspace.work_dir,
        check=True,
        capture_output=True,
    )

    # Create the PR using the gh CLI
    result = subprocess.run(
        [
            "gh", "pr", "create",
            "--title", pr_info["pr_title"],
            "--body", pr_info["pr_body"],
            "--base", base_branch,
            "--head", pr_info["branch"],
        ] + (["--draft"] if draft else []),
        cwd=workspace.work_dir,
        check=True,
        capture_output=True,
        text=True,
    )

    pr_url = result.stdout.strip()
    print(f"Pull request created: {pr_url}")
    return pr_url


# Alternative: using PyGithub for more control
def create_pull_request_api(
    repo_full_name: str,
    pr_info: dict,
    github_token: str,
    base_branch: str = "main",
    draft: bool = True,
) -> str:
    """Create a PR using the PyGithub library."""
    from github import Github

    gh = Github(github_token)
    repo = gh.get_repo(repo_full_name)

    pr = repo.create_pull(
        title=pr_info["pr_title"],
        body=pr_info["pr_body"],
        head=pr_info["branch"],
        base=base_branch,
        draft=draft,
    )

    # Add labels for AI-generated PRs
    pr.add_to_labels("ai-generated", "needs-review")

    print(f"PR #{pr.number}: {pr.html_url}")
    return pr.html_url
Listing 16.16: Pull request creation via the gh CLI and the PyGithub library, with draft mode and AI-generated labels.

Draft mode signals that the change has passed automated validation but still needs human approval. The structured PR body (quality score, file list, acceptance criteria checkboxes) gives reviewers everything they need for efficient evaluation. As covered in Chapter 9, the human's role shifts from writing code to reviewing proposals.

Real-World Application: Graphite's Stacked PRs
Real-World Application: Graphite's Stacked PRs

6. Putting It All Together: A Complete Example

The following example runs the entire pipeline on a concrete task: adding request rate limiting to Flask's development server. This is a realistic feature request that touches multiple files and requires understanding Flask's request handling architecture.

# Complete example: end-to-end repository-scale change
def run_flask_rate_limiting_example():
    """Demonstrate the full pipeline on Flask."""

    # Stage 1: Set up workspace
    workspace = RepositoryWorkspace(
        "https://github.com/pallets/flask.git",
        branch="main",
    ).setup()

    # Stage 2: Define the task
    task = TaskSpecification(
        title="Add request rate limiting middleware",
        description=(
            "Add a configurable rate limiting middleware to Flask "
            "that tracks requests per IP address using an in-memory "
            "sliding window counter. The middleware should be "
            "optional, configurable via app.config, and return "
            "HTTP 429 when the limit is exceeded."
        ),
        acceptance_criteria=[
            "New RateLimiter class in flask/middleware.py",
            "Configuration keys: RATELIMIT_ENABLED, "
            "RATELIMIT_DEFAULT (requests/minute)",
            "HTTP 429 response with Retry-After header",
            "Per-IP tracking with sliding window algorithm",
            "All existing tests continue to pass",
            "New tests in tests/test_rate_limiting.py",
        ],
        hints=[
            "Use collections.deque for the sliding window",
            "Look at flask/wrappers.py for request IP access",
            "Follow the middleware pattern in flask/ctx.py",
        ],
    )

    # Stages 3-7: Execute the pipeline
    pipeline = ChangeOrchestrationPipeline(workspace)
    results = pipeline.execute_change(
        task=task,
        branch_name="ai/add-rate-limiting",
        quality_threshold=0.85,
    )

    # Print summary
    print("\n" + "=" * 60)
    print("PIPELINE SUMMARY")
    print(f"  Task: {results['task']}")
    print(f"  Success: {results['success']}")
    for stage, info in results["stages"].items():
        print(f"  {stage}: {info}")

    return results

# Uncomment to run:
# results = run_flask_rate_limiting_example()
Listing 16.17: Complete end-to-end example executing the rate-limiting feature addition against the Flask repository through all seven pipeline stages.

When this pipeline runs, it produces output similar to the following (condensed for presentation):

Cloned https://github.com/pallets/flask.git to /tmp/repo_change_a3f2k
Indexed: 487 dependency edges, 2341 symbol references
============================================================
STAGE 2-3: Task specification and context gathering
Context: 14 files, ~18,400 tokens (13,600 remaining)
============================================================
STAGE 4: Edit plan generation
Generation order (4 files):
  1. src/flask/middleware.py (no dependencies)
  2. src/flask/app.py (depends on: src/flask/middleware.py)
  3. tests/test_rate_limiting.py (depends on: src/flask/middleware.py)
  4. docs/config.rst (no code dependencies)
============================================================
STAGE 5-6: Code generation and review
Generating: src/flask/middleware.py...
Generating: src/flask/app.py...
Generating: tests/test_rate_limiting.py...
Generating: docs/config.rst...
Iteration 1: score=0.78, errors=1, warnings=2
  Re-generating with feedback...
Iteration 2: score=0.91, errors=0, warnings=1
Quality threshold met at iteration 2
============================================================
STAGE 7: Commit and PR preparation
PIPELINE SUMMARY
  Task: Add request rate limiting middleware
  Success: True
Listing 16.18: Sample pipeline output showing dependency-ordered generation across four files, with quality improvement from 0.78 to 0.91 over two iterations.

Notice that the first iteration scored 0.78 with one error (a test file imported a helper function that the middleware did not export) and two warnings (missing docstrings). The feedback loop resolved the error and one warning on the second iteration, reaching 0.91 and passing the quality gate. A single retry with structured feedback closed a 13-point quality gap; without that loop, the patch would have been rejected and the entire pipeline run wasted.

Beyond fixing scoring deficiencies, the generate-and-review loop sometimes reveals a more remarkable dynamic: the generated tests themselves catching genuine bugs in the generated code.

Fun Note: The Test That Tests Itself

In a test run of this pipeline, the generated test_rate_limiting.py included a test called test_rate_limiter_handles_concurrent_requests that spawned 100 threads to hit the rate limiter simultaneously. The test was well-written and caught a genuine race condition in the generated middleware (the sliding window was not thread-safe). The agent fixed the race condition on retry by adding a threading.Lock. In other words, the AI-generated test found a bug in the AI-generated code, and the AI-generated fix resolved it. The entire cycle (generate, test, find bug, fix) completed without human intervention. This is a glimpse of what Chapter 24: Autonomous Software Organizations explores at scale.

Research Frontier: Agentic Coding Systems on SWE-bench

SWE-bench Verified (Jimenez et al., 2024), a curated benchmark of real GitHub issues paired with ground-truth patches for measuring automated code repair, remains the standard measure for repository-scale change quality. By mid-2025, agentic systems crossed the 50% barrier that seemed out of reach a year earlier. OpenHands (formerly OpenDevin; Wang et al., 2024) introduced a containerized agent runtime that executes code, runs tests, and iterates on fixes inside an isolated environment, reaching 53% on SWE-bench Verified. Amazon's SWE-PolyGlot (Rasheed et al., 2025) extended the benchmark paradigm to multilingual repositories (Java, JavaScript, TypeScript, C#), revealing that localization accuracy degrades sharply outside Python due to weaker static analysis tooling. The leading open-weight approach, Agentless (Xia et al., 2024), deliberately avoids agentic loops and instead uses a localize-then-repair two-phase strategy with hierarchical file and function filtering, demonstrating that a carefully structured pipeline (much like the one in this section) can match or outperform free-form agent exploration. The key takeaway for practitioners: structured localization pipelines with bounded iteration have, so far, outperformed unconstrained agent loops on repository-scale benchmarks, though the gap narrows as agent architectures improve.

7. Discovery Workbench Integration

The complete pipeline becomes the ChangeOrchestrator component of the Discovery Workbench. This component is consumed by agents throughout the rest of Part II: the multi-agent teams of Chapter 17 delegate code generation to this pipeline, the testing agents of Chapter 18 use the impact analysis to determine which tests to run, and the debugging agents of Chapter 19 use the dependency graph to trace error propagation.

class DiscoveryWorkbenchChangeModule:
    """Discovery Workbench integration for repository-scale changes.

    Provides a high-level API that other Workbench components
    and agents can call to propose, validate, and apply changes.
    """

    def __init__(self, repo_root: str):
        self.workspace = RepositoryWorkspace(repo_root).setup()
        self.pipeline = ChangeOrchestrationPipeline(self.workspace)
        self._change_history: list[dict] = []

    def propose_change(self, task: TaskSpecification) -> dict:
        """Generate and validate a change proposal."""
        result = self.pipeline.execute_change(task)
        self._change_history.append(result)
        return result

    def get_impact(
        self, file_path: str, symbols: set[str]
    ) -> dict:
        """Query impact analysis for a hypothetical change."""
        blast = self.workspace.dep_graph.blast_radius(file_path)
        impact = self.workspace.dep_graph.impact_set(
            file_path, symbols
        )
        return {
            "blast_radius": len(blast),
            "impact_set": len(impact),
            "affected_files": sorted(impact.keys()),
            "affected_tests": [
                f for f in impact if "test" in f
            ],
        }

    def list_high_risk_modules(self, top_n: int = 10) -> list[dict]:
        """Identify modules with the largest blast radii."""
        radii = []
        for file_path in self.workspace.dep_graph.dependents:
            blast = self.workspace.dep_graph.blast_radius(file_path)
            radii.append({
                "file": file_path,
                "blast_radius": len(blast),
                "direct_dependents": len(
                    self.workspace.dep_graph.dependents.get(
                        file_path, set()
                    )
                ),
            })
        radii.sort(key=lambda x: x["blast_radius"], reverse=True)
        return radii[:top_n]

    def change_history_summary(self) -> str:
        """Summarize all changes made in this session."""
        if not self._change_history:
            return "No changes proposed yet."
        lines = [f"Changes proposed: {len(self._change_history)}"]
        for i, change in enumerate(self._change_history, 1):
            status = "passed" if change["success"] else "FAILED"
            gen = change["stages"].get("generation", {})
            lines.append(
                f"  {i}. {change['task']} [{status}] "
                f"(score: {gen.get('quality_score', 'N/A')})"
            )
        return "\n".join(lines)
Listing 16.19: Discovery Workbench change module exposing propose_change, get_impact, and list_high_risk_modules APIs for downstream agents.
Library Shortcut: Claude Code Interactive Mode

The pipeline we build in this section orchestrates Claude Code in one-shot mode (--print), managing state externally. For simpler use cases, Claude Code's interactive mode handles multi-file changes natively: you describe the task in natural language, and the agent navigates the repository, edits files, and runs tests within a single conversation. A single command like claude "Add rate limiting middleware to Flask with tests" invokes the agent in a conversational loop that performs many of the same steps (context gathering, edit generation, test execution) automatically. What takes 200 lines of orchestration code in our pipeline, Claude Code delivers as a built-in capability. The explicit pipeline is valuable when you need programmatic control, custom review criteria, or integration with continuous integration / continuous delivery (CI/CD) systems; for interactive development, the built-in agent is more productive.

8. Failure Analysis and Lessons Learned

Running this pipeline across 50 different tasks on three open-source repositories (Flask, FastAPI, and scikit-learn) reveals consistent patterns in what succeeds and what fails:

High success rate (above 80%): adding new features that follow existing patterns (new endpoint, new model field, new test), fixing well-localized bugs with clear error messages, and refactoring that changes names but not behavior.

Medium success rate (50-80%): changes that cross architectural boundaries (adding middleware, changing serialization formats), changes that require understanding implicit contracts (Django signals, Flask before_request hooks), and performance optimizations that require algorithmic reasoning.

Low success rate (below 50%): changes that require understanding runtime behavior (race conditions, memory leaks, async timing), changes to code that relies heavily on metaprogramming or dynamic dispatch, and security-sensitive changes where subtle errors have outsized consequences.

Why the Gap Persists

The correlation between task type and success rate matches findings from SWE-bench evaluations: static, structural tasks are solved reliably; dynamic, behavioral tasks remain challenging. LLMs operate primarily as pattern matchers over text, which means they excel at structural transformations, where the pattern is visible in the syntax. They struggle with behavioral reasoning, where the pattern is visible only at runtime. The testing techniques of Chapter 18 and debugging techniques of Chapter 19 address this gap by providing runtime feedback that the static pipeline cannot.

Try It: Build a Mini Change Pipeline for a Local Project

Pick any small Python project you have on your machine (or clone one under 3,000 lines, such as httpie/cli or pallets/click) and build a simplified version of the pipeline from this section.

Step 1. Write a script that uses Python's ast module to parse every .py file in the project and extract all import statements. Build a dictionary mapping each file to the set of project-internal modules it imports. Print the five files with the most importers (highest blast radius).

Step 2. Write a TaskSpecification dataclass (title, description, acceptance_criteria as a list of strings) and instantiate it for a simple change, such as "add a --verbose flag to the CLI entry point."

Step 3. Implement the keyword extraction function from Listing 16.14. Run it on your task description and verify that it pulls out the identifiers you expect (e.g., verbose, any CamelCase class names mentioned).

Step 4. Combine steps 1 and 3: given the keyword matches, collect the matching files, then expand with one level of imports from your dependency dictionary. Print the final context set and its estimated token count (lines multiplied by 10).

Step 5. Feed the collected file contents and your task specification into an LLM (via the API or Claude Code's --print flag) and ask it to produce a JSON edit plan. Compare the files it proposes to edit against your dependency graph: does every referenced symbol trace back to a file in the plan? If not, identify which file is missing and add it to the context.

Exercise 16.3.1

The pipeline caps generation retries at three iterations and context expansion at double the original budget. Suppose you set the quality threshold to 0.90 and the first three iterations produce scores of 0.74, 0.82, and 0.86. The pipeline reports failure. Without lowering the threshold, describe two concrete changes to the context gathering stage (not the generation stage) that could raise the score above 0.90 on a subsequent run. Explain why each change targets the root cause rather than the symptom.

Hint

Look at the two-pass strategy in gather_context. One pass uses keyword search; the other follows dependency edges. Consider what happens when the task description uses natural language that does not match any identifier in the codebase, or when the relevant module is two hops away in the dependency graph instead of one.

Step-Through: Dependency-Ordered Generation

Trace the edit plan from the Flask rate-limiting example (Listing 16.18) with four files. The dependency edges are: app.py references a symbol defined in middleware.py; test_rate_limiting.py imports from middleware.py; config.rst has no code dependencies.

Step 1. Build the dependency set for each file. middleware.py: {} (empty). app.py: {middleware.py}. test_rate_limiting.py: {middleware.py}. config.rst: {} (empty).

Step 2. Topological sort. Files with no dependencies go first: middleware.py and config.rst (order between them is arbitrary). Next come files whose dependencies are already generated: app.py (depends only on middleware.py, already done) and test_rate_limiting.py (same dependency, already done).

Step 3. Generation order: [1] middleware.py, [2] config.rst, [3] app.py, [4] test_rate_limiting.py. When the LLM generates app.py at position 3, the prompt includes the concrete output of middleware.py from position 1, so every symbol reference resolves to a real definition rather than a hallucinated signature.

Real-World Application: Graphite's Stacked PRs

Graphite, a developer tool used at companies like Robinhood and Notion, implements dependency-ordered change submission through "stacked pull requests." Each PR in a stack declares its parent, forming a directed acyclic graph (DAG) analogous to the edit graph in this section. Graphite's merge engine processes the stack in topological order, rebasing each child onto its merged parent, which is exactly the coordinated generation strategy applied to human-authored code instead of LLM-generated edits.

Lab: Measure How Context Size Affects Patch Quality

Goal: Empirically determine the relationship between the token budget given to gather_context and the quality score of the resulting patch.

Tools needed: Python 3.10+, the ast module (standard library), an LLM API (Claude or any model with a 32k+ context window; as of 2025, most frontier models support 128k to 200k tokens, so budget constraints are increasingly about relevance rather than capacity), and a small open-source repository (Flask or Click recommended, both under 20k lines).

What to vary: Run the pipeline five times with token budgets of 4,000, 8,000, 16,000, 32,000, and 64,000 tokens for the same task specification (e.g., "add a --json output flag to the CLI"). Keep the quality threshold, retry cap, and all other parameters fixed.

What to observe: For each budget, record (a) the number of files in the context set, (b) the quality score after the first generation attempt (before retries), (c) the final quality score after retries, and (d) whether the patch includes all necessary files. Plot budget vs. first-attempt quality score. You should observe diminishing returns: quality improves sharply from 4k to 16k tokens, then plateaus or even degrades slightly as irrelevant files dilute the context.

Exercises

  1. (Conceptual) The pipeline creates pull requests as drafts. Describe three criteria that would justify automatically merging an AI-generated PR without human review. Then describe three scenarios where even a high quality score (0.95+) should still require human review. What does this tell you about the limits of automated quality assessment?
  2. (Coding) Fork a small open-source Python project (under 5,000 lines) and run the RepositoryWorkspace.setup() on it. Build the dependency graph and identify the top 3 modules by blast radius. Then define a TaskSpecification for a simple feature addition (e.g., adding a configuration option) and trace manually which files the context gathering stage would select. Compare your prediction with the actual output.
  3. (Analysis) The pipeline uses a fixed quality threshold of 0.85. Analyze how varying this threshold affects two competing metrics: (a) the percentage of generated patches that pass the gate (pass rate) and (b) the percentage of patches that pass the gate and are actually correct when manually reviewed (precision). Sketch the expected precision-recall trade-off curve and identify the threshold that maximizes the F1 score.

What's Next

You now have a complete pipeline for generating, validating, and proposing multi-file changes at repository scale. But the pipeline runs as a single agent working sequentially through files. In Chapter 17: Multi-Agent Software Teams, we break this single-agent bottleneck by introducing architectures where specialized agents (planner, coder, reviewer, tester) collaborate on changes in parallel. The edit graph from Section 16.1 becomes the coordination structure that assigns work to agents, the impact analysis from Section 16.2 becomes the reviewer agent's primary tool, and the quality gate from this section becomes the merge policy that the orchestrator agent enforces.