Coverage for src/objectstate/object_state_registry.py: 87%
1007 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 07:22 +0000
1"""
2ObjectStateRegistry: Singleton registry of all ObjectState instances.
3Replaces LiveContextService._active_form_managers as the single source of truth.
4"""
6from contextlib import contextmanager
7from contextvars import ContextVar, Token
8from dataclasses import dataclass, is_dataclass, replace
9import logging
10from typing import (
11 Any,
12 Callable,
13 Dict,
14 List,
15 Optional,
16 Set,
17 Tuple,
18 TYPE_CHECKING,
19 Generator,
20 Iterable,
21 TypeAlias,
22)
23import copy
25from objectstate.object_state_metadata import ObjectStateMetadataStore
26from objectstate.construction_binding import StateConstructionBinding
27from objectstate.scope_key import ScopeKey
28from objectstate.parameter_owner import ParameterOwner
29from objectstate.snapshot_model import Snapshot, StateSnapshot, Timeline
30from objectstate.time_travel_profile import TimeTravelProfiler
31from objectstate.transaction_checkpoint import ObjectStateTransactionCheckpoint
33if TYPE_CHECKING:
34 from objectstate.object_state import ObjectState
36TimeTravelStateEntry: TypeAlias = Tuple[str, "ObjectState"]
37TimeTravelCompleteCallback: TypeAlias = Callable[[List[TimeTravelStateEntry], Optional[str]], None]
38RegistryCallback: TypeAlias = Callable[..., None]
39RegistryMutationSignature: TypeAlias = Tuple[
40 Tuple[str, int, Dict[str, Any], Dict[str, Any], StateConstructionBinding], ...
41]
42logger = logging.getLogger(__name__)
45class ObjectStateRegistrySubscription:
46 """Idempotent ownership handle for one registry callback registration."""
48 def __init__(self, release: Callable[[], bool] | None) -> None:
49 self._release = release
51 def release(self) -> bool:
52 """Release this registration once, returning whether it was active."""
54 release = self._release
55 if release is None:
56 return False
57 self._release = None
58 return release()
61@dataclass(frozen=True)
62class TimeTravelTransaction:
63 """Nominal request to move the registry to one snapshot."""
65 snapshot_id: str
66 snapshot: Snapshot
67 source_snapshot_id: Optional[str]
68 snapshot_scopes: Set[str]
69 current_scopes: Set[str]
72@dataclass(frozen=True)
73class TimeTravelScopeReconciliation:
74 """Scope membership changes required before restoring snapshot values."""
76 moved_to_limbo: Set[str]
77 restored_scopes: Set[str]
80@dataclass(frozen=True)
81class PreparedStateRestore:
82 """Prevalidated construction and historical values for one scope."""
84 snapshot: StateSnapshot
85 state: "ObjectState"
86 schema: "ObjectState"
87 diff: "TimeTravelStateDiff"
88 construction_changed: bool
91@dataclass(frozen=True)
92class TimeTravelScopeChange:
93 """Per-scope facts produced by restoring one ObjectState."""
95 changed_paths: Set[str]
96 changed_param_keys: Set[str]
97 meta_changed_keys: Set[str]
98 is_concrete_dirty: bool
99 construction_changed: bool = False
101 @property
102 def needs_navigation(self) -> bool:
103 return bool(self.changed_paths or self.meta_changed_keys or self.construction_changed)
106@dataclass(frozen=True)
107class TimeTravelStateDiff:
108 """Per-state snapshot delta used to keep time-travel restores targeted."""
110 changed_paths: Set[str]
111 changed_param_values: Dict[str, Tuple[Any, Any]]
112 notification_changed_values: Dict[str, Tuple[Any, Any]]
113 meta_changed_values: Dict[str, Tuple[Any, Any]]
114 saved_parameters_changed: bool
115 saved_resolved_changed: bool
116 provenance_changed: bool
118 @property
119 def requires_snapshot_apply(self) -> bool:
120 return bool(
121 self.changed_paths
122 or self.changed_param_values
123 or self.meta_changed_values
124 or self.saved_parameters_changed
125 or self.saved_resolved_changed
126 or self.provenance_changed
127 )
130@dataclass(frozen=True)
131class TimeTravelChangeSet:
132 """Authoritative result of a completed time-travel restore."""
134 triggering_scope: Optional[str]
135 scope_changes: Dict[str, TimeTravelScopeChange]
137 def states_needing_navigation(
138 self, states: dict[str, 'ObjectState']
139 ) -> list[TimeTravelStateEntry]:
140 """Return the registered states this restore changed, as completion callbacks take them."""
141 return [
142 (scope_key, states[scope_key])
143 for scope_key, change in self.scope_changes.items()
144 if change.needs_navigation and scope_key in states
145 ]
148@dataclass(frozen=True)
149class DeferredFieldInvalidation:
150 """One coalescible live-cache invalidation request."""
152 scope_id: str
153 changed_type: ParameterOwner
154 field_name: str
155 invalidate_saved: bool = False
156 include_origin: bool = True
159@dataclass
160class AtomicBlock:
161 """One open ``atomic`` / ``atomic_success`` block: the undo step being coalesced.
163 The block belongs to the context that opened it, exactly as a
164 ``config_context`` layer does: a new thread starts outside every block, and
165 an asyncio task started inside a block inherits it.
166 """
168 label: str
169 triggering_scope: str | None
170 entry_signature: RegistryMutationSignature
171 depth: int = 1
172 snapshot_requested: bool = False
175# Which block, and how many deferrals, the running context is inside. These
176# answer "whose block is this", so they live where ``config_context`` keeps its
177# scope (``contextvars``, see ``context_manager``). What the blocks act on, the
178# registered states, the history and the queue of owed invalidations, is one
179# per process and stays on ``ObjectStateRegistry``.
180_atomic_block: ContextVar[AtomicBlock | None] = ContextVar(
181 "objectstate_atomic_block", default=None
182)
183_live_invalidation_deferrals: ContextVar[int] = ContextVar(
184 "objectstate_live_invalidation_deferrals", default=0
185)
188class ObjectStateRegistry:
189 """Singleton registry of all ObjectState instances.
191 Replaces LiveContextService._active_form_managers as the single source of truth
192 for all configuration state. Keyed by scope_id for efficient lookup.
194 Lifecycle ownership:
195 - PipelineEditor: registers when step added, unregisters when step removed
196 - ImageBrowser: registers when opened, unregisters when closed
197 - Config window: registers PipelineConfig/GlobalPipelineConfig
199 One registry per process, by design: the registered states, the snapshot
200 history, the callbacks and the queue of owed live-cache invalidations are
201 class attributes, so two independent registries cannot exist.
203 Per context (thread or asyncio task), like ``config_context``: the open
204 ``atomic`` / ``atomic_success`` block and the ``defer_live_invalidations``
205 nesting. A thread outside any block never sees another thread's block.
207 Thread safety: the process-wide stores above are not synchronized; callers
208 that mutate the registry from several threads must serialize those calls.
209 """
210 _states: Dict[str, 'ObjectState'] = {} # Keys are always strings (None normalized to "")
212 # Registration lifecycle callbacks - UI subscribes to sync list items with ObjectState lifecycle
213 # Callbacks receive (scope_id: str, object_state: ObjectState)
214 _on_register_callbacks: List[Callable[[str, 'ObjectState'], None]] = []
215 _on_unregister_callbacks: List[Callable[[str, 'ObjectState'], None]] = []
216 _deferred_invalidations: Set[DeferredFieldInvalidation] = set()
217 _snapshot_dirty_scopes: Set[str] = set()
219 # Time-travel completion callbacks - UI subscribes to reopen windows for dirty states
220 # Callbacks receive (dirty_states, triggering_scope) where:
221 # - dirty_states: list of (scope_id, ObjectState) tuples with unsaved changes
222 # - triggering_scope: scope_id that triggered the snapshot (may be None)
223 _on_time_travel_complete_callbacks: List[TimeTravelCompleteCallback] = []
225 # History changed callbacks - fired when history is modified (snapshot added or time-travel)
226 # Used by TimeTravelWidget to stay in sync without polling
227 _on_history_changed_callbacks: List[Callable[[], None]] = []
229 @classmethod
230 def add_register_callback(
231 cls,
232 callback: Callable[[str, 'ObjectState'], None],
233 ) -> ObjectStateRegistrySubscription:
234 """Subscribe to registration events and return its ownership handle."""
235 return cls._subscribe_callback(cls._on_register_callbacks, callback)
237 @classmethod
238 def remove_register_callback(
239 cls,
240 callback: Callable[[str, 'ObjectState'], None],
241 ) -> bool:
242 """Unsubscribe from ObjectState registration events."""
243 return cls._remove_callback(cls._on_register_callbacks, callback)
245 @classmethod
246 def add_unregister_callback(
247 cls,
248 callback: Callable[[str, 'ObjectState'], None],
249 ) -> ObjectStateRegistrySubscription:
250 """Subscribe to unregistration events and return its ownership handle."""
251 return cls._subscribe_callback(cls._on_unregister_callbacks, callback)
253 @classmethod
254 def remove_unregister_callback(
255 cls,
256 callback: Callable[[str, 'ObjectState'], None],
257 ) -> bool:
258 """Unsubscribe from ObjectState unregistration events."""
259 return cls._remove_callback(cls._on_unregister_callbacks, callback)
261 @classmethod
262 def add_time_travel_complete_callback(
263 cls,
264 callback: TimeTravelCompleteCallback,
265 ) -> ObjectStateRegistrySubscription:
266 """Subscribe to time-travel completion events.
268 Callback receives (dirty_states, triggering_scope) where:
269 - dirty_states: list of (scope_id, ObjectState) tuples with unsaved changes
270 - triggering_scope: scope_id that triggered the snapshot (may be None)
271 """
272 return cls._subscribe_callback(
273 cls._on_time_travel_complete_callbacks,
274 callback,
275 )
277 @classmethod
278 def remove_time_travel_complete_callback(
279 cls,
280 callback: TimeTravelCompleteCallback,
281 ) -> bool:
282 """Unsubscribe from time-travel completion events."""
283 return cls._remove_callback(cls._on_time_travel_complete_callbacks, callback)
285 @classmethod
286 def add_history_changed_callback(
287 cls,
288 callback: Callable[[], None],
289 ) -> ObjectStateRegistrySubscription:
290 """Subscribe to history changes and return its ownership handle."""
291 return cls._subscribe_callback(cls._on_history_changed_callbacks, callback)
293 @classmethod
294 def remove_history_changed_callback(cls, callback: Callable[[], None]) -> bool:
295 """Unsubscribe from history change events."""
296 return cls._remove_callback(cls._on_history_changed_callbacks, callback)
298 @classmethod
299 def _subscribe_callback(
300 cls,
301 callbacks: list[RegistryCallback],
302 callback: RegistryCallback,
303 ) -> ObjectStateRegistrySubscription:
304 if callback in callbacks:
305 return ObjectStateRegistrySubscription(None)
306 callbacks.append(callback)
307 return ObjectStateRegistrySubscription(
308 lambda: cls._remove_callback(callbacks, callback)
309 )
311 @staticmethod
312 def _remove_callback(
313 callbacks: list[RegistryCallback],
314 callback: RegistryCallback,
315 ) -> bool:
316 if callback not in callbacks:
317 return False
318 callbacks.remove(callback)
319 return True
321 @classmethod
322 def _fire_history_changed_callbacks(cls) -> None:
323 """Fire all history changed callbacks."""
324 cls._fire_callbacks(cls._on_history_changed_callbacks, "history_changed")
326 @classmethod
327 @contextmanager
328 def defer_live_invalidations(cls) -> Generator[None, None, None]:
329 """Queue descendant live-cache invalidations for an explicit later flush."""
330 token = _live_invalidation_deferrals.set(_live_invalidation_deferrals.get() + 1)
331 try:
332 yield
333 finally:
334 _live_invalidation_deferrals.reset(token)
336 @classmethod
337 def has_deferred_invalidations(cls) -> bool:
338 """Return whether live-cache invalidations are waiting to be flushed."""
339 return bool(cls._deferred_invalidations)
341 @classmethod
342 def flush_deferred_invalidations(cls) -> None:
343 """Apply queued descendant invalidations once, after edit bursts settle."""
344 if not cls._deferred_invalidations:
345 return
347 requests = tuple(cls._deferred_invalidations)
348 cls._deferred_invalidations.clear()
349 token = _live_invalidation_deferrals.set(0)
350 try:
351 for request in requests:
352 cls._invalidate_by_type_and_scope_now(
353 scope_id=request.scope_id,
354 changed_type=request.changed_type,
355 field_name=request.field_name,
356 invalidate_saved=request.invalidate_saved,
357 include_origin=request.include_origin,
358 )
359 finally:
360 _live_invalidation_deferrals.reset(token)
362 @classmethod
363 def _fire_callbacks(cls, callbacks: List[Callable], role: str, *args: Any) -> None:
364 """Fire callback family with uniform error handling."""
365 for callback in callbacks:
366 try:
367 callback(*args)
368 except Exception as e:
369 logger.warning(f"Error in {role} callback: {e}")
371 @classmethod
372 def _fire_register_callbacks(cls, scope_key: str, state: 'ObjectState') -> None:
373 """Fire all registered callbacks for ObjectState registration."""
374 cls._fire_callbacks(cls._on_register_callbacks, "register", scope_key, state)
376 @classmethod
377 def _fire_unregister_callbacks(cls, scope_key: str, state: 'ObjectState') -> None:
378 """Fire all registered callbacks for ObjectState unregistration."""
379 cls._fire_callbacks(cls._on_unregister_callbacks, "unregister", scope_key, state)
381 @classmethod
382 def mark_snapshot_dirty_scope(cls, scope_id: Optional[str]) -> None:
383 """Mark a scope whose materialized state must be copied into the next snapshot."""
384 cls._snapshot_dirty_scopes.add(cls._normalize_scope_id(scope_id))
386 @classmethod
387 def _normalize_scope_id(cls, scope_id: str | None) -> ScopeKey:
388 """Return the registry key: None and "" both represent the global scope."""
389 return ScopeKey(scope_id)
391 @classmethod
392 def register(cls, state: 'ObjectState', _skip_snapshot: bool = False) -> None:
393 """Register an ObjectState in the registry.
395 Args:
396 state: The ObjectState to register.
397 scope_id=None/"" for GlobalPipelineConfig (global scope).
398 scope_id=plate_path for PipelineConfig.
399 scope_id=plate_path::step_N for steps.
400 _skip_snapshot: Internal flag for time-travel (don't record snapshot).
401 """
402 key = cls._normalize_scope_id(state.scope_id)
404 if key in cls._states:
405 logger.warning(f"Overwriting existing ObjectState for scope: {key}")
407 cls._states[key] = state
408 cls.mark_snapshot_dirty_scope(key)
409 logger.debug(f"Registered ObjectState: scope={key}, type={type(state.object_instance).__name__}")
410 cls.resolve_again_below((key,))
412 # Fire callbacks for UI binding
413 cls._fire_register_callbacks(key, state)
415 # Structural registration is not a user-edit boundary. Callers that
416 # create/delete user-visible objects record one explicit operation
417 # snapshot around the mutation.
419 @classmethod
420 def unregister(cls, state: 'ObjectState', _skip_snapshot: bool = False) -> None:
421 """Unregister an ObjectState from the registry.
423 Args:
424 state: The ObjectState to unregister.
425 _skip_snapshot: Internal flag for time-travel (don't record snapshot).
426 """
427 key = cls._normalize_scope_id(state.scope_id)
428 if key in cls._states:
429 removed_state = cls._states.pop(key)
431 # Move to graveyard instead of deleting - enables time travel to past snapshots
432 # that reference this ObjectState. Graveyard persists for session lifetime.
433 cls._graveyard[key] = removed_state
434 logger.debug(f"Unregistered ObjectState: scope={key} (moved to graveyard)")
435 cls.resolve_again_below((key,))
437 # Fire callbacks for UI binding
438 cls._fire_unregister_callbacks(key, state)
440 # Structural unregistration is not a user-edit boundary. The
441 # graveyard preserves state for explicit operation snapshots.
443 @classmethod
444 def global_config_state(cls, config_type: type) -> Optional['ObjectState']:
445 """The state that holds the global config of ``config_type``: the one
446 registered at the global scope, when its object is of exactly that type.
448 ``objectstate.global_config`` reads and replaces the global config through
449 this state while there is one, so registering it is all it takes to make its
450 object the global config, and unregistering it (or time travel to a point
451 where it was not registered) ends that at once.
452 """
453 state = cls._states.get("")
454 if state is not None and type(state._extraction_target) is config_type:
455 return state
456 return None
458 @classmethod
459 def resolve_again_below(cls, changed_scopes: Iterable[ScopeKey]) -> None:
460 """Tell registered states at or below ``changed_scopes`` that what they are
461 resolved from may have changed (a registration, or a stored global config).
463 Each state compares what it is resolved from now with what its resolved values
464 were computed from (``ObjectState.resolve_again_if_sources_changed``), so a
465 state whose ancestors are the same does no work. This call only makes the
466 comparison happen now, so subscribers hear of changed values without waiting
467 for the next read.
468 """
469 changed_scopes = tuple(changed_scopes)
470 for state in tuple(cls._states.values()):
471 if any(scope.contains(state.scope_id) for scope in changed_scopes):
472 state.resolve_again_if_sources_changed()
474 @classmethod
475 def registered_ancestors(
476 cls, scope_id: str | None
477 ) -> tuple[tuple[ScopeKey, 'ObjectState'], ...]:
478 """The states registered at the global scope, each prefix of ``scope_id`` and
479 ``scope_id`` itself, least specific first.
481 Resolved values of a state at ``scope_id`` are computed from these states.
482 """
483 return tuple(
484 (ancestor_key, cls._states[ancestor_key])
485 for ancestor_key in cls._normalize_scope_id(scope_id).ancestors()
486 if ancestor_key in cls._states
487 )
489 @classmethod
490 def unregister_scope_and_descendants(cls, scope_id: Optional[str], _skip_snapshot: bool = False) -> int:
491 """Unregister an ObjectState and all its descendants from the registry.
493 This is used when deleting a plate - we need to cascade delete all child
494 ObjectStates (steps, functions) to prevent memory leaks.
496 Example:
497 When deleting plate "/path/to/plate", this unregisters:
498 - "/path/to/plate" (the plate's PipelineConfig)
499 - "/path/to/plate::step_0" (step ObjectStates)
500 - "/path/to/plate::step_0::func_0" (function ObjectStates)
501 - etc.
503 Args:
504 scope_id: The scope to unregister (along with all descendants).
505 _skip_snapshot: Internal flag for time-travel (don't record snapshot).
507 Returns:
508 Number of ObjectStates unregistered.
509 """
510 scope_key = cls._normalize_scope_id(scope_id)
512 # Find all scopes to delete: exact match + descendants
513 scopes_to_delete = [
514 key
515 for key in cls._states.keys()
516 if key == scope_key or (scope_key and scope_key.contains(key))
517 ]
519 # Delete all matching scopes and fire callbacks
520 # NOTE: Unlike plain unregister(), this is often used for hierarchical deletes.
521 # When history is enabled we must preserve removed states for time-travel (undo/redo),
522 # otherwise snapshots may reference scopes that can no longer be resurrected.
523 removed = [(key, cls._states.pop(key)) for key in scopes_to_delete]
524 cls.resolve_again_below(scopes_to_delete)
525 for key, state in removed:
526 if cls._history_enabled:
527 cls._graveyard[key] = state
528 logger.debug(f"Unregistered ObjectState (cascade): scope={key} (moved to graveyard)")
529 else:
530 logger.debug(f"Unregistered ObjectState (cascade): scope={key}")
531 # Fire callbacks for UI binding
532 cls._fire_unregister_callbacks(key, state)
534 if scopes_to_delete:
535 logger.info(f"Cascade unregistered {len(scopes_to_delete)} ObjectState(s) for scope={scope_key}")
536 # Structural unregister is not a user-edit boundary. User-visible
537 # deletes wrap the full mutation in ObjectStateRegistry.atomic(...).
539 return len(scopes_to_delete)
541 @classmethod
542 def get_by_scope(cls, scope_id: Optional[str]) -> Optional['ObjectState']:
543 """Get ObjectState by scope_id.
545 Args:
546 scope_id: The scope identifier (e.g., "/path::step_0", or None/"" for global scope).
548 Returns:
549 ObjectState if found, None otherwise.
550 """
551 return cls._states.get(cls._normalize_scope_id(scope_id))
553 @classmethod
554 def get_object(cls, scope_id: Optional[str]) -> Optional[Any]:
555 """Get object_instance for a scope, or None if not registered/in limbo.
557 Convenience method for the common pattern of accessing the stored object.
558 When ObjectState uses delegation (via __objectstate_delegate__), this still
559 returns the original object_instance (e.g., orchestrator), not the delegate.
561 Args:
562 scope_id: The scope identifier.
564 Returns:
565 The object_instance if ObjectState is registered, None otherwise.
566 """
567 state = cls.get_by_scope(scope_id)
568 return state.object_instance if state else None
570 @classmethod
571 def get_all(cls) -> List['ObjectState']:
572 """Get all registered ObjectStates.
574 Returns:
575 List of all ObjectState instances. Used by LiveContextService.collect().
576 """
577 return list(cls._states.values())
579 @classmethod
580 def clear(cls) -> None:
581 """Clear all registered state and history. For testing/session reset only."""
582 cls._states.clear()
583 cls._time_travel_limbo.clear()
584 cls._graveyard.clear()
585 cls._deferred_invalidations.clear()
586 _live_invalidation_deferrals.set(0)
587 cls._snapshot_dirty_scopes.clear()
588 cls._snapshots.clear()
589 cls._timelines.clear()
590 cls._current_head = None
591 cls._current_timeline = cls.MAIN_BRANCH
592 _atomic_block.set(None)
593 logger.debug("Cleared ObjectState registry, history, limbo, and graveyard")
595 # ========== TOKEN MANAGEMENT AND CHANGE NOTIFICATION ==========
597 _token: int = 0 # Cache invalidation token
598 _change_callbacks: List[Callable[[], None]] = [] # Change listeners
599 _resolved_changed_callbacks: List[Callable[[str, Set[str]], None]] = []
601 @classmethod
602 def get_token(cls) -> int:
603 """Get current cache invalidation token."""
604 return cls._token
606 @classmethod
607 def increment_token(cls, notify: bool = True) -> None:
608 """Increment cache invalidation token.
610 Args:
611 notify: If True (default), notify listeners of the change.
612 Set to False when you need to invalidate caches but will
613 notify listeners later (e.g., after sibling refresh completes).
614 """
615 cls._token += 1
616 if notify:
617 cls._notify_change()
619 @classmethod
620 def _notify_change(cls) -> None:
621 """Notify all listeners that something changed.
623 UI-agnostic: No PyQt imports. If a callback's object was deleted,
624 the RuntimeError is caught and the callback is removed.
625 """
626 logger.debug(f"🔔 _notify_change: notifying {len(cls._change_callbacks)} listeners")
627 dead_callbacks = []
628 for callback in cls._change_callbacks:
629 try:
630 callback()
631 except RuntimeError as e:
632 # "wrapped C/C++ object has been deleted" - mark for removal
633 if "deleted" in str(e).lower():
634 logger.debug(f" ⚠️ Callback's object was deleted, removing: {e}")
635 dead_callbacks.append(callback)
636 else:
637 logger.warning(f"Change callback failed: {e}")
638 except Exception as e:
639 logger.warning(f"Change callback failed: {e}")
641 # Clean up dead callbacks
642 for cb in dead_callbacks:
643 cls._change_callbacks.remove(cb)
645 @classmethod
646 def connect_listener(
647 cls,
648 callback: Callable[[], None],
649 ) -> ObjectStateRegistrySubscription:
650 """Connect a listener callback that's called on any change.
652 The callback should debounce and call collect() to get fresh values.
653 """
654 subscription = cls._subscribe_callback(cls._change_callbacks, callback)
655 if callback in cls._change_callbacks:
656 logger.debug(f"Connected change listener: {callback}")
657 return subscription
659 @classmethod
660 def disconnect_listener(cls, callback: Callable[[], None]) -> bool:
661 """Disconnect a change listener."""
662 removed = cls._remove_callback(cls._change_callbacks, callback)
663 if removed:
664 logger.debug(f"Disconnected change listener: {callback}")
665 return removed
667 @classmethod
668 def add_resolved_changed_callback(
669 cls,
670 callback: Callable[[str, Set[str]], None],
671 ) -> ObjectStateRegistrySubscription:
672 """Subscribe to resolved changes and return its ownership handle."""
673 return cls._subscribe_callback(cls._resolved_changed_callbacks, callback)
675 @classmethod
676 def remove_resolved_changed_callback(
677 cls,
678 callback: Callable[[str, Set[str]], None],
679 ) -> bool:
680 """Unsubscribe from registry-wide resolved-value changes."""
681 return cls._remove_callback(cls._resolved_changed_callbacks, callback)
683 @classmethod
684 def notify_resolved_changed(
685 cls,
686 scope_id: Optional[str],
687 changed_paths: Set[str],
688 ) -> None:
689 """Publish resolved changes through the registry-level authority."""
690 if not changed_paths:
691 return
692 scope_key = cls._normalize_scope_id(scope_id)
693 cls._fire_callbacks(
694 cls._resolved_changed_callbacks,
695 "resolved_changed",
696 scope_key,
697 set(changed_paths),
698 )
700 # ========== ANCESTOR OBJECT COLLECTION ==========
702 @classmethod
703 def get_ancestor_objects(cls, scope_id: Optional[str], use_saved: bool = False) -> List[Any]:
704 """Get objects from this scope and all ancestors, least→most specific.
706 Replaces LiveContextService.collect() + merge_ancestor_values() for simpler
707 context stack building.
709 Args:
710 scope_id: The scope to get ancestors for (e.g., "/plate::step_0")
711 use_saved: If True, return saved baseline (object_instance) instead of
712 live state (to_object()). Used when computing _saved_resolved
713 to ensure saved baseline only depends on other saved baselines.
715 Returns:
716 List of objects from ancestor scopes, ordered least→most specific.
717 Each object is from state.object_instance (saved) or state.to_object() (live).
718 """
719 return [
720 state.saved_object if use_saved else state.to_object(update_delegate=False)
721 for _, state in cls.registered_ancestors(scope_id)
722 ]
724 @classmethod
725 def get_ancestor_objects_with_scopes(
726 cls,
727 scope_id: Optional[str],
728 use_saved: bool = False,
729 skip_delegate_sync: bool = False,
730 ) -> List[Tuple[str, Any]]:
731 """Get (scope_id, object) tuples from this scope and all ancestors.
733 Similar to get_ancestor_objects() but includes the scope_id for each object.
734 Used for provenance tracking to determine which scope provided a resolved value.
736 ARCHITECTURAL NOTE: This method provides object instances for type structure/MRO resolution,
737 NOT for reading current parameter values. Actual parameter values come from the parameters dict.
738 Using object_instance instead of to_object() avoids:
739 - Expensive reconstruction from parameters
740 - Side effects from delegate auto-sync
741 - Re-entrant calls that invalidate caches mid-computation
743 Args:
744 scope_id: The scope to get ancestors for (e.g., "/plate::step_0")
745 use_saved: If True, use saved baseline. If False, use current live state.
746 skip_delegate_sync: If True, access _extraction_target directly without delegate sync.
747 Use True during cache recomputation to avoid re-entrant invalidation.
748 Use False (default) when you need current values with auto-sync.
750 Returns:
751 List of (scope_id, object) tuples from ancestor scopes, ordered least→most specific.
752 """
753 return [
754 (
755 ancestor_key,
756 state.object_for_resolution(use_saved, sync_delegate=not skip_delegate_sync),
757 )
758 for ancestor_key, state in cls.registered_ancestors(scope_id)
759 ]
761 # ========== SCOPE + TYPE + FIELD AWARE INVALIDATION ==========
763 @classmethod
764 def invalidate_by_type_and_scope(
765 cls,
766 scope_id: Optional[str],
767 changed_type: ParameterOwner,
768 field_name: str,
769 invalidate_saved: bool = False,
770 include_origin: bool = True,
771 ) -> None:
772 """Invalidate a specific field in states that could inherit from changed_type at scope_id.
774 PERFORMANCE: Three-tier filtering:
775 1. SCOPE: State must be at or below changed scope (descendants inherit)
776 2. TYPE: State's type tree must include changed_type
777 3. FIELD: Only invalidate the specific field that changed
779 If changing GlobalPipelineConfig.napari_streaming_config.window_size:
780 - Only states with napari_streaming_config in their tree
781 - Only the 'window_size' field is invalidated, not all 20+ fields
782 - Steps without napari_streaming_config are NOT touched
784 Args:
785 scope_id: The scope that changed (None/"" for global scope)
786 changed_type: The type of the ObjectState that was modified
787 field_name: The specific field that changed
788 invalidate_saved: If True, also invalidate saved_resolved cache for descendants
789 include_origin: Whether to invalidate the state at ``scope_id`` as
790 well as its descendants.
791 """
792 changed_scope = cls._normalize_scope_id(scope_id)
793 if _live_invalidation_deferrals.get() > 0:
794 cls._deferred_invalidations.add(
795 DeferredFieldInvalidation(
796 scope_id=changed_scope,
797 changed_type=changed_type,
798 field_name=field_name,
799 invalidate_saved=invalidate_saved,
800 include_origin=include_origin,
801 )
802 )
803 return
805 cls._invalidate_by_type_and_scope_now(
806 scope_id=changed_scope,
807 changed_type=changed_type,
808 field_name=field_name,
809 invalidate_saved=invalidate_saved,
810 include_origin=include_origin,
811 )
813 @classmethod
814 def _invalidate_by_type_and_scope_now(
815 cls,
816 scope_id: Optional[str],
817 changed_type: ParameterOwner,
818 field_name: str,
819 invalidate_saved: bool = False,
820 include_origin: bool = True,
821 ) -> None:
822 """Immediate implementation for scope/type/field-aware invalidation."""
823 from objectstate.lazy_factory import get_base_type_for_lazy
824 changed_scope = cls._normalize_scope_id(scope_id)
826 # Normalize to base type for comparison (LazyX → X)
827 if isinstance(changed_type, type):
828 base_changed_type = get_base_type_for_lazy(changed_type) or changed_type
829 else:
830 base_changed_type = changed_type
833 for state in cls._states.values():
834 state_scope = cls._normalize_scope_id(state.scope_id)
835 is_self = state_scope == changed_scope
836 if is_self and not include_origin:
837 continue
839 # SCOPE CHECK: must be at or below changed scope
840 if not changed_scope.contains(state_scope):
841 continue
843 # TYPE + FIELD CHECK: find matching nested state and invalidate field
844 cls._invalidate_field_in_matching_states(state, base_changed_type, field_name, invalidate_saved)
846 @classmethod
847 def _invalidate_field_in_matching_states(
848 cls,
849 state: 'ObjectState',
850 target_base_type: ParameterOwner,
851 field_name: str,
852 invalidate_saved: bool = False
853 ) -> None:
854 """Find fields in state that could inherit from target_base_type and invalidate them.
856 With flat storage, we scan _path_to_type to find all dotted paths whose
857 container type matches or inherits from target_base_type.
859 A field should be invalidated if:
860 1. Its container type matches target_base_type exactly, OR
861 2. Its container type inherits from target_base_type (has target_base_type in MRO)
863 This handles sibling inheritance: when PartitionFilterConfig.partition_filter changes,
864 both 'partition_filter_config.partition_filter' and
865 'step_partition_filter_config.partition_filter'
866 are invalidated (since StepPartitionFilterConfig inherits from PartitionFilterConfig).
868 Args:
869 state: ObjectState to check
870 target_base_type: Normalized base type to match
871 field_name: Field to invalidate
872 invalidate_saved: If True, also invalidate saved_resolved cache for this field
873 """
874 invalidated_paths: set[str] = set()
875 _debug = logger.isEnabledFor(logging.DEBUG)
877 # DEBUG: Log entry into type matching
878 if _debug:
879 logger.debug(f"🔍 _invalidate_field_in_matching_states: state_scope={state.scope_id}, target_base_type={target_base_type.__name__}, field_name={field_name}")
880 logger.debug(f"🔍 state.parameter_structure.owner_paths has {len(state.parameter_structure.owner_paths)} entries")
882 for dotted_path in state.inheritance_field_paths(
883 target_base_type,
884 field_name,
885 ):
886 if _debug:
887 logger.debug(f"🔄 INVALIDATING FIELD: {dotted_path} (in parameters)")
888 state.invalidate_field(dotted_path)
889 invalidated_paths.add(dotted_path)
891 if invalidated_paths:
892 if invalidate_saved:
893 # Without recorded sources the saved snapshot is resolved again in full.
894 state._resolved_from.pop(True, None)
895 # Resolve now, so callbacks fire only for values that really changed (no
896 # flash when Reset is clicked on a field that is already reset).
897 changed_paths = state._ensure_live_resolved(notify_flash=False)
898 state._sync_materialized_state(changed_value_fields=changed_paths)
900 # ========== REGISTRY-LEVEL TIME-TRAVEL (DAG Model) ==========
902 # Snapshots: Dict of all snapshots by ID (the DAG - snapshots are NEVER deleted)
903 # Each Snapshot contains id, timestamp, label, triggering_scope, parent_id, all_states
904 # The parent_id forms the DAG edges
905 _snapshots: Dict[str, Snapshot] = {}
907 # Current head: None = at branch head (live), snapshot_id = time-traveling to that snapshot
908 _current_head: Optional[str] = None
910 _history_enabled: bool = True
911 _max_history_size: int = 1000 # Max snapshots in DAG (increased since we don't truncate branches)
912 _in_time_travel: bool = False # Flag for PFM to know to refresh widget values
914 # Limbo: ObjectStates temporarily removed during time-travel
915 # When traveling to a snapshot, ObjectStates not in that snapshot are moved here.
916 # When traveling forward or to head, they're restored from here.
917 _time_travel_limbo: Dict[str, 'ObjectState'] = {}
919 # Graveyard: ObjectStates that were unregistered (deleted) but may be needed for time travel
920 # Unlike limbo (cleared on diverge), graveyard persists for the session lifetime.
921 # This allows time travel to snapshots that reference deleted objects.
922 _graveyard: Dict[str, 'ObjectState'] = {}
924 # Timelines (branches): Named branches that point to a head snapshot.
925 # MAIN_BRANCH is created automatically on the first snapshot.
926 # head_id always points to a valid key in _snapshots (guaranteed by construction)
927 MAIN_BRANCH = "main"
928 _timelines: Dict[str, Timeline] = {}
929 _current_timeline: str = MAIN_BRANCH
931 @classmethod
932 @contextmanager
933 def atomic(cls, label: str, scope_id: Optional[str] = None) -> Generator[None, None, None]:
934 """Context manager for atomic operations that should be a single undo step.
936 All ObjectState changes within this context are coalesced into a single
937 snapshot when the context exits. Nested atomic blocks are supported -
938 only the outermost block records the snapshot.
940 Example:
941 with ObjectStateRegistry.atomic("add step"):
942 # Register step ObjectState
943 ObjectStateRegistry.register(step_state)
944 # Update pipeline's step_scope_ids
945 pipeline_state.update_parameter("step_scope_ids", new_ids)
946 # Register function ObjectState
947 ObjectStateRegistry.register(func_state)
948 # Single snapshot recorded here with label "add step"
950 Args:
951 label: Human-readable label for the coalesced snapshot
952 scope_id: Optional semantic owner of the mutation. When omitted,
953 the first scoped snapshot request inside the atomic block becomes
954 the owner.
955 """
956 block, token = cls._enter_atomic_block(label, scope_id)
957 try:
958 yield
959 finally:
960 if cls._leave_atomic_block(block, token) and cls._atomic_should_record(block):
961 cls.record_snapshot(block.label, block.triggering_scope)
963 @classmethod
964 @contextmanager
965 def atomic_success(
966 cls,
967 label: str,
968 scope_id: Optional[str] = None,
969 ) -> Generator[None, None, None]:
970 """Record one coalesced snapshot only when the atomic body succeeds.
972 This shares the same coalescing state as ``atomic`` so nested ordinary
973 atomic blocks defer into the outer success-only operation. If the body
974 raises, the outermost success-only operation unwinds without recording
975 a partial snapshot.
976 """
977 outermost = cls.current_atomic_block() is None
978 snapshots_before = set(cls._snapshots) if outermost else None
979 timelines_before = copy.deepcopy(cls._timelines) if outermost else None
980 current_head_before = cls._current_head if outermost else None
981 current_timeline_before = cls._current_timeline if outermost else None
982 in_time_travel_before = cls._in_time_travel if outermost else None
983 snapshot_dirty_scopes_before = (
984 set(cls._snapshot_dirty_scopes) if outermost else None
985 )
987 block, token = cls._enter_atomic_block(label, scope_id)
988 succeeded = False
989 try:
990 yield
991 succeeded = True
992 finally:
993 if cls._leave_atomic_block(block, token):
994 if succeeded:
995 if cls._atomic_should_record(block):
996 cls.record_snapshot(block.label, block.triggering_scope)
997 elif snapshots_before is not None and timelines_before is not None:
998 for snapshot_id in set(cls._snapshots) - snapshots_before:
999 del cls._snapshots[snapshot_id]
1000 cls._timelines = timelines_before
1001 cls._current_head = current_head_before
1002 cls._current_timeline = current_timeline_before or cls.MAIN_BRANCH
1003 cls._in_time_travel = bool(in_time_travel_before)
1004 if snapshot_dirty_scopes_before is not None:
1005 cls._snapshot_dirty_scopes = snapshot_dirty_scopes_before
1007 @classmethod
1008 def current_atomic_block(cls) -> AtomicBlock | None:
1009 """Return the block the running context is inside, or None outside any block."""
1010 block = _atomic_block.get()
1011 if block is None or block.depth == 0:
1012 return None
1013 return block
1015 @classmethod
1016 def _enter_atomic_block(
1017 cls, label: str, scope_id: str | None
1018 ) -> tuple[AtomicBlock, Token[AtomicBlock | None] | None]:
1019 """Join the running context's block, opening one when it is outside any."""
1020 block = cls.current_atomic_block()
1021 if block is not None:
1022 block.depth += 1
1023 return block, None
1024 block = AtomicBlock(
1025 label=label,
1026 triggering_scope=scope_id,
1027 entry_signature=cls._registry_mutation_signature(),
1028 )
1029 return block, _atomic_block.set(block)
1031 @classmethod
1032 def _leave_atomic_block(
1033 cls, block: AtomicBlock, token: Token[AtomicBlock | None] | None
1034 ) -> bool:
1035 """Leave ``block``; return whether that closed it, so its step is now due.
1037 A block dropped by ``clear()`` while open is not due: its states are gone.
1038 """
1039 block.depth -= 1
1040 still_current = _atomic_block.get() is block
1041 if token is not None:
1042 _atomic_block.reset(token)
1043 return block.depth == 0 and still_current
1045 @classmethod
1046 def _registry_mutation_signature(
1047 cls,
1048 ) -> RegistryMutationSignature:
1049 """Return the active registry state used to suppress no-op atomics."""
1051 entries: list[Tuple[str, int, Dict[str, Any], Dict[str, Any], StateConstructionBinding]] = (
1052 []
1053 )
1054 for key, state in sorted(cls._states.items()):
1055 parameters, saved_parameters = state.copy_parameter_pair_preserving_callable_identity(
1056 state.parameters,
1057 state._saved_parameters,
1058 )
1059 entries.append(
1060 (key, id(state), parameters, saved_parameters, state.construction_binding())
1061 )
1062 return tuple(entries)
1064 @classmethod
1065 def _atomic_should_record(cls, block: AtomicBlock) -> bool:
1066 """Return whether the closed block produced a user-visible state change."""
1068 if block.snapshot_requested:
1069 return True
1070 return cls._registry_mutation_signature() != block.entry_signature
1072 @classmethod
1073 def ensure_baseline_snapshot(cls, label: str = "init") -> None:
1074 """Create the initial baseline snapshot when history is empty."""
1075 if not cls._history_enabled:
1076 return
1077 if cls._in_time_travel:
1078 return
1079 if cls._snapshots:
1080 return
1081 import time
1083 cls._record_snapshot_internal(label, time.time(), None, prepare_resolved=False)
1085 @classmethod
1086 def get_branch_history(cls, branch_name: Optional[str] = None) -> List[Snapshot]:
1087 """Get ordered history for a branch by walking parent_id chain.
1089 Args:
1090 branch_name: Branch to get history for. If None, uses current branch.
1092 Returns:
1093 List of snapshots from oldest to newest (root first, head last).
1094 Index 0 = oldest (root), index -1 = newest (head).
1095 Empty list if branch has no snapshots yet.
1096 """
1097 if branch_name is None:
1098 branch_name = cls._current_timeline
1100 if branch_name not in cls._timelines:
1101 return []
1103 head_id = cls._timelines[branch_name].head_id
1104 history = []
1105 current_id: Optional[str] = head_id
1107 while current_id is not None:
1108 snapshot = cls._snapshots[current_id] # KeyError if bug
1109 history.append(snapshot)
1110 current_id = snapshot.parent_id
1112 history.reverse() # [oldest, ..., newest] - natural timeline order
1113 return history
1115 @classmethod
1116 def get_current_snapshot_index(cls) -> int:
1117 """Get current position as index into branch history.
1119 Returns:
1120 -1 if at head (live), else index into get_branch_history() list.
1121 Index 0 = oldest, len-1 = head.
1122 """
1123 if cls._current_head is None:
1124 return -1
1126 history = cls.get_branch_history()
1127 for i, snapshot in enumerate(history):
1128 if snapshot.id == cls._current_head:
1129 return i
1131 logger.error(f"⏱️ BRANCH: current_head {cls._current_head} not in branch history")
1132 return -1
1134 @classmethod
1135 def record_snapshot(cls, label: str = "", scope_id: Optional[str] = None) -> None:
1136 """Record current state of ALL ObjectStates to history.
1138 Called automatically after significant state changes.
1139 Each snapshot captures the full system state at a point in time.
1141 Callers that mutate ObjectState values must create the initial baseline
1142 before applying the mutation. This method snapshots the current state
1143 as-is.
1145 Args:
1146 label: Human-readable label (e.g., "edit group_by", "save")
1147 scope_id: Optional scope that triggered the snapshot (for labeling)
1148 """
1149 if not cls._history_enabled:
1150 return
1152 # CRITICAL: Don't record snapshots during time-travel
1153 if cls._in_time_travel:
1154 return
1156 # ATOMIC OPERATIONS: Defer snapshot until atomic block exits
1157 # The atomic() context manager will call record_snapshot() when it completes
1158 block = cls.current_atomic_block()
1159 if block is not None:
1160 block.snapshot_requested = True
1161 if scope_id is not None and block.triggering_scope is None:
1162 block.triggering_scope = scope_id
1163 logger.debug(f"⏱️ ATOMIC: Deferring snapshot '{label}' (depth={block.depth})")
1164 return
1166 import time
1168 full_label = f"{label} [{scope_id}]" if scope_id else label
1169 cls._record_snapshot_internal(full_label, time.time(), scope_id)
1171 @classmethod
1172 def _record_snapshot_internal(
1173 cls,
1174 label: str,
1175 timestamp: float,
1176 triggering_scope: Optional[str],
1177 *,
1178 prepare_resolved: bool = True,
1179 ) -> None:
1180 """Internal method to record a snapshot without baseline logic.
1182 DAG Model:
1183 - Snapshots are added to _snapshots dict (never deleted)
1184 - If _current_head is not None (time-traveling), we're diverging
1185 - On diverge: create auto-branch for old future, new snapshot parents from _current_head
1186 - Branch head_id is updated to point to new snapshot
1187 """
1188 import uuid
1190 if prepare_resolved:
1191 if cls.has_deferred_invalidations():
1192 cls.flush_deferred_invalidations()
1193 cls._prepare_states_for_snapshot()
1195 # Determine parent_id for new snapshot
1196 if cls._current_head is not None:
1197 # We're in the past - diverging from _current_head
1198 parent_id = cls._current_head
1200 # Auto-branch: preserve old future before we diverge
1201 # The old branch head becomes an auto-saved branch
1202 if cls._current_timeline in cls._timelines:
1203 old_head_id = cls._timelines[cls._current_timeline].head_id
1204 if old_head_id != cls._current_head:
1205 # There IS an old future to preserve
1206 branch_name = f"auto-{old_head_id[:8]}"
1207 if branch_name not in cls._timelines:
1208 cls._timelines[branch_name] = Timeline(
1209 name=branch_name,
1210 head_id=old_head_id,
1211 base_id=cls._current_head,
1212 description=f"Auto-saved from diverge at {cls._current_head[:8]}",
1213 )
1214 old_snapshot = cls._snapshots[old_head_id]
1215 logger.info(f"⏱️ AUTO-BRANCH: Created '{branch_name}' (was {old_snapshot.label})")
1217 # Move limbo contents to graveyard before clearing - enables time travel
1218 # to any snapshot that references these states (even on other branches)
1219 cls._graveyard.update(cls._time_travel_limbo)
1220 cls._time_travel_limbo.clear()
1221 else:
1222 # At branch head - parent is current branch's head (if exists)
1223 if cls._current_timeline in cls._timelines:
1224 parent_id = cls._timelines[cls._current_timeline].head_id
1225 else:
1226 parent_id = None
1228 parent_snapshot = cls._snapshots.get(parent_id) if parent_id is not None else None
1230 # Capture registered ObjectStates as StateSnapshot dataclasses.
1231 # Unchanged scopes reuse the parent snapshot's immutable StateSnapshot;
1232 # dirty/new scopes are copied. This keeps snapshots semantically
1233 # complete without revalidating and comparing every open step/config on
1234 # every edit.
1235 current_state_keys = set(cls._states)
1236 if parent_snapshot is None:
1237 all_states: Dict[str, StateSnapshot] = {}
1238 snapshot_candidate_keys = current_state_keys
1239 else:
1240 all_states = {
1241 key: state_snapshot
1242 for key, state_snapshot in parent_snapshot.all_states.items()
1243 if key in current_state_keys
1244 }
1245 snapshot_candidate_keys = (
1246 cls._snapshot_dirty_scopes & current_state_keys
1247 ) | (current_state_keys - set(parent_snapshot.all_states))
1249 for state in cls._states.values():
1250 state.validate_metadata_contracts("snapshot")
1251 capture_values = {
1252 key: StateSnapshot(
1253 saved_resolved=state._saved_resolved,
1254 live_resolved=state._live_resolved,
1255 parameters=state.parameters,
1256 saved_parameters=state._saved_parameters,
1257 provenance=state._live_provenance,
1258 construction=state.construction_binding(),
1259 meta=state.copy_metadata_for_snapshot(),
1260 )
1261 for key, state in cls._states.items()
1262 }
1263 # A changed scope also invalidates immutable snapshots that share a
1264 # copied object with it. Independent clean scopes retain structural
1265 # sharing; connected scopes are copied together with one graph memo.
1266 alias_scopes: Dict[int, Set[str]] = {}
1267 for key, values in capture_values.items():
1268 memo: Dict[int, Any] = {}
1269 copy.deepcopy(values, memo)
1270 for identity in memo:
1271 if identity not in (id(memo), id(values), id(values.construction)):
1272 alias_scopes.setdefault(identity, set()).add(key)
1273 pending = list(snapshot_candidate_keys)
1274 while pending:
1275 key = pending.pop()
1276 for scopes in alias_scopes.values():
1277 if key in scopes:
1278 additional = scopes - snapshot_candidate_keys
1279 snapshot_candidate_keys.update(additional)
1280 pending.extend(additional)
1281 captured = copy.deepcopy({key: capture_values[key] for key in snapshot_candidate_keys})
1283 with TimeTravelProfiler.phase(
1284 "objectstate.snapshot_capture_states",
1285 states=len(cls._states),
1286 has_parent=parent_snapshot is not None,
1287 candidates=len(snapshot_candidate_keys),
1288 ):
1289 for key in snapshot_candidate_keys:
1290 state = cls._states[key]
1291 state.validate_metadata_contracts("snapshot")
1292 all_states[key] = captured[key]
1293 cls._snapshot_dirty_scopes.clear()
1295 # Create new snapshot
1296 snapshot = Snapshot(
1297 id=str(uuid.uuid4()),
1298 timestamp=timestamp,
1299 label=label,
1300 triggering_scope=triggering_scope,
1301 parent_id=parent_id,
1302 all_states=all_states,
1303 )
1305 # Add to DAG
1306 cls._snapshots[snapshot.id] = snapshot
1308 # Update or create current branch to point to new snapshot
1309 if cls._current_timeline not in cls._timelines:
1310 cls._timelines[cls._current_timeline] = Timeline(
1311 name=cls._current_timeline,
1312 head_id=snapshot.id,
1313 base_id=snapshot.id,
1314 description="Main timeline" if cls._current_timeline == cls.MAIN_BRANCH else "",
1315 )
1316 logger.info(f"⏱️ BRANCH: Created '{cls._current_timeline}' timeline")
1317 else:
1318 cls._timelines[cls._current_timeline].head_id = snapshot.id
1320 # Return to head (live state) - we're no longer time-traveling
1321 cls._current_head = None
1323 # Enforce max DAG size (prune oldest snapshots not referenced by any branch)
1324 if len(cls._snapshots) > cls._max_history_size:
1325 cls._prune_unreachable_snapshots()
1327 logger.debug(f"⏱️ SNAPSHOT: Recorded '{label}' (id={snapshot.id[:8]})")
1328 cls._fire_history_changed_callbacks()
1330 @staticmethod
1331 def _copy_snapshot_mapping_with_parent_sharing(
1332 current: Dict[str, Any],
1333 parent: Optional[Dict[str, Any]],
1334 ) -> Dict[str, Any]:
1335 """Copy snapshot values while reusing unchanged parent snapshot entries."""
1336 if parent is None:
1337 return copy.deepcopy(current)
1339 snapshot: Dict[str, Any] = {}
1340 for key, value in current.items():
1341 if key in parent and parent[key] == value:
1342 snapshot[key] = parent[key]
1343 else:
1344 snapshot[key] = copy.deepcopy(value)
1345 return snapshot
1347 @classmethod
1348 def _prepare_states_for_snapshot(cls) -> None:
1349 """Synchronize resolved caches before immutable snapshot capture."""
1351 for _key, state in cls._states_by_dependency_order():
1352 state.prepare_for_snapshot_capture()
1354 @classmethod
1355 def _states_by_dependency_order(cls) -> tuple[tuple[str, 'ObjectState'], ...]:
1356 """Return states ordered so ancestors resolve before descendants."""
1358 return tuple(
1359 sorted(
1360 cls._states.items(),
1361 key=lambda item: (item[0].depth, item[0]),
1362 )
1363 )
1365 @classmethod
1366 def _prune_unreachable_snapshots(cls) -> None:
1367 """Remove snapshots not reachable from any branch head.
1369 Walks from each branch head to find all reachable snapshots,
1370 then removes any that aren't reachable.
1371 """
1372 reachable: Set[str] = set()
1374 for timeline in cls._timelines.values():
1375 current_id: Optional[str] = timeline.head_id
1376 while current_id is not None and current_id not in reachable:
1377 reachable.add(current_id)
1378 snapshot = cls._snapshots.get(current_id)
1379 current_id = snapshot.parent_id if snapshot else None
1381 # Remove unreachable
1382 unreachable = set(cls._snapshots.keys()) - reachable
1383 for snapshot_id in unreachable:
1384 del cls._snapshots[snapshot_id]
1386 if unreachable:
1387 logger.debug(f"⏱️ PRUNE: Removed {len(unreachable)} unreachable snapshots")
1389 @classmethod
1390 def _build_time_travel_transaction(cls, snapshot_id: str) -> TimeTravelTransaction | None:
1391 if snapshot_id not in cls._snapshots:
1392 logger.error(f"⏱️ TIME_TRAVEL: Snapshot {snapshot_id} not found")
1393 return None
1395 snapshot = cls._snapshots[snapshot_id]
1396 return TimeTravelTransaction(
1397 snapshot_id=snapshot_id,
1398 snapshot=snapshot,
1399 source_snapshot_id=cls._current_snapshot_id(),
1400 snapshot_scopes=set(snapshot.all_states.keys()),
1401 current_scopes=set(cls._states.keys()),
1402 )
1404 @classmethod
1405 def _current_snapshot_id(cls) -> Optional[str]:
1406 """Return the snapshot id that currently backs the live registry state."""
1408 if cls._current_head is not None:
1409 return cls._current_head
1410 timeline = cls._timelines.get(cls._current_timeline)
1411 if timeline is None:
1412 return None
1413 return timeline.head_id
1415 @classmethod
1416 def _navigation_triggering_scope(
1417 cls,
1418 transaction: TimeTravelTransaction,
1419 ) -> Optional[str]:
1420 """Return the semantic mutation owner for the time-travel transition.
1422 A snapshot's ``triggering_scope`` describes the operation that produced
1423 that snapshot from its parent. When moving backward, the operation being
1424 undone is the source snapshot, not the target snapshot. When moving
1425 forward, the operation being redone is the target snapshot.
1426 """
1428 source_snapshot = (
1429 cls._snapshots.get(transaction.source_snapshot_id)
1430 if transaction.source_snapshot_id is not None
1431 else None
1432 )
1433 target_snapshot = transaction.snapshot
1435 if source_snapshot is not None:
1436 if source_snapshot.parent_id == target_snapshot.id:
1437 return source_snapshot.triggering_scope
1438 if target_snapshot.parent_id == source_snapshot.id:
1439 return target_snapshot.triggering_scope
1441 return target_snapshot.triggering_scope or (
1442 source_snapshot.triggering_scope
1443 if source_snapshot is not None
1444 else None
1445 )
1447 @classmethod
1448 def _reconcile_time_travel_scopes(
1449 cls,
1450 transaction: TimeTravelTransaction,
1451 prepared: Dict[str, PreparedStateRestore],
1452 ) -> TimeTravelScopeReconciliation:
1453 moved_to_limbo: Set[str] = set()
1454 restored_scopes: Set[str] = set()
1456 for scope_key in transaction.current_scopes - transaction.snapshot_scopes:
1457 state = cls._states.pop(scope_key)
1458 cls._time_travel_limbo[scope_key] = state
1459 moved_to_limbo.add(scope_key)
1460 logger.debug(f"⏱️ TIME_TRAVEL: Moved to limbo: {scope_key}")
1462 for scope_key in transaction.snapshot_scopes - transaction.current_scopes:
1463 state = prepared[scope_key].state
1464 cls._time_travel_limbo.pop(scope_key, None)
1465 cls._graveyard.pop(scope_key, None)
1466 cls._states[scope_key] = state
1467 restored_scopes.add(scope_key)
1469 return TimeTravelScopeReconciliation(
1470 moved_to_limbo=moved_to_limbo,
1471 restored_scopes=restored_scopes,
1472 )
1474 @classmethod
1475 def _prepare_time_travel_states(cls, snapshot: Snapshot) -> Dict[str, PreparedStateRestore]:
1476 """Validate and construct the entire scope graph before publishing it."""
1478 values = copy.deepcopy(snapshot.all_states)
1479 prepared: Dict[str, PreparedStateRestore] = {}
1480 pending = dict(values)
1481 while pending:
1482 progressed = False
1483 for scope, state_snapshot in list(pending.items()):
1484 binding = state_snapshot.construction
1485 if not isinstance(binding, StateConstructionBinding):
1486 raise ValueError(f"Missing typed construction binding at {scope!r}.")
1487 parent_scope = binding.parent_scope
1488 if parent_scope is not None and parent_scope not in values:
1489 raise ValueError(f"Missing historical parent {parent_scope!r} for {scope!r}.")
1490 if parent_scope is not None and parent_scope not in prepared:
1491 continue
1492 state = (
1493 cls._states.get(scope)
1494 or cls._time_travel_limbo.get(scope)
1495 or cls._graveyard.get(scope)
1496 )
1497 parent = prepared[parent_scope].schema if parent_scope is not None else None
1498 schema = binding.prepare(scope, state, parent)
1499 schema.metadata = ObjectStateMetadataStore.from_snapshot(
1500 scope_id=scope, metadata=state_snapshot.meta
1501 )
1502 schema.validate_metadata_contracts("time_travel")
1503 state_snapshot = cls._reconcile_snapshot_schema(
1504 schema,
1505 state_snapshot,
1506 )
1507 cls._apply_time_travel_snapshot_state(schema, state_snapshot, materialize=False)
1508 binding.validate_scope_membership(schema, values.keys())
1509 schema._raw_dirty = schema.parameters != schema._saved_parameters
1510 schema._dirty_fields = schema._compute_dirty_fields()
1511 schema._signature_diff_fields = schema._compute_signature_diff_fields()
1512 prepared[scope] = PreparedStateRestore(
1513 snapshot=state_snapshot,
1514 state=state if state is not None else schema,
1515 schema=schema,
1516 diff=cls._time_travel_state_diff(
1517 state if state is not None else schema, state_snapshot
1518 ),
1519 construction_changed=(
1520 state is not None and state.construction_binding() != binding
1521 ),
1522 )
1523 del pending[scope]
1524 progressed = True
1525 if not progressed:
1526 raise ValueError(f"Cyclic historical ObjectState parents: {sorted(pending)!r}.")
1527 return prepared
1529 @staticmethod
1530 def _reconcile_snapshot_schema(
1531 schema: "ObjectState",
1532 snapshot: StateSnapshot,
1533 ) -> StateSnapshot:
1534 """Project historical values onto the declaration's current field schema.
1536 Construction owns the valid parameter surface. Historical snapshots own
1537 values only for fields still declared on that surface. This gives newly
1538 declared fields their construction-derived defaults and prevents removed
1539 fields from surviving as detached historical parameters.
1540 """
1542 def reconcile(current: dict, historical: dict) -> dict:
1543 return {
1544 key: historical[key] if key in historical else value
1545 for key, value in current.items()
1546 }
1548 return replace(
1549 snapshot,
1550 parameters=reconcile(schema.parameters, snapshot.parameters),
1551 saved_parameters=reconcile(
1552 schema._saved_parameters,
1553 snapshot.saved_parameters,
1554 ),
1555 saved_resolved=reconcile(
1556 schema._saved_resolved,
1557 snapshot.saved_resolved,
1558 ),
1559 live_resolved=reconcile(
1560 schema._live_resolved,
1561 snapshot.live_resolved,
1562 ),
1563 provenance=reconcile(
1564 schema._live_provenance,
1565 snapshot.provenance,
1566 ),
1567 )
1569 @classmethod
1570 def _restore_time_travel_states(
1571 cls,
1572 transaction: TimeTravelTransaction,
1573 reconciliation: TimeTravelScopeReconciliation,
1574 prepared: Dict[str, PreparedStateRestore],
1575 ) -> TimeTravelChangeSet:
1576 scope_changes: Dict[str, TimeTravelScopeChange] = {}
1578 for scope_key, restoration in prepared.items():
1579 state = restoration.state
1580 state_snap = restoration.snapshot
1581 binding = state_snap.construction
1582 parent = cls._states[binding.parent_scope] if binding.parent_scope is not None else None
1583 state.apply_prepared_construction(binding, restoration.schema, parent)
1585 with TimeTravelProfiler.phase("objectstate.restore_scope", scope=scope_key):
1586 scope_changes[scope_key] = cls._restore_time_travel_state(
1587 scope_key=scope_key,
1588 state=state,
1589 state_snap=state_snap,
1590 was_restored_from_limbo=scope_key in reconciliation.restored_scopes,
1591 construction_changed=restoration.construction_changed,
1592 diff=restoration.diff,
1593 )
1595 return TimeTravelChangeSet(
1596 triggering_scope=cls._navigation_triggering_scope(transaction),
1597 scope_changes=scope_changes,
1598 )
1600 @classmethod
1601 def _restore_time_travel_state(
1602 cls,
1603 *,
1604 scope_key: str,
1605 state: "ObjectState",
1606 state_snap: StateSnapshot,
1607 was_restored_from_limbo: bool,
1608 diff: TimeTravelStateDiff,
1609 construction_changed: bool = False,
1610 ) -> TimeTravelScopeChange:
1611 with TimeTravelProfiler.phase(
1612 "objectstate.scope_navigation_meta",
1613 scope=scope_key,
1614 changed_paths=len(diff.changed_paths),
1615 ):
1616 cls._apply_time_travel_navigation_state(
1617 scope_key=scope_key,
1618 state=state,
1619 changed_values=cls._time_travel_navigation_changed_values(
1620 concrete_changed_values=diff.changed_param_values,
1621 notification_changed_values=diff.notification_changed_values,
1622 ),
1623 )
1624 cls._log_time_travel_param_changes(
1625 scope_key=scope_key,
1626 changed_param_values=diff.changed_param_values,
1627 was_restored_from_limbo=was_restored_from_limbo,
1628 )
1629 cls._apply_time_travel_meta_state(
1630 scope_key=scope_key,
1631 state=state,
1632 meta_changed_values=diff.meta_changed_values,
1633 )
1635 with TimeTravelProfiler.phase(
1636 "objectstate.scope_apply_snapshot",
1637 scope=scope_key,
1638 changed_paths=len(diff.changed_paths),
1639 ):
1640 cls._apply_time_travel_snapshot_state(
1641 state, state_snap,
1642 materialize=(diff.requires_snapshot_apply or was_restored_from_limbo or construction_changed),
1643 )
1644 is_concrete_dirty = state.parameters != state._saved_parameters
1645 if is_concrete_dirty:
1646 cls._log_time_travel_concrete_dirty(scope_key, state)
1648 return TimeTravelScopeChange(
1649 changed_paths=diff.changed_paths,
1650 changed_param_keys=set(diff.changed_param_values.keys()),
1651 meta_changed_keys=set(diff.meta_changed_values.keys()),
1652 is_concrete_dirty=is_concrete_dirty,
1653 construction_changed=construction_changed,
1654 )
1656 @classmethod
1657 def _time_travel_state_diff(
1658 cls,
1659 state: 'ObjectState',
1660 state_snap: StateSnapshot,
1661 ) -> TimeTravelStateDiff:
1662 current_params = state.parameters.copy() if state.parameters else {}
1663 current_live = state._live_resolved.copy() if state._live_resolved else {}
1665 live_changed_paths = cls._time_travel_changed_paths(
1666 current_live=current_live,
1667 target_live=state_snap.live_resolved,
1668 )
1669 changed_param_values = cls._time_travel_changed_param_values(
1670 current_params=current_params,
1671 target_params=state_snap.parameters,
1672 )
1673 notification_changed_values = cls._time_travel_changed_live_values(
1674 current_live=current_live,
1675 target_live=state_snap.live_resolved,
1676 changed_paths=live_changed_paths,
1677 )
1678 notification_changed_values.update(changed_param_values)
1679 changed_paths = state.notification_fields_for_value_changes(
1680 notification_changed_values
1681 )
1683 return TimeTravelStateDiff(
1684 changed_paths=changed_paths,
1685 changed_param_values=changed_param_values,
1686 notification_changed_values=notification_changed_values,
1687 meta_changed_values=cls._time_travel_changed_meta_values(
1688 current_meta=state.metadata,
1689 target_meta=state_snap.meta,
1690 ),
1691 saved_parameters_changed=state._saved_parameters
1692 != state_snap.saved_parameters,
1693 saved_resolved_changed=state._saved_resolved
1694 != state_snap.saved_resolved,
1695 provenance_changed=state._live_provenance != state_snap.provenance,
1696 )
1698 @staticmethod
1699 def _time_travel_changed_paths(
1700 *,
1701 current_live: Dict,
1702 target_live: Dict,
1703 ) -> Set[str]:
1704 return {
1705 key
1706 for key in set(target_live.keys()) | set(current_live.keys())
1707 if target_live.get(key) != current_live.get(key)
1708 }
1710 @staticmethod
1711 def _time_travel_changed_param_values(
1712 *,
1713 current_params: Dict,
1714 target_params: Dict,
1715 ) -> Dict[str, Tuple[Any, Any]]:
1716 changed: Dict[str, Tuple[Any, Any]] = {}
1717 for param_key in set(target_params.keys()) | set(current_params.keys()):
1718 before = current_params.get(param_key)
1719 after = target_params.get(param_key)
1720 if is_dataclass(before) or is_dataclass(after):
1721 continue
1722 if before != after:
1723 changed[param_key] = (before, after)
1724 return changed
1726 @staticmethod
1727 def _time_travel_changed_live_values(
1728 *,
1729 current_live: Dict,
1730 target_live: Dict,
1731 changed_paths: Set[str],
1732 ) -> Dict[str, Tuple[Any, Any]]:
1733 return {
1734 path: (
1735 current_live.get(path),
1736 target_live.get(path),
1737 )
1738 for path in changed_paths
1739 }
1741 @classmethod
1742 def _time_travel_navigation_changed_values(
1743 cls,
1744 *,
1745 concrete_changed_values: Dict[str, Tuple[Any, Any]],
1746 notification_changed_values: Dict[str, Tuple[Any, Any]],
1747 ) -> Dict[str, Tuple[Any, Any]]:
1748 """Return concrete changed paths plus their structural descendants for navigation."""
1750 if not concrete_changed_values:
1751 return {}
1752 from objectstate.field_access import DottedFieldPath
1754 concrete_paths = tuple(
1755 DottedFieldPath(path)
1756 for path in concrete_changed_values
1757 )
1758 return {
1759 path: values
1760 for path, values in notification_changed_values.items()
1761 if any(
1762 concrete_path.contains_path(path)
1763 for concrete_path in concrete_paths
1764 )
1765 }
1767 @classmethod
1768 def _apply_time_travel_navigation_state(
1769 cls,
1770 *,
1771 scope_key: str,
1772 state: 'ObjectState',
1773 changed_values: Dict[str, Tuple[Any, Any]],
1774 ) -> None:
1775 if not changed_values:
1776 state._last_changed_field = None
1777 state._last_changed_concrete_paths = set()
1778 state._last_changed_paths = set()
1779 state._last_changed_values = {}
1780 logger.debug(f"⏱️ LAST_CHANGED_FIELD: {scope_key} field=None (no value changes)")
1781 return
1783 state._set_last_changed_values(changed_values)
1784 logger.debug(
1785 "⏱️ LAST_CHANGED_FIELD: %s field=%s total_changes=%d",
1786 scope_key,
1787 state._last_changed_field,
1788 len(changed_values),
1789 )
1791 @staticmethod
1792 def _log_time_travel_param_changes(
1793 *,
1794 scope_key: str,
1795 changed_param_values: Dict[str, Tuple[Any, Any]],
1796 was_restored_from_limbo: bool,
1797 ) -> None:
1798 if was_restored_from_limbo:
1799 return
1800 for param_key, (before, after) in changed_param_values.items():
1801 logger.debug(
1802 "⏱️ PARAM_CHANGE: %s param=%s before=%r after=%r",
1803 scope_key,
1804 param_key,
1805 before,
1806 after,
1807 )
1809 @staticmethod
1810 def _time_travel_changed_meta_values(
1811 *,
1812 current_meta: Dict,
1813 target_meta: Dict,
1814 ) -> Dict[str, Tuple[Any, Any]]:
1815 changed: Dict[str, Tuple[Any, Any]] = {}
1816 for meta_key in set(current_meta.keys()) | set(target_meta.keys()):
1817 before = current_meta.get(meta_key)
1818 after = target_meta.get(meta_key)
1819 if before != after:
1820 changed[meta_key] = (before, after)
1821 return changed
1823 @staticmethod
1824 def _apply_time_travel_meta_state(
1825 *,
1826 scope_key: str,
1827 state: 'ObjectState',
1828 meta_changed_values: Dict[str, Tuple[Any, Any]],
1829 ) -> None:
1830 state._last_changed_meta_keys = set(meta_changed_values.keys())
1831 state._last_changed_meta_values = meta_changed_values
1832 if meta_changed_values:
1833 logger.debug("⏱️ META_CHANGE: %s keys=%s", scope_key, sorted(meta_changed_values))
1835 @staticmethod
1836 def _apply_time_travel_snapshot_state(
1837 state: "ObjectState",
1838 state_snap: StateSnapshot,
1839 *,
1840 materialize: bool = True,
1841 ) -> None:
1842 # The complete historical graph was copied once during preparation.
1843 state._saved_resolved = state_snap.saved_resolved
1844 state._live_resolved = state_snap.live_resolved
1845 state._live_provenance = state_snap.provenance
1846 state.parameters = state_snap.parameters
1847 state._index_parameter_paths()
1848 state._saved_parameters = state_snap.saved_parameters
1849 state._cached_object = None
1850 state._cached_object_applied = False
1851 state.metadata = ObjectStateMetadataStore.from_snapshot(
1852 scope_id=state.scope_id,
1853 metadata=state_snap.meta,
1854 )
1855 if materialize:
1856 state._sync_materialized_state(emit_notifications=False)
1858 @staticmethod
1859 def _log_time_travel_concrete_dirty(scope_key: str, state: "ObjectState") -> None:
1860 for param_key in set(state.parameters.keys()) | set(state._saved_parameters.keys()):
1861 parameter_value = state.parameters.get(param_key)
1862 saved_value = state._saved_parameters.get(param_key)
1863 if parameter_value != saved_value:
1864 logger.debug(
1865 "⏱️ CONCRETE_DIRTY: %s param=%s params=%r saved_params=%r",
1866 scope_key,
1867 param_key,
1868 parameter_value,
1869 saved_value,
1870 )
1872 @classmethod
1873 def _notify_time_travel_state_callbacks(cls, change_set: TimeTravelChangeSet) -> None:
1874 for scope_key, change in change_set.scope_changes.items():
1875 state = cls._states.get(scope_key)
1876 if state is None:
1877 continue
1878 if change.changed_paths:
1879 with TimeTravelProfiler.phase(
1880 "objectstate.notify_resolved",
1881 scope=scope_key,
1882 changed_paths=len(change.changed_paths),
1883 ):
1884 state._notify_resolved_changed(
1885 change.changed_paths,
1886 context="time_travel",
1887 )
1889 @classmethod
1890 def _notify_time_travel_materialized_callbacks(
1891 cls,
1892 change_set: TimeTravelChangeSet,
1893 ) -> None:
1894 for scope_key, change in change_set.scope_changes.items():
1895 state = cls._states.get(scope_key)
1896 if state is None:
1897 continue
1898 if not (
1899 change.changed_paths
1900 or change.changed_param_keys
1901 or change.meta_changed_keys
1902 or change.construction_changed
1903 ):
1904 continue
1905 with TimeTravelProfiler.phase(
1906 "objectstate.notify_state_callback",
1907 scope=scope_key,
1908 changed_paths=len(change.changed_paths),
1909 changed_params=len(change.changed_param_keys),
1910 meta_changes=len(change.meta_changed_keys),
1911 ):
1912 state._notify_state_changed(
1913 change.changed_paths | change.changed_param_keys
1914 )
1916 @classmethod
1917 def _fire_time_travel_complete_callbacks(
1918 cls,
1919 change_set: TimeTravelChangeSet,
1920 ) -> None:
1921 if not cls._on_time_travel_complete_callbacks:
1922 return
1924 states_with_changes = change_set.states_needing_navigation(cls._states)
1925 logger.debug(
1926 "⏱️ TIME_TRAVEL: Firing %d callback(s) with %d changed state(s)",
1927 len(cls._on_time_travel_complete_callbacks),
1928 len(states_with_changes),
1929 )
1930 for callback in cls._on_time_travel_complete_callbacks:
1931 with TimeTravelProfiler.phase(
1932 "objectstate.complete_callback",
1933 callback=f"{type(callback).__module__}.{type(callback).__qualname__}",
1934 changed_states=len(states_with_changes),
1935 ):
1936 callback(states_with_changes, change_set.triggering_scope)
1938 @classmethod
1939 def time_travel_to_snapshot(cls, snapshot_id: str) -> bool:
1940 """Travel ALL ObjectStates to a specific snapshot by ID.
1942 Full time-travel: ObjectStates are registered/unregistered to match the snapshot.
1943 ObjectStates not in the snapshot are moved to limbo. ObjectStates in snapshot
1944 but not in registry are restored from limbo.
1946 Args:
1947 snapshot_id: UUID of snapshot to travel to
1949 Returns:
1950 True if travel succeeded.
1951 """
1952 transaction = cls._build_time_travel_transaction(snapshot_id)
1953 if transaction is None:
1954 return False
1956 prepared = cls._prepare_time_travel_states(transaction.snapshot)
1957 states_before = dict(cls._states)
1958 limbo_before = dict(cls._time_travel_limbo)
1959 graveyard_before = dict(cls._graveyard)
1960 head_before = cls._current_head
1961 dirty_before = set(cls._snapshot_dirty_scopes)
1962 checkpoints = {
1963 scope: ObjectStateTransactionCheckpoint.capture(
1964 restoration.state, preserve_value_identity=True
1965 )
1966 for scope, restoration in prepared.items()
1967 }
1968 cls._current_head = snapshot_id
1969 cls._in_time_travel = True
1970 try:
1971 with TimeTravelProfiler.phase(
1972 "objectstate.reconcile_scopes",
1973 snapshot=snapshot_id[:8],
1974 ):
1975 reconciliation = cls._reconcile_time_travel_scopes(transaction, prepared)
1976 with TimeTravelProfiler.phase(
1977 "objectstate.restore_states",
1978 snapshot=snapshot_id[:8],
1979 states=len(transaction.snapshot.all_states),
1980 ):
1981 change_set = cls._restore_time_travel_states(
1982 transaction,
1983 reconciliation,
1984 prepared,
1985 )
1986 except Exception:
1987 try:
1988 for scope, restoration in reversed(list(prepared.items())):
1989 checkpoints[scope].restore(restoration.state, emit_notifications=False)
1990 finally:
1991 cls._states = states_before
1992 cls._time_travel_limbo = limbo_before
1993 cls._graveyard = graveyard_before
1994 cls._current_head = head_before
1995 cls._snapshot_dirty_scopes = dirty_before
1996 raise
1997 finally:
1998 cls._in_time_travel = False
2000 cls._in_time_travel = True
2001 try:
2002 # Lifecycle subscribers only see a complete, rebound scope graph.
2003 for scope in reconciliation.moved_to_limbo:
2004 cls._fire_unregister_callbacks(scope, cls._time_travel_limbo[scope])
2005 for scope in reconciliation.restored_scopes:
2006 cls._fire_register_callbacks(scope, cls._states[scope])
2007 with TimeTravelProfiler.phase(
2008 "objectstate.notify_resolved_all",
2009 snapshot=snapshot_id[:8],
2010 ):
2011 cls._notify_time_travel_state_callbacks(change_set)
2012 with TimeTravelProfiler.phase(
2013 "objectstate.notify_state_all",
2014 snapshot=snapshot_id[:8],
2015 ):
2016 cls._notify_time_travel_materialized_callbacks(change_set)
2017 with TimeTravelProfiler.phase(
2018 "objectstate.complete_callbacks_all",
2019 snapshot=snapshot_id[:8],
2020 ):
2021 cls._fire_time_travel_complete_callbacks(change_set)
2022 finally:
2023 cls._in_time_travel = False
2025 with TimeTravelProfiler.phase("objectstate.history_callbacks"):
2026 cls._fire_history_changed_callbacks()
2027 logger.info(
2028 "⏱️ TIME_TRAVEL: Traveled to %s (%s)",
2029 transaction.snapshot.label,
2030 snapshot_id[:8],
2031 )
2032 return True
2034 @classmethod
2035 def time_travel_to(cls, index: int) -> bool:
2036 """Travel to a snapshot by index in current branch history.
2038 Convenience method for UI sliders. Uses get_branch_history() to map
2039 index to snapshot_id.
2041 Args:
2042 index: Index into branch history (0 = oldest/root, -1 = newest/head)
2044 Returns:
2045 True if travel succeeded.
2046 """
2047 history = cls.get_branch_history()
2048 if not history:
2049 logger.warning("⏱️ TIME_TRAVEL: No history")
2050 return False
2052 # Normalize negative index
2053 if index < 0:
2054 index = len(history) + index
2056 if index < 0 or index >= len(history):
2057 logger.warning(f"⏱️ TIME_TRAVEL: Index {index} out of range [0, {len(history) - 1}]")
2058 return False
2060 # Index len-1 = head (newest), returning to head means exit time-travel
2061 if index == len(history) - 1:
2062 return cls.time_travel_to_head()
2064 snapshot = history[index]
2065 return cls.time_travel_to_snapshot(snapshot.id)
2067 @classmethod
2068 def time_travel_back(cls) -> bool:
2069 """Travel one step back in history (toward older/lower index)."""
2070 history = cls.get_branch_history()
2071 if not history:
2072 return False
2074 current_index = cls.get_current_snapshot_index()
2075 if current_index == -1:
2076 # At head (len-1) - go one step back
2077 if len(history) < 2:
2078 return False
2079 return cls.time_travel_to_snapshot(history[-2].id)
2081 # Already time-traveling - go one step older (lower index)
2082 if current_index <= 0:
2083 return False # Already at oldest
2084 return cls.time_travel_to_snapshot(history[current_index - 1].id)
2086 @classmethod
2087 def time_travel_forward(cls) -> bool:
2088 """Travel one step forward in history (toward newer/higher index)."""
2089 if cls._current_head is None:
2090 return False # Already at head
2092 history = cls.get_branch_history()
2093 current_index = cls.get_current_snapshot_index()
2095 # Go one step toward head (higher index)
2096 next_index = current_index + 1
2097 if next_index >= len(history) - 1:
2098 # At or past head - return to head
2099 return cls.time_travel_to_head()
2101 return cls.time_travel_to_snapshot(history[next_index].id)
2103 @classmethod
2104 def time_travel_to_head(cls) -> bool:
2105 """Return to the latest state (exit time-travel mode).
2107 Restores state from current branch's head snapshot.
2108 """
2109 if cls._current_head is None:
2110 return True # Already at head
2112 if cls._current_timeline not in cls._timelines:
2113 logger.warning("⏱️ TIME_TRAVEL: No current timeline")
2114 return False
2116 head_id = cls._timelines[cls._current_timeline].head_id
2117 result = cls.time_travel_to_snapshot(head_id)
2118 cls._current_head = None # Mark as at head (not time-traveling)
2119 return result
2121 @classmethod
2122 def get_history_info(cls, filter_fn: Optional[Callable[[str], bool]] = None) -> List[Dict[str, Any]]:
2123 """Get human-readable history for UI display.
2125 Returns history for current branch, oldest first (index 0 = oldest, -1 = head).
2127 Args:
2128 filter_fn: Optional predicate function that takes a scope_id and returns True if the scope
2129 should be included in visible state counts. If None, no filtering is applied.
2130 Example: lambda scope_id: scope_id not in {"", "__plates__"}
2131 """
2132 import datetime
2133 history = cls.get_branch_history()
2134 current_index = cls.get_current_snapshot_index()
2135 head_index = len(history) - 1
2137 result = []
2138 for i, snapshot in enumerate(history):
2139 # Filter visible states based on caller-provided predicate
2140 if filter_fn is not None:
2141 visible_states = [scope_id for scope_id in snapshot.all_states.keys()
2142 if filter_fn(scope_id)]
2143 else:
2144 visible_states = list(snapshot.all_states.keys())
2146 if not visible_states:
2147 # Skip this snapshot - no visible states
2148 continue
2150 is_head = (i == head_index)
2151 is_current = (current_index == -1 and is_head) or (i == current_index)
2152 result.append({
2153 'index': i,
2154 'id': snapshot.id,
2155 'timestamp': datetime.datetime.fromtimestamp(snapshot.timestamp).strftime('%H:%M:%S.%f')[:-3],
2156 'label': snapshot.label or f"Snapshot #{i}",
2157 'is_current': is_current,
2158 'is_head': is_head,
2159 'num_states': len(visible_states),
2160 'parent_id': snapshot.parent_id,
2161 })
2162 return result
2164 @classmethod
2165 def get_history_length(cls) -> int:
2166 """Get number of snapshots in current branch history."""
2167 return len(cls.get_branch_history())
2169 @classmethod
2170 def is_time_traveling(cls) -> bool:
2171 """Check if currently viewing historical state (not at head)."""
2172 return cls._current_head is not None
2174 @classmethod
2175 def is_restoring_snapshot(cls) -> bool:
2176 """Whether a snapshot restore is in progress, so views refresh restored values."""
2177 return cls._in_time_travel
2179 # ========== BRANCH OPERATIONS ==========
2181 @classmethod
2182 def create_branch(cls, name: str, description: str = "") -> Timeline:
2183 """Create a new branch at current position.
2185 Args:
2186 name: Branch name (must be unique)
2187 description: Optional description
2189 Returns:
2190 The created Timeline
2191 """
2192 # Branch from current position
2193 if cls._current_head is not None:
2194 # Time-traveling - branch from current position
2195 head_id = cls._current_head
2196 elif cls._current_timeline in cls._timelines:
2197 # At head of current branch
2198 head_id = cls._timelines[cls._current_timeline].head_id
2199 else:
2200 logger.error("⏱️ BRANCH: No snapshots to branch from")
2201 raise ValueError("No snapshots to branch from")
2203 timeline = Timeline(
2204 name=name,
2205 head_id=head_id,
2206 base_id=head_id,
2207 description=description,
2208 )
2209 cls._timelines[name] = timeline
2210 logger.info(f"⏱️ BRANCH: Created branch '{name}' at {head_id[:8]}")
2211 return timeline
2213 @classmethod
2214 def switch_branch(cls, name: str) -> bool:
2215 """Switch to a different branch.
2217 Args:
2218 name: Branch name
2220 Returns:
2221 True if switch succeeded
2222 """
2223 timeline = cls._timelines[name] # KeyError if branch doesn't exist
2225 # If we're back in time on current branch, preserve the old future as auto-branch
2226 # before switching (same logic as diverge in record_snapshot)
2227 if cls._current_head is not None and cls._current_timeline in cls._timelines:
2228 old_head_id = cls._timelines[cls._current_timeline].head_id
2229 if old_head_id != cls._current_head:
2230 # There IS an old future to preserve
2231 branch_name = f"auto-{old_head_id[:8]}"
2232 if branch_name not in cls._timelines:
2233 cls._timelines[branch_name] = Timeline(
2234 name=branch_name,
2235 head_id=old_head_id,
2236 base_id=cls._current_head,
2237 description=f"Auto-saved before switch from {cls._current_timeline} at {cls._current_head[:8]}",
2238 )
2239 old_snapshot = cls._snapshots[old_head_id]
2240 logger.info(f"⏱️ AUTO-BRANCH: Created '{branch_name}' before branch switch (was {old_snapshot.label})")
2242 # Switch to branch and travel to its head
2243 cls._current_timeline = name
2244 result = cls.time_travel_to_snapshot(timeline.head_id)
2245 cls._current_head = None # At head of new branch
2246 cls._fire_history_changed_callbacks()
2247 logger.info(f"⏱️ BRANCH: Switched to '{name}'")
2248 return result
2250 @classmethod
2251 def delete_branch(cls, name: str) -> None:
2252 """Delete a branch.
2254 Args:
2255 name: Branch name to delete
2257 Note: Snapshots are not deleted - they may be reachable from other branches.
2258 Unreachable snapshots are pruned automatically when DAG exceeds max size.
2259 """
2260 del cls._timelines[name] # KeyError if branch doesn't exist
2261 logger.info(f"⏱️ BRANCH: Deleted branch '{name}'")
2262 cls._fire_history_changed_callbacks()
2264 @classmethod
2265 def list_branches(cls) -> List[Dict[str, Any]]:
2266 """List all branches.
2268 Returns:
2269 List of dicts with branch info
2270 """
2271 return [
2272 {
2273 'name': tl.name,
2274 'head_id': tl.head_id,
2275 'base_id': tl.base_id,
2276 'description': tl.description,
2277 'is_current': tl.name == cls._current_timeline,
2278 }
2279 for tl in cls._timelines.values()
2280 ]
2282 @classmethod
2283 def get_current_branch(cls) -> str:
2284 """Get current branch name."""
2285 return cls._current_timeline
2287 # ========== PERSISTENCE ==========
2289 @classmethod
2290 def export_history_to_dict(cls) -> Dict[str, Any]:
2291 """Export the complete typed history document payload.
2293 Returns:
2294 Dict with 'snapshots', 'timelines', 'current_head', 'current_timeline'.
2295 """
2296 return {
2297 'format_version': 2,
2298 'snapshots': {sid: snap.to_dict() for sid, snap in cls._snapshots.items()},
2299 'timelines': [tl.to_dict() for tl in cls._timelines.values()],
2300 'current_head': cls._current_head,
2301 'current_timeline': cls._current_timeline,
2302 }
2304 @classmethod
2305 def import_history_from_dict(cls, data: dict[str, Any]) -> None:
2306 """Import a complete typed history document payload.
2308 Version 2 includes exact construction declarations. Older documents
2309 cannot prove callable ownership and are rejected without mutation.
2311 Args:
2312 data: Dict with 'snapshots', 'timelines', 'current_head', 'current_timeline'.
2313 """
2314 document = copy.deepcopy(data)
2315 if document.get("format_version") != 2:
2316 raise ValueError(
2317 "Unsupported ObjectState history version; construction ownership is required."
2318 )
2319 snapshots = {
2320 sid: Snapshot.from_dict(payload) for sid, payload in document["snapshots"].items()
2321 }
2322 timelines = {
2323 timeline.name: timeline for timeline in map(Timeline.from_dict, document["timelines"])
2324 }
2325 for sid, snapshot in snapshots.items():
2326 if sid != snapshot.id or (
2327 snapshot.parent_id is not None and snapshot.parent_id not in snapshots
2328 ):
2329 raise ValueError(f"Invalid ObjectState snapshot ancestry at {sid!r}.")
2330 for sid, snapshot in snapshots.items():
2331 ancestry = set()
2332 ancestor = sid
2333 while ancestor is not None:
2334 if ancestor in ancestry:
2335 raise ValueError(f"Cyclic ObjectState snapshot ancestry at {sid!r}.")
2336 ancestry.add(ancestor)
2337 ancestor = snapshots[ancestor].parent_id
2338 cls._prepare_time_travel_states(snapshot)
2339 for timeline in timelines.values():
2340 if timeline.head_id not in snapshots or timeline.base_id not in snapshots:
2341 raise ValueError(f"Invalid ObjectState timeline {timeline.name!r}.")
2342 if timelines and document["current_timeline"] not in timelines:
2343 raise ValueError("Missing current ObjectState timeline.")
2344 if document["current_head"] is not None and document["current_head"] not in snapshots:
2345 raise ValueError("Missing current ObjectState snapshot.")
2346 cls._snapshots = snapshots
2347 cls._timelines = timelines
2348 cls._current_timeline = document["current_timeline"]
2349 cls._current_head = document["current_head"]
2351 @classmethod
2352 def save_history_to_file(cls, filepath: str) -> None:
2353 """Save history as a type-preserving ObjectState document.
2355 The document is intended for trusted local persistence. Dill preserves
2356 arbitrary Python values and graph identity without a parallel type-tag
2357 registry.
2359 Args:
2360 filepath: Path to the binary ObjectState history document.
2361 """
2362 import dill
2364 with open(filepath, "wb") as history_file:
2365 dill.dump(cls.export_history_to_dict(), history_file)
2366 logger.info(f"⏱️ Saved {len(cls._snapshots)} snapshots to {filepath}")
2368 @classmethod
2369 def load_history_from_file(cls, filepath: str) -> None:
2370 """Load and materialize a trusted ObjectState history document.
2372 Args:
2373 filepath: Path to the binary ObjectState history document.
2374 """
2375 import dill
2377 previous_history = (
2378 cls._snapshots,
2379 cls._timelines,
2380 cls._current_timeline,
2381 cls._current_head,
2382 )
2383 try:
2384 with open(filepath, "rb") as history_file:
2385 cls.import_history_from_dict(dill.load(history_file))
2386 restored_at_head = cls._current_head is None
2387 snapshot_id = cls._current_snapshot_id()
2388 if snapshot_id is not None:
2389 if not cls.time_travel_to_snapshot(snapshot_id):
2390 raise RuntimeError(
2391 "Loaded ObjectState history could not restore its current snapshot."
2392 )
2393 if restored_at_head:
2394 cls._current_head = None
2395 except Exception:
2396 cls._snapshots, cls._timelines, cls._current_timeline, cls._current_head = (
2397 previous_history
2398 )
2399 raise
2400 logger.info(f"⏱️ Loaded {len(cls._snapshots)} snapshots from {filepath}")