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

001. Ingestion vertical-slice implementation plan

Purpose

This plan turns the accepted architectural direction in the ADRs into the first working product slice: a tenant-scoped CSV upload is stored in MinIO, represented by durable Postgres records, parsed/chunked/embedded inline in the request (ADR-0017), and indexed as Qdrant points before the response returns.

This is an implementation plan, not an Architecture Decision Record. ADRs explain why major technologies and boundaries were chosen; this document defines the order, scope, and verification criteria for implementing them.

Architecture baseline

The first vertical slice uses these responsibilities:

System	Responsibility
FastAPI	HTTP boundary, validation, authentication, tenant derivation, and job creation.
Postgres	Tenant/auth data, source-file metadata, ingestion attempt records/progress, and audit.
MinIO	Private source-file bytes and retained derived ingestion blobs.
Ingestion service	Parsing, chunking, embedding, ingestion-generated Chunk/Point CRUD, and job status updates — inline in the request.
Qdrant	Tenant-filtered generated chunks and their vectors/payloads.

The controlling ADRs are:

• ADR-0008: FastAPI REST boundary and job-shaped file ingestion contract.
• ADR-0009: Postgres source files, jobs, audit, and migration conventions.
• ADR-0012: resource lifetime, dependency injection, and explicit transaction ownership.
• ADR-0013: MinIO object storage boundary.
• ADR-0017: inline ingestion, batched/bounded-concurrent embedding, anyio.to_thread.run_sync for blocking work, and request bounds. It supersedes ADR-0014, which remains the design to adopt when a broker becomes necessary.

The cited ADRs are currently proposed. Treat them as the implementation baseline only after the project owner accepts them; code should not silently diverge from them.

First release scope

In scope

• POST /v1/files for authenticated tenant-scoped CSV upload.
• File validation, size limits, content hashing, and streaming upload to MinIO.
• Alembic-managed Postgres schema for the minimal tenant/auth, source file, ingestion job, and job event records needed by this slice.
• Inline ingestion in POST /v1/files, with batched/bounded-concurrent embedding, thread-offloaded parsing, and enforced size/timeout/capacity bounds.
• CSV parsing and deterministic chunk creation.
• Tenant-filtered Qdrant point upserts using deterministic point identifiers.
• Job status/progress persistence and GET /v1/files/{file_id} status lookup.
• Structured correlation logging at HTTP and ingestion-stage boundaries.
• Automated tests for the critical state transitions, redelivery, and tenant boundaries.

Explicitly out of scope

• XLSX, DOCX, and legacy DOC ingestion.
• The conversational LangGraph API and SSE streaming.
• Final reranker selection, GPU deployment, or unresolved model licensing from ADR-0005.
• Direct /v1/points CRUD endpoints beyond the reusable service layer required by ingestion.
• Full tenant erasure and hard-deletion workflow.
• A public download API or presigned object URLs.
• A message broker, transactional outbox, job queue, and separate worker/publisher processes (ADR-0014, deferred by ADR-0017).
• late_interaction (jina-colbert-v2) document vectors at ingest — populating them is ADR-0017's primary trigger to move ingestion back off the request.
• Exactly-once end-to-end processing. Job execution must instead be safe when a job runs more than once.

Required invariants

The implementation must preserve these rules from the ADRs:

1. tenant_id is derived from trusted authentication context; it is never accepted from the upload body, query parameters, MinIO metadata, or a dispatch payload as authority.
2. MinIO stores bytes; Postgres stores metadata, lifecycle state, job progress, audit records, and the queue.
3. Dispatch payloads contain stable IDs and correlation metadata only. They never contain file bytes, extracted text, chunks, embeddings, secrets, raw prompts, or raw model output.
4. Ingestion works from the persisted job and source-file records, not from request-supplied values.
5. Parsing, chunking, embedding, and generated Qdrant Chunk/Point CRUD live in the ingestion application service, not in the route handler. Blocking work runs through anyio.to_thread.run_sync, never directly on the event loop.
6. Qdrant reads and mutations are tenant-filtered. Ingestion-generated point IDs are deterministic so retrying a job does not create duplicate logical chunks.
7. No Postgres session or transaction is held open across parse/embed/upsert. The job row is committed running before the work and updated to a terminal status after it, in a second short transaction.
8. Application clients are built at FastAPI startup and closed at shutdown. No mutable clients are opened as import-time globals.
9. Embedding is batched per provider limits and concurrency-bounded by a semaphore; blocking work (parse, chunk, BM25, minio) runs through anyio.to_thread.run_sync with an explicit CapacityLimiter.

Required decisions before implementing the affected phase

The first implementation should use these defaults unless a later ADR changes them:

Source-file idempotency and replacement

• 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 replaced only after the new job completes successfully, so a failed re-ingestion does not remove a working index.
• Preserve the original filename in Postgres metadata. MinIO object keys remain internal ID-based paths.

This policy should be made explicit in ADR-0009 before implementing re-ingestion rather than becoming an accidental repository behavior.

File deletion

For the first release, file deletion should be soft and job-shaped:

1. mark the source file as deletion requested/soft deleted in Postgres;
2. soft-delete the related Qdrant points inline, recording the attempt;
3. retain the MinIO object until an explicit retention or hard-erasure workflow.

Hard deletion requires a later retention/erasure implementation covering MinIO, Qdrant, and the relevant Postgres data.

Ingestion bounds and operations

Before deploying an environment, define and document:

• INGESTION_MAX_CONCURRENCY and the thread-pool capacity limiter, and their relation to the process's CPU/memory budget;
• INGESTION_TIMEOUT_SECONDS, and the proxy/load-balancer/client read timeouts that must exceed it;
• INGESTION_EMBED_BATCH_SIZE and INGESTION_EMBED_CONCURRENCY, sized to the provider's rate limits and the self-hosted embedder's capacity;
• alert thresholds for p95 ingestion duration, 503/504 rates, and jobs left in running past the timeout;
• how failed ingestions are inspected and retried.

These are deployment/runbook settings, not new ADRs unless they change the reliability guarantee or system boundary.

Build order

Phase 1: Foundation and local dependencies

1. Add typed configuration in src/config.py for Postgres, MinIO, Qdrant, ingestion bounds, application limits, and logging.
2. Populate .env.example with non-secret local-development configuration.
3. Add application Docker Compose services for Postgres, MinIO, and Qdrant. Keep application MinIO buckets/credentials separate from Langfuse infrastructure.
4. Add direct Python dependencies and lock them with uv: SQLAlchemy async/Postgres driver, MinIO/S3 client, Qdrant client, and structured logging dependencies chosen by ADR-0011.
5. Add the pytest foundation from ADR-0016: async test configuration, boundary markers, and support for dependency-injected fakes. Add a lifespan smoke test before creating external clients.
6. Create FastAPI lifespan setup and typed dependency helpers without creating schema at startup.

Exit criteria: local infrastructure starts; readiness checks can report each required dependency; clients are opened/closed by process owners; fast unit tests run without Docker or live providers.

Phase 2: Database, migrations, and domain contracts

1. Define SQLAlchemy models and Alembic migrations for the minimum required tables: tenants, api_keys, source_files, ingestion_jobs, and ingestion_job_events, per ADR-0009.
2. Define Pydantic request/response schemas, including the terminal upload response (file_id, ingestion_job_id, status, chunks_indexed).
3. Implement explicit repositories/services with a request/job-lifetime AsyncSession; routes/services own commit/rollback boundaries as specified by ADR-0012.
4. Write the empty-database migration test before each schema revision. Create Testcontainers-based Postgres fixtures for tenants, hashed API keys, database sessions, and migrations. Do not use create_all() in test fixtures.

Exit criteria: migrations create the schema from an empty database; application startup performs no DDL; schema and repository tests verify tenant-scoped reads/writes and valid job transitions.

Phase 3: MinIO upload and durable job creation

1. Implement API-key authentication and AuthContext tenant derivation.
2. Implement POST /v1/files for CSV only, including streaming-size controls, file-type validation, SHA-256 calculation, and a private MinIO upload using an internal object key.
3. In one short Postgres transaction, persist source_files and create ingestion_jobs(status='running'), then commit and release the connection before any parse/embed work.
4. Return 201 Created with file_id, ingestion_job_id, terminal status, and chunks_indexed once ingestion completes.
5. Implement GET /v1/files/{file_id} with tenant filtering and a public status response that does not expose raw storage credentials or internal artifacts.
6. Add cleanup/compensation handling for a MinIO upload that succeeds while the database transaction fails.
7. Add unit/API tests for trusted tenant derivation, CSV validation, idempotency, the terminal 201 Created response, and tenant-scoped status. Add MinIO adapter integration tests for server-derived private object paths and compensation behavior.

Exit criteria: an authenticated CSV upload creates a private object and a running job row committed before any ingestion work; a tenant cannot retrieve another tenant's file status.

Phase 4: Bounded execution primitives

1. Add the async embedding ports and adapters in src/infrastructure/embedding/, with per-provider batching and an asyncio.Semaphore bounding in-flight batches.
2. Route blocking work (parse, chunk, BM25, minio) through anyio.to_thread.run_sync with an explicit CapacityLimiter created at startup, so ingestion cannot exhaust Starlette's thread pool.
3. Enforce the request bounds: INGESTION_MAX_CONCURRENCY (503 + Retry-After when exceeded), INGESTION_TIMEOUT_SECONDS around the whole work phase (504), and the size/chunk-count ceiling (413) checked before work starts.
4. Guarantee a bounded failure always writes a terminal job status — a timeout must never leave a job stuck in running.
5. Add unit tests for batching/concurrency limits, timeout-to-terminal-status, and capacity rejection, using scripted embedder fakes.

Exit criteria: embedding a few hundred chunks issues batched, concurrent requests rather than serial ones; exceeding any bound produces the right status code and a terminal job row.

Phase 5: Ingestion execution and Qdrant Chunk/Point CRUD

1. Implement the ingestion service called by the route, using the application-lifetime database, MinIO, Qdrant, model, and logging clients.
2. Validate the persisted records before fetching the MinIO object.
3. Append progress events, parse CSV, create deterministic chunks, embed them, and upsert tenant-scoped Qdrant points — without holding a Postgres session open across the work.
4. In a second short transaction, mark the job succeeded with counters or failed with a safe error summary, then return the terminal response.
5. Make a retried upload safe: no duplicate logical chunks, no incorrect counters, and no transition from a terminal state back to running.
6. Add unit tests for deterministic CSV chunks, point IDs, and terminal job transitions. Add Testcontainers Qdrant and Postgres integration tests for tenant-filtered upserts, terminal state persistence, retrying an upload, and parser/Qdrant failure handling.

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 retrying the upload produces a correct final state without duplicate chunks.

Phase 6: Operations, integration tests, and documentation

1. Add an operator runbook covering local startup, migrations, MinIO bucket setup, the run command, ingestion-bound tuning, the proxy/client timeout requirement, and how to retry a failed ingestion.
2. Add a serialized Compose-based operational smoke test covering upload through indexed points against the running web process. Testcontainers remains the standard pytest mechanism for individual adapter integration tests.
3. Add end-to-end tests for duplicate upload, retrying a failed upload, tenant isolation, capacity/timeout rejection, and failed parser/Qdrant behavior.
4. Add health/readiness checks that distinguish process health from dependency readiness.
5. Update the README with local-start instructions and links to ADRs, this plan, and the operations runbook.

Exit criteria: a new developer can start the stack, apply migrations, upload a CSV, observe the job through completion, and understand how to investigate or retry a failure.

Definition of done for the vertical slice

The first slice is done when the following path works in local Compose and is covered by automated tests:

POST /v1/files (authenticated CSV upload)
  -> raw bytes stored privately in MinIO
  -> source file and running job committed in Postgres, connection released
  -> parse/chunk on threads, embed in bounded concurrent batches
  -> deterministic Qdrant points upserted
  -> Postgres records progress and terminal job status
  -> 201 Created returns the terminal result in the same request
  -> GET /v1/files/{file_id} reports that status within the owning tenant only

The next implementation work after this slice is direct Point CRUD, retrieval, and then the LangGraph conversational flow. Do not couple those later milestones to the initial ingestion path unless they are needed to preserve one of the invariants above.