feat(mcp): add grep_code MCP tool and backend-agnostic line-accurate vector file reconstruction

REQUIREMENTS.md

@@ -2,7 +2,7 @@
 
 > **Note:** This document is automatically generated and verified against the live test suite by `scripts/generate_requirements.py` and `tests/backend/test_requirements_sync.py`.
 
-**Test Verification Baseline:** **1163 Automated Tests** (805 Pytest Backend + 308 Vitest Frontend + 50 Playwright E2E).
+**Test Verification Baseline:** **1171 Automated Tests** (813 Pytest Backend + 308 Vitest Frontend + 50 Playwright E2E).
 
 ---
 
@@ -1057,6 +1057,16 @@ Asserts both Markdown and PDF docs are matched under doc_type='doc'._
 - `test_sync_single_git_repo_batching_over_25`
 - `test_sync_single_git_repo_clone_error`
 
+#### `tests/test_grep_code.py` (8 tests)
+- `test_file_reader_reconstruction_overlapping_chunks_qdrant` - _Verifies FileReaderService._read_from_vector_store reconstructs overlapping chunks with exact line numbers on QdrantStore._
+- `test_file_reader_reconstruction_overlapping_chunks_pgvector` - _Verifies FileReaderService._read_from_vector_store reconstructs overlapping chunks with exact line numbers on PgVectorStore._
+- `test_grep_code_literal_and_case_sensitivity`
+- `test_grep_code_regex_and_syntax_error`
+- `test_grep_code_repo_and_glob_filters`
+- `test_grep_code_vector_store_fallback_and_deduplication` - _Verifies grep_code works for ephemeral/vector-backed files and deduplicates overlapping chunks._
+- `test_mcp_grep_code_registration_and_execution`
+- `test_api_search_grep_endpoint`
+
 #### `tests/test_incremental_pipeline.py` (3 tests)
 - `test_full_incremental_sync_pipeline` - _End-to-End integration test for the full ingestion & incremental sync pipeline:
 1. Initial full repo ingestion (code + markdown with wikilinks & standard links).

app/api/routers/files.py

@@ -19,6 +19,16 @@ class FileSummarizePayload(BaseModel):
     force_refresh: bool = Field(False, description="Whether to bypass cache and regenerate summary")
 
 
+class GrepSearchPayload(BaseModel):
+    pattern: str = Field(..., max_length=500, description="Substring pattern to search for")
+    repo: Optional[str] = Field(None, description="Optional repository identifier or storage namespace")
+    path_glob: Optional[str] = Field(None, description="Optional fnmatch file glob")
+    is_regex: bool = Field(False, description="Whether pattern is a regex")
+    case_sensitive: bool = Field(False, description="Whether matching is case-sensitive")
+    context_lines: int = Field(2, ge=0, le=10, description="Surrounding context lines")
+    limit: int = Field(30, ge=1, le=100, description="Maximum matches to return")
+
+
 @router.get(
     "/admin/api/files/read",
     dependencies=[Depends(require_role(Role.VIEWER))],
@@ -100,3 +110,46 @@ async def api_summarize_file(payload: FileSummarizePayload):
     except Exception as e:
         logger.error(f"Error summarizing file '{payload.path}': {e}")
         return JSONResponse(status_code=500, content={"error": "Internal server error summarizing file."})
+
+
+@router.post(
+    "/api/search/grep",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
+@router.post(
+    "/admin/api/search/grep",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
+async def api_grep_code(payload: GrepSearchPayload):
+    try:
+        import re
+        raw_pattern = payload.pattern.strip() if payload.pattern else ""
+        if not raw_pattern:
+            return JSONResponse(status_code=400, content={"error": "pattern must be a non-empty string."})
+
+        raw_repo = payload.repo.strip() if payload.repo else None
+        if raw_repo and ("\x00" in raw_repo or any(part == ".." for part in raw_repo.replace("\\", "/").split("/"))):
+            return JSONResponse(status_code=400, content={"error": "Path traversal or invalid repo detected."})
+
+        safe_pattern = re.escape(raw_pattern)
+        reader = get_file_reader_service()
+        result = reader.grep_code(
+            pattern=safe_pattern,
+            repo=raw_repo,
+            path_glob=payload.path_glob,
+            is_regex=True,
+            case_sensitive=payload.case_sensitive,
+            context_lines=payload.context_lines,
+            limit=payload.limit,
+        )
+        result["pattern"] = raw_pattern
+        return result
+    except ForbiddenError as fe:
+        logger.warning(f"Forbidden access during grep_code: {fe}")
+        return JSONResponse(status_code=403, content={"error": "Access denied."})
+    except ValueError as ve:
+        logger.warning(f"Validation error during grep_code: {ve}")
+        return JSONResponse(status_code=400, content={"error": "Invalid search parameters."})
+    except Exception as e:
+        logger.error(f"Error in api_grep_code: {e}")
+        return JSONResponse(status_code=500, content={"error": "Internal server error during grep search."})

app/mcp/handlers/__init__.py

@@ -30,6 +30,7 @@
 from app.mcp.handlers.file_handlers import (
     handle_read_file,
     handle_summarize_file,
+    handle_grep_code,
 )
 from app.mcp.handlers.diagnostic_handlers import (
     handle_diagnose_issue,
@@ -55,5 +56,6 @@
     "handle_what_is_ingested",
     "handle_read_file",
     "handle_summarize_file",
+    "handle_grep_code",
     "handle_diagnose_issue",
 ]

app/mcp/handlers/file_handlers.py

@@ -105,3 +105,39 @@ async def handle_summarize_file(
     except Exception as e:
         logger.error(f"Error summarizing file '{path}': {e}")
         return json.dumps({"error": f"Error summarizing file: {str(e)}"})
+
+
+async def handle_grep_code(
+    pattern: Annotated[str, Field(description="Exact substring or regex pattern to search for across repositories.")],
+    repo: Annotated[Optional[str], Field(description="Optional repository identifier or storage namespace filter.")] = None,
+    path_glob: Annotated[Optional[str], Field(description="Optional fnmatch file path glob (e.g., '*.py' or 'app/services/*').")] = None,
+    is_regex: Annotated[bool, Field(description="If True, treats pattern as a regular expression. Defaults to False (literal string).")] = False,
+    case_sensitive: Annotated[bool, Field(description="If True, performs case-sensitive matching. Defaults to False.")] = False,
+    context_lines: Annotated[int, Field(description="Number of context lines surrounding each match (0-10). Defaults to 2.")] = 2,
+    limit: Annotated[int, Field(description="Maximum number of matched lines to return (1-100). Defaults to 30.")] = 30
+) -> str:
+    """Exact literal and regex pattern matching across one or all indexed repositories returning compact JSON."""
+    try:
+        enforce_tool_permission(Role.VIEWER)
+        reader_fn = _get_tools_attr("get_file_reader_service", get_file_reader_service)
+        reader = reader_fn()
+
+        result = reader.grep_code(
+            pattern=pattern,
+            repo=repo,
+            path_glob=path_glob,
+            is_regex=is_regex,
+            case_sensitive=case_sensitive,
+            context_lines=context_lines,
+            limit=limit,
+        )
+        return json.dumps(result, separators=(",", ":"))
+
+    except ForbiddenError as fe:
+        logger.warning(f"Forbidden error in handle_grep_code: {fe}")
+        return json.dumps({"error": "forbidden", "message": str(fe)}, separators=(",", ":"))
+    except ValueError as ve:
+        return json.dumps({"error": "invalid_argument", "message": str(ve)}, separators=(",", ":"))
+    except Exception as e:
+        logger.error(f"Error searching code with grep: {e}")
+        return json.dumps({"error": str(e)}, separators=(",", ":"))

app/mcp/tools.py

@@ -30,6 +30,7 @@
     handle_what_is_ingested,
     handle_read_file,
     handle_summarize_file,
+    handle_grep_code,
     handle_diagnose_issue,
 )
 
@@ -139,6 +140,12 @@ def register_mcp_tools_and_resources(server=None):
             description="Retrieve an existing summary or generate a structured LLM executive summary for a large file."
         )(handle_summarize_file)
 
+    if "grep_code" not in existing_tools:
+        server.tool(
+            name="grep_code",
+            description="Deterministic literal and regex pattern matching across one or all indexed repositories with surrounding context lines and AST symbol coordinates."
+        )(handle_grep_code)
+
     if "diagnose_issue" not in existing_tools:
         server.tool(
             name="diagnose_issue",