validator: add --validator.cycle_offset_minutes for wall-clock lane scheduling

neurons/validator.py

@@ -118,40 +118,58 @@ def __init__(self, config=None, cycle_name: str = CYCLE_LOW_FREQUENCY):
             )
             self.cycle_name = self.config.validator.cycle_name
 
-        self._apply_assets_filter()
+        self._apply_cycle_overrides()
 
-    def _apply_assets_filter(self):
-        """Narrow the active cycle's asset_list to --validator.assets, if set.
+    def _apply_cycle_overrides(self):
+        """Apply the per-process CLI overrides to the active cycle's
+        PromptConfig: narrow asset_list to --validator.assets and anchor
+        the schedule to a wall-clock lane (--validator.cycle_offset_minutes).
 
-        No-op for the scoring cycle or when the flag is unset. Raises
-        ValueError on assets not in the active cycle's asset_list.
+        No-op for the scoring cycle or when the flags are unset. Raises
+        ValueError on assets not in the active cycle's asset_list, or an
+        offset outside [0, cycle_interval_minutes).
         """
-        assets_filter = self.config.validator.assets
-        if assets_filter == "":
-            return
-
         if self.cycle_name == CYCLE_LOW_FREQUENCY:
             target_config: PromptConfig = LOW_FREQUENCY
         elif self.cycle_name == CYCLE_HIGH_FREQUENCY:
             target_config = HIGH_FREQUENCY
         else:
             return
 
-        requested = [a.strip() for a in assets_filter.split(",") if a.strip()]
-        unknown = set(requested) - set(target_config.asset_list)
-        if unknown:
-            raise ValueError(
-                f"--validator.assets contains unknown assets {sorted(unknown)} "
-                f"for {target_config.label}-frequency cycle. "
-                f"Valid: {target_config.asset_list}"
+        assets_filter = self.config.validator.assets
+        if assets_filter != "":
+            requested = [
+                a.strip() for a in assets_filter.split(",") if a.strip()
+            ]
+            unknown = set(requested) - set(target_config.asset_list)
+            if unknown:
+                raise ValueError(
+                    f"--validator.assets contains unknown assets "
+                    f"{sorted(unknown)} for {target_config.label}-frequency "
+                    f"cycle. Valid: {target_config.asset_list}"
+                )
+            target_config.asset_list = [
+                a for a in target_config.asset_list if a in requested
+            ]
+            bt.logging.info(
+                f"Restricting {target_config.label}-frequency cycle to "
+                f"assets: {target_config.asset_list}"
+            )
+
+        offset = self.config.validator.cycle_offset_minutes
+        if offset is not None:
+            if not 0 <= offset < target_config.cycle_interval_minutes:
+                raise ValueError(
+                    f"--validator.cycle_offset_minutes {offset} must be in "
+                    f"[0, {target_config.cycle_interval_minutes}) for the "
+                    f"{target_config.label}-frequency cycle"
+                )
+            target_config.cycle_offset_minutes = offset
+            bt.logging.info(
+                f"Anchoring {target_config.label}-frequency cycle to "
+                f"wall-clock lane {offset} "
+                f"(mod {target_config.cycle_interval_minutes})"
             )
-        target_config.asset_list = [
-            a for a in target_config.asset_list if a in requested
-        ]
-        bt.logging.info(
-            f"Restricting {target_config.label}-frequency cycle to assets: "
-            f"{target_config.asset_list}"
-        )
 
     def forward_validator(self):
         """Boot metagraph refresh, then run this process's cycle forever."""

synth/utils/config.py

@@ -223,6 +223,12 @@ def add_validator_args(_, parser: argparse.ArgumentParser):
         default="",
     )
 
+    parser.add_argument(
+        "--validator.cycle_offset_minutes",
+        type=int,
+        default=None,
+    )
+
     parser.add_argument(
         "--validator.mode",
         type=str,

synth/utils/sequential_scheduler.py

@@ -66,6 +66,12 @@ def select_delay(
         prompt_config: PromptConfig,
         first_run: bool = False,
     ) -> int:
+        offset = prompt_config.cycle_offset_minutes
+        if offset is not None:
+            return SequentialScheduler.select_grid_delay(
+                cycle_start_time, prompt_config, offset, first_run
+            )
+
         next_cycle = cycle_start_time
         next_cycle = round_time_to_minutes(next_cycle)
         if not first_run:
@@ -83,6 +89,34 @@ def select_delay(
 
         return delay
 
+    @staticmethod
+    def select_grid_delay(
+        cycle_start_time: datetime,
+        prompt_config: PromptConfig,
+        offset: int,
+        first_run: bool = False,
+    ) -> int:
+        """Delay until the next minute boundary T where
+        minutes-since-epoch(T) % cycle_interval_minutes == offset.
+        """
+        interval = prompt_config.cycle_interval_minutes
+
+        now = get_current_time()
+        next_boundary = round_time_to_minutes(now)
+        boundary_minutes = int(next_boundary.timestamp()) // 60
+        next_cycle = next_boundary + timedelta(
+            minutes=(offset - boundary_minutes) % interval
+        )
+
+        if not first_run and next_cycle - cycle_start_time > timedelta(
+            minutes=interval
+        ):
+            bt.logging.warning(
+                "Cycle overran its slot, skipping to the next one in its lane"
+            )
+
+        return int((next_cycle - now).total_seconds())
+
     @staticmethod
     def select_asset(latest_asset: str | None, asset_list: list[str]) -> str:
         asset = asset_list[0]

synth/validator/prompt_config.py

@@ -9,6 +9,7 @@ class PromptConfig:
     time_increment: int
     cycle_interval_minutes: int
     timeout_extra_seconds: int
+    cycle_offset_minutes: int | None = None
     num_simulations: int = 1000
     data_retention_days: int = 30
     # Density tapering: predictions on validator_requests whose start_time is

tests/test_sequential_scheduler.py

@@ -116,5 +116,96 @@ def test_overrun_falls_back_to_next_minute(self):
         warn_mock.assert_called_once()
 
 
+def _grid_config(cycle_interval_minutes: int, offset: int) -> PromptConfig:
+    cfg = _config(cycle_interval_minutes=cycle_interval_minutes)
+    cfg.cycle_offset_minutes = offset
+    return cfg
+
+
+class TestSelectGridDelay(unittest.TestCase):
+    # 2026-05-14 14:00 UTC is minute 29646120 since epoch: even, and ≡ 0 mod 5.
+
+    def test_lanes_interleave(self):
+        # interval 2 at 14:00:30 → lane 0 fires at 14:02, lane 1 at 14:01.
+        # Together the two lanes cover every minute.
+        now = datetime(2026, 5, 14, 14, 0, 30, tzinfo=timezone.utc)
+        with patch(
+            "synth.utils.sequential_scheduler.get_current_time",
+            return_value=now,
+        ):
+            delay_lane_0 = SequentialScheduler.select_delay(
+                cycle_start_time=now,
+                prompt_config=_grid_config(cycle_interval_minutes=2, offset=0),
+                first_run=True,
+            )
+            delay_lane_1 = SequentialScheduler.select_delay(
+                cycle_start_time=now,
+                prompt_config=_grid_config(cycle_interval_minutes=2, offset=1),
+                first_run=True,
+            )
+        self.assertEqual(delay_lane_1, 30)
+        self.assertEqual(delay_lane_0, 90)
+
+    def test_steady_state_keeps_lane(self):
+        # Previous cycle fired on its lane slot (14:00, lane 0 mod 2) and
+        # finished 70 s in → next fire is 14:02, no overrun warning.
+        cycle_start_time = datetime(2026, 5, 14, 14, 0, 0, tzinfo=timezone.utc)
+        now = datetime(2026, 5, 14, 14, 1, 10, tzinfo=timezone.utc)
+        with (
+            patch(
+                "synth.utils.sequential_scheduler.get_current_time",
+                return_value=now,
+            ),
+            patch(
+                "synth.utils.sequential_scheduler.bt.logging.warning"
+            ) as warn_mock,
+        ):
+            delay = SequentialScheduler.select_delay(
+                cycle_start_time=cycle_start_time,
+                prompt_config=_grid_config(cycle_interval_minutes=2, offset=0),
+                first_run=False,
+            )
+        self.assertEqual(delay, 50)
+        warn_mock.assert_not_called()
+
+    def test_overrun_relocks_to_own_lane(self):
+        # Cycle started at 14:00 (lane 0 mod 2) but ran 130 s — past its
+        # 14:02 slot. It must skip to 14:04 (its own lane), not drift to
+        # the odd-minute lane, and warn about the missed slot.
+        cycle_start_time = datetime(2026, 5, 14, 14, 0, 0, tzinfo=timezone.utc)
+        now = datetime(2026, 5, 14, 14, 2, 10, tzinfo=timezone.utc)
+        with (
+            patch(
+                "synth.utils.sequential_scheduler.get_current_time",
+                return_value=now,
+            ),
+            patch(
+                "synth.utils.sequential_scheduler.bt.logging.warning"
+            ) as warn_mock,
+        ):
+            delay = SequentialScheduler.select_delay(
+                cycle_start_time=cycle_start_time,
+                prompt_config=_grid_config(cycle_interval_minutes=2, offset=0),
+                first_run=False,
+            )
+        self.assertEqual(delay, 110)
+        warn_mock.assert_called_once()
+
+    def test_offset_none_keeps_relative_schedule(self):
+        # Guard: default configs (offset unset) must not take the grid path —
+        # delay stays relative to the previous cycle, not the wall clock.
+        now = datetime(2026, 5, 14, 14, 1, 0, tzinfo=timezone.utc)
+        with patch(
+            "synth.utils.sequential_scheduler.get_current_time",
+            return_value=now,
+        ):
+            delay = SequentialScheduler.select_delay(
+                cycle_start_time=now,
+                prompt_config=_config(cycle_interval_minutes=5),
+                first_run=False,
+            )
+        self.assertEqual(delay, 300)
+
+
 if __name__ == "__main__":
     unittest.main()