chore(aio): upgrade posthoganalytics to 7.29.0 and adopt the AI lane switch

ee/hogai/core/runner.py

@@ -37,7 +37,6 @@
 from posthog.event_usage import report_user_action
 from posthog.models import Team, User
 from posthog.ph_client import get_client
-from posthog.settings.ingestion import DedicatedAIEndpointRollout
 from posthog.sync import database_sync_to_async
 from posthog.utils import get_instance_region
 
@@ -206,7 +205,6 @@ def make_client(region: str):
                     region,
                     flush_at=1,
                     before_send=ai_event_truncator,
-                    dedicated_ai_endpoint_stage=DedicatedAIEndpointRollout.RUNNER,
                 )
 
             # Local deployment or hobby

ee/hogai/core/test/test_base_callback_handlers.py

@@ -6,8 +6,6 @@
 import posthoganalytics
 from posthoganalytics.ai.langchain.callbacks import CallbackHandler
 
-from posthog.settings.ingestion import DedicatedAIEndpointRollout
-
 from products.posthog_ai.backend.models.assistant import Conversation
 
 from ee.hogai.chat_agent.runner import ChatAgentRunner
@@ -73,7 +71,6 @@ def test_callback_handler_cloud_us_region(self, mock_get_client, mock_get_region
             "US",
             flush_at=1,
             before_send=ai_event_truncator,
-            dedicated_ai_endpoint_stage=DedicatedAIEndpointRollout.RUNNER,
         )
 
     @patch("ee.hogai.core.runner.is_cloud")
@@ -107,13 +104,11 @@ def get_client_side_effect(region, **kwargs):
             "EU",
             flush_at=1,
             before_send=ai_event_truncator,
-            dedicated_ai_endpoint_stage=DedicatedAIEndpointRollout.RUNNER,
         )
         mock_get_client.assert_any_call(
             "US",
             flush_at=1,
             before_send=ai_event_truncator,
-            dedicated_ai_endpoint_stage=DedicatedAIEndpointRollout.RUNNER,
         )
 
     @patch("ee.hogai.core.runner.is_cloud")

posthog/apps.py

@@ -9,7 +9,6 @@
 from posthoganalytics.client import Client
 
 from posthog.git import get_git_branch, get_git_commit_short
-from posthog.ph_client import enable_dedicated_ai_endpoint_for_default_client
 from posthog.utils import (
     _build_flag_provider,
     get_available_timezones_with_offsets,
@@ -73,11 +72,13 @@ def ready(self):
             "service": settings.OTEL_SERVICE_NAME,
             "environment": os.getenv("OTEL_SERVICE_ENVIRONMENT"),
         }
+        posthoganalytics._use_ai_lane = True  # ty: ignore[invalid-assignment]
+
         # Config for the SDK's `client.metrics` API. The pinned SDK version predates
         # the metrics API and ignores this attr; once posthoganalytics is bumped to
         # >=7.23 it's picked up by setup(), so metrics get a real service.name
         # instead of 'unknown_service'.
-        posthoganalytics.metrics = {  # type: ignore[attr-defined]
+        posthoganalytics.metrics = {  # ty: ignore[invalid-assignment]
             # Same fallback as the OTel trace resource (otel_instrumentation.py) —
             # metrics and traces from one process must share a service identity.
             "service_name": settings.OTEL_SERVICE_NAME or "posthog-django-default",
@@ -138,10 +139,6 @@ def ready(self):
         if not posthoganalytics.disabled and posthoganalytics.feature_flag_definitions() is None:
             posthoganalytics.load_feature_flags()
 
-        # The feature_flag_definitions() call above constructs the default client, so
-        # the dedicated-AI-endpoint flag can only be applied from this point on.
-        enable_dedicated_ai_endpoint_for_default_client()
-
         from posthog.async_migrations.setup import setup_async_migrations
 
         if settings.SKIP_ASYNC_MIGRATIONS_SETUP:

posthog/llm/completions.py

@@ -1,4 +1,5 @@
 import os
+from functools import cache
 from typing import Any, Optional
 
 from django.conf import settings
@@ -7,9 +8,12 @@
 import posthoganalytics
 from posthoganalytics.ai.openai import OpenAI
 
-openai_client = (
-    OpenAI(posthog_client=posthoganalytics, base_url=settings.OPENAI_BASE_URL) if os.getenv("OPENAI_API_KEY") else None  # type: ignore
-)
+
+@cache
+def _get_openai_client() -> Optional[OpenAI]:
+    if not os.getenv("OPENAI_API_KEY"):
+        return None
+    return OpenAI(posthog_client=posthoganalytics.setup(), base_url=settings.OPENAI_BASE_URL)
 
 
 def hit_openai(
@@ -20,6 +24,7 @@ def hit_openai(
     timeout: float | None = None,
     response_format: dict[str, Any] | None = None,
 ) -> tuple[str, int, int]:
+    openai_client = _get_openai_client()
     if not openai_client:
         raise ValueError("OPENAI_API_KEY environment variable not set")
 

posthog/ph_client.py

@@ -4,13 +4,10 @@
 from typing import Any
 from uuid import UUID
 
-from django.conf import settings
-
 import structlog
 import posthoganalytics
 
 from posthog.cloud_utils import is_cloud
-from posthog.settings.ingestion import DedicatedAIEndpointRollout
 from posthog.utils import get_instance_region
 
 PH_US_API_KEY = "sTMFPsFhdP1Ssg"
@@ -21,36 +18,6 @@
 
 logger = structlog.get_logger(__name__)
 
-_DEDICATED_AI_ENDPOINT_STAGES = (DedicatedAIEndpointRollout.RUNNER, DedicatedAIEndpointRollout.ALL)
-
-
-def _use_dedicated_ai_endpoint(caller_stage: DedicatedAIEndpointRollout) -> bool:
-    rollout = settings.POSTHOG_DEDICATED_AI_ENDPOINT_ROLLOUT
-    if rollout is DedicatedAIEndpointRollout.OFF:
-        return False
-    return _DEDICATED_AI_ENDPOINT_STAGES.index(rollout) >= _DEDICATED_AI_ENDPOINT_STAGES.index(caller_stage)
-
-
-def enable_dedicated_ai_endpoint_for_default_client() -> None:
-    """Route the module-level default client's `$ai_*` events to the dedicated AI
-    endpoint at the `all` rollout stage.
-
-    Deliberate workaround: the SDK's lazy `setup()` doesn't accept
-    `_dedicated_ai_endpoint`, and we want to finish testing the endpoint on our own
-    traffic before rethinking the flag as a public option threaded through the
-    SDK's normal construction paths. Mutating the constructed client is safe: it
-    and its consumers read the flag per batch, and the SDK's post-fork consumer
-    rebuild copies it from the old consumers.
-    """
-    if not _use_dedicated_ai_endpoint(DedicatedAIEndpointRollout.ALL):
-        return
-    client = posthoganalytics.default_client
-    if client is None:
-        return
-    client._dedicated_ai_endpoint = True
-    for consumer in client.consumers or []:
-        consumer.dedicated_ai_endpoint = True
-
 
 def feature_enabled_or_false(
     key: str,
@@ -117,12 +84,7 @@ def capture_ph_event(*args: Any, **kwargs: Any) -> None:
         ph_client.shutdown()
 
 
-def get_client(
-    region: str = "US",
-    *,
-    dedicated_ai_endpoint_stage: DedicatedAIEndpointRollout = DedicatedAIEndpointRollout.ALL,
-    **kwargs: Any,
-):
+def get_client(region: str = "US", **kwargs: Any):
     from posthoganalytics import Posthog
 
     api_key = None
@@ -140,6 +102,6 @@ def get_client(
         api_key,
         host=host,
         super_properties={"region": region},
-        _dedicated_ai_endpoint=_use_dedicated_ai_endpoint(dedicated_ai_endpoint_stage),
+        _use_ai_lane=True,
         **kwargs,
     )

posthog/settings/ingestion.py

@@ -1,13 +1,8 @@
 import os
-from enum import StrEnum
-
-import structlog
 
 from posthog.settings.utils import get_from_env, get_list, get_set
 from posthog.utils import str_to_bool
 
-logger = structlog.get_logger(__name__)
-
 INGESTION_LAG_METRIC_TEAM_IDS = get_list(os.getenv("INGESTION_LAG_METRIC_TEAM_IDS", ""))
 
 # KEEP IN SYNC WITH plugin-server/src/config/config.ts
@@ -80,28 +75,6 @@
 NEW_ANALYTICS_CAPTURE_ENDPOINT = os.getenv("NEW_CAPTURE_ENDPOINT", "/i/v0/e/")
 
 
-# Cumulative rollout of the dedicated AI ingestion pipeline for our own `$ai_*` events: each stage
-# also routes the stages before it. Chart-toggled so we can advance or roll back without a deploy.
-class DedicatedAIEndpointRollout(StrEnum):
-    OFF = "off"
-    RUNNER = "runner"
-    ALL = "all"
-
-
-def _parse_dedicated_ai_rollout(value: str) -> "DedicatedAIEndpointRollout":
-    try:
-        return DedicatedAIEndpointRollout(value.strip().lower())
-    except ValueError:
-        logger.warning("invalid_dedicated_ai_endpoint_rollout", value=value)
-        return DedicatedAIEndpointRollout.OFF
-
-
-POSTHOG_DEDICATED_AI_ENDPOINT_ROLLOUT = get_from_env(
-    "POSTHOG_DEDICATED_AI_ENDPOINT_ROLLOUT",
-    DedicatedAIEndpointRollout.OFF,
-    type_cast=_parse_dedicated_ai_rollout,
-)
-
 CAPTURE_V1_INTERNAL_ENDPOINT = os.getenv("CAPTURE_V1_INTERNAL_ENDPOINT", "/i/v1/analytics/events")
 CAPTURE_V1_INTERNAL_MAX_ATTEMPTS = get_from_env("CAPTURE_V1_INTERNAL_MAX_ATTEMPTS", type_cast=int, default=4)
 CAPTURE_V1_INTERNAL_RETRY_AFTER_CAP_SECONDS = get_from_env(

posthog/test/test_ph_client.py

@@ -1,69 +1,17 @@
-from django.test import SimpleTestCase, override_settings
+from django.test import SimpleTestCase
 
 import posthoganalytics
-from parameterized import parameterized
-from posthoganalytics import Posthog
 
-from posthog.ph_client import enable_dedicated_ai_endpoint_for_default_client, get_client
-from posthog.settings.ingestion import (
-    DedicatedAIEndpointRollout as Rollout,
-    _parse_dedicated_ai_rollout,
-)
+from posthog.ph_client import get_client
 
 
-class TestDedicatedAIEndpointRollout(SimpleTestCase):
-    @parameterized.expand(
-        [
-            (Rollout.OFF, Rollout.RUNNER, False),
-            (Rollout.OFF, Rollout.ALL, False),
-            (Rollout.RUNNER, Rollout.RUNNER, True),
-            (Rollout.RUNNER, Rollout.ALL, False),
-            (Rollout.ALL, Rollout.RUNNER, True),
-            (Rollout.ALL, Rollout.ALL, True),
-        ]
-    )
-    def test_dedicated_ai_endpoint_gated_by_rollout_stage(self, rollout, caller_stage, expected):
-        with override_settings(POSTHOG_DEDICATED_AI_ENDPOINT_ROLLOUT=rollout):
-            client = get_client(
-                "US", dedicated_ai_endpoint_stage=caller_stage, send=False, enable_local_evaluation=False
-            )
-            self.assertEqual(client._dedicated_ai_endpoint, expected)
+class TestAILaneOptIn(SimpleTestCase):
+    def test_get_client_opts_into_ai_lane(self):
+        for region in ("US", "EU"):
+            client = get_client(region, send=False, enable_local_evaluation=False)
+            self.assertTrue(client._use_ai_lane)
 
-    def test_general_callers_only_opt_in_at_full_rollout(self):
-        with override_settings(POSTHOG_DEDICATED_AI_ENDPOINT_ROLLOUT=Rollout.RUNNER):
-            self.assertFalse(get_client("US", send=False, enable_local_evaluation=False)._dedicated_ai_endpoint)
-        with override_settings(POSTHOG_DEDICATED_AI_ENDPOINT_ROLLOUT=Rollout.ALL):
-            self.assertTrue(get_client("US", send=False, enable_local_evaluation=False)._dedicated_ai_endpoint)
-
-    @parameterized.expand(
-        [
-            ("off", Rollout.OFF),
-            ("runner", Rollout.RUNNER),
-            ("all", Rollout.ALL),
-            ("  RUNNER  ", Rollout.RUNNER),
-            ("bogus", Rollout.OFF),
-        ]
-    )
-    def test_parse_rollout_falls_back_to_off_on_invalid(self, value, expected):
-        self.assertEqual(_parse_dedicated_ai_rollout(value), expected)
-
-    @parameterized.expand(
-        [
-            (Rollout.OFF, False),
-            (Rollout.RUNNER, False),
-            (Rollout.ALL, True),
-        ]
-    )
-    def test_default_client_routes_ai_events_only_at_full_rollout(self, rollout, expected):
-        client = Posthog("test-key", send=False, enable_local_evaluation=False)
-        original = posthoganalytics.default_client
-        posthoganalytics.default_client = client  # ty: ignore[invalid-assignment]
-        try:
-            with override_settings(POSTHOG_DEDICATED_AI_ENDPOINT_ROLLOUT=rollout):
-                enable_dedicated_ai_endpoint_for_default_client()
-        finally:
-            posthoganalytics.default_client = original
-        self.assertEqual(client._dedicated_ai_endpoint, expected)
-        self.assertTrue(client.consumers)
-        for consumer in client.consumers:
-            self.assertEqual(consumer.dedicated_ai_endpoint, expected)
+    def test_module_attribute_opts_default_client_into_ai_lane(self):
+        self.assertTrue(posthoganalytics._use_ai_lane)
+        client = posthoganalytics.setup()
+        self.assertTrue(client._use_ai_lane)

products/exports/backend/temporal/subscriptions/llm_change_summary.py

@@ -362,7 +362,7 @@ def _attach_images_to_user_message(
 def _get_openai_client() -> OpenAI:
     if not os.environ.get("OPENAI_API_KEY"):
         raise ValueError("OPENAI_API_KEY environment variable not set")
-    return OpenAI(posthog_client=posthoganalytics, base_url=settings.OPENAI_BASE_URL, max_retries=3)  # type: ignore[arg-type]
+    return OpenAI(posthog_client=posthoganalytics.setup(), base_url=settings.OPENAI_BASE_URL, max_retries=3)
 
 
 def generate_change_summary(

products/exports/backend/temporal/subscriptions/test_llm_change_summary.py

@@ -691,12 +691,8 @@ def test_emits_ai_generation_event_with_billing_properties(self, monkeypatch):
         monkeypatch.setenv("OPENAI_API_KEY", "test-fake-key")
         monkeypatch.setattr("posthog.event_usage.SITE_URL", "https://us.posthog.com")
 
-        captured_calls: list[dict] = []
-
-        def fake_capture(*args, **kwargs):
-            captured_calls.append(kwargs)
-
-        monkeypatch.setattr("posthoganalytics.capture", fake_capture)
+        fake_client = MagicMock()
+        monkeypatch.setattr("posthoganalytics.default_client", fake_client)
 
         usage_details = MagicMock()
         usage_details.cached_tokens = 0
@@ -721,6 +717,9 @@ def fake_capture(*args, **kwargs):
         with patch("openai.resources.chat.completions.Completions.create", return_value=fake_response):
             generate_change_summary(None, current, team=team, delivery_id="abc-123")  # type: ignore[arg-type]
 
+        captured_calls = [
+            c.kwargs for method in (fake_client.capture, fake_client._capture_ai) for c in method.call_args_list
+        ]
         ai_generation_calls = [c for c in captured_calls if c.get("event") == "$ai_generation"]
         assert len(ai_generation_calls) == 1, (
             f"expected exactly one $ai_generation capture, got events: {[c.get('event') for c in captured_calls]}"

products/replay_vision/backend/temporal/vision_actions/synthesis.py

@@ -575,7 +575,7 @@ def _run_synthesis(team: Team, action: VisionAction, lines: list[str]) -> str:
     # the LLM gateway (settings.OPENAI_BASE_URL), so the generation lands in LLM analytics tagged to
     # Replay Vision AND bills the team's AI credits ($ai_billable) — the same budget
     # is_team_over_ai_credit_budget gates on above.
-    client = OpenAI(posthog_client=posthoganalytics, base_url=settings.OPENAI_BASE_URL, max_retries=3)  # type: ignore[arg-type]
+    client = OpenAI(posthog_client=posthoganalytics.setup(), base_url=settings.OPENAI_BASE_URL, max_retries=3)
     distinct_id = replay_vision_distinct_id(team.id)
     response = client.chat.completions.create(  # type: ignore[call-overload]
         model=SYNTHESIS_MODEL,

pyproject.toml

@@ -100,7 +100,7 @@ dependencies = [
     "paramiko~=3.5.0",
     "pillow==12.2.0",
     "protobuf~=5.29.6",
-    "posthoganalytics==7.27.0",
+    "posthoganalytics==7.29.0",
     "polars==1.37.1",
     "psycopg2-binary==2.9.10",
     "psycopg[binary]==3.2.4",