feat(aio): dedicated AI capture lane and multimodal passthrough

.sampo/changesets/capture-ai-lane.md

@@ -0,0 +1,7 @@
+---
+pypi/posthog: minor
+---
+
+Refactored capture internals to support multiple delivery lanes per client. Added an internal test lane for heavy AI events.
+
+Events captured after `shutdown()` are now dropped with a warning instead of being silently queued with no consumer to deliver them.

AGENTS.md

@@ -24,7 +24,7 @@ Where the pieces live:
 - `posthog/capture_compression.py` — the `CaptureCompression` enum and `_resolve_capture_compression()` precedence logic (with `gzip` fallback).
 - `posthog/capture_v1.py` — pure transforms (`_to_v1_event`, `_build_v1_batch_body`) and transport (`_post_v1`, `_compress_v1`, `_parse_v1_response`, `_send_v1_batch`, `CaptureV1Error`).
 - Public API surface (enforced by `references/public_api_snapshot.txt`): `CaptureMode`, `CaptureCompression` (both re-exported from `posthog`), `CaptureV1Error`, and the two env var names. Everything else in these modules is underscore-private plumbing.
-- Routing: `Consumer._send_analytics` (async) and `Client._enqueue` (sync) pick the analytics submitter by `capture_mode`. The dedicated `$ai_*` endpoint has no v1 form and always uses the legacy submitter.
+- Routing: `Consumer.request` (async) and `Client._enqueue` (sync) pick the submitter by `capture_mode`.
 
 v1-specific behavior to preserve when editing: sentinel `$`-properties are lifted into `options` (coerced to native JSON types or omitted — a wrong type 400s the whole batch); top-level `$set`/`$set_once` are relocated into `properties`; only events the server tags `retry` are resent (stable `PostHog-Request-Id`/`created_at`, incrementing `PostHog-Attempt`); a server `drop` is a terminal per-event rejection — drops are accumulated across attempts and surfaced via `CaptureV1Error`/`on_error` even on a 2xx with no retries (a success status is not full delivery); `Retry-After` is a *minimum*, not a replacement (the client waits `max(configured_backoff, min(Retry-After, _MAX_BACKOFF_SECONDS))`); `_MAX_BACKOFF_SECONDS` (30s) is the single ceiling for both the exponential backoff and the `Retry-After` clamp; `429` is terminal.
 

posthog/__init__.py

@@ -393,6 +393,14 @@ 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.
+_use_ai_lane = False  # type: bool
+_enable_multimodal_capture = False  # type: bool
 
 default_client = None  # type: Optional[Client]
 
@@ -1201,6 +1209,8 @@ def setup() -> Client:
     default_client.disabled = disabled or not default_client.api_key
     default_client.debug = debug
     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)
     # 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/anthropic/anthropic.py

@@ -250,7 +250,7 @@ def _capture_streaming_event(
 
         # Prepare standardized event data
         formatted_input = format_anthropic_streaming_input(kwargs)
-        sanitized_input = sanitize_anthropic(formatted_input)
+        sanitized_input = sanitize_anthropic(formatted_input, self._client._ph_client)
 
         event_data = StreamingEventData(
             provider="anthropic",

posthog/ai/anthropic/anthropic_async.py

@@ -251,7 +251,7 @@ async def _capture_streaming_event(
 
         # Prepare standardized event data
         formatted_input = format_anthropic_streaming_input(kwargs)
-        sanitized_input = sanitize_anthropic(formatted_input)
+        sanitized_input = sanitize_anthropic(formatted_input, self._client._ph_client)
 
         event_data = StreamingEventData(
             provider="anthropic",

posthog/ai/claude_agent_sdk/processor.py

@@ -25,6 +25,7 @@
     )
 
 from posthog import setup
+from posthog.ai.utils import _capture_ai_event
 from posthog.client import Client
 
 log = logging.getLogger("posthog")
@@ -198,9 +199,10 @@ def _capture_event(
                 **self._properties,
             }
 
-            self._client.capture(
+            _capture_ai_event(
+                self._client,
+                event,
                 distinct_id=distinct_id or "unknown",
-                event=event,
                 properties=final_properties,
                 groups=groups if groups is not None else self._groups,
             )

posthog/ai/gemini/gemini.py

@@ -16,6 +16,7 @@
 from posthog import setup
 from posthog.ai.utils import (
     call_llm_and_track_usage,
+    _capture_ai_event,
     capture_streaming_event,
     merge_usage_stats,
 )
@@ -370,7 +371,7 @@ def _capture_streaming_event(
     ):
         # Prepare standardized event data
         formatted_input = self._format_input(contents, **kwargs)
-        sanitized_input = sanitize_gemini(formatted_input)
+        sanitized_input = sanitize_gemini(formatted_input, self._ph_client)
 
         event_data = StreamingEventData(
             provider="gemini",
@@ -522,9 +523,10 @@ def embed_content(
             if distinct_id is None:
                 event_properties["$process_person_profile"] = False
 
-            self._ph_client.capture(
+            _capture_ai_event(
+                self._ph_client,
+                "$ai_embedding",
                 distinct_id=distinct_id or trace_id,
-                event="$ai_embedding",
                 properties=event_properties,
                 groups=groups,
             )

posthog/ai/gemini/gemini_async.py

@@ -17,6 +17,7 @@
 from posthog import setup
 from posthog.ai.utils import (
     call_llm_and_track_usage_async,
+    _capture_ai_event,
     capture_streaming_event,
     merge_usage_stats,
 )
@@ -374,7 +375,7 @@ def _capture_streaming_event(
     ):
         # Prepare standardized event data
         formatted_input = self._format_input(contents, **kwargs)
-        sanitized_input = sanitize_gemini(formatted_input)
+        sanitized_input = sanitize_gemini(formatted_input, self._ph_client)
 
         event_data = StreamingEventData(
             provider="gemini",
@@ -526,9 +527,10 @@ async def embed_content(
             if distinct_id is None:
                 event_properties["$process_person_profile"] = False
 
-            self._ph_client.capture(
+            _capture_ai_event(
+                self._ph_client,
+                "$ai_embedding",
                 distinct_id=distinct_id or trace_id,
-                event="$ai_embedding",
                 properties=event_properties,
                 groups=groups,
             )

posthog/ai/langchain/callbacks.py

@@ -44,7 +44,7 @@
 from posthog import setup
 from posthog.ai.gateway import warn_if_posthog_ai_gateway
 from posthog.ai.sanitization import sanitize_langchain
-from posthog.ai.utils import get_model_params, with_privacy_mode
+from posthog.ai.utils import _capture_ai_event, get_model_params, with_privacy_mode
 from posthog.client import Client
 
 log = logging.getLogger("posthog")
@@ -536,7 +536,9 @@ def _capture_trace_or_span(
         event_properties = {
             "$ai_trace_id": trace_id,
             "$ai_input_state": with_privacy_mode(
-                self._ph_client, self._privacy_mode, sanitize_langchain(run.input)
+                self._ph_client,
+                self._privacy_mode,
+                sanitize_langchain(run.input, self._ph_client),
             ),
             "$ai_latency": run.latency,
             "$ai_span_name": run.name,
@@ -567,9 +569,10 @@ def _capture_trace_or_span(
         if self._distinct_id is None:
             event_properties["$process_person_profile"] = False
 
-        self._ph_client.capture(
+        _capture_ai_event(
+            self._ph_client,
+            event_name,
             distinct_id=self._distinct_id or run_id,
-            event=event_name,
             properties=event_properties,
             groups=self._groups,
         )
@@ -615,7 +618,9 @@ def _capture_generation(
             "$ai_model": run.model,
             "$ai_model_parameters": run.model_params,
             "$ai_input": with_privacy_mode(
-                self._ph_client, self._privacy_mode, sanitize_langchain(run.input)
+                self._ph_client,
+                self._privacy_mode,
+                sanitize_langchain(run.input, self._ph_client),
             ),
             "$ai_http_status": 200,
             "$ai_latency": run.latency,
@@ -685,9 +690,10 @@ def _capture_generation(
                     if finish_reason is not None:
                         event_properties["$ai_stop_reason"] = finish_reason
 
-        self._ph_client.capture(
+        _capture_ai_event(
+            self._ph_client,
+            "$ai_generation",
             distinct_id=self._distinct_id or trace_id,
-            event="$ai_generation",
             properties=event_properties,
             groups=self._groups,
         )

posthog/ai/openai/openai.py

@@ -13,6 +13,7 @@
 
 from posthog.ai.utils import (
     call_llm_and_track_usage,
+    _capture_ai_event,
     extract_available_tool_calls,
     merge_usage_stats,
     with_privacy_mode,
@@ -244,7 +245,9 @@ def _capture_streaming_event(
 
         # Prepare standardized event data
         formatted_input = format_openai_streaming_input(kwargs, "responses")
-        sanitized_input = sanitize_openai_response(formatted_input)
+        sanitized_input = sanitize_openai_response(
+            formatted_input, self._client._ph_client
+        )
 
         # Use model from kwargs, fallback to model from response
         model = kwargs.get("model") or model_from_response or "unknown"
@@ -513,7 +516,7 @@ def _capture_streaming_event(
 
         # Prepare standardized event data
         formatted_input = format_openai_streaming_input(kwargs, "chat")
-        sanitized_input = sanitize_openai(formatted_input)
+        sanitized_input = sanitize_openai(formatted_input, self._client._ph_client)
 
         # Use model from kwargs, fallback to model from response
         model = kwargs.get("model") or model_from_response or "unknown"
@@ -590,7 +593,7 @@ def create(
             "$ai_input": with_privacy_mode(
                 self._client._ph_client,
                 posthog_privacy_mode,
-                sanitize_openai_response(kwargs.get("input")),
+                sanitize_openai_response(kwargs.get("input"), self._client._ph_client),
             ),
             "$ai_http_status": 200,
             "$ai_input_tokens": usage_stats.get("prompt_tokens", 0),
@@ -605,9 +608,10 @@ def create(
 
         # Send capture event for embeddings
         if hasattr(self._client._ph_client, "capture"):
-            self._client._ph_client.capture(
+            _capture_ai_event(
+                self._client._ph_client,
+                "$ai_embedding",
                 distinct_id=posthog_distinct_id or posthog_trace_id,
-                event="$ai_embedding",
                 properties=event_properties,
                 groups=posthog_groups,
             )

posthog/ai/openai/openai_async.py

@@ -15,6 +15,7 @@
 from posthog import setup
 from posthog.ai.utils import (
     call_llm_and_track_usage_async,
+    _capture_ai_event,
     extract_available_tool_calls,
     get_model_params,
     merge_usage_stats,
@@ -252,7 +253,7 @@ async def _capture_streaming_event(
             "$ai_input": with_privacy_mode(
                 self._client._ph_client,
                 posthog_privacy_mode,
-                sanitize_openai_response(kwargs.get("input")),
+                sanitize_openai_response(kwargs.get("input"), self._client._ph_client),
             ),
             "$ai_output_choices": with_privacy_mode(
                 self._client._ph_client,
@@ -291,9 +292,10 @@ async def _capture_streaming_event(
             event_properties["$process_person_profile"] = False
 
         if hasattr(self._client._ph_client, "capture"):
-            self._client._ph_client.capture(
+            _capture_ai_event(
+                self._client._ph_client,
+                "$ai_generation",
                 distinct_id=posthog_distinct_id or posthog_trace_id,
-                event="$ai_generation",
                 properties=event_properties,
                 groups=posthog_groups,
             )
@@ -547,7 +549,7 @@ async def _capture_streaming_event(
             "$ai_input": with_privacy_mode(
                 self._client._ph_client,
                 posthog_privacy_mode,
-                sanitize_openai(kwargs.get("messages")),
+                sanitize_openai(kwargs.get("messages"), self._client._ph_client),
             ),
             "$ai_output_choices": with_privacy_mode(
                 self._client._ph_client,
@@ -587,9 +589,10 @@ async def _capture_streaming_event(
             event_properties["$process_person_profile"] = False
 
         if hasattr(self._client._ph_client, "capture"):
-            self._client._ph_client.capture(
+            _capture_ai_event(
+                self._client._ph_client,
+                "$ai_generation",
                 distinct_id=posthog_distinct_id or posthog_trace_id,
-                event="$ai_generation",
                 properties=event_properties,
                 groups=posthog_groups,
             )
@@ -647,7 +650,7 @@ async def create(
             "$ai_input": with_privacy_mode(
                 self._client._ph_client,
                 posthog_privacy_mode,
-                sanitize_openai_response(kwargs.get("input")),
+                sanitize_openai_response(kwargs.get("input"), self._client._ph_client),
             ),
             "$ai_http_status": 200,
             "$ai_input_tokens": usage_stats.get("input_tokens", 0),
@@ -662,9 +665,10 @@ async def create(
 
         # Send capture event for embeddings
         if hasattr(self._client._ph_client, "capture"):
-            self._client._ph_client.capture(
+            _capture_ai_event(
+                self._client._ph_client,
+                "$ai_embedding",
                 distinct_id=posthog_distinct_id or posthog_trace_id,
-                event="$ai_embedding",
                 properties=event_properties,
                 groups=posthog_groups,
             )

posthog/ai/openai_agents/processor.py

@@ -21,6 +21,7 @@
 )
 
 from posthog import setup
+from posthog.ai.utils import _capture_ai_event
 from posthog.client import Client
 
 log = logging.getLogger("posthog")
@@ -192,9 +193,10 @@ def _capture_event(
                 **self._properties,
             }
 
-            self._client.capture(
+            _capture_ai_event(
+                self._client,
+                event,
                 distinct_id=distinct_id or "unknown",
-                event=event,
                 properties=final_properties,
                 groups=self._groups,
             )