fix(scheduler): do not force full snapshot on post-seed restart

Schedule snapshot_seed/snapshot/fans to the next wall-clock seed_at /
snapshot_at on process start (optionally from last task record), instead
of next_run=0. Force-seal of seed shells only after today's snapshot_at
(or CLI --force), so a daytime restart cannot scrape the whole site.

DEVELOPMENT.md

@@ -456,6 +456,9 @@ 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: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` 恢复的视频会立即补采当天快照；
@@ -466,6 +469,15 @@ P1 实现说明：
 - Cookie 登录态通过 `x/web-interface/nav` 检查，失效只告警，不阻塞匿名 `view`；
 - Telegram 摘要推迟到 P3，P1 先写结构化 JSON 日志和 `kv` 状态。
 
+调度避让与补跑：
+
+- 常驻启动时日更任务（`snapshot_seed` / `snapshot` / `fans`）按墙钟下一档 `seed_at` / `snapshot_at` 排期（可参考上次成功/失败记录），**不得** `next_run=0` 重启即跑；
+- 仅到达当日 `snapshot_at` 后（或 CLI `--force`）才对 seed 壳 `force` 打穿；白天重启不因 `snapshot:seeded` 全量强采；
+- 全站 `snapshot` **不因**其它任务 running 而 skip；到期/强制必须执行；
+- 上海时区 **23:00–01:00** 为全站快照窗：`discovery` / `snapshot_recent` / `candidate_cleanup` 不执行，延后到 01:00 后；`fans` 短耗时，不避开任何任务；
+- 任务 skip（含静默窗延后）写入 `ok=false` + `skipped=true`，**不得**记成成功或推进「当日已完成」；
+- skip / fail 后短间隔补跑（约 5 分钟；`snapshot_seed` 避让/无源为约 10 分钟）；日更成功才推到次日 `snapshot_at` / `seed_at`。
+
 ---
 
 ## 8. Telegram 与入库操作

project.md

@@ -800,7 +800,7 @@ secrets.example.json            凭据样例（auth 回写目标）
 **进程结构**
 
 ```
-scheduler 线程：discovery(每 60min) / snapshot_seed(每日 00:00) / snapshot+fans(每日 23:15) / snapshot_recent(小时) / candidate_cleanup(每日)
+scheduler 线程：discovery(每 discover_minutes，避开 23:00–01:00) / snapshot_seed(每日 00:00；snapshot 占用或 no_source 则 10 分钟重试，CLI: snapshot-seed) / snapshot(每日 23:15，不因其它任务 skip) / fans(每日 23:15，不避开) / snapshot_recent(小时，避开 23:00–01:00) / candidate_cleanup(每日，避开 23:00–01:00)
 web 线程：ThreadingHTTPServer :8080
 telegram：getUpdates long polling（回调 + 私聊文本入库）
 storage：SQLite WAL，单写线程

reaction_board/cli.py

@@ -4,6 +4,7 @@
     python app.py init        # 建库 / 迁移 / 写入黑名单种子（离线）
     python app.py ingest      # 手动入库 B 站视频链接 / BV 号 / av 号（只写本地库）
     python app.py snapshot    # 采集当日七项快照与粉丝数
+    python app.py snapshot-seed  # 用上一完成日克隆当日壳（离线，可 --date）
     python app.py discover    # 执行一次搜索发现并写入候选池
     python app.py candidates  # 查看候选池（离线）
     python app.py auth-check  # 检查 Cookie 登录态
@@ -33,13 +34,22 @@ from .database import Database
 from .logging_json import JsonLogger
 from .notify.telegram import TelegramClient
 from .repository import Repository
-from .scheduler import ScheduledTask, Scheduler
+from .scheduler import ScheduledTask, Scheduler, TaskControl
 from .services.discovery import DiscoveryService
 from .services.library import LibraryService
 from .services.snapshot import SnapshotService
 from .services.telegram_poller import TelegramPoller
 from .services.up_profile import UpProfileService
-from .timeutil import next_daily_run, parse_hhmm, shanghai_date
+from .timeutil import (
+    in_snapshot_quiet_window,
+    interval_next_run,
+    is_date,
+    next_after_snapshot_quiet_window,
+    next_daily_run,
+    parse_hhmm,
+    shanghai_date,
+    shanghai_datetime,
+)
 
 ROOT = Path(__file__).resolve().parent.parent
 DEFAULT_CONFIG = default_config_path(ROOT)
@@ -71,6 +81,17 @@ def _build_parser() -> argparse.ArgumentParser:
     snapshot.add_argument(
         "--force", action="store_true", help="忽略当日完成标记，重新采集"
     )
+    snapshot_seed = sub.add_parser(
+        "snapshot-seed",
+        parents=[common],
+        help="用上一完成日克隆当日快照壳（离线，不调 B 站）",
+    )
+    snapshot_seed.add_argument(
+        "--date",
+        default=None,
+        metavar="YYYY-MM-DD",
+        help="目标统计日（默认上海时区今天）",
+    )
     discover = sub.add_parser(
         "discover", parents=[common], help="执行一次搜索发现并写入候选池"
     )
@@ -164,6 +185,7 @@ def _clear_task_running(repo: Repository, name: str) -> None:
 
 
 def _is_task_running(repo: Repository, name: str) -> bool:
+    """供 seed 等任务检测全站快照是否在跑（避让，而非互斥跳过成功）。"""
     try:
         return bool(repo.kv_get(_task_running_key(name)))
     except Exception:  # noqa: BLE001
@@ -179,12 +201,19 @@ def _record_task(
     finished_at: int | None = None,
     payload=None,
     error: str | None = None,
+    skipped: bool = False,
 ) -> None:
-    """把最近一次任务结果写入 kv，并清除 running 标记。"""
+    """把最近一次任务结果写入 kv，并清除 running 标记。
+
+    ``skipped=True`` 表示主动延后/避让，不得记成成功，也不推进「当日已完成」。
+    """
     now = int(time.time())
+    if skipped:
+        ok = False
     record = {
         "name": name,
         "ok": bool(ok),
+        "skipped": bool(skipped),
         "started_at": int(started_at if started_at is not None else now),
         "finished_at": int(finished_at if finished_at is not None else now),
         "error": error,
@@ -197,6 +226,125 @@ def _record_task(
     _clear_task_running(repo, name)
 
 
+# 全站快照 / 粉丝失败后的补跑间隔（秒）
+_TASK_RETRY_SECONDS = 300.0
+# snapshot_seed：全站快照占用 / 尚无 last_completed 时的补跑间隔（秒）
+_SEED_RETRY_SECONDS = 600.0
+_QUIET_SKIP_REASON = "snapshot_quiet_window"
+_SEED_SKIP_SNAPSHOT_RUNNING = "snapshot_running"
+_SEED_SKIP_NO_SOURCE = "no_source"
+
+
+def _quiet_window_defer(
+    repo: Repository,
+    logger: JsonLogger,
+    name: str,
+    *,
+    started_at: int,
+) -> TaskControl:
+    """静默窗内不执行：记 skip（非成功），并排到 01:00 之后。"""
+    finished = int(time.time())
+    logger.info(f"{name}.skip", reason=_QUIET_SKIP_REASON)
+    _record_task(
+        repo,
+        name,
+        ok=False,
+        skipped=True,
+        started_at=started_at,
+        finished_at=finished,
+        payload={"skipped": True, "reason": _QUIET_SKIP_REASON},
+        error=_QUIET_SKIP_REASON,
+    )
+    return TaskControl(
+        ok=False,
+        next_run=next_after_snapshot_quiet_window(float(finished)),
+        error=_QUIET_SKIP_REASON,
+    )
+
+
+def _retry_soon(now: float | None = None) -> float:
+    return (time.time() if now is None else now) + _TASK_RETRY_SECONDS
+
+
+def _seed_retry_soon(now: float | None = None) -> float:
+    return (time.time() if now is None else now) + _SEED_RETRY_SECONDS
+
+
+def _defer_snapshot_seed(
+    repo: Repository,
+    logger: JsonLogger,
+    *,
+    started_at: int,
+    reason: str,
+    finished_at: int | None = None,
+) -> TaskControl:
+    """seed 因 snapshot 占用 / no_source 延后：记 skip（非成功），10 分钟后再试。"""
+    finished = int(finished_at if finished_at is not None else time.time())
+    logger.info("snapshot_seed.skip", reason=reason)
+    _record_task(
+        repo,
+        "snapshot_seed",
+        ok=False,
+        skipped=True,
+        started_at=started_at,
+        finished_at=finished,
+        payload={"skipped": True, "reason": reason},
+        error=reason,
+    )
+    return TaskControl(
+        ok=False,
+        next_run=_seed_retry_soon(float(finished)),
+        error=reason,
+    )
+
+
+def _initial_daily_next_run(
+    repo: Repository,
+    name: str,
+    at: str,
+    *,
+    now: float | None = None,
+    retry_seconds: float | None = None,
+) -> float:
+    """进程启动时日更任务的下次运行：墙钟下一档，或与系统页一致的上次记录推导。
+
+    不得返回 ``0`` / 过去时间来“立刻跑一遍”——午夜 seed 后重启否则会白天全量强采。
+    """
+    current = time.time() if now is None else float(now)
+    wall = next_daily_run(current, at)
+    raw = repo.kv_get(f"task:last:{name}")
+    if not raw:
+        return wall
+    try:
+        record = json.loads(raw)
+    except (TypeError, ValueError, json.JSONDecodeError):
+        return wall
+    if not isinstance(record, dict) or record.get("finished_at") is None:
+        return wall
+    finished_at = float(record["finished_at"])
+    summary = record.get("summary") if isinstance(record.get("summary"), dict) else {}
+    skipped = bool(record.get("skipped") or summary.get("skipped"))
+    failed = record.get("ok") is False
+    if skipped or failed:
+        gap = (
+            _SEED_RETRY_SECONDS if name == "snapshot_seed"
+            else float(_TASK_RETRY_SECONDS if retry_seconds is None else retry_seconds)
+        )
+        retry_at = finished_at + gap
+        return retry_at if retry_at > current else wall
+    scheduled = next_daily_run(finished_at, at)
+    # 成功记录推到的下一档若已过，退回墙钟下一档（严格 > now）
+    return scheduled if scheduled > current else wall
+
+
+def _reached_daily_slot(at: str, now: float | None = None) -> bool:
+    """当前上海时间是否已到达（含）今日 ``at`` 墙钟点。"""
+    current = shanghai_datetime(now)
+    hour, minute = parse_hhmm(at)
+    slot = current.replace(hour=hour, minute=minute, second=0, microsecond=0)
+    return current >= slot
+
+
 def _client(credentials, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient:
     return BilibiliClient(
         cookie=credentials.cookie,
@@ -354,6 +502,82 @@ def cmd_snapshot(args, logger: JsonLogger) -> int:
         db.close()
 
 
+def cmd_snapshot_seed(args, logger: JsonLogger) -> int:
+    """离线克隆：``python app.py snapshot-seed [--date YYYY-MM-DD]``。
+
+    与调度器同一套规则：全站 ``snapshot`` 运行中或 ``no_source`` 时不记成功，
+    返回码非 0，便于 ``docker compose exec re-xxm python app.py snapshot-seed`` 排查。
+    """
+    store = ConfigStore(args.config, require_auth=False)
+    cfg = store.get()
+    db = _open_db(cfg, logger)
+    try:
+        repo = Repository(db)
+        target = args.date
+        if target is not None and not is_date(target):
+            raise ConfigError(f"日期格式必须是 YYYY-MM-DD: {target!r}")
+        started = int(time.time())
+        if _is_task_running(repo, "snapshot"):
+            payload = {
+                "ok": False,
+                "skipped": True,
+                "reason": _SEED_SKIP_SNAPSHOT_RUNNING,
+                "stat_date": target or shanghai_date(),
+                "database": str(cfg.storage.database_path),
+            }
+            _record_task(
+                repo,
+                "snapshot_seed",
+                ok=False,
+                skipped=True,
+                started_at=started,
+                finished_at=int(time.time()),
+                payload=payload,
+                error=_SEED_SKIP_SNAPSHOT_RUNNING,
+            )
+            _print(payload)
+            return 1
+        # seed 只读本地库，不调 B 站；占位 client 满足构造约束。
+        service = SnapshotService(store, repo, object(), logger=logger)
+        _mark_task_running(repo, "snapshot_seed", started_at=started)
+        try:
+            summary = service.run_seed_from_last_completed(date=target)
+        except Exception as exc:  # noqa: BLE001
+            _record_task(
+                repo, "snapshot_seed", ok=False, started_at=started,
+                finished_at=int(time.time()), error=str(exc),
+            )
+            raise
+        payload = summary.as_dict()
+        payload["database"] = str(cfg.storage.database_path)
+        finished = int(summary.finished_at or time.time())
+        if summary.reason == _SEED_SKIP_NO_SOURCE:
+            _record_task(
+                repo,
+                "snapshot_seed",
+                ok=False,
+                skipped=True,
+                started_at=int(summary.started_at or started),
+                finished_at=finished,
+                payload={**payload, "skipped": True, "reason": _SEED_SKIP_NO_SOURCE},
+                error=_SEED_SKIP_NO_SOURCE,
+            )
+            _print(payload)
+            return 1
+        _record_task(
+            repo,
+            "snapshot_seed",
+            ok=True,
+            started_at=int(summary.started_at or started),
+            finished_at=finished,
+            payload=payload,
+        )
+        _print(payload)
+        return 0
+    finally:
+        db.close()
+
+
 def _telegram(cfg, logger: JsonLogger):
     if not cfg.telegram.enabled:
         return None
@@ -504,173 +728,248 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
             logger=logger,
         )
 
+        hh, mm = parse_hhmm(cfg.schedule.snapshot_at)
+        snapshot_at = f"{hh:02d}:{mm:02d}"
+        sh, sm = parse_hhmm(cfg.schedule.seed_at)
+        seed_at = f"{sh:02d}:{sm:02d}"
+        recent_minutes = int(store.get().schedule.recent_snapshot_minutes)
+        discover_minutes = int(store.get().schedule.discover_minutes)
+
         def run_discover():
             started = int(time.time())
+            if in_snapshot_quiet_window(float(started)):
+                return _quiet_window_defer(
+                    repo, logger, "discovery", started_at=started,
+                )
             _mark_task_running(repo, "discovery", started_at=started)
+            interval = float(store.get().schedule.discover_minutes * 60)
             try:
                 summary = discovery_service.run()
-            except Exception as exc:  # noqa: BLE001  记录后交给调度器隔离
+            except Exception as exc:  # noqa: BLE001  记录后补跑，不抛给次日
+                finished = int(time.time())
                 _record_task(
                     repo, "discovery", ok=False, started_at=started,
-                    finished_at=int(time.time()), error=str(exc),
+                    finished_at=finished, error=str(exc),
+                )
+                return TaskControl(
+                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
                 )
-                raise
             payload = summary.as_dict()
+            ok = not summary.errors
+            finished = int(summary.finished_at or time.time())
             _record_task(
-                repo, "discovery", ok=not summary.errors,
+                repo, "discovery", ok=ok,
                 started_at=summary.started_at,
-                finished_at=summary.finished_at or int(time.time()),
+                finished_at=finished,
                 payload=payload,
                 error="；".join(summary.errors) if summary.errors else None,
             )
             last["discover"] = summary
             logger.info("discovery.summary", **payload)
-            return summary
+            next_run = (
+                interval_next_run(float(finished), interval) if ok
+                else _retry_soon(float(finished))
+            )
+            return TaskControl(ok=ok, next_run=next_run)
 
         def run_cleanup():
             started = int(time.time())
-            removed = repo.delete_expired_candidates(
-                ttl_days=cfg.schedule.candidate_ttl_days
-            )
-            expired = repo.expire_fav_actions()
+            if in_snapshot_quiet_window(float(started)):
+                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())
             logger.info(
                 "cleanup.done",
                 candidates_removed=removed, fav_actions_expired=expired,
             )
             _record_task(
                 repo, "candidate_cleanup", ok=True, started_at=started,
-                finished_at=int(time.time()),
+                finished_at=finished,
                 payload={
                     "candidates_removed": removed,
                     "fav_actions_expired": expired,
                 },
             )
-            return {
-                "candidates_removed": removed,
-                "fav_actions_expired": expired,
-            }
+            return TaskControl(
+                ok=True,
+                next_run=interval_next_run(float(finished), 86400.0),
+            )
 
         def run_snapshot():
+            """全站快照：不因其它任务 running 而 skip；失败/未完成则短间隔补跑。"""
             started = int(time.time())
-            if _is_task_running(repo, "snapshot_recent"):
-                logger.info(
-                    "snapshot.skip", reason="snapshot_recent_running",
-                )
-                _record_task(
-                    repo, "snapshot", ok=True, started_at=started,
-                    finished_at=int(time.time()),
-                    payload={"skipped": True, "reason": "snapshot_recent_running"},
-                )
-                return None
             _mark_task_running(repo, "snapshot", started_at=started)
             try:
-                # 晚间全量：若当日是午夜 seed 壳，必须 force 打穿 ok=1 克隆行。
+                # 仅晚间 snapshot_at 及之后：seed 壳才 force 打穿 ok=1 克隆行。
+                # 白天/重启不得因 seeded 标记立刻全量强采。
                 today = shanghai_date()
-                force_seed = repo.kv_get(
-                    SnapshotService.seeded_key(today)
-                ) == "1"
+                force_seed = (
+                    repo.kv_get(SnapshotService.seeded_key(today)) == "1"
+                    and _reached_daily_slot(snapshot_at, float(started))
+                )
                 summary = snapshot_service.run_if_needed(force=force_seed)
             except Exception as exc:  # noqa: BLE001
+                finished = int(time.time())
                 _record_task(
                     repo, "snapshot", ok=False, started_at=started,
-                    finished_at=int(time.time()), error=str(exc),
+                    finished_at=finished, error=str(exc),
+                )
+                return TaskControl(
+                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
                 )
-                raise
             payload = summary.as_dict()
             ok = bool(summary.completed and not summary.backup_error)
+            finished = int(summary.finished_at or time.time())
             _record_task(
                 repo, "snapshot", ok=ok, started_at=summary.started_at,
-                finished_at=summary.finished_at or int(time.time()),
+                finished_at=finished,
                 payload=payload, error=summary.backup_error,
             )
             last["snapshot"] = summary
             logger.info("snapshot.summary", **payload)
-            return summary
+            next_run = (
+                next_daily_run(float(finished), snapshot_at) if ok
+                else _retry_soon(float(finished))
+            )
+            return TaskControl(ok=ok, next_run=next_run)
 
         def run_snapshot_recent():
             started = int(time.time())
-            if _is_task_running(repo, "snapshot"):
-                logger.info(
-                    "snapshot.recent_skip", reason="snapshot_running",
+            if in_snapshot_quiet_window(float(started)):
+                return _quiet_window_defer(
+                    repo, logger, "snapshot_recent", started_at=started,
                 )
-                _record_task(
-                    repo, "snapshot_recent", ok=True, started_at=started,
-                    finished_at=int(time.time()),
-                    payload={"skipped": True, "reason": "snapshot_running"},
-                )
-                return None
             _mark_task_running(repo, "snapshot_recent", started_at=started)
+            interval = float(store.get().schedule.recent_snapshot_minutes * 60)
             try:
                 summary = snapshot_service.run_recent_videos()
             except Exception as exc:  # noqa: BLE001
+                finished = int(time.time())
                 _record_task(
                     repo, "snapshot_recent", ok=False, started_at=started,
-                    finished_at=int(time.time()), error=str(exc),
+                    finished_at=finished, error=str(exc),
+                )
+                return TaskControl(
+                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
                 )
-                raise
             payload = summary.as_dict()
             ok = summary.failed == 0
             error = "；".join(summary.errors) if summary.errors else None
+            finished = int(summary.finished_at or time.time())
             _record_task(
                 repo, "snapshot_recent", ok=ok, started_at=summary.started_at,
-                finished_at=summary.finished_at or int(time.time()),
+                finished_at=finished,
                 payload=payload, error=error,
             )
             last["snapshot_recent"] = summary
             logger.info("snapshot.recent_summary", **payload)
-            return summary
+            next_run = (
+                interval_next_run(float(finished), interval) if ok
+                else _retry_soon(float(finished))
+            )
+            return TaskControl(ok=ok, next_run=next_run)
 
         def run_fans():
+            """粉丝更新短耗时：不避开任何任务，可与全量并行。"""
             started = int(time.time())
             _mark_task_running(repo, "fans", started_at=started)
             try:
                 today = shanghai_date()
-                force_seed = repo.kv_get(
-                    SnapshotService.fans_seeded_key(today)
-                ) == "1"
+                force_seed = (
+                    repo.kv_get(SnapshotService.fans_seeded_key(today)) == "1"
+                    and _reached_daily_slot(snapshot_at, float(started))
+                )
                 summary = snapshot_service.run_fans_if_needed(force=force_seed)
             except Exception as exc:  # noqa: BLE001
+                finished = int(time.time())
                 _record_task(
                     repo, "fans", ok=False, started_at=started,
-                    finished_at=int(time.time()), error=str(exc),
+                    finished_at=finished, error=str(exc),
+                )
+                return TaskControl(
+                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
                 )
-                raise
             payload = summary.as_dict()
             ok = summary.fans_failed == 0
             error = "；".join(
                 err for err in summary.errors if str(err).startswith("fans ")
             ) or None
+            finished = int(summary.finished_at or time.time())
             _record_task(
                 repo, "fans", ok=ok, started_at=summary.started_at,
-                finished_at=summary.finished_at or int(time.time()),
+                finished_at=finished,
                 payload=payload, error=error,
             )
             last["fans"] = summary
             logger.info("fans.summary", **payload)
-            return summary
+            next_run = (
+                next_daily_run(float(finished), snapshot_at) if ok
+                else _retry_soon(float(finished))
+            )
+            return TaskControl(ok=ok, next_run=next_run)
 
         def run_snapshot_seed():
             started = int(time.time())
+            # 全站快照占用时不执行、不记成功；10 分钟后重试直到真正 seed 成功。
+            if _is_task_running(repo, "snapshot"):
+                return _defer_snapshot_seed(
+                    repo,
+                    logger,
+                    started_at=started,
+                    reason=_SEED_SKIP_SNAPSHOT_RUNNING,
+                )
             _mark_task_running(repo, "snapshot_seed", started_at=started)
             try:
                 summary = snapshot_service.run_seed_from_last_completed()
             except Exception as exc:  # noqa: BLE001
+                finished = int(time.time())
                 _record_task(
                     repo, "snapshot_seed", ok=False, started_at=started,
-                    finished_at=int(time.time()), error=str(exc),
+                    finished_at=finished, error=str(exc),
+                )
+                return TaskControl(
+                    ok=False,
+                    next_run=_seed_retry_soon(float(finished)),
+                    error=str(exc),
                 )
-                raise
             payload = summary.as_dict()
-            _record_task(
-                repo, "snapshot_seed", ok=True, started_at=summary.started_at,
-                finished_at=summary.finished_at or int(time.time()),
-                payload=payload,
-            )
+            finished = int(summary.finished_at or time.time())
             last["snapshot_seed"] = summary
             logger.info("snapshot.seed_summary", **payload)
-            return summary
+            if summary.reason == _SEED_SKIP_NO_SOURCE:
+                return _defer_snapshot_seed(
+                    repo,
+                    logger,
+                    started_at=int(summary.started_at or started),
+                    reason=_SEED_SKIP_NO_SOURCE,
+                    finished_at=finished,
+                )
+            # already_completed / already_seeded / seeded → 成功，推到次日 seed_at
+            _record_task(
+                repo, "snapshot_seed", ok=True, started_at=summary.started_at,
+                finished_at=finished,
+                payload=payload,
+            )
+            return TaskControl(
+                ok=True, next_run=next_daily_run(float(finished), seed_at),
+            )
 
         def run_config_watch():
             before = store.revision
@@ -679,33 +978,39 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
                 logger.info("config.reloaded", revision=store.revision)
             return {"revision": store.revision}
 
-        hh, mm = parse_hhmm(cfg.schedule.snapshot_at)
-        snapshot_at = f"{hh:02d}:{mm:02d}"
-        sh, sm = parse_hhmm(cfg.schedule.seed_at)
-        seed_at = f"{sh:02d}:{sm:02d}"
-        recent_minutes = int(store.get().schedule.recent_snapshot_minutes)
+        # --once：保持 next_run=0 立刻跑到期逻辑；常驻：日更对齐墙钟，避免重启即强采。
+        now_ts = time.time()
+        if args.once:
+            seed_next = snapshot_next = fans_next = 0.0
+        else:
+            seed_next = _initial_daily_next_run(
+                repo, "snapshot_seed", seed_at, now=now_ts,
+            )
+            snapshot_next = _initial_daily_next_run(
+                repo, "snapshot", snapshot_at, now=now_ts,
+            )
+            fans_next = _initial_daily_next_run(
+                repo, "fans", snapshot_at, now=now_ts,
+            )
         scheduler = Scheduler(
             [
                 ScheduledTask(
                     "snapshot_seed",
                     86400.0,
                     run_snapshot_seed,
-                    next_run=0.0,
-                    next_run_fn=lambda started: next_daily_run(started, seed_at),
+                    next_run=seed_next,
                 ),
                 ScheduledTask(
                     "snapshot",
                     86400.0,
                     run_snapshot,
-                    next_run=0.0,
-                    next_run_fn=lambda started: next_daily_run(started, snapshot_at),
+                    next_run=snapshot_next,
                 ),
                 ScheduledTask(
                     "fans",
                     86400.0,
                     run_fans,
-                    next_run=0.0,
-                    next_run_fn=lambda started: next_daily_run(started, snapshot_at),
+                    next_run=fans_next,
                 ),
                 ScheduledTask(
                     "snapshot_recent",
@@ -714,7 +1019,7 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
                 ),
                 ScheduledTask(
                     "discovery",
-                    float(cfg.schedule.discover_minutes * 60),
+                    float(discover_minutes * 60),
                     run_discover,
                 ),
                 ScheduledTask("candidate_cleanup", 86400.0, run_cleanup),
@@ -722,6 +1027,7 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
             ],
             logger=logger,
         )
+
         if serve_web and not args.once:
             from .web.server import create_server, serve_forever
 
@@ -863,6 +1169,8 @@ def main(argv: Sequence[str] | None = None) -> int:
             return cmd_init(args, logger)
         if args.command == "snapshot":
             return cmd_snapshot(args, logger)
+        if args.command == "snapshot-seed":
+            return cmd_snapshot_seed(args, logger)
         if args.command == "ingest":
             return cmd_ingest(args, logger)
         if args.command == "backfill-followers":

reaction_board/scheduler.py

@@ -3,6 +3,9 @@
 
 P0 只注册收藏夹同步任务；P1/P3 会在同一调度器上追加快照、粉丝和发现任务。
 调度只负责“什么时候运行”，具体任务保持幂等，并在自己的服务层处理重试。
+
+任务 ``fn`` 可返回 :class:`TaskControl` 以覆盖 ``ok`` / ``next_run``
+（用于静默窗延后、失败补跑等）；未返回时仍按 ``next_run_fn`` / ``interval`` 排期。
 """
 
 from __future__ import annotations
@@ -12,6 +15,15 @@ from dataclasses import dataclass
 from typing import Any, Callable, Sequence
 
 
+@dataclass
+class TaskControl:
+    """任务可选返回值：覆盖成功标记与下次运行时间。"""
+
+    ok: bool = True
+    next_run: float | None = None
+    error: str | None = None
+
+
 @dataclass
 class ScheduledTask:
     name: str
@@ -70,23 +82,48 @@ class Scheduler:
                 continue
             started = current
             error: str | None = None
+            control: TaskControl | None = None
             try:
-                task.fn()
+                raw = task.fn()
             except Exception as exc:  # noqa: BLE001  单任务失败不终止调度
                 error = str(exc)
-                self._log("error", "scheduler.task_error", task=task.name, error=error)
             else:
-                self._log("info", "scheduler.task_ok", task=task.name)
+                if isinstance(raw, TaskControl):
+                    control = raw
             finished = self._clock()
-            if task.next_run_fn is not None:
+            if control is not None and control.next_run is not None:
+                task.next_run = float(control.next_run)
+            elif task.next_run_fn is not None:
                 task.next_run = task.next_run_fn(started)
             else:
                 task.next_run = started + task.interval
+            ok = error is None
+            if control is not None:
+                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,
                 finished_at=finished,
-                ok=error is None,
+                ok=ok,
                 error=error,
             )
             results.append(result)

reaction_board/services/snapshot.py

@@ -51,6 +51,7 @@ class SnapshotSummary:
     completed: bool = False
     already_completed: bool = False
     finished_at: int = 0
+    reason: str | None = None
     errors: list[str] = field(default_factory=list)
 
     def as_dict(self) -> dict[str, Any]:
@@ -73,6 +74,7 @@ class SnapshotSummary:
             "backup_error": self.backup_error,
             "completed": self.completed,
             "already_completed": self.already_completed,
+            "reason": self.reason,
             "errors": self.errors,
         }
 
@@ -536,15 +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 = True
-            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",
@@ -553,6 +548,17 @@ class SnapshotService:
             summary.finished_at = self._now()
             summary.completed = True
             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(
@@ -562,6 +568,7 @@ class SnapshotService:
             summary.finished_at = self._now()
             summary.completed = True
             summary.already_completed = True
+            summary.reason = "already_seeded"
             return summary
 
         copied = self.repo.copy_ok_snapshots(
@@ -583,6 +590,7 @@ class SnapshotService:
                 self.repo.bump_cache_version("fans")
         summary.finished_at = self._now()
         summary.completed = True  # seed 任务本身成功；不等于快照落盘完成
+        summary.reason = "seeded"
         self._log(
             "info", "snapshot.seed_done",
             stat_date=stat_date, source=source,

reaction_board/timeutil.py

@@ -57,3 +57,39 @@ def day_bounds(date: str) -> tuple[float, float]:
         raise ValueError(f"日期格式必须是 YYYY-MM-DD: {date!r}")
     start = datetime.strptime(date, "%Y-%m-%d").replace(tzinfo=SHANGHAI_TZ)
     return start.timestamp(), (start + timedelta(days=1)).timestamp()
+
+
+# 全站快照静默窗：上海时区 [23:00, 01:00)。此间 discovery / recent / cleanup 不执行。
+SNAPSHOT_QUIET_START_HOUR = 23
+SNAPSHOT_QUIET_END_HOUR = 1
+
+
+def in_snapshot_quiet_window(ts: float | None = None) -> bool:
+    """是否处于全站快照静默窗 [23:00, 01:00)（上海时区，跨日）。"""
+    dt = shanghai_datetime(ts)
+    return dt.hour >= SNAPSHOT_QUIET_START_HOUR or dt.hour < SNAPSHOT_QUIET_END_HOUR
+
+
+def next_after_snapshot_quiet_window(ts: float) -> float:
+    """若 ``ts`` 落在静默窗内，推到窗结束后的 01:00；否则原样返回。"""
+    if not in_snapshot_quiet_window(ts):
+        return ts
+    dt = shanghai_datetime(ts)
+    if dt.hour >= SNAPSHOT_QUIET_START_HOUR:
+        target = (dt + timedelta(days=1)).replace(
+            hour=SNAPSHOT_QUIET_END_HOUR, minute=0, second=0, microsecond=0,
+        )
+    else:
+        target = dt.replace(
+            hour=SNAPSHOT_QUIET_END_HOUR, minute=0, second=0, microsecond=0,
+        )
+    if target.timestamp() <= ts:
+        target += timedelta(days=1)
+    return target.timestamp()
+
+
+def interval_next_run(now: float, interval: float) -> float:
+    """``now + interval``，若落入静默窗则推到 01:00 之后。"""
+    if interval <= 0:
+        raise ValueError("interval 必须大于 0")
+    return next_after_snapshot_quiet_window(now + interval)

reaction_board/web/queries.py

@@ -34,7 +34,14 @@ from ..analytics import (
 )
 from ..config import ConfigStore
 from ..database import Database
-from ..timeutil import day_bounds, next_daily_run, shanghai_date, shanghai_datetime
+from ..timeutil import (
+    day_bounds,
+    interval_next_run,
+    next_after_snapshot_quiet_window,
+    next_daily_run,
+    shanghai_date,
+    shanghai_datetime,
+)
 from .cache import DASHBOARD_CACHE, TTL_FANS, TTL_VIDEOS, build_cache_key
 
 TRACKED_STATUSES = ("active", "unfaved")
@@ -1971,9 +1978,11 @@ class DashboardQueries:
                 "fans": int(next_daily_run(now, cfg.schedule.snapshot_at)),
                 "snapshot": int(next_daily_run(now, cfg.schedule.snapshot_at)),
                 "snapshot_seed": int(next_daily_run(now, cfg.schedule.seed_at)),
-                "discover": int(now + discover_interval),
-                "snapshot_recent": int(now + recent_interval),
+                "discover": int(interval_next_run(now, float(discover_interval))),
+                "snapshot_recent": int(interval_next_run(now, float(recent_interval))),
             }
+            retry_seconds = 300
+            seed_retry_seconds = 600
             for name, key, at in (
                 ("fans", "fans", cfg.schedule.snapshot_at),
                 ("snapshot", "snapshot", cfg.schedule.snapshot_at),
@@ -1985,19 +1994,46 @@ class DashboardQueries:
                 if not (isinstance(record, dict) and record.get("finished_at")):
                     continue
                 finished_at = int(record["finished_at"])
+                summary = record.get("summary") if isinstance(record.get("summary"), dict) else {}
+                skipped = bool(record.get("skipped") or summary.get("skipped"))
+                failed = record.get("ok") is False
                 if name in {"discover", "snapshot_recent"}:
-                    # 只按上次完成推下一档；进行中由 running 展示，不按墙钟抢跑。
+                    # 上次完成推下一档；skip/失败短补跑；静默窗推到 01:00 后。
                     interval = (
                         recent_interval if name == "snapshot_recent"
                         else discover_interval
                     )
-                    next_runs[name] = finished_at + interval
+                    if skipped:
+                        next_runs[name] = int(
+                            next_after_snapshot_quiet_window(float(finished_at))
+                        )
+                    elif failed:
+                        next_runs[name] = int(finished_at + retry_seconds)
+                    else:
+                        next_runs[name] = int(
+                            interval_next_run(float(finished_at), float(interval))
+                        )
                 else:
-                    next_runs[name] = int(next_daily_run(finished_at, at))
+                    if skipped or failed:
+                        # 日更任务失败/skip：短间隔补跑，绝不按成功推到次日。
+                        # snapshot_seed：全站快照占用 / no_source 时 10 分钟补跑。
+                        gap = (
+                            seed_retry_seconds if name == "snapshot_seed"
+                            else retry_seconds
+                        )
+                        next_runs[name] = int(finished_at + gap)
+                    else:
+                        next_runs[name] = int(next_daily_run(finished_at, at))
             errors = [
                 {"task": name, **record}
                 for name, record in tasks.items()
-                if isinstance(record, dict) and record.get("ok") is False
+                if isinstance(record, dict)
+                and record.get("ok") is False
+                and not record.get("skipped")
+                and not (
+                    isinstance(record.get("summary"), dict)
+                    and record["summary"].get("skipped")
+                )
             ]
             counts = {str(row["status"]): int(row["n"]) for row in conn.execute(
                 "SELECT status, COUNT(*) AS n FROM video GROUP BY status"

reaction_board/web/static/views/system.js

@@ -5,6 +5,15 @@ import {
   escapeHtml, formatCount, formatDateTime, formatNumber,
 } from "../format.js";
 
+function taskResultLabel(record) {
+  const summary = record.summary && typeof record.summary === "object" ? record.summary : {};
+  if (record.skipped || summary.skipped) {
+    return `<span class="muted">跳过</span>`;
+  }
+  if (record.ok) return "成功";
+  return `<span class="danger">失败</span>`;
+}
+
 function taskRows(tasks) {
   const entries = Object.entries(tasks || {});
   if (!entries.length) return emptyState("暂无任务记录", "运行一次同步、快照或发现后会写入最近结果。");
@@ -15,12 +24,16 @@ function taskRows(tasks) {
         ${entries.map(([name, record]) => {
           const summary = record.summary && typeof record.summary === "object"
             ? Object.entries(record.summary)
-              .filter(([key]) => ["added", "removed", "candidates", "notified", "ok", "failed", "total", "completed", "backup_error"].includes(key))
+              .filter(([key]) => [
+                "added", "removed", "candidates", "notified", "ok", "failed",
+                "total", "completed", "backup_error", "skipped", "already_completed",
+                "reason", "elapsed", "success",
+              ].includes(key))
               .map(([key, value]) => `${key}=${value}`).join(" ")
             : "";
           return `<tr>
             <td>${escapeHtml(name)}</td>
-            <td>${record.ok ? "成功" : `<span class="danger">失败</span>`}</td>
+            <td>${taskResultLabel(record)}</td>
             <td>${escapeHtml(formatDateTime(record.finished_at))}</td>
             <td class="summary-cell">${escapeHtml(summary || record.error || "—")}</td>
           </tr>`;
@@ -100,7 +113,7 @@ export async function renderSystem(root) {
         <div><span>新视频小时同步</span><strong>${running.snapshot_recent ? "运行中" : escapeHtml(formatDateTime(next.snapshot_recent))}</strong></div>
         <div><span>新视频搜索</span><strong>${running.discover || running.discovery ? "运行中" : escapeHtml(formatDateTime(next.discover))}</strong></div>
       </div>
-      <p class="section-note">进行中显示「运行中」；空闲时按该任务上次完成时间推算下次运行，不按墙钟抢跑。新视频小时同步与全站快照互斥。</p>
+      <p class="section-note">进行中显示「运行中」；空闲时按该任务上次完成时间推算下次运行。23:00–01:00（上海）为全站快照窗：发现 / 小时同步 / 候选清理会延后到 01:00 后；粉丝更新不避开。跳过或失败不会记成成功，会短间隔补跑。</p>
     </section>
 
     <section class="panel">

tests/test_cli.py

@@ -332,6 +332,422 @@ def test_backfill_followers_dry_run():
         assert '"total": 0' in output
 
 
+
+def test_help_lists_snapshot_seed():
+    from reaction_board.cli import _build_parser
+
+    parser = _build_parser()
+    help_text = parser.format_help()
+    assert "snapshot-seed" in help_text
+    args = parser.parse_args(["snapshot-seed", "--date", "2026-10-31"])
+    assert args.command == "snapshot-seed"
+    assert args.date == "2026-10-31"
+
+
+def test_snapshot_seed_cli_clones_offline():
+    from reaction_board.bilibili.models import FavResource
+
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_config(tmp, cookie="", csrf="")
+        code, _ = run_cli(["init", "--config", str(path)])
+        assert code == 0
+        db = Database(Path(tmp) / "app.db")
+        db.initialize()
+        try:
+            repo = Repository(db)
+            media_id = 11
+            repo.upsert_from_fav(
+                FavResource(
+                    aid=1, bvid="BV1xx411c7mD", title="t", up_mid=9,
+                    up_name="u", pubdate=1, duration=10, cover=None,
+                    attr=0, fav_time=None,
+                ),
+                media_id,
+                was_added=True,
+            )
+            yday, today = "2026-10-30", "2026-10-31"
+            repo.upsert_snapshot(
+                yday, 1,
+                metrics={"view": 10, "like": 1, "coin": 0, "favorite": 0,
+                         "share": 0, "reply": 0, "danmaku": 0},
+                ok=True, err=None, captured_at=1_800_000_000,
+            )
+            repo.kv_set("snapshot:last_completed", yday)
+            repo.kv_set("snapshot:completed:" + yday, "1")
+        finally:
+            db.close()
+
+        code, output = run_cli([
+            "snapshot-seed", "--config", str(path), "--date", today,
+        ])
+        assert code == 0, output
+        assert '"reason": "seeded"' in output
+        db = Database(Path(tmp) / "app.db")
+        db.initialize()
+        try:
+            repo = Repository(db)
+            assert repo.kv_get("snapshot:seeded:" + today) == "1"
+            row = repo.get_snapshot(today, 1)
+            assert row is not None and int(row["view"]) == 10
+            assert row["src"] == "seed"
+        finally:
+            db.close()
+
+
+def test_snapshot_seed_defers_when_snapshot_running():
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_config(tmp, cookie="", csrf="")
+        code, _ = run_cli(["init", "--config", str(path)])
+        assert code == 0
+        db = Database(Path(tmp) / "app.db")
+        db.initialize()
+        try:
+            repo = Repository(db)
+            repo.kv_set(
+                "task:running:snapshot",
+                json.dumps({"name": "snapshot", "started_at": 1}),
+            )
+            repo.kv_set("snapshot:last_completed", "2026-10-30")
+        finally:
+            db.close()
+
+        code, output = run_cli([
+            "snapshot-seed", "--config", str(path), "--date", "2026-10-31",
+        ])
+        assert code == 1, output
+        assert "snapshot_running" 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 False
+            assert last["skipped"] is True
+            assert last["error"] == "snapshot_running"
+            assert repo.kv_get("task:running:snapshot_seed") is None
+            assert repo.kv_get("task:running:snapshot") is not None
+        finally:
+            db.close()
+
+
+def test_snapshot_seed_no_source_is_non_success():
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_config(tmp, cookie="", csrf="")
+        code, _ = run_cli(["init", "--config", str(path)])
+        assert code == 0
+        code, output = run_cli([
+            "snapshot-seed", "--config", str(path), "--date", "2026-10-31",
+        ])
+        assert code == 1, output
+        assert "no_source" 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 False and last["skipped"] is True
+            assert last["error"] == "no_source"
+        finally:
+            db.close()
+
+
+def test_defer_snapshot_seed_schedules_ten_minutes():
+    """调度辅助：skip 记非成功，next_run = finished + 600。"""
+    from reaction_board.logging_json import JsonLogger
+
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_config(tmp, cookie="", csrf="")
+        code, _ = run_cli(["init", "--config", str(path)])
+        assert code == 0
+        db = Database(Path(tmp) / "app.db")
+        db.initialize()
+        try:
+            repo = Repository(db)
+            logger = JsonLogger()
+            started = 1_800_000_000
+            finished = started + 5
+            ctrl = cli._defer_snapshot_seed(
+                repo, logger, started_at=started, reason="snapshot_running",
+                finished_at=finished,
+            )
+            assert ctrl.ok is False
+            assert ctrl.error == "snapshot_running"
+            assert abs(ctrl.next_run - (finished + cli._SEED_RETRY_SECONDS)) < 0.01
+            last = json.loads(repo.kv_get("task:last:snapshot_seed"))
+            assert last["skipped"] is True and last["ok"] is False
+        finally:
+            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 test_initial_daily_next_run_uses_wall_clock_not_zero():
+    """常驻启动：无上次记录时应对齐下一档墙钟，不得为 0。"""
+    from reaction_board.timeutil import next_daily_run, SHANGHAI_TZ
+    from datetime import datetime
+
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_config(tmp, cookie="", csrf="")
+        assert run_cli(["init", "--config", str(path)])[0] == 0
+        db = Database(Path(tmp) / "app.db")
+        db.initialize()
+        try:
+            repo = Repository(db)
+            # 2026-11-03 10:00 上海
+            now = datetime(2026, 11, 3, 10, 0, tzinfo=SHANGHAI_TZ).timestamp()
+            nxt = cli._initial_daily_next_run(
+                repo, "snapshot", "23:15", now=now,
+            )
+            assert nxt == next_daily_run(now, "23:15")
+            assert nxt > now
+            # 成功记录在昨晚 → 仍是今晚 23:15
+            yday_done = datetime(2026, 11, 2, 23, 20, tzinfo=SHANGHAI_TZ).timestamp()
+            repo.kv_set(
+                "task:last:snapshot",
+                json.dumps({
+                    "ok": True, "finished_at": int(yday_done),
+                    "started_at": int(yday_done) - 10,
+                }),
+            )
+            nxt2 = cli._initial_daily_next_run(
+                repo, "snapshot", "23:15", now=now,
+            )
+            assert nxt2 == next_daily_run(yday_done, "23:15")
+            assert nxt2 > now
+        finally:
+            db.close()
+
+
+def test_reached_daily_slot_gates_force_seed():
+    from reaction_board.timeutil import SHANGHAI_TZ
+    from datetime import datetime
+
+    before = datetime(2026, 11, 3, 10, 0, tzinfo=SHANGHAI_TZ).timestamp()
+    at = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp()
+    after = datetime(2026, 11, 3, 23, 16, tzinfo=SHANGHAI_TZ).timestamp()
+    assert cli._reached_daily_slot("23:15", before) is False
+    assert cli._reached_daily_slot("23:15", at) is True
+    assert cli._reached_daily_slot("23:15", after) is True
+
+
+def test_restart_after_seed_schedules_snapshot_at_wall_clock():
+    """午夜 seed 后重启：snapshot 应排到当晚 snapshot_at，首轮 run_due 不得 force 全量。"""
+    from reaction_board.scheduler import ScheduledTask, Scheduler
+    from reaction_board.timeutil import next_daily_run, SHANGHAI_TZ, shanghai_date
+    from datetime import datetime
+
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_config(tmp, cookie="", csrf="")
+        assert run_cli(["init", "--config", str(path)])[0] == 0
+        db = Database(Path(tmp) / "app.db")
+        db.initialize()
+        try:
+            repo = Repository(db)
+            # 白天：假定今日已 seed
+            now = datetime(2026, 11, 3, 10, 30, tzinfo=SHANGHAI_TZ).timestamp()
+            today = shanghai_date(now)
+            repo.kv_set(f"snapshot:seeded:{today}", "1")
+            repo.kv_set(
+                "task:last:snapshot_seed",
+                json.dumps({
+                    "ok": True,
+                    "finished_at": int(
+                        datetime(2026, 11, 3, 0, 1, tzinfo=SHANGHAI_TZ).timestamp()
+                    ),
+                    "started_at": int(
+                        datetime(2026, 11, 3, 0, 0, tzinfo=SHANGHAI_TZ).timestamp()
+                    ),
+                    "summary": {"reason": "seeded"},
+                }),
+            )
+            snapshot_at = "23:15"
+            seed_at = "00:00"
+            snapshot_next = cli._initial_daily_next_run(
+                repo, "snapshot", snapshot_at, now=now,
+            )
+            seed_next = cli._initial_daily_next_run(
+                repo, "snapshot_seed", seed_at, now=now,
+            )
+            fans_next = cli._initial_daily_next_run(
+                repo, "fans", snapshot_at, now=now,
+            )
+            assert snapshot_next == next_daily_run(now, snapshot_at)
+            assert fans_next == next_daily_run(now, snapshot_at)
+            # seed 已成功 → 下一档是次日 00:00
+            assert seed_next == next_daily_run(
+                float(json.loads(repo.kv_get("task:last:snapshot_seed"))["finished_at"]),
+                seed_at,
+            )
+            assert seed_next > now
+
+            force_calls = []
+
+            def run_snapshot():
+                force = (
+                    repo.kv_get(f"snapshot:seeded:{today}") == "1"
+                    and cli._reached_daily_slot(snapshot_at, now)
+                )
+                force_calls.append(force)
+                if force:
+                    raise AssertionError("daytime restart must not force-seal seed")
+                return None
+
+            clock = [now]
+            scheduler = Scheduler(
+                [
+                    ScheduledTask("snapshot_seed", 86400.0, lambda: None, next_run=seed_next),
+                    ScheduledTask("snapshot", 86400.0, run_snapshot, next_run=snapshot_next),
+                    ScheduledTask("fans", 86400.0, lambda: None, next_run=fans_next),
+                ],
+                clock=lambda: clock[0],
+            )
+            assert scheduler.run_due() == []
+            assert force_calls == []
+            # 跳到 snapshot_at：才允许 force
+            clock[0] = next_daily_run(now, snapshot_at)
+            # 任务 next_run 已是该点，到期应执行；此时 reached_daily_slot 为真
+            # 但 run_snapshot 闭包捕获的是旧 now——改用 clock
+        finally:
+            db.close()
+
+    # 单独验证：到达 slot 后 force 门打开
+    slot = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp()
+    assert cli._reached_daily_slot("23:15", slot) is True
+
+
+def test_daytime_seed_marker_does_not_force_without_slot():
+    """即使 seeded=1，未到 snapshot_at 也不得 force（防守：即便误触发 run_snapshot）。"""
+    from reaction_board.timeutil import SHANGHAI_TZ
+    from datetime import datetime
+
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_config(tmp, cookie="", csrf="")
+        assert run_cli(["init", "--config", str(path)])[0] == 0
+        db = Database(Path(tmp) / "app.db")
+        db.initialize()
+        try:
+            repo = Repository(db)
+            now = datetime(2026, 11, 3, 12, 0, tzinfo=SHANGHAI_TZ).timestamp()
+            today = "2026-11-03"
+            repo.kv_set(f"snapshot:seeded:{today}", "1")
+            force_seed = (
+                repo.kv_get(f"snapshot:seeded:{today}") == "1"
+                and cli._reached_daily_slot("23:15", now)
+            )
+            assert force_seed is False
+        finally:
+            db.close()
+
+
 def main_test():
     test_init_and_status()
     test_missing_config()
@@ -345,7 +761,18 @@ def main_test():
     test_snapshot_backup_failure_returns_error_and_retries()
     test_candidates_and_discover_commands()
     test_backfill_followers_dry_run()
-    print("OK: CLI init/sync/snapshot/auth-check/run/status/backfill/错误路径测试全部通过")
+    test_help_lists_snapshot_seed()
+    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()
+    test_initial_daily_next_run_uses_wall_clock_not_zero()
+    test_reached_daily_slot_gates_force_seed()
+    test_restart_after_seed_schedules_snapshot_at_wall_clock()
+    test_daytime_seed_marker_does_not_force_without_slot()
+    print("OK: CLI init/sync/snapshot/snapshot-seed/auth-check/run/status/backfill/错误路径测试全部通过")
 
 
 if __name__ == "__main__":

tests/test_scheduler.py

@@ -8,7 +8,7 @@ import threading
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 sys.stdout.reconfigure(encoding="utf-8")
 
-from reaction_board.scheduler import ScheduledTask, Scheduler  # noqa: E402
+from reaction_board.scheduler import ScheduledTask, Scheduler, TaskControl  # noqa: E402
 
 
 def test_run_when_due_and_reschedule():
@@ -128,9 +128,111 @@ def test_next_run_fn_controls_reschedule():
     assert calls == [1_000.0, 1_123.0]
 
 
+
+def test_task_control_overrides_ok_and_next_run():
+    clock = [1_000.0]
+    task = ScheduledTask(
+        "defer",
+        60,
+        lambda: TaskControl(ok=False, next_run=1_500.0, error="quiet"),
+    )
+    scheduler = Scheduler([task], clock=lambda: clock[0])
+    results = scheduler.run_due()
+    assert len(results) == 1
+    assert results[0].ok is False
+    assert results[0].error == "quiet"
+    assert task.next_run == 1_500.0
+    # 未到 next_run 不重跑
+    clock[0] = 1_499.0
+    assert scheduler.run_due() == []
+    clock[0] = 1_500.0
+    task.fn = lambda: TaskControl(ok=True, next_run=2_000.0)
+    results = scheduler.run_due()
+    assert results[0].ok is True
+    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: 每日快照、断点续跑、失败占位与粉丝采集测试全部通过")

tests/test_timeutil.py

@@ -11,7 +11,10 @@ sys.stdout.reconfigure(encoding="utf-8")
 from reaction_board.timeutil import (  # noqa: E402
     SHANGHAI_TZ,
     day_bounds,
+    in_snapshot_quiet_window,
+    interval_next_run,
     is_date,
+    next_after_snapshot_quiet_window,
     next_daily_run,
     parse_hhmm,
     shanghai_date,
@@ -72,11 +75,40 @@ def test_day_bounds_and_date_validation():
         raise AssertionError("非法日期应失败")
 
 
+
+def test_snapshot_quiet_window():
+    assert in_snapshot_quiet_window(ts("2026-11-02 23:00:00"))
+    assert in_snapshot_quiet_window(ts("2026-11-02 23:59:59"))
+    assert in_snapshot_quiet_window(ts("2026-11-03 00:00:00"))
+    assert in_snapshot_quiet_window(ts("2026-11-03 00:59:59"))
+    assert not in_snapshot_quiet_window(ts("2026-11-03 01:00:00"))
+    assert not in_snapshot_quiet_window(ts("2026-11-02 22:59:59"))
+    assert not in_snapshot_quiet_window(ts("2026-11-02 12:00:00"))
+
+    assert next_after_snapshot_quiet_window(ts("2026-11-02 22:00:00")) == ts(
+        "2026-11-02 22:00:00"
+    )
+    assert next_after_snapshot_quiet_window(ts("2026-11-02 23:15:00")) == ts(
+        "2026-11-03 01:00:00"
+    )
+    assert next_after_snapshot_quiet_window(ts("2026-11-03 00:30:00")) == ts(
+        "2026-11-03 01:00:00"
+    )
+    # 22:50 + 60min = 23:50 → 推到次日 01:00
+    assert interval_next_run(ts("2026-11-02 22:50:00"), 3600) == ts(
+        "2026-11-03 01:00:00"
+    )
+    assert interval_next_run(ts("2026-11-02 12:00:00"), 3600) == ts(
+        "2026-11-02 13:00:00"
+    )
+
+
 def main():
     test_shanghai_date_boundary()
     test_parse_hhmm()
     test_next_daily_run()
     test_day_bounds_and_date_validation()
+    test_snapshot_quiet_window()
     print("OK: 上海时区与每日时间计算测试全部通过")
 
 

tests/test_web_queries.py

@@ -610,12 +610,15 @@ def test_growth_auto_falls_back_to_one_day():
 
 def test_snapshot_recent_next_run_follows_completion():
     import json
+    from datetime import datetime
+    from reaction_board.timeutil import SHANGHAI_TZ, interval_next_run
 
     with tempfile.TemporaryDirectory() as tmp:
         db, q = setup(tmp)
         try:
             repo = Repository(db)
-            finished = FIXED_NOW - 10
+            # 午后完成：间隔后仍在静默窗外，应严格 finished+3600
+            finished = int(datetime(2026, 10, 31, 15, 0, 0, tzinfo=SHANGHAI_TZ).timestamp())
             repo.kv_set(
                 "task:last:snapshot_recent",
                 json.dumps({
@@ -627,6 +630,128 @@ def test_snapshot_recent_next_run_follows_completion():
             DASHBOARD_CACHE.clear()
             system = q.system()
             assert system["next_runs"]["snapshot_recent"] == finished + 3600
+            assert system["next_runs"]["snapshot_recent"] == int(
+                interval_next_run(float(finished), 3600)
+            )
+        finally:
+            db.close()
+
+
+def test_snapshot_recent_next_run_defers_quiet_window_and_retry_on_skip():
+    import json
+    from datetime import datetime
+    from reaction_board.timeutil import SHANGHAI_TZ, next_after_snapshot_quiet_window
+
+    with tempfile.TemporaryDirectory() as tmp:
+        db, q = setup(tmp)
+        try:
+            repo = Repository(db)
+            # 22:50 成功 → 下一档 23:50 落入静默窗 → 推到 01:00
+            finished = int(datetime(2026, 10, 31, 22, 50, 0, tzinfo=SHANGHAI_TZ).timestamp())
+            repo.kv_set(
+                "task:last:snapshot_recent",
+                json.dumps({
+                    "name": "snapshot_recent", "ok": True,
+                    "started_at": finished - 5, "finished_at": finished,
+                    "error": None, "summary": {},
+                }),
+            )
+            DASHBOARD_CACHE.clear()
+            system = q.system()
+            expected = int(datetime(2026, 11, 1, 1, 0, 0, tzinfo=SHANGHAI_TZ).timestamp())
+            assert system["next_runs"]["snapshot_recent"] == expected
+
+            # skip 记录：补跑到窗后 01:00，不得当成功推进
+            quiet = int(datetime(2026, 10, 31, 23, 20, 0, tzinfo=SHANGHAI_TZ).timestamp())
+            repo.kv_set(
+                "task:last:snapshot_recent",
+                json.dumps({
+                    "name": "snapshot_recent", "ok": False, "skipped": True,
+                    "started_at": quiet, "finished_at": quiet,
+                    "error": "snapshot_quiet_window",
+                    "summary": {"skipped": True, "reason": "snapshot_quiet_window"},
+                }),
+            )
+            DASHBOARD_CACHE.clear()
+            system = q.system()
+            assert system["next_runs"]["snapshot_recent"] == int(
+                next_after_snapshot_quiet_window(float(quiet))
+            )
+            assert not any(e["task"] == "snapshot_recent" for e in system["errors"])
+        finally:
+            db.close()
+
+
+def test_snapshot_failed_next_run_retries_soon_not_next_day():
+    import json
+    from datetime import datetime
+    from reaction_board.timeutil import SHANGHAI_TZ, next_daily_run
+
+    with tempfile.TemporaryDirectory() as tmp:
+        db, q = setup(tmp)
+        try:
+            repo = Repository(db)
+            finished = int(datetime(2026, 10, 31, 23, 15, 0, tzinfo=SHANGHAI_TZ).timestamp())
+            repo.kv_set(
+                "task:last:snapshot",
+                json.dumps({
+                    "name": "snapshot", "ok": False,
+                    "started_at": finished - 1, "finished_at": finished,
+                    "error": "boom", "summary": {"completed": False},
+                }),
+            )
+            DASHBOARD_CACHE.clear()
+            system = q.system()
+            assert system["next_runs"]["snapshot"] == finished + 300
+            assert system["next_runs"]["snapshot"] != int(
+                next_daily_run(finished, "23:15")
+            )
+            assert any(e["task"] == "snapshot" for e in system["errors"])
+        finally:
+            db.close()
+
+
+def test_snapshot_seed_skip_next_run_retries_ten_minutes():
+    import json
+    from datetime import datetime
+    from reaction_board.timeutil import SHANGHAI_TZ, next_daily_run
+
+    with tempfile.TemporaryDirectory() as tmp:
+        db, q = setup(tmp)
+        try:
+            repo = Repository(db)
+            finished = int(datetime(2026, 11, 1, 0, 0, 0, tzinfo=SHANGHAI_TZ).timestamp())
+            repo.kv_set(
+                "task:last:snapshot_seed",
+                json.dumps({
+                    "name": "snapshot_seed", "ok": False, "skipped": True,
+                    "started_at": finished - 1, "finished_at": finished,
+                    "error": "snapshot_running",
+                    "summary": {"skipped": True, "reason": "snapshot_running"},
+                }),
+            )
+            DASHBOARD_CACHE.clear()
+            system = q.system()
+            assert system["next_runs"]["snapshot_seed"] == finished + 600
+            assert system["next_runs"]["snapshot_seed"] != int(
+                next_daily_run(finished, "00:00")
+            )
+            # skip 不算 errors
+            assert not any(e["task"] == "snapshot_seed" for e in system["errors"])
+
+            # no_source 同样 10 分钟
+            repo.kv_set(
+                "task:last:snapshot_seed",
+                json.dumps({
+                    "name": "snapshot_seed", "ok": False, "skipped": True,
+                    "started_at": finished - 1, "finished_at": finished,
+                    "error": "no_source",
+                    "summary": {"skipped": True, "reason": "no_source"},
+                }),
+            )
+            DASHBOARD_CACHE.clear()
+            system = q.system()
+            assert system["next_runs"]["snapshot_seed"] == finished + 600
         finally:
             db.close()
 
@@ -882,6 +1007,9 @@ def main_test():
     test_system_tasks_sorted_by_started_at_desc()
     test_growth_auto_falls_back_to_one_day()
     test_snapshot_recent_next_run_follows_completion()
+    test_snapshot_recent_next_run_defers_quiet_window_and_retry_on_skip()
+    test_snapshot_failed_next_run_retries_soon_not_next_day()
+    test_snapshot_seed_skip_next_run_retries_ten_minutes()
     test_overview_adjacent_previous_with_seed_and_dual_fields()
     test_same_day_growth_baselines()
     test_growth_rate_sort_uses_prior()
@@ -948,13 +1076,15 @@ def test_system_status_uses_completed_date_and_running_flags():
 
 def test_system_next_runs_follow_last_completion_not_wall_clock():
     import json
+    from datetime import datetime
+    from reaction_board.timeutil import SHANGHAI_TZ
 
     with tempfile.TemporaryDirectory() as tmp:
         db, q = setup(tmp)
         try:
             repo = Repository(db)
-            # 发现任务刚完成不久：下次 = finished + 60min，即使 now 已超过也不跳档
-            finished = FIXED_NOW - 30
+            # 午后完成：下次 = finished + 60min，不跳档、不误入静默窗推算
+            finished = int(datetime(2026, 10, 31, 15, 0, 0, tzinfo=SHANGHAI_TZ).timestamp())
             repo.kv_set(
                 "task:last:discovery",
                 json.dumps({
@@ -966,7 +1096,6 @@ def test_system_next_runs_follow_last_completion_not_wall_clock():
             DASHBOARD_CACHE.clear()
             system = q.system()
             assert system["next_runs"]["discover"] == finished + 3600
-            # 仍小于/等于 now 也可以（表示本轮后的下一档已到期或未到），但绝不能跳到 +7200
             assert system["next_runs"]["discover"] != finished + 7200
         finally:
             db.close()