实践｜Cron Scheduler，让时间检查产生工作，让原循环执行

先把五分钟走完，不必真的等五分钟

目标仍然是“每五分钟检查 main.py”。先保存一张表，再向同一个调度函数依次传入 09:00:01、09:00:40、09:05:00。预期新增事件数为 1、0、1；让消费者保持 busy 时，两个事件都留下；空闲后才分别交付。

这是注入时钟的实验，时间点由代码提供，文件读取与获准后的 Python 进程则真实执行。模型响应是固定替身，不访问在线服务。思想篇解释了时间表、队列和执行为何要分开。

源码位于独立公开仓库 learn-agent-harness 第十四课。Python3.10+，默认仅标准库：

bash

git clone https://github.com/Chengyunlai/learn-agent-harness.git
cd learn-agent-harness
python3 examples/s14-cron-scheduler/user_code/practice.py --decision allow
python3 examples/s14-cron-scheduler/user_code/practice.py --decision deny

允许路径对两条事件分别读取 expected.txt、运行 main.py，得到两次 42；拒绝路径也交付两轮，但两次都不能执行程序。每次 Agent 循环最多三次模型调用，拥有新的消息历史；这和原对话被后台线程随时修改是两种不同的实现。

宿主首先用 schedule_job 登记表。它验证表达式、提示长度和一次性标记，再写入 state.jobs。本会话最多创建四张，取消也不返还额度。模型需要管理表时，公开源码另提供 schedule_cron、list_crons、cancel_cron 的注册适配器；默认实验直接由宿主登记，不把自然语言解析也混进当前问题。

cron_matches：先弄清楚“现在”属于哪个刻度

函数输入表达式和带 UTC 偏移的 datetime，输出一个布尔值。格式错误会抛出异常，不会被吞成 False；否则“没到点”和“坏时间表”就无法区分。

cron_fields 先验证五项。当前子集仅接受 *、范围内单值，分钟额外接受 */1 到 */59；分钟 0–59、小时 0–23、日 1–31、月 1–12、星期 0–6。范围、列表和星期 7 都明确拒绝。注册时就调用这个 helper，坏表不会先进入正常登记。

真实执行轨迹回放；正式函数、日期与锁原样执行；时钟输入和轮次回调为公开替身，无模型或进程调用。

Pys14/core/runtime.py · cron_matches

未运行

def cron_matches(expression, when):
    """固定UTC+08:00解释时间；日与星期同时受限时任一满足即可。"""
    parts, fields = cron_fields(expression)
    local = checked_time(when).astimezone(LOCAL)
    minute, hour, day, month, weekday = fields
    if local.minute not in minute or local.hour not in hour or local.month not in month:
        return False
    day_ok = local.day in day
    week_ok = (local.weekday() + 1) % 7 in weekday
    if parts[2] == '*':
        return week_ok
    if parts[4] == '*':
        return day_ok
    return day_ok or week_ok

输出

点击行号设置断点，再点“调试”；也可以直接运行。

代码只读；执行发生在浏览器中，不会发送到博客服务器。

教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。

from copy import deepcopy
from datetime import datetime, timedelta, timezone
from threading import RLock
import re
LOCAL=timezone(timedelta(hours=8))
def cron_fields(expression):
    """有意限定子集：*、单值，分钟额外支持*/N；星期0为周日。"""
    if not isinstance(expression, str) or len(expression) > 80:
        raise ValueError('时间表需要有界字符串。')
    parts = expression.split()
    if len(parts) != 5:
        raise ValueError('需要分钟、小时、日、月、星期五项。')
    limits = ((0, 59), (0, 23), (1, 31), (1, 12), (0, 6))
    fields = []
    for index, (part, (low, high)) in enumerate(zip(parts, limits)):
        if part == '*':
            fields.append(set(range(low, high + 1)))
        elif index == 0 and re.fullmatch(r'\*/[1-9][0-9]?', part) and int(part[2:]) <= 59:
            fields.append(set(range(0, 60, int(part[2:]))))
        elif re.fullmatch(r'[0-9]{1,2}', part) and low <= int(part) <= high:
            fields.append({int(part)})
        else:
            raise ValueError('本课仅支持*、范围内单值和分钟*/1到*/59。')
    return parts, fields

def checked_time(value):
    if not isinstance(value, datetime) or value.tzinfo is None or value.utcoffset() is None:
        raise ValueError('时间必须带UTC偏移。')
    return value

def new_state():
    return {'jobs': {}, 'queue': [], 'active': None, 'results': [], 'created': 0}

def schedule_job(state, lock, expression, prompt, once=False):
    cron_fields(expression)
    if not isinstance(prompt, str) or not 1 <= len(prompt.strip()) <= 600 or type(once) is not bool:
        raise ValueError('提示需要1到600字符，once需要布尔值。')
    with lock:
        if state['created'] >= 4:
            raise ValueError('本次会话最多创建4张时间表。')
        state['created'] += 1
        job_id = 'cron-' + str(state['created'])
        state['jobs'][job_id] = {'cron': expression, 'prompt': prompt.strip(), 'once': once, 'last_slot': None}
    return job_id

def cron_matches(expression, when):
    """固定UTC+08:00解释时间；日与星期同时受限时任一满足即可。"""
    parts, fields = cron_fields(expression)
    local = checked_time(when).astimezone(LOCAL)
    minute, hour, day, month, weekday = fields
    if local.minute not in minute or local.hour not in hour or local.month not in month:
        return False
    day_ok = local.day in day
    week_ok = (local.weekday() + 1) % 7 in weekday
    if parts[2] == '*':
        return week_ok
    if parts[4] == '*':
        return day_ok
    return day_ok or week_ok

def enqueue_due(state, now, lock):
    """检查当前分钟，不补发停机期间的时刻；满队列不提前记已触发。"""
    instant = checked_time(now).astimezone(timezone.utc).replace(second=0, microsecond=0)
    slot = instant.isoformat()
    added = []
    with lock:
        for job_id, job in list(state['jobs'].items()):
            if job['last_slot'] is not None and slot <= job['last_slot']:
                continue
            if not cron_matches(job['cron'], instant) or len(state['queue']) >= 8:
                continue
            event = {'id': job_id + '@' + slot, 'job_id': job_id, 'slot': slot, 'prompt': job['prompt']}
            state['queue'].append(event)
            job['last_slot'] = slot
            if job['once']:
                del state['jobs'][job_id]
            added.append(deepcopy(event))
    return added

def dispatch_next(state, run_turn, lock, busy=False):
    """只在空闲时占用一项；锁外运行，失败记录后不自动重跑。"""
    with lock:
        if busy or state['active'] is not None:
            return {'status': 'busy'}
        if not state['queue']:
            return {'status': 'idle'}
        event = state['queue'].pop(0)
        state['active'] = event['id']
    outcome = {'id': event['id'], 'status': 'interrupted', 'text': ''}
    try:
        response = run_turn(deepcopy(event))
        outcome = {'id': event['id'], 'status': str(response['status'])[:40], 'text': str(response['text'])[:1200]}
    except Exception as error:
        outcome = {'id': event['id'], 'status': 'turn_error', 'text': type(error).__name__}
    finally:
        with lock:
            state['results'] = (state['results'] + [outcome])[-8:]
            state['active'] = None
    return deepcopy(outcome)


scenario = 'step'
state, lock = new_state(), RLock()
expression = '*/5 * * * *'
when = datetime.fromisoformat('2026-10-19T09:05:23+08:00')
if scenario=='not_due':when=datetime.fromisoformat('2026-10-19T09:04:00+08:00')
if scenario=='day_or':expression='0 9 1 * 1';when=datetime.fromisoformat('2026-10-19T09:00:00+08:00')
if scenario=='sunday':expression='0 9 * * 0';when=datetime.fromisoformat('2026-10-18T09:00:00+08:00')
if scenario=='invalid':expression='*/0 * * * *'
if scenario=='naive':when=when.replace(tzinfo=None)
if 'cron_matches'!='cron_matches':
    schedule_job(state,lock,expression,'检查main.py，运行仍需本次许可。',once=scenario=='once')
    if 'cron_matches'=='dispatch_next' and scenario!='empty':enqueue_due(state,when,lock)
    if scenario=='duplicate':enqueue_due(state,when,lock)
    if scenario=='backward':
        enqueue_due(state,when,lock)
        when=datetime.fromisoformat('2026-10-19T09:00:00+08:00')
    if scenario=='full':state['queue']=[{'id':'earlier-'+str(i)} for i in range(8)]
    if scenario=='active':state['active']='another-turn'
received=[]
def run_turn(event):
    received.append(deepcopy(event))
    if scenario=='failed':raise OSError('教学轮次失败')
    return {'status':'call_limit' if scenario=='model_stop' else 'answered','text':'模型已停下；程序是否成功需看执行证据。'}
busy=scenario=='busy'


# 调用入口
try:
    observed=cron_matches(expression, when)
except ValueError as error:
    observed={"error":str(error)}
print(observed)

校验五字段，并统一到UTC+08:00。
分钟、小时、月同时匹配。
按日与星期约束交回布尔值。

checked_time 要求输入带偏移，然后转成固定 UTC+08:00。分钟、小时、月份必须同时满足；日与星期的判断随后才进行。Python 的 weekday 把星期一记为 0，表达式把周日记为 0，因此要做加一取余转换。

“日期不符但周一”采用 0 9 1 * 1：2026 年 10 月 19 日不是一号，却是星期一，所以可以匹配。日与星期同时受限时取 OR，和 Cronie 手册描述的这条规则一致。月份不符仍会提前返回 False，不能把所有字段都改成 OR。

“时间缺少偏移”会在转换前拒绝，避免依赖运行机器的隐含时区。本课始终使用固定的八小时时区偏移。*/N 从本小时的第 0 分钟开始匹配刻度，不是“从创建时起经过 N 分钟”；刚登记时当前分钟符合，也可能马上产生一次事件。

enqueue_due：队列先接住，再写已触发标记

state 保存 jobs、queue、active、results 和创建额度；所有访问使用同一把 RLock。enqueue_due 先把 now 转为 UTC 并清掉秒，得到包含日期的 slot，再在锁内遍历登记快照。

一个表先检查 last_slot，再判断时间和容量。slot 小于或等于已成功触发的分钟，就跳过，既覆盖同分钟反复检查，也防止时钟回退时重发旧时刻。没有全局“今天已经执行”开关，第二天仍能正常产生事件。

真实执行轨迹回放；正式函数、日期与锁原样执行；时钟输入和轮次回调为公开替身，无模型或进程调用。

Pys14/core/runtime.py · enqueue_due

未运行

def enqueue_due(state, now, lock):
    """检查当前分钟，不补发停机期间的时刻；满队列不提前记已触发。"""
    instant = checked_time(now).astimezone(timezone.utc).replace(second=0, microsecond=0)
    slot = instant.isoformat()
    added = []
    with lock:
        for job_id, job in list(state['jobs'].items()):
            if job['last_slot'] is not None and slot <= job['last_slot']:
                continue
            if not cron_matches(job['cron'], instant) or len(state['queue']) >= 8:
                continue
            event = {'id': job_id + '@' + slot, 'job_id': job_id, 'slot': slot, 'prompt': job['prompt']}
            state['queue'].append(event)
            job['last_slot'] = slot
            if job['once']:
                del state['jobs'][job_id]
            added.append(deepcopy(event))
    return added

输出

点击行号设置断点，再点“调试”；也可以直接运行。

代码只读；执行发生在浏览器中，不会发送到博客服务器。

教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。

from copy import deepcopy
from datetime import datetime, timedelta, timezone
from threading import RLock
import re
LOCAL=timezone(timedelta(hours=8))
def cron_fields(expression):
    """有意限定子集：*、单值，分钟额外支持*/N；星期0为周日。"""
    if not isinstance(expression, str) or len(expression) > 80:
        raise ValueError('时间表需要有界字符串。')
    parts = expression.split()
    if len(parts) != 5:
        raise ValueError('需要分钟、小时、日、月、星期五项。')
    limits = ((0, 59), (0, 23), (1, 31), (1, 12), (0, 6))
    fields = []
    for index, (part, (low, high)) in enumerate(zip(parts, limits)):
        if part == '*':
            fields.append(set(range(low, high + 1)))
        elif index == 0 and re.fullmatch(r'\*/[1-9][0-9]?', part) and int(part[2:]) <= 59:
            fields.append(set(range(0, 60, int(part[2:]))))
        elif re.fullmatch(r'[0-9]{1,2}', part) and low <= int(part) <= high:
            fields.append({int(part)})
        else:
            raise ValueError('本课仅支持*、范围内单值和分钟*/1到*/59。')
    return parts, fields

def checked_time(value):
    if not isinstance(value, datetime) or value.tzinfo is None or value.utcoffset() is None:
        raise ValueError('时间必须带UTC偏移。')
    return value

def new_state():
    return {'jobs': {}, 'queue': [], 'active': None, 'results': [], 'created': 0}

def schedule_job(state, lock, expression, prompt, once=False):
    cron_fields(expression)
    if not isinstance(prompt, str) or not 1 <= len(prompt.strip()) <= 600 or type(once) is not bool:
        raise ValueError('提示需要1到600字符，once需要布尔值。')
    with lock:
        if state['created'] >= 4:
            raise ValueError('本次会话最多创建4张时间表。')
        state['created'] += 1
        job_id = 'cron-' + str(state['created'])
        state['jobs'][job_id] = {'cron': expression, 'prompt': prompt.strip(), 'once': once, 'last_slot': None}
    return job_id

def cron_matches(expression, when):
    """固定UTC+08:00解释时间；日与星期同时受限时任一满足即可。"""
    parts, fields = cron_fields(expression)
    local = checked_time(when).astimezone(LOCAL)
    minute, hour, day, month, weekday = fields
    if local.minute not in minute or local.hour not in hour or local.month not in month:
        return False
    day_ok = local.day in day
    week_ok = (local.weekday() + 1) % 7 in weekday
    if parts[2] == '*':
        return week_ok
    if parts[4] == '*':
        return day_ok
    return day_ok or week_ok

def enqueue_due(state, now, lock):
    """检查当前分钟，不补发停机期间的时刻；满队列不提前记已触发。"""
    instant = checked_time(now).astimezone(timezone.utc).replace(second=0, microsecond=0)
    slot = instant.isoformat()
    added = []
    with lock:
        for job_id, job in list(state['jobs'].items()):
            if job['last_slot'] is not None and slot <= job['last_slot']:
                continue
            if not cron_matches(job['cron'], instant) or len(state['queue']) >= 8:
                continue
            event = {'id': job_id + '@' + slot, 'job_id': job_id, 'slot': slot, 'prompt': job['prompt']}
            state['queue'].append(event)
            job['last_slot'] = slot
            if job['once']:
                del state['jobs'][job_id]
            added.append(deepcopy(event))
    return added

def dispatch_next(state, run_turn, lock, busy=False):
    """只在空闲时占用一项；锁外运行，失败记录后不自动重跑。"""
    with lock:
        if busy or state['active'] is not None:
            return {'status': 'busy'}
        if not state['queue']:
            return {'status': 'idle'}
        event = state['queue'].pop(0)
        state['active'] = event['id']
    outcome = {'id': event['id'], 'status': 'interrupted', 'text': ''}
    try:
        response = run_turn(deepcopy(event))
        outcome = {'id': event['id'], 'status': str(response['status'])[:40], 'text': str(response['text'])[:1200]}
    except Exception as error:
        outcome = {'id': event['id'], 'status': 'turn_error', 'text': type(error).__name__}
    finally:
        with lock:
            state['results'] = (state['results'] + [outcome])[-8:]
            state['active'] = None
    return deepcopy(outcome)


scenario = 'due'
state, lock = new_state(), RLock()
expression = '*/5 * * * *'
when = datetime.fromisoformat('2026-10-19T09:05:23+08:00')
if scenario=='not_due':when=datetime.fromisoformat('2026-10-19T09:04:00+08:00')
if scenario=='day_or':expression='0 9 1 * 1';when=datetime.fromisoformat('2026-10-19T09:00:00+08:00')
if scenario=='sunday':expression='0 9 * * 0';when=datetime.fromisoformat('2026-10-18T09:00:00+08:00')
if scenario=='invalid':expression='*/0 * * * *'
if scenario=='naive':when=when.replace(tzinfo=None)
if 'enqueue_due'!='cron_matches':
    schedule_job(state,lock,expression,'检查main.py，运行仍需本次许可。',once=scenario=='once')
    if 'enqueue_due'=='dispatch_next' and scenario!='empty':enqueue_due(state,when,lock)
    if scenario=='duplicate':enqueue_due(state,when,lock)
    if scenario=='backward':
        enqueue_due(state,when,lock)
        when=datetime.fromisoformat('2026-10-19T09:00:00+08:00')
    if scenario=='full':state['queue']=[{'id':'earlier-'+str(i)} for i in range(8)]
    if scenario=='active':state['active']='another-turn'
received=[]
def run_turn(event):
    received.append(deepcopy(event))
    if scenario=='failed':raise OSError('教学轮次失败')
    return {'status':'call_limit' if scenario=='model_stop' else 'answered','text':'模型已停下；程序是否成功需看执行证据。'}
busy=scenario=='busy'


# 调用入口
try:
    observed=enqueue_due(state, when, lock)
except ValueError as error:
    observed={"error":str(error)}
print(observed)

构造UTC分钟并检查去重和容量。
队列接收成功后才写last_slot。
一次性表移除不删除待交付事件。

在 queue.append 行停下时，事件尚未进入队列。下一步以后，真实 queue 长度增加；来到 last_slot 赋值行时，才准备把这个分钟记为已触发。图的队列数量直接读取执行快照，没有根据所选案例猜测。

“队列已满”保留八项，并让 last_slot 保持原值。如果本分钟释放了位置，再次检查仍可入队；之后的分钟只检查当下，不回头重放所有错过的时刻。本课因此同时保留了明确的容量边界和明确的不补发规则。

“一次性时间表”在事件入队后删除 jobs 里的表，队列中的事件仍在。这里的 once 是宿主登记参数，不表示一种新的 cron 日期语法。取消则会删除未来表和未交付项；已经由消费者占用的轮次继续运行，不把删除表伪装成程序已停止。

dispatch_next：只把短小的状态变更放进锁

消费者先在锁内检查 busy、active 和 queue。未空闲时直接返回 busy，队列没有变化；空队列返回 idle，也不会请求模型。可以交付时，取首项并把事件 ID 写入 active，这两个动作在同一个临界段内。

随后释放锁，调用 run_turn。它接到的是事件副本，再把事件 ID 和提示组成新的问题，交给唯一的原 Agent 循环。不能为了避免重复交付，把整个模型请求都放在锁内，否则调度器也无法及时追加后续事件。

真实执行轨迹回放；正式函数、日期与锁原样执行；时钟输入和轮次回调为公开替身，无模型或进程调用。

Pys14/core/runtime.py · dispatch_next

未运行

def dispatch_next(state, run_turn, lock, busy=False):
    """只在空闲时占用一项；锁外运行，失败记录后不自动重跑。"""
    with lock:
        if busy or state['active'] is not None:
            return {'status': 'busy'}
        if not state['queue']:
            return {'status': 'idle'}
        event = state['queue'].pop(0)
        state['active'] = event['id']
    outcome = {'id': event['id'], 'status': 'interrupted', 'text': ''}
    try:
        response = run_turn(deepcopy(event))
        outcome = {'id': event['id'], 'status': str(response['status'])[:40], 'text': str(response['text'])[:1200]}
    except Exception as error:
        outcome = {'id': event['id'], 'status': 'turn_error', 'text': type(error).__name__}
    finally:
        with lock:
            state['results'] = (state['results'] + [outcome])[-8:]
            state['active'] = None
    return deepcopy(outcome)

输出

点击行号设置断点，再点“调试”；也可以直接运行。

代码只读；执行发生在浏览器中，不会发送到博客服务器。

教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。

from copy import deepcopy
from datetime import datetime, timedelta, timezone
from threading import RLock
import re
LOCAL=timezone(timedelta(hours=8))
def cron_fields(expression):
    """有意限定子集：*、单值，分钟额外支持*/N；星期0为周日。"""
    if not isinstance(expression, str) or len(expression) > 80:
        raise ValueError('时间表需要有界字符串。')
    parts = expression.split()
    if len(parts) != 5:
        raise ValueError('需要分钟、小时、日、月、星期五项。')
    limits = ((0, 59), (0, 23), (1, 31), (1, 12), (0, 6))
    fields = []
    for index, (part, (low, high)) in enumerate(zip(parts, limits)):
        if part == '*':
            fields.append(set(range(low, high + 1)))
        elif index == 0 and re.fullmatch(r'\*/[1-9][0-9]?', part) and int(part[2:]) <= 59:
            fields.append(set(range(0, 60, int(part[2:]))))
        elif re.fullmatch(r'[0-9]{1,2}', part) and low <= int(part) <= high:
            fields.append({int(part)})
        else:
            raise ValueError('本课仅支持*、范围内单值和分钟*/1到*/59。')
    return parts, fields

def checked_time(value):
    if not isinstance(value, datetime) or value.tzinfo is None or value.utcoffset() is None:
        raise ValueError('时间必须带UTC偏移。')
    return value

def new_state():
    return {'jobs': {}, 'queue': [], 'active': None, 'results': [], 'created': 0}

def schedule_job(state, lock, expression, prompt, once=False):
    cron_fields(expression)
    if not isinstance(prompt, str) or not 1 <= len(prompt.strip()) <= 600 or type(once) is not bool:
        raise ValueError('提示需要1到600字符，once需要布尔值。')
    with lock:
        if state['created'] >= 4:
            raise ValueError('本次会话最多创建4张时间表。')
        state['created'] += 1
        job_id = 'cron-' + str(state['created'])
        state['jobs'][job_id] = {'cron': expression, 'prompt': prompt.strip(), 'once': once, 'last_slot': None}
    return job_id

def cron_matches(expression, when):
    """固定UTC+08:00解释时间；日与星期同时受限时任一满足即可。"""
    parts, fields = cron_fields(expression)
    local = checked_time(when).astimezone(LOCAL)
    minute, hour, day, month, weekday = fields
    if local.minute not in minute or local.hour not in hour or local.month not in month:
        return False
    day_ok = local.day in day
    week_ok = (local.weekday() + 1) % 7 in weekday
    if parts[2] == '*':
        return week_ok
    if parts[4] == '*':
        return day_ok
    return day_ok or week_ok

def enqueue_due(state, now, lock):
    """检查当前分钟，不补发停机期间的时刻；满队列不提前记已触发。"""
    instant = checked_time(now).astimezone(timezone.utc).replace(second=0, microsecond=0)
    slot = instant.isoformat()
    added = []
    with lock:
        for job_id, job in list(state['jobs'].items()):
            if job['last_slot'] is not None and slot <= job['last_slot']:
                continue
            if not cron_matches(job['cron'], instant) or len(state['queue']) >= 8:
                continue
            event = {'id': job_id + '@' + slot, 'job_id': job_id, 'slot': slot, 'prompt': job['prompt']}
            state['queue'].append(event)
            job['last_slot'] = slot
            if job['once']:
                del state['jobs'][job_id]
            added.append(deepcopy(event))
    return added

def dispatch_next(state, run_turn, lock, busy=False):
    """只在空闲时占用一项；锁外运行，失败记录后不自动重跑。"""
    with lock:
        if busy or state['active'] is not None:
            return {'status': 'busy'}
        if not state['queue']:
            return {'status': 'idle'}
        event = state['queue'].pop(0)
        state['active'] = event['id']
    outcome = {'id': event['id'], 'status': 'interrupted', 'text': ''}
    try:
        response = run_turn(deepcopy(event))
        outcome = {'id': event['id'], 'status': str(response['status'])[:40], 'text': str(response['text'])[:1200]}
    except Exception as error:
        outcome = {'id': event['id'], 'status': 'turn_error', 'text': type(error).__name__}
    finally:
        with lock:
            state['results'] = (state['results'] + [outcome])[-8:]
            state['active'] = None
    return deepcopy(outcome)


scenario = 'ready'
state, lock = new_state(), RLock()
expression = '*/5 * * * *'
when = datetime.fromisoformat('2026-10-19T09:05:23+08:00')
if scenario=='not_due':when=datetime.fromisoformat('2026-10-19T09:04:00+08:00')
if scenario=='day_or':expression='0 9 1 * 1';when=datetime.fromisoformat('2026-10-19T09:00:00+08:00')
if scenario=='sunday':expression='0 9 * * 0';when=datetime.fromisoformat('2026-10-18T09:00:00+08:00')
if scenario=='invalid':expression='*/0 * * * *'
if scenario=='naive':when=when.replace(tzinfo=None)
if 'dispatch_next'!='cron_matches':
    schedule_job(state,lock,expression,'检查main.py，运行仍需本次许可。',once=scenario=='once')
    if 'dispatch_next'=='dispatch_next' and scenario!='empty':enqueue_due(state,when,lock)
    if scenario=='duplicate':enqueue_due(state,when,lock)
    if scenario=='backward':
        enqueue_due(state,when,lock)
        when=datetime.fromisoformat('2026-10-19T09:00:00+08:00')
    if scenario=='full':state['queue']=[{'id':'earlier-'+str(i)} for i in range(8)]
    if scenario=='active':state['active']='another-turn'
received=[]
def run_turn(event):
    received.append(deepcopy(event))
    if scenario=='failed':raise OSError('教学轮次失败')
    return {'status':'call_limit' if scenario=='model_stop' else 'answered','text':'模型已停下；程序是否成功需看执行证据。'}
busy=scenario=='busy'


# 调用入口
try:
    observed=dispatch_next(state, run_turn, lock, busy)
except ValueError as error:
    observed={"error":str(error)}
print(observed)

锁内确认空闲并占用一项。
锁外运行原Agent循环。
记录轮次结果并释放active，失败不自动重跑。

在 run_turn 行停下，active 已经被占用，事件已从队列取出，模型却尚未被调用。“已有轮次占用”会在前面直接返回，不会同时启动第二个消费者。图中待交付队列减少和当前占用亮起，是两个位置的真实变化。此时 outcome 中的 interrupted 是异常退出时的保底值，还没有写入 results，不代表已经发生中断。

正常返回时保存 response.status 与短结论；异常转成 turn_error。finally 将结果追加到最近记录并清空 active，两项仍在同一把锁内完成。即使 KeyboardInterrupt 向外传播，也会记录 interrupted 并释放占用。结果最多保留八条，文字最多 1200 字符。

“模型额度用完”保存 call_limit，不会被画成成功回答。answered 同样只代表得到回答：没有许可时也可以正常解释“未执行”。本课失败不重新入队；如果要重试整轮，需要先确认先前工具有没有产生副作用。

真正等待时间时，哪一段还在运行？

默认 demo 不创建线程，只用注入时钟把同一条路径快速走完。显式加上 --serve-seconds，才启动本进程里的真实时间检查：

bash

python3 examples/s14-cron-scheduler/user_code/practice.py --serve-seconds 1

这个入口每分钟触发，默认仍拒绝程序执行。一个调度线程每 0.2 秒调用 enqueue_due；主线程空闲时调用 dispatch_next。创建当分钟符合就可能触发，一秒实验因此不必等到下一整分钟，但它没有安装操作系统定时任务或常驻服务。

python

# user_code/practice.py 的实际交付关系；完整装配见公开源码
result = dispatch_next(
    state,
    lambda event: run_check(event, workspace, allowed),
    lock,
)
# run_check用事件组成question，再调用既有run_loop(ask, execute, ...)。

run_check 复用实际工具注册与第十课提示组装、第十一课父请求恢复，以及原工具权限和 Hook。时间表可以生产工作，真正的 run_main 仍需本次宿主许可。默认定时轮次没有注册新增时间表的三个工具；要开放这些内存工具，也须明确接入宿主策略，旧入口仍默认拒绝未知权限。

这里用同步 run_main 取得每轮结果，第十三课的后台进程入口仍可独立运行。本课新增的是“何时交付一份新问题”，没有把持久任务图、后台程序、历史压缩和定时队列一并堆进演示。

serve 的时长限制主线程继续取出新事件；已经开始的一轮按自己的模型与程序预算结束。退出时发 stop 信号并 join 调度线程。消息历史始终只由消费者所属的原循环处理；生产线程只修改受锁保护的调度状态。会话表、队列和去重标记均不落盘，进程崩溃也不提供恰好一次交付。保存时间表定义、保存待交付事件、恢复正在进行的轮次是后续需要分别设计的持久边界。

用真实并发验证“模型在工作，时钟还能继续”

bash

python3 examples/s14-cron-scheduler/user_code/verify_practice.py
python3 examples/s14-cron-scheduler/user_code/verify_fixtures.py
.venv/bin/python examples/s14-cron-scheduler/user_code/verify_sdk.py

十二组标准库测试包含一个 Event 门控实验：第一个消费者进入回调后等待信号，主测试线程在它占用 active 时调用 enqueue_due，确认下一分钟事件能够入队；第二个消费者则必须返回 busy。最后释放信号，检查结果保存和 active 清空。这个实验使用真实线程，不靠猜测一段 sleep 能否覆盖竞争窗口。

另一组测试让 main.py 每次运行就在计数文件追加 x。两次定时交付后，允许路径恰好为 xx，拒绝路径没有计数文件。队列满后同分钟再试、跨日、时钟回退、一次性表、取消、轮次异常和中断释放也有独立断言。

上面十八组浏览器输入原样执行函数、日期转换和锁；输入时钟与 run_turn 回调是公开可查看的教学替身，不启动浏览器线程、模型或程序。站内实际采集了 183 个断点快照，用同一个游标驱动代码、图和阶段说明。

SDK 验证使用固定 Anthropic1.13.0 与模拟 HTTP。在程序已执行后的报告请求注入一次 529，允许路径四次 HTTP、拒绝路径三次；允许时计数只有一个 x，请求恢复没有把同一个触发事件重新交付。没有在线模型调用。