From c65f267e362e0493f6a1622cd36d02d6778a54f7 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 03:25:36 -0400 Subject: [PATCH 1/8] ScopeKey owns the scope hierarchy grammar Scope ids were plain strings whose '::' grammar was restated at every use: joining, ancestors, descendant tests, depth and last segment. ScopeKey, a str subclass, owns that grammar once; the registry stores its states under ScopeKey keys and every hierarchy operation calls it. Containment is by whole segments, so /plate1 no longer looks like a parent of /plate10 to callers that used a bare prefix test. get_root_from_scope_key and is_scope_affected, which nothing called, are replaced by ScopeKey.root and ScopeKey.contains. Restoring the global scope still restores no descendant parameters: the global scope provides context to every scope and owns none of their state. Co-Authored-By: Claude Opus 5.5 --- src/objectstate/__init__.py | 7 +- src/objectstate/context_manager.py | 50 -------------- src/objectstate/object_state.py | 34 ++++----- src/objectstate/object_state_registry.py | 67 ++++-------------- src/objectstate/scope_key.py | 88 ++++++++++++++++++++++++ tests/test_scope_key.py | 47 +++++++++++++ 6 files changed, 165 insertions(+), 128 deletions(-) create mode 100644 src/objectstate/scope_key.py create mode 100644 tests/test_scope_key.py diff --git a/src/objectstate/__init__.py b/src/objectstate/__init__.py index 6fcb0af..16de44e 100644 --- a/src/objectstate/__init__.py +++ b/src/objectstate/__init__.py @@ -100,9 +100,6 @@ register_hierarchy_relationship, unregister_hierarchy_relationship, get_ancestors_from_hierarchy, - # Scope key utilities - get_root_from_scope_key, - is_scope_affected, # Context stack building (framework-agnostic) build_context_stack, # Threading utility for context propagation @@ -181,8 +178,6 @@ 'extract_all_configs', 'get_base_global_config', 'get_context_type_stack', - 'get_root_from_scope_key', - 'is_scope_affected', 'build_context_stack', 'spawn_thread_with_context', # Placeholder @@ -223,6 +218,7 @@ # Object state management from objectstate.object_state_registry import ObjectStateRegistry +from objectstate.scope_key import ScopeKey from objectstate.object_state import ObjectState from objectstate.object_state_metadata import ( ObjectStateMetadataContract, @@ -259,6 +255,7 @@ # Object state 'ObjectState', 'ObjectStateRegistry', + 'ScopeKey', 'ObjectStateMetadataContract', 'ObjectStateMetadataContractRegistry', 'ObjectStateMetadataStore', diff --git a/src/objectstate/context_manager.py b/src/objectstate/context_manager.py index 6f63e9a..575c686 100644 --- a/src/objectstate/context_manager.py +++ b/src/objectstate/context_manager.py @@ -329,56 +329,6 @@ def _is_global_type(t): _known_hierarchy: dict = {} -def get_root_from_scope_key(scope_key: str) -> str: - """Extract root (plate path) from scope_key for visibility checks. - - scope_key format: - - Pipeline-level: just plate path (e.g., "/path/to/plate") - - Step-level: plate_path::step_token (e.g., "/path/to/plate::step_a") - - Global: empty string - - Returns the portion before "::" (or the whole string if no "::" present). - """ - if not scope_key: - return "" - return scope_key.split("::")[0] - - -def is_scope_affected(target_scope_id: str | None, editing_scope_id: str | None) -> bool: - """Check if target scope is affected by edit at editing scope. - - Uses scope_id hierarchy - no type introspection needed: - - None/"" (global) → affects all - - "plate_path" (pipeline) → affects same plate + all its steps - - "plate_path::token" (step) → affects only that step - - Args: - target_scope_id: The scope being checked for affectedness (None = global) - editing_scope_id: The scope where the edit occurred (None = global) - - Returns: - True if target should refresh when editing_scope changes - """ - # Normalize None to empty string - target = target_scope_id or "" - editing = editing_scope_id or "" - - # Global edit affects all - if not editing: - return True - - # Different plate roots = not affected - target_root = get_root_from_scope_key(target) - editing_root = get_root_from_scope_key(editing) - if target_root != editing_root: - return False - - # Same root: parent affects children, not vice versa - # editing="plate" affects target="plate::step" ✓ - # editing="plate::step" affects target="plate" ✗ - return target == editing or target.startswith(editing + "::") - - def _normalize_type_for_hierarchy(t): """Normalize a type for hierarchy registry, but preserve lazy-global distinction. diff --git a/src/objectstate/object_state.py b/src/objectstate/object_state.py index cdb06f5..b462a4a 100644 --- a/src/objectstate/object_state.py +++ b/src/objectstate/object_state.py @@ -30,6 +30,7 @@ ObjectStateMetadataStore, ) from objectstate.object_state_registry import ObjectStateRegistry +from objectstate.scope_key import ScopeKey from objectstate.parameter_owner import ParameterOwner from objectstate.parameter_structure import ParameterStructure from objectstate.global_config import GlobalContextValues @@ -695,9 +696,9 @@ def on_child_changed(changed_paths): parent_field = field_path or self._parent_field_name if parent_field is None: # Auto-detect from a numeric scope suffix. - parts = self.scope_id.split('::') - if len(parts) >= 2: - last = parts[-1] + scope = ScopeKey(self.scope_id) + if scope.depth >= 2: + last = scope.last import re m = re.match(r'^(.+?)_\d+$', last) parent_field = m.group(1) if m else last @@ -2304,23 +2305,11 @@ def mark_saved(self) -> None: logger.debug(f"🔧 mark_saved: Total states in registry: {len(ObjectStateRegistry._states)}") # Collect descendant scopes first to avoid modifying registry during iteration. - # - # IMPORTANT: Global scope is represented by "" (empty string). - # In that case, *every* non-global scope is a descendant, but the naive - # prefix check ("" + "::" == "::") matches nothing. changed_scope = ObjectStateRegistry._normalize_scope_id(self.scope_id) - if changed_scope == "": - # Global baseline change affects ALL other states. - descendant_scopes = [ - s.scope_id for s in ObjectStateRegistry._states.values() - if ObjectStateRegistry._normalize_scope_id(s.scope_id) != "" - ] - else: - prefix = changed_scope + "::" - descendant_scopes = [ - s.scope_id for s in ObjectStateRegistry._states.values() - if s.scope_id is not None and ObjectStateRegistry._normalize_scope_id(s.scope_id).startswith(prefix) - ] + descendant_scopes = [ + s.scope_id for s in ObjectStateRegistry._states.values() + if changed_scope.strictly_contains(s.scope_id) + ] logger.debug(f"🔧 mark_saved: Found {len(descendant_scopes)} descendant scopes: {descendant_scopes}") for descendant_scope in descendant_scopes: @@ -2456,11 +2445,14 @@ def _restore_saved_impl(self, *, propagate_descendants: bool = True) -> None: # However, for delegation-based parents that act as context providers (e.g. orchestrator -> pipeline_config), # restoring the parent should generally *not* restore descendant raw parameters (steps). Descendants should # only have their resolved caches invalidated so they re-resolve against the restored context. - if propagate_descendants: + # The global scope only provides context to every other scope; it owns none of + # their parameters, so restoring it never restores them. + own_scope = ScopeKey(self.scope_id) + if propagate_descendants and not own_scope.is_global: descendant_scopes = [ scope for scope in ObjectStateRegistry._states.keys() - if scope.startswith(f"{self.scope_id}::") + if own_scope.strictly_contains(scope) ] for descendant_scope in descendant_scopes: state = ObjectStateRegistry._states.get(descendant_scope) diff --git a/src/objectstate/object_state_registry.py b/src/objectstate/object_state_registry.py index 9d98476..19b1c25 100644 --- a/src/objectstate/object_state_registry.py +++ b/src/objectstate/object_state_registry.py @@ -23,6 +23,7 @@ from objectstate.object_state_metadata import ObjectStateMetadataStore from objectstate.construction_binding import StateConstructionBinding from objectstate.history_migration import HistoryMigration +from objectstate.scope_key import ScopeKey from objectstate.global_config import GlobalContextValues from objectstate.context_manager import clear_current_temp_global from objectstate.dual_axis_resolver import clear_mro_resolution_cache @@ -348,9 +349,9 @@ def mark_snapshot_dirty_scope(cls, scope_id: Optional[str]) -> None: cls._snapshot_dirty_scopes.add(cls._normalize_scope_id(scope_id)) @classmethod - def _normalize_scope_id(cls, scope_id: Optional[str]) -> str: - """Normalize scope_id: None and "" both represent global scope.""" - return scope_id if scope_id is not None else "" + def _normalize_scope_id(cls, scope_id: Optional[str]) -> ScopeKey: + """Return the registry key: None and "" both represent the global scope.""" + return ScopeKey(scope_id) @classmethod def register(cls, state: 'ObjectState', _skip_snapshot: bool = False) -> None: @@ -426,14 +427,11 @@ def unregister_scope_and_descendants(cls, scope_id: Optional[str], _skip_snapsho scope_key = cls._normalize_scope_id(scope_id) # Find all scopes to delete: exact match + descendants - scopes_to_delete = [] - for key in cls._states.keys(): - # Exact match - if key == scope_key: - scopes_to_delete.append(key) - # Descendant (starts with scope_key::) - elif scope_key and key.startswith(scope_key + "::"): - scopes_to_delete.append(key) + scopes_to_delete = [ + key + for key in cls._states.keys() + if key == scope_key or (scope_key and scope_key.contains(key)) + ] # Delete all matching scopes and fire callbacks # NOTE: Unlike plain unregister(), this is often used for hierarchical deletes. @@ -640,17 +638,9 @@ def get_ancestor_objects(cls, scope_id: Optional[str], use_saved: bool = False) """ scope_key = cls._normalize_scope_id(scope_id) - # Build list of ancestor scope keys (least-specific to most-specific) - # e.g., "/plate::step_0" -> ["", "/plate", "/plate::step_0"] - ancestors = [""] # Global scope always included - if scope_key: - parts = scope_key.split("::") - for i in range(len(parts)): - ancestors.append("::".join(parts[:i+1])) - - # Get objects from ancestor scopes + # Get objects from ancestor scopes, least-specific to most-specific objects = [] - for ancestor_key in ancestors: + for ancestor_key in scope_key.ancestors(): state = cls._states.get(ancestor_key) if state: if use_saved: @@ -693,16 +683,9 @@ def get_ancestor_objects_with_scopes( """ scope_key = cls._normalize_scope_id(scope_id) - # Build list of ancestor scope keys (least-specific to most-specific) - ancestors = [""] # Global scope always included - if scope_key: - parts = scope_key.split("::") - for i in range(len(parts)): - ancestors.append("::".join(parts[:i+1])) - # Get (scope_id, object) tuples from ancestor scopes results: List[Tuple[str, Any]] = [] - for ancestor_key in ancestors: + for ancestor_key in scope_key.ancestors(): state = cls._states.get(ancestor_key) if state: # Get object - either with or without delegate sync @@ -809,20 +792,8 @@ def _invalidate_by_type_and_scope_now( continue # SCOPE CHECK: must be at or below changed scope - # Global scope (empty string) affects ALL states - if changed_scope == "": - # Global scope - always a descendant (or self if also global) - if field_name == 'well_filter': - logger.debug(f"🔍 Checking state: scope={state_scope!r}, obj_type={type(state.object_instance).__name__}") - logger.debug(f"[SCOPE] Global change affects state scope={state_scope!r}") - else: - # Non-global: check exact match or descendant - prefix = changed_scope + "::" - is_descendant = state_scope.startswith(prefix) - if not (is_self or is_descendant): - logger.debug(f"[SCOPE] SKIP: changed_scope={changed_scope!r} does not affect state_scope={state_scope!r}") - continue - logger.debug(f"[SCOPE] MATCH: changed_scope={changed_scope!r} affects state_scope={state_scope!r}") + if not changed_scope.contains(state_scope): + continue # TYPE + FIELD CHECK: find matching nested state and invalidate field cls._invalidate_field_in_matching_states(state, base_changed_type, field_name, invalidate_saved) @@ -1353,18 +1324,10 @@ def _states_by_dependency_order(cls) -> tuple[tuple[str, 'ObjectState'], ...]: return tuple( sorted( cls._states.items(), - key=lambda item: (cls._scope_depth(item[0]), item[0]), + key=lambda item: (item[0].depth, item[0]), ) ) - @staticmethod - def _scope_depth(scope_id: str) -> int: - """Return hierarchy depth for ObjectState scope identifiers.""" - - if not scope_id: - return 0 - return scope_id.count("::") + 1 - @classmethod def _prune_unreachable_snapshots(cls) -> None: """Remove snapshots not reachable from any branch head. diff --git a/src/objectstate/scope_key.py b/src/objectstate/scope_key.py new file mode 100644 index 0000000..f3f8ffd --- /dev/null +++ b/src/objectstate/scope_key.py @@ -0,0 +1,88 @@ +"""Hierarchical scope identifiers for registered ObjectStates.""" + +from __future__ import annotations + +from typing import ClassVar + + +class ScopeKey(str): + """A scope id: segments joined by ``SEPARATOR``; the empty key is the global scope. + + A scope contains itself and every scope whose segments extend its own; the + global scope contains every scope. ``ScopeKey`` is a ``str``, so it is a + dictionary key and serializes as its text. + """ + + __slots__ = () + + SEPARATOR: ClassVar[str] = "::" + + def __new__(cls, value: str | None = "") -> ScopeKey: + return super().__new__(cls, "" if value is None else value) + + @classmethod + def of(cls, *parts: str) -> ScopeKey: + """Join ``parts`` into one key; each part is a single segment.""" + + for part in parts: + if not part or cls.SEPARATOR in part: + raise ValueError(f"Scope segment {part!r} is empty or contains {cls.SEPARATOR!r}.") + return cls(cls.SEPARATOR.join(parts)) + + @property + def is_global(self) -> bool: + return not self + + @property + def parts(self) -> tuple[str, ...]: + return tuple(self.split(self.SEPARATOR)) if self else () + + @property + def depth(self) -> int: + """Number of segments; the global scope has depth 0.""" + + return len(self.parts) + + @property + def last(self) -> str: + """The final segment; the empty string for the global scope.""" + + return self.parts[-1] if self else "" + + @property + def root(self) -> ScopeKey: + """The first segment as a key; the global scope is its own root.""" + + return ScopeKey(self.parts[0]) if self else self + + @property + def parent(self) -> ScopeKey: + """The key without its final segment; a root's parent is the global scope.""" + + return ScopeKey(self.SEPARATOR.join(self.parts[:-1])) + + def child(self, segment: str) -> ScopeKey: + """The key one segment below this one.""" + + return ScopeKey.of(*self.parts, segment) + + def ancestors(self) -> tuple[ScopeKey, ...]: + """The global scope, then each prefix of this key, ending with the key itself.""" + + parts = self.parts + return (ScopeKey(),) + tuple( + ScopeKey(self.SEPARATOR.join(parts[: index + 1])) for index in range(len(parts)) + ) + + def contains(self, other: str | None) -> bool: + """Whether ``other`` is this scope or below it.""" + + other_key = ScopeKey(other) + if self.is_global or other_key == self: + return True + return other_key.startswith(self + self.SEPARATOR) + + def strictly_contains(self, other: str | None) -> bool: + """Whether ``other`` is below this scope and not the scope itself.""" + + return ScopeKey(other) != self and self.contains(other) diff --git a/tests/test_scope_key.py b/tests/test_scope_key.py new file mode 100644 index 0000000..b59125c --- /dev/null +++ b/tests/test_scope_key.py @@ -0,0 +1,47 @@ +import pickle + +import pytest + +from objectstate import ObjectStateRegistry, ScopeKey + + +def test_scope_key_owns_the_hierarchy_grammar(): + step = ScopeKey.of("/plate1", "step_0") + function = step.child("func_1") + + assert function == "/plate1::step_0::func_1" + assert function.parts == ("/plate1", "step_0", "func_1") + assert (function.depth, function.last) == (3, "func_1") + assert function.root == "/plate1" and function.parent == step + assert function.ancestors() == ("", "/plate1", step, function) + assert ScopeKey(None) == ScopeKey("") and ScopeKey().is_global + assert ScopeKey().parts == () and ScopeKey().ancestors() == ("",) + + +def test_containment_is_by_whole_segments(): + plate = ScopeKey("/plate1") + + assert plate.contains("/plate1") and plate.contains("/plate1::step_0") + assert not plate.contains("/plate10") + assert not plate.contains("/plate10::step_0") + assert not plate.strictly_contains("/plate1") + assert ScopeKey().contains("/anything") and ScopeKey().strictly_contains("/anything") + assert not ScopeKey("/plate1::step_0").contains("/plate1") + + +@pytest.mark.parametrize("segment", ("", "a::b")) +def test_a_segment_is_one_nonempty_segment(segment): + with pytest.raises(ValueError): + ScopeKey("/plate").child(segment) + + +def test_scope_key_is_its_text_as_a_key_and_when_pickled(): + key = ScopeKey.of("/plate", "step") + + assert {"/plate::step": 1}[key] == 1 + restored = pickle.loads(pickle.dumps(key)) + assert type(restored) is ScopeKey and restored == key + + +def test_registry_keys_are_scope_keys(): + assert type(ObjectStateRegistry._normalize_scope_id(None)) is ScopeKey From a4cfcf883d0e8474bfa3a1d1d9a634b748346c9d Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 03:25:36 -0400 Subject: [PATCH 2/8] Delete HistoryMigration and the migration= parameter A history document is current or refused. The one application that passed a converter (OpenHCS's desktop history upgrade) deleted it: desktop histories are runtime state and reset on upgrade. Co-Authored-By: Claude Opus 5.5 --- src/objectstate/history_migration.py | 19 ------------------- src/objectstate/object_state_registry.py | 17 ++++------------- tests/test_construction_history.py | 20 -------------------- 3 files changed, 4 insertions(+), 52 deletions(-) delete mode 100644 src/objectstate/history_migration.py diff --git a/src/objectstate/history_migration.py b/src/objectstate/history_migration.py deleted file mode 100644 index b83a9e1..0000000 --- a/src/objectstate/history_migration.py +++ /dev/null @@ -1,19 +0,0 @@ -"""Explicit trusted upgrades of complete typed history documents.""" - -from abc import ABC, abstractmethod -from collections.abc import Mapping -from typing import Any - -from objectstate.construction_binding import StateConstructionBinding - - -class HistoryMigration(ABC): - """An application-owned proof of historical construction declarations.""" - - @abstractmethod - def migrate( - self, - document: dict[str, Any], - constructions: Mapping[str, StateConstructionBinding], - ) -> dict[str, Any]: - """Return version 2 or reject; current bindings alone are not history proof.""" diff --git a/src/objectstate/object_state_registry.py b/src/objectstate/object_state_registry.py index 19b1c25..2d94e42 100644 --- a/src/objectstate/object_state_registry.py +++ b/src/objectstate/object_state_registry.py @@ -22,7 +22,6 @@ from objectstate.object_state_metadata import ObjectStateMetadataStore from objectstate.construction_binding import StateConstructionBinding -from objectstate.history_migration import HistoryMigration from objectstate.scope_key import ScopeKey from objectstate.global_config import GlobalContextValues from objectstate.context_manager import clear_current_temp_global @@ -2282,9 +2281,7 @@ def export_history_to_dict(cls) -> Dict[str, Any]: } @classmethod - def import_history_from_dict( - cls, data: Dict[str, Any], *, migration: HistoryMigration | None = None - ) -> None: + def import_history_from_dict(cls, data: Dict[str, Any]) -> None: """Import a complete typed history document payload. Version 2 includes exact construction declarations. Older documents @@ -2293,11 +2290,7 @@ def import_history_from_dict( Args: data: Dict with 'snapshots', 'timelines', 'current_head', 'current_timeline'. """ - document, constructions = copy.deepcopy( - (data, {scope: state.construction_binding() for scope, state in cls._states.items()}) - ) - if document.get("format_version") != 2 and migration is not None: - document = migration.migrate(document, constructions) + document = copy.deepcopy(data) if document.get("format_version") != 2: raise ValueError( "Unsupported ObjectState history version; construction ownership is required." @@ -2352,9 +2345,7 @@ def save_history_to_file(cls, filepath: str) -> None: logger.info(f"⏱️ Saved {len(cls._snapshots)} snapshots to {filepath}") @classmethod - def load_history_from_file( - cls, filepath: str, *, migration: HistoryMigration | None = None - ) -> None: + def load_history_from_file(cls, filepath: str) -> None: """Load and materialize a trusted ObjectState history document. Args: @@ -2370,7 +2361,7 @@ def load_history_from_file( ) try: with open(filepath, "rb") as history_file: - cls.import_history_from_dict(dill.load(history_file), migration=migration) + cls.import_history_from_dict(dill.load(history_file)) restored_at_head = cls._current_head is None snapshot_id = cls._current_snapshot_id() if snapshot_id is not None: diff --git a/tests/test_construction_history.py b/tests/test_construction_history.py index 3728e0d..2fd6ccd 100644 --- a/tests/test_construction_history.py +++ b/tests/test_construction_history.py @@ -15,7 +15,6 @@ set_base_config_type, set_global_config_for_editing, ) -from objectstate.history_migration import HistoryMigration @dataclass @@ -480,22 +479,3 @@ def test_failed_attachment_rolls_back_complete_graph_and_loaded_history(tmp_path assert path.read_bytes() == file_bytes -def test_explicit_migration_receives_one_detached_alias_graph_and_must_return_v2(): - shared = [1] - state = register(Config(items=shared), "direct") - record("migration input") - document = ObjectStateRegistry.export_history_to_dict() - document["format_version"] = 1 - document["source_alias"] = shared - - class FixtureMigration(HistoryMigration): - def migrate(self, document, constructions): - assert document["source_alias"] is constructions["direct"].target.items - document["source_alias"].append(7) - document["format_version"] = 2 - return document - - ObjectStateRegistry.import_history_from_dict(document, migration=FixtureMigration()) - assert document["format_version"] == 1 - assert document["source_alias"] == shared == [1] - assert state.object_instance.items is shared From fcf8546cddb73373df4721a5f78cedf0bc354099 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 03:41:40 -0400 Subject: [PATCH 3/8] Name the main branch once; expose whether a snapshot restore is running MAIN_BRANCH replaces four 'main' literals, so callers that fall back to the main branch (deleting the current branch) name it through the registry. is_restoring_snapshot() is the public reading of the restore-in-progress flag that view libraries read privately to refresh widget values after time travel. Co-Authored-By: Claude Opus 5.5 --- src/objectstate/object_state_registry.py | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/src/objectstate/object_state_registry.py b/src/objectstate/object_state_registry.py index 2d94e42..a7597c5 100644 --- a/src/objectstate/object_state_registry.py +++ b/src/objectstate/object_state_registry.py @@ -503,7 +503,7 @@ def clear(cls) -> None: cls._snapshots.clear() cls._timelines.clear() cls._current_head = None - cls._current_timeline = "main" + cls._current_timeline = cls.MAIN_BRANCH cls._atomic_depth = 0 cls._atomic_label = None cls._atomic_triggering_scope = None @@ -896,11 +896,12 @@ def _invalidate_field_in_matching_states( # This allows time travel to snapshots that reference deleted objects. _graveyard: Dict[str, 'ObjectState'] = {} - # Timelines (branches): Named branches that point to a head snapshot - # "main" is always created automatically on first snapshot + # Timelines (branches): Named branches that point to a head snapshot. + # MAIN_BRANCH is created automatically on the first snapshot. # head_id always points to a valid key in _snapshots (guaranteed by construction) + MAIN_BRANCH = "main" _timelines: Dict[str, Timeline] = {} - _current_timeline: str = "main" + _current_timeline: str = MAIN_BRANCH @classmethod @contextmanager @@ -1003,7 +1004,7 @@ def atomic_success( del cls._snapshots[snapshot_id] cls._timelines = timelines_before cls._current_head = current_head_before - cls._current_timeline = current_timeline_before or "main" + cls._current_timeline = current_timeline_before or cls.MAIN_BRANCH cls._in_time_travel = bool(in_time_travel_before) if snapshot_dirty_scopes_before is not None: cls._snapshot_dirty_scopes = snapshot_dirty_scopes_before @@ -1276,7 +1277,7 @@ def _record_snapshot_internal( name=cls._current_timeline, head_id=snapshot.id, base_id=snapshot.id, - description="Main timeline" if cls._current_timeline == "main" else "", + description="Main timeline" if cls._current_timeline == cls.MAIN_BRANCH else "", ) logger.info(f"⏱️ BRANCH: Created '{cls._current_timeline}' timeline") else: @@ -2155,6 +2156,11 @@ def is_time_traveling(cls) -> bool: """Check if currently viewing historical state (not at head).""" return cls._current_head is not None + @classmethod + def is_restoring_snapshot(cls) -> bool: + """Whether a snapshot restore is in progress, so views refresh restored values.""" + return cls._in_time_travel + # ========== BRANCH OPERATIONS ========== @classmethod From 6a84d43963784eaf4d3d4bb24e0374146a3007fa Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 03:51:18 -0400 Subject: [PATCH 4/8] retain_only_authored_paths: lazy dataclass copy marking only authored fields explicit A lazy dataclass rebuilt from ObjectState's flattened fields reads every field as explicitly set. Clean code generation needs only the authored paths, which ObjectState already tracks as signature-diff fields. The function lived in an application's serializer while touching only ObjectState's lazy metadata. Co-Authored-By: Claude Opus 5.5 --- src/objectstate/__init__.py | 2 ++ src/objectstate/lazy_factory.py | 43 +++++++++++++++++++++++++++++++++ tests/test_authored_paths.py | 29 ++++++++++++++++++++++ 3 files changed, 74 insertions(+) create mode 100644 tests/test_authored_paths.py diff --git a/src/objectstate/__init__.py b/src/objectstate/__init__.py index 16de44e..668ea9f 100644 --- a/src/objectstate/__init__.py +++ b/src/objectstate/__init__.py @@ -73,6 +73,7 @@ is_global_config_instance, # Raw-value-preserving utilities replace_raw, + retain_only_authored_paths, ) # Resolver @@ -256,6 +257,7 @@ 'ObjectState', 'ObjectStateRegistry', 'ScopeKey', + 'retain_only_authored_paths', 'ObjectStateMetadataContract', 'ObjectStateMetadataContractRegistry', 'ObjectStateMetadataStore', diff --git a/src/objectstate/lazy_factory.py b/src/objectstate/lazy_factory.py index 498e884..a1ed3ec 100644 --- a/src/objectstate/lazy_factory.py +++ b/src/objectstate/lazy_factory.py @@ -250,6 +250,49 @@ def replace_raw(instance, **changes): return type(instance)(**field_values) + +def retain_only_authored_paths(value, authored_field_paths) -> Any: + """Copy a lazy dataclass, marking only ``authored_field_paths`` as explicitly set. + + An ObjectState rebuild constructs every nested lazy dataclass from all its + flattened fields, so every constructor keyword reads as explicit. Clean + source generation needs the authored paths ObjectState already tracks + (its signature-diff fields); this applies them to the lazy metadata without + changing any field value. + """ + + from objectstate.field_access import DottedFieldPath + + paths = tuple(DottedFieldPath(path) for path in authored_field_paths) + return _copy_with_only_authored_paths(value, DottedFieldPath(""), paths) + + +def _copy_with_only_authored_paths(value, prefix, authored_field_paths): + from objectstate.field_access import DataclassFieldAccess + + field_values = {} + for field_definition in fields(value): + field_value = DataclassFieldAccess.raw_value(value, field_definition.name) + if isinstance(field_value, LazyDataclass): + field_value = _copy_with_only_authored_paths( + field_value, + prefix.child(field_definition.name), + authored_field_paths, + ) + field_values[field_definition.name] = field_value + + projected = type(value)(**field_values) + object.__setattr__( + projected, + "_explicitly_set_fields", + { + field_definition.name + for field_definition in fields(projected) + if prefix.child(field_definition.name).contains_any(authored_field_paths) + }, + ) + return projected + # ContextEventCoordinator removed - replaced with contextvars-based context system diff --git a/tests/test_authored_paths.py b/tests/test_authored_paths.py new file mode 100644 index 0000000..312080e --- /dev/null +++ b/tests/test_authored_paths.py @@ -0,0 +1,29 @@ +from dataclasses import dataclass, field + +from objectstate import LazyDataclassFactory, retain_only_authored_paths + + +@dataclass +class InnerAuthored: + depth: int = 1 + width: int = 2 + + +@dataclass +class OuterAuthored: + name: str = "a" + inner: InnerAuthored = field(default_factory=InnerAuthored) + + +def test_retain_only_authored_paths_keeps_values_and_marks_only_authored_fields(): + lazy_inner = LazyDataclassFactory.make_lazy_simple(InnerAuthored) + lazy_outer = LazyDataclassFactory.make_lazy_simple(OuterAuthored) + rebuilt = lazy_outer(name="b", inner=lazy_inner(depth=5, width=None)) + + projected = retain_only_authored_paths(rebuilt, {"inner.depth"}) + + assert object.__getattribute__(projected, "name") == "b" + assert object.__getattribute__(projected.inner, "depth") == 5 + assert object.__getattribute__(projected.inner, "width") is None + assert projected._explicitly_set_fields == {"inner"} + assert projected.inner._explicitly_set_fields == {"depth"} From af5b4f6186089618671c8254dfaeb227f17d39b6 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 04:03:35 -0400 Subject: [PATCH 5/8] objectstate.inspection: read-only field and state inspection FieldInspection reads one field's raw and resolved value (an unresolvable value is reported unavailable), its dirty and signature-diff state, whether it inherits, its provenance and type names. field_markers/state_markers own the '*' (dirty) and '_' (differs from signature default) markers views prefix to fields; applications wrote that rule in each view. Co-Authored-By: Claude Opus 5.5 --- src/objectstate/inspection.py | 162 ++++++++++++++++++++++++++++++++++ tests/test_inspection.py | 50 +++++++++++ 2 files changed, 212 insertions(+) create mode 100644 src/objectstate/inspection.py create mode 100644 tests/test_inspection.py diff --git a/src/objectstate/inspection.py b/src/objectstate/inspection.py new file mode 100644 index 0000000..cc5df51 --- /dev/null +++ b/src/objectstate/inspection.py @@ -0,0 +1,162 @@ +"""Read-only inspection of registered ObjectStates and their fields.""" + +from __future__ import annotations + +import logging +from collections.abc import Callable +from dataclasses import dataclass + +from objectstate.object_state import ObjectState +from objectstate.object_state_registry import ObjectStateRegistry + +logger = logging.getLogger(__name__) + +DIRTY_MARKER = "*" +"""Marks a field (or a group of fields) whose value differs from the saved value.""" + +SIGNATURE_DIFF_MARKER = "_" +"""Marks a field whose raw value differs from its declared signature default.""" + + +def field_markers(*, dirty: bool, signature_diff: bool) -> tuple[str, ...]: + """The markers views prefix to a field, or to a group of fields, in this order.""" + + return tuple( + marker + for marker, present in ((DIRTY_MARKER, dirty), (SIGNATURE_DIFF_MARKER, signature_diff)) + if present + ) + + +def state_markers(state: ObjectState) -> tuple[str, ...]: + """The markers of a whole ObjectState: any dirty field, any signature difference.""" + + return field_markers( + dirty=bool(state.dirty_fields), + signature_diff=bool(state.signature_diff_fields), + ) + + +def type_name(value: object) -> str: + return "None" if value is None else type(value).__name__ + + +def type_qualname(value_type: type | Callable[..., object]) -> str: + return f"{value_type.__module__}.{value_type.__qualname__}" + + +@dataclass(frozen=True, slots=True) +class FieldProvenance: + """The ancestor scope and declaring type a field's inherited value comes from.""" + + source_scope_id: str | None + source_type: type + field_name: str + + @property + def source_field_path(self) -> str | None: + """The field's path in the source scope's own state, if that scope is registered.""" + + source_state = ObjectStateRegistry.get_by_scope(self.source_scope_id) + if source_state is None: + return None + return source_state.project_ui_visible_field_path(self.source_type, self.field_name) + + +@dataclass(frozen=True, slots=True) +class FieldInspection: + """One field of one ObjectState: raw and resolved value, edit state and provenance.""" + + state: ObjectState + field_path: str + raw_value: object + resolved_value: object + resolved_value_available: bool + provenance: FieldProvenance | None + + @classmethod + def of(cls, state: ObjectState, field_path: str): + """Inspect ``field_path``; a value that fails to resolve is reported unavailable.""" + + resolved_value = None + resolved_value_available = False + try: + resolved_value = state.get_resolved_value(field_path) + resolved_value_available = True + except Exception: + logger.debug( + "Failed to resolve ObjectState field %s on scope %r.", + field_path, + state.scope_id, + exc_info=True, + ) + provenance = None + match state.get_provenance(field_path): + case (source_scope_id, source_type) if source_type is not None: + provenance = FieldProvenance( + source_scope_id=source_scope_id, + source_type=source_type, + field_name=field_path.rsplit(".", 1)[-1], + ) + return cls( + state=state, + field_path=field_path, + raw_value=state.parameters[field_path], + resolved_value=resolved_value, + resolved_value_available=resolved_value_available, + provenance=provenance, + ) + + @property + def dirty(self) -> bool: + return self.field_path in self.state.dirty_fields + + @property + def signature_diff(self) -> bool: + return self.field_path in self.state.signature_diff_fields + + @property + def last_changed(self) -> bool: + return self.field_path == self.state.last_changed_field + + @property + def raw_value_is_none(self) -> bool: + return self.raw_value is None + + @property + def resolved_value_is_none(self) -> bool: + return self.resolved_value_available and self.resolved_value is None + + @property + def inherited_value(self) -> bool: + """The field holds no value of its own and inherits a resolved one.""" + + return self.raw_value is None and self.resolved_value_available and ( + self.resolved_value is not None + ) + + @property + def semantic_markers(self) -> tuple[str, ...]: + return field_markers(dirty=self.dirty, signature_diff=self.signature_diff) + + @property + def field_name(self) -> str: + return self.field_path.rsplit(".", 1)[-1] + + @property + def container_path(self) -> str: + return self.field_path.rsplit(".", 1)[0] if "." in self.field_path else "" + + @property + def raw_value_type(self) -> str: + return type_name(self.raw_value) + + @property + def resolved_value_type(self) -> str | None: + return type_name(self.resolved_value) if self.resolved_value_available else None + + @property + def declaring_type(self) -> str: + """Qualified name of the dataclass that declares this field.""" + + return type_qualname(self.state.type_for_path(self.field_path)) diff --git a/tests/test_inspection.py b/tests/test_inspection.py new file mode 100644 index 0000000..3d6b146 --- /dev/null +++ b/tests/test_inspection.py @@ -0,0 +1,50 @@ +from dataclasses import dataclass + +import pytest + +from objectstate import ObjectState, ObjectStateRegistry, set_base_config_type +from objectstate.inspection import ( + DIRTY_MARKER, + SIGNATURE_DIFF_MARKER, + FieldInspection, + field_markers, + state_markers, +) + + +@dataclass +class Inspected: + count: int = 1 + label: str = "x" + + +@pytest.fixture(autouse=True) +def _registry(): + set_base_config_type(Inspected) + ObjectStateRegistry.clear() + yield + ObjectStateRegistry.clear() + + +def test_markers_are_ordered_dirty_then_signature_difference(): + assert field_markers(dirty=True, signature_diff=True) == (DIRTY_MARKER, SIGNATURE_DIFF_MARKER) + assert field_markers(dirty=False, signature_diff=True) == (SIGNATURE_DIFF_MARKER,) + assert field_markers(dirty=False, signature_diff=False) == () + + +def test_field_inspection_reports_value_edit_state_and_types(): + state = ObjectState(Inspected(count=3), scope_id="/inspected") + ObjectStateRegistry.register(state, _skip_snapshot=True) + state.update_parameter("label", "y") + + label = FieldInspection.of(state, "label") + count = FieldInspection.of(state, "count") + + assert (label.raw_value, label.resolved_value, label.resolved_value_available) == ("y", "y", True) + assert label.dirty and label.last_changed and label.semantic_markers[0] == DIRTY_MARKER + assert count.signature_diff and not count.dirty + assert count.semantic_markers == (SIGNATURE_DIFF_MARKER,) + assert (count.field_name, count.container_path, count.raw_value_type) == ("count", "", "int") + assert count.declaring_type.endswith(".Inspected") + assert not count.inherited_value and count.provenance is None + assert state_markers(state) == (DIRTY_MARKER, SIGNATURE_DIFF_MARKER) From 9e32ef3619aef50f817c77252a8ad7cc01035307 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 04:04:08 -0400 Subject: [PATCH 6/8] Release 1.4.0 Added: ScopeKey; objectstate.inspection (FieldInspection, field_markers, state_markers); retain_only_authored_paths; ObjectStateRegistry.MAIN_BRANCH and is_restoring_snapshot(). Removed: HistoryMigration and the migration= parameter of import_history_from_dict/load_history_from_file (a history document is current or refused); get_root_from_scope_key and is_scope_affected (ScopeKey.root and ScopeKey.contains). Co-Authored-By: Claude Opus 5.5 --- src/objectstate/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/objectstate/__init__.py b/src/objectstate/__init__.py index 668ea9f..8f30e71 100644 --- a/src/objectstate/__init__.py +++ b/src/objectstate/__init__.py @@ -212,7 +212,7 @@ 'ObjectStateEditSession', ] -__version__ = "1.3.1" +__version__ = "1.4.0" __author__ = 'OpenHCS Team' __description__ = 'Generic configuration framework for lazy dataclass resolution' From 4317a8ca91aa8a74b341dbea15eacfbb2c786b9e Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 04:34:55 -0400 Subject: [PATCH 7/8] Satisfy ruff on added lines Co-Authored-By: Claude Opus 5.5 --- src/objectstate/object_state_registry.py | 4 ++-- tests/test_inspection.py | 3 ++- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/objectstate/object_state_registry.py b/src/objectstate/object_state_registry.py index a7597c5..6171311 100644 --- a/src/objectstate/object_state_registry.py +++ b/src/objectstate/object_state_registry.py @@ -348,7 +348,7 @@ def mark_snapshot_dirty_scope(cls, scope_id: Optional[str]) -> None: cls._snapshot_dirty_scopes.add(cls._normalize_scope_id(scope_id)) @classmethod - def _normalize_scope_id(cls, scope_id: Optional[str]) -> ScopeKey: + def _normalize_scope_id(cls, scope_id: str | None) -> ScopeKey: """Return the registry key: None and "" both represent the global scope.""" return ScopeKey(scope_id) @@ -2287,7 +2287,7 @@ def export_history_to_dict(cls) -> Dict[str, Any]: } @classmethod - def import_history_from_dict(cls, data: Dict[str, Any]) -> None: + def import_history_from_dict(cls, data: dict[str, Any]) -> None: """Import a complete typed history document payload. Version 2 includes exact construction declarations. Older documents diff --git a/tests/test_inspection.py b/tests/test_inspection.py index 3d6b146..97d5ddc 100644 --- a/tests/test_inspection.py +++ b/tests/test_inspection.py @@ -40,7 +40,8 @@ def test_field_inspection_reports_value_edit_state_and_types(): label = FieldInspection.of(state, "label") count = FieldInspection.of(state, "count") - assert (label.raw_value, label.resolved_value, label.resolved_value_available) == ("y", "y", True) + assert (label.raw_value, label.resolved_value) == ("y", "y") + assert label.resolved_value_available assert label.dirty and label.last_changed and label.semantic_markers[0] == DIRTY_MARKER assert count.signature_diff and not count.dirty assert count.semantic_markers == (SIGNATURE_DIFF_MARKER,) From 35845f8951748086660bbeca4ea206161ec3cc09 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 05:07:12 -0400 Subject: [PATCH 8/8] Release 2.0.0: removes HistoryMigration, the migration= parameter and two unused scope helpers Co-Authored-By: Claude Opus 5.5 --- src/objectstate/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/objectstate/__init__.py b/src/objectstate/__init__.py index 8f30e71..cdb5b35 100644 --- a/src/objectstate/__init__.py +++ b/src/objectstate/__init__.py @@ -212,7 +212,7 @@ 'ObjectStateEditSession', ] -__version__ = "1.4.0" +__version__ = "2.0.0" __author__ = 'OpenHCS Team' __description__ = 'Generic configuration framework for lazy dataclass resolution'