SIENTIAPDE-1974: fix(slots): harden the tag database path after code review

Code review (code-review skill, xhigh) of efa0d94: 8 findings, 6 fixed here,
2 documented (decisions for the team).

Fixed:
- The tag database is read only when a scouter read tag carries a tag_id, and
  only the referenced tags ($match tag_id $in, int and text forms), through
  aggregations that project only the fields the slot uses. Environments without
  a tag database, or whose pipelines do not use tag_id, never query it (P4), and
  other attributes (e.g. an ObjectId, which the Temporal JSON converter rejects)
  never reach a payload. Fresh aggregation lists are passed, because
  aggregate_documents_in_mongodb appends a stage to the list it gets.
- Incomplete connectors / connector_tags documents are left out of the indexes
  (their read tags are notified as not found) instead of raising KeyError and
  failing the whole run.
- frequency is stored as int, so 15000 vs '15000' across pipelines no longer
  makes the slot depend on the pipeline order (O6).
- A missing, null or empty tag_id means "no tag database": an empty string no
  longer drops a legacy read tag.
- A read tag without an integer frequency (or, without tag_id, without
  server_id / tag_address / tag_name) is skipped and notified
  (ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION) instead of failing
  the slot build.
- Server slot keys can no longer collide (name:server_id is extended until free).

Documented in the README:
- The gateway's plan B (READ_ENGINE=legacy) reads v1 only: switching to it after
  this deploy also requires rolling the Orchestrator back.
- Renaming a tag in connector_tags does not redeploy the scouter (schedules
  update on the pipeline's updated_at).

Validation: 193 unit tests; the real workflow on a local Temporal test server
(sandbox on) with the real Formatters activities and the dev data: slot v2
written without querying the tag database, a second run inserts nothing, and a
tag_id pipeline queries both aggregations with the $in filter; the dev dry run
is still equivalent to the current v1 slot through the gateway's parse_slot.

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

README.md

@@ -121,22 +121,29 @@ The **Orchestrator** workflow is the main coordination workflow that manages pip
 }
 ```
 
-Optional: `connectors_query` and `connector_tags_query` override the tag database queries. The defaults live in the code (`orchestrator/workflows/orchestrator.py`), so the schedule input does not need them:
+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",
-    "filters": {"type": "opc_ua", "status": "Active"}
+    "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",
-    "filters": {"category": "Read", "status": "Active"}
+    "aggregation": [
+      {"$match": {"category": "Read", "status": "Active"}},
+      {"$project": {"_id": 0, "tag_id": 1, "connector_id": 1, "address": 1, "tag_name": 1}}
+    ]
   }
 }
 ```
 
-An empty or missing collection loads nothing, and the pipelines keep working through `opc-servers`.
+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)
 
@@ -171,18 +178,22 @@ How each scouter `read_tags` entry is resolved:
 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. Every read tag needs one, with or without `tag_id`.
+- **`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 the slot keys `name` and `name:server_id`.
+- **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. It is notified once per run:
+- **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;
+  - `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;
+  - `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_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
@@ -675,16 +686,23 @@ Temporal input configuration sample:
     },
     "connectors_query": {
         "collection": "connectors",
-        "filters": {"type": "opc_ua", "status": "Active"}
+        "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",
-        "filters": {"category": "Read", "status": "Active"}
+        "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.
+`connectors_query` and `connector_tags_query` are optional: the values above are the defaults in the code (see "OPC slots" above).
 
 #### Alerts Workflow
 

orchestrator/activities/formatters.py

@@ -22,6 +22,7 @@ with workflow.unsafe.imports_passed_through():
         drift,
         gather_read_tags,
         index_connector_tags,
+        index_connectors,
         minimal_retrain,
         pi_web_api_scouter,
         predictions_batch,
@@ -234,7 +235,8 @@ class Formatters(SientiaMonitoring):
         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, or a server_id missing from opc_servers.
+        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
@@ -259,10 +261,7 @@ class Formatters(SientiaMonitoring):
         pipelines = input_data['pipelines']
         active_ingestors = input_data['active_ingestors']
         opc_servers = {str(server['id']): server for server in input_data['opc_servers']}
-        connectors = {
+        connectors = index_connectors(input_data.get('connectors', []))
-            str(connector['connector_id']): connector
-            for connector in input_data.get('connectors', [])
-        }
         tags_by_id = index_connector_tags(input_data.get('connector_tags', []))
 
         entries, issues = gather_read_tags(pipelines, opc_servers, connectors, tags_by_id)
@@ -286,8 +285,9 @@ class Formatters(SientiaMonitoring):
         Notify the read tags left out of the slots, one notification per kind of issue.
 
         Args:
-            issues (dict[str, set[str]]): Unresolved ids from gather_read_tags, by kind:
+            issues (dict[str, set[str]]): What gather_read_tags left out, by kind:
-                'servers' (opc-servers server_id), 'tags' (tag_id), 'connectors' (connector_id)
+                '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 = (
@@ -299,12 +299,17 @@ class Formatters(SientiaMonitoring):
             (
                 'tags',
                 'ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION',
-                'Tags {} not found among the active Read tags of connector_tags',
+                'Tags {} not found among the active, complete Read tags of connector_tags',
             ),
             (
                 'connectors',
                 'ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION',
-                'Connectors {} not found or not active in connectors',
+                '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:

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
 
@@ -400,11 +400,56 @@ def predictions_batch(config: dict[str, Any]):
     }
 
 
+# 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')
+
+
+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]]): Pipeline configurations
+
+    Returns:
+        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)))
+
+
 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),
@@ -413,7 +458,28 @@ def index_connector_tags(connector_tags: list[dict[str, Any]]) -> dict[str, dict
     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}
+    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(
@@ -438,8 +504,9 @@ def resolve_tag_names(
     resolved = []
     for tag in read_tags:
         name = tag.get('tag_name')
-        if not name and tag.get('tag_id') is not None:
+        tag_id = _tag_id(tag)
-            name = tags_by_id.get(str(tag['tag_id']), {}).get('tag_name')
+        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
 
@@ -476,6 +543,14 @@ def _opc_server(server_id: str, opc_server: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+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]],
@@ -491,8 +566,8 @@ def _locate_read_tag(
     tag_address, with the server from opc-servers (RF4). What cannot be resolved is
     recorded in issues ('tags', 'connectors' or 'servers').
     """
-    if tag.get('tag_id') is not None:
+    tag_id = _tag_id(tag)
-        tag_id = str(tag['tag_id'])
+    if tag_id is not None:
         connector_tag = tags_by_id.get(tag_id)
         if connector_tag is None:
             issues['tags'].add(tag_id)
@@ -512,6 +587,18 @@ def _locate_read_tag(
     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'):
@@ -524,14 +611,16 @@ def _merge_active_read(current: dict[str, Any], requested: dict[str, Any] | None
 
 def _assign_server_keys(servers: dict[str, dict[str, Any]]) -> None:
     """
-    Give each server a unique slot key: its name, or name:server_id when another
+    Give each server a unique slot key: its name, or name:server_id (extended until
-    server_id already took that name. Assigned in server_id order, so the keys are the
+    free) when another server already took it. Assigned in server_id order, so the keys
-    same on every run and in every slot.
+    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'] if server['name'] not in taken else f'{server["name"]}:{server_id}'
+        key = server['name']
+        while key in taken:
+            key = f'{key}:{server_id}'
         server['key'] = key
         taken.add(key)
 
@@ -559,8 +648,8 @@ def gather_read_tags(
             '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 keyed by
+        connectors (dict[str, dict[str, Any]]): Active OPC UA connectors indexed by
-            str(connector_id)
+            index_connectors
         tags_by_id (dict[str, dict[str, Any]]): Active Read connector_tags indexed by
             index_connector_tags
 
@@ -571,14 +660,20 @@ def gather_read_tags(
                   subscription_period_ms (None = derive from the frequencies) and key
                   (the server's unique slot key)
                 - address (str): The OPC node address
-                - frequency: The lowest frequency among the pipelines (RF6)
+                - 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
-            - Ids that could not be resolved: 'tags' (tag_id), 'connectors'
+            - What was left out: 'tags' (tag_id), 'connectors' (connector_id), 'servers'
-              (connector_id) and 'servers' (opc-servers server_id)
+              (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()}
+    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]] = {}
 
@@ -586,7 +681,12 @@ def gather_read_tags(
         if pipeline['workflow_type'] != 'scouter':
             continue
 
-        for tag in resolve_tag_names(pipeline.get('read_tags', []), tags_by_id):
+        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
+
             located = _locate_read_tag(tag, opc_servers, connectors, tags_by_id, issues)
             if located is None:
                 continue
@@ -597,17 +697,18 @@ def gather_read_tags(
             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': tag['frequency'],
+                    'frequency': frequency,
                     'active_read': {'enabled': False},
                     'names': [],
                 },
             )
-            if int(tag['frequency']) < int(entry['frequency']):
+            entry['frequency'] = min(entry['frequency'], frequency)
-                entry['frequency'] = tag['frequency']
             entry['active_read'] = _merge_active_read(entry['active_read'], tag.get('active_read'))
 
             name = {
@@ -671,7 +772,7 @@ def build_tag_config(entries: list[dict[str, Any]]) -> dict[str, Any]:
 
     for server_config in slot_config.values():
         if server_config['subscription_period_ms'] is None:
-            frequencies = [int(tag['frequency']) for tag in server_config['tags'].values()]
+            frequencies = [tag['frequency'] for tag in server_config['tags'].values()]
             server_config['subscription_period_ms'] = min(frequencies) / 2
 
     return slot_config

orchestrator/workflows/orchestrator.py

@@ -7,16 +7,36 @@ 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 referenced_tag_ids
 
 # Tag database (SIENTIAPDE-1569) read by the slot v2 (SIENTIAPDE-1974). The workflow
-# input may override these queries (connectors_query / connector_tags_query).
+# 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',
-    'filters': {'type': 'opc_ua', 'status': 'Active'},
+    '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',
-    'filters': {'category': 'Read', 'status': 'Active'},
+    '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'
@@ -49,11 +69,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:
+                Optional fields (used only when a scouter read tag carries a tag_id):
-                - connectors_query (dict[str, Any]): MongoDB query for the tag database
+                - connectors_query (dict[str, Any]): MongoDB aggregation for the tag
-                  connectors (default: DEFAULT_CONNECTORS_QUERY)
+                  database connectors (default: DEFAULT_CONNECTORS_QUERY)
-                - connector_tags_query (dict[str, Any]): MongoDB query for the tag database
+                - connector_tags_query (dict[str, Any]): MongoDB aggregation for the tag
-                  tags (default: DEFAULT_CONNECTOR_TAGS_QUERY)
+                  database tags (default: DEFAULT_CONNECTOR_TAGS_QUERY)
 
         Returns:
             None: Workflow completes without return value
@@ -120,38 +140,58 @@ class Orchestrator:
             start_to_close_timeout=timedelta(seconds=60),
         )
 
-        # Tag database: empty or missing collections just load nothing, and the read
+        pipeline_config = await pipeline_config_handler
-        # tags without tag_id keep using opc-servers
+
-        connectors_handler = None
+        # Tag database: loaded only when a scouter read tag carries a tag_id, and only
-        connector_tags_handler = None
+        # those tags, so an environment without it never depends on its content (P4) and
-        if workflow.patched(TAG_DATABASE_PATCH):
+        # the payload stays bounded by the pipelines. Read tags without tag_id keep using
+        # opc-servers.
+        connectors: list[dict[str, Any]] = []
+        connector_tags: list[dict[str, Any]] = []
+        tag_ids = referenced_tag_ids(pipeline_config)
+        if workflow.patched(TAG_DATABASE_PATCH) and tag_ids:
+            connectors_query = input_data.get('connectors_query', DEFAULT_CONNECTORS_QUERY)
+            connector_tags_query = input_data.get(
+                'connector_tags_query', DEFAULT_CONNECTOR_TAGS_QUERY
+            )
+
+            # Fresh aggregation lists: the activity appends a stage to the one it gets
             connectors_handler = workflow.start_local_activity_method(
-                Activities.find_documents_in_mongodb,
+                Activities.aggregate_documents_in_mongodb,
                 {
                     **metadata,
-                    'query': input_data.get('connectors_query', DEFAULT_CONNECTORS_QUERY),
+                    '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.find_documents_in_mongodb,
+                Activities.aggregate_documents_in_mongodb,
                 {
                     **metadata,
-                    'query': input_data.get('connector_tags_query', DEFAULT_CONNECTOR_TAGS_QUERY),
+                    'query': {
+                        **connector_tags_query,
+                        'aggregation': [
+                            {'$match': {'tag_id': {'$in': tag_ids}}},
+                            *connector_tags_query['aggregation'],
+                        ],
+                    },
                 },
                 retry_policy=retry_policy,
                 start_to_close_timeout=timedelta(seconds=60),
             )
 
-        pipeline_config = await pipeline_config_handler
+            connectors = await connectors_handler
+            connector_tags = await connector_tags_handler
+
         orchestrated_schedules = await orchestrated_schedules_handler
         current_slot_config = await current_slot_config_handler
         opc_servers = await opc_servers_handler
         active_ingestors = await active_ingestors_handler
-        connectors = await connectors_handler if connectors_handler else []
-        connector_tags = await connector_tags_handler if connector_tags_handler else []
 
         formatted_orchestrated_schedules_handler = workflow.start_local_activity_method(
             Activities.format_schedule_config,

tests/orchestrator/activities/test_formatters.py

@@ -445,6 +445,7 @@ def test_process_slots_notifies_unresolved_tags_once(formatters):
                 {'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'},
             ],
         ),
     ]
@@ -475,20 +476,65 @@ def test_process_slots_notifies_unresolved_tags_once(formatters):
             call(
                 metadata=metadata['metadata'],
                 notification_id='ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION',
-                message='Tags 404 not found among the active Read tags of connector_tags',
+                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 or not active in connectors',
+                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 == 3
+    assert formatters.send_notification.call_count == 4
+
+
+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):

tests/orchestrator/utils/test_orchestrator_functions.py

@@ -7,11 +7,13 @@ from orchestrator.utils.orchestrator_functions import (
     drift,
     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,
@@ -380,7 +382,10 @@ def _gather(pipelines, servers=None, connector_index=None, tag_index=None):
 
 
 def test_index_connector_tags():
-    tags = [{'tag_id': 7, 'tag_name': 'a'}, {'tag_id': '8', 'tag_name': 'b'}]
+    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]}
 
@@ -414,7 +419,7 @@ def test_gather_read_tags_multi_name():
     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()}
+    assert issues == {'tags': set(), 'connectors': set(), 'servers': set(), 'invalid': set()}
 
 
 def test_gather_read_tags_same_name_in_two_pipelines():
@@ -516,7 +521,7 @@ 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()}
+    assert issues == {'tags': {'404'}, 'connectors': set(), 'servers': set(), 'invalid': set()}
 
 
 def test_gather_read_tags_inactive_connector():
@@ -524,7 +529,7 @@ def test_gather_read_tags_inactive_connector():
     entries, issues = _gather([_pipeline('sin', 'm1', [{'tag_id': 8, 'frequency': 1000}])])
 
     assert entries == []
-    assert issues == {'tags': set(), 'connectors': {'5'}, 'servers': set()}
+    assert issues == {'tags': set(), 'connectors': {'5'}, 'servers': set(), 'invalid': set()}
 
 
 def test_gather_read_tags_unknown_server():
@@ -533,7 +538,7 @@ def test_gather_read_tags_unknown_server():
     )
 
     assert entries == []
-    assert issues == {'tags': set(), 'connectors': set(), 'servers': {'9'}}
+    assert issues == {'tags': set(), 'connectors': set(), 'servers': {'9'}, 'invalid': set()}
 
 
 def test_gather_read_tags_mixed_with_and_without_tag_id():
@@ -572,7 +577,7 @@ def test_gather_read_tags_mixed_with_and_without_tag_id():
     # 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()}
+    assert issues == {'tags': set(), 'connectors': set(), 'servers': set(), 'invalid': set()}
 
 
 def test_gather_read_tags_server_name_clash():
@@ -644,7 +649,10 @@ 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()})
+    assert _gather([pipeline]) == (
+        [],
+        {'tags': set(), 'connectors': set(), 'servers': set(), 'invalid': set()},
+    )
 
 
 def test_build_tag_config():
@@ -680,7 +688,7 @@ def test_build_tag_config():
                     'names': [_name('sin', 'a', 'm1')],
                 },
                 'ns=2;s=B': {
-                    'frequency': '300',
+                    'frequency': 300,
                     'active_read': {'enabled': False},
                     'names': [_name('sin', 'b', 'm1')],
                 },
@@ -736,6 +744,121 @@ def test_scouter_skips_tags_without_name():
     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_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():
     config = {
         'workflow_type': 'scouter',

tests/orchestrator/workflows/test_orchestrator.py

@@ -27,8 +27,9 @@ 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',
@@ -116,24 +117,44 @@ 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.find_documents_in_mongodb,
+                Activities.aggregate_documents_in_mongodb,
                 {**metadata, 'query': DEFAULT_CONNECTORS_QUERY},
                 retry_policy=ANY,
                 start_to_close_timeout=ANY,
             ),
             call(
-                Activities.find_documents_in_mongodb,
+                Activities.aggregate_documents_in_mongodb,
-                {**metadata, 'query': DEFAULT_CONNECTOR_TAGS_QUERY},
+                {
+                    **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(
@@ -379,13 +400,35 @@ 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, orchestrator):
+async def test_run_with_tag_database_queries_from_input(workflow_mock, _ids, orchestrator):
-    """The input overrides the default tag database queries (SIENTIAPDE-1974, D9)"""
+    """The input overrides the default tag database aggregations (SIENTIAPDE-1974, D9)"""
     workflow_mock.patched = MagicMock(return_value=True)
-    connectors_query = {'collection': 'connectors', 'filters': {'type': 'opc_ua'}}
+    connectors_query = {'collection': 'connectors', 'aggregation': [{'$match': {'type': 'opc_ua'}}]}
-    connector_tags_query = {'collection': 'connector_tags', 'filters': {'category': 'Read'}}
+    connector_tags_query = {
+        'collection': 'connector_tags',
+        'aggregation': [{'$match': {'category': 'Read'}}],
+    }
     input_data = {
         'pipelines_query': 'SELECT * FROM bucket',
         'opc_servers_query': 'SELECT * FROM servers',
@@ -396,27 +439,44 @@ async def test_run_with_tag_database_queries_from_input(workflow_mock, orchestra
 
     await orchestrator.run(input_data)
 
-    workflow_mock.start_local_activity_method.assert_has_calls(
+    queries = [c.args[1]['query'] for c in _tag_database_calls(workflow_mock)]
-        [
+    assert queries == [
-            call(
+        connectors_query,
-                Activities.find_documents_in_mongodb,
+        {
-                {**metadata, 'query': connectors_query},
+            'collection': 'connector_tags',
-                retry_policy=ANY,
+            'aggregation': [
-                start_to_close_timeout=ANY,
+                {'$match': {'tag_id': {'$in': [7]}}},
-            ),
+                {'$match': {'category': 'Read'}},
-            call(
+            ],
-                Activities.find_documents_in_mongodb,
+        },
-                {**metadata, 'query': connector_tags_query},
+    ]
-                retry_policy=ANY,
-                start_to_close_timeout=ANY,
-            ),
-        ]
-    )
 
 
 @mark.asyncio
+@patch('orchestrator.workflows.orchestrator.referenced_tag_ids', return_value=[])
 @patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock)
-async def test_run_replay_before_tag_database_patch(workflow_mock, orchestrator):
+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 = {
@@ -427,28 +487,7 @@ async def test_run_replay_before_tag_database_patch(workflow_mock, orchestrator)
 
     await orchestrator.run(input_data)
 
-    queries = [
+    assert _tag_database_calls(workflow_mock) == []
-        c.args[1].get('query')
+    slots_input = _process_slots_input(workflow_mock)
-        for c in workflow_mock.start_local_activity_method.call_args_list
+    assert slots_input['connectors'] == []
-        if c.args[0] is Activities.find_documents_in_mongodb
+    assert slots_input['connector_tags'] == []
-    ]
-    assert DEFAULT_CONNECTORS_QUERY not in queries
-    assert DEFAULT_CONNECTOR_TAGS_QUERY not in queries
-
-    workflow_mock.start_local_activity_method.assert_has_calls(
-        [
-            call(
-                Activities.process_slots,
-                {
-                    **metadata,
-                    'opc_servers': workflow_mock.start_local_activity_method.return_value,
-                    'connectors': [],
-                    'connector_tags': [],
-                    'active_ingestors': workflow_mock.start_local_activity_method.return_value,
-                    'pipelines': workflow_mock.start_local_activity_method.return_value,
-                },
-                retry_policy=ANY,
-                start_to_close_timeout=ANY,
-            )
-        ]
-    )