feat: use infrastructure class as global wrapper, add meter event handler, initial integration of nudging event

Author: lc-spxl

config/clients.yaml

@@ -5,6 +5,13 @@
 clients:
   dataset_api:
     class: celine.dt.core.clients.dataset_api:DatasetSqlApiClient
+    scope: dataset.query
     config:
       base_url: "${DATASET_API_BASE_URL:-http://api.celine.localhost/datasets}"
       timeout: 30.0
+  rec_registry_admin:
+    class: celine.sdk.rec_registry:RecRegistryAdminClient
+    scope: "rec-registry.lookup"
+    config:
+      base_url: "${REC_REGISTRY_URL:-http://api.celine.localhost/rec-registry}"
+      timeout: 5.0

src/celine/dt/contracts/__init__.py

@@ -3,8 +3,10 @@
 from celine.dt.contracts.events import DTEvent, EventSource, EventSeverity
 from celine.dt.contracts.component import DTComponent
 from celine.dt.contracts.simulation import DTSimulation, SimulationDescriptor
-from celine.dt.contracts.subscription import SubscriptionSpec, EventHandler, EventContext
+from celine.dt.contracts.subscription import SubscriptionSpec, EventHandler, EventContext, RouteDef
 from celine.dt.contracts.values import ValueFetcherSpec
+from celine.dt.contracts.infrastructure import Infrastructure
+from celine.dt.contracts.app import AppState
 
 # Broker types re-exported from SDK
 from celine.sdk.broker import (
@@ -21,7 +23,9 @@
     "DTEvent", "EventSource", "EventSeverity",
     "DTComponent",
     "DTSimulation", "SimulationDescriptor",
-    "SubscriptionSpec", "EventHandler", "EventContext",
+    "SubscriptionSpec", "EventHandler", "EventContext", "RouteDef",
     "ValueFetcherSpec",
     "Broker", "BrokerMessage", "MqttBroker", "MqttConfig", "PublishResult", "QoS",
+    "Infrastructure",
+    "AppState"
 ]

src/celine/dt/contracts/app.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+
+from starlette.datastructures import State
+from celine.dt.contracts.infrastructure import Infrastructure
+from celine.sdk.auth import TokenProvider
+
+@dataclass
+class AppState(State):
+    infra: Infrastructure
+    token_provider: TokenProvider | None = None

src/celine/dt/contracts/infrastructure.py

@@ -0,0 +1,54 @@
+# celine/dt/core/infrastructure.py
+from __future__ import annotations
+from dataclasses import dataclass, field, replace
+from typing import TYPE_CHECKING, Any, Optional, TypeVar
+
+from celine.dt.core.broker.service import BrokerService
+from celine.dt.core.clients.registry import ClientsRegistry
+from celine.dt.core.values.service import ValuesService, ValuesRegistry
+from celine.dt.core.simulation.registry import SimulationRegistry
+
+if TYPE_CHECKING:
+    from celine.dt.core.domain.registry import DomainRegistry
+    from celine.dt.core.broker.subscriptions import SubscriptionManager
+    from celine.sdk.auth import TokenProvider
+
+
+@dataclass
+class Infrastructure:
+
+    # always present after create_app
+    broker: BrokerService
+    values_service: ValuesService
+    values_registry: ValuesRegistry
+    clients_registry: ClientsRegistry
+    simulation_registry: SimulationRegistry
+
+    # set after lifespan finalization
+    _domain_registry: Optional[DomainRegistry] = field(default=None)
+    _subscription_manager: Optional[SubscriptionManager] = field(default=None)
+    _token_provider: Optional[TokenProvider] = field(default=None)
+
+    overrides: dict[str, Any] = field(default_factory=dict)
+   
+    @property
+    def domain_registry(self) -> DomainRegistry:
+        if self._domain_registry is None:
+            raise RuntimeError("Infrastructure.domains not set yet - domain loading incomplete")
+        return self._domain_registry
+
+    @property
+    def subscription_manager(self) -> SubscriptionManager:
+        if self._subscription_manager is None:
+            raise RuntimeError("Infrastructure.subscription_manager not set yet - lifespan incomplete")
+        return self._subscription_manager
+
+    @property
+    def token_provider(self) -> TokenProvider:
+        if self._token_provider is None:
+            raise RuntimeError("Infrastructure.token_provider not set yet - lifespan incomplete")
+        return self._token_provider
+
+    def with_overrides(self, overrides: dict[str, Any]) -> Infrastructure:
+        """Return a shallow copy with domain-specific overrides applied."""
+        return replace(self, overrides=overrides)

src/celine/dt/contracts/routes.py

@@ -90,9 +90,5 @@ def from_dataclass(obj: Any) -> "FetchResultSchema":
         return FetchResultSchema.model_validate(obj.__dict__)
 
 
-class DescribeResponseSchema(BaseModel):
-    payload: GenericPayload
-
-
 class SummaryResponseSchema(BaseModel):
     payload: GenericPayload

src/celine/dt/contracts/subscription.py

@@ -12,12 +12,15 @@
 from typing import TYPE_CHECKING, Any, Awaitable, Callable, Protocol, runtime_checkable
 from uuid import uuid4
 
-from celine.dt.contracts.events import DTEvent
+
+from celine.dt.contracts import DTEvent
 
 if TYPE_CHECKING:
+    from celine.dt.core.clients.registry import ClientsRegistry
+    from celine.dt.core.domain.registry import DomainRegistry
+    from celine.dt.contracts import Infrastructure
     from celine.dt.core.broker.service import BrokerService
     from celine.dt.core.domain.base import DTDomain
-    from celine.dt.core.domain.registry import DomainRegistry
     from celine.dt.core.values.service import ValuesService
 
 
@@ -32,35 +35,16 @@ async def on_run_completed(event: DTEvent, ctx: EventContext) -> None:
             users = await ctx.values.fetch("pipeline-reactor.affected_users", {...})
             await ctx.broker.publish_event(topic=f"celine/nudging/{user_id}", payload={...})
     """
-
     topic: str
     broker_name: str
     received_at: datetime
-
-    # App-scope infrastructure — always populated by SubscriptionManager
-    broker: BrokerService
-    values: ValuesService
-
-    # Domain registry — for get_dt() lookups
-    registry: DomainRegistry | None = None
-
+    infra: Infrastructure
     entity_id: str | None = None
     message_id: str | None = None
     raw_payload: bytes | None = None
 
     def get_dt(self, domain_type: str) -> DTDomain:
-        """Return the DTDomain instance for the given domain_type.
-
-        Usage in a plain @on_event handler::
-
-            @on_event("pipeline.run.completed", topics=["celine/pipelines/runs/+"])
-            async def on_run_completed(event: DTEvent, ctx: EventContext) -> None:
-                participant = ctx.get_dt("participant")
-                entity = await participant.resolve_entity(participant_id)
-        """
-        if self.registry is None:
-            raise RuntimeError("EventContext has no registry — check SubscriptionManager setup")
-        return self.registry.get_by_type(domain_type)
+        return self.infra.domain_registry.get_by_type(domain_type)
 
 
 EventHandler = Callable[[DTEvent, EventContext], Awaitable[None]]  # type: ignore[type-arg]

src/celine/dt/core/broker/subscriptions.py

@@ -11,11 +11,7 @@
 
 from celine.sdk.broker import QoS, ReceivedMessage, SubscribeResult
 
-from celine.dt.contracts.events import DTEvent, EventSource
-from celine.dt.contracts.subscription import EventContext, EventHandler, RouteDef, SubscriptionSpec
-from celine.dt.core.broker.service import BrokerService
-from celine.dt.core.domain.registry import DomainRegistry
-from celine.dt.core.values.service import ValuesService
+from celine.dt.contracts import DTEvent, EventSource, EventContext, EventHandler, RouteDef, SubscriptionSpec, Infrastructure
 
 logger = logging.getLogger(__name__)
 
@@ -156,23 +152,23 @@ class SubscriptionManager:
     def __init__(
         self,
         *,
-        broker_service: BrokerService,
-        values_service: ValuesService,
-        domain_registry: DomainRegistry | None = None,
+        infra: Infrastructure,
         domains: list[Any] | None = None,
         handler_specs: list[SubscriptionSpec] | None = None,
         default_qos: QoS = QoS.AT_LEAST_ONCE,
         default_broker_name: str | None = None,
     ) -> None:
-        self._broker_service = broker_service
-        self._values_service = values_service
-        self._domain_registry = domain_registry
+        self._infra = infra
         self._domains = domains or []
         self._handler_specs = handler_specs or []
         self._default_qos = default_qos
         self._default_broker_name = default_broker_name
         self._active: list[ActiveSubscription] = []
 
+    @property
+    def infra(self) -> Infrastructure:
+        return self._infra
+
     async def start(self) -> None:
         # Domain instances
         for domain in self._domains:
@@ -208,7 +204,7 @@ async def _register_specs(
                 broker_name=broker_name or "<default>",
             )
 
-            res: SubscribeResult = await self._broker_service.subscribe(
+            res: SubscribeResult = await self.infra.broker.subscribe(
                 topics=topics,
                 handler=handler,
                 broker_name=broker_name,
@@ -240,7 +236,7 @@ async def _register_specs(
     async def stop(self) -> None:
         for sub in list(self._active):
             try:
-                await self._broker_service.unsubscribe(
+                await self.infra.broker.unsubscribe(
                     subscription_id=sub.subscription_id,
                     broker_name=sub.broker_name,
                 )
@@ -257,22 +253,20 @@ def _wrap_handler(
         spec: SubscriptionSpec,
         broker_name: str,
     ) -> Callable[[ReceivedMessage], Awaitable[None]]:
-        broker_service = self._broker_service
-        values_service = self._values_service
-        domain_registry = self._domain_registry
+        broker_service = self.infra.broker
+        values_service = self.infra.values_service
+        domain_registry = self.infra.domain_registry
 
         async def _handler(msg: ReceivedMessage) -> None:
             try:
                 event = _dt_event_from_received(
                     source_name=source_name, spec=spec, msg=msg
                 )
                 ctx = EventContext(
+                    infra=self.infra,
                     topic=msg.topic,
                     broker_name=broker_name,
                     received_at=msg.timestamp or datetime.now(timezone.utc),
-                    broker=broker_service,
-                    values=values_service,
-                    registry=domain_registry,
                     entity_id=spec.metadata.get("entity_id"),
                     message_id=msg.message_id,
                     raw_payload=msg.raw_payload,

src/celine/dt/core/clients/dataset_api.py

@@ -5,11 +5,12 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Any, AsyncIterator, Optional
+from typing import TYPE_CHECKING, Any, AsyncIterator
 import httpx
 
 if TYPE_CHECKING:
     from celine.dt.api.context import Ctx
+    from celine.sdk.auth.provider import TokenProvider
 
 logger = logging.getLogger(__name__)
 
@@ -28,28 +29,33 @@ def __init__(
         *,
         base_url: str,
         timeout: float = 30.0,
-        token_provider: Any | None = None,
+        token_provider: TokenProvider | None = None,
     ) -> None:
         self._base = base_url.rstrip("/")
         self._timeout = timeout
         self._token_provider = token_provider
 
-    async def _headers(self) -> dict[str, str]:
-        if not self._token_provider:
-            return {}
-        token = await self._token_provider.get_token()
-        return {"Authorization": f"Bearer {token.access_token}"}
+    async def _headers(self, user_token: str | None = None) -> dict[str, str]:
+
+        token = user_token
+        if not user_token:
+            if not self._token_provider:
+                return {}
+            client_token = await self._token_provider.get_token()
+            token = client_token.access_token
+
+        return {"Authorization": f"Bearer {token}"}
 
     async def query(
         self, *, sql: str, limit: int = 1000, offset: int = 0, ctx: Ctx | None = None
     ) -> list[dict[str, Any]]:
-        headers = await self._headers()
 
+        token: str | None = None
         if ctx and ctx.token:
             token = ctx.token
-            headers["Authorization"] = (
-                token if token.lower().startswith("bearer") else f"Bearer {token}"
-            )
+            token = token if token.lower().startswith("bearer") else f"Bearer {token}"
+        
+        headers = await self._headers(token)
 
         async with httpx.AsyncClient(timeout=self._timeout) as client:
             try:

src/celine/dt/core/clients/loader.py

@@ -8,17 +8,21 @@
 import logging
 from typing import Any, Iterable
 
+from celine.dt.core.config import settings
 from celine.dt.core.clients.registry import ClientsRegistry
 from celine.dt.core.loader import import_attr, load_yaml_files, substitute_env_vars
+from celine.sdk.auth.provider import TokenProvider
+from celine.sdk.auth import OidcClientCredentialsProvider
 
 logger = logging.getLogger(__name__)
 
 
+
 def load_and_register_clients(
     *,
     patterns: Iterable[str],
     registry: ClientsRegistry,
-    token_provider: Any | None = None,
+    token_provider: TokenProvider | None = None,
 ) -> None:
     """Load client definitions from YAML and register live instances.
 
@@ -42,26 +46,26 @@ def load_and_register_clients(
     for data in yamls:
         for name, spec in (data.get("clients") or {}).items():
             class_path = spec.get("class")
-            if not class_path:
-                raise ValueError(f"Client '{name}' missing 'class' field")
-
+            scope = spec.get("scope")
             raw_config = substitute_env_vars(spec.get("config", {}))
 
-            logger.info("Loading client '%s' from '%s'", name, class_path)
-            try:
-                cls = import_attr(class_path)
-            except (ImportError, AttributeError):
-                logger.exception("Failed to import client class '%s'", class_path)
-                raise
-
+            cls = import_attr(class_path)
             kwargs = dict(raw_config)
 
-            # Inject token_provider if the constructor accepts it
             sig = inspect.signature(cls.__init__)
             if "token_provider" in sig.parameters:
                 kwargs["token_provider"] = token_provider
+                if scope and isinstance(token_provider, OidcClientCredentialsProvider):
+                    if settings.oidc.client_id and settings.oidc.client_secret:
+                        kwargs["token_provider"] = OidcClientCredentialsProvider(
+                            base_url=settings.oidc.base_url,
+                            client_id=settings.oidc.client_id,
+                            client_secret=settings.oidc.client_secret,
+                            scope=scope,
+                            timeout=settings.oidc.timeout,
+                        )
+                    else:
+                        logger.warning(f"Cannot initialize OIDC token provider for aud={scope}: missing client_id / client_secret")
 
-            client = cls(**kwargs)
-            registry.register(name, client)
-
+            registry.register(name, cls(**kwargs))
     logger.info("Registered %d client(s): %s", len(registry.list()), registry.list())