feat(qdrant): add tenant-scoped point storage for ingestion

Why:
- Ingested chunks need to become searchable Qdrant points before the upload
  response returns, with tenant/domain isolation and a safe re-ingestion
  story per ADR-0001/0017.

Changes:
- src/application/points/: index_chunks() is the sole entry point, owning
  payload construction, batched/bounded-concurrency upserts
  (upsert_concurrency semaphore), and a soft-delete sweep for points a
  shorter re-ingestion leaves behind. The sweep runs only after every upsert
  in the attempt succeeds, so a failed attempt can leave a stale prefix but
  never removes content from a working index.
- PointStorage port (application/ports/) + QdrantPointStorage adapter
  (infrastructure/qdrant/points.py), keeping the qdrant_client SDK out of
  application code per ADR-0015.
- FakePointStorage test double for exercising the ordering/idempotency
  guarantees without a real Qdrant.

src/application/points/__init__.py

@@ -0,0 +1,16 @@
+"""Ingestion-generated Qdrant point CRUD (ADR-0001, ADR-0002).
+
+`index_chunks` is the entry point callers outside this package should use: it
+dispatches payload construction, batching, bounded-concurrency upserts, and the
+post-success soft-delete sweep. `build_chunk_payload` and the batching helpers
+stay internal, exported mainly for their own unit tests.
+
+Direct `/v1/points` CRUD (single-point edits, reordering, keyword search) is
+plan 002's surface, not this package's — plan 001 scopes it to "the reusable
+service layer required by ingestion".
+"""
+
+from src.application.points.indexing import IndexingResult, index_chunks
+from src.application.points.models import ChunkPoint
+
+__all__ = ["ChunkPoint", "IndexingResult", "index_chunks"]

src/application/points/indexing.py

@@ -0,0 +1,203 @@
+"""The one caller-facing entry point for indexing embedded chunks (ADR-0001, ADR-0017).
+
+`index_chunks` is the only version of this step callers should reach for. It
+owns the whole composition a correct upsert needs:
+
+- building ADR-0001's payload for every chunk, with `tenant_id`/`domain` taken
+  from server-derived context;
+- offloading that (and the per-chunk content hashing) to a thread, since it is
+  blocking CPU work (ADR-0017);
+- batching at `QDRANT_UPSERT_BATCH_SIZE` inside ADR-0001's 64-256 band;
+- bounding in-flight batches with an `asyncio.Semaphore` rather than an
+  unbounded `gather` (ADR-0017);
+- running the soft-delete sweep for a shortened file **only after every batch
+  has succeeded**.
+
+That last ordering is the point, not an implementation detail — see
+`_deactivate_stale` below. `build_chunk_payload` and `_batches` stay internal;
+pushing that composition onto every call site is exactly the obligation a deep
+module absorbs once (CLAUDE.md, "prefer deep modules").
+"""
+
+import asyncio
+import uuid
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import UTC, datetime
+
+from anyio import CapacityLimiter, to_thread
+
+from src.application.ingestion.errors import PointIndexingError
+from src.application.ingestion.models import EmbeddedChunk
+from src.application.points.models import ChunkPoint
+from src.application.points.payload import build_chunk_payload
+from src.application.ports.embedding import DenseEmbedder, SparseEmbedder
+from src.application.ports.point_storage import PointStorage
+from src.config import QdrantSettings
+
+
+@dataclass(frozen=True)
+class IndexingResult:
+    """What one indexing pass wrote.
+
+    `points_upserted` counts points written, not points *created* — a
+    deterministic-id upsert cannot distinguish an insert from an overwrite, so
+    the ingestion job reports this as `points_created` and leaves
+    `points_updated` at zero rather than guessing.
+    """
+
+    points_upserted: int
+    points_soft_deleted: int
+
+
+def _embedding_model_version(
+    dense_embedders: Sequence[DenseEmbedder], sparse_embedder: SparseEmbedder
+) -> str:
+    """Compose the `embedding_model_version` payload value (ADR-0001).
+
+    Sorted so the string is stable regardless of the order the embedders were
+    wired in — an unstable value would make "which chunks need re-embedding?"
+    unanswerable, which is the field's only reason to exist.
+    """
+    versions = sorted(
+        [embedder.model_version for embedder in dense_embedders] + [sparse_embedder.model_version]
+    )
+    return "+".join(versions)
+
+
+def _batches(points: Sequence[ChunkPoint], size: int) -> list[Sequence[ChunkPoint]]:
+    return [points[i : i + size] for i in range(0, len(points), size)]
+
+
+def _build_points(
+    embedded: Sequence[EmbeddedChunk],
+    *,
+    tenant_id: uuid.UUID,
+    domain: str,
+    file_id: uuid.UUID,
+    source_filename: str,
+    source_type: str,
+    actor: str,
+    embedding_model_version: str,
+    indexed_at: datetime,
+) -> list[ChunkPoint]:
+    """Blocking: hashes every chunk's content. Always called through a thread."""
+    return [
+        ChunkPoint(
+            point_id=item.chunk.chunk_id,
+            dense=item.dense,
+            sparse=item.sparse,
+            payload=build_chunk_payload(
+                item.chunk,
+                tenant_id=tenant_id,
+                domain=domain,
+                file_id=file_id,
+                source_filename=source_filename,
+                source_type=source_type,
+                actor=actor,
+                embedding_model_version=embedding_model_version,
+                indexed_at=indexed_at,
+            ),
+        )
+        for item in embedded
+    ]
+
+
+async def _upsert_bounded(
+    storage: PointStorage, batch: Sequence[ChunkPoint], *, semaphore: asyncio.Semaphore
+) -> None:
+    async with semaphore:
+        try:
+            await storage.upsert_points(batch)
+        except Exception as exc:
+            raise PointIndexingError(f"upserting {len(batch)} points failed: {exc}") from exc
+
+
+async def _deactivate_stale(
+    storage: PointStorage,
+    *,
+    tenant_id: uuid.UUID,
+    file_id: uuid.UUID,
+    from_chunk_index: int,
+    actor: str,
+    deleted_at: datetime,
+) -> int:
+    """Soft-delete points left over from a longer previous version of this file.
+
+    Chunk indices are contiguous from 0, so "index >= the new chunk count" is
+    exactly the set of points the new version no longer produces.
+
+    This runs **only after every upsert has succeeded**, and that ordering is
+    what keeps a failed attempt from damaging a working index. ADR-0001's
+    deterministic point ids mean a re-ingestion overwrites in place, so literal
+    atomic replacement is not available; what *is* guaranteed is that a failed
+    attempt never removes content (it can only leave a prefix updated), and that
+    a retry converges to the correct state. See ADR-0017.
+    """
+    try:
+        return await storage.deactivate_points_from_index(
+            tenant_id=tenant_id,
+            file_id=file_id,
+            from_chunk_index=from_chunk_index,
+            deleted_at=deleted_at,
+            updated_by=actor,
+        )
+    except Exception as exc:
+        raise PointIndexingError(f"soft-deleting stale points failed: {exc}") from exc
+
+
+async def index_chunks(
+    embedded: Sequence[EmbeddedChunk],
+    *,
+    storage: PointStorage,
+    tenant_id: uuid.UUID,
+    domain: str,
+    file_id: uuid.UUID,
+    source_filename: str,
+    source_type: str,
+    actor: str,
+    dense_embedders: Sequence[DenseEmbedder],
+    sparse_embedder: SparseEmbedder,
+    settings: QdrantSettings,
+    thread_limiter: CapacityLimiter,
+) -> IndexingResult:
+    """Upsert every embedded chunk as a tenant-scoped point, then sweep leftovers.
+
+    Raises `PointIndexingError` (502) if any batch or the sweep fails.
+    """
+    if not embedded:
+        return IndexingResult(points_upserted=0, points_soft_deleted=0)
+
+    indexed_at = datetime.now(UTC)
+    points = await to_thread.run_sync(
+        lambda: _build_points(
+            embedded,
+            tenant_id=tenant_id,
+            domain=domain,
+            file_id=file_id,
+            source_filename=source_filename,
+            source_type=source_type,
+            actor=actor,
+            embedding_model_version=_embedding_model_version(dense_embedders, sparse_embedder),
+            indexed_at=indexed_at,
+        ),
+        limiter=thread_limiter,
+    )
+
+    semaphore = asyncio.Semaphore(settings.upsert_concurrency)
+    await asyncio.gather(
+        *(
+            _upsert_bounded(storage, batch, semaphore=semaphore)
+            for batch in _batches(points, settings.upsert_batch_size)
+        )
+    )
+
+    soft_deleted = await _deactivate_stale(
+        storage,
+        tenant_id=tenant_id,
+        file_id=file_id,
+        from_chunk_index=len(points),
+        actor=actor,
+        deleted_at=indexed_at,
+    )
+    return IndexingResult(points_upserted=len(points), points_soft_deleted=soft_deleted)

src/application/points/models.py

@@ -0,0 +1,29 @@
+"""Domain models for Qdrant points (ADR-0001).
+
+Deliberately free of the `qdrant_client` SDK: `src/infrastructure/qdrant/`
+converts these to `PointStruct`/`models.SparseVector` at upsert time
+(ADR-0015 — application code and ports carry no infrastructure imports).
+"""
+
+import uuid
+
+from pydantic import BaseModel, Field
+
+from src.application.ingestion.models import SparseVector
+
+
+class ChunkPoint(BaseModel):
+    """One chunk, ready to upsert: its id, its named vectors, and its payload.
+
+    `point_id` is the chunk's deterministic UUIDv5 (`chunk_id_for`), so
+    re-ingesting a file overwrites its points rather than duplicating them
+    (ADR-0001).
+
+    `dense` is keyed by named-vector name (`dense_nomic`, `dense_openai`).
+    `late_interaction` is absent — ADR-0017 does not compute it at ingest.
+    """
+
+    point_id: uuid.UUID
+    dense: dict[str, list[float]]
+    sparse: SparseVector
+    payload: dict[str, object] = Field(default_factory=dict)

src/application/points/payload.py

@@ -0,0 +1,70 @@
+"""Builds ADR-0001's point payload from a chunk plus its ingestion context.
+
+Internal to `src/application/points/` — callers use `index_chunks`, which owns
+composing this with batching and the deactivation sweep. Exported for its own
+unit tests, not as a surface to build payloads by hand.
+"""
+
+import uuid
+from datetime import datetime
+from hashlib import sha256
+
+from src.application.ingestion.models import Chunk
+
+
+def _optional_id(value: uuid.UUID | None) -> str | None:
+    return str(value) if value is not None else None
+
+
+def build_chunk_payload(
+    chunk: Chunk,
+    *,
+    tenant_id: uuid.UUID,
+    domain: str,
+    file_id: uuid.UUID,
+    source_filename: str,
+    source_type: str,
+    actor: str,
+    embedding_model_version: str,
+    indexed_at: datetime,
+) -> dict[str, object]:
+    """Return ADR-0001's payload for one chunk.
+
+    `tenant_id` and `domain` are passed in from the server-derived `AuthContext`
+    and the validated request — never from anything the client could assert as
+    authority (ADR-0002's non-negotiable isolation rule).
+
+    UUIDs are serialized as strings because the `tenant_id`/`domain`/`file_id`/
+    `previous_chunk_id`/`next_chunk_id` payload indexes are *keyword* indexes;
+    a native UUID would not match a keyword filter.
+
+    **Known gap — `version` is always written as `1`.** ADR-0002 uses this field
+    for optimistic concurrency between ingestion and manual `/v1/points` edits,
+    which needs a read-check-write (one read per point). Ingestion is
+    authoritative for its own file today, so writing `1` is safe until
+    `/v1/points` exists; plan 002 owns closing this.
+    """
+    timestamp = indexed_at.isoformat()
+    return {
+        "tenant_id": str(tenant_id),
+        "domain": domain,
+        "file_id": str(file_id),
+        "chunk_id": str(chunk.chunk_id),
+        "content": chunk.content,
+        "content_type": chunk.content_type.value,
+        "source_filename": source_filename,
+        "source_type": source_type,
+        "order_id": chunk.order_id,
+        "chunk_index": chunk.chunk_index,
+        "previous_chunk_id": _optional_id(chunk.previous_chunk_id),
+        "next_chunk_id": _optional_id(chunk.next_chunk_id),
+        "is_active": True,
+        "deleted_at": None,
+        "created_at": timestamp,
+        "updated_at": timestamp,
+        "created_by": actor,
+        "updated_by": actor,
+        "version": 1,
+        "content_hash": sha256(chunk.content.encode("utf-8")).hexdigest(),
+        "embedding_model_version": embedding_model_version,
+    }

src/application/ports/point_storage.py

@@ -0,0 +1,45 @@
+"""The point-storage port (ADR-0001, ADR-0015).
+
+`src/infrastructure/qdrant/points.py` is the production adapter; tests use a
+hand-written fake (ADR-0016). Application code depends on this Protocol, not on
+the `qdrant_client` SDK.
+
+Deliberately narrow: exactly the two operations ingestion performs. Reads,
+single-point edits, reordering, and keyword search are plan 002's `/v1/points`
+surface and belong on a port of their own rather than accreting here.
+"""
+
+import uuid
+from collections.abc import Sequence
+from datetime import datetime
+from typing import Protocol
+
+from src.application.points.models import ChunkPoint
+
+
+class PointStorage(Protocol):
+    async def upsert_points(self, points: Sequence[ChunkPoint]) -> None:
+        """Upsert one batch of points.
+
+        Callers own batching and concurrency bounding (ADR-0017's
+        `upsert_concurrency` semaphore), not this Protocol — the same division
+        `DenseEmbedder.embed_batch` uses.
+        """
+        ...
+
+    async def deactivate_points_from_index(
+        self,
+        *,
+        tenant_id: uuid.UUID,
+        file_id: uuid.UUID,
+        from_chunk_index: int,
+        deleted_at: datetime,
+        updated_by: str,
+    ) -> int:
+        """Soft-delete this file's points at or past `from_chunk_index`.
+
+        Sets `is_active=false`/`deleted_at` rather than removing the points
+        (ADR-0002: delete is soft by default). Tenant-filtered — a `file_id`
+        alone is never sufficient authority. Returns how many points matched.
+        """
+        ...

src/infrastructure/qdrant/points.py

@@ -0,0 +1,97 @@
+"""Qdrant adapter for the `PointStorage` port (ADR-0001, ADR-0002, ADR-0015).
+
+The `qdrant_client` SDK appears here and nowhere in `application/`. This module
+translates the SDK-free `ChunkPoint` into `PointStruct`s and builds every
+filter — routers and application services never construct Qdrant filters.
+
+`AsyncQdrantClient` is genuinely async, so unlike the `minio` adapter nothing
+here needs a thread offload.
+"""
+
+import uuid
+from collections.abc import Sequence
+from datetime import datetime
+
+from qdrant_client import AsyncQdrantClient, models
+
+from src.application.points.models import ChunkPoint
+from src.infrastructure.qdrant.collection import SPARSE_VECTOR
+
+
+def _tenant_file_filter(
+    tenant_id: uuid.UUID, file_id: uuid.UUID, *, from_chunk_index: int
+) -> models.Filter:
+    """Points of one file, at or past `from_chunk_index`, within one tenant.
+
+    `tenant_id` is always a condition, never optional: a `file_id` alone is not
+    authority to mutate anything (ADR-0002's isolation rule applies to every
+    code path, not just reads).
+    """
+    return models.Filter(
+        must=[
+            models.FieldCondition(key="tenant_id", match=models.MatchValue(value=str(tenant_id))),
+            models.FieldCondition(key="file_id", match=models.MatchValue(value=str(file_id))),
+            models.FieldCondition(key="chunk_index", range=models.Range(gte=from_chunk_index)),
+        ]
+    )
+
+
+class QdrantPointStorage:
+    """A `PointStorage` (see `src/application/ports/point_storage.py`)."""
+
+    def __init__(self, client: AsyncQdrantClient, *, collection: str) -> None:
+        self._client = client
+        self._collection = collection
+
+    async def upsert_points(self, points: Sequence[ChunkPoint]) -> None:
+        if not points:
+            return
+        await self._client.upsert(
+            collection_name=self._collection,
+            points=[
+                models.PointStruct(
+                    id=str(point.point_id),
+                    vector={
+                        **point.dense,
+                        SPARSE_VECTOR: models.SparseVector(
+                            indices=point.sparse.indices, values=point.sparse.values
+                        ),
+                    },
+                    payload=point.payload,
+                )
+                for point in points
+            ],
+        )
+
+    async def deactivate_points_from_index(
+        self,
+        *,
+        tenant_id: uuid.UUID,
+        file_id: uuid.UUID,
+        from_chunk_index: int,
+        deleted_at: datetime,
+        updated_by: str,
+    ) -> int:
+        """Soft-delete via `set_payload` — the points stay for audit (ADR-0002).
+
+        Counts first so the caller can report how many points the sweep
+        touched; `set_payload` itself reports only an operation status.
+        """
+        point_filter = _tenant_file_filter(tenant_id, file_id, from_chunk_index=from_chunk_index)
+        stale = await self._client.count(
+            collection_name=self._collection, count_filter=point_filter, exact=True
+        )
+        if stale.count == 0:
+            return 0
+
+        await self._client.set_payload(
+            collection_name=self._collection,
+            payload={
+                "is_active": False,
+                "deleted_at": deleted_at.isoformat(),
+                "updated_at": deleted_at.isoformat(),
+                "updated_by": updated_by,
+            },
+            points=models.FilterSelector(filter=point_filter),
+        )
+        return stale.count