From d500b2f387eafdf2862ed448a504fce423a729e8 Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Thu, 12 Nov 2026 11:37:56 +0000
Subject: [PATCH 1/6] feat(emails): schedule, move, cancel and list emails

send() takes scheduled_at, an aware datetime or an ISO 8601 string with an
offset, and refuses a naive datetime before any request. Emails and
AsyncEmails gain update(), cancel() and list(). Cancel and update retry like
a send; list never retries.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .github/scripts/check_strict_types.sh      |  26 +-
 .sampo/changesets/scheduled-email-sends.md |   5 +
 posthog/emails.py                          | 445 ++++++++++++++++++++-
 posthog/test/test_emails.py                | 415 ++++++++++++++++++-
 references/public_api_snapshot.txt         |  25 +-
 5 files changed, 904 insertions(+), 12 deletions(-)
 create mode 100644 .sampo/changesets/scheduled-email-sends.md

diff --git a/.github/scripts/check_strict_types.sh b/.github/scripts/check_strict_types.sh
index 4e41c9e8f..13cc955b5 100755
--- a/.github/scripts/check_strict_types.sh
+++ b/.github/scripts/check_strict_types.sh
@@ -15,7 +15,9 @@ import atexit
 
 import posthog
 from posthog import FeatureFlagEvaluations, FlagValue, Posthog
-from posthog.emails import AsyncEmails, Emails, SendEmailResult
+from datetime import datetime, timezone
+
+from posthog.emails import AsyncEmails, CancelEmailResult, EmailList, Emails, SendEmailResult, UpdateEmailResult
 from posthog.ai.evaluations import (
     AsyncOfflineEvaluations,
     BooleanScorerConfig,
@@ -44,12 +46,24 @@ _ = (flag_value, all_flags, enabled, payload, evaluations, active)
 emails = Emails("phs_example")
 template_send: SendEmailResult = emails.send(to="user@example.com", template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b", template_data={"name": "Ada"})
 rendered_send: SendEmailResult = emails.send(to="user@example.com", template_key="password-reset", subject="Reset", html="<p>Reset</p>", from_="Acme <security@example.com>", reply_to=["help@example.com"], capture_engagement=False, max_retries=0)
-_ = (template_send, rendered_send)
+scheduled_send: SendEmailResult = emails.send(to="user@example.com", template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b", scheduled_at=datetime(2026, 12, 1, tzinfo=timezone.utc))
+scheduled_status: str = scheduled_send.status
+scheduled_time: datetime | None = scheduled_send.scheduled_at
+moved: UpdateEmailResult = emails.update(scheduled_send.id, scheduled_at="2026-12-02T09:00:00+02:00")
+canceled: CancelEmailResult = emails.cancel(scheduled_send.id)
+page: EmailList = emails.list(cursor=None, limit=50)
+next_page: EmailList | None = emails.list(cursor=page.next) if page.next else None
+first_time: datetime = page.results[0].scheduled_at
+_ = (template_send, rendered_send, scheduled_status, scheduled_time, moved, canceled, next_page, first_time)
 
 
 async def async_email_types() -> None:
-    sent: SendEmailResult = await AsyncEmails("phs_example").send(to="user@example.com", template_key="welcome", subject="Hi", html="<p>Hi</p>")
-    _ = sent
+    async_emails = AsyncEmails("phs_example")
+    sent: SendEmailResult = await async_emails.send(to="user@example.com", template_key="welcome", subject="Hi", html="<p>Hi</p>", scheduled_at="2026-12-01T09:00:00Z")
+    moved: UpdateEmailResult = await async_emails.update(sent.id, scheduled_at=datetime(2026, 12, 2, tzinfo=timezone.utc))
+    canceled: CancelEmailResult = await async_emails.cancel(sent.id)
+    page: EmailList = await async_emails.list()
+    _ = (sent, moved, canceled, page)
 
 
 async def offline_evaluation_types(
@@ -91,6 +105,10 @@ Emails("phs_example").send(to="user@example.com", template_id="0190f6d2-7c1e-7b3
 Emails("phs_example").send(to="user@example.com", template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b", capture_engagement=False)  # expected-type-error
 Emails("phs_example").send(to="user@example.com", template_key="welcome", subject="Hi", html="<p>Hi</p>", template_data={"name": "Ada"})  # expected-type-error
 Emails("phs_example").send(to="user@example.com", template_key="welcome", subject="Hi", html="<p>Hi</p>", cc="team@example.com")  # expected-type-error
+Emails("phs_example").send(to="user@example.com", template_key="welcome", subject="Hi", html="<p>Hi</p>", scheduled_at=1796115600)  # expected-type-error
+Emails("phs_example").update("0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b")  # expected-type-error
+Emails("phs_example").update("0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b", "2026-12-01T09:00:00Z")  # expected-type-error
+Emails("phs_example").list("scheduled")  # expected-type-error
 
 
 async def invalid_async_email() -> None:
diff --git a/.sampo/changesets/scheduled-email-sends.md b/.sampo/changesets/scheduled-email-sends.md
new file mode 100644
index 000000000..c0170b3c2
--- /dev/null
+++ b/.sampo/changesets/scheduled-email-sends.md
@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Schedule a transactional email with `scheduled_at` on `emails.send()`, up to 30 days ahead, and move, cancel or list scheduled emails with `emails.update()`, `emails.cancel()` and `emails.list()`. A naive `datetime` raises `ValueError` before any request. Send results now carry `status` and `scheduled_at`.
diff --git a/posthog/emails.py b/posthog/emails.py
index 42e70b971..045a5d949 100644
--- a/posthog/emails.py
+++ b/posthog/emails.py
@@ -71,12 +71,29 @@
 - ``client_disabled``: the client is disabled, so it sends nothing.
 - ``client_closed``: the client is shut down.
 
-Status: ``emails.get(message_id)`` reads what happened to a sent email, for
+Status: ``emails.get(message_id)`` reads what happened to an email, for
 30 days after the send, as an :class:`EmailStatus`. A key with only
 ``transactional_email:read`` can read statuses but not send. ``get`` raises
 ``not_found`` (404) for an unknown ID, an email older than 30 days, or for a
 few seconds after the send.
 
+Scheduling: pass ``scheduled_at``, an aware ``datetime`` or an ISO 8601
+string with an offset, to send up to 30 days later. The result's ``status`` is
+``scheduled``. Until the email goes out, ``emails.list()`` finds it,
+``emails.update(id, scheduled_at=...)`` moves it, and ``emails.cancel(id)``
+stops it. A naive ``datetime`` raises ``ValueError`` before any request.
+
+```python
+from datetime import datetime, timedelta, timezone
+
+result = posthog.emails.send(
+    to="user@example.com",
+    template_id="<template UUID>",
+    scheduled_at=datetime.now(timezone.utc) + timedelta(days=1),
+)
+posthog.emails.cancel(result.id)
+```
+
 Cleanup: ``shutdown()`` on the client closes its email connections. Close a
 standalone :class:`Emails` with ``close()`` or a ``with`` block.
 
@@ -98,10 +115,11 @@
 import threading
 import time
 from dataclasses import dataclass
-from datetime import datetime
+from datetime import datetime, timezone
 from types import TracebackType
 from typing import (
     Any,
+    List,
     Callable,
     Mapping,
     Optional,
@@ -111,6 +129,7 @@
     Union,
     overload,
 )
+from urllib.parse import urlencode
 from uuid import UUID, uuid4
 
 import requests
@@ -128,24 +147,37 @@
 
 __all__ = [
     "AsyncEmails",
+    "CancelEmailResult",
     "CodeEmailArgs",
     "EmailError",
+    "EmailList",
     "EmailStatus",
     "EmailWarning",
     "Emails",
+    "ScheduledEmail",
     "SendEmailResult",
     "TemplateEmailArgs",
+    "UpdateEmailResult",
 ]
 
 log = logging.getLogger("posthog")
 
 _SEND_PATH = "/api/projects/@current/messaging/send/email/"
 _STATUS_PATH = "/api/projects/@current/messaging/emails/{message_id}/"
+_CANCEL_PATH = _STATUS_PATH + "cancel/"
+_LIST_PATH = "/api/projects/@current/messaging/emails/"
+_MAX_LIST_LIMIT = 100
+# An ISO 8601 date and time that ends in an offset, such as Z or +02:00. PostHog parses the rest.
+_TIME_WITH_OFFSET = re.compile(
+    r"\d{4}-\d{2}-\d{2}[Tt ]\d{2}:\d{2}(?::\d{2}(?:[.,]\d+)?)?(?:[Zz]|[+-]\d{2}(?::?\d{2})?)"
+)
 _TEMPLATE_KEY = re.compile(r"[a-zA-Z0-9_-]{1,400}")
 _MAX_RETRY_DELAY = 30.0
 _IDEMPOTENCY_KEY = re.compile(r"^[\x21-\x7e]{1,255}$")
 _HEADER_TOKEN = re.compile(r"[\x21-\x7e]+")
 _IN_FLIGHT_CODES = frozenset({"idempotency_key_in_flight"})
+# A cancel or move that comes too late: PostHog is sending the email, or it was canceled.
+_TOO_LATE_CODES = frozenset({"already_sending", "canceled"})
 _QUOTA_CODES = frozenset({"quota_limited"})
 
 
@@ -202,18 +234,83 @@ class EmailWarning:
 
 @dataclass(frozen=True)
 class SendEmailResult:
-    """The queued email. ``id`` is its message ID.
+    """The accepted email. ``id`` is its message ID.
 
     Attributes:
         id: The email's message ID.
         replayed: ``True`` when the response says PostHog replayed an earlier send for
             the same ``idempotency_key`` instead of sending again.
         warnings: Warnings PostHog returned about the request.
+        status: ``"scheduled"`` when the email waits for its ``scheduled_at``,
+            or ``"queued"`` when it goes out now.
+        scheduled_at: When a scheduled email goes out, in UTC. ``None`` for an
+            email that goes out now, also when ``scheduled_at`` was at or
+            before now.
     """
 
     id: str
     replayed: bool = False
     warnings: tuple[EmailWarning, ...] = ()
+    status: str = "queued"
+    scheduled_at: Optional[datetime] = None
+
+
+@dataclass(frozen=True)
+class UpdateEmailResult:
+    """A scheduled email after :meth:`Emails.update` moved it.
+
+    Attributes:
+        id: The email's message ID.
+        status: ``"scheduled"``, until the email goes out.
+        scheduled_at: When the email goes out now, in UTC.
+    """
+
+    id: str
+    status: str
+    scheduled_at: datetime
+
+
+@dataclass(frozen=True)
+class CancelEmailResult:
+    """A scheduled email after :meth:`Emails.cancel` stopped it.
+
+    Attributes:
+        id: The email's message ID.
+        status: ``"canceled"``.
+    """
+
+    id: str
+    status: str
+
+
+@dataclass(frozen=True)
+class ScheduledEmail:
+    """One email that waits for its send time, as :meth:`Emails.list` returns it.
+
+    Attributes:
+        id: The email's message ID.
+        to: The recipient's email address.
+        scheduled_at: When the email goes out, in UTC.
+        created_at: When PostHog accepted the send.
+    """
+
+    id: str
+    to: str
+    scheduled_at: datetime
+    created_at: datetime
+
+
+@dataclass(frozen=True)
+class EmailList:
+    """One page of :meth:`Emails.list`.
+
+    Attributes:
+        results: The emails on this page, the soonest first.
+        next: Pass it as ``cursor`` to read the next page. ``None`` on the last page.
+    """
+
+    results: tuple[ScheduledEmail, ...]
+    next: Optional[str] = None
 
 
 class _RecipientArgs(TypedDict):
@@ -221,6 +318,7 @@ class _RecipientArgs(TypedDict):
     distinct_id: NotRequired[Optional[str]]
     idempotency_key: NotRequired[Optional[str]]
     max_retries: NotRequired[Optional[int]]
+    scheduled_at: NotRequired[Optional[Union[datetime, str]]]
 
 
 class TemplateEmailArgs(_RecipientArgs):
@@ -235,6 +333,8 @@ class TemplateEmailArgs(_RecipientArgs):
         idempotency_key: 1 to 255 visible ASCII characters.
         max_retries: Overrides the client's maximum number of retries for this
             send. ``0`` turns retries off.
+        scheduled_at: When to send the email, up to 30 days ahead. See
+            :meth:`Emails.send`.
     """
 
     template_id: Union[str, UUID]
@@ -273,6 +373,8 @@ class CodeEmailArgs(_RecipientArgs):
             first send's result, also when its workflow was paused since.
         max_retries: Overrides the client's maximum number of retries for this
             send. ``0`` turns retries off.
+        scheduled_at: When to send the email, up to 30 days ahead. See
+            :meth:`Emails.send`.
     """
 
     template_key: str
@@ -291,6 +393,7 @@ class CodeEmailArgs(_RecipientArgs):
         "from_",
         "reply_to",
         "capture_engagement",
+        "scheduled_at",
     }
 )
 
@@ -301,6 +404,8 @@ class EmailStatus:
 
     ``status`` is one of:
 
+    - ``scheduled``: the email waits for its ``scheduled_at``.
+    - ``canceled``: :meth:`Emails.cancel` stopped the email before it went out.
     - ``queued``: the email waits to go out.
     - ``sent``: PostHog handed the email to the email provider.
     - ``failed``: the email was not sent, and ``reason`` says why.
@@ -357,6 +462,7 @@ def _prepare(args: Mapping[str, Any]) -> _PreparedSend:
             args.get("distinct_id"),
             args.get("idempotency_key"),
             args.get("max_retries"),
+            args.get("scheduled_at"),
         )
     return _prepare_code_send(args)
 
@@ -391,6 +497,8 @@ def _prepare_code_send(args: Mapping[str, Any]) -> _PreparedSend:
     if distinct_id is not None:
         payload["distinct_id"] = distinct_id
     payload.update(_sender(args))
+    if "scheduled_at" in args:
+        payload["scheduled_at"] = _scheduled_at(args["scheduled_at"])
     return _PreparedSend(
         path=_SEND_PATH,
         body=_encode(payload),
@@ -451,6 +559,7 @@ def _prepare_send(
     distinct_id: Optional[str],
     idempotency_key: Optional[str],
     max_retries: Optional[int] = None,
+    scheduled_at: Optional[Union[datetime, str]] = None,
 ) -> _PreparedSend:
     to, distinct_id, idempotency_key = _recipient(to, distinct_id, idempotency_key)
     max_retries = _retry_limit(max_retries)
@@ -470,6 +579,8 @@ def _prepare_send(
     }
     if distinct_id is not None:
         payload["distinct_id"] = distinct_id
+    if scheduled_at is not None:
+        payload["scheduled_at"] = _scheduled_at(scheduled_at)
     return _PreparedSend(
         path=_SEND_PATH,
         body=json.dumps(payload, cls=DatetimeSerializer).encode("utf-8"),
@@ -520,6 +631,56 @@ def _status_path(message_id: Union[str, UUID]) -> str:
     return _STATUS_PATH.format(message_id=message_uuid)
 
 
+def _cancel_path(message_id: Union[str, UUID]) -> str:
+    return _status_path(message_id) + "cancel/"
+
+
+def _scheduled_at(value: Any) -> str:
+    """The send time on the wire: an aware ``datetime`` as UTC ISO 8601, a string with an offset as given."""
+    if isinstance(value, datetime):
+        if value.utcoffset() is None:
+            raise ValueError(
+                "scheduled_at has no timezone. Pass an aware datetime, such as "
+                "datetime.now(timezone.utc) + timedelta(hours=1)."
+            )
+        return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
+    if isinstance(value, str):
+        if not _TIME_WITH_OFFSET.fullmatch(value):
+            raise ValueError(
+                "scheduled_at must be an ISO 8601 time with a timezone offset, "
+                "such as 2026-12-01T09:00:00Z or 2026-12-01T09:00:00+02:00."
+            )
+        return value
+    raise TypeError("scheduled_at must be an aware datetime or an ISO 8601 string.")
+
+
+def _prepare_update(scheduled_at: Any) -> bytes:
+    if scheduled_at is None:
+        raise TypeError("update() needs scheduled_at, the email's new send time.")
+    return _encode({"scheduled_at": _scheduled_at(scheduled_at)})
+
+
+def _list_path(status: Any, cursor: Any, limit: Any) -> str:
+    if not isinstance(status, str):
+        raise TypeError("status must be a string, such as 'scheduled'.")
+    if not status:
+        raise ValueError("status must not be empty.")
+    query: List[tuple[str, str]] = [("status", status)]
+    if cursor is not None:
+        if not isinstance(cursor, str):
+            raise TypeError("cursor must be the next value of the previous page.")
+        if not cursor:
+            raise ValueError("cursor must not be empty.")
+        query.append(("cursor", cursor))
+    if limit is not None:
+        if isinstance(limit, bool) or not isinstance(limit, int):
+            raise TypeError("limit must be an integer.")
+        if not 1 <= limit <= _MAX_LIST_LIMIT:
+            raise ValueError(f"limit must be from 1 to {_MAX_LIST_LIMIT}.")
+        query.append(("limit", str(limit)))
+    return f"{_LIST_PATH}?{urlencode(query)}"
+
+
 def _validate_options(
     timeout: float, max_retries: int, migrated_from: Optional[str]
 ) -> None:
@@ -562,7 +723,7 @@ def _error_name(status: Optional[int], code: str) -> str:
     if status == 409:
         return (
             "validation_error"
-            if code == "template_outdated"
+            if code == "template_outdated" or code in _TOO_LATE_CODES
             else "concurrent_idempotent_requests"
         )
     if status == 422:
@@ -608,6 +769,9 @@ def _parse_response(
         return SendEmailResult(
             id=message_id,
             replayed=fields.get("replayed") is True,
+            # A server without scheduled sends answers no status: it queued the email.
+            status=_text(fields.get("status")) or "queued",
+            scheduled_at=_timestamp(fields.get("scheduled_at")),
             warnings=tuple(
                 EmailWarning(
                     code=_text(warning.get("code")) or "warning",
@@ -664,6 +828,80 @@ def _parse_status_response(
     )
 
 
+def _parse_update_response(
+    status: int, payload: Any, retry_after_header: Optional[str]
+) -> UpdateEmailResult:
+    fields = _success_fields(status, payload, retry_after_header)
+    message_id = _text(fields.get("id"))
+    email_status = _text(fields.get("status"))
+    scheduled_at = _timestamp(fields.get("scheduled_at"))
+    if message_id is None or email_status is None or scheduled_at is None:
+        raise _sdk_error(
+            "api_error", "PostHog answered without the email's send time.", status
+        )
+    return UpdateEmailResult(
+        id=message_id, status=email_status, scheduled_at=scheduled_at
+    )
+
+
+def _parse_cancel_response(
+    status: int, payload: Any, retry_after_header: Optional[str]
+) -> CancelEmailResult:
+    fields = _success_fields(status, payload, retry_after_header)
+    message_id = _text(fields.get("id"))
+    email_status = _text(fields.get("status"))
+    if message_id is None or email_status is None:
+        raise _sdk_error(
+            "api_error", "PostHog answered without the email's status.", status
+        )
+    return CancelEmailResult(id=message_id, status=email_status)
+
+
+def _parse_list_response(
+    status: int, payload: Any, retry_after_header: Optional[str]
+) -> EmailList:
+    fields = _success_fields(status, payload, retry_after_header)
+    rows = fields.get("results")
+    if not isinstance(rows, list):
+        raise _sdk_error(
+            "api_error", "PostHog answered without a list of emails.", status
+        )
+    results = []
+    for row in rows:
+        row = row if isinstance(row, dict) else {}
+        message_id = _text(row.get("id"))
+        to = _text(row.get("to"))
+        scheduled_at = _timestamp(row.get("scheduled_at"))
+        created_at = _timestamp(row.get("created_at"))
+        if (
+            message_id is None
+            or to is None
+            or scheduled_at is None
+            or created_at is None
+        ):
+            # Dropping a row would hide an email that still goes out.
+            raise _sdk_error(
+                "api_error",
+                "PostHog answered with an email the client can't read.",
+                status,
+            )
+        results.append(
+            ScheduledEmail(
+                id=message_id, to=to, scheduled_at=scheduled_at, created_at=created_at
+            )
+        )
+    return EmailList(results=tuple(results), next=_text(fields.get("next")))
+
+
+def _success_fields(
+    status: int, payload: Any, retry_after_header: Optional[str]
+) -> Mapping[str, Any]:
+    fields = payload if isinstance(payload, dict) else {}
+    if not 200 <= status < 300:
+        raise _response_error(fields, status, retry_after_header)
+    return fields
+
+
 def _response_error(
     fields: Mapping[str, Any], status: int, retry_after_header: Optional[str]
 ) -> EmailError:
@@ -683,6 +921,7 @@ def _response_error(
 
 def _is_retryable(error: EmailError) -> bool:
     # A send carries an idempotency key, so a retry can't queue a second email. A status read changes nothing.
+    # A cancel or a move ends the same however often it lands, so it needs no key.
     if error.status_code is None:
         return error.code == "network_error"
     if error.status_code == 409:
@@ -734,11 +973,22 @@ def _headers(
 
 _SEND_NOT_DONE = "no email was sent"
 _GET_NOT_DONE = "no email status was read"
+_UPDATE_NOT_DONE = "no email was moved"
+_CANCEL_NOT_DONE = "no email was canceled"
+_LIST_NOT_DONE = "no emails were listed"
 _SEND_KEY_NEEDED = "Sending email needs secret_key, a project secret API key with the transactional_email:write scope."
 _GET_KEY_NEEDED = (
     "Reading an email status needs secret_key, a project secret API key with the "
     "transactional_email:read or transactional_email:write scope."
 )
+_MANAGE_KEY_NEEDED = (
+    "Moving or canceling a scheduled email needs secret_key, a project secret API key "
+    "with the transactional_email:write scope."
+)
+_LIST_KEY_NEEDED = (
+    "Listing scheduled emails needs secret_key, a project secret API key with the "
+    "transactional_email:read or transactional_email:write scope."
+)
 _ResultT = TypeVar("_ResultT")
 _Parse = Callable[[int, Any, Optional[str]], _ResultT]
 
@@ -867,10 +1117,20 @@ def send(self, **kwargs: Any) -> SendEmailResult:
                 outside this call, too.
             max_retries: Overrides the client's maximum number of retries for
                 this send. ``0`` turns retries off.
+            scheduled_at: When to send the email, up to 30 days after this
+                call: an aware ``datetime``, or an ISO 8601 string with an
+                offset, such as ``"2026-12-01T09:00:00+02:00"``. A naive
+                ``datetime`` or a string without an offset raises
+                ``ValueError`` before any request, so the email never goes out
+                in the wrong timezone. A time at or before now sends now. Move
+                or stop a scheduled email with :meth:`update` and
+                :meth:`cancel`, and find it with :meth:`list`. The idempotency
+                key covers the time: the same key with another time raises
+                ``idempotency_key_reused``.
 
         Returns:
-            The queued email. A result means PostHog queued the email, not that
-            it was delivered.
+            The accepted email. A result means PostHog queued or scheduled the
+            email, not that it was delivered. ``status`` says which.
 
         Raises:
             EmailError: PostHog did not queue the email, or the client could
@@ -893,6 +1153,11 @@ def send(self, **kwargs: Any) -> SendEmailResult:
                 html=rendered_html,
                 distinct_id="user_123",
             )
+            emails.send(
+                to="user@example.com",
+                template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b",
+                scheduled_at=datetime.now(timezone.utc) + timedelta(days=1),
+            )
             ```
         """
         prepared = _prepare(kwargs)
@@ -952,6 +1217,119 @@ def get(self, message_id: Union[str, UUID]) -> EmailStatus:
         finally:
             finish()
 
+    def update(
+        self, message_id: Union[str, UUID], *, scheduled_at: Union[datetime, str]
+    ) -> UpdateEmailResult:
+        """Move a scheduled email to another send time.
+
+        Only the time changes. To change the content, the recipient or the
+        sender, :meth:`cancel` the email and send it again.
+
+        Retries like :meth:`send`. A move ends the same however often it
+        lands, so it takes no idempotency key.
+
+        Args:
+            message_id: The ``id`` that :meth:`send` returned.
+            scheduled_at: The new send time, as :meth:`send` takes it. It must
+                be within 30 days after the first send call. A time at or
+                before now sends the email now.
+
+        Returns:
+            The email with its new send time.
+
+        Raises:
+            EmailError: The email was not moved. ``code`` is
+                ``already_sending`` once PostHog has started sending it,
+                ``canceled`` after a cancel, ``invalid_scheduled_at`` for a time
+                too far ahead, and ``not_found`` for an unknown ID or an email
+                that went out.
+            TypeError, ValueError: An argument has the wrong type or format.
+
+        Examples:
+            ```python
+            emails.update(result.id, scheduled_at=datetime.now(timezone.utc) + timedelta(hours=2))
+            ```
+        """
+        path = _status_path(message_id)
+        body = _prepare_update(scheduled_at)
+        finish = self._start(_UPDATE_NOT_DONE)
+        try:
+            headers = _headers(
+                self._secret_key(_MANAGE_KEY_NEEDED), self.migrated_from, has_body=True
+            )
+            return self._call("PATCH", path, body, headers, _parse_update_response)
+        finally:
+            finish()
+
+    def cancel(self, message_id: Union[str, UUID]) -> CancelEmailResult:
+        """Stop a scheduled email before it goes out.
+
+        A canceled email never goes out, and :meth:`get` reads ``canceled``.
+        Canceling it again also succeeds. Retries like :meth:`send`.
+
+        Args:
+            message_id: The ``id`` that :meth:`send` returned.
+
+        Returns:
+            The canceled email.
+
+        Raises:
+            EmailError: The email was not canceled. ``code`` is
+                ``already_sending`` once PostHog has started sending it, and
+                ``not_found`` for an unknown ID or an email that went out.
+            ValueError: ``message_id`` is not a message ID.
+        """
+        path = _cancel_path(message_id)
+        finish = self._start(_CANCEL_NOT_DONE)
+        try:
+            headers = _headers(self._secret_key(_MANAGE_KEY_NEEDED), self.migrated_from)
+            return self._call("POST", path, None, headers, _parse_cancel_response)
+        finally:
+            finish()
+
+    def list(
+        self,
+        *,
+        status: str = "scheduled",
+        cursor: Optional[str] = None,
+        limit: Optional[int] = None,
+    ) -> EmailList:
+        """List the project's emails that wait for their send time, the soonest first.
+
+        Reads one page. Pass the page's ``next`` as ``cursor`` to read the
+        next one. A list is never retried: call it again.
+
+        Args:
+            status: ``"scheduled"``, the only status PostHog lists.
+            cursor: The ``next`` of the previous page.
+            limit: Emails per page, 1 to 100. PostHog's default is 100.
+
+        Returns:
+            One page of scheduled emails.
+
+        Raises:
+            EmailError: PostHog did not answer with a list.
+            TypeError, ValueError: An argument has the wrong type or format.
+
+        Examples:
+            ```python
+            page = emails.list()
+            while True:
+                for email in page.results:
+                    print(email.id, email.to, email.scheduled_at)
+                if page.next is None:
+                    break
+                page = emails.list(cursor=page.next)
+            ```
+        """
+        path = _list_path(status, cursor, limit)
+        finish = self._start(_LIST_NOT_DONE)
+        try:
+            headers = _headers(self._secret_key(_LIST_KEY_NEEDED), self.migrated_from)
+            return self._call("GET", path, None, headers, _parse_list_response, 0)
+        finally:
+            finish()
+
     def _call(
         self,
         method: str,
@@ -1118,6 +1496,61 @@ async def get(self, message_id: Union[str, UUID]) -> EmailStatus:
         finally:
             finish()
 
+    async def update(
+        self, message_id: Union[str, UUID], *, scheduled_at: Union[datetime, str]
+    ) -> UpdateEmailResult:
+        """Move a scheduled email to another send time.
+
+        Takes the same arguments and raises the same errors as
+        :meth:`Emails.update`.
+        """
+        path = _status_path(message_id)
+        body = _prepare_update(scheduled_at)
+        finish = self._start(_UPDATE_NOT_DONE)
+        try:
+            headers = _headers(
+                self._secret_key(_MANAGE_KEY_NEEDED), self.migrated_from, has_body=True
+            )
+            return await self._call(
+                "PATCH", path, body, headers, _parse_update_response
+            )
+        finally:
+            finish()
+
+    async def cancel(self, message_id: Union[str, UUID]) -> CancelEmailResult:
+        """Stop a scheduled email before it goes out.
+
+        Takes the same argument and raises the same errors as
+        :meth:`Emails.cancel`.
+        """
+        path = _cancel_path(message_id)
+        finish = self._start(_CANCEL_NOT_DONE)
+        try:
+            headers = _headers(self._secret_key(_MANAGE_KEY_NEEDED), self.migrated_from)
+            return await self._call("POST", path, None, headers, _parse_cancel_response)
+        finally:
+            finish()
+
+    async def list(
+        self,
+        *,
+        status: str = "scheduled",
+        cursor: Optional[str] = None,
+        limit: Optional[int] = None,
+    ) -> EmailList:
+        """List the project's emails that wait for their send time, the soonest first.
+
+        Takes the same arguments and raises the same errors as
+        :meth:`Emails.list`.
+        """
+        path = _list_path(status, cursor, limit)
+        finish = self._start(_LIST_NOT_DONE)
+        try:
+            headers = _headers(self._secret_key(_LIST_KEY_NEEDED), self.migrated_from)
+            return await self._call("GET", path, None, headers, _parse_list_response, 0)
+        finally:
+            finish()
+
     async def _call(
         self,
         method: str,
diff --git a/posthog/test/test_emails.py b/posthog/test/test_emails.py
index eaf1c0949..390fe2dc8 100644
--- a/posthog/test/test_emails.py
+++ b/posthog/test/test_emails.py
@@ -3,7 +3,7 @@
 import asyncio
 import json
 import logging
-from datetime import datetime, timezone
+from datetime import date, datetime, timedelta, timezone
 from decimal import Decimal
 from typing import Any, Optional, Union
 from unittest import mock
@@ -799,3 +799,416 @@ def test_rejects_a_code_send_that_mixes_or_misses_fields(_name, overrides, error
         )
 
     assert adapter.requests == []
+
+
+# Scheduled sends: schedule, move, cancel and list.
+
+CANCEL_URL = f"{STATUS_URL}cancel/"
+LIST_URL = "https://us.posthog.com/api/projects/@current/messaging/emails/"
+LATER = datetime(2026, 12, 1, 9, 0, tzinfo=timezone(timedelta(hours=2)))
+
+
+@parameterized.expand(
+    [
+        ("aware_datetime_goes_as_utc", LATER, "2026-12-01T07:00:00Z"),
+        ("utc_string_as_given", "2026-12-01T07:00:00Z", "2026-12-01T07:00:00Z"),
+        (
+            "offset_string_as_given",
+            "2026-12-01T09:00:00.250+02:00",
+            "2026-12-01T09:00:00.250+02:00",
+        ),
+    ]
+)
+def test_send_schedules_both_forms_for_a_time_with_an_offset(_name, value, wire):
+    emails, adapter = emails_with(
+        [
+            accepted(status="scheduled", scheduled_at="2026-12-01T07:00:00Z"),
+            accepted(status="scheduled", scheduled_at="2026-12-01T07:00:00Z"),
+        ]
+    )
+
+    template_result = send(emails, scheduled_at=value)
+    code_result = send_code(emails, html=HTML, scheduled_at=value)
+
+    for result in (template_result, code_result):
+        assert result.status == "scheduled"
+        assert result.scheduled_at == datetime(2026, 12, 1, 7, tzinfo=timezone.utc)
+    assert [
+        json.loads(request.body)["scheduled_at"] for request in adapter.requests
+    ] == [
+        wire,
+        wire,
+    ]
+
+
+@parameterized.expand(
+    [
+        ("left_out", {}, {}),
+        ("passed_as_none", {"scheduled_at": None}, {}),
+        ("code_form_as_none", {"scheduled_at": None, "template_key": "x"}, {}),
+    ]
+)
+def test_a_send_now_carries_no_scheduled_at_and_reads_queued(_name, overrides, _):
+    emails, adapter = emails_with([accepted()])
+
+    if "template_key" in overrides:
+        result = send_code(emails, html=HTML, scheduled_at=None)
+    else:
+        result = send(emails, **overrides)
+
+    assert (result.status, result.scheduled_at) == ("queued", None)
+    assert "scheduled_at" not in json.loads(adapter.requests[0].body)
+
+
+def test_a_time_at_or_before_now_reads_as_the_servers_queued_answer():
+    emails, _ = emails_with([accepted(status="queued", scheduled_at=None)])
+
+    result = send(emails, scheduled_at="2020-01-01T00:00:00Z")
+
+    assert (result.status, result.scheduled_at) == ("queued", None)
+
+
+@parameterized.expand(
+    [
+        ("naive_datetime", datetime(2026, 12, 1, 9), ValueError),
+        ("string_without_offset", "2026-12-01T09:00:00", ValueError),
+        ("date_only_string", "2026-12-01", ValueError),
+        ("natural_language", "in 1 min", ValueError),
+        ("empty_string", "", ValueError),
+        ("unix_timestamp", 1796115600, TypeError),
+        ("date", date(2026, 12, 1), TypeError),
+    ]
+)
+def test_refuses_a_time_without_a_timezone_before_any_request(_name, value, error):
+    emails, adapter = emails_with([])
+
+    for call in (
+        lambda: send(emails, scheduled_at=value),
+        lambda: send_code(emails, html=HTML, scheduled_at=value),
+        lambda: emails.update(MESSAGE_ID, scheduled_at=value),
+    ):
+        with pytest.raises(error, match="scheduled_at"):
+            call()
+
+    assert adapter.requests == []
+
+
+def test_update_moves_a_scheduled_email_and_returns_its_new_time():
+    emails, adapter = emails_with(
+        [
+            (
+                200,
+                {
+                    "id": MESSAGE_ID,
+                    "status": "scheduled",
+                    "scheduled_at": "2026-12-01T07:00:00Z",
+                },
+                {},
+            )
+        ]
+    )
+
+    result = emails.update(MESSAGE_ID, scheduled_at=LATER)
+
+    assert (result.id, result.status, result.scheduled_at) == (
+        MESSAGE_ID,
+        "scheduled",
+        datetime(2026, 12, 1, 7, tzinfo=timezone.utc),
+    )
+    [request] = adapter.requests
+    assert (request.method, request.url) == ("PATCH", STATUS_URL)
+    assert json.loads(request.body) == {"scheduled_at": "2026-12-01T07:00:00Z"}
+    assert request.headers["Content-Type"] == "application/json"
+    assert request.headers["Authorization"] == f"Bearer {SECRET_KEY}"
+    assert "Idempotency-Key" not in request.headers
+
+
+def test_cancel_stops_a_scheduled_email():
+    emails, adapter = emails_with([(200, {"id": MESSAGE_ID, "status": "canceled"}, {})])
+
+    result = emails.cancel(MESSAGE_ID)
+
+    assert (result.id, result.status) == (MESSAGE_ID, "canceled")
+    [request] = adapter.requests
+    assert (request.method, request.url, request.body) == ("POST", CANCEL_URL, None)
+    assert "Idempotency-Key" not in request.headers
+
+
+def list_reply(next_cursor: Optional[str] = None) -> Reply:
+    return (
+        200,
+        {
+            "results": [
+                {
+                    "id": MESSAGE_ID,
+                    "to": "user@example.com",
+                    "scheduled_at": "2026-12-01T07:00:00Z",
+                    "created_at": "2026-11-12T10:00:00.123000Z",
+                }
+            ],
+            "next": next_cursor,
+        },
+        {},
+    )
+
+
+@parameterized.expand(
+    [
+        ("first_page", {}, "?status=scheduled", "cursor_2"),
+        (
+            "next_page",
+            {"cursor": "cursor_2", "limit": 50},
+            "?status=scheduled&cursor=cursor_2&limit=50",
+            None,
+        ),
+    ]
+)
+def test_list_reads_scheduled_emails_a_page_at_a_time(_name, args, query, next_cursor):
+    emails, adapter = emails_with([list_reply(next_cursor)])
+
+    page = emails.list(**args)
+
+    [request] = adapter.requests
+    assert (request.method, request.url) == ("GET", LIST_URL + query)
+    [email] = page.results
+    assert (email.id, email.to, email.scheduled_at, email.created_at) == (
+        MESSAGE_ID,
+        "user@example.com",
+        datetime(2026, 12, 1, 7, tzinfo=timezone.utc),
+        datetime(2026, 11, 12, 10, 0, 0, 123000, tzinfo=timezone.utc),
+    )
+    assert page.next == next_cursor
+
+
+@parameterized.expand(
+    [
+        ("zero", {"limit": 0}, ValueError),
+        ("over_a_page", {"limit": 101}, ValueError),
+        ("not_an_int", {"limit": "10"}, TypeError),
+        ("a_bool", {"limit": True}, TypeError),
+        ("empty_cursor", {"cursor": ""}, ValueError),
+        ("empty_status", {"status": ""}, ValueError),
+    ]
+)
+def test_list_refuses_bad_arguments_before_any_request(_name, args, error):
+    emails, adapter = emails_with([])
+
+    with pytest.raises(error):
+        emails.list(**args)
+
+    assert adapter.requests == []
+
+
+@parameterized.expand(
+    [
+        ("update", lambda emails, id: emails.update(id, scheduled_at=LATER)),
+        ("cancel", lambda emails, id: emails.cancel(id)),
+    ]
+)
+def test_update_and_cancel_refuse_an_id_that_is_not_a_message_id(_name, call):
+    emails, adapter = emails_with([])
+
+    with pytest.raises(ValueError, match="message_id"):
+        call(emails, "msg_1")
+
+    assert adapter.requests == []
+
+
+def test_update_needs_a_time():
+    emails, adapter = emails_with([])
+
+    with pytest.raises(TypeError, match="scheduled_at"):
+        emails.update(MESSAGE_ID, scheduled_at=None)
+
+    assert adapter.requests == []
+
+
+MANAGE_CALLS = [
+    (
+        "update",
+        lambda emails: emails.update(MESSAGE_ID, scheduled_at=LATER),
+        (
+            200,
+            {
+                "id": MESSAGE_ID,
+                "status": "scheduled",
+                "scheduled_at": "2026-12-01T07:00:00Z",
+            },
+            {},
+        ),
+    ),
+    (
+        "cancel",
+        lambda emails: emails.cancel(MESSAGE_ID),
+        (200, {"id": MESSAGE_ID, "status": "canceled"}, {}),
+    ),
+]
+
+
+@parameterized.expand(
+    [
+        (f"{name}_{case}", call, success, first_reply, retried)
+        for name, call, success in MANAGE_CALLS
+        for case, first_reply, retried in [
+            ("rate_limited", rejected(429, "throttled"), True),
+            ("service_unavailable", rejected(503, "delivery_unavailable"), True),
+            ("connection_error", requests.exceptions.ConnectionError("down"), True),
+            ("already_sending", rejected(409, "already_sending"), False),
+            ("canceled", rejected(409, "canceled"), False),
+            ("not_found", rejected(404, "not_found"), False),
+            ("too_far", rejected(400, "invalid_scheduled_at"), False),
+            ("over_quota", rejected(429, "quota_limited"), False),
+        ]
+    ]
+)
+def test_update_and_cancel_retry_like_a_send(
+    _name, call, success, first_reply, retried
+):
+    emails, adapter = emails_with([first_reply, success])
+
+    if retried:
+        assert call(emails).id == MESSAGE_ID
+        assert len(adapter.requests) == 2
+    else:
+        with pytest.raises(EmailError):
+            call(emails)
+        assert len(adapter.requests) == 1
+
+
+@parameterized.expand(
+    [
+        ("already_sending", "validation_error"),
+        ("canceled", "validation_error"),
+    ]
+)
+def test_a_cancel_or_move_too_late_is_a_validation_error(code, name):
+    emails, _ = emails_with([rejected(409, code)])
+
+    with pytest.raises(EmailError) as raised:
+        emails.cancel(MESSAGE_ID)
+
+    assert (raised.value.status_code, raised.value.code, raised.value.name) == (
+        409,
+        code,
+        name,
+    )
+
+
+@parameterized.expand(
+    [
+        ("service_unavailable", rejected(503, "delivery_unavailable")),
+        ("rate_limited", rejected(429, "throttled")),
+        ("connection_error", requests.exceptions.ConnectionError("down")),
+    ]
+)
+def test_list_never_retries(_name, first_reply):
+    emails, adapter = emails_with([first_reply, list_reply()])
+
+    with pytest.raises(EmailError):
+        emails.list()
+
+    assert len(adapter.requests) == 1
+
+
+@parameterized.expand(
+    [
+        (
+            "update",
+            lambda emails: emails.update(MESSAGE_ID, scheduled_at=LATER),
+            "missing_secret_key",
+        ),
+        ("cancel", lambda emails: emails.cancel(MESSAGE_ID), "missing_secret_key"),
+        ("list", lambda emails: emails.list(), "missing_secret_key"),
+    ]
+)
+def test_manage_calls_need_the_secret_key_and_an_enabled_client(_name, call, code):
+    adapter = FakeAdapter([])
+    session = requests.Session()
+    session.mount("https://", adapter)
+
+    with pytest.raises(EmailError) as missing:
+        call(Emails(None, session=session))
+    with pytest.raises(EmailError) as disabled:
+        call(Emails(SECRET_KEY, session=session, disabled=True))
+
+    assert (missing.value.code, disabled.value.code) == (code, "client_disabled")
+    assert adapter.requests == []
+
+
+def test_a_malformed_list_page_raises_instead_of_dropping_emails():
+    emails, _ = emails_with(
+        [(200, {"results": [{"id": MESSAGE_ID}], "next": None}, {})]
+    )
+
+    with pytest.raises(EmailError) as raised:
+        emails.list()
+
+    assert raised.value.code == "api_error"
+
+
+@pytest.mark.asyncio
+async def test_async_client_schedules_moves_cancels_and_lists():
+    seen: list[httpx.Request] = []
+    replies = [
+        httpx.Response(
+            202,
+            json={
+                "message_id": MESSAGE_ID,
+                "status": "scheduled",
+                "scheduled_at": "2026-12-01T07:00:00Z",
+            },
+        ),
+        httpx.Response(503, json={"code": "delivery_unavailable"}),
+        httpx.Response(
+            200,
+            json={
+                "id": MESSAGE_ID,
+                "status": "scheduled",
+                "scheduled_at": "2026-12-02T07:00:00Z",
+            },
+        ),
+        httpx.Response(503, json={"code": "delivery_unavailable"}),
+        httpx.Response(200, json={"id": MESSAGE_ID, "status": "canceled"}),
+        httpx.Response(200, json=list_reply()[1]),
+        httpx.Response(503, json={"code": "delivery_unavailable"}),
+    ]
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request)
+        return replies.pop(0)
+
+    emails = AsyncEmails(
+        SECRET_KEY,
+        http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
+    )
+
+    with mock.patch("posthog.emails.asyncio.sleep"):
+        sent = await emails.send(
+            to="user@example.com", template_id=TEMPLATE_ID, scheduled_at=LATER
+        )
+        moved = await emails.update(
+            MESSAGE_ID, scheduled_at="2026-12-02T09:00:00+02:00"
+        )
+        canceled = await emails.cancel(MESSAGE_ID)
+        page = await emails.list(limit=10)
+        with pytest.raises(EmailError):
+            await emails.list()
+
+    assert (sent.status, moved.scheduled_at, canceled.status) == (
+        "scheduled",
+        datetime(2026, 12, 2, 7, tzinfo=timezone.utc),
+        "canceled",
+    )
+    assert [email.id for email in page.results] == [MESSAGE_ID]
+    assert [(request.method, str(request.url)) for request in seen] == [
+        ("POST", SEND_URL),
+        ("PATCH", STATUS_URL),
+        ("PATCH", STATUS_URL),
+        ("POST", CANCEL_URL),
+        ("POST", CANCEL_URL),
+        ("GET", LIST_URL + "?status=scheduled&limit=10"),
+        ("GET", LIST_URL + "?status=scheduled"),
+    ]
+    assert json.loads(seen[0].content)["scheduled_at"] == "2026-12-01T07:00:00Z"
+    assert json.loads(seen[1].content) == {"scheduled_at": "2026-12-02T09:00:00+02:00"}
+    assert "Idempotency-Key" not in seen[1].headers
diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt
index bfffd5d64..d8b16ae7e 100644
--- a/references/public_api_snapshot.txt
+++ b/references/public_api_snapshot.txt
@@ -888,6 +888,8 @@ attribute posthog.debug = False
 attribute posthog.default_client = None
 attribute posthog.disable_geoip = True
 attribute posthog.disabled = False
+attribute posthog.emails.CancelEmailResult.id: str
+attribute posthog.emails.CancelEmailResult.status: str
 attribute posthog.emails.CodeEmailArgs.capture_engagement: NotRequired[Optional[bool]]
 attribute posthog.emails.CodeEmailArgs.from_: NotRequired[Optional[str]]
 attribute posthog.emails.CodeEmailArgs.html: NotRequired[Optional[str]]
@@ -904,6 +906,8 @@ attribute posthog.emails.EmailError.name = name
 attribute posthog.emails.EmailError.retry_after = retry_after
 attribute posthog.emails.EmailError.status_code = status_code
 attribute posthog.emails.EmailError.why = why
+attribute posthog.emails.EmailList.next: Optional[str] = None
+attribute posthog.emails.EmailList.results: tuple[ScheduledEmail, ...]
 attribute posthog.emails.EmailStatus.created_at: datetime
 attribute posthog.emails.EmailStatus.id: str
 attribute posthog.emails.EmailStatus.reason: Optional[str] = None
@@ -912,11 +916,20 @@ attribute posthog.emails.EmailStatus.status_at: datetime
 attribute posthog.emails.EmailWarning.code: str
 attribute posthog.emails.EmailWarning.fix: Optional[str] = None
 attribute posthog.emails.EmailWarning.message: str
+attribute posthog.emails.ScheduledEmail.created_at: datetime
+attribute posthog.emails.ScheduledEmail.id: str
+attribute posthog.emails.ScheduledEmail.scheduled_at: datetime
+attribute posthog.emails.ScheduledEmail.to: str
 attribute posthog.emails.SendEmailResult.id: str
 attribute posthog.emails.SendEmailResult.replayed: bool = False
+attribute posthog.emails.SendEmailResult.scheduled_at: Optional[datetime] = None
+attribute posthog.emails.SendEmailResult.status: str = 'queued'
 attribute posthog.emails.SendEmailResult.warnings: tuple[EmailWarning, ...] = ()
 attribute posthog.emails.TemplateEmailArgs.template_data: NotRequired[Optional[Mapping[str, Any]]]
 attribute posthog.emails.TemplateEmailArgs.template_id: Union[str, UUID]
+attribute posthog.emails.UpdateEmailResult.id: str
+attribute posthog.emails.UpdateEmailResult.scheduled_at: datetime
+attribute posthog.emails.UpdateEmailResult.status: str
 attribute posthog.enable_exception_autocapture = False
 attribute posthog.enable_exception_autocapture_rate_limiting = False
 attribute posthog.enable_full_ai_capture = False
@@ -1343,13 +1356,17 @@ class posthog.client.Client(project_api_key: str, host=None, debug=False, max_qu
 class posthog.consumer.Consumer(queue, api_key, flush_at=100, host=None, on_error=None, flush_interval=5.0, gzip=False, retries=10, timeout: float = 15, historical_migration=False, endpoint=EVENTS_ENDPOINT, max_msg_size=MAX_MSG_SIZE, capture_mode=CaptureMode.V0, capture_compression=CaptureCompression.NONE)
 class posthog.contexts.ContextScope(parent=None, fresh: bool = False, capture_exceptions: bool = True, client: Optional[Client] = None)
 class posthog.emails.AsyncEmails(secret_key: Optional[str], *, host: Optional[str] = None, timeout: float = 15, max_retries: int = 2, disabled: bool = False, http_client: Any = None, migrated_from: Optional[str] = None)
+class posthog.emails.CancelEmailResult(id: str, status: str)
 class posthog.emails.CodeEmailArgs 
 class posthog.emails.EmailError(*, name: str, code: str, message: str, status_code: Optional[int] = None, why: Optional[str] = None, fix: Optional[str] = None, docs_url: Optional[str] = None, field: Optional[str] = None, retry_after: Optional[float] = None)
+class posthog.emails.EmailList(results: tuple[ScheduledEmail, ...], next: Optional[str] = None)
 class posthog.emails.EmailStatus(id: str, status: str, created_at: datetime, status_at: datetime, reason: Optional[str] = None)
 class posthog.emails.EmailWarning(code: str, message: str, fix: Optional[str] = None)
 class posthog.emails.Emails(secret_key: Optional[str], *, host: Optional[str] = None, timeout: float = 15, max_retries: int = 2, disabled: bool = False, session: Optional[requests.Session] = None, migrated_from: Optional[str] = None)
-class posthog.emails.SendEmailResult(id: str, replayed: bool = False, warnings: tuple[EmailWarning, ...] = ())
+class posthog.emails.ScheduledEmail(id: str, to: str, scheduled_at: datetime, created_at: datetime)
+class posthog.emails.SendEmailResult(id: str, replayed: bool = False, warnings: tuple[EmailWarning, ...] = (), status: str = 'queued', scheduled_at: Optional[datetime] = None)
 class posthog.emails.TemplateEmailArgs 
+class posthog.emails.UpdateEmailResult(id: str, status: str, scheduled_at: datetime)
 class posthog.exception_capture.ExceptionCapture(client: Client, rate_limiting_enabled=False, bucket_size=DEFAULT_BUCKET_SIZE, refill_rate=DEFAULT_REFILL_RATE, refill_interval_seconds=DEFAULT_REFILL_INTERVAL_SECONDS)
 class posthog.exception_utils.AnnotatedValue(value, metadata)
 class posthog.exception_utils.VariableSizeLimiter(max_size=DEFAULT_TOTAL_VARIABLES_SIZE_LIMIT)
@@ -1848,11 +1865,17 @@ method posthog.contexts.ContextScope.set_device_id(device_id: str)
 method posthog.contexts.ContextScope.set_distinct_id(distinct_id: str)
 method posthog.contexts.ContextScope.set_session_id(session_id: str)
 method posthog.emails.AsyncEmails.aclose() -> None
+method posthog.emails.AsyncEmails.cancel(message_id: Union[str, UUID]) -> CancelEmailResult
 method posthog.emails.AsyncEmails.get(message_id: Union[str, UUID]) -> EmailStatus
+method posthog.emails.AsyncEmails.list(*, status: str = 'scheduled', cursor: Optional[str] = None, limit: Optional[int] = None) -> EmailList
 method posthog.emails.AsyncEmails.send(**kwargs: Any) -> SendEmailResult
+method posthog.emails.AsyncEmails.update(message_id: Union[str, UUID], *, scheduled_at: Union[datetime, str]) -> UpdateEmailResult
+method posthog.emails.Emails.cancel(message_id: Union[str, UUID]) -> CancelEmailResult
 method posthog.emails.Emails.close() -> None
 method posthog.emails.Emails.get(message_id: Union[str, UUID]) -> EmailStatus
+method posthog.emails.Emails.list(*, status: str = 'scheduled', cursor: Optional[str] = None, limit: Optional[int] = None) -> EmailList
 method posthog.emails.Emails.send(**kwargs: Any) -> SendEmailResult
+method posthog.emails.Emails.update(message_id: Union[str, UUID], *, scheduled_at: Union[datetime, str]) -> UpdateEmailResult
 method posthog.exception_capture.ExceptionCapture.capture_exception(exception, metadata=None)
 method posthog.exception_capture.ExceptionCapture.close()
 method posthog.exception_capture.ExceptionCapture.exception_handler(exc_type, exc_value, exc_traceback)

From 067e51de52bae153b6dd667258a49c0f3282c614 Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Thu, 12 Nov 2026 11:43:03 +0000
Subject: [PATCH 2/6] refactor(emails): check scheduled_at strings with the
 status timestamp parser

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 posthog/emails.py           | 7 ++-----
 posthog/test/test_emails.py | 2 ++
 2 files changed, 4 insertions(+), 5 deletions(-)

diff --git a/posthog/emails.py b/posthog/emails.py
index 045a5d949..1cf933e86 100644
--- a/posthog/emails.py
+++ b/posthog/emails.py
@@ -167,10 +167,6 @@
 _CANCEL_PATH = _STATUS_PATH + "cancel/"
 _LIST_PATH = "/api/projects/@current/messaging/emails/"
 _MAX_LIST_LIMIT = 100
-# An ISO 8601 date and time that ends in an offset, such as Z or +02:00. PostHog parses the rest.
-_TIME_WITH_OFFSET = re.compile(
-    r"\d{4}-\d{2}-\d{2}[Tt ]\d{2}:\d{2}(?::\d{2}(?:[.,]\d+)?)?(?:[Zz]|[+-]\d{2}(?::?\d{2})?)"
-)
 _TEMPLATE_KEY = re.compile(r"[a-zA-Z0-9_-]{1,400}")
 _MAX_RETRY_DELAY = 30.0
 _IDEMPOTENCY_KEY = re.compile(r"^[\x21-\x7e]{1,255}$")
@@ -645,7 +641,8 @@ def _scheduled_at(value: Any) -> str:
             )
         return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
     if isinstance(value, str):
-        if not _TIME_WITH_OFFSET.fullmatch(value):
+        parsed = _timestamp(value)
+        if parsed is None or parsed.utcoffset() is None:
             raise ValueError(
                 "scheduled_at must be an ISO 8601 time with a timezone offset, "
                 "such as 2026-12-01T09:00:00Z or 2026-12-01T09:00:00+02:00."
diff --git a/posthog/test/test_emails.py b/posthog/test/test_emails.py
index 390fe2dc8..563ff1ede 100644
--- a/posthog/test/test_emails.py
+++ b/posthog/test/test_emails.py
@@ -817,6 +817,7 @@ def test_rejects_a_code_send_that_mixes_or_misses_fields(_name, overrides, error
             "2026-12-01T09:00:00.250+02:00",
             "2026-12-01T09:00:00.250+02:00",
         ),
+        ("space_separated_as_given", "2026-12-01 07:00Z", "2026-12-01 07:00Z"),
     ]
 )
 def test_send_schedules_both_forms_for_a_time_with_an_offset(_name, value, wire):
@@ -874,6 +875,7 @@ def test_a_time_at_or_before_now_reads_as_the_servers_queued_answer():
         ("string_without_offset", "2026-12-01T09:00:00", ValueError),
         ("date_only_string", "2026-12-01", ValueError),
         ("natural_language", "in 1 min", ValueError),
+        ("not_a_date", "2026-13-01T09:00:00Z", ValueError),
         ("empty_string", "", ValueError),
         ("unix_timestamp", 1796115600, TypeError),
         ("date", date(2026, 12, 1), TypeError),

From 8a365d276627ecc8409e845f998d39e7e1d6e299 Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Thu, 12 Nov 2026 11:44:04 +0000
Subject: [PATCH 3/6] fix(emails): name a too-late 409 as the shared SDK spec
 does

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 posthog/emails.py           |  4 +---
 posthog/test/test_emails.py | 12 ++++--------
 2 files changed, 5 insertions(+), 11 deletions(-)

diff --git a/posthog/emails.py b/posthog/emails.py
index 1cf933e86..338f4fbd5 100644
--- a/posthog/emails.py
+++ b/posthog/emails.py
@@ -172,8 +172,6 @@
 _IDEMPOTENCY_KEY = re.compile(r"^[\x21-\x7e]{1,255}$")
 _HEADER_TOKEN = re.compile(r"[\x21-\x7e]+")
 _IN_FLIGHT_CODES = frozenset({"idempotency_key_in_flight"})
-# A cancel or move that comes too late: PostHog is sending the email, or it was canceled.
-_TOO_LATE_CODES = frozenset({"already_sending", "canceled"})
 _QUOTA_CODES = frozenset({"quota_limited"})
 
 
@@ -720,7 +718,7 @@ def _error_name(status: Optional[int], code: str) -> str:
     if status == 409:
         return (
             "validation_error"
-            if code == "template_outdated" or code in _TOO_LATE_CODES
+            if code == "template_outdated"
             else "concurrent_idempotent_requests"
         )
     if status == 422:
diff --git a/posthog/test/test_emails.py b/posthog/test/test_emails.py
index 563ff1ede..f77e66785 100644
--- a/posthog/test/test_emails.py
+++ b/posthog/test/test_emails.py
@@ -1077,22 +1077,18 @@ def test_update_and_cancel_retry_like_a_send(
         assert len(adapter.requests) == 1
 
 
-@parameterized.expand(
-    [
-        ("already_sending", "validation_error"),
-        ("canceled", "validation_error"),
-    ]
-)
-def test_a_cancel_or_move_too_late_is_a_validation_error(code, name):
+@parameterized.expand([("already_sending",), ("canceled",)])
+def test_a_cancel_or_move_too_late_keeps_posthogs_code(code):
     emails, _ = emails_with([rejected(409, code)])
 
     with pytest.raises(EmailError) as raised:
         emails.cancel(MESSAGE_ID)
 
+    # The shared SDK spec names every 409 but template_outdated this way.
     assert (raised.value.status_code, raised.value.code, raised.value.name) == (
         409,
         code,
-        name,
+        "concurrent_idempotent_requests",
     )
 
 

From ef74081861344187a170ef4b3b7916e9ef753855 Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Thu, 12 Nov 2026 12:07:17 +0000
Subject: [PATCH 4/6] fix(emails): match posthog-node on old servers, paging
 and too-late conflicts

- A scheduled send that a PostHog without scheduled sends answers with no
  status now carries the scheduling_unsupported warning, as posthog-node does,
  so the caller sees the email went out now.
- list() raises api_error when a page's next cursor is missing or not a
  nonempty string, instead of ending a paging loop early.
- cancel() raises api_error unless PostHog answers status canceled.
- A 409 already_sending or canceled is named validation_error, like
  template_outdated, as posthog-node names it: a retry can't resolve it.
- list() takes only status="scheduled", typed Literal["scheduled"].
- update() documents that a lost answer to a move to now can retry into
  already_sending. Drop the unused cancel path constant and typing.List.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .github/scripts/check_strict_types.sh |   1 +
 posthog/emails.py                     | 113 +++++++++++++++++---------
 posthog/test/test_emails.py           |  87 +++++++++++++++-----
 references/public_api_snapshot.txt    |   4 +-
 4 files changed, 146 insertions(+), 59 deletions(-)

diff --git a/.github/scripts/check_strict_types.sh b/.github/scripts/check_strict_types.sh
index 13cc955b5..77cd24e96 100755
--- a/.github/scripts/check_strict_types.sh
+++ b/.github/scripts/check_strict_types.sh
@@ -109,6 +109,7 @@ Emails("phs_example").send(to="user@example.com", template_key="welcome", subjec
 Emails("phs_example").update("0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b")  # expected-type-error
 Emails("phs_example").update("0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b", "2026-12-01T09:00:00Z")  # expected-type-error
 Emails("phs_example").list("scheduled")  # expected-type-error
+Emails("phs_example").list(status="sent")  # expected-type-error
 
 
 async def invalid_async_email() -> None:
diff --git a/posthog/emails.py b/posthog/emails.py
index 338f4fbd5..b0375ddff 100644
--- a/posthog/emails.py
+++ b/posthog/emails.py
@@ -119,8 +119,8 @@
 from types import TracebackType
 from typing import (
     Any,
-    List,
     Callable,
+    Literal,
     Mapping,
     Optional,
     Sequence,
@@ -164,7 +164,6 @@
 
 _SEND_PATH = "/api/projects/@current/messaging/send/email/"
 _STATUS_PATH = "/api/projects/@current/messaging/emails/{message_id}/"
-_CANCEL_PATH = _STATUS_PATH + "cancel/"
 _LIST_PATH = "/api/projects/@current/messaging/emails/"
 _MAX_LIST_LIMIT = 100
 _TEMPLATE_KEY = re.compile(r"[a-zA-Z0-9_-]{1,400}")
@@ -173,6 +172,10 @@
 _HEADER_TOKEN = re.compile(r"[\x21-\x7e]+")
 _IN_FLIGHT_CODES = frozenset({"idempotency_key_in_flight"})
 _QUOTA_CODES = frozenset({"quota_limited"})
+# Conflicts a retry can't resolve: the email changed under the request, as posthog-node names them.
+_SETTLED_CONFLICT_CODES = frozenset(
+    {"template_outdated", "already_sending", "canceled"}
+)
 
 
 class EmailError(Exception):
@@ -226,6 +229,13 @@ class EmailWarning:
     fix: Optional[str] = None
 
 
+_SCHEDULING_UNSUPPORTED = EmailWarning(
+    code="scheduling_unsupported",
+    message="This PostHog predates scheduled sends, so it ignored scheduled_at and sent the email now.",
+    fix="Update PostHog before you schedule emails with it.",
+)
+
+
 @dataclass(frozen=True)
 class SendEmailResult:
     """The accepted email. ``id`` is its message ID.
@@ -234,7 +244,9 @@ class SendEmailResult:
         id: The email's message ID.
         replayed: ``True`` when the response says PostHog replayed an earlier send for
             the same ``idempotency_key`` instead of sending again.
-        warnings: Warnings PostHog returned about the request.
+        warnings: Warnings PostHog returned about the request. A send with
+            ``scheduled_at`` that a PostHog without scheduled sends delivered
+            at once adds ``scheduling_unsupported``.
         status: ``"scheduled"`` when the email waits for its ``scheduled_at``,
             or ``"queued"`` when it goes out now.
         scheduled_at: When a scheduled email goes out, in UTC. ``None`` for an
@@ -433,6 +445,7 @@ class _PreparedSend:
     body: bytes
     idempotency_key: str
     max_retries: Optional[int] = None
+    scheduled: bool = False
 
 
 def _prepare(args: Mapping[str, Any]) -> _PreparedSend:
@@ -498,6 +511,7 @@ def _prepare_code_send(args: Mapping[str, Any]) -> _PreparedSend:
         body=_encode(payload),
         idempotency_key=idempotency_key,
         max_retries=max_retries,
+        scheduled="scheduled_at" in payload,
     )
 
 
@@ -580,6 +594,7 @@ def _prepare_send(
         body=json.dumps(payload, cls=DatetimeSerializer).encode("utf-8"),
         idempotency_key=idempotency_key,
         max_retries=max_retries,
+        scheduled="scheduled_at" in payload,
     )
 
 
@@ -656,11 +671,11 @@ def _prepare_update(scheduled_at: Any) -> bytes:
 
 
 def _list_path(status: Any, cursor: Any, limit: Any) -> str:
-    if not isinstance(status, str):
-        raise TypeError("status must be a string, such as 'scheduled'.")
-    if not status:
-        raise ValueError("status must not be empty.")
-    query: List[tuple[str, str]] = [("status", status)]
+    if status != "scheduled":
+        raise ValueError(
+            "status must be 'scheduled': only scheduled emails can be listed."
+        )
+    query: list[tuple[str, str]] = [("status", status)]
     if cursor is not None:
         if not isinstance(cursor, str):
             raise TypeError("cursor must be the next value of the previous page.")
@@ -718,7 +733,7 @@ def _error_name(status: Optional[int], code: str) -> str:
     if status == 409:
         return (
             "validation_error"
-            if code == "template_outdated"
+            if code in _SETTLED_CONFLICT_CODES
             else "concurrent_idempotent_requests"
         )
     if status == 422:
@@ -751,7 +766,10 @@ def _retry_after(header: Optional[str]) -> Optional[float]:
 
 
 def _parse_response(
-    status: int, payload: Any, retry_after_header: Optional[str]
+    status: int,
+    payload: Any,
+    retry_after_header: Optional[str],
+    scheduled: bool = False,
 ) -> SendEmailResult:
     fields = payload if isinstance(payload, dict) else {}
     if 200 <= status < 300:
@@ -761,23 +779,27 @@ def _parse_response(
                 "api_error", "PostHog answered without a message ID.", status
             )
         warnings = fields.get("warnings")
+        server_warnings = tuple(
+            EmailWarning(
+                code=_text(warning.get("code")) or "warning",
+                message=_text(warning.get("detail"))
+                or _text(warning.get("message"))
+                or "",
+                fix=_text(warning.get("fix")),
+            )
+            for warning in (warnings if isinstance(warnings, list) else [])
+            if isinstance(warning, dict)
+        )
+        # A PostHog without scheduled sends drops scheduled_at, sends at once and answers no status.
+        unscheduled = (
+            (_SCHEDULING_UNSUPPORTED,) if scheduled and "status" not in fields else ()
+        )
         return SendEmailResult(
             id=message_id,
             replayed=fields.get("replayed") is True,
-            # A server without scheduled sends answers no status: it queued the email.
             status=_text(fields.get("status")) or "queued",
             scheduled_at=_timestamp(fields.get("scheduled_at")),
-            warnings=tuple(
-                EmailWarning(
-                    code=_text(warning.get("code")) or "warning",
-                    message=_text(warning.get("detail"))
-                    or _text(warning.get("message"))
-                    or "",
-                    fix=_text(warning.get("fix")),
-                )
-                for warning in (warnings if isinstance(warnings, list) else [])
-                if isinstance(warning, dict)
-            ),
+            warnings=server_warnings + unscheduled,
         )
     raise _response_error(fields, status, retry_after_header)
 
@@ -844,12 +866,11 @@ def _parse_cancel_response(
 ) -> CancelEmailResult:
     fields = _success_fields(status, payload, retry_after_header)
     message_id = _text(fields.get("id"))
-    email_status = _text(fields.get("status"))
-    if message_id is None or email_status is None:
+    if message_id is None or fields.get("status") != "canceled":
         raise _sdk_error(
-            "api_error", "PostHog answered without the email's status.", status
+            "api_error", "PostHog answered without a canceled email.", status
         )
-    return CancelEmailResult(id=message_id, status=email_status)
+    return CancelEmailResult(id=message_id, status="canceled")
 
 
 def _parse_list_response(
@@ -885,7 +906,13 @@ def _parse_list_response(
                 id=message_id, to=to, scheduled_at=scheduled_at, created_at=created_at
             )
         )
-    return EmailList(results=tuple(results), next=_text(fields.get("next")))
+    next_cursor = fields.get("next", "")
+    if next_cursor is not None and _text(next_cursor) is None:
+        # Losing the cursor would end a caller's paging loop early.
+        raise _sdk_error(
+            "api_error", "PostHog answered without the next page's cursor.", status
+        )
+    return EmailList(results=tuple(results), next=next_cursor)
 
 
 def _success_fields(
@@ -916,7 +943,7 @@ def _response_error(
 
 def _is_retryable(error: EmailError) -> bool:
     # A send carries an idempotency key, so a retry can't queue a second email. A status read changes nothing.
-    # A cancel or a move ends the same however often it lands, so it needs no key.
+    # A cancel, or a move to a later time, ends the same however often it lands, so it needs no key.
     if error.status_code is None:
         return error.code == "network_error"
     if error.status_code == 409:
@@ -1176,7 +1203,9 @@ def _deliver(self, prepared: _PreparedSend, secret_key: str) -> SendEmailResult:
             prepared.path,
             prepared.body,
             headers,
-            _parse_response,
+            lambda status, payload, retry_after: _parse_response(
+                status, payload, retry_after, prepared.scheduled
+            ),
             prepared.max_retries,
         )
 
@@ -1220,8 +1249,11 @@ def update(
         Only the time changes. To change the content, the recipient or the
         sender, :meth:`cancel` the email and send it again.
 
-        Retries like :meth:`send`. A move ends the same however often it
-        lands, so it takes no idempotency key.
+        Retries like :meth:`send`. A move to a later time ends the same
+        however often it lands, so it takes no idempotency key. A move to now
+        starts the send, so when its answer is lost, the retry can raise
+        ``already_sending`` for an email that was moved: read it with
+        :meth:`get`.
 
         Args:
             message_id: The ``id`` that :meth:`send` returned.
@@ -1237,7 +1269,8 @@ def update(
                 ``already_sending`` once PostHog has started sending it,
                 ``canceled`` after a cancel, ``invalid_scheduled_at`` for a time
                 too far ahead, and ``not_found`` for an unknown ID or an email
-                that went out.
+                that went out. ``name`` is ``validation_error`` for the first
+                three: a retry can't change them.
             TypeError, ValueError: An argument has the wrong type or format.
 
         Examples:
@@ -1270,8 +1303,9 @@ def cancel(self, message_id: Union[str, UUID]) -> CancelEmailResult:
 
         Raises:
             EmailError: The email was not canceled. ``code`` is
-                ``already_sending`` once PostHog has started sending it, and
-                ``not_found`` for an unknown ID or an email that went out.
+                ``already_sending``, named ``validation_error``, once PostHog
+                has started sending it, and ``not_found`` for an unknown ID or
+                an email that went out.
             ValueError: ``message_id`` is not a message ID.
         """
         path = _cancel_path(message_id)
@@ -1285,7 +1319,7 @@ def cancel(self, message_id: Union[str, UUID]) -> CancelEmailResult:
     def list(
         self,
         *,
-        status: str = "scheduled",
+        status: Literal["scheduled"] = "scheduled",
         cursor: Optional[str] = None,
         limit: Optional[int] = None,
     ) -> EmailList:
@@ -1303,7 +1337,8 @@ def list(
             One page of scheduled emails.
 
         Raises:
-            EmailError: PostHog did not answer with a list.
+            EmailError: PostHog did not answer with a page it could read, such
+                as one without its ``next`` cursor.
             TypeError, ValueError: An argument has the wrong type or format.
 
         Examples:
@@ -1473,7 +1508,9 @@ async def _deliver(
             prepared.path,
             prepared.body,
             headers,
-            _parse_response,
+            lambda status, payload, retry_after: _parse_response(
+                status, payload, retry_after, prepared.scheduled
+            ),
             prepared.max_retries,
         )
 
@@ -1529,7 +1566,7 @@ async def cancel(self, message_id: Union[str, UUID]) -> CancelEmailResult:
     async def list(
         self,
         *,
-        status: str = "scheduled",
+        status: Literal["scheduled"] = "scheduled",
         cursor: Optional[str] = None,
         limit: Optional[int] = None,
     ) -> EmailList:
diff --git a/posthog/test/test_emails.py b/posthog/test/test_emails.py
index f77e66785..ed1a5399d 100644
--- a/posthog/test/test_emails.py
+++ b/posthog/test/test_emails.py
@@ -844,23 +844,42 @@ def test_send_schedules_both_forms_for_a_time_with_an_offset(_name, value, wire)
 
 @parameterized.expand(
     [
-        ("left_out", {}, {}),
-        ("passed_as_none", {"scheduled_at": None}, {}),
-        ("code_form_as_none", {"scheduled_at": None, "template_key": "x"}, {}),
+        ("left_out", lambda emails: send(emails)),
+        ("passed_as_none", lambda emails: send(emails, scheduled_at=None)),
+        (
+            "code_form_as_none",
+            lambda emails: send_code(emails, html=HTML, scheduled_at=None),
+        ),
     ]
 )
-def test_a_send_now_carries_no_scheduled_at_and_reads_queued(_name, overrides, _):
+def test_a_send_now_carries_no_scheduled_at_and_reads_queued(_name, call):
     emails, adapter = emails_with([accepted()])
 
-    if "template_key" in overrides:
-        result = send_code(emails, html=HTML, scheduled_at=None)
-    else:
-        result = send(emails, **overrides)
+    result = call(emails)
 
-    assert (result.status, result.scheduled_at) == ("queued", None)
+    assert (result.status, result.scheduled_at, result.warnings) == ("queued", None, ())
     assert "scheduled_at" not in json.loads(adapter.requests[0].body)
 
 
+@parameterized.expand(
+    [
+        ("template_form", lambda emails: send(emails, scheduled_at=LATER)),
+        (
+            "code_form",
+            lambda emails: send_code(emails, html=HTML, scheduled_at=LATER),
+        ),
+    ]
+)
+def test_a_server_without_scheduled_sends_warns_that_it_sent_now(_name, call):
+    # A PostHog that predates scheduled sends drops scheduled_at, sends at once and answers no status.
+    emails, _ = emails_with([accepted()])
+
+    result = call(emails)
+
+    assert (result.status, result.scheduled_at) == ("queued", None)
+    assert [w.code for w in result.warnings] == ["scheduling_unsupported"]
+
+
 def test_a_time_at_or_before_now_reads_as_the_servers_queued_answer():
     emails, _ = emails_with([accepted(status="queued", scheduled_at=None)])
 
@@ -990,6 +1009,7 @@ def test_list_reads_scheduled_emails_a_page_at_a_time(_name, args, query, next_c
         ("a_bool", {"limit": True}, TypeError),
         ("empty_cursor", {"cursor": ""}, ValueError),
         ("empty_status", {"status": ""}, ValueError),
+        ("another_status", {"status": "sent"}, ValueError),
     ]
 )
 def test_list_refuses_bad_arguments_before_any_request(_name, args, error):
@@ -1077,19 +1097,26 @@ def test_update_and_cancel_retry_like_a_send(
         assert len(adapter.requests) == 1
 
 
-@parameterized.expand([("already_sending",), ("canceled",)])
-def test_a_cancel_or_move_too_late_keeps_posthogs_code(code):
-    emails, _ = emails_with([rejected(409, code)])
+@parameterized.expand(
+    [
+        ("move_while_sending", MANAGE_CALLS[0][1], "already_sending"),
+        ("move_after_a_cancel", MANAGE_CALLS[0][1], "canceled"),
+        ("cancel_while_sending", MANAGE_CALLS[1][1], "already_sending"),
+    ]
+)
+def test_a_cancel_or_move_too_late_is_a_validation_error(_name, call, code):
+    emails, adapter = emails_with([rejected(409, code)])
 
     with pytest.raises(EmailError) as raised:
-        emails.cancel(MESSAGE_ID)
+        call(emails)
 
-    # The shared SDK spec names every 409 but template_outdated this way.
+    # As with template_outdated, a retry can't resolve it, so posthog-node names it this way too.
     assert (raised.value.status_code, raised.value.code, raised.value.name) == (
         409,
         code,
-        "concurrent_idempotent_requests",
+        "validation_error",
     )
+    assert len(adapter.requests) == 1
 
 
 @parameterized.expand(
@@ -1133,10 +1160,17 @@ def test_manage_calls_need_the_secret_key_and_an_enabled_client(_name, call, cod
     assert adapter.requests == []
 
 
-def test_a_malformed_list_page_raises_instead_of_dropping_emails():
-    emails, _ = emails_with(
-        [(200, {"results": [{"id": MESSAGE_ID}], "next": None}, {})]
-    )
+@parameterized.expand(
+    [
+        ("row_without_fields", {"results": [{"id": MESSAGE_ID}], "next": None}),
+        ("next_not_a_string", {**list_reply()[1], "next": 123}),
+        ("next_empty", {**list_reply()[1], "next": ""}),
+        ("next_left_out", {"results": list_reply()[1]["results"]}),
+    ]
+)
+def test_a_malformed_list_page_raises_instead_of_dropping_emails(_name, body):
+    # A page that loses its next cursor would end a caller's loop early and hide emails that still go out.
+    emails, _ = emails_with([(200, body, {})])
 
     with pytest.raises(EmailError) as raised:
         emails.list()
@@ -1144,6 +1178,21 @@ def test_a_malformed_list_page_raises_instead_of_dropping_emails():
     assert raised.value.code == "api_error"
 
 
+@parameterized.expand(
+    [
+        ("no_status", {"id": MESSAGE_ID}),
+        ("still_scheduled", {"id": MESSAGE_ID, "status": "scheduled"}),
+    ]
+)
+def test_a_cancel_reply_without_a_canceled_email_raises(_name, body):
+    emails, _ = emails_with([(200, body, {})])
+
+    with pytest.raises(EmailError) as raised:
+        emails.cancel(MESSAGE_ID)
+
+    assert raised.value.code == "api_error"
+
+
 @pytest.mark.asyncio
 async def test_async_client_schedules_moves_cancels_and_lists():
     seen: list[httpx.Request] = []
diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt
index d8b16ae7e..6414df984 100644
--- a/references/public_api_snapshot.txt
+++ b/references/public_api_snapshot.txt
@@ -1867,13 +1867,13 @@ method posthog.contexts.ContextScope.set_session_id(session_id: str)
 method posthog.emails.AsyncEmails.aclose() -> None
 method posthog.emails.AsyncEmails.cancel(message_id: Union[str, UUID]) -> CancelEmailResult
 method posthog.emails.AsyncEmails.get(message_id: Union[str, UUID]) -> EmailStatus
-method posthog.emails.AsyncEmails.list(*, status: str = 'scheduled', cursor: Optional[str] = None, limit: Optional[int] = None) -> EmailList
+method posthog.emails.AsyncEmails.list(*, status: Literal['scheduled'] = 'scheduled', cursor: Optional[str] = None, limit: Optional[int] = None) -> EmailList
 method posthog.emails.AsyncEmails.send(**kwargs: Any) -> SendEmailResult
 method posthog.emails.AsyncEmails.update(message_id: Union[str, UUID], *, scheduled_at: Union[datetime, str]) -> UpdateEmailResult
 method posthog.emails.Emails.cancel(message_id: Union[str, UUID]) -> CancelEmailResult
 method posthog.emails.Emails.close() -> None
 method posthog.emails.Emails.get(message_id: Union[str, UUID]) -> EmailStatus
-method posthog.emails.Emails.list(*, status: str = 'scheduled', cursor: Optional[str] = None, limit: Optional[int] = None) -> EmailList
+method posthog.emails.Emails.list(*, status: Literal['scheduled'] = 'scheduled', cursor: Optional[str] = None, limit: Optional[int] = None) -> EmailList
 method posthog.emails.Emails.send(**kwargs: Any) -> SendEmailResult
 method posthog.emails.Emails.update(message_id: Union[str, UUID], *, scheduled_at: Union[datetime, str]) -> UpdateEmailResult
 method posthog.exception_capture.ExceptionCapture.capture_exception(exception, metadata=None)

From 5239890d7e7f288056bd17e4dbd5d11fdb9ed63d Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Thu, 12 Nov 2026 12:11:42 +0000
Subject: [PATCH 5/6] test(emails): a supported scheduled send carries no
 scheduling_unsupported

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 posthog/test/test_emails.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/posthog/test/test_emails.py b/posthog/test/test_emails.py
index ed1a5399d..28b91fecf 100644
--- a/posthog/test/test_emails.py
+++ b/posthog/test/test_emails.py
@@ -834,6 +834,7 @@ def test_send_schedules_both_forms_for_a_time_with_an_offset(_name, value, wire)
     for result in (template_result, code_result):
         assert result.status == "scheduled"
         assert result.scheduled_at == datetime(2026, 12, 1, 7, tzinfo=timezone.utc)
+        assert result.warnings == ()
     assert [
         json.loads(request.body)["scheduled_at"] for request in adapter.requests
     ] == [
@@ -885,7 +886,7 @@ def test_a_time_at_or_before_now_reads_as_the_servers_queued_answer():
 
     result = send(emails, scheduled_at="2020-01-01T00:00:00Z")
 
-    assert (result.status, result.scheduled_at) == ("queued", None)
+    assert (result.status, result.scheduled_at, result.warnings) == ("queued", None, ())
 
 
 @parameterized.expand(

From 7d225c30d632ab0f0c58fc63c6f1a9d46d288977 Mon Sep 17 00:00:00 2001
From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com>
Date: Thu, 12 Nov 2026 17:25:25 +0000
Subject: [PATCH 6/6] test(emails): one async test per scheduled-send behavior

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 posthog/test/test_emails.py | 142 +++++++++++++++++++++++++-----------
 1 file changed, 98 insertions(+), 44 deletions(-)

diff --git a/posthog/test/test_emails.py b/posthog/test/test_emails.py
index 64ec9c811..224278686 100644
--- a/posthog/test/test_emails.py
+++ b/posthog/test/test_emails.py
@@ -1295,32 +1295,8 @@ def test_a_cancel_reply_without_a_canceled_email_raises(_name, body):
     assert raised.value.code == "api_error"
 
 
-@pytest.mark.asyncio
-async def test_async_client_schedules_moves_cancels_and_lists():
+def async_emails_with(replies: list[httpx.Response]):
     seen: list[httpx.Request] = []
-    replies = [
-        httpx.Response(
-            202,
-            json={
-                "message_id": MESSAGE_ID,
-                "status": "scheduled",
-                "scheduled_at": "2026-12-01T07:00:00Z",
-            },
-        ),
-        httpx.Response(503, json={"code": "delivery_unavailable"}),
-        httpx.Response(
-            200,
-            json={
-                "id": MESSAGE_ID,
-                "status": "scheduled",
-                "scheduled_at": "2026-12-02T07:00:00Z",
-            },
-        ),
-        httpx.Response(503, json={"code": "delivery_unavailable"}),
-        httpx.Response(200, json={"id": MESSAGE_ID, "status": "canceled"}),
-        httpx.Response(200, json=list_reply()[1]),
-        httpx.Response(503, json={"code": "delivery_unavailable"}),
-    ]
 
     def handler(request: httpx.Request) -> httpx.Response:
         seen.append(request)
@@ -1330,34 +1306,112 @@ def handler(request: httpx.Request) -> httpx.Response:
         SECRET_KEY,
         http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
     )
+    return emails, seen
+
+
+def unavailable() -> httpx.Response:
+    return httpx.Response(503, json={"code": "delivery_unavailable"})
+
+
+@pytest.mark.asyncio
+async def test_async_send_schedules_an_email():
+    emails, seen = async_emails_with(
+        [
+            httpx.Response(
+                202,
+                json={
+                    "message_id": MESSAGE_ID,
+                    "status": "scheduled",
+                    "scheduled_at": "2026-12-01T07:00:00Z",
+                },
+            )
+        ]
+    )
+
+    sent = await emails.send(
+        to="user@example.com", template_id=TEMPLATE_ID, scheduled_at=LATER
+    )
+
+    assert (sent.status, sent.scheduled_at, sent.warnings) == (
+        "scheduled",
+        datetime(2026, 12, 1, 7, tzinfo=timezone.utc),
+        (),
+    )
+    assert [(request.method, str(request.url)) for request in seen] == [
+        ("POST", SEND_URL)
+    ]
+    assert json.loads(seen[0].content)["scheduled_at"] == "2026-12-01T07:00:00Z"
+
+
+@pytest.mark.asyncio
+async def test_async_update_retries_an_unavailable_service_and_moves_the_email():
+    emails, seen = async_emails_with(
+        [
+            unavailable(),
+            httpx.Response(
+                200,
+                json={
+                    "id": MESSAGE_ID,
+                    "status": "scheduled",
+                    "scheduled_at": "2026-12-02T07:00:00Z",
+                },
+            ),
+        ]
+    )
 
     with mock.patch("posthog.emails.asyncio.sleep"):
-        sent = await emails.send(
-            to="user@example.com", template_id=TEMPLATE_ID, scheduled_at=LATER
-        )
         moved = await emails.update(
             MESSAGE_ID, scheduled_at="2026-12-02T09:00:00+02:00"
         )
-        canceled = await emails.cancel(MESSAGE_ID)
-        page = await emails.list(limit=10)
-        with pytest.raises(EmailError):
-            await emails.list()
 
-    assert (sent.status, moved.scheduled_at, canceled.status) == (
-        "scheduled",
-        datetime(2026, 12, 2, 7, tzinfo=timezone.utc),
-        "canceled",
-    )
-    assert [email.id for email in page.results] == [MESSAGE_ID]
+    assert moved.scheduled_at == datetime(2026, 12, 2, 7, tzinfo=timezone.utc)
     assert [(request.method, str(request.url)) for request in seen] == [
-        ("POST", SEND_URL),
         ("PATCH", STATUS_URL),
         ("PATCH", STATUS_URL),
+    ]
+    assert json.loads(seen[0].content) == {"scheduled_at": "2026-12-02T09:00:00+02:00"}
+    assert "Idempotency-Key" not in seen[0].headers
+
+
+@pytest.mark.asyncio
+async def test_async_cancel_retries_an_unavailable_service_and_cancels_the_email():
+    emails, seen = async_emails_with(
+        [
+            unavailable(),
+            httpx.Response(200, json={"id": MESSAGE_ID, "status": "canceled"}),
+        ]
+    )
+
+    with mock.patch("posthog.emails.asyncio.sleep"):
+        canceled = await emails.cancel(MESSAGE_ID)
+
+    assert canceled.status == "canceled"
+    assert [(request.method, str(request.url)) for request in seen] == [
         ("POST", CANCEL_URL),
         ("POST", CANCEL_URL),
-        ("GET", LIST_URL + "?status=scheduled&limit=10"),
-        ("GET", LIST_URL + "?status=scheduled"),
     ]
-    assert json.loads(seen[0].content)["scheduled_at"] == "2026-12-01T07:00:00Z"
-    assert json.loads(seen[1].content) == {"scheduled_at": "2026-12-02T09:00:00+02:00"}
-    assert "Idempotency-Key" not in seen[1].headers
+
+
+@pytest.mark.asyncio
+async def test_async_list_reads_a_page():
+    emails, seen = async_emails_with([httpx.Response(200, json=list_reply()[1])])
+
+    page = await emails.list(limit=10)
+
+    assert [email.id for email in page.results] == [MESSAGE_ID]
+    assert [str(request.url) for request in seen] == [
+        LIST_URL + "?status=scheduled&limit=10"
+    ]
+
+
+@pytest.mark.asyncio
+async def test_async_list_never_retries():
+    emails, seen = async_emails_with(
+        [unavailable(), httpx.Response(200, json=list_reply()[1])]
+    )
+
+    with mock.patch("posthog.emails.asyncio.sleep"):
+        with pytest.raises(EmailError):
+            await emails.list()
+
+    assert len(seen) == 1