From 069d4cf09cabcec012d875d8290605461fa03e51 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yanis=20Gu=C3=A9rault?= Date: Wed, 22 Jul 2026 12:02:25 +0200 Subject: [PATCH 1/2] feat(serve): add multi-graph support to MCP server Enable serving multiple knowledge graphs from a single MCP endpoint. Always registry-based: single graph = 1-entry registry, multi-graph = directory scan via --graphs-dir flag or GRAPHS_DIR env var. - Add GraphContext dataclass + GraphRegistry (from_path, from_directory) - Refactor _build_server to accept GraphRegistry with per-call resolution - Tool visibility keyed on registry size: list_graphs/use_graph when >1, PR tools when ==1. New graph param on all tool schemas. - Add --graphs-dir CLI flag forcing HTTP transport with auto-rescan - Remove multi_serve.py and graphify-multi-mcp entry point (consolidated) Co-Authored-By: Claude Opus 4.6 (1M context) --- .dockerignore | 3 + .gitignore | 1 + Dockerfile | 16 +- README.md | 37 ++- docker-compose.multi.yml | 18 ++ graphify/serve.py | 558 ++++++++++++++++++++++++++++++--------- tests/test_mcp_cli.py | 175 ++++++++++++ tests/test_serve.py | 487 +++++++++++++++++++++++++++++++++- tests/test_serve_http.py | 109 ++++++-- 9 files changed, 1238 insertions(+), 166 deletions(-) create mode 100644 docker-compose.multi.yml create mode 100644 tests/test_mcp_cli.py diff --git a/.dockerignore b/.dockerignore index 44aa49437f..5cb68e7fe5 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,5 +1,8 @@ .git .github +.env +.env.* +!.env.example .venv venv __pycache__ diff --git a/.gitignore b/.gitignore index 0a6775b2a8..5ef097f618 100644 --- a/.gitignore +++ b/.gitignore @@ -14,6 +14,7 @@ build/ *.egg .graphify/ graphify-out/ +.worktrees/ .graphify_*.json .graphify_python .claude/ diff --git a/Dockerfile b/Dockerfile index a313833c5c..1bde1cd9f0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,25 +1,15 @@ -# graphify MCP server as a shared HTTP service (issue #1143). -# -# Build: docker build -t graphify . -# Run: docker run -p 8080:8080 -v "$(pwd)/graphify-out:/data" graphify \ -# /data/graph.json --transport http --host 0.0.0.0 --api-key "$SECRET" -# -# Builds from source so the image includes the Streamable HTTP transport even -# before it lands on PyPI. The graph.json is mounted at runtime (-v), never -# baked into the image. +# graphify MCP server. Mount a repository containing graphify-out/graph.json. FROM python:3.12-slim - WORKDIR /app COPY . /app # The [mcp] extra pulls mcp + starlette + uvicorn, which the HTTP transport needs. RUN pip install --no-cache-dir ".[mcp]" -# Run as a non-root user — the server is network-exposed. +# Run as a non-root user because the server is network-exposed. RUN useradd --create-home --uid 10001 graphify USER graphify EXPOSE 8080 - ENTRYPOINT ["python", "-m", "graphify.serve"] -CMD ["/data/graph.json", "--transport", "http", "--host", "0.0.0.0", "--port", "8080"] +CMD ["--graphs-dir", "/data", "--transport", "http", "--host", "0.0.0.0", "--port", "8080"] diff --git a/README.md b/README.md index d83ea76c0c..86d0a97314 100644 --- a/README.md +++ b/README.md @@ -505,8 +505,7 @@ The default `127.0.0.1` bind is loopback-only. Set `--host 0.0.0.0` **and** `--a ```bash docker build -t graphify . -docker run -p 8080:8080 -v "$(pwd)/graphify-out:/data" graphify \ - /data/graph.json --transport http --host 0.0.0.0 --api-key "$SECRET" +docker run -p 8080:8080 -e GRAPHIFY_API_KEY="$GRAPHIFY_API_KEY" -v "$(pwd):/data:ro" graphify --graphs-dir /data --transport http --host 0.0.0.0 ``` > **WSL / Linux note:** Ubuntu ships `python3`, not `python`. Use a venv to avoid conflicts: @@ -514,6 +513,38 @@ docker run -p 8080:8080 -v "$(pwd)/graphify-out:/data" graphify \ > python3 -m venv .venv && .venv/bin/pip install "graphifyy[mcp]" > ``` +### Multi-graph MCP server + +Serve multiple repositories containing `graphify-out/graph.json` from one MCP endpoint. This is useful for multi-repo setups, monorepos with per-service graphs, or comparing codebases. + +```bash +# Serve over stdio (the default transport). +python -m graphify.serve --graphs-dir .. + +# Serve over HTTP. Remote HTTP requires a nonblank API key. +python -m graphify.serve --graphs-dir .. --transport http --host 0.0.0.0 --port 8080 --api-key "$SECRET" +``` + +#### Docker Compose + +`docker-compose.multi.yml` serves two pre-built repository graphs over local +HTTP. Each repository must contain `graphify-out/graph.json`. + +1. Edit the two `volumes` entries in `docker-compose.multi.yml` to point at + your repositories. Keep their matching `/repos/...` command arguments in + sync when adding or removing repositories. Mounts are read-only. +2. Start the server with a non-empty API key: + ```bash + GRAPHIFY_API_KEY=your-secret docker compose -f docker-compose.multi.yml up --build + ``` +3. Configure your MCP client with `http://localhost:8080/mcp` and send + `Authorization: Bearer `. + +The Compose port binds to localhost only. Graphify reads the mounted graphs; +it does not extract or modify either repository. + +Tools: same as single-graph (`query_graph`, `get_node`, `get_neighbors`, etc.) plus `list_graphs` and `use_graph`. Each tool accepts an optional `graph` parameter to target a specific graph, or use `use_graph` to set a session default. + --- ## Environment variables @@ -685,7 +716,7 @@ graphify extract ./raw --code-only # index code only — local AST, no API key ( /graphify ./raw --falkordb # generate cypher.txt for FalkorDB /graphify ./raw --falkordb-push falkordb://localhost:6379 /graphify ./raw --watch # auto-sync as files change -/graphify ./raw --mcp # start MCP stdio server +python -m graphify.serve ./raw/graphify-out/graph.json # start MCP stdio server /graphify add https://arxiv.org/abs/1706.03762 /graphify add diff --git a/docker-compose.multi.yml b/docker-compose.multi.yml new file mode 100644 index 0000000000..87d4458662 --- /dev/null +++ b/docker-compose.multi.yml @@ -0,0 +1,18 @@ +services: + graphify-mcp: + build: . + environment: + GRAPHIFY_API_KEY: ${GRAPHIFY_API_KEY:?GRAPHIFY_API_KEY must be set} + ports: + - "127.0.0.1:8080:8080" + volumes: + - ./repos:/repos:ro + command: + - "--graphs-dir" + - "/repos" + - "--transport" + - "http" + - "--host" + - "0.0.0.0" + - "--port" + - "8080" diff --git a/graphify/serve.py b/graphify/serve.py index a9ecd35405..9db608d19b 100644 --- a/graphify/serve.py +++ b/graphify/serve.py @@ -5,16 +5,20 @@ import os import re import sys +import threading +import weakref from array import array from collections import OrderedDict +from contextvars import ContextVar +from dataclasses import dataclass from pathlib import Path -import threading 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.build import edge_data, edge_datas from graphify.paths import default_graph_json as _default_graph_json +from graphify import paths as _paths try: import jieba as _jieba # type: ignore[import-untyped] @@ -23,16 +27,142 @@ class ToolError(Exception): - """Raised by a tool handler to signal an error result. + """Raised by a tool handler to signal an MCP error result.""" + + +class GraphLoadError(Exception): + """A graph loading failure with text suitable for CLI diagnostics.""" + + +@dataclass +class GraphContext: + name: str + path: Path + graph: nx.Graph + communities: dict[int, list[str]] + mtime: float - A normal string return is sent as an ordinary (successful) text result. A - ToolError is instead turned into a tool result with ``isError: true`` so a - client that only checks ``isError`` can tell a genuine failure — e.g. the - ``gh`` CLI missing or a PR that cannot be resolved — from success. - """ +class GraphRegistry: + def __init__(self) -> None: + self._graphs: dict[str, GraphContext] = {} + self._allow_project_paths = True + self._load_learning_overlay = True + self._lock = threading.Lock() -def _load_graph(graph_path: str) -> nx.Graph: + @classmethod + def from_path(cls, graph_path: Path) -> "GraphRegistry": + reg = cls() + graph_path = Path(graph_path).resolve() + G = _load_graph(str(graph_path)) + _get_trigram_index(G) + communities = _communities_from_graph(G) + name = graph_path.parent.name or "default" + mtime = graph_path.stat().st_mtime + reg._graphs[name] = GraphContext( + name=name, path=graph_path, graph=G, + communities=communities, mtime=mtime, + ) + return reg + + @classmethod + def from_paths(cls, graph_paths: list[Path]) -> "GraphRegistry": + reg = cls() + reg._allow_project_paths = False + reg._load_learning_overlay = False + for path in graph_paths: + graph_path = Path(path).resolve() + name = graph_path.parent.parent.name or "default" + if name in reg._graphs: + raise ValueError(f"duplicate graph name {name!r}") + G = _load_graph(str(graph_path), load_learning_overlay=False) + _get_trigram_index(G) + communities = _communities_from_graph(G) + mtime = graph_path.stat().st_mtime + reg._graphs[name] = GraphContext( + name=name, path=graph_path, graph=G, + communities=communities, mtime=mtime, + ) + return reg + + @classmethod + def from_named_paths(cls, graph_paths: dict[str, Path]) -> "GraphRegistry": + reg = cls() + reg._allow_project_paths = False + reg._load_learning_overlay = False + for name, path in graph_paths.items(): + graph_path = Path(path).resolve() + G = _load_graph_or_raise(str(graph_path), load_learning_overlay=False) + _get_trigram_index(G) + reg._graphs[name] = GraphContext( + name=name, + path=graph_path, + graph=G, + communities=_communities_from_graph(G), + mtime=graph_path.stat().st_mtime, + ) + return reg + + def rescan(self) -> None: + with self._lock: + for name, ctx in list(self._graphs.items()): + try: + s = ctx.path.stat() + except FileNotFoundError: + del self._graphs[name] + continue + except OSError: + continue + + if s.st_mtime != ctx.mtime: + try: + G = _load_graph( + str(ctx.path), load_learning_overlay=self._load_learning_overlay + ) + _get_trigram_index(G) + communities = _communities_from_graph(G) + self._graphs[name] = GraphContext( + name=name, path=ctx.path, graph=G, + communities=communities, mtime=s.st_mtime, + ) + except FileNotFoundError: + del self._graphs[name] + except (SystemExit, Exception): + continue + + def get(self, name: str) -> GraphContext | None: + with self._lock: + return self._graphs.get(name) + + def names(self) -> list[str]: + with self._lock: + return sorted(self._graphs.keys()) + + +def _resolve_graph( + registry: GraphRegistry, + *, + graph: str | None = None, + current: str | None = None, +) -> GraphContext: + name = graph or current + if name is not None: + ctx = registry.get(name) + if ctx is None: + raise ValueError(f"graph {name!r} not found. Available: {registry.names()}") + return ctx + names = registry.names() + if not names: + raise ValueError("no graphs loaded") + if len(names) == 1: + return registry.get(names[0]) + raise ValueError( + f"multiple graphs available ({', '.join(names)}), " + "specify with graph param or use_graph()" + ) + + +def _load_graph_or_raise(graph_path: str, *, load_learning_overlay: bool = True) -> nx.Graph: try: resolved = Path(graph_path).resolve() if resolved.suffix != ".json": @@ -65,19 +195,30 @@ def _load_graph(graph_path: str) -> nx.Graph: except TypeError: G = json_graph.node_link_graph(data) G.graph["_logical_directed"] = _logical_directed - # Attach the work-memory overlay (derived sidecar next to graph.json) so - # the query/MCP read surface can annotate NODE lines display-only. Empty - # when no sidecar exists, leaving un-annotated output byte-identical. - try: - from graphify.reflect import load_learning_overlay as _llo - G.graph["_learning_overlay"] = _llo(resolved) - except Exception: + if load_learning_overlay: + # Attach the work-memory overlay (derived sidecar next to graph.json) + # so the query/MCP read surface can annotate NODE lines display-only. + try: + from graphify.reflect import load_learning_overlay as _llo + G.graph["_learning_overlay"] = _llo(resolved) + except Exception: + G.graph["_learning_overlay"] = {} + else: G.graph["_learning_overlay"] = {} return G except json.JSONDecodeError as exc: - print(f"error: graph.json is corrupted ({exc}). Re-run /graphify to rebuild.", file=sys.stderr) - sys.exit(1) + raise GraphLoadError( + f"graph.json is corrupted ({exc}). Re-run /graphify to rebuild." + ) from exc except (ValueError, FileNotFoundError) as exc: + raise GraphLoadError(str(exc)) from exc + + +def _load_graph(graph_path: str, *, load_learning_overlay: bool = True) -> nx.Graph: + """Load a graph for legacy callers that expect invalid input to exit.""" + try: + return _load_graph_or_raise(graph_path, load_learning_overlay=load_learning_overlay) + except GraphLoadError as exc: print(f"error: {exc}", file=sys.stderr) sys.exit(1) @@ -1373,14 +1514,7 @@ def find_node_ambiguity(G: nx.Graph, label: str) -> list[str]: def _resolve_single_node(G: nx.Graph, label: str) -> tuple[str | None, str | None]: - """Shared node resolution for the get_node / get_neighbors tools. - - Returns ``(node_id, None)`` when *label* resolves to a single winner via the - tiered `_find_node` ranking, or ``(None, message)`` when there is no match or - the winning tier spans several source files. Routing both tools through this - keeps get_node from silently returning a `G.nodes()` iteration-order match for - a hub name while get_neighbors reports the same lookup as ambiguous (#ADR-0001). - """ + """Resolve a node only when its best match is unambiguous.""" matches = _find_node(G, label) if not matches: return None, f"No node matching '{label}' found." @@ -1540,14 +1674,25 @@ def _community_header(cid: int, community_name) -> str: return base -def _build_server(graph_path: str): +def _build_server( + graph_path_or_registry: str | GraphRegistry, + *, + session_state: dict | None = None, +): """Build the configured low-level MCP Server (shared by every transport). All graph query tools and resources are registered here over a single ``mcp.server.Server`` instance; the caller picks the transport (stdio or - Streamable HTTP) and runs it. Hot-reload of graph.json works the same way - regardless of transport, since reloads happen inside the tool handlers. + Streamable HTTP) and runs it. Graph resolution goes through the registry, + supporting both single-graph and multi-graph deployments. + + Returns ``(server, handlers)`` — the handlers dict is exposed for testing. """ + registry = ( + GraphRegistry.from_path(Path(graph_path_or_registry)) + if isinstance(graph_path_or_registry, str) + else graph_path_or_registry + ) try: from mcp.server import Server from mcp import types @@ -1560,52 +1705,53 @@ def _build_server(graph_path: str): # AnyUrl (pydantic is an mcp dependency, so this import cannot miss). from pydantic import AnyUrl - from graphify import paths as _paths - - # Graph contexts comprise one pinned configured default plus a bounded LRU - # of project_path graphs. This preserves the configured graph's warm index - # while preventing a shared server from retaining every project it serves. - _default_graph_path = str(Path(graph_path).resolve()) + session_states: weakref.WeakKeyDictionary = weakref.WeakKeyDictionary() + fixed_session_state = session_state + fallback_session_state: dict = {} + active_session: ContextVar[object | None] = ContextVar("graphify_mcp_session", default=None) + is_multi = len(registry.names()) > 1 _ctx_cache = _GraphContextCache(_max_server_contexts()) + default_paths = { + str(ctx.path.resolve()) + for name in registry.names() + if (ctx := registry.get(name)) is not None + } - def _load_ctx(path: str): - """Return the current default or project graph context as a tool error. - - Unlike ``_load_graph``, this never lets a missing or corrupt client - graph terminate the MCP process; it raises so other projects remain - available on the same server. - """ - resolved_path = str(Path(path).resolve()) - return _ctx_cache.load(resolved_path, pinned=resolved_path == _default_graph_path) - - def _resolve_graph_path(project_path) -> str: - """Map an optional project_path to a concrete graph.json path. ``None`` - keeps the server's default graph (backward-compatible); a project_path - resolves to ``//graph.json``, honouring the - GRAPHIFY_OUT override so worktree/shared-output setups keep working.""" - if not project_path: - return _default_graph_path - return str(Path(project_path) / _paths.GRAPHIFY_OUT / "graph.json") - - # Active per-request context, rebound by _select_graph() and read by the tool - # handlers below. No lock needed on the hot path: _select_graph and the - # handler run in one synchronous stretch of each call_tool coroutine (no - # await between them), so a concurrent call never observes a half-applied - # swap. - active_graph_path = _default_graph_path - try: - G, communities = _load_ctx(_default_graph_path) - except (FileNotFoundError, RuntimeError): - # No default graph at startup → run as a pure multi-project server. Tools - # then require project_path; a call without one gets a clear error rather - # than the process refusing to start (which is what _load_graph would do). - G, communities = None, {} - - def _select_graph(project_path) -> None: - nonlocal G, communities, active_graph_path - path = _resolve_graph_path(project_path) - G, communities = _load_ctx(path) - active_graph_path = str(Path(path).resolve()) + def _current_session_state() -> dict: + if fixed_session_state is not None: + return fixed_session_state + session = active_session.get() + if session is not None: + # MCP 2.x creates a ServerSession for each callback but reuses its + # connection for the lifetime of the client HTTP session. + return session_states.setdefault(getattr(session, "_connection", session), {}) + try: + session = server.request_context.session + except (AttributeError, LookupError): + return fallback_session_state + return session_states.setdefault(session, {}) + + def _get_ctx(arguments: dict) -> GraphContext: + registry.rescan() + graph_param = arguments.pop("graph", None) + project_path = arguments.pop("project_path", None) + if project_path: + if not registry._allow_project_paths: + raise ValueError("project_path is not supported by an explicit --mcp registry") + path = str((Path(project_path) / _paths.GRAPHIFY_OUT / "graph.json").resolve()) + graph, communities = _ctx_cache.load(path, pinned=path in default_paths) + return GraphContext( + name=Path(project_path).name or "project", + path=Path(path), + graph=graph, + communities=communities, + mtime=Path(path).stat().st_mtime, + ) + return _resolve_graph( + registry, + graph=graph_param, + current=_current_session_state().get("current_graph"), + ) # NOTE: no decorators here — the handlers below are plain coroutines, # bound to the Server at the END of this function in a version-aware way: @@ -1695,7 +1841,26 @@ async def list_tools() -> list[types.Tool]: "required": ["source", "target"], }, ), - types.Tool( + ] + if is_multi: + _tools.append(types.Tool( + name="list_graphs", + description="List all available knowledge graphs with node/edge/community counts.", + inputSchema={"type": "object", "properties": {}}, + )) + _tools.append(types.Tool( + name="use_graph", + description="Set the default graph for this session.", + inputSchema={ + "type": "object", + "properties": { + "graph": {"type": "string", "description": "Name of the graph to use"}, + }, + "required": ["graph"], + }, + )) + else: + _tools.append(types.Tool( name="list_prs", description=( "List open GitHub PRs with CI status, review state, and graph impact " @@ -1709,8 +1874,8 @@ async def list_tools() -> list[types.Tool]: "repo": {"type": "string", "description": "GitHub repo (owner/repo). Defaults to current repo."}, }, }, - ), - types.Tool( + )) + _tools.append(types.Tool( name="get_pr_impact", description=( "Get detailed graph impact for a specific PR: which files it changes, " @@ -1725,8 +1890,8 @@ async def list_tools() -> list[types.Tool]: }, "required": ["pr_number"], }, - ), - types.Tool( + )) + _tools.append(types.Tool( name="triage_prs", description=( "Return all actionable open PRs (correct base, not stale) with full graph impact data " @@ -1740,31 +1905,36 @@ async def list_tools() -> list[types.Tool]: "repo": {"type": "string", "description": "GitHub repo (owner/repo). Defaults to current repo."}, }, }, - ), - ] - # Multi-project support: every tool accepts an optional project_path. - # Injected here (rather than repeated in 11 literal schemas) so the set - # stays in lockstep as tools are added. Omitting it keeps the historical - # single-graph behaviour, so this is purely additive for existing callers. + )) + # Named graphs and project paths remain optional to preserve existing + # clients while allowing a registry-backed multi-graph server. for _t in _tools: # The constructor accepts the camelCase alias in both majors, but # attribute access is inputSchema on mcp 1.x and input_schema on 2.x. _schema = getattr(_t, "inputSchema", None) if _schema is None: _schema = _t.input_schema - _schema.setdefault("properties", {})["project_path"] = { - "type": "string", - "description": ( - "Absolute path to a project directory containing " - "graphify-out/graph.json. Optional — defaults to the graph " - "this server was started with." - ), - } + if _t.name not in ("list_graphs", "use_graph"): + _schema.setdefault("properties", {})["graph"] = { + "type": "string", + "description": "Target graph name. Overrides session default.", + } + if registry._allow_project_paths and _t.name not in ("list_graphs", "use_graph"): + _schema.setdefault("properties", {})["project_path"] = { + "type": "string", + "description": ( + "Absolute path to a project directory containing " + "graphify-out/graph.json. Optional — defaults to the graph " + "this server was started with." + ), + } return _tools def _tool_query_graph(arguments: dict) -> str: import time as _time from graphify import querylog + ctx = _get_ctx(arguments) + G = ctx.graph question = arguments["question"] mode = arguments.get("mode", "bfs") depth = min(int(arguments.get("depth", 3)), 6) @@ -1778,12 +1948,12 @@ def _tool_query_graph(arguments: dict) -> str: depth=depth, token_budget=budget, context_filters=context_filter, - graph_path=str(active_graph_path), + graph_path=str(ctx.path), ) querylog.log_query( kind="mcp_query", question=question, - corpus=str(active_graph_path), + corpus=str(ctx.path), result=result, mode=mode, depth=depth, @@ -1793,6 +1963,8 @@ def _tool_query_graph(arguments: dict) -> str: return result def _tool_get_node(arguments: dict) -> str: + ctx = _get_ctx(arguments) + G = ctx.graph label = arguments["label"].lower() nid, err = _resolve_single_node(G, label) if err: @@ -1815,11 +1987,24 @@ def _tool_get_node(arguments: dict) -> str: ]) def _tool_get_neighbors(arguments: dict) -> str: + ctx = _get_ctx(arguments) + G = ctx.graph label = arguments["label"].lower() rel_filter = arguments.get("relation_filter", "").lower() - nid, err = _resolve_single_node(G, label) - if err: - return err + matches = _find_node(G, label) + if not matches: + return f"No node matching '{label}' found." + rivals = find_node_ambiguity(G, label) + if rivals: + listing = "\n".join( + f" {G.nodes[r].get('source_file') or r}\n id: {r}" for r in rivals + ) + return ( + f"Ambiguous: '{label}' matches {len(rivals)} nodes in different files.\n" + f"{listing}\n" + "Retry with the repo-relative path or the full node id." + ) + nid = matches[0] lines = [f"Neighbors of {sanitize_label(G.nodes[nid].get('label', nid))}:"] def _edge_at(d: dict) -> str: # Edge location = the relation SITE (call/import line) in the source @@ -1853,6 +2038,9 @@ def _edge_at(d: dict) -> str: ) def _tool_get_community(arguments: dict) -> str: + ctx = _get_ctx(arguments) + G = ctx.graph + communities = ctx.communities cid = int(arguments["community_id"]) nodes = communities.get(cid, []) if not nodes: @@ -1873,12 +2061,17 @@ def _tool_get_community(arguments: dict) -> str: def _tool_god_nodes(arguments: dict) -> str: from graphify.analyze import god_nodes as _god_nodes + ctx = _get_ctx(arguments) + G = ctx.graph nodes = _god_nodes(G, top_n=int(arguments.get("top_n", 10))) lines = ["God nodes (most connected):"] lines += [f" {i}. {n['label']} - {n['degree']} edges" for i, n in enumerate(nodes, 1)] return "\n".join(lines) - def _tool_graph_stats(_: dict) -> str: + def _tool_graph_stats(arguments: dict) -> str: + ctx = _get_ctx(arguments) + G = ctx.graph + communities = ctx.communities confs = [d.get("confidence", "EXTRACTED") for _, _, d in G.edges(data=True)] total = len(confs) or 1 return ( @@ -1891,9 +2084,12 @@ def _tool_graph_stats(_: dict) -> str: ) def _tool_shortest_path(arguments: dict) -> str: + ctx = _get_ctx(arguments) + G = ctx.graph return _shortest_path_text(G, arguments) def _tool_list_prs(arguments: dict) -> str: + arguments.pop("graph", None) # list_prs doesn't route to a graph from graphify.prs import fetch_prs, fetch_worktrees, format_prs_text, _detect_default_branch repo = arguments.get("repo") or None base = arguments.get("base") or _detect_default_branch(repo) @@ -1907,6 +2103,8 @@ def _tool_list_prs(arguments: dict) -> str: return format_prs_text(prs, base) def _tool_get_pr_impact(arguments: dict) -> str: + ctx = _get_ctx(arguments) + G = ctx.graph from graphify.prs import fetch_pr_files, compute_pr_impact, _gh, _parse_ci number = int(arguments["pr_number"]) repo = arguments.get("repo") or None @@ -1937,6 +2135,8 @@ def _tool_get_pr_impact(arguments: dict) -> str: return "\n".join(lines) def _tool_triage_prs(arguments: dict) -> str: + ctx = _get_ctx(arguments) + G = ctx.graph from concurrent.futures import ThreadPoolExecutor, as_completed from graphify.prs import fetch_prs, fetch_worktrees, fetch_pr_files, compute_pr_impact, _STATUS_ORDER, _detect_default_branch repo = arguments.get("repo") or None @@ -1978,7 +2178,33 @@ def _tool_triage_prs(arguments: dict) -> str: ) return "\n\n".join(lines) - _handlers = { + def _tool_list_graphs(arguments: dict) -> str: + registry.rescan() + lines = [] + for name in registry.names(): + ctx = registry.get(name) + if ctx is None: + continue + G = ctx.graph + lines.append( + f"{name}: {G.number_of_nodes()} nodes, " + f"{G.number_of_edges()} edges, " + f"{len(ctx.communities)} communities" + ) + if not lines: + return "No graphs loaded." + return "\n".join(lines) + + def _tool_use_graph(arguments: dict) -> str: + registry.rescan() + name = arguments["graph"] + ctx = registry.get(name) + if ctx is None: + return f"Graph {name!r} not found. Available: {registry.names()}" + _current_session_state()["current_graph"] = name + return f"Switched to graph {name!r} ({ctx.graph.number_of_nodes()} nodes)" + + _handlers: dict = { "query_graph": _tool_query_graph, "get_node": _tool_get_node, "get_neighbors": _tool_get_neighbors, @@ -1986,19 +2212,24 @@ def _tool_triage_prs(arguments: dict) -> str: "god_nodes": _tool_god_nodes, "graph_stats": _tool_graph_stats, "shortest_path": _tool_shortest_path, - "list_prs": _tool_list_prs, - "get_pr_impact": _tool_get_pr_impact, - "triage_prs": _tool_triage_prs, } + if is_multi: + _handlers["list_graphs"] = _tool_list_graphs + _handlers["use_graph"] = _tool_use_graph + else: + _handlers["list_prs"] = _tool_list_prs + _handlers["get_pr_impact"] = _tool_get_pr_impact + _handlers["triage_prs"] = _tool_triage_prs def _load_community_labels() -> dict[int, str]: - labels_path = Path(active_graph_path).parent / ".graphify_labels.json" + ctx = _resolve_graph(registry, current=_current_session_state().get("current_graph")) + labels_path = ctx.path.parent / ".graphify_labels.json" if labels_path.exists(): try: return {int(k): v for k, v in json.loads(labels_path.read_text(encoding="utf-8")).items()} except Exception: pass - return {cid: f"Community {cid}" for cid in communities} + return {cid: f"Community {cid}" for cid in ctx.communities} async def list_resources() -> list[types.Resource]: # Plain-string URIs on purpose: mcp 1.x types the field as AnyUrl and @@ -2013,7 +2244,11 @@ async def list_resources() -> list[types.Resource]: ] async def read_resource(uri: AnyUrl) -> str: - _select_graph(None) # resources read the server's default graph + registry.rescan() + ctx = _resolve_graph(registry, current=_current_session_state().get("current_graph")) + G = ctx.graph + communities = ctx.communities + active_graph_path = str(ctx.path) uri_str = str(uri) if uri_str == "graphify://report": report_path = Path(active_graph_path).parent / "GRAPH_REPORT.md" @@ -2021,9 +2256,22 @@ async def read_resource(uri: AnyUrl) -> str: return report_path.read_text(encoding="utf-8") return "GRAPH_REPORT.md not found. Run graphify extract first." if uri_str == "graphify://stats": - return _tool_graph_stats({}) + confs = [d.get("confidence", "EXTRACTED") for _, _, d in G.edges(data=True)] + total = len(confs) or 1 + return ( + f"Nodes: {G.number_of_nodes()}\n" + f"Edges: {G.number_of_edges()}\n" + f"Communities: {len(communities)}\n" + f"EXTRACTED: {round(confs.count('EXTRACTED')/total*100)}%\n" + f"INFERRED: {round(confs.count('INFERRED')/total*100)}%\n" + f"AMBIGUOUS: {round(confs.count('AMBIGUOUS')/total*100)}%\n" + ) if uri_str == "graphify://god-nodes": - return _tool_god_nodes({"top_n": 10}) + from graphify.analyze import god_nodes as _god_nodes + nodes = _god_nodes(G, top_n=10) + lines = ["God nodes (most connected):"] + lines += [f" {i}. {n['label']} - {n['degree']} edges" for i, n in enumerate(nodes, 1)] + return "\n".join(lines) if uri_str == "graphify://surprises": try: from graphify.analyze import surprising_connections @@ -2065,17 +2313,12 @@ async def read_resource(uri: AnyUrl) -> str: async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: arguments = dict(arguments or {}) - project_path = arguments.pop("project_path", None) handler = _handlers.get(name) if not handler: return [types.TextContent(type="text", text=f"Unknown tool: {name}")] try: - _select_graph(project_path) # bind G/communities to the target graph return [types.TextContent(type="text", text=handler(arguments))] except ToolError: - # A handler-signalled error: propagate so the result is marked - # isError:true (the mcp 1.x decorator wraps a raised exception into - # an error result; the 2.x path catches it in _on_call_tool). raise except Exception as exc: return [types.TextContent(type="text", text=f"Error executing {name}: {exc}")] @@ -2096,6 +2339,7 @@ async def _on_list_tools(ctx, params) -> types.ListToolsResult: return types.ListToolsResult(tools=await list_tools()) async def _on_call_tool(ctx, params) -> types.CallToolResult: + session_token = active_session.set(ctx.session) try: content = await call_tool(params.name, dict(params.arguments or {})) except ToolError as exc: @@ -2103,13 +2347,19 @@ async def _on_call_tool(ctx, params) -> types.CallToolResult: content=[types.TextContent(type="text", text=str(exc))], isError=True, ) + finally: + active_session.reset(session_token) return types.CallToolResult(content=content) async def _on_list_resources(ctx, params) -> types.ListResourcesResult: return types.ListResourcesResult(resources=await list_resources()) async def _on_read_resource(ctx, params) -> types.ReadResourceResult: - text = await read_resource(params.uri) + session_token = active_session.set(ctx.session) + try: + text = await read_resource(params.uri) + finally: + active_session.reset(session_token) mime = "text/markdown" if str(params.uri).startswith("graphify://report") else "text/plain" return types.ReadResourceResult( contents=[types.TextResourceContents(uri=params.uri, mimeType=mime, text=text)] @@ -2129,10 +2379,10 @@ async def _on_read_resource(ctx, params) -> types.ReadResourceResult: on_read_resource=_on_read_resource, ) - return server + return server, _handlers -def serve(graph_path: str | None = None) -> None: +def serve(graph_path: str | None = None, *, registry: GraphRegistry | None = None) -> None: """Start the MCP server over stdio (the default, per-developer transport).""" graph_path = graph_path or _default_graph_json() try: @@ -2141,7 +2391,9 @@ def serve(graph_path: str | None = None) -> None: raise ImportError('mcp not installed. Run: pip install "graphifyy[mcp]"') from e import asyncio - server = _build_server(graph_path) + if registry is None: + registry = GraphRegistry.from_path(Path(graph_path)) + server, _ = _build_server(registry) async def main() -> None: async with stdio_server() as streams: @@ -2209,8 +2461,9 @@ async def __call__(self, scope, receive, send) -> None: def _build_http_app( - graph_path: str, + graph_path: str | None = None, *, + registry: GraphRegistry | None = None, host: str = "127.0.0.1", port: int = 8080, api_key: str | None = None, @@ -2248,7 +2501,9 @@ def _build_http_app( # mistaken for "auth on" — normalize it to None so the gate is unambiguous. api_key = (api_key or "").strip() or None - server = _build_server(graph_path) + if registry is None: + registry = GraphRegistry.from_path(Path(graph_path)) + server, _ = _build_server(registry) # DNS-rebinding protection. When the operator binds a wildcard address they # are intentionally exposing the server, so accept any Host header; for a @@ -2290,9 +2545,15 @@ async def lifespan(_app): ) +def _validate_http_bind(host: str, api_key: str | None) -> None: + if host not in {"127.0.0.1", "::1", "localhost"} and not api_key: + raise ValueError("HTTP binding outside loopback requires --api-key") + + def serve_http( graph_path: str | None = None, *, + registry: GraphRegistry | None = None, host: str = "127.0.0.1", port: int = 8080, api_key: str | None = None, @@ -2311,8 +2572,12 @@ def serve_http( check (``Authorization: Bearer `` or ``X-API-Key: ``). OAuth is a deliberate follow-up. Binding ``0.0.0.0`` exposes the server beyond localhost — set an api_key when you do. + + ``registry`` can be supplied directly for multi-graph mode; when provided + ``graph_path`` is ignored. """ - graph_path = graph_path or _default_graph_json() + if registry is None: + graph_path = graph_path or _default_graph_json() try: import uvicorn except ImportError as e: @@ -2322,9 +2587,11 @@ def serve_http( ) from e api_key = (api_key or "").strip() or None + _validate_http_bind(host, api_key) app = _build_http_app( graph_path, + registry=registry, host=host, port=port, api_key=api_key, @@ -2348,6 +2615,32 @@ def serve_http( uvicorn.run(app, host=host, port=port) +def _discover_graphs(root: Path) -> dict[str, Path]: + root = root.resolve() + if not root.is_dir(): + raise ValueError(f"graphs directory not found: {root}") + + def raise_walk_error(exc: OSError) -> None: + raise exc + + graphs: dict[str, Path] = {} + for current, directories, filenames in os.walk(root, followlinks=False, onerror=raise_walk_error): + current_path = Path(current) + directories[:] = [ + name for name in directories + if not name.startswith(".") and not (current_path / name).is_symlink() + ] + if current_path.name != "graphify-out" or "graph.json" not in filenames: + continue + graph_path = current_path / "graph.json" + if graph_path.is_symlink(): + continue + graphs[current_path.parent.relative_to(root).as_posix()] = graph_path + if not graphs: + raise ValueError(f"no graphify-out/graph.json files found under: {root}") + return dict(sorted(graphs.items())) + + def _main(argv: list[str] | None = None) -> None: import argparse import os @@ -2369,6 +2662,11 @@ def _main(argv: list[str] | None = None) -> None: metavar="PATH", help="Path to graph.json — alias for the positional argument", ) + parser.add_argument( + "--graphs-dir", + metavar="PATH", + help="Recursively serve graphify-out/graph.json files below PATH", + ) parser.add_argument( "--transport", choices=["stdio", "http"], @@ -2400,8 +2698,30 @@ def _main(argv: list[str] | None = None) -> None: help="Reap stateful sessions idle this many seconds (default: 3600; 0 disables)", ) args = parser.parse_args(argv) - graph_path = args.graph_flag or args.graph_path or _default_graph_json() + if args.graphs_dir is not None: + if args.graph_path is not None or args.graph_flag is not None: + parser.error("--graphs-dir cannot be combined with a graph path") + try: + registry = GraphRegistry.from_named_paths(_discover_graphs(Path(args.graphs_dir))) + except (OSError, ValueError, GraphLoadError) as exc: + parser.error(str(exc)) + if args.transport == "http": + serve_http( + registry=registry, + host=args.host, + port=args.port, + api_key=args.api_key, + path=args.path, + json_response=args.json_response, + stateless=args.stateless, + session_timeout=args.session_timeout, + ) + else: + serve(registry=registry) + return + + graph_path = args.graph_flag or args.graph_path or _default_graph_json() if args.transport == "http": serve_http( graph_path, diff --git a/tests/test_mcp_cli.py b/tests/test_mcp_cli.py new file mode 100644 index 0000000000..4ece8898e1 --- /dev/null +++ b/tests/test_mcp_cli.py @@ -0,0 +1,175 @@ +import os +from pathlib import Path +import re +import shutil +import subprocess + +import pytest + + +def _legacy_multi_graph_terms(): + return ( + "--multi" + "-mcp", + "GRAPHS" + "_DIR", + "SCAN" + "_INTERVAL", + "graphify" + "-multi", + ) + + +def test_no_tracked_multi_compose_references(): + tracked = subprocess.run( + ["git", "ls-files"], capture_output=True, check=True, text=True + ).stdout.splitlines() + tracked.remove("tests/test_mcp_cli.py") + tracked = [ + path + for path in tracked + if path != "docker-compose.multi.yml" + and path != "README.md" + and not path.startswith("docs/superpowers/") + ] + result = subprocess.run( + ["git", "grep", "-n", "docker-compose.multi.yml", "--", *tracked], + capture_output=True, + text=True, + ) + assert result.returncode == 1, result.stdout + +def test_dockerfile_uses_the_mcp_server_entrypoint(): + dockerfile = Path("Dockerfile").read_text(encoding="utf-8") + + assert 'ENTRYPOINT ["python", "-m", "graphify.serve"]' in dockerfile + assert 'CMD ["--graphs-dir", "/data", "--transport", "http", "--host", "0.0.0.0", "--port", "8080"]' in dockerfile + + +def test_dockerignore_excludes_local_env_files_but_keeps_example(): + dockerignore = Path(".dockerignore").read_text(encoding="utf-8").splitlines() + + assert ".env" in dockerignore + assert ".env.*" in dockerignore + assert "!.env.example" in dockerignore + + +def test_multi_repo_compose_requires_a_nonempty_api_key(): + if shutil.which("docker") is None: + pytest.skip("Docker is not installed") + + compose_version = subprocess.run( + ["docker", "compose", "version"], capture_output=True, text=True + ) + if compose_version.returncode: + pytest.skip("Docker Compose is not available through subprocess") + + env = os.environ.copy() + env.pop("GRAPHIFY_API_KEY", None) + command = [ + "docker", + "compose", + "-f", + "docker-compose.multi.yml", + "config", + ] + probe = subprocess.run( + command, capture_output=True, text=True, env={**env, "GRAPHIFY_API_KEY": "test-key"} + ) + + unset = subprocess.run(command, capture_output=True, text=True, env=env) + assert unset.returncode != 0 + + empty = subprocess.run( + command, capture_output=True, text=True, env={**env, "GRAPHIFY_API_KEY": ""} + ) + assert empty.returncode != 0 + + assert probe.returncode == 0, probe.stderr + assert "GRAPHIFY_API_KEY: test-key" in probe.stdout + + +def test_multi_repo_compose_discovers_read_only_repository_mount(): + compose = Path("docker-compose.multi.yml").read_text(encoding="utf-8") + + assert '"127.0.0.1:8080:8080"' in compose + assert "GRAPHIFY_API_KEY: ${GRAPHIFY_API_KEY:?GRAPHIFY_API_KEY must be set}" in compose + assert "./repos:/repos:ro" in compose + assert "./repos/repo-a" not in compose + assert "./repos/repo-b" not in compose + assert '"--graphs-dir"' in compose + assert '"/repos"' in compose + assert '"--mcp"' not in compose + assert '"--transport"' in compose + assert '"http"' in compose + assert '"--host"' in compose + assert '"0.0.0.0"' in compose + assert '"--port"' in compose + assert '"8080"' in compose + + +def test_readme_documents_multi_repo_compose_usage(): + readme = Path("README.md").read_text(encoding="utf-8") + + assert "docker-compose.multi.yml" in readme + assert "GRAPHIFY_API_KEY=your-secret docker compose -f docker-compose.multi.yml up --build" in readme + assert "http://localhost:8080/mcp" in readme + assert "Authorization: Bearer " in readme + assert "graphify-out/graph.json" in readme + assert "read-only" in readme + + +def test_readme_documents_supported_multi_graph_mcp_commands(): + readme = Path("README.md").read_text(encoding="utf-8") + + assert "\npython -m graphify.serve --graphs-dir ..\n" in readme + assert ( + 'python -m graphify.serve --graphs-dir .. --transport http --host 0.0.0.0 ' + '--port 8080 --api-key "$SECRET"' + ) in readme + assert "graphify ../frontend ../backend --mcp" not in readme + assert "/graphify ./raw --mcp" not in readme + assert "python -m graphify.serve ./raw/graphify-out/graph.json" in readme + + +def test_readme_documents_public_mcp_docker_cli(): + readme = Path("README.md").read_text(encoding="utf-8") + command = 'docker run -p 8080:8080 -e GRAPHIFY_API_KEY="$GRAPHIFY_API_KEY" -v "$(pwd):/data:ro" graphify --graphs-dir /data --transport http --host 0.0.0.0' + + assert command in readme + assert "--mcp" not in command + + +def test_rendered_skills_have_no_legacy_multi_mcp_terms(): + from tools.skillgen.gen import load_platforms, render_all + + rendered = render_all(load_platforms()) + forbidden = _legacy_multi_graph_terms() + + assert all(term not in artifact.content for term in forbidden for artifact in rendered) + + +def test_no_legacy_multi_graph_entrypoint_references(): + legacy_terms = _legacy_multi_graph_terms() + tracked_files = subprocess.run( + ["git", "ls-files"], + capture_output=True, + check=True, + text=True, + ).stdout.splitlines() + active_files = [ + path + for path in tracked_files + if not path.startswith("docs/superpowers/") + and not Path(path).name.upper().startswith("CHANGELOG") + ] + result = subprocess.run( + [ + "git", + "grep", + "-nE", + "|".join(re.escape(term) for term in legacy_terms), + "--", + *active_files, + ], + capture_output=True, + text=True, + ) + + assert result.returncode == 1, result.stdout diff --git a/tests/test_serve.py b/tests/test_serve.py index 87e71f821d..ffa71a6827 100644 --- a/tests/test_serve.py +++ b/tests/test_serve.py @@ -6,6 +6,8 @@ import networkx as nx from networkx.readwrite import json_graph +from pathlib import Path + from graphify.serve import ( _strip_diacritics, _communities_from_graph, @@ -33,6 +35,9 @@ _community_header, _search_tokens, _shortest_path_text, + GraphContext, + GraphRegistry, + _resolve_graph, ) @@ -1714,9 +1719,6 @@ def test_underscore_query_does_not_let_a_single_token_outrank_the_real_match(): def test_resolve_single_node_shared_by_get_node_and_get_neighbors(): - """ADR-0001 finding 1: the resolver both tools now use returns an Ambiguous - message when the winning tier spans multiple files, a clean node id for a - unique label, and a not-found message otherwise.""" from graphify.serve import _resolve_single_node G = nx.Graph() @@ -1735,3 +1737,482 @@ def test_resolve_single_node_shared_by_get_node_and_get_neighbors(): nid, err = _resolve_single_node(G, "nonexistent") assert nid is None assert "No node matching" in err + + +# --- GraphRegistry tests --- + + +def _write_registry_graph(base: Path, name: str, nodes=None, edges=None): + """Write a minimal graph.json under base/name/graph.json.""" + d = base / name + d.mkdir(parents=True, exist_ok=True) + data = { + "nodes": [ + {"id": f"{name}_n1", "label": "main", "community": 0}, + {"id": f"{name}_n2", "label": "helper", "community": 0}, + ] if nodes is None else nodes, + "links": [ + {"source": f"{name}_n1", "target": f"{name}_n2", "relation": "calls", "confidence": "EXTRACTED"}, + ] if edges is None else edges, + } + (d / "graph.json").write_text(json.dumps(data), encoding="utf-8") + return d / "graph.json" + + +class TestGraphRegistry: + def test_from_path_single_graph(self, tmp_path): + gp = _write_registry_graph(tmp_path, "proj") + reg = GraphRegistry.from_path(gp) + assert len(reg.names()) == 1 + ctx = reg.get(reg.names()[0]) + assert ctx is not None + assert ctx.graph.number_of_nodes() == 2 + + def test_from_paths_loads_multiple_graphs(self, tmp_path): + frontend = _write_registry_graph(tmp_path / "frontend", "graphify-out") + backend = _write_registry_graph(tmp_path / "backend", "graphify-out") + reg = GraphRegistry.from_paths([frontend, backend]) + assert sorted(reg.names()) == ["backend", "frontend"] + + def test_from_paths_rejects_duplicate_repository_names(self, tmp_path): + first = _write_registry_graph(tmp_path / "one" / "api", "graphify-out") + second = _write_registry_graph(tmp_path / "two" / "api", "graphify-out") + + with pytest.raises(ValueError, match="duplicate graph name 'api'"): + GraphRegistry.from_paths([first, second]) + + def test_get_unknown_returns_none(self, tmp_path): + gp = _write_registry_graph(tmp_path, "proj") + reg = GraphRegistry.from_path(gp) + assert reg.get("nonexistent") is None + + def test_registry_reads_acquire_lock(self, tmp_path): + class TrackingLock: + entered = 0 + + def __enter__(self): + self.entered += 1 + + def __exit__(self, *args): + pass + + gp = _write_registry_graph(tmp_path, "proj") + reg = GraphRegistry.from_path(gp) + lock = TrackingLock() + reg._lock = lock + + reg.get("proj") + reg.names() + + assert lock.entered == 2 + + def test_from_path_hot_reload(self, tmp_path): + gp = _write_registry_graph(tmp_path, "proj", nodes=[ + {"id": "a", "label": "a", "community": 0}, + ], edges=[]) + reg = GraphRegistry.from_path(gp) + assert reg.get(reg.names()[0]).graph.number_of_nodes() == 1 + import time; time.sleep(0.05) + _write_registry_graph(tmp_path, "proj", nodes=[ + {"id": "a", "label": "a", "community": 0}, + {"id": "b", "label": "b", "community": 0}, + ], edges=[]) + reg.rescan() + assert reg.get(reg.names()[0]).graph.number_of_nodes() == 2 + + def test_from_path_keeps_graph_on_transient_stat_failure(self, tmp_path, monkeypatch): + graph_path = _write_registry_graph(tmp_path, "proj") + registry = GraphRegistry.from_path(graph_path) + original_stat = Path.stat + + def fail_graph_stat(path): + if path == graph_path: + raise PermissionError("temporarily unavailable") + return original_stat(path) + + monkeypatch.setattr(Path, "stat", fail_graph_stat) + + registry.rescan() + + assert registry.names() == ["proj"] + + def test_from_path_preserves_graph_when_reload_is_malformed(self, tmp_path): + graph_path = _write_registry_graph(tmp_path, "proj", nodes=[ + {"id": "a", "label": "original", "community": 0}, + ], edges=[]) + registry = GraphRegistry.from_path(graph_path) + _, handlers = _build_server(registry) + + graph_path.write_text("{malformed", encoding="utf-8") + import os + original_mtime = graph_path.stat().st_mtime + os.utime(graph_path, (original_mtime + 1, original_mtime + 1)) + + registry.rescan() + + assert "original" in handlers["get_node"]({"label": "original"}) + + def test_from_path_evicts_graph_when_reload_file_is_missing(self, tmp_path, monkeypatch): + graph_path = _write_registry_graph(tmp_path, "proj") + registry = GraphRegistry.from_path(graph_path) + original_mtime = graph_path.stat().st_mtime + + import os + os.utime(graph_path, (original_mtime + 1, original_mtime + 1)) + + def fail_graph_load(path, **kwargs): + if Path(path) == graph_path: + raise FileNotFoundError(path) + return _load_graph(path, **kwargs) + + monkeypatch.setattr("graphify.serve._load_graph", fail_graph_load) + + registry.rescan() + + assert registry.names() == [] + + def test_from_paths_reloads_graph_only_before_each_request(self, tmp_path, monkeypatch): + graph_path = _write_registry_graph( + tmp_path / "repo", "graphify-out", nodes=[ + {"id": "a", "label": "a", "community": 0}, + ], edges=[], + ) + registry = GraphRegistry.from_paths([graph_path]) + _, handlers = _build_server(registry) + from graphify import reflect + + monkeypatch.setattr( + reflect, + "load_learning_overlay", + lambda path: pytest.fail("explicit registry reload must not load sidecars"), + ) + _write_registry_graph( + tmp_path / "repo", "graphify-out", nodes=[ + {"id": "a", "label": "a", "community": 0}, + {"id": "b", "label": "b", "community": 0}, + ], edges=[], + ) + import os + original_mtime = graph_path.stat().st_mtime + os.utime(graph_path, (original_mtime + 1, original_mtime + 1)) + + assert "Nodes: 2" in handlers["graph_stats"]({}) + + def test_registry_has_no_directory_discovery_api(self): + assert not hasattr(GraphRegistry, "from_" + "directory") + + def test_from_path_evicts_deleted(self, tmp_path): + gp = _write_registry_graph(tmp_path, "proj") + reg = GraphRegistry.from_path(gp) + assert len(reg.names()) == 1 + import os + os.remove(gp) + reg.rescan() + assert reg.names() == [] + + def test_from_paths_loads_graph_json_without_learning_overlay(self, tmp_path, monkeypatch): + repo = tmp_path / "repo" + graph_path = repo / "graphify-out" / "graph.json" + graph_path.parent.mkdir(parents=True) + graph_path.write_text('{"nodes": [], "links": []}', encoding="utf-8") + from graphify import reflect + + monkeypatch.setattr( + reflect, + "load_learning_overlay", + lambda path: pytest.fail("explicit registry must not load sidecars"), + ) + + registry = GraphRegistry.from_paths([graph_path]) + + assert registry.names() == ["repo"] + + + +from graphify.serve import _build_server + + +class TestUnifiedBuildServer: + def test_legacy_graph_path_builds_single_graph_server(self, tmp_path): + graph_path = _write_registry_graph(tmp_path, "proj") + + _, handlers = _build_server(str(graph_path)) + + assert "list_prs" in handlers + assert "list_graphs" not in handlers + assert "Nodes: 2" in handlers["graph_stats"]({}) + + def test_single_graph_has_no_list_graphs(self, tmp_path): + gp = _write_registry_graph(tmp_path, "proj") + reg = GraphRegistry.from_path(gp) + server, handlers = _build_server(reg) + assert "list_graphs" not in handlers + assert "use_graph" not in handlers + + def test_multi_graph_has_list_graphs(self, tmp_path): + alpha = _write_registry_graph(tmp_path / "alpha", "graphify-out") + beta = _write_registry_graph(tmp_path / "beta", "graphify-out") + reg = GraphRegistry.from_paths([alpha, beta]) + server, handlers = _build_server(reg) + assert "list_graphs" in handlers + assert "use_graph" in handlers + + def test_single_graph_has_pr_tools(self, tmp_path): + gp = _write_registry_graph(tmp_path, "proj") + reg = GraphRegistry.from_path(gp) + server, handlers = _build_server(reg) + assert "list_prs" in handlers + + def test_multi_graph_no_pr_tools(self, tmp_path): + alpha = _write_registry_graph(tmp_path / "alpha", "graphify-out") + beta = _write_registry_graph(tmp_path / "beta", "graphify-out") + reg = GraphRegistry.from_paths([alpha, beta]) + server, handlers = _build_server(reg) + assert "list_prs" not in handlers + + def test_single_graph_query(self, tmp_path): + _write_registry_graph(tmp_path, "proj", nodes=[ + {"id": "n1", "label": "AuthService", "community": 0, "source_file": "auth.py"}, + {"id": "n2", "label": "Database", "community": 0, "source_file": "db.py"}, + ], edges=[ + {"source": "n1", "target": "n2", "relation": "calls", "confidence": "EXTRACTED"}, + ]) + reg = GraphRegistry.from_path(tmp_path / "proj" / "graph.json") + _, handlers = _build_server(reg) + result = handlers["query_graph"]({"question": "AuthService"}) + assert "AuthService" in result + + def test_multi_graph_use_graph_and_query(self, tmp_path): + alpha = _write_registry_graph(tmp_path / "alpha", "graphify-out", nodes=[ + {"id": "a1", "label": "UserService", "community": 0}, + ], edges=[]) + beta = _write_registry_graph(tmp_path / "beta", "graphify-out", nodes=[ + {"id": "b1", "label": "PaymentGateway", "community": 0}, + ], edges=[]) + reg = GraphRegistry.from_paths([alpha, beta]) + session = {} + _, handlers = _build_server(reg, session_state=session) + + result = handlers["list_graphs"]({}) + assert "alpha" in result + assert "beta" in result + + handlers["use_graph"]({"graph": "alpha"}) + assert session["current_graph"] == "alpha" + + result = handlers["graph_stats"]({}) + assert "Nodes: 1" in result + + result = handlers["graph_stats"]({"graph": "beta"}) + assert "Nodes: 1" in result + + def test_all_tools_have_graph_param(self, tmp_path): + alpha = _write_registry_graph(tmp_path / "alpha", "graphify-out") + beta = _write_registry_graph(tmp_path / "beta", "graphify-out") + reg = GraphRegistry.from_paths([alpha, beta]) + server, _ = _build_server(reg) + from mcp import types as _t + import asyncio + handler = server.request_handlers[_t.ListToolsRequest] + loop = asyncio.new_event_loop() + result = loop.run_until_complete(handler(_t.ListToolsRequest())) + loop.close() + tools = result.root.tools + for t in tools: + if t.name != "list_graphs": + props = t.inputSchema.get("properties", {}) + assert "graph" in props, f"tool {t.name} missing graph param" + + def test_explicit_registry_rejects_project_path(self, tmp_path): + explicit_graph = tmp_path / "repo" / "graphify-out" / "graph.json" + explicit_graph.parent.mkdir(parents=True) + explicit_graph.write_text('{"nodes": [], "links": []}', encoding="utf-8") + other_project = tmp_path / "other" + (other_project / "graphify-out").mkdir(parents=True) + (other_project / "graphify-out" / "graph.json").write_text('{"nodes": [], "links": []}', encoding="utf-8") + registry = GraphRegistry.from_paths([explicit_graph]) + _, handlers = _build_server(registry) + + with pytest.raises(ValueError, match="project_path is not supported"): + handlers["graph_stats"]({"project_path": str(other_project)}) + + +class TestResolveGraph: + def test_explicit_param(self, tmp_path): + alpha = _write_registry_graph(tmp_path / "alpha", "graphify-out") + beta = _write_registry_graph(tmp_path / "beta", "graphify-out") + reg = GraphRegistry.from_paths([alpha, beta]) + ctx = _resolve_graph(reg, graph="alpha", current=None) + assert ctx.name == "alpha" + + def test_session_default(self, tmp_path): + graph_path = _write_registry_graph(tmp_path / "proj", "graphify-out") + reg = GraphRegistry.from_paths([graph_path]) + ctx = _resolve_graph(reg, graph=None, current="proj") + assert ctx.name == "proj" + + def test_single_graph_implicit(self, tmp_path): + gp = _write_registry_graph(tmp_path, "only") + reg = GraphRegistry.from_path(gp) + ctx = _resolve_graph(reg, graph=None, current=None) + assert ctx is not None + + def test_ambiguous_raises(self, tmp_path): + first = _write_registry_graph(tmp_path / "a", "graphify-out") + second = _write_registry_graph(tmp_path / "b", "graphify-out") + reg = GraphRegistry.from_paths([first, second]) + with pytest.raises(ValueError, match="multiple graphs"): + _resolve_graph(reg, graph=None, current=None) + + def test_unknown_name_raises(self, tmp_path): + graph_path = _write_registry_graph(tmp_path / "x", "graphify-out") + reg = GraphRegistry.from_paths([graph_path]) + with pytest.raises(ValueError, match="not found"): + _resolve_graph(reg, graph="nope", current=None) + +class TestMainCLI: + def test_serve_cli_starts_named_registry_for_graphs_dir(self, tmp_path, monkeypatch): + graph_path = _write_registry_graph(tmp_path / "team" / "api", "graphify-out") + captured = {} + monkeypatch.setattr( + "graphify.serve.serve", + lambda graph_path=None, *, registry=None: captured.update(names=registry.names()), + ) + + from graphify.serve import _main + _main(["--graphs-dir", str(tmp_path)]) + + assert captured == {"names": ["team/api"]} + + @pytest.mark.parametrize("graphs_dir", ["missing", "not-a-directory"]) + def test_serve_cli_rejects_missing_or_non_directory_graphs_dir(self, tmp_path, graphs_dir, capsys): + if graphs_dir == "not-a-directory": + (tmp_path / graphs_dir).write_text("not a directory", encoding="utf-8") + + from graphify.serve import _main + + with pytest.raises(SystemExit): + _main(["--graphs-dir", str(tmp_path / graphs_dir)]) + + assert "graphs directory not found" in capsys.readouterr().err + + def test_serve_cli_rejects_graphs_dir_without_a_graph(self, tmp_path, capsys): + from graphify.serve import _main + + with pytest.raises(SystemExit): + _main(["--graphs-dir", str(tmp_path)]) + + assert "no graphify-out/graph.json files found" in capsys.readouterr().err + + def test_serve_cli_ignores_hidden_and_symlinked_graph_directories(self, tmp_path, monkeypatch): + visible = _write_registry_graph(tmp_path / "visible", "graphify-out") + _write_registry_graph(tmp_path / ".hidden", "graphify-out") + (tmp_path / "linked").symlink_to(visible.parent.parent, target_is_directory=True) + captured = {} + monkeypatch.setattr( + "graphify.serve.serve", + lambda graph_path=None, *, registry=None: captured.update(names=registry.names()), + ) + + from graphify.serve import _main + _main(["--graphs-dir", str(tmp_path)]) + + assert captured == {"names": ["visible"]} + + @pytest.mark.parametrize("graph_argument", ["configured.json", "--graph"]) + def test_serve_cli_rejects_graphs_dir_with_graph_path(self, tmp_path, graph_argument, capsys): + from graphify.serve import _main + + argv = ["--graphs-dir", str(tmp_path)] + if graph_argument == "--graph": + argv.extend(["--graph", "configured.json"]) + else: + argv.append(graph_argument) + with pytest.raises(SystemExit): + _main(argv) + + assert "--graphs-dir cannot be combined with a graph path" in capsys.readouterr().err + + def test_serve_cli_forwards_http_options_to_graphs_dir_registry(self, tmp_path, monkeypatch): + _write_registry_graph(tmp_path / "team" / "api", "graphify-out") + captured = {} + monkeypatch.setattr( + "graphify.serve.serve_http", + lambda graph_path=None, **kwargs: captured.update(graph_path=graph_path, **kwargs), + ) + + from graphify.serve import _main + _main([ + "--graphs-dir", str(tmp_path), "--transport", "http", "--host", "0.0.0.0", + "--port", "8080", "--api-key", "secret", "--path", "/registry", + "--json-response", "--stateless", "--session-timeout", "30", + ]) + + assert captured["graph_path"] is None + assert captured["registry"].names() == ["team/api"] + assert {key: value for key, value in captured.items() if key != "registry"} == { + "graph_path": None, + "host": "0.0.0.0", + "port": 8080, + "api_key": "secret", + "path": "/registry", + "json_response": True, + "stateless": True, + "session_timeout": 30.0, + } + + +class TestUnifiedIntegration: + def test_multi_graph_full_flow(self, tmp_path): + alpha = _write_registry_graph(tmp_path / "alpha", "graphify-out", nodes=[ + {"id": "a1", "label": "UserService", "community": 0, "source_file": "user.py", "source_location": "L1", "file_type": "python"}, + {"id": "a2", "label": "AuthService", "community": 0, "source_file": "auth.py", "source_location": "L1", "file_type": "python"}, + ], edges=[ + {"source": "a1", "target": "a2", "relation": "calls", "confidence": "EXTRACTED"}, + ]) + beta = _write_registry_graph(tmp_path / "beta", "graphify-out", nodes=[ + {"id": "b1", "label": "PaymentGateway", "community": 0, "source_file": "pay.py", "source_location": "L1", "file_type": "python"}, + ], edges=[]) + + reg = GraphRegistry.from_paths([alpha, beta]) + session = {} + _, handlers = _build_server(reg, session_state=session) + + listing = handlers["list_graphs"]({}) + assert "alpha" in listing + assert "beta" in listing + + handlers["use_graph"]({"graph": "alpha"}) + assert session["current_graph"] == "alpha" + + result = handlers["query_graph"]({"question": "UserService"}) + assert "UserService" in result + + result = handlers["graph_stats"]({"graph": "beta"}) + assert "Nodes: 1" in result + + assert "list_prs" not in handlers + + def test_single_graph_retro_compat(self, tmp_path): + _write_registry_graph(tmp_path, "proj", nodes=[ + {"id": "n1", "label": "Main", "community": 0, "source_file": "main.py"}, + {"id": "n2", "label": "Helper", "community": 0, "source_file": "helper.py"}, + ], edges=[ + {"source": "n1", "target": "n2", "relation": "calls", "confidence": "EXTRACTED"}, + ]) + reg = GraphRegistry.from_path(tmp_path / "proj" / "graph.json") + _, handlers = _build_server(reg) + + assert "list_graphs" not in handlers + assert "use_graph" not in handlers + assert "list_prs" in handlers + + result = handlers["query_graph"]({"question": "Main"}) + assert "Main" in result + + result = handlers["graph_stats"]({}) + assert "Nodes: 2" in result + + result = handlers["get_node"]({"label": "Helper"}) + assert "Helper" in result + assert "helper.py" in result diff --git a/tests/test_serve_http.py b/tests/test_serve_http.py index a07bbba013..3a81e02bc5 100644 --- a/tests/test_serve_http.py +++ b/tests/test_serve_http.py @@ -7,6 +7,7 @@ from __future__ import annotations import json +import sys from pathlib import Path import pytest @@ -52,6 +53,17 @@ def _graph_file(tmp_path: Path) -> str: return str(p) +def _repository_graph(tmp_path: Path, name: str, node_count: int) -> Path: + graph_path = tmp_path / name / "graphify-out" / "graph.json" + graph_path.parent.mkdir(parents=True) + graph_path.write_text(json.dumps({ + "directed": True, + "nodes": [{"id": f"{name}-{index}", "label": name, "community": 0} for index in range(node_count)], + "edges": [], + }), encoding="utf-8") + return graph_path + + def _client(app) -> TestClient: # Default host is 127.0.0.1, so the DNS-rebinding guard only accepts that # Host header (TestClient otherwise sends the disallowed "testserver"). @@ -200,31 +212,69 @@ def _call_tool(client, headers, name, arguments, rid) -> str: return resp.json()["result"]["content"][0]["text"] -def test_project_path_is_optional_on_every_tool(tmp_path): - """Multi-project support is additive: every tool gains an optional - project_path, and none of them makes it required.""" +def test_graph_param_is_optional_on_every_tool(tmp_path): + """Every tool gains an optional graph param (but list_graphs is exempt since + it sets the graph, not queries one); none makes it required.""" app = serve_mod._build_http_app(_graph_file(tmp_path), json_response=True) with _client(app) as client: headers = _init_session(client) resp = client.post("/mcp", headers=headers, json={"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}) for tool in resp.json()["result"]["tools"]: + if tool["name"] == "list_graphs": + continue props = tool["inputSchema"].get("properties", {}) - assert "project_path" in props, f"{tool['name']} missing project_path" - assert "project_path" not in tool["inputSchema"].get("required", []) + assert "graph" in props, f"{tool['name']} missing graph param" + assert "graph" not in tool["inputSchema"].get("required", []) -def test_project_path_routes_to_that_projects_graph(tmp_path): - """One running server answers against the default graph when project_path is - omitted, and against a project's own graph when it is supplied.""" - proj = _project_with_graph(tmp_path, node_count=3) # default graph has 2 nodes - app = serve_mod._build_http_app(_graph_file(tmp_path), json_response=True) +def test_graph_param_routes_to_that_graph(tmp_path): + """A multi-graph server answers the default graph when graph param is omitted + and a specific graph when graph param is provided.""" + alpha = _repository_graph(tmp_path, "alpha", 1) + beta = _repository_graph(tmp_path, "beta", 3) + registry = serve_mod.GraphRegistry.from_paths([alpha, beta]) + app = serve_mod._build_http_app(registry=registry, json_response=True) + with _client(app) as client: + headers = _init_session(client) + assert "Nodes: 1" in _call_tool(client, headers, "graph_stats", {"graph": "alpha"}, rid=2) + assert "Nodes: 3" in _call_tool(client, headers, "graph_stats", {"graph": "beta"}, rid=3) + + +def test_use_graph_selection_is_isolated_per_http_session(tmp_path): + alpha = _repository_graph(tmp_path, "alpha", 1) + beta = _repository_graph(tmp_path, "beta", 3) + app = serve_mod._build_http_app( + registry=serve_mod.GraphRegistry.from_paths([alpha, beta]), json_response=True + ) + with _client(app) as client: + first = _init_session(client) + second = _init_session(client) + + assert "Switched to graph 'alpha'" in _call_tool(client, first, "use_graph", {"graph": "alpha"}, rid=2) + assert "Switched to graph 'beta'" in _call_tool(client, second, "use_graph", {"graph": "beta"}, rid=3) + assert "Nodes: 1" in _call_tool(client, first, "graph_stats", {}, rid=4) + assert "Nodes: 3" in _call_tool(client, second, "graph_stats", {}, rid=5) + + +def test_mcp_v2_callbacks_keep_session_graph_selection(tmp_path): + """MCP 2.x supplies the session to callbacks instead of Server.request_context.""" + from mcp.server import Server + + if hasattr(Server, "request_context"): + pytest.skip("requires MCP 2.x") + + alpha = _repository_graph(tmp_path, "alpha", 1) + beta = _repository_graph(tmp_path, "beta", 3) + app = serve_mod._build_http_app( + registry=serve_mod.GraphRegistry.from_paths([alpha, beta]), json_response=True + ) with _client(app) as client: headers = _init_session(client) - assert "Nodes: 2" in _call_tool(client, headers, "graph_stats", {}, rid=2) - assert "Nodes: 3" in _call_tool(client, headers, "graph_stats", {"project_path": proj}, rid=3) - # Falling back to the default afterwards still works (no state leak). - assert "Nodes: 2" in _call_tool(client, headers, "graph_stats", {}, rid=4) + assert "Switched to graph 'beta'" in _call_tool( + client, headers, "use_graph", {"graph": "beta"}, rid=2 + ) + assert "Nodes: 3" in _call_tool(client, headers, "graph_stats", {}, rid=3) @pytest.mark.parametrize( @@ -251,46 +301,37 @@ def counting_load(path: str): return original_load(path) monkeypatch.setattr(serve_mod, "_load_graph", counting_load) - projects = [ - _project_with_graph(tmp_path, node_count=i + 3, name=f"project-{i}") - for i in range(3) - ] + projects = [_project_with_graph(tmp_path, node_count=i + 3, name=f"project-{i}") for i in range(3)] default_graph = _graph_file(tmp_path) app = serve_mod._build_http_app(default_graph, json_response=True) with _client(app) as client: headers = _init_session(client) assert "Nodes: 3" in _call_tool(client, headers, "graph_stats", {"project_path": projects[0]}, rid=2) assert "Nodes: 4" in _call_tool(client, headers, "graph_stats", {"project_path": projects[1]}, rid=3) - # A cache hit promotes project-0 above project-1 in LRU recency. assert "Nodes: 3" in _call_tool(client, headers, "graph_stats", {"project_path": projects[0]}, rid=4) assert "Nodes: 5" in _call_tool(client, headers, "graph_stats", {"project_path": projects[2]}, rid=5) - # project-1, not the re-touched project-0, was evicted. assert "Nodes: 4" in _call_tool(client, headers, "graph_stats", {"project_path": projects[1]}, rid=6) - # The configured default graph stays warm even when project capacity is full. assert "Nodes: 2" in _call_tool(client, headers, "graph_stats", {}, rid=7) first_graph = str((Path(projects[0]) / "graphify-out" / "graph.json").resolve()) second_graph = str((Path(projects[1]) / "graphify-out" / "graph.json").resolve()) - default_graph = str(Path(default_graph).resolve()) assert loads[first_graph] == 1 assert loads[second_graph] == 2 - assert loads[default_graph] == 1 -def test_bad_project_path_errors_without_killing_server(tmp_path): - """A missing project graph is a tool error, not a process exit — the server - keeps serving the default graph.""" +def test_bad_graph_param_errors_without_killing_server(tmp_path): + """A bad graph name is a tool error, not a process exit — the server + keeps serving other graphs.""" app = serve_mod._build_http_app(_graph_file(tmp_path), json_response=True) with _client(app) as client: headers = _init_session(client) bad = _call_tool(client, headers, "graph_stats", - {"project_path": str(tmp_path / "does-not-exist")}, rid=2) + {"graph": "does-not-exist"}, rid=2) assert "not found" in bad.lower() assert "Nodes: 2" in _call_tool(client, headers, "graph_stats", {}, rid=3) def test_corrupt_project_graph_is_a_tool_error_without_killing_server(tmp_path): - """A CLI-style SystemExit from a client graph cannot stop the MCP server.""" project = Path(_project_with_graph(tmp_path, node_count=3)) (project / "graphify-out" / "graph.json").write_text("{not json", encoding="utf-8") app = serve_mod._build_http_app(_graph_file(tmp_path), json_response=True) @@ -325,6 +366,18 @@ def test_session_timeout_zero_disables(tmp_path): assert client.post("/mcp", headers=_MCP_HEADERS, json=_INIT_BODY).status_code == 200 +@pytest.mark.parametrize("host", ["0.0.0.0", "192.168.1.10", "::"]) +def test_serve_http_rejects_unauthenticated_non_loopback(host, monkeypatch): + monkeypatch.setitem(sys.modules, "uvicorn", object()) + with pytest.raises(ValueError, match="requires --api-key"): + serve_mod.serve_http("ignored.json", host=host) + + +@pytest.mark.parametrize("host", ["127.0.0.1", "localhost", "::1"]) +def test_validate_http_bind_allows_unauthenticated_loopback(host): + serve_mod._validate_http_bind(host, None) + + # --- CLI argument parsing ------------------------------------------------- def test_cli_defaults_to_stdio(monkeypatch): From b576a3198b83761f1fa34d0893ce3cb9580b0507 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yanis=20Gu=C3=A9rault?= Date: Wed, 2 Sep 2026 14:40:55 +0200 Subject: [PATCH 2/2] feat(serve): discover graphs on interval --- README.md | 6 ++ graphify/serve.py | 66 +++++++++++++++++- tests/test_serve.py | 160 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 230 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 86d0a97314..164b3d4799 100644 --- a/README.md +++ b/README.md @@ -521,10 +521,16 @@ Serve multiple repositories containing `graphify-out/graph.json` from one MCP en # Serve over stdio (the default transport). python -m graphify.serve --graphs-dir .. +# Rescan for added or removed graphs every 30 seconds (use 0 to rescan each request). +python -m graphify.serve --graphs-dir .. --graph-scan-interval 0 + # Serve over HTTP. Remote HTTP requires a nonblank API key. python -m graphify.serve --graphs-dir .. --transport http --host 0.0.0.0 --port 8080 --api-key "$SECRET" ``` +Graph discovery is refreshed every 30 seconds by default. Pass +`--graph-scan-interval 0` to rescan the directory for every MCP request. + #### Docker Compose `docker-compose.multi.yml` serves two pre-built repository graphs over local diff --git a/graphify/serve.py b/graphify/serve.py index 9db608d19b..a7eb1c2d37 100644 --- a/graphify/serve.py +++ b/graphify/serve.py @@ -6,7 +6,9 @@ import re import sys import threading +import time import weakref +from math import isfinite from array import array from collections import OrderedDict from contextvars import ContextVar @@ -49,6 +51,9 @@ def __init__(self) -> None: self._allow_project_paths = True self._load_learning_overlay = True self._lock = threading.Lock() + self._graphs_dir: Path | None = None + self._graph_scan_interval = 30.0 + self._last_discovery = 0.0 @classmethod def from_path(cls, graph_path: Path) -> "GraphRegistry": @@ -103,6 +108,16 @@ def from_named_paths(cls, graph_paths: dict[str, Path]) -> "GraphRegistry": ) return reg + @classmethod + def from_graphs_dir(cls, root: Path, scan_interval: float = 30.0) -> "GraphRegistry": + if not isfinite(scan_interval) or scan_interval < 0: + raise ValueError("graph scan interval must be non-negative") + registry = cls.from_named_paths(_discover_graphs(root)) + registry._graphs_dir = Path(root).resolve() + registry._graph_scan_interval = scan_interval + registry._last_discovery = time.monotonic() + return registry + def rescan(self) -> None: with self._lock: for name, ctx in list(self._graphs.items()): @@ -130,6 +145,34 @@ def rescan(self) -> None: except (SystemExit, Exception): continue + now = time.monotonic() + if ( + self._graphs_dir is not None + and now - self._last_discovery >= self._graph_scan_interval + ): + try: + discovered = _discover_graphs(self._graphs_dir) + except (OSError, ValueError): + return + self._last_discovery = now + for name in set(self._graphs) - set(discovered): + del self._graphs[name] + for name, graph_path in discovered.items(): + if name in self._graphs: + continue + try: + graph = _load_graph_or_raise(str(graph_path), load_learning_overlay=False) + _get_trigram_index(graph) + self._graphs[name] = GraphContext( + name=name, + path=graph_path.resolve(), + graph=graph, + communities=_communities_from_graph(graph), + mtime=graph_path.stat().st_mtime, + ) + except (GraphLoadError, OSError): + continue + def get(self, name: str) -> GraphContext | None: with self._lock: return self._graphs.get(name) @@ -1709,7 +1752,7 @@ def _build_server( fixed_session_state = session_state fallback_session_state: dict = {} active_session: ContextVar[object | None] = ContextVar("graphify_mcp_session", default=None) - is_multi = len(registry.names()) > 1 + is_multi = registry._graphs_dir is not None or len(registry.names()) > 1 _ctx_cache = _GraphContextCache(_max_server_contexts()) default_paths = { str(ctx.path.resolve()) @@ -2645,6 +2688,12 @@ def _main(argv: list[str] | None = None) -> None: import argparse import os + def _non_negative_float(value: str) -> float: + parsed = float(value) + if not isfinite(parsed) or parsed < 0: + raise argparse.ArgumentTypeError("graph scan interval must be non-negative") + return parsed + parser = argparse.ArgumentParser( prog="python -m graphify.serve", description="Serve a graphify knowledge graph over MCP (stdio or Streamable HTTP).", @@ -2667,6 +2716,13 @@ def _main(argv: list[str] | None = None) -> None: metavar="PATH", help="Recursively serve graphify-out/graph.json files below PATH", ) + parser.add_argument( + "--graph-scan-interval", + type=_non_negative_float, + default=None, + metavar="SECONDS", + help="Rescan --graphs-dir for added/removed graphs every N seconds (default: 30; 0 per request)", + ) parser.add_argument( "--transport", choices=["stdio", "http"], @@ -2699,11 +2755,17 @@ def _main(argv: list[str] | None = None) -> None: ) args = parser.parse_args(argv) + if args.graph_scan_interval is not None and args.graphs_dir is None: + parser.error("--graph-scan-interval requires --graphs-dir") + if args.graphs_dir is not None: if args.graph_path is not None or args.graph_flag is not None: parser.error("--graphs-dir cannot be combined with a graph path") try: - registry = GraphRegistry.from_named_paths(_discover_graphs(Path(args.graphs_dir))) + registry = GraphRegistry.from_graphs_dir( + Path(args.graphs_dir), + scan_interval=30.0 if args.graph_scan_interval is None else args.graph_scan_interval, + ) except (OSError, ValueError, GraphLoadError) as exc: parser.error(str(exc)) if args.transport == "http": diff --git a/tests/test_serve.py b/tests/test_serve.py index ffa71a6827..01b45e2f71 100644 --- a/tests/test_serve.py +++ b/tests/test_serve.py @@ -1898,6 +1898,83 @@ def test_from_paths_reloads_graph_only_before_each_request(self, tmp_path, monke assert "Nodes: 2" in handlers["graph_stats"]({}) + def test_graphs_dir_discovers_new_graph_after_interval(self, tmp_path): + _write_registry_graph(tmp_path / "alpha", "graphify-out") + registry = GraphRegistry.from_graphs_dir(tmp_path, scan_interval=30) + _write_registry_graph(tmp_path / "beta", "graphify-out") + registry._last_discovery -= 30 + + registry.rescan() + + assert registry.names() == ["alpha", "beta"] + + def test_graphs_dir_defers_discovery_until_interval(self, tmp_path): + _write_registry_graph(tmp_path / "alpha", "graphify-out") + registry = GraphRegistry.from_graphs_dir(tmp_path, scan_interval=30) + _write_registry_graph(tmp_path / "beta", "graphify-out") + + registry.rescan() + + assert registry.names() == ["alpha"] + + def test_graphs_dir_removes_deleted_graph_after_interval(self, tmp_path): + graph = _write_registry_graph(tmp_path / "alpha", "graphify-out") + registry = GraphRegistry.from_graphs_dir(tmp_path, scan_interval=30) + graph.unlink() + registry._last_discovery -= 30 + + registry.rescan() + + assert registry.names() == [] + + def test_graphs_dir_zero_interval_discovers_on_every_rescan(self, tmp_path): + _write_registry_graph(tmp_path / "alpha", "graphify-out") + registry = GraphRegistry.from_graphs_dir(tmp_path, scan_interval=0) + _write_registry_graph(tmp_path / "beta", "graphify-out") + + registry.rescan() + + assert registry.names() == ["alpha", "beta"] + + @pytest.mark.parametrize("scan_interval", [float("nan"), float("inf"), float("-inf")]) + def test_graphs_dir_rejects_non_finite_scan_interval(self, tmp_path, scan_interval): + _write_registry_graph(tmp_path / "alpha", "graphify-out") + + with pytest.raises(ValueError, match="graph scan interval must be non-negative"): + GraphRegistry.from_graphs_dir(tmp_path, scan_interval=scan_interval) + + def test_graphs_dir_reloads_known_graph_before_discovery_interval(self, tmp_path): + graph = _write_registry_graph( + tmp_path / "alpha", "graphify-out", + nodes=[{"id": "a", "label": "a", "community": 0}], edges=[], + ) + registry = GraphRegistry.from_graphs_dir(tmp_path, scan_interval=30) + _write_registry_graph( + tmp_path / "alpha", "graphify-out", + nodes=[ + {"id": "a", "label": "a", "community": 0}, + {"id": "b", "label": "b", "community": 0}, + ], edges=[], + ) + import os + original_mtime = graph.stat().st_mtime + os.utime(graph, (original_mtime + 1, original_mtime + 1)) + + registry.rescan() + + assert registry.get("alpha").graph.number_of_nodes() == 2 + + def test_graphs_dir_ignores_malformed_new_graph(self, tmp_path): + _write_registry_graph(tmp_path / "alpha", "graphify-out") + registry = GraphRegistry.from_graphs_dir(tmp_path, scan_interval=0) + malformed = _write_registry_graph(tmp_path / "beta", "graphify-out") + malformed.write_text("{malformed", encoding="utf-8") + + registry.rescan() + + assert registry.names() == ["alpha"] + assert registry.get("alpha").graph.number_of_nodes() == 2 + def test_registry_has_no_directory_discovery_api(self): assert not hasattr(GraphRegistry, "from_" + "directory") @@ -1957,6 +2034,21 @@ def test_multi_graph_has_list_graphs(self, tmp_path): assert "list_graphs" in handlers assert "use_graph" in handlers + def test_directory_registry_discovers_and_selects_graph_added_after_server_build(self, tmp_path): + _write_registry_graph(tmp_path / "alpha", "graphify-out") + registry = GraphRegistry.from_graphs_dir(tmp_path, scan_interval=30) + session = {} + _, handlers = _build_server(registry, session_state=session) + _write_registry_graph(tmp_path / "beta", "graphify-out") + registry._last_discovery -= 30 + + listing = handlers["list_graphs"]({}) + selection = handlers["use_graph"]({"graph": "beta"}) + + assert "beta" in listing + assert "Switched to graph 'beta'" in selection + assert session["current_graph"] == "beta" + def test_single_graph_has_pr_tools(self, tmp_path): gp = _write_registry_graph(tmp_path, "proj") reg = GraphRegistry.from_path(gp) @@ -2071,6 +2163,72 @@ def test_unknown_name_raises(self, tmp_path): _resolve_graph(reg, graph="nope", current=None) class TestMainCLI: + def test_serve_cli_defaults_graph_scan_interval(self, tmp_path, monkeypatch): + _write_registry_graph(tmp_path / "api", "graphify-out") + captured = {} + monkeypatch.setattr( + "graphify.serve.serve", + lambda graph_path=None, *, registry=None: captured.update(registry=registry), + ) + + from graphify.serve import _main + _main(["--graphs-dir", str(tmp_path)]) + + assert captured["registry"]._graph_scan_interval == 30 + + def test_serve_cli_forwards_graph_scan_interval(self, tmp_path, monkeypatch): + _write_registry_graph(tmp_path / "api", "graphify-out") + captured = {} + monkeypatch.setattr( + "graphify.serve.serve", + lambda graph_path=None, *, registry=None: captured.update(registry=registry), + ) + + from graphify.serve import _main + _main(["--graphs-dir", str(tmp_path), "--graph-scan-interval", "0"]) + + assert captured["registry"]._graph_scan_interval == 0 + + def test_serve_cli_rejects_negative_graph_scan_interval(self, tmp_path, capsys): + _write_registry_graph(tmp_path / "api", "graphify-out") + + from graphify.serve import _main + + with pytest.raises(SystemExit): + _main(["--graphs-dir", str(tmp_path), "--graph-scan-interval", "-1"]) + + assert "non-negative" in capsys.readouterr().err + + @pytest.mark.parametrize( + "value", ["nan", "inf", "-inf"], ids=["nan", "inf", "negative-inf"] + ) + def test_serve_cli_rejects_non_finite_graph_scan_interval(self, tmp_path, capsys, value): + _write_registry_graph(tmp_path / "api", "graphify-out") + + from graphify.serve import _main + + with pytest.raises(SystemExit): + _main(["--graphs-dir", str(tmp_path), f"--graph-scan-interval={value}"]) + + assert "non-negative" in capsys.readouterr().err + + @pytest.mark.parametrize("transport", ["stdio", "http"]) + def test_serve_cli_rejects_graph_scan_interval_without_graphs_dir( + self, monkeypatch, capsys, transport + ): + monkeypatch.setattr("graphify.serve.serve", lambda *args, **kwargs: None) + monkeypatch.setattr("graphify.serve.serve_http", lambda *args, **kwargs: None) + + from graphify.serve import _main + + argv = ["--graph-scan-interval", "15"] + if transport == "http": + argv.extend(["--transport", "http"]) + with pytest.raises(SystemExit): + _main(argv) + + assert "--graph-scan-interval requires --graphs-dir" in capsys.readouterr().err + def test_serve_cli_starts_named_registry_for_graphs_dir(self, tmp_path, monkeypatch): graph_path = _write_registry_graph(tmp_path / "team" / "api", "graphify-out") captured = {} @@ -2146,10 +2304,12 @@ def test_serve_cli_forwards_http_options_to_graphs_dir_registry(self, tmp_path, "--graphs-dir", str(tmp_path), "--transport", "http", "--host", "0.0.0.0", "--port", "8080", "--api-key", "secret", "--path", "/registry", "--json-response", "--stateless", "--session-timeout", "30", + "--graph-scan-interval", "15", ]) assert captured["graph_path"] is None assert captured["registry"].names() == ["team/api"] + assert captured["registry"]._graph_scan_interval == 15 assert {key: value for key, value in captured.items() if key != "registry"} == { "graph_path": None, "host": "0.0.0.0",