fix(scheduler): drop cross-task mutex, quiet window + retry

Incident: evening full snapshot could be recorded as ok with
already_completed/elapsed=0 after mutual skip with hourly sync, then
next_run jumped to the next day.

- Full-site snapshot never skips because another task is running
- 23:00-01:00 Asia/Shanghai: defer discovery / snapshot_recent /
  candidate_cleanup to after 01:00; fans never defers
- Skip records ok=false + skipped=true (not success / not completed)
- Skip/fail reschedule soon (~5m); daily success still next snapshot_at
- TaskControl return value drives next_run; system UI shows skip fields

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,34 @@ 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
+
+
 def main():
     test_run_when_due_and_reschedule()
     test_next_run_fn_controls_reschedule()
+    test_task_control_overrides_ok_and_next_run()
     test_task_error_does_not_block_other_tasks()
     test_run_forever_with_stop_event()
     test_max_iterations_and_validation()

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,83 @@ 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()
 
@@ -882,6 +962,8 @@ 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_overview_adjacent_previous_with_seed_and_dual_fields()
     test_same_day_growth_baselines()
     test_growth_rate_sort_uses_prior()
@@ -948,13 +1030,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 +1050,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()