chore: bump version to 0.6.7

SIENTIAPDE-2149: model_ids allow-list filter for receiver groups

CLAUDE.md

@@ -45,7 +45,7 @@ CI quality gate (`.github/workflows/quality-gate.yml`) calls the shared reusable
 
 ### Layering
 - **`orchestrator/worker/worker.py`** — entrypoint (`python -m orchestrator.worker.worker`). Builds a single `Activities` instance, connects to Temporal, starts Prometheus (`HTTP_METRICS_PORT`) and Temporal SDK metrics (`HTTP_SDK_METRICS_PORT`) servers, then runs three `Worker`s concurrently via `asyncio.gather`, each on its own task queue: `orchestrator-queue`, `alerts-queue`, `reports-queue`. `prepare_worker.py` derives the queue name from the workflow class name (`camel_to_snake`) and reads per-queue concurrency/poller tuning from env vars prefixed with the workflow name (e.g. `ORCHESTRATOR_MAX_CONCURRENT_ACTIVITIES`).
-- **`orchestrator/workflows/`** — Temporal workflow definitions: `orchestrator.py` (pipeline/slot/schedule orchestration), `alerts.py` (real-time ERROR notifications), `reports.py` (scheduled all-level reports). `alerts.py` and `reports.py` both delegate to the same two subworkflows in `workflows/subworkflows/`: `load_notification_package.py` (load notifications + receiver configs from Mongo, incremental via Redis timestamp) and `process_notifications.py` (build HTML email, send, log to Postgres). The difference between alerts and reports is which filter activity they call (`filter_notification_alerts` vs `filter_notification_reports`) and their base data filter (`level: ERROR` by default vs all levels) — `load_notification_package.py`'s `_compute_query_time_filter` can widen either beyond that default to the union of active receiver groups' own `levels` (SIENTIAPDE-2011); the ERROR-only default only holds when no active group narrows further. Receiver groups also support a `model_ids` allow-list (SIENTIAPDE-2149), independent of the pre-existing `ignore_models_id` deny-list — both apply together (AND) when a group configures both, and `model_ids` follows the same union-at-query-time / precise-per-group-slice split as `levels`. `load_latest_data`'s Mongo query is capped at `NOTIFICATION_QUEUE_BATCH_SIZE` documents per run (default 5000), sorted by `timestamp` ascending so a capped run always returns the oldest pending batch first (SIENTIAPDE-2013) — a hard cap with no drain loop, since the incremental Redis cursor picks up any remainder on the next run.
+- **`orchestrator/workflows/`** — Temporal workflow definitions: `orchestrator.py` (pipeline/slot/schedule orchestration), `alerts.py` (real-time ERROR notifications), `reports.py` (scheduled all-level reports). `alerts.py` and `reports.py` both delegate to the same two subworkflows in `workflows/subworkflows/`: `load_notification_package.py` (load notifications + receiver configs from Mongo, incremental via Redis timestamp) and `process_notifications.py` (build HTML email, send, log to Postgres). The difference between alerts and reports is which filter activity they call (`filter_notification_alerts` vs `filter_notification_reports`) and their base data filter (`level: ERROR` by default vs all levels) — `load_notification_package.py`'s `_compute_query_time_filter` can widen either beyond that default to the union of active receiver groups' own `levels` (SIENTIAPDE-2011); the ERROR-only default only holds when no active group narrows further. Receiver groups also support a `model_ids` allow-list (SIENTIAPDE-2149), independent of the pre-existing `ignore_models_id` deny-list — both apply together (AND) when a group configures both, and `model_ids` follows the same union-at-query-time / precise-per-group-slice split as `levels`. `load_latest_data`'s Mongo query is capped at `NOTIFICATION_QUEUE_BATCH_SIZE` documents per run (default `0`, meaning unlimited — unchanged from pre-SIENTIAPDE-2013 behavior unless a deployment opts in), sorted by `timestamp` ascending so a capped run always returns the oldest pending batch first (SIENTIAPDE-2013) — a hard cap with no drain loop, since the incremental Redis cursor picks up any remainder on the next run.
 - **`orchestrator/activities/activities.py`** — `Activities` is a single class combining every activity mixin via multiple inheritance: `TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres` (Postgres comes from `sientia_do.temporal.activities.postgres`, the external `sientia-dataops-library` dependency). One instance is constructed in `worker.py` and its bound methods are handed to each `Worker` as the activity list — activities are plain methods on this shared object, not separate classes. Each mixin's `__init__` takes its own config dict plus shared `logger`, `notification_handler`, `metrics_controller`. `Activities.shutdown()` closes every mixin's connections.
 - **`orchestrator/utils/`** — `connectors_config.py` builds each config dict from env vars (Temporal/Redis/MongoDB/Postgres/Email); `email_builder.py` renders Jinja2 templates (`utils/templates/email_template.html`, `general_template.html`); `orchestrator_functions.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `converters.py` handles Temporal schedule frequency parsing.
 

orchestrator/activities/mongo_db.py

@@ -451,15 +451,9 @@ class MongoDB(SientiaMonitoring):
         filtering by timestamp to enable incremental data processing. It
         handles connection management and provides comprehensive error reporting.
 
-        SIENTIAPDE-2013: the query is capped at ``notification_queue_batch_size``
-        documents (hard cap per run, no drain loop) and sorted by ``timestamp``
-        ascending, so a capped run always returns the *oldest* pending documents
-        first. That matters because the caller advances its incremental cursor to
-        the max timestamp of whatever this returns -- without the ascending sort,
-        a capped batch could skip over older documents that never get picked up
-        on a later run. This backstops a cold start, backlog buildup, or
-        notification storm from fetching an unbounded (and potentially
-        multi-megabyte) payload in one query.
+        The query is capped at ``notification_queue_batch_size`` documents and sorted by
+        ``timestamp`` ascending, so a capped run returns the oldest pending documents first --
+        the caller advances its cursor only to what was actually returned, so nothing is skipped.
 
         Args:
             input_data (dict[str, Any]): Activity input parameters.

orchestrator/utils/connectors_config.py

@@ -33,7 +33,7 @@ def build_mongodb_config():
         'connection_string': connection_string,
         'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'),
         'ttl_index_seconds': int(getenv('MONGODB_TTL_INDEX_HOURS', '1')) * 3600,
-        'notification_queue_batch_size': int(getenv('NOTIFICATION_QUEUE_BATCH_SIZE', '5000')),
+        'notification_queue_batch_size': int(getenv('NOTIFICATION_QUEUE_BATCH_SIZE', '0')),
     }
 
 

orchestrator/workflows/subworkflows/load_notification_package.py

@@ -147,14 +147,11 @@ class LoadNotificationPackage:
         if not sending_configs:
             return default_filter
 
-        sending_configs_levels = [group.get('levels') for group in sending_configs]
-
-        if all(sending_configs_levels):
-            levels_union: set[str] = set()
-            for levels in sending_configs_levels:
-                if levels:
-                    levels_union.update(levels)
-            query_filter = {**default_filter, 'level': {'$in': sorted(levels_union)}}
+        levels_union = self._union_when_all_restrict(
+            [group.get('levels') for group in sending_configs]
+        )
+        if levels_union is not None:
+            query_filter = {**default_filter, 'level': {'$in': levels_union}}
         else:
             query_filter = dict(default_filter)
 
@@ -166,21 +163,40 @@ class LoadNotificationPackage:
         ]
         models_intersection = set.intersection(*sending_configs_ignore_models_id)
 
-        sending_configs_model_ids = [group.get('model_ids') for group in sending_configs]
+        model_ids_union = self._union_when_all_restrict(
+            [group.get('model_ids') for group in sending_configs]
+        )
 
         model_id_filter: dict[str, Any] = {}
-
-        if all(sending_configs_model_ids):
-            model_ids_union: set[str] = set()
-            for model_ids in sending_configs_model_ids:
-                if model_ids:
-                    model_ids_union.update(model_ids)
-            model_id_filter['$in'] = sorted(model_ids_union)
-
+        if model_ids_union is not None:
+            model_id_filter['$in'] = model_ids_union
         if models_intersection:
             model_id_filter['$nin'] = sorted(models_intersection)
-
         if model_id_filter:
             query_filter['model_id'] = model_id_filter
 
         return query_filter
+
+    @staticmethod
+    def _union_when_all_restrict(per_group_values: list[list[str] | None]) -> list[str] | None:
+        """
+        Union a list of per-group optional restrictions, but only when every group has one.
+
+        Shared by `levels` and `model_ids` (SIENTIAPDE-2149) in `_compute_query_time_filter`,
+        which both follow the same rule: if ANY active group is unrestricted on that axis
+        (missing/empty value), that group wants everything, so the union can't restrict the
+        query either — the caller falls back to no filter on that axis in that case (`None`).
+
+        Args:
+            per_group_values: Each active group's raw value for the axis being unioned
+                (e.g. `group.get('levels')` for every group).
+
+        Returns:
+            Sorted union of every group's values, or None if any group was unrestricted.
+        """
+        if not all(per_group_values):
+            return None
+        union: set[str] = set()
+        for values in per_group_values:
+            union.update(values or [])
+        return sorted(union)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "orchestrator"
-version = "0.6.5"
+version = "0.6.7"
 description = "Sientia DataOps Orchestrator - ML Model Orchestration System"
 readme = "README.md"
 requires-python = ">=3.11"

requirements-local.txt

@@ -4,5 +4,5 @@ sqlalchemy
 redis
 pymongo
 jinja2
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.3
 prometheus-client

tests/orchestrator/utils/test_connectors_config.py

@@ -84,7 +84,7 @@ def test_build_mongo_db_config_with_defaults():
         'connection_string': 'mongodb://root:wKZDbMNU1c@localhost:27018',
         'database_name': 'sientia',
         'ttl_index_seconds': 3600,
-        'notification_queue_batch_size': 5000,
+        'notification_queue_batch_size': 0,
     }