From d927404fb5e8aee6772ed563e075ac1ec5970839 Mon Sep 17 00:00:00 2001 From: citarf <8773429+citarf@users.noreply.github.com> Date: Wed, 24 Jun 2026 22:52:21 +0200 Subject: [PATCH 1/2] graph: add shortest_path and dead_symbols queries + CLI subcommands shortest_path: BFS over the call graph between two resolved symbols, shortest path with min-edge-confidence tie-break, falls back to the reverse direction when no src->dst path exists within max_depth. dead_symbols: lists fan_in==0 functions/methods (excluding classes and module nodes), filterable by repo/subsystem, with an entrypoint caveat. Both wired into the argparse CLI as 'path' and 'dead' subcommands. --- tools/code_index/graph.py | 170 +++++++++++++++++++++++++++ tools/tests/test_code_index_graph.py | 124 +++++++++++++++++++ 2 files changed, 294 insertions(+) diff --git a/tools/code_index/graph.py b/tools/code_index/graph.py index d71a469..dbf5107 100644 --- a/tools/code_index/graph.py +++ b/tools/code_index/graph.py @@ -722,6 +722,123 @@ def code_hotspots(graph: dict, *, top: int = 20, repo: str | None = None, return {"by": metric, "repo": repo, "subsystem": subsystem, "lang": lang, "hotspots": hot} +# ── Requête : shortest_path ───────────────────────────────────────────────────── + +_DEAD_EXCLUDE_KINDS = _CLASS_KINDS | {"module"} + + +def _bfs_path(out: dict, srcs: set[str], dsts: set[str], max_depth: int + ) -> tuple[list[str], float] | None: + """Plus court chemin (en nb d'arêtes) d'une graine `srcs` vers une graine `dsts` dans + `out`. Départage : à longueur égale, on garde le chemin dont la confiance MINIMALE le long + du chemin est la plus haute. Renvoie (liste de nids, min_confidence) ou None. + + BFS par couches : on explore couche par couche (toutes à la même distance) ; dès qu'une + couche contient une cible on s'arrête (longueur minimale garantie). Pour chaque nœud on + mémorise le meilleur (min-conf) prédécesseur à cette distance.""" + if srcs & dsts: # src == dst (ou recouvrement) : chemin trivial + nid = next(iter(srcs & dsts)) + return [nid], 1.0 + # best[nid] = (min_conf du meilleur chemin jusqu'ici, prédécesseur) à la distance courante. + best: dict[str, tuple[float, str | None]] = {s: (1.0, None) for s in srcs} + frontier = set(srcs) + depth = 0 + while frontier and depth < max_depth: + depth += 1 + nxt: dict[str, tuple[float, str | None]] = {} + for nid in frontier: + pconf = best[nid][0] + for tgt, c in out.get(nid, []): + if tgt in best: # déjà atteint plus tôt (distance ≤) → ne pas régresser + continue + path_conf = min(pconf, c) + if tgt not in nxt or path_conf > nxt[tgt][0]: + nxt[tgt] = (path_conf, nid) + if not nxt: + break + # cibles atteintes à cette couche : on prend celle de meilleure confiance minimale. + reached = [t for t in nxt if t in dsts] + if reached: + best.update(nxt) + tgt = max(reached, key=lambda t: nxt[t][0]) + path = [tgt] + while best[path[-1]][1] is not None: + path.append(best[path[-1]][1]) + path.reverse() + return path, best[tgt][0] + best.update(nxt) + frontier = set(nxt) + return None + + +def shortest_path(graph: dict, src: str, dst: str, *, max_depth: int = 8, + sidecar: dict | None = None) -> dict: + """Plus court chemin d'appel entre deux symboles dans le graphe de code. + + BFS sur `graph["out"]` depuis n'importe quelle racine de `src` vers n'importe quelle racine + de `dst`. Départage à longueur égale : confiance MINIMALE la plus haute (chemin le plus + sûr). Si aucun chemin src→dst sous `max_depth`, on tente dst→src et on le signale via + `direction`. Chaque nœud du chemin porte la confiance de son arête entrante.""" + nodes = graph.get("nodes", {}) + out = graph.get("out", {}) + src_roots = resolve_symbol(graph, src) + dst_roots = resolve_symbol(graph, dst) + base = {"found": False, "path": [], "min_confidence": 0.0, + "src_roots": src_roots, "dst_roots": dst_roots, "direction": None} + if not src_roots or not dst_roots: + return base + + def _render(path: list[str], min_conf: float, direction: str) -> dict: + # confiance de l'arête ENTRANTE de chaque nœud (1.0 pour la racine) ; en sens inverse + # le chemin reste exprimé src→…→dst pour la lisibilité, mais les arêtes sont celles + # parcourues (dst→src dans `out`). + confs = [1.0] + for i in range(1, len(path)): + edge = dict(out.get(path[i - 1], [])) + confs.append(edge.get(path[i], 0.0)) + rendered = [_describe(nodes[n], i, confs[i], sidecar) + for i, n in enumerate(path) if n in nodes] + return {"found": True, "path": rendered, "min_confidence": round(min_conf, 2), + "src_roots": src_roots, "dst_roots": dst_roots, "direction": direction} + + fwd = _bfs_path(out, set(src_roots), set(dst_roots), max_depth) + if fwd is not None: + return _render(fwd[0], fwd[1], "src->dst") + rev = _bfs_path(out, set(dst_roots), set(src_roots), max_depth) + if rev is not None: + return _render(rev[0], rev[1], "dst->src") + return base + + +# ── Requête : dead_symbols ────────────────────────────────────────────────────── + +_DEAD_CAVEAT = ("attention : les points d'entrée (routes, mains de cron, hooks, callbacks de " + "framework) ont légitimement 0 appelant intra-projet sous résolution statique " + "et ne doivent pas être supprimés à l'aveugle.") + + +def dead_symbols(graph: dict, *, repo: str | None = None, subsystem: str | None = None, + top: int = 30, sidecar: dict | None = None) -> dict: + """Symboles à fan-in nul (« code potentiellement mort ») : fonctions/méthodes jamais + appelées dans le graphe. Exclut classes/interfaces/traits/enums et nœuds module (un type + sans appelant n'est pas du code mort au même titre qu'une fonction). Filtrable par + repo/sous-système. Tri repo→path→ligne, plafonné à `top`. Voir `caveat` : les points + d'entrée ont légitimement 0 appelant.""" + items = [n for n in graph.get("nodes", {}).values() + if n.get("fan_in", 0) == 0 + and n["kind"] not in _DEAD_EXCLUDE_KINDS + and n["qname"] != MODULE_QNAME] + if repo: + items = [n for n in items if n["repo"] == repo] + if subsystem: + items = [n for n in items if _subsystem(n) == subsystem] + items.sort(key=lambda n: (n["repo"], n["path"], n["start_line"])) + truncated = len(items) > top + symbols = [_describe(n, 0, 1.0, sidecar) for n in items[:max(0, top)]] + return {"count": len(symbols), "truncated": truncated, "symbols": symbols, + "caveat": _DEAD_CAVEAT} + + # ── CLI ───────────────────────────────────────────────────────────────────────── def _cmd_build(args: argparse.Namespace) -> int: @@ -772,6 +889,43 @@ def _cmd_impact(args: argparse.Namespace) -> int: return 0 +def _cmd_path(args: argparse.Namespace) -> int: + graph = load_graph(args.graph) + sidecar = json.loads(Path(args.meta).read_text(encoding="utf-8")) if args.meta else None + res = shortest_path(graph, args.src, args.dst, max_depth=args.max_depth, sidecar=sidecar) + if not res["src_roots"] or not res["dst_roots"]: + miss = args.src if not res["src_roots"] else args.dst + print(f"Symbole introuvable : {miss}", file=sys.stderr) + return 1 + if not res["found"]: + print(f"# Aucun chemin d'appel entre {args.src} et {args.dst} (prof ≤ {args.max_depth}).") + return 1 + arrow = "→" if res["direction"] == "src->dst" else "← (sens inverse)" + print(f"# {args.src} {arrow} {args.dst} — {len(res['path'])} nœuds, " + f"confiance min {res['min_confidence']}") + for n in res["path"]: + url = f" {n['source_url']}" if n["source_url"] else "" + print(f" [{n['tier']}] {n['qname']} ({n['kind']}) " + f"{n['repo']}/{n['path']}:{n['start_line']}{url}") + return 0 + + +def _cmd_dead(args: argparse.Namespace) -> int: + graph = load_graph(args.graph) + sidecar = json.loads(Path(args.meta).read_text(encoding="utf-8")) if args.meta else None + res = dead_symbols(graph, repo=args.repo, subsystem=args.subsystem, top=args.top, + sidecar=sidecar) + print(f"# Symboles à fan-in nul (code potentiellement mort) — {res['count']} affichés" + + (f", repo={args.repo}" if args.repo else "") + + (f", {args.subsystem}" if args.subsystem else "") + + (" — tronqué" if res["truncated"] else "")) + print(f" ⚠ {res['caveat']}") + for n in res["symbols"]: + url = f" {n['source_url']}" if n["source_url"] else "" + print(f" {n['qname']} ({n['kind']}) {n['repo']}/{n['path']}:{n['start_line']}{url}") + return 0 + + def main(argv: list[str]) -> int: ap = argparse.ArgumentParser(description="Graphe d'appels du code Infoclimat (tree-sitter).") sub = ap.add_subparsers(dest="cmd", required=True) @@ -800,6 +954,22 @@ def main(argv: list[str]) -> int: h.add_argument("--meta", default=None) h.set_defaults(func=_cmd_hotspots) + p = sub.add_parser("path", help="Plus court chemin d'appel entre deux symboles.") + p.add_argument("src", help="Symbole de départ (Classe.methode).") + p.add_argument("dst", help="Symbole d'arrivée (Classe.methode).") + p.add_argument("--graph", required=True) + p.add_argument("--max-depth", type=int, default=8) + p.add_argument("--meta", default=None) + p.set_defaults(func=_cmd_path) + + d = sub.add_parser("dead", help="Symboles à fan-in nul (code potentiellement mort).") + d.add_argument("--graph", required=True) + d.add_argument("--top", type=int, default=30) + d.add_argument("--repo", default=None) + d.add_argument("--subsystem", default=None) + d.add_argument("--meta", default=None) + d.set_defaults(func=_cmd_dead) + args = ap.parse_args(argv) return args.func(args) diff --git a/tools/tests/test_code_index_graph.py b/tools/tests/test_code_index_graph.py index f85ad55..289905d 100644 --- a/tools/tests/test_code_index_graph.py +++ b/tools/tests/test_code_index_graph.py @@ -259,3 +259,127 @@ def test_load_graph_roundtrip_and_gzip(tmp_path) -> None: pgz = tmp_path / "graph.json.gz" pgz.write_bytes(gzip.compress(json.dumps(g).encode("utf-8"))) assert graph.load_graph(pgz)["nodes"] == g["nodes"] + + +# ── shortest_path ─────────────────────────────────────────────────────────────── + +def test_shortest_path_src_to_dst() -> None: + pytest.importorskip("tree_sitter_language_pack") + g = _chain_graph() # a → b → c (out) + res = graph.shortest_path(g, "a", "c") + assert res["found"] is True + assert res["direction"] == "src->dst" + assert [n["qname"] for n in res["path"]] == ["a", "b", "c"] + # min_confidence = min des confiances des arêtes du chemin (b→c, a→b sont C_UNIQUE/SAMENS) + assert res["min_confidence"] == pytest.approx(min(n["confidence"] for n in res["path"][1:])) + assert res["src_roots"] and res["dst_roots"] + + +def test_shortest_path_falls_back_to_reverse_direction() -> None: + pytest.importorskip("tree_sitter_language_pack") + g = _chain_graph() # a → b → c (out) : pas de chemin c→a, mais a→c existe + res = graph.shortest_path(g, "c", "a") + assert res["found"] is True + assert res["direction"] == "dst->src" # trouvé en sens inverse (a→…→c) + assert [n["qname"] for n in res["path"]] == ["a", "b", "c"] + + +def test_shortest_path_same_symbol() -> None: + pytest.importorskip("tree_sitter_language_pack") + g = _chain_graph() + res = graph.shortest_path(g, "b", "b") + assert res["found"] is True + assert [n["qname"] for n in res["path"]] == ["b"] + assert res["min_confidence"] == 1.0 + + +def test_shortest_path_unresolved_src() -> None: + pytest.importorskip("tree_sitter_language_pack") + g = _chain_graph() + res = graph.shortest_path(g, "inexistant", "c") + assert res["found"] is False + assert res["src_roots"] == [] and res["direction"] is None + + +def test_shortest_path_no_path_within_depth() -> None: + pytest.importorskip("tree_sitter_language_pack") + # deux composantes disjointes : x→y et z (isolé) → aucun chemin + files = [ + (_sf("r", "p.py", "python"), + "def y():\n return 1\n\ndef x():\n return y()\n"), + (_sf("r", "q.py", "python"), "def z():\n return 0\n"), + ] + g = graph.build_graph(files) + res = graph.shortest_path(g, "x", "z") + assert res["found"] is False + + +def test_shortest_path_tiebreak_prefers_higher_min_confidence() -> None: + pytest.importorskip("tree_sitter_language_pack") + # a atteint d par deux chemins de même longueur ; on construit le graphe à la main pour + # contrôler les confiances. a→b (0.9)→d (0.9) vs a→c (0.4)→d (0.9). + nodes = {} + for q, ln in (("a", 1), ("b", 2), ("c", 3), ("d", 4)): + nid = f"r/m.py:{ln}:{q}" + nodes[nid] = {"name": q, "qname": q, "repo": "r", "path": "m.py", "lang": "python", + "kind": "function", "start_line": ln, "end_line": ln, + "fan_in": 0, "centrality": 0.0} + out = { + "r/m.py:1:a": [["r/m.py:2:b", 0.9], ["r/m.py:3:c", 0.4]], + "r/m.py:2:b": [["r/m.py:4:d", 0.9]], + "r/m.py:3:c": [["r/m.py:4:d", 0.9]], + } + g = {"version": "graph-v2", "nodes": nodes, "by_name": {q: [nid] for nid, q in + ((nid, n["qname"]) for nid, n in nodes.items())}, "by_fqn": {}, "out": out} + res = graph.shortest_path(g, "a", "d") + assert res["found"] is True + assert [n["qname"] for n in res["path"]] == ["a", "b", "d"] # min-conf 0.9 préféré à 0.4 + assert res["min_confidence"] == pytest.approx(0.9) + + +# ── dead_symbols ──────────────────────────────────────────────────────────────── + +def test_dead_symbols_lists_zero_fanin_functions() -> None: + pytest.importorskip("tree_sitter_language_pack") + g = _chain_graph() # a → b → c : seul 'a' a fan_in 0 (et n'est ni classe ni module) + res = graph.dead_symbols(g) + qn = {s["qname"] for s in res["symbols"]} + assert "a" in qn + assert "b" not in qn and "c" not in qn # b et c ont des appelants + assert "" not in qn + assert res["count"] == len(res["symbols"]) + assert "caveat" in res and "point" in res["caveat"].lower() # mentionne les points d'entrée + + +def test_dead_symbols_excludes_classes() -> None: + pytest.importorskip("tree_sitter_language_pack") + files = [(_sf("r", "lib.php", "php"), + " None: + pytest.importorskip("tree_sitter_language_pack") + files = [ + (_sf("r1", "svc/a.py", "python"), "def lonely():\n return 1\n"), + (_sf("r2", "svc/b.py", "python"), "def other():\n return 2\n"), + ] + g = graph.build_graph(files) + res = graph.dead_symbols(g, repo="r1") + assert {s["repo"] for s in res["symbols"]} == {"r1"} + res2 = graph.dead_symbols(g, subsystem="r2/svc") + assert {s["subsystem"] for s in res2["symbols"]} == {"r2/svc"} + + +def test_dead_symbols_top_truncation() -> None: + pytest.importorskip("tree_sitter_language_pack") + src = "".join(f"def f{i}():\n return {i}\n\n" for i in range(5)) + g = graph.build_graph([(_sf("r", "many.py", "python"), src)]) + res = graph.dead_symbols(g, top=2) + assert res["count"] == 2 + assert res["truncated"] is True From f41c3cc9e4c8952f76346670a683b8e03619a62d Mon Sep 17 00:00:00 2001 From: citarf <8773429+citarf@users.noreply.github.com> Date: Wed, 24 Jun 2026 22:52:28 +0200 Subject: [PATCH 2/2] search: expose status/source/since as pre-rerank filters Extend _where() to optionally constrain status (single value or list), source (github/gitlab/other) and since (YYYY-MM-DD lower bound on last_commit). Thread these as keyword-only args through search_code and search_docs, applied before reranking; repos/lang unchanged. Also surface them as --status/--source/--since CLI flags. --- tools/code_index/search.py | 64 +++++++++++++++++++++------ tools/tests/test_code_index_search.py | 36 +++++++++++++++ 2 files changed, 86 insertions(+), 14 deletions(-) diff --git a/tools/code_index/search.py b/tools/code_index/search.py index 3fb71ca..5bcfa4e 100644 --- a/tools/code_index/search.py +++ b/tools/code_index/search.py @@ -145,13 +145,33 @@ def _llm_rerank(client, query: str, results: list[Result], k: int, cfg: Config, time.sleep(min(2 ** attempt, 30)) -def _where(repos: list[str] | None, lang: str | None) -> str | None: +def _q(value: str) -> str: + """Littéral SQL échappé (apostrophes doublées).""" + return "'" + value.replace("'", "''") + "'" + + +def _where(repos: list[str] | None, lang: str | None, *, + status: str | list[str] | None = None, source: str | None = None, + since: str | None = None) -> str | None: + """Clause WHERE de filtrage avant rerank. `status` accepte une valeur ou une liste + (actif/douteux/mort) ; `source` filtre l'origine (github/gitlab/other) ; `since` est une + borne basse YYYY-MM-DD sur `last_commit` (on garde les lignes ≥ `since`).""" clauses = [] if repos: - joined = ", ".join("'" + r.replace("'", "''") + "'" for r in repos) + joined = ", ".join(_q(r) for r in repos) clauses.append(f"repo IN ({joined})") if lang: - clauses.append("lang = '" + lang.replace("'", "''") + "'") + clauses.append("lang = " + _q(lang)) + if status: + if isinstance(status, str): + clauses.append("status = " + _q(status)) + else: + joined = ", ".join(_q(s) for s in status) + clauses.append(f"status IN ({joined})") + if source: + clauses.append("source = " + _q(source)) + if since: + clauses.append("last_commit >= " + _q(since)) return " AND ".join(clauses) if clauses else None @@ -183,17 +203,26 @@ def _run_search(cfg: Config, question: str, k: int, where: str | None) -> list[R def search_code(question: str, k: int = 8, repos: list[str] | None = None, - lang: str | None = None, config: Config | None = None) -> list[Result]: - """Top-`k` chunks de CODE les plus pertinents, filtrables par repo/langage. Le `text` - renvoyé reste le chunk **brut**. Signature stable (réutilisée par le wrapper MCP).""" - return _run_search(config or load_config("code"), question, k, _where(repos, lang)) - - -def search_docs(question: str, k: int = 6, config: Config | None = None) -> list[Result]: + lang: str | None = None, config: Config | None = None, *, + status: str | list[str] | None = None, source: str | None = None, + since: str | None = None) -> list[Result]: + """Top-`k` chunks de CODE les plus pertinents, filtrables par repo/langage et — appliqués + AVANT le rerank — par `status` (actif/douteux/mort), `source` (github/gitlab/other) et + `since` (borne basse YYYY-MM-DD sur le dernier commit). Le `text` renvoyé reste le chunk + **brut**. Signature stable (réutilisée par le wrapper MCP) : `repos`/`lang` inchangés.""" + return _run_search(config or load_config("code"), question, k, + _where(repos, lang, status=status, source=source, since=since)) + + +def search_docs(question: str, k: int = 6, config: Config | None = None, *, + status: str | list[str] | None = None, source: str | None = None, + since: str | None = None) -> list[Result]: """Top-`k` entrées de GOUVERNANCE (contrats, inventory, catalog, glossaire) les plus pertinentes pour `question` — recherche sémantique sur la table `docs_chunks` - (embeddings mistral-embed). Complément des outils lexicaux grep/lineage.""" - return _run_search(config or load_config("docs"), question, k, None) + (embeddings mistral-embed). Complément des outils lexicaux grep/lineage. Filtres + `status`/`source`/`since` appliqués avant le rerank (cf. `search_code`).""" + return _run_search(config or load_config("docs"), question, k, + _where(None, None, status=status, source=source, since=since)) def main(argv: list[str]) -> int: @@ -202,14 +231,21 @@ def main(argv: list[str]) -> int: ap.add_argument("--k", type=int, default=8) ap.add_argument("--repo", action="append", dest="repos", default=None) ap.add_argument("--lang", default=None) + ap.add_argument("--status", action="append", dest="status", default=None, + help="Filtre statut (actif/douteux/mort) ; répétable.") + ap.add_argument("--source", default=None, help="Origine du repo (github/gitlab/other).") + ap.add_argument("--since", default=None, help="Dernier commit ≥ YYYY-MM-DD.") ap.add_argument("--corpus", choices=["code", "docs"], default="code") ap.add_argument("--full", action="store_true", help="Afficher le chunk entier.") args = ap.parse_args(argv) + status = args.status[0] if args.status and len(args.status) == 1 else args.status if args.corpus == "docs": - results = search_docs(args.question, k=args.k) + results = search_docs(args.question, k=args.k, status=status, source=args.source, + since=args.since) else: - results = search_code(args.question, k=args.k, repos=args.repos, lang=args.lang) + results = search_code(args.question, k=args.k, repos=args.repos, lang=args.lang, + status=status, source=args.source, since=args.since) if not results: print("Aucun résultat (index vide ?). Lancer d'abord : python -m code_index.index", file=sys.stderr) diff --git a/tools/tests/test_code_index_search.py b/tools/tests/test_code_index_search.py index 036a9b3..f143ca3 100644 --- a/tools/tests/test_code_index_search.py +++ b/tools/tests/test_code_index_search.py @@ -52,6 +52,25 @@ def test_where_escapes_quotes() -> None: assert _where(["o'brien"], None) == "repo IN ('o''brien')" +def test_where_status_single_and_list() -> None: + assert _where(None, None, status="actif") == "status = 'actif'" + assert _where(None, None, status=["actif", "douteux"]) == "status IN ('actif', 'douteux')" + + +def test_where_source_filter() -> None: + assert _where(None, None, source="github") == "source = 'github'" + + +def test_where_since_lower_bound() -> None: + assert _where(None, None, since="2026-01-01") == "last_commit >= '2026-01-01'" + + +def test_where_combines_all_new_filters() -> None: + clause = _where(["a"], "php", status="actif", source="github", since="2026-01-01") + assert clause == ("repo IN ('a') AND lang = 'php' AND status = 'actif' " + "AND source = 'github' AND last_commit >= '2026-01-01'") + + def _cfg() -> Config: # Pipeline simple (vecteur seul, sans réécriture/rerank) pour tester le mapping isolément ; # les modes contextuel/hybride ont leurs propres tests. @@ -89,6 +108,23 @@ def _fake_search(db, qvec, *, k, where, **kw): # noqa: ANN001 assert r.score == pytest.approx(0.12) +def test_search_code_threads_status_source_since(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(search_mod.embed, "make_client", lambda key: object()) + monkeypatch.setattr(search_mod.embed, "embed_query", lambda *a, **k: [0.1, 0.2]) + monkeypatch.setattr(search_mod.store, "connect", lambda d: object()) + captured = {} + + def _fake_search(db, qvec, *, k, where, **kw): # noqa: ANN001 + captured["where"] = where + return [] + + monkeypatch.setattr(search_mod.store, "search", _fake_search) + search_code("x", k=3, repos=["r"], lang="php", status="actif", source="github", + since="2026-01-01", config=_cfg()) + assert captured["where"] == ("repo IN ('r') AND lang = 'php' AND status = 'actif' " + "AND source = 'github' AND last_commit >= '2026-01-01'") + + def test_search_code_requires_api_key() -> None: cfg = _cfg().__class__(**{**_cfg().__dict__, "api_key": None}) with pytest.raises(RuntimeError):