fix(mcp): rank symbol searches and report truncated pages

README.md

@@ -279,6 +279,15 @@ codeprism serve --no-auto-index
 codeprism serve /path/to/project --transport sse --port 8765
 ```
 
+The MCP `search_symbol` tool ranks substring matches by exact name, then prefix, then other
+matches, with stable name/file ordering. Its `limit` defaults to 50 and accepts 1-500.
+`kind` and `project_path` filters apply before this limit; the response includes `count`
+(returned matches), `total` (all filtered matches) and `truncated` (whether any were cut).
+For example, `search_symbol("run", kind="function", limit=100)` can return more matches;
+narrow the query or filters when `truncated` is true. With semantic embeddings enabled,
+search keeps its existing 20-neighbor cap and ignores `limit`; that response has
+`count`/`matches` only because a complete `total` and `truncated` are unavailable.
+
 The SSE server listens on `127.0.0.1` only and has no built-in authentication; there is no `--host`
 option yet ([#45](https://github.com/knight22-21/CodePrism/issues/45)). To reach it from another machine, use an SSH tunnel or a reverse proxy on
 the same host (see [INTEGRATIONS.md](INTEGRATIONS.md)).

codeprism/core/storage.py

@@ -26,6 +26,9 @@
 # 2: Python calls record call style/receiver; function-local imports are indexed
 INDEX_FORMAT_VERSION = 2
 
+DEFAULT_SEARCH_LIMIT = 50
+MAX_SEARCH_LIMIT = 500
+
 _SCHEMA = """
 PRAGMA journal_mode=WAL;
 PRAGMA foreign_keys=ON;
@@ -331,20 +334,49 @@ async def find_symbols(self, name: str, kind: str | None = None) -> list[SymbolR
         return [_row_to_symbol(r) for r in rows]
 
     async def search_symbols(self, query: str, kind: str | None = None) -> list[SymbolRecord]:
-        pattern = f"%{query}%"
-        kind_param = kind.lower() if kind else None
-        if kind_param:
-            async with self.db.execute(
-                "SELECT * FROM symbols WHERE name LIKE ? AND kind = ? LIMIT 50",
-                (pattern, kind_param),
-            ) as cur:
-                rows = await cur.fetchall()
-        else:
-            async with self.db.execute(
-                "SELECT * FROM symbols WHERE name LIKE ? LIMIT 50", (pattern,)
-            ) as cur:
-                rows = await cur.fetchall()
-        return [_row_to_symbol(r) for r in rows]
+        """Return the default ranked substring page, preserving the list API."""
+        symbols, _ = await self.search_symbols_page(query, kind)
+        return symbols
+
+    async def search_symbols_page(
+        self,
+        query: str,
+        kind: str | None = None,
+        *,
+        limit: int = DEFAULT_SEARCH_LIMIT,
+        file_prefix: str | None = None,
+    ) -> tuple[list[SymbolRecord], int]:
+        """Return ranked substring matches and the total before applying the limit."""
+        if (
+            isinstance(limit, bool)
+            or not isinstance(limit, int)
+            or not 1 <= limit <= MAX_SEARCH_LIMIT
+        ):
+            raise ValueError(f"limit must be an integer between 1 and {MAX_SEARCH_LIMIT}")
+        predicates = ["s.name LIKE ?"]
+        parameters: list[str | int] = [f"%{query}%"]
+        if kind:
+            predicates.append("s.kind = ?")
+            parameters.append(kind.lower())
+        if file_prefix is not None:
+            # Match the existing literal startswith filter, including '%' and '_'
+            # in a directory name, before pagination and counting.
+            predicates.append("substr(f.path, 1, length(?)) = ?")
+            parameters.extend([file_prefix, file_prefix])
+        sql = f"""
+            SELECT s.*, COUNT(*) OVER () AS total_matches
+            FROM symbols AS s JOIN files AS f ON f.id = s.file_id
+            WHERE {" AND ".join(predicates)}
+            ORDER BY s.name = ? DESC, s.name LIKE ? DESC, s.name, f.path, s.id
+            LIMIT ?
+        """
+        parameters.extend([query, f"{query}%", limit])
+        # A window count keeps the page and total in the same SQLite snapshot,
+        # even if a watcher updates the index while the query runs.
+        async with self.db.execute(sql, parameters) as cur:
+            rows = list(await cur.fetchall())
+        total = int(rows[0]["total_matches"]) if rows else 0
+        return [_row_to_symbol(row) for row in rows], total
 
     async def get_all_symbols(self) -> list[SymbolRecord]:
         async with self.db.execute("SELECT * FROM symbols") as cur:

codeprism/mcp/server.py

@@ -403,26 +403,32 @@ async def get_data_flow(file: str, symbol: str) -> dict[str, Any]:
 @mcp.tool(
     description="Find symbols by name (substring match).\n\n"
     "project_path: optional directory prefix to restrict results to a sub-project.\n"
-    "kind: case-insensitive symbol kind; one of " + ", ".join(SYMBOL_KINDS)
+    "kind: case-insensitive symbol kind; one of " + ", ".join(SYMBOL_KINDS) + ".\n"
+    "limit: substring result limit, 1-500 (default 50). Substring results rank exact names, "
+    "then prefixes, then other matches, and include total/truncated.\n"
+    "Semantic search retains its 20-neighbor cap, ignores limit, and omits total/truncated."
 )
 async def search_symbol(
     query: str,
     project_path: str | None = None,
     kind: str | None = None,
+    limit: int = 50,
 ) -> dict[str, Any]:
     """Find symbols by name (substring match).
 
     project_path: optional directory prefix to restrict results to a sub-project.
     kind: case-insensitive symbol kind from SYMBOL_KINDS.
+    limit: 1-500 (default 50), applied after filtering in substring mode.
+    Substring responses include total and truncated. Semantic search retains
+    its 20-neighbor cap and ignores limit; its total/truncated are unavailable.
     """
     try:
-        matches = await _get().search_symbols(query, kind)
+        page = await _get().search_symbols_page(
+            query, kind, limit=limit, file_prefix=_abs(project_path) if project_path else None
+        )
     except ValueError as error:
         return {"error": str(error)}
-    if project_path:
-        prefix = _abs(project_path)
-        matches = [m for m in matches if m.file_path.startswith(prefix)]
-    return search_matches_to_dict(matches)
+    return search_matches_to_dict(page.matches, total=page.total)
 
 
 @mcp.tool()

codeprism/mcp/tools.py

@@ -94,8 +94,11 @@ def file_map_to_dict(r: FileMap) -> dict[str, Any]:
     }
 
 
-def search_matches_to_dict(matches: list[SearchMatch]) -> dict[str, Any]:
-    return {
+def search_matches_to_dict(
+    matches: list[SearchMatch], *, total: int | None = None
+) -> dict[str, Any]:
+    """Serialize matches, adding completeness metadata only when the total is known."""
+    result = {
         "count": len(matches),
         "matches": [
             {
@@ -110,6 +113,9 @@ def search_matches_to_dict(matches: list[SearchMatch]) -> dict[str, Any]:
             for m in matches
         ],
     }
+    if total is not None:
+        result.update(total=total, truncated=total > len(matches))
+    return result
 
 
 def deps_to_dict(r: DependencyResult) -> dict[str, Any]:

codeprism/query/engine.py

@@ -8,7 +8,7 @@
 
 from ..core.graph import GraphEngine
 from ..core.models import SYMBOL_KINDS, FileRecord, NodeKind, SymbolRecord
-from ..core.storage import StorageManager
+from ..core.storage import DEFAULT_SEARCH_LIMIT, MAX_SEARCH_LIMIT, StorageManager
 from . import context as _ctx_mod
 from . import impact as _imp_mod
 from . import summary as _sum_mod
@@ -27,6 +27,14 @@ class SearchMatch:
     docstring_excerpt: str | None = None
 
 
+@dataclass
+class SearchPage:
+    """Search matches with a complete substring total, or None for semantic search."""
+
+    matches: list[SearchMatch]
+    total: int | None = None
+
+
 @dataclass
 class FileMapEntry:
     path: str
@@ -156,21 +164,45 @@ async def get_callees(self, file_path: str, symbol_name: str) -> list[SymbolReco
     # ── Search ────────────────────────────────────────────────────────────────
 
     async def search_symbols(self, query: str, kind: str | None = None) -> list[SearchMatch]:
-        """Search symbols, rejecting unknown kinds before querying either backend."""
+        """Return the default search page while preserving the existing list API."""
+        return (await self.search_symbols_page(query, kind)).matches
+
+    async def search_symbols_page(
+        self,
+        query: str,
+        kind: str | None = None,
+        *,
+        limit: int = DEFAULT_SEARCH_LIMIT,
+        file_prefix: str | None = None,
+    ) -> SearchPage:
+        """Search with filtered substring pagination; semantic results keep their cap.
+
+        file_prefix is an already-resolved literal path prefix. limit and total
+        apply only to substring mode; the semantic backend still requests 20
+        neighbors and supplies no complete total.
+        """
+        if (
+            isinstance(limit, bool)
+            or not isinstance(limit, int)
+            or not 1 <= limit <= MAX_SEARCH_LIMIT
+        ):
+            raise ValueError(f"limit must be an integer between 1 and {MAX_SEARCH_LIMIT}")
         if kind is not None:
             normalized = kind.lower()
             if normalized not in SYMBOL_KINDS:
                 raise ValueError(f"Unknown kind {kind!r}. Valid kinds: {', '.join(SYMBOL_KINDS)}")
             kind = normalized
         if self._embedder is not None and self._embed_store is not None:
-            return await self._semantic_search(query, kind)
-        return await self._substring_search(query, kind)
-
-    async def _substring_search(self, query: str, kind: str | None = None) -> list[SearchMatch]:
-        raw = await self._storage.search_symbols(query, kind)
+            matches = await self._semantic_search(query, kind)
+            if file_prefix is not None:
+                matches = [match for match in matches if match.file_path.startswith(file_prefix)]
+            return SearchPage(matches)
+        raw, total = await self._storage.search_symbols_page(
+            query, kind, limit=limit, file_prefix=file_prefix
+        )
         all_files = await self._storage.get_all_files()
         id_to_path = {f.id: f.path for f in all_files}
-        return [
+        matches = [
             SearchMatch(
                 symbol=sym,
                 file_path=id_to_path.get(sym.file_id, ""),
@@ -179,6 +211,7 @@ async def _substring_search(self, query: str, kind: str | None = None) -> list[S
             )
             for sym in raw
         ]
+        return SearchPage(matches, total)
 
     async def _semantic_search(self, query: str, kind: str | None = None) -> list[SearchMatch]:
         import asyncio as _asyncio

docs/changelog.md

@@ -8,6 +8,10 @@ Versioning follows [Semantic Versioning](https://semver.org/).
 ## [Unreleased]
 
 ### Fixed
+- Substring symbol search ranks exact names and prefixes before limiting results. MCP
+  `search_symbol` now accepts `limit` (1-500, default 50) and reports filtered `total` and
+  `truncated`; kind/project filters apply before pagination. Semantic search keeps its
+  existing cap and response shape. (#38)
 - CLI `callers`, `context`, `impact` and `summary` distinguish missing files and ambiguous paths,
   report candidates on stderr, and exit with status 1 for lookup errors. `callers` no longer
   reports an unknown symbol as a successful zero-callers result. MCP responses stay unchanged.