feat(scheduler): persist task state/progress/due_at across restart (#8)

Replace process-start next_run=0 guesses with kv task:state records (status, due_at, stat_date, progress). Startup recovers stale running, honors future due_at for delayed retries, and resumes the same stat_date. Seed only enqueues evening seal; seal force keeps non-seed ok=1 as progress (keep_scraped). Keeps !7 fail-logging and cleanup retry.
2026-11-03 01:50:00 +08:00
1 parent 51ffa44577
commit f408da93c2
@@ -40,6 +40,7 @@ from .services.library import LibraryService
 from .services.snapshot import SnapshotService
 from .services.telegram_poller import TelegramPoller
 from .services.up_profile import UpProfileService
+from . import task_state as task_state_mod
 from .timeutil import (
    in_snapshot_quiet_window,
    interval_next_run,
@@ -164,17 +165,29 @@ def _task_running_key(name: str) -> str:
    return f"task:running:{name}"


-def _mark_task_running(repo: Repository, name: str, *, started_at: int | None = None) -> None:
-    """任务开始时写入 running 标记，供 /system 显示「运行中」。"""
-    now = int(time.time())
+def _mark_task_running(
+    repo: Repository,
+    name: str,
+    *,
+    started_at: int | None = None,
+    stat_date: str | None = None,
+) -> None:
+    """任务开始：running 标记 + 持久化 status=running（跨重启可续跑）。"""
+    now = int(started_at if started_at is not None else time.time())
    record = {
        "name": name,
-        "started_at": int(started_at if started_at is not None else now),
+        "started_at": now,
    }
    try:
        repo.kv_set(_task_running_key(name), json.dumps(record, ensure_ascii=False))
    except Exception:  # noqa: BLE001
        pass
+    try:
+        task_state_mod.mark_running(
+            repo, name, stat_date=stat_date, now=float(now),
+        )
+    except Exception:  # noqa: BLE001
+        pass


 def _clear_task_running(repo: Repository, name: str) -> None:
@@ -202,20 +215,26 @@ def _record_task(
    payload=None,
    error: str | None = None,
    skipped: bool = False,
+    next_due_at: float | None = None,
+    stat_date: str | None = None,
+    progress=None,
 ) -> None:
    """把最近一次任务结果写入 kv，并清除 running 标记。

    ``skipped=True`` 表示主动延后/避让，不得记成成功，也不推进「当日已完成」。
+    同步写入 ``task:state``：有 ``next_due_at`` 用其排程；否则结束残留
+    ``running`` 并保留原 ``due_at``（CLI 一锤子路径）。
    """
    now = int(time.time())
    if skipped:
        ok = False
+    finished = int(finished_at if finished_at is not None else now)
    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),
+        "finished_at": finished,
        "error": error,
        "summary": payload,
    }
@@ -224,6 +243,52 @@ def _record_task(
    except Exception:  # noqa: BLE001  任务状态记录失败不影响主流程
        pass
    _clear_task_running(repo, name)
+    try:
+        if next_due_at is not None:
+            due = float(next_due_at)
+        else:
+            # CLI 等非调度路径：结束 running，保留原 due_at，避免残留 running。
+            prev = task_state_mod.load_state(repo, name)
+            if prev is None or prev.status != task_state_mod.STATUS_RUNNING:
+                return
+            due = (
+                float(prev.due_at)
+                if prev.due_at and float(prev.due_at) > 0
+                else float(finished)
+            )
+            if stat_date is None:
+                stat_date = prev.stat_date
+        if skipped:
+            task_state_mod.mark_deferred(
+                repo,
+                name,
+                due_at=due,
+                now=float(finished),
+                stat_date=stat_date,
+                progress=progress,
+                error=error,
+            )
+        elif ok:
+            task_state_mod.mark_done(
+                repo,
+                name,
+                due_at=due,
+                now=float(finished),
+                stat_date=stat_date,
+                progress=progress,
+            )
+        else:
+            task_state_mod.mark_failed(
+                repo,
+                name,
+                due_at=due,
+                now=float(finished),
+                stat_date=stat_date,
+                progress=progress,
+                error=error,
+            )
+    except Exception:  # noqa: BLE001
+        pass


 # 全站快照 / 粉丝失败后的补跑间隔（秒）
@@ -245,6 +310,7 @@ def _quiet_window_defer(
    """静默窗内不执行：记 skip（非成功），并排到 01:00 之后。"""
    finished = int(time.time())
    logger.info(f"{name}.skip", reason=_QUIET_SKIP_REASON)
+    next_run = next_after_snapshot_quiet_window(float(finished))
    _record_task(
        repo,
        name,
@@ -254,10 +320,11 @@ def _quiet_window_defer(
        finished_at=finished,
        payload={"skipped": True, "reason": _QUIET_SKIP_REASON},
        error=_QUIET_SKIP_REASON,
+        next_due_at=next_run,
    )
    return TaskControl(
        ok=False,
-        next_run=next_after_snapshot_quiet_window(float(finished)),
+        next_run=next_run,
        error=_QUIET_SKIP_REASON,
    )

@@ -281,6 +348,7 @@ def _defer_snapshot_seed(
    """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)
+    next_run = _seed_retry_soon(float(finished))
    _record_task(
        repo,
        "snapshot_seed",
@@ -290,10 +358,11 @@ def _defer_snapshot_seed(
        finished_at=finished,
        payload={"skipped": True, "reason": reason},
        error=reason,
+        next_due_at=next_run,
    )
    return TaskControl(
        ok=False,
-        next_run=_seed_retry_soon(float(finished)),
+        next_run=next_run,
        error=reason,
    )

@@ -306,35 +375,17 @@ def _initial_daily_next_run(
    now: float | None = None,
    retry_seconds: float | None = None,
 ) -> float:
-    """进程启动时日更任务的下次运行：墙钟下一档，或与系统页一致的上次记录推导。
+    """进程启动时日更任务的下次运行：优先 ``task:state.due_at``。

-    不得返回 ``0`` / 过去时间来“立刻跑一遍”——午夜 seed 后重启否则会白天全量强采。
+    不得返回 ``0`` / 用「无状态」以外的过去时间去误伤——无状态时对齐墙钟。
+    失败且 ``due_at`` 已过：返回该 ``due_at``（调度器会立刻补跑），不退回次日墙钟。
    """
+    del retry_seconds  # 间隔已落在 task:state.due_at
    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
+    return task_state_mod.startup_next_run(
+        repo, name, wall_due_at=wall, now=current,
+    )


 def _reached_daily_slot(at: str, now: float | None = None) -> bool:
@@ -345,6 +396,38 @@ def _reached_daily_slot(at: str, now: float | None = None) -> bool:
    return current >= slot


+def _evening_due_at(now: float, snapshot_at: str) -> float:
+    """尚未到当日 snapshot_at 则挂今晚，否则下一档墙钟。"""
+    hour, minute = parse_hhmm(snapshot_at)
+    dt = shanghai_datetime(now)
+    slot = dt.replace(hour=hour, minute=minute, second=0, microsecond=0).timestamp()
+    if slot > now:
+        return float(slot)
+    return float(next_daily_run(now, snapshot_at))
+
+
+def _enqueue_evening_seal(
+    repo: Repository,
+    *,
+    stat_date: str,
+    snapshot_at: str,
+    now: float,
+) -> None:
+    """seed 成功后：当晚 snapshot / fans 记 pending，due_at=当日 snapshot_at。"""
+    current = shanghai_datetime(now)
+    hour, minute = parse_hhmm(snapshot_at)
+    slot_ts = current.replace(
+        hour=hour, minute=minute, second=0, microsecond=0,
+    ).timestamp()
+    due = slot_ts if slot_ts > now else now
+    task_state_mod.ensure_pending(
+        repo, "snapshot", due_at=due, stat_date=stat_date, now=now,
+    )
+    task_state_mod.ensure_pending(
+        repo, "fans", due_at=due, stat_date=stat_date, now=now,
+    )
+
+
 def _client(credentials, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient:
    return BilibiliClient(
        cookie=credentials.cookie,
@@ -445,51 +528,120 @@ def cmd_snapshot(args, logger: JsonLogger) -> int:
        )
        service = SnapshotService(store, repo, client, logger=logger)
        started = int(time.time())
-        _mark_task_running(repo, "snapshot", started_at=started)
+        hh, mm = parse_hhmm(cfg.schedule.snapshot_at)
+        snapshot_at = f"{hh:02d}:{mm:02d}"
+        _mark_task_running(
+            repo, "snapshot", started_at=started,
+            stat_date=shanghai_date(float(started)),
+        )
        try:
            video_summary = service.run_if_needed(force=args.force)
        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),
+                next_due_at=_retry_soon(float(finished)),
+                stat_date=shanghai_date(float(started)),
            )
            raise
        video_payload = video_summary.as_dict()
        video_payload["database"] = str(cfg.storage.database_path)
        video_payload["snapshot_counts"] = repo.snapshot_counts(video_summary.stat_date)
        video_payload["counts"] = repo.counts()
+        video_finished = int(video_summary.finished_at or time.time())
        video_ok = bool(video_summary.completed and not video_summary.backup_error)
-        _record_task(
-            repo, "snapshot", ok=video_ok,
-            started_at=int(video_summary.started_at or started),
-            finished_at=int(video_summary.finished_at or time.time()),
-            payload=video_payload,
-            error=video_summary.backup_error,
-        )
+        if video_summary.reason == "seeded_pending_seal":
+            evening = _evening_due_at(float(started), snapshot_at)
+            _enqueue_evening_seal(
+                repo,
+                stat_date=video_summary.stat_date,
+                snapshot_at=snapshot_at,
+                now=float(video_finished),
+            )
+            _record_task(
+                repo, "snapshot", ok=False, skipped=True,
+                started_at=int(video_summary.started_at or started),
+                finished_at=video_finished,
+                payload=video_payload,
+                error="seeded_pending_seal",
+                next_due_at=evening,
+                stat_date=video_summary.stat_date,
+            )
+        elif video_ok:
+            _record_task(
+                repo, "snapshot", ok=True,
+                started_at=int(video_summary.started_at or started),
+                finished_at=video_finished,
+                payload=video_payload,
+                error=video_summary.backup_error,
+                next_due_at=next_daily_run(float(video_finished), snapshot_at),
+                stat_date=video_summary.stat_date,
+            )
+        else:
+            _record_task(
+                repo, "snapshot", ok=False,
+                started_at=int(video_summary.started_at or started),
+                finished_at=video_finished,
+                payload=video_payload,
+                error=video_summary.backup_error or "snapshot_incomplete",
+                next_due_at=_retry_soon(float(video_finished)),
+                stat_date=video_summary.stat_date,
+            )

        fans_started = int(time.time())
-        _mark_task_running(repo, "fans", started_at=fans_started)
+        _mark_task_running(
+            repo, "fans", started_at=fans_started,
+            stat_date=video_summary.stat_date,
+        )
        try:
            fans_summary = service.run_fans_if_needed(
                date=video_summary.stat_date, force=args.force,
            )
        except Exception as exc:  # noqa: BLE001
+            finished = int(time.time())
            _record_task(
                repo, "fans", ok=False, started_at=fans_started,
-                finished_at=int(time.time()), error=str(exc),
+                finished_at=finished, error=str(exc),
+                next_due_at=_retry_soon(float(finished)),
+                stat_date=video_summary.stat_date,
            )
            raise
        fans_payload = fans_summary.as_dict()
+        fans_finished = int(fans_summary.finished_at or time.time())
        fans_ok = fans_summary.fans_failed == 0
-        _record_task(
-            repo, "fans", ok=fans_ok,
-            started_at=int(fans_summary.started_at or fans_started),
-            finished_at=int(fans_summary.finished_at or time.time()),
-            payload=fans_payload,
-            error="；".join(
-                err for err in fans_summary.errors if str(err).startswith("fans ")
-            ) or None,
-        )
+        if fans_summary.reason == "seeded_pending_seal":
+            evening = _evening_due_at(float(fans_started), snapshot_at)
+            _record_task(
+                repo, "fans", ok=False, skipped=True,
+                started_at=int(fans_summary.started_at or fans_started),
+                finished_at=fans_finished,
+                payload=fans_payload,
+                error="seeded_pending_seal",
+                next_due_at=evening,
+                stat_date=video_summary.stat_date,
+            )
+        elif fans_ok:
+            _record_task(
+                repo, "fans", ok=True,
+                started_at=int(fans_summary.started_at or fans_started),
+                finished_at=fans_finished,
+                payload=fans_payload,
+                next_due_at=next_daily_run(float(fans_finished), snapshot_at),
+                stat_date=video_summary.stat_date,
+            )
+        else:
+            _record_task(
+                repo, "fans", ok=False,
+                started_at=int(fans_summary.started_at or fans_started),
+                finished_at=fans_finished,
+                payload=fans_payload,
+                error="；".join(
+                    err for err in fans_summary.errors if str(err).startswith("fans ")
+                ) or None,
+                next_due_at=_retry_soon(float(fans_finished)),
+                stat_date=video_summary.stat_date,
+            )
        payload = {
            **video_payload,
            "fans_ok": fans_summary.fans_ok,
@@ -525,32 +677,45 @@ def cmd_snapshot_seed(args, logger: JsonLogger) -> int:
                "stat_date": target or shanghai_date(),
                "database": str(cfg.storage.database_path),
            }
+            finished = int(time.time())
            _record_task(
                repo,
                "snapshot_seed",
                ok=False,
                skipped=True,
                started_at=started,
-                finished_at=int(time.time()),
+                finished_at=finished,
                payload=payload,
                error=_SEED_SKIP_SNAPSHOT_RUNNING,
+                next_due_at=_seed_retry_soon(float(finished)),
+                stat_date=payload["stat_date"],
            )
            _print(payload)
            return 1
        # seed 只读本地库，不调 B 站；占位 client 满足构造约束。
        service = SnapshotService(store, repo, object(), logger=logger)
-        _mark_task_running(repo, "snapshot_seed", started_at=started)
+        seed_date = target or shanghai_date()
+        _mark_task_running(
+            repo, "snapshot_seed", started_at=started, stat_date=seed_date,
+        )
        try:
            summary = service.run_seed_from_last_completed(date=target)
        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),
+                next_due_at=_seed_retry_soon(float(finished)),
+                stat_date=seed_date,
            )
            raise
        payload = summary.as_dict()
        payload["database"] = str(cfg.storage.database_path)
        finished = int(summary.finished_at or time.time())
+        hh, mm = parse_hhmm(cfg.schedule.seed_at)
+        seed_at = f"{hh:02d}:{mm:02d}"
+        sh, sm = parse_hhmm(cfg.schedule.snapshot_at)
+        snapshot_at = f"{sh:02d}:{sm:02d}"
        if summary.reason == _SEED_SKIP_NO_SOURCE:
            _record_task(
                repo,
@@ -561,9 +726,12 @@ def cmd_snapshot_seed(args, logger: JsonLogger) -> int:
                finished_at=finished,
                payload={**payload, "skipped": True, "reason": _SEED_SKIP_NO_SOURCE},
                error=_SEED_SKIP_NO_SOURCE,
+                next_due_at=_seed_retry_soon(float(finished)),
+                stat_date=summary.stat_date or seed_date,
            )
            _print(payload)
            return 1
+        next_due = next_daily_run(float(finished), seed_at)
        _record_task(
            repo,
            "snapshot_seed",
@@ -571,7 +739,16 @@ def cmd_snapshot_seed(args, logger: JsonLogger) -> int:
            started_at=int(summary.started_at or started),
            finished_at=finished,
            payload=payload,
+            next_due_at=next_due,
+            stat_date=summary.stat_date or seed_date,
        )
+        if summary.reason == "seeded":
+            _enqueue_evening_seal(
+                repo,
+                stat_date=summary.stat_date or seed_date,
+                snapshot_at=snapshot_at,
+                now=float(finished),
+            )
        _print(payload)
        return 0
    finally:
@@ -609,24 +786,34 @@ def cmd_discover(args, logger: JsonLogger) -> int:
            logger=logger,
        )
        started = int(time.time())
+        interval = float(cfg.schedule.discover_minutes * 60)
        _mark_task_running(repo, "discovery", started_at=started)
        try:
            summary = service.run()
        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),
+                next_due_at=_retry_soon(float(finished)),
            )
            raise
        payload = summary.as_dict()
        payload["candidate_total"] = len(repo.list_candidates())
        payload["database"] = str(cfg.storage.database_path)
+        finished = int(summary.finished_at or time.time())
+        ok = not summary.errors
+        next_due = (
+            interval_next_run(float(finished), interval) if ok
+            else _retry_soon(float(finished))
+        )
        _record_task(
-            repo, "discovery", ok=not summary.errors,
+            repo, "discovery", ok=ok,
            started_at=int(summary.started_at or started),
-            finished_at=int(summary.finished_at or time.time()),
+            finished_at=finished,
            payload=payload,
            error="；".join(summary.errors) if summary.errors else None,
+            next_due_at=next_due,
        )
        _print(payload)
        return 0 if not summary.errors else 1
@@ -747,29 +934,32 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
                summary = discovery_service.run()
            except Exception as exc:  # noqa: BLE001  记录后补跑，不抛给次日
                finished = int(time.time())
+                next_run = _retry_soon(float(finished))
                _record_task(
                    repo, "discovery", ok=False, started_at=started,
                    finished_at=finished, error=str(exc),
+                    next_due_at=next_run,
                )
                return TaskControl(
-                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
+                    ok=False, next_run=next_run, error=str(exc),
                )
            payload = summary.as_dict()
            ok = not summary.errors
            finished = int(summary.finished_at or time.time())
+            next_run = (
+                interval_next_run(float(finished), interval) if ok
+                else _retry_soon(float(finished))
+            )
            _record_task(
                repo, "discovery", ok=ok,
                started_at=summary.started_at,
                finished_at=finished,
                payload=payload,
                error="；".join(summary.errors) if summary.errors else None,
+                next_due_at=next_run,
            )
            last["discover"] = summary
            logger.info("discovery.summary", **payload)
-            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():
@@ -785,12 +975,14 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
                expired = repo.expire_fav_actions()
            except Exception as exc:  # noqa: BLE001  记录后约 5 分钟补跑
                finished = int(time.time())
+                next_run = _retry_soon(float(finished))
                _record_task(
                    repo, "candidate_cleanup", ok=False, started_at=started,
                    finished_at=finished, error=str(exc),
+                    next_due_at=next_run,
                )
                return TaskControl(
-                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
+                    ok=False, next_run=next_run, error=str(exc),
                )
            if removed or expired:
                repo.bump_cache_version("videos")
@@ -799,6 +991,7 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
                "cleanup.done",
                candidates_removed=removed, fav_actions_expired=expired,
            )
+            next_run = interval_next_run(float(finished), 86400.0)
            _record_task(
                repo, "candidate_cleanup", ok=True, started_at=started,
                finished_at=finished,
@@ -806,48 +999,90 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
                    "candidates_removed": removed,
                    "fav_actions_expired": expired,
                },
+                next_due_at=next_run,
            )
-            return TaskControl(
-                ok=True,
-                next_run=interval_next_run(float(finished), 86400.0),
-            )
+            return TaskControl(ok=True, next_run=next_run)

        def run_snapshot():
            """全站快照：不因其它任务 running 而 skip；失败/未完成则短间隔补跑。"""
            started = int(time.time())
-            _mark_task_running(repo, "snapshot", started_at=started)
+            today = shanghai_date(float(started))
+            stat_date = task_state_mod.resume_stat_date(
+                repo, "snapshot", fallback=today,
+            )
+            _mark_task_running(
+                repo, "snapshot", started_at=started, stat_date=stat_date,
+            )
            try:
-                # 仅晚间 snapshot_at 及之后：seed 壳才 force 打穿 ok=1 克隆行。
-                # 白天/重启不得因 seeded 标记立刻全量强采。
-                today = shanghai_date()
-                force_seed = (
-                    repo.kv_get(SnapshotService.seeded_key(today)) == "1"
-                    and _reached_daily_slot(snapshot_at, float(started))
+                # seed 壳 ≠ 白天强采：仅到达当日 snapshot_at（或显式续跑已到期的
+                # seal pending）才 force。due_at 门闩负责「会不会跑」；这里是
+                # force 的第二道闸。
+                seeded = repo.kv_get(SnapshotService.seeded_key(stat_date)) == "1"
+                force_seed = seeded and _reached_daily_slot(
+                    snapshot_at, float(started),
+                )
+                summary = snapshot_service.run_if_needed(
+                    date=stat_date, force=force_seed,
                )
-                summary = snapshot_service.run_if_needed(force=force_seed)
            except Exception as exc:  # noqa: BLE001
                finished = int(time.time())
+                next_run = _retry_soon(float(finished))
                _record_task(
                    repo, "snapshot", ok=False, started_at=started,
                    finished_at=finished, error=str(exc),
+                    next_due_at=next_run, stat_date=stat_date,
                )
                return TaskControl(
-                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
+                    ok=False, next_run=next_run, error=str(exc),
                )
            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=finished,
-                payload=payload, error=summary.backup_error,
-            )
-            last["snapshot"] = summary
-            logger.info("snapshot.summary", **payload)
+            # 白天误碰到 seed 壳：延到当晚 seal，禁止 5 分钟空转。
+            if (
+                not ok
+                and getattr(summary, "reason", None) == "seeded_pending_seal"
+            ):
+                current = shanghai_datetime(float(finished))
+                hour, minute = parse_hhmm(snapshot_at)
+                slot_ts = current.replace(
+                    hour=hour, minute=minute, second=0, microsecond=0,
+                ).timestamp()
+                next_run = slot_ts if slot_ts > float(finished) else _retry_soon(
+                    float(finished),
+                )
+                _record_task(
+                    repo, "snapshot", ok=False, skipped=True,
+                    started_at=summary.started_at, finished_at=finished,
+                    payload=payload, error="seeded_pending_seal",
+                    next_due_at=next_run,
+                    stat_date=summary.stat_date or stat_date,
+                )
+                last["snapshot"] = summary
+                logger.info("snapshot.summary", **payload)
+                return TaskControl(
+                    ok=False, next_run=next_run, error="seeded_pending_seal",
+                )
            next_run = (
                next_daily_run(float(finished), snapshot_at) if ok
                else _retry_soon(float(finished))
            )
+            progress = {
+                "total": summary.total,
+                "success": summary.success,
+                "failed": summary.failed,
+                "skipped": summary.skipped,
+            }
+            _record_task(
+                repo, "snapshot", ok=ok, started_at=summary.started_at,
+                finished_at=finished,
+                payload=payload, error=summary.backup_error,
+                next_due_at=next_run,
+                stat_date=summary.stat_date or stat_date,
+                progress=progress,
+            )
+            last["snapshot"] = summary
+            logger.info("snapshot.summary", **payload)
            return TaskControl(ok=ok, next_run=next_run)

        def run_snapshot_recent():
@@ -862,71 +1097,109 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
                summary = snapshot_service.run_recent_videos()
            except Exception as exc:  # noqa: BLE001
                finished = int(time.time())
+                next_run = _retry_soon(float(finished))
                _record_task(
                    repo, "snapshot_recent", ok=False, started_at=started,
                    finished_at=finished, error=str(exc),
+                    next_due_at=next_run,
                )
                return TaskControl(
-                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
+                    ok=False, next_run=next_run, error=str(exc),
                )
            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=finished,
-                payload=payload, error=error,
-            )
-            last["snapshot_recent"] = summary
-            logger.info("snapshot.recent_summary", **payload)
            next_run = (
                interval_next_run(float(finished), interval) if ok
                else _retry_soon(float(finished))
            )
+            _record_task(
+                repo, "snapshot_recent", ok=ok, started_at=summary.started_at,
+                finished_at=finished,
+                payload=payload, error=error,
+                next_due_at=next_run,
+            )
+            last["snapshot_recent"] = summary
+            logger.info("snapshot.recent_summary", **payload)
            return TaskControl(ok=ok, next_run=next_run)

        def run_fans():
            """粉丝更新短耗时：不避开任何任务，可与全量并行。"""
            started = int(time.time())
-            _mark_task_running(repo, "fans", started_at=started)
+            today = shanghai_date(float(started))
+            stat_date = task_state_mod.resume_stat_date(
+                repo, "fans", fallback=today,
+            )
+            _mark_task_running(
+                repo, "fans", started_at=started, stat_date=stat_date,
+            )
            try:
-                today = shanghai_date()
-                force_seed = (
-                    repo.kv_get(SnapshotService.fans_seeded_key(today)) == "1"
-                    and _reached_daily_slot(snapshot_at, float(started))
+                seeded = (
+                    repo.kv_get(SnapshotService.fans_seeded_key(stat_date)) == "1"
+                )
+                force_seed = seeded and _reached_daily_slot(
+                    snapshot_at, float(started),
+                )
+                summary = snapshot_service.run_fans_if_needed(
+                    date=stat_date, force=force_seed,
                )
-                summary = snapshot_service.run_fans_if_needed(force=force_seed)
            except Exception as exc:  # noqa: BLE001
                finished = int(time.time())
+                next_run = _retry_soon(float(finished))
                _record_task(
                    repo, "fans", ok=False, started_at=started,
                    finished_at=finished, error=str(exc),
+                    next_due_at=next_run, stat_date=stat_date,
                )
                return TaskControl(
-                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
+                    ok=False, next_run=next_run, error=str(exc),
                )
            payload = summary.as_dict()
+            finished = int(summary.finished_at or time.time())
+            if getattr(summary, "reason", None) == "seeded_pending_seal":
+                current = shanghai_datetime(float(finished))
+                hour, minute = parse_hhmm(snapshot_at)
+                slot_ts = current.replace(
+                    hour=hour, minute=minute, second=0, microsecond=0,
+                ).timestamp()
+                next_run = slot_ts if slot_ts > float(finished) else _retry_soon(
+                    float(finished),
+                )
+                _record_task(
+                    repo, "fans", ok=False, skipped=True,
+                    started_at=summary.started_at, finished_at=finished,
+                    payload=payload, error="seeded_pending_seal",
+                    next_due_at=next_run,
+                    stat_date=summary.stat_date or stat_date,
+                )
+                last["fans"] = summary
+                logger.info("fans.summary", **payload)
+                return TaskControl(
+                    ok=False, next_run=next_run, error="seeded_pending_seal",
+                )
            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=finished,
-                payload=payload, error=error,
-            )
-            last["fans"] = summary
-            logger.info("fans.summary", **payload)
            next_run = (
                next_daily_run(float(finished), snapshot_at) if ok
                else _retry_soon(float(finished))
            )
+            _record_task(
+                repo, "fans", ok=ok, started_at=summary.started_at,
+                finished_at=finished,
+                payload=payload, error=error,
+                next_due_at=next_run,
+                stat_date=summary.stat_date or stat_date,
+            )
+            last["fans"] = summary
+            logger.info("fans.summary", **payload)
            return TaskControl(ok=ok, next_run=next_run)

        def run_snapshot_seed():
            started = int(time.time())
+            today = shanghai_date(float(started))
            # 全站快照占用时不执行、不记成功；10 分钟后重试直到真正 seed 成功。
            if _is_task_running(repo, "snapshot"):
                return _defer_snapshot_seed(
@@ -935,18 +1208,22 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
                    started_at=started,
                    reason=_SEED_SKIP_SNAPSHOT_RUNNING,
                )
-            _mark_task_running(repo, "snapshot_seed", started_at=started)
+            _mark_task_running(
+                repo, "snapshot_seed", started_at=started, stat_date=today,
+            )
            try:
-                summary = snapshot_service.run_seed_from_last_completed()
+                summary = snapshot_service.run_seed_from_last_completed(date=today)
            except Exception as exc:  # noqa: BLE001
                finished = int(time.time())
+                next_run = _seed_retry_soon(float(finished))
                _record_task(
                    repo, "snapshot_seed", ok=False, started_at=started,
                    finished_at=finished, error=str(exc),
+                    next_due_at=next_run, stat_date=today,
                )
                return TaskControl(
                    ok=False,
-                    next_run=_seed_retry_soon(float(finished)),
+                    next_run=next_run,
                    error=str(exc),
                )
            payload = summary.as_dict()
@@ -962,14 +1239,22 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
                    finished_at=finished,
                )
            # already_completed / already_seeded / seeded → 成功，推到次日 seed_at
+            next_run = next_daily_run(float(finished), seed_at)
            _record_task(
                repo, "snapshot_seed", ok=True, started_at=summary.started_at,
                finished_at=finished,
                payload=payload,
+                next_due_at=next_run,
+                stat_date=summary.stat_date or today,
            )
-            return TaskControl(
-                ok=True, next_run=next_daily_run(float(finished), seed_at),
-            )
+            if summary.reason == "seeded":
+                _enqueue_evening_seal(
+                    repo,
+                    stat_date=summary.stat_date or today,
+                    snapshot_at=snapshot_at,
+                    now=float(finished),
+                )
+            return TaskControl(ok=True, next_run=next_run)

        def run_config_watch():
            before = store.revision
@@ -978,10 +1263,11 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
                logger.info("config.reloaded", revision=store.revision)
            return {"revision": store.revision}

-        # --once：保持 next_run=0 立刻跑到期逻辑；常驻：日更对齐墙钟，避免重启即强采。
+        # --once：next_run=0 立刻跑；常驻：一律从 task:state.due_at 恢复，禁止日更 next_run=0。
        now_ts = time.time()
        if args.once:
            seed_next = snapshot_next = fans_next = 0.0
+            recent_next = discover_next = cleanup_next = 0.0
        else:
            seed_next = _initial_daily_next_run(
                repo, "snapshot_seed", seed_at, now=now_ts,
@@ -992,6 +1278,15 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
            fans_next = _initial_daily_next_run(
                repo, "fans", snapshot_at, now=now_ts,
            )
+            recent_next = task_state_mod.startup_next_run(
+                repo, "snapshot_recent", wall_due_at=now_ts, now=now_ts,
+            )
+            discover_next = task_state_mod.startup_next_run(
+                repo, "discovery", wall_due_at=now_ts, now=now_ts,
+            )
+            cleanup_next = task_state_mod.startup_next_run(
+                repo, "candidate_cleanup", wall_due_at=now_ts, now=now_ts,
+            )
        scheduler = Scheduler(
            [
                ScheduledTask(
@@ -1016,13 +1311,20 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
                    "snapshot_recent",
                    float(recent_minutes * 60),
                    run_snapshot_recent,
+                    next_run=recent_next,
                ),
                ScheduledTask(
                    "discovery",
                    float(discover_minutes * 60),
                    run_discover,
+                    next_run=discover_next,
+                ),
+                ScheduledTask(
+                    "candidate_cleanup",
+                    86400.0,
+                    run_cleanup,
+                    next_run=cleanup_next,
                ),
-                ScheduledTask("candidate_cleanup", 86400.0, run_cleanup),
                ScheduledTask("config_watch", 60.0, run_config_watch),
            ],
            logger=logger,
@@ -155,7 +155,10 @@ class SnapshotService:
            raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}")
        seeded = self.repo.kv_get(self.seeded_key(stat_date)) == "1"
        if force:
-            return self.run_videos(date=stat_date, force=True)
+            # 晚间 seal / 续跑：打穿 seed，保留已真实采集的 ok=1 作为进度。
+            return self.run_videos(
+                date=stat_date, force=True, keep_scraped=True,
+            )

        completed = self.repo.kv_get(self.completed_key(stat_date)) == "1"
        if seeded and not completed:
@@ -169,6 +172,7 @@ class SnapshotService:
                finished_at=now,
                completed=False,
                already_completed=False,
+                reason="seeded_pending_seal",
            )
        pending = self.repo.pending_snapshot_aids(stat_date)
        if completed and not pending:
@@ -205,8 +209,19 @@ class SnapshotService:
        summary = self.run_videos(date=date, force=force)
        return self.run_fans(date=summary.stat_date, force=force, summary=summary)

-    def run_videos(self, *, date: str | None = None, force: bool = False) -> SnapshotSummary:
-        """采集视频七项指标并在全部成功后标记完成日 / 备份。"""
+    def run_videos(
+        self,
+        *,
+        date: str | None = None,
+        force: bool = False,
+        keep_scraped: bool = False,
+    ) -> SnapshotSummary:
+        """采集视频七项指标并在全部成功后标记完成日 / 备份。
+
+        ``force``：忽略完成标记，重跑。
+        ``keep_scraped``：force 时仍跳过非 seed 的 ok=1（seal 断点续跑）；
+        CLI ``--force`` 不传此标志，会重采全部成功行。
+        """
        cfg = self.config_store.get()
        client = self._client()
        now = self._now()
@@ -227,11 +242,19 @@ class SnapshotService:
        summary.total = len(videos)
        for index, aid in enumerate(videos, 1):
            existing = self.repo.get_snapshot(stat_date, aid)
-            if existing is not None and int(existing.get("ok") or 0) == 1 and not force:
-                # 断点续跑：同日已成功则跳过。小时任务刷新过的行也是 ok=1，
-                # 自然覆盖「1 小时内 captured_at 已更新则不再次更新」的要求。
-                summary.skipped += 1
-                continue
+            if existing is not None and int(existing.get("ok") or 0) == 1:
+                src = str(existing.get("src") or "")
+                if keep_scraped and src == "seed":
+                    # seal 续跑：必须打穿克隆壳
+                    pass
+                elif keep_scraped and src != "seed":
+                    summary.skipped += 1
+                    continue
+                elif not force:
+                    # 普通续跑：同日已成功则跳过（含小时任务刚刷新的）
+                    summary.skipped += 1
+                    continue
+                # force 且非 keep_scraped：CLI 全量重采，不 skip

            stat, error, unavailable = self._fetch_view(client, aid)
            if unavailable:
@@ -425,6 +448,7 @@ class SnapshotService:
                    completed=False,
                    already_completed=False,
                    fans_skipped=len(cfg.track_fans),
+                    reason="seeded_pending_seal",
                )
            return self.run_fans(date=stat_date, force=True)
        if not self._pending_fan_mids(stat_date, cfg.track_fans):
@@ -0,0 +1,291 @@
+# -*- coding: utf-8 -*-
+"""调度任务状态持久化（跨进程重启 / 容器重建）。
+
+每条任务一条 kv 记录 ``task:state:{name}``，字段见 :class:`TaskState`。
+调度器只认落盘的 ``due_at`` / ``status`` / ``stat_date`` / ``progress``，
+不再靠启动时把 ``next_run=0`` 猜「要不要立刻跑」。
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from dataclasses import asdict, dataclass, field
+from typing import Any, Mapping
+
+from .repository import Repository
+
+STATUS_PENDING = "pending"
+STATUS_RUNNING = "running"
+STATUS_DONE = "done"
+STATUS_FAILED = "failed"
+STATUS_DEFERRED = "deferred"
+
+_ACTIVE = frozenset({STATUS_PENDING, STATUS_RUNNING, STATUS_FAILED, STATUS_DEFERRED})
+
+
+@dataclass
+class TaskState:
+    name: str
+    status: str
+    due_at: float
+    stat_date: str | None = None
+    progress: dict[str, Any] | None = None
+    updated_at: float = 0.0
+    error: str | None = None
+
+    def as_dict(self) -> dict[str, Any]:
+        payload = asdict(self)
+        # JSON 友好：去掉 None progress 也可保留
+        return payload
+
+
+def state_key(name: str) -> str:
+    return f"task:state:{name}"
+
+
+def load_state(repo: Repository, name: str) -> TaskState | None:
+    raw = repo.kv_get(state_key(name))
+    if not raw:
+        return None
+    try:
+        data = json.loads(raw)
+    except (TypeError, ValueError, json.JSONDecodeError):
+        return None
+    if not isinstance(data, dict) or not data.get("name"):
+        return None
+    try:
+        due_at = float(data.get("due_at") or 0)
+        updated_at = float(data.get("updated_at") or 0)
+    except (TypeError, ValueError):
+        return None
+    progress = data.get("progress")
+    if progress is not None and not isinstance(progress, dict):
+        progress = None
+    return TaskState(
+        name=str(data.get("name") or name),
+        status=str(data.get("status") or STATUS_PENDING),
+        due_at=due_at,
+        stat_date=(str(data["stat_date"]) if data.get("stat_date") else None),
+        progress=progress,
+        updated_at=updated_at,
+        error=(str(data["error"]) if data.get("error") else None),
+    )
+
+
+def save_state(repo: Repository, state: TaskState) -> None:
+    repo.kv_set(state_key(state.name), json.dumps(state.as_dict(), ensure_ascii=False))
+
+
+def _now(now: float | None) -> float:
+    return time.time() if now is None else float(now)
+
+
+def recover_stale_running(
+    repo: Repository,
+    name: str,
+    *,
+    now: float | None = None,
+) -> TaskState | None:
+    """进程已死但仍为 ``running``：改为 ``pending``，保留 progress / stat_date。
+
+    ``due_at`` 取 ``min(原 due_at, now)``，使过期/中断任务可立即续跑，
+    未到期的（极少见）仍尊重原 due_at。
+    """
+    current = _now(now)
+    state = load_state(repo, name)
+    if state is None or state.status != STATUS_RUNNING:
+        return state
+    state.status = STATUS_PENDING
+    state.due_at = min(float(state.due_at or current), current)
+    state.updated_at = current
+    save_state(repo, state)
+    return state
+
+
+def mark_running(
+    repo: Repository,
+    name: str,
+    *,
+    stat_date: str | None,
+    now: float | None = None,
+    progress: Mapping[str, Any] | None = None,
+    due_at: float | None = None,
+) -> TaskState:
+    current = _now(now)
+    prev = load_state(repo, name)
+    state = TaskState(
+        name=name,
+        status=STATUS_RUNNING,
+        due_at=float(due_at if due_at is not None else (prev.due_at if prev else current)),
+        stat_date=stat_date if stat_date is not None else (prev.stat_date if prev else None),
+        progress=dict(progress) if progress is not None else (prev.progress if prev else None),
+        updated_at=current,
+        error=None,
+    )
+    save_state(repo, state)
+    return state
+
+
+def mark_done(
+    repo: Repository,
+    name: str,
+    *,
+    due_at: float,
+    now: float | None = None,
+    stat_date: str | None = None,
+    progress: Mapping[str, Any] | None = None,
+) -> TaskState:
+    """成功：记 done，并写入下一次 ``due_at``（下一条墙钟/间隔）。"""
+    current = _now(now)
+    state = TaskState(
+        name=name,
+        status=STATUS_DONE,
+        due_at=float(due_at),
+        stat_date=stat_date,
+        progress=dict(progress) if progress is not None else None,
+        updated_at=current,
+        error=None,
+    )
+    save_state(repo, state)
+    return state
+
+
+def mark_failed(
+    repo: Repository,
+    name: str,
+    *,
+    due_at: float,
+    now: float | None = None,
+    stat_date: str | None = None,
+    progress: Mapping[str, Any] | None = None,
+    error: str | None = None,
+) -> TaskState:
+    current = _now(now)
+    prev = load_state(repo, name)
+    state = TaskState(
+        name=name,
+        status=STATUS_FAILED,
+        due_at=float(due_at),
+        stat_date=stat_date if stat_date is not None else (prev.stat_date if prev else None),
+        progress=(
+            dict(progress) if progress is not None
+            else (prev.progress if prev else None)
+        ),
+        updated_at=current,
+        error=error,
+    )
+    save_state(repo, state)
+    return state
+
+
+def mark_deferred(
+    repo: Repository,
+    name: str,
+    *,
+    due_at: float,
+    now: float | None = None,
+    stat_date: str | None = None,
+    progress: Mapping[str, Any] | None = None,
+    error: str | None = None,
+) -> TaskState:
+    """主动延后（静默窗 / seed 避让等）：保留进度，写未来 due_at。"""
+    current = _now(now)
+    prev = load_state(repo, name)
+    state = TaskState(
+        name=name,
+        status=STATUS_DEFERRED,
+        due_at=float(due_at),
+        stat_date=stat_date if stat_date is not None else (prev.stat_date if prev else None),
+        progress=(
+            dict(progress) if progress is not None
+            else (prev.progress if prev else None)
+        ),
+        updated_at=current,
+        error=error,
+    )
+    save_state(repo, state)
+    return state
+
+
+def ensure_pending(
+    repo: Repository,
+    name: str,
+    *,
+    due_at: float,
+    stat_date: str | None,
+    now: float | None = None,
+    overwrite_done: bool = True,
+) -> TaskState:
+    """保证存在一条 pending（如 seed 成功后挂上当晚 seal）。
+
+    若已有未完成状态且 due_at 更早，保留更早的 due_at；
+    ``overwrite_done`` 时覆盖已完成记录。
+    """
+    current = _now(now)
+    prev = load_state(repo, name)
+    if prev is not None and prev.status in _ACTIVE and prev.status != STATUS_RUNNING:
+        # 未完成：取更早的 due_at，补上缺失的 stat_date
+        due = min(float(prev.due_at or due_at), float(due_at))
+        sd = prev.stat_date or stat_date
+        if due == prev.due_at and sd == prev.stat_date and prev.status == STATUS_PENDING:
+            return prev
+        prev.due_at = due
+        prev.stat_date = sd
+        prev.status = STATUS_PENDING
+        prev.updated_at = current
+        save_state(repo, prev)
+        return prev
+    if prev is not None and prev.status == STATUS_RUNNING:
+        return prev
+    if prev is not None and prev.status == STATUS_DONE and not overwrite_done:
+        return prev
+    state = TaskState(
+        name=name,
+        status=STATUS_PENDING,
+        due_at=float(due_at),
+        stat_date=stat_date,
+        progress=None,
+        updated_at=current,
+        error=None,
+    )
+    save_state(repo, state)
+    return state
+
+
+def startup_next_run(
+    repo: Repository,
+    name: str,
+    *,
+    wall_due_at: float,
+    now: float | None = None,
+) -> float:
+    """常驻启动时的 ``ScheduledTask.next_run``：只认落盘状态。
+
+    - 恢复残留 ``running`` → ``pending``
+    - ``due_at > now`` → 原样返回（失败延时不会被提前）
+    - ``due_at <= now`` 且未完成 → 立刻可跑（返回 ``due_at``，可为过去时间）
+    - ``done`` 且 ``due_at > now`` → 等下一次
+    - ``done`` 且 ``due_at <= now`` → 到期，返回 ``due_at``
+    - 无状态 → 墙钟 ``wall_due_at``
+    """
+    current = _now(now)
+    state = recover_stale_running(repo, name, now=current)
+    if state is None:
+        return float(wall_due_at)
+    due = float(state.due_at or 0)
+    if due <= 0:
+        return float(wall_due_at)
+    # 无论 status：due_at 是唯一时间门闩
+    if due > current:
+        return due
+    # 已到期：未完成要续跑；done 也表示下一轮已到期
+    return due
+
+
+def resume_stat_date(repo: Repository, name: str, *, fallback: str) -> str:
+    """续跑时优先用状态里的归属日，避免跨日重启跑错天。"""
+    state = load_state(repo, name)
+    if state is not None and state.stat_date and state.status in _ACTIVE:
+        return state.stat_date
+    return fallback
@@ -1983,6 +1983,22 @@ class DashboardQueries:
            }
            retry_seconds = 300
            seed_retry_seconds = 600
+            # 优先 task:state.due_at（跨重启唯一时间门闩）；无则回退 task:last 推导。
+            state_rows = conn.execute(
+                "SELECT k, v FROM kv WHERE k LIKE 'task:state:%'"
+            ).fetchall()
+            for row in state_rows:
+                key = str(row["k"]).removeprefix("task:state:")
+                ui_name = "discover" if key == "discovery" else key
+                if ui_name not in next_runs:
+                    continue
+                try:
+                    state = json.loads(str(row["v"]))
+                    due = float(state.get("due_at") or 0)
+                except (TypeError, ValueError, json.JSONDecodeError):
+                    continue
+                if due > 0:
+                    next_runs[ui_name] = int(due)
            for name, key, at in (
                ("fans", "fans", cfg.schedule.snapshot_at),
                ("snapshot", "snapshot", cfg.schedule.snapshot_at),
@@ -1990,6 +2006,14 @@ class DashboardQueries:
                ("discover", "discovery", None),
                ("snapshot_recent", "snapshot_recent", None),
            ):
+                # 已有落盘 due_at 则不再用 last 覆盖
+                state_raw = self._kv(conn, f"task:state:{key}")
+                if state_raw:
+                    try:
+                        if float(json.loads(state_raw).get("due_at") or 0) > 0:
+                            continue
+                    except (TypeError, ValueError, json.JSONDecodeError):
+                        pass
                record = tasks.get(key)
                if not (isinstance(record, dict) and record.get("finished_at")):
                    continue