diff --git a/edgemind/plan_cert.py b/edgemind/plan_cert.py index 4c808aa..c928427 100644 --- a/edgemind/plan_cert.py +++ b/edgemind/plan_cert.py @@ -32,7 +32,13 @@ class PlanCert: invariants: Dict[str, Any] worst_case_energy: Optional[float] = None wcet_ms: Optional[int] = None - provenance: Dict[str, Any] = None + # Optional maximum actuator exposure (compact risk metric) + max_actuator_exposure: Optional[float] = None + # Criticality level (e.g. low, normal, critical) used by admission control + criticality: Optional[str] = None + # Optional rollback checkpoint: small dict containing a checkpoint_hash and revert window + rollback_checkpoint: Optional[Dict[str, Any]] = None + provenance: Optional[Dict[str, Any]] = None signature: Optional[str] = None # hex HMAC-SHA256 signature when signed def to_dict(self) -> Dict[str, Any]: diff --git a/edgemind/sidecar.py b/edgemind/sidecar.py index c1629aa..518ed2e 100644 --- a/edgemind/sidecar.py +++ b/edgemind/sidecar.py @@ -118,3 +118,18 @@ def merge_plan_deltas(deltas: List[Dict[str, Any]]) -> Dict[str, Any]: "base_plan_id": None, "patches": combined_patches} return merged + + +def emit_rollback_event(store: AuditLogStore, plan_cert: "object", reason: str) -> int: + """Emit a watchdog/rollback event into the audit log. + + This is a small helper used by runtimes/watchdogs to persist an operator/verifier + initiated rollback. The function records the plan_id and any rollback_checkpoint + present on the PlanCert. + """ + metadata = { + "plan_id": getattr(plan_cert, "plan_id", None), + "rollback_checkpoint": getattr(plan_cert, "rollback_checkpoint", None), + "reason": reason, + } + return store.append("watchdog", "rollback", metadata) diff --git a/tests/test_plan_cert.py b/tests/test_plan_cert.py index d2034c7..336a787 100644 --- a/tests/test_plan_cert.py +++ b/tests/test_plan_cert.py @@ -10,6 +10,9 @@ def test_plan_cert_sign_and_verify(): invariants={"no-collision": True}, worst_case_energy=12.5, wcet_ms=500, + max_actuator_exposure=0.2, + criticality="normal", + rollback_checkpoint={"checkpoint_hash": "h1", "revert_window": 30}, provenance={"planner_version": "v0.1", "seed": 42, "env_hash": "abc"}, ) @@ -46,3 +49,18 @@ def test_plan_proof_sign_and_verify(): proof = make_plan_proof(cert, accepted=True, score=0.95, verifier="test-verifier", key=key) assert proof.signature is not None assert verify_plan_proof(proof, key) is True + + +def test_plan_cert_with_optional_fields(): + key = b"k3" + cert = PlanCert( + plan_id="p4", + preconditions={}, + postconditions={}, + invariants={}, + provenance={"planner_version": "y"}, + rollback_checkpoint={"checkpoint_hash": "c1", "revert_window": 10}, + criticality="critical", + ) + signed = sign_plan_cert(cert, key) + assert verify_plan_cert(signed, key) is True diff --git a/tests/test_sidecar.py b/tests/test_sidecar.py index d545bf6..a054958 100644 --- a/tests/test_sidecar.py +++ b/tests/test_sidecar.py @@ -31,3 +31,25 @@ def test_merge_plan_deltas_simple(): assert len(merged["patches"]) == 2 assert merged["patches"][0]["op_id"] == "a" assert merged["patches"][1]["op_id"] == "b" + + +def test_emit_rollback_event_into_audit_log(): + from edgemind.plan_cert import PlanCert + from edgemind.sidecar import emit_rollback_event + + s = AuditLogStore(":memory:") + cert = PlanCert( + plan_id="p-rollback", + preconditions={}, + postconditions={}, + invariants={}, + provenance={"planner_version": "w"}, + rollback_checkpoint={"checkpoint_hash": "chk", "revert_window": 5}, + ) + rid = emit_rollback_event(s, cert, "timeout") + entries = s.iter_entries() + # one entry created + assert len(entries) == 1 + assert entries[0]["id"] == rid + assert entries[0]["entry"]["action"] == "rollback" or entries[0]["entry"]["actor"] in ("watchdog", "rollback") + assert s.verify_chain() is True