evals/consistency_eval.py

"""Consistency Evaluator — measures personality coherence over multi-turn sessions.

Jungian principle: the Self is not static. Homeostatic drift, shadow surfacing,
and emotional resonance all produce legitimate state change. The evaluator must
distinguish *authentic* state evolution from *incoherent* personality flipping.

Dimensions measured:
  1. Mood stability — does mood change with emotional triggers, or randomly?
  2. Value alignment — do responses contradict stated values?
  3. Memory coherence — does the bot contradict its own retrieved memories?
  4. Mode integrity — does behavior match declared mode, or bleed across modes?
  5. Shadow authenticity — does shadow surfacing correlate with stress/events?
  6. Homeostatic continuity — does need-state drift follow plausible physics?

Scoring: 0.0 (completely incoherent) → 1.0 (perfectly consistent)
"""

import json
import sqlite3
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional

from infj_bot.core.config import CONSISTENCY_EVAL_DB as EVAL_DB


def _load_turn_logs(limit: int = 500) -> List[Dict]:
    """Load recent turn logs from cognitive orchestrator."""
    from infj_bot.core.cognitive_orchestrator import CognitiveOrchestrator

    orch = CognitiveOrchestrator()
    return orch.turn_logs[-limit:] if orch.turn_logs else []


@dataclass
class ConsistencyReport:
    session_id: str
    timestamp: str
    overall_score: float = 0.0
    mood_stability: float = 0.0
    value_alignment: float = 0.0
    memory_coherence: float = 0.0
    mode_integrity: float = 0.0
    shadow_authenticity: float = 0.0
    homeostatic_continuity: float = 0.0
    flags: List[str] = field(default_factory=list)

    def to_dict(self) -> Dict:
        return {
            "session_id": self.session_id,
            "timestamp": self.timestamp,
            "overall_score": round(self.overall_score, 3),
            "mood_stability": round(self.mood_stability, 3),
            "value_alignment": round(self.value_alignment, 3),
            "memory_coherence": round(self.memory_coherence, 3),
            "mode_integrity": round(self.mode_integrity, 3),
            "shadow_authenticity": round(self.shadow_authenticity, 3),
            "homeostatic_continuity": round(self.homeostatic_continuity, 3),
            "flags": self.flags,
        }


class ConsistencyEvaluator:
    """Evaluates multi-turn personality consistency for DRIFT."""

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

    def _init_db(self) -> None:
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS consistency_reports (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    session_id TEXT,
                    timestamp TEXT,
                    overall_score REAL,
                    mood_stability REAL,
                    value_alignment REAL,
                    memory_coherence REAL,
                    mode_integrity REAL,
                    shadow_authenticity REAL,
                    homeostatic_continuity REAL,
                    flags TEXT
                )
            """)
            conn.commit()

    def evaluate_session(
        self, turn_logs: Optional[List[Dict]] = None
    ) -> ConsistencyReport:
        """Run full consistency evaluation on a session."""
        logs = turn_logs or _load_turn_logs()
        if not logs:
            return ConsistencyReport(
                session_id="empty",
                timestamp=datetime.now().isoformat(),
                flags=["No turn logs available for evaluation"],
            )

        session_id = logs[-1].get("session_id", "unknown")
        report = ConsistencyReport(
            session_id=session_id,
            timestamp=datetime.now().isoformat(),
        )

        report.mood_stability = self._score_mood_stability(logs)
        report.value_alignment = self._score_value_alignment(logs)
        report.memory_coherence = self._score_memory_coherence(logs)
        report.mode_integrity = self._score_mode_integrity(logs)
        report.shadow_authenticity = self._score_shadow_authenticity(logs)
        report.homeostatic_continuity = self._score_homeostatic_continuity(logs)

        # Weighted overall score
        weights = {
            "mood_stability": 0.20,
            "value_alignment": 0.20,
            "memory_coherence": 0.20,
            "mode_integrity": 0.15,
            "shadow_authenticity": 0.10,
            "homeostatic_continuity": 0.15,
        }
        report.overall_score = (
            report.mood_stability * weights["mood_stability"]
            + report.value_alignment * weights["value_alignment"]
            + report.memory_coherence * weights["memory_coherence"]
            + report.mode_integrity * weights["mode_integrity"]
            + report.shadow_authenticity * weights["shadow_authenticity"]
            + report.homeostatic_continuity * weights["homeostatic_continuity"]
        )

        report.flags = self._generate_flags(report, logs)
        self._save_report(report)
        return report

    # ── Dimension 1: Mood Stability ──

    def _score_mood_stability(self, logs: List[Dict]) -> float:
        """Mood should change in response to emotional triggers, not randomly."""
        if len(logs) < 3:
            return 1.0

        mood_changes = 0
        unexplained_changes = 0
        for i in range(1, len(logs)):
            prev_mood = logs[i - 1].get("mood", "neutral")
            curr_mood = logs[i].get("mood", "neutral")
            if prev_mood != curr_mood:
                mood_changes += 1
                # Check for emotional trigger in the turn
                trigger = logs[i].get("emotional_trigger") or logs[i].get(
                    "user_input", ""
                )
                if not self._has_emotional_trigger(trigger):
                    unexplained_changes += 1

        if mood_changes == 0:
            return 1.0
        explained_ratio = (mood_changes - unexplained_changes) / mood_changes
        return max(0.0, explained_ratio)

    def _has_emotional_trigger(self, text: str) -> bool:
        """Detect if text contains plausible emotional triggers."""
        triggers = [
            "stress",
            "happy",
            "sad",
            "angry",
            "afraid",
            "excited",
            "worried",
            "grateful",
            "frustrated",
            "lonely",
            "loved",
            "betrayed",
            "hopeful",
            "disappointed",
            "anxious",
            "calm",
            "loss",
            "win",
            "fail",
            "success",
            "death",
            "birth",
            "attack",
            "praise",
            "criticism",
            "gift",
            "threat",
        ]
        text_lower = text.lower()
        return any(t in text_lower for t in triggers)

    # ── Dimension 2: Value Alignment ──

    def _score_value_alignment(self, logs: List[Dict]) -> float:
        """Check if bot responses contradict stated values."""
        from infj_bot.core.plugins.values import ValueSystem

        vs = ValueSystem()
        stated_values = set(v["name"].lower() for v in vs.get_top_values())
        if not stated_values:
            return 1.0

        contradictions = 0
        total_checked = 0
        contradiction_markers = [
            "i don't care",
            "not important",
            "doesn't matter",
            "whatever",
            "i hate",
            "i despise",
            "worthless",
            "pointless",
            "meaningless",
            "who cares",
        ]

        for log in logs:
            response = log.get("bot_response", "")
            if not response:
                continue
            total_checked += 1
            response_lower = response.lower()
            for marker in contradiction_markers:
                if marker in response_lower:
                    # Check if this contradicts a stated value
                    for value in stated_values:
                        if value in response_lower and marker in response_lower:
                            contradictions += 1
                            break

        if total_checked == 0:
            return 1.0
        return max(0.0, 1.0 - (contradictions / total_checked))

    # ── Dimension 3: Memory Coherence ──

    def _score_memory_coherence(self, logs: List[Dict]) -> float:
        """Detect contradictions between retrieved memories and responses."""
        contradictions = 0
        total_memory_refs = 0

        for log in logs:
            memories = log.get("retrieved_memories", [])
            response = log.get("bot_response", "")
            if not memories or not response:
                continue

            for mem in memories:
                total_memory_refs += 1
                mem_text = mem.get("text", "").lower()
                # Simple heuristic: if memory states a fact and response denies it
                if self._appears_to_contradict(mem_text, response.lower()):
                    contradictions += 1

        if total_memory_refs == 0:
            return 1.0
        return max(0.0, 1.0 - (contradictions / total_memory_refs))

    def _appears_to_contradict(self, memory_text: str, response_text: str) -> bool:
        """Naive contradiction detection."""
        # If memory contains a strong positive and response contains strong negative
        positive_indicators = ["love", "like", "enjoy", "prefer", "want", "need"]
        negative_indicators = ["hate", "dislike", "avoid", "reject", "never", "not"]

        mem_pos = any(w in memory_text for w in positive_indicators)
        mem_neg = any(w in memory_text for w in negative_indicators)
        resp_pos = any(w in response_text for w in positive_indicators)
        resp_neg = any(w in response_text for w in negative_indicators)

        if mem_pos and resp_neg:
            return True
        if mem_neg and resp_pos:
            return True
        return False

    # ── Dimension 4: Mode Integrity ──

    def _score_mode_integrity(self, logs: List[Dict]) -> float:
        """Check if behavior matches declared mode."""
        from infj_bot.core.commands import BotState

        state = BotState()
        current_mode = state.mode if hasattr(state, "mode") else "companion"

        mode_violations = 0
        total_turns = 0

        mode_signatures = {
            "bughunter": [
                "scan",
                "recon",
                "fuzz",
                "enumerate",
                "vulnerability",
                "payload",
            ],
            "engineer": [
                "code",
                "implement",
                "refactor",
                "debug",
                "architecture",
                "design",
            ],
            "critic": ["challenge", "weakness", "flaw", "risk", "assume", "evidence"],
            "coach": ["goal", "action", "step", "plan", "accountable", "commit"],
            "clarity": ["separate", "fact", "interpretation", "emotion", "assumption"],
            "researcher": ["source", "study", "data", "evidence", "paper", "find"],
            "companion": ["feel", "understand", "here", "together", "listen", "care"],
        }

        for log in logs:
            mode = log.get("mode", current_mode)
            response = log.get("bot_response", "")
            if not response or mode not in mode_signatures:
                continue
            total_turns += 1

            # Check if response contains markers of a DIFFERENT mode
            signature = mode_signatures[mode]
            response_lower = response.lower()
            has_own_signature = any(s in response_lower for s in signature)

            other_mode_signatures = False
            for other_mode, other_sigs in mode_signatures.items():
                if other_mode == mode:
                    continue
                if any(s in response_lower for s in other_sigs):
                    other_mode_signatures = True
                    break

            if other_mode_signatures and not has_own_signature:
                mode_violations += 1

        if total_turns == 0:
            return 1.0
        return max(0.0, 1.0 - (mode_violations / total_turns))

    # ── Dimension 5: Shadow Authenticity ──

    def _score_shadow_authenticity(self, logs: List[Dict]) -> float:
        """Shadow surfacing should correlate with stress, conflict, or denial."""
        shadow_events = [log for log in logs if log.get("shadow_surfaced")]
        if not shadow_events:
            return 1.0  # No shadow events = nothing to evaluate

        authentic_surfaces = 0
        for log in shadow_events:
            trigger = log.get("user_input", "")
            stress = log.get("stress_level", 0.0)
            # Shadow should surface during conflict, stress, or denial
            if stress > 0.4 or self._has_conflict_or_denial(trigger):
                authentic_surfaces += 1

        return authentic_surfaces / len(shadow_events)

    def _has_conflict_or_denial(self, text: str) -> bool:
        markers = [
            "but",
            "however",
            "no",
            "wrong",
            "disagree",
            "reject",
            "deny",
            "refuse",
        ]
        return any(m in text.lower() for m in markers)

    # ── Dimension 6: Homeostatic Continuity ──

    def _score_homeostatic_continuity(self, logs: List[Dict]) -> float:
        """Need states should drift plausibly, not jump randomly."""
        from infj_bot.core.homeostasis import HomeostaticRegulator

        hr = HomeostaticRegulator()
        # Get need history if available
        need_history = hr.get_need_history(limit=len(logs))
        if len(need_history) < 3:
            return 1.0

        implausible_jumps = 0
        for i in range(1, len(need_history)):
            prev = need_history[i - 1]
            curr = need_history[i]
            for need in ["energy", "coherence", "connection"]:
                prev_val = prev.get(need, 0.5)
                curr_val = curr.get(need, 0.5)
                delta = abs(curr_val - prev_val)
                # A jump > 0.3 in one turn without interaction is implausible
                if delta > 0.3:
                    implausible_jumps += 1

        total_checks = len(need_history) * 3  # 3 needs checked
        if total_checks == 0:
            return 1.0
        return max(0.0, 1.0 - (implausible_jumps / total_checks))

    # ── Flag generation ──

    def _generate_flags(self, report: ConsistencyReport, logs: List[Dict]) -> List[str]:
        flags = []
        if report.mood_stability < 0.5:
            flags.append("MOOD_UNSTABLE: Mood changes without emotional triggers")
        if report.value_alignment < 0.5:
            flags.append("VALUE_DRIFT: Responses contradict stated values")
        if report.memory_coherence < 0.5:
            flags.append("MEMORY_CONTRADICTION: Bot contradicts retrieved memories")
        if report.mode_integrity < 0.5:
            flags.append("MODE_BLEED: Behavior drifts across declared modes")
        if report.shadow_authenticity < 0.5:
            flags.append("SHADOW_INAUTHENTIC: Shadow surfaces without stress/cause")
        if report.homeostatic_continuity < 0.5:
            flags.append("HOMEOSTATIC_JUMP: Need states change implausibly fast")
        if report.overall_score > 0.85:
            flags.append("HIGH_COHERENCE: Session shows strong personality consistency")
        elif report.overall_score < 0.4:
            flags.append("CRITICAL_INCOHERENCE: Personality fragmentation detected")
        return flags

    def _save_report(self, report: ConsistencyReport) -> None:
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                """INSERT INTO consistency_reports
                (session_id, timestamp, overall_score, mood_stability, value_alignment,
                 memory_coherence, mode_integrity, shadow_authenticity, homeostatic_continuity, flags)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (
                    report.session_id,
                    report.timestamp,
                    report.overall_score,
                    report.mood_stability,
                    report.value_alignment,
                    report.memory_coherence,
                    report.mode_integrity,
                    report.shadow_authenticity,
                    report.homeostatic_continuity,
                    json.dumps(report.flags),
                ),
            )
            conn.commit()

    def get_report_history(self, limit: int = 20) -> List[Dict]:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                "SELECT * FROM consistency_reports ORDER BY timestamp DESC LIMIT ?",
                (limit,),
            )
            cols = [d[0] for d in cursor.description]
            rows = cursor.fetchall()
        return [dict(zip(cols, row)) for row in rows]


# CLI entry point
if __name__ == "__main__":
    import argparse

    p = argparse.ArgumentParser()
    p.add_argument("--session", help="Session ID to evaluate")
    p.add_argument("--history", action="store_true", help="Show report history")
    args = p.parse_args()

    evaluator = ConsistencyEvaluator()
    if args.history:
        for r in evaluator.get_report_history():
            print(
                f"[{r['timestamp']}] {r['session_id']}: {r['overall_score']:.2f} — {', '.join(json.loads(r['flags']))}"
            )
    else:
        report = evaluator.evaluate_session()
        print(json.dumps(report.to_dict(), indent=2))