SIENTIAPDE-1646

SIENTIAPDE-1646 Merge origin/main into release/SIENTIAPDE-1646

Bring main up to date while keeping this branch's sync execution, PluginStore
runtime and sientia_model (SientiaMLflowRepository/SientiaModel) structure.

Resolution decisions:
- OPC: keep the sync asyncua.sync implementation; port main's server_uri
  validation, e2e race fix (lock held until released) and the new unit tests
  converted to sync.
- Model import workflow (.sientia bundles), joblib/transform flavors and the
  local MLFlowRepository (model_repository.py) are dropped, with their tests,
  docs, metrics, env vars and connectors config.
- Worker: drift/simple_metrics stay pinned to the 'core' runtime; port main's
  HOSTNAME pod id, logger-based prometheus startup and sys.exit after finally.
- values.yaml removed, following main (SIENTIAPDE-1945).
- Deps: keep sientia_model; take main's fsspec/s3fs/botocore/boto3 pins and
  sientia_do>=1.12.2; drop import-only deps (pynacl, cryptography, pymongo,
  python-dotenv, pycaret, joblib, gcsfs).
- e2e: keep sientia_data schema and scenario input files.

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

tests/laborious/activities/test_api.py

@@ -218,6 +218,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(

tests/laborious/activities/test_opc.py

@@ -539,6 +539,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/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
 
@@ -8,11 +10,13 @@ from asyncua.crypto import security_policies
 from asyncua.ua.uaerrors import BadNodeIdUnknown, BadSessionIdInvalid
 from sientia_do.notifications.models import NotificationLevel
 
+from laborious import metrics
 from laborious.utils.repository.opc_repository import (
     OpcClientAlreadyExistsError,
     OpcClientNotInitializedError,
     OpcRepository,
     OpcSessionAlreadyConnectedError,
+    _opc_authentication_token_str,
     is_reconnectable_opcua_bad,
 )
 
@@ -86,6 +90,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 +150,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 +205,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'
@@ -423,6 +505,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 +623,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 +750,21 @@ 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

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

tests/laborious/workflows/subworkflows/test_prediction_process.py

@@ -791,6 +791,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):

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,
         {