Add a Go indexer service with a gRPC streaming contract

Extracts the walk-and-parse stage behind rpc IndexRepo(IndexRequest) returns (stream
IndexEvent). Opt-in: IndexingService still uses the Python parser, and nothing changes
until indexer_grpc_enabled is set.

WHY, STATED HONESTLY: NOT SPEED
The obvious claim is false and was measured before writing any Go. On 900 files:
sequential 1115ms, ThreadPool(4) 1092ms (1.02x), ProcessPool(4) 393ms (2.84x).
py-tree-sitter never releases the GIL, so threads gain nothing -- which also means the
obvious anyio.to_thread fix would not have worked. Only ~41% of the parse phase is C, so a
perfect native rewrite ceilings near 2.4x, BELOW what ProcessPoolExecutor on the existing
Python already delivers for half a day of work.

The actual reason is a process and failure boundary: repos.py runs indexing by spinning a
new event loop inside an anyio threadpool thread and blocking it for minutes, in the same
process that serves chat. Server streaming is why gRPC rather than one big response --
progress becomes part of the contract instead of shared mutable state, the same bug class
that made the SSE progress bar show only 0% or 100%.

BINDING CHOICE
Official tree-sitter/go-tree-sitter, not smacker/go-tree-sitter -- which has roughly twice
the stars (562 vs 288) and outranks it in search results but was last pushed 2024-08-27
with 42 open issues and no deprecation notice. Verified against the GitHub API rather than
assumed, the same way the Terraform provider choice was checked.

TWO BUGS NOT REPEATED
Both were fixed in the Python parser earlier; this gets them right from the start.
- .tsx uses LanguageTSX(), not LanguageTypescript(). The plain TS grammar cannot parse JSX.
- Chunk text comes from node.Utf8Text(source) over the source BYTES, because tree-sitter
  offsets are byte offsets and slicing a decoded string with them corrupts every chunk
  after the first multi-byte character.
had_parse_error is on the wire because tree-sitter returns a partial tree rather than
failing, so without it the Python caller cannot decide to fall back to raw indexing.

ONE BUG FIXED IN THE PORT
Python's _find_files max-files cap never worked: its `break` left only the inner filename
loop, so os.walk continued into the next directory and kept appending. The Go walker uses
fs.SkipAll and reports Truncated, so a partial index is not mistaken for full coverage.

VERIFIED END TO END, Go server streaming to the Python client over real code
(apps/api/src): 5 progress events, 3 chunk batches, 509 chunks from 59 files in 99ms, 0
parse errors; kinds {module 17, class 121, function 109, method 262}; spot-check
class 'Level' in api/graphql/schema.py L47-53 matches the file. Correctness cases: a .tsx
component parses with had_parse_error=False under the tsx grammar, and after a line
containing "héllo — wörld 日本語 🎉" the chunks still slice as 'class Café:' and
'def método(self):' rather than misaligned fragments.

Suite: 168 passed, ruff clean. Generated *_pb2*.py are excluded from ruff, since any fix
there is overwritten by the next protoc run.

NOT DONE, deliberately: the service is not wired into the pipeline. That swap needs a
placement decision this commit does not make -- the service must be co-located with the API
on the shared volume, because clones live under ./data/repos/<owner>/<name> and that volume
attaches to exactly one machine. There is also no Go CI job yet (ci.yml has Python 3.11 and
Node 20 only, no cgo toolchain, no grammar-compile caching, no cross-language contract
test), and this build links 5 grammars against the Python side's 9 -- Health() reports which,
so the caller never assumes parity.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: ShreeBohara

apps/api/pyproject.toml

@@ -17,6 +17,9 @@ asyncio_mode = "auto"
 [tool.ruff]
 line-length = 100
 target-version = "py311"
+# Generated by protoc; any fix here is overwritten on the next `protoc` run, so linting it
+# would produce a permanently dirty tree.
+extend-exclude = ["src/core/indexer/indexer_pb2.py", "src/core/indexer/indexer_pb2_grpc.py"]
 
 [tool.ruff.lint]
 select = ["E", "F", "I", "W"]

apps/api/src/core/indexer/__init__.py

@@ -0,0 +1,159 @@
+"""
+Async client for the Go indexer service.
+
+Optional and off by default: the Python walk-and-parse path in IndexingService remains the
+default, and this is used only when indexer_grpc_enabled is set. Both produce the same
+chunk shape, so the caller does not branch beyond choosing a source.
+
+Uses grpc.aio, which grpc documents as stable, and grpcio is already an installed
+dependency (transitively via chromadb) so enabling this adds no new runtime wheel.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import AsyncIterator, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class RemoteChunk:
+    """Mirrors the Chunk message. Deliberately not the ORM model: this is wire data."""
+    file_path: str
+    language: str
+    chunk_type: str
+    name: str
+    content: str
+    start_line: int
+    end_line: int
+    had_parse_error: bool
+
+
+@dataclass
+class IndexProgress:
+    stage: str
+    current_path: str
+    files_processed: int
+    total_files: int
+    percent: float
+
+
+@dataclass
+class IndexSummary:
+    files_walked: int
+    files_parsed: int
+    files_skipped: int
+    chunks_emitted: int
+    files_with_errors: int
+    duration_ms: int
+
+
+class IndexerUnavailable(RuntimeError):
+    """Raised when the service cannot be reached, so the caller can fall back to Python."""
+
+
+class IndexerClient:
+    def __init__(self, target: str, timeout_seconds: float = 900.0):
+        self._target = target
+        self._timeout = timeout_seconds
+
+    async def health(self) -> Optional[dict]:
+        """Returns the service's linked grammars, or None when unreachable."""
+        import grpc
+
+        from src.core.indexer import indexer_pb2, indexer_pb2_grpc
+
+        try:
+            async with grpc.aio.insecure_channel(self._target) as channel:
+                stub = indexer_pb2_grpc.IndexerStub(channel)
+                res = await stub.Health(indexer_pb2.HealthRequest(), timeout=10)
+                return {
+                    "ok": res.ok,
+                    "version": res.version,
+                    # Reported rather than assumed: the Go build links a narrower grammar
+                    # set than the Python parser, and pretending otherwise would silently
+                    # drop languages.
+                    "parsers": {p.language: list(p.extensions) for p in res.parsers},
+                }
+        except Exception as exc:
+            logger.warning("Indexer service health check failed (%s): %s", self._target, exc)
+            return None
+
+    async def index_repo(
+        self,
+        repo_id: str,
+        root_path: str,
+        max_files: int = 0,
+        max_file_size_kb: int = 0,
+        batch_size: int = 0,
+    ) -> AsyncIterator[object]:
+        """
+        Stream IndexProgress / List[RemoteChunk] / IndexSummary as the server produces them.
+
+        Yields heterogeneous types on purpose: the caller wants progress promptly and
+        chunks in batches, and forcing both into one shape would mean buffering the whole
+        repository before anything is usable.
+        """
+        import grpc
+
+        from src.core.indexer import indexer_pb2, indexer_pb2_grpc
+
+        request = indexer_pb2.IndexRequest(
+            repo_id=repo_id,
+            root_path=root_path,
+            max_files=max_files,
+            max_file_size_kb=max_file_size_kb,
+            batch_size=batch_size,
+        )
+
+        try:
+            async with grpc.aio.insecure_channel(self._target) as channel:
+                stub = indexer_pb2_grpc.IndexerStub(channel)
+                async for event in stub.IndexRepo(request, timeout=self._timeout):
+                    which = event.WhichOneof("event")
+                    if which == "progress":
+                        p = event.progress
+                        yield IndexProgress(
+                            stage=p.stage, current_path=p.current_path,
+                            files_processed=p.files_processed, total_files=p.total_files,
+                            percent=p.percent,
+                        )
+                    elif which == "chunks":
+                        yield [
+                            RemoteChunk(
+                                file_path=c.file_path, language=c.language,
+                                chunk_type=c.chunk_type, name=c.name, content=c.content,
+                                start_line=c.start_line, end_line=c.end_line,
+                                had_parse_error=c.had_parse_error,
+                            )
+                            for c in event.chunks.chunks
+                        ]
+                    elif which == "completed":
+                        c = event.completed
+                        yield IndexSummary(
+                            files_walked=c.files_walked, files_parsed=c.files_parsed,
+                            files_skipped=c.files_skipped, chunks_emitted=c.chunks_emitted,
+                            files_with_errors=c.files_with_errors, duration_ms=c.duration_ms,
+                        )
+                    elif which == "failed":
+                        # A server-side failure arrives as a stream event, not a status
+                        # code, so it has to be re-raised here to stop the caller treating
+                        # a truncated stream as a complete index.
+                        raise IndexerUnavailable(
+                            f"indexer failed: {event.failed.message} ({event.failed.path})"
+                        )
+        except IndexerUnavailable:
+            raise
+        except Exception as exc:
+            raise IndexerUnavailable(f"indexer stream failed: {exc}") from exc
+
+
+async def collect_chunks(client: IndexerClient, repo_id: str, root_path: str) -> List[RemoteChunk]:
+    """Convenience for tests and one-shot use; drains the stream into a list."""
+    out: List[RemoteChunk] = []
+    async for item in client.index_repo(repo_id, root_path):
+        if isinstance(item, list):
+            out.extend(item)
+    return out