不指定下一项，先看两位队友会拿走什么
宿主只登记三项固定工作：source 读 main.py，expected 读 expected.txt，verify 依赖前两项，负责运行并核对本次输出。reader、reviewer 都可以处理这些步骤，各自拥有一条工作线程和自己的上下文。
我们不逐项指定负责人。两人进入空闲检查，自己发现并认领可做项，再调用原来的 Agent 循环。思想篇解释了为什么扫描候选不能直接当成认领成功。
源码在独立公开仓库第十七课，Python 3.10+，默认只需标准库：
bash
git clone https://github.com/Chengyunlai/learn-agent-harness.git
cd learn-agent-harness
python3 examples/s17-autonomous-agents/user_code/practice.py --decision allow
python3 examples/s17-autonomous-agents/user_code/practice.py模型为固定替身，双线程、JSON 看板、JSONL 邮箱和获准后的 Python 进程真实执行；没有在线模型调用。允许路径三项 completed，输出 42 与预期一致；默认拒绝运行时，前两项仍完成，verify 保留 in_progress 和负责人。
认领顺序取决于并发时序。每次实际工作有一次工具请求和一次回答，三项加上 Lead 的一次汇总，正常路径共七次逻辑模型请求。空闲检查不会额外询问模型。
scan_ready：这份快照里，哪些项可以开始？
输入是完整看板快照，输出是按 ID 排序的候选列表。checked_board 沿用第十二课的验证：任务字段、状态、负责人和依赖必须一致，缺失依赖或环不能被当成“暂时没有任务”。
函数逐项检查 pending、owner 为空，以及全部依赖 completed。检查只作用于快照，没有写负责人或修改文件。
真实执行轨迹回放；原样函数、锁与Worker虚拟看板/邮箱；等待回调公开注入，不启动线程、模型或程序。
Pys17/core/runtime.py · scan_ready
未运行
def scan_ready(board):
"""只返回快照中的可认领ID，不写负责人，也不运行任务。"""
tasks = checked_board(board)["tasks"]
ready = []
for task_id in sorted(tasks):
task = tasks[task_id]
if task["status"] != "pending" or task["owner"] is not None:
continue
if all(tasks[dep]["status"] == "completed" for dep in task["blocked_by"]):
ready.append(task_id)
return ready输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
from pathlib import Path
from threading import RLock
import json
import re
import os
import tempfile
PREFIX="[team-protocol] "
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}
def member_name(name, members):
if not isinstance(name, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,23}', name) or name not in members:
raise ValueError('只能使用已登记的成员名。')
return name
def create_team():
return {name: {'role': role, 'history': [], 'active': False, 'status': 'idle', 'turns': 0}
for name, role in [('lead', '汇总只读证据'), ('reader', '读取源码并保留调查上下文'),
('reviewer', '读取预期并通知reader')]}
def load_inbox(directory, recipient, members):
member_name(recipient, members)
target = Path(directory) / (recipient + '.jsonl')
if target.is_symlink():
raise ValueError('邮箱不能是符号链接。')
if not target.exists():
return []
with target.open('rb') as stream:
data = stream.read(32001)
if len(data) > 32000:
raise ValueError('邮箱文件超过限制。')
items = [json.loads(line) for line in data.decode('utf-8').splitlines()]
if len(items) > 8:
raise ValueError('单个邮箱最多8条消息。')
for index, item in enumerate(items, 1):
if type(item) is not dict or set(item) != {'id', 'sender', 'recipient', 'body'}:
raise ValueError('邮箱记录字段不符。')
if item['id'] != recipient + '-' + str(index) or item['recipient'] != recipient:
raise ValueError('邮箱序号或收件人不符。')
member_name(item['sender'], members)
if type(item['body']) is not str or not 1 <= len(item['body']) <= 600:
raise ValueError('消息正文长度不符。')
return items
def save_inbox(directory, recipient, items):
root = Path(directory); root.mkdir(parents=True, exist_ok=True)
payload = ''.join(json.dumps(item, ensure_ascii=False) + '\n' for item in items)
if len(payload.encode('utf-8')) > 32000:
raise ValueError('邮箱文件超过限制。')
temporary = None
try:
with tempfile.NamedTemporaryFile('w', encoding='utf-8', dir=root, delete=False) as stream:
temporary = stream.name
stream.write(payload); stream.flush(); os.fsync(stream.fileno())
os.replace(temporary, root / (recipient + '.jsonl'))
finally:
if temporary is not None and Path(temporary).exists():
Path(temporary).unlink()
def send_message(directory, members, sender, recipient, body, lock):
"""先保存到对方邮箱；queued不代表接收方模型已经读取。"""
member_name(sender, members)
member_name(recipient, members)
if not isinstance(body, str) or not 1 <= len(body.strip()) <= 600:
raise ValueError('正文需要1到600字符。')
with lock:
items = load_inbox(directory, recipient, members)
if len(items) >= 8:
raise ValueError('邮箱已满，不覆盖旧消息。')
message = {'id': recipient + '-' + str(len(items) + 1), 'sender': sender,
'recipient': recipient, 'body': body.strip()}
save_inbox(directory, recipient, items + [message])
return {'id': message['id'], 'status': 'queued'}
def encode_protocol(data):
return PREFIX + json.dumps(data, ensure_ascii=False, sort_keys=True)
def read_protocol(message):
if type(message) is not dict or not all(
type(message.get(k)) is str for k in ("sender", "recipient", "body")
):
raise ValueError("消息信封不符。")
if not message["body"].startswith(PREFIX):
return None
data = json.loads(message["body"][len(PREFIX) :])
if (
type(data) is not dict
or type(data.get("type")) is not str
or type(data.get("request_id")) is not str
):
raise ValueError("协议类型或关联ID不符。")
return data
def open_request(state, team, directory, sender, target, kind, payload, lock):
"""邮箱保存成功才登记pending；角色和同类未决请求先验证。"""
member_name(sender, team)
member_name(target, team)
if (
kind not in ("plan", "shutdown")
or (kind == "plan" and not (sender != "lead" and target == "lead"))
or (kind == "shutdown" and not (sender == "lead" and target != "lead"))
):
raise ValueError("协议方向不符。")
if type(payload) is not str or not 1 <= len(payload.strip()) <= 180:
raise ValueError("请求正文需要1到180字符。")
with lock:
if len(state) >= 4 or any(
r["status"] == "pending"
and (r["kind"], r["sender"], r["target"]) == (kind, sender, target)
for r in state.values()
):
raise ValueError("请求额度用完或同类请求仍未决。")
subject = sender if kind == "plan" else target
if team[subject]["status"] == "closed":
raise ValueError("成员已经关闭。")
request_id = "req-" + str(len(state) + 1)
request = {
"type": kind + "_request",
"request_id": request_id,
"payload": payload.strip(),
}
receipt = send_message(
directory, team, sender, target, encode_protocol(request), lock
)
state[request_id] = {
"kind": kind,
"sender": sender,
"target": target,
"payload": payload.strip(),
"status": "pending",
"reply": None,
"receipt": receipt["id"],
}
return {"request_id": request_id, "status": "pending"}
def reply_request(state, team, directory, request_id, actor, approve, feedback, lock):
"""只有目标角色能回复；忙碌成员不能确认关闭，重发不重复存信。"""
member_name(actor, team)
if type(approve) is not bool or type(feedback) is not str or len(feedback) > 120:
raise ValueError("回复需要布尔决定与有界说明。")
with lock:
record = state.get(request_id)
if record is None or record["target"] != actor:
raise ValueError("没有此请求或回复身份不符。")
response = {
"type": record["kind"] + "_response",
"request_id": request_id,
"approve": approve,
"feedback": feedback,
}
if record["reply"] is not None:
if record["reply"] != response:
raise ValueError("同一请求不能改写决定。")
return {"request_id": request_id, "status": "already_sent"}
if record["status"] != "pending":
raise ValueError("请求不再等待回复。")
if record["kind"] == "shutdown" and approve and team[actor]["active"]:
return {"request_id": request_id, "status": "deferred"}
receipt = send_message(
directory, team, actor, record["sender"], encode_protocol(response), lock
)
record["reply"] = response
if record["kind"] == "shutdown" and approve:
team[actor]["status"] = "closed"
return {"request_id": request_id, "status": "queued", "receipt": receipt["id"]}
def shutdown_at_boundary(state, team, mail, name, lock):
"""仅处理发给自己的真实关闭请求；普通消息不唤起空转模型。"""
messages = load_inbox(mail, name, team)
for message in messages:
try:
data = read_protocol(message)
except (ValueError, TypeError):
continue
if data is None or data.get("type") != "shutdown_request":
continue
request_id = data["request_id"]
record = state.get(request_id)
if (
record is None
or record["kind"] != "shutdown"
or record["target"] != name
or record["sender"] != message["sender"]
or record["payload"] != data.get("payload")
or record["status"] != "pending"
):
continue
receipt = reply_request(
state, team, mail, request_id, name, True, "空闲边界确认关闭。", lock
)
return {
"status": "shutdown",
"request_id": request_id,
"reply_status": receipt["status"],
}
return None
def scan_ready(board):
"""只返回快照中的可认领ID，不写负责人，也不运行任务。"""
tasks = checked_board(board)["tasks"]
ready = []
for task_id in sorted(tasks):
task = tasks[task_id]
if task["status"] != "pending" or task["owner"] is not None:
continue
if all(tasks[dep]["status"] == "completed" for dep in task["blocked_by"]):
ready.append(task_id)
return ready
def claim_next(directory, team, name, lock):
"""同一把锁内重新读盘、扫描、认领；成功保存才交出任务。"""
member_name(name, team)
with lock:
member = team[name]
if (
name == "lead"
or member["active"]
or member["status"] != "idle"
or member["turns"] >= 2
):
return {"status": "unavailable"}
board = load_board(directory)
if any(
t["owner"] == name and t["status"] == "in_progress"
for t in board["tasks"].values()
):
return {"status": "busy"}
ready = scan_ready(board)
if not ready:
return {"status": "idle"}
task = claim_task(directory, ready[0], name)
return {"status": "claimed", "task": task}
def idle_poll(directory, team, state, mail, name, lock, wait, checks=3):
"""只做有界宿主检查；关闭优先，没有工作就等待，不调用模型。"""
member_name(name, team)
if type(checks) is not int or not 1 <= checks <= 12:
raise ValueError("检查次数需要1到12。")
for index in range(checks):
with lock:
member = team[name]
if name == "lead" or member["active"] or member["status"] != "idle":
return {"status": "unavailable"}
control = shutdown_at_boundary(state, team, mail, name, lock)
if control is not None:
return control
outcome = claim_next(directory, team, name, lock)
if outcome["status"] != "idle":
return outcome
if index + 1 < checks:
if wait(0.05):
return {"status": "interrupted"}
return {"status": "idle_timeout", "checks": checks}
scenario='ready'
_temp=tempfile.TemporaryDirectory(prefix='s17-auto-');root=Path(_temp.name);directory=root/'board';mail=root/'mail'
team,state,lock=create_team(),{},RLock();name='reader';checks=3;waits=[]
def make_jobs():
create_task(directory,'source','读取源码','固定main.py',[])
create_task(directory,'expected','读取预期','固定expected.txt',[])
create_task(directory,'verify','核对输出','固定程序',['source','expected'])
if not ('scan_ready'=='scan_ready' and scenario=='empty') and not ('scan_ready'=='idle_poll' and scenario in ('timeout','wakeup','interrupted')):make_jobs()
if 'scan_ready'=='scan_ready':
if scenario in ('blocked','unlocked','owned'):
claim_task(directory,'source','reviewer')
if scenario!='owned':claim_task(directory,'expected','reviewer')
if scenario=='unlocked':
complete_task(directory,'source','reviewer','print(6*7)')
complete_task(directory,'expected','reviewer','42')
board=load_board(directory);before=deepcopy(board)
if scenario=='invalid':board['tasks']['verify']['blocked_by']=['missing']
if 'scan_ready'=='claim_next':
if scenario=='stale':claim_task(directory,'expected','reviewer')
if scenario=='busy':claim_task(directory,'source',name)
if scenario=='blocked':
claim_task(directory,'source','reviewer');claim_task(directory,'expected','reviewer')
if scenario=='active':team[name].update(active=True,status='running')
if scenario=='closed':team[name]['status']='closed'
if scenario=='limit':team[name]['turns']=2
if 'scan_ready'=='idle_poll':
if scenario in ('shutdown','shutdown_failure'):
open_request(state,team,mail,'lead',name,'shutdown','本次结束，在空闲边界确认关闭。',lock)
if scenario=='shutdown':team[name]['turns']=2
if scenario=='busy':team[name].update(active=True,status='running')
if scenario=='bad_checks':checks=0
def wait(seconds):
waits.append(seconds)
if scenario=='wakeup' and len(waits)==1:create_task(directory,'source','新到的源码任务','固定main.py',[])
return scenario=='interrupted'
_replace=os.replace
def broken_replace(*_):raise OSError('教学保存失败')
if scenario in ('save_error','shutdown_failure'):os.replace=broken_replace
# 调用入口
try:
observed=scan_ready(board)
except (ValueError,OSError) as error:
observed={"error":str(error)}
finally:
os.replace=_replace
print(observed)
- 验证快照，检查pending、owner和全部依赖。
- 只交回候选，不改负责人。
在 ready.append 行暂停，当前任务已经通过条件，但还没加入返回列表。继续单步后，列表才增加。选择“已有负责人”，那项会跳过；选择“依赖全部完成”，即使 verify 的 blocked_by 不为空，也能进入候选。
扫描验证会复制输入，因此原快照保持。这个副本说明当时有哪些项可做；等另一位队友更新看板，旧副本仍然保持旧样子。要开始工作，还缺少一次对最新状态的确认。
claim_next：在同一把锁内重新读取
函数接收看板目录、成员表、宿主绑定的成员名和共享锁，返回 claimed、idle、busy 或 unavailable。模型不能自行填写负责人。
先确认成员处于 idle、没有活跃轮次，也未耗尽两次交付。随后在锁内重新读取看板；若自己仍持有 in_progress 任务，返回 busy，不继续多取。再用新快照扫描候选。
真实执行轨迹回放；原样函数、锁与Worker虚拟看板/邮箱；等待回调公开注入，不启动线程、模型或程序。
Pys17/core/runtime.py · claim_next
未运行
def claim_next(directory, team, name, lock):
"""同一把锁内重新读盘、扫描、认领；成功保存才交出任务。"""
member_name(name, team)
with lock:
member = team[name]
if (
name == "lead"
or member["active"]
or member["status"] != "idle"
or member["turns"] >= 2
):
return {"status": "unavailable"}
board = load_board(directory)
if any(
t["owner"] == name and t["status"] == "in_progress"
for t in board["tasks"].values()
):
return {"status": "busy"}
ready = scan_ready(board)
if not ready:
return {"status": "idle"}
task = claim_task(directory, ready[0], name)
return {"status": "claimed", "task": task}输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
from pathlib import Path
from threading import RLock
import json
import re
import os
import tempfile
PREFIX="[team-protocol] "
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}
def member_name(name, members):
if not isinstance(name, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,23}', name) or name not in members:
raise ValueError('只能使用已登记的成员名。')
return name
def create_team():
return {name: {'role': role, 'history': [], 'active': False, 'status': 'idle', 'turns': 0}
for name, role in [('lead', '汇总只读证据'), ('reader', '读取源码并保留调查上下文'),
('reviewer', '读取预期并通知reader')]}
def load_inbox(directory, recipient, members):
member_name(recipient, members)
target = Path(directory) / (recipient + '.jsonl')
if target.is_symlink():
raise ValueError('邮箱不能是符号链接。')
if not target.exists():
return []
with target.open('rb') as stream:
data = stream.read(32001)
if len(data) > 32000:
raise ValueError('邮箱文件超过限制。')
items = [json.loads(line) for line in data.decode('utf-8').splitlines()]
if len(items) > 8:
raise ValueError('单个邮箱最多8条消息。')
for index, item in enumerate(items, 1):
if type(item) is not dict or set(item) != {'id', 'sender', 'recipient', 'body'}:
raise ValueError('邮箱记录字段不符。')
if item['id'] != recipient + '-' + str(index) or item['recipient'] != recipient:
raise ValueError('邮箱序号或收件人不符。')
member_name(item['sender'], members)
if type(item['body']) is not str or not 1 <= len(item['body']) <= 600:
raise ValueError('消息正文长度不符。')
return items
def save_inbox(directory, recipient, items):
root = Path(directory); root.mkdir(parents=True, exist_ok=True)
payload = ''.join(json.dumps(item, ensure_ascii=False) + '\n' for item in items)
if len(payload.encode('utf-8')) > 32000:
raise ValueError('邮箱文件超过限制。')
temporary = None
try:
with tempfile.NamedTemporaryFile('w', encoding='utf-8', dir=root, delete=False) as stream:
temporary = stream.name
stream.write(payload); stream.flush(); os.fsync(stream.fileno())
os.replace(temporary, root / (recipient + '.jsonl'))
finally:
if temporary is not None and Path(temporary).exists():
Path(temporary).unlink()
def send_message(directory, members, sender, recipient, body, lock):
"""先保存到对方邮箱；queued不代表接收方模型已经读取。"""
member_name(sender, members)
member_name(recipient, members)
if not isinstance(body, str) or not 1 <= len(body.strip()) <= 600:
raise ValueError('正文需要1到600字符。')
with lock:
items = load_inbox(directory, recipient, members)
if len(items) >= 8:
raise ValueError('邮箱已满，不覆盖旧消息。')
message = {'id': recipient + '-' + str(len(items) + 1), 'sender': sender,
'recipient': recipient, 'body': body.strip()}
save_inbox(directory, recipient, items + [message])
return {'id': message['id'], 'status': 'queued'}
def encode_protocol(data):
return PREFIX + json.dumps(data, ensure_ascii=False, sort_keys=True)
def read_protocol(message):
if type(message) is not dict or not all(
type(message.get(k)) is str for k in ("sender", "recipient", "body")
):
raise ValueError("消息信封不符。")
if not message["body"].startswith(PREFIX):
return None
data = json.loads(message["body"][len(PREFIX) :])
if (
type(data) is not dict
or type(data.get("type")) is not str
or type(data.get("request_id")) is not str
):
raise ValueError("协议类型或关联ID不符。")
return data
def open_request(state, team, directory, sender, target, kind, payload, lock):
"""邮箱保存成功才登记pending；角色和同类未决请求先验证。"""
member_name(sender, team)
member_name(target, team)
if (
kind not in ("plan", "shutdown")
or (kind == "plan" and not (sender != "lead" and target == "lead"))
or (kind == "shutdown" and not (sender == "lead" and target != "lead"))
):
raise ValueError("协议方向不符。")
if type(payload) is not str or not 1 <= len(payload.strip()) <= 180:
raise ValueError("请求正文需要1到180字符。")
with lock:
if len(state) >= 4 or any(
r["status"] == "pending"
and (r["kind"], r["sender"], r["target"]) == (kind, sender, target)
for r in state.values()
):
raise ValueError("请求额度用完或同类请求仍未决。")
subject = sender if kind == "plan" else target
if team[subject]["status"] == "closed":
raise ValueError("成员已经关闭。")
request_id = "req-" + str(len(state) + 1)
request = {
"type": kind + "_request",
"request_id": request_id,
"payload": payload.strip(),
}
receipt = send_message(
directory, team, sender, target, encode_protocol(request), lock
)
state[request_id] = {
"kind": kind,
"sender": sender,
"target": target,
"payload": payload.strip(),
"status": "pending",
"reply": None,
"receipt": receipt["id"],
}
return {"request_id": request_id, "status": "pending"}
def reply_request(state, team, directory, request_id, actor, approve, feedback, lock):
"""只有目标角色能回复；忙碌成员不能确认关闭，重发不重复存信。"""
member_name(actor, team)
if type(approve) is not bool or type(feedback) is not str or len(feedback) > 120:
raise ValueError("回复需要布尔决定与有界说明。")
with lock:
record = state.get(request_id)
if record is None or record["target"] != actor:
raise ValueError("没有此请求或回复身份不符。")
response = {
"type": record["kind"] + "_response",
"request_id": request_id,
"approve": approve,
"feedback": feedback,
}
if record["reply"] is not None:
if record["reply"] != response:
raise ValueError("同一请求不能改写决定。")
return {"request_id": request_id, "status": "already_sent"}
if record["status"] != "pending":
raise ValueError("请求不再等待回复。")
if record["kind"] == "shutdown" and approve and team[actor]["active"]:
return {"request_id": request_id, "status": "deferred"}
receipt = send_message(
directory, team, actor, record["sender"], encode_protocol(response), lock
)
record["reply"] = response
if record["kind"] == "shutdown" and approve:
team[actor]["status"] = "closed"
return {"request_id": request_id, "status": "queued", "receipt": receipt["id"]}
def shutdown_at_boundary(state, team, mail, name, lock):
"""仅处理发给自己的真实关闭请求；普通消息不唤起空转模型。"""
messages = load_inbox(mail, name, team)
for message in messages:
try:
data = read_protocol(message)
except (ValueError, TypeError):
continue
if data is None or data.get("type") != "shutdown_request":
continue
request_id = data["request_id"]
record = state.get(request_id)
if (
record is None
or record["kind"] != "shutdown"
or record["target"] != name
or record["sender"] != message["sender"]
or record["payload"] != data.get("payload")
or record["status"] != "pending"
):
continue
receipt = reply_request(
state, team, mail, request_id, name, True, "空闲边界确认关闭。", lock
)
return {
"status": "shutdown",
"request_id": request_id,
"reply_status": receipt["status"],
}
return None
def scan_ready(board):
"""只返回快照中的可认领ID，不写负责人，也不运行任务。"""
tasks = checked_board(board)["tasks"]
ready = []
for task_id in sorted(tasks):
task = tasks[task_id]
if task["status"] != "pending" or task["owner"] is not None:
continue
if all(tasks[dep]["status"] == "completed" for dep in task["blocked_by"]):
ready.append(task_id)
return ready
def claim_next(directory, team, name, lock):
"""同一把锁内重新读盘、扫描、认领；成功保存才交出任务。"""
member_name(name, team)
with lock:
member = team[name]
if (
name == "lead"
or member["active"]
or member["status"] != "idle"
or member["turns"] >= 2
):
return {"status": "unavailable"}
board = load_board(directory)
if any(
t["owner"] == name and t["status"] == "in_progress"
for t in board["tasks"].values()
):
return {"status": "busy"}
ready = scan_ready(board)
if not ready:
return {"status": "idle"}
task = claim_task(directory, ready[0], name)
return {"status": "claimed", "task": task}
def idle_poll(directory, team, state, mail, name, lock, wait, checks=3):
"""只做有界宿主检查；关闭优先，没有工作就等待，不调用模型。"""
member_name(name, team)
if type(checks) is not int or not 1 <= checks <= 12:
raise ValueError("检查次数需要1到12。")
for index in range(checks):
with lock:
member = team[name]
if name == "lead" or member["active"] or member["status"] != "idle":
return {"status": "unavailable"}
control = shutdown_at_boundary(state, team, mail, name, lock)
if control is not None:
return control
outcome = claim_next(directory, team, name, lock)
if outcome["status"] != "idle":
return outcome
if index + 1 < checks:
if wait(0.05):
return {"status": "interrupted"}
return {"status": "idle_timeout", "checks": checks}
scenario='claimed'
_temp=tempfile.TemporaryDirectory(prefix='s17-auto-');root=Path(_temp.name);directory=root/'board';mail=root/'mail'
team,state,lock=create_team(),{},RLock();name='reader';checks=3;waits=[]
def make_jobs():
create_task(directory,'source','读取源码','固定main.py',[])
create_task(directory,'expected','读取预期','固定expected.txt',[])
create_task(directory,'verify','核对输出','固定程序',['source','expected'])
if not ('claim_next'=='scan_ready' and scenario=='empty') and not ('claim_next'=='idle_poll' and scenario in ('timeout','wakeup','interrupted')):make_jobs()
if 'claim_next'=='scan_ready':
if scenario in ('blocked','unlocked','owned'):
claim_task(directory,'source','reviewer')
if scenario!='owned':claim_task(directory,'expected','reviewer')
if scenario=='unlocked':
complete_task(directory,'source','reviewer','print(6*7)')
complete_task(directory,'expected','reviewer','42')
board=load_board(directory);before=deepcopy(board)
if scenario=='invalid':board['tasks']['verify']['blocked_by']=['missing']
if 'claim_next'=='claim_next':
if scenario=='stale':claim_task(directory,'expected','reviewer')
if scenario=='busy':claim_task(directory,'source',name)
if scenario=='blocked':
claim_task(directory,'source','reviewer');claim_task(directory,'expected','reviewer')
if scenario=='active':team[name].update(active=True,status='running')
if scenario=='closed':team[name]['status']='closed'
if scenario=='limit':team[name]['turns']=2
if 'claim_next'=='idle_poll':
if scenario in ('shutdown','shutdown_failure'):
open_request(state,team,mail,'lead',name,'shutdown','本次结束，在空闲边界确认关闭。',lock)
if scenario=='shutdown':team[name]['turns']=2
if scenario=='busy':team[name].update(active=True,status='running')
if scenario=='bad_checks':checks=0
def wait(seconds):
waits.append(seconds)
if scenario=='wakeup' and len(waits)==1:create_task(directory,'source','新到的源码任务','固定main.py',[])
return scenario=='interrupted'
_replace=os.replace
def broken_replace(*_):raise OSError('教学保存失败')
if scenario in ('save_error','shutdown_failure'):os.replace=broken_replace
# 调用入口
try:
observed=claim_next(directory, team, name, lock)
except (ValueError,OSError) as error:
observed={"error":str(error)}
finally:
os.replace=_replace
print(observed)
board=load_board(directory)
- 在同一锁内重新读取看板。
- 复查成员、已有任务和依赖。
- 保存成功后才交回claimed。
选择“旧候选已被认领”：准备代码保留一份旧快照，然后让 reviewer 先拿走 expected。reader 的 claim_next 会重新读取，改取 source，不会覆盖旧任务的 owner。
在 claim_task 行停下，图中的当前候选仍未被本次调用保存。helper 会重新检查 pending 与依赖，写入 owner、in_progress 和新的 revision，再原子替换 tasks.json。成功返回后，交出的 task 才包含本次认领结果；保存失败则不交出 claimed。
这里复用第十二课的持久存储，不复制第二套任务格式。本课所有认领、完成及宿主写入都使用同一把 RLock，因此多条线程的读改写不会互相覆盖。这个保证只覆盖同一进程；没有跨进程文件锁，也没有成员失联后的自动接管。
认领函数只交出任务，不运行它。模型与工具接在锁外，避免一项工作拖住其他成员的状态更新。本例约定每个成员只有一条工作线程，认领、工作和关闭都由该线程按边界推进。
idle_poll：先查关闭，再看任务，最后等待
函数还接收协议表、自己的邮箱、等待函数与检查次数。默认最多三次检查，参数只接受 1 到 12 的整数。
每次检查先确认成员处于空闲边界，再调用 shutdown_at_boundary。helper 从真实邮箱信封读取关闭请求，核对请求 ID、角色、正文与未决状态，复用第十六课的 reply_request 保存确认。
真实执行轨迹回放；原样函数、锁与Worker虚拟看板/邮箱；等待回调公开注入，不启动线程、模型或程序。
Pys17/core/runtime.py · idle_poll
未运行
def idle_poll(directory, team, state, mail, name, lock, wait, checks=3):
"""只做有界宿主检查；关闭优先，没有工作就等待，不调用模型。"""
member_name(name, team)
if type(checks) is not int or not 1 <= checks <= 12:
raise ValueError("检查次数需要1到12。")
for index in range(checks):
with lock:
member = team[name]
if name == "lead" or member["active"] or member["status"] != "idle":
return {"status": "unavailable"}
control = shutdown_at_boundary(state, team, mail, name, lock)
if control is not None:
return control
outcome = claim_next(directory, team, name, lock)
if outcome["status"] != "idle":
return outcome
if index + 1 < checks:
if wait(0.05):
return {"status": "interrupted"}
return {"status": "idle_timeout", "checks": checks}输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
from pathlib import Path
from threading import RLock
import json
import re
import os
import tempfile
PREFIX="[team-protocol] "
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}
def member_name(name, members):
if not isinstance(name, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,23}', name) or name not in members:
raise ValueError('只能使用已登记的成员名。')
return name
def create_team():
return {name: {'role': role, 'history': [], 'active': False, 'status': 'idle', 'turns': 0}
for name, role in [('lead', '汇总只读证据'), ('reader', '读取源码并保留调查上下文'),
('reviewer', '读取预期并通知reader')]}
def load_inbox(directory, recipient, members):
member_name(recipient, members)
target = Path(directory) / (recipient + '.jsonl')
if target.is_symlink():
raise ValueError('邮箱不能是符号链接。')
if not target.exists():
return []
with target.open('rb') as stream:
data = stream.read(32001)
if len(data) > 32000:
raise ValueError('邮箱文件超过限制。')
items = [json.loads(line) for line in data.decode('utf-8').splitlines()]
if len(items) > 8:
raise ValueError('单个邮箱最多8条消息。')
for index, item in enumerate(items, 1):
if type(item) is not dict or set(item) != {'id', 'sender', 'recipient', 'body'}:
raise ValueError('邮箱记录字段不符。')
if item['id'] != recipient + '-' + str(index) or item['recipient'] != recipient:
raise ValueError('邮箱序号或收件人不符。')
member_name(item['sender'], members)
if type(item['body']) is not str or not 1 <= len(item['body']) <= 600:
raise ValueError('消息正文长度不符。')
return items
def save_inbox(directory, recipient, items):
root = Path(directory); root.mkdir(parents=True, exist_ok=True)
payload = ''.join(json.dumps(item, ensure_ascii=False) + '\n' for item in items)
if len(payload.encode('utf-8')) > 32000:
raise ValueError('邮箱文件超过限制。')
temporary = None
try:
with tempfile.NamedTemporaryFile('w', encoding='utf-8', dir=root, delete=False) as stream:
temporary = stream.name
stream.write(payload); stream.flush(); os.fsync(stream.fileno())
os.replace(temporary, root / (recipient + '.jsonl'))
finally:
if temporary is not None and Path(temporary).exists():
Path(temporary).unlink()
def send_message(directory, members, sender, recipient, body, lock):
"""先保存到对方邮箱；queued不代表接收方模型已经读取。"""
member_name(sender, members)
member_name(recipient, members)
if not isinstance(body, str) or not 1 <= len(body.strip()) <= 600:
raise ValueError('正文需要1到600字符。')
with lock:
items = load_inbox(directory, recipient, members)
if len(items) >= 8:
raise ValueError('邮箱已满，不覆盖旧消息。')
message = {'id': recipient + '-' + str(len(items) + 1), 'sender': sender,
'recipient': recipient, 'body': body.strip()}
save_inbox(directory, recipient, items + [message])
return {'id': message['id'], 'status': 'queued'}
def encode_protocol(data):
return PREFIX + json.dumps(data, ensure_ascii=False, sort_keys=True)
def read_protocol(message):
if type(message) is not dict or not all(
type(message.get(k)) is str for k in ("sender", "recipient", "body")
):
raise ValueError("消息信封不符。")
if not message["body"].startswith(PREFIX):
return None
data = json.loads(message["body"][len(PREFIX) :])
if (
type(data) is not dict
or type(data.get("type")) is not str
or type(data.get("request_id")) is not str
):
raise ValueError("协议类型或关联ID不符。")
return data
def open_request(state, team, directory, sender, target, kind, payload, lock):
"""邮箱保存成功才登记pending；角色和同类未决请求先验证。"""
member_name(sender, team)
member_name(target, team)
if (
kind not in ("plan", "shutdown")
or (kind == "plan" and not (sender != "lead" and target == "lead"))
or (kind == "shutdown" and not (sender == "lead" and target != "lead"))
):
raise ValueError("协议方向不符。")
if type(payload) is not str or not 1 <= len(payload.strip()) <= 180:
raise ValueError("请求正文需要1到180字符。")
with lock:
if len(state) >= 4 or any(
r["status"] == "pending"
and (r["kind"], r["sender"], r["target"]) == (kind, sender, target)
for r in state.values()
):
raise ValueError("请求额度用完或同类请求仍未决。")
subject = sender if kind == "plan" else target
if team[subject]["status"] == "closed":
raise ValueError("成员已经关闭。")
request_id = "req-" + str(len(state) + 1)
request = {
"type": kind + "_request",
"request_id": request_id,
"payload": payload.strip(),
}
receipt = send_message(
directory, team, sender, target, encode_protocol(request), lock
)
state[request_id] = {
"kind": kind,
"sender": sender,
"target": target,
"payload": payload.strip(),
"status": "pending",
"reply": None,
"receipt": receipt["id"],
}
return {"request_id": request_id, "status": "pending"}
def reply_request(state, team, directory, request_id, actor, approve, feedback, lock):
"""只有目标角色能回复；忙碌成员不能确认关闭，重发不重复存信。"""
member_name(actor, team)
if type(approve) is not bool or type(feedback) is not str or len(feedback) > 120:
raise ValueError("回复需要布尔决定与有界说明。")
with lock:
record = state.get(request_id)
if record is None or record["target"] != actor:
raise ValueError("没有此请求或回复身份不符。")
response = {
"type": record["kind"] + "_response",
"request_id": request_id,
"approve": approve,
"feedback": feedback,
}
if record["reply"] is not None:
if record["reply"] != response:
raise ValueError("同一请求不能改写决定。")
return {"request_id": request_id, "status": "already_sent"}
if record["status"] != "pending":
raise ValueError("请求不再等待回复。")
if record["kind"] == "shutdown" and approve and team[actor]["active"]:
return {"request_id": request_id, "status": "deferred"}
receipt = send_message(
directory, team, actor, record["sender"], encode_protocol(response), lock
)
record["reply"] = response
if record["kind"] == "shutdown" and approve:
team[actor]["status"] = "closed"
return {"request_id": request_id, "status": "queued", "receipt": receipt["id"]}
def shutdown_at_boundary(state, team, mail, name, lock):
"""仅处理发给自己的真实关闭请求；普通消息不唤起空转模型。"""
messages = load_inbox(mail, name, team)
for message in messages:
try:
data = read_protocol(message)
except (ValueError, TypeError):
continue
if data is None or data.get("type") != "shutdown_request":
continue
request_id = data["request_id"]
record = state.get(request_id)
if (
record is None
or record["kind"] != "shutdown"
or record["target"] != name
or record["sender"] != message["sender"]
or record["payload"] != data.get("payload")
or record["status"] != "pending"
):
continue
receipt = reply_request(
state, team, mail, request_id, name, True, "空闲边界确认关闭。", lock
)
return {
"status": "shutdown",
"request_id": request_id,
"reply_status": receipt["status"],
}
return None
def scan_ready(board):
"""只返回快照中的可认领ID，不写负责人，也不运行任务。"""
tasks = checked_board(board)["tasks"]
ready = []
for task_id in sorted(tasks):
task = tasks[task_id]
if task["status"] != "pending" or task["owner"] is not None:
continue
if all(tasks[dep]["status"] == "completed" for dep in task["blocked_by"]):
ready.append(task_id)
return ready
def claim_next(directory, team, name, lock):
"""同一把锁内重新读盘、扫描、认领；成功保存才交出任务。"""
member_name(name, team)
with lock:
member = team[name]
if (
name == "lead"
or member["active"]
or member["status"] != "idle"
or member["turns"] >= 2
):
return {"status": "unavailable"}
board = load_board(directory)
if any(
t["owner"] == name and t["status"] == "in_progress"
for t in board["tasks"].values()
):
return {"status": "busy"}
ready = scan_ready(board)
if not ready:
return {"status": "idle"}
task = claim_task(directory, ready[0], name)
return {"status": "claimed", "task": task}
def idle_poll(directory, team, state, mail, name, lock, wait, checks=3):
"""只做有界宿主检查；关闭优先，没有工作就等待，不调用模型。"""
member_name(name, team)
if type(checks) is not int or not 1 <= checks <= 12:
raise ValueError("检查次数需要1到12。")
for index in range(checks):
with lock:
member = team[name]
if name == "lead" or member["active"] or member["status"] != "idle":
return {"status": "unavailable"}
control = shutdown_at_boundary(state, team, mail, name, lock)
if control is not None:
return control
outcome = claim_next(directory, team, name, lock)
if outcome["status"] != "idle":
return outcome
if index + 1 < checks:
if wait(0.05):
return {"status": "interrupted"}
return {"status": "idle_timeout", "checks": checks}
scenario='task'
_temp=tempfile.TemporaryDirectory(prefix='s17-auto-');root=Path(_temp.name);directory=root/'board';mail=root/'mail'
team,state,lock=create_team(),{},RLock();name='reader';checks=3;waits=[]
def make_jobs():
create_task(directory,'source','读取源码','固定main.py',[])
create_task(directory,'expected','读取预期','固定expected.txt',[])
create_task(directory,'verify','核对输出','固定程序',['source','expected'])
if not ('idle_poll'=='scan_ready' and scenario=='empty') and not ('idle_poll'=='idle_poll' and scenario in ('timeout','wakeup','interrupted')):make_jobs()
if 'idle_poll'=='scan_ready':
if scenario in ('blocked','unlocked','owned'):
claim_task(directory,'source','reviewer')
if scenario!='owned':claim_task(directory,'expected','reviewer')
if scenario=='unlocked':
complete_task(directory,'source','reviewer','print(6*7)')
complete_task(directory,'expected','reviewer','42')
board=load_board(directory);before=deepcopy(board)
if scenario=='invalid':board['tasks']['verify']['blocked_by']=['missing']
if 'idle_poll'=='claim_next':
if scenario=='stale':claim_task(directory,'expected','reviewer')
if scenario=='busy':claim_task(directory,'source',name)
if scenario=='blocked':
claim_task(directory,'source','reviewer');claim_task(directory,'expected','reviewer')
if scenario=='active':team[name].update(active=True,status='running')
if scenario=='closed':team[name]['status']='closed'
if scenario=='limit':team[name]['turns']=2
if 'idle_poll'=='idle_poll':
if scenario in ('shutdown','shutdown_failure'):
open_request(state,team,mail,'lead',name,'shutdown','本次结束，在空闲边界确认关闭。',lock)
if scenario=='shutdown':team[name]['turns']=2
if scenario=='busy':team[name].update(active=True,status='running')
if scenario=='bad_checks':checks=0
def wait(seconds):
waits.append(seconds)
if scenario=='wakeup' and len(waits)==1:create_task(directory,'source','新到的源码任务','固定main.py',[])
return scenario=='interrupted'
_replace=os.replace
def broken_replace(*_):raise OSError('教学保存失败')
if scenario in ('save_error','shutdown_failure'):os.replace=broken_replace
# 调用入口
try:
observed=idle_poll(directory, team, state, mail, name, lock, wait, checks)
except (ValueError,OSError) as error:
observed={"error":str(error)}
finally:
os.replace=_replace
print(observed)
- 先处理关闭，再看任务。
- 无可做项时释放锁等待。
- 检查次数、交付和模型调用各有边界。
选择“关闭优先”：即使看板上还有 ready 任务、成员已用完交付额度，空闲关闭请求仍会先被处理。确认保存后成员 closed，返回 shutdown，没有进入认领分支。Lead 的请求仍要等它自己匹配回复，不能把本地关闭直接写成双方状态都完成。
若选择“关闭保存失败”，函数交回错误，不继续取任务，也不把未保存确认写成关闭成功。正在工作的成员则直接返回 unavailable；本函数不在模型调用中间修改历史或中断工具。
没有关闭时才调用 claim_next。有已保存任务就结束空闲阶段；没有可做项则退出锁，调用 wait(0.05)。公开入口传入 Event.wait，等待期间别的线程可以更新任务或发来关闭请求。
选择“等待后有新任务”，公开等待回调会在 Worker 虚拟文件系统里新增一项任务；下一次检查才读到并认领。图跟随原样函数的实际执行轨迹，等待回调可展开查看，不实际延时，也不启动线程、模型或程序。
默认三次检查最多两次等待，最后一次没有工作就返回 idle_timeout。这是有限检查预算，不是严格的墙钟超时；文件读写和检查也要耗时。它不能证明看板全部完成。宿主中止等待则返回 interrupted，后续处理交回调用者。
拿到工作以后，怎样接回旧循环？
公开 demo 创建看板与两条工作线程。每个线程调用 idle_poll，只有拿到 claimed 才将 task 交给 run_teammate，回调 run_job 使用该成员旧历史与唯一的原 run_loop。
python
# user_code/practice.py 的工作交接，省略参数装配与收尾
picked = idle_poll(board_dir, team, protocols, mail, actor, lock, stop.wait)
# 仅在 picked["status"] == "claimed" 时继续
result = run_teammate(team, actor, turn, lock)
saved = finish_job(board_dir, task, actor, box.get("outcome", result), lock)任务调度在循环外，工具请求与反馈仍在原循环里。run_job 按任务只注册 read_file 或 run_main；前置检查核对当前负责人和固定文件路径，再走原权限与 Hook。系统说明继续按真实工具生成，当前模型请求仍复用第十一课的有界恢复。
本例由宿主登记固定步骤，没有新增模型提案，也没有把自主认领当成自动批准计划。运行 main.py 仍取决于本次宿主 --decision。认领成功只是取得工作，不是获得额外执行权限。
回答结束以后，谁来决定任务完成？
finish_job 检查当前轮次实际 tool_result，要求轮次正常回答、工具成功、结果非空。verify 还要把本次 exit_code 与输出和 expected 的已保存读取结果比较；通过后，才在同一把锁内调用 complete_task。
程序被拒绝、执行错误、没有工具证据或输出不符，都保留原 owner 与 in_progress，并返回 needs_attention。模型一句“完成了”不会覆盖这个检查。这里验证的是本课固定任务的具体证据，不是一个适用于所有业务的自动验收器。
失败后不退回 pending，也不换一个成员自动重跑。关闭成员同样不会释放未决任务。这样才能让宿主知道哪里需要处理，避免隐藏一次已经发生过的副作用。
每成员最多两次工作交付，每轮最多四次逻辑模型请求、请求恢复最多二十次实际尝试。达到边界或没有新工作，线程交回控制权。最终示例在 join 后统一请求成员关闭，再让 Lead 匹配确认并接入队友反馈；普通消息在实际工作 prepare 时接入，不单独触发空模型。
默认状态目录结束后清理。--state-dir 可以留下看板和邮箱查看，但必须为空；持久任务进度不等于恢复全部成员历史，本课不自动续跑旧会话。
用并发和失败实验检查，而不是只看一次终端输出
bash
python3 examples/s17-autonomous-agents/user_code/verify_practice.py
python3 examples/s17-autonomous-agents/user_code/verify_fixtures.py十三组标准库测试用 Barrier 确认两个读取任务确实由不同成员并行处理；让两个成员同时认领唯一任务，只有一个 claimed；保留旧快照后认领，仍能拿到最新状态。Event 实验则确认等待不持锁，新任务或关闭到达后，下一次检查能按顺序处理。
测试也检查依赖解锁、坏看板、替换失败保留原字节、空闲不调用模型，以及错误文件路径不能完成任务。允许路径程序计数恰有一个 x，默认拒绝路径没有程序计数；预期改成 43 时，本次程序运行了，verify 仍不能完成。
三个教学区共二十二种输入、308 个实际快照，运行与调试输出一致。可选 verify_sdk.py 使用 Anthropic SDK 1.13.0 与 MockTransport，三条路径各八次模拟 HTTP，含工具完成后的一次 529；恢复当前请求没有重复认领或工具执行，Lead 邮箱保留五条记录。这是模拟 HTTP 验证，没有在线模型调用，也不承诺一般崩溃恢复后的恰好一次。