test(points): hold the fake and the Qdrant adapter to one shared contract

Why:
- Two parallel test files let a fake drift more permissive than the store it
  stands in for, so unit tests stay green while production diverges. Plan 002
  Phase 1's exit criterion is precisely that the two agree.

Changes:
- One scenario suite in tests/support/point_contract.py, run against
  FakePointRepository (unit) and QdrantPointRepository (integration). A
  divergence fails one of the two runs rather than hiding.
- The fake models the behaviours services branch on: the implied is_active read
  filter, value-based cursor pagination, and a stale version guard that matches
  nothing rather than raising -- the no-op Qdrant's filtered set_payload actually
  has, and the reason a service must read back to know its write landed.
- Patched points are re-validated rather than model_copy'd, so the fake holds a
  datetime where a read from real Qdrant returns one.
- The seeded corpus gives each tenant its own file: point IDs derive from
  file_id plus chunk_index alone, so two tenants in one file would collide on a
  single ID and the fixture would assert an impossible state.

Impact:
- 15 scenarios pass against both implementations.

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

tests/fakes.py

@@ -8,6 +8,8 @@ from datetime import datetime
 
 from src.application.ingestion.models import SparseVector
 from src.application.points.models import ChunkPoint
+from src.application.points.point import Point
+from src.application.ports.point_repository import PayloadPatch, PointPage
 
 
 @dataclass
@@ -139,3 +141,161 @@ class FakePointStorage:
             point.payload["is_active"] = False
             point.payload["deleted_at"] = deleted_at.isoformat()
         return len(stale)
+
+
+@dataclass
+class FakePointRepository:
+    """An in-memory `PointRepository`.
+
+    Stores `Point` models keyed by point id. The filter, ordering, and version
+    semantics below are reimplemented in Python rather than stubbed, because
+    every service this fake stands behind is being tested *for* those semantics:
+    a stub that returned points in insertion order would pass a reorder test
+    that real Qdrant fails.
+
+    Three behaviours are modelled deliberately:
+
+    - `is_active` is filtered out unless `include_inactive`, matching ADR-0002's
+      implied read filter.
+    - `list_by_file` sorts by `order_id` and paginates by value, the same cursor
+      mechanism the Qdrant adapter uses, so a test can catch a cursor that skips
+      or repeats.
+    - `apply_patches` honours `expected_version` by silently matching nothing on
+      a mismatch — the no-op-not-error behaviour Qdrant's filtered `set_payload`
+      actually has, which is what makes the service's post-check necessary.
+    """
+
+    points: dict[str, Point] = field(default_factory=dict)
+    patch_batches: list[list[PayloadPatch]] = field(default_factory=list)
+    fail_next: bool = False
+
+    def add(self, point: Point) -> None:
+        """Seed a point. Test-only helper, not part of the port."""
+        self.points[str(point.point_id)] = point
+
+    def _visible(
+        self,
+        *,
+        tenant_id: uuid.UUID,
+        domain: str | None = None,
+        file_id: uuid.UUID | None = None,
+        include_inactive: bool = False,
+    ) -> list[Point]:
+        return [
+            point
+            for point in self.points.values()
+            if point.tenant_id == tenant_id
+            and (domain is None or point.domain == domain)
+            and (file_id is None or point.file_id == file_id)
+            and (include_inactive or point.is_active)
+        ]
+
+    async def get(
+        self, *, tenant_id: uuid.UUID, point_id: uuid.UUID, with_vectors: bool = False
+    ) -> Point | None:
+        point = self.points.get(str(point_id))
+        if point is None or point.tenant_id != tenant_id:
+            return None
+        return point if with_vectors else point.model_copy(update={"vectors": None})
+
+    async def get_many(
+        self, *, tenant_id: uuid.UUID, point_ids: Sequence[uuid.UUID]
+    ) -> tuple[Point, ...]:
+        wanted = {str(point_id) for point_id in point_ids}
+        return tuple(
+            point
+            for key, point in self.points.items()
+            if key in wanted and point.tenant_id == tenant_id
+        )
+
+    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:
+        ordered = sorted(
+            self._visible(tenant_id=tenant_id, file_id=file_id, include_inactive=include_inactive),
+            key=lambda point: point.order_id,
+        )
+        if cursor is not None:
+            ordered = [point for point in ordered if point.order_id > float(cursor)]
+
+        page = tuple(ordered[:limit])
+        next_cursor = repr(page[-1].order_id) if len(page) == limit else None
+        return PointPage(points=page, 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:
+        return len(
+            self._visible(
+                tenant_id=tenant_id,
+                domain=domain,
+                file_id=file_id,
+                include_inactive=include_inactive,
+            )
+        )
+
+    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:
+        # Whole-token containment, lowercased: an approximation of Qdrant's
+        # full-text index good enough to tell "matched" from "did not", which is
+        # all the service layer branches on. Ranking is not modelled because the
+        # real index does not rank either (ADR-0002).
+        terms = query.lower().split()
+        matches = [
+            point
+            for point in self._visible(
+                tenant_id=tenant_id,
+                domain=domain,
+                file_id=file_id,
+                include_inactive=include_inactive,
+            )
+            if all(term in point.content.lower().split() for term in terms)
+        ]
+        start = int(cursor) if cursor is not None else 0
+        page = tuple(matches[start : start + limit])
+        next_start = start + limit
+        return PointPage(
+            points=page, next_cursor=str(next_start) if next_start < len(matches) else None
+        )
+
+    async def apply_patches(self, *, tenant_id: uuid.UUID, patches: Sequence[PayloadPatch]) -> None:
+        self.patch_batches.append(list(patches))
+        if self.fail_next:
+            self.fail_next = False
+            raise RuntimeError("simulated point repository failure")
+
+        for patch in patches:
+            point = self.points.get(str(patch.point_id))
+            if point is None or point.tenant_id != tenant_id:
+                continue
+            if patch.expected_version is not None and point.version != patch.expected_version:
+                continue
+            # Re-validated rather than `model_copy`d, because a patch payload
+            # carries wire values (an ISO `deleted_at`, a stringified
+            # `previous_chunk_id`) exactly as it would reach Qdrant. Copying
+            # without validation would leave the fake holding a `str` where a
+            # read from real Qdrant returns a `datetime`, and a service bug that
+            # depends on the difference would pass here and fail in production.
+            updated = point.model_dump()
+            updated.update(patch.payload)
+            self.points[str(patch.point_id)] = Point.model_validate(updated)

tests/integration/qdrant/test_point_repository.py

@@ -0,0 +1,112 @@
+"""`QdrantPointRepository` against the shared `PointRepository` contract.
+
+The identical scenarios run against the in-memory fake in
+`tests/unit/application/points/test_point_repository_contract.py`. This file is
+the half that decides whether the fake is telling the truth: filter
+construction, `order_by: order_id` scroll, cursor pagination, full-text
+matching, and the filtered `set_payload` version guard are all Qdrant
+behaviours a fake can only approximate, and plan 002 Phase 1's exit criterion is
+that the approximation holds.
+
+Seeding differs from the fake's (real points need vectors), so it lives here;
+everything asserted lives in `tests/support/point_contract.py`.
+"""
+
+import uuid
+from collections.abc import Awaitable, Callable
+
+import pytest
+from qdrant_client import AsyncQdrantClient, models
+
+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.point_repository import QdrantPointRepository
+from src.infrastructure.qdrant.points import QdrantPointStorage
+from tests.support import point_contract
+from tests.support.point_contract import SeedSpec, build_point, seed_specs
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.qdrant,
+    pytest.mark.asyncio(loop_scope="session"),
+]
+
+type Scenario = Callable[..., Awaitable[None]]
+
+TENANT_A = uuid.UUID("11111111-1111-4111-8111-111111111111")
+TENANT_B = uuid.UUID("22222222-2222-4222-8222-222222222222")
+FILE_A = uuid.UUID("33333333-3333-4333-8333-333333333333")
+FILE_B = uuid.UUID("44444444-4444-4444-8444-444444444444")
+FILE_C = uuid.UUID("55555555-5555-4555-8555-555555555555")
+
+IDS = {
+    "tenant_a": TENANT_A,
+    "tenant_b": TENANT_B,
+    "file_a": FILE_A,
+    "file_b": FILE_B,
+    "file_c": FILE_C,
+}
+
+
+def _chunk_point(spec: SeedSpec) -> ChunkPoint:
+    """The contract's `Point` re-expressed as something upsertable.
+
+    `build_point` produces the read model; Qdrant needs vectors and a flat
+    payload dict, so the payload is taken straight off the model to guarantee
+    the two representations cannot drift apart.
+    """
+    point = build_point(spec)
+    payload = point.model_dump(mode="json", exclude={"point_id", "vectors"})
+    return ChunkPoint(
+        point_id=point.point_id,
+        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=payload,
+    )
+
+
+async def _seeded(
+    qdrant_client: AsyncQdrantClient, qdrant_settings: QdrantSettings
+) -> QdrantPointRepository:
+    await ensure_chunks_collection(qdrant_client, collection=qdrant_settings.collection)
+    storage = QdrantPointStorage(qdrant_client, collection=qdrant_settings.collection)
+    await storage.upsert_points(
+        [_chunk_point(spec) for spec in seed_specs(TENANT_A, TENANT_B, FILE_A, FILE_B, FILE_C)]
+    )
+    # Payload indexes are built asynchronously; without waiting, a full-text or
+    # ordered query can run against a half-built index and return short.
+    await _await_indexing(qdrant_client, qdrant_settings.collection)
+    return QdrantPointRepository(qdrant_client, collection=qdrant_settings.collection)
+
+
+async def _await_indexing(client: AsyncQdrantClient, collection: str) -> None:
+    for _ in range(100):
+        info = await client.get_collection(collection)
+        if info.status == models.CollectionStatus.GREEN and info.indexed_vectors_count is not None:
+            return
+    raise AssertionError(f"collection {collection!r} did not finish indexing")
+
+
+def _arguments(scenario: Scenario) -> dict[str, uuid.UUID]:
+    return {name: value for name, value in IDS.items() if name in scenario.__annotations__}
+
+
+@pytest.mark.parametrize(
+    "scenario",
+    [*point_contract.READ_SCENARIOS, *point_contract.WRITE_SCENARIOS],
+    ids=lambda scenario: scenario.__name__.removeprefix("scenario_"),
+)
+async def test_qdrant_point_repository_satisfies_the_contract(
+    scenario: Scenario, qdrant_client: AsyncQdrantClient, qdrant_settings: QdrantSettings
+) -> None:
+    repository = await _seeded(qdrant_client, qdrant_settings)
+    await scenario(repository, **_arguments(scenario))

tests/support/point_contract.py

@@ -0,0 +1,327 @@
+"""One behavioural contract, run against both `PointRepository` implementations.
+
+Plan 002 Phase 1's exit criterion is that the Qdrant adapter and the in-memory
+fake agree on filtering, ordering, and pagination. Asserting that by writing two
+parallel test files invites them to drift — the usual outcome is a fake that
+quietly grows more forgiving than the real store, so unit tests keep passing
+while production does something else.
+
+So the scenarios live here once. `tests/unit/application/points/` runs them
+against `FakePointRepository`; `tests/integration/qdrant/` runs the identical
+functions against real Qdrant. A divergence fails one of the two runs rather
+than hiding.
+
+Each scenario takes an already-seeded repository plus the ids it was seeded
+with, so seeding (which genuinely differs — one writes a dict, the other upserts
+vectors) stays outside the shared code.
+"""
+
+import uuid
+from dataclasses import dataclass
+from datetime import UTC, datetime
+
+from src.application.ingestion.chunking import chunk_id_for
+from src.application.points.point import Point
+from src.application.ports.point_repository import PayloadPatch, PointRepository
+
+# Two tenants and two files, so every scenario can assert both that the right
+# rows come back and that the wrong ones do not.
+SEEDED_AT = datetime(2026, 1, 1, 12, 0, tzinfo=UTC)
+
+
+@dataclass(frozen=True)
+class SeedSpec:
+    """One point to seed. Deliberately flat — the fixture builds the model."""
+
+    tenant_id: uuid.UUID
+    file_id: uuid.UUID
+    chunk_index: int
+    content: str
+    order_id: float
+    is_active: bool = True
+    version: int = 1
+    domain: str = "support"
+
+
+def point_id_for(file_id: uuid.UUID, chunk_index: int) -> uuid.UUID:
+    """The deterministic point id (ADR-0001), re-exported for readability."""
+    return chunk_id_for(file_id, chunk_index)
+
+
+def build_point(spec: SeedSpec) -> Point:
+    point_id = point_id_for(spec.file_id, spec.chunk_index)
+    return Point(
+        point_id=point_id,
+        tenant_id=spec.tenant_id,
+        domain=spec.domain,
+        file_id=spec.file_id,
+        chunk_id=point_id,
+        content=spec.content,
+        content_type="paragraph",
+        source_filename="handbook.docx",
+        source_type="docx",
+        order_id=spec.order_id,
+        chunk_index=spec.chunk_index,
+        previous_chunk_id=None,
+        next_chunk_id=None,
+        is_active=spec.is_active,
+        deleted_at=None if spec.is_active else SEEDED_AT,
+        created_at=SEEDED_AT,
+        updated_at=SEEDED_AT,
+        created_by="ingestion",
+        updated_by="ingestion",
+        version=spec.version,
+        content_hash="0" * 64,
+        embedding_model_version="test-model",
+    )
+
+
+def seed_specs(
+    tenant_a: uuid.UUID,
+    tenant_b: uuid.UUID,
+    file_a: uuid.UUID,
+    file_b: uuid.UUID,
+    file_c: uuid.UUID,
+) -> tuple[SeedSpec, ...]:
+    """The fixed corpus every scenario below assumes.
+
+    Ordering is deliberately not insertion order: `order_id` 3.0 is seeded
+    before 2.0 so a repository that returns rows in write order fails
+    `scenario_list_orders_by_order_id` instead of passing by accident.
+
+    Tenant B owns its own file. Sharing a `file_id` across tenants would be a
+    fixture that cannot occur: point ids are derived from `file_id` plus
+    `chunk_index` alone, so two tenants in one file would collide on a single
+    id and the corpus would be testing an impossible state.
+    """
+    return (
+        SeedSpec(tenant_a, file_a, 0, "onboarding checklist for new staff", 1.0),
+        SeedSpec(tenant_a, file_a, 2, "payroll schedule and bank details", 3.0),
+        SeedSpec(tenant_a, file_a, 1, "expense policy and receipts", 2.0),
+        SeedSpec(tenant_a, file_a, 3, "retired parking policy", 4.0, is_active=False),
+        SeedSpec(tenant_a, file_b, 0, "supplier onboarding contacts", 1.0, domain="ops"),
+        SeedSpec(tenant_b, file_c, 0, "other tenant onboarding secrets", 1.0),
+    )
+
+
+async def scenario_get_returns_point_within_tenant(
+    repository: PointRepository, *, tenant_a: uuid.UUID, file_a: uuid.UUID
+) -> None:
+    point = await repository.get(tenant_id=tenant_a, point_id=point_id_for(file_a, 0))
+    assert point is not None
+    assert point.content == "onboarding checklist for new staff"
+    assert point.chunk_index == 0
+    assert point.order_id == 1.0
+
+
+async def scenario_get_hides_other_tenants_point(
+    repository: PointRepository, *, tenant_b: uuid.UUID, file_a: uuid.UUID
+) -> None:
+    """Tenant B probing a real point id that belongs to tenant A gets nothing.
+
+    The point exists, so this distinguishes a tenant filter from a plain
+    existence check — the difference between `404` and a leak.
+    """
+    assert await repository.get(tenant_id=tenant_b, point_id=point_id_for(file_a, 0)) is None
+
+
+async def scenario_get_reaches_inactive_points(
+    repository: PointRepository, *, tenant_a: uuid.UUID, file_a: uuid.UUID
+) -> None:
+    """A direct `get` is not subject to the implied `is_active` read filter.
+
+    Soft delete and relinking both address points by id and must be able to see
+    a deactivated one; only listings hide them.
+    """
+    point = await repository.get(tenant_id=tenant_a, point_id=point_id_for(file_a, 3))
+    assert point is not None
+    assert point.is_active is False
+
+
+async def scenario_get_many_returns_only_this_tenants_points(
+    repository: PointRepository, *, tenant_a: uuid.UUID, file_a: uuid.UUID
+) -> None:
+    found = await repository.get_many(
+        tenant_id=tenant_a,
+        point_ids=[point_id_for(file_a, 0), point_id_for(file_a, 1), uuid.uuid4()],
+    )
+    assert {point.chunk_index for point in found} == {0, 1}
+
+
+async def scenario_list_orders_by_order_id(
+    repository: PointRepository, *, tenant_a: uuid.UUID, file_a: uuid.UUID
+) -> None:
+    page = await repository.list_by_file(tenant_id=tenant_a, file_id=file_a, limit=10)
+    assert [point.order_id for point in page.points] == [1.0, 2.0, 3.0]
+    assert page.next_cursor is None
+
+
+async def scenario_list_excludes_inactive_by_default(
+    repository: PointRepository, *, tenant_a: uuid.UUID, file_a: uuid.UUID
+) -> None:
+    default = await repository.list_by_file(tenant_id=tenant_a, file_id=file_a, limit=10)
+    assert 4.0 not in [point.order_id for point in default.points]
+
+    opted_in = await repository.list_by_file(
+        tenant_id=tenant_a, file_id=file_a, limit=10, include_inactive=True
+    )
+    assert [point.order_id for point in opted_in.points] == [1.0, 2.0, 3.0, 4.0]
+
+
+async def scenario_list_paginates_without_skipping_or_repeating(
+    repository: PointRepository, *, tenant_a: uuid.UUID, file_a: uuid.UUID
+) -> None:
+    first = await repository.list_by_file(tenant_id=tenant_a, file_id=file_a, limit=2)
+    assert [point.order_id for point in first.points] == [1.0, 2.0]
+    assert first.next_cursor is not None
+
+    second = await repository.list_by_file(
+        tenant_id=tenant_a, file_id=file_a, limit=2, cursor=first.next_cursor
+    )
+    assert [point.order_id for point in second.points] == [3.0]
+    assert second.next_cursor is None
+
+
+async def scenario_list_is_scoped_to_one_file(
+    repository: PointRepository, *, tenant_a: uuid.UUID, file_b: uuid.UUID
+) -> None:
+    page = await repository.list_by_file(tenant_id=tenant_a, file_id=file_b, limit=10)
+    assert [point.content for point in page.points] == ["supplier onboarding contacts"]
+
+
+async def scenario_count_respects_tenant_and_filters(
+    repository: PointRepository,
+    *,
+    tenant_a: uuid.UUID,
+    tenant_b: uuid.UUID,
+    file_a: uuid.UUID,
+) -> None:
+    assert await repository.count(tenant_id=tenant_a) == 4
+    assert await repository.count(tenant_id=tenant_a, include_inactive=True) == 5
+    assert await repository.count(tenant_id=tenant_a, file_id=file_a) == 3
+    assert await repository.count(tenant_id=tenant_a, domain="ops") == 1
+    assert await repository.count(tenant_id=tenant_b) == 1
+
+
+async def scenario_keyword_search_matches_content_within_tenant(
+    repository: PointRepository, *, tenant_a: uuid.UUID, tenant_b: uuid.UUID
+) -> None:
+    found = await repository.keyword_search(tenant_id=tenant_a, query="onboarding", limit=10)
+    assert {point.content for point in found.points} == {
+        "onboarding checklist for new staff",
+        "supplier onboarding contacts",
+    }
+
+    # The same term matches a different tenant's point, which must not leak.
+    other = await repository.keyword_search(tenant_id=tenant_b, query="onboarding", limit=10)
+    assert {point.content for point in other.points} == {"other tenant onboarding secrets"}
+
+
+async def scenario_keyword_search_excludes_inactive_by_default(
+    repository: PointRepository, *, tenant_a: uuid.UUID
+) -> None:
+    """ "retired parking policy" is seeded inactive, so only the opt-in sees it."""
+    default = await repository.keyword_search(tenant_id=tenant_a, query="parking", limit=10)
+    assert default.points == ()
+
+    opted_in = await repository.keyword_search(
+        tenant_id=tenant_a, query="parking", limit=10, include_inactive=True
+    )
+    assert {point.content for point in opted_in.points} == {"retired parking policy"}
+
+
+async def scenario_keyword_search_filters_by_domain(
+    repository: PointRepository, *, tenant_a: uuid.UUID
+) -> None:
+    found = await repository.keyword_search(
+        tenant_id=tenant_a, query="onboarding", limit=10, domain="ops"
+    )
+    assert {point.content for point in found.points} == {"supplier onboarding contacts"}
+
+
+async def scenario_patch_applies_when_version_matches(
+    repository: PointRepository, *, tenant_a: uuid.UUID, file_a: uuid.UUID
+) -> None:
+    target = point_id_for(file_a, 0)
+    await repository.apply_patches(
+        tenant_id=tenant_a,
+        patches=[
+            PayloadPatch(
+                point_id=target,
+                payload={"content": "amended onboarding checklist", "version": 2},
+                expected_version=1,
+            )
+        ],
+    )
+    point = await repository.get(tenant_id=tenant_a, point_id=target)
+    assert point is not None
+    assert point.content == "amended onboarding checklist"
+    assert point.version == 2
+
+
+async def scenario_patch_is_a_noop_when_version_is_stale(
+    repository: PointRepository, *, tenant_a: uuid.UUID, file_a: uuid.UUID
+) -> None:
+    """A stale guard must match nothing rather than raise or clobber.
+
+    This is the behaviour the service's post-apply check depends on: Qdrant
+    reports success for a filtered `set_payload` that matched zero points, so
+    "did my write land?" is only answerable by reading back.
+    """
+    target = point_id_for(file_a, 1)
+    await repository.apply_patches(
+        tenant_id=tenant_a,
+        patches=[
+            PayloadPatch(
+                point_id=target,
+                payload={"content": "should not land"},
+                expected_version=99,
+            )
+        ],
+    )
+    point = await repository.get(tenant_id=tenant_a, point_id=target)
+    assert point is not None
+    assert point.content == "expense policy and receipts"
+
+
+async def scenario_patch_cannot_cross_tenants(
+    repository: PointRepository, *, tenant_a: uuid.UUID, tenant_b: uuid.UUID, file_a: uuid.UUID
+) -> None:
+    """Tenant B patching tenant A's point id must change nothing.
+
+    An unguarded patch (no `expected_version`) addressed by a known-good id is
+    the sharpest form of the probe: only the tenant condition can stop it, so a
+    repository that filtered on id alone would rewrite another tenant's row.
+    """
+    target = point_id_for(file_a, 0)
+    await repository.apply_patches(
+        tenant_id=tenant_b,
+        patches=[PayloadPatch(point_id=target, payload={"content": "hijacked"})],
+    )
+    victim = await repository.get(tenant_id=tenant_a, point_id=target)
+    assert victim is not None
+    assert victim.content == "onboarding checklist for new staff"
+
+
+# Scenarios that only read, safe to run in any order against one seeded corpus.
+READ_SCENARIOS = (
+    scenario_get_returns_point_within_tenant,
+    scenario_get_hides_other_tenants_point,
+    scenario_get_reaches_inactive_points,
+    scenario_get_many_returns_only_this_tenants_points,
+    scenario_list_orders_by_order_id,
+    scenario_list_excludes_inactive_by_default,
+    scenario_list_paginates_without_skipping_or_repeating,
+    scenario_list_is_scoped_to_one_file,
+    scenario_count_respects_tenant_and_filters,
+    scenario_keyword_search_matches_content_within_tenant,
+    scenario_keyword_search_excludes_inactive_by_default,
+    scenario_keyword_search_filters_by_domain,
+)
+
+# Scenarios that mutate, and therefore need a freshly seeded corpus each.
+WRITE_SCENARIOS = (
+    scenario_patch_applies_when_version_matches,
+    scenario_patch_is_a_noop_when_version_is_stale,
+    scenario_patch_cannot_cross_tenants,
+)

tests/unit/application/points/test_point_repository_contract.py

@@ -0,0 +1,57 @@
+"""`FakePointRepository` against the shared `PointRepository` contract.
+
+The same scenarios run against real Qdrant in
+`tests/integration/qdrant/test_point_repository.py`. Keeping both runs green is
+what stops the fake from drifting into something more permissive than the store
+it stands in for — see `tests/support/point_contract.py`.
+"""
+
+import uuid
+from collections.abc import Awaitable, Callable
+
+import pytest
+
+from src.application.ports.point_repository import PointRepository
+from tests.fakes import FakePointRepository
+from tests.support import point_contract
+from tests.support.point_contract import build_point, seed_specs
+
+pytestmark = [pytest.mark.unit, pytest.mark.asyncio]
+
+type Scenario = Callable[..., Awaitable[None]]
+
+TENANT_A = uuid.UUID("11111111-1111-4111-8111-111111111111")
+TENANT_B = uuid.UUID("22222222-2222-4222-8222-222222222222")
+FILE_A = uuid.UUID("33333333-3333-4333-8333-333333333333")
+FILE_B = uuid.UUID("44444444-4444-4444-8444-444444444444")
+FILE_C = uuid.UUID("55555555-5555-4555-8555-555555555555")
+
+IDS = {
+    "tenant_a": TENANT_A,
+    "tenant_b": TENANT_B,
+    "file_a": FILE_A,
+    "file_b": FILE_B,
+    "file_c": FILE_C,
+}
+
+
+def _seeded() -> FakePointRepository:
+    repository = FakePointRepository()
+    for spec in seed_specs(TENANT_A, TENANT_B, FILE_A, FILE_B, FILE_C):
+        repository.add(build_point(spec))
+    return repository
+
+
+def _arguments(scenario: Scenario) -> dict[str, uuid.UUID]:
+    """Pass only the ids a scenario declares, so each one names its own inputs."""
+    return {name: value for name, value in IDS.items() if name in scenario.__annotations__}
+
+
+@pytest.mark.parametrize(
+    "scenario",
+    [*point_contract.READ_SCENARIOS, *point_contract.WRITE_SCENARIOS],
+    ids=lambda scenario: scenario.__name__.removeprefix("scenario_"),
+)
+async def test_fake_point_repository_satisfies_the_contract(scenario: Scenario) -> None:
+    repository: PointRepository = _seeded()
+    await scenario(repository, **_arguments(scenario))