tirth8205--code-review-graph/code_review_graph/tools/analysis_tools.py

"""MCP tool wrappers for graph analysis features."""

from __future__ import annotations

from typing import Any

from ..analysis import (
    find_bridge_nodes,
    find_hub_nodes,
    find_knowledge_gaps,
    find_surprising_connections,
    generate_suggested_questions,
)
from ._common import _get_store


def get_hub_nodes_func(
    repo_root: str | None = None,
    top_n: int = 10,
) -> dict[str, Any]:
    """Find the most connected nodes in the codebase graph.

    Hub nodes have the highest total degree (in + out edges).
    These are architectural hotspots -- changes to them have
    disproportionate blast radius.

    Args:
        repo_root: Repository root (auto-detected if omitted).
        top_n: Number of top hubs to return (default 10).
    """
    store, _root = _get_store(repo_root or None)
    try:
        hubs = find_hub_nodes(store, top_n=top_n)
        return {
            "hub_nodes": hubs,
            "count": len(hubs),
            "next_tool_suggestions": [
                "get_impact_radius -- check blast radius of a hub",
                "query_graph callers_of -- see what calls a hub",
                "get_bridge_nodes -- find architectural chokepoints",
            ],
        }
    finally:
        store.close()


def get_bridge_nodes_func(
    repo_root: str | None = None,
    top_n: int = 10,
) -> dict[str, Any]:
    """Find architectural chokepoints via betweenness centrality.

    Bridge nodes sit on the shortest paths between many node
    pairs. If they break, multiple code regions lose
    connectivity.

    Args:
        repo_root: Repository root (auto-detected if omitted).
        top_n: Number of top bridges to return (default 10).
    """
    store, _root = _get_store(repo_root or None)
    try:
        bridges = find_bridge_nodes(store, top_n=top_n)
        return {
            "bridge_nodes": bridges,
            "count": len(bridges),
            "next_tool_suggestions": [
                "get_hub_nodes -- find most connected nodes",
                "get_impact_radius -- check blast radius",
                "detect_changes -- see if bridges are affected",
            ],
        }
    finally:
        store.close()


def get_knowledge_gaps_func(
    repo_root: str | None = None,
) -> dict[str, Any]:
    """Identify structural weaknesses in the codebase.

    Finds: isolated nodes (disconnected), thin communities
    (< 3 members), untested hotspots (high-degree, no tests),
    and single-file communities.

    Args:
        repo_root: Repository root (auto-detected if omitted).
    """
    store, _root = _get_store(repo_root or None)
    try:
        gaps = find_knowledge_gaps(store)
        total = sum(len(v) for v in gaps.values())
        return {
            "gaps": gaps,
            "total_gaps": total,
            "summary": {
                "isolated_nodes": len(gaps["isolated_nodes"]),
                "thin_communities": len(
                    gaps["thin_communities"]
                ),
                "untested_hotspots": len(
                    gaps["untested_hotspots"]
                ),
                "single_file_communities": len(
                    gaps["single_file_communities"]
                ),
            },
            "next_tool_suggestions": [
                "refactor dead_code -- find unused symbols",
                "get_hub_nodes -- find high-impact nodes",
                "get_suggested_questions -- review prompts",
            ],
        }
    finally:
        store.close()


def get_surprising_connections_func(
    repo_root: str | None = None,
    top_n: int = 15,
) -> dict[str, Any]:
    """Find unexpected architectural coupling in the codebase.

    Scores edges by surprise factors: cross-community,
    cross-language, peripheral-to-hub, cross-test-boundary.

    Args:
        repo_root: Repository root (auto-detected if omitted).
        top_n: Number of top surprises to return (default 15).
    """
    store, _root = _get_store(repo_root or None)
    try:
        surprises = find_surprising_connections(
            store, top_n=top_n
        )
        return {
            "surprising_connections": surprises,
            "count": len(surprises),
            "next_tool_suggestions": [
                "get_architecture_overview -- community structure",
                "query_graph callers_of -- trace the coupling",
                "get_bridge_nodes -- find chokepoints",
            ],
        }
    finally:
        store.close()


def get_suggested_questions_func(
    repo_root: str | None = None,
) -> dict[str, Any]:
    """Auto-generate review questions from graph analysis.

    Produces questions about: bridge nodes, untested hubs,
    surprising connections, thin communities, and untested
    hotspots.

    Args:
        repo_root: Repository root (auto-detected if omitted).
    """
    store, _root = _get_store(repo_root or None)
    try:
        questions = generate_suggested_questions(store)
        by_priority: dict[str, list[dict[str, Any]]] = {
            "high": [], "medium": [], "low": [],
        }
        for q in questions:
            prio = q.get("priority", "medium")
            if prio in by_priority:
                by_priority[prio].append(q)
        return {
            "questions": questions,
            "count": len(questions),
            "by_priority": {
                k: len(v) for k, v in by_priority.items()
            },
            "next_tool_suggestions": [
                "get_knowledge_gaps -- structural weaknesses",
                "detect_changes -- risk-scored review",
                "get_architecture_overview -- community map",
            ],
        }
    finally:
        store.close()