fix(persistence): allocate collision-free checkpoint snapshot IDs

rege/persistence/checkpoint.py

@@ -58,7 +58,7 @@ def create_checkpoint(
             The created StateSnapshot
         """
         now = datetime.now()
-        snapshot_id = f"SNAP_{now.strftime('%Y%m%d_%H%M%S')}"
+        snapshot_id = self._allocate_snapshot_id(now)
 
         snapshot = StateSnapshot(
             snapshot_id=snapshot_id,
@@ -81,6 +81,28 @@ def create_checkpoint(
 
         return snapshot
 
+    def _allocate_snapshot_id(self, when: datetime) -> str:
+        """Allocate a snapshot ID unique across registry and snapshot dir.
+
+        Keeps the whole-second timestamp base (sortable, stable) with a
+        zero-padded sequence suffix (SNAP_YYYYMMDD_HHMMSS_NNN), matching the
+        layout documented in the module docstring. The sequence is derived
+        from IDs already claimed in the checkpoint registry and snapshot
+        files on disk, so two checkpoints created within the same second
+        receive distinct IDs even when the clock does not advance between
+        calls, and allocation stays correct across manager instances and
+        process restarts.
+        """
+        base = f"SNAP_{when.strftime('%Y%m%d_%H%M%S')}"
+        claimed = {e.get("snapshot_id") for e in self.archive.load_checkpoint_registry()}
+        claimed |= {p.stem for p in self._snapshots_dir.glob("SNAP_*.json")}
+        seq = 1
+        candidate = f"{base}_{seq:03d}"
+        while candidate in claimed:
+            seq += 1
+            candidate = f"{base}_{seq:03d}"
+        return candidate
+
     def _save_snapshot(self, snapshot: StateSnapshot) -> str:
         """Save a snapshot to file."""
         file_path = self._snapshots_dir / f"{snapshot.snapshot_id}.json"

rege/tests/test_checkpoint_collision.py

@@ -0,0 +1,116 @@
+"""
+Regression tests for recursive-engine--generative-entity#38:
+same-second checkpoint ID collisions overwriting snapshots.
+
+CheckpointManager.create_checkpoint() derived snapshot_id to whole-second
+precision, so two checkpoints created within the same second shared one ID:
+the second write replaced the first snapshot file while the registry kept
+two entries, and loading the first name returned the second state.
+"""
+
+import re
+from datetime import datetime
+
+import pytest
+
+import rege.persistence.checkpoint as checkpoint_mod
+from rege.persistence.archive import ArchiveManager
+from rege.persistence.checkpoint import CheckpointManager
+from rege.core.exceptions import PersistenceError
+
+
+FIXED_NOW = datetime(2026, 10, 23, 23, 15, 0)
+
+
+class _FrozenDateTime(datetime):
+    """datetime stub with a clock that never advances."""
+
+    @classmethod
+    def now(cls, tz=None):
+        return FIXED_NOW
+
+
+@pytest.fixture
+def frozen_clock(monkeypatch):
+    monkeypatch.setattr(checkpoint_mod, "datetime", _FrozenDateTime)
+
+
+def _manager(tmp_path):
+    archive = ArchiveManager(str(tmp_path / "archive"))
+    return CheckpointManager(archive)
+
+
+def _state(tag):
+    return {"metrics": {"tag": tag}, "organs": {}, "pending": [], "errors": []}
+
+
+def test_same_second_checkpoints_get_distinct_ids_and_files(frozen_clock, tmp_path):
+    """Two checkpoints at the identical wall-clock instant: distinct stable IDs, distinct files."""
+    manager = _manager(tmp_path)
+
+    first = manager.create_checkpoint("first", _state("FIRST"))
+    second = manager.create_checkpoint("second", _state("SECOND"))
+
+    assert first.snapshot_id != second.snapshot_id
+    assert re.fullmatch(r"SNAP_20261023_231500_\d{3}", first.snapshot_id)
+    assert re.fullmatch(r"SNAP_20261023_231500_\d{3}", second.snapshot_id)
+    assert (manager._snapshots_dir / f"{first.snapshot_id}.json").exists()
+    assert (manager._snapshots_dir / f"{second.snapshot_id}.json").exists()
+
+
+def test_same_second_registry_entries_resolve_to_own_state(frozen_clock, tmp_path):
+    """Each registry entry resolves to its own saved state; no stale-name aliasing."""
+    manager = _manager(tmp_path)
+
+    first = manager.create_checkpoint("first", _state("FIRST"))
+    second = manager.create_checkpoint("second", _state("SECOND"))
+
+    registry = manager.list_checkpoints()
+    assert len(registry) == 2
+
+    loaded_first = manager.get_checkpoint_by_name("first")
+    loaded_second = manager.get_checkpoint_by_name("second")
+    assert loaded_first.snapshot_id == first.snapshot_id
+    assert loaded_second.snapshot_id == second.snapshot_id
+    assert loaded_first.system_state["tag"] == "FIRST"
+    assert loaded_second.system_state["tag"] == "SECOND"
+
+
+def test_sequence_continues_across_manager_instances(frozen_clock, tmp_path):
+    """A new manager on the same archive dir must not reuse an allocated ID."""
+    manager = _manager(tmp_path)
+    first = manager.create_checkpoint("first", _state("FIRST"))
+
+    restarted = _manager(tmp_path)
+    second = restarted.create_checkpoint("second", _state("SECOND"))
+
+    assert second.snapshot_id != first.snapshot_id
+    assert len(restarted.list_checkpoints()) == 2
+    assert restarted.get_checkpoint_by_name("first").system_state["tag"] == "FIRST"
+
+
+def test_failed_snapshot_write_creates_no_registry_entry(tmp_path, monkeypatch):
+    """Fail closed: a failed snapshot write must not leave a registry entry."""
+    manager = _manager(tmp_path)
+
+    def _boom(snapshot):
+        raise PersistenceError("simulated disk failure")
+
+    monkeypatch.setattr(manager, "_save_snapshot", _boom)
+
+    with pytest.raises(PersistenceError):
+        manager.create_checkpoint("doomed", _state("DOOMED"))
+
+    assert manager.list_checkpoints() == []
+
+
+def test_preexisting_legacy_id_does_not_collide(frozen_clock, tmp_path):
+    """An old-format (unsuffixed) snapshot on disk does not break allocation."""
+    manager = _manager(tmp_path)
+    legacy_path = manager._snapshots_dir / "SNAP_20261023_231500.json"
+    legacy_path.write_text("{}", encoding="utf-8")
+
+    created = manager.create_checkpoint("new", _state("NEW"))
+
+    assert created.snapshot_id != "SNAP_20261023_231500"
+    assert (manager._snapshots_dir / f"{created.snapshot_id}.json").exists()