From 8a84927ce949ed3d0930a9ed815f741acc07bdd4 Mon Sep 17 00:00:00 2001 From: ciregenz Date: Thu, 6 Aug 2026 14:32:21 -0700 Subject: [PATCH] [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 | 15 ++++++ backend/apps/workflows/workflows.py | 21 ++++++++ backend/tests/test_cloud_workflow_target.py | 54 +++++++++++++++++++++ 3 files changed, 90 insertions(+) diff --git a/backend/apps/workflows/storage.py b/backend/apps/workflows/storage.py index 05e9df3e..fced0452 100644 --- a/backend/apps/workflows/storage.py +++ b/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 diff --git a/backend/apps/workflows/workflows.py b/backend/apps/workflows/workflows.py index b646ffc9..6375f154 100644 --- a/backend/apps/workflows/workflows.py +++ b/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() diff --git a/backend/tests/test_cloud_workflow_target.py b/backend/tests/test_cloud_workflow_target.py index d38849ff..cfa252b6 100644 --- a/backend/tests/test_cloud_workflow_target.py +++ b/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