feat(qdrant): provision the chunks collection as an explicit deployment step

Why:
- The chunks collection needs four named vectors (dense_nomic, dense_openai,
  sparse, late_interaction) and payload indexes defined at creation time per
  ADR-0001; sparse/multivector fields cannot be added to an existing
  collection without recreating it, so schema drift here is expensive.
- Creating it at FastAPI startup would mirror the DDL-at-boot anti-pattern
  ADR-0009 already rejects for Postgres and ADR-0012 rejects for LangGraph's
  setup(), so it is a deployment step instead.

Changes:
- src/infrastructure/qdrant/collection.py: ensure_chunks_collection(),
  idempotent and schema-verifying (raises on dimension/modifier mismatch
  rather than silently accepting a misconfigured collection).
- src/cli/qdrant_bootstrap.py: the operator entry point
  (python -m src.cli.qdrant_bootstrap).
- QdrantSettings gains collection/upsert_batch_size/upsert_concurrency.

Impact:
- Deployments must run the new bootstrap command before the first upload;
  see ADR-0001's new "Collection provisioning" section.

src/cli/__init__.py

@@ -0,0 +1 @@
+"""Operator entry points that run as deployment steps, not at app startup."""

src/cli/qdrant_bootstrap.py

@@ -0,0 +1,54 @@
+"""Create the `chunks` collection — the Qdrant analogue of `alembic upgrade head`.
+
+    uv run python -m src.cli.qdrant_bootstrap
+
+A deployment step, deliberately not part of the FastAPI lifespan: collection
+creation is DDL, which ADR-0009 keeps out of application startup for Postgres
+and ADR-0012 keeps out of it for LangGraph's `.setup()`. See
+`src/infrastructure/qdrant/collection.py` for the full reasoning.
+
+Idempotent and safe to re-run. Exits non-zero if an existing collection
+diverges from the pinned schema, rather than leaving a silently degraded
+sparse index behind.
+"""
+
+import asyncio
+import sys
+
+import structlog
+
+from src.config import Settings
+from src.infrastructure.observability.logging import configure_logging
+from src.infrastructure.qdrant.client import create_client
+from src.infrastructure.qdrant.collection import (
+    CollectionSchemaMismatchError,
+    ensure_chunks_collection,
+)
+
+logger = structlog.get_logger(__name__)
+
+
+async def bootstrap(settings: Settings | None = None) -> int:
+    resolved = settings or Settings()
+    configure_logging(resolved.logging)
+    client = create_client(resolved.qdrant)
+    try:
+        created = await ensure_chunks_collection(client, collection=resolved.qdrant.collection)
+    except CollectionSchemaMismatchError as exc:
+        logger.error("qdrant.bootstrap.schema_mismatch", error=str(exc))
+        return 1
+    finally:
+        await client.close()
+
+    logger.info(
+        "qdrant.bootstrap.completed", collection=resolved.qdrant.collection, created=created
+    )
+    return 0
+
+
+def main() -> None:
+    sys.exit(asyncio.run(bootstrap()))
+
+
+if __name__ == "__main__":
+    main()

src/config.py

@@ -109,12 +109,28 @@ class ChunkingSettings(BaseSettings):
 
 
 class QdrantSettings(BaseSettings):
+    """Qdrant connection and bulk-upsert bounds (ADR-0001, ADR-0017).
+
+    `collection` names the single shared collection all tenants live in
+    (ADR-0001); it is deliberately configurable so tests can point at a
+    disposable one. The vector *dimensions* are not settings -- they are model
+    facts pinned in `src/infrastructure/qdrant/collection.py`, and changing one
+    is a re-embedding migration.
+
+    `upsert_batch_size` sits inside ADR-0001's 64-256 bulk-upload band, and
+    `upsert_concurrency` bounds in-flight batches so ingestion issues parallel
+    streams rather than an unbounded `gather` (ADR-0017).
+    """
+
     model_config = SettingsConfigDict(
         env_prefix="QDRANT_", extra="ignore", env_file=".env", env_ignore_empty=True
     )
 
     url: str = "http://127.0.0.1:6343"
     api_key: str | None = None
+    collection: str = "chunks"
+    upsert_batch_size: int = 128
+    upsert_concurrency: int = 4
 
 
 class NomicEmbeddingSettings(BaseSettings):

src/infrastructure/qdrant/collection.py

@@ -0,0 +1,182 @@
+"""The `chunks` collection schema and its provisioning (ADR-0001).
+
+**This runs as a deployment step, never at FastAPI startup.** Creating a
+collection is DDL, and the project already rules DDL out of the request/boot
+path: ADR-0009 requires Alembic for Postgres schema, and ADR-0012 makes
+LangGraph's `.setup()` a deployment step. Doing it in the lifespan would also
+couple boot to Qdrant being reachable (that is `/readyz`'s job), race across
+replicas, and hide a misconfigured collection until traffic arrives.
+
+Entry point for operators: `uv run python -m src.cli.qdrant_bootstrap`.
+
+Two properties of this schema are load-bearing and fail *silently* if wrong,
+which is why `ensure_chunks_collection` verifies rather than skips:
+
+- **`sparse` must carry `modifier=IDF`.** `src/infrastructure/embedding/bm25.py`
+  computes only BM25's term-frequency saturation; Qdrant supplies IDF from
+  collection-wide statistics. Without the modifier there is no error and no
+  warning — lexical retrieval just quietly loses its IDF term (ADR-0005).
+- **The dense dimensions are pinned**: `dense_nomic` 768,
+  `dense_openai` 3072 (native, `dimensions` deliberately unset). They are
+  constants here rather than settings because they are model facts; changing
+  one is a re-embedding migration, not a config tweak (ADR-0001).
+
+All four named vectors are defined at creation even though `late_interaction`
+stays unpopulated until ADR-0003's rerank work (ADR-0017 does not compute it at
+ingest). Sparse and multivector fields cannot be added to an existing
+collection without recreating it, so deferring them is the one thing this
+schema cannot afford.
+"""
+
+import structlog
+from qdrant_client import AsyncQdrantClient, models
+
+logger = structlog.get_logger(__name__)
+
+DENSE_NOMIC_VECTOR = "dense_nomic"
+DENSE_OPENAI_VECTOR = "dense_openai"
+SPARSE_VECTOR = "sparse"
+LATE_INTERACTION_VECTOR = "late_interaction"
+
+DENSE_NOMIC_DIMENSIONS = 768
+DENSE_OPENAI_DIMENSIONS = 3072
+# jina-colbert-v2's per-token output dimension (ADR-0005).
+LATE_INTERACTION_DIMENSIONS = 128
+
+
+class CollectionSchemaMismatchError(RuntimeError):
+    """An existing collection does not match the schema this code expects.
+
+    Raised loudly instead of returning: silently accepting a collection whose
+    dense size or sparse modifier differs is precisely the failure this
+    explicit bootstrap step exists to prevent.
+    """
+
+
+def _vectors_config() -> dict[str, models.VectorParams]:
+    return {
+        DENSE_NOMIC_VECTOR: models.VectorParams(
+            size=DENSE_NOMIC_DIMENSIONS, distance=models.Distance.COSINE
+        ),
+        DENSE_OPENAI_VECTOR: models.VectorParams(
+            size=DENSE_OPENAI_DIMENSIONS, distance=models.Distance.COSINE
+        ),
+        # Rerank-only: never independently ANN-searched, so its HNSW graph is
+        # disabled (m=0), and stored on disk so its larger footprint does not
+        # degrade dense/sparse query latency (ADR-0001).
+        LATE_INTERACTION_VECTOR: models.VectorParams(
+            size=LATE_INTERACTION_DIMENSIONS,
+            distance=models.Distance.COSINE,
+            multivector_config=models.MultiVectorConfig(
+                comparator=models.MultiVectorComparator.MAX_SIM
+            ),
+            hnsw_config=models.HnswConfigDiff(m=0),
+            on_disk=True,
+        ),
+    }
+
+
+def _sparse_vectors_config() -> dict[str, models.SparseVectorParams]:
+    return {SPARSE_VECTOR: models.SparseVectorParams(modifier=models.Modifier.IDF)}
+
+
+# (field name, schema). `order_id` is float because Qdrant's `Range` conditions
+# and `order_by` only support numeric/datetime payloads -- a keyword key could
+# only be sorted client-side after fetching every chunk (ADR-0001).
+#
+# The full-text index on `content` is deliberately absent: it belongs to plan
+# 002's keyword search, and payload indexes -- unlike vector config -- can be
+# added to a live collection later.
+_PAYLOAD_INDEXES: tuple[tuple[str, models.PayloadSchemaType | models.KeywordIndexParams], ...] = (
+    # `is_tenant` co-locates a tenant's vectors on disk for sequential reads,
+    # which is the whole point of payload-partitioned multitenancy.
+    (
+        "tenant_id",
+        models.KeywordIndexParams(type=models.KeywordIndexType.KEYWORD, is_tenant=True),
+    ),
+    ("domain", models.PayloadSchemaType.KEYWORD),
+    ("file_id", models.PayloadSchemaType.KEYWORD),
+    ("order_id", models.PayloadSchemaType.FLOAT),
+    ("previous_chunk_id", models.PayloadSchemaType.KEYWORD),
+    ("next_chunk_id", models.PayloadSchemaType.KEYWORD),
+)
+
+
+def _verify_existing(collection: str, info: models.CollectionInfo) -> None:
+    params = info.config.params
+    vectors = params.vectors
+    if not isinstance(vectors, dict):
+        raise CollectionSchemaMismatchError(
+            f"collection {collection!r} has an unnamed dense vector; ADR-0001 requires "
+            f"named vectors and this cannot be fixed without recreating the collection"
+        )
+
+    for name, expected_size in (
+        (DENSE_NOMIC_VECTOR, DENSE_NOMIC_DIMENSIONS),
+        (DENSE_OPENAI_VECTOR, DENSE_OPENAI_DIMENSIONS),
+        (LATE_INTERACTION_VECTOR, LATE_INTERACTION_DIMENSIONS),
+    ):
+        existing = vectors.get(name)
+        if existing is None:
+            raise CollectionSchemaMismatchError(
+                f"collection {collection!r} is missing the {name!r} vector"
+            )
+        if existing.size != expected_size:
+            raise CollectionSchemaMismatchError(
+                f"collection {collection!r} has {name!r} at {existing.size} dimensions, "
+                f"expected {expected_size}; re-dimensioning is a re-embedding migration"
+            )
+
+    sparse = (params.sparse_vectors or {}).get(SPARSE_VECTOR)
+    if sparse is None:
+        raise CollectionSchemaMismatchError(
+            f"collection {collection!r} is missing the {SPARSE_VECTOR!r} vector; sparse "
+            f"vectors cannot be added without recreating the collection"
+        )
+    if sparse.modifier != models.Modifier.IDF:
+        raise CollectionSchemaMismatchError(
+            f"collection {collection!r} has {SPARSE_VECTOR!r} with modifier "
+            f"{sparse.modifier!r}, expected 'idf'; without it Qdrant applies no IDF "
+            f"and lexical retrieval silently degrades (ADR-0005)"
+        )
+
+
+async def ensure_chunks_collection(client: AsyncQdrantClient, *, collection: str) -> bool:
+    """Create the `chunks` collection and its payload indexes if absent.
+
+    Idempotent: an existing collection is verified against the pinned schema
+    and left alone. Returns whether it created the collection.
+
+    Raises `CollectionSchemaMismatchError` if an existing collection diverges.
+    """
+    if await client.collection_exists(collection):
+        _verify_existing(collection, await client.get_collection(collection))
+        logger.info("qdrant.collection.verified", collection=collection)
+        # Payload indexes are additive and idempotent, so (re)creating them
+        # here is what lets an index be added to an already-live collection.
+        await _create_payload_indexes(client, collection=collection)
+        return False
+
+    await client.create_collection(
+        collection_name=collection,
+        vectors_config=_vectors_config(),
+        sparse_vectors_config=_sparse_vectors_config(),
+        # m=0 disables the global index; payload_m builds per-tenant graphs
+        # instead, per Qdrant's multitenant guidance (ADR-0001).
+        hnsw_config=models.HnswConfigDiff(m=0, payload_m=16),
+    )
+    logger.info("qdrant.collection.created", collection=collection)
+    await _create_payload_indexes(client, collection=collection)
+    return True
+
+
+async def _create_payload_indexes(client: AsyncQdrantClient, *, collection: str) -> None:
+    for field_name, field_schema in _PAYLOAD_INDEXES:
+        await client.create_payload_index(
+            collection_name=collection, field_name=field_name, field_schema=field_schema
+        )
+    logger.info(
+        "qdrant.collection.payload_indexes.ensured",
+        collection=collection,
+        fields=[name for name, _ in _PAYLOAD_INDEXES],
+    )