test(points): cover soft delete, relinking, and the file sweep

Why:
- The failure modes worth testing are races, and they are cheap to force
  against the fake and expensive to observe anywhere else.

Changes:
- Unit tests for both boundaries, the repeat delete, a missing neighbour, the
  traversal property after several deletes, and — via a repository that bumps a
  rival's version before each apply — both the partial-apply repair and the
  unconvergent 409.
- One new shared contract scenario (a multi-point batch applies every patch,
  including nulling a pointer) so it runs against the fake and real Qdrant.
- HTTP tests against real Postgres and Qdrant for relinking, soft-not-hard
  delete, cross-tenant 404, and scope enforcement on both routes.
- create_source_file factory for tests addressing a file without uploading.

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

tests/integration/qdrant/test_points_deletion_api.py

@@ -0,0 +1,276 @@
+"""Soft delete over HTTP, against real Postgres and real Qdrant (ADR-0002).
+
+The unit tests in `tests/unit/application/points/test_deletion.py` decide
+whether the relink logic is right. This file decides whether it is right
+*through the stack*: real routing and scope checks, a tenant derived from a real
+API key, and a real filtered `set_payload` batch — the last of which is the part
+a fake can only approximate, since Qdrant reports success for a patch that
+matched nothing.
+
+`DELETE /v1/files/{file_id}` is here too rather than with the upload tests: it
+spans both stores, and the assertion that matters is the one about points.
+"""
+
+import uuid
+from collections.abc import AsyncIterator
+
+import pytest
+import pytest_asyncio
+from asgi_lifespan import LifespanManager
+from httpx import ASGITransport, AsyncClient
+from qdrant_client import AsyncQdrantClient
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.bootstrap.dependencies import get_sessionmaker
+from src.config import QdrantSettings, Settings
+from src.infrastructure.postgres.repositories import source_files as source_files_repo
+from src.infrastructure.qdrant.collection import ensure_chunks_collection
+from src.infrastructure.qdrant.points import QdrantPointStorage
+from src.main import create_app
+from tests.support.factories import create_api_key, create_source_file, create_tenant
+from tests.support.point_contract import SeedSpec, chunk_point_for, point_id_for
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.qdrant,
+    pytest.mark.postgres,
+    pytest.mark.asyncio(loop_scope="session"),
+]
+
+FILE_ID = uuid.UUID("cccccccc-cccc-4ccc-8ccc-cccccccccccc")
+
+
+@pytest.fixture
+def api_settings(settings: Settings, qdrant_settings: QdrantSettings) -> Settings:
+    return settings.model_copy(update={"qdrant": qdrant_settings})
+
+
+@pytest_asyncio.fixture(loop_scope="session")
+async def api_client(
+    api_settings: Settings, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> AsyncIterator[AsyncClient]:
+    app = create_app(api_settings)
+    app.dependency_overrides[get_sessionmaker] = lambda: db_sessionmaker
+    async with (
+        LifespanManager(app) as manager,
+        AsyncClient(transport=ASGITransport(app=manager.app), base_url="http://test") as client,
+    ):
+        yield client
+
+
+def _auth(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+async def _tenant_with_key(
+    db_session: AsyncSession, *, scopes: list[str] | None = None
+) -> tuple[uuid.UUID, str]:
+    tenant = await create_tenant(db_session)
+    _, token = await create_api_key(
+        db_session,
+        tenant=tenant,
+        scopes=scopes if scopes is not None else ["points:read", "points:write"],
+    )
+    await db_session.commit()
+    return tenant.id, token
+
+
+async def _seed_chain(
+    qdrant_client: AsyncQdrantClient,
+    qdrant_settings: QdrantSettings,
+    *,
+    tenant_id: uuid.UUID,
+    file_id: uuid.UUID = FILE_ID,
+    length: int = 3,
+) -> list[uuid.UUID]:
+    """A linked run of points, written the way ingestion writes them."""
+    await ensure_chunks_collection(qdrant_client, collection=qdrant_settings.collection)
+    ids = [point_id_for(file_id, index) for index in range(length)]
+    points = []
+    for index in range(length):
+        chunk_point = chunk_point_for(
+            SeedSpec(tenant_id, file_id, index, f"chunk {index}", float(index + 1))
+        )
+        chunk_point.payload["previous_chunk_id"] = str(ids[index - 1]) if index else None
+        chunk_point.payload["next_chunk_id"] = str(ids[index + 1]) if index + 1 < length else None
+        points.append(chunk_point)
+
+    storage = QdrantPointStorage(qdrant_client, collection=qdrant_settings.collection)
+    await storage.upsert_points(points)
+    return ids
+
+
+async def _read_point(
+    api_client: AsyncClient, token: str, point_id: uuid.UUID
+) -> dict[str, object]:
+    response = await api_client.get(f"/v1/points/{point_id}", headers=_auth(token))
+    assert response.status_code == 200
+    payload: dict[str, object] = response.json()
+    return payload
+
+
+async def test_delete_point_deactivates_it_and_relinks_its_neighbours(
+    api_client: AsyncClient,
+    db_session: AsyncSession,
+    qdrant_client: AsyncQdrantClient,
+    qdrant_settings: QdrantSettings,
+) -> None:
+    tenant_id, token = await _tenant_with_key(db_session)
+    first, middle, last = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant_id)
+
+    response = await api_client.delete(f"/v1/points/{middle}", headers=_auth(token))
+
+    assert response.status_code == 200
+    assert response.json()["is_active"] is False
+    assert response.json()["deleted_at"] is not None
+    assert (await _read_point(api_client, token, first))["next_chunk_id"] == str(last)
+    assert (await _read_point(api_client, token, last))["previous_chunk_id"] == str(first)
+
+
+async def test_delete_point_keeps_the_point_in_qdrant(
+    api_client: AsyncClient,
+    db_session: AsyncSession,
+    qdrant_client: AsyncQdrantClient,
+    qdrant_settings: QdrantSettings,
+) -> None:
+    """Soft delete means soft: the point is still there, just not listed."""
+    tenant_id, token = await _tenant_with_key(db_session)
+    _, middle, _ = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant_id)
+
+    await api_client.delete(f"/v1/points/{middle}", headers=_auth(token))
+
+    listed = await api_client.get(
+        "/v1/points", params={"file_id": str(FILE_ID)}, headers=_auth(token)
+    )
+    assert str(middle) not in [point["point_id"] for point in listed.json()["points"]]
+    assert (await _read_point(api_client, token, middle))["is_active"] is False
+
+
+async def test_delete_point_is_a_noop_success_the_second_time(
+    api_client: AsyncClient,
+    db_session: AsyncSession,
+    qdrant_client: AsyncQdrantClient,
+    qdrant_settings: QdrantSettings,
+) -> None:
+    tenant_id, token = await _tenant_with_key(db_session)
+    first, middle, _ = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant_id)
+    await api_client.delete(f"/v1/points/{middle}", headers=_auth(token))
+    version_after_first = (await _read_point(api_client, token, first))["version"]
+
+    again = await api_client.delete(f"/v1/points/{middle}", headers=_auth(token))
+
+    assert again.status_code == 200
+    assert again.json()["is_active"] is False
+    assert (await _read_point(api_client, token, first))["version"] == version_after_first
+
+
+async def test_delete_point_returns_404_for_another_tenants_point(
+    api_client: AsyncClient,
+    db_session: AsyncSession,
+    qdrant_client: AsyncQdrantClient,
+    qdrant_settings: QdrantSettings,
+) -> None:
+    """`404`, not `403`, and nothing is deactivated on the way to saying so."""
+    owner_id, owner_token = await _tenant_with_key(db_session)
+    _, intruder_token = await _tenant_with_key(db_session)
+    _, middle, _ = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=owner_id)
+
+    response = await api_client.delete(f"/v1/points/{middle}", headers=_auth(intruder_token))
+
+    assert response.status_code == 404
+    assert response.json()["error"]["code"] == "not_found"
+    assert (await _read_point(api_client, owner_token, middle))["is_active"] is True
+
+
+async def test_delete_point_requires_the_points_write_scope(
+    api_client: AsyncClient,
+    db_session: AsyncSession,
+    qdrant_client: AsyncQdrantClient,
+    qdrant_settings: QdrantSettings,
+) -> None:
+    """A read-only key can see a point but must not be able to remove it."""
+    tenant_id, token = await _tenant_with_key(db_session, scopes=["points:read"])
+    _, middle, _ = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant_id)
+
+    response = await api_client.delete(f"/v1/points/{middle}", headers=_auth(token))
+
+    assert response.status_code == 403
+    assert (await _read_point(api_client, token, middle))["is_active"] is True
+
+
+async def test_delete_file_deactivates_every_point_and_retires_the_row(
+    api_client: AsyncClient,
+    db_session: AsyncSession,
+    db_sessionmaker: async_sessionmaker[AsyncSession],
+    qdrant_client: AsyncQdrantClient,
+    qdrant_settings: QdrantSettings,
+) -> None:
+    tenant = await create_tenant(db_session)
+    _, token = await create_api_key(
+        db_session, tenant=tenant, scopes=["points:read", "points:write"]
+    )
+    file_id = uuid.uuid4()
+    await create_source_file(db_session, tenant=tenant, source_file_id=file_id)
+    await db_session.commit()
+    await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant.id, file_id=file_id)
+
+    response = await api_client.delete(f"/v1/files/{file_id}", headers=_auth(token))
+
+    assert response.status_code == 200
+    assert response.json()["points_soft_deleted"] == 3
+    listed = await api_client.get(
+        "/v1/points", params={"file_id": str(file_id)}, headers=_auth(token)
+    )
+    assert listed.json()["points"] == []
+
+    async with db_sessionmaker() as session:
+        row = await source_files_repo.get_by_id(
+            session, tenant_id=tenant.id, source_file_id=file_id
+        )
+    assert row is not None
+    assert row.status == "soft_deleted"
+    assert row.deleted_at is not None
+
+
+async def test_delete_file_returns_404_for_another_tenants_file(
+    api_client: AsyncClient,
+    db_session: AsyncSession,
+    qdrant_client: AsyncQdrantClient,
+    qdrant_settings: QdrantSettings,
+) -> None:
+    """The authorization check runs before the sweep, so nothing is deactivated."""
+    owner = await create_tenant(db_session)
+    _, owner_token = await create_api_key(
+        db_session, tenant=owner, scopes=["points:read", "points:write"]
+    )
+    intruder = await create_tenant(db_session)
+    _, intruder_token = await create_api_key(
+        db_session, tenant=intruder, scopes=["points:read", "points:write"]
+    )
+    file_id = uuid.uuid4()
+    await create_source_file(db_session, tenant=owner, source_file_id=file_id)
+    await db_session.commit()
+    await _seed_chain(qdrant_client, qdrant_settings, tenant_id=owner.id, file_id=file_id)
+
+    response = await api_client.delete(f"/v1/files/{file_id}", headers=_auth(intruder_token))
+
+    assert response.status_code == 404
+    assert response.json()["error"]["code"] == "not_found"
+    listed = await api_client.get(
+        "/v1/points", params={"file_id": str(file_id)}, headers=_auth(owner_token)
+    )
+    assert len(listed.json()["points"]) == 3
+
+
+async def test_delete_file_requires_the_points_write_scope(
+    api_client: AsyncClient, db_session: AsyncSession
+) -> None:
+    tenant = await create_tenant(db_session)
+    _, token = await create_api_key(db_session, tenant=tenant, scopes=["files:write"])
+    file_id = uuid.uuid4()
+    await create_source_file(db_session, tenant=tenant, source_file_id=file_id)
+    await db_session.commit()
+
+    response = await api_client.delete(f"/v1/files/{file_id}", headers=_auth(token))
+
+    assert response.status_code == 403

tests/support/factories.py

@@ -11,6 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from src.application.auth.keys import generate_api_key, hash_secret
 from src.infrastructure.postgres.models.api_key import ApiKey
+from src.infrastructure.postgres.models.source_file import SourceFile
 from src.infrastructure.postgres.models.tenant import Tenant
 from src.infrastructure.postgres.models.tenant_domain import TenantDomain
 
@@ -50,6 +51,36 @@ async def create_api_key(
     return api_key, full_key
 
 
+async def create_source_file(
+    session: AsyncSession,
+    *,
+    tenant: Tenant,
+    source_file_id: uuid.UUID | None = None,
+    domain: str = "general",
+    status: str = "active",
+) -> SourceFile:
+    """A `source_files` row for tests that address a file without uploading one.
+
+    `DELETE /v1/files/{file_id}` authorizes against this row before touching a
+    single point, so a delete test needs it even though the interesting state
+    lives in Qdrant.
+    """
+    source_file = SourceFile(
+        id=source_file_id or uuid.uuid4(),
+        tenant_id=tenant.id,
+        domain=domain,
+        source_filename="handbook.docx",
+        source_type="docx",
+        content_sha256="0" * 64,
+        byte_size=1024,
+        storage_uri="s3://bucket/key",
+        status=status,
+    )
+    session.add(source_file)
+    await session.flush()
+    return source_file
+
+
 async def create_tenant_domain(
     session: AsyncSession,
     *,

tests/support/point_contract.py

@@ -329,6 +329,51 @@ async def scenario_patch_cannot_cross_tenants(
     assert victim.content == "onboarding checklist for new staff"
 
 
+async def scenario_patch_batch_applies_every_patch_in_one_call(
+    repository: PointRepository, *, tenant_a: uuid.UUID, file_a: uuid.UUID
+) -> None:
+    """One `apply_patches` call carrying several points applies all of them.
+
+    This is the primitive soft delete is built on: a deactivation plus its two
+    neighbour relinks go out together, and a batch that silently applied only
+    its first operation would leave the pointer chain broken — the exact defect
+    ADR-0002 rules out. Nulling a pointer is included because `None` and
+    "absent" are different payload values, and only one of them clears a link.
+    """
+    deleted, previous, following = (point_id_for(file_a, index) for index in (1, 0, 2))
+    await repository.apply_patches(
+        tenant_id=tenant_a,
+        patches=[
+            PayloadPatch(
+                point_id=deleted,
+                payload={"is_active": False, "version": 2},
+                expected_version=1,
+            ),
+            PayloadPatch(
+                point_id=previous,
+                payload={"next_chunk_id": str(following), "version": 2},
+                expected_version=1,
+            ),
+            PayloadPatch(
+                point_id=following,
+                payload={"previous_chunk_id": None, "version": 2},
+                expected_version=1,
+            ),
+        ],
+    )
+
+    applied = {
+        point.point_id: point
+        for point in await repository.get_many(
+            tenant_id=tenant_a, point_ids=[deleted, previous, following]
+        )
+    }
+    assert applied[deleted].is_active is False
+    assert applied[previous].next_chunk_id == following
+    assert applied[following].previous_chunk_id is None
+    assert [applied[point_id].version for point_id in (deleted, previous, following)] == [2, 2, 2]
+
+
 # Scenarios that only read, safe to run in any order against one seeded corpus.
 READ_SCENARIOS = (
     scenario_get_returns_point_within_tenant,
@@ -350,4 +395,5 @@ WRITE_SCENARIOS = (
     scenario_patch_applies_when_version_matches,
     scenario_patch_is_a_noop_when_version_is_stale,
     scenario_patch_cannot_cross_tenants,
+    scenario_patch_batch_applies_every_patch_in_one_call,
 )

tests/unit/application/points/test_deletion.py

@@ -0,0 +1,379 @@
+"""Soft delete and neighbour relinking against the fake repository (ADR-0016).
+
+The point of these tests is the pointer chain, not the HTTP surface. ADR-0002
+treats a partial relink as a defect, and the ways to produce one are all here:
+deleting at either boundary, deleting the same point twice, losing a version
+race half way through a batch, and pointing at a neighbour that is gone.
+
+`tests/integration/qdrant/test_points_deletion.py` runs the two central cases
+against real Qdrant. What only this file can do cheaply is force the races.
+"""
+
+import uuid
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import override
+
+import pytest
+import structlog.testing
+
+from src.application.points.deletion import soft_delete_file_points, soft_delete_point
+from src.application.points.errors import PointVersionConflictError
+from src.application.points.point import Point, PointNotFoundError
+from src.application.ports.point_repository import PayloadPatch
+from tests.fakes import FakePointRepository
+from tests.support.point_contract import SeedSpec, build_point
+
+pytestmark = [pytest.mark.unit, pytest.mark.asyncio]
+
+TENANT = uuid.UUID("11111111-1111-4111-8111-111111111111")
+OTHER_TENANT = uuid.UUID("22222222-2222-4222-8222-222222222222")
+FILE = uuid.UUID("33333333-3333-4333-8333-333333333333")
+OTHER_FILE = uuid.UUID("44444444-4444-4444-8444-444444444444")
+ACTOR = "api_key:test"
+
+
+def _chain(count: int, *, tenant: uuid.UUID = TENANT, file_id: uuid.UUID = FILE) -> list[Point]:
+    """`count` points of one file, linked head to tail in `order_id` order."""
+    points = [
+        build_point(SeedSpec(tenant, file_id, index, f"chunk {index}", float(index + 1)))
+        for index in range(count)
+    ]
+    return [
+        point.model_copy(
+            update={
+                "previous_chunk_id": points[index - 1].point_id if index else None,
+                "next_chunk_id": (points[index + 1].point_id if index + 1 < len(points) else None),
+            }
+        )
+        for index, point in enumerate(points)
+    ]
+
+
+def _seeded(*points: Point) -> FakePointRepository:
+    repository = FakePointRepository()
+    for point in points:
+        repository.add(point)
+    return repository
+
+
+def _stored(repository: FakePointRepository, point_id: uuid.UUID) -> Point:
+    return repository.points[str(point_id)]
+
+
+def _walk(repository: FakePointRepository, head: uuid.UUID) -> list[uuid.UUID]:
+    """Follow `next_chunk_id` from `head`, guarding against a cycle."""
+    visited: list[uuid.UUID] = []
+    current: uuid.UUID | None = head
+    while current is not None and len(visited) <= len(repository.points):
+        visited.append(current)
+        current = _stored(repository, current).next_chunk_id
+    return visited
+
+
+async def test_soft_delete_point_relinks_the_neighbours_of_a_middle_point() -> None:
+    first, middle, last = _chain(3)
+    repository = _seeded(first, middle, last)
+
+    deleted = await soft_delete_point(
+        repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR
+    )
+
+    assert deleted.is_active is False
+    assert deleted.deleted_at is not None
+    assert _stored(repository, first.point_id).next_chunk_id == last.point_id
+    assert _stored(repository, last.point_id).previous_chunk_id == first.point_id
+
+
+async def test_soft_delete_point_sends_the_deactivation_and_both_relinks_in_one_batch() -> None:
+    """A partial relink is a defect, so the three patches must not be split up."""
+    first, middle, last = _chain(3)
+    repository = _seeded(first, middle, last)
+
+    await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
+
+    assert len(repository.patch_batches) == 1
+    assert {patch.point_id for patch in repository.patch_batches[0]} == {
+        middle.point_id,
+        first.point_id,
+        last.point_id,
+    }
+
+
+async def test_soft_delete_point_bumps_the_version_of_every_point_it_touches() -> None:
+    """A relinked neighbour really changed, so a stale editor of it must `409`."""
+    first, middle, last = _chain(3)
+    repository = _seeded(first, middle, last)
+
+    await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
+
+    assert [_stored(repository, point.point_id).version for point in (first, middle, last)] == [
+        2,
+        2,
+        2,
+    ]
+    assert _stored(repository, first.point_id).updated_by == ACTOR
+
+
+async def test_soft_delete_point_leaves_the_new_head_without_a_previous_pointer() -> None:
+    first, second, third = _chain(3)
+    repository = _seeded(first, second, third)
+
+    await soft_delete_point(repository, tenant_id=TENANT, point_id=first.point_id, actor=ACTOR)
+
+    assert _stored(repository, second.point_id).previous_chunk_id is None
+
+
+async def test_soft_delete_point_leaves_the_new_tail_without_a_next_pointer() -> None:
+    first, second, third = _chain(3)
+    repository = _seeded(first, second, third)
+
+    await soft_delete_point(repository, tenant_id=TENANT, point_id=third.point_id, actor=ACTOR)
+
+    assert _stored(repository, second.point_id).next_chunk_id is None
+
+
+async def test_soft_delete_point_keeps_the_deleted_points_own_pointers() -> None:
+    """Nothing active points at it any more, so its pointers record where it sat.
+
+    That record is what the retry re-plans from, and what a later restore or an
+    audit reader would need to place the point back in the sequence.
+    """
+    first, middle, last = _chain(3)
+    repository = _seeded(first, middle, last)
+
+    await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
+
+    stored = _stored(repository, middle.point_id)
+    assert stored.previous_chunk_id == first.point_id
+    assert stored.next_chunk_id == last.point_id
+
+
+async def test_soft_delete_point_is_a_noop_for_an_already_inactive_point() -> None:
+    """Not a `404`, and not a second relink — no patch is issued at all."""
+    first, middle, last = _chain(3)
+    repository = _seeded(first, middle, last)
+    await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
+    versions_after_first = {
+        point.point_id: _stored(repository, point.point_id).version
+        for point in (first, middle, last)
+    }
+    repository.patch_batches.clear()
+
+    again = await soft_delete_point(
+        repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR
+    )
+
+    assert again.is_active is False
+    assert repository.patch_batches == []
+    assert {
+        point.point_id: _stored(repository, point.point_id).version
+        for point in (first, middle, last)
+    } == versions_after_first
+
+
+async def test_soft_delete_point_raises_not_found_for_another_tenants_point() -> None:
+    first, middle, last = _chain(3)
+    repository = _seeded(first, middle, last)
+
+    with pytest.raises(PointNotFoundError):
+        await soft_delete_point(
+            repository, tenant_id=OTHER_TENANT, point_id=middle.point_id, actor=ACTOR
+        )
+
+    assert repository.patch_batches == []
+    assert _stored(repository, middle.point_id).is_active is True
+
+
+async def test_soft_delete_point_raises_not_found_for_an_unknown_point() -> None:
+    repository = _seeded(*_chain(2))
+
+    with pytest.raises(PointNotFoundError):
+        await soft_delete_point(repository, tenant_id=TENANT, point_id=uuid.uuid4(), actor=ACTOR)
+
+
+async def test_soft_delete_point_skips_a_neighbour_that_is_not_there() -> None:
+    """A pointer naming an absent point means the chain was already broken.
+
+    The delete completes the half of the relink that exists rather than
+    refusing, which would leave the point unremovable through any endpoint.
+    """
+    first, middle, last = _chain(3)
+    repository = _seeded(first, middle)  # `last` is never stored
+
+    deleted = await soft_delete_point(
+        repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR
+    )
+
+    assert deleted.is_active is False
+    assert _stored(repository, first.point_id).next_chunk_id == last.point_id
+
+
+async def test_soft_delete_point_visits_every_active_point_exactly_once_after_several_deletes() -> (
+    None
+):
+    """The plan's traversal property, over a sequence of deletes.
+
+    Walking `next_chunk_id` from the head must reach every surviving point once
+    and never step into a deactivated one.
+    """
+    points = _chain(5)
+    repository = _seeded(*points)
+
+    for index in (1, 3):
+        await soft_delete_point(
+            repository, tenant_id=TENANT, point_id=points[index].point_id, actor=ACTOR
+        )
+
+    walked = _walk(repository, points[0].point_id)
+    assert walked == [points[0].point_id, points[2].point_id, points[4].point_id]
+    assert all(_stored(repository, point_id).is_active for point_id in walked)
+
+
+@dataclass
+class _ContendedRepository(FakePointRepository):
+    """Bumps one point's version just before an apply, as a rival writer would.
+
+    That makes the patch guarding on the old version match nothing while the
+    rest of the batch lands — Qdrant's real behaviour, and the partial apply the
+    service's retry exists to repair. `rounds` bounds how long the rival keeps
+    interfering.
+    """
+
+    rival: uuid.UUID | None = None
+    rounds: int = 0
+
+    @override
+    async def apply_patches(self, *, tenant_id: uuid.UUID, patches: Sequence[PayloadPatch]) -> None:
+        if self.rounds > 0 and self.rival is not None:
+            self.rounds -= 1
+            victim = self.points[str(self.rival)]
+            self.points[str(self.rival)] = victim.model_copy(update={"version": victim.version + 1})
+        await super().apply_patches(tenant_id=tenant_id, patches=patches)
+
+
+async def test_soft_delete_point_repairs_a_partially_applied_batch_on_retry() -> None:
+    first, middle, last = _chain(3)
+    repository = _ContendedRepository(rival=first.point_id, rounds=1)
+    for point in (first, middle, last):
+        repository.add(point)
+
+    await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
+
+    # The first batch left `first` stale; the second re-planned only what was
+    # still missing, rather than re-deactivating the already-inactive point.
+    assert len(repository.patch_batches) == 2
+    assert [patch.point_id for patch in repository.patch_batches[1]] == [first.point_id]
+    assert _stored(repository, first.point_id).next_chunk_id == last.point_id
+    assert _stored(repository, middle.point_id).is_active is False
+
+
+async def test_soft_delete_point_conflicts_when_the_relink_never_settles() -> None:
+    first, middle, last = _chain(3)
+    repository = _ContendedRepository(rival=first.point_id, rounds=99)
+    for point in (first, middle, last):
+        repository.add(point)
+
+    with pytest.raises(PointVersionConflictError):
+        await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
+
+
+async def test_soft_delete_file_points_deactivates_every_active_point_of_the_file() -> None:
+    points = _chain(3)
+    other_file = _chain(2, file_id=OTHER_FILE)
+    other_tenant = _chain(1, tenant=OTHER_TENANT, file_id=uuid.uuid4())
+    repository = _seeded(*points, *other_file, *other_tenant)
+
+    swept = await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)
+
+    assert swept == 3
+    assert all(not _stored(repository, point.point_id).is_active for point in points)
+    assert all(_stored(repository, point.point_id).is_active for point in other_file)
+    assert all(_stored(repository, point.point_id).is_active for point in other_tenant)
+
+
+async def test_soft_delete_file_points_leaves_the_chain_intact() -> None:
+    """No survivor can dangle, so the sweep rewrites no pointer at all."""
+    first, middle, last = _chain(3)
+    repository = _seeded(first, middle, last)
+
+    await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)
+
+    assert _stored(repository, first.point_id).next_chunk_id == middle.point_id
+    assert _stored(repository, middle.point_id).previous_chunk_id == first.point_id
+    assert _stored(repository, last.point_id).previous_chunk_id == middle.point_id
+
+
+async def test_soft_delete_file_points_pages_past_one_batch() -> None:
+    """More points than one sweep page, so the re-listing loop has to run."""
+    points = _chain(230)
+    repository = _seeded(*points)
+
+    swept = await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)
+
+    assert swept == 230
+    assert all(not _stored(repository, point.point_id).is_active for point in points)
+
+
+async def test_soft_delete_file_points_is_a_noop_the_second_time() -> None:
+    repository = _seeded(*_chain(3))
+    await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)
+    repository.patch_batches.clear()
+
+    swept = await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)
+
+    assert swept == 0
+    assert repository.patch_batches == []
+
+
+async def test_soft_delete_file_points_returns_zero_for_an_unknown_file() -> None:
+    repository = _seeded(*_chain(2))
+
+    assert (
+        await soft_delete_file_points(
+            repository, tenant_id=TENANT, file_id=uuid.uuid4(), actor=ACTOR
+        )
+        == 0
+    )
+
+
+async def test_soft_delete_point_logs_its_duration_and_round_count() -> None:
+    """ADR-0011's `duration_ms`, plus the field that explains a slow one.
+
+    Relinking is O(1), so a delete's cost is Qdrant round trips; `rounds` above
+    1 means a concurrent writer forced a re-plan rather than the store being
+    slow, and the two fields are only useful together.
+    """
+    first, middle, last = _chain(3)
+    repository = _seeded(first, middle, last)
+
+    with structlog.testing.capture_logs() as logs:
+        await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
+
+    event = next(entry for entry in logs if entry["event"] == "points.soft_deleted")
+    assert event["rounds"] == 1
+    assert isinstance(event["duration_ms"], float)
+
+
+async def test_soft_delete_file_points_logs_a_round_per_sweep_page() -> None:
+    """The sweep is the delete path whose cost tracks the size of the file."""
+    repository = _seeded(*_chain(230))
+
+    with structlog.testing.capture_logs() as logs:
+        await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)
+
+    event = next(entry for entry in logs if entry["event"] == "points.file_soft_deleted")
+    assert event["points_soft_deleted"] == 230
+    assert event["rounds"] == 3
+    assert isinstance(event["duration_ms"], float)
+
+
+async def test_soft_delete_file_points_conflicts_when_a_round_makes_no_progress() -> None:
+    """A sweep round that attempts the same ids as the one before is stuck."""
+    points = _chain(3)
+    repository = _ContendedRepository(rival=points[0].point_id, rounds=99)
+    for point in points:
+        repository.add(point)
+
+    with pytest.raises(PointVersionConflictError):
+        await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)