Coverage for src/objectstate/object_state.py: 86%
1167 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 07:22 +0000
1"""
2ObjectState: Extracted MODEL from ParameterFormManager.
4This class holds configuration state independently of UI widgets.
5Lifecycle: Created when object added to pipeline, persists until removed.
6PFM attaches to ObjectState when editor opens, detaches when closed.
8FieldProxy: Type-safe proxy for accessing ObjectState fields via dotted attribute syntax.
9"""
11from __future__ import annotations
13import copy
14import logging
15from collections.abc import Callable, Sequence
16from dataclasses import fields as dataclass_fields
17from dataclasses import is_dataclass
18from types import FunctionType
19from typing import Any, Optional
21from objectstate.field_access import DataclassFieldAccess, DottedFieldPath
22from objectstate.construction_binding import (
23 DelegatedStateBinding,
24 DirectStateBinding,
25 StateConstructionBinding,
26)
27from objectstate.object_state_metadata import (
28 ObjectStateMetadataContract,
29 ObjectStateMetadataContractRegistry,
30 ObjectStateMetadataStore,
31)
32from objectstate.object_state_registry import ObjectStateRegistry
33from objectstate.scope_key import ScopeKey
34from objectstate.parameter_owner import ParameterOwner
35from objectstate.parameter_structure import ParameterStructure
36from objectstate.resolution_sources import ResolutionSources
37from objectstate.lazy_factory import is_global_config_type
38from objectstate.subfield_semantics import (
39 MISSING,
40 ObjectStateSubfieldSemanticIndex,
41 build_subfield_semantic_index,
42 changed_structural_leaf_paths,
43)
44from objectstate.time_travel_profile import TimeTravelProfiler
45from objectstate.value_semantics import semantic_values_equal
47logger = logging.getLogger(__name__)
49class FieldProxy:
50 """Type-safe proxy for accessing ObjectState fields via dotted attribute syntax.
52 Provides IDE autocomplete while using flat internal storage:
53 - External API: state.fields.partition_filter_config.partition_filter (type-safe)
54 - Internal: state.parameters['partition_filter_config.partition_filter'] (flat dict)
55 """
57 def __init__(self, state: 'ObjectState', path: str, field_type: type):
58 """Initialize FieldProxy.
60 Args:
61 state: The ObjectState this proxy accesses
62 path: Current dotted path (empty for root)
63 field_type: Type of the object at this path
64 """
65 object.__setattr__(self, '_state', state)
66 object.__setattr__(self, '_path', path)
67 object.__setattr__(self, '_type', field_type)
69 def __getattr__(self, name: str) -> Any:
70 """Get field value or nested FieldProxy.
72 Args:
73 name: Field name to access
75 Returns:
76 FieldProxy for nested dataclass fields, or resolved value for leaf fields
77 """
78 new_path = f'{self._path}.{name}' if self._path else name
80 # Get field info from the type
81 if not is_dataclass(self._type):
82 type_name = getattr(self._type, '__name__', str(self._type))
83 raise AttributeError(f"{type_name} is not a dataclass")
85 field_info = None
86 for f in dataclass_fields(self._type):
87 if f.name == name:
88 field_info = f
89 break
91 if field_info is None:
92 type_name = getattr(self._type, '__name__', str(self._type))
93 raise AttributeError(f"{type_name} has no field '{name}'")
95 # Check if field is a nested dataclass
96 field_type = field_info.type
98 # Handle Optional[DataclassType]
99 from typing import Union, get_args, get_origin
100 origin = get_origin(field_type)
101 if origin is Union:
102 args = get_args(field_type)
103 if len(args) == 2 and type(None) in args:
104 inner_type = next(arg for arg in args if arg is not type(None))
105 if is_dataclass(inner_type) and isinstance(inner_type, type):
106 return FieldProxy(self._state, new_path, inner_type)
108 # Handle direct dataclass type
109 if isinstance(field_type, type) and is_dataclass(field_type):
110 return FieldProxy(self._state, new_path, field_type)
112 # Leaf field - get resolved value
113 return self._state.get_resolved_value(new_path)
115 def __setattr__(self, name: str, value: Any) -> None:
116 """Prevent attribute setting - use state.update_parameter() instead."""
117 _ = (name, value) # Suppress unused warnings
118 raise AttributeError("FieldProxy is read-only. Use state.update_parameter(path, value) to set values.")
121class ObjectState:
122 """
123 UI-independent state for editing and resolving one backing object.
125 Lifecycle:
126 - Created at the backing object's lifecycle boundary, before any editor.
127 - Persists independently of windows that attach to it.
128 - Unregistered when the backing object leaves its owning scope.
130 State ownership:
131 - ``parameters`` is the flat mutable working copy. Nested dataclass fields
132 use dotted paths and do not create recursive ObjectState instances.
133 - ``_saved_parameters`` and ``_saved_resolved`` are the explicit saved
134 baseline; ``_live_resolved`` and ``_live_provenance`` describe the
135 current resolved view.
136 - ``_parent_state`` links independently scoped states for context and
137 notification. It is not structural ownership of a nested dataclass.
138 - ``scope_id`` is the registry identity.
140 Dirty fields and signature differences are materialized from the raw and
141 resolved snapshots. Structural leaf identities are read-only projections
142 of the owning flat field, not additional writable state.
143 """
145 def __init__(
146 self,
147 object_instance: Any,
148 scope_id: str | None = None,
149 parent_state: Optional['ObjectState'] = None,
150 exclude_params: list[str] | None = None,
151 initial_values: dict[str, Any] | None = None,
152 ):
153 """
154 Initialize ObjectState with minimal attributes.
156 Args:
157 object_instance: The object being edited (dataclass, callable, etc.)
158 If the object declares __objectstate_delegate__, parameters
159 are extracted from that attribute instead (delegation pattern).
160 scope_id: Scope identifier for filtering (e.g., "/path::step_0")
161 parent_state: Parent independently scoped ObjectState used for
162 context and parent notification.
163 exclude_params: Parameters to exclude from extraction.
164 initial_values: Initial values to override extracted defaults (e.g., saved kwargs)
165 """
166 # === Core State (3 attributes) ===
167 self.object_instance = object_instance
168 # Use passed scope_id if provided, otherwise inherit from parent
169 # Editors may pass explicit scope_id values for nested parameter scopes.
170 # Nested dataclass configs may omit scope_id and inherit from parent
171 self.scope_id = scope_id if scope_id is not None else (parent_state.scope_id if parent_state else None)
173 # === Delegation Support ===
174 # Check if object declares a delegate for parameter extraction.
175 # This allows storing a lifecycle object (e.g., orchestrator) while
176 # extracting editable parameters from a nested config (e.g., pipeline_config).
177 delegate_attr = getattr(type(object_instance), '__objectstate_delegate__', None)
178 if delegate_attr:
179 self._extraction_target = getattr(object_instance, delegate_attr)
180 self._delegate_attr = delegate_attr
181 logger.debug(f"ObjectState delegation: extracting from '{delegate_attr}' attribute")
182 else:
183 self._extraction_target = object_instance
184 self._delegate_attr = None
186 # Canonical declarations/topology, supplied by the extraction below.
187 self.parameter_structure: ParameterStructure
188 self._cached_object: Any | None = None # Cached result of to_object()
189 self._cached_object_applied: bool = False # True if cached delegate was applied to object_instance
191 # UI / integration metadata (never participates in dirty detection).
192 # Namespaced extension keys are contract-checked so time-travel cannot
193 # silently snapshot or restore extension state it does not understand.
194 self.metadata: ObjectStateMetadataStore = ObjectStateMetadataStore(self.scope_id)
196 # Time-travel navigation helpers (set by ObjectStateRegistry time travel)
197 # Maps param_name -> (before, after) for the last time-travel transition.
198 self._last_changed_values: dict[str, tuple[Any, Any]] = {}
200 # Maps metadata key -> (before, after) for the last time-travel transition.
201 self._last_changed_meta_keys: set[str] = set()
202 self._last_changed_meta_values: dict[str, tuple[Any, Any]] = {}
204 # Extract parameters using FLAT extraction (dotted paths)
205 # This replaces the old UnifiedParameterAnalyzer + _create_nested_states() approach
206 self.parameters: dict[str, Any] = {}
207 extraction_target = self._extraction_target
209 # Flatten parameter extraction - walk nested dataclasses recursively.
210 # Uses _extraction_target (delegate) instead of object_instance for
211 # delegation support. The structure and its immutable path index are
212 # replaced atomically once for this extraction.
213 self._replace_parameter_structure(
214 extraction_target,
215 exclude_params=exclude_params,
216 initial_values=initial_values,
217 )
219 # === Structure (1 attribute) ===
220 self._parent_state: ObjectState | None = parent_state
221 # NOTE: nested_states DELETED - flat storage eliminates nested ObjectState instances
223 # === Cache (3 attributes) ===
224 self._live_resolved: dict[str, Any] | None = None # None = needs full compute
225 self._invalid_fields: set[str] = set() # Fields needing partial recompute
226 # Maps dotted_path → (source_scope_id, source_type) for inherited fields
227 # source_type may differ from local container_type due to MRO inheritance
228 self._live_provenance: dict[str, tuple[str | None, type | None]] = {}
229 # What each resolved snapshot was computed from, keyed by ``use_saved``. A
230 # snapshot is current only while ``_resolution_sources()`` still returns the
231 # same objects; a snapshot without an entry is resolved again in full.
232 self._resolved_from: dict[bool, ResolutionSources] = {}
234 # === Saved baseline (2 attributes) ===
235 self._saved_resolved: dict[str, Any] = {}
236 self._saved_parameters: dict[str, Any] = {} # Immutable snapshot for diff on restore
238 # === Materialized diffs (3 attributes) ===
239 self._raw_dirty = False
240 self._dirty_fields: set[str] = set()
241 self._signature_diff_fields: set[str] = set()
243 # === Change tracking for navigation (2 attributes) ===
244 # Track which field most recently changed VALUE (not just dirty status)
245 # Used for time-travel navigation to scroll to what changed in a transition
246 self._last_changed_field: str | None = None
247 self._last_changed_concrete_paths: set[str] = set()
248 self._last_changed_paths: set[str] = set()
250 # === Flags (kept for batch operations) ===
251 self._in_reset = False
252 self._block_cross_window_updates = False
253 self._forwarding_to_parent = False
254 self._parent_field_name: str | None = None
256 # === State Change Callbacks ===
257 # Callbacks notified when materialized state changes (dirty/signature diffs)
258 self._on_state_changed_callbacks: list[Callable[[set[str]], None]] = []
260 # === Resolved Change Callbacks ===
261 # Callbacks notified when resolved values actually change (for UI flashing)
262 self._on_resolved_changed_callbacks: list[Callable[[set[str]], None]] = []
264 # === Time-Travel Callbacks ===
265 # Callbacks notified when time-travel restores parameters (for widget refresh)
266 self._on_time_travel_callbacks: list[Callable[[], None]] = []
268 # === Time-Travel ===
269 # Instance-level time-travel removed - use ObjectStateRegistry class-level DAG instead
271 # Initialize baselines (nothing to tell anyone during init)
272 self._live_resolved = self._compute_resolved_snapshot()
274 # CRITICAL: Initialize _saved_parameters BEFORE _compute_resolved_snapshot(use_saved=True)
275 # because that method reads from _saved_parameters to get raw values.
276 self._saved_parameters = self._copy_parameters_for_saved_baseline(self.parameters)
278 # CRITICAL: Compute saved_resolved using SAVED ancestor context, not LIVE.
279 # This ensures saved baseline represents "what would this object's values be
280 # if all ancestors were at their saved state", NOT "what are values right now
281 # with ancestor's unsaved edits baked in".
282 #
283 # If we used copy.deepcopy(_live_resolved), we'd bake in ancestor's unsaved
284 # edits, causing inverted dirty state when ancestor is saved/reset.
285 self._saved_resolved = self._compute_resolved_snapshot(use_saved=True)
287 # DEBUG: Log live vs saved at registration
288 logger.debug(f"🔵 INIT_RESOLVED: scope={self.scope_id!r} obj_type={type(self.object_instance).__name__}")
289 for k in sorted(self._live_resolved.keys()):
290 live_val = self._live_resolved.get(k)
291 saved_val = self._saved_resolved.get(k)
292 if live_val != saved_val:
293 logger.debug(f"🔵 INIT_DIFF: {k!r} live={live_val!r} saved={saved_val!r}")
295 # Materialize initial diff sets (no notification during init)
296 # Should be empty for new objects since saved = live
297 self._dirty_fields = self._compute_dirty_fields()
298 self._signature_diff_fields = self._compute_signature_diff_fields()
300 # NOTE: Don't record "init" snapshot here - each ObjectState would create a separate
301 # snapshot missing other ObjectStates created later. Instead, the first edit will
302 # record the baseline state automatically (see record_snapshot logic).
304 def set_extension_metadata(
305 self,
306 contract: ObjectStateMetadataContract,
307 value: Any,
308 ) -> None:
309 """Write extension metadata through a registered time-travel contract."""
311 ObjectStateMetadataContractRegistry.register(contract)
312 self.metadata[contract.key] = value
314 def validate_metadata_contracts(self, operation: str) -> None:
315 """Raise if ObjectState metadata contains unregistered extension keys."""
317 ObjectStateMetadataContractRegistry.validate_mapping(
318 scope_id=self.scope_id,
319 metadata=self.metadata,
320 operation=operation,
321 )
323 def copy_metadata_for_snapshot(self) -> dict[str, Any]:
324 """Return validated metadata payload for StateSnapshot."""
326 return self.metadata.copy_for_snapshot()
328 @property
329 def context_obj(self) -> Any | None:
330 """Derive context_obj from parent_state (no separate attribute needed)."""
331 return self._parent_state.object_instance if self._parent_state else None
333 def _check_and_sync_delegate(self) -> bool:
334 """Check if delegate attribute has changed and sync extraction target if needed.
336 This implements auto-detection of delegate changes (Option 3 from architectural discussion).
337 When object_instance's delegate attribute is replaced with a new instance (e.g., after
338 code that rebuilds the config), this method detects the change and
339 automatically re-extracts parameters from the new delegate.
341 Returns:
342 True if delegate was detected as changed and re-extraction occurred, False otherwise.
343 """
344 if self._delegate_attr is None:
345 # Not using delegation - nothing to check
346 return False
348 try:
349 current_delegate = object.__getattribute__(self.object_instance, self._delegate_attr)
350 except AttributeError:
351 # Delegate attribute no longer exists - this is unexpected but handle gracefully
352 logger.warning(
353 f"Delegate attribute '{self._delegate_attr}' no longer exists on "
354 f"{type(self.object_instance).__name__}. Keeping current extraction target."
355 )
356 return False
358 # Use identity check (is) not equality (==) to detect if it's a new instance
359 if current_delegate is self._extraction_target:
360 # Delegate hasn't changed - no sync needed
361 return False
363 # Delegate has changed to a new instance - sync extraction target and re-extract
364 logger.debug(
365 f"Auto-detected delegate change for ObjectState(scope={self.scope_id!r}): "
366 f"'{self._delegate_attr}' attribute was replaced with new instance. Re-extracting parameters."
367 )
369 self.update_object_instance(current_delegate)
370 return True
372 @property
373 def has_delegate(self) -> bool:
374 """Return whether this state extracts parameters through an object delegate."""
376 return self._delegate_attr is not None
378 @classmethod
379 def _copy_parameters_for_saved_baseline(
380 cls, parameters: dict[str, Any]
381 ) -> dict[str, Any]:
382 """Snapshot parameter values without cloning callable identities.
384 Callables can be semantic identities in object declarations. A plain
385 deepcopy can rebuild callable wrapper instances through ``__reduce__``,
386 making live and saved function specs unequal immediately after load.
387 """
388 memo: dict[int, Any] = {}
389 cls._seed_callable_identity_memo(parameters, memo, set())
390 return copy.deepcopy(parameters, memo)
392 @classmethod
393 def copy_parameter_pair_preserving_callable_identity(
394 cls,
395 parameters: dict[str, Any],
396 saved_parameters: dict[str, Any],
397 ) -> tuple[dict[str, Any], dict[str, Any]]:
398 """Copy live/saved raw parameters as one identity-preserving pair.
400 Time-travel snapshots must preserve callable identity relationships across
401 `parameters` and `_saved_parameters`. Copying the two dicts independently
402 can rebuild callable wrappers twice and make a clean state appear dirty.
403 """
405 memo: dict[int, Any] = {}
406 seen: set[int] = set()
407 cls._seed_callable_identity_memo(parameters, memo, seen)
408 cls._seed_callable_identity_memo(saved_parameters, memo, seen)
409 return copy.deepcopy(parameters, memo), copy.deepcopy(saved_parameters, memo)
411 @classmethod
412 def _seed_callable_identity_memo(
413 cls,
414 value: Any,
415 memo: dict[int, Any],
416 seen: set[int],
417 ) -> None:
418 value_id = id(value)
419 if value_id in seen:
420 return
421 seen.add(value_id)
423 if callable(value):
424 memo[value_id] = value
425 return
427 if isinstance(value, dict):
428 for key, item in value.items():
429 cls._seed_callable_identity_memo(key, memo, seen)
430 cls._seed_callable_identity_memo(item, memo, seen)
431 return
433 if isinstance(value, (list, tuple, set, frozenset)):
434 for item in value:
435 cls._seed_callable_identity_memo(item, memo, seen)
436 return
438 if is_dataclass(value) and not isinstance(value, type):
439 for field in dataclass_fields(value):
440 cls._seed_callable_identity_memo(
441 object.__getattribute__(value, field.name),
442 memo,
443 seen,
444 )
446 @property
447 def saved_object(self) -> Any:
448 """Get the saved baseline object with the correct type.
450 For delegation: returns _extraction_target (the delegate/config)
451 For non-delegation: returns object_instance
453 This is the object that should be used for context resolution when
454 use_saved=True. It represents the "saved" state of the editable object.
455 """
456 # Auto-detect delegate changes before returning extraction target
457 self._check_and_sync_delegate()
458 return self._extraction_target
460 @property
461 def fields(self) -> FieldProxy:
462 """Type-safe field access via FieldProxy.
464 Returns:
465 FieldProxy for accessing fields with IDE autocomplete:
466 state.fields.partition_filter_config.partition_filter → resolved value
467 """
468 return FieldProxy(self, '', type(self.object_instance))
470 def to_resolved_object(self) -> Any:
471 """Reconstruct this state as an object with live resolved values."""
472 self._check_and_sync_delegate()
473 self._ensure_live_resolved()
474 assert self._live_resolved is not None
475 return self._reconstruct_from_resolved(
476 "", self._live_resolved, self._extraction_target, self.parameter_structure
477 )
479 def to_saved_resolved_object(self) -> Any:
480 """Reconstruct this state as an object with saved resolved values."""
481 self._check_and_sync_delegate()
482 self._ensure_saved_resolved()
483 return self._reconstruct_from_resolved(
484 "", self._saved_resolved, self._extraction_target, self.parameter_structure
485 )
487 @classmethod
488 def resolve_saved_object(
489 cls,
490 object_instance: Any,
491 *,
492 scope_id: str | None = None,
493 ancestor_objects_with_scopes: Sequence[tuple[str, Any]] = (),
494 ) -> tuple[Any, dict[str, tuple[str | None, type | None]]]:
495 """Resolve a submitted saved declaration without creating an editing state.
497 Ancestors are explicit saved declarations ordered least to most specific.
498 The saved global context, dual-axis inheritance, raw field extraction and
499 nominal reconstruction are the same operations used by editing states.
500 Returned provenance describes this same saved epoch, not live UI edits.
501 Mutable resolved values are detached together, retaining shared aliases
502 and callable identities. No state is registered or editor initialized.
503 """
504 delegate_attr = getattr(type(object_instance), "__objectstate_delegate__", None)
505 target = (
506 getattr(object_instance, delegate_attr) if delegate_attr else object_instance
507 )
508 structure = ParameterStructure.for_target(target, None)
509 parameters: dict[str, Any] = {}
510 cls._extract_all_parameters_flat(target, parameters, structure)
511 saved_parameters = cls._copy_parameters_for_saved_baseline(parameters)
512 snapshot, provenance = cls._resolve_parameter_snapshot(
513 saved_parameters,
514 saved_parameters,
515 structure,
516 target,
517 scope_id,
518 ancestor_objects_with_scopes,
519 use_saved=True,
520 )
521 # A compilation capture owns inherited mutable values too. The editor's
522 # saved-object API intentionally keeps its existing ancestor lifetime.
523 snapshot = cls._copy_parameters_for_saved_baseline(snapshot)
524 concrete = cls._reconstruct_from_resolved("", snapshot, target, structure)
525 return concrete, provenance
527 @property
528 def parameter_descriptions(self) -> dict[str, str | None]:
529 """Get parameter descriptions for all parameters.
531 Returns:
532 Dictionary mapping dotted parameter paths to their descriptions (value may be None).
533 E.g., {'partition_filter_config.partition_filter': 'Filter partitions by...'}
534 """
535 return self.parameter_structure.descriptions
537 def type_for_path(self, field_path: str) -> ParameterOwner:
538 """Return the ObjectState-recorded type for a dotted field path.
540 Leaf paths return the dataclass/function owner that declares the
541 field. Nested dataclass paths return the nested dataclass type itself.
542 """
543 if field_path == "":
544 return type(self._extraction_target)
545 try:
546 return self.parameter_structure.owner_paths[field_path]
547 except KeyError as exc:
548 raise KeyError(
549 f"ObjectState path is not registered: {field_path!r}"
550 ) from exc
552 def direct_parameter_paths(
553 self,
554 field_path: str = "",
555 ) -> tuple[DottedFieldPath, ...]:
556 """Return immutable direct-child paths for one flat parameter scope.
558 The path topology is rebuilt when ObjectState extracts a replacement
559 object. Values remain authoritative in ``parameters`` and are therefore
560 never cached in this projection.
561 """
563 return self.parameter_structure.direct_paths.get(DottedFieldPath(field_path), ())
565 def has_parameter_descendants(self, field_path: str) -> bool:
566 """Return whether an authoritative flat parameter owns child paths."""
568 return bool(self.direct_parameter_paths(field_path))
570 # === Resolved Change Subscription ===
572 def on_resolved_changed(self, callback: Callable[[set[str]], None]) -> None:
573 """Subscribe to resolved value change notifications.
575 The callback is called when resolved values actually change (not just when
576 cache is invalidated). This enables UI components to flash/highlight
577 specific fields when their resolved values change.
579 Args:
580 callback: Function that takes a Set[str] of changed dotted paths.
581 E.g., {'processing_config.group_by',
582 'partition_filter_config.partition_filter'}
583 """
584 if callback not in self._on_resolved_changed_callbacks:
585 self._on_resolved_changed_callbacks.append(callback)
587 def off_resolved_changed(self, callback: Callable[[set[str]], None]) -> None:
588 """Unsubscribe from resolved value change notifications."""
589 if callback in self._on_resolved_changed_callbacks:
590 self._on_resolved_changed_callbacks.remove(callback)
592 def on_state_changed(self, callback: Callable[[set[str]], None]) -> None:
593 """Subscribe to materialized state change notifications (dirty/signature diffs)."""
594 if callback not in self._on_state_changed_callbacks:
595 self._on_state_changed_callbacks.append(callback)
597 def off_state_changed(self, callback: Callable[[set[str]], None]) -> None:
598 """Unsubscribe from materialized state change notifications."""
599 if callback in self._on_state_changed_callbacks:
600 self._on_state_changed_callbacks.remove(callback)
602 def _notify_state_changed(self, changed_paths: set[str]) -> None:
603 """Fire state change callbacks (best-effort)."""
604 for callback in list(self._on_state_changed_callbacks):
605 try:
606 callback(changed_paths)
607 except Exception as e:
608 logger.warning(f"Error in state_changed callback: {e}")
610 def _notify_resolved_changed(
611 self,
612 changed_paths: set[str],
613 *,
614 context: str,
615 ) -> None:
616 """Publish resolved-value changes to registry and local subscribers."""
617 if not changed_paths:
618 return
620 with TimeTravelProfiler.phase(
621 "objectstate.notify_resolved_emit",
622 scope=self.scope_id,
623 context=context,
624 changed_paths=len(changed_paths),
625 callbacks=len(self._on_resolved_changed_callbacks),
626 ):
627 ObjectStateRegistry.notify_resolved_changed(self.scope_id, changed_paths)
628 if not self._on_resolved_changed_callbacks:
629 return
631 logger.debug(
632 "🔔 CALLBACK_LEAK_DEBUG: Notifying %d callbacks for scope=%s, "
633 "context=%s, changed_paths=%s",
634 len(self._on_resolved_changed_callbacks),
635 self.scope_id,
636 context,
637 changed_paths,
638 )
639 for i, callback in enumerate(list(self._on_resolved_changed_callbacks)):
640 try:
641 with TimeTravelProfiler.phase(
642 "objectstate.notify_resolved_callback",
643 scope=self.scope_id,
644 context=context,
645 callback_index=i,
646 callback_count=len(self._on_resolved_changed_callbacks),
647 changed_paths=len(changed_paths),
648 ):
649 callback(changed_paths)
650 except RuntimeError as e:
651 logger.warning(
652 "🔴 CALLBACK_LEAK_DEBUG: Dead callback #%d detected! "
653 "scope=%s, context=%s, error: %s",
654 i,
655 self.scope_id,
656 context,
657 e,
658 )
659 except Exception as e:
660 logger.warning(
661 "Error in resolved_changed callback #%d during %s: %s",
662 i,
663 context,
664 e,
665 )
667 def forward_to_parent_state(self, field_path: str | None = None) -> None:
668 """Forward child state changes to parent state.
670 Notifies the parent state that a field has conceptually changed, causing
671 the parent's on_resolved_changed callbacks to fire (e.g., for UI flash).
673 Args:
674 field_path: Dotted path of the field that changed (e.g., 'config.value').
675 If None, uses _parent_field_name if set, otherwise defaults to scope suffix.
677 Example:
678 # In child ObjectState callback
679 def on_child_changed(changed_paths):
680 child_state.forward_to_parent_state('config.value')
682 Raises:
683 RuntimeError: If called on a state without a parent state.
684 """
685 if self._parent_state is None:
686 raise RuntimeError(
687 f"Cannot forward to parent: state '{self.scope_id}' has no parent state"
688 )
690 # Reentrancy guard (pattern from ObjectState architectural fixes)
691 if self._forwarding_to_parent:
692 logger.debug(f"[ObjectState] Reentrancy guard blocked for {self.scope_id}")
693 return
695 self._forwarding_to_parent = True
696 try:
697 # Determine which parent field changed
698 # Priority: explicit field_path arg > _parent_field_name > auto-detect
699 parent_field = field_path or self._parent_field_name
700 if parent_field is None:
701 # Auto-detect from a numeric scope suffix.
702 scope = ScopeKey(self.scope_id)
703 if scope.depth >= 2:
704 last = scope.last
705 import re
706 m = re.match(r'^(.+?)_\d+$', last)
707 parent_field = m.group(1) if m else last
709 if parent_field:
710 self._parent_state._notify_resolved_changed(
711 {parent_field},
712 context="forward_to_parent_state",
713 )
715 logger.debug(
716 "[ObjectState] Forwarded %s to parent, field=%r",
717 self.scope_id,
718 parent_field,
719 )
721 finally:
722 self._forwarding_to_parent = False
724 def _ensure_live_resolved(self, notify_flash: bool = True) -> set[str]:
725 """Bring both resolved snapshots up to date; return the changed live paths.
727 Two things make a field out of date: it was invalidated on this state
728 (``_invalid_fields``), or the snapshot was computed from other sources than
729 the state has now (``_fields_with_other_sources``): an ancestor registered,
730 unregistered or replaced, an ancestor's value edited, saved or restored, the
731 global config replaced. Both are found here, on every resolved read, so no
732 caller has to announce a change and a state that is not registered follows
733 its ancestors like a registered one. Only the fields that can be affected are
734 resolved again; when nothing changed the cost is the comparison of the
735 sources.
737 Args:
738 notify_flash: True (reads): when anything was resolved again, dirty and
739 signature fields are updated and subscribers are told here. False:
740 the caller does that with the returned paths
741 (``_sync_materialized_state``).
742 """
743 if self._live_resolved is None:
744 # First access - populate cache; there is no earlier value to compare with.
745 self._live_resolved = self._compute_resolved_snapshot()
746 self._invalid_fields.clear()
747 return set()
749 sources, saved_sources = self._resolution_sources()
750 saved_resolved_again = self._resolve_saved_again_if_sources_changed(saved_sources)
751 if saved_resolved_again:
752 sources, _ = self._resolution_sources()
754 stale = self._fields_with_other_sources(sources, use_saved=False)
755 if stale is None:
756 # Other states are registered above this one: every inherited field may
757 # have another source. Pending field invalidations are covered.
758 previous = self._live_resolved
759 self._live_resolved = self._compute_resolved_snapshot()
760 self._invalid_fields.clear()
761 changed_values = {
762 path: (previous.get(path), value)
763 for path, value in self._live_resolved.items()
764 if previous.get(path) != value
765 }
766 if changed_values:
767 self._set_last_changed_values(changed_values)
768 changed_paths = self.notification_fields_for_value_changes(changed_values)
769 live_resolved_again = True
770 else:
771 self._invalid_fields.update(stale)
772 live_resolved_again = bool(self._invalid_fields)
773 if live_resolved_again:
774 changed_paths = self._recompute_invalid_fields(sources)
775 self._invalid_fields.clear()
776 else:
777 changed_paths = set()
778 self._resolved_from[False] = sources
780 if notify_flash and (live_resolved_again or saved_resolved_again):
781 self._sync_materialized_state(changed_value_fields=changed_paths)
783 return changed_paths
785 def _resolution_sources(
786 self, *, sync_delegates: bool = False
787 ) -> tuple[ResolutionSources, ResolutionSources]:
788 """What this state's fields are resolved from today, live then saved (so the
789 pair is indexed by ``use_saved``): the global config and the objects of the
790 other states registered above.
792 This state's own object is left out (its own parameters are not an outside
793 source), and so are global config objects of ancestors, for which
794 ``build_context_stack`` enters the global config itself.
795 """
796 from objectstate.config import get_base_config_type
797 from objectstate.global_config import get_live_global_config, get_saved_global_config
799 live: list[tuple[str, Any]] = []
800 saved: list[tuple[str, Any]] = []
801 for scope, state in ObjectStateRegistry.registered_ancestors(self.scope_id):
802 if state is self:
803 continue
804 saved_object = state.object_for_resolution(use_saved=True)
805 if is_global_config_type(type(saved_object)):
806 continue
807 saved.append((scope, saved_object))
808 live.append(
809 (scope, state.object_for_resolution(sync_delegate=sync_delegates))
810 )
811 global_config_type = get_base_config_type()
812 return (
813 ResolutionSources(get_live_global_config(global_config_type), tuple(live)),
814 ResolutionSources(get_saved_global_config(global_config_type), tuple(saved)),
815 )
817 def object_for_resolution(self, use_saved: bool = False, sync_delegate: bool = False) -> Any:
818 """The object a state below this one enters as a context layer.
820 For the state of a global config these two objects are the live and the saved
821 global config (``objectstate.global_config``).
823 Saved: the object as last saved, without delegate sync or reconstruction.
824 Live: rebuilt from the current parameters when they changed since the last
825 call; ``sync_delegate`` decides whether a replaced delegate is taken over
826 first, which must not happen while a snapshot is being recomputed.
827 """
828 if use_saved:
829 return self._extraction_target
830 if self._cached_object is not None and not sync_delegate:
831 return self._cached_object
832 return self.to_object(update_delegate=False, sync_delegate=sync_delegate)
834 def _fields_with_other_sources(
835 self, sources: ResolutionSources, *, use_saved: bool
836 ) -> set[str] | None:
837 """Fields of one snapshot that ``sources`` may resolve differently.
839 These are the fields this state leaves unset whose config type inherits a
840 ``(config type, field)`` that an ancestor or the global config stores
841 differently than when the snapshot was computed. A field this state sets
842 itself has no outside source. ``None`` means every field.
843 """
844 resolved_from = self._resolved_from.get(use_saved)
845 if sources.same_objects_as(resolved_from):
846 return set()
847 differing = sources.fields_that_differ(resolved_from)
848 if differing is None:
849 return None
850 own_values = self._saved_parameters if use_saved else self.parameters
851 return {
852 path
853 for config_type, field_name in differing
854 for path in self.inheritance_field_paths(config_type, field_name)
855 if own_values.get(path) is None
856 }
858 def _resolve_saved_again_if_sources_changed(self, sources: ResolutionSources) -> bool:
859 """Resolve the saved snapshot again when its sources changed; say whether it was.
861 The saved sources change when an ancestor is saved, restored or replaced and
862 when the saved global config is replaced, not while values are edited, so the
863 snapshot is resolved in full.
864 """
865 stale = self._fields_with_other_sources(sources, use_saved=True)
866 if stale is not None and not stale:
867 self._resolved_from[True] = sources
868 return False
869 self._saved_resolved = self._compute_resolved_snapshot(use_saved=True)
870 return True
872 def resolve_again_if_sources_changed(self) -> None:
873 """Resolve again now, and tell subscribers, if this state's sources changed.
875 Every resolved read makes this check itself. ``ObjectStateRegistry`` calls it
876 when registration changes so subscribers are told at once.
877 """
878 if self._live_resolved is not None:
879 self._ensure_live_resolved()
881 def _ensure_saved_resolved(self) -> None:
882 """Make ``_saved_resolved`` hold every field, resolved from today's sources."""
883 if self._live_resolved is not None:
884 self._ensure_live_resolved()
885 elif not self._saved_resolved:
886 self._saved_resolved = self._compute_resolved_snapshot(use_saved=True)
888 # DELETED: _create_nested_states() - No longer needed with flat storage
889 # Nested ObjectStates are no longer created - flat storage handles all parameters
891 @staticmethod
892 def _analyze_parameters(
893 obj: Any,
894 exclude_params: list[str] | None = None,
895 ) -> dict[str, Any]:
896 """Analyze object parameters through the canonical introspection owner.
898 Returns dict mapping param_name -> info object with .param_type, .default_value, and .description attributes.
899 """
900 # NOTE: python_introspect is a required dependency; fail loud if missing.
901 # UnifiedParameterAnalyzer is responsible for correctness guarantees:
902 # - defaults come from type/signature (never from instance)
903 # - current instance values are accessed in a lazy-safe way (if needed)
904 from python_introspect import UnifiedParameterAnalyzer
906 return UnifiedParameterAnalyzer.analyze(obj, exclude_params=exclude_params)
908 @staticmethod
909 def _get_nested_dataclass_type(param_type: Any) -> type | None:
910 """Get the nested dataclass type if param_type is a nested dataclass.
912 Args:
913 param_type: The parameter type to check
915 Returns:
916 The dataclass type if nested, None otherwise
917 """
918 from typing import Union, get_args, get_origin
920 # Check Optional[dataclass]
921 origin = get_origin(param_type)
922 if origin is Union:
923 args = get_args(param_type)
924 if len(args) == 2 and type(None) in args:
925 inner_type = next(arg for arg in args if arg is not type(None))
926 if is_dataclass(inner_type):
927 return inner_type
929 # Check direct dataclass (but not the type itself)
930 if is_dataclass(param_type) and not isinstance(param_type, type):
931 # param_type is an instance, not a type - shouldn't happen but handle it
932 return type(param_type)
934 if is_dataclass(param_type):
935 return param_type
937 return None
939 def reset_all_parameters(self) -> None:
940 """Reset all parameters to defaults."""
941 self._in_reset = True
942 try:
943 for param_name in list(self.parameters.keys()):
944 self.reset_parameter(param_name)
945 finally:
946 self._in_reset = False
948 def _dataclass_parameter_updates(self, prefix: str, value: Any) -> dict[str, Any]:
949 """Flatten a dataclass container update into registered child parameters."""
950 if value is None or not is_dataclass(type(value)):
951 return {}
953 prefix_dot = f"{prefix}."
954 if not any(
955 path.startswith(prefix_dot)
956 for path in self.parameters
957 if path != prefix
958 ):
959 return {}
961 updates: dict[str, Any] = {}
962 for field in dataclass_fields(value):
963 field_path = f"{prefix_dot}{field.name}"
964 field_value = DataclassFieldAccess.raw_value(value, field.name)
965 if field_path in self.parameters:
966 updates[field_path] = field_value
967 updates.update(self._dataclass_parameter_updates(field_path, field_value))
968 return updates
970 def _dataclass_ancestor_parameter_updates(
971 self,
972 param_name: str,
973 parameters: dict[str, Any],
974 ) -> dict[str, Any]:
975 """Rebuild dataclass container parameters that own a changed child path."""
977 updates: dict[str, Any] = {}
978 for path in DottedFieldPath(param_name).parent.lineage:
979 prefix = path.value
980 if prefix not in parameters:
981 continue
982 prefix_type = self.parameter_structure.owner_paths.get(prefix)
983 if not (isinstance(prefix_type, type) and is_dataclass(prefix_type)):
984 continue
986 rebuilt_value = self._reconstruct_from_parameter_snapshot(
987 prefix,
988 parameters,
989 )
990 if parameters.get(prefix) != rebuilt_value:
991 updates[prefix] = rebuilt_value
992 parameters[prefix] = rebuilt_value
993 return updates
996 def inheritance_field_paths(
997 self,
998 target_type: ParameterOwner,
999 field_name: str,
1000 ) -> tuple[str, ...]:
1001 """Return flat field paths whose container type can inherit target_type."""
1003 from objectstate.lazy_factory import get_base_type_for_lazy
1005 if isinstance(target_type, type):
1006 target_base_type = get_base_type_for_lazy(target_type) or target_type
1007 else:
1008 target_base_type = target_type
1010 paths: list[str] = []
1011 for dotted_path, container_type in self.parameter_structure.owner_paths.items():
1012 if dotted_path not in self.parameters:
1013 continue
1014 if not (dotted_path.endswith(f".{field_name}") or dotted_path == field_name):
1015 continue
1017 if isinstance(container_type, type):
1018 container_base_type = (
1019 get_base_type_for_lazy(container_type) or container_type
1020 )
1021 else:
1022 container_base_type = container_type
1024 type_matches = container_base_type == target_base_type
1025 if (
1026 not type_matches
1027 and isinstance(container_base_type, type)
1028 and isinstance(target_base_type, type)
1029 ):
1030 type_matches = any(
1031 (get_base_type_for_lazy(mro_class) or mro_class)
1032 == target_base_type
1033 for mro_class in container_base_type.__mro__
1034 )
1036 if type_matches:
1037 paths.append(dotted_path)
1039 return tuple(paths)
1041 def update_parameter(self, param_name: str, value: Any) -> set[str]:
1042 """Update parameter value in state.
1044 Enforces invariants:
1045 1. State mutation → scope+type+field aware cache invalidation
1046 2. State mutation → global token increment (for live context cache)
1048 PERFORMANCE: Three-tier filtering for minimal invalidation:
1049 - SCOPE: Only descendants of this scope (they inherit from us)
1050 - TYPE: Only states with this type in their tree
1051 - FIELD: Only the specific field that changed
1053 Args:
1054 param_name: Name of parameter to update
1055 value: New value
1057 Returns:
1058 Resolved field paths whose visible values changed.
1059 """
1060 # Auto-detect delegate changes before parameter access
1061 self._check_and_sync_delegate()
1063 if param_name not in self.parameters:
1064 logger.warning(
1065 f"⚠️ update_parameter({param_name!r}) called on ObjectState(scope={self.scope_id!r}) "
1066 f"but parameter does not exist. Available: {list(self.parameters.keys())[:5]}..."
1067 )
1068 return set()
1070 child_updates = self._dataclass_parameter_updates(param_name, value)
1071 pending_parameters = dict(self.parameters)
1072 pending_parameters[param_name] = value
1073 for child_path, child_value in child_updates.items():
1074 pending_parameters[child_path] = child_value
1075 ancestor_updates = self._dataclass_ancestor_parameter_updates(
1076 param_name,
1077 pending_parameters,
1078 )
1080 # EARLY EXIT: No change, no invalidation, no flash
1081 current_value = self.parameters[param_name]
1082 changed_child_params = {
1083 child_path
1084 for child_path, child_value in child_updates.items()
1085 if self.parameters.get(child_path) != child_value
1086 }
1087 changed_ancestor_params = {
1088 ancestor_path
1089 for ancestor_path, ancestor_value in ancestor_updates.items()
1090 if self.parameters.get(ancestor_path) != ancestor_value
1091 }
1092 changed_storage_params = set(changed_child_params) | changed_ancestor_params
1093 if current_value != value:
1094 changed_storage_params.add(param_name)
1095 if not changed_storage_params:
1096 return set()
1098 changed_semantic_params = set(changed_child_params)
1099 if current_value != value and not changed_child_params:
1100 changed_semantic_params.add(param_name)
1102 changed_param_values = self._changed_parameter_values(
1103 changed_semantic_params,
1104 pending_parameters,
1105 )
1106 # Establish the pre-mutation resolved authority even for a state that
1107 # has not yet been displayed. Raw edits can preserve the resolved value.
1108 changed_before_the_edit = self._ensure_live_resolved(notify_flash=False)
1109 ObjectStateRegistry.ensure_baseline_snapshot()
1111 # Update state directly (no type conversion - that's VIEW responsibility)
1112 self.parameters[param_name] = value
1113 for child_path, child_value in child_updates.items():
1114 self.parameters[child_path] = child_value
1115 for ancestor_path, ancestor_value in ancestor_updates.items():
1116 self.parameters[ancestor_path] = ancestor_value
1118 # SELF-INVALIDATION: Mark changed fields and same-state inherited siblings.
1119 local_invalidations = set(changed_semantic_params)
1120 for changed_param in changed_semantic_params:
1121 container_type = self.parameter_structure.owner_paths.get(changed_param, type(self.object_instance))
1122 leaf_field_name = DottedFieldPath(changed_param).field_name
1123 local_invalidations.update(
1124 self.inheritance_field_paths(container_type, leaf_field_name)
1125 )
1126 self._invalid_fields.update(local_invalidations)
1127 self._cached_object = None # Invalidate cached reconstructed object
1128 self._cached_object_applied = False
1130 # Materialize the origin once after its live context is authoritative,
1131 # then cascade only to descendants.
1132 changed_paths = self._ensure_live_resolved(notify_flash=False) | changed_before_the_edit
1133 for changed_param in changed_semantic_params:
1134 # SCOPE + TYPE + FIELD AWARE INVALIDATION:
1135 # Get the CONTAINER type for this field (e.g., PartitionFilterConfig for
1136 # 'partition_filter_config.partition_filter')
1137 # This is critical for sibling inheritance: when
1138 # PartitionFilterConfig.partition_filter changes, we need to invalidate
1139 # PathPlanningConfig.partition_filter (which inherits from PartitionFilterConfig)
1140 container_type = self.parameter_structure.owner_paths.get(changed_param, type(self.object_instance))
1142 # Extract leaf field name for invalidation matching
1143 leaf_field_name = DottedFieldPath(changed_param).field_name
1146 ObjectStateRegistry.invalidate_by_type_and_scope(
1147 scope_id=self.scope_id,
1148 changed_type=container_type,
1149 field_name=leaf_field_name,
1150 include_origin=False,
1151 )
1153 # Increment global token for LiveContextService.collect() cache invalidation
1154 ObjectStateRegistry.increment_token(notify=False)
1156 # Raw edits own history/navigation; the resolved delta owns value flashes.
1157 self._set_last_changed_values(changed_param_values)
1158 # Sync materialized state (single point for dirty/sig_diff update + notification)
1159 self._sync_materialized_state(
1160 changed_value_fields=changed_paths
1161 )
1163 # Record snapshot for time-travel (registry-level for coherent system history)
1164 # Leaf edits record directly. Dataclass-container edits record when they
1165 # materialize changed flat child fields; those child paths are the UI and
1166 # time-travel navigation authority for inline editors.
1167 # A field is a container if there are other params that start with "param_name."
1168 param_path = DottedFieldPath(param_name)
1169 is_container = any(
1170 p != param_name and param_path.contains_path(p)
1171 for p in self.parameters.keys()
1172 )
1173 snapshot_field = (
1174 self._select_changed_field(changed_semantic_params)
1175 if is_container and changed_semantic_params
1176 else param_name
1177 )
1178 if snapshot_field:
1179 ObjectStateRegistry.record_snapshot(f"edit {snapshot_field}", self.scope_id)
1181 return changed_paths
1183 def get_resolved_value(self, param_name: str) -> Any:
1184 """Get resolved value for a field from the bulk snapshot.
1186 Args:
1187 param_name: Field name to resolve (can be dotted path like
1188 'path_planning_config.partition_filter')
1190 Returns:
1191 Resolved value from _live_resolved snapshot.
1192 For dataclass container fields, returns a reconstructed dataclass instance
1193 with all sub-fields populated from live resolved values.
1194 """
1195 # Auto-detect delegate changes before resolving values
1196 self._check_and_sync_delegate()
1197 self._ensure_live_resolved()
1198 assert self._live_resolved is not None # Guaranteed by _ensure_live_resolved
1200 # Check if this is a container/dataclass field (has subfields in _live_resolved)
1201 prefix = f"{param_name}."
1202 has_subfields = any(key.startswith(prefix) for key in self._live_resolved.keys())
1204 if has_subfields:
1205 # This is a container field - reconstruct's dataclass from live resolved values
1206 field_type = self.parameter_structure.owner_paths.get(param_name)
1207 if field_type is not None and is_dataclass(field_type):
1208 return self._reconstruct_from_resolved(
1209 param_name, self._live_resolved, self._extraction_target, self.parameter_structure
1210 )
1212 result = self._live_resolved.get(param_name)
1215 return result
1217 def get_saved_resolved_value(self, param_name: str) -> Any:
1218 """Get saved resolved value for a field from the saved snapshot.
1220 Unlike get_resolved_value() which returns live values (including unsaved edits),
1221 this returns the saved baseline with inheritance applied. This is useful for
1222 compilation and other operations that should only consider saved state.
1224 For container fields (dataclasses), this reconstructs the entire nested
1225 dataclass with all sub-fields populated from saved resolved values.
1227 Args:
1228 param_name: Field name to resolve (can be dotted path like
1229 'path_planning_config.partition_filter')
1231 Returns:
1232 Saved resolved value from _saved_resolved snapshot.
1233 For dataclass fields, returns a reconstructed dataclass instance.
1234 """
1235 # Auto-detect delegate changes before resolving values
1236 self._check_and_sync_delegate()
1238 self._ensure_saved_resolved()
1240 # Check if this is a container/dataclass field (has subfields in _saved_resolved)
1241 prefix = f"{param_name}."
1242 has_subfields = any(key.startswith(prefix) for key in self._saved_resolved.keys())
1244 if has_subfields:
1245 # This is a container field - reconstruct's dataclass from saved resolved values
1246 field_type = self.parameter_structure.owner_paths.get(param_name)
1247 if field_type is not None and is_dataclass(field_type):
1248 return self._reconstruct_from_resolved(
1249 param_name, self._saved_resolved, self._extraction_target, self.parameter_structure
1250 )
1252 # Return the simple value (or None if not found)
1253 return self._saved_resolved.get(param_name)
1255 def subfield_semantics(
1256 self,
1257 owner_field_path: DottedFieldPath,
1258 ) -> ObjectStateSubfieldSemanticIndex:
1259 """Project dirty/default/inherited semantics for structural leaves.
1261 The owner field remains the only writable ObjectState field. Returned
1262 leaves are visual/projection identities under that owner, such as
1263 ``source_filters[0].match_type`` for a tuple of dataclass rows.
1264 """
1266 self._check_and_sync_delegate()
1267 self._ensure_live_resolved()
1268 assert self._live_resolved is not None
1269 self._ensure_saved_resolved()
1271 path = owner_field_path
1272 raw_value = self.parameters.get(path.value, MISSING)
1273 resolved_value = self.get_resolved_value(path.value)
1274 saved_resolved_value = self.get_saved_resolved_value(path.value)
1275 declaration = self.parameter_structure.declarations.get(path.value)
1276 signature_default_value = (
1277 declaration.default_value if declaration is not None else MISSING
1278 )
1280 return build_subfield_semantic_index(
1281 owner_field_path=path,
1282 raw_value=raw_value,
1283 resolved_value=resolved_value,
1284 saved_resolved_value=saved_resolved_value,
1285 signature_default_value=signature_default_value,
1286 owner_dirty=path.contains_any(self.dirty_fields),
1287 owner_signature_diff=path.contains_any(self.signature_diff_fields),
1288 )
1291 @classmethod
1292 def _reconstruct_from_resolved(
1293 cls,
1294 prefix: str,
1295 resolved_snapshot: dict[str, Any],
1296 target: Any,
1297 structure: ParameterStructure,
1298 ) -> Any:
1299 """Reconstruct the nominal object and nested dataclasses from a snapshot.
1301 Unified method for both live and saved resolved values. The only difference
1302 is which snapshot dict is passed in (_live_resolved or _saved_resolved).
1304 Args:
1305 prefix: Current path prefix (e.g., 'analysis_consolidation_config')
1306 resolved_snapshot: The snapshot dict to reconstruct from (_live_resolved or _saved_resolved)
1308 Returns:
1309 Reconstructed dataclass instance with resolved values
1310 """
1311 from objectstate.lazy_factory import get_base_type_for_lazy
1313 # Determine the type to reconstruct
1314 if not prefix:
1315 obj_type = type(target)
1316 else:
1317 obj_type = structure.owner_paths.get(prefix)
1318 if obj_type is None:
1319 raise ValueError(f"No type mapping for prefix: {prefix}")
1321 # Normalize to base type for lazy dataclasses
1322 obj_type = get_base_type_for_lazy(obj_type) or obj_type
1324 prefix_dot = f'{prefix}.' if prefix else ''
1326 # Collect direct fields and nested prefixes from resolved snapshot
1327 direct_fields = {}
1328 nested_prefixes = set()
1330 for path, value in resolved_snapshot.items():
1331 if not path.startswith(prefix_dot):
1332 continue
1334 remainder = path[len(prefix_dot):]
1336 if '.' in remainder:
1337 # This is a nested field - collect the first component
1338 first_component = remainder.split('.')[0]
1339 nested_prefixes.add(first_component)
1340 else:
1341 # Direct field of this object
1342 direct_fields[remainder] = value
1344 # Reconstruct nested dataclasses first
1345 for nested_name in nested_prefixes:
1346 nested_path = f'{prefix_dot}{nested_name}'
1347 nested_obj = cls._reconstruct_from_resolved(
1348 nested_path, resolved_snapshot, target, structure
1349 )
1350 direct_fields[nested_name] = nested_obj
1352 # Instantiate the dataclass with all resolved fields
1353 result = obj_type(**direct_fields)
1355 return result
1357 def get_provenance(self, param_name: str) -> tuple[str | None, type] | None:
1358 """Get the source scope_id and type for an inherited field value.
1360 For fields where the local value is None (inherited), returns the scope_id
1361 of the ancestor that provided the value AND the type that has it.
1362 Used for click-to-source navigation in the UI.
1364 The source_type may differ from the local container type due to MRO inheritance.
1365 For example, PartitionFilterConfig.partition_filter might inherit from PathPlanningConfig.
1367 NOTE: Returns provenance even when the resolved value is None (signature default).
1368 A "concrete None" just means the class default is None and nothing overrode it.
1370 Args:
1371 param_name: Field name (can be dotted path like 'path_planning_config.partition_filter')
1373 Returns:
1374 (source_scope_id, source_type): The scope and type that provided the value,
1375 or None if the value is set on this field itself (not inherited).
1377 When another config of this same object provides the value, the scope
1378 is this state's own ``scope_id``. For a state without a scope id that
1379 is ``None``, so ``(None, source_type)`` means "this object, through
1380 ``source_type``".
1381 """
1382 self._ensure_live_resolved()
1383 result = self._live_provenance.get(param_name)
1384 if result is None:
1385 return None # Field is set on this object, not inherited
1386 scope_id, source_type = result
1387 if source_type is None:
1388 return None # No layer and no declared default supplied the field
1389 if scope_id is None and not self._holds_config_type_without_scope(source_type):
1390 return None # No layer held the field; the value is a declared default
1391 return (scope_id, source_type)
1393 def _holds_config_type_without_scope(self, source_type: type) -> bool:
1394 """True when this state has no scope id and its own object holds ``source_type``.
1396 A layer without a scope id supplies values under source scope ``None``. The
1397 only such layer in this state's context stack is its own object, because the
1398 global config and every ancestor are entered under a scope id.
1399 """
1400 return self.scope_id is None and self.find_path_for_type(source_type) is not None
1402 def find_path_for_type(self, container_type: type) -> str | None:
1403 """Find the path prefix for a container type in this ObjectState.
1405 With flat storage, nested configs are identified by their path prefix.
1406 Given a container type (e.g., PathPlanningConfig), returns the path prefix
1407 (e.g., 'path_planning_config').
1409 Handles type normalization: LazyPathPlanningConfig matches PathPlanningConfig.
1411 Args:
1412 container_type: The type to find the path for
1414 Returns:
1415 Path prefix for the type, or None if not found.
1416 Returns "" (empty string) if type is the root object type.
1417 """
1418 from objectstate.lazy_factory import get_base_type_for_lazy
1420 # Normalize the container_type for comparison
1421 container_base = get_base_type_for_lazy(container_type) or container_type
1423 # Check if container_type matches the root object type
1424 root_type = type(self.object_instance)
1425 root_base = get_base_type_for_lazy(root_type) or root_type
1426 if container_base == root_base:
1427 return "" # Root type has no prefix
1429 # Look for paths where the TYPE matches (normalized comparison)
1430 # The path for a nested config is the one WITHOUT a dot suffix that has the type
1431 for path, typ in self.parameter_structure.owner_paths.items():
1432 typ_base = get_base_type_for_lazy(typ) or typ
1433 if not isinstance(typ_base, type):
1434 continue
1435 if typ_base == container_base and '.' not in path:
1436 return path
1438 return None
1440 def project_ui_visible_field_path(self, container_type: type, field_name: str) -> str | None:
1441 """Project a container type and field name to the visible UI field path.
1443 Hidden configuration classes can own inherited values even when the form
1444 renders a visible subclass. ObjectState owns the path/type index, so this
1445 method is the authority for remapping hidden source types to a visible
1446 field path.
1447 """
1448 from objectstate.lazy_factory import get_base_type_for_lazy
1449 from objectstate.ui_visibility import UIVisibilityRegistry
1451 container_base = get_base_type_for_lazy(container_type) or container_type
1452 if UIVisibilityRegistry.is_hidden(container_base):
1453 return self._find_visible_subclass_field_path(container_base, field_name)
1455 path_prefix = self.find_path_for_type(container_type)
1456 if path_prefix is None:
1457 return None
1459 field_path = f"{path_prefix}.{field_name}" if path_prefix else field_name
1460 if field_path not in self.parameter_structure.owner_paths and field_path not in self.parameters:
1461 return None
1462 return field_path
1464 def _find_visible_subclass_field_path(self, hidden_type: type, field_name: str) -> str | None:
1465 """Find a rendered field path for a hidden config type via a visible subclass."""
1466 from objectstate.lazy_factory import get_base_type_for_lazy
1467 from objectstate.ui_visibility import UIVisibilityRegistry
1469 hidden_base = get_base_type_for_lazy(hidden_type) or hidden_type
1470 for path, typ in self.parameter_structure.owner_paths.items():
1471 if "." in path:
1472 continue
1474 typ_base = get_base_type_for_lazy(typ) or typ
1475 if not isinstance(typ_base, type):
1476 continue
1477 if UIVisibilityRegistry.is_hidden(typ_base):
1478 continue
1479 if hidden_base not in typ_base.__mro__:
1480 continue
1482 field_path = f"{path}.{field_name}"
1483 if field_path in self.parameter_structure.owner_paths or field_path in self.parameters:
1484 return field_path
1486 return None
1488 def resolve_for_type(self, container_type: type, field_name: str) -> Any:
1489 """Resolve a field value given the container type and field name.
1491 Convenience method for callers who have a config object but don't know
1492 its path in the flat storage. Finds the path prefix for the container type
1493 and constructs the full dotted path.
1495 Args:
1496 container_type: Type of the containing config (e.g., PathPlanningConfig)
1497 field_name: Field name within that config (e.g., 'partition_filter')
1499 Returns:
1500 Resolved value, or None if not found
1501 """
1502 path_prefix = self.find_path_for_type(container_type)
1503 if path_prefix is None:
1504 # Type not found - try the field_name directly (top-level field)
1505 return self.get_resolved_value(field_name)
1507 full_path = f'{path_prefix}.{field_name}'
1508 return self.get_resolved_value(full_path)
1510 def invalidate_cache(self) -> None:
1511 """Invalidate resolved cache - forces full recompute on next access."""
1512 self._live_resolved = None
1513 self._live_provenance = {} # Provenance must be recomputed with resolved values
1514 self._cached_object = None # Also invalidate cached object
1516 def invalidate_self_and_nested(self) -> None:
1517 """Invalidate this state's cache.
1519 With flat storage, no nested states to invalidate.
1520 """
1521 self._live_resolved = None
1522 self._live_provenance = {} # Provenance must be recomputed with resolved values
1523 self._invalid_fields.clear() # Full invalidation, not field-level
1524 self._cached_object = None
1526 def invalidate_field(self, field_name: str) -> None:
1527 """Mark a specific field as needing recomputation.
1529 PERFORMANCE: Field-level invalidation - only the changed field
1530 needs recomputation, not all 20+ fields in the config.
1531 """
1532 if field_name in self.parameters:
1533 self._invalid_fields.add(field_name)
1535 @staticmethod
1536 def _select_changed_field(changed_paths: set[str]) -> str | None:
1537 """Choose a stable representative changed path for navigation."""
1538 if not changed_paths:
1539 return None
1540 return sorted(changed_paths, key=lambda path: (-path.count("."), path))[0]
1542 def _set_last_changed_values(
1543 self,
1544 changed_values: dict[str, tuple[Any, Any]],
1545 ) -> None:
1546 """Store the latest value changes for UI/time-travel navigation."""
1547 if not changed_values:
1548 return
1549 self._last_changed_concrete_paths = set(changed_values)
1550 self._last_changed_paths = self.notification_fields_for_value_changes(
1551 changed_values
1552 )
1553 self._last_changed_values = copy.deepcopy(changed_values)
1554 self._last_changed_field = self._select_changed_field(
1555 self._last_changed_paths
1556 )
1558 def _changed_parameter_values(
1559 self,
1560 changed_params: set[str],
1561 pending_parameters: dict[str, Any],
1562 ) -> dict[str, tuple[Any, Any]]:
1563 """Return old/new values for changed flat parameter paths."""
1565 changed_values: dict[str, tuple[Any, Any]] = {}
1566 for param_path in pending_parameters:
1567 if param_path not in changed_params:
1568 continue
1569 old_value = self.parameters.get(param_path)
1570 new_value = pending_parameters.get(param_path)
1571 if old_value != new_value:
1572 changed_values[param_path] = (old_value, new_value)
1573 return changed_values
1575 def notification_fields_for_value_changes(
1576 self,
1577 changed_values: dict[str, tuple[Any, Any]],
1578 ) -> set[str]:
1579 """Return ObjectState display paths affected by concrete value changes."""
1581 fields: set[str] = set()
1582 for param_path, (old_value, new_value) in changed_values.items():
1583 if self.parameter_structure.is_container_parameter(param_path, new_value, self.parameters):
1584 continue
1585 fields.add(param_path)
1586 fields.update(
1587 changed_structural_leaf_paths(
1588 owner_field_path=DottedFieldPath(param_path),
1589 old_value=old_value,
1590 new_value=new_value,
1591 )
1592 )
1593 return fields
1595 @staticmethod
1596 def _most_specific_notification_fields(fields: set[str]) -> set[str]:
1597 """Drop container notifications when a concrete descendant path is present."""
1599 if len(fields) < 2:
1600 return set(fields)
1602 paths = tuple(DottedFieldPath(field) for field in fields)
1603 result: set[str] = set()
1604 for field_path in paths:
1605 if any(
1606 field_path.value != candidate.value
1607 and field_path.contains_path(candidate)
1608 for candidate in paths
1609 ):
1610 continue
1611 result.add(field_path.value)
1612 return result
1614 def construction_binding(self) -> StateConstructionBinding:
1615 """Capture construction from this state, not a parallel factory registry."""
1617 arguments = dict(
1618 state_type=type(self),
1619 target=self._extraction_target,
1620 parent_scope=(
1621 self._parent_state.scope_id or "" if self._parent_state is not None else None
1622 ),
1623 exclusions=tuple(self.parameter_structure.exclusions),
1624 parent_field=self._parent_field_name,
1625 )
1626 if self._delegate_attr is not None:
1627 return DelegatedStateBinding(
1628 **arguments,
1629 lifecycle_type=type(self.object_instance),
1630 delegate_attribute=self._delegate_attr,
1631 )
1632 return DirectStateBinding(**arguments)
1634 def apply_prepared_construction(
1635 self,
1636 binding: StateConstructionBinding,
1637 prepared: ObjectState,
1638 parent: ObjectState | None,
1639 ) -> None:
1640 """Attach a prevalidated owner/schema without replacing state callbacks."""
1642 binding.attach(self)
1643 self._parent_state = parent
1644 self._parent_field_name = binding.parent_field
1645 self.parameter_structure = prepared.parameter_structure.clone()
1646 self._cached_object = None
1647 self._cached_object_applied = False
1648 self._invalid_fields.clear()
1650 def update_object_instance(self, new_instance: Any) -> None:
1651 """Replace object_instance with a new instance and re-extract parameters.
1653 This is used when the object being edited is replaced externally (e.g., from
1654 code mode execution). The ObjectState is updated to point to the new instance
1655 and parameters are re-extracted to match the new object's state.
1657 For delegation cases, this updates _extraction_target. For non-delegation cases,
1658 it updates object_instance directly.
1660 Args:
1661 new_instance: The new object instance to extract parameters from
1662 """
1663 self._ensure_live_resolved(notify_flash=False)
1664 previous_parameters = dict(self.parameters)
1665 previous_saved_parameters = dict(self._saved_parameters)
1666 ObjectStateRegistry.mark_snapshot_dirty_scope(self.scope_id)
1667 old_live_resolved = copy.deepcopy(self._live_resolved or {})
1669 if self._delegate_attr is not None:
1670 # Delegation case: verify the new_instance matches the delegate type
1671 if type(new_instance) != type(self._extraction_target):
1672 logger.warning(
1673 f"Type mismatch in update_object_instance for delegated ObjectState: "
1674 f"expected {type(self._extraction_target).__name__}, got {type(new_instance).__name__}"
1675 )
1676 self._extraction_target = new_instance
1677 setattr(self.object_instance, self._delegate_attr, new_instance)
1678 else:
1679 # Non-delegation case: update object_instance directly
1680 self.object_instance = new_instance
1681 self._extraction_target = new_instance
1683 # Re-extract parameters from new instance
1684 self._replace_parameter_structure(
1685 new_instance,
1686 exclude_params=self.parameter_structure.exclusions
1687 )
1689 # Update saved parameters to match
1690 self._saved_parameters = self._copy_parameters_for_saved_baseline(self.parameters)
1692 if not semantic_values_equal(
1693 previous_parameters,
1694 self.parameters,
1695 ) or not semantic_values_equal(
1696 previous_saved_parameters,
1697 self._saved_parameters,
1698 ):
1699 ObjectStateRegistry.mark_snapshot_dirty_scope(self.scope_id)
1701 self._cached_object = None
1702 self._cached_object_applied = False
1703 self._invalid_fields.clear()
1705 self._live_resolved = self._compute_resolved_snapshot(use_saved=False)
1706 self._saved_resolved = self._compute_resolved_snapshot(use_saved=True)
1708 missing = object()
1709 changed_paths = {
1710 path
1711 for path in set(old_live_resolved) | set(self._live_resolved)
1712 if old_live_resolved.get(path, missing)
1713 != self._live_resolved.get(path, missing)
1714 }
1715 changed_values = {
1716 path: (
1717 old_live_resolved.get(path),
1718 self._live_resolved.get(path),
1719 )
1720 for path in changed_paths
1721 }
1723 self._set_last_changed_values(changed_values)
1725 self._notify_resolved_changed(
1726 self.notification_fields_for_value_changes(changed_values),
1727 context="update_object_instance",
1728 )
1730 self._sync_materialized_state()
1732 for path in changed_paths:
1733 container_type = self.parameter_structure.owner_paths.get(path, type(self.object_instance))
1734 leaf_field_name = DottedFieldPath(path).field_name
1735 ObjectStateRegistry.invalidate_by_type_and_scope(
1736 scope_id=self.scope_id,
1737 changed_type=container_type,
1738 field_name=leaf_field_name,
1739 invalidate_saved=True,
1740 )
1742 self._set_last_changed_values(changed_values)
1744 if changed_paths:
1745 ObjectStateRegistry.increment_token(notify=True)
1747 logger.debug(
1748 f"Updated ObjectState(scope={self.scope_id!r}) to new instance of type {type(new_instance).__name__}"
1749 )
1751 def _recompute_invalid_fields(self, sources: ResolutionSources) -> set[str]:
1752 """Recompute only the invalid fields of the live snapshot, from ``sources``.
1754 PERFORMANCE: For explicitly set values, use parameters directly.
1755 Only build context stack for inherited (None) values.
1757 Returns:
1758 Set of paths whose resolved values actually changed (for UI notification).
1759 """
1760 from objectstate.context_manager import build_context_stack
1762 changed_paths: set[str] = set()
1763 changed_values: dict[str, tuple[Any, Any]] = {}
1765 # _live_resolved must exist when this is called (from _ensure_live_resolved)
1766 if self._live_resolved is None:
1767 return changed_paths
1769 # Separate explicit vs inherited fields, skipping container entries
1770 explicit_fields = []
1771 inherited_fields = []
1772 for name in self._invalid_fields:
1773 if name not in self.parameters:
1774 continue
1775 # Safety check: skip any container entries that might have leaked in
1776 # (containers should NOT be in parameters — only leaf fields are tracked)
1777 raw_value = self.parameters[name]
1778 if self.parameter_structure.is_container_parameter(name, raw_value, self.parameters):
1779 continue
1780 if raw_value is not None:
1781 explicit_fields.append(name)
1782 else:
1783 inherited_fields.append(name)
1785 # Explicit values: use parameters directly (no resolution needed)
1786 for name in explicit_fields:
1787 old_val = self._live_resolved.get(name)
1788 explicit_val = self.parameters[name]
1789 if old_val != explicit_val:
1790 changed_paths.add(name)
1791 changed_values[name] = (old_val, explicit_val)
1792 logger.debug(
1793 f"RECOMPUTE EXPLICIT CHANGED [{self.scope_id}] {name}: "
1794 f"old={old_val!r} -> new={explicit_val!r}"
1795 )
1796 self._live_resolved[name] = explicit_val
1797 # Clear provenance for explicit values - they're no longer inherited
1798 if name in self._live_provenance:
1799 del self._live_provenance[name]
1801 # Inherited values: need context stack for lazy resolution + provenance
1802 if inherited_fields:
1803 from objectstate.dual_axis_resolver import resolve_with_provenance
1804 from objectstate.lazy_factory import has_lazy_resolution
1806 # ARCHITECTURAL FIX: Reconstruct current object from parameters for real-time MRO.
1807 # Using object_instance (original saved state) gives stale values, breaking inheritance.
1808 # Calling to_object() with sync_delegate=False reconstructs from CURRENT parameters
1809 # without triggering delegate sync that would invalidate our cache mid-computation.
1810 current_obj = self.to_object(update_delegate=False, sync_delegate=False)
1812 with build_context_stack(
1813 object_instance=current_obj,
1814 ancestor_objects_with_scopes=sources.ancestors,
1815 current_scope_id=self.scope_id,
1816 ):
1817 # For each inherited field, resolve using dual-axis resolution with provenance
1818 for dotted_path in inherited_fields:
1819 container_type = self.parameter_structure.owner_paths.get(dotted_path)
1820 if container_type is None:
1821 logger.debug(f"⚠️ _recompute: {dotted_path} has no container_type in _path_to_type")
1822 continue
1823 # Non-lazy fields with None stay as None rather than
1824 # participating in contextual inheritance.
1825 if not is_dataclass(container_type) or not has_lazy_resolution(container_type):
1826 # Non-lazy field: just use raw value (None)
1827 logger.debug(f"⚠️ _recompute: {dotted_path} has non-lazy container_type={container_type.__name__}, using raw value=None")
1828 old_val = self._live_resolved.get(dotted_path)
1829 raw_val = self.parameters.get(dotted_path)
1830 if old_val != raw_val:
1831 changed_paths.add(dotted_path)
1832 changed_values[dotted_path] = (old_val, raw_val)
1833 self._live_resolved[dotted_path] = raw_val
1834 continue
1835 field_name = DottedFieldPath(dotted_path).field_name
1837 # Use resolve_with_provenance for SINGLE walk that gets both value AND source
1838 value, source_scope_id, source_type = resolve_with_provenance(container_type, field_name)
1840 old_val = self._live_resolved.get(dotted_path)
1841 if old_val != value:
1842 changed_paths.add(dotted_path)
1843 changed_values[dotted_path] = (old_val, value)
1844 logger.debug(
1845 f"RECOMPUTE INHERITED CHANGED [{self.scope_id}] {dotted_path}: "
1846 f"old={old_val!r} -> new={value!r}"
1847 )
1848 else:
1849 logger.debug(
1850 f"RECOMPUTE INHERITED UNCHANGED [{self.scope_id}] {dotted_path}: "
1851 f"old={old_val!r} == new={value!r}"
1852 )
1853 self._live_resolved[dotted_path] = value
1855 # Update provenance for this field
1856 self._live_provenance[dotted_path] = (source_scope_id, source_type)
1858 # Store for navigation only when recomputation observed concrete deltas.
1859 # First-populate paths legitimately have no resolved delta; preserving the
1860 # raw update_parameter() paths keeps time-travel/flash navigation precise.
1861 if changed_values:
1862 self._set_last_changed_values(changed_values)
1863 return self.notification_fields_for_value_changes(changed_values)
1865 @property
1866 def last_changed_field(self) -> str | None:
1867 """Field that most recently changed value (not just dirty status).
1869 This tracks any value change regardless of saved/unsaved state,
1870 useful for time-travel navigation to show what changed in a transition.
1871 """
1872 return self._last_changed_field
1874 def reset_parameter(self, param_name: str) -> None:
1875 """Reset parameter to signature default (None for lazy dataclasses).
1877 Delegates to update_parameter() to ensure consistent invalidation behavior.
1878 """
1879 if param_name not in self.parameters:
1880 return
1882 # Use signature defaults (CLASS defaults), not instance values
1883 # This ensures reset goes back to None for lazy fields, not saved concrete values
1884 declaration = self.parameter_structure.declarations.get(param_name)
1885 default_value = declaration.default_value if declaration is not None else None
1886 self.update_parameter(param_name, default_value)
1888 def get_current_values(self) -> dict[str, Any]:
1889 """
1890 Get current parameter values from state.
1892 With flat storage, this returns the flat dict with dotted paths.
1893 Callers needing nested structure should use to_object() instead.
1895 For ObjectState, this reads directly from self.parameters.
1896 PFM overrides this to also read from widgets.
1897 """
1898 # Auto-detect delegate changes before accessing parameters
1899 self._check_and_sync_delegate()
1900 return dict(self.parameters)
1902 def reconstruct_top_level_parameters(self) -> dict[str, Any]:
1903 """Return top-level parameters with nested dataclass values rebuilt.
1905 ObjectState stores nested fields in a flat dotted-path mapping. Function
1906 pattern kwargs need the top-level call signature values, not the flat UI
1907 storage shape. This method exposes that boundary without requiring UI
1908 widgets or service code to reach into ObjectState private helpers.
1909 """
1910 self._check_and_sync_delegate()
1912 values: dict[str, Any] = {}
1913 root_type = type(self._extraction_target)
1914 top_level_params = {
1915 key
1916 for key in self.parameters
1917 if "." not in key
1918 }
1920 for param_name in top_level_params:
1921 param_type = self.parameter_structure.owner_paths.get(param_name)
1922 nested_dataclass = (
1923 param_type is not None
1924 and is_dataclass(param_type)
1925 and param_type != root_type
1926 )
1927 if nested_dataclass:
1928 values[param_name] = self._reconstruct_from_prefix(param_name)
1929 else:
1930 values[param_name] = self.parameters[param_name]
1932 return values
1934 # ==================== MATERIALIZED DIFFS ====================
1936 @property
1937 def dirty_fields(self) -> set[str]:
1938 """Fields whose live and saved resolved values differ, among today's sources."""
1939 self._ensure_live_resolved()
1940 return self._dirty_fields
1942 @property
1943 def signature_diff_fields(self) -> set[str]:
1944 """Fields where raw != signature_default."""
1945 return self._signature_diff_fields
1947 @property
1948 def is_raw_dirty(self) -> bool:
1949 """Check if raw parameters differ from saved parameters (not resolved values)."""
1950 return self.parameters != self._saved_parameters
1952 def _update_raw_dirty(self) -> bool:
1953 """Recompute raw dirty state, return True if it changed."""
1954 new_raw_dirty = self.is_raw_dirty
1955 if new_raw_dirty != self._raw_dirty:
1956 self._raw_dirty = new_raw_dirty
1957 return True
1958 return False
1960 def _compute_dirty_fields(self) -> set[str]:
1961 """Compute dirty set from live vs saved caches."""
1962 if self._live_resolved is None:
1963 return set()
1964 dirty = set()
1966 def _normalize_callable_value(value: Any) -> Any:
1967 if value is None:
1968 return None
1969 if callable(value):
1970 return value
1971 if isinstance(value, list):
1972 normalized = []
1973 for item in value:
1974 if callable(item):
1975 normalized.append(item)
1976 continue
1977 if isinstance(item, tuple) and len(item) == 2 and callable(item[0]) and isinstance(item[1], dict):
1978 pruned_kwargs = {k: v for k, v in item[1].items() if v is not None}
1979 normalized.append(item[0] if not pruned_kwargs else (item[0], pruned_kwargs))
1980 if len(normalized) == 1 and callable(normalized[0]):
1981 return normalized[0]
1982 return normalized
1983 return value
1984 for k in (self._live_resolved.keys() | self._saved_resolved.keys()):
1985 live_val = _normalize_callable_value(self._live_resolved.get(k))
1986 saved_val = _normalize_callable_value(self._saved_resolved.get(k))
1988 if live_val != saved_val:
1989 dirty.add(k)
1990 logger.debug(f"🔴 DIRTY_FIELD: scope={self.scope_id!r} field={k!r} live={live_val!r} saved={saved_val!r}")
1991 if dirty:
1992 logger.debug(f"🔴 DIRTY_SUMMARY: scope={self.scope_id!r} dirty_fields={dirty}")
1993 return dirty
1995 def _compute_signature_diff_fields(self) -> set[str]:
1996 """Compute signature-diff set from parameters vs defaults.
1998 Any field that differs from its signature default is included.
1999 Defaults come from the same declarations used for reset, including
2000 declared nested containers.
2001 """
2002 result = set()
2003 for k, v in self.parameters.items():
2004 if k in self.parameter_structure.declarations:
2005 sig_default = self.parameter_structure.declarations[k].default_value
2006 is_diff = v != sig_default
2007 if is_diff:
2008 result.add(k)
2009 return result
2011 def _update_dirty_fields(self) -> set[str]:
2012 """Recompute _dirty_fields, return set of fields that changed dirty status.
2014 Returns fields that either became dirty OR became clean.
2015 Empty set means no change.
2016 """
2017 new_dirty = self._compute_dirty_fields()
2018 if new_dirty != self._dirty_fields:
2019 # Symmetric difference: fields that changed dirty status in either direction
2020 changed_fields = new_dirty ^ self._dirty_fields
2021 self._dirty_fields = new_dirty
2022 return changed_fields
2023 return set()
2025 def _update_signature_diff_fields(self) -> set[str]:
2026 """Recompute signature differences and retain the changed field paths."""
2027 new_sig_diff = self._compute_signature_diff_fields()
2028 changed_fields = new_sig_diff ^ self._signature_diff_fields
2029 self._signature_diff_fields = new_sig_diff
2030 return changed_fields
2032 def _sync_materialized_state(
2033 self,
2034 *,
2035 changed_value_fields: set[str] | None = None,
2036 emit_notifications: bool = True,
2037 ) -> bool:
2038 """Single point where materialized diffs are recomputed and notified.
2040 Call this after ANY mutation that could affect:
2041 - _live_resolved (affects dirty_fields)
2042 - _saved_resolved (affects dirty_fields)
2043 - parameters (affects signature_diff_fields)
2045 Correctness guarantee: All mutation paths call this ONE method.
2047 Resolved-value subscribers receive only value changes. Materialized
2048 state subscribers also receive dirty/signature transitions, including
2049 saving or changing explicitness without changing the resolved value.
2050 """
2051 raw_dirty_changed = self._update_raw_dirty()
2052 dirty_status_changed_fields = self._update_dirty_fields()
2053 signature_changed_fields = self._update_signature_diff_fields()
2054 value_notification_fields = self._most_specific_notification_fields(
2055 changed_value_fields or set()
2056 )
2057 notification_fields = self._most_specific_notification_fields(
2058 dirty_status_changed_fields
2059 | signature_changed_fields
2060 | value_notification_fields
2061 )
2063 materialized_changed = raw_dirty_changed or bool(notification_fields)
2064 if materialized_changed:
2065 ObjectStateRegistry.mark_snapshot_dirty_scope(self.scope_id)
2067 if emit_notifications:
2068 self._notify_resolved_changed(
2069 value_notification_fields,
2070 context="_sync_materialized_state",
2071 )
2073 if materialized_changed:
2074 self._notify_state_changed(notification_fields)
2076 return materialized_changed
2078 def prepare_for_snapshot_capture(self) -> None:
2079 """Synchronize only pending resolved work before immutable capture."""
2080 if self._live_resolved is not None and not self._invalid_fields:
2081 return
2083 changed_paths = self._ensure_live_resolved(notify_flash=False)
2084 self._sync_materialized_state(
2085 changed_value_fields=changed_paths,
2086 emit_notifications=False,
2087 )
2089 # ==================== SAVED STATE / DIRTY TRACKING ====================
2091 def _compute_resolved_snapshot(self, use_saved: bool = False) -> dict[str, Any]:
2092 """Resolve editor fields using the selected live or saved ancestor epoch."""
2093 if not isinstance(self.parameters, dict):
2094 raise TypeError("ObjectState.parameters must remain a canonical flat dictionary.")
2095 if not isinstance(self._saved_parameters, dict):
2096 raise TypeError("ObjectState._saved_parameters must remain a canonical flat dictionary.")
2097 sources = self._resolution_sources(sync_delegates=True)[use_saved]
2098 self._resolved_from[use_saved] = sources
2099 current_obj = (
2100 self.saved_object if use_saved else self.to_object(update_delegate=False)
2101 )
2102 parameters = self._saved_parameters if use_saved else self.parameters
2103 snapshot, provenance = self._resolve_parameter_snapshot(
2104 self.parameters,
2105 parameters,
2106 self.parameter_structure,
2107 current_obj,
2108 self.scope_id,
2109 sources.ancestors,
2110 use_saved=use_saved,
2111 )
2112 if not use_saved:
2113 self._live_provenance = provenance
2114 return snapshot
2116 @classmethod
2117 def _resolve_parameter_snapshot(
2118 cls,
2119 parameters: dict[str, Any],
2120 values: dict[str, Any],
2121 structure: ParameterStructure,
2122 current_obj: Any,
2123 scope_id: str | None,
2124 ancestor_objects_with_scopes: Sequence[tuple[str, Any]],
2125 *,
2126 use_saved: bool,
2127 ) -> tuple[dict[str, Any], dict[str, tuple[str | None, type | None]]]:
2128 """Resolve raw fields and their provenance in one scoped semantic epoch."""
2129 from objectstate.context_manager import build_context_stack
2130 from objectstate.dual_axis_resolver import resolve_with_provenance
2131 from objectstate.lazy_factory import has_lazy_resolution
2133 snapshot: dict[str, Any] = {}
2134 provenance: dict[str, tuple[str | None, type | None]] = {}
2136 # UNIFIED: Resolve ALL fields in single context stack
2137 # For each path, check if it has a lazy dataclass container type
2138 with build_context_stack(
2139 object_instance=current_obj,
2140 ancestor_objects_with_scopes=ancestor_objects_with_scopes,
2141 current_scope_id=scope_id,
2142 use_live=not use_saved,
2143 ):
2144 for dotted_path in parameters.keys():
2145 raw_value = values.get(dotted_path)
2146 container_type = structure.owner_paths.get(dotted_path)
2148 # Check if this path is a CONTAINER entry (value is a nested dataclass)
2149 # vs a LEAF field (value is primitive, even if container_type is a dataclass)
2150 if structure.is_container_parameter(dotted_path, raw_value, parameters):
2151 # Container-level entry - SKIP from snapshot
2152 # Containers are kept in parameters for UI rendering but excluded from
2153 # dirty comparison since we compare leaf fields instead
2154 pass
2155 elif (
2156 container_type is not None
2157 and is_dataclass(container_type)
2158 and has_lazy_resolution(container_type)
2159 ):
2160 # Leaf field inside a LAZY dataclass - resolve value AND provenance in ONE walk
2161 # CRITICAL: Only resolve for lazy dataclasses! Non-lazy dataclasses with None
2162 # defaults should keep None as-is, not trigger inheritance resolution.
2163 # This handles both:
2164 # - Nested fields (processing_config.group_by)
2165 # - Top-level fields on root (num_workers on PipelineConfig)
2166 field_name = DottedFieldPath(dotted_path).field_name
2168 if raw_value is None:
2169 # Field needs resolution - use combined resolve + provenance walk
2170 resolved_val, source_scope, source_type = resolve_with_provenance(container_type, field_name)
2171 snapshot[dotted_path] = resolved_val
2173 # Keep provenance in the same epoch as the resolved value.
2174 provenance[dotted_path] = (source_scope, source_type)
2175 else:
2176 # Field has concrete local value - no resolution needed
2177 resolved_val = raw_value
2178 snapshot[dotted_path] = resolved_val
2180 logger.debug(
2181 f"SNAPSHOT [{scope_id}] {dotted_path}: "
2182 f"raw={raw_value!r} -> resolved={resolved_val!r} (type={type(resolved_val).__name__})"
2183 )
2184 else:
2185 # Non-lazy field (regular dataclass, class instance, callable) - use raw value directly
2186 # None stays as None, no inheritance resolution
2187 snapshot[dotted_path] = raw_value
2189 return snapshot, provenance
2191 def mark_saved(self) -> None:
2192 """Mark current state as saved baseline.
2194 UNIFIED: Works for any object_instance type. For the state of a global
2195 config, the saved object also becomes the saved global config.
2197 CRITICAL: Invalidates descendant caches for any parameters that changed.
2198 This ensures that when saving, other windows that inherited from the
2199 old saved values get their caches invalidated so they pick up new values.
2200 This mirrors what restore_saved() does but in the opposite direction.
2202 Invalidation is based on comparing the OLD object_instance (about to be replaced)
2203 with the NEW self.parameters (live values used for reconstruction).
2204 """
2205 logger.debug(f"🐛 mark_saved: ENTER for scope={self.scope_id!r}, obj_type={type(self.object_instance).__name__}, _saved_parameters is None={self._saved_parameters is None}")
2206 # Ensure live cache is populated for accurate dirty computation post-save
2207 self._ensure_live_resolved(notify_flash=False)
2209 # CRITICAL: Extract old values from object_instance BEFORE rebuilding it
2210 # These are the values that descendants might be inheriting from
2211 old_instance_values = {}
2212 if not isinstance(self.object_instance, type):
2213 # Extract raw attribute values from the old object_instance
2214 for param_name in self.parameters.keys():
2215 # Skip container entries (nested dataclass instances)
2216 if param_name in self.parameters:
2217 raw_value = self.parameters.get(param_name)
2218 if self.parameter_structure.is_container_parameter(param_name, raw_value, self.parameters):
2219 continue
2221 # Get the old value by navigating dotted path on the extraction target
2222 # For delegation, parameters are on the delegate, not object_instance
2223 try:
2224 # Navigate through nested attributes for dotted paths
2225 obj = self._extraction_target
2226 parts = param_name.split('.')
2227 for part in parts:
2228 obj = object.__getattribute__(obj, part)
2229 old_instance_values[param_name] = obj
2230 except AttributeError:
2231 # Field doesn't exist on extraction target, skip it
2232 pass
2234 # Find parameters that differ between old object_instance and new live parameters
2235 # These are the fields that changed and need descendant invalidation
2236 changed_params = []
2237 for param_name in self.parameters.keys():
2238 # Skip container entries
2239 raw_value = self.parameters.get(param_name)
2240 if self.parameter_structure.is_container_parameter(param_name, raw_value, self.parameters):
2241 continue
2243 old_value = old_instance_values.get(param_name)
2244 new_value = self.parameters.get(param_name)
2245 if old_value != new_value:
2246 changed_params.append(param_name)
2248 # CRITICAL: Rebuild extraction target BEFORE invalidating descendants
2249 # Descendants will recompute using parent's extraction target, so it must have new values!
2250 if not isinstance(self.object_instance, type):
2251 if self._delegate_attr is not None:
2252 # DELEGATION: to_object() returns the delegate and updates it on object_instance
2253 # as a side effect. Keep object_instance unchanged (it's the lifecycle object).
2254 # _extraction_target is updated to point to the new delegate.
2255 self._extraction_target = self.to_object(update_delegate=True)
2256 else:
2257 # NON-DELEGATION: to_object() returns the reconstructed object_instance
2258 self.object_instance = self.to_object(update_delegate=True)
2259 self._extraction_target = self.object_instance # Keep in sync
2261 # Update saved parameters (after object_instance update, before invalidation)
2262 self._saved_parameters = self._copy_parameters_for_saved_baseline(self.parameters)
2264 # NOW invalidate descendant caches AFTER object_instance is updated
2265 # This ensures descendants see the NEW object_instance when they recompute
2266 # CRITICAL: Also invalidate saved_resolved cache so descendants recompute their saved baseline
2267 logger.debug(f"🔧 mark_saved: Starting invalidation for changed_params={changed_params}, scope={self.scope_id!r}")
2268 for param_name in changed_params:
2269 container_type = self.parameter_structure.owner_paths.get(param_name, type(self.object_instance))
2270 leaf_field_name = DottedFieldPath(param_name).field_name
2271 logger.debug(f"🔧 mark_saved: Invalidating param={param_name}, container_type={container_type.__name__ if hasattr(container_type, '__name__') else type(container_type).__name__}, leaf_field={leaf_field_name}")
2273 ObjectStateRegistry.invalidate_by_type_and_scope(
2274 scope_id=self.scope_id,
2275 changed_type=container_type,
2276 field_name=leaf_field_name,
2277 invalidate_saved=True # Invalidate saved baseline for descendants
2278 )
2280 # Compute new saved resolved using SAVED ancestor baselines (use_saved=True)
2281 # This ensures saved baseline is computed relative to other saved baselines
2282 logger.debug(f"🔧 mark_saved: Computing new saved_resolved for scope={self.scope_id!r}")
2283 new_saved_resolved = self._compute_resolved_snapshot(use_saved=True)
2284 logger.debug(f"🔧 mark_saved: New saved_resolved computed, keys={list(new_saved_resolved.keys())[:5]}...")
2286 # Update saved resolved baseline
2287 self._saved_resolved = new_saved_resolved
2289 # Invalidate cached object so next to_object() call rebuilds
2290 self._cached_object = None
2292 # Sync materialized state (single point for dirty/sig_diff update + notification)
2293 self._sync_materialized_state()
2295 # Record snapshot for time-travel (registry-level) - ONLY if there were actual changes
2296 # This prevents no-op snapshots (e.g., saving a window where only sibling state changed)
2297 if changed_params:
2298 ObjectStateRegistry.record_snapshot("save", self.scope_id)
2300 # CRITICAL FIX: Propagate saved baseline update to ALL descendant states
2301 # When an ancestor's saved baseline changes, all descendants must recompute
2302 # their _saved_resolved to reflect the new ancestor saved values. This ensures that
2303 # when GlobalPipelineConfig is saved, plates/steps clear their dirty markers (*).
2304 logger.debug(f"🔧 mark_saved: Propagating saved baseline to descendants for scope={self.scope_id!r}")
2305 logger.debug(f"🔧 mark_saved: Total states in registry: {len(ObjectStateRegistry._states)}")
2307 # Collect descendant scopes first to avoid modifying registry during iteration.
2308 changed_scope = ObjectStateRegistry._normalize_scope_id(self.scope_id)
2309 descendant_scopes = [
2310 s.scope_id for s in ObjectStateRegistry._states.values()
2311 if changed_scope.strictly_contains(s.scope_id)
2312 ]
2313 logger.debug(f"🔧 mark_saved: Found {len(descendant_scopes)} descendant scopes: {descendant_scopes}")
2315 for descendant_scope in descendant_scopes:
2316 state = ObjectStateRegistry._states.get(descendant_scope)
2317 if state is not None:
2318 logger.debug(f"🔧 mark_saved: Processing descendant state scope={descendant_scope!r}, obj_type={type(state.object_instance).__name__}")
2319 logger.debug(f"🐛 mark_saved descendant: _saved_parameters is None={state._saved_parameters is None}, parameters is None={state.parameters is None}")
2320 # Log dirty fields BEFORE recompute
2321 logger.debug(f"🔧 mark_saved: BEFORE recompute - dirty_fields={state._dirty_fields}, _saved_resolved_keys={list(state._saved_resolved.keys())[:3]}")
2322 # Recompute descendant's saved_resolved using new ancestor saved values
2323 try:
2324 state._saved_resolved = state._compute_resolved_snapshot(use_saved=True)
2325 except AttributeError as e:
2326 logger.error(f"🐛 mark_saved descendant: ERROR in _compute_resolved_snapshot for scope={descendant_scope!r}! Error: {e}")
2327 logger.error(f"🐛 mark_saved descendant: state._saved_parameters type={type(state._saved_parameters).__name__}, is None={state._saved_parameters is None}")
2328 logger.error(f"🐛 mark_saved descendant: state.parameters type={type(state.parameters).__name__}, is None={state.parameters is None}")
2329 raise
2330 # Sync materialized state so dirty fields are recalculated
2331 state._sync_materialized_state()
2332 # Log dirty fields AFTER recompute
2333 logger.debug(f"🔧 mark_saved: AFTER recompute - dirty_fields={state._dirty_fields}, _saved_resolved_keys={list(state._saved_resolved.keys())[:3]}")
2334 else:
2335 logger.warning(f"🔧 mark_saved: Descendant scope {descendant_scope!r} not found in registry!")
2337 def restore_saved(self, *, propagate_descendants: bool = True) -> None:
2338 """Restore parameters to the last saved baseline (from object_instance).
2340 UNIFIED: Works for any object_instance type.
2342 CRITICAL: Invalidates descendant caches for any parameters that changed.
2343 This ensures that when closing a window without saving, other windows
2344 that inherited from the unsaved values get their caches invalidated.
2346 Also emits on_resolved_changed for THIS state so same-level observers
2347 (like list items subscribed to this ObjectState) flash when values revert.
2348 """
2349 if isinstance(self.object_instance, type):
2350 self.invalidate_cache()
2351 self._sync_materialized_state()
2352 return
2354 # If there are no unsaved edits, restoring is a semantic no-op and should not
2355 # create time-travel snapshots (noise).
2356 if not self.is_raw_dirty:
2357 return
2359 # Coalesce all restore side-effects into a single snapshot
2360 with ObjectStateRegistry.atomic(f"restore {self.scope_id}"):
2361 self._restore_saved_impl(propagate_descendants=propagate_descendants)
2362 return
2364 def _restore_saved_impl(self, *, propagate_descendants: bool = True) -> None:
2365 """Internal restore implementation (wrapped by restore_saved atomic block)."""
2367 # Find parameters that differ from saved baseline AND capture their container types
2368 # BEFORE clearing parameters (we need _path_to_type)
2369 changed_params_with_types = []
2370 logger.debug(f"🐛 restore_saved: About to iterate parameters, _saved_parameters type={type(self._saved_parameters).__name__}, is None={self._saved_parameters is None}")
2371 for param_name, current_value in self.parameters.items():
2372 try:
2373 saved_value = self._saved_parameters.get(param_name)
2374 except AttributeError as e:
2375 logger.error(f"🐛 restore_saved: ERROR accessing _saved_parameters.get({param_name!r})! _saved_parameters type={type(self._saved_parameters).__name__}, is None={self._saved_parameters is None}, scope={self.scope_id!r}")
2376 raise
2377 if current_value != saved_value:
2378 container_type = self.parameter_structure.owner_paths.get(param_name, type(self.object_instance))
2379 leaf_field_name = DottedFieldPath(param_name).field_name
2380 changed_params_with_types.append((param_name, container_type, leaf_field_name))
2382 # Clear and re-extract from the saved baseline
2383 # CRITICAL: For delegation, extract from the delegate (pipeline_config), not the lifecycle object.
2384 # This keeps flat parameters aligned with the form's target object after window close/reopen.
2385 # Also refresh _extraction_target in case the delegate attribute was replaced externally.
2386 extraction_target = self._extraction_target
2387 if self._delegate_attr is not None:
2388 try:
2389 extraction_target = getattr(self.object_instance, self._delegate_attr)
2390 self._extraction_target = extraction_target
2391 except Exception:
2392 # Fallback to existing extraction target if delegate access fails
2393 extraction_target = self._extraction_target
2394 self._replace_parameter_structure(
2395 extraction_target,
2396 exclude_params=self.parameter_structure.exclusions,
2397 )
2399 # CRITICAL: Also restore _saved_parameters to match current parameters
2400 # After restore, parameters == saved (both extracted from object_instance)
2401 self._saved_parameters = self._copy_parameters_for_saved_baseline(self.parameters)
2403 self.invalidate_cache()
2404 # Invalidate cached reconstructed object (may contain unsaved edits)
2405 self._cached_object = None
2406 self._cached_object_applied = False
2408 # CRITICAL: Recompute _saved_resolved to match the restored state
2409 # Time travel may have overwritten _saved_resolved with snapshot values,
2410 # but after restore_saved(), _saved_resolved should reflect object_instance
2411 self._saved_resolved = self._compute_resolved_snapshot(use_saved=True)
2413 # Recompute _live_resolved to reflect any unsaved changes from higher-level ObjectStates
2414 # This ensures that unsaved resolved changes from other windows are preserved
2415 self._live_resolved = self._compute_resolved_snapshot(use_saved=False)
2417 # NOW invalidate descendant caches for each changed parameter
2418 # This must happen AFTER restoring parameters so descendants see restored values
2419 for param_name, container_type, leaf_field_name in changed_params_with_types:
2420 ObjectStateRegistry.invalidate_by_type_and_scope(
2421 scope_id=self.scope_id,
2422 changed_type=container_type,
2423 field_name=leaf_field_name
2424 )
2426 # Optionally propagate restore to descendant ObjectStates so their parameters reflect saved baseline.
2427 #
2428 # This is important when restoring a parent that *owns* descendant parameter state (e.g. Step -> function
2429 # ObjectStates) so canceling the parent edit resets child ObjectStates.
2430 #
2431 # However, for delegation-based parents that act as context providers (e.g. orchestrator -> pipeline_config),
2432 # restoring the parent should generally *not* restore descendant raw parameters (steps). Descendants should
2433 # only have their resolved caches invalidated so they re-resolve against the restored context.
2434 # The global scope only provides context to every other scope; it owns none of
2435 # their parameters, so restoring it never restores them.
2436 own_scope = ScopeKey(self.scope_id)
2437 if propagate_descendants and not own_scope.is_global:
2438 descendant_scopes = [
2439 scope
2440 for scope in ObjectStateRegistry._states.keys()
2441 if own_scope.strictly_contains(scope)
2442 ]
2443 for descendant_scope in descendant_scopes:
2444 state = ObjectStateRegistry._states.get(descendant_scope)
2445 if state:
2446 state._restore_saved_impl(propagate_descendants=True)
2448 # Emit on_resolved_changed for changed params so SAME-LEVEL observers flash
2449 # (e.g., list item subscribed to this ObjectState sees the revert as a change)
2450 did_atomic_restore = False
2451 if changed_params_with_types:
2452 changed_paths = {param_name for param_name, _, _ in changed_params_with_types}
2453 # Coalesce any updates triggered by callbacks into a single snapshot
2454 if self._parent_state is None:
2455 with ObjectStateRegistry.atomic(f"restore {self.scope_id}"):
2456 self._notify_resolved_changed(
2457 changed_paths,
2458 context="restore_saved",
2459 )
2460 did_atomic_restore = True
2461 else:
2462 self._notify_resolved_changed(
2463 changed_paths,
2464 context="restore_saved",
2465 )
2467 # Sync materialized state (single point for dirty/sig_diff update + notification)
2468 self._sync_materialized_state()
2470 # Record snapshot for time-travel (registry-level) - ONLY if there were changes
2471 if changed_params_with_types and not did_atomic_restore:
2472 ObjectStateRegistry.record_snapshot("restore", self.scope_id)
2474 def should_skip_updates(self) -> bool:
2475 """Check if updates should be skipped due to batch operations."""
2476 return self._in_reset or self._block_cross_window_updates
2478 # ==================== FLAT STORAGE METHODS (NEW) ====================
2480 @classmethod
2481 def _extract_all_parameters_flat(
2482 cls,
2483 obj: Any,
2484 parameters: dict[str, Any],
2485 structure: ParameterStructure,
2486 prefix: str = '',
2487 exclude_params: list[str] | None = None,
2488 ) -> None:
2489 """Recursively extract parameters into flat dict with dotted paths.
2491 Populate raw parameters and the existing declaration structure through
2492 the canonical parameter analyzer, without constructing an editing state.
2494 Args:
2495 obj: Object to extract from (dataclass instance, callable, or regular object)
2496 prefix: Current path prefix (e.g., 'partition_filter_config')
2497 exclude_params: List of top-level parameter names to exclude
2498 """
2499 exclude_params = exclude_params or []
2501 obj_type = type(obj)
2502 is_function = isinstance(obj, FunctionType)
2503 owner_target: ParameterOwner = obj if is_function else obj_type
2505 # Delegate signature default extraction to python_introspect (it must derive
2506 # defaults from the type/signature and avoid instance attribute reads).
2507 param_info = cls._analyze_parameters(obj, exclude_params if not prefix else [])
2509 logger.debug(f"🔧 _extract_all_parameters_flat: obj_type={obj_type.__name__}, prefix={prefix!r}, param_info keys={list(param_info.keys())}")
2511 for param_name, info in param_info.items():
2512 # Skip excluded parameters (only at top level)
2513 if not prefix and param_name in exclude_params:
2514 continue
2516 # Build dotted path
2517 dotted_path = f'{prefix}.{param_name}' if prefix else param_name
2519 # Get current value
2520 if is_function:
2521 # For functions: use signature default from UnifiedParameterAnalyzer
2522 # (functions don't have instance attributes)
2523 current_value = info.default_value
2524 else:
2525 # For class instances: bypass lazy resolution via object.__getattribute__
2526 try:
2527 current_value = object.__getattribute__(obj, param_name)
2528 except AttributeError:
2529 current_value = info.default_value
2531 structure.declarations[dotted_path] = info
2533 # Check if this is a nested dataclass
2534 # First try from type annotation, then fall back to checking actual value
2535 nested_type = cls._get_nested_dataclass_type(info.param_type)
2537 # The actual dataclass owns both its flattened fields and their
2538 # reconstruction, including concrete nominal subtypes and lazy
2539 # overlays. An annotation supplies the owner only without a value.
2540 if current_value is not None and is_dataclass(type(current_value)):
2541 nested_type = type(current_value)
2543 if nested_type is not None and current_value is not None:
2544 # Store the nested config type reference at this path
2545 structure.owner_paths[dotted_path] = nested_type
2547 # Store the nested dataclass instance in parameters (needed for UI rendering)
2548 parameters[dotted_path] = current_value
2550 # Recurse into nested dataclass for child fields
2551 cls._extract_all_parameters_flat(
2552 current_value, parameters, structure, prefix=dotted_path, exclude_params=[]
2553 )
2554 else:
2555 # Leaf field - store value and container type
2556 parameters[dotted_path] = current_value
2557 # Store the owner target that has this field.
2558 structure.owner_paths[dotted_path] = owner_target
2560 def _replace_parameter_structure(
2561 self,
2562 obj: Any,
2563 *,
2564 exclude_params: list[str] | None = None,
2565 initial_values: dict[str, Any] | None = None,
2566 ) -> None:
2567 """Extract one canonical flat parameter structure and index it once."""
2569 self.parameters.clear()
2570 self.parameter_structure = ParameterStructure.for_target(obj, exclude_params)
2571 self._extract_all_parameters_flat(
2572 obj,
2573 self.parameters,
2574 self.parameter_structure,
2575 prefix="",
2576 exclude_params=exclude_params,
2577 )
2578 if initial_values:
2579 for param_name, value in initial_values.items():
2580 self.parameters[param_name] = value
2581 self.parameters.update(
2582 self._dataclass_parameter_updates(param_name, value)
2583 )
2584 self._index_parameter_paths()
2586 def _index_parameter_paths(self) -> None:
2587 """Index immutable direct-child topology from canonical flat parameters."""
2589 paths_by_owner: dict[DottedFieldPath, list[DottedFieldPath]] = {}
2590 for dotted_path in self.parameters:
2591 path = DottedFieldPath(dotted_path)
2592 paths_by_owner.setdefault(path.parent, []).append(path)
2593 self.parameter_structure.direct_paths = {
2594 owner_path: tuple(parameter_paths)
2595 for owner_path, parameter_paths in paths_by_owner.items()
2596 }
2598 def to_object(self, *, update_delegate: bool = False, sync_delegate: bool = True) -> Any:
2599 """Reconstruct object from flat parameters with updated nested configs.
2601 BOUNDARY METHOD - EXPENSIVE - only call at system boundaries:
2602 - Save operation
2603 - Execute operation
2604 - Serialization
2606 UNIFIED: Works for ANY object_instance type.
2607 - Python functions: can't copy, return original
2608 - Everything else: shallow copy + reconstruct nested dataclass fields
2610 DELEGATION: If __objectstate_delegate__ was used:
2611 - Reconstructs the delegate (e.g., pipeline_config)
2612 - If update_delegate=True, updates the delegate attribute on object_instance
2613 - Returns the reconstructed delegate (NOT object_instance)
2614 - Callers needing the lifecycle object (orchestrator) should use state.object_instance
2616 Args:
2617 update_delegate: If True, apply reconstructed delegate to object_instance
2618 sync_delegate: If True (default), check for delegate changes before reconstruction.
2619 Set to False during cache recomputation to avoid re-entrant invalidation.
2621 Returns:
2622 The reconstructed object that matches the stored parameters.
2623 For delegation, this is the delegate type (config), not the lifecycle object.
2624 """
2625 # Auto-detect delegate changes before reconstruction (unless explicitly disabled)
2626 if sync_delegate:
2627 self._check_and_sync_delegate()
2629 if self._cached_object is not None:
2630 if not update_delegate:
2631 return self._cached_object
2632 if self._delegate_attr is None or self._cached_object_applied:
2633 return self._cached_object
2634 # Apply cached delegate to lifecycle object when requested
2635 setattr(self.object_instance, self._delegate_attr, self._cached_object)
2636 self._cached_object_applied = True
2637 return self._cached_object
2639 # UNIFIED: reconstruct nested dataclass fields
2640 # Works for dataclass, non-dataclass class instances, AND functions
2641 import copy
2643 # For delegation, work with the extraction target (delegate), not object_instance
2644 target = self._extraction_target
2646 # Collect ALL top-level field updates from self.parameters
2647 # This includes both primitive fields AND nested dataclass fields
2648 field_updates = {}
2649 root_type = type(target)
2650 for field_name in self.parameter_structure.owner_paths:
2651 if '.' not in field_name:
2652 # Check if this field's TYPE is a dataclass (not the instance value)
2653 # We need to check the TYPE because the instance value might be stale
2654 # (e.g., self.parameters['partition_filter_config'] might have partition_filter=2
2655 # even though self.parameters['partition_filter_config.partition_filter'] = None)
2656 field_type = self.parameter_structure.owner_paths.get(field_name)
2657 # CRITICAL FIX: _path_to_type stores CONTAINER type for leaf fields,
2658 # but FIELD type for nested dataclass fields. We must distinguish:
2659 # - If field_type == root_type, it's a leaf field (container type stored)
2660 # - If field_type != root_type AND is_dataclass, it's a nested dataclass
2661 is_nested_dataclass = (
2662 field_type is not None and
2663 is_dataclass(field_type) and
2664 field_type != root_type # Not the container type
2665 )
2666 if is_nested_dataclass:
2667 # Nested dataclass: ALWAYS recursively reconstruct from flat storage
2668 # This ensures we pick up changes to nested fields like
2669 # 'partition_filter_config.partition_filter'
2670 logger.debug(f"🔧 to_object: Reconstructing nested dataclass '{field_name}' from flat storage")
2671 reconstructed = self._reconstruct_from_prefix(field_name)
2672 logger.debug(f"🔧 to_object: Reconstructed '{field_name}' type={type(reconstructed).__name__}")
2673 # Log some field values for debugging
2674 if hasattr(reconstructed, 'enabled'):
2675 logger.debug(f"🔧 to_object: '{field_name}'.enabled = {reconstructed.enabled}")
2676 field_updates[field_name] = reconstructed
2677 else:
2678 # Primitive field: use value directly from parameters
2679 value = self.parameters.get(field_name)
2680 field_updates[field_name] = value
2682 # Reconstruct the target object (either object_instance or delegate)
2683 reconstructed = None
2685 # Python functions can't be copied, but we CAN update their attributes
2686 # This is critical for MRO resolution to see edited config values
2687 if isinstance(target, FunctionType):
2688 for field_name, field_value in field_updates.items():
2689 setattr(target, field_name, field_value)
2690 reconstructed = target
2691 elif is_dataclass(target):
2692 # CRITICAL: Use replace_raw to preserve raw None values!
2693 # dataclasses.replace triggers lazy resolution via __getattribute__,
2694 # which resolves None -> concrete defaults and breaks inheritance.
2695 from objectstate.lazy_factory import replace_raw
2696 reconstructed = replace_raw(target, **field_updates)
2697 else:
2698 # Non-dataclass class instance - shallow copy + setattr
2699 obj_copy = copy.copy(target)
2700 obj_type = type(target)
2701 for field_name, field_value in field_updates.items():
2702 # Skip read-only properties (those without setters)
2703 prop = getattr(obj_type, field_name, None)
2704 if isinstance(prop, property) and prop.fset is None:
2705 continue
2706 setattr(obj_copy, field_name, field_value)
2707 reconstructed = obj_copy
2709 # DELEGATION: If using delegation, update the delegate attribute on object_instance
2710 # as a side effect, but return the reconstructed delegate (not object_instance).
2711 # This ensures callers get the correct type (config, not orchestrator).
2712 # Callers who need the lifecycle object should access state.object_instance directly.
2713 if self._delegate_attr is not None:
2714 if update_delegate:
2715 setattr(self.object_instance, self._delegate_attr, reconstructed)
2716 self._cached_object_applied = True
2717 else:
2718 self._cached_object_applied = False
2719 # Return the reconstructed delegate - this is what the parameters represent
2720 self._cached_object = reconstructed
2721 else:
2722 # NON-DELEGATION: Update object_instance to point to reconstructed object
2723 # This ensures that when to_object() is called (e.g., on window save),
2724 # the ObjectState automatically points to the new instance
2725 if update_delegate:
2726 self.object_instance = reconstructed
2727 self._extraction_target = reconstructed
2728 logger.debug(f"Auto-updated object_instance to new reconstructed object for scope={self.scope_id!r}")
2729 self._cached_object = reconstructed
2730 self._cached_object_applied = True
2732 return self._cached_object
2734 def _reconstruct_from_prefix(self, prefix: str) -> Any:
2735 """Recursively reconstruct dataclass from flat parameters.
2737 Args:
2738 prefix: Current path prefix (e.g., 'partition_filter_config')
2740 Returns:
2741 Reconstructed dataclass instance
2742 """
2743 return self._reconstruct_from_parameter_snapshot(prefix, self.parameters)
2745 def _reconstruct_from_parameter_snapshot(
2746 self,
2747 prefix: str,
2748 parameters: dict[str, Any],
2749 ) -> Any:
2750 """Recursively reconstruct dataclass from a flat parameter snapshot."""
2751 # Determine the type to reconstruct
2752 if not prefix:
2753 # Root level - use extraction target type (handles delegation)
2754 obj_type = type(self._extraction_target)
2755 else:
2756 # Nested level - look up type from _path_to_type
2757 obj_type = self.parameter_structure.owner_paths.get(prefix)
2758 if obj_type is None:
2759 raise ValueError(f"No type mapping for prefix: {prefix}")
2761 prefix_dot = f'{prefix}.' if prefix else ''
2763 # Collect direct fields and nested prefixes
2764 direct_fields = {}
2765 nested_prefixes = set()
2767 for path, value in parameters.items():
2768 if not path.startswith(prefix_dot):
2769 continue
2771 remainder = path[len(prefix_dot):]
2773 if '.' in remainder:
2774 # This is a nested field - collect the first component
2775 first_component = remainder.split('.')[0]
2776 nested_prefixes.add(first_component)
2777 else:
2778 # Direct field of this object
2779 direct_fields[remainder] = value
2781 # Reconstruct nested dataclasses first
2782 for nested_name in nested_prefixes:
2783 nested_path = f'{prefix_dot}{nested_name}'
2784 nested_obj = self._reconstruct_from_parameter_snapshot(
2785 nested_path,
2786 parameters,
2787 )
2788 direct_fields[nested_name] = nested_obj
2790 # CRITICAL: Do NOT filter out None values!
2791 # None can be a semantic value, such as "inherit from parent context".
2792 # When a user explicitly resets a field to None, we MUST pass that None
2793 # to the dataclass constructor so lazy resolution can walk up the MRO.
2794 # Filtering None would cause the dataclass to use its class-level default
2795 # instead of the user's explicit None, breaking inheritance.
2797 # At root level, include constructor params excluded from editing.
2798 if not prefix:
2799 direct_fields.update(self.parameter_structure.excluded_values)
2802 # Instantiate the dataclass with ALL fields including None values
2803 result = obj_type(**direct_fields)
2806 return result
2808 def _get_changed_params_with_types(
2809 self, old_target: Any, new_target: Any
2810 ) -> list[tuple[str, type, str]]:
2811 """
2812 Compare old and new extraction targets to find changed parameters.
2814 Returns a list of tuples: (param_name, container_type, leaf_field_name)
2815 """
2816 changed_params = []
2818 # Get all parameter names from the new extraction target
2819 for param_name in self.parameters.keys():
2820 old_value = self._get_param_value_from_target(old_target, param_name)
2821 new_value = self._get_param_value_from_target(new_target, param_name)
2823 if old_value != new_value:
2824 container_type = self.parameter_structure.owner_paths.get(param_name, type(self.object_instance))
2825 leaf_field_name = DottedFieldPath(param_name).field_name
2826 changed_params.append((param_name, container_type, leaf_field_name))
2828 return changed_params
2830 def _get_param_value_from_target(self, target: Any, param_name: str) -> Any:
2831 """
2832 Get a parameter value from an extraction target by dotted path.
2834 Handles nested dataclass attributes.
2835 """
2836 if target is None:
2837 return None
2839 parts = param_name.split('.')
2840 current = target
2842 for part in parts:
2843 if not hasattr(current, part):
2844 return None
2845 current = getattr(current, part)
2847 return current