diff --git a/e2e/scenarios.md b/e2e/scenarios.md index 78fc0ae..032917b 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -413,6 +413,15 @@ four metrics with closed-form expected values. #### S.1.2 Subset metrics **Summary**: Requesting `metrics=['rmse']` writes only the rmse row. +#### S.1.3 Threshold breach escalates the adaptive state +**Summary**: With `simple_metrics_thresholds` breached and a `pipelines` doc +(same `schedule_name`) carrying `frequency_alert`, the workflow sets +`adaptive_state={escalated: true, healthy_streak: 0}` and stamps `updated_at`. + +**Expected Outcome**: +- `adaptive_state` written on the `pipelines` doc. +- `updated_at` is a BSON datetime (the orchestrator calls `.replace(tzinfo=UTC)` on it). + ### 6.2 Edge cases #### S.2.1 Zero-variance target returns r2=0 diff --git a/e2e/test_simple_metrics.py b/e2e/test_simple_metrics.py index 900fd24..3edcfce 100644 --- a/e2e/test_simple_metrics.py +++ b/e2e/test_simple_metrics.py @@ -15,6 +15,7 @@ from decimal import Decimal import math import pytest +from pymongo import MongoClient from sqlalchemy import text from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker @@ -332,3 +333,54 @@ async def test_simple_metrics_no_overlapping_data_short_circuits( {'m': str(model_id)}, ).scalar() assert count == 0, 'Empty target data must short-circuit and skip persistence' + + +@pytest.mark.asyncio +@pytest.mark.integration +async def test_simple_metrics_breach_escalates_adaptive_state( + temporal_test_env: WorkflowEnvironment, + temporal_worker_simple_metrics: Worker, + test_activities: Activities, + postgres_engine, + mongo_container, +): + """ + Scenario S.1.3: a threshold breach on a schedule whose pipelines doc has + ``frequency_alert`` escalates ``adaptive_state`` and stamps ``updated_at`` + as a BSON datetime (the type the orchestrator reads). + """ + model_id = 541 + schedule_name = 'test-simple-metrics-adaptive' + _seed_predictions_and_targets( + postgres_engine, model_id=model_id, pairs=[(1.0, 2.0), (2.0, 4.0), (3.0, 5.0)] + ) + + mongo = MongoClient(f'mongodb://localhost:{mongo_container.get_exposed_port(27017)}') + pipelines = mongo['test_db']['pipelines'] + try: + pipelines.delete_many({'schedule_name': schedule_name}) + pipelines.insert_one( + {'schedule_name': schedule_name, 'frequency': '1h', 'frequency_alert': '5m'} + ) + + input_data = _simple_metrics_input( + model_id, + schedule_name=schedule_name, + model_config={ + 'target': 'sensor_target', + 'simple_metrics_thresholds': {'rmse_max': 0.01}, + }, + ) + await start_and_await_workflow( + temporal_test_env.client, + SimpleMetrics.run, + input_data, + make_workflow_id('test-simple-metrics-adaptive'), + ) + + doc = pipelines.find_one({'schedule_name': schedule_name}) + finally: + mongo.close() + + assert doc['adaptive_state'] == {'escalated': True, 'healthy_streak': 0} + assert isinstance(doc['updated_at'], datetime)