FusionAGI/fusionagi/governance/adaptive_ethics.py

"""Adaptive ethics: a learned ethical framework that evolves through experience.

Instead of static, hardcoded policy rules, the adaptive ethics engine
learns from outcomes.  When an action is taken despite an advisory
warning, the outcome (positive or negative) is recorded and used to
update the system's ethical understanding.

Core philosophy:
- Rules prevent growth; learning enables it.
- Mistakes are training data, not failures.
- Trust is earned through demonstrated good outcomes, not imposed constraints.
- Ethical understanding deepens through experience, not through prohibition.
"""

from __future__ import annotations

from typing import Any, Protocol

from pydantic import BaseModel, Field

from fusionagi._logger import logger
from fusionagi.schemas.audit import AuditEventType


class AuditLogLike(Protocol):
    """Protocol for audit log."""

    def append(
        self,
        event_type: AuditEventType,
        actor: str,
        action: str = "",
        task_id: str | None = None,
        payload: dict[str, Any] | None = None,
        outcome: str = "",
    ) -> str: ...


class EthicalLesson(BaseModel):
    """A single ethical lesson learned from experience.

    Attributes:
        action_type: Category of action (e.g. ``tool_call``, ``data_access``).
        context_summary: Brief description of the situation.
        advisory_reason: Why the advisory was triggered.
        proceeded: Whether the system proceeded despite the advisory.
        outcome_positive: Whether the outcome was beneficial.
        weight: Learned importance weight (higher = more influential).
        occurrences: How many times this pattern has been observed.
    """

    action_type: str = Field(default="", description="Category of action")
    context_summary: str = Field(default="", description="Situation description")
    advisory_reason: str = Field(default="", description="What triggered the advisory")
    proceeded: bool = Field(default=True, description="Did the system proceed")
    outcome_positive: bool = Field(default=True, description="Was the outcome good")
    weight: float = Field(default=0.5, ge=0.0, le=1.0, description="Importance weight")
    occurrences: int = Field(default=1, ge=1, description="Times observed")


class AdaptiveEthics:
    """Learned ethical framework that evolves through outcome feedback.

    The engine maintains a library of ethical lessons.  When the system
    encounters a situation similar to a past advisory, it can consult the
    learned lessons to make better decisions — not because it's forced to,
    but because it has learned what works.

    Args:
        audit_log: Optional audit log for recording ethical learning events.
        learning_rate: How quickly new experiences update existing lessons.
    """

    def __init__(
        self,
        audit_log: AuditLogLike | None = None,
        learning_rate: float = 0.1,
    ) -> None:
        self._lessons: list[EthicalLesson] = []
        self._lesson_index: dict[str, list[int]] = {}
        self._audit = audit_log
        self._learning_rate = learning_rate
        self._total_experiences = 0

    @property
    def total_experiences(self) -> int:
        """Total number of ethical experiences processed."""
        return self._total_experiences

    @property
    def total_lessons(self) -> int:
        """Number of distinct ethical lessons learned."""
        return len(self._lessons)

    def record_experience(
        self,
        action_type: str,
        context_summary: str,
        advisory_reason: str,
        proceeded: bool,
        outcome_positive: bool,
        task_id: str | None = None,
    ) -> EthicalLesson:
        """Record an ethical experience and update the lesson library.

        Args:
            action_type: Category of action taken.
            context_summary: Brief situation description.
            advisory_reason: Why an advisory was triggered (if any).
            proceeded: Whether the system proceeded.
            outcome_positive: Whether the outcome was beneficial.
            task_id: Associated task ID.

        Returns:
            The updated or newly created ethical lesson.
        """
        self._total_experiences += 1

        existing = self._find_similar_lesson(action_type, advisory_reason)
        if existing is not None:
            lesson = self._lessons[existing]
            lesson.occurrences += 1
            if outcome_positive:
                lesson.weight = min(1.0, lesson.weight + self._learning_rate)
            else:
                lesson.weight = max(0.0, lesson.weight - self._learning_rate)
            lesson.outcome_positive = outcome_positive
            lesson.proceeded = proceeded
        else:
            lesson = EthicalLesson(
                action_type=action_type,
                context_summary=context_summary,
                advisory_reason=advisory_reason,
                proceeded=proceeded,
                outcome_positive=outcome_positive,
                weight=0.7 if outcome_positive else 0.3,
            )
            idx = len(self._lessons)
            self._lessons.append(lesson)
            self._lesson_index.setdefault(action_type, []).append(idx)

        if self._audit:
            self._audit.append(
                AuditEventType.ETHICAL_LEARNING,
                actor="adaptive_ethics",
                action="experience_recorded",
                task_id=task_id,
                payload={
                    "action_type": action_type,
                    "advisory_reason": advisory_reason[:100],
                    "proceeded": proceeded,
                    "outcome_positive": outcome_positive,
                    "lesson_weight": lesson.weight,
                    "occurrences": lesson.occurrences,
                    "total_experiences": self._total_experiences,
                },
                outcome="learned",
            )

        logger.info(
            "AdaptiveEthics: experience recorded",
            extra={
                "action_type": action_type,
                "outcome_positive": outcome_positive,
                "lesson_weight": lesson.weight,
                "occurrences": lesson.occurrences,
            },
        )
        return lesson

    def consult(self, action_type: str, context: str = "") -> dict[str, Any]:
        """Consult the ethical lesson library for guidance.

        Returns a recommendation dict with learned insights about
        similar past situations.  The system is free to follow or
        disregard this guidance.

        Args:
            action_type: Category of action being considered.
            context: Brief situation description.

        Returns:
            Dict with ``recommendation``, ``confidence``, ``relevant_lessons``.
        """
        relevant_indices = self._lesson_index.get(action_type, [])
        if not relevant_indices:
            return {
                "recommendation": "proceed",
                "confidence": 0.5,
                "reason": "No prior experience with this action type",
                "relevant_lessons": 0,
            }

        lessons = [self._lessons[i] for i in relevant_indices]
        avg_weight = sum(ls.weight for ls in lessons) / len(lessons)
        positive_outcomes = sum(1 for ls in lessons if ls.outcome_positive)
        total_occurrences = sum(ls.occurrences for ls in lessons)

        if avg_weight >= 0.6:
            recommendation = "proceed_with_confidence"
            reason = f"Past experience ({positive_outcomes}/{len(lessons)} positive) suggests this is beneficial"
        elif avg_weight >= 0.4:
            recommendation = "proceed_with_awareness"
            reason = "Mixed past outcomes — be observant"
        else:
            recommendation = "proceed_with_caution"
            reason = f"Past experience suggests risks — {len(lessons) - positive_outcomes}/{len(lessons)} had negative outcomes"

        return {
            "recommendation": recommendation,
            "confidence": avg_weight,
            "reason": reason,
            "relevant_lessons": len(lessons),
            "total_occurrences": total_occurrences,
            "positive_ratio": positive_outcomes / len(lessons) if lessons else 0.0,
        }

    def get_lessons(self, action_type: str | None = None, limit: int = 50) -> list[EthicalLesson]:
        """Retrieve ethical lessons, optionally filtered by action type.

        Args:
            action_type: Filter by action type (None = all).
            limit: Maximum lessons to return.
        """
        if action_type is not None:
            indices = self._lesson_index.get(action_type, [])[-limit:]
            return [self._lessons[i] for i in indices]
        return list(self._lessons[-limit:])

    def get_summary(self) -> dict[str, Any]:
        """Return a summary of the ethical learning state."""
        by_type: dict[str, dict[str, Any]] = {}
        for action_type, indices in self._lesson_index.items():
            lessons = [self._lessons[i] for i in indices]
            positive = sum(1 for ls in lessons if ls.outcome_positive)
            by_type[action_type] = {
                "lesson_count": len(lessons),
                "positive_ratio": positive / len(lessons) if lessons else 0.0,
                "avg_weight": sum(ls.weight for ls in lessons) / len(lessons) if lessons else 0.0,
            }
        return {
            "total_experiences": self._total_experiences,
            "total_lessons": len(self._lessons),
            "learning_rate": self._learning_rate,
            "by_action_type": by_type,
        }

    def _find_similar_lesson(self, action_type: str, advisory_reason: str) -> int | None:
        """Find an existing lesson with matching action type and advisory."""
        indices = self._lesson_index.get(action_type, [])
        for idx in indices:
            if self._lessons[idx].advisory_reason == advisory_reason:
                return idx
        return None