Pipeline Orchestrator

Found broken on inspection, fixed, then executed successfully end to end for the first time using placeholder estimators — the single most concrete "built it, it was broken, fixed it, ran it" story in this lab.

PurposeWire the state estimation kernel, the estimator suite, and governance into one real, callable pipeline.

StatusVerified — narrowly: the fix is verified and the execution path is verified operational. The numeric outputs are not verified in any predictive sense — they come from placeholder estimators with no real-world claim attached (see Outstanding below).

Built fromSTATE_ESTIMATION_KERNEL_v8.py, base_estimator.py, governance_layer.py — read directly, not assumed

Depends onComponent A Changelog (Fix 9 — the same fit()-before-predict() bug class)

Superseded by—

EvidenceReal execution transcript below, run against real dependency files.

System Data
↓
WhiteboxStateEstimationKernel
↓
Channel Adapter
↓
Estimators
↓
Governance
↓
Decision

Four real bugs, found by reading the dependencies, not assuming they matched

Wrong import name. The original imported StateEstimationKernel. The real class in STATE_ESTIMATION_KERNEL_v8.py is WhiteboxStateEstimationKernel. This would ImportError on the first run, immediately.
The kernel was constructed but never called. Its docstring claimed the kernel ran; the code never invoked it. Fixed with an explicit call to estimate_state(), plus an adapter (_kernel_result_to_channel_data) converting its output into the shape estimators expect — nothing downstream can silently do nothing anymore.
No fit()-before-predict() check. The exact bug class documented as Fix 9 in the Component A Changelog. Now tracked per estimator; predict() is refused, not silently run on an unfit model.
Estimator failures vanished into print() only. Now logged into governance.decision_log, so a failure shows up in the real audit trail, not just stdout.

What was NOT fixed — flagged, not guessed at

The kernel's own channel names (Strength/Dynamics/Interaction/Context) get mapped to S/D/I/C by first letter — the most literal reading, not a confirmed equivalence. This mapping is tracked on the Four-Channel Framework page as Exploratory — a first guess, not yet checked against data, same status as every other unvalidated domain mapping on this site.

Source (fixed)

#!/usr/bin/env python3
"""
pipeline_orchestrator.py (FIXED)
NCFCA Pipeline Orchestrator

Wires together:
- WhiteboxStateEstimationKernel (STATE_ESTIMATION_KERNEL_v8.py)
- Multiple BaseEstimator subclasses
- GovernanceLayer (governance_layer.py)
"""

from __future__ import annotations
from typing import Dict, Any, List, Optional
import numpy as np

from ..kernel.STATE_ESTIMATION_KERNEL_v8 import WhiteboxStateEstimationKernel, StateEstimationResult
from ..estimators.base_estimator import BaseEstimator
from ..governance.governance_layer import GovernanceLayer, GovernanceResult


# First-letter mapping from the kernel's own channel names -- a real
# assumption, not a verified fact. See /four-channel-framework.html.
_KERNEL_CHANNEL_TO_KEY = {
    "Strength": "S", "Dynamics": "D", "Interaction": "I", "Context": "C",
}


def _kernel_result_to_channel_data(result: StateEstimationResult) -> Dict[str, np.ndarray]:
    """Converts a StateEstimationResult's channel list into the
    channel_data dict shape estimators expect. Channels the kernel didn't
    populate are simply absent -- downstream estimators that need a
    specific key fail loudly rather than receive a fabricated zero."""
    channel_data: Dict[str, np.ndarray] = {}
    for ch in result.channels:
        key = _KERNEL_CHANNEL_TO_KEY.get(ch["name"])
        if key is None:
            continue
        channel_data[key] = np.array([ch["aggregate_value"]])
    return channel_data


class PipelineOrchestrator:
    def __init__(self, kernel: Optional[WhiteboxStateEstimationKernel] = None):
        self.kernel = kernel or WhiteboxStateEstimationKernel()
        self.estimators: List[BaseEstimator] = []
        self._fitted: Dict[str, bool] = {}
        self.governance = GovernanceLayer()
        self.last_kernel_result: Optional[StateEstimationResult] = None
        self.last_result: Optional[GovernanceResult] = None

    def register_estimator(self, estimator: BaseEstimator, already_fitted: bool = False):
        self.estimators.append(estimator)
        self._fitted[estimator.name] = already_fitted

    def fit_estimator(self, estimator_name: str, data: np.ndarray, **kwargs):
        for est in self.estimators:
            if est.name == estimator_name:
                est.fit(data, **kwargs)
                self._fitted[estimator_name] = True
                return
        raise KeyError(f"No registered estimator named '{estimator_name}'")

    def run(self, system_id: str, system_data: Dict[str, Any],
            raw_data: Optional[np.ndarray] = None) -> GovernanceResult:
        kernel_result = self.kernel.estimate_state(system_id, system_data)
        self.last_kernel_result = kernel_result
        channel_data = _kernel_result_to_channel_data(kernel_result)

        estimator_results = {}
        for est in self.estimators:
            if not self._fitted.get(est.name, False):
                msg = f"{est.name}: predict() skipped -- fit() was never called"
                self.governance.decision_log.append(f"[PIPELINE] {msg}")
                continue
            try:
                if "channel" in est.name.lower():
                    result = est.predict(channel_data)
                else:
                    result = est.predict(raw_data if raw_data is not None else np.array([]))
                estimator_results[est.name] = result
            except Exception as e:
                msg = f"{est.name}: predict() raised {type(e).__name__}: {e}"
                self.governance.decision_log.append(f"[PIPELINE] {msg}")

        governed = self.governance.govern(estimator_results)
        self.last_result = governed
        return governed

    def get_last_decision_log(self) -> List[str]:
        return self.governance.decision_log

    def get_last_kernel_regime(self) -> Optional[str]:
        if self.last_kernel_result is None:
            return None
        return self.last_kernel_result.regime.value

Real execution — first time this pipeline ran, end to end

Registered three stub estimators (clearly labeled placeholders, not real predictive models — see Backtest-Only Audit for what a real estimator implementation still requires): one exercising the channel-data path, one exercising the raw-data path, one deliberately left unfit to prove the fit-before-predict check actually blocks something.

======================================================================
REAL EXECUTION -- first end-to-end run of the fixed pipeline
======================================================================

[1] Kernel constructed: WhiteboxStateEstimationKernel
[2] Registered 3 estimators: ['stub_channel_estimator', 'stub_raw_estimator', 'stub_failing_estimator']
[3] Fit called for 2/3 estimators (fail_est deliberately left unfit)

[4] Running with system_data: {'spreads_bps': 45.0, 'volatility': 0.22, 'sentiment_bias': 0.15, 'past_returns': 3.2}
----------------------------------------------------------------------
[PIPELINE] stub_failing_estimator: predict() skipped -- fit() was never called
[GOVERNANCE] High disagreement detected: 0.946
[GOVERNANCE] Governance flag raised on extreme + disagreeing result
----------------------------------------------------------------------

[5] Kernel regime: MATURING
[6] Kernel channels produced: [('Strength', 0.0876), ('Dynamics', 0.2042), ('Interaction', 0.0661), ('Context', 0.0778)]
[7] Governance result: GovernanceResult(final_score=1.0544547628161298, confidence=0.5,
    active_estimators=['stub_channel_estimator', 'stub_raw_estimator'],
    violations=['High disagreement between estimators (std=0.946)',
                'Extreme score with high disagreement — governance flag raised'],
    metadata={'num_estimators_run': 2, 'num_estimators_accepted': 2, 'disagreement': 0.9455452371838701})

[8] Full decision log (3 entries):
    [PIPELINE] stub_failing_estimator: predict() skipped -- fit() was never called
    High disagreement detected: 0.946
    Governance flag raised on extreme + disagreeing result

======================================================================
RUN COMPLETE
======================================================================

On final_score = 1.0544, worth explaining rather than leaving unexplained: scores are not normalized to [0,1] here, and that's a real, honest gap, not a deliberate design choice. stub_channel_estimator averages channel values (~0.1 range); stub_raw_estimator naively averages the raw input array [1.0, 2.0, 3.0], which was never scaled to [0,1] at all — it's placeholder data, not a calibrated score. Governance's weighted average of those two lands above 1.0 simply because nothing currently enforces a common scale across estimators before aggregating them. The 0.2/0.8 circuit-breaker thresholds implicitly assume roughly-[0,1]-scaled inputs; this run shows that assumption isn't yet enforced anywhere in the pipeline. A real requirement for whatever estimator implementation replaces the stubs, not yet built.

What this run actually proves

The kernel executed successfully and produced structured output — a regime classification (MATURING) and four real channel values — demonstrating that the execution path is operational, not that the values themselves carry demonstrated predictive significance.
The channel adapter works — kernel output reached the channel-based estimator correctly.
The fit-before-predict check works — the unfit estimator was skipped and logged, not silently run.
Failure logging reaches the real audit trail — visible in the decision log, not lost to stdout.
The governance operator-precedence fix is confirmed against a real run, not just reasoned about on paper: score 1.05 (extreme) with disagreement 0.946 (high) correctly triggered the flag.

Outstanding — stated plainly

Every result above comes from stub estimators — placeholder computations with no predictive claim, built solely to prove the wiring works. No real estimator implementation exists yet. That is the actual current bottleneck on this pipeline, not the orchestrator itself — see the Reasoning Chain and Backtest-Only Audit for what's next.