From e815298ce5b9cd8fbfd3276f26fd8378e187543b Mon Sep 17 00:00:00 2001 From: sinohqb Date: Fri, 31 Jul 2026 02:20:14 +0800 Subject: [PATCH] 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. --- .../agenteval/evaluation/campaign_runner.py | 49 +++++++---- .../evaluation/campaign_scheduler.py | 83 ++++++++++++++++++- tests/unit/test_campaign_scheduler.py | 83 ++++++++++++++++++- 3 files changed, 194 insertions(+), 21 deletions(-) diff --git a/backend/agenteval/evaluation/campaign_runner.py b/backend/agenteval/evaluation/campaign_runner.py index 38e1a9b..b767257 100644 --- a/backend/agenteval/evaluation/campaign_runner.py +++ b/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) diff --git a/backend/agenteval/evaluation/campaign_scheduler.py b/backend/agenteval/evaluation/campaign_scheduler.py index 27edb05..32e1a11 100644 --- a/backend/agenteval/evaluation/campaign_scheduler.py +++ b/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 diff --git a/tests/unit/test_campaign_scheduler.py b/tests/unit/test_campaign_scheduler.py index fd48483..809660a 100644 --- a/tests/unit/test_campaign_scheduler.py +++ b/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