feat(flags): send $experiment_exposure for v2 experiment splits

.sampo/changesets/experiment-exposure.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Send `$experiment_exposure` for config version 2 experiment splits, and add rule attribution to `$feature_flag_called` for config version 2 flags. A read that sends `$feature_flag_called` now also sends `$experiment_exposure` when the flag was evaluated remotely and its config version 2 record is a complete experiment split: reason code `experiment_split`, rule type `experiment`, a rule ID and a variant key. This applies to `get_feature_flag()`, `feature_enabled()`, `get_feature_flag_result()`, and the `is_enabled()` and `get_flag()` reads of an `evaluate_flags()` snapshot. The exposure carries `$feature_flag`, the same `$feature_flag_response` as its `$feature_flag_called`, `$feature_flag_variant`, `$feature_flag_rule_id` and `locally_evaluated: false`, plus the flag ID, flag version, request ID and evaluation time when the response has them. It does not carry the config version, reason code or rule type, which stay on `$feature_flag_called`. Minimal-event mode does not apply to the exposure. The SDK sends one exposure per distinct ID, groups, flag, rule and variant. The value, request ID and evaluation time are not part of this dedupe, a flags reload does not reset it, and it is separate from the `$feature_flag_called` dedupe. No other outcome sends an exposure: not a targeting match, rollout miss, paused experiment, holdout or failed record, and not a config version 1 flag, a flag from a server without typed values or a locally evaluated flag. A config version 2 `$feature_flag_called` carries `$feature_flag_config_version` 2 and `$feature_flag_reason_code`, which replaces `$feature_flag_reason`. A complete experiment split also adds `$feature_flag_rule_id`, `$feature_flag_rule_type` and `$feature_flag_variant`. A targeting match (rule type `targeted_release` or `percentage_rollout`), a rollout miss (`percentage_rollout` or `experiment`), a paused experiment or a holdout (`experiment`) adds `$feature_flag_rule_id` and `$feature_flag_rule_type`. Other reasons and failed records add no rule properties. The rule ID and rule type are sent as a pair or not at all: when a record has only one of them, a rule type its reason does not allow, or a rule ID that is not a hyphenated UUID, the event drops both and keeps the config version and reason code. A split that is missing its rule type, rule ID or variant key sends no exposure and no rule properties. A reason code the SDK does not know is sent unchanged with no rule properties, and a record without a reason code still sends its config version. None of this changes the value a read returns. The `$feature_flag_called` dedupe of a config version 2 flag includes its attribution, so equal-valued arms, successive experiments of one flag and a holdout after a variant are each reported. Minimal `$feature_flag_called` events keep the attribution. The SDK tries the two events separately. If capture of one raises or drops it (a full or closed queue, opt-out or a disabled client), the other is still sent, and the failed event keeps no dedupe entry, so a later read tries it again. An event that `before_send` drops counts as handled: it keeps its dedupe entry, and later reads do not offer it again. `get_feature_flag_payload()` sends no flag events by default; a call with `match_value` reports the caller's value, so it sends no exposure. `send_feature_flag_events=False` suppresses both events. Events of config version 1 flags, flags from servers without typed values and locally evaluated flags do not change, including their `$feature_flag_reason`, and `$feature_flag_response` does not change on any event.

posthog/args.py

@@ -5,6 +5,9 @@
     Any,
     Dict,
     FrozenSet,
+    List,
+    Mapping,
+    NamedTuple,
     Union,
     Tuple,
     Type,
@@ -65,6 +68,27 @@ class OptionalCaptureArgs(TypedDict):
     _property_allowlist: NotRequired[
         Optional[FrozenSet[str]]
     ]  # Internal: strict allowlist applied to the fully-enriched event properties. Used by minimal $feature_flag_called events.
+    _pinned_properties: NotRequired[
+        Optional["_PinnedProperties"]
+    ]  # Internal: properties fixed after super properties are merged. Used by config version 2 flag events.
+    _before_send_drops: NotRequired[
+        Optional[List[str]]
+    ]  # Internal: gets the event name when before_send drops the event. Used by flag event dedupe.
+
+
+class _PinnedProperties(NamedTuple):
+    """Internal: event properties that super properties cannot change.
+
+    Applied to the fully-enriched properties, after super properties are merged and
+    before any allowlist: ``removed`` names are dropped, then ``values`` are set.
+    """
+
+    removed: FrozenSet[str]
+    values: Mapping[str, Any]
+
+    def apply(self, properties: Dict[str, Any]) -> Dict[str, Any]:
+        kept = {k: v for k, v in properties.items() if k not in self.removed}
+        return {**kept, **self.values}
 
 
 class OptionalSetArgs(TypedDict):

posthog/async_client.py

@@ -38,8 +38,8 @@
 from .capture_mode import CaptureMode, _resolve_capture_mode
 from .client import (
     MAX_DICT_SIZE as _MAX_DICT_SIZE,
-    _MINIMAL_FLAG_CALLED_EVENT_PROPERTIES,
     Client as _SyncClient,
+    _capture_flag_events,
     add_context_tags as _add_context_tags,
     get_identity_state as _get_identity_state,
     stringify_id as _stringify_id,
@@ -77,6 +77,7 @@
 )
 from .release_id import _resolve_release_id
 from .request import QuotaLimitError, determine_server_host, normalize_host
+from .types import FeatureFlag as _FeatureFlag
 from .types import (
     FlagValue,
     _has_untyped_flag_records,
@@ -192,6 +193,9 @@ def __init__(
             0, feature_flags_request_max_retries
         )
         self.distinct_ids_feature_flags_reported = SizeLimitedDict(_MAX_DICT_SIZE, set)
+        # $experiment_exposure dedupe, kept apart from $feature_flag_called's so that a
+        # reported call never suppresses its exposure. Same bound and eviction.
+        self._experiment_exposures_reported = SizeLimitedDict(_MAX_DICT_SIZE, set)
         self._feature_flag_evaluations_host_cache: Optional[
             _FeatureFlagEvaluationsHost
         ] = None
@@ -463,6 +467,7 @@ def _prepare_event(
         msg: dict[str, Any],
         disable_geoip: Optional[bool],
         property_allowlist=None,
+        pinned_properties=None,
     ) -> tuple[Optional[dict[str, Any]], Optional[str]]:
         if self.disabled or not self._accepting:
             return None, None
@@ -494,6 +499,10 @@ def _prepare_event(
             msg["properties"].setdefault("$release_id", self._release_id)
         if self.is_server:
             msg["properties"]["$is_server"] = True
+        # After every enrichment step, so a super property never changes or adds
+        # config version 2 flag attribution.
+        if pinned_properties is not None:
+            msg["properties"] = pinned_properties.apply(msg["properties"])
         if property_allowlist is not None:
             msg["properties"] = {
                 key: value
@@ -526,7 +535,7 @@ async def _process_event(self, msg: dict[str, Any]) -> Optional[dict[str, Any]]:
 
     def _build_capture_event(
         self, event: str, kwargs: OptionalCaptureArgs
-    ) -> tuple[dict[str, Any], Optional[bool], Any]:
+    ) -> tuple[dict[str, Any], Optional[bool], Any, Any]:
         properties = {**(kwargs.get("properties") or {}), **system_context()}
         if self.capture_trace_context:
             properties = {**_get_current_otel_span_properties(), **properties}
@@ -562,18 +571,19 @@ def _build_capture_event(
             },
             kwargs.get("disable_geoip"),
             kwargs.get("_property_allowlist"),
+            kwargs.get("_pinned_properties"),
         )
 
     def capture(
         self, event: str, **kwargs: Unpack[OptionalCaptureArgs]
     ) -> Optional[str]:
         """Queue an event without blocking for network delivery."""
         try:
-            msg, disable_geoip, property_allowlist = self._build_capture_event(
-                event, kwargs
+            msg, disable_geoip, property_allowlist, pinned_properties = (
+                self._build_capture_event(event, kwargs)
             )
             prepared, sent_uuid = self._prepare_event(
-                msg, disable_geoip, property_allowlist
+                msg, disable_geoip, property_allowlist, pinned_properties
             )
             if prepared is None or sent_uuid is None:
                 return None
@@ -617,11 +627,11 @@ async def capture_immediate(
         self._immediate_callers[current] = self._immediate_callers.get(current, 0) + 1
         error_batch: list[dict[str, Any]] = []
         try:
-            msg, disable_geoip, property_allowlist = self._build_capture_event(
-                event, kwargs
+            msg, disable_geoip, property_allowlist, pinned_properties = (
+                self._build_capture_event(event, kwargs)
             )
             prepared, sent_uuid = self._prepare_event(
-                msg, disable_geoip, property_allowlist
+                msg, disable_geoip, property_allowlist, pinned_properties
             )
             if prepared is None or sent_uuid is None:
                 return None
@@ -1085,38 +1095,29 @@ def _capture_feature_flag_called_if_needed(
         disable_geoip: Optional[bool] = None,
         has_experiment: Optional[bool] = None,
         minimal_flag_called_events: bool = False,
+        details: Optional[_FeatureFlag] = None,
+        value_from_details: bool = False,
     ) -> None:
-        groups_key = (
-            tuple(sorted((str(k), str(v)) for k, v in groups.items())) if groups else ()
-        )
-        reported_key = (key, response, groups_key)
-        reported_flags = self.distinct_ids_feature_flags_reported.get(distinct_id)
-        if reported_flags is None:
-            reported_flags = set()
-            self.distinct_ids_feature_flags_reported[distinct_id] = reported_flags
-        if reported_key in reported_flags:
-            return
-
-        if has_experiment is not None:
-            properties["$feature_flag_has_experiment"] = has_experiment
-        capture_kwargs: dict[str, Any] = {}
-        if minimal_flag_called_events and has_experiment is False:
-            capture_kwargs["_property_allowlist"] = (
-                _MINIMAL_FLAG_CALLED_EVENT_PROPERTIES
-            )
         capture_groups = {
             str(group_type): _stringify_id(group_key)
             for group_type, group_key in (groups or {}).items()
         }
-        self.capture(
-            "$feature_flag_called",
+        _capture_flag_events(
+            self.capture,
+            self.distinct_ids_feature_flags_reported,
+            self._experiment_exposures_reported,
             distinct_id=distinct_id,
+            key=key,
+            response=response,
             properties=properties,
-            groups=capture_groups,
+            groups=groups,
+            capture_groups=capture_groups,
             disable_geoip=disable_geoip,
-            **capture_kwargs,
+            has_experiment=has_experiment,
+            minimal_flag_called_events=minimal_flag_called_events,
+            details=details,
+            value_from_details=value_from_details,
         )
-        reported_flags.add(reported_key)
 
     def _pending_queue_items(self) -> int:
         return int(getattr(self._queue, "_unfinished_tasks", self._queue.qsize()))
@@ -1278,6 +1279,7 @@ async def shutdown(self) -> None:
                 errors.append(error)
 
             self.distinct_ids_feature_flags_reported.clear()
+            self._experiment_exposures_reported.clear()
             self._closed = True
             if errors and self.debug:
                 raise errors[0]