diff --git a/AGENTS.md b/AGENTS.md index 78b943b..b0a439f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -26,3 +26,4 @@ Rules for contributors: - Write tests for every new feature - Keep interfaces backwards-compatible unless explicitly requested - Update AGENTS.md with notable architectural changes + - Added a minimal sidecar module (edgemind/sidecar.py): SQLite-backed append-only audit log and a simple PlanDelta merge helper. Tests added in tests/test_sidecar.py diff --git a/edgemind/sidecar.py b/edgemind/sidecar.py new file mode 100644 index 0000000..c1629aa --- /dev/null +++ b/edgemind/sidecar.py @@ -0,0 +1,120 @@ +"""Minimal sidecar runtime helpers. + +Provides: +- AuditLogStore: SQLite-backed append-only log with chained hashes (Merkle-like linear chain). +- merge_plan_deltas: simple CRDT-style merge for PlanDelta patches. + +This is intentionally small and dependency-free so it can run on edge devices. +""" +from __future__ import annotations + +import sqlite3 +import hashlib +import json +from typing import Dict, Any, List, Optional + + +def _hash_bytes(b: bytes) -> str: + return hashlib.sha256(b).hexdigest() + + +class AuditLogStore: + """Append-only SQLite-backed audit log with chained hashes. + + Each row contains (id INTEGER PRIMARY KEY AUTOINCREMENT, entry_json TEXT, prev_hash TEXT, hash TEXT) + The hash is sha256(prev_hash || entry_json) so the chain can be verified deterministically. + """ + + def __init__(self, db_path: str = ":memory:"): + self.db_path = db_path + self.conn = sqlite3.connect(self.db_path, check_same_thread=False) + self._init_db() + + def _init_db(self) -> None: + cur = self.conn.cursor() + cur.execute( + """ + CREATE TABLE IF NOT EXISTS audit_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + entry_json TEXT NOT NULL, + prev_hash TEXT, + hash TEXT NOT NULL + ) + """ + ) + self.conn.commit() + + def append(self, actor: str, action: str, metadata: Optional[Dict[str, Any]] = None) -> int: + entry = {"actor": actor, "action": action, "metadata": metadata or {}} + entry_json = json.dumps(entry, sort_keys=True, separators=(",", ":")) + prev_hash = self._get_latest_hash() + h = _hash_bytes((prev_hash or "").encode() + entry_json.encode()) + cur = self.conn.cursor() + cur.execute( + "INSERT INTO audit_log (entry_json, prev_hash, hash) VALUES (?, ?, ?)", + (entry_json, prev_hash, h), + ) + self.conn.commit() + return cur.lastrowid + + def _get_latest_hash(self) -> Optional[str]: + cur = self.conn.cursor() + cur.execute("SELECT hash FROM audit_log ORDER BY id DESC LIMIT 1") + row = cur.fetchone() + return row[0] if row is not None else None + + def iter_entries(self) -> List[Dict[str, Any]]: + cur = self.conn.cursor() + cur.execute("SELECT id, entry_json, prev_hash, hash FROM audit_log ORDER BY id ASC") + out = [] + for row in cur.fetchall(): + out.append({"id": row[0], "entry": json.loads(row[1]), "prev_hash": row[2], "hash": row[3]}) + return out + + def verify_chain(self) -> bool: + """Verify the integrity of the chain. + + Returns True if every row's hash equals sha256(prev_hash || entry_json) and linkage is intact. + """ + cur = self.conn.cursor() + cur.execute("SELECT entry_json, prev_hash, hash FROM audit_log ORDER BY id ASC") + prev = None + for entry_json, prev_hash, h in cur.fetchall(): + expected = _hash_bytes((prev_hash or "").encode() + entry_json.encode()) + if expected != h: + return False + # Also check that prev_hash matches previous row's hash + if prev is not None and prev != (prev_hash or None): + return False + prev = h + return True + + +def merge_plan_deltas(deltas: List[Dict[str, Any]]) -> Dict[str, Any]: + """Merge a list of PlanDelta-like dicts into a single combined PlanDelta. + + Strategy (simple CRDT-style): + - Each delta has an `id`, optional `base_plan_id`, and `patches` list where each patch is a dict + that must include an `op_id` unique identifier. + - We combine all patches and deduplicate by `op_id`, ordering by delta timestamp if present. + - This is intentionally simple; real-world use should use signed ops and version vectors. + """ + # Sort deltas by timestamp if provided, otherwise by id to ensure deterministic order + def _sort_key(d): + return (d.get("timestamp") or "", d.get("id") or "") + + sorted_d = sorted(deltas, key=_sort_key) + seen = set() + combined_patches = [] + for d in sorted_d: + for p in d.get("patches", []): + op_id = p.get("op_id") or json.dumps(p, sort_keys=True) + if op_id in seen: + continue + seen.add(op_id) + combined_patches.append(p) + + merged = {"id": "merged-" + _hash_bytes(json.dumps([d.get("id") for d in sorted_d]).encode()), + "base_plan_id": None, + "patches": combined_patches} + return merged diff --git a/tests/test_sidecar.py b/tests/test_sidecar.py new file mode 100644 index 0000000..d545bf6 --- /dev/null +++ b/tests/test_sidecar.py @@ -0,0 +1,33 @@ +from edgemind.sidecar import AuditLogStore, merge_plan_deltas + + +def test_audit_log_append_and_iterate(): + s = AuditLogStore(":memory:") + id1 = s.append("agentA", "start", {"detail": 1}) + id2 = s.append("agentB", "update", {"detail": 2}) + entries = s.iter_entries() + assert len(entries) == 2 + assert entries[0]["id"] == id1 + assert entries[1]["id"] == id2 + assert s.verify_chain() is True + + +def test_audit_log_chain_tamper_detection(): + s = AuditLogStore(":memory:") + s.append("a", "one", {}) + s.append("b", "two", {}) + # Tamper directly in DB + cur = s.conn.cursor() + cur.execute("UPDATE audit_log SET entry_json = ? WHERE id = 1", ('{"actor":"x"}',)) + s.conn.commit() + assert s.verify_chain() is False + + +def test_merge_plan_deltas_simple(): + d1 = {"id": "d1", "patches": [{"op_id": "a", "op": "set", "k": "x", "v": 1}], "timestamp": "2021-01-01T00:00:00"} + d2 = {"id": "d2", "patches": [{"op_id": "b", "op": "set", "k": "y", "v": 2}], "timestamp": "2021-01-01T00:00:01"} + merged = merge_plan_deltas([d2, d1]) + # Two patches, deterministic order by timestamp + assert len(merged["patches"]) == 2 + assert merged["patches"][0]["op_id"] == "a" + assert merged["patches"][1]["op_id"] == "b"