SIENTIAPDE-1646

Update README, requirements, and E2E tests for improved configuration and functionality

- Enhanced the README with updated model configuration examples, including the addition of an alias for production.
- Removed the `requirements-light.txt` file and updated `requirements-local.txt` and `requirements.txt` to replace `asyncua` with `opcua`.
- Refactored E2E test scenarios to utilize scenario input files for better maintainability and clarity.
- Improved test coverage for MinIO offload functionality and added new helper functions for loading scenario inputs.
- Updated `values.yaml` to reflect new global configurations and environment variables for the laborious worker.

e2e/helpers.py

@@ -3,13 +3,60 @@ Shared helpers for E2E tests (Temporal workflows + PostgreSQL).
 """
 
 import asyncio
+import json
 from datetime import datetime
 from decimal import Decimal
+from pathlib import Path
 from typing import Any
 
+import pandas as pd
 from sqlalchemy import text
 from sqlalchemy.engine import Engine
 
+SCENARIO_INPUTS_DIR = Path(__file__).parent / 'scenario_inputs'
+
+
+def _replace_template_values(payload: Any, model_id: int) -> Any:
+    """
+    Replace string placeholders in scenario payloads with the concrete model id.
+
+    Args:
+        payload: JSON-like structure loaded from scenario input file.
+        model_id: Model id used to render template placeholders.
+
+    Return:
+        Any: Payload with ``{{MODEL_ID}}`` replaced where applicable.
+    """
+    if isinstance(payload, dict):
+        return {key: _replace_template_values(value, model_id) for key, value in payload.items()}
+    if isinstance(payload, list):
+        return [_replace_template_values(item, model_id) for item in payload]
+    if isinstance(payload, str):
+        if payload == '{{MODEL_ID}}':
+            return model_id
+        return payload.replace('{{MODEL_ID}}', str(model_id))
+    return payload
+
+
+def load_scenario_input(file_name: str, model_id: int | None = None) -> dict[str, Any]:
+    """
+    Load a scenario input JSON from ``e2e/scenario_inputs``.
+
+    Args:
+        file_name: JSON file name inside ``e2e/scenario_inputs``.
+        model_id: Optional model id used to render ``{{MODEL_ID}}`` placeholders.
+
+    Return:
+        dict[str, Any]: Input payload ready to be passed to workflow/activity calls.
+    """
+    file_path = SCENARIO_INPUTS_DIR / file_name
+    with file_path.open('r', encoding='utf-8') as f:
+        payload = json.load(f)
+
+    if model_id is not None:
+        return _replace_template_values(payload, model_id)
+    return payload
+
 
 async def start_and_await_workflow(client, workflow_run, input_data: dict, workflow_id: str, timeout: float = 60.0):
     """
@@ -172,3 +219,133 @@ def assert_repeat(postgres_engine: Engine, model_id: int, last_prediction: tuple
 def make_workflow_id(prefix: str) -> str:
     """Build a unique workflow id using a prefix and current timestamp."""
     return f'{prefix}-{datetime.now().timestamp()}'
+
+
+def insert_target_data_for_drift(
+    postgres_engine: Engine,
+    model_id: int,
+    timestamps: list[str],
+    variables_values: dict[str, list[float]],
+) -> None:
+    """
+    Insert one row per (timestamp, variable) pair into ``laborious_data``.
+
+    Used by drift scenarios that need wide-format input where the pivot keeps a
+    full row for every timestamp.
+
+    Args:
+     - postgres_engine: SQLAlchemy engine bound to the test container.
+     - model_id: Model id stamped on every row.
+     - timestamps: ISO-8601 strings used both as ``timestamp`` and ``created_at``.
+     - variables_values: Mapping of variable name to a list of values; each list
+       must be the same length as ``timestamps``.
+    """
+    for var_name, values in variables_values.items():
+        if len(values) != len(timestamps):
+            raise ValueError(
+                f"Variable '{var_name}' has {len(values)} values but {len(timestamps)} timestamps"
+            )
+
+    rows_sql = []
+    for index, ts in enumerate(timestamps):
+        for var_name, values in variables_values.items():
+            rows_sql.append(
+                f"({model_id}, '{var_name}', {values[index]}, '{ts}', '{ts}')"
+            )
+
+    with postgres_engine.begin() as conn:
+        conn.execute(
+            text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}')
+        )
+        if rows_sql:
+            conn.execute(
+                text(
+                    'INSERT INTO predictions_schema.laborious_data '
+                    '(model_id, variable, value, "timestamp", created_at) VALUES '
+                    + ', '.join(rows_sql)
+                )
+            )
+
+
+def build_drift_dataframe(
+    timestamps: list[str],
+    features: list[str],
+    methods: list[str],
+    statistic: float = 1.0,
+    drift_flags: dict[tuple[str, str], bool] | None = None,
+    include_multivariate: bool = True,
+    multivariate_value: float = 16.0,
+    multivariate_drift: bool = True,
+    extra_rows: list[dict[str, Any]] | None = None,
+) -> Any:
+    """
+    Build a deterministic dataframe matching the schema returned by
+    ``sientia_model.analytics.model_analysis.ModelAnalysis.get_drift_metrics_dataframe``
+    so drift E2E tests can pin the exact rows persisted to PostgreSQL.
+
+    The output mirrors the analyzer's canonical schema:
+    ``timestamp, feature, metric, statistic, p_value, alert, chunk_index,
+    chunk_start_date, chunk_end_date``. ``calculate_drift`` then renames
+    ``alert -> drift``, ``chunk_index -> chunk``, ``chunk_end_date ->
+    timestamp_end`` and drops ``p_value`` / ``chunk_start_date``.
+
+    Args:
+     - timestamps: Truncated chunk start timestamps (``'2024-01-01 12:00'`` for ``min``).
+     - features: Univariate feature names (one row per feature/method/timestamp).
+     - methods: Univariate methods such as ``kolmogorov_smirnov``.
+     - statistic: Default univariate statistic value.
+     - drift_flags: Optional override of the ``alert`` flag per ``(feature, method)`` pair.
+     - include_multivariate: Whether to add a final multivariate row block.
+     - multivariate_value: Value placed on multivariate rows.
+     - multivariate_drift: Drift flag placed on multivariate rows.
+     - extra_rows: Additional pre-built rows to append (used for dedup/p_value tests).
+
+    Return:
+     pandas.DataFrame with columns: timestamp, feature, metric, statistic,
+     p_value, alert, chunk_index, chunk_start_date, chunk_end_date.
+    """
+    rows: list[dict[str, Any]] = []
+    drift_flags = drift_flags or {}
+    # Synthetic chunk-end offset that mirrors the high-precision boundary
+    # (``...:59.999999999``) emitted by ``ModelAnalysis`` for minute chunks.
+    # Computing via ``Timedelta`` instead of string concatenation keeps the
+    # helper safe for both minute- and second-precision timestamps.
+    chunk_span = pd.Timedelta(seconds=59, nanoseconds=999999999)
+
+    for chunk_index, ts in enumerate(timestamps):
+        chunk_start = pd.Timestamp(ts)
+        chunk_end = chunk_start + chunk_span
+        for feature in features:
+            for method in methods:
+                rows.append(
+                    {
+                        'timestamp': chunk_start,
+                        'feature': feature,
+                        'metric': method,
+                        'statistic': statistic,
+                        'p_value': 0.5,
+                        'alert': drift_flags.get((feature, method), False),
+                        'chunk_index': chunk_index,
+                        'chunk_start_date': chunk_start,
+                        'chunk_end_date': chunk_end,
+                    }
+                )
+        if include_multivariate:
+            rows.append(
+                {
+                    'timestamp': chunk_start,
+                    'feature': 'multivariate',
+                    'metric': 'multivariate',
+                    'statistic': multivariate_value,
+                    'p_value': 0.0,
+                    'alert': multivariate_drift,
+                    'chunk_index': chunk_index,
+                    'chunk_start_date': chunk_start,
+                    'chunk_end_date': chunk_end,
+                }
+            )
+
+    if extra_rows:
+        rows.extend(extra_rows)
+
+    return pd.DataFrame(rows)

e2e/scenario_inputs/drift_base.json

@@ -0,0 +1,14 @@
+{
+  "schedule_name": "test-schedule",
+  "model_name": "test_model",
+  "model_id": "{{MODEL_ID}}",
+  "schema": "predictions_schema",
+  "source_table_name": "laborious_data",
+  "target_table_name": "drift",
+  "interval": 60,
+  "drift_metrics": ["kolmogorov_smirnov", "jensen_shannon", "wasserstein"],
+  "chunk_period": "min",
+  "model_config": {
+    "target": "sensor_1"
+  }
+}

e2e/scenario_inputs/format_export_base.json

@@ -0,0 +1,37 @@
+{
+  "schedule_name": "test-schedule",
+  "model_name": "test_model",
+  "model_id": "{{MODEL_ID}}",
+  "query": "SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = {{MODEL_ID}}",
+  "schema": "predictions_schema",
+  "table_name": "predictions",
+  "transform_table_name": "transformed_data",
+  "input_filters": {
+    "EMPTY_DATA": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "mlflow_transform_filters": {
+    "API_ERROR": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "mlflow_predict_filters": {
+    "API_ERROR": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "path_priority": ["STOP", "CONTINUE", "REPEAT"],
+  "opc_output_config": {},
+  "pi_web_api_output_config": {},
+  "save_transform": true,
+  "prediction_store_policy": "lts:1",
+  "model_config": {
+    "retention_minutes": 0,
+    "target": "sensor_1"
+  },
+  "datetime_columns": ["timestamp", "created_at"]
+}

e2e/scenario_inputs/main_happy_path.json

@@ -0,0 +1,45 @@
+{
+  "metadata": {
+    "metadata": {
+      "model_id": "{{MODEL_ID}}",
+      "model_name": "test_model",
+      "schedule_name": "test-schedule",
+      "workflow_name": "predictions_batch"
+    }
+  },
+  "schedule_name": "test-schedule",
+  "model_name": "test_model",
+  "model_id": "{{MODEL_ID}}",
+  "query": "SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = {{MODEL_ID}}",
+  "schema": "predictions_schema",
+  "table_name": "predictions",
+  "transform_table_name": "transformed_data",
+  "input_filters": {
+    "EMPTY_DATA": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "mlflow_transform_filters": {
+    "API_ERROR": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "mlflow_predict_filters": {
+    "API_ERROR": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "path_priority": ["STOP", "CONTINUE", "REPEAT"],
+  "opc_output_config": {},
+  "pi_web_api_output_config": {},
+  "save_transform": true,
+  "prediction_store_policy": "lts:1",
+  "model_config": {
+    "target": "sensor_1",
+    "retention_minutes": 0
+  },
+  "datetime_columns": ["timestamp", "created_at"]
+}

e2e/scenario_inputs/main_invalid_datetime.json

@@ -0,0 +1,45 @@
+{
+  "metadata": {
+    "metadata": {
+      "model_id": "{{MODEL_ID}}",
+      "model_name": "test_model",
+      "schedule_name": "test-schedule",
+      "workflow_name": "predictions_batch"
+    }
+  },
+  "schedule_name": "test-schedule",
+  "model_name": "test_model",
+  "model_id": "{{MODEL_ID}}",
+  "query": "SELECT timestamp, variable, value FROM predictions_schema.laborious_data WHERE model_id = {{MODEL_ID}}",
+  "schema": "predictions_schema",
+  "table_name": "predictions",
+  "transform_table_name": "transformed_data",
+  "input_filters": {
+    "EMPTY_DATA": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "mlflow_transform_filters": {
+    "API_ERROR": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "mlflow_predict_filters": {
+    "API_ERROR": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "path_priority": ["STOP", "CONTINUE", "REPEAT"],
+  "opc_output_config": {},
+  "pi_web_api_output_config": {},
+  "save_transform": true,
+  "prediction_store_policy": "lts:1",
+  "model_config": {
+    "target": "sensor_1",
+    "retention_minutes": 0
+  },
+  "datetime_columns": ["nonexistent_column"]
+}

e2e/scenario_inputs/main_missing_required.json

@@ -0,0 +1,16 @@
+{
+  "metadata": {
+    "metadata": {
+      "model_id": "{{MODEL_ID}}",
+      "model_name": "test_model",
+      "schedule_name": "test-schedule",
+      "workflow_name": "predictions_batch"
+    }
+  },
+  "schedule_name": "test-schedule",
+  "model_name": "test_model",
+  "model_id": "{{MODEL_ID}}",
+  "schema": "predictions_schema",
+  "table_name": "predictions",
+  "transform_table_name": "transformed_data"
+}

e2e/scenario_inputs/main_sql_error.json

@@ -0,0 +1,44 @@
+{
+  "metadata": {
+    "metadata": {
+      "model_id": "{{MODEL_ID}}",
+      "model_name": "test_model",
+      "schedule_name": "test-schedule",
+      "workflow_name": "predictions_batch"
+    }
+  },
+  "schedule_name": "test-schedule",
+  "model_name": "test_model",
+  "model_id": "{{MODEL_ID}}",
+  "query": "SELECT * FROM nonexistent_table WHERE invalid_syntax =",
+  "schema": "predictions_schema",
+  "table_name": "predictions",
+  "transform_table_name": "transformed_data",
+  "input_filters": {
+    "EMPTY_DATA": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "mlflow_transform_filters": {
+    "API_ERROR": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "mlflow_predict_filters": {
+    "API_ERROR": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "path_priority": ["STOP", "CONTINUE", "REPEAT"],
+  "opc_output_config": {},
+  "pi_web_api_output_config": {},
+  "save_transform": true,
+  "prediction_store_policy": "lts:1",
+  "model_config": {
+    "target": "sensor_1",
+    "retention_minutes": 0
+  }
+}

e2e/scenario_inputs/minimal_retrain_base.json

@@ -0,0 +1,12 @@
+{
+  "schedule_name": "test-schedule",
+  "model_name": "test_model",
+  "model_id": "{{MODEL_ID}}",
+  "query": "SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = {{MODEL_ID}}",
+  "schema": "predictions_schema",
+  "table_name": "retrain_reports",
+  "datetime_columns": ["timestamp", "created_at"],
+  "model_config": {
+    "target": "sensor_1"
+  }
+}

e2e/scenario_inputs/minio_offload_load_query.json

@@ -0,0 +1,11 @@
+{
+  "metadata": {
+    "schedule_name": "test-schedule",
+    "model_name": "test_model",
+    "model_id": "{{MODEL_ID}}",
+    "workflow_name": "predictions_batch"
+  },
+  "query": "SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = {{MODEL_ID}}",
+  "model_name": "test_model",
+  "datetime_columns": ["timestamp", "created_at"]
+}

e2e/scenario_inputs/minio_offload_workflow.json

@@ -0,0 +1,37 @@
+{
+  "schedule_name": "test-schedule",
+  "model_name": "test_model",
+  "model_id": "{{MODEL_ID}}",
+  "query": "SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = {{MODEL_ID}}",
+  "schema": "predictions_schema",
+  "table_name": "predictions",
+  "transform_table_name": "transformed_data",
+  "input_filters": {
+    "EMPTY_DATA": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "mlflow_transform_filters": {
+    "API_ERROR": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "mlflow_predict_filters": {
+    "API_ERROR": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "path_priority": ["STOP", "CONTINUE", "REPEAT"],
+  "opc_output_config": {},
+  "pi_web_api_output_config": {},
+  "save_transform": false,
+  "prediction_store_policy": "lts:1",
+  "model_config": {
+    "retention_minutes": 0,
+    "target": "sensor_1"
+  },
+  "datetime_columns": ["timestamp", "created_at"]
+}

e2e/scenario_inputs/prediction_process_base.json

@@ -0,0 +1,39 @@
+{
+  "schedule_name": "test-schedule",
+  "model_name": "test_model",
+  "model_id": "{{MODEL_ID}}",
+  "query": "SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = {{MODEL_ID}}",
+  "schema": "predictions_schema",
+  "table_name": "predictions",
+  "transform_table_name": "transformed_data",
+  "input_filters": {
+    "SPECIFIC_VARIABLES_NULL_VALUES": {
+      "POLICY": "CONTINUE",
+      "CONFIG": {
+        "variables": ["sensor_1"]
+      }
+    }
+  },
+  "mlflow_transform_filters": {
+    "API_ERROR": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "mlflow_predict_filters": {
+    "API_ERROR": {
+      "POLICY": "STOP",
+      "CONFIG": {}
+    }
+  },
+  "path_priority": ["STOP", "CONTINUE", "REPEAT"],
+  "opc_output_config": {},
+  "pi_web_api_output_config": {},
+  "save_transform": true,
+  "prediction_store_policy": "lts:1",
+  "model_config": {
+    "retention_minutes": 0,
+    "target": "sensor_1"
+  },
+  "datetime_columns": ["timestamp", "created_at"]
+}

e2e/scenario_inputs/simple_metrics_base.json

@@ -0,0 +1,14 @@
+{
+  "schedule_name": "test-schedule",
+  "model_name": "test_model",
+  "model_id": "{{MODEL_ID}}",
+  "schema": "predictions_schema",
+  "predictions_table_name": "predictions",
+  "data_table_name": "laborious_data",
+  "target_table_name": "simple_metrics_data",
+  "interval_minutes": 60,
+  "metrics": ["rmse", "mse", "mae", "r2"],
+  "model_config": {
+    "target": "sensor_target"
+  }
+}

e2e/scenarios.md

@@ -1,520 +1,402 @@
-# Test Scenarios for Predictions Batch Workflow
+# E2E Scenario Documentation - Predictions Batch
 
-This document describes all possible test scenarios for the `predictions_batch` workflow and its child workflows `prediction_process` and `format_and_export_prediction`.
+This document describes the end-to-end scenarios for `predictions_batch` and its child workflows:
+`prediction_process` and `format_and_export_prediction`.
 
-## Running automated E2E tests (`e2e/`)
+It is a functional reference of scenario behavior, inputs, and expected outcomes.
 
-- **Runtime**: Docker (or a Docker-compatible daemon) must be available so [testcontainers](https://testcontainers.com/) can start **PostgreSQL** and **MinIO** containers.
-- **Dependencies**: install dev requirements (includes `testcontainers[postgres,minio]`).
-- **Invocation**: run only integration-marked tests, for example: `pytest e2e/ -m integration`.
-- **MinIO tests**: `e2e/test_minio_offload.py` exercises real S3 uploads; other E2E modules continue to mock MinIO on the worker used by most scenarios.
+## Execution Context
 
-## Workflow Overview
-
-The `predictions_batch` workflow:
-1. Loads data using a custom SQL query
-2. Prepares prediction configuration
-3. Delegates to `prediction_process` child workflow which:
-   - Retrieves last timestamp for incremental processing
-   - Applies input data quality gates
-   - Executes MLFlow transform operation
-   - Validates transform response
-   - Executes MLFlow predict operation
-   - Validates predict response
-   - Delegates to `format_and_export_prediction` child workflow
-4. The `format_and_export_prediction` workflow:
-   - Formats prediction data (normal or default)
-   - Exports to PI Web API (optional)
-   - Exports to OPC server (optional)
-   - Exports to PostgreSQL
-   - Writes metrics
+- Tests run under `e2e/` and are marked with `@pytest.mark.integration`.
+- PostgreSQL and MinIO are provisioned with testcontainers.
+- `test_minio_offload.py` uses real MinIO I/O; other scenario suites may use stubs/mocks for optional outputs.
 
 ---
 
-## 1. Predictions Batch - Main Workflow Scenarios
+## 1. Main Workflow Scenarios
+Source: `e2e/test_predictions_batch_main_workflow.py`
 
-### 1.1 Success Scenarios
+### 1.1.1 Happy Path - Complete Success
+**Summary**: Full workflow succeeds with valid query and default gate behavior.
 
-#### Scenario 1.1.1: Happy Path - Complete Success
-**Description**: Workflow completes successfully with valid SQL query and all activities succeed
+**Description**:
+- Query returns rows for a model.
+- `prediction_process` runs transform and predict paths.
+- Final prediction and transformed data are persisted.
 
-**Input**:
-- Valid `schedule_name`, `model_name`, `model_id`
-- Valid `query` returning non-empty DataFrame
-- Valid `schema`, `table_name`, `transform_table_name`
-- Optional `datetime_columns` for timestamp parsing
-- Optional `input_filters`, `mlflow_transform_filters`, `mlflow_predict_filters`
-- Optional `path_priority`, `opc_output_config`, `pi_web_api_output_config`
+**Expected Outcome**:
+- Exactly one prediction row is created.
+- Transform rows are created.
+- Confidence/status/comments are success values.
 
-**Expected Behavior**:
-- `load_custom_query` returns DataFrame with data
-- Workflow prepares prediction input with all configurations
-- `prediction_process` child workflow executes successfully
-- All gates pass with no issues
-- Transform and predict operations succeed
-- Data exported to PostgreSQL
-- Metrics written
+### 1.2.1 SQL Query Execution Error
+**Summary**: Invalid SQL leads to no persisted prediction.
 
-**Assertions**:
-- SQL query executed once
-- `prediction_process` workflow called with correct parameters
-- Data exists in PostgreSQL (predictions table)
-- Metrics recorded
-- No errors raised
+**Description**:
+- Input query is invalid.
+- Load step fails and workflow follows error/short-circuit path.
+
+**Expected Outcome**:
+- No prediction rows for the model.
+- Workflow does not require retry-loop assumptions in assertions.
+
+### 1.2.2 Missing Required Parameters
+**Summary**: Missing required fields prevent workflow completion path.
+
+**Description**:
+- Required input key (e.g. `query`) is omitted.
+- Workflow fails to produce actionable input for child flow.
+
+**Expected Outcome**:
+- No prediction rows are persisted.
+- Workflow handle may require explicit terminate in E2E harness.
+
+### 1.2.3 Invalid Datetime Column Specification (de-prioritized)
+**Summary**: Legacy invalid datetime-column case is retained only as low-priority legacy coverage.
+
+**Description**:
+- `datetime_columns` references non-existing columns.
+- Behavior may vary by query shape and parser fallback.
+
+**Expected Outcome**:
+- No predictions persisted in the covered legacy assertion path.
+- Scenario is not considered primary behavior coverage.
 
 ---
 
-### 1.2 Error Scenarios
+## 2. Prediction Process Scenarios
+Source: `e2e/test_predictions_batch_prediction_process.py`
 
-#### Scenario 1.2.1: SQL Query Execution Error
-**Description**: SQL query fails due to syntax error or connection issue
+### 2.1 Input Gate Path Decisions
 
-**Input**:
-- Invalid SQL query (syntax error)
-- Or database connection unavailable
+#### 2.1.1 CONTINUE
+**Summary**: Input filter flags quality issue but allows continuation via default path.
 
-**Expected Behavior**:
-- `load_custom_query` raises exception (caught by Temporal retry policy)
-- Notification sent with SQL error details
-- After retries, activity may return empty data or workflow may fail
-- If empty data returned, workflow completes with early exit via input gate
+**Description**:
+- Input gate returns `CONTINUE`.
+- MLFlow transform/predict are skipped.
+- Export path persists default-style prediction with warning context.
 
-**Assertions**:
-- Error notification sent
-- Workflow completes (either fails or exits early)
-- No data in predictions table
+#### 2.1.2 STOP
+**Summary**: Input filter blocks processing.
+
+**Description**:
+- Input gate returns `STOP`.
+- Workflow exits without export.
+
+#### 2.1.3 REPEAT with history
+**Summary**: Prior prediction is reused.
+
+**Description**:
+- Input gate returns `REPEAT`.
+- `repeat_last_prediction` path is executed using existing historical row.
+
+#### 2.1.4 REPEAT without history
+**Summary**: Repeat requested but no previous prediction exists.
+
+**Description**:
+- Input gate returns `REPEAT`.
+- No prior row is available to duplicate.
+
+**Expected Outcome**:
+- No new prediction rows are created for the model.
+
+### 2.2 Transform Gate Decisions
+
+#### 2.2.1 CONTINUE on transform response error
+**Summary**: Transform response is degraded, but workflow continues.
+
+#### 2.2.2 STOP on transform response error
+**Summary**: Transform response error blocks downstream processing.
+
+#### 2.2.3 REPEAT on transform response error
+**Summary**: Transform response error triggers repeat-last-prediction path.
+
+#### 2.2.4 STOP on transform content NaN
+**Summary**: Content gate (`NAN_VALUES`) blocks on all-NaN transform payload.
+
+### 2.3 Predict Gate Decisions
+
+#### 2.3.1 CONTINUE on predict response error
+**Summary**: Predict response degraded; workflow exports with degraded metadata.
+
+#### 2.3.2 STOP on predict response error
+**Summary**: Predict response error blocks export.
+
+#### 2.3.3 REPEAT on predict response error
+**Summary**: Predict response error routes to repeat-last-prediction.
+
+### 2.4.1 Priority Conflict Resolution
+**Summary**: Deterministic selection when multiple filters produce different flags.
+
+**Description**:
+- Multiple filters may produce `STOP`, `CONTINUE`, and/or `REPEAT`.
+- `path_priority` defines precedence.
+
+**Expected Outcome**:
+- Highest-priority flag is applied consistently.
+- Executed branch matches configured priority ordering.
 
 ---
 
-#### Scenario 1.2.2: Missing Required Parameters
-**Description**: Essential parameters missing from input
+## 3. Format and Export Scenarios
+Source: `e2e/test_predictions_batch_format_export.py`
 
-**Input**:
-- Missing `query` or `model_id` or `schema` or `table_name`
+### 3.1 Output Combination Scenarios
 
-**Expected Behavior**:
-- Workflow or activity raises KeyError or validation error
-- Workflow fails immediately
+#### 3.1.1 Default prediction export
+**Summary**: Non-`None` path flag uses `format_default_prediction`.
 
-**Assertions**:
-- Workflow fails with parameter error
-- Error notification sent
-- No child workflow called
+**Description**:
+- Default prediction is generated.
+- Transform export is skipped.
+- Optional outputs (PI/OPC) still execute when configured.
+
+#### 3.1.2 OPC only
+**Summary**: Postgres + OPC writes, PI Web API disabled.
+
+#### 3.1.3 PI Web API only
+**Summary**: Postgres + PI writes, OPC disabled.
+
+#### 3.1.4 Postgres only
+**Summary**: Both optional outputs disabled; only Postgres persistence and metrics.
+
+#### 3.1.5 No transformed data export
+**Summary**: Prediction is persisted; transformed table is not written.
+
+### 3.2 Degraded-but-successful Completion
+
+#### 3.2.1 PI Web API write error
+**Summary**: PI write failure does not fail workflow.
+
+**Expected Outcome**:
+- Workflow completes.
+- Prediction persisted with degraded confidence/comments (PI error semantics).
+
+#### 3.2.2 OPC write error
+**Summary**: OPC write failure does not fail workflow.
+
+**Expected Outcome**:
+- Workflow completes.
+- Prediction persisted with OPC degraded confidence/comments.
+
+#### 3.2.3 PI Web API partial write error
+**Summary**: Partial PI acknowledgement is treated as degraded success.
+
+**Expected Outcome**:
+- Workflow completes.
+- Prediction persisted with PI error confidence and descriptive comment.
+
+### 3.3.1 Combined Optional Outputs (PI + OPC)
+**Summary**: Both external output channels are enabled together.
+
+**Description**:
+- PI Web API and OPC configs are both present.
+- Output mutation order matters for final persisted payload.
+
+**Expected Outcome**:
+- PI write executes before OPC write in workflow sequence.
+- Final Postgres payload reflects any confidence/comment updates.
+- OPC metrics are emitted when tag writes return response times.
 
 ---
 
-#### Scenario 1.2.3: Invalid Datetime Column Specification
-**Description**: Datetime column specified doesn't exist in query results
+## 4. MinIO Offload Scenarios
+Source: `e2e/test_minio_offload.py`
 
-**Input**:
-- `datetime_columns: ['nonexistent_column']`
-- Query results don't have this column
+### 4.1.1 Forced offload to MinIO
+**Summary**: Very low threshold forces parquet upload.
 
-**Expected Behavior**:
-- `load_custom_query` may raise KeyError or warning
-- Depending on implementation, workflow may fail or continue
-- Error notification sent
+**Description**:
+- Payload is offloaded (`object_key` present, inline data absent/empty).
+- Object is present in MinIO under `prediction_datasets/...`.
+- Retrieval reconstructs the dataframe.
 
-**Assertions**:
-- Error raised or warning logged
-- Workflow behavior depends on error handling policy
+### 4.1.2 Full workflow with offloaded load payload
+**Summary**: Offload path works during full `predictions_batch` execution.
+
+**Expected Outcome**:
+- Workflow completes.
+- Prediction row is persisted.
+
+### 4.2.1 Inline payload below threshold
+**Summary**: Data remains inline when threshold is not exceeded.
+
+**Expected Outcome**:
+- Payload stores inline `data`.
+- `object_key` is `None`.
+- Downstream persistence behavior matches offload scenario semantics.
 
 ---
 
-## 2. Prediction Process - Child Workflow Scenarios
+## 5. Drift Workflow Scenarios
+Source: `e2e/test_drift.py`
 
-### 2.1 Input gate Early Exit Scenarios
+The drift suite mocks `sientia.ModelAnalysis.ModelAnalysis` (not installed; see
+`CODE_ISSUES.md` issue #1) through the controllable `_FakeModelAnalysis` stub
+exposed by the `model_analysis_stub` fixture. The `mlflow_repository_stub`
+provides the reference-data CSV via `download_artifacts`. Every scenario asserts
+postgres rows in `predictions_schema.drift` against this canonical schema:
 
-#### Scenario 2.1.1: Input Gate Triggers CONTINUE
-**Description**: Input gate determines data should use previous prediction
+`id, model_id, feature, method, value, drift, chunk, timestamp, timestamp_end, accurate, created_at, updated_at`.
 
-**Input**:
-- Data that should continue with input data as prediction
-- `input_filters` configured with `POLICY: 'CONTINUE'`
-- `path_priority` includes CONTINUE
+### 5.1 Happy paths
 
-**Expected Behavior**:
-- `input_gate` returns `path_flag='CONTINUE'`
-- `path_flag_handler` calls export workflow with input data directly
-- MLFlow transform and predict skipped
-- Data exported as-is
+#### D.1.1 Full pipeline persists all columns with reference data
+**Summary**: ModelAnalysis returns a deterministic drift dataframe; the
+reference CSV is downloaded from the MLflow stub.
 
-**Assertions**:
-- `input_gate` called
-- MLFlow operations NOT called
-- Export workflow called with original data
-- Workflow completes
+**Expected Outcome**:
+- One row per `(chunk, feature, method)` plus a `multivariate` block per chunk.
+- Every drift column is populated and `accurate=True`.
+- `timestamp_end` preserves the high-precision string (`HH:MM:59.999999999`).
+- `p_value` is dropped before persistence.
+- `drift` flags propagate per `(feature, method)` configuration.
 
+#### D.1.2 30% fallback when reference data is unavailable
+**Summary**: `get_reference_data` fails alias resolution and returns `None`;
+`calculate_drift` uses the first 30% of target rows as reference.
 
-#### Scenario 2.1.2: Input Gate Triggers STOP
-**Description**: Input data quality gate fails with STOP policy
+**Expected Outcome**:
+- Persisted rows carry `accurate=False`.
+- A `MODEL_METRICS_REFERENCE_DATA_WARNING` notification is emitted to MongoDB.
 
-**Input**:
-- Data with EMPTY_DATA or other critical issues
-- `input_filters` configured with `POLICY: 'STOP'`
+### 5.2 Filtering / dedup invariants
 
-**Expected Behavior**:
-- `input_gate` returns `path_flag='STOP'`
-- `path_flag_handler` detects STOP
-- Workflow returns early without calling MLFlow
-- No prediction exported
+#### D.2.1 Deduplication and `p_value` removal
+**Summary**: ModelAnalysis returns duplicate `(timestamp, method, feature)` rows
+plus a `p_value` column.
 
-**Assertions**:
-- `input_gate` called
-- `path_flag_handler` returns True (early exit)
-- MLFlow transform NOT called
-- Export workflow NOT called
-- Workflow completes without error
+**Expected Outcome**:
+- Duplicates are collapsed keeping the first occurrence.
+- `p_value` is absent from the persisted rows.
 
+#### D.2.2 Out-of-range timestamps filtered
+**Summary**: Drift rows whose timestamps are not present in the target window
+must be discarded before persistence.
 
-#### Scenario 2.1.3: Input Gate Triggers REPEAT
-**Description**: Input gate determines data should repeat last prediction
+### 5.3 Failure paths
 
-**Input**:
-- Data with quality issues that require using previous prediction
-- `input_filters` configured with `POLICY: 'REPEAT'`
-- `path_priority` includes REPEAT
+#### D.3.1 Empty target data short-circuits the workflow
+**Summary**: `load_custom_query` returns no rows; ModelAnalysis is never
+instantiated and no drift rows are written.
 
-**Expected Behavior**:
-- `input_gate` returns `path_flag='REPEAT'`
-- `path_flag_handler` calls `repeat_last_prediction` activity
-- MLFlow transform and predict skipped
-- Last prediction repeated and exported
+#### D.3.2 ModelAnalysis raises during dataframe assembly
+**Summary**: `get_drift_metrics_dataframe` raises. The activity catches the
+error, sends a `MODEL_METRICS_GET_DRIFT_METRICS_ERROR` notification, and the
+workflow completes without persisting drift rows.
 
-**Assertions**:
-- `input_gate` called
-- MLFlow operations NOT called
-- `repeat_last_prediction` activity called
-- Workflow completes
+### 5.4 Configuration paths
+
+#### D.4.1 Default drift metrics propagated to analyzer
+**Summary**: Omitting `drift_metrics` defaults to
+`['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']` and forwards the
+exact list to `detect_univariate_drift`.
+
+#### D.4.2 Invalid `chunk_period` raises ValueError
+**Summary**: Anything other than `min` / `s` is rejected by `calculate_drift`.
+
+#### D.4.3 `chunk_period='s'` keeps seconds in timestamp filtering
+**Summary**: Truncated `YYYY-MM-DD HH:MM` rows are filtered out when chunking
+runs at second granularity.
 
 ---
 
-### 2.2 Transform gate Early Exit Scenarios
+## 6. Simple Metrics Workflow Scenarios
+Source: `e2e/test_simple_metrics.py`
 
-#### Scenario 2.2.1: Transform Gate Triggers CONTINUE
-**Description**: Transform response gate determines data should continue despite issues
+Validates `predictions_schema.simple_metrics_data` columns:
+`id, model_id, metric, value, timestamp, data_size, interval_minutes, created_at`.
 
-**Input**:
-- Valid input data
-- Transform response has quality issues but policy is CONTINUE
-- `mlflow_transform_filters` configured with `POLICY: 'CONTINUE'`
-- `path_priority` includes CONTINUE
+### 6.1 Happy paths
 
-**Expected Behavior**:
-- `request_transform` succeeds
-- `mlflow_response_gate` for transform returns `path_flag='CONTINUE'`
-- `path_flag_handler` calls export workflow with transform data
-- MLFlow predict skipped
-- Transform data exported as-is
+#### S.1.1 rmse/mse/mae/r2 happy path
+**Summary**: Prediction/target pairs are inserted; the activity computes all
+four metrics with closed-form expected values.
 
-**Assertions**:
-- Transform completed
-- `mlflow_response_gate` called for transform
-- MLFlow predict NOT called
-- Export workflow called with transform data
-- Workflow completes
+**Expected Outcome**:
+- One row per metric is persisted; all columns populated.
+- `data_size` matches the joined row count and `interval_minutes=60`.
+
+#### S.1.2 Subset metrics
+**Summary**: Requesting `metrics=['rmse']` writes only the rmse row.
+
+### 6.2 Edge cases
+
+#### S.2.1 Zero-variance target returns r2=0
+**Summary**: When all targets are equal, `ss_tot=0`; the activity must guard
+against division by zero and return `r2=0`.
+
+### 6.3 Failure paths
+
+#### S.3.1 No overlapping data short-circuits persistence
+**Summary**: With no `laborious_data` rows for the configured target variable
+the workflow exits before `calculate_simple_metrics` and writes nothing.
 
 ---
 
-#### Scenario 2.2.2: Transform Gate Triggers STOP
-**Description**: Transform response validation fails with STOP policy
+## 7. Minimal Retrain Workflow Scenarios
+Source: `e2e/test_minimal_retrain.py`
 
-**Input**:
-- Valid input data
-- Transform response has critical errors
-- `mlflow_transform_filters` configured with `POLICY: 'STOP'`
+The MLflow registry is fully mocked (no real artifacts in test container).
+Validates `predictions_schema.retrain_reports` columns:
+`id, model_id, model_name, timestamp, status, version, mlflow_run_id, mlflow_experiment_id, created_at`.
 
-**Expected Behavior**:
-- `request_transform` succeeds but response invalid
-- `mlflow_response_gate` for transform returns `path_flag='STOP'`
-- Workflow exits without calling predict or export
+### 7.1 Happy path
 
-**Assertions**:
-- Transform completed but validation failed
-- `mlflow_response_gate` called for transform
-- MLFlow predict NOT called
-- Export workflow NOT called
-- Workflow completes without error
+#### MR.1.1 Successful retrain + promotion
+**Summary**: Training data loads via MinIO offload, `wrapper.retrain` succeeds,
+the new version is promoted to the `production` alias.
+
+**Expected Outcome**:
+- Report row has success status, `version='7'`, `mlflow_run_id='retrain-run-id'`,
+  `mlflow_experiment_id='experiment-id'`.
+- `mlflow.log_artifact` is called with the input CSV.
+- `promote_to_alias` is called once with the resolved version and alias.
+
+### 7.2 Failure paths
+
+#### MR.2.1 Wrapper retrain raises
+**Summary**: `wrapper.retrain` raises `RuntimeError`. The activity returns
+`success=False`, `update_production_model` is NOT invoked.
+
+**Expected Outcome**:
+- Report row carries the error message and `version`/`mlflow_*` columns are NULL.
+
+#### MR.2.2 Missing `model_config.target`
+**Summary**: Empty model config short-circuits before any MLflow call.
+
+**Expected Outcome**:
+- Report row carries the explicit guard message.
+- `get_cached_model` is never invoked.
+
+#### MR.3.1 No training data
+**Summary**: The training query returns no rows; the workflow does not
+persist any report row. The current code raises plain `ValueError` from the
+workflow function, which Temporal treats as a workflow-task failure (see
+`CODE_ISSUES.md` issue MR-1).
 
 ---
 
-#### Scenario 2.2.3: Transform Gate Triggers REPEAT
-**Description**: Transform response gate determines data should repeat last prediction
+## Input Contract Reference
 
-**Input**:
-- Valid input data
-- Transform response has quality issues that require using previous prediction
-- `mlflow_transform_filters` configured with `POLICY: 'REPEAT'`
-- `path_priority` includes REPEAT
+Common scenario input fields:
+- `schedule_name`
+- `model_name`
+- `model_id`
+- `query`
+- `schema`
+- `table_name`
+- `transform_table_name`
+- `input_filters`
+- `mlflow_transform_filters`
+- `mlflow_predict_filters`
+- `path_priority` (default order: `STOP`, `CONTINUE`, `REPEAT`)
+- `save_transform`
+- `prediction_store_policy`
+- `model_config.target`
+- `datetime_columns` (when query returns temporal fields)
 
-**Expected Behavior**:
-- `request_transform` succeeds but response has issues
-- `mlflow_response_gate` for transform returns `path_flag='REPEAT'`
-- `path_flag_handler` calls `repeat_last_prediction` activity
-- MLFlow predict skipped
-- Last prediction repeated and exported
-
-**Assertions**:
-- Transform completed but validation triggered REPEAT
-- `mlflow_response_gate` called for transform
-- MLFlow predict NOT called
-- `repeat_last_prediction` activity called
-- Workflow completes
-
----
-
-### 2.3 Predict gate Early Exit Scenarios
-
-#### Scenario 2.3.1: Predict Gate Triggers CONTINUE
-**Description**: Predict response gate determines data should continue despite issues
-
-**Input**:
-- Valid input and transform data
-- Predict response has quality issues but policy is CONTINUE
-- `mlflow_predict_filters` configured with `POLICY: 'CONTINUE'`
-- `path_priority` includes CONTINUE
-
-**Expected Behavior**:
-- `request_predict` succeeds
-- `mlflow_response_gate` for predict returns `path_flag='CONTINUE'`
-- `path_flag_handler` calls export workflow with predict data
-- Prediction exported despite quality issues
-
-**Assertions**:
-- Transform and predict completed
-- `mlflow_response_gate` called for predict
-- Export workflow called with predict data
-- Workflow completes
-
----
-
-#### Scenario 2.3.2: Predict Gate Triggers STOP
-**Description**: Prediction validation fails with STOP policy
-
-**Input**:
-- Valid input and transform
-- Predict response has critical errors
-- `mlflow_predict_filters` configured with `POLICY: 'STOP'`
-
-**Expected Behavior**:
-- `request_predict` succeeds but response invalid
-- `mlflow_response_gate` for predict returns `path_flag='STOP'`
-- Workflow exits without export
-
-**Assertions**:
-- Transform completed
-- Predict completed but validation failed
-- Export workflow NOT called
-- Workflow completes without error
-
----
-
-#### Scenario 2.3.3: Predict Gate Triggers REPEAT
-**Description**: Predict response gate determines data should repeat last prediction
-
-**Input**:
-- Valid input and transform data
-- Predict response has quality issues that require using previous prediction
-- `mlflow_predict_filters` configured with `POLICY: 'REPEAT'`
-- `path_priority` includes REPEAT
-
-**Expected Behavior**:
-- `request_predict` succeeds but response has issues
-- `mlflow_response_gate` for predict returns `path_flag='REPEAT'`
-- `path_flag_handler` calls `repeat_last_prediction` activity
-- Last prediction repeated and exported
-
-**Assertions**:
-- Transform and predict completed but validation triggered REPEAT
-- `mlflow_response_gate` called for predict
-- `repeat_last_prediction` activity called
-- Export workflow NOT called with current prediction
-- Workflow completes
-
----
-
-## 3. Format and Export Prediction - Child Workflow Scenarios
-
-### 3.1 Success Scenarios
-
-#### Scenario 3.1.1: Default Prediction Export
-**Description**: Error prediction path creates default prediction
-
-**Input**:
-- `path_flag: 'ERROR'` or other non-None value (not STOP/CONTINUE/REPEAT)
-- `comment` provided with error details
-
-**Expected Behavior**:
-- `format_default_prediction` called instead of `format_prediction`
-- Default prediction created with error metadata
-- Exported to PostgreSQL only
-- Transformed data NOT processed
-- Metrics written
-
-**Assertions**:
-- `format_default_prediction` called
-- `format_prediction` NOT called
-- `format_transformed_data` NOT called
-- One PostgreSQL export only
-- Default values in prediction data
-- Comment included
-
----
-
-#### Scenario 3.1.2: Export with OPC only
-**Description**: Export to PostgreSQL and OPC server only (no PI Web API)
-
-**Input**:
-- `path_flag: None`
-- `opc_output_config` configured with valid OPC settings
-- `pi_web_api_output_config: None` or `{}`
-
-**Expected Behavior**:
-- Normal formatting
-- PostgreSQL export executed
-- OPC export executed
-- PI Web API activity skipped
-- Metrics written with OPC metrics
-
-**Assertions**:
-- PI Web API activity NOT called
-- OPC activity called
-- PostgreSQL export called
-- Metrics written with `opc_metrics` populated
-
----
-
-#### Scenario 3.1.3: Export with PI Web API only
-**Description**: Export to PostgreSQL and PI Web API only (no OPC)
-
-**Input**:
-- `path_flag: None`
-- `pi_web_api_output_config` configured with valid PI Web API settings
-- `opc_output_config: None` or `{}`
-
-**Expected Behavior**:
-- Normal formatting
-- PostgreSQL export executed
-- PI Web API export executed
-- OPC activity skipped
-- Metrics written without OPC metrics
-
-**Assertions**:
-- OPC activity NOT called
-- PI Web API activity called
-- PostgreSQL export called
-- Metrics written with empty `opc_metrics`
-
----
-
-#### Scenario 3.1.4: Export Without Optional Outputs
-**Description**: Export only to PostgreSQL (no OPC or PI Web API)
-
-**Input**:
-- `path_flag: None`
-- `opc_output_config: None` or `{}`
-- `pi_web_api_output_config: None` or `{}`
-
-**Expected Behavior**:
-- Normal formatting
-- Only PostgreSQL export executed
-- OPC and PI Web API activities skipped
-- Metrics written without OPC metrics
-
-**Assertions**:
-- PI Web API activity NOT called
-- OPC activity NOT called
-- PostgreSQL export called
-- Metrics written with empty `opc_metrics`
-
----
-
-#### Scenario 3.1.5: Export Without Transformed Data
-**Description**: Only prediction exported, no transform table
-
-**Input**:
-- `path_flag: None`
-- `transformed_data: None` or `save_transform: False`
-- `opc_output_config: None` or `{}`
-- `pi_web_api_output_config: None` or `{}`
-
-**Expected Behavior**:
-- Only prediction formatted and exported
-- Transform export skipped
-- Single PostgreSQL write
-
-**Assertions**:
-- `format_transformed_data` NOT called
-- One PostgreSQL export
-- Transform table remains empty
-
----
-
-### 3.2 Error Scenarios
-
-These paths do **not** rely on Temporal activity retries for export failures: the write activities run once, errors are handled inside the activity, and the **workflow completes successfully** with degraded metadata on the persisted prediction (`prediction_confidence` and `comments`).
-
-#### Scenario 3.2.1: PI Web API Write Error
-**Description**: PI Web API export fails
-
-**Input**:
-- Valid prediction
-- PI Web API service unavailable or invalid config
-
-**Expected Behavior**:
-- `write_pi_web_api_data` surfaces the failure (exception handled in the activity layer)
-- Notification may be sent
-- Workflow **completes** (does not fail)
-- Prediction row is still written to PostgreSQL with error confidence **13** and a comment describing the PI error
-- Subsequent steps (e.g. OPC, Postgres) still run per workflow order with the updated prediction payload
-
-**Assertions**:
-- PI Web API error notification sent (when applicable)
-- Workflow completes
-- PostgreSQL contains the prediction with `prediction_confidence` 13 and expected `comments`
-
----
-
-#### Scenario 3.2.2: OPC Write Error
-**Description**: OPC server write fails
-
-**Input**:
-- Valid prediction
-- OPC server unavailable or invalid configuration
-
-**Expected Behavior**:
-- `write_opc_data` reports failure without aborting the workflow
-- Notification may be sent
-- Workflow **completes** (does not fail)
-- Prediction row is written to PostgreSQL with OPC error confidence **12** and a comment indicating OPC write issues
-
-**Assertions**:
-- OPC error notification sent (when applicable)
-- Workflow completes
-- PostgreSQL contains the prediction with `prediction_confidence` 12 and expected `comments`
-
----
-
-#### Scenario 3.2.3: PI Web API Partial Write Error
-**Description**: Two prediction tags attempt to be written to PI Web API, but only one succeeds
-
-**Input**:
-- Valid prediction
-- Two prediction tags configured
-- PI Web API returns partial success (one tag succeeds, one fails)
-
-**Expected Behavior**:
-- `write_pi_web_api_data` processes response
-- `process_pi_web_api_response` detects partial failure
-- Error confidence set (13)
-- Notification sent for failed tag
-- Workflow completes with error confidence (single activity attempt; no retry loop)
-
-**Assertions**:
-- One tag written successfully
-- One tag failed
-- Error confidence set in prediction
-- Error notification sent
-- Workflow completes
-
----
+Optional outputs:
+- `opc_output_config`
+- `pi_web_api_output_config`

e2e/test_drift.py

@@ -0,0 +1,786 @@
+"""
+End-to-end tests for the Drift workflow.
+
+Coverage focus:
+
+- Full pipeline persists drift rows with **all** columns expected by the
+  ``predictions_schema.drift`` table (model_id, feature, method, value, drift,
+  chunk, timestamp, timestamp_end, accurate, plus DB-managed id/created_at/updated_at).
+- ``get_reference_data`` happy path (CSV downloaded from MLflow stub) and
+  fallback path (30% of target data when reference is unavailable).
+- ``calculate_drift`` invariants: ``p_value`` dropped, duplicates removed, rows
+  outside target timestamps filtered, default ``drift_metrics`` propagated.
+- Failure paths: ``ModelAnalysis.get_drift_metrics_dataframe`` raises ⇒
+  workflow completes without writes; empty target data ⇒ workflow short-circuits;
+  invalid ``chunk_period`` ⇒ activity raises and workflow surfaces the error.
+"""
+
+from datetime import datetime, timedelta, timezone
+from decimal import Decimal
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pandas as pd
+import pytest
+from sqlalchemy import text
+from temporalio.testing import WorkflowEnvironment
+from temporalio.worker import Worker
+
+from e2e.helpers import (
+    build_drift_dataframe,
+    insert_target_data_for_drift,
+    load_scenario_input,
+    make_workflow_id,
+    start_and_await_workflow,
+)
+from laborious.activities.activities import Activities
+from laborious.workflows.drift import Drift
+
+# Drift columns that must be present (and non-null where required) on every row.
+EXPECTED_DRIFT_COLUMNS = [
+    'id',
+    'model_id',
+    'feature',
+    'method',
+    'value',
+    'drift',
+    'chunk',
+    'timestamp',
+    'timestamp_end',
+    'accurate',
+    'created_at',
+    'updated_at',
+]
+
+
+def _drift_input(model_id: int, **overrides) -> dict:
+    """Load the base drift scenario JSON and apply ad-hoc overrides."""
+    input_data = load_scenario_input('drift_base.json', model_id=model_id)
+    input_data.update(overrides)
+    return input_data
+
+
+def _five_minute_window(offset_minutes: int = 6) -> tuple[list[str], list[str]]:
+    """
+    Build five consecutive UTC minute timestamps positioned in the recent past.
+
+    The Drift workflow filters target rows with ``timestamp > NOW() - INTERVAL``,
+    so timestamps must be recent for tests to retrieve any data. We snap to
+    minute precision and back off ``offset_minutes`` minutes so all chunks land
+    well inside the default 60-minute interval defined in ``drift_base.json``.
+
+    Args:
+     - offset_minutes: How many minutes ago the most recent chunk should be.
+
+    Return:
+     Tuple ``(target_timestamps, chunk_timestamps)``:
+
+     - ``target_timestamps``: ISO strings with ``+0000`` used as ``timestamp``
+       and ``created_at`` columns when inserting target rows.
+     - ``chunk_timestamps``: ``YYYY-MM-DD HH:MM`` truncations matching what
+       ``calculate_drift`` filters on for ``chunk_period='min'``.
+    """
+    base = datetime.now(timezone.utc).replace(second=0, microsecond=0) - timedelta(
+        minutes=offset_minutes
+    )
+    target_timestamps = [
+        (base + timedelta(minutes=i)).strftime('%Y-%m-%d %H:%M:%S%z') for i in range(5)
+    ]
+    chunk_timestamps = [
+        (base + timedelta(minutes=i)).strftime('%Y-%m-%d %H:%M') for i in range(5)
+    ]
+    return target_timestamps, chunk_timestamps
+
+
+def _configure_reference_csv(mlflow_repository_stub, reference_rows: pd.DataFrame) -> None:
+    """
+    Wire ``mlflow_repository_stub`` so ``get_reference_data`` returns
+    ``reference_rows`` by writing them to ``dst_path/evaluation_data.csv``.
+    """
+
+    def _download(run_id: str, artifact_path: str, dst_path: str, metadata=None):
+        target = Path(dst_path) / 'evaluation_data.csv'
+        reference_rows.to_csv(target, index=False)
+
+    mlflow_repository_stub._client.get_model_version_by_alias.return_value = MagicMock(
+        run_id='fake-reference-run'
+    )
+    mlflow_repository_stub.download_artifacts.side_effect = _download
+
+
+def _force_reference_unavailable(mlflow_repository_stub) -> None:
+    """Make ``get_reference_data`` return ``None`` by failing alias resolution."""
+    mlflow_repository_stub._client.get_model_version_by_alias.side_effect = Exception(
+        'no production alias registered'
+    )
+
+
+def _assert_drift_columns_complete(rows, *, expected_count: int) -> None:
+    """Validate row count and that the canonical drift columns are populated."""
+    assert len(rows) == expected_count, (
+        f'Expected {expected_count} drift rows persisted, got {len(rows)}'
+    )
+    seen_columns = set(rows[0]._mapping.keys()) if rows else set()
+    for column in EXPECTED_DRIFT_COLUMNS:
+        assert column in seen_columns, f'Missing drift column in postgres: {column}'
+
+    for row in rows:
+        mapping = dict(row._mapping)
+        for column in EXPECTED_DRIFT_COLUMNS:
+            if column == 'value':
+                # ``value`` is nullable in the table; skip null check, only ensure key exists.
+                continue
+            assert mapping[column] is not None, f"Column '{column}' is NULL in {mapping}"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_drift_happy_path_persists_all_columns_with_reference_data(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_drift: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mlflow_repository_stub,
+    model_analysis_stub,
+):
+    """
+    Scenario D.1.1: Happy path with reference data downloaded from MLflow.
+
+    Validates the full drift pipeline produces one row per (chunk, feature, method)
+    combination plus the multivariate row block, with **all** columns required by
+    ``predictions_schema.drift`` populated. Uses a deterministic mocked drift
+    dataframe so the postgres assertions remain stable across runs.
+    """
+    client = temporal_test_env.client
+    model_id = 411
+
+    target_timestamps, chunk_timestamps = _five_minute_window()
+    insert_target_data_for_drift(
+        postgres_engine,
+        model_id=model_id,
+        timestamps=target_timestamps,
+        variables_values={
+            'sensor_1': [10.0, 11.0, 12.0, 13.0, 14.0],
+            'sensor_2': [20.0, 21.0, 22.0, 23.0, 24.0],
+        },
+    )
+
+    reference_df = pd.DataFrame(
+        {
+            'timestamp': ['2023-12-31 11:00:00+00:00', '2023-12-31 11:01:00+00:00'],
+            'sensor_1': [9.5, 9.6],
+            'sensor_2': [19.5, 19.6],
+        }
+    )
+    _configure_reference_csv(mlflow_repository_stub, reference_df)
+
+    features = ['sensor_1', 'sensor_2']
+    methods = ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']
+    drift_flags = {
+        ('sensor_1', 'wasserstein'): True,
+        ('sensor_2', 'wasserstein'): True,
+        ('sensor_2', 'jensen_shannon'): True,
+    }
+
+    model_analysis_stub.set_drift_dataframe(
+        lambda univariate, multivariate: build_drift_dataframe(
+            timestamps=chunk_timestamps,
+            features=features,
+            methods=methods,
+            statistic=0.42,
+            drift_flags=drift_flags,
+            multivariate_value=15.5,
+            multivariate_drift=True,
+        )
+    )
+
+    input_data = _drift_input(model_id)
+    await start_and_await_workflow(
+        client, Drift.run, input_data, make_workflow_id('test-drift-happy-path')
+    )
+
+    with postgres_engine.connect() as conn:
+        rows = (
+            conn.execute(
+                text(
+                    'SELECT * FROM predictions_schema.drift '
+                    'WHERE model_id = :model_id ORDER BY chunk, feature, method'
+                ),
+                {'model_id': model_id},
+            )
+            .mappings()
+            .all()
+        )
+
+    expected_rows = len(chunk_timestamps) * (len(features) * len(methods) + 1)  # univariate + multivariate
+    assert len(rows) == expected_rows
+    for column in EXPECTED_DRIFT_COLUMNS:
+        assert column in rows[0], f'Missing drift column in postgres: {column}'
+
+    for row in rows:
+        for column in EXPECTED_DRIFT_COLUMNS:
+            if column == 'value':
+                continue
+            assert row[column] is not None, f"Column '{column}' is NULL in {dict(row)}"
+
+    multivariate_rows = [r for r in rows if r['feature'] == 'multivariate']
+    univariate_rows = [r for r in rows if r['feature'] != 'multivariate']
+
+    assert len(multivariate_rows) == len(chunk_timestamps)
+    assert all(r['method'] == 'multivariate' for r in multivariate_rows)
+    assert all(r['drift'] is True for r in multivariate_rows)
+    assert all(Decimal(str(r['value'])) == Decimal('15.5') for r in multivariate_rows)
+
+    assert len(univariate_rows) == len(features) * len(methods) * len(chunk_timestamps)
+    assert {r['method'] for r in univariate_rows} == set(methods)
+    assert {r['feature'] for r in univariate_rows} == set(features)
+
+    drift_pairs = {(r['feature'], r['method']): r['drift'] for r in univariate_rows}
+    for (feature, method), expected_drift in drift_flags.items():
+        assert drift_pairs[(feature, method)] is expected_drift
+
+    assert all(r['accurate'] is True for r in rows), 'reference path should mark rows as accurate'
+
+    assert all(
+        r['timestamp_end'].endswith(':59.999999999') and 'T' in r['timestamp_end']
+        for r in rows
+    ), 'timestamp_end should preserve the high-precision ISO string from ModelAnalysis'
+
+    assert all('p_value' not in r for r in rows), 'p_value must be dropped before postgres'
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_drift_uses_30pct_fallback_when_reference_unavailable(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_drift: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mlflow_repository_stub,
+    model_analysis_stub,
+    notification_inserts,
+):
+    """
+    Scenario D.1.2: ``get_reference_data`` returns ``None`` (production alias missing),
+    so ``calculate_drift`` falls back to using the first 30% of target rows as
+    reference. Persisted rows must report ``accurate=False`` and a warning
+    notification must be emitted to mongo.
+    """
+    client = temporal_test_env.client
+    model_id = 412
+
+    target_timestamps, chunk_timestamps = _five_minute_window()
+    insert_target_data_for_drift(
+        postgres_engine,
+        model_id=model_id,
+        timestamps=target_timestamps,
+        variables_values={
+            'sensor_1': [10.0, 10.1, 10.2, 10.3, 10.4],
+            'sensor_2': [20.0, 20.1, 20.2, 20.3, 20.4],
+        },
+    )
+
+    _force_reference_unavailable(mlflow_repository_stub)
+
+    model_analysis_stub.set_drift_dataframe(
+        lambda univariate, multivariate: build_drift_dataframe(
+            timestamps=chunk_timestamps,
+            features=['sensor_1'],
+            methods=['kolmogorov_smirnov'],
+            statistic=0.7,
+            include_multivariate=False,
+        )
+    )
+
+    input_data = _drift_input(model_id)
+
+    await start_and_await_workflow(
+        client, Drift.run, input_data, make_workflow_id('test-drift-fallback')
+    )
+
+    with postgres_engine.connect() as conn:
+        rows = (
+            conn.execute(
+                text('SELECT * FROM predictions_schema.drift WHERE model_id = :m'),
+                {'m': model_id},
+            )
+            .mappings()
+            .all()
+        )
+
+    assert len(rows) == len(chunk_timestamps)
+    assert all(r['accurate'] is False for r in rows), 'fallback path must mark rows as inaccurate'
+    assert all(r['feature'] == 'sensor_1' for r in rows)
+
+    fallback_warnings = [
+        call
+        for call in notification_inserts.call_args_list
+        if call.args
+        and isinstance(call.args[0], dict)
+        and call.args[0].get('notification_id') == 'MODEL_METRICS_REFERENCE_DATA_WARNING'
+    ]
+    assert len(fallback_warnings) >= 1, 'expected reference fallback warning notification'
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_drift_drops_p_value_and_dedupes_by_timestamp_method_feature(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_drift: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mlflow_repository_stub,
+    model_analysis_stub,
+):
+    """
+    Scenario D.2.1: ModelAnalysis returns duplicate (timestamp, method, feature)
+    rows and a populated ``p_value`` column. ``calculate_drift`` must deduplicate
+    keeping the first occurrence and the persisted rows must not contain
+    ``p_value``.
+    """
+    client = temporal_test_env.client
+    model_id = 421
+
+    target_timestamps, chunk_timestamps = _five_minute_window()
+    insert_target_data_for_drift(
+        postgres_engine,
+        model_id=model_id,
+        timestamps=target_timestamps,
+        variables_values={
+            'sensor_1': [1.0, 2.0, 3.0, 4.0, 5.0],
+            'sensor_2': [10.0, 20.0, 30.0, 40.0, 50.0],
+        },
+    )
+
+    duplicates = [
+        # Re-emit the first chunk for sensor_1/kolmogorov_smirnov with a different value
+        # so we can prove the dedup keeps the FIRST row.
+        {
+            'timestamp': pd.Timestamp(chunk_timestamps[0]),
+            'feature': 'sensor_1',
+            'metric': 'kolmogorov_smirnov',
+            'statistic': 0.99,
+            'p_value': 0.02,
+            'alert': True,
+            'chunk_index': 0,
+            'chunk_start_date': pd.Timestamp(chunk_timestamps[0]),
+            'chunk_end_date': pd.Timestamp(f'{chunk_timestamps[0]}:59.999999999'),
+        }
+    ]
+
+    model_analysis_stub.set_drift_dataframe(
+        lambda univariate, multivariate: build_drift_dataframe(
+            timestamps=chunk_timestamps,
+            features=['sensor_1'],
+            methods=['kolmogorov_smirnov'],
+            statistic=0.5,
+            include_multivariate=False,
+            extra_rows=duplicates,
+        )
+    )
+    _force_reference_unavailable(mlflow_repository_stub)
+
+    input_data = _drift_input(model_id)
+    await start_and_await_workflow(
+        client, Drift.run, input_data, make_workflow_id('test-drift-dedupe')
+    )
+
+    with postgres_engine.connect() as conn:
+        rows = (
+            conn.execute(
+                text(
+                    'SELECT chunk, feature, method, value FROM predictions_schema.drift '
+                    'WHERE model_id = :m ORDER BY chunk'
+                ),
+                {'m': model_id},
+            )
+            .mappings()
+            .all()
+        )
+
+    assert len(rows) == len(chunk_timestamps), 'duplicates must be removed before persistence'
+
+    first_chunk_rows = [r for r in rows if r['chunk'] == 0]
+    assert len(first_chunk_rows) == 1
+    assert Decimal(str(first_chunk_rows[0]['value'])) == Decimal('0.5'), (
+        'dedup must keep the FIRST occurrence (statistic=0.5), not the duplicate (statistic=0.99)'
+    )
+
+    columns = set(rows[0].keys())
+    assert 'p_value' not in columns
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_drift_drops_rows_outside_target_timestamps(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_drift: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mlflow_repository_stub,
+    model_analysis_stub,
+):
+    """
+    Scenario D.2.2: Drift rows whose timestamps are not present in the target
+    data (e.g. came from the reference distribution) must be discarded so the
+    saved drift only reflects analysis chunks.
+    """
+    client = temporal_test_env.client
+    model_id = 422
+
+    target_timestamps, chunk_timestamps = _five_minute_window()
+    insert_target_data_for_drift(
+        postgres_engine,
+        model_id=model_id,
+        timestamps=target_timestamps,
+        variables_values={
+            'sensor_1': [1.0, 2.0, 3.0, 4.0, 5.0],
+            'sensor_2': [10.0, 20.0, 30.0, 40.0, 50.0],
+        },
+    )
+
+    out_of_range_timestamp = pd.Timestamp('2099-12-31 23:59')
+    extra = [
+        {
+            'timestamp': out_of_range_timestamp,
+            'feature': 'sensor_1',
+            'metric': 'kolmogorov_smirnov',
+            'statistic': 0.5,
+            'p_value': 0.0,
+            'alert': True,
+            'chunk_index': 99,
+            'chunk_start_date': out_of_range_timestamp,
+            'chunk_end_date': pd.Timestamp('2099-12-31 23:59:59.999999999'),
+        }
+    ]
+
+    model_analysis_stub.set_drift_dataframe(
+        lambda univariate, multivariate: build_drift_dataframe(
+            timestamps=chunk_timestamps,
+            features=['sensor_1'],
+            methods=['kolmogorov_smirnov'],
+            include_multivariate=False,
+            extra_rows=extra,
+        )
+    )
+    _force_reference_unavailable(mlflow_repository_stub)
+
+    input_data = _drift_input(model_id)
+    await start_and_await_workflow(
+        client, Drift.run, input_data, make_workflow_id('test-drift-tts-filter')
+    )
+
+    with postgres_engine.connect() as conn:
+        chunks = [
+            r[0]
+            for r in conn.execute(
+                text(
+                    'SELECT chunk FROM predictions_schema.drift WHERE model_id = :m ORDER BY chunk'
+                ),
+                {'m': model_id},
+            ).all()
+        ]
+
+    assert chunks == [0, 1, 2, 3, 4], 'out-of-range timestamps must be filtered out'
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_drift_empty_target_data_short_circuits_workflow(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_drift: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mlflow_repository_stub,
+    model_analysis_stub,
+):
+    """
+    Scenario D.3.1: When ``load_custom_query`` returns no rows the workflow must
+    return early without invoking ModelAnalysis or writing any drift rows.
+    """
+    client = temporal_test_env.client
+    model_id = 431
+
+    with postgres_engine.begin() as conn:
+        conn.execute(text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}'))
+
+    _force_reference_unavailable(mlflow_repository_stub)
+
+    drift_calls = {'count': 0}
+
+    def _factory(univariate, multivariate):
+        drift_calls['count'] += 1
+        return build_drift_dataframe(
+            timestamps=['2024-01-01 12:00'],
+            features=['sensor_1'],
+            methods=['kolmogorov_smirnov'],
+            include_multivariate=False,
+        )
+
+    model_analysis_stub.set_drift_dataframe(_factory)
+
+    input_data = _drift_input(model_id)
+    await start_and_await_workflow(
+        client, Drift.run, input_data, make_workflow_id('test-drift-empty-target')
+    )
+
+    with postgres_engine.connect() as conn:
+        count = conn.execute(
+            text('SELECT COUNT(*) FROM predictions_schema.drift WHERE model_id = :m'),
+            {'m': model_id},
+        ).scalar()
+    assert count == 0
+    assert drift_calls['count'] == 0, 'ModelAnalysis must not be invoked when target data is empty'
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_drift_model_analysis_failure_keeps_workflow_alive_no_writes(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_drift: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mlflow_repository_stub,
+    model_analysis_stub,
+    notification_inserts,
+):
+    """
+    Scenario D.3.2: ``ModelAnalysis.get_drift_metrics_dataframe`` raises. The
+    activity must catch the error, send an error notification and return ``[]``
+    so the workflow completes without persisting drift rows.
+    """
+    client = temporal_test_env.client
+    model_id = 432
+
+    target_timestamps, _ = _five_minute_window()
+    insert_target_data_for_drift(
+        postgres_engine,
+        model_id=model_id,
+        timestamps=target_timestamps,
+        variables_values={
+            'sensor_1': [1.0, 2.0, 3.0, 4.0, 5.0],
+            'sensor_2': [10.0, 20.0, 30.0, 40.0, 50.0],
+        },
+    )
+
+    _force_reference_unavailable(mlflow_repository_stub)
+    model_analysis_stub.raise_on_get_drift_metrics_dataframe(
+        RuntimeError('drift analyzer crashed')
+    )
+
+    input_data = _drift_input(model_id)
+    await start_and_await_workflow(
+        client, Drift.run, input_data, make_workflow_id('test-drift-analyzer-error')
+    )
+
+    with postgres_engine.connect() as conn:
+        count = conn.execute(
+            text('SELECT COUNT(*) FROM predictions_schema.drift WHERE model_id = :m'),
+            {'m': model_id},
+        ).scalar()
+    assert count == 0
+
+    error_notifications = [
+        call
+        for call in notification_inserts.call_args_list
+        if call.args
+        and isinstance(call.args[0], dict)
+        and call.args[0].get('notification_id') == 'MODEL_METRICS_GET_DRIFT_METRICS_ERROR'
+    ]
+    assert len(error_notifications) >= 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_drift_default_drift_metrics_propagated_to_analyzer(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_drift: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mlflow_repository_stub,
+    model_analysis_stub,
+):
+    """
+    Scenario D.4.1: When ``drift_metrics`` is omitted from input the workflow
+    must default to ``['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']``
+    and forward exactly that list to ``ModelAnalysis.detect_univariate_drift``.
+    """
+    client = temporal_test_env.client
+    model_id = 441
+
+    target_timestamps, chunk_timestamps = _five_minute_window()
+    insert_target_data_for_drift(
+        postgres_engine,
+        model_id=model_id,
+        timestamps=target_timestamps,
+        variables_values={
+            'sensor_1': [1.0, 2.0, 3.0, 4.0, 5.0],
+            'sensor_2': [10.0, 20.0, 30.0, 40.0, 50.0],
+        },
+    )
+
+    _force_reference_unavailable(mlflow_repository_stub)
+    model_analysis_stub.set_drift_dataframe(
+        lambda univariate, multivariate: build_drift_dataframe(
+            timestamps=chunk_timestamps,
+            features=['sensor_1'],
+            methods=['kolmogorov_smirnov'],
+            include_multivariate=False,
+        )
+    )
+
+    input_data = _drift_input(model_id)
+    input_data.pop('drift_metrics', None)
+
+    await start_and_await_workflow(
+        client, Drift.run, input_data, make_workflow_id('test-drift-default-methods')
+    )
+
+    instance = model_analysis_stub.last_instance
+    assert instance is not None, 'ModelAnalysis must have been instantiated'
+    assert len(instance.detect_univariate_drift_calls) == 1
+    forwarded_methods = instance.detect_univariate_drift_calls[0]['methods']
+    assert forwarded_methods == ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_drift_invalid_chunk_period_raises_value_error(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_drift: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mlflow_repository_stub,
+):
+    """
+    Scenario D.4.2: ``calculate_drift`` validates ``chunk_period`` and rejects
+    anything other than ``min`` / ``s``. The workflow must surface the
+    ``ValueError`` to the caller and persist nothing.
+    """
+    client = temporal_test_env.client
+    model_id = 442
+
+    target_timestamps, _ = _five_minute_window()
+    insert_target_data_for_drift(
+        postgres_engine,
+        model_id=model_id,
+        timestamps=target_timestamps,
+        variables_values={
+            'sensor_1': [1.0, 2.0, 3.0, 4.0, 5.0],
+            'sensor_2': [10.0, 20.0, 30.0, 40.0, 50.0],
+        },
+    )
+    _force_reference_unavailable(mlflow_repository_stub)
+
+    input_data = _drift_input(model_id, chunk_period='hour')
+
+    with pytest.raises(Exception) as excinfo:
+        await start_and_await_workflow(
+            client,
+            Drift.run,
+            input_data,
+            make_workflow_id('test-drift-bad-chunk-period'),
+        )
+
+    # Temporal wraps the activity ValueError in WorkflowFailureError; the message
+    # may live on ``.message`` or ``str(exc)`` depending on the SDK error class.
+    cause_descriptions = []
+    current: BaseException | None = excinfo.value
+    while current is not None:
+        cause_descriptions.append(
+            getattr(current, 'message', None) or str(current) or repr(current)
+        )
+        current = current.__cause__
+    assert any('Invalid chunk period' in msg for msg in cause_descriptions), (
+        f'Expected ValueError about chunk period in chain, got: {cause_descriptions}'
+    )
+
+    with postgres_engine.connect() as conn:
+        count = conn.execute(
+            text('SELECT COUNT(*) FROM predictions_schema.drift WHERE model_id = :m'),
+            {'m': model_id},
+        ).scalar()
+    assert count == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_drift_chunk_period_seconds_preserves_seconds_in_timestamp_column(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_drift: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mlflow_repository_stub,
+    model_analysis_stub,
+):
+    """
+    Scenario D.4.3: With ``chunk_period='s'`` the persisted ``timestamp``
+    column must preserve second-level precision instead of being flattened to
+    the start of the minute, and ``chunk_period`` must be propagated to the
+    analyzer so it actually chunks at second granularity.
+    """
+    client = temporal_test_env.client
+    model_id = 443
+
+    base = datetime.now(timezone.utc).replace(second=0, microsecond=0) - timedelta(minutes=2)
+    target_timestamps = [
+        base.strftime('%Y-%m-%d %H:%M:%S%z'),
+        (base + timedelta(seconds=30)).strftime('%Y-%m-%d %H:%M:%S%z'),
+        (base + timedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S%z'),
+    ]
+    chunk_timestamps_full = [
+        base.strftime('%Y-%m-%d %H:%M:%S'),
+        (base + timedelta(seconds=30)).strftime('%Y-%m-%d %H:%M:%S'),
+        (base + timedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S'),
+    ]
+
+    insert_target_data_for_drift(
+        postgres_engine,
+        model_id=model_id,
+        timestamps=target_timestamps,
+        variables_values={
+            'sensor_1': [1.0, 2.0, 3.0],
+            'sensor_2': [10.0, 20.0, 30.0],
+        },
+    )
+    _force_reference_unavailable(mlflow_repository_stub)
+
+    model_analysis_stub.set_drift_dataframe(
+        lambda univariate, multivariate: build_drift_dataframe(
+            timestamps=chunk_timestamps_full,
+            features=['sensor_1'],
+            methods=['kolmogorov_smirnov'],
+            include_multivariate=False,
+        )
+    )
+
+    input_data = _drift_input(model_id, chunk_period='s')
+    await start_and_await_workflow(
+        client,
+        Drift.run,
+        input_data,
+        make_workflow_id('test-drift-chunk-seconds'),
+    )
+
+    with postgres_engine.connect() as conn:
+        rows = (
+            conn.execute(
+                text(
+                    'SELECT chunk, timestamp FROM predictions_schema.drift '
+                    'WHERE model_id = :m ORDER BY chunk'
+                ),
+                {'m': model_id},
+            )
+            .mappings()
+            .all()
+        )
+
+    assert [r['chunk'] for r in rows] == [0, 1, 2]
+    seconds_present = {r['timestamp'].second for r in rows}
+    assert seconds_present == {0, 30}, (
+        f'expected seconds 0 and 30 to be preserved, got {seconds_present}'
+    )
+
+    instance = model_analysis_stub.last_instance
+    assert instance is not None
+    assert instance.detect_univariate_drift_calls[0]['chunk_period'] == 's'

e2e/test_minimal_retrain.py

@@ -0,0 +1,335 @@
+"""
+End-to-end tests for the MinimalRetrain workflow.
+
+The MLflow registry is fully stubbed because no real artifacts exist in a
+test container; we only validate that the workflow:
+
+- Loads training data via ``load_query_with_minio_offload``.
+- Calls ``retrain_model`` with a payload pointing at MinIO.
+- Calls ``update_production_model`` only when retrain succeeds.
+- Persists ``retrain_reports`` rows with all required columns; success rows
+  carry the new ``version`` / ``mlflow_run_id`` / ``mlflow_experiment_id``
+  while failure rows leave them ``NULL``.
+"""
+
+from contextlib import contextmanager
+from datetime import datetime, timedelta, timezone
+from unittest.mock import MagicMock, patch
+
+import pytest
+from sqlalchemy import text
+from temporalio.testing import WorkflowEnvironment
+from temporalio.worker import Worker
+
+from e2e.helpers import (
+    insert_target_data_for_drift,
+    load_scenario_input,
+    make_workflow_id,
+    start_and_await_workflow,
+)
+from laborious.activities.activities import Activities
+from laborious.workflows.minimal_retrain import MinimalRetrain
+
+EXPECTED_RETRAIN_REPORT_COLUMNS = [
+    'id',
+    'model_id',
+    'model_name',
+    'timestamp',
+    'status',
+    'version',
+    'mlflow_run_id',
+    'mlflow_experiment_id',
+    'created_at',
+]
+
+
+def _retrain_input(model_id: int, **overrides) -> dict:
+    """Load and override the minimal-retrain base scenario."""
+    payload = load_scenario_input('minimal_retrain_base.json', model_id=model_id)
+    payload.update(overrides)
+    return payload
+
+
+def _seed_retrain_training_rows(postgres_engine, model_id: int) -> None:
+    """
+    Insert training rows in long format that pivot cleanly into
+    ``index=timestamp`` / ``columns={sensor_1, sensor_2}`` for ``retrain_model``.
+    """
+    target_timestamps = [
+        (datetime.now(timezone.utc).replace(second=0, microsecond=0) - timedelta(minutes=10 - i))
+        .strftime('%Y-%m-%d %H:%M:%S%z')
+        for i in range(5)
+    ]
+    insert_target_data_for_drift(
+        postgres_engine,
+        model_id=model_id,
+        timestamps=target_timestamps,
+        variables_values={
+            'sensor_1': [10.0, 11.0, 12.0, 13.0, 14.0],
+            'sensor_2': [20.0, 21.0, 22.0, 23.0, 24.0],
+        },
+    )
+
+
+def _configure_retrain_happy_path(mlflow_repository_stub) -> None:
+    """
+    Wire ``mlflow_repository_stub`` so retrain + update_production succeed.
+
+    Mocks (in order of consumption):
+
+    - ``_client.get_model_version_by_alias``: returns ``mv`` with a stable
+      ``run_id`` (used as ``source_run_id``).
+    - ``get_cached_model``: returns a wrapper exposing inert ``retrain`` and
+      ``store_model`` methods.
+    - ``start_run``: returns a context manager yielding a ``run_info`` with
+      run/experiment ids.
+    - ``log_params``: inert.
+    - ``_client.search_model_versions``: returns one registry entry whose
+      ``version`` is promoted by ``update_production_model``.
+    - ``promote_to_alias``: inert success.
+    """
+    mv_src = MagicMock()
+    mv_src.run_id = 'source-run-id'
+
+    new_version = MagicMock()
+    new_version.version = '7'
+    new_version.run_id = 'retrain-run-id'
+
+    mlflow_repository_stub._client.get_model_version_by_alias.return_value = mv_src
+
+    cached_wrapper = MagicMock()
+    cached_wrapper.retrain = MagicMock(return_value=None)
+    cached_wrapper.store_model = MagicMock(return_value=None)
+    mlflow_repository_stub.get_cached_model.return_value = cached_wrapper
+
+    @contextmanager
+    def fake_start_run(**kwargs):
+        run_info = MagicMock()
+        run_info.run_id = 'retrain-run-id'
+        run_info.experiment_id = 'experiment-id'
+        yield run_info
+
+    mlflow_repository_stub.start_run.side_effect = fake_start_run
+    mlflow_repository_stub.log_params = MagicMock(return_value=None)
+    mlflow_repository_stub._client.search_model_versions.return_value = [new_version]
+    mlflow_repository_stub.promote_to_alias = MagicMock(return_value=None)
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_minimal_retrain_happy_path_writes_success_report(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_minimal_retrain: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mlflow_repository_stub,
+):
+    """
+    Scenario MR.1.1: Retrain succeeds. ``retrain_reports`` must contain a
+    success row with version/mlflow_run_id/mlflow_experiment_id populated and
+    the registry must have been told to promote the new version to the
+    configured alias.
+    """
+    client = temporal_test_env.client
+    model_id = 711
+
+    _seed_retrain_training_rows(postgres_engine, model_id)
+    _configure_retrain_happy_path(mlflow_repository_stub)
+
+    input_data = _retrain_input(model_id)
+
+    with patch('laborious.activities.mlflow.mlflow.log_artifact') as log_artifact_mock:
+        await start_and_await_workflow(
+            client,
+            MinimalRetrain.run,
+            input_data,
+            make_workflow_id('test-retrain-happy'),
+        )
+
+    assert log_artifact_mock.called, 'retrain_model should log the input CSV artifact'
+
+    with postgres_engine.connect() as conn:
+        rows = (
+            conn.execute(
+                text(
+                    'SELECT * FROM predictions_schema.retrain_reports '
+                    'WHERE model_id = :m'
+                ),
+                {'m': model_id},
+            )
+            .mappings()
+            .all()
+        )
+
+    assert len(rows) == 1
+    for column in EXPECTED_RETRAIN_REPORT_COLUMNS:
+        assert column in rows[0], f'Missing retrain report column: {column}'
+
+    row = rows[0]
+    assert row['model_id'] == model_id
+    assert row['model_name'] == 'test_model'
+    assert row['status'] == 'Model retrained successfully.'
+    assert row['version'] == '7'
+    assert row['mlflow_run_id'] == 'retrain-run-id'
+    assert row['mlflow_experiment_id'] == 'experiment-id'
+    assert row['timestamp'] is not None
+    assert row['created_at'] is not None
+
+    mlflow_repository_stub.promote_to_alias.assert_called_once()
+    promote_kwargs = mlflow_repository_stub.promote_to_alias.call_args.kwargs
+    assert promote_kwargs['model_name'] == 'test_model'
+    assert promote_kwargs['version'] == '7'
+    assert promote_kwargs['alias'] == 'production'
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_minimal_retrain_failure_writes_report_without_version_columns(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_minimal_retrain: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mlflow_repository_stub,
+):
+    """
+    Scenario MR.2.1: ``wrapper.retrain`` raises. The activity must catch the
+    error, return ``success=False`` so ``update_production_model`` is skipped,
+    and ``format_retrain_report`` must produce a row with the error message
+    and NULL version columns.
+    """
+    client = temporal_test_env.client
+    model_id = 721
+
+    _seed_retrain_training_rows(postgres_engine, model_id)
+    _configure_retrain_happy_path(mlflow_repository_stub)
+    mlflow_repository_stub.get_cached_model.return_value.retrain.side_effect = RuntimeError(
+        'training did not converge'
+    )
+
+    input_data = _retrain_input(model_id)
+
+    with patch('laborious.activities.mlflow.mlflow.log_artifact'):
+        await start_and_await_workflow(
+            client,
+            MinimalRetrain.run,
+            input_data,
+            make_workflow_id('test-retrain-failure'),
+        )
+
+    with postgres_engine.connect() as conn:
+        rows = (
+            conn.execute(
+                text(
+                    'SELECT * FROM predictions_schema.retrain_reports '
+                    'WHERE model_id = :m'
+                ),
+                {'m': model_id},
+            )
+            .mappings()
+            .all()
+        )
+
+    assert len(rows) == 1
+    row = rows[0]
+    assert row['model_id'] == model_id
+    assert row['model_name'] == 'test_model'
+    assert 'training did not converge' in row['status']
+    assert row['version'] is None
+    assert row['mlflow_run_id'] is None
+    assert row['mlflow_experiment_id'] is None
+
+    mlflow_repository_stub.promote_to_alias.assert_not_called()
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_minimal_retrain_missing_target_writes_failure_report(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_minimal_retrain: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mlflow_repository_stub,
+):
+    """
+    Scenario MR.2.2: ``model_config`` does not declare ``target``. The retrain
+    activity must short-circuit before any MLflow call and the report row must
+    carry the explicit guard message.
+    """
+    client = temporal_test_env.client
+    model_id = 722
+
+    _seed_retrain_training_rows(postgres_engine, model_id)
+    _configure_retrain_happy_path(mlflow_repository_stub)
+
+    input_data = _retrain_input(model_id, model_config={})
+
+    with patch('laborious.activities.mlflow.mlflow.log_artifact'):
+        await start_and_await_workflow(
+            client,
+            MinimalRetrain.run,
+            input_data,
+            make_workflow_id('test-retrain-missing-target'),
+        )
+
+    with postgres_engine.connect() as conn:
+        row = (
+            conn.execute(
+                text(
+                    'SELECT * FROM predictions_schema.retrain_reports '
+                    'WHERE model_id = :m'
+                ),
+                {'m': model_id},
+            )
+            .mappings()
+            .first()
+        )
+
+    assert row is not None
+    assert 'target' in row['status'].lower(), (
+        f"expected target-missing message, got status={row['status']!r}"
+    )
+    assert row['version'] is None
+    mlflow_repository_stub.get_cached_model.assert_not_called()
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_minimal_retrain_no_training_data_does_not_persist_report(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_minimal_retrain: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mlflow_repository_stub,
+):
+    """
+    Scenario MR.3.1: When the training query returns no rows the workflow must
+    not persist any report row. The workflow currently raises plain
+    ``ValueError`` which Temporal treats as a workflow-task failure (causing
+    indefinite retries until the test environment times out), so the assertion
+    here is constrained to the persistence side-effect. See ``e2e/CODE_ISSUES.md``
+    issue MR-1 for the recommended ``ApplicationError`` fix.
+    """
+    client = temporal_test_env.client
+    model_id = 731
+
+    with postgres_engine.begin() as conn:
+        conn.execute(text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}'))
+
+    _configure_retrain_happy_path(mlflow_repository_stub)
+
+    input_data = _retrain_input(model_id)
+
+    with pytest.raises(Exception), patch('laborious.activities.mlflow.mlflow.log_artifact'):
+        await start_and_await_workflow(
+            client,
+            MinimalRetrain.run,
+            input_data,
+            make_workflow_id('test-retrain-no-data'),
+        )
+
+    with postgres_engine.connect() as conn:
+        count = conn.execute(
+            text('SELECT COUNT(*) FROM predictions_schema.retrain_reports WHERE model_id = :m'),
+            {'m': model_id},
+        ).scalar()
+    assert count == 0

e2e/test_minio_offload.py

@@ -9,7 +9,12 @@ from sqlalchemy import text
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
 
-from e2e.helpers import insert_sample_data, make_workflow_id, start_and_await_workflow
+from e2e.helpers import (
+    insert_sample_data,
+    load_scenario_input,
+    make_workflow_id,
+    start_and_await_workflow,
+)
 from laborious.activities.activities import Activities
 from laborious.utils.models import minio_dataframe_payload as mdp
 from laborious.workflows.predictions_batch import PredictionsBatch
@@ -32,30 +37,14 @@ async def test_load_query_with_minio_offload_writes_object_to_bucket(
         conn.execute(text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}'))
     insert_sample_data(postgres_engine, model_id, [1.0, 2.0])
 
-    metadata = {
-        'metadata': {
-            'schedule_name': 'test-schedule',
-            'model_name': 'test_model',
-            'model_id': model_id,
-            'workflow_name': 'predictions_batch',
-        }
-    }
+    scenario_input = load_scenario_input('minio_offload_load_query.json', model_id=model_id)
+    metadata = {'metadata': scenario_input['metadata']}
     with patch.object(mdp, 'OFFLOAD_THRESHOLD_BYTES', 1):
-        payload = await test_activities_real_minio.load_query_with_minio_offload(
-            {
-                **metadata,
-                'query': (
-                    'SELECT timestamp, variable, value, created_at '
-                    f'FROM predictions_schema.laborious_data WHERE model_id = {model_id}'
-                ),
-                'model_name': 'test_model',
-                'datetime_columns': ['timestamp', 'created_at'],
-            }
-        )
+        payload = test_activities_real_minio.load_query_with_minio_offload(scenario_input)
     assert payload.object_key, 'offloaded payload must reference a MinIO object'
     assert payload.data is None or payload.data == {}, 'large payloads should not inline tabular dict'
 
-    df = await payload.retrieve(test_activities_real_minio.minio_repository, metadata['metadata'])
+    df = payload.retrieve(test_activities_real_minio.minio_repository, metadata['metadata'])
     assert len(df) >= 1
 
     client = minio_container.get_client()
@@ -82,31 +71,7 @@ async def test_predictions_batch_with_minio_offload_path(
         conn.execute(text(f'DELETE FROM predictions_schema.transformed_data WHERE model_id = {model_id}'))
     insert_sample_data(postgres_engine, model_id, [10.0, 20.0, 30.0])
 
-    input_data = {
-        'schedule_name': 'test-schedule',
-        'model_name': 'test_model',
-        'model_id': model_id,
-        'query': (
-            'SELECT timestamp, variable, value, created_at '
-            f'FROM predictions_schema.laborious_data WHERE model_id = {model_id}'
-        ),
-        'schema': 'predictions_schema',
-        'table_name': 'predictions',
-        'transform_table_name': 'transformed_data',
-        'input_filters': {'EMPTY_DATA': {'POLICY': 'STOP', 'CONFIG': {}}},
-        'mlflow_transform_filters': {'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}}},
-        'mlflow_predict_filters': {'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}}},
-        'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
-        'opc_output_config': {},
-        'pi_web_api_output_config': {},
-        'save_transform': True,
-        'prediction_store_policy': 'lts:1',
-        'model_config': {
-            'retention_minutes': 0,
-            'target': 'sensor_1',
-        },
-        'datetime_columns': ['timestamp', 'created_at'],
-    }
+    input_data = load_scenario_input('minio_offload_workflow.json', model_id=model_id)
 
     with patch.object(mdp, 'OFFLOAD_THRESHOLD_BYTES', 1):
         await start_and_await_workflow(
@@ -121,3 +86,23 @@ async def test_predictions_batch_with_minio_offload_path(
             text(f'SELECT COUNT(*) FROM predictions_schema.predictions WHERE model_id = {model_id}')
         ).scalar()
     assert count == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_load_query_with_inline_payload_when_below_threshold(
+    postgres_engine,
+    test_activities_real_minio: Activities,
+):
+    """Scenario 4.2.1: payload stays inline when threshold is high enough."""
+    model_id = 503
+    with postgres_engine.begin() as conn:
+        conn.execute(text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}'))
+    insert_sample_data(postgres_engine, model_id, [1.0, 2.0])
+
+    scenario_input = load_scenario_input('minio_offload_load_query.json', model_id=model_id)
+    with patch.object(mdp, 'OFFLOAD_THRESHOLD_BYTES', 10**9):
+        payload = test_activities_real_minio.load_query_with_minio_offload(scenario_input)
+
+    assert payload.object_key is None
+    assert payload.data is not None

e2e/test_predictions_batch_format_export.py

@@ -4,7 +4,7 @@ End-to-end tests for PredictionsBatch workflow - Format and Export scenarios.
 
 from decimal import Decimal
 from typing import Any, cast
-from unittest.mock import ANY, AsyncMock, call
+from unittest.mock import ANY, call
 
 import pytest
 from sientia_do.notifications.models import NotificationLevel
@@ -12,47 +12,18 @@ from sqlalchemy import text
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
 
-from e2e.helpers import assert_prediction, insert_sample_data, make_workflow_id, start_and_await_workflow
+from e2e.helpers import (
+    assert_prediction,
+    insert_sample_data,
+    load_scenario_input,
+    make_workflow_id,
+    start_and_await_workflow,
+)
 from laborious.activities.activities import Activities
 from laborious.workflows.predictions_batch import PredictionsBatch
 
-base_input_data = {
-    'schedule_name': 'test-schedule',
-    'model_name': 'test_model',
-    'model_id': 301,
-    'query': 'SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = 301',
-    'schema': 'predictions_schema',
-    'table_name': 'predictions',
-    'transform_table_name': 'transformed_data',
-    'input_filters': {
-        'EMPTY_DATA': {'POLICY': 'STOP', 'CONFIG': {}},
-    },
-    'mlflow_transform_filters': {
-        'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}},
-    },
-    'mlflow_predict_filters': {
-        'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}},
-    },
-    'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
-    'opc_output_config': {},
-    'pi_web_api_output_config': {},
-    'save_transform': True,
-    'prediction_store_policy': 'lts:1',
-    'model_config': {
-        'retention_minutes': 0,
-        'target': 'sensor_1',
-    },
-    'datetime_columns': ['timestamp', 'created_at'],
-}
-
-base_query = "SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = {model_id}"
-
 def get_base_input_data(model_id):
-    return {
-        **base_input_data,
-        'model_id': model_id,
-        'query': base_query.format(model_id=model_id),
-    }
+    return load_scenario_input('format_export_base.json', model_id=model_id)
 
 
 @pytest.mark.asyncio
@@ -668,8 +639,8 @@ async def test_scenario_3_2_3_pi_web_api_partial_write_error(
 
     insert_sample_data(postgres_engine, model_id, [23.5, 78.2])
 
-    test_activities.pi_web_api_client.write_value = AsyncMock(
-        side_effect=[
+    test_activities.pi_web_api_client.set_side_effect(
+        [
             # Prediction batch: two web_ids requested, only one acknowledged.
             [{'WebId': 'web_id_1', 'Errors': []}],
             # Confidence write succeeds.
@@ -707,4 +678,42 @@ async def test_scenario_3_2_3_pi_web_api_partial_write_error(
         prediction_confidence=13,
         comments="The number of written tags does not match the number of tag names: Expected ['tag_1', 'tag_3'] tags, but ['tag_1'] tags were written.",
     )
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_scenario_3_3_1_combined_pi_and_opc_outputs(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker: Worker,
+    test_activities: Activities,
+    postgres_engine,
+):
+    """
+    Scenario 3.3.1: PI and OPC enabled together.
+    """
+    client = temporal_test_env.client
+    model_id = 333
+    insert_sample_data(postgres_engine, model_id, [23.5, 78.2])
+
+    input_data = get_base_input_data(model_id)
+    input_data['pi_web_api_output_config'] = {
+        'endpoint': 'test_endpoint',
+        'prediction_tags': {'tag_1': 'web_id_1'},
+        'confidence_tags': {'tag_2': 'web_id_2'},
+    }
+    input_data['opc_output_config'] = {
+        '1': {
+            'prediction_tags': {'addr_1': {'data_type': 'float'}},
+            'confidence_tags': {'addr_2': {'data_type': 'float'}},
+        }
+    }
+
+    await start_and_await_workflow(
+        client, PredictionsBatch.run, input_data, make_workflow_id('test-pi-opc-combined')
+    )
+
+    assert test_activities.pi_web_api_client.write_value.call_count == 2
+    opc_write_data = cast(Any, test_activities.opc_repository['1'].write_data)
+    assert opc_write_data.call_count == 2
+    assert_prediction(postgres_engine, model_id)
     

e2e/test_predictions_batch_main_workflow.py

@@ -10,7 +10,7 @@ from temporalio.worker import Worker
 
 import pytest
 
-from e2e.helpers import make_workflow_id, start_and_await_workflow
+from e2e.helpers import load_scenario_input, make_workflow_id, start_and_await_workflow
 from laborious.activities.activities import Activities
 from laborious.workflows.predictions_batch import PredictionsBatch
 
@@ -37,42 +37,7 @@ async def test_scenario_1_1_1_happy_path_complete_success(
         """
         conn.execute(text(insert_sql))
 
-    input_data = {
-        'metadata': {
-            'metadata': {
-                'model_id': 123,
-                'model_name': 'test_model',
-                'schedule_name': 'test-schedule',
-                'workflow_name': 'predictions_batch',
-            }
-        },
-        'schedule_name': 'test-schedule',
-        'model_name': 'test_model',
-        'model_id': 123,
-        'query': 'SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = 123',
-        'schema': 'predictions_schema',
-        'table_name': 'predictions',
-        'transform_table_name': 'transformed_data',
-        'input_filters': {
-            'EMPTY_DATA': {'POLICY': 'STOP', 'CONFIG': {}},
-        },
-        'mlflow_transform_filters': {
-            'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}},
-        },
-        'mlflow_predict_filters': {
-            'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}},
-        },
-        'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
-        'opc_output_config': {},
-        'pi_web_api_output_config': {},
-        'save_transform': True,
-        'prediction_store_policy': 'lts:1',
-        'model_config': {
-            'target': 'sensor_1',
-            'retention_minutes': 0,
-        },
-        'datetime_columns': ['timestamp', 'created_at'],
-    }
+    input_data = load_scenario_input('main_happy_path.json', model_id=123)
 
     await start_and_await_workflow(
         client,
@@ -125,41 +90,7 @@ async def test_scenario_1_2_1_sql_query_execution_error(
     """Invalid SQL: workflow may complete with early exit; no prediction rows."""
     client = temporal_test_env.client
 
-    input_data = {
-        'metadata': {
-            'metadata': {
-                'model_id': 128,
-                'model_name': 'test_model',
-                'schedule_name': 'test-schedule',
-                'workflow_name': 'predictions_batch',
-            }
-        },
-        'schedule_name': 'test-schedule',
-        'model_name': 'test_model',
-        'model_id': 128,
-        'query': 'SELECT * FROM nonexistent_table WHERE invalid_syntax =',
-        'schema': 'predictions_schema',
-        'table_name': 'predictions',
-        'transform_table_name': 'transformed_data',
-        'input_filters': {
-            'EMPTY_DATA': {'POLICY': 'STOP', 'CONFIG': {}},
-        },
-        'mlflow_transform_filters': {
-            'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}},
-        },
-        'mlflow_predict_filters': {
-            'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}},
-        },
-        'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
-        'opc_output_config': {},
-        'pi_web_api_output_config': {},
-        'save_transform': True,
-        'prediction_store_policy': 'lts:1',
-        'model_config': {
-            'target': 'sensor_1',
-            'retention_minutes': 0,
-        },
-    }
+    input_data = load_scenario_input('main_sql_error.json', model_id=128)
 
     await start_and_await_workflow(
         client,
@@ -186,22 +117,7 @@ async def test_scenario_1_2_2_missing_required_parameters(
     """Missing query: workflow does not produce predictions and is terminated explicitly."""
     client = temporal_test_env.client
 
-    input_data = {
-        'metadata': {
-            'metadata': {
-                'model_id': 129,
-                'model_name': 'test_model',
-                'schedule_name': 'test-schedule',
-                'workflow_name': 'predictions_batch',
-            }
-        },
-        'schedule_name': 'test-schedule',
-        'model_name': 'test_model',
-        'model_id': 129,
-        'schema': 'predictions_schema',
-        'table_name': 'predictions',
-        'transform_table_name': 'transformed_data',
-    }
+    input_data = load_scenario_input('main_missing_required.json', model_id=129)
 
     handle = await client.start_workflow(
         PredictionsBatch.run,
@@ -244,42 +160,7 @@ async def test_scenario_1_2_3_invalid_datetime_column_specification(
             )
         )
 
-    input_data = {
-        'metadata': {
-            'metadata': {
-                'model_id': 130,
-                'model_name': 'test_model',
-                'schedule_name': 'test-schedule',
-                'workflow_name': 'predictions_batch',
-            }
-        },
-        'schedule_name': 'test-schedule',
-        'model_name': 'test_model',
-        'model_id': 130,
-        'query': 'SELECT timestamp, variable, value FROM predictions_schema.laborious_data WHERE model_id = 130',
-        'schema': 'predictions_schema',
-        'table_name': 'predictions',
-        'transform_table_name': 'transformed_data',
-        'input_filters': {
-            'EMPTY_DATA': {'POLICY': 'STOP', 'CONFIG': {}},
-        },
-        'mlflow_transform_filters': {
-            'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}},
-        },
-        'mlflow_predict_filters': {
-            'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}},
-        },
-        'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
-        'opc_output_config': {},
-        'pi_web_api_output_config': {},
-        'save_transform': True,
-        'prediction_store_policy': 'lts:1',
-        'model_config': {
-            'target': 'sensor_1',
-            'retention_minutes': 0,
-        },
-        'datetime_columns': ['nonexistent_column'],
-    }
+    input_data = load_scenario_input('main_invalid_datetime.json', model_id=130)
 
     handle = await client.start_workflow(
         PredictionsBatch.run,

e2e/test_predictions_batch_prediction_process.py

@@ -14,6 +14,7 @@ from temporalio.worker import Worker
 
 from e2e.helpers import (
     assert_continue,
+    load_scenario_input,
     assert_repeat,
     assert_stop,
     insert_sample_data,
@@ -23,47 +24,8 @@ from e2e.helpers import (
 from laborious.activities.activities import Activities
 from laborious.workflows.predictions_batch import PredictionsBatch
 
-base_input_data = {
-    'schedule_name': 'test-schedule',
-    'model_name': 'test_model',
-    'model_id': 201,
-    'query': 'SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = 201',
-    'schema': 'predictions_schema',
-    'table_name': 'predictions',
-    'transform_table_name': 'transformed_data',
-    'input_filters': {
-        'SPECIFIC_VARIABLES_NULL_VALUES': {
-            'POLICY': 'CONTINUE',
-            'CONFIG': {'variables': ['sensor_1']},
-        },
-    },
-    'mlflow_transform_filters': {
-        'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}},
-    },
-    'mlflow_predict_filters': {
-        'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}},
-    },
-    'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
-    'opc_output_config': {},
-    'pi_web_api_output_config': {},
-    'save_transform': True,
-    'prediction_store_policy': 'lts:1',
-    'model_config': {
-        'retention_minutes': 0,
-        'target': 'sensor_1',
-    },
-    'datetime_columns': ['timestamp', 'created_at'],
-}
-
-base_query = 'SELECT timestamp, variable, value, created_at FROM predictions_schema.laborious_data WHERE model_id = {model_id}'
-
-
 def get_base_input_data(model_id):
-    return {
-        **base_input_data,
-        'model_id': model_id,
-        'query': base_query.format(model_id=model_id),
-    }
+    return load_scenario_input('prediction_process_base.json', model_id=model_id)
 
 
 def insert_sample_prediction(postgres_engine, model_id):
@@ -383,3 +345,32 @@ async def test_scenario_2_4_1_input_empty_data_stop(
         client, PredictionsBatch.run, input_data, make_workflow_id('test-empty-data-stop')
     )
     assert_stop(postgres_engine, model_id)
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_scenario_2_4_1_priority_conflict_resolution(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    bad_data_model,
+):
+    """Conflicting filter outputs must honor configured path_priority order."""
+    client = temporal_test_env.client
+    model_id = 242
+    insert_sample_data(postgres_engine, model_id, [60.0, 78.2])
+    input_data = get_base_input_data(model_id)
+    input_data['mlflow_transform_filters'] = {'API_ERROR': {'POLICY': 'CONTINUE', 'CONFIG': {}}}
+    input_data['mlflow_predict_filters'] = {'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}}}
+    input_data['path_priority'] = ['STOP', 'CONTINUE', 'REPEAT']
+
+    await start_and_await_workflow(
+        client, PredictionsBatch.run, input_data, make_workflow_id('test-priority-conflict')
+    )
+    assert_continue(
+        postgres_engine=postgres_engine,
+        model_id=model_id,
+        prediction_confidence=Decimal(10),
+        comments='Unknown MLFlow API error',
+    )

e2e/test_simple_metrics.py

@@ -0,0 +1,322 @@
+"""
+End-to-end tests for the SimpleMetrics workflow.
+
+Coverage focus:
+
+- Happy path computes rmse/mse/mae/r2 from predictions joined against ``laborious_data``
+  and persists rows to ``predictions_schema.simple_metrics_data`` with all required columns.
+- Subset metric selection (only rmse) writes exactly the requested rows.
+- Zero-variance target produces ``r2=0`` per division-by-zero guard.
+- Empty join (no overlapping data) short-circuits without persisting anything.
+"""
+
+from datetime import datetime, timedelta, timezone
+from decimal import Decimal
+
+import math
+import pytest
+from sqlalchemy import text
+from temporalio.testing import WorkflowEnvironment
+from temporalio.worker import Worker
+
+from e2e.helpers import load_scenario_input, make_workflow_id, start_and_await_workflow
+from laborious.activities.activities import Activities
+from laborious.workflows.simple_metrics import SimpleMetrics
+
+EXPECTED_SIMPLE_METRICS_COLUMNS = [
+    'id',
+    'model_id',
+    'metric',
+    'value',
+    'timestamp',
+    'data_size',
+    'interval_minutes',
+    'created_at',
+]
+
+
+def _simple_metrics_input(model_id: int, **overrides) -> dict:
+    """Load and override the simple-metrics base scenario."""
+    payload = load_scenario_input('simple_metrics_base.json', model_id=model_id)
+    payload.update(overrides)
+    return payload
+
+
+def _seed_predictions_and_targets(
+    postgres_engine,
+    model_id: int,
+    pairs: list[tuple[float, float]],
+    target_name: str = 'sensor_target',
+    offset_minutes: int = 6,
+) -> list[str]:
+    """
+    Insert matching prediction/target rows used by the SimpleMetrics SQL JOIN.
+
+    For each ``(prediction, target)`` pair we write a row in ``predictions`` and
+    a matching row in ``laborious_data`` with ``variable=target_name`` so the
+    inner join in the workflow query yields one row per pair.
+
+    Args:
+     - postgres_engine: SQLAlchemy engine bound to the test container.
+     - model_id: Model id stamped on every row.
+     - pairs: ``(prediction, target)`` pairs, one per minute.
+     - target_name: Variable name in ``laborious_data`` representing the target.
+     - offset_minutes: Earliest row sits this many minutes ago so timestamps fall
+       inside the workflow's recent-data window.
+
+    Return:
+     List of timestamp strings written for the inserted rows.
+    """
+    base = datetime.now(timezone.utc).replace(second=0, microsecond=0) - timedelta(
+        minutes=offset_minutes
+    )
+    timestamps = [
+        (base + timedelta(minutes=i)).strftime('%Y-%m-%d %H:%M:%S%z')
+        for i in range(len(pairs))
+    ]
+
+    prediction_rows = []
+    target_rows = []
+    for index, (prediction, target_value) in enumerate(pairs):
+        ts = timestamps[index]
+        prediction_rows.append(
+            f"({model_id}, {prediction}, 0, 0, 'Good', '{ts}', '{ts}')"
+        )
+        target_rows.append(
+            f"({model_id}, '{target_name}', {target_value}, '{ts}', '{ts}')"
+        )
+
+    with postgres_engine.begin() as conn:
+        conn.execute(text(f'DELETE FROM predictions_schema.predictions WHERE model_id = {model_id}'))
+        conn.execute(text(f'DELETE FROM predictions_schema.laborious_data WHERE model_id = {model_id}'))
+        # The SimpleMetrics SQL JOIN only filters ``predictions.model_id``; it does
+        # NOT filter ``laborious_data.model_id`` (see ``e2e/CODE_ISSUES.md`` issue
+        # SM-1). Without this cross-model cleanup, a previous test's target rows
+        # under the same variable name would join into this test's predictions
+        # whenever timestamps happened to overlap.
+        conn.execute(
+            text(
+                "DELETE FROM predictions_schema.laborious_data "
+                "WHERE variable IN (:sensor_default, :target_name) "
+                "AND timestamp >= NOW() - INTERVAL '120 minutes'"
+            ),
+            {'sensor_default': 'sensor_target', 'target_name': target_name},
+        )
+        if prediction_rows:
+            conn.execute(
+                text(
+                    'INSERT INTO predictions_schema.predictions '
+                    '(model_id, prediction, prediction_confidence, response_time, '
+                    'prediction_status, "timestamp", created_at) VALUES '
+                    + ', '.join(prediction_rows)
+                )
+            )
+            conn.execute(
+                text(
+                    'INSERT INTO predictions_schema.laborious_data '
+                    '(model_id, variable, value, "timestamp", created_at) VALUES '
+                    + ', '.join(target_rows)
+                )
+            )
+    return timestamps
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_simple_metrics_happy_path_persists_all_metrics_and_columns(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_simple_metrics: Worker,
+    test_activities: Activities,
+    postgres_engine,
+):
+    """
+    Scenario S.1.1: rmse/mse/mae/r2 are calculated from a deterministic
+    prediction/target pair set and written one row per metric. Every column
+    expected by ``predictions_schema.simple_metrics_data`` must be populated and
+    the numerical values must match closed-form expectations.
+    """
+    client = temporal_test_env.client
+    model_id = 511
+
+    pairs = [
+        (1.0, 2.0),
+        (2.0, 4.0),
+        (3.0, 5.0),
+        (4.0, 9.0),
+        (5.0, 12.0),
+    ]
+    diffs = [target - prediction for prediction, target in pairs]
+    n = len(diffs)
+    expected_rmse = math.sqrt(sum(d * d for d in diffs) / n)
+    expected_mse = sum(d * d for d in diffs) / n
+    expected_mae = sum(abs(d) for d in diffs) / n
+    target_mean = sum(t for _, t in pairs) / n
+    ss_res = sum((target - prediction) ** 2 for prediction, target in pairs)
+    ss_tot = sum((t - target_mean) ** 2 for _, t in pairs)
+    expected_r2 = 1.0 - (ss_res / ss_tot)
+
+    _seed_predictions_and_targets(postgres_engine, model_id=model_id, pairs=pairs)
+
+    input_data = _simple_metrics_input(model_id)
+    await start_and_await_workflow(
+        client,
+        SimpleMetrics.run,
+        input_data,
+        make_workflow_id('test-simple-metrics-happy'),
+    )
+
+    with postgres_engine.connect() as conn:
+        rows = (
+            conn.execute(
+                text(
+                    'SELECT * FROM predictions_schema.simple_metrics_data '
+                    'WHERE model_id = :m ORDER BY metric'
+                ),
+                {'m': model_id},
+            )
+            .mappings()
+            .all()
+        )
+
+    assert len(rows) == 4, f'Expected 4 metric rows, got {len(rows)}'
+    for column in EXPECTED_SIMPLE_METRICS_COLUMNS:
+        assert column in rows[0], f'Missing simple_metrics column: {column}'
+    for row in rows:
+        for column in EXPECTED_SIMPLE_METRICS_COLUMNS:
+            assert row[column] is not None, f"Column '{column}' is NULL in {dict(row)}"
+
+    by_metric = {row['metric']: row for row in rows}
+    assert set(by_metric) == {'rmse', 'mse', 'mae', 'r2'}
+
+    def _decimal_close(actual, expected, places: int = 6) -> bool:
+        return abs(float(actual) - expected) < 10 ** (-places)
+
+    assert _decimal_close(by_metric['rmse']['value'], expected_rmse)
+    assert _decimal_close(by_metric['mse']['value'], expected_mse)
+    assert _decimal_close(by_metric['mae']['value'], expected_mae)
+    assert _decimal_close(by_metric['r2']['value'], expected_r2)
+
+    assert all(row['data_size'] == n for row in rows), 'data_size must equal target row count'
+    assert all(row['interval_minutes'] == 60 for row in rows)
+    assert all(row['model_id'] == model_id for row in rows)
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_simple_metrics_subset_metrics_writes_only_requested_rows(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_simple_metrics: Worker,
+    test_activities: Activities,
+    postgres_engine,
+):
+    """
+    Scenario S.1.2: Requesting ``metrics=['rmse']`` must persist exactly one row
+    with metric ``rmse`` and skip mse/mae/r2.
+    """
+    client = temporal_test_env.client
+    model_id = 512
+
+    pairs = [(1.0, 2.0), (2.0, 4.0), (3.0, 6.0)]
+    _seed_predictions_and_targets(postgres_engine, model_id=model_id, pairs=pairs)
+
+    input_data = _simple_metrics_input(model_id, metrics=['rmse'])
+    await start_and_await_workflow(
+        client,
+        SimpleMetrics.run,
+        input_data,
+        make_workflow_id('test-simple-metrics-subset'),
+    )
+
+    with postgres_engine.connect() as conn:
+        metrics = [
+            r[0]
+            for r in conn.execute(
+                text(
+                    'SELECT metric FROM predictions_schema.simple_metrics_data '
+                    'WHERE model_id = :m'
+                ),
+                {'m': model_id},
+            ).all()
+        ]
+    assert metrics == ['rmse']
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_simple_metrics_zero_variance_target_returns_zero_r2(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_simple_metrics: Worker,
+    test_activities: Activities,
+    postgres_engine,
+):
+    """
+    Scenario S.2.1: When the target column has zero variance the activity must
+    return ``r2 = 0`` (division-by-zero guard) and still persist all four metrics.
+    """
+    client = temporal_test_env.client
+    model_id = 521
+
+    pairs = [(0.0, 5.0), (1.0, 5.0), (2.0, 5.0), (3.0, 5.0)]
+    _seed_predictions_and_targets(postgres_engine, model_id=model_id, pairs=pairs)
+
+    input_data = _simple_metrics_input(model_id)
+    await start_and_await_workflow(
+        client,
+        SimpleMetrics.run,
+        input_data,
+        make_workflow_id('test-simple-metrics-zero-variance'),
+    )
+
+    with postgres_engine.connect() as conn:
+        r2_value = conn.execute(
+            text(
+                "SELECT value FROM predictions_schema.simple_metrics_data "
+                "WHERE model_id = :m AND metric = 'r2'"
+            ),
+            {'m': model_id},
+        ).scalar()
+    assert r2_value is not None
+    assert Decimal(str(r2_value)) == Decimal('0'), f'expected r2=0, got {r2_value!r}'
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_simple_metrics_no_overlapping_data_short_circuits(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_simple_metrics: Worker,
+    test_activities: Activities,
+    postgres_engine,
+):
+    """
+    Scenario S.3.1: When the join produces no rows (no matching laborious_data
+    row for the configured ``target``), the workflow returns early without
+    invoking ``calculate_simple_metrics`` and writes nothing.
+    """
+    client = temporal_test_env.client
+    model_id = 531
+
+    # Insert predictions but no matching target rows for the configured variable.
+    _seed_predictions_and_targets(
+        postgres_engine,
+        model_id=model_id,
+        pairs=[(1.0, 1.0)],
+        target_name='wrong_variable_name',
+    )
+
+    input_data = _simple_metrics_input(model_id)
+    await start_and_await_workflow(
+        client,
+        SimpleMetrics.run,
+        input_data,
+        make_workflow_id('test-simple-metrics-empty-join'),
+    )
+
+    with postgres_engine.connect() as conn:
+        count = conn.execute(
+            text(
+                'SELECT COUNT(*) FROM predictions_schema.simple_metrics_data '
+                'WHERE model_id = :m'
+            ),
+            {'m': model_id},
+        ).scalar()
+    assert count == 0, 'Empty target data must short-circuit and skip persistence'