feat(tasks): report priced sandbox compute credits

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:

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

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 ----------------------------------

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()
+        ],
     )

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.

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")