diff --git a/backend/apps/workflows/executor.py b/backend/apps/workflows/executor.py index f5f51011..2be409c3 100644 --- a/backend/apps/workflows/executor.py +++ b/backend/apps/workflows/executor.py @@ -25,6 +25,15 @@ _running: dict[str, str] = {} _running_lock = asyncio.Lock() +def _ran_late(started_at: datetime, scheduled_for: datetime) -> bool: + """Late means the run STARTED well after its slot (app was closed, event + loop backed up), not that it ran long. Measured from started_at so a + punctual run that simply takes a while isn't mislabeled. Both sides + normalized to UTC; a naive started_at is host-local.""" + delta = started_at.astimezone(timezone.utc) - scheduled_for.astimezone(timezone.utc) + return delta.total_seconds() > 300 + + # run_id -> "stop". Set by the stop endpoint so the executor loop, not the # HTTP handler, owns the run's terminal write. Without this the still-running # executor task could overwrite a "Stopped by user" failure with success. @@ -372,12 +381,7 @@ async def execute( run.status = "failure" run.error = step_error wf.last_run_status = "failure" - elif scheduled_for is not None and (run.finished_at.replace(tzinfo=None) - scheduled_for.replace(tzinfo=None)).total_seconds() > 300: - # Started more than 5 minutes after its slot (app was closed, - # event loop backed up, etc.). Surface in History as ran_late - # so the user can tell apart "fired on time" from "caught up". - # Strip tz before the subtraction so a UTC-aware scheduled_for - # (new code path) and a naive finished_at don't raise. + elif scheduled_for is not None and _ran_late(run.started_at, scheduled_for): run.status = "ran_late" wf.last_run_status = "ran_late" else: diff --git a/backend/apps/workflows/scheduler.py b/backend/apps/workflows/scheduler.py index 650b40f1..7aafe8ed 100644 --- a/backend/apps/workflows/scheduler.py +++ b/backend/apps/workflows/scheduler.py @@ -113,26 +113,38 @@ def is_schedule_configured(sched: ScheduleConfig) -> bool: return True -def _next_fire_after(sched: ScheduleConfig, ref_utc: datetime) -> Optional[datetime]: +def _first_after(anchor: datetime, ref: datetime, step: timedelta) -> datetime: + """First instant on the grid {anchor + k*step} strictly after ref.""" + if anchor > ref: + return anchor + n = (ref - anchor) // step + return anchor + (n + 1) * step + + +def _next_fire_after( + sched: ScheduleConfig, + ref_utc: datetime, + anchor_utc: Optional[datetime] = None, +) -> Optional[datetime]: if not sched.enabled or not is_schedule_configured(sched): return None tz = _resolve_tz(sched.timezone) ref_local = ref_utc.astimezone(tz) base = ref_local.replace(second=0, microsecond=0) + # Anchor recurring phases to a fixed origin (the workflow's creation), so a + # recompute (tick, kick, startup reconcile) lands on the same grid instead + # of re-phasing to "now" and sliding the cadence. Falls back to ref. + anchor_local = (anchor_utc or ref_utc).astimezone(tz) if sched.repeat_unit == "minute": step = max(15, sched.repeat_every) - c = base - while c <= ref_local: - c = c + timedelta(minutes=step) - return c.astimezone(timezone.utc) + grid = anchor_local.replace(second=0, microsecond=0) + return _first_after(grid, ref_local, timedelta(minutes=step)).astimezone(timezone.utc) if sched.repeat_unit == "hour": step = max(1, sched.repeat_every) - c = base.replace(minute=sched.minute) - while c <= ref_local: - c = c + timedelta(hours=step) - return c.astimezone(timezone.utc) + grid = anchor_local.replace(minute=sched.minute, second=0, microsecond=0) + return _first_after(grid, ref_local, timedelta(hours=step)).astimezone(timezone.utc) candidate = base.replace(hour=sched.hour, minute=sched.minute) @@ -164,7 +176,7 @@ def _next_fire_after(sched: ScheduleConfig, ref_utc: datetime) -> Optional[datet if sched.repeat_unit == "week": allowed = sched.on_days step = max(1, sched.repeat_every) - anchor_week = _week_start(ref_local) + anchor_week = _week_start(anchor_local) for _ in range(0, 7 * step + 7): week_delta = (_week_start(candidate).date() - anchor_week.date()).days // 7 if ( @@ -181,7 +193,7 @@ def _next_fire_after(sched: ScheduleConfig, ref_utc: datetime) -> Optional[datet def compute_next_fire(wf: Workflow, ref: Optional[datetime] = None) -> Optional[datetime]: ref_utc = _as_utc(ref) if ref is not None else datetime.now(timezone.utc) - return _next_fire_after(wf.schedule, ref_utc) + return _next_fire_after(wf.schedule, ref_utc, _as_utc(getattr(wf, "created_at", None))) def fires_in_window(wf: Workflow, days: int = 30) -> int: @@ -204,9 +216,10 @@ def fires_in_window(wf: Workflow, days: int = 30) -> int: remaining_budget = ( sched.max_runs - sched.runs_count if sched.max_runs is not None else 5000 ) + anchor_utc = _as_utc(getattr(wf, "created_at", None)) count = 0 while count < min(5000, remaining_budget): - nxt = _next_fire_after(sched, cursor_utc) + nxt = _next_fire_after(sched, cursor_utc, anchor_utc) if nxt is None or nxt > end_utc: break count += 1 @@ -252,7 +265,7 @@ def occurrences_between( limit = max(0, min(cap, remaining)) out: list[datetime] = [] while len(out) < limit: - nxt = _next_fire_after(sched, cursor_utc) + nxt = _next_fire_after(sched, cursor_utc, created_at) if nxt is None or nxt >= end_utc: break if nxt >= start_utc: @@ -301,7 +314,7 @@ async def _tick() -> None: for wf in due: scheduled_for = _as_utc(wf.next_run_at) - nxt = _next_fire_after(wf.schedule, now_utc) + nxt = _next_fire_after(wf.schedule, now_utc, _as_utc(getattr(wf, "created_at", None))) wf.next_run_at = nxt storage.save_workflow(wf) asyncio.create_task(_fire(wf, scheduled_for=scheduled_for)) @@ -452,7 +465,7 @@ def reconcile_on_startup() -> None: 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) + wf.next_run_at = _next_fire_after(wf.schedule, now_utc, _as_utc(getattr(wf, "created_at", None))) storage.save_workflow(wf) diff --git a/backend/tests/test_workflows_semantics.py b/backend/tests/test_workflows_semantics.py index 04d4ab49..9bae4e13 100644 --- a/backend/tests/test_workflows_semantics.py +++ b/backend/tests/test_workflows_semantics.py @@ -287,6 +287,49 @@ def test_weekly_repeat_every_skips_inactive_weeks(): assert nxt == datetime(2026, 7, 6, 9, 0, tzinfo=timezone.utc) +def test_weekly_every_n_weeks_phase_is_stable_across_recompute(): + """The bi-weekly phase must anchor to created_at, not to whenever the + recompute happens. Computing from an 'on' week and from the following + 'off' week must both land on the same created_at-anchored grid; otherwise + a tick/kick in an off week slides the whole cadence by weeks.""" + from backend.apps.workflows.scheduler import compute_next_fire + from backend.apps.workflows.models import ScheduleConfig + # Created on Mon 2026-06-08. Every 2 weeks on Monday => fires 06-08, + # 06-22, 07-06, 07-20 (UTC). 06-15 and 06-29 are 'off' weeks. + wf = _make_wf( + created_at=datetime(2026, 6, 8, tzinfo=timezone.utc), + schedule=ScheduleConfig( + enabled=True, repeat_unit="week", repeat_every=2, on_days=[1], + hour=9, minute=0, timezone="UTC", + ), + ) + # From just after the 06-22 fire (on-week) -> 07-06. + assert compute_next_fire(wf, datetime(2026, 6, 23, tzinfo=timezone.utc)) == \ + datetime(2026, 7, 6, 9, 0, tzinfo=timezone.utc) + # From an off-week (06-30) the next fire is STILL 07-06, not 07-13. + assert compute_next_fire(wf, datetime(2026, 6, 30, tzinfo=timezone.utc)) == \ + datetime(2026, 7, 6, 9, 0, tzinfo=timezone.utc) + # From the off-week between the first two fires (06-16) -> 06-22, not 06-29. + assert compute_next_fire(wf, datetime(2026, 6, 16, tzinfo=timezone.utc)) == \ + datetime(2026, 6, 22, 9, 0, tzinfo=timezone.utc) + + +def test_ran_late_is_measured_from_start_not_finish(): + """A run that STARTS on time is 'success' no matter how long it runs; a run + that starts >5min after its slot is 'ran_late'.""" + from backend.apps.workflows.executor import _ran_late + slot = datetime(2026, 6, 22, 9, 0, tzinfo=timezone.utc) + # Started on time -> not late (even though such a run might finish much later). + assert _ran_late(slot, slot) is False + assert _ran_late(slot + timedelta(minutes=4), slot) is False + # Started well after the slot -> late. + assert _ran_late(slot + timedelta(minutes=6), slot) is True + # Naive started_at (host-local, as datetime.now() produces) is normalized + # to UTC rather than subtracted across the offset. + naive_on_time = slot.astimezone().replace(tzinfo=None) + assert _ran_late(naive_on_time, slot) is False + + def test_frozen_empty_tool_set_does_not_fall_back_to_defaults(): from backend.apps.workflows.executor import _resolve_allowed_tools from backend.apps.workflows.models import ActionsConfig