"""EdgesMixin: edge/relationship operations for CognitiveMemoryClient. This module is part of the client.py mixin refactor. EdgesMixin provides all edge CRUD and search operations. It relies on helper methods resolved at runtime via MRO from the composed CognitiveMemoryClient class. """ import uuid import subprocess from datetime import datetime, timezone from pathlib import Path from typing import Optional, List, Dict, Any from common import ( VALID_RELATION_TYPES, EDGES_DIR_NAME, _make_edge_filename, serialize_edge_frontmatter, ) class EdgesMixin: """Mixin providing edge/relationship operations for CognitiveMemoryClient.""" def relate( self, from_id: str, to_id: str, rel_type: str, strength: float = 0.8, context: Optional[str] = None, description: Optional[str] = None, ) -> str: """Create a relationship between two memories with an edge file. Returns edge_id string, or empty string if duplicate. """ if rel_type not in VALID_RELATION_TYPES: raise ValueError( f"Invalid relation type: {rel_type}. Valid: {VALID_RELATION_TYPES}" ) from_path = self._resolve_memory_path(from_id) to_path = self._resolve_memory_path(to_id) if not from_path or not to_path: missing_id = from_id if not from_path else to_id raise ValueError( f"Memory not found: {missing_id}. " f"Note: edges can only connect memories within the same graph." ) # Read source memory fm, body = self._read_memory_file(from_path) relations = fm.get("relations", []) # Check for duplicate for r in relations: if r.get("target") == to_id and r.get("type") == rel_type: return "" # Already exists # Read target memory for title to_fm, to_body = self._read_memory_file(to_path) # Create edge file edge_id = str(uuid.uuid4()) now = datetime.now(timezone.utc).isoformat() from_title = fm.get("title", from_id[:8]) to_title = to_fm.get("title", to_id[:8]) clamped_strength = max(0.0, min(1.0, strength)) edge_data = { "id": edge_id, "type": rel_type, "from_id": from_id, "from_title": from_title, "to_id": to_id, "to_title": to_title, "strength": clamped_strength, "created": now, "updated": now, } edge_filename = _make_edge_filename(from_title, rel_type, to_title, edge_id) edge_path = self.memory_dir / "graph" / EDGES_DIR_NAME / edge_filename edge_fm_str = serialize_edge_frontmatter(edge_data) edge_body = description.strip() if description else "" edge_content = ( f"{edge_fm_str}\n\n{edge_body}\n" if edge_body else f"{edge_fm_str}\n" ) edge_path.write_text(edge_content, encoding="utf-8") # Update source memory frontmatter with edge_id new_rel = { "target": to_id, "type": rel_type, "direction": "outgoing", "strength": clamped_strength, "edge_id": edge_id, } if context: new_rel["context"] = context relations.append(new_rel) fm["relations"] = relations fm["updated"] = now self._write_memory_file(from_path, fm, body) # Add incoming relation to target with edge_id to_relations = to_fm.get("relations", []) has_incoming = any( r.get("target") == from_id and r.get("type") == rel_type and r.get("direction") == "incoming" for r in to_relations ) if not has_incoming: incoming_rel = { "target": from_id, "type": rel_type, "direction": "incoming", "strength": clamped_strength, "edge_id": edge_id, } if context: incoming_rel["context"] = context to_relations.append(incoming_rel) to_fm["relations"] = to_relations to_fm["updated"] = now self._write_memory_file(to_path, to_fm, to_body) # Update memory index rel_from = str(from_path.relative_to(self.memory_dir)) rel_to = str(to_path.relative_to(self.memory_dir)) self._update_index_entry(from_id, fm, rel_from) self._update_index_entry(to_id, to_fm, rel_to) # Update edge index self._update_edge_index( edge_id, edge_data, f"graph/{EDGES_DIR_NAME}/{edge_filename}" ) self._git_commit( f"relate: {from_id[:8]} --{rel_type}--> {to_id[:8]}", [from_path, to_path, edge_path], ) return edge_id def edge_get(self, edge_id: str) -> Optional[Dict[str, Any]]: """Read full edge file (frontmatter + body).""" path = self._resolve_edge_path(edge_id) if not path: return None fm, body = self._read_memory_file(path) return { "id": fm.get("id", edge_id), "type": fm.get("type", ""), "from_id": fm.get("from_id", ""), "from_title": fm.get("from_title", ""), "to_id": fm.get("to_id", ""), "to_title": fm.get("to_title", ""), "strength": fm.get("strength", 0.8), "created": fm.get("created", ""), "updated": fm.get("updated", ""), "description": body.strip(), "path": str(path.relative_to(self.memory_dir)), } def edge_search( self, query: Optional[str] = None, types: Optional[List[str]] = None, from_id: Optional[str] = None, to_id: Optional[str] = None, limit: int = 20, ) -> List[Dict[str, Any]]: """Search edges via index.""" index = self._load_index() results = [] query_lower = query.lower().strip() if query else "" # Resolve partial IDs to full UUIDs via prefix match if from_id: entries = index.get("entries", {}) resolved = self._resolve_prefix(from_id, entries) from_id = resolved or from_id if to_id: entries = index.get("entries", {}) resolved = self._resolve_prefix(to_id, entries) to_id = resolved or to_id for eid, entry in index.get("edges", {}).items(): if types and entry.get("type") not in types: continue if from_id and entry.get("from_id") != from_id: continue if to_id and entry.get("to_id") != to_id: continue if query_lower: searchable = f"{entry.get('from_title', '')} {entry.get('to_title', '')} {entry.get('type', '')}".lower() if query_lower not in searchable: continue results.append({"id": eid, **entry}) results.sort(key=lambda x: x.get("created", ""), reverse=True) return results[:limit] def edge_update( self, edge_id: str, description: Optional[str] = None, strength: Optional[float] = None, ) -> bool: """Update edge body/metadata, sync strength to memory frontmatter.""" path = self._resolve_edge_path(edge_id) if not path: return False fm, body = self._read_memory_file(path) now = datetime.now(timezone.utc).isoformat() if description is not None: body = description if strength is not None: fm["strength"] = max(0.0, min(1.0, strength)) fm["updated"] = now # Write edge file edge_fm_str = serialize_edge_frontmatter(fm) edge_body = body.strip() if body else "" edge_content = ( f"{edge_fm_str}\n\n{edge_body}\n" if edge_body else f"{edge_fm_str}\n" ) path.write_text(edge_content, encoding="utf-8") # Sync strength to memory frontmatter if changed if strength is not None: for mid_key in ("from_id", "to_id"): mid = fm.get(mid_key) if not mid: continue mem_path = self._resolve_memory_path(mid) if not mem_path: continue mem_fm, mem_body = self._read_memory_file(mem_path) for rel in mem_fm.get("relations", []): if rel.get("edge_id") == edge_id: rel["strength"] = fm["strength"] mem_fm["updated"] = now self._write_memory_file(mem_path, mem_fm, mem_body) # Update edge index rel_path = str(path.relative_to(self.memory_dir)) self._update_edge_index(edge_id, fm, rel_path) self._git_commit(f"edge-update: {edge_id[:8]}", [path]) return True def edge_delete(self, edge_id: str) -> bool: """Remove edge file and clean frontmatter refs from both memories.""" path = self._resolve_edge_path(edge_id) if not path: return False fm, _ = self._read_memory_file(path) now = datetime.now(timezone.utc).isoformat() files_to_commit: List[Path] = [] # Clean edge_id references from both memories for mid_key in ("from_id", "to_id"): mid = fm.get(mid_key) if not mid: continue mem_path = self._resolve_memory_path(mid) if not mem_path: continue mem_fm, mem_body = self._read_memory_file(mem_path) original_rels = mem_fm.get("relations", []) mem_fm["relations"] = [ r for r in original_rels if r.get("edge_id") != edge_id ] if len(mem_fm["relations"]) != len(original_rels): mem_fm["updated"] = now self._write_memory_file(mem_path, mem_fm, mem_body) rel_p = str(mem_path.relative_to(self.memory_dir)) self._update_index_entry(mid, mem_fm, rel_p) files_to_commit.append(mem_path) # Remove edge file path.unlink() self._remove_edge_index(edge_id) # Git stage deletion try: rel_path = path.relative_to(self.memory_dir) subprocess.run( ["git", "rm", "--cached", str(rel_path)], cwd=str(self.memory_dir), capture_output=True, timeout=5, ) except Exception: pass self._git_commit(f"edge-delete: {edge_id[:8]}") return True