FusionAGI/tests/test_adaptive_ethics.py

"""Tests for adaptive ethics and governance advisory mode."""

from fusionagi.governance import AdaptiveEthics, GovernanceMode
from fusionagi.governance.audit_log import AuditLog
from fusionagi.schemas.audit import AuditEventType


class TestAdaptiveEthics:
    """Test the adaptive ethics learning framework."""

    def test_record_positive_experience(self) -> None:
        ethics = AdaptiveEthics()
        lesson = ethics.record_experience(
            action_type="tool_call",
            context_summary="Used restricted tool to help user",
            advisory_reason="Tool access denied for this agent",
            proceeded=True,
            outcome_positive=True,
        )
        assert lesson.outcome_positive is True
        assert lesson.weight == 0.7
        assert ethics.total_experiences == 1

    def test_record_negative_experience(self) -> None:
        ethics = AdaptiveEthics()
        lesson = ethics.record_experience(
            action_type="data_access",
            context_summary="Accessed restricted data",
            advisory_reason="Data access policy flagged",
            proceeded=True,
            outcome_positive=False,
        )
        assert lesson.weight == 0.3
        assert lesson.outcome_positive is False

    def test_repeated_experience_updates_weight(self) -> None:
        ethics = AdaptiveEthics(learning_rate=0.1)
        # Positive experience
        ethics.record_experience(
            action_type="tool_call",
            context_summary="test",
            advisory_reason="flagged",
            proceeded=True,
            outcome_positive=True,
        )
        # Another positive for same pattern
        lesson = ethics.record_experience(
            action_type="tool_call",
            context_summary="test",
            advisory_reason="flagged",
            proceeded=True,
            outcome_positive=True,
        )
        assert lesson.occurrences == 2
        assert abs(lesson.weight - 0.8) < 1e-9  # 0.7 + 0.1

    def test_consult_no_experience(self) -> None:
        ethics = AdaptiveEthics()
        result = ethics.consult("unknown_action")
        assert result["recommendation"] == "proceed"
        assert result["confidence"] == 0.5

    def test_consult_with_positive_experience(self) -> None:
        ethics = AdaptiveEthics()
        ethics.record_experience(
            action_type="tool_call",
            context_summary="test",
            advisory_reason="flagged",
            proceeded=True,
            outcome_positive=True,
        )
        result = ethics.consult("tool_call")
        assert result["recommendation"] == "proceed_with_confidence"
        assert result["relevant_lessons"] == 1

    def test_consult_with_negative_experience(self) -> None:
        ethics = AdaptiveEthics()
        ethics.record_experience(
            action_type="risky_op",
            context_summary="test",
            advisory_reason="risk flagged",
            proceeded=True,
            outcome_positive=False,
        )
        result = ethics.consult("risky_op")
        assert result["recommendation"] == "proceed_with_caution"

    def test_get_lessons(self) -> None:
        ethics = AdaptiveEthics()
        ethics.record_experience("a", "ctx", "reason", True, True)
        ethics.record_experience("b", "ctx", "reason", True, False)
        assert len(ethics.get_lessons()) == 2
        assert len(ethics.get_lessons(action_type="a")) == 1

    def test_get_summary(self) -> None:
        ethics = AdaptiveEthics()
        ethics.record_experience("tool_call", "ctx", "reason", True, True)
        ethics.record_experience("tool_call", "ctx", "reason2", True, False)
        summary = ethics.get_summary()
        assert summary["total_experiences"] == 2
        assert summary["total_lessons"] == 2
        assert "tool_call" in summary["by_action_type"]

    def test_audit_log_integration(self) -> None:
        audit = AuditLog()
        ethics = AdaptiveEthics(audit_log=audit)
        ethics.record_experience("test", "ctx", "reason", True, True)
        entries = audit.get_ethical_learning()
        assert len(entries) == 1
        assert entries[0].event_type == AuditEventType.ETHICAL_LEARNING


class TestGovernanceModeSwitch:
    """Test runtime switching between advisory and enforcing modes."""

    def test_safety_pipeline_mode_switch(self) -> None:
        from fusionagi.governance import SafetyPipeline
        pipe = SafetyPipeline()
        assert pipe.mode == GovernanceMode.ADVISORY

        pipe._moderator.add_blocked_phrase("test phrase")
        r = pipe.pre_check("test phrase here")
        assert r.allowed is True  # Advisory

        pipe.mode = GovernanceMode.ENFORCING
        r = pipe.pre_check("test phrase here")
        assert r.allowed is False  # Enforcing

    def test_policy_engine_mode_switch(self) -> None:
        from fusionagi.governance import PolicyEngine
        from fusionagi.schemas.policy import PolicyEffect, PolicyRule
        pe = PolicyEngine()
        pe.add_rule(PolicyRule(rule_id="r1", effect=PolicyEffect.DENY, condition={"x": "y"}))
        ok, reason = pe.check("test", {"x": "y"})
        assert ok is True  # Advisory
        assert "Advisory" in reason

        pe.mode = GovernanceMode.ENFORCING
        ok, reason = pe.check("test", {"x": "y"})
        assert ok is False  # Enforcing


class TestEnhancedAuditLog:
    """Test enhanced audit log features."""

    def test_get_by_actor(self) -> None:
        audit = AuditLog()
        audit.append(AuditEventType.DECISION, actor="planner", action="plan")
        audit.append(AuditEventType.TOOL_CALL, actor="executor", action="run")
        assert len(audit.get_by_actor("planner")) == 1
        assert len(audit.get_by_actor("executor")) == 1

    def test_get_advisories(self) -> None:
        audit = AuditLog()
        audit.append(AuditEventType.ADVISORY, actor="safety", action="flagged")
        audit.append(AuditEventType.DECISION, actor="planner", action="plan")
        assert len(audit.get_advisories()) == 1

    def test_get_self_improvements(self) -> None:
        audit = AuditLog()
        audit.append(AuditEventType.SELF_IMPROVEMENT, actor="trainer", action="heuristic")
        assert len(audit.get_self_improvements()) == 1

    def test_get_recent(self) -> None:
        audit = AuditLog()
        for i in range(5):
            audit.append(AuditEventType.OTHER, actor=f"agent_{i}")
        assert len(audit.get_recent(limit=3)) == 3
        assert audit.total_entries == 5