From 10a10d1d81755a3334b76435b0dee11cb29e6d4d Mon Sep 17 00:00:00 2001
From: Harley Alexander <43975092+mayteio@users.noreply.github.com>
Date: Thu, 22 Oct 2026 12:58:13 +0100
Subject: [PATCH 1/9] feat(email): send transactional emails through workflows
 behind a flag

The transactional-emails-via-workflows flag picks the recipients and templates that send through the Workflows transactional email API. Everything else, and any send Workflows rejects, still goes through Customer.io.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Generated-By: PostHog Desktop
Task-Id: 20d9dd00-fcbd-4c83-a5c3-68a47fd9be9a
---
 ee/settings.py             |  14 ++-
 posthog/email.py           | 190 ++++++++++++++++++++++++++++++-------
 posthog/test/test_email.py |  89 +++++++++++++++++
 3 files changed, 259 insertions(+), 34 deletions(-)

diff --git a/ee/settings.py b/ee/settings.py
index 23f9f12f198e..4a91423054f0 100644
--- a/ee/settings.py
+++ b/ee/settings.py
@@ -5,7 +5,7 @@
 import os
 
 from posthog.settings import AUTHENTICATION_BACKENDS, DEBUG, DEMO, MIDDLEWARE, SITE_URL
-from posthog.settings.utils import get_from_env
+from posthog.settings.utils import get_from_env, get_list
 from posthog.utils import str_to_bool
 
 # SSO
@@ -63,6 +63,18 @@
 CUSTOMER_IO_API_KEY = get_from_env("CUSTOMER_IO_API_KEY", "", type_cast=str)
 CUSTOMER_IO_API_URL = get_from_env("CUSTOMER_IO_API_URL", "https://api-eu.customer.io", type_cast=str)
 
+# The Workflows transactional email API that sends PostHog's own emails. Every region sends through the US project.
+WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY = get_from_env("WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY", "", type_cast=str)
+WORKFLOWS_TRANSACTIONAL_EMAIL_API_URL = get_from_env(
+    "WORKFLOWS_TRANSACTIONAL_EMAIL_API_URL", "https://us.posthog.com", type_cast=str
+)
+WORKFLOWS_TRANSACTIONAL_EMAIL_PROJECT_ID = get_from_env("WORKFLOWS_TRANSACTIONAL_EMAIL_PROJECT_ID", 2, type_cast=int)
+# Comma-separated `template_name=message_template_id` pairs. They override WORKFLOWS_TEMPLATE_ID_MAP in
+# posthog/email.py, so a local or staging instance can send with its own copies of the templates.
+WORKFLOWS_TRANSACTIONAL_EMAIL_TEMPLATE_IDS: dict[str, str] = dict(
+    pair.split("=", 1) for pair in get_list(os.getenv("WORKFLOWS_TRANSACTIONAL_EMAIL_TEMPLATE_IDS", "")) if "=" in pair
+)
+
 # Schedule to run column materialization on. Follows crontab syntax.
 # Use empty string to prevent from materializing
 MATERIALIZE_COLUMNS_SCHEDULE_CRON = get_from_env("MATERIALIZE_COLUMNS_SCHEDULE_CRON", "0 5 * * SAT")
diff --git a/posthog/email.py b/posthog/email.py
index 86de9fecdf18..e3906b93c0cd 100644
--- a/posthog/email.py
+++ b/posthog/email.py
@@ -1,6 +1,7 @@
 import re
 import html
 import uuid
+import hashlib
 import smtplib
 import datetime
 import dataclasses
@@ -54,9 +55,9 @@ def inline_css(value: str) -> str:
 def is_http_email_service_available() -> bool:
     """
     Returns whether HTTP email services are available on this instance (i.e. settings are in place).
-    This currently only supports Customer.io.
+    These are Customer.io and the Workflows transactional email API.
     """
-    return bool(settings.CUSTOMER_IO_API_KEY)
+    return bool(settings.CUSTOMER_IO_API_KEY or settings.WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY)
 
 
 def is_smtp_email_service_available() -> bool:
@@ -164,6 +165,24 @@ def is_email_available(with_absolute_urls: bool = False) -> bool:
     "loop_run_summary": "77",
 }
 
+# Message template ids in the US project, for templates that can send through Workflows instead of Customer.io.
+# A template sends through Workflows only for recipients the flag below matches.
+WORKFLOWS_TEMPLATE_ID_MAP: dict[str, str] = {
+    "approval_requested": "01a16311-33e6-0000-c533-f383fd8e70f0",
+    "password_reset": "01a1597a-5d91-0000-99fa-9eeb18d293c5",
+    "proxy_provisioned": "01a16311-b3de-0000-8aec-7c865f3e5432",
+}
+
+# Release conditions match the `email` and `transactional_email_template` person properties passed at evaluation,
+# so one flag picks both the recipients and the templates that move.
+WORKFLOWS_TRANSACTIONAL_EMAIL_FLAG = "transactional-emails-via-workflows"
+
+EMAIL_WORKFLOWS_FALLBACK_COUNTER = Counter(
+    "posthog_email_workflows_fallback_total",
+    "Emails that Workflows refused and Customer.io sent instead, by template.",
+    labelnames=["template_name"],
+)
+
 
 def get_email_team_and_org_context(
     team: Optional["Team"] = None,
@@ -204,24 +223,144 @@ def get_customer_io_template_id(template_name: str) -> str:
     return template_id
 
 
-# Note: this http sender is only configure for customer.io right now and it's set up to send
-# via templates so all the configuration is done in the customer.io - i.e. no subject, body, etc.
+class WorkflowsSendRejected(Exception):
+    """Workflows did not take the email, so another provider can send it without a duplicate."""
+
+
+def get_workflows_template_id(template_name: str, dest: dict[str, str]) -> str | None:
+    """The message template to send this email with through Workflows, or None to send it through Customer.io."""
+    template_id = {**WORKFLOWS_TEMPLATE_ID_MAP, **settings.WORKFLOWS_TRANSACTIONAL_EMAIL_TEMPLATE_IDS}.get(
+        template_name
+    )
+    if not template_id or not settings.WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY:
+        return None
+    try:
+        enabled = posthoganalytics.feature_enabled(
+            WORKFLOWS_TRANSACTIONAL_EMAIL_FLAG,
+            dest.get("distinct_id") or dest["raw_email"],
+            person_properties={"email": dest["raw_email"], "transactional_email_template": template_name},
+            send_feature_flag_events=False,
+        )
+    except Exception as err:
+        logger.warning("email_workflows_flag_check_failed", template_name=template_name, error=str(err))
+        return None
+    return template_id if enabled else None
+
+
+def _unescape_for_liquid(value: Any, key: Any = None) -> Any:
+    """Undo the html.escape in sanitize_email_properties, and keep the URL defang.
+
+    Customer.io outputs Liquid values as they are, so the properties arrive pre-escaped. Workflows escapes Liquid
+    output itself, so a pre-escaped value renders as `&amp;amp;` and shows the entity to the reader.
+    """
+    if key is not None and _classify_key(key) == "passthrough":
+        return value
+    if isinstance(value, str):
+        return html.unescape(value)
+    if isinstance(value, dict):
+        return {k: _unescape_for_liquid(v, key=k) for k, v in value.items()}
+    if isinstance(value, list):
+        return [_unescape_for_liquid(item) for item in value]
+    return value
+
+
+def _send_one_via_workflows(
+    dest: dict[str, str], campaign_key: str, template_id: str, properties: dict
+) -> dict[str, Any]:
+    # Workflows templates read only `variables.*`, so the Customer.io `customer.*` attributes travel in the data.
+    customer = {"email": dest["raw_email"]}
+    if dest.get("distinct_id"):
+        customer["id"] = dest["distinct_id"]
+    if dest.get("first_name"):
+        customer["first_name"] = sanitize_email_string(dest["first_name"])
+    payload: dict[str, Any] = {
+        "to": dest["raw_email"],
+        "message_data": _unescape_for_liquid({**properties, "customer": customer}),
+    }
+    if dest.get("distinct_id"):
+        payload["identifiers"] = {"distinct_id": dest["distinct_id"]}
+
+    # A retry after a read timeout must not send the email twice.
+    idempotency_key = hashlib.sha256(f"{campaign_key}\n{dest['raw_email']}".encode()).hexdigest()
+    try:
+        response = requests.post(
+            f"{settings.WORKFLOWS_TRANSACTIONAL_EMAIL_API_URL}/api/projects/"
+            f"{settings.WORKFLOWS_TRANSACTIONAL_EMAIL_PROJECT_ID}/messaging_templates/{template_id}/send/",
+            headers={
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {settings.WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY}",
+                "Idempotency-Key": idempotency_key,
+            },
+            json=payload,
+            timeout=30,
+        )
+    except requests.ConnectionError as err:
+        # The request never reached Workflows. A read timeout is not caught here, because Workflows can have
+        # queued the email before it timed out.
+        raise WorkflowsSendRejected(str(err)) from err
+
+    if response.status_code == 409:
+        # An earlier attempt with this key is still in flight and can still send.
+        raise Exception(f"Workflows API error: {response.status_code} - {response.text}")
+    if response.status_code != 202:
+        raise WorkflowsSendRejected(f"{response.status_code} - {response.text}")
+    return response.json()
+
+
+def _send_one_via_customer_io(dest: dict[str, str], template_name: str, properties: dict) -> dict[str, Any]:
+    if not settings.CUSTOMER_IO_API_KEY:
+        raise Exception("Missing Customer.io API key")
+
+    identifiers: dict[str, str] = {"email": dest["raw_email"]}
+    if dest.get("distinct_id"):
+        identifiers["id"] = dest["distinct_id"]
+
+    payload = {
+        "transactional_message_id": get_customer_io_template_id(template_name),
+        "to": dest["raw_email"],
+        "identifiers": identifiers,
+        "message_data": properties,
+    }
+
+    response = requests.post(
+        f"{settings.CUSTOMER_IO_API_URL}/v1/send/email",
+        headers={
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {settings.CUSTOMER_IO_API_KEY}",
+        },
+        json=payload,
+        timeout=30,
+    )
+
+    if response.status_code != 200:
+        raise Exception(f"Customer.io API error: {response.status_code} - {response.text}")
+
+    return response.json()
+
+
+def _send_one(dest: dict[str, str], campaign_key: str, template_name: str, properties: dict) -> tuple[str, dict]:
+    """Send one email and return the provider that sent it with the provider's response."""
+    workflows_template_id = get_workflows_template_id(template_name, dest)
+    if workflows_template_id:
+        try:
+            return "workflows", _send_one_via_workflows(dest, campaign_key, workflows_template_id, properties)
+        except WorkflowsSendRejected as err:
+            logger.warning("email_workflows_send_rejected", template_name=template_name, error=str(err))
+            EMAIL_WORKFLOWS_FALLBACK_COUNTER.labels(template_name=template_name).inc()
+    return "customer_io", _send_one_via_customer_io(dest, template_name, properties)
+
+
+# Note: this http sender is set up to send via templates, so all the configuration is done in Customer.io or in
+# the Workflows message template - i.e. no subject, body, etc.
 def _send_via_http(
     to: list[dict[str, str]],
     campaign_key: str,
     template_name: str,
     properties: dict,
 ) -> None:
-    """Sends emails using Customer.io API"""
-    customerio_api_key = settings.CUSTOMER_IO_API_KEY
-
-    if not customerio_api_key:
-        raise Exception("Missing Customer.io API key")
-
-    headers = {
-        "Content-Type": "application/json",
-        "Authorization": f"Bearer {customerio_api_key}",
-    }
+    """Sends emails using the Customer.io API, or the Workflows transactional email API where the flag matches"""
+    if not is_http_email_service_available():
+        raise Exception("Missing Customer.io and Workflows API keys")
 
     sent_count = 0
     already_sent_count = 0  # delivered in a prior run — not a failure if the batch later aborts
@@ -237,25 +376,7 @@ def _send_via_http(
                     already_sent_count += 1
                     continue
 
-                identifiers: dict[str, str] = {"email": dest["raw_email"]}
-                if dest.get("distinct_id"):
-                    identifiers["id"] = dest["distinct_id"]
-
-                payload = {
-                    "transactional_message_id": get_customer_io_template_id(template_name),
-                    "to": dest["raw_email"],
-                    "identifiers": identifiers,
-                    "message_data": properties,
-                }
-
-                response = requests.post(
-                    f"{settings.CUSTOMER_IO_API_URL}/v1/send/email", headers=headers, json=payload, timeout=30
-                )
-
-                if response.status_code != 200:
-                    raise Exception(f"Customer.io API error: {response.status_code} - {response.text}")
-
-                provider_response = response.json()
+                provider, provider_response = _send_one(dest, campaign_key, template_name, properties)
 
                 # Mark delivery before any non-essential step so a delivered email is never
                 # recorded as rejected (raise_if_delivery_rejected keys off sent_at). Analytics
@@ -271,6 +392,7 @@ def _send_via_http(
                             "template_name": template_name,
                             "campaign_key": campaign_key,
                             "recipient_email": dest["raw_email"],
+                            "provider": provider,
                             **provider_response,
                         },
                     )
@@ -584,6 +706,8 @@ def add_recipient(self, email: str, name: Optional[str] = None, distinct_id: Opt
         }
         if distinct_id:
             recipient_data["distinct_id"] = distinct_id
+        if name:
+            recipient_data["first_name"] = name
         self.to.append(recipient_data)
 
     def add_user_recipient(self, user: "User", email_override: Optional[str] = None) -> None:
diff --git a/posthog/test/test_email.py b/posthog/test/test_email.py
index dc4d84775623..a70f54b641e2 100644
--- a/posthog/test/test_email.py
+++ b/posthog/test/test_email.py
@@ -339,6 +339,7 @@ def test_send_via_http_success(self, mock_post, mock_capture) -> None:
                     "template_name": "2fa_enabled",
                     "campaign_key": "test_campaign",
                     "recipient_email": "test@posthog.com",
+                    "provider": "customer_io",
                     "delivery_id": "test_delivery_id",
                     "queued_at": 1604977406,
                 },
@@ -399,6 +400,94 @@ def test_send_via_http_api_error(self, mock_post) -> None:
             record = MessagingRecord.objects.filter(campaign_key="test_campaign").first()
             self.assertIsNone(record)
 
+    @parameterized.expand(
+        [
+            ("workflows_accepts", True, [202], ["workflows"], "workflows"),
+            ("workflows_rejects_falls_back", True, [400, 200], ["workflows", "customer_io"], "customer_io"),
+            ("workflows_in_flight_sends_nothing", True, [409], ["workflows"], None),
+            ("flag_off_uses_customer_io", False, [200], ["customer_io"], "customer_io"),
+        ]
+    )
+    @patch("posthoganalytics.capture")
+    @patch("posthog.email.posthoganalytics.feature_enabled")
+    @patch("posthog.email.requests.post")
+    def test_send_via_http_routes_through_workflows_when_flag_matches(
+        self,
+        _name: str,
+        flag_enabled: bool,
+        status_codes: list[int],
+        expected_providers: list[str],
+        sent_by: str | None,
+        mock_post: MagicMock,
+        mock_feature_enabled: MagicMock,
+        mock_capture: MagicMock,
+    ) -> None:
+        mock_feature_enabled.return_value = flag_enabled
+        responses = []
+        for status_code in status_codes:
+            response = MagicMock(status_code=status_code, text="error")
+            response.json.return_value = {"message_id": "m1"} if status_code == 202 else {"delivery_id": "d1"}
+            responses.append(response)
+        mock_post.side_effect = responses
+
+        with (
+            override_instance_config("EMAIL_HOST", "localhost"),
+            self.settings(
+                CUSTOMER_IO_API_KEY="cio-key",
+                WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY="phs_test",
+                WORKFLOWS_TRANSACTIONAL_EMAIL_API_URL="https://us.posthog.com",
+                WORKFLOWS_TRANSACTIONAL_EMAIL_PROJECT_ID=2,
+                WORKFLOWS_TRANSACTIONAL_EMAIL_TEMPLATE_IDS={"2fa_enabled": "tmpl-1"},
+            ),
+        ):
+            message = EmailMessage(
+                campaign_key="routing_campaign",
+                subject="Test subject",
+                template_name="2fa_enabled",
+                template_context={"team_name": "O'Brien & Co", "url": "https://app.posthog.com/r?a=1&b=2"},
+                use_http=True,
+            )
+            message.add_recipient("test@posthog.com", "Test", distinct_id="user-1")
+            message.send(send_async=False)
+
+        providers = [
+            "workflows" if call.args[0].startswith("https://us.posthog.com") else "customer_io"
+            for call in mock_post.call_args_list
+        ]
+        self.assertEqual(providers, expected_providers)
+        mock_feature_enabled.assert_called_once_with(
+            "transactional-emails-via-workflows",
+            "user-1",
+            person_properties={"email": "test@posthog.com", "transactional_email_template": "2fa_enabled"},
+            send_feature_flag_events=False,
+        )
+
+        workflows_call = next((c for c in mock_post.call_args_list if c.args[0].startswith("https://us.")), None)
+        if workflows_call is not None:
+            self.assertEqual(
+                workflows_call.args[0], "https://us.posthog.com/api/projects/2/messaging_templates/tmpl-1/send/"
+            )
+            self.assertEqual(workflows_call.kwargs["headers"]["Authorization"], "Bearer phs_test")
+            self.assertIn("Idempotency-Key", workflows_call.kwargs["headers"])
+            payload = workflows_call.kwargs["json"]
+            self.assertEqual(payload["identifiers"], {"distinct_id": "user-1"})
+            # Workflows escapes Liquid output itself, so values must arrive unescaped or they render as &amp;amp;
+            self.assertEqual(payload["message_data"]["team_name"], "O'Brien & Co")
+            self.assertEqual(payload["message_data"]["url"], "https://app.posthog.com/r?a=1&b=2")
+            self.assertEqual(
+                payload["message_data"]["customer"],
+                {"email": "test@posthog.com", "id": "user-1", "first_name": "Test"},
+            )
+
+        record = MessagingRecord.objects.filter(campaign_key="routing_campaign").first()
+        if sent_by is None:
+            self.assertTrue(record is None or record.sent_at is None)
+            mock_capture.assert_not_called()
+        else:
+            assert record is not None
+            self.assertIsNotNone(record.sent_at)
+            self.assertEqual(mock_capture.call_args.kwargs["properties"]["provider"], sent_by)
+
     def test_sanitize_email_properties(self) -> None:
         # Test with various types of input including potential HTML injection
         properties = {

From 4e1deb7ac5d51c16844ed3aeabf695bfa09df51a Mon Sep 17 00:00:00 2001
From: Harley Alexander <43975092+mayteio@users.noreply.github.com>
Date: Thu, 22 Oct 2026 14:34:34 +0100
Subject: [PATCH 2/9] feat(email): capture failed and workflows-rejected
 transactional emails per template

A failed send only reached error tracking, and the exception carries no template name, so failures could not be counted per template.

- `transactional email failed` fires when a send raises, with `template_name`, `error_type` and `error`.
- `transactional email workflows rejected` fires when Workflows refuses a send and Customer.io sends it instead.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Generated-By: PostHog Desktop
Task-Id: b07b2dac-4768-46c8-8930-75b8c8aecc31
---
 posthog/email.py           | 53 ++++++++++++++++++++++++++++----------
 posthog/test/test_email.py | 28 ++++++++++++++++----
 2 files changed, 62 insertions(+), 19 deletions(-)

diff --git a/posthog/email.py b/posthog/email.py
index e3906b93c0cd..e4f38fe3d9e0 100644
--- a/posthog/email.py
+++ b/posthog/email.py
@@ -338,6 +338,17 @@ def _send_one_via_customer_io(dest: dict[str, str], template_name: str, properti
     return response.json()
 
 
+def _capture_email_event(dest: dict[str, str], event: str, properties: dict[str, Any]) -> None:
+    try:
+        posthoganalytics.capture(
+            distinct_id=dest.get("distinct_id") or dest["raw_email"],
+            event=event,
+            properties={"recipient_email": dest["raw_email"], **properties},
+        )
+    except Exception as analytics_err:
+        logger.warning("email_send_analytics_capture_failed", error=str(analytics_err))
+
+
 def _send_one(dest: dict[str, str], campaign_key: str, template_name: str, properties: dict) -> tuple[str, dict]:
     """Send one email and return the provider that sent it with the provider's response."""
     workflows_template_id = get_workflows_template_id(template_name, dest)
@@ -347,6 +358,11 @@ def _send_one(dest: dict[str, str], campaign_key: str, template_name: str, prope
         except WorkflowsSendRejected as err:
             logger.warning("email_workflows_send_rejected", template_name=template_name, error=str(err))
             EMAIL_WORKFLOWS_FALLBACK_COUNTER.labels(template_name=template_name).inc()
+            _capture_email_event(
+                dest,
+                "transactional email workflows rejected",
+                {"template_name": template_name, "campaign_key": campaign_key, "error": str(err)[:500]},
+            )
     return "customer_io", _send_one_via_customer_io(dest, template_name, properties)
 
 
@@ -364,6 +380,7 @@ def _send_via_http(
 
     sent_count = 0
     already_sent_count = 0  # delivered in a prior run — not a failure if the batch later aborts
+    dest: dict[str, str] | None = None
     try:
         for dest in to:
             with transaction.atomic():
@@ -384,26 +401,34 @@ def _send_via_http(
                 record.sent_at = timezone.now()
                 record.save()
 
-                try:
-                    posthoganalytics.capture(
-                        distinct_id=dest.get("distinct_id") or dest["raw_email"],
-                        event="transactional email triggered",
-                        properties={
-                            "template_name": template_name,
-                            "campaign_key": campaign_key,
-                            "recipient_email": dest["raw_email"],
-                            "provider": provider,
-                            **provider_response,
-                        },
-                    )
-                except Exception as analytics_err:
-                    logger.warning("email_send_analytics_capture_failed", error=str(analytics_err))
+                _capture_email_event(
+                    dest,
+                    "transactional email triggered",
+                    {
+                        "template_name": template_name,
+                        "campaign_key": campaign_key,
+                        "provider": provider,
+                        **provider_response,
+                    },
+                )
 
                 EMAIL_SEND_COUNTER.labels(outcome="sent", transport="http").inc()
                 sent_count += 1
 
     except Exception as err:
         capture_exception(err)  # already logs the traceback via logger.exception
+        # The exception carries no template name, so this event is what lets a dashboard count failures per template.
+        if dest:
+            _capture_email_event(
+                dest,
+                "transactional email failed",
+                {
+                    "template_name": template_name,
+                    "campaign_key": campaign_key,
+                    "error_type": type(err).__name__,
+                    "error": str(err)[:500],
+                },
+            )
         # Count every recipient that did not get through (the failing one + any not yet attempted),
         # so `failed` shares the per-recipient unit with `sent` instead of registering once per batch.
         EMAIL_SEND_COUNTER.labels(outcome="failed", transport="http").inc(len(to) - already_sent_count - sent_count)
diff --git a/posthog/test/test_email.py b/posthog/test/test_email.py
index a70f54b641e2..7b00246530d4 100644
--- a/posthog/test/test_email.py
+++ b/posthog/test/test_email.py
@@ -402,10 +402,24 @@ def test_send_via_http_api_error(self, mock_post) -> None:
 
     @parameterized.expand(
         [
-            ("workflows_accepts", True, [202], ["workflows"], "workflows"),
-            ("workflows_rejects_falls_back", True, [400, 200], ["workflows", "customer_io"], "customer_io"),
-            ("workflows_in_flight_sends_nothing", True, [409], ["workflows"], None),
-            ("flag_off_uses_customer_io", False, [200], ["customer_io"], "customer_io"),
+            ("workflows_accepts", True, [202], ["workflows"], "workflows", ["transactional email triggered"]),
+            (
+                "workflows_rejects_falls_back",
+                True,
+                [400, 200],
+                ["workflows", "customer_io"],
+                "customer_io",
+                ["transactional email workflows rejected", "transactional email triggered"],
+            ),
+            ("workflows_in_flight_sends_nothing", True, [409], ["workflows"], None, ["transactional email failed"]),
+            (
+                "flag_off_uses_customer_io",
+                False,
+                [200],
+                ["customer_io"],
+                "customer_io",
+                ["transactional email triggered"],
+            ),
         ]
     )
     @patch("posthoganalytics.capture")
@@ -418,6 +432,7 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
         status_codes: list[int],
         expected_providers: list[str],
         sent_by: str | None,
+        expected_events: list[str],
         mock_post: MagicMock,
         mock_feature_enabled: MagicMock,
         mock_capture: MagicMock,
@@ -479,10 +494,13 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
                 {"email": "test@posthog.com", "id": "user-1", "first_name": "Test"},
             )
 
+        self.assertEqual([c.kwargs["event"] for c in mock_capture.call_args_list], expected_events)
+        self.assertTrue(
+            all(c.kwargs["properties"]["template_name"] == "2fa_enabled" for c in mock_capture.call_args_list)
+        )
         record = MessagingRecord.objects.filter(campaign_key="routing_campaign").first()
         if sent_by is None:
             self.assertTrue(record is None or record.sent_at is None)
-            mock_capture.assert_not_called()
         else:
             assert record is not None
             self.assertIsNotNone(record.sent_at)

From 68464dffdacd56604dbc5e571480f16106588181 Mon Sep 17 00:00:00 2001
From: Harley Alexander <43975092+mayteio@users.noreply.github.com>
Date: Thu, 22 Oct 2026 15:40:07 +0100
Subject: [PATCH 3/9] refactor(email): match the posthog-python emails.send
 shape and send only password_reset

- `_queue_workflows_email` takes the same arguments as posthog-python's `Emails.send` and raises an error with the same `code` and `status_code`, so the SDK can replace one function body once it ships.
- Sends go to `/api/projects/@current/...`, because the project secret API key names the project. `WORKFLOWS_TRANSACTIONAL_EMAIL_PROJECT_ID` is gone.
- The fallback rule reads `code` and `status_code`: a refused connection or a send that queued nothing falls back to Customer.io, a timeout or an in-flight 409 does not.
- The request timeout drops to 10 seconds, because the caller holds the MessagingRecord row lock while it waits.
- `WORKFLOWS_TEMPLATE_ID_MAP` keeps only `password_reset`, for a single production test.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Generated-By: PostHog Desktop
Task-Id: b07b2dac-4768-46c8-8930-75b8c8aecc31
---
 ee/settings.py             |   4 +-
 posthog/email.py           | 109 ++++++++++++++++++++++++++-----------
 posthog/test/test_email.py |  27 +++++++--
 3 files changed, 102 insertions(+), 38 deletions(-)

diff --git a/ee/settings.py b/ee/settings.py
index 4a91423054f0..5a5aab2be7c1 100644
--- a/ee/settings.py
+++ b/ee/settings.py
@@ -63,12 +63,12 @@
 CUSTOMER_IO_API_KEY = get_from_env("CUSTOMER_IO_API_KEY", "", type_cast=str)
 CUSTOMER_IO_API_URL = get_from_env("CUSTOMER_IO_API_URL", "https://api-eu.customer.io", type_cast=str)
 
-# The Workflows transactional email API that sends PostHog's own emails. Every region sends through the US project.
+# The Workflows transactional email API that sends PostHog's own emails. Every region sends through the US project,
+# which the project secret API key picks.
 WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY = get_from_env("WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY", "", type_cast=str)
 WORKFLOWS_TRANSACTIONAL_EMAIL_API_URL = get_from_env(
     "WORKFLOWS_TRANSACTIONAL_EMAIL_API_URL", "https://us.posthog.com", type_cast=str
 )
-WORKFLOWS_TRANSACTIONAL_EMAIL_PROJECT_ID = get_from_env("WORKFLOWS_TRANSACTIONAL_EMAIL_PROJECT_ID", 2, type_cast=int)
 # Comma-separated `template_name=message_template_id` pairs. They override WORKFLOWS_TEMPLATE_ID_MAP in
 # posthog/email.py, so a local or staging instance can send with its own copies of the templates.
 WORKFLOWS_TRANSACTIONAL_EMAIL_TEMPLATE_IDS: dict[str, str] = dict(
diff --git a/posthog/email.py b/posthog/email.py
index e4f38fe3d9e0..f3868b0f1bd7 100644
--- a/posthog/email.py
+++ b/posthog/email.py
@@ -168,9 +168,7 @@ def is_email_available(with_absolute_urls: bool = False) -> bool:
 # Message template ids in the US project, for templates that can send through Workflows instead of Customer.io.
 # A template sends through Workflows only for recipients the flag below matches.
 WORKFLOWS_TEMPLATE_ID_MAP: dict[str, str] = {
-    "approval_requested": "01a16311-33e6-0000-c533-f383fd8e70f0",
     "password_reset": "01a1597a-5d91-0000-99fa-9eeb18d293c5",
-    "proxy_provisioned": "01a16311-b3de-0000-8aec-7c865f3e5432",
 }
 
 # Release conditions match the `email` and `transactional_email_template` person properties passed at evaluation,
@@ -223,8 +221,27 @@ def get_customer_io_template_id(template_name: str) -> str:
     return template_id
 
 
-class WorkflowsSendRejected(Exception):
-    """Workflows did not take the email, so another provider can send it without a duplicate."""
+class WorkflowsEmailError(Exception):
+    """Workflows did not queue the email, or the caller could not tell whether it did.
+
+    `code` and `status_code` match posthog-python's `EmailError`, so the fallback rule keeps working when the
+    SDK sends instead. `status_code` is None when no response came back.
+    """
+
+    def __init__(self, message: str, *, code: str, status_code: int | None = None) -> None:
+        super().__init__(message)
+        self.code = code
+        self.status_code = status_code
+
+
+def _workflows_error_allows_fallback(err: WorkflowsEmailError) -> bool:
+    """Whether Customer.io can send the email without the risk of a duplicate."""
+    if err.status_code is None:
+        # A refused connection never reached Workflows, and a client that queued nothing sent nothing. After a
+        # timeout, Workflows can have queued the email.
+        return err.code in ("connection_error", "not_queued")
+    # A 409 means an earlier attempt with the same idempotency key is still in flight and can still send.
+    return err.status_code != 409
 
 
 def get_workflows_template_id(template_name: str, dest: dict[str, str]) -> str | None:
@@ -264,47 +281,73 @@ def _unescape_for_liquid(value: Any, key: Any = None) -> Any:
     return value
 
 
-def _send_one_via_workflows(
-    dest: dict[str, str], campaign_key: str, template_id: str, properties: dict
+def _queue_workflows_email(
+    *,
+    to: str,
+    template_id: str,
+    template_data: dict[str, Any],
+    distinct_id: str | None = None,
+    idempotency_key: str,
 ) -> dict[str, Any]:
-    # Workflows templates read only `variables.*`, so the Customer.io `customer.*` attributes travel in the data.
-    customer = {"email": dest["raw_email"]}
-    if dest.get("distinct_id"):
-        customer["id"] = dest["distinct_id"]
-    if dest.get("first_name"):
-        customer["first_name"] = sanitize_email_string(dest["first_name"])
-    payload: dict[str, Any] = {
-        "to": dest["raw_email"],
-        "message_data": _unescape_for_liquid({**properties, "customer": customer}),
-    }
-    if dest.get("distinct_id"):
-        payload["identifiers"] = {"distinct_id": dest["distinct_id"]}
+    """Queue one email from a Workflows transactional template and return its `message_id` and `replayed`.
 
-    # A retry after a read timeout must not send the email twice.
-    idempotency_key = hashlib.sha256(f"{campaign_key}\n{dest['raw_email']}".encode()).hexdigest()
+    Takes the same arguments as posthog-python's `Emails.send`, so the SDK can replace this body once it ships.
+    """
+    payload: dict[str, Any] = {"to": to, "message_data": template_data}
+    if distinct_id:
+        payload["identifiers"] = {"distinct_id": distinct_id}
     try:
         response = requests.post(
-            f"{settings.WORKFLOWS_TRANSACTIONAL_EMAIL_API_URL}/api/projects/"
-            f"{settings.WORKFLOWS_TRANSACTIONAL_EMAIL_PROJECT_ID}/messaging_templates/{template_id}/send/",
+            # The project secret API key names the project, so `@current` resolves to it.
+            f"{settings.WORKFLOWS_TRANSACTIONAL_EMAIL_API_URL}/api/projects/@current/messaging_templates/{template_id}/send/",
             headers={
                 "Content-Type": "application/json",
                 "Authorization": f"Bearer {settings.WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY}",
                 "Idempotency-Key": idempotency_key,
             },
             json=payload,
-            timeout=30,
+            # The caller holds the MessagingRecord row lock for the whole request, so keep it short.
+            timeout=10,
         )
     except requests.ConnectionError as err:
-        # The request never reached Workflows. A read timeout is not caught here, because Workflows can have
-        # queued the email before it timed out.
-        raise WorkflowsSendRejected(str(err)) from err
+        raise WorkflowsEmailError(str(err), code="connection_error") from err
+    except requests.Timeout as err:
+        raise WorkflowsEmailError(str(err), code="timeout") from err
 
-    if response.status_code == 409:
-        # An earlier attempt with this key is still in flight and can still send.
-        raise Exception(f"Workflows API error: {response.status_code} - {response.text}")
     if response.status_code != 202:
-        raise WorkflowsSendRejected(f"{response.status_code} - {response.text}")
-    return response.json()
+        try:
+            code = response.json().get("code") or "api_error"
+        except ValueError:
+            code = "api_error"
+        raise WorkflowsEmailError(
+            f"{response.status_code} - {response.text}", code=code, status_code=response.status_code
+        )
+    body = response.json()
+    return {"message_id": body.get("message_id"), "replayed": bool(body.get("replayed", False))}
+
+
+def _send_one_via_workflows(
+    dest: dict[str, str], campaign_key: str, template_id: str, properties: dict
+) -> dict[str, Any]:
+    # Workflows templates read only `variables.*`, so the Customer.io `customer.*` attributes travel in the data.
+    customer = {"email": dest["raw_email"]}
+    if dest.get("distinct_id"):
+        customer["id"] = dest["distinct_id"]
+    if dest.get("first_name"):
+        customer["first_name"] = sanitize_email_string(dest["first_name"])
+
+    result = _queue_workflows_email(
+        to=dest["raw_email"],
+        template_id=template_id,
+        template_data=_unescape_for_liquid({**properties, "customer": customer}),
+        distinct_id=dest.get("distinct_id") or None,
+        # Derived from the email itself, so a Celery retry of the whole send cannot queue it twice.
+        idempotency_key=hashlib.sha256(f"{campaign_key}\n{dest['raw_email']}".encode()).hexdigest(),
+    )
+    if not result["message_id"]:
+        # A disabled SDK client queues nothing and returns no ID.
+        raise WorkflowsEmailError("Workflows queued nothing", code="not_queued")
+    return result
 
 
 def _send_one_via_customer_io(dest: dict[str, str], template_name: str, properties: dict) -> dict[str, Any]:
@@ -355,7 +398,9 @@ def _send_one(dest: dict[str, str], campaign_key: str, template_name: str, prope
     if workflows_template_id:
         try:
             return "workflows", _send_one_via_workflows(dest, campaign_key, workflows_template_id, properties)
-        except WorkflowsSendRejected as err:
+        except WorkflowsEmailError as err:
+            if not _workflows_error_allows_fallback(err):
+                raise
             logger.warning("email_workflows_send_rejected", template_name=template_name, error=str(err))
             EMAIL_WORKFLOWS_FALLBACK_COUNTER.labels(template_name=template_name).inc()
             _capture_email_event(
diff --git a/posthog/test/test_email.py b/posthog/test/test_email.py
index 7b00246530d4..a1b7a09a94cb 100644
--- a/posthog/test/test_email.py
+++ b/posthog/test/test_email.py
@@ -17,6 +17,7 @@
 from django.test import override_settings
 from django.utils import timezone
 
+import requests
 from parameterized import parameterized
 from prometheus_client import REGISTRY
 
@@ -412,6 +413,22 @@ def test_send_via_http_api_error(self, mock_post) -> None:
                 ["transactional email workflows rejected", "transactional email triggered"],
             ),
             ("workflows_in_flight_sends_nothing", True, [409], ["workflows"], None, ["transactional email failed"]),
+            (
+                "workflows_unreachable_falls_back",
+                True,
+                [requests.ConnectionError("refused"), 200],
+                ["workflows", "customer_io"],
+                "customer_io",
+                ["transactional email workflows rejected", "transactional email triggered"],
+            ),
+            (
+                "workflows_timeout_sends_nothing",
+                True,
+                [requests.ReadTimeout("no response")],
+                ["workflows"],
+                None,
+                ["transactional email failed"],
+            ),
             (
                 "flag_off_uses_customer_io",
                 False,
@@ -429,7 +446,7 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
         self,
         _name: str,
         flag_enabled: bool,
-        status_codes: list[int],
+        status_codes: list[int | Exception],
         expected_providers: list[str],
         sent_by: str | None,
         expected_events: list[str],
@@ -438,8 +455,11 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
         mock_capture: MagicMock,
     ) -> None:
         mock_feature_enabled.return_value = flag_enabled
-        responses = []
+        responses: list[MagicMock | Exception] = []
         for status_code in status_codes:
+            if isinstance(status_code, Exception):
+                responses.append(status_code)
+                continue
             response = MagicMock(status_code=status_code, text="error")
             response.json.return_value = {"message_id": "m1"} if status_code == 202 else {"delivery_id": "d1"}
             responses.append(response)
@@ -451,7 +471,6 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
                 CUSTOMER_IO_API_KEY="cio-key",
                 WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY="phs_test",
                 WORKFLOWS_TRANSACTIONAL_EMAIL_API_URL="https://us.posthog.com",
-                WORKFLOWS_TRANSACTIONAL_EMAIL_PROJECT_ID=2,
                 WORKFLOWS_TRANSACTIONAL_EMAIL_TEMPLATE_IDS={"2fa_enabled": "tmpl-1"},
             ),
         ):
@@ -480,7 +499,7 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
         workflows_call = next((c for c in mock_post.call_args_list if c.args[0].startswith("https://us.")), None)
         if workflows_call is not None:
             self.assertEqual(
-                workflows_call.args[0], "https://us.posthog.com/api/projects/2/messaging_templates/tmpl-1/send/"
+                workflows_call.args[0], "https://us.posthog.com/api/projects/@current/messaging_templates/tmpl-1/send/"
             )
             self.assertEqual(workflows_call.kwargs["headers"]["Authorization"], "Bearer phs_test")
             self.assertIn("Idempotency-Key", workflows_call.kwargs["headers"])

From abaed0e38a652e452200f20e628845ac8ba560f7 Mon Sep 17 00:00:00 2001
From: Harley Alexander <43975092+mayteio@users.noreply.github.com>
Date: Thu, 22 Oct 2026 15:46:24 +0100
Subject: [PATCH 4/9] refactor(email): send with the shared transactional email
 key and host

- Workflows sends use `POSTHOG_TRANSACTIONAL_EMAIL_API_KEY` and `POSTHOG_HOST`, the settings `posthog/transactional_email.py` already reads, so no second key needs provisioning. `WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY` and `_API_URL` are gone.
- HTTP email is available only when Customer.io is configured again. The shared key alone no longer moves an instance off SMTP, so an environment that set it for another feature keeps sending email.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Generated-By: PostHog Desktop
Task-Id: b07b2dac-4768-46c8-8930-75b8c8aecc31
---
 ee/settings.py             |  8 ++------
 posthog/email.py           | 12 ++++++------
 posthog/test/test_email.py |  4 ++--
 3 files changed, 10 insertions(+), 14 deletions(-)

diff --git a/ee/settings.py b/ee/settings.py
index 5a5aab2be7c1..3f5bb6588427 100644
--- a/ee/settings.py
+++ b/ee/settings.py
@@ -63,12 +63,8 @@
 CUSTOMER_IO_API_KEY = get_from_env("CUSTOMER_IO_API_KEY", "", type_cast=str)
 CUSTOMER_IO_API_URL = get_from_env("CUSTOMER_IO_API_URL", "https://api-eu.customer.io", type_cast=str)
 
-# The Workflows transactional email API that sends PostHog's own emails. Every region sends through the US project,
-# which the project secret API key picks.
-WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY = get_from_env("WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY", "", type_cast=str)
-WORKFLOWS_TRANSACTIONAL_EMAIL_API_URL = get_from_env(
-    "WORKFLOWS_TRANSACTIONAL_EMAIL_API_URL", "https://us.posthog.com", type_cast=str
-)
+# Workflows sends PostHog's own emails with POSTHOG_TRANSACTIONAL_EMAIL_API_KEY at POSTHOG_HOST, the same settings as
+# posthog/transactional_email.py. Every region sends through the US project, which the key picks.
 # Comma-separated `template_name=message_template_id` pairs. They override WORKFLOWS_TEMPLATE_ID_MAP in
 # posthog/email.py, so a local or staging instance can send with its own copies of the templates.
 WORKFLOWS_TRANSACTIONAL_EMAIL_TEMPLATE_IDS: dict[str, str] = dict(
diff --git a/posthog/email.py b/posthog/email.py
index f3868b0f1bd7..1e01a04cbbf0 100644
--- a/posthog/email.py
+++ b/posthog/email.py
@@ -55,9 +55,9 @@ def inline_css(value: str) -> str:
 def is_http_email_service_available() -> bool:
     """
     Returns whether HTTP email services are available on this instance (i.e. settings are in place).
-    These are Customer.io and the Workflows transactional email API.
+    This currently only supports Customer.io. Workflows only takes over emails that this path sends.
     """
-    return bool(settings.CUSTOMER_IO_API_KEY or settings.WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY)
+    return bool(settings.CUSTOMER_IO_API_KEY)
 
 
 def is_smtp_email_service_available() -> bool:
@@ -249,7 +249,7 @@ def get_workflows_template_id(template_name: str, dest: dict[str, str]) -> str |
     template_id = {**WORKFLOWS_TEMPLATE_ID_MAP, **settings.WORKFLOWS_TRANSACTIONAL_EMAIL_TEMPLATE_IDS}.get(
         template_name
     )
-    if not template_id or not settings.WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY:
+    if not template_id or not settings.POSTHOG_TRANSACTIONAL_EMAIL_API_KEY:
         return None
     try:
         enabled = posthoganalytics.feature_enabled(
@@ -299,10 +299,10 @@ def _queue_workflows_email(
     try:
         response = requests.post(
             # The project secret API key names the project, so `@current` resolves to it.
-            f"{settings.WORKFLOWS_TRANSACTIONAL_EMAIL_API_URL}/api/projects/@current/messaging_templates/{template_id}/send/",
+            f"{settings.POSTHOG_HOST.rstrip('/')}/api/projects/@current/messaging_templates/{template_id}/send/",
             headers={
                 "Content-Type": "application/json",
-                "Authorization": f"Bearer {settings.WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY}",
+                "Authorization": f"Bearer {settings.POSTHOG_TRANSACTIONAL_EMAIL_API_KEY}",
                 "Idempotency-Key": idempotency_key,
             },
             json=payload,
@@ -421,7 +421,7 @@ def _send_via_http(
 ) -> None:
     """Sends emails using the Customer.io API, or the Workflows transactional email API where the flag matches"""
     if not is_http_email_service_available():
-        raise Exception("Missing Customer.io and Workflows API keys")
+        raise Exception("Missing Customer.io API key")
 
     sent_count = 0
     already_sent_count = 0  # delivered in a prior run — not a failure if the batch later aborts
diff --git a/posthog/test/test_email.py b/posthog/test/test_email.py
index a1b7a09a94cb..34fe3795560c 100644
--- a/posthog/test/test_email.py
+++ b/posthog/test/test_email.py
@@ -469,8 +469,8 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
             override_instance_config("EMAIL_HOST", "localhost"),
             self.settings(
                 CUSTOMER_IO_API_KEY="cio-key",
-                WORKFLOWS_TRANSACTIONAL_EMAIL_API_KEY="phs_test",
-                WORKFLOWS_TRANSACTIONAL_EMAIL_API_URL="https://us.posthog.com",
+                POSTHOG_TRANSACTIONAL_EMAIL_API_KEY="phs_test",
+                POSTHOG_HOST="https://us.posthog.com",
                 WORKFLOWS_TRANSACTIONAL_EMAIL_TEMPLATE_IDS={"2fa_enabled": "tmpl-1"},
             ),
         ):

From 0c14c5ee5b0fbe35aac82e050a1e4515ccf7bac9 Mon Sep 17 00:00:00 2001
From: Harley Alexander <43975092+mayteio@users.noreply.github.com>
Date: Thu, 22 Oct 2026 15:52:03 +0100
Subject: [PATCH 5/9] refactor(email): drop the per-instance workflows template
 id override

`WORKFLOWS_TRANSACTIONAL_EMAIL_TEMPLATE_IDS` only let a local stack point at its own copies of the templates. Production and dev both send through the US project's templates, so `WORKFLOWS_TEMPLATE_ID_MAP` is the only source of template IDs. The feature flag decides whether a template sends through Workflows; the map only says which template to send.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Generated-By: PostHog Desktop
Task-Id: b07b2dac-4768-46c8-8930-75b8c8aecc31
---
 ee/settings.py             | 10 +---------
 posthog/email.py           |  4 +---
 posthog/test/test_email.py |  2 +-
 3 files changed, 3 insertions(+), 13 deletions(-)

diff --git a/ee/settings.py b/ee/settings.py
index 3f5bb6588427..23f9f12f198e 100644
--- a/ee/settings.py
+++ b/ee/settings.py
@@ -5,7 +5,7 @@
 import os
 
 from posthog.settings import AUTHENTICATION_BACKENDS, DEBUG, DEMO, MIDDLEWARE, SITE_URL
-from posthog.settings.utils import get_from_env, get_list
+from posthog.settings.utils import get_from_env
 from posthog.utils import str_to_bool
 
 # SSO
@@ -63,14 +63,6 @@
 CUSTOMER_IO_API_KEY = get_from_env("CUSTOMER_IO_API_KEY", "", type_cast=str)
 CUSTOMER_IO_API_URL = get_from_env("CUSTOMER_IO_API_URL", "https://api-eu.customer.io", type_cast=str)
 
-# Workflows sends PostHog's own emails with POSTHOG_TRANSACTIONAL_EMAIL_API_KEY at POSTHOG_HOST, the same settings as
-# posthog/transactional_email.py. Every region sends through the US project, which the key picks.
-# Comma-separated `template_name=message_template_id` pairs. They override WORKFLOWS_TEMPLATE_ID_MAP in
-# posthog/email.py, so a local or staging instance can send with its own copies of the templates.
-WORKFLOWS_TRANSACTIONAL_EMAIL_TEMPLATE_IDS: dict[str, str] = dict(
-    pair.split("=", 1) for pair in get_list(os.getenv("WORKFLOWS_TRANSACTIONAL_EMAIL_TEMPLATE_IDS", "")) if "=" in pair
-)
-
 # Schedule to run column materialization on. Follows crontab syntax.
 # Use empty string to prevent from materializing
 MATERIALIZE_COLUMNS_SCHEDULE_CRON = get_from_env("MATERIALIZE_COLUMNS_SCHEDULE_CRON", "0 5 * * SAT")
diff --git a/posthog/email.py b/posthog/email.py
index 1e01a04cbbf0..a3a3e6a0aa27 100644
--- a/posthog/email.py
+++ b/posthog/email.py
@@ -246,9 +246,7 @@ def _workflows_error_allows_fallback(err: WorkflowsEmailError) -> bool:
 
 def get_workflows_template_id(template_name: str, dest: dict[str, str]) -> str | None:
     """The message template to send this email with through Workflows, or None to send it through Customer.io."""
-    template_id = {**WORKFLOWS_TEMPLATE_ID_MAP, **settings.WORKFLOWS_TRANSACTIONAL_EMAIL_TEMPLATE_IDS}.get(
-        template_name
-    )
+    template_id = WORKFLOWS_TEMPLATE_ID_MAP.get(template_name)
     if not template_id or not settings.POSTHOG_TRANSACTIONAL_EMAIL_API_KEY:
         return None
     try:
diff --git a/posthog/test/test_email.py b/posthog/test/test_email.py
index 34fe3795560c..a33875357cc1 100644
--- a/posthog/test/test_email.py
+++ b/posthog/test/test_email.py
@@ -471,8 +471,8 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
                 CUSTOMER_IO_API_KEY="cio-key",
                 POSTHOG_TRANSACTIONAL_EMAIL_API_KEY="phs_test",
                 POSTHOG_HOST="https://us.posthog.com",
-                WORKFLOWS_TRANSACTIONAL_EMAIL_TEMPLATE_IDS={"2fa_enabled": "tmpl-1"},
             ),
+            patch.dict("posthog.email.WORKFLOWS_TEMPLATE_ID_MAP", {"2fa_enabled": "tmpl-1"}),
         ):
             message = EmailMessage(
                 campaign_key="routing_campaign",

From df9900e6da820669c84bb467b217689b7a963fd7 Mon Sep 17 00:00:00 2001
From: Harley Alexander <43975092+mayteio@users.noreply.github.com>
Date: Thu, 22 Oct 2026 16:10:37 +0100
Subject: [PATCH 6/9] fix(email): fall back only when workflows cannot have
 queued the email

- A 5xx no longer falls back to Customer.io. The send API also returns 503 when the workflow service queued the email but failed to answer.
- A connection error falls back only when no connection was made. A reset after the request went out can follow a queued email.
- 408 and 422 no longer fall back either. Redirects do, because they come before anything is queued.
- An uncertain Workflows outcome now reaches Celery, which retries with the same idempotency key. Workflows answers a retry with the queued email instead of sending again, and recipients already sent are skipped.
- Send events go through a scoped client that flushes before the Celery task ends.
- Workflows sends refuse redirects and need an HTTPS POSTHOG_HOST outside DEBUG, because the request carries the secret key and the email's contents.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Generated-By: PostHog Desktop
Task-Id: b07b2dac-4768-46c8-8930-75b8c8aecc31
---
 posthog/email.py           | 59 ++++++++++++++++++++++++++++++++------
 posthog/test/test_email.py | 38 ++++++++++++++++++++----
 2 files changed, 82 insertions(+), 15 deletions(-)

diff --git a/posthog/email.py b/posthog/email.py
index a3a3e6a0aa27..0d55e0abfe25 100644
--- a/posthog/email.py
+++ b/posthog/email.py
@@ -19,6 +19,7 @@
 from django.utils import timezone
 from django.utils.module_loading import import_string
 
+import urllib3
 import requests
 import structlog
 import css_inline
@@ -32,6 +33,8 @@
 from posthog.helpers.email_utils import sanitize_display_name, sanitize_email_string
 from posthog.models.instance_setting import get_instance_setting
 from posthog.models.messaging import MessagingRecord, get_email_hashes
+from posthog.ph_client import ScopedCapture, ph_scoped_capture
+from posthog.utils import get_instance_region
 
 logger = structlog.get_logger(__name__)
 
@@ -237,11 +240,24 @@ def __init__(self, message: str, *, code: str, status_code: int | None = None) -
 def _workflows_error_allows_fallback(err: WorkflowsEmailError) -> bool:
     """Whether Customer.io can send the email without the risk of a duplicate."""
     if err.status_code is None:
-        # A refused connection never reached Workflows, and a client that queued nothing sent nothing. After a
-        # timeout, Workflows can have queued the email.
+        # A connection that was never made and a client that queued nothing sent nothing. Any other network
+        # failure can come after Workflows queued the email.
         return err.code in ("connection_error", "not_queued")
-    # A 409 means an earlier attempt with the same idempotency key is still in flight and can still send.
-    return err.status_code != 409
+    # Redirects and 4xx answers come before anything is queued, apart from 408 and the two idempotency answers:
+    # 409 (an attempt with this key is in flight) and 422 (this key already queued an email). A 5xx can follow a
+    # queued email, because the send API also returns 503 when the workflow service queued it but failed to answer.
+    return 300 <= err.status_code < 500 and err.status_code not in (408, 409, 422)
+
+
+def _never_connected(err: requests.ConnectionError) -> bool:
+    """Whether the request failed before a connection existed, so Workflows cannot have received it."""
+    if isinstance(err, requests.ConnectTimeout):
+        return True
+    reason = err.args[0] if err.args else None
+    if isinstance(reason, urllib3.exceptions.MaxRetryError):
+        reason = reason.reason
+    # A reset after the request went out arrives as a ProtocolError instead.
+    return isinstance(reason, urllib3.exceptions.NewConnectionError)
 
 
 def get_workflows_template_id(template_name: str, dest: dict[str, str]) -> str | None:
@@ -249,6 +265,10 @@ def get_workflows_template_id(template_name: str, dest: dict[str, str]) -> str |
     template_id = WORKFLOWS_TEMPLATE_ID_MAP.get(template_name)
     if not template_id or not settings.POSTHOG_TRANSACTIONAL_EMAIL_API_KEY:
         return None
+    if not settings.POSTHOG_HOST.startswith("https://") and not settings.DEBUG:
+        # The request carries the secret key and the email's contents, such as a password reset link.
+        logger.warning("email_workflows_insecure_host", template_name=template_name)
+        return None
     try:
         enabled = posthoganalytics.feature_enabled(
             WORKFLOWS_TRANSACTIONAL_EMAIL_FLAG,
@@ -306,9 +326,11 @@ def _queue_workflows_email(
             json=payload,
             # The caller holds the MessagingRecord row lock for the whole request, so keep it short.
             timeout=10,
+            allow_redirects=False,
         )
     except requests.ConnectionError as err:
-        raise WorkflowsEmailError(str(err), code="connection_error") from err
+        code = "connection_error" if _never_connected(err) else "network_error"
+        raise WorkflowsEmailError(str(err), code=code) from err
     except requests.Timeout as err:
         raise WorkflowsEmailError(str(err), code="timeout") from err
 
@@ -379,9 +401,9 @@ def _send_one_via_customer_io(dest: dict[str, str], template_name: str, properti
     return response.json()
 
 
-def _capture_email_event(dest: dict[str, str], event: str, properties: dict[str, Any]) -> None:
+def _capture_email_event(capture: ScopedCapture, dest: dict[str, str], event: str, properties: dict[str, Any]) -> None:
     try:
-        posthoganalytics.capture(
+        capture(
             distinct_id=dest.get("distinct_id") or dest["raw_email"],
             event=event,
             properties={"recipient_email": dest["raw_email"], **properties},
@@ -390,7 +412,9 @@ def _capture_email_event(dest: dict[str, str], event: str, properties: dict[str,
         logger.warning("email_send_analytics_capture_failed", error=str(analytics_err))
 
 
-def _send_one(dest: dict[str, str], campaign_key: str, template_name: str, properties: dict) -> tuple[str, dict]:
+def _send_one(
+    capture: ScopedCapture, dest: dict[str, str], campaign_key: str, template_name: str, properties: dict
+) -> tuple[str, dict]:
     """Send one email and return the provider that sent it with the provider's response."""
     workflows_template_id = get_workflows_template_id(template_name, dest)
     if workflows_template_id:
@@ -402,6 +426,7 @@ def _send_one(dest: dict[str, str], campaign_key: str, template_name: str, prope
             logger.warning("email_workflows_send_rejected", template_name=template_name, error=str(err))
             EMAIL_WORKFLOWS_FALLBACK_COUNTER.labels(template_name=template_name).inc()
             _capture_email_event(
+                capture,
                 dest,
                 "transactional email workflows rejected",
                 {"template_name": template_name, "campaign_key": campaign_key, "error": str(err)[:500]},
@@ -421,6 +446,14 @@ def _send_via_http(
     if not is_http_email_service_available():
         raise Exception("Missing Customer.io API key")
 
+    # A dedicated client, because the global one may never flush before a Celery worker exits.
+    with ph_scoped_capture(event_region=get_instance_region()) as capture:
+        _send_via_http_batch(capture, to, campaign_key, template_name, properties)
+
+
+def _send_via_http_batch(
+    capture: ScopedCapture, to: list[dict[str, str]], campaign_key: str, template_name: str, properties: dict
+) -> None:
     sent_count = 0
     already_sent_count = 0  # delivered in a prior run — not a failure if the batch later aborts
     dest: dict[str, str] | None = None
@@ -436,7 +469,7 @@ def _send_via_http(
                     already_sent_count += 1
                     continue
 
-                provider, provider_response = _send_one(dest, campaign_key, template_name, properties)
+                provider, provider_response = _send_one(capture, dest, campaign_key, template_name, properties)
 
                 # Mark delivery before any non-essential step so a delivered email is never
                 # recorded as rejected (raise_if_delivery_rejected keys off sent_at). Analytics
@@ -445,6 +478,7 @@ def _send_via_http(
                 record.save()
 
                 _capture_email_event(
+                    capture,
                     dest,
                     "transactional email triggered",
                     {
@@ -461,8 +495,10 @@ def _send_via_http(
     except Exception as err:
         capture_exception(err)  # already logs the traceback via logger.exception
         # The exception carries no template name, so this event is what lets a dashboard count failures per template.
+        retrying = isinstance(err, WorkflowsEmailError)
         if dest:
             _capture_email_event(
+                capture,
                 dest,
                 "transactional email failed",
                 {
@@ -470,11 +506,16 @@ def _send_via_http(
                     "campaign_key": campaign_key,
                     "error_type": type(err).__name__,
                     "error": str(err)[:500],
+                    "retrying": retrying,
                 },
             )
         # Count every recipient that did not get through (the failing one + any not yet attempted),
         # so `failed` shares the per-recipient unit with `sent` instead of registering once per batch.
         EMAIL_SEND_COUNTER.labels(outcome="failed", transport="http").inc(len(to) - already_sent_count - sent_count)
+        if retrying:
+            # Workflows may have queued this email. The Celery retry sends the same idempotency key, so Workflows
+            # returns the queued email instead of sending a second one, and recipients already sent are skipped.
+            raise
 
 
 def _send_via_smtp(
diff --git a/posthog/test/test_email.py b/posthog/test/test_email.py
index a33875357cc1..0f487dc1af1e 100644
--- a/posthog/test/test_email.py
+++ b/posthog/test/test_email.py
@@ -17,6 +17,7 @@
 from django.test import override_settings
 from django.utils import timezone
 
+import urllib3
 import requests
 from parameterized import parameterized
 from prometheus_client import REGISTRY
@@ -24,6 +25,7 @@
 from posthog.email import (
     CUSTOMER_IO_TEMPLATE_ID_MAP,
     EmailMessage,
+    WorkflowsEmailError,
     _send_email,
     _send_via_http,
     _send_via_smtp,
@@ -303,9 +305,10 @@ def test_send_via_http_failed_metric_counts_undelivered_recipients(self, mock_po
         after = REGISTRY.get_sample_value("posthog_email_send_total", failed_labels) or 0.0
         self.assertEqual(after - before, 2.0)  # b failed + c never attempted; a succeeded
 
-    @patch("posthoganalytics.capture")
+    @patch("posthog.email.ph_scoped_capture")
     @patch("posthog.email.requests.post")
-    def test_send_via_http_success(self, mock_post, mock_capture) -> None:
+    def test_send_via_http_success(self, mock_post, mock_scoped_capture) -> None:
+        mock_capture = mock_scoped_capture.return_value.__enter__.return_value
         mock_response = MagicMock()
         mock_response.status_code = 200
         mock_response.json.return_value = {"delivery_id": "test_delivery_id", "queued_at": 1604977406}
@@ -413,14 +416,30 @@ def test_send_via_http_api_error(self, mock_post) -> None:
                 ["transactional email workflows rejected", "transactional email triggered"],
             ),
             ("workflows_in_flight_sends_nothing", True, [409], ["workflows"], None, ["transactional email failed"]),
+            ("workflows_server_error_sends_nothing", True, [503], ["workflows"], None, ["transactional email failed"]),
             (
                 "workflows_unreachable_falls_back",
                 True,
-                [requests.ConnectionError("refused"), 200],
+                [
+                    requests.ConnectionError(
+                        urllib3.exceptions.MaxRetryError(
+                            None, "/", urllib3.exceptions.NewConnectionError(None, "refused")
+                        )
+                    ),
+                    200,
+                ],
                 ["workflows", "customer_io"],
                 "customer_io",
                 ["transactional email workflows rejected", "transactional email triggered"],
             ),
+            (
+                "workflows_reset_after_send_sends_nothing",
+                True,
+                [requests.ConnectionError(urllib3.exceptions.ProtocolError("Connection aborted."))],
+                ["workflows"],
+                None,
+                ["transactional email failed"],
+            ),
             (
                 "workflows_timeout_sends_nothing",
                 True,
@@ -439,7 +458,7 @@ def test_send_via_http_api_error(self, mock_post) -> None:
             ),
         ]
     )
-    @patch("posthoganalytics.capture")
+    @patch("posthog.email.ph_scoped_capture")
     @patch("posthog.email.posthoganalytics.feature_enabled")
     @patch("posthog.email.requests.post")
     def test_send_via_http_routes_through_workflows_when_flag_matches(
@@ -452,8 +471,9 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
         expected_events: list[str],
         mock_post: MagicMock,
         mock_feature_enabled: MagicMock,
-        mock_capture: MagicMock,
+        mock_scoped_capture: MagicMock,
     ) -> None:
+        mock_capture = mock_scoped_capture.return_value.__enter__.return_value
         mock_feature_enabled.return_value = flag_enabled
         responses: list[MagicMock | Exception] = []
         for status_code in status_codes:
@@ -482,7 +502,12 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
                 use_http=True,
             )
             message.add_recipient("test@posthog.com", "Test", distinct_id="user-1")
-            message.send(send_async=False)
+            if sent_by is None:
+                # Workflows may have queued the email, so the error reaches Celery, which retries with the same key.
+                with self.assertRaises(WorkflowsEmailError):
+                    message.send(send_async=False, retry=False)
+            else:
+                message.send(send_async=False, retry=False)
 
         providers = [
             "workflows" if call.args[0].startswith("https://us.posthog.com") else "customer_io"
@@ -503,6 +528,7 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
             )
             self.assertEqual(workflows_call.kwargs["headers"]["Authorization"], "Bearer phs_test")
             self.assertIn("Idempotency-Key", workflows_call.kwargs["headers"])
+            self.assertFalse(workflows_call.kwargs["allow_redirects"])
             payload = workflows_call.kwargs["json"]
             self.assertEqual(payload["identifiers"], {"distinct_id": "user-1"})
             # Workflows escapes Liquid output itself, so values must arrive unescaped or they render as &amp;amp;

From 74b4ceeff434303e1767017192bdb98088c14f21 Mon Sep 17 00:00:00 2001
From: Harley Alexander <43975092+mayteio@users.noreply.github.com>
Date: Thu, 22 Oct 2026 16:32:50 +0100
Subject: [PATCH 7/9] chore(email): pass a real connection pool in the
 unreachable workflows test case

mypy rejects `None` as the pool argument of `urllib3.exceptions.MaxRetryError`.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Generated-By: PostHog Desktop
Task-Id: b07b2dac-4768-46c8-8930-75b8c8aecc31
---
 posthog/test/test_email.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/posthog/test/test_email.py b/posthog/test/test_email.py
index 0f487dc1af1e..dac736f39a3e 100644
--- a/posthog/test/test_email.py
+++ b/posthog/test/test_email.py
@@ -423,7 +423,9 @@ def test_send_via_http_api_error(self, mock_post) -> None:
                 [
                     requests.ConnectionError(
                         urllib3.exceptions.MaxRetryError(
-                            None, "/", urllib3.exceptions.NewConnectionError(None, "refused")
+                            urllib3.HTTPConnectionPool("us.posthog.com"),
+                            "/",
+                            urllib3.exceptions.NewConnectionError(None, "refused"),
                         )
                     ),
                     200,

From b5b85ef2b781ac7a8985d378783a567ce4b4b06e Mon Sep 17 00:00:00 2001
From: Harley Alexander <43975092+mayteio@users.noreply.github.com>
Date: Sat, 24 Oct 2026 09:24:23 +0100
Subject: [PATCH 8/9] fix(email): keep a retried workflows send off customer.io

- Pin a send to Workflows after an attempt that may have queued it, so a retry skips the flag and never falls back.
- Treat a 202 without a message_id as possibly queued instead of falling back.
- Synchronous sends make one Workflows attempt and hand an uncertain one to a worker, instead of eager retries inside the request.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Generated-By: PostHog Desktop
Task-Id: a5c9c2a4-4563-4d98-a4b6-ef13d86f4473
---
 posthog/email.py           |  82 +++++++++++++++++++++++-----
 posthog/test/test_email.py | 109 ++++++++++++++++++++++++++++++++++++-
 2 files changed, 176 insertions(+), 15 deletions(-)

diff --git a/posthog/email.py b/posthog/email.py
index 0d55e0abfe25..eb5c7f2345ac 100644
--- a/posthog/email.py
+++ b/posthog/email.py
@@ -13,6 +13,7 @@
 
 from django.conf import settings
 from django.core import exceptions, mail
+from django.core.cache import cache
 from django.core.mail.backends.smtp import EmailBackend
 from django.db import transaction
 from django.template.loader import get_template
@@ -178,6 +179,9 @@ def is_email_available(with_absolute_urls: bool = False) -> bool:
 # so one flag picks both the recipients and the templates that move.
 WORKFLOWS_TRANSACTIONAL_EMAIL_FLAG = "transactional-emails-via-workflows"
 
+# Matches how long the send API keeps an idempotency key, which outlasts every Celery retry.
+WORKFLOWS_PIN_TTL_SECONDS = 24 * 60 * 60
+
 EMAIL_WORKFLOWS_FALLBACK_COUNTER = Counter(
     "posthog_email_workflows_fallback_total",
     "Emails that Workflows refused and Customer.io sent instead, by template.",
@@ -342,8 +346,35 @@ def _queue_workflows_email(
         raise WorkflowsEmailError(
             f"{response.status_code} - {response.text}", code=code, status_code=response.status_code
         )
-    body = response.json()
-    return {"message_id": body.get("message_id"), "replayed": bool(body.get("replayed", False))}
+    try:
+        body = response.json()
+    except ValueError:
+        body = {}
+    if not body.get("message_id"):
+        # A 202 means Workflows may have queued the email, so this must not fall back to Customer.io.
+        raise WorkflowsEmailError("202 without a message_id", code="invalid_response", status_code=response.status_code)
+    return {"message_id": body["message_id"], "replayed": bool(body.get("replayed", False))}
+
+
+def _workflows_idempotency_key(campaign_key: str, raw_email: str) -> str:
+    # Derived from the email itself, so a Celery retry of the whole send cannot queue it twice.
+    return hashlib.sha256(f"{campaign_key}\n{raw_email}".encode()).hexdigest()
+
+
+def _pinned_workflows_template_id(idempotency_key: str) -> str | None:
+    """The template of an earlier attempt that Workflows may have queued, so a retry sends it the same way."""
+    try:
+        return cache.get(f"email_workflows_pin:{idempotency_key}")
+    except Exception as err:
+        logger.warning("email_workflows_pin_read_failed", error=str(err))
+        return None
+
+
+def _pin_to_workflows(idempotency_key: str, template_id: str) -> None:
+    try:
+        cache.set(f"email_workflows_pin:{idempotency_key}", template_id, timeout=WORKFLOWS_PIN_TTL_SECONDS)
+    except Exception as err:
+        logger.warning("email_workflows_pin_write_failed", error=str(err))
 
 
 def _send_one_via_workflows(
@@ -361,8 +392,7 @@ def _send_one_via_workflows(
         template_id=template_id,
         template_data=_unescape_for_liquid({**properties, "customer": customer}),
         distinct_id=dest.get("distinct_id") or None,
-        # Derived from the email itself, so a Celery retry of the whole send cannot queue it twice.
-        idempotency_key=hashlib.sha256(f"{campaign_key}\n{dest['raw_email']}".encode()).hexdigest(),
+        idempotency_key=_workflows_idempotency_key(campaign_key, dest["raw_email"]),
     )
     if not result["message_id"]:
         # A disabled SDK client queues nothing and returns no ID.
@@ -416,12 +446,17 @@ def _send_one(
     capture: ScopedCapture, dest: dict[str, str], campaign_key: str, template_name: str, properties: dict
 ) -> tuple[str, dict]:
     """Send one email and return the provider that sent it with the provider's response."""
-    workflows_template_id = get_workflows_template_id(template_name, dest)
+    idempotency_key = _workflows_idempotency_key(campaign_key, dest["raw_email"])
+    # After an attempt Workflows may have queued, a retry skips the flag and never falls back. Otherwise a flag
+    # change or a different error on the retry would send the same email again through Customer.io.
+    pinned_template_id = _pinned_workflows_template_id(idempotency_key)
+    workflows_template_id = pinned_template_id or get_workflows_template_id(template_name, dest)
     if workflows_template_id:
         try:
             return "workflows", _send_one_via_workflows(dest, campaign_key, workflows_template_id, properties)
         except WorkflowsEmailError as err:
-            if not _workflows_error_allows_fallback(err):
+            if pinned_template_id or not _workflows_error_allows_fallback(err):
+                _pin_to_workflows(idempotency_key, workflows_template_id)
                 raise
             logger.warning("email_workflows_send_rejected", template_name=template_name, error=str(err))
             EMAIL_WORKFLOWS_FALLBACK_COUNTER.labels(template_name=template_name).inc()
@@ -749,14 +784,34 @@ def _send_email_now(
     reply_to: Optional[str] = None,
     use_http: Optional[bool] = False,
     properties: Optional[dict] = None,
+    retry_in_worker: bool = False,
 ) -> None:
     if use_http and is_http_email_service_available():
-        _send_via_http(
-            to=to,
-            campaign_key=campaign_key,
-            template_name=template_name,
-            properties=properties or {},
-        )
+        try:
+            _send_via_http(
+                to=to,
+                campaign_key=campaign_key,
+                template_name=template_name,
+                properties=properties or {},
+            )
+        except WorkflowsEmailError:
+            if not retry_in_worker:
+                raise
+            # The retry sends the same idempotency key, so Workflows does not send the email twice.
+            _send_email.apply_async(
+                kwargs={
+                    "campaign_key": campaign_key,
+                    "to": to,
+                    "subject": subject,
+                    "headers": headers,
+                    "txt_body": txt_body,
+                    "html_body": html_body,
+                    "template_name": template_name,
+                    "reply_to": reply_to,
+                    "use_http": use_http,
+                    "properties": properties,
+                }
+            )
     elif is_smtp_email_service_available():
         _send_via_smtp(
             to=to,
@@ -848,7 +903,8 @@ def send(self, send_async: bool = True, *, retry: bool = True) -> None:
         if send_async:
             _send_email.apply_async(kwargs=kwargs)
         elif retry:
-            _send_email.apply(kwargs=kwargs)
+            # Celery runs eager retries at once, so each Workflows timeout would hold the request for another 10s.
+            _send_email.apply(kwargs={**kwargs, "retry_in_worker": True})
         else:
             _send_email_now(
                 campaign_key=self.campaign_key,
diff --git a/posthog/test/test_email.py b/posthog/test/test_email.py
index dac736f39a3e..c515705ac180 100644
--- a/posthog/test/test_email.py
+++ b/posthog/test/test_email.py
@@ -13,6 +13,7 @@
 
 from django.conf import settings
 from django.core import mail
+from django.core.cache import cache
 from django.core.exceptions import ImproperlyConfigured
 from django.test import override_settings
 from django.utils import timezone
@@ -91,6 +92,7 @@ def create_person(self, team: Team, base_distinct_id: str = "") -> Person:
     @time_machine.travel("2020-09-21", tick=False)
     def setUp(self):
         super().setUp()
+        cache.clear()
         self.organization = Organization.objects.create()
         self.team = Team.objects.create(organization=self.organization, name="The Bakery")
         self.user = User.objects.create(email="test@posthog.com")
@@ -417,6 +419,14 @@ def test_send_via_http_api_error(self, mock_post) -> None:
             ),
             ("workflows_in_flight_sends_nothing", True, [409], ["workflows"], None, ["transactional email failed"]),
             ("workflows_server_error_sends_nothing", True, [503], ["workflows"], None, ["transactional email failed"]),
+            (
+                "workflows_accepts_without_message_id_sends_nothing",
+                True,
+                [(202, {})],
+                ["workflows"],
+                None,
+                ["transactional email failed"],
+            ),
             (
                 "workflows_unreachable_falls_back",
                 True,
@@ -467,7 +477,7 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
         self,
         _name: str,
         flag_enabled: bool,
-        status_codes: list[int | Exception],
+        status_codes: list[int | tuple[int, dict] | Exception],
         expected_providers: list[str],
         sent_by: str | None,
         expected_events: list[str],
@@ -482,8 +492,12 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
             if isinstance(status_code, Exception):
                 responses.append(status_code)
                 continue
+            if isinstance(status_code, tuple):
+                status_code, body = status_code
+            else:
+                body = {"message_id": "m1"} if status_code == 202 else {"delivery_id": "d1"}
             response = MagicMock(status_code=status_code, text="error")
-            response.json.return_value = {"message_id": "m1"} if status_code == 202 else {"delivery_id": "d1"}
+            response.json.return_value = body
             responses.append(response)
         mock_post.side_effect = responses
 
@@ -553,6 +567,97 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
             self.assertIsNotNone(record.sent_at)
             self.assertEqual(mock_capture.call_args.kwargs["properties"]["provider"], sent_by)
 
+    @parameterized.expand(
+        [
+            ("flag_turned_off", False, MagicMock(status_code=202, json=MagicMock(return_value={"message_id": "m1"}))),
+            (
+                "workflows_unreachable",
+                True,
+                requests.ConnectionError(
+                    urllib3.exceptions.MaxRetryError(
+                        urllib3.HTTPConnectionPool("us.posthog.com"),
+                        "/",
+                        urllib3.exceptions.NewConnectionError(None, "refused"),
+                    )
+                ),
+            ),
+        ]
+    )
+    @patch("posthog.email.ph_scoped_capture")
+    @patch("posthog.email.posthoganalytics.feature_enabled")
+    @patch("posthog.email.requests.post")
+    def test_retry_after_an_uncertain_workflows_send_never_uses_customer_io(
+        self,
+        _name: str,
+        flag_on_retry: bool,
+        retry_response: MagicMock | Exception,
+        mock_post: MagicMock,
+        mock_feature_enabled: MagicMock,
+        _mock_capture: MagicMock,
+    ) -> None:
+        mock_feature_enabled.side_effect = [True, flag_on_retry]
+        mock_post.side_effect = [requests.ReadTimeout("no response"), retry_response]
+
+        with (
+            override_instance_config("EMAIL_HOST", "localhost"),
+            self.settings(
+                CUSTOMER_IO_API_KEY="cio-key",
+                POSTHOG_TRANSACTIONAL_EMAIL_API_KEY="phs_test",
+                POSTHOG_HOST="https://us.posthog.com",
+            ),
+            patch.dict("posthog.email.WORKFLOWS_TEMPLATE_ID_MAP", {"password_reset": "tmpl-1"}),
+        ):
+            message = EmailMessage(
+                campaign_key="pinned_campaign", subject="Test subject", template_name="password_reset", use_http=True
+            )
+            message.add_recipient("test@posthog.com")
+            with self.assertRaises(WorkflowsEmailError):
+                message.send(send_async=False, retry=False)
+            try:
+                message.send(send_async=False, retry=False)
+            except WorkflowsEmailError:
+                pass
+
+        # The first attempt may have queued the email, so Customer.io sending the retry would deliver it twice.
+        self.assertEqual(
+            [call.args[0] for call in mock_post.call_args_list],
+            ["https://us.posthog.com/api/projects/@current/messaging_templates/tmpl-1/send/"] * 2,
+        )
+
+    @patch.object(_send_email, "apply_async")
+    @patch("posthog.email.ph_scoped_capture")
+    @patch("posthog.email.posthoganalytics.feature_enabled", return_value=True)
+    @patch("posthog.email.requests.post", side_effect=requests.ReadTimeout("no response"))
+    def test_inline_send_hands_an_uncertain_workflows_send_to_a_worker(
+        self,
+        mock_post: MagicMock,
+        _mock_feature_enabled: MagicMock,
+        _mock_capture: MagicMock,
+        mock_apply_async: MagicMock,
+    ) -> None:
+        with (
+            override_instance_config("EMAIL_HOST", "localhost"),
+            self.settings(
+                CUSTOMER_IO_API_KEY="cio-key",
+                POSTHOG_TRANSACTIONAL_EMAIL_API_KEY="phs_test",
+                POSTHOG_HOST="https://us.posthog.com",
+            ),
+            patch.dict("posthog.email.WORKFLOWS_TEMPLATE_ID_MAP", {"password_reset": "tmpl-1"}),
+        ):
+            message = EmailMessage(
+                campaign_key="inline_campaign", subject="Test subject", template_name="password_reset", use_http=True
+            )
+            message.add_recipient("test@posthog.com")
+            message.send(send_async=False)
+
+        # Eager retries would post once per retry inside the request, each up to the 10s timeout.
+        self.assertEqual(mock_post.call_count, 1)
+        mock_apply_async.assert_called_once()
+        retry_kwargs = mock_apply_async.call_args.kwargs["kwargs"]
+        self.assertEqual(retry_kwargs["campaign_key"], "inline_campaign")
+        # A worker still on the previous release rejects an argument it does not know.
+        self.assertNotIn("retry_in_worker", retry_kwargs)
+
     def test_sanitize_email_properties(self) -> None:
         # Test with various types of input including potential HTML injection
         properties = {

From 1c4b51e4607cd622fac08a9d02f705f5a66e8bdd Mon Sep 17 00:00:00 2001
From: Harley Alexander <43975092+mayteio@users.noreply.github.com>
Date: Sat, 24 Oct 2026 09:38:32 +0100
Subject: [PATCH 9/9] fix(email): keep the settings checks on a pinned
 workflows retry

A pinned retry now skips only the flag. Without the transactional email key or an https host it sends through Customer.io, so a rollback by settings cannot post the email to an unconfigured or insecure host.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Generated-By: PostHog Desktop
Task-Id: a5c9c2a4-4563-4d98-a4b6-ef13d86f4473
---
 posthog/email.py           | 15 ++++++++++----
 posthog/test/test_email.py | 40 ++++++++++++++++++++++++++++----------
 2 files changed, 41 insertions(+), 14 deletions(-)

diff --git a/posthog/email.py b/posthog/email.py
index eb5c7f2345ac..0beaa96d52ec 100644
--- a/posthog/email.py
+++ b/posthog/email.py
@@ -264,15 +264,22 @@ def _never_connected(err: requests.ConnectionError) -> bool:
     return isinstance(reason, urllib3.exceptions.NewConnectionError)
 
 
-def get_workflows_template_id(template_name: str, dest: dict[str, str]) -> str | None:
-    """The message template to send this email with through Workflows, or None to send it through Customer.io."""
-    template_id = WORKFLOWS_TEMPLATE_ID_MAP.get(template_name)
+def get_workflows_template_id(
+    template_name: str, dest: dict[str, str], pinned_template_id: str | None = None
+) -> str | None:
+    """The message template to send this email with through Workflows, or None to send it through Customer.io.
+
+    A pinned template skips only the flag. The settings checks still apply, because they turn Workflows off.
+    """
+    template_id = pinned_template_id or WORKFLOWS_TEMPLATE_ID_MAP.get(template_name)
     if not template_id or not settings.POSTHOG_TRANSACTIONAL_EMAIL_API_KEY:
         return None
     if not settings.POSTHOG_HOST.startswith("https://") and not settings.DEBUG:
         # The request carries the secret key and the email's contents, such as a password reset link.
         logger.warning("email_workflows_insecure_host", template_name=template_name)
         return None
+    if pinned_template_id:
+        return pinned_template_id
     try:
         enabled = posthoganalytics.feature_enabled(
             WORKFLOWS_TRANSACTIONAL_EMAIL_FLAG,
@@ -450,7 +457,7 @@ def _send_one(
     # After an attempt Workflows may have queued, a retry skips the flag and never falls back. Otherwise a flag
     # change or a different error on the retry would send the same email again through Customer.io.
     pinned_template_id = _pinned_workflows_template_id(idempotency_key)
-    workflows_template_id = pinned_template_id or get_workflows_template_id(template_name, dest)
+    workflows_template_id = get_workflows_template_id(template_name, dest, pinned_template_id)
     if workflows_template_id:
         try:
             return "workflows", _send_one_via_workflows(dest, campaign_key, workflows_template_id, properties)
diff --git a/posthog/test/test_email.py b/posthog/test/test_email.py
index c515705ac180..c17d1891f063 100644
--- a/posthog/test/test_email.py
+++ b/posthog/test/test_email.py
@@ -569,10 +569,17 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
 
     @parameterized.expand(
         [
-            ("flag_turned_off", False, MagicMock(status_code=202, json=MagicMock(return_value={"message_id": "m1"}))),
+            (
+                "flag_turned_off",
+                False,
+                "phs_test",
+                MagicMock(status_code=202, json=MagicMock(return_value={"message_id": "m1"})),
+                ["workflows", "workflows"],
+            ),
             (
                 "workflows_unreachable",
                 True,
+                "phs_test",
                 requests.ConnectionError(
                     urllib3.exceptions.MaxRetryError(
                         urllib3.HTTPConnectionPool("us.posthog.com"),
@@ -580,17 +587,28 @@ def test_send_via_http_routes_through_workflows_when_flag_matches(
                         urllib3.exceptions.NewConnectionError(None, "refused"),
                     )
                 ),
+                ["workflows", "workflows"],
+            ),
+            # Removing the key turns Workflows off, so a pinned retry must not post the email without it.
+            (
+                "api_key_removed",
+                True,
+                "",
+                MagicMock(status_code=200, json=MagicMock(return_value={"delivery_id": "d1"})),
+                ["workflows", "customer_io"],
             ),
         ]
     )
     @patch("posthog.email.ph_scoped_capture")
     @patch("posthog.email.posthoganalytics.feature_enabled")
     @patch("posthog.email.requests.post")
-    def test_retry_after_an_uncertain_workflows_send_never_uses_customer_io(
+    def test_retry_after_an_uncertain_workflows_send_stays_on_workflows_while_it_is_configured(
         self,
         _name: str,
         flag_on_retry: bool,
+        api_key_on_retry: str,
         retry_response: MagicMock | Exception,
+        expected_providers: list[str],
         mock_post: MagicMock,
         mock_feature_enabled: MagicMock,
         _mock_capture: MagicMock,
@@ -613,16 +631,18 @@ def test_retry_after_an_uncertain_workflows_send_never_uses_customer_io(
             message.add_recipient("test@posthog.com")
             with self.assertRaises(WorkflowsEmailError):
                 message.send(send_async=False, retry=False)
-            try:
-                message.send(send_async=False, retry=False)
-            except WorkflowsEmailError:
-                pass
+            with self.settings(POSTHOG_TRANSACTIONAL_EMAIL_API_KEY=api_key_on_retry):
+                try:
+                    message.send(send_async=False, retry=False)
+                except WorkflowsEmailError:
+                    pass
 
         # The first attempt may have queued the email, so Customer.io sending the retry would deliver it twice.
-        self.assertEqual(
-            [call.args[0] for call in mock_post.call_args_list],
-            ["https://us.posthog.com/api/projects/@current/messaging_templates/tmpl-1/send/"] * 2,
-        )
+        providers = [
+            "workflows" if call.args[0].startswith("https://us.posthog.com") else "customer_io"
+            for call in mock_post.call_args_list
+        ]
+        self.assertEqual(providers, expected_providers)
 
     @patch.object(_send_email, "apply_async")
     @patch("posthog.email.ph_scoped_capture")