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 | delete) is persist vs read on local storage — not the capability key. Pipeline output.mode stays records (plural collection envelope).

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.

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" } }
    ]
  }
}

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.

{
  "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 and emit records.

Required: object key output. Prefer mode: records.

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

Optional: limit, includeConfidence.

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