事件协议:turn.completed 增加可选 failure 载荷,失败终态与正常终态同权入锚点

- direct_thread_wire 新增 DirectTurnFailure{kind,message} 类型,给 TurnCompleted 增可选 failure 字段,并补 turn_completed_failed 构造器与 failure 读取器
- with_user_item_id 显式带上 failure:原先把 TurnCompleted 写成 `..` 会静默吞掉失败载荷,身份与原因必须一起流转
- 新增 wire 用例:失败终态带载荷、正常终态不带且回写不补 null、缺载荷的 failed 事件仍可反序列化
- direct_thread_manager 增回归用例:turn.completed(status=failed) 必须顶替更早的 turn.started 成为 lifecycle_anchor,重放不会把已收口的回合看成"还在跑"
- 重新生成 ts-rs 绑定(新增 DirectTurnFailure.ts、DirectThreadEvent.ts 增 failure 字段)并按 prettier 格式化
This commit is contained in:
2026-09-22 16:58:52 +08:00
parent d36f5842b6
commit d32c99c927
4 changed files with 160 additions and 1 deletions
@@ -603,6 +603,30 @@ mod tests {
));
}
/// 失败终态与正常终态同权:`turn.completed(status="failed")` 必须顶替更早的 `turn.started`
/// 成为锚点,否则队列被回收后新订阅只会看到 `turn.started`,把这轮已收口的回合重放成"还在跑"。
#[test]
fn failed_turn_completed_replaces_started_anchor() {
let mut manager = DirectThreadManager::with_limits(100, 100_000);
manager.append("thread-1", DirectThreadEvent::turn_started(1_000));
manager.append(
"thread-1",
DirectThreadEvent::turn_completed_failed(
crate::agent::DirectTurnFailure::new("host-dropped", "回合宿主任务提前结束"),
FIXED_AT_MS,
),
);
let bootstrap = manager.subscribe("thread-1");
assert!(matches!(
bootstrap.events.as_slice(),
[DirectThreadEvent::TurnCompleted { status, failure, at, .. }]
if status == "failed"
&& failure.as_ref().is_some_and(|failure| failure.kind == "host-dropped")
&& *at == Some(FIXED_AT_MS)
));
}
/// 阶段时间必须随事件一起进队列:bootstrap 与重复订阅都拿到**原值**,
/// 重放不得重新取钟(否则每次重连都会把已固定的起止时间改掉)。
#[test]
@@ -211,6 +211,30 @@ impl DirectThreadRequestKind {
}
}
/// 失败终态的可下发载荷(`turn.completed.status == "failed"` 时必有,其余终态没有)。
///
/// `kind` 是稳定分类,只给界面选语气,不参与流程分支;`message` 是**已在宿主侧脱敏并截断**的
/// 可展示原因——失败原因只走这一条通道,前端不再从命令返回或另一条 IPC 里另造文案。
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize, TS)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[ts(export, export_to = concat!(env!("CARGO_MANIFEST_DIR"), "/../src/view/project-development/chat/generated/"))]
pub(crate) struct DirectTurnFailure {
/// 稳定失败分类:`timeout` / `model-failed` / `transport-failed` / `request-rejected` /
/// `host-dropped`。
pub(crate) kind: String,
/// 脱敏 + 截断后的失败原因。
pub(crate) message: String,
}
impl DirectTurnFailure {
pub(crate) fn new(kind: impl Into<String>, message: impl Into<String>) -> Self {
Self {
kind: kind.into(),
message: message.into(),
}
}
}
/// Thread Manager 下发的运行态事件。
///
/// 顺序由数组顺序给出(同一个 subscriber 的 `consume` 按队列顺序返回),因此不需要 `seq`:
@@ -250,7 +274,13 @@ pub(crate) enum DirectThreadEvent {
},
#[serde(rename = "turn.completed")]
TurnCompleted {
/// 终态语义:`completed` / `interrupted` / `aborted` 是正常收场;`failed` 是**失败**
/// 此时必须带 `failure` 载荷。
status: String,
/// 失败载荷:只有 `status == "failed"` 才有;失败原因只从这里下发一次。
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional)]
failure: Option<DirectTurnFailure>,
/// 本轮终态的阶段时间(毫秒):宿主处理终态的毫秒钟,或 `durationMs` + 高精度起点的派生值。
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional, as = "Option<f64>")]
@@ -301,11 +331,30 @@ impl DirectThreadEvent {
pub(crate) fn turn_completed(status: String, at: u64) -> Self {
Self::TurnCompleted {
status,
failure: None,
at: Some(at),
user_item_id: None,
}
}
/// 失败终态:`status` 固定 `"failed"`,原因必须随事件一起带出去。
pub(crate) fn turn_completed_failed(failure: DirectTurnFailure, at: u64) -> Self {
Self::TurnCompleted {
status: "failed".to_string(),
failure: Some(failure),
at: Some(at),
user_item_id: None,
}
}
/// 失败载荷:只有失败终态有。
pub(crate) fn failure(&self) -> Option<&DirectTurnFailure> {
match self {
Self::TurnCompleted { failure, .. } => failure.as_ref(),
_ => None,
}
}
/// 附上本轮开口用户条目的 canonical itemId。
///
/// 只在构造之后补一次身份,避免 `turn.started` / `turn.completed` 的既有调用点(含各处兜底
@@ -317,8 +366,14 @@ impl DirectThreadEvent {
.map(str::to_string);
match self {
Self::TurnStarted { at, .. } => Self::TurnStarted { at, user_item_id },
Self::TurnCompleted { status, at, .. } => Self::TurnCompleted {
Self::TurnCompleted {
status,
failure,
at,
..
} => Self::TurnCompleted {
status,
failure,
at,
user_item_id,
},
@@ -1351,4 +1406,56 @@ mod tests {
);
assert_eq!(item_event.user_item_id(), None);
}
/// 失败终态:`status="failed"` 必须带 `failure{kind,message}`,正常终态不带;载荷跟着身份
/// 一起流转,缺载荷的 `failed` 事件仍能反序列化(前端按"没有原因"处理,不猜)。
#[test]
fn turn_completed_carries_failure_payload_only_when_failed() {
let failed = DirectThreadEvent::turn_completed_failed(
DirectTurnFailure::new("model-failed", "上游返回 500:模型服务暂不可用"),
4_000,
)
.with_user_item_id(Some("direct-codex:turn-1:user"));
assert_eq!(
failed.failure(),
Some(&DirectTurnFailure::new(
"model-failed",
"上游返回 500:模型服务暂不可用"
))
);
assert_eq!(failed.user_item_id(), Some("direct-codex:turn-1:user"));
assert_eq!(
serde_json::to_value(&failed).expect("serialize failed turn"),
json!({
"type": "turn.completed",
"status": "failed",
"failure": {"kind": "model-failed", "message": "上游返回 500:模型服务暂不可用"},
"at": 4_000u64,
"userItemId": "direct-codex:turn-1:user",
})
);
assert_eq!(
serde_json::from_value::<DirectThreadEvent>(
serde_json::to_value(&failed).expect("serialize")
)
.expect("round trip"),
failed
);
// 正常终态不带载荷,也不回写 `failure: null`。
let completed = DirectThreadEvent::turn_completed("completed".to_string(), 5_000);
assert_eq!(completed.failure(), None);
assert_eq!(
serde_json::to_value(&completed).expect("serialize completed turn"),
json!({"type": "turn.completed", "status": "completed", "at": 5_000u64})
);
// 精简 / 旧形状:`failed` 但没有载荷也要能反序列化。
let sparse: DirectThreadEvent = serde_json::from_value(json!({
"type": "turn.completed",
"status": "failed",
}))
.expect("failed turn without failure payload");
assert_eq!(sparse.failure(), None);
}
}
@@ -2,6 +2,7 @@
import type { DirectThreadDeltaKind } from './DirectThreadDeltaKind';
import type { DirectThreadItem } from './DirectThreadItem';
import type { DirectThreadRequestKind } from './DirectThreadRequestKind';
import type { DirectTurnFailure } from './DirectTurnFailure';
/**
* Thread Manager 下发的运行态事件。
@@ -41,7 +42,15 @@ export type DirectThreadEvent =
}
| {
type: 'turn.completed';
/**
* 终态语义:`completed` / `interrupted` / `aborted` 是正常收场;`failed` 是**失败**
* 此时必须带 `failure` 载荷。
*/
status: string;
/**
* 失败载荷:只有 `status == "failed"` 才有;失败原因只从这里下发一次。
*/
failure?: DirectTurnFailure;
/**
* 本轮终态的阶段时间(毫秒):宿主处理终态的毫秒钟,或 `durationMs` + 高精度起点的派生值。
*/
@@ -0,0 +1,19 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
/**
* 失败终态的可下发载荷(`turn.completed.status == "failed"` 时必有,其余终态没有)。
*
* `kind` 是稳定分类,只给界面选语气,不参与流程分支;`message` 是**已在宿主侧脱敏并截断**的
* 可展示原因——失败原因只走这一条通道,前端不再从命令返回或另一条 IPC 里另造文案。
*/
export type DirectTurnFailure = {
/**
* 稳定失败分类:`timeout` / `model-failed` / `transport-failed` / `request-rejected` /
* `host-dropped`。
*/
kind: string;
/**
* 脱敏 + 截断后的失败原因。
*/
message: string;
};