From 6f32861b418571cbebd8a791ce8c6477d9e22a22 Mon Sep 17 00:00:00 2001 From: ciregenz Date: Fri, 31 Jul 2026 11:42:52 -0700 Subject: [PATCH] [eric] workflows: add execution_target so the local scheduler never fires or rolls a cloud-hosted timer --- backend/apps/workflows/models.py | 2 + backend/apps/workflows/scheduler.py | 13 ++ .../tests/test_schedule_execution_target.py | 113 ++++++++++++++++++ 3 files changed, 128 insertions(+) create mode 100644 backend/tests/test_schedule_execution_target.py diff --git a/backend/apps/workflows/models.py b/backend/apps/workflows/models.py index 0e56bd4c..314145bf 100644 --- a/backend/apps/workflows/models.py +++ b/backend/apps/workflows/models.py @@ -91,6 +91,8 @@ class Workflow(BaseModel): steps: list[WorkflowStep] = Field(default_factory=list) actions: ActionsConfig = Field(default_factory=ActionsConfig) schedule: ScheduleConfig = Field(default_factory=ScheduleConfig) + # Where a SCHEDULED fire runs. "cloud" hands the timer to our servers outright, so this machine must never fire it nor roll next_run_at, or the same slot runs twice. Manual Run-now always stays local. + execution_target: Literal["device", "cloud"] = "device" permissions: list[PermissionTier] = Field( default_factory=lambda: [PermissionTier(kind="notify")] ) diff --git a/backend/apps/workflows/scheduler.py b/backend/apps/workflows/scheduler.py index 5bb1f303..671329f0 100644 --- a/backend/apps/workflows/scheduler.py +++ b/backend/apps/workflows/scheduler.py @@ -289,12 +289,21 @@ def _disable_schedule(wf: Workflow) -> None: storage.save_workflow(wf) +def p_timer_is_ours(wf: Workflow) -> bool: + """Whether this process owns a workflow's timer. Both the fire loop and the sleep math must + agree: if only one of them skipped cloud-hosted workflows, an overdue one would never roll + forward and the loop would spin at its 1s floor forever.""" + return wf.execution_target != "cloud" + + async def _tick() -> None: now_utc = datetime.now(timezone.utc) if storage.get_paused(): return due: list[Workflow] = [] for wf in storage.list_workflows(): + if not p_timer_is_ours(wf): + continue if not wf.schedule.enabled: continue if not is_schedule_configured(wf.schedule): @@ -330,6 +339,8 @@ def seconds_to_next_fire() -> Optional[float]: now_utc = datetime.now(timezone.utc) soonest: Optional[datetime] = None for wf in storage.list_workflows(): + if not p_timer_is_ours(wf): + continue if not wf.schedule.enabled: continue nra = _as_utc(wf.next_run_at) @@ -452,6 +463,8 @@ def reconcile_on_startup() -> None: """ now_utc = datetime.now(timezone.utc) for wf in storage.list_workflows(): + if not p_timer_is_ours(wf): + continue if not wf.schedule.enabled: wf.next_run_at = None storage.save_workflow(wf) diff --git a/backend/tests/test_schedule_execution_target.py b/backend/tests/test_schedule_execution_target.py new file mode 100644 index 00000000..5c2af802 --- /dev/null +++ b/backend/tests/test_schedule_execution_target.py @@ -0,0 +1,113 @@ +"""A cloud-hosted workflow's timer belongs to the server, not to this machine. + +The failure these guard against is a double run: if the laptop is awake when a +cloud-hosted slot comes due, both the laptop and the server fire it, and the +user gets two of everything. Ownership is expressed once, on +Workflow.execution_target, and every reader of the schedule must honour it. + +Run: + cd backend && .venv/bin/python -m pytest tests/test_schedule_execution_target.py -v +""" + +from __future__ import annotations + +import asyncio +from datetime import datetime, timedelta, timezone + +import pytest + + +@pytest.fixture(autouse=True) +def p_wf_env(isolated_workflows_data, reset_scheduler_state): + yield + + +def test_default_target_is_this_device(make_wf): + """Every existing workflow predates the field, so the default has to keep them local.""" + wf = make_wf() + assert wf.execution_target == "device" + + +@pytest.mark.asyncio +async def test_tick_does_not_fire_a_cloud_workflow(make_wf, monkeypatch): + from backend.apps.workflows import storage, scheduler + fired: list[str] = [] + + async def p_capture(wf, scheduled_for=None): + fired.append(wf.id) + + monkeypatch.setattr(scheduler, "_fire", p_capture) + overdue = datetime.now(timezone.utc) - timedelta(minutes=5) + + local = make_wf(execution_target="device", next_run_at=overdue) + cloud = make_wf(execution_target="cloud", next_run_at=overdue) + storage.save_workflow(local) + storage.save_workflow(cloud) + + await scheduler._tick() + # _tick hands each fire to create_task, so nothing has actually run until we yield the loop. + await asyncio.sleep(0) + + assert local.id in fired, "a device workflow must still fire; the test is vacuous otherwise" + assert cloud.id not in fired + + +@pytest.mark.asyncio +async def test_tick_does_not_roll_a_cloud_next_run_at(make_wf, monkeypatch): + """Rolling the timer forward locally is its own bug even when nothing fires: the server owns + that field, and a local write silently competes with it.""" + from backend.apps.workflows import storage, scheduler + + async def p_noop(wf, scheduled_for=None): + return None + + monkeypatch.setattr(scheduler, "_fire", p_noop) + overdue = datetime.now(timezone.utc) - timedelta(minutes=5) + cloud = make_wf(execution_target="cloud", next_run_at=overdue) + storage.save_workflow(cloud) + + await scheduler._tick() + + after = storage.get_workflow(cloud.id) + assert after.next_run_at == cloud.next_run_at + + +def test_sleep_math_ignores_cloud_workflows(make_wf): + """_tick and seconds_to_next_fire must agree. If only _tick skipped cloud workflows, an + overdue one would stay overdue forever and pin the loop at its 1s floor, burning a core.""" + from backend.apps.workflows import storage, scheduler + overdue = datetime.now(timezone.utc) - timedelta(minutes=5) + storage.save_workflow(make_wf(execution_target="cloud", next_run_at=overdue)) + + assert scheduler.seconds_to_next_fire() is None + assert scheduler._seconds_until_next() == 60.0 + + +def test_reconcile_leaves_cloud_workflows_untouched(make_wf): + """A closed laptop did not "miss" a cloud run; the server ran it. Capturing it would offer the + user a review card for work that already happened, and rewrite a server-owned field.""" + from backend.apps.workflows import storage, scheduler + anchor = datetime.now(timezone.utc) - timedelta(days=3) + # occurrences_between never enumerates fires from before the workflow existed, so it has to be old. + born = datetime.now(timezone.utc) - timedelta(days=10) + cloud = make_wf(execution_target="cloud", next_run_at=anchor, created_at=born) + storage.save_workflow(cloud) + + scheduler.reconcile_on_startup() + + assert storage.list_missed() == [] + after = storage.get_workflow(cloud.id) + assert after.next_run_at == cloud.next_run_at + + +def test_reconcile_still_captures_device_workflows(make_wf): + """The discriminating half: the same walk must keep working for local workflows.""" + from backend.apps.workflows import storage, scheduler + anchor = datetime.now(timezone.utc) - timedelta(days=3) + born = datetime.now(timezone.utc) - timedelta(days=10) + local = make_wf(execution_target="device", next_run_at=anchor, created_at=born) + storage.save_workflow(local) + + scheduler.reconcile_on_startup() + + assert storage.list_missed() != []