diff --git a/graphify/serve.py b/graphify/serve.py index 33db0108f..f7218d76b 100644 --- a/graphify/serve.py +++ b/graphify/serve.py @@ -6,6 +6,7 @@ import sys from array import array from pathlib import Path +from typing import NamedTuple import networkx as nx from networkx.readwrite import json_graph from graphify.security import sanitize_label, check_graph_file_size_cap @@ -283,8 +284,63 @@ def _trigram_candidates(G: nx.Graph, needles: list[str], *, guard_frac: float = return [ids[i] for i in sorted(cand)] +class _QueryScores(NamedTuple): + """Per-query scoring result, returned by the private `_score_query` helper. + + `ranked` is the existing ordered `(score, node_id)` ranking produced by the + combined query scorer (the value `_score_nodes` always returned). When the + caller asks for it via `collect_per_term_seeds=True`, `best_seed_by_term` + additionally carries the winning node id for each normalized search token — + the seed `_pick_seeds` would have picked for that token via the now-retired + per-token `_score_nodes([token])` rescoring pass — computed in the *same* + per-node traversal so the query path makes exactly one graph scoring pass + regardless of query length. Empty when `collect_per_term_seeds=False`. + """ + ranked: list[tuple[float, str]] + best_seed_by_term: dict[str, str] + + def _score_nodes(G: nx.Graph, terms: list[str]) -> list[tuple[float, str]]: - scored = [] + """Combined query scorer returning the existing ranked `(score, node_id)` list. + + Backwards-compatible thin wrapper around `_score_query` for path, explain, + tests, and every other caller that only needs the combined ranking. The + per-term seed metadata computed by `_score_query` (when requested) is + discarded here so existing callers see no API or runtime-cost change. + """ + return _score_query(G, terms, collect_per_term_seeds=False).ranked + + +def _score_query( + G: nx.Graph, terms: list[str], *, collect_per_term_seeds: bool +) -> _QueryScores: + """Single-pass combined scorer that optionally also records the best seed + for each normalized query token. + + The combined ranking is byte-identical to what `_score_nodes` produced + before the refactor; `_score_nodes` is now a thin wrapper that asks for + `collect_per_term_seeds=False` and returns only `.ranked`. + + When `collect_per_term_seeds=True`, the per-token singleton winner is + computed alongside the combined score in the *same* per-node visit (it + reuses the same `norm_label` / `label_tokens` / `source` already evaluated + for the combined tier), so `_query_graph_text` can feed `best_seed_by_term` + straight into `_pick_seeds` and skip the T additional whole-graph rescoring + passes the old per-token `_score_nodes([token])` loop ran. + + Singleton-winner semantics match the legacy per-token path exactly. The + score itself mirrors `_score_nodes([token])` with `n_terms == 1` (so the + coverage term is 1 and the per-token tier is unscaled) plus the broader + joined-singlet tier (which also checks `label_tokens` and `nid_lower`). + Tie-break order is (1) highest singleton score, (2) highest graph degree, + (3) shortest displayed label, (4) lexicographically smallest node id — + exactly what `max(tied, key=degree)` over a sort by `(-score, label_len, + nid)` produced in the legacy `_pick_seeds` per-token loop. The combined + trigram candidate set (needles `norm_terms + [joined]`) is a superset of + each per-token `[t]` candidate set, so iterating combined candidates + discovers every non-zero singleton-score node for every term. + """ + scored: list[tuple[float, str]] = [] # Dedupe tokens, order-preserving (as _pick_seeds already does): a repeated # query word must not double-count every tier, and with coverage scaling # below it would also inflate the matched-term ratio (#1602). @@ -305,6 +361,16 @@ def _score_nodes(G: nx.Graph, terms: list[str]) -> list[tuple[float, str]]: G.nodes(data=True) if candidate_ids is None else ((nid, G.nodes[nid]) for nid in candidate_ids) ) + # Per-token best tracking, only when the caller (the query path) wants the + # seed metadata. The key tuple is the full multi-key tie-break + # (`(-singleton_score, -degree, label_len, nid)`), so `min` over the + # stored key mirrors the legacy `max(tied, key=degree)` over a + # (-score, label_len, nid)-sorted term_scored list. `None` is comparable + # as "smaller" than every tuple, so the first non-zero candidate seeds the + # entry without a separate `if t not in best_by_term` branch. + best_by_term: dict[str, tuple[tuple, str]] | None = ( + {} if collect_per_term_seeds else None + ) for nid, data in node_iter: norm_label = data.get("norm_label") or _strip_diacritics(data.get("label") or "").lower() bare_label = norm_label.rstrip("()") @@ -315,6 +381,11 @@ def _score_nodes(G: nx.Graph, terms: list[str]) -> list[tuple[float, str]]: # driver". label_tokens = " ".join(_search_tokens(data.get("label") or "")) source = (data.get("source_file") or "").lower() + # `nid_lower` is needed both by the full-query tier (`if joined`) and by + # the per-token singleton tier (joined-singlet exact-match check). When + # neither runs (`joined` empty AND not collecting seeds) skip the call; + # this preserves the single-query-time perf where nid_lower was lazy. + nid_lower = nid.lower() if (joined or collect_per_term_seeds) else "" score = 0.0 # Full-query tier: a multi-word query that equals (or prefixes) the whole # label must dominate the per-token bag-of-words sums below, so `path`/ @@ -323,7 +394,6 @@ def _score_nodes(G: nx.Graph, terms: list[str]) -> list[tuple[float, str]]: # tier never fires, and every node sharing the token set ties -> arbitrary # node-id sort -> wrong/disconnected endpoint -> false "No path found". if joined: - nid_lower = nid.lower() if joined in (norm_label, bare_label, label_tokens, nid_lower): score += _EXACT_MATCH_BONUS * 10 * joined_w elif ( @@ -349,19 +419,55 @@ def _score_nodes(G: nx.Graph, terms: list[str]) -> list[tuple[float, str]]: tiered = 0.0 for t in norm_terms: w = idf.get(t, 1.0) - # Three-tier precedence: exact > prefix > substring (take the - # strongest tier per term so a single term cannot double-count). + # Per-tier contributions for this token, kept separate so the + # singleton tracking below can reuse them without re-evaluating + # the same predicates. Three-tier precedence: exact > prefix > + # substring (take the strongest tier per term so a single term + # cannot double-count). + tier_value = 0.0 + substr_value = 0.0 + source_value = 0.0 if t == norm_label or t == bare_label: - tiered += _EXACT_MATCH_BONUS * w + tier_value = _EXACT_MATCH_BONUS * w matched += 1 elif norm_label.startswith(t) or bare_label.startswith(t): - tiered += _PREFIX_MATCH_BONUS * w + tier_value = _PREFIX_MATCH_BONUS * w matched += 1 elif t in norm_label: - score += _SUBSTRING_MATCH_BONUS * w + substr_value = _SUBSTRING_MATCH_BONUS * w + score += substr_value matched += 1 if t in source: - score += _SOURCE_MATCH_BONUS * w + source_value = _SOURCE_MATCH_BONUS * w + score += source_value + tiered += tier_value + if collect_per_term_seeds and best_by_term is not None: + # Singleton score for [t] on this node, mirroring + # `_score_nodes(G, [t])` exactly (n_terms == 1, no coverage + # scaling). The joined-singlet tier is broader than the per- + # token tier: it also checks `label_tokens` and `nid_lower`, + # matching the legacy single-token `_score_nodes([t])` call + # (where `joined == t`). + if t in (norm_label, bare_label, label_tokens, nid_lower): + singleton = _EXACT_MATCH_BONUS * 10 * w + elif ( + norm_label.startswith(t) + or bare_label.startswith(t) + or label_tokens.startswith(t) + ): + singleton = _PREFIX_MATCH_BONUS * 10 * w + else: + singleton = 0.0 + singleton += tier_value + substr_value + source_value + if singleton > 0: + # Tie-break key mirrors the legacy sort+max(degree): + # (-singleton, -degree, label_len, nid) — the minimum + # tuple wins, exactly matching max(tied, key=degree) + # over (label_len asc, nid asc)-sorted ties. + key = (-singleton, -G.degree(nid), len(data.get("label") or nid), nid) + cur = best_by_term.get(t) + if cur is None or key < cur[0]: + best_by_term[t] = (key, nid) if tiered: score += tiered * (matched / n_terms) ** 2 if score > 0: @@ -369,7 +475,10 @@ def _score_nodes(G: nx.Graph, terms: list[str]) -> list[tuple[float, str]]: # Sort by score desc; break ties toward the shorter label so a concise exact # match beats a longer superset that happens to share the same score. scored.sort(key=lambda s: (-s[0], len(G.nodes[s[1]].get("label") or s[1]), s[1])) - return scored + best_seed_by_term: dict[str, str] = {} + if collect_per_term_seeds and best_by_term: + best_seed_by_term = {t: nid for t, (_key, nid) in best_by_term.items()} + return _QueryScores(ranked=scored, best_seed_by_term=best_seed_by_term) def _pick_scored_endpoint(G: nx.Graph, scored: list[tuple[float, str]], query: str) -> str: @@ -402,7 +511,7 @@ def _pick_seeds( gap_ratio: float = 0.2, *, G: "nx.Graph | None" = None, - terms: list[str] | None = None, + best_seed_by_term: dict[str, str] | None = None, ) -> list[str]: """Select BFS seed nodes, stopping when score drops too far below the top. @@ -421,11 +530,12 @@ def _pick_seeds( seeds, so the BFS traversal only ever explores the neighborhood of the one unrelated exact match — see #1445. - When `G` and `terms` are supplied, this guarantees at least one seed per - distinct query term that has any match at all, so one term's incidental - collision cannot starve out the others. Ties within a term are broken by - graph degree (structural centrality), so an isolated incidental match - doesn't out-rank a real, well-connected hub for that term. + When `G` and `best_seed_by_term` are supplied, this guarantees at least one + seed per distinct query term that has any match at all, so one term's + incidental collision cannot starve out the others. The per-token winners + in `best_seed_by_term` are precomputed by `_score_query` (during the same + traversal that produced `scored`) so this function no longer rescores the + graph per term — see #1445 and the `_score_query` docstring. Coverage scaling in _score_nodes (#1602) now dampens a lone collision's exact tier on multi-term queries, which brings label-matching relevant @@ -464,15 +574,20 @@ def _seed_label_key(nid: str) -> str: seen_labels.add(key) seeds.append(nid) - if G is not None and terms: - norm_terms = sorted({tok for t in terms for tok in _search_tokens(t)}) - for term in norm_terms: - term_scored = _score_nodes(G, [term]) - if not term_scored: - continue - best_score = term_scored[0][0] - tied = [nid for s, nid in term_scored if s == best_score] - best_nid = max(tied, key=lambda n: G.degree(n)) if len(tied) > 1 else term_scored[0][1] + if G is not None and best_seed_by_term: + # Guarantee one seed per distinct query term that has any match at all, + # so an incidental exact match on one term cannot starve matches on + # other terms (#1445). Iterate tokens in a deterministic sorted order + # so seeds added by this loop have a stable order independent of dict + # iteration — preserving the legacy `_pick_seeds(terms=...)` behavior + # which iterated `sorted({tok ...})`. Per-token winners arrive + # precomputed in `best_seed_by_term` from `_score_query`'s single + # traversal, so `_pick_seeds` no longer rescoring the graph per term. + # The per-label dedup cap also gates these additions, so the guarantee + # cannot reintroduce a second copy of an already-seeded generic label + # (#1766). + for term in sorted(best_seed_by_term): + best_nid = best_seed_by_term[term] # Honor the same per-label cap so the per-term guarantee can't # reintroduce a second copy of an already-seeded generic label. key = _seed_label_key(best_nid) @@ -716,8 +831,14 @@ def _query_graph_text( context_filters: list[str] | None = None, ) -> str: terms = _query_terms(question) - scored = _score_nodes(G, terms) - start_nodes = _pick_seeds(scored, G=G, terms=terms) + # One graph scoring pass produces both the combined ranking (used to drive + # the gap-based seed selection below) and the per-token singleton winners + # (used by _pick_seeds' per-term guarantee). Previously this was T+1 passes + # — one combined + one per query token — re-walking the whole graph each + # time; on a 100k-node, three-term benchmark ~71% of scoring time was + # spent in those redundant per-term passes. + qs = _score_query(G, terms, collect_per_term_seeds=True) + start_nodes = _pick_seeds(qs.ranked, G=G, best_seed_by_term=qs.best_seed_by_term) if not start_nodes: return "No matching nodes found." resolved_filters, filter_source = _resolve_context_filters(question, context_filters) diff --git a/tests/bench_query_scoring.py b/tests/bench_query_scoring.py new file mode 100644 index 000000000..375c9449b --- /dev/null +++ b/tests/bench_query_scoring.py @@ -0,0 +1,280 @@ +#!/usr/bin/env python3 +"""Non-CI microbenchmark: single-pass scoring vs legacy per-term rescoring. + +Verifies the single-pass refactor eliminates the T+1 graph-scoring passes a +T-term query used to run, without changing the result. Reports median/min +latency for: + + * legacy — one combined `_score_nodes(G, terms)` call PLUS one + `_score_nodes(G, [token])` call per distinct query token + (the old `_pick_seeds(terms=...)` per-term guarantee loop). + * optimized — one `_score_query(G, terms, collect_per_term_seeds=True)` + call, with the per-token best computed inline in the same + traversal. `_pick_seeds` then consumes `best_seed_by_term` + directly — no rescoring. + +Equality is asserted (ranked, best_seed_by_term, seeds) before timing. The +optimized path's traversal count is invariant in the number of query terms. + +Run it manually; do NOT wire this into CI (wall-clock assertions are flaky): + + uv run python tests/bench_query_scoring.py \\ + --nodes 100000 --term-counts 3,10 --repeats 5 + + uv run python tests/bench_query_scoring.py \\ + --graph graphify-out/graph.json \\ + --query "what calls extract" --query "symbol resolution" \\ + --repeats 10 +""" +from __future__ import annotations + +import argparse +import json +import random +import statistics +import sys +import time +from pathlib import Path + +import networkx as nx + +from graphify.serve import ( + _get_trigram_index, + _pick_seeds, + _query_terms, + _score_nodes, + _score_query, + _search_tokens, +) + + +SYLLABLES = [ + "foo", "bar", "baz", "get", "set", "run", "user", "name", "path", + "build", "report", "extract", "router", "config", "service", + "handler", "token", "auth", "rate", "limit", "widget", "model", +] + +QUERIES_BY_TERM_COUNT: dict[int, list[str]] = { + 1: ["foo"], + 2: ["foo", "bar"], + 3: ["router", "service", "handler"], + 5: ["get", "user", "run", "name", "path"], + 10: ["extract", "build", "report", "router", "config", + "service", "token", "rate", "limit", "widget"], +} + + +def _build_random_graph(n: int, *, seed: int) -> nx.DiGraph: + """Reproducible broad-match DiGraph: short constructed labels + edge noise. + + Labels draw from a small syllable pool so tokens collide across nodes, + forcing the trigram prefilter to be selective and exercising score ties + on common tokens. Edge noise provides degree variance so the legacy + tie-break (`max(tied, key=degree)`) is actually exercised against the + new `(-singleton, -degree, label_len, nid)` key tuple.""" + rng = random.Random(seed) + G: nx.DiGraph = nx.DiGraph() + for i in range(n): + label = "_".join(rng.sample(SYLLABLES, rng.randint(1, 3))) + G.add_node(f"n{i}", label=label, source_file=f"src/{label[:8]}.py") + for _ in range(n * 2): + a, b = rng.randrange(n), rng.randrange(n) + if a != b: + G.add_edge(f"n{a}", f"n{b}", relation="calls", confidence="EXTRACTED") + return G + + +def _load_real_graph(path: str) -> nx.Graph: + """Light wrapper that just builds a NetworkX graph from a real + `graphify-out/graph.json`, skipping the size cap and work-memory overlay + that `serve._load_graph` enforces (the bench is read-only).""" + data = json.loads(Path(path).read_text(encoding="utf-8")) + if "links" not in data and "edges" in data: + data = dict(data, links=data["edges"]) + data = {**data, "directed": True} + try: + return nx.readwrite.json_graph.node_link_graph(data, edges="links") + except TypeError: + return nx.readwrite.json_graph.node_link_graph(data) + + +def _legacy_score_and_pick(G, terms): + """Recreates the pre-refactor flow: combined scoring plus one + `_score_nodes([token])` call per distinct query token, with the legacy + tie-break (`max(tied, key=degree)` over a (-score, label_len, nid)-sorted + list) to derive `best_seed_by_term`. Returns the same triple as the + optimized path so they can be compared for equality.""" + ranked = _score_nodes(G, terms) + norm_terms = sorted({tok for t in terms for tok in _search_tokens(t)}) + best_seed_by_term: dict[str, str] = {} + for term in norm_terms: + term_scored = _score_nodes(G, [term]) + if not term_scored: + continue + best_score = term_scored[0][0] + tied = [nid for s, nid in term_scored if s == best_score] + best_nid = max(tied, key=lambda n: G.degree(n)) if len(tied) > 1 else term_scored[0][1] + best_seed_by_term[term] = best_nid + seeds = _pick_seeds(ranked, G=G, best_seed_by_term=best_seed_by_term) + return ranked, best_seed_by_term, seeds + + +def _optimized_score_and_pick(G, terms): + qs = _score_query(G, terms, collect_per_term_seeds=True) + seeds = _pick_seeds(qs.ranked, G=G, best_seed_by_term=qs.best_seed_by_term) + return qs.ranked, qs.best_seed_by_term, seeds + + +def _warm_caches(G, terms): + """Pre-populate the trigram index and IDF cache so the first timed + iteration doesn't pay the amortized build cost on either path. Both + legacy and optimized share these caches via the graph object, so warming + once is fair to both.""" + _get_trigram_index(G) + # Touch the idf cache for the combined terms and every per-token singleton, + # matching exactly the calls the legacy path will make. + _score_nodes(G, terms) + for term in {tok for t in terms for tok in _search_tokens(t)}: + _score_nodes(G, [term]) + + +def _bench(fn, *, repeats: int) -> list[float]: + # One uncounted warm-up — `_warm_caches` already populated caches, but + # this also amortizes any per-call code-path setup unique to `fn`. + fn() + times: list[float] = [] + for _ in range(repeats): + t0 = time.perf_counter() + fn() + times.append(time.perf_counter() - t0) + return times + + +def _verify_equality(G, terms) -> tuple[int, int]: + leg_rank, leg_best, leg_seeds = _legacy_score_and_pick(G, terms) + opt_rank, opt_best, opt_seeds = _optimized_score_and_pick(G, terms) + assert leg_rank == opt_rank, ( + f"ranked diverged for terms={terms}: legacy[:5]={leg_rank[:5]} opt[:5]={opt_rank[:5]}" + ) + assert leg_best == opt_best, ( + f"best_seed_by_term diverged for terms={terms}: legacy={leg_best} opt={opt_best}" + ) + assert leg_seeds == opt_seeds, ( + f"seeds diverged for terms={terms}: legacy={leg_seeds} opt={opt_seeds}" + ) + return len(leg_rank), len(leg_seeds) + + +def _row(label: str, n_nodes: int, n_terms: int, times: list[float], + traversal_count: int, n_ranked: int, n_seeds: int) -> str: + med = statistics.median(times) * 1000 + mn = min(times) * 1000 + return (f"{label:<10} | n={n_nodes:<7} | terms={n_terms:<3} | " + f"median={med:7.2f}ms | min={mn:7.2f}ms | " + f"passes={traversal_count:<3} | ranked={n_ranked:<6} seeds={n_seeds}") + + +def _legacy_traversal_count(terms) -> int: + # 1 combined pass + one per-token singleton pass. + return 1 + len({tok for t in terms for tok in _search_tokens(t)}) + + +def _run_scenario(G, terms, *, repeats: int) -> tuple[float, float]: + _warm_caches(G, terms) + n_ranked, n_seeds = _verify_equality(G, terms) + + legacy_times = _bench(lambda: _legacy_score_and_pick(G, terms), repeats=repeats) + opt_times = _bench(lambda: _optimized_score_and_pick(G, terms), repeats=repeats) + + n_nodes = G.number_of_nodes() + n_terms = len(set(tok for t in terms for tok in _search_tokens(t))) + print(_row("legacy", n_nodes, n_terms, legacy_times, + _legacy_traversal_count(terms), n_ranked, n_seeds)) + print(_row("optimized", n_nodes, n_terms, opt_times, + 1, n_ranked, n_seeds)) + + med_legacy = statistics.median(legacy_times) + med_opt = statistics.median(opt_times) + speedup = med_legacy / med_opt if med_opt > 0 else float("inf") + print(f"speedup | median: {speedup:.2f}x | " + f"min: {min(legacy_times) / min(opt_times):.2f}x") + return med_legacy, med_opt + + +def _resolve_scenarios(args) -> list[list[str]]: + if args.graph: + # Real-graph mode: each --query is a natural-language sentence, tokenized + # using the same helper the production path uses. + sentences = args.query or ["what calls extract"] + scenarios = [_query_terms(s) for s in sentences] + # Dedupe identical token sets (multiple --query args may tokenize the same). + seen: list[list[str]] = [] + for q in scenarios: + if q not in seen: + seen.append(q) + return seen + term_counts = [int(s.strip()) for s in args.term_counts.split(",") if s.strip()] + scenarios: list[list[str]] = [] + for tc in term_counts: + if tc in QUERIES_BY_TERM_COUNT: + scenarios.append(QUERIES_BY_TERM_COUNT[tc]) + elif tc > 0: + scenarios.append([SYLLABLES[i % len(SYLLABLES)] for i in range(tc)]) + return scenarios + + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser( + description="Microbenchmark single-pass query scoring vs legacy per-term rescoring.", + ) + p.add_argument("--nodes", type=int, default=100_000, + help="node count for the synthetic benchmark graph (default: 100000)") + p.add_argument("--seed", type=int, default=20260714, help="RNG seed for the synthetic graph") + p.add_argument("--term-counts", default="3,10", + help="comma-separated list of term counts to benchmark (synthetic mode)") + p.add_argument("--repeats", type=int, default=5, + help="timed iterations per scenario (after one warm-up)") + p.add_argument("--graph", default=None, + help="optional path to a real graphify-out/graph.json; overrides --nodes") + p.add_argument("--query", action="append", default=None, + help="natural-language query sentence (real-graph mode; repeat for multiple)") + args = p.parse_args(argv) + + if args.graph: + print(f"loading real graph: {args.graph} ...", file=sys.stderr) + t0 = time.perf_counter() + G = _load_real_graph(args.graph) + print(f" loaded in {time.perf_counter() - t0:.2f}s", file=sys.stderr) + else: + print(f"building synthetic graph: n={args.nodes} seed={args.seed} ...", + file=sys.stderr) + t0 = time.perf_counter() + G = _build_random_graph(args.nodes, seed=args.seed) + print(f" built in {time.perf_counter() - t0:.2f}s", file=sys.stderr) + + print() + print(f"graph: {G.number_of_nodes()} nodes, {G.number_of_edges()} edges") + scenarios = _resolve_scenarios(args) + print(f"scenarios: {len(scenarios)} | repeats per scenario: {args.repeats}") + print("-" * 110) + + summaries = [] + for terms in scenarios: + print() + med_legacy, med_opt = _run_scenario(G, terms, repeats=args.repeats) + summaries.append((terms, med_legacy, med_opt)) + + print() + print("-" * 110) + print("summary:") + for terms, med_legacy, med_opt in summaries: + speedup = med_legacy / med_opt if med_opt > 0 else float("inf") + print(f" terms={len(set(tok for t in terms for tok in _search_tokens(t))):>3} | " + f"median legacy={med_legacy*1000:>8.2f}ms | " + f"median optimized={med_opt*1000:>8.2f}ms | " + f"speedup={speedup:>5.2f}x") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/tests/test_serve.py b/tests/test_serve.py index 859cde026..dc4229c24 100644 --- a/tests/test_serve.py +++ b/tests/test_serve.py @@ -7,6 +7,7 @@ from graphify.serve import ( _communities_from_graph, _score_nodes, + _score_query, _compute_idf, _EXACT_MATCH_BONUS, _SOURCE_MATCH_BONUS, @@ -26,6 +27,7 @@ _subgraph_to_text, _load_graph, _community_header, + _search_tokens, ) @@ -734,8 +736,8 @@ def test_pick_seeds_respects_max_k(): def test_pick_seeds_without_diversity_args_is_unchanged(): - """G/terms are optional and default to None: existing callers see identical - behavior to before this change.""" + """G/best_seed_by_term are optional and default to None: existing callers + see identical behavior to before this change.""" scored = [(1000.0, "fbs"), (1.0, "err1"), (0.9, "err2")] assert _pick_seeds(scored) == ["fbs"] @@ -744,9 +746,9 @@ def test_pick_seeds_diversity_recovers_starved_term(monkeypatch): """Reproduces #1445: a vague natural-language query where one term's incidental EXACT match on an unrelated node (e.g. a common word also used as an unrelated field/identifier) outscores every SUBSTRING match on the - query's other, actually-relevant terms by ~1000x. Without G/terms, the - 20%-gap cutoff discards the relevant candidate entirely; with them, it is - recovered as a guaranteed per-term seed. + query's other, actually-relevant terms by ~1000x. Without + G/best_seed_by_term, the 20%-gap cutoff discards the relevant candidate + entirely; with them, it is recovered as a guaranteed per-term seed. """ G = nx.DiGraph() # "unrelated" is an exact label match for the query term "unrelated" and @@ -758,13 +760,17 @@ def test_pick_seeds_diversity_recovers_starved_term(monkeypatch): G.add_edge("other", "target") terms = ["unrelated", "widget"] - scored = _score_nodes(G, terms) + # `_score_query` does the combined scoring and the per-term singleton + # winner tracking in one traversal; `_pick_seeds` consumes its + # `best_seed_by_term` to satisfy the per-term guarantee without rescoring. + qs = _score_query(G, terms, collect_per_term_seeds=True) + scored = qs.ranked # Sanity check the premise: without diversity, only the exact match survives. seeds_before = _pick_seeds(scored) assert seeds_before == ["noise"] - seeds_after = _pick_seeds(scored, G=G, terms=terms) + seeds_after = _pick_seeds(scored, G=G, best_seed_by_term=qs.best_seed_by_term) assert "noise" in seeds_after assert "target" in seeds_after @@ -807,8 +813,9 @@ def test_pick_seeds_per_term_guarantee_does_not_reintroduce_generic_dupe(monkeyp G.add_node(f"get{i}", label="GET", source_file=f"r{i}.py") G.add_node("um", label="users_model", source_file="users.py") G.add_edge("um", "get0") - scored = _score_nodes(G, ["get", "users"]) - seeds = _pick_seeds(scored, G=G, terms=["get", "users"]) + terms = ["get", "users"] + qs = _score_query(G, terms, collect_per_term_seeds=True) + seeds = _pick_seeds(qs.ranked, G=G, best_seed_by_term=qs.best_seed_by_term) get_seeds = [s for s in seeds if s.startswith("get")] assert len(get_seeds) == 1, f"per-term guarantee reintroduced a GET dupe: {seeds}" @@ -974,3 +981,229 @@ def test_community_header_sanitizes_name(): out = _community_header(3, "Pay\x00ments\x1b[31m") assert out.startswith("Community 3 — ") assert "\x00" not in out and "\x1b" not in out + + +# --- single-pass scoring refactor: reference-impl equality + one-traversal --- + + +def _reference_best_seed_by_term(G: nx.Graph, terms: list[str]) -> dict[str, str]: + """Test-only oracle for the legacy per-term `_pick_seeds(terms=...)` loop. + + Re-creates what `_pick_seeds` did before the single-pass refactor: rescore + the whole graph per token via `_score_nodes(G, [token])`, take the top- + scoring ties, and break them by `max(tied, key=degree)` (which, over a + list sorted by `(-score, label_len, nid)`, returns the highest-degree node + with ties broken toward the shortest label then the smallest node id). + This is the semantics `_score_query(..., collect_per_term_seeds=True)` now + produces inline during its single traversal. + """ + norm_terms = sorted({tok for t in terms for tok in _search_tokens(t)}) + best: dict[str, str] = {} + for term in norm_terms: + term_scored = _score_nodes(G, [term]) + if not term_scored: + continue + best_score = term_scored[0][0] + tied = [nid for s, nid in term_scored if s == best_score] + best_nid = max(tied, key=lambda n: G.degree(n)) if len(tied) > 1 else term_scored[0][1] + best[term] = best_nid + return best + + +def _make_random_scoring_graph(n: int, *, seed: int) -> nx.DiGraph: + """Reproducible broad-match DiGraph: short constructed labels + edge noise. + + Labels draw from a small syllable pool so tokens collide across nodes, + forcing the trigram prefilter to be selective and exercising score ties + on common tokens. Edge noise provides degree variance so the legacy + tie-break (`max(tied, key=degree)`) is exercised against the new + `(-singleton, -degree, label_len, nid)` key tuple. + """ + import random + + rng = random.Random(seed) + syllables = [ + "foo", "bar", "baz", "get", "set", "run", "user", "name", "path", + "build", "report", "extract", "router", "config", "service", + "handler", "token", "auth", "rate", "limit", "widget", "model", + ] + G: nx.DiGraph = nx.DiGraph() + for i in range(n): + label = "_".join(rng.sample(syllables, rng.randint(1, 3))) + G.add_node(f"n{i}", label=label, source_file=f"src/{label[:8]}.py") + for _ in range(n * 2): + a, b = rng.randrange(n), rng.randrange(n) + if a != b: + G.add_edge(f"n{a}", f"n{b}", relation="calls", confidence="EXTRACTED") + return G + + +SYLLABLE_QUERIES = [ + ["get"], # single token, exact-match + ["get", "user"], # two distinct tokens + ["router", "service", "handler"], # multi-token identifier + ["extract", "build", "report", "path"], # broad term + ["nonexistent"], # no matches + ["nonexistent", "get"], # one missing term + match + ["bar", "bar"], # repeated token (must dedupe) + ["baz", "run", "set", "auth", "rate", "limit"], # many tokens +] + + +@pytest.mark.parametrize("terms", SYLLABLE_QUERIES) +def test_score_query_ranked_matches_score_nodes_byte_identical(terms): + """`_score_query(..., collect_per_term_seeds=False).ranked` is the byte-for- + byte match of `_score_nodes(G, terms)` — guaranteeing path/explain/tests see + no behavior change from the refactor.""" + G = _make_random_scoring_graph(80, seed=7) + assert _score_query(G, terms, collect_per_term_seeds=False).ranked == _score_nodes(G, terms) + + +@pytest.mark.parametrize("terms", SYLLABLE_QUERIES) +def test_score_query_best_seed_by_term_matches_legacy_singleton_scoring(terms): + """Per-token winner the single-pass scorer records matches the legacy + `_score_nodes([token])` + `max(tied, key=degree)` oracle exactly.""" + G = _make_random_scoring_graph(80, seed=7) + ref = _reference_best_seed_by_term(G, terms) + opt = _score_query(G, terms, collect_per_term_seeds=True).best_seed_by_term + assert ref == opt, f"terms={terms}: legacy={ref} optimized={opt}" + + +@pytest.mark.parametrize("terms", SYLLABLE_QUERIES) +def test_pick_seeds_with_optimized_best_seed_matches_legacy_semantics(terms): + """The seeds produced by `_pick_seeds(qs.ranked, G=G, best_seed_by_term= + qs.best_seed_by_term)` exactly match what the legacy `_pick_seeds(terms=...)` + loop would have produced (recreated via the reference oracle).""" + G = _make_random_scoring_graph(80, seed=7) + qs = _score_query(G, terms, collect_per_term_seeds=True) + ref_best = _reference_best_seed_by_term(G, terms) + # Legacy `_pick_seeds(terms=...)` ran `_score_nodes(G, [term])` per token + # to build ref_best, then deduped by label key. The new `_pick_seeds( + # best_seed_by_term=...)` only swaps the source of the per-token winners, + # so it must produce the same seeds given equivalent inputs. + opt_seeds = _pick_seeds(qs.ranked, G=G, best_seed_by_term=qs.best_seed_by_term) + ref_seeds = _pick_seeds(qs.ranked, G=G, best_seed_by_term=ref_best) + assert opt_seeds == ref_seeds, f"terms={terms}: ref={ref_seeds} opt={opt_seeds}" + # Per-term guarantee: every legacy winner with a non-empty seed slot is + # accounted for — either it appears in the seed list or another node with + # the same normalized label already claimed the slot (#1766 label dedup). + ref_seed_set = set(ref_seeds) + for term, nid in ref_best.items(): + if nid in ref_seed_set: + continue + nid_label = (G.nodes[nid].get("norm_label") + or G.nodes[nid].get("label") + or nid) + seeded_with_same_label = any( + (G.nodes[s].get("norm_label") or G.nodes[s].get("label") or s) == nid_label + for s in ref_seeds + ) + assert seeded_with_same_label, ( + f"term {term!r} winner {nid!r} dropped without label-dedup reason" + ) + + +def test_score_query_matches_legacy_across_random_deterministic_graphs(): + """Across many deterministic random graphs and many random multi-term + queries, the single-pass scorer's combined ranking, per-token winners, + and resulting seed list all match the legacy semantics. Exercises label + collisions, ties, broad terms, missing terms, and graph size variance.""" + import random + + rng = random.Random(42) + syllables = [ + "foo", "bar", "baz", "get", "set", "run", "user", "name", "path", + "build", "report", "extract", "router", "config", "service", + "handler", "token", "auth", "rate", "limit", "widget", "model", + ] + for trial in range(30): + n = rng.randint(20, 200) + G = _make_random_scoring_graph(n, seed=rng.randint(0, 10**9)) + nq = rng.randint(1, 5) + terms = [rng.choice(syllables) for _ in range(nq)] + ref_best = _reference_best_seed_by_term(G, terms) + opt = _score_query(G, terms, collect_per_term_seeds=True) + # (a) Combined ranking unchanged. + assert opt.ranked == _score_nodes(G, terms), ( + f"trial {trial}: combined ranking diverged for terms={terms}" + ) + # (b) Per-token winners match the legacy per-term rescoring loop. + assert opt.best_seed_by_term == ref_best, ( + f"trial {trial}: best_seed_by_term diverged; ref={ref_best} opt={opt.best_seed_by_term}" + ) + # (c) Final seed list is identical under the legacy semantics. + ref_seeds = _pick_seeds(opt.ranked, G=G, best_seed_by_term=ref_best) + opt_seeds = _pick_seeds(opt.ranked, G=G, best_seed_by_term=opt.best_seed_by_term) + assert opt_seeds == ref_seeds, ( + f"trial {trial}: seeds diverged; ref={ref_seeds} opt={opt_seeds}" + ) + + +def test_score_query_matches_legacy_under_full_scan_fallback(monkeypatch): + """When the trigram prefilter falls back to a full-graph scan, the + single-pass path still produces identical rankings and per-term winners. + + Forces `_trigram_candidates` to return None so the combined iterates the + whole graph — mirroring per-token `_score_nodes([token])` which would also + full-scan when its own trigram search isn't selective.""" + monkeypatch.setattr( + "graphify.serve._trigram_candidates", lambda G, needles: None + ) + terms = ["router", "service", "handler"] + G = _make_random_scoring_graph(80, seed=19) + ref_best = _reference_best_seed_by_term(G, terms) + opt = _score_query(G, terms, collect_per_term_seeds=True) + assert opt.ranked == _score_nodes(G, terms) + assert opt.best_seed_by_term == ref_best + + +def test_query_graph_text_makes_exactly_one_score_query_call(monkeypatch): + """`_query_graph_text` must invoke `_score_query` exactly once per query, + regardless of how many tokens the query has — eliminating the legacy + T+1-pass rescoring. `_score_nodes` must NOT be called from the query path + (only path/explain still call it).""" + G = _make_random_scoring_graph(60, seed=23) + original_sq = _score_query + original_sn = _score_nodes + + state = {"sq": 0, "sn": 0} + + def counting_sq(*a, **k): + state["sq"] += 1 + return original_sq(*a, **k) + + def counting_sn(*a, **k): + state["sn"] += 1 + return original_sn(*a, **k) + + monkeypatch.setattr("graphify.serve._score_query", counting_sq) + monkeypatch.setattr("graphify.serve._score_nodes", counting_sn) + + queries = [ + "foo", # one term + "foo bar", # two + "router service handler", # three (the scenario the RFC targets) + "get user run name path", # five + "extract build report router config service token rate limit widget", # ten + ] + for q in queries: + state["sq"] = 0 + state["sn"] = 0 + _query_graph_text(G, q, mode="bfs", depth=1) + assert state["sq"] == 1, ( + f"expected exactly one _score_query call for {q!r}, got {state['sq']}" + ) + assert state["sn"] == 0, ( + f"query path must not call _score_nodes; got {state['sn']} call(s) for {q!r}" + ) + + +def test_score_query_collect_per_term_seeds_false_omits_tracking(monkeypatch): + """`collect_per_term_seeds=False` returns empty `best_seed_by_term` and + does not pay for per-token best tracking — preserving the cost contract + for path/explain/tests callers that only want the combined ranking.""" + G = _make_random_scoring_graph(50, seed=29) + qs = _score_query(G, ["foo", "bar", "baz"], collect_per_term_seeds=False) + assert qs.best_seed_by_term == {} + # And the combined output is still byte-identical to _score_nodes. + assert qs.ranked == _score_nodes(G, ["foo", "bar", "baz"])