fix(scheduler): address !7 REQUEST_CHANGES

- seed: check completed_key before no_source so same-day last_completed
  is already_completed (not 10m retry)
- scheduler: resolve TaskControl ok/error before logging; ok=False logs
  task_fail, not task_ok
- candidate_cleanup: catch failures, record task, retry ~5m like discover

DEVELOPMENT.md

@@ -457,7 +457,8 @@ P1 实现说明：
 - 时区使用固定 UTC+8（中国无夏令时），避免 Windows 缺少 `tzdata`；
 - 午夜 `seed_at`（默认 `00:00`）把 `snapshot:last_completed` 的 ok=1 行克隆为当日壳，写入 `snapshot:seeded:<date>=1` / `fans:seeded:<date>=1`，**不**推进完成日；
 - 也可手动：`python app.py snapshot-seed [--date YYYY-MM-DD]`（Debian / `docker compose exec re-xxm python app.py snapshot-seed`）；
-- 若全站 `snapshot` 正在跑，或 `last_completed` 暂不可用（`no_source`），seed **不记成功**，约 10 分钟后重试直到真正克隆成功；
+- 目标日已有 `snapshot:completed:<date>` 时 seed 记 `already_completed` 成功（即使 `last_completed` 已等于当日），**不得**误判 `no_source`；
+- 若全站 `snapshot` 正在跑，或目标日未完成且 `last_completed` 暂不可用（`no_source`），seed **不记成功**，约 10 分钟后重试直到真正克隆成功；
 - 晚间 `snapshot_at`（默认 `23:15`）对 seeded 日 force 重采；成功后写 `snapshot:completed:<date>=1`、推进 `last_completed` 并清除 seeded；
 - 当日全部视频与粉丝采集成功后在 `kv` 写入 `snapshot:completed:<date>=1`；启动时未完成会补跑（纯 seed 日不误标完成）；
 - 完成标记不是唯一门槛：每次检查 `pending_snapshot_aids(date)`，新入库或从 `dropped` 恢复的视频会立即补采当天快照；

reaction_board/cli.py

@@ -730,10 +730,20 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
                 return _quiet_window_defer(
                     repo, logger, "candidate_cleanup", started_at=started,
                 )
+            try:
                 removed = repo.delete_expired_candidates(
                     ttl_days=cfg.schedule.candidate_ttl_days
                 )
                 expired = repo.expire_fav_actions()
+            except Exception as exc:  # noqa: BLE001  记录后约 5 分钟补跑
+                finished = int(time.time())
+                _record_task(
+                    repo, "candidate_cleanup", ok=False, started_at=started,
+                    finished_at=finished, error=str(exc),
+                )
+                return TaskControl(
+                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
+                )
             if removed or expired:
                 repo.bump_cache_version("videos")
             finished = int(time.time())

reaction_board/scheduler.py

@@ -87,11 +87,9 @@ class Scheduler:
                 raw = task.fn()
             except Exception as exc:  # noqa: BLE001  单任务失败不终止调度
                 error = str(exc)
-                self._log("error", "scheduler.task_error", task=task.name, error=error)
             else:
                 if isinstance(raw, TaskControl):
                     control = raw
-                self._log("info", "scheduler.task_ok", task=task.name)
             finished = self._clock()
             if control is not None and control.next_run is not None:
                 task.next_run = float(control.next_run)
@@ -104,6 +102,23 @@ class Scheduler:
                 ok = bool(control.ok) and error is None
                 if control.error and error is None:
                     error = control.error
+            # Skip/fail ≠ success：先定 ok/error，再记 scheduler 日志。
+            if ok:
+                self._log("info", "scheduler.task_ok", task=task.name)
+            elif control is not None:
+                self._log(
+                    "warning",
+                    "scheduler.task_fail",
+                    task=task.name,
+                    error=error,
+                )
+            else:
+                self._log(
+                    "error",
+                    "scheduler.task_error",
+                    task=task.name,
+                    error=error,
+                )
             result = TaskResult(
                 name=task.name,
                 started_at=started,

reaction_board/services/snapshot.py

@@ -538,16 +538,8 @@ class SnapshotService:
             raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}")
 
         summary = SnapshotSummary(stat_date=stat_date, started_at=now)
-        source = self.repo.kv_get("snapshot:last_completed")
-        if not source or source >= stat_date:
-            self._log(
-                "info", "snapshot.seed_skip",
-                stat_date=stat_date, reason="no_source", source=source,
-            )
-            summary.finished_at = self._now()
-            summary.completed = False
-            summary.reason = "no_source"
-            return summary
+        # 目标日已完成：先于 no_source。全站快照当天完成后
+        # last_completed == stat_date，不可误判为无源再 10 分钟重试。
         if self.repo.kv_get(self.completed_key(stat_date)) == "1":
             self._log(
                 "info", "snapshot.seed_skip",
@@ -558,6 +550,16 @@ class SnapshotService:
             summary.already_completed = True
             summary.reason = "already_completed"
             return summary
+        source = self.repo.kv_get("snapshot:last_completed")
+        if not source or source >= stat_date:
+            self._log(
+                "info", "snapshot.seed_skip",
+                stat_date=stat_date, reason="no_source", source=source,
+            )
+            summary.finished_at = self._now()
+            summary.completed = False
+            summary.reason = "no_source"
+            return summary
         if self.repo.kv_get(self.seeded_key(stat_date)) == "1":
             self._log(
                 "info", "snapshot.seed_skip",

tests/test_cli.py

@@ -479,6 +479,115 @@ def test_defer_snapshot_seed_schedules_ten_minutes():
             db.close()
 
 
+
+
+def test_snapshot_seed_same_day_completed_succeeds():
+    """当日已完成时手动 snapshot-seed 应成功 already_completed，不走 no_source。"""
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_config(tmp, cookie="", csrf="")
+        code, _ = run_cli(["init", "--config", str(path)])
+        assert code == 0
+        today = "2026-10-31"
+        db = Database(Path(tmp) / "app.db")
+        db.initialize()
+        try:
+            repo = Repository(db)
+            repo.kv_set("snapshot:last_completed", today)
+            repo.kv_set("snapshot:completed:" + today, "1")
+        finally:
+            db.close()
+        code, output = run_cli([
+            "snapshot-seed", "--config", str(path), "--date", today,
+        ])
+        assert code == 0, output
+        assert '"reason": "already_completed"' in output
+        assert "no_source" not in output
+        db = Database(Path(tmp) / "app.db")
+        db.initialize()
+        try:
+            repo = Repository(db)
+            last = json.loads(repo.kv_get("task:last:snapshot_seed"))
+            assert last["ok"] is True
+        finally:
+            db.close()
+
+
+def test_candidate_cleanup_failure_retries_soon():
+    """candidate_cleanup 异常应记失败任务并约 5 分钟补跑，而非落到 86400 间隔。"""
+    class FakeClient:
+        def check_login(self):
+            return LoginState(True, 1, "tester", 0, "OK")
+
+    class FakeDiscovery:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def run(self):
+            from reaction_board.services.discovery import DiscoverySummary
+            return DiscoverySummary(started_at=1, finished_at=1)
+
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_config(tmp)
+        data = json.loads(path.read_text(encoding="utf-8"))
+        data["folders"] = []
+        data["track_fans"] = []
+        path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+
+        original_client = cli._client
+        original_discovery = cli.DiscoveryService
+        original_delete = Repository.delete_expired_candidates
+        captured = {}
+
+        def boom(self, *args, **kwargs):
+            raise RuntimeError("cleanup db boom")
+
+        def fake_client(credentials, *args, **kwargs):
+            return FakeClient()
+
+        # 只跑 cleanup：把其它任务 next_run 推远；通过 Scheduler 包装拦截。
+        original_scheduler = cli.Scheduler
+
+        class CaptureScheduler(original_scheduler):
+            def __init__(self, tasks, **kwargs):
+                far = 1_000_000_000_000.0
+                for task in tasks:
+                    if task.name != "candidate_cleanup":
+                        task.next_run = far
+                    else:
+                        task.next_run = 0.0
+                super().__init__(tasks, **kwargs)
+                captured["scheduler"] = self
+
+        cli._client = fake_client
+        cli.DiscoveryService = FakeDiscovery
+        cli.Scheduler = CaptureScheduler
+        Repository.delete_expired_candidates = boom
+        try:
+            code, output = run_cli(["run", "--once", "--config", str(path)])
+        finally:
+            cli._client = original_client
+            cli.DiscoveryService = original_discovery
+            cli.Scheduler = original_scheduler
+            Repository.delete_expired_candidates = original_delete
+
+        assert code == 0, output
+        db = Database(Path(tmp) / "app.db")
+        db.initialize()
+        try:
+            repo = Repository(db)
+            last = json.loads(repo.kv_get("task:last:candidate_cleanup"))
+            assert last["ok"] is False
+            assert "cleanup db boom" in last["error"]
+        finally:
+            db.close()
+        scheduler = captured["scheduler"]
+        cleanup_task = next(t for t in scheduler.tasks if t.name == "candidate_cleanup")
+        # next_run ≈ finished + 300；允许几秒时钟误差
+        assert cleanup_task.next_run > 0
+        delta = cleanup_task.next_run - __import__("time").time()
+        assert 250 <= delta <= 360, delta
+
+
 def main_test():
     test_init_and_status()
     test_missing_config()
@@ -496,7 +605,9 @@ def main_test():
     test_snapshot_seed_cli_clones_offline()
     test_snapshot_seed_defers_when_snapshot_running()
     test_snapshot_seed_no_source_is_non_success()
+    test_snapshot_seed_same_day_completed_succeeds()
     test_defer_snapshot_seed_schedules_ten_minutes()
+    test_candidate_cleanup_failure_retries_soon()
     print("OK: CLI init/sync/snapshot/snapshot-seed/auth-check/run/status/backfill/错误路径测试全部通过")
 
 

tests/test_scheduler.py

@@ -152,10 +152,87 @@ def test_task_control_overrides_ok_and_next_run():
     assert task.next_run == 2_000.0
 
 
+
+
+class _CaptureLogger:
+    def __init__(self):
+        self.events = []
+
+    def info(self, event, **fields):
+        self.events.append(("info", event, fields))
+
+    def warning(self, event, **fields):
+        self.events.append(("warning", event, fields))
+
+    def error(self, event, **fields):
+        self.events.append(("error", event, fields))
+
+
+def test_task_control_ok_false_logs_fail_not_ok():
+    """TaskControl(ok=False) 不得记 scheduler.task_ok（Skip ≠ success）。"""
+    clock = [1_000.0]
+    logger = _CaptureLogger()
+    task = ScheduledTask(
+        "quiet",
+        60,
+        lambda: TaskControl(ok=False, next_run=1_500.0, error="snapshot_quiet_window"),
+    )
+    scheduler = Scheduler([task], clock=lambda: clock[0], logger=logger)
+    results = scheduler.run_due()
+    assert results[0].ok is False
+    assert results[0].error == "snapshot_quiet_window"
+    assert ("info", "scheduler.task_ok", {"task": "quiet"}) not in logger.events
+    fail_events = [e for e in logger.events if e[1] == "scheduler.task_fail"]
+    assert len(fail_events) == 1
+    assert fail_events[0][0] == "warning"
+    assert fail_events[0][2]["task"] == "quiet"
+    assert fail_events[0][2]["error"] == "snapshot_quiet_window"
+
+
+def test_task_control_ok_true_still_logs_task_ok():
+    clock = [1_000.0]
+    logger = _CaptureLogger()
+    task = ScheduledTask(
+        "ok_task",
+        60,
+        lambda: TaskControl(ok=True, next_run=1_200.0),
+    )
+    scheduler = Scheduler([task], clock=lambda: clock[0], logger=logger)
+    results = scheduler.run_due()
+    assert results[0].ok is True
+    assert ("info", "scheduler.task_ok", {"task": "ok_task"}) in logger.events
+    assert not any(e[1] == "scheduler.task_fail" for e in logger.events)
+
+
+def test_raised_exception_logs_task_fail():
+    clock = [100.0]
+    logger = _CaptureLogger()
+
+    def boom():
+        raise RuntimeError("db down")
+
+    scheduler = Scheduler(
+        [ScheduledTask("bad", 10, boom)],
+        clock=lambda: clock[0],
+        logger=logger,
+    )
+    results = scheduler.run_due()
+    assert results[0].ok is False
+    assert results[0].error == "db down"
+    assert not any(e[1] == "scheduler.task_ok" for e in logger.events)
+    assert not any(e[1] == "scheduler.task_fail" for e in logger.events)
+    err_events = [e for e in logger.events if e[1] == "scheduler.task_error"]
+    assert len(err_events) == 1
+    assert err_events[0][0] == "error"
+    assert err_events[0][2]["error"] == "db down"
+
 def main():
     test_run_when_due_and_reschedule()
     test_next_run_fn_controls_reschedule()
     test_task_control_overrides_ok_and_next_run()
+    test_task_control_ok_false_logs_fail_not_ok()
+    test_task_control_ok_true_still_logs_task_ok()
+    test_raised_exception_logs_task_fail()
     test_task_error_does_not_block_other_tasks()
     test_run_forever_with_stop_event()
     test_max_iterations_and_validation()

tests/test_snapshot.py

@@ -537,6 +537,36 @@ def test_full_snapshot_skips_ok_rows_including_hourly_refresh():
 
 
 
+def test_seed_same_day_completed_is_already_completed_not_no_source():
+    """全站完成后 last_completed==target：应 already_completed，不误判 no_source。"""
+    clock = 1_800_000_000
+    with tempfile.TemporaryDirectory() as tmp:
+        transport = SnapshotTransport({1: view_for(1)})
+        db, repo, service = make_service(tmp, transport, track_fans=())
+        try:
+            add_video(repo, 1)
+            today = "2026-10-31"
+            repo.upsert_snapshot(
+                today, 1,
+                metrics={"view": 10, "like": 1, "coin": 0, "favorite": 0,
+                         "share": 0, "reply": 0, "danmaku": 0},
+                ok=True, err=None, captured_at=clock,
+            )
+            # 正常收尾态：last_completed 已推进到当日，且 completed 标记在位
+            repo.kv_set("snapshot:last_completed", today)
+            repo.kv_set(service.completed_key(today), "1")
+
+            summary = service.run_seed_from_last_completed(date=today)
+            assert summary.reason == "already_completed"
+            assert summary.already_completed is True
+            assert summary.completed is True
+            assert summary.reason != "no_source"
+            assert transport.count("/web-interface/view") == 0
+        finally:
+            db.close()
+
+
+
 def test_seed_from_last_completed_does_not_mark_completed():
     clock = 1_800_000_000
     with tempfile.TemporaryDirectory() as tmp:
@@ -646,6 +676,7 @@ def main():
     test_backup_failure_retries_without_api()
     test_run_recent_videos_refreshes_without_completion_markers()
     test_full_snapshot_skips_ok_rows_including_hourly_refresh()
+    test_seed_same_day_completed_is_already_completed_not_no_source()
     test_seed_from_last_completed_does_not_mark_completed()
     test_seeded_day_skips_without_force_and_seals_with_force()
     print("OK: 每日快照、断点续跑、失败占位与粉丝采集测试全部通过")