refactor(ai): share processor privacy and capture policy

posthog/ai/claude_agent_sdk/processor.py

@@ -30,7 +30,11 @@
     format_tool_result_content,
 )
 from posthog.ai.media import ensure_serializable as _ensure_serializable
-from posthog.ai.utils import _capture_ai_event, finalize_ai_content
+from posthog.ai.utils import (
+    _capture_processor_event,
+    finalize_ai_content,
+    with_privacy_mode as _with_privacy_mode,
+)
 from posthog.client import Client
 
 log = logging.getLogger("posthog")
@@ -182,40 +186,21 @@ def _get_distinct_id(
             return str(self._distinct_id)
         return None
 
-    def _with_privacy_mode(self, value: Any) -> Any:
-        if self._privacy_mode or (
-            hasattr(self._client, "privacy_mode") and self._client.privacy_mode
-        ):
-            return None
-        return value
-
     def _capture_event(
         self,
         event: str,
         properties: Dict[str, Any],
         distinct_id: Optional[str] = None,
         groups: Optional[Dict[str, Any]] = None,
     ) -> None:
-        try:
-            if not hasattr(self._client, "capture") or not callable(
-                self._client.capture
-            ):
-                return
-
-            final_properties = {
-                **properties,
-                **self._properties,
-            }
-
-            _capture_ai_event(
-                self._client,
-                event,
-                distinct_id=distinct_id or "unknown",
-                properties=final_properties,
-                groups=groups if groups is not None else self._groups,
-            )
-        except Exception as e:
-            log.debug(f"Failed to capture PostHog event: {e}")
+        _capture_processor_event(
+            self._client,
+            event,
+            properties,
+            default_properties=self._properties,
+            distinct_id=distinct_id,
+            groups=groups if groups is not None else self._groups,
+        )
 
     async def query(
         self,
@@ -252,11 +237,9 @@ async def query(
         distinct_id_override = posthog_distinct_id or self._distinct_id
         trace_id = posthog_trace_id or str(uuid.uuid4())
         extra_props = posthog_properties or {}
-        privacy = (
-            posthog_privacy_mode
-            if posthog_privacy_mode is not None
-            else self._privacy_mode
-        )
+        # Per-call privacy can enable redaction, but cannot disable the
+        # processor-level setting. This preserves the existing precedence.
+        privacy = self._privacy_mode or posthog_privacy_mode is True
         groups = posthog_groups or self._groups
 
         # Ensure partial messages are enabled for per-generation tracking
@@ -434,20 +417,16 @@ def _emit_generation(
         }
 
         if input_messages is not None:
-            properties["$ai_input"] = (
-                None
-                if privacy
-                else self._with_privacy_mode(
-                    finalize_ai_content(input_messages, self._client)
-                )
+            properties["$ai_input"] = _with_privacy_mode(
+                self._client,
+                privacy,
+                finalize_ai_content(input_messages, self._client),
             )
         if output_choices is not None:
-            properties["$ai_output_choices"] = (
-                None
-                if privacy
-                else self._with_privacy_mode(
-                    finalize_ai_content(output_choices, self._client)
-                )
+            properties["$ai_output_choices"] = _with_privacy_mode(
+                self._client,
+                privacy,
+                finalize_ai_content(output_choices, self._client),
             )
 
         if gen.cache_read_input_tokens:
@@ -503,20 +482,16 @@ def _emit_generation_from_result(
         }
 
         if input_messages is not None:
-            properties["$ai_input"] = (
-                None
-                if privacy
-                else self._with_privacy_mode(
-                    finalize_ai_content(input_messages, self._client)
-                )
+            properties["$ai_input"] = _with_privacy_mode(
+                self._client,
+                privacy,
+                finalize_ai_content(input_messages, self._client),
             )
         if output_choices is not None:
-            properties["$ai_output_choices"] = (
-                None
-                if privacy
-                else self._with_privacy_mode(
-                    finalize_ai_content(output_choices, self._client)
-                )
+            properties["$ai_output_choices"] = _with_privacy_mode(
+                self._client,
+                privacy,
+                finalize_ai_content(output_choices, self._client),
             )
 
         cache_read = usage.get("cache_read_input_tokens", 0)
@@ -561,9 +536,7 @@ def _emit_tool_span(
             **extra_props,
         }
 
-        if not privacy and not (
-            hasattr(self._client, "privacy_mode") and self._client.privacy_mode
-        ):
+        if _with_privacy_mode(self._client, privacy, True):
             properties["$ai_input_state"] = finalize_ai_content(
                 _ensure_serializable(block.input), self._client
             )

posthog/ai/openai_agents/processor.py

@@ -22,7 +22,11 @@
 from posthog import setup
 from posthog.ai.media import ensure_serializable as _ensure_serializable
 from posthog.ai.sanitization import _full_ai_capture_enabled, _placeholder
-from posthog.ai.utils import _capture_ai_event, finalize_ai_content
+from posthog.ai.utils import (
+    _capture_processor_event,
+    finalize_ai_content,
+    with_privacy_mode as _with_privacy_mode,
+)
 from posthog.client import Client
 
 log = logging.getLogger("posthog")
@@ -119,11 +123,7 @@ def _get_distinct_id(self, trace: Optional[Trace]) -> Optional[str]:
 
     def _with_privacy_mode(self, value: Any) -> Any:
         """Apply privacy mode redaction if enabled."""
-        if self._privacy_mode or (
-            hasattr(self._client, "privacy_mode") and self._client.privacy_mode
-        ):
-            return None
-        return value
+        return _with_privacy_mode(self._client, self._privacy_mode, value)
 
     def _evict_stale_entries(self) -> None:
         """Evict oldest entries if dicts exceed max size to prevent unbounded growth."""
@@ -159,34 +159,15 @@ def _capture_event(
         properties: Dict[str, Any],
         distinct_id: Optional[str] = None,
     ) -> None:
-        """Capture an event to PostHog with error handling.
-
-        Args:
-            distinct_id: The resolved distinct ID. When the user didn't provide
-                one, callers should pass ``user_distinct_id or fallback_id``
-                (matching the langchain/openai pattern) and separately set
-                ``$process_person_profile`` in properties.
-        """
-        try:
-            if not hasattr(self._client, "capture") or not callable(
-                self._client.capture
-            ):
-                return
-
-            final_properties = {
-                **properties,
-                **self._properties,
-            }
-
-            _capture_ai_event(
-                self._client,
-                event,
-                distinct_id=distinct_id or "unknown",
-                properties=final_properties,
-                groups=self._groups,
-            )
-        except Exception as e:
-            log.debug(f"Failed to capture PostHog event: {e}")
+        """Capture an event without allowing telemetry failures to escape."""
+        _capture_processor_event(
+            self._client,
+            event,
+            properties,
+            default_properties=self._properties,
+            distinct_id=distinct_id,
+            groups=self._groups,
+        )
 
     def on_trace_start(self, trace: Trace) -> None:
         """Called when a new trace begins. Stores metadata for spans; the $ai_trace event is emitted in on_trace_end."""

posthog/ai/utils.py

@@ -1,3 +1,4 @@
+import logging
 import time
 import uuid
 from typing import Any, Callable, Dict, List, Optional, Tuple, cast
@@ -70,6 +71,32 @@ def _capture_ai_event(ph_client, event: str, **kwargs):
     return ph_client.capture(event=event, **kwargs)
 
 
+def _capture_processor_event(
+    ph_client: Any,
+    event: str,
+    properties: Dict[str, Any],
+    *,
+    default_properties: Optional[Dict[str, Any]] = None,
+    distinct_id: Optional[str] = None,
+    groups: Optional[Dict[str, Any]] = None,
+) -> None:
+    """Apply the shared capture policy used by AI SDK processors."""
+    try:
+        capture = getattr(ph_client, "capture", None)
+        if not callable(capture):
+            return
+
+        _capture_ai_event(
+            ph_client,
+            event,
+            distinct_id=distinct_id or "unknown",
+            properties={**properties, **(default_properties or {})},
+            groups=groups,
+        )
+    except Exception as exc:
+        logging.getLogger("posthog").debug("Failed to capture PostHog event: %s", exc)
+
+
 def serialize_raw_usage(raw_usage: Any) -> Optional[Dict[str, Any]]:
     """
     Convert raw provider usage objects to JSON-serializable dicts.
@@ -709,7 +736,7 @@ def finalize_ai_content(value: Any, ph_client: Any = None) -> Any:
 
 
 def with_privacy_mode(ph_client: PostHogClient, privacy_mode: bool, value: Any):
-    if ph_client.privacy_mode or privacy_mode:
+    if getattr(ph_client, "privacy_mode", False) or privacy_mode:
         return None
     return value
 

posthog/test/ai/claude_agent_sdk/test_processor.py

@@ -2,6 +2,7 @@
 
 import base64
 import logging
+from types import SimpleNamespace
 from typing import Any, Dict, List, Optional
 from unittest.mock import MagicMock, patch
 
@@ -457,6 +458,67 @@ async def test_trace_emits_error_status(self, processor, mock_client):
 
 
 class TestPrivacyMode:
+    @pytest.mark.asyncio
+    async def test_client_privacy_mode_redacts_generation_input(self, mock_client):
+        mock_client.privacy_mode = True
+        proc = PostHogClaudeAgentProcessor(client=mock_client, distinct_id="user")
+        messages = [_make_message_start(), _make_message_stop()]
+
+        with patch(
+            "posthog.ai.claude_agent_sdk.processor.original_query",
+            side_effect=lambda **kw: _fake_query(messages),
+        ):
+            async for _ in proc.query(prompt="secret", options=ClaudeAgentOptions()):
+                pass
+
+        properties = mock_client.capture.call_args.kwargs["properties"]
+        assert properties["$ai_input"] is None
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("processor_privacy", "call_privacy"),
+        [(False, True), (True, False)],
+    )
+    async def test_per_call_privacy_mode_cannot_disable_processor_mode(
+        self, mock_client, processor_privacy, call_privacy
+    ):
+        proc = PostHogClaudeAgentProcessor(
+            client=mock_client,
+            distinct_id="user",
+            privacy_mode=processor_privacy,
+        )
+        messages = [_make_message_start(), _make_message_stop()]
+
+        with patch(
+            "posthog.ai.claude_agent_sdk.processor.original_query",
+            side_effect=lambda **kw: _fake_query(messages),
+        ):
+            async for _ in proc.query(
+                prompt="secret",
+                options=ClaudeAgentOptions(),
+                posthog_privacy_mode=call_privacy,
+            ):
+                pass
+
+        properties = mock_client.capture.call_args.kwargs["properties"]
+        assert properties["$ai_input"] is None
+
+    @pytest.mark.asyncio
+    async def test_client_without_privacy_mode_captures_content(self):
+        client = SimpleNamespace(capture=MagicMock())
+        proc = PostHogClaudeAgentProcessor(client=client, distinct_id="user")
+        messages = [_make_message_start(), _make_message_stop()]
+
+        with patch(
+            "posthog.ai.claude_agent_sdk.processor.original_query",
+            side_effect=lambda **kw: _fake_query(messages),
+        ):
+            async for _ in proc.query(prompt="visible", options=ClaudeAgentOptions()):
+                pass
+
+        properties = client.capture.call_args.kwargs["properties"]
+        assert properties["$ai_input"] == [{"role": "user", "content": "visible"}]
+
     @pytest.mark.asyncio
     async def test_privacy_mode_redacts_tool_input(self, mock_client):
         proc = PostHogClaudeAgentProcessor(
@@ -596,12 +658,59 @@ async def test_no_distinct_id_sets_process_person_profile_false(self, mock_clien
             "posthog.ai.claude_agent_sdk.processor.original_query",
             side_effect=lambda **kw: _fake_query(messages),
         ):
-            async for _ in proc.query(prompt="Hi", options=ClaudeAgentOptions()):
+            async for _ in proc.query(
+                prompt="Hi",
+                options=ClaudeAgentOptions(),
+                posthog_trace_id="trace-fallback",
+            ):
                 pass
 
         for call in mock_client.capture.call_args_list:
             props = call.kwargs.get("properties") or call[1].get("properties")
             assert props.get("$process_person_profile") is False
+            assert call.kwargs["distinct_id"] == "trace-fallback"
+
+
+class TestCapturePolicy:
+    def test_explicit_groups_override_preserves_empty_groups(self, mock_client):
+        proc = PostHogClaudeAgentProcessor(
+            client=mock_client,
+            groups={"company": "default"},
+        )
+
+        proc._capture_event("$ai_trace", {}, groups={})
+
+        assert mock_client.capture.call_args.kwargs["groups"] == {}
+
+    def test_default_properties_keep_existing_precedence(self, mock_client):
+        proc = PostHogClaudeAgentProcessor(
+            client=mock_client,
+            properties={"environment": "processor"},
+        )
+
+        proc._capture_event(
+            "$ai_trace",
+            {"environment": "event", "$ai_trace_id": "trace-id"},
+        )
+
+        assert mock_client.capture.call_args.kwargs["properties"] == {
+            "environment": "processor",
+            "$ai_trace_id": "trace-id",
+        }
+
+    def test_client_without_capture_capability_is_ignored(self):
+        proc = PostHogClaudeAgentProcessor(client=object())
+
+        proc._capture_event("$ai_trace", {})
+
+    def test_capture_errors_are_logged_and_suppressed(self, mock_client, caplog):
+        mock_client.capture.side_effect = RuntimeError("capture failed")
+        proc = PostHogClaudeAgentProcessor(client=mock_client)
+
+        with caplog.at_level(logging.DEBUG, logger="posthog"):
+            proc._capture_event("$ai_trace", {})
+
+        assert "Failed to capture PostHog event: capture failed" in caplog.text
 
 
 class TestCustomProperties: