test: 新增 scheduler 限界上下文全量单元测试

新增覆盖以下模块的单元测试：
1. 核心契约与校验函数
2. 持久化层工作单元与ORM映射
3. 运行时处理器注册表
4. 业务用例工具类（退避、抖动、Cron计算）
5. 基础设施容器装配
6. 框架内置清理处理器（运行日志、幂等记录）
7. 测试共享Fixture与桩实现

backend/test/unit/scheduler/adapters/persistence/test_mappers.py

@@ -0,0 +1,286 @@
+"""ORM ↔ dataclass 映射边界单元测试。
+
+覆盖 ``yuxi.scheduler.adapters.persistence.mappers`` 模块：
+- ``orm_to_task``：ORM ScheduledTask → core dataclass
+- ``orm_to_run_log``：ORM ScheduledTaskRunLog → core dataclass
+- ``daily_stat_row_to_dataclass``：仓储层 DailyStat → core dataclass
+- ``handler_summary_row_to_dataclass``：仓储层 HandlerSummary → core dataclass
+- None 入参返回 None
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from types import SimpleNamespace
+
+import pytest
+
+from yuxi.scheduler.adapters.persistence.mappers import (
+    daily_stat_row_to_dataclass,
+    handler_summary_row_to_dataclass,
+    orm_to_run_log,
+    orm_to_task,
+)
+from yuxi.scheduler.core.models import (
+    DailyStat,
+    HandlerSummary,
+    ScheduledTask,
+    ScheduledTaskRunLog,
+)
+
+
+pytestmark = pytest.mark.unit
+
+
+# ─── 测试辅助 ─────────────────────────────────────────────────────────────
+
+
+def _make_task_orm(**overrides) -> SimpleNamespace:
+    """构造类似 ORM ScheduledTask 的 SimpleNamespace。"""
+    defaults = {
+        "id": 1,
+        "task_id": "task-001",
+        "handler_name": "my_handler",
+        "owner_scope": "system",
+        "owner_id": "test:1",
+        "schedule_kind": "cron",
+        "cron_expression": "*/15 * * * *",
+        "run_at": None,
+        "tz": "Asia/Shanghai",
+        "payload": {"k": "v"},
+        "enabled": True,
+        "delete_after_run": False,
+        "block_strategy": "discard_later",
+        "stagger_seconds": 30,
+        "consecutive_errors": 0,
+        "status": "active",
+        "last_run_at": None,
+        "next_run_at": datetime(2024, 1, 1, 0, 15, 0),
+        "last_error": None,
+        "created_by": "system",
+        "updated_by": None,
+        "created_at": datetime(2024, 1, 1, 0, 0, 0),
+        "updated_at": None,
+        "is_deleted": 0,
+        "deleted_at": None,
+    }
+    defaults.update(overrides)
+    return SimpleNamespace(**defaults)
+
+
+def _make_run_log_orm(**overrides) -> SimpleNamespace:
+    """构造类似 ORM ScheduledTaskRunLog 的 SimpleNamespace。"""
+    defaults = {
+        "id": 1,
+        "task_id": "task-001",
+        "run_id": "run-001",
+        "triggered_by": "auto",
+        "status": "success",
+        "error_message": None,
+        "output": {"rows": 10},
+        "started_at": datetime(2024, 1, 1, 0, 15, 0),
+        "finished_at": datetime(2024, 1, 1, 0, 15, 5),
+        "created_by": "system",
+        "updated_by": None,
+        "created_at": datetime(2024, 1, 1, 0, 0, 0),
+        "updated_at": None,
+        "is_deleted": 0,
+        "deleted_at": None,
+    }
+    defaults.update(overrides)
+    return SimpleNamespace(**defaults)
+
+
+# ─── orm_to_task ──────────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestOrmToTask:
+    def test_none_returns_none(self):
+        # Arrange & Act & Assert - 入参为 None 时返回 None
+        assert orm_to_task(None) is None
+
+    def test_basic_mapping(self):
+        # Arrange
+        orm = _make_task_orm()
+        # Act
+        task = orm_to_task(orm)
+        # Assert
+        assert isinstance(task, ScheduledTask)
+        assert task.id == 1
+        assert task.task_id == "task-001"
+        assert task.handler_name == "my_handler"
+        assert task.owner_scope == "system"
+        assert task.schedule_kind == "cron"
+        assert task.cron_expression == "*/15 * * * *"
+        assert task.payload == {"k": "v"}
+        assert task.status == "active"
+        assert task.next_run_at == datetime(2024, 1, 1, 0, 15, 0)
+
+    def test_none_payload_becomes_empty_dict(self):
+        # Arrange - ORM payload 为 None 时 mapper 兜底为空 dict
+        orm = _make_task_orm(payload=None)
+        # Act
+        task = orm_to_task(orm)
+        # Assert
+        assert task.payload == {}
+
+    def test_all_fields_mapped(self):
+        # Arrange - 验证所有字段都被映射
+        orm = _make_task_orm()
+        # Act
+        task = orm_to_task(orm)
+        # Assert
+        assert task.id == orm.id
+        assert task.task_id == orm.task_id
+        assert task.handler_name == orm.handler_name
+        assert task.owner_scope == orm.owner_scope
+        assert task.owner_id == orm.owner_id
+        assert task.schedule_kind == orm.schedule_kind
+        assert task.cron_expression == orm.cron_expression
+        assert task.run_at == orm.run_at
+        assert task.tz == orm.tz
+        assert task.payload == orm.payload
+        assert task.enabled == orm.enabled
+        assert task.delete_after_run == orm.delete_after_run
+        assert task.block_strategy == orm.block_strategy
+        assert task.stagger_seconds == orm.stagger_seconds
+        assert task.consecutive_errors == orm.consecutive_errors
+        assert task.status == orm.status
+        assert task.last_run_at == orm.last_run_at
+        assert task.next_run_at == orm.next_run_at
+        assert task.last_error == orm.last_error
+        assert task.created_by == orm.created_by
+        assert task.updated_by == orm.updated_by
+        assert task.created_at == orm.created_at
+        assert task.updated_at == orm.updated_at
+        assert task.is_deleted == orm.is_deleted
+        assert task.deleted_at == orm.deleted_at
+
+
+# ─── orm_to_run_log ───────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestOrmToRunLog:
+    def test_none_returns_none(self):
+        assert orm_to_run_log(None) is None
+
+    def test_basic_mapping(self):
+        # Arrange
+        orm = _make_run_log_orm()
+        # Act
+        log = orm_to_run_log(orm)
+        # Assert
+        assert isinstance(log, ScheduledTaskRunLog)
+        assert log.id == 1
+        assert log.task_id == "task-001"
+        assert log.run_id == "run-001"
+        assert log.triggered_by == "auto"
+        assert log.status == "success"
+        assert log.output == {"rows": 10}
+        assert log.started_at == datetime(2024, 1, 1, 0, 15, 0)
+
+    def test_all_fields_mapped(self):
+        orm = _make_run_log_orm()
+        log = orm_to_run_log(orm)
+        assert log.id == orm.id
+        assert log.task_id == orm.task_id
+        assert log.run_id == orm.run_id
+        assert log.triggered_by == orm.triggered_by
+        assert log.status == orm.status
+        assert log.error_message == orm.error_message
+        assert log.output == orm.output
+        assert log.started_at == orm.started_at
+        assert log.finished_at == orm.finished_at
+        assert log.created_by == orm.created_by
+        assert log.updated_by == orm.updated_by
+        assert log.created_at == orm.created_at
+        assert log.updated_at == orm.updated_at
+        assert log.is_deleted == orm.is_deleted
+        assert log.deleted_at == orm.deleted_at
+
+
+# ─── daily_stat_row_to_dataclass ──────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestDailyStatRowToDataclass:
+    def test_basic_mapping(self):
+        # Arrange - row 可能是 NamedTuple 或 dataclass 实例
+        row = SimpleNamespace(
+            stat_date=datetime(2024, 1, 1).date(),
+            handler_name="h",
+            success_count=10,
+            failure_count=2,
+            timeout_count=1,
+            dead_letter_count=0,
+        )
+        # Act
+        stat = daily_stat_row_to_dataclass(row)
+        # Assert
+        assert isinstance(stat, DailyStat)
+        assert stat.stat_date == datetime(2024, 1, 1).date()
+        assert stat.handler_name == "h"
+        assert stat.success_count == 10
+        assert stat.failure_count == 2
+        assert stat.timeout_count == 1
+        assert stat.dead_letter_count == 0
+
+    def test_string_counts_converted_to_int(self):
+        # Arrange - 仓储层可能返回字符串计数值
+        row = SimpleNamespace(
+            stat_date=datetime(2024, 1, 1).date(),
+            handler_name="h",
+            success_count="10",
+            failure_count="2",
+            timeout_count="1",
+            dead_letter_count="0",
+        )
+        # Act
+        stat = daily_stat_row_to_dataclass(row)
+        # Assert - int() 转换
+        assert stat.success_count == 10
+        assert stat.failure_count == 2
+        assert stat.timeout_count == 1
+        assert stat.dead_letter_count == 0
+        assert isinstance(stat.success_count, int)
+
+
+# ─── handler_summary_row_to_dataclass ─────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestHandlerSummaryRowToDataclass:
+    def test_basic_mapping(self):
+        # Arrange
+        row = SimpleNamespace(
+            name="my_handler",
+            task_count=5,
+            last_active_at=datetime(2024, 1, 1, 0, 0, 0),
+        )
+        # Act
+        summary = handler_summary_row_to_dataclass(row)
+        # Assert
+        assert isinstance(summary, HandlerSummary)
+        assert summary.name == "my_handler"
+        assert summary.task_count == 5
+        assert summary.last_active_at == datetime(2024, 1, 1, 0, 0, 0)
+
+    def test_string_task_count_converted_to_int(self):
+        # Arrange
+        row = SimpleNamespace(
+            name="h",
+            task_count="7",
+            last_active_at=None,
+        )
+        # Act
+        summary = handler_summary_row_to_dataclass(row)
+        # Assert
+        assert summary.task_count == 7
+        assert isinstance(summary.task_count, int)
+
+    def test_none_last_active_at(self):
+        row = SimpleNamespace(name="h", task_count=0, last_active_at=None)
+        summary = handler_summary_row_to_dataclass(row)
+        assert summary.last_active_at is None

backend/test/unit/scheduler/adapters/persistence/test_repos.py

@@ -0,0 +1,630 @@
+"""SqlAlchemy 仓储适配器单元测试。
+
+覆盖 ``yuxi.scheduler.adapters.persistence`` 下的 4 个仓储适配器：
+- ``SqlAlchemyTaskRepository``：委托 ScheduledTaskRepository
+- ``SqlAlchemyRunLogRepository``：委托 ScheduledTaskRunLogRepository
+- ``SqlAlchemyRunLogDailyRepository``：委托 ScheduledTaskRunLogDailyRepository
+- ``SqlAlchemyIdempotencyRepository``：委托 ScheduledTaskIdempotencyRepository
+
+验证薄委托行为：调用实际仓储方法后通过 mappers 转换为 core dataclass，
+``commit`` 等控制参数透传，不包含业务逻辑。
+"""
+
+from __future__ import annotations
+
+from datetime import date, datetime
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from yuxi.scheduler.adapters.persistence.sqlalchemy_idempotency_repo import (
+    SqlAlchemyIdempotencyRepository,
+)
+from yuxi.scheduler.adapters.persistence.sqlalchemy_run_log_daily_repo import (
+    SqlAlchemyRunLogDailyRepository,
+)
+from yuxi.scheduler.adapters.persistence.sqlalchemy_run_log_repo import (
+    SqlAlchemyRunLogRepository,
+)
+from yuxi.scheduler.adapters.persistence.sqlalchemy_task_repo import (
+    SqlAlchemyTaskRepository,
+)
+from yuxi.scheduler.core.models import (
+    DailyStat,
+    HandlerSummary,
+    ScheduledTask,
+    ScheduledTaskRunLog,
+)
+
+
+pytestmark = pytest.mark.unit
+
+
+# ─── 测试辅助 ─────────────────────────────────────────────────────────────
+
+
+def _make_task_orm(**overrides) -> SimpleNamespace:
+    """构造类似 ORM ScheduledTask 的 SimpleNamespace。"""
+    defaults = {
+        "id": 1,
+        "task_id": "task-001",
+        "handler_name": "my_handler",
+        "owner_scope": "system",
+        "owner_id": "test:1",
+        "schedule_kind": "cron",
+        "cron_expression": "*/15 * * * *",
+        "run_at": None,
+        "tz": "Asia/Shanghai",
+        "payload": {"k": "v"},
+        "enabled": True,
+        "delete_after_run": False,
+        "block_strategy": "discard_later",
+        "stagger_seconds": 30,
+        "consecutive_errors": 0,
+        "status": "active",
+        "last_run_at": None,
+        "next_run_at": datetime(2024, 1, 1, 0, 15, 0),
+        "last_error": None,
+        "created_by": "system",
+        "updated_by": None,
+        "created_at": datetime(2024, 1, 1, 0, 0, 0),
+        "updated_at": None,
+        "is_deleted": 0,
+        "deleted_at": None,
+    }
+    defaults.update(overrides)
+    return SimpleNamespace(**defaults)
+
+
+def _make_run_log_orm(**overrides) -> SimpleNamespace:
+    defaults = {
+        "id": 1,
+        "task_id": "task-001",
+        "run_id": "run-001",
+        "triggered_by": "auto",
+        "status": "success",
+        "error_message": None,
+        "output": {"rows": 10},
+        "started_at": datetime(2024, 1, 1, 0, 15, 0),
+        "finished_at": datetime(2024, 1, 1, 0, 15, 5),
+        "created_by": "system",
+        "updated_by": None,
+        "created_at": datetime(2024, 1, 1, 0, 0, 0),
+        "updated_at": None,
+        "is_deleted": 0,
+        "deleted_at": None,
+    }
+    defaults.update(overrides)
+    return SimpleNamespace(**defaults)
+
+
+# ─── SqlAlchemyTaskRepository ─────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestSqlAlchemyTaskRepository:
+    @pytest.fixture
+    def repo_with_mock(self):
+        """构造 SqlAlchemyTaskRepository，内部 ORM 仓储为 AsyncMock。"""
+        db = MagicMock()
+        with patch(
+            "yuxi.scheduler.adapters.persistence.sqlalchemy_task_repo.ORMScheduledTaskRepository"
+        ) as mock_cls:
+            mock_orm_repo = AsyncMock()
+            mock_cls.return_value = mock_orm_repo
+            repo = SqlAlchemyTaskRepository(db)
+        return repo, mock_orm_repo
+
+    @pytest.mark.asyncio
+    async def test_create_delegates_and_maps(self, repo_with_mock):
+        # Arrange
+        repo, mock_orm = repo_with_mock
+        mock_orm.create = AsyncMock(return_value=_make_task_orm())
+        data = {"task_id": "task-001"}
+        # Act
+        result = await repo.create(data, commit=False)
+        # Assert
+        mock_orm.create.assert_awaited_once_with(data, commit=False)
+        assert isinstance(result, ScheduledTask)
+        assert result.task_id == "task-001"
+
+    @pytest.mark.asyncio
+    async def test_get_by_task_id_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.get_by_task_id = AsyncMock(return_value=_make_task_orm())
+        # Act
+        result = await repo.get_by_task_id("task-001", for_update=True)
+        # Assert
+        mock_orm.get_by_task_id.assert_awaited_once_with("task-001", for_update=True)
+        assert isinstance(result, ScheduledTask)
+
+    @pytest.mark.asyncio
+    async def test_get_by_task_id_returns_none_when_orm_returns_none(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.get_by_task_id = AsyncMock(return_value=None)
+        # Act
+        result = await repo.get_by_task_id("nonexistent")
+        # Assert
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_get_by_id_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.get_by_id = AsyncMock(return_value=_make_task_orm())
+        # Act
+        result = await repo.get_by_id(1, for_update=False)
+        # Assert
+        mock_orm.get_by_id.assert_awaited_once_with(1, for_update=False)
+        assert isinstance(result, ScheduledTask)
+
+    @pytest.mark.asyncio
+    async def test_list_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.list = AsyncMock(return_value=([_make_task_orm(), _make_task_orm()], 2))
+        # Act
+        tasks, total = await repo.list(page=1, page_size=20)
+        # Assert
+        assert len(tasks) == 2
+        assert total == 2
+        assert all(isinstance(t, ScheduledTask) for t in tasks)
+
+    @pytest.mark.asyncio
+    async def test_update_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.update = AsyncMock(return_value=_make_task_orm(status="paused"))
+        # Act
+        result = await repo.update("task-001", {"status": "paused"}, commit=False)
+        # Assert
+        mock_orm.update.assert_awaited_once_with("task-001", {"status": "paused"}, commit=False)
+        assert isinstance(result, ScheduledTask)
+        assert result.status == "paused"
+
+    @pytest.mark.asyncio
+    async def test_delete_delegates(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.delete = AsyncMock(return_value=True)
+        # Act
+        result = await repo.delete("task-001", commit=False)
+        # Assert
+        mock_orm.delete.assert_awaited_once_with("task-001", commit=False)
+        assert result is True
+
+    @pytest.mark.asyncio
+    async def test_list_due_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.list_due = AsyncMock(return_value=[_make_task_orm()])
+        now = datetime(2024, 1, 1, 0, 0, 0)
+        # Act
+        result = await repo.list_due(now, limit=500)
+        # Assert
+        mock_orm.list_due.assert_awaited_once_with(now, limit=500)
+        assert len(result) == 1
+        assert isinstance(result[0], ScheduledTask)
+
+    @pytest.mark.asyncio
+    async def test_list_upcoming_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.list_upcoming = AsyncMock(return_value=[_make_task_orm()])
+        after = datetime(2024, 1, 1, 0, 0, 0)
+        before = datetime(2024, 1, 2, 0, 0, 0)
+        # Act
+        result = await repo.list_upcoming(after=after, before=before, limit=50)
+        # Assert
+        mock_orm.list_upcoming.assert_awaited_once_with(
+            after=after, before=before, limit=50, handler_name=None
+        )
+        assert len(result) == 1
+
+    @pytest.mark.asyncio
+    async def test_list_handler_summary_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        row = SimpleNamespace(name="h", task_count=5, last_active_at=None)
+        mock_orm.list_handler_summary = AsyncMock(return_value=[row])
+        # Act
+        result = await repo.list_handler_summary()
+        # Assert
+        assert len(result) == 1
+        assert isinstance(result[0], HandlerSummary)
+        assert result[0].name == "h"
+
+    @pytest.mark.asyncio
+    async def test_count_by_status_delegates(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.count_by_status = AsyncMock(return_value={"active": 10})
+        # Act
+        result = await repo.count_by_status(owner_scope="system")
+        # Assert
+        mock_orm.count_by_status.assert_awaited_once_with(
+            owner_scope="system", owner_id=None
+        )
+        assert result == {"active": 10}
+
+    @pytest.mark.asyncio
+    async def test_acquire_for_run_delegates(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.acquire_for_run = AsyncMock(return_value=True)
+        now = datetime(2024, 1, 1, 0, 0, 0)
+        next_run = datetime(2024, 1, 1, 0, 15, 0)
+        # Act
+        result = await repo.acquire_for_run(
+            "task-001", run_id="run-001", now=now, next_run_at=next_run
+        )
+        # Assert
+        mock_orm.acquire_for_run.assert_awaited_once_with(
+            "task-001", run_id="run-001", now=now, next_run_at=next_run
+        )
+        assert result is True
+
+    @pytest.mark.asyncio
+    async def test_pause_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.pause = AsyncMock(return_value=_make_task_orm(status="paused"))
+        # Act
+        result = await repo.pause("task-001", commit=False)
+        # Assert
+        mock_orm.pause.assert_awaited_once_with("task-001", commit=False)
+        assert isinstance(result, ScheduledTask)
+        assert result.status == "paused"
+
+    @pytest.mark.asyncio
+    async def test_resume_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.resume = AsyncMock(return_value=_make_task_orm(status="active"))
+        next_run = datetime(2024, 1, 1, 0, 15, 0)
+        # Act
+        result = await repo.resume("task-001", next_run_at=next_run, commit=False)
+        # Assert
+        mock_orm.resume.assert_awaited_once_with(
+            "task-001", next_run_at=next_run, commit=False
+        )
+        assert isinstance(result, ScheduledTask)
+
+    @pytest.mark.asyncio
+    async def test_mark_dead_letter_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.mark_dead_letter = AsyncMock(return_value=_make_task_orm(status="dead_letter"))
+        # Act
+        result = await repo.mark_dead_letter("task-001", commit=False)
+        # Assert
+        mock_orm.mark_dead_letter.assert_awaited_once_with("task-001", commit=False)
+        assert result.status == "dead_letter"
+
+    @pytest.mark.asyncio
+    async def test_reset_from_dead_letter_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.reset_from_dead_letter = AsyncMock(return_value=_make_task_orm(status="active"))
+        next_run = datetime(2024, 1, 1, 0, 15, 0)
+        # Act
+        result = await repo.reset_from_dead_letter("task-001", next_run_at=next_run)
+        # Assert
+        mock_orm.reset_from_dead_letter.assert_awaited_once_with(
+            "task-001", next_run_at=next_run, commit=True
+        )
+
+    @pytest.mark.asyncio
+    async def test_record_run_result_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.record_run_result = AsyncMock(
+            return_value=_make_task_orm(consecutive_errors=1)
+        )
+        # Act
+        result = await repo.record_run_result(
+            "task-001", success=False, error="boom", commit=False
+        )
+        # Assert
+        mock_orm.record_run_result.assert_awaited_once_with(
+            "task-001", success=False, error="boom", backoff_until=None, commit=False
+        )
+        assert result.consecutive_errors == 1
+
+    @pytest.mark.asyncio
+    async def test_get_deleted_by_task_id_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.get_deleted_by_task_id = AsyncMock(return_value=_make_task_orm(is_deleted=1))
+        # Act
+        result = await repo.get_deleted_by_task_id("task-001")
+        # Assert
+        assert isinstance(result, ScheduledTask)
+        assert result.is_deleted == 1
+
+    @pytest.mark.asyncio
+    async def test_list_deleted_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.list_deleted = AsyncMock(return_value=[_make_task_orm(is_deleted=1)])
+        # Act
+        result = await repo.list_deleted(limit=10, offset=0)
+        # Assert
+        assert len(result) == 1
+        assert isinstance(result[0], ScheduledTask)
+
+    @pytest.mark.asyncio
+    async def test_hard_delete_by_id_delegates(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.hard_delete_by_id = AsyncMock(return_value=1)
+        # Act
+        result = await repo.hard_delete_by_id(1)
+        # Assert
+        mock_orm.hard_delete_by_id.assert_awaited_once_with(1, commit=True)
+        assert result == 1
+
+
+# ─── SqlAlchemyRunLogRepository ───────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestSqlAlchemyRunLogRepository:
+    @pytest.fixture
+    def repo_with_mock(self):
+        db = MagicMock()
+        with patch(
+            "yuxi.scheduler.adapters.persistence.sqlalchemy_run_log_repo.ORMScheduledTaskRunLogRepository"
+        ) as mock_cls:
+            mock_orm_repo = AsyncMock()
+            mock_cls.return_value = mock_orm_repo
+            repo = SqlAlchemyRunLogRepository(db)
+        return repo, mock_orm_repo
+
+    @pytest.mark.asyncio
+    async def test_create_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.create = AsyncMock(return_value=_make_run_log_orm())
+        # Act
+        result = await repo.create({"task_id": "t1"}, commit=False)
+        # Assert
+        mock_orm.create.assert_awaited_once_with({"task_id": "t1"}, commit=False)
+        assert isinstance(result, ScheduledTaskRunLog)
+
+    @pytest.mark.asyncio
+    async def test_get_by_run_id_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.get_by_run_id = AsyncMock(return_value=_make_run_log_orm())
+        # Act
+        result = await repo.get_by_run_id("run-001")
+        # Assert
+        mock_orm.get_by_run_id.assert_awaited_once_with("run-001")
+        assert isinstance(result, ScheduledTaskRunLog)
+
+    @pytest.mark.asyncio
+    async def test_get_by_run_id_returns_none_when_not_found(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.get_by_run_id = AsyncMock(return_value=None)
+        # Act
+        result = await repo.get_by_run_id("nonexistent")
+        # Assert
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_update_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.update = AsyncMock(return_value=_make_run_log_orm(status="success"))
+        # Act
+        result = await repo.update("run-001", {"status": "success"}, commit=False)
+        # Assert
+        mock_orm.update.assert_awaited_once_with("run-001", {"status": "success"}, commit=False)
+        assert result.status == "success"
+
+    @pytest.mark.asyncio
+    async def test_list_by_task_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.list_by_task = AsyncMock(return_value=([_make_run_log_orm()], 1))
+        # Act
+        logs, total = await repo.list_by_task("task-001", page=1, page_size=20)
+        # Assert
+        assert len(logs) == 1
+        assert total == 1
+        assert isinstance(logs[0], ScheduledTaskRunLog)
+
+    @pytest.mark.asyncio
+    async def test_list_by_status_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.list_by_status = AsyncMock(return_value=([_make_run_log_orm()], 1))
+        # Act
+        logs, total = await repo.list_by_status(status="running", page=1, page_size=20)
+        # Assert
+        assert len(logs) == 1
+        assert total == 1
+
+    @pytest.mark.asyncio
+    async def test_get_running_by_task_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.get_running_by_task = AsyncMock(return_value=_make_run_log_orm(status="running"))
+        # Act
+        result = await repo.get_running_by_task("task-001")
+        # Assert
+        assert isinstance(result, ScheduledTaskRunLog)
+        assert result.status == "running"
+
+    @pytest.mark.asyncio
+    async def test_count_running_delegates(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.count_running = AsyncMock(return_value=3)
+        # Act
+        result = await repo.count_running()
+        # Assert
+        assert result == 3
+
+    @pytest.mark.asyncio
+    async def test_reclaim_stale_runs_delegates(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.reclaim_stale_runs = AsyncMock(return_value=5)
+        stale_before = datetime(2024, 1, 1, 0, 0, 0)
+        # Act
+        result = await repo.reclaim_stale_runs(stale_before=stale_before)
+        # Assert
+        mock_orm.reclaim_stale_runs.assert_awaited_once_with(
+            stale_before=stale_before,
+            error_message="reclaimed by scheduler restart",
+            commit=True,
+        )
+        assert result == 5
+
+    @pytest.mark.asyncio
+    async def test_cleanup_old_logs_delegates(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.cleanup_old_logs = AsyncMock(return_value=10)
+        before = datetime(2024, 1, 1, 0, 0, 0)
+        # Act
+        result = await repo.cleanup_old_logs(before, commit=False)
+        # Assert
+        mock_orm.cleanup_old_logs.assert_awaited_once_with(before, commit=False)
+        assert result == 10
+
+
+# ─── SqlAlchemyRunLogDailyRepository ──────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestSqlAlchemyRunLogDailyRepository:
+    @pytest.fixture
+    def repo_with_mock(self):
+        db = MagicMock()
+        with patch(
+            "yuxi.scheduler.adapters.persistence.sqlalchemy_run_log_daily_repo.ORMScheduledTaskRunLogDailyRepository"
+        ) as mock_cls:
+            mock_orm_repo = AsyncMock()
+            mock_cls.return_value = mock_orm_repo
+            repo = SqlAlchemyRunLogDailyRepository(db)
+        return repo, mock_orm_repo
+
+    @pytest.mark.asyncio
+    async def test_upsert_daily_stat_delegates(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.upsert_daily_stat = AsyncMock(return_value=None)
+        stat_date = date(2024, 1, 1)
+        # Act
+        await repo.upsert_daily_stat(stat_date, "h", "success", commit=False)
+        # Assert
+        mock_orm.upsert_daily_stat.assert_awaited_once_with(
+            stat_date, "h", "success", commit=False
+        )
+
+    @pytest.mark.asyncio
+    async def test_list_daily_stat_delegates_and_maps(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        row = SimpleNamespace(
+            stat_date=date(2024, 1, 1),
+            handler_name="h",
+            success_count=10,
+            failure_count=2,
+            timeout_count=1,
+            dead_letter_count=0,
+        )
+        mock_orm.list_daily_stat = AsyncMock(return_value=[row])
+        # Act
+        result = await repo.list_daily_stat(
+            start_date=date(2024, 1, 1), end_date=date(2024, 1, 31)
+        )
+        # Assert
+        assert len(result) == 1
+        assert isinstance(result[0], DailyStat)
+        assert result[0].success_count == 10
+
+    @pytest.mark.asyncio
+    async def test_cleanup_old_daily_stats_delegates(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.cleanup_old_daily_stats = AsyncMock(return_value=5)
+        before = date(2023, 1, 1)
+        # Act
+        result = await repo.cleanup_old_daily_stats(before, commit=False)
+        # Assert
+        mock_orm.cleanup_old_daily_stats.assert_awaited_once_with(before, commit=False)
+        assert result == 5
+
+    @pytest.mark.asyncio
+    async def test_get_deleted_by_id_passes_through_any(self, repo_with_mock):
+        # Arrange - core 层无对应 dataclass，按端口契约以 Any 透传 ORM
+        repo, mock_orm = repo_with_mock
+        orm_obj = SimpleNamespace(id=1)
+        mock_orm.get_deleted_by_id = AsyncMock(return_value=orm_obj)
+        # Act
+        result = await repo.get_deleted_by_id(1)
+        # Assert
+        assert result is orm_obj
+
+    @pytest.mark.asyncio
+    async def test_list_deleted_passes_through_any(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        orm_list = [SimpleNamespace(id=1), SimpleNamespace(id=2)]
+        mock_orm.list_deleted = AsyncMock(return_value=orm_list)
+        # Act
+        result = await repo.list_deleted(limit=10)
+        # Assert
+        assert result == orm_list
+
+
+# ─── SqlAlchemyIdempotencyRepository ──────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestSqlAlchemyIdempotencyRepository:
+    @pytest.fixture
+    def repo_with_mock(self):
+        db = MagicMock()
+        with patch(
+            "yuxi.scheduler.adapters.persistence.sqlalchemy_idempotency_repo.ORMScheduledTaskIdempotencyRepository"
+        ) as mock_cls:
+            mock_orm_repo = AsyncMock()
+            mock_cls.return_value = mock_orm_repo
+            repo = SqlAlchemyIdempotencyRepository(db)
+        return repo, mock_orm_repo
+
+    @pytest.mark.asyncio
+    async def test_acquire_delegates(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.acquire = AsyncMock(return_value=True)
+        # Act
+        result = await repo.acquire("key-001", task_id="t1", operation="run")
+        # Assert
+        mock_orm.acquire.assert_awaited_once_with("key-001", task_id="t1", operation="run")
+        assert result is True
+
+    @pytest.mark.asyncio
+    async def test_update_response_delegates(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.update_response = AsyncMock(return_value=None)
+        # Act
+        await repo.update_response("key-001", {"result": "ok"})
+        # Assert
+        mock_orm.update_response.assert_awaited_once_with("key-001", {"result": "ok"})
+
+    @pytest.mark.asyncio
+    async def test_get_delegates(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.get = AsyncMock(return_value={"result": "ok"})
+        # Act
+        result = await repo.get("key-001")
+        # Assert
+        mock_orm.get.assert_awaited_once_with("key-001")
+        assert result == {"result": "ok"}
+
+    @pytest.mark.asyncio
+    async def test_get_returns_none_when_not_found(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.get = AsyncMock(return_value=None)
+        # Act
+        result = await repo.get("nonexistent")
+        # Assert
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_cleanup_old_delegates(self, repo_with_mock):
+        repo, mock_orm = repo_with_mock
+        mock_orm.cleanup_old = AsyncMock(return_value=20)
+        before = datetime(2024, 1, 1, 0, 0, 0)
+        # Act
+        result = await repo.cleanup_old(before, commit=False)
+        # Assert
+        mock_orm.cleanup_old.assert_awaited_once_with(before, commit=False)
+        assert result == 20
+
+    @pytest.mark.asyncio
+    async def test_get_deleted_by_id_passes_through_any(self, repo_with_mock):
+        # Arrange - core 层无对应 dataclass，按端口契约以 Any 透传 ORM
+        repo, mock_orm = repo_with_mock
+        orm_obj = SimpleNamespace(id=1)
+        mock_orm.get_deleted_by_id = AsyncMock(return_value=orm_obj)
+        # Act
+        result = await repo.get_deleted_by_id(1)
+        # Assert
+        assert result is orm_obj

backend/test/unit/scheduler/adapters/persistence/test_unit_of_work.py

@@ -0,0 +1,113 @@
+"""SqlAlchemyUnitOfWork 单元测试。
+
+覆盖 ``yuxi.scheduler.adapters.persistence.unit_of_work`` 模块的
+``SqlAlchemyUnitOfWork``：
+- ``__aenter__`` 返回 self
+- ``__aexit__`` 正常退出调用 commit，异常退出调用 rollback
+- ``commit`` / ``rollback`` 委托给 db session
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from yuxi.scheduler.adapters.persistence.unit_of_work import SqlAlchemyUnitOfWork
+
+
+pytestmark = pytest.mark.unit
+
+
+@pytest.fixture
+def db_session() -> MagicMock:
+    """提供 MagicMock 模拟 AsyncSession。"""
+    session = MagicMock()
+    session.commit = AsyncMock()
+    session.rollback = AsyncMock()
+    return session
+
+
+# ─── __aenter__ / __aexit__ ───────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestSqlAlchemyUnitOfWorkContext:
+    @pytest.mark.asyncio
+    async def test_aenter_returns_self(self, db_session: MagicMock):
+        # Arrange
+        uow = SqlAlchemyUnitOfWork(db_session)
+        # Act
+        async with uow as ctx:
+            # Assert - __aenter__ 返回 self
+            assert ctx is uow
+
+    @pytest.mark.asyncio
+    async def test_aexit_without_exception_commits(self, db_session: MagicMock):
+        # Arrange
+        uow = SqlAlchemyUnitOfWork(db_session)
+        # Act
+        async with uow:
+            pass
+        # Assert - 正常退出调用 commit
+        db_session.commit.assert_awaited_once()
+        db_session.rollback.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_aexit_with_exception_rolls_back(self, db_session: MagicMock):
+        # Arrange
+        uow = SqlAlchemyUnitOfWork(db_session)
+        # Act & Assert - 异常退出调用 rollback
+        with pytest.raises(RuntimeError, match="boom"):
+            async with uow:
+                raise RuntimeError("boom")
+        db_session.rollback.assert_awaited_once()
+        db_session.commit.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_aexit_with_keyboard_interrupt_rolls_back(self, db_session: MagicMock):
+        # Arrange - 任意异常类型都应触发 rollback
+        uow = SqlAlchemyUnitOfWork(db_session)
+        # Act & Assert
+        with pytest.raises(KeyboardInterrupt):
+            async with uow:
+                raise KeyboardInterrupt()
+        db_session.rollback.assert_awaited_once()
+
+
+# ─── commit / rollback ────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestSqlAlchemyUnitOfWorkCommitRollback:
+    @pytest.mark.asyncio
+    async def test_commit_delegates_to_session(self, db_session: MagicMock):
+        # Arrange
+        uow = SqlAlchemyUnitOfWork(db_session)
+        # Act
+        await uow.commit()
+        # Assert
+        db_session.commit.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_rollback_delegates_to_session(self, db_session: MagicMock):
+        # Arrange
+        uow = SqlAlchemyUnitOfWork(db_session)
+        # Act
+        await uow.rollback()
+        # Assert
+        db_session.rollback.assert_awaited_once()
+
+
+# ─── 满足 UnitOfWork Protocol ─────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestSqlAlchemyUnitOfWorkProtocol:
+    def test_satisfies_unit_of_work_protocol(self, db_session: MagicMock):
+        # Arrange & Act
+        from yuxi.scheduler.core.contracts import UnitOfWork
+
+        uow = SqlAlchemyUnitOfWork(db_session)
+        # Assert - 结构化实现 UnitOfWork Protocol，runtime_checkable 允许 isinstance
+        assert isinstance(uow, UnitOfWork)

backend/test/unit/scheduler/conftest.py

@@ -0,0 +1,232 @@
+"""scheduler 限界上下文单元测试共享 fixture。
+
+本文件仅放置跨多个测试文件复用的 fixture 与工厂函数。单文件独有的 helper
+应保留在对应测试文件内，避免过度集中（见 testing-guidelines.md）。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from yuxi.scheduler.core.models import (
+    DailyStat,
+    HandlerSummary,
+    ScheduledTask,
+    ScheduledTaskRunLog,
+)
+
+
+# ─── 工厂函数 ─────────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def make_task() -> type[ScheduledTask]:
+    """返回 ScheduledTask 工厂类型，调用时按需覆盖字段。
+
+    用法::
+
+        task = make_task(task_id="abc", status="paused")
+    """
+
+    def _build(**overrides: Any) -> ScheduledTask:
+        defaults: dict[str, Any] = {
+            "task_id": "task-001",
+            "handler_name": "test_handler",
+            "owner_scope": "system",
+            "owner_id": "test:owner-1",
+            "schedule_kind": "cron",
+            "cron_expression": "*/15 * * * *",
+            "tz": "Asia/Shanghai",
+            "payload": {"key": "value"},
+            "enabled": True,
+            "delete_after_run": False,
+            "block_strategy": "discard_later",
+            "stagger_seconds": 30,
+            "consecutive_errors": 0,
+            "status": "active",
+            "next_run_at": datetime(2024, 1, 1, 0, 15, 0),
+            "id": 1,
+        }
+        defaults.update(overrides)
+        return ScheduledTask(**defaults)
+
+    return _build
+
+
+@pytest.fixture
+def make_run_log() -> type[ScheduledTaskRunLog]:
+    """返回 ScheduledTaskRunLog 工厂类型。"""
+
+    def _build(**overrides: Any) -> ScheduledTaskRunLog:
+        defaults: dict[str, Any] = {
+            "task_id": "task-001",
+            "run_id": "run-001",
+            "triggered_by": "auto",
+            "status": "running",
+            "started_at": datetime(2024, 1, 1, 0, 15, 0),
+            "id": 1,
+        }
+        defaults.update(overrides)
+        return ScheduledTaskRunLog(**defaults)
+
+    return _build
+
+
+@pytest.fixture
+def make_handler_summary() -> type[HandlerSummary]:
+    """返回 HandlerSummary 工厂类型。"""
+
+    def _build(**overrides: Any) -> HandlerSummary:
+        defaults: dict[str, Any] = {
+            "name": "test_handler",
+            "task_count": 5,
+            "last_active_at": datetime(2024, 1, 1, 0, 15, 0),
+        }
+        defaults.update(overrides)
+        return HandlerSummary(**defaults)
+
+    return _build
+
+
+@pytest.fixture
+def make_daily_stat() -> type[DailyStat]:
+    """返回 DailyStat 工厂类型。"""
+
+    def _build(**overrides: Any) -> DailyStat:
+        defaults: dict[str, Any] = {
+            "stat_date": datetime(2024, 1, 1).date(),
+            "handler_name": "test_handler",
+            "success_count": 10,
+            "failure_count": 2,
+            "timeout_count": 1,
+            "dead_letter_count": 0,
+        }
+        defaults.update(overrides)
+        return DailyStat(**defaults)
+
+    return _build
+
+
+# ─── Stub 端口实现 ─────────────────────────────────────────────────────────
+
+
+class FakeRepositories:
+    """内存版 Repositories 桩，所有方法均为 AsyncMock。
+
+    通过 ``task`` / ``run_log`` / ``run_log_daily`` / ``idempotency`` 属性
+    访问各仓储桩，可按需配置 ``return_value`` / ``side_effect``。
+    """
+
+    def __init__(self) -> None:
+        self.task = AsyncMock()
+        self.run_log = AsyncMock()
+        self.run_log_daily = AsyncMock()
+        self.idempotency = AsyncMock()
+
+
+class FakeRuntimeServices:
+    """内存版 RuntimeServices 桩。
+
+    - ``handler_registry``：``MagicMock`` 或 None（模拟 api 进程）
+    - ``arq_pool``：``AsyncMock`` 或 None（模拟 api 进程）
+    - ``config``：``MagicMock``，按需配置 ``scheduler_*`` 字段
+    """
+
+    def __init__(self) -> None:
+        self.handler_registry = MagicMock()
+        self.arq_pool = AsyncMock()
+        # 默认配置与 app.py 默认值对齐
+        self.config = MagicMock()
+        self.config.scheduler_max_payload_size_kb = 64
+        self.config.scheduler_min_cron_interval_minutes = 10
+        self.config.scheduler_stagger_max_seconds = 300
+        self.config.scheduler_backoff_schedule = [10, 30, 120, 600, 3600]
+        self.config.scheduler_max_consecutive_errors = 5
+        self.config.scheduler_task_timeout_seconds = 600
+        self.config.scheduler_run_log_retention_days = 90
+        self.config.scheduler_idempotency_retention_hours = 24
+
+
+class FakeUnitOfWork:
+    """内存版 UnitOfWork 桩，记录 commit / rollback 调用。
+
+    ``async with`` 进入时返回 self，正常退出调用 ``commit``，异常退出调用
+    ``rollback``，与 ``SqlAlchemyUnitOfWork`` 行为一致。
+    """
+
+    def __init__(self) -> None:
+        self.committed = False
+        self.rolled_back = False
+        self.enter_count = 0
+        self.exit_count = 0
+
+    async def __aenter__(self) -> FakeUnitOfWork:
+        self.enter_count += 1
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
+        self.exit_count += 1
+        if exc_type is not None:
+            self.rolled_back = True
+        else:
+            self.committed = True
+
+    async def commit(self) -> None:
+        self.committed = True
+
+    async def rollback(self) -> None:
+        self.rolled_back = True
+
+
+@pytest.fixture
+def fake_repos() -> FakeRepositories:
+    """提供内存版 Repositories 桩。"""
+    return FakeRepositories()
+
+
+@pytest.fixture
+def fake_runtime() -> FakeRuntimeServices:
+    """提供内存版 RuntimeServices 桩。"""
+    return FakeRuntimeServices()
+
+
+@pytest.fixture
+def fake_uow() -> FakeUnitOfWork:
+    """提供内存版 UnitOfWork 桩。"""
+    return FakeUnitOfWork()
+
+
+@pytest.fixture
+def fixed_now(monkeypatch: pytest.MonkeyPatch) -> datetime:
+    """固定 UTC naive 时间，用于时间相关测试的稳定基准。
+
+    同时 monkeypatch ``scheduler_service`` 模块内的 ``utc_now_naive``，
+    使被测服务消费到同一固定时间。
+    """
+    fixed = datetime(2024, 6, 1, 0, 0, 0)
+    import yuxi.scheduler.use_cases.services.scheduler_service as svc_mod
+
+    monkeypatch.setattr(svc_mod, "utc_now_naive", lambda: fixed)
+    return fixed
+
+
+@pytest.fixture
+def fixed_uuid(monkeypatch: pytest.MonkeyPatch) -> str:
+    """固定 ``uuid.uuid4().hex`` 返回值，保证 task_id / run_id 可重复。
+
+    被 ``scheduler_service`` 模块内的 ``uuid.uuid4`` 调用消费。
+    """
+    fixed_hex = "fixed-uuid-hex-001"
+
+    class _FakeUUID:
+        hex = fixed_hex
+
+    monkeypatch.setattr(
+        "yuxi.scheduler.use_cases.services.scheduler_service.uuid.uuid4",
+        lambda: _FakeUUID(),
+    )
+    return fixed_hex

backend/test/unit/scheduler/core/test_contracts.py

@@ -0,0 +1,189 @@
+"""core 层跨层端口契约单元测试。
+
+覆盖 ``yuxi.scheduler.core.contracts`` 模块：
+- ``TaskContext``：handler 执行上下文 dataclass
+- ``TaskResult``：handler 执行结果 dataclass
+- ``TaskHandler``：业务域回调 Protocol（runtime_checkable）
+- ``UnitOfWork``：工作单元 Protocol（runtime_checkable）
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from yuxi.scheduler.core.contracts import (
+    TaskContext,
+    TaskHandler,
+    TaskResult,
+    UnitOfWork,
+)
+
+
+pytestmark = pytest.mark.unit
+
+
+# ─── TaskContext ──────────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestTaskContext:
+    def test_required_fields_must_be_provided(self):
+        with pytest.raises(TypeError):
+            TaskContext()  # type: ignore[call-arg]
+
+    def test_fields_assigned(self):
+        # Arrange & Act
+        ctx = TaskContext(
+            task_id="task-001",
+            run_id="run-001",
+            handler_name="my_handler",
+            payload={"key": "value"},
+            triggered_by="auto",
+            scheduled_at="2024-01-01T00:00:00",
+        )
+        # Assert
+        assert ctx.task_id == "task-001"
+        assert ctx.run_id == "run-001"
+        assert ctx.handler_name == "my_handler"
+        assert ctx.payload == {"key": "value"}
+        assert ctx.triggered_by == "auto"
+        assert ctx.scheduled_at == "2024-01-01T00:00:00"
+
+    def test_payload_accepts_empty_dict(self):
+        ctx = TaskContext(
+            task_id="t",
+            run_id="r",
+            handler_name="h",
+            payload={},
+            triggered_by="auto",
+            scheduled_at=None,
+        )
+        assert ctx.payload == {}
+
+
+# ─── TaskResult ───────────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestTaskResult:
+    def test_success_required(self):
+        with pytest.raises(TypeError):
+            TaskResult()  # type: ignore[call-arg]
+
+    def test_success_only_defaults(self):
+        # Arrange & Act
+        result = TaskResult(success=True)
+        # Assert
+        assert result.success is True
+        assert result.error is None
+        assert result.output == {}
+
+    def test_failure_with_error(self):
+        # Arrange & Act
+        result = TaskResult(success=False, error="boom")
+        # Assert
+        assert result.success is False
+        assert result.error == "boom"
+        assert result.output == {}
+
+    def test_output_default_is_isolated_per_instance(self):
+        # Arrange - 可变默认值必须 per-instance 隔离
+        r1 = TaskResult(success=True)
+        r2 = TaskResult(success=True)
+        # Act
+        r1.output["key"] = "value"
+        # Assert
+        assert r2.output == {}
+
+    def test_output_with_data(self):
+        result = TaskResult(success=True, output={"rows": 10})
+        assert result.output == {"rows": 10}
+
+
+# ─── TaskHandler Protocol ─────────────────────────────────────────────────
+
+
+class _ValidHandler:
+    """符合 TaskHandler Protocol 的测试 handler。"""
+
+    name = "valid_handler"
+    description = "A test handler"
+
+    async def execute(self, ctx: TaskContext) -> TaskResult:
+        return TaskResult(success=True)
+
+
+class _MissingNameHandler:
+    """缺少 name 属性，不符合 TaskHandler Protocol。"""
+
+    description = "no name"
+
+    async def execute(self, ctx: TaskContext) -> TaskResult:
+        return TaskResult(success=True)
+
+
+@pytest.mark.unit
+class TestTaskHandlerProtocol:
+    def test_valid_handler_satisfies_protocol(self):
+        # Arrange & Act & Assert - runtime_checkable 允许 isinstance 检查
+        handler = _ValidHandler()
+        assert isinstance(handler, TaskHandler)
+
+    def test_handler_without_name_does_not_satisfy_protocol(self):
+        # Arrange & Act & Assert - 缺少 name 属性不满足 Protocol
+        handler = _MissingNameHandler()
+        assert not isinstance(handler, TaskHandler)
+
+
+# ─── UnitOfWork Protocol ──────────────────────────────────────────────────
+
+
+class _ValidUnitOfWork:
+    """符合 UnitOfWork Protocol 的测试实现。"""
+
+    async def __aenter__(self) -> _ValidUnitOfWork:
+        return self
+
+    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        pass
+
+    async def commit(self) -> None:
+        pass
+
+    async def rollback(self) -> None:
+        pass
+
+
+class _MissingCommitUnitOfWork:
+    """缺少 commit 方法，不符合 UnitOfWork Protocol。"""
+
+    async def __aenter__(self) -> _MissingCommitUnitOfWork:
+        return self
+
+    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        pass
+
+    async def rollback(self) -> None:
+        pass
+
+
+@pytest.mark.unit
+class TestUnitOfWorkProtocol:
+    def test_valid_uow_satisfies_protocol(self):
+        uow = _ValidUnitOfWork()
+        assert isinstance(uow, UnitOfWork)
+
+    def test_uow_without_commit_does_not_satisfy_protocol(self):
+        uow = _MissingCommitUnitOfWork()
+        assert not isinstance(uow, UnitOfWork)
+
+    @pytest.mark.asyncio
+    async def test_uow_async_context_manager_protocol(self):
+        # Arrange
+        uow = _ValidUnitOfWork()
+        # Act
+        async with uow as ctx:
+            assert ctx is uow
+        # Assert - 无异常即通过

backend/test/unit/scheduler/core/test_exceptions.py

@@ -0,0 +1,310 @@
+"""scheduler 异常层次单元测试。
+
+覆盖 ``yuxi.scheduler.exceptions`` 模块的全部异常类，验证：
+- 异常继承关系（``SchedulerError`` 根异常 + 分层子异常）
+- ``status_code`` 类级属性默认值与覆盖
+- ``message`` / ``details`` 实例属性
+- ``__str__`` 返回 ``message``
+- ``SchedulerValidationError`` 同时继承 ``ValueError``
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from yuxi.scheduler.exceptions import (
+    DuplicateHandlerError,
+    HandlerNotFoundError,
+    InvalidCronExpressionError,
+    SchedulerConflictError,
+    SchedulerEntityNotFoundError,
+    SchedulerError,
+    SchedulerPayloadTooLargeError,
+    SchedulerPermissionDeniedError,
+    SchedulerValidationError,
+    TaskStatusTransitionError,
+)
+
+
+pytestmark = pytest.mark.unit
+
+
+# ─── SchedulerError 根异常 ────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestSchedulerError:
+    def test_default_status_code_is_500(self):
+        # Arrange & Act
+        err = SchedulerError("boom")
+        # Assert
+        assert err.status_code == 500
+
+    def test_message_attribute(self):
+        # Arrange & Act
+        err = SchedulerError("boom")
+        # Assert
+        assert err.message == "boom"
+
+    def test_details_defaults_to_empty_dict(self):
+        # Arrange & Act
+        err = SchedulerError("boom")
+        # Assert
+        assert err.details == {}
+
+    def test_details_passed_through(self):
+        # Arrange
+        details = {"task_id": "abc", "reason": "test"}
+        # Act
+        err = SchedulerError("boom", details=details)
+        # Assert
+        assert err.details == details
+
+    def test_str_returns_message(self):
+        # Arrange
+        err = SchedulerError("boom")
+        # Act & Assert
+        assert str(err) == "boom"
+
+    def test_is_exception_subclass(self):
+        # Arrange & Act & Assert
+        assert issubclass(SchedulerError, Exception)
+
+
+# ─── SchedulerValidationError ─────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestSchedulerValidationError:
+    def test_status_code_is_400(self):
+        err = SchedulerValidationError("invalid")
+        assert err.status_code == 400
+
+    def test_inherits_value_error(self):
+        # Arrange & Act & Assert - 兼容 Pydantic field_validator 习惯
+        assert issubclass(SchedulerValidationError, ValueError)
+
+    def test_inherits_scheduler_error(self):
+        assert issubclass(SchedulerValidationError, SchedulerError)
+
+    def test_can_be_raised_as_value_error(self):
+        # Arrange & Act & Assert - 在 dataclass 薄校验中可直接 raise
+        with pytest.raises(ValueError):
+            raise SchedulerValidationError("invalid")
+
+
+# ─── InvalidCronExpressionError ───────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestInvalidCronExpressionError:
+    def test_status_code_is_400(self):
+        err = InvalidCronExpressionError("bad cron")
+        assert err.status_code == 400
+
+    def test_inherits_scheduler_validation_error(self):
+        assert issubclass(InvalidCronExpressionError, SchedulerValidationError)
+
+    def test_details_carry_expression(self):
+        err = InvalidCronExpressionError(
+            "bad cron",
+            details={"expression": "* * * * *", "reason": "too frequent"},
+        )
+        assert err.details["expression"] == "* * * * *"
+
+
+# ─── SchedulerPayloadTooLargeError ────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestSchedulerPayloadTooLargeError:
+    def test_status_code_is_413(self):
+        err = SchedulerPayloadTooLargeError("too big")
+        assert err.status_code == 413
+
+    def test_inherits_scheduler_validation_error(self):
+        assert issubclass(SchedulerPayloadTooLargeError, SchedulerValidationError)
+
+    def test_details_carry_size_and_limit(self):
+        err = SchedulerPayloadTooLargeError(
+            "too big",
+            details={"size": 102400, "limit": 65536, "unit": "bytes"},
+        )
+        assert err.details["size"] == 102400
+        assert err.details["limit"] == 65536
+
+
+# ─── HandlerNotFoundError ─────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestHandlerNotFoundError:
+    def test_status_code_is_400(self):
+        # Arrange & Act - 语义上属于"调度配置校验失败"，按 spec 归入 ValidationError
+        err = HandlerNotFoundError("not found")
+        # Assert
+        assert err.status_code == 400
+
+    def test_inherits_scheduler_validation_error(self):
+        assert issubclass(HandlerNotFoundError, SchedulerValidationError)
+
+    def test_details_carry_handler_key(self):
+        err = HandlerNotFoundError(
+            "not found",
+            details={"handler_key": "my_handler"},
+        )
+        assert err.details["handler_key"] == "my_handler"
+
+
+# ─── SchedulerEntityNotFoundError ─────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestSchedulerEntityNotFoundError:
+    def test_status_code_is_404(self):
+        err = SchedulerEntityNotFoundError("not exist")
+        assert err.status_code == 404
+
+    def test_inherits_scheduler_error(self):
+        assert issubclass(SchedulerEntityNotFoundError, SchedulerError)
+
+    def test_details_carry_entity_type(self):
+        err = SchedulerEntityNotFoundError(
+            "not exist",
+            details={"entity_type": "task", "identifier": "task-001"},
+        )
+        assert err.details["entity_type"] == "task"
+
+
+# ─── SchedulerPermissionDeniedError ──────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestSchedulerPermissionDeniedError:
+    def test_status_code_is_403(self):
+        err = SchedulerPermissionDeniedError("forbidden")
+        assert err.status_code == 403
+
+    def test_inherits_scheduler_error(self):
+        assert issubclass(SchedulerPermissionDeniedError, SchedulerError)
+
+
+# ─── SchedulerConflictError ───────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestSchedulerConflictError:
+    def test_status_code_is_409(self):
+        err = SchedulerConflictError("conflict")
+        assert err.status_code == 409
+
+    def test_inherits_scheduler_error(self):
+        assert issubclass(SchedulerConflictError, SchedulerError)
+
+
+# ─── DuplicateHandlerError ────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestDuplicateHandlerError:
+    def test_status_code_is_409(self):
+        err = DuplicateHandlerError("dup")
+        assert err.status_code == 409
+
+    def test_inherits_scheduler_conflict_error(self):
+        assert issubclass(DuplicateHandlerError, SchedulerConflictError)
+
+    def test_details_carry_handler_key(self):
+        err = DuplicateHandlerError(
+            "dup",
+            details={"handler_key": "my_handler", "existing_class": "MyHandler"},
+        )
+        assert err.details["handler_key"] == "my_handler"
+
+
+# ─── TaskStatusTransitionError ────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestTaskStatusTransitionError:
+    def test_status_code_is_409(self):
+        err = TaskStatusTransitionError("bad transition")
+        assert err.status_code == 409
+
+    def test_inherits_scheduler_conflict_error(self):
+        assert issubclass(TaskStatusTransitionError, SchedulerConflictError)
+
+    def test_details_carry_state_info(self):
+        err = TaskStatusTransitionError(
+            "bad transition",
+            details={
+                "from_state": "succeeded",
+                "to_state": "running",
+                "allowed_transitions": [],
+            },
+        )
+        assert err.details["from_state"] == "succeeded"
+        assert err.details["to_state"] == "running"
+
+
+# ─── 异常层次关系 ─────────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestExceptionHierarchy:
+    def test_all_exceptions_inherit_scheduler_error(self):
+        # Arrange - 所有子异常都应继承根异常，最外层统一捕获
+        subclasses = [
+            SchedulerValidationError,
+            InvalidCronExpressionError,
+            SchedulerPayloadTooLargeError,
+            HandlerNotFoundError,
+            SchedulerEntityNotFoundError,
+            SchedulerPermissionDeniedError,
+            SchedulerConflictError,
+            DuplicateHandlerError,
+            TaskStatusTransitionError,
+        ]
+        # Act & Assert
+        for cls in subclasses:
+            assert issubclass(cls, SchedulerError), f"{cls.__name__} 未继承 SchedulerError"
+
+    def test_validation_error_subclasses_inherit_value_error(self):
+        # Arrange - SchedulerValidationError 同时继承 ValueError，其子类也应继承
+        subclasses = [
+            InvalidCronExpressionError,
+            SchedulerPayloadTooLargeError,
+            HandlerNotFoundError,
+        ]
+        # Act & Assert
+        for cls in subclasses:
+            assert issubclass(cls, ValueError), f"{cls.__name__} 未继承 ValueError"
+
+    def test_conflict_error_subclasses_inherit_scheduler_conflict_error(self):
+        # Arrange
+        subclasses = [DuplicateHandlerError, TaskStatusTransitionError]
+        # Act & Assert
+        for cls in subclasses:
+            assert issubclass(cls, SchedulerConflictError)
+
+    def test_exception_preserves_cause_chain(self):
+        # Arrange - 异常链（__cause__）应被保留，便于调试
+        original = ValueError("original cause")
+        # Act
+        try:
+            try:
+                raise original
+            except ValueError as e:
+                raise SchedulerValidationError("wrapped") from e
+        except SchedulerValidationError as wrapped:
+            # Assert
+            assert wrapped.__cause__ is original
+
+    def test_exception_details_not_shared_across_instances(self):
+        # Arrange - details 字典不应在实例间共享（避免可变默认值陷阱）
+        err1 = SchedulerError("a", details={"key": "v1"})
+        err2 = SchedulerError("b", details={"key": "v2"})
+        # Act & Assert
+        assert err1.details is not err2.details
+        assert err1.details["key"] == "v1"
+        assert err2.details["key"] == "v2"

backend/test/unit/scheduler/core/test_models.py

@@ -0,0 +1,306 @@
+"""core 层领域数据类单元测试。
+
+覆盖 ``yuxi.scheduler.core.models`` 模块的全部 dataclass：
+- ``ScheduledTask``：定时任务领域模型
+- ``ScheduledTaskRunLog``：执行明细日志领域模型
+- ``HandlerSummary``：handler 聚合摘要
+- ``DailyStat``：日聚合统计
+
+验证字段默认值、可变默认值隔离、字段类型与 ORM 对齐。
+"""
+
+from __future__ import annotations
+
+from dataclasses import fields
+
+import pytest
+
+from yuxi.scheduler.core.models import (
+    DailyStat,
+    HandlerSummary,
+    ScheduledTask,
+    ScheduledTaskRunLog,
+)
+
+
+pytestmark = pytest.mark.unit
+
+
+# ─── ScheduledTask ────────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestScheduledTask:
+    def test_required_fields_must_be_provided(self):
+        # Arrange & Act & Assert - task_id / handler_name / owner_scope / owner_id / schedule_kind 必填
+        with pytest.raises(TypeError):
+            ScheduledTask()  # type: ignore[call-arg]
+
+    def test_required_fields_assigned(self):
+        # Arrange & Act
+        task = ScheduledTask(
+            task_id="task-001",
+            handler_name="my_handler",
+            owner_scope="system",
+            owner_id="test:1",
+            schedule_kind="cron",
+        )
+        # Assert
+        assert task.task_id == "task-001"
+        assert task.handler_name == "my_handler"
+        assert task.owner_scope == "system"
+        assert task.owner_id == "test:1"
+        assert task.schedule_kind == "cron"
+
+    def test_default_values(self):
+        # Arrange & Act
+        task = ScheduledTask(
+            task_id="t1",
+            handler_name="h",
+            owner_scope="system",
+            owner_id="o",
+            schedule_kind="cron",
+        )
+        # Assert - 默认值与 ORM nullable/default 对齐
+        assert task.cron_expression is None
+        assert task.run_at is None
+        assert task.tz == "Asia/Shanghai"
+        assert task.payload == {}
+        assert task.enabled is True
+        assert task.delete_after_run is False
+        assert task.block_strategy == "discard_later"
+        assert task.stagger_seconds is None
+        assert task.consecutive_errors == 0
+        assert task.status == "active"
+        assert task.last_run_at is None
+        assert task.next_run_at is None
+        assert task.last_error is None
+        assert task.created_by == "system"
+        assert task.updated_by is None
+        assert task.id is None
+        assert task.created_at is None
+        assert task.updated_at is None
+        assert task.is_deleted == 0
+        assert task.deleted_at is None
+
+    def test_payload_default_is_isolated_per_instance(self):
+        # Arrange - 可变默认值必须 per-instance 隔离
+        task1 = ScheduledTask(
+            task_id="t1", handler_name="h", owner_scope="s", owner_id="o", schedule_kind="cron"
+        )
+        task2 = ScheduledTask(
+            task_id="t2", handler_name="h", owner_scope="s", owner_id="o", schedule_kind="cron"
+        )
+        # Act
+        task1.payload["key"] = "value"
+        # Assert - 修改 task1.payload 不应影响 task2.payload
+        assert task2.payload == {}
+
+    def test_field_overrides(self):
+        # Arrange & Act
+        task = ScheduledTask(
+            task_id="t1",
+            handler_name="h",
+            owner_scope="business",
+            owner_id="o",
+            schedule_kind="at",
+            cron_expression=None,
+            run_at="2024-01-01T00:00:00",
+            tz="UTC",
+            payload={"k": "v"},
+            enabled=False,
+            delete_after_run=True,
+            block_strategy="discard_later",
+            stagger_seconds=120,
+            consecutive_errors=3,
+            status="dead_letter",
+            created_by="admin",
+            id=42,
+        )
+        # Assert
+        assert task.schedule_kind == "at"
+        assert task.run_at == "2024-01-01T00:00:00"
+        assert task.tz == "UTC"
+        assert task.payload == {"k": "v"}
+        assert task.enabled is False
+        assert task.delete_after_run is True
+        assert task.stagger_seconds == 120
+        assert task.consecutive_errors == 3
+        assert task.status == "dead_letter"
+        assert task.created_by == "admin"
+        assert task.id == 42
+
+    def test_field_names_align_with_orm(self):
+        # Arrange & Act
+        field_names = {f.name for f in fields(ScheduledTask)}
+        # Assert - 关键字段必须存在（与 ORM ScheduledTask 对齐）
+        expected = {
+            "task_id",
+            "handler_name",
+            "owner_scope",
+            "owner_id",
+            "schedule_kind",
+            "cron_expression",
+            "run_at",
+            "tz",
+            "payload",
+            "enabled",
+            "delete_after_run",
+            "block_strategy",
+            "stagger_seconds",
+            "consecutive_errors",
+            "status",
+            "last_run_at",
+            "next_run_at",
+            "last_error",
+            "created_by",
+            "updated_by",
+            "id",
+            "created_at",
+            "updated_at",
+            "is_deleted",
+            "deleted_at",
+        }
+        assert expected.issubset(field_names)
+
+
+# ─── ScheduledTaskRunLog ──────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestScheduledTaskRunLog:
+    def test_required_fields_must_be_provided(self):
+        # Arrange & Act & Assert - task_id / run_id 必填
+        with pytest.raises(TypeError):
+            ScheduledTaskRunLog()  # type: ignore[call-arg]
+
+    def test_required_fields_assigned(self):
+        # Arrange & Act
+        log = ScheduledTaskRunLog(task_id="t1", run_id="r1")
+        # Assert
+        assert log.task_id == "t1"
+        assert log.run_id == "r1"
+
+    def test_default_values(self):
+        # Arrange & Act
+        log = ScheduledTaskRunLog(task_id="t1", run_id="r1")
+        # Assert
+        assert log.triggered_by == "auto"
+        assert log.status == "running"
+        assert log.error_message is None
+        assert log.output is None
+        assert log.started_at is None
+        assert log.finished_at is None
+        assert log.created_by == "system"
+        assert log.updated_by is None
+        assert log.id is None
+        assert log.created_at is None
+        assert log.updated_at is None
+        assert log.is_deleted == 0
+        assert log.deleted_at is None
+
+    def test_field_overrides(self):
+        # Arrange & Act
+        log = ScheduledTaskRunLog(
+            task_id="t1",
+            run_id="r1",
+            triggered_by="manual",
+            status="success",
+            error_message=None,
+            output={"result": "ok"},
+            started_at="2024-01-01T00:00:00",
+            finished_at="2024-01-01T00:00:05",
+        )
+        # Assert
+        assert log.triggered_by == "manual"
+        assert log.status == "success"
+        assert log.output == {"result": "ok"}
+        assert log.started_at == "2024-01-01T00:00:00"
+        assert log.finished_at == "2024-01-01T00:00:05"
+
+    def test_field_names_align_with_orm(self):
+        # Arrange & Act
+        field_names = {f.name for f in fields(ScheduledTaskRunLog)}
+        # Assert - 关键字段必须存在（与 ORM ScheduledTaskRunLog 对齐）
+        expected = {
+            "task_id",
+            "run_id",
+            "triggered_by",
+            "status",
+            "error_message",
+            "output",
+            "started_at",
+            "finished_at",
+            "created_by",
+            "updated_by",
+            "id",
+            "created_at",
+            "updated_at",
+            "is_deleted",
+            "deleted_at",
+        }
+        assert expected.issubset(field_names)
+
+
+# ─── HandlerSummary ───────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestHandlerSummary:
+    def test_required_fields_must_be_provided(self):
+        with pytest.raises(TypeError):
+            HandlerSummary()  # type: ignore[call-arg]
+
+    def test_required_fields_assigned(self):
+        summary = HandlerSummary(name="my_handler", task_count=5)
+        assert summary.name == "my_handler"
+        assert summary.task_count == 5
+
+    def test_last_active_at_defaults_to_none(self):
+        summary = HandlerSummary(name="h", task_count=0)
+        assert summary.last_active_at is None
+
+    def test_field_names_align_with_orm(self):
+        field_names = {f.name for f in fields(HandlerSummary)}
+        expected = {"name", "task_count", "last_active_at"}
+        assert expected.issubset(field_names)
+
+
+# ─── DailyStat ────────────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestDailyStat:
+    def test_required_fields_must_be_provided(self):
+        with pytest.raises(TypeError):
+            DailyStat()  # type: ignore[call-arg]
+
+    def test_required_fields_assigned(self):
+        # Arrange & Act
+        stat = DailyStat(
+            stat_date="2024-01-01",
+            handler_name="h",
+            success_count=10,
+            failure_count=2,
+            timeout_count=1,
+            dead_letter_count=0,
+        )
+        # Assert
+        assert stat.stat_date == "2024-01-01"
+        assert stat.handler_name == "h"
+        assert stat.success_count == 10
+        assert stat.failure_count == 2
+        assert stat.timeout_count == 1
+        assert stat.dead_letter_count == 0
+
+    def test_field_names_align_with_orm(self):
+        field_names = {f.name for f in fields(DailyStat)}
+        expected = {
+            "stat_date",
+            "handler_name",
+            "success_count",
+            "failure_count",
+            "timeout_count",
+            "dead_letter_count",
+        }
+        assert expected.issubset(field_names)

backend/test/unit/scheduler/core/test_validators.py

@@ -0,0 +1,188 @@
+"""core 层校验函数单元测试。
+
+覆盖 ``yuxi.scheduler.core.validators`` 模块的全部公开校验函数：
+- ``validate_cron_expression``：cron 表达式语法 / strict 模式 / 最小触发间隔
+- ``validate_payload_size``：payload 序列化后字节数上限
+- ``validate_owner_scope``：非空 + 长度上限
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from yuxi.scheduler.core.validators import (
+    validate_cron_expression,
+    validate_owner_scope,
+    validate_payload_size,
+)
+from yuxi.scheduler.exceptions import (
+    InvalidCronExpressionError,
+    SchedulerPayloadTooLargeError,
+    SchedulerValidationError,
+)
+
+
+pytestmark = pytest.mark.unit
+
+
+# ─── validate_cron_expression ─────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestValidateCronExpression:
+    @pytest.mark.parametrize(
+        "expr",
+        [
+            "*/15 * * * *",  # 每 15 分钟
+            "0 0 * * *",     # 每天凌晨
+            "0 * * * *",     # 每小时
+            "0 0 * * 1",     # 每周一凌晨
+            "0 0 1 * *",     # 每月 1 号凌晨
+            "*/30 * * * *",  # 每 30 分钟
+        ],
+        ids=["every-15min", "daily", "hourly", "weekly", "monthly", "every-30min"],
+    )
+    def test_valid_cron_expressions_passes(self, expr: str):
+        # Act - 无异常抛出即通过
+        validate_cron_expression(expr)
+
+    @pytest.mark.parametrize(
+        ("expr", "match"),
+        [
+            ("", "不能为空"),
+            ("   ", "不能为空"),
+            ("* * *", "cron 表达式非法"),        # 字段数不足
+            ("99 * * * *", "cron 表达式非法"),   # 分钟字段越界
+            ("invalid", "cron 表达式非法"),       # 非法语法
+        ],
+        ids=["empty", "whitespace", "too-few-fields", "field-out-of-range", "invalid-syntax"],
+    )
+    def test_invalid_cron_expressions_raise(self, expr: str, match: str):
+        with pytest.raises(InvalidCronExpressionError, match=match):
+            validate_cron_expression(expr)
+
+    def test_strict_mode_rejects_6_field_cron(self):
+        # Arrange - strict 模式禁止含秒级字段
+        expr = "0 */15 * * * *"
+        # Act & Assert
+        with pytest.raises(InvalidCronExpressionError, match="strict 模式禁止含秒级字段"):
+            validate_cron_expression(expr, strict=True)
+
+    def test_strict_mode_rejects_every_minute(self):
+        # Arrange - strict 模式禁止每分钟执行
+        expr = "* * * * *"
+        # Act & Assert
+        with pytest.raises(InvalidCronExpressionError, match="strict 模式禁止每分钟执行"):
+            validate_cron_expression(expr, strict=True)
+
+    def test_non_strict_mode_allows_6_field_cron(self):
+        # Arrange - strict=False 时不检查字段数
+        # 注：croniter 对 6 字段 cron 的间隔计算为 1 秒（秒字段递增），
+        # 故需 min_interval_minutes=0 跳过间隔校验，仅验证 strict=False 不拒绝 6 字段
+        expr = "0 */15 * * * *"
+        # Act - 无异常抛出即通过
+        validate_cron_expression(expr, strict=False, min_interval_minutes=0)
+
+    @pytest.mark.parametrize(
+        ("expr", "min_interval", "should_raise"),
+        [
+            ("*/5 * * * *", 10, True),   # 5 分钟 < 10 分钟阈值
+            ("*/5 * * * *", 3, False),   # 5 分钟 >= 3 分钟阈值
+            ("*/5 * * * *", 0, False),   # 0 跳过间隔校验
+            ("*/15 * * * *", 10, False), # 15 分钟 >= 10 分钟阈值
+            ("*/15 * * * *", 20, True),  # 15 分钟 < 20 分钟阈值
+        ],
+        ids=["below-default", "below-custom-ok", "zero-skips", "at-default", "below-higher"],
+    )
+    def test_min_interval_check(
+        self, expr: str, min_interval: int, should_raise: bool
+    ):
+        # Act & Assert
+        if should_raise:
+            with pytest.raises(InvalidCronExpressionError, match="小于最小要求"):
+                validate_cron_expression(expr, min_interval_minutes=min_interval)
+        else:
+            validate_cron_expression(expr, min_interval_minutes=min_interval)
+
+    def test_invalid_cron_raises_invalid_cron_error_subclass(self):
+        # Arrange & Act & Assert - 异常类型为 SchedulerValidationError 子类
+        with pytest.raises(SchedulerValidationError):
+            validate_cron_expression("invalid")
+
+
+# ─── validate_payload_size ────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestValidatePayloadSize:
+    def test_small_payload_passes(self):
+        # Arrange
+        payload = {"key": "value"}
+        # Act - 无异常抛出即通过
+        validate_payload_size(payload, max_size_kb=64)
+
+    def test_empty_payload_passes(self):
+        # Arrange
+        payload: dict = {}
+        # Act
+        validate_payload_size(payload, max_size_kb=64)
+
+    def test_payload_at_limit_passes(self):
+        # Arrange - 构造恰好 1KB 的 payload（max_size_kb=1）
+        # {"k": ""} 序列化为 '{"k": ""}' 共 9 字节，故 value 长度 = 1024 - 9
+        payload = {"k": "a" * (1024 - 9)}  # 9 = len(json.dumps({"k": ""})) 的 JSON 开销
+        # Act
+        validate_payload_size(payload, max_size_kb=1)
+
+    def test_payload_exceeding_limit_raises(self):
+        # Arrange - 构造超过 1KB 的 payload
+        payload = {"k": "a" * 2048}
+        # Act & Assert
+        with pytest.raises(SchedulerPayloadTooLargeError, match="payload 大小超限"):
+            validate_payload_size(payload, max_size_kb=1)
+
+    def test_unicode_payload_size_calculated_by_bytes(self):
+        # Arrange - 中文字符 UTF-8 占 3 字节，确保按字节数而非字符数计算
+        payload = {"k": "中" * 500}  # 500 * 3 = 1500 bytes > 1KB
+        # Act & Assert
+        with pytest.raises(SchedulerPayloadTooLargeError):
+            validate_payload_size(payload, max_size_kb=1)
+
+    def test_exceeding_payload_raises_validation_error_subclass(self):
+        # Arrange & Act & Assert - 异常类型为 SchedulerValidationError 子类
+        with pytest.raises(SchedulerValidationError):
+            validate_payload_size({"k": "a" * 2048}, max_size_kb=1)
+
+    def test_max_size_zero_rejects_any_payload(self):
+        # Arrange - max_size_kb=0 时任何 payload 都超限（空 dict 序列化为 2 字节）
+        payload = {"k": "v"}
+        # Act & Assert
+        with pytest.raises(SchedulerPayloadTooLargeError):
+            validate_payload_size(payload, max_size_kb=0)
+
+
+# ─── validate_owner_scope ─────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestValidateOwnerScope:
+    @pytest.mark.parametrize(
+        "scope",
+        ["system", "business", "custom_scope", "a" * 64],
+        ids=["system", "business", "custom", "at-max-length"],
+    )
+    def test_valid_scope_passes(self, scope: str):
+        # Act - 无异常抛出即通过
+        validate_owner_scope(scope)
+
+    @pytest.mark.parametrize(
+        ("scope", "match"),
+        [
+            ("", "owner_scope 非法"),
+            ("a" * 65, "owner_scope 非法"),  # 长度超过 64
+        ],
+        ids=["empty", "exceeds-max-length"],
+    )
+    def test_invalid_scope_raises(self, scope: str, match: str):
+        with pytest.raises(SchedulerValidationError, match=match):
+            validate_owner_scope(scope)

backend/test/unit/scheduler/framework/handlers/test_idempotency_cleanup_handler.py

@@ -0,0 +1,214 @@
+"""清理过期幂等记录 handler 单元测试。
+
+覆盖 ``yuxi.scheduler.framework.handlers.idempotency_cleanup_handler`` 模块的
+``IdempotencyCleanupHandler``：
+- 正常执行：调用仓储 cleanup_old 方法并返回成功结果
+- payload 覆盖默认保留小时数
+- 异常隔离：仓储异常转换为 TaskResult(success=False)
+"""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from datetime import datetime, timedelta
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from yuxi.scheduler.core.contracts import TaskContext
+from yuxi.scheduler.framework.handlers.idempotency_cleanup_handler import (
+    IdempotencyCleanupHandler,
+    _DEFAULT_RETENTION_HOURS,
+)
+
+
+pytestmark = pytest.mark.unit
+
+
+# ─── 测试辅助 ─────────────────────────────────────────────────────────────
+
+
+def _make_ctx(**payload_overrides) -> TaskContext:
+    """构造 TaskContext，payload 可按需覆盖。"""
+    return TaskContext(
+        task_id="task-cleanup",
+        run_id="run-001",
+        handler_name="idempotency_cleanup",
+        payload=payload_overrides,
+        triggered_by="auto",
+        scheduled_at=datetime(2024, 1, 1, 0, 0, 0),
+    )
+
+
+@asynccontextmanager
+async def _fake_session_factory(db_mock):
+    """构造假的 session_factory，yield db_mock。"""
+    yield db_mock
+
+
+# ─── 类属性 ───────────────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestIdempotencyCleanupHandlerAttributes:
+    def test_name_is_idempotency_cleanup(self):
+        assert IdempotencyCleanupHandler.name == "idempotency_cleanup"
+
+    def test_description_is_non_empty(self):
+        assert IdempotencyCleanupHandler.description
+        assert isinstance(IdempotencyCleanupHandler.description, str)
+
+
+# ─── execute 正常路径 ─────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestIdempotencyCleanupHandlerExecute:
+    @pytest.mark.asyncio
+    async def test_successful_cleanup_returns_success(self):
+        # Arrange
+        db = MagicMock()
+        repos = MagicMock()
+        repos.idempotency = AsyncMock()
+        repos.idempotency.cleanup_old = AsyncMock(return_value=42)
+
+        handler = IdempotencyCleanupHandler(
+            session_factory=lambda: _fake_session_factory(db),
+            default_retention_hours=24,
+        )
+
+        # Act
+        with patch(
+            "yuxi.scheduler.framework.handlers.idempotency_cleanup_handler.create_repositories",
+            return_value=repos,
+        ):
+            result = await handler.execute(_make_ctx())
+
+        # Assert
+        assert result.success is True
+        assert result.output["deleted_count"] == 42
+        assert result.output["retention_hours"] == 24
+
+    @pytest.mark.asyncio
+    async def test_payload_retention_hours_overrides_default(self):
+        # Arrange - payload 指定 retention_hours=12
+        db = MagicMock()
+        repos = MagicMock()
+        repos.idempotency = AsyncMock()
+        repos.idempotency.cleanup_old = AsyncMock(return_value=5)
+
+        handler = IdempotencyCleanupHandler(
+            session_factory=lambda: _fake_session_factory(db),
+            default_retention_hours=24,
+        )
+
+        # Act
+        with patch(
+            "yuxi.scheduler.framework.handlers.idempotency_cleanup_handler.create_repositories",
+            return_value=repos,
+        ):
+            result = await handler.execute(_make_ctx(retention_hours=12))
+
+        # Assert - 使用 payload 中的 12 小时
+        assert result.output["retention_hours"] == 12
+
+    @pytest.mark.asyncio
+    async def test_default_retention_hours_used_when_payload_missing(self):
+        # Arrange - payload 不含 retention_hours
+        db = MagicMock()
+        repos = MagicMock()
+        repos.idempotency = AsyncMock()
+        repos.idempotency.cleanup_old = AsyncMock(return_value=0)
+
+        handler = IdempotencyCleanupHandler(
+            session_factory=lambda: _fake_session_factory(db),
+            default_retention_hours=_DEFAULT_RETENTION_HOURS,
+        )
+
+        # Act
+        with patch(
+            "yuxi.scheduler.framework.handlers.idempotency_cleanup_handler.create_repositories",
+            return_value=repos,
+        ):
+            result = await handler.execute(_make_ctx())
+
+        # Assert
+        assert result.output["retention_hours"] == _DEFAULT_RETENTION_HOURS
+
+    @pytest.mark.asyncio
+    async def test_cleanup_old_called_with_correct_cutoff(self):
+        # Arrange
+        db = MagicMock()
+        repos = MagicMock()
+        repos.idempotency = AsyncMock()
+        repos.idempotency.cleanup_old = AsyncMock(return_value=0)
+
+        handler = IdempotencyCleanupHandler(
+            session_factory=lambda: _fake_session_factory(db),
+            default_retention_hours=24,
+        )
+
+        # Act
+        with patch(
+            "yuxi.scheduler.framework.handlers.idempotency_cleanup_handler.create_repositories",
+            return_value=repos,
+        ), patch(
+            "yuxi.scheduler.framework.handlers.idempotency_cleanup_handler.utc_now_naive",
+            return_value=datetime(2024, 6, 1, 12, 0, 0),
+        ):
+            await handler.execute(_make_ctx())
+
+        # Assert - cutoff = now - 24 hours
+        repos.idempotency.cleanup_old.assert_awaited_once()
+        called_cutoff = repos.idempotency.cleanup_old.call_args[0][0]
+        expected_cutoff = datetime(2024, 6, 1, 12, 0, 0) - timedelta(hours=24)
+        assert called_cutoff == expected_cutoff
+
+
+# ─── execute 异常路径 ─────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestIdempotencyCleanupHandlerErrors:
+    @pytest.mark.asyncio
+    async def test_repository_exception_returns_failure(self):
+        # Arrange - 仓储抛异常
+        db = MagicMock()
+        repos = MagicMock()
+        repos.idempotency = AsyncMock()
+        repos.idempotency.cleanup_old = AsyncMock(side_effect=RuntimeError("db error"))
+
+        handler = IdempotencyCleanupHandler(
+            session_factory=lambda: _fake_session_factory(db),
+        )
+
+        # Act
+        with patch(
+            "yuxi.scheduler.framework.handlers.idempotency_cleanup_handler.create_repositories",
+            return_value=repos,
+        ):
+            result = await handler.execute(_make_ctx())
+
+        # Assert - 异常被捕获并转换为 TaskResult(success=False)
+        assert result.success is False
+        assert "db error" in (result.error or "")
+
+    @pytest.mark.asyncio
+    async def test_session_factory_exception_returns_failure(self):
+        # Arrange - session_factory 自身抛异常
+
+        @asynccontextmanager
+        async def _broken_session_factory():
+            raise RuntimeError("session creation failed")
+            yield  # pragma: no cover
+
+        handler = IdempotencyCleanupHandler(
+            session_factory=_broken_session_factory,
+        )
+
+        # Act
+        result = await handler.execute(_make_ctx())
+
+        # Assert
+        assert result.success is False
+        assert "session creation failed" in (result.error or "")

backend/test/unit/scheduler/framework/handlers/test_run_log_cleanup_handler.py

@@ -0,0 +1,225 @@
+"""清理过期执行日志 handler 单元测试。
+
+覆盖 ``yuxi.scheduler.framework.handlers.run_log_cleanup_handler`` 模块的
+``RunLogCleanupHandler``：
+- 正常执行：调用仓储 cleanup 方法并返回成功结果
+- payload 覆盖默认保留天数
+- 异常隔离：仓储异常转换为 TaskResult(success=False)
+- 会话隔离：通过 session_factory 创建独立会话
+"""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from datetime import datetime, timedelta
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from yuxi.scheduler.core.contracts import TaskContext, TaskResult
+from yuxi.scheduler.framework.handlers.run_log_cleanup_handler import (
+    RunLogCleanupHandler,
+    _DEFAULT_RETENTION_DAYS,
+)
+
+
+pytestmark = pytest.mark.unit
+
+
+# ─── 测试辅助 ─────────────────────────────────────────────────────────────
+
+
+def _make_ctx(**payload_overrides) -> TaskContext:
+    """构造 TaskContext，payload 可按需覆盖。"""
+    return TaskContext(
+        task_id="task-cleanup",
+        run_id="run-001",
+        handler_name="run_log_cleanup",
+        payload=payload_overrides,
+        triggered_by="auto",
+        scheduled_at=datetime(2024, 1, 1, 0, 0, 0),
+    )
+
+
+@asynccontextmanager
+async def _fake_session_factory(db_mock):
+    """构造假的 session_factory，yield db_mock。"""
+    yield db_mock
+
+
+# ─── 类属性 ───────────────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestRunLogCleanupHandlerAttributes:
+    def test_name_is_run_log_cleanup(self):
+        assert RunLogCleanupHandler.name == "run_log_cleanup"
+
+    def test_description_is_non_empty(self):
+        assert RunLogCleanupHandler.description
+        assert isinstance(RunLogCleanupHandler.description, str)
+
+
+# ─── execute 正常路径 ─────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestRunLogCleanupHandlerExecute:
+    @pytest.mark.asyncio
+    async def test_successful_cleanup_returns_success(self):
+        # Arrange
+        db = MagicMock()
+        repos = MagicMock()
+        repos.run_log = AsyncMock()
+        repos.run_log.cleanup_old_logs = AsyncMock(return_value=10)
+        repos.run_log_daily = AsyncMock()
+        repos.run_log_daily.cleanup_old_daily_stats = AsyncMock(return_value=5)
+
+        handler = RunLogCleanupHandler(
+            session_factory=lambda: _fake_session_factory(db),
+            default_retention_days=30,
+        )
+
+        # Act
+        with patch(
+            "yuxi.scheduler.framework.handlers.run_log_cleanup_handler.create_repositories",
+            return_value=repos,
+        ):
+            result = await handler.execute(_make_ctx())
+
+        # Assert
+        assert result.success is True
+        assert result.output["deleted_logs"] == 10
+        assert result.output["deleted_stats"] == 5
+        assert result.output["retention_days"] == 30
+
+    @pytest.mark.asyncio
+    async def test_payload_retention_days_overrides_default(self):
+        # Arrange - payload 指定 retention_days=7
+        db = MagicMock()
+        repos = MagicMock()
+        repos.run_log = AsyncMock()
+        repos.run_log.cleanup_old_logs = AsyncMock(return_value=3)
+        repos.run_log_daily = AsyncMock()
+        repos.run_log_daily.cleanup_old_daily_stats = AsyncMock(return_value=1)
+
+        handler = RunLogCleanupHandler(
+            session_factory=lambda: _fake_session_factory(db),
+            default_retention_days=30,
+        )
+
+        # Act
+        with patch(
+            "yuxi.scheduler.framework.handlers.run_log_cleanup_handler.create_repositories",
+            return_value=repos,
+        ):
+            result = await handler.execute(_make_ctx(retention_days=7))
+
+        # Assert - 使用 payload 中的 7 天，而非默认 30 天
+        assert result.output["retention_days"] == 7
+
+    @pytest.mark.asyncio
+    async def test_default_retention_days_used_when_payload_missing(self):
+        # Arrange - payload 不含 retention_days
+        db = MagicMock()
+        repos = MagicMock()
+        repos.run_log = AsyncMock()
+        repos.run_log.cleanup_old_logs = AsyncMock(return_value=0)
+        repos.run_log_daily = AsyncMock()
+        repos.run_log_daily.cleanup_old_daily_stats = AsyncMock(return_value=0)
+
+        handler = RunLogCleanupHandler(
+            session_factory=lambda: _fake_session_factory(db),
+            default_retention_days=_DEFAULT_RETENTION_DAYS,
+        )
+
+        # Act
+        with patch(
+            "yuxi.scheduler.framework.handlers.run_log_cleanup_handler.create_repositories",
+            return_value=repos,
+        ):
+            result = await handler.execute(_make_ctx())
+
+        # Assert
+        assert result.output["retention_days"] == _DEFAULT_RETENTION_DAYS
+
+    @pytest.mark.asyncio
+    async def test_cleanup_old_logs_called_with_correct_cutoff(self):
+        # Arrange
+        db = MagicMock()
+        repos = MagicMock()
+        repos.run_log = AsyncMock()
+        repos.run_log.cleanup_old_logs = AsyncMock(return_value=0)
+        repos.run_log_daily = AsyncMock()
+        repos.run_log_daily.cleanup_old_daily_stats = AsyncMock(return_value=0)
+
+        handler = RunLogCleanupHandler(
+            session_factory=lambda: _fake_session_factory(db),
+            default_retention_days=90,
+        )
+
+        # Act
+        with patch(
+            "yuxi.scheduler.framework.handlers.run_log_cleanup_handler.create_repositories",
+            return_value=repos,
+        ), patch(
+            "yuxi.scheduler.framework.handlers.run_log_cleanup_handler.utc_now_naive",
+            return_value=datetime(2024, 6, 1, 0, 0, 0),
+        ):
+            await handler.execute(_make_ctx())
+
+        # Assert - cutoff = now - 90 days
+        repos.run_log.cleanup_old_logs.assert_awaited_once()
+        called_cutoff = repos.run_log.cleanup_old_logs.call_args[0][0]
+        expected_cutoff = datetime(2024, 6, 1, 0, 0, 0) - timedelta(days=90)
+        assert called_cutoff == expected_cutoff
+
+
+# ─── execute 异常路径 ─────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestRunLogCleanupHandlerErrors:
+    @pytest.mark.asyncio
+    async def test_repository_exception_returns_failure(self):
+        # Arrange - 仓储抛异常
+        db = MagicMock()
+        repos = MagicMock()
+        repos.run_log = AsyncMock()
+        repos.run_log.cleanup_old_logs = AsyncMock(side_effect=RuntimeError("db error"))
+        repos.run_log_daily = AsyncMock()
+
+        handler = RunLogCleanupHandler(
+            session_factory=lambda: _fake_session_factory(db),
+        )
+
+        # Act
+        with patch(
+            "yuxi.scheduler.framework.handlers.run_log_cleanup_handler.create_repositories",
+            return_value=repos,
+        ):
+            result = await handler.execute(_make_ctx())
+
+        # Assert - 异常被捕获并转换为 TaskResult(success=False)
+        assert result.success is False
+        assert "db error" in (result.error or "")
+
+    @pytest.mark.asyncio
+    async def test_session_factory_exception_returns_failure(self):
+        # Arrange - session_factory 自身抛异常
+
+        @asynccontextmanager
+        async def _broken_session_factory():
+            raise RuntimeError("session creation failed")
+            yield  # pragma: no cover
+
+        handler = RunLogCleanupHandler(
+            session_factory=_broken_session_factory,
+        )
+
+        # Act
+        result = await handler.execute(_make_ctx())
+
+        # Assert
+        assert result.success is False
+        assert "session creation failed" in (result.error or "")

backend/test/unit/scheduler/framework/runtime/test_handler_registry.py

@@ -0,0 +1,143 @@
+"""handler 注册表单元测试。
+
+覆盖 ``yuxi.scheduler.framework.runtime.handler_registry`` 模块的 ``HandlerRegistry``：
+- ``register``：注册 handler，重复注册抛 ``DuplicateHandlerError``
+- ``get``：按 name 查找 handler，未命中返回 None
+- ``list_names``：返回排序后的 handler 名称列表
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import pytest
+
+from yuxi.scheduler.core.contracts import TaskContext, TaskResult
+from yuxi.scheduler.exceptions import DuplicateHandlerError
+from yuxi.scheduler.framework.runtime.handler_registry import HandlerRegistry
+
+
+pytestmark = pytest.mark.unit
+
+
+# ─── 测试用 handler ───────────────────────────────────────────────────────
+
+
+@dataclass
+class _FakeHandler:
+    """符合 TaskHandler Protocol 的测试 handler。"""
+
+    name: str
+    description: str = "test handler"
+
+    async def execute(self, ctx: TaskContext) -> TaskResult:
+        return TaskResult(success=True)
+
+
+# ─── register ─────────────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestHandlerRegistryRegister:
+    def test_register_single_handler(self):
+        # Arrange
+        registry = HandlerRegistry()
+        handler = _FakeHandler(name="my_handler")
+        # Act
+        registry.register(handler)
+        # Assert
+        assert registry.get("my_handler") is handler
+
+    def test_register_multiple_handlers(self):
+        # Arrange
+        registry = HandlerRegistry()
+        h1 = _FakeHandler(name="h1")
+        h2 = _FakeHandler(name="h2")
+        # Act
+        registry.register(h1)
+        registry.register(h2)
+        # Assert
+        assert registry.get("h1") is h1
+        assert registry.get("h2") is h2
+
+    def test_register_duplicate_raises(self):
+        # Arrange - 重复注册抛 DuplicateHandlerError
+        registry = HandlerRegistry()
+        registry.register(_FakeHandler(name="my_handler"))
+        # Act & Assert
+        with pytest.raises(DuplicateHandlerError, match="already registered"):
+            registry.register(_FakeHandler(name="my_handler"))
+
+    def test_register_duplicate_error_carries_handler_key(self):
+        # Arrange
+        registry = HandlerRegistry()
+        registry.register(_FakeHandler(name="my_handler"))
+        # Act & Assert
+        with pytest.raises(DuplicateHandlerError) as exc_info:
+            registry.register(_FakeHandler(name="my_handler"))
+        assert exc_info.value.details["handler_key"] == "my_handler"
+
+    def test_register_uses_handler_name_attribute(self):
+        # Arrange - handler 唯一标识取 handler.name
+        registry = HandlerRegistry()
+        handler = _FakeHandler(name="custom_name")
+        # Act
+        registry.register(handler)
+        # Assert - 不能用其他字符串作为 key
+        assert registry.get("custom_name") is handler
+        assert registry.get("_FakeHandler") is None
+
+
+# ─── get ──────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestHandlerRegistryGet:
+    def test_get_existing_handler(self):
+        # Arrange
+        registry = HandlerRegistry()
+        handler = _FakeHandler(name="my_handler")
+        registry.register(handler)
+        # Act
+        result = registry.get("my_handler")
+        # Assert
+        assert result is handler
+
+    def test_get_nonexistent_returns_none(self):
+        # Arrange
+        registry = HandlerRegistry()
+        # Act
+        result = registry.get("nonexistent")
+        # Assert - 未命中返回 None，由调用方决定是否抛 HandlerNotFoundError
+        assert result is None
+
+    def test_get_on_empty_registry_returns_none(self):
+        registry = HandlerRegistry()
+        assert registry.get("any") is None
+
+
+# ─── list_names ───────────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestHandlerRegistryListNames:
+    def test_empty_registry_returns_empty_list(self):
+        registry = HandlerRegistry()
+        assert registry.list_names() == []
+
+    def test_single_handler(self):
+        registry = HandlerRegistry()
+        registry.register(_FakeHandler(name="h1"))
+        assert registry.list_names() == ["h1"]
+
+    def test_multiple_handlers_sorted(self):
+        # Arrange
+        registry = HandlerRegistry()
+        # 故意乱序注册
+        registry.register(_FakeHandler(name="charlie"))
+        registry.register(_FakeHandler(name="alpha"))
+        registry.register(_FakeHandler(name="bravo"))
+        # Act
+        names = registry.list_names()
+        # Assert - 排序后返回
+        assert names == ["alpha", "bravo", "charlie"]

backend/test/unit/scheduler/infrastructure/test_container.py

@@ -0,0 +1,164 @@
+"""infrastructure/container.py 装配容器单元测试。
+
+覆盖 ``yuxi.scheduler.infrastructure.container`` 模块的两个 factory 函数：
+- ``create_scheduler_service``：api 进程请求级 factory（handler_registry=None）
+- ``create_worker_scheduler_service``：worker 进程请求级 factory（完整 runtime）
+
+验证装配结果 ``SchedulerService`` 实例的依赖注入正确性。
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from yuxi.scheduler.framework.runtime import HandlerRegistry, RuntimeServices
+from yuxi.scheduler.infrastructure.container import (
+    SchedulerService,
+    create_scheduler_service,
+    create_worker_scheduler_service,
+)
+
+
+pytestmark = pytest.mark.unit
+
+
+@pytest.fixture
+def db_session() -> MagicMock:
+    """提供 MagicMock 模拟 AsyncSession。"""
+    return MagicMock()
+
+
+# ─── create_scheduler_service ─────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestCreateSchedulerService:
+    def test_returns_scheduler_service_instance(self, db_session: MagicMock):
+        # Arrange & Act
+        with patch("yuxi.scheduler.infrastructure.container.app_config"):
+            service = create_scheduler_service(db_session)
+        # Assert
+        assert isinstance(service, SchedulerService)
+
+    def test_handler_registry_is_none(self, db_session: MagicMock):
+        # Arrange & Act - api 进程 handler_registry=None
+        with patch("yuxi.scheduler.infrastructure.container.app_config"):
+            service = create_scheduler_service(db_session)
+        # Assert
+        assert service._runtime.handler_registry is None
+
+    def test_arq_pool_defaults_to_none(self, db_session: MagicMock):
+        with patch("yuxi.scheduler.infrastructure.container.app_config"):
+            service = create_scheduler_service(db_session)
+        assert service._runtime.arq_pool is None
+
+    def test_arq_pool_passed_through(self, db_session: MagicMock):
+        # Arrange
+        pool = MagicMock()
+        # Act
+        with patch("yuxi.scheduler.infrastructure.container.app_config"):
+            service = create_scheduler_service(db_session, arq_pool=pool)
+        # Assert
+        assert service._runtime.arq_pool is pool
+
+    def test_repos_injected(self, db_session: MagicMock):
+        with patch("yuxi.scheduler.infrastructure.container.app_config"):
+            service = create_scheduler_service(db_session)
+        # Assert - 4 个仓储实例被注入
+        assert service._repos.task is not None
+        assert service._repos.run_log is not None
+        assert service._repos.run_log_daily is not None
+        assert service._repos.idempotency is not None
+
+    def test_uow_injected(self, db_session: MagicMock):
+        with patch("yuxi.scheduler.infrastructure.container.app_config"):
+            service = create_scheduler_service(db_session)
+        assert service._uow is not None
+
+    def test_config_injected(self, db_session: MagicMock):
+        with patch("yuxi.scheduler.infrastructure.container.app_config") as mock_config:
+            service = create_scheduler_service(db_session)
+        assert service._runtime.config is mock_config
+
+
+# ─── create_worker_scheduler_service ──────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestCreateWorkerSchedulerService:
+    def test_returns_scheduler_service_instance(self, db_session: MagicMock):
+        # Arrange
+        registry = HandlerRegistry()
+        pool = MagicMock()
+        # Act
+        with patch("yuxi.scheduler.infrastructure.container.app_config"):
+            service = create_worker_scheduler_service(
+                db_session, handler_registry=registry, arq_pool=pool
+            )
+        # Assert
+        assert isinstance(service, SchedulerService)
+
+    def test_handler_registry_injected(self, db_session: MagicMock):
+        registry = HandlerRegistry()
+        pool = MagicMock()
+        with patch("yuxi.scheduler.infrastructure.container.app_config"):
+            service = create_worker_scheduler_service(
+                db_session, handler_registry=registry, arq_pool=pool
+            )
+        assert service._runtime.handler_registry is registry
+
+    def test_arq_pool_injected(self, db_session: MagicMock):
+        registry = HandlerRegistry()
+        pool = MagicMock()
+        with patch("yuxi.scheduler.infrastructure.container.app_config"):
+            service = create_worker_scheduler_service(
+                db_session, handler_registry=registry, arq_pool=pool
+            )
+        assert service._runtime.arq_pool is pool
+
+    def test_repos_injected(self, db_session: MagicMock):
+        registry = HandlerRegistry()
+        pool = MagicMock()
+        with patch("yuxi.scheduler.infrastructure.container.app_config"):
+            service = create_worker_scheduler_service(
+                db_session, handler_registry=registry, arq_pool=pool
+            )
+        assert service._repos.task is not None
+        assert service._repos.run_log is not None
+        assert service._repos.run_log_daily is not None
+        assert service._repos.idempotency is not None
+
+    def test_uow_injected(self, db_session: MagicMock):
+        registry = HandlerRegistry()
+        pool = MagicMock()
+        with patch("yuxi.scheduler.infrastructure.container.app_config"):
+            service = create_worker_scheduler_service(
+                db_session, handler_registry=registry, arq_pool=pool
+            )
+        assert service._uow is not None
+
+
+# ─── RuntimeServices 装配 ─────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestRuntimeServicesAssembly:
+    def test_api_runtime_has_none_registry(self, db_session: MagicMock):
+        with patch("yuxi.scheduler.infrastructure.container.app_config"):
+            service = create_scheduler_service(db_session)
+        # Assert - RuntimeServices dataclass 字段正确
+        assert isinstance(service._runtime, RuntimeServices)
+        assert service._runtime.handler_registry is None
+
+    def test_worker_runtime_has_registry_and_pool(self, db_session: MagicMock):
+        registry = HandlerRegistry()
+        pool = MagicMock()
+        with patch("yuxi.scheduler.infrastructure.container.app_config"):
+            service = create_worker_scheduler_service(
+                db_session, handler_registry=registry, arq_pool=pool
+            )
+        assert isinstance(service._runtime, RuntimeServices)
+        assert service._runtime.handler_registry is registry
+        assert service._runtime.arq_pool is pool

backend/test/unit/scheduler/use_cases/test_mappers.py

@@ -0,0 +1,502 @@
+"""领域模型(dataclass) ↔ Output DTO(Pydantic) 转换单元测试。
+
+覆盖 ``yuxi.scheduler.use_cases.mappers`` 模块的全部公开函数：
+- ``to_task_output`` / ``to_run_log_output``：dataclass → Output DTO
+- ``to_handler_summary_output`` / ``to_list_handler_summary_output``
+- ``to_count_by_status_output``：状态分布字典 → DTO
+- ``to_get_health_output``：健康检查参数 → DTO
+- ``daily_stat_to_dict`` / ``to_daily_stat_output`` / ``to_list_daily_stats_output``
+- ``to_list_upcoming_output``
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+import pytest
+
+from yuxi.scheduler.core.models import (
+    DailyStat,
+    HandlerSummary,
+    ScheduledTask,
+    ScheduledTaskRunLog,
+)
+from yuxi.scheduler.use_cases.mappers import (
+    daily_stat_to_dict,
+    to_count_by_status_output,
+    to_daily_stat_output,
+    to_get_health_output,
+    to_handler_summary_output,
+    to_list_daily_stats_output,
+    to_list_handler_summary_output,
+    to_list_upcoming_output,
+    to_run_log_output,
+    to_task_output,
+)
+
+
+pytestmark = pytest.mark.unit
+
+
+# ─── to_task_output ───────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestToTaskOutput:
+    def test_basic_mapping(self):
+        # Arrange
+        task = ScheduledTask(
+            task_id="task-001",
+            handler_name="my_handler",
+            owner_scope="system",
+            owner_id="test:1",
+            schedule_kind="cron",
+            cron_expression="*/15 * * * *",
+            tz="Asia/Shanghai",
+            payload={"k": "v"},
+            enabled=True,
+            stagger_seconds=30,
+            status="active",
+            next_run_at=datetime(2024, 1, 1, 0, 15, 0),
+            id=1,
+        )
+        # Act
+        output = to_task_output(task)
+        # Assert
+        assert output.task_id == "task-001"
+        assert output.handler_name == "my_handler"
+        assert output.owner_scope == "system"
+        assert output.schedule_kind == "cron"
+        assert output.cron_expression == "*/15 * * * *"
+        assert output.payload == {"k": "v"}
+        assert output.stagger_seconds == 30
+        assert output.status == "active"
+        assert output.id == 1
+
+    def test_datetime_fields_formatted_as_iso(self):
+        # Arrange - naive datetime 按 UTC 处理（与 DB 存储约定一致）
+        next_run = datetime(2024, 1, 1, 0, 15, 0)
+        task = ScheduledTask(
+            task_id="t1",
+            handler_name="h",
+            owner_scope="s",
+            owner_id="o",
+            schedule_kind="cron",
+            next_run_at=next_run,
+        )
+        # Act
+        output = to_task_output(task)
+        # Assert - DateTime 字段格式化为 ISO 8601 字符串，naive UTC 不偏移
+        assert isinstance(output.next_run_at, str)
+        assert output.next_run_at == "2024-01-01T00:15:00Z"
+
+    def test_none_datetime_fields_return_none(self):
+        # Arrange
+        task = ScheduledTask(
+            task_id="t1",
+            handler_name="h",
+            owner_scope="s",
+            owner_id="o",
+            schedule_kind="cron",
+        )
+        # Act
+        output = to_task_output(task)
+        # Assert
+        assert output.next_run_at is None
+        assert output.last_run_at is None
+        assert output.run_at is None
+        assert output.created_at is None
+        assert output.updated_at is None
+        assert output.deleted_at is None
+
+    def test_none_payload_returns_empty_dict(self):
+        # Arrange - dataclass payload 为 None 时 mapper 兜底为空 dict
+        task = ScheduledTask(
+            task_id="t1",
+            handler_name="h",
+            owner_scope="s",
+            owner_id="o",
+            schedule_kind="cron",
+            payload=None,  # type: ignore[arg-type]
+        )
+        # Act
+        output = to_task_output(task)
+        # Assert
+        assert output.payload == {}
+
+
+# ─── to_run_log_output ────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestToRunLogOutput:
+    def test_basic_mapping(self):
+        # Arrange
+        log = ScheduledTaskRunLog(
+            task_id="task-001",
+            run_id="run-001",
+            triggered_by="auto",
+            status="success",
+            output={"rows": 10},
+            started_at=datetime(2024, 1, 1, 0, 15, 0),
+            finished_at=datetime(2024, 1, 1, 0, 15, 5),
+            id=1,
+        )
+        # Act
+        output = to_run_log_output(log)
+        # Assert
+        assert output.task_id == "task-001"
+        assert output.run_id == "run-001"
+        assert output.triggered_by == "auto"
+        assert output.status == "success"
+        assert output.output == {"rows": 10}
+        assert output.id == 1
+
+    def test_datetime_fields_formatted_as_iso(self):
+        # Arrange - naive datetime 按 UTC 处理（与 DB 存储约定一致）
+        started = datetime(2024, 1, 1, 0, 15, 0)
+        log = ScheduledTaskRunLog(
+            task_id="t",
+            run_id="r",
+            started_at=started,
+        )
+        # Act
+        output = to_run_log_output(log)
+        # Assert - naive UTC 不偏移
+        assert isinstance(output.started_at, str)
+        assert output.started_at == "2024-01-01T00:15:00Z"
+
+    def test_none_datetime_fields_return_none(self):
+        log = ScheduledTaskRunLog(task_id="t", run_id="r")
+        output = to_run_log_output(log)
+        assert output.started_at is None
+        assert output.finished_at is None
+
+    def test_none_output_returns_none(self):
+        # Arrange - output 为 None 时 mapper 透传 None
+        log = ScheduledTaskRunLog(task_id="t", run_id="r", output=None)
+        # Act
+        output = to_run_log_output(log)
+        # Assert
+        assert output.output is None
+
+    def test_none_error_message_returns_none(self):
+        # Arrange - error_message 为 None 时 mapper 透传 None
+        log = ScheduledTaskRunLog(task_id="t", run_id="r", error_message=None)
+        # Act
+        output = to_run_log_output(log)
+        # Assert
+        assert output.error_message is None
+
+    def test_all_fields_mapped(self):
+        # Arrange - 验证全字段映射
+        log = ScheduledTaskRunLog(
+            task_id="t1",
+            run_id="r1",
+            triggered_by="manual",
+            status="failure",
+            error_message="boom",
+            output={"error": True},
+            started_at=datetime(2024, 1, 1, 0, 15, 0),
+            finished_at=datetime(2024, 1, 1, 0, 15, 5),
+            created_by="admin",
+            updated_by="operator",
+            id=42,
+            created_at=datetime(2024, 1, 1, 0, 0, 0),
+            updated_at=datetime(2024, 1, 1, 0, 0, 1),
+            is_deleted=0,
+            deleted_at=None,
+        )
+        # Act
+        output = to_run_log_output(log)
+        # Assert
+        assert output.task_id == "t1"
+        assert output.run_id == "r1"
+        assert output.triggered_by == "manual"
+        assert output.status == "failure"
+        assert output.error_message == "boom"
+        assert output.output == {"error": True}
+        assert output.created_by == "admin"
+        assert output.updated_by == "operator"
+        assert output.id == 42
+        assert output.is_deleted == 0
+        assert output.deleted_at is None
+
+
+# ─── to_handler_summary_output ────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestToHandlerSummaryOutput:
+    def test_basic_mapping(self):
+        # Arrange
+        summary = HandlerSummary(
+            name="my_handler",
+            task_count=5,
+            last_active_at=datetime(2024, 1, 1, 0, 0, 0),
+        )
+        # Act
+        output = to_handler_summary_output(summary)
+        # Assert
+        assert output.handler_name == "my_handler"
+        assert output.task_count == 5
+        assert isinstance(output.last_active_at, str)
+
+    def test_none_last_active_at(self):
+        summary = HandlerSummary(name="h", task_count=0, last_active_at=None)
+        output = to_handler_summary_output(summary)
+        assert output.last_active_at is None
+
+    def test_description_defaults_empty(self):
+        # Arrange - description 由调用方按需补充，mapper 不聚合
+        summary = HandlerSummary(name="h", task_count=0)
+        # Act
+        output = to_handler_summary_output(summary)
+        # Assert
+        assert output.description == ""
+
+    def test_active_task_count_defaults_zero(self):
+        summary = HandlerSummary(name="h", task_count=0)
+        output = to_handler_summary_output(summary)
+        assert output.active_task_count == 0
+
+
+# ─── to_list_handler_summary_output ───────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestToListHandlerSummaryOutput:
+    def test_empty_list(self):
+        output = to_list_handler_summary_output([])
+        assert output.items == []
+
+    def test_multiple_items(self):
+        summaries = [
+            HandlerSummary(name="h1", task_count=1),
+            HandlerSummary(name="h2", task_count=2),
+        ]
+        output = to_list_handler_summary_output(summaries)
+        assert len(output.items) == 2
+        assert output.items[0].handler_name == "h1"
+        assert output.items[1].handler_name == "h2"
+
+
+# ─── to_count_by_status_output ────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestToCountByStatusOutput:
+    def test_all_statuses_present(self):
+        # Arrange
+        dist = {"active": 10, "paused": 5, "dead_letter": 2}
+        # Act
+        output = to_count_by_status_output(dist)
+        # Assert
+        assert output.active == 10
+        assert output.paused == 5
+        assert output.dead_letter == 2
+
+    def test_missing_statuses_default_zero(self):
+        # Arrange - 仓储层未出现的状态不在字典中
+        dist = {"active": 10}
+        # Act
+        output = to_count_by_status_output(dist)
+        # Assert
+        assert output.active == 10
+        assert output.paused == 0
+        assert output.dead_letter == 0
+
+    def test_empty_dict(self):
+        output = to_count_by_status_output({})
+        assert output.active == 0
+        assert output.paused == 0
+        assert output.dead_letter == 0
+
+
+# ─── to_get_health_output ─────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestToGetHealthOutput:
+    def test_healthy_status(self):
+        # Arrange
+        last_tick = datetime(2024, 1, 1, 0, 0, 0)
+        # Act
+        output = to_get_health_output(
+            last_tick_at=last_tick,
+            active_task_count=10,
+            dead_letter_count=2,
+            running_count=3,
+            healthy=True,
+        )
+        # Assert
+        assert output.status == "healthy"
+        assert output.active_task_count == 10
+        assert output.dead_letter_count == 2
+        assert output.running_count == 3
+        assert isinstance(output.last_tick_at, str)
+
+    def test_unhealthy_status(self):
+        output = to_get_health_output(
+            last_tick_at=None,
+            active_task_count=0,
+            dead_letter_count=0,
+            running_count=0,
+            healthy=False,
+        )
+        assert output.status == "unhealthy"
+        assert output.last_tick_at is None
+
+
+# ─── daily_stat_to_dict ───────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestDailyStatToDict:
+    def test_basic_mapping(self):
+        # Arrange
+        stat = DailyStat(
+            stat_date="2024-01-01",
+            handler_name="h",
+            success_count=10,
+            failure_count=2,
+            timeout_count=1,
+            dead_letter_count=0,
+        )
+        # Act
+        result = daily_stat_to_dict(stat)
+        # Assert
+        assert result["stat_date"] == "2024-01-01"
+        assert result["handler_name"] == "h"
+        assert result["success_count"] == 10
+        assert result["failure_count"] == 2
+        assert result["timeout_count"] == 1
+        assert result["dead_letter_count"] == 0
+
+
+# ─── to_daily_stat_output ─────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestToDailyStatOutput:
+    def test_basic_mapping_with_derived_fields(self):
+        # Arrange
+        stat = DailyStat(
+            stat_date=datetime(2024, 1, 1).date(),
+            handler_name="h",
+            success_count=10,
+            failure_count=2,
+            timeout_count=1,
+            dead_letter_count=0,
+        )
+        # Act
+        output = to_daily_stat_output(stat)
+        # Assert
+        assert output.stat_date == "2024-01-01"
+        assert output.handler_name == "h"
+        assert output.success_count == 10
+        assert output.failure_count == 2
+        assert output.timeout_count == 1
+        assert output.dead_letter_count == 0
+        # 派生字段：total = 10 + 2 + 1 = 13
+        assert output.total_count == 13
+        # success_rate = 10 / 13
+        assert output.success_rate == pytest.approx(10 / 13)
+
+    def test_zero_total_count_success_rate_zero(self):
+        # Arrange - 所有计数为 0
+        stat = DailyStat(
+            stat_date=datetime(2024, 1, 1).date(),
+            handler_name="h",
+            success_count=0,
+            failure_count=0,
+            timeout_count=0,
+            dead_letter_count=0,
+        )
+        # Act
+        output = to_daily_stat_output(stat)
+        # Assert
+        assert output.total_count == 0
+        assert output.success_rate == 0.0
+
+    def test_dead_letter_count_excluded_from_total(self):
+        # Arrange - dead_letter_count 不计入 total_count
+        stat = DailyStat(
+            stat_date=datetime(2024, 1, 1).date(),
+            handler_name="h",
+            success_count=5,
+            failure_count=2,
+            timeout_count=1,
+            dead_letter_count=3,
+        )
+        # Act
+        output = to_daily_stat_output(stat)
+        # Assert - total = 5 + 2 + 1 = 8（不含 dead_letter）
+        assert output.total_count == 8
+        assert output.success_rate == pytest.approx(5 / 8)
+
+
+# ─── to_list_daily_stats_output ───────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestToListDailyStatsOutput:
+    def test_empty_list(self):
+        output = to_list_daily_stats_output([])
+        assert output.items == []
+
+    def test_multiple_items(self):
+        stats = [
+            DailyStat(
+                stat_date=datetime(2024, 1, 1).date(),
+                handler_name="h1",
+                success_count=1,
+                failure_count=0,
+                timeout_count=0,
+                dead_letter_count=0,
+            ),
+            DailyStat(
+                stat_date=datetime(2024, 1, 2).date(),
+                handler_name="h2",
+                success_count=2,
+                failure_count=0,
+                timeout_count=0,
+                dead_letter_count=0,
+            ),
+        ]
+        output = to_list_daily_stats_output(stats)
+        assert len(output.items) == 2
+        assert output.items[0].handler_name == "h1"
+        assert output.items[1].handler_name == "h2"
+
+
+# ─── to_list_upcoming_output ──────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestToListUpcomingOutput:
+    def test_empty_list(self):
+        output = to_list_upcoming_output([])
+        assert output.items == []
+
+    def test_multiple_tasks(self):
+        tasks = [
+            ScheduledTask(
+                task_id="t1",
+                handler_name="h",
+                owner_scope="s",
+                owner_id="o",
+                schedule_kind="cron",
+            ),
+            ScheduledTask(
+                task_id="t2",
+                handler_name="h",
+                owner_scope="s",
+                owner_id="o",
+                schedule_kind="cron",
+            ),
+        ]
+        output = to_list_upcoming_output(tasks)
+        assert len(output.items) == 2
+        assert output.items[0].task_id == "t1"
+        assert output.items[1].task_id == "t2"

backend/test/unit/scheduler/use_cases/utils/test_backoff.py

@@ -0,0 +1,134 @@
+"""指数退避计算工具单元测试。
+
+覆盖 ``yuxi.scheduler.use_cases.utils.backoff`` 模块：
+- ``calc_backoff_seconds``：根据连续错误数从退避序列取退避秒数
+- ``calc_backoff_next_run_at``：计算退避后的下次运行时间
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+
+import pytest
+
+from yuxi.scheduler.use_cases.utils.backoff import (
+    calc_backoff_next_run_at,
+    calc_backoff_seconds,
+)
+
+
+pytestmark = pytest.mark.unit
+
+
+# ─── calc_backoff_seconds ─────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestCalcBackoffSeconds:
+    @pytest.mark.parametrize(
+        ("errors", "expected"),
+        [
+            (0, 0),    # 无错误无退避
+            (-1, 0),   # 负数视为无错误
+            (-100, 0), # 大负数同样视为无错误
+        ],
+        ids=["zero", "negative-one", "large-negative"],
+    )
+    def test_non_positive_errors_return_zero(self, errors: int, expected: int):
+        schedule = [10, 30, 120, 600, 3600]
+        assert calc_backoff_seconds(errors, schedule) == expected
+
+    @pytest.mark.parametrize(
+        ("errors", "expected"),
+        [
+            (1, 10),    # 第 1 次错误取 schedule[0]
+            (2, 30),    # 第 2 次错误取 schedule[1]
+            (3, 120),   # 第 3 次错误取 schedule[2]
+            (4, 600),   # 第 4 次错误取 schedule[3]
+            (5, 3600),  # 第 5 次错误取 schedule[4]
+        ],
+        ids=["first", "second", "third", "fourth", "fifth"],
+    )
+    def test_error_count_within_schedule_returns_indexed_value(
+        self, errors: int, expected: int
+    ):
+        schedule = [10, 30, 120, 600, 3600]
+        assert calc_backoff_seconds(errors, schedule) == expected
+
+    @pytest.mark.parametrize(
+        ("errors", "expected"),
+        [
+            (3, 120),   # 边界：consecutive_errors == len(schedule)
+            (10, 120),  # 超过序列长度取最后一个值
+            (100, 120), # 远超序列长度仍取最后一个值
+        ],
+        ids=["at-length", "exceeds", "far-exceeds"],
+    )
+    def test_error_count_exceeding_schedule_caps_at_last(
+        self, errors: int, expected: int
+    ):
+        schedule = [10, 30, 120]
+        assert calc_backoff_seconds(errors, schedule) == expected
+
+    def test_empty_schedule_returns_zero(self):
+        # Arrange - 空退避序列
+        schedule: list[int] = []
+        # Act
+        result = calc_backoff_seconds(5, schedule)
+        # Assert
+        assert result == 0
+
+    def test_single_element_schedule(self):
+        # Arrange - 单元素序列
+        schedule = [60]
+        # Act & Assert
+        assert calc_backoff_seconds(1, schedule) == 60
+        assert calc_backoff_seconds(5, schedule) == 60
+
+
+# ─── calc_backoff_next_run_at ─────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestCalcBackoffNextRunAt:
+    @pytest.mark.parametrize(
+        ("errors", "expected_offset"),
+        [
+            (0, 0),    # 无错误时退避秒数为 0，返回 base_time
+            (1, 10),   # 第 1 次错误加 10 秒
+            (3, 120),  # 第 3 次错误加 120 秒
+            (100, 120), # 超过序列长度封顶为 120 秒
+        ],
+        ids=["zero-errors", "first-error", "third-error", "exceeds-cap"],
+    )
+    def test_backoff_next_run_at_offsets(
+        self, errors: int, expected_offset: int
+    ):
+        # Arrange
+        base = datetime(2024, 1, 1, 0, 0, 0)
+        schedule = [10, 30, 120]
+        # Act
+        result = calc_backoff_next_run_at(errors, schedule, base_time=base)
+        # Assert
+        assert result == base + timedelta(seconds=expected_offset)
+
+    def test_default_base_time_is_recent(self, monkeypatch: pytest.MonkeyPatch):
+        # Arrange - mock utc_now_naive 返回固定时间
+        fixed_now = datetime(2024, 6, 1, 12, 0, 0)
+        import yuxi.scheduler.use_cases.utils.backoff as backoff_mod
+
+        monkeypatch.setattr(backoff_mod, "utc_now_naive", lambda: fixed_now)
+        schedule = [60]
+        # Act
+        result = calc_backoff_next_run_at(1, schedule)
+        # Assert
+        assert result == fixed_now + timedelta(seconds=60)
+
+    def test_empty_schedule_returns_base_time(self):
+        # Arrange
+        base = datetime(2024, 1, 1, 0, 0, 0)
+        schedule: list[int] = []
+        # Act
+        result = calc_backoff_next_run_at(5, schedule, base_time=base)
+        # Assert
+        assert result == base

backend/test/unit/scheduler/use_cases/utils/test_cron_calc.py

@@ -0,0 +1,157 @@
+"""cron 下次触发时间计算工具单元测试。
+
+覆盖 ``yuxi.scheduler.use_cases.utils.cron_calc`` 模块的 ``calc_next_run_at`` 函数：
+- 无时区参数：base_time 作为 naive datetime 直接交给 croniter
+- 有时区参数：按该时区解释 cron 表达式，结果返回 UTC naive datetime
+- 默认 base_time：``utc_now_naive()``
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from yuxi.scheduler.use_cases.utils.cron_calc import calc_next_run_at
+
+
+pytestmark = pytest.mark.unit
+
+
+# ─── calc_next_run_at（无时区） ────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestCalcNextRunAtNoTimezone:
+    def test_daily_cron_returns_next_day(self):
+        # Arrange - 每天凌晨执行
+        base = datetime(2024, 1, 1, 12, 0, 0)
+        expr = "0 0 * * *"
+        # Act
+        result = calc_next_run_at(expr, base_time=base)
+        # Assert - 下一次执行为 2024-01-02 00:00
+        assert result == datetime(2024, 1, 2, 0, 0, 0)
+
+    def test_hourly_cron_returns_next_hour(self):
+        # Arrange
+        base = datetime(2024, 1, 1, 12, 30, 0)
+        expr = "0 * * * *"
+        # Act
+        result = calc_next_run_at(expr, base_time=base)
+        # Assert
+        assert result == datetime(2024, 1, 1, 13, 0, 0)
+
+    def test_every_15_minutes(self):
+        # Arrange
+        base = datetime(2024, 1, 1, 12, 5, 0)
+        expr = "*/15 * * * *"
+        # Act
+        result = calc_next_run_at(expr, base_time=base)
+        # Assert
+        assert result == datetime(2024, 1, 1, 12, 15, 0)
+
+    def test_specific_time_today_if_future(self):
+        # Arrange - base 时间在指定时刻之前，下一次执行为当天
+        base = datetime(2024, 1, 1, 8, 0, 0)
+        expr = "30 10 * * *"
+        # Act
+        result = calc_next_run_at(expr, base_time=base)
+        # Assert
+        assert result == datetime(2024, 1, 1, 10, 30, 0)
+
+    def test_specific_time_tomorrow_if_past(self):
+        # Arrange - base 时间在指定时刻之后，下一次执行为次日
+        base = datetime(2024, 1, 1, 12, 0, 0)
+        expr = "30 10 * * *"
+        # Act
+        result = calc_next_run_at(expr, base_time=base)
+        # Assert
+        assert result == datetime(2024, 1, 2, 10, 30, 0)
+
+    def test_returns_naive_datetime(self):
+        # Arrange
+        base = datetime(2024, 1, 1, 12, 0, 0)
+        # Act
+        result = calc_next_run_at("0 * * * *", base_time=base)
+        # Assert - 返回 naive datetime（与 DB 字段存储约定一致）
+        assert result.tzinfo is None
+
+    def test_default_base_time_is_recent(self, monkeypatch: pytest.MonkeyPatch):
+        # Arrange - mock utc_now_naive 返回固定时间
+        fixed_now = datetime(2024, 6, 1, 12, 0, 0)
+        import yuxi.scheduler.use_cases.utils.cron_calc as cron_mod
+
+        monkeypatch.setattr(cron_mod, "utc_now_naive", lambda: fixed_now)
+        # Act
+        result = calc_next_run_at("0 * * * *")
+        # Assert
+        assert result == datetime(2024, 6, 1, 13, 0, 0)
+
+
+# ─── calc_next_run_at（有时区） ────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestCalcNextRunAtWithTimezone:
+    def test_tz_returns_utc_naive(self):
+        # Arrange - Asia/Shanghai 时区
+        base = datetime(2024, 1, 1, 12, 0, 0)  # UTC 12:00 = Shanghai 20:00
+        expr = "0 0 * * *"  # 每天 00:00 Shanghai 时间
+        # Act
+        result = calc_next_run_at(expr, base_time=base, tz="Asia/Shanghai")
+        # Assert - Shanghai 00:00 = UTC 16:00 前一天，但 base 是 1/1 12:00 UTC，
+        # 下一次 Shanghai 00:00 是 1/2 00:00 Shanghai = 1/1 16:00 UTC
+        assert result == datetime(2024, 1, 1, 16, 0, 0)
+        assert result.tzinfo is None
+
+    def test_tz_hourly_cron(self):
+        # Arrange - base UTC 12:30 = Shanghai 20:30
+        base = datetime(2024, 1, 1, 12, 30, 0)
+        expr = "0 * * * *"  # 每小时整点 Shanghai 时间
+        # Act
+        result = calc_next_run_at(expr, base_time=base, tz="Asia/Shanghai")
+        # Assert - 下一次 Shanghai 整点是 21:00 = UTC 13:00
+        assert result == datetime(2024, 1, 1, 13, 0, 0)
+
+    def test_tz_utc_zero_offset(self):
+        # Arrange - UTC 时区，与无时区行为一致
+        base = datetime(2024, 1, 1, 12, 0, 0)
+        expr = "0 * * * *"
+        # Act
+        result_no_tz = calc_next_run_at(expr, base_time=base)
+        result_utc = calc_next_run_at(expr, base_time=base, tz="UTC")
+        # Assert
+        assert result_no_tz == result_utc
+
+    def test_tz_default_base_time(self, monkeypatch: pytest.MonkeyPatch):
+        # Arrange
+        fixed_now = datetime(2024, 6, 1, 12, 0, 0)
+        import yuxi.scheduler.use_cases.utils.cron_calc as cron_mod
+
+        monkeypatch.setattr(cron_mod, "utc_now_naive", lambda: fixed_now)
+        # Act
+        result = calc_next_run_at("0 * * * *", tz="Asia/Shanghai")
+        # Assert - 返回 naive datetime
+        assert result.tzinfo is None
+        # Shanghai 20:00 = UTC 12:00，下一次 Shanghai 整点 21:00 = UTC 13:00
+        assert result == datetime(2024, 6, 1, 13, 0, 0)
+
+
+# ─── 异常场景 ─────────────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestCalcNextRunAtErrors:
+    def test_invalid_cron_raises_value_error(self):
+        # Arrange
+        base = datetime(2024, 1, 1, 12, 0, 0)
+        # Act & Assert - croniter 抛出 ValueError
+        with pytest.raises(ValueError):
+            calc_next_run_at("invalid", base_time=base)
+
+    def test_invalid_tz_raises_key_error_or_value_error(self):
+        # Arrange - 非法时区
+        base = datetime(2024, 1, 1, 12, 0, 0)
+        # Act & Assert - ZoneInfo 抛出 KeyError 或 ValueError
+        with pytest.raises((KeyError, ValueError)):
+            calc_next_run_at("0 * * * *", base_time=base, tz="Invalid/Timezone")

backend/test/unit/scheduler/use_cases/utils/test_stagger.py

@@ -0,0 +1,111 @@
+"""整点抖动计算工具单元测试。
+
+覆盖 ``yuxi.scheduler.use_cases.utils.stagger`` 模块的 ``calc_stagger_seconds`` 函数：
+- 基于 task_id 哈希计算固定抖动值
+- 同一 task_id 多次计算得到相同结果（固定性）
+- 抖动值范围 ``[0, max_seconds]``
+"""
+
+from __future__ import annotations
+
+import hashlib
+
+import pytest
+
+from yuxi.scheduler.use_cases.utils.stagger import calc_stagger_seconds
+
+
+pytestmark = pytest.mark.unit
+
+
+# ─── 固定性 ───────────────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestStaggerDeterminism:
+    def test_same_task_id_returns_same_value(self):
+        # Arrange
+        task_id = "task-001"
+        # Act
+        result1 = calc_stagger_seconds(task_id, max_seconds=300)
+        result2 = calc_stagger_seconds(task_id, max_seconds=300)
+        # Assert - 同一 task_id 多次计算得到相同结果
+        assert result1 == result2
+
+    def test_different_task_ids_may_return_different_values(self):
+        # Arrange - 两个不同 task_id，至少有一个不同（哈希分布）
+        # 使用足够大的 max_seconds 确保差异可见
+        task_id1 = "task-001"
+        task_id2 = "task-002"
+        # Act
+        result1 = calc_stagger_seconds(task_id1, max_seconds=300)
+        result2 = calc_stagger_seconds(task_id2, max_seconds=300)
+        # Assert - 大概率不同（哈希分布均匀）
+        assert result1 != result2
+
+
+# ─── 范围 ─────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestStaggerRange:
+    @pytest.mark.parametrize(
+        "max_seconds",
+        [1, 10, 100, 300, 3600],
+        ids=["max-1", "max-10", "max-100", "max-300", "max-3600"],
+    )
+    def test_result_within_range_for_various_max_seconds(self, max_seconds: int):
+        # Act & Assert - 多个 task_id 在不同 max_seconds 下都应在范围内
+        for i in range(50):
+            result = calc_stagger_seconds(f"task-{i:03d}", max_seconds=max_seconds)
+            assert 0 <= result <= max_seconds
+
+    def test_result_within_range_for_many_task_ids(self):
+        # Arrange - 多个 task_id 都应在范围内
+        max_seconds = 100
+        # Act & Assert
+        for i in range(100):
+            result = calc_stagger_seconds(f"task-{i:03d}", max_seconds=max_seconds)
+            assert 0 <= result <= max_seconds
+
+    @pytest.mark.parametrize(
+        ("max_seconds", "valid_values"),
+        [
+            (0, {0}),       # max_seconds=0 时模 1，结果必为 0
+            (1, {0, 1}),    # max_seconds=1 时模 2，结果为 0 或 1
+        ],
+        ids=["max-zero", "max-one"],
+    )
+    def test_boundary_max_seconds(self, max_seconds: int, valid_values: set[int]):
+        # Act & Assert
+        for i in range(50):
+            result = calc_stagger_seconds(f"task-{i}", max_seconds=max_seconds)
+            assert result in valid_values
+
+
+# ─── 哈希算法一致性 ───────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+class TestStaggerHashAlgorithm:
+    def test_uses_md5_hash_of_task_id(self):
+        # Arrange - 手动计算 MD5 哈希取模，验证实现与文档一致
+        task_id = "task-001"
+        max_seconds = 300
+        expected_digest = hashlib.md5(task_id.encode("utf-8")).hexdigest()
+        expected = int(expected_digest, 16) % (max_seconds + 1)
+        # Act
+        result = calc_stagger_seconds(task_id, max_seconds=max_seconds)
+        # Assert
+        assert result == expected
+
+    def test_unicode_task_id_handled(self):
+        # Arrange - task_id 含非 ASCII 字符
+        task_id = "任务-001"
+        max_seconds = 300
+        expected_digest = hashlib.md5(task_id.encode("utf-8")).hexdigest()
+        expected = int(expected_digest, 16) % (max_seconds + 1)
+        # Act
+        result = calc_stagger_seconds(task_id, max_seconds=max_seconds)
+        # Assert
+        assert result == expected