Reconcile historical snapshots with current schemas

src/objectstate/object_state_registry.py

@@ -4,7 +4,7 @@
 """
 
 from contextlib import contextmanager
-from dataclasses import dataclass, is_dataclass
+from dataclasses import dataclass, is_dataclass, replace
 import logging
 from typing import (
     Any,
@@ -1503,6 +1503,10 @@ def _prepare_time_travel_states(cls, snapshot: Snapshot) -> Dict[str, PreparedSt
                     scope_id=scope, metadata=state_snapshot.meta
                 )
                 schema.validate_metadata_contracts("time_travel")
+                state_snapshot = cls._reconcile_snapshot_schema(
+                    schema,
+                    state_snapshot,
+                )
                 cls._apply_time_travel_snapshot_state(schema, state_snapshot, materialize=False)
                 binding.validate_scope_membership(schema, values.keys())
                 schema._raw_dirty = schema.parameters != schema._saved_parameters
@@ -1526,6 +1530,46 @@ def _prepare_time_travel_states(cls, snapshot: Snapshot) -> Dict[str, PreparedSt
                 raise ValueError(f"Cyclic historical ObjectState parents: {sorted(pending)!r}.")
         return prepared
 
+    @staticmethod
+    def _reconcile_snapshot_schema(
+        schema: "ObjectState",
+        snapshot: StateSnapshot,
+    ) -> StateSnapshot:
+        """Project historical values onto the declaration's current field schema.
+
+        Construction owns the valid parameter surface. Historical snapshots own
+        values only for fields still declared on that surface. This gives newly
+        declared fields their construction-derived defaults and prevents removed
+        fields from surviving as detached historical parameters.
+        """
+
+        def reconcile(current: dict, historical: dict) -> dict:
+            return {
+                key: historical[key] if key in historical else value
+                for key, value in current.items()
+            }
+
+        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,
+            ),
+        )
+
     @classmethod
     def _restore_time_travel_states(
         cls,

tests/test_construction_history.py

@@ -188,6 +188,41 @@ def test_equal_signature_owner_change_records_and_notifies_without_fake_value_ch
     assert values == []
 
 
+def test_history_values_reconcile_to_current_construction_schema():
+    register(Config(value=7, excluded="declaration default"), "evolving")
+    snapshot_id = record("historical schema")
+    document = ObjectStateRegistry.export_history_to_dict()
+    for snapshot in document["snapshots"].values():
+        stored = snapshot["states"]["evolving"]
+        for mapping_name in (
+            "parameters",
+            "saved_parameters",
+            "saved_resolved",
+            "live_resolved",
+            "provenance",
+        ):
+            mapping = stored[mapping_name]
+            mapping.pop("excluded", None)
+            mapping["removed_field"] = "historical-only"
+
+    ObjectStateRegistry.clear()
+    fresh = register(Config(value=99, excluded="fresh value"), "evolving")
+    ObjectStateRegistry.import_history_from_dict(document)
+
+    assert ObjectStateRegistry.time_travel_to_snapshot(snapshot_id)
+    assert ObjectStateRegistry.get_by_scope("evolving") is fresh
+    assert fresh.parameters["value"] == 7
+    assert fresh.parameters["excluded"] == "declaration default"
+    assert fresh._saved_parameters["excluded"] == "declaration default"
+    assert fresh._live_resolved["excluded"] == "declaration default"
+    assert fresh._saved_resolved["excluded"] == "declaration default"
+    assert "removed_field" not in fresh.parameters
+    assert "removed_field" not in fresh._saved_parameters
+    assert "removed_field" not in fresh._live_resolved
+    assert "removed_field" not in fresh._saved_resolved
+    assert "removed_field" not in fresh._live_provenance
+
+
 def test_same_scope_registered_replacement_captures_exact_owner_with_equal_parameters():
     initial_state = register(first_callable, "registered-owner")
     initial = record("original registered owner")