Pipeline Expert

OctoMesh Pipeline Expert

Overview

OctoMesh pipelines are YAML-defined ETL data flows executed by adapters. Each pipeline has triggers that start execution and transformations (an ordered list of nodes) that process data through a shared DataContext — a mutable JSON document accessed via JSONPath.

An Adapter (System.Communication/Adapter) is a unified runtime that executes pipelines. Different implementations exist (Mesh, Zenon, Simulation, …) but all share the same CK type and execution model; each registers the nodes it supports — SDK-shared nodes plus adapter-specific ones. Pipelines handle entity CRUD, cross-adapter sync, import/export, notifications, reports, AI queries, anomaly detection, and more.

Source Code Research (MANDATORY)

NEVER guess how a pipeline node behaves. The reference docs are summaries; the C# source is ground truth. Read the source before answering or writing YAML whenever you are not 100% certain of a node's property names/defaults, need to know what it does at runtime, hit a node you don't recognize, or a user reports unexpected behavior. All paths below are relative to the monorepo root — the parent directory of this plugin's repo, containing both octo-sdk/ and octo-mesh-adapter/ (use ../ or search upward).

Where Node Source Code Lives

Naming Conventions & Lookup

Find a node: Grep for NodeName\("CheckDuplicate" in octo-mesh-adapter/src/ (else octo-sdk/src/), or Glob **/CheckDuplicate*Configuration*.cs. Read the config class for properties/defaults/required-vs-optional (from XML docs), then the handler for runtime behavior, DataContext reads/writes, and error conditions.

DataFlows and Pipeline Triggers

A DataFlow (System.Communication/DataFlow) is a logical grouping of related pipelines that work together as part of a single data processing workflow. It serves as the parent container for Pipeline and PipelineTrigger instances.

Migration note: CK migration 3.1.0→3.1.1 unified the old EdgeAdapter/MeshAdapter into a single Adapter type, and EdgePipeline/MeshPipeline into a single Pipeline type. Earlier migrations renamed DataPipeline → DataFlow and DataPipelineTrigger → PipelineTrigger. These are handled automatically by ChangeCkType transforms.

Key concepts:

• Pipelines belong to a DataFlow via System/ParentChild association
• A DataFlow establishes a shared topic exchange in the event hub, enabling inter-pipeline communication
• Pipelines within the same DataFlow can send data to each other using ToPipelineDataEvent@1 (with targetPipelineRtId) and FromPipelineDataEvent@1

A PipelineTrigger (System.Communication/PipelineTrigger) is a child of a DataFlow that triggers pipeline execution on a cron schedule via the Bot Service:

• Has Enabled and CronExpression attributes
• Has a Triggers association linking to one or more Pipeline entities
• The target pipelines must use FromPipelineTriggerEvent@1 as their trigger node
• Cron format: standard 5-field minute hour dayOfMonth month dayOfWeek (no year field; e.g. 0 * * * * = top of every hour). Scheduled via the MassTransit/Hangfire recurring scheduler.

Entity relationships:

DataFlow
  ├── Pipeline (child, via ParentChild) ── executes on ── Adapter
  └── PipelineTrigger (child, via ParentChild) ── triggers ── Pipeline(s)

Pipeline YAML Structure

triggers:
  - type: NodeType@Version      # trigger-specific properties
transformations:
  - type: NodeType@Version      # node-specific properties
    transformations:            # control-flow nodes nest child nodes
      - type: ChildNode@Version

Triggers define how the pipeline starts: polling interval, HTTP request, entity change watch, event hub message, explicit command, or email. Each trigger type populates the DataContext with different initial paths — see references/data-context-guide.md "Trigger DataContext Placement" for the full table. Key examples:

• FromHttpRequest@1: body at $.body, query params at $.query, files at $.files
• FromWatchRtEntity@1: changed entity at $.Document
• FromExecutePipelineCommand@1 / FromPipelineTriggerEvent@1: empty context. FromExecutePipelineCommand@1 now lives in Sdk.Common and is available on all adapters (Edge and Mesh), not Mesh-only — the pipeline must still belong to a DataFlow.

Transformations define the processing steps. Each node reads from and writes to the DataContext. Control flow nodes (ForEach, If, Switch) nest child transformations.

Common Pipeline YAML Mistakes

These errors cause deployment failures. Check your YAML against this table before deploying:

Prevention: before deploying, fetch the schema (octo-cli -c GetPipelineSchema --adapterId <rtId> --outputFile schema.json) and confirm each node's property names and enum values in $defs, or use pipeline_validate.py.

DataContext Essentials

The DataContext is a mutable JSON document. All nodes share it.

Reading: Use path or valuePath properties with JSONPath ($.property, $.array[*], $.nested.path).

Writing: Use targetPath with three modifiers:

• documentMode: Extend (merge into existing, default) or Replace (clear and set)
• targetValueWriteMode: Overwrite (default), Append, Prepend, or Merge
• targetValueKind: Simple (scalar/object, default) or Array (wrap in array)

Common write pattern for collecting items:

targetPath: $.key.updates
targetValueWriteMode: Append
targetValueKind: Array

For full details on write semantics and field filters, read references/data-context-guide.md.

ForEach Iteration

For@1 is different: For@1 deep-clones the parent context directly — it does NOT create $.full/$.key paths. Access data at the same paths as the parent (e.g., $.body.count, not $.full.body.count). Supports both static count and dynamic countPath (JSONPath). See references/data-context-guide.md for a full comparison table.

ForEach creates a child context per array element with three key paths:

Inside ForEach, access data as:

• $.key.Field — current item
• $.full.OtherData — parent context data

Nested ForEach chains $.full:

• $.key — innermost item
• $.full.key — outer item (one level up)
• $.full.full.Config — root context (two levels up)

Parallelism: maxDegreeOfParallelism controls concurrency (0=CPU count, -1=unlimited, >0=explicit). Use 1 for sequential when order matters.

For a deeper explanation of context hierarchy, write modes, and field filters, read references/data-context-guide.md.

Node Categories

This is an operational overview of the most-used nodes. The full property tables for every node live in references/node-reference-sdk.md (SDK nodes, available on all adapters) and references/node-reference-mesh.md (Mesh Adapter nodes). Read those before writing YAML for any node not summarized here.

Triggers

Control Flow (SDK)

Extract (Data In)

Transform

Common: CreateUpdateInfo@1 (build entity Insert/Update/Delete), CreateAssociationUpdate@1 (build association Create/Delete), CreateFileSystemUpdate@1, DataMapping@1 (value mapping), Concat@1, FormatString@1, TransformString@1, Flatten@1, Project@1, Join@1, Math@1, SumAggregation@1, FilterLatestUpdateInfo@1, Distinct@1 (SDK — all adapters), PlaceholderReplace@1, Base64Encode@1/Base64Decode@1, Hash@1, ConvertDataType@1, Map@1, LinearScaler@1, MinMax@1, Logger@1/PrintDebug@1, ExecuteCSharp@1, MakeHttpRequest@1, CheckDuplicate@1, ComputeFileHash@1, QueryResultToMarkdownTable@1.

Specialized (see node-reference-mesh.md): AnthropicAiQuery@1 (Claude AI; MCP-aware), PdfOcrExtraction@1, StatisticalAnomalyDetection@1, MachineLearningAnomalyDetection@1, ImportFromCsv@1, ImportFromExcel@1, ReplyToTeamsChannel@1.

DataPointMapping family (Mesh): ApplyDataPointMappings@1 (evaluate mappings with mXparser expressions), BuildMappingTargets@1 (resolve mappings to acquisition targets), GenerateDataPointMappings@1 (deterministic rule-based generator — non-AI alternative to AnthropicAiQuery@1), ValidateDataPointCoverage@1 (coverage report), MapToRecordArray@1, UpdateRecordArrayItem@1.

Simulation: Simulation@1 is an Extract node (not Transform) — Bogus/Math/Energy generators; SimulateEnergyMeasurements@1 (Mesh) generates 15-min EnergyMeasurement slots from BDEW/PV profiles.

Load (Data Out)

Buffering (SDK)

BufferData@1 (buffer with time-based flush) and BufferRetrievalNode@1 (retrieve buffered data).

Domain-Specific

SAP nodes (SapLogin, GetProductionOrderList, GetProductionOrderDetails), Zenon nodes (variable/AML read-write plus the six archive/Editor nodes ReadZenonArchiveInfo@1, ReadZenonArchiveData@1, ListZenonProjects@1, GetZenonDynamicProperties@1, GetZenonDynamicProperty@1, SetZenonDynamicProperty@1), Microsoft Teams nodes, and Grafana nodes are documented in references/node-reference-mesh.md. EDA energy nodes ship in an external adapter not present in this monorepo — see that file's external-adapter note.

Persisting OutputData (why is OutputData empty?)

Pipeline execution output is only stored when the pipeline includes SetPipelineExecutionResult@1 (SDK Load node). It captures the data-context value at its path and stores it as the PipelineExecution.OutputData retrieved via GetLatestPipelineExecution. Without this node nothing is persisted — this is by design (avoids storing large results from high-frequency pipelines), and it explains an empty/missing OutputData. maxLength (default 1 048 576 chars) truncates oversized results.

- type: SetPipelineExecutionResult@1
  path: $.result        # value to persist as OutputData
  # maxLength: 1048576  # optional cap

RT Entity Data Structures

GetRtEntitiesByType@1 returns an IResultSet object, not a plain array:

{ "TotalCount": 3, "Items": [ { "RtId": "...", "CkTypeId": "...", "RtWellKnownName": "...", "Attributes": { "Name": "value", "StartDateTime": "2026-01-01T06:00:00Z" } }, ... ] }

Key rules:

• Iterate with .Items: iterationPath: $.result.Items (NOT $.result)
• Check count: $.result.TotalCount
• Access attributes inside ForEach: $.key.Attributes.AttributeName (short name, no CK prefix or version suffix)
• System properties: $.key.RtId, $.key.CkTypeId, $.key.RtWellKnownName
• All property names are PascalCase

Attribute Name Casing in CreateUpdateInfo

CRITICAL: The attributeName field in CreateUpdateInfo@1 must use the exact casing from the CK model definition — typically camelCase (e.g., name, machineState, operatingHours).

This is different from the PascalCase you see in GraphQL query results (where attributes appear as $.key.Attributes.Name). The distinction:

Always run ck_explorer.py preflight <type> before writing CreateUpdateInfo to get the exact attribute names.

EntityUpdateInfo JSON Structure

CreateUpdateInfo@1 writes an EntityUpdateInfo: a nested RtEntity (with RtId, RtChangedDateTime, CkTypeId, Attributes) plus top-level RtId, CkTypeId, and ModOption (0=Insert, 1=Update, 2=Delete).

{ "RtEntity": { "RtId": "cc...bb01", "CkTypeId": "Industry.Basic/Machine", "Attributes": { "MachineState": 0 } }, "RtId": "cc...bb01", "CkTypeId": "Industry.Basic/Machine", "ModOption": 0 }

The top-level RtId is populated only when a static rtId is given or an upstream node resolved/generated it. Do NOT use generateRtId: true — obtain IDs first via GetOrCreateRtEntitiesByType@1 or GetRtEntitiesByWellKnownName@1 so they can be referenced in associations.

Mandatory Associations

Some CK types have mandatory outbound associations (multiplicity = ONE). Creating an entity of such a type WITHOUT the required association will fail at ApplyChanges time with: "Inbound association 'X' has minimum multiplicity of 'One'".

Pre-flight check: Always run ck_explorer.py preflight <type> before writing CreateUpdateInfo for a new entity type. If the output shows mandatory associations, you MUST include CreateAssociationUpdate@1 nodes in the same ApplyChanges@2 call.

Common mandatory association: System/ParentChild — types like Machine, TreeNode, and many domain entities require a parent.

The intended workflow for creating entities with associations

Use GetOrCreateRtEntitiesByType@1 to resolve/generate IDs for both parent and child (each writes rtId + modOp paths), then have CreateUpdateInfo@1 and CreateAssociationUpdate@1 reference those IDs via paths — never generateRtId: true:

- type: GetOrCreateRtEntitiesByType@1   # parent → $.parentRtId / $.parentModOp
  ckTypeId: Basic/Tree
  fieldFilters: [{ attributePath: RtWellKnownName, comparisonValue: "My Container" }]
  rtIdTargetPath: $.parentRtId
  modOperationPath: $.parentModOp
- type: GetOrCreateRtEntitiesByType@1   # child → $.childRtId / $.childModOp
  ckTypeId: Industry.Basic/Machine
  fieldFilters: [{ attributePath: RtWellKnownName, comparisonValue: "My Machine" }]
  rtIdTargetPath: $.childRtId
  modOperationPath: $.childModOp
- type: CreateUpdateInfo@1              # child entity update, RtId from path
  targetPath: $.entityUpdates
  targetValueWriteMode: Append
  targetValueKind: Array
  updateKindPath: $.childModOp
  rtIdPath: $.childRtId
  ckTypeId: Industry.Basic/Machine
  attributeUpdates: [{ attributeName: machineState, attributeValueType: Enum, value: 1 }]
- type: If@1                            # create the mandatory assoc ONLY on INSERT (modOp = 0)
  path: $.childModOp
  value: 0
  valueType: Enum
  transformations:
    - type: CreateAssociationUpdate@1
      targetPath: $.assocUpdates
      targetValueWriteMode: Append
      targetValueKind: Array
      updateKind: CREATE
      originRtIdPath: $.childRtId
      originCkTypeId: Industry.Basic/Machine
      targetRtIdPath: $.parentRtId
      targetCkTypeId: Basic/Tree
      associationRoleId: System/ParentChild
- type: ApplyChanges@2                  # persist entities + associations together
  entityUpdatesPath: $.entityUpdates
  associationUpdatesPath: $.assocUpdates

Only create associations on INSERT (guard on modOperationPath = 0); CreateUpdateInfo@1/CreateAssociationUpdate@1 read the IDs via rtIdPath/originRtIdPath/targetRtIdPath.

Common Patterns

Entity CRUD

The standard CRUD shape is: GetRtEntitiesByWellKnownName@1 (with generateInsertOperation: true to write rtId/modOp per item) → ForEach@1 building one CreateUpdateInfo@1 per item into $.key.update → Flatten@1 the per-item updates into $.entityUpdates → ApplyChanges@2. See references/pipeline-examples.md (example 3) for the full annotated version, and the mandatory-association workflow above when the new entity type requires associations.

Inter-Pipeline Communication (DataFlow)

Pipelines within the same DataFlow can chain data to each other — even across different adapter instances (e.g., Zenon Adapter → Mesh Adapter). The sender uses ToPipelineDataEvent@1 with the target pipeline's runtime ID; the receiver uses FromPipelineDataEvent@1 as its trigger.

# Producer pipeline (sends data to consumer)
transformations:
  - type: ToPipelineDataEvent@1
    path: $.sensor
    targetPath: $.input
    targetPipelineRtId: aa0000000000000000000003  # RtId of the consumer pipeline

# Consumer pipeline (receives data)
triggers:
  - type: FromPipelineDataEvent@1

Both pipelines must belong to the same DataFlow (linked via System/ParentChild association). The DataFlow's shared topic exchange routes messages by targetPipelineRtId.

Await-result (synchronous) mode: set awaitResult: true on ToPipelineDataEvent@1 to send a command and block until the target pipeline finishes, placing its result at resultTargetPath (default $.pipelineResult). The receiver FromPipelineDataEvent@1 needs no extra config — it consumes both the fire-and-forget exchange and the command address. Optional timeoutSeconds bounds the wait.

- type: ToPipelineDataEvent@1
  path: $.request
  targetPath: $.input
  targetPipelineRtId: aa0000000000000000000003
  awaitResult: true
  timeoutSeconds: 30
  resultTargetPath: $.pipelineResult

Dual Store (Archive + MongoDB)

Save high-frequency data to both a CrateDB archive and the entity store. SaveStreamDataInArchive@1 requires archiveRtId pointing at an Activated CkArchive; backfill missing columns first if upstream events carry only one attribute.

# Optional: complete each row's columns from the persistent entity
- type: BackfillFromRtEntity@1
  path: $._updates
  archiveRtId: cc0000000000000000000aa1   # the CkArchive

- type: SaveStreamDataInArchive@1
  path: $._updates
  archiveRtId: cc0000000000000000000aa1   # must be Activated

- type: FilterLatestUpdateInfo@1
  path: $._updates
  targetPath: $._updates

- type: ApplyChanges@2
  entityUpdatesPath: $._updates

Pipeline Validation

When validating a pipeline YAML (user-written or generated), use the build-time JSON Schema as the authoritative source. Every OctoMesh adapter generates its own pipeline-schema.json at build time, containing all nodes it supports (SDK nodes + adapter-specific nodes).

Schema locations (relative to monorepo root):

Which schema to use: Pick the adapter that will execute the pipeline (Mesh Adapter has the richest node set). If it hasn't been built locally, fall back to the node reference docs.

Validation: look up each node by its type const in $defs.TriggerNode.oneOf / $defs.TransformationNode.oneOf, then verify all required keys are present, every property exists in the node's properties, and enum values match. For the extraction commands and fallback rules, read references/pipeline-schema-guide.md.

Pipeline Creation Workflow

1. Plan the DataFlow — does this pipeline work alone or chain with others? Group related pipelines under one DataFlow.
2. Identify the trigger — manual (FromExecutePipelineCommand@1), cron (FromPipelineTriggerEvent@1 + PipelineTrigger), inter-pipeline (FromPipelineDataEvent@1), polling, HTTP, entity-watch, or email.
3. Plan the data flow and $.path names for each step's input/output; use ForEach for arrays (plan $.full/$.key).
4. Research node properties — read the C# config class for every node before writing it (see "Source Code Research"). Do NOT rely on memory.
5. Build update operations with CreateUpdateInfo + CreateAssociationUpdate, Flatten before persisting, then ApplyChanges@2 (use Append to collect).
6. Persist OutputData with SetPipelineExecutionResult@1 if the result must be readable via GetLatestPipelineExecution.

For annotated real-world examples covering all these patterns, read references/pipeline-examples.md.

Deploying and Testing Pipelines

After writing pipeline YAML, use the octo skill to deploy and test it. This skill handles YAML authoring; the octo skill handles all operational commands (deployment, execution, status, debugging).

Typical deployment workflow

1. Create runtime entities via octo-cli -c ImportRt -f <file> -w:

   • A System.Communication/DataFlow entity (logical grouping)
   • A System.Communication/Pipeline entity with ParentChild association to the DataFlow and Executes association to the target Adapter
   • Optionally a System.Communication/PipelineTrigger with ParentChild to the DataFlow and Triggers association to the Pipeline(s)

2. Deploy the pipeline YAML — octo-cli -c DeployPipeline --adapterId <id> --pipelineId <id> --file <yaml-path>

3. Verify deployment — octo-cli -c GetPipelineStatus --identifier <pipelineId> --json → confirm state = Deployed

4. Execute — octo-cli -c ExecutePipeline --identifier <pipelineId> → returns execution ID

5. Check execution — octo-cli -c GetLatestPipelineExecution --identifier <pipelineId> --json → status, duration, errors

6. Inspect debug points — octo-cli -c GetPipelineDebugPoints --identifier <pipelineId> --executionId <guid> --json → shows which nodes ran and their data. Debug capture must be enabled for a pipeline to record debug points (see toggle below).

7. Activate triggers (if using cron) — octo-cli -c DeployTriggers

Pipeline debug toggle (no redeploy)

Enable/disable per-node debug capture on a live pipeline without redeploying:

If the adapter is offline, the setting is persisted and applies on the next deploy.

Reassigning and deploying at the DataFlow level

Discovering available nodes at runtime

octo-cli -c GetPipelineSchema --adapterId <rtId> returns a JSON Schema (draft/2020-12) of all nodes available on a specific adapter — use it to discover what the target environment supports (e.g. custom plugins). To hand off any operational command, tell the user to invoke /octo <intent> (e.g. /octo deploy this pipeline).

Pipeline Troubleshooting

Silent failures: adapter returns HTTP 200 on error

The Mesh Adapter returns HTTP 200 for FromHttpRequest-triggered pipelines even when the pipeline fails internally. The GetLatestPipelineExecution may also show Status: null and DurationMs: null on failure.

Status: null does not always mean failure. For FromHttpRequest-triggered executions, the execution tracking may not complete before the HTTP response returns. This means Status: null can appear even on a successful run. Always verify by checking the actual data (e.g., query the entities that should have been created/updated) rather than relying solely on execution status.

Always check the adapter log after unexpected results:

logFiles/MeshAdapter.log

(Located in logFiles/MeshAdapter.log relative to the monorepo root)

Common error patterns in the adapter log

Debugging workflow

1. Trigger the pipeline (HTTP, ExecutePipeline, or scheduled)
2. Check execution: octo-cli -c GetLatestPipelineExecution --identifier <pipelineId> --json
3. If Status is null: the pipeline failed — check the adapter log (e.g., logFiles/MeshAdapter.log) for ERROR entries
4. Check debug tree: octo-cli -c GetPipelineDebugPoints — nodes missing from the tree never executed (pipeline stopped before reaching them)
5. The last node in the tree is usually where the error occurred
6. Read the handler source code for the failing node to understand what conditions cause the error — search for the error message text in the handler class

References

Priority order when understanding a node:

1. C# source code (ground truth) — config classes for properties/defaults, handlers for behavior. See "Source Code Research".
2. Pipeline JSON Schema (auto-generated, authoritative for property names/types/required): references/pipeline-schema-guide.md
3. Reference docs (hand-maintained summaries, may lag): SDK nodes references/node-reference-sdk.md, Mesh Adapter nodes references/node-reference-mesh.md
4. DataContext mechanics (paths, write modes, field filters, iterations): references/data-context-guide.md
5. Real examples (annotated pipelines): references/pipeline-examples.md

If there is ANY doubt about a node's properties or behavior, read the source code. Do not guess.