build(agent): weasel-1#856f80 iteration

cosmic_ledger/delta.py

@@ -118,6 +118,9 @@ class DeltaLog:
     def __init__(self):
         self.entries = []  # each entry is a dict with digest and payload
         self.anchor = None  # optional cloud/ground anchor for global verifiability
+        # Checkpoints store archived roots and the set of digests included at that time
+        # Each checkpoint is a dict: {"root": <hex>, "index": <exclusive_index>, "digests": [<hex>, ...]}
+        self.checkpoints = []
 
     def add_entry(self, entry: dict) -> str:
         # entry must be serializable, and we store a digest for Merkle
@@ -162,3 +165,41 @@ class DeltaLog:
     def root(self) -> str:
         digests = [e["digest"] for e in self.entries]
         return merkle_root(digests)
+
+    def checkpoint(self, prune_before_index: int = 0) -> dict:
+        """Create a checkpoint for entries up to prune_before_index (exclusive) and optionally prune them.
+
+        The checkpoint records the Merkle root over the archived leaves and the list of archived digests.
+        It then removes those entries from the active entries list so storage can be reclaimed.
+
+        Returns the checkpoint dict.
+        """
+        if prune_before_index <= 0:
+            # nothing to archive
+            cp_digests = []
+            cp_root = ""
+        else:
+            cp_digests = [e["digest"] for e in self.entries[:prune_before_index]]
+            cp_root = merkle_root(cp_digests)
+
+        checkpoint = {
+            "root": cp_root,
+            "index": prune_before_index,
+            "digests": cp_digests,
+        }
+        # store checkpoint
+        self.checkpoints.append(checkpoint)
+        # prune entries up to prune_before_index
+        if prune_before_index > 0:
+            self.entries = self.entries[prune_before_index:]
+        return checkpoint
+
+    def has_digest(self, digest: str) -> bool:
+        """Return True if digest is present in active entries or archived checkpoints."""
+        for e in self.entries:
+            if e.get("digest") == digest:
+                return True
+        for cp in self.checkpoints:
+            if digest in cp.get("digests", []):
+                return True
+        return False

tests/test_checkpoint.py

@@ -0,0 +1,34 @@
+import sys
+import pathlib
+
+# Ensure repository root is on sys.path so tests can import the package
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
+
+from cosmic_ledger.delta import DeltaLog, merkle_root
+
+
+def test_checkpoint_prune_and_query():
+    dl = DeltaLog()
+    # add five payloads
+    for i in range(5):
+        dl.add_entry({"v": i})
+
+    # collect initial digests and root
+    initial_digests = [e["digest"] for e in dl.entries]
+    initial_root = merkle_root(initial_digests)
+
+    # create a checkpoint archiving first 2 entries
+    cp = dl.checkpoint(prune_before_index=2)
+    assert cp["root"] == merkle_root(initial_digests[:2])
+    # pruned entries should no longer be active
+    active_digests = [e["digest"] for e in dl.entries]
+    assert len(active_digests) == 3
+
+    # but has_digest should still find pruned digests via checkpoint
+    assert dl.has_digest(initial_digests[0])
+    assert dl.has_digest(initial_digests[1])
+    # and should find active ones too
+    assert dl.has_digest(initial_digests[2])
+
+    # new root reflects remaining entries only
+    assert dl.root() == merkle_root(initial_digests[2:])