refactor(campaign): move tick decisions into the pure scheduler seam

Extend the pure scheduler with elapsed_seconds (clock injected), decide_tick
(offset + due + lifecycle action) and resolve_finalize (cancel-race guard),
so the durable loop stops hand-coding elapsed/finished/status checks and only
does I/O. Deletes the runner's private _elapsed_seconds and converges
current_window_offset onto the one pure elapsed computation. The clock-skew
tolerance and cancel-race guard are now unit-testable at the seam.

backend/agenteval/evaluation/campaign_runner.py

@@ -16,12 +16,18 @@ torn down and rebuilt on restart without losing or duplicating work.
 import asyncio
 import logging
 from dataclasses import dataclass, field
-from datetime import timezone
 from typing import Optional
 
 from sqlmodel import Session
 
-from agenteval.evaluation.campaign_scheduler import clock_offset, decide_schedule
+from agenteval.evaluation.campaign_scheduler import (
+    TickAction,
+    clock_offset,
+    decide_schedule,
+    decide_tick,
+    elapsed_seconds,
+    resolve_finalize,
+)
 from agenteval.evaluation.engine import EvalEngine
 from agenteval.models import Campaign, CampaignStatus, EvalRun, RunStatus, RunTrigger
 from agenteval.storage.db import get_session, utc_now
@@ -55,13 +61,6 @@ class AdvanceResult:
     finished: bool = False
 
 
-def _elapsed_seconds(started_at) -> float:
-    """Real seconds since ``started_at`` (tolerating naive UTC from SQLite)."""
-    if started_at.tzinfo is None:
-        started_at = started_at.replace(tzinfo=timezone.utc)
-    return (utc_now() - started_at).total_seconds()
-
-
 def _spawned_indices(campaign: Campaign) -> set[int]:
     scheduler = (campaign.summary or {}).get("scheduler", {})
     return set(scheduler.get(_SPAWNED_KEY, []))
@@ -76,7 +75,7 @@ def current_window_offset(campaign: Campaign) -> float:
     if campaign.started_at is None:
         return 0.0
     offset = clock_offset(
-        elapsed_seconds=_elapsed_seconds(campaign.started_at),
+        elapsed_seconds=elapsed_seconds(now=utc_now(), started_at=campaign.started_at),
         time_scale=campaign.time_scale,
     )
     return min(offset, float(campaign.window_seconds))
@@ -186,8 +185,8 @@ async def advance_campaign(
 async def run_campaign_loop(campaign_id: str, *, tick_seconds: float = DEFAULT_TICK_SECONDS) -> None:
     """Drive one campaign to completion, ticking on real wall-clock time.
 
-    The loop holds no authority: every tick it reloads the campaign from the DB,
-    recomputes elapsed time from the persisted ``started_at``, and advances.
+    The loop holds no authority and no decision logic: every tick it reloads the
+    campaign, asks the pure ``decide_tick`` what to do, and only performs I/O.
     It exits when the window finishes, the campaign leaves RUNNING (e.g. it was
     cancelled), or the cooperative cancel event fires.
     """
@@ -197,20 +196,34 @@ async def run_campaign_loop(campaign_id: str, *, tick_seconds: float = DEFAULT_T
         while not cancel.is_set():
             repo = CampaignRepository(session)
             campaign = repo.get(campaign_id)
-            if not campaign or campaign.status != CampaignStatus.RUNNING or campaign.started_at is None:
+            if not campaign:
                 return
 
-            elapsed = _elapsed_seconds(campaign.started_at)
-            result = await advance_campaign(
+            now = utc_now()
+            decision = decide_tick(
+                now=now,
+                started_at=campaign.started_at,
+                status=campaign.status,
+                window_seconds=campaign.window_seconds,
+                time_scale=campaign.time_scale,
+                plan=campaign.plan,
+                spawned_indices=_spawned_indices(campaign),
+            )
+            if decision.action is TickAction.STOP:
+                return
+
+            elapsed = elapsed_seconds(now=now, started_at=campaign.started_at)
+            await advance_campaign(
                 campaign_id=campaign_id,
                 elapsed_seconds=elapsed,
                 session=session,
                 cancel_event=cancel,
             )
-            if result and result.finished:
+
+            if decision.action is TickAction.COMPLETE:
                 current = repo.get(campaign_id)
-                # Only complete if still running (not cancelled meanwhile).
-                if current and current.status == CampaignStatus.RUNNING:
+                # Cancel-race guard: only complete if still RUNNING (pure rule).
+                if current and resolve_finalize(current.status) is TickAction.COMPLETE:
                     current.status = CampaignStatus.COMPLETED
                     current.completed_at = utc_now()
                     repo.update(current)

backend/agenteval/evaluation/campaign_scheduler.py

@@ -12,9 +12,11 @@ plan-entry indices, and this function excludes them from the due list.
 """
 
 from dataclasses import dataclass
-from typing import Iterable
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Iterable, Optional
 
-from agenteval.models import CampaignPlanEntry
+from agenteval.models import CampaignPlanEntry, CampaignStatus
 
 
 def clock_offset(*, elapsed_seconds: float, time_scale: float) -> float:
@@ -27,6 +29,19 @@ def clock_offset(*, elapsed_seconds: float, time_scale: float) -> float:
     return elapsed_seconds * time_scale
 
 
+def elapsed_seconds(*, now: datetime, started_at: datetime) -> float:
+    """Real seconds between ``started_at`` and ``now``.
+
+    The clock is injected (``now``) so this stays pure and unit-testable.
+    ``started_at`` from SQLite may be naive; treat it as UTC (AGENTS.md #2).
+    """
+    if started_at.tzinfo is None:
+        started_at = started_at.replace(tzinfo=timezone.utc)
+    if now.tzinfo is None:
+        now = now.replace(tzinfo=timezone.utc)
+    return (now - started_at).total_seconds()
+
+
 @dataclass(frozen=True)
 class DueEntry:
     """A plan entry that is due to spawn, tagged with its plan index."""
@@ -63,3 +78,67 @@ def decide_schedule(
     )
     finished = clock_offset_seconds >= window_seconds
     return ScheduleDecision(due=due, finished=finished)
+
+
+class TickAction(str, Enum):
+    """What the durable loop should do after one tick's decision."""
+
+    CONTINUE = "continue"  # still inside the window — spawn due, then sleep
+    COMPLETE = "complete"  # window ended — mark the campaign completed
+    STOP = "stop"          # not running (cancelled/absent) — exit, touch nothing
+
+
+@dataclass(frozen=True)
+class TickDecision:
+    """The full per-tick decision: window position, what's due, and lifecycle."""
+
+    offset: float
+    due: tuple[DueEntry, ...]
+    finished: bool
+    action: TickAction
+
+
+def decide_tick(
+    *,
+    now: datetime,
+    started_at: Optional[datetime],
+    status: CampaignStatus,
+    window_seconds: int,
+    time_scale: float,
+    plan: list[CampaignPlanEntry],
+    spawned_indices: Iterable[int] = (),
+) -> TickDecision:
+    """Decide everything one loop tick needs, purely from the clock + DB state.
+
+    The shell reads ``now`` and the campaign row, calls this, then only does
+    I/O (spawn due Runs, persist progress, write COMPLETED). A campaign that is
+    not RUNNING or has no ``started_at`` yields ``STOP`` — the loop exits without
+    touching status. Otherwise the window offset drives ``CONTINUE`` vs
+    ``COMPLETE``. The COMPLETE write is still gated by ``resolve_finalize`` on a
+    freshly re-read status, so a cancel arriving mid-tick is never overwritten.
+    """
+    if status != CampaignStatus.RUNNING or started_at is None:
+        return TickDecision(offset=0.0, due=(), finished=False, action=TickAction.STOP)
+
+    offset = clock_offset(
+        elapsed_seconds=elapsed_seconds(now=now, started_at=started_at),
+        time_scale=time_scale,
+    )
+    schedule = decide_schedule(
+        plan=plan,
+        window_seconds=window_seconds,
+        clock_offset_seconds=offset,
+        spawned_indices=spawned_indices,
+    )
+    action = TickAction.COMPLETE if schedule.finished else TickAction.CONTINUE
+    return TickDecision(offset=offset, due=schedule.due, finished=schedule.finished, action=action)
+
+
+def resolve_finalize(current_status: CampaignStatus) -> TickAction:
+    """Cancel-race guard: complete only if the campaign is *still* RUNNING.
+
+    Called by the shell after spawning, on a freshly re-read status, so a
+    cancellation that landed between finish-detection and the COMPLETED write
+    wins — a cancelled campaign stays cancelled.
+    """
+    return TickAction.COMPLETE if current_status == CampaignStatus.RUNNING else TickAction.STOP

tests/unit/test_campaign_scheduler.py

@@ -1,10 +1,16 @@
 """Unit tests for the pure campaign scheduling decision (no I/O)."""
 
+from datetime import datetime, timedelta, timezone
+
 from agenteval.evaluation.campaign_scheduler import (
+    TickAction,
     clock_offset,
     decide_schedule,
+    decide_tick,
+    elapsed_seconds,
+    resolve_finalize,
 )
-from agenteval.models import CampaignPlanEntry
+from agenteval.models import CampaignPlanEntry, CampaignStatus
 
 
 def _plan() -> list[CampaignPlanEntry]:
@@ -74,3 +80,78 @@ def test_not_finished_inside_window():
 def test_finished_at_window_end():
     decision = decide_schedule(plan=_plan(), window_seconds=7200, clock_offset_seconds=7200)
     assert decision.finished is True
+
+
+# ── elapsed_seconds (pure, clock injected) ───────────────────────────────────
+
+T0 = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
+
+
+def test_elapsed_seconds_aware():
+    assert elapsed_seconds(now=T0 + timedelta(seconds=90), started_at=T0) == 90.0
+
+
+def test_elapsed_seconds_tolerates_naive_started_at():
+    # SQLite hands back naive datetimes; treat them as UTC (AGENTS.md #2).
+    naive = T0.replace(tzinfo=None)
+    assert elapsed_seconds(now=T0 + timedelta(seconds=30), started_at=naive) == 30.0
+
+
+def test_elapsed_seconds_negative_when_now_before_start():
+    assert elapsed_seconds(now=T0 - timedelta(seconds=5), started_at=T0) == -5.0
+
+
+# ── decide_tick (lifecycle + due, one pure decision) ─────────────────────────
+
+def _tick(*, now, started_at, status, offset_plan=None, window=7200, scale=1.0, spawned=()):
+    return decide_tick(
+        now=now,
+        started_at=started_at,
+        status=status,
+        window_seconds=window,
+        time_scale=scale,
+        plan=offset_plan if offset_plan is not None else _plan(),
+        spawned_indices=spawned,
+    )
+
+
+def test_tick_stop_when_not_running():
+    d = _tick(now=T0 + timedelta(seconds=1), started_at=T0, status=CampaignStatus.CANCELLED)
+    assert d.action is TickAction.STOP
+
+
+def test_tick_stop_when_no_started_at():
+    d = _tick(now=T0, started_at=None, status=CampaignStatus.RUNNING)
+    assert d.action is TickAction.STOP
+
+
+def test_tick_continue_inside_window_reports_due_and_offset():
+    d = _tick(now=T0, started_at=T0, status=CampaignStatus.RUNNING)
+    assert d.action is TickAction.CONTINUE
+    assert d.finished is False
+    assert d.offset == 0.0
+    assert [due.index for due in d.due] == [0]  # only offset-0 entry due at t=0
+
+
+def test_tick_complete_at_window_end():
+    d = _tick(now=T0 + timedelta(seconds=7200), started_at=T0, status=CampaignStatus.RUNNING)
+    assert d.action is TickAction.COMPLETE
+    assert d.finished is True
+
+
+def test_tick_time_scale_compresses_window():
+    # window 7200s at 7200x → 1 real second reaches the window end.
+    d = _tick(now=T0 + timedelta(seconds=1), started_at=T0, status=CampaignStatus.RUNNING, scale=7200.0)
+    assert d.offset == 7200.0
+    assert d.action is TickAction.COMPLETE
+
+
+# ── resolve_finalize (cancel-race guard, pure) ───────────────────────────────
+
+def test_resolve_finalize_completes_when_still_running():
+    assert resolve_finalize(CampaignStatus.RUNNING) is TickAction.COMPLETE
+
+
+def test_resolve_finalize_stops_when_cancelled_meanwhile():
+    # Cancel landed between finish-detection and the write — do not overwrite it.
+    assert resolve_finalize(CampaignStatus.CANCELLED) is TickAction.STOP