diff --git a/backend/apps/workflows/executor.py b/backend/apps/workflows/executor.py index 5d64f48b..a5229e1a 100644 --- a/backend/apps/workflows/executor.py +++ b/backend/apps/workflows/executor.py @@ -213,9 +213,13 @@ async def execute( # of a toggled-off workflow running itself. Turn it back on to run it. p_live = storage.get_workflow(wf.id) p_refusal = None - if p_live is not None and p_live.deleted_at is not None: + if p_live is None: + # A hard delete leaves nothing to look up, and reading that as "no objection" is how a + # deleted workflow still ran to the end and then wrote itself back to life. p_refusal = "Workflow deleted" - elif p_live is not None and not p_live.schedule.enabled: + elif p_live.deleted_at is not None: + p_refusal = "Workflow deleted" + elif not p_live.schedule.enabled: p_refusal = "Workflow is paused" if p_refusal is not None: p_skipped = WorkflowRun( diff --git a/backend/apps/workflows/storage.py b/backend/apps/workflows/storage.py index fced0452..01cda893 100644 --- a/backend/apps/workflows/storage.py +++ b/backend/apps/workflows/storage.py @@ -36,6 +36,13 @@ _missed_cache: list[MissedRun] = [] _cache_loaded = False _paused = False +# Ids deleted during this process's life. The cache hands out SHARED Workflow instances, so a run +# already in flight when the user deletes still holds one and writes it back when it finishes, and +# save_workflow used to recreate the file AND the cache entry, fully scheduled: the workflow rose +# from the dead every time, which is exactly the "it never dies" field report. Only needs to live in +# memory, because after a restart nothing holds a stale instance to write back. +p_deleted_ids: set[str] = set() + def _resolve_host_tz_name() -> str: """Best-effort IANA name for the host. Mirrors apps/service/client.py.""" @@ -158,8 +165,18 @@ def get_workflow(wid: str) -> Optional[Workflow]: return _workflow_cache.get(wid) -def save_workflow(wf: Workflow) -> Workflow: +def save_workflow(wf: Workflow, untrash: bool = False) -> Workflow: with _io_lock: + if wf.id in p_deleted_ids: + logger.info("ignoring a write-back for deleted workflow %s", wf.id) + return wf + # Trash is one-way too: only /restore passes untrash. Any other save carrying an older copy + # (a run that started before the user hit delete) would otherwise clear deleted_at and put + # the workflow back on the page with its schedule re-armed. + prior = _workflow_cache.get(wf.id) + if not untrash and prior is not None and prior.deleted_at is not None and wf.deleted_at is None: + logger.info("ignoring a write-back that would untrash workflow %s", wf.id) + return wf _ensure_dirs() _workflow_cache[wf.id] = wf p_atomic_write_json(_wf_path(wf.id), wf.model_dump(mode="json")) @@ -171,6 +188,9 @@ def reload_workflow(wid: str) -> Optional[Workflow]: 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: + if wid in p_deleted_ids: + _workflow_cache.pop(wid, None) + return None path = _wf_path(wid) if not os.path.exists(path): _workflow_cache.pop(wid, None) @@ -184,6 +204,7 @@ def reload_workflow(wid: str) -> Optional[Workflow]: def delete_workflow(wid: str) -> bool: with _io_lock: existed = wid in _workflow_cache + p_deleted_ids.add(wid) _workflow_cache.pop(wid, None) _runs_cache.pop(wid, None) wf_file = _wf_path(wid) @@ -217,6 +238,8 @@ def list_all_runs(limit: int = 200) -> list[WorkflowRun]: def record_run(run: WorkflowRun) -> WorkflowRun: with _io_lock: + if run.workflow_id in p_deleted_ids: + return run _ensure_dirs() arr = _runs_cache.setdefault(run.workflow_id, []) # Replace prior entry with same id if we're updating an in-flight run. diff --git a/backend/apps/workflows/workflows.py b/backend/apps/workflows/workflows.py index 6375f154..678c87c4 100644 --- a/backend/apps/workflows/workflows.py +++ b/backend/apps/workflows/workflows.py @@ -906,7 +906,7 @@ async def restore_workflow(workflow_id: str): if not wf or wf.deleted_at is None: raise HTTPException(status_code=404, detail="Workflow not in trash") wf.deleted_at = None - storage.save_workflow(wf) + storage.save_workflow(wf, untrash=True) enriched = _enriched(wf) try: from backend.apps.agents.core.ws_manager import ws_manager @@ -1453,6 +1453,10 @@ async def run_workflow_now(workflow_id: str, body: Optional[dict] = None): wf = storage.get_workflow(workflow_id) if not wf: raise HTTPException(status_code=404, detail="Workflow not found") + # The executor refuses a trashed workflow but writes no history for it, so without this the caller + # got run_id "" with a null status and no idea why nothing happened. + if wf.deleted_at is not None: + raise HTTPException(status_code=409, detail="This workflow is in Trash. Restore it to run it.") # executor.execute() owns the run record. Don't pre-create a stub here or we end up with two rows per manual fire (one orphan "running" row from this handler plus the real one from the executor). pre_ids = {r.id for r in storage.list_runs(wf.id, limit=10)} tested_signature = body.get("signature") if isinstance(body, dict) else None diff --git a/backend/tests/test_deleted_workflow_stays_dead.py b/backend/tests/test_deleted_workflow_stays_dead.py new file mode 100644 index 00000000..cfa2f75a --- /dev/null +++ b/backend/tests/test_deleted_workflow_stays_dead.py @@ -0,0 +1,156 @@ +"""A deleted workflow must never come back, by any route. + +Field report (Haik, 1.7.4): a scheduled workflow survived deleting it, wiping OpenSwarm data, +deleting the app, and reinstalling. It kept firing every 45 minutes. `delete_workflow` removes the +file and the cache entry, but `save_workflow` was an unconditional upsert that recreates BOTH, so any +write-back from a run that was already in flight resurrected it, fully scheduled. Runs that stall for +1200s make that window enormous, and each resurrection re-armed the timer, so it never died. +""" + +import asyncio +from datetime import datetime, timezone + +import pytest + +from backend.apps.workflows import storage +from backend.apps.workflows.models import Workflow, WorkflowRun, WorkflowStep + + +def p_make(title: str = "ghost") -> Workflow: + return Workflow( + title=title, + steps=[WorkflowStep(prompt="do a thing")], + schedule={"enabled": True, "kind": "interval", "every_minutes": 45}, + ) + + +@pytest.fixture(autouse=True) +def p_isolated_store(tmp_path, monkeypatch): + monkeypatch.setattr(storage, "DATA_DIR", str(tmp_path / "workflows")) + monkeypatch.setattr(storage, "RUNS_DIR", str(tmp_path / "workflows" / "runs")) + monkeypatch.setattr(storage, "PAUSED_FILE", str(tmp_path / "workflows" / "paused.json")) + monkeypatch.setattr(storage, "MISSED_FILE", str(tmp_path / "workflows" / "missed.json")) + monkeypatch.setattr(storage, "_workflow_cache", {}) + monkeypatch.setattr(storage, "_runs_cache", {}) + monkeypatch.setattr(storage, "_missed_cache", []) + monkeypatch.setattr(storage, "p_deleted_ids", set(), raising=False) + monkeypatch.setattr(storage, "_cache_loaded", True) + yield + + +def test_stale_save_after_delete_does_not_resurrect(): + """The exact Haik bug: an in-flight run holds the object, the user deletes, the run writes back.""" + wf = storage.save_workflow(p_make()) + in_flight = storage.get_workflow(wf.id) + assert in_flight is not None + + assert storage.delete_workflow(wf.id) is True + + in_flight.last_run_at = datetime.now(timezone.utc) + storage.save_workflow(in_flight) + + assert storage.get_workflow(wf.id) is None + assert [w.id for w in storage.list_workflows()] == [] + import os + assert not os.path.exists(storage._wf_path(wf.id)) + + +def test_repeated_stale_saves_never_resurrect(): + """It never dies: every later write-back must also bounce, not just the first.""" + wf = storage.save_workflow(p_make()) + stale = storage.get_workflow(wf.id) + storage.delete_workflow(wf.id) + for i in range(5): + stale.next_run_at = datetime.now(timezone.utc) + storage.save_workflow(stale) + assert storage.get_workflow(wf.id) is None, f"resurrected on write-back {i + 1}" + + +def test_record_run_after_delete_does_not_recreate_history(): + """Runs are deleted with the workflow; a late run row must not rebuild an orphan history file.""" + wf = storage.save_workflow(p_make()) + storage.delete_workflow(wf.id) + storage.record_run(WorkflowRun(workflow_id=wf.id, status="success")) + assert storage.list_runs(wf.id) == [] + import os + assert not os.path.exists(storage._runs_path(wf.id)) + + +def test_delete_does_not_block_a_different_workflow(): + """The tombstone is per id: deleting one must not stop anything else being saved.""" + dead = storage.save_workflow(p_make("dead")) + storage.delete_workflow(dead.id) + alive = storage.save_workflow(p_make("alive")) + assert storage.get_workflow(alive.id) is not None + assert [w.id for w in storage.list_workflows()] == [alive.id] + + +def test_executor_refuses_a_workflow_that_no_longer_exists(): + """A hard delete makes get_workflow return None, which the pause guard did not treat as refusal, + so a deleted workflow still ran to completion and then resurrected itself on write-back.""" + from backend.apps.workflows import executor + + wf = storage.save_workflow(p_make()) + storage.delete_workflow(wf.id) + + run = asyncio.run(executor.execute(wf, triggered_by="schedule")) + assert run.status == "skipped" + assert "delete" in (run.error or "").lower() + assert storage.get_workflow(wf.id) is None + + +def test_disable_schedule_on_a_deleted_workflow_stays_dead(): + """The scheduler disables end-of-life workflows by saving them; on a deleted one that is a + resurrection with enabled=False, which still shows up on the Workflows page.""" + from backend.apps.workflows import scheduler + + wf = storage.save_workflow(p_make()) + stale = storage.get_workflow(wf.id) + storage.delete_workflow(wf.id) + scheduler._disable_schedule(stale) + assert storage.get_workflow(wf.id) is None + assert [w.id for w in storage.list_workflows()] == [] + + +def test_a_stale_copy_cannot_untrash_a_workflow(): + """Trash is one-way. reload_workflow hands out a NEW instance, so a run that started before the + user hit delete can end up holding a copy whose deleted_at is still None; saving that copy put + the workflow back on the page with its schedule re-armed.""" + wf = storage.save_workflow(p_make()) + stale = storage.get_workflow(wf.id) + storage.reload_workflow(wf.id) # the cache now holds a DIFFERENT instance; the run kept `stale` + trashed = storage.get_workflow(wf.id) + assert trashed is not stale, "test models nothing unless the two copies really diverged" + assert stale.deleted_at is None + + trashed.deleted_at = datetime.now() + trashed.schedule.enabled = False + storage.save_workflow(trashed) + + stale.last_run_at = datetime.now(timezone.utc) + storage.save_workflow(stale) + + live = storage.get_workflow(wf.id) + assert live is not None and live.deleted_at is not None, "a stale copy untrashed it" + assert live.schedule.enabled is False, "the schedule was re-armed by a stale write-back" + + +def test_restore_can_still_untrash(): + """The one-way rule must not brick the Trash > Restore button.""" + wf = storage.save_workflow(p_make()) + trashed = storage.get_workflow(wf.id) + trashed.deleted_at = datetime.now() + storage.save_workflow(trashed) + + back = storage.reload_workflow(wf.id) + back.deleted_at = None + storage.save_workflow(back, untrash=True) + assert storage.get_workflow(wf.id).deleted_at is None + + +def test_delete_survives_a_reload_from_disk(): + """reload_workflow re-reads from disk; on a deleted id it must not repopulate the cache.""" + wf = storage.save_workflow(p_make()) + storage.delete_workflow(wf.id) + assert storage.reload_workflow(wf.id) is None + assert storage.get_workflow(wf.id) is None diff --git a/backend/tests/test_scheduled_stop_on_pause_trash.py b/backend/tests/test_scheduled_stop_on_pause_trash.py index ea2bd7ae..211f060e 100644 --- a/backend/tests/test_scheduled_stop_on_pause_trash.py +++ b/backend/tests/test_scheduled_stop_on_pause_trash.py @@ -93,8 +93,13 @@ def test_paused_scheduled_run_halts_at_next_step(make_wf, fake_agent_manager, mo assert fake_agent_manager.sent_messages == ["step1"] -def test_pause_does_not_halt_a_manual_run(make_wf, fake_agent_manager, monkeypatch): - """Pausing the SCHEDULE must not kill a manual Run Now that's in flight.""" +def test_pause_halts_even_a_manual_run(make_wf, fake_agent_manager, monkeypatch): + """Off means off on every path, including a manual Run Now already in flight. + + This used to assert the opposite, that pausing the schedule left a manual run alone. Eric's call + is that a workflow switched off must not keep running by any route, so the switch now stops the + run at the next step boundary whatever started it. + """ from backend.apps.workflows import storage, executor wf = p_three_step_wf(make_wf) storage.save_workflow(wf) @@ -106,8 +111,8 @@ def test_pause_does_not_halt_a_manual_run(make_wf, fake_agent_manager, monkeypat p_mutate_after_first_step(monkeypatch, fake_agent_manager, pause) run = p_run(executor.execute(wf, triggered_by="manual")) - assert run.status == "success" - assert fake_agent_manager.sent_messages == ["step1", "step2", "step3"] # all ran + assert run.status != "success" + assert fake_agent_manager.sent_messages == ["step1"] # stopped at the boundary after the switch def test_stop_active_run_signals_and_returns_session(make_wf, fake_agent_manager, monkeypatch):