Repository Intelligence 2026: AI Code Understanding for Enterprise Scale

from dataclasses import dataclass
from typing import List, Dict, Set
import ast
import networkx as nx

@dataclass
class CodeEntity:
    """Represents any code entity (function, class, module)"""
    name: str
    type: str  # "function", "class", "module", "service"
    file_path: str
    start_line: int
    end_line: int
    dependencies: Set[str]  # Other entities this depends on
    dependents: Set[str]    # Entities that depend on this
    last_modified: str       # Git commit hash
    change_frequency: int    # Commits touching this entity
    author_count: int        # Unique contributors

class RepositoryIntelligenceEngine:
    """Build persistent view of repository structure and relationships"""

    def __init__(self, repo_path: str):
        self.repo_path = repo_path
        self.dependency_graph = nx.DiGraph()
        self.entity_index: Dict[str, CodeEntity] = {}
        self.module_boundaries: Dict[str, List[str]] = {}

    def build_index(self):
        """Parse repository and build comprehensive entity index"""
        # Phase 1: Discover all entities
        for file_path in self._discover_source_files():
            entities = self._parse_file(file_path)
            for entity in entities:
                self.entity_index[entity.name] = entity
                self.dependency_graph.add_node(
                    entity.name,
                    data=entity
                )

        # Phase 2: Build dependency graph
        for entity_name, entity in self.entity_index.items():
            for dep in entity.dependencies:
                if dep in self.entity_index:
                    self.dependency_graph.add_edge(entity_name, dep)

        # Phase 3: Identify module boundaries
        self.module_boundaries = self._detect_modules()

        # Phase 4: Analyze change patterns
        self._analyze_git_history()

    def _detect_modules(self) -> Dict[str, List[str]]:
        """Identify cohesive modules using graph clustering"""
        # Use Louvain community detection to find natural module boundaries
        communities = nx.community.louvain_communities(
            self.dependency_graph.to_undirected()
        )

        modules = {}
        for idx, community in enumerate(communities):
            module_name = f"module_{idx}"
            modules[module_name] = list(community)

        return modules

    def analyze_pr_impact(
        self,
        changed_files: List[str]
    ) -> Dict:
        """Analyze architectural impact of pull request"""

        affected_entities = set()
        affected_modules = set()
        risk_score = 0.0

        # Find all entities modified in PR
        for file_path in changed_files:
            for entity_name, entity in self.entity_index.items():
                if entity.file_path == file_path:
                    affected_entities.add(entity_name)

        # Calculate impact radius using dependency graph
        for entity_name in affected_entities:
            # Downstream impact (what depends on this?)
            dependents = nx.descendants(self.dependency_graph, entity_name)
            affected_entities.update(dependents)

            # Check if this crosses module boundaries
            for module, entities in self.module_boundaries.items():
                if entity_name in entities:
                    affected_modules.add(module)

        # Calculate risk score
        risk_factors = {
            "entity_count": len(affected_entities),
            "module_span": len(affected_modules),
            "cross_boundary": len(affected_modules) > 1,
            "high_frequency_zone": self._in_hot_zone(affected_entities)
        }

        risk_score = self._calculate_risk(risk_factors)

        return {
            "affected_entities": list(affected_entities),
            "affected_modules": list(affected_modules),
            "risk_score": risk_score,  # 0-100
            "risk_factors": risk_factors,
            "review_recommendations": self._generate_recommendations(risk_factors)
        }

    def _calculate_risk(self, factors: Dict) -> float:
        """Calculate PR risk score 0-100"""
        score = 0.0

        # Entity count impact (0-30 points)
        entity_count = factors["entity_count"]
        score += min(30, entity_count * 0.5)

        # Module spanning (0-25 points)
        if factors["cross_boundary"]:
            score += 25

        # High-change area (0-25 points)
        if factors["high_frequency_zone"]:
            score += 25

        # Size multiplier (0-20 points)
        module_span = factors["module_span"]
        score += min(20, module_span * 5)

        return min(100.0, score)

class CodePatternRecognizer:
    """Identify recurring patterns in codebase"""

    def __init__(self, engine: RepositoryIntelligenceEngine):
        self.engine = engine
        self.patterns = {
            "initialization": [],
            "error_handling": [],
            "api_endpoints": [],
            "database_queries": [],
            "configuration": []
        }

    def learn_patterns(self):
        """Extract common patterns from existing code"""

        for entity_name, entity in self.engine.entity_index.items():
            # Analyze AST for pattern matching
            tree = self._get_ast(entity.file_path)

            # Initialization pattern
            if self._matches_init_pattern(tree):
                self.patterns["initialization"].append({
                    "entity": entity_name,
                    "pattern": self._extract_pattern(tree),
                    "frequency": entity.change_frequency
                })

            # Error handling pattern
            if self._has_error_handling(tree):
                self.patterns["error_handling"].append({
                    "entity": entity_name,
                    "style": self._extract_error_style(tree)
                })

    def suggest_pattern_alignment(
        self,
        new_code: str,
        context_entities: List[str]
    ) -> Dict:
        """Suggest pattern alignment for new code"""

        # Parse new code
        new_tree = ast.parse(new_code)

        # Find dominant patterns in context
        context_patterns = self._get_context_patterns(context_entities)

        # Check for pattern violations
        violations = []

        # Example: Error handling style consistency
        new_error_style = self._extract_error_style(new_tree)
        dominant_style = self._get_dominant_style(
            context_patterns["error_handling"]
        )

        if new_error_style != dominant_style:
            violations.append({
                "type": "error_handling_style_mismatch",
                "current": new_error_style,
                "expected": dominant_style,
                "recommendation": self._generate_alignment_code(dominant_style)
            })

        return {
            "pattern_compliance": len(violations) == 0,
            "violations": violations,
            "context_patterns": context_patterns
        }

    def _get_dominant_style(self, patterns: List[Dict]) -> str:
        """Identify most common pattern in codebase"""
        from collections import Counter

        styles = [p["style"] for p in patterns]
        return Counter(styles).most_common(1)[0][0]

class MultiRepoIntelligence:
    """Coordinate intelligence across multiple repositories"""

    def __init__(self, repos: List[str]):
        self.repos = repos
        self.engines: Dict[str, RepositoryIntelligenceEngine] = {}
        self.cross_repo_deps = nx.DiGraph()

    def build_global_index(self):
        """Build unified index across all repositories"""

        # Build individual repository indexes
        for repo_path in self.repos:
            engine = RepositoryIntelligenceEngine(repo_path)
            engine.build_index()
            self.engines[repo_path] = engine

        # Build cross-repository dependency graph
        self._build_cross_repo_dependencies()

    def _build_cross_repo_dependencies(self):
        """Detect dependencies across repository boundaries"""

        # Example: Service A in Repo1 calls Service B in Repo2
        for repo1_path, engine1 in self.engines.items():
            for entity1_name, entity1 in engine1.entity_index.items():

                # Check if entity1 references entities in other repos
                for repo2_path, engine2 in self.engines.items():
                    if repo1_path == repo2_path:
                        continue

                    for entity2_name, entity2 in engine2.entity_index.items():
                        if self._has_cross_repo_reference(entity1, entity2):
                            self.cross_repo_deps.add_edge(
                                (repo1_path, entity1_name),
                                (repo2_path, entity2_name)
                            )

    def analyze_breaking_change_impact(
        self,
        repo: str,
        changed_entity: str
    ) -> Dict:
        """Analyze impact of breaking changes across repos"""

        # Find all downstream dependents across repos
        affected_repos = set()

        node = (repo, changed_entity)
        if node in self.cross_repo_deps:
            # Get all descendants in cross-repo graph
            descendants = nx.descendants(self.cross_repo_deps, node)

            for downstream_repo, downstream_entity in descendants:
                affected_repos.add(downstream_repo)

        return {
            "breaking_change_propagation": list(affected_repos),
            "affected_services": len(affected_repos),
            "coordination_required": len(affected_repos) > 0,
            "deployment_order": self._calculate_deployment_order(node)
        }