build(agent): jabba#56a767 iteration

missionledger/csfd.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, asdict
+from typing import Any, Dict, List
+import hashlib
+import json
+
+from .dsl import PlanDelta
+
+
+@dataclass
+class CSFDFrame:
+    timestamp_offset: int
+    invariant_id_list: List[str]
+    minimal_state_snapshot_hash: str
+    expected_resource_bounds: Dict[str, Any]
+    proof_stub_hash: str
+
+
+@dataclass
+class CSFDDigest:
+    frames: List[CSFDFrame]
+    signer: str = ""
+    digest_hex: str = ""
+
+
+def _hash_obj(obj: Any) -> str:
+    s = json.dumps(obj, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(s.encode()).hexdigest()
+
+
+def generate_csfd(plan_delta: PlanDelta, signer: str = "", n_frames: int = 3) -> CSFDDigest:
+    """Generate a compact Critical-Safety Frame Digest (CSFD) for a PlanDelta.
+
+    The digest contains up to `n_frames` ordered frames. Each frame is a
+    compact summary referencing invariant ids and a minimal snapshot hash.
+
+    This implementation is intentionally small and deterministic for the
+    MVP; a production implementation should include proper signing (ed25519)
+    and configurable snapshot selectors.
+    """
+    # Source invariants from PlanDelta.metadata (if present)
+    invariants = plan_delta.metadata.get("invariants", {}) if plan_delta and plan_delta.metadata else {}
+
+    # Flatten invariant ids deterministically
+    invariant_ids = sorted(list(map(str, invariants.keys())))
+
+    frames: List[CSFDFrame] = []
+    # Create N frames. For small plan deltas we may repeat or slice invariant ids.
+    for i in range(n_frames):
+        # take a slice of invariant ids for this frame
+        slice_start = (i * 1) % max(1, len(invariant_ids)) if invariant_ids else 0
+        # choose up to 3 invariants per frame
+        ids = invariant_ids[slice_start : slice_start + 3] if invariant_ids else []
+
+        # minimal state snapshot: hash of a few plan delta fields
+        snapshot = {
+            "version": plan_delta.version,
+            "delta_keys": sorted(list(map(str, plan_delta.delta.keys()))) if plan_delta.delta else [],
+            "meta_keys": sorted(list(map(str, plan_delta.metadata.keys()))) if plan_delta.metadata else [],
+            "frame_index": i,
+        }
+
+        minimal_hash = _hash_obj(snapshot)
+
+        # proof_stub_hash: if metadata provides a proof stub map use it, else hash the snapshot
+        proof_stub_hash = (
+            plan_delta.metadata.get("proof_stub_hash", "")
+            or _hash_obj({"snapshot": snapshot, "invariants": ids})
+        )
+
+        # expected_resource_bounds: best-effort from metadata
+        expected_bounds = plan_delta.metadata.get("expected_resource_bounds", {}) or {}
+
+        frame = CSFDFrame(
+            timestamp_offset=i,  # simplistic offset in frames
+            invariant_id_list=ids,
+            minimal_state_snapshot_hash=minimal_hash,
+            expected_resource_bounds=expected_bounds,
+            proof_stub_hash=proof_stub_hash,
+        )
+        frames.append(frame)
+
+    # digest: compact hex over serialized frames
+    digest_src = [asdict(f) for f in frames]
+    digest_hex = hashlib.sha256(json.dumps(digest_src, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+
+    csfd = CSFDDigest(frames=frames, signer=signer, digest_hex=digest_hex)
+
+    # Safety: ensure serialized size is reasonably small (<= 512 bytes for MVP)
+    serialized = json.dumps(asdict(csfd), sort_keys=True, separators=(",", ":"))
+    if len(serialized.encode()) > 512:
+        # If too large, truncate frames to last frame only and recompute digest
+        frames = frames[:1]
+        digest_src = [asdict(f) for f in frames]
+        digest_hex = hashlib.sha256(json.dumps(digest_src, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+        csfd = CSFDDigest(frames=frames, signer=signer, digest_hex=digest_hex)
+
+    return csfd

tests/test_csfd.py

@@ -0,0 +1,33 @@
+from missionledger.dsl import PlanDelta
+from missionledger.csfd import generate_csfd
+
+
+def test_csfd_generation_small_plandelta():
+    pd = PlanDelta(delta={"move": True, "safe_buffer": 0.5}, version=2, metadata={
+        "invariants": {"collision": True, "energy": True},
+        "proof_stub_hash": "stub-abc",
+        "expected_resource_bounds": {"energy": [0, 100]},
+    })
+
+    csfd = generate_csfd(pd, signer="planner-v1", n_frames=3)
+
+    assert csfd is not None
+    assert csfd.signer == "planner-v1"
+    # frames count should be at least 1 and at most requested
+    assert 1 <= len(csfd.frames) <= 3
+    # digest hex should be a 64-char hex string
+    assert isinstance(csfd.digest_hex, str)
+    assert len(csfd.digest_hex) == 64
+
+
+def test_csfd_size_bound():
+    # Create a PlanDelta with large metadata to force truncation path
+    large_meta = {f"k{i}": "v" * 200 for i in range(5)}
+    pd = PlanDelta(delta={"x": 1}, version=1, metadata={"invariants": {"a": True}, **large_meta})
+
+    csfd = generate_csfd(pd, signer="s", n_frames=5)
+    # Ensure serialized representation is <= 512 bytes as enforced
+    import json
+
+    serialized = json.dumps({"frames": [f.__dict__ for f in csfd.frames], "signer": csfd.signer, "digest_hex": csfd.digest_hex}, sort_keys=True, separators=(",", ":"))
+    assert len(serialized.encode()) <= 512