Merge branch 'fix/SIENTIAPDE-2013' of https://github.com/Aignosi/sientia-dataops-orchestrator_temporal into fix/SIENTIAPDE-2013

Vitor asked whether limit=0 really means "no limit" on the Mongo side
and whether that had actually been tested. It's pymongo's own
documented convention, but nothing in the suite exercised it against a
real MongoDB with enough documents to mean anything -- the existing
tests only ever seed 2-5. Seeds 12 (more than any other test in this
file) with the batch size left at its config default (0) and asserts
all 12 come back in one run, against real testcontainers Mongo.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

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.
+- **`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. `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/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.
 

e2e/conftest.py

@@ -330,6 +330,7 @@ async def test_activities(
             'connection_string': f'mongodb://localhost:{mongo_port}',
             'database_name': E2E_DATABASE,
             'ttl_index_seconds': 3600,
+            'notification_queue_batch_size': 0,
         },
         email_config={
             'sender_email': 'e2e@example.com',

e2e/scenarios.md

@@ -6,6 +6,11 @@ Functional reference for orchestrator E2E scenarios. Tests live under `e2e/`, us
 pytest e2e/ --override-ini testpaths=e2e -m e2e
 ```
 
+**Every section below maps 1:1 to an actual test function** (named in each section, or as a bullet
+under a `Source:` heading) — this file describes what the suite verifies today, not planned or
+aspirational coverage. Adding a scenario here without a matching test (or vice versa) is a doc bug;
+fix the mismatch instead of leaving it for later.
+
 ## Execution context
 
 - MongoDB, Redis, PostgreSQL: testcontainers (session-scoped).
@@ -21,31 +26,20 @@ Source: `e2e/test_orchestrator_main_workflow.py`
 
 ### 1.1.1 Happy path
 
-Pipelines in Mongo → schedules created in correct namespace/task queue, Redis slots written, `orchestrated_schedules` updated.
-
-### 1.2.1 No-op tick
-
-Mongo, Redis, and Temporal already match desired state → no new schedules or slot writes.
+Pipelines in Mongo → Temporal schedule created in the `laborious` namespace, `orchestrated_schedules` entries written.
 
 ### 1.3.1 Create-only
 
-New pipeline only → schedules created, timestamps inserted.
-
-### 1.3.2 Update-only
-
-Existing pipeline with newer `updated_at` → schedule updated in Temporal.
+New pipeline only → Temporal schedule created for it.
 
 ### 1.3.3 Delete-only
 
-Pipeline removed from Mongo → schedule deleted from Temporal.
-
-### 1.4.1 Conflict ordering
-
-Pipeline update and slot delete on same OPC server → slot insert before delete (production ordering).
+Pipeline removed from Mongo → previously-created Temporal schedule deleted.
 
 ### 1.5.1 Empty pipelines
 
-No active pipelines → orphan schedules removed, no new orchestration writes.
+No active pipelines → no `orchestrated_schedules` documents written, even with a pre-existing
+unrelated Temporal schedule present.
 
 ### 1.6.1 TTL index bootstrap
 
@@ -65,10 +59,6 @@ ERROR notification → one SMTP message, one `log_report` row, Redis cache key.
 
 Second run with same data and cache seeded → no extra email or log row.
 
-### A.1.3 Persistent escalation
-
-Alert past `notification_ttl` with cache cleared → new email sent.
-
 ### A.2.1 Group filtering
 
 Receiver group `levels` / `ignore_models_id` honored, both halves now covered:
@@ -110,6 +100,8 @@ No notifications → no SMTP, no Postgres row.
 - Prior timestamp → only newer notifications returned.
 - Query-time filter (SIENTIAPDE-2011) → union of active groups' `levels`, intersection of their `ignore_models_id`; a model ignored by only one of several active groups is not excluded from the fetch.
 - Empty Mongo → no Redis timestamp write.
+- Batch size cap (SIENTIAPDE-2013) → with 5 pending notifications and a batch size of 2, each run returns exactly the next-oldest 2 (sorted ascending by timestamp), and the Redis cursor advances only to what was actually returned, so three successive runs drain the backlog in order instead of skipping anything.
+- Batch size 0 means unlimited (SIENTIAPDE-2013) → the configured default (`0`) does not cap the query; 12 pending notifications all come back in one run, verified against a real MongoDB instance rather than assumed from pymongo's documented convention.
 
 ### ProcessNotifications — `e2e/test_subworkflow_process_notifications.py`
 

e2e/test_subworkflow_load_notification_package.py

@@ -228,3 +228,114 @@ async def test_load_package_empty_mongo_no_redis_write(
     )
 
     assert redis_client.get('notification_last_timestamp:Alerts') is None
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_load_package_batch_size_caps_and_returns_oldest_first(
+    temporal_env: WorkflowEnvironment,
+    orchestrator_worker,
+    stub_workers,
+    mongo_uri: str,
+    redis_client: Redis,
+    test_activities,
+):
+    """
+    SIENTIAPDE-2013: a capped run returns the oldest pending documents first,
+    and the incremental cursor picks up the remainder on the next run.
+
+    Five notifications exist, but the batch size is capped at 2. Each run must
+    return exactly the next-oldest 2 (never an arbitrary subset), and the
+    Redis cursor must advance only to the max timestamp of what was actually
+    returned -- not the true max across all 5 -- or the skipped documents
+    would never be fetched again.
+    """
+    test_activities.notification_queue_batch_size = 2
+
+    seed_receiver_groups(mongo_uri, E2E_DATABASE, [default_receiver_group()])
+    seed_notifications(
+        mongo_uri,
+        E2E_DATABASE,
+        [
+            default_notification(
+                notification_id=f'n{i}',
+                timestamp=datetime(2024, 6, 1, 12, i, 0, tzinfo=UTC),
+            )
+            for i in range(1, 6)
+        ],
+    )
+
+    input_data = load_scenario_input('subworkflow_load_notification_package.json')
+    _build_metadata(input_data)
+
+    first_run = await start_and_await_workflow(
+        temporal_env.client,
+        LoadNotificationPackage.run,
+        input_data,
+        make_workflow_id('load-pkg-batch-1'),
+    )
+    assert [n['notification_id'] for n in first_run['notification_package']] == ['n1', 'n2']
+
+    second_run = await start_and_await_workflow(
+        temporal_env.client,
+        LoadNotificationPackage.run,
+        input_data,
+        make_workflow_id('load-pkg-batch-2'),
+    )
+    assert [n['notification_id'] for n in second_run['notification_package']] == ['n3', 'n4']
+
+    third_run = await start_and_await_workflow(
+        temporal_env.client,
+        LoadNotificationPackage.run,
+        input_data,
+        make_workflow_id('load-pkg-batch-3'),
+    )
+    assert [n['notification_id'] for n in third_run['notification_package']] == ['n5']
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_load_package_batch_size_zero_means_unlimited(
+    temporal_env: WorkflowEnvironment,
+    orchestrator_worker,
+    stub_workers,
+    mongo_uri: str,
+    test_activities,
+):
+    """
+    SIENTIAPDE-2013: notification_queue_batch_size=0 (the config default) must
+    NOT cap the query. This exercises pymongo's own documented convention --
+    `Collection.find(..., limit=0)` is equivalent to no limit at all -- against
+    a real MongoDB instance rather than trusting the convention on faith.
+
+    12 notifications is deliberately more than any other test in this file
+    seeds, so a regression that silently caps at some smaller default (e.g. if
+    `limit=0` were ever misread as "return nothing", or pymongo's own behavior
+    changed) would fail loudly here instead of passing by coincidence.
+    """
+    test_activities.notification_queue_batch_size = 0
+
+    seed_receiver_groups(mongo_uri, E2E_DATABASE, [default_receiver_group()])
+    seed_notifications(
+        mongo_uri,
+        E2E_DATABASE,
+        [
+            default_notification(
+                notification_id=f'n{i}',
+                timestamp=datetime(2024, 6, 1, 12, i, 0, tzinfo=UTC),
+            )
+            for i in range(1, 13)
+        ],
+    )
+
+    input_data = load_scenario_input('subworkflow_load_notification_package.json')
+    _build_metadata(input_data)
+
+    result = await start_and_await_workflow(
+        temporal_env.client,
+        LoadNotificationPackage.run,
+        input_data,
+        make_workflow_id('load-pkg-batch-unlimited'),
+    )
+
+    assert len(result['notification_package']) == 12

orchestrator/activities/activities.py

@@ -84,6 +84,7 @@ class Activities(TemporalManager, SlotManager, Formatters, MongoDB, Email, Postg
             connection_string=mongodb_config['connection_string'],
             database_name=mongodb_config['database_name'],
             ttl_index_seconds=mongodb_config['ttl_index_seconds'],
+            notification_queue_batch_size=mongodb_config['notification_queue_batch_size'],
             logger=logger,
             notification_handler=notification_handler,
             metrics_controller=metrics_controller,

orchestrator/activities/mongo_db.py

@@ -31,6 +31,8 @@ class MongoDB(SientiaMonitoring):
         ttl_index_seconds (int): TTL index duration in seconds
         logger (Logger): Application logger instance
         notification_handler (NotificationHandler): Notification management handler
+        notification_queue_batch_size (int): Hard cap on documents fetched per
+            `load_latest_data` run (SIENTIAPDE-2013). `0` means no limit.
     """
 
     def __init__(
@@ -41,6 +43,7 @@ class MongoDB(SientiaMonitoring):
         logger: Logger,
         notification_handler: NotificationHandler,
         metrics_controller: MetricsController,
+        notification_queue_batch_size: int = 0,
     ):
         SientiaMonitoring.__init__(
             self,
@@ -61,6 +64,7 @@ class MongoDB(SientiaMonitoring):
         )
 
         self.ttl_index_seconds = ttl_index_seconds
+        self.notification_queue_batch_size = notification_queue_batch_size
 
         # Initialize MongoDB client here (omitted for brevity)
         logger.info('MongoDB connection initialized')
@@ -447,6 +451,16 @@ 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.
+
         Args:
             input_data (dict[str, Any]): Activity input parameters.
                 Required fields:
@@ -488,7 +502,13 @@ class MongoDB(SientiaMonitoring):
 
             self.debug(f'Data filter: {data_filter}', metadata=metadata)
 
-            data = self.mongo_db_repository.find(collection_name, data_filter, metadata)
+            data = self.mongo_db_repository.find(
+                collection_name,
+                data_filter,
+                metadata,
+                limit=self.notification_queue_batch_size,
+                sort=[('timestamp', 1)],
+            )
 
             self.debug(f'Collected: {data}', metadata=metadata)
 

orchestrator/utils/connectors_config.py

@@ -21,7 +21,8 @@ def build_mongodb_config():
     Build MongoDB configuration from environment variables.
 
     Returns:
-        dict: MongoDB configuration with connection string, database name, and TTL index seconds.
+        dict: MongoDB configuration with connection string, database name, TTL index seconds,
+            and the notification queue's per-run batch size cap.
     """
     username = getenv('MONGODB_USERNAME', 'root')
     password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c')
@@ -32,6 +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')),
     }
 
 

pyproject.toml

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

requirements.txt

@@ -4,5 +4,5 @@ sqlalchemy
 redis
 pymongo
 jinja2
-sientia_do>=1.12.2
+sientia_do>=1.13.0
 prometheus-client

tests/orchestrator/activities/test_activities.py

@@ -27,6 +27,7 @@ def test___init__(
         'connection_string': 'mongodb://localhost:27017',
         'database_name': 'test_db',
         'ttl_index_seconds': 3600,
+        'notification_queue_batch_size': 5000,
     }
 
     redis_config = {'host': 'localhost', 'port': 6379, 'username': 'admin', 'password': 'password'}
@@ -89,6 +90,7 @@ def test___init__(
         connection_string='mongodb://localhost:27017',
         database_name='test_db',
         ttl_index_seconds=3600,
+        notification_queue_batch_size=5000,
         logger=logger,
         notification_handler=notification_handler,
         metrics_controller=mock_metrics_controller.return_value,

tests/orchestrator/activities/test_mongo_db.py

@@ -31,13 +31,14 @@ def test___init__(mongo_mock):
     notification_handler = MagicMock()
     metrics_controller = AsyncMock()
 
-    MongoDB(
+    mongo = MongoDB(
         connection_string='mongodb://localhost:27017',
         database_name='test_db',
         ttl_index_seconds=3600,
         logger=logger,
         notification_handler=notification_handler,
         metrics_controller=metrics_controller,
+        notification_queue_batch_size=5000,
     )
 
     mongo_mock.assert_called_once_with(
@@ -47,6 +48,22 @@ def test___init__(mongo_mock):
         notification_handler=notification_handler,
         metrics_controller=metrics_controller,
     )
+    assert mongo.notification_queue_batch_size == 5000
+
+
+@patch('orchestrator.activities.mongo_db.MongoDBRepository')
+def test___init___notification_queue_batch_size_defaults_to_zero(mongo_mock):
+    """0 means no limit, matching the underlying find()'s own convention."""
+    mongo = MongoDB(
+        connection_string='mongodb://localhost:27017',
+        database_name='test_db',
+        ttl_index_seconds=3600,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=AsyncMock(),
+    )
+
+    assert mongo.notification_queue_batch_size == 0
 
 
 @patch('orchestrator.activities.mongo_db.SientiaMonitoring')
@@ -480,6 +497,8 @@ def test_load_latest_data_none_last_data_timestamp(mongo_db):
         'test_collection',
         {'level': 'ERROR'},
         {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
+        limit=0,
+        sort=[('timestamp', 1)],
     )
 
     assert result == [
@@ -487,6 +506,29 @@ def test_load_latest_data_none_last_data_timestamp(mongo_db):
     ]
 
 
+def test_load_latest_data_passes_configured_batch_size_as_limit(mongo_db):
+    """SIENTIAPDE-2013: the configured notification_queue_batch_size caps the query."""
+    mongo_db.notification_queue_batch_size = 100
+    mongo_db.mongo_db_repository.find = MagicMock(return_value=[])
+
+    mongo_db.load_latest_data(
+        {
+            'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
+            'collection_name': 'test_collection',
+            'last_data_timestamp': None,
+            'base_data_filter': {'level': 'ERROR'},
+        }
+    )
+
+    mongo_db.mongo_db_repository.find.assert_called_once_with(
+        'test_collection',
+        {'level': 'ERROR'},
+        {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
+        limit=100,
+        sort=[('timestamp', 1)],
+    )
+
+
 def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
     """Test load_latest_data compares native datetime against native datetime in $gt"""
 
@@ -516,6 +558,8 @@ def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
             'timestamp': {'$gt': datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)},
         },
         {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
+        limit=0,
+        sort=[('timestamp', 1)],
     )
 
     assert result == [
@@ -529,7 +573,7 @@ def test_load_latest_data_run_boundary_no_skip_or_duplicate(mongo_db):
 
     boundary_timestamp = datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)
 
-    def fake_find(collection_name, data_filter, metadata):
+    def fake_find(collection_name, data_filter, metadata, limit=0, sort=None):
         gt = data_filter.get('timestamp', {}).get('$gt')
         all_docs = [
             {

tests/orchestrator/utils/test_connectors_config.py

@@ -64,10 +64,12 @@ def test_build_mongo_db_config_with_env_vars():
     environ['MONGODB_URL'] = 'localhost:27018'
     environ['MONGODB_DATABASE_NAME'] = 'test_db'
     environ['MONGODB_TTL_INDEX_HOURS'] = '2'
+    environ['NOTIFICATION_QUEUE_BATCH_SIZE'] = '2000'
     assert build_mongodb_config() == {
         'connection_string': 'mongodb://sientia1:sientia1@localhost:27018',
         'database_name': 'test_db',
         'ttl_index_seconds': 7200,
+        'notification_queue_batch_size': 2000,
     }
 
 
@@ -77,10 +79,12 @@ def test_build_mongo_db_config_with_defaults():
     environ.pop('MONGODB_DATABASE_NAME', None)
     environ.pop('MONGODB_URL', None)
     environ.pop('MONGODB_TTL_INDEX_HOURS', None)
+    environ.pop('NOTIFICATION_QUEUE_BATCH_SIZE', None)
     assert build_mongodb_config() == {
         'connection_string': 'mongodb://root:wKZDbMNU1c@localhost:27018',
         'database_name': 'sientia',
         'ttl_index_seconds': 3600,
+        'notification_queue_batch_size': 5000,
     }