Preserve typed ObjectState history across restarts

README.md

@@ -167,8 +167,8 @@ history_dict = ObjectStateRegistry.export_history_to_dict()
 ObjectStateRegistry.import_history_from_dict(history_dict)
 
 # Or save to file
-ObjectStateRegistry.save_history_to_file("history.json")
-ObjectStateRegistry.load_history_from_file("history.json")
+ObjectStateRegistry.save_history_to_file("history.objectstate")
+ObjectStateRegistry.load_history_from_file("history.objectstate")
 ```
 
 ## Automatic Lazy Config Generation with Decorators

docs/conf.py

@@ -1,17 +1,18 @@
 """Sphinx configuration for objectstate."""
-import os
 import sys
 from pathlib import Path
 
 # Add source to path
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
 
+from objectstate import __version__  # noqa: E402
+
 # Project information
 project = "objectstate"
 copyright = "2024, Tristan Simas"
 author = "Tristan Simas"
 version = "1.0"
-release = "1.0.21"
+release = __version__
 
 # General configuration
 extensions = [

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "objectstate"
-version = "1.0.22"
+version = "1.0.23"
 description = "Generic lazy dataclass configuration framework with dual-axis inheritance and contextvars-based resolution"
 authors = [{name = "Tristan Simas", email = "tristan.simas@mail.mcgill.ca"}]
 license = {text = "MIT"}
@@ -24,6 +24,7 @@ classifiers = [
 keywords = ["configuration", "dataclass", "lazy", "inheritance", "contextvars", "hierarchical"]
 
 dependencies = [
+    "dill>=0.4.0",
     "python-introspect>=0.1.6",
 ]
 

src/objectstate/__init__.py

@@ -214,7 +214,7 @@
     'ObjectStateEditSession',
 ]
 
-__version__ = '1.0.21'
+__version__ = "1.0.23"
 __author__ = 'OpenHCS Team'
 __description__ = 'Generic configuration framework for lazy dataclass resolution'
 

src/objectstate/object_state_registry.py

@@ -2097,7 +2097,7 @@ def get_current_branch(cls) -> str:
 
     @classmethod
     def export_history_to_dict(cls) -> Dict[str, Any]:
-        """Export history to a JSON-serializable dict.
+        """Export the complete typed history document payload.
 
         Returns:
             Dict with 'snapshots', 'timelines', 'current_head', 'current_timeline'.
@@ -2111,7 +2111,7 @@ def export_history_to_dict(cls) -> Dict[str, Any]:
 
     @classmethod
     def import_history_from_dict(cls, data: Dict[str, Any]) -> None:
-        """Import history from a dict (e.g., loaded from JSON).
+        """Import a complete typed history document payload.
 
         Only imports state data for scope_ids that currently exist in the registry.
         Scopes in the snapshot but not in the app are skipped.
@@ -2123,16 +2123,7 @@ def import_history_from_dict(cls, data: Dict[str, Any]) -> None:
         cls._timelines.clear()
         current_scopes = set(cls._states.keys())
 
-        # Handle both old list format and new dict format
-        snapshots_data = data['snapshots']
-        if isinstance(snapshots_data, list):
-            # Old format: list of snapshots
-            snapshot_items = [(s['id'], s) for s in snapshots_data]
-        else:
-            # New format: dict of id -> snapshot
-            snapshot_items = snapshots_data.items()
-
-        for _snapshot_id, snapshot_data in snapshot_items:
+        for _snapshot_id, snapshot_data in data['snapshots'].items():
             # Filter to only scopes that exist in current registry
             filtered_states: Dict[str, StateSnapshot] = {}
             for scope_id, state_data in snapshot_data['states'].items():
@@ -2141,68 +2132,53 @@ def import_history_from_dict(cls, data: Dict[str, Any]) -> None:
                         saved_resolved=state_data['saved_resolved'],
                         live_resolved=state_data['live_resolved'],
                         parameters=state_data['parameters'],
-                        # Back-compat: old history may omit saved_parameters or explicitly store it as null.
-                        # In either case, treat it as "same as parameters".
-                        saved_parameters=(
-                            state_data.get('saved_parameters')
-                            if state_data.get('saved_parameters') is not None
-                            else state_data['parameters']
-                        ),
+                        saved_parameters=state_data['saved_parameters'],
                         provenance=state_data['provenance'],
-                        meta=state_data.get('meta') or {},
+                        meta=state_data['meta'],
                     )
 
             snapshot = Snapshot(
                 id=snapshot_data['id'],
                 timestamp=snapshot_data['timestamp'],
                 label=snapshot_data['label'],
-                triggering_scope=snapshot_data.get('triggering_scope'),
-                parent_id=snapshot_data.get('parent_id'),
+                triggering_scope=snapshot_data['triggering_scope'],
+                parent_id=snapshot_data['parent_id'],
                 all_states=filtered_states,
             )
             cls._snapshots[snapshot.id] = snapshot
 
-        # Import timelines
-        if 'timelines' in data:
-            for tl_data in data['timelines']:
-                tl = Timeline.from_dict(tl_data)
-                cls._timelines[tl.name] = tl
-            cls._current_timeline = data.get('current_timeline', 'main')
-        else:
-            cls._current_timeline = 'main'
-
-        # Handle both old index format and new head format
-        if 'current_head' in data:
-            cls._current_head = data['current_head']
-        elif 'current_index' in data:
-            # Old format - convert index to head
-            # Can't reliably convert, just go to head
-            cls._current_head = None
-        else:
-            cls._current_head = None
+        for timeline_data in data['timelines']:
+            timeline = Timeline.from_dict(timeline_data)
+            cls._timelines[timeline.name] = timeline
+        cls._current_timeline = data['current_timeline']
+        cls._current_head = data['current_head']
 
     @classmethod
     def save_history_to_file(cls, filepath: str) -> None:
-        """Save history to a JSON file.
+        """Save history as a type-preserving ObjectState document.
+
+        The document is intended for trusted local persistence. Dill preserves
+        arbitrary Python values and graph identity without a parallel type-tag
+        registry.
 
         Args:
-            filepath: Path to save the JSON file.
+            filepath: Path to the binary ObjectState history document.
         """
-        import json
-        data = cls.export_history_to_dict()
-        with open(filepath, 'w') as f:
-            json.dump(data, f, indent=2)
+        import dill
+
+        with open(filepath, "wb") as history_file:
+            dill.dump(cls.export_history_to_dict(), history_file)
         logger.info(f"⏱️ Saved {len(cls._snapshots)} snapshots to {filepath}")
 
     @classmethod
     def load_history_from_file(cls, filepath: str) -> None:
-        """Load history from a JSON file.
+        """Load a trusted type-preserving ObjectState history document.
 
         Args:
-            filepath: Path to the JSON file.
+            filepath: Path to the binary ObjectState history document.
         """
-        import json
-        with open(filepath, 'r') as f:
-            data = json.load(f)
-        cls.import_history_from_dict(data)
+        import dill
+
+        with open(filepath, "rb") as history_file:
+            cls.import_history_from_dict(dill.load(history_file))
         logger.info(f"⏱️ Loaded {len(cls._snapshots)} snapshots from {filepath}")

src/objectstate/snapshot_model.py

@@ -64,7 +64,7 @@ def create(
         )
 
     def to_dict(self) -> Dict:
-        """Export to JSON-serializable dict."""
+        """Export the typed snapshot document payload."""
         return {
             'id': self.id,
             'timestamp': self.timestamp,
@@ -86,21 +86,15 @@ def to_dict(self) -> Dict:
 
     @classmethod
     def from_dict(cls, data: Dict) -> 'Snapshot':
-        """Import from dict (e.g., loaded from JSON)."""
+        """Import a typed snapshot document payload."""
         all_states = {
             scope_id: StateSnapshot(
                 saved_resolved=state_data['saved_resolved'],
                 live_resolved=state_data['live_resolved'],
                 parameters=state_data['parameters'],
-                # Back-compat: old snapshots may be missing saved_parameters entirely.
-                # Also guard against explicit null saved_parameters in older exported histories.
-                saved_parameters=(
-                    state_data.get('saved_parameters')
-                    if state_data.get('saved_parameters') is not None
-                    else state_data['parameters']
-                ),
+                saved_parameters=state_data['saved_parameters'],
                 provenance=state_data['provenance'],
-                meta=state_data.get('meta') or {},
+                meta=state_data['meta'],
             )
             for scope_id, state_data in data['states'].items()
         }

tests/test_history_persistence.py

@@ -0,0 +1,60 @@
+"""Typed ObjectState history document persistence."""
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from enum import Enum
+from pathlib import Path
+
+from objectstate import ObjectState, ObjectStateRegistry
+
+
+class HistoryMode(Enum):
+    """Representative nominal configuration value."""
+
+    FAST = "fast"
+
+
+def identity(value):
+    """Representative importable callable configuration value."""
+
+    return value
+
+
+@dataclass
+class HistoryConfig:
+    output_path: Path
+    mode: HistoryMode
+    transform: Callable
+
+
+def _reset_registry_history() -> None:
+    ObjectStateRegistry._snapshots.clear()
+    ObjectStateRegistry._timelines.clear()
+    ObjectStateRegistry._current_timeline = "main"
+    ObjectStateRegistry._current_head = None
+
+
+def test_history_file_round_trips_typed_values(tmp_path: Path) -> None:
+    state = ObjectState(
+        HistoryConfig(
+            output_path=tmp_path / "results",
+            mode=HistoryMode.FAST,
+            transform=identity,
+        ),
+        scope_id="typed_history",
+    )
+    ObjectStateRegistry.register(state, _skip_snapshot=True)
+    ObjectStateRegistry.record_snapshot("typed values", scope_id=state.scope_id)
+    history_path = tmp_path / "history.objectstate"
+
+    ObjectStateRegistry.save_history_to_file(str(history_path))
+    _reset_registry_history()
+    ObjectStateRegistry.load_history_from_file(str(history_path))
+
+    snapshot = ObjectStateRegistry.get_branch_history()[-1]
+    parameters = snapshot.all_states[state.scope_id].parameters
+    assert parameters["output_path"] == tmp_path / "results"
+    assert type(parameters["output_path"]) is type(tmp_path)
+    assert parameters["mode"] is HistoryMode.FAST
+    assert parameters["transform"] is identity
+    assert history_path.stat().st_size > 0