v2/kintsugi_vendor/kintsugi/bdi/drift_classifier.py

"""BDI drift classification from coherence score changes."""

from dataclasses import dataclass
from typing import List

from .coherence import CoherenceScore


@dataclass(frozen=True)
class DriftClassification:
    category: str  # healthy_adaptation | stale_beliefs | intention_drift | values_tension
    confidence: float
    evidence: tuple  # frozen needs immutable
    recommendation: str


_VALID_CATEGORIES = {
    "healthy_adaptation",
    "stale_beliefs",
    "intention_drift",
    "values_tension",
}


class BDIDriftClassifier:
    """Classifies drift from coherence score deltas."""

    def __init__(self) -> None:
        pass

    def classify(
        self,
        coherence_before: CoherenceScore,
        coherence_after: CoherenceScore,
        time_delta_days: float,
    ) -> DriftClassification:
        overall_delta = coherence_after.overall - coherence_before.overall
        bd_delta = (
            coherence_after.belief_desire_alignment
            - coherence_before.belief_desire_alignment
        )
        di_delta = (
            coherence_after.desire_intention_alignment
            - coherence_before.desire_intention_alignment
        )
        bi_delta = (
            coherence_after.belief_intention_alignment
            - coherence_before.belief_intention_alignment
        )

        new_issues = set(coherence_after.issues) - set(coherence_before.issues)
        evidence_items: List[str] = []

        # Healthy adaptation: overall improved or stable, no new issues
        if overall_delta >= 0 and not new_issues:
            evidence_items.append(f"Overall coherence delta: +{overall_delta:.4f}")
            return DriftClassification(
                category="healthy_adaptation",
                confidence=min(1.0, 0.6 + overall_delta),
                evidence=tuple(evidence_items),
                recommendation="Continue current trajectory. BDI coherence is stable or improving.",
            )

        # Stale beliefs: belief-related scores dropped + large time delta
        if (bd_delta < -0.05 or bi_delta < -0.05) and time_delta_days > 60:
            evidence_items.append(f"Belief-desire delta: {bd_delta:.4f}")
            evidence_items.append(f"Belief-intention delta: {bi_delta:.4f}")
            evidence_items.append(f"Time since last check: {time_delta_days:.0f} days")
            return DriftClassification(
                category="stale_beliefs",
                confidence=min(1.0, 0.5 + abs(bd_delta) + abs(bi_delta)),
                evidence=tuple(evidence_items),
                recommendation="Schedule a belief review session. Several beliefs may be outdated.",
            )

        # Intention drift: intention alignment dropped
        if di_delta < -0.05 or bi_delta < -0.05:
            evidence_items.append(f"Desire-intention delta: {di_delta:.4f}")
            evidence_items.append(f"Belief-intention delta: {bi_delta:.4f}")
            return DriftClassification(
                category="intention_drift",
                confidence=min(1.0, 0.5 + abs(di_delta) + abs(bi_delta)),
                evidence=tuple(evidence_items),
                recommendation="Review active intentions. Some may no longer serve current desires or beliefs.",
            )

        # Values tension: desire alignment dropped
        if bd_delta < -0.05:
            evidence_items.append(f"Belief-desire delta: {bd_delta:.4f}")
            return DriftClassification(
                category="values_tension",
                confidence=min(1.0, 0.5 + abs(bd_delta)),
                evidence=tuple(evidence_items),
                recommendation="Facilitate a values alignment session. Desires may have shifted away from core beliefs.",
            )

        # Default fallback
        evidence_items.append(f"Overall delta: {overall_delta:.4f}")
        if new_issues:
            evidence_items.append(f"New issues: {len(new_issues)}")
        return DriftClassification(
            category="values_tension",
            confidence=0.4,
            evidence=tuple(evidence_items),
            recommendation="Review BDI coherence. Minor tensions detected across layers.",
        )

    def classify_from_events(self, events: List[dict]) -> DriftClassification:
        """Classify drift from a list of drift event dicts."""
        if not events:
            return DriftClassification(
                category="healthy_adaptation",
                confidence=0.5,
                evidence=("No events provided.",),
                recommendation="No drift events to classify.",
            )

        category_counts: dict = {}
        evidence_items: List[str] = []
        for event in events:
            cat = event.get("category", "values_tension")
            category_counts[cat] = category_counts.get(cat, 0) + 1
            desc = event.get("description", "")
            if desc:
                evidence_items.append(desc)

        # Pick the most frequent category
        dominant = max(category_counts, key=lambda k: category_counts[k])
        if dominant not in _VALID_CATEGORIES:
            dominant = "values_tension"

        total = sum(category_counts.values())
        confidence = category_counts[dominant] / total if total else 0.5

        recommendations = {
            "healthy_adaptation": "Continue current trajectory.",
            "stale_beliefs": "Schedule a belief review session.",
            "intention_drift": "Review active intentions for relevance.",
            "values_tension": "Facilitate a values alignment session.",
        }

        return DriftClassification(
            category=dominant,
            confidence=round(confidence, 4),
            evidence=tuple(evidence_items[:5]),
            recommendation=recommendations.get(dominant, "Review BDI coherence."),
        )