diff --git a/missionledger/__init__.py b/missionledger/__init__.py index c6a5337..b3cc4a1 100644 --- a/missionledger/__init__.py +++ b/missionledger/__init__.py @@ -6,7 +6,7 @@ and Safety/Privacy contracts, a shadow planner, a simple on-device safety proof hook, and a tamper-evident governance ledger. """ -from .dsl import LocalProblem, SafetyContracts, PrivacyContracts, PlanDelta +from .dsl import LocalProblem, SafetyContracts, PrivacyContracts, PlanDelta, PoeticCert, ReplayBundle from .planner import ShadowPlanner, PlanResult from .safety import verify_plan from .governance import GovernanceLedger @@ -16,6 +16,8 @@ __all__ = [ "SafetyContracts", "PrivacyContracts", "PlanDelta", + "PoeticCert", + "ReplayBundle", "ShadowPlanner", "PlanResult", "verify_plan", diff --git a/missionledger/dsl.py b/missionledger/dsl.py index ad0debb..f2c0b8e 100644 --- a/missionledger/dsl.py +++ b/missionledger/dsl.py @@ -47,3 +47,36 @@ class PlanDelta: delta: Dict[str, Any] = field(default_factory=dict) version: int = 0 metadata: Dict[str, Any] = field(default_factory=dict) + + + +@dataclass +class PoeticCert: + """Compact human+machine verifiable certificate for quick triage. + + - invariants: key safety/resource invariants (collision, energy envelopes) + - summary: short natural-language summary for on-call triage + - issued_by: identifier of the issuer (device or service) + - timestamp: optional ISO timestamp string + """ + + invariants: Dict[str, Any] = field(default_factory=dict) + summary: str = "" + issued_by: str = "" + timestamp: str = "" + + +@dataclass +class ReplayBundle: + """Signed, Merkle-linked plan trace bundle used for deterministic replay. + + - entries: list of logged entries (dicts) + - signer: identifier of signer + - merkle_root: compact hex digest representing the bundle contents + - nonce: optional replay nonce + """ + + entries: List[Dict[str, Any]] = field(default_factory=list) + signer: str = "" + merkle_root: str = "" + nonce: str = "" diff --git a/missionledger/governance.py b/missionledger/governance.py index 6513ee1..034ed54 100644 --- a/missionledger/governance.py +++ b/missionledger/governance.py @@ -4,6 +4,10 @@ import hashlib import json from datetime import datetime from typing import List +from typing import Tuple +import json + +from .dsl import PoeticCert, ReplayBundle class GovernanceLedger: @@ -30,6 +34,41 @@ class GovernanceLedger: self._head_hash = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest() self._entries.append(payload) + def emit_poetic_cert(self, invariants: dict, summary: str, issuer: str) -> PoeticCert: + """Emit a compact PoeticCert for quick triage and machine checks. + + This is intentionally minimal and human-readable while remaining + machine-checkable. + """ + timestamp = datetime.utcnow().isoformat() + "Z" + cert = PoeticCert(invariants=invariants, summary=summary, issued_by=issuer, timestamp=timestamp) + # Also log the issuance to the ledger for auditability + self.log({"type": "poetic_cert_issued", "cert_summary": summary, "issued_by": issuer}) + return cert + + def create_replay_bundle(self, entries: List[dict], signer: str, nonce: str = "") -> ReplayBundle: + """Construct a ReplayBundle with a compact merkle-style root. + + For the MVP we compute a deterministic root by hashing each entry and + then hashing the concatenation of sorted entry hashes. This produces a + compact fingerprint suitable for off-device anchoring. + """ + # Normalize and hash entries deterministically + entry_hashes = [] + for e in entries: + h = hashlib.sha256(json.dumps(e, sort_keys=True).encode()).hexdigest() + entry_hashes.append(h) + + # deterministic ordering + entry_hashes.sort() + concat = "".join(entry_hashes).encode() + merkle_root = hashlib.sha256(concat).hexdigest() if concat else "" + + bundle = ReplayBundle(entries=list(entries), signer=signer, merkle_root=merkle_root, nonce=nonce) + # log bundle creation + self.log({"type": "replay_bundle_created", "signer": signer, "merkle_root": merkle_root}) + return bundle + @property def entries(self) -> List[dict]: return self._entries[:] diff --git a/tests/test_cert_and_replay.py b/tests/test_cert_and_replay.py new file mode 100644 index 0000000..8367cc1 --- /dev/null +++ b/tests/test_cert_and_replay.py @@ -0,0 +1,23 @@ +from missionledger.governance import GovernanceLedger + + +def test_poetic_cert_and_replay_bundle_creation(): + gl = GovernanceLedger(device_id="test-device-2") + + invariants = {"collision_envelope": 1.2, "energy_budget": 42} + cert = gl.emit_poetic_cert(invariants=invariants, summary="Safe envelope OK", issuer="planner-v1") + + assert cert.issued_by == "planner-v1" + assert "collision_envelope" in cert.invariants + + entries = [ + {"event": "plan_step", "step": 1}, + {"event": "plan_step", "step": 2}, + ] + bundle = gl.create_replay_bundle(entries=entries, signer="device-A", nonce="n-123") + + assert bundle.signer == "device-A" + assert bundle.merkle_root != "" + assert len(bundle.entries) == 2 + # ensure ledger recorded the bundle creation as an entry + assert any(e.get("entry", {}).get("type") == "replay_bundle_created" for e in gl.entries)