build(agent): weasel-1#856f80 iteration

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
+Provides PlanCert and PlanProof models and a small, safe expression evaluator used by
-that performs basic syntactic and semantic checks using the existing safety
+verify_plan to check preconditions and simple invariants. This is intentionally small
-predicate evaluator. This is intentionally tiny and suitable for edge use.
+and dependency-free beyond pydantic which the project already depends on.
 """
 from __future__ import annotations
 
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Dict, Any, List, Optional
-from pydantic import BaseModel
+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] = Field(default_factory=list)
-    preconditions: List[str] = []
+    invariants: Dict[str, str] = Field(default_factory=dict)
-    postconditions: List[str] = []
+    worst_case_energy: float = 0.0
-    invariants: Dict[str, str] = {}
+    provenance: Dict[str, Any] = Field(default_factory=dict)
-    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] = []
+    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:
+class _SafeEvalError(Exception):
-    """Verify a PlanCert against an initial state and optional energy budget.
+    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,
+def _safe_eval(expr: str, state: Dict[str, Any]) -> bool:
-    deterministic verifier; probabilistic adversarial sampling is outside its
+    """Evaluate simple boolean/comparison expressions against a state dict.
-    scope.
+
+    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)
+            ok = _safe_eval(p, state)
-        except UnsafeExpression as e:
+        except Exception as e:
-            reasons.append(f"precondition[{i}] unsafe: {e}")
+            reasons.append(f"precondition parse error: {p} -> {e}")
             ok = False
         if not ok:
-            reasons.append(f"precondition[{i}] failed: {p}")
+            reasons.append(f"precondition failed: {p}")
-            score -= 0.4
 
     # Check invariants
     for name, inv in cert.invariants.items():
         try:
-            ok = evaluate_predicate(inv, initial_state)
+            ok = _safe_eval(inv, state)
-        except UnsafeExpression as e:
+        except Exception as e:
-            reasons.append(f"invariant[{name}] unsafe: {e}")
+            reasons.append(f"invariant parse error ({name}): {inv} -> {e}")
             ok = False
         if not ok:
-            reasons.append(f"invariant[{name}] failed: {inv}")
+            reasons.append(f"invariant failed ({name}): {inv}")
-            score -= 0.3
 
-    # Energy budget check
+    # Energy check
-    if energy_budget is not None and cert.worst_case_energy is not None:
+    if cert.worst_case_energy is not None and cert.worst_case_energy > energy_budget:
-        try:
+        reasons.append(
-            if cert.worst_case_energy > energy_budget:
+            f"worst_case_energy {cert.worst_case_energy} exceeds budget {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
+    accepted = len(reasons) == 0
 
-    # If rejected, include a minimal counterexample: the parts of state that caused failures
+    # Simple score: start at 1.0, subtract 0.25 per reason, clamp to [0,1]
-    counterexample = None
+    score = max(0.0, 1.0 - 0.25 * len(reasons))
-    if not accepted:
-        counterexample = {k: initial_state.get(k) for k in sorted(initial_state.keys())[:10]}
 
-    # Clamp score
+    proof = PlanProof(accepted=accepted, score=score, reasons=reasons or [])
-    if score < 0:
+    return proof
-        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)
+
+__all__ = ["PlanCert", "PlanProof", "verify_plan"]