From bb5c19891473976bf4d38bf1f19f3a24f5d5ca80 Mon Sep 17 00:00:00 2001 From: Colin Haven <10933733+virtitnerd@users.noreply.github.com> Date: Mon, 12 Oct 2026 14:38:22 -0400 Subject: [PATCH 01/11] Let the routine runner start a job without a scene Split _run_scene so a scheduled job can run the same step executor without syncing scene IDs to the robot, and add start_job(), which refuses to interrupt a routine that is already running on the device. get_routine_runner() gives other callers the per-context instance. Co-Authored-By: Claude Sonnet 5.5 (cherry picked from commit c65764810057f9491d35287e7b707e622eb19405) --- .../bundled_backend/shared/routine_runner.py | 47 ++++++++++++++++++- 1 file changed, 46 insertions(+), 1 deletion(-) diff --git a/src/roborock_local_server/bundled_backend/shared/routine_runner.py b/src/roborock_local_server/bundled_backend/shared/routine_runner.py index 71fcca8e..460ed5e0 100644 --- a/src/roborock_local_server/bundled_backend/shared/routine_runner.py +++ b/src/roborock_local_server/bundled_backend/shared/routine_runner.py @@ -801,6 +801,33 @@ def start_scene(self, scene: dict[str, Any]) -> dict[str, Any]: "stepCount": len(steps), } + def start_job( + self, + *, + device_id: str, + job_id: int, + name: str, + steps: list[RoutineStep], + ) -> dict[str, Any]: + """Run a scheduled job's steps; never interrupts a routine already running.""" + existing = self._tasks_by_device.get(device_id) + if existing is not None and not existing.task.done(): + return { + "accepted": False, + "status": "routine_in_progress", + "jobId": job_id, + "deviceId": device_id, + "activeSceneName": existing.scene_name, + } + scene = {"id": job_id, "name": name, "device_id": device_id} + task = asyncio.get_running_loop().create_task( + self._run_steps(scene=scene, steps=steps, sync_scene_tids=False), + name=f"routine-job-{job_id}-{device_id}", + ) + self._tasks_by_device[device_id] = _ActiveRoutine(task=task, scene_id=job_id, scene_name=name) + task.add_done_callback(lambda finished: self._on_scene_done(device_id, finished)) + return {"accepted": True, "status": "started", "jobId": job_id, "deviceId": device_id, "stepCount": len(steps)} + def _on_scene_done(self, device_id: str, task: asyncio.Task[None]) -> None: current = self._tasks_by_device.get(device_id) if current is not None and current.task is task: @@ -967,6 +994,15 @@ async def _send_step_command( raise async def _run_scene(self, *, scene: dict[str, Any], steps: list[RoutineStep]) -> None: + await self._run_steps(scene=scene, steps=steps, sync_scene_tids=True) + + async def _run_steps( + self, + *, + scene: dict[str, Any], + steps: list[RoutineStep], + sync_scene_tids: bool, + ) -> None: device_id = scene_device_id(scene) device = self._device_record(device_id) logger = logging.LoggerAdapter( @@ -988,7 +1024,8 @@ async def _run_scene(self, *, scene: dict[str, Any], steps: list[RoutineStep]) - client = _RoutineMqttClient(self._context, device, logger) await client.connect() try: - await self._sync_scene_tids(client=client, scene=scene, device_id=device_id, logger=logger) + if sync_scene_tids: + await self._sync_scene_tids(client=client, scene=scene, device_id=device_id, logger=logger) for step_index, step in enumerate(steps): commands = commands_for_step(step) waits_for_step_complete = RoborockDataProtocol.TASK_COMPLETE.value in step.finish_dp_ids @@ -1015,3 +1052,11 @@ async def _run_scene(self, *, scene: dict[str, Any], steps: list[RoutineStep]) - await client.close() logger.info("Completed routine scene=%s device=%s", _scene_name(scene), device_id) + + +def get_routine_runner(context: ServerContext) -> RoutineRunner: + runner = getattr(context, "_routine_runner", None) + if runner is None: + runner = RoutineRunner(context) + setattr(context, "_routine_runner", runner) + return runner From e237e98566aacf5b684c3c91d798faf205bb48e6 Mon Sep 17 00:00:00 2001 From: Colin Haven <10933733+virtitnerd@users.noreply.github.com> Date: Mon, 12 Oct 2026 14:38:22 -0400 Subject: [PATCH 02/11] Add a local executor for cloud-style schedules Roborock's cloud fires server_scheduled_start jobs at nextFireTime. With the vacuum pointed at this server nothing did, so imported schedules never ran. ScheduleRunner evaluates each enabled job's cron in its own timezone and hands due jobs to the routine runner, carrying fan_power, water_box_mode and mop_mode so vac-only and vac+mop schedules behave as configured. One-shot jobs disable themselves after firing. Co-Authored-By: Claude Sonnet 5.5 (cherry picked from commit b5203b2ee58a46f0f41fe1f0f9e0f323b1fc206c) --- .../bundled_backend/shared/schedule_runner.py | 313 ++++++++++++++++++ tests/test_schedule_runner.py | 160 +++++++++ 2 files changed, 473 insertions(+) create mode 100644 src/roborock_local_server/bundled_backend/shared/schedule_runner.py create mode 100644 tests/test_schedule_runner.py diff --git a/src/roborock_local_server/bundled_backend/shared/schedule_runner.py b/src/roborock_local_server/bundled_backend/shared/schedule_runner.py new file mode 100644 index 00000000..c70fd3da --- /dev/null +++ b/src/roborock_local_server/bundled_backend/shared/schedule_runner.py @@ -0,0 +1,313 @@ +"""Local executor for cloud-style schedules (``server_scheduled_start`` jobs). + +Roborock's cloud fires these at ``nextFireTime``. With the vacuum pointed at this +server, nothing else would, so the server evaluates the stored cron entries +itself and hands each due job to the routine runner. +""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from datetime import date, datetime, timedelta, timezone, tzinfo +import json +import logging +from typing import TYPE_CHECKING, Any +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from .data_helpers import as_bool +from .inventory_io import inventory_mutation, load_inventory, write_inventory +from .routine_runner import RoutineExecutionError, RoutineStep, get_routine_runner + +if TYPE_CHECKING: + from .context import ServerContext + +_LOGGER = logging.getLogger(__name__) + +TICK_SECONDS = 20 +_SUPPORTED_JOB_METHOD = "server_scheduled_start" +_CARRIED_ENTRY_KEYS = ("repeat", "fan_power", "water_box_mode", "mop_mode", "mop_template_id") + + +class CronError(ValueError): + """Raised for a cron expression this scheduler cannot evaluate.""" + + +@dataclass(frozen=True) +class CronSpec: + minutes: tuple[int, ...] + hours: tuple[int, ...] + days_of_month: frozenset[int] + months: frozenset[int] + days_of_week: frozenset[int] + dom_restricted: bool + dow_restricted: bool + + def day_matches(self, day: date) -> bool: + if day.month not in self.months: + return False + dom_ok = day.day in self.days_of_month + # cron weekday 0 = Sunday; python's isoweekday() has Sunday as 7. + dow_ok = (day.isoweekday() % 7) in self.days_of_week + if self.dom_restricted and self.dow_restricted: + return dom_ok or dow_ok + return dom_ok and dow_ok + + +def _parse_field(text: str, low: int, high: int, *, name: str) -> tuple[frozenset[int], bool]: + text = text.strip() + if text in ("*", "?"): + return frozenset(range(low, high + 1)), False + values: set[int] = set() + for part in text.split(","): + part = part.strip() + if not part: + raise CronError(f"empty {name} entry in {text!r}") + step = 1 + if "/" in part: + part, _, step_text = part.partition("/") + try: + step = int(step_text) + except ValueError as exc: + raise CronError(f"bad {name} step {step_text!r}") from exc + if step <= 0: + raise CronError(f"{name} step must be positive") + if part in ("*", "?", ""): + start, end = low, high + elif "-" in part: + first, _, last = part.partition("-") + try: + start, end = int(first), int(last) + except ValueError as exc: + raise CronError(f"bad {name} range {part!r}") from exc + else: + try: + start = int(part) + except ValueError as exc: + raise CronError(f"bad {name} value {part!r}") from exc + end = high if step != 1 else start + if start < low or end > high or start > end: + raise CronError(f"{name} value out of range in {text!r}") + values.update(range(start, end + 1, step)) + return frozenset(values), True + + +def parse_cron(expression: str) -> CronSpec: + fields = str(expression or "").split() + if len(fields) != 5: + raise CronError(f"expected 5 cron fields, got {len(fields)}: {expression!r}") + minutes, _ = _parse_field(fields[0], 0, 59, name="minute") + hours, _ = _parse_field(fields[1], 0, 23, name="hour") + dom, dom_restricted = _parse_field(fields[2], 1, 31, name="day-of-month") + months, _ = _parse_field(fields[3], 1, 12, name="month") + raw_dow, dow_restricted = _parse_field(fields[4], 0, 7, name="day-of-week") + dow = frozenset(0 if value == 7 else value for value in raw_dow) + return CronSpec( + minutes=tuple(sorted(minutes)), + hours=tuple(sorted(hours)), + days_of_month=dom, + months=months, + days_of_week=dow, + dom_restricted=dom_restricted, + dow_restricted=dow_restricted, + ) + + +def next_fire(spec: CronSpec, after: datetime) -> datetime | None: + """First fire instant strictly after ``after`` (an aware datetime), in its timezone.""" + zone = after.tzinfo + if zone is None: + raise ValueError("after must be timezone-aware") + start = after.replace(second=0, microsecond=0) + timedelta(minutes=1) + start_utc = start.astimezone(timezone.utc) + first_day = start.date() + for offset in range(0, 800): + day = first_day + timedelta(days=offset) + if not spec.day_matches(day): + continue + for hour in spec.hours: + for minute in spec.minutes: + candidate = datetime(day.year, day.month, day.day, hour, minute, tzinfo=zone) + if candidate.astimezone(timezone.utc) >= start_utc: + return candidate + return None + + +def _zone(name: str, fallback: str) -> tzinfo: + for candidate in (name, fallback): + candidate = str(candidate or "").strip() + if not candidate: + continue + try: + return ZoneInfo(candidate) + except (ZoneInfoNotFoundError, ValueError): + continue + return timezone.utc + + +def _segment_ids(value: Any) -> list[int]: + if isinstance(value, str): + items: list[Any] = [part for part in value.replace(";", ",").split(",") if part.strip()] + elif isinstance(value, (list, tuple)): + items = list(value) + elif value is None: + items = [] + else: + items = [value] + segments: list[int] = [] + for item in items: + try: + number = int(str(item).strip()) + except ValueError: + continue + if number > 0: + segments.append(number) + return segments + + +def job_param(job: dict[str, Any]) -> dict[str, Any]: + param = job.get("param") + if isinstance(param, str): + try: + param = json.loads(param) + except json.JSONDecodeError as exc: + raise RoutineExecutionError("schedule param is not valid JSON") from exc + if not isinstance(param, dict): + raise RoutineExecutionError("schedule is missing its param payload") + return param + + +def steps_for_job(job: dict[str, Any]) -> list[RoutineStep]: + param = job_param(job) + method = str(param.get("method") or "").strip() + if method != _SUPPORTED_JOB_METHOD: + raise RoutineExecutionError(f"unsupported schedule method {method!r}") + params = param.get("params") + entry = params[0] if isinstance(params, list) and params else params + if not isinstance(entry, dict): + raise RoutineExecutionError("schedule params must contain one settings object") + + settings = {key: entry[key] for key in _CARRIED_ENTRY_KEYS if key in entry} + name = str(entry.get("name") or job.get("name") or "").strip() + if not name or name.isdigit(): + name = f"Schedule {job.get('id', '')}".strip() + segments = _segment_ids(entry.get("segments")) + if segments: + return [ + RoutineStep( + step_id=1, + name=name, + method="do_scenes_segments", + params={"data": [{**settings, "segs": segments}]}, + finish_dp_ids=(), + ) + ] + return [ + RoutineStep( + step_id=1, + name=name, + method="do_scenes_app_start", + params=[settings], + finish_dp_ids=(), + ) + ] + + +def next_fire_time_utc(job: dict[str, Any], *, default_timezone: str = "", now: datetime | None = None) -> datetime | None: + if not as_bool(job.get("enabled", True), True): + return None + try: + spec = parse_cron(str(job.get("cron") or "")) + except CronError: + return None + zone = _zone(str(job.get("timeZoneId") or ""), default_timezone) + current = (now or datetime.now(timezone.utc)).astimezone(zone) + upcoming = next_fire(spec, current) + return upcoming.astimezone(timezone.utc) if upcoming is not None else None + + +@inventory_mutation +def _disable_job(ctx: ServerContext, device_id: str, job_id: Any) -> None: + inventory = load_inventory(ctx) + schedules = inventory.get("schedules") + jobs = schedules.get(device_id) if isinstance(schedules, dict) else None + if not isinstance(jobs, list): + return + for job in jobs: + if isinstance(job, dict) and str(job.get("id")) == str(job_id): + job["enabled"] = False + write_inventory(ctx, inventory) + + +class ScheduleRunner: + def __init__(self, context: ServerContext, *, default_timezone: str = "") -> None: + self._context = context + self._default_timezone = default_timezone + self._last_tick: datetime | None = None + self._logger = context.loggers.get("api") or context.loggers.get("unknown") or _LOGGER + + def due_jobs(self, since: datetime, until: datetime) -> list[tuple[str, dict[str, Any]]]: + inventory = load_inventory(self._context) + schedules = inventory.get("schedules") + if not isinstance(schedules, dict): + return [] + due: list[tuple[str, dict[str, Any]]] = [] + for device_id, jobs in schedules.items(): + if not isinstance(jobs, list): + continue + for job in jobs: + if not isinstance(job, dict) or not as_bool(job.get("enabled", True), True): + continue + try: + spec = parse_cron(str(job.get("cron") or "")) + except CronError as exc: + self._logger.warning("Skipping schedule %s on %s: %s", job.get("id"), device_id, exc) + continue + zone = _zone(str(job.get("timeZoneId") or ""), self._default_timezone) + upcoming = next_fire(spec, since.astimezone(zone)) + if upcoming is not None and upcoming.astimezone(timezone.utc) <= until: + due.append((str(device_id), job)) + return due + + def tick(self, now: datetime | None = None) -> list[dict[str, Any]]: + """Fire anything that came due since the previous tick. Needs a running event loop.""" + current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) + previous = self._last_tick + self._last_tick = current + if previous is None: + return [] + results: list[dict[str, Any]] = [] + runner = get_routine_runner(self._context) + for device_id, job in self.due_jobs(previous, current): + job_id = job.get("id") + try: + steps = steps_for_job(job) + result = runner.start_job( + device_id=device_id, + job_id=int(job_id) if str(job_id).isdigit() else 0, + name=steps[0].name, + steps=steps, + ) + except (RoutineExecutionError, ValueError) as exc: + self._logger.warning("Schedule %s on %s not started: %s", job_id, device_id, exc) + result = {"accepted": False, "status": "error", "error": str(exc), "deviceId": device_id} + else: + self._logger.info("Schedule %s on %s fired: %s", job_id, device_id, result.get("status")) + if result.get("accepted") and not as_bool(job.get("repeated", True), True): + _disable_job(self._context, device_id, job_id) + results.append({"jobId": job_id, **result}) + return results + + async def run(self, stop_event: asyncio.Event, *, tick_seconds: float = TICK_SECONDS) -> None: + self.tick() + while not stop_event.is_set(): + try: + await asyncio.wait_for(stop_event.wait(), timeout=tick_seconds) + except TimeoutError: + pass + if stop_event.is_set(): + return + try: + self.tick() + except Exception as exc: # noqa: BLE001 + self._logger.warning("Schedule tick failed: %s", exc) diff --git a/tests/test_schedule_runner.py b/tests/test_schedule_runner.py new file mode 100644 index 00000000..8769898b --- /dev/null +++ b/tests/test_schedule_runner.py @@ -0,0 +1,160 @@ +import json +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +import pytest + +import roborock_local_server.backend # noqa: F401 (puts the bundled `shared` package on sys.path) +from shared import schedule_runner as sr +from shared.routine_runner import RoutineExecutionError + +NEW_YORK = "America/New_York" + + +def _job(**overrides): + job = { + "id": 3549431, + "cron": "00 11 ? * 1,3,5", + "timeZoneId": NEW_YORK, + "repeated": True, + "enabled": True, + "param": { + "id": 1, + "method": "server_scheduled_start", + "params": [ + { + "repeat": 1, + "water_box_mode": 203, + "segments": "0", + "fan_power": 104, + "mop_mode": 300, + "clean_mop": 1, + "map_index": -1, + "name": "1759965417864", + } + ], + }, + } + job.update(overrides) + return job + + +def _utc(*args): + return datetime(*args, tzinfo=timezone.utc) + + +def test_next_fire_matches_the_cloud_reported_fire_times(): + # Values taken from a real cloud snapshot: Monday 2026-09-28 for MWF 11:00 New York. + spec = sr.parse_cron("00 11 ? * 1,3,5") + zone = sr._zone(NEW_YORK, "") + after = _utc(2026, 9, 28, 12, 0).astimezone(zone) + assert sr.next_fire(spec, after).astimezone(timezone.utc) == _utc(2026, 9, 28, 15, 0) + + tue_thu = sr.parse_cron("00 11 ? * 2,4") + assert sr.next_fire(tue_thu, after).astimezone(timezone.utc) == _utc(2026, 9, 29, 15, 0) + + +def test_next_fire_is_strictly_after_and_skips_to_next_matching_day(): + spec = sr.parse_cron("00 11 ? * 1,3,5") + zone = sr._zone(NEW_YORK, "") + on_the_minute = _utc(2026, 9, 28, 15, 0).astimezone(zone) + assert sr.next_fire(spec, on_the_minute).astimezone(timezone.utc) == _utc(2026, 9, 30, 15, 0) + + +def test_weekend_cron_uses_sunday_zero(): + spec = sr.parse_cron("00 11 ? * 0,6") + zone = sr._zone(NEW_YORK, "") + monday = _utc(2026, 9, 28, 12, 0).astimezone(zone) + assert sr.next_fire(spec, monday).astimezone(timezone.utc) == _utc(2026, 10, 3, 15, 0) + + +@pytest.mark.parametrize("expression", ["", "* * * *", "61 * * * *", "* 25 * * *", "a b c d e", "*/0 * * * *"]) +def test_invalid_cron_is_rejected(expression): + with pytest.raises(sr.CronError): + sr.parse_cron(expression) + + +def test_whole_house_job_becomes_an_app_start_step_carrying_vac_and_mop_settings(): + steps = sr.steps_for_job(_job()) + assert len(steps) == 1 + step = steps[0] + assert step.method == "do_scenes_app_start" + assert step.params == [{"repeat": 1, "water_box_mode": 203, "fan_power": 104, "mop_mode": 300}] + assert step.name == "Schedule 3549431" + + +def test_segment_job_becomes_a_segment_clean_step(): + job = _job() + job["param"]["params"][0]["segments"] = "16,17" + step = sr.steps_for_job(job)[0] + assert step.method == "do_scenes_segments" + assert step.params["data"][0]["segs"] == [16, 17] + assert step.params["data"][0]["fan_power"] == 104 + + +def test_unsupported_method_is_refused(): + job = _job() + job["param"]["method"] = "something_else" + with pytest.raises(RoutineExecutionError): + sr.steps_for_job(job) + + +class _FakeRunner: + def __init__(self): + self.started = [] + + def start_job(self, **kwargs): + self.started.append(kwargs) + return {"accepted": True, "status": "started", "deviceId": kwargs["device_id"]} + + +def _runner_with_inventory(tmp_path, monkeypatch, jobs): + inventory = {"schedules": {"duid-1": jobs}} + (tmp_path / "web_api_inventory.json").write_text(json.dumps(inventory), encoding="utf-8") + ctx = SimpleNamespace(http_jsonl=tmp_path / "http.jsonl", loggers={}) + fake = _FakeRunner() + monkeypatch.setattr(sr, "get_routine_runner", lambda _ctx: fake) + return sr.ScheduleRunner(ctx), ctx, fake + + +def test_tick_fires_a_job_that_came_due_since_the_previous_tick(tmp_path, monkeypatch): + runner, _ctx, fake = _runner_with_inventory(tmp_path, monkeypatch, [_job()]) + assert runner.tick(_utc(2026, 9, 28, 14, 59, 40)) == [] # first tick only sets the baseline + results = runner.tick(_utc(2026, 9, 28, 15, 0, 5)) + assert [r["status"] for r in results] == ["started"] + assert fake.started[0]["device_id"] == "duid-1" + assert fake.started[0]["steps"][0].method == "do_scenes_app_start" + + +def test_tick_does_not_fire_twice_or_when_disabled(tmp_path, monkeypatch): + runner, _ctx, fake = _runner_with_inventory( + tmp_path, monkeypatch, [_job(), _job(id=2, enabled=False)] + ) + runner.tick(_utc(2026, 9, 28, 14, 59, 40)) + runner.tick(_utc(2026, 9, 28, 15, 0, 5)) + runner.tick(_utc(2026, 9, 28, 15, 0, 25)) + runner.tick(_utc(2026, 9, 28, 15, 0, 45) + timedelta(minutes=5)) + assert len(fake.started) == 1 + + +def test_one_shot_job_disables_itself_after_firing(tmp_path, monkeypatch): + runner, _ctx, _fake = _runner_with_inventory(tmp_path, monkeypatch, [_job(repeated=False)]) + runner.tick(_utc(2026, 9, 28, 14, 59, 40)) + runner.tick(_utc(2026, 9, 28, 15, 0, 5)) + saved = json.loads((tmp_path / "web_api_inventory.json").read_text(encoding="utf-8")) + assert saved["schedules"]["duid-1"][0]["enabled"] is False + + +@pytest.mark.parametrize("disabled_value", [False, 0, "false", "0"]) +def test_string_typed_disabled_flag_is_respected(tmp_path, monkeypatch, disabled_value): + # Imported inventory stores booleans loosely ("false", 0); "false" must not read as enabled. + runner, _ctx, fake = _runner_with_inventory(tmp_path, monkeypatch, [_job(enabled=disabled_value)]) + runner.tick(_utc(2026, 9, 28, 14, 59, 40)) + runner.tick(_utc(2026, 9, 28, 15, 0, 5)) + assert fake.started == [] + + +def test_next_fire_time_utc_is_none_when_disabled(): + now = _utc(2026, 9, 28, 12, 0) + assert sr.next_fire_time_utc(_job(), now=now) == _utc(2026, 9, 28, 15, 0) + assert sr.next_fire_time_utc(_job(enabled=False), now=now) is None From 36798a3dc037b2baa3d7e86222bc6ecdfce40d0c Mon Sep 17 00:00:00 2001 From: Colin Haven <10933733+virtitnerd@users.noreply.github.com> Date: Mon, 12 Oct 2026 14:38:22 -0400 Subject: [PATCH 03/11] Run the schedule loop with the server and report nextFireTime Start the scheduler alongside the other supervisor tasks and stop it on shutdown. GET /user/devices/{id}/jobs now computes nextFireTime from the cron instead of returning whatever value was captured at import time. Co-Authored-By: Claude Sonnet 5.5 (cherry picked from commit 14ddb5fcd8a7ec1b50fe3e90c8b7cc4cec5b0b97) --- .../https_server/routes/user/devices/service.py | 4 ++++ src/roborock_local_server/server.py | 14 ++++++++++++++ tests/test_schedules.py | 3 +++ 3 files changed, 21 insertions(+) diff --git a/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/service.py b/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/service.py index 0f6990c2..31858449 100644 --- a/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/service.py +++ b/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/service.py @@ -13,6 +13,7 @@ from shared.context import ServerContext from shared.data_helpers import as_bool, as_int, default_product_name, get_value from shared.inventory_io import WEB_API_INVENTORY_FILE, load_inventory +from shared.schedule_runner import next_fire_time_utc _DEFAULT_DEVICE_NAME = "Vacuum 1" @@ -652,6 +653,9 @@ def device_jobs_payload(ctx: ServerContext, device_id: str) -> list[dict[str, An payload["repeated"] = as_bool(get_value(raw, "repeated", default=True), True) payload["enabled"] = as_bool(get_value(raw, "enabled", default=True), True) payload["param"] = get_value(raw, "param", default={}) or {} + upcoming = next_fire_time_utc(payload, default_timezone=str(getattr(ctx, "timezone", "") or "")) + if upcoming is not None: + payload["nextFireTime"] = upcoming.strftime("%Y-%m-%dT%H:%M:%S.000+00:00") schedules.append(payload) return schedules diff --git a/src/roborock_local_server/server.py b/src/roborock_local_server/server.py index 28a57654..53c979aa 100644 --- a/src/roborock_local_server/server.py +++ b/src/roborock_local_server/server.py @@ -53,6 +53,7 @@ ) from shared.inventory_io import atomic_write_inventory, inventory_transaction from shared.protocol_auth import ProtocolAuthStore +from shared.schedule_runner import ScheduleRunner from https_server.routes.user.scene import validity as scene_validity from https_server.routes.auth.service import ( build_login_data_response, @@ -361,6 +362,7 @@ def __init__( self._mqtt_proxy: MqttTlsProxy | None = None self._http_server: ManagedFastApiServer | None = None self._renew_task: asyncio.Task[None] | None = None + self._schedule_task: asyncio.Task[None] | None = None self._stop_event = asyncio.Event() self.certificate_manager = CertificateManager(config=config, paths=paths) @@ -1984,10 +1986,22 @@ async def start(self) -> None: if self._uses_local_tls() and self.config.tls.mode == "cloudflare_acme": self._renew_task = asyncio.create_task(self._renew_loop(), name="tls-renew-loop") + schedule_runner = ScheduleRunner(self.context, default_timezone=self.config.network.timezone) + self._schedule_task = asyncio.create_task( + schedule_runner.run(self._stop_event), + name="schedule-runner", + ) + async def stop(self) -> None: self._stop_event.set() self._turn_relay.stop() self._refresh_turn_health() + if self._schedule_task is not None: + self._schedule_task.cancel() + try: + await self._schedule_task + except asyncio.CancelledError: + pass if self._renew_task is not None: self._renew_task.cancel() try: diff --git a/tests/test_schedules.py b/tests/test_schedules.py index f97b1039..1c0f26f2 100644 --- a/tests/test_schedules.py +++ b/tests/test_schedules.py @@ -104,6 +104,9 @@ def test_device_jobs_endpoint_returns_imported_schedules(tmp_path: Path) -> None payload = response.json() assert payload["success"] is True assert payload["data"] == payload["result"] + # nextFireTime is computed from the cron against "now", so check its shape, not its value. + next_fire_time = payload["result"][0].pop("nextFireTime") + assert next_fire_time.endswith("T13:03:00.000+00:00") assert payload["result"] == [ { **schedule, From 424a315ea502808153a697cf6c9702e0fa99a981 Mon Sep 17 00:00:00 2001 From: Colin Haven <10933733+virtitnerd@users.noreply.github.com> Date: Mon, 12 Oct 2026 16:21:40 -0400 Subject: [PATCH 04/11] Add a schedule store and evaluate crons in the robot's timezone schedule_store holds create/update/delete for the per-device jobs kept in the inventory, so the timer RPC handler, HTTP routes and dashboard can share one writer. Crons without their own timeZoneId now evaluate in the zone the robot is told by /location (America/New_York unless configured) rather than UTC, so a 11:00 schedule fires at 11:00 local. Co-Authored-By: Claude Sonnet 5.5 (cherry picked from commit 0751a7b06b683701a3607eb143258565b548c560) --- .../bundled_backend/shared/schedule_runner.py | 22 +--- .../bundled_backend/shared/schedule_store.py | 119 ++++++++++++++++++ tests/test_schedules.py | 4 +- 3 files changed, 128 insertions(+), 17 deletions(-) create mode 100644 src/roborock_local_server/bundled_backend/shared/schedule_store.py diff --git a/src/roborock_local_server/bundled_backend/shared/schedule_runner.py b/src/roborock_local_server/bundled_backend/shared/schedule_runner.py index c70fd3da..33fa64ce 100644 --- a/src/roborock_local_server/bundled_backend/shared/schedule_runner.py +++ b/src/roborock_local_server/bundled_backend/shared/schedule_runner.py @@ -15,9 +15,11 @@ from typing import TYPE_CHECKING, Any from zoneinfo import ZoneInfo, ZoneInfoNotFoundError +from .constants import DEFAULT_TIMEZONE from .data_helpers import as_bool -from .inventory_io import inventory_mutation, load_inventory, write_inventory +from .inventory_io import load_inventory from .routine_runner import RoutineExecutionError, RoutineStep, get_routine_runner +from .schedule_store import set_enabled if TYPE_CHECKING: from .context import ServerContext @@ -134,7 +136,8 @@ def next_fire(spec: CronSpec, after: datetime) -> datetime | None: def _zone(name: str, fallback: str) -> tzinfo: - for candidate in (name, fallback): + # DEFAULT_TIMEZONE is what /location tells the robot when none is configured. + for candidate in (name, fallback, DEFAULT_TIMEZONE): candidate = str(candidate or "").strip() if not candidate: continue @@ -226,19 +229,6 @@ def next_fire_time_utc(job: dict[str, Any], *, default_timezone: str = "", now: return upcoming.astimezone(timezone.utc) if upcoming is not None else None -@inventory_mutation -def _disable_job(ctx: ServerContext, device_id: str, job_id: Any) -> None: - inventory = load_inventory(ctx) - schedules = inventory.get("schedules") - jobs = schedules.get(device_id) if isinstance(schedules, dict) else None - if not isinstance(jobs, list): - return - for job in jobs: - if isinstance(job, dict) and str(job.get("id")) == str(job_id): - job["enabled"] = False - write_inventory(ctx, inventory) - - class ScheduleRunner: def __init__(self, context: ServerContext, *, default_timezone: str = "") -> None: self._context = context @@ -294,7 +284,7 @@ def tick(self, now: datetime | None = None) -> list[dict[str, Any]]: else: self._logger.info("Schedule %s on %s fired: %s", job_id, device_id, result.get("status")) if result.get("accepted") and not as_bool(job.get("repeated", True), True): - _disable_job(self._context, device_id, job_id) + set_enabled(self._context, device_id, job_id, False) results.append({"jobId": job_id, **result}) return results diff --git a/src/roborock_local_server/bundled_backend/shared/schedule_store.py b/src/roborock_local_server/bundled_backend/shared/schedule_store.py new file mode 100644 index 00000000..1d762fcc --- /dev/null +++ b/src/roborock_local_server/bundled_backend/shared/schedule_store.py @@ -0,0 +1,119 @@ +"""Create, update and delete the per-device schedules kept in the inventory.""" + +from __future__ import annotations + +import copy +from typing import TYPE_CHECKING, Any + +from .inventory_io import inventory_mutation, load_inventory, write_inventory + +if TYPE_CHECKING: + from .context import ServerContext + +DEFAULT_JOB_PARAM: dict[str, Any] = { + "id": 1, + "method": "server_scheduled_start", + "params": [{"repeat": 1, "segments": "0"}], +} + + +def _same_id(job: Any, job_id: Any) -> bool: + return isinstance(job, dict) and str(job.get("id")) == str(job_id) + + +def _jobs_for(inventory: dict[str, Any], duid: str, *, create: bool) -> list[Any] | None: + schedules = inventory.get("schedules") + if not isinstance(schedules, dict): + if not create: + return None + schedules = {} + inventory["schedules"] = schedules + jobs = schedules.get(duid) + if not isinstance(jobs, list): + if not create: + return None + jobs = [] + schedules[duid] = jobs + return jobs + + +def _coerce_id(job_id: Any) -> Any: + text = str(job_id).strip() + return int(text) if text.isdigit() else text + + +def list_jobs(ctx: ServerContext, duid: str) -> list[dict[str, Any]]: + jobs = _jobs_for(load_inventory(ctx), duid, create=False) or [] + return [copy.deepcopy(job) for job in jobs if isinstance(job, dict)] + + +@inventory_mutation +def upsert_job( + ctx: ServerContext, + duid: str, + *, + job_id: Any, + cron: str, + enabled: bool, + time_zone: str = "", + repeated: bool | None = None, + param: dict[str, Any] | None = None, + override_time_zone: bool = False, +) -> dict[str, Any]: + """Create the job, or update cron/enabled on an existing one while keeping its settings. + + An existing job keeps its own timezone unless ``override_time_zone`` is set. + """ + inventory = load_inventory(ctx) + jobs = _jobs_for(inventory, duid, create=True) + assert jobs is not None + existing = next((job for job in jobs if _same_id(job, job_id)), None) + if existing is None: + existing = { + "id": _coerce_id(job_id), + "repeated": True if repeated is None else repeated, + "type": "clean", + "param": copy.deepcopy(param) if param is not None else copy.deepcopy(DEFAULT_JOB_PARAM), + } + if time_zone: + existing["timeZoneId"] = time_zone + jobs.append(existing) + else: + if repeated is not None: + existing["repeated"] = repeated + if param is not None: + existing["param"] = copy.deepcopy(param) + if override_time_zone and time_zone: + existing["timeZoneId"] = time_zone + existing["cron"] = cron + existing["enabled"] = enabled + write_inventory(ctx, inventory) + return copy.deepcopy(existing) + + +@inventory_mutation +def set_enabled(ctx: ServerContext, duid: str, job_id: Any, enabled: bool) -> bool: + inventory = load_inventory(ctx) + jobs = _jobs_for(inventory, duid, create=False) or [] + found = False + for job in jobs: + if _same_id(job, job_id): + job["enabled"] = enabled + found = True + if found: + write_inventory(ctx, inventory) + return found + + +@inventory_mutation +def delete_job(ctx: ServerContext, duid: str, job_id: Any) -> bool: + inventory = load_inventory(ctx) + jobs = _jobs_for(inventory, duid, create=False) + if not jobs: + return False + remaining = [job for job in jobs if not _same_id(job, job_id)] + if len(remaining) == len(jobs): + return False + jobs[:] = remaining + write_inventory(ctx, inventory) + return True diff --git a/tests/test_schedules.py b/tests/test_schedules.py index 1c0f26f2..e4a9ced3 100644 --- a/tests/test_schedules.py +++ b/tests/test_schedules.py @@ -105,8 +105,10 @@ def test_device_jobs_endpoint_returns_imported_schedules(tmp_path: Path) -> None assert payload["success"] is True assert payload["data"] == payload["result"] # nextFireTime is computed from the cron against "now", so check its shape, not its value. + # No timezone is configured, so the cron runs in the zone the robot is told (America/New_York): + # 13:03 local on Dec 15 is 18:03Z. next_fire_time = payload["result"][0].pop("nextFireTime") - assert next_fire_time.endswith("T13:03:00.000+00:00") + assert next_fire_time.endswith("T18:03:00.000+00:00") assert payload["result"] == [ { **schedule, From cdd5c784bab496726896ce41c70903c27f1712e8 Mon Sep 17 00:00:00 2001 From: Colin Haven <10933733+virtitnerd@users.noreply.github.com> Date: Mon, 12 Oct 2026 16:21:40 -0400 Subject: [PATCH 05/11] Answer get/set/upd/del_timer from the schedule store Cloud-managed robots keep their schedules server-side, so a client such as LocalRock that asks the robot for its timers saw nothing and could not change or delete the schedules imported from the cloud. The topic bridge now offers each rr/m request to an interceptor, and timer requests are answered from the store (keeping a schedule's vac/mop settings on edit) instead of being forwarded to the robot. Co-Authored-By: Claude Sonnet 5.5 (cherry picked from commit 2f6f6666f4f374c0e3c2c51ad35b70b6797103ca) --- .../mqtt_broker_server/topic_bridge.py | 19 +- .../bundled_backend/shared/timer_rpc.py | 200 ++++++++++++++++++ src/roborock_local_server/server.py | 5 + tests/test_timer_rpc.py | 172 +++++++++++++++ 4 files changed, 395 insertions(+), 1 deletion(-) create mode 100644 src/roborock_local_server/bundled_backend/shared/timer_rpc.py create mode 100644 tests/test_timer_rpc.py diff --git a/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py b/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py index 7bc087a0..1276126b 100644 --- a/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py +++ b/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py @@ -9,7 +9,7 @@ from pathlib import Path import re import time -from typing import Any +from typing import Any, Callable import aiomqtt @@ -73,7 +73,11 @@ def __init__( fixed_device_mqtt_usr: str = "", runtime_state: Any | None = None, inventory_path: Path | None = None, + request_interceptor: Callable[[str, bytes], bytes | None] | None = None, ) -> None: + # Given (duid, payload) for each rr/m request; returns an encoded reply to + # publish back to the requester instead of forwarding, or None to forward. + self._request_interceptor = request_interceptor self._host = host self._port = port self._logger = logger @@ -389,6 +393,19 @@ async def _handle_cloud_message( cloud_topic: CloudTopicKey, src_topic: str, ) -> None: + if self._request_interceptor is not None: + payload = message.payload + raw = payload if isinstance(payload, (bytes, bytearray)) else str(payload).encode() + try: + reply = self._request_interceptor(cloud_topic.duid, bytes(raw)) + except Exception as exc: # noqa: BLE001 + self._logger.warning("MQTT request interceptor failed for %s: %s", src_topic, exc) + reply = None + if reply is not None: + await client.publish(cloud_topic.topic_out, reply, qos=_extract_qos(message)) + self._logger.info("MQTT bridge answered %s locally -> %s", src_topic, cloud_topic.topic_out) + return + device_topic = self._resolve_device_for_cloud(cloud_topic) if device_topic is None: if cloud_topic.duid not in self._warned_unmapped_duids: diff --git a/src/roborock_local_server/bundled_backend/shared/timer_rpc.py b/src/roborock_local_server/bundled_backend/shared/timer_rpc.py new file mode 100644 index 00000000..f7c867d4 --- /dev/null +++ b/src/roborock_local_server/bundled_backend/shared/timer_rpc.py @@ -0,0 +1,200 @@ +"""Answer the vacuum's legacy timer RPCs (get/set/upd/del_timer) from the schedule store. + +On cloud-managed robots the schedules live server-side, not on the robot, so +clients such as LocalRock that ask the robot for its timers would otherwise see +nothing and could never change the schedules the cloud imported. The topic +bridge hands matching requests here instead of forwarding them to the robot. +""" + +from __future__ import annotations + +import json +import logging +import time +from typing import TYPE_CHECKING, Any + +from roborock.protocol import create_mqtt_decoder, create_mqtt_encoder +from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol + +from .constants import DEFAULT_TIMEZONE +from .schedule_runner import CronError, parse_cron +from .schedule_store import delete_job, list_jobs, set_enabled, upsert_job + +if TYPE_CHECKING: + from .context import ServerContext + +_LOGGER = logging.getLogger(__name__) + +TIMER_METHODS = frozenset({"get_timer", "set_timer", "upd_timer", "del_timer"}) +_OK = ["ok"] + + +class TimerRequestError(ValueError): + """The request used a shape this handler cannot interpret.""" + + +def device_cron(cron: str) -> str | None: + """Cron in the robot's own 'm h dom month dow' form, with '?' written as '*'. + + A one-time schedule pins the day of the month and the month ('15 23 18 10 *'), exactly as + the robot and the official app write it (confirmed live), so it is shown like any other. + A job that pins only one of the two has no such form and is left out of get_timer rather + than shown wrong. + """ + fields = str(cron or "").split() + if len(fields) != 5: + return None + minute, hour, dom, month, dow = (field if field != "?" else "*" for field in fields) + if (dom == "*") != (month == "*"): + return None + return f"{minute} {hour} {dom} {month} {dow}" + + +def _timer_entry(job: dict[str, Any]) -> list[Any] | None: + cron = device_cron(str(job.get("cron") or "")) + if cron is None: + return None + enabled = job.get("enabled", True) + on = enabled not in (False, 0, "0", "false", "False") + return [str(job.get("id")), "on" if on else "off", [cron, ["start_clean", {}]]] + + +def _parse_set_timer(params: Any) -> tuple[str, bool, str]: + entry = params[0] if isinstance(params, list) and params and isinstance(params[0], list) else params + if not isinstance(entry, list) or len(entry) < 2: + raise TimerRequestError("set_timer needs [id, [status, [cron, action]]]") + job_id = str(entry[0]) + if len(entry) >= 3 and isinstance(entry[1], str): + status, detail = entry[1], entry[2] + else: + spec = entry[1] + if not isinstance(spec, list) or len(spec) < 2: + raise TimerRequestError("set_timer spec must be [status, [cron, action]]") + status, detail = spec[0], spec[1] + if not isinstance(detail, list) or not detail: + raise TimerRequestError("set_timer is missing its cron") + return job_id, str(status).lower() == "on", str(detail[0]) + + +def _id_and_status(params: Any) -> tuple[str, str]: + entry = params[0] if isinstance(params, list) and params and isinstance(params[0], list) else params + if not isinstance(entry, list) or len(entry) < 2: + raise TimerRequestError("upd_timer needs [id, status]") + return str(entry[0]), str(entry[1]).lower() + + +def _single_id(params: Any) -> str: + if isinstance(params, list) and params: + first = params[0] + return str(first[0]) if isinstance(first, list) and first else str(first) + if params is None or params == []: + raise TimerRequestError("del_timer needs a timer id") + return str(params) + + +def handle_timer_request( + ctx: ServerContext, + duid: str, + method: str, + params: Any, + *, + default_timezone: str = "", +) -> Any: + """Apply one timer RPC to the store and return the RPC ``result`` value.""" + if method == "get_timer": + entries = (_timer_entry(job) for job in list_jobs(ctx, duid)) + return [entry for entry in entries if entry is not None] + + if method == "set_timer": + job_id, enabled, cron = _parse_set_timer(params) + try: + parse_cron(cron) + except CronError as exc: + raise TimerRequestError(str(exc)) from exc + zone = str(getattr(ctx, "timezone", "") or default_timezone or DEFAULT_TIMEZONE) + upsert_job(ctx, duid, job_id=job_id, cron=cron, enabled=enabled, time_zone=zone) + return list(_OK) + + if method == "upd_timer": + job_id, status = _id_and_status(params) + if status not in ("on", "off"): + raise TimerRequestError(f"unknown timer status {status!r}") + set_enabled(ctx, duid, job_id, status == "on") + return list(_OK) + + if method == "del_timer": + delete_job(ctx, duid, _single_id(params)) + return list(_OK) + + raise TimerRequestError(f"not a timer method: {method!r}") + + +class TimerRpcInterceptor: + """Bridge hook: turn an encrypted timer request into an encrypted reply, or None to forward.""" + + def __init__(self, context: ServerContext, *, default_timezone: str = "") -> None: + self._context = context + self._default_timezone = default_timezone + + def _localkey(self, duid: str) -> str: + creds = self._context.runtime_credentials + device = creds.resolve_device(duid=duid) if creds is not None else None + key = str((device or {}).get("localkey") or "").strip() + return key or str(getattr(self._context, "localkey", "") or "").strip() + + def __call__(self, duid: str, payload: bytes) -> bytes | None: + localkey = self._localkey(duid) + if not localkey: + return None + try: + messages = create_mqtt_decoder(localkey)(payload) + except Exception: # noqa: BLE001 + return None + if len(messages) != 1: + return None + message = messages[0] + protocol = int(getattr(message.protocol, "value", message.protocol)) + if protocol != RoborockMessageProtocol.RPC_REQUEST.value or not message.payload: + return None + request = _request_body(message) + if request is None or request.get("method") not in TIMER_METHODS: + return None + + request_id = request.get("id") + try: + body: dict[str, Any] = { + "id": request_id, + "result": handle_timer_request( + self._context, + duid, + str(request["method"]), + request.get("params"), + default_timezone=self._default_timezone, + ), + } + except TimerRequestError as exc: + _LOGGER.warning("Rejected %s for %s: %s", request.get("method"), duid, exc) + body = {"id": request_id, "error": {"code": -1, "message": str(exc)}} + reply = RoborockMessage( + protocol=RoborockMessageProtocol.RPC_RESPONSE, + payload=json.dumps( + { + "t": int(time.time()), + "dps": {str(RoborockMessageProtocol.RPC_RESPONSE.value): json.dumps(body)}, + }, + separators=(",", ":"), + ).encode(), + seq=message.seq, + version=message.version, + ) + return create_mqtt_encoder(localkey)(reply) + + +def _request_body(message: RoborockMessage) -> dict[str, Any] | None: + try: + outer = json.loads(message.payload) + inner = outer["dps"][str(RoborockMessageProtocol.RPC_REQUEST.value)] + body = json.loads(inner) if isinstance(inner, str) else inner + except (ValueError, KeyError, TypeError): + return None + return body if isinstance(body, dict) else None diff --git a/src/roborock_local_server/server.py b/src/roborock_local_server/server.py index 53c979aa..daefc5f3 100644 --- a/src/roborock_local_server/server.py +++ b/src/roborock_local_server/server.py @@ -54,6 +54,7 @@ from shared.inventory_io import atomic_write_inventory, inventory_transaction from shared.protocol_auth import ProtocolAuthStore from shared.schedule_runner import ScheduleRunner +from shared.timer_rpc import TimerRpcInterceptor from https_server.routes.user.scene import validity as scene_validity from https_server.routes.auth.service import ( build_login_data_response, @@ -1957,6 +1958,10 @@ async def start(self) -> None: logger=self.loggers["mqtt"], runtime_state=self.runtime_state, inventory_path=self.paths.inventory_path, + request_interceptor=TimerRpcInterceptor( + self.context, + default_timezone=self.config.network.timezone, + ), ) await self._topic_bridge.start() self.runtime_state.set_service("mqtt_topic_bridge", running=True, required=False, enabled=True) diff --git a/tests/test_timer_rpc.py b/tests/test_timer_rpc.py new file mode 100644 index 00000000..4d44b028 --- /dev/null +++ b/tests/test_timer_rpc.py @@ -0,0 +1,172 @@ +import asyncio +import json +import logging +from types import SimpleNamespace + +import pytest + +import roborock_local_server.backend # noqa: F401 (puts the bundled `shared` package on sys.path) +from roborock.protocol import create_mqtt_decoder, create_mqtt_encoder +from roborock.protocols.v1_protocol import decode_rpc_response +from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol +from roborock_local_server.bundled_backend.mqtt_broker_server.topic_bridge import ( + CloudTopicKey, + MqttTopicBridge, +) +from shared import schedule_store, timer_rpc + +DUID = "2e2txkqKqxVifcIdyOT1lY" +LOCALKEY = "AoYY9mpxTOIUZtg4" + + +def _cloud_job(job_id, cron, *, enabled=True, **extra): + job = { + "id": job_id, + "cron": cron, + "timeZoneId": "America/New_York", + "repeated": True, + "enabled": enabled, + "param": { + "id": 1, + "method": "server_scheduled_start", + "params": [{"repeat": 1, "water_box_mode": 203, "segments": "0", "fan_power": 104, "mop_mode": 300}], + }, + } + job.update(extra) + return job + + +@pytest.fixture +def ctx(tmp_path): + inventory = { + "schedules": { + DUID: [ + _cloud_job(3549431, "00 11 ? * 1,3,5"), + _cloud_job(3549434, "00 11 ? * 0,6", enabled=False), + _cloud_job(3878757, "03 13 15 12 ?", repeated=False), + ] + } + } + (tmp_path / "web_api_inventory.json").write_text(json.dumps(inventory), encoding="utf-8") + creds = SimpleNamespace(resolve_device=lambda duid="": {"localkey": LOCALKEY} if duid == DUID else None) + return SimpleNamespace( + http_jsonl=tmp_path / "http.jsonl", loggers={}, timezone="America/New_York", + runtime_credentials=creds, localkey="", + ) + + +def test_get_timer_lists_cloud_schedules_in_the_shape_localrock_parses(ctx): + result = timer_rpc.handle_timer_request(ctx, DUID, "get_timer", None) + # The one-off (Dec 15) keeps its day and month, the form the robot and the official app use. + assert result == [ + ["3549431", "on", ["00 11 * * 1,3,5", ["start_clean", {}]]], + ["3549434", "off", ["00 11 * * 0,6", ["start_clean", {}]]], + ["3878757", "on", ["03 13 15 12 *", ["start_clean", {}]]], + ] + + +@pytest.mark.parametrize( + ("cron", "expected"), + [ + ("15 23 18 10 ?", "15 23 18 10 *"), + ("15 23 18 10 *", "15 23 18 10 *"), + ("00 11 ? * 1,3,5", "00 11 * * 1,3,5"), + ("0 8 * * 0,1,2,3,4,5,6", "0 8 * * 0,1,2,3,4,5,6"), + # Only one of day and month pinned has no one-time form. + ("0 8 15 * *", None), + ("0 8 * 10 *", None), + ("not a cron", None), + ], +) +def test_device_cron_keeps_a_one_time_schedules_date(cron, expected): + assert timer_rpc.device_cron(cron) == expected + + +def test_set_timer_creates_a_job_and_get_timer_returns_it(ctx): + params = [["1790000000", ["on", ["30 7 * * 1,2", ["start_clean", {}]]]]] + assert timer_rpc.handle_timer_request(ctx, DUID, "set_timer", params) == ["ok"] + + job = next(j for j in schedule_store.list_jobs(ctx, DUID) if str(j["id"]) == "1790000000") + assert job["cron"] == "30 7 * * 1,2" + assert job["enabled"] is True + assert job["timeZoneId"] == "America/New_York" + assert job["param"]["method"] == "server_scheduled_start" + + listed = timer_rpc.handle_timer_request(ctx, DUID, "get_timer", None) + assert ["1790000000", "on", ["30 7 * * 1,2", ["start_clean", {}]]] in listed + + +def test_editing_a_cloud_schedule_keeps_its_vac_and_mop_settings(ctx): + params = [["3549431", ["on", ["15 9 * * 1,3,5", ["start_clean", {}]]]]] + timer_rpc.handle_timer_request(ctx, DUID, "set_timer", params) + + job = next(j for j in schedule_store.list_jobs(ctx, DUID) if j["id"] == 3549431) + assert job["cron"] == "15 9 * * 1,3,5" + entry = job["param"]["params"][0] + assert (entry["fan_power"], entry["water_box_mode"], entry["mop_mode"]) == (104, 203, 300) + + +def test_upd_timer_toggles_and_del_timer_removes_a_cloud_schedule(ctx): + assert timer_rpc.handle_timer_request(ctx, DUID, "upd_timer", [["3549431", "off"]]) == ["ok"] + job = next(j for j in schedule_store.list_jobs(ctx, DUID) if j["id"] == 3549431) + assert job["enabled"] is False + + assert timer_rpc.handle_timer_request(ctx, DUID, "del_timer", ["3549431"]) == ["ok"] + assert all(j["id"] != 3549431 for j in schedule_store.list_jobs(ctx, DUID)) + + +def test_set_timer_rejects_an_unparseable_cron(ctx): + with pytest.raises(timer_rpc.TimerRequestError): + timer_rpc.handle_timer_request(ctx, DUID, "set_timer", [["1", ["on", ["not a cron", []]]]]) + + +def _encode_request(method, params, request_id=4242): + payload = json.dumps( + {"t": 1790000000, "dps": {"101": json.dumps({"id": request_id, "method": method, "params": params})}} + ).encode() + message = RoborockMessage(protocol=RoborockMessageProtocol.RPC_REQUEST, payload=payload) + return create_mqtt_encoder(LOCALKEY)(message) + + +def _decode_reply(reply): + (message,) = create_mqtt_decoder(LOCALKEY)(reply) + return decode_rpc_response(message) + + +def test_interceptor_answers_get_timer_with_an_encrypted_rpc_response(ctx): + reply = timer_rpc.TimerRpcInterceptor(ctx)(DUID, _encode_request("get_timer", [])) + assert reply is not None + response = _decode_reply(reply) + assert response.request_id == 4242 + assert [entry[0] for entry in response.data] == ["3549431", "3549434", "3878757"] + + +def test_interceptor_forwards_everything_that_is_not_a_timer_request(ctx): + interceptor = timer_rpc.TimerRpcInterceptor(ctx) + assert interceptor(DUID, _encode_request("get_status", [])) is None + assert interceptor(DUID, b"not a roborock frame") is None + assert interceptor("unknown-duid", _encode_request("get_timer", [])) is None + + +def test_bridge_answers_timer_requests_locally_and_does_not_forward_them(ctx): + class Client: + def __init__(self): + self.published = [] + + async def publish(self, topic, payload, *, qos=0, retain=False): + self.published.append((topic, payload)) + + bridge = MqttTopicBridge( + host="127.0.0.1", + port=1883, + logger=logging.getLogger("test.timer_bridge"), + request_interceptor=timer_rpc.TimerRpcInterceptor(ctx), + ) + cloud = CloudTopicKey(rriot_u="u", mqtt_username="m", duid=DUID) + message = SimpleNamespace(payload=_encode_request("get_timer", []), qos=0, retain=False) + client = Client() + + asyncio.run(bridge._handle_cloud_message(client, message, cloud, cloud.topic_in)) + + assert [topic for topic, _ in client.published] == [cloud.topic_out] + assert _decode_reply(client.published[0][1]).request_id == 4242 From 87669e95672d93753daa25b105ac4cd89fe25a93 Mon Sep 17 00:00:00 2001 From: Colin Haven <10933733+virtitnerd@users.noreply.github.com> Date: Mon, 12 Oct 2026 16:21:40 -0400 Subject: [PATCH 06/11] Accept schedule create, update and delete over HTTP POST /user/devices/{duid}/jobs creates a job (the call python-roborock makes), and PUT/PATCH and DELETE on /jobs/{id} edit or remove one. The GET matcher is now method-aware so a write no longer falls through to the list route. Update and delete paths are not confirmed against the official app; they follow the scene routes' conventions. Co-Authored-By: Claude Sonnet 5.5 (cherry picked from commit 5b838531eaebe34fe369b59e511f9aab1f848d29) --- .../https_server/endpoint_rules.py | 9 ++ .../https_server/routes/user/devices/jobs.py | 126 +++++++++++++++++- tests/test_schedules.py | 87 ++++++++++++ 3 files changed, 219 insertions(+), 3 deletions(-) diff --git a/src/roborock_local_server/bundled_backend/https_server/endpoint_rules.py b/src/roborock_local_server/bundled_backend/https_server/endpoint_rules.py index 7aa3c233..e616716b 100644 --- a/src/roborock_local_server/bundled_backend/https_server/endpoint_rules.py +++ b/src/roborock_local_server/bundled_backend/https_server/endpoint_rules.py @@ -114,7 +114,13 @@ from .routes.user.deviceshare import match_received_devices as _match_get_received_devices from .routes.user.deviceshare import match_rooms as _match_get_shared_device_rooms from .routes.user.devices.jobs import build as _build_get_schedules +from .routes.user.devices.jobs import build_create as _build_create_schedule +from .routes.user.devices.jobs import build_delete as _build_delete_schedule +from .routes.user.devices.jobs import build_update as _build_update_schedule from .routes.user.devices.jobs import match as _match_get_schedules +from .routes.user.devices.jobs import match_create as _match_create_schedule +from .routes.user.devices.jobs import match_delete as _match_delete_schedule +from .routes.user.devices.jobs import match_update as _match_update_schedule from .routes.user.devices.newadd import build as _build_add_device from .routes.user.devices.newadd import match as _match_add_device from .routes.user.homes.item import build as _build_get_home_data @@ -596,6 +602,9 @@ def default_endpoint_rules() -> Sequence[EndpointRule]: EndpointRule("get_device", _match_get_device, _build_get_device), EndpointRule("get_device_extra", _match_get_device_extra, _build_get_device_extra), EndpointRule("get_schedules", _match_get_schedules, _build_get_schedules), + EndpointRule("create_schedule", _match_create_schedule, _build_create_schedule), + EndpointRule("update_schedule", _match_update_schedule, _build_update_schedule), + EndpointRule("delete_schedule", _match_delete_schedule, _build_delete_schedule), EndpointRule("post_app_info", _match_post_app_info, _build_post_app_info), EndpointRule("get_inbox_latest", _match_get_inbox_latest, _build_get_inbox_latest), EndpointRule("get_products_v4", _match_get_products_v4, _build_get_products_v4), diff --git a/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/jobs.py b/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/jobs.py index d0ab3239..1aa3ef82 100644 --- a/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/jobs.py +++ b/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/jobs.py @@ -1,17 +1,25 @@ from __future__ import annotations +import json import re +import time from typing import Any +from shared.constants import DEFAULT_TIMEZONE from shared.context import ServerContext +from shared.data_helpers import as_bool from shared.http_helpers import wrap_response +from shared.schedule_runner import CronError, parse_cron +from shared.schedule_store import delete_job, list_jobs, upsert_job from .service import device_jobs_payload +_COLLECTION_RE = re.compile(r"/user/devices/([^/]+)/jobs") +_ITEM_RE = re.compile(r"/user/devices/([^/]+)/jobs/([^/]+)") -def match(path: str) -> bool: - clean = path.rstrip("/") - return bool(re.fullmatch(r"/user/devices/[^/]+/jobs", clean)) + +def match(path: str, method: str = "GET") -> bool: + return method.upper() == "GET" and bool(_COLLECTION_RE.fullmatch(path.rstrip("/"))) def build( @@ -24,3 +32,115 @@ def build( device_id = parts[-2] if len(parts) >= 2 else "" return wrap_response(device_jobs_payload(ctx, device_id)) + +def _error(code: int, message: str) -> dict[str, Any]: + return {"success": False, "code": code, "msg": message, "data": None, "result": None} + + +def _json_body(body_params: dict[str, list[str]]) -> dict[str, Any]: + for raw in body_params.get("__json") or []: + try: + parsed = json.loads(raw) + except (TypeError, json.JSONDecodeError): + continue + if isinstance(parsed, dict): + return parsed + return {} + + +def _parse_param(value: Any) -> dict[str, Any] | None: + if isinstance(value, str) and value.strip(): + try: + value = json.loads(value) + except json.JSONDecodeError: + return None + return value if isinstance(value, dict) else None + + +def match_create(path: str, method: str = "GET") -> bool: + return method.upper() == "POST" and bool(_COLLECTION_RE.fullmatch(path.rstrip("/"))) + + +def build_create( + ctx: ServerContext, + _query_params: dict[str, list[str]], + body_params: dict[str, list[str]], + clean_path: str, +) -> dict[str, Any]: + duid = _COLLECTION_RE.fullmatch(clean_path.rstrip("/")).group(1) # type: ignore[union-attr] + body = _json_body(body_params) + cron = str(body.get("cron") or "").strip() + try: + parse_cron(cron) + except CronError as exc: + return _error(400, f"invalid cron: {exc}") + param = _parse_param(body.get("param")) + if body.get("param") is not None and param is None: + return _error(400, "param must be a JSON object") + job_id = body.get("id") or int(time.time()) + created = upsert_job( + ctx, + duid, + job_id=job_id, + cron=cron, + enabled=as_bool(body.get("enabled", True), True), + time_zone=str(body.get("timeZoneId") or getattr(ctx, "timezone", "") or DEFAULT_TIMEZONE), + repeated=as_bool(body.get("repeated", True), True), + param=param, + ) + return wrap_response(created) + + +def match_update(path: str, method: str = "GET") -> bool: + return method.upper() in ("PUT", "PATCH") and bool(_ITEM_RE.fullmatch(path.rstrip("/"))) + + +def build_update( + ctx: ServerContext, + _query_params: dict[str, list[str]], + body_params: dict[str, list[str]], + clean_path: str, +) -> dict[str, Any]: + match_ = _ITEM_RE.fullmatch(clean_path.rstrip("/")) + duid, job_id = match_.group(1), match_.group(2) # type: ignore[union-attr] + existing = next((job for job in list_jobs(ctx, duid) if str(job.get("id")) == job_id), None) + if existing is None: + return _error(404, "schedule not found") + body = _json_body(body_params) + cron = str(body.get("cron") or existing.get("cron") or "").strip() + try: + parse_cron(cron) + except CronError as exc: + return _error(400, f"invalid cron: {exc}") + param = _parse_param(body.get("param")) if "param" in body else None + if "param" in body and param is None: + return _error(400, "param must be a JSON object") + updated = upsert_job( + ctx, + duid, + job_id=job_id, + cron=cron, + enabled=as_bool(body.get("enabled", existing.get("enabled", True)), True), + time_zone=str(body.get("timeZoneId") or ""), + repeated=as_bool(body["repeated"], True) if "repeated" in body else None, + param=param, + override_time_zone=bool(body.get("timeZoneId")), + ) + return wrap_response(updated) + + +def match_delete(path: str, method: str = "GET") -> bool: + return method.upper() == "DELETE" and bool(_ITEM_RE.fullmatch(path.rstrip("/"))) + + +def build_delete( + ctx: ServerContext, + _query_params: dict[str, list[str]], + _body_params: dict[str, list[str]], + clean_path: str, +) -> dict[str, Any]: + match_ = _ITEM_RE.fullmatch(clean_path.rstrip("/")) + duid, job_id = match_.group(1), match_.group(2) # type: ignore[union-attr] + if not delete_job(ctx, duid, job_id): + return _error(404, "schedule not found") + return wrap_response({"id": job_id}) diff --git a/tests/test_schedules.py b/tests/test_schedules.py index e4a9ced3..f362a6ab 100644 --- a/tests/test_schedules.py +++ b/tests/test_schedules.py @@ -166,3 +166,90 @@ def test_cloud_inventory_build_preserves_device_schedules() -> None: "6HL2zfniaoYYV01CkVuhkO": [schedule], "ignored": [], } + + +def _jobs_client(tmp_path: Path, device_id: str, schedules: list[dict[str, object]]): + config_file = write_release_config(tmp_path) + config = load_config(config_file) + paths = resolve_paths(config_file, config) + _write_json( + paths.inventory_path, + { + "home": {"id": 1233716, "name": "My Home"}, + "devices": [{"duid": device_id, "name": "Roborock Qrevo"}], + "schedules": {device_id: schedules}, + }, + ) + _seed_cloud_snapshot(paths.cloud_snapshot_path) + supervisor = ReleaseSupervisor(config=config, paths=paths) + supervisor.refresh_inventory_state() + return TestClient(supervisor.app), paths + + +def _signed_json(paths, path: str, body: dict[str, object]) -> tuple[bytes, dict[str, str]]: + raw = json.dumps(body, separators=(",", ":")).encode() + user = ProtocolAuthStore(paths.cloud_snapshot_path).availability().user + assert user is not None + headers = { + # Hawk rejects replayed nonces, and _hawk_headers() uses nonce- for plain requests. + "Authorization": build_hawk_authorization( + user=user, path=path, json_body=raw, nonce=f"write-{path.replace('/', '-')}-{len(raw)}" + ), + "Content-Type": "application/json", + } + return raw, headers + + +def test_jobs_can_be_created_updated_and_deleted_over_http(tmp_path: Path) -> None: + device_id = "6HL2zfniaoYYV01CkVuhkO" + existing = { + "id": 3549431, + "cron": "00 11 ? * 1,3,5", + "timeZoneId": "America/New_York", + "repeated": True, + "enabled": True, + "param": { + "id": 1, + "method": "server_scheduled_start", + "params": [{"repeat": 1, "water_box_mode": 203, "segments": "0", "fan_power": 104, "mop_mode": 300}], + }, + } + client, paths = _jobs_client(tmp_path, device_id, [existing]) + collection = f"/user/devices/{device_id}/jobs" + + body, headers = _signed_json(paths, collection, {"cron": "30 7 * * 1,2", "enabled": True}) + created = client.post(collection, content=body, headers=headers) + assert created.status_code == 200 + assert created.json()["success"] is True + new_id = created.json()["result"]["id"] + + listed = client.get(collection, headers=_hawk_headers(paths.cloud_snapshot_path, collection)).json()["result"] + assert {job["id"] for job in listed} == {3549431, new_id} + assert all(job.get("nextFireTime") for job in listed) + + item = f"{collection}/3549431" + body, headers = _signed_json(paths, item, {"cron": "15 9 ? * 1,3,5", "enabled": False}) + updated = client.put(item, content=body, headers=headers) + assert updated.status_code == 200 + result = updated.json()["result"] + assert (result["cron"], result["enabled"]) == ("15 9 ? * 1,3,5", False) + # Vac/mop settings are untouched by a cron/enabled edit. + assert result["param"]["params"][0]["mop_mode"] == 300 + + deleted = client.delete(item, headers=_hawk_headers(paths.cloud_snapshot_path, item)) + assert deleted.status_code == 200 and deleted.json()["success"] is True + user = ProtocolAuthStore(paths.cloud_snapshot_path).availability().user + assert user is not None + second_list = {"Authorization": build_hawk_authorization(user=user, path=collection, nonce="second-list")} + remaining = client.get(collection, headers=second_list).json()["result"] + assert [job["id"] for job in remaining] == [new_id] + + +def test_creating_a_job_with_a_bad_cron_is_rejected(tmp_path: Path) -> None: + device_id = "6HL2zfniaoYYV01CkVuhkO" + client, paths = _jobs_client(tmp_path, device_id, []) + collection = f"/user/devices/{device_id}/jobs" + body, headers = _signed_json(paths, collection, {"cron": "nonsense"}) + response = client.post(collection, content=body, headers=headers) + assert response.json()["success"] is False + assert response.json()["code"] == 400 From 27e20f0afd921fc4c058b01b7ad195c9df4a8028 Mon Sep 17 00:00:00 2001 From: Colin Haven <10933733+virtitnerd@users.noreply.github.com> Date: Mon, 12 Oct 2026 17:17:21 -0400 Subject: [PATCH 07/11] Let timer RPCs carry per-schedule vac and mop settings The legacy set_timer action is ["start_clean", {}]. The object now accepts repeat, fan_power, water_box_mode, mop_mode, mop_template_id, segments and clean_mop, and get_timer returns them, so a client can show and change them. Keys are merged into the job's existing settings, null clears one, and an empty object (what current clients send) leaves the job alone, so editing a schedule's time never resets its vac/mop choice. Co-Authored-By: Claude Sonnet 5.5 (cherry picked from commit dd2d3d16a6c2ca649b57a28623b6077e21fee159) --- .../bundled_backend/shared/schedule_store.py | 62 ++++++++++++++++ .../bundled_backend/shared/timer_rpc.py | 33 +++++++-- tests/test_timer_rpc.py | 74 ++++++++++++++++++- 3 files changed, 159 insertions(+), 10 deletions(-) diff --git a/src/roborock_local_server/bundled_backend/shared/schedule_store.py b/src/roborock_local_server/bundled_backend/shared/schedule_store.py index 1d762fcc..f4bdd5af 100644 --- a/src/roborock_local_server/bundled_backend/shared/schedule_store.py +++ b/src/roborock_local_server/bundled_backend/shared/schedule_store.py @@ -17,6 +17,68 @@ } +JOB_SETTING_KEYS = ( + "repeat", + "fan_power", + "water_box_mode", + "mop_mode", + "mop_template_id", + "segments", + "clean_mop", +) +_INT_SETTING_KEYS = frozenset(JOB_SETTING_KEYS) - {"segments"} + + +class SettingsError(ValueError): + """A schedule setting had a value this server cannot store.""" + + +def _normalize_setting(key: str, value: Any) -> Any: + if key == "segments": + items = value.replace(";", ",").split(",") if isinstance(value, str) else list(value) + try: + ids = [int(str(item).strip()) for item in items if str(item).strip()] + except ValueError as exc: + raise SettingsError(f"segments must be room ids, got {value!r}") from exc + ids = [room for room in ids if room > 0] + return ",".join(str(room) for room in ids) if ids else "0" + if isinstance(value, bool) or not isinstance(value, (int, str)): + raise SettingsError(f"{key} must be a number, got {value!r}") + try: + return int(value) + except ValueError as exc: + raise SettingsError(f"{key} must be a number, got {value!r}") from exc + + +def job_settings(job: dict[str, Any]) -> dict[str, Any]: + """The vac/mop settings a job carries, as a flat dict.""" + param = job.get("param") + params = param.get("params") if isinstance(param, dict) else None + entry = params[0] if isinstance(params, list) and params and isinstance(params[0], dict) else {} + return {key: entry[key] for key in JOB_SETTING_KEYS if key in entry} + + +def merge_job_settings(param: dict[str, Any] | None, settings: dict[str, Any]) -> dict[str, Any]: + """Copy of ``param`` with ``settings`` merged in; a None value clears that setting. + + Settings not mentioned are left alone, so editing one option never resets the rest. + """ + merged = copy.deepcopy(param) if isinstance(param, dict) else copy.deepcopy(DEFAULT_JOB_PARAM) + params = merged.get("params") + if not isinstance(params, list) or not params or not isinstance(params[0], dict): + params = copy.deepcopy(DEFAULT_JOB_PARAM["params"]) + merged["params"] = params + entry = params[0] + for key, value in settings.items(): + if key not in JOB_SETTING_KEYS: + raise SettingsError(f"unknown schedule setting {key!r}") + if value is None: + entry.pop(key, None) + else: + entry[key] = _normalize_setting(key, value) + return merged + + def _same_id(job: Any, job_id: Any) -> bool: return isinstance(job, dict) and str(job.get("id")) == str(job_id) diff --git a/src/roborock_local_server/bundled_backend/shared/timer_rpc.py b/src/roborock_local_server/bundled_backend/shared/timer_rpc.py index f7c867d4..5bbe9aed 100644 --- a/src/roborock_local_server/bundled_backend/shared/timer_rpc.py +++ b/src/roborock_local_server/bundled_backend/shared/timer_rpc.py @@ -18,7 +18,16 @@ from .constants import DEFAULT_TIMEZONE from .schedule_runner import CronError, parse_cron -from .schedule_store import delete_job, list_jobs, set_enabled, upsert_job +from .schedule_store import ( + JOB_SETTING_KEYS, + SettingsError, + delete_job, + job_settings, + list_jobs, + merge_job_settings, + set_enabled, + upsert_job, +) if TYPE_CHECKING: from .context import ServerContext @@ -56,10 +65,10 @@ def _timer_entry(job: dict[str, Any]) -> list[Any] | None: return None enabled = job.get("enabled", True) on = enabled not in (False, 0, "0", "false", "False") - return [str(job.get("id")), "on" if on else "off", [cron, ["start_clean", {}]]] + return [str(job.get("id")), "on" if on else "off", [cron, ["start_clean", job_settings(job)]]] -def _parse_set_timer(params: Any) -> tuple[str, bool, str]: +def _parse_set_timer(params: Any) -> tuple[str, bool, str, dict[str, Any]]: entry = params[0] if isinstance(params, list) and params and isinstance(params[0], list) else params if not isinstance(entry, list) or len(entry) < 2: raise TimerRequestError("set_timer needs [id, [status, [cron, action]]]") @@ -73,7 +82,9 @@ def _parse_set_timer(params: Any) -> tuple[str, bool, str]: status, detail = spec[0], spec[1] if not isinstance(detail, list) or not detail: raise TimerRequestError("set_timer is missing its cron") - return job_id, str(status).lower() == "on", str(detail[0]) + action = detail[1] if len(detail) > 1 else None + settings = action[1] if isinstance(action, list) and len(action) > 1 and isinstance(action[1], dict) else {} + return job_id, str(status).lower() == "on", str(detail[0]), settings def _id_and_status(params: Any) -> tuple[str, str]: @@ -106,13 +117,23 @@ def handle_timer_request( return [entry for entry in entries if entry is not None] if method == "set_timer": - job_id, enabled, cron = _parse_set_timer(params) + job_id, enabled, cron, action_settings = _parse_set_timer(params) try: parse_cron(cron) except CronError as exc: raise TimerRequestError(str(exc)) from exc + # An empty action object (what older clients send) leaves the job's settings alone; + # keys it does carry are merged in, so a client can change one option at a time. + settings = {key: value for key, value in action_settings.items() if key in JOB_SETTING_KEYS} + param = None + if settings: + existing = next((job for job in list_jobs(ctx, duid) if str(job.get("id")) == job_id), None) + try: + param = merge_job_settings(existing.get("param") if existing else None, settings) + except SettingsError as exc: + raise TimerRequestError(str(exc)) from exc zone = str(getattr(ctx, "timezone", "") or default_timezone or DEFAULT_TIMEZONE) - upsert_job(ctx, duid, job_id=job_id, cron=cron, enabled=enabled, time_zone=zone) + upsert_job(ctx, duid, job_id=job_id, cron=cron, enabled=enabled, time_zone=zone, param=param) return list(_OK) if method == "upd_timer": diff --git a/tests/test_timer_rpc.py b/tests/test_timer_rpc.py index 4d44b028..7198bf82 100644 --- a/tests/test_timer_rpc.py +++ b/tests/test_timer_rpc.py @@ -58,10 +58,11 @@ def ctx(tmp_path): def test_get_timer_lists_cloud_schedules_in_the_shape_localrock_parses(ctx): result = timer_rpc.handle_timer_request(ctx, DUID, "get_timer", None) # The one-off (Dec 15) keeps its day and month, the form the robot and the official app use. + settings = {"repeat": 1, "water_box_mode": 203, "segments": "0", "fan_power": 104, "mop_mode": 300} assert result == [ - ["3549431", "on", ["00 11 * * 1,3,5", ["start_clean", {}]]], - ["3549434", "off", ["00 11 * * 0,6", ["start_clean", {}]]], - ["3878757", "on", ["03 13 15 12 *", ["start_clean", {}]]], + ["3549431", "on", ["00 11 * * 1,3,5", ["start_clean", settings]]], + ["3549434", "off", ["00 11 * * 0,6", ["start_clean", settings]]], + ["3878757", "on", ["03 13 15 12 *", ["start_clean", settings]]], ] @@ -93,7 +94,8 @@ def test_set_timer_creates_a_job_and_get_timer_returns_it(ctx): assert job["param"]["method"] == "server_scheduled_start" listed = timer_rpc.handle_timer_request(ctx, DUID, "get_timer", None) - assert ["1790000000", "on", ["30 7 * * 1,2", ["start_clean", {}]]] in listed + default_settings = {"repeat": 1, "segments": "0"} + assert ["1790000000", "on", ["30 7 * * 1,2", ["start_clean", default_settings]]] in listed def test_editing_a_cloud_schedule_keeps_its_vac_and_mop_settings(ctx): @@ -170,3 +172,67 @@ async def publish(self, topic, payload, *, qos=0, retain=False): assert [topic for topic, _ in client.published] == [cloud.topic_out] assert _decode_reply(client.published[0][1]).request_id == 4242 + + +def _job(ctx, job_id): + return next(j for j in schedule_store.list_jobs(ctx, DUID) if str(j["id"]) == str(job_id)) + + +def test_set_timer_carries_vac_and_mop_settings_in_the_action_object(ctx): + action = ["start_clean", {"fan_power": 102, "water_box_mode": 200, "repeat": 2}] + params = [["1790000001", ["on", ["0 9 * * 2,4", action]]]] + assert timer_rpc.handle_timer_request(ctx, DUID, "set_timer", params) == ["ok"] + + entry = _job(ctx, 1790000001)["param"]["params"][0] + assert (entry["fan_power"], entry["water_box_mode"], entry["repeat"]) == (102, 200, 2) + + listed = timer_rpc.handle_timer_request(ctx, DUID, "get_timer", None) + row = next(item for item in listed if item[0] == "1790000001") + assert row[2][1][1]["water_box_mode"] == 200 + + +def test_changing_one_setting_leaves_the_others_alone(ctx): + action = ["start_clean", {"water_box_mode": 200}] # vac only on this cloud schedule + timer_rpc.handle_timer_request(ctx, DUID, "set_timer", [["3549431", ["on", ["00 11 * * 1,3,5", action]]]]) + + entry = _job(ctx, 3549431)["param"]["params"][0] + assert entry["water_box_mode"] == 200 + assert (entry["fan_power"], entry["mop_mode"]) == (104, 300) + + +def test_an_empty_action_object_does_not_reset_existing_settings(ctx): + # What older LocalRock builds send: ["start_clean", {}]. + action = ["start_clean", {}] + timer_rpc.handle_timer_request(ctx, DUID, "set_timer", [["3549431", ["on", ["30 8 * * 1,3,5", action]]]]) + + job = _job(ctx, 3549431) + assert job["cron"] == "30 8 * * 1,3,5" + assert job["param"]["params"][0]["water_box_mode"] == 203 + + +def test_null_clears_a_setting_and_segments_are_normalised(ctx): + action = ["start_clean", {"mop_mode": None, "segments": [16, 17]}] + timer_rpc.handle_timer_request(ctx, DUID, "set_timer", [["3549431", ["on", ["00 11 * * 1,3,5", action]]]]) + + entry = _job(ctx, 3549431)["param"]["params"][0] + assert "mop_mode" not in entry + assert entry["segments"] == "16,17" + + +def test_vac_only_schedule_fires_without_a_mop_template(ctx): + from shared import schedule_runner + + action = ["start_clean", {"water_box_mode": 200, "fan_power": 102}] + timer_rpc.handle_timer_request(ctx, DUID, "set_timer", [["1790000002", ["on", ["0 9 * * *", action]]]]) + + (step,) = schedule_runner.steps_for_job(_job(ctx, 1790000002)) + assert step.params[0]["water_box_mode"] == 200 + assert step.params[0]["fan_power"] == 102 + + +def test_non_numeric_setting_is_rejected_and_nothing_is_saved(ctx): + before = schedule_store.list_jobs(ctx, DUID) + action = ["start_clean", {"fan_power": "loud"}] + with pytest.raises(timer_rpc.TimerRequestError): + timer_rpc.handle_timer_request(ctx, DUID, "set_timer", [["9", ["on", ["0 9 * * *", action]]]]) + assert schedule_store.list_jobs(ctx, DUID) == before From e32d5fb47e1ba8125e49dd04cb0aa08904877217 Mon Sep 17 00:00:00 2001 From: Colin Haven <10933733+virtitnerd@users.noreply.github.com> Date: Sat, 17 Oct 2026 22:33:26 -0400 Subject: [PATCH 08/11] Leave server_timer to the robot, and guard it with a test A robot that speaks server_timer stores those schedules itself, and the official app reads them back from it. Answering get/set/upd/del_server_timer from this store instead hides the robot's real schedules from every client reading through the server and, tried once, caused real data loss: the official app deletes server_timer entries it does not recognize when its schedule screen opens. Adds a parametrized test over all four server_timer methods so the interceptor's method list can't regress into handling them again without a test catching it. Co-Authored-By: Claude Sonnet 5.5 --- .../bundled_backend/shared/timer_rpc.py | 12 ++++++++++++ tests/test_timer_rpc.py | 11 +++++++++++ 2 files changed, 23 insertions(+) diff --git a/src/roborock_local_server/bundled_backend/shared/timer_rpc.py b/src/roborock_local_server/bundled_backend/shared/timer_rpc.py index 5bbe9aed..3d93ef3c 100644 --- a/src/roborock_local_server/bundled_backend/shared/timer_rpc.py +++ b/src/roborock_local_server/bundled_backend/shared/timer_rpc.py @@ -4,6 +4,18 @@ clients such as LocalRock that ask the robot for its timers would otherwise see nothing and could never change the schedules the cloud imported. The topic bridge hands matching requests here instead of forwarding them to the robot. + +Deliberately not handled here: get/set/upd/del_server_timer, the newer sibling +protocol some firmware speaks instead of (or alongside) this one. The robot +stores those entries itself and the official app reads them back from it, so +they must always forward to the robot untouched. Answering them from this store +instead, tried once, hid the robot's real schedules from every client reading +through the server, and the official app then deleted them (it removes +server_timer entries it does not recognize when its schedule screen opens). + +What runs a server_timer schedule is the cloud's own record of it, the job the +official app also writes to /user/devices/{duid}/jobs, which ScheduleRunner +fires like any other. The robot was not seen to run them itself. """ from __future__ import annotations diff --git a/tests/test_timer_rpc.py b/tests/test_timer_rpc.py index 7198bf82..c187caf7 100644 --- a/tests/test_timer_rpc.py +++ b/tests/test_timer_rpc.py @@ -150,6 +150,17 @@ def test_interceptor_forwards_everything_that_is_not_a_timer_request(ctx): assert interceptor("unknown-duid", _encode_request("get_timer", [])) is None +@pytest.mark.parametrize( + "method", ["get_server_timer", "set_server_timer", "upd_server_timer", "del_server_timer"] +) +def test_interceptor_leaves_server_timer_to_the_robot(ctx, method): + # A robot that speaks server_timer stores those entries itself and the official app reads + # them back from it; answering this locally hides the robot's real schedules from every + # client and has caused real data loss. This must always forward, never answer here. + interceptor = timer_rpc.TimerRpcInterceptor(ctx) + assert interceptor(DUID, _encode_request(method, [])) is None + + def test_bridge_answers_timer_requests_locally_and_does_not_forward_them(ctx): class Client: def __init__(self): From c8ada3b56cd8d0f616bb3411b1341240e0c0e2ab Mon Sep 17 00:00:00 2001 From: Colin Haven <10933733+virtitnerd@users.noreply.github.com> Date: Sun, 18 Oct 2026 10:08:18 -0400 Subject: [PATCH 09/11] Keep the official app's link to its server_timer entry on a schedule The official app writes every server_timer schedule to the robot and also to POST/PUT /user/devices/{duid}/jobs, and that record's settings carry a "name" field holding the robot-side server_timer id. job_settings dropped it, since it is not one of the vac/mop settings, so a client reading the schedule back through get_timer had no way to find the robot's entry to edit or delete. It now rides along with the settings. Co-Authored-By: Claude Sonnet 5.5 --- .../bundled_backend/shared/schedule_store.py | 15 +++++++++++-- tests/test_schedule_store.py | 21 +++++++++++++++++++ 2 files changed, 34 insertions(+), 2 deletions(-) create mode 100644 tests/test_schedule_store.py diff --git a/src/roborock_local_server/bundled_backend/shared/schedule_store.py b/src/roborock_local_server/bundled_backend/shared/schedule_store.py index f4bdd5af..8426e54e 100644 --- a/src/roborock_local_server/bundled_backend/shared/schedule_store.py +++ b/src/roborock_local_server/bundled_backend/shared/schedule_store.py @@ -51,11 +51,22 @@ def _normalize_setting(key: str, value: Any) -> Any: def job_settings(job: dict[str, Any]) -> dict[str, Any]: - """The vac/mop settings a job carries, as a flat dict.""" + """The vac/mop settings a job carries, as a flat dict. + + ``name``, when the job has one, rides along too, outside ``JOB_SETTING_KEYS``: it is not a + vac/mop setting a client edits, it is the official app's own link from this job to the + robot's server_timer entry it saved alongside it (POST/PUT /user/devices/{duid}/jobs carries + it in exactly this spot, confirmed live). A client that reads the job back through get_timer + needs it to find the robot's entry to edit or delete. + """ param = job.get("param") params = param.get("params") if isinstance(param, dict) else None entry = params[0] if isinstance(params, list) and params and isinstance(params[0], dict) else {} - return {key: entry[key] for key in JOB_SETTING_KEYS if key in entry} + settings = {key: entry[key] for key in JOB_SETTING_KEYS if key in entry} + name = entry.get("name") + if isinstance(name, str) and name: + settings["name"] = name + return settings def merge_job_settings(param: dict[str, Any] | None, settings: dict[str, Any]) -> dict[str, Any]: diff --git a/tests/test_schedule_store.py b/tests/test_schedule_store.py new file mode 100644 index 00000000..85d00ca5 --- /dev/null +++ b/tests/test_schedule_store.py @@ -0,0 +1,21 @@ +import roborock_local_server.backend # noqa: F401 (puts the bundled `shared` package on sys.path) +from shared.schedule_store import job_settings + + +def _job(**entry_overrides): + entry = {"repeat": 1, "fan_power": 104, "water_box_mode": 203} + entry.update(entry_overrides) + return {"id": "1", "enabled": True, "param": {"params": [entry]}} + + +def test_job_settings_carries_name_when_present(): + # "name" is the official app's link to the robot's server_timer entry, not a JOB_SETTING_KEYS + # vac/mop setting, confirmed live in its POST/PUT /user/devices/{duid}/jobs body. + job = _job(name="1759965417864") + assert job_settings(job) == {"repeat": 1, "fan_power": 104, "water_box_mode": 203, "name": "1759965417864"} + + +def test_job_settings_omits_name_when_absent_or_blank(): + assert "name" not in job_settings(_job()) + assert "name" not in job_settings(_job(name="")) + assert "name" not in job_settings(_job(name=None)) From 59bf2e6de0a0f62b371ac58a41fb44bec5ddc603 Mon Sep 17 00:00:00 2001 From: Colin Haven <10933733+virtitnerd@users.noreply.github.com> Date: Sun, 18 Oct 2026 13:12:00 -0400 Subject: [PATCH 10/11] Run a schedule the way the cloud does: ask the robot to start its timer A schedule's job (POST /user/devices/{duid}/jobs) carries a settings entry whose "name" is the id of the matching timer the official app also saved on the robot. The cloud runs such a job by sending server_scheduled_start with that entry, which is the method name inside the job itself. The robot accepts it, runs the clean (rooms or whole home) and records it as "Schedules" in its history; without the entry it answers -10005 "Invalid timer", and with a bare id "Param should be name + fan + segment". python-roborock does not list the command, so it is sent through a small SchedulerCommand enum. A schedule with no robot-side id (one saved through the legacy set_timer), or one the robot refuses, falls back to the plainer ways, since a refused command changes nothing: a whole-home clean is one app_start carrying the schedule's fields plus source 3, which the robot also records as "Schedules"; a room schedule is the settings commands followed by a segment clean. Scenes and routines are untouched. The new steps are produced only by schedules. Co-Authored-By: Claude Sonnet 5.5 --- .../bundled_backend/shared/routine_runner.py | 118 +++++++++++-- .../bundled_backend/shared/schedule_runner.py | 28 ++- tests/test_routine_runner.py | 162 +++++++++++++++++- tests/test_schedule_runner.py | 34 +++- 4 files changed, 318 insertions(+), 24 deletions(-) diff --git a/src/roborock_local_server/bundled_backend/shared/routine_runner.py b/src/roborock_local_server/bundled_backend/shared/routine_runner.py index 460ed5e0..2c39c2e2 100644 --- a/src/roborock_local_server/bundled_backend/shared/routine_runner.py +++ b/src/roborock_local_server/bundled_backend/shared/routine_runner.py @@ -3,7 +3,8 @@ from __future__ import annotations import asyncio -from dataclasses import dataclass +from dataclasses import dataclass, replace +from enum import StrEnum import json import logging from pathlib import Path @@ -58,6 +59,21 @@ def _ensure_local_python_roborock_on_path() -> None: _ACTION_LOCK_RETRY_DELAY_SECONDS = 5.0 _ACTION_LOCK_RETRY_ATTEMPTS = 6 from .inventory_io import WEB_API_INVENTORY_FILE +class SchedulerCommand(StrEnum): + """A robot command python-roborock's RoborockCommand does not list.""" + + # The cloud's own way to run a schedule: the robot answers it with the schedule's settings entry + # (the same one the cloud job carries, including "name", the robot-side timer id) and records the + # clean as "Schedules" in its history, for whole-home and room schedules alike. Confirmed live; + # without the entry it answers -10005 "Invalid timer", and with a bare id "Param should be + # name + fan + segment". + SERVER_SCHEDULED_START = "server_scheduled_start" + + +# A whole-home clean started by a schedule rather than a scene; see commands_for_step. +SCHEDULE_APP_START = "schedule_app_start" +# Any schedule that knows its robot-side timer id; the robot runs it itself when asked this way. +SCHEDULE_SERVER_START = "schedule_server_start" _SUPPORTED_METHODS = { "do_scenes_app_start", "do_scenes_segments", @@ -67,8 +83,21 @@ def _ensure_local_python_roborock_on_path() -> None: RoborockCommand.APP_START, RoborockCommand.APP_SEGMENT_CLEAN, RoborockCommand.APP_ZONED_CLEAN, + SchedulerCommand.SERVER_SCHEDULED_START, } _ACTION_LOCK_ERROR_CODES = {-10003, -10007} +_PARAM_ERROR_CODE = -10005 +# What app_start takes in one command, as the official app's own schedule entries carry it. A +# robot accepted every one of these together and applied the modes (confirmed live); it answers +# -10005 "Param error" to a value it does not support, such as a route its model lacks. +_APP_START_KEYS = ("repeat", "water_box_mode", "segments", "fan_power", "mop_mode", "clean_mop", "map_index") +_APP_START_DEFAULTS = {"repeat": 1, "segments": "0", "clean_mop": 1, "map_index": -1} +_SETTING_ENTRY_KEYS = ("repeat", "fan_power", "water_box_mode", "mop_mode", "mop_template_id") +# app_start's "source" field is how the robot learns who started a clean, for its history. 3 is +# python-roborock's RoborockStartType.schedule, and a robot recorded the clean as "Schedule" in the +# official app's history; an app start is 2. "start_type", "startType", "start_from", "from" and +# "trigger" were all accepted but ignored, so the clean was recorded as "App". +_SCHEDULE_START_SOURCE = 3 _RESUME_COMMAND_BY_IN_CLEANING: dict[int, RoborockCommand] = { RoborockInCleaning.global_clean_not_complete.value: RoborockCommand.APP_START, RoborockInCleaning.zone_clean_not_complete.value: RoborockCommand.RESUME_ZONED_CLEAN, @@ -82,7 +111,7 @@ class RoutineExecutionError(RuntimeError): @dataclass(frozen=True) class RoutineCommand: - command: RoborockCommand + command: RoborockCommand | SchedulerCommand params: dict[str, Any] | list[Any] | None = None @@ -376,9 +405,44 @@ def commands_for_step(step: RoutineStep) -> list[RoutineCommand]: ), RoutineCommand(RoborockCommand.APP_START, None), ] + if step.method == SCHEDULE_SERVER_START: + entry = _single_start_entry(step) + body = {key: entry[key] for key in (*_APP_START_KEYS, "name") if key in entry} + return [RoutineCommand(SchedulerCommand.SERVER_SCHEDULED_START, [body])] + if step.method == SCHEDULE_APP_START: + entry = _single_start_entry(step) + body = {**_APP_START_DEFAULTS, **{key: entry[key] for key in _APP_START_KEYS if key in entry}} + body["source"] = _SCHEDULE_START_SOURCE + return [RoutineCommand(RoborockCommand.APP_START, [body])] raise RoutineExecutionError(f"Unsupported step method {step.method!r}") +def fallback_commands_for_step(step: RoutineStep) -> list[RoutineCommand] | None: + """The longer way to run [step], for when the robot refuses the short one; None if there is none. + + A schedule is started by asking the robot to run it (server_scheduled_start), and failing that + by one app_start carrying its settings. A refused command changes nothing, so on "Param error" + or "Invalid timer" the next, plainer way is tried: a room schedule as a segment clean, a + whole-home one as separate setting commands, which is how a routine starts a whole-home clean. + """ + if step.method == SCHEDULE_SERVER_START: + entry = _single_start_entry(step) + rooms = [ + int(part) + for part in str(entry.get("segments") or "").replace(";", ",").split(",") + if part.strip().isdigit() and int(part) > 0 + ] + if rooms: + settings = {key: entry[key] for key in _SETTING_ENTRY_KEYS if key in entry} + return commands_for_step( + replace(step, method="do_scenes_segments", params={"data": [{**settings, "segs": rooms}]}) + ) + return commands_for_step(replace(step, method=SCHEDULE_APP_START)) + if step.method != SCHEDULE_APP_START: + return None + return commands_for_step(replace(step, method="do_scenes_app_start")) + + def _is_optional_unsupported_command(command: RoborockCommand, exc: Exception) -> bool: return command == RoborockCommand.SET_MOP_TEMPLATE_ID and isinstance(exc, RoborockUnsupportedFeature) @@ -534,7 +598,7 @@ def _on_message(self, payload: bytes) -> None: async def send_command( self, - command: RoborockCommand, + command: RoborockCommand | SchedulerCommand, params: dict[str, Any] | list[Any] | None = None, ) -> Any: if self._session is None: @@ -993,6 +1057,29 @@ async def _send_step_command( continue raise + async def _send_step_commands( + self, + *, + client: _RoutineMqttClient, + logger: logging.LoggerAdapter, + step: RoutineStep, + commands: list[RoutineCommand], + ) -> None: + for routine_command in commands: + logger.info( + "Routine step=%s method=%s command=%s params=%s", + step.step_id, + step.method, + routine_command.command.value, + routine_command.params, + ) + await self._send_step_command( + client=client, + logger=logger, + step=step, + routine_command=routine_command, + ) + async def _run_scene(self, *, scene: dict[str, Any], steps: list[RoutineStep]) -> None: await self._run_steps(scene=scene, steps=steps, sync_scene_tids=True) @@ -1027,22 +1114,21 @@ async def _run_steps( if sync_scene_tids: await self._sync_scene_tids(client=client, scene=scene, device_id=device_id, logger=logger) for step_index, step in enumerate(steps): - commands = commands_for_step(step) waits_for_step_complete = RoborockDataProtocol.TASK_COMPLETE.value in step.finish_dp_ids - for routine_command in commands: - logger.info( - "Routine step=%s method=%s command=%s params=%s", - step.step_id, - step.method, - routine_command.command.value, - routine_command.params, + try: + await self._send_step_commands( + client=client, logger=logger, step=step, commands=commands_for_step(step) ) - await self._send_step_command( - client=client, - logger=logger, - step=step, - routine_command=routine_command, + except Exception as exc: # noqa: BLE001 + fallback = fallback_commands_for_step(step) + if fallback is None or _exception_error_code(exc) != _PARAM_ERROR_CODE: + raise + logger.warning( + "Robot refused the single start step=%s (%s); sending the settings as separate commands", + step.step_id, + _exception_error_message(exc), ) + await self._send_step_commands(client=client, logger=logger, step=step, commands=fallback) if waits_for_step_complete: logger.info("Waiting for ready state step=%s scene=%s", step.step_id, _scene_name(scene)) await client.wait_for_step_complete() diff --git a/src/roborock_local_server/bundled_backend/shared/schedule_runner.py b/src/roborock_local_server/bundled_backend/shared/schedule_runner.py index 33fa64ce..99259982 100644 --- a/src/roborock_local_server/bundled_backend/shared/schedule_runner.py +++ b/src/roborock_local_server/bundled_backend/shared/schedule_runner.py @@ -18,7 +18,13 @@ from .constants import DEFAULT_TIMEZONE from .data_helpers import as_bool from .inventory_io import load_inventory -from .routine_runner import RoutineExecutionError, RoutineStep, get_routine_runner +from .routine_runner import ( + SCHEDULE_APP_START, + SCHEDULE_SERVER_START, + RoutineExecutionError, + RoutineStep, + get_routine_runner, +) from .schedule_store import set_enabled if TYPE_CHECKING: @@ -29,6 +35,7 @@ TICK_SECONDS = 20 _SUPPORTED_JOB_METHOD = "server_scheduled_start" _CARRIED_ENTRY_KEYS = ("repeat", "fan_power", "water_box_mode", "mop_mode", "mop_template_id") +_START_ENTRY_KEYS = (*_CARRIED_ENTRY_KEYS, "segments", "clean_mop", "map_index") class CronError(ValueError): @@ -191,9 +198,24 @@ def steps_for_job(job: dict[str, Any]) -> list[RoutineStep]: raise RoutineExecutionError("schedule params must contain one settings object") settings = {key: entry[key] for key in _CARRIED_ENTRY_KEYS if key in entry} + # A whole-home start also carries the rest of the entry, so one app_start can take it all. + start_settings = {key: entry[key] for key in _START_ENTRY_KEYS if key in entry} + # The robot-side timer id the official app (or LocalRock) saved this schedule under. A schedule + # that has one is run the way the cloud runs it: the robot is asked to start that timer. + timer_id = entry.get("name") name = str(entry.get("name") or job.get("name") or "").strip() if not name or name.isdigit(): name = f"Schedule {job.get('id', '')}".strip() + if isinstance(timer_id, str) and timer_id.strip(): + return [ + RoutineStep( + step_id=1, + name=name, + method=SCHEDULE_SERVER_START, + params=[{**start_settings, "name": timer_id.strip()}], + finish_dp_ids=(), + ) + ] segments = _segment_ids(entry.get("segments")) if segments: return [ @@ -209,8 +231,8 @@ def steps_for_job(job: dict[str, Any]) -> list[RoutineStep]: RoutineStep( step_id=1, name=name, - method="do_scenes_app_start", - params=[settings], + method=SCHEDULE_APP_START, + params=[start_settings], finish_dp_ids=(), ) ] diff --git a/tests/test_routine_runner.py b/tests/test_routine_runner.py index 40110c01..eb3070c1 100644 --- a/tests/test_routine_runner.py +++ b/tests/test_routine_runner.py @@ -9,7 +9,13 @@ from roborock.exceptions import RoborockException from roborock_local_server.bundled_backend.shared.context import ServerContext import roborock_local_server.bundled_backend.shared.routine_runner as routine_runner_module -from roborock_local_server.bundled_backend.shared.routine_runner import RoutineRunner, parse_scene_steps +from roborock_local_server.bundled_backend.shared.routine_runner import ( + SCHEDULE_APP_START, + SCHEDULE_SERVER_START, + RoutineRunner, + RoutineStep, + parse_scene_steps, +) from roborock.roborock_typing import RoborockCommand @@ -711,3 +717,157 @@ async def exercise() -> None: await client.wait_for_dock_settle(initial_delay_seconds=0.0) asyncio.run(exercise()) + + +def _schedule_step(**entry) -> RoutineStep: + return RoutineStep(step_id=1, name="Schedule 1", method=SCHEDULE_APP_START, params=[entry], finish_dp_ids=()) + + +def _run_schedule_step(tmp_path: Path, monkeypatch, step: RoutineStep, *, refuse_single_start_with: dict | None): + """Run [step] against a fake robot; returns every command it was sent, refused ones included.""" + sent_commands: list[tuple[RoborockCommand, object]] = [] + + class FakeRoutineClient: + def __init__(self, context, device, logger) -> None: + _ = context, device, logger + + async def connect(self) -> None: + return None + + async def close(self) -> None: + return None + + async def send_command(self, command, params=None): + sent_commands.append((command, params)) + if refuse_single_start_with is not None and command == RoborockCommand.APP_START and params is not None: + raise RoborockException(refuse_single_start_with) + return ["ok"] + + async def wait_for_step_complete(self) -> None: + return None + + monkeypatch.setattr(routine_runner_module, "_RoutineMqttClient", FakeRoutineClient) + scene = _scene(scene_id=1, device_id="6HL2zfniaoYYV01CkVuhkO", name="Schedule 1") + runner = RoutineRunner(_test_context(tmp_path)) + return sent_commands, asyncio.run(runner._run_steps(scene=scene, steps=[step], sync_scene_tids=False)) + + +def test_a_schedule_start_is_one_app_start_carrying_the_whole_entry(tmp_path: Path, monkeypatch) -> None: + # A robot accepted app_start with all of these together and applied the modes (confirmed live). + step = _schedule_step(fan_power=104, water_box_mode=203, mop_mode=300) + + sent, _ = _run_schedule_step(tmp_path, monkeypatch, step, refuse_single_start_with=None) + + assert sent == [ + ( + RoborockCommand.APP_START, + [{"fan_power": 104, "water_box_mode": 203, "mop_mode": 300, + "repeat": 1, "segments": "0", "clean_mop": 1, "map_index": -1, + # "source" 3 is what makes the robot's history say "Schedule" instead of "App". + "source": 3}], + ) + ] + + +def test_a_schedule_start_falls_back_to_separate_commands_on_a_param_error(tmp_path: Path, monkeypatch) -> None: + # A route the robot's model lacks makes it answer -10005 "Param error", and a refused command + # changes nothing, so the same settings go out the longer way. + step = _schedule_step(fan_power=104, water_box_mode=203, mop_mode=300) + + sent, _ = _run_schedule_step( + tmp_path, monkeypatch, step, refuse_single_start_with={"code": -10005, "message": "Param error"} + ) + + assert [command for command, _ in sent] == [ + RoborockCommand.APP_START, + RoborockCommand.SET_CUSTOM_MODE, + RoborockCommand.SET_WATER_BOX_CUSTOM_MODE, + RoborockCommand.SET_MOP_MODE, + RoborockCommand.SET_CLEAN_REPEAT_TIMES, + RoborockCommand.APP_START, + ] + assert sent[-1] == (RoborockCommand.APP_START, None) + + +def test_a_schedule_start_does_not_fall_back_on_any_other_refusal(tmp_path: Path, monkeypatch) -> None: + step = _schedule_step(fan_power=104) + + with pytest.raises(RoborockException): + _run_schedule_step( + tmp_path, monkeypatch, step, refuse_single_start_with={"code": -10001, "message": "Busy"} + ) + + +def _run_schedule_step_refusing(tmp_path: Path, monkeypatch, step: RoutineStep, error: dict): + """Like _run_schedule_step, but the robot refuses server_scheduled_start and accepts the rest.""" + sent_commands: list[tuple[object, object]] = [] + + class FakeRoutineClient: + def __init__(self, context, device, logger) -> None: + _ = context, device, logger + + async def connect(self) -> None: + return None + + async def close(self) -> None: + return None + + async def send_command(self, command, params=None): + sent_commands.append((command, params)) + if command == routine_runner_module.SchedulerCommand.SERVER_SCHEDULED_START: + raise RoborockException(error) + return ["ok"] + + async def wait_for_step_complete(self) -> None: + return None + + monkeypatch.setattr(routine_runner_module, "_RoutineMqttClient", FakeRoutineClient) + scene = _scene(scene_id=1, device_id="6HL2zfniaoYYV01CkVuhkO", name="Schedule 1") + runner = RoutineRunner(_test_context(tmp_path)) + return sent_commands, asyncio.run(runner._run_steps(scene=scene, steps=[step], sync_scene_tids=False)) + + +def _server_start_step(**entry) -> RoutineStep: + return RoutineStep(step_id=1, name="Schedule 1", method=SCHEDULE_SERVER_START, params=[entry], finish_dp_ids=()) + + +def test_a_server_start_asks_the_robot_to_run_the_schedule_with_its_whole_entry(tmp_path: Path, monkeypatch) -> None: + # Sent exactly as the cloud job's settings entry, "name" (the robot-side timer id) included; + # confirmed live to start a clean the robot records as "Schedules | Rooms". + step = _server_start_step( + repeat=1, water_box_mode=202, segments="16,17", fan_power=102, mop_mode=300, + clean_mop=1, map_index=0, name="1792336598179", + ) + + sent, _ = _run_schedule_step(tmp_path, monkeypatch, step, refuse_single_start_with=None) + + assert [command for command, _ in sent] == [routine_runner_module.SchedulerCommand.SERVER_SCHEDULED_START] + assert sent[0][1] == [ + {"repeat": 1, "water_box_mode": 202, "segments": "16,17", "fan_power": 102, "mop_mode": 300, + "clean_mop": 1, "map_index": 0, "name": "1792336598179"} + ] + + +def test_a_refused_server_start_falls_back_to_a_segment_clean_for_a_room_schedule(tmp_path: Path, monkeypatch) -> None: + # "Invalid timer" (the robot no longer holds that timer) is a -10005 like any bad parameter. + step = _server_start_step(fan_power=102, water_box_mode=202, segments="16,17", name="gone") + sent, _ = _run_schedule_step_refusing(tmp_path, monkeypatch, step, {"code": -10005, "message": "Invalid timer"}) + + commands = [command for command, _ in sent] + assert commands[0] == routine_runner_module.SchedulerCommand.SERVER_SCHEDULED_START + assert commands[-1] == RoborockCommand.APP_SEGMENT_CLEAN + assert sent[-1][1] == [{"segments": [16, 17], "repeat": 1}] + + +def test_a_refused_server_start_falls_back_to_the_single_app_start_for_a_whole_home_schedule( + tmp_path: Path, monkeypatch +) -> None: + step = _server_start_step(fan_power=102, water_box_mode=202, segments="0", name="gone") + sent, _ = _run_schedule_step_refusing(tmp_path, monkeypatch, step, {"code": -10005, "message": "Invalid timer"}) + + assert sent[0][0] == routine_runner_module.SchedulerCommand.SERVER_SCHEDULED_START + assert sent[1] == ( + RoborockCommand.APP_START, + [{"repeat": 1, "segments": "0", "clean_mop": 1, "map_index": -1, "fan_power": 102, + "water_box_mode": 202, "source": 3}], + ) diff --git a/tests/test_schedule_runner.py b/tests/test_schedule_runner.py index 8769898b..2ee5f098 100644 --- a/tests/test_schedule_runner.py +++ b/tests/test_schedule_runner.py @@ -30,7 +30,6 @@ def _job(**overrides): "mop_mode": 300, "clean_mop": 1, "map_index": -1, - "name": "1759965417864", } ], }, @@ -78,11 +77,38 @@ def test_whole_house_job_becomes_an_app_start_step_carrying_vac_and_mop_settings steps = sr.steps_for_job(_job()) assert len(steps) == 1 step = steps[0] - assert step.method == "do_scenes_app_start" - assert step.params == [{"repeat": 1, "water_box_mode": 203, "fan_power": 104, "mop_mode": 300}] + assert step.method == "schedule_app_start" + assert step.params == [ + {"repeat": 1, "water_box_mode": 203, "segments": "0", "fan_power": 104, "mop_mode": 300, "clean_mop": 1, "map_index": -1} + ] assert step.name == "Schedule 3549431" +def test_job_with_a_robot_side_timer_id_is_run_by_asking_the_robot_to_start_it(): + # The official app saves every schedule on the robot too and records its timer id as "name" in + # the cloud job's settings; the cloud runs such a job with server_scheduled_start (confirmed + # live: the robot records "Schedules" for it, for rooms and whole-home alike). + job = _job() + job["param"]["params"][0]["name"] = "1759965417864" + step = sr.steps_for_job(job)[0] + assert step.method == "schedule_server_start" + assert step.params == [ + { + "repeat": 1, "water_box_mode": 203, "segments": "0", "fan_power": 104, "mop_mode": 300, + "clean_mop": 1, "map_index": -1, "name": "1759965417864", + } + ] + + +def test_room_job_with_a_robot_side_timer_id_carries_its_rooms_in_the_same_step(): + job = _job() + job["param"]["params"][0].update({"name": "1759965417864", "segments": "16,17", "map_index": 0}) + step = sr.steps_for_job(job)[0] + assert step.method == "schedule_server_start" + assert step.params[0]["segments"] == "16,17" + assert step.params[0]["map_index"] == 0 + + def test_segment_job_becomes_a_segment_clean_step(): job = _job() job["param"]["params"][0]["segments"] = "16,17" @@ -123,7 +149,7 @@ def test_tick_fires_a_job_that_came_due_since_the_previous_tick(tmp_path, monkey results = runner.tick(_utc(2026, 9, 28, 15, 0, 5)) assert [r["status"] for r in results] == ["started"] assert fake.started[0]["device_id"] == "duid-1" - assert fake.started[0]["steps"][0].method == "do_scenes_app_start" + assert fake.started[0]["steps"][0].method == "schedule_app_start" def test_tick_does_not_fire_twice_or_when_disabled(tmp_path, monkeypatch): From d5ada08db732a13f082f0e4ee5a8b0f74c118d70 Mon Sep 17 00:00:00 2001 From: pponce Date: Sun, 25 Oct 2026 12:28:15 -0700 Subject: [PATCH 11/11] Synchronize native schedule enablement across clients Reconcile verified robot timer state after forwarded native writes and periodic reads, and verify native enablement for direct job API updates. Preserve schedule identity, settings, and occurrence claims; reject stale or ambiguous updates. Add coverage for concurrent edits, missed acknowledgements, one-time schedules, HTTP updates, and MQTT pass-through. --- docs/schedules.md | 43 ++- .../https_server/routes/user/devices/jobs.py | 97 +++++-- .../mqtt_broker_server/topic_bridge.py | 7 + .../shared/native_schedule_sync.py | 268 ++++++++++++++++++ .../bundled_backend/shared/schedule_store.py | 72 +++++ src/roborock_local_server/server.py | 37 ++- tests/test_native_schedule_sync.py | 235 +++++++++++++++ tests/test_schedules.py | 93 +++++- tests/test_timer_rpc.py | 48 ++++ 9 files changed, 864 insertions(+), 36 deletions(-) create mode 100644 src/roborock_local_server/bundled_backend/shared/native_schedule_sync.py create mode 100644 tests/test_native_schedule_sync.py diff --git a/docs/schedules.md b/docs/schedules.md index 1a18d11f..25208eef 100644 --- a/docs/schedules.md +++ b/docs/schedules.md @@ -20,15 +20,15 @@ restoration, daily reset, and recovery. The server provides the underlying APIs. | --- | --- | | GET `/user/devices/{duid}/jobs` | Stored jobs with calculated UTC `nextFireTime` for enabled, valid jobs | | POST to the collection | Persist a new job with a unique numeric ID | -| PUT/PATCH `/user/devices/{duid}/jobs/{id}` | Update supplied fields; preserve omitted fields and unknown model fields | +| PUT/PATCH `/user/devices/{duid}/jobs/{id}` | Verify native enablement when supplied, then save; preserve omitted fields and unknown model fields | | DELETE an item | Remove the stored job and its occurrence record | | `get/set/upd/del_server_timer` | Forward unchanged to the robot, which owns these timer entries | | Legacy `get/set/upd/del_timer` | Local compatibility view for jobs without a native timer name | | Native scheduled execution | Verify the robot timer is on, then send `server_scheduled_start` with the complete settings entry | A job's numeric `id` is different from the robot timer's string ID in -`param.params[0].name`. Cloud job `enabled: true` can coexist with a robot timer -that is off. The server checks both rather than assuming they are equivalent. +`param.params[0].name`. Imported or previously unsynchronized jobs can disagree with the native timer. +The server now reconciles enablement and still checks both before execution. Native jobs are not duplicated in the legacy `get_timer` view; both captured robots returned an empty legacy list while holding native timers. @@ -40,6 +40,39 @@ strings, and unknown future options survive storage and native dispatch. Invalid cron/timezone/body values are rejected. A persistence failure does not receive a successful write acknowledgement. +## Enablement synchronization across clients + +Native schedule on/off changes are synchronized by the server, independent of +whether Homebridge, LocalRock, the OEM app, or the dashboard made the change. +A forwarded `upd/set/del_server_timer` request prompts a fresh native timer read. +The request and the robot's reply remain unchanged; sending a request or receiving +an `ok` is not used as proof of the final state. No client-specific HTTP write is +required after a native toggle. + +A background read also runs at startup and approximately every 30 seconds to +recover missed traffic, reconnects, and changes made directly over the LAN. +Successful reconciliation changes only `enabled` on existing, uniquely matched +jobs. It preserves IDs, cron/timezone, settings, unknown fields, and occurrence +claims. The dashboard's normal refresh shows the updated saved state. Other +clients see it on their next normal read; this is eventual synchronization, +not an unsolicited push into every app. + +An HTTP job PUT/PATCH that supplies `enabled` verifies the matching native timer +before saving. A timer already in the requested state needs no extra write, +so the OEM app's native-update-then-PUT sequence remains compatible. A lost +acknowledgement is checked with one read-back, without repeating the write. +Time-only requests that omit `enabled` remain store-only edits. + +Background reads and server edits share a per-vacuum lock. A native write that +overlaps a background read invalidates that read; edits/deletions during a read +are checked again before persistence. Missing, malformed, ambiguous or unreachable +timers leave the last saved state intact and never create replacement timers. +A normally polled, still-on timer does not re-enable a consumed one-time job. +An explicit native enable request can do so only after a fresh matching read; +that intent expires after two minutes. Unlinked server-only jobs retain their +existing behavior. Native timer time/settings edits are not imported into job +cron/settings by this enablement reconciliation. + ## Execution and missed runs The scheduler evaluates five-field cron expressions in each job's IANA timezone. @@ -104,8 +137,8 @@ Build the selected Git commit into a versioned Docker image. The CI runs the ful 2. Import current jobs and verify every numeric job ID maps to the expected native timer name. Confirm local jobs reads and robot timer reads separately. 3. Disable a recurring test schedule through Homebridge. Confirm the robot timer - reads off and no clean begins at the scheduled time, even if job `enabled` - remains true. Re-enable it and confirm a later occurrence runs exactly once. + reads off and the saved job becomes disabled without a separate HTTP update. + Re-enable it and confirm both states agree and a later occurrence runs once. 4. Test a room schedule and a whole-home schedule. Compare settings, rooms, map, and history against the app. A refusal should stop dispatch, with no fallback ordinary clean. diff --git a/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/jobs.py b/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/jobs.py index 4e97ad9b..20291c82 100644 --- a/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/jobs.py +++ b/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/jobs.py @@ -10,7 +10,8 @@ from shared.data_helpers import as_bool from shared.http_helpers import wrap_response from shared.schedule_runner import parse_cron -from shared.schedule_store import delete_job, list_jobs, upsert_job +from shared.schedule_store import SettingsError, delete_job, list_jobs, native_timer_name, upsert_job +from shared.native_schedule_sync import TimerSyncError, sync_timer from .service import device_jobs_payload @@ -105,40 +106,92 @@ def match_update(path: str, method: str = "GET") -> bool: return method.upper() in ("PUT", "PATCH") and bool(_ITEM_RE.fullmatch(path.rstrip("/"))) -def build_update( - ctx: ServerContext, - _query_params: dict[str, list[str]], - body_params: dict[str, list[str]], - clean_path: str, -) -> dict[str, Any]: - match_ = _ITEM_RE.fullmatch(clean_path.rstrip("/")) - duid, job_id = match_.group(1), match_.group(2) # type: ignore[union-attr] - existing = next((job for job in list_jobs(ctx, duid) if str(job.get("id")) == job_id), None) - if existing is None: - return _error(404, "schedule not found") +class ScheduleUpdateError(ValueError): + def __init__(self, code: int, message: str) -> None: + super().__init__(message) + self.code = code + + +def _prepare_update(ctx, duid, job_id, body_params): + matches = [job for job in list_jobs(ctx, duid) if str(job.get("id")) == job_id] + if not matches: + raise ScheduleUpdateError(404, "schedule not found") + if len(matches) != 1: + raise ScheduleUpdateError(409, "duplicate schedule IDs") + existing = matches[0] try: body = _json_body(body_params) cron = str(body.get("cron", existing.get("cron")) or "").strip() parse_cron(cron) zone = _validate_zone(body["timeZoneId"]) if "timeZoneId" in body else "" except ValueError as exc: - return _error(400, str(exc)) + raise ScheduleUpdateError(400, str(exc)) from exc param = _parse_param(body.get("param")) if "param" in body else None if "param" in body and param is None: - return _error(400, "param must be a JSON object") + raise ScheduleUpdateError(400, "param must be a JSON object") + values = dict( + job_id=job_id, cron=cron, + enabled=as_bool(body.get("enabled", existing.get("enabled", True)), True), + time_zone=zone, + repeated=as_bool(body["repeated"], True) if "repeated" in body else None, + param=param, override_time_zone=bool(zone), extra_fields=body, expected_job=existing, + ) + return existing, body, values + + +def _save_update(ctx, duid, values): try: - updated = upsert_job( - ctx, duid, job_id=job_id, cron=cron, - enabled=as_bool(body.get("enabled", existing.get("enabled", True)), True), - time_zone=zone, - repeated=as_bool(body["repeated"], True) if "repeated" in body else None, - param=param, override_time_zone=bool(zone), extra_fields=body, expected_job=existing, - ) + return wrap_response(upsert_job(ctx, duid, **values)) except ValueError as exc: return _error(409, str(exc)) except OSError: return _error(500, "unable to persist schedule") - return wrap_response(updated) + + +def build_update(ctx, _query_params, body_params, clean_path): + match_ = _ITEM_RE.fullmatch(clean_path.rstrip("/")) + duid, job_id = match_.group(1), match_.group(2) + try: + _, _, values = _prepare_update(ctx, duid, job_id, body_params) + except ScheduleUpdateError as exc: + return _error(exc.code, str(exc)) + return _save_update(ctx, duid, values) + + +async def build_update_synced(ctx, _query_params, body_params, clean_path, coordinator): + """Authenticated HTTP edits share the same native state as every MQTT client.""" + match_ = _ITEM_RE.fullmatch(clean_path.rstrip("/")) + duid, job_id = match_.group(1), match_.group(2) + async with coordinator.lock(duid): + try: + existing, body, values = _prepare_update(ctx, duid, job_id, body_params) + candidate = {**existing} + if values["param"] is not None: + candidate["param"] = values["param"] + name = native_timer_name(candidate) if "enabled" in body else None + + def check_current(): + jobs = list_jobs(ctx, duid) + if [job for job in jobs if str(job.get("id")) == job_id] != [existing]: + raise TimerSyncError("Schedule changed during timer verification; read it again", 409) + for other in jobs: + if str(other.get("id")) != job_id and native_timer_name(other) == name: + raise TimerSyncError("Multiple schedules reference this native timer", 409) + + if name is not None: + check_current() + await sync_timer(ctx, duid, name, values["enabled"], check_current=check_current) + check_current() + result = _save_update(ctx, duid, values) + if name is not None and not result.get("success"): + result["msg"] = "Robot timer was verified, but the schedule could not be saved; refresh before retrying" + return result + except ScheduleUpdateError as exc: + return _error(exc.code, str(exc)) + except SettingsError as exc: + return _error(409, str(exc)) + except TimerSyncError as exc: + return _error(exc.status, str(exc)) def match_delete(path: str, method: str = "GET") -> bool: diff --git a/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py b/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py index 1276126b..a21b890f 100644 --- a/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py +++ b/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py @@ -74,10 +74,12 @@ def __init__( runtime_state: Any | None = None, inventory_path: Path | None = None, request_interceptor: Callable[[str, bytes], bytes | None] | None = None, + request_observer: Callable[[str, bytes], None] | None = None, ) -> None: # Given (duid, payload) for each rr/m request; returns an encoded reply to # publish back to the requester instead of forwarding, or None to forward. self._request_interceptor = request_interceptor + self._request_observer = request_observer self._host = host self._port = port self._logger = logger @@ -433,6 +435,11 @@ async def _handle_cloud_message( self._d_to_m[device_topic] = cloud_topic target_topic = device_topic.topic_out + if self._request_observer is not None and not bool(getattr(message, "retain", False)): + try: + self._request_observer(cloud_topic.duid, bytes(message.payload)) + except Exception: # Observation must never prevent forwarding. + self._logger.warning("MQTT request observer failed for %s", src_topic, exc_info=True) await self._publish_mirror(client, message, target_topic) self._logger.info( "MQTT bridge m->d %s -> %s bytes=%d", diff --git a/src/roborock_local_server/bundled_backend/shared/native_schedule_sync.py b/src/roborock_local_server/bundled_backend/shared/native_schedule_sync.py new file mode 100644 index 00000000..324e37dc --- /dev/null +++ b/src/roborock_local_server/bundled_backend/shared/native_schedule_sync.py @@ -0,0 +1,268 @@ +"""Reconcile native schedule enablement independently of the client making a change.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Callable +from enum import StrEnum +import logging +import time +from typing import Any + +from .inventory_io import load_inventory +from .schedule_store import SettingsError, list_jobs, native_timer_name, reconcile_native_enabled + +_LOGGER = logging.getLogger(__name__) + + +class TimerSyncError(RuntimeError): + def __init__(self, message: str, status: int = 503) -> None: + super().__init__(message) + self.status = status + + +class TimerCommand(StrEnum): + GET = "get_server_timer" + UPDATE = "upd_server_timer" + + +def _timer_state(timers: Any, name: str) -> str: + if not isinstance(timers, list): + raise TimerSyncError("The vacuum did not return a usable timer list. Refresh schedules and retry.") + matches = [entry for entry in timers if isinstance(entry, list) and entry and entry[0] == name] + if not matches: + raise TimerSyncError("The linked robot timer is missing. Check this schedule in the Roborock app before retrying.", 409) + if len(matches) != 1 or len(matches[0]) < 2 or matches[0][1] not in ("on", "off"): + raise TimerSyncError("The linked robot timer is ambiguous. Check this schedule in the Roborock app before retrying.", 409) + return matches[0][1] + + +async def sync_timer( + context: Any, duid: str, name: str, enabled: bool, *, check_current: Callable[[], None], +) -> str: + """Update only the named timer; a fresh read is the success criterion. + + The app sends upd_server_timer before its HTTP job update. A timed-out write + may still have reached the vacuum, so read it back without repeating the + write. Never create a timer or infer one from a cloud job's numeric ID. + """ + from shared.routine_runner import _RoutineMqttClient + + device = context.runtime_credentials.resolve_device(duid=duid) + if device is None or not device.get("localkey"): + raise TimerSyncError("Connect this vacuum to the local server before changing its schedule.") + desired = "on" if enabled else "off" + client = None + write_attempted = False + try: + async with asyncio.timeout(45): + client = _RoutineMqttClient(context, device, logging.getLogger(__name__)) + await client.connect() + current = _timer_state(await client.send_command(TimerCommand.GET, []), name) + # Recheck the job after the network read, before changing the robot. + check_current() + if current != desired: + write_attempted = True + try: + await client.send_command(TimerCommand.UPDATE, [[name, desired]]) + except Exception: + # A read-back can confirm a write whose acknowledgement was + # lost. It cannot confirm a refused or unapplied write. + pass + current = _timer_state(await client.send_command(TimerCommand.GET, []), name) + if current != desired: + raise TimerSyncError("The vacuum did not confirm the requested timer setting. Refresh schedules and retry.") + return current + except TimerSyncError: + raise + except Exception as exc: + message = ( + "The robot timer may have changed, but verification failed. Refresh schedules and retry." + if write_attempted else + "Could not read this vacuum's timer. Make sure it is connected locally, then retry." + ) + raise TimerSyncError(message) from exc + finally: + if client is not None: + try: + async with asyncio.timeout(5): + await client.close() + except Exception: + logging.getLogger(__name__).warning("Unable to close schedule timer connection") + + +def timer_states(timers: Any) -> dict[str, bool]: + """Ignore ambiguous/malformed entries; absence is not proof of disablement.""" + if not isinstance(timers, list): + raise TimerSyncError("The vacuum did not return a usable timer list") + states: dict[str, bool] = {} + seen: set[str] = set() + invalid: set[str] = set() + for entry in timers: + if not isinstance(entry, list) or not entry or not isinstance(entry[0], str): + continue + name = entry[0] + if name in seen or len(entry) < 2 or entry[1] not in ("on", "off"): + invalid.add(name) + else: + states[name] = entry[1] == "on" + seen.add(name) + return {name: enabled for name, enabled in states.items() if name not in invalid} + + +class NativeScheduleSync: + """One coordinator for API edits and read-only reconciliation from any client. + + Native RPCs still reach the robot unchanged. Observing a write only schedules + a fresh read; no optimistic request or generic 'ok' reply mutates inventory. + Periodic reads also catch LAN changes and recover after missed MQTT traffic. + """ + + INTERVAL = 30.0 + DEBOUNCE = 0.3 + + def __init__(self, context: Any) -> None: + self.context = context + self._locks: dict[str, asyncio.Lock] = {} + self._generation: dict[str, int] = {} + self._enable_intent: dict[str, dict[str, tuple[float, Any]]] = {} + self._dirty: set[str] = set() + self._wake = asyncio.Event() + self._slots = asyncio.Semaphore(2) + + def lock(self, duid: str) -> asyncio.Lock: + return self._locks.setdefault(duid, asyncio.Lock()) + + def changed(self, duid: str, *, enable_name: str | None = None) -> None: + self._generation[duid] = self._generation.get(duid, 0) + 1 + if enable_name is not None: + # Tie one-time re-enablement to the occurrence already consumed when + # the request was observed, never one consumed after this request. + claims = load_inventory(self.context).get("schedule_occurrences", {}).get(duid, {}) + for job in list_jobs(self.context, duid): + try: + matches = native_timer_name(job) == enable_name + except SettingsError: + continue + if matches: + self._enable_intent.setdefault(duid, {})[enable_name] = ( + time.monotonic() + 120, claims.get(str(job.get("id"))), + ) + self._dirty.add(duid) + self._wake.set() + + def observe_request(self, duid: str, payload: bytes) -> None: + from roborock.protocol import create_mqtt_decoder + from .timer_rpc import _request_body + + device = self.context.runtime_credentials.resolve_device(duid=duid) + if not device or not device.get("localkey"): + return + try: + messages = create_mqtt_decoder(device["localkey"])(payload) + for message in messages: + if int(getattr(message.protocol, "value", message.protocol)) != 101: + continue + request = _request_body(message) + if not request or request.get("method") not in { + "upd_server_timer", "set_server_timer", "del_server_timer", + }: + continue + enable_name = None + params = request.get("params") + if (request.get("method") == "upd_server_timer" and isinstance(params, list) + and len(params) == 1 and isinstance(params[0], list) + and len(params[0]) == 2 and isinstance(params[0][0], str)): + name, state = params[0] + # A later off must invalidate an earlier on intent, too. + self._enable_intent.setdefault(duid, {}).pop(name, None) + if state == "on": + enable_name = name + self.changed(duid, enable_name=enable_name) + except Exception: # Unrelated/unsupported frames must always pass through. + return + + async def refresh(self, duid: str) -> list[Any]: + from .routine_runner import _RoutineMqttClient + + async with self._slots, self.lock(duid): + snapshot = list_jobs(self.context, duid) + names = [] + for job in snapshot: + try: + name = native_timer_name(job) + except SettingsError: + continue + if name is not None: + names.append(name) + if not names: + return [] + device = self.context.runtime_credentials.resolve_device(duid=duid) + if not device or not device.get("localkey"): + return [] + generation = self._generation.get(duid, 0) + intents = {name: intent for name, intent in self._enable_intent.get(duid, {}).items() + if intent[0] > time.monotonic()} + self._enable_intent[duid] = intents + explicit_enable = {name: intent[1] for name, intent in intents.items()} + client = None + try: + async with asyncio.timeout(15): + client = _RoutineMqttClient(self.context, device, _LOGGER) + await client.connect() + states = timer_states(await client.send_command(TimerCommand.GET, [])) + if self._generation.get(duid, 0) != generation: + # A newer native write overlapped the read. A later pass + # will confirm it; do not apply this older snapshot. + self._dirty.add(duid) + self._wake.set() + return [] + changed = reconcile_native_enabled( + self.context, duid, snapshot, states, explicit_enable=explicit_enable, + ) + # A read can arrive before an outstanding command is applied. + # Keep a bounded enable intent until it is actually observed on. + self._enable_intent[duid] = {name: until for name, until in intents.items() + if states.get(name) is not True} + if changed: + _LOGGER.info("Synchronized native schedule enablement for %s: %d changed", duid, len(changed)) + return changed + except (OSError, ValueError): + _LOGGER.warning("Could not save native schedule reconciliation for %s", duid, exc_info=True) + return [] + except Exception: + # An offline/unsupported robot keeps its last saved state. + _LOGGER.debug("Native schedule read unavailable for %s", duid, exc_info=True) + return [] + finally: + if client is not None: + try: + async with asyncio.timeout(5): + await client.close() + except Exception: + _LOGGER.debug("Could not close native schedule read for %s", duid, exc_info=True) + + async def run(self, stop_event: asyncio.Event) -> None: + next_scan = 0.0 + while not stop_event.is_set(): + now = time.monotonic() + due = set(self._dirty) + self._dirty.clear() + self._wake.clear() + if now >= next_scan: + schedules = load_inventory(self.context).get("schedules", {}) + if isinstance(schedules, dict): + due.update(str(duid) for duid in schedules) + next_scan = now + self.INTERVAL + if due: + await asyncio.sleep(self.DEBOUNCE) + results = await asyncio.gather(*(self.refresh(duid) for duid in sorted(due)), return_exceptions=True) + for duid, result in zip(sorted(due), results): + if isinstance(result, Exception): + _LOGGER.warning("Native schedule reconciliation unavailable for %s: %s", duid, type(result).__name__) + if self._dirty: + continue + try: + await asyncio.wait_for(self._wake.wait(), timeout=max(0.1, next_scan - time.monotonic())) + except TimeoutError: + pass diff --git a/src/roborock_local_server/bundled_backend/shared/schedule_store.py b/src/roborock_local_server/bundled_backend/shared/schedule_store.py index c0d26800..e59a09f6 100644 --- a/src/roborock_local_server/bundled_backend/shared/schedule_store.py +++ b/src/roborock_local_server/bundled_backend/shared/schedule_store.py @@ -69,6 +69,78 @@ def has_native_timer(job: dict[str, Any]) -> bool: return "name" in _settings_entry(job) +def native_timer_name(job: dict[str, Any]) -> str | None: + """Return an exact native reference, refusing malformed or multiple references.""" + param = job.get("param") + if isinstance(param, str): + try: + param = json.loads(param) + except (ValueError, TypeError): + return None + if not isinstance(param, dict) or param.get("method") != "server_scheduled_start": + return None + params = param.get("params") + if isinstance(params, list): + if len(params) != 1: + if any(isinstance(entry, dict) and "name" in entry for entry in params): + raise SettingsError("Schedule has multiple or ambiguous native timer references") + return None + params = params[0] + if not isinstance(params, dict) or "name" not in params: + return None + name = params["name"] + if not isinstance(name, str) or not name.strip(): + raise SettingsError("Schedule has an invalid native timer reference") + return name + + +@inventory_mutation +def reconcile_native_enabled( + ctx: ServerContext, duid: str, snapshot: list[dict[str, Any]], states: dict[str, bool], + *, explicit_enable: dict[str, Any] | None = None, +) -> list[Any]: + """Merge a fresh robot read only into unchanged, uniquely linked jobs. + + Preserve complete records and occurrence claims. A still-on robot timer + must not resurrect a consumed one-time job after a restart or routine poll. + """ + inventory = load_inventory(ctx) + jobs = _jobs_for(inventory, duid, create=False) or [] + by_name: dict[str, list[dict[str, Any]]] = {} + for job in jobs: + if not isinstance(job, dict): + continue + try: + name = native_timer_name(job) + except SettingsError: + continue + if name is not None: + by_name.setdefault(name, []).append(job) + changed = [] + claims = inventory.get("schedule_occurrences", {}).get(duid, {}) + for name, linked in by_name.items(): + if len(linked) != 1 or name not in states: + continue + job = linked[0] + before = [entry for entry in snapshot if _same_id(entry, job.get("id"))] + if before != [job] or sum(_same_id(entry, job.get("id")) for entry in jobs) != 1: + continue + enabled = states[name] + if as_bool(job.get("enabled", True), True) == enabled: + continue + if (enabled and not as_bool(job.get("repeated", True), True) + and str(job.get("id")) in claims + and (name not in (explicit_enable or {}) + or claims[str(job.get("id"))] != explicit_enable[name])): + continue + job["enabled"] = enabled + job.pop("nextFireTime", None) + changed.append(job["id"]) + if changed: + _persist(ctx, inventory) + return changed + + def job_settings(job: dict[str, Any]) -> dict[str, Any]: """The vac/mop settings a job carries, as a flat dict. diff --git a/src/roborock_local_server/server.py b/src/roborock_local_server/server.py index 47895051..bf0e5818 100644 --- a/src/roborock_local_server/server.py +++ b/src/roborock_local_server/server.py @@ -55,7 +55,9 @@ from shared.inventory_io import atomic_write_inventory, inventory_transaction from shared.protocol_auth import ProtocolAuthStore from shared.schedule_runner import ScheduleRunner +from shared.native_schedule_sync import NativeScheduleSync from shared.timer_rpc import TimerRpcInterceptor +from https_server.routes.user.devices import jobs as schedule_jobs from https_server.routes.user.scene import validity as scene_validity from https_server.routes.auth.service import ( build_login_data_response, @@ -365,6 +367,7 @@ def __init__( self._http_server: ManagedFastApiServer | None = None self._renew_task: asyncio.Task[None] | None = None self._schedule_task: asyncio.Task[None] | None = None + self._schedule_sync_task: asyncio.Task[None] | None = None self._stop_event = asyncio.Event() self.certificate_manager = CertificateManager(config=config, paths=paths) @@ -488,6 +491,7 @@ def __init__( turn_ttl=self.config.turn.ttl, ) self.endpoint_rules = default_endpoint_rules() + self.schedule_sync = NativeScheduleSync(self.context) self.app = self._create_app() def _init_zone_ranges_store(self) -> ZoneRangesStore: @@ -1354,14 +1358,20 @@ async def _handle_roborock_request(self, request: Request) -> Response: ) return response - route_name, response_payload = resolve_route( - rules=self.endpoint_rules, - context=self.context, - clean_path=clean_path, - query_params=query_params, - body_params=body_params, - method=request.method, - ) + if schedule_jobs.match_update(clean_path, request.method): + route_name = "update_schedule" + response_payload = await schedule_jobs.build_update_synced( + self.context, query_params, body_params, clean_path, self.schedule_sync, + ) + else: + route_name, response_payload = resolve_route( + rules=self.endpoint_rules, + context=self.context, + clean_path=clean_path, + query_params=query_params, + body_params=body_params, + method=request.method, + ) entry["route"] = route_name if route_name == "fw_createca": # Persist metadata, but retain the exact credential response sent to firmware. @@ -1971,6 +1981,7 @@ async def start(self) -> None: logger=self.loggers["mqtt"], runtime_state=self.runtime_state, inventory_path=self.paths.inventory_path, + request_observer=self.schedule_sync.observe_request, request_interceptor=TimerRpcInterceptor( self.context, default_timezone=self.config.network.timezone, @@ -2005,6 +2016,9 @@ async def start(self) -> None: self._renew_task = asyncio.create_task(self._renew_loop(), name="tls-renew-loop") schedule_runner = ScheduleRunner(self.context, default_timezone=self.config.network.timezone) + self._schedule_sync_task = asyncio.create_task( + self.schedule_sync.run(self._stop_event), name="native-schedule-sync", + ) self._schedule_task = asyncio.create_task( schedule_runner.run(self._stop_event), name="schedule-runner", @@ -2014,6 +2028,13 @@ async def stop(self) -> None: self._stop_event.set() self._turn_relay.stop() self._refresh_turn_health() + if self._schedule_sync_task is not None: + self._schedule_sync_task.cancel() + try: + await self._schedule_sync_task + except asyncio.CancelledError: + pass + self._schedule_sync_task = None if self._schedule_task is not None: self._schedule_task.cancel() try: diff --git a/tests/test_native_schedule_sync.py b/tests/test_native_schedule_sync.py new file mode 100644 index 00000000..3a7a53a2 --- /dev/null +++ b/tests/test_native_schedule_sync.py @@ -0,0 +1,235 @@ +"""Native state converges across clients without treating a sent command as success.""" + +import asyncio +import copy +import json +from pathlib import Path +import tempfile +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +import roborock_local_server.backend # noqa: F401 (makes bundled shared modules importable) + +from shared import native_schedule_sync as sync +from shared import schedule_store as store + + +def job(enabled=True, *, job_id=701, name="native-1", repeated=True): + return { + "id": job_id, "enabled": enabled, "repeated": repeated, "cron": "0 9 * * *", + "timeZoneId": "America/Los_Angeles", "future": {"keep": [1, 2]}, + "param": {"method": "server_scheduled_start", "params": [{ + "name": name, "segments": "16,18", "map_index": 2, "future": {"keep": True}, + }]}, + } + + +class NativeStoreTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.path = Path(self.temp.name) / "web_api_inventory.json" + self.context = SimpleNamespace(http_jsonl=self.path.parent / "http.jsonl") + self.seed([job()]) + + def seed(self, jobs, claims=None): + self.path.write_text(json.dumps({ + "schedules": {"robot": jobs, "other": [job()]}, + "schedule_occurrences": {"robot": claims or {}}, "keep": {"metadata": 1}, + })) + + def test_reconciles_both_directions_and_preserves_identity_settings_and_other_devices(self): + for enabled in (False, True): + before = json.loads(self.path.read_text()) + snapshot = store.list_jobs(self.context, "robot") + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", snapshot, {"native-1": enabled}), [701]) + before["schedules"]["robot"][0]["enabled"] = enabled + self.assertEqual(json.loads(self.path.read_text()), before) + + def test_noop_does_not_write_inventory(self): + with patch.object(store, "write_inventory", side_effect=AssertionError("unnecessary write")): + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", [job()], {"native-1": True}), []) + + def test_stale_read_cannot_overwrite_job_edited_while_reading_robot(self): + snapshot = store.list_jobs(self.context, "robot") + changed = job() + changed["cron"] = "30 9 * * *" + self.seed([changed]) + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", snapshot, {"native-1": False}), []) + self.assertEqual(store.list_jobs(self.context, "robot"), [changed]) + + def test_deleted_job_is_not_recreated(self): + snapshot = store.list_jobs(self.context, "robot") + self.seed([]) + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", snapshot, {"native-1": False}), []) + self.assertEqual(store.list_jobs(self.context, "robot"), []) + + def test_duplicate_timer_or_job_ids_are_never_guessed(self): + for jobs in ([job(), job(job_id=702)], [job(), job(name="native-2")]): + self.seed(jobs) + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", jobs, {"native-1": False}), []) + self.assertEqual(store.list_jobs(self.context, "robot"), jobs) + + def test_missing_and_malformed_timer_entries_do_not_disable_jobs(self): + for timers in ([], [["native-1", "bad"]], [["native-1", "off"], ["native-1", "on"]], [["native-1"]]): + states = sync.timer_states(timers) + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", [job()], states), []) + + def test_serialized_and_object_params_keep_their_original_form(self): + for serialized in (False, True): + original = job() + if serialized: + original["param"] = json.dumps(original["param"]) + else: + original["param"]["params"] = original["param"]["params"][0] + self.seed([original]) + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", [original], {"native-1": False}), [701]) + self.assertEqual(store.list_jobs(self.context, "robot")[0]["param"], original["param"]) + + def test_consumed_one_time_job_is_not_resurrected_by_periodic_read(self): + original = job(False, repeated=False) + self.seed([original], {"701": 12345}) + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", [original], {"native-1": True}), []) + self.assertEqual(store.reconcile_native_enabled( + self.context, "robot", [original], {"native-1": True}, explicit_enable={"native-1": 12345}, + ), [701]) + self.assertEqual(json.loads(self.path.read_text())["schedule_occurrences"], {"robot": {"701": 12345}}) + + def test_earlier_enable_intent_does_not_resurrect_a_newly_consumed_occurrence(self): + original = job(False, repeated=False) + self.seed([original], {"701": 23456}) + self.assertEqual(store.reconcile_native_enabled( + self.context, "robot", [original], {"native-1": True}, explicit_enable={"native-1": 12345}, + ), []) + + def test_unlinked_server_only_jobs_are_unchanged(self): + original = job() + del original["param"]["params"][0]["name"] + self.seed([original]) + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", [original], {"701": False}), []) + + def test_persistence_failure_does_not_lose_inventory(self): + before = self.path.read_bytes() + with patch.object(store, "write_inventory", return_value=False): + with self.assertRaises(OSError): + store.reconcile_native_enabled(self.context, "robot", [job()], {"native-1": False}) + self.assertEqual(self.path.read_bytes(), before) + + +class NativeCoordinatorTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.path = Path(self.temp.name) / "web_api_inventory.json" + self.path.write_text(json.dumps({"schedules": {"robot": [job()]}})) + self.context = SimpleNamespace( + http_jsonl=self.path.parent / "http.jsonl", + runtime_credentials=SimpleNamespace(resolve_device=lambda **kw: {"duid": "robot", "localkey": "test"}), + ) + self.coordinator = sync.NativeScheduleSync(self.context) + self.robot = SimpleNamespace(timers=[["native-1", "off"]], calls=[], after_read=None, write_error=None, apply=True) + robot = self.robot + + class Client: + def __init__(self, *args): + pass + + async def connect(self): + robot.calls.append("connect") + + async def send_command(self, command, params): + robot.calls.append(str(command)) + if command == "get_server_timer": + result = copy.deepcopy(robot.timers) + if robot.after_read: + await robot.after_read() + return result + if robot.apply: + robot.timers[0][1] = params[0][1] + if robot.write_error: + raise robot.write_error + return ["ok"] + + async def close(self): + robot.calls.append("close") + + self.network = patch.dict("sys.modules", {"shared.routine_runner": SimpleNamespace(_RoutineMqttClient=Client)}) + self.network.start() + self.addCleanup(self.network.stop) + + async def test_external_toggle_reconciliation_reads_only_and_closes_connection(self): + self.assertEqual(await self.coordinator.refresh("robot"), [701]) + self.assertFalse(store.list_jobs(self.context, "robot")[0]["enabled"]) + self.assertEqual(self.robot.calls, ["connect", "get_server_timer", "close"]) + + async def test_later_native_write_invalidates_an_in_flight_read(self): + async def later(): + self.coordinator.changed("robot") + self.robot.timers[0][1] = "on" + self.robot.after_read = later + self.assertEqual(await self.coordinator.refresh("robot"), []) + self.assertTrue(store.list_jobs(self.context, "robot")[0]["enabled"]) + self.robot.after_read = None + self.assertEqual(await self.coordinator.refresh("robot"), []) + + async def test_dashboard_or_http_edit_lock_prevents_background_race(self): + async with self.coordinator.lock("robot"): + pending = asyncio.create_task(self.coordinator.refresh("robot")) + await asyncio.sleep(0) + self.assertEqual(self.robot.calls, []) + self.assertFalse(pending.done()) + self.assertEqual(await pending, [701]) + + async def test_lost_write_ack_requires_a_fresh_matching_read(self): + self.robot.write_error = TimeoutError() + self.assertEqual(await sync.sync_timer(self.context, "robot", "native-1", True, check_current=lambda: None), "on") + self.assertEqual(self.robot.calls.count("upd_server_timer"), 1) + self.assertEqual(self.robot.calls.count("get_server_timer"), 2) + + async def test_unapplied_write_is_not_success(self): + self.robot.apply = False + with self.assertRaises(sync.TimerSyncError): + await sync.sync_timer(self.context, "robot", "native-1", True, check_current=lambda: None) + self.assertEqual(self.robot.calls.count("upd_server_timer"), 1) + + async def test_matching_timer_needs_no_second_write(self): + self.assertEqual(await sync.sync_timer(self.context, "robot", "native-1", False, check_current=lambda: None), "off") + self.assertNotIn("upd_server_timer", self.robot.calls) + + async def test_startup_and_periodic_reads_recover_changes_without_observed_requests(self): + self.coordinator.INTERVAL = 0.02 + self.coordinator.DEBOUNCE = 0 + stop = asyncio.Event() + task = asyncio.create_task(self.coordinator.run(stop)) + try: + for enabled in (False, True): + self.robot.timers[0][1] = "on" if enabled else "off" + async with asyncio.timeout(1): + while store.list_jobs(self.context, "robot")[0]["enabled"] != enabled: + await asyncio.sleep(0.001) + finally: + stop.set() + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + self.assertNotIn("upd_server_timer", self.robot.calls) + + async def test_cancellation_closes_an_in_flight_read(self): + waiting = asyncio.Event() + async def pause(): + waiting.set() + await asyncio.Event().wait() + self.robot.after_read = pause + task = asyncio.create_task(self.coordinator.refresh("robot")) + await waiting.wait() + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + self.assertEqual(self.robot.calls[-1], "close") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_schedules.py b/tests/test_schedules.py index 2036c196..90c27ed3 100644 --- a/tests/test_schedules.py +++ b/tests/test_schedules.py @@ -1,5 +1,7 @@ import json +import copy from pathlib import Path +from types import SimpleNamespace from uuid import uuid4 import pytest @@ -271,7 +273,40 @@ def _native_job(): } -def test_homebridge_delay_and_restore_put_round_trip_preserves_native_job(tmp_path): +@pytest.fixture +def native_robot(monkeypatch): + from shared import routine_runner + robot = SimpleNamespace(timers=[["1800000000001", "on"]], calls=[], apply=True, + error=None, after_read=None) + + class Client: + def __init__(self, context, device, logger): + assert device["duid"] == "robot-test" + + async def connect(self): + pass + + async def send_command(self, command, params): + robot.calls.append((str(command), copy.deepcopy(params))) + if str(command) == "get_server_timer": + if robot.after_read: + robot.after_read() + return copy.deepcopy(robot.timers) + assert str(command) == "upd_server_timer" + if robot.apply: + robot.timers[0][1] = params[0][1] + if robot.error: + raise robot.error + return ["ok"] + + async def close(self): + pass + + monkeypatch.setattr(routine_runner, "_RoutineMqttClient", Client) + return robot + + +def test_homebridge_delay_and_restore_put_round_trip_preserves_native_job(tmp_path, native_robot): original = _native_job() client, paths = _jobs_client(tmp_path, "robot-test", [original]) collection = "/user/devices/robot-test/jobs" @@ -287,6 +322,62 @@ def test_homebridge_delay_and_restore_put_round_trip_preserves_native_job(tmp_pa listed = client.get(collection, headers=_hawk_headers(paths.cloud_snapshot_path, collection)).json()["result"] assert {k: v for k, v in listed[0].items() if k != "nextFireTime"} == original assert listed[0]["param"]["params"][0]["name"] != str(listed[0]["id"]) + assert all(method == "get_server_timer" for method, _ in native_robot.calls) + + +def test_direct_http_enablement_updates_robot_then_store_for_any_client(tmp_path, native_robot): + original = _native_job() + client, paths = _jobs_client(tmp_path, "robot-test", [original]) + item = "/user/devices/robot-test/jobs/42001" + for enabled in (False, True): + raw, headers = _signed_json(paths, item, {"enabled": enabled}) + result = client.put(item, content=raw, headers=headers).json() + assert result["success"] is True + assert result["result"] == {**original, "enabled": enabled} + assert native_robot.timers[0][1] == ("on" if enabled else "off") + assert native_robot.calls[-2:] == [ + ("upd_server_timer", [["1800000000001", "on" if enabled else "off"]]), + ("get_server_timer", []), + ] + + +def test_oem_timer_then_job_put_does_not_write_native_timer_again(tmp_path, native_robot): + native_robot.timers[0][1] = "off" + client, paths = _jobs_client(tmp_path, "robot-test", [_native_job()]) + item = "/user/devices/robot-test/jobs/42001" + raw, headers = _signed_json(paths, item, {"enabled": False}) + result = client.put(item, content=raw, headers=headers).json() + assert result["success"] is True + assert native_robot.calls == [("get_server_timer", [])] + + +@pytest.mark.parametrize("lost_ack", [False, True]) +def test_http_requires_actual_confirmation_and_does_not_retry_write(tmp_path, native_robot, lost_ack): + native_robot.error = TimeoutError("lost ack") if lost_ack else None + native_robot.apply = lost_ack + client, paths = _jobs_client(tmp_path, "robot-test", [_native_job()]) + before = paths.inventory_path.read_bytes() + item = "/user/devices/robot-test/jobs/42001" + raw, headers = _signed_json(paths, item, {"enabled": False}) + result = client.put(item, content=raw, headers=headers).json() + assert result["success"] is lost_ack + assert sum(method == "upd_server_timer" for method, _ in native_robot.calls) == 1 + if not lost_ack: + assert paths.inventory_path.read_bytes() == before + + +def test_http_change_during_timer_read_refuses_stale_native_write(tmp_path, native_robot): + client, paths = _jobs_client(tmp_path, "robot-test", [_native_job()]) + def change(): + data = json.loads(paths.inventory_path.read_text()) + data["schedules"]["robot-test"][0]["cron"] = "0 10 * * *" + paths.inventory_path.write_text(json.dumps(data)) + native_robot.after_read = change + item = "/user/devices/robot-test/jobs/42001" + raw, headers = _signed_json(paths, item, {"enabled": False}) + result = client.put(item, content=raw, headers=headers).json() + assert result["success"] is False and result["code"] == 409 + assert native_robot.calls == [("get_server_timer", [])] def test_put_replaces_param_exactly_when_app_removes_optional_fields(tmp_path): diff --git a/tests/test_timer_rpc.py b/tests/test_timer_rpc.py index 84f1feac..4e951025 100644 --- a/tests/test_timer_rpc.py +++ b/tests/test_timer_rpc.py @@ -178,6 +178,54 @@ def test_interceptor_leaves_server_timer_to_the_robot(ctx, method): assert interceptor(DUID, _encode_request(method, [])) is None +def test_native_write_observation_preserves_wire_payload_and_waits_for_confirmation(ctx): + from shared.native_schedule_sync import NativeScheduleSync + from roborock_local_server.bundled_backend.mqtt_broker_server.topic_bridge import DeviceTopicKey + + coordinator = NativeScheduleSync(ctx) + bridge = MqttTopicBridge( + host="127.0.0.1", port=1883, logger=logging.getLogger("native-sync-test"), + request_interceptor=timer_rpc.TimerRpcInterceptor(ctx), + request_observer=coordinator.observe_request, + ) + device = DeviceTopicKey(did="robot-did", mqtt_usr="robot-user") + bridge._remember_device_seen(device) + before = schedule_store.list_jobs(ctx, DUID) + sent = [] + + class Client: + async def publish(self, topic, payload, **kwargs): + sent.append((topic, payload)) + + async def exercise(): + for session, enabled in (("homebridge", False), ("oem", True), ("localrock", False)): + cloud = CloudTopicKey(rriot_u=session, mqtt_username=session, duid=DUID) + payload = _encode_request("upd_server_timer", [["native-timer", "on" if enabled else "off"]]) + message = SimpleNamespace(payload=payload, qos=0, retain=False) + await bridge._handle_cloud_message(Client(), message, cloud, cloud.topic_in) + assert sent[-1] == (device.topic_out, payload) + assert schedule_store.list_jobs(ctx, DUID) == before + assert coordinator._generation[DUID] == 3 + + asyncio.run(exercise()) + + +def test_retained_native_write_is_not_observed_as_a_new_change(ctx): + from shared.native_schedule_sync import NativeScheduleSync + from roborock_local_server.bundled_backend.mqtt_broker_server.topic_bridge import DeviceTopicKey + coordinator = NativeScheduleSync(ctx) + bridge = MqttTopicBridge(host="127.0.0.1", port=1883, logger=logging.getLogger("native-sync-test"), + request_observer=coordinator.observe_request) + bridge._remember_device_seen(DeviceTopicKey(did="robot-did", mqtt_usr="robot-user")) + cloud = CloudTopicKey(rriot_u="app", mqtt_username="app", duid=DUID) + class Client: + async def publish(self, *args, **kwargs): + pass + message = SimpleNamespace(payload=_encode_request("upd_server_timer", [["native-timer", "on"]]), qos=0, retain=True) + asyncio.run(bridge._handle_cloud_message(Client(), message, cloud, cloud.topic_in)) + assert not coordinator._dirty + + def test_bridge_answers_timer_requests_locally_and_does_not_forward_them(ctx): class Client: def __init__(self):