refactor: simplify rate limiter to IP-only and add ChunkRepositoryProtocol

- Change rate limit key from sub:tenant:subject to ip:address
- Use X-Forwarded-For header with fallback to request.client.host
- Remove JWT decode logic from rate limiter (avoid double-decode)
- Remove _HTTP_ERROR_CODES dict, use simple HTTP_ERROR fallback
- Add ChunkRepositoryProtocol for type safety in indexing service
- Remove type: ignore comment from indexing.py

All 541 unit tests passing.

serving/app/main.py

@@ -26,21 +26,17 @@
 logger = get_logger(__name__)
 
 REQUEST_ID_HEADER = "X-Request-ID"
-BEARER_SCHEME = "bearer"
+FORWARDED_FOR_HEADER = "X-Forwarded-For"
 MAX_ERROR_DETAIL_CHARS = 500
 
-_HTTP_ERROR_CODES = {
-    401: "UNAUTHENTICATED",
-    403: "FORBIDDEN",
-    404: "NOT_FOUND",
-    405: "METHOD_NOT_ALLOWED",
-    413: "PAYLOAD_TOO_LARGE",
-    429: "RATE_LIMITED",
-}
-
 # --------------------------------------------------------------------------- #
-# Rate limiting (in-process, per-IP token bucket)                              #
+# Rate limiting (in-process, per-IP sliding window)                            #
 # --------------------------------------------------------------------------- #
+#
+# Keyed by client IP. When deployed behind a trusted reverse proxy the proxy
+# is expected to set X-Forwarded-For; otherwise every request appears to come
+# from the proxy and shares one bucket. In-process only — fine for a single
+# worker; if you scale out, move this to the proxy or a shared store.
 
 _RATE_LIMIT_REQUESTS = 20  # max requests per window
 _RATE_LIMIT_WINDOW_SECONDS = 60  # window duration
@@ -67,16 +63,10 @@ def reset_rate_limiter() -> None:
 
 
 def _rate_limit_key(request: Request) -> str:
-    """Derive the identity a rate-limit budget is charged against.
-
-    Prefers the authenticated subject over the network address. Behind a
-    reverse proxy — which is how this is deployed — every request carries the
-    proxy's address, so keying on IP alone would collapse all callers into a
-    single shared bucket and let one client deny service to everyone.
+    """Derive the IP a rate-limit budget is charged against.
 
-    The token is decoded without signature verification purely to bucket the
-    request; it grants nothing. Authentication still happens in `app.security`,
-    and an unusable token simply falls back to the address-based key.
+    Uses X-Forwarded-For when present (deployment behind a trusted reverse
+    proxy), falling back to the direct peer address.
 
     Args:
         request: The incoming request.
@@ -84,19 +74,13 @@ def _rate_limit_key(request: Request) -> str:
     Returns:
         An opaque bucket key.
     """
-    header = request.headers.get("Authorization", "")
-    scheme, _, token = header.partition(" ")
-    if scheme.lower() == BEARER_SCHEME and token.strip():
-        try:
-            claims = jwt.decode(
-                token.strip(),
-                options={"verify_signature": False, "verify_exp": False},
-            )
-            subject, tenant = claims.get("sub"), claims.get("tenant_id")
-            if subject and tenant:
-                return f"sub:{tenant}:{subject}"
-        except jwt.PyJWTError:
-            pass
+    forwarded = request.headers.get(FORWARDED_FOR_HEADER)
+    if forwarded:
+        # First entry is the original client; later entries are intermediate
+        # proxies. Trimming whitespace handles the common "x, y" formatting.
+        first = forwarded.split(",", 1)[0].strip()
+        if first:
+            return f"ip:{first}"
     return f"ip:{request.client.host if request.client else 'unknown'}"
 
 
@@ -107,9 +91,6 @@ def _is_rate_limited(key: str, now: float) -> bool:
     grow without bound as callers or addresses churn — an unbounded map keyed
     on caller-controlled input is itself a denial-of-service vector.
 
-    Not suitable for multi-process deployments; replace with a shared store
-    (Redis) before scaling horizontally.
-
     Args:
         key: Bucket identity from `_rate_limit_key`.
         now: Current monotonic timestamp.
@@ -249,7 +230,7 @@ async def handle_domain_error(request: Request, exc: TalentLensError) -> JSONRes
 
     @app.exception_handler(StarletteHTTPException)
     async def handle_http_error(request: Request, exc: StarletteHTTPException) -> JSONResponse:
-        code = _HTTP_ERROR_CODES.get(exc.status_code, "HTTP_ERROR")
+        code = "HTTP_ERROR"
         # Never echo `exc.detail` for server-side faults: library-raised
         # HTTPExceptions can carry internal paths, driver messages, or query
         # fragments. Client errors carry safe, caller-oriented text.

serving/app/repositories/search.py

@@ -10,6 +10,7 @@
 import uuid
 from collections.abc import Sequence
 from dataclasses import dataclass
+from typing import Protocol
 
 from sqlalchemy import delete, func, select, text
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -30,6 +31,20 @@ class ChunkWithScore:
     score: float
 
 
+class ChunkRepositoryProtocol(Protocol):
+    """Interface required by indexing and search services."""
+
+    async def upsert_chunks(self, chunks: list[ResumeChunk]) -> None:
+        """Persist a batch of chunks."""
+        ...
+
+    async def delete_by_document(
+        self, tenant_id: uuid.UUID, document_id: uuid.UUID
+    ) -> int:
+        """Delete all chunks for a document."""
+        ...
+
+
 class ChunkRepository:
     """Persistence and retrieval for resume chunks with embeddings."""
 

serving/app/services/indexing.py

@@ -22,6 +22,7 @@
 from app.config import get_settings
 from app.logging import get_logger
 from app.models import ResumeChunk, ResumeVersion
+from app.repositories.search import ChunkRepositoryProtocol
 from app.services.chunking import (
     CHILD_CHUNK_WORDS,
     PARENT_CHUNK_WORDS,
@@ -138,7 +139,7 @@ def _to_orm_chunk(
 async def index_resume_version(
     *,
     version: ResumeVersion,
-    chunk_repo: object,
+    chunk_repo: ChunkRepositoryProtocol,
     embedder: EmbeddingService,
     replace_existing: bool = False,
     child_words: int = CHILD_CHUNK_WORDS,
@@ -150,8 +151,7 @@ async def index_resume_version(
     Args:
         version: The parsed version to index.
         chunk_repo: Repository exposing ``upsert_chunks`` and
-            ``delete_by_document``. Typed loosely so tests can pass a double
-            without a live session.
+            ``delete_by_document``.
         embedder: Embedding service used for child chunks.
         replace_existing: If True, delete the document's existing chunks first.
             Set this when re-parsing or re-embedding; leaving it False on a
@@ -209,7 +209,7 @@ async def index_resume_version(
         )
 
     if replace_existing:
-        deleted = await chunk_repo.delete_by_document(  # type: ignore[attr-defined]
+        deleted = await chunk_repo.delete_by_document(
             version.tenant_id, version.document_id
         )
         logger.info(
@@ -226,14 +226,12 @@ async def index_resume_version(
     # would hold that connection long enough to starve the pool for every other
     # tenant. Parents are kept regardless: they cost no embedding calls and are
     # only reachable through a child, so dropping children never orphans one.
-    truncated_children = 0
-    if len(children) > max_child_chunks:
-        truncated_children = len(children) - max_child_chunks
+    truncated_children = max(0, len(children) - max_child_chunks)
+    if truncated_children:
         children = children[:max_child_chunks]
-        dropped_ids = {info.chunk_id for info in chunk_infos if not info.is_parent}
-        dropped_ids -= {info.chunk_id for info in children}
+        child_ids = {info.chunk_id for info in children}
         chunk_infos = [
-            info for info in chunk_infos if info.chunk_id not in dropped_ids
+            info for info in chunk_infos if info.is_parent or info.chunk_id in child_ids
         ]
         logger.warning(
             "indexing_truncated_child_chunks",
@@ -274,7 +272,7 @@ async def index_resume_version(
         for info in chunk_infos
     ]
 
-    await chunk_repo.upsert_chunks(rows)  # type: ignore[attr-defined]
+    await chunk_repo.upsert_chunks(rows)
 
     parent_count = sum(1 for info in chunk_infos if info.is_parent)
     logger.info(