za-ek/agents
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Coder preparations

Browse Source
This commit is contained in:
za-ek
2026-09-22 23:21:59 +03:00
parent 3d1df7fee5
commit 1f2dc25e0b
32 changed files with 3205 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agents/codebase_analyst/entrypoint.py
+409
View File
@@ -0,0 +1,409 @@
"""
Idempotent repository entrypoint analyzer (LLMClient edition).
Flow:
1. List repo files (deterministic walk).
2. Ask LLM for primary language.
3. Loop:
a. Ask LLM to pick ONE candidate file.
b. Read an excerpt.
c. Ask LLM: is it the entrypoint / does it forward to another file / wrong?
d. On "entrypoint" -> stop. On "forward" -> probe next_file. On "wrong" -> pick again.
All LLM calls go through the project's `LLMClient.chat_with_schema` with a
strict JSON schema, validated against a Pydantic model. Validation failures are
retried at most 2 times (per project convention). Successful responses are
cached keyed on (schema_name, system, user) so re-running against the same
repository is idempotent and makes zero additional LLM calls.
"""
from __future__ import annotations
import hashlib
import json
import logging
import os
from pathlib import Path
from typing import List, Literal, Optional, Type, TypeVar
from pydantic import BaseModel, Field, ValidationError, TypeAdapter
from agents.coders.repo import RepoContext
from common.llm_client import LLMClient
from contracts.RepoContext import Language, LanguageLiteral
from contracts.coders.VerifyOutput import sanitize_for_openai
log = logging.getLogger("entrypoint_analyzer")
MAX_LLM_RETRIES = 2 # retries AFTER the initial attempt (so <= 3 total)
EXCERPT_CHARS = 4_000
MAX_ITERATIONS = 40
MAX_LISTING_ENTRIES = 2_000
IGNORED_DIRS = {
".git", ".hg", ".svn", "node_modules", "__pycache__", ".venv", "venv",
"env", "dist", "build", ".idea", ".vscode", ".mypy_cache", ".pytest_cache",
".tox", "target", "out", "obj", "bin", "coverage",
}
IGNORED_EXTS = {
".png", ".jpg", ".jpeg", ".gif", ".pdf", ".zip", ".tar", ".gz", ".lock",
".pyc", ".so", ".dll", ".dylib", ".exe", ".woff", ".woff2", ".ico",
".svg", ".mp4", ".mp3", ".class", ".jar",
}
# --------------------------------------------------------------------------- #
# Pydantic schemas #
# --------------------------------------------------------------------------- #
class LanguageGuess(BaseModel):
"""Primary language of the repository."""
language: LanguageLiteral = Field(description="Primary programming language, lowercase (e.g. 'python').")
confidence: float = Field(ge=0.0, le=1.0)
reasoning: str
class FilePick(BaseModel):
"""A single candidate file to inspect, chosen from the listing."""
file_path: str = Field(description="Relative path exactly as it appears in the listing.")
reason: str
class ProbeVerdict(BaseModel):
"""Verdict for the currently inspected file."""
verdict: Literal["entrypoint", "forward", "wrong"]
next_file: Optional[str] = Field(
default=None,
description="If verdict='forward': the next file to inspect (must be in listing).",
)
reason: str
class Attempt(BaseModel):
file_path: str
excerpt: str
verdict: Optional[ProbeVerdict] = None
class AnalysisResult(BaseModel):
language: Language
entrypoint: str
attempts: List[Attempt]
reasoning: str
# --------------------------------------------------------------------------- #
# LLM plumbing: LLMClient wrapper with strict schema + cache + retries #
# --------------------------------------------------------------------------- #
T = TypeVar("T", bound=BaseModel)
class ResponseCache:
"""File-backed cache: prompt-hash -> validated-model JSON. Makes runs idempotent."""
def __init__(self, path: Optional[Path] = None) -> None:
self.path = path
self._mem: dict[str, str] = {}
if path and path.exists():
try:
self._mem = json.loads(path.read_text("utf-8"))
except Exception: # noqa: BLE001
log.warning("Corrupt cache at %s; starting fresh", path)
@staticmethod
def key(system: str, user: str, schema: Type[BaseModel]) -> str:
h = hashlib.sha256()
h.update(schema.__name__.encode())
h.update(b"\x00")
h.update(system.encode())
h.update(b"\x00")
h.update(user.encode())
return h.hexdigest()
def get(self, k: str) -> Optional[str]:
return self._mem.get(k)
def put(self, k: str, v: str) -> None:
self._mem[k] = v
if self.path:
self.path.parent.mkdir(parents=True, exist_ok=True)
tmp = self.path.with_suffix(self.path.suffix + ".tmp")
tmp.write_text(json.dumps(self._mem, indent=2, sort_keys=True), "utf-8")
tmp.replace(self.path)
class StructuredLLM:
"""
Wrap the project's `LLMClient` with:
* strict JSON-schema enforcement (same shape as the reference usage),
* Pydantic validation of the response,
* at most MAX_LLM_RETRIES retries on validation failure,
* a deterministic prompt-keyed cache so reruns are idempotent.
Usage mirrors the reference snippet but is generic over the output model.
"""
def __init__(
self,
client: LLMClient,
cache: Optional[ResponseCache] = None,
max_retries: int = MAX_LLM_RETRIES,
) -> None:
self.client = client
self.cache = cache
self.max_retries = max_retries
def call(self, system: str, user: str, schema: Type[T]) -> T:
schema_adapter = TypeAdapter(schema)
# --- idempotency: serve from cache when prompt+system+schema match ---
cache_key = ResponseCache.key(system, user, schema) if self.cache else None
if cache_key and (hit := self.cache.get(cache_key)) is not None:
try:
return schema.model_validate_json(hit)
except ValidationError:
log.warning("Stale cache entry for %s; ignoring", schema.__name__)
# --- build the strict response_format exactly as in the reference ---
response_format = {
"type": "json_schema",
"json_schema": {
"name": schema.__name__,
"schema": sanitize_for_openai(schema_adapter.json_schema()),
"strict": True,
},
}
last_err: Optional[BaseException] = None
for attempt in range(self.max_retries + 1):
augmented_system = system
if attempt > 0 and last_err is not None:
augmented_system = (
f"{system}\n\n"
f"Your previous response failed schema validation with: {last_err}. "
f"Return ONLY a single JSON object conforming exactly to the "
f"provided schema — no prose, no code fences."
)
try:
raw = self.client.chat_with_schema(
user,
response_format,
system=augmented_system,
)
if isinstance(raw, str):
raw = json.loads(raw)
parsed = schema_adapter.validate_python(raw) # pydantic v2
except (ValidationError, json.JSONDecodeError, ValueError) as exc:
last_err = exc
log.warning(
"Schema %s failed validation (attempt %d/%d): %s",
schema.__name__, attempt + 1, self.max_retries + 1, exc,
)
continue
except Exception as exc: # noqa: BLE001 (transport / API errors)
last_err = exc
log.warning(
"LLM transport error for %s (attempt %d/%d): %s",
schema.__name__, attempt + 1, self.max_retries + 1, exc,
)
continue
# --- memoize under the *original* prompt key ---------------------
if cache_key and self.cache:
self.cache.put(cache_key, parsed.model_dump_json())
return parsed
# Exhausted retries — hard failure, per project convention.
raise RuntimeError(
f"LLM failed to produce a valid {schema.__name__} "
f"after {self.max_retries + 1} attempt(s)"
) from last_err
# --------------------------------------------------------------------------- #
# Repository listing #
# --------------------------------------------------------------------------- #
def list_repo(root: Path) -> List[str]:
"""Deterministic, sorted listing of analyzable files relative to `root`."""
root = root.resolve()
out: List[str] = []
for dirpath, dirnames, filenames in os.walk(root):
dirnames[:] = sorted(d for d in dirnames if d not in IGNORED_DIRS)
for fname in sorted(filenames):
if Path(fname).suffix.lower() in IGNORED_EXTS:
continue
rel = (Path(dirpath) / fname).relative_to(root)
out.append(rel.as_posix())
return out
def _listing_repr(listing: List[str], cap: int = MAX_LISTING_ENTRIES) -> str:
if len(listing) <= cap:
return "\n".join(listing)
return "\n".join(listing[:cap]) + f"\n...<{len(listing) - cap} more entries truncated>..."
# --------------------------------------------------------------------------- #
# Analyzer #
# --------------------------------------------------------------------------- #
class EntrypointAnalyzer:
def __init__(
self,
llm: StructuredLLM,
max_iterations: int = MAX_ITERATIONS,
excerpt_chars: int = EXCERPT_CHARS,
) -> None:
self.llm = llm
self.max_iterations = max_iterations
self.excerpt_chars = excerpt_chars
# -- public ------------------------------------------------------------ #
def analyze(self, repo_root: Path) -> AnalysisResult:
repo_root = Path(repo_root).resolve()
if not repo_root.is_dir():
raise ValueError(f"Not a directory: {repo_root}")
listing = list_repo(repo_root)
if not listing:
raise ValueError(f"No analyzable files under {repo_root}")
guess = self._guess_language(listing)
language: Language = guess.language
log.info("Detected language=%s (confidence=%.2f)", language.value, guess.confidence)
attempts: List[Attempt] = []
carried: Optional[str] = None # "forward" hint from a previous verdict
for iteration in range(1, self.max_iterations + 1):
# Resolve to a definitely-non-None local for this iteration.
current_file: str = (
carried if carried is not None
else self._pick_file(listing, language, attempts)
)
if current_file not in listing:
log.warning(
"LLM suggested %r which is not in the listing; discarding.",
current_file,
)
carried = None
continue
excerpt = self._read_excerpt(repo_root / current_file)
verdict = self._probe(language, current_file, excerpt, attempts)
attempts.append(
Attempt(file_path=current_file, excerpt=excerpt, verdict=verdict)
)
log.info(
"iter=%d file=%s verdict=%s reason=%s",
iteration, current_file, verdict.verdict, verdict.reason,
)
if verdict.verdict == "entrypoint":
return AnalysisResult(
language=language,
entrypoint=current_file,
attempts=attempts,
reasoning=verdict.reason,
)
if verdict.verdict == "forward" and verdict.next_file:
carried = verdict.next_file
else:
if verdict.verdict == "forward":
log.warning("Verdict 'forward' without next_file; falling back to fresh pick.")
carried = None
raise RuntimeError(
f"Entrypoint not found after {self.max_iterations} iterations. "
f"Tried: {[a.file_path for a in attempts]}"
)
# -- LLM steps --------------------------------------------------------- #
def _guess_language(self, listing: List[str]) -> LanguageGuess:
system = (
"You are a senior code analyst. Given a repository's file listing "
"(relative paths), determine the primary programming language of "
"the project. Weigh file extensions, conventional filenames "
"(setup.py, package.json, Cargo.toml, go.mod, pom.xml, ...), and "
"directory structure."
)
user = "Repository file listing:\n" + _listing_repr(listing)
return self.llm.call(system, user, LanguageGuess)
def _pick_file(
self, listing: List[str], language: Language, attempts: List[Attempt]
) -> str:
system = (
"You are a senior code analyst. Pick exactly ONE file from the "
"provided listing that is the most promising candidate to be, or "
"to lead toward, the program's entrypoint. Prefer conventional "
"entrypoint names for the detected language "
"(main.py / __main__.py / app.py / cli.py / manage.py for Python; "
"index.js / server.js / main.ts for JS/TS; cmd/main.go for Go; "
"Program.cs / Startup.cs for C#; main.rs / lib.rs for Rust). "
"You MUST NOT pick a file that has already been tried. Return the "
"path exactly as it appears in the listing."
)
tried = [a.file_path for a in attempts]
user = (
f"Language: {language.value}\n\n"
f"Listing:\n{_listing_repr(listing)}\n\n"
f"Already tried (DO NOT pick these): {tried or '[]'}"
)
return self.llm.call(system, user, FilePick).file_path
def _probe(
self,
language: Language,
path: str,
excerpt: str,
attempts: List[Attempt],
) -> ProbeVerdict:
system = (
f"You are a senior code analyst locating the entrypoint of a "
f"{language.value} project. You are shown one file at a time and must "
f"decide exactly one of:\n"
f" - 'entrypoint': this file IS the program's entrypoint — e.g. it "
f"contains 'if __name__ == \"__main__\"', calls main() at module "
f"scope, does top-level CLI dispatch, is a declared binary target, "
f"or is the bootstrapping module itself.\n"
f" - 'forward': this file is not the entrypoint itself but clearly "
f"imports/dispatches into another file that is closer — set "
f"'next_file' to that file's path (it MUST appear in the listing).\n"
f" - 'wrong': this file is unrelated or a dead end — a different "
f"file must be picked.\n"
f"Be strict: only return 'entrypoint' when the file itself "
f"bootstraps execution."
)
tried = [a.file_path for a in attempts]
user = (
f"File under inspection: {path}\n\n"
f"--- content excerpt ---\n{excerpt}\n--- end excerpt ---\n\n"
f"Previously tried files: {tried or '[]'}"
)
return self.llm.call(system, user, ProbeVerdict)
# -- helpers ----------------------------------------------------------- #
def _read_excerpt(self, path: Path) -> str:
try:
text = path.read_text(encoding="utf-8", errors="replace")
except Exception as exc: # noqa: BLE001
return f"<unreadable: {exc}>"
if len(text) > self.excerpt_chars:
return text[: self.excerpt_chars] + "\n...<truncated>..."
return text
def gather_repo_entrypoint(llm: LLMClient, repo: RepoContext, cache_path: Optional[str] = None) -> AnalysisResult:
return EntrypointAnalyzer(
StructuredLLM(llm, cache=ResponseCache(Path(cache_path)) if cache_path else None)
).analyze(Path(repo.repo_dir))
agents/codebase_analyst/graph_tool.py
+850
View File
@@ -0,0 +1,850 @@
import json
import shutil
import subprocess
from abc import ABC, abstractmethod
from functools import partial
from pathlib import Path
from typing import Optional, Callable
import networkx as nx
import pandas as pd
from networkx.algorithms.community import louvain_communities
from pydantic import BaseModel, Field, ValidationError
from common.llm_client import LLMClient
from contracts.RepoContext import Language, ToolRequirement, DependencyGraph, DependencyNode, DependencyEdge, \
AnalysisResult, AnalysisInput, AnalysisError
def find_tool(command: str) -> Optional[str]:
"""Return the absolute path to `command` if it is on PATH, else None."""
return shutil.which(command)
TOOL_REGISTRY: dict[Language, ToolRequirement] = {
Language.PYTHON: ToolRequirement(
tool_name="pydeps",
command="pydeps",
install_guide="""\
## Install pydeps
```bash
pip install pydeps
You also need Graphviz for rendering:
Ubuntu/Debian: sudo apt install graphviz
macOS: brew install graphviz
Windows: Download from https://graphviz.org/download/
Ensure the dot command is on your PATH.""",
docs_url="https://github.com/thebjorn/pydeps",
),
Language.JAVASCRIPT: ToolRequirement(
tool_name="madge",
command="madge",
install_guide="""\
Install madge
bash
npm -g install madge
Graphviz (optional, for image output):
Ubuntu/Debian: sudo apt install graphviz
macOS: brew install graphviz""",
docs_url="https://github.com/pahen/madge",
),
Language.TYPESCRIPT: ToolRequirement(
tool_name="madge",
command="madge",
install_guide="""\
Install madge (TypeScript support)
bash
npm -g install madge
For TypeScript projects, ensure typescript is also installed:
bash
npm install -g typescript
```""",
docs_url="https://github.com/pahen/madge",
),
Language.JAVA: ToolRequirement(
tool_name="mvn",
command="mvn",
install_guide="""\
## Java dependency analysis
**Maven:**
```bash
mvn dependency:tree -DoutputFile=deps.txt
Gradle:
bash
gradle dependencies > deps.txt
Install Maven: https://maven.apache.org/install.html
Install Gradle: https://gradle.org/install/[reference:9]""",
docs_url="https://maven.apache.org/plugins/maven-dependency-plugin/",
),
Language.GO: ToolRequirement(
tool_name="go",
command="go",
install_guide="""\
Install Go
The Go toolchain provides built-in dependency listing:
bash
# Full module graph
go mod graph
# All modules (direct + indirect)
go list -m all
# Package import graph
go list -f '{{.ImportPath}} {{.Imports}}' ./...
Install Go: https://go.dev/doc/install[reference:10]""",
docs_url="https://go.dev/ref/mod#go-list-m",
),
Language.RUST: ToolRequirement(
tool_name="cargo",
command="cargo",
install_guide="""\
Install Rust / Cargo
cargo tree is built-in since Rust 1.44:
bash
cargo tree
cargo tree --depth 1 # direct deps only
cargo tree -i <crate> # reverse deps
Install Rust: https://www.rust-lang.org/tools/install[reference:11]""",
docs_url="https://doc.rust-lang.org/cargo/commands/cargo-tree.html",
),
Language.CPP: ToolRequirement(
tool_name="include-what-you-use",
command="include-what-you-use",
install_guide="""\
Install include-what-you-use (IWYU)
Ubuntu/Debian: sudo apt-get install iwyu
macOS: brew install include-what-you-use
Arch: sudo pacman -S include-what-you-use
Ensure include-what-you-use is on your PATH.""",
docs_url="https://include-what-you-use.org/",
),
Language.C: ToolRequirement(
tool_name="include-what-you-use",
command="include-what-you-use",
install_guide="""\
Install include-what-you-use (IWYU)
Ubuntu/Debian: sudo apt-get install iwyu
macOS: brew install include-what-you-use
Arch: sudo pacman -S include-what-you-use""",
docs_url="https://include-what-you-use.org/",
),
}
class LanguageAdapter(ABC):
"""Contract for every language-specific analyzer."""
@property
@abstractmethod
def language(self) -> Language: ...
@property
@abstractmethod
def required_tools(self) -> list[ToolRequirement]: ...
@abstractmethod
def build_graph(
self, repo_path: Path, entrypoint: str
) -> DependencyGraph: ...
class PythonAdapter(LanguageAdapter):
@property
def language(self) -> Language:
return Language.PYTHON
@property
def required_tools(self) -> list[ToolRequirement]:
return [TOOL_REGISTRY[Language.PYTHON]]
def build_graph(self, repo_path: Path, entrypoint: str) -> DependencyGraph:
entry_abs = repo_path / entrypoint
# pydeps emits JSON when --show-deps is used
result = subprocess.run(
[
"pydeps",
str(entry_abs),
"--show-deps",
"--no-output",
"--no-config",
"-T", "json",
],
capture_output=True,
text=True,
cwd=str(repo_path),
timeout=120,
)
if result.returncode != 0:
raise RuntimeError(
f"pydeps failed (rc={result.returncode}): {result.stderr.strip()}"
)
raw = json.loads(result.stdout)
nodes: list[DependencyNode] = []
edges: list[DependencyEdge] = []
seen: set[str] = set()
def _add_node(name: str, external: bool = False) -> None:
if name not in seen:
seen.add(name)
nodes.append(
DependencyNode(
id=name,
label=name.split(".")[-1],
language=Language.PYTHON,
is_external=external,
)
)
_add_node(entrypoint)
for mod_name, info in raw.items():
_add_node(mod_name, external=info.get("external", False))
for imported in info.get("imports", []):
_add_node(imported)
edges.append(
DependencyEdge(source=mod_name, target=imported, kind="import")
)
return DependencyGraph(
root=entrypoint,
nodes=nodes,
edges=edges,
metadata={"tool": "pydeps", "entrypoint": entrypoint},
)
class JsTsAdapter(LanguageAdapter):
def __init__(self, language: Language) -> None:
self._language = language
@property
def language(self) -> Language:
return self._language
@property
def required_tools(self) -> list[ToolRequirement]:
return [TOOL_REGISTRY[self._language]]
def build_graph(self, repo_path: Path, entrypoint: str) -> DependencyGraph:
entry_abs = repo_path / entrypoint
cmd = ["madge", "--json", str(entry_abs), "--basedir", str(repo_path)]
if self._language == Language.TYPESCRIPT:
cmd.append("--ts-config")
cmd.append(str(repo_path / "tsconfig.json"))
result = subprocess.run(
cmd,
capture_output=True,
text=True,
cwd=str(repo_path),
timeout=180,
)
if result.returncode != 0:
raise RuntimeError(
f"madge failed (rc={result.returncode}): {result.stderr.strip()}"
)
raw: dict[str, list[str]] = json.loads(result.stdout)
nodes: list[DependencyNode] = []
edges: list[DependencyEdge] = []
seen: set[str] = set()
def _add_node(name: str) -> None:
if name not in seen:
seen.add(name)
nodes.append(
DependencyNode(
id=name,
label=Path(name).name,
language=self._language,
is_external=not name.startswith("."),
)
)
_add_node(entrypoint)
for src, deps in raw.items():
_add_node(src)
for dep in deps:
_add_node(dep)
edges.append(
DependencyEdge(source=src, target=dep, kind="import")
)
return DependencyGraph(
root=entrypoint,
nodes=nodes,
edges=edges,
metadata={"tool": "madge", "entrypoint": entrypoint},
)
class GoAdapter(LanguageAdapter):
@property
def language(self) -> Language:
return Language.GO
@property
def required_tools(self) -> list[ToolRequirement]:
return [TOOL_REGISTRY[Language.GO]]
def build_graph(self, repo_path: Path, entrypoint: str) -> DependencyGraph:
result = subprocess.run(
[
"go", "list",
"-f", "{{.ImportPath}} {{join .Imports \" \"}}",
"./...",
],
capture_output=True,
text=True,
cwd=str(repo_path),
timeout=180,
)
if result.returncode != 0:
raise RuntimeError(
f"go list failed (rc={result.returncode}): {result.stderr.strip()}"
)
nodes: list[DependencyNode] = []
edges: list[DependencyEdge] = []
seen: set[str] = set()
def _add_node(name: str) -> None:
if name not in seen:
seen.add(name)
nodes.append(
DependencyNode(
id=name,
label=name.split("/")[-1],
language=Language.GO,
is_external="/" in name and not name.startswith("."),
)
)
for line in result.stdout.strip().splitlines():
parts = line.split()
if not parts:
continue
pkg = parts[0]
_add_node(pkg)
for imp in parts[1:]:
_add_node(imp)
edges.append(
DependencyEdge(source=pkg, target=imp, kind="import")
)
return DependencyGraph(
root=entrypoint,
nodes=nodes,
edges=edges,
metadata={"tool": "go list", "entrypoint": entrypoint},
)
class RustAdapter(LanguageAdapter):
@property
def language(self) -> Language:
return Language.RUST
@property
def required_tools(self) -> list[ToolRequirement]:
return [TOOL_REGISTRY[Language.RUST]]
def build_graph(self, repo_path: Path, entrypoint: str) -> DependencyGraph:
result = subprocess.run(
["cargo", "tree", "--prefix", "none", "--format", "{p} {f}"],
capture_output=True,
text=True,
cwd=str(repo_path),
timeout=180,
)
if result.returncode != 0:
raise RuntimeError(
f"cargo tree failed (rc={result.returncode}): {result.stderr.strip()}"
)
nodes: list[DependencyNode] = []
edges: list[DependencyEdge] = []
seen: set[str] = set()
def _add_node(name: str) -> None:
if name not in seen:
seen.add(name)
nodes.append(
DependencyNode(
id=name,
label=name,
language=Language.RUST,
is_external=False, # refined below
)
)
# cargo tree --prefix none outputs one package per line,
# with indentation indicating depth. We parse the flat form
# and reconstruct edges from indentation.
lines = result.stdout.strip().splitlines()
stack: list[str] = []
for line in lines:
if not line.strip():
continue
indent = len(line) - len(line.lstrip("│ ├─└ "))
name = line.strip().split(" ")[0]
_add_node(name)
while len(stack) > indent:
stack.pop()
if stack:
edges.append(
DependencyEdge(source=stack[-1], target=name, kind="depends_on")
)
stack.append(name)
return DependencyGraph(
root=entrypoint,
nodes=nodes,
edges=edges,
metadata={"tool": "cargo tree", "entrypoint": entrypoint},
)
class JavaAdapter(LanguageAdapter):
@property
def language(self) -> Language:
return Language.JAVA
@property
def required_tools(self) -> list[ToolRequirement]:
return [TOOL_REGISTRY[Language.JAVA]]
def build_graph(self, repo_path: Path, entrypoint: str) -> DependencyGraph:
if (repo_path / "pom.xml").exists():
return self._maven_graph(repo_path, entrypoint)
if (repo_path / "build.gradle").exists() or (repo_path / "build.gradle.kts").exists():
return self._gradle_graph(repo_path, entrypoint)
raise RuntimeError("No pom.xml or build.gradle found in repository root")
def _maven_graph(self, repo_path: Path, entrypoint: str) -> DependencyGraph:
result = subprocess.run(
["mvn", "dependency:tree", "-DoutputType=text"],
capture_output=True,
text=True,
cwd=str(repo_path),
timeout=300,
)
if result.returncode != 0:
raise RuntimeError(f"mvn dependency:tree failed: {result.stderr.strip()}")
return self._parse_tree_text(result.stdout, entrypoint)
def _gradle_graph(self, repo_path: Path, entrypoint: str) -> DependencyGraph:
result = subprocess.run(
["gradle", "dependencies", "--configuration", "runtimeClasspath"],
capture_output=True,
text=True,
cwd=str(repo_path),
timeout=300,
)
if result.returncode != 0:
raise RuntimeError(f"gradle dependencies failed: {result.stderr.strip()}")
return self._parse_tree_text(result.stdout, entrypoint)
def _parse_tree_text(self, text: str, entrypoint: str) -> DependencyGraph:
# Simplified parser for the textual tree format.
# In production, use the JSON output of the Maven/Gradle plugin.
nodes: list[DependencyNode] = []
edges: list[DependencyEdge] = []
seen: set[str] = set()
def _add_node(name: str) -> None:
if name not in seen:
seen.add(name)
nodes.append(
DependencyNode(
id=name,
label=name.split(":")[-1] if ":" in name else name,
language=Language.JAVA,
is_external=True,
)
)
_add_node(entrypoint)
for line in text.splitlines():
stripped = line.strip()
if not stripped or stripped.startswith("+-") or stripped.startswith("\\-"):
continue
# crude: treat every artifact line as a dependency of the root
name = stripped.split(":")[0]
if name and name != entrypoint:
_add_node(name)
edges.append(
DependencyEdge(source=entrypoint, target=name, kind="depends_on")
)
return DependencyGraph(
root=entrypoint,
nodes=nodes,
edges=edges,
metadata={"tool": "maven/gradle", "entrypoint": entrypoint},
)
class CppAdapter(LanguageAdapter):
@property
def language(self) -> Language:
return Language.CPP
@property
def required_tools(self) -> list[ToolRequirement]:
return [TOOL_REGISTRY[Language.CPP]]
def build_graph(self, repo_path: Path, entrypoint: str) -> DependencyGraph:
entry_abs = repo_path / entrypoint
result = subprocess.run(
[
"include-what-you-use",
"-Xiwyu", "--no_fwd_decls",
str(entry_abs),
],
capture_output=True,
text=True,
cwd=str(repo_path),
timeout=180,
)
# IWYU returns non-zero when it recommends changes; parse stdout anyway
output = result.stdout + result.stderr
nodes: list[DependencyNode] = []
edges: list[DependencyEdge] = []
seen: set[str] = set()
def _add_node(name: str) -> None:
if name not in seen:
seen.add(name)
nodes.append(
DependencyNode(
id=name,
label=Path(name).name,
language=Language.CPP,
is_external=not name.startswith("."),
)
)
_add_node(entrypoint)
# IWYU output contains lines like:
# #include <vector> // for ...
# #include "foo.h" // for ...
import re
pattern = re.compile(r'#include\s+[<"]([^>"]+)[>"]')
for match in pattern.finditer(output):
header = match.group(1)
_add_node(header)
edges.append(
DependencyEdge(source=entrypoint, target=header, kind="include")
)
return DependencyGraph(
root=entrypoint,
nodes=nodes,
edges=edges,
metadata={"tool": "include-what-you-use", "entrypoint": entrypoint},
)
class LLMDependencyInference(BaseModel):
"""Schema the LLM must fill to infer dependencies."""
nodes: list[dict[str, str]]
edges: list[dict[str, str]]
confidence: float = Field(ge=0.0, le=1.0)
SYSTEM_PROMPT = (
"You are a static-analysis assistant. Given the source code of a file, "
"extract all direct dependencies (imports, includes, requires). "
"Return ONLY valid JSON matching the provided schema."
)
def infer_dependencies_with_llm(
llm: LLMClient,
file_content: str,
language: str,
) -> LLMDependencyInference:
prompt = (
f"Language: {language}\n"
f"Source code:\n```\n{file_content}\n```\n"
"Extract all dependencies as a list of nodes and edges."
)
schema = LLMDependencyInference.model_json_schema()
raw = llm.chat_with_schema(
prompt,
{
"type": "json_schema",
"json_schema": {
"name": "LLMDependencyInference",
"schema": schema,
"strict": True,
},
},
system=SYSTEM_PROMPT,
)
try:
if isinstance(raw, str):
raw = json.loads(raw)
return LLMDependencyInference.model_validate(raw)
except ValidationError as exc:
# Hard failure — do not build downstream steps on malformed data.
raise RuntimeError(f"LLM returned invalid dependency schema: {exc}") from exc
def emit_youtrack_comment(result: AnalysisResult, issue_id: str) -> None:
"""Post a single comment to Youtrack describing the analysis outcome."""
if result.success:
# Optionally post a summary; by convention, avoid duplicating comments.
return
err = result.error
body = f"""
**Dependency analysis failed** (`{err.error_code}`)
{err.message}
"""
if err.missing_tools:
for tool in err.missing_tools:
body += f"\n\n### {tool.tool_name}\n\n{tool.install_guide}"
print(body)
AdapterFactory = Callable[[], LanguageAdapter]
ADAPTER_REGISTRY: dict[Language, AdapterFactory] = {
Language.PYTHON: PythonAdapter, # type[LanguageAdapter] is Callable[[], LanguageAdapter]
Language.JAVASCRIPT: partial(JsTsAdapter, Language.JAVASCRIPT), # -> JsTsAdapter
Language.TYPESCRIPT: partial(JsTsAdapter, Language.TYPESCRIPT), # -> JsTsAdapter
Language.JAVA: JavaAdapter,
Language.GO: GoAdapter,
Language.RUST: RustAdapter,
Language.CPP: partial(CppAdapter, Language.CPP), # if CppAdapter takes a language
Language.C: partial(CppAdapter, Language.C),
}
def analyze_repository(inp: AnalysisInput) -> AnalysisResult:
"""
Idempotent entry point.
Given the same AnalysisInput, always returns the same AnalysisResult
(either a graph or a structured error). No files are written to the
repository under analysis.
"""
adapter_factory = ADAPTER_REGISTRY.get(inp.language)
if adapter_factory is None:
return AnalysisResult(
success=False,
error=AnalysisError(
error_code="UNSUPPORTED_LANGUAGE",
message=f"No adapter registered for language '{inp.language}'",
language=inp.language,
entrypoint=inp.entrypoint,
),
)
adapter: LanguageAdapter = adapter_factory()
# --- Tool detection ---
missing: list[ToolRequirement] = []
for req in adapter.required_tools:
if find_tool(req.command) is None:
missing.append(req)
if missing:
return AnalysisResult(
success=False,
error=AnalysisError(
error_code="MISSING_TOOL",
message=(
f"Cannot build dependency graph for '{inp.language.value}': "
f"required tool(s) not found on PATH: "
f"{', '.join(r.tool_name for r in missing)}"
),
missing_tools=missing,
language=inp.language,
entrypoint=inp.entrypoint,
),
)
# --- Build graph ---
try:
graph = adapter.build_graph(inp.repo_path, inp.entrypoint)
except subprocess.TimeoutExpired:
return AnalysisResult(
success=False,
error=AnalysisError(
error_code="TOOL_TIMEOUT",
message=f"Tool timed out while analyzing {inp.entrypoint}",
language=inp.language,
entrypoint=inp.entrypoint,
),
)
except Exception as exc:
return AnalysisResult(
success=False,
error=AnalysisError(
error_code="ANALYSIS_FAILED",
message=str(exc),
language=inp.language,
entrypoint=inp.entrypoint,
),
)
G_all = to_networkx(graph, include_external=True)
G = to_networkx(graph, include_external=False) # internal-only analysis
res_internal = analyze(G, graph, label="_internal")
res_all = analyze(G_all, graph, label="_all")
leaf_clusters = cluster_leaves(G, res_internal["leaf_no_deps"])
return AnalysisResult(success=True, graph=graph, res_all=res_all, leaf_clusters=leaf_clusters)
def to_networkx(dep_graph: DependencyGraph, include_external=False):
G = nx.DiGraph()
# Nodes
for n in dep_graph.nodes:
if not include_external and n.is_external:
continue
G.add_node(n.id, label=n.label, language=n.language.value, is_external=n.is_external)
# Edges (source depends on target)
for e in dep_graph.edges:
if e.source in G and e.target in G:
G.add_edge(e.source, e.target, kind=e.kind)
# Add isolated internal files that never appear in edges
if not include_external:
for n in dep_graph.nodes:
if not n.is_external and n.id not in G:
G.add_node(n.id, label=n.label, language=n.language.value, is_external=False)
return G
def package_of(path: str) -> str:
parts = path.split("/")
return "/".join(parts[:-1]) or "."
COLS = [
"file", "language", "in_degree", "out_degree",
"isolated", "leaf_no_deps", "entry_no_dependents",
"used_by_files", "used_by_packages", "community",
]
def analyze(G, graph, label=""):
in_deg = dict(G.in_degree())
out_deg = dict(G.out_degree())
# 1. Truly isolated: no in and no out edges
isolated = [n for n in G if G.degree(n) == 0]
# 2. Leaf: depends on nothing
leaf_no_deps = [n for n in G if G.out_degree(n) == 0]
# 3. Entrypoint: nothing depends on it
entry_no_dependents = [n for n in G if G.in_degree(n) == 0]
# 4. Widely used: how many distinct files depend on it
widely_used = sorted(in_deg.items(), key=lambda kv: kv[1], reverse=True)
# 5. How many distinct packages depend on it
pkg = {n: package_of(n) for n in G}
used_by_packages = {
n: len({pkg[p] for p in G.predecessors(n)}) for n in G
}
# 6. Weakly connected components = isolated groups
wccs = sorted(nx.weakly_connected_components(G), key=len, reverse=True)
# 7. Strongly connected components = cycles / tightly coupled groups
sccs = sorted(
(c for c in nx.strongly_connected_components(G) if len(c) > 1),
key=len, reverse=True,
)
# 8. Communities via Louvain
U = G.to_undirected()
comms = louvain_communities(U, seed=42) if U.number_of_edges() else []
node2comm = {n: i for i, c in enumerate(comms) for n in c}
# Per-file table
rows = []
for n, data in G.nodes(data=True):
rows.append({
"file": n,
"language": data.get("language"),
"in_degree": in_deg[n],
"out_degree": out_deg[n],
"isolated": G.degree(n) == 0,
"leaf_no_deps": out_deg[n] == 0,
"entry_no_dependents": in_deg[n] == 0,
"used_by_files": in_deg[n],
"used_by_packages": used_by_packages[n],
"community": node2comm.get(n, -1),
})
df = pd.DataFrame(rows, columns=COLS)
if not df.empty:
df = df.sort_values(
["used_by_packages", "used_by_files"],
ascending=False,
)
df.to_csv(f"file_metrics{label}.csv", index=False)
pd.DataFrame([
{"group_id": i, "size": len(c), "members": ";".join(sorted(c))}
for i, c in enumerate(wccs)
]).to_csv(f"isolated_groups{label}.csv", index=False)
pd.DataFrame([
{"scc_id": i, "size": len(c), "members": ";".join(sorted(c))}
for i, c in enumerate(sccs)
]).to_csv(f"cycles{label}.csv", index=False)
return {
"df": df,
"isolated": isolated,
"leaf_no_deps": leaf_no_deps,
"entry_no_dependents": entry_no_dependents,
"widely_used": widely_used,
"used_by_packages": used_by_packages,
"wccs": wccs,
"sccs": sccs,
"node2comm": node2comm,
"comms": comms,
}
def cluster_leaves(G, leaves):
L = nx.Graph()
L.add_nodes_from(leaves)
parent_to_leaves = {}
for leaf in leaves:
for parent in G.predecessors(leaf):
parent_to_leaves.setdefault(parent, []).append(leaf)
# Leaves used by the same parent get connected
for parent, ls in parent_to_leaves.items():
for i in range(len(ls)):
for j in range(i + 1, len(ls)):
L.add_edge(ls[i], ls[j])
clusters = sorted(nx.connected_components(L), key=len, reverse=True)
return clusters