[eric] workflows: a schedule edit on a cloud-hosted workflow now pauses or repushes the CLOUD copy, unreachable fails honest with disk-truth rollback

backend/apps/workflows/storage.py

@@ -166,6 +166,21 @@ def save_workflow(wf: Workflow) -> Workflow:
     return wf
 
 
+def reload_workflow(wid: str) -> Optional[Workflow]:
+    """Re-read one workflow from disk, discarding in-memory mutations. The cache hands out SHARED
+    instances, so a handler that mutated one and then failed must roll back through here or the
+    unsaved change lingers until any later save persists it by accident."""
+    with _io_lock:
+        path = _wf_path(wid)
+        if not os.path.exists(path):
+            _workflow_cache.pop(wid, None)
+            return None
+        with open(path, "r", encoding="utf-8") as f:
+            wf = Workflow(**json.load(f))
+        _workflow_cache[wid] = wf
+        return wf
+
+
 def delete_workflow(wid: str) -> bool:
     with _io_lock:
         existed = wid in _workflow_cache

backend/apps/workflows/workflows.py

@@ -209,6 +209,23 @@ async def list_workflows(dashboard_id: Optional[str] = None):
     return {"workflows": [_enriched(w) for w in items]}
 
 
+async def p_sync_cloud_copy(wf: Workflow, data: dict) -> None:
+    """A cloud-hosted workflow's schedule truth lives in the CLOUD; a PATCH that only edits the
+    local copy pauses nothing (the 'toggled the schedule off but it still runs' bug). Push the
+    edit up before persisting locally; if the cloud cannot be reached, roll the shared cached
+    instance back to disk truth and fail the PATCH so the UI never shows a state the cloud ignores."""
+    if wf.execution_target != "cloud" or not wf.cloud_workflow_id:
+        return
+    if not any(k in data for k in ("schedule", "steps", "title")):
+        return
+    from backend.apps.workflows.cloud.handover import hand_to_cloud
+
+    outcome = await hand_to_cloud(wf, enabled=wf.schedule.enabled)
+    if not outcome.ok:
+        storage.reload_workflow(wf.id)
+        raise HTTPException(status_code=502, detail=outcome.message or "The cloud copy could not be updated; try again.")
+
+
 def _normalize_schedule_state(wf: Workflow, source_allowed_tools: Optional[list[str]] = None) -> None:
     if wf.schedule.timezone == "local" and wf.schedule.enabled:
         wf.schedule.timezone = scheduler.host_timezone_name()
@@ -798,6 +815,8 @@ async def update_workflow(
         await p_relabel_steps(wf, before_draft, wf.draft_steps, wf.model)
         wf.updated_at = datetime.now()
         _normalize_schedule_state(wf)
+        # Steps went to the DRAFT, not live, so only the non-steps fields need the cloud copy synced; commit pushes the steps.
+        await p_sync_cloud_copy(wf, {k: v for k, v in data.items() if k != "steps"})
         storage.save_workflow(wf)
         enriched = _enriched(wf)
         try:
@@ -818,6 +837,7 @@ async def update_workflow(
     if not wf.icon:
         wf.icon = _derive_icon(wf)
     _normalize_schedule_state(wf)
+    await p_sync_cloud_copy(wf, data)
     storage.save_workflow(wf)
     audit.log_change(wf.id, "user", before, wf.model_dump(mode="json"))
     scheduler.kick()
@@ -1184,6 +1204,7 @@ async def commit_draft(workflow_id: str, body: Optional[DraftCommitBody] = None)
     p_sync_model_on_save(wf, body.model if body else None)
     if not (body and body.keep_session):
         await p_end_edit_session(wf)
+    await p_sync_cloud_copy(wf, {"steps": wf.steps})
     storage.save_workflow(wf)
     audit.log_change(wf.id, "user", before, wf.model_dump(mode="json"))
     scheduler.kick()

backend/tests/test_cloud_workflow_target.py

@@ -304,3 +304,57 @@ async def test_a_workflow_the_cloud_still_holds_cannot_be_trashed_into_a_ghost(m
     assert caught.value.status_code == 409
     # Deleting it locally would leave a hosted copy running on its own schedule, billing a user who cannot see it.
     assert storage.get_workflow(wf.id).deleted_at is None
+
+
+@pytest.mark.asyncio
+async def test_toggling_a_cloud_schedule_off_pauses_the_cloud_copy(monkeypatch):
+    """The live report this seals: schedule toggled off, the workflow still ran. The cloud held the
+    timer and the PATCH edited only the local copy, which pauses nothing."""
+    from backend.apps.workflows.models import WorkflowUpdate
+    from backend.apps.workflows.workflows import update_workflow
+
+    wf = p_wf(execution_target="cloud", cloud_workflow_id="cloud-1")
+    seen = p_answer(
+        monkeypatch,
+        lambda method, path, body: p_hosted(enabled=False, next_run_at=None)
+        if path.endswith("/enable")
+        else p_hosted(),
+    )
+    await update_workflow(wf.id, WorkflowUpdate(schedule=p_sched(enabled=False)), if_match=None)
+    enable_calls = [(p, b) for _, p, b in seen if p.endswith("/enable")]
+    assert enable_calls, f"the cloud row was never paused; calls: {[p for _, p, _ in seen]}"
+    assert enable_calls[-1][1] == {"enabled": False}
+    fresh = storage.get_workflow(wf.id)
+    assert fresh.schedule.enabled is False
+    assert fresh.next_run_at is None
+
+
+@pytest.mark.asyncio
+async def test_an_unreachable_cloud_fails_the_toggle_instead_of_lying(monkeypatch):
+    """A toggle the cloud never heard must not render as Off while the cloud keeps firing."""
+    from backend.apps.workflows.models import WorkflowUpdate
+    from backend.apps.workflows.workflows import update_workflow
+
+    wf = p_wf(execution_target="cloud", cloud_workflow_id="cloud-1")
+
+    def boom(method, path, body):
+        raise cloud.CloudUnreachable("no route")
+
+    p_answer(monkeypatch, boom)
+    with pytest.raises(HTTPException) as exc:
+        await update_workflow(wf.id, WorkflowUpdate(schedule=p_sched(enabled=False)), if_match=None)
+    assert exc.value.status_code == 502
+    # The shared cached instance was mutated before the push; disk truth must win back.
+    assert storage.get_workflow(wf.id).schedule.enabled is True
+
+
+@pytest.mark.asyncio
+async def test_a_device_schedule_patch_never_talks_to_the_cloud(monkeypatch):
+    from backend.apps.workflows.models import WorkflowUpdate
+    from backend.apps.workflows.workflows import update_workflow
+
+    wf = p_wf()
+    seen = p_answer(monkeypatch, lambda method, path, body: p_hosted())
+    await update_workflow(wf.id, WorkflowUpdate(schedule=p_sched(enabled=False)), if_match=None)
+    assert seen == []
+    assert storage.get_workflow(wf.id).schedule.enabled is False