From 40f8d8174be524b0e2fbcee8c2a4cbb8b2302de9 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 7 Oct 2026 15:17:53 -0300 Subject: [PATCH 1/9] feat(simple_metrics): use frequency_alert while adaptive_state is escalated (SIENTIAPDE-2190) --- orchestrator/utils/orchestrator_functions.py | 14 ++++++- .../utils/test_orchestrator_functions.py | 38 +++++++++++++++++++ 2 files changed, 51 insertions(+), 1 deletion(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 39ee0aa..e89b354 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -85,6 +85,9 @@ def simple_metrics(config: dict[str, Any]): - interval_minutes (int, optional): Computation interval in minutes (default: 60) - metrics (list[str], optional): List of metrics to compute (default: ['rmse', 'mse', 'mae', 'r2']) + - frequency_alert (str, optional): Frequency used while `adaptive_state.escalated` + is true; same format as `frequency`. Absent = adaptive frequency off. + - adaptive_state (dict, optional): `{escalated, healthy_streak}`, written by laborious. - Additional fields from common_config Returns: @@ -92,14 +95,23 @@ def simple_metrics(config: dict[str, Any]): actual data), target table, time interval, and metrics list. Results stored in 'simple_metrics' table """ + base = common_config(config) + frequency_alert = config.get('frequency_alert') + # SIENTIAPDE-1995: laborious flips `adaptive_state.escalated` on a threshold breach (and back + # after the window is renewed without one) and bumps `updated_at`, which makes this tick + # rewrite the schedule. + if frequency_alert and (config.get('adaptive_state') or {}).get('escalated'): + base['frequency'] = frequency_alert return { - **common_config(config), + **base, 'schema': 'sientia_data', 'predictions_table_name': 'predictions', 'data_table_name': 'laborious_data', 'target_table_name': 'simple_metrics', 'interval_minutes': config.get('interval_minutes', 60), 'metrics': config.get('metrics', ['rmse', 'mse', 'mae', 'r2']), + # Forwarded so create_schedule_config can warn on a value below the floor. + 'frequency_alert': frequency_alert, } diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index 54a9771..5c140a9 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -110,6 +110,7 @@ def test_simple_metrics(): 'target_table_name': 'simple_metrics', 'interval_minutes': 120, 'metrics': ['rmse', 'mse'], + 'frequency_alert': None, 'execution_timeout_seconds': 300, 'task_timeout_seconds': 300, 'runtime': 'legacy', @@ -117,6 +118,43 @@ def test_simple_metrics(): assert result == expected +def _simple_metrics_config(**fields): + return { + 'workflow_type': 'simple_metrics', + 'schedule_name': 'm1-simple_metrics-60m', + 'model_id': 'm1', + 'model': {'name': 'm1'}, + 'frequency': '1h', + **fields, + } + + +def test_simple_metrics_escalated_uses_frequency_alert(): + config = _simple_metrics_config( + frequency_alert='5m', adaptive_state={'escalated': True, 'healthy_streak': 0} + ) + result = simple_metrics(config) + assert result['frequency'] == '5m' + assert result['frequency_alert'] == '5m' + + +def test_simple_metrics_not_escalated_keeps_frequency(): + config = _simple_metrics_config( + frequency_alert='5m', adaptive_state={'escalated': False, 'healthy_streak': 2} + ) + assert simple_metrics(config)['frequency'] == '1h' + + +def test_simple_metrics_escalated_without_frequency_alert_keeps_frequency(): + # Removing frequency_alert turns adaptive mode off even if a stale state says escalated. + config = _simple_metrics_config(adaptive_state={'escalated': True, 'healthy_streak': 0}) + assert simple_metrics(config)['frequency'] == '1h' + + +def test_simple_metrics_frequency_alert_without_state_keeps_frequency(): + assert simple_metrics(_simple_metrics_config(frequency_alert='5m'))['frequency'] == '1h' + + def test_minimal_retrain(): config = { 'workflow_type': 'minimal_retrain', From 7222b5d1ef4e918c2b79dc9f589bf793ece420a2 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 7 Oct 2026 15:28:45 -0300 Subject: [PATCH 2/9] feat(simple_metrics): warn when frequency_alert is below the 1m floor (SIENTIAPDE-2190) --- orchestrator/activities/formatters.py | 19 ++++++++++ orchestrator/utils/orchestrator_functions.py | 21 ++++++++++ .../activities/test_formatters.py | 38 +++++++++++++++++++ .../utils/test_orchestrator_functions.py | 19 ++++++++++ 4 files changed, 97 insertions(+) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index f33b4db..6075554 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -20,6 +20,7 @@ with workflow.unsafe.imports_passed_through(): from orchestrator.utils.orchestrator_functions import ( build_tag_config, drift, + frequency_alert_warning, gather_read_tags, minimal_retrain, pi_web_api_scouter, @@ -409,6 +410,24 @@ class Formatters(SientiaMonitoring): schedules, current_schedules, to_update, to_create, namespace, metadata ) + # Only schedules being created/updated, so this fires once per config change. + changed = {**to_create[self.laborious_namespace], **to_update[self.laborious_namespace]} + for schedule_name, schedule in changed.items(): + warning = frequency_alert_warning(schedule.get('frequency_alert')) + if warning: + self.send_notification( + metadata={ + **metadata, + 'schedule_name': schedule_name, + 'model_id': schedule.get('model_id'), + 'model_name': schedule.get('model_name'), + }, + notification_id='SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR', + message=warning, + block='create_schedule_config', + level=NotificationLevel.WARNING, + ) + for namespace, schedules in current_schedule_config.items(): for schedule_name in schedules: if schedule_name not in schedule_config[namespace]: diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index e89b354..291573c 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -115,6 +115,27 @@ def simple_metrics(config: dict[str, Any]): } +MIN_FREQUENCY_ALERT_SECONDS = 60 + + +def frequency_alert_warning(frequency_alert: str | None) -> str | None: + """ + Warning message when `frequency_alert` is below the 1-minute floor or does not parse, + otherwise None. Warn only: the configured value is kept (SIENTIAPDE-1995 decision 7). + """ + if not frequency_alert: + return None + try: + # str(): a non-string value (e.g. a number typed into the doc) must warn, not raise + # AttributeError and fail the whole create_schedule_config activity. + seconds = parse_frequency(str(frequency_alert)) + except ValueError: + return f'frequency_alert={frequency_alert!r} is not a valid frequency' + if seconds < MIN_FREQUENCY_ALERT_SECONDS: + return f'frequency_alert={frequency_alert} is below the 1m floor; kept as configured' + return None + + def minimal_retrain(config: dict[str, Any]): """ Build minimal retrain configuration from pipeline config. diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index a7aba4a..85e190f 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -1023,3 +1023,41 @@ def test_filter_notification_reports_ignore_models_id_explicit_none_does_not_rai response = formatters.filter_notification_reports(input_data) assert response['test_group_1']['notifications'] == input_data['notification_package'] + + +def _schedule_config_with_frequency_alert(frequency_alert): + return { + 'metadata': {}, + 'current_schedule_config': {'laborious': {}}, + 'schedule_config': { + 'laborious': { + 'm1-simple_metrics-60m': { + 'workflow_type': 'simple_metrics', + 'model_id': 'm1', + 'model_name': 'm1', + 'frequency': '1h', + 'frequency_alert': frequency_alert, + 'updated_at': '2026-10-07', + } + }, + 'scouter': {}, + }, + } + + +def test_create_schedule_config_warns_on_frequency_alert_below_floor(formatters): + result = formatters.create_schedule_config(_schedule_config_with_frequency_alert('30s')) + + # The configured value is kept. + assert result['to_create']['laborious']['m1-simple_metrics-60m']['frequency_alert'] == '30s' + formatters.send_notification.assert_called_once() + kwargs = formatters.send_notification.call_args.kwargs + assert kwargs['notification_id'] == 'SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR' + assert kwargs['level'] == NotificationLevel.WARNING + assert kwargs['metadata']['schedule_name'] == 'm1-simple_metrics-60m' + + +def test_create_schedule_config_no_warning_at_or_above_floor(formatters): + formatters.create_schedule_config(_schedule_config_with_frequency_alert('5m')) + + formatters.send_notification.assert_not_called() diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index 5c140a9..b7008e4 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -1,10 +1,13 @@ from unittest.mock import call, patch +from pytest import mark + from orchestrator.utils.orchestrator_functions import ( base_scouter, build_tag_config, common_config, drift, + frequency_alert_warning, gather_read_tags, minimal_retrain, overlap_filter_config, @@ -800,3 +803,19 @@ def test_pi_web_api_scouter_with_no_timeout(): 'runtime': 'legacy', } assert result == expected + + +@mark.parametrize( + 'frequency_alert, warns', + [ + (None, False), + ('5m', False), + ('1m', False), # the floor itself is allowed + ('60s', False), + ('30s', True), + ('5min', True), # unparseable + (5, True), # not a string: must warn, never raise + ], +) +def test_frequency_alert_warning(frequency_alert, warns): + assert (frequency_alert_warning(frequency_alert) is not None) is warns From 0f902b53a99b65a0d246bb3e05686650d3d83362 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 7 Oct 2026 15:59:29 -0300 Subject: [PATCH 3/9] test(e2e): one simple_metrics schedule per window for the same model (SIENTIAPDE-2193) --- .../orchestrator_simple_metrics_windows.json | 45 +++++++++++++++++++ e2e/scenarios.md | 5 +++ e2e/test_orchestrator_main_workflow.py | 35 +++++++++++++++ 3 files changed, 85 insertions(+) create mode 100644 e2e/scenario_inputs/orchestrator_simple_metrics_windows.json diff --git a/e2e/scenario_inputs/orchestrator_simple_metrics_windows.json b/e2e/scenario_inputs/orchestrator_simple_metrics_windows.json new file mode 100644 index 0000000..8f93c8b --- /dev/null +++ b/e2e/scenario_inputs/orchestrator_simple_metrics_windows.json @@ -0,0 +1,45 @@ +{ + "workflow_input": { + "schedule_name": "orchestrator-e2e-sm-windows", + "pipelines_query": { + "collection": "pipelines", + "aggregation": [{"$match": {"active": true}}] + }, + "opc_servers_query": { + "collection": "opc_servers", + "filters": {"active": true} + } + }, + "pipelines": [ + { + "schedule_name": "model-sm-simple_metrics-60m", + "workflow_type": "simple_metrics", + "model_id": "model-sm", + "model": { + "name": "Model model-sm", + "model_config": {"target": "y", "simple_metrics_thresholds": {"rmse_max": 1.0}} + }, + "active": true, + "updated_at": "@now", + "frequency": "1h", + "frequency_alert": "5m", + "interval_minutes": 60 + }, + { + "schedule_name": "model-sm-simple_metrics-1440m", + "workflow_type": "simple_metrics", + "model_id": "model-sm", + "model": { + "name": "Model model-sm", + "model_config": {"target": "y", "simple_metrics_thresholds": {"rmse_max": 1.0}} + }, + "active": true, + "updated_at": "@now", + "frequency": "1d", + "frequency_alert": "1h", + "interval_minutes": 1440 + } + ], + "opc_servers": [], + "active_ingestors": ["heartbeat:ingestor:1"] +} diff --git a/e2e/scenarios.md b/e2e/scenarios.md index bd66ac4..6f439a0 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -45,6 +45,11 @@ unrelated Temporal schedule present. First run creates TTL index on notification collection used by scouter pipelines. +### 1.7.1 Simple metrics parallel windows + +Two `simple_metrics` pipelines for the same `model_id` (60 min and 1440 min windows) → one +Temporal schedule per window, each with its own `frequency` as interval. + --- ## 2. Alerts workflow diff --git a/e2e/test_orchestrator_main_workflow.py b/e2e/test_orchestrator_main_workflow.py index f2cb635..3f2d0ce 100644 --- a/e2e/test_orchestrator_main_workflow.py +++ b/e2e/test_orchestrator_main_workflow.py @@ -1,5 +1,7 @@ """E2E tests for the Orchestrator main workflow.""" +from datetime import timedelta + import pytest from pymongo import MongoClient from redis import Redis @@ -26,6 +28,13 @@ async def _schedule_ids(host: str, namespace: str) -> list[str]: return [schedule.id async for schedule in await client.list_schedules()] +async def _schedule_every(host: str, schedule_id: str) -> timedelta: + """Return the interval of a laborious-namespace schedule.""" + client = await Client.connect(host, namespace='laborious') + description = await client.get_schedule_handle(schedule_id).describe() + return description.schedule.spec.intervals[0].every + + def _apply_orchestrator_seeds( scenario: dict, mongo_uri: str, @@ -224,3 +233,29 @@ async def test_scenario_1_6_1_ttl_index_bootstrap( assert any('expireAfterSeconds' in info for info in indexes.values()) finally: mongo.close() + + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_scenario_1_7_1_simple_metrics_parallel_windows( + temporal_env: WorkflowEnvironment, + orchestrator_worker, + stub_workers, + mongo_uri: str, + temporal_host: str, + redis_client: Redis, +): + """1.7.1: One pipelines doc per window gives one schedule per window, each with its own interval.""" + scenario = load_scenario_input('orchestrator_simple_metrics_windows') + _apply_orchestrator_seeds(scenario, mongo_uri, redis_client) + + await start_and_await_workflow( + temporal_env.client, + Orchestrator.run, + scenario['workflow_input'], + make_workflow_id('orchestrator-sm-windows'), + ) + + short, long = 'model-sm-simple_metrics-60m', 'model-sm-simple_metrics-1440m' + assert await _schedule_every(temporal_host, short) == timedelta(hours=1) + assert await _schedule_every(temporal_host, long) == timedelta(days=1) From 7b3618d274e7923980515d4a1a99fa21c6279076 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 7 Oct 2026 16:22:47 -0300 Subject: [PATCH 4/9] test(e2e): adaptive frequency escalation and recovery round trip (SIENTIAPDE-2193) --- e2e/scenarios.md | 6 +++ e2e/test_orchestrator_main_workflow.py | 56 +++++++++++++++++++++++++- 2 files changed, 61 insertions(+), 1 deletion(-) diff --git a/e2e/scenarios.md b/e2e/scenarios.md index 6f439a0..b09f2d6 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -50,6 +50,12 @@ First run creates TTL index on notification collection used by scouter pipelines Two `simple_metrics` pipelines for the same `model_id` (60 min and 1440 min windows) → one Temporal schedule per window, each with its own `frequency` as interval. +### 1.7.2 Adaptive frequency escalation/recovery round trip + +Laborious-style write (`adaptive_state.escalated=true` + BSON-datetime `updated_at`) on one +window → next run switches only that schedule to `frequency_alert`; the recovery write switches it +back to `frequency`. + --- ## 2. Alerts workflow diff --git a/e2e/test_orchestrator_main_workflow.py b/e2e/test_orchestrator_main_workflow.py index 3f2d0ce..99e9414 100644 --- a/e2e/test_orchestrator_main_workflow.py +++ b/e2e/test_orchestrator_main_workflow.py @@ -1,6 +1,6 @@ """E2E tests for the Orchestrator main workflow.""" -from datetime import timedelta +from datetime import UTC, datetime, timedelta import pytest from pymongo import MongoClient @@ -259,3 +259,57 @@ async def test_scenario_1_7_1_simple_metrics_parallel_windows( short, long = 'model-sm-simple_metrics-60m', 'model-sm-simple_metrics-1440m' assert await _schedule_every(temporal_host, short) == timedelta(hours=1) assert await _schedule_every(temporal_host, long) == timedelta(days=1) + + +def _laborious_adaptive_write(mongo_uri: str, schedule_name: str, escalated: bool) -> None: + """Replicates laborious `update_simple_metrics_adaptive_state` on an escalated flip.""" + mongo = MongoClient(mongo_uri) + try: + mongo[E2E_DATABASE]['pipelines'].update_one( + {'schedule_name': schedule_name}, + { + '$set': { + 'adaptive_state': {'escalated': escalated, 'healthy_streak': 0}, + # BSON datetime, as laborious writes it; a string would break the tick. + 'updated_at': datetime.now(UTC), + } + }, + ) + finally: + mongo.close() + + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_scenario_1_7_2_simple_metrics_adaptive_frequency_round_trip( + temporal_env: WorkflowEnvironment, + orchestrator_worker, + stub_workers, + mongo_uri: str, + temporal_host: str, + redis_client: Redis, +): + """1.7.2: Escalation written by laborious switches only that window to frequency_alert; recovery switches it back.""" + scenario = load_scenario_input('orchestrator_simple_metrics_windows') + _apply_orchestrator_seeds(scenario, mongo_uri, redis_client) + short, long = 'model-sm-simple_metrics-60m', 'model-sm-simple_metrics-1440m' + + async def run(tag: str) -> None: + await start_and_await_workflow( + temporal_env.client, + Orchestrator.run, + scenario['workflow_input'], + make_workflow_id(f'orchestrator-sm-adaptive-{tag}'), + ) + + await run('create') + assert await _schedule_every(temporal_host, short) == timedelta(hours=1) + + _laborious_adaptive_write(mongo_uri, short, escalated=True) + await run('escalate') + assert await _schedule_every(temporal_host, short) == timedelta(minutes=5) + assert await _schedule_every(temporal_host, long) == timedelta(days=1) + + _laborious_adaptive_write(mongo_uri, short, escalated=False) + await run('recover') + assert await _schedule_every(temporal_host, short) == timedelta(hours=1) From 2c3ce07f5c174e23b70e875b19603201d9c55d53 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 7 Oct 2026 16:25:30 -0300 Subject: [PATCH 5/9] docs: simple metrics parallel windows and adaptive frequency convention (SIENTIAPDE-2193) --- README.md | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/README.md b/README.md index cd37937..5ac0a06 100644 --- a/README.md +++ b/README.md @@ -612,6 +612,35 @@ Temporal input configuration sample: } ``` +#### Simple metrics: parallel windows & adaptive frequency + +One `pipelines` doc per window, all with the same `model_id`. Name them `-simple_metrics-m`; each doc becomes its own Temporal schedule. Thresholds are shared through `model.model_config`. The adaptive state is per doc. + +```json +{ + "schedule_name": "model-x-simple_metrics-60m", + "workflow_type": "simple_metrics", + "model_id": "model-x", + "active": true, + "interval_minutes": 60, + "frequency": "1h", + "frequency_alert": "5m" +} +``` + +| Field | Meaning | Default | +|---|---|---| +| `frequency_alert` | Frequency while escalated, same format as `frequency`. Absent = adaptive mode off. Below `1m` the orchestrator sends one WARNING and keeps the value | — | +| `adaptive_state` | `{escalated, healthy_streak}`, written **only** by laborious (with `updated_at` as a BSON datetime when `escalated` flips) | `{false, 0}` | +| `model.model_config.simple_metrics_thresholds` | `{_max, _min}` shared by all windows | — | +| `model.model_config.simple_metrics_min_samples` | Windows with fewer rows neither alert nor escalate, and count toward recovery | — | + +Rules: +- **Escalation:** laborious escalates on the first breach. +- **Recovery:** after `min(12, ceil(interval_minutes / frequency_alert))` consecutive runs without one. Empty windows count as runs without a breach. +- **Alerts** are per model (trigger `-simple_metrics`), so several windows seeing the same breach produce one alert. +- **Delay:** the schedule switches on the orchestrator tick after laborious flips the state, at most one tick later. + #### Alerts Workflow ```json From 33d99827b179b03d1587031ebb1be5507e39cf96 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 7 Oct 2026 16:37:04 -0300 Subject: [PATCH 6/9] docs: clarify floor warning repeats and laborious owns escalation (SIENTIAPDE-2193) --- README.md | 4 +++- orchestrator/activities/formatters.py | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 5ac0a06..b0e3973 100644 --- a/README.md +++ b/README.md @@ -630,11 +630,13 @@ One `pipelines` doc per window, all with the same `model_id`. Name them ` | Field | Meaning | Default | |---|---|---| -| `frequency_alert` | Frequency while escalated, same format as `frequency`. Absent = adaptive mode off. Below `1m` the orchestrator sends one WARNING and keeps the value | — | +| `frequency_alert` | Frequency while escalated, same format as `frequency`. Absent = adaptive mode off. Below `1m` (or unparseable), the orchestrator sends a WARNING each time it creates or updates the schedule, and keeps the value | — | | `adaptive_state` | `{escalated, healthy_streak}`, written **only** by laborious (with `updated_at` as a BSON datetime when `escalated` flips) | `{false, 0}` | | `model.model_config.simple_metrics_thresholds` | `{_max, _min}` shared by all windows | — | | `model.model_config.simple_metrics_min_samples` | Windows with fewer rows neither alert nor escalate, and count toward recovery | — | +Escalation, recovery and the per-model alert are implemented in laborious (SIENTIAPDE-2189, SIENTIAPDE-2191); the orchestrator only picks `frequency_alert` while `adaptive_state.escalated` is true. + Rules: - **Escalation:** laborious escalates on the first breach. - **Recovery:** after `min(12, ceil(interval_minutes / frequency_alert))` consecutive runs without one. Empty windows count as runs without a breach. diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 6075554..41378eb 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -370,6 +370,8 @@ class Formatters(SientiaMonitoring): Compares the current schedule configuration in Temporal with the new schedule configuration to determine which schedules need to be created, updated, or deleted. Uses timestamp comparison to identify schedules that have changed. + Also sends a SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR WARNING for laborious schedules + whose frequency_alert is below 1m or unparseable (the value is kept). Args: input_data (dict[str, Any]): The input data containing: @@ -410,7 +412,7 @@ class Formatters(SientiaMonitoring): schedules, current_schedules, to_update, to_create, namespace, metadata ) - # Only schedules being created/updated, so this fires once per config change. + # Only schedules being created/updated: fires on each config change (including escalation flips), not every tick. changed = {**to_create[self.laborious_namespace], **to_update[self.laborious_namespace]} for schedule_name, schedule in changed.items(): warning = frequency_alert_warning(schedule.get('frequency_alert')) From b5512498b907f7c08c80abc15a334a71eaacb6b4 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Thu, 8 Oct 2026 08:42:49 -0300 Subject: [PATCH 7/9] fix(simple_metrics): ignore unparseable frequency_alert and malformed adaptive_state (SIENTIAPDE-2190) --- orchestrator/utils/orchestrator_functions.py | 15 ++++++++++++--- .../orchestrator/activities/test_formatters.py | 12 ++++++++++++ .../utils/test_orchestrator_functions.py | 18 ++++++++++++++++++ 3 files changed, 42 insertions(+), 3 deletions(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 291573c..c5df001 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -100,8 +100,16 @@ def simple_metrics(config: dict[str, Any]): # SIENTIAPDE-1995: laborious flips `adaptive_state.escalated` on a threshold breach (and back # after the window is renewed without one) and bumps `updated_at`, which makes this tick # rewrite the schedule. - if frequency_alert and (config.get('adaptive_state') or {}).get('escalated'): - base['frequency'] = frequency_alert + state = config.get('adaptive_state') + # Hand-written docs: a malformed state must not raise and kill the tick for every model. + if frequency_alert and isinstance(state, dict) and state.get('escalated') is True: + try: + # Temporal parses the schedule frequency on create/update; an unparseable value + # would fail it, so keep `frequency` (create_schedule_config still warns). + parse_frequency(str(frequency_alert)) + base['frequency'] = frequency_alert + except ValueError: + pass return { **base, 'schema': 'sientia_data', @@ -121,7 +129,8 @@ MIN_FREQUENCY_ALERT_SECONDS = 60 def frequency_alert_warning(frequency_alert: str | None) -> str | None: """ Warning message when `frequency_alert` is below the 1-minute floor or does not parse, - otherwise None. Warn only: the configured value is kept (SIENTIAPDE-1995 decision 7). + otherwise None. Warn only (SIENTIAPDE-1995 decision 7): a value below the floor is kept; + an unparseable one is ignored by `simple_metrics()`, which keeps `frequency`. """ if not frequency_alert: return None diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index 85e190f..4891eb2 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/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')) diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index b7008e4..45209ee 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/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', From 9f21372ab313a20d26c12c4dc73775787be01f70 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Thu, 8 Oct 2026 08:43:45 -0300 Subject: [PATCH 8/9] docs: floor warning fires on each schedule create/update; unparseable frequency_alert is ignored (SIENTIAPDE-2193) --- README.md | 2 +- orchestrator/activities/formatters.py | 5 +++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index b0e3973..a0d7666 100644 --- a/README.md +++ b/README.md @@ -630,7 +630,7 @@ One `pipelines` doc per window, all with the same `model_id`. Name them ` | Field | Meaning | Default | |---|---|---| -| `frequency_alert` | Frequency while escalated, same format as `frequency`. Absent = adaptive mode off. Below `1m` (or unparseable), the orchestrator sends a WARNING each time it creates or updates the schedule, and keeps the value | — | +| `frequency_alert` | Frequency while escalated, same format as `frequency`. Absent = adaptive mode off. Below `1m` (or unparseable), the orchestrator sends a WARNING each time it creates or updates the schedule. A value below `1m` is kept; an unparseable value is ignored (the schedule keeps `frequency`) | — | | `adaptive_state` | `{escalated, healthy_streak}`, written **only** by laborious (with `updated_at` as a BSON datetime when `escalated` flips) | `{false, 0}` | | `model.model_config.simple_metrics_thresholds` | `{_max, _min}` shared by all windows | — | | `model.model_config.simple_metrics_min_samples` | Windows with fewer rows neither alert nor escalate, and count toward recovery | — | diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 41378eb..17a8187 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -371,7 +371,7 @@ class Formatters(SientiaMonitoring): configuration to determine which schedules need to be created, updated, or deleted. Uses timestamp comparison to identify schedules that have changed. Also sends a SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR WARNING for laborious schedules - whose frequency_alert is below 1m or unparseable (the value is kept). + whose frequency_alert is below 1m (the value is kept) or unparseable (it is ignored). Args: input_data (dict[str, Any]): The input data containing: @@ -412,7 +412,8 @@ class Formatters(SientiaMonitoring): schedules, current_schedules, to_update, to_create, namespace, metadata ) - # Only schedules being created/updated: fires on each config change (including escalation flips), not every tick. + # Fires whenever the schedule is created or updated (each `updated_at` change, + # including escalation flips). changed = {**to_create[self.laborious_namespace], **to_update[self.laborious_namespace]} for schedule_name, schedule in changed.items(): warning = frequency_alert_warning(schedule.get('frequency_alert')) From c96d94a045568ce1f000f47a86d6b12a13eb7265 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Thu, 8 Oct 2026 14:03:23 +0000 Subject: [PATCH 9/9] chore: bump version to 0.7.0 --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 026f790..392f860 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "orchestrator" -version = "0.6.8" +version = "0.7.0" description = "Sientia DataOps Orchestrator - ML Model Orchestration System" readme = "README.md" requires-python = ">=3.11"