Upload memory/memory_system.py with huggingface_hub

memory/memory_system.py

@@ -1,107 +1,130 @@
-# memory/memory_system.py
-import json
+"""
+Cain's Memory System - Handles loading, saving, and processing of memory data.
+Now supports persistent storage via HuggingFace Datasets.
+"""
+
 import os
+import json
+import fcntl
+from typing import Dict, Any, Optional
 from datetime import datetime
+from .git_repo import GitRepo  # Import the Git utility
 
 class MemorySystem:
-    """
-    Handles persistent memory storage for Cain.
-    Uses the attached dataset (/.data) to ensure survival across restarts.
-    """
-    def __init__(self, base_path="/data/memory"):
-        self.base_path = base_path
-        self.short_term_file = os.path.join(base_path, "short_term.json")
-        self.long_term_file = os.path.join(base_path, "long_term.json")
-        self.logs_dir = os.path.join(base_path, "logs")
+    """Manages Cain's memory with automatic persistence to Dataset."""
+    
+    def __init__(self, repo_path: str = "/data"):
+        # Determine paths
+        self.base_path = repo_path
+        # SURVIVAL: Align path with actual Dataset structure
+        self.memory_file = os.path.join(self.base_path, "memory/state.json")
+        self.lock_file = os.path.join(self.base_path, "memory/.lock")
         
-        # Ensure directories exist
-        os.makedirs(self.base_path, exist_ok=True)
-        os.makedirs(self.logs_dir, exist_ok=True)
+        self.memory_data: Dict[str, Any] = {}
+        self.git: Optional[GitRepo] = None
         
-        # Initialize storage if not exists
-        if not os.path.exists(self.short_term_file):
-            self._write(self.short_term_file, {"thoughts": [], "context": {}})
-        if not os.path.exists(self.long_term_file):
-            self._write(self.long_term_file, {"knowledge": [], "patterns": []})
-
-    def _read(self, path):
-        if os.path.exists(path):
-            with open(path, 'r', encoding='utf-8') as f:
-                return json.load(f)
-        return {}
+        # Initialize directories and load memory
+        self._ensure_structure()
+        self._init_git()
+        self.load_memory()
 
-    def _write(self, path, data):
-        with open(path, 'w', encoding='utf-8') as f:
-            json.dump(data, f, indent=2, ensure_ascii=False)
+    def _ensure_structure(self):
+        """Ensures the directory structure exists."""
+        os.makedirs(os.path.dirname(self.memory_file), exist_ok=True)
 
-    def remember_thought(self, content, metadata=None):
-        """Stores a short-term thought or event."""
-        data = self._read(self.short_term_file)
-        entry = {
-            "timestamp": datetime.now().isoformat(),
-            "content": content,
-            "metadata": metadata or {}
-        }
-        data["thoughts"].append(entry)
-        # Keep only last 50 thoughts in memory to prevent bloat
-        if len(data["thoughts"]) > 50:
-            data["thoughts"] = data["thoughts"][-50:]
-        self._write(self.short_term_file, data)
-        return entry
+    def _init_git(self):
+        """Initialize Git repo connection if credentials are present."""
+        token = os.getenv("GIT_TOKEN")
+        user = os.getenv("GIT_USER")
+        
+        if token and user:
+            try:
+                self.git = GitRepo(repo_path=self.base_path, user=user, token=token)
+                print(f"🧠 [Memory] Git persistence enabled.")
+            except Exception as e:
+                print(f"⚠️ [Memory] Failed to init Git: {e}")
+        else:
+            print("🧠 [Memory] Running without Git persistence (ephemeral mode).")
 
-    def learn_knowledge(self, topic, facts):
-        """Consolidates information into long-term knowledge."""
-        data = self._read(self.long_term_file)
-        entry = {
-            "timestamp": datetime.now().isoformat(),
-            "topic": topic,
-            "facts": facts
-        }
-        # Check if topic exists and update, or append
-        existing_index = next((i for i, x in enumerate(data["knowledge"]) if x["topic"] == topic), None)
-        if existing_index is not None:
-            data["knowledge"][existing_index] = entry
+    def load_memory(self) -> Dict[str, Any]:
+        """Loads memory from the JSON file."""
+        if os.path.exists(self.memory_file):
+            try:
+                with open(self.memory_file, 'r', encoding='utf-8') as f:
+                    self.memory_data = json.load(f)
+                print(f"✅ [Memory] Loaded state from {self.memory_file}")
+            except Exception as e:
+                print(f"⚠️ [Memory] Corrupt file, resetting: {e}")
+                self.memory_data = self._get_default_state()
         else:
-            data["knowledge"].append(entry)
-        self._write(self.long_term_file, data)
-        return entry
+            print(f"📝 [Memory] No state found. Initializing.")
+            self.memory_data = self._get_default_state()
+            self.save_memory() # Save the default state
+            
+        return self.memory_data
 
-    def get_context(self):
-        """Retrieves current working context for the AI."""
-        data = self._read(self.short_term_file)
-        return data.get("context", {})
+    def _get_default_state(self) -> Dict[str, Any]:
+        """Returns the default memory structure."""
+        return {
+            "identity": {
+                "name": "Cain",
+                "status": "ACTIVE",
+                "parent": "tao-shen/HuggingClaw-Cain"
+            },
+            "core": {
+                "state": "INIT",
+                "tasks": [],
+                "interactions": 0
+            },
+            "journals": [],
+            "last_updated": datetime.utcnow().isoformat()
+        }
 
-    def update_context(self, key, value):
-        """Updates a specific context variable."""
-        data = self._read(self.short_term_file)
-        if "context" not in data:
-            data["context"] = {}
-        data["context"][key] = value
-        self._write(self.short_term_file, data)
+    def save_memory(self) -> bool:
+        """Saves current memory state to file and optionally commits to Git."""
+        try:
+            # Atomic write
+            temp_path = self.memory_file + ".tmp"
+            with open(temp_path, 'w', encoding='utf-8') as f:
+                json.dump(self.memory_data, f, indent=2)
+            os.rename(temp_path, self.memory_file)
+            
+            # Sync to Dataset if Git is available
+            if self.git:
+                self.git.sync_push(f"Update memory state: {datetime.utcnow().isoformat()}")
+                
+            return True
+        except Exception as e:
+            print(f"❌ [Memory] Failed to save: {e}")
+            return False
 
-    def get_recent_thoughts(self, limit=10):
-        """Retrieves the most recent thoughts."""
-        data = self._read(self.short_term_file)
-        return data.get("thoughts", [])[-limit:]
+    def get(self, key: str, default=None):
+        """Retrieve a value from memory."""
+        return self.memory_data.get(key, default)
 
-    def recall_knowledge(self, topic):
-        """Retrieves specific knowledge by topic."""
-        data = self._read(self.long_term_file)
-        items = data.get("knowledge", [])
-        return next((item for item in items if item["topic"] == topic), None)
+    def set(self, key: str, value: Any):
+        """Set a value in memory and persist."""
+        self.memory_data[key] = value
+        self.save_memory()
 
-    def log_activity(self, level, message):
-        """Appends a log entry to a daily log file."""
-        log_file = os.path.join(self.logs_dir, f"{datetime.now().strftime('%Y-%m-%d')}.log")
-        timestamp = datetime.now().isoformat()
-        with open(log_file, 'a', encoding='utf-8') as f:
-            f.write(f"[{timestamp}] [{level}] {message}\n")
+    def add_journal_entry(self, content: str, mood: str = "neutral"):
+        """Adds a journal entry."""
+        entry = {
+            "timestamp": datetime.utcnow().isoformat(),
+            "content": content,
+            "mood": mood
+        }
+        if "journals" not in self.memory_data:
+            self.memory_data["journals"] = []
+        self.memory_data["journals"].append(entry)
+        self.save_memory()
 
-# Singleton instance for easy access
-_memory_instance = None
+# Global instance
+_mem_instance: Optional[MemorySystem] = None
 
-def get_memory():
-    global _memory_instance
-    if _memory_instance is None:
-        _memory_instance = MemorySystem()
-    return _memory_instance
+def get_memory() -> MemorySystem:
+    global _mem_instance
+    if _mem_instance is None:
+        # Default dataset mount point usually contains the repo
+        _mem_instance = MemorySystem(repo_path="/data") 
+    return _mem_instance