add evaluation context and target app support (#462)

<img width="654" height="135" alt="Screenshot 2025-09-08 at 4 27 56 PM"
src="https://github.com/user-attachments/assets/68267381-cfe3-4072-93ee-dc355810b56b"
/>
statsig-io/kong#3793

Author: kat-statsig

statsig/client_initialize_formatter.py

@@ -4,6 +4,7 @@
 from typing import Any, Dict, Optional, Union
 
 from .config_evaluation import _ConfigEvaluation
+from .evaluation_context import EvaluationContext
 from .spec_store import _SpecStore, EntityType
 from .statsig_metadata import _StatsigMetadata
 from .statsig_user import StatsigUser
@@ -31,8 +32,20 @@ def get_formatted_response(
             evaluator,
             hash_algo: HashingAlgorithm,
             client_sdk_key=None,
-            include_local_override=False
+            include_local_override=False,
+            target_app_id: Optional[str] = None,
     ) -> ClientInitializeResponse:
+        context = EvaluationContext(
+            target_app_id=target_app_id,
+            client_key=client_sdk_key
+        )
+        app_id = None
+        if target_app_id is not None:
+            app_id = target_app_id
+        if client_sdk_key is not None:
+            app_id = spec_store.get_target_app_for_sdk_key(client_sdk_key)
+            context.target_app_id = app_id
+
         def convert_to_entity_type(entity: str, type: str) -> Optional[EntityType]:
             if entity == "layer":
                 return EntityType.LAYER
@@ -43,9 +56,8 @@ def convert_to_entity_type(entity: str, type: str) -> Optional[EntityType]:
             return None
 
         def config_to_response(config_name, config_spec):
-            target_app_id = spec_store.get_target_app_for_sdk_key(client_sdk_key)
             config_target_apps = config_spec.get("targetAppIDs", [])
-            if target_app_id is not None and target_app_id not in config_target_apps:
+            if app_id is not None and app_id not in config_target_apps:
                 return None
 
             eval_result = _ConfigEvaluation()
@@ -61,7 +73,7 @@ def config_to_response(config_name, config_spec):
             if local_override is not None:
                 eval_result = local_override
             else:
-                eval_func(user, config_name, convert_to_entity_type(entity, type), eval_result)
+                eval_func(user, config_name, convert_to_entity_type(entity, type), eval_result, context)
 
             if eval_result is None:
                 return None
@@ -142,7 +154,7 @@ def populate_layer_fields(config_spec, eval_result, result, hash_algo):
             if delegate is not None and delegate != "":
                 delegate_spec = spec_store.get_config(delegate)
                 delegate_result = _ConfigEvaluation()
-                eval_func(user, delegate, EntityType.CONFIG, delegate_result)
+                eval_func(user, delegate, EntityType.CONFIG, delegate_result, context)
 
                 if delegate_spec is not None:
                     result["allocated_experiment_name"] = hash_name(delegate, hash_algo)

statsig/constants.py

@@ -1,7 +1,7 @@
 class Const:
     SUPPORTED_CONDITION_TYPES = ['public', 'fail_gate', 'pass_gate', 'ip_based', 'ua_based', 'user_field',
                                  'environment_field', 'current_time', 'user_bucket', 'unit_id', 'multi_pass_gate',
-                                 'multi_fail_gate']
+                                 'multi_fail_gate', 'target_app']
 
     SUPPORTED_OPERATORS = ['gt', 'gte', 'lt', 'lte',
                            'version_gt', 'version_gte', 'version_lt', 'version_lte',

statsig/evaluation_context.py

@@ -0,0 +1,15 @@
+from typing import Optional
+
+class EvaluationContext:
+    """
+    Context object that holds optional evaluation parameters that need to be passed
+    down through the evaluation function chain.
+    """
+
+    def __init__(self,
+                 sampling_rate: Optional[float] = None,
+                 client_key: Optional[str] = None,
+                 target_app_id: Optional[str] = None):
+        self.sampling_rate = sampling_rate
+        self.client_key = client_key
+        self.target_app_id = target_app_id

statsig/evaluator.py

@@ -9,6 +9,7 @@
 
 from .client_initialize_formatter import ClientInitializeResponseFormatter
 from .config_evaluation import _ConfigEvaluation
+from .evaluation_context import EvaluationContext
 from .evaluation_details import EvaluationDetails, EvaluationReason, DataSource
 from .globals import logger
 from .spec_store import _SpecStore, EntityType
@@ -113,6 +114,7 @@ def get_client_initialize_response(
             hash: HashingAlgorithm,
             client_sdk_key=None,
             include_local_override=False,
+            target_app_id: Optional[str] = None,
     ):
         if not self._spec_store.is_ready_for_checks():
             return None
@@ -122,7 +124,7 @@ def get_client_initialize_response(
 
         return ClientInitializeResponseFormatter \
             .get_formatted_response(self.__eval_config, user, self._spec_store, self, hash, client_sdk_key,
-                                    include_local_override)
+                                    include_local_override, target_app_id)
 
     def _create_evaluation_details(self,
                                    reason: EvaluationReason = EvaluationReason.none,
@@ -217,7 +219,8 @@ def __lookup_override(self, config_overrides, user):
 
     def __lookup_config_mapping(self, user: StatsigUser, config_name: str, spec_type: EntityType,
                                 end_result: _ConfigEvaluation,
-                                maybe_config: Union[Dict[str, Any], None] = None) -> bool:
+                                context: EvaluationContext,
+                                maybe_config: Union[Dict[str, Any], None] = None,) -> bool:
         overrides = self._spec_store.get_overrides()
         if overrides is None or not isinstance(overrides, dict):
             return False
@@ -245,7 +248,8 @@ def __lookup_config_mapping(self, user: StatsigUser, config_name: str, spec_type
                     continue
 
                 end_result.reset()
-                self.__evaluate_rule(user, rule, end_result)
+                context.sampling_rate = rule.get("samplingRate", None)
+                self.__evaluate_rule(user, rule, end_result, context)
                 if not end_result.boolean_value or end_result.evaluation_details.reason in (
                         EvaluationReason.unsupported, EvaluationReason.unrecognized):
                     end_result.reset()
@@ -257,7 +261,7 @@ def __lookup_config_mapping(self, user: StatsigUser, config_name: str, spec_type
                 config_pass = self.__eval_pass_percentage(user, rule, new_config, spec_salt)
                 if config_pass:
                     end_result.override_config_name = override_config_name
-                    self.__evaluate(user, override_config_name, spec_type, end_result)
+                    self.__evaluate(user, override_config_name, spec_type, end_result, context)
                     if end_result.evaluation_details.reason == EvaluationReason.none:
                         return True
         return False
@@ -278,14 +282,16 @@ def unsupported_or_unrecognized(self, config_name, end_result):
             return self._create_evaluation_details(EvaluationReason.unsupported)
         return self._create_evaluation_details(EvaluationReason.unrecognized)
 
-    def check_gate(self, user, gate, end_result=None, is_nested=False):
+    def check_gate(self, user, gate, end_result=None, is_nested=False, context: Optional[EvaluationContext] = None):
         override = self.__lookup_gate_override(user, gate)
         if override is not None:
             return override
 
         if end_result is None:
             end_result = _ConfigEvaluation()
-        self.__eval_config(user, gate, EntityType.GATE, end_result, is_nested)
+        if context is None:
+            context = EvaluationContext()
+        self.__eval_config(user, gate, EntityType.GATE, end_result, context, is_nested)
         return end_result
 
     def get_config(self, user, config_name):
@@ -294,7 +300,7 @@ def get_config(self, user, config_name):
             return override
 
         result = _ConfigEvaluation()
-        self.__eval_config(user, config_name, EntityType.CONFIG, result)
+        self.__eval_config(user, config_name, EntityType.CONFIG, result, EvaluationContext())
         return result
 
     def get_layer(self, user, layer_name):
@@ -303,16 +309,16 @@ def get_layer(self, user, layer_name):
             return override
 
         result = _ConfigEvaluation()
-        self.__eval_config(user, layer_name, EntityType.LAYER, result)
+        self.__eval_config(user, layer_name, EntityType.LAYER, result, EvaluationContext())
         return result
 
-    def __eval_config(self, user, config_name, entity_type: EntityType, end_result, is_nested=False):
+    def __eval_config(self, user, config_name, entity_type: EntityType, end_result, context: EvaluationContext, is_nested=False):
         try:
             if not entity_type:
                 logger.warning("invalid entity type in evaluation: %s", config_name)
                 end_result.rule_id = "error"
                 return
-            self.__evaluate(user, config_name, entity_type, end_result, is_nested)
+            self.__evaluate(user, config_name, entity_type, end_result, context, is_nested)
         except RecursionError:
             raise
         except Exception:
@@ -328,10 +334,10 @@ def __check_id_in_list(self, id, list_name):
             sha256(str(id).encode('utf-8')).digest()).decode('utf-8')[0:8]
         return hashed in ids
 
-    def __evaluate(self, user, config_name, entity_type, end_result, is_nested=False):
+    def __evaluate(self, user, config_name, entity_type, end_result, context: EvaluationContext, is_nested=False):
         maybe_config_spec = self.__get_config_by_entity_type(config_name, entity_type)
 
-        override_config = self.__lookup_config_mapping(user, config_name, entity_type, end_result, maybe_config_spec)
+        override_config = self.__lookup_config_mapping(user, config_name, entity_type, end_result, context, maybe_config_spec)
         if override_config:
             return
 
@@ -344,9 +350,10 @@ def __evaluate(self, user, config_name, entity_type, end_result, is_nested=False
             return
 
         for rule in maybe_config_spec.get("rules", []):
-            self.__evaluate_rule(user, rule, end_result)
+            context.sampling_rate = rule.get("samplingRate", None)
+            self.__evaluate_rule(user, rule, end_result, context)
             if end_result.boolean_value:
-                if self.__evaluate_delegate(user, rule, end_result) is not None:
+                if self.__evaluate_delegate(user, rule, end_result, context) is not None:
                     self.__finalize_exposures(end_result)
                     return
 
@@ -389,15 +396,15 @@ def __finalize_exposures(self, end_result):
         end_result.secondary_exposures = self.clean_exposures(end_result.secondary_exposures)
         end_result.undelegated_secondary_exposures = self.clean_exposures(end_result.undelegated_secondary_exposures)
 
-    def __evaluate_rule(self, user, rule, end_result):
+    def __evaluate_rule(self, user, rule, end_result, context: EvaluationContext):
         total_eval_result = True
         for condition in rule.get("conditions", []):
-            eval_result = self.__evaluate_condition(user, condition, end_result, rule.get("samplingRate", None))
+            eval_result = self.__evaluate_condition(user, condition, end_result, context)
             if not eval_result:
                 total_eval_result = False
         end_result.boolean_value = total_eval_result
 
-    def __evaluate_delegate(self, user, rule, end_result):
+    def __evaluate_delegate(self, user, rule, end_result, context: EvaluationContext):
         config_delegate = rule.get("configDelegate", None)
         if config_delegate is None:
             return None
@@ -408,23 +415,23 @@ def __evaluate_delegate(self, user, rule, end_result):
 
         end_result.undelegated_secondary_exposures = end_result.secondary_exposures[:]
 
-        self.__evaluate(user, config_delegate, EntityType.CONFIG, end_result, True)
+        self.__evaluate(user, config_delegate, EntityType.CONFIG, end_result, context, True)
         end_result.explicit_parameters = config.get(
             "explicitParameters", [])
         end_result.allocated_experiment = config_delegate
         return end_result
 
-    def __evaluate_condition(self, user, condition, end_result, sampling_rate=None):
+    def __evaluate_condition(self, user, condition, end_result, context: EvaluationContext):
         value = None
         type = condition.get("type", "").upper()
         target = condition.get("targetValue")
         field = condition.get("field", "")
         id_Type = condition.get("idType", "userID")
         if type == "PUBLIC":
-            end_result.analytical_condition = sampling_rate is None
+            end_result.analytical_condition = context.sampling_rate is None
             return True
         if type in ("FAIL_GATE", "PASS_GATE"):
-            delegated_gate = self.check_gate(user, target, end_result, True)
+            delegated_gate = self.check_gate(user, target, end_result, True, context)
 
             new_exposure = {
                 "gate": target,
@@ -438,15 +445,15 @@ def __evaluate_condition(self, user, condition, end_result, sampling_rate=None):
 
             pass_gate = delegated_gate.boolean_value if type == "PASS_GATE" else not delegated_gate.boolean_value
 
-            end_result.analytical_condition = sampling_rate is None
+            end_result.analytical_condition = context.sampling_rate is None
             return pass_gate
         if type in ("MULTI_PASS_GATE", "MULTI_FAIL_GATE"):
             if target is None or len(target) == 0:
-                end_result.analytical_condition = sampling_rate is None
+                end_result.analytical_condition = context.sampling_rate is None
                 return False
             pass_gate = False
             for gate in target:
-                other_result = self.check_gate(user, gate)
+                other_result = self.check_gate(user, gate, context=context)
 
                 new_exposure = {
                     "gate": gate,
@@ -461,7 +468,7 @@ def __evaluate_condition(self, user, condition, end_result, sampling_rate=None):
                 if pass_gate:
                     break
 
-            end_result.analytical_condition = sampling_rate is None
+            end_result.analytical_condition = context.sampling_rate is None
             return pass_gate
         if type == "IP_BASED":
             value = self.__get_from_user(user, field)
@@ -478,7 +485,7 @@ def __evaluate_condition(self, user, condition, end_result, sampling_rate=None):
                             self._country_lookup = CountryLookup()
                         value = self._country_lookup.lookupStr(ip)
             if value is None:
-                end_result.analytical_condition = sampling_rate is None
+                end_result.analytical_condition = context.sampling_rate is None
                 return False
         elif type == "UA_BASED":
             value = self.__get_from_user(user, field)
@@ -499,8 +506,13 @@ def __evaluate_condition(self, user, condition, end_result, sampling_rate=None):
                 salt_str + "." + unit_id) % 1000)
         elif type == "UNIT_ID":
             value = self.__get_unit_id(user, id_Type)
+        elif type == "TARGET_APP":
+            if context.client_key is not None:
+                value = context.target_app_id
+            else:
+                value = self._spec_store.get_app_id()
 
-        end_result.analytical_condition = sampling_rate is None
+        end_result.analytical_condition = context.sampling_rate is None
 
         op = condition.get("operator")
         user_bucket = condition.get("user_bucket")

statsig/http_worker.py

@@ -11,7 +11,7 @@
 import ijson
 import requests
 
-from statsig.stream_decompressor import StreamDecompressor
+from .stream_decompressor import StreamDecompressor
 
 from . import globals
 from .diagnostics import Diagnostics, Marker

statsig/spec_store.py

@@ -58,6 +58,7 @@ def __init__(
         self._session_replay_info: Union[None, Dict[str, Dict]] = None
         self._overrides: Union[None, Dict[str, Dict]] = None
         self._override_rules: Union[None, Dict[str, Dict]] = None
+        self._app_id: Union[None, str] = None
 
         self._id_lists: Dict[str, dict] = {}
         self.unsupported_configs: Set[str] = set()
@@ -143,6 +144,9 @@ def get_target_app_for_sdk_key(self, sdk_key=None):
             return target_app_id
         return self._sdk_keys_to_app_ids.get(sdk_key)
 
+    def get_app_id(self):
+        return self._app_id
+
     def get_default_environment(self):
         return self._default_environment
 
@@ -256,6 +260,7 @@ def parse_override_rules(spec_override_rules: Union[None, Dict[str, Dict]]):
         self._session_replay_info = specs_json.get("session_replay_info", None)
         self._overrides = specs_json.get("overrides", None)
         self._override_rules = parse_override_rules(specs_json.get("override_rules", None))
+        self._app_id = specs_json.get("app_id", None)
 
         if self.spec_updater.last_update_time > prev_lcut:
             globals.logger.log_config_sync_update(self.spec_updater.initialized, True,

statsig/statsig.py

@@ -286,7 +286,7 @@ def remove_all_overrides():
 
 def get_client_initialize_response(user: StatsigUser, client_sdk_key: Optional[str] = None,
                                    hash: Optional[HashingAlgorithm] = HashingAlgorithm.SHA256,
-                                   include_local_overrides: Optional[bool] = False) -> ClientInitializeResponse:
+                                   include_local_overrides: Optional[bool] = False, target_app_id: Optional[str] = None) -> ClientInitializeResponse:
     """
     Gets all evaluated values for the given user.
     These values can then be given to a Statsig Client SDK via bootstrapping.
@@ -296,7 +296,7 @@ def get_client_initialize_response(user: StatsigUser, client_sdk_key: Optional[s
     :param client_sdk_key: (Optional) The client sdk key to use for bootstrapping
     :return: An initialize response containing evaluated gates/configs/layers
     """
-    return __instance.get_client_initialize_response(user, client_sdk_key, hash, include_local_overrides)
+    return __instance.get_client_initialize_response(user, client_sdk_key, hash, include_local_overrides, target_app_id)
 
 
 def evaluate_all(user: StatsigUser):

statsig/statsig_server.py

@@ -478,12 +478,13 @@ def get_client_initialize_response(
             client_sdk_key: Optional[str] = None,
             hash: Optional[HashingAlgorithm] = HashingAlgorithm.SHA256,
             include_local_overrides: Optional[bool] = False,
+            target_app_id: Optional[str] = None,
     ):
         hash_value = hash.value if hash is not None else HashingAlgorithm.SHA256.value
 
         def task():
             result = self._evaluator.get_client_initialize_response(
-                self.__normalize_user(user), hash or HashingAlgorithm.SHA256, client_sdk_key, include_local_overrides
+                self.__normalize_user(user), hash or HashingAlgorithm.SHA256, client_sdk_key, include_local_overrides, target_app_id
             )
             if result is None:
                 self._errorBoundary.log_exception("get_client_initialize_response",