fix(simple_metrics): ignore unparseable frequency_alert and malformed adaptive_state (SIENTIAPDE-2190)

tests/orchestrator/activities/test_formatters.py

@@ -1057,6 +1057,18 @@ def test_create_schedule_config_warns_on_frequency_alert_below_floor(formatters)
     assert kwargs['metadata']['schedule_name'] == 'm1-simple_metrics-60m'
 
 
+def test_create_schedule_config_warns_on_frequency_alert_below_floor_on_update(formatters):
+    input_data = _schedule_config_with_frequency_alert('30s')
+    input_data['current_schedule_config'] = {'laborious': {'m1-simple_metrics-60m': '2026-10-06'}}
+
+    result = formatters.create_schedule_config(input_data)
+
+    assert 'm1-simple_metrics-60m' in result['to_update']['laborious']
+    formatters.send_notification.assert_called_once()
+    kwargs = formatters.send_notification.call_args.kwargs
+    assert kwargs['notification_id'] == 'SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR'
+
+
 def test_create_schedule_config_no_warning_at_or_above_floor(formatters):
     formatters.create_schedule_config(_schedule_config_with_frequency_alert('5m'))
 

tests/orchestrator/utils/test_orchestrator_functions.py

@@ -158,6 +158,24 @@ def test_simple_metrics_frequency_alert_without_state_keeps_frequency():
     assert simple_metrics(_simple_metrics_config(frequency_alert='5m'))['frequency'] == '1h'
 
 
+def test_simple_metrics_escalated_unparseable_frequency_alert_keeps_frequency():
+    # An unparseable frequency would make the Temporal schedule create/update fail.
+    config = _simple_metrics_config(frequency_alert='5min', adaptive_state={'escalated': True})
+    assert simple_metrics(config)['frequency'] == '1h'
+
+
+def test_simple_metrics_escalated_below_floor_frequency_alert_is_applied():
+    # Decision 7: below the floor only warns, the value is kept.
+    config = _simple_metrics_config(frequency_alert='30s', adaptive_state={'escalated': True})
+    assert simple_metrics(config)['frequency'] == '30s'
+
+
+@mark.parametrize('adaptive_state', ['yes', {'escalated': 'false'}])
+def test_simple_metrics_malformed_adaptive_state_keeps_frequency(adaptive_state):
+    config = _simple_metrics_config(frequency_alert='5m', adaptive_state=adaptive_state)
+    assert simple_metrics(config)['frequency'] == '1h'
+
+
 def test_minimal_retrain():
     config = {
         'workflow_type': 'minimal_retrain',