feat(campaigns): add scheduling decision and child-Run spawning

Add the pure scheduling seam (campaign_scheduler.decide_schedule) that, given a
static plan and window-clock offset, decides which plan entries are due and
whether the window ended — mirroring judgement.combine_case_outcome, with
time_scale confined to the clock mapping so it never touches judgement/report.

The campaign_runner shell maps injected elapsed time to a window offset, spawns
due child Runs through the existing EvalEngine.run(existing_run=...) path with
campaign_id + RunTrigger.CAMPAIGN, and persists spawned-entry indices per entry
for idempotent, restart-recoverable progress. No auto loop yet (ticket 03).

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

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)

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):

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."""

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

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