Partner event API

The consumer accepts doc.uploaded without aiDocumentId. Aceh RAG generates that ID after persisting its documents row and returns it in stage-status events; it is not an input requirement.

This page is the self-contained contract for the partner backend integration at /docs/api/partner-events.

The consumer is a server-side partner backend that owns crawlers, a DMS, object storage keys, and Kafka job tracking. It is not a browser API. Do not send these events from browser code.

Transport and ownership

Aceh RAG consumes document-upload events from Kafka, reads the referenced bytes from configured object storage, runs the existing parse, OCR, chunk, embed, and index pipeline, then emits stage-status events back to Kafka for partner job tracking.

Direction	Default topic or map	Owner produces	Owner consumes	Purpose
Partner to Aceh RAG	`doc.uploaded`	Partner backend	Aceh RAG Kafka worker	Tell Aceh RAG that bytes are available under `storageKey`.
Aceh RAG to partner	`kafka.stage_topics`	Aceh RAG Kafka worker	Partner backend	Report per-document stage progress and terminal outcomes on the destination selected by the internal routing key.

Kafka is configuration-gated and defaults off for local, test, and CI processes. A production deployment must explicitly enable and run the dedicated Kafka worker process (production: worker recreated on the deployed API image 2cc9a417 — firstmate production audit 2026-10-07). Topic names, broker list, and consumer group are configuration values, so confirm the active values for the target environment before connecting.

Setting	Default	Required deployment confirmation
`kafka.enabled`	`false`	Confirm it is enabled where the partner integration should run.
`kafka.bootstrap_servers`	`localhost:9092`	Confirm the real broker list.
`kafka.doc_uploaded_topic`	`doc.uploaded`	Confirm the input topic name.
`kafka.stage_topics`	Complete stage-to-topic map	Confirm every route and provision every destination.
`kafka.consumer_group`	`aceh-rag-ingestion`	Confirm the consumer group assigned to Aceh RAG.
`kafka.default_source_version`	`m2-kafka`	Legacy setting only; ingress uses `sourceVersion` when supplied, otherwise required `uploadedAt`.
`kafka.max_retries`	`3`	Confirm retry budget for transient fetch and pipeline failures.
`kafka.retry_backoff_seconds`	`2.0`	Confirm the base retry delay.

This contract does not specify broker security settings. Use the platform team's deployed Kafka connection profile.

Default stage-topic map

kafka.stage_topics is a complete mapping: worker configuration fails before startup if a route is missing, blank, or uses an unsupported key. The partner owns the actual DMS values and may override these defaults without changing the publisher.

Event condition	Internal configuration key	Wire `stage/status`	Progress	Default destination
Source bytes fetched	`fetched`	`extracting/done`	10	`doc.extracted`
Extraction / parse complete	`parsed`	`extracting/done`	25	`doc.extracted`
Actual OCR attempt begins	`ocr_pending`	`ocr_pending/started`	25	`doc.ocr.pending`
OCR attempt complete	`ocr`	`ocr_done/done`	45	`doc.ocr.done`
Chunks produced	`chunked`	`chunking/done`	65	`doc.chunked`
Embeddings stored	`embedded`	`embedding/done`	85	`doc.embedded`
Index success or review	`indexed`	`indexed/done or indexed/needs_review`	100	`doc.indexed`
Terminal failure	`failed`	`failed/failed`	100	`doc.indexed`
Structured-source or duplicate no-op	`skipped`	`indexed/skipped`	100	`doc.indexed`

The partner must create the six agreed DMS destinations and grant the worker producer permission before deployment; broker auto-creation is not assumed.

`doc.uploaded` input event

Publish one JSON object per document to the configured input topic. The payload fields use camelCase.

Field table

Field	Required?	Type	Validation, default, and meaning
`jobId`	Yes	UUID string	Partner `IngestionJob` identity. Used for redelivery idempotency, Kafka status-event keys, and correlation.
`documentId`	Yes	UUID string	Partner document/version identity. Default write-lock identity and `source_key` when `sourceKey` is absent.
`registryId`	No	UUID string \| null	Partner registry identity. When absent or `null` in an otherwise valid event, defaults to validated `documentId`; an invalid non-null UUID is rejected.
`storageKey`	Yes	string	Must be nonblank. Object-storage key that Aceh RAG reads to fetch the document bytes.
`fileName`	No	string \| null	Original filename used as the document title after trimming and collapsing whitespace to single spaces. For new ingestion, missing, blank, or non-string values use the storage-key-derived filename fallback; same-byte refresh preserves the stored title instead. This field does not change the storage key or parser filename; its name is provisional pending partner agreement (`kafka_ingestion.PARTNER_FILENAME_FIELD`).
`mimeType`	Yes	string	Supported MIME type. It must agree with the `storageKey` extension.
`collectionId`	Yes	UUID string	Partner collection identity. Persisted as source metadata.
`collectionName`	Yes	string	Normalized to `JDIH`, `OpenData`, `PPID`, `SatuData`, or `upload`. The authoritative mapping remains subject to partner sign-off.
`uploadedBy`	Yes	UUID string	Partner actor or service identity. Persisted as source metadata.
`uploadedAt`	Yes	ISO-8601 timestamp	Timezone-bearing upload time; default `source_version` when `sourceVersion` is absent. Always required.
`sourceType`	No	string enum \| null	If non-null, must agree with a known source `collectionName`; explicit `"JDIH"` is accepted for a region collection such as `"Provinsi Aceh"`.
`format`	No	string \| null	Legacy compatibility only. If non-null, it must agree with `mimeType` and `storageKey`.
`sourceSystem`	No	string \| null	Optional nonblank source system label, for example `jdih-portal` or `ppid-crawler`. Persisted as source metadata when non-null.
`skpaUnit`	No	string \| null	Optional nonblank SKPA or organization unit. Persisted as source metadata when non-null.
`classification`	No	string \| null	Optional nonblank classification label. Persisted as source metadata when non-null.
`category`	No	string \| null	Optional nonblank partner category label. Persisted as source metadata when non-null.
`sourceKey`	No	string \| null	Scheduled-sync stable source identity, e.g. `jdih-<portal uuid>`. Overrides `documentId` as `source_key` and lock identity.
`sourceVersion`	No	string \| null	Overrides `uploadedAt` as `source_version`; a new value preserves a changed file as a new edition.
`sourceMetadata`	No	object \| null	String-valued source provenance merged into `metadata.source`, including `tahun`, `nomor`, `wilayah`, and a citable URL.

fileName and the three scheduled-sync fields are tolerant. For sourceKey, sourceVersion, and each sourceMetadata value, non-string, blank, null, and trimmed "-" values mean absent; non-object sourceMetadata is absent. Absent source keys do not erase stored values. Other optional fields accept omission or null as unset; non-null values must be nonblank strings and satisfy their constraints. For valid events, unset registryId defaults to documentId, unset sourceType and format are inferred, and unset optional metadata is omitted.

Only these descriptive sourceMetadata keys are accepted: judul, tahun, nomor, jenis, wilayah, status_berlaku, bidang_hukum, url, pdf_url, detail_url, bentuk_singkat, catalogue_id, catalogue_source, content_type, year. All other keys are dropped with one structured warning containing names only, never values. These fields cannot control workspace assignment, replacement, storage, transport or access, for either ingestion or same-byte refresh; transport/access fields retain precedence. Unknown top-level fields are ignored by the current consumer and are not persisted as source metadata. Do not put required partner job state only in unknown fields.

Source type values

type DocumentSourceType = "JDIH" | "OpenData" | "PPID" | "SatuData" | "upload";

The field table above defines source-type resolution and the compatibility rule for explicit sourceType values with region collection names. Conflicting source types make the event malformed and produce a failed stage event.

Format values

Supported document formats are:

type DocumentFormat = "txt" | "md" | "html" | "csv" | "docx" | "xlsx" | "pptx" | "pdf" | "image";

format is a hint, not the document bytes. The current Kafka filename derivation uses the hint to force these extensions when storageKey is extensionless:

`format` hint	Derived extension
`pdf`	`.pdf`
`docx`	`.docx`
`xlsx`	`.xlsx`
`csv`	`.csv`
`html`	`.html`
`image`	`.png`

For txt, md, and pptx when format is omitted, make sure the storageKey basename already has the parseable extension, such as records/qanun.txt, records/page.md, or records/slides.pptx. An extensionless key is valid only when the declared supported mimeType resolves the format. A nonempty unsupported extension, unsupported MIME, or conflicting MIME/extension/legacy format is rejected at ingress and emits correlated failed/failed before any OBS fetch.

Required event

{
  "jobId": "550e8400-e29b-41d4-a716-446655440000",
  "documentId": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
  "storageKey": "documents/a1b2c3d4e5f6/perda-no-12-2026.pdf",
  "mimeType": "application/pdf",
  "collectionId": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "collectionName": "JDIH",
  "uploadedBy": "9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d",
  "uploadedAt": "2026-08-05T09:12:03.412Z"
}

This is valid. Aceh RAG reads the OBS object at the configured bucket with the unmodified storageKey, normalizes JDIH, and persists source_key=documentId and source_version=uploadedAt.

Full event example

{
  "jobId": "550e8400-e29b-41d4-a716-446655440000",
  "documentId": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
  "storageKey": "jdih/2026/qanun-aceh-001",
  "mimeType": "application/pdf",
  "collectionId": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "collectionName": "JDIH",
  "uploadedBy": "9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d",
  "uploadedAt": "2026-07-23T10:15:30Z",
  "sourceType": "JDIH",
  "format": "pdf",
  "registryId": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
  "sourceSystem": "jdih-portal",
  "skpaUnit": "dinas-keuangan",
  "classification": "public",
  "category": "qanun",
  "sourceVersion": "2026-07-23T10:15:30Z"
}

Because mimeType and legacy format are pdf, the extensionless storageKey basename is treated as a PDF filename for parser dispatch.

Input validation and failure behavior

Malformed messages are logged and treated as terminal for offset handling. A failed/failed event is emitted only when valid jobId and registryId can be recovered; absent registryId may use valid documentId. This malformed-message recovery differs from successful parsing: explicit invalid/null registry IDs are not replaced. Uncorrelatable failures emit no status record and need operator reconciliation.

Failure	Retry?	Stage events	Partner interpretation
Invalid JSON, non-object payload, or missing/invalid `jobId`	No	No correlated status event	Inspect producer logs and reconcile the job.
Invalid required input, explicit invalid `registryId`, collection mapping, or optional string	No	`failed/failed` only with recoverable job and registry IDs; `aiDocumentId: null`	Fix and republish; reconcile if no event can be correlated.
Object-storage fetch exhausted	Bounded internally	`failed/failed`	Fix object availability before republishing.
Pipeline exhausted after fetch	Bounded internally	`extracting/done`, then `failed/failed`; an OCR-start update may also have occurred	Fix the cause before republishing.

Permanent malformed messages fail fast. Transient object-storage and pipeline errors are retried with exponential backoff using the configured retry settings. Retry attempts do not produce separate retry stage events.

Document identity, versioning, and idempotency

Use jobId as the cross-service job identity and documentId as the source-document identity. aiDocumentId is generated only when Aceh RAG creates its documents row.

Aceh RAG uses two related identities:

In-process duplicate detection uses jobId.
Stored provenance uses source_key = sourceKey or documentId and source_version = sourceVersion or uploadedAt. Admin uploads without these overrides keep their existing SHA-256 + version dedup and fileName title.

Partner rules:

Republish the same jobId only for redelivery or retry of that IngestionJob.
Confirm whether documentId and uploadedAt are immutable for a document version before production deployment.
Track IngestionJob exclusively by jobId.

When Aceh RAG sees a duplicate jobId in the same worker process, it skips reprocessing and emits no new stage events. Migration 0007 enforces (sourceType, source_version, source_key); migration 0006 separately guards checksum + version. With sourceKey, a finished (indexed, needs_review, or archived) shared-corpus row with the same type/key/bytes in any version skips ingestion but merges supplied sourceMetadata, refreshes a usable fileName title, and rebuilds title-search labels. Identity, chunks, access columns and other metadata stay intact. The result is terminal indexed/skipped with the existing aiDocumentId and chunkCount: 0. Changed bytes use existing rules: a new version creates an edition; the same identity corrects in place. No supersede link is inferred. The lock is doc-uploaded:<sourceKey or documentId>; default locks and processed-job state remain in-process.

For source metadata, send string tahun/wilayah for filters and at least one of pdf_url, detail_url, or url for a citable bibliography URL. judul is metadata, not the displayed title. The backend contract §3 owns the single read-only existence query ("sourceType" plus top-level metadata->>'source_key', any version); no lookup endpoint is added. Deploy this AI-engine contract before running sync: older workers ignore the new fields. Consumers must branch on status, not the topic or progress; only indexed/done is success.

`StageStatusEvent` output event

The v1 serializer always emits all ten envelope keys below, including nullable values. Kafka records are keyed by UTF-8 jobId. Routing uses the private internal configuration key in the stage-topic map, not the wire stage alone: two extracting/done updates represent fetch and parse, with progress 10 and 25. There is no total ordering across topics.

JSON schema

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "StageStatusEvent",
  "type": "object",
  "additionalProperties": false,
  "required": [
    "v",
    "jobId",
    "registryId",
    "aiDocumentId",
    "stage",
    "status",
    "progress",
    "error",
    "meta",
    "ts"
  ],
  "properties": {
    "v": {
      "const": 1
    },
    "jobId": {
      "type": "string",
      "minLength": 1
    },
    "registryId": {
      "type": [
        "string",
        "null"
      ]
    },
    "aiDocumentId": {
      "type": [
        "string",
        "null"
      ]
    },
    "stage": {
      "type": "string",
      "enum": [
        "extracting",
        "ocr_pending",
        "ocr_done",
        "chunking",
        "embedding",
        "indexing",
        "indexed",
        "failed"
      ]
    },
    "status": {
      "type": "string",
      "enum": [
        "started",
        "done",
        "failed",
        "skipped",
        "needs_review"
      ]
    },
    "progress": {
      "type": "integer",
      "minimum": 0,
      "maximum": 100
    },
    "error": {
      "type": [
        "string",
        "null"
      ]
    },
    "meta": {
      "type": [
        "object",
        "null"
      ],
      "properties": {
        "ocrConfidence": {
          "type": "number"
        },
        "chunkCount": {
          "type": "integer",
          "minimum": 0
        }
      }
    },
    "ts": {
      "type": "string",
      "format": "date-time"
    }
  },
  "allOf": [
    {
      "if": {
        "anyOf": [
          {
            "properties": {
              "stage": {
                "const": "failed"
              },
              "status": {
                "const": "failed"
              }
            }
          },
          {
            "properties": {
              "stage": {
                "const": "indexed"
              },
              "status": {
                "enum": [
                  "done",
                  "skipped",
                  "needs_review"
                ]
              }
            }
          }
        ]
      },
      "then": {
        "properties": {
          "registryId": {
            "type": "string",
            "minLength": 1
          }
        }
      }
    }
  ]
}

Field meanings

Field	Type	Meaning
`v`	integer	Envelope version, always 1.
`jobId`	string	Partner IngestionJob linkage and Kafka key.
`registryId`	string \| null	Partner registry correlation; valid ingress defaults to `documentId` when absent or `null`; malformed-message recovery uses that fallback only when absent. Required non-null for terminal events.
`aiDocumentId`	string \| null	AI-engine `documents.id`; null before persistence or on pre-persistence failure.
`stage`	enum	Partner wire stage, distinct from internal routing keys.
`status`	enum	Wire outcome; progress 100 alone does not mean success.
`progress`	integer	Milestone percentage 0–100; terminal outcomes use 100.
`error`	string \| null	Failure detail; null when no error applies.
`meta`	object \| null	Metadata, null when none applies. `ocrConfidence` and `chunkCount` are nested here, never top-level keys.
`ts`	string	ISO-8601 producer timestamp with timezone.

meta.ocrConfidence is the minimum per-chunk OCR confidence across all chunks that carry one, including parked (non-indexed) chunks (_emit_success_stages, src/aceh_rag/kafka/kafka_ingestion.py:947-967) — a scan with .95 indexed and .40 parked emits .40. An OCR attempt that falls back to native prose may close with null metadata. meta.chunkCount is total chunks for chunking, indexed chunks for embedding and terminal indexing, and 0 for structured-source or byte-duplicate skips. Review can retain some indexed chunks; it does not guarantee a zero count.

Stages

type Stage = "extracting" | "ocr_pending" | "ocr_done" | "chunking" | "embedding" | "indexing" | "indexed" | "failed";

extracting represents fetch and parse completion. ocr_pending starts an actual OCR attempt; ocr_done closes it. chunking and embedding report completed milestones. indexing is accepted by the v1 envelope but is not emitted by the current milestone adapter. indexed carries success, skip, or review; failed carries terminal failure.

Status meanings

type StageStatus = "started" | "done" | "failed" | "skipped" | "needs_review";

Status	Terminal?	Meaning
`started`	No	Current adapter emits only `ocr_pending/started`, before actual OCR, with progress 25 and null `aiDocumentId`, `error`, and `meta`.
`done`	Only on `indexed`	Completed milestone; only `indexed/done` marks successful completion.
`failed`	Yes	`failed/failed`; surface `error` and fix before republishing.
`skipped`	Yes	`indexed/skipped`; structured-source or byte-duplicate no-op.
`needs_review`	Yes	`indexed/needs_review`; human review required, including partially indexed scans. Do not mark complete.

Serialized success example

{
  "v": 1,
  "jobId": "550e8400-e29b-41d4-a716-446655440000",
  "registryId": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
  "aiDocumentId": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "stage": "indexed",
  "status": "done",
  "progress": 100,
  "error": null,
  "meta": {
    "chunkCount": 12
  },
  "ts": "2026-10-07T09:12:03Z"
}

Ordered lifecycle

Fetch completion is emitted immediately. OCR-start is emitted once per actual attempt, including retries, before OCR work. Parse and other completion updates are emitted after the reused pipeline returns; therefore OCR-start can precede the parse-completion update. Across topics, reconcile by job and progress rather than assuming a global order.

Path	Wire updates
Born-digital success	`extracting/done` (10), `extracting/done` (25), `chunking/done` (65), `embedding/done` (85), `indexed/done` (100)
OCR success	`extracting/done` (10), `ocr_pending/started` (25), `extracting/done` (25), `ocr_done/done` (45), `chunking/done` (65), `embedding/done` (85), `indexed/done` (100)
OCR review	Same OCR milestones, ending `indexed/needs_review` (100)
Fetch failure	`failed/failed` (100); no fetch-completion update
Pipeline failure after fetch	`extracting/done` (10), optional OCR-start updates, `failed/failed` (100)
Structured-source skip	`indexed/skipped` (100), with null `aiDocumentId`
Content-duplicate skip	`extracting/done` (10), `indexed/skipped` (100), with the existing `aiDocumentId`

Retry, deferred, skipped, and publish-error semantics

These states affect partner job tracking even when they are not all visible as output events.

Consumer outcome	Stage event?	Offset handling	Partner tracking rule
`indexed`	Yes, ends with `indexed/done`	Committed	Mark the document version complete.
`needs_review`	Yes, ends with `indexed/needs_review`	Committed	Mark terminal review-needed, not complete.
`error`	Yes, usually `failed/failed`	Committed	Mark failed for this delivery. Republish only after fixing the cause.
`skipped`	`indexed/skipped` for structured-source or content-duplicate no-ops; no new event for an already-processed redelivery	Committed	Treat a `indexed/skipped` event as terminal. For a redelivery already skipped by the in-process dedup store, retain the previous delivery's status.
`deferred`	No success stage event	Not committed for that record immediately	Another worker holds the per-document lock. Keep the job pending, reconcile it as a missing-terminal-status document, and explicitly re-drive it according to the agreed operational policy.

deferred is an internal, non-terminal outcome for lock contention. It emits no success stages and skips the contended record's immediate commit. Current consumer processing does not provide a same-partition redelivery guarantee: if a later record from the same partition is committed, that commit may advance the partition offset past the deferred record. Partner systems must reconcile documents that do not reach indexed/done, indexed/needs_review, or failed/failed and re-drive them explicitly under the agreed operational policy.

If publishing a stage-status event itself fails, Aceh RAG logs and continues processing the document. The document can still be persisted and marked processed even if one or more status events are missed. Partner consumers should make their stage updates idempotent and should have an operator reconciliation path for documents that were produced but have no recent status event.

Producer checklist for `doc.uploaded`

Before producing events:

Confirm the active input topic, complete stage-topic map, consumer group, broker list, and Kafka worker deployment with the platform team.
Ensure the object bytes exist at storageKey before publishing the event.
Use a stable UUID jobId for redelivery of the same IngestionJob, and a UUID documentId for the document version.
Send a timezone-bearing uploadedAt and do not substitute the future AI-engine ID for either partner ID.
Send a mimeType that agrees with the storageKey extension and a recognized collectionName.
Confirm the configured OBS bucket, worker IAM principal, and allowed key prefix; no presigned URL or partner database lookup is used.
Omit unknown optional fields rather than sending null or empty strings.
Keep the full original event in partner logs so a malformed or failed delivery can be corrected and republished.

Consumer checklist for `StageStatusEvent`

When consuming status events:

Key partner job state exclusively by jobId. registryId correlates the partner registry; nullable aiDocumentId is AI-engine provenance, not the update key.
Treat all fields in the schema as required. Store nullable fields as nullable, not missing.
Apply updates idempotently using the full envelope, including progress; the two legitimate extracting/done milestones have progress 10 and 25.
Preserve ordering within each destination topic, but tolerate missing intermediate stages and do not assume a total order across stage topics because stage publishing is best-effort.
Mark indexed/done complete.
Mark indexed/needs_review as terminal review-needed and route to human review.
Mark failed/failed as failed and surface error to operators.
Mark indexed/skipped as the terminal no-op status. Do not wait for a new status event after an already-processed duplicate redelivery; the previous delivery's status is authoritative for that version.
Reconcile pending jobs that have no terminal status event and explicitly re-drive them according to the agreed operational policy; do not rely on deferred lock contention to redeliver the original Kafka record.

Transport and ownership​

Default stage-topic map​

doc.uploaded input event​

Field table​

Source type values​

Format values​

Required event​

Full event example​

Input validation and failure behavior​

Document identity, versioning, and idempotency​

StageStatusEvent output event​

JSON schema​

Field meanings​

Stages​

Status meanings​

Serialized success example​

Ordered lifecycle​

Retry, deferred, skipped, and publish-error semantics​

Producer checklist for doc.uploaded​

Consumer checklist for StageStatusEvent​