Scheduler (Scheduled Tasks)

8.1 Scheduler Design (v2)

• app/services/scheduler.py::HeapScheduler (TaskScheduler alias): min-heap [(next_run_unix, task_path)] + asyncio.Event wake — replaces v1 30s polling
• Started/stopped with main.py lifecycle
• Any next_run_at change calls _wake_event.set()
• 600s full rescan fallback; concurrency capped by SCHEDULER_MAX_CONCURRENT (default 4)
• DISABLE_SCHEDULER=1 skips startup (emergency)

8.2 Task Storage (file tree)

• root_task_id = first path segment; scheduler_tree.task_path_for(task_id) resolves disk path
• Run steps: {task_path}/runs/{run_id}.jsonl (append-only)
• v1 flat {task_id}.json lazy-migrates to tree (scripts/migrate_tasks_to_tree.py)
• list_tasks(..., roots_only=True) returns roots only (UI sidebar); heap reload uses list_all_tasks_flat

8.2.1 Spawn children & context

• spawn_child_task + create_child_task(); ContextVar _current_task_var set/reset in _execute_*_task (must reset in finally)
• Inherits parent reply_to / permissions / capabilities / tz_offset_hours / model by default
• Rate limit: spawn_limits.check_chain_quota (default 30/hour/chain, SCHED_SPAWN_RATE_PER_HOUR)
• L3: descendants_summary appended to ancestors after child finishes (LRU 1500 chars)

8.2.2 Per-task model

• Optional task_config.model (catalog id); empty → _get_default_model(user_id)
• Admin: create_schedule_tool / manage_scheduled_tasks / spawn_child_task support model
• Service: _run_service_agent_task passes config.model as model_override to create_consumer_agent

8.3 Task Types

8.4 Schedule Types

8.5 Timezone Handling

• Each task stores tz_offset_hours field (user's timezone offset at creation time)
• Cron interpreted in user's timezone: _next_cron UTC now → user local → croniter → back to UTC
• once must include timezone suffix: _ensure_tz_suffix provides fallback with tz_offset_hours
• interval not affected by timezone: directly uses second offsets
• Old tasks without the field use _resolve_task_tz_offset(task) falling back to get_tz_offset(user_id) (consistent with preferences default +8)
• React Scheduler must include tz_offset_hours: getTzOffset() in create/update task body

8.6 reply_to Routing

Service tasks control result delivery target via reply_to field:

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

_run_service_agent_task in real-time intercepts send_message tool calls during agent execution loop, sends to WeChat via delivery.py (supports text + media); _deliver_reply only as fallback when agent doesn't use send_message (plain text summary).

8.7 Run Record Steps

Each run record contains steps[]:

8.8 Task Result Persistence

After _run_agent_loop finishes, calls save_message / save_consumer_message to write conversation JSON, including:

• source: "scheduled_task" or "admin_broadcast" marker
• Complete blocks[]

8.9 Sync→Async Main Loop Bridge

LangChain sync tools (like contact_admin) execute via BaseTool._arun in run_in_executor thread pool, which has no event loop. Fix:

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

# When scheduling:
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 Task Tools

Consumer agent via:

• create_service_schedule_tool injects schedule_task
• create_service_manage_tasks_tool injects manage_scheduled_tasks (only when "scheduler" in capabilities)
• Service's manage_scheduled_tasks can only operate tasks for the current conversation_id (permission isolation)

publish_service_task (Admin tool):

• service_ids supports ID and name matching (case-insensitive), returns available Service list when no match
• session_ids optional parameter to target specific WeChat sessions
• run_now scheduling uses _schedule_coro thread-safe mode

8.11 v2 API (tree / quota)

Service mirror: /api/scheduler/services/{service_id}/...