Skip to content

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:

  1. A trigger creates a DocumentIngestionRun + one DocumentIngestionRunItem per file (DB rows; durable from this point). The DocumentProcessingMessage itself is built only by the per-kind factories in trigger.py - build_scrape_message, build_folder_message, build_attachment_message, build_connector_file_message - so every message states its source_type explicitly and the name / URL / content-type rules live in one place. The two chat-scoped source types (conversation_attachment and connector_file, grouped behind DocumentProcessingMessage.is_chat_scoped) 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. A connector_file job processes a shared copy: it carries user_id=None, 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 scrape factory's URL convention: metadata["source_url"] is canonical (the scraper stamps it at save time), metadata["url"] is the legacy key; readers use a source_url-first fallback.
  2. The TwoQueueDispatcher publishes each item onto either the interactive or bulk SQS queue.
  3. 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.
  4. The DocumentIngestionStateMachine drives each item through ordered, idempotent step transitions; every transition is a state-machine row write + a DocumentIngestionRunEvent (cost, duration, error kind).
  5. 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.
  6. On terminal failure the item lands in the Failed Items Inbox (an admin proxy over rows in error_kind ∈ {SchemaInvalid, Fatal}). On retryable failure the worker backs off and retries up to a per-step budget. 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.

A crash mid-file resumes from the last committed step — no LLM-token re-burn.

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 (LocalStack/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. Three tables and four 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: step_started / step_completed / step_failed / retry_scheduled. Holds duration_ms, error_kind, error_message. Cost lives in the ModelUsageEvent ledger, keyed back by run/item id.

Enums (models.TextChoices):

  • DocumentIngestionRunKind — scrape / folder_upload / chat_attachment / connector_import / manual_reprocess.
  • DocumentIngestionRunStatus — pending / running / completed / failed / cancelled / abandoned.
  • DocumentIngestionStep — enqueued / fetched / content_extracted / chunked / chunks_persisted / embedded / entity_extracted / entities_persisted / indexed / done, plus the terminal failed and abandoned. The happy-path order is declared once as STAGES in services/document_ingestion/steps.py; the state machine and the runner both derive from it.
  • DocumentIngestionErrorKind — "" (none) / RateLimited / Transient / SchemaInvalid / Fatal. Mirrored at the application layer by step_result.ErrorKind.

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 typed entities (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. Every attempt records an EntityExtractionSignal row - see §7.
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.

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

The load-bearing source stamps are typed: extract_content builds an IngestProvenance (with nested AttachmentScope / ConnectorFileScope for the chat kinds) from the job onto ctx.provenance, and downstream steps read those fields — never string keys. Every chunk carries source_type (the job's IngestSourceType discriminator, rendered unconditionally) plus original_filename; the remaining stamps are per kind — s3_key/knowledge_folder_id for folder uploads, source_url for scrapes, the conversation scope keys for attachments, the connector identity keys for connector files — and a provenance validator rejects a scope block that disagrees with the discriminator. ctx.extracted_metadata holds only the parser's pass-through dump; StepContext.merged_metadata() is the sole renderer of the persisted dict (chunk metadata JSON and the SourceFile.metadata merge), and its exact per-ingest-kind shape is frozen by tests/test_ingestion_metadata_contract.py — attachment search scoping, the SourceFile.original_filename property, and the URL backfills all read that persisted shape.

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). 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).

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 also owns ErrorKind):

Exception ErrorKind State-machine reaction
RateLimitedError RateLimited Acquire rate budget, retry same step.
TransientError Transient Exponential backoff, retry 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.

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. The backstop is categorize_exception(exc) (and the runner's _categorise), which maps any exception that escapes uncategorized: ImportError → Fatal, then a message heuristic, then a default of Transient (when in doubt, give it one more try; the state machine still caps total attempts). A heuristic firing 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.

result_from_exception(exc, *, stage, ...) wraps an exception into a StepResult for the state machine to ingest.

DLQ behaviour: SQS messages that fail more than max_receive_count times land in the bulk DLQ; the Failed Items Inbox admin lists everything in error_kind ∈ {SchemaInvalid, Fatal} plus everything still pinned in the DLQ.

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.

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.

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.

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.

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. 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. 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.

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 Postgres + TimescaleDB Schema + pgvector + HNSW indexes
redis Redis Rate-budget coordinator, breaker state, cache
minio MinIO S3-compatible (uploads + raw source files)
elasticmq ElasticMQ SQS-compatible (interactive + bulk queues)
localstack LocalStack Generic AWS services (CloudWatch, etc.)
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, its own retry budget, and its own cost attribution.
  • 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); categorize_exception and the runner heuristic are a logged backstop, and unknown exceptions default to retry-once (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 DOCUMENT_PROCESSING_BACKEND=lambda 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.
  • LocalStack Lambda emulation for parity. Local dev no longer needs Lambda emulation. LocalStack is still used for other AWS services (CloudWatch, etc.); ElasticMQ + MinIO handle SQS + S3.

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.