[aidan] test/schedule: cover executor pipeline, storage durability, and recurrence gaps

backend/tests/conftest.py

@@ -4,10 +4,19 @@ Isolate the persistent browser-skill store (and metrics) into throwaway temp
 dirs for the whole test session, so tests never write skills/metrics into the
 real ~/Library/Application Support/OpenSwarm/data tree (which would pollute the
 dev machine and let a stale persisted skill leak across test runs).
+
+The workflow fixtures below (isolated_workflows_data, reset_scheduler_state,
+make_wf, fake_agent_manager) are deliberately NOT autouse: only the scheduled-
+workflows suites opt in, so they never touch unrelated browser/service tests.
+The two pre-existing workflow suites still carry their own local copies; new
+suites lean on these shared ones instead of re-pasting the boilerplate.
 """
 
+import asyncio
 import os
 import tempfile
+from datetime import datetime, timezone
+from types import SimpleNamespace
 
 import pytest
 
@@ -38,3 +47,131 @@ def _isolate_browser_state(monkeypatch):
     _reset()
     yield
     _reset()
+
+
+@pytest.fixture
+def isolated_workflows_data(monkeypatch, tmp_path):
+    """Point the workflow store at a throwaway tmpdir and start every test with
+    empty in-memory caches, so nothing leaks into a real install or across
+    tests. Mirrors the local fixtures in test_workflows_semantics.py and
+    test_schedule_e2e.py; opt in by requesting this fixture (or depend on it
+    from a file-local autouse)."""
+    from backend.apps.workflows import storage as p_storage
+    from backend.apps.workflows import audit as p_audit
+    from backend.apps.workflows import escalation as p_escalation
+    monkeypatch.setattr(p_storage, "DATA_DIR", str(tmp_path / "workflows"))
+    monkeypatch.setattr(p_storage, "RUNS_DIR", str(tmp_path / "workflows" / "runs"))
+    monkeypatch.setattr(p_storage, "PAUSED_FILE", str(tmp_path / "workflows" / "paused.json"))
+    monkeypatch.setattr(p_storage, "MISSED_FILE", str(tmp_path / "workflows" / "missed.json"))
+    monkeypatch.setattr(p_storage, "_workflow_cache", {})
+    monkeypatch.setattr(p_storage, "_runs_cache", {})
+    monkeypatch.setattr(p_storage, "_missed_cache", [])
+    monkeypatch.setattr(p_storage, "_cache_loaded", False)
+    monkeypatch.setattr(p_storage, "_paused", False)
+    monkeypatch.setattr(p_audit, "AUDIT_DIR", str(tmp_path / "workflows" / "audit"))
+    p_escalation._tasks.clear()
+    p_escalation._state.clear()
+    yield
+
+
+@pytest.fixture
+def reset_scheduler_state(monkeypatch):
+    """Reset the scheduler + executor module globals that otherwise survive
+    across tests and cause order-dependent flakes: a stale _wake Event bound to
+    a dead loop, a lingering _loop_task, a cached host tz, or a leftover entry
+    in the _running / control maps."""
+    from backend.apps.workflows import scheduler as p_scheduler
+    from backend.apps.workflows import executor as p_executor
+    p_scheduler._loop_task = None
+    p_scheduler._wake = asyncio.Event()
+    monkeypatch.setattr(p_scheduler, "_host_tz_cache", None)
+    p_executor._running.clear()
+    p_executor._run_control.clear()
+    p_executor._run_pause_override.clear()
+    yield
+    p_executor._running.clear()
+    p_executor._run_control.clear()
+    p_executor._run_pause_override.clear()
+
+
+@pytest.fixture
+def make_wf():
+    """Factory for a Workflow with a sane default daily-9am-LA schedule;
+    override any field via kwargs."""
+    def p_build(**overrides):
+        from backend.apps.workflows.models import Workflow, ScheduleConfig, WorkflowStep
+        base = dict(
+            title="t",
+            steps=[WorkflowStep(text="hi")],
+            schedule=ScheduleConfig(
+                enabled=True, repeat_unit="day", repeat_every=1,
+                hour=9, minute=0, timezone="America/Los_Angeles",
+            ),
+        )
+        base.update(overrides)
+        return Workflow(**base)
+    return p_build
+
+
+class FakeAgentManager:
+    """Stand-in for agent_manager so executor.execute() drives a full run
+    without a live LLM. Records launched configs + every step prompt sent, and
+    lets a test script the per-step terminal status (the value
+    _await_session_idle reads): default 'completed' (-> advance), or 'error' /
+    'stopped' to exercise the failure paths.
+    """
+
+    def __init__(self):
+        self.sessions: dict[str, SimpleNamespace] = {}
+        self.tasks: dict[str, object] = {}
+        self.launched_configs: list[object] = []
+        self.sent_messages: list[str] = []
+        # statuses[i] is the session status after the i-th send_message; absent
+        # entries default to 'completed'. cost_usd lands on the run.
+        self.statuses: list[str] = []
+        self.cost_usd: float = 0.0
+
+    async def launch_agent(self, config):
+        self.launched_configs.append(config)
+        sid = f"sess-{len(self.sessions)}"
+        sess = SimpleNamespace(id=sid, status="completed", cost_usd=self.cost_usd, messages=[])
+        self.sessions[sid] = sess
+        return sess
+
+    async def send_message(self, session_id, text, hidden=False):
+        i = len(self.sent_messages)
+        self.sent_messages.append(text)
+        sess = self.sessions.get(session_id)
+        if sess is not None:
+            sess.status = self.statuses[i] if i < len(self.statuses) else "completed"
+
+    async def close_session(self, session_id):
+        self.sessions.pop(session_id, None)
+
+
+@pytest.fixture
+def fake_agent_manager(monkeypatch):
+    """Patch the agent_manager seam (plus the ws + notifier side effects) that
+    executor.execute imports lazily, and hand back the FakeAgentManager so the
+    test can script statuses and assert on launched configs / sent steps."""
+    from backend.apps.agents import agent_manager as p_am
+    from backend.apps.agents.core.ws_manager import ws_manager as p_ws
+    from backend.apps.workflows import notifier as p_notifier
+
+    fake = FakeAgentManager()
+    monkeypatch.setattr(p_am.agent_manager, "sessions", fake.sessions)
+    monkeypatch.setattr(p_am.agent_manager, "tasks", fake.tasks)
+    monkeypatch.setattr(p_am.agent_manager, "launch_agent", fake.launch_agent)
+    monkeypatch.setattr(p_am.agent_manager, "send_message", fake.send_message)
+    monkeypatch.setattr(p_am.agent_manager, "close_session", fake.close_session)
+
+    async def p_noop_async(*args, **kwargs):
+        return None
+
+    monkeypatch.setattr(p_am, "set_workflow_approval_memory", lambda *a, **k: None)
+    monkeypatch.setattr(p_am, "set_workflow_approval_step", lambda *a, **k: None)
+    monkeypatch.setattr(p_am, "clear_workflow_approval_memory", lambda *a, **k: None)
+    monkeypatch.setattr(p_am, "get_workflow_step_usage", lambda *a, **k: {})
+    monkeypatch.setattr(p_ws, "broadcast_global", p_noop_async)
+    monkeypatch.setattr(p_notifier, "notify_run_complete", p_noop_async)
+    return fake

backend/tests/test_executor_pipeline.py

@@ -0,0 +1,178 @@
+"""Executor pipeline integration: drive executor.execute() end to end with a
+faked agent_manager (the fake_agent_manager fixture) and real on-disk storage.
+
+The existing semantics suite covers the cost-cap skip, p_ran_late in isolation,
+and the _persist_run_fields merge/delete races. This fills the gap those leave:
+the full run lifecycle through a scripted agent, the terminal-status matrix,
+runs_count accounting, step ordering, and the _running lock release on a persist
+failure (the crash-hardening fix).
+
+Run:
+    cd backend && .venv/bin/python -m pytest tests/test_executor_pipeline.py -v
+"""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _wf_env(isolated_workflows_data, reset_scheduler_state):
+    yield
+
+
+def _run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+# --- terminal-status matrix --------------------------------------------------
+
+def test_successful_run_records_success(make_wf, fake_agent_manager):
+    from backend.apps.workflows import storage, executor
+    wf = make_wf()
+    storage.save_workflow(wf)
+    run = _run(executor.execute(wf, triggered_by="schedule"))
+    assert run.status == "success"
+    assert run.finished_at is not None
+    stored = storage.list_runs(wf.id, limit=10)
+    assert stored[0].status == "success"
+    assert storage.get_workflow(wf.id).last_run_status == "success"
+
+
+def test_agent_error_marks_failure(make_wf, fake_agent_manager):
+    from backend.apps.workflows import storage, executor
+    fake_agent_manager.statuses = ["error"]
+    wf = make_wf()
+    storage.save_workflow(wf)
+    run = _run(executor.execute(wf, triggered_by="schedule"))
+    assert run.status == "failure"
+    assert run.error == "Agent session entered error state"
+
+
+def test_scheduled_run_late_start_marks_ran_late(make_wf, fake_agent_manager):
+    """A scheduled fire whose slot is already >5min in the past when it starts
+    is classified ran_late, not success, even though the agent succeeds."""
+    from backend.apps.workflows import storage, executor
+    wf = make_wf()
+    storage.save_workflow(wf)
+    slot = datetime.now(timezone.utc) - timedelta(minutes=10)
+    run = _run(executor.execute(wf, triggered_by="schedule", scheduled_for=slot))
+    assert run.status == "ran_late"
+    assert storage.get_workflow(wf.id).last_run_status == "ran_late"
+
+
+def test_overlap_skipped_when_already_running(make_wf, fake_agent_manager):
+    """A second fire while a run for the same workflow holds _running is skipped
+    with a clear reason, never launching a second agent."""
+    from backend.apps.workflows import storage, executor
+    wf = make_wf()
+    storage.save_workflow(wf)
+    executor._running[wf.id] = "in-flight-run"
+    run = _run(executor.execute(wf, triggered_by="schedule"))
+    assert run.status == "skipped"
+    assert run.error == "Previous run still active"
+    assert fake_agent_manager.launched_configs == []
+
+
+# --- runs_count accounting ---------------------------------------------------
+
+def test_schedule_run_bumps_runs_count(make_wf, fake_agent_manager):
+    from backend.apps.workflows import storage, executor
+    wf = make_wf()
+    storage.save_workflow(wf)
+    _run(executor.execute(wf, triggered_by="schedule"))
+    assert storage.get_workflow(wf.id).schedule.runs_count == 1
+
+
+def test_manual_run_does_not_bump_runs_count(make_wf, fake_agent_manager):
+    """Manual runs are free presses of the Run button; they must not count
+    against max_runs."""
+    from backend.apps.workflows import storage, executor
+    wf = make_wf()
+    storage.save_workflow(wf)
+    _run(executor.execute(wf, triggered_by="manual"))
+    assert storage.get_workflow(wf.id).schedule.runs_count == 0
+
+
+# --- step iteration ----------------------------------------------------------
+
+def test_steps_sent_in_order(make_wf, fake_agent_manager):
+    from backend.apps.workflows import storage, executor
+    from backend.apps.workflows.models import WorkflowStep
+    wf = make_wf(steps=[WorkflowStep(text="one"), WorkflowStep(text="two"), WorkflowStep(text="three")])
+    storage.save_workflow(wf)
+    _run(executor.execute(wf, triggered_by="schedule"))
+    assert fake_agent_manager.sent_messages == ["one", "two", "three"]
+
+
+def test_disabled_and_blank_steps_are_skipped(make_wf, fake_agent_manager):
+    from backend.apps.workflows import storage, executor
+    from backend.apps.workflows.models import WorkflowStep
+    wf = make_wf(steps=[
+        WorkflowStep(text="run me"),
+        WorkflowStep(text="muted", enabled=False),
+        WorkflowStep(text="   "),
+        WorkflowStep(text="me too"),
+    ])
+    storage.save_workflow(wf)
+    _run(executor.execute(wf, triggered_by="schedule"))
+    assert fake_agent_manager.sent_messages == ["run me", "me too"]
+
+
+def test_mid_sequence_error_halts_remaining_steps(make_wf, fake_agent_manager):
+    """An error on step 2 stops the run; step 3 is never dispatched."""
+    from backend.apps.workflows import storage, executor
+    from backend.apps.workflows.models import WorkflowStep
+    fake_agent_manager.statuses = ["completed", "error"]
+    wf = make_wf(steps=[WorkflowStep(text="s1"), WorkflowStep(text="s2"), WorkflowStep(text="s3")])
+    storage.save_workflow(wf)
+    run = _run(executor.execute(wf, triggered_by="schedule"))
+    assert run.status == "failure"
+    assert fake_agent_manager.sent_messages == ["s1", "s2"]
+
+
+def test_no_runnable_steps_fails(make_wf, fake_agent_manager):
+    from backend.apps.workflows import storage, executor
+    from backend.apps.workflows.models import WorkflowStep
+    wf = make_wf(steps=[WorkflowStep(text="", enabled=True)])
+    storage.save_workflow(wf)
+    run = _run(executor.execute(wf, triggered_by="schedule"))
+    assert run.status == "failure"
+    assert "no steps" in (run.error or "").lower()
+    assert fake_agent_manager.launched_configs == []
+
+
+# --- crash hardening: the lock must always be released -----------------------
+
+def test_running_lock_released_on_persist_failure(make_wf, fake_agent_manager, monkeypatch):
+    """If persisting run fields throws after _running is claimed, the finally
+    must still release the lock; otherwise the workflow is wedged 'running'
+    forever and every future fire is skipped. Guards the fix that moved the
+    running-persist inside the try whose finally frees _running."""
+    from backend.apps.workflows import storage, executor
+    wf = make_wf()
+    storage.save_workflow(wf)
+
+    def p_boom(*args, **kwargs):
+        raise RuntimeError("disk gone")
+
+    monkeypatch.setattr(executor, "_persist_run_fields", p_boom)
+    with pytest.raises(RuntimeError):
+        _run(executor.execute(wf, triggered_by="schedule"))
+    assert wf.id not in executor._running
+
+
+def test_resolved_config_uses_workflow_model_and_tools(make_wf, fake_agent_manager):
+    """The launched AgentConfig reflects the workflow's model and, when frozen,
+    its configured tool set rather than the default surface."""
+    from backend.apps.workflows import storage, executor
+    from backend.apps.workflows.models import ActionsConfig
+    wf = make_wf(model="opus", actions=ActionsConfig(freeze=True, configured_sets=["Read", "Grep"]))
+    storage.save_workflow(wf)
+    _run(executor.execute(wf, triggered_by="schedule"))
+    config = fake_agent_manager.launched_configs[0]
+    assert config.model == "opus"
+    assert config.allowed_tools == ["Read", "Grep"]

backend/tests/test_schedule_recovery.py

@@ -0,0 +1,59 @@
+"""Recovery + missed-run edges the e2e suite leaves uncovered.
+
+test_schedule_e2e already covers reconcile capture, the over-cap collapse, and
+the friendly stuck-run message. This adds the boundaries it skips: the summary
+heal on a stuck run, and the exactly-at-cap / empty cases of _capture_missed.
+
+Run:
+    cd backend && .venv/bin/python -m pytest tests/test_schedule_recovery.py -v
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _wf_env(isolated_workflows_data, reset_scheduler_state):
+    yield
+
+
+def test_stuck_run_heals_workflow_summary(make_wf):
+    """When the dead 'running' run is also the workflow's last run, the reaper
+    must fix the summary too, not just the run row, or the detail header keeps
+    showing a spinner forever."""
+    from backend.apps.workflows import storage, scheduler
+    from backend.apps.workflows.models import WorkflowRun
+    wf = make_wf()
+    run = WorkflowRun(workflow_id=wf.id, status="running")
+    wf.last_run_id = run.id
+    wf.last_run_status = "running"
+    storage.save_workflow(wf)
+    storage.record_run(run)
+
+    scheduler._mark_stuck_runs_failed()
+
+    healed = storage.get_workflow(wf.id)
+    assert healed.last_run_status == "failure"
+
+
+def test_capture_missed_exactly_at_cap_keeps_all_no_skipped(make_wf):
+    from backend.apps.workflows import storage, scheduler
+    wf = make_wf()
+    storage.save_workflow(wf)
+    base = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    missed = [base + timedelta(minutes=15 * i) for i in range(scheduler.PER_WORKFLOW_MISSED_CAP)]
+    scheduler._capture_missed(wf, missed)
+    assert len(storage.list_missed()) == scheduler.PER_WORKFLOW_MISSED_CAP
+    assert [r for r in storage.list_runs(wf.id, limit=50) if r.status == "skipped"] == []
+
+
+def test_capture_missed_empty_is_noop(make_wf):
+    from backend.apps.workflows import storage, scheduler
+    wf = make_wf()
+    storage.save_workflow(wf)
+    scheduler._capture_missed(wf, [])
+    assert storage.list_missed() == []
+    assert storage.list_runs(wf.id) == []

backend/tests/test_schedule_recurrence.py

@@ -0,0 +1,121 @@
+"""Pure recurrence + validator units not already covered by the semantics
+suite (which owns the DST, monthly, daily/weekly-interval, and occurrence
+timezone/end-condition cases). These functions take explicit reference
+datetimes and touch no disk, so they need no scaffolding.
+
+Run:
+    cd backend && .venv/bin/python -m pytest tests/test_schedule_recurrence.py -v
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+
+def _sched(**overrides):
+    from backend.apps.workflows.models import ScheduleConfig
+    base = dict(enabled=True, repeat_unit="day", repeat_every=1, hour=9, minute=0, timezone="UTC")
+    base.update(overrides)
+    return ScheduleConfig(**base)
+
+
+# --- minute / hour intervals (day/week/month intervals live in semantics) ----
+
+def test_minute_interval_steps_by_repeat_every_anchored():
+    from backend.apps.workflows.scheduler import _next_fire_after
+    sched = _sched(repeat_unit="minute", repeat_every=30)
+    anchor = datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc)
+    # Asking from 00:10 on the same grid -> next 30-min point is 00:30.
+    ref = datetime(2026, 6, 1, 0, 10, tzinfo=timezone.utc)
+    assert _next_fire_after(sched, ref, anchor) == datetime(2026, 6, 1, 0, 30, tzinfo=timezone.utc)
+
+
+def test_minute_repeat_every_floored_at_15():
+    """A sub-15 interval would be a token-burning loop; the floor lifts it to
+    15 (enforced both by the validator and the scheduler's own max(15, ...))."""
+    from backend.apps.workflows.scheduler import _next_fire_after
+    sched = _sched(repeat_unit="minute", repeat_every=5)
+    anchor = datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc)
+    ref = datetime(2026, 6, 1, 0, 1, tzinfo=timezone.utc)
+    assert _next_fire_after(sched, ref, anchor) == datetime(2026, 6, 1, 0, 15, tzinfo=timezone.utc)
+
+
+def test_hourly_interval_fires_at_configured_minute():
+    from backend.apps.workflows.scheduler import _next_fire_after
+    sched = _sched(repeat_unit="hour", repeat_every=3, minute=20)
+    anchor = datetime(2026, 6, 1, 0, 20, tzinfo=timezone.utc)
+    ref = datetime(2026, 6, 1, 1, 0, tzinfo=timezone.utc)
+    assert _next_fire_after(sched, ref, anchor) == datetime(2026, 6, 1, 3, 20, tzinfo=timezone.utc)
+
+
+# --- p_first_after grid math -------------------------------------------------
+
+def test_p_first_after_before_anchor_returns_anchor():
+    from backend.apps.workflows.scheduler import p_first_after
+    anchor = datetime(2026, 6, 1, 12, 0, tzinfo=timezone.utc)
+    ref = datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc)
+    assert p_first_after(anchor, ref, timedelta(hours=1)) == anchor
+
+
+def test_p_first_after_is_strict_on_exact_grid_point():
+    from backend.apps.workflows.scheduler import p_first_after
+    anchor = datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc)
+    ref = datetime(2026, 6, 1, 2, 0, tzinfo=timezone.utc)  # exactly on the grid
+    # strictly-after means we advance to the next point, not return ref.
+    assert p_first_after(anchor, ref, timedelta(hours=1)) == datetime(2026, 6, 1, 3, 0, tzinfo=timezone.utc)
+
+
+# --- fires_in_window (backs the cost estimate) -------------------------------
+
+def test_fires_in_window_zero_when_disabled():
+    from backend.apps.workflows import scheduler
+    from backend.apps.workflows.models import Workflow, WorkflowStep
+    wf = Workflow(title="t", steps=[WorkflowStep(text="hi")], schedule=_sched(enabled=False))
+    assert scheduler.fires_in_window(wf, days=30) == 0
+
+
+def test_fires_in_window_capped_by_remaining_max_runs():
+    """With 2 runs left on a daily schedule, a 30-day window projects exactly 2
+    fires, not 30."""
+    from backend.apps.workflows import scheduler
+    from backend.apps.workflows.models import Workflow, WorkflowStep
+    wf = Workflow(title="t", steps=[WorkflowStep(text="hi")],
+                  schedule=_sched(max_runs=10, runs_count=8))
+    assert scheduler.fires_in_window(wf, days=30) == 2
+
+
+# --- occurrences_between cap -------------------------------------------------
+
+def test_occurrences_between_respects_cap():
+    from backend.apps.workflows import scheduler
+    from backend.apps.workflows.models import Workflow, WorkflowStep
+    wf = Workflow(title="t", steps=[WorkflowStep(text="hi")], schedule=_sched(repeat_unit="day"))
+    wf.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    fires = scheduler.occurrences_between(
+        wf,
+        datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc),
+        datetime(2026, 6, 30, 0, 0, tzinfo=timezone.utc),
+        cap=5,
+    )
+    assert len(fires) == 5
+
+
+# --- ScheduleConfig validators -----------------------------------------------
+
+def test_clean_on_days_drops_out_of_range_and_dedupes_preserving_order():
+    from backend.apps.workflows.models import ScheduleConfig
+    sched = ScheduleConfig(repeat_unit="week", on_days=[3, 7, -1, 3, 0, 6])
+    assert sched.on_days == [3, 0, 6]
+
+
+def test_interval_bounds_clamp_minute_unit_to_15_1440():
+    from backend.apps.workflows.models import ScheduleConfig
+    assert ScheduleConfig(repeat_unit="minute", repeat_every=5).repeat_every == 15
+    assert ScheduleConfig(repeat_unit="minute", repeat_every=99999).repeat_every == 1440
+
+
+def test_interval_bounds_clamp_other_units_to_365():
+    from backend.apps.workflows.models import ScheduleConfig
+    assert ScheduleConfig(repeat_unit="day", repeat_every=99999).repeat_every == 365

backend/tests/test_workflows_api.py

@@ -0,0 +1,115 @@
+"""Route-handler behavior for the trash lifecycle and the global pause switch,
+called as coroutines (the established pattern in test_workflows_semantics.py).
+CRUD create/patch + If-Match + the calendar endpoint already live in the
+semantics suite; this covers the soft-delete -> restore -> purge flow and
+pause-all / resume-all, including their 404 guards.
+
+Run:
+    cd backend && .venv/bin/python -m pytest tests/test_workflows_api.py -v
+"""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _wf_env(isolated_workflows_data, reset_scheduler_state, monkeypatch):
+    # Silence the ws fan-out the handlers fire; nothing is listening in-test.
+    from backend.apps.agents.core.ws_manager import ws_manager
+
+    async def p_noop(*args, **kwargs):
+        return None
+
+    monkeypatch.setattr(ws_manager, "broadcast_global", p_noop)
+    yield
+
+
+def _run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+def test_soft_delete_hides_and_disables_schedule(make_wf):
+    from backend.apps.workflows import storage
+    from backend.apps.workflows.workflows import delete_workflow
+    wf = make_wf()
+    wf.next_run_at = datetime.now(timezone.utc) + timedelta(hours=1)
+    storage.save_workflow(wf)
+
+    res = _run(delete_workflow(wf.id))
+    assert res == {"ok": True}
+    after = storage.get_workflow(wf.id)
+    assert after.deleted_at is not None
+    assert after.schedule.enabled is False
+    assert after.next_run_at is None
+    assert wf.id not in {w.id for w in storage.list_workflows()}
+    assert wf.id in {w.id for w in storage.list_deleted_workflows()}
+
+
+def test_soft_delete_drops_pending_missed(make_wf):
+    from backend.apps.workflows import storage
+    from backend.apps.workflows.workflows import delete_workflow
+    from backend.apps.workflows.models import MissedRun
+    wf = make_wf()
+    storage.save_workflow(wf)
+    storage.add_missed(MissedRun(workflow_id=wf.id, scheduled_for=datetime.now(timezone.utc)))
+    _run(delete_workflow(wf.id))
+    assert storage.list_missed() == []
+
+
+def test_restore_brings_back_with_schedule_still_off(make_wf):
+    from backend.apps.workflows import storage
+    from backend.apps.workflows.workflows import delete_workflow, restore_workflow
+    wf = make_wf()
+    storage.save_workflow(wf)
+    _run(delete_workflow(wf.id))
+    enriched = _run(restore_workflow(wf.id))
+    assert enriched["id"] == wf.id
+    after = storage.get_workflow(wf.id)
+    assert after.deleted_at is None
+    assert after.schedule.enabled is False  # restore is deliberate; user re-arms
+    assert wf.id in {w.id for w in storage.list_workflows()}
+
+
+def test_purge_only_from_trash_and_removes_record(make_wf):
+    from backend.apps.workflows import storage
+    from backend.apps.workflows.workflows import delete_workflow, purge_workflow
+    wf = make_wf()
+    storage.save_workflow(wf)
+    _run(delete_workflow(wf.id))
+    res = _run(purge_workflow(wf.id))
+    assert res == {"ok": True}
+    assert storage.get_workflow(wf.id) is None
+
+
+def test_delete_restore_purge_guards_404(make_wf):
+    from fastapi import HTTPException
+    from backend.apps.workflows import storage
+    from backend.apps.workflows.workflows import delete_workflow, restore_workflow, purge_workflow
+    wf = make_wf()
+    storage.save_workflow(wf)
+
+    # restore / purge refuse a workflow that isn't in trash.
+    for fn in (restore_workflow, purge_workflow):
+        with pytest.raises(HTTPException) as exc:
+            _run(fn(wf.id))
+        assert exc.value.status_code == 404
+
+    # double-delete is a 404 (already gone).
+    _run(delete_workflow(wf.id))
+    with pytest.raises(HTTPException) as exc:
+        _run(delete_workflow(wf.id))
+    assert exc.value.status_code == 404
+
+
+def test_pause_all_and_resume_all_flip_global_flag(make_wf):
+    from backend.apps.workflows import storage
+    from backend.apps.workflows.workflows import pause_all_schedules, resume_all_schedules
+    assert storage.get_paused() is False
+    assert _run(pause_all_schedules()) == {"paused": True}
+    assert storage.get_paused() is True
+    assert _run(resume_all_schedules()) == {"paused": False}
+    assert storage.get_paused() is False

backend/tests/test_workflows_storage.py

@@ -0,0 +1,120 @@
+"""Storage durability: the crash-safe write path and the bounded caches.
+
+These guard properties the rest of the suite assumes but never exercises: a
+power-off mid-write must not corrupt or orphan files, a record that does end up
+corrupt must be skipped rather than crash the loader, and the per-workflow run
+log + missed-run list must stay bounded.
+
+Run:
+    cd backend && .venv/bin/python -m pytest tests/test_workflows_storage.py -v
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _wf_env(isolated_workflows_data):
+    yield
+
+
+# --- atomic write ------------------------------------------------------------
+
+def test_atomic_write_round_trips(tmp_path):
+    from backend.apps.workflows import storage
+    storage._ensure_dirs()
+    path = os.path.join(storage.DATA_DIR, "thing.json")
+    storage.p_atomic_write_json(path, {"a": 1, "b": [2, 3]})
+    with open(path) as f:
+        assert json.load(f) == {"a": 1, "b": [2, 3]}
+
+
+def test_atomic_write_leaves_no_temp_and_preserves_old_on_failure(monkeypatch):
+    """A failure mid-write must keep the previous complete file intact and drop
+    no .tmp sibling behind (the loader would never read a .tmp, but a litter of
+    them is its own bug)."""
+    from backend.apps.workflows import storage
+    storage._ensure_dirs()
+    path = os.path.join(storage.DATA_DIR, "keep.json")
+    storage.p_atomic_write_json(path, {"v": "original"})
+
+    def p_boom(*args, **kwargs):
+        raise RuntimeError("write died")
+
+    monkeypatch.setattr(storage.json, "dump", p_boom)
+    with pytest.raises(RuntimeError):
+        storage.p_atomic_write_json(path, {"v": "new"})
+
+    with open(path) as f:
+        assert json.load(f) == {"v": "original"}
+    leftovers = [n for n in os.listdir(storage.DATA_DIR) if n.endswith(".tmp")]
+    assert leftovers == []
+
+
+# --- corrupt-record resilience -----------------------------------------------
+
+def test_corrupt_workflow_record_is_skipped_not_fatal(make_wf):
+    """A truncated <id>.json must not take down the whole load; the bad record
+    silently drops out and the good ones still come back."""
+    from backend.apps.workflows import storage
+    good = make_wf(title="good")
+    storage.save_workflow(good)
+    storage._ensure_dirs()
+    with open(os.path.join(storage.DATA_DIR, "broken.json"), "w") as f:
+        f.write('{"id": "broken", "title": "trunc')  # deliberately unterminated
+
+    storage._cache_loaded = False
+    ids = {w.id for w in storage.list_workflows()}
+    assert good.id in ids
+    assert "broken" not in ids
+
+
+def test_corrupt_runs_file_yields_empty_history(make_wf):
+    from backend.apps.workflows import storage
+    wf = make_wf()
+    storage.save_workflow(wf)
+    storage._ensure_dirs()
+    with open(os.path.join(storage.RUNS_DIR, f"{wf.id}.json"), "w") as f:
+        f.write("not json at all")
+
+    storage._cache_loaded = False
+    assert storage.list_runs(wf.id) == []
+
+
+# --- bounded caches ----------------------------------------------------------
+
+def test_missed_cache_capped_to_newest(make_wf):
+    """add_missed past MAX_MISSED keeps the newest by scheduled_for so a card
+    the user never acts on can't grow the file without bound."""
+    from backend.apps.workflows import storage
+    from backend.apps.workflows.models import MissedRun
+    wf = make_wf()
+    storage.save_workflow(wf)
+    base = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    total = storage.MAX_MISSED + 25
+    for i in range(total):
+        storage.add_missed(MissedRun(workflow_id=wf.id, scheduled_for=base + timedelta(minutes=i)))
+    kept = storage.list_missed()
+    assert len(kept) == storage.MAX_MISSED
+    # The oldest 25 fell off; the kept set starts at minute 25.
+    earliest = min(m.scheduled_for for m in kept)
+    assert earliest == base + timedelta(minutes=25)
+
+
+def test_run_history_bounded_per_workflow(make_wf):
+    from backend.apps.workflows import storage
+    from backend.apps.workflows.models import WorkflowRun
+    wf = make_wf()
+    storage.save_workflow(wf)
+    over = storage.RUNS_PER_WORKFLOW + 10
+    for i in range(over):
+        storage.record_run(WorkflowRun(
+            workflow_id=wf.id, status="success",
+            started_at=datetime(2026, 1, 1, tzinfo=timezone.utc) + timedelta(minutes=i),
+        ))
+    assert len(storage._runs_cache[wf.id]) == storage.RUNS_PER_WORKFLOW