From 2c15bfb9f2094a543634a1496e8787bac46e8949 Mon Sep 17 00:00:00 2001 From: Richard Solomou Date: Wed, 10 Jun 2026 22:55:00 +0300 Subject: [PATCH 1/2] feat(ai): warn when AI wrapper is pointed at the PostHog AI Gateway An AI wrapper pointed at gateway.us.posthog.com captures each generation twice (once by the wrapper, once by the gateway), doubling events and cost. Warn on every routed call without dropping the event, since the wrapper event carries groups, custom properties, and trace hierarchy the gateway never sees. Ports posthog-js #3793. Generated-By: PostHog Code Task-Id: d4ca2f3d-2579-4155-b6d4-1f072c7e9ace --- .sampo/changesets/warn-ai-gateway.md | 5 +++ posthog/ai/gateway.py | 43 ++++++++++++++++++++++ posthog/ai/langchain/callbacks.py | 3 ++ posthog/ai/utils.py | 5 +++ posthog/test/ai/test_gateway.py | 54 ++++++++++++++++++++++++++++ 5 files changed, 110 insertions(+) create mode 100644 .sampo/changesets/warn-ai-gateway.md create mode 100644 posthog/ai/gateway.py create mode 100644 posthog/test/ai/test_gateway.py diff --git a/.sampo/changesets/warn-ai-gateway.md b/.sampo/changesets/warn-ai-gateway.md new file mode 100644 index 00000000..f9464ff1 --- /dev/null +++ b/.sampo/changesets/warn-ai-gateway.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: minor +--- + +Warn when an AI wrapper's `base_url` points at the PostHog AI Gateway, which would otherwise capture and bill each LLM generation twice (once by the wrapper, once by the gateway). The wrapper only warns and never drops the event. diff --git a/posthog/ai/gateway.py b/posthog/ai/gateway.py new file mode 100644 index 00000000..6aa0d219 --- /dev/null +++ b/posthog/ai/gateway.py @@ -0,0 +1,43 @@ +import logging +from typing import Any +from urllib.parse import urlparse + +log = logging.getLogger("posthog") + +# Hosts that resolve to the PostHog AI Gateway. Host-only (no scheme/path). +# Detection is host-based, so the scheme, port, and route prefix don't matter. +POSTHOG_AI_GATEWAY_HOSTS = [ + "gateway.us.posthog.com", +] + + +def is_posthog_ai_gateway_url(base_url: Any) -> bool: + """Return True if base_url points at a known PostHog AI Gateway host.""" + if not base_url: + return False + try: + host = urlparse(str(base_url)).hostname + except Exception: + return False + return host in POSTHOG_AI_GATEWAY_HOSTS + + +def warn_if_posthog_ai_gateway(base_url: Any) -> None: + """ + Warn when an AI wrapper is pointed at the PostHog AI Gateway. + + The wrapper and the gateway each capture the LLM generation, which would + double-count (and double-bill) the event. We only warn and never drop the + event, because the wrapper event carries data the gateway never sees + (groups, custom properties, trace hierarchy). We warn on every call rather + than once, since a single startup line is easy to miss. + """ + if is_posthog_ai_gateway_url(base_url): + log.warning( + "Your PostHog AI wrapper is pointed at the PostHog AI Gateway (%s). " + "This will capture and bill each LLM generation twice — once by this " + "wrapper and once by the gateway. Point the wrapper at the model " + "provider's API directly, or remove the wrapper and rely on the " + "gateway. See https://posthog.com/docs/ai-observability", + urlparse(str(base_url)).hostname, + ) diff --git a/posthog/ai/langchain/callbacks.py b/posthog/ai/langchain/callbacks.py index 6d8cda51..7045aa70 100644 --- a/posthog/ai/langchain/callbacks.py +++ b/posthog/ai/langchain/callbacks.py @@ -42,6 +42,7 @@ from pydantic import BaseModel from posthog import setup +from posthog.ai.gateway import warn_if_posthog_ai_gateway from posthog.ai.sanitization import sanitize_langchain from posthog.ai.utils import get_model_params, with_privacy_mode from posthog.client import Client @@ -622,6 +623,8 @@ def _capture_generation( "$ai_framework": "langchain", } + warn_if_posthog_ai_gateway(run.base_url) + if isinstance(run.posthog_properties, dict): event_properties.update(run.posthog_properties) diff --git a/posthog/ai/utils.py b/posthog/ai/utils.py index 7d170f6a..127f5d11 100644 --- a/posthog/ai/utils.py +++ b/posthog/ai/utils.py @@ -3,6 +3,7 @@ from typing import Any, Callable, Dict, List, Optional, cast from posthog import get_tags, identify_context, new_context, tag, contexts +from posthog.ai.gateway import warn_if_posthog_ai_gateway from posthog.ai.sanitization import ( sanitize_anthropic, sanitize_gemini, @@ -423,6 +424,7 @@ def call_llm_and_track_usage( tag("$ai_latency", latency) tag("$ai_trace_id", posthog_trace_id) tag("$ai_base_url", str(base_url)) + warn_if_posthog_ai_gateway(base_url) available_tool_calls = extract_available_tool_calls(provider, kwargs) @@ -572,6 +574,7 @@ async def call_llm_and_track_usage_async( tag("$ai_latency", latency) tag("$ai_trace_id", posthog_trace_id) tag("$ai_base_url", str(base_url)) + warn_if_posthog_ai_gateway(base_url) available_tool_calls = extract_available_tool_calls(provider, kwargs) @@ -706,6 +709,8 @@ def capture_streaming_event( **(event_data.get("properties") or {}), } + warn_if_posthog_ai_gateway(event_data["base_url"]) + # Determine token source: SDK-computed vs externally overridden sdk_token_tags = { "$ai_input_tokens": event_data["usage_stats"].get("input_tokens", 0), diff --git a/posthog/test/ai/test_gateway.py b/posthog/test/ai/test_gateway.py new file mode 100644 index 00000000..524b53f5 --- /dev/null +++ b/posthog/test/ai/test_gateway.py @@ -0,0 +1,54 @@ +import logging + +from parameterized import parameterized + +from posthog.ai.gateway import ( + is_posthog_ai_gateway_url, + warn_if_posthog_ai_gateway, +) + + +@parameterized.expand( + [ + ("bare_host", "https://gateway.us.posthog.com", True), + ("with_path", "https://gateway.us.posthog.com/openai/v1", True), + ("with_port", "http://gateway.us.posthog.com:443/anthropic", True), + ("http_scheme", "http://gateway.us.posthog.com", True), + ("uppercase_host", "https://GATEWAY.US.POSTHOG.COM/v1", True), + ("openai", "https://api.openai.com/v1", False), + ("anthropic", "https://api.anthropic.com", False), + ("suffix_attack", "https://gateway.us.posthog.com.evil.com", False), + ("other_posthog_host", "https://us.i.posthog.com", False), + ("none", None, False), + ("empty", "", False), + ("garbage", "not a url", False), + ] +) +def test_is_posthog_ai_gateway_url(name, base_url, expected): + assert is_posthog_ai_gateway_url(base_url) is expected + + +def test_warn_if_posthog_ai_gateway_warns_for_gateway(caplog): + with caplog.at_level(logging.WARNING, logger="posthog"): + warn_if_posthog_ai_gateway("https://gateway.us.posthog.com/openai/v1") + + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warnings) == 1 + assert "PostHog AI Gateway" in warnings[0].getMessage() + + +def test_warn_if_posthog_ai_gateway_warns_on_every_call(caplog): + with caplog.at_level(logging.WARNING, logger="posthog"): + warn_if_posthog_ai_gateway("https://gateway.us.posthog.com") + warn_if_posthog_ai_gateway("https://gateway.us.posthog.com") + + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warnings) == 2 + + +def test_warn_if_posthog_ai_gateway_silent_for_non_gateway(caplog): + with caplog.at_level(logging.WARNING, logger="posthog"): + for base_url in ("https://api.openai.com/v1", None, ""): + warn_if_posthog_ai_gateway(base_url) + + assert [r for r in caplog.records if r.levelno == logging.WARNING] == [] From 1aea7c6b020c177eb644c5064d9ff4527565299c Mon Sep 17 00:00:00 2001 From: Richard Solomou Date: Wed, 10 Jun 2026 23:10:29 +0300 Subject: [PATCH 2/2] fix(ai): port gateway warning faithfully from posthog-js#3793 Address review on #658 by matching gatewayWarning.ts as written: - List all five deployed gateway hosts, not just gateway.us.posthog.com, so EU and ai-gateway hosts are detected too. - Tolerate base URLs without a scheme (e.g. gateway.us.posthog.com/v1), which urlparse otherwise reads as a path with no hostname. - Use a static warning message, dropping the redundant second urlparse. - Detect the gateway on the OTel span path (server.address / url.full) in the processor and exporter, since those spans bypass the funnels. - Consolidate tests with pytest.mark.parametrize. Generated-By: PostHog Code Task-Id: d4ca2f3d-2579-4155-b6d4-1f072c7e9ace --- .../warn-ai-gateway-double-capture.md | 5 + .sampo/changesets/warn-ai-gateway.md | 5 - posthog/ai/gateway.py | 84 ++++++--- posthog/ai/otel/exporter.py | 3 + posthog/ai/otel/processor.py | 2 + posthog/test/ai/test_gateway.py | 160 ++++++++++++++---- 6 files changed, 199 insertions(+), 60 deletions(-) create mode 100644 .sampo/changesets/warn-ai-gateway-double-capture.md delete mode 100644 .sampo/changesets/warn-ai-gateway.md diff --git a/.sampo/changesets/warn-ai-gateway-double-capture.md b/.sampo/changesets/warn-ai-gateway-double-capture.md new file mode 100644 index 00000000..2da26723 --- /dev/null +++ b/.sampo/changesets/warn-ai-gateway-double-capture.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: minor +--- + +Warn when an AI wrapper's `base_url` points at the PostHog AI Gateway. The gateway emits its own `$ai_generation`, so each call would be captured (and billed) twice. The wrapper only warns and never drops the event. Detection covers the wrapper funnels (OpenAI, Anthropic, LangChain) and the OTel span path. diff --git a/.sampo/changesets/warn-ai-gateway.md b/.sampo/changesets/warn-ai-gateway.md deleted file mode 100644 index f9464ff1..00000000 --- a/.sampo/changesets/warn-ai-gateway.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -pypi/posthog: minor ---- - -Warn when an AI wrapper's `base_url` points at the PostHog AI Gateway, which would otherwise capture and bill each LLM generation twice (once by the wrapper, once by the gateway). The wrapper only warns and never drops the event. diff --git a/posthog/ai/gateway.py b/posthog/ai/gateway.py index 6aa0d219..cdda6c68 100644 --- a/posthog/ai/gateway.py +++ b/posthog/ai/gateway.py @@ -1,43 +1,83 @@ +# Warn when a wrapper's base_url points at the PostHog AI Gateway: the gateway +# emits its own $ai_generation, so each call would be captured (and, for billable +# products, billed) twice. We only warn — the wrapper's event carries data the +# gateway never sees (groups, custom properties, trace hierarchy). + import logging -from typing import Any +import re +from typing import Any, Mapping, Optional from urllib.parse import urlparse log = logging.getLogger("posthog") -# Hosts that resolve to the PostHog AI Gateway. Host-only (no scheme/path). -# Detection is host-based, so the scheme, port, and route prefix don't matter. +# Keep in sync with the gateway's deployed hosts (see services/llm-gateway in the +# main repo). gateway.us.posthog.com is live today; the rest are listed ahead of +# any traffic moving to them. POSTHOG_AI_GATEWAY_HOSTS = [ + "gateway.posthog.com", "gateway.us.posthog.com", + "gateway.eu.posthog.com", + "ai-gateway.us.posthog.com", + "ai-gateway.eu.posthog.com", ] +# Swap for the dedicated AI Gateway page once it ships. +_GATEWAY_DOCS_URL = "https://posthog.com/docs/ai-observability" + +_SCHEME_RE = re.compile(r"^[a-z][a-z0-9+.-]*://", re.IGNORECASE) + +# OTel spans don't pass through the capture funnels, so detect the gateway from +# the span's host/URL attributes instead. These follow the GenAI / HTTP semantic +# conventions: `server.address` is a bare host, `url.full` a full URL, both of +# which is_posthog_ai_gateway_url accepts. +_OTEL_GATEWAY_URL_ATTRIBUTES = ("server.address", "url.full") + + +def _extract_host(base_url: str) -> Optional[str]: + try: + # Tolerate bare hosts that omit a scheme, e.g. "gateway.us.posthog.com/v1". + url = base_url if _SCHEME_RE.match(base_url) else f"https://{base_url}" + host = urlparse(url).hostname + return host.lower() if host else None + except Exception: + return None + def is_posthog_ai_gateway_url(base_url: Any) -> bool: """Return True if base_url points at a known PostHog AI Gateway host.""" if not base_url: return False - try: - host = urlparse(str(base_url)).hostname - except Exception: - return False - return host in POSTHOG_AI_GATEWAY_HOSTS + host = _extract_host(str(base_url)) + return host is not None and host in POSTHOG_AI_GATEWAY_HOSTS def warn_if_posthog_ai_gateway(base_url: Any) -> None: """ Warn when an AI wrapper is pointed at the PostHog AI Gateway. - The wrapper and the gateway each capture the LLM generation, which would - double-count (and double-bill) the event. We only warn and never drop the - event, because the wrapper event carries data the gateway never sees - (groups, custom properties, trace hierarchy). We warn on every call rather - than once, since a single startup line is easy to miss. + Warns on every gateway call by design: the misconfiguration is impossible to + miss that way, and a doubled bill is worse than noisy logs. We only warn and + never drop the event, because the wrapper event carries data the gateway + never sees (groups, custom properties, trace hierarchy). """ - if is_posthog_ai_gateway_url(base_url): - log.warning( - "Your PostHog AI wrapper is pointed at the PostHog AI Gateway (%s). " - "This will capture and bill each LLM generation twice — once by this " - "wrapper and once by the gateway. Point the wrapper at the model " - "provider's API directly, or remove the wrapper and rely on the " - "gateway. See https://posthog.com/docs/ai-observability", - urlparse(str(base_url)).hostname, - ) + if not is_posthog_ai_gateway_url(base_url): + return + log.warning( + "[PostHog] The PostHog AI wrapper is pointed at the PostHog AI Gateway. " + "Both capture $ai_generation, so every call is double-counted and " + "double-billed. Use one or the other — see %s.", + _GATEWAY_DOCS_URL, + ) + + +def warn_if_posthog_ai_gateway_otel_attributes( + attributes: Optional[Mapping[str, Any]], +) -> None: + """Warn at most once per span when its host/URL attributes point at the gateway.""" + if not attributes: + return + for key in _OTEL_GATEWAY_URL_ATTRIBUTES: + value = attributes.get(key) + if isinstance(value, str) and is_posthog_ai_gateway_url(value): + warn_if_posthog_ai_gateway(value) + return diff --git a/posthog/ai/otel/exporter.py b/posthog/ai/otel/exporter.py index bd2bd996..02c1c22e 100644 --- a/posthog/ai/otel/exporter.py +++ b/posthog/ai/otel/exporter.py @@ -11,6 +11,7 @@ from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from ..gateway import warn_if_posthog_ai_gateway_otel_attributes from .spans import DEFAULT_HOST, is_ai_span @@ -65,6 +66,8 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: ai_spans = [span for span in spans if is_ai_span(span)] if not ai_spans: return SpanExportResult.SUCCESS + for span in ai_spans: + warn_if_posthog_ai_gateway_otel_attributes(span.attributes) return self._exporter.export(ai_spans) def shutdown(self) -> None: diff --git a/posthog/ai/otel/processor.py b/posthog/ai/otel/processor.py index 1a1f7bb7..4f520591 100644 --- a/posthog/ai/otel/processor.py +++ b/posthog/ai/otel/processor.py @@ -12,6 +12,7 @@ from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from ..gateway import warn_if_posthog_ai_gateway_otel_attributes from .spans import DEFAULT_HOST, is_ai_span @@ -70,6 +71,7 @@ def on_end(self, span: ReadableSpan) -> None: """ if not is_ai_span(span): return + warn_if_posthog_ai_gateway_otel_attributes(span.attributes) self._processor.on_end(span) def shutdown(self) -> None: diff --git a/posthog/test/ai/test_gateway.py b/posthog/test/ai/test_gateway.py index 524b53f5..584517b0 100644 --- a/posthog/test/ai/test_gateway.py +++ b/posthog/test/ai/test_gateway.py @@ -1,54 +1,148 @@ import logging -from parameterized import parameterized +import pytest from posthog.ai.gateway import ( + POSTHOG_AI_GATEWAY_HOSTS, is_posthog_ai_gateway_url, warn_if_posthog_ai_gateway, + warn_if_posthog_ai_gateway_otel_attributes, ) -@parameterized.expand( +@pytest.mark.parametrize("host", POSTHOG_AI_GATEWAY_HOSTS) +def test_detects_every_gateway_host(host): + assert is_posthog_ai_gateway_url(f"https://{host}/v1") is True + + +@pytest.mark.parametrize( + "url", + [ + "https://gateway.us.posthog.com/v1", + "https://gateway.us.posthog.com/signals/v1", + "https://gateway.us.posthog.com/anthropic", + ], +) +def test_detects_live_host_with_route_prefix(url): + assert is_posthog_ai_gateway_url(url) is True + + +@pytest.mark.parametrize( + "label,url", + [ + ("http scheme", "http://gateway.us.posthog.com/v1"), + ("uppercase host", "https://GATEWAY.US.POSTHOG.COM/v1"), + ("missing scheme", "gateway.us.posthog.com/v1"), + ("port", "http://gateway.us.posthog.com:443/anthropic"), + ], +) +def test_matches_gateway_host_variants(label, url): + assert is_posthog_ai_gateway_url(url) is True + + +@pytest.mark.parametrize( + "label,url", + [ + ("ingestion host", "https://us.i.posthog.com"), + ("app host", "https://eu.posthog.com"), + ("openai", "https://api.openai.com/v1"), + ("anthropic", "https://api.anthropic.com"), + ("look-alike domain", "https://gateway.us.posthog.com.evil.example/v1"), + ], +) +def test_does_not_match_non_gateway_url(label, url): + assert is_posthog_ai_gateway_url(url) is False + + +@pytest.mark.parametrize( + "label,value", [ - ("bare_host", "https://gateway.us.posthog.com", True), - ("with_path", "https://gateway.us.posthog.com/openai/v1", True), - ("with_port", "http://gateway.us.posthog.com:443/anthropic", True), - ("http_scheme", "http://gateway.us.posthog.com", True), - ("uppercase_host", "https://GATEWAY.US.POSTHOG.COM/v1", True), - ("openai", "https://api.openai.com/v1", False), - ("anthropic", "https://api.anthropic.com", False), - ("suffix_attack", "https://gateway.us.posthog.com.evil.com", False), - ("other_posthog_host", "https://us.i.posthog.com", False), - ("none", None, False), - ("empty", "", False), - ("garbage", "not a url", False), - ] -) -def test_is_posthog_ai_gateway_url(name, base_url, expected): - assert is_posthog_ai_gateway_url(base_url) is expected - - -def test_warn_if_posthog_ai_gateway_warns_for_gateway(caplog): + ("empty string", ""), + ("none", None), + ("malformed", "::::not a url"), + ], +) +def test_does_not_match_empty_or_malformed(label, value): + assert is_posthog_ai_gateway_url(value) is False + + +def _warnings(caplog): + return [r for r in caplog.records if r.levelno == logging.WARNING] + + +def test_warns_with_double_counting_message(caplog): with caplog.at_level(logging.WARNING, logger="posthog"): - warn_if_posthog_ai_gateway("https://gateway.us.posthog.com/openai/v1") + warn_if_posthog_ai_gateway("https://gateway.us.posthog.com/v1") - warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + warnings = _warnings(caplog) assert len(warnings) == 1 - assert "PostHog AI Gateway" in warnings[0].getMessage() + message = warnings[0].getMessage() + assert "[PostHog]" in message + assert "PostHog AI Gateway" in message + assert "$ai_generation" in message + assert "double-counted and double-billed" in message + assert "https://posthog.com/docs/ai-observability" in message + + +def test_warns_on_every_gateway_call(caplog): + with caplog.at_level(logging.WARNING, logger="posthog"): + for _ in range(5): + warn_if_posthog_ai_gateway("https://gateway.us.posthog.com/v1") + + assert len(_warnings(caplog)) == 5 + + +@pytest.mark.parametrize( + "base_url", + ["https://api.openai.com/v1", None, ""], +) +def test_does_not_warn_for_non_gateway(base_url, caplog): + with caplog.at_level(logging.WARNING, logger="posthog"): + warn_if_posthog_ai_gateway(base_url) + assert _warnings(caplog) == [] -def test_warn_if_posthog_ai_gateway_warns_on_every_call(caplog): + +@pytest.mark.parametrize( + "label,attributes", + [ + ("server.address bare host", {"server.address": "gateway.us.posthog.com"}), + ( + "url.full full URL", + {"url.full": "https://gateway.us.posthog.com/v1/chat/completions"}, + ), + ], +) +def test_otel_attributes_warn_when_gateway_detected(label, attributes, caplog): with caplog.at_level(logging.WARNING, logger="posthog"): - warn_if_posthog_ai_gateway("https://gateway.us.posthog.com") - warn_if_posthog_ai_gateway("https://gateway.us.posthog.com") + warn_if_posthog_ai_gateway_otel_attributes(attributes) - warnings = [r for r in caplog.records if r.levelno == logging.WARNING] - assert len(warnings) == 2 + assert len(_warnings(caplog)) == 1 + assert "PostHog AI Gateway" in _warnings(caplog)[0].getMessage() -def test_warn_if_posthog_ai_gateway_silent_for_non_gateway(caplog): +def test_otel_attributes_warn_at_most_once_per_span(caplog): + with caplog.at_level(logging.WARNING, logger="posthog"): + warn_if_posthog_ai_gateway_otel_attributes( + { + "server.address": "gateway.us.posthog.com", + "url.full": "https://gateway.us.posthog.com/v1", + } + ) + + assert len(_warnings(caplog)) == 1 + + +@pytest.mark.parametrize( + "label,attributes", + [ + ("none", None), + ("empty", {}), + ("non-gateway", {"server.address": "api.openai.com"}), + ], +) +def test_otel_attributes_silent_for_non_gateway(label, attributes, caplog): with caplog.at_level(logging.WARNING, logger="posthog"): - for base_url in ("https://api.openai.com/v1", None, ""): - warn_if_posthog_ai_gateway(base_url) + warn_if_posthog_ai_gateway_otel_attributes(attributes) - assert [r for r in caplog.records if r.levelno == logging.WARNING] == [] + assert _warnings(caplog) == []