diff --git a/backend/apps/workflows/models.py b/backend/apps/workflows/models.py index c4748189..93b01d0f 100644 --- a/backend/apps/workflows/models.py +++ b/backend/apps/workflows/models.py @@ -31,7 +31,6 @@ class ScheduleConfig(BaseModel): # "local" to the host zone in memory; we leave it on disk until the # user's next save so backup/sync tools don't see spurious churn. timezone: str = "local" - on_missed: Literal["skip", "run_once", "run_all"] = "skip" # Optional end conditions. None = forever / unbounded. Schedule auto- # disables once either is satisfied; scheduler._tick zeroes out # next_run_at and flips enabled=False so the UI reflects reality. @@ -185,6 +184,17 @@ class WorkflowRun(BaseModel): paused: bool = False +class MissedRun(BaseModel): + # A single scheduled fire that elapsed while OpenSwarm was closed. Captured + # at startup and surfaced in the launch-time review card; leaves this store + # only when the user runs it (becomes a ran_late run) or dismisses it + # (becomes a skipped run). scheduled_for is the instant it should have fired. + id: str = Field(default_factory=lambda: uuid4().hex) + workflow_id: str + scheduled_for: datetime + created_at: datetime = Field(default_factory=datetime.now) + + class WorkflowCreate(BaseModel): title: str = "Untitled workflow" auto_named: bool = True @@ -227,6 +237,10 @@ class WorkflowUpdate(BaseModel): step_tool_usage: Optional[dict[str, dict[str, bool]]] = None +class MissedRunAction(BaseModel): + ids: list[str] = Field(default_factory=list) + + class DraftCommitBody(BaseModel): # The model the user settled on in the Edit Agent picker, applied to the # workflow's run model only on Save (save-gated; Discard drops it). diff --git a/backend/apps/workflows/scheduler.py b/backend/apps/workflows/scheduler.py index 28dd949f..8a3f5113 100644 --- a/backend/apps/workflows/scheduler.py +++ b/backend/apps/workflows/scheduler.py @@ -3,8 +3,9 @@ One long-lived asyncio task wakes on the next-due workflow boundary, fires matching workflows, then re-computes. We deliberately avoid one-task-per- workflow (turns rescheduling into a thundering re-spawn problem). On -startup we walk persisted workflows once, decide what to do about missed -fires via on_missed, and queue each. +startup we walk persisted workflows once and capture every fire that +elapsed while the app was closed as a pending MissedRun, so the launch-time +review card can let the user run or dismiss each one. Schedule semantics: unit=minute: fires every repeat_every minutes (15 is the enforced floor) @@ -29,11 +30,19 @@ from datetime import datetime, timedelta, timezone from typing import Optional from zoneinfo import ZoneInfo, ZoneInfoNotFoundError -from backend.apps.workflows.models import Workflow, ScheduleConfig +from backend.apps.workflows.models import Workflow, ScheduleConfig, WorkflowRun, MissedRun from backend.apps.workflows import storage, executor logger = logging.getLogger(__name__) +# How many recent missed fires we keep reviewable per workflow. Older ones +# collapse into a single summarizing "skipped" run so a 15-minute schedule +# that was off for days doesn't flood the card or the run history. +PER_WORKFLOW_MISSED_CAP = 20 +# Bound on the per-workflow enumeration walk at startup. 480 covers ~5 days of +# a 15-minute schedule; past that the exact count stops mattering. +MISSED_ENUM_CAP = 480 + _loop_task: Optional[asyncio.Task] = None _wake = asyncio.Event() @@ -349,14 +358,66 @@ def _mark_stuck_runs_failed() -> None: ) -def reconcile_on_startup() -> None: - """Walk persisted workflows once and resolve missed fires per policy. +def record_skipped(wf: Workflow, scheduled_for: datetime, error: str) -> WorkflowRun: + """Log a missed fire as a 'skipped' run so it leaves a trace in history. - Missed-run policies: - skip -> roll forward to next future fire, ignore missed - run_once -> if any fires were missed, schedule a single catch-up at now - run_all -> not actually run_all in v1 (would burn tokens); same as run_once - but we mark the run.status as ran_late so the UI surfaces it + Used both for over-cap fires at startup and for fires the user dismisses + from the review card. Updates the workflow's last_run_* summary so the + card's status dot reflects reality. + """ + now = datetime.now() + run = WorkflowRun( + workflow_id=wf.id, + status="skipped", + scheduled_for=scheduled_for, + started_at=now, + finished_at=now, + triggered_by="schedule", + error=error, + ) + storage.record_run(run) + wf.last_run_at = now + wf.last_run_status = "skipped" + wf.last_run_id = run.id + storage.save_workflow(wf) + return run + + +def _capture_missed(wf: Workflow, missed: list[datetime]) -> None: + if not missed: + return + recent = missed[-PER_WORKFLOW_MISSED_CAP:] + older = missed[: len(missed) - len(recent)] + if older: + suffix = "+" if len(missed) >= MISSED_ENUM_CAP else "" + record_skipped( + wf, + older[0], + f"Skipped {len(older)}{suffix} earlier missed runs while OpenSwarm was closed", + ) + for sf in recent: + storage.add_missed(MissedRun(workflow_id=wf.id, scheduled_for=sf)) + + +async def run_missed_sequence(wf: Workflow, scheduled_fors: list[datetime]) -> None: + """Run a workflow once per missed fire, sequentially. + + Sequential because the executor refuses concurrent runs of the same + workflow; firing them all at once would skip all but the first. + """ + for sf in scheduled_fors: + try: + await executor.execute(wf, triggered_by="schedule", scheduled_for=sf) + except Exception: + logger.exception("missed-run fire failed for workflow=%s", wf.id) + + +def reconcile_on_startup() -> None: + """Walk persisted workflows once and capture fires missed while closed. + + No auto-firing here anymore: each missed fire becomes a pending MissedRun + the user reviews on launch. We roll next_run_at forward to a future slot so + a dev hot-reload re-running this won't re-enumerate the same misses. """ now_utc = datetime.now(timezone.utc) for wf in storage.list_workflows(): @@ -373,17 +434,12 @@ def reconcile_on_startup() -> None: _disable_schedule(wf) continue - nra = _as_utc(wf.next_run_at) - missed = bool(nra and nra <= now_utc) - if missed and wf.schedule.on_missed in ("run_once", "run_all"): - # Keep next_run_at <= now_utc so the very next tick fires it. - # Normalize to a UTC-aware value so future comparisons don't - # trip on naive legacy datetimes. - wf.next_run_at = nra - storage.save_workflow(wf) - else: - wf.next_run_at = _next_fire_after(wf.schedule, now_utc) - storage.save_workflow(wf) + anchor = _as_utc(wf.next_run_at) + if anchor is not None and anchor <= now_utc: + _capture_missed(wf, occurrences_between(wf, anchor, now_utc, cap=MISSED_ENUM_CAP)) + + wf.next_run_at = _next_fire_after(wf.schedule, now_utc) + storage.save_workflow(wf) async def start() -> None: diff --git a/backend/apps/workflows/storage.py b/backend/apps/workflows/storage.py index 96204c2c..c8d0125a 100644 --- a/backend/apps/workflows/storage.py +++ b/backend/apps/workflows/storage.py @@ -15,15 +15,22 @@ from threading import Lock from typing import Optional from backend.config.paths import DATA_ROOT -from backend.apps.workflows.models import Workflow, WorkflowRun +from backend.apps.workflows.models import Workflow, WorkflowRun, MissedRun DATA_DIR = os.path.join(DATA_ROOT, "workflows") RUNS_DIR = os.path.join(DATA_DIR, "runs") PAUSED_FILE = os.path.join(DATA_DIR, "paused.json") +MISSED_FILE = os.path.join(DATA_DIR, "missed.json") + +# Hard ceiling on pending missed fires kept on disk. The review card only +# shows 50; this just stops the file growing without bound if the user keeps +# quitting without acting on the card. +MAX_MISSED = 200 _io_lock = Lock() _workflow_cache: dict[str, Workflow] = {} _runs_cache: dict[str, list[WorkflowRun]] = {} +_missed_cache: list[MissedRun] = [] _cache_loaded = False _paused = False @@ -97,6 +104,13 @@ def _load_all_from_disk() -> None: _paused = bool(json.load(f).get("paused", False)) except Exception: _paused = False + _missed_cache.clear() + if os.path.exists(MISSED_FILE): + try: + with open(MISSED_FILE) as f: + _missed_cache.extend(MissedRun(**m) for m in json.load(f)) + except Exception: + _missed_cache.clear() _cache_loaded = True @@ -137,6 +151,9 @@ def delete_workflow(wid: str) -> bool: rf = _runs_path(wid) if os.path.exists(rf): os.remove(rf) + if any(m.workflow_id == wid for m in _missed_cache): + _missed_cache[:] = [m for m in _missed_cache if m.workflow_id != wid] + _write_missed() return existed @@ -192,6 +209,51 @@ def set_paused(value: bool) -> bool: return _paused +def _write_missed() -> None: + with open(MISSED_FILE, "w") as f: + json.dump([m.model_dump(mode="json") for m in _missed_cache], f, indent=2) + + +def list_missed() -> list[MissedRun]: + if not _cache_loaded: + init() + return list(_missed_cache) + + +def add_missed(run: MissedRun) -> MissedRun: + if not _cache_loaded: + init() + with _io_lock: + _ensure_dirs() + _missed_cache.append(run) + # Keep the newest MAX_MISSED by scheduled_for so a never-acked card + # can't grow the file forever across repeated launches. + if len(_missed_cache) > MAX_MISSED: + _missed_cache.sort(key=lambda m: m.scheduled_for) + del _missed_cache[: len(_missed_cache) - MAX_MISSED] + _write_missed() + return run + + +def remove_missed(ids: list[str]) -> None: + if not _cache_loaded: + init() + drop = set(ids) + with _io_lock: + _ensure_dirs() + _missed_cache[:] = [m for m in _missed_cache if m.id not in drop] + _write_missed() + + +def clear_missed() -> None: + if not _cache_loaded: + init() + with _io_lock: + _ensure_dirs() + _missed_cache.clear() + _write_missed() + + def update_run(run_id: str, **fields) -> Optional[WorkflowRun]: if not _cache_loaded: init() diff --git a/backend/apps/workflows/workflows.py b/backend/apps/workflows/workflows.py index 3c2f35d5..87f13df9 100644 --- a/backend/apps/workflows/workflows.py +++ b/backend/apps/workflows/workflows.py @@ -14,6 +14,7 @@ from backend.apps.workflows.models import ( WorkflowRun, WorkflowStep, DraftCommitBody, + MissedRunAction, ) from backend.apps.workflows import storage, scheduler, executor, audit, escalation @@ -584,6 +585,66 @@ async def list_all_runs(limit: int = 200): return {"runs": [r.model_dump(mode="json") for r in runs]} +@workflows.router.get("/missed") +async def list_missed_runs(limit: int = 50): + """Pending fires that elapsed while the app was closed, newest-first. + Backs the launch-time review card.""" + missed = sorted(storage.list_missed(), key=lambda m: m.scheduled_for, reverse=True) + out: list[dict] = [] + for m in missed[:limit]: + wf = storage.get_workflow(m.workflow_id) + if not wf: + continue + out.append({ + "id": m.id, + "workflow_id": m.workflow_id, + "workflow_title": wf.title, + "workflow_icon": wf.icon, + "scheduled_for": m.scheduled_for.isoformat() if isinstance(m.scheduled_for, datetime) else m.scheduled_for, + }) + return {"missed": out} + + +@workflows.router.post("/missed/run") +async def run_missed_runs(body: MissedRunAction): + """Run the selected missed fires now. Each lands in History as ran_late. + Fires of the same workflow run sequentially (the executor blocks + concurrent runs of one workflow).""" + wanted = set(body.ids) + selected = [m for m in storage.list_missed() if m.id in wanted] + if not selected: + return {"started": 0} + storage.remove_missed([m.id for m in selected]) + by_wf: dict[str, list[datetime]] = {} + for m in sorted(selected, key=lambda m: m.scheduled_for): + by_wf.setdefault(m.workflow_id, []).append(m.scheduled_for) + started = 0 + for wid, fors in by_wf.items(): + wf = storage.get_workflow(wid) + if not wf: + continue + started += len(fors) + asyncio.create_task(scheduler.run_missed_sequence(wf, fors)) + return {"started": started} + + +@workflows.router.post("/missed/dismiss") +async def dismiss_missed_runs(body: MissedRunAction): + """Drop the selected missed fires, logging each as a skipped run so the + workflow's history still shows it happened.""" + wanted = set(body.ids) + selected = [m for m in storage.list_missed() if m.id in wanted] + storage.remove_missed([m.id for m in selected]) + dismissed = 0 + for m in selected: + wf = storage.get_workflow(m.workflow_id) + if not wf: + continue + scheduler.record_skipped(wf, m.scheduled_for, "You dismissed this missed run") + dismissed += 1 + return {"dismissed": dismissed} + + @workflows.router.get("/calendar") async def list_calendar_events( from_: str = Query(..., alias="from"), diff --git a/backend/tests/test_schedule_e2e.py b/backend/tests/test_schedule_e2e.py index 3723041a..6e50afc0 100644 --- a/backend/tests/test_schedule_e2e.py +++ b/backend/tests/test_schedule_e2e.py @@ -31,8 +31,10 @@ def isolated_data_dir(monkeypatch, tmp_path): 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, "_cache_loaded", False) monkeypatch.setattr(_storage, "_paused", False) monkeypatch.setattr(_audit, "AUDIT_DIR", str(tmp_path / "workflows" / "audit")) @@ -154,33 +156,101 @@ async def test_paused_state_blocks_all_fires(monkeypatch): storage.set_paused(False) -async def test_reconcile_skip_rolls_past_missed(monkeypatch): - """on_missed='skip' + a missed next_run_at => startup rolls forward - to the next future fire without queuing a catch-up.""" +async def test_reconcile_captures_missed_fires(monkeypatch): + """A daily workflow whose next_run_at elapsed while the app was closed => + startup captures the missed fires as pending MissedRuns and rolls + next_run_at forward (no auto-firing).""" from backend.apps.workflows import storage, scheduler - wf = _make_wf() - wf.schedule.on_missed = "skip" - # Stash a missed fire 6 hours ago. - wf.next_run_at = datetime.now(timezone.utc) - timedelta(hours=6) + # created_at must predate the missed window; occurrences_between never + # enumerates fires from before the workflow existed. + wf = _make_wf(created_at=datetime.now(timezone.utc) - timedelta(days=10)) + wf.next_run_at = datetime.now(timezone.utc) - timedelta(days=3) storage.save_workflow(wf) scheduler.reconcile_on_startup() + missed = [m for m in storage.list_missed() if m.workflow_id == wf.id] + assert len(missed) >= 1 + assert all(m.scheduled_for < datetime.now(timezone.utc) for m in missed) after = storage.get_workflow(wf.id) assert after.next_run_at is not None assert after.next_run_at > datetime.now(timezone.utc) -async def test_reconcile_run_once_keeps_missed(monkeypatch): - """on_missed='run_once' => startup leaves next_run_at in the past so - the first tick fires a catch-up.""" +async def test_reconcile_no_missed_when_future(monkeypatch): + """next_run_at in the future => nothing missed, nothing captured.""" from backend.apps.workflows import storage, scheduler - wf = _make_wf() - wf.schedule.on_missed = "run_once" - missed = datetime.now(timezone.utc) - timedelta(hours=6) - wf.next_run_at = missed + wf = _make_wf(created_at=datetime.now(timezone.utc) - timedelta(days=10)) + wf.next_run_at = datetime.now(timezone.utc) + timedelta(hours=6) storage.save_workflow(wf) scheduler.reconcile_on_startup() - after = storage.get_workflow(wf.id) - assert after.next_run_at <= datetime.now(timezone.utc) + assert [m for m in storage.list_missed() if m.workflow_id == wf.id] == [] + + +async def test_reconcile_over_cap_collapses_to_skipped(monkeypatch): + """A 15-minute schedule off for days => only the cap is kept reviewable; + the rest collapse into a single skipped run in history.""" + from backend.apps.workflows import storage, scheduler + from backend.apps.workflows.models import ScheduleConfig + wf = _make_wf( + created_at=datetime.now(timezone.utc) - timedelta(days=30), + schedule=ScheduleConfig( + enabled=True, repeat_unit="minute", repeat_every=15, + timezone="America/Los_Angeles", + ), + ) + wf.next_run_at = datetime.now(timezone.utc) - timedelta(days=2) + storage.save_workflow(wf) + scheduler.reconcile_on_startup() + missed = [m for m in storage.list_missed() if m.workflow_id == wf.id] + assert len(missed) == scheduler.PER_WORKFLOW_MISSED_CAP + skipped = [r for r in storage.list_runs(wf.id, limit=50) if r.status == "skipped"] + assert len(skipped) == 1 + + +async def test_dismiss_missed_records_skipped(monkeypatch): + """Dismissing a missed run drops it from pending and leaves a skipped + run in history.""" + from backend.apps.workflows import storage + from backend.apps.workflows.models import MissedRun, MissedRunAction + from backend.apps.workflows.workflows import dismiss_missed_runs + wf = _make_wf() + storage.save_workflow(wf) + m = MissedRun(workflow_id=wf.id, scheduled_for=datetime.now(timezone.utc) - timedelta(hours=2)) + storage.add_missed(m) + res = await dismiss_missed_runs(MissedRunAction(ids=[m.id])) + assert res["dismissed"] == 1 + assert storage.list_missed() == [] + skipped = [r for r in storage.list_runs(wf.id, limit=10) if r.status == "skipped"] + assert len(skipped) == 1 + + +async def test_run_missed_runs_clears_pending_and_fires(monkeypatch): + """Running selected missed fires removes them from pending and invokes + the executor once per fire, sequentially.""" + from backend.apps.workflows import storage, executor, scheduler + from backend.apps.workflows.models import MissedRun, MissedRunAction + from backend.apps.workflows.workflows import run_missed_runs + + calls: list = [] + + async def fake_execute(wf, triggered_by="schedule", scheduled_for=None): + calls.append(scheduled_for) + from backend.apps.workflows.models import WorkflowRun + return WorkflowRun(workflow_id=wf.id, status="ran_late", scheduled_for=scheduled_for) + + monkeypatch.setattr(executor, "execute", fake_execute) + + wf = _make_wf() + storage.save_workflow(wf) + m1 = MissedRun(workflow_id=wf.id, scheduled_for=datetime.now(timezone.utc) - timedelta(hours=3)) + m2 = MissedRun(workflow_id=wf.id, scheduled_for=datetime.now(timezone.utc) - timedelta(hours=2)) + storage.add_missed(m1) + storage.add_missed(m2) + res = await run_missed_runs(MissedRunAction(ids=[m1.id, m2.id])) + assert res["started"] == 2 + assert storage.list_missed() == [] + # The endpoint spawns the sequence as a background task; give it a beat. + await asyncio.sleep(0.05) + assert len(calls) == 2 async def test_create_workflow_schedules_next_fire(): diff --git a/frontend/src/app/pages/Workflows/ScheduleFacet.tsx b/frontend/src/app/pages/Workflows/ScheduleFacet.tsx index 1068b058..0e2897e7 100644 --- a/frontend/src/app/pages/Workflows/ScheduleFacet.tsx +++ b/frontend/src/app/pages/Workflows/ScheduleFacet.tsx @@ -365,7 +365,7 @@ export default function ScheduleFacet({ draft, setDraft }: { draft: Workflow; se const ends = new Date(s.ends_at).getTime(); if (!Number.isNaN(ends) && ends <= Date.now()) { return ( - + This date is in the past. The schedule will turn itself off. ); @@ -373,24 +373,13 @@ export default function ScheduleFacet({ draft, setDraft }: { draft: Workflow; se } if (endKind === 'after_n' && s.max_runs != null && s.runs_count >= s.max_runs) { return ( - + This workflow has already run {s.runs_count}× (limit {s.max_runs}). Raise the number or reset the counter to re-arm. ); } return null; })()} - - If missed - - {/* Section: What can the agent do? */} @@ -445,11 +434,11 @@ function AppOpenStatusBadge({ info, hour, minute, frequent, onFix }: { info: App return ( - + {good ? 'Will run even if you close OpenSwarm.' : (frequent ? 'OpenSwarm must be open for this to run.' : `OpenSwarm must be open at ${fmt} for this to run.`)} @@ -514,7 +503,7 @@ function PermissionRow({ idx, tier, cloudSmsEnabled, onChange, onRemove }: { )} {!cloudSmsEnabled && ( - + Coming soon. Until cloud SMS ships, this tier falls back to an in-app notify with a "fallback" badge. )} diff --git a/frontend/src/app/pages/Workflows/scheduleUtils.ts b/frontend/src/app/pages/Workflows/scheduleUtils.ts index 53b89d56..921dc3eb 100644 --- a/frontend/src/app/pages/Workflows/scheduleUtils.ts +++ b/frontend/src/app/pages/Workflows/scheduleUtils.ts @@ -18,7 +18,6 @@ export function defaultSchedule(): ScheduleConfig { hour: 9, minute: 0, timezone: tz, - on_missed: 'skip', ends_at: null, max_runs: null, runs_count: 0, diff --git a/frontend/src/shared/state/workflowsSlice.ts b/frontend/src/shared/state/workflowsSlice.ts index 2777b08b..1f3126b7 100644 --- a/frontend/src/shared/state/workflowsSlice.ts +++ b/frontend/src/shared/state/workflowsSlice.ts @@ -19,7 +19,6 @@ export interface ScheduleConfig { hour: number; minute: number; timezone: string; - on_missed: 'skip' | 'run_once' | 'run_all'; /** End conditions; null on both = forever. Scheduler auto-disables on threshold. */ ends_at: string | null; max_runs: number | null;