Document Ingestion Pipeline

Goal

Turn an uploaded document into workspace-scoped, embedded chunk records that are search-ready in OpenSearch.

Pipeline Stages

The current happy path is:

registered -> parsing -> parsed -> chunking -> chunked -> embedding -> indexing -> indexed -> ready

The current degraded path ends at:

degraded

Any non-hard-terminal in-progress stage may also end in:

failed

Stage Breakdown

1. Upload and registration

HTTP entry:

  • POST /api/aibs/documents

Application service:

  • RegisterAndSubmitDocumentService (composed of upload + submit)

What happens:

  • validate workspace id, media type, and upload size

  • stage the browser upload to local disk

  • compute checksum

  • reject duplicates within the same workspace

  • copy the staged file into managed local storage

  • create a Document aggregate in registered state

  • persist the document to the JSON repository

  • submit an OpenSearch-backed ingestion job as part of the same request

  • if job submission fails, rollback registration (delete document JSON + raw file)

Output:

  • document id

  • storage ref

  • registered status

  • backend owns ingestion kick-off; no mandatory /process client call on happy path

2. Parse

Application service:

  • ParseDocumentService

What happens:

  • move document to parsing

  • load raw file from local storage

  • parse it with Docling

  • normalize parser output into the canonical document shape

  • run primary-language detection on a bounded sample of the parsed body text (text blocks and table row-group text, up to 10k characters in reading order), and persist the result as a structured Document.language value object (ISO 639-1 primary code, confidence, detected-at timestamp, classifier identity)

  • attach ParsedDocument

  • persist the aggregate in parsed

If parsing fails:

  • the document is marked failed

Language detection is best-effort and runs outside the parse try/except. If the classifier fails or returns no confident verdict for the sampled text, the document still completes parse with language=None; ingestion is never blocked by a language-detection failure.

3. Chunk

Application service:

  • ChunkDocumentService

What happens:

  • move document to chunking

  • build chunk records from the canonical parsed document

  • wrap them in a ChunkSet

  • attach chunk policy version

  • persist the aggregate in chunked

If chunking emits warnings:

  • the chunk set may be marked partial

  • later indexing may produce a degraded final outcome

If chunking fails:

  • the document is marked failed

4. Embed and index

Application service:

  • IndexDocumentService

What happens:

  • move document to embedding

  • embed each chunk with the local sentence-transformers model

  • move document to indexing

  • create or validate the workspace OpenSearch index

  • replace previously indexed chunks for the same document

  • bulk write current chunk records plus vectors

  • mark indexed

  • mark final state as ready or degraded

  • save the final aggregate, which attempts to update the OpenSearch document catalog projection used by the workspace source list and the retrieval pyramid profile/summary projections used by broad discovery

The final state becomes degraded when, for example:

  • the aggregate was already degraded from earlier stages

  • the chunk set is partial

  • embedding emitted warnings

  • indexing emitted warnings

  • fewer chunks were indexed than expected

If indexing fails after side effects:

  • the service tries to clean up indexed chunks

  • the document is marked failed

If a persistence save fails after successful side effects:

  • the service still tries to clean up indexed chunks

  • the failure is surfaced honestly instead of pretending the aggregate reached a durable final state

Document Catalog Projection

Document JSON files remain the source of truth for the Document aggregate. Large workspace source-list reads use a separate OpenSearch read model instead of scanning every document JSON file for every page.

Projection behavior:

  • index name defaults to aibs-document-catalog

  • environment override: AIBS_DOCUMENT_CATALOG_INDEX

  • one OpenSearch document is stored per uploaded document

  • saves and deletes retry projection writes and then log best-effort failures without rolling back the JSON source of truth

  • catalog/search hydration reads lightweight records from the projection

  • summaries and paging are computed in OpenSearch

Backfill existing workspaces after enabling or renaming the projection:

python -m pdm run python -m aibs_backend.interfaces.cli.rebuild_document_catalog [--workspace <id>] [--batch-size 200] [--dry-run] [--validate-only] [--recreate] [--only-missing]

The projection intentionally duplicates listing/search metadata plus duplicate checksum fields: title, workspace id, status, knowledge flag, language, page/chunk counts, timestamps, failure details, and checksum identity data.

Catalog schema 4 also projects the canonical document identity profile built from existing ingestion artifacts:

  • recovered content title and title provenance

  • aliases

  • aboutness summary and summary provenance

  • document type/family

  • topics, entities, measures, and classification signals

  • source URI/storage reference

  • source organization, publisher, jurisdiction, and country/region

  • publication/effective/coverage dates and document year

  • version label, supersession links, and latest-state flag

  • profile/source-truth completeness and degraded reasons

Catalog text fields use a catalog-specific OpenSearch analyzer that splits common filename separators such as _, -, ., /, and \, so filenames like microsoft_2024_annual_report still match annual report. Schema 4 also stores a compact search_text field built from the same structured profile fields so mixed references such as document code + title keyword + topic can bind quickly. It does not replace chunk/vector indexes and it does not replace aggregate JSON.

Rebuilds upsert the current JSON documents first and delete stale projection rows at the end. That avoids the old clear-then-fill blackout where the source list could briefly show zero documents during a rebuild.

Schema-changing catalog upgrades use the same CLI. The catalog index stores a mapping schema version; startup logs a warning if the live index is stale. A workspace-scoped rebuild refuses stale schemas because recreating the shared index would affect other workspaces. Run the rebuild without --workspace once to recreate the catalog index and backfill all JSON documents.

Validation output includes row drift plus ingestion-quality counters:

  • profile complete/partial/missing

  • source-truth complete/partial/missing

  • missing recovered title

  • missing aboutness summary

  • degraded summary count

Retrieval Pyramid Projection

Document JSON files still remain the source of truth, and the workspace chunk index remains the final evidence layer. The retrieval pyramid is a cheaper OpenSearch read model used before chunk retrieval for broad document discovery.

Projection behavior:

  • document profile index defaults to aibs-document-profiles

  • environment override: AIBS_DOCUMENT_PROFILE_INDEX

  • section/table summary index defaults to aibs-section-summaries

  • environment override: AIBS_SECTION_SUMMARY_INDEX

  • profiles store one row per document with title, recovered content title, title/aboutness provenance, aliases, language, document family/type, profile completeness, source truth completeness, source/date/version signals, status, and parser/chunker signals

  • section/table rows store section title, breadcrumb, summary text, page span, document ID, type, language, topic/entity hints, and completeness context

  • saves and deletes update both projection indexes best-effort after the JSON source of truth changes

  • discovery tries profile/summary hits before broad chunk evidence only when AIBS_RETRIEVAL_PYRAMID_DISCOVERY_ENABLED=true

Backfill existing workspaces after enabling or renaming the projection:

python -m pdm run python -m aibs_backend.interfaces.cli.rebuild_retrieval_pyramid [--workspace <id>] [--batch-size 200] [--dry-run] [--validate-only] [--recreate] [--only-missing] [--profile-version <int>]

The current profile index schema is 5. Schema-changing upgrades require an unscoped --recreate rebuild before workspace-scoped pyramid validation can be trusted.

Full Orchestration

Upload now is backend-owned end-to-end for the happy path:

  • POST /api/aibs/documents uses RegisterAndSubmitDocumentService to register and submit (or reuse) a durable ingestion job before returning.

  • POST /api/aibs/documents/{id}/process remains the manual retry/recovery path for non-terminal documents that need explicit re-submission.

POST /api/aibs/documents/{id}/process still calls SubmitIngestionJobService, which creates or reuses a durable ingestion job in OpenSearch and returns 202 Accepted immediately.

The job is then claimed and executed by IngestionJobRunner inside a worker loop. The runner runs stages idempotently based on the document’s current status:

  • registered or parsing → parse stage

  • parsed or chunking → chunk stage

  • chunked, embedding, or indexing → embed/index stage

  • ready, degraded, or indexed → job marked succeeded, no stage re-run

The worker loop runs either as an in-process background task inside the FastAPI process (when AIBS_INGESTION_WORKER_ENABLED=true) or as a standalone worker CLI (python -m aibs_backend.interfaces.cli.ingestion_worker).

Each active stage is heartbeated on a configurable interval. If the worker process crashes, the lease expires and another worker reaps and retries the job.

Threading And Event Loop Behavior

Some heavy work is still synchronous from a service perspective, but CPU-bound or external-library-heavy calls are pushed through asyncio.to_thread(...).

That means:

  • the service still waits for stage completion internally

  • the event loop does not get blocked by parser/chunker/embedder CPU work

  • the public API does not make the browser wait for full completion

Language Backfill

Documents that were ingested before structured language detection joined the pipeline carry no language metadata. The workspace inventory reports those documents as unknown in its by_language breakdown.

The backfill CLI populates Document.language for those documents without re-running the full parse:

python -m pdm run python -m aibs_backend.interfaces.cli.backfill_language [--workspace <id>] [--dry-run] [--force]

The CLI samples the same parsed body text used at parse time. If the parsed body has no usable text, the CLI falls back to enrichment summaries already stored on the parsed document; if neither has text, the document is reported as skipped_no_text. Per-document failures are isolated and logged so a broken document never aborts the run. With --force the CLI re-detects every document; results that match the existing language record are reported as unchanged rather than detected so the operator sees an honest summary.

Atomic Document Writes And Startup Integrity

The JSON document repository writes each document file through a same-directory temp file followed by os.replace, so a crashed write cannot leave the document in a partially serialized state. On startup, the repository runs an integrity check that loads every persisted document JSON and fails fast if any file is malformed or unloadable, surfacing latent corruption before the API begins serving traffic.

Current Non-Goals

Still deliberately deferred from the ingestion slice:

  • retrieval-time ranking policy configuration

  • cross-workspace document sharing rules

  • production-grade document metadata persistence (currently local JSON plus an OpenSearch catalog read model)