按 item 分段实现文本与工具交替的对话流
- Rust：item 事件保留 itemId，文本段按 (turnId,itemId) 幂等落盘；工具项按 callId 落位置项 - 新增 .agent/conversations/turn-stream.jsonl（schema agc-turn-stream.v1）：seq 为位置真相，首次分配后不回退；updatedAt 单调、同刻取更长文本；脱敏/上限/坏行跳过沿用工具卡片口径 - 发射器新增 emit_with_stream_items（不改 emit 现有签名），事件新增可选 streamItems - 新增读取命令 read_direct_turn_stream（权限 conversation.read） - 前端：打开项目读取历史流，实时按事件合并（同 id 幂等）；按流渲染 text 块与工具块，相邻工具合并、夹文本则分块；回合末渲染结束时间与总耗时 - 回退：该回合没有流项（老项目/缺文件/读取失败）时沿用原渲染，不白屏不丢内容 - 删除长度切点/标点吸附/1 秒时间窗三套启发式 - cargo build 通过、tsc 无新增错误、编码检查通过；未跑测试
This commit is contained in:
@@ -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::*;
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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<String> {
|
||||
self.text_by_item_id.get(item_id).cloned()
|
||||
}
|
||||
|
||||
fn take_partial_items(&mut self) -> impl Iterator<Item = Value> + '_ {
|
||||
std::mem::take(&mut self.text_by_item_id)
|
||||
.into_iter()
|
||||
|
||||
@@ -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<String> {
|
||||
///
|
||||
/// `pub(crate)`：回合流（`direct_turn_stream`）落最终回复前复用同一套可见性投影。
|
||||
pub(crate) fn project_direct_codex_visible_text(value: &str) -> Option<String> {
|
||||
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::<Vec<_>>();
|
||||
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<String, u64>,
|
||||
next_seq: u64,
|
||||
pending_text: Option<DirectTurnStreamPendingText>,
|
||||
}
|
||||
|
||||
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<DirectTurnStreamItem> {
|
||||
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<DirectTurnStreamItem> {
|
||||
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::<Vec<_>>();
|
||||
emitter.emit_with_stream_items(
|
||||
"finalizing",
|
||||
Some("response-finalization"),
|
||||
Some(visible_reply.clone()),
|
||||
None,
|
||||
finalized,
|
||||
);
|
||||
}
|
||||
if direct_codex_output_fingerprint(root) != previous_output_fingerprint {
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -0,0 +1,401 @@
|
||||
//! GameAgent 对话「回合流」的采集、持久化与回读。
|
||||
//!
|
||||
//! 顺序真相放在一处：`<projectRoot>/.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:<turnId>:<itemId>`、工具 `tool:<turnId>:<callId>`。
|
||||
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<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub(crate) call_id: Option<String>,
|
||||
/// 首次出现的写入序号：**顺序真相**，同刻按它排序。
|
||||
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::<String>();
|
||||
truncated.push('…');
|
||||
truncated
|
||||
}
|
||||
|
||||
fn record_line(item: &DirectTurnStreamItem) -> Result<String, String> {
|
||||
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<DirectTurnStreamItem> {
|
||||
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<DirectTurnStreamItem> {
|
||||
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<DirectTurnStreamItem>) -> Vec<DirectTurnStreamItem> {
|
||||
let mut by_id: BTreeMap<String, DirectTurnStreamItem> = 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::<Vec<_>>();
|
||||
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<T>(
|
||||
root: &Path,
|
||||
mutate: impl FnOnce(&mut Vec<DirectTurnStreamItem>) -> T,
|
||||
) -> Result<T, String> {
|
||||
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<Vec<DirectTurnStreamItem>, 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<Option<DirectTurnStreamItem>, 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<Option<DirectTurnStreamItem>, 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)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -62,6 +62,48 @@ impl DirectGameCreatorTurnUpdateEmitter {
|
||||
accumulated_text: Option<String>,
|
||||
tool_calls: Option<Vec<crate::DirectToolCall>>,
|
||||
reasoning_text: Option<String>,
|
||||
) {
|
||||
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<String>,
|
||||
tool_calls: Option<Vec<crate::DirectToolCall>>,
|
||||
stream_items: Vec<crate::DirectTurnStreamItem>,
|
||||
) {
|
||||
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<String>,
|
||||
tool_calls: Option<Vec<crate::DirectToolCall>>,
|
||||
reasoning_text: Option<String>,
|
||||
stream_items: Vec<crate::DirectTurnStreamItem>,
|
||||
) {
|
||||
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,
|
||||
},
|
||||
);
|
||||
|
||||
@@ -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<Vec<DirectTurnStreamItem>, 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,
|
||||
|
||||
@@ -1017,6 +1017,10 @@ struct GameCreatorDirectTurnUpdateEvent {
|
||||
/// 本回合当前累计的思考过程（流式整段替换）；拿不到时字段缺席。
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
reasoning_text: Option<String>,
|
||||
/// 本回合**顺序真相**里本次发生变化的那几条（文本段 / 工具位置标记）。
|
||||
/// `skip_serializing_if`：字段缺席时前端拿到 `undefined`，行为与改造前一致。
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
stream_items: Option<Vec<crate::DirectTurnStreamItem>>,
|
||||
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,
|
||||
|
||||
@@ -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<string, TurnStreamItem>();
|
||||
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
|
||||
@@ -1106,6 +1169,30 @@ export function App({
|
||||
setDirectToolCalls(normalized);
|
||||
}
|
||||
|
||||
// 回合流（文本段 + 工具按**出现顺序**交替）：实时增量与回读历史共用一份状态，
|
||||
// 渲染顺序只由条目的 `seq` 决定，界面不再按文本长度 / 标点 / 时间窗猜切点。
|
||||
const [directTurnStream, setDirectTurnStream] = useState<TurnStreamItem[]>([]);
|
||||
const directTurnStreamRef = useRef<TurnStreamItem[]>([]);
|
||||
|
||||
/** 归并一批回合流条目（实时事件里的 `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 (
|
||||
@@ -2199,6 +2286,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
|
||||
@@ -3880,6 +3971,14 @@ export function App({
|
||||
{ projectPath: nextProjectPath },
|
||||
).catch(() => []);
|
||||
replaceDirectToolCalls(persistedToolCalls);
|
||||
// 回合流（顺序真相）走独立历史文件（`turn-stream.jsonl`）。与工具调用同一处：必须在
|
||||
// 读完项目对话之后、任何提前 return 之前回读。缺命令（老客户端）/ 缺文件 / 读取失败
|
||||
// 都只是"这个回合没有流"，界面回退到原来的渲染，不能因此把整个打开流程判失败。
|
||||
const persistedTurnStream = await invoke<TurnStreamItem[]>(
|
||||
'read_direct_turn_stream',
|
||||
{ projectPath: nextProjectPath },
|
||||
).catch(() => []);
|
||||
replaceTurnStreamItems(persistedTurnStream);
|
||||
// 重进会话时 Rust 侧可能仍登记着上一条 Direct 回合。不接管的话界面既不显示
|
||||
// 过程卡也不给终止入口，用户再发消息只会被守卫拒绝（"已有另一条回合正在运行"）。
|
||||
await restoreRunningDirectCodexTurn(nextProjectPath);
|
||||
@@ -12149,6 +12248,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(
|
||||
@@ -12635,6 +12741,7 @@ export function App({
|
||||
pendingCommand={directCodexProductRuntime ? pendingCommand : null}
|
||||
projectPath={localProject?.projectPath ?? projectPath}
|
||||
toolCalls={visibleToolCalls}
|
||||
turnStreamItems={visibleTurnStreamItems}
|
||||
activeTurnId={
|
||||
directCodexProductRuntime
|
||||
? (activeDirectCodexTurnRef.current?.turnId ?? null)
|
||||
|
||||
@@ -1163,9 +1163,47 @@ export interface GameCreatorDirectTurnUpdateEvent {
|
||||
* 本回合当前累计的思考过程（流式，整段替换）；拿不到时字段缺席。
|
||||
*/
|
||||
reasoningText?: string | null;
|
||||
/**
|
||||
* 「文本段 + 工具」的**顺序真相**里本次发生变化的那几条。
|
||||
*
|
||||
* 顺序由 `seq`（条目首次出现时钉死）决定，与落盘 `turn-stream.jsonl` 完全同一份数据，
|
||||
* 前端不再自己猜切点。可选：老版本事件没有这个字段。
|
||||
*/
|
||||
streamItems?: TurnStreamItem[] | null;
|
||||
updatedAt: number;
|
||||
}
|
||||
|
||||
/** 回合流里的一个 `text` 段；`text` 是该段当前累计全文（会随 delta 增长）。 */
|
||||
export interface TurnStreamTextItem extends TurnStreamItemBase {
|
||||
kind: 'text';
|
||||
text: string;
|
||||
}
|
||||
|
||||
/** 回合流里的一个 `tool` 位置标记；工具正文在 `tool-calls.jsonl`（按 `callId` 关联）。 */
|
||||
export interface TurnStreamToolItem extends TurnStreamItemBase {
|
||||
kind: 'tool';
|
||||
callId: string;
|
||||
}
|
||||
|
||||
interface TurnStreamItemBase {
|
||||
schemaVersion: string;
|
||||
/** 幂等身份：文本段 `text:<turnId>:<itemId>`、工具 `tool:<turnId>:<callId>`。 */
|
||||
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 活跃回合。
|
||||
*
|
||||
|
||||
+217
-20
@@ -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:<turnId>: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<string, TurnStreamItem[]>();
|
||||
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<string, GameCreatorDirectToolCall>();
|
||||
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' ? (
|
||||
<div
|
||||
key={run.key}
|
||||
className={className ? `message message--assistant ${className}` : 'message message--assistant'}
|
||||
>
|
||||
<ChatMarkdownMessage
|
||||
role="assistant"
|
||||
text={run.text}
|
||||
streaming={active && index === runs.length - 1}
|
||||
/>
|
||||
</div>
|
||||
) : (
|
||||
<ToolCallGroup
|
||||
key={run.key}
|
||||
calls={run.callIds
|
||||
.map((callId) => callsById.get(callId))
|
||||
.filter((call): call is GameCreatorDirectToolCall => Boolean(call))}
|
||||
userSentAt={userSentAt}
|
||||
active={active}
|
||||
className="message-tool-call"
|
||||
/>
|
||||
),
|
||||
)}
|
||||
</>
|
||||
);
|
||||
}
|
||||
|
||||
type RuntimePanelProps = ComponentProps<typeof ProjectSupervisorRuntimePanel>;
|
||||
|
||||
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;
|
||||
@@ -222,6 +353,7 @@ export function ProjectSupervisorView({
|
||||
pendingCommand,
|
||||
projectPath,
|
||||
toolCalls = [],
|
||||
turnStreamItems = [],
|
||||
activeTurnId = null,
|
||||
initialSupervisorMessage = '',
|
||||
showProfessionalCollaboration = true,
|
||||
@@ -300,12 +432,28 @@ export function ProjectSupervisorView({
|
||||
// 工具调用折叠块按回合分组，插在**同一回合 assistant 消息之前**（Codex 是「工具在上、答复在下」）。
|
||||
// 历史回合锚到自己那条 `direct-codex:<turnId>:assistant`，不回落到别的回合；
|
||||
// 正在跑的回合还没有 assistant 消息落盘，先落在消息流末尾，等那条消息落盘后回到它之前。
|
||||
// **只有没有回合流的回合才走这条路**：有流的回合按流自己的顺序渲染。
|
||||
const streamTurnId = activeTurnId?.trim() ?? '';
|
||||
// 回合流（`turn-stream.jsonl`）按回合分组：有流的回合按「文本段 → 工具块 → 文本段」顺序渲染，
|
||||
// 没有流的回合（老项目 / 缺文件 / 读取失败）留在老路径上，绝不白屏。
|
||||
const turnStreamByTurn = groupTurnStreamItems(
|
||||
directCodex ? turnStreamItems : [],
|
||||
);
|
||||
const toolCallsByAnchor = new Map<string, GameCreatorDirectToolCall[]>();
|
||||
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,
|
||||
@@ -362,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) => {
|
||||
@@ -510,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
|
||||
@@ -520,6 +681,8 @@ export function ProjectSupervisorView({
|
||||
// 这条消息是该回合的最后一条时，在它后面给出整轮会话的结束时间与耗时。
|
||||
const isTurnEnd =
|
||||
Boolean(anchoredTurnId) && nextTurnId !== anchoredTurnId;
|
||||
const isActiveTurn =
|
||||
Boolean(anchoredTurnId) && anchoredTurnId === activeTurnId;
|
||||
return (
|
||||
<Fragment key={message.messageId ?? `${message.role}-${index}`}>
|
||||
{anchoredToolCalls.length > 0 ? (
|
||||
@@ -531,24 +694,43 @@ export function ProjectSupervisorView({
|
||||
className="message-tool-call"
|
||||
/>
|
||||
) : null}
|
||||
<div className={`message message--${message.role}`}>
|
||||
<ChatMarkdownMessage
|
||||
role={message.role}
|
||||
text={projectSupervisorChatMessageText(message)}
|
||||
{/* 有回合流的回合：assistant 消息这一格交给流自己渲染——文本段 → 工具块 →
|
||||
文本段 → 最终回复，按条目的 `seq` 顺序出现，不再把工具抽到消息前面。 */}
|
||||
{streamCovered ? (
|
||||
<TurnStreamSequence
|
||||
items={turnStream}
|
||||
toolCalls={toolCalls}
|
||||
active={Boolean(activeTurnId) && streamTurnIdForMessage === activeTurnId}
|
||||
userSentAt={userMessageUpdatedAtForTurn(
|
||||
streamTurnIdForMessage ?? '',
|
||||
)}
|
||||
/>
|
||||
{showDesignReasoning && message.reasoningText ? (
|
||||
<details
|
||||
className="design-agent-reasoning"
|
||||
aria-label="策划 Agent 思考过程"
|
||||
>
|
||||
<summary>思考过程</summary>
|
||||
<pre>{message.reasoningText}</pre>
|
||||
</details>
|
||||
) : null}
|
||||
</div>
|
||||
{isTurnEnd && anchoredTurnId && anchoredTurnId !== activeTurnId
|
||||
? renderTurnUsage(anchoredTurnId)
|
||||
: null}
|
||||
) : (
|
||||
<div className={`message message--${message.role}`}>
|
||||
<ChatMarkdownMessage
|
||||
role={message.role}
|
||||
text={projectSupervisorChatMessageText(message)}
|
||||
/>
|
||||
{showDesignReasoning && message.reasoningText ? (
|
||||
<details
|
||||
className="design-agent-reasoning"
|
||||
aria-label="策划 Agent 思考过程"
|
||||
>
|
||||
<summary>思考过程</summary>
|
||||
<pre>{message.reasoningText}</pre>
|
||||
</details>
|
||||
) : null}
|
||||
</div>
|
||||
)}
|
||||
{/* 有流的回合把用量行跟在最后一条**用户**消息之后（结束于 xxx，总耗时 xxx）；
|
||||
没有流的回合保持原样：跟在回合最后一条消息之后。 */}
|
||||
{streamCovered
|
||||
? streamTurnIdForMessage
|
||||
? renderTurnUsage(streamTurnIdForMessage)
|
||||
: null
|
||||
: isTurnEnd && !isActiveTurn && anchoredTurnId
|
||||
? renderTurnUsage(anchoredTurnId)
|
||||
: null}
|
||||
</Fragment>
|
||||
);
|
||||
})}
|
||||
@@ -560,8 +742,22 @@ export function ProjectSupervisorView({
|
||||
className="message-tool-call"
|
||||
/>
|
||||
) : null}
|
||||
{/* 正在跑的回合还没落盘用户消息时，流条目直接接在消息列表末尾；
|
||||
用户消息一旦落盘，同一个回合就由上面那条用户消息负责渲染（不再重复）。 */}
|
||||
{directCodex && liveStreamTurnId && liveStreamItems.length > 0 ? (
|
||||
<>
|
||||
<TurnStreamSequence
|
||||
items={liveStreamItems}
|
||||
toolCalls={toolCalls}
|
||||
active
|
||||
userSentAt={userMessageUpdatedAtForTurn(liveStreamTurnId)}
|
||||
/>
|
||||
{renderTurnUsage(liveStreamTurnId)}
|
||||
</>
|
||||
) : null}
|
||||
{liveToolCallTurnId &&
|
||||
liveToolCallTurnId !== activeTurnId &&
|
||||
!turnStreamByTurn.has(liveToolCallTurnId) &&
|
||||
!visibleMessages.some(
|
||||
(message) =>
|
||||
message.messageId &&
|
||||
@@ -580,8 +776,9 @@ export function ProjectSupervisorView({
|
||||
</details>
|
||||
) : null}
|
||||
{/* 直连回合的流式正文：恢复为对话区里的普通 assistant 消息（不再放进状态卡片），
|
||||
这样"边生成边显示"和"状态卡片只放状态"两件事同时成立。 */}
|
||||
{directCodex && transientReply ? (
|
||||
这样"边生成边显示"和"状态卡片只放状态"两件事同时成立。
|
||||
已有回合流的回合由流自己渲染文本段，这里不再重复一份。 */}
|
||||
{directCodex && transientReply && !liveStreamTurnId ? (
|
||||
<div
|
||||
className="message message--assistant"
|
||||
aria-label="陶泥儿实时回复"
|
||||
|
||||
Reference in New Issue
Block a user