From cd9fb8618ebeae6a152071931e3ac92e7f7f8f9e Mon Sep 17 00:00:00 2001 From: himanshupatro-334 Date: Mon, 17 Aug 2026 23:56:32 +0530 Subject: [PATCH] fix: render rationale in graph read surfaces --- graphify/cli.py | 30 +++- graphify/security.py | 64 ++++++- graphify/serve.py | 62 ++++++- tests/test_explain_cli.py | 58 +++++- tests/test_query_cli.py | 36 ++++ tests/test_security.py | 102 +++++++++++ tests/test_serve.py | 360 ++++++++++++++++++++++++++++++++++++++ 7 files changed, 697 insertions(+), 15 deletions(-) diff --git a/graphify/cli.py b/graphify/cli.py index 95adad4b9c..88fb9e16e7 100644 --- a/graphify/cli.py +++ b/graphify/cli.py @@ -942,7 +942,7 @@ def dispatch_command(cmd: str) -> None: sys.exit(1) elif cmd == "query": if len(sys.argv) < 3: - print("Usage: graphify query \"\" [--dfs] [--context C] [--budget N] [--graph path]", file=sys.stderr) + print("Usage: graphify query \"\" [--dfs] [--context C] [--budget N] [--graph path] [--rationale]", file=sys.stderr) sys.exit(1) from graphify.serve import _query_graph_text from graphify.security import sanitize_label @@ -954,6 +954,7 @@ def dispatch_command(cmd: str) -> None: budget = 2000 graph_path = _default_graph_path() context_filters: list[str] = [] + include_rationale = False args = sys.argv[3:] i = 0 while i < len(args): @@ -980,6 +981,9 @@ def dispatch_command(cmd: str) -> None: elif args[i] == "--graph" and i + 1 < len(args): graph_path = args[i + 1] i += 2 + elif args[i] == "--rationale": + include_rationale = True + i += 1 else: i += 1 gp = Path(graph_path).resolve() @@ -1047,6 +1051,7 @@ def dispatch_command(cmd: str) -> None: depth=2, token_budget=budget, context_filters=context_filters, + include_rationale=include_rationale, ) querylog.log_query( kind="query", @@ -1440,17 +1445,25 @@ def dispatch_command(cmd: str) -> None: elif cmd == "explain": if len(sys.argv) < 3: - print('Usage: graphify explain "" [--graph path]', file=sys.stderr) + print('Usage: graphify explain "" [--graph path] [--rationale]', file=sys.stderr) sys.exit(1) from graphify.serve import _find_node, find_node_ambiguity from networkx.readwrite import json_graph label = sys.argv[2] graph_path = _default_graph_path() + include_rationale = False args = sys.argv[3:] - for i, a in enumerate(args): - if a == "--graph" and i + 1 < len(args): + i = 0 + while i < len(args): + if args[i] == "--graph" and i + 1 < len(args): graph_path = args[i + 1] + i += 2 + elif args[i] == "--rationale": + include_rationale = True + i += 1 + else: + i += 1 gp = Path(graph_path).resolve() if not gp.exists(): print(f"error: graph file not found: {gp}", file=sys.stderr) @@ -1486,6 +1499,15 @@ def dispatch_command(cmd: str) -> None: ) print(f" Type: {d.get('file_type', '')}") print(f" Community: {d.get('community_name') or d.get('community', '')}") + if include_rationale: + from graphify.security import sanitize_rationale, MAX_DETAIL_RATIONALE_CHARS + rat = sanitize_rationale( + d.get("rationale"), + single_line=False, + max_chars=MAX_DETAIL_RATIONALE_CHARS, + ) + if rat: + print(f" Rationale: {rat}") # Work-memory overlay: a derived experiential hint from `graphify reflect`, # merged in display-only from the .graphify_learning.json sidecar next to # graph.json. No line when the node has no overlay entry. diff --git a/graphify/security.py b/graphify/security.py index 2dbe5bd771..7c05524f97 100644 --- a/graphify/security.py +++ b/graphify/security.py @@ -387,7 +387,7 @@ def check_graph_file_size_cap(path: Path) -> None: # Label sanitisation (mirrors code-review-graph's _sanitize_name pattern) # --------------------------------------------------------------------------- -_CONTROL_CHAR_RE = re.compile(r"[\x00-\x1f\x7f]") +_CONTROL_CHAR_RE = re.compile(r"[\x00-\x1f\x7f-\x9f]") _MAX_LABEL_LEN = 256 @@ -405,6 +405,68 @@ def sanitize_label(text: str | None) -> str: return text +MAX_QUERY_RATIONALE_CHARS = 512 +MAX_DETAIL_RATIONALE_CHARS = 2048 + +_CONTROL_CHAR_NO_NL_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]") + + +def sanitize_rationale( + text: str | None, + *, + single_line: bool = False, + max_chars: int | None = None, +) -> str: + """Sanitize and optionally truncate a node rationale string. + + In compact (single-line) mode: + - Normalizes all whitespace runs (newlines, tabs, spaces) to single spaces + - Strips control characters + - Caps output at `max_chars` (default MAX_QUERY_RATIONALE_CHARS = 512) + - Adds an explicit '...' truncation marker when capped + + In detail mode: + - Preserves newlines/paragraphs while stripping unsafe control characters + - Caps output at `max_chars` (default MAX_DETAIL_RATIONALE_CHARS = 2048) + - Adds an explicit '...' truncation marker when capped + + Returns empty string if text is None, empty, or whitespace-only. + """ + if text is None: + return "" + raw = str(text) + if not raw.strip(): + return "" + + if max_chars is None: + max_chars = MAX_QUERY_RATIONALE_CHARS if single_line else MAX_DETAIL_RATIONALE_CHARS + + if single_line: + # Collapse all whitespace sequences to a single space, then strip control chars + flat = " ".join(raw.split()) + cleaned = _CONTROL_CHAR_RE.sub("", flat).strip() + if not cleaned: + return "" + if len(cleaned) > max_chars: + marker = "..." + if max_chars <= len(marker): + return cleaned[:max_chars] + return cleaned[: max_chars - len(marker)].rstrip() + marker + return cleaned + else: + # Preserve newlines/paragraphs, normalize CRLF to LF, strip other control chars + normalized = raw.replace("\r\n", "\n").replace("\r", "\n") + cleaned = _CONTROL_CHAR_NO_NL_RE.sub("", normalized).strip() + if not cleaned: + return "" + if len(cleaned) > max_chars: + marker = "..." + if max_chars <= len(marker): + return cleaned[:max_chars] + return cleaned[: max_chars - len(marker)].rstrip() + marker + return cleaned + + # --------------------------------------------------------------------------- # Metadata sanitisation (recursive, bounded, HTML-safe) # --------------------------------------------------------------------------- diff --git a/graphify/serve.py b/graphify/serve.py index feb89ce399..37b5e74310 100644 --- a/graphify/serve.py +++ b/graphify/serve.py @@ -12,7 +12,13 @@ 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 +from graphify.security import ( + sanitize_label, + check_graph_file_size_cap, + sanitize_rationale, + MAX_QUERY_RATIONALE_CHARS, + MAX_DETAIL_RATIONALE_CHARS, +) from graphify.build import edge_data, edge_datas from graphify.paths import default_graph_json as _default_graph_json @@ -979,7 +985,7 @@ def _dfs(G: nx.Graph, start_nodes: list[str], depth: int) -> tuple[set[str], lis return visited, edges_seen -def _subgraph_to_text(G: nx.Graph, nodes: set[str], edges: list[tuple], token_budget: int = 2000, *, seeds: list[str] | None = None) -> str: +def _subgraph_to_text(G: nx.Graph, nodes: set[str], edges: list[tuple], token_budget: int = 2000, *, seeds: list[str] | None = None, include_rationale: bool = False) -> str: """Render subgraph as text, cutting at token_budget (approx 3 chars/token). seeds: exact-match nodes rendered first before the degree-sorted expansion, @@ -1034,12 +1040,22 @@ def _adj(n): status = sanitize_label(str(entry.get("status", ""))) if status: learning_suffix = f" learning={status}{':stale' if entry.get('stale') else ''}" + why_suffix = "" + if include_rationale: + rationale_text = sanitize_rationale( + d.get("rationale"), + single_line=True, + max_chars=MAX_QUERY_RATIONALE_CHARS, + ) + if rationale_text: + why_suffix = f" WHY {rationale_text}" line = ( f"NODE {sanitize_label(d.get('label', nid))} " f"[src={sanitize_label(str(d.get('source_file', '')))} " f"loc={sanitize_label(str(d.get('source_location', '')))} " f"community={sanitize_label(str(d.get('community_name') or d.get('community', '')))}" f"{learning_suffix}]" + f"{why_suffix}" ) lines.append(line) for u, v in edges: @@ -1171,6 +1187,7 @@ def _query_graph_text( depth: int = 3, token_budget: int = 2000, context_filters: list[str] | None = None, + include_rationale: bool = False, ) -> str: terms = _query_terms(question) # One graph scoring pass produces both the combined ranking (used to drive @@ -1210,7 +1227,14 @@ def _query_graph_text( # Pass the seeds so the queried symbol renders first and survives truncation # (#BUG2): a branch merge had silently dropped this argument, leaving the # seed-first ordering as dead code. - return header + _subgraph_to_text(traversal_graph, nodes, edges, token_budget, seeds=start_nodes) + return header + _subgraph_to_text( + traversal_graph, + nodes, + edges, + token_budget, + seeds=start_nodes, + include_rationale=include_rationale, + ) def _find_node_tiers( @@ -1561,6 +1585,11 @@ async def list_tools() -> list[types.Tool]: "items": {"type": "string"}, "description": "Optional explicit edge-context filter, e.g. ['call', 'field']", }, + "include_rationale": { + "type": "boolean", + "default": False, + "description": "Include WHY rationale on nodes when available", + }, }, "required": ["question"], }, @@ -1570,7 +1599,14 @@ async def list_tools() -> list[types.Tool]: description="Get full details for a specific node by label or ID.", inputSchema={ "type": "object", - "properties": {"label": {"type": "string", "description": "Node label or ID to look up"}}, + "properties": { + "label": {"type": "string", "description": "Node label or ID to look up"}, + "include_rationale": { + "type": "boolean", + "default": False, + "description": "Include node rationale if available", + }, + }, "required": ["label"], }, ), @@ -1702,6 +1738,7 @@ def _tool_query_graph(arguments: dict) -> str: depth = min(int(arguments.get("depth", 3)), 6) budget = int(arguments.get("token_budget", 2000)) context_filter = arguments.get("context_filter") + include_rationale = arguments.get("include_rationale") is True _t0 = _time.perf_counter() result = _query_graph_text( G, @@ -1710,6 +1747,7 @@ def _tool_query_graph(arguments: dict) -> str: depth=depth, token_budget=budget, context_filters=context_filter, + include_rationale=include_rationale, ) querylog.log_query( kind="mcp_query", @@ -1725,20 +1763,30 @@ def _tool_query_graph(arguments: dict) -> str: def _tool_get_node(arguments: dict) -> str: label = arguments["label"].lower() + include_rationale = arguments.get("include_rationale") is True matches = [(nid, d) for nid, d in G.nodes(data=True) if label in (d.get("label") or "").lower() or label == nid.lower()] if not matches: return f"No node matching '{label}' found." nid, d = matches[0] # Sanitise every LLM-derived field before concatenation (F-010). - return "\n".join([ + lines = [ f"Node: {sanitize_label(d.get('label', nid))}", f" ID: {sanitize_label(nid)}", f" Source: {sanitize_label(str(d.get('source_file', '')))} {sanitize_label(str(d.get('source_location', '')))}", f" Type: {sanitize_label(str(d.get('file_type', '')))}", f" Community: {sanitize_label(str(d.get('community_name') or d.get('community', '')))}", - f" Degree: {G.degree(nid)}", - ]) + ] + if include_rationale: + rat = sanitize_rationale( + d.get("rationale"), + single_line=False, + max_chars=MAX_DETAIL_RATIONALE_CHARS, + ) + if rat: + lines.append(f" Rationale: {rat}") + lines.append(f" Degree: {G.degree(nid)}") + return "\n".join(lines) def _tool_get_neighbors(arguments: dict) -> str: label = arguments["label"].lower() diff --git a/tests/test_explain_cli.py b/tests/test_explain_cli.py index 60b3e626e9..50965a8596 100644 --- a/tests/test_explain_cli.py +++ b/tests/test_explain_cli.py @@ -31,10 +31,12 @@ def _write_graph(tmp_path): return p -def _run(monkeypatch, graph_path, label, capsys): +def _run(monkeypatch, graph_path, label, capsys, extra_args=None): monkeypatch.setattr(mainmod, "_check_skill_version", lambda _: None) - monkeypatch.setattr(mainmod.sys, "argv", - ["graphify", "explain", label, "--graph", str(graph_path)]) + argv = ["graphify", "explain", label, "--graph", str(graph_path)] + if extra_args: + argv.extend(extra_args) + monkeypatch.setattr(mainmod.sys, "argv", argv) mainmod.main() return capsys.readouterr().out @@ -320,3 +322,53 @@ def test_explain_matches_within_one_file_are_not_ambiguous(monkeypatch, tmp_path out = _run(monkeypatch, p, "MetricsPort", capsys) assert "Ambiguous" not in out assert "Node: MetricsPort" in out + + +def test_explain_rationale_flag_shows_rationale(monkeypatch, tmp_path, capsys): + graph_data = { + "directed": False, "multigraph": False, "graph": {}, + "nodes": [ + { + "id": "cluster_node", + "label": "Clustering per deposito", + "source_file": "deposito.py", + "source_location": "L10", + "file_type": "code", + "community": 0, + "rationale": "cannot work on ORIGINALS_NORMALIZED as it stands, because timestamps and coordinates are missing", + } + ], + "links": [], + } + p = tmp_path / "graph.json" + p.write_text(json.dumps(graph_data)) + + # Without --rationale flag + out_default = _run(monkeypatch, p, "Clustering per deposito", capsys) + assert "Rationale:" not in out_default + + # With --rationale flag + out_rat = _run(monkeypatch, p, "Clustering per deposito", capsys, extra_args=["--rationale"]) + assert " Rationale: cannot work on ORIGINALS_NORMALIZED as it stands, because timestamps and coordinates are missing" in out_rat + + +def test_explain_rationale_empty_omitted(monkeypatch, tmp_path, capsys): + graph_data = { + "directed": False, "multigraph": False, "graph": {}, + "nodes": [ + { + "id": "empty_rat_node", + "label": "EmptyNode", + "source_file": "empty.py", + "source_location": "L1", + "file_type": "code", + "community": 0, + "rationale": " \n\t ", + } + ], + "links": [], + } + p = tmp_path / "graph.json" + p.write_text(json.dumps(graph_data)) + out = _run(monkeypatch, p, "EmptyNode", capsys, extra_args=["--rationale"]) + assert "Rationale:" not in out diff --git a/tests/test_query_cli.py b/tests/test_query_cli.py index 0db4e6fa8a..d3465ab09c 100644 --- a/tests/test_query_cli.py +++ b/tests/test_query_cli.py @@ -123,3 +123,39 @@ def test_query_cli_rejects_oversized_graph(monkeypatch, tmp_path, capsys): err = capsys.readouterr().err assert "exceeds" in err assert "byte cap" in err + + +def test_query_cli_rationale_flag(monkeypatch, tmp_path, capsys): + G = nx.Graph() + G.add_node( + "auth_node", + label="authenticate", + source_file="auth.py", + source_location="L10", + community=0, + rationale="Decision: Argon2id hashing.\n\nWHY: Resistant to GPU cracking.", + ) + graph_path = tmp_path / "graph.json" + graph_path.write_text(json.dumps(json_graph.node_link_data(G, edges="links"))) + + monkeypatch.setattr(mainmod, "_check_skill_version", lambda _: None) + + # Without --rationale + monkeypatch.setattr( + mainmod.sys, + "argv", + ["graphify", "query", "authenticate", "--graph", str(graph_path)], + ) + mainmod.main() + out_default = capsys.readouterr().out + assert "WHY" not in out_default + + # With --rationale + monkeypatch.setattr( + mainmod.sys, + "argv", + ["graphify", "query", "authenticate", "--graph", str(graph_path), "--rationale"], + ) + mainmod.main() + out_rat = capsys.readouterr().out + assert "WHY Decision: Argon2id hashing. WHY: Resistant to GPU cracking." in out_rat diff --git a/tests/test_security.py b/tests/test_security.py index 74669f9383..500ca394e5 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -461,3 +461,105 @@ def test_sanitize_metadata_bool_not_coerced_to_int(): assert out["flag_t"] is True assert out["flag_f"] is False assert out["num"] == 1 + + +# --------------------------------------------------------------------------- +# sanitize_rationale +# --------------------------------------------------------------------------- + +from graphify.security import ( + sanitize_rationale, + MAX_QUERY_RATIONALE_CHARS, + MAX_DETAIL_RATIONALE_CHARS, +) + + +def test_sanitize_rationale_empty_or_none(): + assert sanitize_rationale(None) == "" + assert sanitize_rationale("") == "" + assert sanitize_rationale(" \n\t ") == "" + + +def test_sanitize_rationale_single_line_normalizes_whitespace(): + raw = "Decision: Use tree-sitter.\n\nWHY: Deterministic\tand fast.\r\nRobust." + out = sanitize_rationale(raw, single_line=True) + assert "\n" not in out + assert "\t" not in out + assert "\r" not in out + assert out == "Decision: Use tree-sitter. WHY: Deterministic and fast. Robust." + + +def test_sanitize_rationale_single_line_strips_control_chars(): + raw = "Good\x00 rationale\x1b[31m with ansi\x07" + out = sanitize_rationale(raw, single_line=True) + assert "\x00" not in out + assert "\x1b" not in out + assert "\x07" not in out + assert "Good rationale[31m with ansi" in out + + +def test_sanitize_rationale_single_line_truncation(): + long_text = "A" * (MAX_QUERY_RATIONALE_CHARS + 100) + out = sanitize_rationale(long_text, single_line=True) + assert len(out) == MAX_QUERY_RATIONALE_CHARS + assert out.endswith("...") + + +def test_sanitize_rationale_detail_mode_preserves_newlines(): + raw = "Decision: Pick SQLite.\n\nWHY: Zero config." + out = sanitize_rationale(raw, single_line=False) + assert out == "Decision: Pick SQLite.\n\nWHY: Zero config." + + +def test_sanitize_rationale_detail_mode_truncation(): + long_text = "B" * (MAX_DETAIL_RATIONALE_CHARS + 200) + out = sanitize_rationale(long_text, single_line=False) + assert len(out) == MAX_DETAIL_RATIONALE_CHARS + assert out.endswith("...") + + +def test_sanitize_rationale_strips_c1_control_chars_compact_and_detail(): + # Test full range of C1 controls U+0080 through U+009F + c1_chars = "".join(chr(cp) for cp in range(0x80, 0xA0)) + raw = f"Prefix{c1_chars}Suffix" + + out_compact = sanitize_rationale(raw, single_line=True) + assert all(chr(cp) not in out_compact for cp in range(0x80, 0xA0)) + + out_detail = sanitize_rationale(raw, single_line=False) + assert all(chr(cp) not in out_detail for cp in range(0x80, 0xA0)) + assert out_detail == "PrefixSuffix" + + # Non-whitespace C1 controls strip without word splitting + c1_non_ws = "".join(chr(cp) for cp in range(0x80, 0xA0) if cp != 0x85) + assert sanitize_rationale(f"Prefix{c1_non_ws}Suffix", single_line=True) == "PrefixSuffix" + + +def test_sanitize_rationale_strips_dangerous_c1_introducers(): + # U+009B (CSI), U+009D (OSC), U+0090 (DCS), U+009F (APC) + raw = "Safe\x9b31mColor\x9dTitle\x90Device\x9fApp" + + out_compact = sanitize_rationale(raw, single_line=True) + assert "\x9b" not in out_compact + assert "\x9d" not in out_compact + assert "\x90" not in out_compact + assert "\x9f" not in out_compact + assert out_compact == "Safe31mColorTitleDeviceApp" + + out_detail = sanitize_rationale(raw, single_line=False) + assert "\x9b" not in out_detail + assert "\x9d" not in out_detail + assert "\x90" not in out_detail + assert "\x9f" not in out_detail + assert out_detail == "Safe31mColorTitleDeviceApp" + + +def test_sanitize_rationale_preserves_printable_unicode(): + raw = "Decisión: café con leche.\n\nWHY: 日本語サポート and fast 🚀 emojis." + + out_compact = sanitize_rationale(raw, single_line=True) + assert out_compact == "Decisión: café con leche. WHY: 日本語サポート and fast 🚀 emojis." + + out_detail = sanitize_rationale(raw, single_line=False) + assert out_detail == "Decisión: café con leche.\n\nWHY: 日本語サポート and fast 🚀 emojis." + assert "\n\n" in out_detail diff --git a/tests/test_serve.py b/tests/test_serve.py index 85f77a59a2..cdbd1b4bb0 100644 --- a/tests/test_serve.py +++ b/tests/test_serve.py @@ -1666,6 +1666,8 @@ def test_shortest_path_tool_undirected_opt_in(): assert "Shortest path (2 hops)" in out assert out.count("<--calls--") == 2 assert "-->" not in out + + def test_underscore_query_matches_hyphenated_label(): r"""Separator-blind seeding: `_` must split like `-` does. @@ -1711,3 +1713,361 @@ def test_underscore_query_does_not_let_a_single_token_outrank_the_real_match(): scored = _score_nodes(G, _query_terms("user_service_client")) assert scored, "the multi-token query must match the full-label node" assert scored[0][1] == "real", f"a single-token node out-ranked the real match: {scored}" + + +# --- Rationale Rendering Tests (#2821) --- + +def test_subgraph_to_text_rationale_disabled_by_default(): + G = nx.DiGraph() + G.add_node("n1", label="Parser", source_file="p.py", source_location="L10", community=0, rationale="Use tree-sitter for speed") + out = _subgraph_to_text(G, {"n1"}, []) + assert "WHY" not in out + assert "tree-sitter" not in out + assert out.strip() == "NODE Parser [src=p.py loc=L10 community=0]" + + +def test_subgraph_to_text_rationale_enabled_renders_why(): + G = nx.DiGraph() + G.add_node("n1", label="Parser", source_file="p.py", source_location="L10", community=0, rationale="Use tree-sitter for speed") + out = _subgraph_to_text(G, {"n1"}, [], include_rationale=True) + assert "WHY Use tree-sitter for speed" in out + assert out.strip() == "NODE Parser [src=p.py loc=L10 community=0] WHY Use tree-sitter for speed" + + +def test_subgraph_to_text_rationale_multiline_normalized(): + G = nx.DiGraph() + G.add_node("n1", label="Auth", source_file="a.py", source_location="L1", community=0, rationale="Decision: JWT tokens.\n\nWHY: Stateless auth across services.") + out = _subgraph_to_text(G, {"n1"}, [], include_rationale=True) + assert "\n" not in out.strip() + assert "WHY Decision: JWT tokens. WHY: Stateless auth across services." in out + + +def test_subgraph_to_text_rationale_empty_or_missing_omitted(): + G = nx.DiGraph() + G.add_node("n1", label="NoRat", source_file="a.py", source_location="L1", community=0) + G.add_node("n2", label="EmptyRat", source_file="b.py", source_location="L2", community=0, rationale=" \n\t ") + G.add_node("n3", label="NoneRat", source_file="c.py", source_location="L3", community=0, rationale=None) + out = _subgraph_to_text(G, {"n1", "n2", "n3"}, [], include_rationale=True) + assert "WHY" not in out + + +def test_subgraph_to_text_rationale_capped_at_512(): + G = nx.DiGraph() + long_rationale = "Word " * 200 + G.add_node("n1", label="Long", source_file="l.py", source_location="L1", community=0, rationale=long_rationale) + out = _subgraph_to_text(G, {"n1"}, [], include_rationale=True) + node_line = out.strip() + why_part = node_line.split(" WHY ", 1)[1] + assert len(why_part) == 512 + assert why_part.endswith("...") + + +def test_subgraph_to_text_rationale_counts_toward_token_budget(): + G = nx.DiGraph() + G.add_node("s1", label="SeedNode", source_file="s.py", source_location="L1", community=0, rationale="A" * 200) + G.add_node("n1", label="Neighbor1", source_file="n.py", source_location="L1", community=0) + G.add_edge("s1", "n1", relation="calls") + # Small budget where rationale pushes neighbor past cut threshold + out_without = _subgraph_to_text(G, {"s1", "n1"}, [("s1", "n1")], token_budget=45, seeds=["s1"], include_rationale=False) + out_with = _subgraph_to_text(G, {"s1", "n1"}, [("s1", "n1")], token_budget=45, seeds=["s1"], include_rationale=True) + assert "Neighbor1" in out_without + assert "TRUNCATED" in out_with + assert "WHY" in out_with + + +def test_query_graph_text_forward_include_rationale(): + G = nx.DiGraph() + G.add_node("auth_mod", label="AuthModule", source_file="auth.py", source_location="L1", community=0, rationale="Central auth gateway") + out_no_rat = _query_graph_text(G, "AuthModule", include_rationale=False) + out_rat = _query_graph_text(G, "AuthModule", include_rationale=True) + assert "WHY" not in out_no_rat + assert "WHY Central auth gateway" in out_rat + + +def test_tool_get_node_and_query_graph_mcp(tmp_path): + import asyncio + import mcp.types as types + from graphify.serve import _build_server + from graphify.export import to_json + + G = nx.DiGraph() + G.add_node( + "auth_service", + label="AuthService", + source_file="auth.py", + source_location="L10", + file_type="code", + community=1, + rationale="Decision: OAuth2 compliance.\n\nWHY: Standard protocol.", + ) + gp = tmp_path / "graph.json" + to_json(G, {1: ["auth_service"]}, str(gp)) + + server = _build_server(str(gp)) + + # Test list_tools schema + list_handler = server.request_handlers[types.ListToolsRequest] + tools_res = asyncio.run(list_handler(types.ListToolsRequest())) + tools_by_name = {t.name: t for t in tools_res.root.tools} + + assert "include_rationale" in tools_by_name["query_graph"].inputSchema["properties"] + assert "include_rationale" in tools_by_name["get_node"].inputSchema["properties"] + + # Test call_tool get_node without rationale (default) + call_handler = server.request_handlers[types.CallToolRequest] + res_no_rat = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="get_node", + arguments={"label": "AuthService", "include_rationale": False}, + ) + ) + ) + ) + text_no_rat = res_no_rat.root.content[0].text + assert "Rationale:" not in text_no_rat + + # Test call_tool get_node with rationale + res_rat = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="get_node", + arguments={"label": "AuthService", "include_rationale": True}, + ) + ) + ) + ) + text_rat = res_rat.root.content[0].text + assert " Rationale: Decision: OAuth2 compliance.\n\nWHY: Standard protocol." in text_rat + + # Test call_tool query_graph without rationale + q_no_rat = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="query_graph", + arguments={"question": "AuthService", "include_rationale": False}, + ) + ) + ) + ) + q_text_no_rat = q_no_rat.root.content[0].text + assert "WHY" not in q_text_no_rat + + # Test call_tool query_graph with rationale + q_rat = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="query_graph", + arguments={"question": "AuthService", "include_rationale": True}, + ) + ) + ) + ) + q_text_rat = q_rat.root.content[0].text + assert "WHY Decision: OAuth2 compliance. WHY: Standard protocol." in q_text_rat + + +def test_mcp_backward_compatibility_omitted_include_rationale(tmp_path): + """Regression test: calling get_node and query_graph without include_rationale in arguments + + must succeed, omit rationale/WHY, and preserve exact previous default output. + """ + import asyncio + import mcp.types as types + from graphify.serve import _build_server + from graphify.export import to_json + + G = nx.DiGraph() + G.add_node( + "auth_service", + label="AuthService", + source_file="auth.py", + source_location="L10", + file_type="code", + community=1, + rationale="Decision: OAuth2 compliance.\n\nWHY: Standard protocol.", + ) + gp = tmp_path / "graph.json" + to_json(G, {1: ["auth_service"]}, str(gp)) + + server = _build_server(str(gp)) + call_handler = server.request_handlers[types.CallToolRequest] + + # Call get_node with ONLY {"label": "AuthService"} + res = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="get_node", + arguments={"label": "AuthService"}, + ) + ) + ) + ) + text = res.root.content[0].text + assert "Rationale:" not in text + expected_lines = [ + "Node: AuthService", + " ID: auth_service", + " Source: auth.py L10", + " Type: code", + " Community: 1", + " Degree: 0", + ] + assert text == "\n".join(expected_lines) + + # Call query_graph with ONLY {"question": "AuthService"} + q_res = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="query_graph", + arguments={"question": "AuthService"}, + ) + ) + ) + ) + q_text = q_res.root.content[0].text + assert "WHY" not in q_text + assert "NODE AuthService [src=auth.py loc=L10 community=1]" in q_text + + +def test_mcp_strict_boolean_include_rationale(tmp_path): + """Regression test: string 'false'/'true' or truthy non-booleans must NOT enable rationale.""" + import asyncio + import mcp.types as types + from graphify.serve import _build_server + from graphify.export import to_json + + G = nx.DiGraph() + G.add_node( + "auth_service", + label="AuthService", + source_file="auth.py", + source_location="L10", + file_type="code", + community=1, + rationale="Decision: OAuth2 compliance.\n\nWHY: Standard protocol.", + ) + gp = tmp_path / "graph.json" + to_json(G, {1: ["auth_service"]}, str(gp)) + + server = _build_server(str(gp)) + call_handler = server.request_handlers[types.CallToolRequest] + + # 1. String "false" + res_get_str_false = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="get_node", + arguments={"label": "AuthService", "include_rationale": "false"}, + ) + ) + ) + ) + assert "Rationale:" not in res_get_str_false.root.content[0].text + + res_q_str_false = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="query_graph", + arguments={"question": "AuthService", "include_rationale": "false"}, + ) + ) + ) + ) + assert "WHY" not in res_q_str_false.root.content[0].text + + # 2. String "true" + res_get_str_true = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="get_node", + arguments={"label": "AuthService", "include_rationale": "true"}, + ) + ) + ) + ) + assert "Rationale:" not in res_get_str_true.root.content[0].text + + res_q_str_true = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="query_graph", + arguments={"question": "AuthService", "include_rationale": "true"}, + ) + ) + ) + ) + assert "WHY" not in res_q_str_true.root.content[0].text + + # 3. None / integer truthy + res_get_int = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="get_node", + arguments={"label": "AuthService", "include_rationale": 1}, + ) + ) + ) + ) + assert "Rationale:" not in res_get_int.root.content[0].text + + # 4. Actual True + res_get_true = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="get_node", + arguments={"label": "AuthService", "include_rationale": True}, + ) + ) + ) + ) + assert " Rationale: Decision: OAuth2 compliance.\n\nWHY: Standard protocol." in res_get_true.root.content[0].text + + res_q_true = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="query_graph", + arguments={"question": "AuthService", "include_rationale": True}, + ) + ) + ) + ) + assert "WHY Decision: OAuth2 compliance. WHY: Standard protocol." in res_q_true.root.content[0].text + + # 5. Actual False + res_get_false = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="get_node", + arguments={"label": "AuthService", "include_rationale": False}, + ) + ) + ) + ) + assert "Rationale:" not in res_get_false.root.content[0].text + + res_q_false = asyncio.run( + call_handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="query_graph", + arguments={"question": "AuthService", "include_rationale": False}, + ) + ) + ) + ) + assert "WHY" not in res_q_false.root.content[0].text