diff --git a/src/cgis/api/mcp_server.py b/src/cgis/api/mcp_server.py index 033f9a2e..d3fd972a 100644 --- a/src/cgis/api/mcp_server.py +++ b/src/cgis/api/mcp_server.py @@ -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}" diff --git a/src/cgis/cli.py b/src/cgis/cli.py index 3e520e0d..dc0b80f8 100644 --- a/src/cgis/cli.py +++ b/src/cgis/cli.py @@ -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 ), @@ -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]") @@ -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]") diff --git a/src/cgis/query/metrics.py b/src/cgis/query/metrics.py index 19d7ad88..dc224f02 100644 --- a/src/cgis/query/metrics.py +++ b/src/cgis/query/metrics.py @@ -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 @@ -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): + 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 @@ -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 @@ -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), @@ -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, @@ -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: @@ -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), ) diff --git a/tests/unit/test_metrics.py b/tests/unit/test_metrics.py index b9ce8058..7bc3da54 100644 --- a/tests/unit/test_metrics.py +++ b/tests/unit/test_metrics.py @@ -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"}