fix(llma): default prompts url (#423)

* fix(llma): small fixes for prompt management

* fix(llma): tests

* fix(llma): tests

.github/workflows/ci.yml

@@ -76,6 +76,29 @@ jobs:
               run: |
                   pytest --verbose --timeout=30
 
+    import-check:
+        name: Python ${{ matrix.python-version }} import check
+        runs-on: ubuntu-latest
+        strategy:
+            matrix:
+                python-version: ['3.10', '3.11', '3.12', '3.13', '3.14']
+
+        steps:
+            - uses: actions/checkout@85e6279cec87321a52edac9c87bce653a07cf6c2
+              with:
+                  fetch-depth: 1
+
+            - name: Set up Python ${{ matrix.python-version }}
+              uses: actions/setup-python@8d9ed9ac5c53483de85588cdf95a591a75ab9f55
+              with:
+                  python-version: ${{ matrix.python-version }}
+
+            - name: Install posthog
+              run: pip install .
+
+            - name: Check import produces no warnings
+              run: python -W error -c "import posthog"
+
     django5-integration:
         name: Django 5 integration tests
         runs-on: ubuntu-latest

.github/workflows/codeql.yml

@@ -0,0 +1,45 @@
+name: 'CodeQL Advanced'
+
+on:
+  push:
+    branches: ['master']
+  pull_request:
+    branches: ['master']
+  schedule:
+    - cron: '32 13 * * 1'
+
+jobs:
+  analyze:
+    name: Analyze (${{ matrix.language }})
+    runs-on: 'ubuntu-latest'
+    permissions:
+      security-events: write
+      # required to fetch internal or private CodeQL packs
+      packages: read
+
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - language: actions
+            build-mode: none
+          - language: python
+            build-mode: none
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v6
+
+      - name: Initialize CodeQL
+        uses: github/codeql-action/init@5d4e8d1aca955e8d8589aabd499c5cae939e33c7 # v4.31.9
+        with:
+          languages: ${{ matrix.language }}
+          build-mode: ${{ matrix.build-mode }}
+          # Disable TRAP caching - it creates a new cache per commit SHA which
+          # is never reused, causing wasted cache space.
+          # See: https://github.com/github/codeql-action/issues/2030
+          trap-caching: false
+
+      - name: Perform CodeQL Analysis
+        uses: github/codeql-action/analyze@5d4e8d1aca955e8d8589aabd499c5cae939e33c7 # v4.31.9
+        with:
+          category: '/language:${{matrix.language}}'

.github/workflows/sdk-compliance.yml

@@ -0,0 +1,21 @@
+name: SDK Compliance Tests
+
+permissions:
+  contents: read
+  packages: read
+  pull-requests: write
+
+on:
+  pull_request:
+  push:
+    branches:
+      - master
+
+jobs:
+  compliance:
+    name: PostHog SDK compliance tests
+    uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@main
+    with:
+      adapter-dockerfile: "sdk_compliance_adapter/Dockerfile"
+      adapter-context: "."
+      test-harness-version: "latest"

CHANGELOG.md

@@ -1,3 +1,33 @@
+# 7.8.2 - 2026-02-04
+
+fix(llma): fix prompts default url
+
+# 7.8.1 - 2026-02-03
+
+fix(llma): small fixes for prompt management
+
+# 7.8.0 - 2026-01-28
+
+feat(llma): add prompt management
+
+Adds the Prompt Management feature. At the time of release, this feature is in a closed alpha.
+
+# 7.7.0 - 2026-01-15
+
+feat(ai): Add OpenAI Agents SDK integration
+
+Automatic tracing for agent workflows, handoffs, tool calls, guardrails, and custom spans. Includes `$ai_total_tokens`, `$ai_error_type` categorization, and `$ai_framework` property.
+
+# 7.6.0 - 2026-01-12
+
+feat: add device_id to flags request payload
+
+Add device_id parameter to all feature flag methods, allowing the server to track device identifiers for flag evaluation. The device_id can be passed explicitly or set via context using `set_context_device_id()`.
+
+# 7.5.1 - 2026-01-07
+
+fix: avoid return from finally block to fix Python 3.14 SyntaxWarning (#361) - thanks @jodal
+
 # 7.5.0 - 2026-01-06
 
 feat: Capture Langchain, OpenAI and Anthropic errors as exceptions (if exception autocapture is enabled)

posthog/__init__.py

@@ -23,6 +23,9 @@ from posthog.contexts import (
 from posthog.contexts import (
     set_code_variables_mask_patterns_context as inner_set_code_variables_mask_patterns_context,
 )
+from posthog.contexts import (
+    set_context_device_id as inner_set_context_device_id,
+)
 from posthog.contexts import (
     set_context_session as inner_set_context_session,
 )
@@ -133,6 +136,26 @@ def set_context_session(session_id: str):
     return inner_set_context_session(session_id)
 
 
+def set_context_device_id(device_id: str):
+    """
+    Set the device ID for the current context, associating all feature flag requests
+    in this or child contexts with the given device ID.
+
+    Args:
+        device_id: The device ID to associate with the current context and its children
+
+    Examples:
+        ```python
+        from posthog import set_context_device_id
+        set_context_device_id("device_123")
+        ```
+
+    Category:
+        Contexts
+    """
+    return inner_set_context_device_id(device_id)
+
+
 def identify_context(distinct_id: str):
     """
     Identify the current context with a distinct ID.
@@ -483,6 +506,7 @@ def feature_enabled(
     only_evaluate_locally=False,  # type: bool
     send_feature_flag_events=True,  # type: bool
     disable_geoip=None,  # type: Optional[bool]
+    device_id=None,  # type: Optional[str]
 ):
     # type: (...) -> bool
     """
@@ -522,6 +546,7 @@ def feature_enabled(
         only_evaluate_locally=only_evaluate_locally,
         send_feature_flag_events=send_feature_flag_events,
         disable_geoip=disable_geoip,
+        device_id=device_id,
     )
 
 
@@ -534,6 +559,7 @@ def get_feature_flag(
     only_evaluate_locally=False,  # type: bool
     send_feature_flag_events=True,  # type: bool
     disable_geoip=None,  # type: Optional[bool]
+    device_id=None,  # type: Optional[str]
 ) -> Optional[FeatureFlag]:
     """
     Get feature flag variant for users. Used with experiments.
@@ -572,6 +598,7 @@ def get_feature_flag(
         only_evaluate_locally=only_evaluate_locally,
         send_feature_flag_events=send_feature_flag_events,
         disable_geoip=disable_geoip,
+        device_id=device_id,
     )
 
 
@@ -582,6 +609,7 @@ def get_all_flags(
     group_properties=None,  # type: Optional[dict]
     only_evaluate_locally=False,  # type: bool
     disable_geoip=None,  # type: Optional[bool]
+    device_id=None,  # type: Optional[str]
 ) -> Optional[dict[str, FeatureFlag]]:
     """
     Get all flags for a given user.
@@ -614,6 +642,7 @@ def get_all_flags(
         group_properties=group_properties or {},
         only_evaluate_locally=only_evaluate_locally,
         disable_geoip=disable_geoip,
+        device_id=device_id,
     )
 
 
@@ -626,6 +655,7 @@ def get_feature_flag_result(
     only_evaluate_locally=False,
     send_feature_flag_events=True,
     disable_geoip=None,  # type: Optional[bool]
+    device_id=None,  # type: Optional[str]
 ):
     # type: (...) -> Optional[FeatureFlagResult]
     """
@@ -657,6 +687,7 @@ def get_feature_flag_result(
         only_evaluate_locally=only_evaluate_locally,
         send_feature_flag_events=send_feature_flag_events,
         disable_geoip=disable_geoip,
+        device_id=device_id,
     )
 
 
@@ -670,6 +701,7 @@ def get_feature_flag_payload(
     only_evaluate_locally=False,
     send_feature_flag_events=True,
     disable_geoip=None,  # type: Optional[bool]
+    device_id=None,  # type: Optional[str]
 ) -> Optional[str]:
     return _proxy(
         "get_feature_flag_payload",
@@ -682,6 +714,7 @@ def get_feature_flag_payload(
         only_evaluate_locally=only_evaluate_locally,
         send_feature_flag_events=send_feature_flag_events,
         disable_geoip=disable_geoip,
+        device_id=device_id,
     )
 
 
@@ -712,6 +745,7 @@ def get_all_flags_and_payloads(
     group_properties=None,  # type: Optional[dict]
     only_evaluate_locally=False,
     disable_geoip=None,  # type: Optional[bool]
+    device_id=None,  # type: Optional[str]
 ) -> FlagsAndPayloads:
     return _proxy(
         "get_all_flags_and_payloads",
@@ -721,6 +755,7 @@ def get_all_flags_and_payloads(
         group_properties=group_properties or {},
         only_evaluate_locally=only_evaluate_locally,
         disable_geoip=disable_geoip,
+        device_id=device_id,
     )
 
 

posthog/ai/__init__.py

@@ -0,0 +1,3 @@
+from posthog.ai.prompts import Prompts
+
+__all__ = ["Prompts"]

posthog/ai/anthropic/anthropic_converter.py

@@ -17,6 +17,7 @@ from posthog.ai.types import (
     TokenUsage,
     ToolInProgress,
 )
+from posthog.ai.utils import serialize_raw_usage
 
 
 def format_anthropic_response(response: Any) -> List[FormattedMessage]:
@@ -221,6 +222,12 @@ def extract_anthropic_usage_from_response(response: Any) -> TokenUsage:
     if web_search_count > 0:
         result["web_search_count"] = web_search_count
 
+    # Capture raw usage metadata for backend processing
+    # Serialize to dict here in the converter (not in utils)
+    serialized = serialize_raw_usage(response.usage)
+    if serialized:
+        result["raw_usage"] = serialized
+
     return result
 
 
@@ -247,6 +254,11 @@ def extract_anthropic_usage_from_event(event: Any) -> TokenUsage:
             usage["cache_read_input_tokens"] = getattr(
                 event.message.usage, "cache_read_input_tokens", 0
             )
+            # Capture raw usage metadata for backend processing
+            # Serialize to dict here in the converter (not in utils)
+            serialized = serialize_raw_usage(event.message.usage)
+            if serialized:
+                usage["raw_usage"] = serialized
 
     # Handle usage stats from message_delta event
     if hasattr(event, "usage") and event.usage:
@@ -262,6 +274,12 @@ def extract_anthropic_usage_from_event(event: Any) -> TokenUsage:
                 if web_search_count > 0:
                     usage["web_search_count"] = web_search_count
 
+        # Capture raw usage metadata for backend processing
+        # Serialize to dict here in the converter (not in utils)
+        serialized = serialize_raw_usage(event.usage)
+        if serialized:
+            usage["raw_usage"] = serialized
+
     return usage
 
 

posthog/ai/gemini/gemini_converter.py

@@ -12,6 +12,7 @@ from posthog.ai.types import (
     FormattedMessage,
     TokenUsage,
 )
+from posthog.ai.utils import serialize_raw_usage
 
 
 class GeminiPart(TypedDict, total=False):
@@ -487,6 +488,12 @@ def _extract_usage_from_metadata(metadata: Any) -> TokenUsage:
         if reasoning_tokens and reasoning_tokens > 0:
             usage["reasoning_tokens"] = reasoning_tokens
 
+    # Capture raw usage metadata for backend processing
+    # Serialize to dict here in the converter (not in utils)
+    serialized = serialize_raw_usage(metadata)
+    if serialized:
+        usage["raw_usage"] = serialized
+
     return usage
 
 

posthog/ai/openai/openai_converter.py

@@ -16,6 +16,7 @@ from posthog.ai.types import (
     FormattedTextContent,
     TokenUsage,
 )
+from posthog.ai.utils import serialize_raw_usage
 
 
 def format_openai_response(response: Any) -> List[FormattedMessage]:
@@ -429,6 +430,12 @@ def extract_openai_usage_from_response(response: Any) -> TokenUsage:
     if web_search_count > 0:
         result["web_search_count"] = web_search_count
 
+    # Capture raw usage metadata for backend processing
+    # Serialize to dict here in the converter (not in utils)
+    serialized = serialize_raw_usage(response.usage)
+    if serialized:
+        result["raw_usage"] = serialized
+
     return result
 
 
@@ -482,6 +489,12 @@ def extract_openai_usage_from_chunk(
                 chunk.usage.completion_tokens_details.reasoning_tokens
             )
 
+        # Capture raw usage metadata for backend processing
+        # Serialize to dict here in the converter (not in utils)
+        serialized = serialize_raw_usage(chunk.usage)
+        if serialized:
+            usage["raw_usage"] = serialized
+
     elif provider_type == "responses":
         # For Responses API, usage is only in chunk.response.usage for completed events
         if hasattr(chunk, "type") and chunk.type == "response.completed":
@@ -516,6 +529,12 @@ def extract_openai_usage_from_chunk(
                     if web_search_count > 0:
                         usage["web_search_count"] = web_search_count
 
+                # Capture raw usage metadata for backend processing
+                # Serialize to dict here in the converter (not in utils)
+                serialized = serialize_raw_usage(response_usage)
+                if serialized:
+                    usage["raw_usage"] = serialized
+
     return usage
 
 

posthog/ai/openai_agents/__init__.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Union
+
+if TYPE_CHECKING:
+    from agents.tracing import Trace
+
+    from posthog.client import Client
+
+try:
+    import agents  # noqa: F401
+except ImportError:
+    raise ModuleNotFoundError(
+        "Please install the OpenAI Agents SDK to use this feature: 'pip install openai-agents'"
+    )
+
+from posthog.ai.openai_agents.processor import PostHogTracingProcessor
+
+__all__ = ["PostHogTracingProcessor", "instrument"]
+
+
+def instrument(
+    client: Optional[Client] = None,
+    distinct_id: Optional[Union[str, Callable[[Trace], Optional[str]]]] = None,
+    privacy_mode: bool = False,
+    groups: Optional[Dict[str, Any]] = None,
+    properties: Optional[Dict[str, Any]] = None,
+) -> PostHogTracingProcessor:
+    """
+    One-liner to instrument OpenAI Agents SDK with PostHog tracing.
+
+    This registers a PostHogTracingProcessor with the OpenAI Agents SDK,
+    automatically capturing traces, spans, and LLM generations.
+
+    Args:
+        client: Optional PostHog client instance. If not provided, uses the default client.
+        distinct_id: Optional distinct ID to associate with all traces.
+            Can also be a callable that takes a trace and returns a distinct ID.
+        privacy_mode: If True, redacts input/output content from events.
+        groups: Optional PostHog groups to associate with events.
+        properties: Optional additional properties to include with all events.
+
+    Returns:
+        PostHogTracingProcessor: The registered processor instance.
+
+    Example:
+        ```python
+        from posthog.ai.openai_agents import instrument
+
+        # Simple setup
+        instrument(distinct_id="user@example.com")
+
+        # With custom properties
+        instrument(
+            distinct_id="user@example.com",
+            privacy_mode=True,
+            properties={"environment": "production"}
+        )
+
+        # Now run agents as normal - traces automatically sent to PostHog
+        from agents import Agent, Runner
+        agent = Agent(name="Assistant", instructions="You are helpful.")
+        result = Runner.run_sync(agent, "Hello!")
+        ```
+    """
+    from agents.tracing import add_trace_processor
+
+    processor = PostHogTracingProcessor(
+        client=client,
+        distinct_id=distinct_id,
+        privacy_mode=privacy_mode,
+        groups=groups,
+        properties=properties,
+    )
+    add_trace_processor(processor)
+    return processor

posthog/ai/openai_agents/processor.py

@@ -0,0 +1,863 @@
+import json
+import logging
+import time
+from datetime import datetime
+from typing import Any, Callable, Dict, Optional, Union
+
+from agents.tracing import Span, Trace
+from agents.tracing.processor_interface import TracingProcessor
+from agents.tracing.span_data import (
+    AgentSpanData,
+    CustomSpanData,
+    FunctionSpanData,
+    GenerationSpanData,
+    GuardrailSpanData,
+    HandoffSpanData,
+    MCPListToolsSpanData,
+    ResponseSpanData,
+    SpeechGroupSpanData,
+    SpeechSpanData,
+    TranscriptionSpanData,
+)
+
+from posthog import setup
+from posthog.client import Client
+
+log = logging.getLogger("posthog")
+
+
+def _ensure_serializable(obj: Any) -> Any:
+    """Ensure an object is JSON-serializable, converting to str as fallback.
+
+    Returns the original object if it's already serializable (dict, list, str,
+    int, etc.), or str(obj) for non-serializable types so that downstream
+    json.dumps() calls won't fail.
+    """
+    if obj is None:
+        return None
+    try:
+        json.dumps(obj)
+        return obj
+    except (TypeError, ValueError):
+        return str(obj)
+
+
+def _parse_iso_timestamp(iso_str: Optional[str]) -> Optional[float]:
+    """Parse ISO timestamp to Unix timestamp."""
+    if not iso_str:
+        return None
+    try:
+        dt = datetime.fromisoformat(iso_str.replace("Z", "+00:00"))
+        return dt.timestamp()
+    except (ValueError, AttributeError):
+        return None
+
+
+class PostHogTracingProcessor(TracingProcessor):
+    """
+    A tracing processor that sends OpenAI Agents SDK traces to PostHog.
+
+    This processor implements the TracingProcessor interface from the OpenAI Agents SDK
+    and maps agent traces, spans, and generations to PostHog's LLM analytics events.
+
+    Example:
+        ```python
+        from agents import Agent, Runner
+        from agents.tracing import add_trace_processor
+        from posthog.ai.openai_agents import PostHogTracingProcessor
+
+        # Create and register the processor
+        processor = PostHogTracingProcessor(
+            distinct_id="user@example.com",
+            privacy_mode=False,
+        )
+        add_trace_processor(processor)
+
+        # Run agents as normal - traces automatically sent to PostHog
+        agent = Agent(name="Assistant", instructions="You are helpful.")
+        result = Runner.run_sync(agent, "Hello!")
+        ```
+    """
+
+    def __init__(
+        self,
+        client: Optional[Client] = None,
+        distinct_id: Optional[Union[str, Callable[[Trace], Optional[str]]]] = None,
+        privacy_mode: bool = False,
+        groups: Optional[Dict[str, Any]] = None,
+        properties: Optional[Dict[str, Any]] = None,
+    ):
+        """
+        Initialize the PostHog tracing processor.
+
+        Args:
+            client: Optional PostHog client instance. If not provided, uses the default client.
+            distinct_id: Either a string distinct ID or a callable that takes a Trace
+                and returns a distinct ID. If not provided, uses the trace_id.
+            privacy_mode: If True, redacts input/output content from events.
+            groups: Optional PostHog groups to associate with all events.
+            properties: Optional additional properties to include with all events.
+        """
+        self._client = client or setup()
+        self._distinct_id = distinct_id
+        self._privacy_mode = privacy_mode
+        self._groups = groups or {}
+        self._properties = properties or {}
+
+        # Track span start times for latency calculation
+        self._span_start_times: Dict[str, float] = {}
+
+        # Track trace metadata for associating with spans
+        self._trace_metadata: Dict[str, Dict[str, Any]] = {}
+
+        # Max entries to prevent unbounded growth if on_span_end/on_trace_end
+        # is never called (e.g., due to an exception in the Agents SDK).
+        self._max_tracked_entries = 10000
+
+    def _get_distinct_id(self, trace: Optional[Trace]) -> Optional[str]:
+        """Resolve the distinct ID for a trace.
+
+        Returns the user-provided distinct ID (string or callable result),
+        or None if no user-provided ID is available. Callers should treat
+        None as a signal to use a fallback ID in personless mode.
+        """
+        if callable(self._distinct_id):
+            if trace:
+                result = self._distinct_id(trace)
+                if result:
+                    return str(result)
+            return None
+        elif self._distinct_id:
+            return str(self._distinct_id)
+        return None
+
+    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
+
+    def _evict_stale_entries(self) -> None:
+        """Evict oldest entries if dicts exceed max size to prevent unbounded growth."""
+        if len(self._span_start_times) > self._max_tracked_entries:
+            # Remove oldest entries by start time
+            sorted_spans = sorted(self._span_start_times.items(), key=lambda x: x[1])
+            for span_id, _ in sorted_spans[: len(sorted_spans) // 2]:
+                del self._span_start_times[span_id]
+            log.debug(
+                "Evicted stale span start times (exceeded %d entries)",
+                self._max_tracked_entries,
+            )
+
+        if len(self._trace_metadata) > self._max_tracked_entries:
+            # Remove half the entries (oldest inserted via dict ordering in Python 3.7+)
+            keys = list(self._trace_metadata.keys())
+            for key in keys[: len(keys) // 2]:
+                del self._trace_metadata[key]
+            log.debug(
+                "Evicted stale trace metadata (exceeded %d entries)",
+                self._max_tracked_entries,
+            )
+
+    def _get_group_id(self, trace_id: str) -> Optional[str]:
+        """Get the group_id for a trace from stored metadata."""
+        if trace_id in self._trace_metadata:
+            return self._trace_metadata[trace_id].get("group_id")
+        return None
+
+    def _capture_event(
+        self,
+        event: str,
+        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,
+            }
+
+            self._client.capture(
+                distinct_id=distinct_id or "unknown",
+                event=event,
+                properties=final_properties,
+                groups=self._groups,
+            )
+        except Exception as e:
+            log.debug(f"Failed to capture PostHog event: {e}")
+
+    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."""
+        try:
+            self._evict_stale_entries()
+            trace_id = trace.trace_id
+            trace_name = trace.name
+            group_id = getattr(trace, "group_id", None)
+            metadata = getattr(trace, "metadata", None)
+
+            distinct_id = self._get_distinct_id(trace)
+
+            # Store trace metadata for later (used by spans and on_trace_end)
+            self._trace_metadata[trace_id] = {
+                "name": trace_name,
+                "group_id": group_id,
+                "metadata": metadata,
+                "distinct_id": distinct_id,
+                "start_time": time.time(),
+            }
+        except Exception as e:
+            log.debug(f"Error in on_trace_start: {e}")
+
+    def on_trace_end(self, trace: Trace) -> None:
+        """Called when a trace completes. Emits the $ai_trace event with full metadata."""
+        try:
+            trace_id = trace.trace_id
+
+            # Pop stored metadata (also cleans up)
+            trace_info = self._trace_metadata.pop(trace_id, {})
+            trace_name = trace_info.get("name") or trace.name
+            group_id = trace_info.get("group_id") or getattr(trace, "group_id", None)
+            metadata = trace_info.get("metadata") or getattr(trace, "metadata", None)
+            distinct_id = trace_info.get("distinct_id") or self._get_distinct_id(trace)
+
+            # Calculate trace-level latency
+            start_time = trace_info.get("start_time")
+            latency = (time.time() - start_time) if start_time else None
+
+            properties = {
+                "$ai_trace_id": trace_id,
+                "$ai_trace_name": trace_name,
+                "$ai_provider": "openai",
+                "$ai_framework": "openai-agents",
+            }
+
+            if latency is not None:
+                properties["$ai_latency"] = latency
+
+            # Include group_id for linking related traces (e.g., conversation threads)
+            if group_id:
+                properties["$ai_group_id"] = group_id
+
+            # Include trace metadata if present
+            if metadata:
+                properties["$ai_trace_metadata"] = _ensure_serializable(metadata)
+
+            if distinct_id is None:
+                properties["$process_person_profile"] = False
+
+            self._capture_event(
+                event="$ai_trace",
+                distinct_id=distinct_id or trace_id,
+                properties=properties,
+            )
+        except Exception as e:
+            log.debug(f"Error in on_trace_end: {e}")
+
+    def on_span_start(self, span: Span[Any]) -> None:
+        """Called when a new span begins."""
+        try:
+            self._evict_stale_entries()
+            span_id = span.span_id
+            self._span_start_times[span_id] = time.time()
+        except Exception as e:
+            log.debug(f"Error in on_span_start: {e}")
+
+    def on_span_end(self, span: Span[Any]) -> None:
+        """Called when a span completes."""
+        try:
+            span_id = span.span_id
+            trace_id = span.trace_id
+            parent_id = span.parent_id
+            span_data = span.span_data
+
+            # Calculate latency
+            start_time = self._span_start_times.pop(span_id, None)
+            if start_time:
+                latency = time.time() - start_time
+            else:
+                # Fall back to parsing timestamps
+                started = _parse_iso_timestamp(span.started_at)
+                ended = _parse_iso_timestamp(span.ended_at)
+                latency = (ended - started) if (started and ended) else 0
+
+            # Get user-provided distinct ID from trace metadata (resolved at trace start).
+            # None means no user-provided ID — use trace_id as fallback in personless mode,
+            # matching the langchain/openai pattern: `distinct_id or trace_id`.
+            trace_info = self._trace_metadata.get(trace_id, {})
+            distinct_id = trace_info.get("distinct_id") or self._get_distinct_id(None)
+
+            # Get group_id from trace metadata for linking
+            group_id = self._get_group_id(trace_id)
+
+            # Get error info if present
+            error_info = span.error
+            error_properties = {}
+            if error_info:
+                if isinstance(error_info, dict):
+                    error_message = error_info.get("message", str(error_info))
+                    error_type_raw = error_info.get("type", "")
+                else:
+                    error_message = str(error_info)
+                    error_type_raw = ""
+
+                # Categorize error type for cross-provider filtering/alerting
+                error_type = "unknown"
+                if (
+                    "ModelBehaviorError" in error_type_raw
+                    or "ModelBehaviorError" in error_message
+                ):
+                    error_type = "model_behavior_error"
+                elif "UserError" in error_type_raw or "UserError" in error_message:
+                    error_type = "user_error"
+                elif (
+                    "InputGuardrailTripwireTriggered" in error_type_raw
+                    or "InputGuardrailTripwireTriggered" in error_message
+                ):
+                    error_type = "input_guardrail_triggered"
+                elif (
+                    "OutputGuardrailTripwireTriggered" in error_type_raw
+                    or "OutputGuardrailTripwireTriggered" in error_message
+                ):
+                    error_type = "output_guardrail_triggered"
+                elif (
+                    "MaxTurnsExceeded" in error_type_raw
+                    or "MaxTurnsExceeded" in error_message
+                ):
+                    error_type = "max_turns_exceeded"
+
+                error_properties = {
+                    "$ai_is_error": True,
+                    "$ai_error": error_message,
+                    "$ai_error_type": error_type,
+                }
+
+            # Personless mode: no user-provided distinct_id, fallback to trace_id
+            if distinct_id is None:
+                error_properties["$process_person_profile"] = False
+                distinct_id = trace_id
+
+            # Dispatch based on span data type
+            if isinstance(span_data, GenerationSpanData):
+                self._handle_generation_span(
+                    span_data,
+                    trace_id,
+                    span_id,
+                    parent_id,
+                    latency,
+                    distinct_id,
+                    group_id,
+                    error_properties,
+                )
+            elif isinstance(span_data, FunctionSpanData):
+                self._handle_function_span(
+                    span_data,
+                    trace_id,
+                    span_id,
+                    parent_id,
+                    latency,
+                    distinct_id,
+                    group_id,
+                    error_properties,
+                )
+            elif isinstance(span_data, AgentSpanData):
+                self._handle_agent_span(
+                    span_data,
+                    trace_id,
+                    span_id,
+                    parent_id,
+                    latency,
+                    distinct_id,
+                    group_id,
+                    error_properties,
+                )
+            elif isinstance(span_data, HandoffSpanData):
+                self._handle_handoff_span(
+                    span_data,
+                    trace_id,
+                    span_id,
+                    parent_id,
+                    latency,
+                    distinct_id,
+                    group_id,
+                    error_properties,
+                )
+            elif isinstance(span_data, GuardrailSpanData):
+                self._handle_guardrail_span(
+                    span_data,
+                    trace_id,
+                    span_id,
+                    parent_id,
+                    latency,
+                    distinct_id,
+                    group_id,
+                    error_properties,
+                )
+            elif isinstance(span_data, ResponseSpanData):
+                self._handle_response_span(
+                    span_data,
+                    trace_id,
+                    span_id,
+                    parent_id,
+                    latency,
+                    distinct_id,
+                    group_id,
+                    error_properties,
+                )
+            elif isinstance(span_data, CustomSpanData):
+                self._handle_custom_span(
+                    span_data,
+                    trace_id,
+                    span_id,
+                    parent_id,
+                    latency,
+                    distinct_id,
+                    group_id,
+                    error_properties,
+                )
+            elif isinstance(
+                span_data, (TranscriptionSpanData, SpeechSpanData, SpeechGroupSpanData)
+            ):
+                self._handle_audio_span(
+                    span_data,
+                    trace_id,
+                    span_id,
+                    parent_id,
+                    latency,
+                    distinct_id,
+                    group_id,
+                    error_properties,
+                )
+            elif isinstance(span_data, MCPListToolsSpanData):
+                self._handle_mcp_span(
+                    span_data,
+                    trace_id,
+                    span_id,
+                    parent_id,
+                    latency,
+                    distinct_id,
+                    group_id,
+                    error_properties,
+                )
+            else:
+                # Unknown span type - capture as generic span
+                self._handle_generic_span(
+                    span_data,
+                    trace_id,
+                    span_id,
+                    parent_id,
+                    latency,
+                    distinct_id,
+                    group_id,
+                    error_properties,
+                )
+
+        except Exception as e:
+            log.debug(f"Error in on_span_end: {e}")
+
+    def _base_properties(
+        self,
+        trace_id: str,
+        span_id: str,
+        parent_id: Optional[str],
+        latency: float,
+        group_id: Optional[str],
+        error_properties: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        """Build the base properties dict shared by all span handlers."""
+        properties = {
+            "$ai_trace_id": trace_id,
+            "$ai_span_id": span_id,
+            "$ai_parent_id": parent_id,
+            "$ai_provider": "openai",
+            "$ai_framework": "openai-agents",
+            "$ai_latency": latency,
+            **error_properties,
+        }
+        if group_id:
+            properties["$ai_group_id"] = group_id
+        return properties
+
+    def _handle_generation_span(
+        self,
+        span_data: GenerationSpanData,
+        trace_id: str,
+        span_id: str,
+        parent_id: Optional[str],
+        latency: float,
+        distinct_id: str,
+        group_id: Optional[str],
+        error_properties: Dict[str, Any],
+    ) -> None:
+        """Handle LLM generation spans - maps to $ai_generation event."""
+        # Extract token usage
+        usage = span_data.usage or {}
+        input_tokens = usage.get("input_tokens") or usage.get("prompt_tokens") or 0
+        output_tokens = (
+            usage.get("output_tokens") or usage.get("completion_tokens") or 0
+        )
+
+        # Extract model config parameters
+        model_config = span_data.model_config or {}
+        model_params = {}
+        for param in [
+            "temperature",
+            "max_tokens",
+            "top_p",
+            "frequency_penalty",
+            "presence_penalty",
+        ]:
+            if param in model_config:
+                model_params[param] = model_config[param]
+
+        properties = {
+            **self._base_properties(
+                trace_id, span_id, parent_id, latency, group_id, error_properties
+            ),
+            "$ai_model": span_data.model,
+            "$ai_model_parameters": model_params if model_params else None,
+            "$ai_input": self._with_privacy_mode(_ensure_serializable(span_data.input)),
+            "$ai_output_choices": self._with_privacy_mode(
+                _ensure_serializable(span_data.output)
+            ),
+            "$ai_input_tokens": input_tokens,
+            "$ai_output_tokens": output_tokens,
+            "$ai_total_tokens": (input_tokens or 0) + (output_tokens or 0),
+        }
+
+        # Add optional token fields if present
+        if usage.get("reasoning_tokens"):
+            properties["$ai_reasoning_tokens"] = usage["reasoning_tokens"]
+        if usage.get("cache_read_input_tokens"):
+            properties["$ai_cache_read_input_tokens"] = usage["cache_read_input_tokens"]
+        if usage.get("cache_creation_input_tokens"):
+            properties["$ai_cache_creation_input_tokens"] = usage[
+                "cache_creation_input_tokens"
+            ]
+
+        self._capture_event("$ai_generation", properties, distinct_id)
+
+    def _handle_function_span(
+        self,
+        span_data: FunctionSpanData,
+        trace_id: str,
+        span_id: str,
+        parent_id: Optional[str],
+        latency: float,
+        distinct_id: str,
+        group_id: Optional[str],
+        error_properties: Dict[str, Any],
+    ) -> None:
+        """Handle function/tool call spans - maps to $ai_span event."""
+        properties = {
+            **self._base_properties(
+                trace_id, span_id, parent_id, latency, group_id, error_properties
+            ),
+            "$ai_span_name": span_data.name,
+            "$ai_span_type": "tool",
+            "$ai_input_state": self._with_privacy_mode(
+                _ensure_serializable(span_data.input)
+            ),
+            "$ai_output_state": self._with_privacy_mode(
+                _ensure_serializable(span_data.output)
+            ),
+        }
+
+        if span_data.mcp_data:
+            properties["$ai_mcp_data"] = _ensure_serializable(span_data.mcp_data)
+
+        self._capture_event("$ai_span", properties, distinct_id)
+
+    def _handle_agent_span(
+        self,
+        span_data: AgentSpanData,
+        trace_id: str,
+        span_id: str,
+        parent_id: Optional[str],
+        latency: float,
+        distinct_id: str,
+        group_id: Optional[str],
+        error_properties: Dict[str, Any],
+    ) -> None:
+        """Handle agent execution spans - maps to $ai_span event."""
+        properties = {
+            **self._base_properties(
+                trace_id, span_id, parent_id, latency, group_id, error_properties
+            ),
+            "$ai_span_name": span_data.name,
+            "$ai_span_type": "agent",
+        }
+
+        if span_data.handoffs:
+            properties["$ai_agent_handoffs"] = span_data.handoffs
+        if span_data.tools:
+            properties["$ai_agent_tools"] = span_data.tools
+        if span_data.output_type:
+            properties["$ai_agent_output_type"] = span_data.output_type
+
+        self._capture_event("$ai_span", properties, distinct_id)
+
+    def _handle_handoff_span(
+        self,
+        span_data: HandoffSpanData,
+        trace_id: str,
+        span_id: str,
+        parent_id: Optional[str],
+        latency: float,
+        distinct_id: str,
+        group_id: Optional[str],
+        error_properties: Dict[str, Any],
+    ) -> None:
+        """Handle agent handoff spans - maps to $ai_span event."""
+        properties = {
+            **self._base_properties(
+                trace_id, span_id, parent_id, latency, group_id, error_properties
+            ),
+            "$ai_span_name": f"{span_data.from_agent} -> {span_data.to_agent}",
+            "$ai_span_type": "handoff",
+            "$ai_handoff_from_agent": span_data.from_agent,
+            "$ai_handoff_to_agent": span_data.to_agent,
+        }
+
+        self._capture_event("$ai_span", properties, distinct_id)
+
+    def _handle_guardrail_span(
+        self,
+        span_data: GuardrailSpanData,
+        trace_id: str,
+        span_id: str,
+        parent_id: Optional[str],
+        latency: float,
+        distinct_id: str,
+        group_id: Optional[str],
+        error_properties: Dict[str, Any],
+    ) -> None:
+        """Handle guardrail execution spans - maps to $ai_span event."""
+        properties = {
+            **self._base_properties(
+                trace_id, span_id, parent_id, latency, group_id, error_properties
+            ),
+            "$ai_span_name": span_data.name,
+            "$ai_span_type": "guardrail",
+            "$ai_guardrail_triggered": span_data.triggered,
+        }
+
+        self._capture_event("$ai_span", properties, distinct_id)
+
+    def _handle_response_span(
+        self,
+        span_data: ResponseSpanData,
+        trace_id: str,
+        span_id: str,
+        parent_id: Optional[str],
+        latency: float,
+        distinct_id: str,
+        group_id: Optional[str],
+        error_properties: Dict[str, Any],
+    ) -> None:
+        """Handle OpenAI Response API spans - maps to $ai_generation event."""
+        response = span_data.response
+        response_id = response.id if response else None
+
+        # Try to extract usage from response
+        usage = getattr(response, "usage", None) if response else None
+        input_tokens = 0
+        output_tokens = 0
+        if usage:
+            input_tokens = getattr(usage, "input_tokens", 0) or 0
+            output_tokens = getattr(usage, "output_tokens", 0) or 0
+
+        # Try to extract model from response
+        model = getattr(response, "model", None) if response else None
+
+        properties = {
+            **self._base_properties(
+                trace_id, span_id, parent_id, latency, group_id, error_properties
+            ),
+            "$ai_model": model,
+            "$ai_response_id": response_id,
+            "$ai_input": self._with_privacy_mode(_ensure_serializable(span_data.input)),
+            "$ai_input_tokens": input_tokens,
+            "$ai_output_tokens": output_tokens,
+            "$ai_total_tokens": input_tokens + output_tokens,
+        }
+
+        # Extract output content from response
+        if response:
+            output_items = getattr(response, "output", None)
+            if output_items:
+                properties["$ai_output_choices"] = self._with_privacy_mode(
+                    _ensure_serializable(output_items)
+                )
+
+        self._capture_event("$ai_generation", properties, distinct_id)
+
+    def _handle_custom_span(
+        self,
+        span_data: CustomSpanData,
+        trace_id: str,
+        span_id: str,
+        parent_id: Optional[str],
+        latency: float,
+        distinct_id: str,
+        group_id: Optional[str],
+        error_properties: Dict[str, Any],
+    ) -> None:
+        """Handle custom user-defined spans - maps to $ai_span event."""
+        properties = {
+            **self._base_properties(
+                trace_id, span_id, parent_id, latency, group_id, error_properties
+            ),
+            "$ai_span_name": span_data.name,
+            "$ai_span_type": "custom",
+            "$ai_custom_data": self._with_privacy_mode(
+                _ensure_serializable(span_data.data)
+            ),
+        }
+
+        self._capture_event("$ai_span", properties, distinct_id)
+
+    def _handle_audio_span(
+        self,
+        span_data: Union[TranscriptionSpanData, SpeechSpanData, SpeechGroupSpanData],
+        trace_id: str,
+        span_id: str,
+        parent_id: Optional[str],
+        latency: float,
+        distinct_id: str,
+        group_id: Optional[str],
+        error_properties: Dict[str, Any],
+    ) -> None:
+        """Handle audio-related spans (transcription, speech) - maps to $ai_span event."""
+        span_type = span_data.type  # "transcription", "speech", or "speech_group"
+
+        properties = {
+            **self._base_properties(
+                trace_id, span_id, parent_id, latency, group_id, error_properties
+            ),
+            "$ai_span_name": span_type,
+            "$ai_span_type": span_type,
+        }
+
+        # Add model info if available
+        if hasattr(span_data, "model") and span_data.model:
+            properties["$ai_model"] = span_data.model
+
+        # Add model config if available (pass-through property)
+        if hasattr(span_data, "model_config") and span_data.model_config:
+            properties["model_config"] = _ensure_serializable(span_data.model_config)
+
+        # Add time to first audio byte for speech spans (pass-through property)
+        if hasattr(span_data, "first_content_at") and span_data.first_content_at:
+            properties["first_content_at"] = span_data.first_content_at
+
+        # Add audio format info (pass-through properties)
+        if hasattr(span_data, "input_format"):
+            properties["audio_input_format"] = span_data.input_format
+        if hasattr(span_data, "output_format"):
+            properties["audio_output_format"] = span_data.output_format
+
+        # Add text input for TTS
+        if (
+            hasattr(span_data, "input")
+            and span_data.input
+            and isinstance(span_data.input, str)
+        ):
+            properties["$ai_input"] = self._with_privacy_mode(span_data.input)
+
+        # Don't include audio data (base64) - just metadata
+        if hasattr(span_data, "output") and isinstance(span_data.output, str):
+            # For transcription, output is the text
+            properties["$ai_output_state"] = self._with_privacy_mode(span_data.output)
+
+        self._capture_event("$ai_span", properties, distinct_id)
+
+    def _handle_mcp_span(
+        self,
+        span_data: MCPListToolsSpanData,
+        trace_id: str,
+        span_id: str,
+        parent_id: Optional[str],
+        latency: float,
+        distinct_id: str,
+        group_id: Optional[str],
+        error_properties: Dict[str, Any],
+    ) -> None:
+        """Handle MCP (Model Context Protocol) spans - maps to $ai_span event."""
+        properties = {
+            **self._base_properties(
+                trace_id, span_id, parent_id, latency, group_id, error_properties
+            ),
+            "$ai_span_name": f"mcp:{span_data.server}",
+            "$ai_span_type": "mcp_tools",
+            "$ai_mcp_server": span_data.server,
+            "$ai_mcp_tools": span_data.result,
+        }
+
+        self._capture_event("$ai_span", properties, distinct_id)
+
+    def _handle_generic_span(
+        self,
+        span_data: Any,
+        trace_id: str,
+        span_id: str,
+        parent_id: Optional[str],
+        latency: float,
+        distinct_id: str,
+        group_id: Optional[str],
+        error_properties: Dict[str, Any],
+    ) -> None:
+        """Handle unknown span types - maps to $ai_span event."""
+        span_type = getattr(span_data, "type", "unknown")
+
+        properties = {
+            **self._base_properties(
+                trace_id, span_id, parent_id, latency, group_id, error_properties
+            ),
+            "$ai_span_name": span_type,
+            "$ai_span_type": span_type,
+        }
+
+        # Try to export span data
+        if hasattr(span_data, "export"):
+            try:
+                exported = span_data.export()
+                properties["$ai_span_data"] = _ensure_serializable(exported)
+            except Exception:
+                pass
+
+        self._capture_event("$ai_span", properties, distinct_id)
+
+    def shutdown(self) -> None:
+        """Clean up resources when the application stops."""
+        try:
+            self._span_start_times.clear()
+            self._trace_metadata.clear()
+
+            # Flush the PostHog client if possible
+            if hasattr(self._client, "flush") and callable(self._client.flush):
+                self._client.flush()
+        except Exception as e:
+            log.debug(f"Error in shutdown: {e}")
+
+    def force_flush(self) -> None:
+        """Force immediate processing of any queued events."""
+        try:
+            if hasattr(self._client, "flush") and callable(self._client.flush):
+                self._client.flush()
+        except Exception as e:
+            log.debug(f"Error in force_flush: {e}")

posthog/ai/prompts.py

@@ -0,0 +1,272 @@
+"""
+Prompt management for PostHog AI SDK.
+
+Fetch and compile LLM prompts from PostHog with caching and fallback support.
+"""
+
+import logging
+import re
+import time
+import urllib.parse
+from typing import Any, Dict, Optional, Union
+
+from posthog.request import USER_AGENT, _get_session
+from posthog.utils import remove_trailing_slash
+
+log = logging.getLogger("posthog")
+
+APP_ENDPOINT = "https://us.posthog.com"
+DEFAULT_CACHE_TTL_SECONDS = 300  # 5 minutes
+
+PromptVariables = Dict[str, Union[str, int, float, bool]]
+
+
+class CachedPrompt:
+    """Cached prompt with metadata."""
+
+    def __init__(self, prompt: str, fetched_at: float):
+        self.prompt = prompt
+        self.fetched_at = fetched_at
+
+
+def _is_prompt_api_response(data: Any) -> bool:
+    """Check if the response is a valid prompt API response."""
+    return (
+        isinstance(data, dict)
+        and "prompt" in data
+        and isinstance(data.get("prompt"), str)
+    )
+
+
+class Prompts:
+    """
+    Fetch and compile LLM prompts from PostHog.
+
+    Can be initialized with a PostHog client or with direct options.
+
+    Examples:
+        ```python
+        from posthog import Posthog
+        from posthog.ai.prompts import Prompts
+
+        # With PostHog client
+        posthog = Posthog('phc_xxx', host='https://us.posthog.com', personal_api_key='phx_xxx')
+        prompts = Prompts(posthog)
+
+        # Or with direct options (no PostHog client needed)
+        prompts = Prompts(personal_api_key='phx_xxx', host='https://us.posthog.com')
+
+        # Fetch with caching and fallback
+        template = prompts.get('support-system-prompt', fallback='You are a helpful assistant.')
+
+        # Compile with variables
+        system_prompt = prompts.compile(template, {
+            'company': 'Acme Corp',
+            'tier': 'premium',
+        })
+        ```
+    """
+
+    def __init__(
+        self,
+        posthog: Optional[Any] = None,
+        *,
+        personal_api_key: Optional[str] = None,
+        host: Optional[str] = None,
+        default_cache_ttl_seconds: Optional[int] = None,
+    ):
+        """
+        Initialize Prompts.
+
+        Args:
+            posthog: PostHog client instance (optional if personal_api_key provided)
+            personal_api_key: Direct API key (optional if posthog provided)
+            host: PostHog host (defaults to app endpoint)
+            default_cache_ttl_seconds: Default cache TTL (defaults to 300)
+        """
+        self._default_cache_ttl_seconds = (
+            default_cache_ttl_seconds or DEFAULT_CACHE_TTL_SECONDS
+        )
+        self._cache: Dict[str, CachedPrompt] = {}
+
+        if posthog is not None:
+            self._personal_api_key = getattr(posthog, "personal_api_key", None) or ""
+            self._host = remove_trailing_slash(
+                getattr(posthog, "raw_host", None) or APP_ENDPOINT
+            )
+        else:
+            self._personal_api_key = personal_api_key or ""
+            self._host = remove_trailing_slash(host or APP_ENDPOINT)
+
+    def get(
+        self,
+        name: str,
+        *,
+        cache_ttl_seconds: Optional[int] = None,
+        fallback: Optional[str] = None,
+    ) -> str:
+        """
+        Fetch a prompt by name from the PostHog API.
+
+        Caching behavior:
+        1. If cache is fresh, return cached value
+        2. If fetch fails and cache exists (stale), return stale cache with warning
+        3. If fetch fails and fallback provided, return fallback with warning
+        4. If fetch fails with no cache/fallback, raise exception
+
+        Args:
+            name: The name of the prompt to fetch
+            cache_ttl_seconds: Cache TTL in seconds (defaults to instance default)
+            fallback: Fallback prompt to use if fetch fails and no cache available
+
+        Returns:
+            The prompt string
+
+        Raises:
+            Exception: If the prompt cannot be fetched and no fallback is available
+        """
+        ttl = (
+            cache_ttl_seconds
+            if cache_ttl_seconds is not None
+            else self._default_cache_ttl_seconds
+        )
+
+        # Check cache first
+        cached = self._cache.get(name)
+        now = time.time()
+
+        if cached is not None:
+            is_fresh = (now - cached.fetched_at) < ttl
+
+            if is_fresh:
+                return cached.prompt
+
+        # Try to fetch from API
+        try:
+            prompt = self._fetch_prompt_from_api(name)
+            fetched_at = time.time()
+
+            # Update cache
+            self._cache[name] = CachedPrompt(prompt=prompt, fetched_at=fetched_at)
+
+            return prompt
+
+        except Exception as error:
+            # Fallback order:
+            # 1. Return stale cache (with warning)
+            if cached is not None:
+                log.warning(
+                    '[PostHog Prompts] Failed to fetch prompt "%s", using stale cache: %s',
+                    name,
+                    error,
+                )
+                return cached.prompt
+
+            # 2. Return fallback (with warning)
+            if fallback is not None:
+                log.warning(
+                    '[PostHog Prompts] Failed to fetch prompt "%s", using fallback: %s',
+                    name,
+                    error,
+                )
+                return fallback
+
+            # 3. Raise error
+            raise
+
+    def compile(self, prompt: str, variables: PromptVariables) -> str:
+        """
+        Replace {{variableName}} placeholders with values.
+
+        Unmatched variables are left unchanged.
+        Supports variable names with hyphens and dots (e.g., user-id, company.name).
+
+        Args:
+            prompt: The prompt template string
+            variables: Object containing variable values
+
+        Returns:
+            The compiled prompt string
+        """
+
+        def replace_variable(match: re.Match) -> str:
+            variable_name = match.group(1)
+
+            if variable_name in variables:
+                return str(variables[variable_name])
+
+            return match.group(0)
+
+        return re.sub(r"\{\{([\w.-]+)\}\}", replace_variable, prompt)
+
+    def clear_cache(self, name: Optional[str] = None) -> None:
+        """
+        Clear cached prompts.
+
+        Args:
+            name: Specific prompt to clear. If None, clears all cached prompts.
+        """
+        if name is not None:
+            self._cache.pop(name, None)
+        else:
+            self._cache.clear()
+
+    def _fetch_prompt_from_api(self, name: str) -> str:
+        """
+        Fetch prompt from PostHog API.
+
+        Endpoint: {host}/api/environments/@current/llm_prompts/name/{encoded_name}/
+        Auth: Bearer {personal_api_key}
+
+        Args:
+            name: The name of the prompt to fetch
+
+        Returns:
+            The prompt string
+
+        Raises:
+            Exception: If the prompt cannot be fetched
+        """
+        if not self._personal_api_key:
+            raise Exception(
+                "[PostHog Prompts] personal_api_key is required to fetch prompts. "
+                "Please provide it when initializing the Prompts instance."
+            )
+
+        encoded_name = urllib.parse.quote(name, safe="")
+        url = f"{self._host}/api/environments/@current/llm_prompts/name/{encoded_name}/"
+
+        headers = {
+            "Authorization": f"Bearer {self._personal_api_key}",
+            "User-Agent": USER_AGENT,
+        }
+
+        response = _get_session().get(url, headers=headers, timeout=10)
+
+        if not response.ok:
+            if response.status_code == 404:
+                raise Exception(f'[PostHog Prompts] Prompt "{name}" not found')
+
+            if response.status_code == 403:
+                raise Exception(
+                    f'[PostHog Prompts] Access denied for prompt "{name}". '
+                    "Check that your personal_api_key has the correct permissions and the LLM prompts feature is enabled."
+                )
+
+            raise Exception(
+                f'[PostHog Prompts] Failed to fetch prompt "{name}": HTTP {response.status_code}'
+            )
+
+        try:
+            data = response.json()
+        except Exception:
+            raise Exception(
+                f'[PostHog Prompts] Invalid response format for prompt "{name}"'
+            )
+
+        if not _is_prompt_api_response(data):
+            raise Exception(
+                f'[PostHog Prompts] Invalid response format for prompt "{name}"'
+            )
+
+        return data["prompt"]

posthog/ai/types.py

@@ -64,6 +64,7 @@ class TokenUsage(TypedDict, total=False):
     cache_creation_input_tokens: Optional[int]
     reasoning_tokens: Optional[int]
     web_search_count: Optional[int]
+    raw_usage: Optional[Any]  # Raw provider usage metadata for backend processing
 
 
 class ProviderResponse(TypedDict, total=False):

posthog/ai/utils.py

@@ -13,6 +13,54 @@ from posthog.ai.types import FormattedMessage, StreamingEventData, TokenUsage
 from posthog.client import Client as PostHogClient
 
 
+def serialize_raw_usage(raw_usage: Any) -> Optional[Dict[str, Any]]:
+    """
+    Convert raw provider usage objects to JSON-serializable dicts.
+
+    Handles Pydantic models (OpenAI/Anthropic) and protobuf-like objects (Gemini)
+    with a fallback chain to ensure we never pass unserializable objects to PostHog.
+
+    Args:
+        raw_usage: Raw usage object from provider SDK
+
+    Returns:
+        Plain dict or None if conversion fails
+    """
+    if raw_usage is None:
+        return None
+
+    # Already a dict
+    if isinstance(raw_usage, dict):
+        return raw_usage
+
+    # Try Pydantic model_dump() (OpenAI/Anthropic)
+    if hasattr(raw_usage, "model_dump") and callable(raw_usage.model_dump):
+        try:
+            return raw_usage.model_dump()
+        except Exception:
+            pass
+
+    # Try to_dict() (some protobuf objects)
+    if hasattr(raw_usage, "to_dict") and callable(raw_usage.to_dict):
+        try:
+            return raw_usage.to_dict()
+        except Exception:
+            pass
+
+    # Try __dict__ / vars() for simple objects
+    try:
+        return vars(raw_usage)
+    except Exception:
+        pass
+
+    # Last resort: convert to string representation
+    # This ensures we always return something rather than failing
+    try:
+        return {"_raw": str(raw_usage)}
+    except Exception:
+        return None
+
+
 def merge_usage_stats(
     target: TokenUsage, source: TokenUsage, mode: str = "incremental"
 ) -> None:
@@ -60,6 +108,17 @@ def merge_usage_stats(
             current = target.get("web_search_count") or 0
             target["web_search_count"] = max(current, source_web_search)
 
+        # Merge raw_usage to avoid losing data from earlier events
+        # For Anthropic streaming: message_start has input tokens, message_delta has output
+        # Note: raw_usage is already serialized by converters, so it's a dict
+        source_raw_usage = source.get("raw_usage")
+        if source_raw_usage is not None and isinstance(source_raw_usage, dict):
+            current_raw_value = target.get("raw_usage")
+            current_raw: Dict[str, Any] = (
+                current_raw_value if isinstance(current_raw_value, dict) else {}
+            )
+            target["raw_usage"] = {**current_raw, **source_raw_usage}
+
     elif mode == "cumulative":
         # Replace with latest values (already cumulative)
         if source.get("input_tokens") is not None:
@@ -76,6 +135,9 @@ def merge_usage_stats(
             target["reasoning_tokens"] = source["reasoning_tokens"]
         if source.get("web_search_count") is not None:
             target["web_search_count"] = source["web_search_count"]
+        # Note: raw_usage is already serialized by converters, so it's a dict
+        if source.get("raw_usage") is not None:
+            target["raw_usage"] = source["raw_usage"]
 
     else:
         raise ValueError(f"Invalid mode: {mode}. Must be 'incremental' or 'cumulative'")
@@ -332,6 +394,11 @@ def call_llm_and_track_usage(
             if web_search_count is not None and web_search_count > 0:
                 tag("$ai_web_search_count", web_search_count)
 
+            raw_usage = usage.get("raw_usage")
+            if raw_usage is not None:
+                # Already serialized by converters
+                tag("$ai_usage", raw_usage)
+
             if posthog_distinct_id is None:
                 tag("$process_person_profile", False)
 
@@ -457,6 +524,11 @@ async def call_llm_and_track_usage_async(
             if web_search_count is not None and web_search_count > 0:
                 tag("$ai_web_search_count", web_search_count)
 
+            raw_usage = usage.get("raw_usage")
+            if raw_usage is not None:
+                # Already serialized by converters
+                tag("$ai_usage", raw_usage)
+
             if posthog_distinct_id is None:
                 tag("$process_person_profile", False)
 
@@ -594,6 +666,12 @@ def capture_streaming_event(
     ):
         event_properties["$ai_web_search_count"] = web_search_count
 
+    # Add raw usage metadata if present (all providers)
+    raw_usage = event_data["usage_stats"].get("raw_usage")
+    if raw_usage is not None:
+        # Already serialized by converters
+        event_properties["$ai_usage"] = raw_usage
+
     # Handle provider-specific fields
     if (
         event_data["provider"] == "openai"

posthog/client.py

@@ -18,6 +18,7 @@ from posthog.contexts import (
     get_capture_exception_code_variables_context,
     get_code_variables_ignore_patterns_context,
     get_code_variables_mask_patterns_context,
+    get_context_device_id,
     get_context_distinct_id,
     get_context_session_id,
     new_context,
@@ -382,6 +383,7 @@ class Client(object):
         group_properties=None,
         disable_geoip=None,
         flag_keys_to_evaluate: Optional[list[str]] = None,
+        device_id: Optional[str] = None,
     ) -> dict[str, Union[bool, str]]:
         """
         Get feature flag variants for a user by calling decide.
@@ -394,6 +396,7 @@ class Client(object):
             disable_geoip: Whether to disable GeoIP for this request.
             flag_keys_to_evaluate: A list of specific flag keys to evaluate. If provided,
                 only these flags will be evaluated, improving performance.
+            device_id: The device ID for this request.
 
         Category:
             Feature flags
@@ -405,6 +408,7 @@ class Client(object):
             group_properties,
             disable_geoip,
             flag_keys_to_evaluate,
+            device_id=device_id,
         )
         return to_values(resp_data) or {}
 
@@ -416,6 +420,7 @@ class Client(object):
         group_properties=None,
         disable_geoip=None,
         flag_keys_to_evaluate: Optional[list[str]] = None,
+        device_id: Optional[str] = None,
     ) -> dict[str, str]:
         """
         Get feature flag payloads for a user by calling decide.
@@ -428,6 +433,7 @@ class Client(object):
             disable_geoip: Whether to disable GeoIP for this request.
             flag_keys_to_evaluate: A list of specific flag keys to evaluate. If provided,
                 only these flags will be evaluated, improving performance.
+            device_id: The device ID for this request.
 
         Examples:
             ```python
@@ -444,6 +450,7 @@ class Client(object):
             group_properties,
             disable_geoip,
             flag_keys_to_evaluate,
+            device_id=device_id,
         )
         return to_payloads(resp_data) or {}
 
@@ -455,6 +462,7 @@ class Client(object):
         group_properties=None,
         disable_geoip=None,
         flag_keys_to_evaluate: Optional[list[str]] = None,
+        device_id: Optional[str] = None,
     ) -> FlagsAndPayloads:
         """
         Get feature flags and payloads for a user by calling decide.
@@ -467,6 +475,7 @@ class Client(object):
             disable_geoip: Whether to disable GeoIP for this request.
             flag_keys_to_evaluate: A list of specific flag keys to evaluate. If provided,
                 only these flags will be evaluated, improving performance.
+            device_id: The device ID for this request.
 
         Examples:
             ```python
@@ -483,6 +492,7 @@ class Client(object):
             group_properties,
             disable_geoip,
             flag_keys_to_evaluate,
+            device_id=device_id,
         )
         return to_flags_and_payloads(resp)
 
@@ -494,6 +504,7 @@ class Client(object):
         group_properties=None,
         disable_geoip=None,
         flag_keys_to_evaluate: Optional[list[str]] = None,
+        device_id: Optional[str] = None,
     ) -> FlagsResponse:
         """
         Get feature flags decision.
@@ -506,6 +517,7 @@ class Client(object):
             disable_geoip: Whether to disable GeoIP for this request.
             flag_keys_to_evaluate: A list of specific flag keys to evaluate. If provided,
                 only these flags will be evaluated, improving performance.
+            device_id: The device ID for this request.
 
         Examples:
             ```python
@@ -522,6 +534,9 @@ class Client(object):
         if distinct_id is None:
             distinct_id = get_context_distinct_id()
 
+        if device_id is None:
+            device_id = get_context_device_id()
+
         if disable_geoip is None:
             disable_geoip = self.disable_geoip
 
@@ -534,6 +549,7 @@ class Client(object):
             "person_properties": person_properties,
             "group_properties": group_properties,
             "geoip_disable": disable_geoip,
+            "device_id": device_id,
         }
 
         if flag_keys_to_evaluate:
@@ -1464,6 +1480,7 @@ class Client(object):
         only_evaluate_locally=False,
         send_feature_flag_events=True,
         disable_geoip=None,
+        device_id: Optional[str] = None,
     ):
         """
         Check if a feature flag is enabled for a user.
@@ -1477,6 +1494,7 @@ class Client(object):
             only_evaluate_locally: Whether to only evaluate locally.
             send_feature_flag_events: Whether to send feature flag events.
             disable_geoip: Whether to disable GeoIP for this request.
+            device_id: The device ID for this request.
 
         Examples:
             ```python
@@ -1499,6 +1517,7 @@ class Client(object):
             only_evaluate_locally=only_evaluate_locally,
             send_feature_flag_events=send_feature_flag_events,
             disable_geoip=disable_geoip,
+            device_id=device_id,
         )
 
         if response is None:
@@ -1530,6 +1549,7 @@ class Client(object):
         only_evaluate_locally=False,
         send_feature_flag_events=True,
         disable_geoip=None,
+        device_id: Optional[str] = None,
     ) -> Optional[FeatureFlagResult]:
         if self.disabled:
             return None
@@ -1584,6 +1604,7 @@ class Client(object):
                         person_properties,
                         group_properties,
                         disable_geoip,
+                        device_id=device_id,
                     )
                 )
                 errors = []
@@ -1656,6 +1677,7 @@ class Client(object):
         only_evaluate_locally=False,
         send_feature_flag_events=True,
         disable_geoip=None,
+        device_id: Optional[str] = None,
     ) -> Optional[FeatureFlagResult]:
         """
         Get a FeatureFlagResult object which contains the flag result and payload for a key by evaluating locally or remotely
@@ -1680,6 +1702,7 @@ class Client(object):
             only_evaluate_locally: Whether to only evaluate locally.
             send_feature_flag_events: Whether to send feature flag events.
             disable_geoip: Whether to disable GeoIP for this request.
+            device_id: The device ID for this request.
 
         Returns:
             Optional[FeatureFlagResult]: The feature flag result or None if disabled/not found.
@@ -1693,6 +1716,7 @@ class Client(object):
             only_evaluate_locally=only_evaluate_locally,
             send_feature_flag_events=send_feature_flag_events,
             disable_geoip=disable_geoip,
+            device_id=device_id,
         )
 
     def get_feature_flag(
@@ -1706,6 +1730,7 @@ class Client(object):
         only_evaluate_locally=False,
         send_feature_flag_events=True,
         disable_geoip=None,
+        device_id: Optional[str] = None,
     ) -> Optional[FlagValue]:
         """
         Get multivariate feature flag value for a user.
@@ -1719,6 +1744,7 @@ class Client(object):
             only_evaluate_locally: Whether to only evaluate locally.
             send_feature_flag_events: Whether to send feature flag events.
             disable_geoip: Whether to disable GeoIP for this request.
+            device_id: The device ID for this request.
 
         Examples:
             ```python
@@ -1741,6 +1767,7 @@ class Client(object):
             only_evaluate_locally=only_evaluate_locally,
             send_feature_flag_events=send_feature_flag_events,
             disable_geoip=disable_geoip,
+            device_id=device_id,
         )
         return feature_flag_result.get_value() if feature_flag_result else None
 
@@ -1794,6 +1821,7 @@ class Client(object):
         only_evaluate_locally=False,
         send_feature_flag_events=False,
         disable_geoip=None,
+        device_id: Optional[str] = None,
     ):
         """
         Get the payload for a feature flag.
@@ -1808,6 +1836,7 @@ class Client(object):
             only_evaluate_locally: Whether to only evaluate locally.
             send_feature_flag_events: Deprecated. Use get_feature_flag() instead if you need events.
             disable_geoip: Whether to disable GeoIP for this request.
+            device_id: The device ID for this request.
 
         Examples:
             ```python
@@ -1840,6 +1869,7 @@ class Client(object):
             only_evaluate_locally=only_evaluate_locally,
             send_feature_flag_events=send_feature_flag_events,
             disable_geoip=disable_geoip,
+            device_id=device_id,
         )
         return feature_flag_result.payload if feature_flag_result else None
 
@@ -1851,6 +1881,7 @@ class Client(object):
         person_properties: dict[str, str],
         group_properties: dict[str, str],
         disable_geoip: Optional[bool],
+        device_id: Optional[str] = None,
     ) -> tuple[Optional[FeatureFlag], Optional[str], Optional[int], bool]:
         """
         Calls /flags and returns the flag details, request id, evaluated at timestamp,
@@ -1863,6 +1894,7 @@ class Client(object):
             group_properties,
             disable_geoip,
             flag_keys_to_evaluate=[key],
+            device_id=device_id,
         )
         request_id = resp_data.get("requestId")
         evaluated_at = resp_data.get("evaluatedAt")
@@ -1987,6 +2019,7 @@ class Client(object):
         only_evaluate_locally=False,
         disable_geoip=None,
         flag_keys_to_evaluate: Optional[list[str]] = None,
+        device_id: Optional[str] = None,
     ) -> Optional[dict[str, Union[bool, str]]]:
         """
         Get all feature flags for a user.
@@ -2000,6 +2033,7 @@ class Client(object):
             disable_geoip: Whether to disable GeoIP for this request.
             flag_keys_to_evaluate: A list of specific flag keys to evaluate. If provided,
                 only these flags will be evaluated, improving performance.
+            device_id: The device ID for this request.
 
         Examples:
             ```python
@@ -2017,6 +2051,7 @@ class Client(object):
             only_evaluate_locally=only_evaluate_locally,
             disable_geoip=disable_geoip,
             flag_keys_to_evaluate=flag_keys_to_evaluate,
+            device_id=device_id,
         )
 
         return response["featureFlags"]
@@ -2031,6 +2066,7 @@ class Client(object):
         only_evaluate_locally=False,
         disable_geoip=None,
         flag_keys_to_evaluate: Optional[list[str]] = None,
+        device_id: Optional[str] = None,
     ) -> FlagsAndPayloads:
         """
         Get all feature flags and their payloads for a user.
@@ -2044,6 +2080,7 @@ class Client(object):
             disable_geoip: Whether to disable GeoIP for this request.
             flag_keys_to_evaluate: A list of specific flag keys to evaluate. If provided,
                 only these flags will be evaluated, improving performance.
+            device_id: The device ID for this request.
 
         Examples:
             ```python
@@ -2079,6 +2116,7 @@ class Client(object):
                     group_properties=group_properties,
                     disable_geoip=disable_geoip,
                     flag_keys_to_evaluate=flag_keys_to_evaluate,
+                    device_id=device_id,
                 )
                 return to_flags_and_payloads(decide_response)
             except Exception as e:

posthog/consumer.py

@@ -84,12 +84,16 @@ class Consumer(Thread):
             self.log.error("error uploading: %s", e)
             success = False
             if self.on_error:
-                self.on_error(e, batch)
+                try:
+                    self.on_error(e, batch)
+                except Exception as e:
+                    self.log.error("on_error handler failed: %s", e)
         finally:
             # mark items as acknowledged from queue
             for item in batch:
                 self.queue.task_done()
-            return success
+
+        return success
 
     def next(self):
         """Return the next batch of items to upload."""

posthog/contexts.py

@@ -21,6 +21,7 @@ class ContextScope:
         self.capture_exceptions = capture_exceptions
         self.session_id: Optional[str] = None
         self.distinct_id: Optional[str] = None
+        self.device_id: Optional[str] = None
         self.tags: Dict[str, Any] = {}
         self.capture_exception_code_variables: Optional[bool] = None
         self.code_variables_mask_patterns: Optional[list] = None
@@ -32,6 +33,9 @@ class ContextScope:
     def set_distinct_id(self, distinct_id: str):
         self.distinct_id = distinct_id
 
+    def set_device_id(self, device_id: str):
+        self.device_id = device_id
+
     def add_tag(self, key: str, value: Any):
         self.tags[key] = value
 
@@ -61,6 +65,13 @@ class ContextScope:
             return self.parent.get_distinct_id()
         return None
 
+    def get_device_id(self) -> Optional[str]:
+        if self.device_id is not None:
+            return self.device_id
+        if self.parent is not None and not self.fresh:
+            return self.parent.get_device_id()
+        return None
+
     def collect_tags(self) -> Dict[str, Any]:
         if self.parent and not self.fresh:
             # We want child tags to take precedence over parent tags,
@@ -275,6 +286,39 @@ def get_context_distinct_id() -> Optional[str]:
     return None
 
 
+def set_context_device_id(device_id: str) -> None:
+    """
+    Set the device ID for the current context, associating all feature flag requests in this or
+    child contexts with the given device ID (unless set_context_device_id is called again).
+    Entering a fresh context will clear the context-level device ID.
+
+    Args:
+        device_id: The device ID to associate with the current context and its children.
+
+    Category:
+        Contexts
+    """
+    current_context = _get_current_context()
+    if current_context:
+        current_context.set_device_id(device_id)
+
+
+def get_context_device_id() -> Optional[str]:
+    """
+    Get the device ID for the current context.
+
+    Returns:
+        The device ID if set, None otherwise
+
+    Category:
+        Contexts
+    """
+    current_context = _get_current_context()
+    if current_context:
+        return current_context.get_device_id()
+    return None
+
+
 def set_capture_exception_code_variables_context(enabled: bool) -> None:
     """
     Set whether code variables are captured for the current context.

posthog/test/ai/anthropic/test_anthropic.py

@@ -1,3 +1,4 @@
+import json
 from unittest.mock import patch
 
 import pytest
@@ -306,6 +307,15 @@ def test_basic_completion(mock_client, mock_anthropic_response):
         assert props["$ai_http_status"] == 200
         assert props["foo"] == "bar"
         assert isinstance(props["$ai_latency"], float)
+        # Verify raw usage metadata is passed for backend processing
+        assert "$ai_usage" in props
+        assert props["$ai_usage"] is not None
+        # Verify it's JSON-serializable
+        json.dumps(props["$ai_usage"])
+        # Verify it has expected structure
+        assert isinstance(props["$ai_usage"], dict)
+        assert "input_tokens" in props["$ai_usage"]
+        assert "output_tokens" in props["$ai_usage"]
 
 
 def test_groups(mock_client, mock_anthropic_response):
@@ -918,6 +928,16 @@ def test_streaming_with_tool_calls(mock_client, mock_anthropic_stream_with_tools
         assert props["$ai_cache_read_input_tokens"] == 5
         assert props["$ai_cache_creation_input_tokens"] == 0
 
+        # Verify raw usage is captured in streaming mode (merged from events)
+        assert "$ai_usage" in props
+        assert props["$ai_usage"] is not None
+        # Verify it's JSON-serializable
+        json.dumps(props["$ai_usage"])
+        # Verify it has expected structure (merged from message_start and message_delta)
+        assert isinstance(props["$ai_usage"], dict)
+        assert "input_tokens" in props["$ai_usage"]
+        assert "output_tokens" in props["$ai_usage"]
+
 
 def test_async_streaming_with_tool_calls(mock_client, mock_anthropic_stream_with_tools):
     """Test that tool calls are properly captured in async streaming mode."""

posthog/test/ai/gemini/test_gemini.py

@@ -1,3 +1,4 @@
+import json
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -34,6 +35,13 @@ def mock_gemini_response():
     # Ensure cache and reasoning tokens are not present (not MagicMock)
     mock_usage.cached_content_token_count = 0
     mock_usage.thoughts_token_count = 0
+    # Make model_dump() return a proper dict for serialization
+    mock_usage.model_dump.return_value = {
+        "prompt_token_count": 20,
+        "candidates_token_count": 10,
+        "cached_content_token_count": 0,
+        "thoughts_token_count": 0,
+    }
     mock_response.usage_metadata = mock_usage
 
     mock_candidate = MagicMock()
@@ -69,6 +77,13 @@ def mock_gemini_response_with_function_calls():
     mock_usage.candidates_token_count = 15
     mock_usage.cached_content_token_count = 0
     mock_usage.thoughts_token_count = 0
+    # Make model_dump() return a proper dict for serialization
+    mock_usage.model_dump.return_value = {
+        "prompt_token_count": 25,
+        "candidates_token_count": 15,
+        "cached_content_token_count": 0,
+        "thoughts_token_count": 0,
+    }
     mock_response.usage_metadata = mock_usage
 
     # Mock function call
@@ -117,6 +132,13 @@ def mock_gemini_response_function_calls_only():
     mock_usage.candidates_token_count = 12
     mock_usage.cached_content_token_count = 0
     mock_usage.thoughts_token_count = 0
+    # Make model_dump() return a proper dict for serialization
+    mock_usage.model_dump.return_value = {
+        "prompt_token_count": 30,
+        "candidates_token_count": 12,
+        "cached_content_token_count": 0,
+        "thoughts_token_count": 0,
+    }
     mock_response.usage_metadata = mock_usage
 
     # Mock function call
@@ -174,6 +196,15 @@ def test_new_client_basic_generation(
     assert props["foo"] == "bar"
     assert "$ai_trace_id" in props
     assert props["$ai_latency"] > 0
+    # Verify raw usage metadata is passed for backend processing
+    assert "$ai_usage" in props
+    assert props["$ai_usage"] is not None
+    # Verify it's JSON-serializable
+    json.dumps(props["$ai_usage"])
+    # Verify it has expected structure
+    assert isinstance(props["$ai_usage"], dict)
+    assert "prompt_token_count" in props["$ai_usage"]
+    assert "candidates_token_count" in props["$ai_usage"]
 
 
 def test_new_client_streaming_with_generate_content_stream(
@@ -810,6 +841,13 @@ def test_streaming_cache_and_reasoning_tokens(mock_client, mock_google_genai_cli
     chunk1_usage.candidates_token_count = 5
     chunk1_usage.cached_content_token_count = 30  # Cache tokens
     chunk1_usage.thoughts_token_count = 0
+    # Make model_dump() return a proper dict for serialization
+    chunk1_usage.model_dump.return_value = {
+        "prompt_token_count": 100,
+        "candidates_token_count": 5,
+        "cached_content_token_count": 30,
+        "thoughts_token_count": 0,
+    }
     chunk1.usage_metadata = chunk1_usage
 
     chunk2 = MagicMock()
@@ -819,6 +857,13 @@ def test_streaming_cache_and_reasoning_tokens(mock_client, mock_google_genai_cli
     chunk2_usage.candidates_token_count = 10
     chunk2_usage.cached_content_token_count = 30  # Same cache tokens
     chunk2_usage.thoughts_token_count = 5  # Reasoning tokens
+    # Make model_dump() return a proper dict for serialization
+    chunk2_usage.model_dump.return_value = {
+        "prompt_token_count": 100,
+        "candidates_token_count": 10,
+        "cached_content_token_count": 30,
+        "thoughts_token_count": 5,
+    }
     chunk2.usage_metadata = chunk2_usage
 
     mock_stream = iter([chunk1, chunk2])
@@ -848,6 +893,16 @@ def test_streaming_cache_and_reasoning_tokens(mock_client, mock_google_genai_cli
     assert props["$ai_cache_read_input_tokens"] == 30
     assert props["$ai_reasoning_tokens"] == 5
 
+    # Verify raw usage is captured in streaming mode (merged from chunks)
+    assert "$ai_usage" in props
+    assert props["$ai_usage"] is not None
+    # Verify it's JSON-serializable
+    json.dumps(props["$ai_usage"])
+    # Verify it has expected structure
+    assert isinstance(props["$ai_usage"], dict)
+    assert "prompt_token_count" in props["$ai_usage"]
+    assert "candidates_token_count" in props["$ai_usage"]
+
 
 def test_web_search_grounding(mock_client, mock_google_genai_client):
     """Test web search detection via grounding_metadata."""

posthog/test/ai/openai/test_openai.py

@@ -1,3 +1,4 @@
+import json
 import time
 from unittest.mock import AsyncMock, patch
 
@@ -496,6 +497,15 @@ def test_basic_completion(mock_client, mock_openai_response):
         assert props["$ai_http_status"] == 200
         assert props["foo"] == "bar"
         assert isinstance(props["$ai_latency"], float)
+        # Verify raw usage metadata is passed for backend processing
+        assert "$ai_usage" in props
+        assert props["$ai_usage"] is not None
+        # Verify it's JSON-serializable
+        json.dumps(props["$ai_usage"])
+        # Verify it has expected structure
+        assert isinstance(props["$ai_usage"], dict)
+        assert "prompt_tokens" in props["$ai_usage"]
+        assert "completion_tokens" in props["$ai_usage"]
 
 
 def test_embeddings(mock_client, mock_embedding_response):
@@ -922,6 +932,16 @@ def test_streaming_with_tool_calls(mock_client, streaming_tool_call_chunks):
         assert props["$ai_input_tokens"] == 20
         assert props["$ai_output_tokens"] == 15
 
+        # Verify raw usage is captured in streaming mode
+        assert "$ai_usage" in props
+        assert props["$ai_usage"] is not None
+        # Verify it's JSON-serializable
+        json.dumps(props["$ai_usage"])
+        # Verify it has expected structure (merged from chunks)
+        assert isinstance(props["$ai_usage"], dict)
+        assert "prompt_tokens" in props["$ai_usage"]
+        assert "completion_tokens" in props["$ai_usage"]
+
 
 # test responses api
 def test_responses_api(mock_client, mock_openai_response_with_responses_api):

posthog/test/ai/openai_agents/__init__.py

@@ -0,0 +1 @@
+# Tests for OpenAI Agents SDK integration

posthog/test/ai/openai_agents/test_processor.py

@@ -0,0 +1,810 @@
+import logging
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+try:
+    from agents.tracing.span_data import (
+        AgentSpanData,
+        CustomSpanData,
+        FunctionSpanData,
+        GenerationSpanData,
+        GuardrailSpanData,
+        HandoffSpanData,
+        ResponseSpanData,
+        SpeechSpanData,
+        TranscriptionSpanData,
+    )
+
+    from posthog.ai.openai_agents import PostHogTracingProcessor, instrument
+
+    OPENAI_AGENTS_AVAILABLE = True
+except ImportError:
+    OPENAI_AGENTS_AVAILABLE = False
+
+
+# Skip all tests if OpenAI Agents SDK is not available
+pytestmark = pytest.mark.skipif(
+    not OPENAI_AGENTS_AVAILABLE, reason="OpenAI Agents SDK is not available"
+)
+
+
+@pytest.fixture(scope="function")
+def mock_client():
+    client = MagicMock()
+    client.privacy_mode = False
+    logging.getLogger("posthog").setLevel(logging.DEBUG)
+    return client
+
+
+@pytest.fixture(scope="function")
+def processor(mock_client):
+    return PostHogTracingProcessor(
+        client=mock_client,
+        distinct_id="test-user",
+        privacy_mode=False,
+    )
+
+
+@pytest.fixture
+def mock_trace():
+    trace = MagicMock()
+    trace.trace_id = "trace_123456789"
+    trace.name = "Test Workflow"
+    trace.group_id = "group_123"
+    trace.metadata = {"key": "value"}
+    return trace
+
+
+@pytest.fixture
+def mock_span():
+    span = MagicMock()
+    span.trace_id = "trace_123456789"
+    span.span_id = "span_987654321"
+    span.parent_id = None
+    span.started_at = "2024-01-01T00:00:00Z"
+    span.ended_at = "2024-01-01T00:00:01Z"
+    span.error = None
+    return span
+
+
+class TestPostHogTracingProcessor:
+    """Tests for the PostHogTracingProcessor class."""
+
+    def test_initialization(self, mock_client):
+        """Test processor initializes correctly."""
+        processor = PostHogTracingProcessor(
+            client=mock_client,
+            distinct_id="user@example.com",
+            privacy_mode=True,
+            groups={"company": "acme"},
+            properties={"env": "test"},
+        )
+
+        assert processor._client == mock_client
+        assert processor._distinct_id == "user@example.com"
+        assert processor._privacy_mode is True
+        assert processor._groups == {"company": "acme"}
+        assert processor._properties == {"env": "test"}
+
+    def test_initialization_with_callable_distinct_id(self, mock_client, mock_trace):
+        """Test processor with callable distinct_id resolver."""
+
+        def resolver(trace):
+            return trace.metadata.get("user_id", "default")
+
+        processor = PostHogTracingProcessor(
+            client=mock_client,
+            distinct_id=resolver,
+        )
+
+        mock_trace.metadata = {"user_id": "resolved-user"}
+        distinct_id = processor._get_distinct_id(mock_trace)
+        assert distinct_id == "resolved-user"
+
+    def test_on_trace_start_stores_metadata(self, processor, mock_client, mock_trace):
+        """Test that on_trace_start stores metadata but does not capture an event."""
+        processor.on_trace_start(mock_trace)
+
+        mock_client.capture.assert_not_called()
+        assert mock_trace.trace_id in processor._trace_metadata
+
+    def test_on_trace_end_captures_ai_trace(self, processor, mock_client, mock_trace):
+        """Test that on_trace_end captures $ai_trace event."""
+        processor.on_trace_start(mock_trace)
+        processor.on_trace_end(mock_trace)
+
+        mock_client.capture.assert_called_once()
+        call_kwargs = mock_client.capture.call_args[1]
+
+        assert call_kwargs["event"] == "$ai_trace"
+        assert call_kwargs["distinct_id"] == "test-user"
+        assert call_kwargs["properties"]["$ai_trace_id"] == "trace_123456789"
+        assert call_kwargs["properties"]["$ai_trace_name"] == "Test Workflow"
+        assert call_kwargs["properties"]["$ai_provider"] == "openai"
+        assert call_kwargs["properties"]["$ai_framework"] == "openai-agents"
+        assert "$ai_latency" in call_kwargs["properties"]
+
+    def test_personless_mode_when_no_distinct_id(self, mock_client, mock_trace):
+        """Test that trace events use personless mode when no distinct_id is provided."""
+        processor = PostHogTracingProcessor(
+            client=mock_client,
+        )
+
+        processor.on_trace_start(mock_trace)
+        processor.on_trace_end(mock_trace)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["properties"]["$process_person_profile"] is False
+        # Should fallback to trace_id as the distinct_id
+        assert call_kwargs["distinct_id"] == mock_trace.trace_id
+
+    def test_personless_mode_for_spans_when_no_distinct_id(
+        self, mock_client, mock_trace, mock_span
+    ):
+        """Test that span events use personless mode when no distinct_id is provided."""
+        processor = PostHogTracingProcessor(
+            client=mock_client,
+        )
+
+        processor.on_trace_start(mock_trace)
+        mock_client.capture.reset_mock()
+
+        span_data = GenerationSpanData(model="gpt-4o")
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["properties"]["$process_person_profile"] is False
+        assert call_kwargs["distinct_id"] == mock_span.trace_id
+
+    def test_personless_mode_when_callable_returns_none(
+        self, mock_client, mock_trace, mock_span
+    ):
+        """Test personless mode when callable distinct_id returns None."""
+
+        def resolver(trace):
+            return None  # Simulate no user ID available
+
+        processor = PostHogTracingProcessor(
+            client=mock_client,
+            distinct_id=resolver,
+        )
+
+        processor.on_trace_start(mock_trace)
+        mock_client.capture.reset_mock()
+
+        span_data = GenerationSpanData(model="gpt-4o")
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["properties"]["$process_person_profile"] is False
+        assert call_kwargs["distinct_id"] == mock_span.trace_id
+
+    def test_person_profile_when_distinct_id_provided(self, mock_client, mock_trace):
+        """Test that events create person profiles when distinct_id is provided."""
+        processor = PostHogTracingProcessor(
+            client=mock_client,
+            distinct_id="real-user",
+        )
+
+        processor.on_trace_start(mock_trace)
+        processor.on_trace_end(mock_trace)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert "$process_person_profile" not in call_kwargs["properties"]
+
+    def test_on_trace_end_clears_metadata(self, processor, mock_client, mock_trace):
+        """Test that on_trace_end clears stored trace metadata."""
+        processor.on_trace_start(mock_trace)
+        assert mock_trace.trace_id in processor._trace_metadata
+
+        processor.on_trace_end(mock_trace)
+        assert mock_trace.trace_id not in processor._trace_metadata
+        # Also verify it captured the event
+        mock_client.capture.assert_called_once()
+
+    def test_on_span_start_tracks_time(self, processor, mock_span):
+        """Test that on_span_start records start time."""
+        processor.on_span_start(mock_span)
+        assert mock_span.span_id in processor._span_start_times
+
+    def test_generation_span_mapping(self, processor, mock_client, mock_span):
+        """Test GenerationSpanData maps to $ai_generation event."""
+        span_data = GenerationSpanData(
+            input=[{"role": "user", "content": "Hello"}],
+            output=[{"role": "assistant", "content": "Hi there!"}],
+            model="gpt-4o",
+            model_config={"temperature": 0.7, "max_tokens": 100},
+            usage={"input_tokens": 10, "output_tokens": 20},
+        )
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        mock_client.capture.assert_called_once()
+        call_kwargs = mock_client.capture.call_args[1]
+
+        assert call_kwargs["event"] == "$ai_generation"
+        assert call_kwargs["properties"]["$ai_trace_id"] == "trace_123456789"
+        assert call_kwargs["properties"]["$ai_span_id"] == "span_987654321"
+        assert call_kwargs["properties"]["$ai_provider"] == "openai"
+        assert call_kwargs["properties"]["$ai_framework"] == "openai-agents"
+        assert call_kwargs["properties"]["$ai_model"] == "gpt-4o"
+        assert call_kwargs["properties"]["$ai_input_tokens"] == 10
+        assert call_kwargs["properties"]["$ai_output_tokens"] == 20
+        assert call_kwargs["properties"]["$ai_input"] == [
+            {"role": "user", "content": "Hello"}
+        ]
+        assert call_kwargs["properties"]["$ai_output_choices"] == [
+            {"role": "assistant", "content": "Hi there!"}
+        ]
+
+    def test_generation_span_with_reasoning_tokens(
+        self, processor, mock_client, mock_span
+    ):
+        """Test GenerationSpanData includes reasoning tokens when present."""
+        span_data = GenerationSpanData(
+            model="o1-preview",
+            usage={
+                "input_tokens": 100,
+                "output_tokens": 500,
+                "reasoning_tokens": 400,
+            },
+        )
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["properties"]["$ai_reasoning_tokens"] == 400
+
+    def test_function_span_mapping(self, processor, mock_client, mock_span):
+        """Test FunctionSpanData maps to $ai_span event with type=tool."""
+        span_data = FunctionSpanData(
+            name="get_weather",
+            input='{"city": "San Francisco"}',
+            output="Sunny, 72F",
+        )
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+
+        assert call_kwargs["event"] == "$ai_span"
+        assert call_kwargs["properties"]["$ai_span_name"] == "get_weather"
+        assert call_kwargs["properties"]["$ai_span_type"] == "tool"
+        assert (
+            call_kwargs["properties"]["$ai_input_state"] == '{"city": "San Francisco"}'
+        )
+        assert call_kwargs["properties"]["$ai_output_state"] == "Sunny, 72F"
+
+    def test_agent_span_mapping(self, processor, mock_client, mock_span):
+        """Test AgentSpanData maps to $ai_span event with type=agent."""
+        span_data = AgentSpanData(
+            name="CustomerServiceAgent",
+            handoffs=["TechnicalAgent", "BillingAgent"],
+            tools=["search", "get_order"],
+            output_type="str",
+        )
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+
+        assert call_kwargs["event"] == "$ai_span"
+        assert call_kwargs["properties"]["$ai_span_name"] == "CustomerServiceAgent"
+        assert call_kwargs["properties"]["$ai_span_type"] == "agent"
+        assert call_kwargs["properties"]["$ai_agent_handoffs"] == [
+            "TechnicalAgent",
+            "BillingAgent",
+        ]
+        assert call_kwargs["properties"]["$ai_agent_tools"] == ["search", "get_order"]
+
+    def test_handoff_span_mapping(self, processor, mock_client, mock_span):
+        """Test HandoffSpanData maps to $ai_span event with type=handoff."""
+        span_data = HandoffSpanData(
+            from_agent="TriageAgent",
+            to_agent="TechnicalAgent",
+        )
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+
+        assert call_kwargs["event"] == "$ai_span"
+        assert call_kwargs["properties"]["$ai_span_type"] == "handoff"
+        assert call_kwargs["properties"]["$ai_handoff_from_agent"] == "TriageAgent"
+        assert call_kwargs["properties"]["$ai_handoff_to_agent"] == "TechnicalAgent"
+        assert (
+            call_kwargs["properties"]["$ai_span_name"]
+            == "TriageAgent -> TechnicalAgent"
+        )
+
+    def test_guardrail_span_mapping(self, processor, mock_client, mock_span):
+        """Test GuardrailSpanData maps to $ai_span event with type=guardrail."""
+        span_data = GuardrailSpanData(
+            name="ContentFilter",
+            triggered=True,
+        )
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+
+        assert call_kwargs["event"] == "$ai_span"
+        assert call_kwargs["properties"]["$ai_span_name"] == "ContentFilter"
+        assert call_kwargs["properties"]["$ai_span_type"] == "guardrail"
+        assert call_kwargs["properties"]["$ai_guardrail_triggered"] is True
+
+    def test_custom_span_mapping(self, processor, mock_client, mock_span):
+        """Test CustomSpanData maps to $ai_span event with type=custom."""
+        span_data = CustomSpanData(
+            name="database_query",
+            data={"query": "SELECT * FROM users", "rows": 100},
+        )
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+
+        assert call_kwargs["event"] == "$ai_span"
+        assert call_kwargs["properties"]["$ai_span_name"] == "database_query"
+        assert call_kwargs["properties"]["$ai_span_type"] == "custom"
+        assert call_kwargs["properties"]["$ai_custom_data"] == {
+            "query": "SELECT * FROM users",
+            "rows": 100,
+        }
+
+    def test_privacy_mode_redacts_content(self, mock_client, mock_span):
+        """Test that privacy_mode redacts input/output content."""
+        processor = PostHogTracingProcessor(
+            client=mock_client,
+            distinct_id="test-user",
+            privacy_mode=True,
+        )
+
+        span_data = GenerationSpanData(
+            input=[{"role": "user", "content": "Secret message"}],
+            output=[{"role": "assistant", "content": "Secret response"}],
+            model="gpt-4o",
+            usage={"input_tokens": 10, "output_tokens": 20},
+        )
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+
+        # Content should be redacted
+        assert call_kwargs["properties"]["$ai_input"] is None
+        assert call_kwargs["properties"]["$ai_output_choices"] is None
+        # Token counts should still be present
+        assert call_kwargs["properties"]["$ai_input_tokens"] == 10
+        assert call_kwargs["properties"]["$ai_output_tokens"] == 20
+
+    def test_error_handling_in_span(self, processor, mock_client, mock_span):
+        """Test that span errors are captured correctly."""
+        span_data = GenerationSpanData(model="gpt-4o")
+        mock_span.span_data = span_data
+        mock_span.error = {"message": "Rate limit exceeded", "data": {"code": 429}}
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+
+        assert call_kwargs["properties"]["$ai_is_error"] is True
+        assert call_kwargs["properties"]["$ai_error"] == "Rate limit exceeded"
+
+    def test_generation_span_includes_total_tokens(
+        self, processor, mock_client, mock_span
+    ):
+        """Test that $ai_total_tokens is calculated and included."""
+        span_data = GenerationSpanData(
+            model="gpt-4o",
+            usage={"input_tokens": 100, "output_tokens": 50},
+        )
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["properties"]["$ai_total_tokens"] == 150
+
+    def test_error_type_categorization_model_behavior(
+        self, processor, mock_client, mock_span
+    ):
+        """Test that ModelBehaviorError is categorized correctly."""
+        span_data = GenerationSpanData(model="gpt-4o")
+        mock_span.span_data = span_data
+        mock_span.error = {
+            "message": "ModelBehaviorError: Invalid JSON output",
+            "type": "ModelBehaviorError",
+        }
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["properties"]["$ai_error_type"] == "model_behavior_error"
+
+    def test_error_type_categorization_user_error(
+        self, processor, mock_client, mock_span
+    ):
+        """Test that UserError is categorized correctly."""
+        span_data = GenerationSpanData(model="gpt-4o")
+        mock_span.span_data = span_data
+        mock_span.error = {"message": "UserError: Tool failed", "type": "UserError"}
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["properties"]["$ai_error_type"] == "user_error"
+
+    def test_error_type_categorization_input_guardrail(
+        self, processor, mock_client, mock_span
+    ):
+        """Test that InputGuardrailTripwireTriggered is categorized correctly."""
+        span_data = GenerationSpanData(model="gpt-4o")
+        mock_span.span_data = span_data
+        mock_span.error = {
+            "message": "InputGuardrailTripwireTriggered: Content blocked"
+        }
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert (
+            call_kwargs["properties"]["$ai_error_type"] == "input_guardrail_triggered"
+        )
+
+    def test_error_type_categorization_output_guardrail(
+        self, processor, mock_client, mock_span
+    ):
+        """Test that OutputGuardrailTripwireTriggered is categorized correctly."""
+        span_data = GenerationSpanData(model="gpt-4o")
+        mock_span.span_data = span_data
+        mock_span.error = {
+            "message": "OutputGuardrailTripwireTriggered: Response blocked"
+        }
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert (
+            call_kwargs["properties"]["$ai_error_type"] == "output_guardrail_triggered"
+        )
+
+    def test_error_type_categorization_max_turns(
+        self, processor, mock_client, mock_span
+    ):
+        """Test that MaxTurnsExceeded is categorized correctly."""
+        span_data = GenerationSpanData(model="gpt-4o")
+        mock_span.span_data = span_data
+        mock_span.error = {"message": "MaxTurnsExceeded: Agent exceeded maximum turns"}
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["properties"]["$ai_error_type"] == "max_turns_exceeded"
+
+    def test_error_type_categorization_unknown(self, processor, mock_client, mock_span):
+        """Test that unknown errors are categorized as unknown."""
+        span_data = GenerationSpanData(model="gpt-4o")
+        mock_span.span_data = span_data
+        mock_span.error = {"message": "Some random error occurred"}
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["properties"]["$ai_error_type"] == "unknown"
+
+    def test_response_span_with_output_and_total_tokens(
+        self, processor, mock_client, mock_span
+    ):
+        """Test ResponseSpanData includes output choices and total tokens."""
+        # Create a mock response object
+        mock_response = MagicMock()
+        mock_response.id = "resp_123"
+        mock_response.model = "gpt-4o"
+        mock_response.output = [{"type": "message", "content": "Hello!"}]
+        mock_response.usage = MagicMock()
+        mock_response.usage.input_tokens = 25
+        mock_response.usage.output_tokens = 10
+
+        span_data = ResponseSpanData(
+            response=mock_response,
+            input="Hello, world!",
+        )
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+
+        assert call_kwargs["event"] == "$ai_generation"
+        assert call_kwargs["properties"]["$ai_total_tokens"] == 35
+        assert call_kwargs["properties"]["$ai_output_choices"] == [
+            {"type": "message", "content": "Hello!"}
+        ]
+        assert call_kwargs["properties"]["$ai_response_id"] == "resp_123"
+
+    def test_speech_span_with_pass_through_properties(
+        self, processor, mock_client, mock_span
+    ):
+        """Test SpeechSpanData includes pass-through properties."""
+        span_data = SpeechSpanData(
+            input="Hello, how can I help you?",
+            output="base64_audio_data",
+            output_format="pcm",
+            model="tts-1",
+            model_config={"voice": "alloy", "speed": 1.0},
+            first_content_at="2024-01-01T00:00:00.500Z",
+        )
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+
+        assert call_kwargs["event"] == "$ai_span"
+        assert call_kwargs["properties"]["$ai_span_type"] == "speech"
+        assert call_kwargs["properties"]["$ai_model"] == "tts-1"
+        # Pass-through properties (no $ai_ prefix)
+        assert (
+            call_kwargs["properties"]["first_content_at"] == "2024-01-01T00:00:00.500Z"
+        )
+        assert call_kwargs["properties"]["audio_output_format"] == "pcm"
+        assert call_kwargs["properties"]["model_config"] == {
+            "voice": "alloy",
+            "speed": 1.0,
+        }
+        # Text input should be captured
+        assert call_kwargs["properties"]["$ai_input"] == "Hello, how can I help you?"
+
+    def test_transcription_span_with_pass_through_properties(
+        self, processor, mock_client, mock_span
+    ):
+        """Test TranscriptionSpanData includes pass-through properties."""
+        span_data = TranscriptionSpanData(
+            input="base64_audio_data",
+            input_format="pcm",
+            output="This is the transcribed text.",
+            model="whisper-1",
+            model_config={"language": "en"},
+        )
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+
+        assert call_kwargs["event"] == "$ai_span"
+        assert call_kwargs["properties"]["$ai_span_type"] == "transcription"
+        assert call_kwargs["properties"]["$ai_model"] == "whisper-1"
+        # Pass-through properties (no $ai_ prefix)
+        assert call_kwargs["properties"]["audio_input_format"] == "pcm"
+        assert call_kwargs["properties"]["model_config"] == {"language": "en"}
+        # Transcription output should be captured
+        assert (
+            call_kwargs["properties"]["$ai_output_state"]
+            == "This is the transcribed text."
+        )
+
+    def test_latency_calculation(self, processor, mock_client, mock_span):
+        """Test that latency is calculated correctly."""
+        span_data = GenerationSpanData(model="gpt-4o")
+        mock_span.span_data = span_data
+
+        with patch("time.time") as mock_time:
+            mock_time.return_value = 1000.0
+            processor.on_span_start(mock_span)
+
+            mock_time.return_value = 1001.5  # 1.5 seconds later
+            processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["properties"]["$ai_latency"] == pytest.approx(1.5, rel=0.01)
+
+    def test_groups_included_in_events(self, mock_client, mock_trace, mock_span):
+        """Test that groups are included in captured events."""
+        processor = PostHogTracingProcessor(
+            client=mock_client,
+            distinct_id="test-user",
+            groups={"company": "acme", "team": "engineering"},
+        )
+
+        processor.on_trace_start(mock_trace)
+        processor.on_trace_end(mock_trace)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["groups"] == {"company": "acme", "team": "engineering"}
+
+    def test_additional_properties_included(self, mock_client, mock_trace):
+        """Test that additional properties are included in events."""
+        processor = PostHogTracingProcessor(
+            client=mock_client,
+            distinct_id="test-user",
+            properties={"environment": "production", "version": "1.0"},
+        )
+
+        processor.on_trace_start(mock_trace)
+        processor.on_trace_end(mock_trace)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["properties"]["environment"] == "production"
+        assert call_kwargs["properties"]["version"] == "1.0"
+
+    def test_shutdown_clears_state(self, processor):
+        """Test that shutdown clears internal state."""
+        processor._span_start_times["span_1"] = 1000.0
+        processor._trace_metadata["trace_1"] = {"name": "test"}
+
+        processor.shutdown()
+
+        assert len(processor._span_start_times) == 0
+        assert len(processor._trace_metadata) == 0
+
+    def test_force_flush_calls_client_flush(self, processor, mock_client):
+        """Test that force_flush calls client.flush()."""
+        processor.force_flush()
+        mock_client.flush.assert_called_once()
+
+    def test_generation_span_with_no_usage(self, processor, mock_client, mock_span):
+        """Test GenerationSpanData with no usage data defaults to zero tokens."""
+        span_data = GenerationSpanData(model="gpt-4o")
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["properties"]["$ai_input_tokens"] == 0
+        assert call_kwargs["properties"]["$ai_output_tokens"] == 0
+        assert call_kwargs["properties"]["$ai_total_tokens"] == 0
+
+    def test_generation_span_with_partial_usage(
+        self, processor, mock_client, mock_span
+    ):
+        """Test GenerationSpanData with only input_tokens present."""
+        span_data = GenerationSpanData(
+            model="gpt-4o",
+            usage={"input_tokens": 42},
+        )
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["properties"]["$ai_input_tokens"] == 42
+        assert call_kwargs["properties"]["$ai_output_tokens"] == 0
+        assert call_kwargs["properties"]["$ai_total_tokens"] == 42
+
+    def test_error_type_categorization_by_type_field_only(
+        self, processor, mock_client, mock_span
+    ):
+        """Test error categorization works when only the type field matches."""
+        span_data = GenerationSpanData(model="gpt-4o")
+        mock_span.span_data = span_data
+        mock_span.error = {
+            "message": "Something went wrong",
+            "type": "ModelBehaviorError",
+        }
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["properties"]["$ai_error_type"] == "model_behavior_error"
+
+    def test_distinct_id_resolved_from_trace_for_spans(
+        self, mock_client, mock_trace, mock_span
+    ):
+        """Test that spans use the distinct_id resolved at trace start."""
+
+        def resolver(trace):
+            return f"user-{trace.name}"
+
+        processor = PostHogTracingProcessor(
+            client=mock_client,
+            distinct_id=resolver,
+        )
+
+        # Start trace - this resolves and stores distinct_id
+        processor.on_trace_start(mock_trace)
+        mock_client.capture.reset_mock()
+
+        # End a span - should use the stored distinct_id from trace
+        span_data = GenerationSpanData(model="gpt-4o")
+        mock_span.span_data = span_data
+
+        processor.on_span_start(mock_span)
+        processor.on_span_end(mock_span)
+
+        call_kwargs = mock_client.capture.call_args[1]
+        assert call_kwargs["distinct_id"] == "user-Test Workflow"
+
+    def test_eviction_of_stale_entries(self, mock_client):
+        """Test that stale entries are evicted when max is exceeded."""
+        processor = PostHogTracingProcessor(
+            client=mock_client,
+            distinct_id="test-user",
+        )
+        processor._max_tracked_entries = 10
+
+        # Fill beyond max
+        for i in range(15):
+            processor._span_start_times[f"span_{i}"] = float(i)
+            processor._trace_metadata[f"trace_{i}"] = {"name": f"trace_{i}"}
+
+        processor._evict_stale_entries()
+
+        # Should have evicted half
+        assert len(processor._span_start_times) <= 10
+        assert len(processor._trace_metadata) <= 10
+
+
+class TestInstrumentHelper:
+    """Tests for the instrument() convenience function."""
+
+    def test_instrument_registers_processor(self, mock_client):
+        """Test that instrument() registers a processor."""
+        with patch("agents.tracing.add_trace_processor") as mock_add:
+            processor = instrument(
+                client=mock_client,
+                distinct_id="test-user",
+            )
+
+            mock_add.assert_called_once_with(processor)
+            assert isinstance(processor, PostHogTracingProcessor)
+
+    def test_instrument_with_privacy_mode(self, mock_client):
+        """Test instrument() respects privacy_mode."""
+        with patch("agents.tracing.add_trace_processor"):
+            processor = instrument(
+                client=mock_client,
+                privacy_mode=True,
+            )
+
+            assert processor._privacy_mode is True
+
+    def test_instrument_with_groups_and_properties(self, mock_client):
+        """Test instrument() accepts groups and properties."""
+        with patch("agents.tracing.add_trace_processor"):
+            processor = instrument(
+                client=mock_client,
+                groups={"company": "acme"},
+                properties={"env": "test"},
+            )
+
+            assert processor._groups == {"company": "acme"}
+            assert processor._properties == {"env": "test"}

posthog/test/ai/test_prompts.py

@@ -0,0 +1,577 @@
+import unittest
+from unittest.mock import MagicMock, patch
+
+from posthog.ai.prompts import Prompts
+
+
+class MockResponse:
+    """Mock HTTP response for testing."""
+
+    def __init__(self, json_data=None, status_code=200, ok=True):
+        self._json_data = json_data
+        self.status_code = status_code
+        self.ok = ok
+
+    def json(self):
+        if self._json_data is None:
+            raise ValueError("No JSON data")
+        return self._json_data
+
+
+class TestPrompts(unittest.TestCase):
+    """Tests for the Prompts class."""
+
+    mock_prompt_response = {
+        "id": 1,
+        "name": "test-prompt",
+        "prompt": "Hello, {{name}}! You are a helpful assistant for {{company}}.",
+        "version": 1,
+        "created_by": "user@example.com",
+        "created_at": "2024-01-01T00:00:00Z",
+        "updated_at": "2024-01-01T00:00:00Z",
+        "deleted": False,
+    }
+
+    def create_mock_posthog(
+        self, personal_api_key="phx_test_key", host="https://us.posthog.com"
+    ):
+        """Create a mock PostHog client."""
+        mock = MagicMock()
+        mock.personal_api_key = personal_api_key
+        mock.raw_host = host
+        return mock
+
+
+class TestPromptsGet(TestPrompts):
+    """Tests for the Prompts.get() method."""
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_successfully_fetch_a_prompt(self, mock_get_session):
+        """Should successfully fetch a prompt."""
+        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)
+
+        result = prompts.get("test-prompt")
+
+        self.assertEqual(result, self.mock_prompt_response["prompt"])
+        mock_get.assert_called_once()
+        call_args = mock_get.call_args
+        self.assertEqual(
+            call_args[0][0],
+            "https://us.posthog.com/api/environments/@current/llm_prompts/name/test-prompt/",
+        )
+        self.assertIn("Authorization", call_args[1]["headers"])
+        self.assertEqual(
+            call_args[1]["headers"]["Authorization"], "Bearer phx_test_key"
+        )
+
+    @patch("posthog.ai.prompts._get_session")
+    @patch("posthog.ai.prompts.time.time")
+    def test_return_cached_prompt_when_fresh(self, mock_time, mock_get_session):
+        """Should return cached prompt when fresh (no API call)."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.return_value = MockResponse(json_data=self.mock_prompt_response)
+        mock_time.return_value = 1000.0
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        # First call - fetches from API
+        result1 = prompts.get("test-prompt", cache_ttl_seconds=300)
+        self.assertEqual(result1, self.mock_prompt_response["prompt"])
+        self.assertEqual(mock_get.call_count, 1)
+
+        # Advance time by 60 seconds (still within TTL)
+        mock_time.return_value = 1060.0
+
+        # Second call - should use cache
+        result2 = prompts.get("test-prompt", cache_ttl_seconds=300)
+        self.assertEqual(result2, self.mock_prompt_response["prompt"])
+        self.assertEqual(mock_get.call_count, 1)  # No additional fetch
+
+    @patch("posthog.ai.prompts._get_session")
+    @patch("posthog.ai.prompts.time.time")
+    def test_refetch_when_cache_is_stale(self, mock_time, mock_get_session):
+        """Should refetch when cache is stale."""
+        mock_get = mock_get_session.return_value.get
+        updated_prompt_response = {
+            **self.mock_prompt_response,
+            "prompt": "Updated prompt: Hello, {{name}}!",
+        }
+
+        mock_get.side_effect = [
+            MockResponse(json_data=self.mock_prompt_response),
+            MockResponse(json_data=updated_prompt_response),
+        ]
+        mock_time.return_value = 1000.0
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        # First call - fetches from API
+        result1 = prompts.get("test-prompt", cache_ttl_seconds=60)
+        self.assertEqual(result1, self.mock_prompt_response["prompt"])
+        self.assertEqual(mock_get.call_count, 1)
+
+        # Advance time past TTL
+        mock_time.return_value = 1061.0
+
+        # Second call - should refetch
+        result2 = prompts.get("test-prompt", cache_ttl_seconds=60)
+        self.assertEqual(result2, updated_prompt_response["prompt"])
+        self.assertEqual(mock_get.call_count, 2)
+
+    @patch("posthog.ai.prompts._get_session")
+    @patch("posthog.ai.prompts.time.time")
+    @patch("posthog.ai.prompts.log")
+    def test_use_stale_cache_on_fetch_failure_with_warning(
+        self, mock_log, mock_time, mock_get_session
+    ):
+        """Should use stale cache on fetch failure with warning."""
+        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)
+
+        # First call - populates cache
+        result1 = prompts.get("test-prompt", cache_ttl_seconds=60)
+        self.assertEqual(result1, self.mock_prompt_response["prompt"])
+
+        # Advance time past TTL
+        mock_time.return_value = 1061.0
+
+        # Second call - should use stale cache
+        result2 = prompts.get("test-prompt", cache_ttl_seconds=60)
+        self.assertEqual(result2, self.mock_prompt_response["prompt"])
+
+        # Check warning was logged
+        mock_log.warning.assert_called()
+        warning_call = mock_log.warning.call_args
+        self.assertIn("using stale cache", warning_call[0][0])
+
+    @patch("posthog.ai.prompts._get_session")
+    @patch("posthog.ai.prompts.log")
+    def test_use_fallback_when_no_cache_and_fetch_fails_with_warning(
+        self, mock_log, mock_get_session
+    ):
+        """Should use fallback when no cache and fetch fails with warning."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.side_effect = Exception("Network error")
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        fallback = "Default system prompt."
+        result = prompts.get("test-prompt", fallback=fallback)
+
+        self.assertEqual(result, fallback)
+
+        # Check warning was logged
+        mock_log.warning.assert_called()
+        warning_call = mock_log.warning.call_args
+        self.assertIn("using fallback", warning_call[0][0])
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_throw_when_no_cache_no_fallback_and_fetch_fails(self, mock_get_session):
+        """Should throw when no cache, no fallback, and fetch fails."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.side_effect = Exception("Network error")
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        with self.assertRaises(Exception) as context:
+            prompts.get("test-prompt")
+
+        self.assertIn("Network error", str(context.exception))
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_handle_404_response(self, mock_get_session):
+        """Should handle 404 response."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.return_value = MockResponse(status_code=404, ok=False)
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        with self.assertRaises(Exception) as context:
+            prompts.get("nonexistent-prompt")
+
+        self.assertIn('Prompt "nonexistent-prompt" not found', str(context.exception))
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_handle_403_response(self, mock_get_session):
+        """Should handle 403 response."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.return_value = MockResponse(status_code=403, ok=False)
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        with self.assertRaises(Exception) as context:
+            prompts.get("restricted-prompt")
+
+        self.assertIn(
+            'Access denied for prompt "restricted-prompt"', str(context.exception)
+        )
+
+    def test_throw_when_no_personal_api_key_configured(self):
+        """Should throw when no personal_api_key is configured."""
+        posthog = self.create_mock_posthog(personal_api_key=None)
+        prompts = Prompts(posthog)
+
+        with self.assertRaises(Exception) as context:
+            prompts.get("test-prompt")
+
+        self.assertIn(
+            "personal_api_key is required to fetch prompts", str(context.exception)
+        )
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_throw_when_api_returns_invalid_response_format(self, mock_get_session):
+        """Should throw when API returns invalid response format."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.return_value = MockResponse(json_data={"invalid": "response"})
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        with self.assertRaises(Exception) as context:
+            prompts.get("test-prompt")
+
+        self.assertIn("Invalid response format", str(context.exception))
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_use_custom_host_from_posthog_options(self, mock_get_session):
+        """Should use custom host from PostHog options."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.return_value = MockResponse(json_data=self.mock_prompt_response)
+
+        posthog = self.create_mock_posthog(host="https://eu.i.posthog.com")
+        prompts = Prompts(posthog)
+
+        prompts.get("test-prompt")
+
+        call_args = mock_get.call_args
+        self.assertTrue(
+            call_args[0][0].startswith("https://eu.i.posthog.com/"),
+            f"Expected URL to start with 'https://eu.i.posthog.com/', got {call_args[0][0]}",
+        )
+
+    @patch("posthog.ai.prompts._get_session")
+    @patch("posthog.ai.prompts.time.time")
+    def test_use_default_cache_ttl_5_minutes(self, mock_time, mock_get_session):
+        """Should use default cache TTL (5 minutes) when not specified."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.return_value = MockResponse(json_data=self.mock_prompt_response)
+        mock_time.return_value = 1000.0
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        # First call
+        prompts.get("test-prompt")
+        self.assertEqual(mock_get.call_count, 1)
+
+        # Advance time by 4 minutes (within default 5-minute TTL)
+        mock_time.return_value = 1000.0 + (4 * 60)
+
+        # Second call - should use cache
+        prompts.get("test-prompt")
+        self.assertEqual(mock_get.call_count, 1)
+
+        # Advance time past 5-minute TTL
+        mock_time.return_value = 1000.0 + (6 * 60)
+
+        # Third call - should refetch
+        prompts.get("test-prompt")
+        self.assertEqual(mock_get.call_count, 2)
+
+    @patch("posthog.ai.prompts._get_session")
+    @patch("posthog.ai.prompts.time.time")
+    def test_use_custom_default_cache_ttl_from_constructor(
+        self, mock_time, mock_get_session
+    ):
+        """Should use custom default cache TTL from constructor."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.return_value = MockResponse(json_data=self.mock_prompt_response)
+        mock_time.return_value = 1000.0
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog, default_cache_ttl_seconds=60)
+
+        # First call
+        prompts.get("test-prompt")
+        self.assertEqual(mock_get.call_count, 1)
+
+        # Advance time past custom TTL
+        mock_time.return_value = 1061.0
+
+        # Second call - should refetch
+        prompts.get("test-prompt")
+        self.assertEqual(mock_get.call_count, 2)
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_url_encode_prompt_names_with_special_characters(self, mock_get_session):
+        """Should URL-encode prompt names with special characters."""
+        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)
+
+        prompts.get("prompt with spaces/and/slashes")
+
+        call_args = mock_get.call_args
+        self.assertEqual(
+            call_args[0][0],
+            "https://us.posthog.com/api/environments/@current/llm_prompts/name/prompt%20with%20spaces%2Fand%2Fslashes/",
+        )
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_work_with_direct_options_no_posthog_client(self, mock_get_session):
+        """Should work with direct options (no PostHog client)."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.return_value = MockResponse(json_data=self.mock_prompt_response)
+
+        prompts = Prompts(personal_api_key="phx_direct_key")
+
+        result = prompts.get("test-prompt")
+
+        self.assertEqual(result, self.mock_prompt_response["prompt"])
+        call_args = mock_get.call_args
+        self.assertEqual(
+            call_args[0][0],
+            "https://us.posthog.com/api/environments/@current/llm_prompts/name/test-prompt/",
+        )
+        self.assertEqual(
+            call_args[1]["headers"]["Authorization"], "Bearer phx_direct_key"
+        )
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_use_custom_host_from_direct_options(self, mock_get_session):
+        """Should use custom host from direct options."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.return_value = MockResponse(json_data=self.mock_prompt_response)
+
+        prompts = Prompts(
+            personal_api_key="phx_direct_key", host="https://eu.posthog.com"
+        )
+
+        prompts.get("test-prompt")
+
+        call_args = mock_get.call_args
+        self.assertEqual(
+            call_args[0][0],
+            "https://eu.posthog.com/api/environments/@current/llm_prompts/name/test-prompt/",
+        )
+
+    @patch("posthog.ai.prompts._get_session")
+    @patch("posthog.ai.prompts.time.time")
+    def test_use_custom_default_cache_ttl_from_direct_options(
+        self, mock_time, mock_get_session
+    ):
+        """Should use custom default cache TTL from direct options."""
+        mock_get = mock_get_session.return_value.get
+        mock_get.return_value = MockResponse(json_data=self.mock_prompt_response)
+        mock_time.return_value = 1000.0
+
+        prompts = Prompts(
+            personal_api_key="phx_direct_key", default_cache_ttl_seconds=60
+        )
+
+        # First call
+        prompts.get("test-prompt")
+        self.assertEqual(mock_get.call_count, 1)
+
+        # Advance time past custom TTL
+        mock_time.return_value = 1061.0
+
+        # Second call - should refetch
+        prompts.get("test-prompt")
+        self.assertEqual(mock_get.call_count, 2)
+
+
+class TestPromptsCompile(TestPrompts):
+    """Tests for the Prompts.compile() method."""
+
+    def test_replace_a_single_variable(self):
+        """Should replace a single variable."""
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        result = prompts.compile("Hello, {{name}}!", {"name": "World"})
+
+        self.assertEqual(result, "Hello, World!")
+
+    def test_replace_multiple_variables(self):
+        """Should replace multiple variables."""
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        result = prompts.compile(
+            "Hello, {{name}}! Welcome to {{company}}. Your tier is {{tier}}.",
+            {"name": "John", "company": "Acme Corp", "tier": "premium"},
+        )
+
+        self.assertEqual(
+            result, "Hello, John! Welcome to Acme Corp. Your tier is premium."
+        )
+
+    def test_handle_numbers(self):
+        """Should handle numbers."""
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        result = prompts.compile("You have {{count}} items.", {"count": 42})
+
+        self.assertEqual(result, "You have 42 items.")
+
+    def test_handle_booleans(self):
+        """Should handle booleans."""
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        result = prompts.compile("Feature enabled: {{enabled}}", {"enabled": True})
+
+        self.assertEqual(result, "Feature enabled: True")
+
+    def test_leave_unmatched_variables_unchanged(self):
+        """Should leave unmatched variables unchanged."""
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        result = prompts.compile(
+            "Hello, {{name}}! Your {{unknown}} is ready.", {"name": "World"}
+        )
+
+        self.assertEqual(result, "Hello, World! Your {{unknown}} is ready.")
+
+    def test_handle_prompts_with_no_variables(self):
+        """Should handle prompts with no variables."""
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        result = prompts.compile("You are a helpful assistant.", {})
+
+        self.assertEqual(result, "You are a helpful assistant.")
+
+    def test_handle_empty_variables_dict(self):
+        """Should handle empty variables dict."""
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        result = prompts.compile("Hello, {{name}}!", {})
+
+        self.assertEqual(result, "Hello, {{name}}!")
+
+    def test_handle_multiple_occurrences_of_same_variable(self):
+        """Should handle multiple occurrences of the same variable."""
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        result = prompts.compile(
+            "Hello, {{name}}! Goodbye, {{name}}!", {"name": "World"}
+        )
+
+        self.assertEqual(result, "Hello, World! Goodbye, World!")
+
+    def test_work_with_direct_options_initialization(self):
+        """Should work with direct options initialization."""
+        prompts = Prompts(personal_api_key="phx_test_key")
+
+        result = prompts.compile("Hello, {{name}}!", {"name": "World"})
+
+        self.assertEqual(result, "Hello, World!")
+
+    def test_handle_variables_with_hyphens(self):
+        """Should handle variables with hyphens."""
+        prompts = Prompts(personal_api_key="phx_test_key")
+
+        result = prompts.compile("User ID: {{user-id}}", {"user-id": "12345"})
+
+        self.assertEqual(result, "User ID: 12345")
+
+    def test_handle_variables_with_dots(self):
+        """Should handle variables with dots."""
+        prompts = Prompts(personal_api_key="phx_test_key")
+
+        result = prompts.compile("Company: {{company.name}}", {"company.name": "Acme"})
+
+        self.assertEqual(result, "Company: Acme")
+
+
+class TestPromptsClearCache(TestPrompts):
+    """Tests for the Prompts.clear_cache() method."""
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_clear_a_specific_prompt_from_cache(self, mock_get_session):
+        """Should clear a specific prompt from cache."""
+        mock_get = mock_get_session.return_value.get
+        other_prompt_response = {**self.mock_prompt_response, "name": "other-prompt"}
+
+        mock_get.side_effect = [
+            MockResponse(json_data=self.mock_prompt_response),
+            MockResponse(json_data=other_prompt_response),
+            MockResponse(json_data=self.mock_prompt_response),
+        ]
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        # Populate cache with two prompts
+        prompts.get("test-prompt")
+        prompts.get("other-prompt")
+        self.assertEqual(mock_get.call_count, 2)
+
+        # Clear only test-prompt
+        prompts.clear_cache("test-prompt")
+
+        # test-prompt should be refetched
+        prompts.get("test-prompt")
+        self.assertEqual(mock_get.call_count, 3)
+
+        # other-prompt should still be cached
+        prompts.get("other-prompt")
+        self.assertEqual(mock_get.call_count, 3)
+
+    @patch("posthog.ai.prompts._get_session")
+    def test_clear_all_prompts_from_cache(self, mock_get_session):
+        """Should clear all prompts from cache when no name is provided."""
+        mock_get = mock_get_session.return_value.get
+        other_prompt_response = {**self.mock_prompt_response, "name": "other-prompt"}
+
+        mock_get.side_effect = [
+            MockResponse(json_data=self.mock_prompt_response),
+            MockResponse(json_data=other_prompt_response),
+            MockResponse(json_data=self.mock_prompt_response),
+            MockResponse(json_data=other_prompt_response),
+        ]
+
+        posthog = self.create_mock_posthog()
+        prompts = Prompts(posthog)
+
+        # Populate cache with two prompts
+        prompts.get("test-prompt")
+        prompts.get("other-prompt")
+        self.assertEqual(mock_get.call_count, 2)
+
+        # Clear all cache
+        prompts.clear_cache()
+
+        # Both prompts should be refetched
+        prompts.get("test-prompt")
+        prompts.get("other-prompt")
+        self.assertEqual(mock_get.call_count, 4)
+
+
+if __name__ == "__main__":
+    unittest.main()

posthog/test/test_client.py

@@ -648,6 +648,7 @@ class TestClient(unittest.TestCase):
                 person_properties={},
                 group_properties={},
                 geoip_disable=True,
+                device_id=None,
             )
 
     @mock.patch("posthog.client.flags")
@@ -712,6 +713,7 @@ class TestClient(unittest.TestCase):
                 person_properties={},
                 group_properties={},
                 geoip_disable=False,
+                device_id=None,
             )
 
     @mock.patch("posthog.client.flags")
@@ -1911,6 +1913,7 @@ class TestClient(unittest.TestCase):
             person_properties={"distinct_id": "some_id"},
             group_properties={},
             geoip_disable=True,
+            device_id=None,
             flag_keys_to_evaluate=["random_key"],
         )
         patch_flags.reset_mock()
@@ -1926,6 +1929,7 @@ class TestClient(unittest.TestCase):
             person_properties={"distinct_id": "feature_enabled_distinct_id"},
             group_properties={},
             geoip_disable=True,
+            device_id=None,
             flag_keys_to_evaluate=["random_key"],
         )
         patch_flags.reset_mock()
@@ -1939,6 +1943,7 @@ class TestClient(unittest.TestCase):
             person_properties={"distinct_id": "all_flags_payloads_id"},
             group_properties={},
             geoip_disable=False,
+            device_id=None,
         )
 
     @mock.patch("posthog.client.Poller")
@@ -1987,6 +1992,7 @@ class TestClient(unittest.TestCase):
                 "instance": {"$group_key": "app.posthog.com"},
             },
             geoip_disable=False,
+            device_id=None,
             flag_keys_to_evaluate=["random_key"],
         )
 
@@ -2014,6 +2020,7 @@ class TestClient(unittest.TestCase):
                 "instance": {"$group_key": "app.posthog.com"},
             },
             geoip_disable=False,
+            device_id=None,
             flag_keys_to_evaluate=["random_key"],
         )
 
@@ -2031,8 +2038,116 @@ class TestClient(unittest.TestCase):
             person_properties={"distinct_id": "some_id"},
             group_properties={},
             geoip_disable=False,
+            device_id=None,
         )
 
+    @parameterized.expand(
+        [
+            # method, method_args, expected_person_props, expected_flag_keys
+            (
+                "get_feature_flag",
+                ["random_key", "some_id"],
+                {"distinct_id": "some_id"},
+                ["random_key"],
+            ),
+            (
+                "feature_enabled",
+                ["random_key", "some_id"],
+                {"distinct_id": "some_id"},
+                ["random_key"],
+            ),
+            (
+                "get_all_flags_and_payloads",
+                ["some_id"],
+                {"distinct_id": "some_id"},
+                None,
+            ),
+            ("get_all_flags", ["some_id"], {"distinct_id": "some_id"}, None),
+            ("get_flags_decision", ["some_id"], {}, None),
+        ]
+    )
+    @mock.patch("posthog.client.flags")
+    def test_device_id_is_passed_to_flags_request(
+        self,
+        method,
+        method_args,
+        expected_person_props,
+        expected_flag_keys,
+        patch_flags,
+    ):
+        """Test that device_id is properly passed to the flags request when provided."""
+        patch_flags.return_value = {"featureFlags": {"beta-feature": "random-variant"}}
+        client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail)
+
+        getattr(client, method)(*method_args, device_id="test-device-123")
+
+        expected_call = {
+            "distinct_id": "some_id",
+            "groups": {},
+            "person_properties": expected_person_props,
+            "group_properties": {},
+            "geoip_disable": True,
+            "device_id": "test-device-123",
+        }
+        if expected_flag_keys:
+            expected_call["flag_keys_to_evaluate"] = expected_flag_keys
+
+        patch_flags.assert_called_with(
+            "random_key", "https://us.i.posthog.com", timeout=3, **expected_call
+        )
+
+    @mock.patch("posthog.client.flags")
+    def test_device_id_from_context_is_used_in_flags_request(self, patch_flags):
+        """Test that device_id from context is used in flags request when not explicitly provided."""
+        from posthog.contexts import new_context, set_context_device_id
+
+        patch_flags.return_value = {
+            "featureFlags": {
+                "beta-feature": "random-variant",
+            }
+        }
+        client = Client(
+            FAKE_TEST_API_KEY,
+            on_error=self.set_fail,
+        )
+
+        # Test that device_id from context is used
+        with new_context():
+            set_context_device_id("context-device-id")
+            client.get_feature_flag("random_key", "some_id")
+            patch_flags.assert_called_with(
+                "random_key",
+                "https://us.i.posthog.com",
+                timeout=3,
+                distinct_id="some_id",
+                groups={},
+                person_properties={"distinct_id": "some_id"},
+                group_properties={},
+                geoip_disable=True,
+                device_id="context-device-id",
+                flag_keys_to_evaluate=["random_key"],
+            )
+
+        # Test that explicit device_id overrides context
+        patch_flags.reset_mock()
+        with new_context():
+            set_context_device_id("context-device-id")
+            client.get_feature_flag(
+                "random_key", "some_id", device_id="explicit-device-id"
+            )
+            patch_flags.assert_called_with(
+                "random_key",
+                "https://us.i.posthog.com",
+                timeout=3,
+                distinct_id="some_id",
+                groups={},
+                person_properties={"distinct_id": "some_id"},
+                group_properties={},
+                geoip_disable=True,
+                device_id="explicit-device-id",
+                flag_keys_to_evaluate=["random_key"],
+            )
+
     @parameterized.expand(
         [
             # name, sys_platform, version_info, expected_runtime, expected_version, expected_os, expected_os_version, platform_method, platform_return, distro_info

posthog/version.py

@@ -1,4 +1,4 @@
-VERSION = "7.5.0"
+VERSION = "7.8.2"
 
 if __name__ == "__main__":
     print(VERSION, end="")  # noqa: T201

pyproject.toml

@@ -84,9 +84,12 @@ packages = [
     "posthog.ai",
     "posthog.ai.langchain",
     "posthog.ai.openai",
+    "posthog.ai.openai_agents",
     "posthog.ai.anthropic",
     "posthog.ai.gemini",
     "posthog.test",
+    "posthog.test.ai",
+    "posthog.test.ai.openai_agents",
     "posthog.integrations",
 ]
 

sdk_compliance_adapter/Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+# Copy the SDK source code
+COPY posthog/ /app/sdk/posthog/
+COPY setup.py pyproject.toml README.md LICENSE /app/sdk/
+
+# Install the SDK from source
+RUN cd /app/sdk && pip install --no-cache-dir -e .
+
+# Install adapter dependencies
+RUN pip install --no-cache-dir flask python-dateutil
+
+# Copy adapter code
+COPY sdk_compliance_adapter/adapter.py /app/adapter.py
+
+# Expose port 8080
+EXPOSE 8080
+
+# Run the adapter
+CMD ["python", "/app/adapter.py"]

sdk_compliance_adapter/README.md

@@ -0,0 +1,76 @@
+# PostHog Python SDK Test Adapter
+
+This adapter wraps the posthog-python SDK for compliance testing with the [PostHog SDK Test Harness](https://github.com/PostHog/posthog-sdk-test-harness).
+
+## What is This?
+
+This is a simple Flask app that:
+1. Wraps the posthog-python SDK
+2. Exposes a REST API for the test harness to control
+3. Tracks internal SDK state for test assertions
+
+## Running Tests
+
+Tests run automatically in CI via GitHub Actions. See the test harness repo for details.
+
+### Locally with Docker Compose
+
+```bash
+# From the posthog-python/sdk_compliance_adapter directory
+docker-compose up --build --abort-on-container-exit
+```
+
+This will:
+1. Build the Python SDK adapter
+2. Pull the test harness image
+3. Run all compliance tests
+4. Show results
+
+### Manually with Docker
+
+```bash
+# Create network
+docker network create test-network
+
+# Build and run adapter
+docker build -f sdk_compliance_adapter/Dockerfile -t posthog-python-adapter .
+docker run -d --name sdk-adapter --network test-network -p 8080:8080 posthog-python-adapter
+
+# Run test harness
+docker run --rm \
+  --name test-harness \
+  --network test-network \
+  ghcr.io/posthog/sdk-test-harness:latest \
+  run --adapter-url http://sdk-adapter:8080 --mock-url http://test-harness:8081
+
+# Cleanup
+docker stop sdk-adapter && docker rm sdk-adapter
+docker network rm test-network
+```
+
+## Adapter Implementation
+
+See [adapter.py](adapter.py) for the implementation.
+
+The adapter implements the standard SDK adapter interface defined in the [test harness CONTRACT](https://github.com/PostHog/posthog-sdk-test-harness/blob/main/CONTRACT.yaml):
+
+- `GET /health` - Return SDK information
+- `POST /init` - Initialize SDK with config
+- `POST /capture` - Capture an event
+- `POST /flush` - Flush pending events
+- `GET /state` - Return internal state
+- `POST /reset` - Reset SDK state
+
+### Key Implementation Details
+
+**Request Tracking**: The adapter monkey-patches `batch_post` to track all HTTP requests made by the SDK, including retries.
+
+**State Management**: Thread-safe state tracking for events captured vs sent, retry attempts, and errors.
+
+**UUID Tracking**: Extracts and tracks UUIDs from batches to verify deduplication.
+
+## Documentation
+
+For complete documentation on the test harness and how to implement adapters, see:
+- [PostHog SDK Test Harness](https://github.com/PostHog/posthog-sdk-test-harness)
+- [Adapter Implementation Guide](https://github.com/PostHog/posthog-sdk-test-harness/blob/main/ADAPTER_GUIDE.md)

sdk_compliance_adapter/adapter.py

@@ -0,0 +1,383 @@
+"""
+PostHog Python SDK Test Adapter
+
+This adapter implements the SDK Test Adapter Interface defined in the PostHog Capture API Contract.
+It wraps the posthog-python SDK and exposes a REST API for the test harness to exercise.
+"""
+
+import logging
+import os
+import threading
+import time
+from typing import Any, Dict, List, Optional
+
+from flask import Flask, jsonify, request
+
+from posthog import Client
+from posthog.request import batch_post as original_batch_post
+from posthog.version import VERSION
+
+# Configure logging
+logging.basicConfig(
+    level=logging.DEBUG, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+app = Flask(__name__)
+
+
+class RequestInfo:
+    """Information about an HTTP request made by the SDK"""
+
+    def __init__(
+        self,
+        timestamp_ms: int,
+        status_code: int,
+        retry_attempt: int,
+        event_count: int,
+        uuid_list: List[str],
+    ):
+        self.timestamp_ms = timestamp_ms
+        self.status_code = status_code
+        self.retry_attempt = retry_attempt
+        self.event_count = event_count
+        self.uuid_list = uuid_list
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "timestamp_ms": self.timestamp_ms,
+            "status_code": self.status_code,
+            "retry_attempt": self.retry_attempt,
+            "event_count": self.event_count,
+            "uuid_list": self.uuid_list,
+        }
+
+
+class SDKState:
+    """Tracks SDK internal state for test assertions"""
+
+    def __init__(self):
+        self.lock = threading.Lock()
+        self.pending_events = 0
+        self.total_events_captured = 0
+        self.total_events_sent = 0
+        self.total_retries = 0
+        self.last_error: Optional[str] = None
+        self.requests_made: List[RequestInfo] = []
+        self.client: Optional[Client] = None
+        self.retry_attempts: Dict[str, int] = {}  # Track retry attempts by batch ID
+
+    def reset(self):
+        """Reset all state"""
+        with self.lock:
+            self.pending_events = 0
+            self.total_events_captured = 0
+            self.total_events_sent = 0
+            self.total_retries = 0
+            self.last_error = None
+            self.requests_made = []
+            self.retry_attempts = {}
+            if self.client:
+                # Flush and shutdown existing client
+                try:
+                    self.client.shutdown()
+                except Exception as e:
+                    logger.warning(f"Error shutting down client: {e}")
+                self.client = None
+
+    def increment_captured(self):
+        """Increment total events captured"""
+        with self.lock:
+            self.total_events_captured += 1
+            self.pending_events += 1
+
+    def record_request(self, status_code: int, batch: List[Dict], batch_id: str):
+        """Record an HTTP request made by the SDK"""
+        with self.lock:
+            # Determine retry attempt for this batch
+            retry_attempt = self.retry_attempts.get(batch_id, 0)
+
+            # Extract UUIDs from batch
+            uuid_list = [event.get("uuid", "") for event in batch]
+
+            request_info = RequestInfo(
+                timestamp_ms=int(time.time() * 1000),
+                status_code=status_code,
+                retry_attempt=retry_attempt,
+                event_count=len(batch),
+                uuid_list=uuid_list,
+            )
+            self.requests_made.append(request_info)
+
+            # Update counters
+            if status_code == 200:
+                # Success - clear pending events
+                self.total_events_sent += len(batch)
+                self.pending_events = max(0, self.pending_events - len(batch))
+                # Remove batch from retry tracking
+                self.retry_attempts.pop(batch_id, None)
+            else:
+                # Failure - increment retry count
+                self.retry_attempts[batch_id] = retry_attempt + 1
+                if retry_attempt > 0:
+                    self.total_retries += 1
+
+    def record_error(self, error: str):
+        """Record an error"""
+        with self.lock:
+            self.last_error = error
+
+    def get_state(self) -> Dict[str, Any]:
+        """Get current state as dict"""
+        with self.lock:
+            return {
+                "pending_events": self.pending_events,
+                "total_events_captured": self.total_events_captured,
+                "total_events_sent": self.total_events_sent,
+                "total_retries": self.total_retries,
+                "last_error": self.last_error,
+                "requests_made": [r.to_dict() for r in self.requests_made],
+            }
+
+
+# Global state
+state = SDKState()
+
+
+def create_batch_id(batch: List[Dict]) -> str:
+    """Create a unique ID for a batch based on UUIDs"""
+    uuids = sorted([event.get("uuid", "") for event in batch])
+    return "-".join(uuids[:3])  # Use first 3 UUIDs as batch ID
+
+
+def patched_batch_post(
+    api_key: str,
+    host: Optional[str] = None,
+    gzip: bool = False,
+    timeout: int = 15,
+    **kwargs,
+):
+    """Patched version of batch_post that tracks requests"""
+    batch = kwargs.get("batch", [])
+    batch_id = create_batch_id(batch)
+
+    try:
+        # Call original batch_post
+        response = original_batch_post(api_key, host, gzip, timeout, **kwargs)
+        # Record successful request
+        state.record_request(200, batch, batch_id)
+        return response
+    except Exception as e:
+        # Record failed request
+        status_code = (
+            getattr(e, "status_code", 500) if hasattr(e, "status_code") else 500
+        )
+        state.record_request(status_code, batch, batch_id)
+        state.record_error(str(e))
+        raise
+
+
+# Monkey-patch the batch_post function
+import posthog.request  # noqa: E402
+
+posthog.request.batch_post = patched_batch_post
+
+# Also patch in consumer module
+import posthog.consumer  # noqa: E402
+
+posthog.consumer.batch_post = patched_batch_post
+
+
+@app.route("/health", methods=["GET"])
+def health():
+    """Health check endpoint"""
+    return jsonify(
+        {
+            "sdk_name": "posthog-python",
+            "sdk_version": VERSION,
+            "adapter_version": "1.0.0",
+        }
+    )
+
+
+@app.route("/init", methods=["POST"])
+def init():
+    """Initialize the SDK client"""
+    try:
+        data = request.json or {}
+
+        # Reset state
+        state.reset()
+
+        # Extract config
+        api_key = data.get("api_key")
+        host = data.get("host")
+        flush_at = data.get("flush_at", 100)
+        flush_interval_ms = data.get("flush_interval_ms", 500)
+        max_retries = data.get("max_retries", 3)
+        enable_compression = data.get("enable_compression", False)
+
+        if not api_key:
+            return jsonify({"error": "api_key is required"}), 400
+        if not host:
+            return jsonify({"error": "host is required"}), 400
+
+        # Convert flush_interval from ms to seconds
+        flush_interval = flush_interval_ms / 1000.0
+
+        # Create client
+        client = Client(
+            project_api_key=api_key,
+            host=host,
+            flush_at=flush_at,
+            flush_interval=flush_interval,
+            gzip=enable_compression,
+            max_retries=max_retries,
+            debug=True,
+        )
+
+        state.client = client
+
+        logger.info(
+            f"Initialized SDK with api_key={api_key[:10]}..., host={host}, "
+            f"flush_at={flush_at}, flush_interval={flush_interval}, "
+            f"max_retries={max_retries}, gzip={enable_compression}"
+        )
+
+        return jsonify({"success": True})
+    except Exception as e:
+        logger.exception("Error initializing SDK")
+        return jsonify({"error": str(e)}), 500
+
+
+@app.route("/capture", methods=["POST"])
+def capture():
+    """Capture a single event"""
+    try:
+        if not state.client:
+            return jsonify({"error": "SDK not initialized"}), 400
+
+        data = request.json or {}
+
+        distinct_id = data.get("distinct_id")
+        event = data.get("event")
+        properties = data.get("properties")
+        timestamp = data.get("timestamp")
+
+        if not distinct_id:
+            return jsonify({"error": "distinct_id is required"}), 400
+        if not event:
+            return jsonify({"error": "event is required"}), 400
+
+        # Capture event
+        kwargs = {"distinct_id": distinct_id, "properties": properties}
+        if timestamp:
+            # Parse ISO8601 timestamp
+            from dateutil.parser import parse  # type: ignore[import-untyped]
+
+            kwargs["timestamp"] = parse(timestamp)
+
+        uuid = state.client.capture(event, **kwargs)
+
+        # Track that we captured an event
+        state.increment_captured()
+
+        logger.info(f"Captured event: {event} for {distinct_id}, uuid={uuid}")
+
+        return jsonify({"success": True, "uuid": uuid})
+    except Exception as e:
+        logger.exception("Error capturing event")
+        state.record_error(str(e))
+        return jsonify({"error": str(e)}), 500
+
+
+@app.route("/identify", methods=["POST"])
+def identify():
+    """Identify a user"""
+    try:
+        if not state.client:
+            return jsonify({"error": "SDK not initialized"}), 400
+
+        data = request.json or {}
+
+        distinct_id = data.get("distinct_id")
+        properties = data.get("properties")
+        properties_set_once = data.get("properties_set_once")
+
+        if not distinct_id:
+            return jsonify({"error": "distinct_id is required"}), 400
+
+        # Use the identify pattern - set + set_once
+        if properties:
+            state.client.set(distinct_id=distinct_id, properties=properties)
+            state.increment_captured()
+
+        if properties_set_once:
+            state.client.set_once(
+                distinct_id=distinct_id, properties=properties_set_once
+            )
+            state.increment_captured()
+
+        logger.info(f"Identified user: {distinct_id}")
+
+        return jsonify({"success": True})
+    except Exception as e:
+        logger.exception("Error identifying user")
+        state.record_error(str(e))
+        return jsonify({"error": str(e)}), 500
+
+
+@app.route("/flush", methods=["POST"])
+def flush():
+    """Force flush all pending events"""
+    try:
+        if not state.client:
+            return jsonify({"error": "SDK not initialized"}), 400
+
+        # Flush and wait
+        state.client.flush()
+
+        # Wait a bit for flush to complete
+        # The flush() method triggers queue.join() which blocks until all items are processed
+        time.sleep(0.5)
+
+        logger.info("Flushed pending events")
+
+        return jsonify({"success": True, "events_flushed": state.total_events_sent})
+    except Exception as e:
+        logger.exception("Error flushing events")
+        state.record_error(str(e))
+        return jsonify({"error": str(e), "errors": [str(e)]}, 500)
+
+
+@app.route("/state", methods=["GET"])
+def get_state():
+    """Get internal SDK state"""
+    try:
+        return jsonify(state.get_state())
+    except Exception as e:
+        logger.exception("Error getting state")
+        return jsonify({"error": str(e)}), 500
+
+
+@app.route("/reset", methods=["POST"])
+def reset():
+    """Reset SDK state"""
+    try:
+        state.reset()
+        logger.info("Reset SDK state")
+        return jsonify({"success": True})
+    except Exception as e:
+        logger.exception("Error resetting state")
+        return jsonify({"error": str(e)}), 500
+
+
+def main():
+    """Main entry point"""
+    port = int(os.environ.get("PORT", 8080))
+    logger.info(f"Starting SDK Test Adapter on port {port}")
+    app.run(host="0.0.0.0", port=port, debug=False)
+
+
+if __name__ == "__main__":
+    main()

sdk_compliance_adapter/docker-compose.yml

@@ -0,0 +1,25 @@
+version: "3.8"
+
+services:
+  # PostHog Python SDK adapter
+  sdk-adapter:
+    build:
+      context: ..
+      dockerfile: sdk_compliance_adapter/Dockerfile
+    ports:
+      - "8080:8080"
+    networks:
+      - test-network
+
+  # Test harness
+  test-harness:
+    image: ghcr.io/posthog/sdk-test-harness:latest
+    command: ["run", "--adapter-url", "http://sdk-adapter:8080", "--mock-url", "http://test-harness:8081"]
+    networks:
+      - test-network
+    depends_on:
+      - sdk-adapter
+
+networks:
+  test-network:
+    driver: bridge

sdk_compliance_adapter/requirements.txt

@@ -0,0 +1,3 @@
+# SDK Test Adapter dependencies
+flask>=3.0.0
+python-dateutil>=2.8.0