SIENTIAPDE-1974

SIENTIAPDE-1974  Split orchestrator_functions into pipeline configs, tag database and OPC slots
- orchestrator_functions.py held four unrelated concerns (pipeline
  builders, tag database indexes, PI Web API resolution, OPC slot v2)
  and had grown to 1135 lines. Move them, unchanged, into:
  pipeline_configs.py, tag_database.py and opc_slots.py.
- Make the helpers shared across modules public: is_complete,
  tag_id_text and pi_connection.
- Split the utils tests the same way, with the pipeline and tag
  database documents they share in tag_database_fixtures.py.
- Point formatters, the orchestrator workflow, README and CLAUDE.md at
  the new modules.

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

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`); `pipeline_configs.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `tag_database.py` indexes the tag database (`connectors` + `connector_tags`) and resolves tags by `tag_id` (OPC UA names, PI Web API WebIDs and connection); `opc_slots.py` builds the OPC slot v2 the gateway reads (`gather_read_tags` / `build_tag_config`, contract C1: one entry per address with a `names[]` list; see the README section "OPC slots"). `resolve_tag_names` (in `tag_database.py`) is the single source of tag names for both the scouter's `model_tags` and the slot's `names` (SIENTIAPDE-1974 SC1); `converters.py` handles Temporal schedule frequency parsing.
 
 ### Temporal-specific conventions (all workflow/activity files follow this)
 - Every workflow and activity-consuming module wraps non-deterministic/side-effecting imports in `with workflow.unsafe.imports_passed_through(): ...` right after `from temporalio import workflow`, because Temporal's sandbox restricts what workflow code can import directly. When adding a new workflow file, follow the existing pattern (see `orchestrator/workflows/orchestrator.py`, `orchestrator/activities/activities.py`) rather than importing at module top level.

README.md

@@ -959,7 +959,9 @@ orchestrator/
 ├── utils/                   # Utility functions
 │   ├── connectors_config.py # Database configuration
 │   ├── email_builder.py    # Email template generation
-│   ├── orchestrator_functions.py # Pipeline utilities
+│   ├── pipeline_configs.py # Schedule input builders per pipeline type
+│   ├── tag_database.py     # Tag database indexes, OPC names and PI Web API resolution
+│   ├── opc_slots.py        # OPC slot v2 (contract C1)
 │   ├── converters.py       # Data type conversion utilities
 │   └── templates/          # Email HTML templates
 │       ├── email_template.html    # Report email template

orchestrator/activities/formatters.py

@@ -17,24 +17,25 @@ with workflow.unsafe.imports_passed_through():
         now,
     )
 
-    from orchestrator.utils.orchestrator_functions import (
-        PI_OUTPUT_WORKFLOW_TYPES,
-        build_tag_config,
+    from orchestrator.utils.opc_slots import build_tag_config, gather_read_tags
+    from orchestrator.utils.pipeline_configs import (
         chunk_period_warning,
         drift,
         frequency_alert_warning,
-        gather_read_tags,
-        index_connector_tags,
-        index_connectors,
-        index_pi_connectors,
         minimal_retrain,
         pi_web_api_scouter,
         predictions_batch,
+        scouter,
+        simple_metrics,
+    )
+    from orchestrator.utils.tag_database import (
+        PI_OUTPUT_WORKFLOW_TYPES,
+        index_connector_tags,
+        index_connectors,
+        index_pi_connectors,
         resolve_pi_output_tags,
         resolve_pi_read_tags,
         resolve_tag_names,
-        scouter,
-        simple_metrics,
     )
 
 topic_separator = '\n ========== \n'

orchestrator/utils/opc_slots.py

@@ -0,0 +1,282 @@
+"""OPC slot v2 (contract C1): read tags grouped by server and address for the gateway."""
+
+from typing import Any, cast
+
+from orchestrator.utils.tag_database import is_complete, resolve_tag_names, tag_id_text
+
+
+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_text(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_text(tag) is not None:
+        return False
+    return not is_complete(tag, ('server_id', 'tag_address', 'tag_name'))
+
+
+def _merge_active_read(current: dict[str, Any], requested: dict[str, Any] | None) -> dict[str, Any]:
+    """Active read of an address: on if any pipeline asks, with the lowest cycle_multiple (D3)."""
+    if not requested or not requested.get('enabled'):
+        return current
+    cycle_multiple = int(requested.get('cycle_multiple', 1))
+    if current.get('enabled'):
+        cycle_multiple = min(cycle_multiple, current['cycle_multiple'])
+    return {'enabled': True, 'cycle_multiple': cycle_multiple}
+
+
+def _assign_server_keys(servers: dict[str, dict[str, Any]]) -> None:
+    """
+    Give each server a unique slot key: its name, or name:server_id (extended until
+    free) when another server already took it. Assigned in server_id order, so the keys
+    are the same on every run and in every slot.
+    """
+    taken: set[str] = set()
+    for server_id in sorted(servers):
+        server = servers[server_id]
+        key = server['name']
+        while key in taken:
+            key = f'{key}:{server_id}'
+        server['key'] = key
+        taken.add(key)
+
+
+def _add_read_tag(
+    entries: dict[tuple[str, str], dict[str, Any]],
+    servers: dict[str, dict[str, Any]],
+    pipeline: dict[str, Any],
+    tag: dict[str, Any],
+    server: dict[str, Any],
+    address: str,
+) -> None:
+    """Merge one resolved read tag into its (server, address) entry and its server."""
+    server_id = server['server_id']
+    known = servers.get(server_id)
+    if known is None or (server['from_tag_database'] and not known['from_tag_database']):
+        servers[server_id] = server
+
+    # int, so the slot does not depend on how each pipeline stored the number
+    frequency = cast(int, _frequency(tag))
+    entry = entries.setdefault(
+        (server_id, address),
+        {
+            'address': address,
+            'frequency': frequency,
+            'active_read': {'enabled': False},
+            'names': [],
+        },
+    )
+    entry['frequency'] = min(entry['frequency'], frequency)
+    entry['active_read'] = _merge_active_read(entry['active_read'], tag.get('active_read'))
+
+    name = {
+        'collection': f'raw_{pipeline["schedule_name"]}',
+        'tag_name': tag['tag_name'],
+        'model_id': pipeline['model_id'],
+        'model_name': pipeline['model']['name'],
+    }
+    if name not in entry['names']:
+        entry['names'].append(name)
+
+
+def gather_read_tags(
+    pipelines: list[dict[str, Any]],
+    opc_servers: dict[str, dict[str, Any]],
+    connectors: dict[str, dict[str, Any]],
+    tags_by_id: dict[str, dict[str, Any]],
+) -> tuple[list[dict[str, Any]], dict[str, set[str]]]:
+    """
+    Group the read tags of all scouter pipelines by (server, address) for the slot v2.
+
+    An address read by N pipelines becomes ONE entry with N items in names[]: the
+    gateway reads it once and writes it to N collections (SIENTIAPDE-1974, RF2). Each
+    read tag is resolved through the tag database when it has a tag_id (RF3), or through
+    the current server_id + tag_address + opc-servers fields when it has not (RF4).
+
+    A server known through both connectors and opc-servers under the same server_id is
+    the same server (the gateway keys its connections by server_id); the connectors
+    fields win.
+
+    Args:
+        pipelines (list[dict[str, Any]]): Pipeline configurations. Only workflow_type
+            'scouter' is read; each needs schedule_name, model_id, model.name and
+            read_tags (with frequency, and optionally tag_id and active_read)
+        opc_servers (dict[str, dict[str, Any]]): opc-servers documents keyed by str(id)
+        connectors (dict[str, dict[str, Any]]): Active OPC UA connectors indexed by
+            index_connectors
+        tags_by_id (dict[str, dict[str, Any]]): Active Read connector_tags indexed by
+            index_connector_tags
+
+    Returns:
+        tuple[list[dict[str, Any]], dict[str, set[str]]]:
+            - One entry per (server, address), sorted by server_id then address, with:
+                - server (dict): server_id, name, url, server_uri, use_security,
+                  subscription_period_ms (None = derive from the frequencies) and key
+                  (the server's unique slot key)
+                - address (str): The OPC node address
+                - frequency (int): The lowest frequency among the pipelines (RF6)
+                - active_read (dict): {'enabled': False} unless a pipeline enables it (D3)
+                - names (list[dict]): collection ('raw_{schedule_name}'), tag_name,
+                  model_id and model_name, one per pipeline name, sorted
+            - What was left out: 'tags' (tag_id), 'connectors' (connector_id), 'servers'
+              (opc-servers server_id) and 'invalid' (schedule_name/read tag without a
+              usable frequency or legacy fields)
+    """
+    issues: dict[str, set[str]] = {
+        'tags': set(),
+        'connectors': set(),
+        'servers': set(),
+        'invalid': set(),
+    }
+    servers: dict[str, dict[str, Any]] = {}
+    entries: dict[tuple[str, str], dict[str, Any]] = {}
+
+    for pipeline in pipelines:
+        if pipeline['workflow_type'] != 'scouter':
+            continue
+
+        for tag in resolve_tag_names(pipeline.get('read_tags') or [], tags_by_id):
+            if _invalid_read_tag(tag):
+                ref = tag_id_text(tag) or tag.get('tag_address') or tag.get('tag_name')
+                issues['invalid'].add(f'{pipeline["schedule_name"]}/{ref}')
+                continue
+
+            located = _locate_read_tag(tag, opc_servers, connectors, tags_by_id, issues)
+            if located is None:
+                continue
+            _add_read_tag(entries, servers, pipeline, tag, *located)
+
+    _assign_server_keys(servers)
+
+    gathered = []
+    for (server_id, _address), entry in sorted(entries.items()):
+        # A stable order keeps the slot identical between runs, so the gateway does
+        # not re-subscribe an unchanged slot (O6)
+        entry['names'].sort(key=lambda n: (n['collection'], n['tag_name'], str(n['model_id'])))
+        gathered.append({'server': servers[server_id], **entry})
+
+    return gathered, issues
+
+
+def build_tag_config(entries: list[dict[str, Any]]) -> dict[str, Any]:
+    """
+    Build one slot in the slot v2 format (contract C1) from address entries.
+
+    The slot no longer carries cert_path, private_key_path or server_cert_path: the
+    gateway takes the certificate from MinIO (SIENTIAPDE-1974, O4).
+
+    Args:
+        entries (list[dict[str, Any]]): Address entries from gather_read_tags for this slot
+
+    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.
+    """
+    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'],
+                'tags': {},
+            }
+
+        slot_config[server['key']]['tags'][entry['address']] = {
+            'frequency': entry['frequency'],
+            'active_read': entry['active_read'],
+            'names': entry['names'],
+        }
+
+    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
+
+    return slot_config

orchestrator/utils/pipeline_configs.py

@@ -0,0 +1,513 @@
+"""Pipeline configuration builders: one Temporal schedule input per pipeline type."""
+
+from math import ceil
+from typing import Any
+
+from orchestrator.utils.converters import parse_frequency
+from orchestrator.utils.tag_database import PI_QUERY_DEFAULTS, pi_connection
+
+
+def common_config(config: dict[str, Any]):
+    """
+    Extract common configuration parameters from a pipeline configuration.
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - workflow_type (str): Type of workflow (e.g., 'scouter', 'predictions_batch', 'drift')
+            - schedule_name (str): Unique name identifier for this schedule
+            - model_id (str): MongoDB ID of the associated model
+            - model (dict): Model configuration containing:
+                - name (str): Human-readable name of the model
+                - model_config (dict, optional): Additional model-specific configuration
+            - frequency (str, optional): Execution frequency (default: '1m')
+                Format: '{number}{unit}' where unit is 's', 'm', 'h', or 'd'
+            - offset (str, optional): Schedule offset/delay (default: '0m')
+            - max_retry_policy (int, optional): Maximum retry attempts on failure (default: 1)
+            - execution_timeout_seconds (int, optional): Workflow execution timeout (default: 300)
+            - task_timeout_seconds (int, optional): Individual task timeout (default: 300)
+
+    Returns:
+        dict[str, Any]: Common configuration dictionary with standardized parameters
+            for Temporal workflow execution
+    """
+    model = config['model']
+    return {
+        'workflow_type': config['workflow_type'],
+        'schedule_name': config['schedule_name'],
+        'frequency': config.get('frequency', '1m'),
+        'offset': config.get('offset', '0m'),
+        'max_retry_policy': config.get('max_retry_policy', 1),
+        'model_id': config['model_id'],
+        'model_name': model['name'],
+        'model_config': model.get('model_config', {}),
+        'execution_timeout_seconds': config.get('execution_timeout_seconds', 300),
+        'task_timeout_seconds': config.get('task_timeout_seconds', 300),
+        'on_conflict': config.get('on_conflict', 'error'),
+        'runtime': config.get('runtime', 'legacy'),
+    }
+
+
+# Kept in sync by hand with the `chunk_period` validation in
+# sientia-dataops-laborious/laborious/activities/model_metrics.py (no shared contract).
+DRIFT_CHUNK_PERIODS = ('min', 's')
+
+
+def _drift_interval_minutes(config: dict[str, Any], frequency: Any) -> int:
+    """
+    Drift window in minutes: `interval_minutes` as configured, otherwise the schedule frequency
+    rounded up to whole minutes (min 1). Consecutive default windows do not overlap, so each
+    drift chunk is stored once.
+    """
+    if 'interval_minutes' in config:
+        return config['interval_minutes']
+    try:
+        seconds = parse_frequency(str(frequency))
+    except ValueError:
+        # create_schedules fails this schedule on its own; don't kill the tick for every model.
+        return 60
+    return max(1, ceil(seconds / 60))
+
+
+def drift(config: dict[str, Any]):
+    """
+    Build drift configuration from pipeline config.
+
+    Creates a drift detection workflow configuration with database table mappings
+    and drift metric specifications for monitoring data distribution changes.
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - interval_minutes (int, optional): Time window for data comparison in minutes,
+                used as-is. Default: `frequency` rounded up to whole minutes (min 1), so
+                consecutive windows do not overlap (1 with the default '1m');
+                `interval_minutes: 60` keeps the legacy window.
+            - chunk_period (str, optional): Drift bucket size, 'min' or 's' (default: 'min').
+                Any other value falls back to 'min' and create_schedule_config warns.
+            - drift_metrics (list[str], optional): Statistical metrics to compute (default:
+                ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein'])
+            - Additional fields from common_config
+
+    Returns:
+        dict[str, Any]: Drift detection configuration with source/target tables,
+            time interval, chunk period and metrics specifications. Results stored in
+            'drift_metrics' table
+    """
+    base = common_config(config)
+    configured_chunk_period = config.get('chunk_period')
+    chunk_period_valid = configured_chunk_period in DRIFT_CHUNK_PERIODS
+    output = {
+        **base,
+        'schema': 'sientia_data',
+        'source_table_name': 'laborious_data',
+        'target_table_name': 'drift_metrics',
+        'interval': _drift_interval_minutes(config, base['frequency']),
+        'chunk_period': configured_chunk_period if chunk_period_valid else 'min',
+        'drift_metrics': config.get(
+            'drift_metrics', ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']
+        ),
+    }
+    if configured_chunk_period is not None and not chunk_period_valid:
+        # Forwarded so create_schedule_config can warn without re-reading the pipeline.
+        output['chunk_period_configured'] = configured_chunk_period
+    return output
+
+
+def chunk_period_warning(chunk_period: Any) -> str | None:
+    """
+    Warning message when a configured drift `chunk_period` is not supported by laborious,
+    otherwise None. Warn only: `drift()` already replaced it with 'min'.
+    """
+    if chunk_period is None or chunk_period in DRIFT_CHUNK_PERIODS:
+        return None
+    return (
+        f"chunk_period={chunk_period!r} is not one of {DRIFT_CHUNK_PERIODS}; falling back to 'min'"
+    )
+
+
+def simple_metrics(config: dict[str, Any]):
+    """
+    Build simple metrics configuration from pipeline config.
+
+    Creates a simple metrics computation workflow configuration for calculating
+    model performance metrics like RMSE, MSE, MAE, and R².
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - interval_minutes (int, optional): Computation interval in minutes (default: 60)
+            - metrics (list[str], optional): List of metrics to compute
+                (default: ['rmse', 'mse', 'mae', 'r2'])
+            - frequency_alert (str, optional): Frequency used while `adaptive_state.escalated`
+                is true; same format as `frequency`. Absent = adaptive frequency off.
+            - adaptive_state (dict, optional): `{escalated, healthy_streak}`, written by laborious.
+            - Additional fields from common_config
+
+    Returns:
+        dict[str, Any]: Simple metrics configuration with source tables (predictions and
+            actual data), target table, time interval, and metrics list. Results stored in
+            'simple_metrics' table
+    """
+    base = common_config(config)
+    frequency_alert = config.get('frequency_alert')
+    # SIENTIAPDE-1995: laborious flips `adaptive_state.escalated` on a threshold breach (and back
+    # after the window is renewed without one) and bumps `updated_at`, which makes this tick
+    # rewrite the schedule.
+    state = config.get('adaptive_state')
+    # Hand-written docs: a malformed state must not raise and kill the tick for every model.
+    if frequency_alert and isinstance(state, dict) and state.get('escalated') is True:
+        try:
+            # Temporal parses the schedule frequency on create/update; an unparseable value
+            # would fail it, so keep `frequency` (create_schedule_config still warns).
+            parse_frequency(str(frequency_alert))
+            base['frequency'] = frequency_alert
+        except ValueError:
+            pass
+    return {
+        **base,
+        'schema': 'sientia_data',
+        'predictions_table_name': 'predictions',
+        'data_table_name': 'laborious_data',
+        'target_table_name': 'simple_metrics',
+        'interval_minutes': config.get('interval_minutes', 60),
+        'metrics': config.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
+        # Forwarded so create_schedule_config can warn on a value below the floor.
+        'frequency_alert': frequency_alert,
+    }
+
+
+MIN_FREQUENCY_ALERT_SECONDS = 60
+
+
+def frequency_alert_warning(frequency_alert: str | None) -> str | None:
+    """
+    Warning message when `frequency_alert` is below the 1-minute floor or does not parse,
+    otherwise None. Warn only (SIENTIAPDE-1995 decision 7): a value below the floor is kept;
+    an unparseable one is ignored by `simple_metrics()`, which keeps `frequency`.
+    """
+    if not frequency_alert:
+        return None
+    try:
+        # str(): a non-string value (e.g. a number typed into the doc) must warn, not raise
+        # AttributeError and fail the whole create_schedule_config activity.
+        seconds = parse_frequency(str(frequency_alert))
+    except ValueError:
+        return f'frequency_alert={frequency_alert!r} is not a valid frequency'
+    if seconds < MIN_FREQUENCY_ALERT_SECONDS:
+        return f'frequency_alert={frequency_alert} is below the 1m floor; kept as configured'
+    return None
+
+
+def minimal_retrain(config: dict[str, Any]):
+    """
+    Build minimal retrain configuration from pipeline config.
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - query (str): SQL query to retrieve training data. Should return features
+                and target variable in expected format
+            - datetime_columns (list[str], optional): Column names to parse as datetime
+                for proper temporal handling (default: [])
+            - Additional fields from common_config
+
+    Returns:
+        dict[str, Any]: Minimal retrain configuration with SQL query, database settings,
+            and datetime column specifications. Retraining logs are stored in 'log_retrain' table
+    """
+    return {
+        **common_config(config),
+        'query': config['query'],
+        'schema': 'sientia_data',
+        'table_name': 'log_retrain',
+        'datetime_columns': config.get('datetime_columns', []),
+    }
+
+
+def base_scouter(config: dict[str, Any]):
+    """
+    Build base scouter configuration shared by all scouter workflow types.
+
+    Creates the foundational configuration for OPC data collection workflows,
+    including filter policies, database settings, and data retention parameters.
+    This configuration is extended by specific scouter implementations (OPC UA, PI Web API).
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - filters (list[dict], optional): List of filter configurations with:
+                - filter_name (str): Name of the filter
+                - policy (str): Filter policy to apply
+            - tag_retention_minutes (int, optional): Tag retention time in minutes (default: 60)
+            - debug_data_package (bool, optional): Enable debug data package logging (default: False)
+            - fill_missing_tags (bool, optional): Fill missing tags with interpolation (default: False)
+            - Additional fields from common_config
+
+    Returns:
+        dict[str, Any]: Base scouter configuration with filters, database settings,
+            and retention policies
+    """
+
+    filters = {}
+    for f in config.get('filters', []):
+        filters[f['filter_name']] = {'policy': f['policy']}
+
+    return {
+        **common_config(config),
+        'trigger_laborious': False,
+        'filters': filters,
+        'schema': 'sientia_data',
+        'table_name': 'laborious_data',
+        'retention_time': config.get('tag_retention_minutes', 60) * 60,
+        'debug_data_package': config.get('debug_data_package', False),
+        'fill_missing_tags': config.get('fill_missing_tags', False),
+    }
+
+
+def scouter(config: dict[str, Any]):
+    """
+    Build scouter configuration from pipeline config.
+
+    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
+                - 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])
+            - Additional fields from base_scouter
+
+    Returns:
+        dict[str, Any]: OPC UA scouter configuration with Kafka topic, tag mappings,
+            filters, and retention settings. Topic name follows pattern: 'raw_{schedule_name}'
+    """
+
+    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]),
+        }
+
+    return {
+        **base_scouter(config),
+        'topic': f'raw_{config["schedule_name"]}',
+        'model_tags': tags,
+    }
+
+
+def pi_web_api_scouter(config: dict[str, Any]):
+    """
+    Build PI Web API scouter configuration from pipeline config.
+
+    Creates a data collection workflow configuration for OSIsoft PI servers using
+    the PI Web API REST interface. Configures tag mappings with WebIDs, aggregation
+    functions, and API query parameters including timeout management.
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - read_tags (dict[str, dict]): Tag configurations keyed by model tag name, already
+              passed through resolve_pi_read_tags (process_schedules does it), with:
+                - webid (str): PI Web API WebID for the tag
+                - aggr_func (str, optional): Aggregation function (default: 'lts')
+                - data_range (list[int], optional): Valid data range (default: [-100, 100])
+            - pi_web_api_config (dict): PI Web API query settings with:
+                - endpoint (str): PI Web API endpoint URL
+                - period (str, optional): Time period for data retrieval (default: '*-1d')
+                - max_count (int, optional): Maximum number of values to retrieve (default: 1)
+                - api_timeout (int, optional): API request timeout in seconds
+            - pi_web_api_connector (dict, optional): The tag database connector of the
+              pipeline's resolved tags (set by process_schedules). Its pi_web_api period,
+              max_count and api_timeout are defaults under pi_web_api_config
+            - Additional fields from base_scouter
+
+    Returns:
+        dict[str, Any]: PI Web API scouter configuration with tag mappings and query settings.
+            API timeout is automatically adjusted to not exceed workflow frequency. With a
+            connector, it also carries connection: {connector_id, base_url, auth} (auth holds
+            the secret reference only, never the credential).
+    """
+    tags = {}
+    for tag, tag_config in config['read_tags'].items():
+        tags[tag] = {
+            'webid': tag_config['webid'],
+            'aggr_func': tag_config.get('aggr_func', 'lts'),
+            'data_range': tag_config.get('data_range', [-100, 100]),
+        }
+
+    base_config = base_scouter(config)
+
+    connector = config.get('pi_web_api_connector')
+    pi_web_api_config = config['pi_web_api_config']
+    if connector:
+        connector_defaults = connector['pi_web_api']
+        pi_web_api_config = {
+            **{
+                key: connector_defaults[key]
+                for key in PI_QUERY_DEFAULTS
+                if key in connector_defaults
+            },
+            **pi_web_api_config,
+        }
+
+    config_timeout = pi_web_api_config.get('api_timeout', None)
+    frequency = parse_frequency(base_config['frequency'])
+
+    if config_timeout is None or config_timeout > frequency:
+        config_timeout = frequency
+
+    output = {
+        **base_config,
+        'model_tags': tags,
+        'pi_web_api_query': {
+            'endpoint': pi_web_api_config['endpoint'],
+            'period': pi_web_api_config.get('period', '*-1d'),
+            'max_count': pi_web_api_config.get('max_count', 1),
+            'api_timeout': config_timeout,
+        },
+    }
+    if connector:
+        output['connection'] = pi_connection(connector)
+    return output
+
+
+def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[str, Any]]):
+    """
+    Merge filter configurations with base filter configuration.
+
+    Extends the base filter configuration dictionary by adding or overwriting
+    filters from the provided configuration list. Used in predictions_batch
+    workflows to combine default filters with user-defined custom filters.
+
+    Args:
+        base_filter_config (dict[str, Any]): Base filter configuration dictionary to extend.
+            Each filter entry contains 'policy' and optionally 'config' keys.
+        config (list[dict[str, Any]]): List of filter configurations to merge, each containing:
+            - filter_name (str): Name of the filter to add or update
+            - policy (str): Filter policy (e.g., 'STOP', 'CONTINUE', 'REPEAT')
+            - config (dict, optional): Additional filter-specific configuration
+
+    Returns:
+        dict[str, Any]: Extended filter configuration dictionary with merged filters.
+            Filters from config list overwrite or add to base_filter_config entries.
+    """
+    for fil in config:
+        base_filter_config[fil['filter_name']] = {
+            'policy': fil['policy'],
+            'config': fil.get('config', {}),
+        }
+
+    return base_filter_config
+
+
+def process_path_priority(path_priority: list[str]):
+    """
+    Process and normalize path priority list for filter policy execution order.
+
+    Validates and normalizes the priority list used to determine the order in which
+    filter policies are evaluated in prediction workflows. Invalid priorities are
+    removed, missing required priorities are appended, and the list is truncated to
+    exactly 3 elements.
+
+    Valid priorities define workflow behavior when filters are triggered:
+        - STOP: Halt workflow execution immediately
+        - CONTINUE: Proceed to next step despite filter trigger
+        - REPEAT: Retry the current step
+
+    Args:
+        path_priority (list[str]): User-provided list of path priorities. May contain
+            invalid values or be incomplete.
+
+    Returns:
+        list[str]: Normalized path priority list with exactly 3 elements in user-specified
+            or default order. Default order when priorities are missing: ["STOP", "CONTINUE", "REPEAT"]
+    """
+    for priority in path_priority[:]:
+        if priority not in ['STOP', 'CONTINUE', 'REPEAT']:
+            path_priority.remove(priority)
+
+    for priority in ['STOP', 'CONTINUE', 'REPEAT']:
+        if priority not in path_priority:
+            path_priority.append(priority)
+
+    return path_priority[0:3]
+
+
+def predictions_batch(config: dict[str, Any]):
+    """
+    Build predictions batch configuration from pipeline config.
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - query (str): SQL query to retrieve input data for predictions
+            - write_tags (list[dict]): List of OPC tag configurations for write-back with:
+                - server_id (str): ID of the target OPC server
+                - type (str): Tag type - 'prediction' (model output) or 'confidence' (prediction confidence)
+                - addr (str): OPC tag address/path
+                - data_type (str, optional): OPC data type (default: 'float')
+            - datetime_columns (list[str], optional): Column names to parse as datetime (default: [])
+            - path_priority (list[str], optional): Filter policy execution order (default: ["STOP", "CONTINUE", "REPEAT"])
+            - input_filters (list[dict], optional): Input data validation filters
+            - mlflow_transform_filters (list[dict], optional): Transform stage filters
+            - mlflow_predict_filters (list[dict], optional): Prediction stage filters
+            - model_retention_minutes (int, optional): Data retention time in minutes (default: 60)
+            - save_transform (bool, optional): Save transformed data to database (default: True)
+            - predictions_storage_policy (str, optional): Prediction storage policy (default: 'lts:1')
+            - pi_web_api_output_config (dict, optional): PI Web API output configuration for write-back (default: {}),
+              already passed through resolve_pi_output_tags (process_schedules does it)
+            - Additional fields from common_config
+
+    Returns:
+        dict[str, Any]: Complete predictions batch configuration with OPC output mappings,
+            PI Web API output configuration, multi-stage filters, SQL query, and retention policies
+    """
+    tags: dict[str, Any] = {}
+    for tag in config.get('write_tags', []):
+        if tag['server_id'] not in tags:
+            tags[tag['server_id']] = {}
+
+        tag_type = tag['type']
+
+        if tag_type == 'prediction' or tag_type == 'confidence':
+            tag_type_str = f'{tag_type}_tags'
+
+            if tag_type_str not in tags[tag['server_id']]:
+                tags[tag['server_id']][tag_type_str] = {}
+
+            tags[tag['server_id']][tag_type_str][tag['addr']] = {
+                'data_type': tag.get('data_type', 'float'),
+            }
+
+    path_priority = process_path_priority(
+        config.get('path_priority', ['STOP', 'CONTINUE', 'REPEAT'])
+    )
+
+    return {
+        **common_config(config),
+        'query': config['query'],
+        'datetime_columns': config.get('datetime_columns', []),
+        'schema': 'sientia_data',
+        'table_name': 'predictions',
+        'save_transform': config.get('save_transform', True),
+        'transform_table_name': 'transformed_data',
+        'retention_time': config.get('model_retention_minutes', 60) * 60,
+        'opc_output_config': tags,
+        'pi_web_api_output_config': config.get('pi_web_api_output_config', {}),
+        'input_filters': overlap_filter_config(
+            {'EMPTY_DATA': {'policy': 'STOP', 'config': {}}}, config.get('input_filters', [])
+        ),
+        'mlflow_transform_filters': overlap_filter_config(
+            {
+                'EMPTY_DATA': {'policy': 'STOP', 'config': {}},
+                'API_ERROR': {'policy': 'STOP', 'config': {}},
+            },
+            config.get('mlflow_transform_filters', []),
+        ),
+        'mlflow_predict_filters': overlap_filter_config(
+            {'API_ERROR': {'policy': 'STOP', 'config': {}}},
+            config.get('mlflow_predict_filters', []),
+        ),
+        'path_priority': path_priority,
+        'predictions_storage_policy': config.get('predictions_storage_policy', 'lts:1'),
+    }

orchestrator/utils/tag_database.py

@@ -0,0 +1,359 @@
+"""
+Tag database (connectors + connector_tags): which tags the pipelines reference, the
+indexes of what was loaded, and the resolution of OPC UA names and PI Web API tags.
+"""
+
+from typing import Any, cast
+
+# 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 is_complete(document: dict[str, Any], fields: tuple[str, ...]) -> bool:
+    return all(document.get(field) not in (None, '') for field in fields)
+
+
+def tag_id_text(tag: dict[str, Any]) -> str | None:
+    """The read tag's tag_id as text, or None when it has none (missing, null or empty)."""
+    tag_id = tag.get('tag_id')
+    return None if tag_id in (None, '') else str(tag_id)
+
+
+def referenced_tag_ids(pipelines: list[dict[str, Any]]) -> list[Any]:
+    """
+    The tag_ids the scouter pipelines reference, for the connector_tags query.
+
+    Each id goes in both forms, int and text, because the tag database stores int and
+    pipelines may hold either. Sorted, so the query is the same on every run.
+
+    Args:
+        pipelines (list[dict[str, Any]]): Pipeline configurations
+
+    Returns:
+        list[Any]: The referenced tag_ids (empty when no read tag carries a tag_id)
+    """
+    tags = [
+        tag
+        for pipeline in pipelines
+        if pipeline.get('workflow_type') == 'scouter'
+        for tag in pipeline.get('read_tags') or []
+    ]
+    return _query_tag_ids(tags)
+
+
+def _query_tag_ids(tags: list[Any]) -> list[Any]:
+    """The tag_ids of the tags in both forms (int and text), sorted for a stable query."""
+    ids: set[Any] = set()
+    for tag in tags:
+        tag_id = tag_id_text(tag) if isinstance(tag, dict) else None
+        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)))
+
+
+# Workflow types whose pi_web_api_output_config may reference the tag database
+PI_OUTPUT_WORKFLOW_TYPES = ('predictions_batch', 'xgboost_predictions_batch')
+
+
+# connector_tags category each PI Web API output mapping accepts
+PI_OUTPUT_CATEGORIES = {'prediction_tags': 'Write', 'confidence_tags': 'Prediction Confidence'}
+
+
+# Connection defaults a pi_web_api connector may carry for the scouter query
+PI_QUERY_DEFAULTS = ('period', 'max_count', 'api_timeout')
+
+
+def referenced_pi_tag_ids(pipelines: list[dict[str, Any]]) -> list[Any]:
+    """
+    The tag_ids the PI Web API pipelines reference, for the PI connector_tags query.
+
+    Read tags of pi_web_api_scouter pipelines and the prediction_tags / confidence_tags
+    entries of a predictions_batch pi_web_api_output_config given as a list. Same forms
+    and order as referenced_tag_ids.
+
+    Args:
+        pipelines (list[dict[str, Any]]): Pipeline configurations
+
+    Returns:
+        list[Any]: The referenced tag_ids (empty when no PI Web API tag carries a tag_id)
+    """
+    tags: list[Any] = []
+    for pipeline in pipelines:
+        workflow_type = pipeline.get('workflow_type')
+        if workflow_type == 'pi_web_api_scouter':
+            read_tags = pipeline.get('read_tags')
+            if isinstance(read_tags, dict):
+                tags.extend(read_tags.values())
+        elif workflow_type in PI_OUTPUT_WORKFLOW_TYPES:
+            output_config = pipeline.get('pi_web_api_output_config') or {}
+            for key in PI_OUTPUT_CATEGORIES:
+                if isinstance(output_config.get(key), list):
+                    tags.extend(output_config[key])
+    return _query_tag_ids(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 is_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 is_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_text(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 _pi_auth(connector: dict[str, Any]) -> dict[str, Any] | None:
+    """The connector's pi_web_api.auth when it names a type and a secret {name, key}."""
+    block = connector.get('pi_web_api')
+    auth = block.get('auth') if isinstance(block, dict) else None
+    if not isinstance(auth, dict) or auth.get('type') in (None, ''):
+        return None
+    secret = auth.get('secret')
+    if not isinstance(secret, dict) or not is_complete(secret, ('name', 'key')):
+        return None
+    return auth
+
+
+def index_pi_connectors(connectors: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
+    """
+    Index the tag database's pi_web_api connectors by connector_id (as text).
+
+    Documents without connector_id, url or a pi_web_api.auth with type and secret
+    {name, key} are left out, so their tags are notified as connector not found.
+
+    Args:
+        connectors (list[dict[str, Any]]): Active pi_web_api connectors documents
+
+    Returns:
+        dict[str, dict[str, Any]]: connectors documents keyed by str(connector_id)
+    """
+    return {
+        str(connector['connector_id']): connector
+        for connector in connectors
+        if is_complete(connector, ('connector_id', 'url')) and _pi_auth(connector) is not None
+    }
+
+
+def pi_connection(connector: dict[str, Any]) -> dict[str, Any]:
+    """
+    Connection block of a pi_web_api connector for the schedule input.
+
+    Only the secret reference is copied, field by field, so nothing else the document
+    may hold under auth ever reaches the schedule input or the Temporal history (D3).
+    """
+    auth = cast(dict[str, Any], _pi_auth(connector))
+    connection = {
+        'connector_id': str(connector['connector_id']),
+        'base_url': connector['url'],
+        'auth': {
+            'type': auth['type'],
+            'secret': {'name': auth['secret']['name'], 'key': auth['secret']['key']},
+        },
+    }
+    data_server_webid = connector['pi_web_api'].get('data_server_webid')
+    if data_server_webid:
+        connection['data_server_webid'] = data_server_webid
+    return connection
+
+
+def _new_pi_issues() -> dict[str, set[str]]:
+    return {'tags': set(), 'categories': set(), 'connectors': set(), 'mixed': set()}
+
+
+def _locate_pi_tag(
+    tag_id: str,
+    category: str,
+    pi_tags: dict[str, dict[str, Any]],
+    pi_connectors: dict[str, dict[str, Any]],
+    issues: dict[str, set[str]],
+) -> dict[str, Any] | None:
+    """
+    The connector_tags document of a PI Web API tag_id, or None when it cannot be used:
+    unknown tag_id ('tags'), another category ('categories'), or a connector that is not
+    an active, complete pi_web_api connector ('connectors').
+    """
+    connector_tag = pi_tags.get(tag_id)
+    if connector_tag is None:
+        issues['tags'].add(tag_id)
+        return None
+    if connector_tag.get('category') != category:
+        issues['categories'].add(tag_id)
+        return None
+    connector_id = str(connector_tag['connector_id'])
+    if connector_id not in pi_connectors:
+        issues['connectors'].add(connector_id)
+        return None
+    return connector_tag
+
+
+def resolve_pi_read_tags(
+    read_tags: dict[str, dict[str, Any]],
+    pi_tags: dict[str, dict[str, Any]],
+    pi_connectors: dict[str, dict[str, Any]],
+) -> tuple[dict[str, dict[str, Any]], dict[str, Any] | None, dict[str, set[str]]]:
+    """
+    Resolve the read tags of a pi_web_api_scouter pipeline through the tag database.
+
+    An entry with tag_id gets its webid from connector_tags.address (Read category); its
+    key (the model tag name), aggr_func and data_range stay the pipeline's. An entry
+    without tag_id is kept as is (legacy webid). An entry that cannot be resolved is left
+    out. All the resolved tags must belong to one connector (D6).
+
+    Args:
+        read_tags (dict[str, dict[str, Any]]): The pipeline's read_tags
+        pi_tags (dict[str, dict[str, Any]]): PI connector_tags indexed by index_connector_tags
+        pi_connectors (dict[str, dict[str, Any]]): Indexed by index_pi_connectors
+
+    Returns:
+        tuple: the resolved read_tags; the connector of the resolved tags (None when no tag
+            was resolved through the tag database or they span several connectors); and
+            what was left out by kind: 'tags' / 'categories' (tag_id), 'connectors'
+            (connector_id) and 'mixed' (the connector_ids, joined, when there are several)
+    """
+    issues = _new_pi_issues()
+    resolved: dict[str, dict[str, Any]] = {}
+    connector_ids: set[str] = set()
+    for name, tag in read_tags.items():
+        tag_id = tag_id_text(tag)
+        if tag_id is None:
+            resolved[name] = tag
+            continue
+        connector_tag = _locate_pi_tag(tag_id, 'Read', pi_tags, pi_connectors, issues)
+        if connector_tag is None:
+            continue
+        connector_ids.add(str(connector_tag['connector_id']))
+        resolved[name] = {**tag, 'webid': connector_tag['address']}
+
+    if len(connector_ids) > 1:
+        issues['mixed'].add(', '.join(sorted(connector_ids)))
+        return resolved, None, issues
+    connector = pi_connectors[connector_ids.pop()] if connector_ids else None
+    return resolved, connector, issues
+
+
+def resolve_pi_output_tags(
+    output_config: dict[str, Any],
+    pi_tags: dict[str, dict[str, Any]],
+    pi_connectors: dict[str, dict[str, Any]],
+) -> tuple[dict[str, Any], dict[str, set[str]]]:
+    """
+    Resolve the prediction_tags / confidence_tags of a pi_web_api_output_config.
+
+    A mapping given as a list is turned into {tag_name: WebID}: an entry with tag_id takes
+    tag_name and address from connector_tags ('Write' for prediction_tags, 'Prediction
+    Confidence' for confidence_tags); an entry without tag_id is read as a legacy
+    {tag_name: webid} fragment. A mapping already given as a dict passes through unchanged,
+    so the laborious input keeps today's shape.
+
+    Args:
+        output_config (dict[str, Any]): The pipeline's pi_web_api_output_config
+        pi_tags (dict[str, dict[str, Any]]): PI connector_tags indexed by index_connector_tags
+        pi_connectors (dict[str, dict[str, Any]]): Indexed by index_pi_connectors
+
+    Returns:
+        tuple: the resolved output config and what was left out, by kind ('tags',
+            'categories', 'connectors'), as in resolve_pi_read_tags
+    """
+    issues = _new_pi_issues()
+    resolved = dict(output_config)
+    for key, category in PI_OUTPUT_CATEGORIES.items():
+        entries = output_config.get(key)
+        if not isinstance(entries, list):
+            continue
+        mapping: dict[str, Any] = {}
+        for entry in entries:
+            if not isinstance(entry, dict):
+                continue
+            tag_id = tag_id_text(entry)
+            if tag_id is None:
+                mapping.update(entry)
+                continue
+            connector_tag = _locate_pi_tag(tag_id, category, pi_tags, pi_connectors, issues)
+            if connector_tag is not None:
+                mapping[connector_tag['tag_name']] = connector_tag['address']
+        resolved[key] = mapping
+    return resolved, issues

orchestrator/workflows/orchestrator.py

@@ -8,7 +8,7 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.temporal.policies import retry_policy
 
     from orchestrator.activities.activities import Activities
-    from orchestrator.utils.orchestrator_functions import (
+    from orchestrator.utils.tag_database import (
         filter_by_tag_ids,
         referenced_pi_tag_ids,
         referenced_tag_ids,

tests/orchestrator/activities/test_formatters.py

@@ -6,7 +6,7 @@ from pytest import fixture
 from sientia_do.notifications.models import NotificationLevel
 
 from orchestrator.activities.formatters import Formatters
-from orchestrator.utils.orchestrator_functions import pi_web_api_scouter, predictions_batch
+from orchestrator.utils.pipeline_configs import pi_web_api_scouter, predictions_batch
 
 
 @fixture

tests/orchestrator/utils/tag_database_fixtures.py

@@ -0,0 +1,64 @@
+"""Pipeline and tag database documents shared by the utils tests."""
+
+
+def scouter_pipeline_doc(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,
+    }
+
+
+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'},
+}
+
+
+# PI Web API through the tag database (connectors-pi-web-api)
+
+PI_SECRET = {'name': 'pi-web-api-auth-token', 'key': 'token'}
+
+
+def pi_connector_doc(connector_id='pi-usina', **pi_web_api):
+    return {
+        'connector_id': connector_id,
+        'name': 'PI Usina',
+        'url': f'https://{connector_id}.example/piwebapi',
+        'pi_web_api': {'auth': {'type': 'bearer', 'secret': dict(PI_SECRET)}, **pi_web_api},
+    }
+
+
+def pi_tag_doc(tag_id, address, tag_name, category='Read', connector_id='pi-usina'):
+    return {
+        'tag_id': tag_id,
+        'connector_id': connector_id,
+        'address': address,
+        'tag_name': tag_name,
+        'category': category,
+    }
+
+
+def pi_scouter_pipeline_doc(read_tags, **fields):
+    return {
+        'workflow_type': 'pi_web_api_scouter',
+        'schedule_name': 'pi_schedule',
+        'model_id': 'model_id',
+        'model': {'name': 'model_name'},
+        'frequency': '1m',
+        'read_tags': read_tags,
+        'pi_web_api_config': {'endpoint': '/streamsets/recorded'},
+        **fields,
+    }

tests/orchestrator/utils/test_opc_slots.py

@@ -0,0 +1,474 @@
+from orchestrator.utils.opc_slots import (
+    build_tag_config,
+    gather_read_tags,
+)
+from tests.orchestrator.utils.tag_database_fixtures import (
+    TAGS_BY_ID,
+    legacy_tag,
+    scouter_pipeline_doc,
+)
+
+
+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,
+    }
+}
+
+
+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_gather_read_tags_multi_name():
+    """One address read by two pipelines with different names: one entry, two names"""
+    pipelines = [
+        scouter_pipeline_doc('sin', 'm1', [legacy_tag('ns=2;s=A', 'Counter', frequency=15000)]),
+        scouter_pipeline_doc('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 = [
+        scouter_pipeline_doc('sin', 'm1', [legacy_tag('ns=2;s=A', 'Counter')]),
+        scouter_pipeline_doc('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 = [
+        scouter_pipeline_doc(
+            '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 = [
+        scouter_pipeline_doc(
+            '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([scouter_pipeline_doc('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(
+        [scouter_pipeline_doc('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(
+        [scouter_pipeline_doc('sin', 'm1', [{'tag_id': 7, 'frequency': 1000}])]
+    )
+
+    assert entries == [
+        {
+            'server': {
+                'server_id': '3',
+                'name': 'plant',
+                'url': 'opc.tcp://plant:4840',
+                'server_uri': 'urn:plant',
+                'use_security': True,
+                'subscription_period_ms': 1500,
+                'from_tag_database': True,
+                'key': 'plant',
+            },
+            'address': 'ns=3;i=1001',
+            'frequency': 1000,
+            'active_read': {'enabled': False},
+            'names': [_name('sin', 'Pressure', 'm1')],
+        }
+    ]
+
+
+def test_gather_read_tags_pipeline_name_wins_over_tag_database():
+    entries, _issues = _gather(
+        [scouter_pipeline_doc('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(
+        [scouter_pipeline_doc('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(
+        [scouter_pipeline_doc('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(
+        [scouter_pipeline_doc('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 = [
+        scouter_pipeline_doc(
+            'sin', 'm1', [legacy_tag('ns=2;s=A', 'a'), legacy_tag('ns=2;s=B', 'b')]
+        ),
+        scouter_pipeline_doc(
+            '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 = [
+        scouter_pipeline_doc(
+            '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 = [
+        scouter_pipeline_doc('p1', 'm1', [legacy_tag('ns=2;s=A', 'a')]),
+        scouter_pipeline_doc(
+            'p2',
+            'm2',
+            [legacy_tag('ns=2;s=A', 'a', active_read={'enabled': True, 'cycle_multiple': 4})],
+        ),
+        scouter_pipeline_doc(
+            'p3',
+            'm3',
+            [legacy_tag('ns=2;s=A', 'a', active_read={'enabled': True, 'cycle_multiple': 2})],
+        ),
+        scouter_pipeline_doc(
+            'p4',
+            'm4',
+            [legacy_tag('ns=2;s=B', 'b', active_read={'enabled': False, 'cycle_multiple': 1})],
+        ),
+        scouter_pipeline_doc(
+            'p5', 'm5', [legacy_tag('ns=2;s=C', 'c', active_read={'enabled': True})]
+        ),
+    ]
+
+    entries, _issues = _gather(pipelines)
+
+    assert [e['active_read'] for e in entries] == [
+        {'enabled': True, 'cycle_multiple': 2},
+        {'enabled': False},
+        {'enabled': True, 'cycle_multiple': 1},
+    ]
+
+
+def test_gather_read_tags_filters_non_scouter_workflows():
+    """Only scouter pipelines feed the slots"""
+    pipelines = [
+        scouter_pipeline_doc('sin', 'm1', [legacy_tag('ns=2;s=A', 'a')]),
+        scouter_pipeline_doc(
+            'pred', 'm2', [legacy_tag('ns=2;s=P', 'p')], workflow_type='predictions_batch'
+        ),
+        scouter_pipeline_doc(
+            'retrain', 'm3', [legacy_tag('ns=2;s=R', 'r')], workflow_type='minimal_retrain'
+        ),
+        scouter_pipeline_doc('drift', 'm4', [legacy_tag('ns=2;s=D', 'd')], workflow_type='drift'),
+    ]
+
+    entries, _issues = _gather(pipelines)
+
+    assert [e['address'] for e in entries] == ['ns=2;s=A']
+
+
+def test_gather_read_tags_without_read_tags():
+    pipeline = scouter_pipeline_doc('sin', 'm1', [])
+    del pipeline['read_tags']
+
+    assert _gather([pipeline]) == (
+        [],
+        {'tags': set(), 'connectors': set(), 'servers': set(), 'invalid': set()},
+    )
+
+
+def test_build_tag_config():
+    """Slot v2 (contract C1): no certificate paths, subscription period per server"""
+    pipelines = [
+        scouter_pipeline_doc(
+            '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': {
+            'server_id': '1',
+            'name': 'kepserver',
+            'url': 'opc.tcp://kep:49320',
+            'server_uri': 'urn:kep',
+            'use_security': True,
+            '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': {
+            'server_id': '2',
+            'name': 'simulator',
+            'url': 'opc.tcp://sim:4840',
+            'server_uri': 'urn:sim',
+            'use_security': False,
+            'subscription_period_ms': 100,
+            'tags': {
+                'ns=2;s=C': {
+                    'frequency': 200,
+                    'active_read': {'enabled': False},
+                    'names': [_name('sin', 'c', 'm1')],
+                },
+            },
+        },
+        'plant': {
+            'server_id': '3',
+            'name': 'plant',
+            'url': 'opc.tcp://plant:4840',
+            'server_uri': 'urn:plant',
+            'use_security': True,
+            'subscription_period_ms': 1500,
+            'tags': {
+                'ns=3;i=1001': {
+                    'frequency': 100,
+                    'active_read': {'enabled': False},
+                    'names': [_name('sin', 'Pressure', 'm1')],
+                },
+            },
+        },
+    }
+
+
+def test_build_tag_config_empty():
+    assert build_tag_config([]) == {}
+
+
+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(
+        [scouter_pipeline_doc('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 = [
+        scouter_pipeline_doc(
+            '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 = scouter_pipeline_doc('sin', 'm1', [legacy_tag('ns=2;s=A', 'a', frequency=15000)])
+    as_text = scouter_pipeline_doc('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 = [
+        scouter_pipeline_doc(
+            '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())

tests/orchestrator/utils/test_pipeline_configs.py

@@ -0,0 +1,778 @@
+from unittest.mock import call, patch
+
+from pytest import mark
+
+from orchestrator.utils.pipeline_configs import (
+    base_scouter,
+    chunk_period_warning,
+    common_config,
+    drift,
+    frequency_alert_warning,
+    minimal_retrain,
+    overlap_filter_config,
+    pi_web_api_scouter,
+    predictions_batch,
+    process_path_priority,
+    scouter,
+    simple_metrics,
+)
+from tests.orchestrator.utils.tag_database_fixtures import (
+    PI_SECRET,
+    pi_connector_doc,
+    pi_scouter_pipeline_doc,
+)
+
+
+def test_common_config():
+    config = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+    }
+    result = common_config(config)
+    expected = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'frequency': '1m',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'on_conflict': 'error',
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'runtime': 'legacy',
+    }
+    assert result == expected
+
+
+def test_common_config_preserves_explicit_runtime():
+    config = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name'},
+        'runtime': 'tenant-x',
+    }
+    assert common_config(config)['runtime'] == 'tenant-x'
+
+
+def test_drift():
+    config = {
+        'workflow_type': 'drift',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+        'interval_minutes': 120,
+        'drift_metrics': ['kolmogorov_smirnov', 'jensen_shannon'],
+    }
+    result = drift(config)
+    expected = {
+        'workflow_type': 'drift',
+        'schedule_name': 'test_schedule',
+        'frequency': '1m',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'on_conflict': 'error',
+        'schema': 'sientia_data',
+        'source_table_name': 'laborious_data',
+        'target_table_name': 'drift_metrics',
+        'interval': 120,
+        'chunk_period': 'min',
+        'drift_metrics': ['kolmogorov_smirnov', 'jensen_shannon'],
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'runtime': 'legacy',
+    }
+    assert result == expected
+
+
+def _drift_config(**fields):
+    return {
+        'workflow_type': 'drift',
+        'schedule_name': 'm1-drift',
+        'model_id': 'm1',
+        'model': {'name': 'm1'},
+        **fields,
+    }
+
+
+def test_drift_default_window_is_frequency():
+    result = drift(_drift_config())
+
+    assert result['frequency'] == '1m'
+    assert result['interval'] == 1
+
+
+@mark.parametrize(
+    ('frequency', 'interval'),
+    [('1h', 60), ('30s', 1), ('90s', 2)],
+)
+def test_drift_window_derived_from_frequency(frequency, interval):
+    assert drift(_drift_config(frequency=frequency))['interval'] == interval
+
+
+def test_drift_explicit_interval_minutes_wins():
+    result = drift(_drift_config(frequency='1m', interval_minutes=60))
+
+    assert result['interval'] == 60
+
+
+def test_drift_invalid_frequency_falls_back_to_legacy_window():
+    result = drift(_drift_config(frequency='abc'))
+
+    assert result['frequency'] == 'abc'
+    assert result['interval'] == 60
+
+
+def test_drift_chunk_period_defaults_to_min():
+    result = drift(_drift_config())
+
+    assert result['chunk_period'] == 'min'
+    assert 'chunk_period_configured' not in result
+
+
+def test_drift_chunk_period_seconds_kept():
+    result = drift(_drift_config(chunk_period='s'))
+
+    assert result['chunk_period'] == 's'
+    assert 'chunk_period_configured' not in result
+
+
+@mark.parametrize('chunk_period', ['hour', 5])
+def test_drift_invalid_chunk_period_falls_back_to_min(chunk_period):
+    result = drift(_drift_config(chunk_period=chunk_period))
+
+    assert result['chunk_period'] == 'min'
+    assert result['chunk_period_configured'] == chunk_period
+
+
+@mark.parametrize('chunk_period', [None, 'min', 's'])
+def test_chunk_period_warning_none_when_absent_or_valid(chunk_period):
+    assert chunk_period_warning(chunk_period) is None
+
+
+@mark.parametrize('chunk_period', ['hour', 5])
+def test_chunk_period_warning_message_when_invalid(chunk_period):
+    warning = chunk_period_warning(chunk_period)
+
+    assert warning is not None
+    assert repr(chunk_period) in warning
+
+
+def test_simple_metrics():
+    config = {
+        'workflow_type': 'simple_metrics',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+        'interval_minutes': 120,
+        'metrics': ['rmse', 'mse'],
+    }
+    result = simple_metrics(config)
+    expected = {
+        'workflow_type': 'simple_metrics',
+        'schedule_name': 'test_schedule',
+        'frequency': '1m',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'on_conflict': 'error',
+        'schema': 'sientia_data',
+        'predictions_table_name': 'predictions',
+        'data_table_name': 'laborious_data',
+        'target_table_name': 'simple_metrics',
+        'interval_minutes': 120,
+        'metrics': ['rmse', 'mse'],
+        'frequency_alert': None,
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'runtime': 'legacy',
+    }
+    assert result == expected
+
+
+def _simple_metrics_config(**fields):
+    return {
+        'workflow_type': 'simple_metrics',
+        'schedule_name': 'm1-simple_metrics-60m',
+        'model_id': 'm1',
+        'model': {'name': 'm1'},
+        'frequency': '1h',
+        **fields,
+    }
+
+
+def test_simple_metrics_escalated_uses_frequency_alert():
+    config = _simple_metrics_config(
+        frequency_alert='5m', adaptive_state={'escalated': True, 'healthy_streak': 0}
+    )
+    result = simple_metrics(config)
+    assert result['frequency'] == '5m'
+    assert result['frequency_alert'] == '5m'
+
+
+def test_simple_metrics_not_escalated_keeps_frequency():
+    config = _simple_metrics_config(
+        frequency_alert='5m', adaptive_state={'escalated': False, 'healthy_streak': 2}
+    )
+    assert simple_metrics(config)['frequency'] == '1h'
+
+
+def test_simple_metrics_escalated_without_frequency_alert_keeps_frequency():
+    # Removing frequency_alert turns adaptive mode off even if a stale state says escalated.
+    config = _simple_metrics_config(adaptive_state={'escalated': True, 'healthy_streak': 0})
+    assert simple_metrics(config)['frequency'] == '1h'
+
+
+def test_simple_metrics_frequency_alert_without_state_keeps_frequency():
+    assert simple_metrics(_simple_metrics_config(frequency_alert='5m'))['frequency'] == '1h'
+
+
+def test_simple_metrics_escalated_unparseable_frequency_alert_keeps_frequency():
+    # An unparseable frequency would make the Temporal schedule create/update fail.
+    config = _simple_metrics_config(frequency_alert='5min', adaptive_state={'escalated': True})
+    assert simple_metrics(config)['frequency'] == '1h'
+
+
+def test_simple_metrics_escalated_below_floor_frequency_alert_is_applied():
+    # Decision 7: below the floor only warns, the value is kept.
+    config = _simple_metrics_config(frequency_alert='30s', adaptive_state={'escalated': True})
+    assert simple_metrics(config)['frequency'] == '30s'
+
+
+@mark.parametrize('adaptive_state', ['yes', {'escalated': 'false'}])
+def test_simple_metrics_malformed_adaptive_state_keeps_frequency(adaptive_state):
+    config = _simple_metrics_config(frequency_alert='5m', adaptive_state=adaptive_state)
+    assert simple_metrics(config)['frequency'] == '1h'
+
+
+def test_minimal_retrain():
+    config = {
+        'workflow_type': 'minimal_retrain',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+        'query': 'select * from sientia_data.laborious_data order by "timestamp" desc limit 30;',
+        'datetime_columns': ['timestamp'],
+    }
+    result = minimal_retrain(config)
+    expected = {
+        'workflow_type': 'minimal_retrain',
+        'schedule_name': 'test_schedule',
+        'frequency': '1m',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'on_conflict': 'error',
+        'query': 'select * from sientia_data.laborious_data order by "timestamp" desc limit 30;',
+        'schema': 'sientia_data',
+        'table_name': 'log_retrain',
+        'datetime_columns': ['timestamp'],
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'runtime': 'legacy',
+    }
+    assert result == expected
+
+
+def test_scouter():
+    config = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+        'filters': [{'filter_name': 'test_filter_name', 'policy': 'test_policy'}],
+        'read_tags': [
+            {'tag_name': 'test_tag_name', 'aggr_func': 'test_aggr_func', 'data_range': [1, 2]}
+        ],
+        'tag_retention_minutes': 10,
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'runtime': 'legacy',
+    }
+    result = scouter(config)
+    expected = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'frequency': '1m',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'on_conflict': 'error',
+        'topic': 'raw_test_schedule',
+        'trigger_laborious': False,
+        'filters': {'test_filter_name': {'policy': 'test_policy'}},
+        'schema': 'sientia_data',
+        'table_name': 'laborious_data',
+        'retention_time': 10 * 60,
+        'model_tags': {'test_tag_name': {'aggr_func': 'test_aggr_func', 'data_range': [1, 2]}},
+        'debug_data_package': False,
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'fill_missing_tags': False,
+        'runtime': 'legacy',
+    }
+    assert result == expected
+
+
+def test_overlap_filter_config():
+    config = [
+        {'filter_name': 'test_filter_name', 'policy': 'test_policy'},
+        {'filter_name': 'test_filter_name2', 'policy': 'test_policy2'},
+    ]
+    result = overlap_filter_config({'test_filter_name': {'policy': 'test_policy'}}, config)
+    expected = {
+        'test_filter_name': {'policy': 'test_policy', 'config': {}},
+        'test_filter_name2': {'policy': 'test_policy2', 'config': {}},
+    }
+    assert result == expected
+
+
+def test_process_path_priority():
+    config = ['OTHER', 'STOP', 'CONTINUE']
+    result = process_path_priority(config)
+    expected = ['STOP', 'CONTINUE', 'REPEAT']
+    assert result == expected
+
+
+@patch(
+    'orchestrator.utils.pipeline_configs.overlap_filter_config',
+    return_value={'test_filter_name': {'policy': 'test_policy', 'config': {}}},
+)
+@patch(
+    'orchestrator.utils.pipeline_configs.process_path_priority',
+    return_value=['STOP', 'CONTINUE', 'REPEAT'],
+)
+def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_config):
+    config = {
+        'schedule_name': 'test_schedule',
+        'workflow_type': 'predictions_batch',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+        'query': 'test_query',
+        'write_tags': [
+            {'server_id': 'test_server_id', 'type': 'prediction', 'addr': 'test_addr'},
+            {'server_id': 'test_server_id', 'type': 'confidence', 'addr': 'test_addr'},
+        ],
+        'pi_web_api_output_config': {
+            'endpoint': 'test_endpoint',
+            'prediction_tags': {
+                'tag_1': 'webid_1',
+            },
+            'confidence_tags': {
+                'tag_2': 'webid_2',
+            },
+        },
+        'input_filters': [{'filter_name': 'test_filter_name', 'policy': 'test_policy'}],
+        'mlflow_transform_filters': [{'filter_name': 'test_filter_name', 'policy': 'test_policy'}],
+        'mlflow_predict_filters': [{'filter_name': 'test_filter_name', 'policy': 'test_policy'}],
+        'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
+        'datetime_columns': ['timestamp'],
+        'predictions_storage_policy': 'erl:1',
+    }
+
+    result = predictions_batch(config)
+
+    mock_overlap_filter_config.assert_has_calls(
+        [call({'EMPTY_DATA': {'policy': 'STOP', 'config': {}}}, config['input_filters'])]
+    )
+    mock_overlap_filter_config.assert_has_calls(
+        [
+            call(
+                {
+                    'EMPTY_DATA': {'policy': 'STOP', 'config': {}},
+                    'API_ERROR': {'policy': 'STOP', 'config': {}},
+                },
+                config['mlflow_transform_filters'],
+            )
+        ]
+    )
+    mock_overlap_filter_config.assert_has_calls(
+        [call({'API_ERROR': {'policy': 'STOP', 'config': {}}}, config['mlflow_predict_filters'])]
+    )
+    mock_process_path_priority.assert_called_once_with(config['path_priority'])
+
+    expected = {
+        'workflow_type': 'predictions_batch',
+        'schedule_name': 'test_schedule',
+        'frequency': '1m',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'on_conflict': 'error',
+        'query': 'test_query',
+        'schema': 'sientia_data',
+        'table_name': 'predictions',
+        'save_transform': True,
+        'transform_table_name': 'transformed_data',
+        'retention_time': 60 * 60,
+        'opc_output_config': {
+            'test_server_id': {
+                'prediction_tags': {'test_addr': {'data_type': 'float'}},
+                'confidence_tags': {'test_addr': {'data_type': 'float'}},
+            }
+        },
+        'pi_web_api_output_config': {
+            'endpoint': 'test_endpoint',
+            'prediction_tags': {
+                'tag_1': 'webid_1',
+            },
+            'confidence_tags': {
+                'tag_2': 'webid_2',
+            },
+        },
+        'input_filters': {'test_filter_name': {'policy': 'test_policy', 'config': {}}},
+        'mlflow_transform_filters': {'test_filter_name': {'policy': 'test_policy', 'config': {}}},
+        'mlflow_predict_filters': {'test_filter_name': {'policy': 'test_policy', 'config': {}}},
+        'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
+        'datetime_columns': ['timestamp'],
+        'predictions_storage_policy': 'erl:1',
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'runtime': 'legacy',
+    }
+    assert result == expected
+
+
+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_base_scouter():
+    config = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+        'filters': [
+            {'filter_name': 'test_filter_name', 'policy': 'test_policy'},
+            {'filter_name': 'test_filter_name2', 'policy': 'test_policy2'},
+        ],
+        'tag_retention_minutes': 30,
+        'debug_data_package': True,
+        'fill_missing_tags': True,
+    }
+    result = base_scouter(config)
+    expected = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'frequency': '1m',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'on_conflict': 'error',
+        'trigger_laborious': False,
+        'filters': {
+            'test_filter_name': {'policy': 'test_policy'},
+            'test_filter_name2': {'policy': 'test_policy2'},
+        },
+        'schema': 'sientia_data',
+        'table_name': 'laborious_data',
+        'retention_time': 30 * 60,
+        'debug_data_package': True,
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'fill_missing_tags': True,
+        'runtime': 'legacy',
+    }
+    assert result == expected
+
+
+def test_pi_web_api_scouter():
+    config = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+        'filters': [{'filter_name': 'test_filter_name', 'policy': 'test_policy'}],
+        'read_tags': {
+            'test_tag_name': {
+                'webid': 'test_webid',
+                'aggr_func': 'test_aggr_func',
+                'data_range': [1, 2],
+            }
+        },
+        'tag_retention_minutes': 10,
+        'pi_web_api_config': {
+            'endpoint': 'https://test-endpoint.com',
+            'period': '*-2d',
+            'max_count': 5,
+            'api_timeout': 30,
+        },
+        'frequency': '1m',
+    }
+    result = pi_web_api_scouter(config)
+    expected = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'frequency': '1m',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'on_conflict': 'error',
+        'trigger_laborious': False,
+        'filters': {'test_filter_name': {'policy': 'test_policy'}},
+        'schema': 'sientia_data',
+        'table_name': 'laborious_data',
+        'retention_time': 10 * 60,
+        'model_tags': {
+            'test_tag_name': {
+                'webid': 'test_webid',
+                'aggr_func': 'test_aggr_func',
+                'data_range': [1, 2],
+            }
+        },
+        'debug_data_package': False,
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'fill_missing_tags': False,
+        'pi_web_api_query': {
+            'endpoint': 'https://test-endpoint.com',
+            'period': '*-2d',
+            'max_count': 5,
+            'api_timeout': 30,
+        },
+        'runtime': 'legacy',
+    }
+    assert result == expected
+
+
+def test_pi_web_api_scouter_with_timeout_greater_than_frequency():
+    config = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+        'filters': [],
+        'read_tags': {
+            'test_tag_name': {
+                'webid': 'test_webid',
+            }
+        },
+        'tag_retention_minutes': 10,
+        'pi_web_api_config': {
+            'endpoint': 'https://test-endpoint.com',
+            'api_timeout': 120,
+        },
+        'frequency': '1m',
+    }
+    result = pi_web_api_scouter(config)
+    expected = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'frequency': '1m',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'on_conflict': 'error',
+        'trigger_laborious': False,
+        'filters': {},
+        'schema': 'sientia_data',
+        'table_name': 'laborious_data',
+        'retention_time': 10 * 60,
+        'model_tags': {
+            'test_tag_name': {'webid': 'test_webid', 'aggr_func': 'lts', 'data_range': [-100, 100]}
+        },
+        'debug_data_package': False,
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'fill_missing_tags': False,
+        'pi_web_api_query': {
+            'endpoint': 'https://test-endpoint.com',
+            'period': '*-1d',
+            'max_count': 1,
+            'api_timeout': 60,
+        },
+        'runtime': 'legacy',
+    }
+    assert result == expected
+
+
+def test_pi_web_api_scouter_with_no_timeout():
+    config = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+        'filters': [],
+        'read_tags': {
+            'test_tag_name': {
+                'webid': 'test_webid',
+            }
+        },
+        'tag_retention_minutes': 10,
+        'pi_web_api_config': {
+            'endpoint': 'https://test-endpoint.com',
+        },
+        'frequency': '30s',
+    }
+    result = pi_web_api_scouter(config)
+    expected = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'frequency': '30s',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'on_conflict': 'error',
+        'trigger_laborious': False,
+        'filters': {},
+        'schema': 'sientia_data',
+        'table_name': 'laborious_data',
+        'retention_time': 10 * 60,
+        'model_tags': {
+            'test_tag_name': {'webid': 'test_webid', 'aggr_func': 'lts', 'data_range': [-100, 100]}
+        },
+        'debug_data_package': False,
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'fill_missing_tags': False,
+        'pi_web_api_query': {
+            'endpoint': 'https://test-endpoint.com',
+            'period': '*-1d',
+            'max_count': 1,
+            'api_timeout': 30,
+        },
+        'runtime': 'legacy',
+    }
+    assert result == expected
+
+
+@mark.parametrize(
+    'frequency_alert, warns',
+    [
+        (None, False),
+        ('5m', False),
+        ('1m', False),  # the floor itself is allowed
+        ('60s', False),
+        ('30s', True),
+        ('5min', True),  # unparseable
+        (5, True),  # not a string: must warn, never raise
+    ],
+)
+def test_frequency_alert_warning(frequency_alert, warns):
+    assert (frequency_alert_warning(frequency_alert) is not None) is warns
+
+
+def test_pi_web_api_scouter_legacy_shape():
+    """Without a connector the output is exactly the pre-tag-database one"""
+    config = pi_scouter_pipeline_doc(
+        {'TI-101': {'webid': 'F1DP-X'}},
+        pi_web_api_config={'endpoint': '/streamsets/recorded', 'max_count': 5, 'api_timeout': 10},
+    )
+
+    result = pi_web_api_scouter(config)
+
+    assert 'connection' not in result
+    assert result['model_tags'] == {
+        'TI-101': {'webid': 'F1DP-X', 'aggr_func': 'lts', 'data_range': [-100, 100]}
+    }
+    assert result['pi_web_api_query'] == {
+        'endpoint': '/streamsets/recorded',
+        'period': '*-1d',
+        'max_count': 5,
+        'api_timeout': 10,
+    }
+
+
+def test_pi_web_api_scouter_connector_defaults_with_pipeline_override():
+    connector = pi_connector_doc(
+        period='*-2h', max_count=10, api_timeout=600, data_server_webid='DS1'
+    )
+    config = pi_scouter_pipeline_doc(
+        {'TI-101': {'tag_id': 42, 'webid': 'F1DP-A', 'aggr_func': 'avg'}},
+        pi_web_api_config={'endpoint': '/streamsets/recorded', 'max_count': 5},
+        pi_web_api_connector=connector,
+    )
+
+    result = pi_web_api_scouter(config)
+
+    assert result['model_tags'] == {
+        'TI-101': {'webid': 'F1DP-A', 'aggr_func': 'avg', 'data_range': [-100, 100]}
+    }
+    assert result['pi_web_api_query'] == {
+        'endpoint': '/streamsets/recorded',
+        'period': '*-2h',
+        'max_count': 5,
+        # The connector default is still capped at the 1m frequency
+        'api_timeout': 60,
+    }
+    assert result['connection'] == {
+        'connector_id': 'pi-usina',
+        'base_url': 'https://pi-usina.example/piwebapi',
+        'auth': {'type': 'bearer', 'secret': PI_SECRET},
+        'data_server_webid': 'DS1',
+    }
+
+
+def test_pi_web_api_scouter_connection_carries_only_the_secret_reference():
+    connector = pi_connector_doc()
+    connector['pi_web_api']['auth']['secret']['value'] = 'leaked-credential'
+    connector['pi_web_api']['auth']['token'] = 'leaked-credential'
+    config = pi_scouter_pipeline_doc(
+        {'TI-101': {'tag_id': 42, 'webid': 'F1DP-A'}}, pi_web_api_connector=connector
+    )
+
+    result = pi_web_api_scouter(config)
+
+    assert result['connection']['auth'] == {'type': 'bearer', 'secret': PI_SECRET}
+    assert 'leaked-credential' not in str(result)
+
+
+def test_predictions_batch_groups_write_tags_by_server_and_type():
+    config = {
+        'workflow_type': 'predictions_batch',
+        'schedule_name': 'batch',
+        'model_id': 'model_id',
+        'model': {'name': 'model_name'},
+        'query': 'SELECT 1',
+        'write_tags': [
+            {'server_id': '1', 'type': 'prediction', 'addr': 'ns=2;s=P1'},
+            {'server_id': '1', 'type': 'prediction', 'addr': 'ns=2;s=P2', 'data_type': 'int'},
+            # Neither prediction nor confidence: the server is known, the tag is ignored
+            {'server_id': '2', 'type': 'other', 'addr': 'ns=2;s=O1'},
+        ],
+    }
+
+    result = predictions_batch(config)
+
+    assert result['opc_output_config'] == {
+        '1': {
+            'prediction_tags': {
+                'ns=2;s=P1': {'data_type': 'float'},
+                'ns=2;s=P2': {'data_type': 'int'},
+            }
+        },
+        '2': {},
+    }

tests/orchestrator/utils/test_tag_database.py

@@ -0,0 +1,291 @@
+from orchestrator.utils.tag_database import (
+    filter_by_tag_ids,
+    index_connector_tags,
+    index_connectors,
+    index_pi_connectors,
+    referenced_pi_tag_ids,
+    referenced_tag_ids,
+    resolve_pi_output_tags,
+    resolve_pi_read_tags,
+    resolve_tag_names,
+)
+from tests.orchestrator.utils.tag_database_fixtures import (
+    TAGS_BY_ID,
+    legacy_tag,
+    pi_connector_doc,
+    pi_scouter_pipeline_doc,
+    pi_tag_doc,
+    scouter_pipeline_doc,
+)
+
+
+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_referenced_tag_ids():
+    """Ids in int and text form, scouter pipelines only, empty or null tag_id ignored"""
+    pipelines = [
+        scouter_pipeline_doc(
+            'sin', 'm1', [{'tag_id': 7}, {'tag_id': '12'}, {'tag_id': ''}, {'tag_id': None}]
+        ),
+        scouter_pipeline_doc('cos', 'm2', [{'tag_id': 'abc'}, legacy_tag('ns=2;s=A', 'a')]),
+        scouter_pipeline_doc('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 = scouter_pipeline_doc('sin', 'm1', [legacy_tag('ns=2;s=A', 'a')])
+    no_read_tags = scouter_pipeline_doc('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_index_pi_connectors_leaves_out_incomplete():
+    complete = pi_connector_doc()
+    no_url = {**pi_connector_doc('pi-a'), 'url': ''}
+    no_auth = {**pi_connector_doc('pi-b'), 'pi_web_api': {}}
+    inline_token = {
+        **pi_connector_doc('pi-c'),
+        'pi_web_api': {'auth': {'type': 'bearer', 'token': 'secret-value'}},
+    }
+    no_secret_key = {
+        **pi_connector_doc('pi-d'),
+        'pi_web_api': {'auth': {'type': 'basic', 'secret': {'name': 'x'}}},
+    }
+
+    index = index_pi_connectors([complete, no_url, no_auth, inline_token, no_secret_key])
+
+    assert index == {'pi-usina': complete}
+
+
+def test_referenced_pi_tag_ids():
+    pipelines = [
+        pi_scouter_pipeline_doc({'TI-101': {'tag_id': 42}, 'TI-102': {'webid': 'F1DP-X'}}),
+        {
+            'workflow_type': 'predictions_batch',
+            'pi_web_api_output_config': {
+                'prediction_tags': [{'tag_id': '77'}],
+                'confidence_tags': {'CONF-1': 'F1DP-C'},
+            },
+        },
+        {
+            'workflow_type': 'xgboost_predictions_batch',
+            'pi_web_api_output_config': {'confidence_tags': [{'tag_id': 78}]},
+        },
+        # OPC scouter tag_ids belong to the OPC loads
+        {'workflow_type': 'scouter', 'read_tags': [{'tag_id': 7}]},
+        {'workflow_type': 'drift'},
+    ]
+
+    assert referenced_pi_tag_ids(pipelines) == [42, 77, 78, '42', '77', '78']
+
+
+def test_referenced_pi_tag_ids_none():
+    assert referenced_pi_tag_ids([pi_scouter_pipeline_doc({'TI-101': {'webid': 'F1DP-X'}})]) == []
+
+
+def test_resolve_pi_read_tags_by_tag_id():
+    connectors = index_pi_connectors([pi_connector_doc()])
+    tags = index_connector_tags([pi_tag_doc(42, 'F1DP-A', 'TI-101-db')])
+
+    resolved, connector, issues = resolve_pi_read_tags(
+        {'TI-101': {'tag_id': 42, 'aggr_func': 'avg'}}, tags, connectors
+    )
+
+    # The key stays the model tag name; only the WebID comes from the tag database
+    assert resolved == {'TI-101': {'tag_id': 42, 'aggr_func': 'avg', 'webid': 'F1DP-A'}}
+    assert connector == connectors['pi-usina']
+    assert not any(issues.values())
+
+
+def test_resolve_pi_read_tags_legacy_untouched():
+    legacy = {'TI-101': {'webid': 'F1DP-X'}}
+
+    resolved, connector, issues = resolve_pi_read_tags(legacy, {}, {})
+
+    assert resolved == legacy
+    assert resolved['TI-101'] is legacy['TI-101']
+    assert connector is None
+    assert not any(issues.values())
+
+
+def test_resolve_pi_read_tags_issues():
+    connectors = index_pi_connectors([pi_connector_doc()])
+    tags = index_connector_tags(
+        [
+            pi_tag_doc(42, 'F1DP-A', 'TI-101'),
+            pi_tag_doc(43, 'F1DP-W', 'PRED-1', category='Write'),
+            pi_tag_doc(44, 'F1DP-O', 'TI-OFF', connector_id='pi-inactive'),
+        ]
+    )
+
+    resolved, connector, issues = resolve_pi_read_tags(
+        {
+            'TI-101': {'tag_id': 42},
+            'UNKNOWN': {'tag_id': 99},
+            'WRITE': {'tag_id': 43},
+            'OFF': {'tag_id': 44},
+            'LEGACY': {'webid': 'F1DP-X'},
+        },
+        tags,
+        connectors,
+    )
+
+    # Unresolvable tags are left out; the rest of the pipeline is still scheduled
+    assert set(resolved) == {'TI-101', 'LEGACY'}
+    assert connector == connectors['pi-usina']
+    assert issues == {
+        'tags': {'99'},
+        'categories': {'43'},
+        'connectors': {'pi-inactive'},
+        'mixed': set(),
+    }
+
+
+def test_resolve_pi_read_tags_mixed_connectors():
+    connectors = index_pi_connectors([pi_connector_doc('pi-a'), pi_connector_doc('pi-b')])
+    tags = index_connector_tags(
+        [
+            pi_tag_doc(1, 'F1DP-1', 'A', connector_id='pi-a'),
+            pi_tag_doc(2, 'F1DP-2', 'B', connector_id='pi-b'),
+        ]
+    )
+
+    _resolved, connector, issues = resolve_pi_read_tags(
+        {'A': {'tag_id': 1}, 'B': {'tag_id': 2}}, tags, connectors
+    )
+
+    assert connector is None
+    assert issues['mixed'] == {'pi-a, pi-b'}
+
+
+def test_resolve_pi_output_tags_by_tag_id():
+    connectors = index_pi_connectors([pi_connector_doc()])
+    tags = index_connector_tags(
+        [
+            pi_tag_doc(77, 'F1DP-P', 'PRED-101', category='Write'),
+            pi_tag_doc(78, 'F1DP-C', 'CONF-101', category='Prediction Confidence'),
+        ]
+    )
+
+    resolved, issues = resolve_pi_output_tags(
+        {
+            'prediction_tags': [{'tag_id': 77}],
+            'confidence_tags': [{'tag_id': 78}],
+            'other': 'kept',
+        },
+        tags,
+        connectors,
+    )
+
+    assert resolved == {
+        'prediction_tags': {'PRED-101': 'F1DP-P'},
+        'confidence_tags': {'CONF-101': 'F1DP-C'},
+        'other': 'kept',
+    }
+    assert not any(issues.values())
+
+
+def test_resolve_pi_output_tags_legacy_untouched():
+    legacy = {'prediction_tags': {'PRED-101': 'F1DP-P'}, 'confidence_tags': {}}
+
+    resolved, issues = resolve_pi_output_tags(legacy, {}, {})
+
+    assert resolved == legacy
+    assert resolved['prediction_tags'] is legacy['prediction_tags']
+    assert not any(issues.values())
+
+
+def test_resolve_pi_output_tags_issues():
+    connectors = index_pi_connectors([pi_connector_doc()])
+    tags = index_connector_tags(
+        [
+            pi_tag_doc(42, 'F1DP-R', 'TI-101', category='Read'),
+            pi_tag_doc(77, 'F1DP-P', 'PRED-101', category='Write'),
+        ]
+    )
+
+    resolved, issues = resolve_pi_output_tags(
+        {
+            # A Read tag as prediction, a Write tag as confidence, an unknown tag_id
+            'prediction_tags': [{'tag_id': 42}, {'tag_id': 99}, {'PRED-LEGACY': 'F1DP-L'}],
+            'confidence_tags': [{'tag_id': 77}],
+        },
+        tags,
+        connectors,
+    )
+
+    assert resolved['prediction_tags'] == {'PRED-LEGACY': 'F1DP-L'}
+    assert resolved['confidence_tags'] == {}
+    assert issues['tags'] == {'99'}
+    assert issues['categories'] == {'42', '77'}
+
+
+def test_referenced_pi_tag_ids_ignores_malformed_read_tags():
+    assert referenced_pi_tag_ids([pi_scouter_pipeline_doc([{'tag_id': 42}])]) == []
+
+
+def test_resolve_pi_output_tags_ignores_non_dict_entries():
+    resolved, issues = resolve_pi_output_tags(
+        {'prediction_tags': ['F1DP-P', None, {'PRED-101': 'F1DP-P'}]}, {}, {}
+    )
+
+    assert resolved == {'prediction_tags': {'PRED-101': 'F1DP-P'}}
+    assert not any(issues.values())