rag-context-worker — business features

Makes an uploaded document searchable, then hands the relevant passages to whoever is writing the course · ← platform hub · entry points verified against tools/feature-docs/out/rag-context-worker.json

1. What it is

When an admin uploads a manual, this worker is what turns those bytes into something the rest of the platform can actually use: it sends the file to RAGFlow to be parsed and chunked, tracks how far that got so the admin sees a real progress bar, and afterwards answers questions of the form "give me the passages in this tenant's documents that are about forklift battery changes, within this token budget". Three services consume those passages over HTTP: authoring-service-v2, content-worker and suggest-pipeline-worker.

Who uses itauthoring-service-v2 (NATS + HTTP), authoring-content-worker (HTTP, edge H6), suggest-pipeline-worker (HTTP, edge H10). No admin or mobile client reaches it directly.
RuntimeFastAPI (5 business routes + 2 probes) + 4 JetStream consumers · Deployment rag-context-worker, replicas: 1, strategy: Recreate (rag-context-worker/k8s/deployment.yaml:9-16) — its durables are single-binding, so two pods cannot coexist
DatabaseOwn Postgres database rag_context_db (ARCHITECTURE.md §3.4 — sole writer); alembic, migrated by an init container
RedisNone.
NATS streamsConsumes on AUTHORING_RAG (authoring.rag.>, 7 days, 1 MB messages). It also owns the AUTHORING_USAGE stream definition (nats/streams/authoring-usage.json) that five workers publish cost rows onto — a platform-wide stream defined in this service's repo directory.
External APIsRAGFlow (in the ragflow namespace), Modal MinerU parser warm-up, S3 for context blobs, Cloudflare CDN for document downloads (ARCHITECTURE.md §3.6)
Entry points7 HTTP routes (5 business + 2 probes) · 4 NATS consumers · 6 published subjects · 5 in-process tasks

Its HTTP paths are unversioned and live at the root

/documents, /documents/{id}, /contexts/{id}, /contexts/{id}/blob, /contexts/{id}/stream — no /v1, no /internal prefix, unlike every other Oper service (ARCHITECTURE.md §3.2 notes the platform's two prefix conventions; this service follows neither). Callers hardcode the bare paths, so any future versioning is a coordinated change across three services. Tenant scoping comes from the X-Tenant-Id header; when the internal key is unset, that degrades to header-spoofable access (bug-hunt-reports/rag-context-worker.md #5).

2. Feature map

flowchart LR
  ADMIN["Admin uploads a document"] --> AM[asset-manager]
  AM -. "upload.completed" .-> AV2[authoring-service-v2]
  AV2 -. "rag.indexing.requested" .-> F1["Index a document"]
  F1 --> RF["RAGFlow parse + chunk"]
  F1 -. "indexing progress / ready / failed" .-> AV2
  AV2 -. "rag.context.requested" .-> F2["Retrieve context for a topic"]
  F2 -. "context.ready / failed" .-> AV2
  F2 --> BLOB["context blob in S3"]
  BLOB --> HTTP["GET /contexts/{id}/stream"]
  HTTP --> CW["content-worker + suggest-pipeline"]

3. Features

Admin

Watch a document become usable live

admin

Parsing a 200-page PDF takes minutes, and an admin staring at a silent spinner assumes it broke. This worker reports parsing progress as it goes and a final ready-or-failed verdict with a page count, which authoring-v2 pushes to the admin's browser. The same document row can also be polled over HTTP as a fallback.

Entry points
publishes authoring.rag.indexing.progress / .ready / .failed; GET /documents/{document_id}
Touches
rag_documents, RAGFlow
Related
A transient NATS failure on a progress tick used to fail an otherwise healthy index — fixed (bug-hunt-reports/rag-context-worker.md #7)
Evidence
publishes app/publishers/events_publisher.py:40,48,54,57 · consumed at authoring-service-v2/app/consumers/register.py:21-23, forwarded to the browser at authoring-service-v2/app/consumers/rag_consumer.py:70

Employee (mobile)

None.

Internal (other services)

Index a document so it can be searched live

authoring-service-v2

The moment an upload completes, authoring-v2 hands the file here. The worker creates its own document record, downloads the bytes from the CDN, streams them to RAGFlow for parsing and chunking, and polls until the document is queryable — optionally kicking off a MinerU parse in parallel for better layout handling. Nothing downstream — no outline, no lesson, no suggestion — can use a document until this finishes.

Entry points
authoring.rag.indexing.requested (durable rag-indexing)
Touches
rag_documents, RAGFlow datasets, Modal MinerU, CDN download, S3
Related
Global concurrency 4 (INDEXING_GLOBAL_CONCURRENCY, k8s/configmap.yaml:36); poll ceiling 600 s (:27). The document row carries both RAGFlow ids because extraction-worker needs them to read chunks (app/api.py:132-137).
Evidence
consumer app/consumers/indexing_consumer.py:194 · publisher authoring-service-v2/app/consumers/asset_consumer.py:83 (on authoring.asset.upload.completed) and the index-document tool (libs/oper-tools/oper_tools/catalog.json:1202)

Retrieve the passages that matter for a topic live

authoring-service-v2

The retrieval half. Given a topic and the documents a tenant owns, the worker queries RAGFlow, assembles the best chunks into one context blob inside a token budget, stores it, and announces a context id. Everything that writes course material then pulls that blob — which is what makes generated lessons cite the customer's own manual instead of the model's general knowledge.

Entry points
authoring.rag.context.requested (durable rag-retrieval) → authoring.rag.context.ready / .failed; blob served by GET /contexts/{context_id}/blob and /stream
Touches
contexts, context_chunks, RAGFlow query, S3 blob
Related
Known gap: retrieval is scoped by RAGFlow dataset rather than by the requested doc_ids, so it can be over-broad within a tenant (bug-hunt-reports/rag-context-worker.md #2)
Evidence
consumer app/consumers/retrieval_consumer.py:73 · publisher authoring-service-v2/app/workflows/doc_to_module.py:357 and the retrieve-context tool (libs/oper-tools/oper_tools/catalog.json:1250) · results consumed at authoring-service-v2/app/consumers/register.py:24-25

Serve the context blob to the writers live

content-worker · suggest-pipeline-worker · authoring-service-v2

A context blob is roughly 32 KB of plain text, so it travels over HTTP rather than through a message. content-worker fetches it before writing each lesson, suggest-pipeline before drafting an outline, and authoring-v2 redirects the browser to a presigned URL when a draft needs to show its sources. All three treat a failure as "generate without retrieved context" rather than an error, so a blip degrades quality instead of breaking a run.

Entry points
GET /contexts/{context_id}/stream (text), GET /contexts/{context_id}/blob (307 to presigned S3)
Touches
contexts, S3
Related
HTTP edges H5, H6 and H10 in ARCHITECTURE.md §3.2; gzip on responses over 1 KB (app/api.py:76)
Evidence
routes app/api.py:172,183 · callers authoring-content-worker/app/services/context_fetcher.py:48, suggest-pipeline-worker/app/pipeline/outline.py:97, authoring-service-v2/app/core/http.py:706

Forget a document, or stop indexing one dead

authoring-service-v2

Two consumers exist for operations an admin would obviously want: delete a source document (removing it from RAGFlow and this worker's records) and cancel an index that is still running. Both are fully implemented here and documented in this service's own contract — and nothing in the platform publishes either subject, so neither has ever run in production. A deleted document therefore stays in the vector store, and a cancelled upload keeps parsing.

Entry points
authoring.rag.document.delete.requested (durable rag-delete), authoring.rag.indexing.cancel.requested (durable rag-cancel)
Touches
rag_documents, RAGFlow; the in-flight indexing task for cancel
Related
See §8 for the search evidence. Both consumers also carry an ack-before-work defect (bug-hunt-reports/rag-context-worker.md #1 fixed, #3 open) — worth fixing only if a publisher is actually built.
Evidence
consumers app/consumers/delete_consumer.py:53, app/consumers/cancel_consumer.py:34 · documented as authoring-published in contracts/authoring-service.md:16-17, but no publisher exists in any service

Report what indexing and retrieval cost live

authoring-service-v2

Parsing and embedding a large document is not free, so each indexing and retrieval operation reports a usage row on the same stream the generating workers use. That is how the cost of preparing a document shows up next to the cost of writing its lessons in a tenant's bill.

Entry points
authoring.usage.event (publish only)
Touches
AUTHORING_USAGE — whose stream definition this service owns
Related
Five of the six authoring pipeline workers publish this subject; authoring-v2's usage consumer is its only reader.
Evidence
app/publishers/events_publisher.py:69 · stream nats/streams/authoring-usage.json · consumed at authoring-service-v2/app/consumers/register.py:68

Background

Long-running work kept off the delivery loop live

background

Indexing and retrieval both outlast any sane message deadline, so each delivered message is run as its own task with a keepalive that extends the acknowledgement deadline while it works, and the message is only acknowledged after the work is done. That is why a pod restart mid-parse resumes instead of losing the document — and why the concurrency cap matters more than it looks.

Entry points
tasks _run_one, process_indexing_event, process_retrieval_event, parse_and_store (all per-message, none scheduled)
Touches
RAGFlow, MinerU, S3, own database
Related
The eager MinerU parse is fire-and-forget alongside the RAGFlow upload and is flag-gated (EAGER_PARSE_ENABLED, k8s/configmap.yaml:51)
Evidence
app/consumers/indexing_consumer.py:103,186,271, app/consumers/retrieval_consumer.py:88

NATS bootstrap that never gives up live

background

If the durables cannot be bound, the worker closes and retries every 30 seconds rather than crash-looping. Because that retry loop swallows failures, readiness deliberately checks NATS as well as the database — otherwise a pod that consumes nothing would answer health checks cheerfully forever.

Entry points
lifespan task connect_nats_with_retry
Touches
All four durables
Related
A durable already bound by another pod raises on purpose — a pod that binds nothing must not report success (app/consumers/subscribe_helper.py:100-106)
Evidence
app/main.py:115 · readiness rationale app/api.py:100-112

4. API reference

MethodPathAuthFeatureCallersVerdict
GET/documents/{document_id}X-Tenant-Id (+ X-Oper-Key when set)Poll one document's indexing state and RAGFlow idsauthoring-service-v2 — app/core/http.py:699 (edge H5)live
GET/contexts/{context_id}/blobX-Tenant-Id307 redirect to the presigned blobauthoring-service-v2 — app/core/http.py:706live
GET/contexts/{context_id}/streamX-Tenant-IdServe the context text inlinecontent-worker app/services/context_fetcher.py:48 (H6); suggest-pipeline app/pipeline/outline.py:97 (H10)live
GET/documentsX-Tenant-IdPaged document list for a tenantNo in-repo caller. contracts/authoring-service.md:34 assigns it to the admin document-picker UI, which is not in this repo.suspect
GET/contexts/{context_id}X-Tenant-IdContext metadata: chunk count, size, similarity range, expiryNo in-repo caller; documented in contracts/README.mdsuspect
GET/healthnoneLiveness (DB check)kubelet livenessProbe (k8s/deployment.yaml:108-112)live
GET/readynoneReadiness — DB and all four durables boundkubelet readinessProbe (k8s/deployment.yaml:118-122)live

5. Async contracts

All four durables are single-binding by design — no queue group. The helper's long comment explains why: nats-py rejects a queue name that differs from the durable before it even reaches the server, so the four consumers were always created without a deliver group, and matching that reality is the honest fix. It is also why the Deployment is replicas: 1 with strategy: Recreate: a second pod binding the same durable raises on purpose (app/consumers/subscribe_helper.py:66-106).

Consumes

SubjectStreamDurablePublished byFeatureVerdict
authoring.rag.indexing.requestedAUTHORING_RAGrag-indexing (max_ack_pending = indexing concurrency)authoring-service-v2 — app/consumers/asset_consumer.py:83; tool index-documentIndex a documentlive
authoring.rag.context.requestedAUTHORING_RAGrag-retrievalauthoring-service-v2 — app/workflows/doc_to_module.py:357; tool retrieve-contextRetrieve contextlive
authoring.rag.document.delete.requestedAUTHORING_RAGrag-deleteNobody. Contract says authoring-v2 (contracts/authoring-service.md:16); no publisher exists in any serviceForget a documentdead
authoring.rag.indexing.cancel.requestedAUTHORING_RAGrag-cancelNobody. Contract says authoring-v2 (contracts/authoring-service.md:17); no publisher exists in any serviceCancel an indexdead

Publishes

Earlier scans reported this service as publish-free; it is not. It reports every indexing and retrieval outcome back to authoring-service-v2 — those six subjects are how an admin's upload progress and a draft's context ever become visible.

SubjectConsumed byFeatureVerdict
authoring.rag.indexing.progress events_publisher.py:40,48authoring-service-v2 rag_consumer.on_indexing_progress (register.py:21) → WS "parsing %"Watch a document become usablelive
authoring.rag.indexing.ready events_publisher.py:54authoring-service-v2 (register.py:22)Document indexed — unblocks extractionlive
authoring.rag.indexing.failed events_publisher.py:57authoring-service-v2 (register.py:23)Indexing failedlive
authoring.rag.context.ready events_publisher.py:60authoring-service-v2 (register.py:24)Context available — carries the context idlive
authoring.rag.context.failed events_publisher.py:63authoring-service-v2 (register.py:25)Retrieval failedlive
authoring.usage.event events_publisher.py:69authoring-service-v2 usage_consumer (register.py:68) on AUTHORING_USAGECredit meteringlive

Background jobs

JobScheduleWhat it doesVerdict
connect_nats_with_retry (app/main.py:115)Once per pod start, retrying every 30 sBinds all four durables; failures are swallowed, which is why /ready checks NATSlive
_run_one (app/consumers/indexing_consumer.py:186)Not scheduled — one task per indexing message, under a semaphoreRuns one document index without blocking the delivery looplive
process_indexing_event (app/consumers/indexing_consumer.py:271)Not scheduled — inside the keepalive wrapperThe index itself; the keepalive extends the ack deadline so a long parse is not redeliveredlive
parse_and_store (app/consumers/indexing_consumer.py:103)Not scheduled — fire-and-forget per document when EAGER_PARSE_ENABLEDMinerU parse alongside the RAGFlow uploadlive
process_retrieval_event (app/consumers/retrieval_consumer.py:88)Not scheduled — inside the keepalive wrapperRAGFlow query + blob write + DB write for one retrievallive

6. Data it owns

Sole writer of rag_context_db (ARCHITECTURE.md §3.4). No other service reads or writes it, and this worker writes no other service's tables — everything it shares travels as an event, an HTTP response or an S3 object.

TableWhat it holdsWritten by
rag_documentsOne row per uploaded source document: tenant, filename, indexing status and error, page count, and the RAGFlow dataset + document ids that everything downstream needs to read chunks.This worker only (alembic/versions/2025_03_07_1000-initial_schema_create_tables.py:24)
contextsOne row per retrieval: the topic, the documents asked for, a request hash for reuse, chunk count, byte size, similarity range, blob location, and an expiry (30 days by default).This worker only (alembic/versions/2026_05_28_1200-rag_worker_v2_contexts.py:63)
context_chunksThe ordered chunks inside a context, each with its source document, RAGFlow chunk id, similarity and byte offset into the blob — the provenance trail behind a generated lesson's citations.This worker only (alembic/versions/2026_05_28_1200-rag_worker_v2_contexts.py:120)
ai_drafts droppedA v1 draft table, deleted by the v2 migration — named here so nobody goes looking for it.Nobody (alembic/versions/2026_05_28_1200-rag_worker_v2_contexts.py:144-146)

Context text is not in Postgres: the blob lives in S3 and the row points at it, which is why the HTTP blob route can hand out a presigned redirect instead of proxying bytes.

7. Dependencies

flowchart LR
  AV2[authoring-service-v2] -. "indexing · context requested" .-> RAG[rag-context-worker]
  AV2 -- "HTTP H5: document meta · blob" --> RAG
  CW[content-worker] -- "HTTP H6: context stream" --> RAG
  SPW[suggest-pipeline-worker] -- "HTTP H10: context stream" --> RAG
  RAG -. "indexing + context progress/ready/failed · usage" .-> AV2
  RAG -- parse + query --> RF["RAGFlow (ragflow ns)"]
  RAG -- layout parse --> MU["Modal MinerU"]
  RAG -- "blobs · downloads" --> S3["S3 + Cloudflare CDN"]
  RAG -- "documents · contexts · chunks" --> DB[("rag_context_db")]

8. Dead-code verdicts

Every entry point with no in-repo caller. Deleting is a separate decision — see the hub roll-up.

Entry pointKindVerdictEvidence
authoring.rag.document.delete.requested (durable rag-delete)NATS consumerdeadConsumer exists: app/consumers/delete_consumer.py:53, subject at :14. No publisher, and none possible: repo-wide search for the subject string across .py/.yaml/.json/.md returns only this worker, its own contracts/ docs, its smoketests/smoke_b_to_f.py:385,437 (tests do not count) and New-Design planning notes. authoring-v2 is the only service that publishes on AUTHORING_RAG and has no send site for it (its tool catalog exposes only index-document and retrieve-context). So deleting a document never reaches RAGFlow.
authoring.rag.indexing.cancel.requested (durable rag-cancel)NATS consumerdeadConsumer exists: app/consumers/cancel_consumer.py:34, subject at :11. No publisher: same search, same result — only this worker, contracts/authoring-service.md:17, smoketests/smoke_b_to_f.py:443 and design notes. Note authoring-v2 does publish a cancel for tool calls (authoring.tool-call.cancelled, which image-worker consumes) — it simply never wired the RAG-specific one, so cancelling an upload leaves the parse running and billable.
GET /documentsHTTP routesuspectSearched every service's client code for the path and for a RAG_WORKER base URL usage: the only in-repo consumers of this service call /documents/{id}, /contexts/{id}/blob and /contexts/{id}/stream. This service's own contract assigns the list route to the admin document-picker UI (contracts/authoring-service.md:34), and the admin web app is not in this repository — so it may well be live in production. Resolved by checking the admin frontend or the route's access logs.
GET /contexts/{context_id}HTTP routesuspectSame search: no in-repo caller. Documented as the metadata read in contracts/README.md:134-143; every in-repo consumer goes straight to the blob instead. Resolved the same way — a frontend check or access logs.

All six published subjects have a named authoring-v2 consumer, and the indexing/retrieval consumers have named authoring-v2 publishers.

9. Sources