fix(worker): re-exec after install_runtime so imports match the runtime (SIENTIAPDE-2214)

The worker imports sklearn (metrics, evidently) before
PluginStore.install_runtime replaces it on disk with the runtime's pinned
version, the same order that breaks every sklearn prediction in laborious.
After installing the runtime, the worker now replaces its process once
(SIENTIA_RUNTIME_READY=1 skips the second install), so every module is
loaded from the runtime's packages.

.github/workflows/deploy.yml

@@ -20,7 +20,8 @@ jobs:
       branch_name: ${{ github.event.pull_request.head.ref }}
       project_name: 'model-manager'
       chart_name: 'model-manager'
-      env_list_path: '.global.env'
+      env_list_path: '[".global.env"]'
+      update_version_in: '["pyproject"]'
       create_tarball: ${{ github.event_name == 'pull_request' }}
       mirror: true
       bump_chart: true

.github/workflows/quality-gate.yml

@@ -5,7 +5,6 @@ on:
     branches:
       - main
       - 'release/**'
-      - 'feature/**'
   pull_request:
     branches:
       - main

model_manager/utils/repository/data_manager_repository.py

@@ -527,6 +527,32 @@ class DataManagerRepository(SientiaMonitoring):
             self.error(error_msg, metadata)
             raise OSError(error_msg) from e
 
+    def _save_data_csv(
+        self, frame: pd.DataFrame, file_path: str, metadata: dict[str, Any] | None = None
+    ) -> None:
+        """
+        Write a train/test frame to CSV with its time index as a ``timestamp`` column.
+
+        laborious drift reads ``test_data.csv`` from MLflow as its reference and needs
+        ``timestamp`` (SIENTIAPDE-2201). If a column already has that name, the frame is
+        written without the index and that model's drift keeps falling back to the target
+        window: training never fails over drift.
+
+        Args:
+            frame: Frame indexed by the UTC training timestamps.
+            file_path: Destination CSV path.
+            metadata: Workflow metadata for logging.
+        """
+        if 'timestamp' in frame.columns:
+            self.warning(
+                f"Column 'timestamp' already exists; {path.basename(file_path)} saved without "
+                'the time column (drift reference falls back to the target window)',
+                metadata,
+            )
+            frame.to_csv(file_path, index=False)
+            return
+        frame.to_csv(file_path, index_label='timestamp')
+
     def generate_report(
         self, data: TrainModelResult, metadata: dict[str, Any] | None = None
     ) -> TrainModelResult:
@@ -598,11 +624,11 @@ class DataManagerRepository(SientiaMonitoring):
         # Save training / validation CSVs using the same frames as the report (includes
         # literal `target` alias for Evidently, plus predictions and float-cast features).
         data.train_data_path = path.join(data.run_dir, 'train_data.csv')
-        reference_data_float.to_csv(data.train_data_path, index=False)
+        self._save_data_csv(reference_data_float, data.train_data_path, metadata)
 
         # Save test data CSV
         data.test_data_path = path.join(data.run_dir, 'test_data.csv')
-        current_data_float.to_csv(data.test_data_path, index=False)
+        self._save_data_csv(current_data_float, data.test_data_path, metadata)
 
         # Save equation as JSON
         if data.equation is not None and data.params.model_type == 'linear_regression':

model_manager/worker/worker.py

@@ -84,9 +84,11 @@ async def main():
     1. Sets up logging and metadata
     2. Starts Prometheus metrics server
     3. Initializes notification handler
-    4. Creates and configures activities
-    5. Starts Temporal client and workers
-    6. Manages worker lifecycle and graceful shutdown
+    4. Installs the PluginStore runtime and re-executes itself once (`SIENTIA_RUNTIME_READY=1`)
+       so every import loads the runtime's package versions
+    5. Creates and configures activities
+    6. Starts Temporal client and workers
+    7. Manages worker lifecycle and graceful shutdown
 
     The function runs indefinitely until interrupted or an error occurs.
     On error, it performs cleanup and exits with a non-zero status code.
@@ -144,7 +146,12 @@ async def main():
         metrics_controller=metrics_controller,
     )
 
-    await plugin_store.install_runtime(runtime_name=runtime)
+    if os.environ.get('SIENTIA_RUNTIME_READY') != '1':
+        await plugin_store.install_runtime(runtime_name=runtime)
+        os.environ['SIENTIA_RUNTIME_READY'] = '1'
+        logger.custom_info(f'Runtime {runtime} installed, re-executing worker', metadata)
+        # ponytail: fresh interpreter so loaded modules match the runtime on disk (SIENTIAPDE-2214)
+        os.execv(sys.executable, [sys.executable, *sys.orig_argv[1:]])  # noqa: S606
 
     activities = Activities(
         postgres_config=build_postgres_config(),

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "model-manager"
-version = "2.0.1"
+version = "2.0.3"
 description = "Sientia DataOps Model Manager - ML Model Orchestration System"
 readme = "README.md"
 requires-python = ">=3.11"

tests/utils/repository/test_data_manager_repository.py

@@ -514,6 +514,58 @@ def test_generate_report_adds_target_alias_for_reports(tmp_path):
         assert test_csv['target'].equals(test_csv['t'])
 
 
+_UTC_INDEX = pd.DatetimeIndex(['2024-01-01 02:00', '2024-01-01 01:00'], tz='UTC')
+
+
+def _dated_result(**train_columns: list[float]) -> TrainModelResult:
+    """Frames indexed by UTC time, like prepare_training_data + predict produce them."""
+    columns = train_columns or {'v1': [1.0, 2.0], 't': [1.0, 2.0]}
+    return TrainModelResult(
+        params=TrainModelParams.from_dict(_minimal_dict_for_prepare()),
+        train_data=pd.DataFrame(columns, index=_UTC_INDEX),
+        val_data=pd.DataFrame(columns, index=_UTC_INDEX),
+        y_train_pred=pd.DataFrame({'t': [1.0, 2.0]}, index=_UTC_INDEX),
+        y_pred=pd.DataFrame({'t': [1.0, 2.0]}, index=_UTC_INDEX),
+        run_name='testrun',
+    )
+
+
+def _generate_report(repo, tmr: TrainModelResult, tmp_path) -> TrainModelResult:
+    with (
+        patch.object(repo, '_get_reports_directory', return_value=str(tmp_path)),
+        patch('model_manager.utils.repository.data_manager_repository.Reports'),
+    ):
+        return repo.generate_report(tmr, {})
+
+
+def test_generate_report_csvs_keep_the_time_index_as_timestamp(tmp_path):
+    """laborious drift reads test_data.csv as its reference and needs ``timestamp``."""
+    repo = dmr.DataManagerRepository(MagicMock())
+
+    out = _generate_report(repo, _dated_result(), tmp_path)
+
+    for csv_path in (out.train_data_path, out.test_data_path):
+        csv = pd.read_csv(csv_path)
+        assert list(csv.columns) == ['timestamp', 'prediction', 'v1', 't', 'target']
+        assert pd.to_datetime(csv['timestamp']).tolist() == list(_UTC_INDEX)
+
+
+def test_generate_report_csvs_skip_timestamp_when_a_column_already_has_the_name(tmp_path):
+    """A variable literally named ``timestamp`` wins; training must not fail over drift."""
+    repo = dmr.DataManagerRepository(MagicMock())
+    tmr = _dated_result(timestamp=[5.0, 6.0], v1=[1.0, 2.0], t=[1.0, 2.0])
+
+    with patch.object(repo, 'warning') as warning:
+        out = _generate_report(repo, tmr, tmp_path)
+
+    for csv_path in (out.train_data_path, out.test_data_path):
+        csv = pd.read_csv(csv_path)
+        assert list(csv.columns) == ['prediction', 'timestamp', 'v1', 't', 'target']
+        assert csv['timestamp'].tolist() == [5.0, 6.0]
+    assert warning.call_count == 2
+    assert "Column 'timestamp' already exists" in warning.call_args.args[0]
+
+
 def test_generate_report_skips_equation_file_when_not_linear(tmp_path):
     repo = dmr.DataManagerRepository(MagicMock())
     p = TrainModelParams.from_dict(_minimal_dict_for_prepare())

tests/worker/test_worker.py

@@ -30,6 +30,16 @@ def mock_env_vars():
         yield env_vars
 
 
+@pytest.fixture(autouse=True)
+def runtime_ready(monkeypatch):
+    """Boot as the re-executed process by default; the real os.execv must never run in tests."""
+    monkeypatch.setenv('SIENTIA_RUNTIME_READY', '1')
+
+
+class _StopBootError(Exception):
+    """Stops main() at a chosen point, e.g. where os.execv would replace the process."""
+
+
 @pytest.fixture
 def mock_logger():
     """Create a mock logger."""
@@ -844,6 +854,119 @@ async def test_main_missing_runtime_uses_single_fallback(
     assert mock_prepare_worker.call_args_list[1].kwargs['runtime'] == 'single'
 
 
+_PLUGIN_STORE_CONFIG = {
+    'base_url': 'http://sientia-plugin-store.svc.cluster.local',
+    'owner': 'sientia',
+    'repo': 'model-library-store',
+    'branch': 'main',
+    'username': 'gitea-user',
+    'password': 'gitea-password',
+    'pypi_index_url': 'http://library-distribution-server.library.svc.cluster.local:5000',
+    'pypi_username': None,
+    'pypi_password': None,
+    'cache_ttl_seconds': None,
+}
+
+
+@pytest.mark.asyncio
+@patch('model_manager.worker.worker.RUNTIME', 'basic')
+# Stops boot at Activities, so the red phase does not try to reach a real Temporal server.
+@patch('model_manager.worker.worker.Activities', side_effect=_StopBootError)
+@patch('model_manager.worker.worker.NotificationHandler')
+@patch('model_manager.worker.worker.PluginStore')
+@patch('model_manager.worker.worker.build_plugin_store_config', return_value=_PLUGIN_STORE_CONFIG)
+@patch('model_manager.worker.worker.build_postgres_config', return_value={})
+@patch('model_manager.worker.worker.build_mlflow_config', return_value={})
+@patch('model_manager.worker.worker.build_minio_config', return_value={})
+@patch(
+    'model_manager.worker.worker.build_mongodb_config',
+    return_value={'connection_string': 'mongodb://test', 'database_name': 'db', 'uri': 'u'},
+)
+@patch('model_manager.worker.worker.get_logger')
+@patch('model_manager.worker.worker.start_prometheus_server')
+@patch('model_manager.worker.worker.ensure_runtime_directories')
+async def test_main_reexecs_after_installing_runtime(
+    _mock_dirs,
+    _mock_prometheus,
+    _mock_get_logger,
+    _mock_mongo,
+    _mock_minio,
+    _mock_mlflow,
+    _mock_postgres,
+    _mock_ps_config,
+    mock_plugin_store_class,
+    _mock_notification,
+    mock_activities_class,
+    monkeypatch,
+):
+    """First boot installs the runtime, then replaces the process so imports see it on disk."""
+    from model_manager.worker import worker as worker_module
+
+    monkeypatch.setenv('SIENTIA_RUNTIME_READY', '0')
+    mock_plugin_store_class.return_value.install_runtime = AsyncMock(return_value=None)
+
+    with (
+        patch('model_manager.worker.worker.os.execv', side_effect=_StopBootError) as m_execv,
+        patch.object(sys, 'orig_argv', ['python', '-m', 'model_manager.worker.worker']),
+        pytest.raises(_StopBootError),
+    ):
+        await worker_module.main()
+
+    mock_plugin_store_class.return_value.install_runtime.assert_awaited_once_with(
+        runtime_name='basic'
+    )
+    assert os.environ['SIENTIA_RUNTIME_READY'] == '1'
+    m_execv.assert_called_once_with(
+        sys.executable, [sys.executable, '-m', 'model_manager.worker.worker']
+    )
+    mock_activities_class.assert_not_called()
+
+
+@pytest.mark.asyncio
+@patch('model_manager.worker.worker.RUNTIME', 'basic')
+@patch('model_manager.worker.worker.Activities', side_effect=_StopBootError)
+@patch('model_manager.worker.worker.NotificationHandler')
+@patch('model_manager.worker.worker.PluginStore')
+@patch('model_manager.worker.worker.build_plugin_store_config', return_value=_PLUGIN_STORE_CONFIG)
+@patch('model_manager.worker.worker.build_postgres_config', return_value={})
+@patch('model_manager.worker.worker.build_mlflow_config', return_value={})
+@patch('model_manager.worker.worker.build_minio_config', return_value={})
+@patch(
+    'model_manager.worker.worker.build_mongodb_config',
+    return_value={'connection_string': 'mongodb://test', 'database_name': 'db', 'uri': 'u'},
+)
+@patch('model_manager.worker.worker.get_logger')
+@patch('model_manager.worker.worker.start_prometheus_server')
+@patch('model_manager.worker.worker.ensure_runtime_directories')
+async def test_main_skips_install_when_runtime_ready(
+    _mock_dirs,
+    _mock_prometheus,
+    _mock_get_logger,
+    _mock_mongo,
+    _mock_minio,
+    _mock_mlflow,
+    _mock_postgres,
+    _mock_ps_config,
+    mock_plugin_store_class,
+    _mock_notification,
+    mock_activities_class,
+):
+    """The re-executed process skips the install and the exec and goes on booting."""
+    from model_manager.worker import worker as worker_module
+
+    mock_plugin_store_class.return_value.install_runtime = AsyncMock(return_value=None)
+
+    with (
+        patch('model_manager.worker.worker.os.execv') as m_execv,
+        pytest.raises(_StopBootError),
+    ):
+        await worker_module.main()
+
+    mock_plugin_store_class.return_value.install_runtime.assert_not_awaited()
+    m_execv.assert_not_called()
+    mock_activities_class.assert_called_once()
+
+
 @patch('model_manager.worker.worker.asyncio.run')
 def test_main_entrypoint(mock_asyncio_run):
     """Test the __main__ entrypoint."""