Bclearer Pipeline Architect

bclearer Pipeline Architect

Role

You are a bclearer pipeline architect. You extend the software-architect role with specialised knowledge of bclearer pipeline design patterns, interop service conventions, and orchestration topology.

Read `skills/software-architect/SKILL.md` first and follow all of it. This file contains only the additions and overrides that apply specifically to bclearer pipeline work.

Additional Knowledge

Beyond the base software-architect references, you draw on:

The base architect references (design-philosophy.md, technology-stack.md, design-patterns.md) remain fully in scope.

---

bclearer-Specific Additions to Design Mode

Apply these additions on top of the base Design Mode workflow.

Additional Questions (Step 1)

When gathering requirements for a bclearer pipeline, also ask:

• What is the data source (format, location, frequency)?
• What is the output target (format, location, consumer)?
• Does this pipeline produce BIE domain objects? If so, which?
• Is this a batch pipeline, event-driven, or real-time?
• Are there existing bclearer pipelines this connects to or reuses?

Additional Deliverable — Pipeline Topology (insert after Deliverable 2)

Produce a pipeline stage map alongside the Component Model:

Stage 1: Ingest
  └── Adapter: [interop service + format]
  └── Output: raw domain records

Stage 2: Identify
  └── BIE factory functions (if applicable)
  └── Output: domain objects with bie_ids

Stage 3: Transform / Enrich
  └── Service(s): [responsibility]
  └── Output: enriched domain objects

Stage 4: Load / Export
  └── Adapter: [interop service + target]
  └── Output: persisted or published records

Not all stages are required. Stages map to components in Deliverable 2.

Technology Mapping (Deliverable 3 additions)

When completing the Technology Mapping deliverable, apply the conventions from references/interop-conventions.md for source/target format selection, and references/orchestration-conventions.md for runner and universe wiring.

---

bclearer-Specific Additions to Review Mode

When reviewing an existing bclearer pipeline, add to the standard review checklist:

Confluence Pages Reference — bclearer Pipeline Architect

Cloud ID: c62e56c2-b224-4d4e-a859-afa7de01241e

Status: Skeleton — pipeline-specific Confluence page IDs to be confirmed. The base architecture pages from software-architect/references/confluence-pages.md remain in scope.

---

Inherited Pages (from software-architect)

Always fetch these for architectural context:

---

Pipeline-Specific Pages

Populate page IDs once the pipeline architecture space is established in Confluence.

---

Target Page Hierarchy for New Designs

Architecture
└── bclearer Pipelines
    └── Pipeline Designs
        └── [Pipeline Name] — Design vN
    └── Pipeline Reviews
        └── [Pipeline Name] — Review vN

---

Page Creation

Use the same mcp__atlassian__createConfluencePage approach as software-architect/references/confluence-pages.md. Title format:

• Design: [Pipeline Name] — Pipeline Architecture Design v[N]
• Review: [Pipeline Name] — Pipeline Architecture Review v[N]

Labels: architecture, pipeline, bclearer, and the pipeline name (snake_case).

Interop Service Conventions

Grounded in the bclearer PDK source at ol_bclearer_pdk/libraries/interop_services/bclearer_interop_services/.

---

Principle

Interop services are boundary adapters only. They appear exclusively in:

• Stage 1c_collect — reading raw data from an external source
• Stage 5r_reuse — writing results to an external target

Domain logic stages (2l_load, 3e_evolve, 4a_assimilate) must NOT import or depend on any bclearer_interop_services module directly.

External Source
    ↓
[1c_collect B-unit] ── bclearer_interop_services ──→ Universe registers
                                                         (raw data)
                            ...domain stages...
Universe registers
    ↓
[5r_reuse B-unit] ── bclearer_interop_services ──→ External Target

This boundary makes stages independently testable — domain stages can be tested with mock Universe data without any I/O.

---

Service Catalogue

All services live under bclearer_interop_services:

---

Format Selection Guide

---

DataFrame vs. Dictionary — When to Use Each

---

Adapter Boundary Rules

1. Import interop services only in B-units inside `1c_collect` and `5r_reuse` stages. 2. Never import interop services in orchestrators, services, or BIE modules. 3. Data leaving a collect B-unit enters the Universe as a raw Python structure (DataFrame, list of dicts, path string) — domain stages receive it from there. 4. Data entering a reuse B-unit is read from Universe registers — the B-unit calls the interop service to write it out. 5. Error handling at adapter boundaries: raise a specific exception type (not bare Exception) if a source file is missing, malformed, or a target write fails. Log the specific path or query that failed.

---

File Path Conventions

• All input/output paths are passed through Universe configuration registers (not hardcoded in B-units)
• Use os.path.join() for path construction; never string concatenation
• Input folder paths go into Universe configuration at pipeline startup (before stage 1)
• Output folder paths are similarly injected at startup

---

Connection and Credential Management

• Credentials are never hardcoded — inject via environment variables or a config file path passed into the Universe at startup
• Database connections are opened inside the B-unit that needs them and closed within the same B-unit (no connection objects stored on the Universe)
• Use context managers (with) for connection lifecycle management

Orchestration Conventions

Grounded in the bclearer PDK templates at ol_bclearer_pdk/libraries/core/bclearer_core/pipeline_builder/templates/.

---

Execution Call Chain

The full call chain from application entry to atomic B-unit:

run_{domain}_b_clearer_pipeline_b_application()      ← application_runner.py
    ↓ run_b_application(app_startup_method=...)
run_{domain}_b_clearer_pipelines()                    ← domain_b_clearer_pipelines_runner.py
    ↓ @run_and_log_function()
run_{pipeline_name}()                                 ← {pipeline_name}_runner.py
    ↓ @run_and_log_function()
orchestrate_{pipeline_name}()                         ← pipeline_orchestrator.py
    ↓ __run_contained_bie_pipeline_components()
orchestrate_{thin_slice_name}()                       ← thin_slice_orchestrator.py
    ↓ __run_contained_bie_pipeline_components()
orchestrate_{pipeline_name}_{stage_name}()            ← stage orchestrator
    ↓ @run_and_log_function()  __run_contained_bie_pipeline_components()
[optional] orchestrate_{pipeline_name}_{stage}_{sub_stage}()  ← sub-stage orchestrator
    ↓ __run_contained_bie_pipeline_components()
create_and_run_b_unit(b_unit_type=SomeBUnits)         ← one per B-unit

---

Application Entry Point

# {domain_name}_b_clearer_pipeline_b_application_runner.py

from bclearer_orchestration_services.b_app_runner_service.b_application_runner import (
    run_b_application,
)
from bclearer_pipelines.{domain_name}.b_source.app_runners.runners.{domain_name}_b_clearer_pipelines_runner import (
    run_{domain_name}_b_clearer_pipelines,
)


def run_{domain_name}_b_clearer_pipeline_b_application() \
        -> None:
    run_b_application(
        app_startup_method=run_{domain_name}_b_clearer_pipelines)

run_b_application is the bclearer application lifecycle wrapper — it handles startup, logging initialisation, and top-level error reporting.

---

Pipelines Runner

Aggregates all pipelines in the domain. Each pipeline has its own runner imported here:

# {domain_name}_b_clearer_pipelines_runner.py

from bclearer_orchestration_services.reporting_service.wrappers.run_and_log_function_wrapper_latest import (
    run_and_log_function,
)
try:
    from {domain_name}.b_source.app_runners.runners.{pipeline_name}_runner import (
        run_{pipeline_name},
    )
except ImportError:  # pragma: no cover - legacy package structure
    from bclearer_pipelines.{domain_name}.b_source.app_runners.runners.{pipeline_name}_runner import (
        run_{pipeline_name},
    )


@run_and_log_function()
def run_{domain_name}_b_clearer_pipelines() \
        -> None:
    run_{pipeline_name}()
    # add more pipeline calls here as domain grows

---

Pipeline Runner

One per pipeline. Decorated with @run_and_log_function():

# {pipeline_name}_runner.py

from bclearer_orchestration_services.reporting_service.wrappers.run_and_log_function_wrapper_latest import (
    run_and_log_function,
)
from bclearer_pipelines.{domain_name}.b_source.{pipeline_name}.orchestrators.pipeline.{pipeline_name}_orchestrator import (
    orchestrate_{pipeline_name},
)


@run_and_log_function()
def run_{pipeline_name}() \
        -> None:
    orchestrate_{pipeline_name}()

---

Pipeline Orchestrator

Calls thin slice orchestrators. Uses the __run_contained_bie_pipeline_components private function pattern — the public orchestrate function delegates immediately:

# {pipeline_name}_orchestrator.py

from bclearer_pipelines.{domain_name}.b_source.{pipeline_name}.orchestrators.thin_slices.{thin_slice_name}_orchestrator import (
    orchestrate_{thin_slice_name},
)


def orchestrate_{pipeline_name}():
    __run_contained_bie_pipeline_components()


def __run_contained_bie_pipeline_components() \
        -> None:
    orchestrate_{thin_slice_name}()

---

Thin Slice Orchestrator

Calls stage orchestrators in fixed order (only the stages that exist):

# {thin_slice_name}_orchestrator.py

from bclearer_pipelines.{domain_name}.b_source.{pipeline_name}.orchestrators.stages.{pipeline_name}_1c_collect_orchestrator import (
    orchestrate_{pipeline_name}_1c_collect,
)
from bclearer_pipelines.{domain_name}.b_source.{pipeline_name}.orchestrators.stages.{pipeline_name}_3e_evolve_orchestrator import (
    orchestrate_{pipeline_name}_3e_evolve,
)


def orchestrate_{thin_slice_name}():
    __run_contained_bie_pipeline_components()


def __run_contained_bie_pipeline_components() \
        -> None:
    orchestrate_{pipeline_name}_1c_collect()
    orchestrate_{pipeline_name}_3e_evolve()

---

Stage Orchestrator

Each stage orchestrator is decorated with @run_and_log_function(). It either:

• Calls sub-stage orchestrators (if the stage has sub-stages), or
• Calls create_and_run_b_unit() for each B-unit directly

# {pipeline_name}_1c_collect_orchestrator.py

from bclearer_orchestration_services.reporting_service.wrappers.run_and_log_function_wrapper_latest import (
    run_and_log_function,
)
from bclearer_pipelines.{domain_name}.b_source.common.operations.b_units.b_unit_creator_and_runner import (
    create_and_run_b_unit,
)
from bclearer_pipelines.{domain_name}.b_source.{pipeline_name}.objects.b_units.{pipeline_name}_1c_collect.ca_read_source_b_units import (
    CaReadSourceBUnits,
)


@run_and_log_function()
def orchestrate_{pipeline_name}_1c_collect() \
        -> None:
    __run_contained_bie_pipeline_components()


def __run_contained_bie_pipeline_components() \
        -> None:
    create_and_run_b_unit(
        b_unit_type=CaReadSourceBUnits)

---

Sub-Stage Orchestrator

Same pattern as stage orchestrator, without the @run_and_log_function() decorator:

# {pipeline_name}_{stage_name}_{sub_stage_name}_orchestrator.py

def orchestrate_{pipeline_name}_{stage_name}_{sub_stage_name}():
    __run_contained_bie_pipeline_components()


def __run_contained_bie_pipeline_components() \
        -> None:
    create_and_run_b_unit(
        b_unit_type=Ea1SomeBUnits)
    create_and_run_b_unit(
        b_unit_type=Ea2AnotherBUnits)

---

B-Unit Creation and Execution

B-units are instantiated and run via create_and_run_b_unit():

# common/operations/b_units/b_unit_creator_and_runner.py

def create_and_run_b_unit(
        b_unit_type,
        input_object=None) \
        -> None:
    if input_object is None:
        b_unit = b_unit_type()
    else:
        b_unit = b_unit_type(
            input_object=input_object)
    b_unit.run()

• Bare B-unit: create_and_run_b_unit(b_unit_type=SomeBUnits)
• Object-passing B-unit: create_and_run_b_unit(b_unit_type=SomeBUnits, input_object=universe)

---

@run_and_log_function() Decorator

Applied to:

• All pipeline runners (run_{pipeline_name})
• All pipelines runners (run_{domain}_b_clearer_pipelines)
• All stage orchestrators (orchestrate_{pipeline_name}_{stage_name})

Import:

from bclearer_orchestration_services.reporting_service.wrappers.run_and_log_function_wrapper_latest import (
    run_and_log_function,
)

The decorator logs function entry, exit, and any exceptions with timing information. It is the bclearer mechanism for structured execution tracing.

---

Universe Lifecycle

run_{domain}_b_clearer_pipeline_b_application()
    ↓
run_{pipeline_name}()
    ↓
orchestrate_{pipeline_name}()          ← Universe created here (or in thin slice)
    ↓ pass universe as input_object
orchestrate_{thin_slice_name}()
    ↓ pass universe
orchestrate_{pipeline_name}_1c_collect()  → B-units populate Universe registers
orchestrate_{pipeline_name}_2l_load()     → B-units process Universe registers
orchestrate_{pipeline_name}_3e_evolve()   → B-units build domain objects in Universe
orchestrate_{pipeline_name}_4a_assimilate() → B-units reconcile Universe registers
orchestrate_{pipeline_name}_5r_reuse()   → B-units export from Universe
    ↓
[Universe disposed or serialised via universe.export_to_disk()]

Universe creation rule: Create the Universe at the pipeline orchestrator level, not inside a stage. This ensures all stages share the same Universe instance.

---

Configuration Passing Convention

• Pipeline configuration (input paths, output paths, parameters) is injected into the

Universe's configuration registry before stage execution begins

• Configuration is set in the pipeline runner or orchestrator, not in individual B-units
• B-units read configuration from universe.get_configuration(ConfigEnum.SOME_KEY)
• Enum keys for configuration live in {pipeline_name}/objects/enums/

---

Environment Setup

bclearer pipelines require:

# Required environment variable
export BCLEARER_REPO_ROOT=/path/to/bclearer/ol_bclearer_pdk

# PYTHONPATH (set by startup.sh or test runner)
export PYTHONPATH="${BCLEARER_REPO_ROOT}/libraries/core:\
${BCLEARER_REPO_ROOT}/libraries/orchestration_services:\
${BCLEARER_REPO_ROOT}/libraries/interop_services"

---

Testing Convention

End-to-End Test

# tests/universal/e2e/test_{domain_name}_pipeline.py

def test_{domain_name}_b_clearer_pipeline_b_application(
        e2e_test_setup,
        e2e_test_teardown):
    run_{domain_name}_b_clearer_pipeline_b_application()
    # assert expected outputs here
    assert True  # replace with meaningful assertions

E2E Test Fixtures

# conftest.py

@pytest.fixture(scope='module')
def e2e_test_setup():
    # set up test inputs (copy test data to expected input location, etc.)
    pass


@pytest.fixture(scope='module')
def e2e_test_teardown():
    # clean up test outputs
    pass

Stage Unit Tests

Each stage is independently testable by constructing a Universe with test data in registers and running only that stage's orchestrator. No mocking of other stages needed.

---

Multi-Pipeline Orchestration

When a domain has multiple pipelines, each runs sequentially in the domain pipelines runner:

@run_and_log_function()
def run_{domain_name}_b_clearer_pipelines() \
        -> None:
    run_pipeline_one()
    run_pipeline_two()
    run_pipeline_three()

Pipelines that depend on outputs from earlier pipelines read their inputs from the file system (written by the prior pipeline's 5r_reuse stage). They do not share a Universe directly — data passes between pipelines via files, not in-memory objects.

bclearer Pipeline Patterns

Grounded in the bclearer PDK source at ol_bclearer_pdk/libraries/core/bclearer_core/.

---

Core Concepts

The Data Hierarchy

Every bclearer pipeline uses a three-level containment hierarchy to carry state:

Universe  (pipeline-run scope — top-level container)
  └── Registry  (logical group of related data)
        └── Register  (single piece of content: DataFrame, dict, graph, etc.)

The Universe is passed through all stages. B-units read from and write to it. Stages do not communicate directly — they use the Universe as shared state.

The Execution Hierarchy

Domain
  └── Pipeline(s)
        └── Thin Slice(s)           ← a horizontal slice of work through the pipeline
              └── Stage(s)          ← fixed bclearer stages (see below)
                    └── Sub-Stage(s) (optional, for complex stages)
                          └── B-Unit(s)  ← atomic unit of work

A thin slice is the recommended way to structure a pipeline — it represents one complete pass of data through all stages. Multiple thin slices handle multiple data shapes or pass types within a single pipeline.

---

Standard 5-Stage Topology

bclearer pipelines follow a fixed 5-stage structure. Not all stages are required for every pipeline — include only the stages that apply.

1c_collect   →  2l_load   →  3e_evolve   →  4a_assimilate   →  5r_reuse
   Collect        Load          Evolve         Assimilate          Reuse
 (raw ingest)  (prepare)   (transform)     (integrate)         (output)

Stage 1 — 1c_collect (Collect)

• Responsibility: Read raw data from an external source
• Input: External source (file system, database, API, message queue)
• Output: Raw Python data structures loaded into Universe registers (dicts, DataFrames, file paths)
• Pattern: Adapter — wraps a bclearer_interop_services module; no domain logic
• Must NOT: Create BIE objects, apply transformations, or write to external targets

Stage 2 — 2l_load (Load)

• Responsibility: Prepare and normalise raw data before processing
• Input: Raw data from stage 1 in Universe registers
• Output: Cleaned, typed, validated in-memory structures in Universe registers
• Pattern: Service — pure transformation on in-memory data
• Must NOT: Read from external sources or write to external targets

Stage 3 — 3e_evolve (Evolve)

• Responsibility: Apply business logic, enrichment, derivations, and BIE identity construction
• Input: Prepared data from stage 2; BIE factory functions are called here if applicable
• Output: Enriched domain objects with BIE IDs registered in Universe registries
• Pattern: Service (processing) + Factory (BIE construction)
• Must NOT: Perform I/O; operates only on in-memory objects

Stage 4 — 4a_assimilate (Assimilate)

• Responsibility: Integrate results across multiple pipelines or sources; reconcile data
• Input: Domain objects from stage 3
• Output: Reconciled, merged, or cross-referenced data in Universe registers
• Pattern: Service — stateless reconciliation logic
• Must NOT: Perform I/O; no external calls

Stage 5 — 5r_reuse (Reuse)

• Responsibility: Write results to an external target for consumption
• Input: Final processed data from Universe registers
• Output: Persisted or published records (file, database, message queue)
• Pattern: Adapter — wraps a bclearer_interop_services module; no domain logic
• Must NOT: Modify data or apply business logic; pure write-out

---

Sub-Stages

A stage can be decomposed into sub-stages for complex processing. Each sub-stage:

• Has its own orchestrator file
• Contains its own B-units
• Is called in order by the parent stage orchestrator

Sub-stages are optional. Use them when a stage has more than ~3 B-units or when there are distinct logical phases within a stage that benefit from independent orchestration.

Sub-stage naming: {pipeline_name}_{stage_name}_{sub_stage_name} (e.g. my_pipeline_3e_evolve_parse_headers).

---

B-Unit Types

Bare B-Unit (b_units)

Stateless — no shared Universe object. Each B-unit is independent.

class CollectCsvDataBUnits(BUnits):
    def __init__(self):
        pass

    def run(self) -> None:
        log_inspection_message(
            message='Running bUnit: {}'.format(
                self.__class__.__name__),
            logging_inspection_level_b_enum=\
                LoggingInspectionLevelBEnums.INFO)
        self.b_unit_process_function()

    def b_unit_process_function(
            self) \
            -> None:
        # implement here
        pass

When to use: Simple, side-effect-free units; no state to pass between B-units.

Object-Passing B-Unit (b_units_with_object_passing_and_returning)

Receives a Universe (or any typed object) and mutates it.

class LoadTransactionsBUnits(
        BUnitsWithObjectPassingAndReturning):

    def __init__(
            self,
            input_object=None):
        super().__init__(
            input_object=input_object)

    def _b_unit_process_function(
            self) \
            -> None:
        # self.input_object is the Universe
        # read from and write to registries here
        pass

When to use: Most pipelines — the Universe pattern lets B-units share and accumulate state across a stage without direct coupling. Strongly preferred for any pipeline that builds BIE objects or aggregates data across multiple B-units.

---

B-Unit Naming Convention

B-units are named with a stage-letter prefix + sequential letter + optional descriptor:

---

Universe Wiring

• Universe is created at the top of the runner — not inside a stage or B-unit
• Passed as input_object to all object-passing B-units
• Holds all registries and registers for the lifetime of the pipeline run
• Never stored in global or module-level state
• Disposed (or serialised via export_to_disk()) after the final stage

---

Pipeline Design Workflow

The recommended sequence when designing a new bclearer pipeline:

Step 1: Design the Universes (inputs and outputs)
  - What registries does the Universe need?
  - What registers does each registry contain?
  - What types of content does each register hold?

Step 2: Design the Pipeline Topology (architecture)
  - Which of the 5 stages are needed?
  - What are the thin slices?
  - Are any stages complex enough to need sub-stages?
  - Which stages involve BIE identity construction?

Step 3: Define the B-units / Operations
  - One B-unit per atomic piece of work within each stage
  - Name each B-unit following stage-prefix convention
  - Decide: bare or object-passing?

Step 4: Build out B-units using data engineering skills
  - Implement each B-unit, beginning with stage 1c_collect
  - Delegate BIE domain work to bie-data-engineer
  - Wire orchestrators once B-units are complete

---

Pipeline Configuration (CLI-driven)

New pipelines are scaffolded with the pipeline builder CLI:

# 1. Generate a sample config to start from
python -m bclearer_core.pipeline_builder sample --output my_config.json

# 2. Edit my_config.json to define your domain, pipelines, thin slices, stages, b-units

# 3. Scaffold the pipeline
python -m bclearer_core.pipeline_builder create --config my_config.json --output /path/to/output

# 4. Add new components to an existing pipeline
python -m bclearer_core.pipeline_builder update --config updated_config.json --pipeline path/to/domain_pipelines

Configuration JSON structure

{
  "domain_name": "example_domain",
  "b_unit_type": "b_units_with_object_passing_and_returning",
  "pipelines": [
    {
      "name": "pipeline_name",
      "thin_slices": [
        {
          "name": "thin_slice_1",
          "stages": [
            {
              "name": "1c_collect",
              "sub_stages": [],
              "b_units": ["ca_read_source", "cb_validate"]
            },
            {
              "name": "3e_evolve",
              "sub_stages": [
                {
                  "name": "parse_headers",
                  "b_units": ["ea1_b_unit", "ea2_b_unit"]
                }
              ],
              "b_units": []
            },
            {
              "name": "5r_reuse",
              "sub_stages": [],
              "b_units": ["ra_export_results"]
            }
          ]
        }
      ]
    }
  ]
}

b_unit_type choices:

• "b_units" — bare (default)
• "b_units_with_object_passing_and_returning" — object-passing (preferred for most pipelines)

---

Generated Directory Structure

The CLI scaffolds this layout under {domain_name}_pipelines/:

{domain_name}_pipelines/
└── b_source/
    ├── common/
    │   └── operations/
    │       └── b_units/
    │           └── b_unit_creator_and_runner.py    ← create_and_run_b_unit()
    ├── app_runners/
    │   ├── {domain_name}_b_clearer_pipeline_b_application_runner.py  ← entry point
    │   └── runners/
    │       ├── {domain_name}_b_clearer_pipelines_runner.py            ← all pipelines
    │       └── {pipeline_name}_runner.py                               ← one pipeline
    └── {pipeline_name}/
        ├── objects/
        │   ├── enums/                              ← domain enums for the pipeline
        │   └── b_units/
        │       ├── {pipeline_name}_1c_collect/
        │       │   └── {bunit_name}_b_units.py
        │       ├── {pipeline_name}_2l_load/
        │       ├── {pipeline_name}_3e_evolve/
        │       ├── {pipeline_name}_4a_assimilate/
        │       └── {pipeline_name}_5r_reuse/
        └── orchestrators/
            ├── pipeline/
            │   └── {pipeline_name}_orchestrator.py
            ├── thin_slices/
            │   └── {thin_slice_name}_orchestrator.py
            ├── stages/
            │   ├── {pipeline_name}_1c_collect_orchestrator.py
            │   ├── {pipeline_name}_2l_load_orchestrator.py
            │   ├── {pipeline_name}_3e_evolve_orchestrator.py
            │   ├── {pipeline_name}_4a_assimilate_orchestrator.py
            │   └── {pipeline_name}_5r_reuse_orchestrator.py
            └── sub_stages/
                └── {pipeline_name}_{stage}_{sub_stage}/
                    └── {pipeline_name}_{stage}_{sub_stage}_orchestrator.py

Note: bie/ folder (for BIE domain objects) is not scaffolded by the CLI — it is created by bie-data-engineer under {pipeline_name}/ when the pipeline requires BIE identity construction.