memory.py

"""
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