feat: capture_ai public beta

.sampo/changesets/public-capture-ai-beta.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Public beta `capture_ai`: AI events on the dedicated AI endpoint with the event UUID returned; new `enable_full_ai_capture` flag (old private flags kept as deprecated aliases).

posthog/__init__.py

@@ -401,12 +401,10 @@ def get_tags() -> Dict[str, Any]:
 # Capture wire protocol for the global client. None defers to POSTHOG_CAPTURE_MODE
 # then CaptureMode.V0. See posthog.capture_mode.CaptureMode.
 capture_mode = None  # type: Optional[CaptureMode]
-# Internal, no stability guarantees. `_use_ai_lane` routes AI SDK wrapper events
-# through the dedicated AI capture lane; `_enable_multimodal_capture` additionally
-# skips media redaction (and implies the lane). Module attributes so the lazily
-# auto-instantiated default client can be configured without constructing it.
-# Like `debug`/`disabled`, these are authoritative for the default client:
-# `setup()` re-syncs them onto it on every call, overwriting direct assignments.
+# Routes AI SDK wrapper events through the dedicated AI capture lane, skips
+# truncation, and passes media unredacted. `privacy_mode` always wins.
+enable_full_ai_capture = False  # type: bool
+# Deprecated aliases for `enable_full_ai_capture`.
 _use_ai_lane = False  # type: bool
 _enable_multimodal_capture = False  # type: bool
 
@@ -503,6 +501,41 @@ def capture(event: str, **kwargs: Unpack[OptionalCaptureArgs]) -> Optional[str]:
     return _proxy("capture", event, **kwargs)
 
 
+def capture_ai(event: str, **kwargs: Unpack[OptionalCaptureArgs]) -> Optional[str]:
+    """
+    Capture an AI event on the dedicated AI capture endpoint.
+
+    Beta: the signature is stable; operational limits (per-event size cap,
+    batching, endpoint) may change without notice.
+
+    Takes the same arguments and returns the same value as `capture()`: the
+    event UUID, or None when the event was not admitted (disabled client, or
+    dropped by `before_send`). The event is delivered on an isolated queue
+    with its own consumer pool and a higher per-event size cap, posting to
+    the dedicated AI ingestion endpoint. The payload is sent as given — no
+    redaction or truncation is applied here.
+
+    Args:
+        event: The event name, normally one of the `$ai_*` event names.
+        **kwargs: Same optional arguments as `capture()`.
+
+    Examples:
+        ```python
+        from posthog import capture_ai
+
+        uuid = capture_ai(
+            "$ai_generation",
+            distinct_id="user_123",
+            properties={"$ai_model": "gpt-5"},
+        )
+        ```
+
+    Category:
+        Events
+    """
+    return _proxy("capture_ai", event, **kwargs)
+
+
 def set(**kwargs: Unpack[OptionalSetArgs]) -> Optional[str]:
     """
     Set properties on a user record.
@@ -1237,8 +1270,11 @@ def setup() -> Client:
     default_client.debug = debug
     default_client.privacy_mode = bool(privacy_mode)
     default_client._set_before_send(before_send)
-    default_client._use_ai_lane = bool(_use_ai_lane)
-    default_client._enable_multimodal_capture = bool(_enable_multimodal_capture)
+    default_client.enable_full_ai_capture = (
+        bool(enable_full_ai_capture)
+        or bool(_use_ai_lane)
+        or bool(_enable_multimodal_capture)
+    )
     # Metrics config is consumed lazily on first `.metrics` access, so late
     # module-attr assignment (e.g. a Django ready() hook running after something
     # already forced setup()) still applies until the metrics API is first used.

posthog/ai/openai_agents/processor.py

@@ -21,7 +21,7 @@
 
 from posthog import setup
 from posthog.ai.media import ensure_serializable as _ensure_serializable
-from posthog.ai.sanitization import _multimodal_capture_enabled, _placeholder
+from posthog.ai.sanitization import _full_ai_capture_enabled, _placeholder
 from posthog.ai.utils import _capture_ai_event, finalize_ai_content
 from posthog.client import Client
 
@@ -801,7 +801,7 @@ def _handle_audio_span(
             if span_type == "transcription":
                 audio_input: Any = (
                     span_data.input
-                    if _multimodal_capture_enabled(self._client)
+                    if _full_ai_capture_enabled(self._client)
                     else _placeholder(
                         getattr(span_data, "input_format", None) or "audio"
                     )

posthog/ai/sanitization.py

@@ -32,10 +32,10 @@
 _MEDIA_URL_CONTAINER_KEYS = {"image_url", "imageUrl", "video_url", "videoUrl"}
 
 
-def _multimodal_capture_enabled(ph_client: Any = None) -> bool:
-    """Media passthrough: on only when the client opted into multimodal capture."""
+def _full_ai_capture_enabled(ph_client: Any = None) -> bool:
+    """Full AI capture: no truncation and media passthrough, on only when the client opted in."""
     return (
-        getattr(ph_client, "_enable_multimodal_capture", False) is True
+        getattr(ph_client, "enable_full_ai_capture", False) is True
     )  # is True: tolerate unspecced Mock clients whose auto-generated attrs are truthy
 
 
@@ -131,7 +131,7 @@ def _redact_string(
 def redact_media(
     value: Any, max_string_len: Optional[int] = None, ph_client: Any = None
 ) -> Any:
-    passthrough = _multimodal_capture_enabled(ph_client)
+    passthrough = _full_ai_capture_enabled(ph_client)
     stack: set = set()
 
     def walk(

posthog/ai/utils.py

@@ -4,7 +4,7 @@
 
 from posthog import get_tags, identify_context, new_context, tag, contexts
 from posthog.ai.gateway import warn_if_posthog_ai_gateway
-from posthog.ai.sanitization import _multimodal_capture_enabled, redact_media
+from posthog.ai.sanitization import _full_ai_capture_enabled, redact_media
 from posthog.ai.sanitization import sanitize_messages  # noqa: F401 -- re-exported for back-compat
 from posthog.ai.types import FormattedMessage, StreamingEventData, TokenUsage
 from posthog.client import Client as PostHogClient
@@ -57,21 +57,14 @@ def _get_tokens_source(
 
 
 def _ai_lane_enabled(ph_client) -> bool:
-    """The client's private, unstable AI-lane opt-in; multimodal implies it."""
-    # `is True` tolerates unspecced Mock clients whose auto-generated attrs are truthy.
-    opted_in = getattr(ph_client, "_use_ai_lane", False) is True
-    return opted_in or _multimodal_capture_enabled(ph_client)
+    """The client's full-AI-capture opt-in routes wrapper events onto the AI lane."""
+    return _full_ai_capture_enabled(ph_client)
 
 
 def _capture_ai_event(ph_client, event: str, **kwargs):
-    """Capture a wrapper-emitted AI event.
-
-    When the client opted into the AI lane, the event rides it via
-    `_capture_ai`. Otherwise — including duck-typed client-likes without the
-    lane — events keep the plain `capture()` path they have today.
-    """
+    """Capture a wrapper-emitted AI event, falling back to `capture()` for duck-typed clients without `capture_ai`."""
     if _ai_lane_enabled(ph_client):
-        capture_ai = getattr(ph_client, "_capture_ai", None)
+        capture_ai = getattr(ph_client, "capture_ai", None)
         if callable(capture_ai):
             return capture_ai(event=event, **kwargs)
     return ph_client.capture(event=event, **kwargs)

posthog/client.py

@@ -633,6 +633,7 @@ def __init__(
         capture_compression: Optional[Union[CaptureCompression, str]] = None,
         secret_key=None,
         metrics: Optional[dict] = None,
+        enable_full_ai_capture=False,
         _use_ai_lane=False,
         _enable_multimodal_capture=False,
     ):
@@ -693,6 +694,11 @@ def __init__(
                 captured exceptions. Defaults to the current working directory.
             privacy_mode: For AI observability, capture usage metadata without
                 prompt inputs or outputs.
+            enable_full_ai_capture: Route PostHog AI wrapper events through
+                the dedicated AI capture endpoint and capture full AI content:
+                skips string truncation and passes media (base64/data URIs)
+                through unredacted. ``privacy_mode`` always wins. Defaults to
+                False.
             before_send: Optional callback that can modify or drop events before
                 upload. Return ``None`` to drop an event.
             flag_fallback_cache_url: Optional feature flag fallback cache URL,
@@ -814,12 +820,12 @@ def __init__(
         self._metrics_config = metrics
         self._metrics: Optional[PostHogMetrics] = None
         self._metrics_lock = threading.Lock()
-        # Internal, no stability guarantees. `_use_ai_lane` routes all AI SDK
-        # wrapper events through the dedicated AI lane; `_enable_multimodal_capture`
-        # additionally skips media redaction (and implies the lane). Both are
-        # read per event by wrapper-layer code, never by `capture()` itself.
-        self._use_ai_lane = bool(_use_ai_lane)
-        self._enable_multimodal_capture = bool(_enable_multimodal_capture)
+        # `_use_ai_lane` / `_enable_multimodal_capture` are deprecated aliases.
+        self.enable_full_ai_capture = (
+            enable_full_ai_capture is True
+            or _use_ai_lane is True
+            or _enable_multimodal_capture is True
+        )
         self.is_server = is_server
         self.historical_migration = historical_migration
         # Selects the capture wire protocol (V0 legacy `/batch/` vs V1
@@ -992,6 +998,24 @@ def consumers(self) -> Optional[List[Consumer]]:
             return None
         return [consumer for lane in self._lanes for consumer in lane.consumers]
 
+    @property
+    def _use_ai_lane(self) -> bool:
+        """Deprecated alias for `enable_full_ai_capture`."""
+        return self.enable_full_ai_capture
+
+    @_use_ai_lane.setter
+    def _use_ai_lane(self, value) -> None:
+        self.enable_full_ai_capture = value is True
+
+    @property
+    def _enable_multimodal_capture(self) -> bool:
+        """Deprecated alias for `enable_full_ai_capture`."""
+        return self.enable_full_ai_capture
+
+    @_enable_multimodal_capture.setter
+    def _enable_multimodal_capture(self, value) -> None:
+        self.enable_full_ai_capture = value is True
+
     def _warn_if_duplicate_async_client(self):
         if self.disabled or not self.send or self.sync_mode or not self.api_key:
             return
@@ -1453,30 +1477,35 @@ def capture(
         return self._capture(event, self._analytics_lane, **kwargs)
 
     @no_throw()
-    def _capture_ai(
+    def capture_ai(
         self, event: str, **kwargs: Unpack[OptionalCaptureArgs]
     ) -> Optional[str]:
-        """Capture an AI event on the dedicated AI lane.
+        """Capture an AI event on the dedicated AI capture endpoint.
+
+        Beta: the signature is stable; operational limits (per-event size
+        cap, batching, endpoint) may change without notice.
 
-        Internal and experimental, with no stability guarantees: the signature
-        and lane behavior may change while the AI capture lane is validated on
-        PostHog's own traffic.
+        Takes the same arguments and returns the same value as `capture()`:
+        the event UUID, or None when the event was not admitted (disabled
+        client, or dropped by `before_send`). The event is queued on an
+        isolated AI lane with its own consumer pool and a higher per-event
+        size cap, posting to the dedicated AI ingestion endpoint. The payload
+        is sent as given — no redaction or truncation is applied here.
 
-        Takes the same arguments and returns the same value as `capture()`,
-        but the event is queued on the AI lane, which posts to the dedicated
-        AI endpoint with its own consumer pool and per-event size cap.
+        Category:
+            Capture
         """
         if not event.startswith("$ai_"):
             self.log.debug(
-                "_capture_ai called with non-AI event name %r; routing it to the AI endpoint anyway.",
+                "capture_ai called with non-AI event name %r; routing it to the AI endpoint anyway.",
                 event,
             )
         return self._capture(event, self._ai_lane, **kwargs)
 
     def _capture(
         self, event: str, lane: _Lane, **kwargs: Unpack[OptionalCaptureArgs]
     ) -> Optional[str]:
-        """Shared message-building body of `capture()` and `_capture_ai()`; `lane` picks the wire destination."""
+        """Shared message-building body of `capture()` and `capture_ai()`; `lane` picks the wire destination."""
         distinct_id = kwargs.get("distinct_id", None)
         properties = kwargs.get("properties", None)
         timestamp = kwargs.get("timestamp", None)
@@ -2118,6 +2147,21 @@ def _reinit_after_fork(self):
         else:
             self.poller = None
 
+    def _normalize_event_uuid(self, msg):
+        # type: (...) -> None
+        """Ensure `msg["uuid"]` is a valid uuid string, generating one if missing or invalid."""
+        if "uuid" in msg:
+            uuid = msg.pop("uuid")
+            if uuid is not None:
+                try:
+                    msg["uuid"] = _stringify_event_uuid(uuid)
+                except ValueError as e:
+                    self.log.error("%s Falling back to a generated UUID.", e)
+
+        if "uuid" not in msg:
+            # Always send a uuid, so we can always return one
+            msg["uuid"] = stringify_id(uuid4())
+
     def _enqueue(self, msg, disable_geoip, lane=None, property_allowlist=None):
         # type: (...) -> Optional[str]
         """Push a new `msg` onto a lane's queue (analytics when unspecified), return the event uuid or None."""
@@ -2136,19 +2180,7 @@ def _enqueue(self, msg, disable_geoip, lane=None, property_allowlist=None):
         timestamp = guess_timezone(timestamp)
         msg["timestamp"] = timestamp.isoformat()
 
-        if "uuid" in msg:
-            uuid = msg.pop("uuid")
-            if uuid is not None:
-                try:
-                    msg["uuid"] = _stringify_event_uuid(uuid)
-                except ValueError as e:
-                    self.log.error("%s Falling back to a generated UUID.", e)
-
-        if "uuid" not in msg:
-            # Always send a uuid, so we can always return one
-            msg["uuid"] = stringify_id(uuid4())
-
-        sent_uuid = msg["uuid"]
+        self._normalize_event_uuid(msg)
 
         if not msg.get("properties"):
             msg["properties"] = {}
@@ -2194,6 +2226,11 @@ def _enqueue(self, msg, disable_geoip, lane=None, property_allowlist=None):
                 self.log.exception(f"Error in before_send callback: {e}")
                 # Continue with the original message if callback fails
 
+        # Re-normalized after before_send, which may have replaced or removed
+        # msg["uuid"], so the returned uuid always matches the wire event.
+        self._normalize_event_uuid(msg)
+        sent_uuid = msg["uuid"]
+
         self.log.debug("queueing: %s", msg)
 
         # if send is False, return msg as if it was successfully queued, unless

posthog/test/ai/anthropic/test_anthropic.py

@@ -1970,7 +1970,7 @@ async def mock_async_create(**kwargs):
 
 
 def test_ai_lane_client_routes_through_capture_ai(mock_client, mock_anthropic_response):
-    mock_client._use_ai_lane = True
+    mock_client.enable_full_ai_capture = True
     with patch(
         "anthropic.resources.Messages.create", return_value=mock_anthropic_response
     ):
@@ -1982,12 +1982,12 @@ def test_ai_lane_client_routes_through_capture_ai(mock_client, mock_anthropic_re
         )
 
     mock_client.capture.assert_not_called()
-    assert mock_client._capture_ai.call_count == 1
-    assert mock_client._capture_ai.call_args[1]["event"] == "$ai_generation"
+    assert mock_client.capture_ai.call_count == 1
+    assert mock_client.capture_ai.call_args[1]["event"] == "$ai_generation"
 
 
 def test_multimodal_client_skips_media_redaction(mock_client, mock_anthropic_response):
-    mock_client._enable_multimodal_capture = True
+    mock_client.enable_full_ai_capture = True
     image_data = "A" * 64
 
     with patch(
@@ -2015,6 +2015,6 @@ def test_multimodal_client_skips_media_redaction(mock_client, mock_anthropic_res
         )
 
     mock_client.capture.assert_not_called()
-    assert mock_client._capture_ai.call_count == 1
-    props = mock_client._capture_ai.call_args[1]["properties"]
+    assert mock_client.capture_ai.call_count == 1
+    props = mock_client.capture_ai.call_args[1]["properties"]
     assert props["$ai_input"][0]["content"][0]["source"]["data"] == image_data

posthog/test/ai/claude_agent_sdk/test_processor.py

@@ -785,10 +785,10 @@ async def test_non_config_errors_propagate(self):
 
 
 def test_ai_lane_client_routes_through_capture_ai(mock_client):
-    mock_client._use_ai_lane = True
+    mock_client.enable_full_ai_capture = True
     processor = PostHogClaudeAgentProcessor(client=mock_client, distinct_id="test-user")
     processor._capture_event(event="$ai_trace", properties={}, distinct_id="d")
 
     mock_client.capture.assert_not_called()
-    mock_client._capture_ai.assert_called_once()
-    assert mock_client._capture_ai.call_args[1]["event"] == "$ai_trace"
+    mock_client.capture_ai.assert_called_once()
+    assert mock_client.capture_ai.call_args[1]["event"] == "$ai_trace"

posthog/test/ai/claude_agent_sdk/test_processor_content.py

@@ -73,7 +73,7 @@ def test_empty_string_tool_content_not_none(mod):
 def test_passthrough_tool_result_media_not_truncated(mod):
     # A client opted into multimodal capture must not get its tool-result media
     # cut at max_string_len — a 5000-char slice through base64 corrupts it.
-    client = types.SimpleNamespace(_enable_multimodal_capture=True)
+    client = types.SimpleNamespace(enable_full_ai_capture=True)
     long_b64 = "A" * 6000
     block = FakeToolResultBlock(
         content=[

posthog/test/ai/gemini/test_gemini.py

@@ -1584,7 +1584,7 @@ def test_ai_lane_client_routes_through_capture_ai(
 ):
     mock_google_genai_client.models.generate_content.return_value = mock_gemini_response
 
-    mock_client._use_ai_lane = True
+    mock_client.enable_full_ai_capture = True
     client = Client(api_key="test-key", posthog_client=mock_client)
     client.models.generate_content(
         model="gemini-2.0-flash",
@@ -1593,5 +1593,5 @@ def test_ai_lane_client_routes_through_capture_ai(
     )
 
     mock_client.capture.assert_not_called()
-    assert mock_client._capture_ai.call_count == 1
-    assert mock_client._capture_ai.call_args[1]["event"] == "$ai_generation"
+    assert mock_client.capture_ai.call_count == 1
+    assert mock_client.capture_ai.call_args[1]["event"] == "$ai_generation"