feat(workflows): subtract sandbox sends from the sending tier

docs/internal/workflows-email-sending-tiers.md

@@ -62,6 +62,7 @@ None of them are wired there yet; the charts change is part of turning the rollo
 
 In order: staff suspension or an AWS-paused tenant drops to tier 0; a dirty 7-day window, a workflow auto-pause, or a HIGH tenant reputation impact demotes one tier (at most once per cooldown, anchored on the last rate demotion); a tier above 0 with zero sends for the decay period drops one step per period; promotion needs the dwell served, real use of the tier (at least the use ratio of the daily cap on the minimum number of separate days, counted only after the tier anchor), a clean 30-day window, and a clean tenant reputation.
 Rates only count on a meaningful denominator; below the complaint floor, the absolute complaint backstop still applies, including in windows with no sends.
+Sandbox sender sends never count as sends: the sweep subtracts `email_sandbox_sent` from `email_sent` per day, floored at zero. Their bounces and complaints still count.
 Pinned teams never move automatically.
 
 While the tier is enforced, the sweep notifies teams it moved: a rate demotion sends an in-app notification plus an email to project admins, and an earned promotion sends an in-app notification only.
@@ -117,4 +118,5 @@ A committed state change reaches the workers through the `reload-sandbox-sender-
 Each delivered send emits a `workflows sandbox email sent` event.
 Test sends never record `email_sent`, so they do not count toward the sending tier.
 Live workflow sends still count as `email_sent`, which the sending tier reads, until sandbox sends get their own metric.
+The sending tier subtracts `email_sandbox_sent` from `email_sent` per day, and the team's sending allowance per workflow, both floored at zero. Sandbox bounces and complaints still count.
 Use the switch in production for test sends only until that metric is deployed.

products/workflows/backend/services/email_health.py

@@ -23,6 +23,7 @@
 from products.workflows.backend.models.hog_flow.hog_flow import HogFlow
 from products.workflows.backend.providers.ses import SESProvider
 from products.workflows.backend.services.email_sending_attribution import fold_email_totals_by_flow
+from products.workflows.backend.services.email_sending_tier import SANDBOX_SENT_METRIC, SENT_METRIC, own_sends
 from products.workflows.backend.utils.email_sending_tiers import max_email_sending_tier, resolve_team_email_sending_tier
 
 logger = structlog.get_logger(__name__)
@@ -90,9 +91,9 @@ def team_email_sending_allowance(team_id: int) -> EmailSendingAllowance:
 
 def _team_email_sends_since(team_id: int, after: datetime) -> int:
     totals = fetch_app_metric_totals_by_team_and_source(
-        app_source="hog_flow", name=["email_sent"], after=after, team_ids=[team_id]
+        app_source="hog_flow", name=[SENT_METRIC, SANDBOX_SENT_METRIC], after=after, team_ids=[team_id]
     )
-    return sum(counts.get("email_sent", 0) for counts in totals.get(team_id, {}).values())
+    return sum(own_sends(counts) for counts in totals.get(team_id, {}).values())
 
 
 AWS_TENANT_REPUTATION_CACHE_SECONDS = 5 * 60

products/workflows/backend/services/email_sending_tier.py

@@ -27,6 +27,7 @@
 APP_SOURCE = "hog_flow"
 
 SENT_METRIC = "email_sent"
+SANDBOX_SENT_METRIC = "email_sandbox_sent"
 HARD_BOUNCE_METRIC = "email_bounced_hard"
 # SES complaint events are recorded under this name (see the SES webhook handler in the email
 # worker), so a complaint rate reads `email_blocked`, not a metric called "complaint".
@@ -332,7 +333,7 @@ def _fetch_daily_metrics(
     workload: Workload = Workload.DEFAULT,
 ) -> dict[int, dict[str, dict[str, int]]]:
     auto_pause_metrics = [name for name in settings.WORKFLOWS_EMAIL_TIER_AUTO_PAUSE_METRIC_NAMES if name]
-    metric_names = [SENT_METRIC, HARD_BOUNCE_METRIC, COMPLAINT_METRIC, *auto_pause_metrics]
+    metric_names = [SENT_METRIC, SANDBOX_SENT_METRIC, HARD_BOUNCE_METRIC, COMPLAINT_METRIC, *auto_pause_metrics]
     # Tag the query so it is attributable in query-cost analysis and does not trip the untagged-query
     # guard in local dev. Celery adds only task identity, not a product or feature. This derives tier
     # state in the background, so it is enrichment, not a customer-facing query. One context here
@@ -348,17 +349,22 @@ def _fetch_daily_metrics(
         )
 
 
+def own_sends(counts: dict[str, int]) -> int:
+    return max(0, counts.get(SENT_METRIC, 0) - counts.get(SANDBOX_SENT_METRIC, 0))
+
+
 def _history_from_daily(team_id: int, days: dict[str, dict[str, int]]) -> TeamSendingHistory:
     # The tier decision only reads team totals, so metrics recorded under a batch job id and under
     # its parent workflow id sum the same either way and need no per-source resolution.
     auto_pause_metrics = [name for name in settings.WORKFLOWS_EMAIL_TIER_AUTO_PAUSE_METRIC_NAMES if name]
+    daily_sends = {day: sent for day, counts in days.items() if (sent := own_sends(counts)) > 0}
     return TeamSendingHistory(
         team_id=team_id,
-        sent=sum(counts.get(SENT_METRIC, 0) for counts in days.values()),
+        sent=sum(daily_sends.values()),
         hard_bounced=sum(counts.get(HARD_BOUNCE_METRIC, 0) for counts in days.values()),
         complained=sum(counts.get(COMPLAINT_METRIC, 0) for counts in days.values()),
         auto_paused=any(counts.get(metric, 0) > 0 for counts in days.values() for metric in auto_pause_metrics),
-        daily_sends={day: counts.get(SENT_METRIC, 0) for day, counts in days.items() if counts.get(SENT_METRIC, 0)},
+        daily_sends=daily_sends,
     )
 
 

products/workflows/backend/test/test_email_sending_tier.py

@@ -1,6 +1,7 @@
 from datetime import timedelta
 
-from posthog.test.base import BaseTest
+import time_machine
+from posthog.test.base import BaseTest, ClickhouseTestMixin
 from unittest.mock import patch
 
 from django.test import SimpleTestCase, override_settings
@@ -9,6 +10,7 @@
 from parameterized import parameterized
 
 from posthog.models.activity_logging.activity_log import ActivityLog
+from posthog.test.fixtures import create_app_metric2
 
 from products.workflows.backend.management.commands.backfill_workflows_email_sending_tiers import (
     Command as BackfillCommand,
@@ -20,6 +22,7 @@
     TeamSendingHistory,
     TierDecision,
     apply_tier_decision,
+    build_sending_histories,
     decide_tier,
     highest_qualifying_tier,
     recompute_email_sending_tier_for_team,
@@ -470,6 +473,96 @@ def test_pinned_team_never_moves(self, _name: str, tier: int) -> None:
         assert TeamWorkflowsConfig.objects.get(team=self.team).email_sending_tier == tier
 
 
+@time_machine.travel("2026-10-05T12:00:00Z", tick=False)
+@override_settings(**TIER_SETTINGS)
+class TestSandboxSendsInTierHistory(ClickhouseTestMixin, BaseTest):
+    def _record_metrics(self, *, days_ago: int, **counts: int) -> None:
+        for metric_name, count in counts.items():
+            if count:
+                create_app_metric2(
+                    team_id=self.team.pk,
+                    app_source="hog_flow",
+                    metric_kind="email",
+                    metric_name=metric_name,
+                    count=count,
+                    timestamp=timezone.now() - timedelta(days=days_ago),
+                )
+
+    def _record_sends(self, *, days_ago: int, sent: int, sandbox_sent: int) -> None:
+        self._record_metrics(days_ago=days_ago, email_sent=sent, email_sandbox_sent=sandbox_sent)
+
+    @parameterized.expand(
+        [
+            ("no sends decay", 1, 0, 0, 0, "inactive"),
+            ("only sandbox sends decay like no sends", 1, 600, 600, 0, "inactive"),
+            ("sandbox sends do not lift own sends over the use bar", 0, 600, 300, 0, "tier_not_used_enough"),
+            ("own sends over the use bar still promote", 0, 900, 300, 1, "clean_and_used"),
+        ]
+    )
+    def test_only_own_sends_count_toward_tier_history(
+        self, _name: str, current_tier: int, sent: int, sandbox_sent: int, expected_tier: int, expected_reason: str
+    ) -> None:
+        TeamWorkflowsConfig.objects.update_or_create(
+            team=self.team,
+            defaults={
+                "email_sending_tier": current_tier,
+                "email_sending_tier_updated_at": timezone.now() - timedelta(days=40),
+            },
+        )
+        for days_ago in (1, 2):
+            self._record_sends(days_ago=days_ago, sent=sent, sandbox_sent=sandbox_sent)
+
+        decision = recompute_email_sending_tier_for_team(self.team.id)
+
+        assert decision is not None
+        assert (decision.new_tier, decision.reason) == (expected_tier, expected_reason)
+        assert TeamWorkflowsConfig.objects.get(team=self.team).email_sending_tier == expected_tier
+
+    @parameterized.expand(
+        [
+            (
+                "sandbox excess on one day does not erase own sends of another day",
+                {
+                    1: {"email_sent": 100, "email_sandbox_sent": 300},
+                    2: {"email_sent": 200, "email_sandbox_sent": 50},
+                },
+                {2: 150},
+                0,
+                0,
+            ),
+            (
+                "bounces and complaints of sandbox sends still count",
+                {1: {"email_sent": 50, "email_sandbox_sent": 50, "email_bounced_hard": 2, "email_blocked": 1}},
+                {},
+                2,
+                1,
+            ),
+        ]
+    )
+    def test_sandbox_subtraction_is_per_day_and_keeps_feedback(
+        self,
+        _name: str,
+        metrics_by_days_ago: dict[int, dict[str, int]],
+        expected_daily_sends_by_days_ago: dict[int, int],
+        expected_hard_bounced: int,
+        expected_complained: int,
+    ) -> None:
+        for days_ago, counts in metrics_by_days_ago.items():
+            self._record_metrics(days_ago=days_ago, **counts)
+
+        history = build_sending_histories(after=timezone.now() - timedelta(days=30), team_ids=[self.team.pk])[
+            self.team.pk
+        ]
+
+        expected_daily_sends = {
+            (timezone.now() - timedelta(days=days_ago)).strftime("%Y-%m-%d"): count
+            for days_ago, count in expected_daily_sends_by_days_ago.items()
+        }
+        assert history.daily_sends == expected_daily_sends
+        assert history.sent == sum(expected_daily_sends.values())
+        assert (history.hard_bounced, history.complained) == (expected_hard_bounced, expected_complained)
+
+
 @override_settings(**TIER_SETTINGS)
 class TestBackfillEmailSendingTiers(BaseTest):
     def test_suspended_team_is_not_promoted_despite_qualifying_history(self) -> None:

products/workflows/backend/tests/api/test_email_reputation.py

@@ -86,6 +86,28 @@ def test_reputation_endpoint_returns_empty_shape_when_nothing_was_sent(self):
             "email_sending_suspension_reason": "",
         }
 
+    def test_sending_allowance_leaves_out_sandbox_sends(self):
+        sends_by_source = {
+            self.team.id: {"flow-a": {"email_sent": 1, "email_sandbox_sent": 3}, "flow-b": {"email_sent": 2}}
+        }
+
+        def requested_metrics(*, name: list[str], **_kwargs: object) -> dict[int, dict[str, dict[str, int]]]:
+            return {
+                team_id: {
+                    source: {metric: count for metric, count in counts.items() if metric in name}
+                    for source, counts in sources.items()
+                }
+                for team_id, sources in sends_by_source.items()
+            }
+
+        with patch(
+            "products.workflows.backend.services.email_health.fetch_app_metric_totals_by_team_and_source",
+            side_effect=requested_metrics,
+        ):
+            allowance = self._get_reputation({})["sending_allowance"]
+        assert allowance["emails_sent_last_hour"] == 2
+        assert allowance["emails_sent_last_day"] == 2
+
     @parameterized.expand(
         [
             ("no_findings", "ENABLED", None, "healthy"),