feat: SSE support in provider

clients/python/provider/examples/sse_example.py

@@ -0,0 +1,59 @@
+import asyncio
+import logging
+import os
+import sys
+
+_PYTHON_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+sys.path.insert(0, os.path.join(_PYTHON_DIR, "sdk"))
+sys.path.insert(0, os.path.join(_PYTHON_DIR, "bindings"))
+
+from openfeature.evaluation_context import EvaluationContext
+from superposition_provider import LocalResolutionProvider, HttpDataSource
+from superposition_provider.types import SuperpositionOptions, SseStrategy
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s  %(message)s")
+logging.getLogger("superposition_sdk").setLevel(logging.CRITICAL)
+logging.getLogger("superposition_provider").setLevel(logging.WARNING)
+log = logging.getLogger(__name__)
+
+
+ENDPOINT  = os.environ.get("SUPERPOSITION_ENDPOINT", "http://localhost:8080")
+TOKEN     = os.environ.get("SUPERPOSITION_TOKEN", "token")
+ORG       = os.environ.get("SUPERPOSITION_ORG_ID", "localorg")
+WORKSPACE = os.environ.get("SUPERPOSITION_WORKSPACE", "dev")
+
+
+async def main():
+    loop = asyncio.get_event_loop()
+    _orig = loop.default_exception_handler
+    loop.set_exception_handler(
+        lambda l, ctx: None if ctx.get("message") == "Unclosed client session" else _orig(ctx)
+    )
+
+    options = SuperpositionOptions(endpoint=ENDPOINT, token=TOKEN, org_id=ORG, workspace_id=WORKSPACE)
+
+    def on_config_change(before, after):
+        for key, value in after.items():
+            if before.get(key) != value:
+                log.info(f"[UPDATE] {key}: {before.get(key)!r} -> {value!r}")
+
+    provider = LocalResolutionProvider(
+        primary_source=HttpDataSource(options),
+        refresh_strategy=SseStrategy(superposition_options=options, reconnect_delay=5),
+        on_config_change=on_config_change,
+    )
+
+    await provider.initialize(EvaluationContext())
+    log.info(f"Initial config: {provider.resolve_all_features(EvaluationContext())}")
+    log.info("Listening for SSE config changes (Ctrl-C to stop)")
+
+    try:
+        await asyncio.Event().wait()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        await provider.shutdown()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

clients/python/provider/superposition_provider/__init__.py

@@ -4,7 +4,7 @@
 Provides OpenFeature-compliant feature flag providers with support for:
 - Local resolution with caching (LocalResolutionProvider)
 - Remote evaluation without caching (SuperpositionAPIProvider)
-- Configurable refresh strategies (Polling, OnDemand, Watch, Manual)
+- Configurable refresh strategies (Polling, OnDemand, Watch, SSE, Manual)
 - File-based and HTTP-based data sources
 - Full FFI integration for performance
 """
@@ -29,6 +29,7 @@
     PollingStrategy,
     OnDemandStrategy,
     WatchStrategy,
+    SseStrategy,
     ManualStrategy,
     RefreshStrategy as RefreshStrategyType,
     ConfigurationOptions,
@@ -59,6 +60,7 @@
     "PollingStrategy",
     "OnDemandStrategy",
     "WatchStrategy",
+    "SseStrategy",
     "ManualStrategy",
     "RefreshStrategyType",
     "ConfigurationOptions",

clients/python/provider/superposition_provider/http_data_source.py

@@ -197,6 +197,13 @@ def supports_experiments(self) -> bool:
         return True
 
     async def close(self) -> None:
-        """Close the HTTP client."""
+        """Close the HTTP client and its underlying aiohttp session."""
         if self.client:
+            try:
+                http_client = getattr(self.client._config, "http_client", None)
+                session = getattr(http_client, "_session", None)
+                if session and not session.closed:
+                    await session.close()
+            except Exception:
+                pass
             self.client = None

clients/python/provider/superposition_provider/local_provider.py

@@ -10,7 +10,7 @@
 import json
 import weakref
 from datetime import datetime, timezone
-from typing import Dict, List, Optional, Any, Tuple, Union, Sequence, Mapping
+from typing import Callable, Dict, List, Optional, Any, Tuple, Union, Sequence, Mapping
 
 from openfeature.provider import (
     AbstractProvider,
@@ -26,7 +26,7 @@
 from . import FetchResponse
 from .data_source import SuperpositionDataSource, ConfigData, ExperimentData
 from .interfaces import AllFeatureProvider, FeatureExperimentMeta
-from .types import RefreshStrategy, OnDemandStrategy, WatchStrategy, PollingStrategy, ManualStrategy, default_on_demand_strategy
+from .types import RefreshStrategy, OnDemandStrategy, WatchStrategy, PollingStrategy, ManualStrategy, SseStrategy, default_on_demand_strategy
 
 logger = logging.getLogger(__name__)
 
@@ -45,6 +45,7 @@ def __init__(
         primary_source: SuperpositionDataSource,
         fallback_source: Optional[SuperpositionDataSource] = None,
         refresh_strategy: RefreshStrategy = default_on_demand_strategy(),
+        on_config_change: Optional[Callable[[Dict[str, Any], Dict[str, Any]], None]] = None,
     ):
         """Initialize local resolution provider.
 
@@ -56,6 +57,7 @@ def __init__(
         self.primary_source = primary_source
         self.fallback_source = fallback_source
         self.refresh_strategy = refresh_strategy
+        self.on_config_change = on_config_change
 
         self.metadata = Metadata(name="LocalResolutionProvider")
         self.status = ProviderStatus.NOT_READY
@@ -68,6 +70,8 @@ def __init__(
 
         # Background task for refresh strategy
         self._background_task: Optional[asyncio.Task] = None
+        # Set once the SSE connection is established (SseStrategy only)
+        self._sse_connected_event: Optional[asyncio.Event] = None
 
     async def initialize(self, context: EvaluationContext):
         """Initialize the provider.
@@ -358,7 +362,21 @@ async def refresh(self) -> None:
 
         Useful for MANUAL refresh strategy.
         """
-        await asyncio.gather(self._fetch_and_cache_config(), self._fetch_and_cache_experiments())
+        before = self.resolve_all_features(EvaluationContext()) if self.on_config_change and self.ffi_cache else None
+
+        results = await asyncio.gather(
+            self._fetch_and_cache_config(),
+            self._fetch_and_cache_experiments(),
+            return_exceptions=True,
+        )
+        for result in results:
+            if isinstance(result, Exception):
+                logger.warning(f"Error during refresh: {result}")
+
+        if self.on_config_change and before is not None:
+            after = self.resolve_all_features(EvaluationContext())
+            if before != after:
+                self.on_config_change(before, after)
 
     # --- Private helpers ---
 
@@ -574,7 +592,75 @@ async def _watch_loop() -> None:
             case WatchStrategy():
                 self._background_task = asyncio.create_task(_watch_loop())
             case PollingStrategy():
-                self._background_task = asyncio.create_task(_polling_loop())
+                self._background_task = asyncio.create_task(self._polling_loop())
+            case SseStrategy():
+                import weakref
+                strategy: SseStrategy = self.refresh_strategy
+                options = strategy.superposition_options
+                debounce_s = strategy.debounce_ms / 1000
+                reconnect_delay = max(strategy.reconnect_delay, 1)
+                sse_url = f"{options.endpoint.rstrip('/')}/{options.org_id}/{options.workspace_id}/stream"
+                sse_headers = {"Authorization": f"Bearer {options.token}"}
+                if options.org_id:
+                    sse_headers["x-org-id"] = options.org_id
+                weak_self = weakref.ref(self)
+                connected_event = asyncio.Event()
+                self._sse_connected_event = connected_event
+
+                async def _sse_loop():
+                    import aiohttp
+                    logger.info(f"Starting SSE refresh (url={sse_url})")
+                    try:
+                        async with aiohttp.ClientSession() as session:
+                            while True:
+                                if weak_self() is None:
+                                    logger.info("Provider garbage collected, stopping SSE loop.")
+                                    return
+                                try:
+                                    async with session.get(
+                                        sse_url,
+                                        headers=sse_headers,
+                                        timeout=aiohttp.ClientTimeout(total=None, sock_read=30),
+                                    ) as resp:
+                                        if resp.status != 200:
+                                            logger.warning(f"SSE endpoint returned {resp.status}, retrying in {reconnect_delay}s")
+                                            await asyncio.sleep(reconnect_delay)
+                                            continue
+                                        logger.info("SSE connection established")
+                                        connected_event.set()
+                                        self_ref = weak_self()
+                                        if self_ref is not None:
+                                            try:
+                                                await self_ref.refresh()
+                                            except Exception as e:
+                                                logger.warning(f"Reconnect refresh failed: {e}")
+                                        debounce_task = None
+                                        async for line_bytes in resp.content:
+                                            self_ref = weak_self()
+                                            if self_ref is None:
+                                                logger.info("Provider garbage collected, stopping SSE loop.")
+                                                return
+                                            line = line_bytes.decode("utf-8", errors="replace").strip()
+                                            logger.debug(f"SSE raw line: {line!r}")
+                                            if not line or line.startswith(":"):
+                                                continue
+                                            if line.startswith("event:") or line.startswith("data:"):
+                                                async def _do_refresh(ref=self_ref):
+                                                    await asyncio.sleep(debounce_s)
+                                                    try:
+                                                        await ref.refresh()
+                                                    except Exception as e:
+                                                        logger.warning(f"SSE-triggered refresh failed: {e}")
+                                                if debounce_task and not debounce_task.done():
+                                                    debounce_task.cancel()
+                                                debounce_task = asyncio.create_task(_do_refresh())
+                                except (aiohttp.ClientError, asyncio.TimeoutError) as e:
+                                    logger.warning(f"SSE connection error: {e}, retrying in {reconnect_delay}s")
+                                    await asyncio.sleep(reconnect_delay)
+                    except asyncio.CancelledError:
+                        logger.info("SSE loop cancelled")
+
+                self._background_task = asyncio.create_task(_sse_loop())
             case ManualStrategy():
                 logger.debug("MANUAL strategy: caller must invoke refresh()")
             case OnDemandStrategy():

clients/python/provider/superposition_provider/types.py

@@ -71,6 +71,19 @@ class WatchStrategy:
 def default_watch_strategy():
     return WatchStrategy(500)
 
+@dataclass
+class SseStrategy:
+    """SSE-based refresh strategy.
+
+    Connects to the server's SSE endpoint and refreshes when a change event
+    is received. Reconnects automatically on connection failure.
+
+    Requires SuperpositionOptions to build the SSE endpoint URL and authenticate.
+    """
+    superposition_options: SuperpositionOptions
+    reconnect_delay: int = 5  # seconds between reconnect attempts
+    debounce_ms: int = 500  # debounce rapid successive events
+
 @dataclass
 class ManualStrategy:
     """Manual refresh strategy.
@@ -81,7 +94,7 @@ class ManualStrategy:
 
 
 # Union type for all refresh strategies
-RefreshStrategy = Union[PollingStrategy, OnDemandStrategy, WatchStrategy, ManualStrategy]
+RefreshStrategy = Union[PollingStrategy, OnDemandStrategy, WatchStrategy, SseStrategy, ManualStrategy]
 
 
 # ============================================================================

crates/context_aware_config/src/api/context/handlers.rs

@@ -15,7 +15,7 @@ use diesel::{
 use serde_json::{Map, Value};
 use service_utils::{
     helpers::{
-        WebhookData, execute_webhook_call, fetch_dimensions_info_map, parse_config_tags,
+        WebhookData, fetch_dimensions_info_map, notify_change, parse_config_tags,
     },
     middlewares::auth_z::{Action as AuthZAction, AuthZ},
     service::types::{
@@ -186,7 +186,7 @@ async fn create_handler(
     };
 
     let webhook_status =
-        execute_webhook_call(data, &workspace_context, &state, &mut conn).await;
+        notify_change(data, &workspace_context, &state, &mut conn).await;
 
     let mut http_resp = if webhook_status {
         HttpResponse::Ok()
@@ -322,7 +322,7 @@ async fn update_handler(
     };
 
     let webhook_status =
-        execute_webhook_call(data, &workspace_context, &state, &mut conn).await;
+        notify_change(data, &workspace_context, &state, &mut conn).await;
 
     let mut http_resp = if webhook_status {
         HttpResponse::Ok()
@@ -467,7 +467,7 @@ async fn move_handler(
     };
 
     let webhook_status =
-        execute_webhook_call(data, &workspace_context, &state, &mut conn).await;
+        notify_change(data, &workspace_context, &state, &mut conn).await;
 
     let mut http_resp = if webhook_status {
         HttpResponse::Ok()
@@ -754,7 +754,7 @@ async fn delete_handler(
     };
 
     let webhook_status =
-        execute_webhook_call(data, &workspace_context, &state, &mut conn).await;
+        notify_change(data, &workspace_context, &state, &mut conn).await;
 
     let mut http_resp = if webhook_status {
         HttpResponse::NoContent()
@@ -1129,7 +1129,7 @@ async fn bulk_operations_handler(
     };
 
     let webhook_status =
-        execute_webhook_call(data, &workspace_context, &state, &mut conn).await;
+        notify_change(data, &workspace_context, &state, &mut conn).await;
 
     let mut resp_builder = if webhook_status {
         HttpResponse::Ok()
@@ -1245,7 +1245,7 @@ async fn weight_recompute_handler(
     };
 
     let webhook_status =
-        execute_webhook_call(data, &workspace_context, &state, &mut conn).await;
+        notify_change(data, &workspace_context, &state, &mut conn).await;
 
     let mut http_resp = if webhook_status {
         HttpResponse::Ok()

crates/context_aware_config/src/api/default_config/handlers.rs

@@ -12,7 +12,7 @@ use diesel::{
 use jsonschema::ValidationError;
 use serde_json::Value;
 use service_utils::{
-    helpers::{WebhookData, execute_webhook_call, parse_config_tags},
+    helpers::{WebhookData, notify_change, parse_config_tags},
     service::types::{
         AppHeader, AppState, CustomHeaders, DbConnection, EncryptionKey, SchemaName,
         WorkspaceContext,
@@ -188,7 +188,7 @@ async fn create_handler(
     };
 
     let webhook_status =
-        execute_webhook_call(data, &workspace_context, &state, &mut conn).await;
+        notify_change(data, &workspace_context, &state, &mut conn).await;
 
     let mut http_resp = if webhook_status {
         HttpResponse::Ok()
@@ -348,7 +348,7 @@ async fn update_handler(
     };
 
     let webhook_status =
-        execute_webhook_call(data, &workspace_context, &state, &mut conn).await;
+        notify_change(data, &workspace_context, &state, &mut conn).await;
 
     let mut http_resp = if webhook_status {
         HttpResponse::Ok()
@@ -586,7 +586,7 @@ async fn delete_handler(
         };
 
         let webhook_status =
-            execute_webhook_call(data, &workspace_context, &state, &mut conn).await;
+            notify_change(data, &workspace_context, &state, &mut conn).await;
 
         let mut http_resp = if webhook_status {
             HttpResponse::Ok()