feat(capture): add capture_mode config scaffolding (capture v1, 1/6)

.github/workflows/sdk-compliance.yml

@@ -13,9 +13,19 @@ on:
 
 jobs:
   compliance:
-    name: PostHog SDK compliance tests
+    name: PostHog SDK compliance tests (capture v0)
     uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@02c049e529001d02f37a534745678e057d371fb0
     with:
       adapter-dockerfile: "sdk_compliance_adapter/Dockerfile"
       adapter-context: "."
       test-harness-version: "0.10.0"
+      report-name: "sdk-compliance-report-v0"
+
+  compliance-v1:
+    name: PostHog SDK compliance tests (capture v1)
+    uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@02c049e529001d02f37a534745678e057d371fb0
+    with:
+      adapter-dockerfile: "sdk_compliance_adapter/Dockerfile.v1"
+      adapter-context: "."
+      test-harness-version: "0.10.0"
+      report-name: "sdk-compliance-report-v1"

.sampo/changesets/capture-v1-mode.md

@@ -0,0 +1,9 @@
+---
+pypi/posthog: minor
+---
+
+Add an opt-in `capture_mode` for the Capture V1 ingestion protocol (`POST /i/v1/analytics/events`). Set `capture_mode="v1"` on the client (or the `POSTHOG_CAPTURE_MODE=v1` environment variable) to use Bearer auth, per-event results, and partial retry. Defaults to `"v0"` (the legacy `/batch/` endpoint), so existing setups are unaffected.
+
+When using `capture_mode="v1"`, request bodies can be compressed via `capture_compression` (or `POSTHOG_CAPTURE_COMPRESSION`): `"gzip"`, `"deflate"`, `"zstd"` (requires the optional `posthog[zstd]` extra), or `"none"` (default). The legacy `gzip=True` flag is honored as a fallback.
+
+Per-event server verdicts are surfaced through the existing `on_error` handler: events the backend explicitly drops, or fails to accept after retries, raise a `CaptureV1Error` carrying the affected event UUIDs — so a rejection is never silently lost, even when the HTTP request itself succeeded.

AGENTS.md

@@ -9,6 +9,27 @@ Guidance for coding agents working in `posthog-python`.
 - The project uses `uv` for local development. See `CONTRIBUTING.md` for setup.
 - Keep edits targeted and follow existing patterns. Prefer adding or updating tests near the behavior you change.
 
+## Capture protocol (`capture_mode`)
+
+The client supports two ingestion wire protocols, selected by `capture_mode` (precedence: explicit `Client(capture_mode=...)` kwarg > `POSTHOG_CAPTURE_MODE` env var > default).
+
+- `"v0"` (default) — legacy `POST /batch/`. Upgrades stay transparent; existing callers are unaffected.
+- `"v1"` — `POST /i/v1/analytics/events`: Bearer auth, a typed event `options` object, per-event results, and partial retry.
+
+v1 request bodies can additionally be compressed via `capture_compression` (precedence: explicit `Client(capture_compression=...)` kwarg > `POSTHOG_CAPTURE_COMPRESSION` env var > the legacy `gzip` flag > none). Supported values are `"none"`, `"gzip"`, `"deflate"` (zlib-wrapped, RFC 1950, to match the server's decoder and the Go/Rust SDKs), and `"zstd"` (requires the optional `posthog[zstd]` extra; explicit zstd without the package raises, env-var zstd warns and falls back). v0 keeps using its own `gzip` flag; `capture_compression` is v1-only.
+
+Where the pieces live:
+
+- `posthog/capture_mode.py` — the `CaptureMode` enum and `_resolve_capture_mode()` precedence logic.
+- `posthog/capture_compression.py` — the `CaptureCompression` enum and `_resolve_capture_compression()` precedence logic (with `gzip` fallback).
+- `posthog/capture_v1.py` — pure transforms (`_to_v1_event`, `_build_v1_batch_body`) and transport (`_post_v1`, `_compress_v1`, `_parse_v1_response`, `_send_v1_batch`, `CaptureV1Error`).
+- Public API surface (enforced by `references/public_api_snapshot.txt`): `CaptureMode`, `CaptureCompression` (both re-exported from `posthog`), `CaptureV1Error`, and the two env var names. Everything else in these modules is underscore-private plumbing.
+- Routing: `Consumer._send_analytics` (async) and `Client._enqueue` (sync) pick the analytics submitter by `capture_mode`. The dedicated `$ai_*` endpoint has no v1 form and always uses the legacy submitter.
+
+v1-specific behavior to preserve when editing: sentinel `$`-properties are lifted into `options` (coerced to native JSON types or omitted — a wrong type 400s the whole batch); top-level `$set`/`$set_once` are relocated into `properties`; only events the server tags `retry` are resent (stable `PostHog-Request-Id`/`created_at`, incrementing `PostHog-Attempt`); a server `drop` is a terminal per-event rejection — drops are accumulated across attempts and surfaced via `CaptureV1Error`/`on_error` even on a 2xx with no retries (a success status is not full delivery); `Retry-After` is a *minimum*, not a replacement (the client waits `max(configured_backoff, min(Retry-After, _MAX_BACKOFF_SECONDS))`); `_MAX_BACKOFF_SECONDS` (30s) is the single ceiling for both the exponential backoff and the `Retry-After` clamp; `429` is terminal.
+
+Retry blocking matches v0: in the default async mode retries happen on the background consumer thread, but with `sync_mode=True` the partial-retry loop (including its backoff sleeps) runs inline on the calling thread, so a slow/erroring endpoint blocks the caller until retries are exhausted.
+
 ## Validation
 
 Useful checks:

posthog/__init__.py

@@ -9,6 +9,8 @@
     OptionalCaptureArgs,
     OptionalSetArgs,
 )
+from posthog.capture_compression import CaptureCompression as CaptureCompression
+from posthog.capture_mode import CaptureMode as CaptureMode
 from posthog.client import Client
 from posthog.exception_capture import ExceptionCapture
 from posthog.contexts import (
@@ -381,6 +383,9 @@ def get_tags() -> Dict[str, Any]:
 # We recommend setting this to False if you are only using the personalApiKey for evaluating remote config payloads via `get_remote_config_payload` and not using local evaluation.
 enable_local_evaluation = True  # type: bool
 flag_definition_cache_provider = None  # type: Optional[FlagDefinitionCacheProvider]
+# Capture wire protocol for the global client. None defers to POSTHOG_CAPTURE_MODE
+# then CaptureMode.V0. See posthog.capture_mode.CaptureMode.
+capture_mode = None  # type: Optional[CaptureMode]
 
 default_client = None  # type: Optional[Client]
 
@@ -1180,6 +1185,7 @@ def setup() -> Client:
             exception_autocapture_bucket_size=exception_autocapture_bucket_size,
             exception_autocapture_refill_rate=exception_autocapture_refill_rate,
             exception_autocapture_refill_interval_seconds=exception_autocapture_refill_interval_seconds,
+            capture_mode=capture_mode,
         )
 
     # Always set in case user changes it. Preserve Client's auto-disabled state

posthog/capture_compression.py

@@ -0,0 +1,128 @@
+import logging
+import os
+from enum import Enum
+from typing import Any, Optional, Union
+
+_zstandard: Any | None
+try:
+    import zstandard
+
+    _zstandard = zstandard
+except ImportError:
+    _zstandard = None
+
+__all__ = ["CAPTURE_COMPRESSION_ENV_VAR", "CaptureCompression"]
+
+log = logging.getLogger("posthog")
+
+CAPTURE_COMPRESSION_ENV_VAR = "POSTHOG_CAPTURE_COMPRESSION"
+
+
+class CaptureCompression(str, Enum):
+    """Selects the request-body compression for capture-v1 uploads.
+
+    Only honored when ``capture_mode`` is ``V1``; the legacy ``/batch/`` path
+    keeps using its own ``gzip`` flag. ``NONE`` sends the body uncompressed.
+    ``GZIP`` and ``DEFLATE`` (zlib, RFC 1950) are both stdlib / zero-dependency;
+    ``ZSTD`` is faster and compresses better but needs the optional zstandard
+    package (``pip install posthog[zstd]``) until stdlib support lands in
+    Python 3.14. Each maps to the matching ``Content-Encoding`` token the v1
+    server decodes (``br`` is accepted by the server too but is intentionally
+    left out for now). Inheriting from ``str`` keeps the members comparable to
+    and serializable as their token values.
+    """
+
+    NONE = "none"
+    GZIP = "gzip"
+    DEFLATE = "deflate"
+    ZSTD = "zstd"
+
+
+# Accepted spellings for both the kwarg and the env var. ``identity`` mirrors
+# the HTTP token for "no encoding".
+_ALIASES: dict[str, CaptureCompression] = {
+    "none": CaptureCompression.NONE,
+    "identity": CaptureCompression.NONE,
+    "gzip": CaptureCompression.GZIP,
+    "deflate": CaptureCompression.DEFLATE,
+    "zstd": CaptureCompression.ZSTD,
+}
+
+
+def _zstd_available() -> bool:
+    return _zstandard is not None
+
+
+def _coerce_explicit(
+    value: Union[CaptureCompression, str],
+) -> CaptureCompression:
+    """Normalize an explicitly-supplied compression to a ``CaptureCompression``.
+
+    An explicit but unrecognized value is a programming error, so it raises
+    ``ValueError`` rather than silently defaulting (unlike the env var, which is
+    operator-supplied and defaults defensively).
+    """
+    if isinstance(value, CaptureCompression):
+        return value
+    if isinstance(value, str):
+        resolved = _ALIASES.get(value.strip().lower())
+        if resolved is not None:
+            return resolved
+    raise ValueError(
+        f"invalid capture_compression {value!r}; expected a CaptureCompression "
+        f"or one of {sorted(_ALIASES)}"
+    )
+
+
+def _resolve_capture_compression(
+    capture_compression: Optional[Union[CaptureCompression, str]] = None,
+    *,
+    gzip_fallback: bool = False,
+) -> CaptureCompression:
+    """Resolve the effective v1 compression.
+
+    Precedence: explicit ``capture_compression`` argument >
+    ``POSTHOG_CAPTURE_COMPRESSION`` env var > the legacy ``gzip`` flag
+    (``GZIP`` when set) > ``NONE``. An unrecognized env value logs a warning and
+    falls back to the ``gzip`` flag, so a typo never silently changes encoding.
+
+    ``ZSTD`` requires the optional zstandard package: explicitly requesting it
+    without the package raises ``ValueError`` (programming error, fail loud),
+    while requesting it via the env var warns and falls back (operator-supplied
+    config must never silently break capture).
+    """
+    if capture_compression is not None:
+        resolved = _coerce_explicit(capture_compression)
+        if resolved is CaptureCompression.ZSTD and not _zstd_available():
+            raise ValueError(
+                "capture_compression 'zstd' requires the zstandard package; "
+                "install posthog[zstd]"
+            )
+        return resolved
+
+    fallback = CaptureCompression.GZIP if gzip_fallback else CaptureCompression.NONE
+
+    raw = os.environ.get(CAPTURE_COMPRESSION_ENV_VAR)
+    if raw is None or raw.strip() == "":
+        return fallback
+
+    env_resolved = _ALIASES.get(raw.strip().lower())
+    if env_resolved is None:
+        log.warning(
+            "Unrecognized %s=%r; falling back to %s. Expected one of %s.",
+            CAPTURE_COMPRESSION_ENV_VAR,
+            raw,
+            fallback.value,
+            sorted(_ALIASES),
+        )
+        return fallback
+    if env_resolved is CaptureCompression.ZSTD and not _zstd_available():
+        log.warning(
+            "%s=%r requires the zstandard package (install posthog[zstd]); "
+            "falling back to %s.",
+            CAPTURE_COMPRESSION_ENV_VAR,
+            raw,
+            fallback.value,
+        )
+        return fallback
+    return env_resolved

posthog/capture_mode.py

@@ -0,0 +1,84 @@
+import logging
+import os
+from enum import Enum
+from typing import Optional, Union
+
+__all__ = ["CAPTURE_MODE_ENV_VAR", "CaptureMode"]
+
+log = logging.getLogger("posthog")
+
+CAPTURE_MODE_ENV_VAR = "POSTHOG_CAPTURE_MODE"
+
+
+class CaptureMode(str, Enum):
+    """Selects the capture wire protocol used for event ingestion.
+
+    ``V0`` is the legacy ``POST /batch/`` endpoint and the default, so upgrading
+    is transparent to existing callers. ``V1`` opts into
+    ``POST /i/v1/analytics/events`` (Bearer auth, per-event results, partial
+    retry). Inheriting from ``str`` keeps the members directly comparable to and
+    serializable as their ``"v0"`` / ``"v1"`` values.
+    """
+
+    V0 = "v0"
+    V1 = "v1"
+
+
+# Accepted spellings for both the explicit kwarg and the env var. Aliases mirror
+# the posthog-go naming (``legacy`` / ``analytics_v1``) so the two SDKs are
+# configured with the same vocabulary.
+_ALIASES: dict[str, CaptureMode] = {
+    "v0": CaptureMode.V0,
+    "legacy": CaptureMode.V0,
+    "v1": CaptureMode.V1,
+    "analytics_v1": CaptureMode.V1,
+}
+
+
+def _coerce_explicit(value: Union[CaptureMode, str]) -> CaptureMode:
+    """Normalize an explicitly-supplied capture mode to a ``CaptureMode``.
+
+    Accepts a ``CaptureMode`` or one of the string aliases. An explicit but
+    unrecognized value is a programming error, so it raises ``ValueError`` rather
+    than silently defaulting (unlike the env var, which is operator-supplied and
+    defaults defensively).
+    """
+    if isinstance(value, CaptureMode):
+        return value
+    if isinstance(value, str):
+        resolved = _ALIASES.get(value.strip().lower())
+        if resolved is not None:
+            return resolved
+    raise ValueError(
+        f"invalid capture_mode {value!r}; expected a CaptureMode or one of "
+        f"{sorted(_ALIASES)}"
+    )
+
+
+def _resolve_capture_mode(
+    capture_mode: Optional[Union[CaptureMode, str]] = None,
+) -> CaptureMode:
+    """Resolve the effective capture mode.
+
+    Precedence: explicit ``capture_mode`` argument > ``POSTHOG_CAPTURE_MODE`` env
+    var > ``CaptureMode.V0``. An unrecognized env value logs a warning and falls
+    back to ``V0`` so a typo never silently flips the wire protocol.
+    """
+    if capture_mode is not None:
+        return _coerce_explicit(capture_mode)
+
+    raw = os.environ.get(CAPTURE_MODE_ENV_VAR)
+    if raw is None or raw.strip() == "":
+        return CaptureMode.V0
+
+    resolved = _ALIASES.get(raw.strip().lower())
+    if resolved is None:
+        log.warning(
+            "Unrecognized %s=%r; falling back to %s. Expected one of %s.",
+            CAPTURE_MODE_ENV_VAR,
+            raw,
+            CaptureMode.V0.value,
+            sorted(_ALIASES),
+        )
+        return CaptureMode.V0
+    return resolved