fix(query): scope and paginate file maps

README.md

@@ -268,6 +268,15 @@ reload it ([#35](https://github.com/knight22-21/CodePrism/issues/35)).
 
 ### MCP server
 
+`get_file_map(project_path="", limit=200, offset=0)` returns file roles under the served
+project root by default. Pass a project-relative directory such as `src`, or an absolute
+directory, to narrow it. Files are sorted by path; `limit` accepts 1-1000 and `offset` skips
+that many files. `total_files` and `total_symbols` count the whole selected directory,
+while `truncated` is true whenever the page omits files (including earlier pages).
+A directory with no indexed files returns an error. To read the next page, increase
+`offset` by `limit`. Python callers of `QueryEngine.get_file_map` can use `limit=None`
+for a complete map; internal CLI statistics and summaries do this.
+
 ```bash
 # stdio (Claude Code, Codex, Cursor, ...). With no path: the project containing the current directory
 codeprism serve

codeprism/cli.py

@@ -442,7 +442,7 @@ async def _stats(project: str, verbose: bool, json_output: bool = False) -> None
     engine, storage = await _open_session(project)
     try:
         data = await engine.get_stats()
-        file_map = await engine.get_file_map(project) if verbose else None
+        file_map = await engine.get_file_map(limit=None) if verbose else None
     finally:
         await storage.close()
 
@@ -673,7 +673,7 @@ def _setup(agent: str, project: str, global_: bool) -> None:
 get_callers(file, function)             → every call site with line numbers
 get_callees(file, function)             → every function this one calls
 search_symbol(query, kind=None)         → find symbols by name substring
-get_file_map(project_path)              → full file tree with role summaries
+get_file_map(project_path, limit, offset) → directory file roles (200 per page by default)
 get_dependencies(file)                  → imports: internal vs external
 scan_diff(original, proposed, file)     → security check before any write
 record_read(session_id, file, symbol)   → log what you've already read
@@ -1587,7 +1587,7 @@ async def _scan(target: str, all_: bool, diff: str | None, project: str) -> None
     if all_:
         engine, storage = await _open_session(project)
         try:
-            fm = await engine.get_file_map(project)
+            fm = await engine.get_file_map(limit=None)
         finally:
             await storage.close()
 

codeprism/mcp/server.py

@@ -327,9 +327,19 @@ async def get_module_summary(file: str) -> dict[str, Any]:
 
 
 @mcp.tool()
-async def get_file_map(project_path: str = "") -> dict[str, Any]:
-    """Return the file tree with per-file role summaries (token-efficient entry point)."""
-    result = await _get().get_file_map(project_path)
+async def get_file_map(project_path: str = "", limit: int = 200, offset: int = 0) -> dict[str, Any]:
+    """Return a directory's file roles in path order, relative to the served project.
+
+    limit: 1-1000 files (default 200). offset: non-negative number of files to skip.
+    Totals describe the entire directory; truncated means this page omits files.
+    """
+    if limit is None:
+        return {"error": "limit must be between 1 and 1000"}
+    path = _abs(project_path or ".")
+    try:
+        result = await _get().get_file_map(path, limit=limit, offset=offset)
+    except ValueError as exc:
+        return {"error": str(exc), "project_path": path}
     return file_map_to_dict(result)
 
 

codeprism/mcp/tools.py

@@ -79,6 +79,9 @@ def file_map_to_dict(r: FileMap) -> dict[str, Any]:
         "project_path": r.project_path,
         "total_files": r.total_files,
         "total_symbols": r.total_symbols,
+        "limit": r.limit,
+        "offset": r.offset,
+        "truncated": r.truncated,
         "files": [
             {
                 "path": e.path,

codeprism/query/engine.py

@@ -52,6 +52,9 @@ class FileMap:
     entries: list[FileMapEntry] = field(default_factory=list)
     total_files: int = 0
     total_symbols: int = 0
+    limit: int | None = None
+    offset: int = 0
+    truncated: bool = False
 
 
 @dataclass
@@ -237,8 +240,20 @@ async def _semantic_search(self, query: str, kind: str | None = None) -> list[Se
 
     # ── File map ──────────────────────────────────────────────────────────────
 
-    async def get_file_map(self, project_path: str = "") -> FileMap:
+    async def get_file_map(
+        self, project_path: str = "", *, limit: int | None = 200, offset: int = 0
+    ) -> FileMap:
+        """Return a sorted directory-scoped page; limit=None requests the complete map."""
+        if limit is not None and not 1 <= limit <= 1000:
+            raise ValueError("limit must be between 1 and 1000")
+        if offset < 0:
+            raise ValueError("offset must be non-negative")
         all_files = await self._storage.get_all_files()
+        if project_path:
+            root = Path(project_path).resolve()
+            all_files = [f for f in all_files if root in Path(f.path).parents]
+            if not all_files:
+                raise ValueError(f"No indexed files under directory '{project_path}'")
         all_syms = await self._storage.get_all_symbols()
 
         # Build per-file symbol counts
@@ -247,8 +262,12 @@ async def get_file_map(self, project_path: str = "") -> FileMap:
             if sym.file_id in file_syms:
                 file_syms[sym.file_id].append(sym)
 
+        total_symbols = sum(len(syms) for syms in file_syms.values())
+        ordered_files = sorted(all_files, key=lambda x: x.path)
+        page = ordered_files[offset : None if limit is None else offset + limit]
+
         entries: list[FileMapEntry] = []
-        for f in sorted(all_files, key=lambda x: x.path):
+        for f in page:
             syms = file_syms[f.id]
             n_class = sum(1 for s in syms if s.kind == NodeKind.CLASS)
             n_func = sum(1 for s in syms if s.kind == NodeKind.FUNCTION)
@@ -276,7 +295,10 @@ async def get_file_map(self, project_path: str = "") -> FileMap:
             project_path=project_path,
             entries=entries,
             total_files=len(all_files),
-            total_symbols=len(all_syms),
+            total_symbols=total_symbols,
+            limit=limit,
+            offset=offset,
+            truncated=len(entries) < len(all_files),
         )
 
     # ── Dependencies ──────────────────────────────────────────────────────────

docs/changelog.md

@@ -8,6 +8,10 @@ Versioning follows [Semantic Versioning](https://semver.org/).
 ## [Unreleased]
 
 ### Fixed
+- `get_file_map` now filters indexed files by directory and returns a sorted page of up to 200
+  files by default. MCP callers can use `limit` (1-1000) and `offset`; totals cover the selected
+  directory and `truncated` reports omitted files. A directory with no indexed files returns
+  an error. Internal CLI consumers still request complete maps. (#49)
 - CLI `index` and the `index_project` MCP tool now load `.codeprism.toml`, matching the server's
   startup index. Explicit language and embedding options override the file, while omitted options
   preserve it; `index --no-embeddings` can disable configured embeddings. Automatic CLI/MCP parser

tests/test_file_map.py

@@ -0,0 +1,152 @@
+"""Directory-scoped file maps with stable, explicit pagination."""
+
+from io import StringIO
+from pathlib import Path
+
+import pytest
+from fastmcp import Client
+from rich.console import Console
+
+import codeprism.cli as cli
+import codeprism.mcp.server as server
+from codeprism.core.models import FileRecord, NodeKind, SymbolRecord
+from codeprism.query.engine import QueryEngine
+
+
+@pytest.fixture
+async def mapped_project(storage, graph, tmp_path, monkeypatch):
+    root = tmp_path / "project"
+    files = []
+    for relative in ("src/a.py", "src/nested/b.py", "src_extra/c.py", "z.py"):
+        path = root / relative
+        file = FileRecord.create(str(path), language="python", line_count=10)
+        await storage.upsert_file(file)
+        await storage.upsert_symbol(
+            SymbolRecord.create(
+                file_path=str(path),
+                file_id=file.id,
+                name="work",
+                kind=NodeKind.FUNCTION,
+                line_start=1,
+                line_end=2,
+            )
+        )
+        files.append(str(path))
+    engine = QueryEngine(graph, storage)
+    monkeypatch.setattr(server, "_engine", engine)
+    monkeypatch.setattr(server, "_project_path", str(root))
+    return engine, root, files
+
+
+async def test_directory_filter_and_totals(mapped_project):
+    engine, root, files = mapped_project
+    result = await engine.get_file_map(str(root / "src"))
+    assert [e.path for e in result.entries] == files[:2]
+    assert result.total_files == result.total_symbols == 2
+    assert all(e.function_count == e.symbol_count == 1 for e in result.entries)
+
+
+async def test_mcp_relative_directory_uses_served_root(mapped_project, tmp_path, monkeypatch):
+    _, root, files = mapped_project
+    monkeypatch.chdir(tmp_path)
+    result = await server.get_file_map("src/../src/")
+    assert [e["path"] for e in result["files"]] == files[:2]
+    assert result["project_path"] == str(root / "src")
+    assert result["total_symbols"] == 2
+
+
+@pytest.mark.parametrize("relative", ["missing", "src_extra2", "src/a.py"])
+async def test_unindexed_directory_is_an_error(mapped_project, relative):
+    engine, root, _ = mapped_project
+    with pytest.raises(ValueError, match="No indexed files"):
+        await engine.get_file_map(str(root / relative))
+    result = await server.get_file_map(relative)
+    assert "No indexed files" in result["error"]
+
+
+async def test_pages_keep_scoped_totals_and_order(mapped_project):
+    engine, root, files = mapped_project
+    first = await server.get_file_map(limit=2)
+    second = await server.get_file_map(limit=2, offset=2)
+    past_end = await server.get_file_map(limit=2, offset=4)
+    assert [e["path"] for e in first["files"] + second["files"]] == files
+    for page in (first, second, past_end):
+        assert page["total_files"] == page["total_symbols"] == 4
+        assert page["limit"] == 2
+        assert page["truncated"] is True
+    assert first["offset"] == 0
+    assert second["offset"] == 2
+    assert past_end["files"] == []
+    whole = await engine.get_file_map(str(root), limit=None)
+    assert len(whole.entries) == 4
+    assert whole.truncated is False
+
+
+async def test_default_and_maximum_caps(storage, graph, tmp_path, monkeypatch):
+    root = tmp_path / "big"
+    for i in reversed(range(1001)):
+        await storage.upsert_file(FileRecord.create(str(root / f"{i:04d}.py")))
+    engine = QueryEngine(graph, storage)
+    monkeypatch.setattr(server, "_engine", engine)
+    monkeypatch.setattr(server, "_project_path", str(root))
+    result = await server.get_file_map()
+    assert len(result["files"]) == result["limit"] == 200
+    assert result["total_files"] == 1001
+    assert result["truncated"] is True
+    assert Path(result["files"][0]["path"]).name == "0000.py"
+    assert len((await server.get_file_map(limit=1000))["files"]) == 1000
+    assert len((await engine.get_file_map(str(root), limit=None)).entries) == 1001
+
+
+@pytest.mark.parametrize("command", ["stats", "scan"])
+async def test_cli_consumers_keep_complete_index(storage, graph, tmp_path, monkeypatch, command):
+    """Whole-index CLI output includes files beyond the page cap and outside the root."""
+    root = tmp_path / "project"
+    root.mkdir()
+    paths = [root / f"{i:04d}.py" for i in range(201)] + [tmp_path / "external.py"]
+    for path in paths:
+        path.write_text("value = 1\n", encoding="utf-8")
+        await storage.upsert_file(FileRecord.create(str(path), language="python", line_count=1))
+    engine = QueryEngine(graph, storage)
+
+    async def open_session(project):
+        return engine, storage
+
+    output = StringIO()
+    monkeypatch.setattr(cli, "_open_session", open_session)
+    monkeypatch.setattr(cli, "console", Console(file=output, width=120))
+    if command == "stats":
+        await cli._stats(str(root), verbose=True)
+        table = output.getvalue().rsplit("Files", 1)[1]
+        rows = [line for line in table.splitlines() if line.lstrip().startswith("│")]
+        assert len(rows) == len(paths)
+        assert "0200.py" in table
+    else:
+        await cli._scan(".", all_=True, diff=None, project=str(root))
+        assert "202 files · 0 issue(s) found" in output.getvalue()
+
+
+@pytest.mark.parametrize("kwargs", [{"limit": 0}, {"limit": -1}, {"limit": 1001}, {"offset": -1}])
+async def test_invalid_pagination_is_an_error(mapped_project, kwargs):
+    engine, root, _ = mapped_project
+    with pytest.raises(ValueError):
+        await engine.get_file_map(str(root), **kwargs)
+    assert "error" in await server.get_file_map(**kwargs)
+
+
+async def test_empty_index_and_protocol_arguments(storage, graph, tmp_path, monkeypatch):
+    import codeprism.core.paths as paths
+
+    monkeypatch.setattr(server, "_engine", QueryEngine(graph, storage))
+    monkeypatch.setattr(server, "_project_path", str(tmp_path))
+    monkeypatch.setattr(server, "_auto_index", False)
+    monkeypatch.setattr(paths, "get_db_path", lambda project: tmp_path / "protocol.db")
+    assert "error" in await server.get_file_map()
+    async with Client(server.mcp) as client:
+        tools = await client.list_tools()
+        tool = next(t for t in tools if t.name == "get_file_map")
+        schema = getattr(tool, "input_schema", None) or tool.inputSchema
+        assert schema["properties"]["limit"]["default"] == 200
+        assert "offset" in schema["properties"]
+        result = await client.call_tool("get_file_map", {"limit": 1, "offset": 0})
+        assert "No indexed files" in result.data["error"]