宿主：逻辑回合的边界交给 Thread Manager，接单即成对
- direct_thread_manager 新增逻辑回合占用：接单在同一个临界区里拒并发 + 登记占用 + 追加 turn.started，返回已占用的 token - direct_thread_manager 拆出深层终态出口与占用兜底出口，notify 从 append 里抽出来复用 - 新增 direct_turn_accept：接单对象持有这一轮的终态出口，Drop 兜底补 host-dropped - direct_turn_failure 删除 DirectTurnFailureGuard，终态改成显式构造的 DirectTurnTerminal - codex_app_server 不再镜像 Codex 原生回合：删掉 run_turn 内的 turn.started 与守卫武装，终态改走 complete_direct_thread_turn - 用户条目事件仍由 run_turn 下发，顺序固定为逻辑回合开始 → 用户消息 → 起 codex
This commit is contained in:
@@ -34,6 +34,7 @@ mod direct_thread_wire;
|
||||
mod direct_tool_bridge;
|
||||
mod direct_tool_calls;
|
||||
mod direct_tools_mcp;
|
||||
mod direct_turn_accept;
|
||||
mod direct_turn_error;
|
||||
mod direct_turn_failure;
|
||||
mod direct_turn_metrics;
|
||||
|
||||
@@ -3538,30 +3538,21 @@ impl CodexAppServerConnection {
|
||||
}
|
||||
turn_start_guard.armed = false;
|
||||
let direct_thread_id = direct_thread_id_for_project(history_root);
|
||||
// 回合边界的阶段时间：Turn 上游只有**秒**级 `startedAt` / `completedAt`，秒级截断
|
||||
// 撑不起前端 0.1 秒粒度的展示，也可能让完成时刻落进该轮用户消息的同一秒、落在真实
|
||||
// 发送时间之前，被判成无效边界后整轮新回合被吞掉。因此这里只在宿主处理对应阶段时取
|
||||
// 毫秒钟（与条目侧"没有原生阶段时间就用宿主钟"同一口径），不再读上游秒字段。
|
||||
let direct_turn_started_at_ms = direct_tool_call_now_ms();
|
||||
// 本轮开口用户条目的 canonical id：只从已落盘的那条条目上读身份（`id`，工具条目才用
|
||||
// `call_id`），不在事件侧重造一份。拿不到就留空，让前端按"归属不可证明"处理。
|
||||
let direct_turn_user_item_id = direct_persisted_user_item
|
||||
.as_ref()
|
||||
.and_then(direct_thread_item_identity);
|
||||
// 回合终态兜底：`turn.started` 进队列之后就武装，写完终态即解除。宿主在这两者之间任何
|
||||
// 提前收场（panic、future 被丢弃、以后新增的早退）都由它补一条失败终态，否则前端只能
|
||||
// 永远停在"还在跑"。
|
||||
let mut direct_turn_failure_guard: Option<DirectTurnFailureGuard> = None;
|
||||
// 逻辑回合的**边界**不在这里：开始事件由接单动作发出、兜底由接单占用对象持有
|
||||
// （`direct_turn_accept.rs`）。这里只把本轮的用户条目作为第一条运行态条目下发，
|
||||
// 于是顺序天然是"逻辑回合开始 → 用户消息 → 起 codex"。
|
||||
//
|
||||
// 下面这个毫秒钟与逻辑回合无关，只服务模型终态的**完成时刻**：上游 Turn 的
|
||||
// `startedAt` / `completedAt` 只有秒级，秒级截断撑不起前端 0.1 秒粒度的展示，也可能
|
||||
// 让完成时刻落进该轮用户消息的同一秒。因此这里在进入模型往返前取一次宿主毫秒钟，与
|
||||
// `durationMs` 相加得到终态时刻；拿不到 `durationMs` 时退回观察时刻。
|
||||
let direct_turn_started_at_ms = direct_tool_call_now_ms();
|
||||
if self.inner.workspace_mode == CodexAppServerWorkspaceMode::DirectProject {
|
||||
append_direct_thread_event(
|
||||
&direct_thread_id,
|
||||
DirectThreadEvent::turn_started(direct_turn_started_at_ms)
|
||||
.with_user_item_id(direct_turn_user_item_id.as_deref()),
|
||||
);
|
||||
direct_turn_failure_guard = Some(DirectTurnFailureGuard::arm(
|
||||
direct_thread_id.clone(),
|
||||
direct_turn_user_item_id.clone(),
|
||||
));
|
||||
if let Some(user_item) = direct_persisted_user_item.as_ref() {
|
||||
if let Some(entry_item) = direct_thread_event_item(history_root, user_item) {
|
||||
// 这里的条目时间可能是启动应答后的观测时间；前端按同一用户条目身份
|
||||
@@ -4119,13 +4110,11 @@ impl CodexAppServerConnection {
|
||||
turn_failure.as_ref(),
|
||||
history_root,
|
||||
);
|
||||
append_direct_thread_event(
|
||||
// 终态走 Thread Manager 的深出口：解除这一轮的占用并写下 `turn.completed`。
|
||||
complete_direct_thread_turn(
|
||||
&direct_thread_id,
|
||||
terminal.event(completed_at, direct_turn_user_item_id.as_deref()),
|
||||
);
|
||||
if let Some(guard) = direct_turn_failure_guard.as_mut() {
|
||||
guard.disarm();
|
||||
}
|
||||
}
|
||||
let text = collect_result?;
|
||||
guard.armed = false;
|
||||
@@ -7795,6 +7784,13 @@ done
|
||||
let _active_invocation =
|
||||
crate::agent::DirectTaonierActiveInvocationGuard::enter(&project, "turn-0001")
|
||||
.expect("enter direct invocation");
|
||||
// 逻辑回合的开始事件由**接单动作**发出（命令侧），不是 run_turn 内部：这里补上同一步，
|
||||
// 于是这一轮的边界仍在同一个订阅里成对出现。
|
||||
let _reservation = crate::agent::direct_turn_accept::DirectTurnReservation::accept(
|
||||
&thread_id,
|
||||
Some("direct-codex:turn-0001:user"),
|
||||
)
|
||||
.expect("accept logical turn");
|
||||
let execution = super::super::direct_execution::open_at(
|
||||
&temp.path().join("host"),
|
||||
&project,
|
||||
|
||||
@@ -35,6 +35,15 @@ struct SubscriberState {
|
||||
cursor: u64,
|
||||
}
|
||||
|
||||
/// 一条正在跑的逻辑回合的占用：接单时登记，终态写出时解除。
|
||||
///
|
||||
/// `token` 是这一次接单的稳定身份：终态出口只有拿着同一个 token 的占用对象才能写兜底终态，
|
||||
/// 避免迟到的旧占用把新回合的边界顶掉。
|
||||
#[derive(Clone, Debug)]
|
||||
struct ActiveDirectTurn {
|
||||
token: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
struct ThreadState {
|
||||
next_seq: u64,
|
||||
@@ -43,6 +52,8 @@ struct ThreadState {
|
||||
total_bytes: usize,
|
||||
active_items: HashSet<String>,
|
||||
unresolved_requests: HashSet<String>,
|
||||
/// 未收口的逻辑回合。`None` 表示这个 thread 没有正在跑的回合。
|
||||
active_turn: Option<ActiveDirectTurn>,
|
||||
/// 最近一条 `turn.started` / `turn.completed` 的独立拷贝。
|
||||
///
|
||||
/// TODO(thread-manager): 这里有意只保留"锚点"，因为 replay 队列会回收可回收事件，
|
||||
@@ -63,6 +74,7 @@ impl Default for ThreadState {
|
||||
total_bytes: 0,
|
||||
active_items: HashSet::new(),
|
||||
unresolved_requests: HashSet::new(),
|
||||
active_turn: None,
|
||||
lifecycle_anchor: None,
|
||||
subscribers: HashMap::new(),
|
||||
}
|
||||
@@ -155,6 +167,75 @@ impl DirectThreadManager {
|
||||
}
|
||||
}
|
||||
|
||||
/// 接单：同一个临界区里拒绝并发、登记占用、追加逻辑回合开始事件。
|
||||
///
|
||||
/// 返回 `Err(existing_token)` 表示这个 thread 已经有一条没收口的回合——此时不动队列，
|
||||
/// 由调用方把它投影成接单拒绝。
|
||||
fn accept_turn(
|
||||
&mut self,
|
||||
thread_id: &str,
|
||||
token: &str,
|
||||
user_item_id: Option<&str>,
|
||||
started_at_ms: u64,
|
||||
) -> Result<DirectThreadEvent, String> {
|
||||
{
|
||||
let thread = self.threads.entry(thread_id.to_string()).or_default();
|
||||
if let Some(active) = thread.active_turn.as_ref() {
|
||||
return Err(active.token.clone());
|
||||
}
|
||||
thread.active_turn = Some(ActiveDirectTurn {
|
||||
token: token.to_string(),
|
||||
});
|
||||
}
|
||||
Ok(self.append(
|
||||
thread_id,
|
||||
DirectThreadEvent::turn_started(started_at_ms).with_user_item_id(user_item_id),
|
||||
))
|
||||
}
|
||||
|
||||
/// 深层的终态出口：解除占用并写下 `turn.completed`。
|
||||
///
|
||||
/// 不校验 token：这一条由真正跑完这一轮的代码调用，终态就是它算出来的那个（CLI 这类没有
|
||||
/// 占用登记的入口也走这里，保持"终态一定下发"的既有语义）。
|
||||
fn complete_turn(&mut self, thread_id: &str, event: DirectThreadEvent) -> DirectThreadEvent {
|
||||
if let Some(thread) = self.threads.get_mut(thread_id) {
|
||||
thread.active_turn = None;
|
||||
}
|
||||
self.append(thread_id, event)
|
||||
}
|
||||
|
||||
/// 占用对象的兜底出口：只有当这个 thread 仍被同一个 token 占用时才写。
|
||||
///
|
||||
/// 返回是否真的写了。深层已经写出终态时返回 `false`——兜底不覆盖真实结果。
|
||||
fn complete_turn_if_reserved(
|
||||
&mut self,
|
||||
thread_id: &str,
|
||||
token: &str,
|
||||
event: DirectThreadEvent,
|
||||
) -> bool {
|
||||
let reserved = match self.threads.get_mut(thread_id) {
|
||||
Some(thread) => match thread.active_turn.as_ref() {
|
||||
Some(active) if active.token == token => {
|
||||
thread.active_turn = None;
|
||||
true
|
||||
}
|
||||
_ => false,
|
||||
},
|
||||
None => false,
|
||||
};
|
||||
if !reserved {
|
||||
return false;
|
||||
}
|
||||
self.append(thread_id, event);
|
||||
true
|
||||
}
|
||||
|
||||
fn turn_is_active(&self, thread_id: &str) -> bool {
|
||||
self.threads
|
||||
.get(thread_id)
|
||||
.is_some_and(|thread| thread.active_turn.is_some())
|
||||
}
|
||||
|
||||
fn subscriber_ids(&self, thread_id: &str) -> Vec<String> {
|
||||
self.threads
|
||||
.get(thread_id)
|
||||
@@ -377,13 +458,76 @@ pub(crate) fn append_direct_thread_event(
|
||||
thread_id: &str,
|
||||
event: DirectThreadEvent,
|
||||
) -> DirectThreadEvent {
|
||||
let (event, subscriber_ids) = {
|
||||
let mut manager = global_direct_thread_manager()
|
||||
let event = {
|
||||
global_direct_thread_manager()
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
.append(thread_id, event)
|
||||
};
|
||||
notify_direct_thread_subscribers(thread_id);
|
||||
event
|
||||
}
|
||||
|
||||
/// 接单：拒绝并发 + 登记占用 + 发逻辑回合开始事件（见 [`DirectThreadManager::accept_turn`]）。
|
||||
pub(crate) fn accept_direct_thread_turn(
|
||||
thread_id: &str,
|
||||
token: &str,
|
||||
user_item_id: Option<&str>,
|
||||
started_at_ms: u64,
|
||||
) -> Result<(), String> {
|
||||
{
|
||||
global_direct_thread_manager()
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
.accept_turn(thread_id, token, user_item_id, started_at_ms)?;
|
||||
}
|
||||
notify_direct_thread_subscribers(thread_id);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// 深层终态出口：解除占用并写 `turn.completed`。
|
||||
pub(crate) fn complete_direct_thread_turn(thread_id: &str, event: DirectThreadEvent) {
|
||||
{
|
||||
global_direct_thread_manager()
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
.complete_turn(thread_id, event);
|
||||
}
|
||||
notify_direct_thread_subscribers(thread_id);
|
||||
}
|
||||
|
||||
/// 占用对象的兜底出口：仍被同一 token 占用时才写，返回是否写了。
|
||||
pub(crate) fn complete_direct_thread_turn_if_reserved(
|
||||
thread_id: &str,
|
||||
token: &str,
|
||||
event: DirectThreadEvent,
|
||||
) -> bool {
|
||||
let written = {
|
||||
global_direct_thread_manager()
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
.complete_turn_if_reserved(thread_id, token, event)
|
||||
};
|
||||
if written {
|
||||
notify_direct_thread_subscribers(thread_id);
|
||||
}
|
||||
written
|
||||
}
|
||||
|
||||
/// 这个 thread 是否还有没收口的逻辑回合。
|
||||
pub(crate) fn direct_thread_turn_is_active(thread_id: &str) -> bool {
|
||||
global_direct_thread_manager()
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
.turn_is_active(thread_id)
|
||||
}
|
||||
|
||||
fn notify_direct_thread_subscribers(thread_id: &str) {
|
||||
let subscriber_ids = {
|
||||
let manager = global_direct_thread_manager()
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner());
|
||||
let event = manager.append(thread_id, event);
|
||||
let subscriber_ids = manager.subscriber_ids(thread_id);
|
||||
(event, subscriber_ids)
|
||||
manager.subscriber_ids(thread_id)
|
||||
};
|
||||
if let Some(app) = DIRECT_THREAD_MANAGER_APP_HANDLE.get() {
|
||||
for subscription_id in subscriber_ids {
|
||||
@@ -394,7 +538,6 @@ pub(crate) fn append_direct_thread_event(
|
||||
);
|
||||
}
|
||||
}
|
||||
event
|
||||
}
|
||||
|
||||
pub(crate) fn subscribe_direct_thread(thread_id: &str) -> DirectThreadSubscriptionBootstrap {
|
||||
|
||||
@@ -0,0 +1,213 @@
|
||||
//! DirectProject 的接单：把"一条用户消息被接单"变成 Thread Manager 里一对必然成对的逻辑回合事件。
|
||||
//!
|
||||
//! 这个模块只有一件事，别再往里加第二件：**接单成立的那一刻**在同一个临界区里拒绝并发、登记占用、
|
||||
//! 发出逻辑回合开始事件；占用对象持有这一轮的终态出口——正常 / 失败 / 接单后的前置失败谁先写谁算，
|
||||
//! 都没写时由 `Drop` 补一条 `host-dropped`。
|
||||
//!
|
||||
//! 为什么回合边界不能继续镜像 Codex 原生回合：`turn/start` 之前的失败（连不上 app-server、配置未
|
||||
//! 就绪失败、历史注入失败）根本没有原生回合可以镜像，而它们同样是"这一轮已经成立"。设计见
|
||||
//! `docs/adr/【ADR】DirectProject命令接单化-2026-09-23.md`。
|
||||
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::{
|
||||
accept_direct_thread_turn, complete_direct_thread_turn_if_reserved, direct_tool_call_now_ms,
|
||||
DirectTurnError, DirectTurnTerminal,
|
||||
};
|
||||
|
||||
/// 一次接单的占用。持有它就代表这一轮还没收口。
|
||||
///
|
||||
/// 生命周期由调用方决定：命令把整轮任务 spawn 出去时把它一起搬进任务，任务结束（正常或失败）
|
||||
/// 时它随任务一起 drop。**持有顺序要与单飞锁一致**：单飞锁先声明、占用后声明，drop 时占用先收尾，
|
||||
/// 新回合不可能插到中间。
|
||||
pub(crate) struct DirectTurnReservation {
|
||||
thread_id: String,
|
||||
token: String,
|
||||
user_item_id: Option<String>,
|
||||
}
|
||||
|
||||
impl DirectTurnReservation {
|
||||
/// 接单：登记占用并发出逻辑回合开始事件。
|
||||
///
|
||||
/// 失败表示这个 thread 已经有一条没收口的回合（并发接单），此时不改队列、不发事件。
|
||||
pub(crate) fn accept(
|
||||
thread_id: &str,
|
||||
user_item_id: Option<&str>,
|
||||
) -> Result<Self, DirectTurnError> {
|
||||
let token = Uuid::new_v4().to_string();
|
||||
accept_direct_thread_turn(thread_id, &token, user_item_id, direct_tool_call_now_ms())
|
||||
.map_err(|existing| DirectTurnError::TurnAlreadyRunning {
|
||||
existing_invocation_id: existing,
|
||||
incoming_invocation_id: token.clone(),
|
||||
})?;
|
||||
Ok(Self {
|
||||
thread_id: thread_id.to_string(),
|
||||
token,
|
||||
user_item_id: user_item_id.map(str::to_string),
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn thread_id(&self) -> &str {
|
||||
&self.thread_id
|
||||
}
|
||||
|
||||
/// 接单之后还没走到深层终态就失败的收口口：只有这一轮仍被自己占用时才写。
|
||||
///
|
||||
/// 深层（真正跑完这一轮的代码）已经写出终态时返回 `false`，兜底不覆盖真实结果。
|
||||
pub(crate) fn finish_if_unfinished(&self, terminal: DirectTurnTerminal) -> bool {
|
||||
complete_direct_thread_turn_if_reserved(
|
||||
&self.thread_id,
|
||||
&self.token,
|
||||
terminal.event(direct_tool_call_now_ms(), self.user_item_id.as_deref()),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for DirectTurnReservation {
|
||||
fn drop(&mut self) {
|
||||
// 兜底：任务 panic、future 被丢弃、或今后在终态之前新增的 `?` 早退。
|
||||
// 这类失败说不出原因，只给分类；能说清原因的错误必须由调用方在更早的地方显式收口。
|
||||
let _ = self.finish_if_unfinished(DirectTurnTerminal::host_dropped());
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::agent::{
|
||||
consume_direct_thread, direct_thread_turn_is_active, subscribe_direct_thread,
|
||||
DirectThreadEvent, DirectTurnFailure,
|
||||
};
|
||||
|
||||
/// 订阅并把 bootstrap 拿掉：之后的 `consume` 只返回这次订阅之后产生的事件。
|
||||
fn watch(thread_id: &str) -> String {
|
||||
let bootstrap = subscribe_direct_thread(thread_id);
|
||||
let _ = consume_direct_thread(&bootstrap.subscription_id);
|
||||
bootstrap.subscription_id
|
||||
}
|
||||
|
||||
fn pending(subscription_id: &str) -> Vec<DirectThreadEvent> {
|
||||
consume_direct_thread(subscription_id)
|
||||
.expect("consume")
|
||||
.events
|
||||
}
|
||||
|
||||
fn turn_completed_events(events: &[DirectThreadEvent]) -> Vec<&DirectThreadEvent> {
|
||||
events
|
||||
.iter()
|
||||
.filter(|event| matches!(event, DirectThreadEvent::TurnCompleted { .. }))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn unique_thread(label: &str) -> String {
|
||||
format!("accept-test-{label}-{}", Uuid::new_v4())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accept_emits_a_logical_turn_started_and_holds_the_turn() {
|
||||
let thread = unique_thread("started");
|
||||
let subscription = watch(&thread);
|
||||
|
||||
let reservation = DirectTurnReservation::accept(&thread, Some("u-1")).expect("accept");
|
||||
|
||||
let events = pending(&subscription);
|
||||
assert_eq!(events.len(), 1, "{events:?}");
|
||||
match &events[0] {
|
||||
DirectThreadEvent::TurnStarted { user_item_id, .. } => {
|
||||
assert_eq!(user_item_id.as_deref(), Some("u-1"));
|
||||
}
|
||||
other => panic!("expected turn.started, got {other:?}"),
|
||||
}
|
||||
assert!(direct_thread_turn_is_active(&thread));
|
||||
drop(reservation);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_second_accept_is_rejected_while_the_turn_is_open() {
|
||||
let thread = unique_thread("busy");
|
||||
let subscription = watch(&thread);
|
||||
let reservation = DirectTurnReservation::accept(&thread, Some("u-1")).expect("accept");
|
||||
// 先取走第一条接单自己的开始事件，之后的"空"才只说明被拒的这一次没写东西。
|
||||
assert_eq!(pending(&subscription).len(), 1);
|
||||
|
||||
let rejected = DirectTurnReservation::accept(&thread, Some("u-2"));
|
||||
|
||||
assert!(matches!(
|
||||
rejected,
|
||||
Err(DirectTurnError::TurnAlreadyRunning { .. })
|
||||
));
|
||||
let events = pending(&subscription);
|
||||
assert_eq!(events.len(), 0, "被拒的接单不许产生事件：{events:?}");
|
||||
drop(reservation);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn drop_without_a_terminal_writes_a_host_dropped_terminal() {
|
||||
let thread = unique_thread("drop");
|
||||
let subscription = watch(&thread);
|
||||
let reservation = DirectTurnReservation::accept(&thread, Some("u-1")).expect("accept");
|
||||
assert!(reservation.finish_if_unfinished(DirectTurnTerminal::host_dropped()));
|
||||
assert!(!direct_thread_turn_is_active(&thread));
|
||||
|
||||
// 显式收口之后 Drop 不再补第二条：兜底只负责"没人写过"的那一种。
|
||||
drop(reservation);
|
||||
|
||||
let events = pending(&subscription);
|
||||
let completed = turn_completed_events(&events);
|
||||
assert_eq!(completed.len(), 1, "{events:?}");
|
||||
match completed[0] {
|
||||
DirectThreadEvent::TurnCompleted {
|
||||
user_item_id,
|
||||
failure: Some(failure),
|
||||
..
|
||||
} => {
|
||||
assert_eq!(failure.kind, "host-dropped");
|
||||
assert_eq!(user_item_id.as_deref(), Some("u-1"));
|
||||
}
|
||||
other => panic!("expected a failed terminal, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_deep_terminal_wins_and_the_fallback_stays_silent() {
|
||||
let thread = unique_thread("deep");
|
||||
let subscription = watch(&thread);
|
||||
let reservation = DirectTurnReservation::accept(&thread, Some("u-1")).expect("accept");
|
||||
|
||||
// 深层收口：真正跑完这一轮的代码算出来的终态。
|
||||
let deep = DirectThreadEvent::turn_completed_failed(
|
||||
DirectTurnFailure::new("timeout".to_string(), "等待模型回执超时".to_string()),
|
||||
2_000,
|
||||
)
|
||||
.with_user_item_id(Some("u-1"));
|
||||
crate::agent::complete_direct_thread_turn(&thread, deep);
|
||||
|
||||
assert!(
|
||||
!reservation.finish_if_unfinished(DirectTurnTerminal::host_dropped()),
|
||||
"深层已收口时兜底不许再写"
|
||||
);
|
||||
drop(reservation);
|
||||
|
||||
let events = pending(&subscription);
|
||||
let completed = turn_completed_events(&events);
|
||||
assert_eq!(completed.len(), 1, "一轮只许有一条终态：{events:?}");
|
||||
match completed[0] {
|
||||
DirectThreadEvent::TurnCompleted { failure, .. } => {
|
||||
assert_eq!(failure.as_ref().map(|f| f.kind.as_str()), Some("timeout"));
|
||||
}
|
||||
other => panic!("expected a terminal, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_thread_can_be_accepted_again_after_the_turn_is_settled() {
|
||||
let thread = unique_thread("again");
|
||||
let first = DirectTurnReservation::accept(&thread, Some("u-1")).expect("accept");
|
||||
drop(first);
|
||||
|
||||
let second = DirectTurnReservation::accept(&thread, Some("u-2")).expect("second accept");
|
||||
|
||||
assert!(direct_thread_turn_is_active(&thread));
|
||||
drop(second);
|
||||
}
|
||||
}
|
||||
@@ -3,8 +3,10 @@
|
||||
//!
|
||||
//! 这个模块只有三件事，别再往里加第四件：
|
||||
//! 1. [`direct_turn_terminal`]：拿这一轮的事实判定终态——是不是失败、原因是什么、状态写什么；
|
||||
//! 2. [`DirectTurnTerminal::event`]：把终态投影成 `turn.completed` 事件；
|
||||
//! 3. [`DirectTurnFailureGuard`]：`turn.started` 之后武装、写完终态解除的 Drop 兜底。
|
||||
//! 2. [`DirectTurnTerminal::event`]：把终态投影成 `turn.completed` 事件。
|
||||
//!
|
||||
//! 终态的**出口**（谁写、什么时候兜底）不在这里，在 `direct_turn_accept.rs` 的接单占用对象里：
|
||||
//! 这个模块只负责"什么算失败、原因怎么写"。
|
||||
//!
|
||||
//! 失败载荷的**形状**属于线上协议，定义在 `direct_thread_wire.rs`（`DirectTurnFailure`）；
|
||||
//! 载荷的 `kind` 与 `message` 由 [`DirectTurnError`] 投影而来（`kind` 的取值表见
|
||||
@@ -13,10 +15,7 @@
|
||||
|
||||
use std::path::Path;
|
||||
|
||||
use super::{
|
||||
append_direct_thread_event, direct_tool_call_now_ms, redact_agent_runtime_error,
|
||||
DirectThreadEvent, DirectTurnError, DirectTurnFailure,
|
||||
};
|
||||
use super::{redact_agent_runtime_error, DirectThreadEvent, DirectTurnError, DirectTurnFailure};
|
||||
|
||||
/// `turn.completed.failure.message` 的字符上限：与本地错误文案同一档——够说清原因，又不至于
|
||||
/// 把整段上游报文塞进事件队列。
|
||||
@@ -78,7 +77,18 @@ pub(crate) fn direct_turn_terminal(
|
||||
(None, Ok(_)) => None,
|
||||
};
|
||||
match failure {
|
||||
Some(failure) => DirectTurnTerminal {
|
||||
Some(failure) => DirectTurnTerminal::failed(history_root, &failure),
|
||||
None => DirectTurnTerminal {
|
||||
status: session_status.to_string(),
|
||||
failure: None,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
impl DirectTurnTerminal {
|
||||
/// 一次失败终态：`kind` 与 `message` 只在这一个出口从 typed 错误投影。
|
||||
pub(crate) fn failed(history_root: &Path, failure: &DirectTurnError) -> Self {
|
||||
Self {
|
||||
status: "failed".to_string(),
|
||||
failure: Some(DirectTurnFailure::new(
|
||||
failure.wire_kind().unwrap_or("model-failed").to_string(),
|
||||
@@ -88,65 +98,21 @@ pub(crate) fn direct_turn_terminal(
|
||||
DIRECT_TURN_FAILURE_MESSAGE_MAX_CHARS,
|
||||
),
|
||||
)),
|
||||
},
|
||||
None => DirectTurnTerminal {
|
||||
status: session_status.to_string(),
|
||||
failure: None,
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 回合终态兜底守卫：`turn.started` 发出去之后，这一轮在宿主侧只剩两条收场路径——正常路径
|
||||
/// 写完终态事件（然后 [`Self::disarm`]），或者这个守卫的 `Drop`。
|
||||
///
|
||||
/// 兜底覆盖三种"走不到终态"的情况：panic 展开、future 被丢弃（任务 / 进程取消），以及今后在
|
||||
/// 终态事件之前新增的 `?` 早退。它们都再也没有机会补终态事件，前端只能永远停在"还在跑"；
|
||||
/// 这里在 Drop 里补一条 `status="failed"` + `host-dropped` 的终态，让前端拿到收口依据。
|
||||
///
|
||||
/// 与 `CodexTurnGuard` / `CodexTurnStartGuard` 是**三件事**，不要合并：那两个守卫管的是
|
||||
/// app-server 连接与 `turn/start` 请求的回收，Drop 里不产出任何事件。
|
||||
///
|
||||
/// 已知边界（不为它加路径）：宿主进程被强杀（`kill -9`）时没有任何 `Drop` 会执行，前端仍会停在
|
||||
/// 运行态；`turn.started` 之前的早退根本不武装这个守卫——没有开始就没有"未收口的回合"。
|
||||
pub(crate) struct DirectTurnFailureGuard {
|
||||
thread_id: String,
|
||||
user_item_id: Option<String>,
|
||||
armed: bool,
|
||||
}
|
||||
|
||||
impl DirectTurnFailureGuard {
|
||||
/// 武装：调用点必须是 `turn.started` **已经**进入队列之后。
|
||||
pub(crate) fn arm(thread_id: String, user_item_id: Option<String>) -> Self {
|
||||
/// 宿主任务提前结束（panic / future 被丢弃 / 取消）的兜底终态。
|
||||
///
|
||||
/// 这类收场说不出原因，只给分类；能说清原因的一律走 [`Self::failed`]。
|
||||
pub(crate) fn host_dropped() -> Self {
|
||||
Self {
|
||||
thread_id,
|
||||
user_item_id,
|
||||
armed: true,
|
||||
status: "failed".to_string(),
|
||||
failure: Some(DirectTurnFailure::new(
|
||||
DIRECT_TURN_FAILURE_HOST_DROPPED_KIND.to_string(),
|
||||
DIRECT_TURN_FAILURE_HOST_DROPPED_MESSAGE.to_string(),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
/// 解除：终态事件（正常或失败）已经写完，兜底不再需要。
|
||||
pub(crate) fn disarm(&mut self) {
|
||||
self.armed = false;
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for DirectTurnFailureGuard {
|
||||
fn drop(&mut self) {
|
||||
if !self.armed {
|
||||
return;
|
||||
}
|
||||
append_direct_thread_event(
|
||||
&self.thread_id,
|
||||
DirectThreadEvent::turn_completed_failed(
|
||||
DirectTurnFailure::new(
|
||||
DIRECT_TURN_FAILURE_HOST_DROPPED_KIND,
|
||||
DIRECT_TURN_FAILURE_HOST_DROPPED_MESSAGE,
|
||||
),
|
||||
direct_tool_call_now_ms(),
|
||||
)
|
||||
.with_user_item_id(self.user_item_id.as_deref()),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
@@ -279,41 +245,13 @@ stderrClass=nonempty；stderrBytes=1000";
|
||||
));
|
||||
}
|
||||
|
||||
/// 兜底：守卫武装后没被解除就 Drop，必须补一条失败终态（panic / future 被丢弃走的就是这条）。
|
||||
/// 兜底终态：说不出原因的那一种只给分类，不冒充真实原因。
|
||||
#[test]
|
||||
fn armed_guard_appends_host_dropped_terminal_on_drop() {
|
||||
let thread_id = "test-thread-failure-guard-armed";
|
||||
let subscription = subscribe_direct_thread(thread_id);
|
||||
let guard = DirectTurnFailureGuard::arm(
|
||||
thread_id.to_string(),
|
||||
Some("direct-codex:turn-1:user".to_string()),
|
||||
);
|
||||
drop(guard);
|
||||
|
||||
let events = consume_direct_thread(&subscription.subscription_id)
|
||||
.expect("consume guard terminal")
|
||||
.events;
|
||||
assert!(matches!(
|
||||
events.as_slice(),
|
||||
[DirectThreadEvent::TurnCompleted { status, failure, user_item_id, .. }]
|
||||
if status == "failed"
|
||||
&& failure.as_ref().is_some_and(|failure| failure.kind == "host-dropped")
|
||||
&& user_item_id.as_deref() == Some("direct-codex:turn-1:user")
|
||||
));
|
||||
}
|
||||
|
||||
/// 解除之后就闭嘴：正常写完终态的回合不得再多出一条兜底终态。
|
||||
#[test]
|
||||
fn disarmed_guard_appends_nothing() {
|
||||
let thread_id = "test-thread-failure-guard-disarmed";
|
||||
let subscription = subscribe_direct_thread(thread_id);
|
||||
let mut guard = DirectTurnFailureGuard::arm(thread_id.to_string(), None);
|
||||
guard.disarm();
|
||||
drop(guard);
|
||||
|
||||
assert!(consume_direct_thread(&subscription.subscription_id)
|
||||
.expect("consume disarmed guard")
|
||||
.events
|
||||
.is_empty());
|
||||
fn host_dropped_terminal_only_carries_the_classification() {
|
||||
let terminal = DirectTurnTerminal::host_dropped();
|
||||
assert_eq!(terminal.status, "failed");
|
||||
let failure = terminal.failure.expect("host-dropped must fail the turn");
|
||||
assert_eq!(failure.kind, "host-dropped");
|
||||
assert!(!failure.message.trim().is_empty());
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user