"""Tests for temporal validity functionality (MemStrata pattern). Tests bi-temporal ledger with valid_from, valid_to, superseded_by fields and deterministic supersedence checks. """ import time from unittest.mock import MagicMock import pytest from cyberbrain26.data_plane.sqlite_store import SQLiteDatabase from cyberbrain26.knowledge_plane.contracts import KnowledgeDocument, DocumentStatus from cyberbrain26.knowledge_plane.memory_service import MemoryService from cyberbrain26.knowledge_plane.navigation import KnowledgeNavigator def test_knowledge_document_temporal_fields(): """Test that KnowledgeDocument has temporal fields with defaults.""" doc = KnowledgeDocument( document_id="test-doc", source_id="test-source", source_uri="file:///test.txt", content_sha256="abc123", media_type="text/plain", trust="reviewed", sensitivity="personal", status=DocumentStatus.INDEXED, metadata={}, created_at=time.time(), updated_at=time.time(), ) assert hasattr(doc, "valid_from") assert hasattr(doc, "valid_to") assert hasattr(doc, "superseded_by") assert doc.valid_from == 0.0 assert doc.valid_to == 0.0 assert doc.superseded_by == "" def test_knowledge_document_temporal_fields_explicit(): """Test KnowledgeDocument with explicit temporal values.""" now = time.time() doc = KnowledgeDocument( document_id="test-doc", source_id="test-source", source_uri="file:///test.txt", content_sha256="abc123", media_type="text/plain", trust="reviewed", sensitivity="personal", status=DocumentStatus.INDEXED, metadata={}, created_at=now, updated_at=now, valid_from=now, valid_to=now + 3600, superseded_by="newer-doc", ) assert doc.valid_from == now assert doc.valid_to == now + 3600 assert doc.superseded_by == "newer-doc" def test_check_temporal_validity_active_document(): """Test check_temporal_validity returns True for active document.""" navigator = KnowledgeNavigator() doc = KnowledgeDocument( document_id="active-doc", source_id="test-source", source_uri="file:///test.txt", content_sha256="abc123", media_type="text/plain", trust="reviewed", sensitivity="personal", status=DocumentStatus.INDEXED, metadata={}, created_at=time.time(), updated_at=time.time(), valid_from=time.time() - 100, valid_to=0.0, superseded_by="", ) result = navigator.check_temporal_validity(doc) assert result is True def test_check_temporal_validity_superseded_document(): """Test check_temporal_validity returns False for superseded document.""" navigator = KnowledgeNavigator() doc = KnowledgeDocument( document_id="old-doc", source_id="test-source", source_uri="file:///test.txt", content_sha256="abc123", media_type="text/plain", trust="reviewed", sensitivity="personal", status=DocumentStatus.INDEXED, metadata={}, created_at=time.time(), updated_at=time.time(), valid_from=time.time() - 100, valid_to=time.time() - 10, superseded_by="new-doc", ) result = navigator.check_temporal_validity(doc) assert result is False def test_get_valid_documents_filters_superseded(): """Test get_valid_documents filters out superseded documents.""" navigator = KnowledgeNavigator() docs = [ KnowledgeDocument( document_id=f"doc-{i}", source_id="test-source", source_uri=f"file:///test{i}.txt", content_sha256="abc123", media_type="text/plain", trust="reviewed", sensitivity="personal", status=DocumentStatus.INDEXED, metadata={}, created_at=time.time(), updated_at=time.time(), valid_from=time.time() - 100, valid_to=0.0 if i < 2 else time.time() - 10, superseded_by="" if i < 2 else "newer-doc", ) for i in range(4) ] valid = navigator.get_valid_documents(docs) assert len(valid) == 2 assert all(d.superseded_by == "" for d in valid) def test_memory_service_temporal_filter_direct(): """Test that MemoryService temporal filtering works via navigator.""" from unittest.mock import MagicMock from cyberbrain26.data_plane.sqlite_store import SQLiteDatabase mock_db = MagicMock(spec=SQLiteDatabase) service = MemoryService( database=mock_db, audit_log=MagicMock(), ) mock_results = [ {"id": "doc-1", "score": 0.9, "text": "active doc"}, {"id": "doc-2", "score": 0.8, "text": "superseded doc"}, ] current_time = time.time() def mock_get_document(doc_id): if doc_id == "doc-1": return KnowledgeDocument( document_id="doc-1", source_id="test-source", source_uri="file:///test1.txt", content_sha256="abc123", media_type="text/plain", trust="reviewed", sensitivity="personal", status=DocumentStatus.INDEXED, metadata={}, created_at=current_time - 100, updated_at=current_time - 100, valid_from=current_time - 100, valid_to=0.0, superseded_by="", ) elif doc_id == "doc-2": return KnowledgeDocument( document_id="doc-2", source_id="test-source", source_uri="file:///test2.txt", content_sha256="def456", media_type="text/plain", trust="reviewed", sensitivity="personal", status=DocumentStatus.INDEXED, metadata={}, created_at=current_time - 100, updated_at=current_time - 100, valid_from=current_time - 100, valid_to=current_time - 10, superseded_by="doc-3", ) return None service.get_document = mock_get_document # Filter results using navigator temporal validity valid_results = [] for result in mock_results: doc_id = result.get("id") document = service.get_document(doc_id) if document and service.navigator.check_temporal_validity(document, current_time): valid_results.append(result) assert len(valid_results) == 1 assert valid_results[0]["id"] == "doc-1" def test_supersede_document(): """Test supersede_document method in KnowledgeNavigator.""" navigator = KnowledgeNavigator() doc = KnowledgeDocument( document_id="doc-to-supersede", source_id="test-source", source_uri="file:///test.txt", content_sha256="abc123", media_type="text/plain", trust="reviewed", sensitivity="personal", status=DocumentStatus.INDEXED, metadata={}, created_at=time.time(), updated_at=time.time(), ) superseded_doc = navigator.supersede_document(doc, "new-doc-id") assert superseded_doc.document_id == "doc-to-supersede" assert superseded_doc.superseded_by == "new-doc-id" assert superseded_doc.valid_to > 0 assert superseded_doc.valid_to <= time.time() + 1 # approximately now def test_temporal_validity_edge_cases(): """Test edge cases for temporal validity.""" navigator = KnowledgeNavigator() # Document with valid_to but no superseded_by (expired but not superseded) doc1 = KnowledgeDocument( document_id="expired-doc", source_id="test-source", source_uri="file:///test.txt", content_sha256="abc123", media_type="text/plain", trust="reviewed", sensitivity="personal", status=DocumentStatus.INDEXED, metadata={}, created_at=time.time(), updated_at=time.time(), valid_from=time.time() - 100, valid_to=time.time() - 10, superseded_by="", ) # According to MemStrata pattern, superseded_by is the authoritative flag # (no similarity threshold for stale facts) result = navigator.check_temporal_validity(doc1) assert result is True # Not superseded, so valid # Document with superseded_by set but no valid_to doc2 = KnowledgeDocument( document_id="superseded-doc", source_id="test-source", source_uri="file:///test.txt", content_sha256="abc123", media_type="text/plain", trust="reviewed", sensitivity="personal", status=DocumentStatus.INDEXED, metadata={}, created_at=time.time(), updated_at=time.time(), valid_from=time.time() - 100, valid_to=0.0, superseded_by="newer-doc", ) result = navigator.check_temporal_validity(doc2) assert result is False # Superseded, so invalid if __name__ == "__main__": pytest.main([__file__, "-v"])