# Ingestion Worker CLI

Module: `aibs_backend.interfaces.cli.ingestion_worker`

## Purpose

Run a standalone ingestion worker process. The worker claims durable
`IngestionJob` records and advances each document through:

1. parse
2. chunk
3. index
4. semantic_profile

Use this CLI when ingestion work must be isolated from the API process, for
example on a GPU/model host, a separate worker VM, or a scaled ingestion lane.

## Command

```bash
pdm run python -m aibs_backend.interfaces.cli.ingestion_worker
```

or, from an activated virtual environment:

```bash
python -m aibs_backend.interfaces.cli.ingestion_worker
```

## Required Environment

The worker loads `.env` from the repository root. It must be able to reach:

- document repository storage
- upload/document roots used by the API
- OpenSearch chunk, catalog, retrieval-pyramid, and job indexes
- embedding service/model
- reranker service/model when reranking is enabled
- parser dependencies
- LLM lanes needed by enabled ingestion features, including semantic profiles

Important knobs:

| Setting | Meaning |
| --- | --- |
| `AIBS_INGESTION_WORKER_COUNT` | Number of worker loops in this process. `0` disables the pool. |
| `AIBS_INGESTION_WORKER_IDLE_SLEEP_SECONDS` | Sleep when no job is claimed. |
| `AIBS_INGESTION_WORKER_ERROR_SLEEP_SECONDS` | Sleep after a loop-level exception. |
| `AIBS_INGESTION_WORKER_LEASE_REAP_INTERVAL_SECONDS` | How often expired job leases are reclaimed. |
| `AIBS_INGESTION_WORKER_CLAIM_SCAN_SIZE` | Number of claimable jobs scanned per claim attempt. |
| `AIBS_INGESTION_MAX_CONCURRENT_PARSE_STAGES` | Process-local bound for parse stages. |
| `AIBS_INGESTION_MAX_CONCURRENT_CHUNK_STAGES` | Process-local bound for chunk stages. |
| `AIBS_INGESTION_MAX_CONCURRENT_INDEX_STAGES` | Process-local bound for index stages. |
| `AIBS_INGESTION_MAX_CONCURRENT_SEMANTIC_PROFILE_STAGES` | Process-local bound for semantic-profile stages. |
| `AIBS_INGESTION_MAX_CONCURRENT_HEAVY_MODEL_STAGES` | Process-local bound for heavy model-backed ingestion stages. |

## Startup Flow

1. Load `.env` with override enabled.
2. Build `AppSettings` and configure structured logging.
3. Build the application container.
4. Run readiness checks:
   - document repository integrity, when available
   - chunk OpenSearch connection
   - document catalog index connection
   - retrieval pyramid index connection, when configured
   - embedding readiness
   - reranker readiness when `AIBS_RERANKER_ENABLED=true`
   - parser warmup
   - answer assistant readiness when abstractive summaries require LLM startup checks
5. Reap expired ingestion leases once before the worker pool starts.
6. Start `run_ingestion_worker_pool`.

Readiness failure exits before any job is claimed.

## Semantic Profile Stage

The semantic profile is a persisted ingestion artifact stored on the document
metadata under `semantic_profile`. It is separate from processing metadata such
as `document_family`.

The stage runs after indexing. That means a document can become searchable before
semantic enrichment finishes. If model-backed semantic enrichment fails during
normal ingestion, the stage retries and then persists an unavailable semantic
profile so the durable ingestion job can complete degraded. The maintenance
backfill CLI behaves differently: it reports failures and does not silently
persist unavailable profiles.

## Worker Loop Behavior

Each worker loop uses an ID like:

```text
aibs-worker-<hostname>-<pid>-<lane>
```

The loop repeatedly:

1. reaps expired leases when the interval has elapsed
2. claims one job through `IngestionJobRunner.claim_and_run_once`
3. runs the next stage for that job
4. renews/observes durable job state through the runner
5. sleeps when no job is available

`asyncio.CancelledError` stops the loop cleanly. Other loop exceptions are logged
and retried after `AIBS_INGESTION_WORKER_ERROR_SLEEP_SECONDS`.

## Verification

After startup, confirm:

- readiness logs are successful
- worker pool started with the expected worker count
- uploaded documents leave queued/running states
- job stage results include parse, chunk, index, and semantic_profile
- degraded semantic profiles are reviewed and later repaired with
  [backfill_semantic_profiles.md](backfill-semantic-profiles.md)

## Safety Notes

- In split deployment, set `AIBS_INGESTION_WORKER_ENABLED=false` on API nodes if
  ingestion should run only on worker nodes.
- Multiple workers can run against the same durable job index, but model
  concurrency must be sized for the cluster. The process-local stage semaphores
  are not a replacement for global model admission.
- Do not use production indexes from a local development `.env`.