diff --git a/hindsight-api/hindsight_api/engine/cross_encoder.py b/hindsight-api/hindsight_api/engine/cross_encoder.py index 51957180..7d70c147 100644 --- a/hindsight-api/hindsight_api/engine/cross_encoder.py +++ b/hindsight-api/hindsight_api/engine/cross_encoder.py @@ -163,11 +163,101 @@ class LocalSTCrossEncoder(CrossEncoderModel): else: logger.info("Reranker: local provider initialized (using existing executor)") + def _is_xpc_error(self, error: Exception) -> bool: + """ + Check if an error is an XPC connection error (macOS daemon issue). + + On macOS, long-running daemons can lose XPC connections to system services + when the process is idle for extended periods. + """ + error_str = str(error).lower() + return "xpc_error_connection_invalid" in error_str or "xpc error" in error_str + + def _reinitialize_model_sync(self) -> None: + """ + Clear and reinitialize the cross-encoder model synchronously. + + This is used to recover from XPC errors on macOS where the + PyTorch/MPS backend loses its connection to system services. + """ + logger.warning(f"Reinitializing reranker model {self.model_name} due to backend error") + + # Clear existing model + self._model = None + + # Force garbage collection to free resources + import gc + + import torch + + gc.collect() + + # If using CUDA/MPS, clear the cache + if torch.cuda.is_available(): + torch.cuda.empty_cache() + elif hasattr(torch.backends, "mps") and torch.backends.mps.is_available(): + try: + torch.mps.empty_cache() + except AttributeError: + pass # Method might not exist in all PyTorch versions + + # Reinitialize the model + try: + from sentence_transformers import CrossEncoder + except ImportError: + raise ImportError( + "sentence-transformers is required for LocalSTCrossEncoder. " + "Install it with: pip install sentence-transformers" + ) + + # Determine device based on hardware availability + has_gpu = torch.cuda.is_available() or (hasattr(torch.backends, "mps") and torch.backends.mps.is_available()) + + if has_gpu: + device = None # Let sentence-transformers auto-detect GPU/MPS + else: + device = "cpu" + + self._model = CrossEncoder( + self.model_name, + device=device, + model_kwargs={"low_cpu_mem_usage": False}, + ) + + logger.info("Reranker: local provider reinitialized successfully") + + def _predict_with_recovery(self, pairs: list[tuple[str, str]]) -> list[float]: + """ + Predict with automatic recovery from XPC errors. + + This runs synchronously in the thread pool. + """ + max_retries = 1 + for attempt in range(max_retries + 1): + try: + scores = self._model.predict(pairs, show_progress_bar=False) + return scores.tolist() if hasattr(scores, "tolist") else list(scores) + except Exception as e: + # Check if this is an XPC error (macOS daemon issue) + if self._is_xpc_error(e) and attempt < max_retries: + logger.warning(f"XPC error detected in reranker (attempt {attempt + 1}): {e}") + try: + self._reinitialize_model_sync() + logger.info("Reranker reinitialized successfully, retrying prediction") + continue + except Exception as reinit_error: + logger.error(f"Failed to reinitialize reranker: {reinit_error}") + raise Exception(f"Failed to recover from XPC error: {str(e)}") + else: + # Not an XPC error or out of retries + raise + async def predict(self, pairs: list[tuple[str, str]]) -> list[float]: """ Score query-document pairs for relevance. Uses a dedicated thread pool with limited workers to prevent CPU thrashing. + Automatically recovers from XPC errors on macOS by reinitializing the model. Args: pairs: List of (query, document) tuples to score @@ -180,11 +270,11 @@ class LocalSTCrossEncoder(CrossEncoderModel): # Use dedicated executor - limited workers naturally limits concurrency loop = asyncio.get_event_loop() - scores = await loop.run_in_executor( + return await loop.run_in_executor( LocalSTCrossEncoder._executor, - lambda: self._model.predict(pairs, show_progress_bar=False), + self._predict_with_recovery, + pairs, ) - return scores.tolist() if hasattr(scores, "tolist") else list(scores) class RemoteTEICrossEncoder(CrossEncoderModel): diff --git a/hindsight-api/hindsight_api/engine/embeddings.py b/hindsight-api/hindsight_api/engine/embeddings.py index 3ab3aefc..dddee9d5 100644 --- a/hindsight-api/hindsight_api/engine/embeddings.py +++ b/hindsight-api/hindsight_api/engine/embeddings.py @@ -151,10 +151,75 @@ class LocalSTEmbeddings(Embeddings): self._dimension = self._model.get_sentence_embedding_dimension() logger.info(f"Embeddings: local provider initialized (dim: {self._dimension})") + def _is_xpc_error(self, error: Exception) -> bool: + """ + Check if an error is an XPC connection error (macOS daemon issue). + + On macOS, long-running daemons can lose XPC connections to system services + when the process is idle for extended periods. + """ + error_str = str(error).lower() + return "xpc_error_connection_invalid" in error_str or "xpc error" in error_str + + def _reinitialize_model_sync(self) -> None: + """ + Clear and reinitialize the embedding model synchronously. + + This is used to recover from XPC errors on macOS where the + PyTorch/MPS backend loses its connection to system services. + """ + logger.warning(f"Reinitializing embedding model {self.model_name} due to backend error") + + # Clear existing model + self._model = None + + # Force garbage collection to free resources + import gc + + import torch + + gc.collect() + + # If using CUDA/MPS, clear the cache + if torch.cuda.is_available(): + torch.cuda.empty_cache() + elif hasattr(torch.backends, "mps") and torch.backends.mps.is_available(): + try: + torch.mps.empty_cache() + except AttributeError: + pass # Method might not exist in all PyTorch versions + + # Reinitialize the model (inline version of initialize() but synchronous) + try: + from sentence_transformers import SentenceTransformer + except ImportError: + raise ImportError( + "sentence-transformers is required for LocalSTEmbeddings. " + "Install it with: pip install sentence-transformers" + ) + + # Determine device based on hardware availability + has_gpu = torch.cuda.is_available() or (hasattr(torch.backends, "mps") and torch.backends.mps.is_available()) + + if has_gpu: + device = None # Let sentence-transformers auto-detect GPU/MPS + else: + device = "cpu" + + self._model = SentenceTransformer( + self.model_name, + device=device, + model_kwargs={"low_cpu_mem_usage": False}, + ) + + logger.info("Embeddings: local provider reinitialized successfully") + def encode(self, texts: list[str]) -> list[list[float]]: """ Generate embeddings for a list of texts. + Automatically recovers from XPC errors on macOS by reinitializing the model. + Args: texts: List of text strings to encode @@ -163,8 +228,27 @@ class LocalSTEmbeddings(Embeddings): """ if self._model is None: raise RuntimeError("Embeddings not initialized. Call initialize() first.") - embeddings = self._model.encode(texts, convert_to_numpy=True, show_progress_bar=False) - return [emb.tolist() for emb in embeddings] + + # Try encoding with automatic recovery from XPC errors + max_retries = 1 + for attempt in range(max_retries + 1): + try: + embeddings = self._model.encode(texts, convert_to_numpy=True, show_progress_bar=False) + return [emb.tolist() for emb in embeddings] + except Exception as e: + # Check if this is an XPC error (macOS daemon issue) + if self._is_xpc_error(e) and attempt < max_retries: + logger.warning(f"XPC error detected in embedding generation (attempt {attempt + 1}): {e}") + try: + self._reinitialize_model_sync() + logger.info("Model reinitialized successfully, retrying embedding generation") + continue + except Exception as reinit_error: + logger.error(f"Failed to reinitialize model: {reinit_error}") + raise Exception(f"Failed to recover from XPC error: {str(e)}") + else: + # Not an XPC error or out of retries + raise class RemoteTEIEmbeddings(Embeddings): diff --git a/hindsight-api/tests/test_cross_encoder_xpc_recovery.py b/hindsight-api/tests/test_cross_encoder_xpc_recovery.py new file mode 100644 index 00000000..2bb38c58 --- /dev/null +++ b/hindsight-api/tests/test_cross_encoder_xpc_recovery.py @@ -0,0 +1,148 @@ +""" +Tests for XPC error recovery in LocalSTCrossEncoder. + +This tests the automatic reinitialization of the cross-encoder model when +XPC connection errors occur on macOS (common in long-running daemon processes). +""" + +import asyncio +from unittest.mock import MagicMock, patch + +import pytest + +from hindsight_api.engine.cross_encoder import LocalSTCrossEncoder + + +class TestCrossEncoderXPCErrorRecovery: + """Tests for XPC error detection and recovery in LocalSTCrossEncoder.""" + + @pytest.fixture + def cross_encoder(self): + """Create a LocalSTCrossEncoder instance.""" + return LocalSTCrossEncoder(model_name="cross-encoder/ms-marco-TinyBERT-L-2-v2") + + def test_is_xpc_error_detection(self, cross_encoder): + """Test that XPC errors are correctly detected.""" + # Test various XPC error message formats + xpc_error = Exception("Compiler encountered XPC_ERROR_CONNECTION_INVALID (is the OS shutting down?)") + assert cross_encoder._is_xpc_error(xpc_error) + + xpc_error2 = Exception("XPC error occurred") + assert cross_encoder._is_xpc_error(xpc_error2) + + # Test that non-XPC errors are not detected + normal_error = Exception("Some other error") + assert not cross_encoder._is_xpc_error(normal_error) + + @pytest.mark.asyncio + async def test_predict_with_xpc_recovery(self, cross_encoder): + """Test that predict() recovers from XPC errors by reinitializing.""" + # Initialize the cross-encoder + await cross_encoder.initialize() + + # Track calls to reinitialize + reinit_called = False + original_reinit = cross_encoder._reinitialize_model_sync + + def track_reinit(): + nonlocal reinit_called + reinit_called = True + original_reinit() + + # Track predict attempts + predict_attempts = [] + original_predict = cross_encoder._model.predict + + def mock_predict(*args, **kwargs): + predict_attempts.append(1) + # Only fail on first attempt + if len(predict_attempts) == 1: + raise RuntimeError("Compiler encountered XPC_ERROR_CONNECTION_INVALID (is the OS shutting down?)") + else: + # After reinit: succeed + return original_predict(*args, **kwargs) + + # Mock the initial predict to fail, reinit happens, then new model succeeds + with patch.object(cross_encoder, "_reinitialize_model_sync", side_effect=track_reinit): + with patch.object(cross_encoder._model, "predict", side_effect=mock_predict): + # This should trigger XPC error on first attempt, then recover and succeed + result = await cross_encoder.predict([("query", "document")]) + + # Verify we got a result + assert result is not None + assert len(result) == 1 + assert isinstance(result[0], float) + assert reinit_called # Should have reinitialized + assert len(predict_attempts) >= 1 # At least one attempt was made + + @pytest.mark.asyncio + async def test_predict_fails_on_non_xpc_error(self, cross_encoder): + """Test that predict() does not retry for non-XPC errors.""" + # Initialize the cross-encoder + await cross_encoder.initialize() + + # Create a mock that raises a non-XPC error + def mock_predict(*args, **kwargs): + raise RuntimeError("Some other error") + + # Patch the model's predict method + with patch.object(cross_encoder._model, "predict", side_effect=mock_predict): + # This should fail without retry + with pytest.raises(RuntimeError) as exc_info: + await cross_encoder.predict([("query", "document")]) + + assert "Some other error" in str(exc_info.value) + + @pytest.mark.asyncio + async def test_reinitialize_clears_model(self, cross_encoder): + """Test that _reinitialize_model_sync properly clears and reinits the model.""" + # Initialize the cross-encoder + await cross_encoder.initialize() + + original_model = cross_encoder._model + assert original_model is not None + + # Reinitialize + cross_encoder._reinitialize_model_sync() + + # Model should be reinitialized (new instance) + assert cross_encoder._model is not None + assert cross_encoder._model is not original_model + + # Should still work + result = await cross_encoder.predict([("test query", "test document")]) + assert len(result) == 1 + assert isinstance(result[0], float) + + @pytest.mark.asyncio + async def test_xpc_recovery_exhausts_retries(self, cross_encoder): + """Test that XPC recovery gives up after max retries.""" + # Initialize the cross-encoder + await cross_encoder.initialize() + + # Track reinit calls + reinit_count = 0 + original_reinit = cross_encoder._reinitialize_model_sync + + def track_and_fail_reinit(): + nonlocal reinit_count + reinit_count += 1 + # Call original reinit, but the new model will also be mocked to fail + original_reinit() + # After reinit, patch the new model too + cross_encoder._model.predict = MagicMock( + side_effect=RuntimeError("Compiler encountered XPC_ERROR_CONNECTION_INVALID") + ) + + # Mock that always raises XPC error + cross_encoder._model.predict = MagicMock( + side_effect=RuntimeError("Compiler encountered XPC_ERROR_CONNECTION_INVALID") + ) + + with patch.object(cross_encoder, "_reinitialize_model_sync", side_effect=track_and_fail_reinit): + # Should try once, reinitialize, try again, and fail + with pytest.raises(Exception) as exc_info: + await cross_encoder.predict([("query", "document")]) + + assert "XPC_ERROR_CONNECTION_INVALID" in str(exc_info.value) or "Failed to recover" in str(exc_info.value) + assert reinit_count == 1 # Should have tried to reinitialize once diff --git a/hindsight-api/tests/test_embedding_xpc_recovery.py b/hindsight-api/tests/test_embedding_xpc_recovery.py new file mode 100644 index 00000000..38045146 --- /dev/null +++ b/hindsight-api/tests/test_embedding_xpc_recovery.py @@ -0,0 +1,148 @@ +""" +Tests for XPC error recovery in LocalSTEmbeddings. + +This tests the automatic reinitialization of the embedding model when +XPC connection errors occur on macOS (common in long-running daemon processes). +""" + +import asyncio +from unittest.mock import MagicMock, patch + +import pytest + +from hindsight_api.engine.embeddings import LocalSTEmbeddings + + +class TestXPCErrorRecovery: + """Tests for XPC error detection and recovery in LocalSTEmbeddings.""" + + @pytest.fixture + def embeddings(self): + """Create a LocalSTEmbeddings instance.""" + return LocalSTEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2") + + def test_is_xpc_error_detection(self, embeddings): + """Test that XPC errors are correctly detected.""" + # Test various XPC error message formats + xpc_error = Exception("Compiler encountered XPC_ERROR_CONNECTION_INVALID (is the OS shutting down?)") + assert embeddings._is_xpc_error(xpc_error) + + xpc_error2 = Exception("XPC error occurred") + assert embeddings._is_xpc_error(xpc_error2) + + # Test that non-XPC errors are not detected + normal_error = Exception("Some other error") + assert not embeddings._is_xpc_error(normal_error) + + @pytest.mark.asyncio + async def test_encode_with_xpc_recovery(self, embeddings): + """Test that encode() recovers from XPC errors by reinitializing.""" + # Initialize the embeddings + await embeddings.initialize() + + # Track calls to reinitialize + reinit_called = False + original_reinit = embeddings._reinitialize_model_sync + + def track_reinit(): + nonlocal reinit_called + reinit_called = True + original_reinit() + + # Track encode attempts + encode_attempts = [] + original_encode = embeddings._model.encode + + def mock_encode(*args, **kwargs): + encode_attempts.append(1) + # Only fail on first attempt + if len(encode_attempts) == 1: + raise RuntimeError("Compiler encountered XPC_ERROR_CONNECTION_INVALID (is the OS shutting down?)") + else: + # After reinit: succeed + return original_encode(*args, **kwargs) + + # Mock the initial encode to fail, reinit happens, then new model succeeds + with patch.object(embeddings, "_reinitialize_model_sync", side_effect=track_reinit): + with patch.object(embeddings._model, "encode", side_effect=mock_encode): + # This should trigger XPC error on first attempt, then recover and succeed + result = embeddings.encode(["test text"]) + + # Verify we got a result + assert result is not None + assert len(result) == 1 + assert len(result[0]) > 0 # Should have embedding vector + assert reinit_called # Should have reinitialized + assert len(encode_attempts) >= 1 # At least one attempt was made + + @pytest.mark.asyncio + async def test_encode_fails_on_non_xpc_error(self, embeddings): + """Test that encode() does not retry for non-XPC errors.""" + # Initialize the embeddings + await embeddings.initialize() + + # Create a mock that raises a non-XPC error + def mock_encode(*args, **kwargs): + raise RuntimeError("Some other error") + + # Patch the model's encode method + with patch.object(embeddings._model, "encode", side_effect=mock_encode): + # This should fail without retry + with pytest.raises(RuntimeError) as exc_info: + embeddings.encode(["test text"]) + + assert "Some other error" in str(exc_info.value) + + @pytest.mark.asyncio + async def test_reinitialize_clears_model(self, embeddings): + """Test that _reinitialize_model_sync properly clears and reinits the model.""" + # Initialize the embeddings + await embeddings.initialize() + + original_model = embeddings._model + assert original_model is not None + + # Reinitialize + embeddings._reinitialize_model_sync() + + # Model should be reinitialized (new instance) + assert embeddings._model is not None + assert embeddings._model is not original_model + + # Should still work + result = embeddings.encode(["test"]) + assert len(result) == 1 + assert len(result[0]) > 0 + + @pytest.mark.asyncio + async def test_xpc_recovery_exhausts_retries(self, embeddings): + """Test that XPC recovery gives up after max retries.""" + # Initialize the embeddings + await embeddings.initialize() + + # Track reinit calls + reinit_count = 0 + original_reinit = embeddings._reinitialize_model_sync + + def track_and_fail_reinit(): + nonlocal reinit_count + reinit_count += 1 + # Call original reinit, but the new model will also be mocked to fail + original_reinit() + # After reinit, patch the new model too + embeddings._model.encode = MagicMock( + side_effect=RuntimeError("Compiler encountered XPC_ERROR_CONNECTION_INVALID") + ) + + # Mock that always raises XPC error + embeddings._model.encode = MagicMock( + side_effect=RuntimeError("Compiler encountered XPC_ERROR_CONNECTION_INVALID") + ) + + with patch.object(embeddings, "_reinitialize_model_sync", side_effect=track_and_fail_reinit): + # Should try once, reinitialize, try again, and fail + with pytest.raises(RuntimeError) as exc_info: + embeddings.encode(["test"]) + + assert "XPC_ERROR_CONNECTION_INVALID" in str(exc_info.value) + assert reinit_count == 1 # Should have tried to reinitialize once