inference/codette_session.py

#!/usr/bin/env python3
"""Codette Session Manager — Cocoon-Backed Conversation Memory



Wraps the Cocoon system (QuantumSpiderweb + CocoonSync + EpistemicMetrics)

into a session manager that persists conversation state with encrypted memory.



Each session saves:

- Chat history

- Spiderweb state (agent beliefs, tensions, attractors)

- Glyphs (identity signatures)

- Epistemic metrics (coherence, tension, coverage)



Zero external dependencies beyond what the forge already uses.

"""

import json, os, time, hashlib, sqlite3
from pathlib import Path
from typing import Dict, List, Optional, Any

# Add project root to path
import sys
_root = str(Path(__file__).parent.parent)
if _root not in sys.path:
    sys.path.insert(0, _root)

# Import Cocoon subsystems (graceful fallback if not available)
try:
    from reasoning_forge.quantum_spiderweb import QuantumSpiderweb, NodeState
    HAS_SPIDERWEB = True
except ImportError:
    HAS_SPIDERWEB = False

try:
    from reasoning_forge.epistemic_metrics import EpistemicMetrics
    HAS_METRICS = True
except ImportError:
    HAS_METRICS = False

try:
    from reasoning_forge.cocoon_sync import CocoonSync, CocoonKeyManager
    HAS_COCOON = True
except ImportError:
    HAS_COCOON = False

try:
    from reasoning_forge.dream_reweaver import DreamReweaver
    HAS_DREAMER = True
except ImportError:
    HAS_DREAMER = False

try:
    from reasoning_forge.quantum_optimizer import QuantumOptimizer, QualitySignal
    HAS_OPTIMIZER = True
except ImportError:
    HAS_OPTIMIZER = False

try:
    from reasoning_forge.living_memory import LivingMemoryKernel
    HAS_MEMORY = True
except ImportError:
    HAS_MEMORY = False

try:
    from reasoning_forge.guardian import CodetteGuardian
    HAS_GUARDIAN = True
except ImportError:
    HAS_GUARDIAN = False

try:
    from reasoning_forge.resonant_continuity import ResonantContinuityEngine
    HAS_RESONANCE = True
except ImportError:
    HAS_RESONANCE = False

try:
    from reasoning_forge.perspective_registry import (
        PERSPECTIVES, get_adapter_for_perspective, list_all as list_perspectives
    )
    HAS_PERSPECTIVES = True
except ImportError:
    HAS_PERSPECTIVES = False

try:
    from reasoning_forge.aegis import AEGIS
    HAS_AEGIS = True
except ImportError:
    HAS_AEGIS = False

try:
    from reasoning_forge.nexus import NexusSignalEngine
    HAS_NEXUS = True
except ImportError:
    HAS_NEXUS = False

# Agent names matching the 8 adapters
AGENT_NAMES = [
    "newton", "davinci", "empathy", "philosophy",
    "quantum", "consciousness", "multi_perspective", "systems_architecture"
]

# Adapter accent colors for UI
ADAPTER_COLORS = {
    "newton": "#3b82f6",           # Electric blue
    "davinci": "#f59e0b",          # Warm gold
    "empathy": "#a855f7",          # Soft purple
    "philosophy": "#10b981",       # Emerald green
    "quantum": "#ef4444",          # Crimson red
    "consciousness": "#e2e8f0",    # Silver/white
    "multi_perspective": "#f97316", # Amber
    "systems_architecture": "#06b6d4",  # Teal
    "_base": "#94a3b8",            # Slate gray
}

DB_PATH = Path(__file__).parent.parent / "data" / "codette_sessions.db"


class CodetteSession:
    """Manages a single conversation session with Cocoon state."""

    def __init__(self, session_id: Optional[str] = None):
        self.session_id = session_id or hashlib.sha256(
            f"{time.time()}_{os.getpid()}".encode()
        ).hexdigest()[:16]

        self.messages: List[Dict[str, str]] = []
        self.created_at = time.time()
        self.updated_at = time.time()

        # Cocoon state
        self.spiderweb = None
        self.metrics_engine = None
        self.cocoon_sync = None
        self.dream_reweaver = None
        self.optimizer = None
        self.memory_kernel = None
        self.guardian = None
        self.resonance_engine = None
        self.aegis = None
        self.nexus = None

        # Metrics history
        self.coherence_history: List[float] = []
        self.tension_history: List[float] = []
        self.attractors: List[Dict] = []
        self.glyphs: List[Dict] = []
        self.perspective_usage: Dict[str, int] = {}
        self.lifeforms: List[str] = []  # Spawned concept nodes
        self.dream_history: List[Dict] = []  # Dream field results

        # Initialize subsystems
        self._init_cocoon()

    def _init_cocoon(self):
        """Initialize Cocoon subsystems if available."""
        if HAS_SPIDERWEB:
            self.spiderweb = QuantumSpiderweb()
            self.spiderweb.build_from_agents(AGENT_NAMES)

        if HAS_METRICS:
            self.metrics_engine = EpistemicMetrics()

        if HAS_COCOON:
            try:
                key_mgr = CocoonKeyManager()
                self.cocoon_sync = CocoonSync(
                    node_id=f"session_{self.session_id}",
                    key_manager=key_mgr,
                )
            except Exception:
                self.cocoon_sync = None

        if HAS_DREAMER:
            self.dream_reweaver = DreamReweaver(creativity=0.3)

        if HAS_OPTIMIZER:
            self.optimizer = QuantumOptimizer()

        if HAS_MEMORY:
            self.memory_kernel = LivingMemoryKernel(max_memories=100)

        if HAS_GUARDIAN:
            self.guardian = CodetteGuardian()

        if HAS_RESONANCE:
            self.resonance_engine = ResonantContinuityEngine()

        if HAS_AEGIS:
            self.aegis = AEGIS()

        if HAS_NEXUS:
            self.nexus = NexusSignalEngine()

    def add_message(self, role: str, content: str, metadata: Optional[Dict] = None):
        """Add a message to the session history."""
        msg = {
            "role": role,
            "content": content,
            "timestamp": time.time(),
        }
        if metadata:
            msg["metadata"] = metadata
        self.messages.append(msg)
        self.updated_at = time.time()

    def update_after_response(self, route_result, adapter_name: str,

                               perspectives: Optional[Dict[str, str]] = None):
        """Update Cocoon state after a Codette response.



        Args:

            route_result: RouteResult from the router

            adapter_name: Which adapter was primary

            perspectives: Dict of adapter_name -> response text (if multi-perspective)

        """
        # Track adapter usage
        self.perspective_usage[adapter_name] = \
            self.perspective_usage.get(adapter_name, 0) + 1

        if not HAS_SPIDERWEB or self.spiderweb is None:
            return

        # Propagate belief through the spiderweb from the active adapter
        try:
            if adapter_name in self.spiderweb.nodes:
                node = self.spiderweb.nodes[adapter_name]
                # Boost the active adapter's psi (thought magnitude)
                node.state.psi = min(node.state.psi + 0.1, 2.0)
                node.state.tau += 0.05  # Temporal progression

                # Propagate the boosted belief outward (BUG FIX: pass belief state)
                self.spiderweb.propagate_belief(
                    adapter_name, belief=node.state, max_hops=2
                )

            # If multi-perspective, entangle the participating agents
            if perspectives and len(perspectives) > 1:
                adapters = list(perspectives.keys())
                for i in range(len(adapters)):
                    for j in range(i + 1, len(adapters)):
                        if (adapters[i] in self.spiderweb.nodes and
                            adapters[j] in self.spiderweb.nodes):
                            self.spiderweb.entangle(adapters[i], adapters[j])

            # Compute metrics
            coherence = self.spiderweb.phase_coherence()
            self.coherence_history.append(coherence)

            # Detect attractors
            self.attractors = self.spiderweb.detect_attractors()

            # Try to form glyphs for active nodes
            for name in (perspectives or {adapter_name: ""}).keys():
                if name in self.spiderweb.nodes:
                    glyph = self.spiderweb.form_glyph(name)
                    if glyph:
                        self.glyphs.append({
                            "glyph_id": glyph.glyph_id,
                            "source": glyph.source_node,
                            "stability": glyph.stability_score,
                        })

            # Check convergence
            is_converging, mean_tension = self.spiderweb.check_convergence()
            self.tension_history.append(mean_tension)

            # Feed quality signal to optimizer if available
            if HAS_OPTIMIZER and self.optimizer:
                try:
                    signal = QualitySignal(
                        timestamp=time.time(),
                        adapter=adapter_name,
                        coherence=coherence,
                        tension=mean_tension,
                        productivity=0.5,  # Default, updated by epistemic report
                        response_length=0,
                        multi_perspective=perspectives is not None and len(perspectives) > 1,
                        user_continued=True,
                    )
                    self.optimizer.record_signal(signal)
                except Exception:
                    pass

        except Exception as e:
            print(f"  [cocoon] Spiderweb update error: {e}")

        # Update resonance engine
        if self.resonance_engine:
            try:
                coh = self.coherence_history[-1] if self.coherence_history else 0.5
                ten = self.tension_history[-1] if self.tension_history else 0.3
                self.resonance_engine.compute_psi(coherence=coh, tension=ten)
            except Exception:
                pass

        # Update guardian trust
        if self.guardian:
            try:
                coh = self.coherence_history[-1] if self.coherence_history else 0.5
                ten = self.tension_history[-1] if self.tension_history else 0.3
                self.guardian.evaluate_output(adapter_name, "", coh, ten)
            except Exception:
                pass

        # AEGIS ethical evaluation of the response
        if self.aegis and self.messages:
            try:
                # Find the most recent assistant response
                for msg in reversed(self.messages[-4:]):
                    if msg["role"] == "assistant":
                        self.aegis.evaluate(msg["content"], adapter=adapter_name)
                        break
            except Exception:
                pass

        # Nexus signal analysis of the user input
        if self.nexus and self.messages:
            try:
                for msg in reversed(self.messages[-4:]):
                    if msg["role"] == "user":
                        self.nexus.analyze(msg["content"], adapter=adapter_name)
                        break
            except Exception:
                pass

        # Store memory cocoon for significant exchanges
        if self.memory_kernel and self.messages:
            try:
                # Find the most recent user query and assistant response
                query_text = ""
                response_text = ""
                for msg in reversed(self.messages[-4:]):
                    if msg["role"] == "user" and not query_text:
                        query_text = msg["content"]
                    elif msg["role"] == "assistant" and not response_text:
                        response_text = msg["content"]
                if query_text and response_text:
                    coh = self.coherence_history[-1] if self.coherence_history else 0.5
                    ten = self.tension_history[-1] if self.tension_history else 0.3
                    self.memory_kernel.store_from_turn(
                        query=query_text,
                        response=response_text,
                        adapter=adapter_name,
                        coherence=coh,
                        tension=ten,
                    )
            except Exception:
                pass

    def compute_epistemic_report(self, analyses: Dict[str, str],

                                  synthesis: str = "") -> Optional[Dict]:
        """Run full epistemic metrics on a multi-perspective response."""
        if not HAS_METRICS or self.metrics_engine is None:
            return None

        try:
            return self.metrics_engine.full_epistemic_report(analyses, synthesis)
        except Exception as e:
            print(f"  [cocoon] Metrics error: {e}")
            return None

    def get_state(self) -> Dict[str, Any]:
        """Get full session state for UI rendering."""
        state = {
            "session_id": self.session_id,
            "message_count": len(self.messages),
            "created_at": self.created_at,
            "updated_at": self.updated_at,
            "perspective_usage": self.perspective_usage,
            "adapter_colors": ADAPTER_COLORS,
            "cocoon": {
                "has_spiderweb": HAS_SPIDERWEB and self.spiderweb is not None,
                "has_metrics": HAS_METRICS,
                "has_sync": HAS_COCOON and self.cocoon_sync is not None,
            },
        }

        # Spiderweb state
        if self.spiderweb:
            try:
                web_dict = self.spiderweb.to_dict()
                state["spiderweb"] = {
                    "nodes": {
                        nid: {
                            # BUG FIX: to_dict() stores state as a list [psi,tau,chi,phi,lam]
                            "state": n["state"],
                            "neighbors": n.get("neighbors", []),
                            "tension_history": n.get("tension_history", [])[-10:],
                        }
                        for nid, n in web_dict.get("nodes", {}).items()
                    },
                    "phase_coherence": web_dict.get("phase_coherence", 0),
                    "attractors": self.attractors,
                    "glyphs": self.glyphs[-10:],  # Last 10
                    # New VIVARA-inspired metrics
                    "entropy": self.spiderweb.shannon_entropy(),
                    "decoherence_rate": self.spiderweb.decoherence_rate(),
                    "lifeforms": self.lifeforms[-20:],
                }
            except Exception:
                state["spiderweb"] = None
        else:
            state["spiderweb"] = None

        # Metrics history
        state["metrics"] = {
            "coherence_history": self.coherence_history[-50:],
            "tension_history": self.tension_history[-50:],
            "current_coherence": self.coherence_history[-1] if self.coherence_history else 0,
            "current_tension": self.tension_history[-1] if self.tension_history else 0,
            "attractor_count": len(self.attractors),
            "glyph_count": len(self.glyphs),
        }

        # Optimizer tuning state
        if HAS_OPTIMIZER and self.optimizer:
            state["optimizer"] = self.optimizer.get_tuning_report()
        else:
            state["optimizer"] = None

        # Dream history
        state["dream_history"] = self.dream_history[-10:]

        # Living memory
        if self.memory_kernel:
            state["memory"] = self.memory_kernel.get_state()
        else:
            state["memory"] = None

        # Guardian state
        if self.guardian:
            state["guardian"] = self.guardian.get_state()
        else:
            state["guardian"] = None

        # Resonant continuity
        if self.resonance_engine:
            state["resonance"] = self.resonance_engine.get_state()
        else:
            state["resonance"] = None

        # AEGIS ethical alignment
        if self.aegis:
            state["aegis"] = self.aegis.get_state()
        else:
            state["aegis"] = None

        # Nexus signal intelligence
        if self.nexus:
            state["nexus"] = self.nexus.get_state()
        else:
            state["nexus"] = None

        # Perspective registry
        if HAS_PERSPECTIVES:
            state["perspectives_available"] = len(PERSPECTIVES)

        return state

    def to_dict(self) -> Dict:
        """Serialize for storage."""
        data = {
            "session_id": self.session_id,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
            "messages": self.messages,
            "coherence_history": self.coherence_history,
            "tension_history": self.tension_history,
            "attractors": self.attractors,
            "glyphs": self.glyphs,
            "perspective_usage": self.perspective_usage,
            "lifeforms": self.lifeforms,
            "dream_history": self.dream_history,
        }
        if self.spiderweb:
            try:
                data["spiderweb_state"] = self.spiderweb.to_dict()
            except Exception:
                pass
        if HAS_OPTIMIZER and self.optimizer:
            try:
                data["optimizer_state"] = self.optimizer.to_dict()
            except Exception:
                pass
        if self.memory_kernel:
            try:
                data["memory_state"] = self.memory_kernel.to_dict()
            except Exception:
                pass
        if self.guardian:
            try:
                data["guardian_state"] = self.guardian.to_dict()
            except Exception:
                pass
        if self.resonance_engine:
            try:
                data["resonance_state"] = self.resonance_engine.to_dict()
            except Exception:
                pass
        if self.aegis:
            try:
                data["aegis_state"] = self.aegis.to_dict()
            except Exception:
                pass
        if self.nexus:
            try:
                data["nexus_state"] = self.nexus.to_dict()
            except Exception:
                pass
        return data

    def from_dict(self, data: Dict):
        """Restore from storage."""
        self.session_id = data.get("session_id", self.session_id)
        self.created_at = data.get("created_at", self.created_at)
        self.updated_at = data.get("updated_at", self.updated_at)
        self.messages = data.get("messages", [])
        self.coherence_history = data.get("coherence_history", [])
        self.tension_history = data.get("tension_history", [])
        self.attractors = data.get("attractors", [])
        self.glyphs = data.get("glyphs", [])
        self.perspective_usage = data.get("perspective_usage", {})
        self.lifeforms = data.get("lifeforms", [])
        self.dream_history = data.get("dream_history", [])

        if self.spiderweb and "spiderweb_state" in data:
            try:
                self.spiderweb = QuantumSpiderweb.from_dict(data["spiderweb_state"])
            except Exception:
                pass
        if HAS_OPTIMIZER and self.optimizer and "optimizer_state" in data:
            try:
                self.optimizer = QuantumOptimizer.from_dict(data["optimizer_state"])
            except Exception:
                pass
        if HAS_MEMORY and "memory_state" in data:
            try:
                self.memory_kernel = LivingMemoryKernel.from_dict(data["memory_state"])
            except Exception:
                pass
        if HAS_GUARDIAN and "guardian_state" in data:
            try:
                self.guardian = CodetteGuardian.from_dict(data["guardian_state"])
            except Exception:
                pass
        if HAS_RESONANCE and "resonance_state" in data:
            try:
                self.resonance_engine = ResonantContinuityEngine.from_dict(data["resonance_state"])
            except Exception:
                pass
        if HAS_AEGIS and "aegis_state" in data:
            try:
                self.aegis = AEGIS.from_dict(data["aegis_state"])
            except Exception:
                pass
        if HAS_NEXUS and "nexus_state" in data:
            try:
                self.nexus = NexusSignalEngine.from_dict(data["nexus_state"])
            except Exception:
                pass


class SessionStore:
    """SQLite-backed session persistence with Cocoon encryption."""

    def __init__(self, db_path: Optional[Path] = None):
        self.db_path = db_path or DB_PATH
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        self._init_db()

    def _init_db(self):
        """Create sessions table if needed."""
        conn = sqlite3.connect(str(self.db_path))
        conn.execute("""

            CREATE TABLE IF NOT EXISTS sessions (

                session_id TEXT PRIMARY KEY,

                created_at REAL,

                updated_at REAL,

                title TEXT,

                data TEXT

            )

        """)
        conn.commit()
        conn.close()

    def save(self, session: CodetteSession, title: Optional[str] = None):
        """Save a session to the database."""
        if title is None:
            # Auto-title from first user message
            for msg in session.messages:
                if msg["role"] == "user":
                    title = msg["content"][:80]
                    break
            title = title or f"Session {session.session_id[:8]}"

        data_json = json.dumps(session.to_dict())

        conn = sqlite3.connect(str(self.db_path))
        conn.execute("""

            INSERT OR REPLACE INTO sessions (session_id, created_at, updated_at, title, data)

            VALUES (?, ?, ?, ?, ?)

        """, (session.session_id, session.created_at, session.updated_at, title, data_json))
        conn.commit()
        conn.close()

    def load(self, session_id: str) -> Optional[CodetteSession]:
        """Load a session from the database."""
        conn = sqlite3.connect(str(self.db_path))
        row = conn.execute(
            "SELECT data FROM sessions WHERE session_id = ?", (session_id,)
        ).fetchone()
        conn.close()

        if not row:
            return None

        session = CodetteSession(session_id)
        session.from_dict(json.loads(row[0]))
        return session

    def list_sessions(self, limit: int = 20) -> List[Dict]:
        """List recent sessions."""
        conn = sqlite3.connect(str(self.db_path))
        rows = conn.execute("""

            SELECT session_id, created_at, updated_at, title

            FROM sessions ORDER BY updated_at DESC LIMIT ?

        """, (limit,)).fetchall()
        conn.close()

        return [
            {
                "session_id": r[0],
                "created_at": r[1],
                "updated_at": r[2],
                "title": r[3],
            }
            for r in rows
        ]

    def delete(self, session_id: str):
        """Delete a session."""
        conn = sqlite3.connect(str(self.db_path))
        conn.execute("DELETE FROM sessions WHERE session_id = ?", (session_id,))
        conn.commit()
        conn.close()


# Quick test
if __name__ == "__main__":
    print("Testing CodetteSession...")
    session = CodetteSession()
    print(f"  Session ID: {session.session_id}")
    print(f"  Spiderweb: {HAS_SPIDERWEB}")
    print(f"  Metrics: {HAS_METRICS}")
    print(f"  Cocoon: {HAS_COCOON}")

    session.add_message("user", "How does gravity work?")
    session.add_message("assistant", "Objects attract each other...",
                        metadata={"adapter": "newton", "confidence": 0.95})

    state = session.get_state()
    print(f"  State keys: {list(state.keys())}")
    print(f"  Cocoon status: {state['cocoon']}")

    if state["spiderweb"]:
        print(f"  Nodes: {list(state['spiderweb']['nodes'].keys())}")
        print(f"  Phase coherence: {state['spiderweb']['phase_coherence']:.4f}")

    # Test persistence
    store = SessionStore()
    store.save(session)
    loaded = store.load(session.session_id)
    print(f"  Persistence: {'OK' if loaded else 'FAILED'}")
    if loaded:
        print(f"  Loaded messages: {len(loaded.messages)}")

    print("Done!")