DirectProject 待发消息条目改为事件投影，放行时重投影 canonical 与 prompt

- queue.rs：PendingTurn 退化成 queue.enqueued 的投影（new / from_event / enqueued_event 往返），删掉 canonical_user_item 与 prompt 两个宿主私有产物字段
- thread_manager/mod.rs：删 StoredEvent.pending（append_inner 并回 append）；pending_turns 改为一次事件折叠（有 queue.enqueued、没有配对 queue.removed），容量、判重、remove、claim 全读这一次折叠；mark_queue_events_cleanable 删掉“产物还在就不回收”的防御
- dispatch.rs：放行的 canonical = serde_json::to_value(&pending.user_item)，prompt = direct_codex_user_item_to_prompt(&pending.user_item)——不写盘、不读 manifest、不重跑校验
- user_input.rs：入队只造条目本身，不再冻结 prompt / canonical 形状
- wire.rs：queue.enqueued 注释改为“这条事件就是这条待发消息的全部事实”
- 验证：cargo test -- thread_manager::（62 passed，单线程跑进程级计数器用例）、-- agent::（948 passed，另一条 design_runtime 用例为既有并发 flaky，单跑通过）、-- direct_runtime:: analytics:: direct_codex_user_item（182 passed）、npx vitest run appSurface/directThreadChat/directHistoryPaging/directProjectTurn（249 passed）

apps/ai-game-creator-shell/src-tauri/src/agent/direct_runtime/user_input.rs

@@ -100,18 +100,9 @@ async fn enqueue_direct_codex_turn_typed(
             let detail = redact_agent_runtime_error(root, &error, 1800);
             DirectTurnError::EnvironmentNotReady { detail }
         })?;
-    // 入队：到这里这一条已经过了全部检查，剩下的就是排队等放行。prompt 与 canonical 形状在这里
-    // 冻结——放行不重算，所以放行没有失败出口。
-    let pending = PendingTurn::prepare(
-        turn_id,
-        user_item,
-        user_prompt,
-        creation_type,
-        direct_tool_call_now_ms(),
-    )
-    .map_err(|error| DirectTurnError::InputRejected {
-        detail: error.to_string(),
-    })?;
+    // 入队：到这里这一条已经过了全部检查，剩下的就是排队等放行。条目只带走它自己的事实
+    // （用户条目、创建类型、入队时刻），canonical 形状与 prompt 放行时从它重投影——放行没有失败出口。
+    let pending = PendingTurn::new(turn_id, user_item, creation_type, direct_tool_call_now_ms());
     match enqueue_pending_turn(&thread_id, pending) {
         Ok(_) => {}
         Err(EnqueueRejection::QueueFull) => {

apps/ai-game-creator-shell/src-tauri/src/agent/thread_manager/dispatch.rs

@@ -17,7 +17,7 @@ use std::path::{Path, PathBuf};
 use crate::agent::PendingTurn;
 use crate::agent::{
     append_direct_project_user_message_at, claim_pending_turn, complete_turn_if_reserved,
-    direct_tool_call_now_ms, redact_agent_runtime_error,
+    direct_codex_user_item_to_prompt, direct_tool_call_now_ms, redact_agent_runtime_error,
     run_direct_game_creator_turn_at_with_creation_type_and_emitter, thread_id_for_project,
     DirectGameCreatorTurnUpdateEmitter, DirectTurnError, DirectTurnTerminal, DispatchedTurn,
 };
@@ -62,7 +62,7 @@ impl TurnReservation {
     /// 测试用：直接开一轮占用，等价于"入队 + 认领"，不经过 Tauri 命令与真实的用户条目。
     #[cfg(test)]
     pub(crate) fn accept_for_test(thread_id: &str, client_turn_id: &str) -> Self {
-        let pending = PendingTurn::prepare(
+        let pending = PendingTurn::new(
             client_turn_id.to_string(),
             serde_json::from_value(serde_json::json!({
                 "type": "message",
@@ -71,11 +71,9 @@ impl TurnReservation {
                 "id": format!("direct-codex:{client_turn_id}:user"),
             }))
             .expect("canonical user item"),
-            "测试消息".to_string(),
             None,
             direct_tool_call_now_ms(),
-        )
-        .expect("prepare pending turn");
+        );
         super::enqueue_pending_turn(thread_id, pending).expect("enqueue test turn");
         let dispatched = claim_pending_turn(thread_id).expect("claim test turn");
         Self::resume(thread_id, &dispatched)
@@ -128,10 +126,12 @@ async fn run_dispatched_direct_turn(
     release_identity_generation: u64,
 ) {
     let turn_id = dispatched.pending.client_turn_id.clone();
+    // 放行只搬运条目自己的事实：canonical 形状与 prompt 都从冻结过的条目重投影，不重算、不写盘。
+    let canonical_user_item = serde_json::to_value(&dispatched.pending.user_item)
+        .expect("冻结过的 canonical 条目一定可序列化");
+    let prompt = direct_codex_user_item_to_prompt(&dispatched.pending.user_item);
     // 落盘即回合成立：放行之后必须留下这条用户消息，哪怕这一轮随后失败。
-    if let Err(error) =
-        append_direct_project_user_message_at(&root, &dispatched.pending.canonical_user_item)
-    {
+    if let Err(error) = append_direct_project_user_message_at(&root, &canonical_user_item) {
         // 不继续起整轮：历史是这条对话的单一事实源，用户消息没落盘时继续跑只会得到一条没有开口用户
         // 消息的助手回复，而且失败会被静默掉。
         let failure = DirectTurnError::EnvironmentNotReady {
@@ -146,18 +146,15 @@ async fn run_dispatched_direct_turn(
     }
     // 用户条目落盘成功即下发：这一轮从"放行"到"起 codex"之间的一切失败（连不上 app-server、执行器
     // 未通过验收、历史注入失败）都靠它把失败说明挂回自己那一轮。
-    crate::agent::codex_app_server::emit_direct_thread_user_item(
-        &root,
-        &dispatched.pending.canonical_user_item,
-    );
+    crate::agent::codex_app_server::emit_direct_thread_user_item(&root, &canonical_user_item);
     let capture = crate::analytics::gui::capture_writer_context();
     let emitter = DirectGameCreatorTurnUpdateEmitter::new(&root, turn_id);
     let outcome = run_direct_game_creator_turn_at_with_creation_type_and_emitter(
         &root,
-        &dispatched.pending.prompt,
+        &prompt,
         dispatched.pending.creation_type.as_deref(),
         Some(&emitter),
-        Some(dispatched.pending.canonical_user_item.clone()),
+        Some(canonical_user_item.clone()),
         capture,
         release_identity_generation,
     )
@@ -202,9 +199,9 @@ mod tests {
             .collect()
     }
 
-    /// 一条待发消息。放行路径只读它的身份与产物，所以这里用最小可用形状。
+    /// 一条待发消息。放行路径只读它自己的事实（身份、条目、创建类型），所以这里用最小可用形状。
     fn pending_turn(client_turn_id: &str) -> PendingTurn {
-        PendingTurn::prepare(
+        PendingTurn::new(
             client_turn_id.to_string(),
             serde_json::from_value(serde_json::json!({
                 "type": "message",
@@ -213,11 +210,9 @@ mod tests {
                 "id": format!("direct-codex:{client_turn_id}:user"),
             }))
             .expect("canonical user item"),
-            format!("消息 {client_turn_id}"),
             None,
             direct_tool_call_now_ms(),
         )
-        .expect("prepare pending turn")
     }
 
     /// 真实存在的项目目录不是这条用例的判据，用唯一的假路径当线程身份即可：认领与占用的语义只认

apps/ai-game-creator-shell/src-tauri/src/agent/thread_manager/mod.rs

@@ -38,12 +38,6 @@ struct StoredEvent {
     seq: u64,
     bytes: usize,
     cleanable: bool,
-    /// 这条事件在队期间挂着的宿主侧产物。
-    ///
-    /// 只有 `queue.enqueued` 会带它，而且**它本身就是队列成员身份**：`Some` = 这条待发消息还在队，
-    /// 被取走 = 已经离开队列。队列的先后就是事件的先后，所以没有第二张队列表要对齐，也不会出现
-    /// "事件说在队、队列表说不在"的中间态（见 `thread_manager::queue`）。
-    pending: Option<PendingTurn>,
 }
 
 #[derive(Clone, Debug)]
@@ -194,16 +188,6 @@ impl ThreadManager {
     }
 
     pub(crate) fn append(&mut self, thread_id: &str, event: ThreadEvent) -> ThreadEvent {
-        self.append_inner(thread_id, event, None)
-    }
-
-    /// 追加事件。`pending` 用来把宿主侧产物挂到这条事件上（只有入队会传）。
-    fn append_inner(
-        &mut self,
-        thread_id: &str,
-        event: ThreadEvent,
-        pending: Option<PendingTurn>,
-    ) -> ThreadEvent {
         let thread = self.threads.entry(thread_id.to_string()).or_default();
         thread.next_seq = thread.next_seq.saturating_add(1);
         let seq = thread.next_seq;
@@ -217,7 +201,6 @@ impl ThreadManager {
             seq,
             bytes,
             cleanable,
-            pending,
         });
         if let Some(item_id) = event.item_id() {
             Self::mark_item_events_cleanable(thread, item_id, seq);
@@ -418,13 +401,26 @@ impl ThreadManager {
     }
 
     /// 在队的待发消息，队首在前。
-    fn pending_turns(thread: &ThreadState) -> Vec<&PendingTurn> {
-        thread
-            .events
-            .iter()
-            .skip(thread.head)
-            .filter_map(|stored| stored.pending.as_ref())
-            .collect()
+    ///
+    /// 队列就是事件列表：一条待发消息在队 = 事件窗口里"有 `queue.enqueued`、没有配对
+    /// `queue.removed`"。一次折叠既判成员也定顺序，不需要第二张表；在队条目的入队事件永不可回收，
+    /// 所以它不会被前缀回收吃掉（见 `thread_manager::queue` 与 [`Self::observe_event`]）。
+    fn pending_turns(thread: &ThreadState) -> Vec<PendingTurn> {
+        let mut pending: Vec<PendingTurn> = Vec::new();
+        for stored in thread.events.iter().skip(thread.head) {
+            match &stored.event {
+                ThreadEvent::QueueEnqueued { .. } => {
+                    if let Some(turn) = PendingTurn::from_event(&stored.event) {
+                        pending.push(turn);
+                    }
+                }
+                ThreadEvent::QueueRemoved { client_turn_id, .. } => {
+                    pending.retain(|turn| &turn.client_turn_id != client_turn_id);
+                }
+                _ => {}
+            }
+        }
+        pending
     }
 
     /// 入队：把一条已经通过入队检查的待发消息排到队尾。
@@ -436,6 +432,7 @@ impl ThreadManager {
         thread_id: &str,
         turn: PendingTurn,
     ) -> Result<EnqueueOutcome, EnqueueRejection> {
+        let event = turn.enqueued_event();
         {
             let thread = self.threads.entry(thread_id.to_string()).or_default();
             let already_known = Self::pending_turns(thread)
@@ -450,8 +447,7 @@ impl ThreadManager {
             }
             queue_has_room(Self::pending_turns(thread).len())?;
         }
-        let event = turn.enqueued_event();
-        self.append_inner(thread_id, event, Some(turn));
+        self.append(thread_id, event);
         Ok(EnqueueOutcome::Enqueued)
     }
 
@@ -464,35 +460,24 @@ impl ThreadManager {
         client_turn_id: &str,
         at_ms: u64,
     ) -> QueueRemovalOutcome {
-        let removed = {
-            let Some(thread) = self.threads.get_mut(thread_id) else {
-                return QueueRemovalOutcome::NotFound;
-            };
-            if thread
-                .active_turn
-                .as_ref()
-                .is_some_and(|active| active.turn_id == client_turn_id)
-            {
-                return QueueRemovalOutcome::AlreadyDispatched;
-            }
-            let head = thread.head;
-            let mut removed = false;
-            for stored in thread.events.iter_mut().skip(head) {
-                if stored
-                    .pending
-                    .as_ref()
-                    .is_some_and(|pending| pending.client_turn_id == client_turn_id)
-                {
-                    // 取走产物即离开队列：这一幕在临界区里发生，外面看不到"事件还在、条目已经不在"。
-                    stored.pending = None;
-                    removed = true;
-                }
-            }
-            removed
+        let Some(thread) = self.threads.get(thread_id) else {
+            return QueueRemovalOutcome::NotFound;
         };
-        if !removed {
+        if thread
+            .active_turn
+            .as_ref()
+            .is_some_and(|active| active.turn_id == client_turn_id)
+        {
+            return QueueRemovalOutcome::AlreadyDispatched;
+        }
+        if !Self::pending_turns(thread)
+            .iter()
+            .any(|pending| pending.client_turn_id == client_turn_id)
+        {
             return QueueRemovalOutcome::NotFound;
         }
+        // 离开队列这件事由 `queue.removed` 自己表达：追加即出队（同一条入队事件就此转成可回收），
+        // 所以入队成功之后这里一定找得到那一条。
         self.append(
             thread_id,
             ThreadEvent::queue_removed(
@@ -515,17 +500,11 @@ impl ThreadManager {
         started_at_ms: u64,
     ) -> Option<DispatchedTurn> {
         let pending = {
-            let thread = self.threads.get_mut(thread_id)?;
+            let thread = self.threads.get(thread_id)?;
             if thread.active_turn.is_some() {
                 return None;
             }
-            let head = thread.head;
-            let stored = thread
-                .events
-                .iter_mut()
-                .skip(head)
-                .find(|stored| stored.pending.is_some())?;
-            stored.pending.take().expect("filtered on a pending turn")
+            Self::pending_turns(thread).into_iter().next()?
         };
         let token = Uuid::new_v4().to_string();
         let turn_id = pending.client_turn_id.clone();
@@ -745,10 +724,8 @@ impl ThreadManager {
     /// 一条待发消息离开队列：它和它对应的入队事件一起变成可回收。
     fn mark_queue_events_cleanable(thread: &mut ThreadState, client_turn_id: &str) {
         for stored in &mut thread.events {
-            // 仍挂着产物的条目永远不算可回收：在队条目被回收就等于队列悄悄丢了一条消息。
-            if stored.pending.is_none()
-                && stored.event.queue_client_turn_id() == Some(client_turn_id)
-            {
+            // 配对的两条一起转可回收——在队条目的入队事件本来就不会走到这里。
+            if stored.event.queue_client_turn_id() == Some(client_turn_id) {
                 stored.cleanable = true;
             }
         }
@@ -1064,7 +1041,10 @@ pub(crate) fn unsubscribe_thread(subscription_id: &str) {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::agent::{ThreadDeltaKind, ThreadItem, ThreadRequestKind, MAX_PENDING_TURNS};
+    use crate::agent::{
+        direct_codex_user_item_to_prompt, ThreadDeltaKind, ThreadItem, ThreadRequestKind,
+        MAX_PENDING_TURNS,
+    };
 
     fn message(item_id: &str) -> ThreadItem {
         ThreadItem::Message {
@@ -1471,7 +1451,7 @@ mod tests {
     }
 
     fn pending_turn(client_turn_id: &str) -> PendingTurn {
-        PendingTurn::prepare(
+        PendingTurn::new(
             client_turn_id.to_string(),
             serde_json::from_value(serde_json::json!({
                 "type": "message",
@@ -1480,11 +1460,9 @@ mod tests {
                 "id": format!("direct-codex:{client_turn_id}:user"),
             }))
             .expect("canonical user item"),
-            format!("消息 {client_turn_id}"),
             None,
             FIXED_AT_MS,
         )
-        .expect("prepare pending turn")
     }
 
     fn enqueued_ids(events: &[ThreadEvent]) -> Vec<&str> {
@@ -1680,7 +1658,10 @@ mod tests {
             .claim_pending_turn("thread-1", 2_000)
             .expect("claim head");
         assert_eq!(claimed.pending.client_turn_id, "turn-1");
-        assert_eq!(claimed.pending.prompt, "消息 turn-1");
+        assert_eq!(
+            direct_codex_user_item_to_prompt(&claimed.pending.user_item),
+            "消息 turn-1"
+        );
         assert_eq!(manager.pending_turn_ids("thread-1"), vec!["turn-2"]);
 
         let events = manager

apps/ai-game-creator-shell/src-tauri/src/agent/thread_manager/queue.rs

@@ -4,11 +4,10 @@
 //! `queue.enqueued` 事件不可回收；离开队列（取消或放行）时才转成可回收。新订阅者的 bootstrap 因此
 //! 天然看得见当前队列，不需要第二张队列表，也不会有"事件与队列不一致"的窗口。
 //!
-//! 这个模块只放三件事：一条待发消息带走什么（[`PendingTurn`]）、容量规则
-//! （[`MAX_PENDING_TURNS`]）、以及它在线上长什么样（[`PendingTurn::enqueued_event`]）。
-//! 它不碰锁、不碰 Tauri、不写盘：入队检查在命令侧，放行顺序在 Thread Manager。
-
-use serde_json::Value;
+//! 这个模块只放三件事：一条待发消息在线上长什么样（[`PendingTurn`] 与
+//! [`PendingTurn::enqueued_event`]）、容量规则（[`MAX_PENDING_TURNS`]）、以及"它还在不在队"
+//! 这条判据的成员折叠（`ThreadManager::pending_turns`）。
+//! 它不碰锁、不碰 Tauri、不写盘、不重算 prompt：入队检查在命令侧，放行顺序在 Thread Manager。
 
 use crate::agent::{
     direct_codex_user_item_id_for_client_turn_id, DirectCodexUserItem, ThreadEvent,
@@ -22,48 +21,57 @@ pub(crate) const MAX_PENDING_TURNS: usize = 5;
 
 /// 一条已经通过入队检查、正在等放行的用户消息。
 ///
-/// 只在内存里，进程重启即消失（与 ADR 记的边界一致）。它同时是**放行时要用的全部输入**：放行
-/// 没有失败出口，所以检查产物在入队时就地冻结，放行只搬运、不重算。
+/// 它**就是 `queue.enqueued` 的投影**（[`PendingTurn::from_event`] / [`PendingTurn::enqueued_event`]）：
+/// 字段与事件载荷一一对应，宿主不为它另存第二份产物——canonical 形状与 prompt 都在放行时从这条条目
+/// 重投影。在队 = 事件窗口里"有 `queue.enqueued`、没有配对 `queue.removed`"，成员与顺序因此只从事件
+/// 折出来，没有"事件说在队、队列表说不在"的中间态。
 #[derive(Clone, Debug)]
 pub(crate) struct PendingTurn {
     /// 这条消息的回合身份；放行后同一轮的 `turn.started` / `turn.completed` 用它。
     pub(crate) client_turn_id: String,
     /// canonical 用户条目：事件与界面 chip 都读它，Rust 不渲染展示形状。
     pub(crate) user_item: DirectCodexUserItem,
-    /// canonical 用户条目的 JSON 形状：入队时算好，放行时直接落盘。
-    pub(crate) canonical_user_item: Value,
-    /// 入队检查产出的 prompt：放行不重算。
-    pub(crate) prompt: String,
     pub(crate) creation_type: Option<String>,
     /// 入队那一刻的宿主毫秒钟。
     pub(crate) at: u64,
 }
 
 impl PendingTurn {
-    /// 组一条待发消息：入队检查已经全部通过，这里只把放行要用的产物冻结下来。
-    ///
-    /// 冻结是刻意的：放行没有失败出口，所以任何可能在放行时才失败的计算都必须提前到这里
-    /// （canonical 形状与 prompt 都是）。
-    pub(crate) fn prepare(
+    pub(crate) fn new(
         client_turn_id: String,
         user_item: DirectCodexUserItem,
-        prompt: String,
         creation_type: Option<String>,
         at: u64,
-    ) -> Result<Self, serde_json::Error> {
-        let canonical_user_item = serde_json::to_value(&user_item)?;
-        Ok(Self {
+    ) -> Self {
+        Self {
             client_turn_id,
             user_item,
-            canonical_user_item,
-            prompt,
             creation_type,
             at,
+        }
+    }
+
+    /// 从 `queue.enqueued` 事件折出这条待发消息；其它事件返回 `None`。
+    pub(crate) fn from_event(event: &ThreadEvent) -> Option<Self> {
+        let ThreadEvent::QueueEnqueued {
+            client_turn_id,
+            user_item,
+            creation_type,
+            at,
+        } = event
+        else {
+            return None;
+        };
+        Some(Self {
+            client_turn_id: client_turn_id.clone(),
+            user_item: user_item.clone(),
+            creation_type: creation_type.clone(),
+            at: *at,
         })
     }
 
-    /// 入队事件的投影：带 canonical 用户条目与 `creationType`，**不带 prompt**（prompt 只留在宿主的
-    /// 队列条目里，它不是要下发的展示形状）。
+    /// 入队事件的投影：带 canonical 用户条目与 `creationType`，**不带 prompt**（prompt 是入队检查的
+    /// 产物，放行时由条目重投影，不是要下发的展示形状），也不带埋点身份（成绩由宿主自己结算）。
     pub(crate) fn enqueued_event(&self) -> ThreadEvent {
         ThreadEvent::queue_enqueued(
             self.client_turn_id.clone(),
@@ -119,14 +127,12 @@ mod tests {
     }
 
     fn pending(client_turn_id: &str, creation_type: Option<&str>) -> PendingTurn {
-        PendingTurn::prepare(
+        PendingTurn::new(
             client_turn_id.to_string(),
             user_item("生成一个游戏", "direct-codex:turn-1:user"),
-            "生成一个游戏".to_string(),
             creation_type.map(str::to_string),
             1_700_000_000_123,
         )
-        .expect("prepare pending turn")
     }
 
     /// 入队事件只带 canonical 用户条目与 `creationType`：prompt 是宿主的入队检查产物，
@@ -162,18 +168,21 @@ mod tests {
         assert_eq!(event.queue_removal_reason(), None);
     }
 
-    /// canonical 形状在入队时就冻结：放行时落盘的就是这一份，不再重算。
+    /// 条目就是事件投影：由自己的入队事件折回来的条目与原件逐字段相同。
     #[test]
-    fn prepare_freezes_the_canonical_item() {
-        let turn = pending("turn-1", None);
-        assert_eq!(
-            turn.canonical_user_item["id"],
-            json!("direct-codex:turn-1:user")
-        );
-        assert_eq!(
-            turn.canonical_user_item,
-            serde_json::to_value(&turn.user_item).expect("serialize user item")
-        );
+    fn pending_turn_round_trips_through_its_enqueued_event() {
+        for turn in [pending("turn-1", Some("web-game")), pending("turn-2", None)] {
+            let restored =
+                PendingTurn::from_event(&turn.enqueued_event()).expect("queue.enqueued projects");
+            assert_eq!(restored.client_turn_id, turn.client_turn_id);
+            assert_eq!(restored.creation_type, turn.creation_type);
+            assert_eq!(restored.at, turn.at);
+            assert_eq!(
+                serde_json::to_value(&restored.user_item).expect("serialize user item"),
+                serde_json::to_value(&turn.user_item).expect("serialize user item")
+            );
+            assert!(PendingTurn::from_event(&ThreadEvent::turn_started(1)).is_none());
+        }
     }
 
     /// 用户条目身份由 `clientTurnId` 派生，与落盘 / 下发用的是同一个函数。

apps/ai-game-creator-shell/src-tauri/src/agent/thread_manager/wire.rs

@@ -354,7 +354,8 @@ pub(crate) enum ThreadEvent {
     /// 这条事件在条目仍在队期间**不可回收**，离开队列（取消或放行）时才转成可回收——新订阅者
     /// 靠这一点在 bootstrap 里看到当前队列，`is_bootstrap_event` 不需要为它加特例。
     ///
-    /// 事件不带 prompt：prompt 是入队检查的产物，只留在宿主的队列条目里。
+    /// 事件就是这条待发消息的**全部**事实：宿主不为它另存产物，prompt 与 canonical 形状都在放行时
+    /// 从这条条目重投影。
     #[serde(rename = "queue.enqueued")]
     QueueEnqueued {
         /// 这条待发消息的回合身份；放行后同一轮的 `turn.started` / `turn.completed` 用它。