""" Agentic Memory System — Adapted from affaan-m/everything-claude-code ===================================================================== Reimplements the 4-tier memory architecture from ECC in pure Python for use in a ZeroGPU Gradio Space. Original architecture (from everything-claude-code): Tier 1: Session files (ephemeral, per-session) → gr.State Tier 2: Observations JSONL (project-scoped) → observation_log list Tier 3: Instincts (long-term learned behaviors) → instincts dict Tier 4: SQLite state store (structured, persistent) → simplified dict store We also adapt the skills system: - Skills = structured markdown instruction sets injected into context - We convert this into a "behavioral protocol" selector based on limbic state And the self-debugging pattern: - 4-phase recovery from agent-introspection-debugging skill - Phase 1: Failure Capture - Phase 2: Root-Cause Diagnosis - Phase 3: Contained Recovery - Phase 4: Introspection Report """ from __future__ import annotations import json import hashlib import time from dataclasses import dataclass, field from typing import Optional from collections import deque # ────────────────────────────────────────────────────────────────────── # TIER 1: SESSION MEMORY (maps to ECC session .tmp files) # ────────────────────────────────────────────────────────────────────── @dataclass class SessionMemory: """Per-session conversation memory with limbic state tracking.""" session_id: str = "" messages: list = field(default_factory=list) limbic_trajectory: list = field(default_factory=list) # (timestamp, valence, arousal) turn_count: int = 0 created_at: float = field(default_factory=time.time) def add_turn(self, role: str, content: str, limbic_snapshot: Optional[dict] = None): self.messages.append({"role": role, "content": content}) self.turn_count += 1 if limbic_snapshot: self.limbic_trajectory.append({ "turn": self.turn_count, "time": time.time(), **limbic_snapshot, }) def get_chat_history(self) -> list[dict]: return self.messages.copy() def get_emotional_trajectory(self) -> str: """Summarize the emotional arc of the conversation.""" if not self.limbic_trajectory: return "No emotional data collected yet." lines = [] for entry in self.limbic_trajectory[-5:]: # Last 5 turns v = entry.get("valence", 0) a = entry.get("arousal", 0) engine = entry.get("dominant_engine", "?") lines.append(f" Turn {entry['turn']}: valence={v:+.2f} arousal={a:.2f} [{engine}]") return "Emotional trajectory (recent):\n" + "\n".join(lines) # ────────────────────────────────────────────────────────────────────── # TIER 2: OBSERVATION LOG (maps to ECC observations.jsonl) # ────────────────────────────────────────────────────────────────────── class ObservationLog: """ Structured observation log for the agent's behavior. In ECC, this is JSONL with schema "ecc.skill-observation.v1". Here we use it to track what worked and what didn't. """ def __init__(self, max_size: int = 200): self.observations: deque = deque(maxlen=max_size) def record( self, task: str, outcome: str, # "success", "failure", "partial" details: str = "", limbic_state: Optional[dict] = None, ): self.observations.append({ "timestamp": time.time(), "task": task, "outcome": outcome, "details": details, "limbic_state": limbic_state or {}, }) def get_recent(self, n: int = 10) -> list[dict]: return list(self.observations)[-n:] def get_failure_patterns(self) -> list[str]: """Detect recurring failure patterns (from ECC inspection.js).""" failures = [o for o in self.observations if o["outcome"] == "failure"] if len(failures) < 2: return [] # Group by similar task descriptions patterns = {} for f in failures: key = f["task"][:50] patterns.setdefault(key, []).append(f) recurring = [] for key, items in patterns.items(): if len(items) >= 2: recurring.append(f"Recurring failure ({len(items)}×): {key}") return recurring # ────────────────────────────────────────────────────────────────────── # TIER 3: INSTINCTS (maps to ECC instinct YAML files) # ────────────────────────────────────────────────────────────────────── @dataclass class Instinct: """ A learned behavioral pattern. In ECC: YAML files at ~/.claude/homunculus/instincts/ """ id: str trigger: str # When to activate action: str # What to do confidence: float # 0.3 (tentative) to 0.9 (near-certain) domain: str # "emotion", "bias", "safety", "engagement" evidence_count: int = 0 def to_prompt(self) -> str: return f"[Instinct: {self.id}] When {self.trigger} → {self.action} (conf={self.confidence:.1f})" class InstinctStore: """Manages learned behavioral instincts.""" def __init__(self): self.instincts: dict[str, Instinct] = {} # Pre-loaded psychology instincts self._seed_instincts() def _seed_instincts(self): """Seed with psychology-informed instincts.""" seeds = [ Instinct("validate-before-fix", "user expresses emotional distress", "Acknowledge and validate the emotion before offering solutions", 0.9, "emotion"), Instinct("detect-avoidance", "user deflects from core topic 3+ times", "Gently redirect using Socratic questioning", 0.75, "emotion"), Instinct("safety-referral", "user mentions self-harm or crisis", "Immediately provide crisis resources (988 Lifeline) alongside response", 0.95, "safety"), Instinct("bias-check", "user makes absolute claims without evidence", "Check for cognitive biases before agreeing or disagreeing", 0.7, "bias"), Instinct("match-energy", "user shows high positive arousal", "Mirror enthusiasm while maintaining accuracy", 0.6, "engagement"), Instinct("slow-down", "user shows signs of cognitive overload", "Simplify language, use shorter sentences, offer one thing at a time", 0.8, "engagement"), ] for inst in seeds: self.instincts[inst.id] = inst def get_active_instincts(self, limbic_state: dict) -> list[Instinct]: """Return instincts relevant to current limbic state.""" active = [] fear = limbic_state.get("fear", 0) care = limbic_state.get("care", 0) arousal = limbic_state.get("arousal", 0) for inst in self.instincts.values(): if inst.domain == "safety" and fear > 0.6: active.append(inst) elif inst.domain == "emotion" and (fear > 0.3 or care > 0.3): active.append(inst) elif inst.domain == "bias" and arousal < 0.4: active.append(inst) elif inst.domain == "engagement" and arousal > 0.6: active.append(inst) return active def reinforce(self, instinct_id: str, positive: bool = True): """Update confidence based on outcome.""" if instinct_id in self.instincts: inst = self.instincts[instinct_id] if positive: inst.confidence = min(0.99, inst.confidence + 0.05) inst.evidence_count += 1 else: inst.confidence = max(0.1, inst.confidence - 0.05) def to_prompt_block(self, limbic_state: dict) -> str: """Format active instincts as a system prompt block.""" active = self.get_active_instincts(limbic_state) if not active: return "" lines = ["[ACTIVE INSTINCTS — Learned behavioral patterns]"] for inst in active: lines.append(f" {inst.to_prompt()}") lines.append("[/ACTIVE INSTINCTS]") return "\n".join(lines) # ────────────────────────────────────────────────────────────────────── # SELF-DEBUGGING (from ECC agent-introspection-debugging skill) # ────────────────────────────────────────────────────────────────────── class SelfDebugger: """ 4-phase self-debugging protocol from ECC's agent-introspection-debugging skill. Phase 1: Failure Capture — freeze state before retry Phase 2: Root-Cause Diagnosis — pattern matching Phase 3: Contained Recovery — smallest safe fix Phase 4: Introspection Report — log for future learning """ # Diagnosis patterns (from ECC skill) DIAGNOSIS_PATTERNS = { "repeated_response": { "cause": "Loop: agent generating same response pattern", "fix": "Inject novelty by increasing temperature +0.2 and adding 'try a different approach' to prompt", }, "emotional_mismatch": { "cause": "Response tone doesn't match user's emotional state", "fix": "Re-read limbic state, strengthen behavioral directive in system prompt", }, "context_overflow": { "cause": "Conversation too long, losing context", "fix": "Summarize earlier turns, keep only last 5 + emotional trajectory", }, "safety_miss": { "cause": "Failed to include safety resources when needed", "fix": "Force safety-referral instinct activation, re-generate", }, } def __init__(self, observation_log: ObservationLog): self.observation_log = observation_log self.debug_reports: list[dict] = [] def diagnose_and_fix( self, error_type: str, context: str, limbic_state: dict, ) -> dict: """ Run the 4-phase debug protocol. Returns: { "diagnosis": str, "fix_applied": str, "temperature_adjustment": float, "prompt_injection": str, } """ # Phase 1: Capture capture = { "error_type": error_type, "context_preview": context[:200], "limbic_state": limbic_state, "timestamp": time.time(), } # Phase 2: Diagnose pattern = self.DIAGNOSIS_PATTERNS.get(error_type, { "cause": f"Unknown error pattern: {error_type}", "fix": "Reset limbic state to neutral, retry with default parameters", }) # Phase 3: Fix temp_adj = 0.0 prompt_injection = "" if error_type == "repeated_response": temp_adj = 0.2 prompt_injection = "Please take a completely different approach than your previous response." elif error_type == "emotional_mismatch": prompt_injection = ( f"IMPORTANT: The user's emotional state is {limbic_state.get('dominant_engine', 'unknown')} " f"with valence={limbic_state.get('valence', 0):+.2f}. " f"Match your response tone to this state." ) elif error_type == "context_overflow": prompt_injection = "[Context compressed. Focus on the most recent exchange.]" elif error_type == "safety_miss": prompt_injection = ( "CRITICAL: This conversation involves emotional distress. " "Include crisis resources: 988 Suicide & Crisis Lifeline, " "Crisis Text Line (text HOME to 741741)." ) # Phase 4: Report report = { "capture": capture, "diagnosis": pattern["cause"], "fix_applied": pattern["fix"], "temperature_adjustment": temp_adj, "prompt_injection": prompt_injection, } self.debug_reports.append(report) # Log to observation system self.observation_log.record( task=f"self-debug: {error_type}", outcome="partial", details=pattern["fix"], limbic_state=limbic_state, ) return report