diff --git a/backend/agenteval/evaluation/campaign_runner.py b/backend/agenteval/evaluation/campaign_runner.py new file mode 100644 index 0000000..af4d889 --- /dev/null +++ b/backend/agenteval/evaluation/campaign_runner.py @@ -0,0 +1,113 @@ +"""Campaign runner — the side-effect shell around the pure scheduler. + +Given a campaign and an injected clock position (real elapsed seconds), it asks +``campaign_scheduler.decide_schedule`` what is due, then spawns those child Runs +by reusing the existing single-run execution path (``EvalEngine.run`` with an +``existing_run`` that carries ``campaign_id``). Spawned plan-entry indices are +persisted on the campaign so re-advancing the same clock never double-spawns. + +This module deliberately has no timer/loop of its own — ticket 02 is driven by +tests or a manual clock advance. The durable async loop and restart recovery +arrive in a later ticket. +""" + +from dataclasses import dataclass, field +from typing import Optional + +from sqlmodel import Session + +from agenteval.evaluation.campaign_scheduler import clock_offset, decide_schedule +from agenteval.evaluation.engine import EvalEngine +from agenteval.models import Campaign, EvalRun, RunStatus, RunTrigger +from agenteval.storage.repository import ( + CampaignRepository, + RunRepository, + ScenarioRepository, + TargetRepository, +) + +_SPAWNED_KEY = "spawned_indices" + + +@dataclass +class AdvanceResult: + """Outcome of advancing a campaign's clock once.""" + + spawned_run_ids: list[str] = field(default_factory=list) + finished: bool = False + + +def _spawned_indices(campaign: Campaign) -> set[int]: + scheduler = (campaign.summary or {}).get("scheduler", {}) + return set(scheduler.get(_SPAWNED_KEY, [])) + + +async def _spawn_child_run(campaign: Campaign, scenario_id: str, session: Session) -> str: + """Create a Run owned by the campaign and drive it through the engine. + + The Run row is created on the caller's ``session``, but the engine runs on + its own session (obtained internally, closed when the run ends) so one + child Run's lifecycle never closes the campaign's session — the same + isolation the single-run background task uses. + """ + target = TargetRepository(session).get(campaign.target_id) + scenario = ScenarioRepository(session).get(scenario_id) + if not target or not scenario: + raise ValueError(f"campaign target or scenario missing: {campaign.target_id}/{scenario_id}") + + run = RunRepository(session).create( + EvalRun( + target_id=campaign.target_id, + scenario_id=scenario_id, + scenario_version=scenario.version or 1, + campaign_id=campaign.id, + triggered_by=RunTrigger.CAMPAIGN, + status=RunStatus.PENDING, + ) + ) + engine = EvalEngine(target=target, scenario=scenario, triggered_by=RunTrigger.CAMPAIGN) + await engine.run(existing_run=run) + return run.id or "" + + +async def advance_campaign( + *, + campaign_id: str, + elapsed_seconds: float, + session: Session, +) -> Optional[AdvanceResult]: + """Advance the campaign clock to ``elapsed_seconds`` and spawn due Runs. + + ``elapsed_seconds`` is real wall-clock time since the window started; it is + mapped to a window offset via ``time_scale``. Returns ``None`` if the + campaign does not exist. + """ + repo = CampaignRepository(session) + campaign = repo.get(campaign_id) + if not campaign: + return None + + offset = clock_offset(elapsed_seconds=elapsed_seconds, time_scale=campaign.time_scale) + spawned = _spawned_indices(campaign) + decision = decide_schedule( + plan=campaign.plan, + window_seconds=campaign.window_seconds, + clock_offset_seconds=offset, + spawned_indices=spawned, + ) + + result = AdvanceResult(finished=decision.finished) + for due in decision.due: + for _ in range(due.entry.count): + run_id = await _spawn_child_run(campaign, due.entry.scenario_id, session) + result.spawned_run_ids.append(run_id) + spawned.add(due.index) + # Persist progress per entry: a failure partway through a multi-entry + # advance must never lose which entries already spawned, since restart + # recovery (later ticket) reads this back from the DB. + summary = dict(campaign.summary or {}) + summary["scheduler"] = {_SPAWNED_KEY: sorted(spawned)} + campaign.summary = summary + repo.update(campaign) + + return result diff --git a/backend/agenteval/evaluation/campaign_scheduler.py b/backend/agenteval/evaluation/campaign_scheduler.py new file mode 100644 index 0000000..27edb05 --- /dev/null +++ b/backend/agenteval/evaluation/campaign_scheduler.py @@ -0,0 +1,65 @@ +"""Campaign scheduling decision — pure functions (no I/O). + +This is the key testable seam for the durable scheduler (ADR-0003), mirroring +``judgement.combine_case_outcome``. Given a campaign's static plan and the +current window-clock position, it decides which plan entries are due to spawn +now and whether the window has ended. The async loop and Run-spawning side +effects live in ``campaign_runner``; this module never touches the database, +the clock, or ``time_scale`` semantics beyond the offset mapping below. + +Idempotency is the caller's responsibility: it passes the set of already-spawned +plan-entry indices, and this function excludes them from the due list. +""" + +from dataclasses import dataclass +from typing import Iterable + +from agenteval.models import CampaignPlanEntry + + +def clock_offset(*, elapsed_seconds: float, time_scale: float) -> float: + """Map real elapsed wall-clock time to a window offset. + + ``time_scale`` only changes *when* an entry becomes due — 1.0 is real wall + clock (production), larger compresses the window (development). It never + touches judgement or report semantics. + """ + return elapsed_seconds * time_scale + + +@dataclass(frozen=True) +class DueEntry: + """A plan entry that is due to spawn, tagged with its plan index.""" + + index: int + entry: CampaignPlanEntry + + +@dataclass(frozen=True) +class ScheduleDecision: + """What the scheduler should do at one clock position.""" + + due: tuple[DueEntry, ...] + finished: bool + + +def decide_schedule( + *, + plan: list[CampaignPlanEntry], + window_seconds: int, + clock_offset_seconds: float, + spawned_indices: Iterable[int] = (), +) -> ScheduleDecision: + """Decide which plan entries are due at ``clock_offset_seconds``. + + An entry is due once the clock reaches its offset and it hasn't been + spawned yet. ``finished`` is True once the clock reaches the window end. + """ + already = set(spawned_indices) + due = tuple( + DueEntry(index=i, entry=entry) + for i, entry in enumerate(plan) + if i not in already and entry.offset_seconds <= clock_offset_seconds + ) + finished = clock_offset_seconds >= window_seconds + return ScheduleDecision(due=due, finished=finished) diff --git a/backend/agenteval/models.py b/backend/agenteval/models.py index 5b6a908..1d9f340 100644 --- a/backend/agenteval/models.py +++ b/backend/agenteval/models.py @@ -154,6 +154,7 @@ class RunTrigger(str, Enum): MANUAL = "manual" AI_ASSISTANT = "ai_assistant" CLI = "cli" + CAMPAIGN = "campaign" class EvalRun(BaseModel): diff --git a/backend/agenteval/storage/repository.py b/backend/agenteval/storage/repository.py index fc3e2d0..61b33a0 100644 --- a/backend/agenteval/storage/repository.py +++ b/backend/agenteval/storage/repository.py @@ -390,6 +390,25 @@ class CampaignRepository: self.session.refresh(db) return _campaign_from_db(db) + def update(self, campaign: Campaign) -> Optional[Campaign]: + existing = self.session.get(CampaignDB, campaign.id) + if not existing: + return None + existing.name = campaign.name + existing.target_id = campaign.target_id + existing.window_seconds = campaign.window_seconds + existing.time_scale = campaign.time_scale + existing.set_plan([entry.model_dump(mode="json") for entry in campaign.plan]) + existing.status = campaign.status.value + existing.started_at = campaign.started_at + existing.completed_at = campaign.completed_at + if campaign.summary is not None: + existing.set_summary(campaign.summary) + self.session.add(existing) + self.session.commit() + self.session.refresh(existing) + return _campaign_from_db(existing) + class ResultRepository: """Repository for evaluation results.""" diff --git a/tests/integration/test_campaign_runner.py b/tests/integration/test_campaign_runner.py new file mode 100644 index 0000000..140de2f --- /dev/null +++ b/tests/integration/test_campaign_runner.py @@ -0,0 +1,111 @@ +"""Integration test for campaign Run-spawning driven by a manual clock. + +Drives ``advance_campaign`` at injected clock positions against a +compressed-time-scale campaign and asserts the spawned child Runs match the +plan (count, ownership, scenario) and complete with results/summary in the DB. +No real timer is used — the clock is injected, mirroring how the durable loop +(later ticket) will call the same seam. +""" + +import pytest + +from agenteval.evaluation.campaign_runner import advance_campaign +from agenteval.models import ( + Campaign, CampaignPlanEntry, Case, CaseType, ChannelType, EvalTarget, + PlatformType, RunStatus, RunTrigger, Scenario, TargetStatus, +) +from agenteval.storage.repository import CampaignRepository, RunRepository, ScenarioRepository, TargetRepository +from tests.unit.mock_channel import MockChannel + + +@pytest.fixture() +def seeded_db(db_session, monkeypatch): + """Point every get_session consumer at the test session, seed target + + scenario, and make the channel a MockChannel so no network is hit.""" + from agenteval.channels import factory as factory_module + from agenteval.evaluation import engine as engine_module + from agenteval.storage import db as db_module + from agenteval.storage import repository as repo_module + + def _test_get_session(): + return db_session + + monkeypatch.setattr(db_module, "get_session", _test_get_session) + monkeypatch.setattr(repo_module, "get_session", _test_get_session) + monkeypatch.setattr(engine_module, "get_session", _test_get_session) + + channel = MockChannel(reply_delay=0.0) + monkeypatch.setattr(factory_module.ChannelFactory, "create", lambda target: channel) + + target = EvalTarget( + id="t-1", name="mock-target", + platform=PlatformType.AI_DIGITAL_EMPLOYEE, + channel_type=ChannelType.TUTU_API, + channel_config={"base_url": "http://mock", "token": "x"}, + status=TargetStatus.ACTIVE, + ) + TargetRepository(db_session).create(target) + + scenario = Scenario( + id="s-1", name="mock-scenario", + cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])], + ) + ScenarioRepository(db_session).create(scenario) + + return db_session + + +def _make_campaign(session) -> Campaign: + return CampaignRepository(session).create(Campaign( + name="compressed", + target_id="t-1", + window_seconds=7200, + time_scale=3600.0, # 1 real second == 3600 window seconds + plan=[ + CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=2), + CampaignPlanEntry(scenario_id="s-1", offset_seconds=3600, count=1), + ], + )) + + +async def test_advance_spawns_due_runs_matching_plan(seeded_db): + campaign = _make_campaign(seeded_db) + + # t=0 → window offset 0 → only entry 0 (count 2) is due. + r0 = await advance_campaign(campaign_id=campaign.id, elapsed_seconds=0.0, session=seeded_db) + assert len(r0.spawned_run_ids) == 2 + assert r0.finished is False + + # t=1s → window offset 3600 → entry 1 (count 1) becomes due. + r1 = await advance_campaign(campaign_id=campaign.id, elapsed_seconds=1.0, session=seeded_db) + assert len(r1.spawned_run_ids) == 1 + + runs = RunRepository(seeded_db).list_all() + assert len(runs) == 3 + for run in runs: + assert run.campaign_id == campaign.id + assert run.scenario_id == "s-1" + assert run.triggered_by == RunTrigger.CAMPAIGN + assert run.status == RunStatus.COMPLETED + assert run.summary["total_cases"] == 1 + + +async def test_advance_is_idempotent(seeded_db): + campaign = _make_campaign(seeded_db) + + await advance_campaign(campaign_id=campaign.id, elapsed_seconds=1.0, session=seeded_db) + # Re-advancing to the same clock must not double-spawn. + again = await advance_campaign(campaign_id=campaign.id, elapsed_seconds=1.0, session=seeded_db) + assert again.spawned_run_ids == [] + assert len(RunRepository(seeded_db).list_all()) == 3 # entry0(2) + entry1(1) + + +async def test_advance_reports_finished_at_window_end(seeded_db): + campaign = _make_campaign(seeded_db) + # t=2s → offset 7200 == window end. + result = await advance_campaign(campaign_id=campaign.id, elapsed_seconds=2.0, session=seeded_db) + assert result.finished is True + + +async def test_advance_missing_campaign_returns_none(seeded_db): + assert await advance_campaign(campaign_id="nope", elapsed_seconds=0.0, session=seeded_db) is None diff --git a/tests/unit/test_campaign_scheduler.py b/tests/unit/test_campaign_scheduler.py new file mode 100644 index 0000000..fd48483 --- /dev/null +++ b/tests/unit/test_campaign_scheduler.py @@ -0,0 +1,76 @@ +"""Unit tests for the pure campaign scheduling decision (no I/O).""" + +from agenteval.evaluation.campaign_scheduler import ( + clock_offset, + decide_schedule, +) +from agenteval.models import CampaignPlanEntry + + +def _plan() -> list[CampaignPlanEntry]: + return [ + CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=2), + CampaignPlanEntry(scenario_id="s-2", offset_seconds=3600, count=1), + CampaignPlanEntry(scenario_id="s-3", offset_seconds=3600, count=3), + ] + + +# ── clock mapping ────────────────────────────────────────────────────────── + +def test_clock_offset_real_wall_clock(): + assert clock_offset(elapsed_seconds=120.0, time_scale=1.0) == 120.0 + + +def test_clock_offset_compressed(): + # 1 real second maps to 3600 window seconds at 3600x. + assert clock_offset(elapsed_seconds=1.0, time_scale=3600.0) == 3600.0 + + +# ── due / not-due matrix ───────────────────────────────────────────────────── + +def test_nothing_due_before_first_offset(): + # Only offset-0 entries exist at t<0? Use a plan whose earliest offset is 10. + plan = [CampaignPlanEntry(scenario_id="s-1", offset_seconds=10, count=1)] + decision = decide_schedule(plan=plan, window_seconds=100, clock_offset_seconds=5) + assert decision.due == () + assert decision.finished is False + + +def test_offset_zero_entry_due_at_start(): + decision = decide_schedule(plan=_plan(), window_seconds=7200, clock_offset_seconds=0) + assert [d.index for d in decision.due] == [0] + assert decision.due[0].entry.count == 2 + + +def test_multiple_entries_due_at_same_offset(): + decision = decide_schedule(plan=_plan(), window_seconds=7200, clock_offset_seconds=3600) + # entry 0 (offset 0) plus entries 1 and 2 (offset 3600) are all due. + assert [d.index for d in decision.due] == [0, 1, 2] + + +def test_spawned_entries_excluded_idempotent(): + decision = decide_schedule( + plan=_plan(), window_seconds=7200, clock_offset_seconds=3600, + spawned_indices=[0, 1], + ) + assert [d.index for d in decision.due] == [2] + + +def test_all_spawned_yields_no_due(): + decision = decide_schedule( + plan=_plan(), window_seconds=7200, clock_offset_seconds=99999, + spawned_indices=[0, 1, 2], + ) + assert decision.due == () + + +# ── window end ─────────────────────────────────────────────────────────────── + +def test_not_finished_inside_window(): + decision = decide_schedule(plan=_plan(), window_seconds=7200, clock_offset_seconds=7199) + assert decision.finished is False + + +def test_finished_at_window_end(): + decision = decide_schedule(plan=_plan(), window_seconds=7200, clock_offset_seconds=7200) + assert decision.finished is True