Merge release/SIENTIAPDE-1646 into feature/SIENTIAPDE-1986-simple-metrics-rewrite

Requirements follow the release; only sientia_model>=0.12.0 (RegressionMetrics),
model-library@0.13.1 and scikit-learn==1.8.0 (local) are kept from this branch.

tests/laborious/activities/test_activities.py

@@ -6,6 +6,7 @@ from laborious.activities.gates import Gates
 from laborious.activities.mlflow import MLFlow
 from laborious.activities.model_metrics import ModelMetrics
 from laborious.activities.opc import OPC
+from laborious.activities.prediction_metadata import PredictionMetadata
 from laborious.activities.storage import Storage
 
 
@@ -13,6 +14,7 @@ from laborious.activities.storage import Storage
 @patch('laborious.activities.activities.MLFlow.__init__')
 @patch('laborious.activities.activities.OPC.__init__')
 @patch('laborious.activities.activities.Gates.__init__')
+@patch('laborious.activities.activities.PredictionMetadata.__init__')
 @patch('laborious.activities.activities.ModelMetrics.__init__')
 @patch('laborious.activities.activities.API.__init__')
 @patch('laborious.activities.activities.MinioRepository')
@@ -22,6 +24,7 @@ def test___init__(
     mock_minio_repository,
     mock_api_init,
     mock_model_metrics_init,
+    mock_prediction_metadata_init,
     mock_gates_init,
     mock_opc_init,
     mock_mlflow_init,
@@ -47,6 +50,7 @@ def test___init__(
     }
 
     mlflow_repository = MagicMock()
+    mongo_repository = MagicMock()
     plugin_store = MagicMock()
 
     opc_config = {
@@ -73,6 +77,7 @@ def test___init__(
         logger=logger,
         notification_handler=notification_handler,
         mlflow_repository=mlflow_repository,
+        mongo_repository=mongo_repository,
     )
 
     assert isinstance(activities, Activities)
@@ -82,6 +87,7 @@ def test___init__(
     assert isinstance(activities, Gates)
     assert isinstance(activities, ModelMetrics)
     assert isinstance(activities, API)
+    assert isinstance(activities, PredictionMetadata)
 
     mock_storage_init.assert_called_once_with(
         ANY,
@@ -125,6 +131,14 @@ def test___init__(
         metrics_controller=mock_metrics_controller.return_value,
     )
 
+    mock_prediction_metadata_init.assert_called_once_with(
+        ANY,
+        mongo_repository=mongo_repository,
+        logger=logger,
+        notification_handler=notification_handler,
+        metrics_controller=mock_metrics_controller.return_value,
+    )
+
     mock_model_metrics_init.assert_called_once_with(
         ANY,
         logger=logger,
@@ -158,6 +172,7 @@ def test___init__(
 @patch('laborious.activities.activities.MLFlow')
 @patch('laborious.activities.activities.OPC')
 @patch('laborious.activities.activities.Gates')
+@patch('laborious.activities.activities.PredictionMetadata')
 @patch('laborious.activities.activities.ModelMetrics')
 @patch('laborious.activities.activities.API')
 @patch('laborious.activities.activities.MinioRepository')
@@ -165,6 +180,7 @@ def test_shutdown(
     _mock_minio_repository,
     mock_api_init,
     mock_model_metrics_init,
+    mock_prediction_metadata_init,
     mock_gates_init,
     mock_opc_init,
     mock_mlflow_init,
@@ -217,6 +233,7 @@ def test_shutdown(
         logger=logger,
         notification_handler=notification_handler,
         mlflow_repository=mlflow_repository,
+        mongo_repository=MagicMock(),
     )
 
     activities.shutdown()
@@ -226,6 +243,7 @@ def test_shutdown(
     mock_gates_init.close.assert_called_once()
     mock_model_metrics_init.close.assert_called_once()
     mock_api_init.close.assert_called_once()
+    mock_prediction_metadata_init.close.assert_called_once()
 
 
 @patch('laborious.activities.activities.SientiaMLflowRepository')
@@ -234,6 +252,7 @@ def test_shutdown(
 @patch('laborious.activities.activities.MLFlow.__init__')
 @patch('laborious.activities.activities.OPC.__init__')
 @patch('laborious.activities.activities.Gates.__init__')
+@patch('laborious.activities.activities.PredictionMetadata.__init__')
 @patch('laborious.activities.activities.ModelMetrics.__init__')
 @patch('laborious.activities.activities.API.__init__')
 @patch('laborious.activities.activities.MinioRepository')
@@ -243,6 +262,7 @@ def test___init___builds_mlflow_repository_when_not_provided(
     mock_minio_repository,
     _mock_api_init,
     _mock_model_metrics_init,
+    _mock_prediction_metadata_init,
     _mock_gates_init,
     _mock_opc_init,
     _mock_mlflow_init,
@@ -286,6 +306,7 @@ def test___init___builds_mlflow_repository_when_not_provided(
         pi_web_api_config=pi_web_api_config,
         logger=logger,
         notification_handler=notification_handler,
+        mongo_repository=MagicMock(),
     )
 
     mock_build_mlflow_config.assert_called_once()
@@ -297,3 +318,92 @@ def test___init___builds_mlflow_repository_when_not_provided(
         notification_handler=notification_handler,
         metrics_controller=mock_metrics_controller.return_value,
     )
+
+
+@patch('laborious.activities.activities.MongoDBRepository')
+@patch('laborious.activities.activities.build_mongodb_config')
+@patch('laborious.activities.activities.Storage.__init__')
+@patch('laborious.activities.activities.MLFlow.__init__')
+@patch('laborious.activities.activities.OPC.__init__')
+@patch('laborious.activities.activities.Gates.__init__')
+@patch('laborious.activities.activities.PredictionMetadata.__init__')
+@patch('laborious.activities.activities.ModelMetrics.__init__')
+@patch('laborious.activities.activities.API.__init__')
+@patch('laborious.activities.activities.MinioRepository')
+@patch('laborious.activities.activities.MetricsController')
+def test___init___builds_mongo_repository_when_not_provided(
+    mock_metrics_controller,
+    mock_minio_repository,
+    _mock_api_init,
+    _mock_model_metrics_init,
+    mock_prediction_metadata_init,
+    mock_gates_init,
+    _mock_opc_init,
+    _mock_mlflow_init,
+    _mock_storage_init,
+    mock_build_mongodb_config,
+    mock_mongo_repository_cls,
+):
+    postgres_config = {
+        'host': 'localhost',
+        'port': 5432,
+        'user': 'postgres',
+        'password': 'postgres',
+        'dbname': 'postgres',
+        'min_connections': 1,
+        'max_connections': 10,
+    }
+    minio_config = {
+        'endpoint_url': 'localhost:9000',
+        'access_key': 'minio',
+        'secret_key': 'minio123',
+        'default_bucket': 'test',
+        'retention_hours': 24,
+        'secure': False,
+    }
+    opc_config = {'bootstrap_servers': 'localhost:9092', 'polling_time': 1000, 'group_id': 'test'}
+    pi_web_api_config = {'base_url': 'https://pi', 'auth_type': 'bearer', 'auth_token': 'token'}
+    logger = MagicMock()
+    notification_handler = MagicMock()
+    plugin_store = MagicMock()
+    mlflow_repository = MagicMock()
+    mock_build_mongodb_config.return_value = {
+        'connection_string': 'mongodb://root:pass@localhost:27018',
+        'database_name': 'sientia',
+    }
+
+    Activities(
+        postgres_config=postgres_config,
+        plugin_store=plugin_store,
+        minio_config=minio_config,
+        opc_config=opc_config,
+        pi_web_api_config=pi_web_api_config,
+        logger=logger,
+        notification_handler=notification_handler,
+        mlflow_repository=mlflow_repository,
+    )
+
+    mock_build_mongodb_config.assert_called_once()
+    mock_mongo_repository_cls.assert_called_once_with(
+        connection_string='mongodb://root:pass@localhost:27018',
+        database_name='sientia',
+        logger=logger,
+        notification_handler=notification_handler,
+        metrics_controller=mock_metrics_controller.return_value,
+    )
+    mock_gates_init.assert_called_once_with(
+        ANY,
+        minio_repository=mock_minio_repository.return_value,
+        logger=logger,
+        notification_handler=notification_handler,
+        metrics_controller=mock_metrics_controller.return_value,
+    )
+    # The same repository built here is the one handed to PredictionMetadata, not a second
+    # instance.
+    mock_prediction_metadata_init.assert_called_once_with(
+        ANY,
+        mongo_repository=mock_mongo_repository_cls.return_value,
+        logger=logger,
+        notification_handler=notification_handler,
+        metrics_controller=mock_metrics_controller.return_value,
+    )

tests/laborious/activities/test_api.py

@@ -2,6 +2,7 @@ from unittest.mock import ANY, MagicMock, call, patch
 
 from pytest import fixture
 from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.metrics_controller import MetricsController
 
 from laborious.activities.api import API, PI_WEB_API_PREDICTION_ERROR_CONFIDENCE
 
@@ -70,7 +71,7 @@ def test_get_pi_web_api_core_labels_without_operation_type(mock_pi_web_api_clien
         auth_token='test_token',
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     with patch.object(
         SientiaMonitoring,
@@ -104,7 +105,7 @@ def test_get_pi_web_api_core_labels_with_operation_type(mock_pi_web_api_client):
         auth_token='test_token',
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     with patch.object(
         SientiaMonitoring,
@@ -131,7 +132,7 @@ def test__init__():
         auth_token='test_token',
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
 
     assert api.pi_web_api_client is not None
@@ -152,7 +153,7 @@ def api(mock_pi_web_api_client):
         auth_token='test_token',
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     api_instance.send_notification = MagicMock()
     api_instance.info = MagicMock()
@@ -218,6 +219,43 @@ def test_write_pi_web_api_data_success(mock_dataframe, api, base_input_data):
     }
 
 
+@patch('laborious.activities.api.DataFrame')
+def test_write_pi_web_api_data_partial_prediction_mismatch_downgrades_confidence(
+    mock_dataframe, api, base_input_data
+):
+    """A partial write (some tags report Errors) doesn't raise — it's caught by the tag-count
+    mismatch check, which still must downgrade confidence and attach the mismatch message."""
+    input_data = {
+        **base_input_data,
+        'pi_web_api_output_config': {
+            'endpoint': 'https://test-pi-server.com/piwebapi',
+            'prediction_tags': {'tag1': 'web_id_1', 'tag2': 'web_id_2'},
+            'confidence_tags': {'tag3': 'web_id_3'},
+        },
+    }
+    mock_df = _create_mock_dataframe()
+    mock_dataframe.return_value = mock_df
+
+    api.pi_web_api_client.write_value.side_effect = [
+        [{'WebId': 'web_id_1', 'Errors': []}, {'WebId': 'web_id_2', 'Errors': ['bad tag']}],
+        [{'WebId': 'web_id_3', 'Errors': []}],
+    ]
+
+    api.write_pi_web_api_data(input_data)
+
+    mock_df.__setitem__.assert_any_call(
+        'prediction_confidence', PI_WEB_API_PREDICTION_ERROR_CONFIDENCE
+    )
+    mock_df.__setitem__.assert_any_call('comments', ANY)
+    api.send_notification.assert_called_once_with(
+        metadata=metadata['metadata'],
+        notification_id='WRITE_PI_WEB_API_PREDICTION_ERROR',
+        message=ANY,
+        block='write_pi_web_api_data',
+        level=NotificationLevel.ERROR,
+    )
+
+
 @patch('laborious.activities.api.DataFrame')
 def test_write_pi_web_api_data_prediction_error(mock_dataframe, api, base_input_data):
     mock_dataframe.return_value = _create_mock_dataframe(
@@ -348,7 +386,8 @@ def test_close(mock_shutdown, api):
     api.close()
 
     api.pi_web_api_client.close.assert_called_once()
-    mock_shutdown.assert_called_once_with(api)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(api)) == 1
 
 
 def test_process_pi_web_api_response_success(api):

tests/laborious/activities/test_gates.py

@@ -3,9 +3,10 @@ from unittest.mock import ANY, MagicMock, call, patch
 from pandas import DataFrame
 from pytest import fixture
 from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.metrics_controller import MetricsController
 from sientia_do.observability.sientia_monitoring import SientiaMonitoring
 
-from laborious.activities.gates import Gates
+from laborious.activities.gates import Gates, input_filter_counters, input_filter_functions
 
 
 @fixture(autouse=True)
@@ -38,7 +39,7 @@ def gates_activity():
     gates = Gates(
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     gates.error = MagicMock()
     gates.debug = MagicMock()
@@ -60,6 +61,13 @@ metadata = {
 }
 
 
+def test_input_filter_counters_covers_every_input_filter_function():
+    # A filter added to input_filter_functions without a matching entry here would degrade to
+    # counter=0 (see the `.get(..., lambda *_: 0)` fallback in input_gate) instead of crashing —
+    # but this test should still catch the gap before it ships (review finding, 2026-09-30).
+    assert set(input_filter_counters) == set(input_filter_functions)
+
+
 def test_input_gate_invalid_filter(gates_activity):
     # Arrange
     input_data = {
@@ -73,7 +81,7 @@ def test_input_gate_invalid_filter(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.error.assert_called_once_with(
         'Filter INVALID_FILTER not found', metadata['metadata']
     )
@@ -97,7 +105,7 @@ def test_input_gate_filter_exception(mock_input_filter_functions, gates_activity
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.send_notification.assert_called_once_with(
         metadata=metadata['metadata'],
         notification_id='INTPUT_GATE_ERROR__EMPTY_DATA',
@@ -121,7 +129,7 @@ def test_input_gate_no_filters(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.debug.assert_called()
 
 
@@ -135,10 +143,14 @@ def test_input_gate_with_filter(gates_activity):
     }
 
     # Act
-    result = gates_activity.input_gate(input_data)
+    path_flag, confidence, comment, document = gates_activity.input_gate(input_data)
 
-    # Assert
-    assert result == ('STOP', -1, 'Input data with bad quality')
+    # Assert: the gate only computes the document now — it never writes anywhere, and never
+    # checks path_flag to decide whether to build one (that decision moved to the caller, see
+    # `PredictionMetadata.write_prediction_metadata` and `prediction_process.py`), so even a STOP
+    # resolution still gets its document built.
+    assert (path_flag, confidence, comment) == ('STOP', -1, 'Input data with bad quality')
+    assert document == {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}
     gates_activity.debug.assert_called()
 
 
@@ -155,7 +167,12 @@ def test_input_gate_with_filter_lowercase_keys(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == ('STOP', -1, 'Input data with bad quality')
+    assert result == (
+        'STOP',
+        -1,
+        'Input data with bad quality',
+        {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]},
+    )
 
 
 def test_input_gate_with_filter_capitalized_keys(gates_activity):
@@ -171,7 +188,12 @@ def test_input_gate_with_filter_capitalized_keys(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == ('STOP', -1, 'Input data with bad quality')
+    assert result == (
+        'STOP',
+        -1,
+        'Input data with bad quality',
+        {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]},
+    )
 
 
 def test_input_gate_with_filter_not_caught(gates_activity):
@@ -187,10 +209,94 @@ def test_input_gate_with_filter_not_caught(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.debug.assert_called()
 
 
+def test_input_gate_repeat_also_builds_document(gates_activity):
+    # Arrange: REPEAT used to be treated like STOP (no document at all) — it isn't anymore, since
+    # REPEAT does insert a new `predictions` row (reusing the previous row's values), so it needs
+    # a document/metadata_id just like CONTINUE (2026-10-01 redesign).
+    input_data = {
+        **metadata,
+        'filters': {'EMPTY_DATA': {'POLICY': 'REPEAT', 'CONFIG': {}}},
+        'data': _minio_payload(DataFrame({'value': []})),
+        'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
+    }
+
+    # Act
+    result = gates_activity.input_gate(input_data)
+
+    # Assert
+    assert result == (
+        'REPEAT',
+        -1,
+        'Input data with bad quality',
+        {'filters': [{'name': 'EMPTY_DATA', 'policy': 'REPEAT', 'counter': 0}]},
+    )
+
+
+def test_input_gate_continue_builds_document_with_counter(gates_activity):
+    # Arrange: two rows with a null in one of the configured variables ('x'); a third row is
+    # clean, to confirm the counter reflects only the rows that actually matched the condition.
+    input_data = {
+        **metadata,
+        'filters': {
+            'SPECIFIC_VARIABLES_NULL_VALUES': {
+                'POLICY': 'CONTINUE',
+                'CONFIG': {'variables': ['x']},
+            }
+        },
+        'data': _minio_payload(
+            DataFrame({'variable': ['x', 'x', 'y'], 'value': [None, None, None]})
+        ),
+        'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
+    }
+
+    # Act
+    path_flag, confidence, comment, document = gates_activity.input_gate(input_data)
+
+    # Assert
+    assert (path_flag, confidence, comment) == ('CONTINUE', 2, 'Input data with bad quality')
+    assert document == {
+        'filters': [{'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2}]
+    }
+
+
+def test_input_gate_multiple_filters_same_policy_lists_each(gates_activity):
+    # Arrange: force both input filters to trigger (not reachable with today's real filter
+    # conditions — EMPTY_DATA and SPECIFIC_VARIABLES_NULL_VALUES are mutually exclusive) to lock
+    # in the list-building behavior approved by Vitor (2026-09-30): one entry per filter sharing
+    # the winning policy, instead of a single joined/aggregated value.
+    with patch('laborious.activities.gates.input_filter_functions') as mock_filters:
+        mock_filters.__contains__.return_value = True
+        mock_filters.__getitem__.return_value = MagicMock(return_value=True)
+
+        input_data = {
+            **metadata,
+            'filters': {
+                'SPECIFIC_VARIABLES_NULL_VALUES': {
+                    'POLICY': 'CONTINUE',
+                    'CONFIG': {'variables': ['x']},
+                },
+                'EMPTY_DATA': {'POLICY': 'CONTINUE', 'CONFIG': {}},
+            },
+            'data': _minio_payload(DataFrame({'variable': ['x'], 'value': [None]})),
+            'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
+        }
+
+        # Act
+        path_flag, _, _, document = gates_activity.input_gate(input_data)
+
+    # Assert
+    assert path_flag == 'CONTINUE'
+    assert {f['name'] for f in document['filters']} == {
+        'SPECIFIC_VARIABLES_NULL_VALUES',
+        'EMPTY_DATA',
+    }
+    assert all(f['policy'] == 'CONTINUE' for f in document['filters'])
+
+
 def test_mlflow_response_gate_invalid_filter(gates_activity):
     # Arrange
     input_data = {
@@ -985,12 +1091,13 @@ def test_close_disposes_minio_repository(mock_shutdown):
         minio_repository=minio,
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     gates.close()
     minio.close.assert_called_once()
     assert gates.minio_repository is None
-    mock_shutdown.assert_called_once_with(gates)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(gates)) == 1
 
 
 @patch.object(SientiaMonitoring, 'shutdown')
@@ -1000,7 +1107,8 @@ def test_close_without_minio_repository(mock_shutdown):
         minio_repository=None,
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     gates.close()
-    mock_shutdown.assert_called_once_with(gates)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(gates)) == 1

tests/laborious/activities/test_mlflow.py

@@ -5,6 +5,7 @@ import numpy as np
 import pandas as pd
 from pytest import fixture, raises
 from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.metrics_controller import MetricsController
 from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
 
 from laborious.activities.mlflow import MLFlow
@@ -22,7 +23,7 @@ def _passthrough_from_dict():
 def test___init__(mock_minio_repository):
     logger = MagicMock()
     notification_handler = MagicMock()
-    metrics_controller = MagicMock()
+    metrics_controller = MagicMock(spec=MetricsController)
     mlflow_repo = MagicMock()
     plugin_store = MagicMock()
 
@@ -63,7 +64,7 @@ def test___init__(mock_minio_repository):
 def mlflow(mock_minio_repository):
     logger = MagicMock()
     notification_handler = MagicMock()
-    metrics_controller = MagicMock()
+    metrics_controller = MagicMock(spec=MetricsController)
     mlflow_repo = MagicMock()
     plugin_store = MagicMock()
 
@@ -374,9 +375,9 @@ def test_request_predict_failure(mock_to_datetime, mock_from_dataframe, mlflow):
 @patch('laborious.activities.mlflow.rmtree')
 @patch('laborious.activities.mlflow.to_datetime')
 def test_retrain_model_success_data_success_retrain(
-    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow
+    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow, tmp_path
 ):
-    mock_mkdtemp.return_value = 'tmp'
+    mock_mkdtemp.return_value = str(tmp_path)
     mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx)
 
     mv_alias = MagicMock()
@@ -427,9 +428,9 @@ def test_retrain_model_success_data_success_retrain(
 @patch('laborious.activities.mlflow.rmtree')
 @patch('laborious.activities.mlflow.to_datetime')
 def test_retrain_model_success_with_payload_data(
-    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow
+    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow, tmp_path
 ):
-    mock_mkdtemp.return_value = 'tmp'
+    mock_mkdtemp.return_value = str(tmp_path)
     mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx)
     mv_alias = MagicMock(run_id='src')
     mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv_alias
@@ -472,9 +473,9 @@ def test_retrain_model_success_with_payload_data(
 @patch('laborious.activities.mlflow.rmtree')
 @patch('laborious.activities.mlflow.to_datetime')
 def test_retrain_model_always_uses_retrain_even_with_full_retrain_flag(
-    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow
+    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow, tmp_path
 ):
-    mock_mkdtemp.return_value = 'tmp'
+    mock_mkdtemp.return_value = str(tmp_path)
     mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx)
     mv_alias = MagicMock(run_id='src')
     mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv_alias
@@ -854,3 +855,22 @@ def test_get_reference_data_exception(mlflow):
     result = mlflow.get_reference_data(input_data)
 
     assert result is None
+
+
+def test_find_downloaded_csv_prefers_the_direct_path(mlflow, tmp_path):
+    direct = tmp_path / 'evaluation_data.csv'
+    direct.write_text('a\n1\n')
+
+    assert mlflow._find_downloaded_csv(str(tmp_path), 'evaluation_data.csv') == direct
+
+
+def test_find_downloaded_csv_searches_nested_directories(mlflow, tmp_path):
+    nested = tmp_path / 'artifacts' / 'data' / 'evaluation_data.csv'
+    nested.parent.mkdir(parents=True)
+    nested.write_text('a\n1\n')
+
+    assert mlflow._find_downloaded_csv(str(tmp_path), 'evaluation_data.csv') == nested
+
+
+def test_find_downloaded_csv_returns_none_when_missing(mlflow, tmp_path):
+    assert mlflow._find_downloaded_csv(str(tmp_path), 'evaluation_data.csv') is None

tests/laborious/activities/test_model_metrics.py

@@ -1,8 +1,9 @@
 from unittest.mock import ANY, MagicMock, patch
 
-from pandas import DataFrame, Timestamp
-from pytest import fixture, raises
+from pandas import DataFrame, Series, Timestamp
+from pytest import fixture, mark, raises
 from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.metrics_controller import MetricsController
 from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
 from sientia_model.analytics.drift_analysis import DriftInsufficientDataError
 
@@ -14,7 +15,7 @@ def model_metrics_activity():
     model_metrics = ModelMetrics(
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     model_metrics.error = MagicMock()
     model_metrics.debug = MagicMock()
@@ -1169,3 +1170,50 @@ def test_calculate_simple_metrics_no_thresholds_no_alert(model_metrics_activity)
 
     assert len(result) == 1
     model_metrics_activity.send_notification.assert_not_called()
+
+
+@mark.parametrize(
+    'failing_stage',
+    ['detect_univariate_drift', 'detect_multivariate_drift', 'get_drift_metrics_dataframe'],
+)
+@patch('laborious.activities.model_metrics.DriftAnalysis')
+def test_get_drift_metrics_reraises_insufficient_data_untouched(
+    mock_drift_analysis, failing_stage, model_metrics_activity
+):
+    """Insufficient data is not a stage failure: it propagates without the stage error report."""
+    error = DriftInsufficientDataError('not enough rows', analysis_rows=1)
+    getattr(mock_drift_analysis.return_value, failing_stage).side_effect = error
+    data = DataFrame({'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0]})
+    model_metrics_activity._drift_analyze_stage_error = MagicMock()
+
+    with raises(DriftInsufficientDataError) as exc_info:
+        model_metrics_activity.get_drift_metrics(
+            reference_data=data,
+            target_data=data,
+            target_name='target',
+            reference_columns=data.drop(columns=['target', 'timestamp']).columns,
+            drift_metrics=['ks_test'],
+            chunk_period='min',
+            metadata=metadata['metadata'],
+        )
+
+    assert exc_info.value is error
+    model_metrics_activity._drift_analyze_stage_error.assert_not_called()
+
+
+def test_to_naive_utc_converts_aware_timestamps_to_naive_utc():
+    series = Series(['2024-01-01T03:00:00-03:00', '2024-01-01T09:00:00-03:00'])
+
+    result = ModelMetrics._to_naive_utc(series)
+
+    assert result.dt.tz is None
+    assert result.tolist() == [Timestamp('2024-01-01 06:00:00'), Timestamp('2024-01-01 12:00:00')]
+
+
+def test_to_naive_utc_keeps_naive_timestamps():
+    series = Series(['2024-01-01 06:00:00'])
+
+    result = ModelMetrics._to_naive_utc(series)
+
+    assert result.dt.tz is None
+    assert result.tolist() == [Timestamp('2024-01-01 06:00:00')]

tests/laborious/activities/test_opc.py

@@ -4,6 +4,7 @@ import pytest
 from pandas import DataFrame
 from pytest import mark
 from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.metrics_controller import MetricsController
 
 from laborious.activities.opc import (
     OPC,
@@ -32,7 +33,7 @@ def test__init__():
         opc_servers=servers,
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
 
     assert opc.opc_servers == servers
@@ -43,7 +44,7 @@ def test__init__():
 @patch('laborious.activities.opc.OPC.send_notification')
 def test_init_opc(mock_send_notification, mock_opc_repository):
     mock_logger = MagicMock()
-    mock_metrics_controller = MagicMock()
+    mock_metrics_controller = MagicMock(spec=MetricsController)
     server1 = MagicMock(
         connect=MagicMock(return_value=(True, {})), write_data=MagicMock(return_value=(True, {}))
     )
@@ -192,7 +193,7 @@ def opc(mock_opc_repository):
         opc_servers=servers,
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     opc.init_opc()
     opc.send_notification = MagicMock()
@@ -539,6 +540,38 @@ def test_manage_output_tags_do_nothing(opc):
     opc._write_tags_from_config.assert_not_called()
 
 
+@patch('laborious.activities.opc.DataFrame')
+def test_write_opc_data_propagates_session_bad_and_reconnect_in_progress(mock_dataframe, opc):
+    """A single server reporting session_bad/reconnect_in_progress must flip the aggregated
+    flags passed to process_confidence, even though nothing raised."""
+    input_data = {
+        **metadata,
+        'data': {'prediction': [0.75], 'prediction_confidence': [0.95]},
+        'opc_output_config': {
+            'server1': {
+                'prediction_tags': {'tag1': {'data_type': 'float'}},
+                'confidence_tags': {'tag2': {'data_type': 'float'}},
+            }
+        },
+    }
+
+    opc.manage_output_tags = MagicMock(
+        return_value=(False, {'tag1': 0.1, 'tag2': 0.2}, True, 'BadSessionClosed', True)
+    )
+    opc.process_confidence = MagicMock(return_value={'data': 'data'})
+
+    opc.write_opc_data(input_data)
+
+    opc.process_confidence.assert_called_once_with(
+        mock_dataframe.return_value,
+        False,
+        metadata['metadata'],
+        session_bad=True,
+        opc_status='BadSessionClosed',
+        reconnect_in_progress=True,
+    )
+
+
 @patch('laborious.activities.opc.DataFrame')
 def test_write_opc_data_success(mock_dataframe, opc):
     # Arrange

tests/laborious/activities/test_prediction_metadata.py

@@ -0,0 +1,179 @@
+from unittest.mock import MagicMock, call, patch
+
+from pytest import fixture
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+
+from laborious.activities.prediction_metadata import PredictionMetadata
+
+
+@fixture
+def mongo() -> MagicMock:
+    return MagicMock()
+
+
+@fixture
+def prediction_metadata(mongo: MagicMock):
+    activity = PredictionMetadata(
+        mongo_repository=mongo,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    activity.error = MagicMock()
+    return activity
+
+
+metadata = {
+    'metadata': {
+        'model_id': 'test_model',
+        'model_name': 'test_model',
+        'workflow_name': 'test_workflow',
+        'schema_name': 'test_schedule',
+    },
+}
+
+
+def test_write_prediction_metadata_no_documents_skips_mongo(prediction_metadata, mongo: MagicMock):
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': []})
+
+    # Assert
+    assert result is None
+    mongo.insert.assert_not_called()
+
+
+def test_write_prediction_metadata_missing_documents_key_skips_mongo(
+    prediction_metadata, mongo: MagicMock
+):
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata})
+
+    # Assert
+    assert result is None
+    mongo.insert.assert_not_called()
+
+
+def test_write_prediction_metadata_documents_with_no_filters_skips_mongo(
+    prediction_metadata, mongo: MagicMock
+):
+    # Arrange: a falsy document (None) and one with an empty `filters` list — neither contributes
+    # anything to write.
+    documents = [None, {'filters': []}]
+
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': documents})
+
+    # Assert
+    assert result is None
+    mongo.insert.assert_not_called()
+
+
+def test_write_prediction_metadata_not_configured_returns_none():
+    # Arrange
+    activity = PredictionMetadata(
+        mongo_repository=None,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    activity.error = MagicMock()
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}
+
+    # Act
+    result = activity.write_prediction_metadata({**metadata, 'documents': [document]})
+
+    # Assert
+    assert result is None
+    activity.error.assert_not_called()
+
+
+def test_write_prediction_metadata_success_generates_id_and_inserts(
+    prediction_metadata, mongo: MagicMock
+):
+    # Arrange
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 3}]}
+
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': [document]})
+
+    # Assert: a plain 24-hex-char string id (matches predictions.metadata_id's varchar(24)),
+    # not a BSON ObjectId — `insert(...)` never hands one back, so it's generated client-side.
+    assert result is not None
+    assert len(result) == 24
+    int(result, 16)  # raises ValueError if not hex
+
+    collection, inserted_document, inserted_metadata = mongo.insert.call_args.args
+    assert collection == PredictionMetadata._PREDICTION_METADATA_COLLECTION
+    assert inserted_document == {'filters': document['filters'], '_id': result}
+    assert inserted_metadata == metadata['metadata']
+
+
+def test_write_prediction_metadata_combines_multiple_documents(
+    prediction_metadata, mongo: MagicMock
+):
+    # Arrange: two documents (e.g. from two different sources contributing metadata for the same
+    # prediction cycle) — their `filters` entries are flattened into one combined document.
+    documents = [
+        {'filters': [{'name': 'EMPTY_DATA', 'policy': 'CONTINUE', 'counter': 0}]},
+        {
+            'filters': [
+                {'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2}
+            ]
+        },
+    ]
+
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': documents})
+
+    # Assert
+    assert result is not None
+    _, inserted_document, _ = mongo.insert.call_args.args
+    assert inserted_document['filters'] == [
+        {'name': 'EMPTY_DATA', 'policy': 'CONTINUE', 'counter': 0},
+        {'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2},
+    ]
+
+
+def test_write_prediction_metadata_insert_failure_swallowed(prediction_metadata, mongo: MagicMock):
+    # Arrange
+    mongo.insert.side_effect = RuntimeError('connection reset')
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}
+
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': [document]})
+
+    # Assert: the prediction write must never be blocked or failed by this
+    assert result is None
+    prediction_metadata.error.assert_called_once()
+
+
+@patch.object(SientiaMonitoring, 'shutdown')
+def test_close_disposes_mongo_repository(mock_shutdown):
+    """``PredictionMetadata.close`` should close the Mongo client and clear the reference."""
+    mongo = MagicMock()
+    activity = PredictionMetadata(
+        mongo_repository=mongo,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    activity.close()
+    mongo.close.assert_called_once()
+    assert activity.mongo_repository is None
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(activity)) == 1
+
+
+@patch.object(SientiaMonitoring, 'shutdown')
+def test_close_without_mongo_repository(mock_shutdown):
+    """When no Mongo repository is configured, ``close`` only shuts down monitoring."""
+    activity = PredictionMetadata(
+        mongo_repository=None,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    activity.close()
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(activity)) == 1

tests/laborious/activities/test_storage.py

@@ -1,9 +1,10 @@
 import datetime
 import os
-from unittest.mock import ANY, MagicMock, patch
+from unittest.mock import ANY, MagicMock, call, patch
 
 from pytest import fixture, raises
 from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.metrics_controller import MetricsController
 from sientia_do.observability.sientia_monitoring import SientiaMonitoring
 from sientia_do.temporal.activities.postgres_sync import Postgres
 
@@ -52,7 +53,7 @@ def storage(mock_minio_repository):
         minio_repository=mock_minio_repository.return_value,
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
 
 
@@ -60,7 +61,7 @@ def storage(mock_minio_repository):
 def test___init___not_hasattr(mock_minio_repository):
     logger = MagicMock()
     notification_handler = MagicMock()
-    metrics_controller = MagicMock()
+    metrics_controller = MagicMock(spec=MetricsController)
     minio_repo = mock_minio_repository.return_value
     storage = Storage(
         host='localhost',
@@ -87,7 +88,7 @@ def test___init___none_minio_repository(mock_minio_repository, storage):
     storage.minio_repository = None
     logger = MagicMock()
     notification_handler = MagicMock()
-    metrics_controller = MagicMock()
+    metrics_controller = MagicMock(spec=MetricsController)
     storage.__init__(
         host='localhost',
         port=5432,
@@ -121,7 +122,7 @@ def test___init___done_repository(mock_minio_repository, storage):
         minio_repository=mock_minio_repository.return_value,
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     mock_minio_repository.assert_not_called()
     assert storage.minio_repository is not None
@@ -133,7 +134,8 @@ def test_close(storage, _patch_monitoring_shutdown):
     storage.close()
 
     assert storage.minio_repository is None
-    _patch_monitoring_shutdown.assert_called_once_with(storage)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert _patch_monitoring_shutdown.call_args_list.count(call(storage)) == 1
 
 
 def test_close_when_minio_repository_already_none(storage, _patch_monitoring_shutdown):
@@ -143,7 +145,8 @@ def test_close_when_minio_repository_already_none(storage, _patch_monitoring_shu
     storage.close()
 
     assert storage.minio_repository is None
-    _patch_monitoring_shutdown.assert_called_once_with(storage)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert _patch_monitoring_shutdown.call_args_list.count(call(storage)) == 1
 
 
 def test_load_query_with_minio_offload_no_rows(storage):
@@ -254,6 +257,86 @@ def test_export_payload_to_postgres(storage):
     assert result == {'success': True}
 
 
+def test_build_repeat_last_prediction_query(storage):
+    query = storage._build_repeat_last_prediction_query('public', 'predictions')
+    query_str = str(query)
+
+    assert 'INSERT INTO "public".predictions' in query_str
+    assert 'metadata_id' in query_str
+    assert ':last_timestamp' in query_str
+    assert ':metadata_id' in query_str
+    assert ':model' in query_str
+
+
+@patch('laborious.activities.storage.text')
+def test_repeat_last_prediction_success(text_mock, storage):
+    storage.emit_metric_sync = MagicMock()
+    storage.observe_lag_sync = MagicMock()
+    query_items = {
+        **metadata,
+        'schema': 'public',
+        'table_name': 'predictions',
+        'model': 1,
+        'last_timestamp': '2022-01-01',
+        'metadata_id': 'abc123',
+    }
+
+    with patch('sqlalchemy.orm.session.Session.execute') as mock_execute:
+        storage.repeat_last_prediction(query_items)
+
+    text_mock.assert_called_once()
+    mock_execute.assert_called_once_with(
+        text_mock.return_value,
+        {'last_timestamp': '2022-01-01', 'model': 1, 'metadata_id': 'abc123'},
+    )
+    storage.observe_lag_sync.assert_called_once()
+    storage.emit_metric_sync.assert_called_once()
+
+
+@patch('laborious.activities.storage.text')
+def test_repeat_last_prediction_without_metadata_id(text_mock, storage):
+    storage.emit_metric_sync = MagicMock()
+    storage.observe_lag_sync = MagicMock()
+    query_items = {
+        **metadata,
+        'schema': 'public',
+        'table_name': 'predictions',
+        'model': 1,
+        'last_timestamp': '2022-01-01',
+    }
+
+    with patch('sqlalchemy.orm.session.Session.execute') as mock_execute:
+        storage.repeat_last_prediction(query_items)
+
+    mock_execute.assert_called_once_with(
+        text_mock.return_value,
+        {'last_timestamp': '2022-01-01', 'model': 1, 'metadata_id': None},
+    )
+
+
+@patch('laborious.activities.storage.text')
+def test_repeat_last_prediction_error(text_mock, storage):
+    storage.emit_metric_sync = MagicMock()
+    storage.send_notification = MagicMock()
+    query_items = {
+        **metadata,
+        'schema': 'public',
+        'table_name': 'predictions',
+        'model': 1,
+        'last_timestamp': '2022-01-01',
+        'metadata_id': 'abc123',
+    }
+
+    with (
+        patch('sqlalchemy.orm.session.Session.execute', side_effect=ValueError('db')),
+        raises(ValueError),
+    ):
+        storage.repeat_last_prediction(query_items)
+
+    storage.send_notification.assert_called_once()
+    storage.emit_metric_sync.assert_called()
+
+
 def test_cleanup_minio_objects_expired_minio_not_initialized(storage):
     storage.minio_repository = None
 
@@ -322,3 +405,17 @@ def test_cleanup_minio_objects_expired_list_objects_error(mock_now, storage):
         attachment_content=ANY,
     )
     storage.error.assert_called_once()
+
+
+def test_close_without_engine_skips_postgres(storage, _patch_monitoring_shutdown):
+    """A Storage whose Postgres init never ran (no engine) still closes MinIO."""
+    del storage.engine
+    minio = MagicMock()
+    storage.minio_repository = minio
+
+    with patch('laborious.activities.storage.Postgres.close') as mock_postgres_close:
+        storage.close()
+
+    mock_postgres_close.assert_not_called()
+    minio.close.assert_called_once()
+    assert storage.minio_repository is None

tests/laborious/utils/models/test_minio_dataframe_payload.py

@@ -220,6 +220,26 @@ def test_from_dataframe_offloaded(mock_now):
     minio.upload_file.assert_called_once()
 
 
+@patch('laborious.utils.models.minio_dataframe_payload.OFFLOAD_THRESHOLD_BYTES', 10**9)
+def test_from_dataframe_inline_keeps_a_provided_last_timestamp():
+    """An explicit last_timestamp must survive as-is, skipping the max() derivation."""
+    minio = MagicMock()
+    logger = MagicMock()
+    df = _mock_dataframe({'timestamp': ['2024-01-01'], 'value': [42]})
+
+    result = MinioDataFramePayload.from_dataframe(
+        dataframe=df,
+        minio_repo=minio,
+        model_name='m',
+        operation='initial',
+        last_timestamp='2030-01-01T00:00:00+00:00',
+        logger=logger,
+    )
+
+    assert result.last_timestamp == '2030-01-01T00:00:00+00:00'
+    logger.custom_debug.assert_called()
+
+
 def test_from_dict_inline():
     raw = {
         'last_timestamp': '2024-01-01T00:00:00+00:00',

tests/laborious/utils/repository/test_opc_repository.py

@@ -1,5 +1,7 @@
 import concurrent.futures
 import json
+import threading
+import time
 from datetime import datetime
 from unittest.mock import ANY, MagicMock, Mock, patch
 
@@ -7,12 +9,15 @@ import pytest
 from asyncua.crypto import security_policies
 from asyncua.ua.uaerrors import BadNodeIdUnknown, BadSessionIdInvalid
 from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.metrics_controller import MetricsController
 
+from laborious import metrics
 from laborious.utils.repository.opc_repository import (
     OpcClientAlreadyExistsError,
     OpcClientNotInitializedError,
     OpcRepository,
     OpcSessionAlreadyConnectedError,
+    _opc_authentication_token_str,
     is_reconnectable_opcua_bad,
 )
 
@@ -35,7 +40,7 @@ def opc_repository(mock_logger):
         cert_path='/path/to/cert.pem',
         private_key_path='/path/to/key.pem',
         server_cert_path='/path/to/server_cert.pem',
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     repository.disconnection_interval = 0.1
     repository.send_notification = MagicMock()
@@ -86,6 +91,40 @@ def test_init(opc_repository):
     assert opc_repository.last_reconnection_time is None
 
 
+def test_opc_authentication_token_str_no_protocol():
+    client = MagicMock()
+    client.aio_obj.uaclient.protocol = None
+
+    assert _opc_authentication_token_str(client) == 'unknown'
+
+
+def test_opc_authentication_token_str_no_token():
+    client = MagicMock()
+    client.aio_obj.uaclient.protocol = MagicMock(authentication_token=None)
+
+    assert _opc_authentication_token_str(client) == 'unknown'
+
+
+def test_opc_authentication_token_str_swallows_errors():
+    class _ExplodingClient:
+        @property
+        def aio_obj(self):
+            raise RuntimeError('boom')
+
+    assert _opc_authentication_token_str(_ExplodingClient()) == 'unknown'
+
+
+def test_is_session_open_swallows_errors(opc_repository):
+    class _ExplodingClient:
+        @property
+        def aio_obj(self):
+            raise RuntimeError('boom')
+
+    opc_repository.client = _ExplodingClient()
+
+    assert opc_repository._is_session_open() is False
+
+
 def test_set_security(opc_repository, mock_client):
     opc_repository.client = mock_client
     opc_repository.set_security()
@@ -112,6 +151,13 @@ def test_set_security_missing_certificates(opc_repository):
         assert str(e) == 'Certificate and private key paths must be provided for secure connection.'
 
 
+def test_set_security_missing_server_uri(opc_repository):
+    opc_repository.server_uri = None
+
+    with pytest.raises(ValueError, match='server_uri must be provided for secure connection.'):
+        opc_repository.set_security()
+
+
 def test_set_security_missing_client(opc_repository):
     opc_repository.client = None
     try:
@@ -160,6 +206,43 @@ def test_create_client_raises_when_client_exists(opc_repository, mock_client):
         opc_repository._create_client()
 
 
+def test_create_client_builds_a_secured_client(opc_repository, mock_client):
+    opc_repository.pod_id = 'pod-1'
+    opc_repository.set_security = MagicMock()
+
+    opc_repository._create_client()
+
+    assert opc_repository.client is mock_client
+    assert mock_client.aio_obj.name == 'pod-1'
+    assert mock_client.application_uri == 'pod:1'
+    assert mock_client.aio_obj.product_uri == 'pod:1'
+    opc_repository.set_security.assert_called_once()
+
+
+def test_create_client_skips_security_without_a_cert_path(opc_repository, mock_client):
+    opc_repository.cert_path = None
+    opc_repository.set_security = MagicMock()
+
+    opc_repository._create_client()
+
+    assert opc_repository.client is mock_client
+    opc_repository.set_security.assert_not_called()
+
+
+def test_connect_locked_reuses_an_existing_client(opc_repository, mock_client):
+    """A client already set (e.g. surviving a prior failed open) is not recreated."""
+    opc_repository.client = mock_client
+    opc_repository._is_session_open = MagicMock(return_value=False)
+    opc_repository._create_client = MagicMock()
+    opc_repository._open_session = MagicMock(return_value=(True, {}))
+
+    result = opc_repository._connect_locked()
+
+    opc_repository._create_client.assert_not_called()
+    opc_repository._open_session.assert_called_once()
+    assert result == (True, {})
+
+
 def test_open_session_success(opc_repository):
     closed_proto = MagicMock()
     closed_proto.state = 'closed'
@@ -361,6 +444,22 @@ def test_write_data_get_node_failed(opc_repository):
     assert error_data['attachment_content'] is not None
 
 
+def test_write_data_without_client_fails_as_get_node_error(opc_repository):
+    """A client torn down between validation and the write is reported, not asserted away."""
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
+    opc_repository.client = None
+
+    is_success, error_data = opc_repository.write_data(
+        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
+    )
+
+    assert is_success is False
+    assert error_data['notification_id'] == f'OPC_WRITE_GET_NODE_ERROR_{opc_repository.id}'
+    assert 'OPC client is not initialized' in error_data['message']
+    last_call = opc_repository.emit_metric_sync.call_args_list[-1]
+    assert last_call.args[1]['result'] == 'GetNodeError:OpcClientNotInitializedError'
+
+
 def test_write_data_invalid_data_type(opc_repository, mock_client):
     opc_repository.validate_connection = MagicMock(return_value=(True, {}))
     opc_repository.client = mock_client
@@ -423,6 +522,75 @@ def test_write_data_write_value_failed(opc_repository, mock_client):
     assert error_data['attachment_content'] is not None
 
 
+def test_write_data_with_empty_metadata_uses_unknown_labels(opc_repository, mock_client):
+    """No model_id/model_name in the write metadata: the metric labels fall back to 'unknown'."""
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
+    opc_repository.client = mock_client
+    mock_client.get_node = MagicMock(return_value=MagicMock())
+
+    result = opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', {})
+
+    assert result == (True, {'response_time': ANY})
+    last_call = opc_repository.emit_metric_sync.call_args_list[-1]
+    assert last_call.args[0] is metrics.OPC_WRITE_ATTEMPTS_TOTAL
+    assert last_call.args[1]['model_id'] == 'unknown'
+    assert last_call.args[1]['model_name'] == 'unknown'
+
+
+def test_write_data_get_node_bad_session_id_schedules_reconnect(opc_repository, mock_client):
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
+    opc_repository.client = mock_client
+    opc_repository._start_reconnect = MagicMock()
+    mock_client.get_node = MagicMock(side_effect=BadSessionIdInvalid())
+
+    is_success, error_data = opc_repository.write_data(
+        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
+    )
+
+    opc_repository._start_reconnect.assert_called_once()
+    assert is_success is False
+    assert error_data['opc_error_kind'] == 'session_bad'
+    assert error_data['opc_status'] == 'BadSessionIdInvalid'
+
+
+def test_write_data_logs_inter_arrival_and_flags_a_stale_session(opc_repository, mock_client):
+    """The second write on the same session logs the gap since the previous one, and flags it
+    when the gap outlives the session's revised timeout."""
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
+    opc_repository.client = mock_client
+    mock_client.aio_obj.session_timeout = 0.001  # 1 microsecond: any real gap will exceed it
+    mock_client.get_node = MagicMock(return_value=MagicMock())
+
+    opc_repository.write_data('ns=2;s=TestNode', 1.0, 'float', metadata['metadata'])
+    opc_repository.write_data('ns=2;s=TestNode', 2.0, 'float', metadata['metadata'])
+
+    logged = [c.args[0] for c in opc_repository.info.call_args_list]
+    assert any('OPC write inter-arrival' in message for message in logged)
+    emitted_metrics = [c.args[0] for c in opc_repository.emit_metric_sync.call_args_list]
+    assert metrics.OPC_WRITE_INTER_ARRIVAL_OVER_SESSION_TIMEOUT_TOTAL in emitted_metrics
+
+
+def test_log_write_inter_arrival_skips_the_metric_without_a_client(opc_repository):
+    opc_repository._last_write_mono = time.monotonic() - 1
+    opc_repository.client = None
+
+    opc_repository._log_write_inter_arrival('session-1', 'ns=2;s=Node')
+
+    opc_repository.emit_metric_sync.assert_not_called()
+
+
+def test_log_write_inter_arrival_skips_the_metric_within_the_session_timeout(
+    opc_repository, mock_client
+):
+    opc_repository._last_write_mono = time.monotonic()
+    opc_repository.client = mock_client
+    mock_client.aio_obj.session_timeout = 600_000  # 600s: comfortably longer than this test takes
+
+    opc_repository._log_write_inter_arrival('session-1', 'ns=2;s=Node')
+
+    opc_repository.emit_metric_sync.assert_not_called()
+
+
 def test_is_reconnectable_opcua_bad():
     assert is_reconnectable_opcua_bad(BadSessionIdInvalid()) is True
     assert is_reconnectable_opcua_bad(BadNodeIdUnknown()) is False
@@ -472,6 +640,23 @@ def test_start_reconnect_skips_within_interval(opc_repository):
     assert opc_repository._reconnect_thread is None
 
 
+def test_start_reconnect_skips_when_already_in_progress(opc_repository):
+    opc_repository.last_reconnection_time = None
+    release = threading.Event()
+    in_progress_thread = threading.Thread(target=release.wait, daemon=True)
+    in_progress_thread.start()
+    opc_repository._reconnect_thread = in_progress_thread
+
+    try:
+        opc_repository._start_reconnect('BadSessionIdInvalid', 'tok')
+
+        assert opc_repository._reconnect_thread is in_progress_thread
+        opc_repository.warning.assert_called_once()
+    finally:
+        release.set()
+        in_progress_thread.join(timeout=5.0)
+
+
 def test_write_data_protocol_closed_schedules_reconnect(opc_repository):
     opc_repository.client = MagicMock()
     opc_repository.client.aio_obj.uaclient.protocol = MagicMock(state='closed')
@@ -582,3 +767,77 @@ def test_reconnect_locked_sets_last_reconnection_time(mock_datetime, opc_reposit
     opc_repository._connect_locked.assert_called_once()
     assert result == (True, {})
     assert opc_repository.last_reconnection_time == datetime(2025, 1, 1, 12, 0, 0)
+
+
+def test_run_reconnect_failed_without_error_detail_logs_once(opc_repository):
+    """A falsy `error` (no message to add) must not log a second, empty line."""
+    opc_repository._reconnect_locked = MagicMock(return_value=(False, {}))
+
+    opc_repository._run_reconnect('BadSessionIdInvalid', 'tok')
+
+    assert opc_repository.error.call_count == 1
+
+
+def test_run_reconnect_swallows_an_unexpected_exception(opc_repository):
+    """A bug inside the reconnect itself must not escape the background thread."""
+    opc_repository._reconnect_locked = MagicMock(side_effect=RuntimeError('boom'))
+
+    opc_repository._run_reconnect('BadSessionIdInvalid', 'tok')
+
+    assert opc_repository.error.call_count >= 2
+
+
+def test_create_client_without_watchdog_attribute(opc_repository, mock_client):
+    """Older asyncua clients have no `watchdog_intervall`; it is not invented on them."""
+    opc_repository.cert_path = None
+    mock_client.aio_obj = MagicMock(spec=['name', 'description', 'product_uri', 'uaclient'])
+
+    opc_repository._create_client()
+
+    assert opc_repository.client is mock_client
+    assert not hasattr(mock_client.aio_obj, 'watchdog_intervall')
+
+
+def test_run_reconnect_success_logs_no_error(opc_repository):
+    opc_repository._reconnect_locked = MagicMock(return_value=(True, {}))
+
+    opc_repository._run_reconnect('BadSessionIdInvalid', 'tok')
+
+    opc_repository._reconnect_locked.assert_called_once()
+    opc_repository.error.assert_not_called()
+
+
+@pytest.mark.parametrize('failure', [AttributeError('no write_value'), TypeError('bad args')])
+def test_write_node_value_falls_back_to_set_value(failure):
+    node = MagicMock()
+    node.write_value.side_effect = failure
+
+    OpcRepository._write_node_value(node, 'ua-data', 42.0, 'variant')
+
+    node.write_value.assert_called_once_with('ua-data')
+    node.set_value.assert_called_once_with(42.0, 'variant')
+
+
+def test_write_node_value_uses_set_value_without_write_value():
+    node = MagicMock(spec=['set_value'])
+
+    OpcRepository._write_node_value(node, 'ua-data', 42.0, 'variant')
+
+    node.set_value.assert_called_once_with(42.0, 'variant')
+
+
+def test_write_data_session_not_ready_reports_the_reconnect_it_started(opc_repository):
+    """When the SessionNotReady reconnect starts, the write reports reconnect_in_progress."""
+    opc_repository._session_ready.clear()
+    opc_repository._reconnect_thread_in_progress = MagicMock(side_effect=[False, True])
+    opc_repository._start_reconnect = MagicMock()
+    opc_repository.validate_connection = MagicMock()
+
+    is_success, error_data = opc_repository.write_data(
+        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
+    )
+
+    opc_repository._start_reconnect.assert_called_once_with('SessionNotReady', ANY)
+    opc_repository.validate_connection.assert_not_called()
+    assert is_success is False
+    assert error_data['opc_error_kind'] == 'reconnect_in_progress'

tests/laborious/utils/test_dataframe_debug.py

@@ -1,12 +1,40 @@
+"""Debug message formatting: full dataframe when small, a summary when it would flood the log."""
+
 from pandas import DataFrame
 
-from laborious.utils.dataframe_debug import build_dataframe_debug_message
+from laborious.utils.dataframe_debug import (
+    DEFAULT_MAX_DEBUG_DATAFRAME_ROWS,
+    build_dataframe_debug_message,
+)
 
 
-def test_build_dataframe_debug_message_skips_large_dataframe():
-    df = DataFrame({'a': [1, 2, 3]})
+def test_non_dataframe_payloads_are_stringified():
+    assert build_dataframe_debug_message('payload:', {'a': 1}) == "payload: {'a': 1}"
 
-    msg = build_dataframe_debug_message('payload', df, max_rows=1)
 
-    assert 'skipped because dataframe has 3 rows' in msg
-    assert '(max: 1)' in msg
+def test_a_small_dataframe_is_logged_in_full():
+    data = DataFrame({'a': [1, 2]})
+
+    message = build_dataframe_debug_message('payload:', data)
+
+    assert message.startswith('payload:\n')
+    assert data.to_csv() in message
+
+
+def test_a_dataframe_over_the_row_limit_is_summarized_instead():
+    data = DataFrame({'a': range(DEFAULT_MAX_DEBUG_DATAFRAME_ROWS + 1)})
+
+    message = build_dataframe_debug_message('payload:', data)
+
+    assert 'skipped because dataframe has' in message
+    assert str(data.shape) in message
+    assert data.to_csv() not in message
+
+
+def test_the_row_limit_is_configurable():
+    data = DataFrame({'a': [1, 2, 3]})
+
+    message = build_dataframe_debug_message('payload:', data, max_rows=1)
+
+    assert 'skipped because dataframe has 3 rows' in message
+    assert '(max: 1)' in message

tests/laborious/utils/test_time.py

@@ -0,0 +1,19 @@
+from datetime import UTC, datetime
+
+from laborious.utils.time import to_naive
+
+
+def test_to_naive_returns_naive_input_unchanged() -> None:
+    moment = datetime(2024, 1, 1, 12, 30, 45)
+
+    assert to_naive(moment) == moment
+    assert to_naive(moment).tzinfo is None
+
+
+def test_to_naive_strips_tzinfo_from_aware_input() -> None:
+    moment = datetime(2024, 1, 1, 12, 30, 45, tzinfo=UTC)
+
+    result = to_naive(moment)
+
+    assert result.tzinfo is None
+    assert result == datetime(2024, 1, 1, 12, 30, 45)

tests/laborious/worker/test_worker.py

@@ -48,21 +48,25 @@ def _build_fake_worker(async_result=None, async_error: Exception | None = None):
 
 @patch('laborious.worker.worker.start_http_server')
 def test_start_prometheus_server_success(mock_start_http):
+    logger = MagicMock()
     with patch.object(worker.metrics.APP_UP, 'labels') as labels:
         gauge = MagicMock()
         labels.return_value = gauge
         with patch('laborious.worker.worker.os.getenv', return_value='9090'):
-            worker.start_prometheus_server()
+            worker.start_prometheus_server(logger, {})
     mock_start_http.assert_called_once_with(9090)
     gauge.set.assert_called_once_with(1)
+    logger.custom_info.assert_called_once_with('Prometheus server started on port 9090.', {})
 
 
 @patch('laborious.worker.worker.start_http_server', side_effect=RuntimeError('nope'))
 def test_start_prometheus_server_error_exits(_mock_start_http):
+    logger = MagicMock()
     with patch('laborious.worker.worker.os._exit', side_effect=SystemExit(1)) as m_exit:
         with raises(SystemExit):
-            worker.start_prometheus_server()
+            worker.start_prometheus_server(logger, {})
     m_exit.assert_called_once_with(1)
+    logger.custom_critical.assert_called_once_with('Failed to start Prometheus server: nope', {})
 
 
 @mark.asyncio
@@ -181,6 +185,9 @@ async def test_main_success_exit_zero(monkeypatch):
     prepare_calls = m_prepare.call_args_list
     assert prepare_calls[0].kwargs['runtime'] == 'single'
     assert prepare_calls[3].kwargs['runtime'] == 'single'
+    # Drift and SimpleMetrics load no model code, so every runtime shares the `core` queues.
+    assert prepare_calls[1].kwargs['runtime'] == 'core'
+    assert prepare_calls[2].kwargs['runtime'] == 'core'
 
 
 @mark.asyncio
@@ -241,3 +248,45 @@ async def test_main_worker_gather_error_exits_one(monkeypatch):
             await worker.main()
 
     assert m_logger.return_value.custom_error.called
+
+
+@mark.asyncio
+async def test_main_legacy_runtime_installs_single(monkeypatch):
+    """The `legacy` runtime is served by the `single` store runtime."""
+    monkeypatch.setenv('RUNTIME', 'legacy')
+    fake_plugin = MagicMock()
+    fake_plugin.install_runtime = AsyncMock(side_effect=RuntimeError('stop after install'))
+    with (
+        patch('laborious.worker.worker.start_prometheus_server'),
+        patch('laborious.worker.worker.get_logger'),
+        patch(
+            'laborious.worker.worker.build_mongodb_config',
+            return_value={'connection_string': 'cs', 'database_name': 'db'},
+        ),
+        patch('laborious.worker.worker.NotificationHandler'),
+        patch('laborious.worker.worker.MetricsController'),
+        patch(
+            'laborious.worker.worker.build_plugin_store_config',
+            return_value={
+                'base_url': '',
+                'owner': '',
+                'repo': '',
+                'username': None,
+                'password': None,
+                'branch': None,
+                'cache_ttl_seconds': None,
+                'pypi_index_url': '',
+                'pypi_username': None,
+                'pypi_password': None,
+            },
+        ),
+        patch('laborious.worker.worker.PluginStore', return_value=fake_plugin),
+        patch.object(worker.metrics.APP_UP, 'labels') as labels,
+        patch('laborious.worker.worker.sys.exit', side_effect=SystemExit(1)),
+    ):
+        labels.return_value = MagicMock()
+        with raises(SystemExit):
+            await worker.main()
+
+    fake_plugin.install_runtime.assert_awaited_once()
+    assert fake_plugin.install_runtime.await_args.kwargs['runtime_name'] == 'single'

tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py

@@ -326,6 +326,7 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction
                     'model_id': input_data['model_id'],
                     'prediction_confidence': input_data['prediction_confidence'],
                     'comment': input_data['comment'],
+                    'metadata_id': None,
                 },
                 retry_policy=ANY,
                 start_to_close_timeout=ANY,
@@ -389,6 +390,56 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction
     assert workflow_mock.execute_local_activity_method.call_count == 1
 
 
+@mark.asyncio
+@patch(
+    'laborious.workflows.sub_workflows.format_and_export_prediction.workflow',
+    new_callable=AsyncMock,
+)
+async def test_run_default_path_flag_passes_metadata_id(
+    workflow_mock, format_and_export_prediction
+):
+    # Arrange: metadata_id reaches format_default_prediction, which is the row that ends up in
+    # Postgres via export_data_to_postgres (CONTINUE/REPEAT path).
+    input_data = {
+        'metadata': metadata,
+        'path_flag': 'default',
+        'data': {'test': 'data'},
+        'timestamp': '2021-01-01',
+        'model_id': 1,
+        'model_name': metadata['metadata']['model_name'],
+        'prediction_confidence': 0,
+        'schema': 'test_schema',
+        'table_name': 'test_table',
+        'opc_output_config': {'test': 'config'},
+        'comment': 'test_comment',
+        'metadata_id': 'a' * 24,
+    }
+
+    workflow_mock.execute_activity_method.side_effect = [
+        (MagicMock(), MagicMock()),
+        MagicMock(),
+        MagicMock(),
+    ]
+
+    # Act
+    await format_and_export_prediction.run(input_data)
+
+    # Assert
+    workflow_mock.execute_local_activity_method.assert_any_call(
+        Activities.format_default_prediction,
+        {
+            **metadata,
+            'timestamp': input_data['timestamp'],
+            'model_id': input_data['model_id'],
+            'prediction_confidence': input_data['prediction_confidence'],
+            'comment': input_data['comment'],
+            'metadata_id': 'a' * 24,
+        },
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+
+
 @mark.asyncio
 @patch(
     'laborious.workflows.sub_workflows.format_and_export_prediction.workflow',
@@ -638,6 +689,20 @@ async def test_run_default_path_flag_with_pi_web_api(workflow_mock, format_and_e
 
     await format_and_export_prediction.run(input_data)
 
+    workflow_mock.execute_local_activity_method.assert_any_call(
+        Activities.format_default_prediction,
+        {
+            **metadata,
+            'timestamp': input_data['timestamp'],
+            'model_id': input_data['model_id'],
+            'prediction_confidence': input_data['prediction_confidence'],
+            'comment': input_data['comment'],
+            'metadata_id': None,
+        },
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+
     workflow_mock.execute_activity_method.assert_has_calls(
         [
             call(

tests/laborious/workflows/subworkflows/test_prediction_process.py

@@ -57,7 +57,7 @@ async def test_run(workflow_mock, prediction_process):
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('continue', 0.95, 'Input data with bad quality'),  # input_gate
+        ('continue', 0.95, 'Input data with bad quality', None),  # input_gate
         # mlflow_response_gate (transform)
         ('continue', 0.95, 'Error'),
         # mlflow_content_gate (transform)
@@ -217,7 +217,7 @@ async def test_run_stop_at_input_gate(workflow_mock, prediction_process):
 
     # Mock the activity responses
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('stop', 0.95, 'Input data with bad quality'),  # input_gate
+        ('stop', 0.95, 'Input data with bad quality', None),  # input_gate
     ]
 
     # Act
@@ -275,7 +275,7 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('repeat', 0.95, 'Input data with bad quality'),  # input_gate
+        ('repeat', 0.95, 'Input data with bad quality', None),  # input_gate
         ('continue', 0.95, 'Error'),  # mlflow_response_gate (transform)
     ]
 
@@ -365,7 +365,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('continue', 0.95, 'Input data with bad quality'),  # input_gate
+        ('continue', 0.95, 'Input data with bad quality', None),  # input_gate
         # mlflow_response_gate (transform)
         ('continue', 0.95, 'Error'),
         # mlflow_content_gate (transform)
@@ -476,7 +476,7 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('continue', 0.95, 'Input data with bad quality'),  # input_gate
+        ('continue', 0.95, 'Input data with bad quality', None),  # input_gate
         # mlflow_response_gate (transform)
         ('continue', 0.95, 'Error'),
         # mlflow_content_gate (transform)
@@ -598,6 +598,7 @@ async def test_path_flag_handler_stop(workflow_mock, prediction_process):
     last_timestamp = '2024-01-01'
     model_name = 'test_model_name'
     model_config = {'retention': '30'}
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}
 
     # Act
     result = await prediction_process.path_flag_handler(
@@ -616,18 +617,22 @@ async def test_path_flag_handler_stop(workflow_mock, prediction_process):
         confidence,
         last_timestamp,
         '',
+        document,
     )
 
-    # Assert
+    # Assert: STOP never reaches a write, even though it now gets a `document` same as CONTINUE
+    # and REPEAT do — it just never has anywhere to use it.
     assert result is True
     workflow_mock.execute_local_activity_method.assert_not_called()
+    workflow_mock.execute_activity_method.assert_not_called()
     workflow_mock.execute_child_workflow.assert_not_called()
 
 
 @mark.asyncio
 @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
 async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
-    # Arrange
+    # Arrange: REPEAT now writes metadata just like CONTINUE — it does insert a new `predictions`
+    # row (reusing the previous row's values), so it needs a `metadata_id` too.
     data = {'test': 'data'}
     path_flag = 'repeat'
     confidence = 0.95
@@ -637,6 +642,7 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
     last_timestamp = '2024-01-01'
     model_name = 'test_model_name'
     model_config = {'retention': '30'}
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'REPEAT', 'counter': 0}]}
 
     # Act
     result = await prediction_process.path_flag_handler(
@@ -655,11 +661,20 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
         confidence,
         last_timestamp,
         '',
+        document,
     )
 
     # Assert
     assert result is True
-    workflow_mock.execute_activity_method.assert_called_once_with(
+    metadata_id = workflow_mock.execute_activity_method.return_value
+    assert workflow_mock.execute_activity_method.call_count == 2
+    workflow_mock.execute_activity_method.assert_any_call(
+        Activities.write_prediction_metadata,
+        {**metadata, 'documents': [document]},
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+    workflow_mock.execute_activity_method.assert_any_call(
         Activities.repeat_last_prediction,
         {
             **metadata,
@@ -667,6 +682,7 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
             'table_name': table_name,
             'model': model,
             'last_timestamp': last_timestamp,
+            'metadata_id': metadata_id,
         },
         retry_policy=ANY,
         start_to_close_timeout=ANY,
@@ -674,6 +690,56 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
     workflow_mock.execute_child_workflow.assert_not_called()
 
 
+@mark.asyncio
+@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
+async def test_path_flag_handler_repeat_no_document_skips_write_activity_call(
+    workflow_mock, prediction_process
+):
+    # Arrange: no document (e.g. REPEAT triggered by a gate that doesn't build one yet — today,
+    # `mlflow_response_gate`/`mlflow_content_gate`) must not even dispatch the remote
+    # write_prediction_metadata activity — that would be a wasted Temporal round-trip on every
+    # REPEAT caused by those two gates, not a rare edge case (review finding, 2026-10-01).
+    data = {'test': 'data'}
+
+    # Act
+    result = await prediction_process.path_flag_handler(
+        data,
+        'repeat',
+        {
+            'metadata': metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'transform_table_name': 'test_transform_table',
+            'model_id': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'model_name': 'test_model_name',
+            'model_config': {'retention': '30'},
+        },
+        0.95,
+        '2024-01-01',
+        '',
+    )
+
+    # Assert
+    assert result is True
+    called_activities = [c.args[0] for c in workflow_mock.execute_activity_method.call_args_list]
+    assert Activities.write_prediction_metadata not in called_activities
+    # repeat_last_prediction still happens, just with metadata_id=None.
+    workflow_mock.execute_activity_method.assert_any_call(
+        Activities.repeat_last_prediction,
+        {
+            **metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'model': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'metadata_id': None,
+        },
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+
+
 @mark.asyncio
 @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
 async def test_path_flag_handler_continue(workflow_mock, prediction_process):
@@ -688,6 +754,9 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
     model_name = 'test_model_name'
     model_config = {'retention': '30'}
     prediction_store_policy = 'erl:1'
+    document = {
+        'filters': [{'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2}]
+    }
 
     # Act
     result = await prediction_process.path_flag_handler(
@@ -713,11 +782,18 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
         confidence,
         last_timestamp,
         'Prediction Process',
+        document,
     )
 
     # Assert
+    metadata_id = workflow_mock.execute_activity_method.return_value
     assert result is True
-    workflow_mock.execute_activity_method.assert_not_called()
+    workflow_mock.execute_activity_method.assert_called_once_with(
+        Activities.write_prediction_metadata,
+        {**metadata, 'documents': [document]},
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
     workflow_mock.execute_child_workflow.assert_called_once_with(
         'subworkflow.format_and_export_prediction',
         {
@@ -733,6 +809,7 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
             'table_name': table_name,
             'transform_table_name': 'test_transform_table',
             'comment': 'Prediction Process',
+            'metadata_id': metadata_id,
             'opc_output_config': {'test': 'config'},
             'pi_web_api_output_config': {
                 'endpoint': 'https://test-pi-server.com',
@@ -745,6 +822,153 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
     )
 
 
+@mark.asyncio
+@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
+async def test_path_flag_handler_continue_no_document_skips_write_activity_call(
+    workflow_mock, prediction_process
+):
+    # Arrange: same gap as the REPEAT case above (review finding, 2026-10-01) — no document must
+    # not dispatch the remote write_prediction_metadata activity.
+    data = {'test': 'data'}
+
+    # Act
+    result = await prediction_process.path_flag_handler(
+        data,
+        'continue',
+        {
+            'metadata': metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'transform_table_name': 'test_transform_table',
+            'model_id': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'model_name': 'test_model_name',
+            'model_config': {'retention': '30'},
+            'opc_output_config': {'test': 'config'},
+            'pi_web_api_output_config': {
+                'endpoint': 'https://test-pi-server.com',
+                'prediction_tags': {},
+                'confidence_tags': {},
+            },
+            'prediction_store_policy': 'erl:1',
+        },
+        0.95,
+        '2024-01-01',
+        '',
+    )
+
+    # Assert
+    assert result is True
+    workflow_mock.execute_activity_method.assert_not_called()
+    workflow_mock.execute_child_workflow.assert_called_once_with(
+        'subworkflow.format_and_export_prediction',
+        ANY,
+    )
+    _, export_input = workflow_mock.execute_child_workflow.call_args.args
+    assert export_input['metadata_id'] is None
+
+
+@mark.asyncio
+@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
+async def test_path_flag_handler_continue_write_metadata_failure_does_not_block_prediction(
+    workflow_mock, prediction_process
+):
+    # Arrange: write_prediction_metadata's retry_policy has maximum_attempts=1 — a timeout, an
+    # unavailable worker, or any dispatch failure raises straight into the workflow with no
+    # retry. Business rule 1 says this must never block the prediction write (review finding,
+    # 2026-10-01) — confirm the CONTINUE branch still proceeds instead of letting it propagate.
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'CONTINUE', 'counter': 0}]}
+    workflow_mock.execute_activity_method.side_effect = RuntimeError('activity dispatch failed')
+
+    # Act
+    result = await prediction_process.path_flag_handler(
+        {'test': 'data'},
+        'continue',
+        {
+            'metadata': metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'transform_table_name': 'test_transform_table',
+            'model_id': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'model_name': 'test_model_name',
+            'model_config': {'retention': '30'},
+            'opc_output_config': {'test': 'config'},
+            'pi_web_api_output_config': {
+                'endpoint': 'https://test-pi-server.com',
+                'prediction_tags': {},
+                'confidence_tags': {},
+            },
+            'prediction_store_policy': 'erl:1',
+        },
+        0.95,
+        '2024-01-01',
+        '',
+        document,
+    )
+
+    # Assert: the prediction write still happens, with metadata_id=None, instead of raising.
+    assert result is True
+    workflow_mock.execute_child_workflow.assert_called_once_with(
+        'subworkflow.format_and_export_prediction',
+        ANY,
+    )
+    _, export_input = workflow_mock.execute_child_workflow.call_args.args
+    assert export_input['metadata_id'] is None
+
+
+@mark.asyncio
+@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
+async def test_path_flag_handler_repeat_write_metadata_failure_does_not_block_prediction(
+    workflow_mock, prediction_process
+):
+    # Arrange: only the write_prediction_metadata call (first) fails — repeat_last_prediction
+    # (second) must still run normally; its own failure is a separate, intentional
+    # non-best-effort path, not what this test is about.
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'REPEAT', 'counter': 0}]}
+    workflow_mock.execute_activity_method.side_effect = [
+        RuntimeError('activity dispatch failed'),
+        MagicMock(),
+    ]
+
+    # Act
+    result = await prediction_process.path_flag_handler(
+        {'test': 'data'},
+        'repeat',
+        {
+            'metadata': metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'transform_table_name': 'test_transform_table',
+            'model_id': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'model_name': 'test_model_name',
+            'model_config': {'retention': '30'},
+        },
+        0.95,
+        '2024-01-01',
+        '',
+        document,
+    )
+
+    # Assert: repeat_last_prediction still runs, with metadata_id=None, instead of raising.
+    assert result is True
+    assert workflow_mock.execute_activity_method.call_count == 2
+    workflow_mock.execute_activity_method.assert_any_call(
+        Activities.repeat_last_prediction,
+        {
+            **metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'model': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'metadata_id': None,
+        },
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+
+
 @mark.asyncio
 @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
 async def test_path_flag_handler_unknown(workflow_mock, prediction_process):
@@ -791,6 +1015,30 @@ async def test_path_flag_handler_unknown(workflow_mock, prediction_process):
     workflow_mock.execute_child_workflow.assert_not_called()
 
 
+@mark.asyncio
+@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
+async def test_run_cleans_up_and_reraises_on_pipeline_failure(workflow_mock, prediction_process):
+    """A failure anywhere in the pipeline must still release the MinIO offload before the
+    workflow fails — cleanup is not conditional on success."""
+    prediction_process._run_prediction_pipeline = AsyncMock(side_effect=RuntimeError('boom'))
+    input_data = {
+        'metadata': metadata,
+        'data': {'some': 'data'},
+        'model_id': 1,
+        'model_name': 'test_model_name',
+    }
+
+    with raises(RuntimeError, match='boom'):
+        await prediction_process.run(input_data)
+
+    workflow_mock.execute_activity_method.assert_called_once_with(
+        Activities.cleanup_minio_objects_expired,
+        {**metadata, 'data': input_data['data']},
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+
+
 @mark.asyncio
 @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
 async def test_run_with_cleanup_prefixes(workflow_mock, prediction_process):
@@ -826,7 +1074,7 @@ async def test_run_with_cleanup_prefixes(workflow_mock, prediction_process):
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('continue', 0.95, 'ok'),
+        ('continue', 0.95, 'ok', None),  # input_gate
         ('continue', 0.95, ''),
         ('continue', 0.95, ''),
         ('continue', 0.95, ''),

tests/laborious/workflows/test_minimal_retrain.py

@@ -1,6 +1,7 @@
 from unittest.mock import ANY, AsyncMock, call, patch
 
-from pytest import fixture, mark
+from pytest import fixture, mark, raises
+from temporalio.exceptions import ApplicationError
 
 from laborious.activities.activities import Activities
 from laborious.workflows.minimal_retrain import MinimalRetrain
@@ -192,11 +193,12 @@ async def test_run_storage_fail(workflow_mock: AsyncMock, minimal_retrain: Minim
         ]
     )
 
-    from pytest import raises
-
-    with raises(ValueError, match='No data returned from query'):
+    with raises(ApplicationError, match='No data returned from query') as exc_info:
         await minimal_retrain.run(input_data)
 
+    assert exc_info.value.non_retryable is True
+    assert exc_info.value.type == 'NoDataReturnedError'
+
     workflow_mock.execute_activity_method.assert_called_once_with(
         Activities.load_query_with_minio_offload,
         {