objectstate.context_manager

Generic contextvars-based context management system for lazy configuration.

This module provides explicit context scoping using Python’s contextvars to enable hierarchical configuration resolution without explicit parameter passing.

Key features: 1. Explicit context scoping with config_context() manager 2. Config extraction from functions, dataclasses, and objects 3. Config merging for context hierarchy 4. Clean separation between UI windows and contexts

Key components: - current_temp_global: ContextVar holding current merged global config - config_context(): Context manager for creating context scopes - extract_config_overrides(): Extract config values from any object type - merge_configs(): Merge overrides into base config

Functions

`build_context_stack`(object_instance[, ...])	Build a complete context stack for placeholder resolution.
`clear_current_temp_global`()	Clear current context (for testing/debugging).
`clear_extract_all_configs_cache`()	Clear the extract_all_configs cache.
`config_context`(obj[, mask_with_none, ...])	Create new context scope with obj's matching fields merged into base config.
`extract_all_configs`(context_obj)	Extract all config instances from a context object using type-driven approach.
`extract_all_configs_from_context`()	Extract all *_config attributes from current context.
`extract_from_dataclass_fields`(obj)	Get non-None fields as config overrides.
`extract_from_function_signature`(func)	Get parameter defaults as config overrides.
`extract_from_object_attributes`(obj)	Extract config attributes from step/pipeline objects.
`get_ancestors_from_hierarchy`(target_type)	Get all ancestor types for target_type by walking up the known hierarchy.
`get_base_global_config`([use_live])	Get the base global config (fallback when no context set).
`get_context_info`()	Get information about current context for debugging.
`get_context_layer_stack`()	Get the current layer stack for provenance tracking.
`get_context_type_stack`()	Get the current stack of context types (outermost to innermost).
`get_current_temp_global`()	Get current context or None.
`get_normalized_stack`()	Get the context type stack with normalized (base) types, excluding global configs.
`get_root_from_scope_key`(scope_key)	Extract root (plate path) from scope_key for visibility checks.
`get_types_before_in_stack`(target_type)	Get all non-global types that come before target_type in the hierarchy.
`is_ancestor_in_context`(ancestor_type, ...)	Check if ancestor_type comes before descendant_type in the context hierarchy.
`is_same_type_in_context`(type_a, type_b)	Check if two types are the same when normalized.
`is_scope_affected`(target_scope_id, ...)	Check if target scope is affected by edit at editing scope.
`merge_configs`(base, overrides)	Merge overrides into base config, creating new immutable instance.
`register_hierarchy_relationship`(...)	Register that context_obj_type is the parent of object_instance_type in the hierarchy.
`set_current_temp_global`(config)	Set current context (for testing/debugging).
`spawn_thread_with_context`(target[, daemon, name])	Spawn a thread with current context propagated.
`unregister_hierarchy_relationship`(...)	Remove a type from the hierarchy registry when its editor closes.

objectstate.context_manager.config_context(obj, mask_with_none: bool = False, use_live_global: bool = True, scope_id: str | None = None)[source]

Create new context scope with obj’s matching fields merged into base config.

This is the universal context manager for all config context needs. It works by: 1. Finding fields that exist on both obj and the base config type 2. Using matching field values to create a temporary merged config 3. Setting that as the current context

Parameters:

obj – Object with config fields (pipeline_config, step, etc.)
mask_with_none – If True, None values override/mask base config values. If False (default), None values are ignored (normal inheritance). Use True when editing GlobalPipelineConfig to mask thread-local loaded instance with static class defaults.
use_live_global – If True (default), use LIVE global config (UI sees unsaved edits). If False, use SAVED global config (compiler sees saved values).
scope_id – Optional scope identifier for provenance tracking (e.g., “plate_123::step_0”). When provided, the (scope_id, obj) tuple is pushed to context_layer_stack to enable inheritance source lookup via get_field_provenance().

Usage:

# UI operations (default - uses LIVE) with config_context(orchestrator.pipeline_config):

# UI sees unsaved global config edits

# Compilation (explicit - uses SAVED) with config_context(orchestrator.pipeline_config, use_live_global=False):

# Compiler uses saved global config only

# With provenance tracking with config_context(step, scope_id=”plate_123::step_0”):

# Layer tracked for inheritance source lookup

objectstate.context_manager.get_context_type_stack()[source]

Get the current stack of context types (outermost to innermost).

This enables generic hierarchy inference without hardcoding specific types. The stack represents the order in which config_context() calls were nested.

Returns:: Tuple of types representing the context hierarchy. Empty tuple if no context is active.

Example

with config_context(global_config): # stack = (GlobalPipelineConfig,)

with config_context(pipeline_config): # stack = (GlobalPipelineConfig, PipelineConfig)

with config_context(step): # stack = (GlobalPipelineConfig, PipelineConfig, Step): types = get_context_type_stack() # types == (GlobalPipelineConfig, PipelineConfig, Step)

objectstate.context_manager.get_context_layer_stack() → Tuple[Tuple[str, Any], ...][source]

Get the current layer stack for provenance tracking.

Returns a tuple of (scope_id, obj) tuples, ordered from outermost to innermost. Only includes layers where scope_id was explicitly provided to config_context().

Used by get_field_provenance() to determine which scope provided a resolved value.

Returns:: Tuple of (scope_id, obj) tuples representing the context hierarchy. Empty tuple if no layers with scope_ids are active.

Example

with config_context(plate, scope_id=”plate_123”):

with config_context(step, scope_id=”plate_123::step_0”):: layers = get_context_layer_stack() # layers == ((“plate_123”, plate), (“plate_123::step_0”, step))

objectstate.context_manager.get_root_from_scope_key(scope_key: str) → str[source]

Extract root (plate path) from scope_key for visibility checks.

scope_key format: - Pipeline-level: just plate path (e.g., “/path/to/plate”) - Step-level: plate_path::step_token (e.g., “/path/to/plate::step_a”) - Global: empty string

Returns the portion before “::” (or the whole string if no “::” present).

objectstate.context_manager.is_scope_affected(target_scope_id: str | None, editing_scope_id: str | None) → bool[source]

Check if target scope is affected by edit at editing scope.

Uses scope_id hierarchy - no type introspection needed: - None/”” (global) → affects all - “plate_path” (pipeline) → affects same plate + all its steps - “plate_path::token” (step) → affects only that step

Parameters:

target_scope_id – The scope being checked for affectedness (None = global)
editing_scope_id – The scope where the edit occurred (None = global)

Returns:

True if target should refresh when editing_scope changes

objectstate.context_manager.register_hierarchy_relationship(context_obj_type, object_instance_type)[source]

Register that context_obj_type is the parent of object_instance_type in the hierarchy.

Called by ParameterFormManager when a root manager registers. This builds up the known hierarchy from actual usage patterns.

Parameters:

context_obj_type – The parent/context type (e.g., PipelineConfig for Step editor)
object_instance_type – The child type being edited (e.g., Step)

Note

Types are normalized to base types (e.g., LazyPathPlanningConfig → PathPlanningConfig), but lazy-to-global normalization is prevented to preserve distinct hierarchy levels (PipelineConfig stays as PipelineConfig, not collapsed to GlobalPipelineConfig).

objectstate.context_manager.unregister_hierarchy_relationship(object_instance_type)[source]

Remove a type from the hierarchy registry when its editor closes.

Parameters:: object_instance_type – The type to remove from the registry

objectstate.context_manager.get_ancestors_from_hierarchy(target_type)[source]

Get all ancestor types for target_type by walking up the known hierarchy.

Returns ancestors in order from outermost to innermost (excluding target_type itself).

Parameters:: target_type – The type to find ancestors for
Returns:: List of ancestor types in hierarchy order (grandparent, parent, …)

objectstate.context_manager.get_normalized_stack()[source]

Get the context type stack with normalized (base) types, excluding global configs.

Returns:: List of base types in hierarchy order (outermost to innermost), with global config types filtered out.

objectstate.context_manager.get_types_before_in_stack(target_type)[source]

Get all non-global types that come before target_type in the hierarchy.

First tries the active context_type_stack (for resolution during config_context), then falls back to the known hierarchy registry (for cross-window updates).

Parameters:: target_type – The type to find ancestors for (will be normalized)
Returns:: List of base types that come before target_type in the hierarchy. Empty list if no ancestors found.

objectstate.context_manager.is_ancestor_in_context(ancestor_type, descendant_type)[source]

Check if ancestor_type comes before descendant_type in the context hierarchy.

This determines whether changes to ancestor_type should affect descendant_type.

Parameters:

ancestor_type – The potential ancestor type
descendant_type – The potential descendant type

Returns:

True if ancestor_type is an ancestor of descendant_type, False otherwise.

objectstate.context_manager.is_same_type_in_context(type_a, type_b)[source]

Check if two types are the same when normalized.

Handles lazy vs base type equivalence.

Parameters:

type_a – First type to compare
type_b – Second type to compare

Returns:

True if both types normalize to the same base type.

objectstate.context_manager.build_context_stack(object_instance: object, ancestor_objects: list[object] | None = None, ancestor_objects_with_scopes: list[tuple[str, object]] | None = None, current_scope_id: str | None = None, use_live: bool = True)[source]

Build a complete context stack for placeholder resolution.

SINGLE SOURCE OF TRUTH: Uses ancestor_objects from ObjectStateRegistry as the sole mechanism for parent hierarchy. No separate context_obj parameter.

Layer order (innermost to outermost when entered): 1. Global context layer (from ancestors or thread-local) 2. Ancestor objects (from ObjectStateRegistry.get_ancestor_objects()) 3. Current object_instance

Parameters:

object_instance – The object being edited (type used to infer global editing mode)
ancestor_objects – List of ancestor objects from least→most specific (from ObjectStateRegistry). This is the SINGLE SOURCE OF TRUTH for parent hierarchy. DEPRECATED: Use ancestor_objects_with_scopes for provenance tracking.
ancestor_objects_with_scopes – List of (scope_id, object) tuples from least→most specific. Enables provenance tracking via context_layer_stack.
current_scope_id – Scope ID for the current object_instance. Used for provenance tracking.

Returns:

ExitStack with all context layers entered. Caller must manage the stack lifecycle.

objectstate.context_manager.extract_from_function_signature(func) → Dict[str, Any][source]

Get parameter defaults as config overrides.

This enables functions to provide config context through their parameter defaults. Useful for step functions that want to specify their own config values.

Parameters:: func – Function to extract parameter defaults from
Returns:: Dict of parameter_name -> default_value for parameters with defaults

objectstate.context_manager.extract_from_dataclass_fields(obj) → Dict[str, Any][source]

Get non-None fields as config overrides.

This extracts concrete values from dataclass instances, ignoring None values which represent fields that should inherit from context.

Parameters:: obj – Dataclass instance to extract field values from
Returns:: Dict of field_name -> value for non-None fields

objectstate.context_manager.extract_from_object_attributes(obj) → Dict[str, Any][source]

Extract config attributes from step/pipeline objects.

This handles orchestrators, steps, and other objects that have *_config attributes. It flattens the config hierarchy into a single dict of field overrides.

Parameters:: obj – Object to extract config attributes from
Returns:: Dict of field_name -> value for all non-None config fields

objectstate.context_manager.merge_configs(base, overrides: Dict[str, Any])[source]

Merge overrides into base config, creating new immutable instance.

This creates a new config instance with override values merged in, preserving immutability of the original base config.

Parameters:

base – Base config instance (base config type)
overrides – Dict of field_name -> value to override

Returns:

New config instance with overrides applied

objectstate.context_manager.get_base_global_config(use_live: bool = True)[source]

Get the base global config (fallback when no context set).

This provides the global config that was set up with ensure_global_config_context(), or a default if none was set. Used as the base for merging operations.

Parameters:: use_live – If True (default), return LIVE config (UI sees unsaved edits). If False, return SAVED config (compiler sees saved values).
Returns:: Current global config instance or default instance of base config type

objectstate.context_manager.get_current_temp_global()[source]

Get current context or None.

This is the primary interface for resolution functions to access the current context. Returns None if no context is active.

Returns:: Current merged global config or None

objectstate.context_manager.set_current_temp_global(config)[source]

Set current context (for testing/debugging).

This is primarily for testing purposes. Normal code should use config_context() manager instead.

Parameters:: config – Global config instance to set as current context
Returns:: Token for resetting the context

objectstate.context_manager.clear_current_temp_global()[source]

Clear current context (for testing/debugging).

This removes any active context, causing resolution to fall back to default behavior.

objectstate.context_manager.get_context_info() → Dict[str, Any][source]

Get information about current context for debugging.

Returns:: Dict with context information including type, field count, etc.

objectstate.context_manager.extract_all_configs_from_context() → Dict[str, Any][source]

Extract all *_config attributes from current context.

This is used by the resolution system to get all available configs for cross-dataclass inheritance resolution.

Returns:: Dict of config_name -> config_instance for all *_config attributes

objectstate.context_manager.extract_all_configs(context_obj) → Dict[str, Any][source]

Extract all config instances from a context object using type-driven approach.

PERFORMANCE: Results are cached per context object id. Cache is cleared on context changes via clear_extract_all_configs_cache().

This function leverages dataclass field type annotations to efficiently extract config instances, avoiding string matching and runtime attribute scanning.

Parameters:: context_obj – Object to extract configs from (orchestrator, merged config, etc.)
Returns:: Dict mapping config type names to config instances

objectstate.context_manager.clear_extract_all_configs_cache() → None[source]: Clear the extract_all_configs cache. Call when context changes.

objectstate.context_manager.spawn_thread_with_context(target, daemon: bool = True, name: str = None, *args, **kwargs)[source]

Spawn a thread with current context propagated.

This ensures thread-local context (like GlobalPipelineConfig from config framework) is accessible in spawned thread.

Python’s threading module creates isolated thread-local storage, so contextvars don’t automatically propagate to spawned threads. This utility wraps threading.Thread to propagate the current context using contextvars.copy_context().

Parameters:

target – The function to run in thread
daemon – Whether thread should be a daemon thread (default: True)
name – Optional thread name
*args – Positional arguments to pass to target
**kwargs – Keyword arguments to pass to target

Returns:

The spawned Thread object

Example

>>> spawn_thread_with_context(my_function, arg1, kwarg=value)
>>> # Equivalent to:
>>> ctx = contextvars.copy_context()
>>> threading.Thread(target=ctx.run, args=(my_function, arg1), kwargs={'kwarg': value}).start()