diff --git a/src/objectstate/object_state.py b/src/objectstate/object_state.py
index c00efe7..1ae0afe 100644
--- a/src/objectstate/object_state.py
+++ b/src/objectstate/object_state.py
@@ -12,7 +12,7 @@
 
 import copy
 import logging
-from collections.abc import Callable, Sequence
+from collections.abc import Callable, Iterable, Sequence
 from dataclasses import fields as dataclass_fields
 from dataclasses import is_dataclass
 from types import FunctionType
@@ -231,8 +231,9 @@ def __init__(
         # same objects; a snapshot without an entry is resolved again in full.
         self._resolved_from: dict[bool, ResolutionSources] = {}
 
-        # === Saved baseline (2 attributes) ===
+        # === Saved baseline ===
         self._saved_resolved: dict[str, Any] = {}
+        self._saved_structure = self.parameter_structure.clone()
         self._saved_parameters: dict[str, Any] = {}  # Immutable snapshot for diff on restore
 
         # === Materialized diffs (3 attributes) ===
@@ -271,7 +272,7 @@ def __init__(
         # Initialize baselines (nothing to tell anyone during init)
         self._live_resolved = self._compute_resolved_snapshot()
 
-        # CRITICAL: Initialize _saved_parameters BEFORE _compute_resolved_snapshot(use_saved=True)
+        # Initialize saved raw values before resolving their saved epoch.
         # because that method reads from _saved_parameters to get raw values.
         self._saved_parameters = self._copy_parameters_for_saved_baseline(self.parameters)
 
@@ -282,7 +283,7 @@ def __init__(
         #
         # If we used copy.deepcopy(_live_resolved), we'd bake in ancestor's unsaved
         # edits, causing inverted dirty state when ancestor is saved/reset.
-        self._saved_resolved = self._compute_resolved_snapshot(use_saved=True)
+        self._refresh_saved_resolved()
 
         # DEBUG: Log live vs saved at registration
         logger.debug(f"🔵 INIT_RESOLVED: scope={self.scope_id!r} obj_type={type(self.object_instance).__name__}")
@@ -481,7 +482,8 @@ def to_saved_resolved_object(self) -> Any:
         self._check_and_sync_delegate()
         self._ensure_saved_resolved()
         return self._reconstruct_from_resolved(
-            "", self._saved_resolved, self._extraction_target, self.parameter_structure
+            "", self._saved_resolved, self._extraction_target,
+            self._saved_structure,
         )
 
     @classmethod
@@ -508,14 +510,15 @@ def resolve_saved_object(
         structure = ParameterStructure.for_target(target, None)
         parameters: dict[str, Any] = {}
         cls._extract_all_parameters_flat(target, parameters, structure)
+        structure.index(parameters)
         saved_parameters = cls._copy_parameters_for_saved_baseline(parameters)
         snapshot, provenance = cls._resolve_parameter_snapshot(
-            saved_parameters,
             saved_parameters,
             structure,
             target,
             scope_id,
             ancestor_objects_with_scopes,
+            paths=structure.leaf_paths(),
             use_saved=True,
         )
         # A compilation capture owns inherited mutable values too. The editor's
@@ -560,7 +563,7 @@ def direct_parameter_paths(
         never cached in this projection.
         """
 
-        return self.parameter_structure.direct_paths.get(DottedFieldPath(field_path), ())
+        return self.parameter_structure.children(field_path)
 
     def has_parameter_descendants(self, field_path: str) -> bool:
         """Return whether an authoritative flat parameter owns child paths."""
@@ -758,11 +761,9 @@ def _ensure_live_resolved(self, notify_flash: bool = True) -> set[str]:
             previous = self._live_resolved
             self._live_resolved = self._compute_resolved_snapshot()
             self._invalid_fields.clear()
-            changed_values = {
-                path: (previous.get(path), value)
-                for path, value in self._live_resolved.items()
-                if previous.get(path) != value
-            }
+            changed_values = self._changed_parameter_values(
+                previous, self._live_resolved
+            )
             if changed_values:
                 self._set_last_changed_values(changed_values)
             changed_paths = self.notification_fields_for_value_changes(changed_values)
@@ -848,10 +849,11 @@ def _fields_with_other_sources(
         if differing is None:
             return None
         own_values = self._saved_parameters if use_saved else self.parameters
+        structure = self._saved_structure if use_saved else self.parameter_structure
         return {
             path
             for config_type, field_name in differing
-            for path in self.inheritance_field_paths(config_type, field_name)
+            for path in structure.inheritance_field_paths(config_type, field_name)
             if own_values.get(path) is None
         }
 
@@ -866,7 +868,7 @@ def _resolve_saved_again_if_sources_changed(self, sources: ResolutionSources) ->
         if stale is not None and not stale:
             self._resolved_from[True] = sources
             return False
-        self._saved_resolved = self._compute_resolved_snapshot(use_saved=True)
+        self._refresh_saved_resolved()
         return True
 
     def resolve_again_if_sources_changed(self) -> None:
@@ -883,7 +885,7 @@ def _ensure_saved_resolved(self) -> None:
         if self._live_resolved is not None:
             self._ensure_live_resolved()
         elif not self._saved_resolved:
-            self._saved_resolved = self._compute_resolved_snapshot(use_saved=True)
+            self._refresh_saved_resolved()
 
     # DELETED: _create_nested_states() - No longer needed with flat storage
     # Nested ObjectStates are no longer created - flat storage handles all parameters
@@ -905,37 +907,6 @@ def _analyze_parameters(
 
         return UnifiedParameterAnalyzer.analyze(obj, exclude_params=exclude_params)
 
-    @staticmethod
-    def _get_nested_dataclass_type(param_type: Any) -> type | None:
-        """Get the nested dataclass type if param_type is a nested dataclass.
-
-        Args:
-            param_type: The parameter type to check
-
-        Returns:
-            The dataclass type if nested, None otherwise
-        """
-        from typing import Union, get_args, get_origin
-
-        # Check Optional[dataclass]
-        origin = get_origin(param_type)
-        if origin is Union:
-            args = get_args(param_type)
-            if len(args) == 2 and type(None) in args:
-                inner_type = next(arg for arg in args if arg is not type(None))
-                if is_dataclass(inner_type):
-                    return inner_type
-
-        # Check direct dataclass (but not the type itself)
-        if is_dataclass(param_type) and not isinstance(param_type, type):
-            # param_type is an instance, not a type - shouldn't happen but handle it
-            return type(param_type)
-
-        if is_dataclass(param_type):
-            return param_type
-
-        return None
-
     def reset_all_parameters(self) -> None:
         """Reset all parameters to defaults."""
         self._in_reset = True
@@ -945,98 +916,63 @@ def reset_all_parameters(self) -> None:
         finally:
             self._in_reset = False
 
-    def _dataclass_parameter_updates(self, prefix: str, value: Any) -> dict[str, Any]:
-        """Flatten a dataclass container update into registered child parameters."""
-        if value is None or not is_dataclass(type(value)):
-            return {}
+    def _prepare_parameter_update(
+        self, prefix: str, value: Any
+    ) -> tuple[dict[str, Any], ParameterStructure, object]:
+        """Prepare the subtree and rebuilt root before publishing any edit."""
 
-        prefix_dot = f"{prefix}."
-        if not any(
-            path.startswith(prefix_dot)
-            for path in self.parameters
-            if path != prefix
-        ):
-            return {}
+        structure = self.parameter_structure.without_descendants(prefix)
+        changed = DottedFieldPath(prefix)
+        parameters = {
+            path: current for path, current in self.parameters.items()
+            if path == prefix or not changed.contains_path(path)
+        }
+        parent = changed.parent.value
+        owner = parameters[parent] if parent else self._extraction_target
+        self._admit_parameter(owner, prefix, value, parameters, structure)
+        structure.index(parameters)
+        rebuilt = self._dataclass_ancestor_parameter_updates(prefix, parameters, structure)
+        return parameters, structure, rebuilt
+
+    def _parameter_epoch(
+        self, values: dict[str, Any]
+    ) -> tuple[dict[str, Any], ParameterStructure]:
+        """Derive one raw epoch through the same declaration/value extractor.
+
+        Native containers in values select their own children. The current root
+        declaration still filters genuinely removed constructor fields.
+        """
 
-        updates: dict[str, Any] = {}
-        for field in dataclass_fields(value):
-            field_path = f"{prefix_dot}{field.name}"
-            field_value = DataclassFieldAccess.raw_value(value, field.name)
-            if field_path in self.parameters:
-                updates[field_path] = field_value
-            updates.update(self._dataclass_parameter_updates(field_path, field_value))
-        return updates
+        parameters: dict[str, Any] = {}
+        structure = ParameterStructure.for_target(
+            self._extraction_target, self.parameter_structure.exclusions
+        )
+        self._extract_all_parameters_flat(
+            self._extraction_target, parameters, structure,
+            exclude_params=structure.exclusions, values=values,
+        )
+        structure.index(parameters)
+        for path in structure.reconstruction_paths():
+            parameters[path.value] = self._reconstruct_from_parameter_snapshot(
+                path.value, parameters, structure
+            )
+        return parameters, structure
 
     def _dataclass_ancestor_parameter_updates(
         self,
         param_name: str,
         parameters: dict[str, Any],
-    ) -> dict[str, Any]:
-        """Rebuild dataclass container parameters that own a changed child path."""
-
-        updates: dict[str, Any] = {}
-        for path in DottedFieldPath(param_name).parent.lineage:
-            prefix = path.value
-            if prefix not in parameters:
-                continue
-            prefix_type = self.parameter_structure.owner_paths.get(prefix)
-            if not (isinstance(prefix_type, type) and is_dataclass(prefix_type)):
-                continue
-
-            rebuilt_value = self._reconstruct_from_parameter_snapshot(
-                prefix,
-                parameters,
+        structure: ParameterStructure,
+    ) -> Any:
+        """Rebuild the admitted ancestors, returning the validated root for use."""
+        for path in structure.reconstruction_ancestors(param_name):
+            parameters[path.value] = self._reconstruct_from_parameter_snapshot(
+                path.value, parameters, structure
             )
-            if parameters.get(prefix) != rebuilt_value:
-                updates[prefix] = rebuilt_value
-                parameters[prefix] = rebuilt_value
-        return updates
-
-
-    def inheritance_field_paths(
-        self,
-        target_type: ParameterOwner,
-        field_name: str,
-    ) -> tuple[str, ...]:
-        """Return flat field paths whose container type can inherit target_type."""
-
-        from objectstate.lazy_factory import get_base_type_for_lazy
-
-        if isinstance(target_type, type):
-            target_base_type = get_base_type_for_lazy(target_type) or target_type
-        else:
-            target_base_type = target_type
-
-        paths: list[str] = []
-        for dotted_path, container_type in self.parameter_structure.owner_paths.items():
-            if dotted_path not in self.parameters:
-                continue
-            if not (dotted_path.endswith(f".{field_name}") or dotted_path == field_name):
-                continue
-
-            if isinstance(container_type, type):
-                container_base_type = (
-                    get_base_type_for_lazy(container_type) or container_type
-                )
-            else:
-                container_base_type = container_type
-
-            type_matches = container_base_type == target_base_type
-            if (
-                not type_matches
-                and isinstance(container_base_type, type)
-                and isinstance(target_base_type, type)
-            ):
-                type_matches = any(
-                    (get_base_type_for_lazy(mro_class) or mro_class)
-                    == target_base_type
-                    for mro_class in container_base_type.__mro__
-                )
-
-            if type_matches:
-                paths.append(dotted_path)
+        if structure.is_container_parameter(''):
+            return self._reconstruct_from_parameter_snapshot('', parameters, structure)
+        return None
 
-        return tuple(paths)
 
     def update_parameter(self, param_name: str, value: Any) -> set[str]:
         """Update parameter value in state.
@@ -1067,53 +1003,37 @@ def update_parameter(self, param_name: str, value: Any) -> set[str]:
             )
             return set()
 
-        child_updates = self._dataclass_parameter_updates(param_name, value)
-        pending_parameters = dict(self.parameters)
-        pending_parameters[param_name] = value
-        for child_path, child_value in child_updates.items():
-            pending_parameters[child_path] = child_value
-        ancestor_updates = self._dataclass_ancestor_parameter_updates(
-            param_name,
-            pending_parameters,
+        pending_parameters, pending_structure, pending_object = (
+            self._prepare_parameter_update(param_name, value)
         )
-
-        # EARLY EXIT: No change, no invalidation, no flash
-        current_value = self.parameters[param_name]
-        changed_child_params = {
-            child_path
-            for child_path, child_value in child_updates.items()
-            if self.parameters.get(child_path) != child_value
-        }
-        changed_ancestor_params = {
-            ancestor_path
-            for ancestor_path, ancestor_value in ancestor_updates.items()
-            if self.parameters.get(ancestor_path) != ancestor_value
-        }
-        changed_storage_params = set(changed_child_params) | changed_ancestor_params
-        if current_value != value:
-            changed_storage_params.add(param_name)
-        if not changed_storage_params:
+        previous_structure = self.parameter_structure
+        changed_storage_values = self._changed_parameter_values(
+            self.parameters, pending_parameters
+        )
+        changed_storage_params = changed_storage_values.keys()
+        topology_changed = (
+            self.parameters.keys() != pending_parameters.keys()
+            or previous_structure.owner_paths != pending_structure.owner_paths
+        )
+        if not changed_storage_params and not topology_changed:
             return set()
+        value_paths = previous_structure.value_paths | pending_structure.value_paths
+        changed_semantic_params = {
+            path for path in changed_storage_params & value_paths
+            if DottedFieldPath(param_name).contains_path(path)
+        }
 
-        changed_semantic_params = set(changed_child_params)
-        if current_value != value and not changed_child_params:
-            changed_semantic_params.add(param_name)
-
-        changed_param_values = self._changed_parameter_values(
-            changed_semantic_params,
-            pending_parameters,
-        )
+        changed_param_values = {
+            path: changed_storage_values[path] for path in changed_semantic_params
+        }
         # Establish the pre-mutation resolved authority even for a state that
         # has not yet been displayed. Raw edits can preserve the resolved value.
         changed_before_the_edit = self._ensure_live_resolved(notify_flash=False)
         ObjectStateRegistry.ensure_baseline_snapshot()
 
-        # Update state directly (no type conversion - that's VIEW responsibility)
-        self.parameters[param_name] = value
-        for child_path, child_value in child_updates.items():
-            self.parameters[child_path] = child_value
-        for ancestor_path, ancestor_value in ancestor_updates.items():
-            self.parameters[ancestor_path] = ancestor_value
+        # Publish values and topology together; saved values remain their own epoch.
+        self.parameters = pending_parameters
+        self.parameter_structure = pending_structure
 
         # SELF-INVALIDATION: Mark changed fields and same-state inherited siblings.
         local_invalidations = set(changed_semantic_params)
@@ -1121,15 +1041,21 @@ def update_parameter(self, param_name: str, value: Any) -> set[str]:
             container_type = self.parameter_structure.owner_paths.get(changed_param, type(self.object_instance))
             leaf_field_name = DottedFieldPath(changed_param).field_name
             local_invalidations.update(
-                self.inheritance_field_paths(container_type, leaf_field_name)
+                self.parameter_structure.inheritance_field_paths(container_type, leaf_field_name)
             )
         self._invalid_fields.update(local_invalidations)
-        self._cached_object = None  # Invalidate cached reconstructed object
+        self._cached_object = pending_object  # Reuse the root validated during preparation
         self._cached_object_applied = False
 
         # Materialize the origin once after its live context is authoritative,
         # then cascade only to descendants.
-        changed_paths = self._ensure_live_resolved(notify_flash=False) | changed_before_the_edit
+        if topology_changed:
+            # The old sources described another field surface. Keep their values
+            # for the diff; the refresh owner retires paths and provenance.
+            self._resolved_from.pop(False, None)
+        changed_paths = self._ensure_live_resolved(notify_flash=False)
+        changed_paths |= changed_before_the_edit
+        changed_paths = self._most_specific_notification_fields(changed_paths)
         for changed_param in changed_semantic_params:
             # SCOPE + TYPE + FIELD AWARE INVALIDATION:
             # Get the CONTAINER type for this field (e.g., PartitionFilterConfig for
@@ -1137,18 +1063,18 @@ def update_parameter(self, param_name: str, value: Any) -> set[str]:
             # This is critical for sibling inheritance: when
             # PartitionFilterConfig.partition_filter changes, we need to invalidate
             # PathPlanningConfig.partition_filter (which inherits from PartitionFilterConfig)
-            container_type = self.parameter_structure.owner_paths.get(changed_param, type(self.object_instance))
-
-            # Extract leaf field name for invalidation matching
-            leaf_field_name = DottedFieldPath(changed_param).field_name
-
-
-            ObjectStateRegistry.invalidate_by_type_and_scope(
-                scope_id=self.scope_id,
-                changed_type=container_type,
-                field_name=leaf_field_name,
-                include_origin=False,
-            )
+            owners = {
+                structure.owner_paths[changed_param]
+                for structure in (previous_structure, pending_structure)
+                if changed_param in structure.owner_paths
+            }
+            for container_type in owners:
+                ObjectStateRegistry.invalidate_by_type_and_scope(
+                    scope_id=self.scope_id,
+                    changed_type=container_type,
+                    field_name=DottedFieldPath(changed_param).field_name,
+                    include_origin=False,
+                )
 
         # Increment global token for LiveContextService.collect() cache invalidation
         ObjectStateRegistry.increment_token(notify=False)
@@ -1164,15 +1090,10 @@ def update_parameter(self, param_name: str, value: Any) -> set[str]:
         # Leaf edits record directly. Dataclass-container edits record when they
         # materialize changed flat child fields; those child paths are the UI and
         # time-travel navigation authority for inline editors.
-        # A field is a container if there are other params that start with "param_name."
-        param_path = DottedFieldPath(param_name)
-        is_container = any(
-            p != param_name and param_path.contains_path(p)
-            for p in self.parameters.keys()
-        )
         snapshot_field = (
             self._select_changed_field(changed_semantic_params)
-            if is_container and changed_semantic_params
+            if self.parameter_structure.is_container_parameter(param_name)
+            and changed_semantic_params
             else param_name
         )
         if snapshot_field:
@@ -1197,22 +1118,10 @@ def get_resolved_value(self, param_name: str) -> Any:
         self._ensure_live_resolved()
         assert self._live_resolved is not None  # Guaranteed by _ensure_live_resolved
 
-        # Check if this is a container/dataclass field (has subfields in _live_resolved)
-        prefix = f"{param_name}."
-        has_subfields = any(key.startswith(prefix) for key in self._live_resolved.keys())
-
-        if has_subfields:
-            # This is a container field - reconstruct's dataclass from live resolved values
-            field_type = self.parameter_structure.owner_paths.get(param_name)
-            if field_type is not None and is_dataclass(field_type):
-                return self._reconstruct_from_resolved(
-                    param_name, self._live_resolved, self._extraction_target, self.parameter_structure
-                )
-
-        result = self._live_resolved.get(param_name)
-
-
-        return result
+        return self._reconstruct_from_resolved(
+            param_name, self._live_resolved, self._extraction_target,
+            self.parameter_structure,
+        )
 
     def get_saved_resolved_value(self, param_name: str) -> Any:
         """Get saved resolved value for a field from the saved snapshot.
@@ -1237,20 +1146,10 @@ def get_saved_resolved_value(self, param_name: str) -> Any:
 
         self._ensure_saved_resolved()
 
-        # Check if this is a container/dataclass field (has subfields in _saved_resolved)
-        prefix = f"{param_name}."
-        has_subfields = any(key.startswith(prefix) for key in self._saved_resolved.keys())
-
-        if has_subfields:
-            # This is a container field - reconstruct's dataclass from saved resolved values
-            field_type = self.parameter_structure.owner_paths.get(param_name)
-            if field_type is not None and is_dataclass(field_type):
-                return self._reconstruct_from_resolved(
-                    param_name, self._saved_resolved, self._extraction_target, self.parameter_structure
-                )
-
-        # Return the simple value (or None if not found)
-        return self._saved_resolved.get(param_name)
+        return self._reconstruct_from_resolved(
+            param_name, self._saved_resolved, self._extraction_target,
+            self._saved_structure,
+        )
 
     def subfield_semantics(
         self,
@@ -1310,6 +1209,9 @@ def _reconstruct_from_resolved(
         """
         from objectstate.lazy_factory import get_base_type_for_lazy
 
+        if prefix and not structure.is_container_parameter(prefix):
+            return resolved_snapshot.get(prefix)
+
         # Determine the type to reconstruct
         if not prefix:
             obj_type = type(target)
@@ -1321,34 +1223,15 @@ def _reconstruct_from_resolved(
         # Normalize to base type for lazy dataclasses
         obj_type = get_base_type_for_lazy(obj_type) or obj_type
 
-        prefix_dot = f'{prefix}.' if prefix else ''
-
-        # Collect direct fields and nested prefixes from resolved snapshot
-        direct_fields = {}
-        nested_prefixes = set()
-
-        for path, value in resolved_snapshot.items():
-            if not path.startswith(prefix_dot):
-                continue
-
-            remainder = path[len(prefix_dot):]
-
-            if '.' in remainder:
-                # This is a nested field - collect the first component
-                first_component = remainder.split('.')[0]
-                nested_prefixes.add(first_component)
-            else:
-                # Direct field of this object
-                direct_fields[remainder] = value
-
-        # Reconstruct nested dataclasses first
-        for nested_name in nested_prefixes:
-            nested_path = f'{prefix_dot}{nested_name}'
-            nested_obj = cls._reconstruct_from_resolved(
-                nested_path, resolved_snapshot, target, structure
+        direct_fields = {
+            path.field_name: cls._reconstruct_from_resolved(
+                path.value, resolved_snapshot, target, structure
             )
-            direct_fields[nested_name] = nested_obj
+            for path in structure.children(prefix)
+        }
 
+        if not prefix:
+            direct_fields.update(structure.excluded_values)
         # Instantiate the dataclass with all resolved fields
         result = obj_type(**direct_fields)
 
@@ -1537,7 +1420,10 @@ def _select_changed_field(changed_paths: set[str]) -> str | None:
         """Choose a stable representative changed path for navigation."""
         if not changed_paths:
             return None
-        return sorted(changed_paths, key=lambda path: (-path.count("."), path))[0]
+        return min(
+            changed_paths,
+            key=lambda path: (-len(DottedFieldPath(path).lineage), path),
+        )
 
     def _set_last_changed_values(
         self,
@@ -1555,22 +1441,20 @@ def _set_last_changed_values(
             self._last_changed_paths
         )
 
+    @staticmethod
     def _changed_parameter_values(
-        self,
-        changed_params: set[str],
-        pending_parameters: dict[str, Any],
+        previous: dict[str, Any],
+        current: dict[str, Any],
     ) -> dict[str, tuple[Any, Any]]:
-        """Return old/new values for changed flat parameter paths."""
+        """Compare raw or resolved epochs without conflating absence with None."""
 
-        changed_values: dict[str, tuple[Any, Any]] = {}
-        for param_path in pending_parameters:
-            if param_path not in changed_params:
-                continue
-            old_value = self.parameters.get(param_path)
-            new_value = pending_parameters.get(param_path)
-            if old_value != new_value:
-                changed_values[param_path] = (old_value, new_value)
-        return changed_values
+        return {
+            path: (previous.get(path), current.get(path))
+            for path in previous.keys() | current.keys()
+            if path not in previous or path not in current
+            or previous[path] != current[path]
+            or type(previous[path]) is not type(current[path])
+        }
 
     def notification_fields_for_value_changes(
         self,
@@ -1579,9 +1463,9 @@ def notification_fields_for_value_changes(
         """Return ObjectState display paths affected by concrete value changes."""
 
         fields: set[str] = set()
-        for param_path, (old_value, new_value) in changed_values.items():
-            if self.parameter_structure.is_container_parameter(param_path, new_value, self.parameters):
-                continue
+        containers = self.parameters.keys() - self.parameter_structure.value_paths
+        for param_path in changed_values.keys() - containers:
+            old_value, new_value = changed_values[param_path]
             fields.add(param_path)
             fields.update(
                 changed_structural_leaf_paths(
@@ -1643,6 +1527,7 @@ def apply_prepared_construction(
         self._parent_state = parent
         self._parent_field_name = binding.parent_field
         self.parameter_structure = prepared.parameter_structure.clone()
+        self._saved_structure = prepared._saved_structure.clone()
         self._cached_object = None
         self._cached_object_applied = False
         self._invalid_fields.clear()
@@ -1702,8 +1587,8 @@ def update_object_instance(self, new_instance: Any) -> None:
         self._cached_object_applied = False
         self._invalid_fields.clear()
 
-        self._live_resolved = self._compute_resolved_snapshot(use_saved=False)
-        self._saved_resolved = self._compute_resolved_snapshot(use_saved=True)
+        self._live_resolved = self._compute_resolved_snapshot()
+        self._refresh_saved_resolved()
 
         missing = object()
         changed_paths = {
@@ -1757,109 +1642,30 @@ def _recompute_invalid_fields(self, sources: ResolutionSources) -> set[str]:
         Returns:
             Set of paths whose resolved values actually changed (for UI notification).
         """
-        from objectstate.context_manager import build_context_stack
-
-        changed_paths: set[str] = set()
-        changed_values: dict[str, tuple[Any, Any]] = {}
-
-        # _live_resolved must exist when this is called (from _ensure_live_resolved)
-        if self._live_resolved is None:
-            return changed_paths
-
-        # Separate explicit vs inherited fields, skipping container entries
-        explicit_fields = []
-        inherited_fields = []
-        for name in self._invalid_fields:
-            if name not in self.parameters:
-                continue
-            # Safety check: skip any container entries that might have leaked in
-            # (containers should NOT be in parameters — only leaf fields are tracked)
-            raw_value = self.parameters[name]
-            if self.parameter_structure.is_container_parameter(name, raw_value, self.parameters):
-                continue
-            if raw_value is not None:
-                explicit_fields.append(name)
-            else:
-                inherited_fields.append(name)
-
-        # Explicit values: use parameters directly (no resolution needed)
-        for name in explicit_fields:
-            old_val = self._live_resolved.get(name)
-            explicit_val = self.parameters[name]
-            if old_val != explicit_val:
-                changed_paths.add(name)
-                changed_values[name] = (old_val, explicit_val)
-                logger.debug(
-                    f"RECOMPUTE EXPLICIT CHANGED [{self.scope_id}] {name}: "
-                    f"old={old_val!r} -> new={explicit_val!r}"
-                )
-            self._live_resolved[name] = explicit_val
-            # Clear provenance for explicit values - they're no longer inherited
-            if name in self._live_provenance:
-                del self._live_provenance[name]
-
-        # Inherited values: need context stack for lazy resolution + provenance
-        if inherited_fields:
-            from objectstate.dual_axis_resolver import resolve_with_provenance
-            from objectstate.lazy_factory import has_lazy_resolution
-
-            # ARCHITECTURAL FIX: Reconstruct current object from parameters for real-time MRO.
-            # Using object_instance (original saved state) gives stale values, breaking inheritance.
-            # Calling to_object() with sync_delegate=False reconstructs from CURRENT parameters
-            # without triggering delegate sync that would invalidate our cache mid-computation.
-            current_obj = self.to_object(update_delegate=False, sync_delegate=False)
-
-            with build_context_stack(
-                object_instance=current_obj,
-                ancestor_objects_with_scopes=sources.ancestors,
-                current_scope_id=self.scope_id,
-            ):
-                # For each inherited field, resolve using dual-axis resolution with provenance
-                for dotted_path in inherited_fields:
-                    container_type = self.parameter_structure.owner_paths.get(dotted_path)
-                    if container_type is None:
-                        logger.debug(f"⚠️ _recompute: {dotted_path} has no container_type in _path_to_type")
-                        continue
-                    # Non-lazy fields with None stay as None rather than
-                    # participating in contextual inheritance.
-                    if not is_dataclass(container_type) or not has_lazy_resolution(container_type):
-                        # Non-lazy field: just use raw value (None)
-                        logger.debug(f"⚠️ _recompute: {dotted_path} has non-lazy container_type={container_type.__name__}, using raw value=None")
-                        old_val = self._live_resolved.get(dotted_path)
-                        raw_val = self.parameters.get(dotted_path)
-                        if old_val != raw_val:
-                            changed_paths.add(dotted_path)
-                            changed_values[dotted_path] = (old_val, raw_val)
-                        self._live_resolved[dotted_path] = raw_val
-                        continue
-                    field_name = DottedFieldPath(dotted_path).field_name
-
-                    # Use resolve_with_provenance for SINGLE walk that gets both value AND source
-                    value, source_scope_id, source_type = resolve_with_provenance(container_type, field_name)
-
-                    old_val = self._live_resolved.get(dotted_path)
-                    if old_val != value:
-                        changed_paths.add(dotted_path)
-                        changed_values[dotted_path] = (old_val, value)
-                        logger.debug(
-                            f"RECOMPUTE INHERITED CHANGED [{self.scope_id}] {dotted_path}: "
-                            f"old={old_val!r} -> new={value!r}"
-                        )
-                    else:
-                        logger.debug(
-                            f"RECOMPUTE INHERITED UNCHANGED [{self.scope_id}] {dotted_path}: "
-                            f"old={old_val!r} == new={value!r}"
-                        )
-                    self._live_resolved[dotted_path] = value
-
-                    # Update provenance for this field
-                    self._live_provenance[dotted_path] = (source_scope_id, source_type)
-
-        # Store for navigation only when recomputation observed concrete deltas.
-        # First-populate paths legitimately have no resolved delta; preserving the
-        # raw update_parameter() paths keeps time-travel/flash navigation precise.
-        if changed_values:
-            self._set_last_changed_values(changed_values)
+        assert self._live_resolved is not None
+        paths = tuple(
+            path for path in self.parameter_structure.leaf_paths()
+            if path.value in self._invalid_fields
+        )
+        previous = {
+            path.value: self._live_resolved[path.value]
+            for path in paths if path.value in self._live_resolved
+        }
+        resolved, provenance = self._resolve_parameter_snapshot(
+            self.parameters,
+            self.parameter_structure,
+            self.to_object(update_delegate=False, sync_delegate=False),
+            self.scope_id,
+            sources.ancestors,
+            paths=paths,
+            use_saved=False,
+        )
+        changed_values = self._changed_parameter_values(previous, resolved)
+        self._live_resolved.update(resolved)
+        for path in paths:
+            self._live_provenance.pop(path.value, None)
+        self._live_provenance.update(provenance)
+        self._set_last_changed_values(changed_values)
         return self.notification_fields_for_value_changes(changed_values)
 
     @property
@@ -1908,28 +1714,20 @@ def reconstruct_top_level_parameters(self) -> dict[str, Any]:
         widgets or service code to reach into ObjectState private helpers.
         """
         self._check_and_sync_delegate()
+        return self._top_level_values()
 
-        values: dict[str, Any] = {}
-        root_type = type(self._extraction_target)
-        top_level_params = {
-            key
-            for key in self.parameters
-            if "." not in key
-        }
-
-        for param_name in top_level_params:
-            param_type = self.parameter_structure.owner_paths.get(param_name)
-            nested_dataclass = (
-                param_type is not None
-                and is_dataclass(param_type)
-                and param_type != root_type
+    def _top_level_values(self) -> dict[str, Any]:
+        """Reconstruct admitted values without synchronizing a delegate."""
+        return {
+            path.value: (
+                self._reconstruct_from_parameter_snapshot(
+                    path.value, self.parameters, self.parameter_structure
+                )
+                if self.parameter_structure.is_container_parameter(path.value)
+                else self.parameters[path.value]
             )
-            if nested_dataclass:
-                values[param_name] = self._reconstruct_from_prefix(param_name)
-            else:
-                values[param_name] = self.parameters[param_name]
-
-        return values
+            for path in self.parameter_structure.children()
+        }
 
     # ==================== MATERIALIZED DIFFS ====================
 
@@ -1985,7 +1783,10 @@ def _normalize_callable_value(value: Any) -> Any:
             live_val = _normalize_callable_value(self._live_resolved.get(k))
             saved_val = _normalize_callable_value(self._saved_resolved.get(k))
 
-            if live_val != saved_val:
+            if (
+                k not in self._live_resolved or k not in self._saved_resolved
+                or live_val != saved_val
+            ):
                 dirty.add(k)
                 logger.debug(f"🔴 DIRTY_FIELD: scope={self.scope_id!r} field={k!r} live={live_val!r} saved={saved_val!r}")
         if dirty:
@@ -2088,104 +1889,91 @@ def prepare_for_snapshot_capture(self) -> None:
 
     # ==================== SAVED STATE / DIRTY TRACKING ====================
 
-    def _compute_resolved_snapshot(self, use_saved: bool = False) -> dict[str, Any]:
-        """Resolve editor fields using the selected live or saved ancestor epoch."""
+    def _compute_resolved_snapshot(self) -> dict[str, Any]:
+        """Resolve the current live graph through its admitted topology."""
         if not isinstance(self.parameters, dict):
-            raise TypeError("ObjectState.parameters must remain a canonical flat dictionary.")
-        if not isinstance(self._saved_parameters, dict):
-            raise TypeError("ObjectState._saved_parameters must remain a canonical flat dictionary.")
-        sources = self._resolution_sources(sync_delegates=True)[use_saved]
-        self._resolved_from[use_saved] = sources
+            raise TypeError(
+                "ObjectState.parameters must remain a canonical flat dictionary."
+            )
+        snapshot, self._live_provenance = self._resolve_epoch(
+            self.parameters, self.parameter_structure,
+            self.to_object(update_delegate=False), use_saved=False,
+        )
+        return snapshot
+
+    def _refresh_saved_resolved(self) -> None:
+        """Resolve and publish saved values and topology as one epoch."""
+        parameters, structure = self._parameter_epoch(self._saved_parameters)
         current_obj = (
-            self.saved_object if use_saved else self.to_object(update_delegate=False)
+            self._reconstruct_from_parameter_snapshot('', parameters, structure)
+            if structure.is_container_parameter('') else self.saved_object
         )
-        parameters = self._saved_parameters if use_saved else self.parameters
-        snapshot, provenance = self._resolve_parameter_snapshot(
-            self.parameters,
-            parameters,
-            self.parameter_structure,
-            current_obj,
-            self.scope_id,
-            sources.ancestors,
-            use_saved=use_saved,
+        snapshot, _ = self._resolve_epoch(
+            parameters, structure, current_obj, use_saved=True
         )
-        if not use_saved:
-            self._live_provenance = provenance
-        return snapshot
+        self._saved_resolved, self._saved_structure = snapshot, structure
+
+    def _resolve_epoch(
+        self,
+        parameters: dict[str, Any],
+        structure: ParameterStructure,
+        current_obj: object,
+        *,
+        use_saved: bool,
+    ) -> tuple[dict[str, Any], dict[str, tuple[str | None, type | None]]]:
+        """Resolve either graph against its corresponding ancestor sources."""
+        sources = self._resolution_sources(sync_delegates=True)[use_saved]
+        resolved = self._resolve_parameter_snapshot(
+            parameters, structure, current_obj, self.scope_id,
+            sources.ancestors, paths=structure.leaf_paths(), use_saved=use_saved,
+        )
+        self._resolved_from[use_saved] = sources
+        return resolved
 
     @classmethod
     def _resolve_parameter_snapshot(
         cls,
         parameters: dict[str, Any],
-        values: dict[str, Any],
         structure: ParameterStructure,
         current_obj: Any,
         scope_id: str | None,
         ancestor_objects_with_scopes: Sequence[tuple[str, Any]],
         *,
+        paths: Iterable[DottedFieldPath],
         use_saved: bool,
     ) -> tuple[dict[str, Any], dict[str, tuple[str | None, type | None]]]:
-        """Resolve raw fields and their provenance in one scoped semantic epoch."""
+        """Resolve admitted values in one epoch, shared by full and partial refresh."""
         from objectstate.context_manager import build_context_stack
         from objectstate.dual_axis_resolver import resolve_with_provenance
         from objectstate.lazy_factory import has_lazy_resolution
 
         snapshot: dict[str, Any] = {}
         provenance: dict[str, tuple[str | None, type | None]] = {}
+        inherited: list[DottedFieldPath] = []
+        for path in paths:
+            value = parameters[path.value]
+            snapshot[path.value] = value
+            owner = structure.owner_paths[path.value]
+            # Extraction supplies the foreign declaration. Only lazy dataclass
+            # None fields inherit; ordinary None and concrete values stay native.
+            if value is None and is_dataclass(owner) and has_lazy_resolution(owner):
+                inherited.append(path)
+
+        if not inherited:
+            return snapshot, provenance
 
-        # UNIFIED: Resolve ALL fields in single context stack
-        # For each path, check if it has a lazy dataclass container type
         with build_context_stack(
             object_instance=current_obj,
             ancestor_objects_with_scopes=ancestor_objects_with_scopes,
             current_scope_id=scope_id,
             use_live=not use_saved,
         ):
-            for dotted_path in parameters.keys():
-                raw_value = values.get(dotted_path)
-                container_type = structure.owner_paths.get(dotted_path)
-
-                # Check if this path is a CONTAINER entry (value is a nested dataclass)
-                # vs a LEAF field (value is primitive, even if container_type is a dataclass)
-                if structure.is_container_parameter(dotted_path, raw_value, parameters):
-                    # Container-level entry - SKIP from snapshot
-                    # Containers are kept in parameters for UI rendering but excluded from
-                    # dirty comparison since we compare leaf fields instead
-                    pass
-                elif (
-                    container_type is not None
-                    and is_dataclass(container_type)
-                    and has_lazy_resolution(container_type)
-                ):
-                    # Leaf field inside a LAZY dataclass - resolve value AND provenance in ONE walk
-                    # CRITICAL: Only resolve for lazy dataclasses! Non-lazy dataclasses with None
-                    # defaults should keep None as-is, not trigger inheritance resolution.
-                    # This handles both:
-                    # - Nested fields (processing_config.group_by)
-                    # - Top-level fields on root (num_workers on PipelineConfig)
-                    field_name = DottedFieldPath(dotted_path).field_name
-
-                    if raw_value is None:
-                        # Field needs resolution - use combined resolve + provenance walk
-                        resolved_val, source_scope, source_type = resolve_with_provenance(container_type, field_name)
-                        snapshot[dotted_path] = resolved_val
-
-                        # Keep provenance in the same epoch as the resolved value.
-                        provenance[dotted_path] = (source_scope, source_type)
-                    else:
-                        # Field has concrete local value - no resolution needed
-                        resolved_val = raw_value
-                        snapshot[dotted_path] = resolved_val
-
-                    logger.debug(
-                        f"SNAPSHOT [{scope_id}] {dotted_path}: "
-                        f"raw={raw_value!r} -> resolved={resolved_val!r} (type={type(resolved_val).__name__})"
-                    )
-                else:
-                    # Non-lazy field (regular dataclass, class instance, callable) - use raw value directly
-                    # None stays as None, no inheritance resolution
-                    snapshot[dotted_path] = raw_value
-
+            for path in inherited:
+                value, source_scope, source_type = resolve_with_provenance(
+                    structure.owner_paths[path.value], path.field_name
+                )
+                snapshot[path.value] = value
+                provenance[path.value] = (source_scope, source_type)
         return snapshot, provenance
 
     def mark_saved(self) -> None:
@@ -2211,13 +1999,8 @@ def mark_saved(self) -> None:
         old_instance_values = {}
         if not isinstance(self.object_instance, type):
             # Extract raw attribute values from the old object_instance
-            for param_name in self.parameters.keys():
-                # Skip container entries (nested dataclass instances)
-                if param_name in self.parameters:
-                    raw_value = self.parameters.get(param_name)
-                    if self.parameter_structure.is_container_parameter(param_name, raw_value, self.parameters):
-                        continue
-
+            for path in self.parameter_structure.leaf_paths():
+                param_name = path.value
                 # Get the old value by navigating dotted path on the extraction target
                 # For delegation, parameters are on the delegate, not object_instance
                 try:
@@ -2234,12 +2017,8 @@ def mark_saved(self) -> None:
         # Find parameters that differ between old object_instance and new live parameters
         # These are the fields that changed and need descendant invalidation
         changed_params = []
-        for param_name in self.parameters.keys():
-            # Skip container entries
-            raw_value = self.parameters.get(param_name)
-            if self.parameter_structure.is_container_parameter(param_name, raw_value, self.parameters):
-                continue
-
+        for path in self.parameter_structure.leaf_paths():
+            param_name = path.value
             old_value = old_instance_values.get(param_name)
             new_value = self.parameters.get(param_name)
             if old_value != new_value:
@@ -2280,11 +2059,8 @@ def mark_saved(self) -> None:
         # Compute new saved resolved using SAVED ancestor baselines (use_saved=True)
         # This ensures saved baseline is computed relative to other saved baselines
         logger.debug(f"🔧 mark_saved: Computing new saved_resolved for scope={self.scope_id!r}")
-        new_saved_resolved = self._compute_resolved_snapshot(use_saved=True)
-        logger.debug(f"🔧 mark_saved: New saved_resolved computed, keys={list(new_saved_resolved.keys())[:5]}...")
-
-        # Update saved resolved baseline
-        self._saved_resolved = new_saved_resolved
+        self._refresh_saved_resolved()
+        logger.debug(f"🔧 mark_saved: New saved epoch computed, keys={list(self._saved_resolved.keys())[:5]}...")
 
         # Invalidate cached object so next to_object() call rebuilds
         self._cached_object = None
@@ -2321,7 +2097,7 @@ def mark_saved(self) -> None:
                 logger.debug(f"🔧 mark_saved: BEFORE recompute - dirty_fields={state._dirty_fields}, _saved_resolved_keys={list(state._saved_resolved.keys())[:3]}")
                 # Recompute descendant's saved_resolved using new ancestor saved values
                 try:
-                    state._saved_resolved = state._compute_resolved_snapshot(use_saved=True)
+                    state._refresh_saved_resolved()
                 except AttributeError as e:
                     logger.error(f"🐛 mark_saved descendant: ERROR in _compute_resolved_snapshot for scope={descendant_scope!r}! Error: {e}")
                     logger.error(f"🐛 mark_saved descendant: state._saved_parameters type={type(state._saved_parameters).__name__}, is None={state._saved_parameters is None}")
@@ -2407,12 +2183,12 @@ def _restore_saved_impl(self, *, propagate_descendants: bool = True) -> None:
 
         # CRITICAL: Recompute _saved_resolved to match the restored state
         # Time travel may have overwritten _saved_resolved with snapshot values,
-        # but after restore_saved(), _saved_resolved should reflect object_instance
-        self._saved_resolved = self._compute_resolved_snapshot(use_saved=True)
+        # but after restore_saved(), the saved epoch should reflect object_instance
+        self._refresh_saved_resolved()
 
         # Recompute _live_resolved to reflect any unsaved changes from higher-level ObjectStates
         # This ensures that unsaved resolved changes from other windows are preserved
-        self._live_resolved = self._compute_resolved_snapshot(use_saved=False)
+        self._live_resolved = self._compute_resolved_snapshot()
 
         # NOW invalidate descendant caches for each changed parameter
         # This must happen AFTER restoring parameters so descendants see restored values
@@ -2485,6 +2261,7 @@ def _extract_all_parameters_flat(
         structure: ParameterStructure,
         prefix: str = '',
         exclude_params: list[str] | None = None,
+        values: dict[str, Any] | None = None,
     ) -> None:
         """Recursively extract parameters into flat dict with dotted paths.
 
@@ -2500,7 +2277,6 @@ def _extract_all_parameters_flat(
 
         obj_type = type(obj)
         is_function = isinstance(obj, FunctionType)
-        owner_target: ParameterOwner = obj if is_function else obj_type
 
         # Delegate signature default extraction to python_introspect (it must derive
         # defaults from the type/signature and avoid instance attribute reads).
@@ -2528,34 +2304,35 @@ def _extract_all_parameters_flat(
                 except AttributeError:
                     current_value = info.default_value
 
+            if values is not None and dotted_path in values:
+                current_value = values[dotted_path]
             structure.declarations[dotted_path] = info
 
-            # Check if this is a nested dataclass
-            # First try from type annotation, then fall back to checking actual value
-            nested_type = cls._get_nested_dataclass_type(info.param_type)
-
-            # The actual dataclass owns both its flattened fields and their
-            # reconstruction, including concrete nominal subtypes and lazy
-            # overlays. An annotation supplies the owner only without a value.
-            if current_value is not None and is_dataclass(type(current_value)):
-                nested_type = type(current_value)
-
-            if nested_type is not None and current_value is not None:
-                # Store the nested config type reference at this path
-                structure.owner_paths[dotted_path] = nested_type
+            cls._admit_parameter(
+                obj, dotted_path, current_value, parameters, structure, values=values
+            )
 
-                # Store the nested dataclass instance in parameters (needed for UI rendering)
-                parameters[dotted_path] = current_value
+    @classmethod
+    def _admit_parameter(
+        cls,
+        owner: object,
+        path: str,
+        value: object,
+        parameters: dict[str, Any],
+        structure: ParameterStructure,
+        *,
+        values: dict[str, Any] | None = None,
+    ) -> None:
+        """One admission operation for extraction and native subtree replacement."""
 
-                # Recurse into nested dataclass for child fields
-                cls._extract_all_parameters_flat(
-                    current_value, parameters, structure, prefix=dotted_path, exclude_params=[]
-                )
-            else:
-                # Leaf field - store value and container type
-                parameters[dotted_path] = current_value
-                # Store the owner target that has this field.
-                structure.owner_paths[dotted_path] = owner_target
+        parameters[path] = value
+        if structure.admit_container(path, value):
+            structure.owner_paths[path] = type(value)
+            cls._extract_all_parameters_flat(
+                value, parameters, structure, prefix=path, values=values,
+            )
+        else:
+            structure.owner_paths[path] = owner if isinstance(owner, FunctionType) else type(owner)
 
     def _replace_parameter_structure(
         self,
@@ -2573,27 +2350,9 @@ def _replace_parameter_structure(
             self.parameters,
             self.parameter_structure,
             prefix="",
-            exclude_params=exclude_params,
+            exclude_params=exclude_params, values=initial_values,
         )
-        if initial_values:
-            for param_name, value in initial_values.items():
-                self.parameters[param_name] = value
-                self.parameters.update(
-                    self._dataclass_parameter_updates(param_name, value)
-                )
-        self._index_parameter_paths()
-
-    def _index_parameter_paths(self) -> None:
-        """Index immutable direct-child topology from canonical flat parameters."""
-
-        paths_by_owner: dict[DottedFieldPath, list[DottedFieldPath]] = {}
-        for dotted_path in self.parameters:
-            path = DottedFieldPath(dotted_path)
-            paths_by_owner.setdefault(path.parent, []).append(path)
-        self.parameter_structure.direct_paths = {
-            owner_path: tuple(parameter_paths)
-            for owner_path, parameter_paths in paths_by_owner.items()
-        }
+        self.parameter_structure.index(self.parameters)
 
     def to_object(self, *, update_delegate: bool = False, sync_delegate: bool = True) -> Any:
         """Reconstruct object from flat parameters with updated nested configs.
@@ -2643,41 +2402,7 @@ def to_object(self, *, update_delegate: bool = False, sync_delegate: bool = True
         # For delegation, work with the extraction target (delegate), not object_instance
         target = self._extraction_target
 
-        # Collect ALL top-level field updates from self.parameters
-        # This includes both primitive fields AND nested dataclass fields
-        field_updates = {}
-        root_type = type(target)
-        for field_name in self.parameter_structure.owner_paths:
-            if '.' not in field_name:
-                # Check if this field's TYPE is a dataclass (not the instance value)
-                # We need to check the TYPE because the instance value might be stale
-                # (e.g., self.parameters['partition_filter_config'] might have partition_filter=2
-                # even though self.parameters['partition_filter_config.partition_filter'] = None)
-                field_type = self.parameter_structure.owner_paths.get(field_name)
-                # CRITICAL FIX: _path_to_type stores CONTAINER type for leaf fields,
-                # but FIELD type for nested dataclass fields. We must distinguish:
-                # - If field_type == root_type, it's a leaf field (container type stored)
-                # - If field_type != root_type AND is_dataclass, it's a nested dataclass
-                is_nested_dataclass = (
-                    field_type is not None and
-                    is_dataclass(field_type) and
-                    field_type != root_type  # Not the container type
-                )
-                if is_nested_dataclass:
-                    # Nested dataclass: ALWAYS recursively reconstruct from flat storage
-                    # This ensures we pick up changes to nested fields like
-                    # 'partition_filter_config.partition_filter'
-                    logger.debug(f"🔧 to_object: Reconstructing nested dataclass '{field_name}' from flat storage")
-                    reconstructed = self._reconstruct_from_prefix(field_name)
-                    logger.debug(f"🔧 to_object: Reconstructed '{field_name}' type={type(reconstructed).__name__}")
-                    # Log some field values for debugging
-                    if hasattr(reconstructed, 'enabled'):
-                        logger.debug(f"🔧 to_object: '{field_name}'.enabled = {reconstructed.enabled}")
-                    field_updates[field_name] = reconstructed
-                else:
-                    # Primitive field: use value directly from parameters
-                    value = self.parameters.get(field_name)
-                    field_updates[field_name] = value
+        field_updates = self._top_level_values()
 
         # Reconstruct the target object (either object_instance or delegate)
         reconstructed = None
@@ -2731,79 +2456,26 @@ def to_object(self, *, update_delegate: bool = False, sync_delegate: bool = True
 
         return self._cached_object
 
-    def _reconstruct_from_prefix(self, prefix: str) -> Any:
-        """Recursively reconstruct dataclass from flat parameters.
-
-        Args:
-            prefix: Current path prefix (e.g., 'partition_filter_config')
-
-        Returns:
-            Reconstructed dataclass instance
-        """
-        return self._reconstruct_from_parameter_snapshot(prefix, self.parameters)
-
     def _reconstruct_from_parameter_snapshot(
         self,
         prefix: str,
         parameters: dict[str, Any],
+        structure: ParameterStructure,
     ) -> Any:
-        """Recursively reconstruct dataclass from a flat parameter snapshot."""
-        # Determine the type to reconstruct
-        if not prefix:
-            # Root level - use extraction target type (handles delegation)
-            obj_type = type(self._extraction_target)
-        else:
-            # Nested level - look up type from _path_to_type
-            obj_type = self.parameter_structure.owner_paths.get(prefix)
-            if obj_type is None:
-                raise ValueError(f"No type mapping for prefix: {prefix}")
-
-        prefix_dot = f'{prefix}.' if prefix else ''
-
-        # Collect direct fields and nested prefixes
-        direct_fields = {}
-        nested_prefixes = set()
+        """Reconstruct raw values through this epoch's admitted topology."""
 
-        for path, value in parameters.items():
-            if not path.startswith(prefix_dot):
-                continue
-
-            remainder = path[len(prefix_dot):]
-
-            if '.' in remainder:
-                # This is a nested field - collect the first component
-                first_component = remainder.split('.')[0]
-                nested_prefixes.add(first_component)
-            else:
-                # Direct field of this object
-                direct_fields[remainder] = value
-
-        # Reconstruct nested dataclasses first
-        for nested_name in nested_prefixes:
-            nested_path = f'{prefix_dot}{nested_name}'
-            nested_obj = self._reconstruct_from_parameter_snapshot(
-                nested_path,
-                parameters,
+        obj_type = type(self._extraction_target) if not prefix else structure.owner_paths[prefix]
+        direct_fields = {
+            path.field_name: (
+                self._reconstruct_from_parameter_snapshot(path.value, parameters, structure)
+                if structure.is_container_parameter(path.value) else parameters[path.value]
             )
-            direct_fields[nested_name] = nested_obj
-
-        # CRITICAL: Do NOT filter out None values!
-        # None can be a semantic value, such as "inherit from parent context".
-        # When a user explicitly resets a field to None, we MUST pass that None
-        # to the dataclass constructor so lazy resolution can walk up the MRO.
-        # Filtering None would cause the dataclass to use its class-level default
-        # instead of the user's explicit None, breaking inheritance.
-
-        # At root level, include constructor params excluded from editing.
+            for path in structure.children(prefix)
+        }
         if not prefix:
-            direct_fields.update(self.parameter_structure.excluded_values)
-
-
-        # Instantiate the dataclass with ALL fields including None values
-        result = obj_type(**direct_fields)
-
-
-        return result
+            direct_fields.update(structure.excluded_values)
+        # Raw None is meaningful; never replace it with the signature default.
+        return obj_type(**direct_fields)
 
     def _get_changed_params_with_types(
         self, old_target: Any, new_target: Any
diff --git a/src/objectstate/object_state_registry.py b/src/objectstate/object_state_registry.py
index bf7dd33..ad934b0 100644
--- a/src/objectstate/object_state_registry.py
+++ b/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
diff --git a/src/objectstate/parameter_structure.py b/src/objectstate/parameter_structure.py
index c075eee..a9b810f 100644
--- a/src/objectstate/parameter_structure.py
+++ b/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.
diff --git a/src/objectstate/transaction_checkpoint.py b/src/objectstate/transaction_checkpoint.py
index e25fcb6..432d764 100644
--- a/src/objectstate/transaction_checkpoint.py
+++ b/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,
diff --git a/tests/test_construction_history.py b/tests/test_construction_history.py
index 898bcd1..54ec867 100644
--- a/tests/test_construction_history.py
+++ b/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):
diff --git a/tests/test_native_member_topology.py b/tests/test_native_member_topology.py
new file mode 100644
index 0000000..e413f0e
--- /dev/null
+++ b/tests/test_native_member_topology.py
@@ -0,0 +1,161 @@
+"""Native member edits retain distinct live, saved and historical topologies."""
+
+from dataclasses import dataclass, field
+
+import pytest
+
+from objectstate import DottedFieldPath, ObjectState, ObjectStateRegistry
+
+
+@dataclass(frozen=True)
+class TextMember:
+    text: str = 'original'
+
+
+@dataclass(frozen=True)
+class RangeMember:
+    lower: int = 2
+    upper: int = 8
+
+    def __post_init__(self) -> None:
+        if self.lower > self.upper:
+            raise ValueError('lower exceeds upper')
+
+
+@dataclass(frozen=True)
+class EmptyMember:
+    pass
+
+
+@dataclass
+class MemberRoot:
+    member: TextMember | RangeMember | EmptyMember | int | None = field(default_factory=TextMember)
+    sibling: str = 'unchanged'
+    excluded: str = 'retained'
+
+
+@pytest.fixture(autouse=True)
+def registry():
+    ObjectStateRegistry.clear()
+    yield
+    ObjectStateRegistry.clear()
+
+
+def assert_epoch(state: ObjectState, value, children: set[str]) -> None:
+    expected = {'member', 'sibling'} | {f'member.{name}' for name in children}
+    assert set(state.parameters) == expected
+    assert set(state.parameter_structure.declarations) == expected
+    assert set(state.parameter_structure.owner_paths) == expected
+    assert set(state.parameter_structure.defaults) == expected
+    assert set(state.parameter_structure.descriptions) == expected
+    assert {path.value for path in state.direct_parameter_paths('member')} == {
+        f'member.{name}' for name in children
+    }
+    assert state.get_resolved_value('member') == value
+    assert type(state.get_resolved_value('member')) is type(value)
+    assert state.to_object().member == value
+    assert state.to_resolved_object().member == value
+    assert state.to_object().excluded == 'retained'
+    assert state.get_resolved_value('sibling') == 'unchanged'
+    assert state.parameter_structure.exclusions == ['excluded']
+
+
+def test_unrelated_members_replace_all_live_metadata_but_not_saved_epoch():
+    state = ObjectState(MemberRoot(), scope_id='members', exclude_params=['excluded'])
+    ObjectStateRegistry.register(state, _skip_snapshot=True)
+    events: list[set[str]] = []
+    state.on_resolved_changed(lambda paths: events.append(set(paths)))
+
+    changed = state.update_parameter('member', RangeMember())
+    assert changed == {'member.text', 'member.lower', 'member.upper'}
+    assert events == [changed]
+    assert_epoch(state, RangeMember(), {'lower', 'upper'})
+    saved_structure = state._saved_structure
+    saved_values = state._saved_resolved
+    for _ in range(3):
+        assert state.get_saved_resolved_value('member') == TextMember()
+        assert state.to_saved_resolved_object().member == TextMember()
+        assert state._saved_structure is saved_structure
+        assert state._saved_resolved is saved_values
+    assert state.is_raw_dirty
+    assert state.dirty_fields == {'member.text', 'member.lower', 'member.upper'}
+
+    state.update_parameter('member.lower', 3)
+    assert_epoch(state, RangeMember(lower=3), {'lower', 'upper'})
+    assert state.to_saved_resolved_object().member == TextMember()
+    state.mark_saved()
+    assert state.get_saved_resolved_value('member') == RangeMember(lower=3)
+    assert state.to_saved_resolved_object().member == RangeMember(lower=3)
+    assert not state.is_raw_dirty
+    assert not state.dirty_fields
+    assert state._saved_structure is not saved_structure
+    assert set(state._saved_structure.declarations) == set(state.parameters)
+
+
+@pytest.mark.parametrize('value,children,container', [
+    (RangeMember(), {'lower', 'upper'}, True),
+    (EmptyMember(), set(), True),
+    (None, set(), False),
+    (17, set(), False),
+    (TextMember('return'), {'text'}, True),
+])
+def test_member_shape_round_trip_from_different_prior_shapes(value, children, container):
+    state = ObjectState(MemberRoot(), exclude_params=['excluded'])
+    state.update_parameter('member', RangeMember())
+    state.update_parameter('member', None)
+    state.update_parameter('member', value)
+    assert_epoch(state, value, children)
+    assert state.parameter_structure.is_container_parameter('member') == container
+    assert state.to_saved_resolved_object().member == TextMember()
+    assert state.get_saved_resolved_value('member.text') == 'original'
+    assert state.is_raw_dirty
+    state.mark_saved()
+    assert state.get_saved_resolved_value('member') == value
+    assert type(state.get_saved_resolved_value('member')) is type(value)
+    assert state.to_saved_resolved_object().member == value
+    assert not state.is_raw_dirty
+    assert not state.dirty_fields
+
+
+def test_failed_native_ancestor_preparation_changes_nothing():
+    state = ObjectState(MemberRoot(RangeMember()), scope_id='members', exclude_params=['excluded'])
+    ObjectStateRegistry.register(state, _skip_snapshot=True)
+    ObjectStateRegistry.record_snapshot('before refusal')
+    parameters = dict(state.parameters)
+    saved = dict(state._saved_parameters)
+    structure = state.parameter_structure
+    history = [entry.id for entry in ObjectStateRegistry.get_branch_history()]
+    token = ObjectStateRegistry._token
+    events = []
+    state.on_resolved_changed(lambda paths: events.append(paths))
+    with pytest.raises(ValueError, match='lower exceeds upper'):
+        state.update_parameter('member.lower', 99)
+    assert state.parameters == parameters
+    assert state._saved_parameters == saved
+    assert state.parameter_structure is structure
+    assert ObjectStateRegistry._token == token
+    assert [entry.id for entry in ObjectStateRegistry.get_branch_history()] == history
+    assert events == []
+    assert_epoch(state, RangeMember(), {'lower', 'upper'})
+
+
+def test_registry_undo_redo_restores_native_members_and_their_own_baselines():
+    state = ObjectState(MemberRoot(), scope_id='members', exclude_params=['excluded'])
+    ObjectStateRegistry.register(state, _skip_snapshot=True)
+    ObjectStateRegistry.record_snapshot('original')
+    state.update_parameter('member', RangeMember())
+    state.update_parameter('member.lower', 3)
+    assert ObjectStateRegistry.time_travel_back()
+    assert_epoch(state, RangeMember(), {'lower', 'upper'})
+    assert state.to_saved_resolved_object().member == TextMember()
+    assert ObjectStateRegistry.time_travel_back()
+    assert_epoch(state, TextMember(), {'text'})
+    assert not state.is_raw_dirty
+    assert ObjectStateRegistry.time_travel_forward()
+    assert_epoch(state, RangeMember(), {'lower', 'upper'})
+    assert state.to_saved_resolved_object().member == TextMember()
+    assert state.is_raw_dirty
+    assert ObjectStateRegistry.time_travel_forward()
+    assert_epoch(state, RangeMember(lower=3), {'lower', 'upper'})
+    assert state.parameter_structure.path('member.lower') == DottedFieldPath('member.lower')
+    assert state.get_saved_resolved_value('member.text') == 'original'
diff --git a/tests/test_object_state_update_instance.py b/tests/test_object_state_update_instance.py
index e893533..06d87cb 100644
--- a/tests/test_object_state_update_instance.py
+++ b/tests/test_object_state_update_instance.py
@@ -207,21 +207,12 @@ def test_resolved_snapshot_rejects_partial_parameter_state_without_recovery():
     assert state.direct_parameter_paths("nested") == expected_topology
 
 
-def test_replacement_rebuilds_direct_parameter_topology_once(monkeypatch):
+def test_replacement_rebuilds_direct_parameter_topology():
     state = ObjectState(PlainConfig(), scope_id="config")
-    index_calls = 0
-    original_index = state._index_parameter_paths
-
-    def counted_index() -> None:
-        nonlocal index_calls
-        index_calls += 1
-        original_index()
-
-    monkeypatch.setattr(state, "_index_parameter_paths", counted_index)
-
     state.update_object_instance(ConfigWithNestedTopology())
 
-    assert index_calls == 1
+    assert set(state.parameter_structure.declarations) == set(state.parameters)
+    assert set(state.parameter_structure.owner_paths) == set(state.parameters)
     assert state.direct_parameter_paths() == (
         DottedFieldPath("enabled"),
         DottedFieldPath("nested"),
@@ -249,7 +240,7 @@ def on_registry_change(scope_id: str, changed_paths: set[str]) -> None:
     assert events == [("config", {"threshold"})]
 
 
-def test_dataclass_leaf_value_is_not_treated_as_flat_container():
+def test_native_dataclass_admits_its_actual_children_from_an_absent_baseline():
     LazyConfigWithDataclassLeaf = LazyDataclassFactory.make_lazy_simple(
         ConfigWithDataclassLeaf
     )
@@ -260,7 +251,9 @@ def test_dataclass_leaf_value_is_not_treated_as_flat_container():
     state.update_parameter("leaf", value)
 
     assert state.get_resolved_value("leaf") == value
-    assert state.dirty_fields == {"leaf"}
+    assert state.parameters['leaf.name'] == 'explicit'
+    assert state.get_saved_resolved_value('leaf') is None
+    assert state.dirty_fields == {"leaf", "leaf.name"}
 
     state.mark_saved()