Ingestion Pipeline

Audience: developer who needs to understand how data enters Cortex, what validations are applied, and how the pipeline ensures idempotency.

Overview

Ingestion is the only structured write path for catalogue data. It is centralized in domain/services/ingestion-service.ts and delegates each step to focused sub-modules under domain/services/ingestion/.

The pipeline handles two entity types:

Blocs and atoms via ingestBloc() — uses the arka.hcm.block.v1 schema.
HYOS profiles via ingestProfil() — uses the arka.hcm.profil.v1 schema.

Both follow the same eight-step pipeline:

Input (JSON)
  |
  v
1. VALIDATE       Zod schema + cross-field checks
  |
  v
2. HASH           Compute content_hash (SHA-256)
  |
  v
3. LOCK           In-memory lock per nodeId
  |
  v
4. DIFF           Compare hash with existing node
  |
  v
5. ROUTE          Dispatch to bloc table or atom table
  |
  v
6. PERSIST        Create or update via GraphPort
  |
  v
7. TAXONOMY       Ensure famille/sous-famille nodes + edges
  |
  v
8. EMBED          Enrich with bge-m3 vector (best-effort)

Step 1 — Validate

Validation is handled by ingestion/ingestion-validation.ts. Two functions:

validateBlockStructure(input) — parses against HyosBlockV1Schema (Zod). Validates:
- Required fields: registre.id, registre.type, registre.branche, registre.label, registre.description, registre.sous_famille, registre.artifact_ref.
- Type enum: must be one of the 13 HYOS types (skill, expertise, methode, tool, scope, mode, reglement, pipeline, template, schema, service, arkadoc_bloc, arkadoc_type).
- Branch enum: must be one of the 5 branches (equipement, gouvernance, service, arkadoc_bloc, arka_deck).
- Conditional validation: arkadoc_bloc type triggers ArkadocBlocArtefactSchema on the artefact field. arkadoc_type type triggers ArkadocTypeArtefactSchema.
- Cross-field coherence: arkadoc_bloc must have famille=ARKADOC_BLOC and branche=arkadoc_bloc. arkadoc_type must have famille=ARKA_DECK and branche=arka_deck.
validateProfilStructure(input) — parses against HyosProfilV1Schema (Zod). Validates:
- Required: registre.id, registre.label, registre.description, registre.sous_famille.
- Fixed values: registre.type=profil_hyos, registre.branche=profil.
- Composition entries: each must have a bloc_id and a poids of core or support.

On validation failure, the pipeline returns immediately with { status: 'error', action: 'error', error: { code, message, field } }.

Step 2 — Hash

Content hashing is handled by ingestion/ingestion-mappers.ts:

computeBlockContentHash(block) — SHA-256 of the block’s canonical representation.
computeProfilContentHash(profil) — SHA-256 of the profile’s canonical representation.

The hash is the primary mechanism for idempotent ingestion: if the computed hash matches the existing node’s content_hash, the pipeline returns { action: 'ignored' } without any write.

Step 3 — Lock

The IngestionLockManager (ingestion/ingestion-lock.ts) prevents concurrent writes to the same node. It uses an in-memory Map<string, string> where the key is the node ID and the value is the content hash being written.

acquire(nodeId, hash) — returns true if the lock was acquired, false if another ingestion is in progress for the same node.
release(nodeId) — releases the lock.

If the lock cannot be acquired, the pipeline returns { action: 'error', error: { code: 'INGESTION_IN_PROGRESS' } }.

Step 4 — Diff

The pipeline fetches the existing node from the graph (graph.getNode(nodeId)) and compares:

If no existing node: the atom will be created.
If existing node with the same content_hash: the atom is ignored (idempotent).
If existing node with a different hash: the atom will be updated.
If existing node has a different type: returns { error: { code: 'ID_CONFLICT' } }.

For updates, detectChangeType() from semver-bump.ts classifies the change as patch, minor, or major based on which fields changed.

Step 5 — Route

The pipeline determines the target table based on the atom type:

Bloc types (skill, expertise, methode, tool, scope): routed to the bloc table via catalogueService.depositBloc() or catalogueService.updateBloc().
Atom types (mode, reglement, pipeline, template, schema, service, arkadoc_bloc, arkadoc_type): routed to the atom table via direct graph.addNode() or graph.updateNode().

The routing decision uses the HYOS_TYPES_BLOC constant:

const isBloc = HYOS_TYPES_BLOC.includes(block.registre.type);
const nodeId = isBloc ? `bloc:${block.registre.id}` : `atom:${block.registre.id}`;

Step 6 — Persist

For blocs (via CatalogueService)

Create: catalogueService.depositBloc(mapToBlocInput(block)) — maps the ingestion schema to the internal Bloc type and persists.
Update: catalogueService.updateBloc(nodeId, mapToBlocUpdate(block), changeType) — applies delta and bumps semver based on change classification.

For atoms (direct graph writes)

Create: graph.addNode({ id, type: 'atom', label, data, ... }) with versionSemver: '1.0.0'.
Update: graph.updateNode(nodeId, { ...atomData, version_semver: nextVersion }) where nextVersion = bumpSemver(previousVersion, changeType).

After creation, the pipeline also creates:

A belongs_to_sous_famille edge from the atom to its sub-family node.
An authored_by edge from the atom to the contributor node (if registre.auteur is set).

For profiles

Create: graph.addNode({ id, type: 'profil_hyos', label, data, ... }).
Update: graph.updateNode(nodeId, { ...profilData, version_semver: nextVersion }). Before update, existing equipe_de edges are removed and recreated.

After persistence, profilEdges.createComposition(nodeId, profil) creates equipe_de edges from each referenced bloc to the profile node.

Step 7 — Taxonomy

The IngestionTaxonomyService (ingestion/ingestion-taxonomy.ts) ensures taxonomy nodes and edges exist for the ingested atom:

Ensure the famille node exists (e.g. famille:canonique_gouvernance).
Ensure the sous_famille node exists (e.g. sous_famille:safety_baseline).
Ensure a belongs_to_famille edge connects the sous-famille to the famille.

All operations are idempotent — if nodes/edges already exist, no error is raised.

Step 8 — Embed

The IngestionEmbeddingService (ingestion/ingestion-embedding.ts) enriches the node with a vector embedding:

Uses the Ollama adapter with the bge-m3 model (1024 dimensions).
Builds an input string from the atom’s label, description, tags, and content.
Truncates input to INGESTION_EMBEDDING_MAX_INPUT_CHARS (default: 8000).
Timeout: INGESTION_EMBEDDING_TIMEOUT_MS (default: 10000ms).
Retries: INGESTION_EMBEDDING_RETRY_MAX (default: 3).

Embedding is best-effort and non-blocking: if the Ollama service is unavailable, the pipeline still succeeds. The embedding will be null until a subsequent enrichment pass.

Before a batch ingestion, the pipeline checks isProviderHealthy() and logs a warning if the embedding provider is unreachable.

Batch ingestion

Both ingestBlocBatch(inputs) and ingestProfilBatch(inputs) process items sequentially (not concurrently) to respect the in-memory lock manager and avoid overwhelming the embedding provider.

The batch endpoint (POST /api/blocs/batch) is rate-limited to 600 requests per minute and capped at INGESTION_BATCH_MAX_BLOCS items per request (default: 200).

Dry run

The dryRun(input) method validates the input and checks what action would be taken without actually persisting:

interface DryRunResult {
  valid: boolean;
  id: string;
  wouldAction?: 'created' | 'updated' | 'ignored';
  contentHash?: string;
  errors?: Array<{ code: string; message: string; field?: string }>;
}

Available via POST /api/blocs/dry-run and cortex_catalogue_ingest with dry_run=true.

Ingestion result

Every ingestion call returns:

interface IngestionResult {
  status: 'ok' | 'error';
  id: string;
  version?: string;
  action: 'created' | 'updated' | 'ignored' | 'error';
  error?: { code: string; message: string; field?: string };
}

Error codes:

Code	Meaning
`VALIDATION_ERROR`	Zod schema validation failed
`INGESTION_IN_PROGRESS`	Concurrent ingestion for the same node
`ID_CONFLICT`	Node ID exists with a different type
`UNEXPECTED`	Unhandled exception during persistence

Domain events

After successful persistence, the pipeline emits a node.deposited domain event with:

{
  type: 'node.deposited',
  timestamp: string,
  payload: {
    nodeId: string,
    nodeType: string,
    label: string,
    edgesCreated: number,
    versionSemver: string,
    contentHash: string,
    isUpdate: boolean,
  }
}

Listeners can react to this event for MeiliSearch sync, resonance scoring, or external notifications.