feat(ingestion): add bounded, benchmark-aligned embedding execution

Why:
- Plan 001 Phase 4 needs batched, concurrency-bounded embedding wired into
  the inline upload path, with process-wide capacity/timeout/chunk-limit
  guards (ADR-0017).
- The BM25 analyzer and dense-model config are ported from the `emet`
  evaluation lab, which benchmarked them against the real Farsi corpus
  (bm25-fa-norm-stop; nomic-embed-text-v2-moe at 768-dim; text-embedding-3-large
  at native 3072-dim), closing open items in ADR-0001/ADR-0005.

Changes:
- New: embedding ports, orchestration (embed_chunks), request-bounds
  helpers, and dense/sparse adapters (analyzers.py, bm25.py,
  openai_compatible.py).
- upload.py now parses/chunks/embeds inline behind INGESTION_MAX_CONCURRENCY
  (503), INGESTION_TIMEOUT_SECONDS (504), and the chunk-count ceiling (413);
  every failure path still writes a terminal job row.
- Lifespan builds and warms both dense embedders at startup (fail-soft) and
  creates the sparse embedder and concurrency semaphore.
- httpx moves from dev to main dependencies (adapters use it directly).

Impact:
- Qdrant point upserts are still Phase 5 -- chunks_indexed stays 0.
- New EMBEDDING_* env vars documented in .env.example; safe defaults.

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

src/application/files/upload.py

@@ -5,21 +5,27 @@ one request-scoped session, because the request is two units of work
 (ADR-0012, ADR-0017):
 
     txn A (short):  source_files [+ ingestion_jobs(status='running')], commit
-    no txn:          store bytes in MinIO
+    no txn:          store bytes in MinIO, parse/chunk (threads),
+                     embed dense+sparse (bounded/batched)
     txn B (short):  ingestion_jobs -> succeeded/failed, append event, commit
 
-No Postgres session is open during the MinIO write. A storage failure between
+No Postgres session is open during phase 2. A failure at any point between
 txn A and txn B still leaves a durable, inspectable `failed` job — never a
-job stuck in `running`.
+job stuck in `running`. The whole request additionally holds one of
+`INGESTION_MAX_CONCURRENCY` process-wide slots (`503` when exhausted) and
+phase 2 is bounded by `INGESTION_TIMEOUT_SECONDS` (`504`) (ADR-0017, plan 001
+Phase 4).
 
-Parsing/chunking/Qdrant indexing are Phase 4/5 work, not implemented here:
-this phase stores bytes only, so a successful job reports `chunks_indexed=0`.
+Qdrant point upserts are Phase 5 work, not implemented here: this phase
+parses, chunks, and embeds, so a successful job still reports
+`chunks_indexed=0` — nothing is searchable yet.
 """
 
 import uuid
+from collections.abc import Sequence
 
 import structlog
-from anyio import CapacityLimiter, to_thread
+from anyio import CapacityLimiter, Semaphore, fail_after, to_thread
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from src.application.auth.context import AuthContext
@@ -27,7 +33,22 @@ from src.application.files.errors import InvalidUploadError
 from src.application.files.models import UploadResult
 from src.application.files.storage_keys import source_file_object_key
 from src.application.files.validation import validate_and_hash_upload
+from src.application.ingestion import (
+    ChunkTooLargeError,
+    DocumentParseError,
+    UnsupportedSourceTypeError,
+    parse_and_chunk_document,
+)
+from src.application.ingestion.bounds import acquire_ingestion_slot, enforce_chunk_limit
+from src.application.ingestion.embedding import embed_chunks
+from src.application.ingestion.errors import (
+    ChunkLimitExceededError,
+    EmbedderError,
+    IngestionTimeoutError,
+)
+from src.application.ports.embedding import DenseEmbedder, SparseEmbedder
 from src.application.ports.object_storage import ObjectStorage
+from src.config import ChunkingSettings, IngestionSettings
 from src.infrastructure.postgres.repositories import ingestion_jobs as jobs_repo
 from src.infrastructure.postgres.repositories import source_files as source_files_repo
 
@@ -71,9 +92,12 @@ async def upload_source_file(
     domain: str,
     filename: str,
     data: bytes,
-    max_upload_size_bytes: int,
-    chunking_strategy: str,
-    validation_limiter: CapacityLimiter,
+    ingestion_settings: IngestionSettings,
+    chunking_settings: ChunkingSettings,
+    thread_limiter: CapacityLimiter,
+    concurrency_limiter: Semaphore,
+    dense_embedders: Sequence[DenseEmbedder],
+    sparse_embedder: SparseEmbedder,
 ) -> UploadResult:
     domain = domain.strip()
     if not domain:
@@ -81,114 +105,186 @@ async def upload_source_file(
 
     validated = await to_thread.run_sync(
         lambda: validate_and_hash_upload(
-            filename=filename, data=data, max_size_bytes=max_upload_size_bytes
+            filename=filename, data=data, max_size_bytes=ingestion_settings.max_upload_size_bytes
         ),
-        limiter=validation_limiter,
+        limiter=thread_limiter,
     )
 
-    async with sessionmaker() as session:
-        existing = await source_files_repo.find_active_by_content_hash(
-            session,
-            tenant_id=auth.tenant_id,
-            domain=domain,
-            content_sha256=validated.content_sha256,
-        )
-
-        if existing is not None:
-            latest_job = await jobs_repo.get_latest_for_source_file(
-                session, tenant_id=auth.tenant_id, source_file_id=existing.id
-            )
-            if latest_job is not None and latest_job.status == "succeeded":
-                logger.info(
-                    "files.upload.duplicate",
-                    tenant_id=str(auth.tenant_id),
-                    file_id=str(existing.id),
-                )
-                return UploadResult(
-                    file_id=existing.id,
-                    ingestion_job_id=latest_job.id,
-                    status=latest_job.status,
-                    chunks_indexed=latest_job.points_created,
-                    is_new_attempt=False,
-                )
-            source_file_id = existing.id
-            object_key = existing.storage_uri or source_file_object_key(
-                auth.tenant_id, source_file_id
-            )
-        else:
-            source_file_id = uuid.uuid4()
-            object_key = source_file_object_key(auth.tenant_id, source_file_id)
-            source_files_repo.create(
+    async with acquire_ingestion_slot(concurrency_limiter):
+        async with sessionmaker() as session:
+            existing = await source_files_repo.find_active_by_content_hash(
                 session,
-                source_file_id=source_file_id,
                 tenant_id=auth.tenant_id,
                 domain=domain,
-                source_filename=filename,
-                source_type=validated.source_type,
                 content_sha256=validated.content_sha256,
-                byte_size=len(data),
-                storage_uri=object_key,
-                created_by_api_key_id=auth.api_key_id,
             )
-            # `ingestion_jobs.source_file_id` FKs to this row; flush so the
-            # insert below sees it, since the two mapped classes carry no
-            # ORM relationship for the unit of work to order by itself.
-            await session.flush()
 
-        job = jobs_repo.create_running(
-            session,
-            tenant_id=auth.tenant_id,
-            source_file_id=source_file_id,
-            requested_by_api_key_id=auth.api_key_id,
-            chunking_strategy=chunking_strategy,
-        )
-        jobs_repo.append_event(
-            session,
-            tenant_id=auth.tenant_id,
-            ingestion_job_id=job.id,
-            level="info",
-            stage="received",
-            message="upload accepted, storing object",
-        )
-        await session.commit()
-        ingestion_job_id = job.id
+            if existing is not None:
+                latest_job = await jobs_repo.get_latest_for_source_file(
+                    session, tenant_id=auth.tenant_id, source_file_id=existing.id
+                )
+                if latest_job is not None and latest_job.status == "succeeded":
+                    logger.info(
+                        "files.upload.duplicate",
+                        tenant_id=str(auth.tenant_id),
+                        file_id=str(existing.id),
+                    )
+                    return UploadResult(
+                        file_id=existing.id,
+                        ingestion_job_id=latest_job.id,
+                        status=latest_job.status,
+                        chunks_indexed=latest_job.points_created,
+                        is_new_attempt=False,
+                    )
+                source_file_id = existing.id
+                object_key = existing.storage_uri or source_file_object_key(
+                    auth.tenant_id, source_file_id
+                )
+            else:
+                source_file_id = uuid.uuid4()
+                object_key = source_file_object_key(auth.tenant_id, source_file_id)
+                source_files_repo.create(
+                    session,
+                    source_file_id=source_file_id,
+                    tenant_id=auth.tenant_id,
+                    domain=domain,
+                    source_filename=filename,
+                    source_type=validated.source_type,
+                    content_sha256=validated.content_sha256,
+                    byte_size=len(data),
+                    storage_uri=object_key,
+                    created_by_api_key_id=auth.api_key_id,
+                )
+                # `ingestion_jobs.source_file_id` FKs to this row; flush so the
+                # insert below sees it, since the two mapped classes carry no
+                # ORM relationship for the unit of work to order by itself.
+                await session.flush()
 
-    # Phase 2: no Postgres session open across this work (ADR-0017).
-    try:
-        await storage.put_object(key=object_key, data=data, content_type=validated.content_type)
-    except Exception as exc:
-        logger.warning(
-            "files.upload.storage_failed",
-            tenant_id=str(auth.tenant_id),
-            file_id=str(source_file_id),
-            ingestion_job_id=str(ingestion_job_id),
-        )
-        await _mark_job_failed(
-            sessionmaker,
-            tenant_id=auth.tenant_id,
-            ingestion_job_id=ingestion_job_id,
-            error_code="storage_upload_failed",
-            error_message=f"failed to store object: {exc}",
-        )
-        raise
+            job = jobs_repo.create_running(
+                session,
+                tenant_id=auth.tenant_id,
+                source_file_id=source_file_id,
+                requested_by_api_key_id=auth.api_key_id,
+                chunking_strategy=chunking_settings.strategy,
+            )
+            jobs_repo.append_event(
+                session,
+                tenant_id=auth.tenant_id,
+                ingestion_job_id=job.id,
+                level="info",
+                stage="received",
+                message="upload accepted, storing object",
+            )
+            await session.commit()
+            ingestion_job_id = job.id
 
-    async with sessionmaker() as session:
-        await jobs_repo.mark_terminal(
-            session,
-            tenant_id=auth.tenant_id,
-            ingestion_job_id=ingestion_job_id,
-            status="succeeded",
-            points_created=0,
-        )
-        jobs_repo.append_event(
-            session,
-            tenant_id=auth.tenant_id,
-            ingestion_job_id=ingestion_job_id,
-            level="info",
-            stage="completed",
-            message="object stored; parsing/embedding/indexing not yet implemented",
-        )
-        await session.commit()
+        # Phase 2: no Postgres session open across this work (ADR-0017),
+        # bounded end-to-end by INGESTION_TIMEOUT_SECONDS.
+        try:
+            with fail_after(ingestion_settings.timeout_seconds):
+                try:
+                    await storage.put_object(
+                        key=object_key, data=data, content_type=validated.content_type
+                    )
+                except Exception as exc:
+                    logger.warning(
+                        "files.upload.storage_failed",
+                        tenant_id=str(auth.tenant_id),
+                        file_id=str(source_file_id),
+                        ingestion_job_id=str(ingestion_job_id),
+                    )
+                    await _mark_job_failed(
+                        sessionmaker,
+                        tenant_id=auth.tenant_id,
+                        ingestion_job_id=ingestion_job_id,
+                        error_code="storage_upload_failed",
+                        error_message=f"failed to store object: {exc}",
+                    )
+                    raise
+
+                try:
+                    chunks = await parse_and_chunk_document(
+                        data,
+                        source_type=validated.source_type,
+                        file_id=source_file_id,
+                        settings=chunking_settings,
+                        limiter=thread_limiter,
+                    )
+                except (DocumentParseError, UnsupportedSourceTypeError, ChunkTooLargeError) as exc:
+                    await _mark_job_failed(
+                        sessionmaker,
+                        tenant_id=auth.tenant_id,
+                        ingestion_job_id=ingestion_job_id,
+                        error_code="parse_failed",
+                        error_message=str(exc),
+                    )
+                    raise
+
+                try:
+                    enforce_chunk_limit(chunks, max_chunks=ingestion_settings.max_chunks_per_file)
+                except ChunkLimitExceededError as exc:
+                    await _mark_job_failed(
+                        sessionmaker,
+                        tenant_id=auth.tenant_id,
+                        ingestion_job_id=ingestion_job_id,
+                        error_code="chunk_limit_exceeded",
+                        error_message=str(exc),
+                    )
+                    raise
+
+                try:
+                    embedded = await embed_chunks(
+                        chunks,
+                        dense_embedders=dense_embedders,
+                        sparse_embedder=sparse_embedder,
+                        settings=ingestion_settings,
+                        thread_limiter=thread_limiter,
+                    )
+                except EmbedderError as exc:
+                    await _mark_job_failed(
+                        sessionmaker,
+                        tenant_id=auth.tenant_id,
+                        ingestion_job_id=ingestion_job_id,
+                        error_code="embedding_failed",
+                        error_message=str(exc),
+                    )
+                    raise
+        except TimeoutError:
+            logger.warning(
+                "files.upload.timeout",
+                tenant_id=str(auth.tenant_id),
+                file_id=str(source_file_id),
+                ingestion_job_id=str(ingestion_job_id),
+            )
+            await _mark_job_failed(
+                sessionmaker,
+                tenant_id=auth.tenant_id,
+                ingestion_job_id=ingestion_job_id,
+                error_code="timeout",
+                error_message=f"ingestion exceeded {ingestion_settings.timeout_seconds}s",
+            )
+            raise IngestionTimeoutError(
+                f"ingestion exceeded {ingestion_settings.timeout_seconds}s"
+            ) from None
+
+        async with sessionmaker() as session:
+            await jobs_repo.mark_terminal(
+                session,
+                tenant_id=auth.tenant_id,
+                ingestion_job_id=ingestion_job_id,
+                status="succeeded",
+                points_created=0,
+            )
+            jobs_repo.append_event(
+                session,
+                tenant_id=auth.tenant_id,
+                ingestion_job_id=ingestion_job_id,
+                level="info",
+                stage="completed",
+                message="chunks parsed and embedded; Qdrant indexing not yet implemented",
+                details={"chunks_parsed": len(chunks), "chunks_embedded": len(embedded)},
+            )
+            await session.commit()
 
     logger.info(
         "files.upload.succeeded",

src/application/ingestion/bounds.py

@@ -0,0 +1,48 @@
+"""Request bounds for inline ingestion (ADR-0017).
+
+Three independent bounds, each mapping to its own status code: the chunk
+ceiling (`413`, checked before embedding starts), process-wide concurrency
+(`503` + `Retry-After`, rejected rather than queued), and the work-phase
+deadline (`504`, and the caller must still write a terminal job status).
+"""
+
+from collections.abc import AsyncIterator, Sequence
+from contextlib import asynccontextmanager
+
+from anyio import Semaphore, WouldBlock
+
+from src.application.ingestion.errors import ChunkLimitExceededError, IngestionAtCapacityError
+from src.application.ingestion.models import Chunk
+
+
+def enforce_chunk_limit(chunks: Sequence[Chunk], *, max_chunks: int) -> None:
+    """Raise `ChunkLimitExceededError` if `chunks` exceeds `max_chunks`.
+
+    Call this immediately after parsing/chunking and before any embedding
+    call — the ceiling must be discovered up front, not mid-batch.
+    """
+    if len(chunks) > max_chunks:
+        raise ChunkLimitExceededError(
+            f"document produced {len(chunks)} chunks, over the {max_chunks}-chunk limit"
+        )
+
+
+@asynccontextmanager
+async def acquire_ingestion_slot(limiter: Semaphore) -> AsyncIterator[None]:
+    """Hold one of `INGESTION_MAX_CONCURRENCY` process-wide slots for the block.
+
+    `limiter` is an `anyio.Semaphore` created once in the lifespan. Rejects
+    immediately with `IngestionAtCapacityError` when the process is already at
+    capacity, rather than queueing the request behind an unbounded wait
+    (ADR-0017) — the semaphore's own async `acquire()` would do the latter.
+    """
+    try:
+        limiter.acquire_nowait()
+    except WouldBlock:
+        raise IngestionAtCapacityError(
+            "ingestion is at capacity; retry after the configured backoff"
+        ) from None
+    try:
+        yield
+    finally:
+        limiter.release()

src/application/ingestion/embedding.py

@@ -0,0 +1,112 @@
+"""The one caller-facing entry point for embedding chunks (ADR-0001, ADR-0017).
+
+`embed_chunks` is the only version of this step callers should reach for: it
+owns batching, the `embed_concurrency` semaphore bounding in-flight dense
+batches, and the `anyio.to_thread.run_sync` + `CapacityLimiter` offload for
+the blocking BM25 pipeline. Composing these correctly at every call site is
+exactly the obligation a deep module absorbs once (see CLAUDE.md's "prefer
+deep modules").
+
+Per-provider text shaping — task prefixes, `keep_alive`, request payload —
+belongs to the adapters in `src/infrastructure/embedding/`, not here. This
+module knows only that an embedder turns texts into vectors.
+"""
+
+import asyncio
+from collections.abc import Sequence
+from functools import partial
+
+from anyio import CapacityLimiter, to_thread
+
+from src.application.ingestion.errors import EmbedderError
+from src.application.ingestion.models import Chunk, EmbeddedChunk
+from src.application.ports.embedding import DenseEmbedder, SparseEmbedder
+from src.config import IngestionSettings
+
+
+def _batches(texts: Sequence[str], size: int) -> list[Sequence[str]]:
+    return [texts[i : i + size] for i in range(0, len(texts), size)]
+
+
+async def _embed_dense_bounded(
+    embedder: DenseEmbedder,
+    batch: Sequence[str],
+    *,
+    semaphore: asyncio.Semaphore,
+) -> list[list[float]]:
+    async with semaphore:
+        try:
+            return await embedder.embed_batch(batch)
+        except Exception as exc:
+            raise EmbedderError(f"{embedder.name} embedding batch failed: {exc}") from exc
+
+
+async def _embed_dense_all(
+    embedder: DenseEmbedder,
+    texts: Sequence[str],
+    *,
+    batch_size: int,
+    semaphore: asyncio.Semaphore,
+) -> list[list[float]]:
+    batches = _batches(texts, batch_size)
+    results = await asyncio.gather(
+        *(_embed_dense_bounded(embedder, batch, semaphore=semaphore) for batch in batches)
+    )
+    return [vector for batch_result in results for vector in batch_result]
+
+
+def _embed_sparse_sync(embedder: SparseEmbedder, texts: Sequence[str]):
+    try:
+        return embedder.embed_batch(texts)
+    except Exception as exc:
+        raise EmbedderError(f"{embedder.name} embedding batch failed: {exc}") from exc
+
+
+async def embed_chunks(
+    chunks: Sequence[Chunk],
+    *,
+    dense_embedders: Sequence[DenseEmbedder],
+    sparse_embedder: SparseEmbedder,
+    settings: IngestionSettings,
+    thread_limiter: CapacityLimiter,
+) -> list[EmbeddedChunk]:
+    """Embed every chunk into all dense vectors plus the sparse vector.
+
+    Dense embedders run concurrently with each other; each one's batches are
+    concurrent among themselves too, bounded by one `embed_concurrency`
+    semaphore shared across all dense embedders (ADR-0017: the limit exists
+    for both providers' rate limits and the self-hosted server's capacity —
+    not a per-provider budget). The sparse (BM25) pass is blocking and runs
+    once, off the event loop.
+
+    Raises `EmbedderError` (502) if any embedder call fails.
+    """
+    if not chunks:
+        return []
+
+    texts = [chunk.content for chunk in chunks]
+    semaphore = asyncio.Semaphore(settings.embed_concurrency)
+
+    dense_task = asyncio.gather(
+        *(
+            _embed_dense_all(
+                embedder, texts, batch_size=settings.embed_batch_size, semaphore=semaphore
+            )
+            for embedder in dense_embedders
+        )
+    )
+    sparse_task = to_thread.run_sync(
+        partial(_embed_sparse_sync, sparse_embedder, texts), limiter=thread_limiter
+    )
+    dense_results, sparse_vectors = await asyncio.gather(dense_task, sparse_task)
+
+    dense_by_name = {
+        embedder.name: vectors
+        for embedder, vectors in zip(dense_embedders, dense_results, strict=True)
+    }
+
+    embedded: list[EmbeddedChunk] = []
+    for index, chunk in enumerate(chunks):
+        dense = {name: vectors[index] for name, vectors in dense_by_name.items()}
+        embedded.append(EmbeddedChunk(chunk=chunk, dense=dense, sparse=sparse_vectors[index]))
+    return embedded

src/application/ingestion/errors.py

@@ -39,3 +39,26 @@ class ChunkTooLargeError(IngestionError):
     against is silent — `nomic-embed-text-v2-moe` truncates over-long input
     without raising (ADR-0004).
     """
+
+
+class EmbedderError(IngestionError):
+    """A dense or sparse embedder call failed (transport error, non-2xx, or
+    a malformed response).
+
+    Maps to `502` per ADR-0017.
+    """
+
+
+class IngestionAtCapacityError(IngestionError):
+    """`INGESTION_MAX_CONCURRENCY` in-process ingestions are already running.
+
+    Maps to `503` with `Retry-After`, not a queued wait (ADR-0017).
+    """
+
+
+class IngestionTimeoutError(IngestionError):
+    """The work phase (parse/embed/upsert) exceeded `INGESTION_TIMEOUT_SECONDS`.
+
+    Maps to `504`. The caller must still write a terminal `failed` job status
+    before this propagates (ADR-0017).
+    """

src/application/ingestion/models.py

@@ -64,3 +64,28 @@ class Chunk(BaseModel):
     next_chunk_id: uuid.UUID | None = None
     token_count: int
     character_count: int
+
+
+class SparseVector(BaseModel):
+    """A sparse (term-index -> weight) vector, Qdrant's `modifier="idf"` shape.
+
+    Kept free of the `qdrant_client` SDK (ADR-0015: ports carry no infra
+    imports) — `src/infrastructure/qdrant/` converts this to the SDK's own
+    `SparseVector` type at upsert time (Phase 5).
+    """
+
+    indices: list[int]
+    values: list[float]
+
+
+class EmbeddedChunk(BaseModel):
+    """A chunk plus every vector it will be upserted with (ADR-0001).
+
+    `dense` is keyed by named-vector name (`dense_nomic`, `dense_openai`).
+    `late_interaction` is deliberately absent — not computed at ingest
+    (ADR-0017).
+    """
+
+    chunk: Chunk
+    dense: dict[str, list[float]]
+    sparse: SparseVector

src/application/ports/embedding.py

@@ -0,0 +1,49 @@
+"""Embedding ports (ADR-0001, ADR-0017).
+
+`src/infrastructure/embedding/` holds the production adapters; tests use
+scripted fakes (ADR-0016). Application code depends on these Protocols, not
+on `httpx`/provider SDKs directly.
+"""
+
+from collections.abc import Sequence
+from typing import Protocol
+
+from src.application.ingestion.models import SparseVector
+
+
+class DenseEmbedder(Protocol):
+    """One named dense vector's embedding client (`dense_nomic`/`dense_openai`).
+
+    `embed_batch` is a single batched network call — callers own concurrency
+    bounding (ADR-0017's `embed_concurrency` semaphore), not this Protocol.
+    """
+
+    name: str
+
+    async def embed_batch(self, texts: Sequence[str]) -> list[list[float]]:
+        """Return one vector per input text, same order. Raises `EmbedderError`
+        (see `src/application/ingestion/errors.py`) on transport/response
+        failure.
+        """
+        ...
+
+
+class SparseEmbedder(Protocol):
+    """The `sparse` (BM25) vector's embedding client.
+
+    Blocking/CPU-bound (ADR-0017): callers offload it via
+    `anyio.to_thread.run_sync` with the ingestion `CapacityLimiter`, not call
+    it directly from an `async def`.
+    """
+
+    name: str
+
+    def embed_batch(self, texts: Sequence[str], *, query: bool = False) -> list[SparseVector]:
+        """Return one sparse vector per input text, same order.
+
+        `query=True` selects the query-side weighting, which omits document
+        length normalization. Ingestion always passes `False`; the flag exists
+        so retrieval (ADR-0003) encodes queries through this same port rather
+        than growing a second, silently divergent implementation.
+        """
+        ...