Interrupt 的第一步不是取消网络请求,而是从 Session 原子取走 ActiveTurn。这个动作先改变归属:新的 Steer 不再能写入旧 Turn;随后才通过 CancellationToken、宽限等待和 Tokio abort 向实际工作传播。
从 Op 到 Turn 终态的顺序
abort_all_tasks(Interrupted) 取走整个 ActiveTurn,从 RunningTask 保存 TurnContext,然后调用 handle_task_abort。完成后触发 abort lifecycle,清 pending waiters/items;若有 mailbox trigger 或 durable sleep,再检查是否应启动新的工作。
async def interrupt_session():
active = await take_active_turn() # 原子变成 None
if active is None:
return
task = active.take_task()
if task is not None:
await handle_task_abort(task, INTERRUPTED)
await emit_turn_abort_lifecycle(task.turn)
await input_queue.clear_pending(active)
await maybe_start_turn_for_pending_work()
按指定 turn id 的 abort_turn_if_active 使用 compare-and-take:只有当前 task id 相同才取得 ActiveTurn并返回 true。它防止延迟的取消命令误杀后继 Turn。
模型流怎样观察取消
RegularTask 把父 token 的 child 交给 run_turn,再交给 sampling request。流建立和每次 stream.next() 都通过可取消组合等待;token 触发时返回 CodexErr::TurnAborted。MCP capture、required server 解析等准备步骤也使用相同取消树。
工具 Runtime 为每个调用再建 child token。取消发生时,它先检查工具是否已经产生 terminal outcome;若已完成就收下结果,避免“结果刚完成却被误写成 aborted”。未完成工具根据 runtime 能力决定等待内部 teardown 或 abort dispatch handle,随后生成模型可配对的 aborted response 并发 tool-aborted 通知。
async def run_tool_with_cancel(call, token):
handle = spawn(dispatch(call, token.clone()))
winner = await select(handle.result, token.cancelled())
if winner.is_result:
return winner.result
if terminal_outcome_already_reached() or handle.is_finished():
return await handle
if tool_waits_for_runtime_cancellation(call):
claim_terminal_outcome()
await handle # 等 runtime kill/close 进程
else:
handle.abort()
await notify_tool_aborted(call.id)
return aborted_output(call.id)
子进程不是由 Tokio abort 自动杀死
丢弃 Rust future 不保证操作系统子进程退出。Shell/UnifiedExec runtime 必须把 cancellation 映射为 process terminate,并等待 exit watcher完成必要输出。需要长驻的 background terminal 有独立 ProcessManager 所有权;Turn 中断是否终止某个进程取决于工具 runtime 的 terminal policy,Session Shutdown 则无条件 terminate_all_processes。
这一区别避免两种错误:把前台命令留成孤儿,以及把用户有意后台化的 terminal仅因一次模型采样结束就杀掉。详细进程策略会在第六章展开,本节只固定生命周期接口。
评论
登录后即可评论