Document Ingestion Pipeline¶

How CampusCore turns documents (scraped pages, uploaded files, connector imports, chat attachments) into chunked, embedded, entity-typed records that the agent can retrieve from. This doc describes the running system; for engineer-level detail on the state machine (step diagram, error categories, how to add a step), see document-ingestion-state-machine.md.

1. Overview¶

Every ingestion - regardless of source - flows through one unified pipeline:

A trigger creates a DocumentIngestionRun + one DocumentIngestionRunItem per file (DB rows; durable from this point). DocumentProcessingMessage is a union of one model per ingest kind - FolderUploadMessage, WebScrapeMessage, AttachmentMessage, ConnectorFileMessage - discriminated on source_type, and each kind is built only by its own factory in trigger.py: build_folder_message, build_scrape_message, build_attachment_message, build_connector_file_message. A kind declares only the fields that kind has, and every one of them is required, so what a kind needs is stated by its class rather than by prose; consumers narrow with isinstance or match instead of null-checking optionals, and the worker parses every body through parse_document_processing_message. The two chat-scoped kinds, AttachmentMessage and ConnectorFileMessage, share an interactive ingest profile: fast PDF parsing, no LLM metadata, no entities, no summary chunk, no parsed-file persist, and answerability the moment text chunks persist. The kind vocabulary those tags come from - IngestSourceType and CHAT_SCOPED_SOURCE_TYPES - is declared in schemas/ingest_kinds.py, a leaf module that imports nothing, because the persisted chunk metadata, the observability buckets and the search-result labels hold the same values long after the message is gone; is_chat_scoped answers from that set, and test_queue_message_kinds pins the declaration against the message classes. The whole schemas/ package holds to the same rule - only the standard library, Pydantic, typing_extensions and siblings - pinned by test_schemas_import_direction, with entities.py's import of EntityContentType as the one recorded exception. A ConnectorFileMessage processes a shared copy: the kind carries no user_id at all, its status flips address every ConversationAttachment link of the file, and cancellation counts links - the copy is abandoned only when no live link remains (see connectors.md, "Chat-Attachment File Bridge"). The scraper writes SourceFile.source_url at save time and the scrape factory reads it.
The TwoQueueDispatcher publishes each item onto either the interactive or bulk SQS queue.
The worker (ECS Fargate) polls interactive-first, pulls a batch under an AdmissionController memory gate, and runs each item through the state machine concurrently via asyncio.
The DocumentIngestionStateMachine drives each item through ordered, idempotent step transitions; every transition is a state-machine row write + a DocumentIngestionRunEvent (cost, duration, error kind).
Each step is a pure async function (fetch, extract_content, chunk_and_embed, extract_entities, persist, finalize). They all run in-process in the worker - there is no separate compute fabric.
On terminal failure the item lands in the Failed Items Inbox (an admin proxy over items whose outcome is failed or abandoned). On a retryable failure the worker republishes the item with a wait that depends on the queue, 15 s then 60 s on the interactive queue and 60 s then 300 s on the bulk queue, whatever the failure's kind, and every failure counts against one budget for the whole item, max_retries attempts in all (see §6). The file itself records why it failed in SourceFile.error_message, which a row constraint requires on a failed file and forbids on every other status, so a file that failed and was retried into extracted no longer carries the old reason. What it records is one sentence per failure kind, chosen by failure_reason_for in services/document_ingestion/file_status.py, because the folder list and the chat chip show that text to the person who uploaded the file.
Fatal - "The file could not be processed. Check that it opens on your computer, then upload it again."
SchemaInvalid - "The file's contents could not be processed. Upload it again, or try another format."
Transient - "Processing stopped after several attempts. Upload the file again."
RateLimited - "Processing stopped because a service it needs was too busy. Upload the file again later."

The raw message the step raised stays on the item's last_error_message, which is what the Failed Items Inbox shows an operator, so a library's own wording never reaches the folder list. A caller that knows the failure better than its kind does passes its own sentence instead: trigger.enqueue hands terminate_item the file_reason PUBLISH_FAILED_REASON for a message that never reached SQS. A failure found where no item speaks for the file is written through write_failure_for_file in the same module, which settles the file and every chat link together: the connector copy's two publish undos and its bounded wait in connector_file_ingest.py, the admin "Set status to 'Failed'" action, and the failed-prior restores in process_content and extraction_review all use it.

A crash mid-file resumes from the last committed step, and a requeue at or past chunks_persisted rebuilds its context from the chunk rows rather than re-fetching, re-extracting and re-chunking the source. Entity extraction is still re-run until the item reaches entities_persisted, because the batch is durable nowhere before that, and embed_chunks re-embeds the whole file whenever a single vector is NULL.

2. Architecture¶

                       trigger.enqueue(kind, items)
                                  │
                                  ▼
            ┌─────────────────────────────────────────────┐
            │ DocumentIngestionRun + Item rows (DB)       │
            │ - durable from this point                   │
            └────────────────────┬────────────────────────┘
                                 │
                                 ▼
                      ┌──────────────────────┐
                      │ TwoQueueDispatcher   │
                      └──────────┬───────────┘
                                 │
                       ┌─────────┴─────────┐
                       ▼                   ▼
                  SQS interactive       SQS bulk
                       │                   │
                       └────────┬──────────┘  (interactive drained first)
                                ▼
                   ┌──────────────────────────┐
                   │  ECS worker (Fargate)    │
                   │  asyncio batch,          │
                   │  max_concurrency = 10,   │
                   │  AdmissionController gate│
                   └────────────┬─────────────┘
                                │ (one task per item)
                                ▼
              ┌──────────────────────────────────────┐
              │  DocumentIngestionStateMachine       │
              │   .run_step(target, callable)        │
              │                                      │
              │   for each step (idempotent):        │
              │     ▸ fetch                          │
              │     ▸ extract_content                │
              │     ▸ chunk_and_embed                │
              │     ▸ extract_entities               │
              │     ▸ persist                        │
              │     ▸ finalize                       │
              └────────────┬─────────────────────────┘
                           │
                           ▼
              DocumentIngestionStep advances
              + DocumentIngestionRunEvent written
              (step / duration / model_id / tokens / cost / error_kind)

External boundaries the worker talks to:

Postgres (TimescaleDB image) - the durable layer for DocumentIngestionRun*, Document, DocumentChunk, SourceFile, the entity satellite tables, and the embeddings (pgvector with HNSW).
SQS - the work queue (ElasticMQ in dev; real SQS in cloud).
S3 - for raw source files and uploads. Not for per-step staging (the original Lambda-era staging bucket was removed; see §10).
Gemini / OpenAI - LLM/embedding providers. Calls go through apps/main_app/clients/.
Redis - rate-budget coordinator (Lua-scripted token bucket per provider).

3. Data model¶

Defined in apps/main_app/models/document_ingestion.py. Four tables and eight TextChoices enums:

Table	Purpose
`DocumentIngestionRun`	One row per ingestion operation (scrape, folder upload, chat attachment, connector import, manual reprocess). Aggregates item counts.
`DocumentIngestionRunItem`	One row per file. Carries the current `step`, retry count, last error, FK to the `SourceFile`.
`DocumentIngestionRunEvent`	Append-only per-step lifecycle log. Both `event_type` (`DocumentIngestionRunEventType`) and `step` (`DocumentIngestionStep`) are constrained to their enum. Holds `duration_ms`, `error_kind`, `error_message`. A constraint ties the duration to the event type: `step_completed` always carries a duration; `step_failed` carries one when the failure ends a measured attempt, and `terminate_item`'s administrative failure rows carry none. A start or a scheduled retry never carries one. Cost lives in the `ModelUsageEvent` ledger, keyed back by run/item id.
`DocumentIngestionStepSpan`	One row per sub-step per step attempt, hanging off its `DocumentIngestionRunEvent`. Holds a `name`, a `total_ms` that is the union of that sub-step's intervals so concurrent same-name work is counted once, and the `count` of intervals that went into it. Unique on `(event, name)`, so an attempt reports each sub-step once.

Enums (models.TextChoices):

DocumentIngestionRunKind - scrape / folder_upload / chat_attachment / connector_import / manual_reprocess.
DocumentIngestionRunStatus - running / completed / failed / abandoned. Every creator starts a run that is already under way, so running is the only non-terminal member and a run is never written in any other opening state. A check constraint holds the pairing with the clock: a terminal run has a completed_at, a running one does not, and started_at is never null.
DocumentIngestionStep - enqueued / fetched / content_extracted / chunked / chunks_persisted / embedded / entity_extracted / entities_persisted / indexed / done. Every member is a position; the happy-path order is declared once as STAGES in services/document_ingestion/steps.py, and the state machine and the runner both derive from it.
DocumentIngestionOutcome - in_progress / done / failed / abandoned, on the same row as the step. It is how the ingest ended, so a failed item still reports the last step it committed, and every "has this settled?" question reads it.
DocumentIngestionRunEventType - step_started / step_completed / step_failed / retry_scheduled, on DocumentIngestionRunEvent. What happened to one step attempt, held apart from the step itself.
DocumentIngestionErrorKind - "" (none) / RateLimited / Transient / SchemaInvalid / Fatal. Declared once here and imported by everything that carries an error kind, StepResult included.
DocumentIngestionSpanName - html_to_md_llm / libreoffice_convert / parse_file / metadata_llm / summary_llm / chunk / classify / pymupdf / liteparse / rasterize / ocr, on DocumentIngestionStepSpan. Each member answers kind, which is what a breakdown totals on.
SpanKind - llm / cpu / container. container is parse_file, whose interval contains the parser's own spans, so adding it to either total would count that work twice.

Admin entry points: /cc_admin/ → Pipeline Manager (/cc_admin/pipeline/), Failed Items Inbox.

4. Pipeline steps¶

The state machine doesn't know what each step does - it just orchestrates. Each step is one async function under apps/main_app/services/document_ingestion/pipeline/, mapped to its step in runner.py's _STEP_FNS table and assembled into PIPELINE in the order steps.STAGES declares. Most are pure logic; the LLM-bearing ones are flagged.

Step	Module	LLM?	What it does
`fetched`	`pipeline/fetch.py`	-	Pulls raw bytes from S3 / HTTP into `ctx.raw_bytes` and resolves the content type.
`content_extracted`	`pipeline/extract.py`	yes	Dispatches into `services/content_extraction`: HTML → markdown via Gemini, files → markdown via the parser registry, markdown → passthrough. Produces `ctx.markdown`, `ctx.raw_chunks`, the typed `ctx.provenance` + `ctx.summary`, and the parser pass-through `ctx.extracted_metadata`. Ends by persisting `ctx.markdown` to `SourceFile.parsed_file` for every ingest kind except chat attachments (see "The parsed artifact" below).
`chunked`	`pipeline/chunk_and_embed.py`	-	`package_chunks` finalizes the chunks produced upstream (or splits raw markdown via `chunk_markdown`).
`chunks_persisted`	`pipeline/persist.py`	-	`persist_chunks` writes chunk TEXT rows (embedding NULL) - a chat attachment becomes readable/answerable here while the vector + entity work continues behind it. The link reads `available` when the state machine records this step, not from inside it.
`embedded` ‖ `entity_extracted`	`pipeline/chunk_and_embed.py`, `pipeline/entity_extract.py`	yes	Parallel group. `embed_chunks` calls the OpenAI embeddings client in batches (model + dimensions come from the `EMBEDDING_MODEL_ID` / `EMBEDDING_DIMENSIONS` constants in `apps/main_app/clients/openai.py`) and backfills vectors into the persisted rows; `extract_entities` runs an enumerate-then-paginate flow of up to three Gemini calls → an `EntityBatch` of `ExtractedEntity` envelopes carrying typed payloads (Article / Event / Course / …): the first call names every entity on the page and returns a first batch of documents, and follow-up calls fetch whatever the roll call named but no document covered. Trivial markdown (<200 chars) bypasses the LLM for a synthesized `GenericPage`. Text too long for the model is skipped rather than failed: when the provider refuses the first call with `ContextWindowExceededError`, the extractor makes no further call and returns a `GenericPage` batch, which stores no entities, so the file still reaches Ready and is searchable through its chunks; the step logs one warning with the file's id and text length and increments `ingest.entity_extraction_too_long`. A follow-up call that overflows keeps what the first call extracted. Every attempt records an `EntityExtractionSignal` row - see §7. Each entity in the batch becomes an `EntityDocument` row whose `content_type` carries a membership constraint on `EntityContentType.values`, so only a member of the enum reaches storage.
`entities_persisted`	`pipeline/persist.py`	-	`persist_entities` upserts `Document` + entity satellite rows in one transaction.
`indexed`	`pipeline/finalize.py`	-	`finalize_indexes` merges the extracted metadata, retires superseded files, and emits per-source telemetry; the `SourceFile` reads `extracted` and its links `ready` when the state machine records this step, after the function returns - the HNSW/FTS indexes are already current from chunk insert. Retiring first detaches the victims' entity documents (`DocumentWriter.detach_source_files` reassigns docs merged into them to their surviving files), so a folder "replace" keeps the replacement's entities and the FK cascade takes only docs whose sole backing was the victim.
`done`	`pipeline/runner.py`	-	Terminal marker - closes the item, and the run if it was the last item in flight.

Entity identity¶

DocumentWriter dedups at insert time on the key each payload type declares for itself (identity_key on the classes in schemas/entities.py, stored as EntityDocument.identity_primary), with a soft qualifier that lets a payload missing a field merge into one that has it. A type whose key form depends on where the thing was seen declares the other forms on its key: an article listed on the news page is keyed by its URL and names its title+date form as an IdentityKey.aliases entry, and the same article extracted from its own page is keyed by title+date and offers the page URL as IdentityKey.page_alias. _find_existing tries the primary, then each alias, so the writer finds the row whichever ingest came first. The page alias is only matched when the page is the entity: write_entity_batch honours it for a type the batch holds exactly one page-level payload of, and _find_existing matches it only against a row of the same normalized name, so an archive page's unlinked entries never fold into each other or into a listing row keyed by the archive's URL, whether the archive holds one entry or many. The title alias needs a date: two undated stories with one title stay two rows. Aliases are matched, never stored - a row keeps the key it was first stored under, and every later payload of either form finds it through the aliases (re-keying to the latest form would strand a later payload of the first form, such as a listing item with no date). Rows persisted before the aliases existed are folded by manage.py reconcile_entities --content-type article, which pairs a title-keyed row with the URL-keyed row stored at its source_url when the titles agree, and re-keys each survivor to the URL form; it expects a drained worker, the way the person backfill is run. Both modes run from the Operations console under "Reconcile duplicate entities" (services/admin_operations/registry.py), which requires a dry run before the apply and surfaces the quiet-worker checklist.

The parsed artifact¶

Every non-attachment ingest stores the canonical markdown on SourceFile.parsed_file as all_files/parsed/sourcefile_{id}.md (the file-storage bucket in cloud, MEDIA_ROOT in dev). Chat attachments are excluded: their removal contract deletes every artifact, and a parsed copy nothing reads would outlive it. Files ingested before this write existed have the field empty; the admin surfaces fall back to the chunk reconstruction for them. A post_delete receiver on SourceFile (signals/storage_signals.py) deletes the blob whenever a row dies - service deletes, folder-delete cascades, benchmark cleanup - because Model.delete() never touches FileField storage.

The two LLM-bearing services (content_extraction.process_* and EntityExtractor) live as their own packages with their own front doors and tests - the pipeline only composes them. The extraction prompt's per-type shape blocks are rendered at import from the payload classes in apps/main_app/schemas/entities.py (prompts/scraper/entity_extraction.py), so a schema field change updates the prompt automatically and adding an entity type without prompt wording fails at import.

The inter-step metadata contract¶

A chunk's metadata column is one model, ChunkMetadata (schemas/chunk_metadata.py), with four keys: parser, provenance, context_summary and is_summary. parser is the extractor's DocumentMetadata (schemas/document.py), stored as its own block and never merged with anything, so a parser key can no longer collide with a provenance name. It carries the four blocks the metadata call fills - event, source, temporal, classification - plus the two the extractor stamps afterwards: summary, and clean_route naming which of the three HTML routes ran (null off the HTML paths). Every model in that family sets extra="forbid", so a renamed extractor key fails at the parse instead of reading as absent for the four readers that take it. The one reader that still asks the database rather than the model goes through DocumentChunk.parser_lookup("clean_route"), which builds the JSON path from DocumentMetadata.model_fields and raises on a field name the model does not have. Migration 0161 is what made the existing rows fit: it stripped every key outside the type, at the top level and inside each block, and refused to start on a row no rewrite could bring inside the type's shape. provenance is an IngestProvenance dump: the ingest kind, plus an AttachmentScope or a ConnectorFileScope for the two chat kinds, and nothing else. It deliberately does not carry the file name, the source URL, the S3 key or the folder id - the parent SourceFile and the chunk's own columns already own those, and a second copy is a second thing to keep in step; the chat scope blocks are the exception, because the link rows they mirror can be deleted while the chunks live on.

The model is the only writer and the only reader. extract_content builds the provenance from the job with provenance_from_job, StepContext.chunk_metadata() builds the whole model, DocumentWriter dumps it with to_row(), and every reader parses it back. extra="forbid" makes a row of the pre-split flat shape fail to parse rather than read as an empty provenance: the two search tools drop such a hit through the shared helper in services/agent/hit_metadata.py, logging at ERROR with the chunk id and counting chunk_metadata.invalid_row, and the resume rebuild is the one reader that tolerates it, falling back to a full run. Migration 0159 rewrote every existing row into this shape, inferring the kind from the chunk's own access columns where no stamp existed. ctx.extracted_metadata holds that same DocumentMetadata, None until extract_content sets it and required after - every step that reads it raises when it is still None, the way they do for ctx.source_file - and its dump alone is what finalize merges into SourceFile.metadata; finalize also writes the message's name into SourceFile.display_name. The exact per-ingest-kind row is frozen by tests/test_ingestion_metadata_contract.py.

The file-format registry¶

SUPPORTED_FORMATS in apps/main_app/utils/content_types.py is the single source of truth for what a file type is: its extensions, content type, extraction route, parser, upload permission, legacy-conversion target, and the HTTP MIME values that identify it when no extension is available. Everything else derives from it - upload validation, the scraper's download gate, parser dispatch, MIME_TO_CONTENT_TYPE header detection, and the SPA's upload allowlist.

To add or change a format, edit SUPPORTED_FORMATS, then run scripts/sync_upload_contract.sh to regenerate the SPA's web/src/types/generated/upload-contract.ts (allowlist, size limits and the branding image caps). The file-size limit is not in the registry: it is MAX_FILE_BYTES in campus_core/shared_utils/file_size.py, exactly 30,000,000 bytes, and it bounds folder uploads, chat attachments and files loaded into a chat through a connector. Sizes read in decimal units everywhere, so the limit reads "30 MB": format_file_size in the same module writes every server and connector message, and the SPA's formatBytes follows the same rule, with both test suites reading one example file, web/src/lib/upload/fileSizeExamples.json. Three drift guards keep the derivations honest: tests/test_file_format_registry.py pins the derived lists and MIME map to the registry, and the sync script's --check mode runs in gate.sh, pre-commit, and CI - a registry edit that is not regenerated is a red build, the same as an OpenAPI drift.

5. Coordination primitives¶

The pieces that decide what runs when, separate from the per-file pipeline above.

TwoQueueDispatcher¶

Two SQS queues, one interactive and one bulk. The worker drains interactive first; a 25k-page scrape never sits in front of a chat attachment. trigger.enqueue picks the queue by DocumentIngestionRunKind via queue.priority_for_kind - chat_attachment → interactive; scrape / folder_upload / connector_import / manual_reprocess → bulk (unknown kinds default to bulk).

The worker takes two kinds of message off those queues, not one. parse_worker_message (services/queue_service.py) is a discriminated union on source_type: an ingestion body becomes a DocumentProcessingMessage and runs the state machine, and an artifact_export body becomes an ArtifactExportMessage and runs the export job (artifacts.md). An export goes on the interactive queue, because a person is watching a dialog for it, which is the same reason a chat attachment does; LibreOffice runs under a semaphore of two in the worker so an export cannot take every slot from ingestion.

Which client a publish goes out on is a choice each caller makes by picking a constructor. TwoQueueDispatcher.for_request_path() is bounded - a 2 second connect, a 3 second read and one attempt - so an unreachable queue ends the call in seconds instead of holding it open through botocore's default minutes. TwoQueueDispatcher.for_background_work() is the retrying client, four attempts on botocore's default timeouts, for a caller that would rather wait than lose the message. trigger.enqueue publishes on for_request_path() unless the caller hands it a dispatcher, because a caller that does not choose is the one with a person waiting on the answer. The callers with nobody waiting pass for_background_work(): the crawl, the process_content and reprocess_from_raw commands, the connector cloud import thread, and an artifact export delivered into a knowledge folder. The worker builds the same background dispatcher for its poll and its retry resends, because a read timeout shorter than the long poll's own wait would end every quiet cycle in an error.

A publish that raises terminates the item enqueue has just created, so the SourceFile and every live ConversationAttachment link read failed with PUBLISH_FAILED_REASON, "The file could not be queued for processing." What enqueue raises next says which kind of failure it was. _queue_could_not_take_it is true for a network failure, botocore's ConnectionError or HTTPClientError, and for a service-side refusal: a status of 500 or above, or one of the throttle codes RequestThrottled, ThrottlingException and KmsThrottled. For those enqueue logs one error-level line carrying the item id and the exception's class name, then raises IngestPublishError from the original. That log line is how an outage still reaches Sentry, because a deliberate status writes none of its own. Every other failure - missing credentials, no region, a denied permission, a queue that does not exist - is re-raised as itself and stays a logged 500, because a fault in our own code or deployment must not tell a person to try again.

domain_errors() in apis/files_api.py turns IngestPublishError into ApiServiceUnavailableError, so POST /api/uploads/confirm, POST /api/uploads/multipart/complete and POST /api/chat/attachments/{id}/confirm answer 503 with the code service_unavailable and the sentence "The file could not be queued for processing. Try uploading it again." The SPA shows that sentence on the pending upload row in a knowledge folder and on the attachment chip in the composer.

AdmissionController¶

A memory-watermark gate around SQS pulls. The worker checks container memory usage before each batch pull:

0.85 watermark → stop pulling new work (back-pressure).
0.65 watermark → resume pulling.

Prevents OOM-kills on bulk runs where embedding payloads spike memory.

Rate-budget coordinator¶

A Redis Lua-script token bucket, with separate RPM and TPM buckets per provider/model (e.g. gemini.flash-lite.rpm, gemini.flash-lite.tpm, openai.embeddings.rpm, openai.embeddings.tpm). Refill rates are tuned to ~90% of each published quota.

Every LLM/embedding call site wraps its invocation in the gate(budget, provider, model_family) async context manager:

Pre-call - acquires one slot from the .rpm bucket. If the bucket is empty, gate sleeps for the bucket's reported wait_ms and retries (the RPM cap becomes a queue, not an error). The work eventually runs rather than failing.
Post-call - the call site invokes await g.charge(response) (or charge_tokens(n) for embeddings, which don't expose usage_metadata), debiting the actual token count from the .tpm bucket. Over-budget TPM logs a warning but doesn't block - the work is already done.

The backend is selected at process start by get_rate_budget(): RedisRateBudget when settings.REDIS_URL is set (fleet-wide coordination), else InProcessRateBudget (dev/single-worker). The 6 wired call sites are the entity extractor, Gemini OCR parser, HTML→markdown conversion, metadata extraction, summary generation, and OpenAI chunk embeddings.

ECS scale-up pre-warm (`scale_workers_for_bulk`)¶

When a bulk-trigger (scrape, large folder upload) creates a run, trigger.py bumps the ECS worker service's desired_count so workers are ready when the SQS lag hits. Tied to INGEST_BULK_WORKER_DESIRED_COUNT and friends in settings.

6. Error handling¶

The 4-way exception taxonomy in services/error_handling/exceptions.py, which maps onto the DocumentIngestionErrorKind values declared on the model:

Exception	`DocumentIngestionErrorKind`	State-machine reaction
`RateLimitedError`	`RateLimited`	Republished with a wait, retried up to `max_retries`; the retry's calls wait on the rate budget.
`TransientError`	`Transient`	Republished with a wait, retried up to `max_retries`.
`SchemaInvalidError`	`SchemaInvalid`	Surface to Failed Items Inbox. Do not retry - same input → same output.
`FatalError`	`Fatal`	Surface to Failed Items Inbox. Operator intervention.

Both retryable kinds wait the same before their retry. The worker republishes the item with an SQS DelaySeconds that depends on the queue it came from, whatever the failure's kind: 15 s then 60 s on the interactive queue, and 60 s then 300 s on the bulk queue. The interactive queue waits less because a person is watching the chat composer, which stops polling an attachment after 10 minutes. The schedules live in RETRY_DELAY_SECONDS in run_document_worker.py, and the last value of each repeats for any later retry a higher --max-retries allows. A RateLimited retry waits too, because nothing feeds a provider's 429 back into the rate budget, and a retry at once would run into the same limit. max_retries (3 by default, the worker's --max-retries) is one budget for the whole item, counted across every step. A retry waiting out its delay counts in the queue depth as delayed, which the abandoned-runs reconciler and the Scraping Manager dashboard both read, so a waiting retry never reads as an empty queue.

Categorization is two-tier. Producers own it at the raise site: every parser (the DocumentParser contract in content_extraction/interfaces.py), the LibreOffice converter, and the dispatch boundary raise a categorized exception chosen by the type of the underlying failure - a corrupt file or missing system dependency is Fatal, a Gemini quota error is RateLimited, a poppler timeout is Transient. A provider error has one rule, as_categorized_error(exc, *, context="") in services/error_handling/exceptions.py, which the Gemini OCR parser, the PDF page loops and the runner all call. It returns an error that is already categorized as itself, files an instance of UNRETRYABLE_EXCEPTION_TYPES as Fatal, files a litellm.RateLimitError or any error whose text names a rate limit, a quota or 429 as RateLimited, and files everything else as Transient (when in doubt, give it one more try; the state machine still caps total attempts). context only prefixes the message, such as "Gemini OCR: ...", so a log line says where the error came from. categorize_exception(exc) reads the kind of whatever as_categorized_error returns, so the two cannot disagree.

The runner's _categorise is the backstop for an exception that escapes a step uncategorized. It files the exception through as_categorized_error, and it also owns one rule of its own: a Transient answer whose text names "schema", "json", "parse" or "validation" becomes SchemaInvalid, because that text means malformed LLM output. That rule never applies to an error a producer already categorized, so a Gemini server error whose body mentions JSON stays Transient. An escape means some producer violated the contract, so the runner logs a warning naming the original exception type and increments the ingest.uncategorised_error_fallback metric.

A scanned page is never left empty. Both PDF parsers OCR each scanned page through ocr_page_with_retries in parsers/pdf/image_based_pdf.py, which retries a Transient failure after 1 s and then 2 s and raises the third failure. A RateLimited, Fatal or SchemaInvalid page error escapes on its first attempt, because the worker's retry budget owns those waits. The first page that fails stops every page that has not started yet, and the parse raises the most terminal of the failures, in the order Fatal, SchemaInvalid, RateLimited, Transient. A page fault that arrives uncategorized, such as a page image that cannot be read back from disk, is filed as Transient and counted on the same metric. So a quota error on a scanned page retries the whole file, and no PDF route ships a file as Ready with a page or a whole scanned branch missing.

UNRETRYABLE_EXCEPTION_TYPES, in services/error_handling/exceptions.py, holds the exceptions no retry can change, and as_categorized_error reads it, so the rule has one home. It holds ImportError, because a missing dependency fails the same way in the same image. It also holds litellm's ContextWindowExceededError and ContentPolicyViolationError, because litellm maps each from the provider's explicit text about the request itself, and a retry sends the same request. The rule names these two subclasses and not their base, BadRequestError, on purpose. litellm raises the base class for any Gemini error whose text contains "403" anywhere, because the check is a substring match on the whole error text, and it runs before the rate-limit check. So a rate-limit error can arrive as a BadRequestError, for example a 429 whose retry delay reads "14.034037s", and it must stay retryable. Embeddings go through litellm too, so the rule covers every provider call in the pipeline. It is a typed rule, not a guess, so it logs no heuristic warning.

Provider calls also retry inside litellm, before the pipeline sees an error. gemini_complete (clients/gemini.py) is the one Gemini wrapper for every ingestion call, and it passes num_retries (5) for failures that can heal, such as rate limits, timeouts and server errors, and a retry_policy that sends a rejected request once. So a context-window error is sent once and fails its step as Fatal on the first attempt, except in entity extraction, which catches it and skips the page's entities instead (§4). A rate-limit error is retried by litellm first, unless litellm filed it as a BadRequestError, which the retry_policy sends once. If it still fails, as_categorized_error makes it RateLimited when its text names a rate limit, a quota or 429, and Transient otherwise, on every path including OCR; either way the item is republished within its max_retries budget. An invalid API key raises AuthenticationError, which the rule does not name, so it stays Transient and is retried, because it heals when an operator fixes the key.

result_from_exception(exc, *, stage, ...) wraps an exception into a StepResult for the state machine to ingest. StepResult says how a step ended with one field, error_kind: an empty kind is success, and the state machine reads nothing else to choose between retry and the inbox. An empty kind beside an error_message is refused at the constructor, so a success cannot carry text describing a failure that did not happen.

DLQ behaviour: SQS messages that fail more than max_receive_count times land in the bulk DLQ; the Failed Items Inbox admin lists every item whose outcome is failed or abandoned, and the DLQ is read separately on the Pipeline Manager. A message whose item has already settled never gets that far - see document-ingestion-state-machine.md §"Idempotency" for that rule. Nothing consumes the DLQ, so an item whose message reached it is settled as abandoned by the daily reconciler, and the inbox requeue resumes it from the step it had committed. Its file and that file's chat links read failed with _ABANDONED_FILE_REASON, "Processing never started before the request expired. Upload the file again.", while the item keeps _ABANDONED_REASON, which names the expired SQS message for the operator.

Database connection loss is handled at the process level, not in the taxonomy. All of the worker's ORM traffic runs on one shared executor thread, and _process_message recycles that thread's connection at the start of every message (close_old_connections, made effective by CONN_HEALTH_CHECKS), so each message begins on a health-checked connection. A connection that dies mid-message propagates - the pipeline categorizes it as Transient, or, when the state machine's own bookkeeping is what died, the error escapes the task and SQS redelivery retries the whole message on a fresh connection.

See document-ingestion-state-machine.md §"Error categories" for the per-step decision table.

7. Observability + cost telemetry¶

Every step transition emits one DocumentIngestionRunEvent row carrying duration_ms plus error_kind / error_message on step_failed - pure lifecycle, no cost columns.

Sub-step spans¶

A step's duration says how long the step took, not where the time went. content_extracted on a scanned PDF spends most of its time in OCR and the rest in the parser, and the event row cannot tell those apart. So each step attempt records spans: run_step installs a tracker for the duration of the callable (services/llm_infra/span_tracker.py), the pipeline functions open a named span(...) around the work they do, and the tracker's per-name totals are written as DocumentIngestionStepSpan rows beside the event. The totals are unions rather than sums, so two OCR calls running at the same time count the wall time they actually occupied rather than twice it. A span opened on a thread the parser hands work to is only visible to the tracker because copy_context().run carries the current context onto it - run_in_executor does not do that on its own.

Timing figures¶

services/document_ingestion/timing/ computes every timing figure the Pipeline Manager shows, on demand from the event, item, and span rows the pipeline already writes; nothing there stores a second copy. Views import from the package, never from one of its modules.

step_latency(scope) - p50 and p95 per step, for one run (RunScope) or for a window across every run (WindowScope). Postgres computes the percentiles through a Percentile aggregate that renders PERCENTILE_CONT(f) WITHIN GROUP (ORDER BY ...), so the durations are ordered in the database rather than pulled into the request.
item_durations(run_id, item_ids=...) - two clocks per finished document plus the run's percentiles of each. Wall is queue to searchable: the item's created to the completion of its done step. Processing is the time the attempts really spent working, with the stretch the embedded and entity_extracted pair spent running concurrently subtracted once.
span_breakdown(item_ids) - an item's sub-step totals, longest first, with the parse_file container held apart from the leaves it contains. An item processed before spans existed is absent from the result, which is how it renders blank.
throughput_trend(window, now=...) - documents that reached done per bucket over the window, empty buckets included so gaps render.
active_runs(now=...) - one row per running run: how far it has got, its completion rate read from the last ten minutes rather than from the whole run, and the finish that rate projects.

TimingWindow is where the three windows the dashboard offers are spelled: 1h, 24h, 7d, each carrying its span, its bucket width, the database truncation the counts group by, and its label format.

What the Pipeline Manager shows¶

/cc_admin/pipeline/ opens on Dashboard, scoped by a window chip. It carries the health badge and its six tiles, the retry-churn and DLQ tiles, the active-runs table, the throughput trend, and per-step p50/p95 for the window. It refreshes itself every ten seconds while a run is in flight and every thirty when nothing is, so an idle pipeline costs one request roughly twice a minute, and that request is about ten queries plus one SQS GetQueueAttributes call.

Run detail (/cc_admin/pipeline/runs/<id>/) renders every run kind, scrape included. A scrape adds a crawl block reading its ScrapeRun, so the crawl clock and the ingestion clock sit on one page. The items table carries both per-document clocks and the top of each item's span breakdown; opening a row shows the full breakdown, the untruncated source URL, and the untruncated error.

Cost is recorded once per LLM call, centrally: the runner wraps each step in an attributed(f"ingestion:{step}", run_id=..., item_id=..., user_id=...) scope, and the litellm metering callback writes one ModelUsageEvent row per call with that attribution, real provider token usage, and a micro-USD price from the registry (see model-cost-tracking.md). The Pipeline Manager cost pane, run detail, recent pane, and the Scraping Manager ingest summary all aggregate those raw rows on demand through services/document_ingestion/cost_reporting.py - the one place that defines which features, statuses, and groupings count as ingestion spend. A run or step with no ledger rows renders an em dash, never $0.00: an unknown cost must not read as free.

The entity-extraction prompts are shaped to earn prompt-cache hits: the first-pass and follow-up prompts open with one byte-identical static block (role sentence + rules, prompts/scraper/entity_extraction.py), so Gemini's implicit prefix caching can fire across every extraction call. That shared-prefix structure is pinned by a drift test in apps/main_app/tests/test_entity_extraction.py.

Entity-extraction signals¶

Chunk indexing and entity extraction have different failure shapes. A file whose chunks are indexed is findable in chat whatever the extractor did; what a short extraction costs is structured lookup by name, and that failure is invisible from the outside. So the entity_extracted step writes one EntityExtractionSignal row per attempt (services/document_ingestion/extraction_telemetry.py), keyed on (source_file, run_item) so a crash-resumed step updates its row instead of duplicating it. A row carries three counts: how many entities the model's roll call named, how many list items, table rows and headings the page's markup contains, and how many classifiable documents came out. It also stores the roll call itself - one entry per listed entity, with whether a document covered it and which one. The counts say a shortfall happened; only the names say which entity was lost, and only the covering document reveals a collapse, where one document stands in for two listed names and the counts look healthy.

A row also says how the attempt ended, in outcome, an EntityExtractionOutcome with three values. extracted is a normal attempt, unusable_output means the model's output could not be used even after one corrective retry, and too_long means the provider refused the page as longer than the model's context window. Two check constraints hold the column: it takes only those three values, and any outcome other than extracted stores zero entities. A too_long row has an empty roll call and keeps the page's real item count. The column replaced the used_fallback boolean, which stays in the table with DEFAULT false for one release so the previous release can still write rows during a rollout, and a later release drops it.

Only two conditions raise an alert, because only those two are things a machine can judge. Either the extractor could not account for every entity it listed, or nothing classifiable survived at all. EntityExtractionSignalQuerySet.flagged() is the sole definition, shared by the nightly sweep and the admin filter. A too_long or unusable_output row stored no entities, so the second condition flags it.

Deliberately absent is any comparison between the counts. One document can legitimately cover several listed names, so a healthy page routinely extracts fewer documents than it listed, and a rule built on that arithmetic would alert on working pages. Deciding that a page looks far bigger than the extractor claimed needs a person, so the review surface sorts by the gap between page items and listed entities and leaves the judgement there. That gap is annotated by EntityExtractionSignalQuerySet.with_item_gap(), which the review queue and the Django admin changelist both sort by. A too_long row has no roll call to measure a gap against, so its gap reads 0, and it ranks below every row with a real gap instead of above them all.

Read the alert for what it is: silence means nothing tripped those two conditions, not that nothing was lost. is_complete is a database-computed column: Postgres reads the stored roll call and reports true when no entry is left unmatched, so the flag can never disagree with the evidence it summarises. An empty roll call therefore reads complete, which is what an attempt written before the roll call existed carries; the review tool says so rather than claiming every name matched. A roll call that under-names the page in the first place also reports complete - that is the original failure in this issue, and it is caught by eye on the review surface, not by the sweep.

sweep_under_extracted runs nightly (07:00 UTC, infrastructure/app/schedules.tf), posts one Slack digest of counts, and stamps the rows it reported so a file is named once. The digest counts too_long rows under their own "Too long to extract" field, and its "Nothing extracted" field leaves them out, so no row is counted twice. Together with the ingest.entity_extraction_too_long metric, that field says how often long files lose their entities, which is the number that decides whether long-text extraction is worth building. Operators review in the Extraction Review tool (/cc_admin/extraction-review/, apis/admin_extraction_review_apis.py), reached from the Admin Center tile and from the digest link. It exists because ranking files by a number is not error analysis: the tool puts the file's indexed text beside the model's roll call, marks every listed name matched or missing, and lets an operator click a missing name to find it in the page - which is what turns "the counts look wrong" into "this entity was on the page and we lost it". A second tab lists the entity rows that actually persisted, and the right rail carries the four actions. A too_long row wears a "too long" pill in the queue, and its verdict says the text is too long for the extraction model, that the file is still searchable, and that reprocessing will not change it, with no note about a gap. The detail pane's Outcome row names the outcome, and its Complete row follows one rule: no roll call means unknown. An empty roll call makes is_complete true, so a too-long row, an unusable output that listed nothing and a row written before roll calls existed would all read as complete. The model's roll_call_complete property returns None for an empty roll call. The Complete row then reads "unknown", and the Django admin reads "Unknown" in its Complete column, its change form and its Complete filter. The filter and the column's sort both read ROLL_CALL_COMPLETENESS, the same rule as a query expression, so an unknown row groups on its own instead of sorting as complete. An unusable output that did list entities keeps its roll call, so its Complete row reads yes or no. When the roll call is empty, the Roll Call pane says why for each outcome: a too-long text was refused and reprocessing will not change that, an unusable output named no entities, and an older extracted row predates the roll call and gets one on reprocessing. The page text is the stored parse (SourceFile.parsed_file - exactly the markdown the extractor read) when the file has one. For files ingested before the parse was persisted it falls back to a labelled reconstruction from DocumentChunk, and a file with neither still gets the "nothing stored, not searchable" diagnostic - which also survives parse-first: a parse with zero chunks renders the text plus that warning.

The four actions - acknowledge, re-open, tag for eval, reprocess - live in services/extraction_review.py and are shared with the Django admin changelist, which stays registered for raw model access. Acknowledging suppresses the file, not the attempt, so re-scraping an acknowledged file does not re-flag it; re-open therefore clears every acknowledged row on the file, because an undo scoped to one row would leave a sibling still suppressing it and report success anyway.

A reprocess rebuilds a file through the same message kind it was ingested as, chosen by build_reprocess_message. A conversation attachment gets an AttachmentMessage built from its owner and its live ConversationAttachment link, which names its conversation once it has one, so it keeps the chat-scoped ingest profile (no entities, no summary chunk) and its cost is still attributed to the person who uploaded it. The queue is chosen by the run kind, not the message kind, so a reprocess still goes to the bulk queue. An attachment with no live link, or with no s3_key, has nothing to rebuild that message from and is reported unroutable rather than re-ingested as a folder upload.

Two limits are worth knowing. A page naming more than 192 entities (MAX_RESPONSE_DOCUMENTS × MAX_PASSES) cannot complete within the pass budget, so very large directory pages are permanently incomplete by construction; the digest and the review tool both mark those separately so they read as a known ceiling rather than a fault. And the table grows by one row per extraction attempt with no janitor yet. Storing the roll call made each row substantially larger - a directory page carries up to 512 entries of JSON rather than a handful of integers - so a retention job matters more than it did when the rows were only counts.

OTel traces stitch across the worker's step calls, the LLM client calls, and Postgres queries. The agent-run tagger scopes spans to actual agent loops. Traces export to the in-DB span store that powers the agent-traces admin. See observability-stack.md.

8. Local environment¶

Docker Compose (docker-compose.yaml at repo root) brings up:

Service	Image	Purpose
`db`	pgvector (PostgreSQL 18)	Schema + pgvector + HNSW indexes
`redis`	Redis	Rate-budget coordinator, breaker state, cache
`s3`	Versity Gateway	Local S3 server (uploads + raw source files)
`s3-init`	python	One-shot `manage.py prepare_local_upload_bucket`: creates the upload bucket and its browser-upload CORS rule
`elasticmq`	ElasticMQ	SQS-compatible (interactive + bulk queues)
`tailwind`	node	CSS build watcher
`web`	python	Django ASGI
`worker`	python	The ingestion worker (`manage.py run_document_worker`)
`dozzle`	dozzle	Container log viewer

Container names are the service name prefixed with the compose project (campuscore_web in the main checkout; cc-<slug>_web in a per-worktree stack - see parallel-dev-stacks.md), so address services with docker compose exec -T <service> rather than fixed container names.

No Lambda in local. The worker calls Gemini and OpenAI directly. Streaming chat needs ASGI - run via uvicorn campus_core.asgi:application (handled by the web service).

9. Decisions recorded¶

Compact list of architectural decisions that shaped the current shape. Not a plan - these are the shipped choices, captured here so future engineers don't re-litigate them without context.

Workers, not Lambda, do the LLM work. Direct in-process Gemini calls from ECS Fargate. Lower per-doc latency, simpler observability, one execution environment. Lambda was the original plan; never wired up; cleanup removed all of it (see §10).
Two queues with priority. Interactive (chat attachments) drains before bulk (scrapes, folder uploads, connector imports, reprocesses). Stops a 25k-page bulk job from blocking a single-document user request.
Per-file state machine with idempotent transitions. A worker crash mid-file resumes from the last committed step. Each step's success is a DB write + an event; restart finds the right resume point.
Per-step transitions, not opaque mega-steps. Each pipeline step (fetch_source, extract_content, package_chunks, persist_chunks, embed_chunks, extract_entities, persist_entities, finalize_indexes) is its own state-machine transition with its own event pair and its own cost attribution; the retry budget belongs to the item.
Categorized errors over try/except-Exception. The 4-way taxonomy makes "retry vs Failed Items Inbox" a typed decision, not a guess. Producers categorize at the raise site (parsers, converter, dispatch), and a provider error goes through the one shared rule, as_categorized_error; the runner's _categorise is a logged backstop with its own schema-text rule, an instance of UNRETRYABLE_EXCEPTION_TYPES fails at once, and other unknown exceptions default to Transient so a flapping provider doesn't terminate.
Centralized rate-budget coordinator (Redis Lua) over per-process backoff. Multiple worker tasks share one bucket per provider; total RPS to Gemini/OpenAI is bounded regardless of how many workers exist.
Memory watermark admission control. SQS pulls pause at 0.85 container memory, resume at 0.65. Embedding bursts no longer OOM-kill the worker.
Failed Items Inbox over silent retries. Items with SchemaInvalid or Fatal surface in an admin proxy with their full error context. Operators triage; the worker doesn't loop on them.
Cost recorded once, centrally. Every LLM call is metered into the ModelUsageEvent ledger by the litellm callback, attributed per step by the runner's attributed() scope; the Pipeline Manager aggregates the raw rows on demand via services/document_ingestion/cost_reporting.py.
One unified DocumentIngestionRun model for every ingestion kind. One admin screen lists every scrape, folder upload, chat attachment, connector import, and manual reprocess. No per-source admin variants.

10. What we evaluated but didn't ship¶

Recorded so future engineers don't re-introduce these without remembering why we don't have them.

Modal (third-party GPU/compute platform). Predecessor of the current shape. Removed because it required every BYOC client to set up a separate Modal billing relationship - at odds with the "deploy to your cloud" promise. Plain Python in ECS Fargate replaced it.
AWS Lambda execution backend. The original design split heavy LLM work to container-image Lambdas (extract-content + extract-typed) behind a feature flag. Built end-to-end (handlers, LambdaClient, Terraform module, IAM, ECR repo, SSM Gemini-key) but never wired into the pipeline. The state-machine rework collapsed the dispatcher into direct in-process calls and the Lambda path was never restored. We removed it entirely (commit cleanup) because (a) for a single document Lambda only adds latency - the Gemini call dominates either way; (b) for throughput the real ceiling is Gemini's rate limit, not worker concurrency, so Lambda's scale-out doesn't help; (c) the operational overhead (separate deploy pipeline, IAM, monitoring across two runtimes) wasn't worth it at pilot scale.
S3 staging hand-off (*-ingest-staging bucket + payload_s3_key event field). Existed to ferry Lambda content-extraction outputs back to the worker without hitting Lambda's 6 MB sync-response cap. Gone with the Lambda removal - the worker holds outputs in-process now.

If a future workload - burst-heavy with Gemini quota to spare, or a need for execution isolation per document - makes scale-out worth re-evaluating, the first lever is horizontal ECS worker scaling (raise desired_count), not re-introducing Lambda. One execution environment beats two.