CIP step catalog

Reference for $defs.cipStep* on Business Entity execution.cip.operations.*.steps. Each step is a single keyed object — one of the types below. Authoring context: Configure data flow. Certification: CIP execution.

Prerequisites

• Configure data flow open while editing execution.cip
• Schema synced via Builder validate (or local aifabrix validate <systemKey>)

Where it lives

How to set

1. Open the operation under execution.cip.operations that matches your capability key.
2. Append steps as single-key objects only — never invent keys outside this catalog.
3. Prefer the common list pattern: fetch → paginate → map → filter → output.
4. Re-run aifabrix validate <systemKey> after each structural change.

Defaults and examples

Step types at a glance

Common optional step wrappers: enabled, order, stepId, onError, lineage (where schema allows).

fetch

Purpose: Load a payload for the operation. source is where I/O happens. CIP operation shape (create | update | upsert | delete) is persist vs read on local storage — not the capability key. Pipeline output.mode is records for ordinary record output or documents for binary document persistence.

Required: object key fetch with source. Closed enum:

source: record is singular. Do not write fetch.source: records. output.mode: records remains plural.

CIP fetch.source compiles to COM source (same pattern as writeShape). Named perspectives are source: viewpoint. Local table and vendor fetches are the other tokens — capability keys such as list / search / get are examples, not a closed set.

fetch.responseType is json by default. Use binary only with source: openapi or source: http for a file response; it carries one binary record to document output. Bytes stay out of traces and returned evidence.

Useful optional fields (vendor/nested): query (parameter object), bodyTemplate, headers, expectedItemsPath, accumulate, skipUnauthorized, parameters (datasource). query is a SQL string only for viewpoint.

{
  "fetch": {
    "source": "openapi",
    "openapiRef": "list",
    "query": { "limit": "100" }
  }
}

Viewpoint (SQL against governed record storage):

{
  "fetch": {
    "source": "viewpoint",
    "query": "SELECT customer_id AS \"customerId\", SUM(amount) AS total FROM records_example_orders GROUP BY customer_id"
  }
}

Local table read (list, get, or any other published read capability — no shape):

{
  "fetch": {
    "source": "record"
  }
}

Local table persist — shape on the operation, not inferred from the key:

{
  "registerEmployee": {
    "shape": "create",
    "steps": [
      { "fetch": { "source": "record" } },
      { "output": { "mode": "records" } }
    ]
  }
}

upsert is an operation shape, not a new step type. It must be declared and
compose exactly one datasource create fetch and one datasource update
fetch for the same datasource. Identity comes from
validation.repeatingValues entries with strategy: "reject"; every identity
field must be present in the request. CIP resolves record existence once and
executes only the permitted half. Missing or ambiguous identity, an unresolved
record, or denied access stops without falling back. Validate the production
transport used by the integration; platform connector and worker acceptance
is tracked separately from the schema and datasource execution contract.

When to use: Vendor or child-entity HTTP (openapi / http / datasource); calculated read-only perspectives (viewpoint); local record-storage list/get/create/update with no vendor HTTP (record).
When not: Do not invent a sixth source. record is not vendor HTTP. viewpoint is not a table write. Do not use a capability key (create, list, …) as the fetch source.

paginate

Purpose: Continue fetching pages after an initial fetch (typically list).

Required: paginate.strategy — cursor | page | offset | none.

{
  "paginate": {
    "strategy": "cursor",
    "cursorField": "$.paging.next.after",
    "cursorParam": "after",
    "pageSizeParam": "limit",
    "pageSize": 100,
    "maxPages": 50
  }
}

When to use: List operations with vendor pagination.
When not: Single-record get / create / update / delete unless the vendor truly pages those responses.

map

Purpose: Transform raw API records into the normalized metadata model.

Required: object key map (inner properties optional; defaults apply).

Prefer useFieldMappings: true so fieldMappings.attributes drive expressions. Optional inputPath (default $.items[*]) and inline overrides. Pipe functions (toLower, split, index, and the rest of the closed list) are documented in Configure field-mapping expressions.

{
  "map": {
    "useFieldMappings": true,
    "inputPath": "$.results[*]"
  }
}

When to use: Almost every list/get after fetch.
When not: Do not invent a transform step — use map (or pythonInline as last resort).

merge

Purpose: Merge accumulated fetch outputs (fetch.accumulate: true) — data transformation only.

Required: merge.strategy — object | array | union | byKey (optional keyField).

{
  "merge": {
    "strategy": "array"
  }
}

When to use: Multi-fetch orchestration into one batch.
When not: Simple single-fetch pipelines — skip merge.

filter

Purpose: Filter mapped records before output (ABAC and/or JMESPath).

Required: object key filter.

{
  "filter": {
    "abac": {
      "mode": "mandatory",
      "policyScope": "datasource"
    }
  }
}

abac.mode: mandatory | optional | disabled. Optional expression (+ expressionLanguage: jmespath).

When to use: Enforce or explain ABAC on list/get results.
When not: Do not disable ABAC in production tenants without a documented exception.

output

Purpose: Finalize the pipeline. records emits governed records; documents persists binary content through the document pipeline and emits document references.

Required: object key output. Prefer mode: records.

{
  "output": {
    "mode": "records"
  }
}

Optional: limit, includeConfidence.

For keyed file retrieval, pair fetch.responseType: binary with
output.mode: documents. The datasource must use entityType: documentStorage
and documentStorage.enabled: true; map its stable key from request parameters
or response headers. The document pipeline retains metadata, projects protection
and applies ingestion eligibility. ingestAfterSync does not bypass approval or
required-link checks. A simulation does not prove persistence.

Operation fragment (requires the surrounding datasource and matching OpenAPI
get declaration):

{
  "get": {
    "steps": [
      { "fetch": { "source": "openapi", "openapiRef": "get", "responseType": "binary" } },
      { "output": { "mode": "documents" } }
    ]
  }
}

Use Build a document storage entity
for keyed-get testing and storage prerequisites.

When to use: End of every operation that should return governed records.
When not: Leaving output off often breaks sync and agent consumption — include it unless validate proves otherwise.

pythonInline

Purpose: Sandboxed inline Python on the current record batch when declarative steps cannot express the transform.

Required: pythonInline.code — small pure snippet; no imports, IO, or network (runtime must enforce sandboxing).

{
  "pythonInline": {
    "code": "return [r for r in records if r.get('status') == 'active']"
  }
}

When to use: Rare edge transforms after map.
When not: Prefer map / filter first. Prefer execution.engine: python only for full custom handlers, not routine field mapping.

Validate

aifabrix validate <systemKey>

Reject invented step keys; use this catalog and Builder help topic schemaCatalog when property names fail.

Common mistakes