Orchestrator Configuration Management

Pipeline-specific configuration with automatic context management and sibling inheritance preservation.

Status: STABLE Module: openhcs.core.orchestrator.orchestrator

The Problem: Configuration Synchronization Complexity

Pipelines need to override global configuration defaults (e.g., use a specific GPU, different memory backend) without affecting other pipelines. This requires synchronizing pipeline-specific config to thread-local context so that steps can access it. Without automatic synchronization, developers must manually call sync methods scattered throughout the code, leading to bugs where config changes aren’t propagated. Additionally, serialization needs fully-resolved config (no None values), while UI operations need inheritance-preserving config (None values indicate “use parent default”).

The Solution: Automatic Context Sync with Dual-Mode Access

The PipelineOrchestrator implements automatic synchronization: whenever pipeline config changes, it immediately updates thread-local context. Additionally, it provides dual-mode configuration access: one mode preserves None values for inheritance, another resolves all values for serialization. This eliminates manual sync calls and provides the right config format for each use case.

Overview

The PipelineOrchestrator implements sophisticated configuration management that bridges the gap between global application defaults and pipeline-specific overrides. This system ensures that configuration changes are automatically synchronized to thread-local context while preserving the None values necessary for sibling inheritance.

The orchestrator provides three key configuration management capabilities:

Automatic Context Synchronization - Pipeline config changes automatically sync to thread-local storage
Dual-Mode Configuration Access - Support for both inheritance-preserving and serialization-ready config access
Explicit Context Management - Context managers for scoped configuration operations

Auto-Sync Configuration Pattern

The orchestrator uses a property/setter pattern to automatically synchronize pipeline configuration changes to thread-local context, eliminating the need for manual synchronization calls.

Auto-sync property pattern from openhcs/core/orchestrator/orchestrator.py

    @pipeline_config.setter
    def pipeline_config(self, value: Optional["PipelineConfig"]) -> None:
        """Set pipeline configuration with auto-sync to thread-local context."""
        self._pipeline_config = value
        # CRITICAL FIX: Also update public attribute for dual-axis resolver discovery
        # This ensures the resolver can always find the current pipeline config
        if hasattr(self, "__dict__"):  # Avoid issues during __init__
            self.__dict__["pipeline_config"] = value
        if self._auto_sync_enabled and value is not None:
            self._sync_to_thread_local()

Benefits:

Eliminates Manual Sync Calls - No more scattered apply_pipeline_config() calls
Fail-Loud Behavior - Immediate errors if context setup fails
Thread Safety - Proper synchronization control during updates

Configuration Access Patterns

The orchestrator provides unified configuration access with explicit control over inheritance behavior.

Dual-Mode Configuration Access

Unified configuration access from openhcs/core/orchestrator/orchestrator.py

    def get_effective_config(
        self, *, for_serialization: bool = False
    ) -> GlobalPipelineConfig:
        """
        Get effective configuration for this orchestrator.

        Args:
            for_serialization: If True, resolves all values for pickling/storage.
                              If False, preserves None values for sibling inheritance.
        """

        if for_serialization:
            result = self.pipeline_config.to_base_config()
            return result
        else:
            # Reuse existing merged config logic from apply_pipeline_config
            shared_context = get_current_global_config(GlobalPipelineConfig)
            if not shared_context:
                raise RuntimeError(
                    "No global configuration context available for merging"
                )

            result = _create_merged_config(self.pipeline_config, shared_context)
            return result

Usage Patterns:

# For UI operations (preserves sibling inheritance)
ui_config = orchestrator.get_effective_config(for_serialization=False)

# For compilation/storage (resolves all values)
storage_config = orchestrator.get_effective_config(for_serialization=True)

Pure Function Configuration Merging

Configuration merging is implemented as a pure function following OpenHCS stateless architecture principles:

Pure function for configuration merging from openhcs/core/orchestrator/orchestrator.py

def _create_merged_config(
    pipeline_config: "PipelineConfig", global_config: GlobalPipelineConfig
) -> GlobalPipelineConfig:
    """
    Pure function for creating merged config that preserves None values for sibling inheritance.

    Follows OpenHCS stateless architecture principles - no side effects, explicit dependencies.
    Extracted from apply_pipeline_config to eliminate code duplication.
    """
    logger.debug(
        f"Starting merge with pipeline_config={type(pipeline_config)} and global_config={type(global_config)}"
    )

    merged_config_values = {}
    for field in fields(GlobalPipelineConfig):
        # CRITICAL: Access raw stored value from __dict__ to avoid lazy resolution fallback to MRO defaults
        # For lazy dataclasses, getattr() triggers resolution which falls back to GlobalPipelineConfig defaults
        # We need the actual None value to know if it should inherit from global config
        pipeline_value = pipeline_config.__dict__.get(field.name)

        if pipeline_value is not None:
            # CRITICAL FIX: For lazy configs, merge with global config BEFORE converting to base
            # This ensures None values in lazy configs resolve to global values
            # Then convert to base config to store in thread-local context
            if hasattr(pipeline_value, "to_base_config"):
                # This is a lazy config - merge with global config first
                global_value = getattr(global_config, field.name)
                from dataclasses import is_dataclass

                if is_dataclass(global_value):
                    # Merge lazy config with global config to resolve None values
                    merged_lazy = _merge_nested_dataclass(pipeline_value, global_value)
                    # Now convert merged result to base config
                    converted_value = (
                        merged_lazy.to_base_config()
                        if hasattr(merged_lazy, "to_base_config")
                        else merged_lazy
                    )
                    merged_config_values[field.name] = converted_value
                else:
                    # No global value to merge with, just convert to base
                    converted_value = pipeline_value.to_base_config()
                    merged_config_values[field.name] = converted_value
            else:
                # CRITICAL FIX: For base dataclass configs, merge nested fields
                # This ensures None values in nested configs resolve to global values
                global_value = getattr(global_config, field.name)
                from dataclasses import is_dataclass

                if is_dataclass(pipeline_value) and is_dataclass(global_value):
                    merged_config_values[field.name] = _merge_nested_dataclass(
                        pipeline_value, global_value
                    )
                else:
                    # Regular value - use as-is
                    merged_config_values[field.name] = pipeline_value
        else:
            global_value = getattr(global_config, field.name)
            merged_config_values[field.name] = global_value

    result = GlobalPipelineConfig(**merged_config_values)
    return result

Design Principles:

Stateless Function - No side effects, explicit dependencies
Fail-Loud Behavior - No defensive programming with getattr fallbacks
Code Reuse - Eliminates duplication between methods

Context Manager Pattern

The orchestrator provides explicit context managers for operations requiring specific configuration contexts.

Scoped Configuration Context

Usage Examples:

# UI operations requiring sibling inheritance
with orchestrator.config_context(for_serialization=False):
    editor = StepEditorWindow(step_data, orchestrator=orchestrator)

# Compilation operations requiring resolved values
with orchestrator.config_context(for_serialization=True):
    compiled_context = compile_pipeline_step(step, context)

Configuration Inheritance Preservation

The system carefully preserves None values in configuration objects to maintain sibling inheritance chains.

Merged vs Resolved Configuration

Merged Configuration (for_serialization=False):

Preserves None values from pipeline config
Enables sibling inheritance (materialization_defaults → path_planning)
Used for UI operations and step editing

Resolved Configuration (for_serialization=True):

Resolves all None values to concrete values
Suitable for serialization and storage
Used for compilation and pickling operations

Critical Implementation Detail:

# ✅ CORRECT: Preserves None values for sibling inheritance
merged_config = _create_merged_config(pipeline_config, global_config)

# ❌ INCORRECT: Resolves None values, breaking inheritance
resolved_config = pipeline_config.to_base_config()

Integration with Lazy Configuration System

The orchestrator configuration management integrates seamlessly with the lazy configuration system documented in Dynamic Dataclass Factory System.

Thread-Local Context Flow

Pipeline Config Assignment - Auto-sync triggers when orchestrator.pipeline_config is set
Merged Config Creation - Pure function creates config preserving None values
Thread-Local Update - Merged config becomes active context for lazy resolution
Sibling Inheritance - Lazy configs resolve using preserved None values

Resolution Chain:

# Step-level lazy config resolution chain:

# 1. Check step's explicit value
step_value = step.materialization_config.output_dir_suffix
if step_value is not None:
    return step_value

# 2. Resolve from orchestrator context (merged config with None preservation)
orchestrator_context = get_current_global_config(GlobalPipelineConfig)
orchestrator_value = orchestrator_context.materialization_defaults.output_dir_suffix
if orchestrator_value is not None:
    return orchestrator_value

# 3. Sibling inheritance (materialization_defaults → path_planning)
sibling_value = orchestrator_context.path_planning.output_dir_suffix
return sibling_value

Benefits and Design Rationale

Architectural Benefits:

Eliminates Code Duplication - Single pure function for config merging
Explicit Dependencies - Clear parameter-based function contracts
Fail-Loud Behavior - Immediate errors instead of silent degradation
Stateless Design - Pure functions with no hidden state

User Experience Benefits:

Automatic Synchronization - No manual context management required
Preserved User Edits - Sibling inheritance maintains user intentions
Explicit Scoping - Context managers make dependencies clear