chore: bump version to 0.6.8

fix(orchestrator): do not stamp or untrack every schedule when all updates/deletes fail (SIENTIAPDE-2192)

CLAUDE.md

@@ -47,7 +47,7 @@ CI quality gate (`.github/workflows/quality-gate.yml`) calls the shared reusable
 - **`orchestrator/worker/worker.py`** — entrypoint (`python -m orchestrator.worker.worker`). Builds a single `Activities` instance, connects to Temporal, starts Prometheus (`HTTP_METRICS_PORT`) and Temporal SDK metrics (`HTTP_SDK_METRICS_PORT`) servers, then runs three `Worker`s concurrently via `asyncio.gather`, each on its own task queue: `orchestrator-queue`, `alerts-queue`, `reports-queue`. `prepare_worker.py` derives the queue name from the workflow class name (`camel_to_snake`) and reads per-queue concurrency/poller tuning from env vars prefixed with the workflow name (e.g. `ORCHESTRATOR_MAX_CONCURRENT_ACTIVITIES`).
 - **`orchestrator/workflows/`** — Temporal workflow definitions: `orchestrator.py` (pipeline/slot/schedule orchestration), `alerts.py` (real-time ERROR notifications), `reports.py` (scheduled all-level reports). `alerts.py` and `reports.py` both delegate to the same two subworkflows in `workflows/subworkflows/`: `load_notification_package.py` (load notifications + receiver configs from Mongo, incremental via Redis timestamp) and `process_notifications.py` (build HTML email, send, log to Postgres). The difference between alerts and reports is which filter activity they call (`filter_notification_alerts` vs `filter_notification_reports`) and their base data filter (`level: ERROR` by default vs all levels) — `load_notification_package.py`'s `_compute_query_time_filter` can widen either beyond that default to the union of active receiver groups' own `levels` (SIENTIAPDE-2011); the ERROR-only default only holds when no active group narrows further. 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`) 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.
+- **`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.
 
 ### 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, active ingestors, and the tag database (`connectors` and `connector_tags`)
+1. **Parallel Data Loading**: Concurrently loads pipelines, OPC servers, orchestrated schedules, OPC slots, and active ingestors
 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,81 +121,6 @@ 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
@@ -683,27 +608,10 @@ 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

orchestrator/activities/formatters.py

@@ -21,12 +21,9 @@ with workflow.unsafe.imports_passed_through():
         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,
     )
@@ -176,8 +173,6 @@ 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
@@ -188,7 +183,6 @@ 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: {},
@@ -202,13 +196,6 @@ 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']
@@ -229,25 +216,14 @@ 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 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).
+        returns a slot config dictionary. If no ingestor is available, only one slot
+        is created.
 
         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[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.
+                - opc_servers (list[str]): The OPC servers to create ingestor config.
                 - active_ingestors (list[str]): The active ingestors to divide into slots.
 
         Returns:
@@ -259,69 +235,55 @@ 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', []))
 
-        entries, issues = gather_read_tags(pipelines, opc_servers, connectors, tags_by_id)
-        self._notify_slot_issues(issues, metadata)
+        opc_servers = {}
+        for server in opc_servers_list:
+            opc_servers[server['id']] = {
+                **server,
+            }
 
+        tags = list(gather_read_tags(pipelines).values())
+
+        number_of_tags = len(tags)
         number_of_slots = len(active_ingestors) if active_ingestors else 1
-        entries_per_slot = ceil(len(entries) / number_of_slots)
+        tags_per_slot = ceil(number_of_tags / number_of_slots)
 
         slot_config: dict[str, Any] = {}
-        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)
+        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,
+            )
 
         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]:
         """
@@ -468,21 +430,15 @@ 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,
-                  keyed by Redis key (slot:opc_tags:{id}) or by slot id
+                - current_slot_config (dict[str, Any]): The current slot config in Redis
                 - 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 are new or changed
+                - to_insert (dict): Slots that need to be inserted/updated
                 - to_delete (list[str]): Slot IDs that need to be deleted
         """
 
@@ -500,16 +456,7 @@ 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)]
 
-        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}
+        output = {'to_delete': to_delete, 'to_insert': slot_config}
 
         self.info('Created slot config', metadata=metadata)
         self.debug(json.dumps(output, indent=4, sort_keys=True), metadata=metadata)

orchestrator/activities/mongo_db.py

@@ -242,7 +242,12 @@ class MongoDB(SientiaMonitoring):
             for pipeline in updated_pipelines
             if pipeline['success']
         ]
-        data_filter = {'$or': argument} if argument else {}
+        if not argument:
+            # Every update failed: stamp nothing so they are retried next tick (same guard as
+            # create_pipelines_timestamps). An empty filter would match every tracked schedule.
+            self.info('No successful pipeline updates to timestamp', metadata=metadata)
+            return
+        data_filter = {'$or': argument}
 
         try:
             self.mongo_db_repository.update_many(
@@ -343,7 +348,13 @@ class MongoDB(SientiaMonitoring):
             for pipeline in deleted_pipelines
             if pipeline['success']
         ]
-        data_filter = {'$or': argument} if argument else {}
+        if not argument:
+            # Every delete failed: remove nothing (same guard as create_pipelines_timestamps).
+            # An empty filter would wipe orchestrated_schedules and make the next tick delete
+            # and recreate every orchestrated schedule.
+            self.info('No successful pipeline deletions to untrack', metadata=metadata)
+            return
+        data_filter = {'$or': argument}
 
         try:
             self.mongo_db_repository.delete_many('orchestrated_schedules', data_filter, metadata)

orchestrator/utils/orchestrator_functions.py

@@ -1,4 +1,4 @@
-from typing import Any, cast
+from typing import Any
 
 from orchestrator.utils.converters import parse_frequency
 
@@ -174,10 +174,8 @@ 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, 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
+            - read_tags (list[dict]): List of tag configurations with:
+                - tag_name (str): Name of the tag to read
                 - 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])
@@ -190,8 +188,6 @@ 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]),
@@ -400,408 +396,110 @@ 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]:
+def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]:
     """
-    The tag_ids the scouter pipelines reference, for the connector_tags query.
+    Gather all read tags from scouter pipeline configurations.
 
-    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.
+    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.
 
     Args:
-        pipelines (list[dict[str, Any]]): Pipeline configurations
+        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.
 
     Returns:
-        list[Any]: The referenced tag_ids (empty when no read tag carries a tag_id)
+        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}')
     """
-    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 = {}
 
-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]] = {}
-
+    # Get all read tags from pipelines
     for pipeline in pipelines:
         if pipeline['workflow_type'] != 'scouter':
             continue
 
-        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
+        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': []}
 
-            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)
+            tags[tag_string]['topics'].append(f'raw_{pipeline["schedule_name"]}')
 
-    _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
+    return tags
 
 
-def build_tag_config(entries: list[dict[str, Any]]) -> dict[str, Any]:
+def build_tag_config(
+    tags: list[dict[str, Any]], opc_servers: dict[str, Any]
+) -> tuple[dict[str, Any], list]:
     """
-    Build one slot in the slot v2 format (contract C1) from address entries.
+    Build tag configuration for a specific slot and OPC server.
 
-    The slot no longer carries cert_path, private_key_path or server_cert_path: the
-    gateway takes the certificate from MinIO (SIENTIAPDE-1974, O4).
+    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.
 
     Args:
-        entries (list[dict[str, Any]]): Address entries from gather_read_tags for this slot
+        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
 
     Returns:
-        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.
+        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
     """
-    slot_config: dict[str, Any] = {}
 
-    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'],
+    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),
                 'tags': {},
             }
 
-        slot_config[server['key']]['tags'][entry['address']] = {
-            'frequency': entry['frequency'],
-            'active_read': entry['active_read'],
-            'names': entry['names'],
+        slot_config[server_name]['tags'][tag['tag_address']] = {
+            **tag,
         }
 
-    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
+    for server_name in slot_config:
+        frequencies = [int(x['frequency']) for x in slot_config[server_name]['tags'].values()]
 
-    return slot_config
+        min_frequency = min(frequencies) if frequencies else 1000
+
+        slot_config[server_name]['subscription_period_ms'] = min_frequency / 2
+
+    return slot_config, notifications

orchestrator/workflows/orchestrator.py

@@ -7,39 +7,6 @@ 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')
@@ -53,57 +20,6 @@ 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]):
         """
@@ -120,11 +36,6 @@ 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
@@ -192,11 +103,6 @@ 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
@@ -211,7 +117,7 @@ class Orchestrator:
 
         schedules_config_handler = workflow.start_local_activity_method(
             Activities.process_schedules,
-            {**metadata, 'pipelines': pipeline_config, 'connector_tags': connector_tags},
+            {**metadata, 'pipelines': pipeline_config},
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(seconds=60),
         )
@@ -221,8 +127,6 @@ class Orchestrator:
             {
                 **metadata,
                 'opc_servers': opc_servers,
-                'connectors': connectors,
-                'connector_tags': connector_tags,
                 'active_ingestors': active_ingestors,
                 'pipelines': pipeline_config,
             },

pyproject.toml

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

tests/orchestrator/activities/test_email.py

@@ -85,9 +85,7 @@ def test_close(sientia_monitoring_mock, email):
     email.close()
 
     email.server.quit.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
+    sientia_monitoring_mock.shutdown.assert_called_once()
 
 
 def test_build_email_html(email):

tests/orchestrator/activities/test_formatters.py

@@ -218,395 +218,110 @@ def test_process_schedules_with_invalid_workflow_type(formatters):
     mock_scouter.assert_any_call(pipelines[2])
 
 
-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},
-            ],
-        ),
-    ]
+@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):
     input_data = {
-        'opc_servers': slot_opc_servers,
-        'active_ingestors': ['heartbeat:ingestor:1', 'heartbeat:ingestor:2'],
-        'pipelines': pipelines,
+        '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',
         **metadata,
     }
 
-    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 == {
+    expected_opc_servers = {
         '1': {
-            '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',
-                            },
-                        ],
-                    },
-                },
-            }
+            'id': '1',
+            'server_name': 'test_server_name',
+            'url': 'test_url',
+            'uri': 'test_uri',
         },
         '2': {
-            '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',
-                            },
-                        ],
-                    },
+            'id': '2',
+            'server_name': 'test_server_name2',
+            'url': 'test_url2',
+            'uri': 'test_uri2',
         },
     }
-        },
-    }
-    formatters.send_notification.assert_not_called()
 
-
-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,
+    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,
         }
+    }
+
+    mock_build_tag_config.return_value = (slot_mock, ['2'])
 
     result = formatters.process_slots(input_data)
 
-    assert list(result) == ['1']
-    assert list(result['1']['kepserver']['tags']) == ['ns=2;s=A']
+    tags = list(mock_gather_read_tags.return_value.values())
 
+    mock_gather_read_tags.assert_called_once_with(input_data['pipelines'])
 
-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',
+    mock_build_tag_config.assert_has_calls(
         [
-                {'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'},
-            ],
-        ),
+            call(tags[:2], expected_opc_servers),
+            call(tags[2:], expected_opc_servers),
         ]
-    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 9 not found in opc_servers',
+                message='Servers 2 not found in opc_servers',
                 block='orchestrator',
                 level=NotificationLevel.ERROR,
             ),
             call(
                 metadata=metadata['metadata'],
-                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'
-                ),
+                notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
+                message='Servers 2 not found in opc_servers',
                 block='orchestrator',
                 level=NotificationLevel.ERROR,
             ),
         ]
     )
-    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,
+    assert result == {
+        '1': slot_mock,
+        '2': slot_mock,
     }
 
-    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 = {
@@ -695,31 +410,6 @@ 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,

tests/orchestrator/activities/test_mongo_db.py

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

tests/orchestrator/utils/test_orchestrator_functions.py

@@ -5,17 +5,12 @@ 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,
 )
@@ -316,562 +311,249 @@ def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_confi
     assert result == expected
 
 
-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,
-    }
-
-
-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"""
+def test_gather_read_tags():
     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',
+            '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'],
         },
-            '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},
-    ]
+    assert result == expected
 
 
 def test_gather_read_tags_filters_non_scouter_workflows():
-    """Only scouter pipelines feed the slots"""
+    """Test that gather_read_tags only processes scouter workflow types and ignores others"""
     pipelines = [
-        _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'),
+        {
+            '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',
+                }
+            ],
+        },
     ]
 
-    entries, _issues = _gather(pipelines)
+    result = gather_read_tags(pipelines)
 
-    assert [e['address'] for e in entries] == ['ns=2;s=A']
+    # 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'],
+        },
+    }
 
-
-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()},
-    )
+    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_build_tag_config():
-    """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)
-
-    result = build_tag_config(entries)
-
-    assert result == {
-        'kepserver': {
+    tags = [
+        {
             'server_id': '1',
-            'name': 'kepserver',
-            'url': 'opc.tcp://kep:49320',
-            'server_uri': 'urn:kep',
-            'use_security': True,
+            '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,
+        },
+    ]
+
+    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(tags, opc_servers)
+
+    expected = {
+        '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,
+            '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,
             'subscription_period_ms': 150,
             'tags': {
-                'ns=2;s=A': {
-                    'frequency': 1000,
-                    'active_read': {'enabled': False},
-                    'names': [_name('sin', 'a', 'm1')],
-                },
-                'ns=2;s=B': {
-                    'frequency': 300,
-                    'active_read': {'enabled': False},
-                    'names': [_name('sin', 'b', 'm1')],
-                },
-            },
-        },
-        'simulator': {
+                'test_tag_address2': {
                     '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')],
+                    'server_name': 'test_server_name2',
+                    'tag_address': 'test_tag_address2',
+                    'frequency': 1000,
                 },
-            },
-        },
-        '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')],
+                'test_tag_address3': {
+                    'server_id': '2',
+                    'server_name': 'test_server_name2',
+                    'tag_address': 'test_tag_address3',
+                    'frequency': 300,
                 },
             },
         },
     }
 
-
-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())
+    assert result == (expected, ['3'])
 
 
 def test_base_scouter():

tests/orchestrator/workflows/test_orchestrator.py

@@ -1,14 +1,9 @@
-from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
+from unittest.mock import ANY, AsyncMock, call, patch
 
 from pytest import fixture, mark
 
 from orchestrator.activities.activities import Activities
-from orchestrator.workflows.orchestrator import (
-    DEFAULT_CONNECTOR_TAGS_QUERY,
-    DEFAULT_CONNECTORS_QUERY,
-    TAG_DATABASE_PATCH,
-    Orchestrator,
-)
+from orchestrator.workflows.orchestrator import Orchestrator
 
 
 @fixture
@@ -27,10 +22,8 @@ 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, referenced_tag_ids_mock, orchestrator):
-    workflow_mock.patched = MagicMock(return_value=True)
+async def test_run(workflow_mock, orchestrator):
     input_data = {
         'pipelines_query': 'SELECT * FROM bucket',
         'opc_servers_query': 'SELECT * FROM servers',
@@ -116,54 +109,11 @@ async def test_run(workflow_mock, referenced_tag_ids_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,
-                    'connector_tags': workflow_mock.start_local_activity_method.return_value,
-                },
+                {**metadata, 'pipelines': workflow_mock.start_local_activity_method.return_value},
                 retry_policy=ANY,
                 start_to_close_timeout=ANY,
             )
@@ -177,8 +127,6 @@ async def test_run(workflow_mock, referenced_tag_ids_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,
                 },
@@ -398,96 +346,3 @@ async def test_run(workflow_mock, referenced_tag_ids_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'] == []