Agent Harness 工程：定时任务与 Scheduler

上一篇为 Agent Harness 增加了后台 Job。耗时命令启动后会立即返回 job_id，Agent 可以继续处理其它工作，运行时则在进程结束后把结果重新交给 Agent。

但后台执行只解决了“启动以后如何不阻塞”，没有解决“什么时候启动”。

如果希望 Agent 每天早上 8 点检查服务器运行状态，每天晚上 11 点运行自动化测试，仅靠当前能力仍然需要用户按时发一条消息。让模型自己调用 sleep(86400) 也不可靠：Harness 一旦重启，等待就会消失；执行时间还会随着每轮耗时逐渐漂移。

我们需要在 Agent Loop 和后台 Job 之外，再增加一个确定性的 Scheduler。它负责保存时间规则、判断某个时刻是否到期，并把每次触发转换为一个可恢复的 Task。

本章解决五个问题：

Schedule、ScheduleRun、Task 和 Job 分别负责什么；
如何正确处理 Cron、时区与夏令时；
Harness 停机期间错过的触发应该怎样恢复；
上一次任务还没结束时，下一次触发应该怎么办；
如何避免重启和重复轮询创建两份相同任务。

Schedule 不是 Task

“定时任务”这个词容易把时间规则和实际工作混在一起。实现之前，先把四个概念分开：

概念	回答的问题	生命周期	示例
Schedule	什么时候触发	长期存在，可以暂停和修改	每天 8:00，`Asia/Shanghai`
ScheduleRun	这一次是否已经触发	对应一个计划执行时刻	7 月 15 日 8:00 的服务器巡检
Task	这一次要完成什么	从创建到完成、失败或取消	检查服务器并汇总异常
Job	某个具体操作执行得怎样	从进程启动到退出	执行健康检查脚本或测试命令

Schedule 不应该反复“重新打开”同一个 Task。Task 一旦完成就应该保持完成，因此每个计划时刻都要生成独立的 ScheduleRun，再创建一个新的 Task。

Scheduler 只决定何时产生工作，不直接承担工作本身。ScheduleRun 负责去重和审计，Task System 管理目标，耗时操作继续交给后台 Job。

这条链路可以概括为：

Schedule -> ScheduleRun -> Task -> Agent -> Job
   何时         哪一次         做什么    决策    具体执行

这样做还有一个重要好处：Scheduler 本身不需要调用模型。时间计算、去重和状态推进都由普通程序完成，只有新的 Task 真正就绪时才唤醒 Agent。

一条 Schedule 需要保存什么

先看一条“每天早上 8 点检查服务器”的配置：

{
  "id": "schedule-server-health",
  "name": "每日服务器巡检",
  "cron": "0 8 * * *",
  "timezone": "Asia/Shanghai",
  "task_title": "执行每日服务器巡检",
  "task_description": "检查服务器运行状态、磁盘空间和关键服务，并汇总异常。",
  "misfire_policy": "run_once",
  "misfire_grace_seconds": 300,
  "overlap_policy": "skip",
  "max_catch_up": 3,
  "enabled": true,
  "next_run_at": "2026-07-15T00:00:00+00:00",
  "schema_version": 1
}

用户看到的是上海时间 8:00，内部的 next_run_at 则统一保存为 UTC。Schedule 必须同时保留原始时区，因为下一次执行时间仍然要按照当地日历计算，不能简单地在上一次 UTC 时间上加 24 小时。

用数据类表达这些字段：

from dataclasses import asdict, dataclass
from typing import Literal


MisfirePolicy = Literal["skip", "run_once", "catch_up"]
OverlapPolicy = Literal["skip", "allow"]
ScheduleRunStatus = Literal["pending", "dispatched", "skipped"]


@dataclass
class Schedule:
    id: str
    name: str
    cron: str
    timezone: str
    task_title: str
    task_description: str
    next_run_at: str
    misfire_policy: MisfirePolicy = "run_once"
    misfire_grace_seconds: int = 300
    overlap_policy: OverlapPolicy = "skip"
    max_catch_up: int = 3
    enabled: bool = True
    created_at: str = ""
    updated_at: str = ""
    schema_version: int = 1

    def to_dict(self) -> dict:
        return asdict(self)


@dataclass
class ScheduleRun:
    id: str
    schedule_id: str
    scheduled_for: str
    task_title: str
    task_description: str
    status: ScheduleRunStatus = "pending"
    task_id: str | None = None
    reason: str | None = None
    created_at: str = ""
    dispatched_at: str | None = None
    schema_version: int = 1

    def to_dict(self) -> dict:
        return asdict(self)

ScheduleRun 会复制当时的标题和描述，而不是只保存 schedule_id。如果 Schedule 在触发后、Task 创建前被修改，这次已经发生的触发仍然使用旧快照，不会悄悄改变含义。

它只保存 pending、dispatched 和 skipped 三种分发状态。进入 dispatched 后，最终是完成、失败还是取消，都从关联 Task 推导，不在 ScheduleRun 中再维护一份重复状态。

目录结构继续放在 .agent 中：

.agent/
├── schedules/
│   └── schedule-server-health.json
└── schedule-runs/
    └── schedule-run-94b2c0d9fd4a2c31.json

Schedule Store 可以直接复用 Task Store 的基础设施：校验 ID、使用文件锁保护读改写，并通过临时文件、fsync() 和 os.replace() 原子替换 JSON。这里不再重复整套文件存储代码。

不要自己解析 Cron

Cron 看起来只有五个字段，真正的边界却不少：月份天数、闰年、星期语义和夏令时都可能影响结果。这里使用 croniter 解析表达式，时区使用 Python 标准库的 zoneinfo：

pip install croniter tzdata

tzdata 为缺少系统 IANA 时区数据库的环境提供一致的数据来源。

第一版只接受标准五字段 Cron，不开放秒和年份字段：

from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

from croniter import croniter


UTC = timezone.utc
MIN_SCHEDULE_INTERVAL = timedelta(minutes=1)


def parse_utc(value: str) -> datetime:
    result = datetime.fromisoformat(value)
    if result.tzinfo is None:
        raise ValueError("timestamp must include a timezone")
    return result.astimezone(UTC)


def next_fire_time(
    expression: str,
    timezone_name: str,
    after_utc: datetime,
) -> datetime:
    zone = ZoneInfo(timezone_name)
    local_base = after_utc.astimezone(zone)
    local_next = croniter(expression, local_base, day_or=True).get_next(datetime)
    return local_next.astimezone(UTC)


def validate_schedule_expression(
    expression: str,
    timezone_name: str,
    now_utc: datetime,
) -> datetime:
    if len(expression.split()) != 5 or not croniter.is_valid(expression, strict=True):
        raise ValueError("cron must be a valid five-field expression")

    try:
        ZoneInfo(timezone_name)
    except ZoneInfoNotFoundError as exc:
        raise ValueError(f"unknown IANA timezone: {timezone_name}") from exc

    first = next_fire_time(expression, timezone_name, now_utc)
    second = next_fire_time(expression, timezone_name, first)
    if second - first < MIN_SCHEDULE_INTERVAL:
        raise ValueError("schedule interval is too short")

    return first

day_or=True 明确采用常见 Cron 语义：当“每月第几天”和“星期几”同时受限时，任意一个匹配就会触发。这个规则很容易被误解，因此创建 Schedule 时除了保存表达式，还应该把接下来几次本地执行时间展示给用户确认。

例如：

Cron: 0 8 * * *
Timezone: Asia/Shanghai

接下来三次执行：
- 2026-07-15 08:00 +08:00
- 2026-07-16 08:00 +08:00
- 2026-07-17 08:00 +08:00

所有内部比较都使用带时区的 UTC datetime。不要混用 naive datetime、服务器本地时区和字符串排序。

夏令时不是加 24 小时

Asia/Shanghai 没有当前夏令时切换，但 Scheduler 不能假设所有时区都如此。

在采用夏令时的地区：

春季会有一段本地时间不存在；
秋季会有一段本地时间出现两次；
时区规则还可能随着法规更新。

因此，“每天凌晨 2:30”并不总是一个无歧义的时刻。zoneinfo 会根据 IANA 时区数据处理 UTC 偏移，croniter 也要求基准时间带有时区；但产品仍然要明确自己的 DST 策略，并为使用到的时区编写跨切换日测试。

本章的最小代码沿用 croniter 对时区感知时间的处理结果。若产品要求在不存在的时刻“跳过”或“顺延”，在重复时刻“执行一次”或“执行两次”，就要在 Cron 计算外再增加显式策略，不能依赖未说明的默认行为。ScheduleRun 使用 UTC scheduled_for 生成 ID，因此重复的本地时间即使拥有不同 UTC 偏移，也能被区分。

最稳妥的约束是：

保存 IANA 时区名称，例如 America/New_York，不要只保存 UTC-5；
每次根据 Cron 和时区计算下一个日历时刻；
将结果转换为 UTC 后持久化；
时区数据库升级后，重新计算尚未触发的 next_run_at；
对不存在或重复的本地时刻采用固定策略，并在创建时告知用户。

对于服务器内部维护任务，如果业务并不关心当地日历，可以直接使用 UTC，避开大部分 DST 歧义。

处理错过执行

假设自动化测试原本应该每天 23:00 运行，但 Harness 从 22:30 停机到第二天 01:00。重启后至少有三种处理方式：

策略	行为	适合场景
`skip`	跳过已经错过的时刻	高频采样、过期后没有价值的任务
`run_once`	立即补一次，忽略更早的遗漏	每日巡检、夜间测试
`catch_up`	补跑最近若干次，但设置数量上限	每次执行都代表独立业务周期的任务

还需要一个 misfire_grace_seconds。Scheduler 晚几秒醒来是正常调度延迟，不应该立刻被视为停机恢复；只有超过宽限时间才应用 misfire 策略。

错过执行决定“过去的触发要不要补”，重叠策略决定“上一次还没结束时能不能再开一次”。两者是独立维度。

先实现查找最近计划时刻的辅助函数：

def previous_fire_time(
    expression: str,
    timezone_name: str,
    at_or_before_utc: datetime,
) -> datetime:
    zone = ZoneInfo(timezone_name)
    local_base = (at_or_before_utc + timedelta(microseconds=1)).astimezone(zone)
    local_previous = croniter(
        expression,
        local_base,
        day_or=True,
    ).get_prev(datetime)
    return local_previous.astimezone(UTC)


def latest_due_times(
    schedule: Schedule,
    now_utc: datetime,
    limit: int,
) -> list[datetime]:
    lower_bound = parse_utc(schedule.next_run_at)
    cursor = now_utc
    result = []

    while len(result) < limit:
        candidate = previous_fire_time(
            schedule.cron,
            schedule.timezone,
            cursor,
        )
        if candidate < lower_bound:
            break
        result.append(candidate)
        cursor = candidate - timedelta(microseconds=1)

    return sorted(result)

然后根据策略计算本轮需要生成哪些 ScheduleRun，以及新的游标：

def plan_due_times(
    schedule: Schedule,
    now_utc: datetime,
) -> tuple[list[datetime], datetime]:
    scheduled_for = parse_utc(schedule.next_run_at)
    if scheduled_for > now_utc:
        return [], scheduled_for

    lateness = (now_utc - scheduled_for).total_seconds()
    is_misfire = lateness > schedule.misfire_grace_seconds

    if not is_misfire:
        due_times = [scheduled_for]
        next_base = scheduled_for
    elif schedule.misfire_policy == "skip":
        due_times = []
        next_base = now_utc
    elif schedule.misfire_policy == "run_once":
        due_times = latest_due_times(schedule, now_utc, limit=1)
        next_base = now_utc
    elif schedule.misfire_policy == "catch_up":
        limit = max(1, min(schedule.max_catch_up, 20))
        due_times = latest_due_times(schedule, now_utc, limit=limit)
        next_base = now_utc
    else:
        raise ValueError(f"unknown misfire policy: {schedule.misfire_policy}")

    next_run_at = next_fire_time(
        schedule.cron,
        schedule.timezone,
        next_base,
    )
    return due_times, next_run_at

catch_up 只补最近的有限次数，不应该从数月前开始无限回放。重试也不能通过修改 next_run_at 实现：Schedule 游标描述下一次计划时刻，某次执行失败后的重试属于对应 Task 或 Job 的生命周期。

为每次触发生成稳定 ID

Scheduler 可能在同一分钟轮询多次，也可能在写入后、返回前崩溃。如果每次都生成随机 ID，同一个计划时刻就会产生多份 Task。

更可靠的幂等键是：

schedule_id + scheduled_for_utc

用 UUID v5 生成稳定的 ScheduleRun ID：

import uuid


def schedule_run_id(schedule_id: str, scheduled_for: datetime) -> str:
    canonical_time = scheduled_for.astimezone(UTC).isoformat(timespec="seconds")
    key = f"{schedule_id}:{canonical_time}"
    digest = uuid.uuid5(uuid.NAMESPACE_URL, key).hex[:16]
    return f"schedule-run-{digest}"


def scheduled_task_id(run_id: str) -> str:
    return f"task-scheduled-{run_id.removeprefix('schedule-run-')}"

相同 Schedule 在相同 UTC 时刻无论被扫描多少次，都会得到相同 ID。ScheduleRun 文件已经存在时，写入逻辑只需验证定义一致并返回原记录；定义不同则报告冲突，不能覆盖。

先记录触发，再推进游标

一次 Scheduler tick 的写入顺序很重要：

在锁内重新读取 Schedule；
计算本轮到期时刻；
为每个时刻写入稳定 ID 的 ScheduleRun；
最后更新 Schedule 的 next_run_at。

不能先推进游标再写 ScheduleRun。否则进程在两次写入之间崩溃，这个计划时刻会永久丢失。

下面省略了上一章已经实现过的锁和原子 JSON 细节，只保留核心流程：

def collect_due_runs(
    schedule_store: ScheduleStore,
    task_store: TaskStore,
    now_utc: datetime,
) -> list[ScheduleRun]:
    created = []

    with schedule_store.locked():
        for schedule in schedule_store.list_enabled_unlocked():
            due_times, next_run_at = plan_due_times(schedule, now_utc)

            for scheduled_for in due_times:
                run_id = schedule_run_id(schedule.id, scheduled_for)
                existing = schedule_store.find_run_unlocked(run_id)
                if existing:
                    continue

                overlap = (
                    schedule.overlap_policy == "skip"
                    and has_active_run(schedule_store, task_store, schedule.id)
                )
                run = ScheduleRun(
                    id=run_id,
                    schedule_id=schedule.id,
                    scheduled_for=scheduled_for.isoformat(timespec="seconds"),
                    task_title=schedule.task_title,
                    task_description=schedule.task_description,
                    status="skipped" if overlap else "pending",
                    reason="previous run is still active" if overlap else None,
                    created_at=utc_now_text(),
                )
                schedule_store.write_run_unlocked(run)
                created.append(run)

            schedule.next_run_at = next_run_at.isoformat(timespec="seconds")
            schedule.updated_at = utc_now_text()
            schedule_store.write_schedule_unlocked(schedule)

    return created

虽然 ScheduleRun 和 Schedule 是两个 JSON 文件，稳定 ID 仍然让流程具备可恢复性：

如果写完 ScheduleRun、更新游标前崩溃，重启后会再次得到同一个 run ID，跳过重复记录后继续推进游标；
如果所有记录都还没写入，重启后 next_run_at 仍然到期，会重新执行整个步骤；
不会出现游标已经越过某个时刻、对应触发却完全不存在的情况。

单机文件锁只能保护同一台机器。多实例部署应该使用数据库事务，并在 (schedule_id, scheduled_for) 上建立唯一约束。

把 ScheduleRun 分发为 Task

ScheduleRun 的 pending 状态可以看作一个持久化 Outbox：只要它还没有关联 Task，Dispatcher 就会继续尝试。

Task ID 同样由 run ID 确定，因此“Task 已经创建，但 ScheduleRun 还没更新”也能安全恢复：

def dispatch_schedule_run(
    schedule_store: ScheduleStore,
    task_store: TaskStore,
    runtime_events: Queue[dict],
    run: ScheduleRun,
) -> None:
    if run.status != "pending":
        return

    task_id = scheduled_task_id(run.id)
    task = create_task(
        store=task_store,
        task_id=task_id,
        title=run.task_title,
        description=run.task_description,
    )

    with schedule_store.locked():
        current = schedule_store.read_run_unlocked(run.id)
        if current.status != "pending":
            return
        current.status = "dispatched"
        current.task_id = task.id
        current.dispatched_at = utc_now_text()
        schedule_store.write_run_unlocked(current)

    runtime_events.put(
        {
            "event_id": f"scheduled-task-ready:{run.id}",
            "kind": "scheduled_task_ready",
            "schedule_id": run.schedule_id,
            "schedule_run_id": run.id,
            "scheduled_for": run.scheduled_for,
            "task_id": task.id,
        }
    )

如果进程在 create_task() 后崩溃，Dispatcher 重试时会使用同一个 Task ID。上一章的 create_task() 已经具备幂等检查，相同定义会返回现有 Task，不会重复创建。

事件队列只是唤醒提示，Task 文件才是持久化事实。如果进程在 Task 落盘后、事件入队前重启，Runtime 启动时扫描由 ScheduleRun 关联且尚未进入终态的 Task，再补发唤醒事件即可。

Prompt Assembler 可以把事件渲染为短暂的运行时 Section：

Scheduled task is ready:
- schedule_id: schedule-server-health
- schedule_run_id: schedule-run-94b2c0d9fd4a2c31
- scheduled_for: 2026-07-15T00:00:00+00:00
- task_id: task-scheduled-94b2c0d9fd4a2c31

这不是一条新的用户消息，而是用户之前授权的 Schedule 在约定时间产生的内部事件。Agent 随后正常认领 Task，使用 TodoWrite 规划步骤；需要执行测试或脚本时，再创建上一章的后台 Job。

上一次还没完成怎么办

错过执行和运行重叠经常被混为一谈：

Misfire：计划时刻已经过去，但 Scheduler 当时没有正常触发；
Overlap：新的计划时刻到达时，上一次 Task 仍未结束。

第一版只提供两种重叠策略：

策略	行为	适合场景
`skip`	记录一次 `skipped`，不建 Task	巡检、构建、可能产生副作用的任务
`allow`	为本次触发创建独立 Task	各周期互不影响、允许并行的任务

has_active_run() 不应该只看 ScheduleRun 的 dispatched 字段，而要读取关联 Task 的真实状态：pending 和 in_progress 仍然活跃，completed、failed、cancelled 才是终态。这样不会在 ScheduleRun 和 Task 中维护两套互相矛盾的完成状态。

不建议第一版实现 replace。取消旧 Task 并不代表已经启动的外部请求、部署或后台进程能够安全回滚。是否允许替换必须由具体工具和业务语义决定。

Scheduler 主循环

Scheduler Loop 不参与模型推理，只做三件事：收集到期时刻、分发 pending run、等待下一次检查。

def scheduler_loop(state: HarnessState) -> None:
    while not state.shutdown_requested:
        now_utc = utc_now()
        collect_due_runs(
            state.schedule_store,
            state.task_store,
            now_utc,
        )

        for run in state.schedule_store.list_pending_runs():
            dispatch_schedule_run(
                state.schedule_store,
                state.task_store,
                state.runtime_event_queue,
                run,
            )

        state.shutdown_event.wait(timeout=1.0)

和后台 Job 监控器一样，它可以运行在独立线程或异步任务中。shutdown_event.wait() 比 time.sleep() 更适合这里，因为 Harness 退出时可以立即唤醒循环。

一秒轮询是为了便于理解。实际实现可以读取最早的 next_run_at 计算等待时间，同时设置一个最大睡眠上限，以便 Schedule 被创建、修改或恢复时及时重新计算。

启动顺序也很重要：

加载并校验 Schedule 与 ScheduleRun；
恢复 pending run，补齐已经存在的 Task；
扫描已分发但仍未结束的定时 Task，重建唤醒提示；
根据当前时间执行 misfire 计算；
启动持续调度循环。

暴露 Schedule 工具

Agent 需要通过专用工具管理 Schedule，不能直接编辑 .agent/schedules/*.json：

工具	作用
`schedule_create`	创建 Schedule，并返回接下来几次执行时间
`schedule_get`	查看规则、策略和下一次触发
`schedule_list`	列出启用、暂停或全部 Schedule
`schedule_update`	修改时间规则、任务模板或策略
`schedule_pause`	停止产生新的 ScheduleRun
`schedule_resume`	重新计算 `next_run_at` 后恢复
`schedule_delete`	停止未来触发，但保留历史记录
`schedule_run_now`	立即创建一次独立 ScheduleRun
`schedule_history`	查看触发记录和关联 Task

schedule_create 的 Schema 可以保持明确而克制：

SCHEDULE_CREATE_TOOL = {
    "type": "function",
    "function": {
        "name": "schedule_create",
        "description": "Create a persistent recurring task schedule.",
        "parameters": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "cron": {
                    "type": "string",
                    "description": "Five-field cron expression.",
                },
                "timezone": {
                    "type": "string",
                    "description": "IANA timezone such as Asia/Shanghai.",
                },
                "task_title": {"type": "string"},
                "task_description": {"type": "string"},
                "misfire_policy": {
                    "type": "string",
                    "enum": ["skip", "run_once", "catch_up"],
                },
                "overlap_policy": {
                    "type": "string",
                    "enum": ["skip", "allow"],
                },
            },
            "required": [
                "name",
                "cron",
                "timezone",
                "task_title",
                "task_description",
                "misfire_policy",
                "overlap_policy",
            ],
        },
    },
}

修改 Cron 或时区后，要从当前时间重新计算 next_run_at，不能沿用旧游标。修改不会重写已经存在的 ScheduleRun；暂停和删除也不会自动取消正在运行的 Task 或 Job。

schedule_run_now 不属于 Cron 序列，不能只用调用时的当前秒生成 ID。处理函数应该使用稳定的 tool_call_id 或客户端 request_id 作为幂等键，使同一次工具调用在响应丢失后重试时仍然返回同一个 ScheduleRun。

定时执行是一种长期授权

创建 Schedule 和普通调用一次工具不同。用户实际上是在授权 Harness 将来反复唤醒 Agent，因此权限边界必须更严格。

至少要遵守这些规则：

schedule_create、修改、恢复和删除需要明确确认；
确认界面展示本地时区、接下来几次执行时间和任务内容；
Schedule 只保存任务描述，不直接保存 token、密码和其它凭据；
每次触发仍然经过当前工具权限、路径限制和网络策略；
一次性的工具批准不能自动升级为永久批准；
无人值守运行只能使用预先定义的权限 Profile 和资源预算；
限制 Schedule 数量、最小触发间隔、单日模型调用次数和最大费用；
Schedule 的创建、修改、暂停、触发和删除都写入审计日志。

对于“每天晚上运行自动化测试”，Schedule 只负责创建测试 Task。Agent 是否可以直接执行测试命令，仍然由触发时生效的 Permission 与 Hook 决定。后台执行和定时执行都不能成为绕过安全检查的通道。

多实例与时钟问题

本章的文件存储适合单机 Harness。只要所有 Scheduler 共享同一把文件锁，稳定 run ID 就能防止同一时刻重复写入。

多机器部署需要更强的协调能力：

使用数据库保存 Schedule 和 ScheduleRun；
对 (schedule_id, scheduled_for) 建立唯一约束；
通过租约或行锁选出当前调度者；
把 pending run 放入支持确认与重投的消息系统；
Worker 使用稳定 run ID 作为下游幂等键。

即便如此，也不要宣称外部副作用具备“恰好一次”语义。数据库可以保证同一个 ScheduleRun 只创建一条记录，但网络请求可能已经成功、响应却丢失。真正的恰好一次仍然需要目标系统支持幂等键或事务。

Scheduler 使用的是日历时间，因此必须读取系统 UTC 时钟；time.monotonic() 适合计算超时，不能用来表达“每天 8 点”。生产环境还应该同步系统时钟，并监控时间大幅跳变。

常见坑

第一，用 sleep(86400) 实现每天执行。 它会漂移，也无法在进程重启后恢复。

第二，只保存本地时间，不保存 IANA 时区。 夏令时和服务器迁移后都会产生歧义。

第三，把 Schedule 当成一个反复重开的 Task。 历史结果会被覆盖，也无法区分每次触发。

第四，先推进 next_run_at，再记录 ScheduleRun。 中途崩溃会永久丢掉一次执行。

第五，每次触发都使用随机 ID。 重复轮询和崩溃重试会创建多份相同 Task。

第六，停机恢复后无限补跑。 长时间停机可能瞬间制造数千个任务，必须设置策略和上限。

第七，默认允许运行重叠。 测试、部署和巡检可能互相争抢资源，甚至产生重复副作用。

第八，用修改 Schedule 游标表示失败重试。 计划周期和单次执行状态会被混在一起，重试应该由 Task 或 Job 管理。

第九，Schedule 直接执行 shell。 这会绕过 Agent、Task System、权限检查和结果验证。

第十，暂停 Schedule 时顺便终止现有工作。 暂停只影响未来触发，取消已经运行的 Task 或 Job 必须是独立操作。

第十一，多实例只依赖进程内锁。 两台机器仍然可能同时触发，需要共享存储、唯一约束和调度租约。

小结

本文在后台 Job 之后，为 Agent Harness 增加了一个最小 Scheduler：

用 Schedule 保存 Cron、时区、任务模板和下一次执行时间；
用 ScheduleRun 表达一次具体触发，避免复用已经完成的 Task；
使用 croniter 和 zoneinfo 计算带时区的日历时间，内部统一保存 UTC；
通过 skip、run_once 和有限 catch_up 处理错过执行；
通过 skip 或 allow 明确处理运行重叠；
用 schedule_id + scheduled_for 生成稳定 ID，抵抗重复轮询和崩溃重试；
先持久化 ScheduleRun，再推进 next_run_at，避免丢失触发；
将 pending run 幂等地分发为 Task，再由 Agent 和后台 Job 完成实际工作；
让 Schedule 工具经过明确确认，并为无人值守运行设置权限与资源预算；
在多实例环境中使用数据库唯一约束、租约和可重投消息代替本地文件锁。

至此，Agent 不仅能管理长期任务、等待后台操作，还能在没有用户即时输入的情况下，按照约定时间自动开始新工作。

当这些能力一起运行后，Task System 中会持续出现新的就绪任务。当前 Harness 仍然主要依赖一个 Agent 逐项处理；即使任务之间没有依赖关系，也无法真正利用多个 Agent 并行推进。

下一章讲 Agent Team 与多 Agent 协作：让多个 Agent 共享同一张任务图，通过角色与能力选择合适的任务，使用认领、租约和消息机制协调进度，并把各自的结果安全地汇总回主任务。