convertcom · usmanabbas7 · Jun 8, 2026 · Jun 8, 2026 · Jun 8, 2026 · Jun 8, 2026
diff --git a/src/convert_sdk/__init__.py b/src/convert_sdk/__init__.py
@@ -23,6 +23,7 @@
 from convert_sdk.domain.results import (
     ConversionResult,
     ConversionStatus,
+    CustomSegmentsResult,
     ExperienceResult,
     FeatureResult,
     FeatureStatus,
@@ -63,6 +64,8 @@
     # Conversion tracking foundation (Story 2.1).
     "ConversionResult",
     "ConversionStatus",
+    # Story 3.3 custom-segment evaluation typed result (FR15).
+    "CustomSegmentsResult",
     # Lifecycle events public surface (Story 2.4): consumers need this enum to
     # call Core.on(LifecycleEvent.API_QUEUE_RELEASED, ...).
     "LifecycleEvent",

diff --git a/src/convert_sdk/context.py b/src/convert_sdk/context.py
@@ -27,12 +27,23 @@
 from typing import TYPE_CHECKING, Any, List, Mapping, Optional
 
 from convert_sdk.domain.context_state import ContextState
-from convert_sdk.domain.results import ConversionResult, ExperienceResult, FeatureResult
+from convert_sdk.domain.results import (
+    ConversionResult,
+    CustomSegmentsResult,
+    ExperienceResult,
+    FeatureResult,
+)
 from convert_sdk.evaluation.experiences import select_experience
 from convert_sdk.evaluation.features import resolve_feature, resolve_features
+from convert_sdk.evaluation.segments import select_custom_segments
 from convert_sdk.ports.storage import visitor_state_key
 from convert_sdk.tracking.conversions import create_conversion
 
+# The distinct key under which matched custom-segment IDs are recorded inside
+# the default-segment state (JS ``SegmentsKeys.CUSTOM_SEGMENTS`` parity). Kept in
+# sync with the ``customSegments`` allowlist key in ``tracking/payloads.py``.
+_CUSTOM_SEGMENTS_KEY = "customSegments"
+
 if TYPE_CHECKING:  # pragma: no cover - typing only
     from convert_sdk.domain.config_snapshot import ConfigSnapshot
     from convert_sdk.ports.storage import DataStore
@@ -64,16 +75,20 @@ def __init__(
         snapshot: "ConfigSnapshot",
         *,
         visitor_attributes: Optional[Mapping[str, Any]] = None,
+        default_segments: Optional[Mapping[str, Any]] = None,
         location_attributes: Optional[Mapping[str, Any]] = None,
         tracker: Optional["Tracker"] = None,
         data_store: Optional["DataStore"] = None,
     ) -> None:
-        # Visitor identity + stored attributes + snapshot linkage live in the
-        # typed ContextState (visitor state stays separate from the snapshot).
+        # Visitor identity + stored attributes + default segments + snapshot
+        # linkage live in the typed ContextState (visitor state stays separate
+        # from the snapshot; default segments are a DISTINCT field from raw
+        # attributes — Story 3.3, Critical Warning #7).
         self._state = ContextState(
             visitor_id=visitor_id,
             snapshot=snapshot,
             visitor_attributes=visitor_attributes,
+            default_segments=default_segments,
         )
         self._snapshot = snapshot
         # Story 2.3: shared tracking orchestrator (dedup + queue). When None,
@@ -108,6 +123,17 @@ def attributes(self) -> Mapping[str, Any]:
         """
         return self._state.visitor_attributes
 
+    @property
+    def default_segments(self) -> Mapping[str, Any]:
+        """A read-only view of this visitor's associated default segments.
+
+        Distinct from :attr:`visitor_attributes` (Story 3.3 / FR14): default
+        segments are a separate visitor-state concern that feeds reporting and
+        conversion attribution, not raw audience traits. Observable for
+        reporting/tracking without exposing the internal ``ContextState``.
+        """
+        return self._state.default_segments
+
     # --- mutable visitor state (Story 3.2) ---------------------------------
 
     def set_attributes(self, attributes: dict[str, Any]) -> None:
@@ -145,22 +171,121 @@ def set_attributes(self, attributes: dict[str, Any]) -> None:
         self._state = self._state.with_attributes(attributes)
         self._persist_visitor_state()
 
+    def set_segments(self, segments: dict[str, Any]) -> None:
+        """Persistently associate default visitor segments with this context (FR14).
+
+        Shallow-merges ``segments`` into the context's DISTINCT default-segment
+        state — new keys override touched keys, untouched keys persist — and
+        REBINDS the context's
+        :class:`~convert_sdk.domain.context_state.ContextState` to the merged
+        immutable copy (the same frozen-dataclass rebind + persist-through-store
+        pattern Story 3.2 used for :meth:`set_attributes`). The original frozen
+        state is never mutated in place, the shared immutable ``ConfigSnapshot``
+        is never touched, and the segments are kept STRICTLY SEPARATE from
+        :attr:`visitor_attributes` (Critical Warning #7).
+
+        Default segments feed reporting/conversion state — a subsequently tracked
+        conversion's ``segments`` payload reflects the visitor's active default
+        segments at conversion time — and a later
+        ``create_context(visitor_id)`` for the same visitor rehydrates them
+        through the injected ``DataStore`` (the same per-``Core`` persistence
+        boundary and visitor-scoped key Story 3.1 established). Deterministic
+        bucketing inputs (visitor identity + config snapshot) are unaffected
+        (FR25). This is the Python analogue of the JS
+        ``Context.setDefaultSegments`` → ``SegmentsManager.putSegments`` write
+        path.
+
+        Args:
+            segments: Default visitor segments to merge into the stored state.
+
+        Returns:
+            ``None``.
+        """
+        self._state = self._state.with_segments(segments)
+        self._persist_visitor_state()
+
+    def run_custom_segments(
+        self,
+        segment_keys: list[str],
+        rule_data: Optional[Mapping[str, Any]] = None,
+    ) -> CustomSegmentsResult:
+        """Evaluate custom segment matches for this visitor (FR15).
+
+        Resolves the named segments from the immutable
+        :class:`~convert_sdk.domain.config_snapshot.ConfigSnapshot` and matches
+        each segment's rule against the visitor's segment-rule input through the
+        SAME pure-Python rule engine
+        (:func:`convert_sdk.evaluation.rules.is_rule_matched`) the SDK uses for
+        audience qualification — delegating to
+        :func:`convert_sdk.evaluation.segments.select_custom_segments`.
+        Evaluation is fully LOCAL and deterministic: it reads only the loaded
+        snapshot plus visitor-scoped state and performs NO network I/O (Critical
+        Warning #5).
+
+        The per-call ``rule_data`` is an EPHEMERAL request-time overlay on the
+        visitor's stored attributes (request value > persisted state precedence,
+        reusing the Story 3.2 :meth:`ContextState.with_overlay` seam). It is
+        NEVER written back into :attr:`visitor_attributes` (AC #5); only the
+        resulting matched segment IDs are recorded — under a ``customSegments``
+        list inside the DISTINCT default-segment state (JS ``VisitorSegments``
+        parity) — and persisted through the injected ``DataStore`` so a later
+        ``create_context(visitor_id)`` rehydrates them. Already-recorded segment
+        IDs are not re-added (duplicates skipped).
+
+        Args:
+            segment_keys: The segment keys to evaluate.
+            rule_data: Optional per-call segment-rule input. Overlays the stored
+                visitor attributes for THIS call only.
+
+        Returns:
+            A typed :class:`~convert_sdk.domain.results.CustomSegmentsResult`
+            carrying the newly matched segment IDs. A normal no-match returns a
+            result with an empty ``matched_segment_ids`` — a typed, non-exception
+            outcome (never a raw dict, never raises on a normal miss).
+        """
+        # Ephemeral request-time overlay (request > persisted), reusing the
+        # Story 3.2 seam — never written back into visitor_attributes.
+        segment_rule = self._state.with_overlay(rule_data)
+
+        existing = self._state.default_segments.get(_CUSTOM_SEGMENTS_KEY) or []
+        matched = select_custom_segments(
+            self._snapshot,
+            segment_keys,
+            segment_rule,
+            existing_ids=existing,
+        )
+        if matched:
+            updated = list(existing) + matched
+            self._state = self._state.with_segments({_CUSTOM_SEGMENTS_KEY: updated})
+            self._persist_visitor_state()
+        return CustomSegmentsResult(matched_segment_ids=tuple(matched))
+
     def _persist_visitor_state(self) -> None:
         """Persist the current visitor state through the injected ``DataStore``.
 
         No-op when no store is injected (a ``Context`` constructed directly
         rather than via ``Core``). The write is visitor-scoped: it targets only
         this visitor's state key (:func:`visitor_state_key`), never another
-        visitor's and never a ``Core``-global key. The persisted value is the
-        plain merged attribute ``dict`` so a later ``create_context(visitor_id)``
-        rehydrates it through the same store. The ``DataStore`` four-method
-        surface is unchanged — a plain ``set`` of serialized state; no business
-        logic lives in the store.
+        visitor's and never a ``Core``-global key.
+
+        The persisted value is a structured envelope
+        ``{"attributes": {...}, "segments": {...}}`` so a later
+        ``create_context(visitor_id)`` round-trips BOTH the visitor attributes
+        (Story 3.2) and the default segments (Story 3.3) through the same store
+        and hydrate route. The ``DataStore`` four-method surface is unchanged —
+        a plain ``set`` of serialized state; no business logic lives in the
+        store.
         """
         if self._data_store is None:
             return None
         key = visitor_state_key(self._state.visitor_id)
-        self._data_store.set(key, dict(self._state.visitor_attributes))
+        self._data_store.set(
+            key,
+            {
+                "attributes": dict(self._state.visitor_attributes),
+                "segments": dict(self._state.default_segments),
+            },
+        )
         return None
 
     # --- evaluation surface ------------------------------------------------
@@ -349,6 +474,7 @@ def track_conversion(
                 revenue=revenue,
                 conversion_data=conversion_data,
                 visitor_attributes=self._state.visitor_attributes,
+                default_segments=self._state.default_segments,
                 force_multiple=force_multiple,
             )
         # Fallback: stateless create_conversion (no dedup/queue) for a Context
@@ -360,4 +486,5 @@ def track_conversion(
             revenue=revenue,
             conversion_data=conversion_data,
             visitor_attributes=self._state.visitor_attributes,
+            default_segments=self._state.default_segments,
         )
diff --git a/src/convert_sdk/core.py b/src/convert_sdk/core.py
@@ -136,37 +136,64 @@ def create_context(
         # context override matching keys (explicit construction wins). The read
         # is strictly visitor-scoped and goes through the DataStore protocol
         # only — Core (L4) owns the concrete store; downstream sees the protocol.
-        hydrated = self._hydrate_visitor_attributes(visitor_id, visitor_attributes)
+        hydrated, segments = self._hydrate_visitor_state(visitor_id, visitor_attributes)
         return Context(
             visitor_id,
             self._snapshot,
             visitor_attributes=hydrated,
+            default_segments=segments,
             location_attributes=location_attributes,
             tracker=self._tracker,
             data_store=self._data_store,
         )
 
-    def _hydrate_visitor_attributes(
+    def _hydrate_visitor_state(
         self,
         visitor_id: str,
         visitor_attributes: Optional[Mapping[str, Any]],
-    ) -> Optional[Mapping[str, Any]]:
-        """Merge any persisted visitor attributes with the caller-supplied ones.
-
-        Reads this visitor's persisted ``ContextState`` attributes (written by
-        :meth:`convert_sdk.context.Context.set_attributes`) through the single
-        per-Core ``DataStore`` and overlays the caller-supplied
-        ``visitor_attributes`` on top (explicit construction wins). Returns the
-        caller value unchanged when nothing is persisted, so contexts for
-        visitors that never called ``set_attributes`` behave exactly as before.
+    ) -> tuple[Optional[Mapping[str, Any]], Optional[Mapping[str, Any]]]:
+        """Rehydrate persisted visitor attributes AND default segments.
+
+        Reads this visitor's persisted ``ContextState`` envelope (written by
+        :meth:`convert_sdk.context.Context.set_attributes` /
+        :meth:`convert_sdk.context.Context.set_segments`) through the single
+        per-Core ``DataStore`` and returns ``(attributes, default_segments)``.
+
+        The persisted value is the structured envelope
+        ``{"attributes": {...}, "segments": {...}}`` (Story 3.3). For backward
+        compatibility a legacy Story 3.2 plain-attributes ``dict`` (no envelope)
+        is treated as attributes-only with empty segments. Caller-supplied
+        ``visitor_attributes`` for this fresh context overlay the persisted
+        attributes (explicit construction wins). The read is strictly
+        visitor-scoped and goes through the ``DataStore`` protocol only — Core
+        (L4) owns the concrete store; downstream sees the protocol. Returns the
+        caller value unchanged (and no segments) when nothing is persisted, so
+        contexts for visitors that never persisted state behave exactly as
+        before.
         """
         stored = self._data_store.get(visitor_state_key(visitor_id))
-        if not isinstance(stored, Mapping) or not stored:
-            return visitor_attributes
-        merged = dict(stored)
-        if visitor_attributes:
-            merged.update(visitor_attributes)
-        return merged
+        stored_attributes: Mapping[str, Any] = {}
+        stored_segments: Optional[Mapping[str, Any]] = None
+        if isinstance(stored, Mapping) and stored:
+            if "attributes" in stored or "segments" in stored:
+                # Story 3.3 structured envelope.
+                raw_attrs = stored.get("attributes")
+                stored_attributes = raw_attrs if isinstance(raw_attrs, Mapping) else {}
+                raw_segments = stored.get("segments")
+                if isinstance(raw_segments, Mapping) and raw_segments:
+                    stored_segments = dict(raw_segments)
+            else:
+                # Legacy Story 3.2 plain-attributes dict (attributes-only).
+                stored_attributes = stored
+
+        if not stored_attributes and not visitor_attributes:
+            attributes: Optional[Mapping[str, Any]] = visitor_attributes
+        else:
+            merged = dict(stored_attributes)
+            if visitor_attributes:
+                merged.update(visitor_attributes)
+            attributes = merged
+        return attributes, stored_segments
 
     # --- tracking flush ----------------------------------------------------
 

diff --git a/src/convert_sdk/domain/context_state.py b/src/convert_sdk/domain/context_state.py
@@ -45,6 +45,7 @@ class ContextState:
     visitor_id: str
     snapshot: "ConfigSnapshot"
     visitor_attributes: Mapping[str, Any] = field(default_factory=dict)
+    default_segments: Mapping[str, Any] = field(default_factory=dict)
 
     def __post_init__(self) -> None:
         # Store a defensive, read-only copy so caller-side mutation cannot leak
@@ -55,6 +56,15 @@ def __post_init__(self) -> None:
                 "visitor_attributes",
                 MappingProxyType(dict(self.visitor_attributes or {})),
             )
+        # Story 3.3: default segments are a DISTINCT visitor-state concern, kept
+        # strictly separate from visitor_attributes (Critical Warning #7). They
+        # are copied defensively and wrapped read-only, exactly like attributes.
+        if not isinstance(self.default_segments, MappingProxyType):
+            object.__setattr__(
+                self,
+                "default_segments",
+                MappingProxyType(dict(self.default_segments or {})),
+            )
 
     def with_overlay(self, overlay: Optional[Mapping[str, Any]]) -> Mapping[str, Any]:
         """Return stored visitor attributes overlaid with request-time values.
@@ -102,4 +112,38 @@ def with_attributes(self, new_attributes: Optional[Mapping[str, Any]]) -> "Conte
             visitor_id=self.visitor_id,
             snapshot=self.snapshot,
             visitor_attributes=merged,
+            default_segments=self.default_segments,
+        )
+
+    def with_segments(self, new_segments: Optional[Mapping[str, Any]]) -> "ContextState":
+        """Return a NEW :class:`ContextState` with ``new_segments`` merged in.
+
+        This is the immutable, PERSISTENT default-segment association operation
+        (Story 3.3 / FR14). It shallow-merges the stored default segments with
+        ``new_segments`` — new keys override touched keys, untouched keys persist
+        — mirroring the JS ``SegmentsManager.putSegments`` shallow-merge of the
+        stored ``segments`` with the new segment values.
+
+        The update targets ONLY the DISTINCT :attr:`default_segments` field;
+        ``visitor_attributes`` are carried through unchanged so segment state and
+        raw attribute state stay strictly separate (Critical Warning #7). The
+        original instance is never mutated: a fresh frozen ``ContextState`` is
+        returned, carrying the same ``visitor_id`` and the same
+        :class:`ConfigSnapshot` by reference (the snapshot is shared, never
+        copied or mutated per visitor — Critical Warning #10). When
+        ``new_segments`` is empty/``None`` the merge is a content-equal no-op
+        copy, preserving determinism (AC #4 / FR25).
+
+        This mirrors :meth:`with_attributes` for the segment field and is the
+        single shared segment-merge seam — callers persist the returned state
+        through the ``DataStore`` exactly as they do for attribute updates.
+        """
+        merged = dict(self.default_segments)
+        if new_segments:
+            merged.update(new_segments)
+        return ContextState(
+            visitor_id=self.visitor_id,
+            snapshot=self.snapshot,
+            visitor_attributes=self.visitor_attributes,
+            default_segments=merged,
         )