test(scheduler): 补 core/scheduler 服务层测试(审查 Top3 缺口)+ DB 测试门控收紧

- tests/test_scheduler.py:纯逻辑(cron 校验/next_run 墙钟时区折算/run message
  包装)+ DB 级(create 校验、claim 推进 next_run 不重复认领、disabled/软删/
  过期不认领、连续失败自动停用、skipped 不动计数)。认领类用例统一用 2000 年
  纪元时间,claim_due_jobs(now=纪元) 只可能命中测试 job,永不触碰真实 job
- **DB 测试门控收紧(test_scheduler + test_usage_report)**:只认显式
  ZCBOT_TEST_DB_URL,绝不回退 .env 的 ZCBOT_DB_URL —— 实锤教训:.env 的 URL
  经隧道指向生产库,测试插入的到点 job 被生产 green 实例调度守护认领并真跑了
  agent(4 次一句话迷你调用,费用几厘,产物已按测试专属 user 全量清理,零残留)
- RUN.md:环境段补 ZCBOT_TEST_DB_URL 说明,故障兜底表加一行

301 测试全过(未设测试库时 DB 组自动 skip)。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

RUN.md

@@ -144,6 +144,7 @@
   - **未绑定成员发消息 → 回绑定指引**(不再静默):聊天优先布局下新员工第一动作就是打字,回调对未绑定成员的 text/图片/文件消息每条回一句"先去控制台绑定"(事件不回)。未绑定成员点菜单「工作台」则落在绑定提示页(不自动建号)。
 - **channel 长会话上下文(微信/企业微信通用,0019)**:常驻会话不再无限膨胀。① **自动分段**——入站时距上次消息超过 `config.json` 的 `channel.session_gap_hours`(默 **6** 小时,设 `<=0` 关闭)→ 软重置:只把「最后一条 user 消息起」喂模型(保留上一轮做续聊锚点),之前的历史仍全留 DB,网页端照旧翻完整记录;② **手动新话题**——用户在微信/企业微信里直接发「新话题 / 新会话 / `/new` / 清空上下文」→ 硬重置,彻底从零(回执提示已归档)。两者都**不删任何消息**,只移动「喂给模型的窗口起点」`tasks.context_base_idx`。网页端「清空对话」(`POST /v1/tasks/{id}/clear`)仍整清并把 base 归 0。需 `main.py db upgrade head` 带上 `0019`。
 - **PG**:`ZCBOT_DB_URL` 必填。本地 docker compose / 远端 dev / 生产任选;未设置时启动清晰报错,不引导 docker(§7.4)。
+- **测试库(可选,`ZCBOT_TEST_DB_URL`)**:DB 级单测(`tests/test_usage_report.py` / `tests/test_scheduler.py`)**只认这个显式变量、绝不回退 `.env` 的 `ZCBOT_DB_URL`**——后者可能经隧道指向生产库,测试插入的到点 job 会被生产实例调度守护真跑一次(2026-07-23 实锤)。未设则这两组自动 skip;要跑就建个专用库(`createdb zcbot_test` + `env ZCBOT_DB_URL=<测试库> main.py db upgrade head`)后 `set ZCBOT_TEST_DB_URL=...`。
 - **Auth env**:`PLATFORM_KEY` + `JWT_SECRET` 任一缺失 web 启动 fail-fast。生成随机串:`python -c "import secrets; print(secrets.token_urlsafe(48))"`。
 - **用户管理**(`users.email/password_hash/role`,0005 UNIQUE(email)、0009 role):dev SPA 登录后端。发用户两条路径任选:CLI `main.py user add`(下方),或在登录页右下角"+ 管理员添加用户"链接(需先设 `ZCBOT_ADMIN_TOKEN` env,弹窗输入 email/密码/管理员口令/角色)。撤用户 `DELETE FROM users WHERE email=...`(先 DELETE 该 user 的 tasks)。**用户自助改密**:登录后顶栏「改密码」按钮(走 `POST /v1/auth/change_password`,需知道旧密码);改邮箱 / 用户忘了旧密码无法自助 → 手动 SQL(见故障兜底)。
 - **角色与管理后台**(`users.role` ∈ `user`/`admin`):admin 才显顶栏"管理"入口 → `/static/admin.html`(非 admin 403)。页面:左侧目录(点击滚到对应区)+ 运行态/任务/用户用量/按模型/各用户用量/存储;「按模型」「各用户用量」支持时间筛选(全部/近7天/近30天)+ 排序(按成本/按用量),「各用户用量」「存储」分页;顶栏「导出 PDF」走浏览器打印(在打印对话框选"另存为 PDF",列表取前 10)。提管理员 `main.py user role --email X --role admin`(改完即时生效,role 走 DB 查不进 JWT)。`ZCBOT_ADMIN_TOKEN` 是另一回事(发用户共享口令),与 role 互不相干。
@@ -840,6 +841,7 @@ sudo xfs_quota -x -c "limit -p bhard=10g zcbot_<user_uuid>" /opt
 | 现象 | 原因 / 处理 |
 |---|---|
 | `ZCBOT_DB_URL is not set` | `.env` 没写 / litellm 链路没触发。直跑脚本时 `import litellm` 或 `export ZCBOT_DB_URL=...` |
+| DB 级单测把行写进了生产库 | 测试只认 `ZCBOT_TEST_DB_URL`(见「环境」段);别把它指向 `.env` 里的隧道 URL。已发生的:按测试专属 email(`test-*@invalid.local`)过滤清理 usage_events/tasks/scheduled_jobs/users |
 | `ModuleNotFoundError: litellm` | 用了全局 `python`,改 `.venv/Scripts/python.exe ...` |
 | Windows 控制台 emoji 崩 | Python stdout 是 GBK。用 `[OK]` / `[ng]` 等 ASCII 标签(见 memory) |
 | research 报 `paper_server auth failed (HTTP 400, not_authenticated)` | paper_server GET 接口已要求 API Key。`.env` 配 `PAPER_SERVER_API_KEY=`(paper_server Django admin 里建 ApiKey);docker 模式会自动透传进容器,改完重启 web 进程生效 |

tests/test_scheduler.py

@@ -0,0 +1,183 @@
+"""core/scheduler.py 服务层测试(DESIGN §8.5)—— 补审查 Top3 缺口(此前零测试)。
+
+覆盖两层:
+- 纯逻辑(无 DB):cron 校验、next_run 时区折算(墙钟语义)、run message 包装
+- DB 级(仅当显式设 ZCBOT_TEST_DB_URL 才跑;只插/删测试专属 user 的行):
+  create/claim 推进 next_run(不重复认领)、disabled/软删/过期不认领、
+  record_result 连续失败自动停用
+
+端到端(守护循环认领 → 真跑 agent → LLM 回复)仍走 scripts/smoke_scheduler.py
+(需起服务 + 真实调用,不进单测)。
+"""
+from __future__ import annotations
+
+import os
+import unittest
+import uuid
+from datetime import datetime, timedelta, timezone
+
+
+def _test_db_ready() -> bool:
+    """DB 级测试只认显式 `ZCBOT_TEST_DB_URL`,绝不回退 .env 的 ZCBOT_DB_URL。
+
+    教训(2026-07-23 实锤):.env 的 ZCBOT_DB_URL 可能经隧道指向**生产库**(本次
+    即 127.0.0.1:6012 → 生产 PG),测试若沿用它,插入的 enabled 到点 job 会被生产
+    实例的调度守护认领并真跑 agent。测试库须显式指定:
+        set ZCBOT_TEST_DB_URL=postgresql+psycopg://user:pass@127.0.0.1:5432/zcbot_test
+    (建库后 `alembic upgrade head` 一次);未设 → 整组 skip。
+    """
+    url = os.environ.get("ZCBOT_TEST_DB_URL", "").strip()
+    if not url:
+        return False
+    os.environ["ZCBOT_DB_URL"] = url  # 本测试进程内覆盖,engine 单例随之指向测试库
+    return True
+
+
+from core import scheduler  # noqa: E402  (纯逻辑部分不碰 DB,导入安全)
+
+try:
+    if not _test_db_ready():
+        raise RuntimeError("ZCBOT_TEST_DB_URL 未设")
+    from core.storage import session_scope
+    from core.storage.models import ScheduledJob, User
+
+    with session_scope() as _s:
+        _s.execute(__import__("sqlalchemy").select(1))
+    _DB_OK = True
+except Exception:
+    _DB_OK = False
+
+
+class SchedulerPureTests(unittest.TestCase):
+    def test_validate_cron(self):
+        scheduler.validate_cron("0 8 * * *")
+        scheduler.validate_cron("*/5 * * * *")
+        for bad in ("", "not a cron", "99 99 * * *", "* * * *"):
+            with self.assertRaises(ValueError):
+                scheduler.validate_cron(bad)
+
+    def test_compute_next_run_wall_clock_tz(self):
+        """'0 8 * * *' 是 job 时区的早 8 点,不是 UTC 8 点(§8.5 时区坑)。"""
+        # 2026-01-05 00:00 UTC = 上海 08:00 —— 下一个上海 08:00 是 1 月 6 日
+        after = datetime(2026, 1, 5, 0, 0, tzinfo=timezone.utc)
+        nxt = scheduler.compute_next_run("0 8 * * *", "Asia/Shanghai", after=after)
+        self.assertEqual(nxt.tzinfo, timezone.utc)
+        self.assertEqual((nxt.year, nxt.month, nxt.day, nxt.hour), (2026, 1, 6, 0))
+        # 同一时刻按 UTC 时区算,则当天 08:00 UTC
+        nxt_utc = scheduler.compute_next_run("0 8 * * *", "UTC", after=after)
+        self.assertEqual((nxt_utc.day, nxt_utc.hour), (5, 8))
+
+    def test_build_run_message_wraps_prompt(self):
+        msg = scheduler.build_run_message(
+            {"name": "早安简报", "prompt": "回一句早安", "tz": "Asia/Shanghai"}
+        )
+        self.assertIn("[定时任务「早安简报」自动触发", msg)
+        self.assertIn("回一句早安", msg)
+
+
+@unittest.skipUnless(_DB_OK, "ZCBOT_TEST_DB_URL 未设或测试库不可达,跳过 DB 级测试(绝不用 .env 的库)")
+class SchedulerDbTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.uid = uuid.uuid4()
+        with session_scope() as s:
+            s.add(User(user_id=cls.uid, email=f"test-scheduler-{cls.uid.hex[:8]}@invalid.local"))
+
+    @classmethod
+    def tearDownClass(cls):
+        from sqlalchemy import delete
+        with session_scope() as s:
+            s.execute(delete(ScheduledJob).where(ScheduledJob.user_id == cls.uid))
+            s.execute(delete(User).where(User.user_id == cls.uid))
+
+    def _mk_job(self, **kw) -> uuid.UUID:
+        d = scheduler.create_job(
+            self.uid, name=kw.pop("name", "t-job"), prompt=kw.pop("prompt", "p"),
+            cron=kw.pop("cron", "0 8 * * *"), **kw,
+        )
+        return uuid.UUID(d["job_id"])
+
+    def _set(self, jid: uuid.UUID, **values) -> None:
+        from sqlalchemy import update
+        with session_scope() as s:
+            s.execute(update(ScheduledJob).where(ScheduledJob.job_id == jid).values(**values))
+
+    def _row(self, jid: uuid.UUID) -> ScheduledJob:
+        with session_scope() as s:
+            return s.get(ScheduledJob, jid)
+
+    # 认领一律用过去纪元:claim_due_jobs 是全库扫描,若用真实 now 会把真实用户
+    # 到点的 job 一并认领(推进 next_run_at 但不执行 = 吃掉一次真实触发)。
+    # 测试 job 的 next_run_at 都摆在 2000 年,真实 job(next_run 恒近未来)永不命中。
+    EPOCH_DUE = datetime(2000, 1, 1, tzinfo=timezone.utc)
+    EPOCH_NOW = datetime(2000, 1, 2, tzinfo=timezone.utc)
+
+    def _claim_ids(self) -> set:
+        return {snap["job_id"] for snap in scheduler.claim_due_jobs(now=self.EPOCH_NOW)}
+
+    def test_create_sets_next_run_and_validates(self):
+        jid = self._mk_job()
+        row = self._row(jid)
+        self.assertIsNotNone(row.next_run_at)
+        self.assertTrue(row.enabled)
+        with self.assertRaises(scheduler.JobError):
+            scheduler.create_job(self.uid, name="", prompt="p", cron="0 8 * * *")
+        with self.assertRaises(ValueError):
+            scheduler.create_job(self.uid, name="x", prompt="p", cron="bad cron")
+
+    def test_claim_advances_next_run_no_double_claim(self):
+        jid = self._mk_job(name="due-job")
+        self._set(jid, next_run_at=self.EPOCH_DUE)
+        self.assertIn(jid, self._claim_ids())        # 到点 → 认领
+        row = self._row(jid)
+        self.assertGreater(row.next_run_at, self.EPOCH_NOW)  # 已按 cron 推进到 now 之后
+        self.assertNotIn(jid, self._claim_ids())     # 同一到点不重复认领
+
+    def test_disabled_and_deleted_not_claimed(self):
+        jid = self._mk_job(name="off-job")
+        self._set(jid, next_run_at=self.EPOCH_DUE)
+        scheduler.set_enabled(self.uid, str(jid), False)
+        self.assertNotIn(jid, self._claim_ids())
+
+        jid2 = self._mk_job(name="del-job")
+        self._set(jid2, next_run_at=self.EPOCH_DUE)
+        scheduler.cancel_job(self.uid, str(jid2))
+        self.assertNotIn(jid2, self._claim_ids())
+        self.assertIsNotNone(self._row(jid2).deleted_at)  # 软删,不物理删
+
+    def test_expired_job_auto_disabled(self):
+        jid = self._mk_job(name="expired-job")
+        self._set(jid, next_run_at=self.EPOCH_DUE,
+                  expires_at=self.EPOCH_DUE + timedelta(hours=1))
+        self.assertNotIn(jid, self._claim_ids())     # 过期不返回
+        row = self._row(jid)
+        self.assertFalse(row.enabled)
+        self.assertEqual(row.last_status, "expired")
+
+    def test_record_result_failure_streak_auto_disable(self):
+        jid = self._mk_job(name="fail-job")
+        for i in range(scheduler.FAILURE_DISABLE_THRESHOLD - 1):
+            scheduler.record_result(jid, status="error", task_id=None, error=f"boom {i}")
+        row = self._row(jid)
+        self.assertTrue(row.enabled)                 # 还差一次,不停
+        # 一次 ok 清零计数
+        scheduler.record_result(jid, status="ok", task_id=None)
+        self.assertEqual(self._row(jid).consecutive_failures, 0)
+        # 连续失败到阈值 → 自动停用,last_error 带说明
+        for i in range(scheduler.FAILURE_DISABLE_THRESHOLD):
+            scheduler.record_result(jid, status="error", task_id=None, error="boom")
+        row = self._row(jid)
+        self.assertFalse(row.enabled)
+        self.assertIn("已自动停用", row.last_error or "")
+
+    def test_skipped_does_not_touch_counters(self):
+        jid = self._mk_job(name="skip-job")
+        scheduler.record_result(jid, status="skipped", task_id=None, error="busy")
+        row = self._row(jid)
+        self.assertEqual(row.run_count or 0, 0)
+        self.assertEqual(row.consecutive_failures or 0, 0)
+        self.assertEqual(row.last_status, "skipped")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_usage_report.py

@@ -16,24 +16,25 @@ from datetime import datetime, timedelta, timezone
 from decimal import Decimal
 
 
-def _ensure_db_url() -> None:
+def _test_db_ready() -> bool:
-    """ZCBOT_DB_URL 平时靠 `import litellm` 的隐式 dotenv 加载进 env(engine.py 自己
+    """DB 级测试只认显式 `ZCBOT_TEST_DB_URL`,绝不回退 .env 的 ZCBOT_DB_URL。
-    不读 .env)。测试不背 litellm 这个 8s 重依赖,显式从仓库根 .env 抠这一个 key。"""
+
-    if os.environ.get("ZCBOT_DB_URL", "").strip():
+    教训(2026-07-23 实锤):.env 的 ZCBOT_DB_URL 可能经隧道指向**生产库**(本次
-        return
+    即 127.0.0.1:6012 → 生产 PG),测试若沿用它,插入的 enabled 到点 job 会被生产
-    from core.paths import ROOT
+    实例的调度守护认领并真跑 agent。测试库须显式指定:
-    envf = ROOT / ".env"
+        set ZCBOT_TEST_DB_URL=postgresql+psycopg://user:pass@127.0.0.1:5432/zcbot_test
-    if not envf.is_file():
+    (建库后 `alembic upgrade head` 一次);未设 → 整组 skip。
-        return
+    """
-    for line in envf.read_text(encoding="utf-8").splitlines():
+    url = os.environ.get("ZCBOT_TEST_DB_URL", "").strip()
-        line = line.strip()
+    if not url:
-        if line.startswith("ZCBOT_DB_URL="):
+        return False
-            os.environ["ZCBOT_DB_URL"] = line.split("=", 1)[1].strip().strip("'\"")
+    os.environ["ZCBOT_DB_URL"] = url  # 本测试进程内覆盖,engine 单例随之指向测试库
-            return
+    return True
 
 
 try:
-    _ensure_db_url()
+    if not _test_db_ready():
+        raise RuntimeError("ZCBOT_TEST_DB_URL 未设")
     from core.storage import session_scope
     from core.storage.models import Task, UsageEvent, User
     from core.storage import usage_report
@@ -45,7 +46,7 @@ except Exception:
     _DB_OK = False
 
 
-@unittest.skipUnless(_DB_OK, "PG 不可达(ZCBOT_DB_URL 未配或库没起),跳过 DB 级测试")
+@unittest.skipUnless(_DB_OK, "ZCBOT_TEST_DB_URL 未设或测试库不可达,跳过 DB 级测试(绝不用 .env 的库)")
 class UsageReportTests(unittest.TestCase):
     """一个测试专属 user + 两个 task,插一组已知 usage_events,验证三条读路径。"""