feat(scheduler): implement R92 delegated mode, anti-skip cadence invariants, and compute-error safeguards

api/schedules.py

@@ -8,10 +8,6 @@ from typing import Optional
 
 from aiohttp import web
 
-from ..services.scheduler.models import Schedule, TriggerType
-from ..services.scheduler.storage import get_schedule_store
-from ..services.templates import is_template_allowed
-
 # Import discipline:
 # - ComfyUI runtime: package-relative imports only (prevents collisions with other custom nodes).
 # - Unit tests: allow top-level fallbacks.
@@ -19,13 +15,35 @@ from ..services.templates import is_template_allowed
 # IMPORTANT: Avoid a broad `try/except ImportError` here. Falling back to `services.*` in ComfyUI
 # can silently import another pack's module and break auth/approval semantics.
 if __package__ and "." in __package__:
+    from ..services.scheduler.models import Schedule, TriggerType
+    from ..services.scheduler.storage import get_schedule_store
+    from ..services.templates import is_template_allowed
     from ..services.webhook_auth import AuthError
 else:  # pragma: no cover (test-only import mode)
+    from services.scheduler.models import Schedule, TriggerType  # type: ignore
+    from services.scheduler.storage import get_schedule_store  # type: ignore
+    from services.templates import is_template_allowed  # type: ignore
     from services.webhook_auth import AuthError  # type: ignore
 
 logger = logging.getLogger("ComfyUI-OpenClaw.api.schedules")
 
 
+def _get_scheduler_runner():
+    if __package__ and "." in __package__:
+        from ..services.scheduler.runner import get_scheduler_runner
+    else:  # pragma: no cover (test-only import mode)
+        from services.scheduler.runner import get_scheduler_runner  # type: ignore
+    return get_scheduler_runner()
+
+
+def _get_run_history():
+    if __package__ and "." in __package__:
+        from ..services.scheduler.history import get_run_history
+    else:  # pragma: no cover (test-only import mode)
+        from services.scheduler.history import get_run_history  # type: ignore
+    return get_run_history()
+
+
 class ScheduleHandlers:
     """
     CRUD handlers for /moltbot/schedules endpoints.
@@ -278,9 +296,18 @@ class ScheduleHandlers:
         # Trigger immediate execution via scheduler runner
         import time
 
-        from ..services.scheduler.runner import get_scheduler_runner
+        runner = _get_scheduler_runner()
+        if runner.is_execution_delegated():
+            # IMPORTANT: in public+split mode, embedded scheduler execution must remain blocked.
+            return web.json_response(
+                {
+                    "error": "Scheduler execution is delegated to external control plane",
+                    "code": "scheduler_delegated",
+                    "remediation": "Use external scheduler control plane in split mode.",
+                },
+                status=503,
+            )
 
-        runner = get_scheduler_runner()
         try:
             runner._execute_schedule(schedule, time.time())
             return web.json_response(
@@ -309,9 +336,7 @@ class ScheduleHandlers:
         if not self._store.get(schedule_id):
             return web.json_response({"error": "Schedule not found"}, status=404)
 
-        from ..services.scheduler.history import get_run_history
-
-        history = get_run_history()
+        history = _get_run_history()
         limit = int(request.query.get("limit", "100"))
         offset = int(request.query.get("offset", "0"))
         status = request.query.get("status")
@@ -340,9 +365,7 @@ class ScheduleHandlers:
         except Exception:
             return web.json_response({"error": "Unauthorized"}, status=403)
 
-        from ..services.scheduler.history import get_run_history
-
-        history = get_run_history()
+        history = _get_run_history()
         limit = int(request.query.get("limit", "100"))
         offset = int(request.query.get("offset", "0"))
         status = request.query.get("status")

services/runtime_config.py

@@ -103,6 +103,8 @@ ALLOWED_SCHEDULER_KEYS = {
     "startup_jitter_sec",
     "max_runs_per_tick",
     "skip_missed_intervals",
+    "execution_mode",
+    "compute_error_disable_threshold",
 }
 
 # Default values
@@ -122,6 +124,8 @@ DEFAULTS = {
         "startup_jitter_sec": 30,
         "max_runs_per_tick": 5,
         "skip_missed_intervals": False,
+        "execution_mode": "auto",
+        "compute_error_disable_threshold": 3,
     },
 }
 
@@ -135,6 +139,7 @@ CONSTRAINTS = {
 SCHEDULER_CONSTRAINTS = {
     "startup_jitter_sec": (0, 300),
     "max_runs_per_tick": (1, 100),
+    "compute_error_disable_threshold": (1, 20),
 }
 
 # Environment variable mappings (new, legacy)
@@ -157,6 +162,11 @@ SCHEDULER_ENV_MAPPINGS = {
     "startup_jitter_sec": ("OPENCLAW_SCHEDULER_STARTUP_JITTER_SEC", ""),
     "max_runs_per_tick": ("OPENCLAW_SCHEDULER_MAX_RUNS_PER_TICK", ""),
     "skip_missed_intervals": ("OPENCLAW_SCHEDULER_SKIP_MISSED", ""),
+    "execution_mode": ("OPENCLAW_SCHEDULER_EXECUTION_MODE", ""),
+    "compute_error_disable_threshold": (
+        "OPENCLAW_SCHEDULER_COMPUTE_ERROR_DISABLE_THRESHOLD",
+        "",
+    ),
 }
 
 # IMPORTANT:

services/scheduler/models.py

@@ -65,6 +65,10 @@ class Schedule:
     last_tick_ts: Optional[float] = None
     last_run_id: Optional[str] = None
 
+    # R92: bounded compute-error tracking for due/recompute failures
+    compute_error_count: int = 0
+    last_compute_error: Optional[str] = None
+
     def __post_init__(self):
         """Validation after initialization."""
         self.validate()
@@ -109,6 +113,9 @@ class Schedule:
                 if not url.startswith(("http://", "https://")):
                     raise ValueError("delivery.url must be http(s)")
 
+        if self.compute_error_count < 0:
+            raise ValueError("compute_error_count must be >= 0")
+
     def to_dict(self) -> Dict[str, Any]:
         """Convert to dict for JSON serialization."""
         d = asdict(self)
@@ -134,6 +141,28 @@ class Schedule:
         self.last_run_id = run_id
         self.updated_at = datetime.now(timezone.utc).isoformat()
 
+    def clear_compute_error(self) -> None:
+        """Reset transient compute-error state after successful due evaluation."""
+        if self.compute_error_count or self.last_compute_error:
+            self.compute_error_count = 0
+            self.last_compute_error = None
+            self.updated_at = datetime.now(timezone.utc).isoformat()
+
+    def record_compute_error(self, error_message: str, disable_threshold: int) -> bool:
+        """
+        Record a due/recompute error and optionally disable this schedule.
+
+        Returns:
+            True if the schedule is disabled by threshold, otherwise False.
+        """
+        self.compute_error_count += 1
+        self.last_compute_error = str(error_message or "")[:200]
+        disabled = self.compute_error_count >= max(1, disable_threshold)
+        if disabled:
+            self.enabled = False
+        self.updated_at = datetime.now(timezone.utc).isoformat()
+        return disabled
+
 
 # Maximum schedules allowed
 MAX_SCHEDULES = 200

services/scheduler/runner.py

@@ -6,6 +6,7 @@ Background tick loop for executing due schedules.
 import asyncio
 import hashlib
 import logging
+import os
 import random
 import threading
 import time
@@ -19,6 +20,41 @@ from .storage import get_schedule_store
 
 logger = logging.getLogger("ComfyUI-OpenClaw.services.scheduler")
 
+SCHEDULER_EXECUTION_EMBEDDED = "embedded"
+SCHEDULER_EXECUTION_DELEGATED = "delegated"
+
+
+def resolve_scheduler_execution_mode(config: Optional[dict] = None) -> str:
+    """
+    Resolve scheduler execution mode for current runtime.
+
+    Modes:
+    - embedded: in-process scheduler executes due jobs.
+    - delegated: in-process scheduler is read-only; execution is delegated.
+    """
+    if config is None:
+        config = get_scheduler_config()
+
+    explicit = str(config.get("execution_mode", "auto")).strip().lower()
+    if explicit in {SCHEDULER_EXECUTION_EMBEDDED, SCHEDULER_EXECUTION_DELEGATED}:
+        return explicit
+
+    profile = os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local").strip().lower()
+    try:
+        from ..control_plane import ControlPlaneMode, resolve_control_plane_mode
+
+        if (
+            profile == "public"
+            and resolve_control_plane_mode(profile) == ControlPlaneMode.SPLIT
+        ):
+            return SCHEDULER_EXECUTION_DELEGATED
+    except Exception:
+        # Fail-open to embedded for non-public/local workflows if control-plane
+        # module cannot be resolved in this runtime.
+        pass
+
+    return SCHEDULER_EXECUTION_EMBEDDED
+
 
 def compute_idempotency_key(schedule_id: str, tick_ts: float) -> str:
     """
@@ -141,12 +177,133 @@ class SchedulerRunner:
 
         self._store = get_schedule_store()
 
+    def is_execution_delegated(self, config: Optional[dict] = None) -> bool:
+        """Return True when in-process scheduler execution is delegated/blocked."""
+        return resolve_scheduler_execution_mode(config) == SCHEDULER_EXECUTION_DELEGATED
+
+    @staticmethod
+    def _compute_cursor_tick_ts(schedule: Schedule, tick_ts: float) -> float:
+        """
+        Compute the persisted cursor timestamp for this execution.
+
+        R92 invariant:
+        - Interval schedules advance by one interval step from previous cursor
+          (when available), rather than jumping directly to `now`.
+        - This prevents cadence drift/long-jump behavior for daily intervals.
+        """
+        if (
+            schedule.trigger_type == TriggerType.INTERVAL
+            and schedule.interval_sec
+            and schedule.last_tick_ts is not None
+        ):
+            next_tick = schedule.last_tick_ts + float(schedule.interval_sec)
+            if next_tick <= tick_ts:
+                return next_tick
+        return tick_ts
+
+    @staticmethod
+    def _compute_startup_skip_cursor_ts(schedule: Schedule, now_ts: float) -> float:
+        """
+        Compute cursor advancement for startup skip-missed maintenance.
+
+        For interval schedules, advance to the latest interval boundary <= now.
+        For all other cases, fall back to now.
+        """
+        if (
+            schedule.trigger_type == TriggerType.INTERVAL
+            and schedule.interval_sec
+            and schedule.last_tick_ts is not None
+            and schedule.interval_sec > 0
+        ):
+            elapsed = now_ts - schedule.last_tick_ts
+            if elapsed <= 0:
+                return now_ts
+            intervals = int(elapsed // float(schedule.interval_sec))
+            if intervals > 0:
+                return schedule.last_tick_ts + (
+                    intervals * float(schedule.interval_sec)
+                )
+        return now_ts
+
+    def _record_compute_error(
+        self, schedule: Schedule, exc: Exception, disable_threshold: int
+    ) -> None:
+        """Persist per-schedule due/recompute error and deterministic disable state."""
+        disabled = schedule.record_compute_error(str(exc), disable_threshold)
+        self._store.update(schedule)
+
+        if disabled:
+            logger.error(
+                "R92_SCHED_COMPUTE_DISABLED: schedule=%s disabled after %s due/recompute errors",
+                schedule.schedule_id,
+                schedule.compute_error_count,
+            )
+        else:
+            logger.warning(
+                "R92_SCHED_COMPUTE_ERROR: schedule=%s error_count=%s error=%s",
+                schedule.schedule_id,
+                schedule.compute_error_count,
+                str(exc),
+            )
+
+    def _evaluate_schedule_due(
+        self,
+        schedule: Schedule,
+        now: datetime,
+        now_ts: float,
+        disable_threshold: int,
+    ) -> bool:
+        """Evaluate due status with bounded error tracking."""
+        try:
+            is_due = False
+            if schedule.trigger_type == TriggerType.CRON:
+                is_due = is_cron_due(schedule.cron_expr, schedule.last_tick_ts, now)
+            elif schedule.trigger_type == TriggerType.INTERVAL:
+                is_due = is_interval_due(
+                    schedule.interval_sec, schedule.last_tick_ts, now_ts
+                )
+            had_compute_error = bool(
+                schedule.compute_error_count or schedule.last_compute_error
+            )
+            schedule.clear_compute_error()
+            if had_compute_error:
+                self._store.update(schedule)
+            return is_due
+        except Exception as e:
+            self._record_compute_error(schedule, e, disable_threshold)
+            return False
+
+    def _collect_due_schedules(
+        self,
+        schedules: list[Schedule],
+        now: datetime,
+        now_ts: float,
+        disable_threshold: int,
+    ) -> list[Schedule]:
+        """Execution-path recompute: collect due schedules without cursor mutation."""
+        due_schedules: list[Schedule] = []
+        for schedule in schedules:
+            if not schedule.enabled:
+                continue
+            if self._evaluate_schedule_due(schedule, now, now_ts, disable_threshold):
+                due_schedules.append(schedule)
+        return due_schedules
+
     def start(self) -> None:
         """Start the scheduler background loop."""
         if self._running:
             logger.warning("Scheduler already running")
             return
 
+        startup_config = get_scheduler_config()
+        if self.is_execution_delegated(startup_config):
+            logger.info(
+                "R92: Scheduler execution delegated (public+split). Embedded runner is disabled."
+            )
+            self._running = False
+            self._thread = None
+            return
+
         self._running = True
         self._stop_event.clear()
 
@@ -177,6 +334,13 @@ class SchedulerRunner:
         # R34: Read config once at startup for jitter/skip behavior
         config = get_scheduler_config()
 
+        if self.is_execution_delegated(config):
+            logger.info(
+                "R92: Scheduler loop startup blocked by delegated execution mode."
+            )
+            self._running = False
+            return
+
         # 1. Startup Jitter
         jitter_sec = config.get("startup_jitter_sec", 0)
         if jitter_sec > 0:
@@ -193,11 +357,17 @@ class SchedulerRunner:
         if config.get("skip_missed_intervals"):
             logger.info("Skip Missed Intervals enabled: advancing cursors...")
             try:
-                self._skip_missed_ticks()
+                self._skip_missed_ticks(config=config)
             except Exception as e:
                 logger.error(f"Failed to skip missed ticks: {e}")
 
         while not self._stop_event.is_set():
+            if self.is_execution_delegated():
+                logger.warning(
+                    "R92: execution mode switched to delegated; stopping embedded scheduler loop."
+                )
+                self._running = False
+                break
             try:
                 self._tick()
             except Exception as e:
@@ -208,34 +378,34 @@ class SchedulerRunner:
 
         logger.debug("Scheduler loop exited")
 
-    def _skip_missed_ticks(self) -> None:
+    def _skip_missed_ticks(self, config: Optional[dict] = None) -> None:
         """
         Advance all due schedules to now without executing them.
         Prevents backlog burst after downtime.
         """
+        if config is None:
+            config = get_scheduler_config()
+
         now = datetime.now(timezone.utc)
         now_ts = now.timestamp()
+        disable_threshold = int(config.get("compute_error_disable_threshold", 3))
         schedules = self._store.list_all()
 
         skipped_count = 0
-        for schedule in schedules:
-            if not schedule.enabled:
-                continue
+        due_schedules = self._collect_due_schedules(
+            schedules=schedules,
+            now=now,
+            now_ts=now_ts,
+            disable_threshold=disable_threshold,
+        )
 
-            is_due = False
-            if schedule.trigger_type == TriggerType.CRON:
-                is_due = is_cron_due(schedule.cron_expr, schedule.last_tick_ts, now)
-            elif schedule.trigger_type == TriggerType.INTERVAL:
-                is_due = is_interval_due(
-                    schedule.interval_sec, schedule.last_tick_ts, now_ts
-                )
-
-            if is_due:
-                # Update cursor without running
-                # Use a special run_id to indicate skip
-                schedule.update_cursor(now_ts, "skipped_startup")
-                self._store.update(schedule)
-                skipped_count += 1
+        for schedule in due_schedules:
+            # Update cursor without running (startup skip policy)
+            # Use a special run_id to indicate skip.
+            skip_ts = self._compute_startup_skip_cursor_ts(schedule, now_ts)
+            schedule.update_cursor(skip_ts, "skipped_startup")
+            self._store.update(schedule)
+            skipped_count += 1
 
         if skipped_count > 0:
             logger.info(
@@ -250,25 +420,21 @@ class SchedulerRunner:
         # R34: Dynamic config read for runtime tuning
         config = get_scheduler_config()
         max_runs = config.get("max_runs_per_tick", 5)
+        disable_threshold = int(config.get("compute_error_disable_threshold", 3))
+
+        if self.is_execution_delegated(config):
+            logger.debug(
+                "R92: delegated mode active, skipping in-process scheduler tick execution."
+            )
+            return
 
         schedules = self._store.list_all()
-        due_schedules = []
-
-        for schedule in schedules:
-            if not schedule.enabled:
-                continue
-
-            is_due = False
-
-            if schedule.trigger_type == TriggerType.CRON:
-                is_due = is_cron_due(schedule.cron_expr, schedule.last_tick_ts, now)
-            elif schedule.trigger_type == TriggerType.INTERVAL:
-                is_due = is_interval_due(
-                    schedule.interval_sec, schedule.last_tick_ts, now_ts
-                )
-
-            if is_due:
-                due_schedules.append(schedule)
+        due_schedules = self._collect_due_schedules(
+            schedules=schedules,
+            now=now,
+            now_ts=now_ts,
+            disable_threshold=disable_threshold,
+        )
 
         if due_schedules:
             logger.debug(f"Found {len(due_schedules)} due schedules")
@@ -289,6 +455,9 @@ class SchedulerRunner:
 
     def _execute_schedule(self, schedule: Schedule, tick_ts: float) -> None:
         """Execute a single due schedule."""
+        if self.is_execution_delegated():
+            raise RuntimeError("scheduler_delegated")
+
         idempotency_key = compute_idempotency_key(schedule.schedule_id, tick_ts)
 
         # R9: Check if already processed via history
@@ -354,7 +523,8 @@ class SchedulerRunner:
                 run_record.skip("No submit function")
 
             # Update cursor
-            schedule.update_cursor(tick_ts, run_id)
+            effective_tick_ts = self._compute_cursor_tick_ts(schedule, tick_ts)
+            schedule.update_cursor(effective_tick_ts, run_id)
             self._store.update(schedule)
 
             # R9: Record run
@@ -368,7 +538,8 @@ class SchedulerRunner:
             history.add_run(run_record)
 
             # Still update cursor to avoid retry storm
-            schedule.update_cursor(tick_ts, run_id)
+            effective_tick_ts = self._compute_cursor_tick_ts(schedule, tick_ts)
+            schedule.update_cursor(effective_tick_ts, run_id)
             self._store.update(schedule)
 
 

tests/test_api_schedules_r92.py

@@ -0,0 +1,63 @@
+import json
+import unittest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+try:
+    from aiohttp import web  # noqa: F401
+
+    AIOHTTP_AVAILABLE = True
+except ModuleNotFoundError:
+    AIOHTTP_AVAILABLE = False
+
+
+@unittest.skipIf(not AIOHTTP_AVAILABLE, "aiohttp not available")
+class TestScheduleApiR92(unittest.IsolatedAsyncioTestCase):
+    async def test_run_now_blocked_in_delegated_mode(self):
+        from api.schedules import ScheduleHandlers
+
+        mock_store = MagicMock()
+        mock_store.get.return_value = MagicMock(template_id="tmpl_1")
+        mock_runner = MagicMock()
+        mock_runner.is_execution_delegated.return_value = True
+
+        with patch("api.schedules.get_schedule_store", return_value=mock_store):
+            handlers = ScheduleHandlers()
+
+        request = AsyncMock()
+        request.match_info = {"schedule_id": "sched_1"}
+
+        with patch("api.schedules._get_scheduler_runner", return_value=mock_runner):
+            resp = await handlers.run_now(request)
+
+        self.assertEqual(resp.status, 503)
+        body = json.loads(resp.body)
+        self.assertEqual(body["code"], "scheduler_delegated")
+        mock_runner._execute_schedule.assert_not_called()
+
+    async def test_run_now_executes_when_embedded_mode(self):
+        from api.schedules import ScheduleHandlers
+
+        mock_schedule = MagicMock(template_id="tmpl_2")
+        mock_store = MagicMock()
+        mock_store.get.return_value = mock_schedule
+        mock_runner = MagicMock()
+        mock_runner.is_execution_delegated.return_value = False
+
+        with patch("api.schedules.get_schedule_store", return_value=mock_store):
+            handlers = ScheduleHandlers()
+
+        request = AsyncMock()
+        request.match_info = {"schedule_id": "sched_2"}
+
+        with patch("api.schedules._get_scheduler_runner", return_value=mock_runner):
+            resp = await handlers.run_now(request)
+
+        self.assertEqual(resp.status, 200)
+        body = json.loads(resp.body)
+        self.assertTrue(body["triggered"])
+        self.assertEqual(body["schedule_id"], "sched_2")
+        mock_runner._execute_schedule.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_scheduler_r92.py

@@ -0,0 +1,146 @@
+import os
+import unittest
+from unittest.mock import MagicMock, patch
+
+from services.scheduler.models import Schedule, TriggerType
+from services.scheduler.runner import (
+    SCHEDULER_EXECUTION_DELEGATED,
+    SCHEDULER_EXECUTION_EMBEDDED,
+    SchedulerRunner,
+    resolve_scheduler_execution_mode,
+)
+
+
+class TestSchedulerR92(unittest.TestCase):
+    def setUp(self):
+        self.store_patcher = patch("services.scheduler.runner.get_schedule_store")
+        self.mock_get_store = self.store_patcher.start()
+        self.mock_store = MagicMock()
+        self.mock_get_store.return_value = self.mock_store
+
+        self.history_patcher = patch("services.scheduler.runner.get_run_history")
+        self.mock_get_history = self.history_patcher.start()
+        self.mock_history = MagicMock()
+        self.mock_history.is_processed.return_value = False
+        self.mock_get_history.return_value = self.mock_history
+
+    def tearDown(self):
+        self.store_patcher.stop()
+        self.history_patcher.stop()
+
+    def test_resolve_execution_mode_auto_public_split(self):
+        with patch.dict(
+            os.environ,
+            {
+                "OPENCLAW_DEPLOYMENT_PROFILE": "public",
+                "OPENCLAW_CONTROL_PLANE_MODE": "split",
+            },
+            clear=True,
+        ):
+            mode = resolve_scheduler_execution_mode({"execution_mode": "auto"})
+            self.assertEqual(mode, SCHEDULER_EXECUTION_DELEGATED)
+
+    def test_resolve_execution_mode_explicit_embedded(self):
+        mode = resolve_scheduler_execution_mode({"execution_mode": "embedded"})
+        self.assertEqual(mode, SCHEDULER_EXECUTION_EMBEDDED)
+
+    def test_start_is_noop_when_execution_delegated(self):
+        with patch(
+            "services.scheduler.runner.get_scheduler_config",
+            return_value={"execution_mode": "delegated"},
+        ):
+            runner = SchedulerRunner(submit_fn=None, tick_interval=30.0)
+            runner.start()
+            self.assertFalse(runner._running)
+            self.assertIsNone(runner._thread)
+
+    def test_interval_cursor_advances_single_interval_step(self):
+        schedule = Schedule(
+            schedule_id="sched_r92_interval",
+            name="R92 interval",
+            template_id="tmpl",
+            trigger_type=TriggerType.INTERVAL,
+            interval_sec=86400,
+            last_tick_ts=0.0,
+        )
+
+        with patch(
+            "services.scheduler.runner.get_scheduler_config",
+            return_value={"execution_mode": "embedded"},
+        ):
+            runner = SchedulerRunner(submit_fn=None, tick_interval=30.0)
+            runner._execute_schedule(schedule, tick_ts=200000.0)
+
+        # R92 invariant: one-step interval cursor progression (anti long-jump drift).
+        self.assertEqual(schedule.last_tick_ts, 86400.0)
+
+    def test_due_compute_error_threshold_disables_schedule(self):
+        schedule = Schedule(
+            schedule_id="sched_r92_cron",
+            name="R92 cron",
+            template_id="tmpl",
+            trigger_type=TriggerType.CRON,
+            cron_expr="5 * * * *",
+        )
+        self.mock_store.list_all.return_value = [schedule]
+
+        with (
+            patch(
+                "services.scheduler.runner.get_scheduler_config",
+                return_value={
+                    "execution_mode": "embedded",
+                    "max_runs_per_tick": 5,
+                    "compute_error_disable_threshold": 2,
+                },
+            ),
+            patch(
+                "services.scheduler.runner.is_cron_due",
+                side_effect=ValueError("boom"),
+            ),
+        ):
+            runner = SchedulerRunner(submit_fn=None, tick_interval=30.0)
+            runner._execute_schedule = MagicMock()
+            runner._tick()
+            self.assertTrue(schedule.enabled)
+            self.assertEqual(schedule.compute_error_count, 1)
+
+            runner._tick()
+            self.assertFalse(schedule.enabled)
+            self.assertEqual(schedule.compute_error_count, 2)
+            self.assertIn("boom", schedule.last_compute_error)
+            runner._execute_schedule.assert_not_called()
+
+    def test_tick_due_recompute_does_not_advance_cursor_before_execute(self):
+        schedule = Schedule(
+            schedule_id="sched_r92_due",
+            name="R92 due separation",
+            template_id="tmpl",
+            trigger_type=TriggerType.INTERVAL,
+            interval_sec=60,
+            last_tick_ts=1000.0,
+        )
+        self.mock_store.list_all.return_value = [schedule]
+
+        with (
+            patch(
+                "services.scheduler.runner.get_scheduler_config",
+                return_value={
+                    "execution_mode": "embedded",
+                    "max_runs_per_tick": 5,
+                    "compute_error_disable_threshold": 3,
+                },
+            ),
+            patch("services.scheduler.runner.is_interval_due", return_value=True),
+        ):
+            runner = SchedulerRunner(submit_fn=None, tick_interval=30.0)
+            seen_cursor = []
+            runner._execute_schedule = MagicMock(
+                side_effect=lambda s, _ts: seen_cursor.append(s.last_tick_ts)
+            )
+            runner._tick()
+
+        self.assertEqual(seen_cursor, [1000.0])
+
+
+if __name__ == "__main__":
+    unittest.main()