diff --git a/src/application/points/__init__.py b/src/application/points/__init__.py new file mode 100644 index 0000000..5cb1ca8 --- /dev/null +++ b/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"] diff --git a/src/application/points/indexing.py b/src/application/points/indexing.py new file mode 100644 index 0000000..4d58756 --- /dev/null +++ b/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) diff --git a/src/application/points/models.py b/src/application/points/models.py new file mode 100644 index 0000000..6391785 --- /dev/null +++ b/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) diff --git a/src/application/points/payload.py b/src/application/points/payload.py new file mode 100644 index 0000000..11efec9 --- /dev/null +++ b/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, + } diff --git a/src/application/ports/point_storage.py b/src/application/ports/point_storage.py new file mode 100644 index 0000000..bd4bcd2 --- /dev/null +++ b/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. + """ + ... diff --git a/src/infrastructure/qdrant/points.py b/src/infrastructure/qdrant/points.py new file mode 100644 index 0000000..3a68f86 --- /dev/null +++ b/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 diff --git a/tests/fakes.py b/tests/fakes.py index d67f183..3a80dfb 100644 --- a/tests/fakes.py +++ b/tests/fakes.py @@ -1,10 +1,13 @@ """Hand-written fakes for narrow application-owned ports (ADR-0016).""" import asyncio +import uuid from collections.abc import Sequence from dataclasses import dataclass, field +from datetime import datetime from src.application.ingestion.models import SparseVector +from src.application.points.models import ChunkPoint @dataclass @@ -61,3 +64,75 @@ class FakeSparseEmbedder: raise RuntimeError("simulated embedder failure") return [SparseVector(indices=[], values=[]) for _ in texts] + +@dataclass +class FakePointStorage: + """An in-memory `PointStorage`. + + `points` is keyed by point id, so a re-upsert of the same deterministic id + overwrites rather than accumulating — the property a test asserting "a + retry produces no duplicate points" needs the fake to actually model. + + `fail_on_batch` fails the Nth (0-based) upsert batch, which is how a test + checks that the soft-delete sweep never runs after a partial failure. + """ + + points: dict[str, ChunkPoint] = field(default_factory=dict) + upsert_batches: list[int] = field(default_factory=list) + deactivate_calls: list[dict[str, object]] = field(default_factory=list) + fail_on_batch: int | None = None + fail_deactivate: bool = False + max_in_flight: int = 0 + _in_flight: int = 0 + + async def upsert_points(self, points: Sequence[ChunkPoint]) -> None: + self._in_flight += 1 + self.max_in_flight = max(self.max_in_flight, self._in_flight) + try: + # Yield so concurrent batches actually overlap; without this the + # in-flight ceiling is trivially 1 and the bound goes untested. + await asyncio.sleep(0) + index = len(self.upsert_batches) + self.upsert_batches.append(len(points)) + if self.fail_on_batch is not None and index == self.fail_on_batch: + raise RuntimeError("simulated point storage failure") + for point in points: + self.points[str(point.point_id)] = point + finally: + self._in_flight -= 1 + + 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: + self.deactivate_calls.append( + { + "tenant_id": tenant_id, + "file_id": file_id, + "from_chunk_index": from_chunk_index, + "updated_by": updated_by, + } + ) + if self.fail_deactivate: + raise RuntimeError("simulated deactivate failure") + + def is_stale(point: ChunkPoint) -> bool: + chunk_index = point.payload.get("chunk_index") + return ( + point.payload.get("file_id") == str(file_id) + and point.payload.get("tenant_id") == str(tenant_id) + and point.payload.get("is_active") is True + and isinstance(chunk_index, int) + and chunk_index >= from_chunk_index + ) + + stale = [point for point in self.points.values() if is_stale(point)] + for point in stale: + point.payload["is_active"] = False + point.payload["deleted_at"] = deleted_at.isoformat() + return len(stale) diff --git a/tests/integration/qdrant/test_points.py b/tests/integration/qdrant/test_points.py new file mode 100644 index 0000000..a7a51da --- /dev/null +++ b/tests/integration/qdrant/test_points.py @@ -0,0 +1,182 @@ +"""`QdrantPointStorage` against a real Qdrant (ADR-0001, ADR-0002). + +Reads here go through the raw client rather than the port: `PointStorage` is +deliberately write-only, because point reads are plan 002's `/v1/points` +surface. The reads below are the test's own verification, not a preview of an +API this slice ships. +""" + +import uuid +from datetime import UTC, datetime + +import pytest +from qdrant_client import AsyncQdrantClient, models + +from src.application.ingestion.chunking import chunk_id_for +from src.application.ingestion.models import SparseVector +from src.application.points.models import ChunkPoint +from src.config import QdrantSettings +from src.infrastructure.qdrant.collection import ( + DENSE_NOMIC_DIMENSIONS, + DENSE_OPENAI_DIMENSIONS, + ensure_chunks_collection, +) +from src.infrastructure.qdrant.points import QdrantPointStorage + +pytestmark = [ + pytest.mark.integration, + pytest.mark.qdrant, + pytest.mark.asyncio(loop_scope="session"), +] + + +def _point(tenant_id: uuid.UUID, file_id: uuid.UUID, chunk_index: int) -> ChunkPoint: + return ChunkPoint( + point_id=chunk_id_for(file_id, chunk_index), + dense={ + "dense_nomic": [0.1] * DENSE_NOMIC_DIMENSIONS, + "dense_openai": [0.2] * DENSE_OPENAI_DIMENSIONS, + }, + sparse=SparseVector(indices=[1, 2], values=[0.5, 0.25]), + payload={ + "tenant_id": str(tenant_id), + "domain": "fire", + "file_id": str(file_id), + "chunk_id": str(chunk_id_for(file_id, chunk_index)), + "chunk_index": chunk_index, + "order_id": float(chunk_index + 1), + "content": f"chunk {chunk_index}", + "is_active": True, + }, + ) + + +async def _storage( + qdrant_client: AsyncQdrantClient, qdrant_settings: QdrantSettings +) -> QdrantPointStorage: + await ensure_chunks_collection(qdrant_client, collection=qdrant_settings.collection) + return QdrantPointStorage(qdrant_client, collection=qdrant_settings.collection) + + +async def _count_for_tenant( + client: AsyncQdrantClient, collection: str, tenant_id: uuid.UUID +) -> int: + result = await client.count( + collection_name=collection, + count_filter=models.Filter( + must=[ + models.FieldCondition( + key="tenant_id", match=models.MatchValue(value=str(tenant_id)) + ) + ] + ), + exact=True, + ) + return result.count + + +async def test_upsert_points_stores_points_readable_under_the_owning_tenant_filter( + qdrant_client: AsyncQdrantClient, qdrant_settings: QdrantSettings +) -> None: + storage = await _storage(qdrant_client, qdrant_settings) + tenant_id, file_id = uuid.uuid4(), uuid.uuid4() + + await storage.upsert_points([_point(tenant_id, file_id, i) for i in range(3)]) + + assert await _count_for_tenant(qdrant_client, qdrant_settings.collection, tenant_id) == 3 + + +async def test_upsert_points_are_invisible_to_another_tenants_filter( + qdrant_client: AsyncQdrantClient, qdrant_settings: QdrantSettings +) -> None: + """The Qdrant-level form of "cross-tenant access finds nothing" (ADR-0002).""" + storage = await _storage(qdrant_client, qdrant_settings) + owner, other, file_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4() + + await storage.upsert_points([_point(owner, file_id, i) for i in range(3)]) + + assert await _count_for_tenant(qdrant_client, qdrant_settings.collection, other) == 0 + + +async def test_upsert_points_is_idempotent_for_deterministic_ids( + qdrant_client: AsyncQdrantClient, qdrant_settings: QdrantSettings +) -> None: + storage = await _storage(qdrant_client, qdrant_settings) + tenant_id, file_id = uuid.uuid4(), uuid.uuid4() + points = [_point(tenant_id, file_id, i) for i in range(4)] + + await storage.upsert_points(points) + await storage.upsert_points(points) + + assert await _count_for_tenant(qdrant_client, qdrant_settings.collection, tenant_id) == 4 + + +async def test_deactivate_points_from_index_soft_deletes_only_the_tail( + qdrant_client: AsyncQdrantClient, qdrant_settings: QdrantSettings +) -> None: + storage = await _storage(qdrant_client, qdrant_settings) + tenant_id, file_id = uuid.uuid4(), uuid.uuid4() + await storage.upsert_points([_point(tenant_id, file_id, i) for i in range(5)]) + + deactivated = await storage.deactivate_points_from_index( + tenant_id=tenant_id, + file_id=file_id, + from_chunk_index=2, + deleted_at=datetime.now(UTC), + updated_by="api_key:test", + ) + + assert deactivated == 3 + records, _ = await qdrant_client.scroll( + collection_name=qdrant_settings.collection, + scroll_filter=models.Filter( + must=[models.FieldCondition(key="file_id", match=models.MatchValue(value=str(file_id)))] + ), + limit=10, + with_payload=True, + ) + by_index = { + record.payload["chunk_index"]: record.payload["is_active"] + for record in records + if record.payload is not None + } + assert by_index == {0: True, 1: True, 2: False, 3: False, 4: False} + # Soft delete, not removal -- the points stay for audit (ADR-0002). + assert await _count_for_tenant(qdrant_client, qdrant_settings.collection, tenant_id) == 5 + + +async def test_deactivate_points_from_index_does_not_touch_another_tenants_points( + qdrant_client: AsyncQdrantClient, qdrant_settings: QdrantSettings +) -> None: + """A file_id alone is never authority to mutate (ADR-0002).""" + storage = await _storage(qdrant_client, qdrant_settings) + owner, other, file_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4() + await storage.upsert_points([_point(owner, file_id, i) for i in range(3)]) + + deactivated = await storage.deactivate_points_from_index( + tenant_id=other, + file_id=file_id, + from_chunk_index=0, + deleted_at=datetime.now(UTC), + updated_by="api_key:intruder", + ) + + assert deactivated == 0 + + +async def test_deactivate_points_from_index_returns_zero_when_nothing_is_stale( + qdrant_client: AsyncQdrantClient, qdrant_settings: QdrantSettings +) -> None: + storage = await _storage(qdrant_client, qdrant_settings) + tenant_id, file_id = uuid.uuid4(), uuid.uuid4() + await storage.upsert_points([_point(tenant_id, file_id, i) for i in range(3)]) + + deactivated = await storage.deactivate_points_from_index( + tenant_id=tenant_id, + file_id=file_id, + from_chunk_index=3, + deleted_at=datetime.now(UTC), + updated_by="api_key:test", + ) + + assert deactivated == 0 diff --git a/tests/unit/application/points/__init__.py b/tests/unit/application/points/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/application/points/test_indexing.py b/tests/unit/application/points/test_indexing.py new file mode 100644 index 0000000..a4f4902 --- /dev/null +++ b/tests/unit/application/points/test_indexing.py @@ -0,0 +1,227 @@ +"""`index_chunks`: payload correctness, bounded batching, and the ordering +that keeps a failed attempt from damaging a working index (ADR-0001, ADR-0017). +""" + +import uuid + +import pytest +from anyio import CapacityLimiter + +from src.application.ingestion.chunking import chunk_id_for +from src.application.ingestion.errors import PointIndexingError +from src.application.ingestion.models import Chunk, ContentType, EmbeddedChunk, SparseVector +from src.application.points import index_chunks +from src.config import QdrantSettings +from tests.fakes import FakeDenseEmbedder, FakePointStorage, FakeSparseEmbedder + +pytestmark = [pytest.mark.unit, pytest.mark.asyncio] + +_TENANT_ID = uuid.UUID("11111111-1111-1111-1111-111111111111") +_FILE_ID = uuid.UUID("22222222-2222-2222-2222-222222222222") +_API_KEY_ID = uuid.UUID("33333333-3333-3333-3333-333333333333") +_ACTOR = f"api_key:{_API_KEY_ID}" + + +def _embedded(count: int) -> list[EmbeddedChunk]: + chunks = [ + Chunk( + chunk_id=chunk_id_for(_FILE_ID, index), + chunk_index=index, + order_id=float(index + 1), + content=f"chunk {index}", + content_type=ContentType.PARAGRAPH, + token_count=2, + character_count=7, + ) + for index in range(count) + ] + for position, chunk in enumerate(chunks): + if position > 0: + chunk.previous_chunk_id = chunks[position - 1].chunk_id + if position < len(chunks) - 1: + chunk.next_chunk_id = chunks[position + 1].chunk_id + return [ + EmbeddedChunk( + chunk=chunk, + dense={"dense_nomic": [0.0] * 4, "dense_openai": [1.0] * 4}, + sparse=SparseVector(indices=[7], values=[0.5]), + ) + for chunk in chunks + ] + + +async def _index( + storage: FakePointStorage, + *, + count: int = 3, + settings: QdrantSettings | None = None, + domain: str = "fire", +): + return await index_chunks( + _embedded(count), + storage=storage, + tenant_id=_TENANT_ID, + domain=domain, + file_id=_FILE_ID, + source_filename="policy.docx", + source_type="docx", + actor=_ACTOR, + dense_embedders=[ + FakeDenseEmbedder(name="dense_nomic", model_version="nomic-embed-text-v2-moe"), + FakeDenseEmbedder(name="dense_openai", model_version="text-embedding-3-large"), + ], + sparse_embedder=FakeSparseEmbedder(model_version="bm25-fa_norm_stop"), + settings=settings or QdrantSettings(), + thread_limiter=CapacityLimiter(2), + ) + + +async def test_index_chunks_writes_every_adr_0001_payload_field() -> None: + storage = FakePointStorage() + + await _index(storage, count=3) + + payload = storage.points[str(chunk_id_for(_FILE_ID, 1))].payload + assert payload["tenant_id"] == str(_TENANT_ID) + assert payload["domain"] == "fire" + assert payload["file_id"] == str(_FILE_ID) + assert payload["chunk_id"] == str(chunk_id_for(_FILE_ID, 1)) + assert payload["content"] == "chunk 1" + assert payload["content_type"] == "paragraph" + assert payload["source_filename"] == "policy.docx" + assert payload["source_type"] == "docx" + assert payload["order_id"] == 2.0 + assert payload["chunk_index"] == 1 + assert payload["previous_chunk_id"] == str(chunk_id_for(_FILE_ID, 0)) + assert payload["next_chunk_id"] == str(chunk_id_for(_FILE_ID, 2)) + assert payload["is_active"] is True + assert payload["deleted_at"] is None + assert payload["created_by"] == _ACTOR + assert payload["updated_by"] == _ACTOR + assert payload["version"] == 1 + assert payload["created_at"] == payload["updated_at"] + assert isinstance(payload["content_hash"], str) + # Sorted, so wiring order cannot change the value (ADR-0001). + assert payload["embedding_model_version"] == ( + "bm25-fa_norm_stop+nomic-embed-text-v2-moe+text-embedding-3-large" + ) + + +async def test_index_chunks_uses_null_neighbours_at_the_file_ends() -> None: + storage = FakePointStorage() + + await _index(storage, count=3) + + assert storage.points[str(chunk_id_for(_FILE_ID, 0))].payload["previous_chunk_id"] is None + assert storage.points[str(chunk_id_for(_FILE_ID, 2))].payload["next_chunk_id"] is None + + +async def test_index_chunks_derives_tenant_and_domain_from_the_caller_not_the_chunk() -> None: + """Tenant identity is server-derived; nothing in the chunk can assert it.""" + storage = FakePointStorage() + + await _index(storage, count=1, domain="car") + + payload = storage.points[str(chunk_id_for(_FILE_ID, 0))].payload + assert payload["tenant_id"] == str(_TENANT_ID) + assert payload["domain"] == "car" + + +async def test_index_chunks_uses_deterministic_point_ids() -> None: + storage = FakePointStorage() + + result = await _index(storage, count=4) + + assert result.points_upserted == 4 + assert set(storage.points) == {str(chunk_id_for(_FILE_ID, i)) for i in range(4)} + + +async def test_index_chunks_repeated_run_produces_no_duplicate_points() -> None: + storage = FakePointStorage() + + await _index(storage, count=4) + await _index(storage, count=4) + + assert len(storage.points) == 4 + + +async def test_index_chunks_batches_at_the_configured_size() -> None: + storage = FakePointStorage() + settings = QdrantSettings(upsert_batch_size=2, upsert_concurrency=4) + + await _index(storage, count=5, settings=settings) + + assert storage.upsert_batches == [2, 2, 1] + + +async def test_index_chunks_bounds_in_flight_batches() -> None: + storage = FakePointStorage() + settings = QdrantSettings(upsert_batch_size=1, upsert_concurrency=2) + + await _index(storage, count=8, settings=settings) + + assert len(storage.upsert_batches) == 8 + assert storage.max_in_flight <= 2 + + +async def test_index_chunks_soft_deletes_only_points_past_the_new_chunk_count() -> None: + storage = FakePointStorage() + await _index(storage, count=5) + + result = await _index(storage, count=2) + + assert result.points_soft_deleted == 3 + assert storage.points[str(chunk_id_for(_FILE_ID, 1))].payload["is_active"] is True + assert storage.points[str(chunk_id_for(_FILE_ID, 2))].payload["is_active"] is False + assert storage.points[str(chunk_id_for(_FILE_ID, 4))].payload["is_active"] is False + + +async def test_index_chunks_does_not_soft_delete_when_an_upsert_batch_fails() -> None: + """A failed attempt must never remove content from a working index.""" + storage = FakePointStorage() + await _index(storage, count=5) + storage.deactivate_calls.clear() + storage.fail_on_batch = 1 + + with pytest.raises(PointIndexingError): + await _index(storage, count=2, settings=QdrantSettings(upsert_batch_size=1)) + + assert storage.deactivate_calls == [] + assert all(point.payload["is_active"] is True for point in storage.points.values()) + + +async def test_index_chunks_raises_point_indexing_error_when_a_batch_fails() -> None: + storage = FakePointStorage(fail_on_batch=0) + + with pytest.raises(PointIndexingError, match="upserting"): + await _index(storage, count=2) + + +async def test_index_chunks_raises_point_indexing_error_when_the_sweep_fails() -> None: + storage = FakePointStorage(fail_deactivate=True) + + with pytest.raises(PointIndexingError, match="soft-deleting"): + await _index(storage, count=2) + + +async def test_index_chunks_on_empty_input_touches_no_storage() -> None: + storage = FakePointStorage() + + result = await index_chunks( + [], + storage=storage, + tenant_id=_TENANT_ID, + domain="fire", + file_id=_FILE_ID, + source_filename="empty.csv", + source_type="csv", + actor=_ACTOR, + dense_embedders=[FakeDenseEmbedder(name="dense_nomic")], + sparse_embedder=FakeSparseEmbedder(), + settings=QdrantSettings(), + thread_limiter=CapacityLimiter(2), + ) + + assert result.points_upserted == 0 + assert storage.upsert_batches == [] + assert storage.deactivate_calls == []