Context Management System

Configuration resolution requires tracking which configs are active at any point during execution. OpenHCS uses Python’s contextvars for clean context stacking without frame introspection.

from openhcs.config_framework import config_context

with config_context(global_config):
    with config_context(pipeline_config):
        # Both configs available for resolution
        lazy_instance.field_name  # Resolves through both contexts

The config_context() manager extracts dataclass fields and merges them into the context stack, enabling lazy resolution without explicit parameter passing.

Context Stacking

Contexts stack via contextvars.ContextVar:

# openhcs/config_framework/context_manager.py
_config_context_base: ContextVar[Optional[Dict[str, Any]]] = ContextVar(
    'config_context_base',
    default=None
)

# Track types pushed via config_context() for hierarchy queries
context_type_stack: ContextVar[Tuple[Type, ...]] = ContextVar(
    'context_type_stack',
    default=()
)

@contextmanager
def config_context(obj):
    """Stack a configuration context."""
    # Extract all dataclass fields from obj
    new_configs = extract_all_configs(obj)

    # Get current context
    current = _config_context_base.get()

    # Merge with current context
    merged = merge_configs(current, new_configs) if current else new_configs

    # Track type in stack for hierarchy queries
    current_types = context_type_stack.get()
    new_types = current_types + (type(obj),) if obj is not None else current_types

    # Set new context
    token = _config_context_base.set(merged)
    type_token = context_type_stack.set(new_types)
    try:
        yield
    finally:
        _config_context_base.reset(token)
        context_type_stack.reset(type_token)

Each with config_context() block adds configs to the stack. On exit, the context is automatically restored.

Context Type Stack

The context_type_stack ContextVar tracks which types have been pushed via config_context(). This enables generic hierarchy queries without hardcoding type names:

from openhcs.config_framework import get_context_type_stack

with config_context(global_config):
    with config_context(pipeline_config):
        with config_context(step):
            # Query the active type stack
            stack = get_context_type_stack()
            # Returns: (GlobalPipelineConfig, PipelineConfig, StepType)

Hierarchy Registry

For cross-window updates (when parent config windows are open but not actively resolving), the system maintains a persistent hierarchy registry:

# Registry maps child_type → parent_type
_known_hierarchy: dict = {}

def register_hierarchy_relationship(parent_type, child_type):
    """Register that parent_type is the parent of child_type in the config hierarchy."""
    parent_base = _normalize_type(parent_type)
    child_base = _normalize_type(child_type)
    if parent_base != child_base and not _is_global_type(parent_base):
        _known_hierarchy[child_base] = parent_base

def unregister_hierarchy_relationship(child_type):
    """Remove hierarchy entry when form closes."""
    child_base = _normalize_type(child_type)
    _known_hierarchy.pop(child_base, None)

Form managers register their hierarchy when they open:

# Step editor opens with context_obj=pipeline_config
register_hierarchy_relationship(type(context_obj), type(object_instance))
# Registers: PipelineConfig → Step

This allows get_types_before_in_stack() to return correct ancestors even when the parent window isn’t actively inside a config_context() call.

Hierarchy Query Functions

The config framework provides generic functions to query the hierarchy:

from openhcs.config_framework import (
    get_types_before_in_stack,
    is_ancestor_in_context,
    is_same_type_in_context,
    get_ancestors_from_hierarchy
)

# Get all types that come before Step in the hierarchy
ancestors = get_types_before_in_stack(Step)
# Returns: [PipelineConfig] (uses active stack or registry fallback)

# Check if PipelineConfig is an ancestor of Step
is_ancestor = is_ancestor_in_context(PipelineConfig, Step)
# Returns: True

# Check if two types are equivalent (handles lazy wrappers)
is_same = is_same_type_in_context(LazyPipelineConfig, PipelineConfig)
# Returns: True

These functions first check the active context_type_stack, then fall back to the _known_hierarchy registry for cross-window scenarios.

Context Extraction

The system extracts all dataclass fields from the provided object:

def extract_all_configs(context_obj) -> Dict[str, Any]:
    """Extract all dataclass configs from an object."""
    configs = {}

    # Extract dataclass fields
    if is_dataclass(context_obj):
        for field in fields(context_obj):
            value = getattr(context_obj, field.name)
            if is_dataclass(value):
                # Store by type name
                configs[type(value).__name__] = value

    return configs

This flattens nested configs into a single dictionary keyed by type name.

Global Config Context

Global config uses thread-local storage for stability:

# Thread-local storage for global config
_global_config_storage: Dict[Type, threading.local] = {}

def ensure_global_config_context(config_type: Type, config_instance: Any):
    """Establish global config as base context."""
    # Store in thread-local
    set_current_global_config(config_type, config_instance)

    # Also inject into contextvars base
    with config_context(config_instance):
        # Global context now available

This hybrid approach uses thread-local for the global base and contextvars for dynamic stacking.

Resolution Integration

The dual-axis resolver receives the merged context:

def resolve_field_inheritance(obj, field_name, available_configs):
    """Resolve field through dual-axis algorithm.

    available_configs: Merged context from config_context() stack
    """
    # Walk MRO
    for mro_class in type(obj).__mro__:
        # Check if this MRO class has a config instance in context
        for config_name, config_instance in available_configs.items():
            if type(config_instance) == mro_class:
                value = object.__getattribute__(config_instance, field_name)
                if value is not None:
                    return value
    return None

The available_configs dict contains all configs from the context stack, flattened and ready for MRO traversal.

Usage Pattern

From tests/integration/test_main.py:

# Establish global context
global_config = GlobalPipelineConfig(num_workers=4)
ensure_global_config_context(GlobalPipelineConfig, global_config)

# Create pipeline config
pipeline_config = PipelineConfig(
    path_planning_config=LazyPathPlanningConfig(output_dir_suffix="_custom")
)

# Stack contexts
with config_context(pipeline_config):
    # Both global and pipeline configs available
    # Lazy fields resolve through merged context
    orchestrator = Orchestrator(pipeline_config)

The orchestrator and all lazy configs inside it can resolve fields through both global_config and pipeline_config contexts.

Context Cleanup

Contextvars automatically handle cleanup:

with config_context(pipeline_config):
    # Context active
    pass
# Context automatically restored to previous state

No manual cleanup needed - Python’s context manager protocol handles it.

Framework-Agnostic Context Stack Building

For UI placeholder resolution, the build_context_stack() function provides a framework-agnostic way to build complete context stacks:

from openhcs.config_framework import build_context_stack

# Build context stack for placeholder resolution
stack = build_context_stack(
    context_obj=pipeline_config,           # Parent context
    overlay=manager.parameters,            # Current form values
    dataclass_type=manager.dataclass_type, # Type being edited
    live_context=live_context_dict,        # Live values from other forms
    root_form_values=root_values,          # Root form's values (for sibling inheritance)
    root_form_type=root_type,              # Root form's dataclass type
)

with stack:
    # Context layers are active
    placeholder = resolve_placeholder(field_name)

The stack builds layers in order:

Global context layer - Thread-local global config or live editor values
Intermediate layers - Ancestors from get_types_before_in_stack()
Parent context - The context_obj parameter
Root form layer - For sibling inheritance (see below)
Overlay - Current form values

Sibling Inheritance via Root Form

When nested configs need to inherit from siblings (e.g., well_filter_config inheriting from step_well_filter_config), the root form’s values enable this:

# Root form (Step) contains both sibling configs
root_values = {
    'step_well_filter_config': LazyStepWellFilterConfig(well_filter=123),
    'well_filter_config': LazyWellFilterConfig(well_filter=None),
    ...
}

# When resolving well_filter_config.well_filter:
# 1. stack includes root_values
# 2. LazyWellFilterConfig.well_filter resolution walks MRO
# 3. Finds StepWellFilterConfig (superclass) in context
# 4. Uses step_well_filter_config.well_filter = 123

For non-dataclass roots (e.g., FunctionStep), the function wraps values in SimpleNamespace to maintain a unified code path:

if root_form_type and is_dataclass(root_form_type):
    root_instance = root_form_type(**root_form_values)
else:
    # Non-dataclass root - wrap in SimpleNamespace
    from types import SimpleNamespace
    root_instance = SimpleNamespace(**root_form_values)

stack.enter_context(config_context(root_instance))

This enables sibling inheritance for any root type, including function step parameters.

See Field Change Dispatcher Architecture for how the UI uses this for live placeholder updates.