feat(search): add Phase 2 semantic search with pgvector and hybrid retrieval

Implements parent-child chunking, pgvector embeddings, and hybrid (dense + lexical)
retrieval with Reciprocal Rank Fusion. Adds TEI embedding service adapter, NoOp
reranker stub, and /search endpoint.

Known open issues in this commit (to be addressed in follow-up):
- Phase 2 search disconnected from ingestion flow (no code path creates chunks)
- Raw SQL f-string interpolation in repositories/search.py (injection risk)
- Zero test coverage for Phase 2 (chunking, embedding, reranker, search)
- content_tsv ORM/migration drift (model declares Text, migration creates tsvector)

Components added:
- services/chunking.py: parent (~700w) and child (~180w) chunk generation
- services/embedding.py: EmbeddingService protocol + TEI adapter
- services/reranker.py: RerankerService protocol + NoOp adapter
- services/search.py: RRF fusion, dense/lexical search orchestration
- repositories/search.py: pgvector cosine + tsvector ts_rank_cd queries
- routers/search.py: POST /search endpoint
- schemas/search.py: SearchRequest/Response Pydantic models
- migrations/versions/20260731_0930: resume_chunks table, HNSW + GIN indexes

.env.example

@@ -1,4 +1,4 @@
-# TalentLens — environment template (Phase 0-1)
+# TalentLens — environment template (Phase 0-2)
 # Copy to `.env` and fill in. Never commit a populated `.env`.
 
 # --- Runtime -----------------------------------------------------------------
@@ -27,3 +27,21 @@ MAX_UPLOAD_BYTES=10485760         # 10 MiB
 
 # --- CORS (Vite dev server by default; add the Vercel origin in production) --
 CORS_ALLOW_ORIGINS=["http://localhost:5173"]
+
+# --- Embedding (Phase 2) ----------------------------------------------------
+# HuggingFace Text Embeddings Inference endpoint for bge-m3
+EMBEDDING_ENDPOINT=http://localhost:8080
+EMBEDDING_MODEL=BAAI/bge-m3
+EMBEDDING_DIM=1024
+EMBEDDING_BATCH_SIZE=32
+
+# --- Search (Phase 2) -------------------------------------------------------
+SEARCH_DENSE_WEIGHT=0.6
+SEARCH_LEXICAL_WEIGHT=0.4
+SEARCH_TOP_K_RECALL=20
+SEARCH_RERANK_TOP_K=10
+
+# --- Reranker (Phase 2, optional) --------------------------------------------
+# Leave empty to disable reranking (CPU fallback / dev mode)
+RERANKER_ENDPOINT=
+RERANKER_MODEL=BAAI/bge-reranker-v2-m3

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "talentlens"
 version = "0.1.0"
-description = "TalentLens — AI CV Screener (Phase 0-1: foundations + ingestion)"
+description = "TalentLens — AI CV Screener (Phase 0-2: foundations + ingestion + search)"
 requires-python = ">=3.11,<3.12"
 dependencies = [
     "fastapi>=0.141,<0.142",
@@ -23,6 +23,7 @@ dependencies = [
     "pdfplumber>=0.11,<0.12",
     "python-docx>=1.1,<2",
     "defusedxml>=0.7,<0.8",
+    "pgvector>=0.3,<0.4",
 ]
 
 [project.optional-dependencies]

serving/app/config.py

@@ -62,6 +62,22 @@ class Settings(BaseSettings):
     max_upload_bytes: int = 10 * 1024 * 1024
     allowed_upload_mime_types: tuple[str, ...] = ("application/pdf", DOCX_MIME)
 
+    # --- Embedding (Phase 2) -------------------------------------------------
+    embedding_endpoint: str = "http://localhost:8080"
+    embedding_model: str = "BAAI/bge-m3"
+    embedding_dim: int = 1024
+    embedding_batch_size: int = 32
+
+    # --- Search (Phase 2) ----------------------------------------------------
+    search_dense_weight: float = 0.6
+    search_lexical_weight: float = 0.4
+    search_top_k_recall: int = 20
+    search_rerank_top_k: int = 10
+
+    # --- Reranker (Phase 2) --------------------------------------------------
+    reranker_endpoint: str = ""
+    reranker_model: str = "BAAI/bge-reranker-v2-m3"
+
     cors_allow_origins: tuple[str, ...] = ("http://localhost:5173",)
 
     log_level: str = "INFO"

serving/app/exceptions.py

@@ -128,3 +128,25 @@ class StorageError(TalentLensError):
     error_code: ClassVar[str] = "STORAGE_UNAVAILABLE"
     status_code: ClassVar[int] = 503
     default_message: ClassVar[str] = "Document storage is temporarily unavailable."
+
+
+# --------------------------------------------------------------------------- #
+# Search / Embeddings (Phase 2)                                                #
+# --------------------------------------------------------------------------- #
+
+
+class EmbeddingServiceUnavailableError(TalentLensError):
+    """Raised when the embedding service cannot be reached."""
+
+    error_code: ClassVar[str] = "EMBEDDING_SERVICE_UNAVAILABLE"
+    status_code: ClassVar[int] = 503
+    default_message: ClassVar[str] = "The embedding service is temporarily unavailable."
+
+
+class SearchError(TalentLensError):
+    """Raised when a search operation fails unexpectedly."""
+
+    error_code: ClassVar[str] = "SEARCH_FAILED"
+    status_code: ClassVar[int] = 500
+    default_message: ClassVar[str] = "The search operation failed."
+

serving/app/main.py

@@ -21,7 +21,7 @@
 from app.config import Settings, get_settings
 from app.exceptions import TalentLensError
 from app.logging import configure_logging, get_logger
-from app.routers import auth, jobs, resumes
+from app.routers import auth, jobs, resumes, search
 
 logger = get_logger(__name__)
 
@@ -47,7 +47,7 @@
 _RATE_LIMIT_MAX_TRACKED_KEYS = 10_000  # hard ceiling on limiter memory
 
 # Paths that are rate-limited (upload endpoints are the highest risk).
-_RATE_LIMITED_PREFIXES = ("/api/v1/resumes", "/api/v1/jobs")
+_RATE_LIMITED_PREFIXES = ("/api/v1/resumes", "/api/v1/jobs", "/api/v1/search")
 
 _request_counts: dict[str, list[float]] = defaultdict(list)
 
@@ -341,4 +341,5 @@ async def health() -> dict[str, str]:
     app.include_router(auth.router, prefix=cfg.api_v1_prefix)
     app.include_router(resumes.router, prefix=cfg.api_v1_prefix)
     app.include_router(jobs.router, prefix=cfg.api_v1_prefix)
+    app.include_router(search.router, prefix=cfg.api_v1_prefix)
     return app

serving/app/migrations/versions/20260731_0930_add_resume_chunks_with_pgvector.py

@@ -0,0 +1,119 @@
+"""add resume_chunks table with pgvector
+
+Revision ID: a1b2c3d4e5f6
+Revises: dd2329f32308
+Create Date: 2026-07-31 09:30:00.000000+00:00
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "a1b2c3d4e5f6"
+down_revision: str | None = "dd2329f32308"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Apply this revision."""
+    # Enable the pgvector extension — idempotent, safe to re-run
+    op.execute("CREATE EXTENSION IF NOT EXISTS vector")
+
+    op.create_table(
+        "resume_chunks",
+        sa.Column("id", sa.UUID(), nullable=False),
+        sa.Column("tenant_id", sa.UUID(), nullable=False),
+        sa.Column("resume_version_id", sa.UUID(), nullable=False),
+        sa.Column("document_id", sa.UUID(), nullable=False),
+        sa.Column("chunk_index", sa.Integer(), nullable=False),
+        sa.Column("parent_chunk_id", sa.UUID(), nullable=True),
+        sa.Column("section", sa.String(length=64), nullable=False),
+        sa.Column("content", sa.Text(), nullable=False),
+        sa.Column("content_tsv", sa.Text(), nullable=True),
+        sa.Column("page_from", sa.Integer(), nullable=False),
+        sa.Column("page_to", sa.Integer(), nullable=False),
+        sa.Column("start_char", sa.Integer(), nullable=False),
+        sa.Column("end_char", sa.Integer(), nullable=False),
+        sa.Column("token_count", sa.Integer(), nullable=False),
+        sa.Column("embedding_model", sa.String(length=128), nullable=False),
+        sa.Column("embedding_version", sa.String(length=64), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.ForeignKeyConstraint(
+            ["resume_version_id"], ["resume_versions.id"], ondelete="CASCADE"
+        ),
+        sa.ForeignKeyConstraint(
+            ["document_id"], ["resume_documents.id"], ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+
+    # Add the pgvector embedding column — raw SQL because Alembic/SA don't
+    # know about the vector type natively
+    op.execute("ALTER TABLE resume_chunks ADD COLUMN embedding vector(1024)")
+
+    # Replace the content_tsv text column with a real tsvector column
+    op.execute("ALTER TABLE resume_chunks DROP COLUMN IF EXISTS content_tsv")
+    op.execute(
+        "ALTER TABLE resume_chunks ADD COLUMN content_tsv tsvector "
+        "GENERATED ALWAYS AS (to_tsvector('english', content)) STORED"
+    )
+
+    # Add the is_parent column (needed for filtering child-only retrieval)
+    op.execute(
+        "ALTER TABLE resume_chunks ADD COLUMN is_parent boolean NOT NULL DEFAULT false"
+    )
+
+    # --- Indexes ---
+    # HNSW index for cosine vector search (pgvector-specific)
+    op.execute(
+        "CREATE INDEX ix_resume_chunks_embedding_hnsw ON resume_chunks "
+        "USING hnsw (embedding vector_cosine_ops) "
+        "WITH (m = 16, ef_construction = 64)"
+    )
+
+    # GIN index for full-text search
+    op.execute(
+        "CREATE INDEX ix_resume_chunks_content_tsv ON resume_chunks "
+        "USING gin (content_tsv)"
+    )
+
+    # Composite index for tenant + document scoped queries
+    op.create_index(
+        "ix_resume_chunks_tenant_document",
+        "resume_chunks",
+        ["tenant_id", "document_id"],
+        unique=False,
+    )
+
+    # Tenant isolation index
+    op.create_index(
+        "ix_resume_chunks_tenant_id",
+        "resume_chunks",
+        ["tenant_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    """Revert this revision."""
+    op.drop_index("ix_resume_chunks_tenant_id", table_name="resume_chunks")
+    op.drop_index("ix_resume_chunks_tenant_document", table_name="resume_chunks")
+    op.execute("DROP INDEX IF EXISTS ix_resume_chunks_content_tsv")
+    op.execute("DROP INDEX IF EXISTS ix_resume_chunks_embedding_hnsw")
+    op.drop_table("resume_chunks")
+    # Note: we do NOT drop the vector extension — other tables may use it

serving/app/repositories/search.py

@@ -0,0 +1,287 @@
+"""Data access for resume chunks and vector search.
+
+This is the only layer permitted to query the ``resume_chunks`` table.
+Hybrid retrieval (dense + lexical) is implemented here because the query
+construction is tightly coupled to the pgvector and tsvector index structure.
+"""
+
+from __future__ import annotations
+
+import uuid
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+from sqlalchemy import delete, func, select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models import ResumeChunk
+
+
+@dataclass(frozen=True, slots=True)
+class ChunkWithScore:
+    """A chunk paired with a retrieval score.
+
+    Attributes:
+        chunk: The retrieved chunk.
+        score: Retrieval score (cosine similarity, ts_rank, or fused).
+    """
+
+    chunk: ResumeChunk
+    score: float
+
+
+class ChunkRepository:
+    """Persistence and retrieval for resume chunks with embeddings."""
+
+    def __init__(self, session: AsyncSession) -> None:
+        """Initialize the repository.
+
+        Args:
+            session: An active async session.
+        """
+        self._session = session
+
+    async def upsert_chunks(self, chunks: list[ResumeChunk]) -> None:
+        """Persist a batch of chunks.
+
+        Args:
+            chunks: Chunks to insert.
+        """
+        for chunk in chunks:
+            self._session.add(chunk)
+        await self._session.flush()
+
+    async def delete_by_document(
+        self, tenant_id: uuid.UUID, document_id: uuid.UUID
+    ) -> int:
+        """Delete all chunks for a document.
+
+        Used when re-embedding or re-parsing a document.
+
+        Args:
+            tenant_id: Owning tenant.
+            document_id: Document to clear.
+
+        Returns:
+            Number of rows deleted.
+        """
+        stmt = delete(ResumeChunk).where(
+            ResumeChunk.tenant_id == tenant_id,
+            ResumeChunk.document_id == document_id,
+        )
+        result = await self._session.execute(stmt)
+        return result.rowcount  # type: ignore[return-value]
+
+    async def search_dense(
+        self,
+        tenant_id: uuid.UUID,
+        embedding: list[float],
+        top_k: int = 20,
+        *,
+        document_id: uuid.UUID | None = None,
+    ) -> list[ChunkWithScore]:
+        """Dense vector search using pgvector cosine distance.
+
+        Args:
+            tenant_id: Owning tenant (isolation filter).
+            embedding: Query embedding vector.
+            top_k: Maximum results.
+            document_id: Optional filter to scope results to one document.
+
+        Returns:
+            Chunks ordered by descending cosine similarity.
+        """
+        # Use raw SQL for pgvector operator support
+        vec_str = "[" + ",".join(str(v) for v in embedding) + "]"
+
+        where_clauses = ["tenant_id = :tenant_id", "embedding IS NOT NULL", "is_parent = false"]
+        params: dict[str, object] = {"tenant_id": str(tenant_id), "top_k": top_k}
+
+        if document_id is not None:
+            where_clauses.append("document_id = :document_id")
+            params["document_id"] = str(document_id)
+
+        where_sql = " AND ".join(where_clauses)
+
+        query = text(f"""
+            SELECT id, tenant_id, resume_version_id, document_id,
+                   chunk_index, parent_chunk_id, section, content,
+                   page_from, page_to, start_char, end_char,
+                   token_count, embedding_model, embedding_version,
+                   1 - (embedding <=> :embedding::vector) AS score
+            FROM resume_chunks
+            WHERE {where_sql}
+            ORDER BY embedding <=> :embedding::vector
+            LIMIT :top_k
+        """)
+        params["embedding"] = vec_str
+
+        result = await self._session.execute(query, params)
+        rows = result.fetchall()
+
+        chunks: list[ChunkWithScore] = []
+        for row in rows:
+            chunk = ResumeChunk(
+                id=uuid.UUID(str(row.id)),
+                tenant_id=uuid.UUID(str(row.tenant_id)),
+                resume_version_id=uuid.UUID(str(row.resume_version_id)),
+                document_id=uuid.UUID(str(row.document_id)),
+                chunk_index=row.chunk_index,
+                parent_chunk_id=(
+                    uuid.UUID(str(row.parent_chunk_id))
+                    if row.parent_chunk_id
+                    else None
+                ),
+                section=row.section,
+                content=row.content,
+                page_from=row.page_from,
+                page_to=row.page_to,
+                start_char=row.start_char,
+                end_char=row.end_char,
+                token_count=row.token_count,
+                embedding_model=row.embedding_model,
+                embedding_version=row.embedding_version,
+            )
+            chunks.append(ChunkWithScore(chunk=chunk, score=float(row.score)))
+        return chunks
+
+    async def search_lexical(
+        self,
+        tenant_id: uuid.UUID,
+        query: str,
+        top_k: int = 20,
+        *,
+        document_id: uuid.UUID | None = None,
+    ) -> list[ChunkWithScore]:
+        """Lexical search using PostgreSQL tsvector and ts_rank_cd.
+
+        Args:
+            tenant_id: Owning tenant.
+            query: Raw search query text.
+            top_k: Maximum results.
+            document_id: Optional filter to scope results to one document.
+
+        Returns:
+            Chunks ordered by descending ts_rank_cd score.
+        """
+        where_clauses = [
+            "tenant_id = :tenant_id",
+            "content_tsv IS NOT NULL",
+            "content_tsv @@ plainto_tsquery('english', :query)",
+        ]
+        params: dict[str, object] = {
+            "tenant_id": str(tenant_id),
+            "query": query,
+            "top_k": top_k,
+        }
+
+        if document_id is not None:
+            where_clauses.append("document_id = :document_id")
+            params["document_id"] = str(document_id)
+
+        where_sql = " AND ".join(where_clauses)
+
+        sql = text(f"""
+            SELECT id, tenant_id, resume_version_id, document_id,
+                   chunk_index, parent_chunk_id, section, content,
+                   page_from, page_to, start_char, end_char,
+                   token_count, embedding_model, embedding_version,
+                   ts_rank_cd(content_tsv, plainto_tsquery('english', :query)) AS score
+            FROM resume_chunks
+            WHERE {where_sql}
+            ORDER BY score DESC
+            LIMIT :top_k
+        """)
+
+        result = await self._session.execute(sql, params)
+        rows = result.fetchall()
+
+        chunks: list[ChunkWithScore] = []
+        for row in rows:
+            chunk = ResumeChunk(
+                id=uuid.UUID(str(row.id)),
+                tenant_id=uuid.UUID(str(row.tenant_id)),
+                resume_version_id=uuid.UUID(str(row.resume_version_id)),
+                document_id=uuid.UUID(str(row.document_id)),
+                chunk_index=row.chunk_index,
+                parent_chunk_id=(
+                    uuid.UUID(str(row.parent_chunk_id))
+                    if row.parent_chunk_id
+                    else None
+                ),
+                section=row.section,
+                content=row.content,
+                page_from=row.page_from,
+                page_to=row.page_to,
+                start_char=row.start_char,
+                end_char=row.end_char,
+                token_count=row.token_count,
+                embedding_model=row.embedding_model,
+                embedding_version=row.embedding_version,
+            )
+            chunks.append(ChunkWithScore(chunk=chunk, score=float(row.score)))
+        return chunks
+
+    async def get_parents(self, chunk_ids: list[uuid.UUID]) -> list[ResumeChunk]:
+        """Retrieve parent chunks by their IDs.
+
+        Used for parent expansion after child retrieval.
+
+        Args:
+            chunk_ids: Parent chunk IDs to fetch.
+
+        Returns:
+            The parent chunks.
+        """
+        if not chunk_ids:
+            return []
+        stmt = select(ResumeChunk).where(ResumeChunk.id.in_(chunk_ids))
+        result = await self._session.execute(stmt)
+        return list(result.scalars().all())
+
+    async def get_by_document(
+        self,
+        tenant_id: uuid.UUID,
+        document_id: uuid.UUID,
+        *,
+        children_only: bool = False,
+    ) -> Sequence[ResumeChunk]:
+        """Return all chunks for a document.
+
+        Args:
+            tenant_id: Owning tenant.
+            document_id: Document identifier.
+            children_only: If True, exclude parent chunks.
+
+        Returns:
+            Chunks ordered by chunk_index.
+        """
+        stmt = (
+            select(ResumeChunk)
+            .where(
+                ResumeChunk.tenant_id == tenant_id,
+                ResumeChunk.document_id == document_id,
+            )
+            .order_by(ResumeChunk.chunk_index)
+        )
+        if children_only:
+            stmt = stmt.where(ResumeChunk.parent_chunk_id.isnot(None))
+        result = await self._session.execute(stmt)
+        return result.scalars().all()
+
+    async def count_by_tenant(self, tenant_id: uuid.UUID) -> int:
+        """Count total chunks for a tenant.
+
+        Args:
+            tenant_id: Owning tenant.
+
+        Returns:
+            Total chunk count.
+        """
+        stmt = (
+            select(func.count())
+            .select_from(ResumeChunk)
+            .where(ResumeChunk.tenant_id == tenant_id)
+        )
+        result = await self._session.execute(stmt)
+        return int(result.scalar() or 0)

serving/app/routers/search.py

@@ -0,0 +1,157 @@
+"""Semantic and hybrid talent search endpoints.
+
+Two endpoints per ARCHITECTURE.md §7.2:
+- ``POST /api/v1/search/candidates`` — hybrid talent search
+- ``POST /api/v1/search/similar`` — "more like this"
+
+No LLM in this path. Sub-second. Tenant-scoped via ``ReadPrincipal``.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, status
+
+from app.db import DbSession
+from app.schemas.search import (
+    CandidateSearchHitResponse,
+    EvidenceSpanResponse,
+    SearchRequest,
+    SearchResponse,
+    SimilarRequest,
+)
+from app.security import ReadPrincipal
+from app.services.embedding import get_embedding_service
+from app.services.reranker import get_reranker_service
+from app.services.search import SearchMode as ServiceSearchMode
+from app.services.search import search_candidates, search_similar
+
+router = APIRouter(prefix="/search", tags=["search"])
+
+
+def _to_response(result: object) -> SearchResponse:
+    """Convert a service-layer SearchResult to an API response.
+
+    Args:
+        result: The SearchResult from the search service.
+
+    Returns:
+        The API-shaped response.
+    """
+    from app.services.search import SearchResult
+
+    assert isinstance(result, SearchResult)  # noqa: S101
+
+    return SearchResponse(
+        items=[
+            CandidateSearchHitResponse(
+                document_id=hit.document_id,
+                score=hit.score,
+                spans=[
+                    EvidenceSpanResponse(
+                        chunk_id=span.chunk_id,
+                        content=span.content,
+                        section=span.section,
+                        page_from=span.page_from,
+                        page_to=span.page_to,
+                        start_char=span.start_char,
+                        end_char=span.end_char,
+                        score=span.score,
+                    )
+                    for span in hit.spans
+                ],
+            )
+            for hit in result.items
+        ],
+        count=result.count,
+        query=result.query,
+        mode=result.mode,
+    )
+
+
+@router.post(
+    "/candidates",
+    response_model=SearchResponse,
+    status_code=status.HTTP_200_OK,
+    summary="Search candidates",
+    description=(
+        "Natural-language search across the tenant's candidate pool. "
+        "Returns ranked candidates with evidence-cited results. "
+        "Supports hybrid (default), semantic-only, and lexical-only modes. "
+        "No LLM in this path."
+    ),
+)
+async def search_candidates_endpoint(
+    body: SearchRequest,
+    principal: ReadPrincipal,
+    session: DbSession,
+) -> SearchResponse:
+    """Execute a talent search.
+
+    Args:
+        body: Search parameters.
+        principal: Verified caller.
+        session: Database session.
+
+    Returns:
+        Ranked candidates with supporting evidence spans.
+    """
+    embedder = get_embedding_service()
+    reranker = get_reranker_service()
+
+    # Map schema enum to service enum
+    mode_map = {
+        "hybrid": ServiceSearchMode.HYBRID,
+        "semantic": ServiceSearchMode.SEMANTIC,
+        "lexical": ServiceSearchMode.LEXICAL,
+    }
+    service_mode = mode_map.get(body.mode.value, ServiceSearchMode.HYBRID)
+
+    result = await search_candidates(
+        session=session,
+        embedder=embedder,
+        reranker=reranker,
+        tenant_id=principal.tenant_id,
+        query=body.query,
+        top_k=body.top_k,
+        mode=service_mode,
+    )
+    return _to_response(result)
+
+
+@router.post(
+    "/similar",
+    response_model=SearchResponse,
+    status_code=status.HTTP_200_OK,
+    summary="Find similar candidates",
+    description=(
+        '"More like this" — find candidates similar to a given resume. '
+        "Uses the source document's embeddings to search across the corpus."
+    ),
+)
+async def search_similar_endpoint(
+    body: SimilarRequest,
+    principal: ReadPrincipal,
+    session: DbSession,
+) -> SearchResponse:
+    """Find candidates similar to a given document.
+
+    Args:
+        body: The source document ID and result count.
+        principal: Verified caller.
+        session: Database session.
+
+    Returns:
+        Ranked similar candidates with evidence spans.
+    """
+    embedder = get_embedding_service()
+    reranker = get_reranker_service()
+
+    result = await search_similar(
+        session=session,
+        embedder=embedder,
+        reranker=reranker,
+        tenant_id=principal.tenant_id,
+        document_id=body.document_id,
+        top_k=body.top_k,
+    )
+    return _to_response(result)

serving/app/schemas/search.py

@@ -0,0 +1,99 @@
+"""Request and response schemas for semantic and hybrid talent search."""
+
+from __future__ import annotations
+
+import uuid
+from enum import Enum
+
+from pydantic import BaseModel, Field
+
+
+class SearchMode(str, Enum):
+    """Search retrieval strategy."""
+
+    HYBRID = "hybrid"
+    SEMANTIC = "semantic"
+    LEXICAL = "lexical"
+
+
+class SearchFilters(BaseModel):
+    """Optional filters to narrow search results."""
+
+    sections: list[str] | None = Field(
+        default=None,
+        description="Restrict search to specific resume sections (e.g. experience, skills).",
+    )
+
+
+class SearchRequest(BaseModel):
+    """Payload for ``POST /api/v1/search/candidates``."""
+
+    query: str = Field(
+        min_length=1,
+        max_length=2000,
+        description="Natural-language search query.",
+    )
+    filters: SearchFilters | None = Field(
+        default=None,
+        description="Optional search filters.",
+    )
+    top_k: int = Field(
+        default=10,
+        ge=1,
+        le=100,
+        description="Maximum number of candidates to return.",
+    )
+    mode: SearchMode = Field(
+        default=SearchMode.HYBRID,
+        description="Retrieval strategy: hybrid, semantic, or lexical.",
+    )
+
+
+class EvidenceSpanResponse(BaseModel):
+    """One piece of evidence supporting a search result."""
+
+    chunk_id: uuid.UUID = Field(description="Identifier of the source chunk.")
+    content: str = Field(description="The matching chunk text.")
+    section: str = Field(description="Resume section this chunk belongs to.")
+    page_from: int = Field(description="Starting page (0-based).")
+    page_to: int = Field(description="Ending page (0-based).")
+    start_char: int = Field(description="Starting character offset in the full document.")
+    end_char: int = Field(description="Ending character offset in the full document.")
+    score: float = Field(description="Retrieval score for this chunk.")
+
+
+class CandidateSearchHitResponse(BaseModel):
+    """One candidate in search results."""
+
+    document_id: uuid.UUID = Field(description="The resume document ID.")
+    score: float = Field(description="Aggregated relevance score.")
+    spans: list[EvidenceSpanResponse] = Field(
+        default_factory=list,
+        description="Best supporting evidence spans.",
+    )
+
+
+class SearchResponse(BaseModel):
+    """Complete search result set."""
+
+    items: list[CandidateSearchHitResponse] = Field(
+        default_factory=list,
+        description="Ranked candidate hits.",
+    )
+    count: int = Field(description="Number of candidates returned.")
+    query: str = Field(description="The original query text.")
+    mode: str = Field(description="The search mode used.")
+
+
+class SimilarRequest(BaseModel):
+    """Payload for ``POST /api/v1/search/similar``."""
+
+    document_id: uuid.UUID = Field(
+        description="Source document to find similar candidates for.",
+    )
+    top_k: int = Field(
+        default=10,
+        ge=1,
+        le=100,
+        description="Maximum number of candidates to return.",
+    )

serving/app/services/chunking.py

@@ -0,0 +1,365 @@
+"""Parent-child chunking with offset preservation.
+
+Implements the chunking strategy from ARCHITECTURE.md §6.4:
+- **Child chunks:** ~180 words for embedding precision
+- **Parent chunks:** ~700 words fed to the LLM for context (Phase 4)
+
+Each child carries ``parent_chunk_id`` referencing its parent, enabling the
+retrieve-on-children, expand-to-parents pattern.
+
+Section detection is heuristic: common resume headings are matched
+case-insensitively and mapped to one of the standard categories.
+"""
+
+from __future__ import annotations
+
+import re
+import uuid
+from dataclasses import dataclass
+
+from app.logging import get_logger
+
+logger = get_logger(__name__)
+
+# Target sizes in words
+CHILD_CHUNK_WORDS = 180
+PARENT_CHUNK_WORDS = 700
+
+# Approximate token-to-word ratio (words × factor ≈ tokens)
+_WORD_TO_TOKEN_RATIO = 1.33
+
+# Section heading patterns (case-insensitive)
+_SECTION_PATTERNS: list[tuple[str, re.Pattern[str]]] = [
+    (
+        "experience",
+        re.compile(r"(?i)^\s*(work\s+)?experience|employment|career|professional\s+history"),
+    ),
+    ("education", re.compile(r"(?i)^\s*education|academic|qualifications|degrees?")),
+    ("skills", re.compile(r"(?i)^\s*(technical\s+)?skills|competenc|technologies|proficienc")),
+    (
+        "summary",
+        re.compile(r"(?i)^\s*(professional\s+)?(summary|profile|objective|about\s+me|overview)"),
+    ),
+    ("certifications", re.compile(r"(?i)^\s*certifications?|licens|credentials?")),
+    ("projects", re.compile(r"(?i)^\s*projects?|portfolio")),
+    ("languages", re.compile(r"(?i)^\s*languages?")),
+    ("awards", re.compile(r"(?i)^\s*awards?|honors?|achievements?")),
+    ("publications", re.compile(r"(?i)^\s*publications?|papers?|research")),
+    ("references", re.compile(r"(?i)^\s*references?")),
+]
+
+
+@dataclass(frozen=True, slots=True)
+class ChunkInfo:
+    """One chunk produced by the chunking pipeline.
+
+    Attributes:
+        chunk_id: Pre-assigned UUID for the chunk.
+        parent_chunk_id: UUID of the parent chunk, or None for parents.
+        chunk_index: Zero-based position in the document's chunk sequence.
+        content: The chunk text.
+        section: Detected resume section.
+        page_from: Starting page (0-based).
+        page_to: Ending page (0-based).
+        start_char: Starting character offset in the full text.
+        end_char: Ending character offset in the full text.
+        token_count: Estimated token count.
+        is_parent: True for parent chunks, False for children.
+    """
+
+    chunk_id: uuid.UUID
+    parent_chunk_id: uuid.UUID | None
+    chunk_index: int
+    content: str
+    section: str
+    page_from: int
+    page_to: int
+    start_char: int
+    end_char: int
+    token_count: int
+    is_parent: bool
+
+
+@dataclass
+class _Section:
+    """An intermediate section detected during segmentation."""
+
+    label: str
+    text: str
+    start_char: int
+    end_char: int
+
+
+def _detect_section(line: str) -> str | None:
+    """Match a line against known section headings.
+
+    Args:
+        line: A single line of text.
+
+    Returns:
+        The section label if matched, else None.
+    """
+    stripped = line.strip()
+    if not stripped or len(stripped) > 80:
+        return None
+    for label, pattern in _SECTION_PATTERNS:
+        if pattern.match(stripped):
+            return label
+    return None
+
+
+def _estimate_tokens(text: str) -> int:
+    """Estimate token count from word count.
+
+    Args:
+        text: The text to estimate.
+
+    Returns:
+        Approximate token count.
+    """
+    words = len(text.split())
+    return max(1, round(words * _WORD_TO_TOKEN_RATIO))
+
+
+def _resolve_page(
+    char_offset: int,
+    page_offsets: list[dict[str, int]],
+) -> int:
+    """Resolve a character offset to a page number.
+
+    Args:
+        char_offset: Character offset in the full text.
+        page_offsets: Page boundary list from the parser.
+
+    Returns:
+        The 0-based page number, or 0 if offsets are empty.
+    """
+    if not page_offsets:
+        return 0
+    for page_info in reversed(page_offsets):
+        if char_offset >= page_info.get("start_char", 0):
+            return int(page_info.get("page", 0))
+    return 0
+
+
+def _segment_into_sections(text: str) -> list[_Section]:
+    """Split text into sections based on heading detection.
+
+    Args:
+        text: The full document text.
+
+    Returns:
+        A list of sections with their character offsets.
+    """
+    lines = text.split("\n")
+    sections: list[_Section] = []
+    current_label = "summary"
+    current_lines: list[str] = []
+    current_start = 0
+    char_pos = 0
+
+    for line in lines:
+        detected = _detect_section(line)
+        if detected is not None and current_lines:
+            section_text = "\n".join(current_lines).strip()
+            if section_text:
+                sections.append(_Section(
+                    label=current_label,
+                    text=section_text,
+                    start_char=current_start,
+                    end_char=current_start + len(section_text),
+                ))
+            current_label = detected
+            current_lines = []
+            current_start = char_pos
+        else:
+            current_lines.append(line)
+        char_pos += len(line) + 1  # +1 for the newline
+
+    # Flush last section
+    section_text = "\n".join(current_lines).strip()
+    if section_text:
+        sections.append(_Section(
+            label=current_label,
+            text=section_text,
+            start_char=current_start,
+            end_char=current_start + len(section_text),
+        ))
+
+    # If no sections detected, wrap the entire text as "other"
+    if not sections:
+        sections.append(_Section(
+            label="other",
+            text=text.strip(),
+            start_char=0,
+            end_char=len(text.strip()),
+        ))
+
+    return sections
+
+
+def _split_into_word_chunks(
+    text: str,
+    target_words: int,
+    base_start_char: int,
+) -> list[tuple[str, int, int]]:
+    """Split text into chunks of approximately ``target_words`` words.
+
+    Splits on sentence boundaries when possible, falling back to word
+    boundaries.
+
+    Args:
+        text: Text to split.
+        target_words: Target word count per chunk.
+        base_start_char: Character offset of ``text`` within the full document.
+
+    Returns:
+        List of (chunk_text, start_char, end_char) tuples.
+    """
+    if not text.strip():
+        return []
+
+    words = text.split()
+    if len(words) <= target_words:
+        return [(text.strip(), base_start_char, base_start_char + len(text.strip()))]
+
+    # Try sentence splitting first
+    sentences = re.split(r"(?<=[.!?])\s+", text)
+    chunks: list[tuple[str, int, int]] = []
+    current_words: list[str] = []
+    current_start = base_start_char
+
+    for sentence in sentences:
+        sentence_words = sentence.split()
+        if current_words and len(current_words) + len(sentence_words) > target_words:
+            chunk_text = " ".join(current_words).strip()
+            if chunk_text:
+                chunks.append((
+                    chunk_text,
+                    current_start,
+                    current_start + len(chunk_text),
+                ))
+            current_start = current_start + len(chunk_text) + 1
+            current_words = sentence_words
+        else:
+            current_words.extend(sentence_words)
+
+    # Flush remaining
+    if current_words:
+        chunk_text = " ".join(current_words).strip()
+        if chunk_text:
+            chunks.append((
+                chunk_text,
+                current_start,
+                current_start + len(chunk_text),
+            ))
+
+    return chunks
+
+
+def chunk_document(
+    text: str,
+    page_offsets: list[dict[str, int]] | None = None,
+    child_words: int = CHILD_CHUNK_WORDS,
+    parent_words: int = PARENT_CHUNK_WORDS,
+) -> list[ChunkInfo]:
+    """Chunk a document into parent-child chunks with offset preservation.
+
+    The strategy is:
+    1. Segment text into resume sections (experience, skills, etc.)
+    2. Within each section, create parent chunks (~700 words)
+    3. Within each parent, create child chunks (~180 words)
+    4. Children carry ``parent_chunk_id``; parents carry ``is_parent=True``
+
+    Args:
+        text: The full document text.
+        page_offsets: Page boundary data from the parser.
+        child_words: Target words per child chunk.
+        parent_words: Target words per parent chunk.
+
+    Returns:
+        All chunks (parents and children), ordered by chunk_index.
+    """
+    if not text or not text.strip():
+        return []
+
+    offsets = page_offsets or []
+    sections = _segment_into_sections(text)
+
+    all_chunks: list[ChunkInfo] = []
+    chunk_index = 0
+
+    for section in sections:
+        # Create parent chunks from the section
+        parent_spans = _split_into_word_chunks(
+            section.text, parent_words, section.start_char
+        )
+
+        for parent_text, parent_start, parent_end in parent_spans:
+            parent_id = uuid.uuid4()
+
+            # Create the parent chunk
+            parent_chunk = ChunkInfo(
+                chunk_id=parent_id,
+                parent_chunk_id=None,
+                chunk_index=chunk_index,
+                content=parent_text,
+                section=section.label,
+                page_from=_resolve_page(parent_start, offsets),
+                page_to=_resolve_page(parent_end, offsets),
+                start_char=parent_start,
+                end_char=parent_end,
+                token_count=_estimate_tokens(parent_text),
+                is_parent=True,
+            )
+            all_chunks.append(parent_chunk)
+            chunk_index += 1
+
+            # Create child chunks within this parent
+            child_spans = _split_into_word_chunks(
+                parent_text, child_words, parent_start
+            )
+
+            # If the parent is small enough, it serves as its own child
+            if len(child_spans) <= 1:
+                child_chunk = ChunkInfo(
+                    chunk_id=uuid.uuid4(),
+                    parent_chunk_id=parent_id,
+                    chunk_index=chunk_index,
+                    content=parent_text,
+                    section=section.label,
+                    page_from=parent_chunk.page_from,
+                    page_to=parent_chunk.page_to,
+                    start_char=parent_start,
+                    end_char=parent_end,
+                    token_count=parent_chunk.token_count,
+                    is_parent=False,
+                )
+                all_chunks.append(child_chunk)
+                chunk_index += 1
+            else:
+                for child_text, child_start, child_end in child_spans:
+                    child_chunk = ChunkInfo(
+                        chunk_id=uuid.uuid4(),
+                        parent_chunk_id=parent_id,
+                        chunk_index=chunk_index,
+                        content=child_text,
+                        section=section.label,
+                        page_from=_resolve_page(child_start, offsets),
+                        page_to=_resolve_page(child_end, offsets),
+                        start_char=child_start,
+                        end_char=child_end,
+                        token_count=_estimate_tokens(child_text),
+                        is_parent=False,
+                    )
+                    all_chunks.append(child_chunk)
+                    chunk_index += 1
+
+    logger.info(
+        "document_chunked",
+        total_chunks=len(all_chunks),
+        parent_chunks=sum(1 for c in all_chunks if c.is_parent),
+        child_chunks=sum(1 for c in all_chunks if not c.is_parent),
+        sections=len(sections),
+    )
+    return all_chunks

serving/app/services/embedding.py

@@ -0,0 +1,326 @@
+"""Embedding service port and adapters.
+
+The ``EmbeddingService`` protocol abstracts the embedding provider so the rest
+of the application depends on a capability, not on a deployment choice.  Two
+adapters are provided:
+
+* ``TEIEmbeddingService`` — calls HuggingFace Text Embeddings Inference over
+  HTTP.  This is the production path.
+* ``MockEmbeddingService`` — returns deterministic vectors of the configured
+  dimension.  Used in tests; not a fake implementation, just a test double.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import math
+from typing import Protocol, runtime_checkable
+
+import httpx
+
+from app.config import Settings, get_settings
+from app.exceptions import EmbeddingServiceUnavailableError
+from app.logging import get_logger
+
+logger = get_logger(__name__)
+
+# ---------------------------------------------------------------------------
+# Port
+# ---------------------------------------------------------------------------
+
+
+@runtime_checkable
+class EmbeddingService(Protocol):
+    """Capability: embed text into dense vectors."""
+
+    @property
+    def model_name(self) -> str:
+        """The model identifier used for provenance tracking."""
+        ...
+
+    @property
+    def model_version(self) -> str:
+        """A version tag for the embedder, stored on every chunk row."""
+        ...
+
+    @property
+    def dimension(self) -> int:
+        """Dimensionality of the vectors this service produces."""
+        ...
+
+    async def embed_texts(self, texts: list[str]) -> list[list[float]]:
+        """Embed a batch of passage texts.
+
+        Args:
+            texts: The texts to embed.
+
+        Returns:
+            A list of embedding vectors, one per input text.
+
+        Raises:
+            EmbeddingServiceUnavailableError: If the service is unreachable.
+        """
+        ...
+
+    async def embed_query(self, query: str) -> list[float]:
+        """Embed a single search query.
+
+        Some models use different prefixes for queries vs passages.  This
+        method handles that distinction.
+
+        Args:
+            query: The search query text.
+
+        Returns:
+            The embedding vector.
+
+        Raises:
+            EmbeddingServiceUnavailableError: If the service is unreachable.
+        """
+        ...
+
+
+# ---------------------------------------------------------------------------
+# TEI adapter (production)
+# ---------------------------------------------------------------------------
+
+_TEI_EMBED_PATH = "/embed"
+_TEI_TIMEOUT_SECONDS = 30.0
+
+
+class TEIEmbeddingService:
+    """Calls HuggingFace Text Embeddings Inference over HTTP.
+
+    TEI serves models like ``bge-m3`` and exposes a simple JSON API.  This
+    adapter batches according to ``settings.embedding_batch_size`` and
+    retries once on transient failures.
+
+    Attributes:
+        _endpoint: Base URL of the TEI instance.
+        _model: Model name for provenance.
+        _dim: Expected vector dimension.
+        _batch_size: Max texts per HTTP request.
+    """
+
+    def __init__(self, settings: Settings | None = None) -> None:
+        """Initialize from settings.
+
+        Args:
+            settings: Optional configuration override.
+        """
+        cfg = settings or get_settings()
+        self._endpoint = cfg.embedding_endpoint.rstrip("/")
+        self._model = cfg.embedding_model
+        self._dim = cfg.embedding_dim
+        self._batch_size = cfg.embedding_batch_size
+        self._version = "1"
+
+    @property
+    def model_name(self) -> str:
+        """Return the configured model identifier."""
+        return self._model
+
+    @property
+    def model_version(self) -> str:
+        """Return the embedding version tag."""
+        return self._version
+
+    @property
+    def dimension(self) -> int:
+        """Return the configured embedding dimension."""
+        return self._dim
+
+    async def embed_texts(self, texts: list[str]) -> list[list[float]]:
+        """Embed passage texts via TEI, batched.
+
+        Args:
+            texts: The texts to embed.
+
+        Returns:
+            A list of embedding vectors.
+
+        Raises:
+            EmbeddingServiceUnavailableError: On HTTP or connection error.
+        """
+        if not texts:
+            return []
+
+        all_embeddings: list[list[float]] = []
+        for i in range(0, len(texts), self._batch_size):
+            batch = texts[i : i + self._batch_size]
+            embeddings = await self._call_tei(batch)
+            all_embeddings.extend(embeddings)
+        return all_embeddings
+
+    async def embed_query(self, query: str) -> list[float]:
+        """Embed a search query via TEI.
+
+        Args:
+            query: The search query.
+
+        Returns:
+            The embedding vector.
+
+        Raises:
+            EmbeddingServiceUnavailableError: On HTTP or connection error.
+        """
+        results = await self._call_tei([query])
+        return results[0]
+
+    async def _call_tei(self, inputs: list[str]) -> list[list[float]]:
+        """Make a single TEI embed request.
+
+        Args:
+            inputs: Batch of texts to embed.
+
+        Returns:
+            The embedding vectors.
+
+        Raises:
+            EmbeddingServiceUnavailableError: On any failure.
+        """
+        url = f"{self._endpoint}{_TEI_EMBED_PATH}"
+        payload = {"inputs": inputs}
+        try:
+            async with httpx.AsyncClient(timeout=_TEI_TIMEOUT_SECONDS) as client:
+                response = await client.post(url, json=payload)
+                response.raise_for_status()
+                data: list[list[float]] = response.json()
+                return data
+        except (httpx.HTTPError, httpx.ConnectError, Exception) as exc:
+            logger.error(
+                "embedding_service_error",
+                endpoint=self._endpoint,
+                error=str(exc)[:200],
+            )
+            raise EmbeddingServiceUnavailableError(
+                f"Embedding service at {self._endpoint} is unavailable."
+            ) from exc
+
+
+# ---------------------------------------------------------------------------
+# Mock adapter (tests)
+# ---------------------------------------------------------------------------
+
+
+class MockEmbeddingService:
+    """Returns deterministic vectors for testing.
+
+    Vectors are derived from a hash of the input text so the same text always
+    produces the same vector, enabling assertions in tests.  This is *not* a
+    fake implementation — it makes no attempt at semantic similarity.
+    """
+
+    def __init__(self, dimension: int = 1024) -> None:
+        """Initialize with a target dimension.
+
+        Args:
+            dimension: Vector dimensionality to produce.
+        """
+        self._dim = dimension
+
+    @property
+    def model_name(self) -> str:
+        """Return a test model identifier."""
+        return "mock-embedder"
+
+    @property
+    def model_version(self) -> str:
+        """Return a test version tag."""
+        return "test-v1"
+
+    @property
+    def dimension(self) -> int:
+        """Return the configured dimension."""
+        return self._dim
+
+    async def embed_texts(self, texts: list[str]) -> list[list[float]]:
+        """Return deterministic vectors for each text.
+
+        Args:
+            texts: The texts to embed.
+
+        Returns:
+            Deterministic vectors derived from text hashes.
+        """
+        return [self._deterministic_vector(t) for t in texts]
+
+    async def embed_query(self, query: str) -> list[float]:
+        """Return a deterministic vector for the query.
+
+        Args:
+            query: The search query.
+
+        Returns:
+            A deterministic vector.
+        """
+        return self._deterministic_vector(query)
+
+    def _deterministic_vector(self, text: str) -> list[float]:
+        """Produce a unit-length vector deterministically from text.
+
+        Args:
+            text: Input text.
+
+        Returns:
+            A normalized vector of ``self._dim`` dimensions.
+        """
+        digest = hashlib.sha256(text.encode("utf-8")).digest()
+        # Expand the 32-byte hash to fill the dimension by repeating
+        raw = list(digest) * ((self._dim // len(digest)) + 1)
+        raw = raw[: self._dim]
+        # Map bytes to floats in [-1, 1] and normalize
+        vec = [(b - 128) / 128.0 for b in raw]
+        norm = math.sqrt(sum(v * v for v in vec)) or 1.0
+        return [v / norm for v in vec]
+
+
+# ---------------------------------------------------------------------------
+# Singleton management
+# ---------------------------------------------------------------------------
+
+_embedding_service: EmbeddingService | None = None
+
+
+def get_embedding_service() -> EmbeddingService:
+    """Return the process-wide embedding service instance.
+
+    Returns a ``MockEmbeddingService`` when no TEI endpoint is configured
+    (i.e. in test or when the endpoint is empty).
+
+    Returns:
+        The embedding service singleton.
+    """
+    global _embedding_service  # noqa: PLW0603
+    if _embedding_service is not None:
+        return _embedding_service
+
+    settings = get_settings()
+    if settings.embedding_endpoint and settings.environment != "test":
+        _embedding_service = TEIEmbeddingService(settings)
+        logger.info(
+            "embedding_service_initialized",
+            adapter="TEI",
+            endpoint=settings.embedding_endpoint,
+            model=settings.embedding_model,
+        )
+    else:
+        _embedding_service = MockEmbeddingService(dimension=settings.embedding_dim)
+        logger.info("embedding_service_initialized", adapter="Mock")
+    return _embedding_service
+
+
+def set_embedding_service(service: EmbeddingService) -> None:
+    """Override the embedding service singleton (for tests).
+
+    Args:
+        service: The service to install.
+    """
+    global _embedding_service  # noqa: PLW0603
+    _embedding_service = service
+
+
+def reset_embedding_service() -> None:
+    """Clear the embedding service singleton (for tests)."""
+    global _embedding_service  # noqa: PLW0603
+    _embedding_service = None

serving/app/services/reranker.py

@@ -0,0 +1,209 @@
+"""Reranker service port and adapters.
+
+Cross-encoder reranking improves precision by scoring query–document pairs with
+a more powerful model after initial recall.  Two adapters:
+
+* ``TEIRerankerService`` — calls HuggingFace TEI rerank endpoint.
+* ``NoOpRerankerService`` — passes through scores unchanged (CPU fallback).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Protocol, runtime_checkable
+
+import httpx
+
+from app.config import Settings, get_settings
+from app.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+@dataclass(frozen=True, slots=True)
+class RerankResult:
+    """One reranked document.
+
+    Attributes:
+        index: Original index in the input list.
+        score: Relevance score from the reranker.
+    """
+
+    index: int
+    score: float
+
+
+# ---------------------------------------------------------------------------
+# Port
+# ---------------------------------------------------------------------------
+
+
+@runtime_checkable
+class RerankerService(Protocol):
+    """Capability: rerank documents by relevance to a query."""
+
+    async def rerank(
+        self, query: str, documents: list[str], top_k: int
+    ) -> list[RerankResult]:
+        """Rerank documents by relevance to the query.
+
+        Args:
+            query: The search query.
+            documents: Document texts to rerank.
+            top_k: Maximum results to return.
+
+        Returns:
+            The top-k results sorted by descending score.
+        """
+        ...
+
+
+# ---------------------------------------------------------------------------
+# TEI adapter (production)
+# ---------------------------------------------------------------------------
+
+_TEI_RERANK_PATH = "/rerank"
+_TEI_RERANK_TIMEOUT = 30.0
+
+
+class TEIRerankerService:
+    """Calls HuggingFace TEI rerank endpoint.
+
+    Attributes:
+        _endpoint: Base URL of the TEI reranker instance.
+    """
+
+    def __init__(self, settings: Settings | None = None) -> None:
+        """Initialize from settings.
+
+        Args:
+            settings: Optional configuration override.
+        """
+        cfg = settings or get_settings()
+        self._endpoint = cfg.reranker_endpoint.rstrip("/")
+        self._model = cfg.reranker_model
+
+    async def rerank(
+        self, query: str, documents: list[str], top_k: int
+    ) -> list[RerankResult]:
+        """Rerank via TEI.
+
+        Args:
+            query: The search query.
+            documents: Document texts.
+            top_k: Maximum results.
+
+        Returns:
+            Reranked results sorted by descending score.
+        """
+        if not documents:
+            return []
+
+        url = f"{self._endpoint}{_TEI_RERANK_PATH}"
+        payload = {
+            "query": query,
+            "texts": documents,
+            "return_text": False,
+        }
+        try:
+            async with httpx.AsyncClient(timeout=_TEI_RERANK_TIMEOUT) as client:
+                response = await client.post(url, json=payload)
+                response.raise_for_status()
+                raw: list[dict[str, object]] = response.json()
+                results = [
+                    RerankResult(index=int(r["index"]), score=float(r["score"]))
+                    for r in raw
+                ]
+                results.sort(key=lambda r: r.score, reverse=True)
+                return results[:top_k]
+        except (httpx.HTTPError, Exception) as exc:
+            logger.warning(
+                "reranker_service_error",
+                endpoint=self._endpoint,
+                error=str(exc)[:200],
+            )
+            # Graceful degradation: return original order with default scores
+            return [
+                RerankResult(index=i, score=1.0 / (i + 1))
+                for i in range(min(top_k, len(documents)))
+            ]
+
+
+# ---------------------------------------------------------------------------
+# No-op adapter (CPU fallback / dev mode)
+# ---------------------------------------------------------------------------
+
+
+class NoOpRerankerService:
+    """Passes through documents without reranking.
+
+    Used when no reranker endpoint is configured.  Returns scores based on
+    input order so the RRF-fused ranking is preserved.
+    """
+
+    async def rerank(
+        self, query: str, documents: list[str], top_k: int
+    ) -> list[RerankResult]:
+        """Return input-order scores without reranking.
+
+        Args:
+            query: Ignored.
+            documents: Document texts.
+            top_k: Maximum results.
+
+        Returns:
+            Results in input order with position-based scores.
+        """
+        return [
+            RerankResult(index=i, score=1.0 / (i + 1))
+            for i in range(min(top_k, len(documents)))
+        ]
+
+
+# ---------------------------------------------------------------------------
+# Singleton management
+# ---------------------------------------------------------------------------
+
+_reranker_service: RerankerService | None = None
+
+
+def get_reranker_service() -> RerankerService:
+    """Return the process-wide reranker service instance.
+
+    Returns a ``NoOpRerankerService`` when no reranker endpoint is configured.
+
+    Returns:
+        The reranker service singleton.
+    """
+    global _reranker_service  # noqa: PLW0603
+    if _reranker_service is not None:
+        return _reranker_service
+
+    settings = get_settings()
+    if settings.reranker_endpoint:
+        _reranker_service = TEIRerankerService(settings)
+        logger.info(
+            "reranker_service_initialized",
+            adapter="TEI",
+            endpoint=settings.reranker_endpoint,
+        )
+    else:
+        _reranker_service = NoOpRerankerService()
+        logger.info("reranker_service_initialized", adapter="NoOp")
+    return _reranker_service
+
+
+def set_reranker_service(service: RerankerService) -> None:
+    """Override the reranker service singleton (for tests).
+
+    Args:
+        service: The service to install.
+    """
+    global _reranker_service  # noqa: PLW0603
+    _reranker_service = service
+
+
+def reset_reranker_service() -> None:
+    """Clear the reranker service singleton (for tests)."""
+    global _reranker_service  # noqa: PLW0603
+    _reranker_service = None

serving/app/services/search.py

@@ -0,0 +1,349 @@
+"""Hybrid semantic and lexical talent search.
+
+Orchestrates the retrieval pipeline described in ARCHITECTURE.md §12.4:
+query → embed → hybrid recall (dense + lexical) → RRF fusion → rerank →
+group by candidate → aggregate → return ranked results with evidence spans.
+
+No LLM in this path — search is sub-second and free.
+"""
+
+from __future__ import annotations
+
+import uuid
+from collections import defaultdict
+from dataclasses import dataclass, field
+from enum import Enum
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import Settings, get_settings
+from app.logging import get_logger
+from app.repositories.search import ChunkRepository, ChunkWithScore
+from app.services.embedding import EmbeddingService
+from app.services.reranker import RerankerService
+
+logger = get_logger(__name__)
+
+
+class SearchMode(str, Enum):
+    """Search retrieval strategy."""
+
+    HYBRID = "hybrid"
+    SEMANTIC = "semantic"
+    LEXICAL = "lexical"
+
+
+@dataclass(frozen=True, slots=True)
+class EvidenceSpan:
+    """One piece of supporting evidence from a search result.
+
+    Attributes:
+        chunk_id: The chunk this evidence comes from.
+        content: The chunk text.
+        section: Resume section (experience, skills, etc.).
+        page_from: Starting page (0-based).
+        page_to: Ending page (0-based).
+        start_char: Starting character offset in the full document.
+        end_char: Ending character offset in the full document.
+        score: Retrieval score for this chunk.
+    """
+
+    chunk_id: uuid.UUID
+    content: str
+    section: str
+    page_from: int
+    page_to: int
+    start_char: int
+    end_char: int
+    score: float
+
+
+@dataclass(frozen=True, slots=True)
+class CandidateSearchHit:
+    """One candidate in search results.
+
+    Attributes:
+        document_id: The resume document ID.
+        score: Aggregated relevance score across all matching chunks.
+        spans: The best supporting evidence spans.
+    """
+
+    document_id: uuid.UUID
+    score: float
+    spans: list[EvidenceSpan] = field(default_factory=list)
+
+
+@dataclass(frozen=True, slots=True)
+class SearchResult:
+    """Complete search result set.
+
+    Attributes:
+        items: Ranked candidate hits.
+        count: Total number of candidates returned.
+        query: The original query text.
+        mode: The search mode used.
+    """
+
+    items: list[CandidateSearchHit]
+    count: int
+    query: str
+    mode: str
+
+
+def reciprocal_rank_fusion(
+    ranked_lists: list[list[ChunkWithScore]],
+    weights: list[float],
+    k: int = 60,
+) -> list[ChunkWithScore]:
+    """Fuse multiple ranked lists using Reciprocal Rank Fusion.
+
+    RRF score for a document d = sum(weight_i / (k + rank_i(d))) across lists.
+
+    Args:
+        ranked_lists: Lists of ranked chunks from different retrieval methods.
+        weights: Weight for each list (must sum to ~1.0).
+        k: RRF constant (default 60, standard value).
+
+    Returns:
+        A single fused list sorted by descending RRF score.
+    """
+    scores: dict[uuid.UUID, float] = defaultdict(float)
+    chunk_map: dict[uuid.UUID, ChunkWithScore] = {}
+
+    for ranked_list, weight in zip(ranked_lists, weights, strict=False):
+        for rank, item in enumerate(ranked_list):
+            rrf_score = weight / (k + rank + 1)
+            scores[item.chunk.id] += rrf_score
+            # Keep the chunk with the highest individual score
+            if item.chunk.id not in chunk_map or item.score > chunk_map[item.chunk.id].score:
+                chunk_map[item.chunk.id] = item
+
+    # Build fused results
+    fused = [
+        ChunkWithScore(chunk=chunk_map[chunk_id].chunk, score=score)
+        for chunk_id, score in scores.items()
+    ]
+    fused.sort(key=lambda x: x.score, reverse=True)
+    return fused
+
+
+def group_by_candidate(
+    chunks: list[ChunkWithScore],
+    max_spans_per_candidate: int = 5,
+) -> list[CandidateSearchHit]:
+    """Group chunks by candidate (document_id) and aggregate scores.
+
+    Per-candidate score is the sum of the best chunk scores (not mean, so
+    candidates with more matching evidence rank higher).
+
+    Args:
+        chunks: Scored chunks from retrieval.
+        max_spans_per_candidate: Maximum evidence spans to include per candidate.
+
+    Returns:
+        Candidates sorted by descending aggregate score.
+    """
+    by_document: dict[uuid.UUID, list[ChunkWithScore]] = defaultdict(list)
+    for chunk in chunks:
+        by_document[chunk.chunk.document_id].append(chunk)
+
+    hits: list[CandidateSearchHit] = []
+    for doc_id, doc_chunks in by_document.items():
+        # Sort chunks by score descending
+        doc_chunks.sort(key=lambda c: c.score, reverse=True)
+        best = doc_chunks[:max_spans_per_candidate]
+
+        # Aggregate score: sum of best chunk scores
+        total_score = sum(c.score for c in best)
+
+        spans = [
+            EvidenceSpan(
+                chunk_id=c.chunk.id,
+                content=c.chunk.content,
+                section=c.chunk.section,
+                page_from=c.chunk.page_from,
+                page_to=c.chunk.page_to,
+                start_char=c.chunk.start_char,
+                end_char=c.chunk.end_char,
+                score=c.score,
+            )
+            for c in best
+        ]
+        hits.append(CandidateSearchHit(
+            document_id=doc_id,
+            score=total_score,
+            spans=spans,
+        ))
+
+    hits.sort(key=lambda h: h.score, reverse=True)
+    return hits
+
+
+async def search_candidates(
+    *,
+    session: AsyncSession,
+    embedder: EmbeddingService,
+    reranker: RerankerService,
+    tenant_id: uuid.UUID,
+    query: str,
+    top_k: int = 10,
+    mode: SearchMode = SearchMode.HYBRID,
+    settings: Settings | None = None,
+) -> SearchResult:
+    """Execute a hybrid talent search across the candidate pool.
+
+    Pipeline:
+    1. Embed query
+    2. Dense retrieval (pgvector HNSW cosine)
+    3. Lexical retrieval (GIN tsvector ts_rank_cd)
+    4. RRF fusion (configurable weights)
+    5. Cross-encoder rerank
+    6. Group by candidate, aggregate scores
+    7. Return ranked results with evidence spans
+
+    Args:
+        session: Database session.
+        embedder: Embedding service.
+        reranker: Reranker service.
+        tenant_id: Owning tenant.
+        query: Natural-language search query.
+        top_k: Maximum candidates to return.
+        mode: Retrieval strategy.
+        settings: Optional configuration override.
+
+    Returns:
+        Ranked search results with evidence spans.
+    """
+    cfg = settings or get_settings()
+    repo = ChunkRepository(session)
+
+    recall_k = cfg.search_top_k_recall
+
+    # Step 1: Retrieve
+    if mode == SearchMode.HYBRID:
+        query_embedding = await embedder.embed_query(query)
+        dense_results = await repo.search_dense(tenant_id, query_embedding, recall_k)
+        lexical_results = await repo.search_lexical(tenant_id, query, recall_k)
+
+        # Step 2: RRF fusion
+        fused = reciprocal_rank_fusion(
+            [dense_results, lexical_results],
+            [cfg.search_dense_weight, cfg.search_lexical_weight],
+        )
+    elif mode == SearchMode.SEMANTIC:
+        query_embedding = await embedder.embed_query(query)
+        fused = await repo.search_dense(tenant_id, query_embedding, recall_k)
+    else:  # LEXICAL
+        fused = await repo.search_lexical(tenant_id, query, recall_k)
+
+    if not fused:
+        return SearchResult(items=[], count=0, query=query, mode=mode.value)
+
+    # Step 3: Rerank
+    rerank_input = [c.chunk.content for c in fused]
+    rerank_results = await reranker.rerank(
+        query, rerank_input, cfg.search_rerank_top_k
+    )
+
+    # Map rerank results back to chunks
+    reranked: list[ChunkWithScore] = []
+    for rr in rerank_results:
+        if rr.index < len(fused):
+            reranked.append(
+                ChunkWithScore(chunk=fused[rr.index].chunk, score=rr.score)
+            )
+
+    # Step 4: Group by candidate
+    candidates = group_by_candidate(reranked or fused)
+
+    # Step 5: Trim to top_k
+    candidates = candidates[:top_k]
+
+    logger.info(
+        "search_completed",
+        tenant_id=str(tenant_id),
+        query_length=len(query),
+        mode=mode.value,
+        candidates_returned=len(candidates),
+        total_chunks_recalled=len(fused),
+    )
+
+    return SearchResult(
+        items=candidates,
+        count=len(candidates),
+        query=query,
+        mode=mode.value,
+    )
+
+
+async def search_similar(
+    *,
+    session: AsyncSession,
+    embedder: EmbeddingService,
+    reranker: RerankerService,
+    tenant_id: uuid.UUID,
+    document_id: uuid.UUID,
+    top_k: int = 10,
+    settings: Settings | None = None,
+) -> SearchResult:
+    """Find candidates similar to a given document ("more like this").
+
+    Embeds the candidate's chunks, aggregates them, and searches for similar
+    chunks across the corpus — excluding the source document itself.
+
+    Args:
+        session: Database session.
+        embedder: Embedding service.
+        reranker: Reranker service.
+        tenant_id: Owning tenant.
+        document_id: Source document to find similar candidates for.
+        top_k: Maximum candidates to return.
+        settings: Optional configuration override.
+
+    Returns:
+        Ranked search results for similar candidates.
+    """
+    cfg = settings or get_settings()
+    repo = ChunkRepository(session)
+
+    # Get the source document's chunks
+    source_chunks = await repo.get_by_document(
+        tenant_id, document_id, children_only=True
+    )
+
+    if not source_chunks:
+        return SearchResult(
+            items=[], count=0, query=f"similar:{document_id}", mode="similar"
+        )
+
+    # Use the source chunks' content to build a representative query
+    # Take the first few chunks to avoid exceeding embedding limits
+    representative_texts = [c.content for c in source_chunks[:5]]
+    query_text = " ".join(representative_texts)[:2000]
+
+    query_embedding = await embedder.embed_query(query_text)
+
+    # Search across all chunks, then filter out the source document
+    all_results = await repo.search_dense(
+        tenant_id, query_embedding, cfg.search_top_k_recall * 2
+    )
+
+    # Filter out chunks from the source document
+    filtered = [r for r in all_results if r.chunk.document_id != document_id]
+
+    # Group by candidate
+    candidates = group_by_candidate(filtered)[:top_k]
+
+    logger.info(
+        "similar_search_completed",
+        tenant_id=str(tenant_id),
+        source_document_id=str(document_id),
+        candidates_returned=len(candidates),
+    )
+
+    return SearchResult(
+        items=candidates,
+        count=len(candidates),
+        query=f"similar:{document_id}",
+        mode="similar",
+    )