# 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:

```powershell
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:

```powershell
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:

```powershell
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)