Merge remote-tracking branch 'origin/main' into fix/SIENTIAPDE-2215

fix(worker): pin the runtime versions and refuse to boot on a version swap (SIENTIAPDE-2214)

CLAUDE.md

@@ -45,7 +45,7 @@ Coverage gate in CI is 80% minimum; this repo targets 100% in practice — don't
 ## Architecture
 
 ### Temporal layering
-- **Worker** (`model_manager/worker/worker.py`): entrypoint. Builds config from env, connects Postgres/MLFlow/MinIO/Plugin Store, installs the `RUNTIME` model runtime via `PluginStore.install_runtime`, starts two Temporal workers (one per task queue) plus Prometheus metrics server and the cleanup schedule. All heavy imports are wrapped in `workflow.unsafe.imports_passed_through()` — required because Temporal workflow sandboxing forbids non-deterministic imports at module scope outside that block.
+- **Worker** (`model_manager/worker/worker.py`): entrypoint. Builds config from env, connects Postgres/MLFlow/MinIO/Plugin Store, installs the `RUNTIME` model runtime via `PluginStore.install_runtime` (boot fails if that changes the version of an already-imported package, so the `requirements.txt` pins must match the factory runtimes; `app_up` goes to 1 only after this check), starts two Temporal workers (one per task queue) plus Prometheus metrics server and the cleanup schedule. All heavy imports are wrapped in `workflow.unsafe.imports_passed_through()` — required because Temporal workflow sandboxing forbids non-deterministic imports at module scope outside that block.
 - **Workflows** (`model_manager/workflows/`): pure orchestration, no I/O beyond `execute_activity_method` calls.
   - `train_model.py` (`TrainModel`, workflow name `train_model`): validate `experiment_run_id` → `load_model_metadata` → `validate_train_params` → update DB status `ORCHESTRATOR_WAITING_PROC` → `train_model` activity → update DB status `TRAINING_SUCCESS`/`TRAINING_ERROR` → `cleanup_resources` (always runs in `finally` if a `run_dir` exists, even on failure). Uses distinct `RetryPolicy`s per operation class: `network_retry_policy` (transient I/O), `no_retry_policy` (training/validation — data errors are permanent, and retrying would duplicate MLflow writes), `database_retry_policy` (Postgres). Timeouts are env-driven (`TIMEOUT_*`), extend `TIMEOUT_TRAIN_MODEL` instead of adding retries for slow training.
   - `cleanup_files.py` (`CleanupFiles`): scheduled (`cleanup_schedule.py`, default daily at midnight UTC via `CLEANUP_CRON`) deletion of stale local temp dirs matching `name_YYYYMMDD_HHMMSS_microseconds`, older than `CLEANUP_RETENTION_HOURS`.

README.md

@@ -149,6 +149,7 @@ The Model Manager system uses a Temporal-based workflow architecture with clear
   - Prometheus metrics server initialization
   - Notification handler setup and configuration
   - **Runtime Installation**: Automatically installs the required model runtime from the Plugin Store
+  - **Runtime version guard**: boot fails if the install changes the version of a package the worker already imported, so the `requirements.txt` pins must match `model-plugin-factory` `runtime/basic.yaml` and `runtime/xgboost.yaml`; `app_up` goes to 1 only after this check
 - **Key Features**:
   - Automatic scaling with `PollerBehaviorAutoscaling`
   - Health check endpoints for Kubernetes liveness/readiness probes
@@ -1249,6 +1250,10 @@ sientia-dataops-model-manager/
    - Check training parameter validation errors
    - Verify input data format and required fields
 
+5. **Worker exits at boot with `Runtime ... changed packages already imported by the worker`**
+   - A runtime pin in `model-plugin-factory` `runtime/*.yaml` no longer matches `requirements.txt`; the message lists each package as `old -> new`
+   - Set the runtime version in `requirements.txt` and `requirements-local.txt` here and in `sientia-dataops-laborious`, then redeploy
+
 ### Debug Mode
 
 Enable debug logging by setting the log level:

model_manager/utils/repository/data_manager_repository.py

@@ -357,7 +357,8 @@ class DataManagerRepository(SientiaMonitoring):
         This helper mirrors the previous TrainingRepository.after_train_calculation
         behavior, assuming that predictions (y_pred/y_train_pred) are already on the
         correct scale for metric calculation (any scaling is handled inside the
-        model wrapper).
+        model wrapper). Pairs with a NaN target or a NaN prediction are left out of
+        mse/mae/r2. With a single valid pair left, r2_val is NaN (R² needs 2 samples).
 
         Args:
             tmr: Training result containing:
@@ -368,6 +369,10 @@ class DataManagerRepository(SientiaMonitoring):
 
         Return:
             TrainModelResult: Same object with mse_val, mae_val and r2_val set.
+
+        Raises:
+            ValueError: If y_pred is not set, val_data and y_pred share no indices,
+                or no pair has both a valid target and a valid prediction.
         """
         if tmr.y_pred is None:
             raise ValueError('y_pred must be set before computing regression metrics')
@@ -405,6 +410,24 @@ class DataManagerRepository(SientiaMonitoring):
         y_true_val = y_true_val.loc[common_index]
         y_pred_val = y_pred_val.loc[common_index]
 
+        # NaN target, or NaN prediction for rows with NaN features, carries no error signal
+        # (SIENTIAPDE-2215).
+        valid = y_true_val.notna() & y_pred_val.notna()
+        if not valid.any():
+            raise ValueError(
+                'No validation rows with both target and prediction: all '
+                f'{len(valid)} overlapping rows have NaN'
+            )
+        if not valid.all():
+            n_valid = int(valid.sum())
+            message = (
+                f'Ignoring {len(valid) - n_valid} of {len(valid)} validation row(s) with NaN '
+                f'target or prediction; regression metrics use the remaining {n_valid}'
+            )
+            self.warning(message, metadata)
+        y_true_val = y_true_val[valid]
+        y_pred_val = y_pred_val[valid]
+
         # Metrics helpers already round to 2 decimals.
         tmr.mse_val = mse(y_true_val, y_pred_val)
         tmr.mae_val = mae(y_true_val, y_pred_val)

model_manager/worker/worker.py

@@ -32,6 +32,7 @@ from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig
 
 with workflow.unsafe.imports_passed_through():
     import asyncio
+    import importlib.metadata
     import os
     import sys
 
@@ -76,6 +77,36 @@ def _get_runtime(runtime: str | None) -> str:
     return normalized_runtime or 'single'
 
 
+def _loaded_distribution_versions() -> dict[str, str]:
+    """Version on disk of every installed distribution whose top-level module is imported."""
+    # Iterate a copy: another thread importing while we iterate would raise RuntimeError.
+    loaded = {name.partition('.')[0] for name in sys.modules.copy()}
+    return {
+        dist: importlib.metadata.version(dist)
+        for module, dists in importlib.metadata.packages_distributions().items()
+        if module in loaded
+        for dist in dists
+    }
+
+
+def _runtime_changed_loaded_packages(before: dict[str, str]) -> dict[str, str]:
+    """
+    Packages that were imported before ``install_runtime`` and whose version changed on disk.
+
+    The process keeps the old code in memory, so a later lazy import mixes two versions
+    (SIENTIAPDE-2214: sklearn 1.9.1 in memory, 1.8.0 on disk, ``_cyutility`` import error).
+    """
+    changed = {}
+    for dist, old in before.items():
+        try:
+            new = importlib.metadata.version(dist)
+        except importlib.metadata.PackageNotFoundError:
+            new = 'removed'
+        if new != old:
+            changed[dist] = f'{old} -> {new}'
+    return changed
+
+
 async def main():
     """
     Main entry point for the Model Manager worker application.
@@ -83,7 +114,9 @@ async def main():
     This function initializes and starts all components of the worker:
     1. Sets up logging and metadata
     2. Starts Prometheus metrics server
-    3. Initializes notification handler
+    3. Initializes notification handler, then installs the runtime; exits if the install
+       changed the version of a package the worker already imported, and only then sets
+       ``app_up`` (what the probes check) to 1
     4. Creates and configures activities
     5. Starts Temporal client and workers
     6. Manages worker lifecycle and graceful shutdown
@@ -144,8 +177,23 @@ async def main():
         metrics_controller=metrics_controller,
     )
 
+    loaded_before_install = _loaded_distribution_versions()
     await plugin_store.install_runtime(runtime_name=runtime)
 
+    changed = _runtime_changed_loaded_packages(loaded_before_install)
+    if changed:
+        logger.custom_critical(
+            f'Runtime {runtime} changed packages already imported by the worker: {changed}. '
+            'Pin them in requirements.txt to the runtime versions.',
+            metadata,
+        )
+        metrics.APP_UP.labels(pod_id=POD_ID).set(0)
+        sys.exit(1)
+
+    # The probes only check app_up, so it goes to 1 after the guard: a pod that would fail it
+    # never turns Ready, and a rollout keeps the old pod.
+    metrics.APP_UP.labels(pod_id=POD_ID).set(1)
+
     activities = Activities(
         postgres_config=build_postgres_config(),
         mlflow_config=build_mlflow_config(),
@@ -232,8 +280,8 @@ def start_prometheus_server(logger: SientiaLogger, metadata: dict[str, str | Non
     """
     Starts the Prometheus metrics server for monitoring and observability.
 
-    This function initializes the Prometheus HTTP server on the configured port
-    and sets the application health metric to indicate the service is running.
+    This function initializes the Prometheus HTTP server on the configured port.
+    ``main`` sets ``app_up`` to 1 later, once the runtime install passed the version guard.
 
     The server exposes metrics that can be scraped by Prometheus for monitoring
     the health and performance of the Model Manager worker.
@@ -249,7 +297,6 @@ def start_prometheus_server(logger: SientiaLogger, metadata: dict[str, str | Non
         port = int(os.getenv('HTTP_METRICS_PORT', 9090))
         start_http_server(port)
         logger.custom_info(f'Prometheus server initialized on port {port}.', metadata)
-        metrics.APP_UP.labels(pod_id=POD_ID).set(1)  # Mark app as UP
     except Exception as e:  # noqa: BLE001
         logger.custom_critical(f'Failed to start Prometheus server: {e}', metadata)
         os._exit(1)

pyproject.toml

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

requirements-local.txt

@@ -8,4 +8,15 @@ git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.14.0
 prometheus-client==0.23.1
 beautifulsoup4==4.12.3
 evidently==0.6.7
-jsonschema==4.26.0
+jsonschema==4.26.0
+# Same versions as the plugin runtimes (model-plugin-factory runtime/basic.yaml and xgboost.yaml).
+# The worker imports these before install_runtime; if the runtime installed a different version,
+# the old code would stay in memory, so the worker refuses to boot (SIENTIAPDE-2214).
+# Changing a runtime pin means changing it here and in requirements.txt too.
+mlflow==3.8.1
+pathspec==1.0.4
+pandas==2.3.3
+cloudpickle==3.1.2
+scikit-learn==1.8.0
+joblib==1.5.3
+rich==14.3.2

requirements.txt

@@ -4,8 +4,19 @@ sqlalchemy==2.0.50
 boto3==1.42.70
 botocore==1.42.70
 sientia_do>=1.12.2
-sientia_model>=0.8.1
+sientia_model>=0.14.0
 prometheus-client==0.23.1
 beautifulsoup4==4.12.3
 evidently==0.6.7
 jsonschema==4.26.0
+# Same versions as the plugin runtimes (model-plugin-factory runtime/basic.yaml and xgboost.yaml).
+# The worker imports these before install_runtime; if the runtime installed a different version,
+# the old code would stay in memory, so the worker refuses to boot (SIENTIAPDE-2214).
+# Changing a runtime pin means changing it here and in requirements-local.txt too.
+mlflow==3.8.1
+pathspec==1.0.4
+pandas==2.3.3
+cloudpickle==3.1.2
+scikit-learn==1.8.0
+joblib==1.5.3
+rich==14.3.2

tests/utils/repository/test_data_manager_repository.py

@@ -309,6 +309,46 @@ def test_compute_regression_metrics_no_overlap():
         repo.compute_regression_metrics(tmr, MagicMock())
 
 
+def test_compute_regression_metrics_ignores_nan_pairs():
+    """NaN target or NaN prediction rows are left out of mse/mae/r2 (SIENTIAPDE-2215)."""
+    repo = dmr.DataManagerRepository(MagicMock())
+    p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
+    idx = pd.Index([0, 1, 2, 3])
+    tmr = TrainModelResult(
+        params=p,
+        train_data=pd.DataFrame({'t': [1.0, 2.0, 3.0, 4.0]}, index=idx),
+        val_data=pd.DataFrame({'t': [1.0, np.nan, 3.0, 4.0]}, index=idx),
+        y_pred=pd.DataFrame({'p': [1.0, 2.0, np.nan, 4.0]}, index=idx),
+    )
+
+    wrapper = MagicMock()
+    wrapper.model = object()  # no sklearn .regr -> equation extraction is skipped
+    with patch.object(repo, 'warning') as warn:
+        out = repo.compute_regression_metrics(tmr, wrapper)
+
+    assert out.mse_val == pytest.approx(0.0)
+    assert out.mae_val == pytest.approx(0.0)
+    assert out.r2_val == pytest.approx(1.0)
+    warn.assert_called_once()
+    assert 'Ignoring 2 of 4' in warn.call_args.args[0]
+
+
+def test_compute_regression_metrics_all_nan_pairs_raises():
+    repo = dmr.DataManagerRepository(MagicMock())
+    p = TrainModelParams.from_dict(_minimal_dict_for_prepare())
+    idx = pd.Index([0, 1])
+    tmr = TrainModelResult(
+        params=p,
+        train_data=pd.DataFrame({'t': [1.0, 2.0]}, index=idx),
+        val_data=pd.DataFrame({'t': [1.0, 2.0]}, index=idx),
+        y_pred=pd.DataFrame({'p': [np.nan, np.nan]}, index=idx),
+    )
+    wrapper = MagicMock()
+
+    with pytest.raises(ValueError, match='No validation rows with both target and prediction'):
+        repo.compute_regression_metrics(tmr, wrapper)
+
+
 def test_compute_regression_metrics_linear_equation():
     repo = dmr.DataManagerRepository(MagicMock())
     p = TrainModelParams.from_dict(_minimal_dict_for_prepare())

tests/worker/test_worker.py

@@ -3,7 +3,7 @@
 import asyncio
 import os
 import sys
-from unittest.mock import AsyncMock, Mock, patch
+from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import pytest
 
@@ -132,9 +132,8 @@ def test_start_prometheus_server_success(
     # Verify HTTP server started
     mock_start_http_server.assert_called_once_with(9090)
 
-    # Verify APP_UP metric was set to 1
-    mock_metrics.APP_UP.labels.assert_called_once_with(pod_id='test-pod-123')
-    mock_app_up.set.assert_called_once_with(1)
+    # app_up goes to 1 in main(), after the runtime version guard.
+    mock_app_up.set.assert_not_called()
     mock_logger.custom_info.assert_called_once()
 
 
@@ -284,7 +283,8 @@ async def test_main_successful_startup(
     # Verify cleanup was performed
     mock_notification_handler.shutdown.assert_called_once()
     mock_activities.shutdown.assert_called_once()
-    mock_app_up.set.assert_called_with(0)
+    # Ready after the guard, down again on shutdown.
+    assert [c.args[0] for c in mock_app_up.set.call_args_list] == [1, 0]
 
 
 @pytest.mark.asyncio
@@ -848,12 +848,10 @@ async def test_main_missing_runtime_uses_single_fallback(
 def test_main_entrypoint(mock_asyncio_run):
     """Test the __main__ entrypoint."""
     # Import and execute the main block
-    with patch.object(sys, 'argv', ['worker.py']):
-        import model_manager.worker.worker as worker_module
-
-        # Simulate running the module
-        worker_module.main = AsyncMock()
+    import model_manager.worker.worker as worker_module
 
+    # patch.object restores main afterwards; assigning it leaked the mock into later tests.
+    with patch.object(sys, 'argv', ['worker.py']), patch.object(worker_module, 'main', AsyncMock()):
         # This would normally be called by asyncio.run(main())
         # We just verify the pattern is correct
         assert callable(worker_module.main)
@@ -905,3 +903,105 @@ def test_start_prometheus_server_prints_failure(
     # Agora o erro é logado via logger crítico
     mock_logger.custom_critical.assert_called_once()
     mock_exit.assert_called_once_with(1)
+
+
+def test_loaded_distribution_versions_reads_imported_packages():
+    """An imported module maps to its distribution's installed version."""
+    import importlib.metadata
+
+    from model_manager.worker.worker import _loaded_distribution_versions
+
+    assert _loaded_distribution_versions()['pytest'] == importlib.metadata.version('pytest')
+
+
+def test_runtime_changed_loaded_packages_reports_only_changes():
+    """Only distributions whose version on disk differs after the install are reported."""
+    import importlib.metadata
+
+    from model_manager.worker.worker import _runtime_changed_loaded_packages
+
+    on_disk = {'scikit-learn': '1.8.0', 'numpy': '1.26.4'}
+
+    def fake_version(dist):
+        if dist not in on_disk:
+            raise importlib.metadata.PackageNotFoundError(dist)
+        return on_disk[dist]
+
+    before = {'scikit-learn': '1.9.1', 'numpy': '1.26.4', 'gone': '1.0'}
+    with patch('model_manager.worker.worker.importlib.metadata.version', side_effect=fake_version):
+        changed = _runtime_changed_loaded_packages(before)
+
+    assert changed == {'scikit-learn': '1.9.1 -> 1.8.0', 'gone': '1.0 -> removed'}
+
+
+@pytest.mark.asyncio
+async def test_main_runtime_changed_loaded_package_exits(mock_env_vars, mock_logger):
+    """Boot stops before Activities when install_runtime changed an imported package."""
+    import model_manager.worker.worker as worker_module
+
+    plugin_store = Mock()
+    plugin_store.install_runtime = AsyncMock(return_value=None)
+    before = {'scikit-learn': '1.9.1'}
+
+    def snapshot():
+        # Taken after install_runtime, it would already see 1.8.0 and the guard would pass.
+        assert plugin_store.install_runtime.await_count == 0
+        return before
+
+    mongo = {'connection_string': 'mongodb://test', 'database_name': 'test_db', 'uri': 'x'}
+    with (
+        patch.object(worker_module, 'RUNTIME', 'basic'),
+        patch.object(worker_module, 'ensure_runtime_directories'),
+        patch.object(worker_module, 'metrics') as mock_metrics,
+        patch.object(worker_module, 'start_prometheus_server'),
+        patch.object(worker_module, 'get_logger', return_value=mock_logger),
+        patch.object(worker_module, 'build_mongodb_config', return_value=mongo),
+        patch.object(worker_module, 'NotificationHandler'),
+        patch.object(worker_module, 'build_plugin_store_config', return_value=MagicMock()),
+        patch.object(worker_module, 'PluginStore', return_value=plugin_store),
+        patch.object(worker_module, '_loaded_distribution_versions', side_effect=snapshot),
+        patch.object(
+            worker_module,
+            '_runtime_changed_loaded_packages',
+            return_value={'scikit-learn': '1.9.1 -> 1.8.0'},
+        ) as mock_changed,
+        patch.object(worker_module, 'Activities') as mock_activities_class,
+        patch.object(worker_module.sys, 'exit', side_effect=SystemExit(1)) as mock_exit,
+    ):
+        with pytest.raises(SystemExit):
+            await worker_module.main()
+
+    mock_exit.assert_called_once_with(1)
+    mock_activities_class.assert_not_called()
+    plugin_store.install_runtime.assert_awaited_once()
+    mock_changed.assert_called_once_with(before)
+    mock_metrics.APP_UP.labels.return_value.set.assert_called_once_with(0)
+    message = mock_logger.custom_critical.call_args.args[0]
+    assert "'scikit-learn': '1.9.1 -> 1.8.0'" in message
+
+
+def test_guard_detects_version_change_on_disk(tmp_path, monkeypatch):
+    """The real importlib.metadata path: a package swapped on disk after import is reported."""
+    import types
+
+    from model_manager.worker.worker import (
+        _loaded_distribution_versions,
+        _runtime_changed_loaded_packages,
+    )
+
+    info = tmp_path / 'fakepkg_2214-1.0.dist-info'
+    info.mkdir()
+    (info / 'METADATA').write_text('Metadata-Version: 2.1\nName: fakepkg-2214\nVersion: 1.0\n')
+    (info / 'top_level.txt').write_text('fakepkg_2214\n')
+    monkeypatch.syspath_prepend(str(tmp_path))
+    # The guard only reads sys.modules keys; setitem removes the entry again on teardown.
+    monkeypatch.setitem(sys.modules, 'fakepkg_2214', types.ModuleType('fakepkg_2214'))
+
+    before = _loaded_distribution_versions()
+    assert before['fakepkg-2214'] == '1.0'
+
+    # What pip does on an upgrade: the old dist-info goes away and a new one takes its place.
+    new_info = info.rename(tmp_path / 'fakepkg_2214-2.0.dist-info')
+    (new_info / 'METADATA').write_text('Metadata-Version: 2.1\nName: fakepkg-2214\nVersion: 2.0\n')
+
+    assert _runtime_changed_loaded_packages(before) == {'fakepkg-2214': '1.0 -> 2.0'}