From e05ec1c0a93578634b76c8cc8d312f6ae71c868b Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 7 Oct 2026 15:54:44 -0300 Subject: [PATCH] feat(simple_metrics): update adaptive state after each run, empty windows included (SIENTIAPDE-2191) --- e2e/conftest.py | 1 + laborious/worker/worker.py | 1 + laborious/workflows/simple_metrics.py | 90 +++++++++++------- tests/laborious/worker/test_worker.py | 6 ++ .../workflows/test_simple_metrics.py | 94 ++++++++++++++++++- 5 files changed, 152 insertions(+), 40 deletions(-) diff --git a/e2e/conftest.py b/e2e/conftest.py index 807a5ce..a108127 100644 --- a/e2e/conftest.py +++ b/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, ] diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index b44d829..c33fe9b 100644 --- a/laborious/worker/worker.py +++ b/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', diff --git a/laborious/workflows/simple_metrics.py b/laborious/workflows/simple_metrics.py index fad9a89..279d77e 100644 --- a/laborious/workflows/simple_metrics.py +++ b/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). - enough_samples = not min_samples or len(target_data) >= min_samples - if not enough_samples: + # window must neither alert nor escalate (SIENTIAPDE-1995). + enough_samples = bool(target_data) and (not min_samples or len(target_data) >= min_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( - Activities.calculate_simple_metrics, - { - **metadata, - 'model_id': model_id, - 'target_data': target_data, - 'metrics': input_data.get('metrics', ['rmse', 'mse', 'mae', 'r2']), - 'interval_minutes': interval_minutes, - 'model_type': model_type, - 'thresholds': thresholds if enough_samples else None, - }, - retry_policy=retry_policy, - start_to_close_timeout=timedelta(seconds=300), - ) + simple_metrics = None + if target_data: + simple_metrics = await workflow.execute_local_activity_method( + Activities.calculate_simple_metrics, + { + **metadata, + 'model_id': model_id, + 'target_data': target_data, + 'metrics': input_data.get('metrics', ['rmse', 'mse', 'mae', 'r2']), + 'interval_minutes': interval_minutes, + 'model_type': model_type, + 'thresholds': thresholds if enough_samples else None, + }, + 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( - 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, + try: + await workflow.execute_activity_method( + Activities.update_simple_metrics_adaptive_state, + { + **metadata, + # Only a valid breach escalates: an empty or under-sampled window is sent as + # no rows, which counts as a healthy run (spec decision 8). + 'simple_metrics': (simple_metrics or []) if enough_samples else [], + 'thresholds': thresholds, + 'interval_minutes': interval_minutes, }, - }, - retry_policy=retry_policy, - start_to_close_timeout=timedelta(seconds=300), - ) + retry_policy=retry_policy, + start_to_close_timeout=timedelta(minutes=1), + ) + 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}') diff --git a/tests/laborious/worker/test_worker.py b/tests/laborious/worker/test_worker.py index a4c1898..c9ded87 100644 --- a/tests/laborious/worker/test_worker.py +++ b/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 diff --git a/tests/laborious/workflows/test_simple_metrics.py b/tests/laborious/workflows/test_simple_metrics.py index bafd007..dc9b6bf 100644 --- a/tests/laborious/workflows/test_simple_metrics.py +++ b/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. - assert workflow_mock.execute_activity_method.call_count == 2 + # Persist, don't alert: load, export, adaptive state. + 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()