fix(pipeline): re-upload of identical content unconditionally re-embeds all chunks, burning API credits and creating section gap #278

Summary

When the same content is re-uploaded (content hash matches existing document), the pipeline reuses the doc_id but does not short-circuit. It re-parses, re-TOC-extracts, re-chunks, re-embeds all chunks (burning OpenAI API credits), and calls upsert_document_and_replace_sections which deletes and re-inserts all section rows — creating a brief window where the document has zero sections and routing queries return empty results.

Root Cause

ingestion/pipeline.py around line 437–480:

existing = await find_doc_id_by_hash(content_hash)   # finds existing doc
doc_id = existing.doc_id                              # reuse ID

# No short-circuit — pipeline continues:
parsed = await parse_document(data)        # full PDF parse
toc = await extract_toc(parsed)            # full TOC extraction
sections = derive_sections(toc, parsed)    # full section derivation
chunks = chunk_sections(sections)          # full chunking
vectors = embedder.embed([c.text for c in chunks])   # ← full API call, full cost
await section_store.upsert_document_and_replace_sections(...)  # ← DELETE + re-INSERT gap
await vector_store.upsert_points(...)      # full Qdrant upsert

Failure Scenario

1. Tenant uploads a 300-page PDF (1,500 chunks). Fully indexed. Cost: $0.15 in OpenAI embedding fees.
2. Tenant re-uploads the same PDF (accidentally, or from a sync).
3. Pipeline re-embeds all 1,500 chunks: another $0.15 wasted.
4. replace_sections deletes 50 section rows and re-inserts them. During the DELETE:
   • A concurrent routing query sees 0 sections → returns empty results.
5. High-traffic systems with periodic re-uploads accumulate significant wasted API costs.

Proposed Fix

Short-circuit on confirmed dedup:

existing = await find_doc_id_by_hash(content_hash)
if existing and not doc.force_reingest:
    logger.info(
        "Content hash matches existing document — skipping re-ingest",
        extra={"doc_id": existing.doc_id, "tenant_id": tenant_id},
    )
    return IngestResult(
        doc_id=existing.doc_id,
        pre_existing=True,
        ingest_status=existing.ingest_status,
        section_count=existing.section_count,
    )

Add force_reingest: bool = False to IngestInput for callers that explicitly want to re-process.