Main Lane (user input):        User Input --> lane_lock.acquire() -------> LLM --> Print                       (blocking: always wins)    Heartbeat Lane (background thread, 1s poll):        should_run()?            |no --> sleep 1s            |yes        _execute():            lane_lock.acquire(blocking=False)                |fail --> yield (user has priority)                |success            build prompt from HEARTBEAT.md + SOUL.md + MEMORY.md                |            run_agent_single_turn()                |            parse: "HEARTBEAT_OK"? --> suppress                   meaningful text? --> duplicate? --> suppress                                           |no                                       output_queue.append()    Cron Service (background thread, 1s tick):        CRON.json --> load jobs --> tick() every 1s            |        for each job: enabled? --> due? --> _run_job()            |        error? --> consecutive_errors++ --> >=5? --> auto-disable            |ok        consecutive_errors = 0 --> log to cron-runs.jsonl

@dataclassclass CronJob:    id: str    name: str    enabled: bool    schedule_kind: str       # "at" | "every" | "cron"    schedule_config: dict    payload: dict            # {"kind": "agent_turn", "message": "..."}    consecutive_errors: int = 0def _compute_next(self, job, now):    if job.schedule_kind == "at":        ts = datetime.fromisoformat(cfg.get("at", "")).timestamp()        return ts if ts > now else 0.0    if job.schedule_kind == "every":        every = cfg.get("every_seconds", 3600)        # 对齐到锚点, 保证触发时间可预测        steps = int((now - anchor) / every) + 1        return anchor + steps * every    if job.schedule_kind == "cron":        return croniter(expr, datetime.fromtimestamp(now)).get_next(datetime).timestamp()

if status == "error":    job.consecutive_errors += 1    if job.consecutive_errors >= 5:        job.enabled = Falseelse:    job.consecutive_errors = 0

---name: crondescription: Schedule reminders and recurring tasks.---# CronUse the `cron` tool to schedule reminders or recurring tasks.## Three Modes1. **Reminder** - message is sent directly to user2. **Task** - message is a task description, agent executes and sends result3. **One-time** - runs once at a specific time, then auto-deletes## ExamplesFixed reminder:```cron(action="add", message="Time to take a break!", every_seconds=1200)```Dynamic task (agent executes each time):```cron(action="add", message="Check HKUDS/nanobot GitHub stars and report", every_seconds=600)```One-time scheduled task (compute ISO datetime from current time):```cron(action="add", message="Remind me about the meeting", at="<ISO datetime>")```Timezone-aware cron:```cron(action="add", message="Morning standup", cron_expr="0 9 * * 1-5", tz="America/Vancouver")```List/remove:```cron(action="list")cron(action="remove", job_id="abc123")```## Time Expressions| User says | Parameters ||-----------|------------|| every 20 minutes | every_seconds: 1200 || every hour | every_seconds: 3600 || every day at 8am | cron_expr: "0 8 * * *" || weekdays at 5pm | cron_expr: "0 17 * * 1-5" || 9am Vancouver time daily | cron_expr: "0 9 * * *", tz: "America/Vancouver" || at a specific time | at: ISO datetime string (compute from current time) |## TimezoneUse `tz` with `cron_expr` to schedule in a specific IANA timezone. Without `tz`, the server's local timezone is used.

   CronService._on_timer() → _execute_job(job) → on_job(job)                                                     ↓                                               AgentLoop.process_direct()                                                     ↓                                                返回结果

ounter(line  _execute_job() → 更新 job.state → _save_store() → JSON

CronTool.execute() → 执行cron工具    ↓根据action参数判断操作类型：├── add:[CronService:add_job()]├── list:[CronService:list_jobs()]├── remove:[CronService:remove_job()]├── run:[CronService:run_job()]├── enable/disable:[CronService:enable_job()]    ↓CronService._save_store() → 保存任务到磁盘    ↓返回操作结果给LLM

@dataclassclass CronSchedule:    """Schedule definition for a cron job."""    kind: Literal["at", "every", "cron"]    # For "at": timestamp in ms    at_ms: int | None = None    # For "every": interval in ms    every_ms: int | None = None    # For "cron": cron expression (e.g. "0 9 * * *")    expr: str | None = None    # Timezone for cron expressions    tz: str | None = None@dataclassclass CronJob:    """A scheduled job."""    id: str    name: str    enabled: bool = True    schedule: CronSchedule = field(default_factory=lambda: CronSchedule(kind="every"))    payload: CronPayload = field(default_factory=CronPayload)    state: CronJobState = field(default_factory=CronJobState)    created_at_ms: int = 0    updated_at_ms: int = 0    delete_after_run: bool = False# ===================== 核心类定义 =====================class CronTool(Tool):    """Tool to schedule reminders and recurring tasks."""    def __init__(self, cron_service: CronService):        # 关联底层定时任务服务实例（核心依赖）        self._cron = cron_service        # 初始化会话渠道（用于任务消息投递）        self._channel = ""        # 初始化会话ID（用于任务消息精准投递到指定聊天）        self._chat_id = ""    def set_context(self, channel: str, chat_id: str) -> None:        """Set the current session context for delivery."""        # 设置当前会话的渠道信息（如钉钉/微信渠道）        self._channel = channel        # 设置当前会话的聊天ID（如群ID/用户ID）        self._chat_id = chat_id    @property    def name(self) -> str:        # 工具名称（Agent调用时的标识）        return "cron"    @property    def description(self) -> str:        # 工具描述（告知Agent该工具的作用和支持的操作）        return "Schedule reminders and recurring tasks. Actions: add, list, remove."    @property    def parameters(self) -> dict[str, Any]:        # 工具参数定义（JSON Schema格式，供Agent生成工具调用参数）        return {            "type": "object",  # 参数整体为对象类型            "properties": {                # 核心参数：操作类型（add/list/remove）                "action": {                    "type": "string",                    "enum": ["add", "list", "remove"],  # 仅支持这三种操作                    "description": "Action to perform"                },                # 任务消息（仅add操作需要）                "message": {                    "type": "string",                    "description": "Reminder message (for add)"                },                # 周期执行间隔（秒，仅add操作的every类型需要）                "every_seconds": {                    "type": "integer",                    "description": "Interval in seconds (for recurring tasks)"                },                # Cron表达式（仅add操作的cron类型需要）                "cron_expr": {                    "type": "string",                    "description": "Cron expression like '0 9 * * *' (for scheduled tasks)"                },                # 时区（仅add操作的cron类型需要）                "tz": {                    "type": "string",                    "description": "IANA timezone for cron expressions (e.g. 'America/Vancouver')"                },                # 一次性执行时间（ISO格式，仅add操作的at类型需要）                "at": {                    "type": "string",                    "description": "ISO datetime for one-time execution (e.g. '2026-02-12T10:30:00')"                },                # 任务ID（仅remove操作需要）                "job_id": {                    "type": "string",                    "description": "Job ID (for remove)"                }            },            "required": ["action"]  # 必选参数：操作类型        }    async def execute(        self,        action: str,        message: str = "",        every_seconds: int | None = None,        cron_expr: str | None = None,        tz: str | None = None,        at: str | None = None,        job_id: str | None = None,        **kwargs: Any    ) -> str:        # 根据操作类型分发到对应处理方法        if action == "add":            return self._add_job(message, every_seconds, cron_expr, tz, at)        elif action == "list":            return self._list_jobs()        elif action == "remove":            return self._remove_job(job_id)        # 未知操作时返回提示        return f"Unknown action: {action}"    def _add_job(        self,        message: str,        every_seconds: int | None,        cron_expr: str | None,        tz: str | None,        at: str | None,    ) -> str:        # 校验1：添加任务必须提供消息内容        if not message:            return "Error: message is required for add"        # 校验2：必须先设置会话上下文（渠道/聊天ID），否则无法投递消息        if not self._channel or not self._chat_id:            return "Error: no session context (channel/chat_id)"        # 校验3：时区参数仅能与Cron表达式配合使用        if tz and not cron_expr:            return "Error: tz can only be used with cron_expr"        # 校验4：时区参数合法性校验        if tz:            from zoneinfo import ZoneInfo  # 按需导入时区模块（减少初始化开销）            try:                # 验证时区是否为有效IANA时区                ZoneInfo(tz)            except (KeyError, Exception):                return f"Error: unknown timezone '{tz}'"        # 构建调度规则（三种类型互斥）        delete_after = False  # 标记是否执行后删除（仅at类型为True）        # 类型1：周期性任务（every_seconds）        if every_seconds:            schedule = CronSchedule(kind="every", every_ms=every_seconds * 1000)        # 类型2：Cron表达式任务（cron_expr）        elif cron_expr:            schedule = CronSchedule(kind="cron", expr=cron_expr, tz=tz)        # 类型3：一次性定时任务（at）        elif at:            from datetime import datetime  # 按需导入时间模块            # 将ISO格式时间转换为datetime对象            dt = datetime.fromisoformat(at)            # 转换为毫秒级时间戳            at_ms = int(dt.timestamp() * 1000)            schedule = CronSchedule(kind="at", at_ms=at_ms)            delete_after = True  # 一次性任务执行后删除        # 无有效调度规则时返回错误        else:            return "Error: either every_seconds, cron_expr, or at is required"        # 调用底层CronService添加任务        job = self._cron.add_job(            name=message[:30],  # 任务名称截取前30字符（避免过长）            schedule=schedule,  # 调度规则            message=message,    # 任务消息内容            deliver=True,       # 标记需要投递消息到指定渠道            channel=self._channel,  # 投递渠道            to=self._chat_id,       # 投递目标            delete_after_run=delete_after,  # 执行后是否删除        )        # 返回创建成功提示（包含任务名称和ID）        return f"Created job '{job.name}' (id: {job.id})"    def _list_jobs(self) -> str:        # 调用底层CronService获取所有启用的任务        jobs = self._cron.list_jobs()        # 无任务时返回提示        if not jobs:            return "No scheduled jobs."        # 格式化任务列表（名称+ID+调度类型）        lines = [f"- {j.name} (id: {j.id}, {j.schedule.kind})" for j in jobs]        # 返回用户友好的任务列表        return "Scheduled jobs:\n" + "\n".join(lines)    def _remove_job(self, job_id: str | None) -> str:        # 校验：删除任务必须提供任务ID        if not job_id:            return "Error: job_id is required for remove"        # 调用底层CronService删除任务        if self._cron.remove_job(job_id):            return f"Removed job {job_id}"        # 任务ID不存在时返回提示        return f"Job {job_id} not found"

CronService._compute_next_run()├── "at": 直接返回指定时间戳（如果未过期）├── "every": 当前时间 + 间隔时间└── "cron": 使用croniter库解析表达式计算下次执行时间_

CronService._arm_timer()├── 计算下一个唤醒时间├── 创建异步任务等待唤醒└── 定时器到期时触发_on_timer()

CronService._on_timer()├── 检查哪些任务已到期├── 对每个到期任务执行_execute_job()└── 重新安排下一次唤醒

class CronService:    """Service for managing and executing scheduled jobs."""    def __init__(        self,        store_path: Path,        on_job: Callable[[CronJob], Coroutine[Any, Any, str | None]] | None = None    ):        # 定时任务存储文件路径（JSON格式）        self.store_path = store_path        # 任务执行回调函数：外部传入，负责实际执行任务逻辑，返回响应文本        self.on_job = on_job        # 定时任务存储实例（懒加载，首次使用时加载）        self._store: CronStore | None = None        # 定时器任务（asyncio.Task）：用于调度下次任务执行        self._timer_task: asyncio.Task | None = None        # 服务运行状态标记：True=运行中，False=已停止        self._running = False    def _load_store(self) -> CronStore:        """Load jobs from disk."""        # 若已加载存储实例，直接返回（避免重复加载）        if self._store:            return self._store        # 存储文件存在时，从文件加载任务        if self.store_path.exists():            try:                # 读取文件内容并解析为JSON                data = json.loads(self.store_path.read_text(encoding="utf-8"))                jobs = []                # 遍历JSON中的任务列表，转换为CronJob对象                for j in data.get("jobs", []):                    jobs.append(CronJob(                        id=j["id"],  # 任务ID                        name=j["name"],  # 任务名称                        enabled=j.get("enabled", True),  # 是否启用（默认True）                        # 解析调度规则                        schedule=CronSchedule(                            kind=j["schedule"]["kind"],                            at_ms=j["schedule"].get("atMs"),                            every_ms=j["schedule"].get("everyMs"),                            expr=j["schedule"].get("expr"),                            tz=j["schedule"].get("tz"),                        ),                        # 解析任务负载                        payload=CronPayload(                            kind=j["payload"].get("kind", "agent_turn"),                            message=j["payload"].get("message", ""),                            deliver=j["payload"].get("deliver", False),                            channel=j["payload"].get("channel"),                            to=j["payload"].get("to"),                        ),                        # 解析任务状态                        state=CronJobState(                            next_run_at_ms=j.get("state", {}).get("nextRunAtMs"),                            last_run_at_ms=j.get("state", {}).get("lastRunAtMs"),                            last_status=j.get("state", {}).get("lastStatus"),                            last_error=j.get("state", {}).get("lastError"),                        ),                        created_at_ms=j.get("createdAtMs", 0),  # 创建时间戳                        updated_at_ms=j.get("updatedAtMs", 0),  # 更新时间戳                        delete_after_run=j.get("deleteAfterRun", False),  # 执行后删除                    ))                # 初始化存储实例                self._store = CronStore(jobs=jobs)            except Exception as e:                # 加载失败时记录警告，初始化空存储                logger.warning("Failed to load cron store: {}", e)                self._store = CronStore()        else:            # 存储文件不存在时，初始化空存储            self._store = CronStore()        return self._store    def _save_store(self) -> None:        """Save jobs to disk."""        # 存储实例未初始化时直接返回        if not self._store:            return        # 确保存储文件所在目录存在（不存在则创建）        self.store_path.parent.mkdir(parents=True, exist_ok=True)        # 将存储实例序列化为JSON格式数据        data = {            "version": self._store.version,  # 存储版本            "jobs": [                {                    "id": j.id,                    "name": j.name,                    "enabled": j.enabled,                    # 序列化调度规则                    "schedule": {                        "kind": j.schedule.kind,                        "atMs": j.schedule.at_ms,                        "everyMs": j.schedule.every_ms,                        "expr": j.schedule.expr,                        "tz": j.schedule.tz,                    },                    # 序列化任务负载                    "payload": {                        "kind": j.payload.kind,                        "message": j.payload.message,                        "deliver": j.payload.deliver,                        "channel": j.payload.channel,                        "to": j.payload.to,                    },                    # 序列化任务状态                    "state": {                        "nextRunAtMs": j.state.next_run_at_ms,                        "lastRunAtMs": j.state.last_run_at_ms,                        "lastStatus": j.state.last_status,                        "lastError": j.state.last_error,                    },                    "createdAtMs": j.created_at_ms,                    "updatedAtMs": j.updated_at_ms,                    "deleteAfterRun": j.delete_after_run,                }                for j in self._store.jobs  # 遍历所有任务            ]        }        # 将JSON数据写入文件（缩进2格，确保非ASCII字符不转义）        self.store_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")    async def start(self) -> None:        """Start the cron service."""        # 标记服务为运行状态        self._running = True        # 加载任务存储（从文件读取）        self._load_store()        # 重新计算所有启用任务的下次执行时间        self._recompute_next_runs()        # 保存更新后的任务状态到文件        self._save_store()        # 设置定时器（触发下次任务执行）        self._arm_timer()        # 记录启动日志：当前任务数量        logger.info("Cron service started with {} jobs", len(self._store.jobs if self._store else []))    def stop(self) -> None:        """Stop the cron service."""        # 标记服务为停止状态        self._running = False        # 取消当前定时器任务（避免继续触发）        if self._timer_task:            self._timer_task.cancel()            self._timer_task = None    def _recompute_next_runs(self) -> None:        """Recompute next run times for all enabled jobs."""        # 存储实例未初始化时直接返回        if not self._store:            return        # 获取当前时间戳（毫秒）        now = _now_ms()        # 遍历所有任务，重新计算启用任务的下次执行时间        for job in self._store.jobs:            if job.enabled:                job.state.next_run_at_ms = _compute_next_run(job.schedule, now)    def _get_next_wake_ms(self) -> int | None:        """Get the earliest next run time across all jobs."""        # 存储实例未初始化时返回None        if not self._store:            return None        # 收集所有启用任务的下次执行时间戳        times = [j.state.next_run_at_ms for j in self._store.jobs                  if j.enabled and j.state.next_run_at_ms]        # 返回最早的执行时间（无任务时返回None）        return min(times) if times else None    def _arm_timer(self) -> None:        """Schedule the next timer tick."""        # 取消当前定时器任务（避免重复调度）        if self._timer_task:            self._timer_task.cancel()        # 获取最早的下次执行时间        next_wake = self._get_next_wake_ms()        # 无执行时间或服务已停止时，不设置定时器        if not next_wake or not self._running:            return        # 计算定时器延迟时间（毫秒）：确保非负        delay_ms = max(0, next_wake - _now_ms())        # 转换为秒（asyncio.sleep接收秒为单位）        delay_s = delay_ms / 1000        # 定义定时器回调函数：延迟后执行任务检查        async def tick():            await asyncio.sleep(delay_s)            # 服务仍在运行时，处理定时器触发逻辑            if self._running:                await self._on_timer()        # 创建并保存定时器任务        self._timer_task = asyncio.create_task(tick())    async def _on_timer(self) -> None:        """Handle timer tick - run due jobs."""        # 存储实例未初始化时直接返回        if not self._store:            return        # 获取当前时间戳        now = _now_ms()        # 筛选出已到执行时间的任务：启用 + 有下次执行时间 + 当前时间≥执行时间        due_jobs = [            j for j in self._store.jobs            if j.enabled and j.state.next_run_at_ms and now >= j.state.next_run_at_ms        ]        # 遍历并执行所有到期任务        for job in due_jobs:            await self._execute_job(job)        # 保存任务状态到文件        self._save_store()        # 重新设置定时器（触发下一次任务执行）        self._arm_timer()    async def _execute_job(self, job: CronJob) -> None:        """Execute a single job."""        # 记录任务开始执行时间戳        start_ms = _now_ms()        # 记录执行日志        logger.info("Cron: executing job '{}' ({})", job.name, job.id)        try:            response = None            # 存在执行回调时，调用回调执行任务            if self.on_job:                response = await self.on_job(job)            # 更新任务状态：执行成功            job.state.last_status = "ok"            job.state.last_error = None            logger.info("Cron: job '{}' completed", job.name)        except Exception as e:            # 更新任务状态：执行失败            job.state.last_status = "error"            job.state.last_error = str(e)            logger.error("Cron: job '{}' failed: {}", job.name, e)        # 更新任务最后执行时间和更新时间        job.state.last_run_at_ms = start_ms        job.updated_at_ms = _now_ms()        # 处理一次性任务（at类型）        if job.schedule.kind == "at":            # 执行后删除任务            if job.delete_after_run:                self._store.jobs = [j for j in self._store.jobs if j.id != job.id]            # 执行后禁用任务（不删除）            else:                job.enabled = False                job.state.next_run_at_ms = None        else:            # 周期性任务：计算下次执行时间            job.state.next_run_at_ms = _compute_next_run(job.schedule, _now_ms())    # ========== Public API ==========    def list_jobs(self, include_disabled: bool = False) -> list[CronJob]:        """List all jobs."""        # 加载任务存储        store = self._load_store()        # 筛选任务：是否包含禁用任务        jobs = store.jobs if include_disabled else [j for j in store.jobs if j.enabled]        # 按下次执行时间排序（无执行时间的排最后）        return sorted(jobs, key=lambda j: j.state.next_run_at_ms or float('inf'))    def add_job(        self,        name: str,        schedule: CronSchedule,        message: str,        deliver: bool = False,        channel: str | None = None,        to: str | None = None,        delete_after_run: bool = False,    ) -> CronJob:        """Add a new job."""        # 加载任务存储        store = self._load_store()        # 校验调度规则合法性        _validate_schedule_for_add(schedule)        # 获取当前时间戳        now = _now_ms()        # 创建新任务实例        job = CronJob(            id=str(uuid.uuid4())[:8],  # 生成8位短UUID作为任务ID            name=name,  # 任务名称            enabled=True,  # 默认启用            schedule=schedule,  # 调度规则            # 任务负载            payload=CronPayload(                kind="agent_turn",                message=message,                deliver=deliver,                channel=channel,                to=to,            ),            # 任务状态（计算下次执行时间）            state=CronJobState(next_run_at_ms=_compute_next_run(schedule, now)),            created_at_ms=now,  # 创建时间            updated_at_ms=now,  # 更新时间            delete_after_run=delete_after_run,  # 执行后是否删除        )        # 将新任务添加到存储        store.jobs.append(job)        # 保存到文件        self._save_store()        # 重新设置定时器        self._arm_timer()        # 记录添加日志        logger.info("Cron: added job '{}' ({})", name, job.id)        return job    def remove_job(self, job_id: str) -> bool:        """Remove a job by ID."""        # 加载任务存储        store = self._load_store()        # 记录删除前的任务数量        before = len(store.jobs)        # 过滤掉指定ID的任务        store.jobs = [j for j in store.jobs if j.id != job_id]        # 判断是否成功删除        removed = len(store.jobs) < before        # 删除成功时，保存并重新设置定时器        if removed:            self._save_store()            self._arm_timer()            logger.info("Cron: removed job {}", job_id)        return removed    def enable_job(self, job_id: str, enabled: bool = True) -> CronJob | None:        """Enable or disable a job."""        # 加载任务存储        store = self._load_store()        # 遍历任务查找指定ID        for job in store.jobs:            if job.id == job_id:                # 更新启用状态                job.enabled = enabled                # 更新任务更新时间                job.updated_at_ms = _now_ms()                # 启用时重新计算下次执行时间                if enabled:                    job.state.next_run_at_ms = _compute_next_run(job.schedule, _now_ms())                # 禁用时清空下次执行时间                else:                    job.state.next_run_at_ms = None                # 保存并重新设置定时器                self._save_store()                self._arm_timer()                return job        # 未找到任务时返回None        return None    async def run_job(self, job_id: str, force: bool = False) -> bool:        """Manually run a job."""        # 加载任务存储        store = self._load_store()        # 遍历任务查找指定ID        for job in store.jobs:            if job.id == job_id:                # 非强制模式下，禁用的任务无法执行                if not force and not job.enabled:                    return False                # 执行任务                await self._execute_job(job)                # 保存并重新设置定时器                self._save_store()                self._arm_timer()                return True        # 未找到任务时返回False        return False    def status(self) -> dict:        """Get service status."""        # 加载任务存储        store = self._load_store()        # 返回服务状态：运行状态、任务数量、下次唤醒时间        return {            "enabled": self._running,            "jobs": len(store.jobs),            "next_wake_at_ms": self._get_next_wake_ms(),        }

0x06 Dream机制

Dream 是nanobot 的异步深度记忆整合器，相当于人类睡眠时大脑对白天记忆的巩固与清理。它不是实时运行的，而是定时或手动触发，对 history.jsonl做批量分析并通过LLM更新长期记忆文件。

6.1 架构

触发方式如下：

• 定时触发：CronService按DreamConfig.interval_h（默认2h）调度dream.run()
• 手动触发：用户发送/dream命令，cmd_dream()创建asynciotask立即执行

6.2 流程

其关键设计决策如下：

6.3 特色

Cron 任务与 AgentLoop 的关系如下。普通 cron 任务完全经过 AgentLoop, 跟用户手动发消息走的是同一条路, 只是 session key 是 cron:<job_id> 而非用户的对话 key。Dream 则是绕过 AgentLoop 的特殊内部任务。

普通 cron 任务完整流程如下：

两条路径详细对比如下：