Fix native member topology across edits, saved state and undo

src/objectstate/object_state_registry.py

@@ -879,7 +879,7 @@ def _invalidate_field_in_matching_states(
             logger.debug(f"🔍 _invalidate_field_in_matching_states: state_scope={state.scope_id}, target_base_type={target_base_type.__name__}, field_name={field_name}")
             logger.debug(f"🔍   state.parameter_structure.owner_paths has {len(state.parameter_structure.owner_paths)} entries")
 
-        for dotted_path in state.inheritance_field_paths(
+        for dotted_path in state.parameter_structure.inheritance_field_paths(
             target_base_type,
             field_name,
         ):
@@ -888,7 +888,12 @@ def _invalidate_field_in_matching_states(
             state.invalidate_field(dotted_path)
             invalidated_paths.add(dotted_path)
 
-        if invalidated_paths:
+        saved_invalidated_paths: tuple[str, ...] = ()
+        if invalidate_saved:
+            saved_invalidated_paths = state._saved_structure.inheritance_field_paths(
+                target_base_type, field_name
+            )
+        if invalidated_paths or saved_invalidated_paths:
             if invalidate_saved:
                 # Without recorded sources the saved snapshot is resolved again in full.
                 state._resolved_from.pop(True, None)
@@ -1545,25 +1550,31 @@ def reconcile(current: dict, historical: dict) -> dict:
                 for key, value in current.items()
             }
 
+        live_parameters, live_structure = schema._parameter_epoch(snapshot.parameters)
+        saved_parameters, saved_structure = schema._parameter_epoch(snapshot.saved_parameters)
+
+        live_leaves = {
+            path.value: live_parameters[path.value]
+            for path in live_structure.leaf_paths()
+        }
+        saved_leaves = {
+            path.value: saved_parameters[path.value]
+            for path in saved_structure.leaf_paths()
+        }
+        # Attach historical topology to the detached restoration schema; the
+        # snapshot application below publishes its corresponding values once.
+        schema.parameter_structure = live_structure
+        schema._saved_structure = saved_structure
         return replace(
             snapshot,
-            parameters=reconcile(schema.parameters, snapshot.parameters),
-            saved_parameters=reconcile(
-                schema._saved_parameters,
-                snapshot.saved_parameters,
-            ),
-            saved_resolved=reconcile(
-                schema._saved_resolved,
-                snapshot.saved_resolved,
-            ),
-            live_resolved=reconcile(
-                schema._live_resolved,
-                snapshot.live_resolved,
-            ),
-            provenance=reconcile(
-                schema._live_provenance,
-                snapshot.provenance,
-            ),
+            parameters=live_parameters,
+            saved_parameters=saved_parameters,
+            saved_resolved=reconcile(saved_leaves, snapshot.saved_resolved),
+            live_resolved=reconcile(live_leaves, snapshot.live_resolved),
+            provenance={
+                path: value for path, value in snapshot.provenance.items()
+                if path in live_leaves
+            },
         )
 
     @classmethod
@@ -1844,7 +1855,6 @@ def _apply_time_travel_snapshot_state(
         state._live_resolved = state_snap.live_resolved
         state._live_provenance = state_snap.provenance
         state.parameters = state_snap.parameters
-        state._index_parameter_paths()
         state._saved_parameters = state_snap.saved_parameters
         state._cached_object = None
         state._cached_object_applied = False

src/objectstate/parameter_structure.py

@@ -22,6 +22,7 @@ class ParameterStructure:
         default_factory=dict
     )
     declarations: dict[str, ParameterInfo] = field(default_factory=dict)
+    container_paths: set[DottedFieldPath] = field(default_factory=set)
     exclusions: list[str] = field(default_factory=list)
     excluded_values: dict[str, Any] = field(default_factory=dict)
 
@@ -37,11 +38,40 @@ def descriptions(self) -> dict[str, str | None]:
 
         return {path: info.description for path, info in self.declarations.items()}
 
+    def index(self, parameters: dict[str, Any]) -> None:
+        """Derive direct-child topology from this epoch's admitted values."""
+
+        paths_by_owner: dict[DottedFieldPath, list[DottedFieldPath]] = {}
+        for name in parameters:
+            path = DottedFieldPath(name)
+            paths_by_owner.setdefault(path.parent, []).append(path)
+        self.direct_paths = {
+            owner: tuple(paths) for owner, paths in paths_by_owner.items()
+        }
+
+    def without_descendants(self, prefix: str) -> ParameterStructure:
+        """Prepare a subtree replacement without publishing live metadata."""
+
+        prepared = self.clone()
+        parent = DottedFieldPath(prefix)
+        prepared.declarations = {
+            path: info for path, info in self.declarations.items()
+            if path == prefix or not parent.contains_path(path)
+        }
+        prepared.owner_paths = {
+            path: owner for path, owner in self.owner_paths.items()
+            if path == prefix or not parent.contains_path(path)
+        }
+        prepared.container_paths = {
+            path for path in self.container_paths if not parent.contains_path(path)
+        }
+        return prepared
+
     def path(self, name: str) -> DottedFieldPath:
         """Resolve a name to its admitted path, without analyzing declarations."""
 
         requested = DottedFieldPath(name)
-        for path in self.direct_paths.get(requested.parent, ()):
+        for path in self.children(requested.parent.value):
             if path.value == name:
                 return path
         raise KeyError(f"ObjectState path is not registered: {name!r}")
@@ -51,24 +81,83 @@ def ancestors(self, path: DottedFieldPath) -> tuple[DottedFieldPath, ...]:
 
         return tuple(self.path(step.value) for step in reversed(path.lineage))
 
+    def children(self, parent: str = "") -> tuple[DottedFieldPath, ...]:
+        """Return this epoch's admitted direct fields."""
+
+        return self.direct_paths.get(DottedFieldPath(parent), ())
+
     def leaf_paths(self, parent: str = "") -> Iterator[DottedFieldPath]:
-        """Enumerate leaves through the extracted topology, never the source type."""
+        """Enumerate values, including childless records that also reconstruct."""
 
-        for path in self.direct_paths.get(DottedFieldPath(parent), ()):
-            if self.direct_paths.get(path):
+        for path in self.children(parent):
+            children = self.children(path.value)
+            if children:
                 yield from self.leaf_paths(path.value)
             else:
                 yield path
 
-    def is_container_parameter(
-        self, name: str, value: Any, parameters: dict[str, Any]
-    ) -> bool:
-        """Identify a nested dataclass entry from its actual extracted children."""
+    @property
+    def value_paths(self) -> frozenset[str]:
+        """The flat value surface; populated containers reconstruct their leaves."""
+
+        return frozenset(path.value for path in self.leaf_paths())
+
+    def reconstruction_paths(self) -> tuple[DottedFieldPath, ...]:
+        """Admitted records deepest first, excluding the separately rebuilt root."""
+
+        return tuple(sorted(
+            (path for path in self.container_paths if path.value),
+            key=lambda path: len(path.lineage),
+            reverse=True,
+        ))
+
+    def reconstruction_ancestors(self, name: str) -> Iterator[DottedFieldPath]:
+        """Records containing an edited field, nearest first."""
+
+        for path in DottedFieldPath(name).parent.lineage:
+            if self.is_container_parameter(path.value):
+                yield path
+
+    def inheritance_field_paths(
+        self, target_type: ParameterOwner, field_name: str
+    ) -> tuple[str, ...]:
+        """Find inheriting fields in this structure's epoch, not another one."""
+        from objectstate.lazy_factory import get_base_type_for_lazy
 
-        if value is None or not is_dataclass(type(value)):
+        target_base = (
+            get_base_type_for_lazy(target_type) or target_type
+            if isinstance(target_type, type) else target_type
+        )
+        paths: list[str] = []
+        for path, owner in self.owner_paths.items():
+            if DottedFieldPath(path).field_name != field_name:
+                continue
+            owner_base = (
+                get_base_type_for_lazy(owner) or owner
+                if isinstance(owner, type) else owner
+            )
+            matches = owner_base == target_base
+            if not matches and isinstance(owner_base, type) and isinstance(target_base, type):
+                matches = any(
+                    (get_base_type_for_lazy(base) or base) == target_base
+                    for base in owner_base.__mro__
+                )
+            if matches:
+                paths.append(path)
+        return tuple(paths)
+
+    def admit_container(self, path: str, value: object) -> bool:
+        """Classify foreign values once during extraction, including empty records."""
+
+        if not is_dataclass(type(value)):
             return False
-        prefix = f"{name}."
-        return any(path.startswith(prefix) for path in parameters)
+        self.container_paths.add(DottedFieldPath(path))
+        return True
+
+    def is_container_parameter(self, name: str) -> bool:
+        """Read the container membership recorded by extraction."""
+
+        return DottedFieldPath(name) in self.container_paths
 
     @classmethod
     def for_target(
@@ -77,12 +166,14 @@ def for_target(
         """Start extraction from the actual target's excluded constructor values."""
 
         names = list(exclusions or [])
-        return cls(
+        structure = cls(
             exclusions=names,
             excluded_values={
                 name: getattr(target, name) for name in names if hasattr(target, name)
             },
         )
+        structure.admit_container('', target)
+        return structure
 
     def clone(self) -> ParameterStructure:
         """Detach owned maps without cloning declaration/target identities.

src/objectstate/transaction_checkpoint.py

@@ -29,6 +29,7 @@ class ObjectStateTransactionCheckpoint:
     construction: StateConstructionBinding
     parent_state: ObjectState | None
     structure: ParameterStructure
+    saved_structure: ParameterStructure
     preserve_value_identity: bool
     parameters: dict[str, Any]
     saved_parameters: dict[str, Any]
@@ -74,6 +75,7 @@ def capture(
             construction=state.construction_binding(),
             parent_state=state._parent_state,
             structure=state.parameter_structure.clone(),
+            saved_structure=state._saved_structure.clone(),
             preserve_value_identity=preserve_value_identity,
             parameters=copied["parameters"],
             saved_parameters=copied["saved_parameters"],
@@ -124,10 +126,11 @@ def restore(self, state: ObjectState, *, emit_notifications: bool = True) -> Non
         state._parent_field_name = self.construction.parent_field
         state.parameter_structure = self.structure.clone()
         state.parameters = copied["parameters"]
-        state._index_parameter_paths()
         state._saved_parameters = copied["saved_parameters"]
         state._live_resolved = copied["live_resolved"]
-        state._saved_resolved = copied["saved_resolved"]
+        state._saved_resolved, state._saved_structure = (
+            copied["saved_resolved"], self.saved_structure.clone()
+        )
         state._live_provenance = copied["live_provenance"]
         state.metadata = ObjectStateMetadataStore.from_snapshot(
             scope_id=state.scope_id,

tests/test_construction_history.py

@@ -409,7 +409,7 @@ def test_history_restores_saved_live_global_context_after_real_invalidation(
         assert get_live_global_config(ContextConfig).threshold == live
         child.invalidate_cache()
         assert child.get_resolved_value("threshold") == live
-        assert child._compute_resolved_snapshot(use_saved=True)["threshold"] == 3
+        assert child.get_saved_resolved_value("threshold") == 3
 
 
 def test_failed_attachment_rolls_back_complete_graph_and_loaded_history(tmp_path: Path):