API Reference

This section contains the complete API documentation for objectstate.

Core Modules

objectstate.config	Framework configuration for pluggable base config type.
objectstate.lazy_factory	Generic lazy dataclass factory using flexible resolution.
objectstate.context_manager	Generic contextvars-based context management system for lazy configuration.
objectstate.dual_axis_resolver	Generic dual-axis resolver for lazy configuration inheritance.
objectstate.global_config	Generic global configuration context management.
objectstate.placeholder	Generic lazy placeholder service for UI integration.
objectstate.live_context_resolver	Live context resolution service for configuration framework.
objectstate.token_cache	Token-based cache invalidation service.
objectstate.object_state	ObjectState: Extracted MODEL from ParameterFormManager.
objectstate.snapshot_model	Snapshot and Timeline dataclasses for git-like time-travel history.
objectstate.parametric_axes	Parametric Axes Prototype - Arbitrary semantic axes for type construction.
objectstate.reified_generics	Reified Generics Prototype - Runtime-preserved type parameters.

Module Documentation

objectstate.config

Framework configuration for pluggable base config type.

This module provides the configuration interface for the lazy configuration framework, allowing applications to specify their base configuration type.

The framework uses pure MRO-based resolution. The dual-axis resolution works by: 1. X-axis: Context flattening (Step → Pipeline → Global contexts merged) 2. Y-axis: MRO traversal (most specific → least specific class in inheritance chain)

You only need to call set_base_config_type() once at application startup.

objectstate.config.set_base_config_type(config_type: Type) → None

Set the base configuration type for the framework.

This type is used as the root of the configuration hierarchy and should be the top-level configuration dataclass for your application.

Parameters:

config_type – The base configuration dataclass type

Example

>>> from myapp.config import GlobalConfig
>>> from objectstate.config import set_base_config_type
>>> set_base_config_type(GlobalConfig)

objectstate.config.get_base_config_type() → Type

Get the base configuration type.

Returns:

The base configuration type

Raises:

RuntimeError – If base config type has not been set

objectstate.lazy_factory

Generic lazy dataclass factory using flexible resolution.

objectstate.lazy_factory.get_inherited_field_names(cls: Type) → set

Get names of fields inherited from parent dataclasses (not defined in cls itself).

A field is “inherited” if it exists in a parent’s __dataclass_fields__ but is NOT in this class’s own __annotations__ (i.e., not redefined here).

objectstate.lazy_factory.rebuild_with_none_defaults(cls: Type, field_names_to_none: set | None = None, new_name: str | None = None) → Type

Rebuild a dataclass via make_dataclass with None defaults for specified fields.

This is the UNIFIED approach for both base classes (inherit_as_none) and lazy classes. Instead of patching Field objects after @dataclass, we rebuild with correct defaults.

Parameters:

• cls – The dataclass to rebuild

• field_names_to_none – Fields that should have default=None. If None, ALL fields get default=None (for lazy classes).

• new_name – Optional new class name (for lazy classes)

Returns:

A new class with the same fields but modified defaults

objectstate.lazy_factory.replace_raw(instance, **changes)

Replace dataclass fields while preserving raw None values.

Unlike dataclasses.replace(), this function uses object.__getattribute__ to get field values, preventing lazy resolution from being triggered. This is critical for lazy dataclasses where None means “inherit from parent” and must not be resolved during copy operations.

Parameters:

• instance – The dataclass instance to copy

• **changes – Field values to override

Returns:

A new instance with raw values preserved (not resolved)

objectstate.lazy_factory.register_lazy_type_mapping(lazy_type: Type, base_type: Type) → None

Register mapping between lazy dataclass type and its base type.

objectstate.lazy_factory.get_base_type_for_lazy(lazy_type: Type) → Type | None

Get the base type for a lazy dataclass type.

objectstate.lazy_factory.register_lazy_type(cls: Type) → Type

Register a lazy dataclass type for constructor patching.

This decorator/function marks a lazy dataclass as needing constructor patching when used in code execution contexts (e.g., exec()).

Parameters:

cls – The lazy dataclass type to register

Returns:

The class unchanged (allows use as decorator)

Example

register_lazy_type(LazyPipelineConfig) register_lazy_type(LazyZarrConfig)

# Or as decorator: @register_lazy_type class LazyCustomConfig:

…

objectstate.lazy_factory.get_registered_lazy_types() → frozenset

Get all registered lazy types for constructor patching.

Returns:

Immutable set of registered lazy dataclass types

objectstate.lazy_factory.is_lazy_dataclass(obj_or_type) → bool

Check if an object or type is a lazy dataclass.

ANTI-DUCK-TYPING: Uses isinstance() check against LazyDataclass base class instead of hasattr() attribute sniffing.

Works with both instances and types, and naturally handles Optional types without unwrapping.

Parameters:

obj_or_type – Either a dataclass instance or a dataclass type

Returns:

True if the object/type is a lazy dataclass

Examples

>>> is_lazy_dataclass(PipelineConfig)  # True (type check)
>>> is_lazy_dataclass(GlobalPipelineConfig)  # False
>>> is_lazy_dataclass(pipeline_config_instance)  # True (instance check)
>>> is_lazy_dataclass(LazyPathPlanningConfig)  # True
>>> is_lazy_dataclass(PathPlanningConfig)  # False

# Works with Optional without unwrapping! >>> config: Optional[PipelineConfig] = PipelineConfig() >>> is_lazy_dataclass(config) # True - checks the instance, not the type annotation

class objectstate.lazy_factory.GlobalConfigMeta

Bases: type

Metaclass that makes isinstance(obj, GlobalConfigBase) work by checking _is_global_config marker.

This enables type-safe isinstance checks without inheritance:

if isinstance(config, GlobalConfigBase): # Returns True for GlobalPipelineConfig

# Returns False for PipelineConfig (lazy version)

class objectstate.lazy_factory.GlobalConfigBase

Bases: object

Virtual base class for all global config types.

Uses custom metaclass to check _is_global_config marker instead of actual inheritance. This prevents lazy versions (PipelineConfig) from being considered global configs.

Usage:

if isinstance(config, GlobalConfigBase): # Generic, works for any global config

Instead of:

if isinstance(config, GlobalPipelineConfig): # Hardcoded, breaks extensibility

class objectstate.lazy_factory.LazyDataclass

Bases: object

Base class for all lazy dataclasses created by LazyDataclassFactory.

This enables isinstance() checks without duck typing or unwrapping:

isinstance(config, LazyDataclass) # Works! isinstance(optional_config, LazyDataclass) # Works even for Optional!

All lazy dataclasses inherit from this, regardless of naming convention: - PipelineConfig (lazy version of GlobalPipelineConfig) - LazyPathPlanningConfig - LazyWellFilterConfig - etc.

ANTI-DUCK-TYPING: Use isinstance(obj, LazyDataclass) instead of hasattr() checks.

objectstate.lazy_factory.is_global_config_type(config_type: Type) → bool

Check if a config type is a global config (marked by @auto_create_decorator).

GENERIC SCOPE RULE: Use this instead of hardcoding class name checks like:

if config_class == GlobalPipelineConfig:

Instead use:

if is_global_config_type(config_class):

Parameters:

config_type – The config class to check

Returns:

True if the type is marked as a global config, False otherwise

objectstate.lazy_factory.is_global_config_instance(config_instance: Any) → bool

Check if a config instance is an instance of a global config class.

GENERIC SCOPE RULE: Use this instead of hardcoding isinstance checks like:

if isinstance(config, GlobalPipelineConfig):

Instead use:

if is_global_config_instance(config):

Or use the virtual base class:

if isinstance(config, GlobalConfigBase):

Parameters:

config_instance – The config instance to check

Returns:

True if the instance is of a global config type, False otherwise

objectstate.lazy_factory.get_lazy_type_for_base(base_type: Type) → Type | None

Get the lazy type for a base dataclass type.

objectstate.lazy_factory.bind_lazy_resolution_to_class(cls: Type) → None

Add lazy __getattribute__ to an existing class.

This enables concrete classes (like WellFilterConfig stored in GlobalPipelineConfig) to resolve None values via MRO without changing their static defaults.

Parameters:

cls – The class to add lazy resolution to

class objectstate.lazy_factory.LazyMethodBindings

Bases: object

Declarative method bindings for lazy dataclasses.

static create_resolver() → Callable[[Any, str], Any]

Create field resolver method using new pure function interface.

static create_getattribute() → Callable[[Any, str], Any]

Create lazy __getattribute__ method using new context system.

static create_to_base_config(base_class: Type) → Callable[[Any], Any]

Create base config converter method.

static create_class_methods() → Dict[str, Any]

Create class-level utility methods.

__init__() → None

class objectstate.lazy_factory.LazyDataclassFactory

Bases: object

Generic factory for creating lazy dataclasses with flexible resolution.

static make_lazy_simple(base_class: Type, lazy_class_name: str = None) → Type

Create lazy dataclass using new contextvars system.

SIMPLIFIED: No complex hierarchy providers or field path detection needed. Uses new contextvars system for all resolution.

Parameters:

• base_class – Base dataclass to make lazy

• lazy_class_name – Optional name for the lazy class

Returns:

Generated lazy dataclass with contextvars-based resolution

objectstate.lazy_factory.patch_lazy_constructors(types: List[Type] | None = None)

Context manager that patches lazy dataclass constructors to preserve None vs concrete distinction.

This is critical for code editors that use exec() to create dataclass instances. Without patching, lazy dataclasses would resolve None values to concrete defaults during construction, making it impossible to distinguish between explicitly set values and inherited values.

The patched constructor sets fields provided in kwargs and otherwise uses the dataclass defaults/default_factory (or None if none exist). This preserves the None vs concrete distinction while still instantiating nested lazy configs.

Parameters:

types – Optional list of lazy types to patch. If None, uses all registered lazy types.

Usage:

# Register types at module level register_lazy_type(LazyPipelineConfig) register_lazy_type(LazyZarrConfig)

# Patch during code execution with patch_lazy_constructors():

exec(code_string, namespace) # Lazy dataclasses created during exec() will preserve None values

Example

# Without patching: LazyZarrConfig(compression=’gzip’) # All unspecified fields resolve to defaults

# With patching: with patch_lazy_constructors():

LazyZarrConfig(compression=’gzip’) # Only compression is set, rest are None

objectstate.lazy_factory.ensure_global_config_context(global_config_type: Type, global_config_instance: Any) → None

Ensure proper thread-local storage setup for any global config type.

objectstate.lazy_factory.resolve_lazy_configurations_for_serialization(data: Any) → Any

Recursively resolve lazy dataclass instances to concrete values for serialization.

CRITICAL: This function must be called WITHIN a config_context() block! The context provides the hierarchy for lazy resolution.

How it works: 1. For lazy dataclasses: Access fields with getattr() to trigger resolution 2. The lazy __getattribute__ uses the active config_context() to resolve None values 3. Convert resolved values to base config for pickling

Example (from README.md):

with config_context(orchestrator.pipeline_config):

# Lazy resolution happens here via context resolved_steps = resolve_lazy_configurations_for_serialization(steps)

objectstate.lazy_factory.create_dataclass_for_editing(dataclass_type: Type[T], source_config: Any, preserve_values: bool = False, context_provider: Callable[[Any], None] | None = None) → T

Create dataclass for editing with configurable value preservation.

objectstate.lazy_factory.rebuild_lazy_config_with_new_global_reference(existing_lazy_config: Any, new_global_config: Any, global_config_type: Type | None = None) → Any

Rebuild lazy config to reference new global config while preserving field states.

This function preserves the exact field state of the existing lazy config: - Fields that are None (using lazy resolution) remain None - Fields that have been explicitly set retain their concrete values - Nested dataclass fields are recursively rebuilt to reference new global config - The underlying global config reference is updated for None field resolution

Parameters:

• existing_lazy_config – Current lazy config instance

• new_global_config – New global config to reference for lazy resolution

• global_config_type – Type of the global config (defaults to type of new_global_config)

Returns:

New lazy config instance with preserved field states and updated global reference

class objectstate.lazy_factory.abbreviation(name: str)

Bases: object

Decorator/marker for abbreviations in config classes.

Can be used as: 1. Class decorator: @abbreviation(‘wfc’) to set class abbreviation 2. In Annotated types: Annotated[Type, abbreviation(‘wf’)] for field abbreviation

Examples

@abbreviation(‘wfc’) @global_pipeline_config @dataclass class WellFilterConfig:

well_filter: Annotated[Optional[int], abbreviation(‘wf’)] = None well_filter_mode: Annotated[WellFilterMode, abbreviation(‘wfm’)] = WellFilterMode.INCLUDE

Parameters:

name – The abbreviation string

Returns:

the class with _abbreviation attribute set When used in Annotated: a marker object storing the abbreviation

Return type:

When used as class decorator

__init__(name: str)

__call__(cls: Type) → Type

Class decorator usage: @abbreviation(‘wfc’)

objectstate.lazy_factory.get_group_abbreviation(config_type: str | type) → str

Look up group abbreviation from GROUP_ABBREVIATIONS_REGISTRY.

Handles both type objects and type name strings.

Parameters:

config_type – Config class or type name to get abbreviation for

Returns:

Abbreviation string if found, otherwise falls back to class name prefix

objectstate.lazy_factory.create_global_default_decorator(target_config_class: Type)

Create a decorator factory for a specific global config class.

The decorator accumulates all decorations, then injects all fields at once when the module finishes loading. Also creates lazy versions of all decorated configs.

objectstate.lazy_factory.auto_create_decorator(global_config_class)

Decorator that automatically creates: 1. A field injection decorator for other configs to use 2. A lazy version of the global config itself

Global config classes must start with “Global” prefix.

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

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

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()

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], ...]

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

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

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)

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)

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)

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()

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)

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)

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)

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)

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]

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]

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]

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])

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)

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()

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)

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()

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]

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]

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]

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

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)

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()

objectstate.dual_axis_resolver

Generic dual-axis resolver for lazy configuration inheritance.

This module provides the core inheritance resolution logic as a pure function, supporting both context hierarchy (X-axis) and sibling inheritance (Y-axis).

The resolver is completely generic and has no application-specific dependencies.

objectstate.dual_axis_resolver.clear_mro_resolution_cache() → None

Clear the MRO resolution cache. Call when context fundamentally changes.

objectstate.dual_axis_resolver.invalidate_mro_cache_for_field(changed_type: type, field_name: str) → None

Invalidate cache entries for a specific field that could be affected by a type change.

PERFORMANCE: Only clears cache entries where: 1. The field_name matches 2. The obj_type has changed_type in its MRO (could inherit from it)

This is much more targeted than clear_mro_resolution_cache().

objectstate.dual_axis_resolver.resolve_field_inheritance(obj, field_name: str, available_configs: Dict[str, Any]) → Any

MRO-based inheritance resolution.

ALGORITHM: For LazyDataclass types:

1. Check if same-type config exists in context with concrete value

2. Walk MRO to find parent class configs with concrete value

3. Fall back to static class defaults if nothing found

For concrete classes with lazy resolution:

1. SKIP same-type lookup (if you created ProcessingConfig(group_by=None), you want None)

2. Walk MRO to find PARENT class configs with concrete value (sibling inheritance)

3. Return None if nothing found (no static default fallback)

Parameters:

• obj – The object requesting field resolution

• field_name – Name of the field to resolve

• available_configs – Dict mapping config type names to config instances

Returns:

Resolved field value or None if not found

objectstate.dual_axis_resolver.resolve_with_provenance(container_type: type, field_name: str) → Tuple[Any, str | None, type | None]

Resolve a field value AND find its provenance source in ONE walk.

TWO-PHASE DUAL-AXIS RESOLUTION:

Phase 1 - Hierarchy (same-type only, outer to inner):

Walk scopes from global → pipeline → step, checking ONLY the same-type config. If a concrete value is found, return it immediately. This gives hierarchy precedence for directly-set values.

Phase 2 - MRO fallback (only if Phase 1 found nothing):

Walk scopes from inner to outer (step → pipeline → global), doing MRO walk. This allows sibling inheritance when no concrete value exists in the hierarchy.

This ensures: - GlobalPipelineConfig.well_filter_config.well_filter overrides the same field

at PipelineConfig level (hierarchy precedence for same-type)

• MRO inheritance only applies when NO concrete value exists in the hierarchy for the specific config type being resolved

PERFORMANCE: Single walk instead of separate resolve + provenance calls.

IMPORTANT: Must be called within a config_context() that has scope_ids set up. The layer stack is built by build_context_stack() with ancestor_objects_with_scopes.

Parameters:

• container_type – The type containing the field (e.g., LazyPathPlanningConfig)

• field_name – Name of the field to find provenance for (e.g., “well_filter”)

Returns:

The resolved value, scope that provided it, and the TYPE that has the concrete value (may differ from container_type due to MRO inheritance, e.g., PathPlanningConfig instead of WellFilterConfig). If no concrete value found, returns (None, None, None).

Return type:

(resolved_value, source_scope_id, source_type)

objectstate.dual_axis_resolver.get_field_provenance(container_type: type, field_name: str) → Tuple[str | None, type | None]

Find which scope AND type provided the concrete value for a field.

CONVENIENCE WRAPPER: Calls resolve_with_provenance() and returns scope + type. Use resolve_with_provenance() directly when you also need the value.

Parameters:

• container_type – The type containing the field (e.g., LazyPathPlanningConfig)

• field_name – Name of the field to find provenance for (e.g., “well_filter”)

Returns:

The scope_id and type that provided the value. source_type may differ from container_type due to MRO inheritance. Returns (None, None) if no layer has a concrete value.

Return type:

(source_scope_id, source_type)

objectstate.global_config

Generic global configuration context management.

Provides thread-local storage for global configuration state. This is used as the base context for all lazy configuration resolution.

DUAL THREAD-LOCAL PATTERN: - _saved_global_config_contexts: SAVED config (what descendants/compiler see) - _live_global_config_contexts: LIVE config (what UI sees during editing)

Default behavior: UI uses LIVE (sees unsaved edits) Explicit override: Compilation uses SAVED (via use_live_global=False)

objectstate.global_config.set_saved_global_config(config_type: Type, config_instance: Any) → None

Set SAVED global config (what descendants/compiler see).

Called when: - App startup loads cached config - User SAVES GlobalPipelineConfig in ConfigWindow - Tests set up saved state

Parameters:

• config_type – The config type to set

• config_instance – The SAVED config instance

objectstate.global_config.set_live_global_config(config_type: Type, config_instance: Any) → None

Set LIVE global config (what UI sees during editing).

Called when: - User types in GlobalPipelineConfig field (every keystroke) - UI needs to show live preview of unsaved changes

Parameters:

• config_type – The config type to set

• config_instance – The LIVE (unsaved) config instance

objectstate.global_config.get_saved_global_config(config_type: Type) → Any | None

Get SAVED global config (what descendants/compiler see).

Parameters:

config_type – The config type to retrieve

Returns:

Saved config instance or None

objectstate.global_config.get_live_global_config(config_type: Type) → Any | None

Get LIVE global config (what UI sees during editing).

Parameters:

config_type – The config type to retrieve

Returns:

Live config instance or None

objectstate.global_config.set_current_global_config(config_type: Type, config_instance: Any, *, caller_context: str = None) → None

DEPRECATED: Use set_saved_global_config() or set_live_global_config() explicitly.

For backward compatibility, this sets BOTH saved and live.

objectstate.global_config.set_global_config_for_editing(config_type: Type, config_instance: Any) → None

Set global config for editing (sets BOTH saved and live).

Use this for app startup and initial setup. For live editing updates, use set_live_global_config(). For saving, use set_saved_global_config().

Parameters:

• config_type – The config type to set

• config_instance – The config instance to set

objectstate.global_config.get_current_global_config(config_type: Type, use_live: bool = True) → Any | None

Get current global config.

Parameters:

• config_type – The config type to retrieve

• use_live – If True (default), return LIVE config (UI sees unsaved edits). If False, return SAVED config (compiler sees saved values).

Returns:

Config instance or None

objectstate.placeholder

Generic lazy placeholder service for UI integration.

Provides placeholder text resolution for lazy configuration dataclasses using contextvars-based context management.

class objectstate.placeholder.LazyDefaultPlaceholderService

Bases: object

Simplified placeholder service using new contextvars system.

Provides consistent placeholder pattern for lazy configuration classes using the same resolution mechanism as the compiler.

PLACEHOLDER_PREFIX = 'Default'

NONE_VALUE_TEXT = '(none)'

static has_lazy_resolution(dataclass_type: type) → bool

Check if a type has lazy resolution capability.

Returns True for: 1. LazyDataclass types (all None defaults, used in PipelineConfig) 2. Concrete types with _has_lazy_resolution (used in GlobalPipelineConfig)

The distinction matters: - is_lazy_dataclass() → only LazyDataclass types - has_lazy_resolution() → any type that can resolve None via MRO

static get_lazy_resolved_placeholder(dataclass_type: type, field_name: str, placeholder_prefix: str | None = None, context_obj: Any | None = None) → str | None

Get placeholder text using the new contextvars system.

Parameters:

• dataclass_type – The dataclass type to resolve for

• field_name – Name of the field to resolve

• placeholder_prefix – Optional prefix for placeholder text

• context_obj – Optional context object (orchestrator, step, dataclass instance, etc.) - unused since context should be set externally

Returns:

Formatted placeholder text or None if no resolution possible

objectstate.placeholder.get_lazy_resolved_placeholder(*args, **kwargs)

Backward compatibility wrapper.

objectstate.live_context_resolver

Live context resolution service for configuration framework.

Provides cached resolution of config attributes using live values, avoiding redundant context building and resolution operations.

This service is completely generic and UI-agnostic: - No knowledge of ParameterFormManager or any UI components - No knowledge of steps, orchestrators, or domain concepts - Only knows about dataclasses, context stacks, and attribute resolution - Caller is responsible for providing live context data

class objectstate.live_context_resolver.LiveContextResolver

Bases: object

Pure service for resolving config attributes with live values.

Caches resolved config values to avoid expensive context stack building + resolution. Token-based invalidation ensures cache coherency.

Completely generic - works with any dataclasses and any context hierarchy. UI layer is responsible for: - Collecting live context from form managers - Providing the current token - Building the context stack

__init__()

resolve_config_attr(config_obj: object, attr_name: str, context_stack: list, live_context: Dict[Type, Dict[str, Any]], cache_token: int) → Any

Resolve config attribute through context hierarchy with caching.

Completely generic - no knowledge of UI, steps, orchestrators, or domain concepts.

Parameters:

• config_obj – Config dataclass instance to resolve attribute from

• attr_name – Attribute name to resolve (e.g., ‘enabled’)

• context_stack – List of context objects to resolve through (e.g., [global_config, pipeline_config, step])

• live_context – Live values from form managers, keyed by type

• cache_token – Current cache token for invalidation

Returns:

Resolved attribute value

invalidate() → None

Invalidate all caches.

reconstruct_live_values(live_values: Dict[str, Any]) → Dict[str, Any]

Return live values unchanged.

objectstate.token_cache

Token-based cache invalidation service.

Provides a reusable abstraction for caching values that should be invalidated when a global token changes (e.g., when any form value changes).

class objectstate.token_cache.CacheKey(components: Tuple[Any, ...])

Bases: object

Immutable cache key that can include multiple components.

components: Tuple[Any, ...]

classmethod from_args(*args) → CacheKey

Create cache key from variable arguments.

__init__(components: Tuple[Any, ...]) → None

class objectstate.token_cache.TokenCache(token_provider: Callable[[], int])

Bases: Generic[T]

Generic token-based cache with automatic invalidation.

The cache is invalidated when the token changes. This is useful for caching values that depend on global state (e.g., form values) that can change.

Example

# Create cache with token provider cache = TokenCache(lambda: ParameterFormManager._live_context_token_counter)

# Get or compute value value = cache.get_or_compute(

key=CacheKey.from_args(‘scope’, ‘param_name’), compute_fn=lambda: expensive_computation()

)

# Cache is automatically invalidated when token changes

__init__(token_provider: Callable[[], int])

Initialize token cache.

Parameters:

token_provider – Function that returns the current token value

get_or_compute(key: CacheKey, compute_fn: Callable[[], T]) → T

Get cached value or compute and cache it.

Parameters:

• key – Cache key

• compute_fn – Function to compute value if cache miss

Returns:

Cached or computed value

invalidate()

Manually invalidate the entire cache.

get(key: CacheKey) → T | None

Get cached value without computing.

Parameters:

key – Cache key

Returns:

Cached value or None if not found or token changed

put(key: CacheKey, value: T)

Put value in cache.

Parameters:

• key – Cache key

• value – Value to cache

class objectstate.token_cache.SingleValueTokenCache(token_provider: Callable[[], int])

Bases: Generic[T]

Simplified token cache for single values (no key needed).

Useful when you only need to cache one value that depends on a token.

Example

cache = SingleValueTokenCache(lambda: token_counter) value = cache.get_or_compute(lambda: expensive_computation())

__init__(token_provider: Callable[[], int])

Initialize single-value token cache.

Parameters:

token_provider – Function that returns the current token value

get_or_compute(compute_fn: Callable[[], T]) → T

Get cached value or compute and cache it.

Parameters:

compute_fn – Function to compute value if cache miss

Returns:

Cached or computed value

invalidate()

Manually invalidate the cache.

objectstate.object_state

ObjectState: Extracted MODEL from ParameterFormManager.

This class holds configuration state independently of UI widgets. Lifecycle: Created when object added to pipeline, persists until removed. PFM attaches to ObjectState when editor opens, detaches when closed.

FieldProxy: Type-safe proxy for accessing ObjectState fields via dotted attribute syntax.

class objectstate.object_state.FieldProxy(state: ObjectState, path: str, field_type: type)

Bases: object

Type-safe proxy for accessing ObjectState fields via dotted attribute syntax.

Provides IDE autocomplete while using flat internal storage: - External API: state.fields.well_filter_config.well_filter (type-safe) - Internal: state.parameters[‘well_filter_config.well_filter’] (flat dict)

__init__(state: ObjectState, path: str, field_type: type)

Initialize FieldProxy.

Parameters:

• state – The ObjectState this proxy accesses

• path – Current dotted path (empty for root)

• field_type – Type of the object at this path

__getattr__(name: str) → Any

Get field value or nested FieldProxy.

Parameters:

name – Field name to access

Returns:

FieldProxy for nested dataclass fields, or resolved value for leaf fields

__setattr__(name: str, value: Any) → None

Prevent attribute setting - use state.update_parameter() instead.

class objectstate.object_state.ObjectState(object_instance: Any, scope_id: str | None = None, parent_state: ObjectState | None = None, exclude_params: List[str] | None = None, initial_values: Dict[str, Any] | None = None)

Bases: object

Extracted MODEL from ParameterFormManager - pure Python state without PyQt dependencies.

Lifecycle: - Created when object added to pipeline (before any UI) - Persists until object removed from pipeline - PFM attaches to ObjectState when editor opens, detaches when closed

Core Attributes (8 total): - object_instance: Backing object (updated on Save) - parameters: Mutable working copy (None = unset, value = user-set) - _saved_resolved: Resolved snapshot at save time - _live_resolved: Resolved snapshot using live hierarchy (None = needs compute) - _invalid_fields: Fields needing partial recompute - nested_states: Recursive containment - _parent_state: Parent for context derivation - scope_id: Scope for registry lookup

Everything else is derived: - context_obj → _parent_state.object_instance - dirty_fields → _live_resolved != _saved_resolved - signature_diff_fields → parameters != signature defaults - user_set_fields → {k for k,v in parameters.items() if v is not None}

__init__(object_instance: Any, scope_id: str | None = None, parent_state: ObjectState | None = None, exclude_params: List[str] | None = None, initial_values: Dict[str, Any] | None = None)

Initialize ObjectState with minimal attributes.

Parameters:

• object_instance – The object being edited (dataclass, callable, etc.) If the object declares __objectstate_delegate__, parameters are extracted from that attribute instead (delegation pattern).

• scope_id – Scope identifier for filtering (e.g., “/path::step_0”)

• parent_state – Parent ObjectState for nested forms

• exclude_params – Parameters to exclude from extraction (e.g., [‘func’] for FunctionStep)

• initial_values – Initial values to override extracted defaults (e.g., saved kwargs)

set_extension_metadata(contract: ObjectStateMetadataContract, value: Any) → None

Write extension metadata through a registered time-travel contract.

validate_metadata_contracts(operation: str) → None

Raise if ObjectState metadata contains unregistered extension keys.

copy_metadata_for_snapshot() → Dict[str, Any]

Return validated metadata payload for StateSnapshot.

property context_obj: Any | None

Derive context_obj from parent_state (no separate attribute needed).

property has_delegate: bool

Return whether this state extracts parameters through an object delegate.

classmethod copy_parameter_pair_preserving_callable_identity(parameters: Dict[str, Any], saved_parameters: Dict[str, Any]) → Tuple[Dict[str, Any], Dict[str, Any]]

Copy live/saved raw parameters as one identity-preserving pair.

Time-travel snapshots must preserve callable identity relationships across parameters and _saved_parameters. Copying the two dicts independently can rebuild callable wrappers twice and make a clean state appear dirty.

property saved_object: Any

Get the saved baseline object with the correct type.

For delegation: returns _extraction_target (the delegate/config) For non-delegation: returns object_instance

This is the object that should be used for context resolution when use_saved=True. It represents the “saved” state of the editable object.

property fields: FieldProxy

Type-safe field access via FieldProxy.

Returns:

state.fields.well_filter_config.well_filter → resolved value

Return type:

FieldProxy for accessing fields with IDE autocomplete

property parameter_descriptions: Dict[str, str | None]

Get parameter descriptions for all parameters.

Returns:

Dictionary mapping dotted parameter paths to their descriptions (value may be None). E.g., {‘well_filter_config.well_filter’: ‘Filter wells by…’}

on_resolved_changed(callback: Callable[[Set[str]], None]) → None

Subscribe to resolved value change notifications.

The callback is called when resolved values actually change (not just when cache is invalidated). This enables UI components to flash/highlight specific fields when their resolved values change.

Parameters:

callback – Function that takes a Set[str] of changed dotted paths. E.g., {‘processing_config.group_by’, ‘well_filter_config.well_filter’}

off_resolved_changed(callback: Callable[[Set[str]], None]) → None

Unsubscribe from resolved value change notifications.

on_state_changed(callback: Callable[[], None]) → None

Subscribe to materialized state change notifications (dirty/signature diffs).

off_state_changed(callback: Callable[[], None]) → None

Unsubscribe from materialized state change notifications.

forward_to_parent_state(field_path: str | None = None) → None

Forward child state changes to parent state.

Notifies the parent state that a field has conceptually changed, causing the parent’s on_resolved_changed callbacks to fire (e.g., for UI flash).

Parameters:

field_path – Dotted path of the field that changed (e.g., ‘func’, ‘config.value’). If None, uses _parent_field_name if set, otherwise defaults to scope suffix.

Example

# In child ObjectState callback def on_child_changed(changed_paths):

child_state.forward_to_parent_state(‘func’) # Notify parent its ‘func’ changed

Raises:

RuntimeError – If called on a state without a parent state.

reset_all_parameters() → None

Reset all parameters to defaults.

update_parameter(param_name: str, value: Any) → None

Update parameter value in state.

Enforces invariants: 1. State mutation → scope+type+field aware cache invalidation 2. State mutation → global token increment (for live context cache)

PERFORMANCE: Three-tier filtering for minimal invalidation: - SCOPE: Only descendants of this scope (they inherit from us) - TYPE: Only states with this type in their tree - FIELD: Only the specific field that changed

Parameters:

• param_name – Name of parameter to update

• value – New value

get_resolved_value(param_name: str) → Any

Get resolved value for a field from the bulk snapshot.

Parameters:

param_name – Field name to resolve (can be dotted path like ‘path_planning_config.well_filter’)

Returns:

Resolved value from _live_resolved snapshot. For dataclass container fields, returns a reconstructed dataclass instance with all sub-fields populated from live resolved values.

get_saved_resolved_value(param_name: str) → Any

Get saved resolved value for a field from the saved snapshot.

Unlike get_resolved_value() which returns live values (including unsaved edits), this returns the saved baseline with inheritance applied. This is useful for compilation and other operations that should only consider saved state.

For container fields (dataclasses), this reconstructs the entire nested dataclass with all sub-fields populated from saved resolved values.

Parameters:

param_name – Field name to resolve (can be dotted path like ‘path_planning_config.well_filter’)

Returns:

Saved resolved value from _saved_resolved snapshot. For dataclass fields, returns a reconstructed dataclass instance.

get_provenance(param_name: str) → Tuple[str, type] | None

Get the source scope_id and type for an inherited field value.

For fields where the local value is None (inherited), returns the scope_id of the ancestor that provided the value AND the type that has it. Used for click-to-source navigation in the UI.

The source_type may differ from the local container type due to MRO inheritance. For example, WellFilterConfig.well_filter might inherit from PathPlanningConfig.

NOTE: Returns provenance even when the resolved value is None (signature default). A “concrete None” just means the class default is None and nothing overrode it.

Parameters:

param_name – Field name (can be dotted path like ‘path_planning_config.well_filter’)

Returns:

The scope and type that provided the value, or None if the value is local (not inherited).

Return type:

(source_scope_id, source_type)

find_path_for_type(container_type: type) → str | None

Find the path prefix for a container type in this ObjectState.

With flat storage, nested configs are identified by their path prefix. Given a container type (e.g., PathPlanningConfig), returns the path prefix (e.g., ‘path_planning_config’).

Handles type normalization: LazyPathPlanningConfig matches PathPlanningConfig.

Parameters:

container_type – The type to find the path for

Returns:

Path prefix for the type, or None if not found. Returns “” (empty string) if type is the root object type.

project_ui_visible_field_path(container_type: type, field_name: str) → str | None

Project a container type and field name to the visible UI field path.

Hidden configuration classes can own inherited values even when the form renders a visible subclass. ObjectState owns the path/type index, so this method is the authority for remapping hidden source types to a visible field path.

resolve_for_type(container_type: type, field_name: str) → Any

Resolve a field value given the container type and field name.

Convenience method for callers who have a config object but don’t know its path in the flat storage. Finds the path prefix for the container type and constructs the full dotted path.

Parameters:

• container_type – Type of the containing config (e.g., PathPlanningConfig)

• field_name – Field name within that config (e.g., ‘well_filter’)

Returns:

Resolved value, or None if not found

invalidate_cache() → None

Invalidate resolved cache - forces full recompute on next access.

invalidate_self_and_nested() → None

Invalidate this state’s cache.

With flat storage, no nested states to invalidate.

invalidate_field(field_name: str) → None

Mark a specific field as needing recomputation.

PERFORMANCE: Field-level invalidation - only the changed field needs recomputation, not all 20+ fields in the config.

update_object_instance(new_instance: Any) → None

Replace object_instance with a new instance and re-extract parameters.

This is used when the object being edited is replaced externally (e.g., from code mode execution). The ObjectState is updated to point to the new instance and parameters are re-extracted to match the new object’s state.

For delegation cases, this updates _extraction_target. For non-delegation cases, it updates object_instance directly.

Parameters:

new_instance – The new object instance to extract parameters from

property last_changed_field: str | None

Field that most recently changed value (not just dirty status).

This tracks any value change regardless of saved/unsaved state, useful for time-travel navigation to show what changed in a transition.

reset_parameter(param_name: str) → None

Reset parameter to signature default (None for lazy dataclasses).

Delegates to update_parameter() to ensure consistent invalidation behavior.

get_current_values() → Dict[str, Any]

Get current parameter values from state.

With flat storage, this returns the flat dict with dotted paths. Callers needing nested structure should use to_object() instead.

For ObjectState, this reads directly from self.parameters. PFM overrides this to also read from widgets.

property dirty_fields: Set[str]

Fields where resolved_live != resolved_saved.

property signature_diff_fields: Set[str]

Fields where raw != signature_default.

property is_raw_dirty: bool

Check if raw parameters differ from saved parameters (not resolved values).

mark_saved() → None

Mark current state as saved baseline.

UNIFIED: Works for any object_instance type.

CRITICAL: Invalidates descendant caches for any parameters that changed. This ensures that when saving, other windows that inherited from the old saved values get their caches invalidated so they pick up new values. This mirrors what restore_saved() does but in the opposite direction.

Invalidation is based on comparing the OLD object_instance (about to be replaced) with the NEW self.parameters (live values used for reconstruction).

restore_saved(*, propagate_descendants: bool = True) → None

Restore parameters to the last saved baseline (from object_instance).

UNIFIED: Works for any object_instance type.

CRITICAL: Invalidates descendant caches for any parameters that changed. This ensures that when closing a window without saving, other windows that inherited from the unsaved values get their caches invalidated.

Also emits on_resolved_changed for THIS state so same-level observers (like list items subscribed to this ObjectState) flash when values revert.

should_skip_updates() → bool

Check if updates should be skipped due to batch operations.

to_object(*, update_delegate: bool = False, sync_delegate: bool = True) → Any

Reconstruct object from flat parameters with updated nested configs.

BOUNDARY METHOD - EXPENSIVE - only call at system boundaries: - Save operation - Execute operation - Serialization

UNIFIED: Works for ANY object_instance type. - Python functions: can’t copy, return original - Everything else: shallow copy + reconstruct nested dataclass fields

DELEGATION: If __objectstate_delegate__ was used: - Reconstructs the delegate (e.g., pipeline_config) - If update_delegate=True, updates the delegate attribute on object_instance - Returns the reconstructed delegate (NOT object_instance) - Callers needing the lifecycle object (orchestrator) should use state.object_instance

Parameters:

• update_delegate – If True, apply reconstructed delegate to object_instance

• sync_delegate – If True (default), check for delegate changes before reconstruction. Set to False during cache recomputation to avoid re-entrant invalidation.

Returns:

The reconstructed object that matches the stored parameters. For delegation, this is the delegate type (config), not the lifecycle object.

objectstate.snapshot_model

Snapshot and Timeline dataclasses for git-like time-travel history.

This module provides typed data structures for the ObjectStateRegistry’s time-travel system, replacing the tuple-based implementation.

Design Philosophy: Correct by Construction - No Optional fields for required data - Immutable snapshots (frozen dataclass) - UUID-based identity for snapshots - Direct attribute access (no getattr fallbacks)

class objectstate.snapshot_model.StateSnapshot(saved_resolved: Dict, live_resolved: Dict, parameters: Dict, saved_parameters: Dict, provenance: Dict, meta: Dict = <factory>)

Bases: object

Immutable snapshot of a single ObjectState’s data.

Captures the resolved values, parameters, and provenance at a point in time. No object references - data only for serializability.

saved_resolved: Dict

live_resolved: Dict

parameters: Dict

saved_parameters: Dict

provenance: Dict

meta: Dict

__init__(saved_resolved: Dict, live_resolved: Dict, parameters: Dict, saved_parameters: Dict, provenance: Dict, meta: Dict = <factory>) → None

class objectstate.snapshot_model.Snapshot(id: str, timestamp: float, label: str, triggering_scope: str | None, parent_id: str | None, all_states: Dict[str, StateSnapshot])

Bases: object

Immutable snapshot of ALL ObjectStates at a point in time.

Analogous to a git commit - captures the entire system state.

id: str

timestamp: float

label: str

triggering_scope: str | None

parent_id: str | None

all_states: Dict[str, StateSnapshot]

classmethod create(label: str, all_states: Dict[str, StateSnapshot], triggering_scope: str | None = None, parent_id: str | None = None) → Snapshot

Create a new snapshot with auto-generated ID and timestamp.

to_dict() → Dict

Export to JSON-serializable dict.

classmethod from_dict(data: Dict) → Snapshot

Import from dict (e.g., loaded from JSON).

__init__(id: str, timestamp: float, label: str, triggering_scope: str | None, parent_id: str | None, all_states: Dict[str, StateSnapshot]) → None

class objectstate.snapshot_model.Timeline(name: str, head_id: str, base_id: str, created_at: float = <factory>, description: str = '')

Bases: object

Named branch of history - analogous to a git branch.

Points to a head snapshot and tracks its base (branch point).

name: str

head_id: str

base_id: str

created_at: float

description: str = ''

to_dict() → Dict

Export to JSON-serializable dict.

classmethod from_dict(data: Dict) → Timeline

Import from dict.

__init__(name: str, head_id: str, base_id: str, created_at: float = <factory>, description: str = '') → None

objectstate.parametric_axes

Parametric Axes Prototype - Arbitrary semantic axes for type construction.

This module prototypes extending type() to support arbitrary axes beyond (B, S). Proof-of-concept for PEP proposal: extending type() with axes= parameter.

In the real CPython implementation, this logic would be in type.__new__ itself, and every class would have __axes__ by default (empty MappingProxyType).

This prototype uses AxesMeta as a stand-in for what type() would do natively. Once any class uses metaclass=AxesMeta, all subclasses inherit it automatically.

Usage:

from objectstate.parametric_axes import AxesMeta

# One class in hierarchy uses metaclass=AxesMeta class Step(metaclass=AxesMeta):

pass

# All subclasses automatically get axes support (metaclass inherited) class MyStep(Step, axes={“scope”: “/pipeline/step_0”}):

pass

MyStep.__axes__[“scope”] # “/pipeline/step_0”

# Per-key MRO inheritance works automatically class ChildStep(MyStep, axes={“priority”: 1}):

pass

ChildStep.__axes__[“scope”] # “/pipeline/step_0” (inherited) ChildStep.__axes__[“priority”] # 1 (defined here)

class objectstate.parametric_axes.AxesMeta(name: str, bases: tuple, namespace: dict, axes: Dict[str, Any] | None = None, **kwargs)

Bases: type

Metaclass prototype for type() with axes support.

In the real CPython implementation, this logic would be in type.__new__, and every class would have __axes__ = MappingProxyType({}) by default.

This metaclass is the stand-in: once any class uses metaclass=AxesMeta, all subclasses inherit it automatically (standard metaclass inheritance).

static __new__(mcs, name: str, bases: tuple, namespace: dict, axes: Dict[str, Any] | None = None, **kwargs)

Create type with axes metadata and per-key MRO inheritance.

objectstate.parametric_axes.with_axes(**axes: Any) → Callable[[type], type]

Class decorator that attaches axes metadata.

The decorated class is recreated using AxesMeta so it gains: - __axes__ (MappingProxyType) - convenience attributes (__scope__, __priority__, etc.)

class objectstate.parametric_axes.AxesBase

Bases: object

Opt-in base class enabling class Foo(AxesBase, axes={...}) syntax.

objectstate.parametric_axes.axes_type(name: str, bases: tuple, namespace: dict, **axes) → type

Create a type with arbitrary axes - prototype for extended type().

This mimics the proposed signature:

type(name, bases, namespace, **axes)

Parameters:

• name – Class name

• bases – Base classes tuple

• namespace – Class namespace dict

• **axes – Arbitrary axes (scope=, registry=, version=, etc.)

Returns:

New type with __axes__ containing all axis values

Example

MyStep = axes_type(“MyStep”, (Step,), {“process”: fn},

scope=”/pipeline/step_0”, registry=step_registry)

MyStep.__axes__[“scope”] # “/pipeline/step_0” MyStep.__scope__ # “/pipeline/step_0” (convenience)

objectstate.parametric_axes.get_axes(cls: type) → Dict[str, Any]

Get all axes from a type.

objectstate.parametric_axes.get_axis(cls: type, name: str, default: Any = None) → Any

Get a specific axis value.

objectstate.parametric_axes.has_axis(cls: type, name: str) → bool

Check if type has a specific axis.

objectstate.reified_generics

Reified Generics Prototype - Runtime-preserved type parameters.

This module provides a proof-of-concept implementation of reified generics, where List[int] and List[str] are distinct types at runtime.

Core concepts: - ReifiedMeta: Metaclass with custom __instancecheck__ and __subclasscheck__ - Type caching: Same parameterization returns same type object - Covariance: issubclass(List[int], List[object]) works correctly

Usage:

from objectstate.reified_generics import List, Dict

x = List[int]([1, 2, 3]) isinstance(x, List[int]) # True isinstance(x, List[str]) # False type(x).__args__ # (int,)

class objectstate.reified_generics.ReifiedMeta

Bases: type

Metaclass for reified generic types.

Provides custom isinstance/issubclass behavior that respects type parameters.

__instancecheck__(instance: Any) → bool

Check if instance is of this reified type.

__subclasscheck__(subclass: type) → bool

Check subclass with covariance support.

__hash__() → int

Reified types are hashable for use as dict keys.

__eq__(other: Any) → bool

Equality based on origin and args.

class objectstate.reified_generics.ReifiedList(iterable=(), /)

Bases: list

Reified list that preserves type parameters at runtime.

Usage:

x = List[int]([1, 2, 3]) type(x) # List[int] isinstance(x, List[int]) # True

classmethod __class_getitem__(params)

Create or retrieve cached reified type.

class objectstate.reified_generics.ReifiedDict

Bases: dict

Reified dict that preserves type parameters at runtime.

Usage:

x = Dict[str, int]({“a”: 1}) type(x) # Dict[str, int] isinstance(x, Dict[str, int]) # True

classmethod __class_getitem__(params)

Create or retrieve cached reified type.

class objectstate.reified_generics.ReifiedSet

Bases: set

Reified set that preserves type parameters at runtime.

class objectstate.reified_generics.ReifiedTuple(iterable=(), /)

Bases: tuple

Reified tuple that preserves type parameters at runtime.

objectstate.reified_generics.List

alias of ReifiedList

objectstate.reified_generics.Dict

alias of ReifiedDict

objectstate.reified_generics.Set

alias of ReifiedSet

objectstate.reified_generics.Tuple

alias of ReifiedTuple

objectstate.reified_generics.is_reified(t: type) → bool

Check if a type is a reified generic.

objectstate.reified_generics.get_reified_args(t: type) → tuple

Get type arguments from a reified type.

objectstate.reified_generics.get_reified_origin(t: type) → type

Get origin type from a reified type.

objectstate.reified_generics.clear_cache() → None

Clear the reified type cache (for testing).

objectstate.reified_generics.reified(cls: type) → type

Decorator to make a class support reified generics.

Usage:

@reified class Container:

def __init__(self, items):

self.items = items

x = Container[int]([1, 2, 3]) isinstance(x, Container[int]) # True