程云来 / 杭州
返回文章·Agent 与 Harness
·约 9 分钟

实践｜Task System，把依赖、认领和完成落到文件里

展开三个真实函数，沿断点观察原记录、候选与持久提交，用独立进程验证任务解锁和不自动重跑的边界。

结束一个进程，再来继续下一项

把“检查程序并报告”拆成 check → run → report，先保存依赖，再分别认领与完成。三个 work 命令是三个独立进程：后一个进程依靠任务文件接着做，而不是靠前一个 Python 进程还活着。思想篇解释了为什么完成只意味着下游可以开始。

源码在独立公开仓库 learn-agent-harness。Python3.10+，默认标准库，无需 API Key：

bash
git clone https://github.com/Chengyunlai/learn-agent-harness.git
cd learn-agent-harness
python3 examples/s12-task-system/user_code/practice.py init
python3 examples/s12-task-system/user_code/practice.py show
python3 examples/s12-task-system/user_code/practice.py work --task check
python3 examples/s12-task-system/user_code/practice.py work --task run --decision allow
python3 examples/s12-task-system/user_code/practice.py work --task report

可开始项依次为 check、run、report，最后为空。run 重新读取当前预期，获准执行后取得 42 并核对；report 引用已存的 run 结果，不再启动程序。初始化后的修订号为 3，三次认领和三次完成后为 9。

本例由 init 建好已知任务图，模型响应为固定脚本；真实读写任务文件、真实读取工作区、获准后真实运行程序。它验证任务机制，不证明真实模型已经学会拆解复杂项目。init 只对新任务目录运行；需要重做实验时，用 --store 指定另一空目录，已有 ID 不会被重置。

create_task：新记录先进入候选，不直接改原文件

每项记录保存 id、subject、description、status、owner、blocked_by 和 result。ID 用宿主给出的有界小写标识，在同一图里唯一；本课没有删除或重命名操作。创建时 status 固定为 pending，owner 为空，结果说明也为空。

create_task 重新读盘、拒绝重复 ID，在深复制的 candidate 中加入新项，再校验整份图。依赖必须已经存在且不能重复或指向自身。创建只引用旧项，后来又不能改边，这让新增边保持先后关系；加载文件仍额外检查环，避免手动改坏后继续运行。

真实执行轨迹回放；直接读写Worker临时虚拟任务文件，不访问宿主目录或调用模型。

Pys12/core/runtime.py · create_task
未运行
def create_task(directory, task_id, subject, description, blocked_by):
    """依赖只能引用已存在任务；新项从pending开始，不能覆盖旧ID。"""
    task_id = identifier(task_id)
    board = load_board(directory)
    if task_id in board['tasks']:
        raise ValueError('任务ID已经存在，不覆盖。')
    item = {'id': task_id, 'subject': subject, 'description': description, 'status': 'pending',
            'owner': None, 'blocked_by': deepcopy(blocked_by), 'result': ''}
    candidate = deepcopy(board)
    candidate['tasks'][task_id] = item
    candidate['revision'] += 1
    checked_board(candidate)
    save_board(directory, candidate)
    return deepcopy(item)
输出

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

ID / DESCRIPTIONSTATUS / OWNERSTATEOWNERDEPENDENCIESCANDIDATE COPYSTATUS / OWNEREXISTING IDSBLOCKED BYPERSISTED STATETASKS.JSONATOMIC REPLACESINGLE WRITER任务与依赖校验整份图新增候选原任务图持久文件
代码只读；执行发生在浏览器中，不会发送到博客服务器。
教学准备与调用入口

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

from copy import deepcopy
import json
import os
from pathlib import Path
import re
import tempfile
scenario='new'
_temp=tempfile.TemporaryDirectory(prefix='lesson12-')
directory=Path(_temp.name)

def identifier(value):
    if type(value) is not str or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}',value):
        raise ValueError('标识必须是1到32位小写字母、数字或连字符。')
    return value

def checked_board(board):
    if type(board) is not dict or set(board)!={'version','revision','tasks'} or type(board['version']) is not int or board['version']!=1 or type(board['revision']) is not int or not 0<=board['revision']<=10000:
        raise ValueError('任务文件版本或修订号无效。')
    tasks=board['tasks']
    if type(tasks) is not dict or len(tasks)>20:
        raise ValueError('最多保存20项任务。')
    for key,item in tasks.items():
        identifier(key)
        if type(item) is not dict or set(item)!={'id','subject','description','status','owner','blocked_by','result'} or item['id']!=key:
            raise ValueError('任务字段或ID不一致。')
        for field,limit in [('subject',120),('description',600),('result',1200)]:
            if type(item[field]) is not str or len(item[field])>limit or (field=='subject' and not item[field].strip()):
                raise ValueError('任务文字为空或超过上限。')
        deps=item['blocked_by']
        if type(deps) is not list or len(deps)>20 or any(type(d) is not str or d not in tasks or d==key for d in deps) or len(set(deps))!=len(deps):
            raise ValueError('依赖必须存在、唯一且不能指向自身。')
        if item['status'] not in ('pending','in_progress','completed'):
            raise ValueError('未知任务状态。')
        if item['status']=='pending':
            if item['owner'] is not None or item['result']:
                raise ValueError('待认领任务不能已有负责人或结果。')
        else:
            identifier(item['owner'])
            if bool(item['result'].strip())!=(item['status']=='completed'):
                raise ValueError('只有完成任务需要非空结果说明。')
    for item in tasks.values():
        if item['status']!='pending' and any(tasks[d]['status']!='completed' for d in item['blocked_by']):
            raise ValueError('进行中或完成任务仍有未完成依赖。')
    def visit(key,visiting,done):
        if key in visiting:raise ValueError('依赖存在环。')
        if key in done:return
        visiting.add(key)
        for dep in tasks[key]['blocked_by']:visit(dep,visiting,done)
        visiting.remove(key);done.add(key)
    done=set()
    for key in tasks:visit(key,set(),done)
    return deepcopy(board)

def load_board(directory):
    target=Path(directory)/'tasks.json'
    try:
        with target.open('rb') as stream:raw=stream.read(64001)
    except FileNotFoundError:return {'version':1,'revision':0,'tasks':{}}
    if len(raw)>64000:raise ValueError('任务文件超过64000字节。')
    return checked_board(json.loads(raw.decode('utf-8')))

def save_board(directory,board):
    board=checked_board(board)
    payload=json.dumps(board,ensure_ascii=False,indent=2)
    if len(payload.encode('utf-8'))>64000:raise ValueError('新任务文件超过64000字节。')
    directory=Path(directory);directory.mkdir(parents=True,exist_ok=True);temporary=None
    try:
        with tempfile.NamedTemporaryFile(mode='w',encoding='utf-8',dir=directory,prefix='.tasks-',delete=False) as stream:
            temporary=Path(stream.name);stream.write(payload);stream.flush();os.fsync(stream.fileno())
        os.replace(temporary,directory/'tasks.json')
    finally:
        if temporary is not None:temporary.unlink(missing_ok=True)

def ready_ids(board):
    tasks=checked_board(board)['tasks']
    return sorted(key for key,item in tasks.items() if item['status']=='pending' and all(tasks[d]['status']=='completed' for d in item['blocked_by']))

def create_task(directory, task_id, subject, description, blocked_by):
    """依赖只能引用已存在任务；新项从pending开始，不能覆盖旧ID。"""
    task_id = identifier(task_id)
    board = load_board(directory)
    if task_id in board['tasks']:
        raise ValueError('任务ID已经存在，不覆盖。')
    item = {'id': task_id, 'subject': subject, 'description': description, 'status': 'pending',
            'owner': None, 'blocked_by': deepcopy(blocked_by), 'result': ''}
    candidate = deepcopy(board)
    candidate['tasks'][task_id] = item
    candidate['revision'] += 1
    checked_board(candidate)
    save_board(directory, candidate)
    return deepcopy(item)

def claim_task(directory, task_id, owner):
    """重新读盘，检查状态与全部依赖，再写入负责人；单写者约定。"""
    task_id, owner = identifier(task_id), identifier(owner)
    board = load_board(directory)
    if task_id not in board['tasks']:
        raise ValueError('未知任务。')
    item = board['tasks'][task_id]
    if item['status'] != 'pending':
        raise ValueError('任务已被认领或完成，不能再次认领。')
    blocked = [dep for dep in item['blocked_by'] if board['tasks'][dep]['status'] != 'completed']
    if blocked:
        raise ValueError('依赖尚未完成：' + ','.join(blocked))
    candidate = deepcopy(board)
    candidate['tasks'][task_id].update(status='in_progress', owner=owner)
    candidate['revision'] += 1
    save_board(directory, candidate)
    return deepcopy(candidate['tasks'][task_id])

def complete_task(directory, task_id, owner, result):
    """同一负责人提交结果说明；完成后重新求可开始项，不自动执行下游。"""
    task_id, owner = identifier(task_id), identifier(owner)
    board = load_board(directory)
    if task_id not in board['tasks']:
        raise ValueError('未知任务。')
    item = board['tasks'][task_id]
    if item['status'] != 'in_progress' or item['owner'] != owner:
        raise ValueError('只有进行中任务的负责人可以提交完成。')
    before = ready_ids(board)
    candidate = deepcopy(board)
    candidate['tasks'][task_id].update(status='completed', result=result)
    candidate['revision'] += 1
    checked_board(candidate)
    save_board(directory, candidate)
    unlocked = [key for key in ready_ids(candidate) if key not in before]
    return {'task': deepcopy(candidate['tasks'][task_id]), 'unlocked': unlocked}

create_task(directory,'check','检查文件','读取当前文件，不等于已经运行。',[])
create_task(directory,'run','运行并核对','取得本次许可和执行证据。',['check'])
create_task(directory,'report','报告已存结论','引用run记录，不自动重新运行。',['run'])
task_id='review' if 'create_task'=='create_task' else 'run'
owner='worker-a'
result='本次获准运行，exit_code=0，实际输出42。'
deps=['check']
if scenario not in ('blocked','pending') and 'create_task'!='create_task':
    claim_task(directory,'check',owner)
    complete_task(directory,'check',owner,'已读文件，尚未运行。')
if 'create_task'=='claim_task' and scenario=='taken':claim_task(directory,'run',owner)
if 'create_task'=='complete_task' and scenario!='pending':claim_task(directory,'run',owner)
if scenario=='duplicate':task_id='check'
if scenario=='missing':deps=['missing']
if scenario=='self':deps=['review']
if scenario=='bad_id':task_id='../other'
if scenario=='unknown':task_id='missing'
if scenario=='wrong_owner':owner='worker-b'
if scenario=='empty':result=''
before=(directory/'tasks.json').read_bytes()
_replace=os.replace
def broken_replace(*_):raise OSError('教学替换失败')
if scenario=='replace_error':os.replace=broken_replace


# 调用入口
try:
    observed=create_task(directory,task_id,'复查记录','只更新任务图，不执行程序。',deps)
    print(observed)
except (ValueError,OSError) as error:
    observed={"error":str(error)}
    print(str(error))
    assert (directory/"tasks.json").read_bytes()==before
finally:
    os.replace=_replace
  1. 读取原任务图并拒绝重复ID。
  2. 新项加入候选，校验已有依赖、状态与边。
  3. 完整校验后持久保存，创建不执行。

在加入候选的赋值行停住，可以看到原 board 与 candidate 尚未包含新项；单步后只在 candidate 中出现。再到 save_board 调用行，图会明确“下一步提交”，不会因为内存已经变化就把磁盘画成成功。

save_board 是三个写操作共用的存储边界：再次校验候选，序列化后检查 64000 字节上限，写同目录临时文件，flush/fsync 后 os.replace。最多保存 20 项任务；标题 120、描述 600、结果 1200 字符。替换失败时旧文件保持，临时文件清理；损坏原文件则拒绝操作，不把错误解释成空任务图。

claim_task：先确认能做，再把负责人写进去

函数输入任务 ID 和宿主负责人，先确认记录仍为 pending，然后逐个检查 blocked_by。只要有一项前置尚未完成，整个认领失败，原文件保持。空依赖会得到空的 blocked 列表，不需要额外放行标记。

通过后复制整份图，把目标候选改成 in_progress 并绑定 owner，增加修订号，再提交。修订号只表示本例发生过一次成功变更，不承担并发比较或锁的功能。返回记录是副本，调用者改返回值不会改磁盘。

真实执行轨迹回放；直接读写Worker临时虚拟任务文件，不访问宿主目录或调用模型。

Pys12/core/runtime.py · claim_task
未运行
def claim_task(directory, task_id, owner):
    """重新读盘，检查状态与全部依赖，再写入负责人；单写者约定。"""
    task_id, owner = identifier(task_id), identifier(owner)
    board = load_board(directory)
    if task_id not in board['tasks']:
        raise ValueError('未知任务。')
    item = board['tasks'][task_id]
    if item['status'] != 'pending':
        raise ValueError('任务已被认领或完成，不能再次认领。')
    blocked = [dep for dep in item['blocked_by'] if board['tasks'][dep]['status'] != 'completed']
    if blocked:
        raise ValueError('依赖尚未完成：' + ','.join(blocked))
    candidate = deepcopy(board)
    candidate['tasks'][task_id].update(status='in_progress', owner=owner)
    candidate['revision'] += 1
    save_board(directory, candidate)
    return deepcopy(candidate['tasks'][task_id])
输出

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

EXISTING IDSBLOCKED BYPERSISTED STATESTATEOWNERDEPENDENCIESCANDIDATE COPYSTATUS / OWNERCURRENT RECORDSTATUS / OWNERTASKS.JSONATOMIC REPLACESINGLE WRITER依赖状态认领检查认领候选当前记录持久文件
代码只读；执行发生在浏览器中，不会发送到博客服务器。
教学准备与调用入口

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

from copy import deepcopy
import json
import os
from pathlib import Path
import re
import tempfile
scenario='ready'
_temp=tempfile.TemporaryDirectory(prefix='lesson12-')
directory=Path(_temp.name)

def identifier(value):
    if type(value) is not str or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}',value):
        raise ValueError('标识必须是1到32位小写字母、数字或连字符。')
    return value

def checked_board(board):
    if type(board) is not dict or set(board)!={'version','revision','tasks'} or type(board['version']) is not int or board['version']!=1 or type(board['revision']) is not int or not 0<=board['revision']<=10000:
        raise ValueError('任务文件版本或修订号无效。')
    tasks=board['tasks']
    if type(tasks) is not dict or len(tasks)>20:
        raise ValueError('最多保存20项任务。')
    for key,item in tasks.items():
        identifier(key)
        if type(item) is not dict or set(item)!={'id','subject','description','status','owner','blocked_by','result'} or item['id']!=key:
            raise ValueError('任务字段或ID不一致。')
        for field,limit in [('subject',120),('description',600),('result',1200)]:
            if type(item[field]) is not str or len(item[field])>limit or (field=='subject' and not item[field].strip()):
                raise ValueError('任务文字为空或超过上限。')
        deps=item['blocked_by']
        if type(deps) is not list or len(deps)>20 or any(type(d) is not str or d not in tasks or d==key for d in deps) or len(set(deps))!=len(deps):
            raise ValueError('依赖必须存在、唯一且不能指向自身。')
        if item['status'] not in ('pending','in_progress','completed'):
            raise ValueError('未知任务状态。')
        if item['status']=='pending':
            if item['owner'] is not None or item['result']:
                raise ValueError('待认领任务不能已有负责人或结果。')
        else:
            identifier(item['owner'])
            if bool(item['result'].strip())!=(item['status']=='completed'):
                raise ValueError('只有完成任务需要非空结果说明。')
    for item in tasks.values():
        if item['status']!='pending' and any(tasks[d]['status']!='completed' for d in item['blocked_by']):
            raise ValueError('进行中或完成任务仍有未完成依赖。')
    def visit(key,visiting,done):
        if key in visiting:raise ValueError('依赖存在环。')
        if key in done:return
        visiting.add(key)
        for dep in tasks[key]['blocked_by']:visit(dep,visiting,done)
        visiting.remove(key);done.add(key)
    done=set()
    for key in tasks:visit(key,set(),done)
    return deepcopy(board)

def load_board(directory):
    target=Path(directory)/'tasks.json'
    try:
        with target.open('rb') as stream:raw=stream.read(64001)
    except FileNotFoundError:return {'version':1,'revision':0,'tasks':{}}
    if len(raw)>64000:raise ValueError('任务文件超过64000字节。')
    return checked_board(json.loads(raw.decode('utf-8')))

def save_board(directory,board):
    board=checked_board(board)
    payload=json.dumps(board,ensure_ascii=False,indent=2)
    if len(payload.encode('utf-8'))>64000:raise ValueError('新任务文件超过64000字节。')
    directory=Path(directory);directory.mkdir(parents=True,exist_ok=True);temporary=None
    try:
        with tempfile.NamedTemporaryFile(mode='w',encoding='utf-8',dir=directory,prefix='.tasks-',delete=False) as stream:
            temporary=Path(stream.name);stream.write(payload);stream.flush();os.fsync(stream.fileno())
        os.replace(temporary,directory/'tasks.json')
    finally:
        if temporary is not None:temporary.unlink(missing_ok=True)

def ready_ids(board):
    tasks=checked_board(board)['tasks']
    return sorted(key for key,item in tasks.items() if item['status']=='pending' and all(tasks[d]['status']=='completed' for d in item['blocked_by']))

def create_task(directory, task_id, subject, description, blocked_by):
    """依赖只能引用已存在任务；新项从pending开始，不能覆盖旧ID。"""
    task_id = identifier(task_id)
    board = load_board(directory)
    if task_id in board['tasks']:
        raise ValueError('任务ID已经存在，不覆盖。')
    item = {'id': task_id, 'subject': subject, 'description': description, 'status': 'pending',
            'owner': None, 'blocked_by': deepcopy(blocked_by), 'result': ''}
    candidate = deepcopy(board)
    candidate['tasks'][task_id] = item
    candidate['revision'] += 1
    checked_board(candidate)
    save_board(directory, candidate)
    return deepcopy(item)

def claim_task(directory, task_id, owner):
    """重新读盘，检查状态与全部依赖，再写入负责人；单写者约定。"""
    task_id, owner = identifier(task_id), identifier(owner)
    board = load_board(directory)
    if task_id not in board['tasks']:
        raise ValueError('未知任务。')
    item = board['tasks'][task_id]
    if item['status'] != 'pending':
        raise ValueError('任务已被认领或完成，不能再次认领。')
    blocked = [dep for dep in item['blocked_by'] if board['tasks'][dep]['status'] != 'completed']
    if blocked:
        raise ValueError('依赖尚未完成：' + ','.join(blocked))
    candidate = deepcopy(board)
    candidate['tasks'][task_id].update(status='in_progress', owner=owner)
    candidate['revision'] += 1
    save_board(directory, candidate)
    return deepcopy(candidate['tasks'][task_id])

def complete_task(directory, task_id, owner, result):
    """同一负责人提交结果说明；完成后重新求可开始项，不自动执行下游。"""
    task_id, owner = identifier(task_id), identifier(owner)
    board = load_board(directory)
    if task_id not in board['tasks']:
        raise ValueError('未知任务。')
    item = board['tasks'][task_id]
    if item['status'] != 'in_progress' or item['owner'] != owner:
        raise ValueError('只有进行中任务的负责人可以提交完成。')
    before = ready_ids(board)
    candidate = deepcopy(board)
    candidate['tasks'][task_id].update(status='completed', result=result)
    candidate['revision'] += 1
    checked_board(candidate)
    save_board(directory, candidate)
    unlocked = [key for key in ready_ids(candidate) if key not in before]
    return {'task': deepcopy(candidate['tasks'][task_id]), 'unlocked': unlocked}

create_task(directory,'check','检查文件','读取当前文件，不等于已经运行。',[])
create_task(directory,'run','运行并核对','取得本次许可和执行证据。',['check'])
create_task(directory,'report','报告已存结论','引用run记录，不自动重新运行。',['run'])
task_id='review' if 'claim_task'=='create_task' else 'run'
owner='worker-a'
result='本次获准运行，exit_code=0，实际输出42。'
deps=['check']
if scenario not in ('blocked','pending') and 'claim_task'!='create_task':
    claim_task(directory,'check',owner)
    complete_task(directory,'check',owner,'已读文件，尚未运行。')
if 'claim_task'=='claim_task' and scenario=='taken':claim_task(directory,'run',owner)
if 'claim_task'=='complete_task' and scenario!='pending':claim_task(directory,'run',owner)
if scenario=='duplicate':task_id='check'
if scenario=='missing':deps=['missing']
if scenario=='self':deps=['review']
if scenario=='bad_id':task_id='../other'
if scenario=='unknown':task_id='missing'
if scenario=='wrong_owner':owner='worker-b'
if scenario=='empty':result=''
before=(directory/'tasks.json').read_bytes()
_replace=os.replace
def broken_replace(*_):raise OSError('教学替换失败')
if scenario=='replace_error':os.replace=broken_replace


# 调用入口
try:
    observed=claim_task(directory,task_id,owner)
    print(observed)
except (ValueError,OSError) as error:
    observed={"error":str(error)}
    print(str(error))
    assert (directory/"tasks.json").read_bytes()==before
finally:
    os.replace=_replace
  1. 重新读盘，确认pending及全部依赖完成。
  2. 将负责人和in_progress写入候选。
  3. 提交后才认领成功，不承担并发锁。

“依赖未完成”会显示真实 blocked 列表，且没有持久提交；“已经被认领”在状态检查处就失败。进入候选更新以后，左侧当前记录仍为 pending，右侧候选才变为 in_progress。两个位置表达的是原记录与待提交记录，不能合并成一次提前发生的状态变化。

模型工具的 task_claim 参数只有 task_id；owner 与目录在注册时由宿主绑定。它不能通过在参数里加 owner=worker-b 来替换负责人，也不能指定另一个文件路径。这个约定仍需要单写者环境，不能据此宣称并发认领已受锁保护。

complete_task：完成先保存，解锁项随后重新计算

完成要求当前状态为 in_progress，并且提交者与原 owner 相同。结果说明不能为空，但字段校验只能保证格式，不能证明文字一定属实。固定流程在实际核对成功以后才调用完成；直接调用函数的人仍要为结果说明提供证据。

函数先求提交前的 ready_ids，再准备 completed 候选并保存。保存返回以后，重新求可开始项，与原集合比较，只返回本次新增的 unlocked。原本就能开始的独立任务，不会被重复算成本次解锁成果。

真实执行轨迹回放；直接读写Worker临时虚拟任务文件，不访问宿主目录或调用模型。

Pys12/core/runtime.py · complete_task
未运行
def complete_task(directory, task_id, owner, result):
    """同一负责人提交结果说明；完成后重新求可开始项，不自动执行下游。"""
    task_id, owner = identifier(task_id), identifier(owner)
    board = load_board(directory)
    if task_id not in board['tasks']:
        raise ValueError('未知任务。')
    item = board['tasks'][task_id]
    if item['status'] != 'in_progress' or item['owner'] != owner:
        raise ValueError('只有进行中任务的负责人可以提交完成。')
    before = ready_ids(board)
    candidate = deepcopy(board)
    candidate['tasks'][task_id].update(status='completed', result=result)
    candidate['revision'] += 1
    checked_board(candidate)
    save_board(directory, candidate)
    unlocked = [key for key in ready_ids(candidate) if key not in before]
    return {'task': deepcopy(candidate['tasks'][task_id]), 'unlocked': unlocked}
输出

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

CURRENT RECORDSTATUS / OWNERSTATEOWNERDEPENDENCIESCANDIDATE COPYSTATUS / OWNERNEW READY IDSNO AUTO EXECUTIONTASKS.JSONATOMIC REPLACESINGLE WRITER当前记录状态与负责人完成候选新可开始项持久文件
代码只读；执行发生在浏览器中，不会发送到博客服务器。
教学准备与调用入口

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

from copy import deepcopy
import json
import os
from pathlib import Path
import re
import tempfile
scenario='done'
_temp=tempfile.TemporaryDirectory(prefix='lesson12-')
directory=Path(_temp.name)

def identifier(value):
    if type(value) is not str or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}',value):
        raise ValueError('标识必须是1到32位小写字母、数字或连字符。')
    return value

def checked_board(board):
    if type(board) is not dict or set(board)!={'version','revision','tasks'} or type(board['version']) is not int or board['version']!=1 or type(board['revision']) is not int or not 0<=board['revision']<=10000:
        raise ValueError('任务文件版本或修订号无效。')
    tasks=board['tasks']
    if type(tasks) is not dict or len(tasks)>20:
        raise ValueError('最多保存20项任务。')
    for key,item in tasks.items():
        identifier(key)
        if type(item) is not dict or set(item)!={'id','subject','description','status','owner','blocked_by','result'} or item['id']!=key:
            raise ValueError('任务字段或ID不一致。')
        for field,limit in [('subject',120),('description',600),('result',1200)]:
            if type(item[field]) is not str or len(item[field])>limit or (field=='subject' and not item[field].strip()):
                raise ValueError('任务文字为空或超过上限。')
        deps=item['blocked_by']
        if type(deps) is not list or len(deps)>20 or any(type(d) is not str or d not in tasks or d==key for d in deps) or len(set(deps))!=len(deps):
            raise ValueError('依赖必须存在、唯一且不能指向自身。')
        if item['status'] not in ('pending','in_progress','completed'):
            raise ValueError('未知任务状态。')
        if item['status']=='pending':
            if item['owner'] is not None or item['result']:
                raise ValueError('待认领任务不能已有负责人或结果。')
        else:
            identifier(item['owner'])
            if bool(item['result'].strip())!=(item['status']=='completed'):
                raise ValueError('只有完成任务需要非空结果说明。')
    for item in tasks.values():
        if item['status']!='pending' and any(tasks[d]['status']!='completed' for d in item['blocked_by']):
            raise ValueError('进行中或完成任务仍有未完成依赖。')
    def visit(key,visiting,done):
        if key in visiting:raise ValueError('依赖存在环。')
        if key in done:return
        visiting.add(key)
        for dep in tasks[key]['blocked_by']:visit(dep,visiting,done)
        visiting.remove(key);done.add(key)
    done=set()
    for key in tasks:visit(key,set(),done)
    return deepcopy(board)

def load_board(directory):
    target=Path(directory)/'tasks.json'
    try:
        with target.open('rb') as stream:raw=stream.read(64001)
    except FileNotFoundError:return {'version':1,'revision':0,'tasks':{}}
    if len(raw)>64000:raise ValueError('任务文件超过64000字节。')
    return checked_board(json.loads(raw.decode('utf-8')))

def save_board(directory,board):
    board=checked_board(board)
    payload=json.dumps(board,ensure_ascii=False,indent=2)
    if len(payload.encode('utf-8'))>64000:raise ValueError('新任务文件超过64000字节。')
    directory=Path(directory);directory.mkdir(parents=True,exist_ok=True);temporary=None
    try:
        with tempfile.NamedTemporaryFile(mode='w',encoding='utf-8',dir=directory,prefix='.tasks-',delete=False) as stream:
            temporary=Path(stream.name);stream.write(payload);stream.flush();os.fsync(stream.fileno())
        os.replace(temporary,directory/'tasks.json')
    finally:
        if temporary is not None:temporary.unlink(missing_ok=True)

def ready_ids(board):
    tasks=checked_board(board)['tasks']
    return sorted(key for key,item in tasks.items() if item['status']=='pending' and all(tasks[d]['status']=='completed' for d in item['blocked_by']))

def create_task(directory, task_id, subject, description, blocked_by):
    """依赖只能引用已存在任务；新项从pending开始，不能覆盖旧ID。"""
    task_id = identifier(task_id)
    board = load_board(directory)
    if task_id in board['tasks']:
        raise ValueError('任务ID已经存在，不覆盖。')
    item = {'id': task_id, 'subject': subject, 'description': description, 'status': 'pending',
            'owner': None, 'blocked_by': deepcopy(blocked_by), 'result': ''}
    candidate = deepcopy(board)
    candidate['tasks'][task_id] = item
    candidate['revision'] += 1
    checked_board(candidate)
    save_board(directory, candidate)
    return deepcopy(item)

def claim_task(directory, task_id, owner):
    """重新读盘，检查状态与全部依赖，再写入负责人；单写者约定。"""
    task_id, owner = identifier(task_id), identifier(owner)
    board = load_board(directory)
    if task_id not in board['tasks']:
        raise ValueError('未知任务。')
    item = board['tasks'][task_id]
    if item['status'] != 'pending':
        raise ValueError('任务已被认领或完成，不能再次认领。')
    blocked = [dep for dep in item['blocked_by'] if board['tasks'][dep]['status'] != 'completed']
    if blocked:
        raise ValueError('依赖尚未完成：' + ','.join(blocked))
    candidate = deepcopy(board)
    candidate['tasks'][task_id].update(status='in_progress', owner=owner)
    candidate['revision'] += 1
    save_board(directory, candidate)
    return deepcopy(candidate['tasks'][task_id])

def complete_task(directory, task_id, owner, result):
    """同一负责人提交结果说明；完成后重新求可开始项，不自动执行下游。"""
    task_id, owner = identifier(task_id), identifier(owner)
    board = load_board(directory)
    if task_id not in board['tasks']:
        raise ValueError('未知任务。')
    item = board['tasks'][task_id]
    if item['status'] != 'in_progress' or item['owner'] != owner:
        raise ValueError('只有进行中任务的负责人可以提交完成。')
    before = ready_ids(board)
    candidate = deepcopy(board)
    candidate['tasks'][task_id].update(status='completed', result=result)
    candidate['revision'] += 1
    checked_board(candidate)
    save_board(directory, candidate)
    unlocked = [key for key in ready_ids(candidate) if key not in before]
    return {'task': deepcopy(candidate['tasks'][task_id]), 'unlocked': unlocked}

create_task(directory,'check','检查文件','读取当前文件，不等于已经运行。',[])
create_task(directory,'run','运行并核对','取得本次许可和执行证据。',['check'])
create_task(directory,'report','报告已存结论','引用run记录，不自动重新运行。',['run'])
task_id='review' if 'complete_task'=='create_task' else 'run'
owner='worker-a'
result='本次获准运行，exit_code=0，实际输出42。'
deps=['check']
if scenario not in ('blocked','pending') and 'complete_task'!='create_task':
    claim_task(directory,'check',owner)
    complete_task(directory,'check',owner,'已读文件，尚未运行。')
if 'complete_task'=='claim_task' and scenario=='taken':claim_task(directory,'run',owner)
if 'complete_task'=='complete_task' and scenario!='pending':claim_task(directory,'run',owner)
if scenario=='duplicate':task_id='check'
if scenario=='missing':deps=['missing']
if scenario=='self':deps=['review']
if scenario=='bad_id':task_id='../other'
if scenario=='unknown':task_id='missing'
if scenario=='wrong_owner':owner='worker-b'
if scenario=='empty':result=''
before=(directory/'tasks.json').read_bytes()
_replace=os.replace
def broken_replace(*_):raise OSError('教学替换失败')
if scenario=='replace_error':os.replace=broken_replace


# 调用入口
try:
    observed=complete_task(directory,task_id,owner,result)
    print(observed)
except (ValueError,OSError) as error:
    observed={"error":str(error)}
    print(str(error))
    assert (directory/"tasks.json").read_bytes()==before
finally:
    os.replace=_replace
  1. 确认进行中状态及相同负责人。
  2. 保存完成与结果说明。
  3. 比较提交前后ready，只返回新解锁项，不自动执行。

在保存调用行停下，磁盘仍是进行中记录；下一步保存返回，才进入解锁计算。完成 run 的正常输入只解锁 report，report 的状态仍是 pending。函数没有去认领它，也没有启动它的任何动作。

负责人不符、尚未认领、缺少结果说明和替换文件失败，都不会写出完成记录。“替换文件失败”的教学输入专门注入了一个公开可查看的接口故障，不能把候选中的 completed 当作持久文件已经完成。

五个任务工具，怎样接进已有的循环？

三个写函数之外，还有 task_list 与 task_get，分别读取概要与完整描述/结果。工具名刻意区分第六课的 task：task 是一次子调查调用，task_claim 等操作的是持久工作记录。本例仍只有原来的 run_loop 负责保存模型请求、关联工具结果和决定继续或停止。

装配时把目录与负责人绑定到任务工具，再并入原注册表；第十课沿真实注册生成 system 与 tools，因此这次新增的能力也进入同一份请求。任务写操作走既有参数分发和 Hook，原 run_main 继续经过第三课许可。

python
# s12/core/runtime.py 的任务工具绑定形状（完整注册见公开源码）
"task_claim": {
    "description": "检查pending及依赖后认领；负责人由宿主绑定。",
    "parameters": {"task_id": str},
    "run": lambda task_id: text(claim_task(directory, task_id, owner)),
}

# s12/user_code/practice.py 的装配
task_tools = build_task_tools(directory, owner)
tools.update(task_tools)
# ask仍复用s11请求恢复；完整响应返回给原run_loop，再分发工具。

当前工具参数沿用既有的必填字符串契约，因此 task_create 的 dependencies 是 JSON 数组文本，例如 '["check"]'，适配器解析以后交给 create_task 的 blocked_by 列表。函数正文和工具参数形状是两个边界，不能把字符串直接当作依赖列表使用。

第八课的历史整理、第十课的提示组装、第十一课的父请求恢复继续复用；原清单、技能和子任务注册也保留。当前固定脚本只调用这次需要的工具。SDK 验证在完成提交后的模型请求注入 529，重发只发生在请求层，任务不会重复完成，程序也不会重跑。

拒绝、重启和并发，分别验证到哪里？

从新的任务目录初始化，先完成 check，再把 run 的许可设为 deny。模型可以正常回答“未执行”，旧循环状态甚至是 answered，但 run 仍为 in_progress，report 仍被阻塞。CLI 会根据持久任务是否完成返回退出码 1，不能把“拿到了回答”当作“任务已经完成”。

bash
python3 examples/s12-task-system/user_code/practice.py work --task run --decision deny
python3 examples/s12-task-system/user_code/practice.py show
python3 examples/s12-task-system/user_code/verify_practice.py
python3 examples/s12-task-system/user_code/verify_fixtures.py

再次启动 work --task run 不会自动重跑，因为认领检查发现任务已经进行中。若上次已执行但未提交，应先查实际产物与记录，再通过 complete 命令明确提交说明；本课没有自动回退、抢占或恰好一次执行机制。完整输出与预期不一致时也保留进行中，等待检查。

标准库测试会将真实程序改成每次运行就在计数文件追加 x，三个独立 work 结束后必须恰好只有一个 x。14 组测试还覆盖损坏文件、依赖环、原子替换失败和拒绝不执行；15 组浏览器绑定直接在 Worker 临时虚拟文件中执行原样函数与同源 helper，替换故障才使用明确注入，不访问宿主目录。

SDK 使用固定 Anthropic1.13.0 与模拟 HTTP，允许加过载恢复路径为 15 次请求，拒绝与阻塞路径为 11 次；任务文件和进程真实执行，没有在线模型调用。本课不提供 --live，先把可验证的任务边界跑清楚。

同一任务目录必须串行使用。临时文件加原子替换让单次提交完整，却没有锁住“读取、检查、写入”这一整段；需要多个工作者同时认领时，应另加锁或数据库事务。任务状态告诉你下一件事是否可开始，证据与权限决定这件事究竟能否、是否已经完成。