Merge pull request #51 from convertcom/fix/bucketing-activation-events-parity

fix: emit experience-activation (bucketing) tracking events with per-call enable_tracking control

Author: abbaseya

src/convert_sdk/context.py

@@ -416,6 +416,7 @@ def run_experience(
         *,
         attributes: Optional[Mapping[str, Any]] = None,
         location_attributes: Optional[Mapping[str, Any]] = None,
+        enable_tracking: bool = True,
     ) -> Optional[ExperienceResult]:
         """Evaluate a single experience by key for this visitor.
 
@@ -425,6 +426,19 @@ def run_experience(
         qualifies and buckets into a variation, or ``None`` for any normal miss
         (missing experience, unqualified visitor, no active variation). Never
         raises for normal evaluation outcomes and performs no network I/O.
+
+        Args:
+            experience_key: The experience key to evaluate.
+            attributes: Optional per-call visitor attribute overlay (ephemeral).
+            location_attributes: Optional per-call location attribute overlay
+                (ephemeral).
+            enable_tracking: When ``True`` (the default), a successfully-bucketed
+                result enqueues a bucketing activation event via the tracker
+                (Story 2.5). When ``False``, no bucketing event is enqueued and
+                no ``LifecycleEvent.BUCKETING`` is emitted. The returned
+                :class:`~convert_sdk.domain.results.ExperienceResult` is
+                identical regardless of this flag — it gates tracking only, not
+                evaluation. Must be passed as a keyword argument.
         """
         visitor_attributes = self._state.with_overlay(attributes)
         location = self._merge(self._location_attributes, location_attributes)
@@ -436,20 +450,36 @@ def run_experience(
             location_attributes=location,
         )
         self._log_bucketing(result)
+        if result is not None and enable_tracking and self._tracker is not None:
+            self._tracker.track_bucketing(
+                visitor_id=self._state.visitor_id,
+                experience_id=result.experience_id,
+                variation_id=result.variation_id,
+            )
         return result
 
     def run_experiences(
         self,
         *,
         attributes: Optional[Mapping[str, Any]] = None,
         location_attributes: Optional[Mapping[str, Any]] = None,
+        enable_tracking: bool = True,
     ) -> List[ExperienceResult]:
         """Evaluate all applicable experiences for this visitor.
 
         Returns the list of typed results for the experiences the visitor
         qualifies for and buckets into; experiences that do not resolve are
         omitted (no ``None`` entries). Evaluation stays local to the snapshot —
         no network I/O.
+
+        Args:
+            attributes: Optional per-call visitor attribute overlay (ephemeral).
+            location_attributes: Optional per-call location attribute overlay
+                (ephemeral).
+            enable_tracking: When ``True`` (the default), each successfully-bucketed
+                result enqueues a bucketing activation event via the tracker
+                (Story 2.5). When ``False``, no bucketing events are enqueued
+                for any resolved experience. Must be passed as a keyword argument.
         """
         visitor_attributes = self._state.with_overlay(attributes)
         location = self._merge(self._location_attributes, location_attributes)
@@ -468,6 +498,12 @@ def run_experiences(
             if result is not None:
                 results.append(result)
                 self._log_bucketing(result)
+                if enable_tracking and self._tracker is not None:
+                    self._tracker.track_bucketing(
+                        visitor_id=self._state.visitor_id,
+                        experience_id=result.experience_id,
+                        variation_id=result.variation_id,
+                    )
         return results
 
     # --- feature resolution ------------------------------------------------

src/convert_sdk/domain/results.py

@@ -308,6 +308,28 @@ class EntityDiagnostic(_Diagnostic):
     """
 
 
+@dataclass(frozen=True)
+class BucketingEvent:
+    """An in-process bucketing activation event tied to a visitor and experience.
+
+    Story 2.5 creates this locally when a visitor is bucketed into a variation —
+    it carries the stable identity fields needed for later payload shaping. Parallel
+    to :class:`ConversionEvent` but simpler: no revenue, no goalData, no segments
+    here (those are conversion-only concerns). All fields stay **snake_case internal**
+    — no wire-name mapping happens on this value object (that is the serializer's
+    sole job, per the architecture's data boundary).
+
+    Attributes:
+        visitor_id: The visitor the bucketing is attributed to.
+        experience_id: The resolved experience's id (stable downstream identity).
+        variation_id: The selected variation's id.
+    """
+
+    visitor_id: str
+    experience_id: str
+    variation_id: str
+
+
 @dataclass(frozen=True)
 class CustomSegmentsResult:
     """The typed outcome of :meth:`convert_sdk.context.Context.run_custom_segments`.

src/convert_sdk/events.py

@@ -10,9 +10,10 @@
 Enum identifiers are frozen by the PRD (``prd.md#API-Surface``) and aligned with
 the JS ``SystemEvents`` parity subset:
 
-* ``READY`` / ``CONFIG_UPDATED`` / ``BUCKETING`` are defined for completeness and
-  JS parity but are emitted by the initialization/config and evaluation layers —
-  Story 2.4 does NOT emit them (Critical Warning #12).
+* ``READY`` / ``CONFIG_UPDATED`` are defined for completeness and JS parity but
+  are emitted by the initialization/config layers.
+* ``BUCKETING`` is emitted from the tracking enqueue path (Story 2.5) after a
+  visitor is bucketed into a variation and the bucketing event is enqueued.
 * ``CONVERSION`` is emitted from the tracking enqueue path on a tracked
   (non-suppressed) conversion.
 * ``API_QUEUE_RELEASED`` is emitted from the single shared release path on every
@@ -74,6 +75,21 @@ class ConversionEventPayload:
     goal_key: str
 
 
+@dataclass(frozen=True)
+class BucketingEventPayload:
+    """Domain-relevant context for a ``BUCKETING`` lifecycle event (Story 2.5).
+
+    Carries ONLY internal snake_case domain identity fields — never raw visitor
+    attributes, the wire payload, or any transport object (NFR6). Built directly
+    from the in-process bucketing event, not from the wire serializer, so emission
+    stays off the serialization path. Parallel to :class:`ConversionEventPayload`.
+    """
+
+    visitor_id: str
+    experience_id: str
+    variation_id: str
+
+
 @dataclass(frozen=True)
 class QueueReleasedPayload:
     """Diagnostic context for an ``API_QUEUE_RELEASED`` lifecycle event.

src/convert_sdk/tracking/deduplication.py

@@ -83,6 +83,48 @@ class DedupDecision:
     already_tracked: bool
 
 
+def bucketing_marker_key(visitor_id: str, experience_id: str) -> str:
+    """Build the DataStore key for the ``(visitor_id, experience_id)`` bucketing marker.
+
+    Uses a collision-safe namespaced key
+    ``f"bucketing:{json.dumps([visitor_id, experience_id])}"``. JSON serialization of a
+    two-element list guarantees no collision regardless of separator characters in the
+    values — e.g. ``("a:b", "c")`` → ``'bucketing:["a:b", "c"]'`` and
+    ``("a", "b:c")`` → ``'bucketing:["a", "b:c"]'`` are guaranteed distinct, where a
+    naive ``f"{visitor_id}:{experience_id}"`` composite would collide. Parallel to
+    :func:`goal_marker_key` but uses the ``bucketing:`` prefix to keep concerns separate.
+    """
+    return f"bucketing:{json.dumps([visitor_id, experience_id])}"
+
+
+def evaluate_bucketing_dedup(
+    store: DataStore,
+    *,
+    visitor_id: str,
+    experience_id: str,
+) -> bool:
+    """Check and persist a bucketing deduplication marker for ``(visitor_id, experience_id)``.
+
+    Returns ``True`` (should enqueue) when the marker is absent (first time for this
+    visitor+experience pair) and persists the marker via ``store.set(key, True)``.
+    Returns ``False`` (suppress) when the marker is already present.
+
+    Bucketing dedup has no ``force_multiple`` / ``goalData`` analog — it is a plain
+    "first time?" check scoped to ``(visitor_id, experience_id)``. This satisfies
+    Story 2.5 AC#3: client-side dedup ensures each visitor-experience pair produces
+    at most one bucketing event per DataStore scope. Note that the JS SDK enqueues
+    bucketing events unconditionally and relies on server-side ``enrichData`` — the
+    Python SDK honors the story contract with an explicit client-side dedup marker
+    instead, consistent with the Python SDK's ``DataStore``-backed dedup approach
+    for conversions.
+    """
+    key = bucketing_marker_key(visitor_id, experience_id)
+    if store.has(key):
+        return False
+    store.set(key, True)
+    return True
+
+
 def evaluate_dedup(
     store: DataStore,
     *,

src/convert_sdk/tracking/payloads.py

@@ -38,7 +38,7 @@
 
 from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
-from convert_sdk.domain.results import ConversionEvent
+from convert_sdk.domain.results import BucketingEvent, ConversionEvent
 
 if TYPE_CHECKING:  # pragma: no cover - typing only
     from convert_sdk.domain.config_snapshot import ConfigSnapshot
@@ -188,6 +188,61 @@ def _build_visitor_entry(event: ConversionEvent) -> Dict[str, Any]:
     return visitor
 
 
+def build_bucketing_payload(
+    snapshot: "ConfigSnapshot",
+    event: BucketingEvent,
+    *,
+    data_store: Optional[Any] = None,
+) -> Dict[str, Any]:
+    """Assemble the JS-SDK outbound bucketing-event payload for one bucketing activation.
+
+    Wire contract anchored against ``types.gen.ts:2467-2473`` (VisitorTrackingEvents
+    wrapper) and ``types.gen.ts:2486-2497`` (BucketingEvent body) and
+    ``architecture.md §"Bucketing Event Tracking Format"``:
+
+        {eventType: "bucketing", data: {experienceId: str, variationId: str}}
+
+    No ``timestamp``, no extra keys beyond ``eventType`` and ``data``, no ``segments``
+    on the visitor entry (bucketing events carry no segment attribution). The stable
+    envelope fields (``accountId`` / ``projectId`` / ``source`` / ``enrichData``) are
+    computed identically to :func:`build_tracking_payload` so a mixed batch serializes
+    coherently through :meth:`~convert_sdk.tracking.tracker.Tracker._build_batch_payload`.
+
+    Args:
+        snapshot: The immutable config snapshot supplying ``accountId`` /
+            ``projectId``.
+        event: The internal snake_case :class:`~convert_sdk.domain.results.BucketingEvent`
+            to serialize.
+        data_store: The configured DataStore, if any. ``enrichData`` is computed as
+            ``data_store is None`` (F-002 parity with :func:`build_tracking_payload`).
+
+    Returns:
+        A JSON-serializable ``dict`` matching the JS-SDK ``SendTrackingEventsRequestData``
+        envelope with a single bucketing event entry. In-memory only — no batching,
+        queue, or network I/O happens here.
+    """
+    return {
+        "accountId": snapshot.account_id,
+        "projectId": snapshot.project_id,
+        "enrichData": data_store is None,
+        "source": TRACKING_SOURCE,
+        "visitors": [
+            {
+                "visitorId": event.visitor_id,
+                "events": [
+                    {
+                        "eventType": "bucketing",
+                        "data": {
+                            "experienceId": str(event.experience_id),
+                            "variationId": str(event.variation_id),
+                        },
+                    }
+                ],
+            }
+        ],
+    }
+
+
 def build_tracking_payload(
     snapshot: "ConfigSnapshot",
     event: ConversionEvent,

src/convert_sdk/tracking/queue.py

@@ -30,9 +30,15 @@
 import enum
 import threading
 from dataclasses import dataclass, field
-from typing import Any, List, Mapping, Optional
+from typing import Any, List, Mapping, Optional, Union
 
-from convert_sdk.domain.results import ConversionEvent
+from convert_sdk.domain.results import BucketingEvent, ConversionEvent
+
+# Union type for all events the queue can hold (Story 2.5). Both ConversionEvent
+# and BucketingEvent carry a ``visitor_id`` field so per-visitor grouping works
+# unchanged. The queue holds snake_case domain objects only — wire serialization
+# is dispatched per-type at flush time in ``tracking/tracker.py``.
+TrackedEvent = Union[ConversionEvent, BucketingEvent]
 
 
 class ReleaseReason(str, enum.Enum):
@@ -64,7 +70,7 @@ class VisitorQueueItem:
     """
 
     visitor_id: str
-    events: List[ConversionEvent] = field(default_factory=list)
+    events: List[TrackedEvent] = field(default_factory=list)
     segments: Optional[Mapping[str, Any]] = None
 
 
@@ -125,18 +131,23 @@ def update_snapshot_metadata(
 
     def enqueue(
         self,
-        event: ConversionEvent,
+        event: TrackedEvent,
         *,
         segments: Optional[Mapping[str, Any]] = None,
     ) -> bool:
-        """Append a conversion event for its visitor; lightweight and synchronous.
+        """Append a tracked event (conversion or bucketing) for its visitor; lightweight and sync.
 
         Groups the event under its ``visitor_id`` and records the visitor's
         active ``segments`` (latest write wins). Performs no network I/O and no
         wire serialization (NFR5). Returns ``True`` when this enqueue brings the
         total event count to the configured ``batch_size`` — the signal that the
         caller should release the queue via the shared release path with
         :attr:`ReleaseReason.SIZE`. Returns ``False`` otherwise.
+
+        Both :class:`~convert_sdk.domain.results.ConversionEvent` and
+        :class:`~convert_sdk.domain.results.BucketingEvent` are accepted; they
+        coexist in the same per-visitor batch and are dispatched per-type at flush
+        time in :meth:`~convert_sdk.tracking.tracker.Tracker._build_batch_payload`.
         """
         with self._lock:
             item = self._items.get(event.visitor_id)