feat(R07): task repository, schedule calculator, Qt clock adapter, task/AI-planner services

Team Hoa, EPIC R07 (Scheduling & Workflow Runtime) - Team Hoa scope only
(R07-T01 -> T05; R07-T06 Co4EWorkflowService is Team Nam's).

- R07-T01: infrastructure/persistence/json/task_repository_impl.py wraps
  core/tasks.py's CRUD; core/tasks.py::save_task now writes through
  atomic_write.write_json (same durability fix as R06-T02, save_task was
  still doing a plain write_text).
- R07-T02: domain/tasks/schedule_calculator.py::ScheduleCalculator - the
  cron/interval/daily/weekly/monthly due-time math extracted from
  core/tasks.py, pure Python with is_holiday/make_cron injected so domain/
  never imports core (ADR-001 I2). core/tasks.py keeps its old function
  names as thin wrappers so every existing caller is unchanged. This was
  previously untested; now has its own unit suite.
- R07-T03: infrastructure/qt/qt_scheduler_clock.py::QtSchedulerClock wraps
  the QTimer TaskScheduler used to own directly, injected via a new
  `clock=` constructor param (defaults to a real one). Originally planned
  at platform/qt/... ; moved after confirming that name shadows the
  stdlib platform module (used by core/windows_sandbox_vm.py,
  core/appcontainer_sandbox.py) whenever the repo root is on sys.path.
  tests/fakes/fake_clock.py lets scheduler dispatch be tested tick-by-tick
  with no Qt event loop.
- R07-T04: application/scheduling/task_application_service.py centralizes
  run_now/duplicate/pause/delete/bulk_delete and the Kanban drag-drop
  business rules (move_to_status), currently only reachable by driving
  the real ui/schedule_task_tab.py widget.
- R07-T05: application/scheduling/ai_task_planner_service.py wraps
  core/ai_task_planner.py::plan_tasks and core/task_import.py::import_tasks
  as a seam, plus the attachment-stamping step that used to only exist
  inside the AI-create dialog's worker closure.

pytest: 328 pass (same 4 pre-existing failures as the R05/R06 baseline,
unrelated to this work - see docs/refactor/BaoCao_TeamHoa_R05_R06.md).
scripts/check_imports.py: PASS.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

tests/unit/test_ai_task_planner_service.py

@@ -0,0 +1,87 @@
+"""EPIC R07-T05: AiTaskPlannerService — AI-generate + import, no Qt, no network.
+
+``core/ai_task_planner.py::plan_tasks`` and ``core/task_import.py::
+import_tasks`` are exercised through a FakeProvider / real tmp files rather
+than reimplemented — this service is a seam, not a new planner.
+"""
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from cowork_local.application.scheduling.ai_task_planner_service import (
+    AiTaskPlannerService,
+)
+from tests.fakes import FakeProvider, ScriptedTurn
+
+_PLAN_REPLY = json.dumps({
+    "tasks": [
+        {"title": "Draft report", "description": "d", "task_type": "cowork",
+         "priority": "medium", "schedule": {"enabled": False}}
+    ]
+})
+
+
+def test_plan_uses_the_constructor_injected_provider_factory():
+    provider = FakeProvider([ScriptedTurn(text=_PLAN_REPLY)])
+    service = AiTaskPlannerService(provider_factory=lambda: provider)
+
+    tasks = service.plan("Write a weekly report")
+
+    assert len(tasks) == 1
+    assert tasks[0]["title"] == "Draft report"
+    assert provider.call_count == 1
+
+
+def test_plan_prefers_an_explicit_provider_over_the_factory():
+    factory_provider = FakeProvider([ScriptedTurn(text=_PLAN_REPLY)], strict=False)
+    explicit_provider = FakeProvider([ScriptedTurn(text=_PLAN_REPLY)])
+    service = AiTaskPlannerService(provider_factory=lambda: factory_provider)
+
+    service.plan("Write a weekly report", provider=explicit_provider)
+
+    assert explicit_provider.call_count == 1
+    assert factory_provider.call_count == 0
+
+
+def test_plan_without_any_provider_raises_runtime_error():
+    service = AiTaskPlannerService(provider_factory=None)
+    with pytest.raises(RuntimeError):
+        service.plan("Write a weekly report")
+
+
+def test_plan_stamps_attachments_onto_every_generated_task():
+    two_tasks_reply = json.dumps({"tasks": [
+        {"title": "A", "task_type": "cowork"},
+        {"title": "B", "task_type": "cowork"},
+    ]})
+    provider = FakeProvider([ScriptedTurn(text=two_tasks_reply)])
+    service = AiTaskPlannerService(provider_factory=lambda: provider)
+
+    tasks = service.plan("do two things", file_paths=["a.txt"], links=["https://x"])
+
+    assert len(tasks) == 2
+    for t in tasks:
+        assert t["input"]["file_paths"] == ["a.txt"]
+        assert t["input"]["links"] == ["https://x"]
+
+
+def test_import_file_delegates_to_core_task_import(tmp_path):
+    csv_path = tmp_path / "tasks.csv"
+    csv_path.write_text("title,task_type,priority\nMy Task,cowork,medium\n", encoding="utf-8")
+    service = AiTaskPlannerService()
+
+    tasks = service.import_file(csv_path)
+
+    assert len(tasks) == 1
+    assert tasks[0]["title"] == "My Task"
+
+
+def test_import_file_raises_value_error_on_unsupported_extension(tmp_path):
+    bogus = tmp_path / "tasks.txt"
+    bogus.write_text("nope", encoding="utf-8")
+    service = AiTaskPlannerService()
+
+    with pytest.raises(ValueError):
+        service.import_file(bogus)

tests/unit/test_schedule_calculator.py

@@ -0,0 +1,164 @@
+"""EPIC R07-T02: ScheduleCalculator — pure due-time/cron math.
+
+This is the one piece of scheduling logic core/tasks.py's own docstring
+claimed was "Qt-free so it can be unit-tested headlessly" but had NO unit
+test at all before this task (confirmed by grepping tests/ for
+"schedule_calculator"/"compute_next_run"/"cron" — nothing matched). These
+tests exercise domain/tasks/schedule_calculator.py directly, with no Qt, no
+filesystem, and fake holiday/cron callables so the module stays provably
+zero-I/O.
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+import pytest
+
+from cowork_local.domain.tasks.schedule_calculator import ScheduleCalculator
+
+
+def _sched(**overrides):
+    base = {
+        "enabled": True,
+        "run_at": "2026-08-24 09:00",   # a Monday
+        "repeat_type": "none",
+        "cron_expression": None,
+        "working_days_only": False,
+        "skip_holidays": False,
+        "holiday_country": "VN",
+    }
+    base.update(overrides)
+    return base
+
+
+def _task(**sched_overrides):
+    return {"status": "scheduled", "schedule": _sched(**sched_overrides)}
+
+
+class _FakeCron:
+    """A cron stub that fires every day at a fixed hour:minute — enough to
+    exercise the cron branch without depending on core/cron.py::Cron."""
+
+    def __init__(self, expression: str):
+        if expression == "bad":
+            raise ValueError("bad cron expression")
+        self.hour, self.minute = 10, 0
+
+    def next_after(self, after: datetime) -> datetime:
+        candidate = after.replace(hour=self.hour, minute=self.minute, second=0, microsecond=0)
+        if candidate <= after:
+            from datetime import timedelta
+            candidate += timedelta(days=1)
+        return candidate
+
+
+def _is_weekend_holiday(date_, country):
+    # A deterministic fake: only 2026-08-29 (a Saturday) counts as a holiday.
+    return date_.isoformat() == "2026-08-29"
+
+
+def test_daily_advances_by_one_day():
+    calc = ScheduleCalculator()
+    task = _task(repeat_type="daily")
+    nxt = calc.compute_next_run(task, datetime(2026, 8, 24, 9, 0))
+    assert nxt == datetime(2026, 8, 25, 9, 0)
+
+
+def test_weekly_advances_by_seven_days():
+    calc = ScheduleCalculator()
+    task = _task(repeat_type="weekly")
+    nxt = calc.compute_next_run(task, datetime(2026, 8, 24, 9, 0))
+    assert nxt == datetime(2026, 8, 31, 9, 0)
+
+
+def test_monthly_clamps_day_to_shorter_month():
+    calc = ScheduleCalculator()
+    task = _task(run_at="2026-01-31 09:00", repeat_type="monthly")
+    nxt = calc.compute_next_run(task, datetime(2026, 1, 31, 9, 0))
+    assert nxt == datetime(2026, 2, 28, 9, 0)   # Feb 2026 has 28 days
+
+
+def test_one_shot_repeat_type_none_has_no_next_run():
+    calc = ScheduleCalculator()
+    task = _task(repeat_type="none")
+    assert calc.compute_next_run(task, datetime(2026, 8, 24, 9, 0)) is None
+
+
+def test_cron_without_injected_factory_returns_none():
+    """No make_cron wired in -> a cron schedule simply never produces a next
+    run, instead of crashing the caller."""
+    calc = ScheduleCalculator()
+    task = _task(repeat_type="cron", cron_expression="0 10 * * *")
+    assert calc.compute_next_run(task, datetime(2026, 8, 24, 9, 0)) is None
+
+
+def test_cron_uses_injected_factory():
+    calc = ScheduleCalculator(make_cron=_FakeCron)
+    task = _task(repeat_type="cron", cron_expression="0 10 * * *")
+    nxt = calc.compute_next_run(task, datetime(2026, 8, 24, 9, 0))
+    assert nxt == datetime(2026, 8, 24, 10, 0)
+
+
+def test_malformed_cron_expression_returns_none_not_raise():
+    calc = ScheduleCalculator(make_cron=_FakeCron)
+    task = _task(repeat_type="cron", cron_expression="bad")
+    assert calc.compute_next_run(task, datetime(2026, 8, 24, 9, 0)) is None
+
+
+def test_working_days_only_skips_weekend():
+    calc = ScheduleCalculator()
+    # 2026-08-28 is a Friday; +1 day (daily) would land on Saturday 08-29.
+    task = _task(run_at="2026-08-28 09:00", repeat_type="daily", working_days_only=True)
+    nxt = calc.compute_next_run(task, datetime(2026, 8, 28, 9, 0))
+    assert nxt.weekday() < 5   # Monday 08-31, not the weekend
+
+
+def test_skip_holidays_uses_injected_is_holiday():
+    calc = ScheduleCalculator(is_holiday=_is_weekend_holiday)
+    # 2026-08-28 (Fri) + 1 day = 2026-08-29, which the fake marks a holiday.
+    task = _task(run_at="2026-08-28 09:00", repeat_type="daily", skip_holidays=True)
+    nxt = calc.compute_next_run(task, datetime(2026, 8, 28, 9, 0))
+    assert nxt == datetime(2026, 8, 30, 9, 0)   # skipped past the holiday
+
+
+def test_skip_holidays_without_injected_is_holiday_degrades_gracefully():
+    calc = ScheduleCalculator()   # no is_holiday wired in
+    task = _task(run_at="2026-08-28 09:00", repeat_type="daily", skip_holidays=True)
+    nxt = calc.compute_next_run(task, datetime(2026, 8, 28, 9, 0))
+    assert nxt == datetime(2026, 8, 29, 9, 0)   # no holiday check applied at all
+
+
+def test_shift_off_excluded_days_moves_weekend_forward():
+    calc = ScheduleCalculator()
+    sched = _sched(working_days_only=True)
+    saturday = datetime(2026, 8, 29, 9, 0)
+    shifted = calc.shift_off_excluded_days(saturday, sched)
+    assert shifted.weekday() < 5
+    assert shifted >= saturday
+
+
+def test_add_month_end_of_year_rolls_to_january():
+    calc = ScheduleCalculator()
+    assert calc.add_month(datetime(2026, 12, 15, 9, 0)) == datetime(2027, 1, 15, 9, 0)
+
+
+def test_due_tasks_filters_by_status_enabled_and_run_at():
+    calc = ScheduleCalculator()
+    now = datetime(2026, 8, 27, 12, 0)
+    due_now = _task(run_at="2026-08-27 09:00")
+    due_now["title"] = "due"
+    future = _task(run_at="2026-08-28 09:00")
+    future["title"] = "future"
+    disabled = _task(run_at="2026-08-27 09:00", enabled=False)
+    disabled["title"] = "disabled"
+    not_scheduled = _task(run_at="2026-08-27 09:00")
+    not_scheduled["title"] = "backlog"
+    not_scheduled["status"] = "backlog"
+
+    result = calc.due_tasks([due_now, future, disabled, not_scheduled], now)
+    assert [t["title"] for t in result] == ["due"]
+
+
+def test_due_tasks_empty_when_no_tasks():
+    calc = ScheduleCalculator()
+    assert calc.due_tasks([], datetime(2026, 8, 27, 12, 0)) == []

tests/unit/test_task_application_service.py

@@ -0,0 +1,191 @@
+"""EPIC R07-T04: TaskApplicationService — CRUD + dispatch rules, no Qt.
+
+Everything here used to be exercised only by driving the real
+``ui/schedule_task_tab.py`` widget (a QListWidget drag gesture, a QMenu
+click). These tests drive the same rules directly through the service.
+"""
+from __future__ import annotations
+
+from cowork_local.application.scheduling.task_application_service import (
+    TaskApplicationService,
+)
+from cowork_local.infrastructure.persistence.json import TaskRepository
+
+
+def _service(tmp_path, run_now=None):
+    return TaskApplicationService(TaskRepository(tmp_path), run_now=run_now)
+
+
+def _new_saved_task(repo: TaskRepository, **overrides):
+    task = repo.create("T", **overrides)
+    repo.save(task)
+    return task
+
+
+def test_run_now_rejects_manual_task_type(tmp_path):
+    repo = TaskRepository(tmp_path)
+    task = _new_saved_task(repo, task_type="manual")
+    service = TaskApplicationService(repo, run_now=lambda tid: True)
+
+    result = service.run_now(task["task_id"])
+
+    assert result.ok is False
+    assert result.reason == "manual_task"
+
+
+def test_run_now_without_scheduler_wired_reports_no_scheduler(tmp_path):
+    repo = TaskRepository(tmp_path)
+    task = _new_saved_task(repo, task_type="cowork")
+    service = TaskApplicationService(repo, run_now=None)
+
+    result = service.run_now(task["task_id"])
+
+    assert result.ok is False
+    assert result.reason == "no_scheduler"
+
+
+def test_run_now_delegates_to_injected_scheduler(tmp_path):
+    repo = TaskRepository(tmp_path)
+    task = _new_saved_task(repo, task_type="cowork")
+    seen = []
+    service = TaskApplicationService(repo, run_now=lambda tid: seen.append(tid) or True)
+
+    result = service.run_now(task["task_id"])
+
+    assert result.ok is True
+    assert seen == [task["task_id"]]
+
+
+def test_run_now_missing_task_reports_not_found(tmp_path):
+    service = _service(tmp_path, run_now=lambda tid: True)
+    result = service.run_now("does-not-exist")
+    assert result.ok is False
+    assert result.reason == "not_found"
+
+
+def test_duplicate_saves_a_copy_with_fresh_identity(tmp_path):
+    repo = TaskRepository(tmp_path)
+    task = _new_saved_task(repo, description="d")
+    service = TaskApplicationService(repo)
+
+    dup = service.duplicate(task["task_id"])
+
+    assert dup is not None
+    assert dup["task_id"] != task["task_id"]
+    assert dup["description"] == "d"
+    assert repo.get(dup["task_id"]) is not None   # actually persisted, not just returned
+
+
+def test_toggle_pause_then_resume_goes_to_backlog(tmp_path):
+    repo = TaskRepository(tmp_path)
+    task = _new_saved_task(repo)
+    service = TaskApplicationService(repo)
+
+    paused = service.toggle_pause(task["task_id"])
+    assert paused["status"] == "paused"
+
+    resumed = service.toggle_pause(task["task_id"])
+    assert resumed["status"] == "backlog"
+
+
+def test_delete_reports_whether_the_task_existed(tmp_path):
+    repo = TaskRepository(tmp_path)
+    task = _new_saved_task(repo)
+    service = TaskApplicationService(repo)
+
+    assert service.delete(task["task_id"]) is True
+    assert repo.get(task["task_id"]) is None
+    assert service.delete(task["task_id"]) is False   # already gone
+
+
+def test_bulk_delete_counts_only_tasks_that_existed(tmp_path):
+    repo = TaskRepository(tmp_path)
+    a = _new_saved_task(repo)
+    b = _new_saved_task(repo)
+    service = TaskApplicationService(repo)
+
+    count = service.bulk_delete([a["task_id"], b["task_id"], "ghost-id"])
+
+    assert count == 2
+    assert repo.list() == []
+
+
+def test_move_to_status_running_task_is_blocked(tmp_path):
+    repo = TaskRepository(tmp_path)
+    task = _new_saved_task(repo)
+    task["status"] = "running"
+    repo.save(task)
+    service = TaskApplicationService(repo)
+
+    result = service.move_to_status(task["task_id"], "backlog")
+
+    assert result.blocked is True
+    assert repo.get(task["task_id"])["status"] == "running"   # untouched
+
+
+def test_move_to_status_running_lane_dispatches_run_now(tmp_path):
+    repo = TaskRepository(tmp_path)
+    task = _new_saved_task(repo, task_type="cowork")
+    seen = []
+    service = TaskApplicationService(repo, run_now=lambda tid: seen.append(tid) or True)
+
+    result = service.move_to_status(task["task_id"], "running")
+
+    assert result.ran_now is True
+    assert result.run_now_result.ok is True
+    assert seen == [task["task_id"]]
+
+
+def test_move_to_status_done_disables_the_schedule(tmp_path):
+    repo = TaskRepository(tmp_path)
+    task = _new_saved_task(repo)
+    task["schedule"]["enabled"] = True
+    task["schedule"]["run_at"] = "2026-08-28 09:00"
+    repo.save(task)
+    service = TaskApplicationService(repo)
+
+    result = service.move_to_status(task["task_id"], "done")
+
+    assert result.task["status"] == "done"
+    assert result.task["schedule"]["enabled"] is False
+    assert repo.get(task["task_id"])["schedule"]["enabled"] is False
+
+
+def test_move_to_status_scheduled_without_run_at_needs_schedule(tmp_path):
+    repo = TaskRepository(tmp_path)
+    task = _new_saved_task(repo)   # fresh task has schedule.run_at = None
+    service = TaskApplicationService(repo)
+
+    result = service.move_to_status(task["task_id"], "scheduled")
+
+    assert result.needs_schedule is True
+    assert repo.get(task["task_id"])["status"] == "scheduled"
+    assert repo.get(task["task_id"])["schedule"]["enabled"] is False
+
+
+def test_move_to_status_scheduled_with_run_at_enables_it(tmp_path):
+    repo = TaskRepository(tmp_path)
+    task = _new_saved_task(repo)
+    task["schedule"]["run_at"] = "2026-08-28 09:00"
+    repo.save(task)
+    service = TaskApplicationService(repo)
+
+    result = service.move_to_status(task["task_id"], "scheduled")
+
+    assert result.needs_schedule is False
+    assert repo.get(task["task_id"])["schedule"]["enabled"] is True
+
+
+def test_move_to_status_plain_change(tmp_path):
+    repo = TaskRepository(tmp_path)
+    task = _new_saved_task(repo)
+    service = TaskApplicationService(repo)
+
+    result = service.move_to_status(task["task_id"], "backlog")
+
+    assert result.task["status"] == "backlog"
+
+
+def test_move_to_status_missing_task_returns_none(tmp_path):
+    service = _service(tmp_path)
+    assert service.move_to_status("does-not-exist", "backlog") is None

tests/unit/test_task_repository.py

@@ -0,0 +1,70 @@
+"""EPIC R07-T01: TaskRepository + core/tasks.py::save_task atomic write.
+
+The motivating bug: ``core/tasks.py::save_task`` used to
+``path.write_text(json.dumps(...))`` — two syscalls, no atomicity, same class
+of bug already fixed for projects/conversations at R06-T02. A failure between
+the write and the replace must never leave a half-written task JSON file on
+disk; that is the one property the crash-injection test exists to pin.
+"""
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from cowork_local.core.tasks import new_task
+from cowork_local.infrastructure.persistence.json import TaskRepository
+
+
+def test_task_repository_crud_round_trip(tmp_path):
+    repo = TaskRepository(tmp_path)
+    task = repo.create("My Task")
+    repo.save(task)
+
+    assert [t["task_id"] for t in repo.list()] == [task["task_id"]]
+    assert repo.get(task["task_id"])["title"] == "My Task"
+
+    task["title"] = "Renamed"
+    repo.save(task)
+    assert repo.get(task["task_id"])["title"] == "Renamed"
+
+    repo.delete(task["task_id"])
+    assert repo.get(task["task_id"]) is None
+    assert repo.list() == []
+
+
+def test_task_repository_duplicate_keeps_config_resets_identity(tmp_path):
+    repo = TaskRepository(tmp_path)
+    task = repo.create("Original", description="d")
+    repo.save(task)
+
+    dup = repo.duplicate(task)
+    repo.save(dup)
+
+    assert dup["task_id"] != task["task_id"]
+    assert dup["description"] == "d"
+    assert {t["task_id"] for t in repo.list()} == {task["task_id"], dup["task_id"]}
+
+
+def test_save_task_never_corrupts_existing_file_on_crash(tmp_path, monkeypatch):
+    """Same guarantee as ``test_atomic_write_and_repositories.py``'s crash
+    test, exercised through ``core/tasks.py::save_task`` directly (not just
+    through the repository) since that is the function every existing
+    scheduler/executor call site still uses."""
+    from cowork_local.core.tasks import save_task, task_path
+
+    task = new_task("Stable")
+    save_task(task, tmp_path)
+
+    import cowork_local.infrastructure.persistence.json.atomic_write as mod
+
+    def boom(*_a, **_k):
+        raise OSError("simulated crash between write and replace")
+
+    monkeypatch.setattr(mod.os, "replace", boom)
+    task["title"] = "Corrupted?"
+    with pytest.raises(OSError):
+        save_task(task, tmp_path)
+
+    on_disk = json.loads(task_path(task["task_id"], tmp_path).read_text(encoding="utf-8"))
+    assert on_disk["title"] == "Stable"

tests/unit/test_task_scheduler_clock_wiring.py

@@ -0,0 +1,66 @@
+"""EPIC R07-T03: TaskScheduler <-> clock wiring, entirely through
+tests/fakes/fake_clock.py::FakeClock — no real QTimer, no Qt event loop.
+
+Scope note: this only exercises the clock injection seam (start arms+starts
+the clock with `tick`, stop stops it), not the full dispatch/execution
+pipeline (`_start` -> `AgentWorker` -> `execute_task`), which needs a real
+``ctx``/provider and is exactly the kind of Qt-adjacent, thread-heavy path
+better left to an offscreen integration test if/when R08 touches this file
+again — recorded here rather than silently left untested.
+"""
+from __future__ import annotations
+
+from cowork_local.core.task_scheduler import TICK_MS, TaskScheduler
+from tests.fakes import FakeClock
+
+
+def test_start_arms_and_starts_the_injected_clock(tmp_path):
+    clock = FakeClock()
+    scheduler = TaskScheduler(ctx=None, tasks_dir=tmp_path, clock=clock)
+    ticks = []
+    # Instance-attribute override, set BEFORE start(): TaskScheduler.start()
+    # reads `self.tick`, which Python resolves to this override rather than
+    # the class method, so we can count calls without a real due task/ctx.
+    scheduler.tick = lambda: ticks.append(1)
+
+    scheduler.start()
+
+    assert clock.running is True
+    assert clock.interval_ms == TICK_MS
+    assert ticks == [1]   # the catch-up tick() call at startup
+
+
+def test_clock_fire_drives_another_tick(tmp_path):
+    clock = FakeClock()
+    scheduler = TaskScheduler(ctx=None, tasks_dir=tmp_path, clock=clock)
+    ticks = []
+    scheduler.tick = lambda: ticks.append(1)
+    scheduler.start()
+
+    clock.fire()
+
+    assert ticks == [1, 1]
+
+
+def test_stop_stops_the_clock(tmp_path):
+    clock = FakeClock()
+    scheduler = TaskScheduler(ctx=None, tasks_dir=tmp_path, clock=clock)
+    scheduler.tick = lambda: None
+    scheduler.start()
+
+    scheduler.stop()
+
+    assert clock.running is False
+
+
+def test_fire_after_stop_does_not_call_tick(tmp_path):
+    clock = FakeClock()
+    scheduler = TaskScheduler(ctx=None, tasks_dir=tmp_path, clock=clock)
+    ticks = []
+    scheduler.tick = lambda: ticks.append(1)
+    scheduler.start()
+    scheduler.stop()
+
+    clock.fire()
+
+    assert ticks == [1]   # only the startup catch-up tick, nothing after stop