diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 3a0982562045..8f7e93c64e19 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1741,7 +1741,7 @@ def _ensure_hermes_home_managed(home: Path): "user_char_limit": 1375, # ~500 tokens at 2.75 chars/token # External memory provider plugin (empty = built-in only). # Set to a provider name to activate: "openviking", "mem0", - # "hindsight", "holographic", "retaindb", "byterover". + # "hindsight", "holographic", "memorygraph", "retaindb", "byterover". # Only ONE external provider is allowed at a time. "provider": "", }, diff --git a/hermes_cli/subcommands/memory.py b/hermes_cli/subcommands/memory.py index 23fe0b857641..2ddd7314a1c7 100644 --- a/hermes_cli/subcommands/memory.py +++ b/hermes_cli/subcommands/memory.py @@ -17,7 +17,7 @@ def build_memory_parser(subparsers, *, cmd_memory: Callable) -> None: description=( "Set up and manage external memory provider plugins.\n\n" "Available providers: honcho, openviking, mem0, hindsight,\n" - "holographic, retaindb, byterover.\n\n" + "holographic, memorygraph, retaindb, byterover.\n\n" "Only one external provider can be active at a time.\n" "Built-in memory (MEMORY.md/USER.md) is always active." ), diff --git a/plugins/memory/memorygraph/DESIGN.md b/plugins/memory/memorygraph/DESIGN.md new file mode 100644 index 000000000000..abdcfd0037be --- /dev/null +++ b/plugins/memory/memorygraph/DESIGN.md @@ -0,0 +1,117 @@ +# Memory Graph v2 — Design + +Status: v0.1 implemented (this plugin). This document records the +architecture decisions behind the implemented core and the **design-only** +roadmap for parts whose architecture is not yet stable enough to build. + +## Goals + +Transform Hermes memory from searchable notes into a **governed knowledge +graph** while preserving every existing memory behavior: + +- Built-in `MEMORY.md` / `USER.md` file memory: untouched. The graph mirrors + its writes via the existing `on_memory_write` provider hook. +- The `MemoryProvider` ABC and `MemoryManager`: untouched. memorygraph is a + standard bundled provider, subject to the one-external-provider rule, and + inert unless `memory.provider: memorygraph` is configured. +- Other providers (honcho, hindsight, holographic, ...): untouched. + +## Implemented core (architecture stable) + +### Data model + +``` +entities ──< claims >── evidence + │ │ + └──< relationships >────┘ (evidence attaches to claims, relationships + └──< entity_aliases and entities via (subject_kind, subject_id)) +governance_log (append-only audit) +``` + +- **Entities** are typed (person, project, goal, skill, business, + organization, place, tool, concept) and resolved by normalized name key + with alias support. Projects, goals, skills, businesses and people are + first-class *entity types*, not separate tables: they share governance, + evidence and relationship semantics, and new types can be added without + migration. +- **Claims** are the unit of knowledge: `entity.attribute = value` plus + confidence, tier, status, temporal validity and provenance. This is the + classic property-graph-with-reified-statements shape: it lets governance + operate on statements (contradict, supersede, age, promote) rather than + on opaque note blobs. +- **Time-aware knowledge** is modeled with `valid_from`/`valid_to` windows + on both claims and relationships, plus a `superseded_by` chain. History is + never deleted — `timeline` reconstructs how knowledge evolved. + +### Governance lifecycle + +``` + re-assertion (dup) aging sweep + write ──► reinforce ──► promote ──► decay ──► demote + │ + ├─ exclusive conflict, clearly newer ──► supersede (time-aware) + └─ exclusive conflict, ambiguous ──► contradicted (review queue) +``` + +- Write-time governance (dedupe, contradiction) keeps the graph clean at + the source. Sweep-time governance (aging, promotion) is idempotent and + runs at session end or on demand, never in the hot path. +- Confidence is a bounded scalar in [0, 1]; every mutation is clamped and + audited. Half-life decay (default 90 days since last reinforcement) + favors knowledge that keeps getting used. +- Promotion tiers gate trust: `candidate` (new, unproven) → `established` + (confident + independently evidenced) → `core` (repeatedly reinforced, + aged, uncontradicted). Consumers can filter recall by tier. + +### Why a bundled provider, not a core change + +The provider seam (`agent/memory_provider.py`) already carries every signal +the graph needs: turn sync, built-in write mirroring, pre-compression +extraction, session boundaries, tool exposure. Building v2 as a provider +means zero risk to existing memory behavior and a clean rollback path +(unset `memory.provider`). + +## Design-only (architecture not yet stable — do not build) + +These are specified here so future work is consistent, but deliberately not +implemented in v0.1: + +1. **Semantic duplicate/contradiction detection.** Current detection is + lexical (normalized keys + Jaccard/sequence similarity). Embedding-based + similarity would catch paraphrases, but Hermes has no core embedding + dependency and each memory provider currently makes its own choice. + Design: an optional `Embedder` protocol injected into + `GovernanceEngine`, falling back to lexical similarity when absent. + Blocked on: choice of a local, dependency-light embedding path. + +2. **Automatic turn extraction (NER → graph).** `sync_turn` / + `on_pre_compress` could extract entities and claims from conversation + automatically. Extraction quality gates trust in the whole graph, so + v0.1 keeps writes explicit (model-invoked `remember`/`link`, plus + mirrored built-in writes). Design: extraction lands as `candidate` + claims with `kind='extraction'` evidence and a lower initial confidence + (0.4), so promotion gates filter noise. Blocked on: an evaluated + extraction prompt/pipeline. + +3. **Cross-provider federation.** Mirroring graph knowledge into an active + cloud provider (or importing from hindsight's entity graph) conflicts + with the one-external-provider rule by design. Any federation should be + an explicit `hermes memory export/import` CLI flow, not a runtime + bridge. + +4. **Promotion into built-in MEMORY.md.** Auto-writing `core` claims into + `MEMORY.md` would change built-in memory behavior, which v2 must not do. + Design: a `promote_review` surface (already queryable via `stats` / + tier filters) that the *model* can act on with the existing memory tool, + keeping the human/model in the loop. + +5. **Multi-hop graph reasoning.** `neighbors()` exists in the store; + compositional queries ("claims connected to X and Y within 2 hops") + need recursive CTEs plus result ranking. Deferred until real usage + shows which query shapes matter. + +## Migration & versioning + +`meta.schema_version` (currently 1) gates future migrations. Migrations +must be additive (new tables/columns) — history tables are append-only and +never rewritten. diff --git a/plugins/memory/memorygraph/README.md b/plugins/memory/memorygraph/README.md new file mode 100644 index 000000000000..966e4b7f52ec --- /dev/null +++ b/plugins/memory/memorygraph/README.md @@ -0,0 +1,76 @@ +# memorygraph — Governed Knowledge Graph Memory + +Transforms Hermes memory from searchable notes into a **governed knowledge +graph**: typed entities, time-aware relationships, evidence-linked claims, +confidence tracking, contradiction/duplicate detection, knowledge aging and +promotion. + +Local-only (stdlib `sqlite3`), no network, no credentials. Database at +`$HERMES_HOME/memory_graph.db` (profile-scoped). + +## Activate + +```yaml +# config.yaml +memory: + provider: memorygraph +``` + +The built-in `memory` tool (MEMORY.md / USER.md) is unchanged. Writes to it +are mirrored into the graph via the `on_memory_write` hook, so both stay +consistent. + +## Model + +| Concept | Table | Notes | +|---|---|---| +| Entities | `entities` | Typed: person, project, goal, skill, business, organization, place, tool, concept. Alias resolution via `entity_aliases`. | +| Relationships | `relationships` | Typed edges with `valid_from`/`valid_to` windows and confidence. | +| Claims | `claims` | `entity.attribute = value` with confidence, tier (candidate → established → core), status (active / superseded / retracted / contradicted), temporal validity. | +| Evidence | `evidence` | Provenance links (session, built-in memory write, tool, URL, quote). | +| Audit | `governance_log` | Append-only log of every governed mutation. | + +## Governance + +- **Duplicate detection** — re-asserting known knowledge reinforces the + existing claim (confidence up, reinforcement count up) instead of + duplicating. Normalized-key match plus fuzzy similarity (default ≥ 0.88). +- **Contradiction detection** — conflicting values for an *exclusive* + attribute are superseded time-aware when the new value is clearly newer + (old claim's validity window closes, linked via `superseded_by`); + otherwise both are flagged `contradicted` for review (`contradictions` / + `resolve` actions). +- **Confidence tracking** — reinforcement +0.10, feedback ±0.15, + contradiction −0.15, clamped to [0, 1]. +- **Knowledge aging** — confidence decays with a 90-day half-life since last + reinforcement (configurable); decayed established/core claims are demoted. +- **Knowledge promotion** — candidate → established at confidence ≥ 0.70 + with ≥ 2 evidence links; established → core at confidence ≥ 0.85 with + ≥ 3 reinforcements and ≥ 7 days of age. Contradicted claims never promote. + +Sweeps run at session end (configurable) or on demand via the `sweep` +action. + +## Tool + +One tool, `graph_memory`, with actions: +`remember`, `link`, `unlink`, `about`, `query`, `timeline`, +`contradictions`, `resolve`, `duplicates`, `feedback`, `forget`, `sweep`, +`stats`. + +## Config (optional) + +`$HERMES_HOME/memorygraph.json`: + +```json +{ + "db_path": "/custom/path/memory_graph.db", + "half_life_days": 90, + "duplicate_similarity": 0.88, + "prefetch_enabled": true, + "sweep_on_session_end": true +} +``` + +See `DESIGN.md` for the governance model rationale and the design-only +roadmap. diff --git a/plugins/memory/memorygraph/__init__.py b/plugins/memory/memorygraph/__init__.py new file mode 100644 index 000000000000..41a8d777277e --- /dev/null +++ b/plugins/memory/memorygraph/__init__.py @@ -0,0 +1,507 @@ +"""memorygraph — governed knowledge graph memory provider for Hermes. + +Transforms memory from searchable notes into a governed knowledge graph: +typed entities (people, projects, goals, skills, businesses, ...), typed +time-aware relationships, attribute claims with evidence links, confidence +tracking, contradiction + duplicate detection, knowledge aging, and +candidate → established → core promotion. + +Local-only: stdlib sqlite3, no network, no credentials. The database is +profile-scoped at ``$HERMES_HOME/memory_graph.db`` by default. + +Activate with ``memory.provider: memorygraph`` in config.yaml. The built-in +``memory`` tool (MEMORY.md / USER.md) is unchanged and keeps working; writes +to it are mirrored into the graph via the ``on_memory_write`` hook. + +Optional config file ``$HERMES_HOME/memorygraph.json``: + { + "db_path": "...", # default: $HERMES_HOME/memory_graph.db + "half_life_days": 90, # knowledge aging half-life + "duplicate_similarity": 0.88, # near-duplicate threshold [0, 1] + "prefetch_enabled": true, # inject graph recall before each turn + "sweep_on_session_end": true # run aging+promotion at session end + } + +See DESIGN.md in this directory for the governance model and the +design-only roadmap (semantic dedupe, automatic turn extraction, federation). +""" + +from __future__ import annotations + +import json +import logging +import os +import threading +from typing import Any, Dict, List, Optional + +from agent.memory_provider import MemoryProvider + +from .governance import GovernanceEngine, GovernancePolicy +from .store import ENTITY_TYPES, GraphStore + +logger = logging.getLogger(__name__) + +_CONFIG_FILENAME = "memorygraph.json" +_DB_FILENAME = "memory_graph.db" + +GRAPH_MEMORY_SCHEMA: Dict[str, Any] = { + "name": "graph_memory", + "description": ( + "Governed knowledge graph memory. Use alongside the built-in memory " + "tool — memory for always-on notes, graph_memory for structured, " + "evidence-linked knowledge about people, projects, goals, skills and " + "businesses.\n\n" + "ACTIONS:\n" + "• remember — Store a claim about an entity (creates the entity if " + "needed). Duplicates reinforce; conflicting exclusive values are " + "superseded time-aware or flagged as contradictions.\n" + "• link — Create a typed relationship between two entities.\n" + "• unlink — Close a relationship's validity window (it happened, " + "but is no longer current).\n" + "• about — Everything known about an entity: claims by tier, open " + "relationships, evidence counts.\n" + "• query — Keyword search across entities and claims.\n" + "• timeline — Chronological claim history for an entity, including " + "superseded values (time-aware knowledge).\n" + "• contradictions — List contradicted claim groups needing review.\n" + "• resolve — Resolve a contradiction by choosing the winning claim.\n" + "• duplicates — Audit for near-duplicate active claims.\n" + "• feedback — Rate a claim after use (helpful/unhelpful) to tune " + "confidence.\n" + "• forget — Retract a claim (kept in history, never surfaced).\n" + "• sweep — Run governance now: aging (confidence decay + demotion) " + "then promotion (candidate → established → core).\n" + "• stats — Graph size, tier distribution, governance counters.\n\n" + "Before answering questions about known people or projects, use " + "'about' or 'query' first." + ), + "parameters": { + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": [ + "remember", "link", "unlink", "about", "query", "timeline", + "contradictions", "resolve", "duplicates", "feedback", + "forget", "sweep", "stats", + ], + }, + "entity": {"type": "string", "description": "Entity name (remember/about/timeline)."}, + "entity_type": { + "type": "string", + "enum": ENTITY_TYPES, + "description": "Entity type hint (default: concept).", + }, + "attribute": { + "type": "string", + "description": "Claim attribute, e.g. 'employer', 'status', 'deadline' (remember/timeline).", + }, + "value": {"type": "string", "description": "Claim value (remember)."}, + "exclusive": { + "type": "boolean", + "description": "True if the attribute holds one current value (e.g. employer, status).", + }, + "confidence": {"type": "number", "description": "Initial confidence 0-1 (default 0.6)."}, + "source": {"type": "string", "description": "Evidence reference: URL, file, quote origin."}, + "quote": {"type": "string", "description": "Supporting quote for the evidence link."}, + "src": {"type": "string", "description": "Source entity name (link)."}, + "dst": {"type": "string", "description": "Target entity name (link)."}, + "rel_type": { + "type": "string", + "description": "Relationship type, e.g. 'works_at', 'owns', 'part_of' (link).", + }, + "relationship_id": {"type": "integer", "description": "Relationship id (unlink)."}, + "claim_id": {"type": "integer", "description": "Claim id (feedback/forget/resolve)."}, + "helpful": {"type": "boolean", "description": "Feedback direction (feedback)."}, + "query": {"type": "string", "description": "Search text (query)."}, + "limit": {"type": "integer", "description": "Max results (default 10)."}, + }, + "required": ["action"], + }, +} + + +def _coerce_bool(raw: Any, default: bool = False) -> bool: + """Robust boolean coercion for tool args — bool("false") is True.""" + if isinstance(raw, bool): + return raw + if isinstance(raw, str): + return raw.strip().lower() in {"1", "true", "yes", "on"} + if isinstance(raw, (int, float)): + return bool(raw) + return default + + +def _load_json_config(hermes_home: str) -> Dict[str, Any]: + path = os.path.join(hermes_home, _CONFIG_FILENAME) + try: + with open(path, encoding="utf-8") as f: + data = json.load(f) + return data if isinstance(data, dict) else {} + except (OSError, ValueError): + return {} + + +class MemoryGraphProvider(MemoryProvider): + """Governed knowledge graph memory provider (local SQLite).""" + + def __init__(self): + self._store: Optional[GraphStore] = None + self._engine: Optional[GovernanceEngine] = None + self._session_id: str = "" + self._hermes_home: str = "" + self._agent_context: str = "primary" + self._prefetch_enabled: bool = True + self._sweep_on_session_end: bool = True + self._prefetch_cache: Dict[str, str] = {} + self._lock = threading.Lock() + + @property + def store(self) -> GraphStore: + """Initialized store accessor — narrows Optional for callers.""" + if self._store is None: + raise ValueError("memorygraph is not initialized") + return self._store + + @property + def engine(self) -> GovernanceEngine: + """Initialized governance engine accessor.""" + if self._engine is None: + raise ValueError("memorygraph is not initialized") + return self._engine + + # -- identity / availability ------------------------------------------------ + + @property + def name(self) -> str: + return "memorygraph" + + def is_available(self) -> bool: + # Local, stdlib-only: always available. No network, no credentials. + return True + + # -- lifecycle ---------------------------------------------------------------- + + def initialize(self, session_id: str, **kwargs) -> None: + self._session_id = session_id + self._hermes_home = str( + kwargs.get("hermes_home") or os.path.expanduser("~/.hermes") + ) + self._agent_context = str(kwargs.get("agent_context") or "primary") + config = _load_json_config(self._hermes_home) + + db_path = str(config.get("db_path") or os.path.join(self._hermes_home, _DB_FILENAME)) + policy = GovernancePolicy() + try: + # Clamp to sane ranges: a zero/negative half-life would invert + # the decay curve; similarity must stay within [0, 1]. + policy.half_life_days = max( + 0.1, float(config.get("half_life_days", policy.half_life_days)) + ) + policy.duplicate_similarity = min( + 1.0, + max(0.0, float(config.get("duplicate_similarity", + policy.duplicate_similarity))), + ) + except (TypeError, ValueError): + logger.warning("memorygraph: invalid numeric config value; using defaults") + self._prefetch_enabled = bool(config.get("prefetch_enabled", True)) + self._sweep_on_session_end = bool(config.get("sweep_on_session_end", True)) + + self._store = GraphStore(db_path) + self._engine = GovernanceEngine(self._store, policy) + logger.info("memorygraph initialized (db=%s, session=%s)", db_path, session_id) + + def shutdown(self) -> None: + with self._lock: + if self._store is not None: + self.store.close() + self._store = None + self._engine = None + + def on_session_switch(self, new_session_id: str, **kwargs) -> None: + self._session_id = new_session_id + self._prefetch_cache.clear() + + def on_session_end(self, messages: List[Dict[str, Any]]) -> None: + if self._engine is None or not self._sweep_on_session_end: + return + if self._agent_context != "primary": + return + try: + # Hold the provider lock so shutdown() cannot close the store + # mid-sweep (MemoryManager drains background hooks with a + # bounded timeout before calling shutdown()). + with self._lock: + if self._engine is None: + return + result = self._engine.sweep() + logger.debug("memorygraph session-end sweep: %s", result) + except Exception: + logger.exception("memorygraph: session-end governance sweep failed") + + # -- prompt / recall ------------------------------------------------------------ + + def system_prompt_block(self) -> str: + if self._store is None: + return "" + try: + stats = self.store.stats() + except Exception: + return "" + active = sum(stats["active_claims_by_tier"].values()) + contradicted = stats["claims_by_status"].get("contradicted", 0) + lines = [ + "## Knowledge Graph Memory (memorygraph)", + f"Graph: {stats['entities']} entities, {stats['open_relationships']} " + f"relationships, {active} active claims.", + "Use the graph_memory tool to recall ('about', 'query', 'timeline') " + "before answering questions about known people, projects, goals, " + "skills or businesses, and to store new durable knowledge " + "('remember', 'link').", + ] + if contradicted: + lines.append( + f"⚠ {contradicted} contradicted claims await review — " + "use graph_memory action='contradictions' then 'resolve'." + ) + return "\n".join(lines) + + def prefetch(self, query: str, *, session_id: str = "") -> str: + if self._store is None or not self._prefetch_enabled or not query: + return "" + try: + hits = self.store.search(query, limit=5) + except Exception: + logger.exception("memorygraph: prefetch search failed") + return "" + if not hits: + return "" + lines = ["[memorygraph recall]"] + for hit in hits: + if hit["_kind"] == "entity": + lines.append(f"- entity: {hit['name']} ({hit['type']}) — {hit['summary']}".rstrip(" —")) + else: + lines.append( + f"- {hit['entity_name']}.{hit['attribute']} = {hit['value']} " + f"(confidence {float(hit['confidence']):.2f}, {hit['tier']})" + ) + return "\n".join(lines) + + # -- built-in memory mirroring --------------------------------------------------- + + def on_memory_write( + self, + action: str, + target: str, + content: str, + metadata: Optional[Dict[str, Any]] = None, + ) -> None: + """Mirror built-in MEMORY.md / USER.md writes into the graph. + + Mirrored entries land as claims on the 'Hermes Notes' or 'User' + entity with evidence kind 'builtin_memory', so file memory and + graph memory stay consistent without changing built-in behavior. + """ + if self._engine is None or self._store is None: + return + if action not in ("add", "replace") or not (content or "").strip(): + return + try: + entity_name = "User" if target == "user" else "Hermes Notes" + entity_type = "person" if target == "user" else "concept" + entity = self.store.upsert_entity(entity_name, entity_type) + session_id = str((metadata or {}).get("session_id") or self._session_id) + self.engine.assert_claim( + entity["id"], "note", content.strip(), confidence=0.6, + evidence={ + "kind": "builtin_memory", + "ref": f"{target}:{action}", + "quote": content.strip()[:200], + "session_id": session_id, + }, + ) + except Exception: + logger.exception("memorygraph: failed to mirror built-in memory write") + + # -- tools ----------------------------------------------------------------------- + + def get_tool_schemas(self) -> List[Dict[str, Any]]: + return [GRAPH_MEMORY_SCHEMA] + + def handle_tool_call(self, tool_name: str, args: Dict[str, Any], **kwargs) -> str: + if tool_name != "graph_memory": + return json.dumps({"error": f"unknown tool: {tool_name}"}) + if self._engine is None or self._store is None: + return json.dumps({"error": "memorygraph is not initialized"}) + action = str(args.get("action") or "") + handler = getattr(self, f"_action_{action}", None) + if handler is None: + return json.dumps({"error": f"unknown action: {action}"}) + try: + result = handler(args) + except ValueError as e: + return json.dumps({"error": str(e)}) + except Exception: + logger.exception("memorygraph: action %s failed", action) + return json.dumps({"error": f"action {action} failed; see logs"}) + return json.dumps(result, ensure_ascii=False, default=str) + + # Individual actions. Each returns a JSON-serializable dict. + + def _require(self, args: Dict[str, Any], *names: str) -> List[Any]: + values = [] + for n in names: + v = args.get(n) + if v is None or (isinstance(v, str) and not v.strip()): + raise ValueError(f"'{n}' is required for this action") + values.append(v) + return values + + def _evidence_from_args(self, args: Dict[str, Any]) -> Dict[str, Any]: + return { + "kind": "tool", + "ref": str(args.get("source") or ""), + "quote": str(args.get("quote") or ""), + "session_id": self._session_id, + } + + def _action_remember(self, args: Dict[str, Any]) -> Dict[str, Any]: + entity_name, attribute, value = self._require(args, "entity", "attribute", "value") + entity = self.store.upsert_entity( + str(entity_name), str(args.get("entity_type") or "concept") + ) + result = self.engine.assert_claim( + entity["id"], + str(attribute), + str(value), + confidence=float(args.get("confidence") or 0.6), + exclusive=_coerce_bool(args.get("exclusive")), + evidence=self._evidence_from_args(args), + ) + return { + "outcome": result["outcome"], + "claim": result["claim"], + "entity": {"id": entity["id"], "name": entity["name"], "type": entity["type"]}, + "conflicts": result.get("conflicts", []), + } + + def _action_link(self, args: Dict[str, Any]) -> Dict[str, Any]: + src_name, dst_name, rel_type = self._require(args, "src", "dst", "rel_type") + src = self.store.upsert_entity(str(src_name), str(args.get("entity_type") or "concept")) + dst = self.store.upsert_entity(str(dst_name)) + rel = self.store.add_relationship( + src["id"], dst["id"], str(rel_type), + confidence=float(args.get("confidence") or 0.6), + ) + self.store.add_evidence("relationship", rel["id"], **self._evidence_from_args(args)) + return {"relationship": rel, "src": src["name"], "dst": dst["name"]} + + def _action_unlink(self, args: Dict[str, Any]) -> Dict[str, Any]: + (rel_id,) = self._require(args, "relationship_id") + ok = self.store.end_relationship(int(rel_id)) + return {"ended": ok, "relationship_id": int(rel_id)} + + def _action_about(self, args: Dict[str, Any]) -> Dict[str, Any]: + (entity_name,) = self._require(args, "entity") + entity = self.store.resolve_entity(str(entity_name)) + if not entity: + return {"found": False, "entity": str(entity_name)} + claims = self.store.claims_for(entity["id"]) + for claim in claims: + claim["evidence_count"] = self.store.evidence_count("claim", claim["id"]) + return { + "found": True, + "entity": entity, + "claims": sorted(claims, key=lambda c: (c["tier"] != "core", + c["tier"] != "established", + -float(c["confidence"]))), + "relationships": self.store.relationships_for(entity["id"]), + } + + def _action_query(self, args: Dict[str, Any]) -> Dict[str, Any]: + (query,) = self._require(args, "query") + limit = int(args.get("limit") or 10) + return {"results": self.store.search(str(query), limit=limit)} + + def _action_timeline(self, args: Dict[str, Any]) -> Dict[str, Any]: + (entity_name,) = self._require(args, "entity") + entity = self.store.resolve_entity(str(entity_name)) + if not entity: + return {"found": False, "entity": str(entity_name)} + claims = self.store.claims_for( + entity["id"], attribute=str(args.get("attribute") or ""), include_history=True + ) + return {"found": True, "entity": entity["name"], "timeline": claims} + + def _action_contradictions(self, args: Dict[str, Any]) -> Dict[str, Any]: + return {"groups": self.engine.find_contradictions()} + + def _action_resolve(self, args: Dict[str, Any]) -> Dict[str, Any]: + (claim_id,) = self._require(args, "claim_id") + return self.engine.resolve_contradiction(int(claim_id)) + + def _action_duplicates(self, args: Dict[str, Any]) -> Dict[str, Any]: + return {"groups": self.engine.find_duplicates()} + + def _action_feedback(self, args: Dict[str, Any]) -> Dict[str, Any]: + (claim_id,) = self._require(args, "claim_id") + if args.get("helpful") is None: + raise ValueError("'helpful' is required for this action") + claim = self.engine.feedback(int(claim_id), _coerce_bool(args["helpful"])) + if claim is None: + raise ValueError(f"claim {claim_id} not found") + return {"claim": claim} + + def _action_forget(self, args: Dict[str, Any]) -> Dict[str, Any]: + (claim_id,) = self._require(args, "claim_id") + ok = self.engine.retract(int(claim_id), reason="user request") + return {"retracted": ok, "claim_id": int(claim_id)} + + def _action_sweep(self, args: Dict[str, Any]) -> Dict[str, Any]: + return self.engine.sweep() + + def _action_stats(self, args: Dict[str, Any]) -> Dict[str, Any]: + stats = self.store.stats() + stats["recent_governance_events"] = [ + {"ts": e["ts"], "event": e["event"], "subject_id": e["subject_id"]} + for e in self.store.recent_events(10) + ] + return stats + + # -- setup wizard ------------------------------------------------------------------- + + def get_config_schema(self) -> List[Dict[str, Any]]: + return [ + { + "key": "half_life_days", + "description": "Knowledge aging half-life in days (confidence decay).", + "required": False, + "default": "90", + }, + { + "key": "duplicate_similarity", + "description": "Near-duplicate similarity threshold (0-1).", + "required": False, + "default": "0.88", + }, + ] + + def save_config(self, values: Dict[str, Any], hermes_home: str) -> None: + path = os.path.join(hermes_home, _CONFIG_FILENAME) + config = _load_json_config(hermes_home) + for key in ("half_life_days", "duplicate_similarity"): + if key in values and values[key] not in (None, ""): + try: + config[key] = float(values[key]) + except (TypeError, ValueError): + continue + os.makedirs(hermes_home, exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + json.dump(config, f, indent=2) + logger.info("memorygraph: config saved to %s", path) + + +def register(ctx) -> None: + """Plugin entry point — called by the memory provider loader.""" + ctx.register_memory_provider(MemoryGraphProvider()) diff --git a/plugins/memory/memorygraph/governance.py b/plugins/memory/memorygraph/governance.py new file mode 100644 index 000000000000..52dd670cd10c --- /dev/null +++ b/plugins/memory/memorygraph/governance.py @@ -0,0 +1,351 @@ +"""Governance engine for the memorygraph provider. + +Turns raw writes into *governed* knowledge: + + duplicate detection — re-asserting known knowledge reinforces instead + of duplicating (normalized-key + fuzzy match) + contradiction detection — conflicting values for an exclusive attribute are + either superseded (time-aware update) or flagged + as contradicted for review + confidence tracking — reinforcement raises confidence, feedback and + contradiction lower it, all clamped to [0, 1] + knowledge aging — confidence decays with a configurable half-life + since last reinforcement; decayed core/established + knowledge is demoted + knowledge promotion — candidate → established → core, gated on + confidence, evidence count, reinforcement count + and age, never while contradicted + +Every governed mutation is written to the store's governance_log so the +graph stays auditable. +""" + +from __future__ import annotations + +import difflib +from dataclasses import dataclass +from typing import Any, Dict, List, Optional + +from .store import GraphStore, normalize_key, normalize_ts, parse_ts + + +@dataclass +class GovernancePolicy: + """Tunable thresholds. Defaults are deliberately conservative.""" + + # Duplicate detection + duplicate_similarity: float = 0.88 + + # Confidence tracking + reinforce_delta: float = 0.10 + contradiction_penalty: float = 0.15 + feedback_delta: float = 0.15 + min_confidence: float = 0.0 + max_confidence: float = 1.0 + + # Aging + half_life_days: float = 90.0 + stale_confidence_floor: float = 0.15 + demotion_confidence: float = 0.40 + + # Promotion + promote_established_confidence: float = 0.70 + promote_established_evidence: int = 2 + promote_core_confidence: float = 0.85 + promote_core_reinforcements: int = 3 + promote_core_min_age_days: float = 7.0 + + +def _clamp(value: float, lo: float = 0.0, hi: float = 1.0) -> float: + return max(lo, min(hi, value)) + + +def text_similarity(a: str, b: str) -> float: + """Similarity in [0, 1] combining token Jaccard and sequence ratio.""" + ka, kb = normalize_key(a), normalize_key(b) + if not ka or not kb: + return 0.0 + if ka == kb: + return 1.0 + ta, tb = set(ka.split()), set(kb.split()) + jaccard = len(ta & tb) / len(ta | tb) if (ta | tb) else 0.0 + ratio = difflib.SequenceMatcher(None, ka, kb).ratio() + return max(jaccard, ratio) + + +class GovernanceEngine: + """Applies governance policy to a GraphStore.""" + + def __init__(self, store: GraphStore, policy: Optional[GovernancePolicy] = None): + self.store = store + self.policy = policy or GovernancePolicy() + + # -- governed write ------------------------------------------------------ + + def assert_claim( + self, + entity_id: int, + attribute: str, + value: str, + confidence: float = 0.6, + exclusive: bool = False, + valid_from: str = "", + evidence: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """Governed claim write. Returns {claim, outcome, [conflicts]}. + + Outcomes: + created — new knowledge + reinforced — duplicate of existing active claim (confidence up) + superseded — exclusive attribute updated time-aware (old claim's + validity window closed, linked via superseded_by) + contradicted — conflicting exclusive values with no clear temporal + ordering; both flagged for review + """ + attribute_key = normalize_key(attribute) or "note" + existing = self.store.claims_for(entity_id, attribute_key, status="active") + + # 1. Duplicate detection → reinforcement. + # Exclusive (single-valued) attributes hold short distinguishing + # values ("Facility A" vs "Facility B") where fuzzy matching would + # swallow genuine updates — restrict them to exact-key matches. + exact_only = exclusive or any(c["exclusive"] for c in existing) + dup = self._find_duplicate(existing, value, exact_only=exact_only) + if dup is not None: + reinforced = self.reinforce(dup["id"]) + if evidence: + self.store.add_evidence("claim", dup["id"], **evidence) + self.store.log_event("claim_reinforced", "claim", dup["id"], + {"similarity_value": value}) + return {"claim": reinforced, "outcome": "reinforced"} + + # 2. Contradiction detection (exclusive = single-valued attribute) + conflicts = [c for c in existing if c["exclusive"] or exclusive] + if conflicts and (exclusive or any(c["exclusive"] for c in conflicts)): + new_from = normalize_ts(valid_from, self.store.now()) + newer = all(new_from >= c["valid_from"] for c in conflicts) + if newer: + # Time-aware update: new value supersedes older ones. + claim = self.store.insert_claim( + entity_id, attribute_key, value, confidence, + exclusive=True, valid_from=new_from, + ) + for old in conflicts: + self.store.update_claim( + old["id"], status="superseded", + valid_to=new_from, superseded_by=claim["id"], + ) + self.store.log_event("claim_superseded", "claim", old["id"], + {"by": claim["id"]}) + if evidence: + self.store.add_evidence("claim", claim["id"], **evidence) + return { + "claim": self.store.get_claim(claim["id"]), + "outcome": "superseded", + "conflicts": [old["id"] for old in conflicts], + } + # Ambiguous temporal ordering — flag everything for review. + claim = self.store.insert_claim( + entity_id, attribute_key, value, confidence, + exclusive=True, valid_from=new_from, + ) + penalty = self.policy.contradiction_penalty + for c in [claim, *conflicts]: + self.store.update_claim( + c["id"], status="contradicted", + confidence=_clamp(float(c["confidence"]) - penalty), + ) + self.store.log_event("claim_contradicted", "claim", c["id"], + {"group": [claim["id"], *(x["id"] for x in conflicts)]}) + if evidence: + self.store.add_evidence("claim", claim["id"], **evidence) + return { + "claim": self.store.get_claim(claim["id"]), + "outcome": "contradicted", + "conflicts": [c["id"] for c in conflicts], + } + + # 3. Plain create + claim = self.store.insert_claim( + entity_id, attribute_key, value, confidence, + exclusive=exclusive, valid_from=valid_from, + ) + if evidence: + self.store.add_evidence("claim", claim["id"], **evidence) + return {"claim": claim, "outcome": "created"} + + def _find_duplicate( + self, + existing: List[Dict[str, Any]], + value: str, + exact_only: bool = False, + ) -> Optional[Dict[str, Any]]: + value_key = normalize_key(value) + best, best_score = None, 0.0 + for claim in existing: + if claim["value_key"] == value_key: + return claim + if exact_only: + continue + score = text_similarity(claim["value"], value) + if score > best_score: + best, best_score = claim, score + if best is not None and best_score >= self.policy.duplicate_similarity: + return best + return None + + # -- confidence tracking --------------------------------------------------- + + def reinforce(self, claim_id: int) -> Optional[Dict[str, Any]]: + claim = self.store.get_claim(claim_id) + if not claim: + return None + self.store.update_claim( + claim_id, + confidence=_clamp(float(claim["confidence"]) + self.policy.reinforce_delta), + reinforcement_count=int(claim["reinforcement_count"]) + 1, + last_reinforced_at=self.store.now(), + ) + return self.store.get_claim(claim_id) + + def feedback(self, claim_id: int, helpful: bool) -> Optional[Dict[str, Any]]: + claim = self.store.get_claim(claim_id) + if not claim: + return None + delta = self.policy.feedback_delta if helpful else -self.policy.feedback_delta + fields: Dict[str, Any] = {"confidence": _clamp(float(claim["confidence"]) + delta)} + if helpful: + fields["last_reinforced_at"] = self.store.now() + fields["reinforcement_count"] = int(claim["reinforcement_count"]) + 1 + self.store.update_claim(claim_id, **fields) + self.store.log_event("claim_feedback", "claim", claim_id, {"helpful": helpful}) + return self.store.get_claim(claim_id) + + def retract(self, claim_id: int, reason: str = "") -> bool: + claim = self.store.get_claim(claim_id) + if not claim or claim["status"] == "retracted": + return False + self.store.update_claim(claim_id, status="retracted", valid_to=self.store.now()) + self.store.log_event("claim_retracted", "claim", claim_id, {"reason": reason}) + return True + + def resolve_contradiction(self, winner_id: int) -> Dict[str, Any]: + """Keep one claim from a contradicted group; supersede the rest.""" + winner = self.store.get_claim(winner_id) + if not winner: + return {"error": f"claim {winner_id} not found"} + losers = [ + c for c in self.store.claims_for( + winner["entity_id"], winner["attribute"], status="contradicted") + if c["id"] != winner_id + ] + self.store.update_claim(winner_id, status="active") + for loser in losers: + self.store.update_claim( + loser["id"], status="superseded", + valid_to=self.store.now(), superseded_by=winner_id, + ) + self.store.log_event("contradiction_resolved", "claim", winner_id, + {"superseded": [loser["id"] for loser in losers]}) + return {"winner": self.store.get_claim(winner_id), + "superseded": [loser["id"] for loser in losers]} + + # -- audits ------------------------------------------------------------------ + + def find_contradictions(self) -> List[List[Dict[str, Any]]]: + """Groups of claims currently flagged as contradicted.""" + flagged = self.store.claims_by_status("contradicted") + groups: Dict[tuple, List[Dict[str, Any]]] = {} + for claim in flagged: + groups.setdefault((claim["entity_id"], claim["attribute"]), []).append(claim) + return [g for g in groups.values() if len(g) >= 2] + + def find_duplicates(self) -> List[List[Dict[str, Any]]]: + """Near-duplicate active claim groups that slipped past write-time checks.""" + by_bucket: Dict[tuple, List[Dict[str, Any]]] = {} + for claim in self.store.all_active_claims(): + by_bucket.setdefault((claim["entity_id"], claim["attribute"]), []).append(claim) + out: List[List[Dict[str, Any]]] = [] + for claims in by_bucket.values(): + if len(claims) < 2: + continue + used: set = set() + for i, a in enumerate(claims): + if a["id"] in used: + continue + group = [a] + for b in claims[i + 1:]: + if b["id"] in used: + continue + if text_similarity(a["value"], b["value"]) >= self.policy.duplicate_similarity: + group.append(b) + used.add(b["id"]) + if len(group) >= 2: + used.add(a["id"]) + out.append(group) + return out + + # -- aging + promotion sweeps ----------------------------------------------- + # + # Sweeps are O(active claims) full scans (promotion also does a per-claim + # evidence-count query). Fine for a personal knowledge graph running at + # session end; revisit with batched queries past a few thousand claims. + + def age_knowledge(self, now: str = "") -> Dict[str, int]: + """Decay confidence by half-life since last reinforcement; demote decayed.""" + now_ts = parse_ts(now or self.store.now()) + half_life = max(1e-6, self.policy.half_life_days) + decayed = demoted = 0 + for claim in self.store.all_active_claims(): + age_days = (now_ts - parse_ts(claim["last_reinforced_at"])).total_seconds() / 86400.0 + if age_days <= 0: + continue + factor = 0.5 ** (age_days / half_life) + new_conf = max( + self.policy.stale_confidence_floor, + round(float(claim["confidence"]) * factor, 4), + ) + if new_conf >= float(claim["confidence"]): + continue + fields: Dict[str, Any] = {"confidence": new_conf} + if (claim["tier"] in ("established", "core") + and new_conf < self.policy.demotion_confidence): + demote_to = "established" if claim["tier"] == "core" else "candidate" + fields["tier"] = demote_to + demoted += 1 + self.store.log_event("claim_demoted", "claim", claim["id"], + {"from": claim["tier"], "to": demote_to}) + self.store.update_claim(claim["id"], **fields) + decayed += 1 + self.store.set_meta("last_aging_run", now or self.store.now()) + return {"decayed": decayed, "demoted": demoted} + + def promote_knowledge(self, now: str = "") -> Dict[str, int]: + """Promote well-evidenced, reinforced, uncontradicted knowledge.""" + now_ts = parse_ts(now or self.store.now()) + promoted_established = promoted_core = 0 + for claim in self.store.all_active_claims(): + conf = float(claim["confidence"]) + if claim["tier"] == "candidate": + evidence_n = self.store.evidence_count("claim", claim["id"]) + if (conf >= self.policy.promote_established_confidence + and evidence_n >= self.policy.promote_established_evidence): + self.store.update_claim(claim["id"], tier="established") + self.store.log_event("claim_promoted", "claim", claim["id"], + {"to": "established"}) + promoted_established += 1 + elif claim["tier"] == "established": + age_days = (now_ts - parse_ts(claim["created_at"])).total_seconds() / 86400.0 + if (conf >= self.policy.promote_core_confidence + and int(claim["reinforcement_count"]) >= self.policy.promote_core_reinforcements + and age_days >= self.policy.promote_core_min_age_days): + self.store.update_claim(claim["id"], tier="core") + self.store.log_event("claim_promoted", "claim", claim["id"], {"to": "core"}) + promoted_core += 1 + return {"established": promoted_established, "core": promoted_core} + + def sweep(self, now: str = "") -> Dict[str, Any]: + """Full governance sweep: aging then promotion.""" + aged = self.age_knowledge(now) + promoted = self.promote_knowledge(now) + return {"aging": aged, "promotion": promoted} diff --git a/plugins/memory/memorygraph/plugin.yaml b/plugins/memory/memorygraph/plugin.yaml new file mode 100644 index 000000000000..c5c8c18b1249 --- /dev/null +++ b/plugins/memory/memorygraph/plugin.yaml @@ -0,0 +1,7 @@ +name: memorygraph +version: 0.1.0 +description: "Governed knowledge graph memory — typed entities and time-aware relationships with evidence links, confidence tracking, contradiction/duplicate detection, knowledge aging and promotion. Local SQLite, no credentials." +hooks: + - on_memory_write + - on_session_end + - on_session_switch diff --git a/plugins/memory/memorygraph/store.py b/plugins/memory/memorygraph/store.py new file mode 100644 index 000000000000..5b5bf8d71c03 --- /dev/null +++ b/plugins/memory/memorygraph/store.py @@ -0,0 +1,610 @@ +"""SQLite-backed governed knowledge graph store for the memorygraph provider. + +Schema overview (schema_version 1): + + entities — typed nodes (person, project, goal, skill, business, ...) + entity_aliases — alternate names resolving to a canonical entity + relationships — typed, time-aware, confidence-scored edges + claims — attribute/value knowledge about an entity with + confidence, tier (candidate/established/core), status + (active/superseded/retracted/contradicted) and temporal + validity (valid_from/valid_to) + evidence — provenance links for claims / relationships / entities + governance_log — append-only audit trail of every governed mutation + meta — schema version and sweep bookkeeping + +All timestamps are UTC ISO-8601 strings. The store takes an optional +``now`` callable so tests can control the clock deterministically. + +The store is deliberately stdlib-only (sqlite3) and profile-scoped: the +database lives under ``$HERMES_HOME`` by default so it never leaks across +Hermes profiles. +""" + +from __future__ import annotations + +import json +import re +import sqlite3 +import threading +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional + +SCHEMA_VERSION = 1 + +ENTITY_TYPES = [ + "person", + "project", + "goal", + "skill", + "business", + "organization", + "place", + "tool", + "concept", +] + +CLAIM_TIERS = ["candidate", "established", "core"] +CLAIM_STATUSES = ["active", "superseded", "retracted", "contradicted"] + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS entities ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + name_key TEXT NOT NULL, + type TEXT NOT NULL DEFAULT 'concept', + summary TEXT NOT NULL DEFAULT '', + attrs TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE (name_key, type) +); +CREATE INDEX IF NOT EXISTS idx_entities_name_key ON entities (name_key); + +CREATE TABLE IF NOT EXISTS entity_aliases ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + entity_id INTEGER NOT NULL REFERENCES entities (id) ON DELETE CASCADE, + alias TEXT NOT NULL, + alias_key TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE (entity_id, alias_key) +); +CREATE INDEX IF NOT EXISTS idx_aliases_key ON entity_aliases (alias_key); + +CREATE TABLE IF NOT EXISTS relationships ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + src_id INTEGER NOT NULL REFERENCES entities (id) ON DELETE CASCADE, + dst_id INTEGER NOT NULL REFERENCES entities (id) ON DELETE CASCADE, + rel_type TEXT NOT NULL, + confidence REAL NOT NULL DEFAULT 0.6, + valid_from TEXT NOT NULL, + valid_to TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_rel_src ON relationships (src_id); +CREATE INDEX IF NOT EXISTS idx_rel_dst ON relationships (dst_id); + +CREATE TABLE IF NOT EXISTS claims ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + entity_id INTEGER NOT NULL REFERENCES entities (id) ON DELETE CASCADE, + attribute TEXT NOT NULL, + value TEXT NOT NULL, + value_key TEXT NOT NULL, + confidence REAL NOT NULL DEFAULT 0.6, + tier TEXT NOT NULL DEFAULT 'candidate', + status TEXT NOT NULL DEFAULT 'active', + exclusive INTEGER NOT NULL DEFAULT 0, + reinforcement_count INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + last_reinforced_at TEXT NOT NULL, + valid_from TEXT NOT NULL, + valid_to TEXT, + superseded_by INTEGER +); +CREATE INDEX IF NOT EXISTS idx_claims_entity ON claims (entity_id); +CREATE INDEX IF NOT EXISTS idx_claims_attr ON claims (entity_id, attribute); +CREATE INDEX IF NOT EXISTS idx_claims_status ON claims (status); + +CREATE TABLE IF NOT EXISTS evidence ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + subject_kind TEXT NOT NULL, + subject_id INTEGER NOT NULL, + kind TEXT NOT NULL DEFAULT 'session', + ref TEXT NOT NULL DEFAULT '', + quote TEXT NOT NULL DEFAULT '', + session_id TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_evidence_subject ON evidence (subject_kind, subject_id); + +CREATE TABLE IF NOT EXISTS governance_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, + event TEXT NOT NULL, + subject_kind TEXT NOT NULL DEFAULT '', + subject_id INTEGER, + details TEXT NOT NULL DEFAULT '' +); + +CREATE TABLE IF NOT EXISTS meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL +); +""" + +_KEY_RE = re.compile(r"[^a-z0-9]+") + +# Confidence bump when an identical open relationship is re-asserted. +# Mirrors GovernancePolicy.reinforce_delta for claims; kept here because +# the store is policy-free by design. +_REL_REINFORCE_DELTA = 0.1 + + +def normalize_key(text: str) -> str: + """Normalize a name/value into a stable comparison key.""" + return _KEY_RE.sub(" ", (text or "").lower()).strip() + + +def utcnow_iso() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def parse_ts(ts: str) -> datetime: + """Parse a stored ISO-8601 UTC timestamp back into a datetime.""" + return datetime.strptime(ts, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) + + +def normalize_ts(ts: str, fallback: str) -> str: + """Return ``ts`` if it is a valid stored-format timestamp, else ``fallback``. + + Timestamp ordering elsewhere relies on lexicographic comparison of the + fixed ``%Y-%m-%dT%H:%M:%SZ`` format, so caller-supplied ``valid_from`` + values must be validated at the write boundary. + """ + if not ts: + return fallback + try: + parse_ts(ts) + return ts + except ValueError: + return fallback + + +def _row_to_dict(row: sqlite3.Row) -> Dict[str, Any]: + return {k: row[k] for k in row.keys()} + + +class GraphStore: + """Thread-safe SQLite knowledge graph store.""" + + def __init__(self, db_path: str, now: Optional[Callable[[], str]] = None): + self.db_path = str(db_path) + self._now = now or utcnow_iso + self._lock = threading.RLock() + Path(self.db_path).parent.mkdir(parents=True, exist_ok=True) + self._conn = sqlite3.connect(self.db_path, check_same_thread=False) + self._conn.row_factory = sqlite3.Row + self._conn.execute("PRAGMA foreign_keys = ON") + with self._lock, self._conn: + self._conn.executescript(_SCHEMA) + self._conn.execute( + "INSERT OR IGNORE INTO meta (key, value) VALUES ('schema_version', ?)", + (str(SCHEMA_VERSION),), + ) + + def close(self) -> None: + with self._lock: + self._conn.close() + + def now(self) -> str: + return self._now() + + # -- meta / audit ------------------------------------------------------ + + def get_meta(self, key: str, default: str = "") -> str: + with self._lock: + row = self._conn.execute("SELECT value FROM meta WHERE key = ?", (key,)).fetchone() + return row["value"] if row else default + + def set_meta(self, key: str, value: str) -> None: + with self._lock, self._conn: + self._conn.execute( + "INSERT INTO meta (key, value) VALUES (?, ?) " + "ON CONFLICT (key) DO UPDATE SET value = excluded.value", + (key, value), + ) + + def log_event( + self, + event: str, + subject_kind: str = "", + subject_id: Optional[int] = None, + details: Optional[Dict[str, Any]] = None, + ) -> None: + with self._lock, self._conn: + self._conn.execute( + "INSERT INTO governance_log (ts, event, subject_kind, subject_id, details) " + "VALUES (?, ?, ?, ?, ?)", + (self.now(), event, subject_kind, subject_id, + json.dumps(details or {}, ensure_ascii=False)), + ) + + def recent_events(self, limit: int = 20) -> List[Dict[str, Any]]: + with self._lock: + rows = self._conn.execute( + "SELECT * FROM governance_log ORDER BY id DESC LIMIT ?", (limit,) + ).fetchall() + return [_row_to_dict(r) for r in rows] + + # -- entities ---------------------------------------------------------- + + def resolve_entity(self, name: str, entity_type: str = "") -> Optional[Dict[str, Any]]: + """Resolve a name (or alias) to an entity dict, or None.""" + key = normalize_key(name) + if not key: + return None + with self._lock: + if entity_type: + row = self._conn.execute( + "SELECT * FROM entities WHERE name_key = ? AND type = ?", + (key, entity_type), + ).fetchone() + else: + row = self._conn.execute( + "SELECT * FROM entities WHERE name_key = ? ORDER BY id LIMIT 1", (key,) + ).fetchone() + if row: + return _row_to_dict(row) + alias = self._conn.execute( + "SELECT e.* FROM entity_aliases a JOIN entities e ON e.id = a.entity_id " + "WHERE a.alias_key = ? ORDER BY a.id LIMIT 1", + (key,), + ).fetchone() + return _row_to_dict(alias) if alias else None + + def upsert_entity( + self, + name: str, + entity_type: str = "concept", + summary: str = "", + attrs: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """Create or update an entity; resolves aliases first.""" + if entity_type not in ENTITY_TYPES: + entity_type = "concept" + key = normalize_key(name) + if not key: + raise ValueError("entity name must be non-empty") + ts = self.now() + existing = self.resolve_entity(name, entity_type) or self.resolve_entity(name) + with self._lock, self._conn: + if existing: + new_summary = summary or existing["summary"] + merged = json.loads(existing["attrs"] or "{}") + merged.update(attrs or {}) + self._conn.execute( + "UPDATE entities SET summary = ?, attrs = ?, updated_at = ? WHERE id = ?", + (new_summary, json.dumps(merged, ensure_ascii=False), ts, existing["id"]), + ) + return self.get_entity(existing["id"]) # type: ignore[return-value] + cur = self._conn.execute( + "INSERT INTO entities (name, name_key, type, summary, attrs, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (name.strip(), key, entity_type, summary, + json.dumps(attrs or {}, ensure_ascii=False), ts, ts), + ) + entity_id = cur.lastrowid + self.log_event("entity_created", "entity", entity_id, {"name": name, "type": entity_type}) + return self.get_entity(entity_id) # type: ignore[return-value] + + def get_entity(self, entity_id: int) -> Optional[Dict[str, Any]]: + with self._lock: + row = self._conn.execute( + "SELECT * FROM entities WHERE id = ?", (entity_id,) + ).fetchone() + return _row_to_dict(row) if row else None + + def add_alias(self, entity_id: int, alias: str) -> bool: + key = normalize_key(alias) + if not key: + return False + with self._lock, self._conn: + try: + self._conn.execute( + "INSERT INTO entity_aliases (entity_id, alias, alias_key, created_at) " + "VALUES (?, ?, ?, ?)", + (entity_id, alias.strip(), key, self.now()), + ) + except sqlite3.IntegrityError: + return False + self.log_event("alias_added", "entity", entity_id, {"alias": alias}) + return True + + def list_entities(self, entity_type: str = "", limit: int = 50) -> List[Dict[str, Any]]: + with self._lock: + if entity_type: + rows = self._conn.execute( + "SELECT * FROM entities WHERE type = ? ORDER BY updated_at DESC LIMIT ?", + (entity_type, limit), + ).fetchall() + else: + rows = self._conn.execute( + "SELECT * FROM entities ORDER BY updated_at DESC LIMIT ?", (limit,) + ).fetchall() + return [_row_to_dict(r) for r in rows] + + # -- relationships ----------------------------------------------------- + + def add_relationship( + self, + src_id: int, + dst_id: int, + rel_type: str, + confidence: float = 0.6, + valid_from: str = "", + ) -> Dict[str, Any]: + """Add a typed edge. Re-adding an identical open edge reinforces it.""" + ts = self.now() + rel_key = normalize_key(rel_type) + with self._lock, self._conn: + row = self._conn.execute( + "SELECT * FROM relationships WHERE src_id = ? AND dst_id = ? " + "AND rel_type = ? AND valid_to IS NULL", + (src_id, dst_id, rel_key), + ).fetchone() + if row: + new_conf = min(1.0, row["confidence"] + _REL_REINFORCE_DELTA) + self._conn.execute( + "UPDATE relationships SET confidence = ?, updated_at = ? WHERE id = ?", + (new_conf, ts, row["id"]), + ) + rel_id = row["id"] + else: + cur = self._conn.execute( + "INSERT INTO relationships " + "(src_id, dst_id, rel_type, confidence, valid_from, valid_to, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, NULL, ?, ?)", + (src_id, dst_id, rel_key, max(0.0, min(1.0, confidence)), + normalize_ts(valid_from, ts), ts, ts), + ) + rel_id = cur.lastrowid + self.log_event("relationship_created", "relationship", rel_id, + {"src": src_id, "dst": dst_id, "type": rel_key}) + out = self._conn.execute( + "SELECT * FROM relationships WHERE id = ?", (rel_id,) + ).fetchone() + return _row_to_dict(out) + + def end_relationship(self, rel_id: int, valid_to: str = "") -> bool: + """Close a relationship's validity window (time-aware retirement).""" + with self._lock, self._conn: + cur = self._conn.execute( + "UPDATE relationships SET valid_to = ?, updated_at = ? " + "WHERE id = ? AND valid_to IS NULL", + (valid_to or self.now(), self.now(), rel_id), + ) + if cur.rowcount: + self.log_event("relationship_ended", "relationship", rel_id, {}) + return bool(cur.rowcount) + + def relationships_for( + self, entity_id: int, include_ended: bool = False + ) -> List[Dict[str, Any]]: + q = ( + "SELECT r.*, s.name AS src_name, d.name AS dst_name FROM relationships r " + "JOIN entities s ON s.id = r.src_id JOIN entities d ON d.id = r.dst_id " + "WHERE (r.src_id = ? OR r.dst_id = ?)" + ) + if not include_ended: + q += " AND r.valid_to IS NULL" + q += " ORDER BY r.updated_at DESC" + with self._lock: + rows = self._conn.execute(q, (entity_id, entity_id)).fetchall() + return [_row_to_dict(r) for r in rows] + + def neighbors(self, entity_id: int) -> List[Dict[str, Any]]: + """Entities directly connected to entity_id via open edges.""" + with self._lock: + rows = self._conn.execute( + "SELECT DISTINCT e.* FROM relationships r " + "JOIN entities e ON e.id = CASE WHEN r.src_id = ? THEN r.dst_id ELSE r.src_id END " + "WHERE (r.src_id = ? OR r.dst_id = ?) AND r.valid_to IS NULL", + (entity_id, entity_id, entity_id), + ).fetchall() + return [_row_to_dict(r) for r in rows] + + # -- claims -------------------------------------------------------------- + + def insert_claim( + self, + entity_id: int, + attribute: str, + value: str, + confidence: float = 0.6, + exclusive: bool = False, + valid_from: str = "", + ) -> Dict[str, Any]: + """Insert a raw claim row (governance handled by GovernanceEngine).""" + ts = self.now() + with self._lock, self._conn: + cur = self._conn.execute( + "INSERT INTO claims (entity_id, attribute, value, value_key, confidence, tier, " + "status, exclusive, reinforcement_count, created_at, updated_at, " + "last_reinforced_at, valid_from, valid_to, superseded_by) " + "VALUES (?, ?, ?, ?, ?, 'candidate', 'active', ?, 0, ?, ?, ?, ?, NULL, NULL)", + (entity_id, normalize_key(attribute) or "note", value.strip(), + normalize_key(value), max(0.0, min(1.0, confidence)), + 1 if exclusive else 0, ts, ts, ts, normalize_ts(valid_from, ts)), + ) + claim_id = cur.lastrowid + self.log_event("claim_created", "claim", claim_id, + {"entity_id": entity_id, "attribute": attribute}) + return self.get_claim(claim_id) # type: ignore[return-value] + + def get_claim(self, claim_id: int) -> Optional[Dict[str, Any]]: + with self._lock: + row = self._conn.execute("SELECT * FROM claims WHERE id = ?", (claim_id,)).fetchone() + return _row_to_dict(row) if row else None + + def update_claim(self, claim_id: int, **fields: Any) -> None: + allowed = { + "value", "value_key", "confidence", "tier", "status", "exclusive", + "reinforcement_count", "last_reinforced_at", "valid_from", "valid_to", + "superseded_by", + } + cols = {k: v for k, v in fields.items() if k in allowed} + if not cols: + return + cols["updated_at"] = self.now() + sets = ", ".join(f"{k} = ?" for k in cols) + with self._lock, self._conn: + self._conn.execute( + f"UPDATE claims SET {sets} WHERE id = ?", # noqa: S608 — cols whitelisted + (*cols.values(), claim_id), + ) + + def claims_for( + self, + entity_id: int, + attribute: str = "", + status: str = "active", + include_history: bool = False, + ) -> List[Dict[str, Any]]: + q = "SELECT * FROM claims WHERE entity_id = ?" + params: List[Any] = [entity_id] + if attribute: + q += " AND attribute = ?" + params.append(normalize_key(attribute)) + if not include_history: + q += " AND status = ?" + params.append(status) + q += " ORDER BY valid_from ASC, id ASC" + with self._lock: + rows = self._conn.execute(q, params).fetchall() + return [_row_to_dict(r) for r in rows] + + def claims_by_status(self, status: str, limit: int = 100) -> List[Dict[str, Any]]: + with self._lock: + rows = self._conn.execute( + "SELECT * FROM claims WHERE status = ? ORDER BY updated_at DESC LIMIT ?", + (status, limit), + ).fetchall() + return [_row_to_dict(r) for r in rows] + + def all_active_claims(self) -> List[Dict[str, Any]]: + with self._lock: + rows = self._conn.execute( + "SELECT * FROM claims WHERE status = 'active' ORDER BY id" + ).fetchall() + return [_row_to_dict(r) for r in rows] + + # -- evidence ------------------------------------------------------------ + + def add_evidence( + self, + subject_kind: str, + subject_id: int, + kind: str = "session", + ref: str = "", + quote: str = "", + session_id: str = "", + ) -> int: + with self._lock, self._conn: + cur = self._conn.execute( + "INSERT INTO evidence (subject_kind, subject_id, kind, ref, quote, session_id, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (subject_kind, subject_id, kind, ref, quote[:500], session_id, self.now()), + ) + return int(cur.lastrowid or 0) + + def evidence_for(self, subject_kind: str, subject_id: int) -> List[Dict[str, Any]]: + with self._lock: + rows = self._conn.execute( + "SELECT * FROM evidence WHERE subject_kind = ? AND subject_id = ? ORDER BY id", + (subject_kind, subject_id), + ).fetchall() + return [_row_to_dict(r) for r in rows] + + def evidence_count(self, subject_kind: str, subject_id: int) -> int: + with self._lock: + row = self._conn.execute( + "SELECT COUNT(*) AS n FROM evidence WHERE subject_kind = ? AND subject_id = ?", + (subject_kind, subject_id), + ).fetchone() + return int(row["n"]) + + # -- search -------------------------------------------------------------- + + def search(self, query: str, limit: int = 10) -> List[Dict[str, Any]]: + """Keyword search across entities and active claims. + + Word-boundary match on significant tokens (length >= 3, falling back + to all tokens for short queries). Any-token match, ranked by token + coverage; claim hits are additionally weighted by confidence so + trusted knowledge surfaces first. Works both for explicit keyword + queries and for whole-message prefetch queries. + """ + all_tokens = [t for t in normalize_key(query).split() if t] + tokens = [t for t in all_tokens if len(t) >= 3] or all_tokens + if not tokens: + return [] + results: List[Dict[str, Any]] = [] + with self._lock: + ent_rows = self._conn.execute("SELECT * FROM entities").fetchall() + claim_rows = self._conn.execute( + "SELECT c.*, e.name AS entity_name FROM claims c " + "JOIN entities e ON e.id = c.entity_id WHERE c.status = 'active'" + ).fetchall() + for row in ent_rows: + words = set(normalize_key(f"{row['name']} {row['type']} {row['summary']}").split()) + hits = sum(1 for t in tokens if t in words) + if hits: + d = _row_to_dict(row) + d["_kind"] = "entity" + d["_score"] = hits / len(tokens) + results.append(d) + for row in claim_rows: + words = set( + normalize_key(f"{row['entity_name']} {row['attribute']} {row['value']}").split() + ) + hits = sum(1 for t in tokens if t in words) + if hits: + d = _row_to_dict(row) + d["_kind"] = "claim" + d["_score"] = (hits / len(tokens)) * (0.5 + 0.5 * float(row["confidence"])) + results.append(d) + results.sort(key=lambda d: d["_score"], reverse=True) + return results[:limit] + + # -- stats ----------------------------------------------------------------- + + def stats(self) -> Dict[str, Any]: + with self._lock: + n_ent = self._conn.execute("SELECT COUNT(*) AS n FROM entities").fetchone()["n"] + n_rel = self._conn.execute( + "SELECT COUNT(*) AS n FROM relationships WHERE valid_to IS NULL" + ).fetchone()["n"] + by_status = { + r["status"]: r["n"] + for r in self._conn.execute( + "SELECT status, COUNT(*) AS n FROM claims GROUP BY status" + ).fetchall() + } + by_tier = { + r["tier"]: r["n"] + for r in self._conn.execute( + "SELECT tier, COUNT(*) AS n FROM claims WHERE status = 'active' GROUP BY tier" + ).fetchall() + } + n_ev = self._conn.execute("SELECT COUNT(*) AS n FROM evidence").fetchone()["n"] + return { + "entities": n_ent, + "open_relationships": n_rel, + "claims_by_status": by_status, + "active_claims_by_tier": by_tier, + "evidence": n_ev, + "schema_version": int(self.get_meta("schema_version", "1")), + } diff --git a/scripts/release.py b/scripts/release.py index cdebc8e10af5..727253fd2c83 100755 --- a/scripts/release.py +++ b/scripts/release.py @@ -1555,6 +1555,7 @@ "bsmith@bramarstrategicservices.com": "bcsmith528", # PR #20589 salvage (register_slack_action_handler plugin API) "sunsky.lau@gmail.com": "liuhao1024", # PR #45494 salvage (claim session slot before auto-resume task; #45456) "andrewdmwalker@gmail.com": "capt-marbles", # PR #38440 salvage (resolve xAI OAuth credentials across profiles; #43589) + "chadsm@umich.edu": "chadsm-sys", # memorygraph provider (Memory Graph v2) } diff --git a/tests/plugins/memory/test_memorygraph_governance.py b/tests/plugins/memory/test_memorygraph_governance.py new file mode 100644 index 000000000000..cdf737068c78 --- /dev/null +++ b/tests/plugins/memory/test_memorygraph_governance.py @@ -0,0 +1,312 @@ +"""Tests for the memorygraph GovernanceEngine: duplicate detection, +contradiction detection, confidence tracking, aging, and promotion.""" + +from datetime import datetime, timedelta, timezone + +import pytest + +from plugins.memory.memorygraph.governance import ( + GovernanceEngine, + GovernancePolicy, + text_similarity, +) +from plugins.memory.memorygraph.store import GraphStore + + +class FakeClock: + def __init__(self, start="2026-07-01T00:00:00Z"): + self.current = datetime.strptime(start, "%Y-%m-%dT%H:%M:%SZ").replace( + tzinfo=timezone.utc + ) + + def __call__(self): + return self.current.strftime("%Y-%m-%dT%H:%M:%SZ") + + def advance(self, days=0, hours=0): + self.current += timedelta(days=days, hours=hours) + + +@pytest.fixture +def clock(): + return FakeClock() + + +@pytest.fixture +def store(tmp_path, clock): + s = GraphStore(str(tmp_path / "graph.db"), now=clock) + yield s + s.close() + + +@pytest.fixture +def engine(store): + return GovernanceEngine(store, GovernancePolicy()) + + +@pytest.fixture +def entity(store): + return store.upsert_entity("Chad", "person") + + +def test_text_similarity(): + assert text_similarity("works at Facility A", "works at Facility A") == 1.0 + assert text_similarity("Works at Facility A!", "works at facility a") == 1.0 + assert text_similarity("bench press goal 200", "bench press goal 200 lbs") > 0.8 + assert text_similarity("apples", "submarine") < 0.5 + assert text_similarity("", "anything") == 0.0 + + +# -- duplicate detection ------------------------------------------------------ + + +def test_duplicate_reinforces_instead_of_duplicating(engine, store, entity): + r1 = engine.assert_claim(entity["id"], "employer", "Facility A", confidence=0.6) + assert r1["outcome"] == "created" + r2 = engine.assert_claim(entity["id"], "employer", "facility a") + assert r2["outcome"] == "reinforced" + assert r2["claim"]["id"] == r1["claim"]["id"] + assert r2["claim"]["confidence"] == pytest.approx(0.7) + assert r2["claim"]["reinforcement_count"] == 1 + assert len(store.claims_for(entity["id"], "employer")) == 1 + + +def test_near_duplicate_fuzzy_match(engine, entity): + engine.assert_claim(entity["id"], "goal", "bench press 200 lbs by December") + r = engine.assert_claim(entity["id"], "goal", "bench press 200lbs by December!") + assert r["outcome"] == "reinforced" + + +def test_distinct_values_do_not_dedupe(engine, store, entity): + engine.assert_claim(entity["id"], "skill", "regional anesthesia") + r = engine.assert_claim(entity["id"], "skill", "pediatric airway management") + assert r["outcome"] == "created" + assert len(store.claims_for(entity["id"], "skill")) == 2 + + +def test_duplicate_evidence_accumulates(engine, store, entity): + r1 = engine.assert_claim( + entity["id"], "employer", "Facility A", + evidence={"kind": "session", "ref": "s1"}, + ) + engine.assert_claim( + entity["id"], "employer", "Facility A", + evidence={"kind": "session", "ref": "s2"}, + ) + assert store.evidence_count("claim", r1["claim"]["id"]) == 2 + + +def test_find_duplicates_audit(engine, store, entity): + # Insert raw (ungoverned) near-duplicates, then audit. + store.insert_claim(entity["id"], "note", "drinks black coffee every morning") + store.insert_claim(entity["id"], "note", "drinks black coffee every morning!!") + store.insert_claim(entity["id"], "note", "allergic to shellfish") + groups = engine.find_duplicates() + assert len(groups) == 1 + assert len(groups[0]) == 2 + + +# -- contradiction detection ------------------------------------------------------ + + +def test_exclusive_newer_value_supersedes(engine, store, entity, clock): + r1 = engine.assert_claim(entity["id"], "employer", "Facility A", exclusive=True) + clock.advance(days=10) + r2 = engine.assert_claim(entity["id"], "employer", "Facility B", exclusive=True) + assert r2["outcome"] == "superseded" + old = store.get_claim(r1["claim"]["id"]) + assert old["status"] == "superseded" + assert old["superseded_by"] == r2["claim"]["id"] + assert old["valid_to"] == r2["claim"]["valid_from"] + active = store.claims_for(entity["id"], "employer") + assert [c["value"] for c in active] == ["Facility B"] + + +def test_ambiguous_conflict_flags_contradiction(engine, store, entity, clock): + engine.assert_claim(entity["id"], "employer", "Facility A", exclusive=True) + clock.advance(days=10) + engine.assert_claim(entity["id"], "employer", "Facility B", exclusive=True) + # Backdated claim: older than the current active one → ambiguous. + r3 = engine.assert_claim( + entity["id"], "employer", "Facility C", exclusive=True, + valid_from="2026-07-05T00:00:00Z", + ) + assert r3["outcome"] == "contradicted" + assert store.get_claim(r3["claim"]["id"])["status"] == "contradicted" + groups = engine.find_contradictions() + assert len(groups) == 1 + assert len(groups[0]) == 2 + + +def test_contradiction_lowers_confidence(engine, store, entity): + r1 = engine.assert_claim( + entity["id"], "employer", "Facility A", exclusive=True, confidence=0.8, + valid_from="2026-07-02T00:00:00Z", + ) + r2 = engine.assert_claim( + entity["id"], "employer", "Facility B", exclusive=True, confidence=0.8, + valid_from="2026-07-01T00:00:00Z", # backdated → ambiguous + ) + assert r2["outcome"] == "contradicted" + assert store.get_claim(r1["claim"]["id"])["confidence"] == pytest.approx(0.65) + assert store.get_claim(r2["claim"]["id"])["confidence"] == pytest.approx(0.65) + + +def test_resolve_contradiction(engine, store, entity): + engine.assert_claim( + entity["id"], "employer", "Facility A", exclusive=True, + valid_from="2026-07-02T00:00:00Z", + ) + r2 = engine.assert_claim( + entity["id"], "employer", "Facility B", exclusive=True, + valid_from="2026-07-01T00:00:00Z", + ) + result = engine.resolve_contradiction(r2["claim"]["id"]) + assert result["winner"]["status"] == "active" + assert len(result["superseded"]) == 1 + assert engine.find_contradictions() == [] + + +def test_non_exclusive_claims_coexist(engine, store, entity): + engine.assert_claim(entity["id"], "hobby", "hockey with Colin") + r = engine.assert_claim(entity["id"], "hobby", "golf") + assert r["outcome"] == "created" + assert len(store.claims_for(entity["id"], "hobby")) == 2 + + +# -- confidence tracking --------------------------------------------------------- + + +def test_confidence_clamped_at_one(engine, entity): + r = engine.assert_claim(entity["id"], "fact", "stable truth", confidence=0.99) + for _ in range(5): + r = {"claim": engine.reinforce(r["claim"]["id"])} + assert r["claim"]["confidence"] == 1.0 + + +def test_feedback_adjusts_confidence(engine, entity): + r = engine.assert_claim(entity["id"], "fact", "something", confidence=0.5) + up = engine.feedback(r["claim"]["id"], helpful=True) + assert up["confidence"] == pytest.approx(0.65) + assert up["reinforcement_count"] == 1 + down = engine.feedback(r["claim"]["id"], helpful=False) + assert down["confidence"] == pytest.approx(0.5) + assert engine.feedback(9999, helpful=True) is None + + +def test_retract(engine, store, entity): + r = engine.assert_claim(entity["id"], "fact", "wrong thing") + assert engine.retract(r["claim"]["id"], "user asked") is True + assert engine.retract(r["claim"]["id"]) is False # already retracted + assert store.get_claim(r["claim"]["id"])["status"] == "retracted" + assert store.claims_for(entity["id"], "fact") == [] + + +# -- knowledge aging ----------------------------------------------------------------- + + +def test_aging_decays_confidence(engine, store, entity, clock): + r = engine.assert_claim(entity["id"], "fact", "ages over time", confidence=0.8) + clock.advance(days=90) # exactly one half-life + result = engine.age_knowledge() + assert result["decayed"] == 1 + aged = store.get_claim(r["claim"]["id"]) + assert aged["confidence"] == pytest.approx(0.4, abs=0.01) + + +def test_aging_has_confidence_floor(engine, store, entity, clock): + r = engine.assert_claim(entity["id"], "fact", "ancient", confidence=0.8) + clock.advance(days=3650) + engine.age_knowledge() + assert store.get_claim(r["claim"]["id"])["confidence"] == pytest.approx(0.15) + + +def test_aging_skips_fresh_claims(engine, entity, clock): + engine.assert_claim(entity["id"], "fact", "brand new") + assert engine.age_knowledge()["decayed"] == 0 + + +def test_reinforcement_resets_aging(engine, store, entity, clock): + r = engine.assert_claim(entity["id"], "fact", "kept alive", confidence=0.8) + clock.advance(days=89) + engine.reinforce(r["claim"]["id"]) # refreshes last_reinforced_at + engine.age_knowledge() + # No decay: last reinforcement is "now". + assert store.get_claim(r["claim"]["id"])["confidence"] == pytest.approx(0.9) + + +def test_aging_demotes_decayed_established(engine, store, entity, clock): + r = engine.assert_claim(entity["id"], "fact", "was solid", confidence=0.8) + store.update_claim(r["claim"]["id"], tier="established") + clock.advance(days=180) # two half-lives → 0.2 < demotion threshold 0.4 + result = engine.age_knowledge() + assert result["demoted"] == 1 + assert store.get_claim(r["claim"]["id"])["tier"] == "candidate" + + +# -- knowledge promotion ---------------------------------------------------------------- + + +def test_promotion_candidate_to_established(engine, store, entity): + r = engine.assert_claim( + entity["id"], "fact", "well evidenced", confidence=0.6, + evidence={"kind": "session", "ref": "s1"}, + ) + claim_id = r["claim"]["id"] + engine.assert_claim( # duplicate → reinforce + second evidence link + entity["id"], "fact", "well evidenced", + evidence={"kind": "url", "ref": "https://example.com"}, + ) + result = engine.promote_knowledge() + assert result["established"] == 1 + assert store.get_claim(claim_id)["tier"] == "established" + + +def test_promotion_requires_evidence(engine, store, entity): + r = engine.assert_claim(entity["id"], "fact", "confident but unevidenced", + confidence=0.95) + engine.promote_knowledge() + assert store.get_claim(r["claim"]["id"])["tier"] == "candidate" + + +def test_promotion_established_to_core_requires_age_and_reinforcement( + engine, store, entity, clock +): + r = engine.assert_claim( + entity["id"], "fact", "core truth", confidence=0.6, + evidence={"kind": "session", "ref": "s1"}, + ) + claim_id = r["claim"]["id"] + store.add_evidence("claim", claim_id, kind="session", ref="s2") + for _ in range(3): + engine.reinforce(claim_id) # 0.6 → 0.9, count 3 + assert engine.promote_knowledge()["established"] == 1 + + # Not old enough for core yet. + assert engine.promote_knowledge()["core"] == 0 + clock.advance(days=8) + # Aging over 8 days pulls 0.9 down a bit but stays above 0.85. + engine.reinforce(claim_id) # keep it fresh: 1.0, count 4 + assert engine.promote_knowledge()["core"] == 1 + assert store.get_claim(claim_id)["tier"] == "core" + + +def test_contradicted_claims_never_promote(engine, store, entity): + engine.assert_claim( + entity["id"], "employer", "Facility A", exclusive=True, confidence=0.9, + valid_from="2026-07-02T00:00:00Z", + ) + engine.assert_claim( + entity["id"], "employer", "Facility B", exclusive=True, confidence=0.9, + valid_from="2026-07-01T00:00:00Z", + ) + result = engine.promote_knowledge() + assert result["established"] == 0 and result["core"] == 0 + + +def test_sweep_runs_aging_then_promotion(engine, store, entity, clock): + engine.assert_claim(entity["id"], "fact", "sweep me", confidence=0.8) + clock.advance(days=90) + result = engine.sweep() + assert result["aging"]["decayed"] == 1 + assert "promotion" in result + assert store.get_meta("last_aging_run") == clock() diff --git a/tests/plugins/memory/test_memorygraph_provider.py b/tests/plugins/memory/test_memorygraph_provider.py new file mode 100644 index 000000000000..cafc9c496db8 --- /dev/null +++ b/tests/plugins/memory/test_memorygraph_provider.py @@ -0,0 +1,401 @@ +"""Tests for the MemoryGraphProvider: lifecycle, tool surface, built-in +memory mirroring, prefetch, config, and loader discovery.""" + +import json +import os + +import pytest + +from plugins.memory.memorygraph import ( + GRAPH_MEMORY_SCHEMA, + MemoryGraphProvider, + register, +) + + +@pytest.fixture +def provider(tmp_path): + p = MemoryGraphProvider() + p.initialize("session-1", hermes_home=str(tmp_path), platform="cli", + agent_context="primary") + yield p + p.shutdown() + + +def call(provider, **args): + return json.loads(provider.handle_tool_call("graph_memory", args)) + + +# -- lifecycle ----------------------------------------------------------------- + + +def test_availability_without_initialize(): + p = MemoryGraphProvider() + assert p.name == "memorygraph" + assert p.is_available() is True # local-only: no creds needed + + +def test_constructor_has_no_side_effects(tmp_path): + # discover_memory_providers() instantiates providers just to list them — + # the constructor must not create the database or config files. + MemoryGraphProvider() + assert list(tmp_path.rglob("memory_graph.db")) == [] + assert list(tmp_path.rglob("memorygraph.json")) == [] + + +def test_initialize_creates_profile_scoped_db(provider, tmp_path): + assert os.path.exists(tmp_path / "memory_graph.db") + + +def test_tool_call_before_initialize_is_safe(): + p = MemoryGraphProvider() + result = json.loads(p.handle_tool_call("graph_memory", {"action": "stats"})) + assert "error" in result + + +def test_shutdown_then_tool_call_is_safe(tmp_path): + p = MemoryGraphProvider() + p.initialize("s", hermes_home=str(tmp_path)) + p.shutdown() + assert "error" in json.loads(p.handle_tool_call("graph_memory", {"action": "stats"})) + + +def test_session_switch_clears_prefetch_cache(provider): + provider._prefetch_cache["q"] = "stale" + provider.on_session_switch("session-2", reset=True) + assert provider._prefetch_cache == {} + assert provider._session_id == "session-2" + + +# -- tool schema ------------------------------------------------------------------ + + +def test_tool_schema_shape(provider): + schemas = provider.get_tool_schemas() + assert len(schemas) == 1 + schema = schemas[0] + assert schema["name"] == "graph_memory" + assert schema is GRAPH_MEMORY_SCHEMA + actions = schema["parameters"]["properties"]["action"]["enum"] + for action in ("remember", "link", "about", "query", "timeline", + "contradictions", "resolve", "duplicates", "feedback", + "forget", "sweep", "stats"): + assert action in actions + assert schema["parameters"]["required"] == ["action"] + + +def test_unknown_tool_and_action(provider): + assert "error" in json.loads(provider.handle_tool_call("nope", {})) + assert "error" in call(provider, action="explode") + assert "error" in call(provider) # missing action + + +def test_missing_required_args_reported(provider): + result = call(provider, action="remember", entity="Chad") + assert "required" in result["error"] + + +# -- core actions round trip --------------------------------------------------------- + + +def test_remember_and_about(provider): + result = call( + provider, action="remember", entity="Hermes", entity_type="project", + attribute="status", value="P1 reliability push", confidence=0.8, + source="backlog.md", quote="P1 Hermes reliability", + ) + assert result["outcome"] == "created" + assert result["entity"]["type"] == "project" + + about = call(provider, action="about", entity="hermes") + assert about["found"] is True + assert about["claims"][0]["value"] == "P1 reliability push" + assert about["claims"][0]["evidence_count"] == 1 + + missing = call(provider, action="about", entity="Unknown Entity") + assert missing["found"] is False + + +def test_remember_duplicate_reinforces(provider): + call(provider, action="remember", entity="Jamie", entity_type="person", + attribute="birthday", value="March 30") + result = call(provider, action="remember", entity="Jamie", + entity_type="person", attribute="birthday", value="march 30") + assert result["outcome"] == "reinforced" + + +def test_remember_exclusive_supersedes_and_timeline(provider): + call(provider, action="remember", entity="Chad", entity_type="person", + attribute="employer", value="Facility A", exclusive=True) + result = call(provider, action="remember", entity="Chad", + entity_type="person", attribute="employer", + value="Facility B", exclusive=True) + assert result["outcome"] == "superseded" + + timeline = call(provider, action="timeline", entity="Chad", + attribute="employer") + assert timeline["found"] is True + statuses = [c["status"] for c in timeline["timeline"]] + assert statuses == ["superseded", "active"] + + +def test_link_unlink_and_relationships_in_about(provider): + result = call(provider, action="link", src="Chad", dst="C Smith Anesthesia", + rel_type="owns", entity_type="person") + rel_id = result["relationship"]["id"] + about = call(provider, action="about", entity="Chad") + assert len(about["relationships"]) == 1 + assert about["relationships"][0]["rel_type"] == "owns" + + ended = call(provider, action="unlink", relationship_id=rel_id) + assert ended["ended"] is True + about = call(provider, action="about", entity="Chad") + assert about["relationships"] == [] + + +def test_query(provider): + call(provider, action="remember", entity="Gusto", entity_type="tool", + attribute="purpose", value="S-Corp W-2 payroll") + result = call(provider, action="query", query="payroll") + assert result["results"] + assert result["results"][0]["_kind"] == "claim" + + +def test_contradiction_flow(provider): + call(provider, action="remember", entity="Chad", entity_type="person", + attribute="employer", value="Facility A", exclusive=True) + # handle_tool_call has no valid_from arg → backdating happens via the + # engine; simulate ambiguity by writing directly through the engine. + entity = provider._store.resolve_entity("Chad") + provider._engine.assert_claim( + entity["id"], "employer", "Facility B", exclusive=True, + valid_from="2020-01-01T00:00:00Z", + ) + groups = call(provider, action="contradictions")["groups"] + assert len(groups) == 1 + winner_id = groups[0][0]["id"] + resolved = call(provider, action="resolve", claim_id=winner_id) + assert resolved["winner"]["status"] == "active" + assert call(provider, action="contradictions")["groups"] == [] + + +def test_feedback_and_forget(provider): + result = call(provider, action="remember", entity="X", attribute="fact", + value="something", confidence=0.5) + claim_id = result["claim"]["id"] + fb = call(provider, action="feedback", claim_id=claim_id, helpful=True) + assert fb["claim"]["confidence"] == pytest.approx(0.65) + assert "error" in call(provider, action="feedback", claim_id=claim_id) + + gone = call(provider, action="forget", claim_id=claim_id) + assert gone["retracted"] is True + about = call(provider, action="about", entity="X") + assert about["claims"] == [] + + +def test_duplicates_audit_action(provider): + entity = provider._store.upsert_entity("X") + provider._store.insert_claim(entity["id"], "note", "same fact here") + provider._store.insert_claim(entity["id"], "note", "same fact here!") + assert len(call(provider, action="duplicates")["groups"]) == 1 + + +def test_sweep_and_stats(provider): + call(provider, action="remember", entity="Hermes", entity_type="project", + attribute="status", value="P1") + result = call(provider, action="sweep") + assert "aging" in result and "promotion" in result + + stats = call(provider, action="stats") + assert stats["entities"] == 1 + assert stats["claims_by_status"]["active"] == 1 + assert any(e["event"] == "claim_created" + for e in stats["recent_governance_events"]) + + +# -- prompt / prefetch --------------------------------------------------------------- + + +def test_system_prompt_block(provider): + assert provider.system_prompt_block().startswith("## Knowledge Graph Memory") + call(provider, action="remember", entity="Hermes", entity_type="project", + attribute="status", value="P1") + block = provider.system_prompt_block() + assert "1 entities" in block + assert "graph_memory" in block + + +def test_system_prompt_block_uninitialized(): + assert MemoryGraphProvider().system_prompt_block() == "" + + +def test_prefetch_returns_recall(provider): + call(provider, action="remember", entity="Gusto", entity_type="tool", + attribute="purpose", value="payroll for the S-Corp") + text = provider.prefetch("how do I run payroll?") + assert "[memorygraph recall]" in text + assert "payroll" in text + assert provider.prefetch("completely unrelated zebra") == "" + assert provider.prefetch("") == "" + + +def test_prefetch_disabled_by_config(tmp_path): + with open(tmp_path / "memorygraph.json", "w", encoding="utf-8") as f: + json.dump({"prefetch_enabled": False}, f) + p = MemoryGraphProvider() + p.initialize("s", hermes_home=str(tmp_path)) + try: + p.handle_tool_call("graph_memory", { + "action": "remember", "entity": "Gusto", "attribute": "purpose", + "value": "payroll", + }) + assert p.prefetch("payroll") == "" + finally: + p.shutdown() + + +# -- built-in memory mirroring ---------------------------------------------------------- + + +def test_on_memory_write_mirrors_to_graph(provider): + provider.on_memory_write("add", "memory", "Chad prefers 4:30 AM training", + metadata={"session_id": "s-77"}) + about = call(provider, action="about", entity="Hermes Notes") + assert about["found"] is True + assert about["claims"][0]["value"] == "Chad prefers 4:30 AM training" + evidence = provider._store.evidence_for("claim", about["claims"][0]["id"]) + assert evidence[0]["kind"] == "builtin_memory" + assert evidence[0]["session_id"] == "s-77" + + +def test_on_memory_write_user_target(provider): + provider.on_memory_write("add", "user", "Married to Jamie") + about = call(provider, action="about", entity="User") + assert about["found"] is True + assert about["entity"]["type"] == "person" + + +def test_on_memory_write_ignores_removes_and_blanks(provider): + provider.on_memory_write("remove", "memory", "whatever") + provider.on_memory_write("add", "memory", " ") + assert call(provider, action="stats")["entities"] == 0 + + +def test_on_memory_write_before_initialize_is_safe(): + MemoryGraphProvider().on_memory_write("add", "memory", "content") # no raise + + +def test_on_session_end_runs_sweep(provider): + call(provider, action="remember", entity="X", attribute="fact", value="v") + provider.on_session_end([]) # must not raise + events = [e["event"] for e in provider._store.recent_events(50)] + assert "claim_created" in events + + +def test_on_session_end_skipped_for_non_primary(tmp_path): + p = MemoryGraphProvider() + p.initialize("s", hermes_home=str(tmp_path), agent_context="cron") + try: + p._store.set_meta("last_aging_run", "sentinel") + p.on_session_end([]) + assert p._store.get_meta("last_aging_run") == "sentinel" + finally: + p.shutdown() + + +# -- config ------------------------------------------------------------------------------ + + +def test_config_schema_and_save(tmp_path): + p = MemoryGraphProvider() + keys = [f["key"] for f in p.get_config_schema()] + assert "half_life_days" in keys + p.save_config({"half_life_days": "30", "duplicate_similarity": "bad"}, + str(tmp_path)) + with open(tmp_path / "memorygraph.json", encoding="utf-8") as f: + saved = json.load(f) + assert saved == {"half_life_days": 30.0} + + p.initialize("s", hermes_home=str(tmp_path)) + try: + assert p._engine.policy.half_life_days == 30.0 + finally: + p.shutdown() + + +def test_custom_db_path(tmp_path): + custom = tmp_path / "custom" / "kg.db" + with open(tmp_path / "memorygraph.json", "w", encoding="utf-8") as f: + json.dump({"db_path": str(custom)}, f) + p = MemoryGraphProvider() + p.initialize("s", hermes_home=str(tmp_path)) + try: + assert custom.exists() + finally: + p.shutdown() + + +def test_corrupt_config_falls_back_to_defaults(tmp_path): + with open(tmp_path / "memorygraph.json", "w", encoding="utf-8") as f: + f.write("{not json") + p = MemoryGraphProvider() + p.initialize("s", hermes_home=str(tmp_path)) + try: + assert p._engine.policy.half_life_days == 90.0 + finally: + p.shutdown() + + +# -- registration / discovery ---------------------------------------------------------- + + +def test_register_entry_point(): + class Ctx: + provider = None + + def register_memory_provider(self, p): + self.provider = p + + ctx = Ctx() + register(ctx) + assert isinstance(ctx.provider, MemoryGraphProvider) + + +def test_loader_discovers_memorygraph(): + from plugins.memory import find_provider_dir, load_memory_provider + + assert find_provider_dir("memorygraph") is not None + provider = load_memory_provider("memorygraph") + assert provider is not None + assert provider.name == "memorygraph" + + +def test_bool_args_coerce_string_forms(provider): + # bool("false") is True — tool args may arrive as strings. + result = call(provider, action="remember", entity="Chad", + entity_type="person", attribute="employer", + value="Facility A", exclusive="false") + assert result["claim"]["exclusive"] == 0 + + result = call(provider, action="remember", entity="Chad", + entity_type="person", attribute="status", + value="active CRNA", exclusive="true") + assert result["claim"]["exclusive"] == 1 + + r = call(provider, action="remember", entity="X", attribute="fact", + value="v", confidence=0.5) + fb = call(provider, action="feedback", claim_id=r["claim"]["id"], + helpful="false") # string "false" must mean unhelpful + assert fb["claim"]["confidence"] == pytest.approx(0.35) + + +def test_config_clamps_invalid_ranges(tmp_path): + with open(tmp_path / "memorygraph.json", "w", encoding="utf-8") as f: + json.dump({"half_life_days": -5, "duplicate_similarity": 3.0}, f) + p = MemoryGraphProvider() + p.initialize("s", hermes_home=str(tmp_path)) + try: + assert p._engine.policy.half_life_days == 0.1 + assert p._engine.policy.duplicate_similarity == 1.0 + finally: + p.shutdown() diff --git a/tests/plugins/memory/test_memorygraph_store.py b/tests/plugins/memory/test_memorygraph_store.py new file mode 100644 index 000000000000..8471ff7dcd17 --- /dev/null +++ b/tests/plugins/memory/test_memorygraph_store.py @@ -0,0 +1,246 @@ +"""Tests for the memorygraph GraphStore (entities, relationships, claims, +evidence, time-aware windows, search, audit log).""" + +from datetime import datetime, timedelta, timezone + +import pytest + +from plugins.memory.memorygraph.store import ( + ENTITY_TYPES, + GraphStore, + normalize_key, + parse_ts, +) + + +class FakeClock: + """Deterministic UTC clock the store accepts as its ``now`` callable.""" + + def __init__(self, start="2026-07-01T00:00:00Z"): + self.current = datetime.strptime(start, "%Y-%m-%dT%H:%M:%SZ").replace( + tzinfo=timezone.utc + ) + + def __call__(self): + return self.current.strftime("%Y-%m-%dT%H:%M:%SZ") + + def advance(self, days=0, hours=0, seconds=0): + self.current += timedelta(days=days, hours=hours, seconds=seconds) + + +@pytest.fixture +def clock(): + return FakeClock() + + +@pytest.fixture +def store(tmp_path, clock): + s = GraphStore(str(tmp_path / "graph.db"), now=clock) + yield s + s.close() + + +def test_normalize_key(): + assert normalize_key(" C Smith Anesthesia, LLC! ") == "c smith anesthesia llc" + assert normalize_key("") == "" + assert normalize_key("ABC-123") == "abc 123" + + +def test_parse_ts_roundtrip(clock): + ts = clock() + assert parse_ts(ts).strftime("%Y-%m-%dT%H:%M:%SZ") == ts + + +# -- entities --------------------------------------------------------------- + + +def test_entity_types_cover_required_domains(): + for required in ("person", "project", "goal", "skill", "business"): + assert required in ENTITY_TYPES + + +def test_upsert_entity_creates_and_resolves(store): + e = store.upsert_entity("Jamie Smith", "person", summary="spouse") + assert e["id"] > 0 + assert e["type"] == "person" + resolved = store.resolve_entity("jamie smith") + assert resolved is not None and resolved["id"] == e["id"] + + +def test_upsert_entity_is_idempotent_and_merges(store): + a = store.upsert_entity("Hermes", "project", summary="gateway") + b = store.upsert_entity("Hermes", "project", attrs={"priority": "P1"}) + assert a["id"] == b["id"] + assert b["summary"] == "gateway" # preserved + assert '"priority"' in b["attrs"] + + +def test_upsert_entity_rejects_empty_name(store): + with pytest.raises(ValueError): + store.upsert_entity(" ") + + +def test_unknown_entity_type_falls_back_to_concept(store): + e = store.upsert_entity("Thing", "starship") + assert e["type"] == "concept" + + +def test_alias_resolution(store): + e = store.upsert_entity("C Smith Anesthesia Staffing LLC", "business") + assert store.add_alias(e["id"], "CSA") is True + assert store.add_alias(e["id"], "CSA") is False # duplicate alias + resolved = store.resolve_entity("csa") + assert resolved is not None and resolved["id"] == e["id"] + + +def test_list_entities_filters_by_type(store): + store.upsert_entity("Colin", "person") + store.upsert_entity("Bench 200", "goal") + people = store.list_entities("person") + assert [e["name"] for e in people] == ["Colin"] + + +# -- relationships ------------------------------------------------------------ + + +def test_relationship_lifecycle_time_aware(store, clock): + a = store.upsert_entity("Chad", "person") + b = store.upsert_entity("Facility A", "business") + rel = store.add_relationship(a["id"], b["id"], "works_at", confidence=0.7) + assert rel["valid_to"] is None + assert rel["valid_from"] == clock() + + # Re-adding the same open edge reinforces, not duplicates + again = store.add_relationship(a["id"], b["id"], "works_at") + assert again["id"] == rel["id"] + assert again["confidence"] > rel["confidence"] + + clock.advance(days=30) + assert store.end_relationship(rel["id"]) is True + assert store.end_relationship(rel["id"]) is False # already ended + ended = store.relationships_for(a["id"], include_ended=True)[0] + assert ended["valid_to"] == clock() + assert store.relationships_for(a["id"]) == [] # open-only view + + +def test_neighbors(store): + a = store.upsert_entity("Chad", "person") + b = store.upsert_entity("Hermes", "project") + c = store.upsert_entity("Emma", "person") + store.add_relationship(a["id"], b["id"], "owns") + store.add_relationship(c["id"], a["id"], "child_of") + names = {e["name"] for e in store.neighbors(a["id"])} + assert names == {"Hermes", "Emma"} + + +# -- claims + evidence ----------------------------------------------------------- + + +def test_claim_insert_defaults(store, clock): + e = store.upsert_entity("Hermes", "project") + c = store.insert_claim(e["id"], "Status", "P1 active", confidence=0.9) + assert c["attribute"] == "status" # normalized + assert c["tier"] == "candidate" + assert c["status"] == "active" + assert c["valid_from"] == clock() + assert c["confidence"] == 0.9 + + +def test_claim_confidence_clamped(store): + e = store.upsert_entity("X") + c = store.insert_claim(e["id"], "a", "v", confidence=7.5) + assert c["confidence"] == 1.0 + + +def test_claims_for_history_and_status_filters(store): + e = store.upsert_entity("X") + c1 = store.insert_claim(e["id"], "a", "old") + store.insert_claim(e["id"], "a", "new") + store.update_claim(c1["id"], status="superseded") + active = store.claims_for(e["id"], "a") + assert [c["value"] for c in active] == ["new"] + history = store.claims_for(e["id"], "a", include_history=True) + assert len(history) == 2 + + +def test_update_claim_ignores_unknown_fields(store): + e = store.upsert_entity("X") + c = store.insert_claim(e["id"], "a", "v") + store.update_claim(c["id"], entity_id=999, bogus="nope", confidence=0.25) + updated = store.get_claim(c["id"]) + assert updated["entity_id"] == e["id"] + assert updated["confidence"] == 0.25 + + +def test_evidence_links(store): + e = store.upsert_entity("Hermes", "project") + c = store.insert_claim(e["id"], "status", "P1") + store.add_evidence("claim", c["id"], kind="session", ref="s-1", quote="it is P1") + store.add_evidence("claim", c["id"], kind="url", ref="https://example.com") + assert store.evidence_count("claim", c["id"]) == 2 + kinds = [ev["kind"] for ev in store.evidence_for("claim", c["id"])] + assert kinds == ["session", "url"] + + +# -- search / stats / audit ------------------------------------------------------- + + +def test_search_matches_entities_and_claims(store): + e = store.upsert_entity("Gusto Payroll", "tool", summary="W-2 payroll") + store.insert_claim(e["id"], "saves", "57K per year via S-Corp", confidence=0.8) + hits = store.search("gusto payroll") + assert hits and hits[0]["_kind"] == "entity" + hits = store.search("s corp 57k") + assert hits and hits[0]["_kind"] == "claim" + assert store.search("") == [] + assert store.search("nonexistent zebra") == [] + + +def test_search_excludes_retracted_claims(store): + e = store.upsert_entity("X") + c = store.insert_claim(e["id"], "a", "unique zebra fact") + store.update_claim(c["id"], status="retracted") + assert store.search("zebra") == [] + + +def test_stats_and_governance_log(store): + e = store.upsert_entity("Hermes", "project") + store.insert_claim(e["id"], "status", "P1") + stats = store.stats() + assert stats["entities"] == 1 + assert stats["claims_by_status"]["active"] == 1 + assert stats["active_claims_by_tier"]["candidate"] == 1 + assert stats["schema_version"] == 1 + events = [ev["event"] for ev in store.recent_events()] + assert "entity_created" in events + assert "claim_created" in events + + +def test_meta_roundtrip(store): + store.set_meta("k", "v1") + store.set_meta("k", "v2") + assert store.get_meta("k") == "v2" + assert store.get_meta("missing", "default") == "default" + + +def test_persistence_across_reopen(tmp_path, clock): + path = str(tmp_path / "graph.db") + s1 = GraphStore(path, now=clock) + s1.upsert_entity("Persist Me", "concept") + s1.close() + s2 = GraphStore(path, now=clock) + assert s2.resolve_entity("persist me") is not None + s2.close() + + +def test_normalize_ts_validates_format(store, clock): + from plugins.memory.memorygraph.store import normalize_ts + + assert normalize_ts("2026-07-01T00:00:00Z", "FB") == "2026-07-01T00:00:00Z" + assert normalize_ts("", "FB") == "FB" + assert normalize_ts("2026-07-01 00:00:00", "FB") == "FB" + assert normalize_ts("2026-07-01T00:00:00.123Z", "FB") == "FB" + # Write-boundary enforcement: bad valid_from falls back to now() + e = store.upsert_entity("X") + c = store.insert_claim(e["id"], "a", "v", valid_from="not-a-timestamp") + assert c["valid_from"] == clock() diff --git a/website/docs/user-guide/features/memory-providers.md b/website/docs/user-guide/features/memory-providers.md index 476bd46696dd..82ed96f995a9 100644 --- a/website/docs/user-guide/features/memory-providers.md +++ b/website/docs/user-guide/features/memory-providers.md @@ -1,12 +1,12 @@ --- sidebar_position: 4 title: "Memory Providers" -description: "External memory provider plugins — Honcho, OpenViking, Mem0, Hindsight, Holographic, RetainDB, ByteRover, Supermemory" +description: "External memory provider plugins — Honcho, OpenViking, Mem0, Hindsight, Holographic, Memory Graph, RetainDB, ByteRover, Supermemory" --- # Memory Providers -Hermes Agent ships with 8 external memory provider plugins that give the agent persistent, cross-session knowledge beyond the built-in MEMORY.md and USER.md. Only **one** external provider can be active at a time — the built-in memory is always active alongside it. +Hermes Agent ships with external memory provider plugins that give the agent persistent, cross-session knowledge beyond the built-in MEMORY.md and USER.md. Only **one** external provider can be active at a time — the built-in memory is always active alongside it. ## Quick Start @@ -22,7 +22,7 @@ Or set manually in `~/.hermes/config.yaml`: ```yaml memory: - provider: openviking # or honcho, mem0, hindsight, holographic, retaindb, byterover, supermemory + provider: openviking # or honcho, mem0, hindsight, holographic, memorygraph, retaindb, byterover, supermemory ``` ## How It Works @@ -421,6 +421,45 @@ hermes config set memory.provider holographic --- +### Memory Graph (memorygraph) + +Governed knowledge graph: typed entities (people, projects, goals, skills, businesses, ...), time-aware relationships, and evidence-linked claims with confidence tracking, contradiction/duplicate detection, knowledge aging, and candidate → established → core promotion. + +| | | +|---|---| +| **Best for** | Structured, auditable knowledge with provenance and lifecycle governance, fully local | +| **Requires** | Nothing (stdlib SQLite only) | +| **Data storage** | Local SQLite (`$HERMES_HOME/memory_graph.db`) | +| **Cost** | Free | + +**Tools:** `graph_memory` (13 actions: remember, link, unlink, about, query, timeline, contradictions, resolve, duplicates, feedback, forget, sweep, stats) + +**Setup:** +```bash +hermes memory setup # select "memorygraph" +# Or manually: +hermes config set memory.provider memorygraph +``` + +**Config:** `$HERMES_HOME/memorygraph.json` (optional) + +| Key | Default | Description | +|-----|---------|-------------| +| `db_path` | `$HERMES_HOME/memory_graph.db` | SQLite database path | +| `half_life_days` | `90` | Knowledge aging half-life (confidence decay) | +| `duplicate_similarity` | `0.88` | Near-duplicate detection threshold (0–1) | +| `prefetch_enabled` | `true` | Inject graph recall before each turn | +| `sweep_on_session_end` | `true` | Run aging + promotion at session end | + +**Unique capabilities:** +- Time-aware knowledge — claims and relationships carry validity windows; superseded values stay queryable via `timeline` +- Write-time governance — duplicates reinforce instead of duplicating; conflicting exclusive values are superseded or flagged as contradictions for review +- Promotion tiers — knowledge graduates candidate → established → core based on confidence, evidence links, reinforcement, and age +- Evidence links + append-only governance audit log +- Mirrors built-in MEMORY.md/USER.md writes into the graph automatically + +--- + ### RetainDB Cloud memory API with hybrid search (Vector + BM25 + Reranking), 7 memory types, and delta compression. @@ -567,6 +606,7 @@ hermes memory setup | **Mem0** | Cloud | Paid | 3 | `mem0ai` | Server-side LLM extraction | | **Hindsight** | Cloud/Local | Free/Paid | 3 | `hindsight-client` | Knowledge graph + reflect synthesis | | **Holographic** | Local | Free | 2 | None | HRR algebra + trust scoring | +| **Memory Graph** | Local | Free | 1 | None | Governed knowledge graph: evidence, contradiction detection, aging, promotion | | **RetainDB** | Cloud | $20/mo | 5 | `requests` | Delta compression | | **ByteRover** | Local/Cloud | Free/Paid | 3 | `brv` CLI | Pre-compression extraction | | **Supermemory** | Cloud | Paid | 4 | `supermemory` | Context fencing + session graph ingest + multi-container | @@ -576,7 +616,7 @@ hermes memory setup Each provider's data is isolated per [profile](/user-guide/profiles): -- **Local storage providers** (Holographic, ByteRover) use `$HERMES_HOME/` paths which differ per profile +- **Local storage providers** (Holographic, Memory Graph, ByteRover) use `$HERMES_HOME/` paths which differ per profile - **Config file providers** (Honcho, Mem0, Hindsight, Supermemory) store config in `$HERMES_HOME/` so each profile has its own credentials - **Cloud providers** (RetainDB) auto-derive profile-scoped project names - **Env var providers** (OpenViking) are configured via each profile's `.env` file