From b3a0f7d2ad2ededad73add66a48979e9e6101ef3 Mon Sep 17 00:00:00 2001 From: rookiestar28 Date: Tue, 24 Feb 2026 02:20:05 +0800 Subject: [PATCH] feat(scheduler): implement R92 delegated mode, anti-skip cadence invariants, and compute-error safeguards --- api/schedules.py | 47 ++++-- services/runtime_config.py | 10 ++ services/scheduler/models.py | 29 ++++ services/scheduler/runner.py | 247 +++++++++++++++++++++++++++----- tests/test_api_schedules_r92.py | 63 ++++++++ tests/test_scheduler_r92.py | 146 +++++++++++++++++++ 6 files changed, 492 insertions(+), 50 deletions(-) create mode 100644 tests/test_api_schedules_r92.py create mode 100644 tests/test_scheduler_r92.py diff --git a/api/schedules.py b/api/schedules.py index bfc1941..6ae571b 100644 --- a/api/schedules.py +++ b/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") diff --git a/services/runtime_config.py b/services/runtime_config.py index b06374e..2bb1e18 100644 --- a/services/runtime_config.py +++ b/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: diff --git a/services/scheduler/models.py b/services/scheduler/models.py index 7af2634..da92e07 100644 --- a/services/scheduler/models.py +++ b/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 diff --git a/services/scheduler/runner.py b/services/scheduler/runner.py index 3d92954..c7a8a49 100644 --- a/services/scheduler/runner.py +++ b/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) diff --git a/tests/test_api_schedules_r92.py b/tests/test_api_schedules_r92.py new file mode 100644 index 0000000..5d99142 --- /dev/null +++ b/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() diff --git a/tests/test_scheduler_r92.py b/tests/test_scheduler_r92.py new file mode 100644 index 0000000..0cb2755 --- /dev/null +++ b/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()