Lab 08: Planning Agent Implementation

Advanced Due: 2026-04-28

Objectives

• Implement a CodebaseAnalyzer that analyzes real codebases
• Have PlannerAgent generate a concrete spec.md based on analysis results
• Manage task dependency graphs and priority ordering

The Role of PlannerAgent

The planning agent is the brain of the pipeline. A good planner does three things:

1. Assess the codebase — what files exist and where are the functions
2. Decompose the objective — break a large goal into independent, verifiable subtasks
3. Generate the specification — write a spec.md that the coder can work from without ambiguity

Implementation Requirements

1. codebase_analyzer.py — Codebase Analyzer

codebase_analyzer.py

import ast
import subprocess
from pathlib import Path
from dataclasses import dataclass, field

@dataclass
class FileInfo:
    path: str
    size_lines: int
    functions: list[str]
    classes: list[str]
    imports: list[str]
    has_tests: bool

@dataclass
class CodebaseSnapshot:
    root: str
    total_files: int
    total_lines: int
    files: list[FileInfo]
    test_files: list[str]
    entry_points: list[str]

    def to_summary(self, max_files: int = 20) -> str:
        """Generates a concise summary to pass to the LLM."""
        lines = [
            f"Codebase: {self.root}",
            f"Total files: {self.total_files} | Total lines: {self.total_lines:,}",
            f"Test files: {len(self.test_files)}",
            "",
            "Key files:"
        ]
        for fi in sorted(self.files, key=lambda x: x.size_lines, reverse=True)[:max_files]:
            funcs = ", ".join(fi.functions[:5])
            lines.append(f"  {fi.path} ({fi.size_lines} lines) — {funcs}")
        return "\n".join(lines)


class CodebaseAnalyzer:
    IGNORE_DIRS = {".git", "__pycache__", ".venv", "venv", "node_modules", ".mypy_cache"}

    def __init__(self, root: str):
        self.root = Path(root)

    def analyze(self) -> CodebaseSnapshot:
        files: list[FileInfo] = []
        for py_file in self.root.rglob("*.py"):
            if any(d in py_file.parts for d in self.IGNORE_DIRS):
                continue
            info = self._analyze_file(py_file)
            files.append(info)

        test_files = [f.path for f in files if f.has_tests]
        entry_points = self._find_entry_points(files)

        return CodebaseSnapshot(
            root=str(self.root),
            total_files=len(files),
            total_lines=sum(f.size_lines for f in files),
            files=files,
            test_files=test_files,
            entry_points=entry_points
        )

    def _analyze_file(self, path: Path) -> FileInfo:
        source = path.read_text(errors="replace")
        lines = source.splitlines()

        functions, classes, imports = [], [], []
        try:
            tree = ast.parse(source)
            for node in ast.walk(tree):
                if isinstance(node, ast.FunctionDef):
                    functions.append(node.name)
                elif isinstance(node, ast.ClassDef):
                    classes.append(node.name)
                elif isinstance(node, (ast.Import, ast.ImportFrom)):
                    if isinstance(node, ast.Import):
                        imports.extend(alias.name for alias in node.names)
                    else:
                        imports.append(node.module or "")
        except SyntaxError:
            pass

        rel_path = str(path.relative_to(self.root))
        return FileInfo(
            path=rel_path,
            size_lines=len(lines),
            functions=functions,
            classes=classes,
            imports=[i for i in imports if i],
            has_tests=rel_path.startswith("test") or "/test" in rel_path
        )

    def _find_entry_points(self, files: list[FileInfo]) -> list[str]:
        """Identifies files with a main() function as entry points."""
        return [f.path for f in files if "main" in f.functions]

2. spec_generator.py — Specification Generator

spec_generator.py

import anthropic
from datetime import datetime
from pathlib import Path
from codebase_analyzer import CodebaseSnapshot

SPEC_SYSTEM = """
You are a senior software architect creating implementation specifications.
Given a codebase analysis and an objective, produce a detailed spec.md.

The spec must include:
1. Objective (one sentence)
2. Scope (which files will change)
3. Implementation plan (numbered steps, each verifiable)
4. Test strategy (how to verify each step)
5. Risk assessment (what could go wrong)

Be concrete. Reference actual file names and function names from the codebase.
"""

class SpecGenerator:
    def __init__(self):
        self.client = anthropic.Anthropic()

    def generate(self, snapshot: CodebaseSnapshot, objective: str) -> str:
        prompt = f"""
Codebase Analysis:
{snapshot.to_summary()}

Objective: {objective}

Generate a detailed spec.md for this task.
"""
        response = self.client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=SPEC_SYSTEM,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text

    def save(self, content: str, path: str = "spec.md"):
        header = f"<!-- Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')} -->\n\n"
        Path(path).write_text(header + content)
        print(f"[SpecGenerator] spec.md saved: {path}")

3. dependency_graph.py — Task Dependency Management

dependency_graph.py

from collections import defaultdict, deque
from dataclasses import dataclass

@dataclass
class Task:
    id: str
    description: str
    assignee: str
    depends_on: list[str]
    priority: int = 0

class DependencyGraph:
    """Manages task dependencies as a DAG and performs topological sorting."""

    def __init__(self, tasks: list[Task]):
        self.tasks = {t.id: t for t in tasks}
        self._validate()

    def _validate(self):
        """Checks for dependency cycles."""
        visited, in_stack = set(), set()

        def dfs(task_id: str):
            visited.add(task_id)
            in_stack.add(task_id)
            for dep in self.tasks[task_id].depends_on:
                if dep not in self.tasks:
                    raise ValueError(f"Unknown dependency: {dep}")
                if dep in in_stack:
                    raise ValueError(f"Cycle detected: {task_id} -> {dep}")
                if dep not in visited:
                    dfs(dep)
            in_stack.remove(task_id)

        for tid in self.tasks:
            if tid not in visited:
                dfs(tid)

    def topological_sort(self) -> list[Task]:
        """Returns tasks sorted in dependency order."""
        in_degree = defaultdict(int)
        for task in self.tasks.values():
            for dep in task.depends_on:
                in_degree[task.id] += 1

        queue = deque([
            t for t in self.tasks.values() if in_degree[t.id] == 0
        ])
        result = []

        while queue:
            task = queue.popleft()
            result.append(task)
            for other in self.tasks.values():
                if task.id in other.depends_on:
                    in_degree[other.id] -= 1
                    if in_degree[other.id] == 0:
                        queue.append(other)

        if len(result) != len(self.tasks):
            raise ValueError("Cycle exists — topological sort not possible")
        return result

    def get_ready_tasks(self, completed: set[str]) -> list[Task]:
        """Returns tasks that are immediately executable given the completed set."""
        return [
            t for t in self.tasks.values()
            if t.id not in completed
            and all(dep in completed for dep in t.depends_on)
        ]

4. planner_agent.py — Completed Planner

# planner_agent.py (extended from Lab 07)
from codebase_analyzer import CodebaseAnalyzer
from spec_generator import SpecGenerator
from dependency_graph import DependencyGraph, Task
import json
from pathlib import Path

class PlannerAgent:
    def __init__(self, codebase_root: str):
        self.analyzer = CodebaseAnalyzer(codebase_root)
        self.spec_gen = SpecGenerator()

    def plan(self, objective: str, output_dir: str = ".") -> dict:
        print("[Planner] Analyzing codebase...")
        snapshot = self.analyzer.analyze()

        print("[Planner] Generating spec.md...")
        spec_content = self.spec_gen.generate(snapshot, objective)
        self.spec_gen.save(spec_content, f"{output_dir}/spec.md")

        print("[Planner] Decomposing tasks...")
        tasks = self._decompose(spec_content, objective)

        graph = DependencyGraph(tasks)
        ordered = graph.topological_sort()

        plan = {
            "objective": objective,
            "codebase_summary": snapshot.to_summary(max_files=10),
            "spec_path": f"{output_dir}/spec.md",
            "tasks": [
                {"id": t.id, "description": t.description,
                 "assignee": t.assignee, "depends_on": t.depends_on}
                for t in ordered
            ]
        }
        Path(f"{output_dir}/plan.json").write_text(
            json.dumps(plan, indent=2, ensure_ascii=False)
        )
        print(f"[Planner] Done: {len(tasks)} tasks, plan.json saved")
        return plan

    def _decompose(self, spec: str, objective: str) -> list[Task]:
        """Parses spec.md to create a Task list. In practice, uses an LLM call."""
        # Implement yourself in the lab — parse the spec content or re-request the LLM
        return [
            Task("t-01", "Explore codebase and identify relevant files", "researcher", []),
            Task("t-02", "Implement core functionality", "coder", ["t-01"]),
            Task("t-03", "Write and run unit tests", "qa", ["t-02"]),
            Task("t-04", "Code review", "reviewer", ["t-03"]),
        ]

1. Implement codebase_analyzer.py and test it against your own codebase
2. Implement spec_generator.py — confirm actual spec.md is generated
3. Implement dependency_graph.py — include cycle detection tests
4. Run PlannerAgent.plan(): python -c "from planner_agent import PlannerAgent; PlannerAgent('.').plan('Fix calculator.py to pass all tests')"
5. Review the generated spec.md and plan.json

Deliverables

Submit a PR to assignments/lab-08/[student-id]/:

• codebase_analyzer.py — FileInfo, CodebaseSnapshot, CodebaseAnalyzer
• spec_generator.py — LLM-based spec.md generation
• dependency_graph.py — Topological sort and cycle detection
• planner_agent.py — Complete version integrating the three modules
• spec.md — Example of an actually generated specification
• plan.json — Example of an actually generated plan JSON
• README.md — Codebase analysis summary and planner behavior description

spec.md Quality

The quality of spec.md determines the performance of the entire pipeline. A good spec should clearly indicate which function in which file the coder needs to modify and how.