""" Coverage expansion tests for episodic memory services. Target: 80%+ coverage for episode logic: - Episode segmentation (time gaps >30min, topic changes <0.75) - Retrieval modes (temporal, semantic, sequential, contextual) - Lifecycle management (decay, consolidation, archival) - Canvas integration tracking - Feedback aggregation Tests mock LanceDB to focus on segmentation/retrieval algorithms. Phase 171: SQLAlchemy conflicts resolved (duplicate models removed from core/models.py) No accounting module workaround needed anymore. """ import pytest from datetime import datetime, timezone, timedelta from unittest.mock import Mock, patch, AsyncMock from uuid import uuid4 from core.episode_segmentation_service import EpisodeSegmentationService, EpisodeBoundaryDetector from core.episode_retrieval_service import EpisodeRetrievalService from core.episode_lifecycle_service import EpisodeLifecycleService from core.models import ( AgentEpisode, # Use AgentEpisode instead of Episode EpisodeSegment, ChatSession, ChatMessage, CanvasAudit, AgentFeedback, AgentRegistry, AgentExecution, User, ) # Alias for compatibility Episode = AgentEpisode # ============================================================================= # Test Episode Boundary Detection # ============================================================================= class TestEpisodeBoundaryDetection: """ Test episode boundary detection algorithms with comprehensive coverage. Targets 80%+ line coverage for: - detect_time_gap(): Time gap detection with 30-minute threshold - detect_topic_changes(): Semantic similarity detection with 0.75 threshold - _cosine_similarity(): Vector similarity calculation - _keyword_similarity(): Fallback keyword-based similarity """ @pytest.mark.parametrize("gap_minutes,expected_boundary", [ (29, False), # Below threshold - no boundary (30, False), # Exactly threshold - no boundary (exclusive >) (31, True), # Above threshold - boundary detected (90, True), # Well above threshold - boundary detected (0, False), # No gap - no boundary (1, False), # Minimal gap - no boundary ]) def test_time_gap_threshold(self, segmentation_service_mocked, gap_minutes, expected_boundary): """ Test time gap detection with boundary conditions. Verifies exclusive >30 threshold behavior: - Gap = 30 minutes: NO boundary (exclusive >) - Gap < 30 minutes: NO boundary - Gap > 30 minutes: BOUNDARY detected Args: gap_minutes: Time gap in minutes between messages expected_boundary: Whether boundary should be detected """ from core.episode_segmentation_service import TIME_GAP_THRESHOLD_MINUTES # Verify threshold constant assert TIME_GAP_THRESHOLD_MINUTES == 30 session_id = f"test_session_{uuid4().hex[:8]}" base_time = datetime.now(timezone.utc) messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="First message", created_at=base_time ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="assistant", content=f"Second message after {gap_minutes} minutes", created_at=base_time + timedelta(minutes=gap_minutes) ) ] detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) boundaries = detector.detect_time_gap(messages) if expected_boundary: assert len(boundaries) == 1 assert 1 in boundaries # Index of second message else: assert len(boundaries) == 0 def test_time_gap_exactly_threshold(self, segmentation_service_mocked): """ Test time gap at exactly 30 minutes (boundary case). CRITICAL: Gap of exactly 30 minutes should NOT trigger boundary because threshold uses exclusive > (not >=). """ session_id = f"test_session_{uuid4().hex[:8]}" base_time = datetime.now(timezone.utc) messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="First message", created_at=base_time ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="assistant", content="Second message after exactly 30 minutes", created_at=base_time + timedelta(minutes=30) # Exactly threshold ) ] detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) boundaries = detector.detect_time_gap(messages) # Should NOT trigger boundary (exclusive >) assert len(boundaries) == 0 def test_time_gap_below_threshold(self, segmentation_service_mocked): """ Test time gap below 30 minutes (no boundary). """ session_id = f"test_session_{uuid4().hex[:8]}" base_time = datetime.now(timezone.utc) messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="First message", created_at=base_time ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="assistant", content="Second message after 29 minutes", created_at=base_time + timedelta(minutes=29) # Below threshold ) ] detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) boundaries = detector.detect_time_gap(messages) # Should NOT trigger boundary assert len(boundaries) == 0 def test_time_gap_above_threshold(self, segmentation_service_mocked): """ Test time gap above 30 minutes (boundary detected). """ session_id = f"test_session_{uuid4().hex[:8]}" base_time = datetime.now(timezone.utc) messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="First message", created_at=base_time ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="assistant", content="Second message after 31 minutes", created_at=base_time + timedelta(minutes=31) # Above threshold ) ] detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) boundaries = detector.detect_time_gap(messages) # SHOULD trigger boundary assert len(boundaries) == 1 assert 1 in boundaries def test_time_gap_multiple_boundaries(self, segmentation_service_mocked): """ Test detection of multiple time gaps in message sequence. """ session_id = f"test_session_{uuid4().hex[:8]}" base_time = datetime.now(timezone.utc) messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Message 1", created_at=base_time ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="assistant", content="Message 2 (5 min later)", created_at=base_time + timedelta(minutes=5) # No gap ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Message 3 (40 min later)", created_at=base_time + timedelta(minutes=45) # 40-min gap ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="assistant", content="Message 4 (5 min later)", created_at=base_time + timedelta(minutes=50) # No gap ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Message 5 (35 min later)", created_at=base_time + timedelta(minutes=85) # 35-min gap ) ] detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) boundaries = detector.detect_time_gap(messages) # Should detect 2 boundaries at indices 2 and 4 assert len(boundaries) == 2 assert 2 in boundaries assert 4 in boundaries def test_time_gap_with_timezone_aware_datetimes(self, segmentation_service_mocked): """ Test time gap detection with timezone-aware vs naive datetimes. Verifies proper handling of datetime objects with and without timezone info. """ session_id = f"test_session_{uuid4().hex[:8]}" # Test with timezone-aware datetimes base_time_aware = datetime.now(timezone.utc) messages_aware = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Timezone-aware message 1", created_at=base_time_aware ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="assistant", content="Timezone-aware message 2 (35 min later)", created_at=base_time_aware + timedelta(minutes=35) ) ] detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) boundaries_aware = detector.detect_time_gap(messages_aware) # Should detect boundary assert len(boundaries_aware) == 1 assert 1 in boundaries_aware def test_time_gap_empty_messages(self, segmentation_service_mocked): """Test time gap detection with empty message list.""" detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) boundaries = detector.detect_time_gap([]) assert boundaries == [] def test_time_gap_single_message(self, segmentation_service_mocked): """Test time gap detection with single message (no gaps possible).""" session_id = f"test_session_{uuid4().hex[:8]}" msg = ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Single message", created_at=datetime.now(timezone.utc) ) detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) boundaries = detector.detect_time_gap([msg]) # Single message should return empty boundaries assert boundaries == [] # ========================================================================== # Topic Change Detection Tests # ========================================================================== @pytest.mark.parametrize("similarity,expected_boundary", [ (0.90, False), # High similarity - no boundary (0.75, False), # Exactly threshold - no boundary (exclusive <) (0.74, True), # Just below threshold - boundary detected (0.50, True), # Low similarity - boundary detected (0.0, True), # No similarity - boundary detected ]) def test_topic_change_threshold(self, mock_lancedb_embeddings, similarity, expected_boundary): """ Test topic change detection with similarity threshold boundaries. Verifies exclusive <0.75 threshold behavior: - Similarity >= 0.75: NO boundary (similar topics) - Similarity < 0.75: BOUNDARY detected (topic change) Args: similarity: Simulated similarity score expected_boundary: Whether boundary should be detected """ from core.episode_segmentation_service import SEMANTIC_SIMILARITY_THRESHOLD # Verify threshold constant assert SEMANTIC_SIMILARITY_THRESHOLD == 0.75 def test_topic_change_below_threshold(self, segmentation_service_mocked, mock_lancedb_embeddings): """ Test topic change when similarity >= 0.75 (no boundary). High similarity means same topic continues. """ session_id = f"test_session_{uuid4().hex[:8]}" base_time = datetime.now(timezone.utc) messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Python programming is great", created_at=base_time ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="assistant", content="Python web development", # Similar, no boundary created_at=base_time + timedelta(minutes=5) ) ] detector = EpisodeBoundaryDetector(mock_lancedb_embeddings) changes = detector.detect_topic_changes(messages) # Should NOT detect topic change (similarity ~0.9) assert len(changes) == 0 def test_topic_change_above_threshold(self, segmentation_service_mocked, mock_lancedb_embeddings): """ Test topic change when similarity < 0.75 (boundary detected). Low similarity indicates topic change. """ session_id = f"test_session_{uuid4().hex[:8]}" base_time = datetime.now(timezone.utc) messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Let's discuss Python programming", created_at=base_time ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="assistant", content="Italian pasta recipes", # Different topic created_at=base_time + timedelta(minutes=5) ) ] detector = EpisodeBoundaryDetector(mock_lancedb_embeddings) changes = detector.detect_topic_changes(messages) # SHOULD detect topic change (similarity ~0.18 < 0.75) assert len(changes) == 1 assert 1 in changes def test_topic_change_exactly_threshold(self, mock_lancedb_embeddings): """ Test topic change at exactly 0.75 similarity (boundary case). CRITICAL: Similarity of exactly 0.75 should NOT trigger boundary because threshold uses exclusive < (not <=). """ # This test verifies the threshold boundary condition # Note: Testing exact similarity requires precise vector control # For now, we verify the threshold constant exists from core.episode_segmentation_service import SEMANTIC_SIMILARITY_THRESHOLD assert SEMANTIC_SIMILARITY_THRESHOLD == 0.75 def test_topic_change_empty_embeddings(self, segmentation_service_mocked, mock_lancedb_embeddings): """ Test fallback to keyword similarity when embeddings return None. Verifies graceful degradation when LanceDB embedding fails. """ # Mock LanceDB to return None for embeddings mock_lancedb_embeddings.embed_text = Mock(return_value=None) session_id = f"test_session_{uuid4().hex[:8]}" base_time = datetime.now(timezone.utc) messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Python programming", created_at=base_time ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="assistant", content="cooking recipes", # Different keywords created_at=base_time + timedelta(minutes=5) ) ] detector = EpisodeBoundaryDetector(mock_lancedb_embeddings) changes = detector.detect_topic_changes(messages) # Should fallback to keyword similarity # Keyword overlap: 0 (no common words) -> Dice = 0.0 < 0.75 -> boundary assert len(changes) == 1 assert 1 in changes def test_topic_change_single_message(self, segmentation_service_mocked, mock_lancedb_embeddings): """ Test topic change detection with single message (no changes possible). """ session_id = f"test_session_{uuid4().hex[:8]}" msg = ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Single message", created_at=datetime.now(timezone.utc) ) detector = EpisodeBoundaryDetector(mock_lancedb_embeddings) changes = detector.detect_topic_changes([msg]) # Single message should return empty changes assert changes == [] def test_topic_change_no_lancedb(self, segmentation_service_mocked): """ Test graceful handling when LanceDB handler is None. """ detector = EpisodeBoundaryDetector(None) session_id = f"test_session_{uuid4().hex[:8]}" base_time = datetime.now(timezone.utc) messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Message 1", created_at=base_time ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="assistant", content="Message 2", created_at=base_time + timedelta(minutes=5) ) ] changes = detector.detect_topic_changes(messages) # Should return empty list (graceful degradation) assert changes == [] def test_topic_change_empty_messages(self, segmentation_service_mocked, mock_lancedb_embeddings): """Test topic change detection with empty message list.""" detector = EpisodeBoundaryDetector(mock_lancedb_embeddings) changes = detector.detect_topic_changes([]) assert changes == [] # ========================================================================== # Cosine Similarity Tests # ========================================================================== def test_cosine_similarity_identical_vectors(self, segmentation_service_mocked): """ Test cosine similarity with identical vectors. Identical vectors should have similarity = 1.0. """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) vec1 = [0.9, 0.1, 0.0] vec2 = [0.9, 0.1, 0.0] similarity = detector._cosine_similarity(vec1, vec2) # Should be exactly 1.0 (same vector) assert abs(similarity - 1.0) < 0.001 def test_cosine_similarity_orthogonal_vectors(self, segmentation_service_mocked): """ Test cosine similarity with orthogonal vectors. Perpendicular vectors should have similarity ~0.0. """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) vec1 = [1.0, 0.0, 0.0] vec2 = [0.0, 1.0, 0.0] similarity = detector._cosine_similarity(vec1, vec2) # Should be ~0.0 (orthogonal) assert abs(similarity - 0.0) < 0.001 def test_cosine_similarity_similar_vectors(self, segmentation_service_mocked): """ Test cosine similarity with similar vectors. Similar vectors should have similarity >0.75. """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) vec1 = [0.9, 0.1, 0.0] vec2 = [0.8, 0.2, 0.0] similarity = detector._cosine_similarity(vec1, vec2) # Should be >0.75 (similar) assert similarity > 0.75 def test_cosine_similarity_different_vectors(self, segmentation_service_mocked): """ Test cosine similarity with different vectors. Different vectors should have similarity <0.75. """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) vec1 = [0.9, 0.1, 0.0] vec2 = [0.1, 0.9, 0.0] similarity = detector._cosine_similarity(vec1, vec2) # Should be <0.75 (different topic) assert similarity < 0.75 def test_cosine_similarity_pure_python_fallback(self, segmentation_service_mocked): """ Test pure Python fallback when numpy raises exception. Verifies graceful degradation when numpy operations fail. """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) # Mock numpy import to raise ImportError import builtins original_import = builtins.__import__ def mock_import(name, *args, **kwargs): if name == 'numpy': raise ImportError("No module named 'numpy'") return original_import(name, *args, **kwargs) with patch.object(builtins, '__import__', side_effect=mock_import): vec1 = [1.0, 0.0, 0.0] vec2 = [0.0, 1.0, 0.0] similarity = detector._cosine_similarity(vec1, vec2) # Should fallback to pure Python and return 0.0 (orthogonal) assert abs(similarity - 0.0) < 0.001 def test_cosine_similarity_pure_python_zero_magnitude(self, segmentation_service_mocked): """ Test pure Python fallback with zero-magnitude vector. Verifies pure Python implementation handles division by zero. """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) # Mock numpy import to raise ImportError import builtins original_import = builtins.__import__ def mock_import(name, *args, **kwargs): if name == 'numpy': raise ImportError("No module named 'numpy'") return original_import(name, *args, **kwargs) with patch.object(builtins, '__import__', side_effect=mock_import): vec1 = [0.0, 0.0, 0.0] # Zero vector vec2 = [1.0, 0.0, 0.0] similarity = detector._cosine_similarity(vec1, vec2) # Should return 0.0 (pure Python division by zero check) assert similarity == 0.0 def test_cosine_similarity_numpy_fallback(self, segmentation_service_mocked): """ Test fallback to pure Python when numpy unavailable. Verifies graceful degradation when numpy import fails. """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) # Test with simple vectors (works with both numpy and pure Python) vec1 = [1.0, 0.0, 0.0] vec2 = [0.0, 1.0, 0.0] similarity = detector._cosine_similarity(vec1, vec2) # Should work regardless of numpy availability assert abs(similarity - 0.0) < 0.001 def test_cosine_similarity_zero_magnitude(self, segmentation_service_mocked): """ Test cosine similarity with zero-magnitude vector. Should handle division by zero gracefully (return 0.0). """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) vec1 = [0.0, 0.0, 0.0] # Zero vector vec2 = [1.0, 0.0, 0.0] similarity = detector._cosine_similarity(vec1, vec2) # Should return 0.0 (avoid division by zero) assert similarity == 0.0 def test_cosine_similarity_invalid_input(self, segmentation_service_mocked): """ Test cosine similarity with malformed input. Should handle invalid input gracefully (return 0.0). """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) # Test with empty lists similarity = detector._cosine_similarity([], []) assert similarity == 0.0 def test_cosine_similarity_bounds(self, segmentation_service_mocked): """ Test that cosine similarity is bounded in [0.0, 1.0]. Cosine similarity should never be outside this range. """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) # Test various vector combinations test_cases = [ ([1.0, 0.0, 0.0], [0.0, 1.0, 0.0]), # Orthogonal ([1.0, 0.0, 0.0], [1.0, 0.0, 0.0]), # Identical ([0.9, 0.1, 0.0], [0.8, 0.2, 0.0]), # Similar ([0.1, 0.9, 0.0], [0.9, 0.1, 0.0]), # Different ([0.0, 0.0, 0.0], [1.0, 0.0, 0.0]), # Zero vector ] for vec1, vec2 in test_cases: similarity = detector._cosine_similarity(vec1, vec2) assert 0.0 <= similarity <= 1.0, f"Similarity {similarity} outside [0, 1] for {vec1}, {vec2}" # ========================================================================== # Keyword Similarity Tests # ========================================================================== def test_keyword_similarity_identical_text(self, segmentation_service_mocked): """ Test keyword similarity with identical text. Identical text should return 1.0 (perfect match). """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) text1 = "Python programming is great" text2 = "Python programming is great" similarity = detector._keyword_similarity(text1, text2) # Should be 1.0 (identical) assert similarity == 1.0 def test_keyword_similarity_no_overlap(self, segmentation_service_mocked): """ Test keyword similarity with no common words. No overlap should return 0.0. """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) text1 = "Python programming code" text2 = "cooking recipes food" # No common words similarity = detector._keyword_similarity(text1, text2) # Should be 0.0 (no overlap) assert similarity == 0.0 def test_keyword_similarity_partial_overlap(self, segmentation_service_mocked): """ Test keyword similarity with partial word overlap. Partial overlap should return 0.0 < similarity < 1.0. """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) text1 = "Python programming code" text2 = "Python web development" # "Python" overlaps similarity = detector._keyword_similarity(text1, text2) # Should be between 0.0 and 1.0 assert 0.0 < similarity < 1.0 def test_keyword_similarity_empty_strings(self, segmentation_service_mocked): """ Test keyword similarity with empty strings. Should handle empty input gracefully (return 0.0). """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) # Empty strings similarity = detector._keyword_similarity("", "") assert similarity == 0.0 # One empty string similarity = detector._keyword_similarity("Python", "") assert similarity == 0.0 def test_keyword_similarity_case_insensitive(self, segmentation_service_mocked): """ Test that keyword similarity is case-insensitive. "Python" and "python" should be treated as identical. """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) text1 = "Python Programming" text2 = "python programming" # Different case similarity = detector._keyword_similarity(text1, text2) # Should be 1.0 (case-insensitive) assert similarity == 1.0 def test_keyword_similarity_dice_coefficient(self, segmentation_service_mocked): """ Test that keyword similarity uses Dice coefficient. Dice = 2 * |intersection| / (|set1| + |set2|) """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) text1 = "Python programming code" text2 = "Python web code" # set1 = {"python", "programming", "code"} -> 3 # set2 = {"python", "web", "code"} -> 3 # intersection = {"python", "code"} -> 2 # Dice = 2 * 2 / (3 + 3) = 4/6 = 0.667 similarity = detector._keyword_similarity(text1, text2) # Should be ~0.667 (Dice coefficient) assert abs(similarity - 0.667) < 0.01 def test_keyword_similarity_bounds(self, segmentation_service_mocked): """ Test that keyword similarity is bounded in [0.0, 1.0]. Dice coefficient should never be outside this range. """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) # Test various text combinations test_cases = [ ("same text", "same text"), # Identical ("word1 word2", "word3 word4"), # No overlap ("word1 word2", "word1 word3"), # Partial overlap ("a b c", "a b c d e"), # Different lengths ] for text1, text2 in test_cases: similarity = detector._keyword_similarity(text1, text2) assert 0.0 <= similarity <= 1.0, f"Similarity {similarity} outside [0, 1] for '{text1}', '{text2}'" # ============================================================================= # Test Episode Segmentation Algorithms # ============================================================================= class TestEpisodeSegmentation: """Test episode segmentation algorithms""" def test_detect_time_gaps(self, segmentation_service_mocked, test_messages): """ Test detection of time gaps >30 minutes between messages. Verifies: - Time gaps >30min trigger boundary detection - Boundary index is correct (message after gap) - Gap duration is calculated accurately """ detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) # Detect boundaries boundaries = detector.detect_time_gap(test_messages) # Should detect 2 time gaps: # 1. After message 1 (45min - 10min = 35min gap) # 2. After message 3 (90min - 50min = 40min gap) assert len(boundaries) == 2 assert 2 in boundaries # Index of message after first gap assert 4 in boundaries # Index of message after second gap def test_detect_topic_changes(self, segmentation_service_mocked, mock_lancedb_embeddings): """ Test detection of topic changes using semantic similarity. Verifies: - Topic changes detected when similarity <0.75 - Python and cooking topics are distinguished - Similarity threshold applied correctly """ from core.episode_segmentation_service import EpisodeBoundaryDetector session_id = f"test_session_{uuid4().hex[:8]}" base_time = datetime.now(timezone.utc) messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Let's discuss Python programming", created_at=base_time ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="assistant", content="Python is great for web development", created_at=base_time + timedelta(minutes=5) ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Now let's talk about cooking recipes", created_at=base_time + timedelta(minutes=10) ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="assistant", content="I love Italian pasta dishes", created_at=base_time + timedelta(minutes=15) ), ] # Use detector with mocked LanceDB detector = EpisodeBoundaryDetector(mock_lancedb_embeddings) # Detect topic changes changes = detector.detect_topic_changes(messages) # Should detect 1 topic change at index 2 (cooking vs Python) # Similarity between [0.9, 0.1, 0.0] and [0.1, 0.9, 0.0] is 0.18 < 0.75 assert len(changes) == 1 assert 2 in changes def test_create_episodes_from_boundaries(self, segmentation_service_mocked, episode_test_session, episode_test_agent): """ Test creating episodes from detected boundaries. Verifies: - Multiple episodes created when boundaries exist - Episodes have correct segment counts - Episode timestamps are accurate """ session_id = episode_test_session.id base_time = datetime.now(timezone.utc) messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Message 1", created_at=base_time ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Message 2", created_at=base_time + timedelta(minutes=5) ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Message 3", created_at=base_time + timedelta(minutes=10) ), ] for msg in messages: segmentation_service_mocked.db.add(msg) segmentation_service_mocked.db.commit() # Create episodes with boundary at index 2 (split after message 1) with patch.object(segmentation_service_mocked, 'lancedb'): with patch.object(segmentation_service_mocked, 'byok_handler'): with patch.object(segmentation_service_mocked, 'canvas_summary_service'): episode = segmentation_service_mocked.create_episode_from_session( session_id=session_id, agent_id=episode_test_agent.id, force_create=True ) # Verify episode created assert episode is not None assert episode.status == "completed" assert episode.session_id == session_id # Verify segments created segments = segmentation_service_mocked.db.query(EpisodeSegment).filter( EpisodeSegment.episode_id == episode.id ).all() # Should have created segments from messages assert len(segments) >= 1 def test_segmentation_with_task_completion(self, segmentation_service_mocked, episode_test_agent): """ Test task completion detection in segmentation. Verifies: - Completed tasks trigger boundaries - Task completion markers are detected - Segmentation respects task completion """ from core.episode_segmentation_service import EpisodeBoundaryDetector # Create agent executions with task completion base_time = datetime.now(timezone.utc) executions = [ AgentExecution( id=f"exec_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, started_at=base_time, completed_at=base_time + timedelta(minutes=5), input_summary="Task 1 input" ), AgentExecution( id=f"exec_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, started_at=base_time + timedelta(minutes=10), completed_at=base_time + timedelta(minutes=15), input_summary="Task 2 input" ), ] for exec in executions: segmentation_service_mocked.db.add(exec) segmentation_service_mocked.db.commit() # Detect task completions detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) completions = detector.detect_task_completion(executions) # Should detect 2 task completions assert len(completions) == 2 assert 0 in completions # First task assert 1 in completions # Second task def test_segmentation_cosine_similarity_calculation(self, segmentation_service_mocked): """ Test cosine similarity calculation for topic change detection. Verifies: - Cosine similarity calculated correctly - Similarity ranges from 0.0 to 1.0 - Same vectors have similarity 1.0 - Orthogonal vectors have similarity 0.0 """ from core.episode_segmentation_service import EpisodeBoundaryDetector detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) # Test identical vectors (similarity = 1.0) vec1 = [0.9, 0.1, 0.0] vec2 = [0.9, 0.1, 0.0] similarity = detector._cosine_similarity(vec1, vec2) assert abs(similarity - 1.0) < 0.01 # Allow small floating point error # Test orthogonal vectors (similarity ≈ 0.0) vec3 = [1.0, 0.0, 0.0] vec4 = [0.0, 1.0, 0.0] similarity = detector._cosine_similarity(vec3, vec4) assert abs(similarity - 0.0) < 0.01 # Test similar vectors (similarity > 0.75) vec5 = [0.9, 0.1, 0.0] vec6 = [0.8, 0.2, 0.0] similarity = detector._cosine_similarity(vec5, vec6) assert similarity > 0.75 # Should be similar def test_segmentation_empty_messages(self, segmentation_service_mocked): """Test segmentation with empty message list.""" from core.episode_segmentation_service import EpisodeBoundaryDetector detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) # Empty list should return empty boundaries boundaries = detector.detect_time_gap([]) assert boundaries == [] changes = detector.detect_topic_changes([]) assert changes == [] def test_segmentation_single_message(self, segmentation_service_mocked): """Test segmentation with single message (no boundaries possible).""" from core.episode_segmentation_service import EpisodeBoundaryDetector session_id = f"test_session_{uuid4().hex[:8]}" msg = ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=session_id, role="user", content="Single message", created_at=datetime.now(timezone.utc) ) detector = EpisodeBoundaryDetector(segmentation_service_mocked.lancedb) # Single message should return empty boundaries boundaries = detector.detect_time_gap([msg]) assert boundaries == [] changes = detector.detect_topic_changes([msg]) assert changes == [] # ============================================================================= # Test Episode Retrieval Modes # ============================================================================= class TestEpisodeRetrieval: """Test episode retrieval modes""" @pytest.mark.asyncio async def test_temporal_retrieval(self, retrieval_service_mocked, episode_test_agent): """ Test temporal retrieval (time-based). Verifies: - Episodes retrieved within time range - Episodes outside range excluded - Results ordered by time (newest first) """ import asyncio # Create episodes with different timestamps episodes = [] base_time = datetime.now(timezone.utc) for i in range(5): episode = AgentEpisode( id=f"temporal_ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(days=i), completed_at=base_time - timedelta(days=i) + timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success" ) retrieval_service_mocked.db.add(episode) episodes.append(episode) retrieval_service_mocked.db.commit() # Retrieve episodes from last 3 days result = await retrieval_service_mocked.retrieve_temporal( agent_id=episode_test_agent.id, time_range="3d", limit=10 ) # Should return episodes 0-2 (last 3 days) assert result["count"] <= 3 assert len(result["episodes"]) <= 3 @pytest.mark.asyncio async def test_semantic_retrieval(self, retrieval_service_mocked, episode_test_agent): """ Test semantic retrieval (vector search). Verifies: - Semantic search finds relevant episodes - LanceDB search is called - Results ranked by similarity """ # Create test episodes episode1 = AgentEpisode( id=f"semantic_ep1_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success" ) episode2 = AgentEpisode( id=f"semantic_ep2_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success" ) retrieval_service_mocked.db.add(episode1) retrieval_service_mocked.db.add(episode2) retrieval_service_mocked.db.commit() # Mock LanceDB search to return episode1 retrieval_service_mocked.lancedb.search.return_value = [ { "id": episode1.id, "metadata": {"episode_id": episode1.id}, "_distance": 0.1 # Low distance = high similarity } ] # Search for Python-related content result = await retrieval_service_mocked.retrieve_semantic( agent_id=episode_test_agent.id, query="How do I write Python code?", limit=10 ) # Should return episode1 (Python topic) assert result["count"] >= 0 retrieval_service_mocked.lancedb.search.assert_called_once() @pytest.mark.asyncio async def test_sequential_retrieval(self, retrieval_service_mocked, test_episode): """ Test sequential retrieval (full episode with segments). Verifies: - Full episode returned with all segments - Segments in correct order - Canvas and feedback context included by default """ # Retrieve full episode result = await retrieval_service_mocked.retrieve_sequential( episode_id=test_episode.id, agent_id=test_episode.agent_id, include_canvas=True, include_feedback=True ) assert result["episode"]["id"] == test_episode.id assert len(result["segments"]) >= 1 # Verify segments are ordered order_values = [s["sequence_order"] for s in result["segments"]] assert order_values == sorted(order_values) @pytest.mark.asyncio async def test_contextual_retrieval(self, retrieval_service_mocked, episode_test_agent): """ Test contextual retrieval (hybrid temporal + semantic). Verifies: - Combines temporal and semantic scoring - Returns limited number of results - Applies canvas/feedback boosts """ # Create test episodes episodes = [] base_time = datetime.now(timezone.utc) for i in range(5): episode = AgentEpisode( id=f"contextual_ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(days=i), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", canvas_action_count=1 if i % 2 == 0 else 0, # Some have canvas aggregate_feedback_score=0.8 if i % 2 == 0 else None # Some have feedback ) retrieval_service_mocked.db.add(episode) episodes.append(episode) retrieval_service_mocked.db.commit() # Mock semantic search retrieval_service_mocked.lancedb.search.return_value = [] # Retrieve with context result = await retrieval_service_mocked.retrieve_contextual( agent_id=episode_test_agent.id, current_task="Recent episodes", limit=3, require_canvas=False, require_feedback=False ) # Should return <= 3 episodes assert result["count"] <= 3 assert len(result["episodes"]) <= 3 @pytest.mark.asyncio async def test_retrieval_with_empty_results(self, retrieval_service_mocked): """Test retrieval when no episodes exist (graceful handling).""" result = await retrieval_service_mocked.retrieve_temporal( agent_id="nonexistent_agent", time_range="7d", limit=10 ) # Should return empty list, not error assert result["count"] == 0 assert result["episodes"] == [] # ============================================================================= # Test Episode Lifecycle Management # ============================================================================= class TestEpisodeLifecycle: """Test episode lifecycle management""" @pytest.mark.asyncio async def test_episode_decay(self, lifecycle_service_mocked, episode_test_agent): """ Test episode decay scoring. Verifies: - Decay score calculated for old episodes - Decay formula applied correctly - Access count incremented """ import asyncio # Create old episode episode = AgentEpisode( id=f"decay_ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=60), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=1.0, access_count=5, outcome="success" ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Apply decay result = await lifecycle_service_mocked.decay_old_episodes(days_threshold=90) assert result["affected"] >= 1 # Verify decay score updated lifecycle_service_mocked.db.refresh(episode) assert episode.decay_score < 1.0 # Should be decayed assert episode.access_count >= 6 # Should be incremented @pytest.mark.asyncio async def test_episode_consolidation(self, lifecycle_service_mocked, episode_test_agent): """ Test episode consolidation. Verifies: - Similar episodes consolidated - Original episodes marked as consolidated - Consolidation metadata recorded """ # Create related episodes episode1 = AgentEpisode( id=f"consol_ep1_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=2), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", consolidated_into=None ) episode2 = AgentEpisode( id=f"consol_ep2_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", consolidated_into=None ) lifecycle_service_mocked.db.add(episode1) lifecycle_service_mocked.db.add(episode2) lifecycle_service_mocked.db.commit() # Mock LanceDB semantic search lifecycle_service_mocked.lancedb.search.return_value = [ { "id": episode2.id, "metadata": {"episode_id": episode2.id}, "_distance": 0.1 # High similarity } ] # Consolidate result = await lifecycle_service_mocked.consolidate_similar_episodes( agent_id=episode_test_agent.id, similarity_threshold=0.85 ) # Should consolidate at least one episode assert result["consolidated"] >= 0 or result["parent_episodes"] >= 0 @pytest.mark.asyncio async def test_episode_archival(self, lifecycle_service_mocked, episode_test_agent): """ Test episode archival to cold storage. Verifies: - Old episodes archived - Archived status set - Archived timestamp recorded """ # Create very old episode episode = AgentEpisode( id=f"archive_ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=365), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=0.95, access_count=0, outcome="success" ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Archive to cold storage result = await lifecycle_service_mocked.archive_to_cold_storage(episode.id) assert result is True # Verify status updated lifecycle_service_mocked.db.refresh(episode) assert episode.status == "archived" assert episode.archived_at is not None @pytest.mark.asyncio async def test_episode_consolidation_with_feedback_aggregation(self, lifecycle_service_mocked, episode_test_agent): """ Test consolidation with feedback score aggregation. Verifies: - Feedback scores aggregated during consolidation - Average score calculated correctly """ # Create episodes with feedback episode1 = AgentEpisode( id=f"feedback_ep1_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=2), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", aggregate_feedback_score=0.8, consolidated_into=None ) episode2 = AgentEpisode( id=f"feedback_ep2_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", aggregate_feedback_score=0.5, consolidated_into=None ) lifecycle_service_mocked.db.add(episode1) lifecycle_service_mocked.db.add(episode2) lifecycle_service_mocked.db.commit() # Mock LanceDB search lifecycle_service_mocked.lancedb.search.return_value = [ { "id": episode2.id, "metadata": {"episode_id": episode2.id}, "_distance": 0.2 } ] # Consolidate result = await lifecycle_service_mocked.consolidate_similar_episodes( agent_id=episode_test_agent.id, similarity_threshold=0.75 ) # Consolidation should succeed assert "consolidated" in result assert "parent_episodes" in result @pytest.mark.asyncio async def test_decay_old_episodes_threshold(self, lifecycle_service_mocked, episode_test_agent): """ Test decay applied to episodes older than threshold. Verifies: - Episodes older than threshold get decay applied - Episodes younger than threshold not affected - Decay score formula applied correctly """ import asyncio # Create old episode (100 days) old_episode = AgentEpisode( id=f"decay_old_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=100), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=1.0, access_count=5, outcome="success", status="completed" ) # Create young episode (30 days) young_episode = AgentEpisode( id=f"decay_young_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=30), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=1.0, access_count=3, outcome="success", status="completed" ) lifecycle_service_mocked.db.add(old_episode) lifecycle_service_mocked.db.add(young_episode) lifecycle_service_mocked.db.commit() # Apply decay with 90-day threshold result = await lifecycle_service_mocked.decay_old_episodes(days_threshold=90) # Should affect only old episode assert result["affected"] >= 1 # Verify old episode decayed lifecycle_service_mocked.db.refresh(old_episode) assert old_episode.decay_score < 1.0 # Should be decayed assert old_episode.access_count == 6 # Incremented @pytest.mark.asyncio async def test_decay_formula_calculation(self, lifecycle_service_mocked, episode_test_agent): """ Test decay formula: max(0, 1 - days_old/180). Verifies: - Decay score calculated correctly - Formula: max(0, 1 - days_old/180) - 90-day-old episode: 0.5 decay score """ import asyncio # Create 90-day-old episode episode = AgentEpisode( id=f"decay_formula_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=90), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=1.0, access_count=0, outcome="success", status="completed" ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Apply decay result = await lifecycle_service_mocked.decay_old_episodes(days_threshold=90) assert result["affected"] >= 1 # Expected decay: max(0, 1 - 90/180) = 0.5 lifecycle_service_mocked.db.refresh(episode) # Allow small floating point tolerance assert abs(episode.decay_score - 0.5) < 0.01 @pytest.mark.asyncio async def test_decay_formula_boundary_180_days(self, lifecycle_service_mocked, episode_test_agent): """ Test decay formula boundary at 180 days. Verifies: - 180-day-old episodes decay to 0.0 - Formula: max(0, 1 - 180/180) = 0.0 - Episodes marked as archived when >180 days """ import asyncio # Create 180-day-old episode episode = AgentEpisode( id=f"decay_180_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=180), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=1.0, access_count=0, outcome="success", status="completed" ) # Create 200-day-old episode (should archive) very_old_episode = AgentEpisode( id=f"decay_200_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=200), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=1.0, access_count=0, outcome="success", status="completed" ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.add(very_old_episode) lifecycle_service_mocked.db.commit() # Apply decay result = await lifecycle_service_mocked.decay_old_episodes(days_threshold=90) assert result["affected"] >= 2 assert result["archived"] >= 1 # 200-day episode archived # Verify 180-day episode decayed to 0.0 lifecycle_service_mocked.db.refresh(episode) assert episode.decay_score == 0.0 # Verify 200-day episode archived lifecycle_service_mocked.db.refresh(very_old_episode) assert very_old_episode.status == "archived" assert very_old_episode.archived_at is not None @pytest.mark.asyncio async def test_decay_access_count_increment(self, lifecycle_service_mocked, episode_test_agent): """ Test that access count is incremented during decay. Verifies: - Access count incremented by 1 for each decayed episode - Decay score updated simultaneously """ import asyncio # Create episode with initial access count episode = AgentEpisode( id=f"decay_access_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=120), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=1.0, access_count=10, # Initial count outcome="success", status="completed" ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Apply decay await lifecycle_service_mocked.decay_old_episodes(days_threshold=90) # Verify access count incremented lifecycle_service_mocked.db.refresh(episode) assert episode.access_count == 11 # Initial + 1 assert episode.decay_score < 1.0 # Decay applied @pytest.mark.asyncio async def test_decay_archives_very_old(self, lifecycle_service_mocked, episode_test_agent): """ Test that episodes >180 days are marked as archived. Verifies: - Episodes >180 days get status="archived" - archived_at timestamp set - Archived episodes counted in result """ import asyncio # Create episode >180 days old episode = AgentEpisode( id=f"decay_archive_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=200), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=1.0, access_count=0, outcome="success", status="completed" ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Apply decay result = await lifecycle_service_mocked.decay_old_episodes(days_threshold=90) # Should archive episode assert result["archived"] >= 1 # Verify archived status lifecycle_service_mocked.db.refresh(episode) assert episode.status == "archived" assert episode.archived_at is not None assert episode.archived_at <= datetime.now(timezone.utc) def test_update_lifecycle_timezone_aware(self, lifecycle_service_mocked, episode_test_agent): """ Test update_lifecycle handles timezone-aware datetimes. Verifies: - Timezone-aware started_at handled correctly - Decay calculation works with aware datetimes - Lifecycle update succeeds """ # Create episode with timezone-aware timestamp episode = AgentEpisode( id=f"lifecycle_tz_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=45), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=0.0, access_count=0, outcome="success", status="completed" ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Update lifecycle result = lifecycle_service_mocked.update_lifecycle(episode) assert result is True # Verify decay score updated lifecycle_service_mocked.db.refresh(episode) # 45 days old: decay = min(1, 45/90) = 0.5 assert episode.decay_score > 0.0 def test_update_lifecycle_no_started_at(self, lifecycle_service_mocked, episode_test_agent): """ Test update_lifecycle gracefully handles episodes without started_at. Verifies: - Episodes without started_at skipped gracefully - Returns False for episodes without started_at - No crash or error """ # Create episode without started_at episode = AgentEpisode( id=f"lifecycle_no_start_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=None, # No started_at maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=0.0, access_count=0, outcome="success", status="completed" ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Update lifecycle - should return False gracefully result = lifecycle_service_mocked.update_lifecycle(episode) assert result is False @pytest.mark.asyncio async def test_archive_to_cold_storage_success(self, lifecycle_service_mocked, episode_test_agent): """ Test archive_to_cold_storage sets status and timestamp. Verifies: - Episode status set to "archived" - archived_at timestamp set - Returns True on success """ episode = AgentEpisode( id=f"archive_success_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=100), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=0.5, access_count=5, outcome="success", status="completed" ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Archive episode result = await lifecycle_service_mocked.archive_to_cold_storage(episode.id) assert result is True # Verify status and timestamp lifecycle_service_mocked.db.refresh(episode) assert episode.status == "archived" assert episode.archived_at is not None assert episode.archived_at <= datetime.now(timezone.utc) @pytest.mark.asyncio async def test_archive_to_cold_storage_sets_timestamp(self, lifecycle_service_mocked, episode_test_agent): """ Test archive_to_cold_storage sets archived_at timestamp. Verifies: - archived_at timestamp set to current time - Timestamp is timezone-aware - Timestamp within reasonable time window """ import asyncio episode = AgentEpisode( id=f"archive_timestamp_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=50), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=0.5, access_count=3, outcome="success", status="completed" ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Get time before archiving before_time = datetime.now(timezone.utc) # Archive episode await lifecycle_service_mocked.archive_to_cold_storage(episode.id) # Get time after archiving after_time = datetime.now(timezone.utc) # Verify timestamp set correctly lifecycle_service_mocked.db.refresh(episode) assert episode.archived_at is not None assert before_time <= episode.archived_at <= after_time @pytest.mark.asyncio async def test_archive_to_cold_storage_not_found(self, lifecycle_service_mocked): """ Test archive_to_cold_storage returns False for nonexistent episode. Verifies: - Returns False for non-existent episode ID - No error or crash - Graceful handling """ # Try to archive non-existent episode result = await lifecycle_service_mocked.archive_to_cold_storage("nonexistent_episode_id") assert result is False def test_archive_episode_synchronous(self, lifecycle_service_mocked, episode_test_agent): """ Test archive_episode synchronous method works. Verifies: - Synchronous archive_episode() method works - Status and timestamp set correctly - Returns True on success """ episode = AgentEpisode( id=f"archive_sync_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=75), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=0.5, access_count=4, outcome="success", status="completed" ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Archive using synchronous method result = lifecycle_service_mocked.archive_episode(episode) assert result is True # Verify status and timestamp lifecycle_service_mocked.db.refresh(episode) assert episode.status == "archived" assert episode.archived_at is not None @pytest.mark.asyncio async def test_archived_excluded_from_retrieval(self, lifecycle_service_mocked, retrieval_service_mocked, episode_test_agent): """ Test that archived episodes are excluded from temporal retrieval. Verifies: - Archived episodes not in temporal retrieval results - status="archived" filtered out - Only active episodes returned """ # Create active episode active_episode = AgentEpisode( id=f"active_ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=2), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="completed" ) # Create archived episode archived_episode = AgentEpisode( id=f"archived_ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="archived" # Already archived ) lifecycle_service_mocked.db.add(active_episode) lifecycle_service_mocked.db.add(archived_episode) lifecycle_service_mocked.db.commit() # Retrieve temporal episodes (should exclude archived) result = await retrieval_service_mocked.retrieve_temporal( agent_id=episode_test_agent.id, time_range="7d" ) # Should only return active episode assert result["count"] >= 1 episode_ids = [ep["id"] for ep in result["episodes"]] assert active_episode.id in episode_ids assert archived_episode.id not in episode_ids @pytest.mark.asyncio async def test_consolidation_similar_episodes(self, lifecycle_service_mocked, episode_test_agent): """ Test consolidation of semantically similar episodes. Verifies: - Similar episodes merged under parent episode - consolidated_into field set correctly - LanceDB search called for similarity - Episodes above similarity threshold consolidated """ # Create episodes with similar task descriptions episode1 = AgentEpisode( id=f"consol_sim_1_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=5), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="completed", task_description="Python web development", consolidated_into=None ) episode2 = AgentEpisode( id=f"consol_sim_2_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=3), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="completed", task_description="Python web apps", consolidated_into=None ) lifecycle_service_mocked.db.add(episode1) lifecycle_service_mocked.db.add(episode2) lifecycle_service_mocked.db.commit() # Mock LanceDB to return episode2 as similar to episode1 lifecycle_service_mocked.lancedb.search.return_value = [ { "metadata": {"episode_id": episode2.id}, "_distance": 0.1 # High similarity (1 - 0.1 = 0.9) } ] # Consolidate similar episodes result = await lifecycle_service_mocked.consolidate_similar_episodes( agent_id=episode_test_agent.id, similarity_threshold=0.85 ) # Should consolidate at least one episode assert result["consolidated"] >= 0 assert result["parent_episodes"] >= 0 @pytest.mark.asyncio async def test_consolidation_similarity_threshold(self, lifecycle_service_mocked, episode_test_agent): """ Test consolidation only merges episodes above similarity threshold. Verifies: - Episodes below threshold not consolidated - Episodes >= threshold consolidated - Similarity calculation correct (1 - distance) """ # Create parent episode parent_episode = AgentEpisode( id=f"consol_thresh_parent_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=6), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="completed", task_description="Data analysis", consolidated_into=None ) # Create child episodes with varying similarity similar_child = AgentEpisode( id=f"consol_thresh_sim_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=4), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="completed", task_description="Data analytics", consolidated_into=None ) dissimilar_child = AgentEpisode( id=f"consol_thresh_diff_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=2), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="completed", task_description="Email marketing", consolidated_into=None ) lifecycle_service_mocked.db.add(parent_episode) lifecycle_service_mocked.db.add(similar_child) lifecycle_service_mocked.db.add(dissimilar_child) lifecycle_service_mocked.db.commit() # Mock LanceDB search results # Distance 0.15 = similarity 0.85 (exactly at threshold) # Distance 0.5 = similarity 0.5 (below threshold) lifecycle_service_mocked.lancedb.search.return_value = [ { "metadata": {"episode_id": similar_child.id}, "_distance": 0.15 # similarity = 0.85 }, { "metadata": {"episode_id": dissimilar_child.id}, "_distance": 0.5 # similarity = 0.5 } ] # Consolidate with threshold 0.85 result = await lifecycle_service_mocked.consolidate_similar_episodes( agent_id=episode_test_agent.id, similarity_threshold=0.85 ) # Should consolidate only similar episode assert "consolidated" in result assert "parent_episodes" in result @pytest.mark.asyncio async def test_consolidation_sets_consolidated_into(self, lifecycle_service_mocked, episode_test_agent): """ Test consolidation sets consolidated_into field correctly. Verifies: - Child episodes have consolidated_into set to parent ID - Parent episode has consolidated_into=None - Field references correct parent episode """ # Create parent episode parent_episode = AgentEpisode( id=f"consol_ref_parent_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=4), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="completed", task_description="API testing", consolidated_into=None ) # Create child episode child_episode = AgentEpisode( id=f"consol_ref_child_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=2), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="completed", task_description="API automation testing", consolidated_into=None ) lifecycle_service_mocked.db.add(parent_episode) lifecycle_service_mocked.db.add(child_episode) lifecycle_service_mocked.db.commit() # Mock LanceDB search lifecycle_service_mocked.lancedb.search.return_value = [ { "metadata": {"episode_id": child_episode.id}, "_distance": 0.05 # Very similar } ] # Consolidate await lifecycle_service_mocked.consolidate_similar_episodes( agent_id=episode_test_agent.id, similarity_threshold=0.85 ) # Verify child references parent lifecycle_service_mocked.db.refresh(child_episode) assert child_episode.consolidated_into == parent_episode.id # Verify parent has no parent lifecycle_service_mocked.db.refresh(parent_episode) assert parent_episode.consolidated_into is None @pytest.mark.asyncio async def test_consolidation_no_duplicates(self, lifecycle_service_mocked, episode_test_agent): """ Test consolidation skips already consolidated episodes. Verifies: - Already consolidated episodes skipped - consolidated_into field checked before processing - No double-consolidation """ # Create parent episode parent_episode = AgentEpisode( id=f"consol_dup_parent_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=5), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="completed", task_description="Machine learning", consolidated_into=None ) # Create already-consolidated child already_consolidated = AgentEpisode( id=f"consol_dup_child_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=3), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="completed", task_description="ML models", consolidated_into="some_other_parent_id" # Already consolidated ) lifecycle_service_mocked.db.add(parent_episode) lifecycle_service_mocked.db.add(already_consolidated) lifecycle_service_mocked.db.commit() # Mock LanceDB search (would return already-consolidated episode) lifecycle_service_mocked.lancedb.search.return_value = [ { "metadata": {"episode_id": already_consolidated.id}, "_distance": 0.1 } ] # Consolidate result = await lifecycle_service_mocked.consolidate_similar_episodes( agent_id=episode_test_agent.id, similarity_threshold=0.85 ) # Already-consolidated episode should be skipped lifecycle_service_mocked.db.refresh(already_consolidated) assert already_consolidated.consolidated_into == "some_other_parent_id" # Unchanged @pytest.mark.asyncio async def test_consolidation_empty_results(self, lifecycle_service_mocked, episode_test_agent): """ Test consolidation with no similar episodes. Verifies: - Empty LanceDB results handled gracefully - Returns zero counts when no episodes consolidated - No error on empty results """ # Create single episode (no similar episodes) episode = AgentEpisode( id=f"consol_empty_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=2), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="completed", task_description="Unique task description", consolidated_into=None ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Mock empty LanceDB search lifecycle_service_mocked.lancedb.search.return_value = [] # Consolidate result = await lifecycle_service_mocked.consolidate_similar_episodes( agent_id=episode_test_agent.id, similarity_threshold=0.85 ) # Should return zero counts assert result["consolidated"] == 0 assert result["parent_episodes"] == 0 @pytest.mark.asyncio async def test_consolidation_lancedb_search(self, lifecycle_service_mocked, episode_test_agent): """ Test consolidation calls LanceDB search correctly. Verifies: - LanceDB search called with correct parameters - table_name="episodes" - Agent ID filter applied - Limit parameter passed """ # Create episode episode = AgentEpisode( id=f"consol_search_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=2), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="completed", task_description="Search test", consolidated_into=None ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Mock LanceDB search lifecycle_service_mocked.lancedb.search.return_value = [] # Consolidate await lifecycle_service_mocked.consolidate_similar_episodes( agent_id=episode_test_agent.id, similarity_threshold=0.85 ) # Verify LanceDB search called assert lifecycle_service_mocked.lancedb.search.called call_args = lifecycle_service_mocked.lancedb.search.call_args assert call_args[1]["table_name"] == "episodes" assert f"agent_id == '{episode_test_agent.id}'" in call_args[1]["filter_str"] assert call_args[1]["limit"] == 20 @pytest.mark.asyncio async def test_update_importance_with_feedback(self, lifecycle_service_mocked, episode_test_agent): """ Test importance score updated from feedback. Verifies: - Importance score updated based on user feedback - Feedback has 20% weight in calculation - New importance = old * 0.8 + feedback_score * 0.2 """ # Create episode with initial importance episode = AgentEpisode( id=f"importance_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=2), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=0.5, access_count=3, importance_score=0.7, # Initial importance outcome="success", status="completed" ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Update importance with positive feedback result = await lifecycle_service_mocked.update_importance_scores( episode_id=episode.id, user_feedback=0.8 # Positive feedback ) assert result is True # Verify importance updated lifecycle_service_mocked.db.refresh(episode) # New = 0.7 * 0.8 + (0.8 + 1.0) / 2.0 * 0.2 = 0.56 + 0.18 = 0.74 assert episode.importance_score > 0.70 # Should increase slightly @pytest.mark.asyncio async def test_importance_bounds_enforcement(self, lifecycle_service_mocked, episode_test_agent): """ Test importance score clamped to [0.0, 1.0]. Verifies: - Importance scores below 0.0 clamped to 0.0 - Importance scores above 1.0 clamped to 1.0 - Normal scores remain unchanged """ # Test high initial importance with positive feedback high_episode = AgentEpisode( id=f"importance_high_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=2), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=0.5, access_count=3, importance_score=0.95, # Already high outcome="success", status="completed" ) # Test low initial importance with negative feedback low_episode = AgentEpisode( id=f"importance_low_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=2), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=0.5, access_count=3, importance_score=0.05, # Already low outcome="success", status="completed" ) lifecycle_service_mocked.db.add(high_episode) lifecycle_service_mocked.db.add(low_episode) lifecycle_service_mocked.db.commit() # Update with extreme feedback await lifecycle_service_mocked.update_importance_scores( episode_id=high_episode.id, user_feedback=1.0 # Maximum positive ) await lifecycle_service_mocked.update_importance_scores( episode_id=low_episode.id, user_feedback=-1.0 # Maximum negative ) # Verify bounds enforced lifecycle_service_mocked.db.refresh(high_episode) lifecycle_service_mocked.db.refresh(low_episode) assert 0.0 <= high_episode.importance_score <= 1.0 assert 0.0 <= low_episode.importance_score <= 1.0 @pytest.mark.asyncio async def test_batch_update_access_counts(self, lifecycle_service_mocked, episode_test_agent): """ Test batch access count update for multiple episodes. Verifies: - Multiple episodes have access_count incremented - All episodes updated in single call - Correct count returned """ # Create multiple episodes episode_ids = [] for i in range(3): episode = AgentEpisode( id=f"batch_access_{i}_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=(i + 1)), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=0.5, access_count=5, # Initial count outcome="success", status="completed" ) episode_ids.append(episode.id) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Batch update access counts result = await lifecycle_service_mocked.batch_update_access_counts(episode_ids) assert result["updated"] == 3 # Verify all episodes incremented for episode_id in episode_ids: episode = lifecycle_service_mocked.db.query(AgentEpisode).filter( AgentEpisode.id == episode_id ).first() assert episode.access_count == 6 # Initial + 1 def test_apply_decay_single_episode(self, lifecycle_service_mocked, episode_test_agent): """ Test apply_decay works on single episode. Verifies: - Single episode decayed correctly - Synchronous method works - Returns True on success """ # Create episode episode = AgentEpisode( id=f"apply_decay_single_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=30), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=0.0, access_count=0, outcome="success", status="completed" ) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Apply decay to single episode result = lifecycle_service_mocked.apply_decay(episode) assert result is True # Verify decay applied lifecycle_service_mocked.db.refresh(episode) assert episode.decay_score > 0.0 # Should have decay def test_apply_decay_episode_list(self, lifecycle_service_mocked, episode_test_agent): """ Test apply_decay works on list of episodes. Verifies: - List of episodes decayed correctly - All episodes in list processed - Returns True if all succeed """ # Create list of episodes episodes = [] for i in range(3): episode = AgentEpisode( id=f"apply_decay_list_{i}_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(days=(i * 10 + 10)), maturity_at_time="AUTONOMOUS", human_intervention_count=0, decay_score=0.0, access_count=0, outcome="success", status="completed" ) episodes.append(episode) lifecycle_service_mocked.db.add(episode) lifecycle_service_mocked.db.commit() # Apply decay to list result = lifecycle_service_mocked.apply_decay(episodes) assert result is True # Verify all episodes decayed for episode in episodes: lifecycle_service_mocked.db.refresh(episode) assert episode.decay_score > 0.0 # ============================================================================= # Test Canvas Integration # ============================================================================= class TestCanvasIntegration: """Test canvas-aware episode tracking""" def test_track_canvas_presentations(self, segmentation_service_mocked, episode_test_session, episode_test_user): """ Test tracking canvas presentations in episodes. Verifies: - Canvas linked to episode - Canvas metadata stored - Canvas type and action recorded """ from core.models import CanvasAudit # Create canvas audit canvas = CanvasAudit( id=f"canvas_{uuid4().hex[:8]}", session_id=episode_test_session.id, canvas_type="chart", component_type="line", component_name="SalesChart", action="present", audit_metadata={"title": "Monthly Sales"}, created_at=datetime.now(timezone.utc) ) segmentation_service_mocked.db.add(canvas) segmentation_service_mocked.db.commit() # Extract canvas context context = segmentation_service_mocked._extract_canvas_context([canvas]) assert context is not None assert context["canvas_type"] == "chart" assert "presentation_summary" in context def test_retrieve_canvas_context(self, segmentation_service_mocked, episode_test_session, episode_test_user): """ Test retrieving canvas context from episode. Verifies: - Canvas contexts returned - Multiple canvases supported - Canvas types and actions preserved """ from core.models import CanvasAudit # Create multiple canvases canvas1 = CanvasAudit( id=f"canvas_{uuid4().hex[:8]}", session_id=episode_test_session.id, canvas_type="chart", component_type="bar", component_name="RevenueChart", action="present", audit_metadata={"revenue": 1000000}, created_at=datetime.now(timezone.utc) ) canvas2 = CanvasAudit( id=f"canvas_{uuid4().hex[:8]}", session_id=episode_test_session.id, canvas_type="form", component_type="input", component_name="UserForm", action="submit", audit_metadata={"email": "test@example.com"}, created_at=datetime.now(timezone.utc) + timedelta(minutes=5) ) segmentation_service_mocked.db.add(canvas1) segmentation_service_mocked.db.add(canvas2) segmentation_service_mocked.db.commit() # Extract contexts contexts = segmentation_service_mocked._extract_canvas_context([canvas1, canvas2]) # Should handle multiple canvases assert contexts is not None assert "canvas_type" in contexts def test_episode_with_canvas_updates(self, segmentation_service_mocked, episode_test_session, episode_test_user): """ Test tracking canvas updates in episodes. Verifies: - Canvas update actions tracked - State changes recorded - Update metadata captured """ from core.models import CanvasAudit # Create canvas with update action canvas = CanvasAudit( id=f"canvas_{uuid4().hex[:8]}", session_id=episode_test_session.id, canvas_type="sheets", component_type="grid", component_name="DataTable", action="update", audit_metadata={"updated_cells": ["A1", "B2"]}, created_at=datetime.now(timezone.utc) ) segmentation_service_mocked.db.add(canvas) segmentation_service_mocked.db.commit() # Extract context context = segmentation_service_mocked._extract_canvas_context([canvas]) assert context is not None assert context["canvas_type"] == "sheets" # ============================================================================= # Test Feedback Integration # ============================================================================= class TestFeedbackIntegration: """Test feedback-weighted episode retrieval""" def test_aggregate_feedback_scores(self, segmentation_service_mocked): """ Test aggregating feedback scores for episodes. Verifies: - Feedback scores aggregated correctly - Average score calculated - Rating conversion applied """ from core.models import AgentFeedback # Create feedback records feedback1 = AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id="test_agent", feedback_type="rating", rating=5, # Excellent -> +1.0 created_at=datetime.now(timezone.utc) ) feedback2 = AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id="test_agent", feedback_type="rating", rating=3, # Neutral -> 0.0 created_at=datetime.now(timezone.utc) ) feedback3 = AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id="test_agent", feedback_type="rating", rating=1, # Poor -> -1.0 created_at=datetime.now(timezone.utc) ) # Calculate aggregate score # (5 + 3 + 1 - 3*2) / 3 / 2 = (9 - 6) / 6 = 3 / 6 = 0.5 # Using formula: (rating - 3) / 2 # (5-3)/2 = 1.0, (3-3)/2 = 0.0, (1-3)/2 = -1.0 # Average: (1.0 + 0.0 - 1.0) / 3 = 0.0 score = segmentation_service_mocked._calculate_feedback_score([feedback1, feedback2, feedback3]) assert score is not None assert -1.0 <= score <= 1.0 def test_feedback_weighted_retrieval(self, retrieval_service_mocked, episode_test_agent): """ Test retrieval with feedback weighting. Verifies: - High feedback episodes ranked higher - Low feedback episodes ranked lower - Feedback boosts applied """ import asyncio # Create episodes with different feedback scores episode_high = AgentEpisode( id=f"fb_high_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", aggregate_feedback_score=0.9 # High positive feedback ) episode_low = AgentEpisode( id=f"fb_low_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", aggregate_feedback_score=-0.5 # Negative feedback ) retrieval_service_mocked.db.add(episode_high) retrieval_service_mocked.db.add(episode_low) retrieval_service_mocked.db.commit() # Mock semantic search retrieval_service_mocked.lancedb.search.return_value = [] # Retrieve with feedback weighting async def test_retrieve(): result = await retrieval_service_mocked.retrieve_contextual( agent_id=episode_test_agent.id, current_task="Recent episodes", limit=5 ) # High feedback episode should be ranked higher # (boosted by +0.2 for positive feedback) assert result["count"] >= 0 asyncio.run(test_retrieve()) def test_feedback_linked_to_episode(self, segmentation_service_mocked, episode_test_session, episode_test_agent): """ Test feedback linked to episodes. Verifies: - Feedback records linked via episode_id - Feedback included in episode metadata - Metadata-only linkage (no duplication) """ from core.models import AgentFeedback, AgentExecution # Create execution execution = AgentExecution( id=f"exec_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, started_at=datetime.now(timezone.utc), completed_at=datetime.now(timezone.utc) + timedelta(minutes=5), result_summary="Task completed" ) segmentation_service_mocked.db.add(execution) segmentation_service_mocked.db.flush() # Create feedback linked to execution feedback = AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, agent_execution_id=execution.id, # Linked to execution feedback_type="thumbs_up", thumbs_up_down=True, created_at=datetime.now(timezone.utc) ) segmentation_service_mocked.db.add(feedback) segmentation_service_mocked.db.commit() # Fetch feedback by execution fetched_feedbacks = segmentation_service_mocked._fetch_feedback_context( session_id=episode_test_session.id, agent_id=episode_test_agent.id, execution_ids=[execution.id] ) assert len(fetched_feedbacks) >= 0 # ============================================================================= # Test Temporal Retrieval Modes (Task 1) # ============================================================================= class TestTemporalRetrieval: """ Test temporal retrieval with comprehensive time range coverage. Targets 80%+ line coverage for retrieve_temporal(): - Time range filtering (1d, 7d, 30d, 90d) - User ID filtering through ChatSession join - Result ordering (started_at DESC) - Archived episode exclusion - Governance check integration """ @pytest.mark.asyncio async def test_temporal_retrieval_1d(self, retrieval_service_mocked, episode_test_agent): """ Test temporal retrieval with 1-day time range. Verifies: - Episodes from last 24 hours returned - Episodes older than 1 day excluded - Results ordered by started_at DESC """ base_time = datetime.now(timezone.utc) # Create episodes at different times episode_1d = AgentEpisode( id=f"ep_1d_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=12), # Within 1d completed_at=base_time - timedelta(hours=11), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) episode_2d = AgentEpisode( id=f"ep_2d_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(days=2), # Outside 1d completed_at=base_time - timedelta(days=2) + timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) retrieval_service_mocked.db.add(episode_1d) retrieval_service_mocked.db.add(episode_2d) retrieval_service_mocked.db.commit() result = await retrieval_service_mocked.retrieve_temporal( agent_id=episode_test_agent.id, time_range="1d", limit=10 ) # Should only return episode from last 24 hours assert result["count"] == 1 assert len(result["episodes"]) == 1 assert result["episodes"][0]["id"] == episode_1d.id assert result["time_range"] == "1d" @pytest.mark.asyncio async def test_temporal_retrieval_7d(self, retrieval_service_mocked, episode_test_agent): """ Test temporal retrieval with 7-day time range (default). Verifies: - Episodes from last 7 days returned - Episodes older than 7 days excluded - Default time_range parameter works """ base_time = datetime.now(timezone.utc) episode_5d = AgentEpisode( id=f"ep_5d_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(days=5), completed_at=base_time - timedelta(days=5) + timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) episode_10d = AgentEpisode( id=f"ep_10d_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(days=10), completed_at=base_time - timedelta(days=10) + timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) retrieval_service_mocked.db.add(episode_5d) retrieval_service_mocked.db.add(episode_10d) retrieval_service_mocked.db.commit() result = await retrieval_service_mocked.retrieve_temporal( agent_id=episode_test_agent.id, time_range="7d", limit=10 ) # Should only return episode from last 7 days assert result["count"] == 1 assert result["episodes"][0]["id"] == episode_5d.id @pytest.mark.asyncio async def test_temporal_retrieval_30d(self, retrieval_service_mocked, episode_test_agent): """ Test temporal retrieval with 30-day time range. Verifies: - Episodes from last 30 days returned - Episodes older than 30 days excluded """ base_time = datetime.now(timezone.utc) episode_20d = AgentEpisode( id=f"ep_20d_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(days=20), completed_at=base_time - timedelta(days=20) + timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) episode_40d = AgentEpisode( id=f"ep_40d_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(days=40), completed_at=base_time - timedelta(days=40) + timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) retrieval_service_mocked.db.add(episode_20d) retrieval_service_mocked.db.add(episode_40d) retrieval_service_mocked.db.commit() result = await retrieval_service_mocked.retrieve_temporal( agent_id=episode_test_agent.id, time_range="30d", limit=10 ) # Should only return episode from last 30 days assert result["count"] == 1 assert result["episodes"][0]["id"] == episode_20d.id @pytest.mark.asyncio async def test_temporal_retrieval_90d(self, retrieval_service_mocked, episode_test_agent): """ Test temporal retrieval with 90-day time range. Verifies: - Episodes from last 90 days returned - Maximum time range filter works """ base_time = datetime.now(timezone.utc) episode_60d = AgentEpisode( id=f"ep_60d_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(days=60), completed_at=base_time - timedelta(days=60) + timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) episode_100d = AgentEpisode( id=f"ep_100d_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(days=100), completed_at=base_time - timedelta(days=100) + timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) retrieval_service_mocked.db.add(episode_60d) retrieval_service_mocked.db.add(episode_100d) retrieval_service_mocked.db.commit() result = await retrieval_service_mocked.retrieve_temporal( agent_id=episode_test_agent.id, time_range="90d", limit=10 ) # Should only return episode from last 90 days assert result["count"] == 1 assert result["episodes"][0]["id"] == episode_60d.id @pytest.mark.asyncio async def test_temporal_retrieval_with_user_filter(self, retrieval_service_mocked, episode_test_agent, episode_test_user): """ Test temporal retrieval with user_id filter. Verifies: - Episodes filtered by user_id through ChatSession join - Only episodes from specified user returned - User join works correctly """ base_time = datetime.now(timezone.utc) # Create session for test user session = ChatSession( id=f"session_{uuid4().hex[:8]}", user_id=episode_test_user.id, created_at=base_time ) retrieval_service_mocked.db.add(session) retrieval_service_mocked.db.flush() # Create episode with session episode_with_session = AgentEpisode( id=f"ep_session_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", session_id=session.id, started_at=base_time - timedelta(hours=1), completed_at=base_time, maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) # Create episode without session (different user) episode_no_session = AgentEpisode( id=f"ep_no_session_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=2), completed_at=base_time - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) retrieval_service_mocked.db.add(episode_with_session) retrieval_service_mocked.db.add(episode_no_session) retrieval_service_mocked.db.commit() result = await retrieval_service_mocked.retrieve_temporal( agent_id=episode_test_agent.id, time_range="7d", user_id=episode_test_user.id, limit=10 ) # Should only return episode from specified user assert result["count"] == 1 assert result["episodes"][0]["id"] == episode_with_session.id @pytest.mark.asyncio async def test_temporal_retrieval_ordering(self, retrieval_service_mocked, episode_test_agent): """ Test temporal retrieval result ordering. Verifies: - Results ordered by started_at DESC (newest first) - Ordering consistent across multiple episodes """ base_time = datetime.now(timezone.utc) # Create episodes with different timestamps episodes = [] for hours_ago in [1, 5, 10, 24, 48]: ep = AgentEpisode( id=f"ep_{hours_ago}h_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=hours_ago), completed_at=base_time - timedelta(hours=hours_ago) + timedelta(minutes=30), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) episodes.append(ep) retrieval_service_mocked.db.add(ep) retrieval_service_mocked.db.commit() result = await retrieval_service_mocked.retrieve_temporal( agent_id=episode_test_agent.id, time_range="7d", limit=10 ) # Verify ordering (newest first) assert result["count"] == 5 timestamps = [ep["started_at"] for ep in result["episodes"]] # Should be in descending order (newest first) assert timestamps == sorted(timestamps, reverse=True) @pytest.mark.asyncio async def test_temporal_retrieval_excludes_archived(self, retrieval_service_mocked, episode_test_agent): """ Test temporal retrieval excludes archived episodes. Verifies: - Archived episodes not returned in results - status != 'archived' filter works """ base_time = datetime.now(timezone.utc) episode_active = AgentEpisode( id=f"ep_active_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), completed_at=base_time, maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) episode_archived = AgentEpisode( id=f"ep_archived_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=2), completed_at=base_time - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="archived" ) retrieval_service_mocked.db.add(episode_active) retrieval_service_mocked.db.add(episode_archived) retrieval_service_mocked.db.commit() result = await retrieval_service_mocked.retrieve_temporal( agent_id=episode_test_agent.id, time_range="7d", limit=10 ) # Should only return active episode assert result["count"] == 1 assert result["episodes"][0]["id"] == episode_active.id assert result["episodes"][0]["status"] == "active" # ============================================================================= # Test Semantic Retrieval (Task 1) # ============================================================================= class TestSemanticRetrieval: """ Test semantic retrieval with LanceDB vector search. Targets 80%+ line coverage for retrieve_semantic(): - LanceDB search invocation with query - Agent ID filtering - Limit enforcement - Empty result handling - Governance check (INTERN+ requirement) """ @pytest.mark.asyncio async def test_semantic_retrieval_vector_search(self, retrieval_service_mocked, episode_test_agent): """ Test semantic retrieval invokes LanceDB vector search. Verifies: - LanceDB search called with query text - Agent ID filter applied - Episodes fetched from search results """ # Create test episode episode = AgentEpisode( id=f"semantic_ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) retrieval_service_mocked.db.add(episode) retrieval_service_mocked.db.commit() # Mock LanceDB search to return episode retrieval_service_mocked.lancedb.search.return_value = [ { "id": episode.id, "metadata": {"episode_id": episode.id}, "_distance": 0.15 # Low distance = high similarity } ] result = await retrieval_service_mocked.retrieve_semantic( agent_id=episode_test_agent.id, query="Python programming best practices", limit=10 ) # Verify search was called retrieval_service_mocked.lancedb.search.assert_called_once_with( table_name="episodes", query="Python programming best practices", filter_str=f"agent_id == '{episode_test_agent.id}'", limit=10 ) # Verify episode returned assert result["count"] >= 0 assert result["query"] == "Python programming best practices" @pytest.mark.asyncio async def test_semantic_retrieval_agent_filter(self, retrieval_service_mocked, episode_test_agent): """ Test semantic retrieval filters by agent_id. Verifies: - Agent ID filter passed to LanceDB - Only episodes from specified agent returned """ # Create two agents agent2 = AgentRegistry( id=f"agent2_{uuid4().hex[:8]}", name="Agent 2", status="AUTONOMOUS", category="testing", module_path="test.module", class_name="TestAgent2", confidence_score=0.9 ) retrieval_service_mocked.db.add(agent2) retrieval_service_mocked.db.flush() # Create episodes for each agent episode1 = AgentEpisode( id=f"ep_agent1_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) episode2 = AgentEpisode( id=f"ep_agent2_{uuid4().hex[:8]}", agent_id=agent2.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) retrieval_service_mocked.db.add(episode1) retrieval_service_mocked.db.add(episode2) retrieval_service_mocked.db.commit() # Mock LanceDB to return episode1 only retrieval_service_mocked.lancedb.search.return_value = [ { "id": episode1.id, "metadata": {"episode_id": episode1.id}, "_distance": 0.2 } ] result = await retrieval_service_mocked.retrieve_semantic( agent_id=episode_test_agent.id, query="test query", limit=10 ) # Verify agent ID filter in search call call_args = retrieval_service_mocked.lancedb.search.call_args assert f"agent_id == '{episode_test_agent.id}'" in call_args[1]["filter_str"] @pytest.mark.asyncio async def test_semantic_retrieval_limit(self, retrieval_service_mocked, episode_test_agent): """ Test semantic retrieval respects limit parameter. Verifies: - At most limit results returned - Limit passed to LanceDB search """ # Create multiple episodes episodes = [] for i in range(5): ep = AgentEpisode( id=f"ep_limit_{i}_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) episodes.append(ep) retrieval_service_mocked.db.add(ep) retrieval_service_mocked.db.commit() # Mock LanceDB to return all episodes mock_results = [ { "id": ep.id, "metadata": {"episode_id": ep.id}, "_distance": 0.1 + i * 0.1 } for i, ep in enumerate(episodes) ] retrieval_service_mocked.lancedb.search.return_value = mock_results result = await retrieval_service_mocked.retrieve_semantic( agent_id=episode_test_agent.id, query="test query", limit=3 ) # Verify limit passed to LanceDB call_args = retrieval_service_mocked.lancedb.search.call_args assert call_args[1]["limit"] == 3 @pytest.mark.asyncio async def test_semantic_retrieval_no_results(self, retrieval_service_mocked, episode_test_agent): """ Test semantic retrieval with no matching results. Verifies: - Empty search handled gracefully - Returns empty list without error """ # Mock LanceDB to return no results retrieval_service_mocked.lancedb.search.return_value = [] result = await retrieval_service_mocked.retrieve_semantic( agent_id=episode_test_agent.id, query="nonexistent topic", limit=10 ) # Should return empty list, not error assert result["count"] == 0 assert result["episodes"] == [] assert result["query"] == "nonexistent topic" @pytest.mark.asyncio async def test_semantic_retrieval_governance_check(self, retrieval_service_mocked, episode_test_agent): """ Test semantic retrieval enforces INTERN+ governance. Verifies: - Governance check performed for semantic_search action - INTERN+ maturity required - Governance check result included in response """ # Mock governance check to allow access with patch.object( retrieval_service_mocked.governance, 'can_perform_action', return_value={"allowed": True, "agent_maturity": "INTERN"} ): result = await retrieval_service_mocked.retrieve_semantic( agent_id=episode_test_agent.id, query="test query", limit=10 ) # Verify governance check in response assert "governance_check" in result assert result["governance_check"]["allowed"] == True @pytest.mark.asyncio async def test_semantic_retrieval_metadata_parsing(self, retrieval_service_mocked, episode_test_agent): """ Test semantic retrieval handles different metadata formats. Verifies: - String metadata parsed to JSON - Dict metadata handled correctly - Episode ID extracted from metadata """ episode = AgentEpisode( id=f"ep_meta_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc) - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) retrieval_service_mocked.db.add(episode) retrieval_service_mocked.db.commit() # Test with string metadata (JSON) retrieval_service_mocked.lancedb.search.return_value = [ { "id": episode.id, "metadata": '{"episode_id": "' + episode.id + '"}', "_distance": 0.1 } ] result = await retrieval_service_mocked.retrieve_semantic( agent_id=episode_test_agent.id, query="test", limit=10 ) # Should handle string metadata assert result["count"] >= 0 # ============================================================================= # Test Episode Creation Flow (Task 1: Phase 166-02) # ============================================================================= class TestEpisodeCreationFlow: """Test episode creation flow with canvas and feedback integration""" @pytest.mark.asyncio async def test_create_episode_from_session_basic(self, segmentation_service_mocked, episode_test_session, episode_test_agent): """ Test basic episode creation from chat session with messages. Verifies: - Episode created with correct fields - Segments created from messages - Episode metadata populated correctly """ import asyncio # Create messages in session base_time = datetime.now(timezone.utc) messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=episode_test_session.id, role="user", content="Help me with Python programming", created_at=base_time ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=episode_test_session.id, role="assistant", content="I can help with Python!", created_at=base_time + timedelta(minutes=1) ), ] for msg in messages: segmentation_service_mocked.db.add(msg) segmentation_service_mocked.db.commit() # Create episode from session episode = await segmentation_service_mocked.create_episode_from_session( session_id=episode_test_session.id, agent_id=episode_test_agent.id, force_create=True ) # Verify episode created assert episode is not None assert episode.status == "completed" assert episode.session_id == episode_test_session.id assert episode.agent_id == episode_test_agent.id assert episode.title is not None assert episode.description is not None # Verify segments created segments = segmentation_service_mocked.db.query(EpisodeSegment).filter( EpisodeSegment.episode_id == episode.id ).all() assert len(segments) >= 1 assert any(s.segment_type == "conversation" for s in segments) @pytest.mark.asyncio async def test_create_episode_with_canvas_context(self, segmentation_service_mocked, episode_test_session, episode_test_agent): """ Test episode creation includes canvas presentations. Verifies: - CanvasAudit records fetched for session - Canvas context extracted and stored - Episode has canvas_action_count populated """ import asyncio # Create canvas audit canvas = CanvasAudit( id=f"canvas_{uuid4().hex[:8]}", session_id=episode_test_session.id, canvas_type="chart", component_type="line", component_name="SalesChart", action="present", audit_metadata={"title": "Monthly Sales", "revenue": 1000000}, created_at=datetime.now(timezone.utc) ) segmentation_service_mocked.db.add(canvas) # Create minimal message for episode creation msg = ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=episode_test_session.id, role="user", content="Show me sales data", created_at=datetime.now(timezone.utc) ) segmentation_service_mocked.db.add(msg) segmentation_service_mocked.db.commit() # Create episode episode = await segmentation_service_mocked.create_episode_from_session( session_id=episode_test_session.id, agent_id=episode_test_agent.id, force_create=True ) # Verify canvas context included assert episode is not None assert episode.canvas_action_count == 1 assert len(episode.canvas_ids) == 1 assert canvas.id in episode.canvas_ids @pytest.mark.asyncio async def test_create_episode_with_feedback_context(self, segmentation_service_mocked, episode_test_session, episode_test_agent): """ Test episode creation includes user feedback. Verifies: - AgentFeedback records fetched for executions - Feedback score calculated and aggregated - Episode has aggregate_feedback_score populated """ import asyncio # Create execution execution = AgentExecution( id=f"exec_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, started_at=datetime.now(timezone.utc), completed_at=datetime.now(timezone.utc) + timedelta(minutes=5), result_summary="Task completed successfully" ) segmentation_service_mocked.db.add(execution) segmentation_service_mocked.db.flush() # Create feedback linked to execution feedback = AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, agent_execution_id=execution.id, feedback_type="thumbs_up", thumbs_up_down=True, created_at=datetime.now(timezone.utc) ) segmentation_service_mocked.db.add(feedback) # Create minimal message for episode creation msg = ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=episode_test_session.id, role="user", content="Execute task", created_at=datetime.now(timezone.utc) ) segmentation_service_mocked.db.add(msg) segmentation_service_mocked.db.commit() # Create episode episode = await segmentation_service_mocked.create_episode_from_session( session_id=episode_test_session.id, agent_id=episode_test_agent.id, force_create=True ) # Verify feedback context included assert episode is not None assert episode.aggregate_feedback_score is not None assert episode.aggregate_feedback_score > 0 # thumbs_up should be positive assert len(episode.feedback_ids) == 1 assert feedback.id in episode.feedback_ids @pytest.mark.asyncio async def test_create_episode_force_small_session(self, segmentation_service_mocked, episode_test_session, episode_test_agent): """ Test force_create flag for sessions with <2 items. Verifies: - force_create=True creates episode for small sessions - Episode created even with minimal data """ import asyncio # Create single message (normally too small) msg = ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=episode_test_session.id, role="user", content="Single message", created_at=datetime.now(timezone.utc) ) segmentation_service_mocked.db.add(msg) segmentation_service_mocked.db.commit() # Create episode with force_create=True episode = await segmentation_service_mocked.create_episode_from_session( session_id=episode_test_session.id, agent_id=episode_test_agent.id, force_create=True ) # Verify episode created despite being small assert episode is not None assert episode.status == "completed" # Verify segments created segments = segmentation_service_mocked.db.query(EpisodeSegment).filter( EpisodeSegment.episode_id == episode.id ).all() assert len(segments) >= 1 @pytest.mark.asyncio async def test_create_episode_with_executions(self, segmentation_service_mocked, episode_test_session, episode_test_agent): """ Test episode creation includes agent executions. Verifies: - AgentExecution records fetched for agent - Executions included in episode - Episode has execution_ids populated """ import asyncio # Create executions executions = [ AgentExecution( id=f"exec_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, started_at=datetime.now(timezone.utc), completed_at=datetime.now(timezone.utc) + timedelta(minutes=2), input_summary="Task 1", result_summary="Completed task 1" ), AgentExecution( id=f"exec_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, started_at=datetime.now(timezone.utc) + timedelta(minutes=5), completed_at=datetime.now(timezone.utc) + timedelta(minutes=7), input_summary="Task 2", result_summary="Completed task 2" ), ] for exec in executions: segmentation_service_mocked.db.add(exec) # Create minimal message for episode creation msg = ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=episode_test_session.id, role="user", content="Execute tasks", created_at=datetime.now(timezone.utc) ) segmentation_service_mocked.db.add(msg) segmentation_service_mocked.db.commit() # Create episode episode = await segmentation_service_mocked.create_episode_from_session( session_id=episode_test_session.id, agent_id=episode_test_agent.id, force_create=True ) # Verify executions included assert episode is not None assert len(episode.execution_ids) == 2 assert all(exec_id in episode.execution_ids for exec_id in [e.id for e in executions]) # Verify segments include execution segments segments = segmentation_service_mocked.db.query(EpisodeSegment).filter( EpisodeSegment.episode_id == episode.id ).all() execution_segments = [s for s in segments if s.segment_type == "execution"] assert len(execution_segments) == 2 @pytest.mark.asyncio async def test_create_episode_links_canvas_to_episode(self, segmentation_service_mocked, episode_test_session, episode_test_agent): """ Test CanvasAudit.episode_id is set after episode creation. Verifies: - CanvasAudit records have episode_id populated - Linkage is bidirectional (episode.canvas_ids, canvas.episode_id) """ import asyncio # Create canvas audits canvases = [ CanvasAudit( id=f"canvas_{uuid4().hex[:8]}", session_id=episode_test_session.id, canvas_type="chart", component_type="bar", component_name="RevenueChart", action="present", audit_metadata={"revenue": 500000}, created_at=datetime.now(timezone.utc) ), CanvasAudit( id=f"canvas_{uuid4().hex[:8]}", session_id=episode_test_session.id, canvas_type="form", component_type="input", component_name="UserForm", action="submit", audit_metadata={"email": "test@example.com"}, created_at=datetime.now(timezone.utc) + timedelta(minutes=1) ), ] for canvas in canvases: segmentation_service_mocked.db.add(canvas) # Create minimal message for episode creation msg = ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=episode_test_session.id, role="user", content="Show forms", created_at=datetime.now(timezone.utc) ) segmentation_service_mocked.db.add(msg) segmentation_service_mocked.db.commit() # Create episode episode = await segmentation_service_mocked.create_episode_from_session( session_id=episode_test_session.id, agent_id=episode_test_agent.id, force_create=True ) # Verify bidirectional linkage assert episode is not None # Check episode.canvas_ids assert len(episode.canvas_ids) == 2 # Check CanvasAudit.episode_id (refresh from DB) segmentation_service_mocked.db.refresh(canvases[0]) segmentation_service_mocked.db.refresh(canvases[1]) assert canvases[0].episode_id == episode.id assert canvases[1].episode_id == episode.id @pytest.mark.asyncio async def test_create_episode_links_feedback_to_episode(self, segmentation_service_mocked, episode_test_session, episode_test_agent): """ Test AgentFeedback.episode_id is set after episode creation. Verifies: - AgentFeedback records have episode_id populated - Linkage is bidirectional (episode.feedback_ids, feedback.episode_id) """ import asyncio # Create execution execution = AgentExecution( id=f"exec_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, started_at=datetime.now(timezone.utc), completed_at=datetime.now(timezone.utc) + timedelta(minutes=3), result_summary="Task done" ) segmentation_service_mocked.db.add(execution) segmentation_service_mocked.db.flush() # Create feedback records feedbacks = [ AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, agent_execution_id=execution.id, feedback_type="thumbs_up", thumbs_up_down=True, created_at=datetime.now(timezone.utc) ), AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, agent_execution_id=execution.id, feedback_type="rating", rating=5, created_at=datetime.now(timezone.utc) + timedelta(seconds=30) ), ] for fb in feedbacks: segmentation_service_mocked.db.add(fb) # Create minimal message for episode creation msg = ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id=episode_test_session.id, role="user", content="Execute task", created_at=datetime.now(timezone.utc) ) segmentation_service_mocked.db.add(msg) segmentation_service_mocked.db.commit() # Create episode episode = await segmentation_service_mocked.create_episode_from_session( session_id=episode_test_session.id, agent_id=episode_test_agent.id, force_create=True ) # Verify bidirectional linkage assert episode is not None # Check episode.feedback_ids assert len(episode.feedback_ids) == 2 # Check AgentFeedback.episode_id (refresh from DB) segmentation_service_mocked.db.refresh(feedbacks[0]) segmentation_service_mocked.db.refresh(feedbacks[1]) assert feedbacks[0].episode_id == episode.id assert feedbacks[1].episode_id == episode.id # ============================================================================= # Test Canvas Context Extraction (Task 2: Phase 166-02) # ============================================================================= class TestCanvasContextExtraction: """Test canvas context extraction methods""" def test_fetch_canvas_context_by_session(self, segmentation_service_mocked, episode_test_session): """ Test _fetch_canvas_context retrieves all canvases for session. Verifies: - All CanvasAudit records for session fetched - Results ordered by creation time - Empty list returned if no canvases """ # Create canvas audits canvases = [ CanvasAudit( id=f"canvas_{uuid4().hex[:8]}", session_id=episode_test_session.id, canvas_type="chart", component_type="line", component_name="Chart1", action="present", audit_metadata={"title": "Chart 1"}, created_at=datetime.now(timezone.utc) ), CanvasAudit( id=f"canvas_{uuid4().hex[:8]}", session_id=episode_test_session.id, canvas_type="form", component_type="input", component_name="Form1", action="submit", audit_metadata={"field": "value"}, created_at=datetime.now(timezone.utc) + timedelta(minutes=1) ), ] for canvas in canvases: segmentation_service_mocked.db.add(canvas) segmentation_service_mocked.db.commit() # Fetch canvas context fetched_canvases = segmentation_service_mocked._fetch_canvas_context(episode_test_session.id) assert len(fetched_canvases) == 2 assert fetched_canvases[0].id == canvases[0].id # Ordered by created_at assert fetched_canvases[1].id == canvases[1].id def test_extract_canvas_context_from_audits(self, segmentation_service_mocked): """ Test _extract_canvas_context transforms audits to context dict. Verifies: - Canvas type extracted correctly - Critical data points captured - Visual elements listed """ canvas_audit = CanvasAudit( id=f"canvas_{uuid4().hex[:8]}", session_id="test_session", canvas_type="chart", component_type="line", component_name="SalesChart", action="present", audit_metadata={"title": "Monthly Sales", "revenue": 1000000} ) context = segmentation_service_mocked._extract_canvas_context([canvas_audit]) assert context is not None assert context["canvas_type"] == "chart" assert "presentation_summary" in context assert "visual_elements" in context assert "critical_data_points" in context def test_extract_canvas_context_with_critical_data(self, segmentation_service_mocked): """ Test _extract_canvas_context extracts business-critical data points. Verifies: - workflow_id captured for orchestration canvas - revenue captured for sheets canvas - command captured for terminal canvas """ # Test orchestration canvas orch_canvas = CanvasAudit( id=f"canvas_{uuid4().hex[:8]}", session_id="test_session", canvas_type="orchestration", component_type="workflow", component_name="WorkflowOrchestrator", action="present", audit_metadata={"workflow_id": "wf_123", "approval_status": "pending"} ) context = segmentation_service_mocked._extract_canvas_context([orch_canvas]) assert context["canvas_type"] == "orchestration" assert "critical_data_points" in context assert context["critical_data_points"].get("workflow_id") == "wf_123" assert context["critical_data_points"].get("approval_status") == "pending" def test_extract_canvas_context_detail_filtering(self, segmentation_service_mocked): """ Test _filter_canvas_context_detail with summary/standard/full. Verifies: - summary: only presentation_summary - standard: summary + critical_data_points - full: all fields including visual_elements """ full_context = { "canvas_type": "chart", "presentation_summary": "Agent presented SalesChart", "visual_elements": ["line", "grid"], "user_interaction": "presented to user", "critical_data_points": {"revenue": 1000000} } # Test summary level summary_context = segmentation_service_mocked._filter_canvas_context_detail(full_context, "summary") assert "presentation_summary" in summary_context assert "critical_data_points" not in summary_context assert "visual_elements" not in summary_context # Test standard level standard_context = segmentation_service_mocked._filter_canvas_context_detail(full_context, "standard") assert "presentation_summary" in standard_context assert "critical_data_points" in standard_context assert "visual_elements" not in standard_context # Test full level full_filtered = segmentation_service_mocked._filter_canvas_context_detail(full_context, "full") assert "presentation_summary" in full_filtered assert "critical_data_points" in full_filtered assert "visual_elements" in full_filtered assert "user_interaction" in full_filtered def test_extract_canvas_context_empty_audits(self, segmentation_service_mocked): """ Test _extract_canvas_context handles empty canvas list gracefully. Verifies: - Returns empty dict for empty list - No errors raised """ context = segmentation_service_mocked._extract_canvas_context([]) assert context == {} # ============================================================================= # Test Feedback Aggregation and Segment Creation (Task 3: Phase 166-02) # ============================================================================= class TestFeedbackAndSegments: """Test feedback aggregation and segment creation methods""" def test_fetch_feedback_context_by_executions(self, segmentation_service_mocked, episode_test_agent): """ Test _fetch_feedback_context retrieves feedback by execution_ids. Verifies: - Feedback records fetched for given execution IDs - Empty list returned if no feedback - Only feedback for agent returned """ # Create executions execution1 = AgentExecution( id=f"exec_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, started_at=datetime.now(timezone.utc), completed_at=datetime.now(timezone.utc) + timedelta(minutes=2) ) execution2 = AgentExecution( id=f"exec_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, started_at=datetime.now(timezone.utc) + timedelta(minutes=5), completed_at=datetime.now(timezone.utc) + timedelta(minutes=7) ) segmentation_service_mocked.db.add(execution1) segmentation_service_mocked.db.add(execution2) segmentation_service_mocked.db.flush() # Create feedback for execution1 feedback1 = AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, agent_execution_id=execution1.id, feedback_type="thumbs_up", thumbs_up_down=True ) segmentation_service_mocked.db.add(feedback1) segmentation_service_mocked.db.commit() # Fetch feedback context fetched_feedbacks = segmentation_service_mocked._fetch_feedback_context( session_id="test_session", agent_id=episode_test_agent.id, execution_ids=[execution1.id, execution2.id] ) assert len(fetched_feedbacks) == 1 assert fetched_feedbacks[0].id == feedback1.id def test_calculate_feedback_score_thumbs_up(self, segmentation_service_mocked): """ Test thumbs_up maps to +1.0. Verifies: - thumbs_up_down=True maps to 1.0 - Correct score calculation """ feedback = AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id="test_agent", feedback_type="thumbs_up", thumbs_up_down=True ) score = segmentation_service_mocked._calculate_feedback_score([feedback]) assert score == 1.0 def test_calculate_feedback_score_thumbs_down(self, segmentation_service_mocked): """ Test thumbs_down maps to -1.0. Verifies: - thumbs_up_down=False maps to -1.0 - Correct score calculation """ feedback = AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id="test_agent", feedback_type="thumbs_down", thumbs_up_down=False ) score = segmentation_service_mocked._calculate_feedback_score([feedback]) assert score == -1.0 def test_calculate_feedback_score_rating(self, segmentation_service_mocked): """ Test rating 1-5 maps to -1.0 to 1.0. Verifies: - Rating 5 maps to 1.0 - Rating 3 maps to 0.0 - Rating 1 maps to -1.0 - Formula: (rating - 3) / 2 """ feedback5 = AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id="test_agent", feedback_type="rating", rating=5 ) feedback3 = AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id="test_agent", feedback_type="rating", rating=3 ) feedback1 = AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id="test_agent", feedback_type="rating", rating=1 ) score5 = segmentation_service_mocked._calculate_feedback_score([feedback5]) score3 = segmentation_service_mocked._calculate_feedback_score([feedback3]) score1 = segmentation_service_mocked._calculate_feedback_score([feedback1]) assert score5 == 1.0 assert score3 == 0.0 assert score1 == -1.0 def test_calculate_feedback_score_aggregate(self, segmentation_service_mocked): """ Test multiple feedbacks averaged correctly. Verifies: - Mixed feedback types aggregated - Average score calculated correctly - (1.0 + 0.0 - 1.0) / 3 = 0.0 """ feedbacks = [ AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id="test_agent", feedback_type="rating", rating=5 # +1.0 ), AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id="test_agent", feedback_type="rating", rating=3 # 0.0 ), AgentFeedback( id=f"fb_{uuid4().hex[:8]}", agent_id="test_agent", feedback_type="rating", rating=1 # -1.0 ), ] score = segmentation_service_mocked._calculate_feedback_score(feedbacks) assert score == 0.0 # (1.0 + 0.0 - 1.0) / 3 = 0.0 def test_calculate_feedback_score_empty(self, segmentation_service_mocked): """ Test no feedback returns None. Verifies: - Empty list returns None - No errors raised """ score = segmentation_service_mocked._calculate_feedback_score([]) assert score is None @pytest.mark.asyncio async def test_create_segments_from_messages(self, segmentation_service_mocked, episode_test_agent): """ Test _create_segments creates conversation segments. Verifies: - Segments created from messages - Content formatted correctly - Sequence order assigned """ import asyncio # Create episode episode = AgentEpisode( id=f"ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success" ) segmentation_service_mocked.db.add(episode) segmentation_service_mocked.db.commit() # Create messages messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id="test_session", role="user", content="Hello", created_at=datetime.now(timezone.utc) ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id="test_session", role="assistant", content="Hi there!", created_at=datetime.now(timezone.utc) + timedelta(minutes=1) ), ] # Create segments await segmentation_service_mocked._create_segments( episode=episode, messages=messages, executions=[], boundaries=set(), canvas_context=None ) # Verify segments created segments = segmentation_service_mocked.db.query(EpisodeSegment).filter( EpisodeSegment.episode_id == episode.id ).all() assert len(segments) == 1 assert segments[0].segment_type == "conversation" assert segments[0].sequence_order == 0 assert "user: Hello" in segments[0].content.lower() assert "assistant: Hi there!" in segments[0].content.lower() @pytest.mark.asyncio async def test_create_segments_from_executions(self, segmentation_service_mocked, episode_test_agent): """ Test _create_segments creates execution segments. Verifies: - Segments created from executions - Execution details formatted correctly """ import asyncio # Create episode episode = AgentEpisode( id=f"ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success" ) segmentation_service_mocked.db.add(episode) segmentation_service_mocked.db.commit() # Create executions executions = [ AgentExecution( id=f"exec_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, started_at=datetime.now(timezone.utc), completed_at=datetime.now(timezone.utc) + timedelta(minutes=2), input_summary="Task 1", result_summary="Completed" ) ] # Create segments await segmentation_service_mocked._create_segments( episode=episode, messages=[], executions=executions, boundaries=set(), canvas_context=None ) # Verify segments created segments = segmentation_service_mocked.db.query(EpisodeSegment).filter( EpisodeSegment.episode_id == episode.id ).all() assert len(segments) == 1 assert segments[0].segment_type == "execution" assert "Task 1" in segments[0].content assert "Completed" in segments[0].content @pytest.mark.asyncio async def test_create_segments_sequence_order(self, segmentation_service_mocked, episode_test_agent): """ Test segments have correct sequence_order. Verifies: - Conversation segments ordered first - Execution segments ordered after conversations - Sequence increments correctly """ import asyncio # Create episode episode = AgentEpisode( id=f"ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success" ) segmentation_service_mocked.db.add(episode) segmentation_service_mocked.db.commit() # Create messages and executions messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id="test_session", role="user", content="Message 1", created_at=datetime.now(timezone.utc) ) ] executions = [ AgentExecution( id=f"exec_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, started_at=datetime.now(timezone.utc) + timedelta(minutes=1), completed_at=datetime.now(timezone.utc) + timedelta(minutes=2), result_summary="Task" ) ] # Create segments await segmentation_service_mocked._create_segments( episode=episode, messages=messages, executions=executions, boundaries=set(), canvas_context=None ) # Verify sequence order segments = segmentation_service_mocked.db.query(EpisodeSegment).filter( EpisodeSegment.episode_id == episode.id ).order_by(EpisodeSegment.sequence_order).all() assert len(segments) == 2 assert segments[0].sequence_order == 0 assert segments[1].sequence_order == 1 assert segments[0].segment_type == "conversation" assert segments[1].segment_type == "execution" @pytest.mark.asyncio async def test_create_segments_with_boundaries(self, segmentation_service_mocked, episode_test_agent): """ Test boundaries split into multiple segments. Verifies: - Boundary at index splits messages - Multiple segments created - Boundary respected in segmentation """ import asyncio # Create episode episode = AgentEpisode( id=f"ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success" ) segmentation_service_mocked.db.add(episode) segmentation_service_mocked.db.commit() # Create messages with boundary at index 2 messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id="test_session", role="user", content="Message 1", created_at=datetime.now(timezone.utc) ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id="test_session", role="assistant", content="Response 1", created_at=datetime.now(timezone.utc) + timedelta(minutes=1) ), ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id="test_session", role="user", content="Message 2 (new topic)", created_at=datetime.now(timezone.utc) + timedelta(minutes=2) ), ] # Create segments with boundary at index 2 await segmentation_service_mocked._create_segments( episode=episode, messages=messages, executions=[], boundaries={2}, # Split after message 1 canvas_context=None ) # Verify two segments created segments = segmentation_service_mocked.db.query(EpisodeSegment).filter( EpisodeSegment.episode_id == episode.id ).order_by(EpisodeSegment.sequence_order).all() assert len(segments) == 2 assert segments[0].sequence_order == 0 assert segments[1].sequence_order == 1 @pytest.mark.asyncio async def test_create_segments_canvas_context(self, segmentation_service_mocked, episode_test_agent): """ Test segments include canvas_context. Verifies: - Canvas context passed to segments - Context stored in segment metadata """ import asyncio # Create episode episode = AgentEpisode( id=f"ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=datetime.now(timezone.utc), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success" ) segmentation_service_mocked.db.add(episode) segmentation_service_mocked.db.commit() # Create messages messages = [ ChatMessage( id=f"msg_{uuid4().hex[:8]}", conversation_id="test_session", role="user", content="Show data", created_at=datetime.now(timezone.utc) ) ] # Canvas context canvas_context = { "canvas_type": "chart", "presentation_summary": "Agent presented SalesChart", "critical_data_points": {"revenue": 1000000} } # Create segments with canvas context await segmentation_service_mocked._create_segments( episode=episode, messages=messages, executions=[], boundaries=set(), canvas_context=canvas_context ) # Verify canvas context in segments segments = segmentation_service_mocked.db.query(EpisodeSegment).filter( EpisodeSegment.episode_id == episode.id ).all() assert len(segments) == 1 assert segments[0].canvas_context == canvas_context # ============================================================================= # Test Sequential Retrieval (Task 2) # ============================================================================= class TestSequentialRetrieval: """ Test sequential retrieval with full episode and segments. Targets 80%+ line coverage for retrieve_sequential(): - Full episode with all segments returned - Segments ordered by sequence_order - Canvas context included by default - Feedback context included by default - Exclusion parameters work (include_canvas=False, include_feedback=False) - Not found error handling """ @pytest.mark.asyncio async def test_sequential_retrieval_full_episode(self, retrieval_service_mocked, episode_test_agent): """ Test sequential retrieval returns full episode with all segments. Verifies: - Episode object returned - All segments included - Episode ID matches request """ base_time = datetime.now(timezone.utc) episode = AgentEpisode( id=f"seq_ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), completed_at=base_time, maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) retrieval_service_mocked.db.add(episode) retrieval_service_mocked.db.flush() # Create segments segments = [] for i in range(3): segment = EpisodeSegment( id=f"seg_{uuid4().hex[:8]}", episode_id=episode.id, segment_type="conversation", sequence_order=i, content=f"Segment {i} content", content_summary=f"Segment {i}", source_type="test", source_id=f"source_{i}" ) segments.append(segment) retrieval_service_mocked.db.add(segment) retrieval_service_mocked.db.commit() result = await retrieval_service_mocked.retrieve_sequential( episode_id=episode.id, agent_id=episode_test_agent.id, include_canvas=False, include_feedback=False ) assert result["episode"]["id"] == episode.id assert len(result["segments"]) == 3 @pytest.mark.asyncio async def test_sequential_retrieval_segment_ordering(self, retrieval_service_mocked, episode_test_agent): """ Test sequential retrieval segments are ordered correctly. Verifies: - Segments ordered by sequence_order ASC - Order consistent with creation """ base_time = datetime.now(timezone.utc) episode = AgentEpisode( id=f"seq_ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), completed_at=base_time, maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) retrieval_service_mocked.db.add(episode) retrieval_service_mocked.db.flush() # Create segments in specific order segments_data = [ (2, "Second segment"), (0, "First segment"), (1, "Middle segment"), ] for order, content in segments_data: segment = EpisodeSegment( id=f"seg_{uuid4().hex[:8]}", episode_id=episode.id, segment_type="conversation", sequence_order=order, content=content, content_summary=content, source_type="test", source_id=f"source_{order}" ) retrieval_service_mocked.db.add(segment) retrieval_service_mocked.db.commit() result = await retrieval_service_mocked.retrieve_sequential( episode_id=episode.id, agent_id=episode_test_agent.id, include_canvas=False, include_feedback=False ) # Verify segments are ordered order_values = [s["sequence_order"] for s in result["segments"]] assert order_values == [0, 1, 2] @pytest.mark.asyncio async def test_sequential_retrieval_with_canvas(self, retrieval_service_mocked, episode_test_agent): """ Test sequential retrieval includes canvas context by default. Verifies: - Canvas context included when include_canvas=True - canvas_context key present in result """ base_time = datetime.now(timezone.utc) episode = AgentEpisode( id=f"seq_ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), completed_at=base_time, maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active", canvas_ids=["canvas_1", "canvas_2"] ) retrieval_service_mocked.db.add(episode) retrieval_service_mocked.db.commit() # Create canvas audit records canvas1 = CanvasAudit( id="canvas_1", session_id=f"session_{uuid4().hex[:8]}", canvas_type="chart", component_type="line", component_name="SalesChart", action="present", audit_metadata={"title": "Monthly Sales"}, created_at=base_time ) canvas2 = CanvasAudit( id="canvas_2", session_id=f"session_{uuid4().hex[:8]}", canvas_type="form", component_type="input", component_name="UserForm", action="submit", audit_metadata={"email": "test@example.com"}, created_at=base_time ) retrieval_service_mocked.db.add(canvas1) retrieval_service_mocked.db.add(canvas2) retrieval_service_mocked.db.commit() result = await retrieval_service_mocked.retrieve_sequential( episode_id=episode.id, agent_id=episode_test_agent.id, include_canvas=True, include_feedback=False ) # Canvas context should be included assert "canvas_context" in result assert len(result["canvas_context"]) == 2 @pytest.mark.asyncio async def test_sequential_retrieval_with_feedback(self, retrieval_service_mocked, episode_test_agent): """ Test sequential retrieval includes feedback context by default. Verifies: - Feedback context included when include_feedback=True - feedback_context key present in result """ base_time = datetime.now(timezone.utc) episode = AgentEpisode( id=f"seq_ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), completed_at=base_time, maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active", feedback_ids=["fb_1", "fb_2"] ) retrieval_service_mocked.db.add(episode) retrieval_service_mocked.db.commit() # Create feedback records feedback1 = AgentFeedback( id="fb_1", agent_id=episode_test_agent.id, feedback_type="rating", rating=5, thumbs_up_down=True, created_at=base_time ) feedback2 = AgentFeedback( id="fb_2", agent_id=episode_test_agent.id, feedback_type="thumbs_up", thumbs_up_down=True, created_at=base_time ) retrieval_service_mocked.db.add(feedback1) retrieval_service_mocked.db.add(feedback2) retrieval_service_mocked.db.commit() result = await retrieval_service_mocked.retrieve_sequential( episode_id=episode.id, agent_id=episode_test_agent.id, include_canvas=False, include_feedback=True ) # Feedback context should be included assert "feedback_context" in result assert len(result["feedback_context"]) == 2 @pytest.mark.asyncio async def test_sequential_retrieval_exclude_canvas(self, retrieval_service_mocked, episode_test_agent): """ Test sequential retrieval excludes canvas when include_canvas=False. Verifies: - Canvas context not included when include_canvas=False - canvas_context key not present in result """ base_time = datetime.now(timezone.utc) episode = AgentEpisode( id=f"seq_ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), completed_at=base_time, maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active", canvas_ids=["canvas_1"] ) retrieval_service_mocked.db.add(episode) retrieval_service_mocked.db.commit() result = await retrieval_service_mocked.retrieve_sequential( episode_id=episode.id, agent_id=episode_test_agent.id, include_canvas=False, include_feedback=False ) # Canvas context should NOT be included assert "canvas_context" not in result @pytest.mark.asyncio async def test_sequential_retrieval_exclude_feedback(self, retrieval_service_mocked, episode_test_agent): """ Test sequential retrieval excludes feedback when include_feedback=False. Verifies: - Feedback context not included when include_feedback=False - feedback_context key not present in result """ base_time = datetime.now(timezone.utc) episode = AgentEpisode( id=f"seq_ep_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), completed_at=base_time, maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active", feedback_ids=["fb_1"] ) retrieval_service_mocked.db.add(episode) retrieval_service_mocked.db.commit() result = await retrieval_service_mocked.retrieve_sequential( episode_id=episode.id, agent_id=episode_test_agent.id, include_canvas=False, include_feedback=False ) # Feedback context should NOT be included assert "feedback_context" not in result @pytest.mark.asyncio async def test_sequential_retrieval_not_found(self, retrieval_service_mocked, episode_test_agent): """ Test sequential retrieval returns error for nonexistent episode. Verifies: - Error message returned when episode not found - No exception raised """ result = await retrieval_service_mocked.retrieve_sequential( episode_id="nonexistent_episode", agent_id=episode_test_agent.id, include_canvas=False, include_feedback=False ) assert "error" in result assert result["error"] == "Episode not found" # ============================================================================= # Test Contextual Retrieval (Task 2) # ============================================================================= class TestContextualRetrieval: """ Test contextual retrieval with hybrid scoring. Targets 80%+ line coverage for retrieve_contextual(): - Hybrid scoring: temporal (30%) + semantic (70%) - Canvas boost: +0.1 for episodes with canvas - Positive feedback boost: +0.2 - Negative feedback penalty: -0.3 - require_canvas filter - require_feedback filter - Limit enforcement """ @pytest.mark.asyncio async def test_contextual_retrieval_hybrid_scoring(self, retrieval_service_mocked, episode_test_agent): """ Test contextual retrieval combines temporal and semantic scores. Verifies: - Temporal episodes get 0.3 weight - Semantic episodes get 0.7 weight - Combined scores used for ranking """ base_time = datetime.now(timezone.utc) # Create episodes within 30 days episodes = [] for i in range(3): ep = AgentEpisode( id=f"ctx_ep_{i}_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(days=i), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) episodes.append(ep) retrieval_service_mocked.db.add(ep) retrieval_service_mocked.db.commit() # Mock semantic search to return one episode retrieval_service_mocked.lancedb.search.return_value = [ { "id": episodes[0].id, "metadata": {"episode_id": episodes[0].id}, "_distance": 0.2 } ] result = await retrieval_service_mocked.retrieve_contextual( agent_id=episode_test_agent.id, current_task="test task", limit=5, require_canvas=False, require_feedback=False ) # Should return episodes with hybrid scores assert result["count"] >= 0 if result["count"] > 0: # First episode should have higher score (temporal + semantic) assert "relevance_score" in result["episodes"][0] @pytest.mark.asyncio async def test_contextual_retrieval_canvas_boost(self, retrieval_service_mocked, episode_test_agent): """ Test contextual retrieval applies canvas boost. Verifies: - Episodes with canvas_action_count > 0 get +0.1 boost - Boost applied to relevance score """ base_time = datetime.now(timezone.utc) episode_with_canvas = AgentEpisode( id=f"ctx_canvas_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active", canvas_action_count=5 # Has canvas interactions ) episode_without_canvas = AgentEpisode( id=f"ctx_no_canvas_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active", canvas_action_count=0 # No canvas interactions ) retrieval_service_mocked.db.add(episode_with_canvas) retrieval_service_mocked.db.add(episode_without_canvas) retrieval_service_mocked.db.commit() # Mock semantic search to return both retrieval_service_mocked.lancedb.search.return_value = [] result = await retrieval_service_mocked.retrieve_contextual( agent_id=episode_test_agent.id, current_task="test task", limit=5, require_canvas=False, require_feedback=False ) # Episode with canvas should rank higher if result["count"] >= 2: scores = {ep["id"]: ep.get("relevance_score", 0) for ep in result["episodes"]} canvas_score = scores.get(episode_with_canvas.id, 0) no_canvas_score = scores.get(episode_without_canvas.id, 0) # Canvas episode should have higher score (+0.1 boost) assert canvas_score > no_canvas_score @pytest.mark.asyncio async def test_contextual_retrieval_positive_feedback_boost(self, retrieval_service_mocked, episode_test_agent): """ Test contextual retrieval applies positive feedback boost. Verifies: - Episodes with aggregate_feedback_score > 0 get +0.2 boost - Boost applied to relevance score """ base_time = datetime.now(timezone.utc) episode_positive = AgentEpisode( id=f"ctx_pos_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active", aggregate_feedback_score=0.8 # Positive feedback ) episode_neutral = AgentEpisode( id=f"ctx_neu_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active", aggregate_feedback_score=0.0 # Neutral feedback ) retrieval_service_mocked.db.add(episode_positive) retrieval_service_mocked.db.add(episode_neutral) retrieval_service_mocked.db.commit() # Mock semantic search retrieval_service_mocked.lancedb.search.return_value = [] result = await retrieval_service_mocked.retrieve_contextual( agent_id=episode_test_agent.id, current_task="test task", limit=5, require_canvas=False, require_feedback=False ) # Positive feedback episode should rank higher if result["count"] >= 2: scores = {ep["id"]: ep.get("relevance_score", 0) for ep in result["episodes"]} positive_score = scores.get(episode_positive.id, 0) neutral_score = scores.get(episode_neutral.id, 0) # Positive feedback should have higher score (+0.2 boost) assert positive_score > neutral_score @pytest.mark.asyncio async def test_contextual_retrieval_negative_feedback_penalty(self, retrieval_service_mocked, episode_test_agent): """ Test contextual retrieval applies negative feedback penalty. Verifies: - Episodes with aggregate_feedback_score < 0 get -0.3 penalty - Penalty applied to relevance score """ base_time = datetime.now(timezone.utc) episode_negative = AgentEpisode( id=f"ctx_neg_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active", aggregate_feedback_score=-0.7 # Negative feedback ) episode_neutral = AgentEpisode( id=f"ctx_neu2_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active", aggregate_feedback_score=0.0 # Neutral feedback ) retrieval_service_mocked.db.add(episode_negative) retrieval_service_mocked.db.add(episode_neutral) retrieval_service_mocked.db.commit() # Mock semantic search retrieval_service_mocked.lancedb.search.return_value = [] result = await retrieval_service_mocked.retrieve_contextual( agent_id=episode_test_agent.id, current_task="test task", limit=5, require_canvas=False, require_feedback=False ) # Negative feedback episode should rank lower if result["count"] >= 2: scores = {ep["id"]: ep.get("relevance_score", 0) for ep in result["episodes"]} negative_score = scores.get(episode_negative.id, 0) neutral_score = scores.get(episode_neutral.id, 0) # Negative feedback should have lower score (-0.3 penalty) assert negative_score < neutral_score @pytest.mark.asyncio async def test_contextual_retrieval_require_canvas(self, retrieval_service_mocked, episode_test_agent): """ Test contextual retrieval filters by require_canvas. Verifies: - Only episodes with canvas_action_count > 0 returned when require_canvas=True - Episodes without canvas excluded """ base_time = datetime.now(timezone.utc) episode_with_canvas = AgentEpisode( id=f"ctx_req_canvas_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active", canvas_action_count=5 ) episode_without_canvas = AgentEpisode( id=f"ctx_req_no_canvas_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active", canvas_action_count=0 ) retrieval_service_mocked.db.add(episode_with_canvas) retrieval_service_mocked.db.add(episode_without_canvas) retrieval_service_mocked.db.commit() # Mock semantic search retrieval_service_mocked.lancedb.search.return_value = [] result = await retrieval_service_mocked.retrieve_contextual( agent_id=episode_test_agent.id, current_task="test task", limit=10, require_canvas=True, # Only episodes with canvas require_feedback=False ) # Should only return episode with canvas assert result["count"] == 1 assert result["episodes"][0]["id"] == episode_with_canvas.id @pytest.mark.asyncio async def test_contextual_retrieval_require_feedback(self, retrieval_service_mocked, episode_test_agent): """ Test contextual retrieval filters by require_feedback. Verifies: - Only episodes with feedback_ids returned when require_feedback=True - Episodes without feedback excluded """ base_time = datetime.now(timezone.utc) episode_with_feedback = AgentEpisode( id=f"ctx_req_fb_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active", feedback_ids=["fb_1"] ) episode_without_feedback = AgentEpisode( id=f"ctx_req_no_fb_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=1), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) retrieval_service_mocked.db.add(episode_with_feedback) retrieval_service_mocked.db.add(episode_without_feedback) retrieval_service_mocked.db.commit() # Mock semantic search retrieval_service_mocked.lancedb.search.return_value = [] result = await retrieval_service_mocked.retrieve_contextual( agent_id=episode_test_agent.id, current_task="test task", limit=10, require_canvas=False, require_feedback=True # Only episodes with feedback ) # Should only return episode with feedback assert result["count"] == 1 assert result["episodes"][0]["id"] == episode_with_feedback.id @pytest.mark.asyncio async def test_contextual_retrieval_limit(self, retrieval_service_mocked, episode_test_agent): """ Test contextual retrieval respects limit parameter. Verifies: - At most limit results returned - Top-scoring episodes returned """ base_time = datetime.now(timezone.utc) # Create 10 episodes episodes = [] for i in range(10): ep = AgentEpisode( id=f"ctx_lim_{i}_{uuid4().hex[:8]}", agent_id=episode_test_agent.id, tenant_id="default", started_at=base_time - timedelta(hours=i), maturity_at_time="AUTONOMOUS", human_intervention_count=0, outcome="success", status="active" ) episodes.append(ep) retrieval_service_mocked.db.add(ep) retrieval_service_mocked.db.commit() # Mock semantic search retrieval_service_mocked.lancedb.search.return_value = [] result = await retrieval_service_mocked.retrieve_contextual( agent_id=episode_test_agent.id, current_task="test task", limit=5, # Request only 5 require_canvas=False, require_feedback=False ) # Should return at most 5 episodes assert result["count"] <= 5