122 lines
4.8 KiB
Python
122 lines
4.8 KiB
Python
|
|
#!/usr/bin/env python3
|
|
"""Impact Analysis Tool - Prevents regressions by identifying affected downstream components."""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
from tools.registry import registry, tool_error, tool_result
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
VERIFY_IMPACT_SCHEMA = {
|
|
"name": "verify_impact",
|
|
"description": "Analyze the impact of your recent changes. Checks for usages of modified functions/classes across the codebase to identify potential regressions. Use this before claiming a task is done to ensure you haven't broken downstream components.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {"type": "string", "description": "Optional: Path to the specific file you want to analyze. If omitted, analyzes all currently staged/modified files in git."},
|
|
"depth": {"type": "integer", "description": "Search depth for usages (default: 1)", "default": 1}
|
|
}
|
|
}
|
|
}
|
|
|
|
def analyze_impact(path: str = None, depth: int = 1, task_id: str = "default"):
|
|
"""Identify downstream usages of modified code elements."""
|
|
try:
|
|
# 1. Identify changed files and symbols
|
|
if path:
|
|
files_to_check = [path]
|
|
else:
|
|
# Use git to find modified files if not specified
|
|
try:
|
|
cmd = ["git", "diff", "--name-only", "HEAD"]
|
|
files_to_check = subprocess.check_output(cmd).decode().splitlines()
|
|
if not files_to_check:
|
|
# Try staged files
|
|
cmd = ["git", "diff", "--cached", "--name-only"]
|
|
files_to_check = subprocess.check_output(cmd).decode().splitlines()
|
|
except:
|
|
return tool_error("Git not available or not a repository. Please specify 'path' explicitly.")
|
|
|
|
if not files_to_check:
|
|
return tool_result(message="No changes detected in git. Try specifying a 'path' if you have uncommitted changes.")
|
|
|
|
# 2. Extract potential symbols (functions/classes) from changes
|
|
# For simplicity, we'll use a heuristic: grep for 'def ' and 'class ' in diffs
|
|
affected_symbols = set()
|
|
for f in files_to_check:
|
|
try:
|
|
diff_cmd = ["git", "diff", "HEAD", "--", f]
|
|
diff = subprocess.check_output(diff_cmd).decode()
|
|
for line in diff.splitlines():
|
|
if line.startswith("+") and ("def " in line or "class " in line):
|
|
# Extract name
|
|
parts = line.split()
|
|
if len(parts) > 1:
|
|
name = parts[1].split("(")[0].split(":")[0]
|
|
affected_symbols.add(name)
|
|
except:
|
|
continue
|
|
|
|
# 3. Search for these symbols in the codebase (excluding the original files)
|
|
impact_report = {}
|
|
for symbol in affected_symbols:
|
|
if not symbol or len(symbol) < 3: continue
|
|
|
|
# Use ripgrep/grep to find usages
|
|
try:
|
|
exclude_args = []
|
|
for f in files_to_check:
|
|
exclude_args.extend(["--exclude", f])
|
|
|
|
# Search for usages
|
|
search_cmd = ["grep", "-r", "-l", symbol, "."] + exclude_args
|
|
usages = subprocess.check_output(search_cmd).decode().splitlines()
|
|
if usages:
|
|
impact_report[symbol] = usages[:10] # Limit per symbol
|
|
except subprocess.CalledProcessError:
|
|
# No matches found
|
|
continue
|
|
|
|
if not impact_report:
|
|
return tool_result(
|
|
status="Clean",
|
|
message="No obvious downstream usages found for modified symbols. Changes appear contained.",
|
|
files_analyzed=files_to_check
|
|
)
|
|
|
|
summary = (
|
|
f"IDENTIFIED POTENTIAL IMPACTS:\n"
|
|
f"You modified {len(affected_symbols)} key symbols. The following files use them and might be affected:\n"
|
|
)
|
|
for sym, files in impact_report.items():
|
|
summary += f"- {sym}: used in {', '.join(files)}\n"
|
|
|
|
return tool_result(
|
|
status="Attention Required",
|
|
summary=summary,
|
|
impact_map=impact_report,
|
|
recommendation="Review the identified files to ensure your changes didn't break their functionality."
|
|
)
|
|
|
|
except Exception as e:
|
|
return tool_error(f"Impact analysis failed: {str(e)}")
|
|
|
|
def _handle_verify_impact(args, **kw):
|
|
return analyze_impact(
|
|
path=args.get("path"),
|
|
depth=args.get("depth", 1),
|
|
task_id=kw.get("task_id", "default")
|
|
)
|
|
|
|
registry.register(
|
|
name="verify_impact",
|
|
toolset="qa",
|
|
schema=VERIFY_IMPACT_SCHEMA,
|
|
handler=_handle_verify_impact,
|
|
emoji="🛡️"
|
|
)
|
|
|