读懂 Codex（5.25）：取消模型流时连接和活动 Item 怎样收尾

Turn 取消不是把网络 Task 粗暴遗忘。采样等待点把 CancellationToken 映射为 TurnAborted，随后 flush 解析器、取消并 drain 已启动工具、丢弃 ResponseStream 以通知映射任务；只有已经收到 Item Done 的事实留在 History。

具体问题与启用条件

本节聚焦普通 Responses 流取消。Session 怎样最终只发一个 TurnAborted 已在第三章说明，工具子进程具体清理在第六章。

条件来源	决定字段或状态	对本机制的影响
Turn 控制	Interrupt/替换/父任务取消	取消当前 token
采样等待	stream.next().or_cancel	立即跳出为 TurnAborted
工具执行	child cancellation token	传播到 in-flight future

协议、类型与状态所有权

状态或协议	所有者	生命周期	关键不变量
Turn CancellationToken	活动 Task	一个 Turn	父取消传播到采样与工具 child
ResponseStream consumer_dropped	Core stream wrapper	流对象	Drop 时取消 mapper task 分支
active_item/parser	采样局部状态	当前流	取消后 flush 展示尾部但不提交半项
已 Done ResponseItem	History/Rollout	持久	取消不会回滚已完成事实

机制调用链如下：

Interrupt
→ Turn token cancelled
→ stream.next or stream open 的 or_cancel 返回 Cancelled
→ outcome=TurnAborted
→ flush all assistant parser state
→ child tokens 取消 in-flight tools
→ drain FuturesOrdered 并记录必要 output
→ Drop ResponseStream
→ consumer_dropped 让 mapper 记录 cancelled/停止转发
→ 返回 TurnAborted 生命周期

机制怎样工作

连接建立和每次 stream.next() 都包在可取消等待中。取消命中时事件循环不把它包装成 Stream 网络错误，而是直接得到 CodexErr::TurnAborted，因此不会进入重试。离开 loop 后仍会 flush Assistant parser，随后等待 in-flight 工具完成取消收尾；token count 在已有 usage 时仍可发送。最后再次检查 cancellation token，确保即便工具 drain 正常返回也不会把采样当成功。

Core ResponseStream 的 Drop 会 cancel consumer_dropped。映射任务 select 到它后记录 inference attempt cancelled，并停止从 codex-api stream 取事件；如果发送通道先发现消费者消失也走相同记录。底层 SSE/WS parser随后因接收端关闭或自身终态退出。当前源码不承诺“取消一定保留 WebSocket”；因此复刻时应把连接复用视为优化，只有明确完成且协议同步的连接才可保留。

Python 风格伪代码

这段伪代码保留生产实现中会改变结果的状态、分支和异步边界；认证 SDK、遥测字段和 Rust 所有权样板被折叠为明确的领域对象。

async def sample_with_cancellation(stream: ResponseStream, cancel: asyncio.Event):
    state = StreamState()
    try:
        while True:
            event = await next_event_or_cancel(stream, cancel)
            await state.handle(event)
            if isinstance(event, Completed):
                break
    except TurnCancelled:
        state.outcome = TurnAborted()
    finally:
        await state.flush_text_parsers()  # 只完成 UI 暂态
        for task in state.tool_tasks:
            task.cancel()
        await drain_tool_tasks_and_record_outputs(state.tool_tasks)
        await stream.close()  # signal consumer_dropped

    if cancel.is_set():
        raise TurnAborted()
    return state.result()

class ResponseStream:
    async def close(self):
        self.consumer_dropped.set()
        await self.mapper_task.stop_forwarding()

失败、取消与恢复

故障或边界	已发生的状态	对上层的结果	能否直接重试	恢复动作
取消发生在连接建立	尚无内容项	TurnAborted	否	停止请求尝试
取消发生在 active Delta	UI 可能有片段，History 无半项	TurnAborted	否	flush/关闭暂态 Item
取消发生在 Done ToolCall 后	Call 已记录且工具已启动	取消 child 并 drain	不自动回滚	记录配对 output
取消与 Completed 竞争	usage/Completed 可能已处理	最终 token 再检查	否	Turn 终态仍由上层唯一化

设计取舍与验证

协作式取消比直接 abort 所有 Task 慢，因为要 drain 工具并保住配对与 usage；收益是 History、Rollout 和子进程不留悬空状态。暂态 UI 可能已经显示一部分文本，这是流式体验的固有代价，但它不会冒充持久 Assistant Message。

可验证契约	证据方式	预期结果
Interrupt 很快产生 TurnAborted	挂起模型/工具后取消	不出现重试或 TurnComplete
半截流不写完整 Message	Delta 后取消	History 无该 Message Done 项
已完成 Call 仍有取消结果	ToolCall Done 后 interrupt	Rollout 保持 call/output 配对

Mini Codex 对照

Mini Codex 应把 cancel token同时传给模型流和工具 runtime，并在 finally 中 drain。若 Python WebSocket 库不能安全复用取消后的连接，直接关闭重连比冒险保留更正确。

本节边界

普通 Responses 生命周期到此闭合。最后一节比较 Realtime：它同样使用 WebSocket，却有完全不同的会话、媒体和 handoff 状态机。