Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 17 additions & 7 deletions src/cgis/api/mcp_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -380,21 +380,31 @@ def cgis_context(


@mcp.tool()
def cgis_metrics(db_path: str = _DEFAULT_DB, limit: int = 10) -> str:
def cgis_metrics(
db_path: str = _DEFAULT_DB, limit: int = 10, exclude: list[str] | None = None
) -> str:
"""Whole-graph architectural metrics β€” coupling bottlenecks + God classes (#16).

Returns JSON ``{bottlenecks, god_classes}`` computed with vectorized DuckDB
aggregations over the whole graph (fan-in/fan-out coupling, declared-member
counts) β€” the global "what are the hotspots?" view that complements the
node-local trace/impact/context tools. Requires the optional ``duckdb``
extra; an unavailable dependency is reported as a normal ❌ message.
Returns JSON ``{bottlenecks, god_classes, critical}`` computed with vectorized
DuckDB aggregations over the whole graph (fan-in/fan-out coupling,
declared-member counts, PageRank) β€” the global "what are the hotspots?" view
that complements the node-local trace/impact/context tools. Requires the
optional ``duckdb`` extra; an unavailable dependency is reported as a normal
❌ message.

``exclude`` drops any node whose FQN contains one of the given dot-segments
(e.g. ``["tests"]`` removes both ``tests.*`` and ``domains.*.tests.*``) so
test/vendor scaffolding stays out of the rankings.
"""
if not Path(db_path).exists():
return f"❌ Database not found at: {db_path}. Run cgis_ingest first."
try:
with DuckDBAnalyzer(db_path) as analyzer:
report = analyzer.architecture_report(
bottleneck_limit=limit, god_limit=limit, critical_limit=limit
bottleneck_limit=limit,
god_limit=limit,
critical_limit=limit,
exclude=exclude or [],
)
except Exception as exc:
return f"❌ {exc}"
Expand Down
12 changes: 11 additions & 1 deletion src/cgis/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -1141,6 +1141,15 @@ def _render_metrics(report: ArchitectureReport) -> None:
def metrics(
db: str = typer.Option(_DEFAULT_DB, "--db", "-d", help=_DEFAULT_DB_HELP),
limit: int = typer.Option(10, "--limit", min=1, help="Top-N rows per section."),
exclude: list[str] = typer.Option(
[],
"--exclude",
"-x",
help=(
"Drop nodes whose FQN contains this dot-segment anywhere, e.g. "
"'-x tests' removes both tests.* and domains.*.tests.* (repeatable)."
),
),
output_format: OutputFormat = typer.Option(
OutputFormat.TEXT, "--format", "-f", help=_TEXT_JSON_FORMAT_HELP
),
Expand All @@ -1149,6 +1158,7 @@ def metrics(

Runs vectorized aggregations over the graph via an optional DuckDB layer.
Install it with `pip install 'codegraph-brain[analytics]'` if missing.
Use `--exclude tests` to keep test scaffolding out of the rankings.
"""
if output_format == OutputFormat.MERMAID:
console.print("[bold red]❌ metrics supports --format text or json only.[/bold red]")
Expand All @@ -1159,7 +1169,7 @@ def metrics(
try:
with DuckDBAnalyzer(db) as analyzer:
report = analyzer.architecture_report(
bottleneck_limit=limit, god_limit=limit, critical_limit=limit
bottleneck_limit=limit, god_limit=limit, critical_limit=limit, exclude=exclude
)
except Exception as e: # duckdb missing, extension fetch, or a non-SQLite file
console.print(f"[bold red]❌ {e}[/bold red]")
Expand Down
90 changes: 73 additions & 17 deletions src/cgis/query/metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
raises a clear error so the CLI can degrade gracefully.
"""

from collections.abc import Sequence
from pathlib import Path
from typing import Any

Expand All @@ -33,7 +34,38 @@
"Install it with: pip install 'codegraph-brain[analytics]' (or: uv add duckdb)"
)

_COUPLING_QUERY = f"""

def _segment_exclusion(id_expr: str, segments: Sequence[str]) -> tuple[str, list[str]]:
"""Build a WHERE fragment excluding FQNs that contain any of ``segments``.

A segment matches a whole dot-delimited component **anywhere** in the id, so
``tests`` drops both ``tests.utils.x`` and ``domains.resv.tests.test_x`` while
keeping ``domains.testservice`` (substring, not a segment). Returns
``("", [])`` when ``segments`` is empty. The fragment uses ``?`` placeholders
(the segments ride in as query parameters, never string-interpolated) and
``LIKE … ESCAPE '\\'`` with ``%``/``_``/``\\`` escaped, so a segment is matched
literally and the path stays injection-safe.
"""
if not segments: # reachable empty case (default ()) β€” and guards a None caller
return "", []
clauses: list[str] = []
params: list[str] = []
# dict.fromkeys dedupes while preserving order; skip empty/whitespace segments
# (they would yield a '%..%' pattern that matches no real FQN β€” a silent no-op).
for seg in dict.fromkeys(segments):
Comment thread
zaebee marked this conversation as resolved.
if not seg.strip():
continue
escaped = seg.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
clauses.append(f"('.' || {id_expr} || '.') NOT LIKE ? ESCAPE '\\'")
params.append(f"%.{escaped}.%")
if not clauses:
return "", []
return " AND " + " AND ".join(clauses), params


def _coupling_query(exclude_sql: str) -> str:
"""Coupling query with an optional segment-exclusion fragment on the node list."""
return f"""
WITH incoming AS (
SELECT target AS node_id, COUNT(*) AS in_deg
FROM edges WHERE type = 'CALLS' GROUP BY target
Expand All @@ -57,21 +89,25 @@
LEFT JOIN outgoing o ON n.id = o.node_id
WHERE n.namespace = 'INTERNAL'
AND n.type IN ('FUNCTION', 'METHOD')
AND n.file_path != '{VIRTUAL_FILE_PATH}'
AND n.file_path != '{VIRTUAL_FILE_PATH}'{exclude_sql}
ORDER BY (COALESCE(i.in_deg, 0) + COALESCE(o.out_deg, 0)) DESC, n.id
LIMIT ?
"""

_GOD_CLASS_QUERY = """

def _god_class_query(exclude_sql: str) -> str:
"""God-class query with an optional segment-exclusion fragment on the class id."""
return f"""
SELECT e.source AS node_id, COUNT(*) AS declares
FROM edges e
JOIN nodes n ON e.source = n.id
WHERE e.type = 'DECLARES' AND n.type = 'CLASS'
WHERE e.type = 'DECLARES' AND n.type = 'CLASS'{exclude_sql}
GROUP BY e.source
ORDER BY declares DESC, e.source
LIMIT ?
"""


# Vectorized PageRank: 20 fixed iterations of the standard damped formula, run as
# DuckDB temp-table joins (Python only drives the loop β€” the data never leaves the
# in-memory DuckDB). Dangling nodes (no outgoing internal CALLS) redistribute their
Expand Down Expand Up @@ -183,24 +219,30 @@ def _rows_to_metrics(self, rows: list[tuple[Any, ...]]) -> list[NodeMetric]:
for node_id, node_type, in_degree, out_degree in rows
]

def get_coupling_metrics(self, limit: int = 10) -> list[NodeMetric]:
def get_coupling_metrics(
self, limit: int = 10, exclude: Sequence[str] = ()
) -> list[NodeMetric]:
"""Top INTERNAL functions/methods by total coupling (fan-in + fan-out).

High in-degree marks a critical bottleneck (many callers); high out-degree
marks an over-orchestrating or complex unit. External/stdlib and
unresolved ``raw_call:`` targets are excluded so ``len``/``print`` noise
never tops the list.
never tops the list. ``exclude`` drops any node whose FQN contains one of
the given dot-segments (e.g. ``["tests"]``) from the ranked list.
"""
rows = self.conn.execute(_COUPLING_QUERY, [limit]).fetchall()
where, params = _segment_exclusion("n.id", exclude)
rows = self.conn.execute(_coupling_query(where), [*params, limit]).fetchall()
return self._rows_to_metrics(rows)

def get_god_classes(self, limit: int = 5) -> list[NodeMetric]:
def get_god_classes(self, limit: int = 5, exclude: Sequence[str] = ()) -> list[NodeMetric]:
"""Top classes by declared-member count (DECLARES fan-out).

``out_degree`` carries the number of methods/attributes the class
declares β€” a high value is the classic God-object smell.
declares β€” a high value is the classic God-object smell. ``exclude`` drops
classes whose FQN contains one of the given dot-segments.
"""
rows = self.conn.execute(_GOD_CLASS_QUERY, [limit]).fetchall()
where, params = _segment_exclusion("n.id", exclude)
rows = self.conn.execute(_god_class_query(where), [*params, limit]).fetchall()
return [
NodeMetric(
node_id=str(node_id),
Expand All @@ -211,7 +253,7 @@ def get_god_classes(self, limit: int = 5) -> list[NodeMetric]:
for node_id, declares in rows
]

def get_pagerank(self, limit: int = 10) -> list[NodeMetric]:
def get_pagerank(self, limit: int = 10, exclude: Sequence[str] = ()) -> list[NodeMetric]:
"""Top INTERNAL nodes by PageRank over the internal CALLS graph.

PageRank weights a node by the *transitive* importance of what reaches it,
Expand All @@ -223,11 +265,17 @@ def get_pagerank(self, limit: int = 10) -> list[NodeMetric]:
also includes CLASS nodes β€” a constructor call is a CALLS edge to the
class, so a heavily-instantiated class (e.g. a core data model) is
legitimately central. Uncalled classes simply rest at the floor rank.

``exclude`` removes any node whose FQN contains one of the given
dot-segments from the PageRank universe entirely (so excluded nodes leave
the propagation graph, not just the final ranking).
"""
conn = self.conn
where, params = _segment_exclusion("id", exclude)
conn.execute(
"CREATE OR REPLACE TEMP TABLE pr_nodes AS "
f"SELECT id FROM nodes WHERE {_PAGERANK_INTERNAL}"
f"SELECT id FROM nodes WHERE {_PAGERANK_INTERNAL}{where}",
params,
)
n = int(conn.execute("SELECT COUNT(*) FROM pr_nodes").fetchone()[0])
if n == 0:
Expand Down Expand Up @@ -262,11 +310,19 @@ def get_pagerank(self, limit: int = 10) -> list[NodeMetric]:
]

def architecture_report(
self, bottleneck_limit: int = 10, god_limit: int = 5, critical_limit: int = 10
self,
bottleneck_limit: int = 10,
god_limit: int = 5,
critical_limit: int = 10,
exclude: Sequence[str] = (),
) -> ArchitectureReport:
"""Bundle the coupling bottlenecks, God classes, and PageRank-critical nodes."""
"""Bundle the coupling bottlenecks, God classes, and PageRank-critical nodes.

``exclude`` is threaded into all three sections β€” any node whose FQN
contains one of the given dot-segments (e.g. ``["tests"]``) is dropped.
"""
return ArchitectureReport(
bottlenecks=self.get_coupling_metrics(bottleneck_limit),
god_classes=self.get_god_classes(god_limit),
critical=self.get_pagerank(critical_limit),
bottlenecks=self.get_coupling_metrics(bottleneck_limit, exclude),
god_classes=self.get_god_classes(god_limit, exclude),
critical=self.get_pagerank(critical_limit, exclude),
)
108 changes: 108 additions & 0 deletions tests/unit/test_metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -241,3 +241,111 @@ def test_pagerank_empty_graph_returns_empty(tmp_path: Path) -> None:
db = _write_db(tmp_path, [], [])
with DuckDBAnalyzer(db) as analyzer:
assert analyzer.get_pagerank() == []


def _exclude_fixture(tmp_path: Path) -> str:
"""Graph mixing prod nodes, top-level tests, nested tests, and a 'testservice'
false-positive β€” used by the --exclude segment tests (#234)."""
nodes = [
_node("core.svc.handler"),
_node("tests.utils.rnd"),
_node("domains.resv.tests.test_routes"),
_node("domains.testservice.run"),
_node("core.models.Thing", NodeType.CLASS),
_node("core.models.Thing.m", NodeType.METHOD),
_node("tests.fakes.FakeThing", NodeType.CLASS),
_node("tests.fakes.FakeThing.m", NodeType.METHOD),
]
edges = [
# test helpers call the prod handler (so they have coupling and would rank)
Edge(id="e1", source="tests.utils.rnd", target="core.svc.handler", type=EdgeType.CALLS),
Edge(
id="e2",
source="domains.resv.tests.test_routes",
target="core.svc.handler",
type=EdgeType.CALLS,
),
Edge(
id="e3",
source="core.svc.handler",
target="domains.testservice.run",
type=EdgeType.CALLS,
),
Edge(
id="d1",
source="core.models.Thing",
target="core.models.Thing.m",
type=EdgeType.DECLARES,
),
Edge(
id="d2",
source="tests.fakes.FakeThing",
target="tests.fakes.FakeThing.m",
type=EdgeType.DECLARES,
),
]
return _write_db(tmp_path, nodes, edges)


def test_exclude_segment_drops_tests_from_all_sections(tmp_path: Path) -> None:
"""`exclude=['tests']` removes top-level AND nested test FQNs from every
section, while keeping a 'testservice' that only contains the substring (#234)."""
db = _exclude_fixture(tmp_path)
with DuckDBAnalyzer(db) as analyzer:
report = analyzer.architecture_report(exclude=["tests"])

ids = (
{m.node_id for m in report.bottlenecks}
| {m.node_id for m in report.god_classes}
| {m.node_id for m in report.critical}
)
# no FQN carries a 'tests' dot-segment anywhere
assert not any(".tests." in f".{i}." for i in ids)
assert "tests.utils.rnd" not in ids
assert "domains.resv.tests.test_routes" not in ids
assert "tests.fakes.FakeThing" not in ids
# substring-but-not-a-segment survivor stays, as does the real prod node
assert "domains.testservice.run" in ids
assert "core.svc.handler" in ids


def test_exclude_is_opt_in_default_keeps_tests(tmp_path: Path) -> None:
"""Without exclude, test nodes still appear β€” the flag changed behaviour (#234)."""
db = _exclude_fixture(tmp_path)
with DuckDBAnalyzer(db) as analyzer:
report = analyzer.architecture_report()
bottleneck_ids = {m.node_id for m in report.bottlenecks}
god_ids = {m.node_id for m in report.god_classes}
assert "tests.utils.rnd" in bottleneck_ids
assert "tests.fakes.FakeThing" in god_ids


def test_exclude_segment_escapes_like_metacharacters(tmp_path: Path) -> None:
"""An underscore in a segment is matched literally, not as the LIKE wildcard (#234)."""
nodes = [_node("a.b_c.fn"), _node("a.bxc.fn")]
db = _write_db(tmp_path, nodes, [])
with DuckDBAnalyzer(db) as analyzer:
ids = {m.node_id for m in analyzer.get_coupling_metrics(exclude=["b_c"])}
# 'b_c' must drop only the literal 'b_c' segment, never 'bxc'
assert "a.b_c.fn" not in ids
assert "a.bxc.fn" in ids


def test_exclude_multiple_segments_drops_any_match(tmp_path: Path) -> None:
"""Repeatable exclude is AND-composed: a node is dropped if it matches ANY
given segment; an unrelated node survives (#234, #235 review)."""
nodes = [_node("a.tests.x"), _node("b.vendor.y"), _node("c.core.z")]
db = _write_db(tmp_path, nodes, [])
with DuckDBAnalyzer(db) as analyzer:
ids = {m.node_id for m in analyzer.get_coupling_metrics(exclude=["tests", "vendor"])}
assert ids == {"c.core.z"}


def test_exclude_empty_or_whitespace_segment_is_noop(tmp_path: Path) -> None:
"""Empty / whitespace-only segments are skipped, never excluding everything
(#235 review β€” gemini/zaebee robustness nit)."""
nodes = [_node("a.core.x"), _node("b.svc.y")]
db = _write_db(tmp_path, nodes, [])
with DuckDBAnalyzer(db) as analyzer:
ids = {m.node_id for m in analyzer.get_coupling_metrics(exclude=["", " "])}
assert ids == {"a.core.x", "b.svc.y"}
Loading