diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent.rs b/apps/ai-game-creator-shell/src-tauri/src/agent.rs index 13dab607d..823729aed 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent.rs @@ -22,6 +22,7 @@ mod direct_project_turn_history; mod direct_runtime; mod direct_tool_bridge; mod direct_tool_calls; +mod direct_turn_stream; mod direct_tools_mcp; mod generation; mod interaction; @@ -53,6 +54,7 @@ pub(crate) use direct_project_turn_history::*; pub(crate) use direct_runtime::*; pub(crate) use direct_tool_bridge::*; pub(crate) use direct_tool_calls::*; +pub(crate) use direct_turn_stream::*; pub(crate) use direct_tools_mcp::*; pub(crate) use generation::*; pub(crate) use interaction::*; diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server.rs index 03980d27f..e64141a08 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server.rs @@ -572,6 +572,14 @@ enum CodexTurnEvent { #[derive(Clone, Debug, Eq, PartialEq)] pub(crate) enum DirectCodexTurnObservation { AccumulatedText(String), + /// 一个 assistant 文本段的当前累计全文。 + /// + /// `item_id` 是一次 assistant 消息的稳定身份：同一个 id 的后续 delta 属于**同一段**， + /// id 变了就是新的一段。回合流的"文本段 + 工具"顺序用它来分段，而不是按 delta 分。 + AgentMessageSegment { + item_id: String, + accumulated_text: String, + }, IntermediateText(String), /// 模型的思考过程（reasoning item 的明文摘要）：流式阶段整段替换下发。 Reasoning(String), @@ -2927,6 +2935,17 @@ impl CodexAppServerConnection { observer(DirectCodexTurnObservation::AccumulatedText( streamed_text.clone(), )); + // 同一个 assistant item 的当前累计全文：回合流按 item 分段， + // 段内只追加、段间才换行，不能拿"整轮累计"当一段。 + let segment_text = direct_project_history + .accumulated_text_for(&item_id) + .unwrap_or_else(|| delta.clone()); + if !segment_text.trim().is_empty() { + observer(DirectCodexTurnObservation::AgentMessageSegment { + item_id: item_id.clone(), + accumulated_text: segment_text, + }); + } } if let Some(callback) = on_agent_message_delta.as_deref_mut() { callback(&platform_llm::LlmStreamDelta { diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_turn_history.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_turn_history.rs index c2c317692..c41b19de0 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_turn_history.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_turn_history.rs @@ -22,6 +22,14 @@ impl DirectProjectHistoryAccumulator { } } + /// 某个 assistant item 目前累计到的全文。 + /// + /// 回合流按 item 分段：同一个 item 的后续 delta 是同一段的增长，item 变了才是新的一段。 + /// 没有这条 item（非 DirectProject 工作区、或已经 complete）时返回 `None`。 + pub(crate) fn accumulated_text_for(&self, item_id: &str) -> Option { + self.text_by_item_id.get(item_id).cloned() + } + fn take_partial_items(&mut self) -> impl Iterator + '_ { std::mem::take(&mut self.text_by_item_id) .into_iter() diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_runtime.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_runtime.rs index 69c95a8ad..81358e479 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_runtime.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_runtime.rs @@ -1,5 +1,6 @@ use super::*; use base64::Engine as _; +use std::collections::BTreeMap; use std::collections::HashMap; use std::future::Future; use std::io::Write; @@ -4102,7 +4103,9 @@ fn sync_direct_codex_project_outputs_at( /// Project Codex text for the user-visible DirectProject stream and reply. /// Reasoning wrappers are still removed because they are not reply text, but /// the user owns the project and the resulting reply is not redacted here. -fn project_direct_codex_visible_text(value: &str) -> Option { +/// +/// `pub(crate)`：回合流（`direct_turn_stream`）落最终回复前复用同一套可见性投影。 +pub(crate) fn project_direct_codex_visible_text(value: &str) -> Option { let stripped = strip_incomplete_direct_thinking_marker(&strip_llm_thinking_blocks(value)); if stripped.trim().is_empty() { return None; @@ -4354,7 +4357,19 @@ async fn run_direct_game_creator_turn_at_with_creation_type_and_emitter( let _ = persist_direct_codex_failure_context(root, emitter.turn_id(), &error); } if let Some(emitter) = turn_emitter { - emitter.emit("failed", Some("none"), None, None); + // 失败说明也是这一回合的内容：按出现顺序追加到回合流末尾， + // 这样"流里已经是完整内容"这一点对失败回合同样成立。 + let failure_item = append_direct_turn_stream_text_at( + root, + emitter.turn_id(), + DIRECT_TURN_STREAM_FAILURE_ITEM_ID, + &error, + ) + .ok() + .flatten() + .into_iter() + .collect::>(); + emitter.emit_with_stream_items("failed", Some("none"), None, None, failure_item); } Err(error) } @@ -4397,6 +4412,125 @@ fn persist_collected_direct_tool_calls(root: &Path, collector: &DirectToolCallCo } } +/// 回合流条目落盘：与工具调用同一口径（阻塞线程池 + 项目锁）。 +fn spawn_persist_direct_turn_stream_item(root: &Path, item: &DirectTurnStreamItem) { + let root = root.to_path_buf(); + let item = item.clone(); + tauri::async_runtime::spawn_blocking(move || upsert_direct_turn_stream_item_at(&root, &item)); +} + +/// 文本段落盘/下发的节流间隔：文本段是"整段累计 + 原地替换"，不需要逐 delta 落盘。 +const DIRECT_TURN_STREAM_TEXT_THROTTLE_MS: u128 = 300; + +/// 正在增长的那一段文本。 +struct DirectTurnStreamPendingText { + /// 这一段对应的 Codex assistant item id（段身份）。 + item_id: String, + item: DirectTurnStreamItem, + last_flush: std::time::Instant, +} + +/// 回合流的写入与下发状态（观察者持有）。 +/// +/// `seq_by_id` 是**顺序真相的本体**：条目 id 第一次出现时分配序号，之后所有更新都带同一个 +/// 序号，所以并发落盘的先后不会改变渲染顺序（不会出现"新工具插到旧文本前面"）。 +struct DirectTurnStreamWriter { + turn_id: String, + seq_by_id: BTreeMap, + next_seq: u64, + pending_text: Option, +} + +impl DirectTurnStreamWriter { + fn new(turn_id: String) -> Self { + Self { + turn_id, + seq_by_id: BTreeMap::new(), + next_seq: 0, + pending_text: None, + } + } + + /// 条目 id 对应的固定序号：首次出现时分配，之后永远不变。 + fn seq_for(&mut self, id: &str) -> u64 { + if let Some(seq) = self.seq_by_id.get(id) { + return *seq; + } + self.next_seq += 1; + self.seq_by_id.insert(id.to_string(), self.next_seq); + self.next_seq + } + + /// 文本段推进。返回需要落盘 + 下发的那份快照（节流窗口内返回 `None`）。 + /// + /// 段身份（`item_id`）变化时先把上一段的收尾快照交出去：上一段最后一段文字不能丢。 + fn push_text( + &mut self, + root: &Path, + item_id: &str, + visible_text: &str, + now_ms: u64, + ) -> Option { + let now = std::time::Instant::now(); + let mut flushed = None; + if self + .pending_text + .as_ref() + .is_some_and(|pending| pending.item_id != item_id) + { + flushed = self.take_pending_snapshot(); + } + match self.pending_text.as_mut() { + Some(pending) if pending.item_id == item_id => { + pending.item.text = Some(sanitize_stream_text(root, visible_text)); + pending.item.updated_at = now_ms.max(pending.item.updated_at); + if now.duration_since(pending.last_flush).as_millis() + >= DIRECT_TURN_STREAM_TEXT_THROTTLE_MS + { + pending.last_flush = now; + return Some(pending.item.clone()); + } + } + _ => { + let seq = self.seq_for(&direct_turn_stream_text_item_id(&self.turn_id, item_id)); + let item = direct_turn_stream_text_item( + root, + &self.turn_id, + item_id, + visible_text, + seq, + now_ms, + now_ms, + ); + self.pending_text = Some(DirectTurnStreamPendingText { + item_id: item_id.to_string(), + item, + last_flush: now, + }); + // 新段一出现就立刻落盘 + 下发：位置由这一刻钉死。 + return self.pending_text.as_ref().map(|pending| pending.item.clone()); + } + } + flushed + } + + /// 取出当前段的收尾快照（段结束 / 回合结束时调用），不再持有它。 + fn take_pending_snapshot(&mut self) -> Option { + self.pending_text.take().map(|pending| pending.item) + } + + /// 工具条目：只记位置，正文仍来自 `tool-calls.jsonl`。 + fn push_tool(&mut self, call: &DirectToolCall, now_ms: u64) -> DirectTurnStreamItem { + let seq = self.seq_for(&direct_turn_stream_tool_item_id(&self.turn_id, &call.id)); + let at = if call.started_at > 0 { + call.started_at + } else { + now_ms + }; + direct_turn_stream_tool_item(&self.turn_id, call, seq, at) + } +} + async fn run_direct_game_creator_turn_inner( root: &Path, prompt: &str, @@ -4426,6 +4560,8 @@ async fn run_direct_game_creator_turn_inner( let emitter = emitter.clone(); let turn_root = root.to_path_buf(); let turn_tool_calls = Arc::clone(&tool_calls); + // 回合流：文本段与工具按**出现顺序**各占一行，位置（seq）在首次出现时钉死。 + let mut stream_writer = DirectTurnStreamWriter::new(client_turn_id.clone()); let mut observer = move |observation: DirectCodexTurnObservation| { let status = direct_codex_observation_status(&observation, stream_enabled); match observation { @@ -4437,6 +4573,27 @@ async fn run_direct_game_creator_turn_inner( } emitter.emit(status, None, visible_text, None); } + DirectCodexTurnObservation::AgentMessageSegment { + item_id, + accumulated_text, + } => { + // 可见文本段：同一 item 的后续 delta 就地增长，item 变了才新起一段。 + let Some(visible_text) = project_direct_codex_visible_text(&accumulated_text) + else { + return; + }; + let Some(item) = stream_writer.push_text( + &turn_root, + &item_id, + &visible_text, + direct_tool_call_now_ms(), + ) else { + return; + }; + // 事件与落盘共用同一份快照：前端的顺序真相与文件里的顺序真相一致。 + spawn_persist_direct_turn_stream_item(&turn_root, &item); + emitter.emit_with_stream_items(status, None, None, None, vec![item]); + } DirectCodexTurnObservation::IntermediateText(intermediate_text) => { let visible_text = if stream_enabled || is_direct_codex_item_started_work_detail(&intermediate_text) @@ -4480,7 +4637,16 @@ async fn run_direct_game_creator_turn_inner( collected.retain(|existing| existing.id != tool_call.id); collected.push(tool_call.clone()); } - emitter.emit(status, None, None, Some(vec![tool_call.clone()])); + // 回合流：工具是**普通元素**，位置在文本段之后（或与相邻工具成块）。 + let stream_item = stream_writer.push_tool(&tool_call, direct_tool_call_now_ms()); + spawn_persist_direct_turn_stream_item(&turn_root, &stream_item); + emitter.emit_with_stream_items( + status, + None, + None, + Some(vec![tool_call.clone()]), + vec![stream_item], + ); // 落盘"最新的那一份"：started 让卡片刷新后立刻出现，终态覆盖同一行。 spawn_persist_direct_tool_call(&turn_root, &tool_call); } @@ -4517,11 +4683,23 @@ async fn run_direct_game_creator_turn_inner( ) })?; if let Some(emitter) = turn_emitter { - emitter.emit( + // 最终回复落到本回合最后一条文本段上（原地更新，不新起一段），并随事件下发： + // 前端据此把最后一段替换成最终可见回复，流式尾巴与最终回复不会重复。 + let finalized = finalize_direct_turn_stream_reply_at( + root, + emitter.turn_id(), + &visible_reply, + ) + .ok() + .flatten() + .into_iter() + .collect::>(); + emitter.emit_with_stream_items( "finalizing", Some("response-finalization"), Some(visible_reply.clone()), None, + finalized, ); } if direct_codex_output_fingerprint(root) != previous_output_fingerprint { diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_tool_calls.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_tool_calls.rs index bf9e35b05..b1dae74b0 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_tool_calls.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_tool_calls.rs @@ -205,7 +205,9 @@ fn relativize_project_root_paths(root: &Path, value: &str) -> String { /// `Authorization: Bearer …`、`Cookie: …`、`api_key=…`、`client_secret=…` 这类键值凭据； /// 含 `--password` / `--token` / `--secret` 这类敏感 CLI 标志的行按既有 fail-closed /// 约定整行替换成 `[redacted sensitive context]`（与 `sanitize_agent_runtime_text` 一致）。 -fn sanitize_detail_text(root: &Path, value: &str) -> String { +/// +/// `pub(crate)`：回合流（`direct_turn_stream`）的文本段复用同一套脱敏，避免两处口径分叉。 +pub(crate) fn sanitize_detail_text(root: &Path, value: &str) -> String { let without_project_root = relativize_project_root_paths(root, value); let without_absolute = redact_absolute_path_tokens(&without_project_root); let without_secret = redact_secret_tokens(&without_absolute); diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/direct_turn_stream.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_turn_stream.rs new file mode 100644 index 000000000..4b9bdbfb7 --- /dev/null +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/direct_turn_stream.rs @@ -0,0 +1,401 @@ +//! GameAgent 对话「回合流」的采集、持久化与回读。 +//! +//! 顺序真相放在一处：`/.agent/conversations/turn-stream.jsonl` 按**出现顺序** +//! 记录一个回合里的文本段与工具调用。工具条目只记位置标记（`callId`），工具本身的正文 +//! 仍然来自 `tool-calls.jsonl`（同一 id 幂等合并只有一处实现）。 +//! +//! 位置稳定：每条条目的 `seq` 在**首次出现**时由观察方分配并落盘，后续更新（同一 id 的 +//! 文本追加 / 工具状态变化）只改内容不改 `seq`。因此并发落盘的先后顺序不会让"新工具插到 +//! 旧文本前面"——渲染顺序只由 `seq` 决定。 +//! +//! 为什么不复用 `project.jsonl`：那条链路一个回合只投影一条 assistant 消息（整段回复）， +//! 中途的文本段没有独立记录，装不下"文本段与工具交替"的顺序。 + +use crate::agent::sanitize_detail_text; +use crate::config::{prepare_game_creator_private_path_for_read, write_game_creator_private_file}; +use crate::project::{enforce_project_permission_policy, project_append_lock_for}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::BTreeMap; +use std::fs::File; +use std::io::{BufRead, BufReader}; +use std::path::{Path, PathBuf}; + +/// 行信封类型，与既有历史文件同构（`{"type": …, "payload": {…}}`）。 +pub(crate) const DIRECT_TURN_STREAM_RECORD_TYPE: &str = "turn_stream_item"; +/// 条目 schema 版本。 +pub(crate) const DIRECT_TURN_STREAM_SCHEMA_VERSION: &str = "agc-turn-stream.v1"; +/// 回读上限：只保留最后这么多条（按 `seq` 取最新）。 +pub(crate) const DIRECT_TURN_STREAM_LIMIT: usize = 400; +/// 单条文本段的字符上限（与工具明细同口径的截断，避免单段失控）。 +const DIRECT_TURN_STREAM_TEXT_MAX_CHARS: usize = 8000; +/// 没有流式分段时，最终回复那一段的固定 item id。 +const DIRECT_TURN_STREAM_FINAL_ITEM_ID: &str = "final"; +/// 回合失败说明那一段的固定 item id：失败说明也是这一回合的内容，排在流末尾。 +pub(crate) const DIRECT_TURN_STREAM_FAILURE_ITEM_ID: &str = "failure"; + +/// 文本段。 +pub(crate) const DIRECT_TURN_STREAM_KIND_TEXT: &str = "text"; +/// 工具调用的位置标记。 +pub(crate) const DIRECT_TURN_STREAM_KIND_TOOL: &str = "tool"; + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct DirectTurnStreamItem { + pub(crate) schema_version: String, + /// 幂等身份：文本段 `text::`、工具 `tool::`。 + pub(crate) id: String, + pub(crate) turn_id: String, + /// `text` | `tool` + pub(crate) kind: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) text: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) call_id: Option, + /// 首次出现的写入序号：**顺序真相**，同刻按它排序。 + pub(crate) seq: u64, + /// 条目首次出现的本机毫秒时刻。 + pub(crate) at: u64, + pub(crate) updated_at: u64, +} + +impl DirectTurnStreamItem { + fn order_key(&self) -> (u64, u64, &str) { + (self.seq, self.at, self.id.as_str()) + } +} + +fn turn_stream_path(root: &Path) -> PathBuf { + root.join(".agent/conversations/turn-stream.jsonl") +} + +/// 文本段条目的幂等 id：同一个 Codex assistant item 只占一行。 +pub(crate) fn direct_turn_stream_text_item_id(turn_id: &str, item_id: &str) -> String { + format!("text:{}:{}", turn_id.trim(), item_id.trim()) +} + +/// 工具条目（位置标记）的幂等 id：同一个 callId 只占一行。 +pub(crate) fn direct_turn_stream_tool_item_id(turn_id: &str, call_id: &str) -> String { + format!("tool:{}:{}", turn_id.trim(), call_id.trim()) +} + +/// 构造一条文本段条目：脱敏 + 截断与 `tool-calls.jsonl` 同口径。 +pub(crate) fn direct_turn_stream_text_item( + root: &Path, + turn_id: &str, + item_id: &str, + text: &str, + seq: u64, + at: u64, + updated_at: u64, +) -> DirectTurnStreamItem { + DirectTurnStreamItem { + schema_version: DIRECT_TURN_STREAM_SCHEMA_VERSION.to_string(), + id: direct_turn_stream_text_item_id(turn_id, item_id), + turn_id: turn_id.trim().to_string(), + kind: DIRECT_TURN_STREAM_KIND_TEXT.to_string(), + text: Some(sanitize_stream_text(root, text)), + call_id: None, + seq, + at, + updated_at, + } +} + +/// 构造一条工具条目：只记位置，正文仍来自 `DirectToolCall`。 +pub(crate) fn direct_turn_stream_tool_item( + turn_id: &str, + call: &crate::DirectToolCall, + seq: u64, + at: u64, +) -> DirectTurnStreamItem { + DirectTurnStreamItem { + schema_version: DIRECT_TURN_STREAM_SCHEMA_VERSION.to_string(), + id: direct_turn_stream_tool_item_id(turn_id, &call.id), + turn_id: turn_id.trim().to_string(), + kind: DIRECT_TURN_STREAM_KIND_TOOL.to_string(), + text: None, + call_id: Some(call.id.trim().to_string()), + seq, + at, + updated_at: call.updated_at, + } +} + +/// 文本脱敏 + 截断：与 `tool-calls.jsonl` 同一套 `sanitize_detail_text`。 +pub(crate) fn sanitize_stream_text(root: &Path, text: &str) -> String { + let sanitized = sanitize_detail_text(root, text); + if sanitized.chars().count() <= DIRECT_TURN_STREAM_TEXT_MAX_CHARS { + return sanitized; + } + let mut truncated = sanitized + .chars() + .take(DIRECT_TURN_STREAM_TEXT_MAX_CHARS) + .collect::(); + truncated.push('…'); + truncated +} + +fn record_line(item: &DirectTurnStreamItem) -> Result { + serde_json::to_string(&serde_json::json!({ + "type": DIRECT_TURN_STREAM_RECORD_TYPE, + "payload": item, + })) + .map_err(|error| format!("序列化回合流条目失败：{error}")) +} + +/// 解析一行信封；坏行 / 非本文件条目都返回 `None`（尽力而为的展示数据，不整体失败）。 +fn stream_item_from_line(line: &str) -> Option { + let trimmed = line.trim(); + if trimmed.is_empty() { + return None; + } + let parsed: Value = serde_json::from_str(trimmed).ok()?; + if parsed.get("type").and_then(Value::as_str) != Some(DIRECT_TURN_STREAM_RECORD_TYPE) { + return None; + } + let payload = parsed.get("payload")?; + let mut item: DirectTurnStreamItem = serde_json::from_value(payload.clone()).ok()?; + if item.id.trim().is_empty() || item.turn_id.trim().is_empty() { + return None; + } + if !matches!( + item.kind.as_str(), + DIRECT_TURN_STREAM_KIND_TEXT | DIRECT_TURN_STREAM_KIND_TOOL + ) { + return None; + } + if item.schema_version.trim().is_empty() { + item.schema_version = DIRECT_TURN_STREAM_SCHEMA_VERSION.to_string(); + } + Some(item) +} + +fn read_stream_lines(path: &Path) -> Vec { + let Ok(file) = File::open(path) else { + return Vec::new(); + }; + let mut reader = BufReader::new(file); + let mut buffer = Vec::new(); + let mut items = Vec::new(); + loop { + buffer.clear(); + match reader.read_until(b'\n', &mut buffer) { + Ok(0) => break, + // 单行解码失败（非法 UTF-8）只跳过这一行，继续读后面的行。 + Ok(_) => match std::str::from_utf8(&buffer) { + Ok(line) => { + if let Some(item) = stream_item_from_line(line) { + items.push(item); + } + } + Err(_) => continue, + }, + // 读 I/O 错误：无法再定位下一行边界，停止读取（已读到的照常返回）。 + Err(_) => break, + } + } + items +} + +/// 同一 id 的重复行合并：`seq` 取最早（位置钉死，后到的不得回退），`at` 取最早非零， +/// `updated_at` 取最大；文本只在更新（或同刻更长）的快照上替换。 +fn merge_stream_snapshot( + existing: &DirectTurnStreamItem, + incoming: &DirectTurnStreamItem, +) -> DirectTurnStreamItem { + let text_len = |item: &DirectTurnStreamItem| { + item.text + .as_deref() + .map(str::chars) + .map(Iterator::count) + .unwrap_or_default() + }; + let take_incoming = incoming.updated_at > existing.updated_at + || (incoming.updated_at == existing.updated_at && text_len(incoming) > text_len(existing)); + let mut merged = existing.clone(); + if take_incoming { + merged.text = incoming.text.clone(); + merged.updated_at = incoming.updated_at; + } + if merged.call_id.is_none() { + merged.call_id = incoming.call_id.clone(); + } + merged.seq = merged.seq.min(incoming.seq); + merged.at = [merged.at, incoming.at] + .into_iter() + .filter(|at| *at > 0) + .min() + .unwrap_or_default(); + merged +} + +/// 按 id 归并，再按 `seq` 正序裁剪到最近 `DIRECT_TURN_STREAM_LIMIT` 条。 +fn normalize_stream_items(items: Vec) -> Vec { + let mut by_id: BTreeMap = BTreeMap::new(); + for item in items { + let merged = match by_id.remove(&item.id) { + Some(previous) => merge_stream_snapshot(&previous, &item), + None => item, + }; + by_id.insert(merged.id.clone(), merged); + } + let mut normalized = by_id.into_values().collect::>(); + normalized.sort_by(|left, right| left.order_key().cmp(&right.order_key())); + if normalized.len() > DIRECT_TURN_STREAM_LIMIT { + normalized.drain(..normalized.len() - DIRECT_TURN_STREAM_LIMIT); + } + normalized +} + +/// 锁内读改写：整文件重写（追加与就地更新混用，没有纯追加的 JSONL 语义）。 +/// 文件规模由 400 条上限与 8000 字符截断兜住。 +fn with_locked_stream_items( + root: &Path, + mutate: impl FnOnce(&mut Vec) -> T, +) -> Result { + let path = turn_stream_path(root); + let _project_lock = crate::acquire_game_creator_agent_runtime_project_write_lock_with_wait( + root, + "conversation.write", + )?; + let lock = project_append_lock_for(&path)?; + let _append_guard = lock.lock("回合流写入")?; + let mut items = read_stream_lines(&path); + let outcome = mutate(&mut items); + let normalized = normalize_stream_items(items); + let mut body = String::new(); + for item in &normalized { + body.push_str(&record_line(item)?); + body.push('\n'); + } + write_game_creator_private_file(&path, body.as_bytes(), "回合流历史")?; + Ok(outcome) +} + +/// 幂等 upsert 一条回合流条目。 +/// +/// 位置（`seq` / `at`）只在第一次出现时确定：同一 id 的后续快照不得回退位置， +/// 也不得把已经写下的文本改短（并发落盘下"后到的旧快照"不会覆盖新快照）。 +pub(crate) fn upsert_direct_turn_stream_item_at( + root: &Path, + item: &DirectTurnStreamItem, +) -> Result<(), String> { + enforce_project_permission_policy(root, "conversation.write")?; + with_locked_stream_items(root, |items| { + items.retain(|existing| existing.id != item.id); + items.push(item.clone()); + }) +} + +/// 回读：文件缺失返回空数组；单行损坏跳过；按 `seq` 正序，最多最后 400 条。 +pub(crate) fn read_direct_turn_stream_at(root: &Path) -> Result, String> { + let path = turn_stream_path(root); + if !prepare_game_creator_private_path_for_read(&path, false, "回合流历史")? { + return Ok(Vec::new()); + } + Ok(normalize_stream_items(read_stream_lines(&path))) +} + +/// 追加一段固定身份的文本段（失败说明等）：位置排在当前流末尾。 +/// +/// 幂等：同一 `(turnId, itemId)` 已经存在时只更新文本与 `updatedAt`（回合重放 / 重复收尾 +/// 不会多出一段）。返回写下的那一条，调用方用它下发同一份快照。 +pub(crate) fn append_direct_turn_stream_text_at( + root: &Path, + turn_id: &str, + item_id: &str, + text: &str, +) -> Result, String> { + let turn_id = turn_id.trim(); + let text = text.trim(); + if turn_id.is_empty() || text.is_empty() { + return Ok(None); + } + let sanitized = sanitize_stream_text(root, text); + let item_id = item_id.trim(); + enforce_project_permission_policy(root, "conversation.write")?; + let now = crate::agent::direct_tool_call_now_ms(); + with_locked_stream_items(root, |items| { + let existing_id = direct_turn_stream_text_item_id(turn_id, item_id); + if let Some(existing) = items.iter_mut().find(|item| item.id == existing_id) { + // 位置不动：只替换文本与 updatedAt。 + existing.text = Some(sanitized.clone()); + existing.updated_at = now.max(existing.updated_at); + return Some(existing.clone()); + } + // 首次出现：位置钉在末尾（当前最大 seq + 1）。 + let next_seq = items.iter().map(|item| item.seq).max().unwrap_or(0) + 1; + let item = DirectTurnStreamItem { + schema_version: DIRECT_TURN_STREAM_SCHEMA_VERSION.to_string(), + id: existing_id, + turn_id: turn_id.to_string(), + kind: DIRECT_TURN_STREAM_KIND_TEXT.to_string(), + text: Some(sanitized), + call_id: None, + seq: next_seq, + at: now, + updated_at: now, + }; + items.push(item.clone()); + Some(item) + }) +} + +/// 回合结束：把**可见回复**落到本回合最后一条文本段上。 +/// +/// - 本回合已有文本段（流式逐段落的那些）：原地把最后一段替换成最终可见回复， +/// 不新起一段——否则最终回复会和最后一段重复。 +/// - 本回合没有文本段（非流式、或整轮没有文本）：追加一段，位置排在最后。 +/// +/// `visible_reply` 由调用方先做可见性投影（去思考块、去空）。返回被写入的那一条， +/// 调用方用它下发同一份快照。 +pub(crate) fn finalize_direct_turn_stream_reply_at( + root: &Path, + turn_id: &str, + visible_reply: &str, +) -> Result, String> { + let turn_id = turn_id.trim(); + if turn_id.is_empty() || visible_reply.trim().is_empty() { + return Ok(None); + } + // 入口再做一次可见性投影：调用方给的是原始回复时，思考块不能落进对话流。 + let visible_reply = crate::agent::project_direct_codex_visible_text(visible_reply) + .unwrap_or_else(|| visible_reply.trim().to_string()); + let visible_reply = visible_reply.as_str(); + enforce_project_permission_policy(root, "conversation.write")?; + let now = crate::agent::direct_tool_call_now_ms(); + with_locked_stream_items(root, |items| { + let last_text_id = items + .iter() + .filter(|item| item.turn_id == turn_id && item.kind == DIRECT_TURN_STREAM_KIND_TEXT) + .max_by(|left, right| left.order_key().cmp(&right.order_key())) + .map(|item| item.id.clone()); + match last_text_id { + Some(id) => { + let item = items + .iter_mut() + .find(|item| item.id == id) + .expect("最后一条文本段必须还在集合里"); + // 位置不动：只替换文本与 updatedAt（原地更新，不重排）。 + item.text = Some(sanitize_stream_text(root, visible_reply)); + item.updated_at = now.max(item.updated_at); + Some(item.clone()) + } + None => { + let next_seq = items.iter().map(|item| item.seq).max().unwrap_or(0) + 1; + let item = direct_turn_stream_text_item( + root, + turn_id, + DIRECT_TURN_STREAM_FINAL_ITEM_ID, + visible_reply, + next_seq, + now, + now, + ); + items.push(item.clone()); + Some(item) + } + } + }) +} diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_driver/entrypoints.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_driver/entrypoints.rs index 9089d7748..afad36523 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_driver/entrypoints.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_driver/entrypoints.rs @@ -62,6 +62,48 @@ impl DirectGameCreatorTurnUpdateEmitter { accumulated_text: Option, tool_calls: Option>, reasoning_text: Option, + ) { + self.emit_full( + status, + activity, + accumulated_text, + tool_calls, + reasoning_text, + Vec::new(), + ); + } + + /// 带回合流的回合更新：`stream_items` 是"顺序真相"里本次变化的那几条。 + /// + /// 前端按这些条目的 `seq` 顺序渲染，所以它们必须来自与落盘同一份数据， + /// 不能在前端各算一套顺序。 + pub(crate) fn emit_with_stream_items( + &self, + status: &'static str, + activity: Option<&'static str>, + accumulated_text: Option, + tool_calls: Option>, + stream_items: Vec, + ) { + self.emit_full( + status, + activity, + accumulated_text, + tool_calls, + None, + stream_items, + ); + } + + #[allow(clippy::too_many_arguments)] + fn emit_full( + &self, + status: &'static str, + activity: Option<&'static str>, + accumulated_text: Option, + tool_calls: Option>, + reasoning_text: Option, + stream_items: Vec, ) { let status_is_allowed = matches!( status, @@ -107,6 +149,7 @@ impl DirectGameCreatorTurnUpdateEmitter { accumulated_text, tool_calls, reasoning_text, + stream_items: (!stream_items.is_empty()).then_some(stream_items), updated_at, }, ); diff --git a/apps/ai-game-creator-shell/src-tauri/src/commands.rs b/apps/ai-game-creator-shell/src-tauri/src/commands.rs index a385eca42..fdde71030 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/commands.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/commands.rs @@ -5311,6 +5311,19 @@ pub(crate) async fn read_direct_tool_calls( .map_err(|error| format!("读取工具调用历史后台任务失败：{error}"))? } +#[tauri::command] +pub(crate) async fn read_direct_turn_stream( + project_path: String, +) -> Result, String> { + tauri::async_runtime::spawn_blocking(move || { + let root = Path::new(project_path.trim()); + enforce_project_permission_policy(root, "conversation.read")?; + read_direct_turn_stream_at(root) + }) + .await + .map_err(|error| format!("读取回合流历史后台任务失败：{error}"))? +} + #[tauri::command] pub(crate) fn append_local_conversation_message( project_path: String, diff --git a/apps/ai-game-creator-shell/src-tauri/src/main.rs b/apps/ai-game-creator-shell/src-tauri/src/main.rs index 426a713c5..b69d8f19b 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/main.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/main.rs @@ -1017,6 +1017,10 @@ struct GameCreatorDirectTurnUpdateEvent { /// 本回合当前累计的思考过程（流式整段替换）；拿不到时字段缺席。 #[serde(skip_serializing_if = "Option::is_none")] reasoning_text: Option, + /// 本回合**顺序真相**里本次发生变化的那几条（文本段 / 工具位置标记）。 + /// `skip_serializing_if`：字段缺席时前端拿到 `undefined`，行为与改造前一致。 + #[serde(skip_serializing_if = "Option::is_none")] + stream_items: Option>, updated_at: u64, } @@ -2779,6 +2783,7 @@ fn main() { read_local_conversation, read_direct_project_conversation, read_direct_tool_calls, + read_direct_turn_stream, read_agent_runtime_error_detail, append_local_conversation_message, append_direct_project_conversation_message, diff --git a/apps/ai-game-creator-shell/src/App.tsx b/apps/ai-game-creator-shell/src/App.tsx index f4498546d..95053dc3e 100644 --- a/apps/ai-game-creator-shell/src/App.tsx +++ b/apps/ai-game-creator-shell/src/App.tsx @@ -99,6 +99,7 @@ import type { ProjectPermissionPolicyView, SyncCanvasProjectAssetsResult, TauriInvoke, + TurnStreamItem, UploadLocalAssetResult, } from './app/types'; import { useWindowChrome } from './components/windowChromeContext'; @@ -504,6 +505,68 @@ function directCodexTurnIdFromAssistantMessageId(messageId: string) { export const MAX_CHAT_COMPOSER_ATTACHMENTS = 8; +/** + * 回合流排序：`seq`（条目首次出现时钉死）优先，其次 `at`，最后按 id 兜底。 + * 与 Rust 侧同一口径——前端不自己发明顺序。 + */ +function sortTurnStreamItems(items: readonly TurnStreamItem[]) { + return [...items].sort( + (left, right) => + left.seq - right.seq || + left.at - right.at || + left.id.localeCompare(right.id), + ); +} + +/** + * 归并一批回合流条目：同 id 幂等覆盖（`updatedAt` 单调，同刻取更长文本），新 id 追加。 + * 实时增量与回读历史共用这一处，所以界面上的顺序只有一份来源。 + */ +function mergeTurnStreamItems( + existing: readonly TurnStreamItem[], + incoming: readonly TurnStreamItem[], +): TurnStreamItem[] { + if (incoming.length === 0) { + return [...existing]; + } + const byId = new Map(); + for (const item of existing) { + const id = item.id?.trim(); + if (id) { + byId.set(id, item); + } + } + for (const item of incoming) { + const id = item.id?.trim(); + if (!id) { + continue; + } + const previous = byId.get(id); + const normalized: TurnStreamItem = { ...item, id }; + if (!previous) { + byId.set(id, normalized); + continue; + } + // 内容只在更新（或同刻更长）的快照上替换；`seq` 取最早，位置不许回退。 + const textLength = (value: TurnStreamItem) => + value.kind === 'text' ? (value.text?.length ?? 0) : 0; + const takeIncoming = + normalized.updatedAt > previous.updatedAt || + (normalized.updatedAt === previous.updatedAt && + textLength(normalized) > textLength(previous)); + byId.set(id, { + ...(takeIncoming ? normalized : previous), + id, + seq: Math.min(previous.seq, normalized.seq), + at: + previous.at > 0 && normalized.at > 0 + ? Math.min(previous.at, normalized.at) + : Math.max(previous.at, normalized.at), + } as TurnStreamItem); + } + return sortTurnStreamItems([...byId.values()]); +} + export function isDirectCodexTurnAlreadyRunningError(error: unknown) { const message = error instanceof Error ? error.message : String(error); return message @@ -884,13 +947,6 @@ export function App({ const [directCodexTransientReasoning, setDirectCodexTransientReasoning] = useState(''); /** 实时回合里"某个工具首次出现时，已生成正文的长度"——用它把正文与工具交替排列。 */ - const [directToolTextOffsets, setDirectToolTextOffsets] = useState< - Record> - >({}); - const directToolTextOffsetsRef = useRef>>( - {}, - ); - const directAccumulatedTextLengthRef = useRef(0); const [ directCodexTransientReplyUpdatedAt, setDirectCodexTransientReplyUpdatedAt, @@ -992,9 +1048,6 @@ export function App({ setDirectCodexProgressUpdatedAt(null); setDirectCodexTransientReply(''); setDirectCodexTransientReasoning(''); - directToolTextOffsetsRef.current = {}; - setDirectToolTextOffsets({}); - directAccumulatedTextLengthRef.current = 0; directCodexTransientReplyRef.current = ''; setDirectCodexTransientReplyUpdatedAt(null); } @@ -1094,9 +1147,6 @@ export function App({ setDirectCodexProgressUpdatedAt(Date.now()); setDirectCodexTransientReply(''); setDirectCodexTransientReasoning(''); - directToolTextOffsetsRef.current = {}; - setDirectToolTextOffsets({}); - directAccumulatedTextLengthRef.current = 0; directCodexTransientReplyRef.current = ''; setDirectCodexTransientReplyUpdatedAt(null); setProjectSupervisorRuntimeError(''); @@ -1120,6 +1170,32 @@ export function App({ setDirectToolCalls(normalized); } + // 回合流（文本段 + 工具按**出现顺序**交替）：实时增量与回读历史共用一份状态， + // 渲染顺序只由条目的 `seq` 决定，界面不再按文本长度 / 标点 / 时间窗猜切点。 + const [directTurnStream, setDirectTurnStream] = useState( + [], + ); + const directTurnStreamRef = useRef([]); + + /** 归并一批回合流条目（实时事件里的 `streamItems`）。 */ + function applyTurnStreamItems(incoming: readonly TurnStreamItem[]) { + if (incoming.length === 0) { + return; + } + const merged = mergeTurnStreamItems(directTurnStreamRef.current, incoming); + directTurnStreamRef.current = merged; + setDirectTurnStream(merged); + } + + /** 回合流按项目维度作废：换项目 / 重开历史时整体替换，避免串项目。 */ + function replaceTurnStreamItems(next: readonly TurnStreamItem[]) { + const normalized = sortTurnStreamItems( + next.filter((item) => Boolean(item.id?.trim())), + ); + directTurnStreamRef.current = normalized; + setDirectTurnStream(normalized); + } + function clearDirectCodexTransientReply(projectPath: string, turnId: string) { const activeTurn = activeDirectCodexTurnRef.current; if ( @@ -1131,9 +1207,6 @@ export function App({ activeDirectCodexTurnRef.current = null; setDirectCodexTransientReply(''); setDirectCodexTransientReasoning(''); - directToolTextOffsetsRef.current = {}; - setDirectToolTextOffsets({}); - directAccumulatedTextLengthRef.current = 0; directCodexTransientReplyRef.current = ''; setDirectCodexTransientReplyUpdatedAt(null); return true; @@ -2205,29 +2278,7 @@ export function App({ ); } // 工具调用增量：字段可选，老事件（undefined）走原路径，行为不变。 - if (typeof payload.accumulatedText === 'string') { - directAccumulatedTextLengthRef.current = payload.accumulatedText.length; - } if (payload.toolCalls?.length) { - // 每个工具只记第一次出现时的正文长度：这就是它在对话流里的位置。 - const turnOffsets = - directToolTextOffsetsRef.current[payload.turnId] ?? {}; - let offsetsChanged = false; - for (const call of payload.toolCalls) { - const callId = (call as { id?: string }).id; - if (!callId || turnOffsets[callId] !== undefined) { - continue; - } - turnOffsets[callId] = directAccumulatedTextLengthRef.current; - offsetsChanged = true; - } - if (offsetsChanged) { - directToolTextOffsetsRef.current = { - ...directToolTextOffsetsRef.current, - [payload.turnId]: { ...turnOffsets }, - }; - setDirectToolTextOffsets(directToolTextOffsetsRef.current); - } applyDirectToolCalls( payload.toolCalls.map((call) => ({ ...call, @@ -2238,6 +2289,10 @@ export function App({ if (typeof payload.reasoningText === 'string') { setDirectCodexTransientReasoning(payload.reasoningText); } + // 回合流的顺序真相：字段可选，老事件（undefined）走原路径。 + if (payload.streamItems?.length) { + applyTurnStreamItems(payload.streamItems); + } const updatedAt = Number.isFinite(payload.updatedAt) && payload.updatedAt > 0 ? payload.updatedAt @@ -2251,9 +2306,6 @@ export function App({ setDirectCodexProgressUpdatedAt(updatedAt); setDirectCodexTransientReply(''); setDirectCodexTransientReasoning(''); - directToolTextOffsetsRef.current = {}; - setDirectToolTextOffsets({}); - directAccumulatedTextLengthRef.current = 0; setDirectCodexTransientReplyUpdatedAt(null); return; } @@ -3922,6 +3974,14 @@ export function App({ { projectPath: nextProjectPath }, ).catch(() => []); replaceDirectToolCalls(persistedToolCalls); + // 回合流（顺序真相）走独立历史文件（`turn-stream.jsonl`）。与工具调用同一处：必须在 + // 读完项目对话之后、任何提前 return 之前回读。缺命令（老客户端）/ 缺文件 / 读取失败 + // 都只是"这个回合没有流"，界面回退到原来的渲染，不能因此把整个打开流程判失败。 + const persistedTurnStream = await invoke( + 'read_direct_turn_stream', + { projectPath: nextProjectPath }, + ).catch(() => []); + replaceTurnStreamItems(persistedTurnStream); // 重进会话时 Rust 侧可能仍登记着上一条 Direct 回合。不接管的话界面既不显示 // 过程卡也不给终止入口，用户再发消息只会被守卫拒绝（"已有另一条回合正在运行"）。 await restoreRunningDirectCodexTurn(nextProjectPath); @@ -6902,9 +6962,6 @@ export function App({ setDirectCodexProgress('正在等待陶泥儿开始'); setDirectCodexTransientReply(''); setDirectCodexTransientReasoning(''); - directToolTextOffsetsRef.current = {}; - setDirectToolTextOffsets({}); - directAccumulatedTextLengthRef.current = 0; setDirectCodexProgressUpdatedAt(Date.now()); directCodexTransientReplyRef.current = ''; setDirectCodexTransientReplyUpdatedAt(null); @@ -12194,6 +12251,13 @@ export function App({ visibleTurnIds.has(call.turnId) || call.turnId === activeDirectCodexTurnRef.current?.turnId, ); + // 回合流只保留「当前消息列表里确实有这个回合」的那些，外加正在跑的回合； + // 没有流条目的回合由视图自己回退到原来的渲染。 + const visibleTurnStreamItems = directTurnStream.filter( + (item) => + visibleTurnIds.has(item.turnId) || + item.turnId === activeDirectCodexTurnRef.current?.turnId, + ); const projectSupervisorTransientReply = projectSupervisorResponseStream?.accumulatedText.trim() ?? ''; const projectSupervisorNeedsUserInput = agentRuntimeNeedsUserInput( @@ -12639,7 +12703,6 @@ export function App({ if (projectSupervisorOnly) { return ( :`、工具 `tool::`。 */ + id: string; + turnId: string; + /** 首次出现的写入序号：**顺序真相**，按它升序渲染。 */ + seq: number; + /** 条目首次出现的时刻（Unix 毫秒），同 `seq` 时用它排序。 */ + at: number; + updatedAt: number; +} + +/** + * 回合流条目（`read_direct_turn_stream` 的返回元素）。 + * + * 与 Rust `DirectTurnStreamItem` 同形：`text` 段 ↔ `tool` 位置标记。 + */ +export type TurnStreamItem = TurnStreamTextItem | TurnStreamToolItem; + /** * `read_direct_codex_active_turn` 的返回值：Rust 进程内当前登记的 Direct 活跃回合。 * diff --git a/apps/ai-game-creator-shell/src/features/project-workspace/ProjectSupervisorView.tsx b/apps/ai-game-creator-shell/src/features/project-workspace/ProjectSupervisorView.tsx index e50ee4028..440d43832 100644 --- a/apps/ai-game-creator-shell/src/features/project-workspace/ProjectSupervisorView.tsx +++ b/apps/ai-game-creator-shell/src/features/project-workspace/ProjectSupervisorView.tsx @@ -17,6 +17,7 @@ import type { PendingUiConfirmation, PlanGddDecisionAction, PlanGddStateViewV1, + TurnStreamItem, } from '../../app/types'; import type { DesignClarificationRequest, @@ -90,6 +91,129 @@ function directCodexTurnIdFromAssistantMessageId(messageId: string) { return messageId.slice(prefix.length, -suffix.length); } +/** 从 `direct-codex::user` 反解回合 id；不是这个形状返回 `null`。 */ +function directCodexTurnIdFromUserMessageId(messageId: string) { + const prefix = 'direct-codex:'; + const suffix = ':user'; + if (!messageId.startsWith(prefix) || !messageId.endsWith(suffix)) { + return null; + } + return messageId.slice(prefix.length, -suffix.length); +} + +/** 按回合分组，组内按 `seq` 升序（排序已在 App 侧完成，这里保持原顺序）。 */ +function groupTurnStreamItems(items: readonly TurnStreamItem[]) { + const byTurn = new Map(); + for (const item of items) { + const turnId = item.turnId?.trim(); + if (!turnId) { + continue; + } + const bucket = byTurn.get(turnId); + if (bucket) { + bucket.push(item); + } else { + byTurn.set(turnId, [item]); + } + } + return byTurn; +} + +/** 相邻的工具流项合并成一个块；中间夹了文本段就另起一块。 */ +type TurnStreamToolRun = { + kind: 'tools'; + key: string; + callIds: string[]; +}; +type TurnStreamRun = + | { kind: 'text'; key: string; text: string } + | TurnStreamToolRun; + +function turnStreamRuns(items: readonly TurnStreamItem[]) { + const runs: TurnStreamRun[] = []; + for (const item of items) { + if (item.kind === 'text') { + const text = item.text ?? ''; + if (!text.trim()) { + continue; + } + runs.push({ kind: 'text', key: item.id, text }); + continue; + } + const callId = item.callId?.trim() ?? ''; + if (!callId) { + continue; + } + const last = runs[runs.length - 1]; + if (last?.kind === 'tools') { + // 连续的工具合并成一块：块的身份用首个工具条目 id，块头顺序不变。 + last.callIds.push(callId); + continue; + } + runs.push({ kind: 'tools', key: item.id, callIds: [callId] }); + } + return runs; +} + +/** + * 一个回合的对话流：**按条目顺序**渲染文本段与工具块。 + * + * 工具是普通元素，位置在它出现的地方：文本段 → 工具块 → 文本段 → …，最后是最终回复； + * 连续的工具合并成一块，中间夹了文本就分块。顺序只来自 `seq`，不做任何切点猜测。 + * 回合末尾由调用方补 `renderTurnUsage`（结束于 xxx，总耗时 xxx）。 + */ +function TurnStreamSequence({ + items, + toolCalls, + active, + userSentAt, + className, +}: { + items: readonly TurnStreamItem[]; + toolCalls: readonly GameCreatorDirectToolCall[]; + /** 这个回合是否正在跑（决定工具行显示"执行中"）。 */ + active: boolean; + userSentAt: number; + className?: string; +}) { + const runs = turnStreamRuns(items); + const callsById = new Map(); + for (const call of toolCalls) { + const id = call.id?.trim(); + if (id && !callsById.has(id)) { + callsById.set(id, call); + } + } + return ( + <> + {runs.map((run, index) => + run.kind === 'text' ? ( +
+ +
+ ) : ( + callsById.get(callId)) + .filter((call): call is GameCreatorDirectToolCall => Boolean(call))} + userSentAt={userSentAt} + active={active} + className="message-tool-call" + /> + ), + )} + + ); +} + type RuntimePanelProps = ComponentProps; function directStatusTitle(status: string | null | undefined) { @@ -152,6 +276,13 @@ type ProjectSupervisorViewProps = RuntimePanelProps & { projectPath: string; /** 本回合（含历史回读）的工具调用卡片，按 `startedAt` 升序，同一 id 只会出现一次。 */ toolCalls?: GameCreatorDirectToolCall[]; + /** + * 「文本段 + 工具」的顺序真相（`turn-stream.jsonl` / 回合事件里的 `streamItems`）。 + * + * 有值的回合按它渲染（文本段 → 工具块 → 文本段 → …），没有值的回合回退到 + * 「工具块 + 整轮消息 + 整轮用量」的老渲染——老项目、缺文件、读取失败都不能白屏。 + */ + turnStreamItems?: TurnStreamItem[]; /** 当前正在跑的回合 id；卡片在 assistant 消息落盘前锚到它。 */ activeTurnId?: string | null; initialSupervisorMessage?: string; @@ -159,8 +290,6 @@ type ProjectSupervisorViewProps = RuntimePanelProps & { transientReply: string; /** 流式思考过程（direct-codex）：拿不到就为空，空则不渲染。 */ transientReasoning?: string; - /** 实时回合里"某工具首次出现时正文的长度"：用于把正文与工具按顺序交替渲染。 */ - directToolTextOffsets?: Record>; showDesignReasoning?: boolean; designReasoning?: string; designReasoningEntries?: DesignReasoningEntry[]; @@ -224,12 +353,12 @@ export function ProjectSupervisorView({ pendingCommand, projectPath, toolCalls = [], + turnStreamItems = [], activeTurnId = null, initialSupervisorMessage = '', showProfessionalCollaboration = true, transientReply, transientReasoning = '', - directToolTextOffsets = {}, showDesignReasoning = false, designReasoning = '', designReasoningEntries = [], @@ -303,12 +432,28 @@ export function ProjectSupervisorView({ // 工具调用折叠块按回合分组，插在**同一回合 assistant 消息之前**（Codex 是「工具在上、答复在下」）。 // 历史回合锚到自己那条 `direct-codex::assistant`，不回落到别的回合； // 正在跑的回合还没有 assistant 消息落盘，先落在消息流末尾，等那条消息落盘后回到它之前。 + // **只有没有回合流的回合才走这条路**：有流的回合按流自己的顺序渲染。 const streamTurnId = activeTurnId?.trim() ?? ''; + // 回合流（`turn-stream.jsonl`）按回合分组：有流的回合按「文本段 → 工具块 → 文本段」顺序渲染， + // 没有流的回合（老项目 / 缺文件 / 读取失败）留在老路径上，绝不白屏。 + const turnStreamByTurn = groupTurnStreamItems( + directCodex ? turnStreamItems : [], + ); const toolCallsByAnchor = new Map(); const liveToolCalls: GameCreatorDirectToolCall[] = []; let liveToolCallTurnId = ''; + // 正在跑的回合：它的流条目直接渲染在消息列表末尾，直到用户消息落盘、由那条消息接管。 + const liveStreamTurnId = streamTurnId && turnStreamByTurn.has(streamTurnId) + ? streamTurnId + : ''; + const liveStreamItems = liveStreamTurnId + ? (turnStreamByTurn.get(liveStreamTurnId) ?? []) + : []; if (directCodex) { for (const call of toolCalls) { + if (turnStreamByTurn.has(call.turnId)) { + continue; + } const expected = directCodexTurnMessageId(call.turnId, 'assistant'); const hasPersistedAssistant = visibleMessages.some( (message) => message.messageId === expected, @@ -365,7 +510,12 @@ export function ProjectSupervisorView({ const starts = toolCalls .filter((call) => call.turnId === turnId && Number(call.startedAt) > 0) .map((call) => Number(call.startedAt)); - return starts.length > 0 ? Math.min(...starts) : 0; + if (starts.length > 0) return Math.min(...starts); + // 纯文本回合没有工具调用：用流里最早一条的出现时刻兜底，否则整轮用量行不显示。 + const streamStarts = (turnStreamByTurn.get(turnId) ?? []) + .map((item) => Number(item.at)) + .filter((at) => Number.isFinite(at) && at > 0); + return streamStarts.length > 0 ? Math.min(...streamStarts) : 0; }; const clockTimeWithSeconds = (timestamp: number) => { @@ -374,97 +524,6 @@ export function ProjectSupervisorView({ return `${pad(date.getHours())}:${pad(date.getMinutes())}:${pad(date.getSeconds())}`; }; - /** - * 把实时正文与工具块按"工具出现时正文长度"交替排列： - * 文本段 → 该位置的工具块 → 下一段文本……同一切分点的工具合并成一个块。 - * 没有切分点（历史回合）时退化为"正文在前、工具块在后"，与老项目兼容。 - */ - const renderLiveStream = (liveText: string, calls: typeof liveToolCalls) => { - const offsets = directToolTextOffsets[liveToolCallTurnId] ?? {}; - /** - * 把切分点吸附到句子边界：工具事件与文本增量是交替到达的，直接按"当时的长度"切 - * 会把句子切碎（历史上出现过"我来做一次轻｜量｜的"这种）。这里向后找最近的 - * 句末标点或换行，同一句里的多个工具因此落到同一个切点、自动合并成一块。 - */ - const snapToBoundary = (source: string, index: number) => { - const start = Math.max(0, Math.min(index, source.length)); - const limit = Math.min(source.length, start + 200); - for (let cursor = start; cursor < limit; cursor += 1) { - const char = source[cursor]; - if (char === '\n') { - return cursor + 1; - } - if ('。！？；!?;'.includes(char)) { - return cursor + 1; - } - if ( - char === '.' && - (source[cursor + 1] === ' ' || source[cursor + 1] === undefined) - ) { - return cursor + 1; - } - } - return start; - }; - const groups = new Map(); - for (const call of calls) { - const offset = offsets[call.id]; - const key = snapToBoundary( - liveText, - Number.isFinite(offset) ? Number(offset) : liveText.length, - ); - const bucket = groups.get(key) ?? []; - bucket.push(call); - groups.set(key, bucket); - } - if (groups.size === 0) { - return null; - } - const positions = [...groups.keys()].sort((left, right) => left - right); - const parts: ReactNode[] = []; - let cursor = 0; - for (const position of positions) { - const segment = liveText.slice(cursor, Math.max(cursor, position)); - if (segment.length > 0) { - parts.push( -
- -
, - ); - } - const bucket = groups.get(position) ?? []; - parts.push( - , - ); - cursor = Math.max(cursor, position); - } - const tail = liveText.slice(cursor); - if (tail.length > 0) { - parts.push( -
- -
, - ); - } - return parts; - }; - /** 整轮会话的结束时间与耗时（进行中时用 tick 驱的 now，所以秒数会实时跳动）。 */ const renderTurnUsage = (turnId: string) => { if (!turnId) { @@ -604,6 +663,14 @@ export function ProjectSupervisorView({ const anchoredTurnId = message.messageId ? directCodexTurnIdFromAssistantMessageId(message.messageId) : null; + const userTurnId = message.messageId + ? directCodexTurnIdFromUserMessageId(message.messageId) + : null; + const streamTurnIdForMessage = userTurnId ?? anchoredTurnId; + const turnStream = streamTurnIdForMessage + ? (turnStreamByTurn.get(streamTurnIdForMessage) ?? []) + : []; + const streamCovered = turnStream.length > 0; const nextTurnId = index + 1 < visibleMessages.length && visibleMessages[index + 1]?.messageId @@ -614,23 +681,10 @@ export function ProjectSupervisorView({ // 这条消息是该回合的最后一条时，在它后面给出整轮会话的结束时间与耗时。 const isTurnEnd = Boolean(anchoredTurnId) && nextTurnId !== anchoredTurnId; + const isActiveTurn = + Boolean(anchoredTurnId) && anchoredTurnId === activeTurnId; return ( -
- - {showDesignReasoning && message.reasoningText ? ( -
- 思考过程 - 
{message.reasoningText}
-
- ) : null} -
{anchoredToolCalls.length > 0 ? ( ) : null} - {isTurnEnd && anchoredTurnId && anchoredTurnId !== activeTurnId - ? renderTurnUsage(anchoredTurnId) - : null} + {/* 有回合流的回合：assistant 消息这一格交给流自己渲染——文本段 → 工具块 → + 文本段 → 最终回复，按条目的 `seq` 顺序出现，不再把工具抽到消息前面。 */} + {streamCovered ? ( + + ) : ( +
+ + {showDesignReasoning && message.reasoningText ? ( +
+ 思考过程 + 
{message.reasoningText}
+
+ ) : null} +
+ )} + {/* 有流的回合把用量行跟在最后一条**用户**消息之后（结束于 xxx，总耗时 xxx）； + 没有流的回合保持原样：跟在回合最后一条消息之后。 */} + {streamCovered + ? streamTurnIdForMessage + ? renderTurnUsage(streamTurnIdForMessage) + : null + : isTurnEnd && !isActiveTurn && anchoredTurnId + ? renderTurnUsage(anchoredTurnId) + : null} ); })} + {liveToolCalls.length > 0 ? ( + + ) : null} + {/* 正在跑的回合还没落盘用户消息时，流条目直接接在消息列表末尾； + 用户消息一旦落盘，同一个回合就由上面那条用户消息负责渲染（不再重复）。 */} + {directCodex && liveStreamTurnId && liveStreamItems.length > 0 ? ( + <> + + {renderTurnUsage(liveStreamTurnId)} + + ) : null} {liveToolCallTurnId && liveToolCallTurnId !== activeTurnId && + !turnStreamByTurn.has(liveToolCallTurnId) && !visibleMessages.some( (message) => message.messageId && @@ -666,11 +776,9 @@ export function ProjectSupervisorView({ ) : null} {/* 直连回合的流式正文：恢复为对话区里的普通 assistant 消息（不再放进状态卡片）， - 这样"边生成边显示"和"状态卡片只放状态"两件事同时成立。 */} - {directCodex && transientReply && liveToolCalls.length > 0 - ? renderLiveStream(transientReply, liveToolCalls) - : null} - {directCodex && transientReply && liveToolCalls.length === 0 ? ( + 这样"边生成边显示"和"状态卡片只放状态"两件事同时成立。 + 已有回合流的回合由流自己渲染文本段，这里不再重复一份。 */} + {directCodex && transientReply && !liveStreamTurnId ? (
) : null} - {/* 有正文时，工具块的顺序由 renderLiveStream 决定，这里不再重复渲染 */} - {liveToolCalls.length > 0 && !(directCodex && transientReply) ? ( - - ) : null} {showDesignReasoning && designReasoningEntries .filter((entry) => !entry.messageId)