Unified Exec会话状态机
Unified Exec 没有一个包罗所有阶段的 enum SessionState。它把生命周期分散在三处：reserved_process_ids 表示 ID 已占用但条目未必建立；ProcessStore.processes 表示可交互会话；ProcessState 表示底层进程是否退出、失败或被 sandbox 拒绝。真正的状态机只能从这些字段的写入者和消费者重建。
本文承接Unified Exec数据结构和Unified Exec创建进程，面向理解 watch、Arc、异步锁和取消 token 的读者。本文不展开输出截断或 watcher 的事件字段，而是回答：ID 何时有效、何时从 store 移除，退出与失败怎样保留，write_stdin 为什么可能在 map 已删除后仍返回退出结果。
1. 三层状态
1.1 ID状态
ID 在 spawn 前被加入 reserved_process_ids。启动失败调用 release_process_id；长命令启动成功后才进入 processes。因此“已预留”不等于“可交互”。
源码位置：codex-rs/core/src/unified_exec/process_manager.rs :: allocate_process_id、release_process_id
pub(crate) async fn allocate_process_id(&self) -> i32 {
loop {
let mut store = self.process_store.lock().await;
let process_id = if should_use_deterministic_process_ids() {
store
.reserved_process_ids
.iter()
.copied()
.max()
.map(|m| std::cmp::max(m, 999) + 1)
.unwrap_or(1000)
} else {
rand::rng().random_range(1_000..100_000)
};
if store.reserved_process_ids.contains(&process_id) {
continue;
}
store.reserved_process_ids.insert(process_id);
return process_id;
}
}测试使用从 1000 递增的确定性 ID，生产使用随机范围并在锁内检查冲突。数字本身不编码状态；判断能否交互必须查询 processes。
1.2 Store状态
进程创建后只要仍存活，manager 就在首次 yield 等待之前插入 ProcessEntry。这一步故意早于工具响应：即使 Turn 在等待期间被取消，store 中的 Arc 仍能保活后台进程。initial_exec_command_active 在 ID 尚未交付给模型时保护条目；refresh_process_state 则是“运行中 → 已退出并移出 store”的集中转换点。
源码位置：codex-rs/core/src/unified_exec/process_manager.rs :: UnifiedExecProcessManager::exec_command
let process_started_alive = !process.has_exited() && process.exit_code().is_none();
let mut initial_exec_command_guard = if process_started_alive {
let initial_exec_command_active = Arc::new(AtomicBool::new(true));
self.store_process(
Arc::clone(&process),
context,
&request.command,
request.hook_command.clone(),
cwd.clone(),
plugin_attribution.clone(),
start,
request.process_id,
request.tty,
deferred_network_approval.clone(),
network_denial_monitor,
metrics_sidecar,
Arc::clone(&transcript),
Arc::clone(&initial_exec_command_active),
)
.await;
InitialExecCommandGuard {
active: Some(initial_exec_command_active),
metrics_sidecar: None,
}
} else {
InitialExecCommandGuard {
active: None,
metrics_sidecar,
}
};guard 析构时才把 active flag 清为 false。长命令此后继续留在 store；若命令在首次等待内退出，refresh 可以先移出 entry，同时仍利用其中的网络审批和 metrics sidecar 完成收尾。
源码位置：codex-rs/core/src/unified_exec/process_manager.rs :: refresh_process_state
let Some(entry) = store.processes.get_mut(&process_id) else {
return ProcessStatus::Unknown;
};
let exit_code = entry.process.exit_code();
let process_id = entry.process_id;
if entry.process.has_exited() {
let Some(entry) = store.remove(process_id) else {
return ProcessStatus::Unknown;
};
ProcessStatus::Exited {
exit_code,
entry: Box::new(entry),
}
} else {
ProcessStatus::Alive {
exit_code,
call_id: entry.call_id.clone(),
process_id,
}
}Exited 携带被移出的完整 entry，调用方仍可结束网络审批、读取 call ID 和进程错误；Unknown 只说明当前 store 没有该 ID，不自动证明底层进程从未存在。
1.3 进程状态
ProcessState 是本地与远程共同的最新快照。退出、失败和 sandbox denial 可以同时存在，而不是互斥枚举。
源码位置：codex-rs/core/src/unified_exec/process_state.rs :: ProcessState
pub(crate) struct ProcessState {
pub(crate) has_exited: bool,
pub(crate) exit_code: Option<i32>,
pub(crate) failure_message: Option<String>,
pub(crate) sandbox_denied: bool,
}
pub(crate) fn failed(&self, message: String) -> Self {
Self {
has_exited: true,
exit_code: self.exit_code,
failure_message: Some(message),
sandbox_denied: self.sandbox_denied,
}
}失败会把 has_exited 置真并保留已有退出码；后到的退出通知又会保留 failure。读者不能用 exit_code.is_some() 代替 has_exited。
2. 退出发布
2.1 signal_exit
本地 exit receiver 和远程事件最终调用 signal_exit。它替换状态快照并取消 output token，唤醒等待退出的 watcher。
源码位置：codex-rs/core/src/unified_exec/process.rs :: signal_exit
fn signal_exit(&self, exit_code: Option<i32>) {
let state = self.state_rx.borrow().clone();
let _ = self.state_tx.send_replace(state.exited(exit_code));
self.output.cancellation_token.cancel();
}取消 token 在这里表示“进程生命周期已结束”，不是用户一定发出了 cancel。watcher 以它作为退出屏障，再等待输出 drain。
2.2 终态屏障
后台 watcher 必须依次等待退出 token、输出 drain、延迟网络拒绝监控和 interaction_lock，之后才能发布唯一的结束事件。这里的 output drain 不只是“收到一次通知”：streaming task 在 exit token 后继续等待 producer close，若 close 通知缺失，才使用 trailing grace 作为兜底。
源码位置：codex-rs/core/src/unified_exec/async_watcher.rs :: spawn_exit_watcher
exit_token.cancelled().await;
output_drained.notified().await;
if let Some(network_denial_monitor) = network_denial_monitor {
let _ = network_denial_monitor.await;
}
let _interaction_guard = interaction_lock.lock_owned().await;
let duration = Instant::now().saturating_duration_since(started_at);
let plugin_metrics_sidecar = plugin_metrics_sidecar
.as_ref()
.and_then(take_plugin_metrics_sidecar);如果只等 child exit，最后一段 stdout 可能还未进入 transcript；如果不等网络 monitor，刚退出的请求可能被错误发布为成功。交互锁又防止 watcher 与正在轮询的 write_stdin 同时消费终态。
metrics sidecar 也在这个屏障后获取。它以 Arc<Mutex<Option<T>>> 共享给首次响应与 watcher，take 保证只有先观察到终态的一方完成测量。失败状态不会伪造正常 exit code：watcher 丢弃 sidecar 后发布 failed end；成功状态才用最终 exit code 完成测量。
3. 交互转换
3.1 同一会话串行
write_stdin 先从 store 克隆进程，再取得 interaction_lock，然后重新调用 prepare_process_handles 校验同一个 Arc。不同 ID 可并发，同一 ID 的读写与终止不能重叠。
源码位置：codex-rs/core/src/unified_exec/process_manager.rs :: write_stdin
let locked_process = {
let store = self.process_store.lock().await;
let entry = store
.processes
.get(&process_id)
.ok_or(UnifiedExecError::UnknownProcessId { process_id })?;
Arc::clone(&entry.process)
};
let _interaction_guard = locked_process.interaction_lock().lock_owned().await;
let PreparedProcessHandles { process, output, .. } = self
.prepare_process_handles(process_id, &locked_process)
.await?;两次查询之间条目可能被替换或删除，所以第二次校验不是多余的。Arc::ptr_eq 失败时返回 Unknown，避免旧任务操作新会话。
3.2 TTY与pipe
非空输入在 TTY 模式写字节；pipe 模式一般关闭 stdin，但允许控制字符 INTERRUPT 转成 signal。写入失败后 manager 刷新状态：若进程已退出，继续构造终态响应；若是通信失败且仍未退出，则终止并释放 ID。
源码位置：codex-rs/core/src/unified_exec/process_manager.rs :: write_stdin
if !request.input.is_empty() {
if !tty {
if request.input == INTERRUPT {
process.interrupt().await?;
} else {
return Err(UnifiedExecError::StdinClosed);
}
} else {
match process.write(request.input.as_bytes()).await {
Ok(()) => tokio::time::sleep(Duration::from_millis(100)).await,
Err(err) => {
let status = self.refresh_process_state(process_id).await;
if matches!(status, ProcessStatus::Exited { .. }) {
status_after_write = Some(status);
} else if matches!(err, UnifiedExecError::ProcessFailed { .. }) {
process.terminate();
self.release_process_id(process_id).await;
return Err(err);
} else {
return Err(err);
}
}
}
}
}这段分支把“stdin 已关闭但进程正常结束”和“远程通信失败且进程状态未知”区分开来，不能统一写成工具失败。
3.3 轮询终态
收集输出后，manager 再刷新状态。Alive 返回同一 process ID；Exited 返回 process_id: None 并结束网络审批；Unknown 若手中的 Arc 已退出，仍可返回退出码，否则才报告 Unknown。
源码位置：codex-rs/core/src/unified_exec/process_manager.rs :: write_stdin
let (process_id, exit_code, event_call_id) = match status {
ProcessStatus::Alive { exit_code, call_id, process_id } => {
(Some(process_id), exit_code, call_id)
}
ProcessStatus::Exited { exit_code, entry } => {
let call_id = entry.call_id.clone();
finish_network_approval_after_process_exit_for_entry(&entry).await?;
(None, exit_code, call_id)
}
ProcessStatus::Unknown => {
if process.has_exited() {
(None, process.exit_code(), call_id)
} else {
return Err(UnifiedExecError::UnknownProcessId {
process_id: request.process_id,
});
}
}
};Unknown 的 fallback 是典型竞态处理：另一个任务可能已从 map 移除条目，但当前轮询仍持有有效 Arc，因此可以安全返回终态，而不是丢掉退出码。
4. 失败状态
4.1 远程stdin关闭
远程 backend 返回 UnknownProcess 或 StdinClosed 时，统一进程把当前快照转成 exited、取消 token，并返回 WriteToStdin。manager 随后刷新 store 状态，可能把它转成正常终态响应。
源码位置：codex-rs/core/src/unified_exec/process.rs :: write
WriteStatus::UnknownProcess | WriteStatus::StdinClosed => {
let state = self.state_rx.borrow().clone();
let _ = self.state_tx.send_replace(state.exited(state.exit_code));
self.output.cancellation_token.cancel();
Err(UnifiedExecError::WriteToStdin)
}4.2 失败只写一次
fail_and_terminate 只在没有 failure message 时写入，随后终止底层进程。重复错误不会覆盖第一个根因。
源码位置：codex-rs/core/src/unified_exec/process.rs :: fail_and_terminate
pub(super) fn fail_and_terminate(&self, message: String) {
let state = self.state_rx.borrow().clone();
if state.failure_message.is_none() {
let _ = self.state_tx.send_replace(state.failed(message));
}
self.terminate();
}这对网络拒绝和 transport failure 很重要：后到的 terminate 错误不应掩盖最初的 policy failure。
4.3 延迟网络拒绝
网络策略结果可能稍晚于 child exit 到达。manager 为此保留一个 100ms grace：如果进程 token 先取消，它仍短暂等待 network cancellation；确认拒绝后调用 fail_and_terminate，让 watcher 在发布成功事件前看到失败状态。
源码位置：codex-rs/core/src/unified_exec/process_manager.rs :: wait_for_late_network_denial、terminate_process_on_network_denial
async fn wait_for_late_network_denial(network_cancelled: Option<CancellationToken>) -> bool {
let Some(network_cancelled) = network_cancelled else {
return false;
};
if network_cancelled.is_cancelled() {
return true;
}
tokio::select! {
_ = network_cancelled.cancelled() => true,
_ = tokio::time::sleep(LATE_NETWORK_DENIAL_GRACE_PERIOD) => false,
}
}这不是延迟所有成功命令 100ms。只有 deferred network approval 存在时才有 cancellation token；exit watcher 还会等待对应 monitor 完成，从而把“OS 已退出”和“策略分类已完成”分成两个状态事实。
5. 显式终止
terminate_process 先在锁内复制 Arc，锁外等待 terminate_confirmed，再重新加锁校验同一个对象。首轮响应仍活跃时保留条目，防止 process ID 尚未交付就被删除。
源码位置：codex-rs/core/src/unified_exec/process_manager.rs :: terminate_process
if !already_exited && process.terminate_confirmed().await.is_err() {
return false;
}
let entry = {
let mut store = self.process_store.lock().await;
let Some(entry) = store.processes.get(&process_id) else {
return true;
};
if !Arc::ptr_eq(&entry.process, &process) {
return true;
}
if entry.initial_exec_command_active.load(Ordering::Acquire) {
return true;
}
let Some(entry) = store.remove(process_id) else {
return false;
};
entry
};
unregister_network_approval_for_entry(&entry).await;
true返回 true 表示终止请求已处理，并不总表示条目立即删除；首轮保护与并发替换都会让清理交给其他路径完成。条目移出后，网络审批在锁外注销；metrics sidecar 的共享副本仍由 exit watcher 取得，避免终止 API 在没有完整 transcript 和终态分类时提前完成测量。
6. 四组验证
6.1 完成后复用
reusing_completed_process_returns_unknown_process 启动交互 shell，写入 exit，等待回收后再次轮询旧 ID。断言返回同一 ID 的 UnknownProcessId，并且 store 为空。它证明已完成会话不可复用，不证明数字 ID 永不在未来重新分配。
源码位置：codex-rs/core/src/unified_exec/mod_tests.rs :: reusing_completed_process_returns_unknown_process
cd codex-rs
cargo test -p codex-core --lib reusing_completed_process_returns_unknown_process -- --test-threads=16.2 轮询与终止
terminating_during_stdin_poll_returns_exited_response 让 poll 取得进程句柄后从 store 释放 ID，再终止底层 fake process。断言 poll 返回 process_id: None，而不是 Unknown。这正是 Unknown + Arc已退出 fallback 的证明范围。
源码位置：codex-rs/core/src/unified_exec/mod_tests.rs :: terminating_during_stdin_poll_returns_exited_response
cd codex-rs
cargo test -p codex-core --lib terminating_during_stdin_poll_returns_exited_response -- --test-threads=16.3 远程失败投影
remote_write_unknown_process_marks_process_exited 构造返回 WriteStatus::UnknownProcess 的 fake exec-server process，写入任意字节后断言得到 WriteToStdin，同时统一状态已经 exited。fail_and_terminate_preserves_failure_message 连续写入两个失败原因，断言第二个不会覆盖第一个。
源码位置：
codex-rs/core/src/unified_exec/process_tests.rs :: remote_write_unknown_process_marks_process_exitedcodex-rs/core/src/unified_exec/process_tests.rs :: fail_and_terminate_preserves_failure_message
cd codex-rs
cargo test -p codex-core --lib remote_write_unknown_process_marks_process_exited -- --test-threads=1
cargo test -p codex-core --lib fail_and_terminate_preserves_failure_message -- --test-threads=1这两个测试验证 UnifiedExecProcess 的局部状态转换，不覆盖 manager 的 store 移除或 handler 输出格式。
6.4 终态屏障
输出关闭测试先发送 process exit，再延迟 50ms 关闭 producer，断言 watcher 收到 close 后立即 drain，不必等完整 grace；相邻测试故意不关闭 producer，断言只能在 grace 到期后收尾。网络测试让 denial 在 exit 后 10ms 才写入，断言最终事件为 failed；manager 的 grace 测试则直接证明 exit 后短窗口仍能观察 cancellation。
源码位置：
codex-rs/core/src/unified_exec/async_watcher_tests.rs :: streaming_output_finishes_on_close_without_waiting_for_gracecodex-rs/core/src/unified_exec/async_watcher_tests.rs :: streaming_output_keeps_grace_as_fallback_without_closecodex-rs/core/src/unified_exec/async_watcher_tests.rs :: exit_watcher_waits_for_late_network_denial_before_classifying_endcodex-rs/core/src/unified_exec/process_manager_tests.rs :: late_network_denial_grace_observes_cancellation_after_exit
cd codex-rs
cargo test -p codex-core --lib streaming_output_finishes_on_close_without_waiting_for_grace -- --test-threads=1
cargo test -p codex-core --lib streaming_output_keeps_grace_as_fallback_without_close -- --test-threads=1
cargo test -p codex-core --lib exit_watcher_waits_for_late_network_denial_before_classifying_end -- --test-threads=1
cargo test -p codex-core --lib late_network_denial_grace_observes_cancellation_after_exit -- --test-threads=17. 故障复述
看到 UnknownProcessId 时，先问它发生在第一次 store 查询、prepare_process_handles 的 Arc::ptr_eq，还是 poll 结束后的 refresh。看到结束事件晚于 child exit 时，应检查 output drain、network denial monitor 和 interaction lock，而不是把延迟误判为 watcher 丢失。
rg -n "refresh_process_state|UnknownProcessId|ProcessStatus" codex-rs/core/src/unified_exec
rg -n "signal_exit|fail_and_terminate|spawn_exit_watcher" codex-rs/core/src/unified_exec
rg -n "reusing_completed_process|terminating_during_stdin_poll" codex-rs/core/src/unified_exec完整生命周期可以概括为：ID 先预留，活进程才入 store；退出先发布到 ProcessState，输出完成后 watcher 发终态事件；refresh 或显式终止移出条目并释放 ID。下一篇将只聚焦 write_stdin 的输入、控制字符、EOF 与非 TTY 行为。