SIENTIAPDE-2212

SIENTIAPDE-2212-drift-janela-e-chunk-period: Remove default 1-minute margin from drift window calculation. The default interval now directly matches the frequency, ensuring consecutive windows do not overlap.

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).

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'
 

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:

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