SIENTIAPDE-1974: refactor(slots): fix the SonarQube quality gate findings

SonarQube on PR #45 reported 3 new issues; all fixed without behavior change:
- S3776 gather_read_tags (cognitive complexity 22): the merge of one resolved
  read tag into its (server, address) entry moves to _add_read_tag (now 14).
- S3776 Orchestrator.run (16): the tag database load moves to
  Orchestrator._load_tag_database (run back to 14). workflow.patched() is still
  evaluated first on every execution, so its marker is always recorded.
- S1192 '$match' duplicated in the workflow module: the connector_tags tag_id
  filter moves to orchestrator_functions.filter_by_tag_ids, which also returns a
  new aggregation list.

194 unit tests (new: filter_by_tag_ids); real workflow on a local Temporal test
server still passes the 3 scenarios; the dev dry run is still equivalent.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.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,9 +45,9 @@ 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. Receiver groups also support a `model_ids` allow-list (SIENTIAPDE-2149), independent of the pre-existing `ignore_models_id` deny-list — both apply together (AND) when a group configures both, and `model_ids` follows the same union-at-query-time / precise-per-group-slice split as `levels`. `load_latest_data`'s Mongo query is capped at `NOTIFICATION_QUEUE_BATCH_SIZE` documents per run (default `0`, meaning unlimited — unchanged from pre-SIENTIAPDE-2013 behavior unless a deployment opts in), sorted by `timestamp` ascending so a capped run always returns the oldest pending batch first (SIENTIAPDE-2013) — a hard cap with no drain loop, since the incremental Redis cursor picks up any remainder on the next run.
 - **`orchestrator/activities/activities.py`** — `Activities` is a single class combining every activity mixin via multiple inheritance: `TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres` (Postgres comes from `sientia_do.temporal.activities.postgres`, the external `sientia-dataops-library` dependency). One instance is constructed in `worker.py` and its bound methods are handed to each `Worker` as the activity list — activities are plain methods on this shared object, not separate classes. Each mixin's `__init__` takes its own config dict plus shared `logger`, `notification_handler`, `metrics_controller`. `Activities.shutdown()` closes every mixin's connections.
-- **`orchestrator/utils/`** — `connectors_config.py` builds each config dict from env vars (Temporal/Redis/MongoDB/Postgres/Email); `email_builder.py` renders Jinja2 templates (`utils/templates/email_template.html`, `general_template.html`); `orchestrator_functions.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `converters.py` handles Temporal schedule frequency parsing.
+- **`orchestrator/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`) and builds the OPC slot v2 the gateway reads (`gather_read_tags` / `build_tag_config`, contract C1: one entry per address with a `names[]` list, tags resolved through the tag database when they carry a `tag_id`; see the README section "OPC slots"). `resolve_tag_names` is the single source of tag names for both the scouter's `model_tags` and the slot's `names` (SIENTIAPDE-1974 SC1); `converters.py` handles Temporal schedule frequency parsing.
 
 ### Temporal-specific conventions (all workflow/activity files follow this)
 - Every workflow and activity-consuming module wraps non-deterministic/side-effecting imports in `with workflow.unsafe.imports_passed_through(): ...` right after `from temporalio import workflow`, because Temporal's sandbox restricts what workflow code can import directly. When adding a new workflow file, follow the existing pattern (see `orchestrator/workflows/orchestrator.py`, `orchestrator/activities/activities.py`) rather than importing at module top level.

README.md

@@ -97,7 +97,7 @@ The **Orchestrator** workflow is the main coordination workflow that manages pip
 - **Infrastructure Management**: Creates, updates, and deletes workflow schedules
 
 **Execution Flow**:
-1. **Parallel Data Loading**: Concurrently loads pipelines, OPC servers, orchestrated schedules, OPC slots, and active ingestors
+1. **Parallel Data Loading**: Concurrently loads pipelines, OPC servers, orchestrated schedules, OPC slots, active ingestors, and the tag database (`connectors` and `connector_tags`)
 2. **Parallel Processing**: Formats orchestrated schedules and processes new schedules and slots
 3. **Parallel Config Creation**: Creates schedule and slot action configurations, normalizes schedules, and creates collections with TTL indexes
 4. **Parallel Operations**: Executes slot deletions, slot updates, schedule deletions, schedule creations, and schedule updates concurrently
@@ -121,6 +121,81 @@ The **Orchestrator** workflow is the main coordination workflow that manages pip
 }
 ```
 
+Optional: `connectors_query` and `connector_tags_query` override the tag database aggregations. The defaults live in the code (`orchestrator/workflows/orchestrator.py`), so the schedule input does not need them:
+
+```json
+{
+  "connectors_query": {
+    "collection": "connectors",
+    "aggregation": [
+      {"$match": {"type": "opc_ua", "status": "Active"}},
+      {"$project": {"_id": 0, "connector_id": 1, "name": 1, "url": 1, "server_uri": 1,
+                    "use_security": 1, "subscription_period_ms": 1}}
+    ]
+  },
+  "connector_tags_query": {
+    "collection": "connector_tags",
+    "aggregation": [
+      {"$match": {"category": "Read", "status": "Active"}},
+      {"$project": {"_id": 0, "tag_id": 1, "connector_id": 1, "address": 1, "tag_name": 1}}
+    ]
+  }
+}
+```
+
+The tag database is read only when a scouter read tag carries a `tag_id`, and the workflow puts `{"$match": {"tag_id": {"$in": [...]}}}` in front of `connector_tags_query`, so it loads only the referenced tags. Environments without a tag database, or whose pipelines do not use `tag_id`, never query it. An override should keep a `$project` of these fields: any other attribute may hold a BSON type a Temporal payload cannot carry (such as `ObjectId`).
+
+#### OPC slots (slot v2, contract C1)
+
+`process_slots` writes one slot per active ingestor to Redis (`slot:opc_tags:{n}`). The OPC Gateway reads it (SIENTIAPDE-1974). Each server carries its connection fields, and each tag address carries the list of names it is written under:
+
+```json
+{
+  "kepserver": {
+    "server_id": "1", "name": "kepserver", "url": "opc.tcp://host:49320",
+    "server_uri": "urn:...", "use_security": true, "subscription_period_ms": 7500.0,
+    "tags": {
+      "ns=2;s=Channel1.Device1.Tag1": {
+        "frequency": 15000,
+        "active_read": {"enabled": false},
+        "names": [
+          {"collection": "raw_model_a", "tag_name": "Counter",  "model_id": "1", "model_name": "model a"},
+          {"collection": "raw_model_b", "tag_name": "Contador", "model_id": "2", "model_name": "model b"}
+        ]
+      }
+    }
+  }
+}
+```
+
+How each scouter `read_tags` entry is resolved:
+
+| Read tag | Address and server | Name |
+|---|---|---|
+| With `tag_id` | `connector_tags` (`address`, `connector_id`) and `connectors` (`url`, `server_uri`, `use_security`, `subscription_period_ms`) | The read tag's `tag_name`, else the `connector_tags` one |
+| Without `tag_id` | `server_id` + `tag_address`, server from `opc-servers` | The read tag's `tag_name` |
+
+Rules:
+- **One entry per address.** An address read by N pipelines is read once and written to N collections: one `names` item per pipeline (`collection` = `raw_{schedule_name}`).
+- **Same name source as the scouter.** `resolve_tag_names` names both the slot's `names` and the scouter's `model_tags`, so the gateway always writes the name the scouter looks for.
+- **`frequency`:** the lowest among the pipelines of the address, as an integer. Every read tag needs one, with or without `tag_id`.
+- **`tag_id`:** a missing, `null` or empty `tag_id` means the read tag does not use the tag database.
+- **`subscription_period_ms`:** the connector's value. For `opc-servers`, it is half the server's lowest tag frequency in the slot, as before.
+- **`use_security`:** the connector's value. For `opc-servers`, it is `true` when the server has a `cert_path`. The certificate itself comes from MinIO on the gateway, so the slot carries no certificate path.
+- **`active_read`:** off by default. It is on when any pipeline sets `read_tags[].active_read.enabled`, with the lowest `cycle_multiple`.
+- **Same `server_id` in both sources:** a server known through `connectors` and `opc-servers` under the same id is one server, and the `connectors` fields win. Two different servers with the same name get distinct slot keys (`name`, then `name:server_id`).
+- **Slot division:** the addresses are split across the slots, sorted by server and address, and the names of an address always stay in the same slot.
+- **Unresolved tags:** a read tag that cannot be resolved is left out of the slot, and the rest of the slot is still written. It is notified once per run:
+  - `ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION`: the `tag_id` is not an active Read tag, or its document lacks `connector_id`, `address` or `tag_name`;
+  - `ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION`: the connector is missing or inactive, or lacks `name`, `url` or `server_uri`;
+  - `ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION`: the `server_id` is not in `opc-servers`;
+  - `ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION`: the read tag has no integer `frequency` or, without `tag_id`, lacks `server_id`, `tag_address` or `tag_name`.
+- **Only changed slots are written.** The slot is built in a stable order, so unchanged pipelines produce an identical slot, and the gateway does not re-subscribe. A slot still in the v1 format (`topics`, `cert_path`) is replaced once.
+
+**Gateway plan B:** the gateway's legacy read engine (`READ_ENGINE=legacy`, kept until the end of SIENTIAPDE-1974) reads the slot v1 only. Once this version writes the slot v2, switching the gateway to the legacy engine also requires rolling the Orchestrator back to a version that writes v1.
+
+**Known limitation:** scouter schedules are only redeployed when the pipeline's `updated_at` changes. If a tag is renamed in `connector_tags`, the slot picks up the new name on the next run, but the scouter keeps the old `model_tags` until the pipeline is updated. When renaming, also update the pipeline, or keep the name in `read_tags[].tag_name`.
+
 #### Architecture
 
 ```mermaid
@@ -608,10 +683,27 @@ Temporal input configuration sample:
         "filters": {
 
         }
+    },
+    "connectors_query": {
+        "collection": "connectors",
+        "aggregation": [
+            {"$match": {"type": "opc_ua", "status": "Active"}},
+            {"$project": {"_id": 0, "connector_id": 1, "name": 1, "url": 1, "server_uri": 1,
+                          "use_security": 1, "subscription_period_ms": 1}}
+        ]
+    },
+    "connector_tags_query": {
+        "collection": "connector_tags",
+        "aggregation": [
+            {"$match": {"category": "Read", "status": "Active"}},
+            {"$project": {"_id": 0, "tag_id": 1, "connector_id": 1, "address": 1, "tag_name": 1}}
+        ]
     }
 }
 ```
 
+`connectors_query` and `connector_tags_query` are optional: the values above are the defaults in the code (see "OPC slots" above).
+
 #### 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

@@ -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,8 @@ 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,
+    model_ids: list[str] | None = None,
 ) -> dict[str, Any]:
     """Minimal active receiver_groups document."""
     return {
@@ -352,5 +356,43 @@ 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 [],
+        'model_ids': model_ids 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,20 @@ 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.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
 
@@ -103,7 +104,11 @@ 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

@@ -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,162 @@ 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_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_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,129 @@ 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_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(
@@ -152,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

@@ -7,19 +7,26 @@ 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,
         drift,
         gather_read_tags,
+        index_connector_tags,
+        index_connectors,
         minimal_retrain,
         pi_web_api_scouter,
         predictions_batch,
+        resolve_tag_names,
         scouter,
         simple_metrics,
     )
@@ -169,6 +176,8 @@ class Formatters(SientiaMonitoring):
             - input_data (dict[str, Any]): The input data containing
               the schedules to process.
                 - pipelines (list[dict[str, Any]]): The schedules to process.
+                - connector_tags (list[dict[str, Any]], optional): Active Read tags of
+                  the tag database, to name the scouter read tags that carry only a tag_id.
 
         Returns:
             - dict[str, Any]: The schedule configuration dictionary keyed by namespace
@@ -179,6 +188,7 @@ class Formatters(SientiaMonitoring):
         self.info('Processing schedules...', metadata=metadata)
 
         pipelines = input_data['pipelines']
+        tags_by_id = index_connector_tags(input_data.get('connector_tags', []))
 
         schedule_config: dict[str, dict[str, Any]] = {
             self.scouter_namespace: {},
@@ -192,6 +202,13 @@ class Formatters(SientiaMonitoring):
                 self.error(f'Workflow type {workflow_type} not supported', metadata=metadata)
                 continue
 
+            if workflow_type == 'scouter' and 'read_tags' in pipeline:
+                # Same names as the slot's names[] (process_slots): SIENTIAPDE-1974 SC1
+                pipeline = {
+                    **pipeline,
+                    'read_tags': resolve_tag_names(pipeline['read_tags'], tags_by_id),
+                }
+
             schedule_type = schedule_types[workflow_type]
             namespace = schedule_type['namespace']
             function = schedule_type['function']
@@ -212,14 +229,25 @@ class Formatters(SientiaMonitoring):
     def process_slots(self, input_data: dict[str, Any]) -> dict[str, Any]:
         """
         Extracts all read tags from input pipelines, divides them into slots and
-        returns a slot config dictionary. If no ingestor is available, only one slot
-        is created.
+        returns a slot config dictionary in the slot v2 format (contract C1). If no
+        ingestor is available, only one slot is created.
+
+        The division is by address: all the names of an address stay in the same slot
+        (SIENTIAPDE-1974, O5). Read tags that cannot be resolved are left out and
+        notified once per run: a tag_id missing from connector_tags, a connector missing
+        or inactive, a server_id missing from opc_servers, or a read tag without a usable
+        frequency (or, without tag_id, without server_id, tag_address or tag_name).
 
         Args:
             - input_data (dict[str, Any]): The input data containing
               the schedules to process.
                 - pipelines (list[dict[str, Any]]): The schedules to process.
-                - opc_servers (list[str]): The OPC servers to create ingestor config.
+                - opc_servers (list[dict[str, Any]]): The OPC servers (opc-servers), for
+                  read tags without tag_id.
+                - connectors (list[dict[str, Any]], optional): Active OPC UA connectors of
+                  the tag database, for read tags with tag_id.
+                - connector_tags (list[dict[str, Any]], optional): Active Read tags of the
+                  tag database, for read tags with tag_id.
                 - active_ingestors (list[str]): The active ingestors to divide into slots.
 
         Returns:
@@ -231,55 +259,69 @@ class Formatters(SientiaMonitoring):
         self.info('Processing slots...', metadata=metadata)
 
         pipelines = input_data['pipelines']
-        opc_servers_list = input_data['opc_servers']
         active_ingestors = input_data['active_ingestors']
+        opc_servers = {str(server['id']): server for server in input_data['opc_servers']}
+        connectors = index_connectors(input_data.get('connectors', []))
+        tags_by_id = index_connector_tags(input_data.get('connector_tags', []))
 
-        opc_servers = {}
-        for server in opc_servers_list:
-            opc_servers[server['id']] = {
-                **server,
-            }
+        entries, issues = gather_read_tags(pipelines, opc_servers, connectors, tags_by_id)
+        self._notify_slot_issues(issues, metadata)
 
-        tags = list(gather_read_tags(pipelines).values())
-
-        number_of_tags = len(tags)
         number_of_slots = len(active_ingestors) if active_ingestors else 1
-        tags_per_slot = ceil(number_of_tags / number_of_slots)
+        entries_per_slot = ceil(len(entries) / number_of_slots)
 
         slot_config: dict[str, Any] = {}
-        last_index = 0
-
-        for i in range(1, number_of_slots):
-            slot_tags = tags[last_index : last_index + tags_per_slot]
-            slot_config[f'{i}'], notifications = build_tag_config(slot_tags, opc_servers)
-
-            if notifications:
-                self.send_notification(
-                    metadata=metadata,
-                    notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
-                    message=f'Servers {", ".join(notifications)} not found in opc_servers',
-                    block='orchestrator',
-                    level=NotificationLevel.ERROR,
-                )
-
-            last_index += tags_per_slot
-
-        slot_tags = tags[last_index:]
-        slot_config[f'{number_of_slots}'], notifications = build_tag_config(slot_tags, opc_servers)
-        if notifications:
-            self.send_notification(
-                metadata=metadata,
-                notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
-                message=f'Servers {", ".join(notifications)} not found in opc_servers',
-                block='orchestrator',
-                level=NotificationLevel.ERROR,
-            )
+        for i in range(number_of_slots):
+            slot_entries = entries[i * entries_per_slot : (i + 1) * entries_per_slot]
+            slot_config[f'{i + 1}'] = build_tag_config(slot_entries)
 
         self.info('Processed slots', metadata=metadata)
         self.debug(json.dumps(slot_config, indent=4, sort_keys=True), metadata=metadata)
 
         return slot_config
 
+    def _notify_slot_issues(self, issues: dict[str, set[str]], metadata: dict[str, Any]) -> None:
+        """
+        Notify the read tags left out of the slots, one notification per kind of issue.
+
+        Args:
+            issues (dict[str, set[str]]): What gather_read_tags left out, by kind:
+                'servers' (opc-servers server_id), 'tags' (tag_id), 'connectors'
+                (connector_id) and 'invalid' (schedule_name/read tag)
+            metadata (dict[str, Any]): Metadata for the notifications
+        """
+        kinds = (
+            (
+                'servers',
+                'ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
+                'Servers {} not found in opc_servers',
+            ),
+            (
+                'tags',
+                'ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION',
+                'Tags {} not found among the active, complete Read tags of connector_tags',
+            ),
+            (
+                'connectors',
+                'ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION',
+                'Connectors {} not found, not active or incomplete in connectors',
+            ),
+            (
+                'invalid',
+                'ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION',
+                'Read tags {} have no integer frequency or lack server_id, tag_address or tag_name',
+            ),
+        )
+        for kind, notification_id, message in kinds:
+            if issues.get(kind):
+                self.send_notification(
+                    metadata=metadata,
+                    notification_id=notification_id,
+                    message=message.format(', '.join(sorted(issues[kind]))),
+                    block='orchestrator',
+                    level=NotificationLevel.ERROR,
+                )
+
     @activity.defn(name='format_schedule_config')
     def format_schedule_config(self, input_data: dict[str, Any]) -> dict[str, Any]:
         """
@@ -426,15 +468,21 @@ class Formatters(SientiaMonitoring):
         to determine which slots need to be inserted or deleted. Slots are identified
         by numeric IDs, and excess slots are marked for deletion.
 
+        Only slots whose content changed are inserted, so an unchanged slot is not
+        rewritten (nor reported) on every run. gather_read_tags builds the slot in a
+        stable order, so the same pipelines always give the same slot (SIENTIAPDE-1974, O6).
+        A slot in the v1 format differs from its v2 rewrite and is replaced once.
+
         Args:
             input_data (dict[str, Any]): The input data containing:
-                - current_slot_config (dict[str, Any]): The current slot config in Redis
+                - current_slot_config (dict[str, Any]): The current slot config in Redis,
+                  keyed by Redis key (slot:opc_tags:{id}) or by slot id
                 - slot_config (dict[str, Any]): The slot config to process
                 - metadata (dict): Metadata for logging purposes
 
         Returns:
             dict[str, Any]: A dictionary containing:
-                - to_insert (dict): Slots that need to be inserted/updated
+                - to_insert (dict): Slots that are new or changed
                 - to_delete (list[str]): Slot IDs that need to be deleted
         """
 
@@ -452,7 +500,16 @@ class Formatters(SientiaMonitoring):
         if number_of_current_slots > number_of_slots:
             to_delete = [str(i) for i in range(number_of_slots + 1, number_of_current_slots + 1)]
 
-        output = {'to_delete': to_delete, 'to_insert': slot_config}
+        current_by_id = {
+            str(key).rsplit(':', 1)[-1]: config for key, config in current_slot_config.items()
+        }
+        to_insert = {
+            slot: config
+            for slot, config in slot_config.items()
+            if slot not in current_by_id or current_by_id[slot] != config
+        }
+
+        output = {'to_delete': to_delete, 'to_insert': to_insert}
 
         self.info('Created slot config', metadata=metadata)
         self.debug(json.dumps(output, indent=4, sort_keys=True), metadata=metadata)
@@ -755,9 +812,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 +877,9 @@ 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
+                    - model_ids (list, optional): Model IDs to allow (SIENTIAPDE-2149)
 
         Returns:
             dict[str, Any]: Filtered receiver groups with their notifications, keyed by group_name.
@@ -828,6 +903,9 @@ 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 []
+            model_ids_list = receiver_group.get('model_ids')
 
             for notification in notification_package:
                 alert_type = 'reports'
@@ -840,6 +918,12 @@ 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
+                    )
+                    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')
@@ -447,13 +451,17 @@ 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:
                 - 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,17 +483,26 @@ 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)
 
-            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

@@ -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,9 @@ 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
+                    - model_ids (list, optional): Model IDs to allow (SIENTIAPDE-2149)
                 - notification_ttl (int): Seconds before considering notification persistent
 
         Returns:
@@ -399,48 +418,125 @@ 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 []
+        model_ids_list = receiver_group.get('model_ids')
+
+        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_model_filters(
+                    notification, levels_list, ignore_models_id_list, model_ids_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_model_filters(
+        notification, levels_list, ignore_models_id_list, model_ids_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
+        ignore_models_id_check = (
+            not ignore_models_id_list or notification.get('model_id') not in ignore_models_id_list
+        )
+        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(
+        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/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

@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, cast
 
 from orchestrator.utils.converters import parse_frequency
 
@@ -174,8 +174,10 @@ def scouter(config: dict[str, Any]):
     Args:
         config (dict[str, Any]): Pipeline configuration containing:
             - schedule_name (str): Name of the schedule (used for topic generation)
-            - read_tags (list[dict]): List of tag configurations with:
-                - tag_name (str): Name of the tag to read
+            - read_tags (list[dict]): List of tag configurations, already passed through
+              resolve_tag_names (process_schedules does it), with:
+                - tag_name (str): Name of the tag to read. A tag without one (a tag_id the
+                  tag database cannot resolve) is left out; process_slots notifies it
                 - aggr_func (str, optional): Aggregation function for data collection (default: 'lts')
                     Common values: 'lts' (last), 'avg' (average), 'min', 'max', 'sum'
                 - data_range (list[int], optional): Valid data range [min, max] (default: [-100, 100])
@@ -188,6 +190,8 @@ def scouter(config: dict[str, Any]):
 
     tags = {}
     for tag in config['read_tags']:
+        if not tag.get('tag_name'):
+            continue
         tags[tag['tag_name']] = {
             'aggr_func': tag.get('aggr_func', 'lts'),
             'data_range': tag.get('data_range', [-100, 100]),
@@ -396,110 +400,408 @@ def predictions_batch(config: dict[str, Any]):
     }
 
 
-def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]:
-    """
-    Gather all read tags from scouter pipeline configurations.
+# Fields a tag database document needs to take part in a slot. Documents missing any
+# of them are left out of the index, so their read tags are notified as not found
+# instead of failing the whole run.
+CONNECTOR_TAG_FIELDS = ('tag_id', 'connector_id', 'address', 'tag_name')
+CONNECTOR_FIELDS = ('connector_id', 'name', 'url', 'server_uri')
 
-    Collects all read tags from scouter-type pipelines and organizes them by
-    server_id and tag_address, tracking which Kafka topics each tag is associated with.
-    This function is used during slot configuration to aggregate tags across multiple
-    scouter pipelines for efficient OPC server slot allocation.
+
+def _complete(document: dict[str, Any], fields: tuple[str, ...]) -> bool:
+    return all(document.get(field) not in (None, '') for field in fields)
+
+
+def _tag_id(tag: dict[str, Any]) -> str | None:
+    """The read tag's tag_id as text, or None when it has none (missing, null or empty)."""
+    tag_id = tag.get('tag_id')
+    return None if tag_id in (None, '') else str(tag_id)
+
+
+def referenced_tag_ids(pipelines: list[dict[str, Any]]) -> list[Any]:
+    """
+    The tag_ids the scouter pipelines reference, for the connector_tags query.
+
+    Each id goes in both forms, int and text, because the tag database stores int and
+    pipelines may hold either. Sorted, so the query is the same on every run.
 
     Args:
-        pipelines (list[dict[str, Any]]): List of pipeline configurations to process.
-            Only pipelines with workflow_type 'scouter' are processed. Each scouter
-            pipeline should contain a 'read_tags' list with tag configurations.
+        pipelines (list[dict[str, Any]]): Pipeline configurations
 
     Returns:
-        dict[str, Any]: Dictionary of read tags keyed by "server_id:tag_address",
-            where each entry contains:
-            - All original tag configuration fields
-            - topics (list[str]): List of Kafka topic names associated with this tag
-                (format: 'raw_{schedule_name}')
+        list[Any]: The referenced tag_ids (empty when no read tag carries a tag_id)
     """
+    ids: set[Any] = set()
+    for pipeline in pipelines:
+        if pipeline.get('workflow_type') != 'scouter':
+            continue
+        for tag in pipeline.get('read_tags') or []:
+            tag_id = _tag_id(tag)
+            if tag_id is None:
+                continue
+            ids.add(tag_id)
+            if tag_id.lstrip('-').isdigit():
+                ids.add(int(tag_id))
+    return sorted(ids, key=lambda value: (isinstance(value, str), str(value)))
 
-    tags = {}
 
-    # Get all read tags from pipelines
+def filter_by_tag_ids(query: dict[str, Any], tag_ids: list[Any]) -> dict[str, Any]:
+    """
+    Restrict a connector_tags aggregation to the referenced tag_ids.
+
+    Args:
+        query (dict[str, Any]): The connector_tags query ({'collection', 'aggregation'})
+        tag_ids (list[Any]): The ids from referenced_tag_ids
+
+    Returns:
+        dict[str, Any]: A new query whose aggregation starts with a tag_id $match. The
+            aggregation list is new too: aggregate_documents_in_mongodb appends a stage
+            to the list it gets.
+    """
+    return {
+        **query,
+        'aggregation': [{'$match': {'tag_id': {'$in': tag_ids}}}, *query['aggregation']],
+    }
+
+
+def index_connector_tags(connector_tags: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
+    """
+    Index the tag database's connector_tags by tag_id.
+
+    The key is the tag_id as text, because pipelines may store it as int or str.
+    Documents without tag_id, connector_id, address or tag_name are left out.
+
+    Args:
+        connector_tags (list[dict[str, Any]]): connector_tags documents (SIENTIAPDE-1569),
+            each with tag_id, connector_id, address and tag_name
+
+    Returns:
+        dict[str, dict[str, Any]]: connector_tags documents keyed by str(tag_id)
+    """
+    return {
+        str(tag['tag_id']): tag for tag in connector_tags if _complete(tag, CONNECTOR_TAG_FIELDS)
+    }
+
+
+def index_connectors(connectors: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
+    """
+    Index the tag database's connectors by connector_id (as text).
+
+    Documents without connector_id, name, url or server_uri are left out.
+
+    Args:
+        connectors (list[dict[str, Any]]): connectors documents (SIENTIAPDE-1569)
+
+    Returns:
+        dict[str, dict[str, Any]]: connectors documents keyed by str(connector_id)
+    """
+    return {
+        str(connector['connector_id']): connector
+        for connector in connectors
+        if _complete(connector, CONNECTOR_FIELDS)
+    }
+
+
+def resolve_tag_names(
+    read_tags: list[dict[str, Any]], tags_by_id: dict[str, dict[str, Any]]
+) -> list[dict[str, Any]]:
+    """
+    Fill in the tag_name of each read tag of a scouter pipeline.
+
+    The name is the pipeline's own tag_name when it has one, else the connector_tags
+    name of its tag_id. The scouter's model_tags (scouter()) and the slot's names[]
+    (gather_read_tags()) both take the name from here, so the name the gateway writes
+    is always the one the scouter looks for (SIENTIAPDE-1974, SC1).
+
+    Args:
+        read_tags (list[dict[str, Any]]): The pipeline's read_tags
+        tags_by_id (dict[str, dict[str, Any]]): connector_tags indexed by index_connector_tags
+
+    Returns:
+        list[dict[str, Any]]: The read tags, each with tag_name set when it could be
+            resolved. A tag_id missing from tags_by_id leaves the tag without a name.
+    """
+    resolved = []
+    for tag in read_tags:
+        name = tag.get('tag_name')
+        tag_id = _tag_id(tag)
+        if not name and tag_id is not None:
+            name = tags_by_id.get(tag_id, {}).get('tag_name')
+        resolved.append({**tag, 'tag_name': name} if name else tag)
+    return resolved
+
+
+def _connector_server(connector: dict[str, Any]) -> dict[str, Any]:
+    """Slot server fields of a tag database connector (connectors collection)."""
+    return {
+        'server_id': str(connector['connector_id']),
+        'name': connector['name'],
+        'url': connector['url'],
+        'server_uri': connector['server_uri'],
+        'use_security': bool(connector.get('use_security', False)),
+        'subscription_period_ms': connector.get('subscription_period_ms'),
+        'from_tag_database': True,
+    }
+
+
+def _opc_server(server_id: str, opc_server: dict[str, Any]) -> dict[str, Any]:
+    """
+    Slot server fields of a legacy opc-servers document.
+
+    The certificate now always comes from MinIO on the gateway side, so the file paths
+    are not carried: a cert_path only tells that the server uses security (RF4). The
+    subscription period is left to build_tag_config (half the lowest tag frequency).
+    """
+    return {
+        'server_id': server_id,
+        'name': opc_server['server_name'],
+        'url': opc_server['url'],
+        'server_uri': opc_server['uri'],
+        'use_security': bool(opc_server.get('cert_path')),
+        'subscription_period_ms': None,
+        'from_tag_database': False,
+    }
+
+
+def _frequency(tag: dict[str, Any]) -> int | None:
+    """The read tag's frequency as int, or None when it is missing or not an integer."""
+    try:
+        return int(tag['frequency'])
+    except (KeyError, TypeError, ValueError):
+        return None
+
+
+def _locate_read_tag(
+    tag: dict[str, Any],
+    opc_servers: dict[str, dict[str, Any]],
+    connectors: dict[str, dict[str, Any]],
+    tags_by_id: dict[str, dict[str, Any]],
+    issues: dict[str, set[str]],
+) -> tuple[dict[str, Any], str] | None:
+    """
+    Server and address of a read tag, or None when they cannot be resolved.
+
+    A tag with tag_id takes its address and connector from connector_tags and its server
+    from connectors (RF3). A tag without one keeps the current fields: server_id +
+    tag_address, with the server from opc-servers (RF4). What cannot be resolved is
+    recorded in issues ('tags', 'connectors' or 'servers').
+    """
+    tag_id = _tag_id(tag)
+    if tag_id is not None:
+        connector_tag = tags_by_id.get(tag_id)
+        if connector_tag is None:
+            issues['tags'].add(tag_id)
+            return None
+        connector_id = str(connector_tag['connector_id'])
+        connector = connectors.get(connector_id)
+        if connector is None:
+            issues['connectors'].add(connector_id)
+            return None
+        return _connector_server(connector), connector_tag['address']
+
+    server_id = str(tag['server_id'])
+    opc_server = opc_servers.get(server_id)
+    if opc_server is None:
+        issues['servers'].add(server_id)
+        return None
+    return _opc_server(server_id, opc_server), tag['tag_address']
+
+
+def _invalid_read_tag(tag: dict[str, Any]) -> bool:
+    """
+    A read tag the slot cannot use: no integer frequency, or (without tag_id) no
+    server_id, tag_address or tag_name.
+    """
+    if _frequency(tag) is None:
+        return True
+    if _tag_id(tag) is not None:
+        return False
+    return not _complete(tag, ('server_id', 'tag_address', 'tag_name'))
+
+
+def _merge_active_read(current: dict[str, Any], requested: dict[str, Any] | None) -> dict[str, Any]:
+    """Active read of an address: on if any pipeline asks, with the lowest cycle_multiple (D3)."""
+    if not requested or not requested.get('enabled'):
+        return current
+    cycle_multiple = int(requested.get('cycle_multiple', 1))
+    if current.get('enabled'):
+        cycle_multiple = min(cycle_multiple, current['cycle_multiple'])
+    return {'enabled': True, 'cycle_multiple': cycle_multiple}
+
+
+def _assign_server_keys(servers: dict[str, dict[str, Any]]) -> None:
+    """
+    Give each server a unique slot key: its name, or name:server_id (extended until
+    free) when another server already took it. Assigned in server_id order, so the keys
+    are the same on every run and in every slot.
+    """
+    taken: set[str] = set()
+    for server_id in sorted(servers):
+        server = servers[server_id]
+        key = server['name']
+        while key in taken:
+            key = f'{key}:{server_id}'
+        server['key'] = key
+        taken.add(key)
+
+
+def _add_read_tag(
+    entries: dict[tuple[str, str], dict[str, Any]],
+    servers: dict[str, dict[str, Any]],
+    pipeline: dict[str, Any],
+    tag: dict[str, Any],
+    server: dict[str, Any],
+    address: str,
+) -> None:
+    """Merge one resolved read tag into its (server, address) entry and its server."""
+    server_id = server['server_id']
+    known = servers.get(server_id)
+    if known is None or (server['from_tag_database'] and not known['from_tag_database']):
+        servers[server_id] = server
+
+    # int, so the slot does not depend on how each pipeline stored the number
+    frequency = cast(int, _frequency(tag))
+    entry = entries.setdefault(
+        (server_id, address),
+        {
+            'address': address,
+            'frequency': frequency,
+            'active_read': {'enabled': False},
+            'names': [],
+        },
+    )
+    entry['frequency'] = min(entry['frequency'], frequency)
+    entry['active_read'] = _merge_active_read(entry['active_read'], tag.get('active_read'))
+
+    name = {
+        'collection': f'raw_{pipeline["schedule_name"]}',
+        'tag_name': tag['tag_name'],
+        'model_id': pipeline['model_id'],
+        'model_name': pipeline['model']['name'],
+    }
+    if name not in entry['names']:
+        entry['names'].append(name)
+
+
+def gather_read_tags(
+    pipelines: list[dict[str, Any]],
+    opc_servers: dict[str, dict[str, Any]],
+    connectors: dict[str, dict[str, Any]],
+    tags_by_id: dict[str, dict[str, Any]],
+) -> tuple[list[dict[str, Any]], dict[str, set[str]]]:
+    """
+    Group the read tags of all scouter pipelines by (server, address) for the slot v2.
+
+    An address read by N pipelines becomes ONE entry with N items in names[]: the
+    gateway reads it once and writes it to N collections (SIENTIAPDE-1974, RF2). Each
+    read tag is resolved through the tag database when it has a tag_id (RF3), or through
+    the current server_id + tag_address + opc-servers fields when it has not (RF4).
+
+    A server known through both connectors and opc-servers under the same server_id is
+    the same server (the gateway keys its connections by server_id); the connectors
+    fields win.
+
+    Args:
+        pipelines (list[dict[str, Any]]): Pipeline configurations. Only workflow_type
+            'scouter' is read; each needs schedule_name, model_id, model.name and
+            read_tags (with frequency, and optionally tag_id and active_read)
+        opc_servers (dict[str, dict[str, Any]]): opc-servers documents keyed by str(id)
+        connectors (dict[str, dict[str, Any]]): Active OPC UA connectors indexed by
+            index_connectors
+        tags_by_id (dict[str, dict[str, Any]]): Active Read connector_tags indexed by
+            index_connector_tags
+
+    Returns:
+        tuple[list[dict[str, Any]], dict[str, set[str]]]:
+            - One entry per (server, address), sorted by server_id then address, with:
+                - server (dict): server_id, name, url, server_uri, use_security,
+                  subscription_period_ms (None = derive from the frequencies) and key
+                  (the server's unique slot key)
+                - address (str): The OPC node address
+                - frequency (int): The lowest frequency among the pipelines (RF6)
+                - active_read (dict): {'enabled': False} unless a pipeline enables it (D3)
+                - names (list[dict]): collection ('raw_{schedule_name}'), tag_name,
+                  model_id and model_name, one per pipeline name, sorted
+            - What was left out: 'tags' (tag_id), 'connectors' (connector_id), 'servers'
+              (opc-servers server_id) and 'invalid' (schedule_name/read tag without a
+              usable frequency or legacy fields)
+    """
+    issues: dict[str, set[str]] = {
+        'tags': set(),
+        'connectors': set(),
+        'servers': set(),
+        'invalid': set(),
+    }
+    servers: dict[str, dict[str, Any]] = {}
+    entries: dict[tuple[str, str], dict[str, Any]] = {}
+
     for pipeline in pipelines:
         if pipeline['workflow_type'] != 'scouter':
             continue
 
-        for tag in pipeline.get('read_tags', []):
-            tag_string = f'{tag["server_id"]}:{tag["tag_address"]}'
-            if tag_string not in tags:
-                tags[tag_string] = {**tag, 'topics': []}
+        for tag in resolve_tag_names(pipeline.get('read_tags') or [], tags_by_id):
+            if _invalid_read_tag(tag):
+                ref = _tag_id(tag) or tag.get('tag_address') or tag.get('tag_name')
+                issues['invalid'].add(f'{pipeline["schedule_name"]}/{ref}')
+                continue
 
-            tags[tag_string]['topics'].append(f'raw_{pipeline["schedule_name"]}')
+            located = _locate_read_tag(tag, opc_servers, connectors, tags_by_id, issues)
+            if located is None:
+                continue
+            _add_read_tag(entries, servers, pipeline, tag, *located)
 
-    return tags
+    _assign_server_keys(servers)
+
+    gathered = []
+    for (server_id, _address), entry in sorted(entries.items()):
+        # A stable order keeps the slot identical between runs, so the gateway does
+        # not re-subscribe an unchanged slot (O6)
+        entry['names'].sort(key=lambda n: (n['collection'], n['tag_name'], str(n['model_id'])))
+        gathered.append({'server': servers[server_id], **entry})
+
+    return gathered, issues
 
 
-def build_tag_config(
-    tags: list[dict[str, Any]], opc_servers: dict[str, Any]
-) -> tuple[dict[str, Any], list]:
+def build_tag_config(entries: list[dict[str, Any]]) -> dict[str, Any]:
     """
-    Build tag configuration for a specific slot and OPC server.
+    Build one slot in the slot v2 format (contract C1) from address entries.
 
-    Organizes tags by OPC server name and calculates the minimum subscription period
-    based on tag frequencies. Validates that all server IDs exist in the OPC
-    servers configuration. The subscription period is set to half of the minimum
-    tag frequency to ensure efficient data collection.
+    The slot no longer carries cert_path, private_key_path or server_cert_path: the
+    gateway takes the certificate from MinIO (SIENTIAPDE-1974, O4).
 
     Args:
-        tags (list[dict[str, Any]]): List of tag configurations containing:
-            - server_id (str): ID of the OPC server
-            - tag_address (str): Address/path of the OPC tag
-            - frequency (int): Tag read frequency in milliseconds
-            - Additional tag-specific configuration fields
-        opc_servers (dict[str, Any]): Dictionary of OPC server configurations keyed by server_id.
-            Each server configuration should contain:
-            - server_name (str): Human-readable server name
-            - url (str): OPC server URL
-            - uri (str): OPC server URI
-            - cert_path (str, optional): Certificate file path
-            - private_key_path (str, optional): Private key file path
-            - server_cert_path (str, optional): Server certificate file path
+        entries (list[dict[str, Any]]): Address entries from gather_read_tags for this slot
 
     Returns:
-        tuple[dict[str, Any], list]: A tuple containing:
-            - Slot configuration dictionary organized by server_name, where each server
-              contains connection details, tags dictionary, and subscription_period_ms
-            - List of server IDs (str) that were not found in opc_servers configuration
+        dict[str, Any]: The slot keyed by server key, each server with server_id, name,
+            url, server_uri, use_security, subscription_period_ms and tags. Each tag (keyed
+            by address) has frequency, active_read and names. A server without a
+            subscription period of its own (opc-servers) gets half its lowest tag
+            frequency in this slot, as before.
     """
+    slot_config: dict[str, Any] = {}
 
-    slot_config = {}
-    notifications = []
-
-    for tag in tags:
-        server_id = tag['server_id']
-
-        if server_id not in opc_servers:
-            notifications.append(server_id)
-            continue
-
-        server_name = opc_servers[server_id]['server_name']
-        if server_name not in slot_config:
-            slot_config[server_name] = {
-                'server_id': server_id,
-                'name': server_name,
-                'url': opc_servers[server_id]['url'],
-                'server_uri': opc_servers[server_id]['uri'],
-                'cert_path': opc_servers[server_id].get('cert_path', None),
-                'private_key_path': opc_servers[server_id].get('private_key_path', None),
-                'server_cert_path': opc_servers[server_id].get('server_cert_path', None),
+    for entry in entries:
+        server = entry['server']
+        if server['key'] not in slot_config:
+            slot_config[server['key']] = {
+                'server_id': server['server_id'],
+                'name': server['name'],
+                'url': server['url'],
+                'server_uri': server['server_uri'],
+                'use_security': server['use_security'],
+                'subscription_period_ms': server['subscription_period_ms'],
                 'tags': {},
             }
 
-        slot_config[server_name]['tags'][tag['tag_address']] = {
-            **tag,
+        slot_config[server['key']]['tags'][entry['address']] = {
+            'frequency': entry['frequency'],
+            'active_read': entry['active_read'],
+            'names': entry['names'],
         }
 
-    for server_name in slot_config:
-        frequencies = [int(x['frequency']) for x in slot_config[server_name]['tags'].values()]
+    for server_config in slot_config.values():
+        if server_config['subscription_period_ms'] is None:
+            frequencies = [tag['frequency'] for tag in server_config['tags'].values()]
+            server_config['subscription_period_ms'] = min(frequencies) / 2
 
-        min_frequency = min(frequencies) if frequencies else 1000
-
-        slot_config[server_name]['subscription_period_ms'] = min_frequency / 2
-
-    return slot_config, notifications
+    return slot_config

orchestrator/workflows/orchestrator.py

@@ -7,6 +7,39 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.temporal.policies import retry_policy
 
     from orchestrator.activities.activities import Activities
+    from orchestrator.utils.orchestrator_functions import filter_by_tag_ids, referenced_tag_ids
+
+# Tag database (SIENTIAPDE-1569) read by the slot v2 (SIENTIAPDE-1974). The workflow
+# input may override these aggregations (connectors_query / connector_tags_query); the
+# workflow puts a tag_id $match in front of connector_tags_query. Only the fields the
+# slot uses are projected: other attributes may hold BSON types a Temporal payload
+# cannot carry.
+DEFAULT_CONNECTORS_QUERY = {
+    'collection': 'connectors',
+    'aggregation': [
+        {'$match': {'type': 'opc_ua', 'status': 'Active'}},
+        {
+            '$project': {
+                '_id': 0,
+                'connector_id': 1,
+                'name': 1,
+                'url': 1,
+                'server_uri': 1,
+                'use_security': 1,
+                'subscription_period_ms': 1,
+            }
+        },
+    ],
+}
+DEFAULT_CONNECTOR_TAGS_QUERY = {
+    'collection': 'connector_tags',
+    'aggregation': [
+        {'$match': {'category': 'Read', 'status': 'Active'}},
+        {'$project': {'_id': 0, 'tag_id': 1, 'connector_id': 1, 'address': 1, 'tag_name': 1}},
+    ],
+}
+# Runs started before the tag database loads existed replay without them
+TAG_DATABASE_PATCH = 'SIENTIAPDE-1974-tag-database'
 
 
 @workflow.defn(name='orchestrator')
@@ -20,6 +53,57 @@ class Orchestrator:
     and Redis infrastructure.
     """
 
+    async def _load_tag_database(
+        self,
+        input_data: dict[str, Any],
+        metadata: dict[str, Any],
+        pipelines: list[dict[str, Any]],
+    ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+        """
+        Load the tag database (SIENTIAPDE-1974) for the read tags that carry a tag_id.
+
+        Only the referenced tags are loaded, with the fields the slot uses, so an
+        environment without a tag database never depends on its content (P4) and the
+        payload stays bounded by the pipelines. Nothing is loaded when no scouter read
+        tag has a tag_id, or when replaying a run started before TAG_DATABASE_PATCH.
+
+        Args:
+            input_data (dict[str, Any]): Workflow input (optional connectors_query and
+                connector_tags_query)
+            metadata (dict[str, Any]): Activity metadata ({'metadata': {...}})
+            pipelines (list[dict[str, Any]]): The loaded pipeline configurations
+
+        Returns:
+            tuple[list[dict[str, Any]], list[dict[str, Any]]]: connectors and connector_tags
+        """
+        tag_ids = referenced_tag_ids(pipelines)
+        # patched() first: it must run on every execution to record its marker
+        if not workflow.patched(TAG_DATABASE_PATCH) or not tag_ids:
+            return [], []
+
+        connectors_query = input_data.get('connectors_query', DEFAULT_CONNECTORS_QUERY)
+        connector_tags_query = input_data.get('connector_tags_query', DEFAULT_CONNECTOR_TAGS_QUERY)
+
+        connectors_handler = workflow.start_local_activity_method(
+            Activities.aggregate_documents_in_mongodb,
+            {
+                **metadata,
+                # A fresh aggregation list: the activity appends a stage to the one it gets
+                'query': {**connectors_query, 'aggregation': list(connectors_query['aggregation'])},
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        connector_tags_handler = workflow.start_local_activity_method(
+            Activities.aggregate_documents_in_mongodb,
+            {**metadata, 'query': filter_by_tag_ids(connector_tags_query, tag_ids)},
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        return await connectors_handler, await connector_tags_handler
+
     @workflow.run
     async def run(self, input_data: dict[str, Any]):
         """
@@ -36,6 +120,11 @@ class Orchestrator:
                 - schedule_name (str): Name of the orchestration schedule
                 - pipelines_query (dict[str, Any]): MongoDB query for pipeline configurations
                 - opc_servers_query (dict[str, Any]): MongoDB query for OPC server data
+                Optional fields (used only when a scouter read tag carries a tag_id):
+                - connectors_query (dict[str, Any]): MongoDB aggregation for the tag
+                  database connectors (default: DEFAULT_CONNECTORS_QUERY)
+                - connector_tags_query (dict[str, Any]): MongoDB aggregation for the tag
+                  database tags (default: DEFAULT_CONNECTOR_TAGS_QUERY)
 
         Returns:
             None: Workflow completes without return value
@@ -103,6 +192,11 @@ class Orchestrator:
         )
 
         pipeline_config = await pipeline_config_handler
+
+        connectors, connector_tags = await self._load_tag_database(
+            input_data, metadata, pipeline_config
+        )
+
         orchestrated_schedules = await orchestrated_schedules_handler
         current_slot_config = await current_slot_config_handler
         opc_servers = await opc_servers_handler
@@ -117,7 +211,7 @@ class Orchestrator:
 
         schedules_config_handler = workflow.start_local_activity_method(
             Activities.process_schedules,
-            {**metadata, 'pipelines': pipeline_config},
+            {**metadata, 'pipelines': pipeline_config, 'connector_tags': connector_tags},
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(seconds=60),
         )
@@ -127,6 +221,8 @@ class Orchestrator:
             {
                 **metadata,
                 'opc_servers': opc_servers,
+                'connectors': connectors,
+                'connector_tags': connector_tags,
                 'active_ingestors': active_ingestors,
                 'pipelines': pipeline_config,
             },

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,92 @@ 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.
+        - 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
+
+        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)
+
+        # `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)
+
+        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:
+            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.0.0"
+version = "0.6.7"
 description = "Sientia DataOps Orchestrator - ML Model Orchestration System"
 readme = "README.md"
 requires-python = ">=3.11"

requirements-local.txt

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

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

@@ -85,7 +85,9 @@ def test_close(sientia_monitoring_mock, email):
     email.close()
 
     email.server.quit.assert_called_once()
-    sientia_monitoring_mock.shutdown.assert_called_once()
+    # Count only this instance: an Email left by another test may be garbage-collected
+    # (Email.__del__ -> close) while the patch is active
+    assert sientia_monitoring_mock.shutdown.call_args_list.count(call(email)) == 1
 
 
 def test_build_email_html(email):

tests/orchestrator/activities/test_formatters.py

@@ -218,110 +218,395 @@ def test_process_schedules_with_invalid_workflow_type(formatters):
     mock_scouter.assert_any_call(pipelines[2])
 
 
-@patch(
-    'orchestrator.activities.formatters.gather_read_tags',
-    return_value={
-        '1:test_tag_address': {
-            'server_id': '1',
-            'server_name': 'test_server_name',
-            'tag_address': 'test_tag_address',
-            'topics': ['raw_test_schedule'],
-            'frequency': 1000,
-        },
-        '2:test_tag_address2': {
-            'server_id': '2',
-            'server_name': 'test_server_name2',
-            'tag_address': 'test_tag_address2',
-            'topics': ['raw_test_schedule2'],
-            'frequency': 1000,
-        },
-        '2:test_tag_address3': {
-            'server_id': '2',
-            'server_name': 'test_server_name2',
-            'tag_address': 'test_tag_address3',
-            'topics': ['raw_test_schedule2'],
-            'frequency': 1000,
-        },
-    },
-)
-@patch('orchestrator.activities.formatters.build_tag_config')
-def test_process_slots(mock_build_tag_config, mock_gather_read_tags, formatters):
+def _scouter_pipeline(schedule_name, model_id, read_tags):
+    return {
+        'workflow_type': 'scouter',
+        'schedule_name': schedule_name,
+        'model_id': model_id,
+        'model': {'name': f'model {model_id}'},
+        'read_tags': read_tags,
+    }
+
+
+slot_opc_servers = [
+    {'id': '1', 'server_name': 'kepserver', 'url': 'opc.tcp://kep:49320', 'uri': 'urn:kep'},
+]
+
+
+def test_process_slots(formatters):
+    """Slot v2 split by address: the names of an address never land in different slots"""
+    pipelines = [
+        _scouter_pipeline(
+            'sin',
+            'm1',
+            [
+                {'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000},
+                {'server_id': '1', 'tag_address': 'ns=2;s=B', 'tag_name': 'b', 'frequency': 1000},
+            ],
+        ),
+        _scouter_pipeline(
+            'cos',
+            'm2',
+            [
+                {'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'x', 'frequency': 500},
+                {'server_id': '1', 'tag_address': 'ns=2;s=C', 'tag_name': 'c', 'frequency': 4000},
+            ],
+        ),
+    ]
     input_data = {
-        'opc_servers': [
-            {'id': '1', 'server_name': 'test_server_name', 'url': 'test_url', 'uri': 'test_uri'},
-            {'id': '2', 'server_name': 'test_server_name2', 'url': 'test_url2', 'uri': 'test_uri2'},
-        ],
-        'active_ingestors': ['test_active_ingestor1', 'test_active_ingestor2'],
-        'pipelines': 'test_gather_read_tags',
+        'opc_servers': slot_opc_servers,
+        'active_ingestors': ['heartbeat:ingestor:1', 'heartbeat:ingestor:2'],
+        'pipelines': pipelines,
         **metadata,
     }
 
-    expected_opc_servers = {
+    result = formatters.process_slots(input_data)
+
+    server = {
+        'server_id': '1',
+        'name': 'kepserver',
+        'url': 'opc.tcp://kep:49320',
+        'server_uri': 'urn:kep',
+        'use_security': False,
+    }
+    assert result == {
         '1': {
-            'id': '1',
-            'server_name': 'test_server_name',
-            'url': 'test_url',
-            'uri': 'test_uri',
+            'kepserver': {
+                **server,
+                'subscription_period_ms': 250,
+                'tags': {
+                    'ns=2;s=A': {
+                        'frequency': 500,
+                        'active_read': {'enabled': False},
+                        'names': [
+                            {
+                                'collection': 'raw_cos',
+                                'tag_name': 'x',
+                                'model_id': 'm2',
+                                'model_name': 'model m2',
+                            },
+                            {
+                                'collection': 'raw_sin',
+                                'tag_name': 'a',
+                                'model_id': 'm1',
+                                'model_name': 'model m1',
+                            },
+                        ],
+                    },
+                    'ns=2;s=B': {
+                        'frequency': 1000,
+                        'active_read': {'enabled': False},
+                        'names': [
+                            {
+                                'collection': 'raw_sin',
+                                'tag_name': 'b',
+                                'model_id': 'm1',
+                                'model_name': 'model m1',
+                            },
+                        ],
+                    },
+                },
+            }
         },
         '2': {
-            'id': '2',
-            'server_name': 'test_server_name2',
-            'url': 'test_url2',
-            'uri': 'test_uri2',
+            'kepserver': {
+                **server,
+                'subscription_period_ms': 2000,
+                'tags': {
+                    'ns=2;s=C': {
+                        'frequency': 4000,
+                        'active_read': {'enabled': False},
+                        'names': [
+                            {
+                                'collection': 'raw_cos',
+                                'tag_name': 'c',
+                                'model_id': 'm2',
+                                'model_name': 'model m2',
+                            },
+                        ],
+                    },
+                },
+            }
         },
     }
+    formatters.send_notification.assert_not_called()
 
-    slot_mock = {
-        'test_server_name': {
-            'server_id': '1',
-            'name': 'test_server_name',
-            'url': 'test_url',
-            'server_uri': 'test_uri',
-            'cert_path': None,
-            'private_key_path': None,
-            'server_cert_path': None,
-        }
+
+def test_process_slots_without_ingestors_builds_one_slot(formatters):
+    pipelines = [
+        _scouter_pipeline(
+            'sin',
+            'm1',
+            [{'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000}],
+        )
+    ]
+    input_data = {
+        'opc_servers': slot_opc_servers,
+        'active_ingestors': [],
+        'pipelines': pipelines,
+        **metadata,
     }
 
-    mock_build_tag_config.return_value = (slot_mock, ['2'])
-
     result = formatters.process_slots(input_data)
 
-    tags = list(mock_gather_read_tags.return_value.values())
+    assert list(result) == ['1']
+    assert list(result['1']['kepserver']['tags']) == ['ns=2;s=A']
 
-    mock_gather_read_tags.assert_called_once_with(input_data['pipelines'])
 
-    mock_build_tag_config.assert_has_calls(
-        [
-            call(tags[:2], expected_opc_servers),
-            call(tags[2:], expected_opc_servers),
-        ]
-    )
+def test_process_slots_more_ingestors_than_addresses(formatters):
+    """Extra ingestors get empty slots, as before"""
+    pipelines = [
+        _scouter_pipeline(
+            'sin',
+            'm1',
+            [{'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000}],
+        )
+    ]
+    input_data = {
+        'opc_servers': slot_opc_servers,
+        'active_ingestors': ['i1', 'i2', 'i3'],
+        'pipelines': pipelines,
+        **metadata,
+    }
 
+    result = formatters.process_slots(input_data)
+
+    assert list(result) == ['1', '2', '3']
+    assert list(result['1']['kepserver']['tags']) == ['ns=2;s=A']
+    assert result['2'] == {}
+    assert result['3'] == {}
+
+
+def test_process_slots_with_tag_database(formatters):
+    """A read tag with tag_id takes address, name and server from the tag database"""
+    pipelines = [
+        _scouter_pipeline('sin', 'm1', [{'tag_id': 7, 'frequency': 1000}]),
+    ]
+    input_data = {
+        'opc_servers': slot_opc_servers,
+        'connectors': [
+            {
+                'connector_id': 3,
+                'type': 'opc_ua',
+                'name': 'plant',
+                'url': 'opc.tcp://plant:4840',
+                'server_uri': 'urn:plant',
+                'use_security': True,
+                'subscription_period_ms': 1500,
+                'status': 'Active',
+            }
+        ],
+        'connector_tags': [
+            {'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1001', 'tag_name': 'Pressure'}
+        ],
+        'active_ingestors': ['i1'],
+        'pipelines': pipelines,
+        **metadata,
+    }
+
+    result = formatters.process_slots(input_data)
+
+    assert result == {
+        '1': {
+            'plant': {
+                'server_id': '3',
+                'name': 'plant',
+                'url': 'opc.tcp://plant:4840',
+                'server_uri': 'urn:plant',
+                'use_security': True,
+                'subscription_period_ms': 1500,
+                'tags': {
+                    'ns=3;i=1001': {
+                        'frequency': 1000,
+                        'active_read': {'enabled': False},
+                        'names': [
+                            {
+                                'collection': 'raw_sin',
+                                'tag_name': 'Pressure',
+                                'model_id': 'm1',
+                                'model_name': 'model m1',
+                            }
+                        ],
+                    }
+                },
+            }
+        }
+    }
+
+
+def test_process_slots_notifies_unresolved_tags_once(formatters):
+    """Each kind of unresolved read tag is notified once per run, not once per slot"""
+    pipelines = [
+        _scouter_pipeline(
+            'sin',
+            'm1',
+            [
+                {'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000},
+                {'server_id': '9', 'tag_address': 'ns=2;s=B', 'tag_name': 'b', 'frequency': 1000},
+                {'tag_id': 404, 'frequency': 1000},
+                {'tag_id': 8, 'frequency': 1000},
+                {'server_id': '1', 'tag_address': 'ns=2;s=C', 'tag_name': 'c'},
+            ],
+        ),
+    ]
+    input_data = {
+        'opc_servers': slot_opc_servers,
+        'connectors': [],
+        'connector_tags': [
+            {'tag_id': 8, 'connector_id': 5, 'address': 'ns=3;i=8', 'tag_name': 'Level'}
+        ],
+        'active_ingestors': ['i1', 'i2'],
+        'pipelines': pipelines,
+        **metadata,
+    }
+
+    result = formatters.process_slots(input_data)
+
+    assert list(result['1']['kepserver']['tags']) == ['ns=2;s=A']
+    assert result['2'] == {}
     formatters.send_notification.assert_has_calls(
         [
             call(
                 metadata=metadata['metadata'],
                 notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
-                message='Servers 2 not found in opc_servers',
+                message='Servers 9 not found in opc_servers',
                 block='orchestrator',
                 level=NotificationLevel.ERROR,
             ),
             call(
                 metadata=metadata['metadata'],
-                notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
-                message='Servers 2 not found in opc_servers',
+                notification_id='ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION',
+                message='Tags 404 not found among the active, complete Read tags of connector_tags',
+                block='orchestrator',
+                level=NotificationLevel.ERROR,
+            ),
+            call(
+                metadata=metadata['metadata'],
+                notification_id='ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION',
+                message='Connectors 5 not found, not active or incomplete in connectors',
+                block='orchestrator',
+                level=NotificationLevel.ERROR,
+            ),
+            call(
+                metadata=metadata['metadata'],
+                notification_id='ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION',
+                message=(
+                    'Read tags sin/ns=2;s=C have no integer frequency '
+                    'or lack server_id, tag_address or tag_name'
+                ),
                 block='orchestrator',
                 level=NotificationLevel.ERROR,
             ),
         ]
     )
+    assert formatters.send_notification.call_count == 4
 
-    assert result == {
-        '1': slot_mock,
-        '2': slot_mock,
+
+def test_process_slots_survives_malformed_tag_database(formatters):
+    """An incomplete connector or tag leaves only its own read tags out of the slot"""
+    pipelines = [
+        _scouter_pipeline(
+            'sin',
+            'm1',
+            [
+                {'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000},
+                {'tag_id': 7, 'frequency': 1000},
+                {'tag_id': 8, 'frequency': 1000},
+            ],
+        ),
+    ]
+    input_data = {
+        'opc_servers': slot_opc_servers,
+        'connectors': [{'connector_id': 3, 'name': 'plant', 'server_uri': 'urn:plant'}],
+        'connector_tags': [
+            {'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1', 'tag_name': 'P'},
+            {'tag_id': 8, 'connector_id': 3, 'tag_name': 'no address'},
+        ],
+        'active_ingestors': [],
+        'pipelines': pipelines,
+        **metadata,
     }
 
+    result = formatters.process_slots(input_data)
+
+    assert list(result['1']) == ['kepserver']
+    notified = {c.kwargs['notification_id'] for c in formatters.send_notification.call_args_list}
+    assert notified == {
+        'ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION',
+        'ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION',
+    }
+
+
+def test_process_slots_is_stable_across_pipeline_order(formatters):
+    """The same pipelines in another order give the same slots (O6)"""
+    pipelines = [
+        _scouter_pipeline(
+            'sin',
+            'm1',
+            [{'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000}],
+        ),
+        _scouter_pipeline(
+            'cos',
+            'm2',
+            [
+                {'server_id': '1', 'tag_address': 'ns=2;s=B', 'tag_name': 'b', 'frequency': 1000},
+                {'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'x', 'frequency': 1000},
+            ],
+        ),
+    ]
+
+    def run(order):
+        return formatters.process_slots(
+            {
+                'opc_servers': slot_opc_servers,
+                'active_ingestors': ['i1', 'i2'],
+                'pipelines': order,
+                **metadata,
+            }
+        )
+
+    assert json.dumps(run(pipelines)) == json.dumps(run(list(reversed(pipelines))))
+
+
+def test_process_schedules_names_scouter_tags_like_the_slot(formatters):
+    """model_tags and the slot's names[] take the tag name from the same source (SC1)"""
+    pipeline = {
+        **_scouter_pipeline('sin', 'm1', [{'tag_id': 7, 'frequency': 1000, 'aggr_func': 'avg'}]),
+        'updated_at': '2021-01-01',
+    }
+    connector_tags = [
+        {'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1001', 'tag_name': 'Pressure'}
+    ]
+    connectors = [
+        {
+            'connector_id': 3,
+            'name': 'plant',
+            'url': 'opc.tcp://plant:4840',
+            'server_uri': 'urn:plant',
+        }
+    ]
+
+    schedules = formatters.process_schedules(
+        {'pipelines': [pipeline], 'connector_tags': connector_tags, **metadata}
+    )
+    slots = formatters.process_slots(
+        {
+            'opc_servers': [],
+            'connectors': connectors,
+            'connector_tags': connector_tags,
+            'active_ingestors': [],
+            'pipelines': [pipeline],
+            **metadata,
+        }
+    )
+
+    model_tags = schedules['scouter']['sin']['model_tags']
+    names = slots['1']['plant']['tags']['ns=3;i=1001']['names']
+    assert list(model_tags) == ['Pressure']
+    assert model_tags['Pressure']['aggr_func'] == 'avg'
+    assert [name['tag_name'] for name in names] == ['Pressure']
+    assert names[0]['collection'] == schedules['scouter']['sin']['topic']
+
 
 def test_format_schedule_config(formatters):
     input_data = {
@@ -410,6 +695,31 @@ def test_create_slot_config(formatters):
     }
 
 
+def test_create_slot_config_skips_unchanged_slots(formatters):
+    """Only new or changed slots are written (O6); current slots come keyed by Redis key"""
+    slot_v2 = {'kepserver': {'server_id': '1', 'tags': {'ns=2;s=A': {'names': []}}}}
+    slot_v1 = {'kepserver': {'server_id': '1', 'tags': {'ns=2;s=A': {'topics': ['raw_x']}}}}
+    input_data = {
+        'current_slot_config': {
+            'slot:opc_tags:1': slot_v2,
+            'slot:opc_tags:2': slot_v1,
+        },
+        'slot_config': {'1': slot_v2, '2': slot_v2, '3': {}},
+    }
+
+    result = formatters.create_slot_config(input_data)
+
+    assert result == {'to_delete': [], 'to_insert': {'2': slot_v2, '3': {}}}
+
+
+def test_create_slot_config_without_current_slots(formatters):
+    slot_config = {'1': {'kepserver': {'tags': {}}}, '2': {}}
+
+    result = formatters.create_slot_config({'current_slot_config': {}, 'slot_config': slot_config})
+
+    assert result == {'to_delete': [], 'to_insert': slot_config}
+
+
 def test_send_success_report(formatters):
     formatters.send_success_report(
         metadata=metadata,
@@ -716,7 +1026,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 +1076,260 @@ 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_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,
+):
+    # 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

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

@@ -5,12 +5,17 @@ from orchestrator.utils.orchestrator_functions import (
     build_tag_config,
     common_config,
     drift,
+    filter_by_tag_ids,
     gather_read_tags,
+    index_connector_tags,
+    index_connectors,
     minimal_retrain,
     overlap_filter_config,
     pi_web_api_scouter,
     predictions_batch,
     process_path_priority,
+    referenced_tag_ids,
+    resolve_tag_names,
     scouter,
     simple_metrics,
 )
@@ -311,249 +316,562 @@ def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_confi
     assert result == expected
 
 
-def test_gather_read_tags():
-    pipelines = [
-        {
-            'workflow_type': 'scouter',
-            'schedule_name': 'test_schedule',
-            'read_tags': [
-                {
-                    'server_id': '1',
-                    'server_name': 'test_server_name',
-                    'tag_address': 'test_tag_address',
-                }
-            ],
-        },
-        {
-            'workflow_type': 'scouter',
-            'schedule_name': 'test_schedule2',
-            'read_tags': [
-                {
-                    'server_id': '2',
-                    'server_name': 'test_server_name2',
-                    'tag_address': 'test_tag_address2',
-                },
-                {
-                    'server_id': '2',
-                    'server_name': 'test_server_name2',
-                    'tag_address': 'test_tag_address3',
-                },
-            ],
-        },
-    ]
-
-    result = gather_read_tags(pipelines)
-
-    expected = {
-        '1:test_tag_address': {
-            'server_id': '1',
-            'server_name': 'test_server_name',
-            'tag_address': 'test_tag_address',
-            'topics': ['raw_test_schedule'],
-        },
-        '2:test_tag_address2': {
-            'server_id': '2',
-            'server_name': 'test_server_name2',
-            'tag_address': 'test_tag_address2',
-            'topics': ['raw_test_schedule2'],
-        },
-        '2:test_tag_address3': {
-            'server_id': '2',
-            'server_name': 'test_server_name2',
-            'tag_address': 'test_tag_address3',
-            'topics': ['raw_test_schedule2'],
-        },
+def _pipeline(schedule_name, model_id, read_tags, workflow_type='scouter'):
+    return {
+        'workflow_type': workflow_type,
+        'schedule_name': schedule_name,
+        'model_id': model_id,
+        'model': {'name': f'model {model_id}'},
+        'read_tags': read_tags,
     }
 
-    assert result == expected
+
+def _legacy_tag(address, name, frequency=1000, server_id='1', **extra):
+    return {
+        'server_id': server_id,
+        'tag_address': address,
+        'tag_name': name,
+        'frequency': frequency,
+        **extra,
+    }
+
+
+def _name(schedule_name, tag_name, model_id):
+    return {
+        'collection': f'raw_{schedule_name}',
+        'tag_name': tag_name,
+        'model_id': model_id,
+        'model_name': f'model {model_id}',
+    }
+
+
+opc_servers = {
+    '1': {
+        'id': '1',
+        'server_name': 'kepserver',
+        'url': 'opc.tcp://kep:49320',
+        'uri': 'urn:kep',
+        'cert_path': 'certs/client.pem',
+    },
+    '2': {'id': '2', 'server_name': 'simulator', 'url': 'opc.tcp://sim:4840', 'uri': 'urn:sim'},
+}
+
+connectors = {
+    '3': {
+        'connector_id': 3,
+        'name': 'plant',
+        'url': 'opc.tcp://plant:4840',
+        'server_uri': 'urn:plant',
+        'use_security': True,
+        'subscription_period_ms': 1500,
+    }
+}
+
+tags_by_id = {
+    '7': {'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1001', 'tag_name': 'Pressure'},
+    '8': {'tag_id': 8, 'connector_id': 5, 'address': 'ns=3;i=1002', 'tag_name': 'Level'},
+}
+
+
+def _gather(pipelines, servers=None, connector_index=None, tag_index=None):
+    return gather_read_tags(
+        pipelines,
+        opc_servers if servers is None else servers,
+        connectors if connector_index is None else connector_index,
+        tags_by_id if tag_index is None else tag_index,
+    )
+
+
+def test_index_connector_tags():
+    tags = [
+        {'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1', 'tag_name': 'a'},
+        {'tag_id': '8', 'connector_id': 3, 'address': 'ns=3;i=2', 'tag_name': 'b'},
+    ]
+
+    assert index_connector_tags(tags) == {'7': tags[0], '8': tags[1]}
+
+
+def test_resolve_tag_names():
+    read_tags = [
+        {'tag_id': 7},
+        {'tag_id': '7', 'tag_name': 'own name'},
+        {'tag_id': 404},
+        {'tag_name': 'legacy'},
+    ]
+
+    assert resolve_tag_names(read_tags, tags_by_id) == [
+        {'tag_id': 7, 'tag_name': 'Pressure'},
+        {'tag_id': '7', 'tag_name': 'own name'},
+        {'tag_id': 404},
+        {'tag_name': 'legacy'},
+    ]
+
+
+def test_gather_read_tags_multi_name():
+    """One address read by two pipelines with different names: one entry, two names"""
+    pipelines = [
+        _pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'Counter', frequency=15000)]),
+        _pipeline('cos', 'm2', [_legacy_tag('ns=2;s=A', 'Contador', frequency=5000)]),
+    ]
+
+    entries, issues = _gather(pipelines)
+
+    assert len(entries) == 1
+    assert entries[0]['address'] == 'ns=2;s=A'
+    assert entries[0]['frequency'] == 5000
+    assert entries[0]['names'] == [_name('cos', 'Contador', 'm2'), _name('sin', 'Counter', 'm1')]
+    assert issues == {'tags': set(), 'connectors': set(), 'servers': set(), 'invalid': set()}
+
+
+def test_gather_read_tags_same_name_in_two_pipelines():
+    """The same name read by two pipelines still goes to both collections"""
+    pipelines = [
+        _pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'Counter')]),
+        _pipeline('cos', 'm2', [_legacy_tag('ns=2;s=A', 'Counter')]),
+    ]
+
+    entries, _issues = _gather(pipelines)
+
+    assert entries[0]['names'] == [_name('cos', 'Counter', 'm2'), _name('sin', 'Counter', 'm1')]
+
+
+def test_gather_read_tags_duplicate_tag_in_one_pipeline():
+    pipelines = [
+        _pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a'), _legacy_tag('ns=2;s=A', 'a')]),
+    ]
+
+    entries, _issues = _gather(pipelines)
+
+    assert entries[0]['names'] == [_name('sin', 'a', 'm1')]
+
+
+def test_gather_read_tags_same_address_on_two_servers():
+    pipelines = [
+        _pipeline(
+            'sin',
+            'm1',
+            [_legacy_tag('ns=2;s=A', 'a', server_id='2'), _legacy_tag('ns=2;s=A', 'a')],
+        ),
+    ]
+
+    entries, _issues = _gather(pipelines)
+
+    assert [(e['server']['server_id'], e['address']) for e in entries] == [
+        ('1', 'ns=2;s=A'),
+        ('2', 'ns=2;s=A'),
+    ]
+
+
+def test_gather_read_tags_legacy_server():
+    """Without tag_id: opc-servers fields, no certificate paths, security from cert_path"""
+    entries, _issues = _gather([_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a')])])
+
+    assert entries[0]['server'] == {
+        'server_id': '1',
+        'name': 'kepserver',
+        'url': 'opc.tcp://kep:49320',
+        'server_uri': 'urn:kep',
+        'use_security': True,
+        'subscription_period_ms': None,
+        'from_tag_database': False,
+        'key': 'kepserver',
+    }
+
+
+def test_gather_read_tags_numeric_server_id():
+    """A numeric server_id in the pipeline still finds its opc-servers document"""
+    entries, issues = _gather([_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a', server_id=1)])])
+
+    assert entries[0]['server']['server_id'] == '1'
+    assert not issues['servers']
+
+
+def test_gather_read_tags_with_tag_database():
+    """With tag_id: address, name and server come from connector_tags + connectors"""
+    entries, _issues = _gather([_pipeline('sin', 'm1', [{'tag_id': 7, 'frequency': 1000}])])
+
+    assert entries == [
+        {
+            'server': {
+                'server_id': '3',
+                'name': 'plant',
+                'url': 'opc.tcp://plant:4840',
+                'server_uri': 'urn:plant',
+                'use_security': True,
+                'subscription_period_ms': 1500,
+                'from_tag_database': True,
+                'key': 'plant',
+            },
+            'address': 'ns=3;i=1001',
+            'frequency': 1000,
+            'active_read': {'enabled': False},
+            'names': [_name('sin', 'Pressure', 'm1')],
+        }
+    ]
+
+
+def test_gather_read_tags_pipeline_name_wins_over_tag_database():
+    entries, _issues = _gather(
+        [_pipeline('sin', 'm1', [{'tag_id': 7, 'tag_name': 'P1', 'frequency': 1000}])]
+    )
+
+    assert entries[0]['names'] == [_name('sin', 'P1', 'm1')]
+
+
+def test_gather_read_tags_unknown_tag_id():
+    entries, issues = _gather([_pipeline('sin', 'm1', [{'tag_id': 404, 'frequency': 1000}])])
+
+    assert entries == []
+    assert issues == {'tags': {'404'}, 'connectors': set(), 'servers': set(), 'invalid': set()}
+
+
+def test_gather_read_tags_inactive_connector():
+    """Connector 5 is not among the active connectors: its tag is left out"""
+    entries, issues = _gather([_pipeline('sin', 'm1', [{'tag_id': 8, 'frequency': 1000}])])
+
+    assert entries == []
+    assert issues == {'tags': set(), 'connectors': {'5'}, 'servers': set(), 'invalid': set()}
+
+
+def test_gather_read_tags_unknown_server():
+    entries, issues = _gather(
+        [_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a', server_id='9')])]
+    )
+
+    assert entries == []
+    assert issues == {'tags': set(), 'connectors': set(), 'servers': {'9'}, 'invalid': set()}
+
+
+def test_gather_read_tags_mixed_with_and_without_tag_id():
+    """Both paths in one run; the same node via tag_id and via tag_address merges"""
+    connector_1 = {
+        'connector_id': 1,
+        'name': 'kepserver',
+        'url': 'opc.tcp://kep:49320',
+        'server_uri': 'urn:kep',
+        'use_security': True,
+        'subscription_period_ms': 2000,
+    }
+    tag_index = {
+        '20': {'tag_id': 20, 'connector_id': 1, 'address': 'ns=2;s=A', 'tag_name': 'A from db'},
+    }
+    pipelines = [
+        _pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a'), _legacy_tag('ns=2;s=B', 'b')]),
+        _pipeline(
+            'cos', 'm2', [{'tag_id': 20, 'frequency': 500}, {'tag_id': 7, 'frequency': 1000}]
+        ),
+    ]
+
+    entries, issues = _gather(
+        pipelines,
+        connector_index={**connectors, '1': connector_1},
+        tag_index={**tags_by_id, **tag_index},
+    )
+
+    assert [(e['server']['server_id'], e['address']) for e in entries] == [
+        ('1', 'ns=2;s=A'),
+        ('1', 'ns=2;s=B'),
+        ('3', 'ns=3;i=1001'),
+    ]
+    assert entries[0]['names'] == [_name('cos', 'A from db', 'm2'), _name('sin', 'a', 'm1')]
+    assert entries[0]['frequency'] == 500
+    # Same server_id in connectors and opc-servers: the connectors fields win
+    assert entries[0]['server']['subscription_period_ms'] == 2000
+    assert entries[1]['server'] is entries[0]['server']
+    assert issues == {'tags': set(), 'connectors': set(), 'servers': set(), 'invalid': set()}
+
+
+def test_gather_read_tags_server_name_clash():
+    """Two servers with the same name get distinct slot keys, stable by server_id"""
+    servers = {
+        '1': opc_servers['1'],
+        '2': {**opc_servers['2'], 'server_name': 'kepserver'},
+    }
+    pipelines = [
+        _pipeline(
+            'sin',
+            'm1',
+            [_legacy_tag('ns=2;s=A', 'a', server_id='2'), _legacy_tag('ns=2;s=B', 'b')],
+        ),
+    ]
+
+    entries, _issues = _gather(pipelines, servers=servers)
+
+    assert [e['server']['key'] for e in entries] == ['kepserver', 'kepserver:2']
+    assert [e['server']['name'] for e in entries] == ['kepserver', 'kepserver']
+
+
+def test_gather_read_tags_active_read():
+    """Active read: on if any pipeline asks, with the lowest cycle_multiple (D3)"""
+    pipelines = [
+        _pipeline('p1', 'm1', [_legacy_tag('ns=2;s=A', 'a')]),
+        _pipeline(
+            'p2',
+            'm2',
+            [_legacy_tag('ns=2;s=A', 'a', active_read={'enabled': True, 'cycle_multiple': 4})],
+        ),
+        _pipeline(
+            'p3',
+            'm3',
+            [_legacy_tag('ns=2;s=A', 'a', active_read={'enabled': True, 'cycle_multiple': 2})],
+        ),
+        _pipeline(
+            'p4',
+            'm4',
+            [_legacy_tag('ns=2;s=B', 'b', active_read={'enabled': False, 'cycle_multiple': 1})],
+        ),
+        _pipeline('p5', 'm5', [_legacy_tag('ns=2;s=C', 'c', active_read={'enabled': True})]),
+    ]
+
+    entries, _issues = _gather(pipelines)
+
+    assert [e['active_read'] for e in entries] == [
+        {'enabled': True, 'cycle_multiple': 2},
+        {'enabled': False},
+        {'enabled': True, 'cycle_multiple': 1},
+    ]
 
 
 def test_gather_read_tags_filters_non_scouter_workflows():
-    """Test that gather_read_tags only processes scouter workflow types and ignores others"""
+    """Only scouter pipelines feed the slots"""
     pipelines = [
-        {
-            'workflow_type': 'scouter',
-            'schedule_name': 'test_scouter_schedule',
-            'read_tags': [
-                {
-                    'server_id': '1',
-                    'server_name': 'test_server_name',
-                    'tag_address': 'test_tag_address',
-                }
-            ],
-        },
-        {
-            'workflow_type': 'predictions_batch',
-            'schedule_name': 'test_predictions_schedule',
-            'read_tags': [
-                {
-                    'server_id': '2',
-                    'server_name': 'test_server_name2',
-                    'tag_address': 'test_tag_address_predictions',
-                }
-            ],
-        },
-        {
-            'workflow_type': 'minimal_retrain',
-            'schedule_name': 'test_retrain_schedule',
-            'read_tags': [
-                {
-                    'server_id': '3',
-                    'server_name': 'test_server_name3',
-                    'tag_address': 'test_tag_address_retrain',
-                }
-            ],
-        },
-        {
-            'workflow_type': 'pi_web_api_scouter',
-            'schedule_name': 'test_pi_web_api_schedule',
-            'read_tags': [
-                {
-                    'server_id': '4',
-                    'server_name': 'test_server_name4',
-                    'tag_address': 'test_tag_address_pi_web_api',
-                }
-            ],
-        },
-        {
-            'workflow_type': 'drift',
-            'schedule_name': 'test_drift_schedule',
-            'read_tags': [
-                {
-                    'server_id': '5',
-                    'server_name': 'test_server_name5',
-                    'tag_address': 'test_tag_address_drift',
-                }
-            ],
-        },
-        {
-            'workflow_type': 'scouter',
-            'schedule_name': 'test_scouter_schedule2',
-            'read_tags': [
-                {
-                    'server_id': '1',
-                    'server_name': 'test_server_name',
-                    'tag_address': 'test_tag_address2',
-                }
-            ],
-        },
+        _pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a')]),
+        _pipeline('pred', 'm2', [_legacy_tag('ns=2;s=P', 'p')], workflow_type='predictions_batch'),
+        _pipeline('retrain', 'm3', [_legacy_tag('ns=2;s=R', 'r')], workflow_type='minimal_retrain'),
+        _pipeline('drift', 'm4', [_legacy_tag('ns=2;s=D', 'd')], workflow_type='drift'),
     ]
 
-    result = gather_read_tags(pipelines)
+    entries, _issues = _gather(pipelines)
 
-    # Only scouter workflow types should be included
-    expected = {
-        '1:test_tag_address': {
-            'server_id': '1',
-            'server_name': 'test_server_name',
-            'tag_address': 'test_tag_address',
-            'topics': ['raw_test_scouter_schedule'],
-        },
-        '1:test_tag_address2': {
-            'server_id': '1',
-            'server_name': 'test_server_name',
-            'tag_address': 'test_tag_address2',
-            'topics': ['raw_test_scouter_schedule2'],
-        },
-    }
+    assert [e['address'] for e in entries] == ['ns=2;s=A']
 
-    assert result == expected
-    # Ensure non-scouter pipelines are not included
-    assert '2:test_tag_address_predictions' not in result
-    assert '3:test_tag_address_retrain' not in result
-    assert '4:test_tag_address_pi_web_api' not in result
-    assert '5:test_tag_address_drift' not in result
+
+def test_gather_read_tags_without_read_tags():
+    pipeline = _pipeline('sin', 'm1', [])
+    del pipeline['read_tags']
+
+    assert _gather([pipeline]) == (
+        [],
+        {'tags': set(), 'connectors': set(), 'servers': set(), 'invalid': set()},
+    )
 
 
 def test_build_tag_config():
-    tags = [
-        {
-            'server_id': '1',
-            'server_name': 'test_server_name',
-            'tag_address': 'test_tag_address',
-            'frequency': 1000,
-        },
-        {
-            'server_id': '2',
-            'server_name': 'test_server_name2',
-            'tag_address': 'test_tag_address2',
-            'frequency': 1000,
-        },
-        {
-            'server_id': '2',
-            'server_name': 'test_server_name2',
-            'tag_address': 'test_tag_address3',
-            'frequency': 300,
-        },
-        {
-            'server_id': '3',
-            'server_name': 'test_server_name3',
-            'tag_address': 'test_tag_address4',
-            'frequency': 200,
-        },
+    """Slot v2 (contract C1): no certificate paths, subscription period per server"""
+    pipelines = [
+        _pipeline(
+            'sin',
+            'm1',
+            [
+                _legacy_tag('ns=2;s=A', 'a', frequency=1000),
+                _legacy_tag('ns=2;s=B', 'b', frequency='300'),
+                _legacy_tag('ns=2;s=C', 'c', frequency=200, server_id='2'),
+                {'tag_id': 7, 'frequency': 100},
+            ],
+        )
     ]
+    entries, _issues = _gather(pipelines)
 
-    opc_servers = {
-        '1': {
-            'server_name': 'test_server_name',
-            'url': 'test_url',
-            'uri': 'test_uri',
-        },
-        '2': {
-            'server_name': 'test_server_name2',
-            'url': 'test_url2',
-            'uri': 'test_uri2',
-        },
-    }
+    result = build_tag_config(entries)
 
-    result = build_tag_config(tags, opc_servers)
-
-    expected = {
-        'test_server_name': {
+    assert result == {
+        'kepserver': {
             'server_id': '1',
-            'name': 'test_server_name',
-            'url': 'test_url',
-            'server_uri': 'test_uri',
-            'cert_path': None,
-            'private_key_path': None,
-            'server_cert_path': None,
-            'subscription_period_ms': 500,
-            'tags': {
-                'test_tag_address': {
-                    'server_id': '1',
-                    'server_name': 'test_server_name',
-                    'tag_address': 'test_tag_address',
-                    'frequency': 1000,
-                },
-            },
-        },
-        'test_server_name2': {
-            'server_id': '2',
-            'name': 'test_server_name2',
-            'url': 'test_url2',
-            'server_uri': 'test_uri2',
-            'cert_path': None,
-            'private_key_path': None,
-            'server_cert_path': None,
+            'name': 'kepserver',
+            'url': 'opc.tcp://kep:49320',
+            'server_uri': 'urn:kep',
+            'use_security': True,
             'subscription_period_ms': 150,
             'tags': {
-                'test_tag_address2': {
-                    'server_id': '2',
-                    'server_name': 'test_server_name2',
-                    'tag_address': 'test_tag_address2',
+                'ns=2;s=A': {
                     'frequency': 1000,
+                    'active_read': {'enabled': False},
+                    'names': [_name('sin', 'a', 'm1')],
                 },
-                'test_tag_address3': {
-                    'server_id': '2',
-                    'server_name': 'test_server_name2',
-                    'tag_address': 'test_tag_address3',
+                'ns=2;s=B': {
                     'frequency': 300,
+                    'active_read': {'enabled': False},
+                    'names': [_name('sin', 'b', 'm1')],
+                },
+            },
+        },
+        'simulator': {
+            'server_id': '2',
+            'name': 'simulator',
+            'url': 'opc.tcp://sim:4840',
+            'server_uri': 'urn:sim',
+            'use_security': False,
+            'subscription_period_ms': 100,
+            'tags': {
+                'ns=2;s=C': {
+                    'frequency': 200,
+                    'active_read': {'enabled': False},
+                    'names': [_name('sin', 'c', 'm1')],
+                },
+            },
+        },
+        'plant': {
+            'server_id': '3',
+            'name': 'plant',
+            'url': 'opc.tcp://plant:4840',
+            'server_uri': 'urn:plant',
+            'use_security': True,
+            'subscription_period_ms': 1500,
+            'tags': {
+                'ns=3;i=1001': {
+                    'frequency': 100,
+                    'active_read': {'enabled': False},
+                    'names': [_name('sin', 'Pressure', 'm1')],
                 },
             },
         },
     }
 
-    assert result == (expected, ['3'])
+
+def test_build_tag_config_empty():
+    assert build_tag_config([]) == {}
+
+
+def test_scouter_skips_tags_without_name():
+    """A tag_id the tag database cannot resolve has no name and stays out of model_tags"""
+    config = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'sin',
+        'model_id': 'm1',
+        'model': {'name': 'model m1'},
+        'read_tags': [{'tag_name': 'a'}, {'tag_id': 404}],
+    }
+
+    assert list(scouter(config)['model_tags']) == ['a']
+
+
+def test_referenced_tag_ids():
+    """Ids in int and text form, scouter pipelines only, empty or null tag_id ignored"""
+    pipelines = [
+        _pipeline('sin', 'm1', [{'tag_id': 7}, {'tag_id': '12'}, {'tag_id': ''}, {'tag_id': None}]),
+        _pipeline('cos', 'm2', [{'tag_id': 'abc'}, _legacy_tag('ns=2;s=A', 'a')]),
+        _pipeline('pred', 'm3', [{'tag_id': 99}], workflow_type='predictions_batch'),
+    ]
+
+    # Sorted by text (ints first), so the query is the same on every run
+    assert referenced_tag_ids(pipelines) == [12, 7, '12', '7', 'abc']
+
+
+def test_referenced_tag_ids_without_tag_ids():
+    pipeline = _pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a')])
+    no_read_tags = _pipeline('cos', 'm2', [])
+    del no_read_tags['read_tags']
+
+    assert referenced_tag_ids([pipeline, no_read_tags]) == []
+
+
+def test_filter_by_tag_ids():
+    """A tag_id $match goes in front, on a new aggregation list"""
+    query = {'collection': 'connector_tags', 'aggregation': [{'$project': {'_id': 0}}]}
+
+    result = filter_by_tag_ids(query, [7, '7'])
+
+    assert result == {
+        'collection': 'connector_tags',
+        'aggregation': [{'$match': {'tag_id': {'$in': [7, '7']}}}, {'$project': {'_id': 0}}],
+    }
+    assert result['aggregation'] is not query['aggregation']
+    assert query['aggregation'] == [{'$project': {'_id': 0}}]
+
+
+def test_index_skips_incomplete_documents():
+    """A malformed tag database document is left out instead of failing the run"""
+    tags = [
+        {'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1', 'tag_name': 'a'},
+        {'connector_id': 3, 'address': 'ns=3;i=2', 'tag_name': 'no id'},
+        {'tag_id': 9, 'connector_id': 3, 'address': '', 'tag_name': 'no address'},
+    ]
+    connector_list = [
+        {'connector_id': 3, 'name': 'plant', 'url': 'opc.tcp://p:4840', 'server_uri': 'urn:p'},
+        {'connector_id': 4, 'name': 'no url', 'server_uri': 'urn:x'},
+        {'name': 'no id', 'url': 'opc.tcp://x:4840', 'server_uri': 'urn:x'},
+    ]
+
+    assert list(index_connector_tags(tags)) == ['7']
+    assert list(index_connectors(connector_list)) == ['3']
+
+
+def test_gather_read_tags_empty_tag_id_keeps_legacy_fields():
+    """tag_id '' is no tag database reference: the current fields are used (P4)"""
+    entries, issues = _gather([_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a', tag_id='')])])
+
+    assert [(e['server']['server_id'], e['address']) for e in entries] == [('1', 'ns=2;s=A')]
+    assert issues == {'tags': set(), 'connectors': set(), 'servers': set(), 'invalid': set()}
+
+
+def test_gather_read_tags_invalid_read_tags():
+    """Unusable read tags are reported and skipped; the others still make the slot"""
+    pipelines = [
+        _pipeline(
+            'sin',
+            'm1',
+            [
+                _legacy_tag('ns=2;s=A', 'a'),
+                {'server_id': '1', 'tag_address': 'ns=2;s=B', 'tag_name': 'b'},
+                _legacy_tag('ns=2;s=C', 'c', frequency='fast'),
+                {'server_id': '1', 'tag_address': 'ns=2;s=D', 'frequency': 1000},
+                {'tag_address': 'ns=2;s=E', 'tag_name': 'e', 'frequency': 1000},
+                {'tag_id': 7},
+            ],
+        )
+    ]
+
+    entries, issues = _gather(pipelines)
+
+    assert [e['address'] for e in entries] == ['ns=2;s=A']
+    assert issues['invalid'] == {
+        'sin/7',
+        'sin/ns=2;s=B',
+        'sin/ns=2;s=C',
+        'sin/ns=2;s=D',
+        'sin/ns=2;s=E',
+    }
+
+
+def test_gather_read_tags_frequency_type_does_not_depend_on_order():
+    """15000 and '15000' give the same int frequency in any pipeline order (O6)"""
+    as_int = _pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a', frequency=15000)])
+    as_text = _pipeline('cos', 'm2', [_legacy_tag('ns=2;s=A', 'a', frequency='15000')])
+
+    first, _ = _gather([as_int, as_text])
+    second, _ = _gather([as_text, as_int])
+
+    assert first[0]['frequency'] == second[0]['frequency'] == 15000
+    assert isinstance(first[0]['frequency'], int)
+    assert isinstance(second[0]['frequency'], int)
+
+
+def test_gather_read_tags_server_key_never_collides():
+    """A server literally named 'x:3' does not share its key with server 3"""
+    servers = {
+        '1': {**opc_servers['1'], 'server_name': 'x:3'},
+        '2': {**opc_servers['2'], 'server_name': 'x'},
+        '3': {'id': '3', 'server_name': 'x', 'url': 'opc.tcp://three:4840', 'uri': 'urn:3'},
+    }
+    pipelines = [
+        _pipeline(
+            'sin',
+            'm1',
+            [
+                _legacy_tag('ns=2;s=A', 'a', server_id='1'),
+                _legacy_tag('ns=2;s=B', 'b', server_id='2'),
+                _legacy_tag('ns=2;s=C', 'c', server_id='3'),
+            ],
+        ),
+    ]
+
+    entries, _issues = _gather(pipelines, servers=servers)
+    slot = build_tag_config(entries)
+
+    keys = [e['server']['key'] for e in entries]
+    assert len(set(keys)) == 3
+    assert {server['server_id'] for server in slot.values()} == {'1', '2', '3'}
+    assert all(len(server['tags']) == 1 for server in slot.values())
 
 
 def test_base_scouter():

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,162 @@ 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_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,
+):
+    # 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

tests/orchestrator/workflows/test_orchestrator.py

@@ -1,9 +1,14 @@
-from unittest.mock import ANY, AsyncMock, call, patch
+from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
 
 from pytest import fixture, mark
 
 from orchestrator.activities.activities import Activities
-from orchestrator.workflows.orchestrator import Orchestrator
+from orchestrator.workflows.orchestrator import (
+    DEFAULT_CONNECTOR_TAGS_QUERY,
+    DEFAULT_CONNECTORS_QUERY,
+    TAG_DATABASE_PATCH,
+    Orchestrator,
+)
 
 
 @fixture
@@ -22,8 +27,10 @@ metadata = {
 
 
 @mark.asyncio
+@patch('orchestrator.workflows.orchestrator.referenced_tag_ids', return_value=[7, '7'])
 @patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock)
-async def test_run(workflow_mock, orchestrator):
+async def test_run(workflow_mock, referenced_tag_ids_mock, orchestrator):
+    workflow_mock.patched = MagicMock(return_value=True)
     input_data = {
         'pipelines_query': 'SELECT * FROM bucket',
         'opc_servers_query': 'SELECT * FROM servers',
@@ -109,11 +116,54 @@ async def test_run(workflow_mock, orchestrator):
         ]
     )
 
+    workflow_mock.patched.assert_called_once_with(TAG_DATABASE_PATCH)
+    referenced_tag_ids_mock.assert_called_once_with(
+        workflow_mock.start_local_activity_method.return_value
+    )
+
+    workflow_mock.start_local_activity_method.assert_has_calls(
+        [
+            call(
+                Activities.aggregate_documents_in_mongodb,
+                {**metadata, 'query': DEFAULT_CONNECTORS_QUERY},
+                retry_policy=ANY,
+                start_to_close_timeout=ANY,
+            ),
+            call(
+                Activities.aggregate_documents_in_mongodb,
+                {
+                    **metadata,
+                    'query': {
+                        **DEFAULT_CONNECTOR_TAGS_QUERY,
+                        'aggregation': [
+                            {'$match': {'tag_id': {'$in': [7, '7']}}},
+                            *DEFAULT_CONNECTOR_TAGS_QUERY['aggregation'],
+                        ],
+                    },
+                },
+                retry_policy=ANY,
+                start_to_close_timeout=ANY,
+            ),
+        ]
+    )
+
+    # The activity appends a stage to the aggregation it gets: never the module default
+    tag_database_calls = _tag_database_calls(workflow_mock)
+    assert len(tag_database_calls) == 2
+    for c in tag_database_calls:
+        aggregation = c.args[1]['query']['aggregation']
+        assert aggregation is not DEFAULT_CONNECTORS_QUERY['aggregation']
+        assert aggregation is not DEFAULT_CONNECTOR_TAGS_QUERY['aggregation']
+
     workflow_mock.start_local_activity_method.assert_has_calls(
         [
             call(
                 Activities.process_schedules,
-                {**metadata, 'pipelines': workflow_mock.start_local_activity_method.return_value},
+                {
+                    **metadata,
+                    'pipelines': workflow_mock.start_local_activity_method.return_value,
+                    'connector_tags': workflow_mock.start_local_activity_method.return_value,
+                },
                 retry_policy=ANY,
                 start_to_close_timeout=ANY,
             )
@@ -127,6 +177,8 @@ async def test_run(workflow_mock, orchestrator):
                 {
                     **metadata,
                     'opc_servers': workflow_mock.start_local_activity_method.return_value,
+                    'connectors': workflow_mock.start_local_activity_method.return_value,
+                    'connector_tags': workflow_mock.start_local_activity_method.return_value,
                     'active_ingestors': workflow_mock.start_local_activity_method.return_value,
                     'pipelines': workflow_mock.start_local_activity_method.return_value,
                 },
@@ -346,3 +398,96 @@ async def test_run(workflow_mock, orchestrator):
             )
         ]
     )
+
+
+def _tag_database_calls(workflow_mock):
+    return [
+        c
+        for c in workflow_mock.start_local_activity_method.call_args_list
+        if c.args[0] is Activities.aggregate_documents_in_mongodb
+        and isinstance(c.args[1]['query'], dict)
+        and c.args[1]['query'].get('collection') in ('connectors', 'connector_tags')
+    ]
+
+
+def _process_slots_input(workflow_mock):
+    return next(
+        c.args[1]
+        for c in workflow_mock.start_local_activity_method.call_args_list
+        if c.args[0] is Activities.process_slots
+    )
+
+
+@mark.asyncio
+@patch('orchestrator.workflows.orchestrator.referenced_tag_ids', return_value=[7])
+@patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock)
+async def test_run_with_tag_database_queries_from_input(workflow_mock, _ids, orchestrator):
+    """The input overrides the default tag database aggregations (SIENTIAPDE-1974, D9)"""
+    workflow_mock.patched = MagicMock(return_value=True)
+    connectors_query = {'collection': 'connectors', 'aggregation': [{'$match': {'type': 'opc_ua'}}]}
+    connector_tags_query = {
+        'collection': 'connector_tags',
+        'aggregation': [{'$match': {'category': 'Read'}}],
+    }
+    input_data = {
+        'pipelines_query': 'SELECT * FROM bucket',
+        'opc_servers_query': 'SELECT * FROM servers',
+        'connectors_query': connectors_query,
+        'connector_tags_query': connector_tags_query,
+        'schedule_name': 'test-schedule-name',
+    }
+
+    await orchestrator.run(input_data)
+
+    queries = [c.args[1]['query'] for c in _tag_database_calls(workflow_mock)]
+    assert queries == [
+        connectors_query,
+        {
+            'collection': 'connector_tags',
+            'aggregation': [
+                {'$match': {'tag_id': {'$in': [7]}}},
+                {'$match': {'category': 'Read'}},
+            ],
+        },
+    ]
+
+
+@mark.asyncio
+@patch('orchestrator.workflows.orchestrator.referenced_tag_ids', return_value=[])
+@patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock)
+async def test_run_without_tag_ids_skips_tag_database(workflow_mock, _ids, orchestrator):
+    """No read tag with tag_id: the tag database is not queried at all (P4)"""
+    workflow_mock.patched = MagicMock(return_value=True)
+    input_data = {
+        'pipelines_query': 'SELECT * FROM bucket',
+        'opc_servers_query': 'SELECT * FROM servers',
+        'schedule_name': 'test-schedule-name',
+    }
+
+    await orchestrator.run(input_data)
+
+    workflow_mock.patched.assert_called_once_with(TAG_DATABASE_PATCH)
+    assert _tag_database_calls(workflow_mock) == []
+    slots_input = _process_slots_input(workflow_mock)
+    assert slots_input['connectors'] == []
+    assert slots_input['connector_tags'] == []
+
+
+@mark.asyncio
+@patch('orchestrator.workflows.orchestrator.referenced_tag_ids', return_value=[7])
+@patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock)
+async def test_run_replay_before_tag_database_patch(workflow_mock, _ids, orchestrator):
+    """A run started before the patch replays without the tag database loads"""
+    workflow_mock.patched = MagicMock(return_value=False)
+    input_data = {
+        'pipelines_query': 'SELECT * FROM bucket',
+        'opc_servers_query': 'SELECT * FROM servers',
+        'schedule_name': 'test-schedule-name',
+    }
+
+    await orchestrator.run(input_data)
+
+    assert _tag_database_calls(workflow_mock) == []
+    slots_input = _process_slots_input(workflow_mock)
+    assert slots_input['connectors'] == []
+    assert slots_input['connector_tags'] == []