diff --git a/missionledger/csfd.py b/missionledger/csfd.py new file mode 100644 index 0000000..a2de1d2 --- /dev/null +++ b/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 diff --git a/tests/test_csfd.py b/tests/test_csfd.py new file mode 100644 index 0000000..75ca7b9 --- /dev/null +++ b/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