From 194990d424c0ba34d6fbc3e21882d33bc9392026 Mon Sep 17 00:00:00 2001 From: Cal Corum Date: Sat, 28 Feb 2026 22:07:05 -0600 Subject: [PATCH] feat: auto-create edges on memory_store in MCP server Automatically find and link related memories after every store operation, removing reliance on Claude manually creating edges. Uses recall + type heuristics to choose edge types (SOLVES, BUILDS_ON, RELATED_TO) and returns created edge IDs in the store response for optional review. Key design choices: - Keyword-only fallback requires tag overlap to prevent spurious edges - Similarity threshold (0.4) filters before max-edge cap (3) - Edge description arrow matches actual from/to direction - Failures never break the store operation Co-Authored-By: Claude Opus 4.6 --- mcp_server.py | 110 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 110 insertions(+) diff --git a/mcp_server.py b/mcp_server.py index d7679a8..ec1d757 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -20,6 +20,23 @@ from common import resolve_graph_path, list_graphs SYNC_SCRIPT = Path(__file__).parent / "scripts" / "memory-git-sync.sh" +# Auto-edge heuristics: (new_type, match_type) -> (rel_type, direction) +# direction "ab" = new->match, "ba" = match->new +AUTO_EDGE_HEURISTICS = { + ("fix", "problem"): ("SOLVES", "ab"), + ("solution", "problem"): ("SOLVES", "ab"), + ("solution", "error"): ("SOLVES", "ab"), + ("fix", "error"): ("SOLVES", "ab"), + ("insight", "solution"): ("BUILDS_ON", "ab"), + ("insight", "decision"): ("BUILDS_ON", "ab"), + ("decision", "solution"): ("BUILDS_ON", "ab"), + ("fix", "solution"): ("BUILDS_ON", "ab"), + ("code_pattern", "solution"): ("BUILDS_ON", "ab"), + ("procedure", "workflow"): ("BUILDS_ON", "ab"), +} +AUTO_EDGE_SIMILARITY_THRESHOLD = 0.4 +AUTO_EDGE_MAX = 3 + _clients: Dict[str, CognitiveMemoryClient] = {} @@ -527,6 +544,97 @@ def create_tools() -> list: ] +def _auto_create_edges( + client: CognitiveMemoryClient, + memory_id: str, + title: str, + mem_type: str, + tags: Optional[list] = None, +) -> list: + """Auto-create edges between a newly stored memory and related existing memories. + + Uses recall to find similar memories and heuristic type-pairs to choose + relationship types. Returns list of created edge info dicts. + Never raises — failures are silently swallowed so store always succeeds. + """ + try: + # Build recall query from title + tags for better precision (#4) + query = title + if tags: + query = f"{title} {' '.join(tags)}" + results = client.recall(query, limit=5, semantic=True) + + # Filter out the just-stored memory + results = [r for r in results if r.get("id") != memory_id] + + # Filter by similarity threshold before slicing to AUTO_EDGE_MAX (#5) + # When embeddings are absent, keyword-only results lack "similarity" — + # require at least a tag overlap to avoid spurious edges (#1) + filtered = [] + new_tags = set(tags or []) + for result in results: + similarity = result.get("similarity") + if similarity is not None: + # Semantic path: use similarity threshold + if similarity < AUTO_EDGE_SIMILARITY_THRESHOLD: + continue + else: + # Keyword-only path: require at least one shared tag + match_tags = set(result.get("tags") or []) + if not new_tags or not (new_tags & match_tags): + continue + filtered.append(result) + + created_edges = [] + for result in filtered[:AUTO_EDGE_MAX]: + match_type = result.get("type", "") + match_id = result["id"] + match_title = result.get("title", "") + + # Look up heuristic for (new_type, match_type) in both orderings + key_ab = (mem_type, match_type) + key_ba = (match_type, mem_type) + + if key_ab in AUTO_EDGE_HEURISTICS: + rel_type, direction = AUTO_EDGE_HEURISTICS[key_ab] + from_id = memory_id if direction == "ab" else match_id + to_id = match_id if direction == "ab" else memory_id + elif key_ba in AUTO_EDGE_HEURISTICS: + rel_type, direction = AUTO_EDGE_HEURISTICS[key_ba] + # Reverse: if heuristic says ab for (match, new), then match->new + from_id = match_id if direction == "ab" else memory_id + to_id = memory_id if direction == "ab" else match_id + else: + rel_type = "RELATED_TO" + from_id = memory_id + to_id = match_id + + # Build description matching actual edge direction (#2) + from_title = title if from_id == memory_id else match_title + to_title = match_title if to_id == match_id else title + desc = f"Auto-edge: {from_title} → {to_title}" + edge_id = client.relate( + from_id=from_id, + to_id=to_id, + rel_type=rel_type, + description=desc, + ) + + if edge_id: # Empty string means duplicate + created_edges.append( + { + "edge_id": edge_id, + "rel_type": rel_type, + "linked_memory_id": match_id, + "linked_title": match_title, + } + ) + + return created_edges + except Exception: + return [] + + def handle_tool_call(tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: """Dispatch MCP tool calls to the CognitiveMemoryClient.""" @@ -561,12 +669,14 @@ def handle_tool_call(tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any tags=tags, ) episode_logged = True + auto_edges = _auto_create_edges(client, memory_id, title, mem_type, tags) _trigger_git_sync() return ok( { "success": True, "memory_id": memory_id, "episode_logged": episode_logged, + "auto_edges": auto_edges, } )