feat(ingestion): index embedded chunks into Qdrant on upload

Why:
- POST /v1/files was reporting chunks_indexed=0/points_created=0 unconditionally
  — chunks were parsed and embedded but never written to Qdrant, so nothing
  was actually searchable after upload.

Changes:
- upload_source_file() now calls index_chunks() after embedding, inside the
  same INGESTION_TIMEOUT_SECONDS window, and marks the job failed
  (error_code=index_failed, 502) if it raises.
- Job counters (points_created, points_soft_deleted) and the response's
  chunks_indexed now reflect the real indexing result instead of a hardcoded
  zero.
- Wired PointStorage through AppResources/lifespan/the files router.

Impact:
- A successful upload is now searchable in Qdrant by the time 201 returns.

CLAUDE.md

@@ -18,10 +18,14 @@ the chunk-count ceiling (`413`). The embedding configuration is **ported from
 the `emet` evaluation lab** (`~/code/talie/emet`), which benchmarked these
 models and analyzers on the real Farsi corpus — the analyzer and BM25 weights
 are verified token-for-token against it, so treat them as a measured artifact
-and re-benchmark rather than tune them in place (ADR-0005). Not built yet: Qdrant collection bootstrap,
+and re-benchmark rather than tune them in place (ADR-0005). Also working: the
-Qdrant point upserts (so uploaded chunks are parsed/embedded but not yet
+`chunks` collection bootstrap (`src/infrastructure/qdrant/collection.py`, run as
-searchable), and `src/agent/`. That maps to plan 001 Phases 1-4 done, Phase 5
+a deployment step via `uv run python -m src.cli.qdrant_bootstrap` — never at
-not started.
+startup) and tenant-scoped point upserts (`src/application/points/` behind the
+`PointStorage` port), so an upload is searchable by the time `201` returns. Not
+built yet: `/v1/points` CRUD and keyword search (plan 002), `tenant_domains`
+validation of the `domain` field, and `src/agent/`. That maps to plan 001
+Phases 1-5 done.
 
 Architecture decisions live in `docs/adr/` (18 ADRs plus the 0000 template;
 0001–0004 are `Accepted` — 0004 amended by 0018; 0014 is `Superseded by 0017`;
@@ -140,9 +144,13 @@ as its only caller-facing entry point. It dispatches on source type and owns
 the `anyio.to_thread.run_sync` + `CapacityLimiter` offload ADR-0017 requires;
 `parse_docx`/`parse_csv`/`parse_xlsx`/`chunk_document` stay in the package,
 exported mainly for their own tests, not for outside callers to reach for
-directly. Follow this pattern in `application/` as new packages are added
+directly. `src/application/points/` follows the same shape: `index_chunks` is
-there — `points/`, `retrieval/`, `threads/` — rather than exposing their
+the only caller-facing entry point, owning payload construction, batching,
-internals as the primary surface.
+the `upsert_concurrency` semaphore, and the ordering rule that the soft-delete
+sweep runs only after every upsert succeeds; `build_chunk_payload` stays
+internal. Follow this pattern in `application/` as new packages are added
+there — `retrieval/`, `threads/` — rather than exposing their internals as the
+primary surface.
 
 ### Resource lifetime rules (ADR-0012)
 

README.md

@@ -3,6 +3,23 @@
 Architecture decisions live in [`docs/adr`](docs/adr). The first implementation
 milestone is documented in the [ingestion vertical-slice plan](docs/plans/001-ingestion-vertical-slice.md).
 
+## Provisioning the datastores
+
+Both schema steps run as explicit deployment steps. The application performs no
+DDL at startup — not for Postgres (ADR-0009) and not for Qdrant (ADR-0001,
+"Collection provisioning").
+
+```bash
+docker compose up -d                          # Postgres, MinIO, Qdrant
+uv run alembic upgrade head                   # Postgres schema
+uv run python -m src.cli.qdrant_bootstrap     # the `chunks` collection
+uv run fastapi dev src/main.py
+```
+
+Both commands are idempotent and safe to re-run. `qdrant_bootstrap` verifies an
+existing collection against the pinned schema and exits non-zero on a mismatch,
+rather than leaving a silently degraded sparse index in place.
+
 ## Local Langfuse
 
 This repo includes a root-level development Compose file for Langfuse:

docs/adr/0017-synchronous-ingestion-in-the-request-path.md

@@ -159,8 +159,25 @@ retry, and phase 2 has no transaction protecting it:
   return the existing file/job rather than re-ingesting (plan 001).
 - `tenant_id` comes from `AuthContext`, never from the request body.
 - A terminal job is never transitioned back to `running`.
-- Qdrant points from a failed attempt do not replace the previous successful
+- A failed attempt never *removes* content from a working index. The
-  index; replacement happens only after a successful attempt.
+  soft-delete sweep that retires a shortened file's leftover points runs only
+  after every upsert in the attempt has succeeded.
+
+  This is deliberately weaker than "replacement happens only after a successful
+  attempt", which an earlier revision of this ADR claimed. That guarantee is not
+  achievable alongside ADR-0001's deterministic point ids: those ids are exactly
+  what makes a retry idempotent, and they also mean a re-ingestion overwrites
+  points **in place**, so a crash partway through leaves a prefix updated and the
+  remainder still on the old content. Buying literal atomicity would mean
+  generation-suffixed ids and an activation flip, which contradicts ADR-0001 and
+  ADR-0002's stable point ids. Staging the new points as `is_active=false` and
+  flipping them on success is strictly worse — the in-place overwrite would
+  deactivate the previously live points, silently emptying a working index if the
+  attempt were interrupted.
+
+  What holds instead: the index is never emptied, never partially deleted, and a
+  retry converges — deterministic ids rewrite every point and the sweep re-runs,
+  reaching the exact correct state.
 
 ### Failures are HTTP failures
 

docs/plans/001-ingestion-vertical-slice.md

@@ -125,9 +125,12 @@ them:
 - Use `(tenant_id, domain, content_sha256)` to recognize identical uploads.
 - An identical active upload should return the existing source-file/job reference
   rather than create a duplicate ingestion.
-- A changed upload creates a new ingestion job. Existing active Qdrant points are
+- A changed upload creates a new ingestion job. A failed re-ingestion never
-  replaced only after the new job completes successfully, so a failed re-ingestion
+  removes a working index: the soft-delete sweep for a shortened file runs only
-  does not remove a working index.
+  after every upsert has succeeded. Because ADR-0001's point ids are
+  deterministic, upserts overwrite in place, so an interrupted attempt can leave
+  a prefix updated — it cannot empty or partially delete the index, and a retry
+  converges. See ADR-0017, "Re-running an ingestion stays safe".
 - Preserve the original filename in Postgres metadata. MinIO object keys remain
   internal ID-based paths.
 
@@ -272,6 +275,11 @@ code and a terminal job row.
    tenant-filtered upserts, terminal state persistence, retrying an upload, and
    parser/Qdrant failure handling.
 
+The `chunks` collection itself is provisioned by a deployment step —
+`uv run python -m src.cli.qdrant_bootstrap` — not by FastAPI startup, for the
+same reason ADR-0009 keeps Alembic out of startup and ADR-0012 makes LangGraph's
+`.setup()` a deployment step. See ADR-0001, "Collection provisioning".
+
 **Exit criteria:** a successful upload returns `201` with a terminal status, and
 its points are retrievable only under the owning tenant's Qdrant filter. A forced
 failure mid-ingestion produces a `failed` job and the right HTTP status, and

src/api/errors.py

@@ -23,6 +23,7 @@ from src.application.ingestion.errors import (
     EmbedderError,
     IngestionAtCapacityError,
     IngestionTimeoutError,
+    PointIndexingError,
     UnsupportedSourceTypeError,
 )
 
@@ -44,6 +45,7 @@ _MAPPING: tuple[tuple[type[Exception], int, str], ...] = (
     (FileTooLargeError, status.HTTP_413_CONTENT_TOO_LARGE, "payload_too_large"),
     (ChunkLimitExceededError, status.HTTP_413_CONTENT_TOO_LARGE, "payload_too_large"),
     (EmbedderError, status.HTTP_502_BAD_GATEWAY, "embedder_error"),
+    (PointIndexingError, status.HTTP_502_BAD_GATEWAY, "index_error"),
     (IngestionTimeoutError, status.HTTP_504_GATEWAY_TIMEOUT, "ingestion_timeout"),
 )
 

src/api/routers/files.py

@@ -19,11 +19,13 @@ from src.application.files.status import get_file_status
 from src.application.files.upload import upload_source_file
 from src.application.ports.embedding import DenseEmbedder, SparseEmbedder
 from src.application.ports.object_storage import ObjectStorage
+from src.application.ports.point_storage import PointStorage
 from src.bootstrap.dependencies import (
     get_dense_embedders,
     get_ingestion_concurrency_limiter,
     get_ingestion_limiter,
     get_object_storage,
+    get_point_storage,
     get_sessionmaker,
     get_settings,
     get_sparse_embedder,
@@ -35,6 +37,7 @@ router = APIRouter(prefix="/files", tags=["files"])
 _RequireFilesWrite = Annotated[AuthContext, Depends(require_scope("files:write"))]
 _SessionmakerDep = Annotated[async_sessionmaker[AsyncSession], Depends(get_sessionmaker)]
 _ObjectStorageDep = Annotated[ObjectStorage, Depends(get_object_storage)]
+_PointStorageDep = Annotated[PointStorage, Depends(get_point_storage)]
 _SettingsDep = Annotated[Settings, Depends(get_settings)]
 _IngestionLimiterDep = Annotated[CapacityLimiter, Depends(get_ingestion_limiter)]
 _ConcurrencyLimiterDep = Annotated[Semaphore, Depends(get_ingestion_concurrency_limiter)]
@@ -50,6 +53,7 @@ async def upload_file(
     auth: _RequireFilesWrite,
     sessionmaker: _SessionmakerDep,
     storage: _ObjectStorageDep,
+    point_storage: _PointStorageDep,
     settings: _SettingsDep,
     limiter: _IngestionLimiterDep,
     concurrency_limiter: _ConcurrencyLimiterDep,
@@ -60,12 +64,14 @@ async def upload_file(
     result = await upload_source_file(
         sessionmaker=sessionmaker,
         storage=storage,
+        point_storage=point_storage,
         auth=auth,
         domain=domain,
         filename=file.filename or "",
         data=data,
         ingestion_settings=settings.ingestion,
         chunking_settings=settings.chunking,
+        qdrant_settings=settings.qdrant,
         thread_limiter=limiter,
         concurrency_limiter=concurrency_limiter,
         dense_embedders=dense_embedders,

src/application/files/upload.py

@@ -16,9 +16,10 @@ job stuck in `running`. The whole request additionally holds one of
 phase 2 is bounded by `INGESTION_TIMEOUT_SECONDS` (`504`) (ADR-0017, plan 001
 Phase 4).
 
-Qdrant point upserts are Phase 5 work, not implemented here: this phase
+Phase 2 ends by upserting the embedded chunks as tenant-scoped Qdrant points
-parses, chunks, and embeds, so a successful job still reports
+(`src/application/points/`), so a successful upload is searchable by the time
-`chunks_indexed=0` — nothing is searchable yet.
+the `201` returns. The collection those points land in is provisioned by a
+deployment step, not by this path — see `src/cli/qdrant_bootstrap.py`.
 """
 
 import uuid
@@ -45,10 +46,13 @@ from src.application.ingestion.errors import (
     ChunkLimitExceededError,
     EmbedderError,
     IngestionTimeoutError,
+    PointIndexingError,
 )
+from src.application.points import index_chunks
 from src.application.ports.embedding import DenseEmbedder, SparseEmbedder
 from src.application.ports.object_storage import ObjectStorage
-from src.config import ChunkingSettings, IngestionSettings
+from src.application.ports.point_storage import PointStorage
+from src.config import ChunkingSettings, IngestionSettings, QdrantSettings
 from src.infrastructure.postgres.repositories import ingestion_jobs as jobs_repo
 from src.infrastructure.postgres.repositories import source_files as source_files_repo
 
@@ -88,12 +92,14 @@ async def upload_source_file(
     *,
     sessionmaker: async_sessionmaker[AsyncSession],
     storage: ObjectStorage,
+    point_storage: PointStorage,
     auth: AuthContext,
     domain: str,
     filename: str,
     data: bytes,
     ingestion_settings: IngestionSettings,
     chunking_settings: ChunkingSettings,
+    qdrant_settings: QdrantSettings,
     thread_limiter: CapacityLimiter,
     concurrency_limiter: Semaphore,
     dense_embedders: Sequence[DenseEmbedder],
@@ -249,6 +255,31 @@ async def upload_source_file(
                         error_message=str(exc),
                     )
                     raise
+
+                try:
+                    indexed = await index_chunks(
+                        embedded,
+                        storage=point_storage,
+                        tenant_id=auth.tenant_id,
+                        domain=domain,
+                        file_id=source_file_id,
+                        source_filename=filename,
+                        source_type=validated.source_type,
+                        actor=f"api_key:{auth.api_key_id}",
+                        dense_embedders=dense_embedders,
+                        sparse_embedder=sparse_embedder,
+                        settings=qdrant_settings,
+                        thread_limiter=thread_limiter,
+                    )
+                except PointIndexingError as exc:
+                    await _mark_job_failed(
+                        sessionmaker,
+                        tenant_id=auth.tenant_id,
+                        ingestion_job_id=ingestion_job_id,
+                        error_code="index_failed",
+                        error_message=str(exc),
+                    )
+                    raise
         except TimeoutError:
             logger.warning(
                 "files.upload.timeout",
@@ -273,7 +304,11 @@ async def upload_source_file(
                 tenant_id=auth.tenant_id,
                 ingestion_job_id=ingestion_job_id,
                 status="succeeded",
-                points_created=0,
+                # An upsert with deterministic ids cannot tell an insert from
+                # an overwrite, so every written point is reported here and
+                # `points_updated` stays 0 rather than being guessed at.
+                points_created=indexed.points_upserted,
+                points_soft_deleted=indexed.points_soft_deleted,
             )
             jobs_repo.append_event(
                 session,
@@ -281,8 +316,13 @@ async def upload_source_file(
                 ingestion_job_id=ingestion_job_id,
                 level="info",
                 stage="completed",
-                message="chunks parsed and embedded; Qdrant indexing not yet implemented",
+                message="chunks parsed, embedded, and indexed",
-                details={"chunks_parsed": len(chunks), "chunks_embedded": len(embedded)},
+                details={
+                    "chunks_parsed": len(chunks),
+                    "chunks_embedded": len(embedded),
+                    "points_upserted": indexed.points_upserted,
+                    "points_soft_deleted": indexed.points_soft_deleted,
+                },
             )
             await session.commit()
 
@@ -291,11 +331,12 @@ async def upload_source_file(
         tenant_id=str(auth.tenant_id),
         file_id=str(source_file_id),
         ingestion_job_id=str(ingestion_job_id),
+        points_indexed=indexed.points_upserted,
     )
     return UploadResult(
         file_id=source_file_id,
         ingestion_job_id=ingestion_job_id,
         status="succeeded",
-        chunks_indexed=0,
+        chunks_indexed=indexed.points_upserted,
         is_new_attempt=True,
     )

src/application/ingestion/errors.py

@@ -49,6 +49,15 @@ class EmbedderError(IngestionError):
     """
 
 
+class PointIndexingError(IngestionError):
+    """Upserting or soft-deleting Qdrant points failed.
+
+    Maps to `502` — like `EmbedderError`, this is an upstream dependency
+    failing, not a malformed request. Kept distinct from `EmbedderError` so the
+    job's `error_code` says which dependency broke.
+    """
+
+
 class IngestionAtCapacityError(IngestionError):
     """`INGESTION_MAX_CONCURRENCY` in-process ingestions are already running.
 

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_storage import PointStorage
 from src.config import Settings
 
 
@@ -20,6 +21,7 @@ class AppResources:
     minio_client: Minio
     qdrant_client: AsyncQdrantClient
     object_storage: ObjectStorage
+    point_storage: PointStorage
     ingestion_limiter: CapacityLimiter
     dense_embedders: Sequence[DenseEmbedder]
     sparse_embedder: SparseEmbedder
@@ -46,6 +48,10 @@ def get_object_storage(request: Request) -> ObjectStorage:
     return _resources(request).object_storage
 
 
+def get_point_storage(request: Request) -> PointStorage:
+    return _resources(request).point_storage
+
+
 def get_ingestion_limiter(request: Request) -> CapacityLimiter:
     return _resources(request).ingestion_limiter
 

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.points import QdrantPointStorage
 
 logger = structlog.get_logger(__name__)
 
@@ -77,6 +78,13 @@ def create_lifespan(
         logger.info("lifespan.minio.client.created")
 
         qdrant_client = create_qdrant_client(resolved_settings.qdrant)
+        # No collection DDL here: `ensure_chunks_collection` is a deployment
+        # step (`python -m src.cli.qdrant_bootstrap`), for the same reason
+        # ADR-0009 keeps Alembic out of startup and ADR-0012 makes LangGraph's
+        # `.setup()` a deployment step.
+        point_storage = QdrantPointStorage(
+            qdrant_client, collection=resolved_settings.qdrant.collection
+        )
         logger.info("lifespan.qdrant.client.created")
 
         nomic_settings = resolved_settings.embedding.nomic
@@ -136,6 +144,7 @@ def create_lifespan(
             minio_client=minio_client,
             qdrant_client=qdrant_client,
             object_storage=object_storage,
+            point_storage=point_storage,
             ingestion_limiter=ingestion_limiter,
             dense_embedders=dense_embedders,
             sparse_embedder=sparse_embedder,

tests/integration/postgres/test_upload_service.py

@@ -13,11 +13,21 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 from src.application.auth.context import AuthContext
 from src.application.files.models import UploadResult
 from src.application.files.upload import upload_source_file
-from src.application.ingestion.errors import IngestionAtCapacityError, IngestionTimeoutError
+from src.application.ingestion.errors import (
+    IngestionAtCapacityError,
+    IngestionTimeoutError,
+    PointIndexingError,
+)
 from src.application.ports.object_storage import ObjectStorage
-from src.config import ChunkingSettings, IngestionSettings
+from src.application.ports.point_storage import PointStorage
+from src.config import ChunkingSettings, IngestionSettings, QdrantSettings
 from src.infrastructure.postgres.models.ingestion_job import IngestionJob
-from tests.fakes import FakeDenseEmbedder, FakeObjectStorage, FakeSparseEmbedder
+from tests.fakes import (
+    FakeDenseEmbedder,
+    FakeObjectStorage,
+    FakePointStorage,
+    FakeSparseEmbedder,
+)
 from tests.support.factories import create_api_key, create_tenant
 
 pytestmark = [
@@ -34,6 +44,7 @@ async def _upload(
     sessionmaker: async_sessionmaker[AsyncSession],
     storage: ObjectStorage,
     auth: AuthContext,
+    point_storage: PointStorage | None = None,
     domain: str = "general",
     filename: str = "report.csv",
     data: bytes = _CSV_BYTES,
@@ -41,12 +52,14 @@ async def _upload(
     return await upload_source_file(
         sessionmaker=sessionmaker,
         storage=storage,
+        point_storage=point_storage if point_storage is not None else FakePointStorage(),
         auth=auth,
         domain=domain,
         filename=filename,
         data=data,
         ingestion_settings=IngestionSettings(),
         chunking_settings=ChunkingSettings(),
+        qdrant_settings=QdrantSettings(),
         thread_limiter=CapacityLimiter(2),
         concurrency_limiter=Semaphore(2),
         dense_embedders=[
@@ -83,7 +96,7 @@ async def test_upload_source_file_commits_running_job_before_storage_write(
     result = await _upload(sessionmaker=db_sessionmaker, storage=storage, auth=auth)
 
     assert result.status == "succeeded"
-    assert result.chunks_indexed == 0
+    assert result.chunks_indexed == 1
     assert result.is_new_attempt
 
     async with db_sessionmaker() as verify_session:
@@ -170,12 +183,14 @@ async def test_upload_source_file_timeout_writes_failed_job_and_raises(
         await upload_source_file(
             sessionmaker=db_sessionmaker,
             storage=FakeObjectStorage(),
+            point_storage=FakePointStorage(),
             auth=auth,
             domain="general",
             filename="report.csv",
             data=_CSV_BYTES,
             ingestion_settings=IngestionSettings(timeout_seconds=0.05),
             chunking_settings=ChunkingSettings(),
+            qdrant_settings=QdrantSettings(),
             thread_limiter=CapacityLimiter(2),
             concurrency_limiter=Semaphore(2),
             dense_embedders=[slow_embedder, FakeDenseEmbedder(name="dense_openai")],
@@ -212,12 +227,14 @@ async def test_upload_source_file_at_capacity_rejects_before_any_job_row(
         await upload_source_file(
             sessionmaker=db_sessionmaker,
             storage=FakeObjectStorage(),
+            point_storage=FakePointStorage(),
             auth=auth,
             domain="general",
             filename="report.csv",
             data=_CSV_BYTES,
             ingestion_settings=IngestionSettings(),
             chunking_settings=ChunkingSettings(),
+            qdrant_settings=QdrantSettings(),
             thread_limiter=CapacityLimiter(2),
             concurrency_limiter=concurrency_limiter,
             dense_embedders=[
@@ -238,3 +255,143 @@ async def test_upload_source_file_at_capacity_rejects_before_any_job_row(
             .all()
         )
     assert jobs == []
+
+
+async def test_upload_source_file_indexes_points_and_records_real_counters(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    """The Phase 5 payoff: a successful upload is searchable, and the job row
+    says how many points it wrote rather than a hardcoded zero.
+    """
+    auth = await _auth_for(db_session)
+    point_storage = FakePointStorage()
+
+    result = await _upload(
+        sessionmaker=db_sessionmaker,
+        storage=FakeObjectStorage(),
+        auth=auth,
+        point_storage=point_storage,
+        data=b"name,value\nfirst,1\nsecond,2\nthird,3\n",
+    )
+
+    assert result.status == "succeeded"
+    assert result.chunks_indexed == len(point_storage.points)
+    assert result.chunks_indexed > 0
+
+    async with db_sessionmaker() as verify_session:
+        job = await verify_session.get(IngestionJob, result.ingestion_job_id)
+    assert job is not None
+    assert job.points_created == result.chunks_indexed
+    # An upsert cannot tell an insert from an overwrite, so this stays 0.
+    assert job.points_updated == 0
+
+
+async def test_upload_source_file_indexes_points_under_the_authenticated_tenant(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    auth = await _auth_for(db_session)
+    point_storage = FakePointStorage()
+
+    result = await _upload(
+        sessionmaker=db_sessionmaker,
+        storage=FakeObjectStorage(),
+        auth=auth,
+        point_storage=point_storage,
+        domain="fire",
+    )
+
+    payloads = [point.payload for point in point_storage.points.values()]
+    assert payloads
+    for payload in payloads:
+        assert payload["tenant_id"] == str(auth.tenant_id)
+        assert payload["domain"] == "fire"
+        assert payload["file_id"] == str(result.file_id)
+        assert payload["created_by"] == f"api_key:{auth.api_key_id}"
+
+
+async def test_upload_source_file_index_failure_writes_terminal_failed_job(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    auth = await _auth_for(db_session)
+    point_storage = FakePointStorage(fail_on_batch=0)
+
+    with pytest.raises(PointIndexingError):
+        await _upload(
+            sessionmaker=db_sessionmaker,
+            storage=FakeObjectStorage(),
+            auth=auth,
+            point_storage=point_storage,
+        )
+
+    async with db_sessionmaker() as verify_session:
+        jobs = (await verify_session.execute(select(IngestionJob))).scalars().all()
+    job = next(job for job in jobs if job.tenant_id == auth.tenant_id)
+    assert job.status == "failed"
+    assert job.error_code == "index_failed"
+    assert job.completed_at is not None
+
+
+async def test_upload_source_file_failed_index_does_not_soft_delete_existing_points(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    """A failed attempt must never remove content from a working index."""
+    auth = await _auth_for(db_session)
+    point_storage = FakePointStorage()
+    await _upload(
+        sessionmaker=db_sessionmaker,
+        storage=FakeObjectStorage(),
+        auth=auth,
+        point_storage=point_storage,
+        data=b"name,value\nfirst,1\nsecond,2\n",
+    )
+    # fail_on_batch indexes into upsert_batches, which accumulates across
+    # uploads -- reset it so "batch 0" means the retry's first batch.
+    point_storage.deactivate_calls.clear()
+    point_storage.upsert_batches.clear()
+    point_storage.fail_on_batch = 0
+
+    with pytest.raises(PointIndexingError):
+        await _upload(
+            sessionmaker=db_sessionmaker,
+            storage=FakeObjectStorage(),
+            auth=auth,
+            point_storage=point_storage,
+            data=b"name,value\nonly,1\n",
+        )
+
+    assert point_storage.deactivate_calls == []
+    assert all(point.payload["is_active"] is True for point in point_storage.points.values())
+
+
+async def test_upload_source_file_retry_after_index_failure_produces_no_duplicate_points(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    auth = await _auth_for(db_session)
+    point_storage = FakePointStorage(fail_on_batch=0)
+
+    with pytest.raises(PointIndexingError):
+        await _upload(
+            sessionmaker=db_sessionmaker,
+            storage=FakeObjectStorage(),
+            auth=auth,
+            point_storage=point_storage,
+        )
+
+    point_storage.fail_on_batch = None
+    retry = await _upload(
+        sessionmaker=db_sessionmaker,
+        storage=FakeObjectStorage(),
+        auth=auth,
+        point_storage=point_storage,
+    )
+
+    assert retry.status == "succeeded"
+    assert len(point_storage.points) == retry.chunks_indexed
+
+    async with db_sessionmaker() as verify_session:
+        jobs = (await verify_session.execute(select(IngestionJob))).scalars().all()
+    tenant_jobs = [job for job in jobs if job.tenant_id == auth.tenant_id]
+    # A terminal job never returns to `running` (ADR-0017); the retry is a new row.
+    assert len(tenant_jobs) == 2
+    assert {job.status for job in tenant_jobs} == {"failed", "succeeded"}
+