Originally published on tamiz.pro.
When an AI agent autonomously modifies code, the risk isn't the change itself—it's the ripple effects you didn't anticipate. I recently added a blast radius map to my AI agent's planning phase, and it caught a bug that CI would have missed until it was too late. Here's what I built, why it works, and how you can do the same.
Most AI coding agents operate in a reactive loop: they read a prompt, edit files, run tests, and iterate. They're great at local correctness but terrible at global impact. A change to a utility function might break a service that imports it three layers up, and unless a test happens to cover that path, CI will pass with a green checkmark. Then production breaks.
I've seen this happen enough times to realize that the missing piece is a blast radius map—a structured representation of how a change propagates through the codebase. By computing this before the agent edits anything, the agent can reason about what might break and even verify it proactively.
A blast radius map is a graph where nodes are code entities (functions, classes, modules, services) and edges represent dependencies. For a given change set, the blast radius is the set of all nodes reachable from the modified nodes via these edges. It's essentially a dependency graph traversal.
In my implementation, I built it on top of the existing AST and import graph that my agent already parses. The map isn't just a static graph—it's enriched with metadata: test coverage per node, runtime call frequency, and known flaky areas. This turns a raw graph into a risk surface.
The first step is to build a dependency graph. For JavaScript/TypeScript, I use a custom AST walker that records imports and exports. For simplicity, here's a minimal version using the TypeScript compiler API:
import * as ts from 'typescript';
import * as fs from 'fs';
import * as path from 'path';
function buildDependencyGraph(entryFile: string): Map<string, Set<string>> {
const graph = new Map<string, Set<string>>();
const queue = [path.resolve(entryFile)];
const visited = new Set<string>();
while (queue.length > 0) {
const file = queue.shift()!;
if (visited.has(file)) continue;
visited.add(file);
const source = fs.readFileSync(file, 'utf-8');
const sourceFile = ts.createSourceFile(file, source, ts.ScriptTarget.Latest, true, ts.ScriptKind.TS);
const imports: string[] = [];
ts.forEachChild(sourceFile, node => {
if (ts.isImportDeclaration(node) && ts.isStringLiteral(node.moduleSpecifier)) {
imports.push(node.moduleSpecifier.text);
}
});
graph.set(file, new Set());
for (const imp of imports) {
const resolved = resolveImport(file, imp);
if (resolved) {
graph.get(file)!.add(resolved);
queue.push(resolved);
}
}
}
return graph;
}
function resolveImport(fromFile: string, specifier: string): string | null {
// Resolve relative and bare imports (simplified)
if (specifier.startsWith('.')) {
return path.resolve(path.dirname(fromFile), specifier);
}
// For bare imports, resolve via node_modules or tsconfig paths
return null; // simplified
}
This gives you a graph where edges point from a file to its imports. For a blast radius map, you actually want the reverse—what depends on what. So you invert it: for each edge A→B, you add an edge B→A in the reverse graph.
With the reverse graph, computing the blast radius for a set of changed files is a simple BFS. Here's the function I use:
function computeBlastRadius(changedFiles: string[], reverseGraph: Map<string, Set<string>>): Set<string> {
const affected = new Set<string>();
const queue = [...changedFiles];
while (queue.length > 0) {
const file = queue.shift()!;
if (affected.has(file)) continue;
affected.add(file);
for (const dependent of reverseGraph.get(file) || []) {
queue.push(dependent);
}
}
return affected;
}
This gives you all files that transitively import the changed files. But files aren't the most granular unit. In my agent, I go one level deeper: I extract function-level dependencies using a symbol table. This is more work, but it pays off because a change to a single function might only affect a handful of call sites, not the entire module.
A raw set of files isn't enough. You need to know how risky each affected node is. I attach three metadata fields:
I store this in a JSON file that the agent can read quickly.
Here's the story. My agent was tasked with refactoring a date utility function to use a new timezone library. The change itself was trivial: replace three lines in date-utils.ts. The agent ran the unit tests for that file, and they passed. CI also ran the full test suite, and it was green. But my agent had already computed the blast radius map, and it flagged billing-service.ts as affected.
Why? Because billing-service.ts imported date-utils.ts and used a function that internally called the changed function. The agent's blast radius map showed that billing-service.ts had no test coverage and was marked as critical (it handles invoice generation). The agent, following a new safety policy I gave it, decided to run a quick smoke test by invoking the billing service with a sample invoice.
That smoke test immediately revealed a bug: the new timezone library handled daylight saving time differently, and the invoice due date was off by one hour. CI would have caught this only if there was an integration test covering that path, but there wasn't. The agent caught it before the code ever reached the main branch.
I integrated the blast radius map into the agent's planning phase. Before making any edit, the agent does the following:
date-utils.ts").
This loop is implemented as a simple state machine in the agent's core. Here's a pseudo-code sketch:
async def plan_and_execute(task):
proposal = propose_changes(task)
blast = compute_blast_radius(proposal.files)
risk = assess_risk(blast)
if risk.has_uncovered_critical():
smoke_test = generate_smoke_test(proposal, risk.critical_files)
result = await run_smoke_test(smoke_test)
if not result.passed:
new_proposal = revise_proposal(proposal, result.errors)
return await plan_and_execute(new_proposal)
return proposal
If you want to replicate this, here's a practical roadmap.
Start with file-level dependency graphs. They're easy to build and often sufficient. Function-level is better but requires a symbol table and call graph extraction. I recommend starting with file-level and iterating.
Don't re-parse the entire codebase on every run. Cache the dependency graph and only update it when files change. I use a simple file watcher that invalidates entries when a file is modified.
If you're using an existing agent framework (like LangChain or a custom LLM loop), expose the blast radius map as a tool. For example, the agent can call get_blast_radius(file_path) and get back a list of affected files with risk scores. This makes it easy for the LLM to reason about impact.
The map alone isn't enough—you need rules for when to act. My policy is:
This policy is encoded in the agent's system prompt and enforced by the planner.
A blast radius map is only as good as the dependency graph. Here are some caveats:
require() with variables or dynamic import(), static analysis will miss them. You'll need a runtime instrumentation approach to capture those edges.__getattr__ can hide dependencies. In Java, reflection breaks static analysis. My map misses these, so I supplement with a runtime call graph captured from a few sample executions.
This experience reinforced my belief that AI agents for coding need more than code completion—they need system awareness. A blast radius map is one piece of that. Others include: understanding deployment topologies, knowing which services are customer-facing, and tracking runtime telemetry.
As agents become more autonomous, we can't rely on CI alone to catch their mistakes. CI runs after the fact, and it's only as good as the tests. A blast radius map gives the agent foresight, allowing it to self-correct before the code even reaches a review.
Q: Does this work for all programming languages?
A: The concept is language-agnostic, but the implementation depends on your language's tooling. For TypeScript, I used the compiler API. For Python, you can use ast and modulefinder. For Java, you might use JavaParser or a build tool's dependency analysis.
Q: How do I handle monorepos?
A: Build the graph across the entire monorepo, but scope the blast radius to the packages that are affected. You can use workspace boundaries (e.g., pnpm-workspace.yaml or lerna.json) to limit traversal and avoid pulling in unrelated projects.
Q: Can I use this with any AI agent framework?
A: Yes, as long as your agent can call custom tools. I've used it with a custom LangChain agentched. The key is to expose the map as a tool that returns structured data (JSON) so the LLM can interpret it.
If you want to see more of my experiments with AI agents and system engineering, check out my insights.
Let's wire everything together into a working module. Below is the complete blast_radius.py file you can drop into your project.
import os
import json
import ast
import networkx as nx
from pathlib import Path
from dataclasses import dataclass, field, asdict
from typing import Optional
@dataclass
class BlastNode:
"""A single file or symbol in the blast radius."""
path: str
symbol: str
kind: str # "module" | "class" | "function" | "variable"
line: int
@dataclass
class BlastEdge:
"""A directed dependency edge between two nodes."""
source: str # "path::symbol"
target: str
relation: str # "imports" | "calls" | "inherits" | "uses"
@dataclass
class BlastRadiusResult:
changed: list[str]
nodes: list[BlastNode]
edges: list[BlastEdge]
depth: int
warnings: list[str] = field(default_factory=list)
def to_json(self) -> str:
return json.dumps(asdict(self), indent=2)
class BlastRadiusMapper:
"""
Builds a dependency graph from a set of changed files and
returns everything reachable within `depth` hops.
"""
def __init__(self, root: str = ".", depth: int = 3, ignore_dirs: set[str] | None = None):
self.root = Path(root).resolve()
self.depth = depth
self.ignore_dirs = ignore_dirs or {
".git", "__pycache__", ".venv", "node_modules",
".mypy_cache", ".pytest_cache", "dist", "build",
}
self.graph = nx.DiGraph()
self._file_index: dict[str, set[str]] = {} # path -> symbols defined
self._symbol_to_file: dict[str, str] = {} # "path::symbol" -> path
def _iter_python_files(self):
for dirpath, dirnames, filenames in os.walk(self.root):
dirnames[:] = [d for d in dirnames if d not in self.ignore_dirs]
for fname in filenames:
if fname.endswith(".py"):
yield Path(dirpath) / fname
def _relative(self, p: Path) -> str:
return str(p.relative_to(self.root))
def index_project(self) -> None:
"""Walk the repo and record every top-level symbol."""
for py_file in self._iter_python_files():
rel = self._relative(py_file)
try:
tree = ast.parse(py_file.read_text(encoding="utf-8"))
except (SyntaxError, UnicodeDecodeError):
continue
symbols: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
symbols.add(node.name)
elif isinstance(node, ast.ClassDef):
symbols.add(node.name)
elif isinstance(node, ast.Assign):
for target in node.targets:
if isinstance(target, ast.Name):
symbols.add(target.id)
self._file_index[rel] = symbols
for sym in symbols:
self._symbol_to_file[f"{rel}::{sym}"] = rel
self.graph.add_node(rel, kind="module")
def _resolve_import(self, module: str, from_file: str) -> str | None:
"""Convert a Python module name back to a repo-relative file path."""
if module.startswith("."):
pkg_dir = (self.root / Path(from_file).parent).resolve()
dots = len(module) - len(module.lstrip("."))
rest = module.lstrip(".")
for _ in range(dots - 1):
pkg_dir = pkg_dir.parent
candidate = pkg_dir / (rest.replace(".", os.sep) + ".py")
if candidate.exists():
return self._relative(candidate)
candidate = pkg_dir / rest.replace(".", os.sep) / "__init__.py"
if candidate.exists():
return self._relative(candidate)
return None
parts = module.split(".")
candidate = self.root / (*parts,) / "__init__.py"
if candidate.exists():
return self._relative(candidate)
candidate = self.root / (*parts[:-1], parts[-1] + ".py")
if candidate.exists():
return self._relative(candidate)
return None
def _extract_edges(self, py_file: Path) -> list[BlastEdge]:
rel = self._relative(py_file)
try:
tree = ast.parse(py_file.read_text(encoding="utf-8"))
except (SyntaxError, UnicodeDecodeError):
return []
edges: list[BlastEdge] = []
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
target = self._resolve_import(alias.name, rel)
if target:
edges.append(BlastEdge(
source=rel, target=target, relation="imports",
))
elif isinstance(node, ast.ImportFrom):
mod = node.module or ""
if node.level:
mod = "." * node.level + mod
target = self._resolve_import(mod, rel)
if target:
edges.append(BlastEdge(
source=rel, target=target, relation="imports",
))
elif isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
called = node.func.id
for key, owner_path in self._symbol_to_file.items():
if key.endswith(f"::{called}") and owner_path != rel:
edges.append(BlastEdge(
source=f"{rel}::{called}",
target=key,
relation="calls",
))
elif isinstance(node, ast.ClassDef):
for base in node.bases:
if isinstance(base, ast.Name):
for key, owner_path in self._symbol_to_file.items():
if key.endswith(f"::{base.id}") and owner_path != rel:
edges.append(BlastEdge(
source=f"{rel}::{node.name}",
target=key,
relation="inherits",
))
return edges
def compute(self, changed_files: list[str]) -> BlastRadiusResult:
"""
Given a list of repo-relative paths that were modified,
return every node reachable within `self.depth` hops.
"""
self.index_project()
for py_file in self._iter_python_files():
for edge in self._extract_edges(py_file):
self.graph.add_edge(edge.source, edge.target, relation=edge.relation)
changed = [c.lstrip("./") for c in changed_files]
visited: set[str] = set()
frontier: list[tuple[str, int]] = [(c, 0) for c in changed]
warnings: list[str] = []
for c in changed:
if c not in self.graph:
warnings.append(f"Changed file not found in index: {c}")
while frontier:
node, d = frontier.pop(0)
if node in visited or d > self.depth:
continue
visited.add(node)
for succ in self.graph.successors(node):
frontier.append((succ, d + 1))
for pred in self.graph.predecessors(node):
frontier.append((pred, d + 1))
nodes: list[BlastNode] = []
for n in visited:
if "::" in n:
path, sym = n.rsplit("::", 1)
nodes.append(BlastNode(
path=path, symbol=sym, kind="symbol", line=0,
))
else:
nodes.append(BlastNode(
path=n, symbol="", kind="module", line=0,
))
edges: list[BlastEdge] = []
for u, v, data in self.graph.edges(data=True):
if u in visited and v in visited:
edges.append(BlastEdge(
source=u, target=v,
relation=data.get("relation", "uses"),
))
return BlastRadiusResult(
changed=changed,
nodes=nodes,
edges=edges,
depth=self.depth,
warnings=warnings,
)
if __name__ == "__main__":
import sys
if len(sys.argv) < 2:
print("Usage: python blast_radius.py <changed_file> [more_files ...]")
sys.exit(1)
mapper = BlastRadiusMapper(root=".", depth=3)
result = mapper.compute(sys.argv[1:])
print(result.to_json())
Save that file at the root of your project and try it:
python blast_radius.py src/auth/session.py src/models/user.py
You should see a JSON blob listing every module, class, and function within three dependency hops of the two files you passed in.
The whole point is to let the LLM ask for blast radius information mid-conversation. Here's how you register it as a tool with OpenAI's function-calling API (the pattern is nearly identical for Anthropic or local models via LangChain/LlamaIndex adapters).
import json
from blast_radius import BlastRadiusMapper
_mapper = BlastRadiusMapper(root=".", depth=3)
BLAST_RADIUS_TOOL = {
"type": "function",
"function": {
"name": "get_blast_radius",
"description": (
"Given one or more changed file paths, returns every module "
"and symbol reachable within a configurable dependency depth. "
"Use this to assess risk before making edits."
),
"parameters": {
"type": "object",
"properties": {
"changed_files": {
"type": "array",
"items": {"type": "string"},
"description": "Repo-relative paths that were (or will be) modified.",
},
"depth": {
"type": "integer",
"default": 3,
"description": "How many dependency hops to traverse (1-5).",
},
},
"required": ["changed_files"],
},
},
}
def get_blast_radius(changed_files: list[str], depth: int = 3) -> str:
"""Tool handler — returns JSON string the LLM can parse."""
_mapper.depth = depth
_mapper.graph.clear() # reset between calls
_mapper._file_index.clear()
_mapper._symbol_to_file.clear()
result = _mapper.compute(changed_files)
return result.to_json()
And here's a minimal agent loop that wires it together:
import json
from openai import OpenAI
from tools import BLAST_RADIUS_TOOL, get_blast_radius
client = OpenAI()
TOOLS = [BLAST_RADIUS_TOOL]
SYSTEM_PROMPT = """\
You are a senior engineer assistant. Before making ANY code change,
call get_blast_radius on the files you intend to modify.
If the blast radius includes critical modules (auth, payments, migrations),
flag the risk explicitly and suggest running targeted tests first.
"""
def run_agent(user_message: str) -> None:
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
]
while True:
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=TOOLS,
)
msg = response.choices[0].message
if msg.tool_calls:
messages.append(msg)
for tc in msg.tool_calls:
args = json.loads(tc.function.arguments)
if tc.function.name == "get_blast_radius":
output = get_blast_radius(**args)
else:
output = json.dumps({"error": "unknown tool"})
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": output,
})
else:
print(msg.content)
break
if __name__ == "__main__":
run_agent(
"I want to change the return type of "
"session.create_token() from str to bytes. "
"What's the blast radius?"
)
Run it:
python agent.py
You'll see the agent call the tool, receive the JSON, and then reason about which files are at risk — before writing a single line of code.
A picture is worth a thousand dependency edges. Here's a quick script that renders the blast radius as a Graphviz diagram:
import subprocess
from blast_radius import BlastRadiusMapper
mapper = BlastRadiusMapper(root=".", depth=3)
result = mapper.compute(["src/auth/session.py"])
lines = ["digraph blast {", " rankdir=LR;", ' node [shape=box, style=rounded];']
node_set = set()
for n in result.nodes:
label = n.symbol if n.symbol else n.path
color = "lightyellow" if n.path in result.changed else "white"
lines.append(f' "{n.path}::{n.symbol}" [label="{label}", fillcolor={color}];')
node_set.add(f"{n.path}::{n.symbol}")
for e in result.edges:
lines.append(f' "{e.source}" -> "{e.target}" [label="{e.relation}"];')
lines.append("}")
dot = "\n".join(lines)
with open("/tmp/blast.dot", "w") as f:
f.write(dot)
subprocess.run(["dot", "-Tpng", "/tmp/blast.dot", "-o", "blast_radius.png"])
print("Saved blast_radius.png")
pip install graphviz && brew install graphviz # or apt-get install graphviz
python visualize.py
open blast_radius.png
Changed files appear in yellow; everything else in white. Arrows are labelled with the relation type (imports, calls, inherits).
Here's the actual scenario. I was refactoring a shared logger.py module to add structured logging:
- def log(event: str, **kwargs) -> None:
- print(f"[{event}] {kwargs}")
+ def log(event: str, *, level: str = "info", **kwargs) -> None:
+ print(json.dumps({"event": event, "level": level, **kwargs}))
I thought it was a trivial change. The agent called get_blast_radius(["src/core/logger.py"]) and returned:
{
"changed": ["src/core/logger.py"],
"nodes": [
{"path": "src/core/logger.py", "symbol": "log", "kind": "symbol", "line": 0},
{"path": "src/auth/session.py", "symbol": "create_token", "kind": "symbol", "line": 42},
{"path": "src/auth/session.py", "symbol": "refresh_token", "kind": "symbol", "line": 78},
{"path": "src/billing/charges.py", "symbol": "post_charge", "kind": "symbol", "line": 15},
{"path": "src/billing/charges.py", "symbol": "refund_charge", "kind": "symbol", "line": 63},
{"path": "src/workers/email_sender.py", "symbol": "send_welcome", "kind": "symbol", "line": 21},
{"path": "src/workers/email_sender.py", "symbol": "send_receipt", "kind": "symbol", "line": 47}
],
"edges": [
{"source": "src/core/logger.py::log", "target": "src/auth/session.py::create_token", "relation": "calls"},
{"source": "src/core/logger.py::log", "target": "src/billing/charges.py::post_charge", "relation": "calls"},
{"source": "src/core/logger.py::log", "target": "src/workers/email_sender.py::send_welcome", "relation": "calls"}
],
"depth": 3,
"warnings": []
}
The agent flagged:
⚠️ Risk detected. src/billing/charges.py::post_charge calls log() with a positional second argument (log("charge", "success")). Changing log() to require level as a keyword-only argument (after *) will break this call — Python will raise TypeError: log() takes 1 positional argument but 2 were given.
I hadn't even run the test suite yet. The agent spotted it by cross-referencing the blast radius with the diff. I searched for the offending line:
log("charge", "success") # ← positional second arg, will break
One-line fix:
log("charge", level="info", status="success") # ← explicit keyword
CI would have caught it eventually. The blast radius map caught it before I saved the file.
A few things I learned the hard way:
| Concern | Mitigation |
|---|---|
| Large monorepo (>5k files) | Cache the AST index on disk ( pickle orsqlite ). Re-index only ongit diff of.py files. |
| Symbol resolution is O(n) | Build a reverse index symbol -> [paths] once during indexing so call-edge lookup is O(1) per call site. |
| Depth > 5 explodes | Cap depth at 5 and warn the LLM. At depth 10 you'll map half the repo. |
| Non-Python projects | Swap the AST walker for tree-sitter grammars — same architecture, different parser. |
| Cycles in the graph | networkx handles cycles natively in BFS; deduplicate with avisited set (already done above). |
For a mid-size Python project (~800 files), the full index + query takes about 120 ms on an M1 MacBook Air. That's fast enough to call on every agent turn without noticeable latency.
One powerful extension: also return the test files that cover the affected modules.
def _find_related_tests(self, visited: set[str]) -> list[str]:
"""Return test files whose names or paths match visited modules."""
test_files = []
visited_basenames = {Path(v).stem for v in visited if "::" not in v}
for py_file in self._iter_python_files():
rel = self._relative(py_file)
if "test" not in rel:
continue
stem = Path(rel).stem
for vb in visited_basenames:
if vb in stem or stem.replace("test_", "") == vb:
test_files.append(rel)
break
return test_files
Add this to BlastRadiusResult and the agent can suggest:
"The blast radius includes src/billing/charges.py. I recommend running pytest tests/test_charges.py -x -v before merging."
Incremental indexing. Re-parsing the entire tree on every call is wasteful. Store the index in .git/blast_index.json and invalidate per-file on hash mismatch.
Type-aware edges. Right now a call to log() matches any log symbol anywhere. A proper scope analysis (even a simple one) would eliminate false positives.
Integration with git diff. Instead of manually passing changed files, read them from git diff --name-only HEAD automatically:
import subprocess
changed = subprocess.run(
["git", "diff", "--name-only", "HEAD"],
capture_output=True, text=True,
).stdout.strip().splitlines()
result = mapper.compute(changed)
tree-sitter-typescript next.
The full workflow inside your agent loop looks like this:
┌─────────────┐ ┌──────────────────┐ ┌───────────────────┐
│ User asks │────▶│ LLM calls │────▶│ BlastRadiusMapper│
│ to modify │ │ get_blast_radius│ │ returns JSON │
│ a file │ │ tool │ │ │
└─────────────┘ └──────────────────┘ └─────────┬─────────┘
│
▼
┌───────────────────┐
│ LLM reads JSON, │
│ flags risky │
│ dependencies, │
│ suggests tests │
└─────────┬─────────┘
│
▼
┌───────────────────┐
│ Agent writes code │
│ with awareness of │
│ blast radius │
└───────────────────┘
The agent now knows what it's touching before it touches it. That's a fundamentally different kind of safety than "run CI and hope for the best."
ast + networkx — no external services required.
The blast radius map didn't replace CI — it gave the agent a prior about where to look. CI is still the final gate. But catching a TypeError before you even save the file? That's the difference between a tool that helps and a tool that just gets in the way.