feat(prompts): add capture_errors option for error tracking

.sampo/changesets/prompts-capture-errors.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Add `capture_errors` option to `Prompts` that reports prompt fetch failures to PostHog error tracking via `capture_exception()` when enabled.

posthog/ai/prompts.py

@@ -94,6 +94,9 @@ class Prompts:
             host='https://us.posthog.com',
         )
 
+        # With error tracking: prompt fetch failures are reported to PostHog
+        prompts = Prompts(posthog, capture_errors=True)
+
         # Fetch with caching and fallback
         template = prompts.get('support-system-prompt', fallback='You are a helpful assistant.')
 
@@ -116,6 +119,7 @@ def __init__(
         project_api_key: Optional[str] = None,
         host: Optional[str] = None,
         default_cache_ttl_seconds: Optional[int] = None,
+        capture_errors: bool = False,
     ):
         """
         Initialize Prompts.
@@ -126,12 +130,16 @@ def __init__(
             project_api_key: Direct project API key (optional if posthog provided)
             host: PostHog host (defaults to app endpoint)
             default_cache_ttl_seconds: Default cache TTL (defaults to 300)
+            capture_errors: If True and a PostHog client is provided, prompt fetch
+                failures are reported to PostHog error tracking via capture_exception().
         """
         self._default_cache_ttl_seconds = (
             default_cache_ttl_seconds or DEFAULT_CACHE_TTL_SECONDS
         )
         self._cache: Dict[PromptCacheKey, CachedPrompt] = {}
         self._has_warned_deprecation = False
+        self._client = posthog
+        self._capture_errors = capture_errors
 
         if posthog is not None:
             self._personal_api_key = getattr(posthog, "personal_api_key", None) or ""
@@ -296,6 +304,8 @@ def _get_internal(
             )
 
         except Exception as error:
+            self._maybe_capture_error(error, name=name, version=version)
+
             prompt_reference = _prompt_reference(name, version)
             # Return stale cache (with warning)
             if cached is not None:
@@ -363,6 +373,27 @@ def clear_cache(
         for key in keys_to_clear:
             self._cache.pop(key, None)
 
+    def _maybe_capture_error(
+        self, error: Exception, *, name: str, version: Optional[int]
+    ) -> None:
+        """Report a prompt fetch error to PostHog error tracking if enabled."""
+        if not self._capture_errors or self._client is None:
+            return
+        if not hasattr(self._client, "capture_exception"):
+            return
+        try:
+            self._client.capture_exception(
+                error,
+                properties={
+                    "$lib_feature": "ai.prompts",
+                    "prompt_name": name,
+                    "prompt_version": version,
+                    "posthog_host": self._host,
+                },
+            )
+        except Exception:
+            log.debug("[PostHog Prompts] Failed to capture exception to error tracking")
+
     def _fetch_prompt_from_api(
         self, name: str, version: Optional[int] = None
     ) -> Dict[str, Any]:

posthog/test/ai/test_prompts.py

@@ -848,6 +848,131 @@ def test_handle_variables_with_dots(self):
         self.assertEqual(result, "Company: Acme")
 
 
+class TestPromptsCaptureErrors(TestPrompts):
+    """Tests for the capture_errors option."""
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_capture_exception_called_on_fetch_failure_with_fallback(
+        self, mock_get_session
+    ):
+        """Should call capture_exception on fetch failure when capture_errors=True."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.side_effect = Exception("Network error")
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog, capture_errors=True)
+
+        result = prompts.get("test-prompt", fallback="fallback prompt", version=3)
+
+        self.assertEqual(result, "fallback prompt")
+        posthog.capture_exception.assert_called_once()
+        captured_exc = posthog.capture_exception.call_args[0][0]
+        self.assertIn("Network error", str(captured_exc))
+
+        properties = posthog.capture_exception.call_args.kwargs["properties"]
+        self.assertEqual(properties["$lib_feature"], "ai.prompts")
+        self.assertEqual(properties["prompt_name"], "test-prompt")
+        self.assertEqual(properties["prompt_version"], 3)
+        self.assertEqual(properties["posthog_host"], "https://us.posthog.com")
+
+    @patch("posthog.ai.prompts._get_session")
+    @patch("posthog.ai.prompts.time.time")
+    def test_capture_exception_called_on_fetch_failure_with_stale_cache(
+        self, mock_time, mock_get_session
+    ):
+        """Should call capture_exception when falling back to stale cache."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.side_effect = [
+            MockResponse(json_data=self.mock_prompt_response),
+            Exception("Network error"),
+        ]
+        mock_time.return_value = 1000.0
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog, capture_errors=True)
+
+        # First call populates cache
+        prompts.get("test-prompt", cache_ttl_seconds=60)
+
+        # Expire cache
+        mock_time.return_value = 1061.0
+
+        # Second call falls back to stale cache
+        result = prompts.get("test-prompt", cache_ttl_seconds=60)
+        self.assertEqual(result, self.mock_prompt_response["prompt"])
+        posthog.capture_exception.assert_called_once()
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_capture_exception_called_when_error_is_raised(self, mock_get_session):
+        """Should call capture_exception even when the error is re-raised (no fallback, no cache)."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.side_effect = Exception("Network error")
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog, capture_errors=True)
+
+        with self.assertRaises(Exception):
+            prompts.get("test-prompt")
+
+        posthog.capture_exception.assert_called_once()
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_no_capture_exception_when_capture_errors_is_false(self, mock_get_session):
+        """Should NOT call capture_exception when capture_errors=False (default)."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.side_effect = Exception("Network error")
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        prompts.get("test-prompt", fallback="fallback prompt")
+
+        posthog.capture_exception.assert_not_called()
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_no_capture_exception_without_client(self, mock_get_session):
+        """Should not error when capture_errors=True but no client provided."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.side_effect = Exception("Network error")
+
+        prompts = Prompts(
+            personal_api_key="phx_test_key",
+            project_api_key="phc_test_key",
+            capture_errors=True,
+        )
+
+        result = prompts.get("test-prompt", fallback="fallback prompt")
+
+        self.assertEqual(result, "fallback prompt")
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_no_capture_exception_on_successful_fetch(self, mock_get_session):
+        """Should NOT call capture_exception on successful fetch."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.return_value = MockResponse(json_data=self.mock_prompt_response)
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog, capture_errors=True)
+
+        prompts.get("test-prompt")
+
+        posthog.capture_exception.assert_not_called()
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_capture_exception_failure_does_not_affect_fallback(self, mock_get_session):
+        """If capture_exception itself throws, the fallback should still be returned."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.side_effect = Exception("Network error")
+
+        posthog = self.create_mock_posthog()
+        posthog.capture_exception.side_effect = Exception("capture failed")
+        prompts = Prompts(posthog, capture_errors=True)
+
+        result = prompts.get("test-prompt", fallback="fallback prompt")
+
+        self.assertEqual(result, "fallback prompt")
+
+
 class TestPromptsClearCache(TestPrompts):
     """Tests for the Prompts.clear_cache() method."""