From d1786085d0f8c91172efcb0ee6ba430b753ac794 Mon Sep 17 00:00:00 2001
From: vitor-aignosi <vitor.santos@aignosi.com.br>
Date: Tue, 27 Oct 2026 16:19:51 -0300
Subject: [PATCH 1/3] SIENTIAPDE-2212 
 SIENTIAPDE-2212-drift-janela-e-chunk-period: Add configurable chunk period
 and dynamic window calculation for drift, with warnings for invalid periods.

---
 .gitignore                                    |  3 +-
 orchestrator/activities/formatters.py         | 19 ++++-
 orchestrator/activities/temporal_manager.py   |  3 +
 orchestrator/utils/orchestrator_functions.py  | 60 +++++++++++++--
 .../activities/test_formatters.py             | 43 +++++++++++
 .../activities/test_temporal_manager.py       | 39 ++++++++++
 .../utils/test_orchestrator_functions.py      | 75 +++++++++++++++++++
 7 files changed, 235 insertions(+), 7 deletions(-)

diff --git a/.gitignore b/.gitignore
index 372699a..f8e8af5 100644
--- a/.gitignore
+++ b/.gitignore
@@ -49,4 +49,5 @@ git_log
 .env
 
 openspec/
-.cursor/
\ No newline at end of file
+.cursor/
+.claude/
\ No newline at end of file
diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py
index 17a8187..8a06fea 100644
--- a/orchestrator/activities/formatters.py
+++ b/orchestrator/activities/formatters.py
@@ -19,6 +19,7 @@ with workflow.unsafe.imports_passed_through():
 
     from orchestrator.utils.orchestrator_functions import (
         build_tag_config,
+        chunk_period_warning,
         drift,
         frequency_alert_warning,
         gather_read_tags,
@@ -371,7 +372,9 @@ 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 (the value is kept) or unparseable (it is ignored).
+        whose frequency_alert is below 1m (the value is kept) or unparseable (it is ignored),
+        and a DRIFT_CHUNK_PERIOD_INVALID WARNING for drift schedules whose configured
+        chunk_period is not supported by laborious (replaced by 'min').
 
         Args:
             input_data (dict[str, Any]): The input data containing:
@@ -430,6 +433,20 @@ class Formatters(SientiaMonitoring):
                     block='create_schedule_config',
                     level=NotificationLevel.WARNING,
                 )
+            warning = chunk_period_warning(schedule.get('chunk_period_configured'))
+            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='DRIFT_CHUNK_PERIOD_INVALID',
+                    message=warning,
+                    block='create_schedule_config',
+                    level=NotificationLevel.WARNING,
+                )
 
         for namespace, schedules in current_schedule_config.items():
             for schedule_name in schedules:
diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py
index 8fcc95c..ec622e4 100644
--- a/orchestrator/activities/temporal_manager.py
+++ b/orchestrator/activities/temporal_manager.py
@@ -220,6 +220,9 @@ class TemporalManager(SientiaMonitoring):
                     task_queue_name = self._build_task_queue_name(workflow_type, schedule)
                     schedule['task_queue'] = task_queue_name
 
+                    # No `policy`: schedules rely on Temporal's default SKIP overlap. The
+                    # laborious `DriftAnalysis` cache assumes at most one running drift
+                    # workflow per model; revisit its locking if a policy is ever added here.
                     await client.create_schedule(
                         schedule_name,
                         Schedule(
diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py
index c5df001..bdbb592 100644
--- a/orchestrator/utils/orchestrator_functions.py
+++ b/orchestrator/utils/orchestrator_functions.py
@@ -1,3 +1,4 @@
+from math import ceil
 from typing import Any
 
 from orchestrator.utils.converters import parse_frequency
@@ -43,6 +44,28 @@ 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.
+    """
+    if 'interval_minutes' in config:
+        return config['interval_minutes']
+    try:
+        seconds = parse_frequency(str(frequency))
+    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
+
+
 def drift(config: dict[str, Any]):
     """
     Build drift configuration from pipeline config.
@@ -52,25 +75,52 @@ 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 (default: 60)
+            - 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.
+            - 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:
                 ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein'])
             - Additional fields from common_config
 
     Returns:
         dict[str, Any]: Drift detection configuration with source/target tables,
-            time interval, and metrics specifications. Results stored in 'drift_metrics' table
+            time interval, chunk period and metrics specifications. Results stored in
+            'drift_metrics' table
     """
-    return {
-        **common_config(config),
+    base = common_config(config)
+    configured_chunk_period = config.get('chunk_period')
+    chunk_period_valid = configured_chunk_period in DRIFT_CHUNK_PERIODS
+    output = {
+        **base,
         'schema': 'sientia_data',
         'source_table_name': 'laborious_data',
         'target_table_name': 'drift_metrics',
-        'interval': config.get('interval_minutes', 60),
+        'interval': _drift_interval_minutes(config, base['frequency']),
+        'chunk_period': configured_chunk_period if chunk_period_valid else 'min',
         'drift_metrics': config.get(
             'drift_metrics', ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']
         ),
     }
+    if configured_chunk_period is not None and not chunk_period_valid:
+        # Forwarded so create_schedule_config can warn without re-reading the pipeline.
+        output['chunk_period_configured'] = configured_chunk_period
+    return output
+
+
+def chunk_period_warning(chunk_period: Any) -> str | None:
+    """
+    Warning message when a configured drift `chunk_period` is not supported by laborious,
+    otherwise None. Warn only: `drift()` already replaced it with 'min'.
+    """
+    if chunk_period is None or chunk_period in DRIFT_CHUNK_PERIODS:
+        return None
+    return (
+        f"chunk_period={chunk_period!r} is not one of {DRIFT_CHUNK_PERIODS}; falling back to 'min'"
+    )
 
 
 def simple_metrics(config: dict[str, Any]):
diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py
index 4891eb2..320852c 100644
--- a/tests/orchestrator/activities/test_formatters.py
+++ b/tests/orchestrator/activities/test_formatters.py
@@ -1073,3 +1073,46 @@ 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()
+
+
+def _drift_schedule_config(chunk_period_configured):
+    return {
+        'metadata': {},
+        'current_schedule_config': {'laborious': {}},
+        'schedule_config': {
+            'laborious': {
+                'm1-drift-1m': {
+                    'workflow_type': 'drift',
+                    'model_id': 'm1',
+                    'model_name': 'm1',
+                    'frequency': '1m',
+                    'chunk_period': 'min',
+                    'chunk_period_configured': chunk_period_configured,
+                    'updated_at': '2026-10-27',
+                }
+            },
+            'scouter': {},
+        },
+    }
+
+
+def test_create_schedule_config_warns_on_invalid_chunk_period(formatters):
+    result = formatters.create_schedule_config(_drift_schedule_config('hour'))
+
+    assert 'm1-drift-1m' in result['to_create']['laborious']
+    formatters.send_notification.assert_called_once()
+    kwargs = formatters.send_notification.call_args.kwargs
+    assert kwargs['notification_id'] == 'DRIFT_CHUNK_PERIOD_INVALID'
+    assert kwargs['level'] == NotificationLevel.WARNING
+    assert kwargs['metadata']['schedule_name'] == 'm1-drift-1m'
+    assert kwargs['metadata']['model_id'] == 'm1'
+    assert kwargs['metadata']['model_name'] == 'm1'
+
+
+def test_create_schedule_config_no_chunk_period_warning_when_unchanged(formatters):
+    input_data = _drift_schedule_config('hour')
+    input_data['current_schedule_config'] = {'laborious': {'m1-drift-1m': '2026-10-27'}}
+
+    formatters.create_schedule_config(input_data)
+
+    formatters.send_notification.assert_not_called()
diff --git a/tests/orchestrator/activities/test_temporal_manager.py b/tests/orchestrator/activities/test_temporal_manager.py
index 337ff1a..f3c44ed 100644
--- a/tests/orchestrator/activities/test_temporal_manager.py
+++ b/tests/orchestrator/activities/test_temporal_manager.py
@@ -3,6 +3,7 @@ from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
 
 from pytest import fixture, mark
 from sientia_do.notifications.models import NotificationLevel
+from temporalio.client import ScheduleOverlapPolicy
 
 from orchestrator.activities.temporal_manager import TemporalManager
 from orchestrator.utils.converters import parse_frequency
@@ -684,3 +685,41 @@ async def test_create_schedules_tenant_runtime_queue(
         task_timeout=timedelta(seconds=300),
         typed_search_attributes=_mock_typed_search_attributes.return_value,
     )
+
+
+@mark.asyncio
+@patch('orchestrator.activities.temporal_manager.Schedule')
+@patch('orchestrator.activities.temporal_manager.ScheduleActionStartWorkflow')
+@patch('orchestrator.activities.temporal_manager.ScheduleIntervalSpec')
+@patch('orchestrator.activities.temporal_manager.ScheduleSpec')
+@patch('orchestrator.activities.temporal_manager.TypedSearchAttributes')
+@patch('orchestrator.activities.temporal_manager.SearchAttributePair')
+async def test_create_schedules_drift_keeps_default_skip_overlap(
+    _mock_search_attribute_pair,
+    _mock_typed_search_attributes,
+    _mock_schedule_spec,
+    _mock_schedule_interval_spec,
+    _mock_schedule_action_start_workflow,
+    mock_schedule,
+    temporal_manager,
+):
+    # The laborious DriftAnalysis cache assumes no two drift runs of a model overlap.
+    input_data = {
+        'schedules': {
+            'laborious': {
+                'm1-drift-1m': {
+                    'model_id': 'm1',
+                    'model_name': 'm1',
+                    'workflow_type': 'drift',
+                    'frequency': '1m',
+                }
+            }
+        }
+    }
+    temporal_manager.temporal_clients['laborious'].create_schedule = AsyncMock()
+
+    await temporal_manager.create_schedules(input_data)
+
+    mock_schedule.assert_called_once()
+    policy = mock_schedule.call_args.kwargs.get('policy')
+    assert policy is None or policy.overlap == ScheduleOverlapPolicy.SKIP
diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py
index 45209ee..769c679 100644
--- a/tests/orchestrator/utils/test_orchestrator_functions.py
+++ b/tests/orchestrator/utils/test_orchestrator_functions.py
@@ -5,6 +5,7 @@ from pytest import mark
 from orchestrator.utils.orchestrator_functions import (
     base_scouter,
     build_tag_config,
+    chunk_period_warning,
     common_config,
     drift,
     frequency_alert_warning,
@@ -79,6 +80,7 @@ def test_drift():
         'source_table_name': 'laborious_data',
         'target_table_name': 'drift_metrics',
         'interval': 120,
+        'chunk_period': 'min',
         'drift_metrics': ['kolmogorov_smirnov', 'jensen_shannon'],
         'execution_timeout_seconds': 300,
         'task_timeout_seconds': 300,
@@ -87,6 +89,79 @@ def test_drift():
     assert result == expected
 
 
+def _drift_config(**fields):
+    return {
+        'workflow_type': 'drift',
+        'schedule_name': 'm1-drift',
+        'model_id': 'm1',
+        'model': {'name': 'm1'},
+        **fields,
+    }
+
+
+def test_drift_default_window_is_frequency_plus_margin():
+    result = drift(_drift_config())
+
+    assert result['frequency'] == '1m'
+    assert result['interval'] == 2
+
+
+@mark.parametrize(
+    ('frequency', 'interval'),
+    [('1h', 61), ('30s', 2), ('90s', 3)],
+)
+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():
+    result = drift(_drift_config(frequency='1m', interval_minutes=60))
+
+    assert result['interval'] == 60
+
+
+def test_drift_invalid_frequency_falls_back_to_legacy_window():
+    result = drift(_drift_config(frequency='abc'))
+
+    assert result['frequency'] == 'abc'
+    assert result['interval'] == 60
+
+
+def test_drift_chunk_period_defaults_to_min():
+    result = drift(_drift_config())
+
+    assert result['chunk_period'] == 'min'
+    assert 'chunk_period_configured' not in result
+
+
+def test_drift_chunk_period_seconds_kept():
+    result = drift(_drift_config(chunk_period='s'))
+
+    assert result['chunk_period'] == 's'
+    assert 'chunk_period_configured' not in result
+
+
+@mark.parametrize('chunk_period', ['hour', 5])
+def test_drift_invalid_chunk_period_falls_back_to_min(chunk_period):
+    result = drift(_drift_config(chunk_period=chunk_period))
+
+    assert result['chunk_period'] == 'min'
+    assert result['chunk_period_configured'] == chunk_period
+
+
+@mark.parametrize('chunk_period', [None, 'min', 's'])
+def test_chunk_period_warning_none_when_absent_or_valid(chunk_period):
+    assert chunk_period_warning(chunk_period) is None
+
+
+@mark.parametrize('chunk_period', ['hour', 5])
+def test_chunk_period_warning_message_when_invalid(chunk_period):
+    warning = chunk_period_warning(chunk_period)
+
+    assert warning is not None
+    assert repr(chunk_period) in warning
+
+
 def test_simple_metrics():
     config = {
         'workflow_type': 'simple_metrics',

From 9e160655f2e8512bf2acdb19fd473ebcfc305438 Mon Sep 17 00:00:00 2001
From: vitor-aignosi <vitor.santos@aignosi.com.br>
Date: Tue, 27 Oct 2026 16:27:58 -0300
Subject: [PATCH 2/3] SIENTIAPDE-2212  Add e2e scenario for drift window and
 chunk_period

- Add scenario 1.8.1: two drift pipelines on frequency 1m. One has no
  interval_minutes and an invalid chunk_period='hour', so it expects
  interval=2, chunk_period='min' and one DRIFT_CHUNK_PERIOD_INVALID
  notification. The other sets interval_minutes=60 and chunk_period='s',
  and both are forwarded unchanged with no warning (SIENTIAPDE-2212).
- Add a _schedule_input helper that decodes the workflow input of a
  laborious-namespace schedule.
- Reformat the ensure_namespace_search_attributes signature in
  conftest.py to a single line.
---
 e2e/conftest.py                               |  4 +-
 .../orchestrator_drift_window.json            | 38 ++++++++++++++
 e2e/scenarios.md                              |  8 +++
 e2e/test_orchestrator_main_workflow.py        | 51 +++++++++++++++++++
 4 files changed, 98 insertions(+), 3 deletions(-)
 create mode 100644 e2e/scenario_inputs/orchestrator_drift_window.json

diff --git a/e2e/conftest.py b/e2e/conftest.py
index ab81973..0a67ccf 100644
--- a/e2e/conftest.py
+++ b/e2e/conftest.py
@@ -92,9 +92,7 @@ async def register_namespace_if_missing(env: WorkflowEnvironment, namespace: str
     raise TimeoutError(f'Namespace {namespace} not ready within 5s')
 
 
-async def ensure_namespace_search_attributes(
-    env: WorkflowEnvironment, namespace: str
-) -> None:
+async def ensure_namespace_search_attributes(env: WorkflowEnvironment, namespace: str) -> None:
     """
     Register the orchestrator search attributes on a namespace, if missing.
 
diff --git a/e2e/scenario_inputs/orchestrator_drift_window.json b/e2e/scenario_inputs/orchestrator_drift_window.json
new file mode 100644
index 0000000..a034a9c
--- /dev/null
+++ b/e2e/scenario_inputs/orchestrator_drift_window.json
@@ -0,0 +1,38 @@
+{
+  "workflow_input": {
+    "schedule_name": "orchestrator-e2e-drift-window",
+    "pipelines_query": {
+      "collection": "pipelines",
+      "aggregation": [{"$match": {"active": true}}]
+    },
+    "opc_servers_query": {
+      "collection": "opc_servers",
+      "filters": {"active": true}
+    }
+  },
+  "pipelines": [
+    {
+      "schedule_name": "model-dr-drift-default",
+      "workflow_type": "drift",
+      "model_id": "model-dr",
+      "model": {"name": "Model model-dr"},
+      "active": true,
+      "updated_at": "@now",
+      "frequency": "1m",
+      "chunk_period": "hour"
+    },
+    {
+      "schedule_name": "model-dr-drift-explicit",
+      "workflow_type": "drift",
+      "model_id": "model-dr",
+      "model": {"name": "Model model-dr"},
+      "active": true,
+      "updated_at": "@now",
+      "frequency": "1m",
+      "interval_minutes": 60,
+      "chunk_period": "s"
+    }
+  ],
+  "opc_servers": [],
+  "active_ingestors": ["heartbeat:ingestor:1"]
+}
diff --git a/e2e/scenarios.md b/e2e/scenarios.md
index b09f2d6..f0b74d5 100644
--- a/e2e/scenarios.md
+++ b/e2e/scenarios.md
@@ -56,6 +56,14 @@ Laborious-style write (`adaptive_state.escalated=true` + BSON-datetime `updated_
 window → next run switches only that schedule to `frequency_alert`; the recovery write switches it
 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='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).
+
 ---
 
 ## 2. Alerts workflow
diff --git a/e2e/test_orchestrator_main_workflow.py b/e2e/test_orchestrator_main_workflow.py
index 99e9414..bdbc505 100644
--- a/e2e/test_orchestrator_main_workflow.py
+++ b/e2e/test_orchestrator_main_workflow.py
@@ -313,3 +313,54 @@ async def test_scenario_1_7_2_simple_metrics_adaptive_frequency_round_trip(
     _laborious_adaptive_write(mongo_uri, short, escalated=False)
     await run('recover')
     assert await _schedule_every(temporal_host, short) == timedelta(hours=1)
+
+
+async def _schedule_input(host: str, schedule_id: str) -> dict:
+    """Return the decoded workflow input of a laborious-namespace schedule."""
+    client = await Client.connect(host, namespace='laborious')
+    description = await client.get_schedule_handle(schedule_id).describe()
+    action = description.schedule.action
+    assert isinstance(action, ScheduleActionStartWorkflow)
+    [input_data] = await client.data_converter.decode(action.args)
+    return input_data
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_scenario_1_8_1_drift_window_and_chunk_period(
+    temporal_env: WorkflowEnvironment,
+    orchestrator_worker,
+    stub_workers,
+    mongo_uri: str,
+    temporal_host: str,
+    redis_client: Redis,
+    notification_handler,
+):
+    """1.8.1: Drift window defaults to frequency + 1 min margin; invalid chunk_period warns and falls back."""
+    scenario = load_scenario_input('orchestrator_drift_window')
+    _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-drift-window'),
+    )
+
+    default = await _schedule_input(temporal_host, 'model-dr-drift-default')
+    assert default['interval'] == 2
+    assert default['chunk_period'] == 'min'
+    assert default['chunk_period_configured'] == 'hour'
+
+    explicit = await _schedule_input(temporal_host, 'model-dr-drift-explicit')
+    assert explicit['interval'] == 60
+    assert explicit['chunk_period'] == 's'
+    assert 'chunk_period_configured' not in explicit
+
+    warnings = list(
+        notification_handler.mongo_collection.find(
+            {'notification_id': 'DRIFT_CHUNK_PERIOD_INVALID'}
+        )
+    )
+    assert len(warnings) == 1
+    assert warnings[0]['model_id'] == 'model-dr'

From 2d794d02b2cc76fc41b9890dcf2d75d27f66e30c Mon Sep 17 00:00:00 2001
From: vitor-aignosi <vitor.santos@aignosi.com.br>
Date: Tue, 27 Oct 2026 16:42:48 -0300
Subject: [PATCH 3/3] 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                                 |  2 +-
 e2e/test_orchestrator_main_workflow.py           |  4 ++--
 orchestrator/utils/orchestrator_functions.py     | 16 +++++++---------
 .../utils/test_orchestrator_functions.py         |  8 ++++----
 4 files changed, 14 insertions(+), 16 deletions(-)

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