diff --git a/edgemind/plan_cert.py b/edgemind/plan_cert.py index f2389b5..ca7953a 100644 --- a/edgemind/plan_cert.py +++ b/edgemind/plan_cert.py @@ -1,94 +1,150 @@ -"""PlanCert and PlanProof utilities. +"""PlanCert schema and lightweight verifier. -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. +Provides PlanCert and PlanProof models and a small, safe expression evaluator used by +verify_plan to check preconditions and simple invariants. This is intentionally small +and dependency-free beyond pydantic which the project already depends on. """ from __future__ import annotations -from typing import Any, Dict, List, Optional, Tuple -from pydantic import BaseModel +from typing import Dict, Any, List, Optional +import ast -from .safety import evaluate_predicate, UnsafeExpression +from pydantic import BaseModel, Field 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 + preconditions: List[str] = Field(default_factory=list) + invariants: Dict[str, str] = Field(default_factory=dict) + worst_case_energy: float = 0.0 + provenance: Dict[str, Any] = Field(default_factory=dict) class PlanProof(BaseModel): - plan_id: str accepted: bool score: float - reasons: List[str] = [] + reasons: List[str] = Field(default_factory=list) 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. +class _SafeEvalError(Exception): + pass - 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. +def _safe_eval(expr: str, state: Dict[str, Any]) -> bool: + """Evaluate simple boolean/comparison expressions against a state dict. + + Allowed expressions: comparisons (>, >=, <, <=, ==, !=), boolean and/or, numeric + literals, names that map to keys in `state` and simple arithmetic (+,-,*,/). + This uses ast to prevent arbitrary code execution. + """ + node = ast.parse(expr, mode="eval") + + def _eval(n): + if isinstance(n, ast.Expression): + return _eval(n.body) + if isinstance(n, ast.BoolOp): + if isinstance(n.op, ast.And): + return all(_eval(v) for v in n.values) + if isinstance(n.op, ast.Or): + return any(_eval(v) for v in n.values) + raise _SafeEvalError("unsupported boolean op") + if isinstance(n, ast.Compare): + left = _eval(n.left) + for op, comp in zip(n.ops, n.comparators): + right = _eval(comp) + if isinstance(op, ast.Gt): + ok = left > right + elif isinstance(op, ast.GtE): + ok = left >= right + elif isinstance(op, ast.Lt): + ok = left < right + elif isinstance(op, ast.LtE): + ok = left <= right + elif isinstance(op, ast.Eq): + ok = left == right + elif isinstance(op, ast.NotEq): + ok = left != right + else: + raise _SafeEvalError("unsupported comparison") + if not ok: + return False + left = right + return True + if isinstance(n, ast.BinOp): + l = _eval(n.left) + r = _eval(n.right) + if isinstance(n.op, ast.Add): + return l + r + if isinstance(n.op, ast.Sub): + return l - r + if isinstance(n.op, ast.Mult): + return l * r + if isinstance(n.op, ast.Div): + return l / r + raise _SafeEvalError("unsupported binary operator") + if isinstance(n, ast.UnaryOp): + if isinstance(n.op, ast.USub): + return -_eval(n.operand) + raise _SafeEvalError("unsupported unary op") + if isinstance(n, ast.Name): + if n.id in state: + return state[n.id] + raise _SafeEvalError(f"unknown name: {n.id}") + if isinstance(n, ast.Constant): + return n.value + if isinstance(n, ast.Num): # type: ignore + return n.n # type: ignore + raise _SafeEvalError(f"unsupported AST node: {type(n)}") + + return bool(_eval(node)) + + +def verify_plan(cert: PlanCert, state: Dict[str, Any], energy_budget: float) -> PlanProof: + """Run a lightweight verification of the PlanCert against a given state and energy budget. + + Checks: + - All preconditions evaluate to True in `state`. + - All invariants evaluate to True in `state`. + - worst_case_energy <= energy_budget + + Returns a PlanProof with accepted bool, a simple confidence score (0..1), and reasons. """ reasons: List[str] = [] - score = 1.0 # Check preconditions - for i, p in enumerate(cert.preconditions): + for p in cert.preconditions: try: - ok = evaluate_predicate(p, initial_state) - except UnsafeExpression as e: - reasons.append(f"precondition[{i}] unsafe: {e}") + ok = _safe_eval(p, state) + except Exception as e: + reasons.append(f"precondition parse error: {p} -> {e}") ok = False if not ok: - reasons.append(f"precondition[{i}] failed: {p}") - score -= 0.4 + reasons.append(f"precondition failed: {p}") # 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 = _safe_eval(inv, state) + except Exception as e: + reasons.append(f"invariant parse error ({name}): {inv} -> {e}") ok = False if not ok: - reasons.append(f"invariant[{name}] failed: {inv}") - score -= 0.3 + reasons.append(f"invariant failed ({name}): {inv}") - # 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 + # Energy check + if cert.worst_case_energy is not None and cert.worst_case_energy > energy_budget: + reasons.append( + f"worst_case_energy {cert.worst_case_energy} exceeds budget {energy_budget}" + ) - accepted = score > 0.0 and len([r for r in reasons if r.startswith("precondition") or r.startswith("invariant")]) == 0 + accepted = len(reasons) == 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]} + # Simple score: start at 1.0, subtract 0.25 per reason, clamp to [0,1] + score = max(0.0, 1.0 - 0.25 * len(reasons)) - # Clamp score - if score < 0: - score = 0.0 - if score > 1.0: - score = 1.0 + proof = PlanProof(accepted=accepted, score=score, reasons=reasons or []) + return proof - return PlanProof(plan_id=cert.plan_id, accepted=accepted, score=score, reasons=reasons, counterexample=counterexample) + +__all__ = ["PlanCert", "PlanProof", "verify_plan"]