chore: bump version to 0.6.5

SIENTIAPDE-2012: per-group levels/ignore_models_id filtering

.github/workflows/deploy.yml

@@ -0,0 +1,29 @@
+name: Deploy DataOps Worker
+
+on:
+  pull_request:
+    types: [closed]
+    branches: [main, 'release/**', 'feature/**']
+  workflow_dispatch:
+
+permissions:
+  contents: write
+  id-token: write
+  pull-requests: read
+
+jobs:
+  deploy:
+    name: Deploy
+    if: github.event_name == 'workflow_dispatch' || github.event.pull_request.merged == true
+    uses: Aignosi/github_workflow_templates/.github/workflows/reusable-deploy-dataops.yml@main
+    with:
+      branch_name: ${{ github.event.pull_request.head.ref }}
+      project_name: 'orchestrator'
+      chart_name: 'orchestrator-worker'
+      env_list_path: '[".env"]'
+      update_version_in: '["pyproject"]'
+      create_tarball: ${{ github.event_name == 'pull_request' }}
+      mirror: true
+      bump_chart: true
+      deploy: true
+    secrets: inherit

.github/workflows/release.yml

@@ -1,16 +0,0 @@
-name: Create Release on Merge to Main
-
-on:
-  pull_request:
-    types: [closed]
-    branches:
-      - main
-
-jobs:
-  release:
-    if: github.event.pull_request.merged == true
-    uses: Aignosi/github_workflow_templates/.github/workflows/dataops-module-release.yml@main
-    permissions: write-all
-    with:
-      project_name: 'orchestrator'
-    secrets: inherit

.github/workflows/require-chart-ref-label.yml

@@ -0,0 +1,16 @@
+name: Require chart-ref label
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened, labeled, unlabeled]
+    branches:
+      - main
+      - 'release/**'
+      - 'feature/**'
+
+permissions:
+  pull-requests: write
+
+jobs:
+  require-chart-ref-label:
+    uses: Aignosi/github_workflow_templates/.github/workflows/require-chart-ref-label.yml@main

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` vs all levels).
+- **`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/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/helpers.py

@@ -7,6 +7,7 @@ import json
 import re
 import uuid
 from datetime import UTC, datetime, timedelta
+from email.message import EmailMessage, Message
 from pathlib import Path
 from typing import Any
 
@@ -29,7 +30,6 @@ MONGO_COLLECTIONS = (
 )
 
 DATETIME_FORMAT_MS_WITH_TZ = '%Y-%m-%d %H:%M:%S.%f%z'
-DATETIME_FORMAT_WITH_TZ = '%Y-%m-%d %H:%M:%S%z'
 
 _TIMESTAMP_MARKER_PATTERN = re.compile(r'^@now(?:([+-])(\d+)([smhd]))?$')
 _TIMESTAMP_UNIT_TO_KWARG = {'s': 'seconds', 'm': 'minutes', 'h': 'hours', 'd': 'days'}
@@ -222,17 +222,19 @@ def seed_last_timestamp(redis_client: Redis, mail_type: str, value: str) -> None
     Set notification_last_timestamp for a mail type, JSON-encoded.
 
     Values must be JSON-encoded so ``redis_repository.get`` (which calls
-    ``json.loads`` on the raw payload) can deserialize them. The value
-    must follow the exact format ``sientia_do.notifications.models.Notification``
-    writes into ``notification_queue.timestamp``: ``DATETIME_FORMAT_WITH_TZ``
-    (e.g. ``"2026-05-22 16:47:02+0000"``) — no microseconds and no colon in
-    the timezone offset.
+    ``json.loads`` on the raw payload) can deserialize them. The value must
+    follow the exact format ``put_last_data_timestamp``
+    (``orchestrator/activities/slot_manager.py``) writes into Redis today:
+    ``DATETIME_FORMAT_MS_WITH_TZ`` (e.g. ``"2026-05-22 16:47:02.000000+0000"``)
+    — microseconds included, no colon in the timezone offset. ``load_latest_data``
+    parses this same format back into a ``datetime`` for the ``$gt`` comparison,
+    so the two sides must agree exactly.
 
     Args:
         redis_client: Redis client connected to the test instance.
         mail_type: Mail type identifier (e.g. ``"Alerts"``, ``"Reports"``).
-        value: Timestamp string in ``DATETIME_FORMAT_WITH_TZ``
-            (e.g. ``"2024-06-01 10:30:00+0000"``).
+        value: Timestamp string in ``DATETIME_FORMAT_MS_WITH_TZ``
+            (e.g. ``"2024-06-01 10:30:00.000000+0000"``).
     """
     redis_client.set(f'notification_last_timestamp:{mail_type}', json.dumps(value))
 
@@ -295,34 +297,35 @@ def default_notification(
     *,
     notification_id: str,
     level: str = 'ERROR',
-    timestamp: str | None = None,
+    timestamp: datetime | None = None,
     model_name: str = 'model-a',
     model_id: str = '1',
 ) -> dict[str, Any]:
     """
     Build a minimal notification_queue document mirroring production layout.
 
-    The ``timestamp`` field is stored as a string in ``DATETIME_FORMAT_WITH_TZ``
-    because that is exactly what ``sientia_do.notifications.models.Notification``
-    writes into ``notification_queue`` in production (``now().strftime(
-    DATETIME_FORMAT_WITH_TZ)``). Tests intentionally use this same format so we
+    The ``timestamp`` field is stored as a native, timezone-aware ``datetime``
+    (BSON Date) because that is exactly what
+    ``sientia_do.notifications.models.Notification`` writes into
+    ``notification_queue`` in production since SIENTIAPDE-2007
+    (``self.timestamp = now()`` — a real ``datetime``, no more string
+    formatting). Tests intentionally use this same representation so we
     surface, rather than hide, real production behavior in downstream
-    activities.
+    activities (e.g. ``put_last_data_timestamp``'s ``DataFrame(...)['timestamp']
+    .max().strftime(...)``, which requires a real ``datetime``, not a string).
 
     Args:
         notification_id: Unique identifier for the notification.
         level: Notification level (e.g. ``"ERROR"``, ``"WARNING"``).
-        timestamp: Optional production-format timestamp string. ``None`` falls
-            back to a fixed sample value.
+        timestamp: Optional timezone-aware datetime. ``None`` falls back to a
+            fixed sample value.
         model_name: Model name attached to the notification.
         model_id: Model id attached to the notification.
 
     Return:
         dict[str, Any]: A notification document ready for insertion.
     """
-    ts = timestamp if timestamp is not None else datetime(
-        2024, 6, 1, 12, 0, 0, tzinfo=UTC
-    ).strftime(DATETIME_FORMAT_WITH_TZ)
+    ts = timestamp if timestamp is not None else datetime(2024, 6, 1, 12, 0, 0, tzinfo=UTC)
     return {
         'notification_id': notification_id,
         'level': level,
@@ -343,7 +346,7 @@ def default_receiver_group(
     members: list[str] | None = None,
     levels: list[str] | None = None,
     contents: list[str] | None = None,
-    ignore_models: list[str] | None = None,
+    ignore_models_id: list[str] | None = None,
 ) -> dict[str, Any]:
     """Minimal active receiver_groups document."""
     return {
@@ -352,5 +355,42 @@ def default_receiver_group(
         'members': members or ['admin@example.com'],
         'levels': levels or ['ERROR', 'WARNING', 'INFO'],
         'contents': contents or ['core_alerts', 'persistent_alerts', 'reports'],
-        'ignore_models': ignore_models or [],
+        'ignore_models_id': ignore_models_id or [],
     }
+
+
+def extract_html_body(message: EmailMessage | Message) -> str:
+    """
+    Return the text/html portion of an email message as a decoded string.
+
+    Walks every part looking for the first text/html payload, decoding it
+    according to the part's transfer encoding and charset. Falls back to
+    the message's own ``get_content``/raw payload when no HTML part is
+    present so callers can still inspect plain-text reports. Shared between
+    the Alerts and Reports e2e suites, both of which send mail through the
+    same ``build_email_html``/``send_email`` activities.
+
+    Args:
+        message: Captured email message returned by the test SMTP server.
+
+    Return:
+        str: HTML body content, or an empty string when nothing decodable
+            is found.
+    """
+    if message.is_multipart():
+        for part in message.walk():
+            if part.get_content_type() != 'text/html':
+                continue
+            payload = part.get_payload(decode=True)
+            if payload is None:
+                continue
+            charset = part.get_content_charset() or 'utf-8'
+            return payload.decode(charset, errors='replace')
+    payload = message.get_payload(decode=True)
+    if payload is not None:
+        charset = message.get_content_charset() or 'utf-8'
+        return payload.decode(charset, errors='replace')
+    try:
+        return message.get_content()
+    except (AttributeError, KeyError):
+        return str(message.get_payload() or '')

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,13 +59,14 @@ 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` honored.
+Receiver group `levels` / `ignore_models_id` honored, both halves now covered:
+the query-time union/intersection across active groups (SIENTIAPDE-2011, see
+`LoadNotificationPackage` below) and the per-group precise slice
+(SIENTIAPDE-2012) — a group only receives notifications matching its own
+`levels`/`ignore_models_id`, even when the query-time filter widened the fetch
+to satisfy another active group's broader configuration.
 
 ### A.3.1 Empty queue
 
@@ -103,6 +98,7 @@ 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.
 - Empty Mongo → no Redis timestamp write.
 
 ### ProcessNotifications — `e2e/test_subworkflow_process_notifications.py`

e2e/test_alerts_main_workflow.py

@@ -12,6 +12,7 @@ from e2e.helpers import (
     count_log_report_rows,
     default_notification,
     default_receiver_group,
+    extract_html_body,
     fetch_log_report,
     load_scenario_input,
     make_workflow_id,
@@ -92,6 +93,95 @@ async def test_scenario_a_1_2_duplicate_suppressed(
     assert count_log_report_rows(postgres_engine, 'Alerts') == 0
 
 
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_scenario_a_2_1_per_group_levels_and_ignore_models_id(
+    temporal_env: WorkflowEnvironment,
+    orchestrator_worker,
+    stub_workers,
+    mongo_uri: str,
+    smtp_server,
+):
+    """
+    A.2.1 (SIENTIAPDE-2012): each receiver group gets only its own precise
+    slice of notifications, on top of SIENTIAPDE-2011's query-time fetch.
+
+    Two active groups widen the query-time union beyond Alerts' ERROR-only
+    default: group-errors (levels=['ERROR']) and group-warnings
+    (levels=['WARNING'], ignore_models_id=['model-x']). Because the query-time
+    filter only unions/intersects across ALL active groups, both ERROR and
+    WARNING notifications get fetched for every group, and model-x is not
+    excluded at the query level either (group-errors has no ignore_models_id
+    restriction, so the intersection is empty). The per-group filter added in
+    this subtask is what actually narrows each group back down to what it
+    configured.
+    """
+    seed_receiver_groups(
+        mongo_uri,
+        E2E_DATABASE,
+        [
+            default_receiver_group(
+                group_name='group-errors', contents=['core_alerts'], levels=['ERROR']
+            ),
+            default_receiver_group(
+                group_name='group-warnings',
+                contents=['core_alerts'],
+                levels=['WARNING'],
+                ignore_models_id=['model-x'],
+            ),
+        ],
+    )
+    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='warn-a',
+                level='WARNING',
+                model_name='model-a',
+                model_id='model-a',
+                timestamp=datetime(2024, 6, 1, 12, 1, 0, tzinfo=UTC),
+            ),
+            default_notification(
+                notification_id='warn-x',
+                level='WARNING',
+                model_name='model-x',
+                model_id='model-x',
+                timestamp=datetime(2024, 6, 1, 12, 2, 0, tzinfo=UTC),
+            ),
+        ],
+    )
+
+    await start_and_await_workflow(
+        temporal_env.client,
+        Alerts.run,
+        load_scenario_input('alerts_happy_path.json'),
+        make_workflow_id('alerts-2012'),
+    )
+
+    # One email per group -- each group ended up with exactly one notification.
+    assert len(smtp_server.messages) == 2
+    bodies = [extract_html_body(message) for message in smtp_server.messages]
+    combined = '\n'.join(bodies)
+
+    # group-errors (levels=['ERROR']): received err-a, never warn-a/warn-x
+    # despite both being fetched by the widened query-time union.
+    errors_body = next(b for b in bodies if 'Errors detected:' in b)
+    assert 'Warnings detected:' not in errors_body
+
+    # group-warnings (levels=['WARNING'], ignore_models_id=['model-x']):
+    # received warn-a only -- warn-x was fetched (query-time intersection is
+    # empty, since group-errors doesn't ignore any model) but dropped here by
+    # the per-group ignore_models_id check.
+    warnings_body = next(b for b in bodies if 'Warnings detected:' in b)
+    assert 'Errors detected:' not in warnings_body
+    assert 'model-a' in combined
+    assert 'model-x' not in combined
+
+
 @pytest.mark.e2e
 @pytest.mark.asyncio
 async def test_scenario_a_3_1_empty_queue(

e2e/test_reports_main_workflow.py

@@ -1,6 +1,6 @@
 """E2E tests for the Reports main workflow."""
 
-from email.message import EmailMessage, Message
+from datetime import UTC, datetime
 
 import pytest
 from temporalio.testing import WorkflowEnvironment
@@ -10,6 +10,7 @@ from e2e.helpers import (
     count_log_report_rows,
     default_notification,
     default_receiver_group,
+    extract_html_body,
     load_scenario_input,
     make_workflow_id,
     seed_notifications,
@@ -19,41 +20,6 @@ from e2e.helpers import (
 from orchestrator.workflows.reports import Reports
 
 
-def _extract_html_body(message: EmailMessage | Message) -> str:
-    """
-    Return the text/html portion of an email message as a decoded string.
-
-    Walks every part looking for the first text/html payload, decoding it
-    according to the part's transfer encoding and charset. Falls back to
-    the message's own ``get_content``/raw payload when no HTML part is
-    present so callers can still inspect plain-text reports.
-
-    Args:
-        message: Captured email message returned by the test SMTP server.
-
-    Return:
-        str: HTML body content, or an empty string when nothing decodable
-            is found.
-    """
-    if message.is_multipart():
-        for part in message.walk():
-            if part.get_content_type() != 'text/html':
-                continue
-            payload = part.get_payload(decode=True)
-            if payload is None:
-                continue
-            charset = part.get_content_charset() or 'utf-8'
-            return payload.decode(charset, errors='replace')
-    payload = message.get_payload(decode=True)
-    if payload is not None:
-        charset = message.get_content_charset() or 'utf-8'
-        return payload.decode(charset, errors='replace')
-    try:
-        return message.get_content()
-    except (AttributeError, KeyError):
-        return str(message.get_payload() or '')
-
-
 @pytest.mark.e2e
 @pytest.mark.asyncio
 async def test_scenario_r_1_1_happy_path(
@@ -79,13 +45,13 @@ async def test_scenario_r_1_1_happy_path(
                 notification_id='r-warn',
                 level='WARNING',
                 model_name='m2',
-                timestamp='2024-06-01 12:01:00+0000',
+                timestamp=datetime(2024, 6, 1, 12, 1, 0, tzinfo=UTC),
             ),
             default_notification(
                 notification_id='r-info',
                 level='INFO',
                 model_name='m3',
-                timestamp='2024-06-01 12:02:00+0000',
+                timestamp=datetime(2024, 6, 1, 12, 2, 0, tzinfo=UTC),
             ),
         ],
     )
@@ -98,7 +64,7 @@ async def test_scenario_r_1_1_happy_path(
     )
 
     assert len(smtp_server.messages) == 1
-    body = _extract_html_body(smtp_server.messages[0])
+    body = extract_html_body(smtp_server.messages[0])
     assert 'Errors detected:' in body
     assert 'Warnings detected:' in body
     assert 'Infos detected:' in body
@@ -137,7 +103,7 @@ async def test_scenario_r_1_2_error_section_only(
     )
 
     assert smtp_server.messages, 'Expected at least one report email'
-    body = _extract_html_body(smtp_server.messages[0])
+    body = extract_html_body(smtp_server.messages[0])
     assert 'Errors detected:' in body
     assert 'Warnings detected:' not in body
     assert 'Infos detected:' not in body

e2e/test_subworkflow_load_notification_package.py

@@ -1,5 +1,7 @@
 """E2E tests for LoadNotificationPackage subworkflow."""
 
+from datetime import UTC, datetime
+
 import pytest
 from redis import Redis
 from temporalio.testing import WorkflowEnvironment
@@ -42,10 +44,12 @@ async def test_load_package_without_prior_timestamp(
     """
     No Redis timestamp → returns notifications and stores max timestamp.
 
-    Notifications are seeded with the exact production format produced by
-    ``sientia_do.notifications.models.Notification`` (string in
-    ``DATETIME_FORMAT_WITH_TZ``, e.g. ``"2024-06-01 11:00:00+0000"``). The
-    cached "last timestamp" must mirror that representation.
+    Notifications are seeded with the exact production representation written
+    by ``sientia_do.notifications.models.Notification`` since SIENTIAPDE-2007:
+    a native ``datetime`` (BSON Date), not a formatted string. The cached
+    "last timestamp" written back to Redis by ``put_last_data_timestamp`` is a
+    string formatted per ``DATETIME_FORMAT_MS_WITH_TZ`` (microseconds
+    included) — that is the format this test asserts against.
     """
     seed_receiver_groups(mongo_uri, E2E_DATABASE, [default_receiver_group()])
     seed_notifications(
@@ -54,11 +58,11 @@ async def test_load_package_without_prior_timestamp(
         [
             default_notification(
                 notification_id='n1',
-                timestamp='2024-06-01 10:00:00+0000',
+                timestamp=datetime(2024, 6, 1, 10, 0, 0, tzinfo=UTC),
             ),
             default_notification(
                 notification_id='n2',
-                timestamp='2024-06-01 11:00:00+0000',
+                timestamp=datetime(2024, 6, 1, 11, 0, 0, tzinfo=UTC),
             ),
         ],
     )
@@ -75,7 +79,7 @@ async def test_load_package_without_prior_timestamp(
 
     assert len(result['notification_package']) == 2
     stored = redis_client.get('notification_last_timestamp:Alerts') or ''
-    assert stored.strip('"') == '2024-06-01 11:00:00+0000'
+    assert stored.strip('"') == '2024-06-01 11:00:00.000000+0000'
 
 
 @pytest.mark.e2e
@@ -90,27 +94,27 @@ async def test_load_package_with_prior_timestamp(
     """
     Existing timestamp → only newer notifications returned.
 
-    Both the seeded "last timestamp" (Redis) and the notification timestamps
-    (Mongo) follow the production format used by
-    ``sientia_do.notifications.models.Notification`` (string in
-    ``DATETIME_FORMAT_WITH_TZ``). ``load_latest_data`` will translate the
-    Redis value into a Python ``datetime`` and apply ``{$gt: <Date>}``
-    against the Mongo string timestamps; this exercise surfaces the real
-    BSON comparison semantics rather than a synthetic ideal.
+    The seeded "last timestamp" (Redis) follows the format
+    ``put_last_data_timestamp`` actually writes (``DATETIME_FORMAT_MS_WITH_TZ``,
+    microseconds included); the notification timestamps (Mongo) are native
+    ``datetime`` values, matching current production behavior since
+    SIENTIAPDE-2007. ``load_latest_data`` parses the Redis string back into a
+    ``datetime`` and applies ``{$gt: <Date>}`` against the Mongo BSON dates —
+    this exercises the real comparison semantics rather than a synthetic ideal.
     """
     seed_receiver_groups(mongo_uri, E2E_DATABASE, [default_receiver_group()])
-    seed_last_timestamp(redis_client, 'Alerts', '2024-06-01 10:30:00+0000')
+    seed_last_timestamp(redis_client, 'Alerts', '2024-06-01 10:30:00.000000+0000')
     seed_notifications(
         mongo_uri,
         E2E_DATABASE,
         [
             default_notification(
                 notification_id='old',
-                timestamp='2024-06-01 10:00:00+0000',
+                timestamp=datetime(2024, 6, 1, 10, 0, 0, tzinfo=UTC),
             ),
             default_notification(
                 notification_id='new',
-                timestamp='2024-06-01 11:00:00+0000',
+                timestamp=datetime(2024, 6, 1, 11, 0, 0, tzinfo=UTC),
             ),
         ],
     )
@@ -129,6 +133,78 @@ async def test_load_package_with_prior_timestamp(
     assert ids == {'new'}
 
 
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_load_package_query_time_filter_union_and_intersection(
+    temporal_env: WorkflowEnvironment,
+    orchestrator_worker,
+    stub_workers,
+    mongo_uri: str,
+    redis_client: Redis,
+):
+    """
+    SIENTIAPDE-2011: query-time filter unions active groups' levels and
+    intersects their ignore_models_id, instead of fetching everything.
+
+    Two active receiver groups narrow the fetch differently:
+      - group_a: levels=['ERROR'], ignore_models_id=['model-x']
+      - group_b: levels=['WARNING'], ignore_models_id=['model-x', 'model-y']
+
+    The Mongo query must return exactly what the UNION of levels allows
+    (ERROR or WARNING, never INFO) and must exclude only models in the
+    INTERSECTION of every active group's ignore list (model-x, ignored by
+    both). A model ignored by only one group (model-y) must still be
+    fetched, since the other group still wants it — per-group correctness
+    for that case is SIENTIAPDE-2012's job, not this query-time filter's.
+    """
+    seed_receiver_groups(
+        mongo_uri,
+        E2E_DATABASE,
+        [
+            default_receiver_group(
+                group_name='group_a', levels=['ERROR'], ignore_models_id=['model-x']
+            ),
+            default_receiver_group(
+                group_name='group_b',
+                levels=['WARNING'],
+                ignore_models_id=['model-x', 'model-y'],
+            ),
+        ],
+    )
+    seed_notifications(
+        mongo_uri,
+        E2E_DATABASE,
+        [
+            default_notification(
+                notification_id='error-included', level='ERROR', model_id='model-a'
+            ),
+            default_notification(
+                notification_id='warning-included', level='WARNING', model_id='model-a'
+            ),
+            default_notification(notification_id='info-excluded', level='INFO', model_id='model-a'),
+            default_notification(
+                notification_id='modelx-excluded', level='ERROR', model_id='model-x'
+            ),
+            default_notification(
+                notification_id='modely-included', level='WARNING', model_id='model-y'
+            ),
+        ],
+    )
+
+    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-filter'),
+    )
+
+    ids = {n['notification_id'] for n in result['notification_package']}
+    assert ids == {'error-included', 'warning-included', 'modely-included'}
+
+
 @pytest.mark.e2e
 @pytest.mark.asyncio
 async def test_load_package_empty_mongo_no_redis_write(

orchestrator/activities/formatters.py

@@ -7,11 +7,15 @@ with workflow.unsafe.imports_passed_through():
     from math import ceil
     from typing import Any, TypedDict, cast
 
-    from pandas import DataFrame
+    from pandas import DataFrame, to_datetime
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.notifications.models import NotificationLevel
     from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring
-    from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now
+    from sientia_do.temporal.constants import (
+        DATETIME_FORMAT_MS_WITH_TZ,
+        DATETIME_FORMAT_WITH_TZ,
+        now,
+    )
 
     from orchestrator.utils.orchestrator_functions import (
         build_tag_config,
@@ -755,9 +759,24 @@ class Formatters(SientiaMonitoring):
                 key = f'{notification_id}:{trigger}'
 
                 if key not in data:
+                    # ``notification['timestamp']`` started out as a native
+                    # ``datetime``/BSON Date in Mongo, but by the time it gets here
+                    # it has crossed at least one Temporal activity boundary:
+                    # Temporal's default data converter serializes ``datetime``
+                    # values to plain ISO-8601 strings and does not restore them on
+                    # decode, and pymongo returns naive datetimes for BSON Date
+                    # fields (no tzinfo), so the ISO string may carry no UTC offset
+                    # at all. Normalizing with ``utc=True`` and re-formatting to
+                    # ``DATETIME_FORMAT_WITH_TZ`` (same pattern as
+                    # ``SlotManager.put_last_data_timestamp``) keeps the value a
+                    # fixed-format string that ``export_data_to_postgres``'s
+                    # ``timestamp_conversion`` can parse after the next boundary
+                    # crossing, regardless of what shape it arrived in here.
                     data[key] = {
                         'status': group_config['status'],
-                        'timestamp': notification['timestamp'],
+                        'timestamp': to_datetime(notification['timestamp'], utc=True).strftime(
+                            DATETIME_FORMAT_WITH_TZ
+                        ),
                         'groups': [group_name],
                         'message': notification['message'],
                         'level': notification['level'],
@@ -805,6 +824,8 @@ class Formatters(SientiaMonitoring):
                     - group_name (str): Name of the receiver group
                     - contents (list): Content types to include (must contain "reports")
                     - ignore (list, optional): Notification IDs to exclude
+                    - levels (list, optional): Notification levels to include
+                    - ignore_models_id (list, optional): Model IDs to exclude
 
         Returns:
             dict[str, Any]: Filtered receiver groups with their notifications, keyed by group_name.
@@ -828,6 +849,8 @@ class Formatters(SientiaMonitoring):
             already_added_keys = []
 
             ignore_list = receiver_group.get('ignore', [])
+            levels_list = receiver_group.get('levels')
+            ignore_models_id_list = receiver_group.get('ignore_models_id') or []
 
             for notification in notification_package:
                 alert_type = 'reports'
@@ -840,6 +863,11 @@ class Formatters(SientiaMonitoring):
                     alert_type in receiver_group['contents']
                     and notification_id not in ignore_list
                     and key not in already_added_keys
+                    and (not levels_list or notification.get('level') in levels_list)
+                    and (
+                        not ignore_models_id_list
+                        or notification.get('model_id') not in ignore_models_id_list
+                    )
                 ):
                     receiver_groups[group_name]['notifications'].append(notification)
                     already_added_keys.append(key)

orchestrator/activities/mongo_db.py

@@ -453,7 +453,7 @@ class MongoDB(SientiaMonitoring):
                 - metadata (dict[str, Any]): Workflow execution metadata
                 - collection_name (str): Name of the MongoDB collection
                 - last_data_timestamp (str | None): Last processed timestamp
-                  (ISO-format string) for filtering
+                  (formatted per ``DATETIME_FORMAT_MS_WITH_TZ``) for filtering
                 - base_data_filter (dict[str, Any]): Base query filter conditions
 
         Returns:
@@ -475,12 +475,15 @@ class MongoDB(SientiaMonitoring):
             else:
                 # ``notification_queue.timestamp`` is stored as a native BSON
                 # ``Date`` (``Notification.timestamp`` is a ``datetime``).
-                # ``last_data_timestamp`` arrives here as an ISO-format string
-                # (round-tripped through Redis), so it must be parsed back to
-                # ``datetime`` for the ``$gt`` comparison to be type-correct.
+                # ``last_data_timestamp`` arrives here as a string formatted per
+                # ``DATETIME_FORMAT_MS_WITH_TZ`` (round-tripped through Redis),
+                # so it must be parsed back to ``datetime`` for the ``$gt``
+                # comparison to be type-correct.
                 data_filter = {
                     **base_data_filter,
-                    'timestamp': {'$gt': datetime.fromisoformat(last_data_timestamp)},
+                    'timestamp': {
+                        '$gt': datetime.strptime(last_data_timestamp, DATETIME_FORMAT_MS_WITH_TZ)
+                    },
                 }
 
             self.debug(f'Data filter: {data_filter}', metadata=metadata)

orchestrator/activities/slot_manager.py

@@ -7,7 +7,7 @@ with workflow.unsafe.imports_passed_through():
     from logging import Logger
     from typing import Any
 
-    from pandas import DataFrame
+    from pandas import DataFrame, to_datetime
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.notifications.models import NotificationLevel
     from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring
@@ -331,17 +331,33 @@ class SlotManager(SientiaMonitoring):
             self.warning('No data to insert', metadata=metadata)
             return None
 
-        last_data_timestamp = data['timestamp'].max()
+        # ``timestamp`` started out as a native ``datetime``/BSON Date in Mongo,
+        # but by the time it gets here it has crossed a Temporal activity
+        # boundary (this activity's ``input_data`` is whatever
+        # ``load_latest_data`` returned): Temporal's default data converter
+        # serializes ``datetime`` values to plain ISO-8601 strings and does not
+        # restore them on decode, so ``data['timestamp']`` is a ``str`` in a
+        # real workflow run even though unit tests mocking the activity in
+        # isolation never see that conversion. pymongo also returns naive
+        # datetimes for BSON Date fields (no tzinfo), so the ISO string may
+        # carry no UTC offset at all. ``utc=True`` normalizes every case
+        # (native datetime/Timestamp, offset-aware or naive ISO string) into a
+        # UTC-aware ``Timestamp`` — naive values are BSON Dates, always UTC.
+        last_data_timestamp = to_datetime(data['timestamp'], utc=True).max()
 
         self.debug(f'Last collected timestamp to insert: {last_data_timestamp}', metadata=metadata)
 
-        # ``timestamp`` is a native ``datetime``/``Timestamp`` (BSON Date from
-        # Mongo), which the Redis repository's plain ``json.dumps`` cannot
-        # serialize. Store it as an ISO-format string instead.
-        last_data_timestamp = last_data_timestamp.isoformat()
+        # The Redis repository's plain ``json.dumps`` cannot serialize a
+        # ``Timestamp``/``datetime`` directly, so store it as a string
+        # formatted per ``DATETIME_FORMAT_MS_WITH_TZ`` instead. Kept as a
+        # separate variable (rather than reassigning ``last_data_timestamp``)
+        # since it's a different type — mypy infers a local's type from its
+        # first assignment and flags a same-name reassignment to an
+        # incompatible type as an error.
+        last_data_timestamp_str = last_data_timestamp.strftime(DATETIME_FORMAT_MS_WITH_TZ)
 
         try:
-            self.redis_repository.set(key, last_data_timestamp, ttl=60 * 60 * 5)
+            self.redis_repository.set(key, last_data_timestamp_str, ttl=60 * 60 * 5)
         except Exception as e:
             self.send_notification(
                 metadata=metadata,
@@ -353,7 +369,7 @@ class SlotManager(SientiaMonitoring):
             )
             raise e
 
-        return last_data_timestamp
+        return last_data_timestamp_str
 
     @activity.defn(name='filter_notification_alerts')
     def filter_notification_alerts(self, input_data: dict[str, Any]) -> dict[str, Any]:
@@ -380,6 +396,8 @@ class SlotManager(SientiaMonitoring):
                     - group_name (str): Name of the receiver group
                     - contents (list): Alert types to include (core_alerts, persistent_alerts)
                     - ignore (list, optional): Notification IDs to exclude
+                    - levels (list, optional): Notification levels to include
+                    - ignore_models_id (list, optional): Model IDs to exclude
                 - notification_ttl (int): Seconds before considering notification persistent
 
         Returns:
@@ -399,48 +417,118 @@ class SlotManager(SientiaMonitoring):
 
         for receiver_group in sending_configs:
             group_name = receiver_group['group_name']
-            receiver_groups[group_name] = {**receiver_group, 'notifications': []}
-            receiver_groups[group_name]['notifications'] = []
-
-            already_added_keys = []
-
-            ignore_list = receiver_group.get('ignore', [])
-
-            for notification in notification_package:
-                alert_type = 'do_nothing'
-                notification_id = notification['notification_id']
-                # Check if notification was recently sent
-                key = f'{notification["trigger"]}:{notification_id}'
-
-                last_sent = self.redis_repository.get(key)
-
-                if last_sent is None:
-                    alert_type = 'core_alerts'
-
-                else:
-                    last_sent = datetime.strptime(last_sent, DATETIME_FORMAT_MS_WITH_TZ)
-
-                    # Check if "notification_ttl" seconds has passed since last sent
-                    if (now() - last_sent) > timedelta(seconds=notification_ttl):
-                        alert_type = 'persistent_alerts'
-
-                # Check if this group must be notified
-                if (
-                    alert_type in receiver_group['contents']
-                    and notification_id not in ignore_list
-                    and key not in already_added_keys
-                ):
-                    receiver_groups[group_name]['notifications'].append(notification)
-                    already_added_keys.append(key)
+            notifications = self._filter_group_notifications(
+                receiver_group, notification_package, notification_ttl
+            )
+            receiver_groups[group_name] = {**receiver_group, 'notifications': notifications}
 
         # Remove groups with no notifications
-        receiver_groups = {
+        return {
             group_name: group
             for group_name, group in receiver_groups.items()
             if group['notifications']
         }
 
-        return receiver_groups
+    def _filter_group_notifications(
+        self,
+        receiver_group: dict[str, Any],
+        notification_package: list[dict[str, Any]],
+        notification_ttl: int,
+    ) -> list[dict[str, Any]]:
+        """
+        Select the notifications this receiver group should get.
+
+        Args:
+            receiver_group (dict[str, Any]): The receiver group configuration.
+            notification_package (list[dict[str, Any]]): Candidate notifications.
+            notification_ttl (int): Seconds before considering a notification persistent.
+
+        Returns:
+            list[dict[str, Any]]: Notifications matching this group's filters.
+        """
+        already_added_keys: list[str] = []
+        ignore_list = receiver_group.get('ignore', [])
+        levels_list = receiver_group.get('levels')
+        ignore_models_id_list = receiver_group.get('ignore_models_id') or []
+
+        notifications = []
+        for notification in notification_package:
+            notification_id = notification['notification_id']
+            key = f'{notification["trigger"]}:{notification_id}'
+            alert_type = self._determine_alert_type(key, notification_ttl)
+
+            if (
+                alert_type in receiver_group['contents']
+                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
+                )
+            ):
+                notifications.append(notification)
+                already_added_keys.append(key)
+
+        return notifications
+
+    def _determine_alert_type(self, key: str, notification_ttl: int) -> str:
+        """
+        Classify a notification as new, persistent, or not-yet-due.
+
+        Args:
+            key (str): The Redis cache key for this notification.
+            notification_ttl (int): Seconds before considering a notification persistent.
+
+        Returns:
+            str: 'core_alerts', 'persistent_alerts', or 'do_nothing'.
+        """
+        last_sent = self.redis_repository.get(key)
+        if last_sent is None:
+            return 'core_alerts'
+
+        last_sent = datetime.strptime(last_sent, DATETIME_FORMAT_MS_WITH_TZ)
+        if (now() - last_sent) > timedelta(seconds=notification_ttl):
+            return 'persistent_alerts'
+
+        return 'do_nothing'
+
+    @staticmethod
+    def _check_levels_and_ignore_models_id(notification, levels_list, ignore_models_id_list):
+        """
+        Check if the notification's level is in the levels_list and its model_id is not in the ignore_models_id_list.
+
+        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.
+
+        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 = (
+            not ignore_models_id_list or notification.get('model_id') not in ignore_models_id_list
+        )
+        return level_check and model_id_check
+
+    @staticmethod
+    def _check_keys_and_notif_id(
+        notification_id, ignore_list, key, already_added_keys: list[str]
+    ) -> bool:
+        """
+        Check if the key has already been added to the list of already added keys.
+
+        Args:
+            notification_id (str): The ID of the notification.
+            ignore_list (list): List of notification IDs to ignore.
+            key (str): The key to check.
+            already_added_keys (list[str]): List of keys that have already been added.
+
+        Returns:
+            bool: True if the key has not been added yet, False otherwise.
+
+        """
+        return notification_id not in ignore_list and key not in already_added_keys
 
     @activity.defn(name='store_notification_cache')
     def store_notification_cache(self, input_data: dict[str, Any]) -> None:

orchestrator/workflows/subworkflows/load_notification_package.py

@@ -68,18 +68,23 @@ class LoadNotificationPackage:
         # Load notification package from collection "notification_queue", using a
         # static filter
 
+        sending_configs = await sending_configs_handler
+
+        base_data_filter = self._compute_query_time_filter(
+            sending_configs, input_data['base_data_filter']
+        )
+
         notification_package = await workflow.start_local_activity_method(
             Activities.load_latest_data,
             {
                 **metadata,
                 'collection_name': 'notification_queue',
                 'last_data_timestamp': last_timestamp,
-                'base_data_filter': input_data['base_data_filter'],
+                'base_data_filter': base_data_filter,
             },
             start_to_close_timeout=timedelta(seconds=60),
             retry_policy=retry_policy,
         )
-        sending_configs = await sending_configs_handler
 
         if not sending_configs or not notification_package:
             return {
@@ -106,3 +111,56 @@ class LoadNotificationPackage:
             'notification_package': notification_package,
             'sending_configs': sending_configs,
         }
+
+    def _compute_query_time_filter(
+        self, sending_configs: list[dict[str, Any]], default_filter: dict[str, Any]
+    ) -> dict[str, Any]:
+        """
+        Compute the Mongo query-time filter from all active receiver groups.
+
+        - levels: union of every group's `levels`. If ANY active group has no
+          restriction (missing/empty `levels`), that group wants every level, so
+          the query can't restrict levels at all -> falls back to whatever level
+          restriction `default_filter` itself carries (e.g. ERROR-only for Alerts).
+
+          When every active group DOES restrict `levels`, the union REPLACES
+          `default_filter`'s own level restriction rather than ANDing with it —
+          this is deliberate, not an oversight: SIENTIAPDE-2011's design doc and
+          ticket text both call for Alerts' hard-coded `level: ERROR` to become
+          "whatever levels the active groups actually care about", falling back
+          to ERROR-only only when no group narrows further. Until SIENTIAPDE-2012
+          (per-group `levels` filtering) lands, this means a single active group
+          configuring `levels` to include WARNING/INFO widens the Alerts fetch for
+          every active group, not just that one — an accepted interim gap, not a
+          bug in this function.
+        - ignore_models_id: intersection of every group's `ignore_models_id`. A model
+          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.
+        """
+        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)}}
+        else:
+            query_filter = dict(default_filter)
+
+        # `or []`, not a `.get(..., [])` default: a document with `ignore_models_id`
+        # explicitly set to `null` (no schema validation on this collection) makes
+        # `.get()` return None, and `set(None)` would raise TypeError.
+        sending_configs_ignore_models_id = [
+            set(group.get('ignore_models_id') or []) for group in sending_configs
+        ]
+        models_intersection = set.intersection(*sending_configs_ignore_models_id)
+
+        if models_intersection:
+            query_filter['model_id'] = {'$nin': sorted(models_intersection)}
+
+        return query_filter

pyproject.toml

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

tests/orchestrator/activities/test_formatters.py

@@ -716,7 +716,7 @@ def test_format_log_report(formatters):
         [
             {
                 'status': 'sent',
-                'timestamp': '2021-01-01',
+                'timestamp': '2021-01-01 00:00:00+0000',
                 'groups': ['test_receiver_group', 'test_receiver_group2'],
                 'message': 'test_message',
                 'level': 'test_level',
@@ -766,3 +766,143 @@ def test_filter_notification_reports(formatters):
             ],
         }
     }
+
+
+def test_filter_notification_reports_levels_restriction(formatters):
+    # SIENTIAPDE-2012: a group with `levels` set only receives notifications
+    # matching one of those levels.
+    input_data = {
+        **metadata,
+        'notification_package': [
+            {
+                'trigger': 'test_trigger_1',
+                'notification_id': 'test_notification_id_1',
+                'level': 'ERROR',
+            },
+            {
+                'trigger': 'test_trigger_2',
+                'notification_id': 'test_notification_id_2',
+                'level': 'INFO',
+            },
+        ],
+        'sending_configs': [
+            {'group_name': 'test_group_1', 'contents': ['reports'], 'levels': ['ERROR']},
+        ],
+    }
+
+    response = formatters.filter_notification_reports(input_data)
+
+    assert response == {
+        'test_group_1': {
+            'group_name': 'test_group_1',
+            'contents': ['reports'],
+            'levels': ['ERROR'],
+            'notifications': [
+                {
+                    'trigger': 'test_trigger_1',
+                    'notification_id': 'test_notification_id_1',
+                    'level': 'ERROR',
+                },
+            ],
+        },
+    }
+
+
+def test_filter_notification_reports_ignore_models_id_restriction(formatters):
+    # SIENTIAPDE-2012: a group with `ignore_models_id` set does not receive
+    # notifications from those models, but still receives everything else.
+    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'],
+                'ignore_models_id': ['model_x'],
+            },
+        ],
+    }
+
+    response = formatters.filter_notification_reports(input_data)
+
+    assert response == {
+        'test_group_1': {
+            'group_name': 'test_group_1',
+            'contents': ['reports'],
+            '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_no_levels_or_ignore_models_id_passes_everything(
+    formatters,
+):
+    # SIENTIAPDE-2012: a group configuring neither field is unaffected -- same
+    # behavior as before this subtask (only `contents`/`ignore` apply).
+    input_data = {
+        **metadata,
+        'notification_package': [
+            {
+                'trigger': 'test_trigger_1',
+                'notification_id': 'test_notification_id_1',
+                'level': 'ERROR',
+                'model_id': 'model_x',
+            },
+            {
+                'trigger': 'test_trigger_2',
+                'notification_id': 'test_notification_id_2',
+                'level': 'INFO',
+                'model_id': 'model_y',
+            },
+        ],
+        'sending_configs': [
+            {'group_name': 'test_group_1', 'contents': ['reports']},
+        ],
+    }
+
+    response = formatters.filter_notification_reports(input_data)
+
+    assert response['test_group_1']['notifications'] == input_data['notification_package']
+
+
+def test_filter_notification_reports_ignore_models_id_explicit_none_does_not_raise(
+    formatters,
+):
+    # receiver_groups has no schema validation -- ignore_models_id can be
+    # explicitly null (key present, value None) instead of omitted. Must not
+    # crash (see the equivalent SIENTIAPDE-2011 fix in load_notification_package).
+    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'], 'ignore_models_id': None},
+        ],
+    }
+
+    response = formatters.filter_notification_reports(input_data)
+
+    assert response['test_group_1']['notifications'] == input_data['notification_package']

tests/orchestrator/activities/test_mongo_db.py

@@ -504,7 +504,7 @@ def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
         {
             'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
             'collection_name': 'test_collection',
-            'last_data_timestamp': '2023-01-01T12:00:00+00:00',
+            'last_data_timestamp': '2023-01-01 12:00:00.000000+0000',
             'base_data_filter': {'level': 'ERROR'},
         }
     )
@@ -561,7 +561,7 @@ def test_load_latest_data_run_boundary_no_skip_or_duplicate(mongo_db):
         {
             'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
             'collection_name': 'test_collection',
-            'last_data_timestamp': boundary_timestamp.isoformat(),
+            'last_data_timestamp': boundary_timestamp.strftime(DATETIME_FORMAT_MS_WITH_TZ),
             'base_data_filter': {'level': 'ERROR'},
         }
     )
@@ -578,7 +578,7 @@ def test_load_latest_data_error(mongo_db):
             {
                 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
                 'collection_name': 'test_collection',
-                'last_data_timestamp': '2023-01-01T12:00:00+00:00',
+                'last_data_timestamp': '2023-01-01 12:00:00.000000+0000',
                 'base_data_filter': {'level': 'ERROR'},
             }
         )

tests/orchestrator/activities/test_slot_manager.py

@@ -297,10 +297,10 @@ def test_put_last_data_timestamp_not_empty_dataframe(slot_manager):
 
     result = slot_manager.put_last_data_timestamp(test_data)
 
-    assert result == '2023-01-01T12:00:01+00:00'
+    assert result == '2023-01-01 12:00:01.000000+0000'
 
     slot_manager.redis_repository.set.assert_called_once_with(
-        'notification_last_timestamp:test_mail_type', '2023-01-01T12:00:01+00:00', ttl=18000
+        'notification_last_timestamp:test_mail_type', '2023-01-01 12:00:01.000000+0000', ttl=18000
     )
 
 
@@ -389,6 +389,162 @@ def test_filter_notification_alerts(slot_manager):
     }
 
 
+def test_filter_notification_alerts_levels_restriction(slot_manager):
+    # SIENTIAPDE-2012: a group with `levels` set only receives notifications
+    # matching one of those levels.
+    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',
+                'level': 'ERROR',
+            },
+            {
+                'trigger': 'test_trigger_2',
+                'notification_id': 'test_notification_id_2',
+                'level': 'WARNING',
+            },
+        ],
+        'sending_configs': [
+            {'group_name': 'test_group_1', 'contents': ['core_alerts'], 'levels': ['ERROR']},
+        ],
+        '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'],
+            'levels': ['ERROR'],
+            'notifications': [
+                {
+                    'trigger': 'test_trigger_1',
+                    'notification_id': 'test_notification_id_1',
+                    'level': 'ERROR',
+                },
+            ],
+        },
+    }
+
+
+def test_filter_notification_alerts_ignore_models_id_restriction(slot_manager):
+    # SIENTIAPDE-2012: a group with `ignore_models_id` set does not receive
+    # notifications from those models, but still receives everything else.
+    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'],
+                'ignore_models_id': ['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'],
+            '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_no_levels_or_ignore_models_id_passes_everything(
+    slot_manager,
+):
+    # SIENTIAPDE-2012: a group configuring neither field is unaffected -- same
+    # behavior as before this subtask (only `contents`/`ignore` apply).
+    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',
+                'level': 'ERROR',
+                'model_id': 'model_x',
+            },
+            {
+                'trigger': 'test_trigger_2',
+                'notification_id': 'test_notification_id_2',
+                'level': 'INFO',
+                'model_id': 'model_y',
+            },
+        ],
+        'sending_configs': [
+            {'group_name': 'test_group_1', 'contents': ['core_alerts']},
+        ],
+        '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_ignore_models_id_explicit_none_does_not_raise(
+    slot_manager,
+):
+    # receiver_groups has no schema validation -- ignore_models_id can be
+    # explicitly null (key present, value None) instead of omitted. Must not
+    # crash (see the equivalent SIENTIAPDE-2011 fix in load_notification_package).
+    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'], 'ignore_models_id': 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_store_notification_cache(slot_manager):
     """Test store_notification_cache"""
     test_data = {

tests/orchestrator/workflows/subworkflows/test_load_notification_package.py

@@ -32,16 +32,19 @@ async def test_run(workflow_mock, load_notification_package):
         'base_data_filter': {'level': 'ERROR'},
     }
 
+    # Await order is now: last_timestamp -> sending_configs -> notification_package
+    # (fixed ordering bug: sending_configs must resolve before load_latest_data is
+    # called, so its levels/ignore_models_id can feed the query-time filter).
     workflow_mock.start_local_activity_method.side_effect = [
         '2023-01-01 12:00:00',
         [
             {
-                'id': '1',
+                'id_r': '1',
             }
         ],
         [
             {
-                'id_r': '1',
+                'id': '1',
             }
         ],
     ]
@@ -158,7 +161,9 @@ async def test_run_has_data(workflow_mock, load_notification_package):
         'mail_type': 'test_mail_type',
     }
 
-    workflow_mock.start_local_activity_method.side_effect = ['2023-01-01 12:00:00', ['data'], []]
+    # sending_configs (awaited 2nd) is empty here on purpose, to exercise the
+    # early-return branch below regardless of notification_package having data.
+    workflow_mock.start_local_activity_method.side_effect = ['2023-01-01 12:00:00', [], ['data']]
 
     output = await load_notification_package.run(input_data)
 
@@ -169,3 +174,103 @@ async def test_run_has_data(workflow_mock, load_notification_package):
     }
 
     workflow_mock.start_activity_method.assert_not_called()
+
+
+def test_compute_query_time_filter_empty_active_groups(load_notification_package):
+    default_filter = {'level': 'ERROR'}
+
+    result = load_notification_package._compute_query_time_filter([], default_filter)
+
+    assert result == default_filter
+
+
+def test_compute_query_time_filter_mixed_groups_no_levels_restriction(load_notification_package):
+    # One active group restricts levels/ignore_models_id, the other has neither set
+    # (the common case today) -> that group wants everything, so no levels
+    # restriction can be applied at query time; falls back to default_filter.
+    sending_configs = [
+        {'group_name': 'group_a', 'levels': ['ERROR'], 'ignore_models_id': ['model_a']},
+        {'group_name': 'group_b'},
+    ]
+    default_filter = {'level': 'ERROR'}
+
+    result = load_notification_package._compute_query_time_filter(sending_configs, default_filter)
+
+    assert result == {'level': 'ERROR'}
+
+
+def test_compute_query_time_filter_union_replaces_default_level_restriction(
+    load_notification_package,
+):
+    # default_filter already restricts level (e.g. Alerts' ERROR-only default),
+    # but every active group specifies levels of its own -> the union REPLACES
+    # default_filter's level restriction, it does not AND with it. This is
+    # deliberate per SIENTIAPDE-2011's design doc and ticket text ("Alerts'
+    # hard-coded level: ERROR should become whatever levels the active groups
+    # actually care about"), not a bug -- this test locks the behavior in so a
+    # future "fix" reintroducing AND semantics fails loudly instead of passing
+    # silently.
+    sending_configs = [
+        {'group_name': 'group_a', 'levels': ['WARNING'], 'ignore_models_id': []},
+    ]
+    default_filter = {'level': 'ERROR'}
+
+    result = load_notification_package._compute_query_time_filter(sending_configs, default_filter)
+
+    assert result == {'level': {'$in': ['WARNING']}}
+
+
+def test_compute_query_time_filter_model_ignored_by_only_one_group_not_excluded(
+    load_notification_package,
+):
+    # model_a is ignored by group_a but group_b still wants it -> must stay in
+    # the fetch (the per-group filter is what drops it for group_a specifically).
+    sending_configs = [
+        {'group_name': 'group_a', 'levels': ['ERROR'], 'ignore_models_id': ['model_a']},
+        {'group_name': 'group_b', 'levels': ['ERROR'], 'ignore_models_id': []},
+    ]
+    default_filter = {}
+
+    result = load_notification_package._compute_query_time_filter(sending_configs, default_filter)
+
+    assert result == {'level': {'$in': ['ERROR']}}
+    assert 'model_id' not in result
+
+
+def test_compute_query_time_filter_union_levels_and_intersection_models(
+    load_notification_package,
+):
+    # Every active group restricts levels -> union of levels applies.
+    # model_a is ignored by every active group -> safe to exclude at query time.
+    sending_configs = [
+        {'group_name': 'group_a', 'levels': ['ERROR'], 'ignore_models_id': ['model_a', 'model_b']},
+        {'group_name': 'group_b', 'levels': ['WARNING', 'ERROR'], 'ignore_models_id': ['model_a']},
+    ]
+    default_filter = {}
+
+    result = load_notification_package._compute_query_time_filter(sending_configs, default_filter)
+
+    assert result == {
+        'level': {'$in': ['ERROR', 'WARNING']},
+        'model_id': {'$nin': ['model_a']},
+    }
+
+
+def test_compute_query_time_filter_ignore_models_id_explicit_none_does_not_raise(
+    load_notification_package,
+):
+    # receiver_groups has no schema validation, so a document can have
+    # ignore_models_id explicitly set to null (key present, value None) rather
+    # than omitted. `group.get('ignore_models_id', [])` would still return None
+    # in that case (the default only applies when the key is missing), and
+    # `set(None)` raises TypeError -- must not crash the whole activity.
+    sending_configs = [
+        {'group_name': 'group_a', 'levels': ['ERROR'], 'ignore_models_id': None},
+        {'group_name': 'group_b', 'levels': ['ERROR'], 'ignore_models_id': ['model_a']},
+    ]
+    default_filter = {}
+
+    result = load_notification_package._compute_query_time_filter(sending_configs, default_filter)
+
+    assert result == {'level': {'$in': ['ERROR']}}
+    assert 'model_id' not in result