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

edgemind/cli.py

@@ -0,0 +1,60 @@
+"""Small CLI helpers to exercise sidecar and safety primitives.
+
+This module provides small, well-tested functions used by unit tests and
+simple scripts to exercise core runtime pieces without pulling in a full
+CLI dependency. The functions are intentionally thin wrappers around
+existing primitives in edgemind.sidecar and edgemind.safety so tests can
+verify integration points.
+"""
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List, Optional
+
+from .sidecar import AuditLogStore, merge_plan_deltas
+from .safety import check_contracts
+
+
+_STORE_CACHE: Dict[str, AuditLogStore] = {}
+
+
+def _get_store(db_path: str) -> AuditLogStore:
+    # Reuse store instances per db_path to allow in-memory DBs to be shared
+    if db_path not in _STORE_CACHE:
+        _STORE_CACHE[db_path] = AuditLogStore(db_path)
+    return _STORE_CACHE[db_path]
+
+
+def append_audit(db_path: str, actor: str, action: str, metadata: Optional[Dict[str, Any]] = None) -> int:
+    """Append an audit entry to the SQLite store and return the row id.
+
+    Uses a simple per-db_path cache so ":memory:" usages in tests share the same
+    connection instance.
+    """
+    store = _get_store(db_path)
+    return store.append(actor, action, metadata)
+
+
+def list_audit(db_path: str) -> List[Dict[str, Any]]:
+    """Return all audit entries as a list of dicts."""
+    store = _get_store(db_path)
+    return store.iter_entries()
+
+
+def verify_audit(db_path: str) -> bool:
+    """Verify the integrity of the audit chain."""
+    store = _get_store(db_path)
+    return store.verify_chain()
+
+
+def merge_deltas(deltas: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """Merge plan deltas using the sidecar helper."""
+    return merge_plan_deltas(deltas)
+
+
+def evaluate_contracts(contracts: Dict[str, Dict[str, Any]], state: Dict[str, Any]):
+    """Evaluate a dict of contracts against a state mapping and return results.
+
+    Returns the same tuple list as edgemind.safety.check_contracts: [(id, passed, fail_safe), ...]
+    """
+    return check_contracts(contracts, state)

tests/test_cli.py

@@ -0,0 +1,31 @@
+from edgemind.cli import append_audit, list_audit, verify_audit, merge_deltas, evaluate_contracts
+
+
+def test_audit_cli_in_memory():
+    # Using in-memory DB path ensures isolation
+    db_path = ":memory:"
+    id1 = append_audit(db_path, "cliAgent", "init", {"v": 1})
+    id2 = append_audit(db_path, "cliAgent", "step", {"v": 2})
+    entries = list_audit(db_path)
+    assert len(entries) == 2
+    assert entries[0]["id"] == id1
+    assert entries[1]["id"] == id2
+    assert verify_audit(db_path) is True
+
+
+def test_merge_deltas_cli():
+    d1 = {"id": "d1", "patches": [{"op_id": "x", "op": "set"}], "timestamp": "2020-01-01T00:00:00"}
+    d2 = {"id": "d2", "patches": [{"op_id": "y", "op": "set"}], "timestamp": "2020-01-01T00:00:01"}
+    merged = merge_deltas([d2, d1])
+    assert len(merged["patches"]) == 2
+
+
+def test_evaluate_contracts_cli():
+    contracts = {
+        "c1": {"predicate": "battery > 10", "fail_safe": "halt"},
+        "c2": {"predicate": "temp < 100", "fail_safe": "safe-mode"},
+    }
+    state = {"battery": 50, "temp": 20}
+    results = evaluate_contracts(contracts, state)
+    # ensure both contracts passed
+    assert all(passed for (_cid, passed, _fs) in results)