Implement intent classification and Turn message update tasks with error handling

Author: erikh360

src/tasks.py

@@ -1,27 +1,149 @@
 """Celery tasks for async processing."""
 
 import logging
+from pathlib import Path
+
+from celery import chain
 
 from src.celery_app import celery_app
+from src.intent_classifier import IntentClassifier
+from src.turn_client import TurnAPIClient, TurnAPIServerError
 
 logger = logging.getLogger(__name__)
 
+# Load classifier once at module level for efficiency
+# Model is large and we're classifying thousands of messages
+_DATA_DIR = Path(__file__).parent / "data"
+_EMBEDDINGS_PATH = _DATA_DIR / "intent_embeddings.json"
+_NLU_PATH = _DATA_DIR / "nlu.yaml"
+
+try:
+    classifier = IntentClassifier(
+        embeddings_path=_EMBEDDINGS_PATH,
+        nlu_path=_NLU_PATH,
+    )
+    logger.info("Intent classifier loaded successfully")
+except Exception as e:
+    logger.error(f"Failed to load intent classifier: {e}")
+    classifier = None
+
+
+@celery_app.task(
+    acks_late=True,  # Only ack after task completes (safer for worker crashes)
+)
+def classify_turn_message(message_id: str, message_text: str) -> dict:
+    """
+    Classify incoming message text.
+
+    This task:
+    1. Classifies the message text using the IntentClassifier
+    2. Does NOT call the Turn API (no label update here)
+
+    Args:
+        message_id: Turn message ID to classify
+        message_text: Message text to classify
+
+    Returns:
+        dict: Classification result with keys:
+            - message_id: The Turn message ID
+            - intent: Classified intent label
+            - confidence: Classification confidence score
+            - error: Optional error string
+
+    Raises:
+        ValueError: Classifier not loaded or invalid input
+    """
+    # Validate classifier is loaded
+    if classifier is None:
+        error_msg = "Intent classifier not loaded. Cannot classify message."
+        logger.error(f"{error_msg} Message ID: {message_id}")
+        raise ValueError(error_msg)
+
+    try:
+        # Step 1: Classify the message
+        logger.info(f"Classifying message {message_id}: {message_text[:100]}...")
+        intent, confidence = classifier.classify(message_text)
+        logger.info(
+            f"Classified message {message_id} as '{intent}' with confidence {confidence:.4f}"
+        )
+
+        return {
+            "message_id": message_id,
+            "intent": intent,
+            "confidence": confidence,
+        }
+    except Exception as e:
+        # Log other errors but don't retry (likely classification or client errors)
+        logger.error(
+            f"Error processing message {message_id}: {e}",
+            exc_info=True,
+        )
+        raise
+
 
-@celery_app.task(name="src.tasks.hello_world")
-def hello_world(name: str = "World") -> dict:
+@celery_app.task(
+    acks_late=True,  # Only ack after task completes (safer for worker crashes)
+    autoretry_for=(TurnAPIServerError,),  # Retry on 5xx server errors
+    max_retries=3,
+    retry_backoff=True,  # Exponential backoff
+    retry_backoff_max=600,  # Max 10 minutes between retries
+    retry_jitter=True,  # Add randomness to prevent thundering herd
+)
+def update_turn_message_label(classification: dict) -> dict:
     """
-    Simple hello world task for testing Celery connection.
+    Update Turn message label.
+
+    This task:
+    1. Updates the message label in Turn via the API
+    2. Automatically retries on server errors (5xx) with exponential backoff
+    3. Does NOT retry on client errors (4xx) - bad data won't improve on retry
 
     Args:
-        name: Name to greet (default: "World")
+        classification: dict with keys message_id and intent
 
     Returns:
-        dict: Status message with greeting
+        dict: Update result with keys:
+            - message_id: The Turn message ID
+            - intent: Intent label sent to Turn
+            - turn_response: Response from Turn API
+
+    Raises:
+        TurnAPIServerError: Turn API server error (5xx) - will auto-retry
+        TurnAPIClientError: Turn API client error (4xx) - will NOT retry
     """
-    message = f"Hello, {name}!"
-    logger.info(f"Hello world task executed: {message}")
-    return {"status": "success", "message": message}
+    message_id = classification["message_id"]
+    intent = classification["intent"]
+
+    try:
+        turn_client = TurnAPIClient()
+        turn_response = turn_client.update_message_label(message_id, intent)
+
+        logger.info(f"Successfully updated Turn message {message_id} with label: {intent}")
+
+        return {
+            "message_id": message_id,
+            "intent": intent,
+            "turn_response": turn_response,
+        }
+    except TurnAPIServerError:
+        # Re-raise server errors to trigger retry
+        logger.warning(
+            f"Turn API server error for message {message_id}. "
+            f"Retry {update_turn_message_label.request.retries}/{update_turn_message_label.max_retries}"
+        )
+        raise
+    except Exception as e:
+        # Log other errors but don't retry (likely client errors)
+        logger.error(
+            f"Error updating Turn message {message_id}: {e}",
+            exc_info=True,
+        )
+        raise
 
 
-# Future tasks will be added here:
-# - classify_and_update_turn: Main classification task that calls Turn API
+def build_classify_and_update_chain(message_id: str, message_text: str):
+    """Build a Celery chain that classifies then updates the Turn label."""
+    return chain(
+        classify_turn_message.s(message_id, message_text),
+        update_turn_message_label.s(),
+    )

tests/test_tasks.py

@@ -1,32 +1,20 @@
 """Tests for Celery tasks."""
 
+import pytest
+
 from src.celery_app import celery_app
-from src.tasks import hello_world
+from src.tasks import (
+    build_classify_and_update_chain,
+    classify_turn_message,
+    update_turn_message_label,
+)
+from src.turn_client import TurnAPIClientError, TurnAPIServerError
 
 # Enable eager mode for all tests in this module (run tasks synchronously)
 celery_app.conf.task_always_eager = True
 celery_app.conf.task_eager_propagates = True
 
 
-class TestHelloWorldTask:
-    """Tests for the hello_world task."""
-
-    def test_hello_world_default(self):
-        """Test hello world task with default parameter."""
-        result = hello_world()
-        assert result == {"status": "success", "message": "Hello, World!"}
-
-    def test_hello_world_custom_name(self):
-        """Test hello world task with custom name."""
-        result = hello_world("Celery")
-        assert result == {"status": "success", "message": "Hello, Celery!"}
-
-    def test_hello_world_synchronous(self):
-        """Test hello world task called synchronously."""
-        result = hello_world("Testing")
-        assert result == {"status": "success", "message": "Hello, Testing!"}
-
-
 class TestCeleryConfiguration:
     """Tests for Celery app configuration."""
 
@@ -45,3 +33,133 @@ def test_task_serializer_is_json(self):
         """Test that task serializer is set to JSON."""
         assert celery_app.conf.task_serializer == "json"
         assert "json" in celery_app.conf.accept_content
+
+
+class TestClassifyTurnMessageTask:
+    """Tests for the classify_turn_message task."""
+
+    def test_classify_success(self, mocker):
+        """Test successful classification."""
+        # Mock classifier
+        mock_classifier = mocker.Mock()
+        mock_classifier.classify.return_value = ("compliment", 0.95)
+        mocker.patch("src.tasks.classifier", mock_classifier)
+
+        # Execute task
+        result = classify_turn_message("msg-123", "Thank you so much!")
+
+        # Verify classifier was called
+        mock_classifier.classify.assert_called_once_with("Thank you so much!")
+
+        # Verify result
+        assert result["message_id"] == "msg-123"
+        assert result["intent"] == "compliment"
+        assert result["confidence"] == 0.95
+
+    def test_classify_classifier_not_loaded(self, mocker):
+        """Test error when classifier is not loaded."""
+        # Mock classifier as None (not loaded)
+        mocker.patch("src.tasks.classifier", None)
+
+        # Execute task - should raise ValueError
+        with pytest.raises(ValueError, match="Intent classifier not loaded"):
+            classify_turn_message("msg-789", "Hello world")
+
+    def test_classify_unclassified_intent(self, mocker):
+        """Test handling of unclassified intent."""
+        # Mock classifier to return Unclassified
+        mock_classifier = mocker.Mock()
+        mock_classifier.classify.return_value = ("Unclassified", 0.42)
+        mocker.patch("src.tasks.classifier", mock_classifier)
+
+        # Execute task
+        result = classify_turn_message("msg-unclass", "asdfghjkl")
+
+        # Verify result
+        assert result["intent"] == "Unclassified"
+        assert result["confidence"] == 0.42
+
+
+class TestUpdateTurnMessageLabelTask:
+    """Tests for the update_turn_message_label task."""
+
+    def test_update_turn_label_success(self, mocker, monkeypatch):
+        """Test successful Turn API update."""
+        # Set up Turn API env vars
+        monkeypatch.setenv("TURN_API_BASE_URL", "https://api.turn.io")
+        monkeypatch.setenv("TURN_API_TOKEN", "test-token")
+
+        # Mock Turn API client
+        mock_turn_client = mocker.Mock()
+        mock_turn_client.update_message_label.return_value = {
+            "status": "success",
+            "message_id": "msg-123",
+        }
+        mocker.patch("src.tasks.TurnAPIClient", return_value=mock_turn_client)
+
+        # Execute task
+        result = update_turn_message_label(
+            {"message_id": "msg-123", "intent": "compliment"}
+        )
+
+        # Verify Turn API was called
+        mock_turn_client.update_message_label.assert_called_once_with(
+            "msg-123", "compliment"
+        )
+
+        # Verify result
+        assert result["message_id"] == "msg-123"
+        assert result["intent"] == "compliment"
+        assert result["turn_response"]["status"] == "success"
+
+    def test_update_turn_label_client_error(self, mocker, monkeypatch):
+        """Test that 4xx client errors do NOT trigger retry."""
+        # Set up Turn API env vars
+        monkeypatch.setenv("TURN_API_BASE_URL", "https://api.turn.io")
+        monkeypatch.setenv("TURN_API_TOKEN", "test-token")
+
+        # Mock Turn API client to raise client error (4xx)
+        mock_turn_client = mocker.Mock()
+        mock_turn_client.update_message_label.side_effect = TurnAPIClientError(
+            "Turn API client error (HTTP 400): Invalid label"
+        )
+        mocker.patch("src.tasks.TurnAPIClient", return_value=mock_turn_client)
+
+        # Execute task - should raise TurnAPIClientError without retry
+        with pytest.raises(TurnAPIClientError, match="Turn API client error"):
+            update_turn_message_label({"message_id": "msg-400", "intent": "greeting"})
+
+        # Verify Turn API was called only once (no retry)
+        mock_turn_client.update_message_label.assert_called_once()
+
+    def test_update_turn_label_server_error_retry(self, mocker, monkeypatch):
+        """Test that 5xx server errors trigger automatic retry."""
+        # Set up Turn API env vars
+        monkeypatch.setenv("TURN_API_BASE_URL", "https://api.turn.io")
+        monkeypatch.setenv("TURN_API_TOKEN", "test-token")
+
+        # Mock Turn API client to raise server error (5xx)
+        mock_turn_client = mocker.Mock()
+        mock_turn_client.update_message_label.side_effect = TurnAPIServerError(
+            "Turn API server error (HTTP 500): Internal server error"
+        )
+        mocker.patch("src.tasks.TurnAPIClient", return_value=mock_turn_client)
+
+        # Execute task - should raise TurnAPIServerError (retry will be handled by Celery)
+        with pytest.raises(TurnAPIServerError, match="Turn API server error"):
+            update_turn_message_label({"message_id": "msg-500", "intent": "question"})
+
+        # Verify Turn API was called
+        mock_turn_client.update_message_label.assert_called_once()
+
+
+class TestClassifyAndUpdateChain:
+    """Tests for the classify and update chain helper."""
+
+    def test_build_chain_signatures(self):
+        """Ensure chain is composed of classify then update tasks."""
+        chain_sig = build_classify_and_update_chain("msg-123", "Hello there")
+
+        assert len(chain_sig.tasks) == 2
+        assert chain_sig.tasks[0].name == classify_turn_message.name
+        assert chain_sig.tasks[1].name == update_turn_message_label.name