diff --git a/posthog/tasks/test/test_usage_report.py b/posthog/tasks/test/test_usage_report.py index 87b93022d051..0338d366137d 100644 --- a/posthog/tasks/test/test_usage_report.py +++ b/posthog/tasks/test/test_usage_report.py @@ -4790,6 +4790,7 @@ def test_counts_attributed_in_period_usage_only(self) -> None: self.assertEqual(usage.seconds, [(self.team.id, 7 * 3600)]) self.assertEqual(usage.cpu_core_seconds, [(self.team.id, 7 * 3600 * 4)]) self.assertEqual(usage.memory_gib_seconds, [(self.team.id, 7 * 3600 * 16)]) + self.assertEqual(usage.sandbox_compute_credits, [(self.team.id, 74)]) def test_has_non_zero_usage_counts_task_sandbox_seconds(self) -> None: import dataclasses @@ -4801,6 +4802,17 @@ def test_has_non_zero_usage_counts_task_sandbox_seconds(self) -> None: self.assertFalse(has_non_zero_usage(UsageReportCounters(**zero))) self.assertTrue(has_non_zero_usage(UsageReportCounters(**{**zero, "task_sandbox_seconds_in_period": 5}))) + def test_has_non_zero_usage_counts_sandbox_compute_credits(self) -> None: + import dataclasses + + from posthog.tasks.usage_report import UsageReportCounters, has_non_zero_usage + + zero = {field.name: 0 for field in dataclasses.fields(UsageReportCounters)} + + self.assertFalse(has_non_zero_usage(UsageReportCounters(**zero))) + report = UsageReportCounters(**{**zero, "sandbox_compute_credits_used_in_period": 1}) + self.assertTrue(has_non_zero_usage(report)) + class TestSendUsage(LicensedTestMixin, ClickhouseDestroyTablesMixin, APIBaseTest): def setUp(self) -> None: diff --git a/posthog/tasks/usage_report.py b/posthog/tasks/usage_report.py index 95cdf69fe286..d976afc55262 100644 --- a/posthog/tasks/usage_report.py +++ b/posthog/tasks/usage_report.py @@ -246,9 +246,9 @@ class UsageReportCounters: # PostHog Desktop Billing Credits (PostHog Desktop product usage — same cost math as ai_credits, scoped to ai_product='posthog_code') posthog_code_credits_used_in_period: int + sandbox_compute_credits_used_in_period: int - # Cloud task sandbox compute, all task origins (raw user-attributed usage from the - # SandboxSession ledger — unpriced until a billing model is decided; pre-warm time excluded) + # Cloud task sandbox compute, all task origins (raw user-attributed usage; pre-warm time excluded) task_sandbox_seconds_in_period: int task_sandbox_cpu_core_seconds_in_period: int task_sandbox_memory_gib_seconds_in_period: int @@ -2486,6 +2486,7 @@ def has_non_zero_usage(report: UsageReportCounters) -> bool: or report.ai_credits_used_in_period > 0 or report.signals_credits_used_in_period > 0 or report.posthog_code_credits_used_in_period > 0 + or report.sandbox_compute_credits_used_in_period > 0 or report.task_sandbox_seconds_in_period > 0 or report.logs_bytes_in_period > 0 or report.apm_tracing_bytes_in_period > 0 @@ -2769,6 +2770,7 @@ def _get_all_usage_data(period_start: datetime, period_end: datetime) -> dict[st "teams_with_posthog_code_credits_used_in_period": get_teams_with_posthog_code_credits_used_in_period( period_start, period_end ), + "teams_with_sandbox_compute_credits_used_in_period": task_sandbox_usage.sandbox_compute_credits, "teams_with_task_sandbox_seconds_in_period": task_sandbox_usage.seconds, "teams_with_task_sandbox_cpu_core_seconds_in_period": task_sandbox_usage.cpu_core_seconds, "teams_with_task_sandbox_memory_gib_seconds_in_period": task_sandbox_usage.memory_gib_seconds, @@ -2962,6 +2964,9 @@ def _get_team_report(all_data: dict[str, Any], team: Team) -> UsageReportCounter ai_credits_used_in_period=all_data["teams_with_ai_credits_used_in_period"].get(team.id, 0), signals_credits_used_in_period=all_data["teams_with_signals_credits_used_in_period"].get(team.id, 0), posthog_code_credits_used_in_period=all_data["teams_with_posthog_code_credits_used_in_period"].get(team.id, 0), + sandbox_compute_credits_used_in_period=all_data["teams_with_sandbox_compute_credits_used_in_period"].get( + team.id, 0 + ), task_sandbox_seconds_in_period=all_data["teams_with_task_sandbox_seconds_in_period"].get(team.id, 0), task_sandbox_cpu_core_seconds_in_period=all_data["teams_with_task_sandbox_cpu_core_seconds_in_period"].get( team.id, 0 diff --git a/posthog/temporal/usage_report/queries.py b/posthog/temporal/usage_report/queries.py index 318fcfeaf744..b1fa617619ce 100644 --- a/posthog/temporal/usage_report/queries.py +++ b/posthog/temporal/usage_report/queries.py @@ -213,6 +213,7 @@ def _task_sandbox_usage(begin: datetime, end: datetime) -> dict[str, list[tuple[ "seconds": usage.seconds, "cpu_core_seconds": usage.cpu_core_seconds, "memory_gib_seconds": usage.memory_gib_seconds, + "sandbox_compute_credits": usage.sandbox_compute_credits, } @@ -486,6 +487,7 @@ class QuerySpec: "seconds": "teams_with_task_sandbox_seconds_in_period", "cpu_core_seconds": "teams_with_task_sandbox_cpu_core_seconds_in_period", "memory_gib_seconds": "teams_with_task_sandbox_memory_gib_seconds_in_period", + "sandbox_compute_credits": "teams_with_sandbox_compute_credits_used_in_period", }, ), # ---- ClickHouse: workflows / messaging ---------------------------------- diff --git a/products/tasks/backend/logic/services/sandbox_usage.py b/products/tasks/backend/logic/services/sandbox_usage.py index 943f670a6689..25a090e7d3f3 100644 --- a/products/tasks/backend/logic/services/sandbox_usage.py +++ b/products/tasks/backend/logic/services/sandbox_usage.py @@ -1,11 +1,10 @@ -"""Raw usage ledger for cloud task sandboxes. +"""Usage ledger and aggregation for cloud task sandboxes. One ``SandboxSession`` row per provisioned sandbox records its resource shape and the boundary timestamps of its lifetime (provisioned / user-attributed / last user -activity / ended). The ledger stores raw usage only — no pricing or credit -conversion — so any billable-window policy can be computed later without a -backfill. Pre-warm time is PostHog's cost: a warm sandbox stays unattributed until -a user claims its run with their first message. +activity / ended). Aggregation preserves raw usage and prices user-created compute +with provisional rates. Pre-warm time is PostHog's cost: a warm sandbox stays +unattributed until a user claims its run with their first message. The write helpers swallow and log every failure: the ledger must never break sandbox provisioning, cleanup, or user-message delivery. @@ -14,7 +13,9 @@ from collections.abc import Callable from dataclasses import dataclass from datetime import datetime, timedelta +from decimal import ROUND_CEILING, Decimal from functools import wraps +from math import ceil from typing import ParamSpec, TypeVar from uuid import UUID @@ -25,10 +26,20 @@ import structlog from products.tasks.backend.logic.services.sandbox import SandboxConfig -from products.tasks.backend.models import ComputeSource, SandboxSession, TaskRun +from products.tasks.backend.models import ComputeSource, SandboxSession, Task, TaskRun logger = structlog.get_logger(__name__) +PROVISIONAL_MODAL_CPU_USD_PER_CORE_SECOND_WITH_MARGIN = Decimal("0.00001572") +PROVISIONAL_MODAL_MEMORY_USD_PER_GIB_SECOND_WITH_MARGIN = Decimal("0.000002664") +CREDITS_PER_USD = Decimal(100) +BILLABLE_DIRECT_ORIGINS = frozenset( + { + Task.OriginProduct.IMAGE_BUILDER, + Task.OriginProduct.AUTOMATION, + } +) + P = ParamSpec("P") R = TypeVar("R") @@ -145,11 +156,12 @@ def record_task_run_user_activity( @dataclass(frozen=True) class SandboxUsageByTeam: - """Raw per-team sandbox usage over a period, as (team_id, amount) rows.""" + """Per-team sandbox usage over a period, as (team_id, amount) rows.""" seconds: list[tuple[int, int]] cpu_core_seconds: list[tuple[int, int]] memory_gib_seconds: list[tuple[int, int]] + sandbox_compute_credits: list[tuple[int, int]] def get_task_sandbox_usage_by_team(begin: datetime, end: datetime) -> SandboxUsageByTeam: @@ -162,8 +174,8 @@ def get_task_sandbox_usage_by_team(begin: datetime, end: datetime) -> SandboxUsa workflows), stamped late, or the session is genuinely live (clamped to now). Open rows whose TTL expired before the period are excluded in the query itself, so missed close stamps can't grow the scan without bound. Resource-second - metrics use the configured limits; burstable request floors are recorded on the - row for future pricing policy but don't affect raw usage. + metrics use configured limits. Compute credits use burstable request floors or + the fixed shape and only include work initiated or configured in the Code app. """ now = timezone.now() # Unscoped: the usage report aggregates across every team in the region. @@ -177,6 +189,7 @@ def get_task_sandbox_usage_by_team(begin: datetime, end: datetime) -> SandboxUsa ) usage: dict[int, list[float]] = {} + compute_cost_usd: dict[int, Decimal] = {} for session in sessions.iterator(): assert session.user_attributed_at is not None start = max(session.user_attributed_at, begin) @@ -189,9 +202,34 @@ def get_task_sandbox_usage_by_team(begin: datetime, end: datetime) -> SandboxUsa team_usage[0] += seconds team_usage[1] += seconds * session.cpu_cores team_usage[2] += seconds * session.memory_gb + is_billable_loop = session.origin_product == Task.OriginProduct.LOOP and session.loop_internal is False + is_desktop_run = ( + session.origin_product == Task.OriginProduct.USER_CREATED + and session.compute_source == ComputeSource.POSTHOG_DESKTOP + ) + if session.origin_product in BILLABLE_DIRECT_ORIGINS or is_desktop_run or is_billable_loop: + billable_seconds = Decimal(ceil(seconds)) + if session.burstable: + assert session.cpu_request_cores is not None + assert session.memory_request_mb is not None + cpu_cores = session.cpu_request_cores if session.burstable else session.cpu_cores + memory_gib = ( + Decimal(str(session.memory_request_mb)) / Decimal(1024) + if session.burstable + else Decimal(str(session.memory_gb)) + ) + session_cost = billable_seconds * ( + Decimal(str(cpu_cores)) * PROVISIONAL_MODAL_CPU_USD_PER_CORE_SECOND_WITH_MARGIN + + memory_gib * PROVISIONAL_MODAL_MEMORY_USD_PER_GIB_SECOND_WITH_MARGIN + ) + compute_cost_usd[session.team_id] = compute_cost_usd.get(session.team_id, Decimal(0)) + session_cost return SandboxUsageByTeam( seconds=[(team_id, round(totals[0])) for team_id, totals in usage.items()], cpu_core_seconds=[(team_id, round(totals[1])) for team_id, totals in usage.items()], memory_gib_seconds=[(team_id, round(totals[2])) for team_id, totals in usage.items()], + sandbox_compute_credits=[ + (team_id, int((cost * CREDITS_PER_USD).to_integral_value(rounding=ROUND_CEILING))) + for team_id, cost in compute_cost_usd.items() + ], ) diff --git a/products/tasks/backend/logic/services/tests/test_sandbox_usage.py b/products/tasks/backend/logic/services/tests/test_sandbox_usage.py index 87f5ef15d3f6..da5cbcd24640 100644 --- a/products/tasks/backend/logic/services/tests/test_sandbox_usage.py +++ b/products/tasks/backend/logic/services/tests/test_sandbox_usage.py @@ -24,13 +24,15 @@ def _config(**overrides) -> SandboxConfig: class SandboxUsageBase(APIBaseTest): - def _run(self, *, state: dict | None = None, compute_source: ComputeSource | None = None) -> TaskRun: - task = Task.objects.create( - team=self.team, - title="t", - description="", - origin_product=Task.OriginProduct.USER_CREATED, - ) + def _run( + self, + *, + state: dict | None = None, + origin_product: Task.OriginProduct = Task.OriginProduct.USER_CREATED, + loop: Loop | None = None, + compute_source: ComputeSource | None = None, + ) -> TaskRun: + task = Task.objects.create(team=self.team, title="t", description="", origin_product=origin_product, loop=loop) return TaskRun.objects.create( task=task, team=self.team, @@ -232,12 +234,20 @@ class TestSandboxUsageAggregation(SandboxUsageBase): BEGIN = datetime(2026, 1, 2, tzinfo=UTC) END = datetime(2026, 1, 3, tzinfo=UTC) - def _session(self, **overrides) -> SandboxSession: - run = self._run() + def _session( + self, + *, + task_origin_product: Task.OriginProduct = Task.OriginProduct.USER_CREATED, + loop: Loop | None = None, + **overrides, + ) -> SandboxSession: + run = self._run(origin_product=task_origin_product, loop=loop) defaults: dict = { "team": self.team, "task_run": run, "origin_product": Task.OriginProduct.USER_CREATED, + "compute_source": ComputeSource.POSTHOG_DESKTOP, + "loop_internal": loop.internal if loop is not None else None, "cpu_cores": 4.0, "memory_gb": 16.0, "ttl_seconds": 6 * 60 * 60, @@ -263,6 +273,94 @@ def test_sums_attributed_window_with_resource_multipliers(self): assert usage.seconds == [(self.team.id, 3600)] assert usage.cpu_core_seconds == [(self.team.id, 3600 * 4)] assert usage.memory_gib_seconds == [(self.team.id, 3600 * 16)] + assert usage.sandbox_compute_credits == [(self.team.id, 11)] + + def test_prices_burstable_request_floors(self): + self._session( + burstable=True, + cpu_request_cores=0.5, + memory_request_mb=1024, + ) + + usage = get_task_sandbox_usage_by_team(self.BEGIN, self.END) + + assert usage.sandbox_compute_credits == [(self.team.id, 4)] + assert isinstance(usage.sandbox_compute_credits[0][1], int) + + def test_compute_credits_include_direct_app_origins(self): + self._session(sandbox_id="sb-user-created") + self._session( + sandbox_id="sb-image-builder", + task_origin_product=Task.OriginProduct.IMAGE_BUILDER, + origin_product=Task.OriginProduct.IMAGE_BUILDER, + ) + self._session( + sandbox_id="sb-automation", + task_origin_product=Task.OriginProduct.AUTOMATION, + origin_product=Task.OriginProduct.AUTOMATION, + ) + + usage = get_task_sandbox_usage_by_team(self.BEGIN, self.END) + + assert usage.sandbox_compute_credits == [(self.team.id, 32)] + + def test_compute_credits_default_deny_other_origins(self): + self._session(sandbox_id="sb-null", origin_product=None) + self._session(sandbox_id="sb-known-other", origin_product=Task.OriginProduct.POSTHOG_AI) + self._session(sandbox_id="sb-unknown", origin_product="future_product") + self._session(sandbox_id="sb-signal", origin_product=Task.OriginProduct.SIGNAL_REPORT) + + usage = get_task_sandbox_usage_by_team(self.BEGIN, self.END) + + assert usage.seconds == [(self.team.id, 4 * 3600)] + assert usage.sandbox_compute_credits == [] + + def test_user_created_compute_requires_desktop_source(self): + self._session(sandbox_id="sb-desktop", compute_source=ComputeSource.POSTHOG_DESKTOP) + self._session(sandbox_id="sb-other", compute_source=None) + + usage = get_task_sandbox_usage_by_team(self.BEGIN, self.END) + + assert usage.sandbox_compute_credits == [(self.team.id, 11)] + + def test_compute_credits_include_only_non_internal_loops(self): + user_loop = Loop.objects.unscoped().create( + team=self.team, name="User loop", instructions="Run", runtime_adapter="claude", internal=False + ) + internal_loop = Loop.objects.unscoped().create( + team=self.team, name="Internal loop", instructions="Run", runtime_adapter="claude", internal=True + ) + self._session( + sandbox_id="sb-user-loop", + task_origin_product=Task.OriginProduct.LOOP, + origin_product=Task.OriginProduct.LOOP, + loop=user_loop, + ) + self._session( + sandbox_id="sb-internal-loop", + task_origin_product=Task.OriginProduct.LOOP, + origin_product=Task.OriginProduct.LOOP, + loop=internal_loop, + ) + user_loop.internal = True + user_loop.save(update_fields=["internal", "updated_at"]) + internal_loop.internal = False + internal_loop.save(update_fields=["internal", "updated_at"]) + + usage = get_task_sandbox_usage_by_team(self.BEGIN, self.END) + + assert usage.sandbox_compute_credits == [(self.team.id, 11)] + + def test_rounds_each_duration_and_final_credits_up(self): + self._session( + cpu_cores=0.01, + memory_gb=0.01, + ended_at=datetime(2026, 1, 2, 1, 0, 0, 1, tzinfo=UTC), + ) + + usage = get_task_sandbox_usage_by_team(self.BEGIN, self.END) + + assert usage.sandbox_compute_credits == [(self.team.id, 1)] def test_apportions_sessions_spanning_period_boundaries(self): # Attributed the previous day, ends mid-period: only the in-period slice counts. diff --git a/products/tasks/backend/models.py b/products/tasks/backend/models.py index 755876dc80f9..2591fb606a93 100644 --- a/products/tasks/backend/models.py +++ b/products/tasks/backend/models.py @@ -2313,7 +2313,7 @@ class EndedReason(models.TextChoices): # Resource shape at creation, already clamped by SandboxConfig. Limits are what the # sandbox may consume — raw usage metrics derive from these; the burstable request - # floors are recorded for future pricing-policy work only (Modal bills max(request, actual)). + # floors determine priced compute for burstable sessions (Modal bills max(request, actual)). cpu_cores = models.FloatField(help_text="CPU core limit") memory_gb = models.FloatField(help_text="Memory limit in GiB") ttl_seconds = models.IntegerField(help_text="Hard TTL after which the provider kills the sandbox")