File size: 6,987 Bytes
81e3673 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 | """
Test ReflectionEngine pattern detection.
Tests cover:
- Event listening
- Repeated failure pattern detection
- Triggering MementoEngine after threshold
- Pattern frequency tracking
- Resetting patterns after successful promotion
"""
import pytest
from unittest.mock import AsyncMock, MagicMock
from core.auto_dev.reflection_engine import ReflectionEngine
from core.auto_dev.event_hooks import TaskEvent
class TestReflectionEngineEventListening:
"""Test ReflectionEngine listens to events."""
def test_subscribes_to_task_fail_events(self, auto_dev_db_session):
"""Test subscribes to task_fail events."""
engine = ReflectionEngine(db=auto_dev_db_session)
engine.register()
# Should have registered handler
assert len(engine._failure_buffer) == 0
def test_tracks_event_metadata(self, auto_dev_db_session, sample_task_event):
"""Test tracks event metadata."""
engine = ReflectionEngine(db=auto_dev_db_session)
# Mock capability gate to allow processing
engine._should_process_agent = lambda agent_id, tenant_id: True
import asyncio
asyncio.run(engine.process_failure(sample_task_event))
assert len(engine._failure_buffer[sample_task_event.agent_id]) == 1
class TestReflectionEnginePatternDetection:
"""Test detects repeated failure patterns."""
@pytest.mark.asyncio
async def test_identifies_repeated_failure_patterns(self, auto_dev_db_session, sample_task_event):
"""Test identifies repeated failure patterns."""
from unittest.mock import AsyncMock
engine = ReflectionEngine(db=auto_dev_db_session, failure_threshold=2)
# Mock capability gate to allow processing
engine._should_process_agent = lambda agent_id, tenant_id: True
# Mock _trigger_memento to avoid database query
engine._trigger_memento = AsyncMock()
# Process first failure
await engine.process_failure(sample_task_event)
# Check buffer has 1 failure
assert len(engine._failure_buffer[sample_task_event.agent_id]) == 1
# Process second failure (should trigger)
await engine.process_failure(sample_task_event)
# Verify _trigger_memento was called (pattern detected)
engine._trigger_memento.assert_called_once()
# Note: Buffer is cleared after triggering, so we check the call happened
@pytest.mark.asyncio
async def test_groups_by_error_type(self, auto_dev_db_session):
"""Test groups by error type."""
from unittest.mock import AsyncMock
engine = ReflectionEngine(db=auto_dev_db_session, failure_threshold=3)
# Mock capability gate to allow processing
engine._should_process_agent = lambda agent_id, tenant_id: True
# Mock _trigger_memento to avoid database query
engine._trigger_memento = AsyncMock()
event1 = TaskEvent(
episode_id="ep-001",
agent_id="agent-001",
tenant_id="tenant-001",
task_description="Process sales data",
error_trace="ValueError: Invalid format",
outcome="failure",
)
event2 = TaskEvent(
episode_id="ep-002",
agent_id="agent-001",
tenant_id="tenant-001",
task_description="Process sales data again",
error_trace="ValueError: Invalid format",
outcome="failure",
)
await engine.process_failure(event1)
await engine.process_failure(event2)
# Should find similar failures
similar = engine._find_similar_failures("agent-001", "Process sales data")
assert len(similar) >= 1
class TestReflectionEngineTriggerThreshold:
"""Test triggers MementoEngine after threshold."""
@pytest.mark.asyncio
async def test_triggers_after_n_failures(self, auto_dev_db_session, sample_task_event, monkeypatch):
"""Test triggers MementoEngine after N failures."""
engine = ReflectionEngine(db=auto_dev_db_session, failure_threshold=2)
# Mock capability gate to allow processing
engine._should_process_agent = lambda agent_id, tenant_id: True
# Mock MementoEngine to avoid database requirements
mock_memento = MagicMock()
mock_candidate = MagicMock()
mock_candidate.skill_name = "test_skill"
mock_memento.generate_skill_candidate = AsyncMock(return_value=mock_candidate)
import sys
original_module = sys.modules.get("core.auto_dev.memento_engine")
class MockMementoEngine:
def __init__(self, db):
pass
sys.modules["core.auto_dev.memento_engine"] = MockMementoEngine
sys.modules["core.auto_dev.memento_engine"].MementoEngine = lambda db: mock_memento
try:
# Process two failures
await engine.process_failure(sample_task_event)
await engine.process_failure(sample_task_event)
# Should have triggered
mock_memento.generate_skill_candidate.assert_called_once()
finally:
if original_module:
sys.modules["core.auto_dev.memento_engine"] = original_module
else:
sys.modules.pop("core.auto_dev.memento_engine", None)
@pytest.mark.asyncio
async def test_prevents_duplicate_triggers(self, auto_dev_db_session, sample_task_event):
"""Test prevents duplicate triggers."""
engine = ReflectionEngine(db=auto_dev_db_session, failure_threshold=2)
# Process same event multiple times
for _ in range(5):
await engine.process_failure(sample_task_event)
# Buffer should be cleared after trigger
assert len(engine._failure_buffer[sample_task_event.agent_id]) < 5
class TestReflectionEnginePatternTracking:
"""Test pattern frequency tracking."""
@pytest.mark.asyncio
async def test_stores_pattern_metadata(self, auto_dev_db_session, sample_task_event):
"""Test stores pattern metadata."""
engine = ReflectionEngine(db=auto_dev_db_session)
# Mock capability gate to allow processing
engine._should_process_agent = lambda agent_id, tenant_id: True
await engine.process_failure(sample_task_event)
buffer = engine._failure_buffer[sample_task_event.agent_id]
assert len(buffer) == 1
assert buffer[0]["episode_id"] == sample_task_event.episode_id
assert buffer[0]["task_description"] == sample_task_event.task_description
class TestReflectionEngineIntegration:
"""Test ReflectionEngine integration."""
def test_reflection_engine_initialization(self, auto_dev_db_session):
"""Test ReflectionEngine initializes correctly."""
engine = ReflectionEngine(db=auto_dev_db_session)
assert engine.db == auto_dev_db_session
assert engine.failure_threshold == 2 # Default
assert len(engine._failure_buffer) == 0
|