定时任务（Scheduler）

8.1 调度器设计（v2）

• app/services/scheduler.py::HeapScheduler（TaskScheduler 别名）单例：最小堆 [(next_run_unix, task_path)] + asyncio.Event 唤醒，替代 v1 的 30s 轮询
• main.py startup 启动，shutdown 优雅停止
• 任何 next_run_at 变化（create/update/spawn/delete/run-now）调用 _wake_event.set()
• 每 600s 全量 rescan 兜底；并发执行受 SCHEDULER_MAX_CONCURRENT（默认 4）信号量限制
• 环境变量 DISABLE_SCHEDULER=1 可跳过启动（紧急救火）

8.2 任务存储（文件树）

• root_task_id = 路径第一段；scheduler_tree.task_path_for(task_id) 反查磁盘路径
• 运行步骤：{task_path}/runs/{run_id}.jsonl（append-only）
• v1 扁平 {task_id}.json lazy 迁移到树结构（scripts/migrate_tasks_to_tree.py）
• list_tasks(..., roots_only=True) 默认只返回 root（UI 侧栏）；堆 reload 仍用 list_all_tasks_flat（子任务须进堆）

8.2.1 Spawn 子任务与上下文

• spawn_child_task 工具 + create_child_task()；ContextVar _current_task_var 在 _execute_*_task 进出 set/reset（finally 必须 reset）
• 默认继承父任务 reply_to / permissions / capabilities / tz_offset_hours / model
• 限频：spawn_limits.check_chain_quota（默认每小时 30 次/链，SCHED_SPAWN_RATE_PER_HOUR）
• L3：descendants_summary 在子任务完成后向上祖先追加摘要（LRU 1500 字）

8.2.2 任务级模型

• task_config.model 可选 catalog id；空则 _get_default_model(user_id)（尊重用户 capability_defaults.llm）
• Admin：create_schedule_tool / manage_scheduled_tasks / spawn_child_task 均支持 model 参数与校验
• Service：_run_service_agent_task 将 config.model 作为 model_override 传入 create_consumer_agent

8.3 任务类型

8.4 调度类型

8.5 时区处理

• 每个任务存 tz_offset_hours 字段（创建时的用户时区偏移）
• cron 按用户时区解释：_next_cron UTC now → 用户本地 → croniter → 转回 UTC
• once 必须带时区后缀：无 tz 时按 tz_offset_hours 补充，工具层 _ensure_tz_suffix 兜底
• interval 不受时区影响：直接按秒数偏移
• 缺字段的旧任务用 _resolve_task_tz_offset(task) 回退 get_tz_offset(user_id)（与 preferences 默认 +8 一致）
• React Scheduler 创建/更新任务时 body 必须带 tz_offset_hours: getTzOffset()

8.6 reply_to 路由

Service 任务通过 reply_to 字段控制结果推送目标：

{
  "reply_to": {
    "channel": "wechat | inbox | admin_chat",
    "admin_id": "...",
    "service_id": "...",
    "conversation_id": "...",
    "session_id": "wechat_user_xxx"
  }
}

_run_service_agent_task 在 agent 执行循环中实时拦截 send_message 工具调用，通过 delivery.py 发送到 WeChat（支持文本+媒体）；_deliver_reply 仅在 agent 未使用 send_message 时作为兜底（纯文本摘要）。

8.7 运行记录步骤

每条 run record 含 steps[]：

8.8 任务结果持久化

_run_agent_loop 结束后调用 save_message / save_consumer_message 写入对话 JSON，含：

• source: "scheduled_task" 或 "admin_broadcast" 标记
• 完整 blocks[]

8.9 sync→async 主循环桥接

LangChain sync tool（如 contact_admin）通过 BaseTool._arun 在 run_in_executor 线程池执行，无 event loop。修复：

# inbox.py / scheduler.py
def set_main_loop(loop: asyncio.AbstractEventLoop):
    global _main_loop
    _main_loop = loop

# 调度时：
try:
    loop = asyncio.get_running_loop()
    loop.create_task(coro)
except RuntimeError:
    if _main_loop is not None and _main_loop.is_running():
        asyncio.run_coroutine_threadsafe(coro, _main_loop)

8.10 Service 任务专用工具

Consumer agent 通过：

• create_service_schedule_tool 注入 schedule_task
• create_service_manage_tasks_tool 注入 manage_scheduled_tasks（仅 "scheduler" 在 capabilities 中时）
• Service 的 manage_scheduled_tasks 仅能操作当前 conversation_id 的任务（权限隔离）

publish_service_task（Admin 工具）：

• service_ids 支持 ID 和名称匹配（大小写不敏感），未匹配时返回可用 Service 列表
• session_ids 可选参数，精确到单个微信会话
• run_now 调度使用 _schedule_coro 线程安全模式

8.11 v2 API 端点（谱系 / 配额）

Service 镜像路径：/api/scheduler/services/{service_id}/...