实践｜Team Protocols，把同意匹配到这一份请求

先故意让两种“允许”不一致

本课继续处理一个具体问题：“读取预期，运行 main.py，比较结果。”reader 先提交计划，Lead 给出决定；reader 匹配回复后，才进入实际工作。程序执行仍需要宿主许可。

先运行三次，比先背协议字段更容易发现边界。源码在独立公开仓库第十六课，Python 3.10+，默认只需标准库：

bash

git clone https://github.com/Chengyunlai/learn-agent-harness.git
cd learn-agent-harness
python3 examples/s16-team-protocols/user_code/practice.py --decision allow
python3 examples/s16-team-protocols/user_code/practice.py --plan reject --decision allow
python3 examples/s16-team-protocols/user_code/practice.py

第一条真实运行程序，输出 42，并与 expected.txt 比较。第二条即使宿主允许，也因为计划拒绝而不能运行；第三条计划批准，仍因为默认宿主拒绝而不能运行。模型响应都是固定替身，JSONL 邮箱、文件读取和获准后的 Python 进程真实执行，没有在线模型调用。

每条路径在回复保存后、匹配前都会打印 pending。结束工作后，Lead 再请求关闭 reader；空闲成员保存确认并关闭，Lead 匹配后才结束这份请求。思想篇解释这些观察位置为什么要分开。

open_request：先存进邮箱，再登记 pending

函数接收请求表 state、成员表 team、邮箱位置、宿主绑定的 sender、target、协议 kind、正文与共享锁。返回 request_id 和 pending，而不是批准结果。

方向先限制：plan 由队友发给 lead，shutdown 由 lead 发给队友。正文最多 180 字符，会话最多四份请求，同方向同类已有 pending 时拒绝继续新增。已经 closed 的成员不能再发起计划或被请求关闭。

真实执行轨迹回放；原样协议函数与锁、Worker虚拟邮箱；保存故障为公开教学注入，不调用模型或程序。

Pys16/core/runtime.py · open_request

未运行

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"}

输出

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

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

教学准备与调用入口

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

from copy import deepcopy
from pathlib import Path
from threading import RLock
import json
import re
import os
import tempfile
PREFIX="[team-protocol] "
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 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 create_team():
    return {name: {'role': role, 'history': [], 'active': False, 'status': 'idle', 'turns': 0}
            for name, role in [('lead', '汇总只读证据'), ('reader', '读取源码并保留调查上下文'),
                               ('reviewer', '读取预期并通知reader')]}

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 match_response(state, message, recipient, lock):
    """信封角色、类型、ID和已发送决定全部一致，才更新请求状态。"""
    data = read_protocol(message)
    if data is None:
        return {"status": "ignored", "reason": "ordinary_message"}
    if (
        set(data) != {"type", "request_id", "approve", "feedback"}
        or type(data["approve"]) is not bool
        or type(data["feedback"]) is not str
        or len(data["feedback"]) > 120
    ):
        return {"status": "ignored", "reason": "invalid_response"}
    with lock:
        record = state.get(data["request_id"])
        if record is None or data["type"] != record["kind"] + "_response":
            return {"status": "ignored", "reason": "id_or_type"}
        if (
            recipient != record["sender"]
            or message["recipient"] != recipient
            or message["sender"] != record["target"]
        ):
            return {"status": "ignored", "reason": "role_or_recipient"}
        if record["reply"] != data:
            return {"status": "ignored", "reason": "not_saved_reply"}
        if record["status"] != "pending":
            return {"request_id": data["request_id"], "status": "duplicate"}
        record["status"] = "approved" if data["approve"] else "rejected"
    return {"request_id": data["request_id"], "status": record["status"]}


scenario='plan'
state,team,lock={},create_team(),RLock()
_temp=tempfile.TemporaryDirectory(prefix='s16-mail-');directory=Path(_temp.name)
sender,target,kind,payload='reader','lead','plan','读取预期，再运行main.py。'
actor,approve,feedback='lead',True,'仅批准本次计划，权限另行判断。'
if 'open_request'!='open_request':
    kind='shutdown' if scenario in ('busy','closed') else 'plan'
    sender,target=('lead','reader') if kind=='shutdown' else ('reader','lead')
    if kind=='shutdown':payload,feedback='本次工作结束，请在空闲边界关闭。','空闲边界确认本次关闭。'
    request_id=open_request(state,team,directory,sender,target,kind,payload,lock)['request_id']
    actor=target
if 'open_request'=='open_request':
    if scenario=='shutdown':sender,target,kind,payload='lead','reader','shutdown','本次工作结束，请在空闲边界关闭。'
    if scenario=='direction':sender,target='lead','reader'
    if scenario=='pending':open_request(state,team,directory,sender,target,kind,payload,lock)
    if scenario=='limit':
        for i in range(4):state['old-'+str(i)]={'status':'rejected'}
if scenario in ('reject','rejected'):approve=False
if 'open_request'=='reply_request':
    if scenario=='wrong_role':actor='reviewer'
    if scenario=='busy':team['reader'].update(active=True,status='running')
    if scenario in ('duplicate','conflict'):
        reply_request(state,team,directory,request_id,actor,True,feedback,lock)
        if scenario=='conflict':approve=False
if 'open_request'=='match_response':
    reply_request(state,team,directory,request_id,actor,approve,feedback,lock)
    message=deepcopy(load_inbox(directory,sender,team)[-1]);recipient=sender
    data=read_protocol(message)
    if scenario=='wrong_type':data['type']='shutdown_response'
    if scenario=='wrong_role':message['sender']='reviewer'
    if scenario=='unknown':data['request_id']='req-404'
    if scenario=='invalid':data['approve']='true'
    if scenario=='forged':data['feedback']='不同的未保存决定'
    message['body']=encode_protocol(data)
    if scenario=='duplicate':match_response(state,message,recipient,lock)
before=deepcopy(state)
_replace=os.replace
def broken_replace(*_):raise OSError('教学保存失败')
if scenario=='save_error':os.replace=broken_replace


# 调用入口
try:
    observed=open_request(state, team, directory, sender, target, kind, payload, lock)
except (ValueError,OSError) as error:
    observed={"error":str(error)}
finally:
    os.replace=_replace
print(observed)

验证方向与额度。
保存请求后才登记pending。
保存失败保留原记录。

在 send_message 行设断点，请求卡片已经有类型、ID 和正文，内存 state 里还没有新记录。单步跨过保存 helper 后，才来到 state 赋值处。继续单步完成这次赋值，记录才出现。这两个动作使用同一把 RLock，接收线程不会在本进程临界段中间处理一半状态。

请求沿用上一课的 send_message：同目录临时文件写完再替换，失败保留旧邮箱；收件箱仍有八条、600 字符正文和 32000 字节限制。类型化内容放在带 [team-protocol] 前缀的正文里，外层发件人由宿主保存，不从 JSON 正文认领身份。

选择“保存失败”，可以看到没有虚构 pending。但内存请求表与文件不是跨崩溃事务：若进程在保存信件后、内存登记前崩溃，重新启动不能只靠邮箱恢复全部状态。本课入口要求空目录，不宣称已经实现恢复。

reply_request：保存决定之前，谁有资格回答？

输入是原 request_id、宿主绑定的 actor、布尔 approve 和最多 120 字符反馈。目标角色必须等于原记录的 target。只有 Lead 能回复 reader 提交给它的计划；只有被请求关闭的队友能确认自己退出。

函数在同一把锁内检查已有决定。record.reply 已存在时，相同决定返回 already_sent，不再存信；不同决定直接拒绝，保留第一份记录。协议正文中的字符串 true 不能冒充布尔 True。

真实执行轨迹回放；原样协议函数与锁、Worker虚拟邮箱；保存故障为公开教学注入，不调用模型或程序。

Pys16/core/runtime.py · reply_request

未运行

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"]}

输出

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

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

教学准备与调用入口

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

from copy import deepcopy
from pathlib import Path
from threading import RLock
import json
import re
import os
import tempfile
PREFIX="[team-protocol] "
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 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 create_team():
    return {name: {'role': role, 'history': [], 'active': False, 'status': 'idle', 'turns': 0}
            for name, role in [('lead', '汇总只读证据'), ('reader', '读取源码并保留调查上下文'),
                               ('reviewer', '读取预期并通知reader')]}

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 match_response(state, message, recipient, lock):
    """信封角色、类型、ID和已发送决定全部一致，才更新请求状态。"""
    data = read_protocol(message)
    if data is None:
        return {"status": "ignored", "reason": "ordinary_message"}
    if (
        set(data) != {"type", "request_id", "approve", "feedback"}
        or type(data["approve"]) is not bool
        or type(data["feedback"]) is not str
        or len(data["feedback"]) > 120
    ):
        return {"status": "ignored", "reason": "invalid_response"}
    with lock:
        record = state.get(data["request_id"])
        if record is None or data["type"] != record["kind"] + "_response":
            return {"status": "ignored", "reason": "id_or_type"}
        if (
            recipient != record["sender"]
            or message["recipient"] != recipient
            or message["sender"] != record["target"]
        ):
            return {"status": "ignored", "reason": "role_or_recipient"}
        if record["reply"] != data:
            return {"status": "ignored", "reason": "not_saved_reply"}
        if record["status"] != "pending":
            return {"request_id": data["request_id"], "status": "duplicate"}
        record["status"] = "approved" if data["approve"] else "rejected"
    return {"request_id": data["request_id"], "status": record["status"]}


scenario='approve'
state,team,lock={},create_team(),RLock()
_temp=tempfile.TemporaryDirectory(prefix='s16-mail-');directory=Path(_temp.name)
sender,target,kind,payload='reader','lead','plan','读取预期，再运行main.py。'
actor,approve,feedback='lead',True,'仅批准本次计划，权限另行判断。'
if 'reply_request'!='open_request':
    kind='shutdown' if scenario in ('busy','closed') else 'plan'
    sender,target=('lead','reader') if kind=='shutdown' else ('reader','lead')
    if kind=='shutdown':payload,feedback='本次工作结束，请在空闲边界关闭。','空闲边界确认本次关闭。'
    request_id=open_request(state,team,directory,sender,target,kind,payload,lock)['request_id']
    actor=target
if 'reply_request'=='open_request':
    if scenario=='shutdown':sender,target,kind,payload='lead','reader','shutdown','本次工作结束，请在空闲边界关闭。'
    if scenario=='direction':sender,target='lead','reader'
    if scenario=='pending':open_request(state,team,directory,sender,target,kind,payload,lock)
    if scenario=='limit':
        for i in range(4):state['old-'+str(i)]={'status':'rejected'}
if scenario in ('reject','rejected'):approve=False
if 'reply_request'=='reply_request':
    if scenario=='wrong_role':actor='reviewer'
    if scenario=='busy':team['reader'].update(active=True,status='running')
    if scenario in ('duplicate','conflict'):
        reply_request(state,team,directory,request_id,actor,True,feedback,lock)
        if scenario=='conflict':approve=False
if 'reply_request'=='match_response':
    reply_request(state,team,directory,request_id,actor,approve,feedback,lock)
    message=deepcopy(load_inbox(directory,sender,team)[-1]);recipient=sender
    data=read_protocol(message)
    if scenario=='wrong_type':data['type']='shutdown_response'
    if scenario=='wrong_role':message['sender']='reviewer'
    if scenario=='unknown':data['request_id']='req-404'
    if scenario=='invalid':data['approve']='true'
    if scenario=='forged':data['feedback']='不同的未保存决定'
    message['body']=encode_protocol(data)
    if scenario=='duplicate':match_response(state,message,recipient,lock)
before=deepcopy(state)
_replace=os.replace
def broken_replace(*_):raise OSError('教学保存失败')
if scenario=='save_error':os.replace=broken_replace


# 调用入口
try:
    observed=reply_request(state, team, directory, request_id, actor, approve, feedback, lock)
except (ValueError,OSError) as error:
    observed={"error":str(error)}
finally:
    os.replace=_replace
print(observed)

只允许目标角色决定。
相同回复不重复存信，冲突拒绝。
成员空闲且确认已保存后才关闭。

正常回复先进入请求方邮箱，再写入 record.reply。注意 record.status 此时没有改成 approved。这份记录分别保存“已有发出的决定”和“请求方已经匹配的状态”，读同一个字段无法表达这两个事实。

选择“成员仍在工作”，这是关闭请求：active 仍为 True，函数返回 deferred，没有发确认，也没有把成员改成 closed。真实线程实验会先用 Event 挡住一轮工作，确认关闭被推迟，再让轮次结束，重新处理同一份请求。

空闲且同意关闭时，先存信，再设置成员 closed。这个变更与新轮次的占用使用同一把锁；后来的 run_teammate 会拒绝交付。若选择“回复保存失败”，关闭不会提前发生。拒绝关闭同样保留成员原状态。

本入口在宿主明确的空闲边界处理关闭，没有新增永久等待的线程。第十五课保留的是逻辑成员与自己的历史；本课关闭的也是这个成员的后续交付资格。

match_response：看见回复以后，还要逐项对上

函数接收实际邮箱信封 message、当前消费角色 recipient、请求表与共享锁。read_protocol 先解析前缀和 JSON；普通文本返回 ignored，不从一句“同意”推断协作状态。

接着核对回复字段与布尔决定，再按 request_id 查原记录。类型必须是原 kind 对应的 response，信封发送者必须是原 target，实际收件人必须是原 sender。最后，回复内容还必须等于成功保存的 record.reply。

真实执行轨迹回放；原样协议函数与锁、Worker虚拟邮箱；保存故障为公开教学注入，不调用模型或程序。

Pys16/core/runtime.py · match_response

未运行

def match_response(state, message, recipient, lock):
    """信封角色、类型、ID和已发送决定全部一致，才更新请求状态。"""
    data = read_protocol(message)
    if data is None:
        return {"status": "ignored", "reason": "ordinary_message"}
    if (
        set(data) != {"type", "request_id", "approve", "feedback"}
        or type(data["approve"]) is not bool
        or type(data["feedback"]) is not str
        or len(data["feedback"]) > 120
    ):
        return {"status": "ignored", "reason": "invalid_response"}
    with lock:
        record = state.get(data["request_id"])
        if record is None or data["type"] != record["kind"] + "_response":
            return {"status": "ignored", "reason": "id_or_type"}
        if (
            recipient != record["sender"]
            or message["recipient"] != recipient
            or message["sender"] != record["target"]
        ):
            return {"status": "ignored", "reason": "role_or_recipient"}
        if record["reply"] != data:
            return {"status": "ignored", "reason": "not_saved_reply"}
        if record["status"] != "pending":
            return {"request_id": data["request_id"], "status": "duplicate"}
        record["status"] = "approved" if data["approve"] else "rejected"
    return {"request_id": data["request_id"], "status": record["status"]}

输出

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

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

教学准备与调用入口

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

from copy import deepcopy
from pathlib import Path
from threading import RLock
import json
import re
import os
import tempfile
PREFIX="[team-protocol] "
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 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 create_team():
    return {name: {'role': role, 'history': [], 'active': False, 'status': 'idle', 'turns': 0}
            for name, role in [('lead', '汇总只读证据'), ('reader', '读取源码并保留调查上下文'),
                               ('reviewer', '读取预期并通知reader')]}

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 match_response(state, message, recipient, lock):
    """信封角色、类型、ID和已发送决定全部一致，才更新请求状态。"""
    data = read_protocol(message)
    if data is None:
        return {"status": "ignored", "reason": "ordinary_message"}
    if (
        set(data) != {"type", "request_id", "approve", "feedback"}
        or type(data["approve"]) is not bool
        or type(data["feedback"]) is not str
        or len(data["feedback"]) > 120
    ):
        return {"status": "ignored", "reason": "invalid_response"}
    with lock:
        record = state.get(data["request_id"])
        if record is None or data["type"] != record["kind"] + "_response":
            return {"status": "ignored", "reason": "id_or_type"}
        if (
            recipient != record["sender"]
            or message["recipient"] != recipient
            or message["sender"] != record["target"]
        ):
            return {"status": "ignored", "reason": "role_or_recipient"}
        if record["reply"] != data:
            return {"status": "ignored", "reason": "not_saved_reply"}
        if record["status"] != "pending":
            return {"request_id": data["request_id"], "status": "duplicate"}
        record["status"] = "approved" if data["approve"] else "rejected"
    return {"request_id": data["request_id"], "status": record["status"]}


scenario='approved'
state,team,lock={},create_team(),RLock()
_temp=tempfile.TemporaryDirectory(prefix='s16-mail-');directory=Path(_temp.name)
sender,target,kind,payload='reader','lead','plan','读取预期，再运行main.py。'
actor,approve,feedback='lead',True,'仅批准本次计划，权限另行判断。'
if 'match_response'!='open_request':
    kind='shutdown' if scenario in ('busy','closed') else 'plan'
    sender,target=('lead','reader') if kind=='shutdown' else ('reader','lead')
    if kind=='shutdown':payload,feedback='本次工作结束，请在空闲边界关闭。','空闲边界确认本次关闭。'
    request_id=open_request(state,team,directory,sender,target,kind,payload,lock)['request_id']
    actor=target
if 'match_response'=='open_request':
    if scenario=='shutdown':sender,target,kind,payload='lead','reader','shutdown','本次工作结束，请在空闲边界关闭。'
    if scenario=='direction':sender,target='lead','reader'
    if scenario=='pending':open_request(state,team,directory,sender,target,kind,payload,lock)
    if scenario=='limit':
        for i in range(4):state['old-'+str(i)]={'status':'rejected'}
if scenario in ('reject','rejected'):approve=False
if 'match_response'=='reply_request':
    if scenario=='wrong_role':actor='reviewer'
    if scenario=='busy':team['reader'].update(active=True,status='running')
    if scenario in ('duplicate','conflict'):
        reply_request(state,team,directory,request_id,actor,True,feedback,lock)
        if scenario=='conflict':approve=False
if 'match_response'=='match_response':
    reply_request(state,team,directory,request_id,actor,approve,feedback,lock)
    message=deepcopy(load_inbox(directory,sender,team)[-1]);recipient=sender
    data=read_protocol(message)
    if scenario=='wrong_type':data['type']='shutdown_response'
    if scenario=='wrong_role':message['sender']='reviewer'
    if scenario=='unknown':data['request_id']='req-404'
    if scenario=='invalid':data['approve']='true'
    if scenario=='forged':data['feedback']='不同的未保存决定'
    message['body']=encode_protocol(data)
    if scenario=='duplicate':match_response(state,message,recipient,lock)
before=deepcopy(state)
_replace=os.replace
def broken_replace(*_):raise OSError('教学保存失败')
if scenario=='save_error':os.replace=broken_replace


# 调用入口
try:
    observed=match_response(state, message, recipient, lock)
except (ValueError,OSError) as error:
    observed={"error":str(error)}
finally:
    os.replace=_replace
print(observed)

读取类型化回复。
核对ID、类型、信封角色及已保存决定。
只推进pending，重放不重复更新。

在 record.status 赋值行暂停：所有关联检查已经通过，快照中仍是 pending。下一步才变成 approved 或 rejected；图中状态也在同一步更新。这个函数没有调用模型或程序，返回终态只完成协议关联。

选择“协议类型错误”“信封角色错误”或“未保存的决定”，原请求保持 pending。选择“重复匹配”，请求已经有终态，函数只返回 duplicate。追加日志可以重复消费，不会因此重复执行程序。

consume_responses 是统一消费入口，在原循环的 prepare 边界先处理协议，再沿用 prepare_inbox 将消息作为普通文本接入自己的历史。它保留邮箱日志，并将坏 JSON 记录为 ignored，避免同一封坏信让后续记录永远无法处理。浏览器执行原样函数、锁与 Worker 虚拟邮箱；保存故障来自公开的注入，不执行 SDK 或外部程序。

approved 怎样接到真正的执行入口？

公开 run_member 继续使用唯一的原 Agent 循环。reader 第一轮提交计划，Lead 轮次调用 review_plan，reader 第二轮保留原历史并提出读取与运行请求。工具中的身份由宿主闭包绑定，模型参数里没有 sender 或 actor。

运行入口先用 plan_gate 核对本次 request_id、kind、发起人和 approved，再通过旧 execute_hooked 与 execute_guarded。这里有两条连续的门控，不把计划审批写成用户授权：

python

# user_code/practice.py 的真实执行关系，省略装配参数
_s07._s04.execute_hooked(
    call,
    tools,
    lambda *_: allowed,
    {"before": [plan_gate(state, plan_id, actor, lock)]},
    guard=guard,
)

计划与审批阶段的 guard 只分发各自明确注册的 submit_plan 或 review_plan；工作阶段 guard 为 None，使用原默认权限入口。run_main 仍是固定文件程序，不提供任意 shell。批准属于这份不可改的计划，不能拿别人的 ID 或另一份旧批准放行。

正常实验共七次逻辑模型请求。程序成功执行以后，固定响应根据本次工具结果与读取的预期进行比较；预期改成 43 时，仍会诚实报告“不符”，不会因为协议 approved 就把验收也写成通过。

工作完成后，宿主发关闭请求，并在空闲边界处理确认。本地 closed 时，Lead 请求仍可保持 pending；Lead 自己匹配后才 approved。再次交付返回 unavailable，不会重新启动回调。若不收到回复，当前实现不自动重试或超时，只保留未决状态和有界额度。

用什么证据判断这一课真的接通了？

bash

python3 examples/s16-team-protocols/user_code/verify_practice.py
python3 examples/s16-team-protocols/user_code/verify_fixtures.py

十一组标准库测试核对三条完整执行路径、真实输出与预期不符、保存失败、错 ID/类型/角色、重复与冲突决定、并发回复，以及真实活跃轮次结束后的关闭。允许路径的程序计数文件恰有一个 x，两个拒绝路径都没有该文件。

三个教学区共二十二种输入、364 个真实快照，运行和调试输出一致。可选 verify_sdk.py 以 Anthropic SDK 1.13.0 和 MockTransport 验证三条路径，各八次模拟 HTTP，包含批准回复工具完成之后的一次 529。恢复当前请求没有重复发信，四封协议消息保留，程序仅在双重允许时执行一次；这不是在线模型实验。

读到这里，应该可以独立检查一份协作实现：有没有绑定这次请求，有没有把保存当成匹配，有没有把协作批准当成执行许可，以及关闭时有没有等待真实的空闲边界。