cf-pipeline-engine

The Cogniflow pipeline engine implemented in C++ (compiler, scheduler, runtime).

This folder is structured as a Python package for consistency with the other cf_* components, even though the core implementation is native C++ and built via CMake.

Build (CMake)

cmake -S . -B build
cmake --build build

Python package wrapper

The Python package is intentionally thin and provides access to the packaged native engine assets:

• cf_pipeline_engine.cf_pipeline_v2_path()
• cf_pipeline_engine.cf_type_registry_path()
• cf_pipeline_engine.resolve_cf_pipeline_v2_executable()

Published distribution name:

pip install cf-pipeline-engine

The published wheel installs:

• bin/cf_pipeline_v2(.exe)
• bin/type_registry.v0.json

The engine compiles natively only against cf_package_contracts. Owner providers from cf-pipeline-sdk, cf-opcua-server, and cf-datahive are resolved at runtime via the provider registry and exported contract symbols.

cfio:OpcuaReaderStep is executed by the cf_basic_io plugin and consumes the installed cf-opcua-server owner contract injected by the engine. If the owner provider needs a custom script location, set CF_OPCUA_SERVER_CMD to the absolute script or executable path.

Python CLI

The legacy standalone Python wrapper package has been retired. Its CLI surface now lives in this cf_pipeline_engine package.

Direct invocation remains policy-gated by CF_ALLOW_DIRECT_ENGINE:

CF_ALLOW_DIRECT_ENGINE=1 python -m cf_pipeline_engine.cli \
  --pipeline sandcastle/cf_pipeline/cf_pipeline_engine/examples/opcua_fifo_avg_to_duckdb_parquet_triggered.nq \
  --interval 1 \
  --duration 10

Equivalent module entrypoint:

CF_ALLOW_DIRECT_ENGINE=1 python -m cf_pipeline_engine --help

Publishing

cf_pipeline_engine is published with the dedicated Windows workflow:

• Workflow: .github/workflows/cf_pipeline_engine_windows_publish.yml
• Package directory: sandcastle/cf_pipeline/cf_pipeline_engine
• PyPI tag: cf-pipeline-engine-v<version>
• TestPyPI tag: cf-pipeline-engine-v<version>-test

Local preflight:

powershell -ExecutionPolicy Bypass -File scripts/mimic_windows_python_publish_workflow.ps1 `
  -WorkflowFile .github/workflows/cf_pipeline_engine_windows_publish.yml `
  -PackageDir sandcastle/cf_pipeline/cf_pipeline_engine `
  -PythonExe py `
  -PythonVersion 3.14

Queue a dry-run dispatch:

powershell -ExecutionPolicy Bypass -File scripts/queue_windows_python_publish_workflow.ps1 `
  -WorkflowFile .github/workflows/cf_pipeline_engine_windows_publish.yml `
  -PackageDir sandcastle/cf_pipeline/cf_pipeline_engine `
  -PublishTarget testpypi `
  -Ref main `
  -RequireLocalPass `
  -DryRun

OPC UA Demo Pipeline Sink

Packaged demo resources are owned by cf_setup. The repo keeps source examples under examples/ for development and test use.

The existing demo pipeline examples/opcua_fifo_avg_to_duckdb_parquet_triggered.nq uses plugin-executed cfio:OpcuaReaderStep and cfsink:DataHiveParquetSinkStep.

cf_basic_io consumes the injected OPC UA owner contract exported by cf-opcua-server, and cf_basic_sinks consumes the injected Data Hive sink contract exported by cf-datahive, producing one committed data hive run with 20 rows (cycle_id 0..19).

Run the one-click demo:

cf install --dev --clean --run-demo --show-details

Expected Data Hive evidence after one demo session, owned by the DataHive contracts:

• workspace/<data_hive>/opcua_fifo_avg/latest.txt
• workspace/<data_hive>/opcua_fifo_avg/latest_session.txt
• workspace/<data_hive>/opcua_fifo_avg/<yyyy>/<mm>/<dd>/<session_id>/runs/<run_id>/manifest.json
• workspace/<data_hive>/opcua_fifo_avg/<yyyy>/<mm>/<dd>/<session_id>/tables/measurements/run_id=<run_id>/part-*.parquet

Consumers should inspect this output through DataHiveRunsContractV1 rather than reimplementing path discovery in package code.