diff --git a/products/ai_observability/backend/api/personal_spend.py b/products/ai_observability/backend/api/personal_spend.py index 6d3d934dff11..49b6417210ec 100644 --- a/products/ai_observability/backend/api/personal_spend.py +++ b/products/ai_observability/backend/api/personal_spend.py @@ -7,7 +7,7 @@ query param; see `SUPPORTED_PRODUCTS` for the currently accepted values. Endpoint: -- GET /api/llm_analytics/@me/spend/?product=&date_from=-30d&date_to=&limit=50&refresh=false +- GET /api/llm_analytics/@me/spend/?product=&date_from=-30d&date_to=&limit=50&refresh=false&bucket_minutes=5 """ from __future__ import annotations @@ -63,6 +63,11 @@ MAX_WINDOW_DAYS = 90 # The most calendar days a MAX_WINDOW_DAYS window can touch: partial days at both edges. BY_DAY_MAX_ROWS = MAX_WINDOW_DAYS + 1 +# Sub-day series are only useful (and cheap) over short windows; a "last 24h" +# view is the intended consumer. The cap bounds the series length regardless of +# bucket size: 600 buckets is 50 hours at 5-minute buckets and 25 days hourly. +BUCKET_MINUTES_CHOICES = [5, 15, 30, 60] +MAX_TIME_BUCKETS = 600 _RELATIVE_DATE_RE = re.compile(r"^-?\d+[hdwmqyHDWMQY](Start|End)?$") MIN_LIMIT = 1 @@ -78,9 +83,13 @@ def _internal_team_id() -> int: return settings.LLM_ANALYTICS_INTERNAL_TEAM_ID -def _cache_key(email: str, date_from: str, date_to: str | None, product: str, limit: int) -> str: +def _cache_key( + email: str, date_from: str, date_to: str | None, product: str, limit: int, bucket_minutes: int | None +) -> str: to_slot = date_to or "_now" - return f"personal_spend:{email}:{date_from}:{to_slot}:{product}:{limit}" + # Suffix only when set, so bucketless requests keep their pre-bucket_minutes cache keys. + bucket_slot = f":{bucket_minutes}" if bucket_minutes else "" + return f"personal_spend:{email}:{date_from}:{to_slot}:{product}:{limit}{bucket_slot}" def _parse_date_param(value: str, field: str, now: datetime.datetime) -> datetime.datetime: @@ -114,6 +123,40 @@ def _resolve_window(date_from: str, date_to: str | None) -> tuple[datetime.datet return from_dt, to_dt +def _resolve_and_validate_window( + date_from: str, date_to: str | None, bucket_minutes: int | None +) -> tuple[datetime.datetime, datetime.datetime]: + """Resolve the window and enforce the bucket cap on the resolved bounds. Shared by + the US compute path and the EU proxy, which must run this before its cache lookup: + with relative dates the same raw cache key can be cached while the resolved window + is under the cap and resolve over it moments later.""" + from_dt, to_dt = _resolve_window(date_from, date_to) + if bucket_minutes is not None: + bucket_seconds = bucket_minutes * 60 + # Count the bucket starts the window can actually produce rows for (unaligned + # edges add partial buckets), so `by_bucket` never exceeds MAX_TIME_BUCKETS rows. + from_bucket = int(from_dt.timestamp() // bucket_seconds) + to_ts = to_dt.timestamp() + last_bucket = int(to_ts // bucket_seconds) + if to_ts % bucket_seconds == 0: + # `date_to` is exclusive (`timestamp < date_to`), so an end aligned exactly + # on a bucket boundary can never produce a row in its own bucket. + last_bucket -= 1 + n_buckets = last_bucket - from_bucket + 1 + if n_buckets > MAX_TIME_BUCKETS: + max_hours = bucket_minutes * MAX_TIME_BUCKETS // 60 + raise exceptions.ValidationError( + { + "bucket_minutes": ( + f"A window this large would span more than {MAX_TIME_BUCKETS} buckets at " + f"{bucket_minutes}-minute resolution; narrow the window to under {max_hours} hours, " + "or pick a larger bucket size." + ) + } + ) + return from_dt, to_dt + + class _SpendQueryParamsSerializer(serializers.Serializer): date_from = serializers.CharField( required=False, @@ -158,6 +201,21 @@ class _SpendQueryParamsSerializer(serializers.Serializer): default=False, help_text="If true, bypass the result cache and re-run the underlying queries against ClickHouse.", ) + # No allow_null and no default: either would mark the generated schema nullable, and + # typed clients would then advertise `bucket_minutes: null`, which serializes to the + # literal string "null" in a GET query and gets rejected. Omitted means "no by_bucket". + bucket_minutes = serializers.ChoiceField( + choices=BUCKET_MINUTES_CHOICES, + required=False, + help_text=( + "When set, additionally return a `by_bucket` breakdown: a time-ascending UTC cost series for " + "the scoped product at this bucket size in minutes, with per-bucket cost split into uncached " + "input / output / cache read / cache creation components plus the matching token sums. " + f"Supported bucket sizes: {', '.join(str(c) for c in BUCKET_MINUTES_CHOICES)}. The window may " + f"span at most {MAX_TIME_BUCKETS} buckets of the chosen size (e.g. 50 hours at 5-minute " + "buckets)." + ), + ) def validate_product(self, value: str) -> str: if value not in SUPPORTED_PRODUCTS: @@ -226,6 +284,55 @@ class _DayBreakdownRowSerializer(serializers.Serializer): cost_usd = serializers.FloatField(help_text="Total cost in USD on this day for the scoped product.") +class _BucketBreakdownRowSerializer(serializers.Serializer): + bucket_start = serializers.DateTimeField( + help_text="UTC start of the time bucket the events fall in (`toStartOfInterval(timestamp, ...)`)." + ) + event_count = serializers.IntegerField( + help_text="Number of $ai_generation + $ai_embedding events in this bucket for the scoped product." + ) + cost_usd = serializers.FloatField( + help_text=( + "Total cost in USD in this bucket (sum of `$ai_total_cost_usd`). Authoritative: the component " + "columns below can sum to less than this when the cost breakdown was unavailable for some " + "events; render any remainder as uncategorized rather than assuming the components reconcile." + ) + ) + input_cost_usd = serializers.FloatField( + help_text=( + "Cost of uncached (full-price) input tokens in USD, derived per event as `$ai_input_cost_usd` " + "minus the cache read/write costs (the stored input cost includes them), clamped at zero. " + "The four component columns are disjoint: they sum to `cost_usd` when the full breakdown is " + "present, so they can be stacked without double counting cache costs." + ) + ) + output_cost_usd = serializers.FloatField(help_text="Cost of output tokens in USD (sum of `$ai_output_cost_usd`).") + cache_read_cost_usd = serializers.FloatField( + help_text="Cost of prompt-cache reads in USD (sum of `$ai_cache_read_cost_usd`)." + ) + cache_creation_cost_usd = serializers.FloatField( + help_text=( + "Cost of prompt-cache writes in USD (sum of `$ai_cache_creation_cost_usd`). A spike here with " + "near-zero cache reads is the signature of a cold session being revived: the full conversation " + "context is re-written to the cache at the cache-write rate instead of being read back cheaply." + ) + ) + input_tokens = serializers.IntegerField( + help_text=( + "Sum of `$ai_input_tokens` in this bucket. Whether cached tokens are included follows the " + "provider's reporting (`$ai_cache_reporting_exclusive`): Anthropic-style events exclude them, " + "OpenAI-style events include them, so don't stack this with the cache token sums." + ) + ) + output_tokens = serializers.IntegerField(help_text="Sum of `$ai_output_tokens` in this bucket.") + cache_read_input_tokens = serializers.IntegerField( + help_text="Sum of `$ai_cache_read_input_tokens` (prompt tokens served from cache) in this bucket." + ) + cache_creation_input_tokens = serializers.IntegerField( + help_text="Sum of `$ai_cache_creation_input_tokens` (prompt tokens written to cache) in this bucket." + ) + + class _TopTraceRowSerializer(serializers.Serializer): trace_id = serializers.CharField( allow_null=True, @@ -310,6 +417,26 @@ class _DayBreakdownSerializer(serializers.Serializer): ) +class _BucketBreakdownSerializer(serializers.Serializer): + items = _BucketBreakdownRowSerializer( + many=True, + help_text=( + "One row per UTC time bucket that has events, ordered by bucket start ascending. Buckets with " + "no events are omitted; zero-fill client-side when rendering a continuous series." + ), + ) + bucket_minutes = serializers.IntegerField( + help_text="Bucket size in minutes the series was computed at; echoes the request `bucket_minutes`." + ) + truncated = serializers.BooleanField( + help_text=( + "Effectively always false: `by_bucket` ignores `limit` because truncating a time series by " + f"cost would be meaningless, and the {MAX_TIME_BUCKETS}-bucket window cap already bounds the " + "series length." + ) + ) + + class _TopTracesSerializer(serializers.Serializer): items = _TopTraceRowSerializer(many=True, help_text="Rows of top traces by cost, ordered by cost descending.") truncated = serializers.BooleanField( @@ -329,6 +456,13 @@ class PersonalSpendAnalysisResponseSerializer(serializers.Serializer): by_day = _DayBreakdownSerializer( help_text="Spend grouped by UTC day, ordered ascending. Scoped to `product`. Not subject to `limit`." ) + by_bucket = _BucketBreakdownSerializer( + required=False, + help_text=( + "Spend grouped by UTC time bucket with per-bucket cost/token components, ordered ascending. " + "Scoped to `product`. Only present when the request set `bucket_minutes`." + ), + ) top_traces = _TopTracesSerializer( help_text=( "Deprecated — always returns `{items: [], truncated: false}`. Trace IDs are opaque strings " @@ -623,6 +757,88 @@ def _fetch_by_day( return _truncate(rows, BY_DAY_MAX_ROWS) +def _fetch_by_bucket( + team: Team, + email: str, + from_dt: datetime.datetime, + to_dt: datetime.datetime, + product: str, + bucket_minutes: int, +) -> dict[str, Any]: + # The stored $ai_input_cost_usd INCLUDES prompt-cache read/write costs: on events + # carrying the cache cost columns, input + output reconciles to total and the cache + # costs sit inside input (verified against production events; the ingestion cost + # pipeline prices cache tokens inside input cost the same way). Uncached input is + # therefore derived per event as input minus cache read/write, clamped at zero so a + # future switch to exclusive reporting degrades to undercounting rather than + # double-subtracting. Events without the cache columns carry no cached tokens, so + # the subtraction is a no-op there. Fallback-priced events carry only + # $ai_total_cost_usd, so the components can undershoot cost_usd; the serializer + # help_text tells clients to render the remainder as uncategorized. + query = parse_select( + """ + SELECT + toStartOfInterval(timestamp, toIntervalMinute({bucket_minutes})) AS bucket_start, + count() AS event_count, + round(sum(toFloat(properties.$ai_total_cost_usd)), 6) AS cost_usd, + round(sum(greatest( + toFloat(properties.$ai_input_cost_usd) + - coalesce(toFloat(properties.$ai_cache_read_cost_usd), 0) + - coalesce(toFloat(properties.$ai_cache_creation_cost_usd), 0), + 0 + )), 6) AS input_cost_usd, + round(sum(toFloat(properties.$ai_output_cost_usd)), 6) AS output_cost_usd, + round(sum(toFloat(properties.$ai_cache_read_cost_usd)), 6) AS cache_read_cost_usd, + round(sum(toFloat(properties.$ai_cache_creation_cost_usd)), 6) AS cache_creation_cost_usd, + sum(toFloat(properties.$ai_input_tokens)) AS input_tokens, + sum(toFloat(properties.$ai_output_tokens)) AS output_tokens, + sum(toFloat(properties.$ai_cache_read_input_tokens)) AS cache_read_input_tokens, + sum(toFloat(properties.$ai_cache_creation_input_tokens)) AS cache_creation_input_tokens + FROM events + WHERE {event_in} + AND {product_filter} + AND {email_filter} + AND {timestamp_filter} + GROUP BY bucket_start + ORDER BY bucket_start ASC + LIMIT {limit} + """ + ) + result = execute_hogql_query( + query=query, + placeholders={ + "bucket_minutes": ast.Constant(value=bucket_minutes), + "event_in": _event_in(["$ai_generation", "$ai_embedding"]), + "product_filter": _product_filter(product), + "email_filter": _email_filter(email), + "timestamp_filter": _timestamp_filter(from_dt, to_dt), + # Not the request `limit`; same reasoning as by_day. +1 is the truncation probe row. + "limit": ast.Constant(value=MAX_TIME_BUCKETS + 1), + }, + team=team, + # Buckets are documented as UTC; pin them like by_day does. + modifiers=HogQLQueryModifiers(convertToProjectTimezone=False), + query_type="PersonalSpendByBucket", + ) + rows = [ + { + "bucket_start": row[0], + "event_count": int(row[1] or 0), + "cost_usd": float(row[2] or 0.0), + "input_cost_usd": float(row[3] or 0.0), + "output_cost_usd": float(row[4] or 0.0), + "cache_read_cost_usd": float(row[5] or 0.0), + "cache_creation_cost_usd": float(row[6] or 0.0), + "input_tokens": int(row[7] or 0), + "output_tokens": int(row[8] or 0), + "cache_read_input_tokens": int(row[9] or 0), + "cache_creation_input_tokens": int(row[10] or 0), + } + for row in (result.results or []) + ] + return {**_truncate(rows, MAX_TIME_BUCKETS), "bucket_minutes": bucket_minutes} + + def _compute_spend_analysis( *, email: str, @@ -631,12 +847,15 @@ def _compute_spend_analysis( product: str, limit: int, refresh: bool, + # Defaulted: callers splat validated_data, which omits the key entirely when + # the request didn't set it (the field has no serializer-level default). + bucket_minutes: int | None = None, ) -> dict[str, Any]: """Cached, email-scoped spend analysis shared by the US viewset and the cross-region receiver. Expects already-validated params.""" - from_dt, to_dt = _resolve_window(date_from, date_to) + from_dt, to_dt = _resolve_and_validate_window(date_from, date_to, bucket_minutes) - cache_key = _cache_key(email, date_from, date_to, product, limit) + cache_key = _cache_key(email, date_from, date_to, product, limit, bucket_minutes) if not refresh: cached = cache.get(cache_key) @@ -669,6 +888,11 @@ def _compute_spend_analysis( "by_tool": by_tool, "by_model": _fetch_by_model(team, email, from_dt, to_dt, product, limit), "by_day": _fetch_by_day(team, email, from_dt, to_dt, product), + **( + {"by_bucket": _fetch_by_bucket(team, email, from_dt, to_dt, product, bucket_minutes)} + if bucket_minutes is not None + else {} + ), # Deprecated — trace IDs are opaque and unactionable in the UI. Returned empty so # existing consumers don't crash while they remove the rendering. Drop the field # entirely once no consumer reads it. @@ -754,7 +978,10 @@ class PersonalSpendViewSet(_PersonalSpendUserViewSet): "query param is required and scopes the tool / model / day / trace breakdowns to a single " f"product; supported values: {', '.join(sorted(SUPPORTED_PRODUCTS))}. `by_product` is " "always returned for cross-product visibility. `by_day` returns a day-ascending spend " - "series for the scoped product. Use `refresh=true` to bypass the 5-minute response cache." + "series for the scoped product. Pass `bucket_minutes` (5, 15, 30, or 60; the window may span " + f"at most {MAX_TIME_BUCKETS} buckets) to additionally get `by_bucket`, a time-ascending " + "series with per-bucket cost split into uncached input / output / cache read / cache creation " + "components. Use `refresh=true` to bypass the 5-minute response cache." ), tags=["AI observability"], ) @@ -769,7 +996,7 @@ def list(self, request: Request) -> Response: _EU_REDIRECT_TARGET = "https://us.posthog.com/api/llm_analytics/@me/spend/" -_EU_REDIRECT_FORWARDED_PARAMS = frozenset({"date_from", "date_to", "product", "limit", "refresh"}) +_EU_REDIRECT_FORWARDED_PARAMS = frozenset({"date_from", "date_to", "product", "limit", "refresh", "bucket_minutes"}) def personal_spend_eu_redirect(request: HttpRequest) -> HttpResponseRedirect: @@ -896,19 +1123,27 @@ def list(self, request: Request) -> HttpResponseBase: email = self._require_email(request) - # Validate in-region so bad requests 400 without paying for the hop. + # Validate in-region so bad requests 400 without paying for the hop. The window + # check must precede the cache lookup: the cache is keyed on the raw date + # strings, so a relative window cached while under the bucket cap could + # otherwise keep serving after it resolves over the cap. params = _SpendQueryParamsSerializer(data=request.query_params) params.is_valid(raise_exception=True) data = params.validated_data + _resolve_and_validate_window(data["date_from"], data["date_to"], data.get("bucket_minutes")) # Same cache as the US compute path, so repeat loads skip the cross-region hop. - cache_key = _cache_key(email, data["date_from"], data["date_to"], data["product"], data["limit"]) + cache_key = _cache_key( + email, data["date_from"], data["date_to"], data["product"], data["limit"], data.get("bucket_minutes") + ) if not data["refresh"]: cached = cache.get(cache_key) if cached is not None: return Response(cached, status=status.HTTP_200_OK) - body = json.dumps({**data, "email": email}).encode("utf-8") + # Omit None-valued params (serializer defaults) so the internal receiver's + # non-nullable fields (e.g. bucket_minutes) accept the payload. + body = json.dumps({**{k: v for k, v in data.items() if v is not None}, "email": email}).encode("utf-8") signature, ts = sign_cross_region_spend_request(body, secret) try: upstream = requests.post( diff --git a/products/ai_observability/backend/api/test/test_personal_spend.py b/products/ai_observability/backend/api/test/test_personal_spend.py index f30fbf760858..6431c915f89a 100644 --- a/products/ai_observability/backend/api/test/test_personal_spend.py +++ b/products/ai_observability/backend/api/test/test_personal_spend.py @@ -138,6 +138,39 @@ def test_date_to_before_date_from_rejected(self) -> None: response = self.client.get(f"{ENDPOINT}?{PRODUCT_QS}&date_from=-7d&date_to=-30d") assert response.status_code == status.HTTP_400_BAD_REQUEST + @parameterized.expand( + [ + ("hourly_within_cap", 60, "-7d", status.HTTP_200_OK), + ("hourly_over_cap", 60, "-30d", status.HTTP_400_BAD_REQUEST), + ("five_min_within_cap", 5, "-1d", status.HTTP_200_OK), + ("five_min_over_cap", 5, "-7d", status.HTTP_400_BAD_REQUEST), + ] + ) + def test_bucket_window_cap(self, _label: str, bucket_minutes: int, date_from: str, expected: int) -> None: + response = self.client.get(f"{ENDPOINT}?{PRODUCT_QS}&date_from={date_from}&bucket_minutes={bucket_minutes}") + assert response.status_code == expected + + def test_unsupported_bucket_size_rejected(self) -> None: + response = self.client.get(f"{ENDPOINT}?{PRODUCT_QS}&bucket_minutes=7") + assert response.status_code == status.HTTP_400_BAD_REQUEST + + @parameterized.expand( + [ + # 25 days is exactly 600 hourly buckets of duration, but the half-hour offset + # touches 601 bucket starts; a duration-only check would let 601 rows through. + ("unaligned_at_cap", "2026-06-01T00:30:00", "2026-06-26T00:30:00", status.HTTP_400_BAD_REQUEST), + ("aligned_under_cap", "2026-06-01T00:00:00", "2026-06-25T23:30:00", status.HTTP_200_OK), + # date_to is exclusive, so an end aligned exactly on a bucket boundary never + # reaches its own bucket: the full advertised 600-bucket window must pass. + ("aligned_at_cap_exclusive_end", "2026-06-01T00:00:00", "2026-06-26T00:00:00", status.HTTP_200_OK), + ] + ) + def test_bucket_cap_counts_partial_edge_buckets( + self, _label: str, date_from: str, date_to: str, expected: int + ) -> None: + response = self.client.get(f"{ENDPOINT}?{PRODUCT_QS}&date_from={date_from}&date_to={date_to}&bucket_minutes=60") + assert response.status_code == expected + def test_product_too_long_rejected(self) -> None: response = self.client.get(f"{ENDPOINT}?product={'x' * 100}") assert response.status_code == status.HTTP_400_BAD_REQUEST @@ -206,6 +239,7 @@ def _create_generation( output_tokens: int = 500, event_name: str = "$ai_generation", timestamp: datetime | None = None, + extra_props: dict | None = None, ) -> None: props: dict = { "$ai_input_tokens": input_tokens, @@ -218,6 +252,8 @@ def _create_generation( props["$ai_total_cost_usd"] = cost if tool is not None: props["$ai_tools_called"] = tool + if extra_props: + props.update(extra_props) kwargs: dict = {} if timestamp is not None: kwargs["timestamp"] = timestamp @@ -460,6 +496,125 @@ def test_by_day_uses_utc_days_regardless_of_team_timezone(self) -> None: response = self.client.get(f"{ENDPOINT}?{PRODUCT_QS}&date_from=2026-06-10&date_to=2026-06-16") assert [r["day"] for r in response.json()["by_day"]["items"]] == ["2026-06-15"] + def test_by_bucket_absent_unless_requested(self) -> None: + response = self.client.get(ENDPOINT_OK) + assert response.status_code == status.HTTP_200_OK + assert "by_bucket" not in response.json() + + def test_by_bucket_groups_cost_components_per_utc_hour(self) -> None: + warm = datetime(2026, 6, 15, 9, 30, tzinfo=UTC) + cold = datetime(2026, 6, 15, 11, 5, tzinfo=UTC) + # As stored on real events, $ai_input_cost_usd INCLUDES the cache read/write + # costs (input + output = total); the endpoint must derive the uncached split. + # Warm turn: most of the prompt served from cache (0.1 uncached inside 0.8). + self._create_generation( + cost=1.0, + trace_id="warm", + timestamp=warm, + input_tokens=1000, + output_tokens=500, + extra_props={ + "$ai_input_cost_usd": 0.8, + "$ai_output_cost_usd": 0.2, + "$ai_cache_read_cost_usd": 0.6, + "$ai_cache_creation_cost_usd": 0.1, + "$ai_cache_read_input_tokens": 400000, + "$ai_cache_creation_input_tokens": 20000, + }, + ) + # Cold-revival turn: the whole context re-written to cache, nothing read back + # (0.2 uncached inside 2.7). + self._create_generation( + cost=3.0, + trace_id="cold", + timestamp=cold, + input_tokens=2000, + output_tokens=800, + extra_props={ + "$ai_input_cost_usd": 2.7, + "$ai_output_cost_usd": 0.3, + "$ai_cache_read_cost_usd": 0.0, + "$ai_cache_creation_cost_usd": 2.5, + "$ai_cache_read_input_tokens": 0, + "$ai_cache_creation_input_tokens": 500000, + }, + ) + # Same hour, another product: must not leak into the scoped series. + self._create_generation(ai_product="background_agents", cost=99.0, timestamp=cold) + flush_persons_and_events() + + response = self.client.get(f"{ENDPOINT}?{PRODUCT_QS}&date_from=2026-06-15&date_to=2026-06-16&bucket_minutes=60") + by_bucket = response.json()["by_bucket"] + assert by_bucket["truncated"] is False + assert by_bucket["bucket_minutes"] == 60 + assert by_bucket["items"] == [ + { + "bucket_start": "2026-06-15T09:00:00Z", + "event_count": 1, + "cost_usd": 1.0, + "input_cost_usd": 0.1, + "output_cost_usd": 0.2, + "cache_read_cost_usd": 0.6, + "cache_creation_cost_usd": 0.1, + "input_tokens": 1000, + "output_tokens": 500, + "cache_read_input_tokens": 400000, + "cache_creation_input_tokens": 20000, + }, + { + "bucket_start": "2026-06-15T11:00:00Z", + "event_count": 1, + "cost_usd": 3.0, + "input_cost_usd": 0.2, + "output_cost_usd": 0.3, + "cache_read_cost_usd": 0.0, + "cache_creation_cost_usd": 2.5, + "input_tokens": 2000, + "output_tokens": 800, + "cache_read_input_tokens": 0, + "cache_creation_input_tokens": 500000, + }, + ] + + def test_by_bucket_five_minute_buckets_split_within_the_hour(self) -> None: + # Two calls 15 minutes apart share an hourly bucket but must split at 5-minute + # resolution — this is what isolates a cold-revival spike from surrounding traffic. + self._create_generation(cost=1.0, trace_id="a", timestamp=datetime(2026, 6, 15, 9, 2, tzinfo=UTC)) + self._create_generation(cost=3.0, trace_id="b", timestamp=datetime(2026, 6, 15, 9, 17, tzinfo=UTC)) + flush_persons_and_events() + + response = self.client.get(f"{ENDPOINT}?{PRODUCT_QS}&date_from=2026-06-15&date_to=2026-06-16&bucket_minutes=5") + by_bucket = response.json()["by_bucket"] + assert by_bucket["bucket_minutes"] == 5 + assert [(r["bucket_start"], r["cost_usd"]) for r in by_bucket["items"]] == [ + ("2026-06-15T09:00:00Z", 1.0), + ("2026-06-15T09:15:00Z", 3.0), + ] + + def test_by_bucket_defaults_components_to_zero_when_breakdown_missing(self) -> None: + # Fallback-priced events carry only $ai_total_cost_usd — components must be 0, not an error. + self._create_generation(cost=1.5, timestamp=datetime(2026, 6, 15, 9, 30, tzinfo=UTC)) + flush_persons_and_events() + + response = self.client.get(f"{ENDPOINT}?{PRODUCT_QS}&date_from=2026-06-15&date_to=2026-06-16&bucket_minutes=60") + items = response.json()["by_bucket"]["items"] + assert len(items) == 1 + assert items[0]["cost_usd"] == 1.5 + assert items[0]["input_cost_usd"] == 0.0 + assert items[0]["cache_creation_cost_usd"] == 0.0 + assert items[0]["cache_read_input_tokens"] == 0 + + def test_cache_key_includes_bucket_minutes(self) -> None: + # Without `bucket_minutes` in the cache key, this second call would be served + # the cached bucketless payload and silently drop `by_bucket`. The window must + # stay under the 600-bucket cap, so pin it to a day rather than the 30d default. + with patch("products.ai_observability.backend.api.personal_spend.execute_hogql_query") as mock_exec: + mock_exec.return_value.results = [] + self.client.get(f"{ENDPOINT}?{PRODUCT_QS}&date_from=-1d") + response = self.client.get(f"{ENDPOINT}?{PRODUCT_QS}&date_from=-1d&bucket_minutes=60") + assert response.status_code == status.HTTP_200_OK + assert "by_bucket" in response.json() + def test_by_day_counts_embeddings_and_costless_events(self) -> None: day_one = datetime(2026, 6, 13, 9, 0, tzinfo=UTC) self._create_generation(cost=1.0, timestamp=day_one) @@ -701,6 +856,17 @@ def test_invalid_params_rejected_without_upstream_call(self) -> None: assert response.status_code == status.HTTP_400_BAD_REQUEST post.assert_not_called() + def test_over_cap_bucket_window_rejected_without_upstream_call(self) -> None: + # The window cap must be enforced EU-side before the cache lookup, not + # delegated to the US receiver. + with override_settings(PERSONAL_SPEND_CROSS_REGION_SECRET=CROSS_REGION_SECRET): + with patch("products.ai_observability.backend.api.personal_spend.requests.post") as post: + response = self._get( + {"product": "posthog_code", "date_from": "-30d", "bucket_minutes": "60"}, user=self.user + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + post.assert_not_called() + def test_relays_upstream_success_and_signs_asserted_email(self) -> None: upstream_payload = {"summary": {"scoped_cost_usd": 1.25}} with override_settings(PERSONAL_SPEND_CROSS_REGION_SECRET=CROSS_REGION_SECRET): diff --git a/products/ai_observability/frontend/generated/api.schemas.ts b/products/ai_observability/frontend/generated/api.schemas.ts index d6a4d3ed4c98..0ff0411109bf 100644 --- a/products/ai_observability/frontend/generated/api.schemas.ts +++ b/products/ai_observability/frontend/generated/api.schemas.ts @@ -105,6 +105,40 @@ export interface _DayBreakdownApi { truncated: boolean } +export interface _BucketBreakdownRowApi { + /** UTC start of the time bucket the events fall in (`toStartOfInterval(timestamp, ...)`). */ + bucket_start: string + /** Number of $ai_generation + $ai_embedding events in this bucket for the scoped product. */ + event_count: number + /** Total cost in USD in this bucket (sum of `$ai_total_cost_usd`). Authoritative: the component columns below can sum to less than this when the cost breakdown was unavailable for some events; render any remainder as uncategorized rather than assuming the components reconcile. */ + cost_usd: number + /** Cost of uncached (full-price) input tokens in USD, derived per event as `$ai_input_cost_usd` minus the cache read/write costs (the stored input cost includes them), clamped at zero. The four component columns are disjoint: they sum to `cost_usd` when the full breakdown is present, so they can be stacked without double counting cache costs. */ + input_cost_usd: number + /** Cost of output tokens in USD (sum of `$ai_output_cost_usd`). */ + output_cost_usd: number + /** Cost of prompt-cache reads in USD (sum of `$ai_cache_read_cost_usd`). */ + cache_read_cost_usd: number + /** Cost of prompt-cache writes in USD (sum of `$ai_cache_creation_cost_usd`). A spike here with near-zero cache reads is the signature of a cold session being revived: the full conversation context is re-written to the cache at the cache-write rate instead of being read back cheaply. */ + cache_creation_cost_usd: number + /** Sum of `$ai_input_tokens` in this bucket. Whether cached tokens are included follows the provider's reporting (`$ai_cache_reporting_exclusive`): Anthropic-style events exclude them, OpenAI-style events include them, so don't stack this with the cache token sums. */ + input_tokens: number + /** Sum of `$ai_output_tokens` in this bucket. */ + output_tokens: number + /** Sum of `$ai_cache_read_input_tokens` (prompt tokens served from cache) in this bucket. */ + cache_read_input_tokens: number + /** Sum of `$ai_cache_creation_input_tokens` (prompt tokens written to cache) in this bucket. */ + cache_creation_input_tokens: number +} + +export interface _BucketBreakdownApi { + /** One row per UTC time bucket that has events, ordered by bucket start ascending. Buckets with no events are omitted; zero-fill client-side when rendering a continuous series. */ + items: _BucketBreakdownRowApi[] + /** Bucket size in minutes the series was computed at; echoes the request `bucket_minutes`. */ + bucket_minutes: number + /** Effectively always false: `by_bucket` ignores `limit` because truncating a time series by cost would be meaningless, and the 600-bucket window cap already bounds the series length. */ + truncated: boolean +} + export interface _TopTraceRowApi { /** * `$ai_trace_id` of the session — opaque string scoped to the originating product. Format is not stable: most are UUIDs but some SDK wrappers emit JSON-shaped strings like `{"device_id":"...","session_id":"..."}`. Callers should treat this as an opaque identifier (URL-encode before linking to a trace view). @@ -143,6 +177,8 @@ export interface PersonalSpendAnalysisResponseApi { by_model: _ModelBreakdownApi /** Spend grouped by UTC day, ordered ascending. Scoped to `product`. Not subject to `limit`. */ by_day: _DayBreakdownApi + /** Spend grouped by UTC time bucket with per-bucket cost/token components, ordered ascending. Scoped to `product`. Only present when the request set `bucket_minutes`. */ + by_bucket?: _BucketBreakdownApi /** Deprecated — always returns `{items: [], truncated: false}`. Trace IDs are opaque strings that aren't actionable in the UI. Kept in the response shape so existing consumers don't crash; remove your rendering of this field and we'll drop it from the response entirely in a follow-up. */ top_traces: _TopTracesApi } @@ -2393,6 +2429,15 @@ export interface TestHogTaggerResponseApi { } export type LlmAnalyticsPersonalSpendListParams = { + /** + * When set, additionally return a `by_bucket` breakdown: a time-ascending UTC cost series for the scoped product at this bucket size in minutes, with per-bucket cost split into uncached input / output / cache read / cache creation components plus the matching token sums. Supported bucket sizes: 5, 15, 30, 60. The window may span at most 600 buckets of the chosen size (e.g. 50 hours at 5-minute buckets). + * + * * `5` - 5 + * * `15` - 15 + * * `30` - 30 + * * `60` - 60 + */ + bucket_minutes?: LlmAnalyticsPersonalSpendListBucketMinutes /** * Start of the spend window. Accepts absolute dates (`2026-04-23`) or relative strings (`-7d`, `-1m`, etc.) — same parser used elsewhere in PostHog. Defaults to `-30d`. The window between `date_from` and `date_to` cannot exceed 90 days. * @minLength 1 @@ -2423,6 +2468,16 @@ export type LlmAnalyticsPersonalSpendListParams = { refresh?: boolean } +export type LlmAnalyticsPersonalSpendListBucketMinutes = + (typeof LlmAnalyticsPersonalSpendListBucketMinutes)[keyof typeof LlmAnalyticsPersonalSpendListBucketMinutes] + +export const LlmAnalyticsPersonalSpendListBucketMinutes = { + Number5: 5, + Number15: 15, + Number30: 30, + Number60: 60, +} as const + export type DatasetItemsListParams = { /** * Filter by dataset ID diff --git a/products/ai_observability/frontend/generated/api.ts b/products/ai_observability/frontend/generated/api.ts index 0fb34a81abf7..c40f7ab54417 100644 --- a/products/ai_observability/frontend/generated/api.ts +++ b/products/ai_observability/frontend/generated/api.ts @@ -140,7 +140,7 @@ export const getLlmAnalyticsPersonalSpendListUrl = (params: LlmAnalyticsPersonal } /** - * Return a structured personal LLM spend analysis for the requesting user. Pass `date_from` / `date_to` (absolute like `2026-04-23` or relative like `-7d`) to bound the window — defaults to the last 30 days, max 90 days. The `product=` query param is required and scopes the tool / model / day / trace breakdowns to a single product; supported values: posthog_code. `by_product` is always returned for cross-product visibility. `by_day` returns a day-ascending spend series for the scoped product. Use `refresh=true` to bypass the 5-minute response cache. + * Return a structured personal LLM spend analysis for the requesting user. Pass `date_from` / `date_to` (absolute like `2026-04-23` or relative like `-7d`) to bound the window — defaults to the last 30 days, max 90 days. The `product=` query param is required and scopes the tool / model / day / trace breakdowns to a single product; supported values: posthog_code. `by_product` is always returned for cross-product visibility. `by_day` returns a day-ascending spend series for the scoped product. Pass `bucket_minutes` (5, 15, 30, or 60; the window may span at most 600 buckets) to additionally get `by_bucket`, a time-ascending series with per-bucket cost split into uncached input / output / cache read / cache creation components. Use `refresh=true` to bypass the 5-minute response cache. */ export const llmAnalyticsPersonalSpendList = async ( params: LlmAnalyticsPersonalSpendListParams, diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts index b3105b0d6d8a..7667190438ff 100644 --- a/services/mcp/src/api/generated.ts +++ b/services/mcp/src/api/generated.ts @@ -45147,6 +45147,40 @@ export namespace Schemas { truncated: boolean; } + export interface _BucketBreakdownRow { + /** UTC start of the time bucket the events fall in (`toStartOfInterval(timestamp, ...)`). */ + bucket_start: string; + /** Number of $ai_generation + $ai_embedding events in this bucket for the scoped product. */ + event_count: number; + /** Total cost in USD in this bucket (sum of `$ai_total_cost_usd`). Authoritative: the component columns below can sum to less than this when the cost breakdown was unavailable for some events; render any remainder as uncategorized rather than assuming the components reconcile. */ + cost_usd: number; + /** Cost of uncached (full-price) input tokens in USD, derived per event as `$ai_input_cost_usd` minus the cache read/write costs (the stored input cost includes them), clamped at zero. The four component columns are disjoint: they sum to `cost_usd` when the full breakdown is present, so they can be stacked without double counting cache costs. */ + input_cost_usd: number; + /** Cost of output tokens in USD (sum of `$ai_output_cost_usd`). */ + output_cost_usd: number; + /** Cost of prompt-cache reads in USD (sum of `$ai_cache_read_cost_usd`). */ + cache_read_cost_usd: number; + /** Cost of prompt-cache writes in USD (sum of `$ai_cache_creation_cost_usd`). A spike here with near-zero cache reads is the signature of a cold session being revived: the full conversation context is re-written to the cache at the cache-write rate instead of being read back cheaply. */ + cache_creation_cost_usd: number; + /** Sum of `$ai_input_tokens` in this bucket. Whether cached tokens are included follows the provider's reporting (`$ai_cache_reporting_exclusive`): Anthropic-style events exclude them, OpenAI-style events include them, so don't stack this with the cache token sums. */ + input_tokens: number; + /** Sum of `$ai_output_tokens` in this bucket. */ + output_tokens: number; + /** Sum of `$ai_cache_read_input_tokens` (prompt tokens served from cache) in this bucket. */ + cache_read_input_tokens: number; + /** Sum of `$ai_cache_creation_input_tokens` (prompt tokens written to cache) in this bucket. */ + cache_creation_input_tokens: number; + } + + export interface _BucketBreakdown { + /** One row per UTC time bucket that has events, ordered by bucket start ascending. Buckets with no events are omitted; zero-fill client-side when rendering a continuous series. */ + items: _BucketBreakdownRow[]; + /** Bucket size in minutes the series was computed at; echoes the request `bucket_minutes`. */ + bucket_minutes: number; + /** Effectively always false: `by_bucket` ignores `limit` because truncating a time series by cost would be meaningless, and the 600-bucket window cap already bounds the series length. */ + truncated: boolean; + } + export interface _TopTraceRow { /** * `$ai_trace_id` of the session — opaque string scoped to the originating product. Format is not stable: most are UUIDs but some SDK wrappers emit JSON-shaped strings like `{"device_id":"...","session_id":"..."}`. Callers should treat this as an opaque identifier (URL-encode before linking to a trace view). @@ -45185,6 +45219,8 @@ export namespace Schemas { by_model: _ModelBreakdown; /** Spend grouped by UTC day, ordered ascending. Scoped to `product`. Not subject to `limit`. */ by_day: _DayBreakdown; + /** Spend grouped by UTC time bucket with per-bucket cost/token components, ordered ascending. Scoped to `product`. Only present when the request set `bucket_minutes`. */ + by_bucket?: _BucketBreakdown; /** Deprecated — always returns `{items: [], truncated: false}`. Trace IDs are opaque strings that aren't actionable in the UI. Kept in the response shape so existing consumers don't crash; remove your rendering of this field and we'll drop it from the response entirely in a follow-up. */ top_traces: _TopTraces; } @@ -63754,6 +63790,15 @@ export namespace Schemas { }; export type LlmAnalyticsPersonalSpendListParams = { + /** + * When set, additionally return a `by_bucket` breakdown: a time-ascending UTC cost series for the scoped product at this bucket size in minutes, with per-bucket cost split into uncached input / output / cache read / cache creation components plus the matching token sums. Supported bucket sizes: 5, 15, 30, 60. The window may span at most 600 buckets of the chosen size (e.g. 50 hours at 5-minute buckets). + * + * * `5` - 5 + * * `15` - 15 + * * `30` - 30 + * * `60` - 60 + */ + bucket_minutes?: LlmAnalyticsPersonalSpendListBucketMinutes; /** * Start of the spend window. Accepts absolute dates (`2026-04-23`) or relative strings (`-7d`, `-1m`, etc.) — same parser used elsewhere in PostHog. Defaults to `-30d`. The window between `date_from` and `date_to` cannot exceed 90 days. * @minLength 1 @@ -63784,6 +63829,16 @@ export namespace Schemas { refresh?: boolean; }; + export type LlmAnalyticsPersonalSpendListBucketMinutes = typeof LlmAnalyticsPersonalSpendListBucketMinutes[keyof typeof LlmAnalyticsPersonalSpendListBucketMinutes]; + + + export const LlmAnalyticsPersonalSpendListBucketMinutes = { + Number5: 5, + Number15: 15, + Number30: 30, + Number60: 60, + } as const; + export type ListParams = { /** * Number of results to return per page. diff --git a/services/mcp/src/generated/ai_observability/api.ts b/services/mcp/src/generated/ai_observability/api.ts index 54386c70ff4f..295f821869f8 100644 --- a/services/mcp/src/generated/ai_observability/api.ts +++ b/services/mcp/src/generated/ai_observability/api.ts @@ -9,7 +9,7 @@ import * as zod from 'zod' /** - * Return a structured personal LLM spend analysis for the requesting user. Pass `date_from` / `date_to` (absolute like `2026-04-23` or relative like `-7d`) to bound the window — defaults to the last 30 days, max 90 days. The `product=` query param is required and scopes the tool / model / day / trace breakdowns to a single product; supported values: posthog_code. `by_product` is always returned for cross-product visibility. `by_day` returns a day-ascending spend series for the scoped product. Use `refresh=true` to bypass the 5-minute response cache. + * Return a structured personal LLM spend analysis for the requesting user. Pass `date_from` / `date_to` (absolute like `2026-04-23` or relative like `-7d`) to bound the window — defaults to the last 30 days, max 90 days. The `product=` query param is required and scopes the tool / model / day / trace breakdowns to a single product; supported values: posthog_code. `by_product` is always returned for cross-product visibility. `by_day` returns a day-ascending spend series for the scoped product. Pass `bucket_minutes` (5, 15, 30, or 60; the window may span at most 600 buckets) to additionally get `by_bucket`, a time-ascending series with per-bucket cost split into uncached input / output / cache read / cache creation components. Use `refresh=true` to bypass the 5-minute response cache. */ export const llmAnalyticsPersonalSpendListQueryDateFromDefault = `-30d` export const llmAnalyticsPersonalSpendListQueryDateFromMax = 32 @@ -24,6 +24,12 @@ export const llmAnalyticsPersonalSpendListQueryProductMax = 64 export const llmAnalyticsPersonalSpendListQueryRefreshDefault = false export const LlmAnalyticsPersonalSpendListQueryParams = /* @__PURE__ */ zod.object({ + bucket_minutes: zod + .union([zod.literal(5), zod.literal(15), zod.literal(30), zod.literal(60)]) + .optional() + .describe( + 'When set, additionally return a `by_bucket` breakdown: a time-ascending UTC cost series for the scoped product at this bucket size in minutes, with per-bucket cost split into uncached input / output / cache read / cache creation components plus the matching token sums. Supported bucket sizes: 5, 15, 30, 60. The window may span at most 600 buckets of the chosen size (e.g. 50 hours at 5-minute buckets).\n\n* `5` - 5\n* `15` - 15\n* `30` - 30\n* `60` - 60' + ), date_from: zod .string() .min(1) diff --git a/services/mcp/src/tools/generated/ai_observability.ts b/services/mcp/src/tools/generated/ai_observability.ts index 80dfd99c59a4..1a73c32b725a 100644 --- a/services/mcp/src/tools/generated/ai_observability.ts +++ b/services/mcp/src/tools/generated/ai_observability.ts @@ -721,6 +721,7 @@ const llmaPersonalSpend = (): ToolBase