[aidan] fix/schedule: measure ran_late from start and anchor recurrences to created_at

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:

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)
 
 

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