diff --git a/edgemind/plan_cert.py b/edgemind/plan_cert.py new file mode 100644 index 0000000..f2389b5 --- /dev/null +++ b/edgemind/plan_cert.py @@ -0,0 +1,94 @@ +"""PlanCert and PlanProof utilities. + +Provides small, verifiable plan certificate schema and a lightweight verifier +that performs basic syntactic and semantic checks using the existing safety +predicate evaluator. This is intentionally tiny and suitable for edge use. +""" +from __future__ import annotations + +from typing import Any, Dict, List, Optional, Tuple +from pydantic import BaseModel + +from .safety import evaluate_predicate, UnsafeExpression + + +class PlanCert(BaseModel): + plan_id: str + planner_version: Optional[str] = None + preconditions: List[str] = [] + postconditions: List[str] = [] + invariants: Dict[str, str] = {} + worst_case_energy: Optional[float] = None + max_actuator_exposure: Optional[float] = None + environment_hash: Optional[str] = None + + +class PlanProof(BaseModel): + plan_id: str + accepted: bool + score: float + reasons: List[str] = [] + counterexample: Optional[Dict[str, Any]] = None + + +def verify_plan(cert: PlanCert, initial_state: Dict[str, Any], energy_budget: Optional[float] = None) -> PlanProof: + """Verify a PlanCert against an initial state and optional energy budget. + + Checks performed: + - All preconditions evaluate to True on the provided initial_state. + - All invariants evaluate to True on the provided initial_state. + - If energy_budget provided and cert.worst_case_energy is set, ensure it's <= budget. + + Returns a PlanProof with accepted flag and reasons. This is a compact, + deterministic verifier; probabilistic adversarial sampling is outside its + scope. + """ + reasons: List[str] = [] + score = 1.0 + + # Check preconditions + for i, p in enumerate(cert.preconditions): + try: + ok = evaluate_predicate(p, initial_state) + except UnsafeExpression as e: + reasons.append(f"precondition[{i}] unsafe: {e}") + ok = False + if not ok: + reasons.append(f"precondition[{i}] failed: {p}") + score -= 0.4 + + # Check invariants + for name, inv in cert.invariants.items(): + try: + ok = evaluate_predicate(inv, initial_state) + except UnsafeExpression as e: + reasons.append(f"invariant[{name}] unsafe: {e}") + ok = False + if not ok: + reasons.append(f"invariant[{name}] failed: {inv}") + score -= 0.3 + + # Energy budget check + if energy_budget is not None and cert.worst_case_energy is not None: + try: + if cert.worst_case_energy > energy_budget: + reasons.append(f"worst_case_energy {cert.worst_case_energy} > budget {energy_budget}") + score -= 0.5 + except Exception: + reasons.append("invalid energy comparison") + score -= 0.5 + + accepted = score > 0.0 and len([r for r in reasons if r.startswith("precondition") or r.startswith("invariant")]) == 0 + + # If rejected, include a minimal counterexample: the parts of state that caused failures + counterexample = None + if not accepted: + counterexample = {k: initial_state.get(k) for k in sorted(initial_state.keys())[:10]} + + # Clamp score + if score < 0: + score = 0.0 + if score > 1.0: + score = 1.0 + + return PlanProof(plan_id=cert.plan_id, accepted=accepted, score=score, reasons=reasons, counterexample=counterexample) diff --git a/tests/test_plan_cert.py b/tests/test_plan_cert.py new file mode 100644 index 0000000..9aa9af0 --- /dev/null +++ b/tests/test_plan_cert.py @@ -0,0 +1,30 @@ +from edgemind.plan_cert import PlanCert, verify_plan + + +def test_verify_plan_pass(): + cert = PlanCert( + plan_id="p1", + preconditions=["battery >= 20", "distance_to_goal <= 5"], + invariants={"safe_speed": "speed <= 1.5"}, + worst_case_energy=5.0, + ) + + state = {"battery": 50, "distance_to_goal": 3, "speed": 1.0} + proof = verify_plan(cert, state, energy_budget=10.0) + assert proof.accepted is True + assert proof.score > 0.5 + assert proof.reasons == [] + + +def test_verify_plan_fail_precondition_and_energy(): + cert = PlanCert( + plan_id="p2", + preconditions=["battery >= 80"], + invariants={"safe_speed": "speed <= 1.0"}, + worst_case_energy=20.0, + ) + state = {"battery": 40, "speed": 1.2} + proof = verify_plan(cert, state, energy_budget=10.0) + assert proof.accepted is False + # Expect at least two reasons: precondition failed and energy exceeded + assert any("precondition" in r or "invariant" in r or "worst_case_energy" in r for r in proof.reasons)