让 reader 记得自己刚才读过什么
实验沿用思想篇的只读问题：reader 查 main.py，reviewer 查 expected.txt；后者把预期发给前者，reader 再结合旧源码记录补充报告。
目标不只是出现两条线程。我们要验证：第一轮确实并行；消息先保存、后接入；reader 第二轮继续自己的历史，没有重新读取 main.py；Lead 最后确实在自己的模型输入里看见报告。
源码在独立公开仓库第十五课，本文摘录固定到该提交。Python 3.10+，默认只需标准库：
bash
git clone https://github.com/Chengyunlai/learn-agent-harness.git
cd learn-agent-harness
python3 examples/s15-agent-teams/user_code/practice.py模型响应由固定替身提供，文件读取、线程和 JSONL 邮箱真实执行；没有调用在线模型，也没有运行 main.py。正常路径有九次逻辑模型请求，lead 收到两封，reader 收到一封，reviewer 没有来信。第一阶段两人的结束顺序可以不同，不应把某一份终端输出顺序当成协议。
默认使用临时目录，结束后清理。加上 --maildir /tmp/my-new-team-mail 可以保留邮箱查看，目录必须为空；这不是恢复旧团队的入口。
send_message：先保存，再交收据
输入是登记成员、宿主绑定的 sender、recipient、正文与共享锁。模型只能提供收件人和正文，不能自行宣称 sender 是 Lead。函数输出短收据，例如 id 为 reader-1、status 为 queued。
先验证成员和正文，再在锁内读取收件箱、检查容量、生成下一个 ID。读取和保存属于同一次读改写，必须使用同一把锁，否则两个发送者可能读到相同旧记录，后保存的一方覆盖前一方。
真实执行轨迹回放；原样函数、锁和Worker虚拟收件箱；轮次回调是公开替身，不启动模型或线程。
Pys15/core/runtime.py · send_message
未运行
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'}输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
from pathlib import Path
from threading import RLock
import json
import os
import re
import tempfile
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 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 create_team():
return {name: {'role': role, 'history': [], 'active': False, 'status': 'idle', 'turns': 0}
for name, role in [('lead', '汇总只读证据'), ('reader', '读取源码并保留调查上下文'),
('reviewer', '读取预期并通知reader')]}
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 prepare_inbox(history, directory, members, recipient, lock):
"""读消息不删除；以已经提交的历史去重，丢弃候选不会丢信。"""
with lock:
items = load_inbox(directory, recipient, members)
candidate = deepcopy(history)
for message in items:
text = '[team-message] ' + json.dumps(message, ensure_ascii=False, sort_keys=True)
present = any(item.get('role') == 'user' and item.get('content') == text for item in candidate)
if not present:
candidate.append({'role': 'user', 'content': text})
if len(candidate) > 80:
raise ValueError('队友历史超过80条，交回宿主检查。')
return candidate
def run_teammate(team, name, run_turn, 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 {'member': name, 'status': 'unavailable'}
history = deepcopy(member['history'])
member.update(active=True, status='running', turns=member['turns'] + 1)
try:
outcome = run_turn(history)
if (type(outcome) is not dict or type(outcome.get('messages')) is not list
or len(outcome['messages']) > 80 or type(outcome.get('status')) is not str
or type(outcome.get('text')) is not str):
raise ValueError('轮次没有交回有界历史。')
with lock:
member['history'] = deepcopy(outcome['messages'])
member['status'] = 'idle' if outcome['status'] == 'answered' else 'stopped'
return {'member': name, 'status': outcome['status'], 'text': str(outcome['text'])[:600]}
except Exception as error:
with lock:
member['status'] = 'failed'
return {'member': name, 'status': 'failed', 'text': type(error).__name__}
finally:
with lock:
member['active'] = False
if member['status'] == 'running':
member['status'] = 'stopped'
scenario = 'queued'
team, lock = create_team(), RLock()
_temp = tempfile.TemporaryDirectory(prefix='s15-mail-')
directory = Path(_temp.name)
sender, recipient, body = 'reviewer', 'reader', '当前expected.txt记录为42；尚未运行。'
history = [{'role': 'user', 'content': '保留上轮源码，等队友补充预期。'}]
if 'send_message'=='prepare_inbox' and scenario!='empty':send_message(directory,team,sender,recipient,body,lock)
if scenario=='already':history=prepare_inbox(history,directory,team,recipient,lock)
if scenario=='discarded':prepare_inbox(history,directory,team,recipient,lock)
if scenario=='corrupt':(directory/'reader.jsonl').write_text('not-json')
if scenario=='limit':history=[{'role':'user','content':'已存消息'+str(i)} for i in range(80)]
if scenario=='unknown':recipient='../other'
if scenario=='sender':sender='unknown'
if scenario=='invalid':body=''
if scenario=='full':
for i in range(8):send_message(directory,team,sender,recipient,'旧记录'+str(i),lock)
name='reader'
if scenario=='continued':
team[name]['history']=[{'role':'user','content':'先前读取main.py。'},{'role':'assistant','content':'已保存源码；没有运行。'}]
team[name]['turns']=1
if scenario=='busy':team[name].update(active=True,status='running')
if scenario=='exhausted':team[name]['turns']=2
received=[]
def run_turn(previous):
received.append(deepcopy(previous))
if scenario=='failed':raise OSError('教学轮次失败')
return {'status':'call_limit' if scenario=='stopped' else 'answered','text':'保存补充说明，仍没有运行。',
'messages':previous+[{'role':'user','content':'本轮追问'},{'role':'assistant','content':'依据已有记录补充回答。'}]}
before=deepcopy(history)
old_mail=(directory/'reader.jsonl').read_bytes() if (directory/'reader.jsonl').exists() else None
_replace=os.replace
def broken_replace(*_):raise OSError('教学保存失败')
if scenario=='replace_error':os.replace=broken_replace
# 调用入口
try:
observed=send_message(directory, team, sender, recipient, body, lock)
except (ValueError,OSError) as error:
observed={"error":str(error)}
finally:
os.replace=_replace
assert history==before
print(observed)
- 验证登记成员、正文与容量。
- 追加候选保存后，交回queued收据。
- 收据不代表接收方模型已读。
在 save_inbox 行设断点：此时内存候选已经追加新信，磁盘保存还没发生。单步跨过 helper 以后，才能把新记录画成已存，并继续生成收据。图只跟随这个实际执行位置变化。
save_inbox 在同目录写临时文件，flush、fsync 后用 os.replace 替换；出错清理临时文件，原文件保留。load_inbox 则校验顺序 ID、登记成员和字段；损坏文件不会被当成空邮箱继续覆盖。单箱最多八条、正文 600 字符、文件 32000 字节。选择“收件箱已满”或“保存失败”，可以看到拒绝后旧字节没有变化。
原子替换解决一次文件更新的可见性，共享 RLock 解决本进程多个线程之间的读改写；本课没有跨进程写锁，也不承诺崩溃后的恰好一次投递。返回 queued 同样没有启动接收方模型。
prepare_inbox：构造候选，别提前确认
输入是接收方自己的历史和邮箱位置，输出是历史副本。函数先在锁内读取邮箱快照，随后释放锁，再复制历史、查找缺少的消息。
每封信转成带 [team-message] 前缀的普通 user 文本，内容包含消息 ID、发送者、收件人和正文。它没有伪装成 tool_result；发送方的工具收据，和接收方后来收到的通知，是两次不同交接。
真实执行轨迹回放；原样函数、锁和Worker虚拟收件箱；轮次回调是公开替身，不启动模型或线程。
Pys15/core/runtime.py · prepare_inbox
未运行
def prepare_inbox(history, directory, members, recipient, lock):
"""读消息不删除；以已经提交的历史去重，丢弃候选不会丢信。"""
with lock:
items = load_inbox(directory, recipient, members)
candidate = deepcopy(history)
for message in items:
text = '[team-message] ' + json.dumps(message, ensure_ascii=False, sort_keys=True)
present = any(item.get('role') == 'user' and item.get('content') == text for item in candidate)
if not present:
candidate.append({'role': 'user', 'content': text})
if len(candidate) > 80:
raise ValueError('队友历史超过80条，交回宿主检查。')
return candidate输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
from pathlib import Path
from threading import RLock
import json
import os
import re
import tempfile
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 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 create_team():
return {name: {'role': role, 'history': [], 'active': False, 'status': 'idle', 'turns': 0}
for name, role in [('lead', '汇总只读证据'), ('reader', '读取源码并保留调查上下文'),
('reviewer', '读取预期并通知reader')]}
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 prepare_inbox(history, directory, members, recipient, lock):
"""读消息不删除；以已经提交的历史去重，丢弃候选不会丢信。"""
with lock:
items = load_inbox(directory, recipient, members)
candidate = deepcopy(history)
for message in items:
text = '[team-message] ' + json.dumps(message, ensure_ascii=False, sort_keys=True)
present = any(item.get('role') == 'user' and item.get('content') == text for item in candidate)
if not present:
candidate.append({'role': 'user', 'content': text})
if len(candidate) > 80:
raise ValueError('队友历史超过80条，交回宿主检查。')
return candidate
def run_teammate(team, name, run_turn, 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 {'member': name, 'status': 'unavailable'}
history = deepcopy(member['history'])
member.update(active=True, status='running', turns=member['turns'] + 1)
try:
outcome = run_turn(history)
if (type(outcome) is not dict or type(outcome.get('messages')) is not list
or len(outcome['messages']) > 80 or type(outcome.get('status')) is not str
or type(outcome.get('text')) is not str):
raise ValueError('轮次没有交回有界历史。')
with lock:
member['history'] = deepcopy(outcome['messages'])
member['status'] = 'idle' if outcome['status'] == 'answered' else 'stopped'
return {'member': name, 'status': outcome['status'], 'text': str(outcome['text'])[:600]}
except Exception as error:
with lock:
member['status'] = 'failed'
return {'member': name, 'status': 'failed', 'text': type(error).__name__}
finally:
with lock:
member['active'] = False
if member['status'] == 'running':
member['status'] = 'stopped'
scenario = 'new'
team, lock = create_team(), RLock()
_temp = tempfile.TemporaryDirectory(prefix='s15-mail-')
directory = Path(_temp.name)
sender, recipient, body = 'reviewer', 'reader', '当前expected.txt记录为42；尚未运行。'
history = [{'role': 'user', 'content': '保留上轮源码，等队友补充预期。'}]
if 'prepare_inbox'=='prepare_inbox' and scenario!='empty':send_message(directory,team,sender,recipient,body,lock)
if scenario=='already':history=prepare_inbox(history,directory,team,recipient,lock)
if scenario=='discarded':prepare_inbox(history,directory,team,recipient,lock)
if scenario=='corrupt':(directory/'reader.jsonl').write_text('not-json')
if scenario=='limit':history=[{'role':'user','content':'已存消息'+str(i)} for i in range(80)]
if scenario=='unknown':recipient='../other'
if scenario=='sender':sender='unknown'
if scenario=='invalid':body=''
if scenario=='full':
for i in range(8):send_message(directory,team,sender,recipient,'旧记录'+str(i),lock)
name='reader'
if scenario=='continued':
team[name]['history']=[{'role':'user','content':'先前读取main.py。'},{'role':'assistant','content':'已保存源码；没有运行。'}]
team[name]['turns']=1
if scenario=='busy':team[name].update(active=True,status='running')
if scenario=='exhausted':team[name]['turns']=2
received=[]
def run_turn(previous):
received.append(deepcopy(previous))
if scenario=='failed':raise OSError('教学轮次失败')
return {'status':'call_limit' if scenario=='stopped' else 'answered','text':'保存补充说明，仍没有运行。',
'messages':previous+[{'role':'user','content':'本轮追问'},{'role':'assistant','content':'依据已有记录补充回答。'}]}
before=deepcopy(history)
old_mail=(directory/'reader.jsonl').read_bytes() if (directory/'reader.jsonl').exists() else None
_replace=os.replace
def broken_replace(*_):raise OSError('教学保存失败')
if scenario=='replace_error':os.replace=broken_replace
# 调用入口
try:
observed=prepare_inbox(history, directory, team, recipient, lock)
except (ValueError,OSError) as error:
observed={"error":str(error)}
finally:
os.replace=_replace
assert history==before
print(observed)
- 读取收件箱但不删除。
- 用已提交历史去重，只追加缺少的通知。
- 候选交回原循环后才进入请求。
在 candidate.append 行停住时，新的通知还没进入候选；下一步才会增加。原 history 与磁盘记录始终保留。选择“丢弃上次候选”重新运行，仍能读到消息；选择“历史已经包含”，完整文本已在传入历史中，就不会再次追加。
这里以已提交历史中的完整通知文本去重，前提是消息记录追加后不改写，本例也不清理这些历史通知。若以后引入摘要或移除旧历史，应另外持久保存确认位置，不能假设这条去重规则仍然足够。
候选超过 80 条会拒绝交出。这样错误发生在替换历史之前，而不是先删信、再发现装不下。准备完成也不表示模型已经收到：还要由原循环提交这份候选，再发下一次请求。
run_teammate：占用身份，带着旧历史继续
team 中每名成员保存 role、history、active、status、turns。run_teammate 接收成员名与 run_turn 回调，先在锁内检查：是不是可交付成员、是否正在工作、是否 idle、是否已经交付两次。
可开始时，复制该成员自己的历史，设置 active 与 running，再增加次数。接下来必须离开锁才调用 run_turn；否则 reviewer 发信或其他成员开始一轮，也会被一个慢请求挡住。
真实执行轨迹回放；原样函数、锁和Worker虚拟收件箱；轮次回调是公开替身，不启动模型或线程。
Pys15/core/runtime.py · run_teammate
未运行
def run_teammate(team, name, run_turn, 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 {'member': name, 'status': 'unavailable'}
history = deepcopy(member['history'])
member.update(active=True, status='running', turns=member['turns'] + 1)
try:
outcome = run_turn(history)
if (type(outcome) is not dict or type(outcome.get('messages')) is not list
or len(outcome['messages']) > 80 or type(outcome.get('status')) is not str
or type(outcome.get('text')) is not str):
raise ValueError('轮次没有交回有界历史。')
with lock:
member['history'] = deepcopy(outcome['messages'])
member['status'] = 'idle' if outcome['status'] == 'answered' else 'stopped'
return {'member': name, 'status': outcome['status'], 'text': str(outcome['text'])[:600]}
except Exception as error:
with lock:
member['status'] = 'failed'
return {'member': name, 'status': 'failed', 'text': type(error).__name__}
finally:
with lock:
member['active'] = False
if member['status'] == 'running':
member['status'] = 'stopped'输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
from pathlib import Path
from threading import RLock
import json
import os
import re
import tempfile
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 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 create_team():
return {name: {'role': role, 'history': [], 'active': False, 'status': 'idle', 'turns': 0}
for name, role in [('lead', '汇总只读证据'), ('reader', '读取源码并保留调查上下文'),
('reviewer', '读取预期并通知reader')]}
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 prepare_inbox(history, directory, members, recipient, lock):
"""读消息不删除；以已经提交的历史去重，丢弃候选不会丢信。"""
with lock:
items = load_inbox(directory, recipient, members)
candidate = deepcopy(history)
for message in items:
text = '[team-message] ' + json.dumps(message, ensure_ascii=False, sort_keys=True)
present = any(item.get('role') == 'user' and item.get('content') == text for item in candidate)
if not present:
candidate.append({'role': 'user', 'content': text})
if len(candidate) > 80:
raise ValueError('队友历史超过80条，交回宿主检查。')
return candidate
def run_teammate(team, name, run_turn, 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 {'member': name, 'status': 'unavailable'}
history = deepcopy(member['history'])
member.update(active=True, status='running', turns=member['turns'] + 1)
try:
outcome = run_turn(history)
if (type(outcome) is not dict or type(outcome.get('messages')) is not list
or len(outcome['messages']) > 80 or type(outcome.get('status')) is not str
or type(outcome.get('text')) is not str):
raise ValueError('轮次没有交回有界历史。')
with lock:
member['history'] = deepcopy(outcome['messages'])
member['status'] = 'idle' if outcome['status'] == 'answered' else 'stopped'
return {'member': name, 'status': outcome['status'], 'text': str(outcome['text'])[:600]}
except Exception as error:
with lock:
member['status'] = 'failed'
return {'member': name, 'status': 'failed', 'text': type(error).__name__}
finally:
with lock:
member['active'] = False
if member['status'] == 'running':
member['status'] = 'stopped'
scenario = 'first'
team, lock = create_team(), RLock()
_temp = tempfile.TemporaryDirectory(prefix='s15-mail-')
directory = Path(_temp.name)
sender, recipient, body = 'reviewer', 'reader', '当前expected.txt记录为42；尚未运行。'
history = [{'role': 'user', 'content': '保留上轮源码，等队友补充预期。'}]
if 'run_teammate'=='prepare_inbox' and scenario!='empty':send_message(directory,team,sender,recipient,body,lock)
if scenario=='already':history=prepare_inbox(history,directory,team,recipient,lock)
if scenario=='discarded':prepare_inbox(history,directory,team,recipient,lock)
if scenario=='corrupt':(directory/'reader.jsonl').write_text('not-json')
if scenario=='limit':history=[{'role':'user','content':'已存消息'+str(i)} for i in range(80)]
if scenario=='unknown':recipient='../other'
if scenario=='sender':sender='unknown'
if scenario=='invalid':body=''
if scenario=='full':
for i in range(8):send_message(directory,team,sender,recipient,'旧记录'+str(i),lock)
name='reader'
if scenario=='continued':
team[name]['history']=[{'role':'user','content':'先前读取main.py。'},{'role':'assistant','content':'已保存源码；没有运行。'}]
team[name]['turns']=1
if scenario=='busy':team[name].update(active=True,status='running')
if scenario=='exhausted':team[name]['turns']=2
received=[]
def run_turn(previous):
received.append(deepcopy(previous))
if scenario=='failed':raise OSError('教学轮次失败')
return {'status':'call_limit' if scenario=='stopped' else 'answered','text':'保存补充说明，仍没有运行。',
'messages':previous+[{'role':'user','content':'本轮追问'},{'role':'assistant','content':'依据已有记录补充回答。'}]}
before=deepcopy(history)
old_mail=(directory/'reader.jsonl').read_bytes() if (directory/'reader.jsonl').exists() else None
_replace=os.replace
def broken_replace(*_):raise OSError('教学保存失败')
if scenario=='replace_error':os.replace=broken_replace
# 调用入口
try:
observed=run_teammate(team, name, run_turn, lock)
except (ValueError,OSError) as error:
observed={"error":str(error)}
finally:
os.replace=_replace
assert history==before
print(observed)
- 锁内验证并占用成员，复制自己的历史。
- 锁外运行原Agent循环。
- 保存本轮历史和状态，finally释放占用。
选择“沿用上下文”，在 run_turn 行暂停。此时成员已占用，传入 history 含上轮记录，回调尚未执行。图中的历史数量、状态和占用来自真实快照，不根据案例名称预填。“已经占用”则在前面返回 unavailable，根本不会走到回调。
回调返回后，先检查 messages、status、text 的结构和历史容量，再写回成员历史。answered 对应 idle，其他正常返回状态对应 stopped；“额度用完”因此不会被画成成功回答。异常对应 failed，并保留旧历史；finally 一定释放 active。已发出的消息等副作用不会随历史一起回滚，所以失败后不自动重跑整轮。
浏览器执行原样函数与存储 helper，使用 Worker 虚拟文件；run_turn 是公开可查看的教学替身，不启动线程或模型。断点展示真实执行轨迹中的下一行、少量变量与输出。真正并发的证明在下面的公开 Python 测试中。
三个函数怎样接回已有循环？
公开入口的 run_member 负责装配：把成员旧历史作为 prefix 保留，本轮交给已有 run_loop。每次请求把 prefix 与本轮 history 合起来；prepare 钩子用同一组合接入邮箱，再只将本轮后缀交回原循环。轮次结束，完整历史才交给 run_teammate 保存。
这样保留一个实际 Agent 循环，也保留第十课的工具提示和第十一课的模型请求恢复。不是为每个队友再抄一份循环，也不是在另外一条线程里随时修改正在发出的 messages。
宿主先并发交付 reader、reviewer 并 join，再给同一个 reader 追问，最后让 Lead 汇总。每个成员仅注册 read_file 与 send_message，工具 ID 带成员与轮次，避免续接历史中的调用编号冲突。旧权限与 Hook 仍经过真实分发路径；本课没有程序执行入口。
同一成员最多两次交付，每次原循环最多四次逻辑请求，请求恢复最多二十次实际尝试。停止状态不自动继续，邮箱日志也不等于可恢复的成员历史。进程重启、跨进程锁与长期邮箱清理留在当前实现边界之外。
怎样证明它没有悄悄丢信或重新调查？
bash
python3 examples/s15-agent-teams/user_code/verify_practice.py
python3 examples/s15-agent-teams/user_code/verify_fixtures.py十二组标准库测试包括：用 Event 门控证明不同成员同时工作、同一成员不能重入；八个发送者并发存信不丢失；满箱、坏文件和替换失败保留旧数据；候选丢弃后能再接入；错误或中断后释放占用。完整入口另核对 reader 第二轮含旧 read_file 结果、没有再次读源码，Lead 的实际请求含两份报告。
三个交互区共有十八种输入、206 个实际快照，运行与调试输出一致。可选的 verify_sdk.py 使用 Anthropic SDK 1.13.0 与 MockTransport 验证十次模拟 HTTP，其中一次 529 发生在发信之后；恢复当前模型请求，没有重复发信，最终仍是三条邮箱记录。这是 SDK 协议验证，没有在线模型调用。
最终我们得到的是一个有限的小团队：各自调查，明确存信，在下一次请求边界接信，并沿用自己的上下文继续。判断实现是否正确，回到这些具体证据就足够了。