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>

orchestrator/utils/orchestrator_functions.py

@@ -444,6 +444,25 @@ def referenced_tag_ids(pipelines: list[dict[str, Any]]) -> list[Any]:
     return sorted(ids, key=lambda value: (isinstance(value, str), str(value)))
 
 
+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.
@@ -625,6 +644,44 @@ def _assign_server_keys(servers: dict[str, dict[str, Any]]) -> None:
         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]],
@@ -690,35 +747,7 @@ def gather_read_tags(
             located = _locate_read_tag(tag, opc_servers, connectors, tags_by_id, issues)
             if located is None:
                 continue
-            server, address = located
+            _add_read_tag(entries, servers, pipeline, tag, *located)
-
-            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)
 
     _assign_server_keys(servers)
 

orchestrator/workflows/orchestrator.py

@@ -7,7 +7,7 @@ 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
+    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
@@ -53,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]):
         """
@@ -142,51 +193,9 @@ class Orchestrator:
 
         pipeline_config = await pipeline_config_handler
 
-        # Tag database: loaded only when a scouter read tag carries a tag_id, and only
+        connectors, connector_tags = await self._load_tag_database(
-        # those tags, so an environment without it never depends on its content (P4) and
+            input_data, metadata, pipeline_config
-        # 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.aggregate_documents_in_mongodb,
-                {
-                    **metadata,
-                    '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': {
-                        **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),
-            )
-
-            connectors = await connectors_handler
-            connector_tags = await connector_tags_handler
 
         orchestrated_schedules = await orchestrated_schedules_handler
         current_slot_config = await current_slot_config_handler

tests/orchestrator/utils/test_orchestrator_functions.py

@@ -5,6 +5,7 @@ from orchestrator.utils.orchestrator_functions import (
     build_tag_config,
     common_config,
     drift,
+    filter_by_tag_ids,
     gather_read_tags,
     index_connector_tags,
     index_connectors,
@@ -764,6 +765,20 @@ def test_referenced_tag_ids_without_tag_ids():
     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 = [