流式 Agent 的进度显示，什么时候才能写完成
让“已完成”有可检查的依据：先确认运行终态，再验收最终产物。
你能带走什么
- 文本片段不签发任务完成
- 单次响应结束仍可能进入下一轮
- 迭代器结束后仍需检查审批与异常
- 应用验收绑定最终候选
- 冲突和缺失证据保守显示未知
1. 先让“完成”这两个字有出处
终端里刚出现“巡检记录已整理”，右下角就亮起完成。几秒后，工具还在返回结果；再过一会儿，运行停在人工批准环节。读者看见的文字没错，界面给出的承诺却走快了。流式显示最难的地方就在这里：消息一段段到达，任务的结论可能要晚很多。
这次做一个完全离线的进度回放器。它读取合成事件，把文本、工具状态、等待批准、取消和应用验收放进同一份显示状态。最后一段文字只负责追加文字；运行结束以后还要单独验收，才能写“已完成”。把夹具中间一行删掉、原样重放一行，或插入另一个 run 的回调，预期显示也都能重新检查。
实验只用 Python 标准库，不调用模型、不连接网络、不执行真实工具，没有外部日志收集器。代码是文章配套的教学程序，不涉及任何网站控件。正文中的 run.ended、app.checked 等名字全部属于本例自定义协议。
2. 先分清三个结束位置
OpenAI 的 Responses 流式事件参考列出 response.output_text.delta、response.output_text.done、response.completed、response.failed 和 response.incomplete。text.done 结束一个文本部分，response.completed 结束一次模型响应。Agent 接下来仍可能执行工具或进入下一次模型调用。以 response.completed 直接点亮整项任务的绿色勾，会把这个范围放大。
Python Agents SDK 的 Streaming 指南要求消费 result.stream_events() 到迭代器结束；最后可见文字之后仍可能有运行收尾。暂停审批同样会结束当前迭代器，待审内容放在 interruptions 中。因此宿主需要在循环结束后检查这些结果，而不是在 print 最后一段文字时给任务盖章。
本文研究的是用户可见状态的证据边界。agent-loop 中的调用执行、重试和预算策略仍由执行器负责；trace 脱敏关心排障信息如何保留；usage 统计核算消耗；结构化最终输出负责验收具体业务字段。本例只给这些结果留入口，不再实现一遍它们。
3. 固定参考版本，避免把自定义事件认成 SDK 接口
本次查阅日期为 2026-10-07。SDK 的源码参考固定在 openai-agents-python 的 v0.23.1 标签，版本记录可回查。没有安装或运行这个 SDK；以下适配器测试使用人工构造的字段投影，不能算 SDK 集成测试。Responses 文档没有随这个 SDK 标签一起冻结，真正接线时还要记录自己的 OpenAI 客户端版本与接口契约。
固定版本的 stream_events.py定义 raw_response_event、run_item_stream_event、agent_updated_stream_event。工具调用项使用 tool_called，工具输出项使用 tool_output。本例只适配普通 function_call / function_call_output；工具调用项一出现，界面可以写“已收到工具调用”，实际开始时间需要工具执行入口或钩子的证据。收到输出项时记“已返回”，不据此声称业务成功。
本例的协议分成两个入口。projected_sdk_event 接收少量合成 SDK 字段投影，把它们翻译成显示事件；owner_finished 接收宿主已经确认的生命周期结果。tool.started 来自本地工具入口，approval.wait 来自待审状态，app.checked 来自应用验收器。它们都不是官方 SSE 名称，也没有所谓自动到来的 agent.completed 事件。
还有一个容易漏掉的细节：固定版本的 result.py提供 run_loop_exception，供消费者在流结束后检查后台运行异常。真实接线需要同时处理迭代器抛错以及这个属性，检查 is_complete、interruptions 和宿主自己的取消记录。error 标志不能根据模型文字填写。
4. 显示状态如何收紧
状态分成三层：运行活动保留在 tools、text 和 waiting；terminal 记录宿主确认的运行结论；verdict 记录应用验收。status 每次从这些事实推导，不让各个回调随手修改一个 completed 布尔值。看到一个模型响应结束只会登记 response_id，既不会结算工具，也不会跳过应用检查。
- running：文字和工具仍可更新，任务尚未完成
- waiting：待审或本轮暂停；审批恢复要建立新的 run 显示实例
- checking：成功候选已经固定，等待应用检查
- done：本轮运行的成功候选通过了本例的本地验收
- failed / incomplete / cancelled：保留各自结论，避免统一写成完成
- unknown：日志存在冲突，或连接结束时没有足够证据；需要核对结果
本地验收规则刻意很小：最终文本必须等于“合成巡检：已检查 2 项”。验收票据带上最终文本的 SHA-256 摘要，避免把旧产物的通过结论套到新产物上。这只验证合成夹具的字符串约定，没有证明巡检事实真实，更没有评估模型质量。真实应用应在自己的验收器中检查预期文件、回执或领域规则。摘要提供一致性绑定，不提供来源认证。
允许工具错误被上层处理。tool.failed 说明工具那一步出错；若执行器最终给出可验收的产物，应用仍可接受。相反，只要还有 requested 或 running 的工具，succeeded 就与当前证据矛盾。本例将它降为 unknown，不偷偷清空未结束的工具。
5. 重复、乱序与跨 run 的处理
每个事件都有 run、id、seq、kind、data。run 在受信任的宿主创建消费者时绑定，不从模型文本或工具返回内容中读取。id 在生产事件时分配，并随重传保持不变。seq 是这份完整本地显示日志的连续序号，从 1 开始；它与 Responses 的 sequence_number 没有互换关系。适配器过滤上游事件以后，应重新分配自己的序号。
完全相同的信封再次到达只增加 duplicates，文字不会追加第二遍。同一个 id 携带不同内容会记录 id_conflict。新 id 带来缺口或旧序号时也保守降级。这份程序没有乱序缓冲器；想允许乱序，需要先在另一层重排并证明没有缺失，再向 reducer 投递完整序列。
跨 run 的事件直接隔离，不污染当前内容。一次暂停恢复可以保留业务 task ID，但新的流消费应分配新的 run，并用独立状态记录。切换界面后旧回调迟到是常见场景：前端关闭旧订阅还不够，归属检查也要保留。这里的隔离只解决显示混流，不能代替鉴权或多租户权限检查。
未知事件采用保守策略：unknown_event 会阻止显示成功。因为无法确定新事件是否意味着新的终止约束，这个小程序宁愿要求核对。生产环境可以维护明确的“仅供观测、可安全忽略”白名单，但不要靠事件名以 done 或 completed 结尾来猜语义。程序也不会依据 token 个数推算百分比；Agent 后续还有多少步骤通常尚未确定。
6. 先跑一遍，再故意删掉完成证据
将下文四个 Python 源码块和两个 JSON 文件分别以 UTF-8 编码保存到同一个空目录，使用 Python 3.10 及以上版本。实测环境为 Python 3.12.14，42 项 unittest 全部通过。运行以下命令不会联网，也不会读取任何凭据。
python3 -B -m unittest -v
python3 -B replay.py normal
python3 -B replay.py token_only
python3 -B replay.py approval_pause
python3 -B replay.py cancel_confirmed
python3 -B replay.py allnormal 的最后三行事件状态依次是 model.completed → running、run.ended → checking、app.checked → done；EOF 仍为 done。token_only 在文字到达时显示“正在处理，尚未完成”，EOF 后变成“完成证据不足，请核对运行结果”。approval_pause 则一直停在等待批准。
可以复制 fixtures.json 中 normal 的一组事件，删除 run.ended 后重放。后面的 app.checked 会因为缺少终态被拒绝，序号缺口也会留下记录。再把同一个 text.delta 信封原样插入，文字只出现一次。若给重复文字换一个新 id 和新 seq，程序会把它视为新内容；因此幂等依赖生产端保留身份，无法从两段相同文字推测它们是不是重传。
7. 完整 reducer：状态只能来自已收到的证据
feed 先检查信封与归属，再去重和检查序号，最后才进入状态分支。close_transport 只记录输入结束，绝不补写 run.ended。issues 一旦出现就保留，后续“成功”不能抹去此前的缺口。
progress.py · 完整源码
"""Original offline display reducer. Event names here are application-owned."""
from dataclasses import dataclass, field
import hashlib
import json
EXPECTED = "合成巡检：已检查 2 项"
def digest(text):
return hashlib.sha256(text.encode("utf-8")).hexdigest()
MESSAGES = {
"running": "正在处理，尚未完成",
"waiting": "等待批准，任务尚未完成",
"cancelling": "已请求取消，正在等待结束确认",
"cancelled": "已取消；已发生的工具效果需另行核对",
"failed": "运行失败，任务未完成",
"incomplete": "响应未完整生成，任务未完成",
"unknown": "完成证据不足，请核对运行结果",
"checking": "运行已结束，等待应用验收",
"rejected": "运行已结束，应用验收未通过",
"done": "已完成：运行结束且本地验收通过",
}
@dataclass
class Progress:
run: str
seen: dict = field(default_factory=dict)
last_seq: int = 0
tools: dict = field(default_factory=dict)
text: str = ""
model_responses: list = field(default_factory=list)
issues: list = field(default_factory=list)
foreign: int = 0
duplicates: int = 0
waiting: bool = False
cancel_requested: bool = False
terminal: str | None = None
final_text: str | None = None
verdict: bool | None = None
closed: bool = False
@property
def status(self):
if self.issues:
return "unknown"
if self.terminal in {"cancelled", "failed", "incomplete"}:
return self.terminal
if self.terminal == "paused" or self.waiting:
return "waiting"
if self.terminal == "succeeded":
return "checking" if self.verdict is None else ("done" if self.verdict else "rejected")
if self.closed:
return "unknown"
return "cancelling" if self.cancel_requested else "running"
@property
def message(self):
return MESSAGES[self.status]
def issue(self, code):
if code not in self.issues:
self.issues.append(code)
def feed(self, event):
# Envelope identity is assigned by the trusted application producer.
# This is not an SSE ID, SDK event schema, or authentication mechanism.
if not isinstance(event, dict) or set(event) != {"run", "id", "seq", "kind", "data"}:
self.issue("bad_envelope")
return
if not all(isinstance(event[k], str) and event[k] for k in ("run", "id", "kind")):
self.issue("bad_envelope")
return
if type(event["seq"]) is not int or event["seq"] < 1 or not isinstance(event["data"], dict):
self.issue("bad_envelope")
return
if event["run"] != self.run:
self.foreign += 1
return
try:
signature = json.dumps(event, sort_keys=True, ensure_ascii=False, allow_nan=False)
except (TypeError, ValueError):
self.issue("not_json")
return
if event["id"] in self.seen:
if self.seen[event["id"]] != signature:
self.issue("id_conflict")
else:
self.duplicates += 1
return
self.seen[event["id"]] = signature
if event["seq"] != self.last_seq + 1:
self.issue("sequence_gap_or_reorder")
self.last_seq = max(self.last_seq, event["seq"])
kind, d = event["kind"], event["data"]
if self.closed:
self.issue("event_after_transport_close")
return
if self.terminal is not None and kind != "app.checked":
self.issue("event_after_terminal")
return
try:
self.apply(kind, d)
except (ValueError, TypeError, KeyError):
self.issue("bad_payload_or_transition")
@staticmethod
def fields(d, names):
if set(d) != set(names):
raise ValueError("fields")
@staticmethod
def nonempty(value):
if not isinstance(value, str) or not value:
raise ValueError("string")
def apply(self, kind, d):
if kind == "text.delta":
self.fields(d, ["text"])
if not isinstance(d["text"], str):
raise ValueError("text")
self.text += d["text"]
elif kind == "model.completed":
self.fields(d, ["response_id"])
self.nonempty(d["response_id"])
self.model_responses.append(d["response_id"])
elif kind in {"tool.requested", "tool.started", "tool.finished", "tool.failed"}:
self.fields(d, ["call_id"])
call = d["call_id"]
self.nonempty(call)
previous = self.tools.get(call)
if kind == "tool.requested":
if previous is not None:
raise ValueError("reused call")
self.tools[call] = "requested"
elif kind == "tool.started":
if previous != "requested":
raise ValueError("missing request")
self.tools[call] = "running"
else:
# A result item can arrive without a local start hook.
if previous not in {"requested", "running"}:
raise ValueError("unmatched result")
self.tools[call] = "returned" if kind == "tool.finished" else "failed"
elif kind == "approval.wait":
self.fields(d, [])
self.waiting = True
elif kind == "cancel.requested":
self.fields(d, [])
self.cancel_requested = True
elif kind == "run.ended":
self.fields(d, ["outcome", "final_text"])
outcome = d["outcome"]
if outcome not in {"succeeded", "paused", "cancelled", "failed", "incomplete"}:
raise ValueError("outcome")
if d["final_text"] is not None and not isinstance(d["final_text"], str):
raise ValueError("final_text")
if outcome == "succeeded":
if self.waiting or self.cancel_requested or d["final_text"] is None:
raise ValueError("contradictory success")
if any(v not in {"returned", "failed"} for v in self.tools.values()):
raise ValueError("unsettled tool")
self.terminal = outcome
self.final_text = d["final_text"]
self.waiting = outcome == "paused"
elif kind == "app.checked":
self.fields(d, ["digest", "ok"])
if self.terminal != "succeeded" or self.verdict is not None:
raise ValueError("check timing")
if type(d["ok"]) is not bool or d["digest"] != digest(self.final_text):
raise ValueError("check binding")
self.verdict = d["ok"]
else:
self.issue("unknown_event")
def close_transport(self):
# EOF alone cannot manufacture a run.ended event.
self.closed = True
def acceptance(progress):
"""Tiny synthetic contract, not a factual/model-quality evaluator."""
if progress.terminal != "succeeded":
raise ValueError("no final candidate")
return {"digest": digest(progress.final_text), "ok": progress.final_text == EXPECTED}
8. 适配层：只演示已经核对过的映射
adapter.py 是可运行的离线字段投影示例，不是可以直接接入线上 SDK 的全量适配器。传给 projected_sdk_event 的 dict 是测试手工构造的最小字段，真实 SDK 对象需要先读取相应属性再投影。response.created、消息项、handoff、搜索及其他工具均未覆盖；这些输入目前会保守触发 adapter.unsupported_event。不要直接接上一个完整官方流再期待它通过。
普通函数工具的 tool_called 映射成 tool.requested；tool_output 映射成 tool.finished，状态值是 returned。真正的执行开始与错误需要宿主另投 tool.started / tool.failed。单次 raw response 的 failed、incomplete 或 error 在这里记为未解决终止事件，阻止成功；如何归类整轮 Agent 的恢复结果，需要真实执行器补充信息。fixture 中 run.ended/incomplete 仅表示一个宿主已经确认不完整的合成结论。
owner_finished 的 exhausted 只能在消费循环正常结束后置真，不能因为本地 for 循环 break 或 UI 取消订阅就置真。runtime_failed 表示已确认的运行失败；网络断开而远端状态不明时应让 exhausted=False，保持未知。取消请求进入 cancelled 分支前仍需等消费者收尾。本例选择保守处理完成与取消的竞争：只要这次消费已经登记取消请求，就不再签发成功候选。
adapter.py · 完整源码
"""Offline protocol sketch, not a complete OpenAI SDK integration.
Input dictionaries are synthetic projections of documented event fields.
SDK source reference: openai-agents-python tag v0.23.1.
"""
from dataclasses import dataclass
@dataclass
class Producer:
run: str
seq: int = 0
def emit(self, kind, **data):
self.seq += 1
return {"run": self.run, "id": f"{self.run}:{self.seq}",
"seq": self.seq, "kind": kind, "data": data}
def projected_sdk_event(self, event):
"""Only a small allowlist; caller must project actual SDK objects first."""
if event.get("type") == "raw_response_event":
raw = event["data"]
t = raw.get("type")
if t == "response.output_text.delta":
return self.emit("text.delta", text=raw["delta"])
if t == "response.completed":
return self.emit("model.completed", response_id=raw["response"]["id"])
# Failures are handled by owner_finished, not guessed from text.
if t in {"response.failed", "response.incomplete", "error"}:
return self.emit("adapter.unresolved_terminal")
elif event.get("type") == "run_item_stream_event":
name = event.get("name")
raw_item = event["item"]["raw_item"]
# This demonstration covers ordinary function tools only.
if name == "tool_called" and raw_item.get("type") == "function_call":
return self.emit("tool.requested", call_id=raw_item["call_id"])
if name == "tool_output" and raw_item.get("type") == "function_call_output":
return self.emit("tool.finished", call_id=raw_item["call_id"])
return self.emit("adapter.unsupported_event")
def owner_finished(self, *, exhausted, is_complete, interruptions,
cancel_requested, runtime_failed, final_text):
"""Application lifecycle observation, never an official wire event.
Called after consuming the iterator. All arguments are synthetic here.
runtime_failed means a confirmed execution failure, including a non-None
v0.23.1 result.run_loop_exception after draining. A socket disconnect
alone must use exhausted=False, not runtime_failed=True.
"""
if runtime_failed:
return self.emit("run.ended", outcome="failed", final_text=None)
if not exhausted or not is_complete:
return None
if cancel_requested:
outcome = "cancelled"
elif interruptions:
outcome = "paused"
elif final_text is not None:
outcome = "succeeded"
else:
return None
return self.emit("run.ended", outcome=outcome, final_text=final_text)
9. 把每组故障的预期显示写在夹具里
fixtures.json 有 30 组确定输入。expected_steps 逐个规定事件后的状态，expected_final 规定 EOF 后的状态。预期值手工填写，没有调用 reducer 生成；测试同时核对重复计数、跨 run 计数和指定工具状态，避免只验证“没有抛异常”。
| 夹具 | 事件后状态 → EOF | 要暴露的问题 |
|---|---|---|
| token_only | running / EOF: unknown | 最后一段文字到达后断流，不能写完成 |
| model_only | running → running / EOF: unknown | 单次 Responses 结束，Agent 仍缺少整体终止证据 |
| normal | running → running → running → running → running → checking → done / EOF: done | 工具返回后等运行结束，再验收 |
| no_verdict | checking / EOF: checking | 运行结束但应用验收尚未执行 |
| rejected | checking → rejected / EOF: rejected | 业务验收失败不能重标为成功 |
| approval_pause | waiting → waiting / EOF: waiting | 迭代器正常结束也可能只是暂停 |
| approval_from_snapshot | waiting / EOF: waiting | 无需假造 Python SDK 的通用审批事件 |
| cancel_pending | cancelling / EOF: unknown | 取消请求没有结束确认，EOF 后结果未知 |
| cancel_confirmed | running → running → cancelling → cancelled / EOF: cancelled | 取消不抹掉工具已开始的事实 |
| runtime_error | running → failed / EOF: failed | 宿主捕获的明确运行错误 |
| response_incomplete | running → incomplete / EOF: incomplete | 单响应宿主确认 incomplete 的合成生命周期结果 |
| empty_eof | / EOF: unknown | 连接关闭而没有任何完成证据 |
| duplicate_delta | running → running → checking → done / EOF: done | 相同信封重放一次，文本只追加一次 |
| duplicate_final | checking → done → done / EOF: done | 同一验收事件的原样重传仍幂等 |
| id_conflict | running → unknown → unknown → unknown / EOF: unknown | 同 ID 不同内容阻止成功 |
| unknown_event | unknown → unknown → unknown / EOF: unknown | 未知事件名不会被后缀 completed 猜成成功 |
| foreign_run | running → running → checking → done / EOF: done | 旧运行回调不修改新运行 |
| tool_unsettled | running → running → unknown → unknown / EOF: unknown | 有未结束工具时拒绝成功证据 |
| tool_return_only | unknown / EOF: unknown | 缺少对应调用项 |
| tool_no_start_hook | running → running → checking → done / EOF: done | 收到结果足以记录已返回，不补造开始时刻 |
| tool_failure_recovered | running → running → running → checking → done / EOF: done | 工具错误可被上层处理，本例验收独立判定 |
| premature_check | unknown → unknown / EOF: unknown | 提前到达的应用通过票据无效 |
| wrong_digest | checking → unknown / EOF: unknown | 验收必须绑定当前最终候选 |
| late_delta | checking → done → unknown / EOF: unknown | 终止后新事件提示协议冲突，撤下完成标签 |
| pause_then_success | waiting → unknown → unknown / EOF: unknown | 恢复需要新 run 实例，不能在暂停态硬塞成功 |
| cancel_then_success | cancelling → unknown → unknown / EOF: unknown | 本例保守处理取消与成功竞争 |
| sequence_gap | running → unknown → unknown / EOF: unknown | 自有完整日志的序号出现缺口 |
| sequence_reorder | running → unknown / EOF: unknown | 新 ID 携带旧序号 |
| bad_payload | unknown / EOF: unknown | 文本字段类型错误 |
| bad_envelope | unknown / EOF: unknown | 拒绝信封混入额外字段 |
下面给出完整 fixtures.json，共 30 组输入及逐步预期。复制整个代码块保存为同名文件，不要只保留 normal 一组。随后保存 expected-messages.json，其中每个状态都绑定固定中文显示消息；测试会逐字比较这些消息。
fixtures.json · 完整源码
[
{
"name": "token_only",
"note": "最后一段文字到达后断流，不能写完成",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "text.delta",
"data": {
"text": "合成"
}
}
],
"expected_steps": [
"running"
],
"expected_final": "unknown",
"assertions": {
"text": "合成"
}
},
{
"name": "model_only",
"note": "单次 Responses 结束，Agent 仍缺少整体终止证据",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "text.delta",
"data": {
"text": "合成"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "model.completed",
"data": {
"response_id": "resp-demo-1"
}
}
],
"expected_steps": [
"running",
"running"
],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "normal",
"note": "工具返回后等运行结束，再验收",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "text.delta",
"data": {
"text": "合成"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "tool.requested",
"data": {
"call_id": "call-1"
}
},
{
"run": "run-A",
"id": "run-A:3",
"seq": 3,
"kind": "tool.started",
"data": {
"call_id": "call-1"
}
},
{
"run": "run-A",
"id": "run-A:4",
"seq": 4,
"kind": "tool.finished",
"data": {
"call_id": "call-1"
}
},
{
"run": "run-A",
"id": "run-A:5",
"seq": 5,
"kind": "model.completed",
"data": {
"response_id": "resp-demo-1"
}
},
{
"run": "run-A",
"id": "run-A:6",
"seq": 6,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:7",
"seq": 7,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
}
],
"expected_steps": [
"running",
"running",
"running",
"running",
"running",
"checking",
"done"
],
"expected_final": "done",
"assertions": {}
},
{
"name": "no_verdict",
"note": "运行结束但应用验收尚未执行",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
}
],
"expected_steps": [
"checking"
],
"expected_final": "checking",
"assertions": {}
},
{
"name": "rejected",
"note": "业务验收失败不能重标为成功",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：只有 1 项"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "app.checked",
"data": {
"digest": "cc95ce132ddc0f0e813c53273743a4aa3b6fe6c6af8b047f7f2cea623f4bf139",
"ok": false
}
}
],
"expected_steps": [
"checking",
"rejected"
],
"expected_final": "rejected",
"assertions": {}
},
{
"name": "approval_pause",
"note": "迭代器正常结束也可能只是暂停",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "approval.wait",
"data": {}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "run.ended",
"data": {
"outcome": "paused",
"final_text": null
}
}
],
"expected_steps": [
"waiting",
"waiting"
],
"expected_final": "waiting",
"assertions": {}
},
{
"name": "approval_from_snapshot",
"note": "无需假造 Python SDK 的通用审批事件",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "run.ended",
"data": {
"outcome": "paused",
"final_text": null
}
}
],
"expected_steps": [
"waiting"
],
"expected_final": "waiting",
"assertions": {}
},
{
"name": "cancel_pending",
"note": "取消请求没有结束确认，EOF 后结果未知",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "cancel.requested",
"data": {}
}
],
"expected_steps": [
"cancelling"
],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "cancel_confirmed",
"note": "取消不抹掉工具已开始的事实",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "tool.requested",
"data": {
"call_id": "call-1"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "tool.started",
"data": {
"call_id": "call-1"
}
},
{
"run": "run-A",
"id": "run-A:3",
"seq": 3,
"kind": "cancel.requested",
"data": {}
},
{
"run": "run-A",
"id": "run-A:4",
"seq": 4,
"kind": "run.ended",
"data": {
"outcome": "cancelled",
"final_text": null
}
}
],
"expected_steps": [
"running",
"running",
"cancelling",
"cancelled"
],
"expected_final": "cancelled",
"assertions": {
"tools": {
"call-1": "running"
}
}
},
{
"name": "runtime_error",
"note": "宿主捕获的明确运行错误",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "text.delta",
"data": {
"text": "合成"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "run.ended",
"data": {
"outcome": "failed",
"final_text": null
}
}
],
"expected_steps": [
"running",
"failed"
],
"expected_final": "failed",
"assertions": {}
},
{
"name": "response_incomplete",
"note": "单响应宿主确认 incomplete 的合成生命周期结果",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "text.delta",
"data": {
"text": "合成"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "run.ended",
"data": {
"outcome": "incomplete",
"final_text": null
}
}
],
"expected_steps": [
"running",
"incomplete"
],
"expected_final": "incomplete",
"assertions": {}
},
{
"name": "empty_eof",
"note": "连接关闭而没有任何完成证据",
"run": "run-A",
"events": [],
"expected_steps": [],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "duplicate_delta",
"note": "相同信封重放一次，文本只追加一次",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "text.delta",
"data": {
"text": "合成"
}
},
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "text.delta",
"data": {
"text": "合成"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:3",
"seq": 3,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
}
],
"expected_steps": [
"running",
"running",
"checking",
"done"
],
"expected_final": "done",
"assertions": {
"text": "合成",
"duplicates": 1
}
},
{
"name": "duplicate_final",
"note": "同一验收事件的原样重传仍幂等",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
}
],
"expected_steps": [
"checking",
"done",
"done"
],
"expected_final": "done",
"assertions": {
"duplicates": 1
}
},
{
"name": "id_conflict",
"note": "同 ID 不同内容阻止成功",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "text.delta",
"data": {
"text": "合成"
}
},
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "text.delta",
"data": {
"text": "被替换"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:3",
"seq": 3,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
}
],
"expected_steps": [
"running",
"unknown",
"unknown",
"unknown"
],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "unknown_event",
"note": "未知事件名不会被后缀 completed 猜成成功",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "future.completed",
"data": {}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:3",
"seq": 3,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
}
],
"expected_steps": [
"unknown",
"unknown",
"unknown"
],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "foreign_run",
"note": "旧运行回调不修改新运行",
"run": "run-A",
"events": [
{
"run": "run-B",
"id": "run-B:1",
"seq": 1,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "text.delta",
"data": {
"text": "合成"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:3",
"seq": 3,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
}
],
"expected_steps": [
"running",
"running",
"checking",
"done"
],
"expected_final": "done",
"assertions": {
"foreign": 1,
"text": "合成"
}
},
{
"name": "tool_unsettled",
"note": "有未结束工具时拒绝成功证据",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "tool.requested",
"data": {
"call_id": "call-1"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "tool.started",
"data": {
"call_id": "call-1"
}
},
{
"run": "run-A",
"id": "run-A:3",
"seq": 3,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:4",
"seq": 4,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
}
],
"expected_steps": [
"running",
"running",
"unknown",
"unknown"
],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "tool_return_only",
"note": "缺少对应调用项",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "tool.finished",
"data": {
"call_id": "call-1"
}
}
],
"expected_steps": [
"unknown"
],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "tool_no_start_hook",
"note": "收到结果足以记录已返回，不补造开始时刻",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "tool.requested",
"data": {
"call_id": "call-1"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "tool.finished",
"data": {
"call_id": "call-1"
}
},
{
"run": "run-A",
"id": "run-A:3",
"seq": 3,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:4",
"seq": 4,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
}
],
"expected_steps": [
"running",
"running",
"checking",
"done"
],
"expected_final": "done",
"assertions": {}
},
{
"name": "tool_failure_recovered",
"note": "工具错误可被上层处理，本例验收独立判定",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "tool.requested",
"data": {
"call_id": "call-1"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "tool.started",
"data": {
"call_id": "call-1"
}
},
{
"run": "run-A",
"id": "run-A:3",
"seq": 3,
"kind": "tool.failed",
"data": {
"call_id": "call-1"
}
},
{
"run": "run-A",
"id": "run-A:4",
"seq": 4,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:5",
"seq": 5,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
}
],
"expected_steps": [
"running",
"running",
"running",
"checking",
"done"
],
"expected_final": "done",
"assertions": {
"tools": {
"call-1": "failed"
}
}
},
{
"name": "premature_check",
"note": "提前到达的应用通过票据无效",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
}
],
"expected_steps": [
"unknown",
"unknown"
],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "wrong_digest",
"note": "验收必须绑定当前最终候选",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "app.checked",
"data": {
"digest": "c5bfad0a397846efcc0985125641c02f9b43844b6e1b33bed9e1bea449c612f1",
"ok": true
}
}
],
"expected_steps": [
"checking",
"unknown"
],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "late_delta",
"note": "终止后新事件提示协议冲突，撤下完成标签",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
},
{
"run": "run-A",
"id": "run-A:3",
"seq": 3,
"kind": "text.delta",
"data": {
"text": "合成"
}
}
],
"expected_steps": [
"checking",
"done",
"unknown"
],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "pause_then_success",
"note": "恢复需要新 run 实例，不能在暂停态硬塞成功",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "approval.wait",
"data": {}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:3",
"seq": 3,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
}
],
"expected_steps": [
"waiting",
"unknown",
"unknown"
],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "cancel_then_success",
"note": "本例保守处理取消与成功竞争",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "cancel.requested",
"data": {}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 2,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:3",
"seq": 3,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
}
],
"expected_steps": [
"cancelling",
"unknown",
"unknown"
],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "sequence_gap",
"note": "自有完整日志的序号出现缺口",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "text.delta",
"data": {
"text": "合成"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 3,
"kind": "run.ended",
"data": {
"outcome": "succeeded",
"final_text": "合成巡检：已检查 2 项"
}
},
{
"run": "run-A",
"id": "run-A:3",
"seq": 4,
"kind": "app.checked",
"data": {
"digest": "5ad405cc35fd834b36815b3a56fb80adf9068b6fea2aa2188db2b00fa27a74bc",
"ok": true
}
}
],
"expected_steps": [
"running",
"unknown",
"unknown"
],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "sequence_reorder",
"note": "新 ID 携带旧序号",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "text.delta",
"data": {
"text": "合成"
}
},
{
"run": "run-A",
"id": "run-A:2",
"seq": 1,
"kind": "text.delta",
"data": {
"text": "巡检"
}
}
],
"expected_steps": [
"running",
"unknown"
],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "bad_payload",
"note": "文本字段类型错误",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "text.delta",
"data": {
"text": 42
}
}
],
"expected_steps": [
"unknown"
],
"expected_final": "unknown",
"assertions": {}
},
{
"name": "bad_envelope",
"note": "拒绝信封混入额外字段",
"run": "run-A",
"events": [
{
"run": "run-A",
"id": "run-A:1",
"seq": 1,
"kind": "text.delta",
"data": {
"text": "合成"
},
"unexpected": true
}
],
"expected_steps": [
"unknown"
],
"expected_final": "unknown",
"assertions": {}
}
]
expected-messages.json · 完整源码
{
"running": "正在处理，尚未完成",
"waiting": "等待批准，任务尚未完成",
"cancelling": "已请求取消，正在等待结束确认",
"cancelled": "已取消；已发生的工具效果需另行核对",
"failed": "运行失败，任务未完成",
"incomplete": "响应未完整生成，任务未完成",
"unknown": "完成证据不足，请核对运行结果",
"checking": "运行已结束，等待应用验收",
"rejected": "运行已结束，应用验收未通过",
"done": "已完成：运行结束且本地验收通过"
}
10. 文本回放入口与全部测试
replay.py 逐条打印事件名称、推导状态和固定中文消息。正文片段只保存在 state.text 中，正文和异常文本不直接输出。事件 id、kind 与夹具 name、note 仍原样打印，因此这里只适用于受信任的合成夹具，尚未做终端控制字符净化。本例也不提供日志大小限制、持久化去重表或不可信上传入口。
replay.py · 完整源码
"""Deterministic replay. No clocks, network, SDK, model or credentials."""
from pathlib import Path
import argparse,json
from progress import Progress
def replay(case):
state = Progress(case['run'])
lines = []
for e in case['events']:
state.feed(e)
lines.append(f"{e['id']} {e['kind']} => {state.status} | {state.message}")
state.close_transport()
lines.append(f"EOF => {state.status} | {state.message}")
return state,lines
def main():
cases=json.loads(Path(__file__).with_name('fixtures.json').read_text(encoding='utf-8'))
parser=argparse.ArgumentParser()
parser.add_argument('case',nargs='?',default='normal',choices=['all']+[c['name'] for c in cases])
args=parser.parse_args()
for c in cases:
if args.case in ('all',c['name']):
state,lines=replay(c)
print(f"\n[{c['name']}] {c['note']}")
print('\n'.join(lines))
print('issues='+','.join(state.issues))
if __name__=='__main__':main()
test_progress.py · 完整源码
from pathlib import Path
import json,unittest
from progress import Progress,EXPECTED,acceptance
from adapter import Producer
CASES=json.loads(Path(__file__).with_name('fixtures.json').read_text(encoding='utf-8'))
EXPECTED_MESSAGES=json.loads(Path(__file__).with_name('expected-messages.json').read_text(encoding='utf-8'))
class FixtureTests(unittest.TestCase):
pass
def make_test(case):
def test(self):
p=Progress(case['run'])
self.assertEqual(len(case['events']),len(case['expected_steps']))
for e,expected in zip(case['events'],case['expected_steps']):
p.feed(e)
self.assertEqual(p.status,expected,(case['name'],e))
self.assertEqual(p.message,EXPECTED_MESSAGES[expected])
p.close_transport()
self.assertEqual(p.status,case['expected_final'])
self.assertEqual(p.message,EXPECTED_MESSAGES[case['expected_final']])
for k,v in case['assertions'].items():
self.assertEqual(getattr(p,k),v)
return test
for case in CASES:
setattr(FixtureTests,'test_'+case['name'],make_test(case))
class AdapterTests(unittest.TestCase):
def test_response_complete_is_only_model_complete(self):
e=Producer('r').projected_sdk_event({'type':'raw_response_event','data':{'type':'response.completed','response':{'id':'resp-1'}}})
p=Progress('r');p.feed(e)
self.assertEqual(p.status,'running')
self.assertIsNone(p.terminal)
def test_text_mapping(self):
e=Producer('r').projected_sdk_event({'type':'raw_response_event','data':{'type':'response.output_text.delta','delta':'x'}})
self.assertEqual(e['kind'],'text.delta')
def test_tool_items_do_not_claim_actual_start(self):
prod=Producer('r');p=Progress('r')
for name,t in [('tool_called','function_call'),('tool_output','function_call_output')]:
p.feed(prod.projected_sdk_event({'type':'run_item_stream_event','name':name,'item':{'raw_item':{'type':t,'call_id':'c'}}}))
self.assertEqual(p.tools,{'c':'returned'})
self.assertEqual(p.status,'running')
def test_unknown_sdk_event_is_conservative(self):
p=Progress('r');p.feed(Producer('r').projected_sdk_event({'type':'future'}))
self.assertEqual(p.status,'unknown')
def test_raw_failure_needs_owner_classification(self):
for t in ('response.failed','response.incomplete','error'):
p=Progress('r');p.feed(Producer('r').projected_sdk_event({'type':'raw_response_event','data':{'type':t}}))
self.assertEqual(p.status,'unknown')
def snapshot(self,**changes):
d=dict(exhausted=True,is_complete=True,interruptions=False,cancel_requested=False,runtime_failed=False,final_text=EXPECTED)
d.update(changes)
return Producer('r').owner_finished(**d)
def test_no_exhaustion_no_success(self):
self.assertIsNone(self.snapshot(exhausted=False))
def test_not_complete_no_success(self):
self.assertIsNone(self.snapshot(is_complete=False))
def test_approval_snapshot(self):
self.assertEqual(self.snapshot(interruptions=True,final_text=None)['data']['outcome'],'paused')
def test_cancel_and_error_snapshot(self):
self.assertEqual(self.snapshot(cancel_requested=True)['data']['outcome'],'cancelled')
self.assertEqual(self.snapshot(runtime_failed=True)['data']['outcome'],'failed')
def test_missing_final_candidate(self):
self.assertIsNone(self.snapshot(final_text=None))
def test_acceptance_binds_final_candidate(self):
prod=Producer('r');p=Progress('r')
p.feed(prod.emit('run.ended',outcome='succeeded',final_text=EXPECTED))
p.feed(prod.emit('app.checked',**acceptance(p)))
self.assertEqual(p.status,'done')
def test_post_close_event(self):
p=Progress('r');p.close_transport()
p.feed(Producer('r').emit('run.ended',outcome='succeeded',final_text=EXPECTED))
self.assertEqual(p.status,'unknown')
self.assertIn('event_after_transport_close',p.issues)
if __name__=='__main__':unittest.main()
文件编码与终端输出要分开验证
2026-10-20 复跑发现，旧版 replay.py 与 test_progress.py 的三处 read_text() 依赖进程默认编码。在 Linux、Python 3.12.14 中将默认编码确认为 ASCII，并关闭 UTF-8 模式后，两个入口都会在读取含中文的 fixtures.json 时抛出 UnicodeDecodeError，尚未进入状态回放。现在三处均明确使用 encoding='utf-8'；reducer、30 组夹具和验收规则没有改变。Path.read_text 的编码参数与文本文件读取规则一致，源码文件本身能被 Python 正确读取，不代表另一个 JSON 文件也会自动按 UTF-8 解码。
在下载包根目录可用下面的 Linux 命令检查这个边界。第一条应显示 ANSI_X3.4-1968、0、utf-8，分别是本次环境的 locale 编码、关闭的 UTF-8 模式和标准输出编码。Python 的启动环境说明解释了为何要同时关闭 locale 自动转换和 UTF-8 模式。PYTHONIOENCODING 这里只保证中文输出可写入 UTF-8 管道，不会替 JSON 文件指定编码。
env LC_ALL=C PYTHONCOERCECLOCALE=0 PYTHONUTF8=0 PYTHONIOENCODING=utf-8 python -B -c "import locale,sys; print(locale.getencoding(),sys.flags.utf8_mode,sys.stdout.encoding)"
env LC_ALL=C PYTHONCOERCECLOCALE=0 PYTHONUTF8=0 PYTHONIOENCODING=utf-8 python -B -m unittest discover -s original -p "test_progress.py" -v
env LC_ALL=C PYTHONCOERCECLOCALE=0 PYTHONUTF8=0 PYTHONIOENCODING=utf-8 python -B original/replay.py all修正后的正文源码与 ZIP 解压副本，在普通环境和上述 ASCII 环境、普通与 -O 模式下都通过原有 42 项测试；all 回放的输出逐字节一致。另将三处编码参数各自单独去掉，在普通与 -O 模式下共运行六个反例，均重新触发 UnicodeDecodeError。这六次用于确认回归检查能发现遗漏，不计入 42 项状态机测试。验证仅在 Linux、Python 3.12.14 完成，没有测试 Windows、终端字体显示或其他 Python 版本；上面的 locale.getencoding() 探针需 Python 3.11 及以上。SSE 字节分帧与 SDK 集成的未验范围仍保持不变。
11. 取消和断流以后，界面还欠用户什么
取消按钮按下后先显示“已请求取消”，直到宿主确认本轮已停。即便最终显示“已取消”，先前开始的工具也可能已经产生效果；取消回调无法证明远端回滚。例子特意保留 call-1 的 running 状态，提醒后续需要查真实业务回执。本文没有实现取消执行器、补偿事务或安全重试。
连接断开时，文字可以留着供用户阅读，但应显示完成证据不足。若执行器支持查询，可以按稳定的业务标识去查最终结果，再在一个明确的对账流程中更新界面；不要简单重跑一个可能带副作用的工具。这里的 replay 在 EOF 后关闭输入，不接受补发事件，目的是让这份离线日志有清楚的边界。真实重连需要另行定义恢复游标和权威快照协议。
如果 done 之后又收到新的非重复事件，本例会撤下完成标签并显示 unknown。这个选择揭示了协议矛盾，也方便测试发现生产端在错误的时间签发终态。它不能保证网络永远不会乱序；只能保证当前 reducer 不掩盖自己已经观察到的矛盾。
12. 接入前保留这份验证清单
- 锁定实际安装的 Agents SDK 与 OpenAI 客户端版本，针对真实对象测试字段投影；本稿只核对 v0.23.1 源码
- 消费到 SDK 迭代器结束，处理迭代异常并检查固定版本的 run_loop_exception，随后读取 interruptions 与最终候选
- 把工具请求项、实际开始和业务结果分开记录；覆盖并发工具、嵌套 Agent、handoff 与恢复审批后的新 run
- 确认稳定 ID 在重传时不会重新生成；为实际传输实现重排、背压、日志上限和恢复游标
- 把验收器放在受信任的应用侧，绑定最终候选；通过标准要覆盖真实任务，不照搬合成字符串比较
- 拿真实断流、取消竞争和 SDK 升级做集成测试，再决定哪些未知事件能够安全忽略
离线回放已经验证的是显示逻辑与这些合成序列。真实 SSE 传输、SDK 对象、模型行为、实际取消清理、工具副作用和外部系统可靠性都没有测试。把这个范围写清楚，后面接线时才知道哪些证据已经有了，哪些还需要补。
13. 补充：字节输入要先还原为完整的 SSE 事件
前文的 42 项测试从已经整理好的字典开始。如果输入来自字节流，还需要检查一层更早的边界：一次读取可能停在“你”的三个 UTF-8 字节中间，也可能把一个 CRLF 分到两次读取里。把每个 chunk 单独解码，或者直接交给 json.loads，都可能在完整事件到达之前失败。
下面是 2026 年 10 月 13 日验证的独立输入实验。它只把固定字节还原为带有 event、data、id 的字典，暂不解析 data 里的 JSON，也不接入前文 reducer。原有状态合同与 42 项字典测试保持不变；新增证据只覆盖这份教学分帧代码。
先比较三个很短的输入。retry: 3000 后跟空行，只更新本例保存的 retry_ms，没有 data 字段，因此不会分派数据事件。data: 后跟空行则会分派一个事件，data 的值是空字符串。data: {"done":true} 后只有一个换行，到 EOF 时还缺结束事件的空行，所以不会分派，即使大括号里的 JSON 已经完整。
第四个演示故意把中文、emoji 和两对 CRLF 切开。skip_lf 只记住前一个字符是 CR：下一块若以 LF 开头，就消费掉这一半配对；若是别的字符，继续正常处理。这里第四块末尾的 CR 已经结束空行并分派事件，第五块的 LF 不会再分派一次。
demo_sse.py
"""Fixed inputs only. Print the complete framed result, without parsing JSON."""
import json
from sse_fixture import decode
CASES = [
('metadata_only', [b'retry: 3000\nid: cursor-1\nevent: status\n\n']),
('empty_data', [b'data:\n\n']),
('json_at_eof', [b'data: {"done":true}\n']),
('split_unicode_crlf', [
b'data: \xe4',
b'\xbd\xa0\xe5\xa5\xbd\xf0\x9f',
b'\x99\x82\r',
b'\ndata: second\r\n\r',
b'\n',
]),
]
if __name__ == '__main__':
for name, chunks in CASES:
print(name + ' => ' + json.dumps(decode(chunks), ensure_ascii=False))
运行 python src/demo_sse.py，完整固定输出如下。JSON 中的 \n 表示 data 内的一次换行。
expected-demo.txt
metadata_only => {"events": [], "retry_ms": 3000}
empty_data => {"events": [{"event": "message", "data": "", "id": ""}], "retry_ms": null}
json_at_eof => {"events": [], "retry_ms": null}
split_unicode_crlf => {"events": [{"event": "message", "data": "你好🙂\nsecond", "id": ""}], "retry_ms": null}
14. 分开保存 UTF-8、行与 data 的未完成状态
FixtureDecoder 依次维护三层状态。增量 UTF-8 解码器暂存尚不完整的字符；line 暂存尚未结束的行；data 保存本事件已经读完的 data 行。CR、LF 和 CRLF 都能结束一行，只有空行才触发分派。多条 data 用一个 LF 连接，字段名只认精确的小写写法，冒号后的一个可选空格会移除，其余内容保留。
这里用列表保存 data，因此 [] 与 [''] 可以区分。前者表示没有出现 data 字段；后者表示出现过一条值为空的 data。若改成 if payload.strip()，后者会被误删，单个空格这样的有效文本也会消失。JSON 是下一层的应用格式；一个正确分帧的空字符串能否作为业务消息接收，仍由该格式决定。
本例选用严格增量 utf-8，再仅移去文本开头的一个 U+FEFF。这样即使 BOM 的三个字节分开到达，也能先完成解码；流中的 U+FEFF 则原样保留。finish 用 final=True 要求解码器处理尾部残留字节，然后丢弃尚未由空行结束的行和事件，不人工追加换行。
严格拒绝非法字节是本实验的应用政策。浏览器 EventSource 使用的 UTF-8 解码有替换错误处理，不能把本例的报错结果称作浏览器一致性。后面的政策检查专门包括 EOF 处只有 EF 或 EF BB 的截断前缀，避免它们被当成已经处理完的开头标记。
sse_fixture.py
"""Small SSE byte-framing lab for trusted fixtures, not a network client.
Strict UTF-8 is an application choice, not browser replacement-error behavior.
No size limits, reconnect, JSON parsing, or Agent lifecycle decisions.
"""
import codecs
class FixtureDecoder:
def __init__(self):
self.utf8 = codecs.getincrementaldecoder('utf-8')(errors='strict')
self.at_start = True
self.line = []
self.skip_lf = False
self.data = []
self.event_type = ''
self.id_buffer = ''
self.retry_ms = None
self.events = []
self.closed = False
def feed(self, chunk):
if self.closed:
raise RuntimeError('closed')
self._text(self.utf8.decode(chunk, final=False))
def _text(self, text):
for char in text:
if self.at_start:
self.at_start = False
if char == '\ufeff':
continue
if self.skip_lf:
self.skip_lf = False
if char == '\n':
continue
if char in '\r\n':
self._line(''.join(self.line))
self.line.clear()
self.skip_lf = char == '\r'
else:
self.line.append(char)
def _line(self, line):
if not line:
if self.data:
self.events.append({'event': self.event_type or 'message',
'data': '\n'.join(self.data),
'id': self.id_buffer})
self.data.clear()
self.event_type = ''
return
if line.startswith(':'):
return
field, _, value = line.partition(':')
if value.startswith(' '):
value = value[1:]
if field == 'data':
self.data.append(value)
elif field == 'event':
self.event_type = value
elif field == 'id' and '\0' not in value:
self.id_buffer = value
elif field == 'retry' and value and all('0' <= c <= '9' for c in value):
self.retry_ms = int(value)
def finish(self):
if self.closed:
raise RuntimeError('closed')
self.closed = True
self._text(self.utf8.decode(b'', final=True))
# EOF is not a blank line: drop an unfinished event, even valid JSON.
self.line.clear()
self.data.clear()
self.event_type = ''
return {'events': self.events, 'retry_ms': self.retry_ms}
def decode(chunks):
parser = FixtureDecoder()
for chunk in chunks:
parser.feed(chunk)
return parser.finish()
15. 用固定预期检查分块，再让五种错误策略现形
fixtures.json 把 16 份文本输入和完整输出一起写死，不从分帧器的运行结果生成预期。测试将文本编码为 UTF-8，每份分别按整块、每字节一块，以及每个单切点的双块输入重跑。包括首尾空块在内，一共得到 540 次判定；部分切法重复，数量也不代表所有可能的多切点组合。
夹具覆盖中文和 emoji、多条 data、三种行结束、注释、未知字段、字段大小写、起始与内部 BOM，以及 EOF 未结束的末事件。id 会跨事件保留，空 id 会重置，含 NUL 的 id 不覆盖旧值；retry 只把本例遇到的非空 ASCII 数字保存成整数。只有 id、retry、event 的块可以改变这些缓冲内容，但没有 data 就不会产生数据事件。event 类型在空行后重置。
测试还实际运行五种错误做法：逐块独立解码会在“你”的中间报错；只按 LF LF 切块会漏掉两个 CR-only 事件；EOF 补空行会凭空产生一个事件；strip 过滤会删掉真实空 data；把元数据块送进 JSON 解析会触发 JSONDecodeError。每项都有明确的失败或差异判定。检查使用显式异常，不依赖会被 python -O 移除的 assert。
另外五项只检查本例自己的政策：非法 UTF-8、普通字符的截断尾字节、EF 与 EF BB 两种截断前缀，以及 finish 后的新输入。这些输入与正常分帧矩阵分开计数，不把应用拒绝政策混成协议覆盖率。
fixtures.json
[
{"name":"unicode_multiline_crlf","wire":"event: text\r\ndata: 你好🙂\r\ndata: 第二行\r\n\r\n","expected":{"events":[{"event":"text","data":"你好🙂\n第二行","id":""}],"retry_ms":null}},
{"name":"meta_only_then_data","wire":"retry: 3000\r\nid: cursor-1\r\nevent: ignored\r\n\r\ndata: ok\r\n\r\n","expected":{"events":[{"event":"message","data":"ok","id":"cursor-1"}],"retry_ms":3000}},
{"name":"empty_data_is_event","wire":"data:\n\n","expected":{"events":[{"event":"message","data":"","id":""}],"retry_ms":null}},
{"name":"whitespace_is_data","wire":"data: \n\n","expected":{"events":[{"event":"message","data":" ","id":""}],"retry_ms":null}},
{"name":"json_without_final_blank","wire":"data: {\"done\":true}\n","expected":{"events":[],"retry_ms":null}},
{"name":"partial_final_line","wire":"data: {\"done\":true}","expected":{"events":[],"retry_ms":null}},
{"name":"cr_only","wire":"data: one\r\rdata: two\r\r","expected":{"events":[{"event":"message","data":"one","id":""},{"event":"message","data":"two","id":""}],"retry_ms":null}},
{"name":"mixed_line_endings","wire":"data: first\rdata: second\n\r\ndata: third\r\n\r\n","expected":{"events":[{"event":"message","data":"first\nsecond","id":""},{"event":"message","data":"third","id":""}],"retry_ms":null}},
{"name":"comment_unknown_case","wire":": keepalive\nData: ignored\nx: ignored\ndata: ok\n\n","expected":{"events":[{"event":"message","data":"ok","id":""}],"retry_ms":null}},
{"name":"one_leading_bom","wire":"data: first\n\ndata: inside\n\n","expected":{"events":[{"event":"message","data":"first","id":""},{"event":"message","data":"inside","id":""}],"retry_ms":null}},
{"name":"id_persistence_reset_null","wire":"id: A\ndata: one\n\ndata: two\n\nid: bad\u0000id\ndata: three\n\nid:\ndata: four\n\n","expected":{"events":[{"event":"message","data":"one","id":"A"},{"event":"message","data":"two","id":"A"},{"event":"message","data":"three","id":"A"},{"event":"message","data":"four","id":""}],"retry_ms":null}},
{"name":"retry_ascii_only","wire":"retry: 40\nretry: +5\nretry: ５\nretry:\nretry: 9ms\n\ndata: ok\n\n","expected":{"events":[{"event":"message","data":"ok","id":""}],"retry_ms":40}},
{"name":"no_colon_data","wire":"data\n\n","expected":{"events":[{"event":"message","data":"","id":""}],"retry_ms":null}},
{"name":"one_space_and_first_colon","wire":"data: a:b\n\n","expected":{"events":[{"event":"message","data":" a:b","id":""}],"retry_ms":null}},
{"name":"unterminated_second_event","wire":"data: first\n\ndata: second\n","expected":{"events":[{"event":"message","data":"first","id":""}],"retry_ms":null}},
{"name":"done_is_ordinary_payload","wire":"data: [DONE]\n\n","expected":{"events":[{"event":"message","data":"[DONE]","id":""}],"retry_ms":null}}
]
test_fixture.py
"""Literal expectations and deliberate mistakes; also works with python -O."""
import argparse
import json
import platform
from pathlib import Path
from sse_fixture import FixtureDecoder, decode
ROOT = Path(__file__).resolve().parent
def check(condition, detail):
if not condition:
raise AssertionError(detail)
def must_raise(error, action):
try:
action()
except error:
return error.__name__
raise AssertionError('expected ' + error.__name__)
def run():
cases = json.loads((ROOT / 'fixtures.json').read_text(encoding='utf-8'))
checks = []
for case in cases:
raw = case['wire'].encode('utf-8')
chunks = [[raw], [bytes([b]) for b in raw]]
chunks += [[raw[:i], raw[i:]] for i in range(len(raw) + 1)]
for i, pieces in enumerate(chunks):
actual = decode(pieces)
check(actual == case['expected'], (case['name'], i, actual, case['expected']))
checks.append({'case': case['name'], 'partition_checks': len(chunks), 'passed': True})
negative = []
raw = 'data: 你好🙂\n\n'.encode('utf-8')
cut = raw.index('你'.encode('utf-8')) + 1
error = must_raise(UnicodeDecodeError,
lambda: ''.join(c.decode('utf-8') for c in [raw[:cut], raw[cut:]]))
negative.append({'strategy': 'decode_each_chunk', 'detected': error})
raw = b'data: first\r\rdata: second\r\r'
wrong_blocks = raw.split(b'\n\n')[:-1]
check(len(wrong_blocks) == 0 and len(decode([raw])['events']) == 2,
'LF-only splitter should miss both CR events')
negative.append({'strategy': 'split_only_lf_lf', 'detected': 'misses_two_CR_events'})
raw = b'data: {"done":true}\n'
wrong_eof = decode([raw + b'\n'])
check(len(wrong_eof['events']) == 1 and not decode([raw])['events'],
'adding a blank line should invent one event')
negative.append({'strategy': 'invent_blank_line_at_eof', 'detected': 'invented_event'})
empty = decode([b'data:\n\n'])['events']
wrong_skip = [e for e in empty if e['data'].strip()]
check(len(empty) == 1 and not wrong_skip, 'strip should erase one real event')
negative.append({'strategy': 'skip_empty_or_whitespace_payload',
'detected': 'drops_real_empty_data_event'})
# This wrong strategy assumes every delimited block contains JSON data.
raw = b'retry: 3000\n\n'
block = raw.decode('utf-8').split('\n\n')[0]
wrong_data = '\n'.join(line[5:].lstrip(' ') for line in block.splitlines()
if line.startswith('data:'))
must_raise(json.JSONDecodeError, lambda: json.loads(wrong_data))
check(decode([raw])['events'] == [], 'metadata-only should not dispatch')
negative.append({'strategy': 'parse_every_block_as_json',
'detected': 'JSONDecodeError_on_meta_only_block'})
policy = []
for name, raw in [('malformed_utf8', b'data: \xff\n\n'),
('truncated_utf8', b'data: \xe4'),
('partial_bom_one_byte', b'\xef'),
('partial_bom_two_bytes', b'\xef\xbb')]:
must_raise(UnicodeDecodeError, lambda raw=raw: decode([raw]))
policy.append({'case': name, 'passed': True})
parser = FixtureDecoder()
parser.finish()
must_raise(RuntimeError, lambda: parser.feed(b'data: late\n\n'))
policy.append({'case': 'input_after_finish_rejected', 'passed': True})
return {'python': platform.python_version(), 'fixture_count': len(cases),
'partition_checks': sum(c['partition_checks'] for c in checks),
'checks': checks, 'negative_controls': negative,
'application_policy_checks': policy,
'scope': 'Trusted synthetic bytes only. No SDK, network, browser or Agent integration.'}
if __name__ == '__main__':
args = argparse.ArgumentParser()
args.add_argument('--report', type=Path)
options = args.parse_args()
result = run()
if options.report is not None:
options.report.write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n',
encoding='utf-8')
print(f"{result['fixture_count']} fixtures; {result['partition_checks']} partition checks passed")
print(f"{len(result['negative_controls'])} negative controls detected; "
f"{len(result['application_policy_checks'])} application-policy checks passed")
16. 复跑结果与接入前仍需补上的边界
下载下方 ZIP，在解压目录运行以下命令。src 是新增实验；original 是从当前正文提取的六个字典回放文件，已同步上述三处显式 UTF-8 读取修正。也可以把正文中标明文件名的源码按相同目录放好，再分别运行。两套程序没有相互调用。
python src/demo_sse.py
python src/test_fixture.py
python -O src/test_fixture.py
python -m unittest discover -s original -p "test_progress.py" -v
新增测试在 Linux、Python 3.12.14 的普通和 -O 模式下均输出下面两行；原字典回放仍通过 42 项测试。源码、从正文提取的文件与 ZIP 解压副本运行结果相同，演示输出逐字节匹配 expected-demo.txt。报告版本号写在可选的 --report results.json 中；该选项会覆盖指定本地文件，使用时请选一个新的文件名。
expected-tests.txt
16 fixtures; 540 partition checks passed
5 negative controls detected; 5 application-policy checks passed
这里的 id_buffer 只是读到的字段值，retry_ms 也只是观察结果。代码没有发出 Last-Event-ID 请求、等待重试间隔或实现断线重连；未定义恢复游标、持久化去重或服务器确认。它会把整份事件列表留在内存里，也没有限制行长、事件大小和 retry 数字长度，不适合直接作为不可信网络输入的生产解析器。
分帧得到的 data 仍是普通文本。夹具中的 [DONE] 会原样返回；哪怕某个完整 JSON 写着 done=true 或 run.ended，也不能单凭它签发前文的任务终态。仍需受信任宿主的运行证据与绑定最终候选的应用验收。这里没有实现或验证 SSE 到原适配层的接线。
本次没有运行 SDK、模型、API、真实网络、浏览器 EventSource 或真实工具，也没有验证 HTTP 缓冲、取消、超时、背压和业务副作用。原创 SVG 按 1120 × 870 自然比例渲染检查，HTML 做了静态结构和源码一致性检查；没有浏览器像素验收。实际接入时，应在锁定的 SDK 与宿主版本上补齐这些集成测试，再决定哪些输入能进入显示状态机。
response.completed 到达后可以显示任务完成吗？
它结束的是一次模型响应。Agent 可能还要执行工具或继续调用模型，整体终态与应用验收仍需确认。
为什么取消后还显示工具需要核对？
停止本轮消费不证明已经发生的工具效果已撤销。真实业务结果应通过回执或对账查询。
本例的 run.ended 是官方事件吗？
不是。它是本文应用层自定义的规范化事件，由受信任的宿主根据生命周期证据签发。
42 项测试通过能证明真实网络可靠性吗？
不能。测试只回放合成字典，真实 SDK、SSE、模型、工具和网络均未运行。