diff --git a/src/api/schemas/points.py b/src/api/schemas/points.py new file mode 100644 index 0000000..5537e80 --- /dev/null +++ b/src/api/schemas/points.py @@ -0,0 +1,145 @@ +"""Public request/response models for `/v1/points` (ADR-0002, ADR-0008). + +The shape callers see, kept separate from `application/points`' domain models +(ADR-0015). Two rules are encoded here rather than left to route code: + +- **Vectors are opt-in.** `PointResponse` omits them unless the caller asked, + so a listing does not ship megabytes of floats nobody reads (ADR-0008). +- **Server-owned fields are not accepted on input.** The request models simply + do not declare `tenant_id`, `version`, or `chunk_index`, and forbid extra + keys, so a client that sends one gets `422` from Pydantic instead of having + it silently ignored — ADR-0002's isolation rule enforced at the boundary. +""" + +import uuid +from datetime import datetime + +from pydantic import BaseModel, ConfigDict, Field + +from src.application.points.point import Point +from src.application.ports.point_repository import PointPage + + +class PointResponse(BaseModel): + point_id: uuid.UUID + domain: str + file_id: uuid.UUID + chunk_id: uuid.UUID + + content: str + content_type: str + source_filename: str + source_type: str + + order_id: float + chunk_index: int + previous_chunk_id: uuid.UUID | None + next_chunk_id: uuid.UUID | None + + is_active: bool + deleted_at: datetime | None + created_at: datetime + updated_at: datetime + created_by: str + updated_by: str + + version: int + content_hash: str + embedding_model_version: str + + vectors: dict[str, object] | None = None + + @classmethod + def from_point(cls, point: Point) -> "PointResponse": + # `tenant_id` is present on `Point` and deliberately absent here: the + # caller already knows which tenant it authenticated as, and echoing it + # back invites clients to start sending it. + return cls.model_validate(point.model_dump(exclude={"tenant_id"})) + + +class PointListResponse(BaseModel): + """A page of points plus the cursor for the next one. + + Cursor-based rather than `limit`/`offset`: an offset cursor silently skips + or repeats rows when a concurrent insert shifts positions, which is exactly + the pagination defect plan 002 requires a test for. + """ + + points: list[PointResponse] + next_cursor: str | None = None + + @classmethod + def from_page(cls, page: PointPage) -> "PointListResponse": + return cls( + points=[PointResponse.from_point(point) for point in page.points], + next_cursor=page.next_cursor, + ) + + +class PointCountResponse(BaseModel): + count: int + + +class PointSearchResponse(PointListResponse): + """Results of a **keyword** match, not of semantic retrieval. + + Named and documented so it cannot be mistaken for ADR-0003's hybrid + retrieval: these points matched a full-text filter on `content`, they are + not ranked by relevance, and there is no score to report. Anything that + wants ranked results wants the agent retrieval path in plan 003. + """ + + query: str + + +class PointCreateRequest(BaseModel): + """Create one point. The server assigns identity, ordering, and provenance. + + `after_point_id` positions the new point rather than a raw `order_id`: the + caller says where in the sequence it goes and the server computes the + fractional key and relinks neighbours, which a client-supplied `order_id` + could not do correctly (ADR-0002). `None` means "at the start of the file". + """ + + model_config = ConfigDict(extra="forbid") + + file_id: uuid.UUID + content: str = Field(min_length=1) + content_type: str = "paragraph" + after_point_id: uuid.UUID | None = None + + +class PointReplaceRequest(BaseModel): + """Replace a point's content under a version guard. + + `version` here is the *expected* version, not a value being written — the + optimistic-concurrency precondition. A mismatch is `409`. + """ + + model_config = ConfigDict(extra="forbid") + + content: str = Field(min_length=1) + content_type: str | None = None + version: int + + +class PointPayloadPatchRequest(BaseModel): + """Payload-only update of caller-writable fields, under a version guard.""" + + model_config = ConfigDict(extra="forbid") + + payload: dict[str, object] + version: int + + +class PointReorderRequest(BaseModel): + """Move a point to sit immediately after `after_point_id`. + + `None` moves it to the front of the file. Expressed as a neighbour rather + than an `order_id` for the same reason as `PointCreateRequest`. + """ + + model_config = ConfigDict(extra="forbid") + + after_point_id: uuid.UUID | None = None + version: int diff --git a/src/application/points/point.py b/src/application/points/point.py new file mode 100644 index 0000000..fc374ae --- /dev/null +++ b/src/application/points/point.py @@ -0,0 +1,126 @@ +"""The caller-facing point model for `/v1/points` (ADR-0001, ADR-0002). + +`ChunkPoint` in `models.py` is the *write* shape ingestion upserts: an id, its +named vectors, and an opaque payload dict. This module is the *read/edit* shape +the `/v1/points` surface works in, where the payload's individual fields matter +and the distinction between what a caller may write and what the server owns is +a security boundary rather than a convention. + +That split is the reason this is a model and not a dict. ADR-0002's isolation +rule ("never accepted as client-supplied input") and its optimistic-concurrency +guard both fail open if a caller can smuggle `tenant_id` or `version` through a +payload update, so the writable field set is enumerated in one place here and +every write path validates against it. +""" + +import uuid +from collections.abc import Mapping +from datetime import datetime +from typing import Self + +from pydantic import BaseModel, ConfigDict, Field + +# Fields the server derives and a caller may never set, patch, or override. +# `tenant_id` is authority, `version` is the concurrency guard, `chunk_index` +# derives the point id, and the rest are provenance the server timestamps. +SERVER_OWNED_FIELDS: frozenset[str] = frozenset( + { + "tenant_id", + "version", + "chunk_index", + "chunk_id", + "is_active", + "deleted_at", + "created_at", + "created_by", + "updated_at", + "updated_by", + "content_hash", + "embedding_model_version", + } +) + +# Fields a caller may supply on create, replace, or payload patch. `order_id` +# is writable on create but moves only through `PATCH /v1/points/{id}/order` +# afterwards, because a bare `order_id` write would not relink neighbours. +CALLER_WRITABLE_FIELDS: frozenset[str] = frozenset( + { + "content", + "content_type", + "domain", + "source_filename", + "source_type", + "order_id", + "previous_chunk_id", + "next_chunk_id", + } +) + + +class Point(BaseModel): + """One Qdrant point, read back with its ADR-0001 payload fields typed. + + Vectors are deliberately absent: ADR-0008 returns them only when explicitly + requested, and every read path that does not ask for them should not pay to + deserialize them. + """ + + model_config = ConfigDict(frozen=True) + + point_id: uuid.UUID + + tenant_id: uuid.UUID + domain: str + file_id: uuid.UUID + chunk_id: uuid.UUID + + content: str + content_type: str + source_filename: str + source_type: str + + order_id: float + chunk_index: int + previous_chunk_id: uuid.UUID | None = None + next_chunk_id: uuid.UUID | None = None + + is_active: bool = True + deleted_at: datetime | None = None + + created_at: datetime + updated_at: datetime + created_by: str + updated_by: str + + version: int + content_hash: str + embedding_model_version: str + + # Only populated when the caller explicitly asked for vectors. + vectors: dict[str, object] | None = Field(default=None) + + @classmethod + def from_payload( + cls, + point_id: uuid.UUID, + payload: Mapping[str, object], + *, + vectors: Mapping[str, object] | None = None, + ) -> Self: + """Build a `Point` from a raw Qdrant payload dict. + + Lives here rather than in the Qdrant adapter so the payload field names + are declared once, next to the model that mirrors them. The adapter + stays responsible for talking to the SDK, not for knowing ADR-0001's + schema twice. + """ + return cls.model_validate({**payload, "point_id": point_id, "vectors": vectors}) + + +class PointNotFoundError(LookupError): + """No such point *within the requesting tenant*. + + Routes map this to `404`, never `403` — a caller must not be able to probe + for the existence of another tenant's point ids (ADR-0016). The error + deliberately carries no hint about which of the two cases occurred. + """ diff --git a/src/application/ports/point_repository.py b/src/application/ports/point_repository.py new file mode 100644 index 0000000..9d58c9c --- /dev/null +++ b/src/application/ports/point_repository.py @@ -0,0 +1,131 @@ +"""The point read/edit port for `/v1/points` (ADR-0002, ADR-0015). + +Separate from `PointStorage`, which stays exactly the two bulk operations +ingestion performs. Reads, single-point edits, reordering, and keyword search +have a different caller, a different failure vocabulary, and a different +tenant-filter obligation, so they get their own port rather than accreting onto +the ingestion one. + +`tenant_id` is a required keyword argument on **every** method. That is not +style: ADR-0002's isolation rule has to hold on every code path that touches the +collection, and an optional tenant filter is one forgotten argument away from a +cross-tenant read. Making it required moves that from a review question to a +type error. + +`src/infrastructure/qdrant/point_repository.py` is the production adapter; +`tests.fakes.FakePointRepository` is the test double. +""" + +import uuid +from collections.abc import Sequence +from typing import Protocol + +from pydantic import BaseModel + +from src.application.points.point import Point + + +class PointPage(BaseModel): + """One page of points plus the cursor that continues it. + + `next_cursor` is opaque to callers and encoded by the adapter: ordered + scrolls and keyword searches paginate by different Qdrant mechanisms, and + neither is a plain integer offset. `None` means the listing is exhausted. + """ + + points: tuple[Point, ...] + next_cursor: str | None = None + + +class PayloadPatch(BaseModel): + """Set these payload fields on one point, optionally guarded by `version`. + + When `expected_version` is set, the adapter attaches it to the operation's + filter, so a concurrent write that has already moved the version on means + this patch matches nothing rather than clobbering it. The guard is what + makes a lost update impossible; detecting that it fired is the service's + job (see `apply_patches`). + """ + + point_id: uuid.UUID + payload: dict[str, object] + expected_version: int | None = None + + +class PointRepository(Protocol): + async def get( + self, *, tenant_id: uuid.UUID, point_id: uuid.UUID, with_vectors: bool = False + ) -> Point | None: + """One point, or `None` if it does not exist *under this tenant*. + + The two cases are deliberately indistinguishable — the route maps both + to `404` so a caller cannot probe for another tenant's point ids. + """ + ... + + async def get_many( + self, *, tenant_id: uuid.UUID, point_ids: Sequence[uuid.UUID] + ) -> tuple[Point, ...]: + """The subset of `point_ids` that exists under this tenant. + + Order is not guaranteed and missing ids are silently absent: callers are + neighbour-relinking and batch precondition checks, both of which match + on id rather than position. + """ + ... + + async def list_by_file( + self, + *, + tenant_id: uuid.UUID, + file_id: uuid.UUID, + limit: int, + cursor: str | None = None, + include_inactive: bool = False, + ) -> PointPage: + """One file's points in `order_id` order (ADR-0008's `scroll`). + + Scoped to a single file because the cursor is an `order_id` value, and + `order_id` is only unique within a file. + """ + ... + + async def count( + self, + *, + tenant_id: uuid.UUID, + domain: str | None = None, + file_id: uuid.UUID | None = None, + include_inactive: bool = False, + ) -> int: ... + + async def keyword_search( + self, + *, + tenant_id: uuid.UUID, + query: str, + limit: int, + cursor: str | None = None, + domain: str | None = None, + file_id: uuid.UUID | None = None, + include_inactive: bool = False, + ) -> PointPage: + """Full-text payload match on `content`, plus structured filters. + + Keyword matching, **not** semantic retrieval (ADR-0002). Qdrant's + full-text index is a filter, not a scorer, so results carry no relevance + ranking and their order is unspecified. + """ + ... + + async def apply_patches(self, *, tenant_id: uuid.UUID, patches: Sequence[PayloadPatch]) -> None: + """Apply every patch in one Qdrant `points/batch` request. + + Qdrant has no multi-point transaction, so this is not atomic and does + not pretend to be. ADR-0002's all-or-nothing rule is implemented one + layer up as validate-every-precondition-then-apply; the per-patch + `expected_version` guard here is what makes the residual window safe, + turning a lost update into a no-op the service can detect rather than a + silent clobber. + """ + ... diff --git a/src/bootstrap/dependencies.py b/src/bootstrap/dependencies.py index b1a2ee0..9f7a57b 100644 --- a/src/bootstrap/dependencies.py +++ b/src/bootstrap/dependencies.py @@ -9,6 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker from src.application.ports.embedding import DenseEmbedder, SparseEmbedder from src.application.ports.object_storage import ObjectStorage +from src.application.ports.point_repository import PointRepository from src.application.ports.point_storage import PointStorage from src.config import Settings @@ -22,6 +23,7 @@ class AppResources: qdrant_client: AsyncQdrantClient object_storage: ObjectStorage point_storage: PointStorage + point_repository: PointRepository ingestion_limiter: CapacityLimiter dense_embedders: Sequence[DenseEmbedder] sparse_embedder: SparseEmbedder @@ -52,6 +54,10 @@ def get_point_storage(request: Request) -> PointStorage: return _resources(request).point_storage +def get_point_repository(request: Request) -> PointRepository: + return _resources(request).point_repository + + def get_ingestion_limiter(request: Request) -> CapacityLimiter: return _resources(request).ingestion_limiter diff --git a/src/bootstrap/lifespan.py b/src/bootstrap/lifespan.py index 7b2c7b9..c11b01a 100644 --- a/src/bootstrap/lifespan.py +++ b/src/bootstrap/lifespan.py @@ -20,6 +20,7 @@ from src.infrastructure.minio.storage import MinioObjectStorage from src.infrastructure.observability.logging import configure_logging from src.infrastructure.postgres.database import create_engine, create_sessionmaker from src.infrastructure.qdrant.client import create_client as create_qdrant_client +from src.infrastructure.qdrant.point_repository import QdrantPointRepository from src.infrastructure.qdrant.points import QdrantPointStorage logger = structlog.get_logger(__name__) @@ -85,6 +86,9 @@ def create_lifespan( point_storage = QdrantPointStorage( qdrant_client, collection=resolved_settings.qdrant.collection ) + point_repository = QdrantPointRepository( + qdrant_client, collection=resolved_settings.qdrant.collection + ) logger.info("lifespan.qdrant.client.created") nomic_settings = resolved_settings.embedding.nomic @@ -145,6 +149,7 @@ def create_lifespan( qdrant_client=qdrant_client, object_storage=object_storage, point_storage=point_storage, + point_repository=point_repository, ingestion_limiter=ingestion_limiter, dense_embedders=dense_embedders, sparse_embedder=sparse_embedder, diff --git a/src/infrastructure/qdrant/point_repository.py b/src/infrastructure/qdrant/point_repository.py new file mode 100644 index 0000000..b321907 --- /dev/null +++ b/src/infrastructure/qdrant/point_repository.py @@ -0,0 +1,247 @@ +"""Qdrant adapter for the `PointRepository` port (ADR-0002, ADR-0015). + +Every Qdrant filter for the `/v1/points` surface is built here. Routers and +application services never import `qdrant_client` — they pass a `tenant_id` and +get `Point` models back. + +Two mechanics are worth reading before changing anything: + +**Reads go through `scroll`, not `retrieve`.** `retrieve` fetches by id and +takes no filter, which would force the tenant check to happen *after* Qdrant +answered — exactly the "check it in Python afterwards" shape ADR-0002's +isolation rule exists to prevent. `scroll` with a `HasIdCondition` plus the +tenant condition pushes the check server-side, so a foreign id returns an empty +page rather than a row this code has to remember to reject. + +**Ordered listing paginates by `order_id` value, not by offset.** Qdrant does +not return a `next_page_offset` when `order_by` is set, and an offset-based +cursor would skip or repeat rows when a concurrent insert shifts positions +underneath the reader. Ranging on `order_id > cursor` is stable instead: a point +inserted ahead of the cursor was already returned, and one inserted after it +shows up on a later page. This relies on `order_id` being unique within a file, +which ADR-0002 guarantees by rejecting a reorder whose gap would collapse onto a +neighbour value. +""" + +import uuid +from collections.abc import Sequence + +from qdrant_client import AsyncQdrantClient, models + +from src.application.points.point import Point +from src.application.ports.point_repository import PayloadPatch, PointPage + +# Qdrant's scroll returns `(records, next_page_offset)`; a record's id is a +# `str | int` union in the SDK, and every id this service writes is a UUID. +type _ScrollRecord = models.Record + + +def _uuid_of(record: _ScrollRecord) -> uuid.UUID: + return uuid.UUID(str(record.id)) + + +def _to_point(record: _ScrollRecord, *, with_vectors: bool) -> Point: + vectors = record.vector if with_vectors and isinstance(record.vector, dict) else None + return Point.from_payload(_uuid_of(record), record.payload or {}, vectors=vectors) + + +def _conditions( + *, + tenant_id: uuid.UUID, + domain: str | None = None, + file_id: uuid.UUID | None = None, + include_inactive: bool = False, +) -> list[models.Condition]: + """The base filter every point query carries. + + `tenant_id` is unconditional and first. `is_active` is added unless the + caller explicitly opted into inactive points, which is ADR-0002's + "`is_active: true` implied" rule expressed once rather than per method. + """ + conditions: list[models.Condition] = [ + models.FieldCondition(key="tenant_id", match=models.MatchValue(value=str(tenant_id))) + ] + if domain is not None: + conditions.append( + models.FieldCondition(key="domain", match=models.MatchValue(value=domain)) + ) + if file_id is not None: + conditions.append( + models.FieldCondition(key="file_id", match=models.MatchValue(value=str(file_id))) + ) + if not include_inactive: + conditions.append( + models.FieldCondition(key="is_active", match=models.MatchValue(value=True)) + ) + return conditions + + +class QdrantPointRepository: + """A `PointRepository` (see `src/application/ports/point_repository.py`).""" + + def __init__(self, client: AsyncQdrantClient, *, collection: str) -> None: + self._client = client + self._collection = collection + + async def get( + self, *, tenant_id: uuid.UUID, point_id: uuid.UUID, with_vectors: bool = False + ) -> Point | None: + records, _ = await self._client.scroll( + collection_name=self._collection, + scroll_filter=models.Filter( + must=[ + *_conditions(tenant_id=tenant_id, include_inactive=True), + models.HasIdCondition(has_id=[str(point_id)]), + ] + ), + limit=1, + with_payload=True, + with_vectors=with_vectors, + ) + if not records: + return None + return _to_point(records[0], with_vectors=with_vectors) + + async def get_many( + self, *, tenant_id: uuid.UUID, point_ids: Sequence[uuid.UUID] + ) -> tuple[Point, ...]: + if not point_ids: + return () + records, _ = await self._client.scroll( + collection_name=self._collection, + scroll_filter=models.Filter( + must=[ + *_conditions(tenant_id=tenant_id, include_inactive=True), + models.HasIdCondition(has_id=[str(point_id) for point_id in point_ids]), + ] + ), + limit=len(point_ids), + with_payload=True, + with_vectors=False, + ) + return tuple(_to_point(record, with_vectors=False) for record in records) + + async def list_by_file( + self, + *, + tenant_id: uuid.UUID, + file_id: uuid.UUID, + limit: int, + cursor: str | None = None, + include_inactive: bool = False, + ) -> PointPage: + conditions = _conditions( + tenant_id=tenant_id, file_id=file_id, include_inactive=include_inactive + ) + if cursor is not None: + conditions.append( + models.FieldCondition(key="order_id", range=models.Range(gt=float(cursor))) + ) + + records, _ = await self._client.scroll( + collection_name=self._collection, + scroll_filter=models.Filter(must=conditions), + order_by=models.OrderBy(key="order_id", direction=models.Direction.ASC), + limit=limit, + with_payload=True, + with_vectors=False, + ) + points = tuple(_to_point(record, with_vectors=False) for record in records) + # A short page means the listing is exhausted. A full page might be, but + # claiming so would need an extra round trip; handing back a cursor that + # yields an empty final page is the cheaper honest answer. + next_cursor = repr(points[-1].order_id) if len(points) == limit else None + return PointPage(points=points, next_cursor=next_cursor) + + async def count( + self, + *, + tenant_id: uuid.UUID, + domain: str | None = None, + file_id: uuid.UUID | None = None, + include_inactive: bool = False, + ) -> int: + result = await self._client.count( + collection_name=self._collection, + count_filter=models.Filter( + must=_conditions( + tenant_id=tenant_id, + domain=domain, + file_id=file_id, + include_inactive=include_inactive, + ) + ), + exact=True, + ) + return result.count + + async def keyword_search( + self, + *, + tenant_id: uuid.UUID, + query: str, + limit: int, + cursor: str | None = None, + domain: str | None = None, + file_id: uuid.UUID | None = None, + include_inactive: bool = False, + ) -> PointPage: + conditions = _conditions( + tenant_id=tenant_id, + domain=domain, + file_id=file_id, + include_inactive=include_inactive, + ) + conditions.append(models.FieldCondition(key="content", match=models.MatchText(text=query))) + + # No `order_by` here, so Qdrant does return a page offset: the full-text + # index filters rather than scores, and imposing `order_id` ordering + # across files would be meaningless (`order_id` is per-file). + records, next_offset = await self._client.scroll( + collection_name=self._collection, + scroll_filter=models.Filter(must=conditions), + limit=limit, + offset=cursor, + with_payload=True, + with_vectors=False, + ) + return PointPage( + points=tuple(_to_point(record, with_vectors=False) for record in records), + next_cursor=str(next_offset) if next_offset is not None else None, + ) + + async def apply_patches(self, *, tenant_id: uuid.UUID, patches: Sequence[PayloadPatch]) -> None: + if not patches: + return + await self._client.batch_update_points( + collection_name=self._collection, + update_operations=[ + models.SetPayloadOperation( + set_payload=models.SetPayload( + payload=patch.payload, + filter=models.Filter(must=self._patch_conditions(tenant_id, patch)), + ) + ) + for patch in patches + ], + ) + + @staticmethod + def _patch_conditions(tenant_id: uuid.UUID, patch: PayloadPatch) -> list[models.Condition]: + """Address one point by id, under this tenant, at an expected version. + + `include_inactive=True`: soft-deleting and relinking both have to reach + points the default read filter hides, and a patch is addressed by an + explicit id rather than discovered by a listing. + """ + conditions: list[models.Condition] = [ + *_conditions(tenant_id=tenant_id, include_inactive=True), + models.HasIdCondition(has_id=[str(patch.point_id)]), + ] + if patch.expected_version is not None: + conditions.append( + models.FieldCondition( + key="version", match=models.MatchValue(value=patch.expected_version) + ) + ) + return conditions