chore: bump version to 0.7.0

test(simple_metrics): parallel windows convention + adaptive frequency e2e (SIENTIAPDE-2193)

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. 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.
 

README.md

@@ -612,6 +612,37 @@ Temporal input configuration sample:
 }
 ```
 
+#### Simple metrics: parallel windows & adaptive frequency
+
+One `pipelines` doc per window, all with the same `model_id`. Name them `<model>-simple_metrics-<interval_minutes>m`; each doc becomes its own Temporal schedule. Thresholds are shared through `model.model_config`. The adaptive state is per doc.
+
+```json
+{
+    "schedule_name": "model-x-simple_metrics-60m",
+    "workflow_type": "simple_metrics",
+    "model_id": "model-x",
+    "active": true,
+    "interval_minutes": 60,
+    "frequency": "1h",
+    "frequency_alert": "5m"
+}
+```
+
+| Field | Meaning | Default |
+|---|---|---|
+| `frequency_alert` | Frequency while escalated, same format as `frequency`. Absent = adaptive mode off. Below `1m` (or unparseable), the orchestrator sends a WARNING each time it creates or updates the schedule. A value below `1m` is kept; an unparseable value is ignored (the schedule keeps `frequency`) | — |
+| `adaptive_state` | `{escalated, healthy_streak}`, written **only** by laborious (with `updated_at` as a BSON datetime when `escalated` flips) | `{false, 0}` |
+| `model.model_config.simple_metrics_thresholds` | `{<metric>_max, <metric>_min}` shared by all windows | — |
+| `model.model_config.simple_metrics_min_samples` | Windows with fewer rows neither alert nor escalate, and count toward recovery | — |
+
+Escalation, recovery and the per-model alert are implemented in laborious (SIENTIAPDE-2189, SIENTIAPDE-2191); the orchestrator only picks `frequency_alert` while `adaptive_state.escalated` is true.
+
+Rules:
+- **Escalation:** laborious escalates on the first breach.
+- **Recovery:** after `min(12, ceil(interval_minutes / frequency_alert))` consecutive runs without one. Empty windows count as runs without a breach.
+- **Alerts** are per model (trigger `<model_id>-simple_metrics`), so several windows seeing the same breach produce one alert.
+- **Delay:** the schedule switches on the orchestrator tick after laborious flips the state, at most one tick later.
+
 #### Alerts Workflow
 
 ```json

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/helpers.py

@@ -347,6 +347,7 @@ def default_receiver_group(
     levels: list[str] | None = None,
     contents: list[str] | None = None,
     ignore_models_id: list[str] | None = None,
+    model_ids: list[str] | None = None,
 ) -> dict[str, Any]:
     """Minimal active receiver_groups document."""
     return {
@@ -356,6 +357,7 @@ def default_receiver_group(
         'levels': levels or ['ERROR', 'WARNING', 'INFO'],
         'contents': contents or ['core_alerts', 'persistent_alerts', 'reports'],
         'ignore_models_id': ignore_models_id or [],
+        'model_ids': model_ids or [],
     }
 
 

e2e/scenario_inputs/orchestrator_simple_metrics_windows.json

@@ -0,0 +1,45 @@
+{
+  "workflow_input": {
+    "schedule_name": "orchestrator-e2e-sm-windows",
+    "pipelines_query": {
+      "collection": "pipelines",
+      "aggregation": [{"$match": {"active": true}}]
+    },
+    "opc_servers_query": {
+      "collection": "opc_servers",
+      "filters": {"active": true}
+    }
+  },
+  "pipelines": [
+    {
+      "schedule_name": "model-sm-simple_metrics-60m",
+      "workflow_type": "simple_metrics",
+      "model_id": "model-sm",
+      "model": {
+        "name": "Model model-sm",
+        "model_config": {"target": "y", "simple_metrics_thresholds": {"rmse_max": 1.0}}
+      },
+      "active": true,
+      "updated_at": "@now",
+      "frequency": "1h",
+      "frequency_alert": "5m",
+      "interval_minutes": 60
+    },
+    {
+      "schedule_name": "model-sm-simple_metrics-1440m",
+      "workflow_type": "simple_metrics",
+      "model_id": "model-sm",
+      "model": {
+        "name": "Model model-sm",
+        "model_config": {"target": "y", "simple_metrics_thresholds": {"rmse_max": 1.0}}
+      },
+      "active": true,
+      "updated_at": "@now",
+      "frequency": "1d",
+      "frequency_alert": "1h",
+      "interval_minutes": 1440
+    }
+  ],
+  "opc_servers": [],
+  "active_ingestors": ["heartbeat:ingestor:1"]
+}

e2e/scenarios.md

@@ -45,6 +45,17 @@ unrelated Temporal schedule present.
 
 First run creates TTL index on notification collection used by scouter pipelines.
 
+### 1.7.1 Simple metrics parallel windows
+
+Two `simple_metrics` pipelines for the same `model_id` (60 min and 1440 min windows) → one
+Temporal schedule per window, each with its own `frequency` as interval.
+
+### 1.7.2 Adaptive frequency escalation/recovery round trip
+
+Laborious-style write (`adaptive_state.escalated=true` + BSON-datetime `updated_at`) on one
+window → next run switches only that schedule to `frequency_alert`; the recovery write switches it
+back to `frequency`.
+
 ---
 
 ## 2. Alerts workflow
@@ -68,6 +79,12 @@ the query-time union/intersection across active groups (SIENTIAPDE-2011, see
 `levels`/`ignore_models_id`, even when the query-time filter widened the fetch
 to satisfy another active group's broader configuration.
 
+### A.2.2 Model_id allow-list
+
+Receiver group `model_ids` (SIENTIAPDE-2149) honored: a group configuring it only
+receives notifications from those models, even when another active group with
+no such restriction widens the query-time fetch to include every model.
+
 ### A.3.1 Empty queue
 
 No notifications → no SMTP, no Postgres row.
@@ -99,7 +116,10 @@ No notifications → no SMTP, no Postgres row.
 - No prior Redis timestamp → all matching notifications returned, max timestamp stored.
 - 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.
+- Model_id allow-list union at query time (SIENTIAPDE-2149) → when every active group sets `model_ids`, the fetch is restricted to the union of every group's allow-list; a model outside that union is never fetched at all, even for a group that would otherwise want it via a different axis.
 - 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_alerts_main_workflow.py

@@ -182,6 +182,73 @@ async def test_scenario_a_2_1_per_group_levels_and_ignore_models_id(
     assert 'model-x' not in combined
 
 
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_scenario_a_2_2_model_ids_allow_list(
+    temporal_env: WorkflowEnvironment,
+    orchestrator_worker,
+    stub_workers,
+    mongo_uri: str,
+    smtp_server,
+):
+    """
+    A.2.2 (SIENTIAPDE-2149): a group configuring `model_ids` only receives
+    notifications from those models, on top of the existing query-time/per-group
+    filtering (SIENTIAPDE-2011/2012).
+
+    group-allow (model_ids=['model-a']) coexists with group-everything (no
+    model_ids restriction). Because the query-time filter only restricts to an
+    allow-list union when EVERY active group has one, group-everything's lack
+    of a model_ids restriction means the query itself fetches both models --
+    the per-group filter added by this ticket is what narrows group-allow back
+    down to model-a only.
+    """
+    seed_receiver_groups(
+        mongo_uri,
+        E2E_DATABASE,
+        [
+            default_receiver_group(
+                group_name='group-allow', contents=['core_alerts'], model_ids=['model-a']
+            ),
+            default_receiver_group(group_name='group-everything', contents=['core_alerts']),
+        ],
+    )
+    seed_notifications(
+        mongo_uri,
+        E2E_DATABASE,
+        [
+            default_notification(
+                notification_id='err-a', level='ERROR', model_name='model-a', model_id='model-a'
+            ),
+            default_notification(
+                notification_id='err-b',
+                level='ERROR',
+                model_name='model-b',
+                model_id='model-b',
+                timestamp=datetime(2024, 6, 1, 12, 1, 0, tzinfo=UTC),
+            ),
+        ],
+    )
+
+    await start_and_await_workflow(
+        temporal_env.client,
+        Alerts.run,
+        load_scenario_input('alerts_happy_path.json'),
+        make_workflow_id('alerts-2149'),
+    )
+
+    # One email per group -- group-allow got only err-a, group-everything got both.
+    assert len(smtp_server.messages) == 2
+    bodies = [extract_html_body(message) for message in smtp_server.messages]
+
+    # Distinguish the two emails by content rather than order: only
+    # group-everything's email mentions model-b.
+    allow_body = next(b for b in bodies if 'model-b' not in b)
+    everything_body = next(b for b in bodies if 'model-b' in b)
+    assert 'model-a' in allow_body
+    assert 'model-a' in everything_body
+
+
 @pytest.mark.e2e
 @pytest.mark.asyncio
 async def test_scenario_a_3_1_empty_queue(

e2e/test_orchestrator_main_workflow.py

@@ -1,5 +1,7 @@
 """E2E tests for the Orchestrator main workflow."""
 
+from datetime import UTC, datetime, timedelta
+
 import pytest
 from pymongo import MongoClient
 from redis import Redis
@@ -26,6 +28,13 @@ async def _schedule_ids(host: str, namespace: str) -> list[str]:
     return [schedule.id async for schedule in await client.list_schedules()]
 
 
+async def _schedule_every(host: str, schedule_id: str) -> timedelta:
+    """Return the interval of a laborious-namespace schedule."""
+    client = await Client.connect(host, namespace='laborious')
+    description = await client.get_schedule_handle(schedule_id).describe()
+    return description.schedule.spec.intervals[0].every
+
+
 def _apply_orchestrator_seeds(
     scenario: dict,
     mongo_uri: str,
@@ -224,3 +233,83 @@ async def test_scenario_1_6_1_ttl_index_bootstrap(
         assert any('expireAfterSeconds' in info for info in indexes.values())
     finally:
         mongo.close()
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_scenario_1_7_1_simple_metrics_parallel_windows(
+    temporal_env: WorkflowEnvironment,
+    orchestrator_worker,
+    stub_workers,
+    mongo_uri: str,
+    temporal_host: str,
+    redis_client: Redis,
+):
+    """1.7.1: One pipelines doc per window gives one schedule per window, each with its own interval."""
+    scenario = load_scenario_input('orchestrator_simple_metrics_windows')
+    _apply_orchestrator_seeds(scenario, mongo_uri, redis_client)
+
+    await start_and_await_workflow(
+        temporal_env.client,
+        Orchestrator.run,
+        scenario['workflow_input'],
+        make_workflow_id('orchestrator-sm-windows'),
+    )
+
+    short, long = 'model-sm-simple_metrics-60m', 'model-sm-simple_metrics-1440m'
+    assert await _schedule_every(temporal_host, short) == timedelta(hours=1)
+    assert await _schedule_every(temporal_host, long) == timedelta(days=1)
+
+
+def _laborious_adaptive_write(mongo_uri: str, schedule_name: str, escalated: bool) -> None:
+    """Replicates laborious `update_simple_metrics_adaptive_state` on an escalated flip."""
+    mongo = MongoClient(mongo_uri)
+    try:
+        mongo[E2E_DATABASE]['pipelines'].update_one(
+            {'schedule_name': schedule_name},
+            {
+                '$set': {
+                    'adaptive_state': {'escalated': escalated, 'healthy_streak': 0},
+                    # BSON datetime, as laborious writes it; a string would break the tick.
+                    'updated_at': datetime.now(UTC),
+                }
+            },
+        )
+    finally:
+        mongo.close()
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_scenario_1_7_2_simple_metrics_adaptive_frequency_round_trip(
+    temporal_env: WorkflowEnvironment,
+    orchestrator_worker,
+    stub_workers,
+    mongo_uri: str,
+    temporal_host: str,
+    redis_client: Redis,
+):
+    """1.7.2: Escalation written by laborious switches only that window to frequency_alert; recovery switches it back."""
+    scenario = load_scenario_input('orchestrator_simple_metrics_windows')
+    _apply_orchestrator_seeds(scenario, mongo_uri, redis_client)
+    short, long = 'model-sm-simple_metrics-60m', 'model-sm-simple_metrics-1440m'
+
+    async def run(tag: str) -> None:
+        await start_and_await_workflow(
+            temporal_env.client,
+            Orchestrator.run,
+            scenario['workflow_input'],
+            make_workflow_id(f'orchestrator-sm-adaptive-{tag}'),
+        )
+
+    await run('create')
+    assert await _schedule_every(temporal_host, short) == timedelta(hours=1)
+
+    _laborious_adaptive_write(mongo_uri, short, escalated=True)
+    await run('escalate')
+    assert await _schedule_every(temporal_host, short) == timedelta(minutes=5)
+    assert await _schedule_every(temporal_host, long) == timedelta(days=1)
+
+    _laborious_adaptive_write(mongo_uri, short, escalated=False)
+    await run('recover')
+    assert await _schedule_every(temporal_host, short) == timedelta(hours=1)

e2e/test_subworkflow_load_notification_package.py

@@ -205,6 +205,57 @@ async def test_load_package_query_time_filter_union_and_intersection(
     assert ids == {'error-included', 'warning-included', 'modely-included'}
 
 
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_load_package_query_time_filter_model_ids_union(
+    temporal_env: WorkflowEnvironment,
+    orchestrator_worker,
+    stub_workers,
+    mongo_uri: str,
+    redis_client: Redis,
+):
+    """
+    SIENTIAPDE-2149: query-time filter unions active groups' `model_ids`
+    allow-list when every active group sets one, excluding any model outside
+    that union from the fetch entirely -- not just from a single group's
+    per-group slice.
+
+    group_a wants only model-a, group_b wants only model-b -- both restrict
+    levels to ERROR so only the model_ids axis is under test. model-c is
+    outside the union of both allow-lists and must never be fetched.
+    """
+    seed_receiver_groups(
+        mongo_uri,
+        E2E_DATABASE,
+        [
+            default_receiver_group(group_name='group_a', levels=['ERROR'], model_ids=['model-a']),
+            default_receiver_group(group_name='group_b', levels=['ERROR'], model_ids=['model-b']),
+        ],
+    )
+    seed_notifications(
+        mongo_uri,
+        E2E_DATABASE,
+        [
+            default_notification(notification_id='a-included', level='ERROR', model_id='model-a'),
+            default_notification(notification_id='b-included', level='ERROR', model_id='model-b'),
+            default_notification(notification_id='c-excluded', level='ERROR', model_id='model-c'),
+        ],
+    )
+
+    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-model-ids-union'),
+    )
+
+    ids = {n['notification_id'] for n in result['notification_package']}
+    assert ids == {'a-included', 'b-included'}
+
+
 @pytest.mark.e2e
 @pytest.mark.asyncio
 async def test_load_package_empty_mongo_no_redis_write(
@@ -228,3 +279,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/formatters.py

@@ -20,6 +20,7 @@ with workflow.unsafe.imports_passed_through():
     from orchestrator.utils.orchestrator_functions import (
         build_tag_config,
         drift,
+        frequency_alert_warning,
         gather_read_tags,
         minimal_retrain,
         pi_web_api_scouter,
@@ -369,6 +370,8 @@ class Formatters(SientiaMonitoring):
         Compares the current schedule configuration in Temporal with the new schedule
         configuration to determine which schedules need to be created, updated, or deleted.
         Uses timestamp comparison to identify schedules that have changed.
+        Also sends a SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR WARNING for laborious schedules
+        whose frequency_alert is below 1m (the value is kept) or unparseable (it is ignored).
 
         Args:
             input_data (dict[str, Any]): The input data containing:
@@ -409,6 +412,25 @@ class Formatters(SientiaMonitoring):
                 schedules, current_schedules, to_update, to_create, namespace, metadata
             )
 
+        # Fires whenever the schedule is created or updated (each `updated_at` change,
+        # including escalation flips).
+        changed = {**to_create[self.laborious_namespace], **to_update[self.laborious_namespace]}
+        for schedule_name, schedule in changed.items():
+            warning = frequency_alert_warning(schedule.get('frequency_alert'))
+            if warning:
+                self.send_notification(
+                    metadata={
+                        **metadata,
+                        'schedule_name': schedule_name,
+                        'model_id': schedule.get('model_id'),
+                        'model_name': schedule.get('model_name'),
+                    },
+                    notification_id='SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR',
+                    message=warning,
+                    block='create_schedule_config',
+                    level=NotificationLevel.WARNING,
+                )
+
         for namespace, schedules in current_schedule_config.items():
             for schedule_name in schedules:
                 if schedule_name not in schedule_config[namespace]:
@@ -826,6 +848,7 @@ class Formatters(SientiaMonitoring):
                     - ignore (list, optional): Notification IDs to exclude
                     - levels (list, optional): Notification levels to include
                     - ignore_models_id (list, optional): Model IDs to exclude
+                    - model_ids (list, optional): Model IDs to allow (SIENTIAPDE-2149)
 
         Returns:
             dict[str, Any]: Filtered receiver groups with their notifications, keyed by group_name.
@@ -851,6 +874,7 @@ class Formatters(SientiaMonitoring):
             ignore_list = receiver_group.get('ignore', [])
             levels_list = receiver_group.get('levels')
             ignore_models_id_list = receiver_group.get('ignore_models_id') or []
+            model_ids_list = receiver_group.get('model_ids')
 
             for notification in notification_package:
                 alert_type = 'reports'
@@ -868,6 +892,7 @@ class Formatters(SientiaMonitoring):
                         not ignore_models_id_list
                         or notification.get('model_id') not in ignore_models_id_list
                     )
+                    and (not model_ids_list or notification.get('model_id') in model_ids_list)
                 ):
                     receiver_groups[group_name]['notifications'].append(notification)
                     already_added_keys.append(key)

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')
@@ -238,7 +242,12 @@ class MongoDB(SientiaMonitoring):
             for pipeline in updated_pipelines
             if pipeline['success']
         ]
-        data_filter = {'$or': argument} if argument else {}
+        if not argument:
+            # Every update failed: stamp nothing so they are retried next tick (same guard as
+            # create_pipelines_timestamps). An empty filter would match every tracked schedule.
+            self.info('No successful pipeline updates to timestamp', metadata=metadata)
+            return
+        data_filter = {'$or': argument}
 
         try:
             self.mongo_db_repository.update_many(
@@ -339,7 +348,13 @@ class MongoDB(SientiaMonitoring):
             for pipeline in deleted_pipelines
             if pipeline['success']
         ]
-        data_filter = {'$or': argument} if argument else {}
+        if not argument:
+            # Every delete failed: remove nothing (same guard as create_pipelines_timestamps).
+            # An empty filter would wipe orchestrated_schedules and make the next tick delete
+            # and recreate every orchestrated schedule.
+            self.info('No successful pipeline deletions to untrack', metadata=metadata)
+            return
+        data_filter = {'$or': argument}
 
         try:
             self.mongo_db_repository.delete_many('orchestrated_schedules', data_filter, metadata)
@@ -447,6 +462,10 @@ class MongoDB(SientiaMonitoring):
         filtering by timestamp to enable incremental data processing. It
         handles connection management and provides comprehensive error reporting.
 
+        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.
                 Required fields:
@@ -488,7 +507,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/activities/slot_manager.py

@@ -398,6 +398,7 @@ class SlotManager(SientiaMonitoring):
                     - ignore (list, optional): Notification IDs to exclude
                     - levels (list, optional): Notification levels to include
                     - ignore_models_id (list, optional): Model IDs to exclude
+                    - model_ids (list, optional): Model IDs to allow (SIENTIAPDE-2149)
                 - notification_ttl (int): Seconds before considering notification persistent
 
         Returns:
@@ -450,6 +451,7 @@ class SlotManager(SientiaMonitoring):
         ignore_list = receiver_group.get('ignore', [])
         levels_list = receiver_group.get('levels')
         ignore_models_id_list = receiver_group.get('ignore_models_id') or []
+        model_ids_list = receiver_group.get('model_ids')
 
         notifications = []
         for notification in notification_package:
@@ -462,8 +464,8 @@ class SlotManager(SientiaMonitoring):
                 and self._check_keys_and_notif_id(
                     notification_id, ignore_list, key, already_added_keys
                 )
-                and self._check_levels_and_ignore_models_id(
-                    notification, levels_list, ignore_models_id_list
+                and self._check_levels_and_model_filters(
+                    notification, levels_list, ignore_models_id_list, model_ids_list
                 )
             ):
                 notifications.append(notification)
@@ -493,23 +495,29 @@ class SlotManager(SientiaMonitoring):
         return 'do_nothing'
 
     @staticmethod
-    def _check_levels_and_ignore_models_id(notification, levels_list, ignore_models_id_list):
+    def _check_levels_and_model_filters(
+        notification, levels_list, ignore_models_id_list, model_ids_list
+    ):
         """
-        Check if the notification's level is in the levels_list and its model_id is not in the ignore_models_id_list.
+        Check the notification's level and model_id against a group's filters.
 
         Args:
             notification (dict): The notification to check.
             levels_list (list, optional): List of levels to include. If None, all levels are included.
             ignore_models_id_list (list, optional): List of model_ids to exclude. If None, no models are excluded.
+            model_ids_list (list, optional): List of model_ids to allow (SIENTIAPDE-2149).
+                If None/empty, all models are allowed. Independent of ignore_models_id_list --
+                both apply together when both are configured.
 
         Returns:
             bool: True if the notification passes the checks, False otherwise.
         """
         level_check = not levels_list or notification.get('level') in levels_list
-        model_id_check = (
+        ignore_models_id_check = (
             not ignore_models_id_list or notification.get('model_id') not in ignore_models_id_list
         )
-        return level_check and model_id_check
+        model_ids_check = not model_ids_list or notification.get('model_id') in model_ids_list
+        return level_check and ignore_models_id_check and model_ids_check
 
     @staticmethod
     def _check_keys_and_notif_id(

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', '0')),
     }
 
 

orchestrator/utils/orchestrator_functions.py

@@ -85,6 +85,9 @@ def simple_metrics(config: dict[str, Any]):
             - interval_minutes (int, optional): Computation interval in minutes (default: 60)
             - metrics (list[str], optional): List of metrics to compute
                 (default: ['rmse', 'mse', 'mae', 'r2'])
+            - frequency_alert (str, optional): Frequency used while `adaptive_state.escalated`
+                is true; same format as `frequency`. Absent = adaptive frequency off.
+            - adaptive_state (dict, optional): `{escalated, healthy_streak}`, written by laborious.
             - Additional fields from common_config
 
     Returns:
@@ -92,17 +95,56 @@ def simple_metrics(config: dict[str, Any]):
             actual data), target table, time interval, and metrics list. Results stored in
             'simple_metrics' table
     """
+    base = common_config(config)
+    frequency_alert = config.get('frequency_alert')
+    # SIENTIAPDE-1995: laborious flips `adaptive_state.escalated` on a threshold breach (and back
+    # after the window is renewed without one) and bumps `updated_at`, which makes this tick
+    # rewrite the schedule.
+    state = config.get('adaptive_state')
+    # Hand-written docs: a malformed state must not raise and kill the tick for every model.
+    if frequency_alert and isinstance(state, dict) and state.get('escalated') is True:
+        try:
+            # Temporal parses the schedule frequency on create/update; an unparseable value
+            # would fail it, so keep `frequency` (create_schedule_config still warns).
+            parse_frequency(str(frequency_alert))
+            base['frequency'] = frequency_alert
+        except ValueError:
+            pass
     return {
-        **common_config(config),
+        **base,
         'schema': 'sientia_data',
         'predictions_table_name': 'predictions',
         'data_table_name': 'laborious_data',
         'target_table_name': 'simple_metrics',
         'interval_minutes': config.get('interval_minutes', 60),
         'metrics': config.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
+        # Forwarded so create_schedule_config can warn on a value below the floor.
+        'frequency_alert': frequency_alert,
     }
 
 
+MIN_FREQUENCY_ALERT_SECONDS = 60
+
+
+def frequency_alert_warning(frequency_alert: str | None) -> str | None:
+    """
+    Warning message when `frequency_alert` is below the 1-minute floor or does not parse,
+    otherwise None. Warn only (SIENTIAPDE-1995 decision 7): a value below the floor is kept;
+    an unparseable one is ignored by `simple_metrics()`, which keeps `frequency`.
+    """
+    if not frequency_alert:
+        return None
+    try:
+        # str(): a non-string value (e.g. a number typed into the doc) must warn, not raise
+        # AttributeError and fail the whole create_schedule_config activity.
+        seconds = parse_frequency(str(frequency_alert))
+    except ValueError:
+        return f'frequency_alert={frequency_alert!r} is not a valid frequency'
+    if seconds < MIN_FREQUENCY_ALERT_SECONDS:
+        return f'frequency_alert={frequency_alert} is below the 1m floor; kept as configured'
+    return None
+
+
 def minimal_retrain(config: dict[str, Any]):
     """
     Build minimal retrain configuration from pipeline config.

orchestrator/workflows/subworkflows/load_notification_package.py

@@ -137,18 +137,21 @@ class LoadNotificationPackage:
           is only safe to drop from the fetch if EVERY active group ignores it.
           A group with no `ignore_models_id` is treated as "ignores nothing" (empty
           set), which naturally zeroes out the intersection via set math.
+        - model_ids (SIENTIAPDE-2149): union of every group's `model_ids` allow-list,
+          mirroring `levels`' own union-or-nothing logic. If ANY active group has no
+          restriction (missing/empty `model_ids`), that group wants every model, so
+          the query can't restrict to an allow-list at all. `model_ids` and
+          `ignore_models_id` are independent axes and can both narrow the same
+          `model_id` field at once (Mongo supports multiple operators on one field).
         """
         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)
 
@@ -160,7 +163,40 @@ class LoadNotificationPackage:
         ]
         models_intersection = set.intersection(*sending_configs_ignore_models_id)
 
+        model_ids_union = self._union_when_all_restrict(
+            [group.get('model_ids') for group in sending_configs]
+        )
+
+        model_id_filter: dict[str, Any] = {}
+        if model_ids_union is not None:
+            model_id_filter['$in'] = model_ids_union
         if models_intersection:
-            query_filter['model_id'] = {'$nin': sorted(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.7.0"
 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

requirements.txt

@@ -4,5 +4,5 @@ sqlalchemy
 redis
 pymongo
 jinja2
-sientia_do>=1.12.2
+sientia_do>=1.12.3
 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_formatters.py

@@ -852,6 +852,123 @@ def test_filter_notification_reports_ignore_models_id_restriction(formatters):
     }
 
 
+def test_filter_notification_reports_model_ids_allow_list_restriction(formatters):
+    # SIENTIAPDE-2149: a group with `model_ids` set only receives notifications
+    # from those models, unlike `ignore_models_id` which is a deny-list.
+    input_data = {
+        **metadata,
+        'notification_package': [
+            {
+                'trigger': 'test_trigger_1',
+                'notification_id': 'test_notification_id_1',
+                'model_id': 'model_x',
+            },
+            {
+                'trigger': 'test_trigger_2',
+                'notification_id': 'test_notification_id_2',
+                'model_id': 'model_y',
+            },
+        ],
+        'sending_configs': [
+            {
+                'group_name': 'test_group_1',
+                'contents': ['reports'],
+                'model_ids': ['model_x'],
+            },
+        ],
+    }
+
+    response = formatters.filter_notification_reports(input_data)
+
+    assert response == {
+        'test_group_1': {
+            'group_name': 'test_group_1',
+            'contents': ['reports'],
+            'model_ids': ['model_x'],
+            'notifications': [
+                {
+                    'trigger': 'test_trigger_1',
+                    'notification_id': 'test_notification_id_1',
+                    'model_id': 'model_x',
+                },
+            ],
+        },
+    }
+
+
+def test_filter_notification_reports_model_ids_and_ignore_models_id_together(formatters):
+    # SIENTIAPDE-2149: the allow-list and deny-list apply together (AND), not
+    # as alternatives -- a model must pass both checks.
+    input_data = {
+        **metadata,
+        'notification_package': [
+            {
+                'trigger': 'test_trigger_1',
+                'notification_id': 'test_notification_id_1',
+                'model_id': 'model_x',
+            },
+            {
+                'trigger': 'test_trigger_2',
+                'notification_id': 'test_notification_id_2',
+                'model_id': 'model_y',
+            },
+            {
+                'trigger': 'test_trigger_3',
+                'notification_id': 'test_notification_id_3',
+                'model_id': 'model_z',
+            },
+        ],
+        'sending_configs': [
+            {
+                'group_name': 'test_group_1',
+                'contents': ['reports'],
+                'model_ids': ['model_x', 'model_y'],
+                'ignore_models_id': ['model_x'],
+            },
+        ],
+    }
+
+    response = formatters.filter_notification_reports(input_data)
+
+    assert response == {
+        'test_group_1': {
+            'group_name': 'test_group_1',
+            'contents': ['reports'],
+            'model_ids': ['model_x', 'model_y'],
+            'ignore_models_id': ['model_x'],
+            'notifications': [
+                {
+                    'trigger': 'test_trigger_2',
+                    'notification_id': 'test_notification_id_2',
+                    'model_id': 'model_y',
+                },
+            ],
+        },
+    }
+
+
+def test_filter_notification_reports_model_ids_explicit_none_does_not_raise(formatters):
+    # Same defensive concern as ignore_models_id: receiver_groups has no schema
+    # validation, so model_ids can be explicitly null instead of omitted.
+    input_data = {
+        **metadata,
+        'notification_package': [
+            {
+                'trigger': 'test_trigger_1',
+                'notification_id': 'test_notification_id_1',
+                'model_id': 'model_x',
+            },
+        ],
+        'sending_configs': [
+            {'group_name': 'test_group_1', 'contents': ['reports'], 'model_ids': None},
+        ],
+    }
+
+    response = formatters.filter_notification_reports(input_data)
+
+    assert response['test_group_1']['notifications'] == input_data['notification_package']
+
+
 def test_filter_notification_reports_no_levels_or_ignore_models_id_passes_everything(
     formatters,
 ):
@@ -906,3 +1023,53 @@ def test_filter_notification_reports_ignore_models_id_explicit_none_does_not_rai
     response = formatters.filter_notification_reports(input_data)
 
     assert response['test_group_1']['notifications'] == input_data['notification_package']
+
+
+def _schedule_config_with_frequency_alert(frequency_alert):
+    return {
+        'metadata': {},
+        'current_schedule_config': {'laborious': {}},
+        'schedule_config': {
+            'laborious': {
+                'm1-simple_metrics-60m': {
+                    'workflow_type': 'simple_metrics',
+                    'model_id': 'm1',
+                    'model_name': 'm1',
+                    'frequency': '1h',
+                    'frequency_alert': frequency_alert,
+                    'updated_at': '2026-10-07',
+                }
+            },
+            'scouter': {},
+        },
+    }
+
+
+def test_create_schedule_config_warns_on_frequency_alert_below_floor(formatters):
+    result = formatters.create_schedule_config(_schedule_config_with_frequency_alert('30s'))
+
+    # The configured value is kept.
+    assert result['to_create']['laborious']['m1-simple_metrics-60m']['frequency_alert'] == '30s'
+    formatters.send_notification.assert_called_once()
+    kwargs = formatters.send_notification.call_args.kwargs
+    assert kwargs['notification_id'] == 'SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR'
+    assert kwargs['level'] == NotificationLevel.WARNING
+    assert kwargs['metadata']['schedule_name'] == 'm1-simple_metrics-60m'
+
+
+def test_create_schedule_config_warns_on_frequency_alert_below_floor_on_update(formatters):
+    input_data = _schedule_config_with_frequency_alert('30s')
+    input_data['current_schedule_config'] = {'laborious': {'m1-simple_metrics-60m': '2026-10-06'}}
+
+    result = formatters.create_schedule_config(input_data)
+
+    assert 'm1-simple_metrics-60m' in result['to_update']['laborious']
+    formatters.send_notification.assert_called_once()
+    kwargs = formatters.send_notification.call_args.kwargs
+    assert kwargs['notification_id'] == 'SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR'
+
+
+def test_create_schedule_config_no_warning_at_or_above_floor(formatters):
+    formatters.create_schedule_config(_schedule_config_with_frequency_alert('5m'))
+
+    formatters.send_notification.assert_not_called()

tests/orchestrator/activities/test_mongo_db.py

@@ -1,7 +1,7 @@
 from datetime import UTC, datetime
 from unittest.mock import ANY, AsyncMock, MagicMock, patch
 
-from pytest import fixture
+from pytest import fixture, mark
 from sientia_do.notifications.models import NotificationLevel
 from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ
 
@@ -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')
@@ -248,6 +265,41 @@ def test_update_pipelines_timestamps_success(now_mock, mongo_db):
     )
 
 
+@patch('orchestrator.activities.mongo_db.now')
+@mark.parametrize(
+    'pipelines', [[], [{'schedule_name': 'test1', 'namespace': 'test1', 'success': False}]]
+)
+def test_update_pipelines_timestamps_all_failed_stamps_nothing(now_mock, mongo_db, pipelines):
+    input_data = {'updated_pipelines': pipelines}
+    mongo_db.mongo_db_repository.update_many = MagicMock()
+
+    mongo_db.update_pipelines_timestamps(input_data)
+
+    # An empty filter would match and stamp every tracked schedule, so the failed one
+    # would never be retried.
+    mongo_db.mongo_db_repository.update_many.assert_not_called()
+
+
+@patch('orchestrator.activities.mongo_db.now')
+def test_update_pipelines_timestamps_mixed_stamps_only_successes(now_mock, mongo_db):
+    input_data = {
+        'updated_pipelines': [
+            {'schedule_name': 'test1', 'namespace': 'test1', 'success': True},
+            {'schedule_name': 'test2', 'namespace': 'test2', 'success': False},
+        ]
+    }
+    mongo_db.mongo_db_repository.update_many = MagicMock()
+
+    mongo_db.update_pipelines_timestamps(input_data)
+
+    mongo_db.mongo_db_repository.update_many.assert_called_once_with(
+        'orchestrated_schedules',
+        {'$or': [{'schedule_name': 'test1', 'namespace': 'test1'}]},
+        {'$set': {'updated_at': now_mock.return_value}},
+        {},
+    )
+
+
 @patch('orchestrator.activities.mongo_db.now')
 def test_update_pipelines_timestamps_failure(now_mock, mongo_db):
     input_data = {
@@ -345,6 +397,38 @@ def test_delete_pipelines_timestamps_success(now_mock, mongo_db):
     )
 
 
+@mark.parametrize(
+    'pipelines', [[], [{'schedule_name': 'test1', 'namespace': 'test1', 'success': False}]]
+)
+def test_delete_pipelines_timestamps_all_failed_deletes_nothing(mongo_db, pipelines):
+    input_data = {'deleted_pipelines': pipelines}
+    mongo_db.mongo_db_repository.delete_many = MagicMock()
+
+    mongo_db.delete_pipelines_timestamps(input_data)
+
+    # An empty filter would wipe orchestrated_schedules; the next tick's normalize_schedules
+    # would then delete and recreate every orchestrated Temporal schedule.
+    mongo_db.mongo_db_repository.delete_many.assert_not_called()
+
+
+def test_delete_pipelines_timestamps_mixed_deletes_only_successes(mongo_db):
+    input_data = {
+        'deleted_pipelines': [
+            {'schedule_name': 'test1', 'namespace': 'test1', 'success': True},
+            {'schedule_name': 'test2', 'namespace': 'test2', 'success': False},
+        ]
+    }
+    mongo_db.mongo_db_repository.delete_many = MagicMock()
+
+    mongo_db.delete_pipelines_timestamps(input_data)
+
+    mongo_db.mongo_db_repository.delete_many.assert_called_once_with(
+        'orchestrated_schedules',
+        {'$or': [{'schedule_name': 'test1', 'namespace': 'test1'}]},
+        {},
+    )
+
+
 @patch('orchestrator.activities.mongo_db.now')
 def test_delete_pipelines_timestamps_failure(now_mock, mongo_db):
     input_data = {
@@ -480,6 +564,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 +573,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 +625,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 +640,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/activities/test_slot_manager.py

@@ -481,6 +481,137 @@ def test_filter_notification_alerts_ignore_models_id_restriction(slot_manager):
     }
 
 
+def test_filter_notification_alerts_model_ids_allow_list_restriction(slot_manager):
+    # SIENTIAPDE-2149: a group with `model_ids` set only receives notifications
+    # from those models, unlike `ignore_models_id` which is a deny-list.
+    slot_manager.redis_repository.get = MagicMock(side_effect=[None, None])
+
+    input_data = {
+        **metadata,
+        'notification_package': [
+            {
+                'trigger': 'test_trigger_1',
+                'notification_id': 'test_notification_id_1',
+                'model_id': 'model_x',
+            },
+            {
+                'trigger': 'test_trigger_2',
+                'notification_id': 'test_notification_id_2',
+                'model_id': 'model_y',
+            },
+        ],
+        'sending_configs': [
+            {
+                'group_name': 'test_group_1',
+                'contents': ['core_alerts'],
+                'model_ids': ['model_x'],
+            },
+        ],
+        'notification_ttl': 300,
+        'mail_type': 'test_mail_type',
+    }
+
+    response = slot_manager.filter_notification_alerts(input_data)
+
+    assert response == {
+        'test_group_1': {
+            'group_name': 'test_group_1',
+            'contents': ['core_alerts'],
+            'model_ids': ['model_x'],
+            'notifications': [
+                {
+                    'trigger': 'test_trigger_1',
+                    'notification_id': 'test_notification_id_1',
+                    'model_id': 'model_x',
+                },
+            ],
+        },
+    }
+
+
+def test_filter_notification_alerts_model_ids_and_ignore_models_id_together(slot_manager):
+    # SIENTIAPDE-2149: the allow-list and deny-list apply together (AND), not
+    # as alternatives -- a model must pass both checks.
+    slot_manager.redis_repository.get = MagicMock(side_effect=[None, None, None])
+
+    input_data = {
+        **metadata,
+        'notification_package': [
+            {
+                'trigger': 'test_trigger_1',
+                'notification_id': 'test_notification_id_1',
+                'model_id': 'model_x',
+            },
+            {
+                'trigger': 'test_trigger_2',
+                'notification_id': 'test_notification_id_2',
+                'model_id': 'model_y',
+            },
+            {
+                'trigger': 'test_trigger_3',
+                'notification_id': 'test_notification_id_3',
+                'model_id': 'model_z',
+            },
+        ],
+        'sending_configs': [
+            {
+                'group_name': 'test_group_1',
+                'contents': ['core_alerts'],
+                'model_ids': ['model_x', 'model_y'],
+                'ignore_models_id': ['model_x'],
+            },
+        ],
+        'notification_ttl': 300,
+        'mail_type': 'test_mail_type',
+    }
+
+    response = slot_manager.filter_notification_alerts(input_data)
+
+    # model_x: allowed but ignored -> excluded. model_y: allowed and not
+    # ignored -> included. model_z: not in the allow-list -> excluded.
+    assert response == {
+        'test_group_1': {
+            'group_name': 'test_group_1',
+            'contents': ['core_alerts'],
+            'model_ids': ['model_x', 'model_y'],
+            'ignore_models_id': ['model_x'],
+            'notifications': [
+                {
+                    'trigger': 'test_trigger_2',
+                    'notification_id': 'test_notification_id_2',
+                    'model_id': 'model_y',
+                },
+            ],
+        },
+    }
+
+
+def test_filter_notification_alerts_model_ids_explicit_none_does_not_raise(slot_manager):
+    # Same defensive concern as ignore_models_id: receiver_groups has no schema
+    # validation, so model_ids can be explicitly null instead of omitted.
+    slot_manager.redis_repository.get = MagicMock(side_effect=[None])
+
+    input_data = {
+        **metadata,
+        'notification_package': [
+            {
+                'trigger': 'test_trigger_1',
+                'notification_id': 'test_notification_id_1',
+                'model_id': 'model_x',
+            },
+        ],
+        'sending_configs': [
+            {'group_name': 'test_group_1', 'contents': ['core_alerts'], 'model_ids': None},
+        ],
+        'notification_ttl': 300,
+        'mail_type': 'test_mail_type',
+    }
+
+    response = slot_manager.filter_notification_alerts(input_data)
+
+    assert response['test_group_1']['notifications'] == input_data['notification_package']
+
+
 def test_filter_notification_alerts_no_levels_or_ignore_models_id_passes_everything(
     slot_manager,
 ):

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': 0,
     }
 
 

tests/orchestrator/utils/test_orchestrator_functions.py

@@ -1,10 +1,13 @@
 from unittest.mock import call, patch
 
+from pytest import mark
+
 from orchestrator.utils.orchestrator_functions import (
     base_scouter,
     build_tag_config,
     common_config,
     drift,
+    frequency_alert_warning,
     gather_read_tags,
     minimal_retrain,
     overlap_filter_config,
@@ -110,6 +113,7 @@ def test_simple_metrics():
         'target_table_name': 'simple_metrics',
         'interval_minutes': 120,
         'metrics': ['rmse', 'mse'],
+        'frequency_alert': None,
         'execution_timeout_seconds': 300,
         'task_timeout_seconds': 300,
         'runtime': 'legacy',
@@ -117,6 +121,61 @@ def test_simple_metrics():
     assert result == expected
 
 
+def _simple_metrics_config(**fields):
+    return {
+        'workflow_type': 'simple_metrics',
+        'schedule_name': 'm1-simple_metrics-60m',
+        'model_id': 'm1',
+        'model': {'name': 'm1'},
+        'frequency': '1h',
+        **fields,
+    }
+
+
+def test_simple_metrics_escalated_uses_frequency_alert():
+    config = _simple_metrics_config(
+        frequency_alert='5m', adaptive_state={'escalated': True, 'healthy_streak': 0}
+    )
+    result = simple_metrics(config)
+    assert result['frequency'] == '5m'
+    assert result['frequency_alert'] == '5m'
+
+
+def test_simple_metrics_not_escalated_keeps_frequency():
+    config = _simple_metrics_config(
+        frequency_alert='5m', adaptive_state={'escalated': False, 'healthy_streak': 2}
+    )
+    assert simple_metrics(config)['frequency'] == '1h'
+
+
+def test_simple_metrics_escalated_without_frequency_alert_keeps_frequency():
+    # Removing frequency_alert turns adaptive mode off even if a stale state says escalated.
+    config = _simple_metrics_config(adaptive_state={'escalated': True, 'healthy_streak': 0})
+    assert simple_metrics(config)['frequency'] == '1h'
+
+
+def test_simple_metrics_frequency_alert_without_state_keeps_frequency():
+    assert simple_metrics(_simple_metrics_config(frequency_alert='5m'))['frequency'] == '1h'
+
+
+def test_simple_metrics_escalated_unparseable_frequency_alert_keeps_frequency():
+    # An unparseable frequency would make the Temporal schedule create/update fail.
+    config = _simple_metrics_config(frequency_alert='5min', adaptive_state={'escalated': True})
+    assert simple_metrics(config)['frequency'] == '1h'
+
+
+def test_simple_metrics_escalated_below_floor_frequency_alert_is_applied():
+    # Decision 7: below the floor only warns, the value is kept.
+    config = _simple_metrics_config(frequency_alert='30s', adaptive_state={'escalated': True})
+    assert simple_metrics(config)['frequency'] == '30s'
+
+
+@mark.parametrize('adaptive_state', ['yes', {'escalated': 'false'}])
+def test_simple_metrics_malformed_adaptive_state_keeps_frequency(adaptive_state):
+    config = _simple_metrics_config(frequency_alert='5m', adaptive_state=adaptive_state)
+    assert simple_metrics(config)['frequency'] == '1h'
+
+
 def test_minimal_retrain():
     config = {
         'workflow_type': 'minimal_retrain',
@@ -762,3 +821,19 @@ def test_pi_web_api_scouter_with_no_timeout():
         'runtime': 'legacy',
     }
     assert result == expected
+
+
+@mark.parametrize(
+    'frequency_alert, warns',
+    [
+        (None, False),
+        ('5m', False),
+        ('1m', False),  # the floor itself is allowed
+        ('60s', False),
+        ('30s', True),
+        ('5min', True),  # unparseable
+        (5, True),  # not a string: must warn, never raise
+    ],
+)
+def test_frequency_alert_warning(frequency_alert, warns):
+    assert (frequency_alert_warning(frequency_alert) is not None) is warns

tests/orchestrator/workflows/subworkflows/test_load_notification_package.py

@@ -237,6 +237,65 @@ def test_compute_query_time_filter_model_ignored_by_only_one_group_not_excluded(
     assert 'model_id' not in result
 
 
+def test_compute_query_time_filter_model_ids_union_when_all_groups_restrict(
+    load_notification_package,
+):
+    # SIENTIAPDE-2149: every active group sets a model_ids allow-list -> the
+    # union applies at query time, mirroring how `levels` already behaves.
+    sending_configs = [
+        {'group_name': 'group_a', 'model_ids': ['model_a']},
+        {'group_name': 'group_b', 'model_ids': ['model_b']},
+    ]
+    default_filter = {}
+
+    result = load_notification_package._compute_query_time_filter(sending_configs, default_filter)
+
+    assert result == {'model_id': {'$in': ['model_a', 'model_b']}}
+
+
+def test_compute_query_time_filter_model_ids_no_restriction_when_any_group_missing(
+    load_notification_package,
+):
+    # One active group has no model_ids restriction -> it wants every model,
+    # so the query can't restrict to an allow-list at all.
+    sending_configs = [
+        {'group_name': 'group_a', 'model_ids': ['model_a']},
+        {'group_name': 'group_b'},
+    ]
+    default_filter = {}
+
+    result = load_notification_package._compute_query_time_filter(sending_configs, default_filter)
+
+    assert 'model_id' not in result
+
+
+def test_compute_query_time_filter_model_ids_union_and_ignore_models_id_intersection_together(
+    load_notification_package,
+):
+    # SIENTIAPDE-2149: the allow-list union ($in) and the ignore-list
+    # intersection ($nin) are independent axes and can both narrow the same
+    # model_id field at once.
+    sending_configs = [
+        {
+            'group_name': 'group_a',
+            'model_ids': ['model_a', 'model_b'],
+            'ignore_models_id': ['model_a'],
+        },
+        {
+            'group_name': 'group_b',
+            'model_ids': ['model_b', 'model_c'],
+            'ignore_models_id': ['model_a'],
+        },
+    ]
+    default_filter = {}
+
+    result = load_notification_package._compute_query_time_filter(sending_configs, default_filter)
+
+    assert result == {
+        'model_id': {'$in': ['model_a', 'model_b', 'model_c'], '$nin': ['model_a']},
+    }
+
+
 def test_compute_query_time_filter_union_levels_and_intersection_models(
     load_notification_package,
 ):