feat(simple_metrics): update adaptive state after each run, empty windows included (SIENTIAPDE-2191)

e2e/conftest.py

@@ -445,6 +445,7 @@ def _simple_metrics_worker_activity_list(test_activities: Activities):
         test_activities.load_custom_query,
         test_activities.calculate_simple_metrics,
         test_activities.export_data_to_postgres,
+        test_activities.update_simple_metrics_adaptive_state,
     ]
 
 

laborious/worker/worker.py

@@ -217,6 +217,7 @@ async def main():
                 activities.load_custom_query,
                 activities.calculate_simple_metrics,
                 activities.export_data_to_postgres,
+                activities.update_simple_metrics_adaptive_state,
             ],
             logger=logger,
             runtime='core',

laborious/workflows/simple_metrics.py

@@ -71,48 +71,68 @@ class SimpleMetrics:
             start_to_close_timeout=timedelta(seconds=300),
         )
 
-        if not target_data:
-            return
-
         # Too few rows for a reliable value: metrics are still persisted, but a noisy short
-        # window must not alert (SIENTIAPDE-1995).
+        # window must neither alert nor escalate (SIENTIAPDE-1995).
-        enough_samples = not min_samples or len(target_data) >= min_samples
+        enough_samples = bool(target_data) and (not min_samples or len(target_data) >= min_samples)
-        if not enough_samples:
+        if target_data and not enough_samples:
             workflow.logger.info(
                 f'{len(target_data)} rows < simple_metrics_min_samples={min_samples}: '
                 'threshold alerting skipped'
             )
 
-        simple_metrics = await workflow.execute_local_activity_method(
+        simple_metrics = None
-            Activities.calculate_simple_metrics,
+        if target_data:
-            {
+            simple_metrics = await workflow.execute_local_activity_method(
-                **metadata,
+                Activities.calculate_simple_metrics,
-                'model_id': model_id,
+                {
-                'target_data': target_data,
+                    **metadata,
-                'metrics': input_data.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
+                    'model_id': model_id,
-                'interval_minutes': interval_minutes,
+                    'target_data': target_data,
-                'model_type': model_type,
+                    'metrics': input_data.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
-                'thresholds': thresholds if enough_samples else None,
+                    'interval_minutes': interval_minutes,
-            },
+                    'model_type': model_type,
-            retry_policy=retry_policy,
+                    'thresholds': thresholds if enough_samples else None,
-            start_to_close_timeout=timedelta(seconds=300),
+                },
-        )
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=300),
+            )
 
-        if not simple_metrics:
+        if simple_metrics:
+            await workflow.execute_activity_method(
+                Activities.export_data_to_postgres,
+                {
+                    **metadata,
+                    'data': simple_metrics,
+                    'schema': input_data['schema'],
+                    'table_name': input_data['target_table_name'],
+                    'timestamp_conversion': {
+                        'column': 'timestamp',
+                        'format': DATETIME_FORMAT_WITH_TZ,
+                    },
+                },
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=300),
+            )
+
+        if not thresholds:
+            # No thresholds = no breach possible, so skip the Mongo round trip.
             return
 
-        await workflow.execute_activity_method(
+        try:
-            Activities.export_data_to_postgres,
+            await workflow.execute_activity_method(
-            {
+                Activities.update_simple_metrics_adaptive_state,
-                **metadata,
+                {
-                'data': simple_metrics,
+                    **metadata,
-                'schema': input_data['schema'],
+                    # Only a valid breach escalates: an empty or under-sampled window is sent as
-                'table_name': input_data['target_table_name'],
+                    # no rows, which counts as a healthy run (spec decision 8).
-                'timestamp_conversion': {
+                    'simple_metrics': (simple_metrics or []) if enough_samples else [],
-                    'column': 'timestamp',
+                    'thresholds': thresholds,
-                    'format': DATETIME_FORMAT_WITH_TZ,
+                    'interval_minutes': interval_minutes,
                 },
-            },
+                retry_policy=retry_policy,
-            retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(minutes=1),
-            start_to_close_timeout=timedelta(seconds=300),
+            )
-        )
+        except Exception as e:
+            # Best-effort: metrics are already persisted. retry_policy has maximum_attempts=1, so
+            # a dispatch failure would otherwise fail the workflow (same as prediction_process).
+            workflow.logger.warning(f'Failed to update adaptive state, continuing anyway: {e}')

tests/laborious/worker/test_worker.py

@@ -16,6 +16,7 @@ def _build_fake_activities():
     inst.export_data_to_postgres = MagicMock()
     inst.load_custom_query = MagicMock()
     inst.calculate_simple_metrics = MagicMock()
+    inst.update_simple_metrics_adaptive_state = MagicMock()
     inst.get_reference_data = MagicMock()
     inst.calculate_drift = MagicMock()
     inst.request_predict = MagicMock()
@@ -188,6 +189,11 @@ async def test_main_success_exit_zero(monkeypatch):
     # 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'
+    # prepare_calls[1] is the SimpleMetrics worker (order: MinimalRetrain, SimpleMetrics, Drift, PredictionsBatch).
+    assert (
+        fake_activities.update_simple_metrics_adaptive_state
+        in prepare_calls[1].kwargs['activities']
+    )
 
 
 @mark.asyncio

tests/laborious/workflows/test_simple_metrics.py

@@ -245,7 +245,7 @@ _METRICS = [{'metric': 'rmse', 'value': 9.0}]
 @patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
 async def test_run_below_min_samples_drops_thresholds(workflow_mock, simple_metrics, n_rows):
     workflow_mock.logger = MagicMock()
-    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * n_rows, None])
+    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * n_rows, None, None])
     workflow_mock.execute_local_activity_method = AsyncMock(return_value=_METRICS)
 
     await simple_metrics.run(
@@ -254,8 +254,8 @@ async def test_run_below_min_samples_drops_thresholds(workflow_mock, simple_metr
 
     activity_input = workflow_mock.execute_local_activity_method.call_args.args[1]
     assert activity_input['thresholds'] is None
-    # Persist, don't alert.
+    # Persist, don't alert: load, export, adaptive state.
-    assert workflow_mock.execute_activity_method.call_count == 2
+    assert workflow_mock.execute_activity_method.call_count == 3
     export_call = workflow_mock.execute_activity_method.call_args_list[1]
     assert export_call.args[0] == Activities.export_data_to_postgres
     assert export_call.args[1]['data'] == _METRICS
@@ -265,7 +265,7 @@ async def test_run_below_min_samples_drops_thresholds(workflow_mock, simple_metr
 @patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
 async def test_run_at_min_samples_keeps_thresholds(workflow_mock, simple_metrics):
     workflow_mock.logger = MagicMock()
-    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 5, None])
+    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 5, None, None])
     workflow_mock.execute_local_activity_method = AsyncMock(return_value=_METRICS)
 
     await simple_metrics.run(
@@ -274,6 +274,43 @@ async def test_run_at_min_samples_keeps_thresholds(workflow_mock, simple_metrics
 
     activity_input = workflow_mock.execute_local_activity_method.call_args.args[1]
     assert activity_input['thresholds'] == {'rmse_max': 1.0}
+    assert workflow_mock.execute_activity_method.call_count == 3
+
+
+_BREACH_ROWS = [{'metric': 'rmse', 'value': 9.0}]
+_THRESHOLDS = {'rmse_max': 1.0}
+
+
+@mark.asyncio
+@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
+async def test_run_updates_adaptive_state_after_export(workflow_mock, simple_metrics):
+    workflow_mock.logger = MagicMock()
+    workflow_mock.execute_activity_method = AsyncMock(
+        side_effect=[[_ROW] * 3, None, {'escalated': True, 'healthy_streak': 0}]
+    )
+    workflow_mock.execute_local_activity_method = AsyncMock(return_value=_BREACH_ROWS)
+
+    await simple_metrics.run(_input(simple_metrics_thresholds=_THRESHOLDS))
+
+    calls = workflow_mock.execute_activity_method.call_args_list
+    assert calls[1].args[0] is Activities.export_data_to_postgres
+    assert calls[2].args[0] is Activities.update_simple_metrics_adaptive_state
+    assert calls[2].args[1] == {
+        **metadata,
+        'simple_metrics': _BREACH_ROWS,
+        'thresholds': _THRESHOLDS,
+        'interval_minutes': 60,
+    }
+
+
+@mark.asyncio
+@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
+async def test_run_without_thresholds_skips_adaptive_state(workflow_mock, simple_metrics):
+    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 3, None])
+    workflow_mock.execute_local_activity_method = AsyncMock(return_value=_BREACH_ROWS)
+
+    await simple_metrics.run(_input())
+
     assert workflow_mock.execute_activity_method.call_count == 2
 
 
@@ -284,7 +321,7 @@ async def test_run_at_min_samples_keeps_thresholds(workflow_mock, simple_metrics
 @patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
 async def test_run_guard_off_keeps_thresholds(workflow_mock, simple_metrics, min_samples, warns):
     workflow_mock.logger = MagicMock()
-    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 2, None])
+    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 2, None, None])
     workflow_mock.execute_local_activity_method = AsyncMock(return_value=_METRICS)
 
     await simple_metrics.run(
@@ -294,3 +331,50 @@ async def test_run_guard_off_keeps_thresholds(workflow_mock, simple_metrics, min
     activity_input = workflow_mock.execute_local_activity_method.call_args.args[1]
     assert activity_input['thresholds'] == {'rmse_max': 1.0}
     assert workflow_mock.logger.warning.call_count == int(warns)
+
+
+@mark.asyncio
+@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
+async def test_run_empty_window_counts_as_healthy_run(workflow_mock, simple_metrics):
+    workflow_mock.logger = MagicMock()
+    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[], None])
+    workflow_mock.execute_local_activity_method = AsyncMock()
+
+    await simple_metrics.run(_input(simple_metrics_thresholds=_THRESHOLDS))
+
+    workflow_mock.execute_local_activity_method.assert_not_called()
+    calls = workflow_mock.execute_activity_method.call_args_list
+    assert len(calls) == 2  # load + adaptive state, no export
+    assert calls[1].args[0] is Activities.update_simple_metrics_adaptive_state
+    assert calls[1].args[1]['simple_metrics'] == []
+
+
+@mark.asyncio
+@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
+async def test_run_below_min_samples_sends_no_rows_to_adaptive_state(workflow_mock, simple_metrics):
+    workflow_mock.logger = MagicMock()
+    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 2, None, None])
+    workflow_mock.execute_local_activity_method = AsyncMock(return_value=_BREACH_ROWS)
+
+    await simple_metrics.run(
+        _input(simple_metrics_thresholds=_THRESHOLDS, simple_metrics_min_samples=5)
+    )
+
+    calls = workflow_mock.execute_activity_method.call_args_list
+    assert calls[1].args[0] is Activities.export_data_to_postgres  # metrics still persisted
+    assert calls[2].args[0] is Activities.update_simple_metrics_adaptive_state
+    assert calls[2].args[1]['simple_metrics'] == []  # never a breach
+
+
+@mark.asyncio
+@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
+async def test_run_adaptive_state_failure_does_not_fail_workflow(workflow_mock, simple_metrics):
+    workflow_mock.logger = MagicMock()
+    workflow_mock.execute_activity_method = AsyncMock(
+        side_effect=[[_ROW] * 3, None, RuntimeError('dispatch failed')]
+    )
+    workflow_mock.execute_local_activity_method = AsyncMock(return_value=_BREACH_ROWS)
+
+    await simple_metrics.run(_input(simple_metrics_thresholds=_THRESHOLDS))
+
+    workflow_mock.logger.warning.assert_called_once()