import logging from .working import WorkingMemory from .episodic import EpisodicMemory from .semantic import SemanticMemory from .reflection import ReflectionMemory _logger = logging.getLogger("memory.manager") class MemoryManager: """ S569: Unified Memory Manager (ARCH-K2.3). Coordina i 4 layer di memoria dell'agente. """ def __init__(self, sb_client=None, chroma_client=None): self.working = WorkingMemory() self.episodic = EpisodicMemory() self.semantic = SemanticMemory(sb_client, chroma_client) self.reflection = ReflectionMemory() async def init(self): """Inizializzazione asincrona (es. caricamento snapshot).""" await self.semantic.init() # S569: Auto-restore semantica se vuota await self._auto_restore_semantic() _logger.info("[MemoryManager] Layer inizializzati: working, episodic, semantic (pgvector=%s), reflection", getattr(self.semantic, '_pgvector', False)) async def save_working(self, goal: str, plan: list, facts: list): self.working.update(goal, plan, facts) # S569: backup periodico della working memory su episodic await self.save_episode("checkpoint", goal, f"Plan: {len(plan)} steps, Facts: {len(facts)}", True) async def save_episode(self, type_: str, task: str, output: str, success: bool, tags: list | None = None): self.episodic.add(type_, task, output, success, tags) if self.semantic.available and task: self.semantic.add(task, {"type": type_, "success": success}) async def search(self, query: str, n: int = 5, layer: str | None = None) -> list[dict]: results = [] if layer in (None, "semantic") and self.semantic.available: for h in self.semantic.search(query, n_results=n): results.append({**h, "layer": "semantic"}) if layer in (None, "episodic"): for ep in self.episodic.search_text(query, n=n): results.append({ "content": f"{ep.task} → {ep.output[:300]}", "layer": "episodic", "type": ep.type, "success": ep.success, }) if layer == "reflection": lessons = self.reflection.get_relevant_lessons(query, n=n) results.extend([{**l, "layer": "reflection"} for l in lessons]) return results[:n] async def get_context(self, query: str, code_length: int = 0, n: int = 5) -> str: """Return a bounded text context for consumers such as UnifiedAgentLoop. The loop needs a context-shaped view, while the public manager API exposes structured search results. Keep this adapter here so callers do not reach into individual memory layers or depend on their implementation details. """ if not query: return "" hits = await self.search(query, n=n) if not hits: return "" # Leave room for the current prompt/context; never inject an unbounded # memory payload into a long-running agent loop. max_chars = max(1000, min(4000, 4000 - max(0, code_length))) parts: list[str] = [] used = 0 for hit in hits: content = str(hit.get("content", "")).strip() if not content: continue layer = str(hit.get("layer", "memory")) block = f"[{layer}] {content}" remaining = max_chars - used if remaining <= 0: break parts.append(block[:remaining]) used += len(parts[-1]) + 1 return "\n".join(parts).strip() async def reflect(self, task: str, output: str, success: bool, error: str | None = None) -> dict: if success: self.reflection.record_success(task, output[:500]) await self.save_episode("fix", task, output, True) else: self.reflection.record_failure(task, error or output[:500]) await self.save_episode("error", task, error or output[:500], False) return { "recorded": True, "top_patterns": self.reflection.get_top_patterns(5), "lessons": self.reflection.get_relevant_lessons(task, 4), } async def _auto_restore_semantic(self) -> None: """Auto-restore: se la semantic memory è vuota, carica l'ultimo snapshot da GitHub.""" import asyncio as _asyncio, os if not self.semantic.available: return count = await _asyncio.to_thread(self.semantic.count) if count > 0: return token = os.environ.get("GITHUB_TOKEN", "") if not token: return try: import urllib.request as _urq, json as _json, base64 as _b64 req = _urq.Request( "https://api.github.com/repos/Baida98/AI/contents/data/semantic_snapshot.json", headers={ "Authorization": f"Bearer {token}", "Accept": "application/vnd.github.v3+json", "User-Agent": "agente-ai-backend", }, ) with _urq.urlopen(req, timeout=10) as resp: meta = _json.loads(resp.read()) records = _json.loads(_b64.b64decode(meta["content"]).decode("utf-8")) if not records: return result = await _asyncio.to_thread(self.semantic.import_all, records) _logger.info("[MemoryManager] ✓ Auto-restore semantica: %d record da GitHub snapshot", result["imported"]) except Exception as exc: _logger.debug("[MemoryManager] Auto-restore semantica: snapshot non disponibile (%s)", exc.__class__.__name__) def stats(self) -> dict: return { "working": self.working.stats(), "episodic": self.episodic.stats(), "semantic": self.semantic.stats(), "reflection": self.reflection.stats(), } async def clear(self, layer: str | None = None): if layer in (None, "working"): self.working.clear() if layer in (None, "episodic"): import sqlite3 if self.episodic._db: try: self.episodic._db.execute("DELETE FROM episodes") self.episodic._db.commit() except Exception as _e: try: self.episodic._db.rollback() except Exception: pass raise RuntimeError(f"clear episodic fallito: {_e}") from _e # ── Singleton globale — inizializzato in main.py _on_startup (GAP-5-fix) ───── _global_manager: 'MemoryManager | None' = None