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

edgemind/safety.py

@@ -0,0 +1,124 @@
+"""Runtime safety contract evaluator.
+
+Provides a small, auditable evaluator that can safely evaluate boolean
+predicates expressed over a LocalProblem's state variables. This is intentionally
+minimal and does not allow function calls, attribute access, or other unsafe
+operations. It raises ValueError for disallowed expressions.
+
+The evaluator accepts expressions like:
+  "battery > 20 and distance_to_goal < 5"
+and looks up names from a provided state mapping.
+"""
+from __future__ import annotations
+
+import ast
+from typing import Any, Dict
+
+
+ALLOWED_NODES = {
+    ast.Expression,
+    ast.BoolOp,
+    ast.UnaryOp,
+    ast.BinOp,
+    ast.Compare,
+    ast.Name,
+    ast.Load,
+    ast.Constant,
+    ast.And,
+    ast.Or,
+    ast.Not,
+    ast.Add,
+    ast.Sub,
+    ast.Mult,
+    ast.Div,
+    ast.Mod,
+    ast.Pow,
+    ast.Eq,
+    ast.NotEq,
+    ast.Lt,
+    ast.LtE,
+    ast.Gt,
+    ast.GtE,
+    ast.Is,
+    ast.IsNot,
+    ast.In,
+    ast.NotIn,
+    ast.UnaryOp,
+    ast.USub,
+    ast.UAdd,
+}
+
+
+class UnsafeExpression(ValueError):
+    pass
+
+
+def _ensure_safe(node: ast.AST) -> None:
+    """Recursively ensure AST contains only allowed nodes.
+
+    Raises UnsafeExpression on disallowed nodes.
+    """
+    node_type = type(node)
+    if node_type not in ALLOWED_NODES:
+        raise UnsafeExpression(f"Disallowed AST node: {node_type.__name__}")
+
+    # Recurse into supported child fields
+    for child in ast.iter_child_nodes(node):
+        _ensure_safe(child)
+
+
+def evaluate_predicate(expr: str, state: Dict[str, Any]) -> bool:
+    """Evaluate a boolean predicate string against a state mapping.
+
+    Only names present in `state` are allowed identifiers. Any unknown name
+    raises UnsafeExpression.
+
+    Examples:
+        evaluate_predicate("battery > 20 and dist < 5", {"battery": 50, "dist": 3})
+    """
+    try:
+        tree = ast.parse(expr, mode="eval")
+    except SyntaxError as e:
+        raise UnsafeExpression(f"Syntax error in predicate: {e}")
+
+    _ensure_safe(tree)
+
+    # Prepare a restricted eval environment: only allow names from state
+    class NameResolver(ast.NodeTransformer):
+        def visit_Name(self, node: ast.Name) -> ast.AST:
+            if node.id not in state:
+                raise UnsafeExpression(f"Unknown name in predicate: {node.id}")
+            # Replace the Name node with a Constant node containing the value
+            value = state[node.id]
+            return ast.copy_location(ast.Constant(value=value), node)
+
+    resolved = NameResolver().visit(tree)
+    ast.fix_missing_locations(resolved)
+
+    compiled = compile(resolved, filename="<safety-predicate>", mode="eval")
+    result = eval(compiled, {"__builtins__": {}}, {})
+    if not isinstance(result, (bool, int, float)):
+        # We expect a boolean-like result; numbers are allowed (0/1)
+        raise UnsafeExpression("Predicate did not evaluate to a boolean or numeric value")
+
+    return bool(result)
+
+
+def check_contracts(contracts: Dict[str, Dict[str, Any]], state: Dict[str, Any]):
+    """Evaluate a dict of contracts against state.
+
+    contracts: mapping contract_id -> {"predicate": str, "fail_safe": str}
+
+    Returns a list of (contract_id, passed: bool, fail_safe) tuples.
+    """
+    results = []
+    for cid, c in contracts.items():
+        pred = c.get("predicate") or c.get("predicate_str") or "False"
+        fail_safe = c.get("fail_safe", "halt")
+        try:
+            passed = evaluate_predicate(pred, state)
+        except UnsafeExpression:
+            # Treat unsafe/malformed predicate as violation
+            passed = False
+        results.append((cid, passed, fail_safe))
+    return results

tests/test_safety.py

@@ -0,0 +1,41 @@
+from edgemind import safety
+
+
+def test_evaluate_predicate_basic_true():
+    state = {"battery": 50, "dist": 3}
+    assert safety.evaluate_predicate("battery > 20 and dist < 5", state) is True
+
+
+def test_evaluate_predicate_basic_false():
+    state = {"battery": 10, "dist": 10}
+    assert safety.evaluate_predicate("battery > 20 or dist < 5", state) is False
+
+
+def test_evaluate_predicate_unknown_name():
+    state = {"battery": 50}
+    try:
+        safety.evaluate_predicate("battery > 20 and foo < 5", state)
+        assert False, "Expected UnsafeExpression for unknown name"
+    except safety.UnsafeExpression:
+        pass
+
+
+def test_evaluate_predicate_malicious_code():
+    state = {"battery": 50}
+    # Attempt to call builtins or import should be rejected
+    try:
+        safety.evaluate_predicate("__import__('os').system('echo hi') == 0", state)
+        assert False, "Expected UnsafeExpression for malicious code"
+    except safety.UnsafeExpression:
+        pass
+
+
+def test_check_contracts_aggregates_results():
+    contracts = {
+        "c1": {"predicate": "battery > 20", "fail_safe": "halt"},
+        "c2": {"predicate": "battery > 80", "fail_safe": "safe-mode"},
+    }
+    state = {"battery": 50}
+    results = dict((cid, (passed, fs)) for cid, passed, fs in safety.check_contracts(contracts, state))
+    assert results["c1"][0] is True and results["c1"][1] == "halt"
+    assert results["c2"][0] is False and results["c2"][1] == "safe-mode"