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

cosmic_ledger/adapter.py

@@ -0,0 +1,65 @@
+from dataclasses import dataclass
+from typing import List, Optional, Dict, Any
+from .delta import verify_delta_block, verify_inclusion, merkle_path_for_index
+
+
+@dataclass
+class DeltaBlock:
+    """A small wrapper representing a compact delta-block used by adapters.
+
+    Fields mirror the structure produced by DeltaLog.export_delta_block to
+    provide a minimal, well-typed API surface for adapters and demos.
+    """
+    entries: List[Dict[str, Any]]
+    digests: List[str]
+    root: str
+    anchor: Optional[str]
+    aggregated_sig: Optional[str]
+    since_index: int = 0
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "entries": self.entries,
+            "digests": self.digests,
+            "root": self.root,
+            "anchor": self.anchor,
+            "aggregated_sig": self.aggregated_sig,
+            "since_index": self.since_index,
+        }
+
+    @staticmethod
+    def from_dict(d: Dict[str, Any]) -> "DeltaBlock":
+        return DeltaBlock(
+            entries=d.get("entries", []),
+            digests=d.get("digests", []),
+            root=d.get("root", ""),
+            anchor=d.get("anchor"),
+            aggregated_sig=d.get("aggregated_sig"),
+            since_index=d.get("since_index", 0),
+        )
+
+
+def verify_block(block: DeltaBlock, signer_key: Optional[bytes] = None) -> bool:
+    """Convenience wrapper around verify_delta_block for DeltaBlock instances."""
+    return verify_delta_block(block.to_dict(), signer_key=signer_key)
+
+
+def verify_entry_inclusion(entry_digest: str, block: DeltaBlock, index: int) -> bool:
+    """Verify that a given entry digest is included in the block's Merkle root.
+
+    This function reconstructs the merkle path information from the block's
+    entries for the requested index and calls the library-level verify_inclusion
+    helper. This is useful for adapters that need to prove a single entry's
+    inclusion without re-computing the whole tree.
+    """
+    if index < 0 or index >= len(block.digests):
+        return False
+    # find the merkle_path attached to the corresponding entry if present
+    try:
+        entry = block.entries[index]
+    except Exception:
+        return False
+
+    merkle_path = entry.get("merkle_path") if isinstance(entry, dict) else None
+    root = block.root
+    return verify_inclusion(entry_digest, merkle_path or [], root, index)

tests/test_adapter.py

@@ -0,0 +1,36 @@
+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
+from cosmic_ledger.adapter import DeltaBlock, verify_block, verify_entry_inclusion
+
+
+def test_deltablock_create_and_verify():
+    dl = DeltaLog()
+    # add a few simple entries
+    for i in range(5):
+        dl.add_entry({"v": i})
+
+    key = b'test-key-for-adapter'
+    raw_block = dl.export_delta_block(since_index=0, signer_key=key)
+
+    block = DeltaBlock.from_dict(raw_block)
+
+    # verify aggregated block signature and structure
+    assert block.root == raw_block["root"]
+    assert len(block.digests) == len(block.entries)
+    assert verify_block(block, signer_key=key)
+
+
+def test_verify_single_entry_inclusion():
+    dl = DeltaLog()
+    for i in range(3):
+        dl.add_entry({"k": i})
+
+    block = DeltaBlock.from_dict(dl.export_delta_block(since_index=0))
+    # verify inclusion for each entry
+    for idx, digest in enumerate(block.digests):
+        assert verify_entry_inclusion(digest, block, idx)