diff --git a/e2e/scenarios.md b/e2e/scenarios.md index f0b74d5..aa3e9af 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -59,7 +59,7 @@ back to `frequency`. ### 1.8.1 Drift window and chunk_period Two `drift` pipelines on `frequency='1m'`: one without `interval_minutes` and with an invalid -`chunk_period='hour'` → workflow input `interval=2` (frequency + 1-minute margin), +`chunk_period='hour'` → workflow input `interval=1` (frequency, no overlap), `chunk_period='min'` and exactly one `DRIFT_CHUNK_PERIOD_INVALID` notification; one with `interval_minutes=60` and `chunk_period='s'` → both forwarded unchanged, no warning (SIENTIAPDE-2212). diff --git a/e2e/test_orchestrator_main_workflow.py b/e2e/test_orchestrator_main_workflow.py index bdbc505..86d0091 100644 --- a/e2e/test_orchestrator_main_workflow.py +++ b/e2e/test_orchestrator_main_workflow.py @@ -336,7 +336,7 @@ async def test_scenario_1_8_1_drift_window_and_chunk_period( redis_client: Redis, notification_handler, ): - """1.8.1: Drift window defaults to frequency + 1 min margin; invalid chunk_period warns and falls back.""" + """1.8.1: Drift window defaults to frequency; invalid chunk_period warns and falls back.""" scenario = load_scenario_input('orchestrator_drift_window') _apply_orchestrator_seeds(scenario, mongo_uri, redis_client) @@ -348,7 +348,7 @@ async def test_scenario_1_8_1_drift_window_and_chunk_period( ) default = await _schedule_input(temporal_host, 'model-dr-drift-default') - assert default['interval'] == 2 + assert default['interval'] == 1 assert default['chunk_period'] == 'min' assert default['chunk_period_configured'] == 'hour' diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index bdbb592..ac13082 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -47,14 +47,13 @@ def common_config(config: dict[str, Any]): # Kept in sync by hand with the `chunk_period` validation in # sientia-dataops-laborious/laborious/activities/model_metrics.py (no shared contract). DRIFT_CHUNK_PERIODS = ('min', 's') -# Re-reads the tail of the previous window so rows ingested up to 1 minute late still count. -DRIFT_WINDOW_MARGIN_MINUTES = 1 def _drift_interval_minutes(config: dict[str, Any], frequency: Any) -> int: """ - Drift window in minutes: `interval_minutes` as configured (no margin), otherwise the - schedule frequency rounded up to whole minutes (min 1) plus DRIFT_WINDOW_MARGIN_MINUTES. + Drift window in minutes: `interval_minutes` as configured, otherwise the schedule frequency + rounded up to whole minutes (min 1). Consecutive default windows do not overlap, so each + drift chunk is stored once. """ if 'interval_minutes' in config: return config['interval_minutes'] @@ -63,7 +62,7 @@ def _drift_interval_minutes(config: dict[str, Any], frequency: Any) -> int: except ValueError: # create_schedules fails this schedule on its own; don't kill the tick for every model. return 60 - return max(1, ceil(seconds / 60)) + DRIFT_WINDOW_MARGIN_MINUTES + return max(1, ceil(seconds / 60)) def drift(config: dict[str, Any]): @@ -76,10 +75,9 @@ def drift(config: dict[str, Any]): Args: config (dict[str, Any]): Pipeline configuration containing: - interval_minutes (int, optional): Time window for data comparison in minutes, - used as-is (no margin). Default: `frequency` rounded up to whole minutes - (min 1) plus a 1-minute margin for late data (2 with the default '1m'). - Pipelines whose ingestion lag exceeds 1 minute should set it explicitly - (frequency + lag); `interval_minutes: 60` keeps the legacy window. + used as-is. Default: `frequency` rounded up to whole minutes (min 1), so + consecutive windows do not overlap (1 with the default '1m'); + `interval_minutes: 60` keeps the legacy window. - chunk_period (str, optional): Drift bucket size, 'min' or 's' (default: 'min'). Any other value falls back to 'min' and create_schedule_config warns. - drift_metrics (list[str], optional): Statistical metrics to compute (default: diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index 769c679..bda5c74 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -99,22 +99,22 @@ def _drift_config(**fields): } -def test_drift_default_window_is_frequency_plus_margin(): +def test_drift_default_window_is_frequency(): result = drift(_drift_config()) assert result['frequency'] == '1m' - assert result['interval'] == 2 + assert result['interval'] == 1 @mark.parametrize( ('frequency', 'interval'), - [('1h', 61), ('30s', 2), ('90s', 3)], + [('1h', 60), ('30s', 1), ('90s', 2)], ) def test_drift_window_derived_from_frequency(frequency, interval): assert drift(_drift_config(frequency=frequency))['interval'] == interval -def test_drift_explicit_interval_minutes_wins_without_margin(): +def test_drift_explicit_interval_minutes_wins(): result = drift(_drift_config(frequency='1m', interval_minutes=60)) assert result['interval'] == 60