From 76de8139de06d9d95f2f2a8d540cdac1ffdfa468 Mon Sep 17 00:00:00 2001 From: xiaoxixi Date: Sun, 19 Jul 2026 13:51:14 +0800 Subject: [PATCH] perf: calibrate turn history incrementally --- AGENTS.md | 2 +- docs/ARCHITECTURE.md | 4 +- src/bus/message.rs | 25 +++++++ src/bus/mod.rs | 5 +- src/channels/base.rs | 12 +++- src/channels/cli_chat.rs | 111 +++++++++++++++++++++-------- src/client/mod.rs | 12 +++- src/client/tui/app.rs | 122 +++++++++++++++++++++++++++++++- src/delivery/service.rs | 17 +++++ src/protocol.rs | 85 ++++++++++++++++++++++ src/session/persistence.rs | 20 +++++- src/session/session.rs | 55 ++++++++++++-- webui/src/pages/ChatPage.svelte | 27 ++++++- 13 files changed, 448 insertions(+), 49 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 7b5cd7a..2e48d29 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -89,7 +89,7 @@ Scheduler → SessionManager scheduled execution → AgentLoop → Scheduler del - **AgentLoop** is stateless across turns; it receives prepared history, calls LLM providers, executes tools, and returns one result - **WebUI management APIs** only expose allowlisted config/profile/log/storage operations; keep response limits, secret redaction, atomic config writes, and profile path allowlists intact - **WebUI slash completion** must consume the existing `get_slash_commands` WebSocket response; do not duplicate the backend command list in frontend source -- **WebUI chat rendering** sanitizes Markdown before inserting HTML; session history must preserve structured tool-call metadata so calls and results remain independently collapsible +- **WebUI chat rendering** sanitizes Markdown before inserting HTML; durable `turn_committed` deltas calibrate normal terminal Turns without a full history reload, and history must preserve structured tool-call metadata so calls and results remain independently collapsible - **WebUI/TUI file transfer** streams bytes over authenticated HTTP and sends only short-lived upload IDs/attachment metadata over WebSocket; messages persist local media paths without guaranteeing later availability, and client responses must never expose those paths - **WebUI authentication** protects every management API and `/ws`; only static pairing assets, public health/status, and pairing submission may bypass device auth. Pair-code issuance requires both a real loopback peer and the filesystem-held admin token; never put bearer tokens in URLs or logs - **Providers** are pure HTTP clients; no bus/session/channel awareness diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 322afd2..79a1711 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -199,7 +199,7 @@ SessionManager 负责组装会话上下文：系统提示、Skills、召回的 K - 5 秒 busy timeout。 - schema version 迁移。 -持久化范围包括 sessions、messages、memories、task plans/items、scheduled jobs、job runs 和 background tasks。消息保存可展示 `reasoning_content`、Provider 私有回放状态、`turn_id`、iteration 和 completion status；私有 Provider 状态不进入 WebSocket/Channel，且只允许回放给同一 Provider。Provider 诊断和失败审计只记录模型、消息数、工具数等请求摘要，不保存正文、reasoning 或签名 payload。修改 schema 时应： +持久化范围包括 sessions、messages、memories、task plans/items、scheduled jobs、job runs 和 background tasks。消息保存可展示 `reasoning_content`、Provider 私有回放状态、`turn_id`、iteration 和 completion status；私有 Provider 状态不进入 WebSocket/Channel，且只允许回放给同一 Provider。成功持久化一个 Turn 后，交互 Channel 收到只包含公开字段的 `CommittedTurnDelta`，其中 `history_revision` 是本批次最高 durable sequence；客户端按 revision 幂等合并，正常完成不重新加载整段历史，断线重连和失败/取消仍使用 `SessionHistory` 校准。Provider 诊断和失败审计只记录模型、消息数、工具数等请求摘要，不保存正文、reasoning 或签名 payload。修改 schema 时应： 1. 更新集中式 schema/迁移逻辑。 2. 保留已有数据库的升级路径。 @@ -236,7 +236,7 @@ Turn delivery 使用 `spawn_graceful`。全局取消发生时先停止读取快 ### WebUI 与管理 API -Gateway 在 `/` 提供随二进制编译的 HTML/CSS/JavaScript，不依赖外部 CDN 或前端运行服务。前端源码位于 `webui/`，由 Svelte 5 + Vite 构建，Bits UI 提供无样式的可访问组件原语。`build.rs` 监听前端源码、锁文件和构建配置，增量地将生产资源生成到 Cargo `OUT_DIR`，Rust 再从该目录编译嵌入；前端产物不进入仓库，最终用户使用发布二进制时不需要 Node.js。浏览器聊天继续使用 `/ws` 和 `cli_chat` 渠道，因此复用现有 dialog scope、每会话串行 worker、历史持久化和出站 lane；会话历史帧保留工具调用 ID、名称、参数和工具结果角色，WebUI 在本轮完成后刷新历史并将其渲染为默认折叠的工具卡片；聊天页的 Todo 侧栏默认隐藏，按 session 保存快照和未读状态，并通过结构化 `session_plan`/`plan_updated` 帧刷新，计划变化不会写入聊天历史；斜杠命令补全通过 `get_slash_commands` 获取 Gateway 的实时命令与别名清单，不在前端重复定义；WebUI 不直接调用 Provider 或 SessionManager。 +Gateway 在 `/` 提供随二进制编译的 HTML/CSS/JavaScript，不依赖外部 CDN 或前端运行服务。前端源码位于 `webui/`，由 Svelte 5 + Vite 构建，Bits UI 提供无样式的可访问组件原语。`build.rs` 监听前端源码、锁文件和构建配置，增量地将生产资源生成到 Cargo `OUT_DIR`，Rust 再从该目录编译嵌入；前端产物不进入仓库，最终用户使用发布二进制时不需要 Node.js。浏览器聊天继续使用 `/ws` 和 `cli_chat` 渠道，因此复用现有 dialog scope、每会话串行 worker、历史持久化和出站 lane；会话历史帧保留工具调用 ID、名称、参数和工具结果角色，WebUI 在正常完成时合并 `turn_committed` 增量并将工具信息渲染为默认折叠的工具卡片；聊天页的 Todo 侧栏默认隐藏，按 session 保存快照和未读状态，并通过结构化 `session_plan`/`plan_updated` 帧刷新，计划变化不会写入聊天历史；斜杠命令补全通过 `get_slash_commands` 获取 Gateway 的实时命令与别名清单，不在前端重复定义；WebUI 不直接调用 Provider 或 SessionManager。 WebUI/TUI 文件字节通过受鉴权的 HTTP 接口流式传输，WebSocket 只携带短期 `upload_id` 和结构化附件描述。`UploadRegistry` 在内存中按 `cli_chat` chat scope 校验并消费待发送上传；消息继续以 `media_refs` 保存 Gateway 本地路径，不建立永久附件资产。下载接口必须通过 client、session、message 和附件序号反查路径，不能接受客户端路径。历史附件路径失效属于正常状态，不得影响历史文本读取。待发送但未进入消息的上传由 `TaskSupervisor` 所有的限时清理任务回收。Agent 上下文会为所有媒体注入内部路径清单，客户端响应不得暴露该路径。 diff --git a/src/bus/message.rs b/src/bus/message.rs index 4085d71..9a92569 100644 --- a/src/bus/message.rs +++ b/src/bus/message.rs @@ -374,6 +374,31 @@ pub struct ChannelContext { pub private: HashMap, } +/// Public, durable projection of a newly committed conversation message. +/// Provider replay state and source identities are deliberately excluded. +#[derive(Debug, Clone)] +pub struct CommittedMessage { + pub id: String, + pub seq: i64, + pub role: String, + pub content: String, + pub reasoning_content: Option, + pub completion_status: CompletionStatus, + pub media_refs: Vec, + pub created_at: i64, + pub tool_call_id: Option, + pub tool_name: Option, + pub tool_calls: Option>, +} + +#[derive(Debug, Clone)] +pub struct CommittedTurnDelta { + pub session_id: String, + /// Highest durable message sequence included in this commit. + pub history_revision: i64, + pub messages: Vec, +} + #[derive(Debug, Clone)] pub struct InboundMessage { pub channel: String, diff --git a/src/bus/mod.rs b/src/bus/mod.rs index 4050e8f..0a4b1d2 100644 --- a/src/bus/mod.rs +++ b/src/bus/mod.rs @@ -3,8 +3,9 @@ pub mod message; pub use dispatcher::OutboundDispatcher; pub use message::{ - ChannelContext, ChatMessage, CompletionStatus, ContentBlock, ControlMessage, InboundMessage, - MediaItem, MediaRef, MessageSource, OutboundMessage, ProviderReasoningState, SourceKind, + ChannelContext, ChatMessage, CommittedMessage, CommittedTurnDelta, CompletionStatus, + ContentBlock, ControlMessage, InboundMessage, MediaItem, MediaRef, MessageSource, + OutboundMessage, ProviderReasoningState, SourceKind, }; use std::sync::Arc; diff --git a/src/channels/base.rs b/src/channels/base.rs index aa460c6..29519c5 100644 --- a/src/channels/base.rs +++ b/src/channels/base.rs @@ -3,7 +3,7 @@ use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; -use crate::bus::{BusError, InboundMessage, MessageBus, OutboundMessage}; +use crate::bus::{BusError, CommittedTurnDelta, InboundMessage, MessageBus, OutboundMessage}; use crate::delivery::PresentationPolicy; use crate::session::TurnSnapshot; @@ -90,6 +90,16 @@ pub trait Channel: Send + Sync + 'static { ))) } + /// Deliver a durable history delta after a Turn commit. Channels without + /// local history views intentionally ignore this event. + async fn commit_turn( + &self, + _target: &TurnTarget, + _delta: CommittedTurnDelta, + ) -> Result<(), ChannelError> { + Ok(()) + } + /// Send a message to the channel (called by OutboundDispatcher) async fn send(&self, msg: OutboundMessage) -> Result<(), ChannelError>; diff --git a/src/channels/cli_chat.rs b/src/channels/cli_chat.rs index a1f625b..2005094 100644 --- a/src/channels/cli_chat.rs +++ b/src/channels/cli_chat.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use std::time::Duration; use tokio::sync::{Mutex, mpsc}; -use crate::bus::{ControlMessage, InboundMessage, MessageBus, OutboundMessage}; +use crate::bus::{CommittedTurnDelta, ControlMessage, InboundMessage, MessageBus, OutboundMessage}; use crate::gateway::uploads::UploadRegistry; use crate::protocol::{ HistoryMessage, MessageAttachment, SlashCommandInfo, WsInbound, WsOutbound, parse_inbound, @@ -433,36 +433,7 @@ impl CliChatChannel { || message.tool_calls.is_some() || message.role == "tool" }) - .map(|message| { - let attachments = message - .media_refs - .as_deref() - .and_then(|refs| { - serde_json::from_str::>(refs).ok() - }) - .unwrap_or_default() - .iter() - .enumerate() - .map(|(index, media_ref)| { - MessageAttachment::from_media_ref(index, media_ref) - }) - .collect(); - HistoryMessage { - id: message.id, - seq: message.seq, - role: message.role, - content: message.content, - reasoning_content: message.reasoning_content, - completion_status: message.completion_status, - created_at: message.created_at, - tool_call_id: message.tool_call_id, - tool_name: message.tool_name, - tool_calls: message - .tool_calls - .and_then(|calls| serde_json::from_str(&calls).ok()), - attachments, - } - }) + .map(HistoryMessage::from_message_meta) .collect(); let _ = client .sender @@ -892,6 +863,29 @@ impl Channel for CliChatChannel { })) } + async fn commit_turn( + &self, + target: &TurnTarget, + delta: CommittedTurnDelta, + ) -> Result<(), ChannelError> { + let client = self.clients.lock().await.get(&target.chat_id).cloned(); + let Some(client) = client else { + return Ok(()); + }; + let frame = WsOutbound::TurnCommitted { + session_id: delta.session_id, + history_revision: delta.history_revision, + messages: delta + .messages + .into_iter() + .map(HistoryMessage::from) + .collect(), + }; + client.sender.send(frame).await.map_err(|_| { + ChannelError::ConnectionError("CLI client disconnected during turn commit".to_string()) + }) + } + async fn send(&self, msg: OutboundMessage) -> Result<(), ChannelError> { let client = self.clients.lock().await.get(&msg.chat_id).cloned(); let Some(client) = client else { @@ -1143,4 +1137,59 @@ mod tests { other => panic!("unexpected outbound: {other:?}"), } } + + #[tokio::test] + async fn committed_turn_is_projected_as_incremental_history_frame() { + let channel = CliChatChannel::new(); + let (sender, mut receiver) = mpsc::channel(1); + let client = Arc::new(Client { + sender, + chat_id: "client".into(), + current_session_id: Mutex::new(None), + }); + channel.clients.lock().await.insert("client".into(), client); + let target = TurnTarget { + channel: "cli_chat".into(), + chat_id: "client".into(), + session_id: "cli_chat:client:dialog".into(), + reply_to: None, + metadata: HashMap::new(), + }; + + channel + .commit_turn( + &target, + CommittedTurnDelta { + session_id: target.session_id.clone(), + history_revision: 4, + messages: vec![crate::bus::CommittedMessage { + id: "message".into(), + seq: 4, + role: "assistant".into(), + content: "done".into(), + reasoning_content: None, + completion_status: crate::bus::CompletionStatus::Completed, + media_refs: Vec::new(), + created_at: 1, + tool_call_id: None, + tool_name: None, + tool_calls: None, + }], + }, + ) + .await + .unwrap(); + + match receiver.recv().await.unwrap() { + WsOutbound::TurnCommitted { + history_revision, + messages, + .. + } => { + assert_eq!(history_revision, 4); + assert_eq!(messages[0].id, "message"); + } + other => panic!("unexpected outbound: {other:?}"), + } + } } diff --git a/src/client/mod.rs b/src/client/mod.rs index 50d3dcf..59feeba 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -255,6 +255,7 @@ async fn handle_ws_message(app: &mut App, outbound: WsOutbound) { match outbound { WsOutbound::TurnUpdated { snapshot } => { let terminal = snapshot.status != crate::session::TurnStatus::Running; + let completed = snapshot.status == crate::session::TurnStatus::Completed; let session_id = snapshot.session_id.clone(); if terminal { app.pending_responses = app.pending_responses.saturating_sub(1); @@ -263,7 +264,9 @@ async fn handle_ws_message(app: &mut App, outbound: WsOutbound) { if terminal { app.status_message = None; if app.current_session_id.as_deref() == Some(&session_id) { - request_history(app, session_id).await; + if !completed { + request_history(app, session_id).await; + } } else { app.status_message = Some("另一个会话已完成响应".to_string()); } @@ -275,6 +278,13 @@ async fn handle_ws_message(app: &mut App, outbound: WsOutbound) { app.status_message = Some("另一个会话已完成响应".to_string()); } } + WsOutbound::TurnCommitted { + session_id, + history_revision, + messages, + } => { + app.apply_turn_commit(&session_id, history_revision, messages); + } WsOutbound::AssistantResponse { id, content, diff --git a/src/client/tui/app.rs b/src/client/tui/app.rs index 642b4da..26c0c28 100644 --- a/src/client/tui/app.rs +++ b/src/client/tui/app.rs @@ -66,6 +66,7 @@ pub struct App { pub selected_session: usize, pub show_archived: bool, pub messages: VecDeque, + pub history_revision: i64, pub active_turn: Option, pub input: String, /// UTF-8 byte offset. It is always maintained at a character boundary. @@ -99,6 +100,7 @@ impl App { selected_session: 0, show_archived: false, messages: VecDeque::new(), + history_revision: 0, active_turn: None, input: String::new(), input_cursor_pos: 0, @@ -151,6 +153,11 @@ impl App { if self.current_session_id.as_deref() != Some(session_id) { return; } + let history_revision = messages + .iter() + .map(|message| message.seq) + .max() + .unwrap_or(0); let calibrates_terminal = self.active_turn.as_ref().is_some_and(|turn| { turn.status != TurnStatus::Running && messages.iter().any(|message| message.id == turn.message_id) @@ -178,12 +185,65 @@ impl App { self.messages.pop_front(); } self.chat_scroll_from_bottom = 0; + self.history_revision = history_revision; self.status_message = None; if calibrates_terminal { self.active_turn = None; } } + pub fn apply_turn_commit( + &mut self, + session_id: &str, + history_revision: i64, + messages: Vec, + ) -> bool { + if self.current_session_id.as_deref() != Some(session_id) + || history_revision <= self.history_revision + { + return false; + } + let calibrates_terminal = self.active_turn.as_ref().is_some_and(|turn| { + turn.status != TurnStatus::Running + && messages.iter().any(|message| message.id == turn.message_id) + }); + for message in messages { + let role = match message.role.as_str() { + "user" => MessageRole::User, + "assistant" => MessageRole::Assistant, + "system" | "tool" => MessageRole::System, + _ => continue, + }; + let projected = ChatMessage { + id: message.id.clone(), + role, + content: message.content, + reasoning_content: message.reasoning_content, + completion_status: message.completion_status, + attachments: message.attachments, + }; + if let Some(existing) = self + .messages + .iter_mut() + .find(|existing| existing.id == message.id) + { + *existing = projected; + } else { + self.messages.push_back(projected); + } + } + while self.messages.len() > MAX_MESSAGES { + self.messages.pop_front(); + } + self.history_revision = history_revision; + self.chat_scroll_from_bottom = 0; + self.status_message = None; + if calibrates_terminal { + self.active_turn = None; + } + true + } + pub fn set_sessions(&mut self, sessions: Vec) { self.sessions = sessions; if let Some(current) = &self.current_session_id @@ -202,6 +262,7 @@ impl App { self.current_session_id = session_id; self.messages.clear(); self.active_turn = None; + self.history_revision = 0; self.pending_uploads.clear(); self.chat_scroll_from_bottom = 0; } @@ -225,7 +286,12 @@ impl App { { return false; } - self.active_turn = Some(snapshot); + let already_committed = snapshot.status != TurnStatus::Running + && self + .messages + .iter() + .any(|message| message.id == snapshot.message_id); + self.active_turn = (!already_committed).then_some(snapshot); self.chat_scroll_from_bottom = 0; true } @@ -491,6 +557,60 @@ mod tests { assert!(app.active_turn.is_none()); } + #[test] + fn committed_delta_calibrates_terminal_without_reloading_history() { + let mut app = App::new(); + app.set_current_session(Some("current".into())); + app.apply_turn_snapshot(turn(3, TurnStatus::Completed)); + + assert!(app.apply_turn_commit( + "current", + 2, + vec![HistoryMessage { + id: "message".into(), + seq: 2, + role: "assistant".into(), + content: "done".into(), + reasoning_content: None, + completion_status: crate::bus::CompletionStatus::Completed, + created_at: 1, + tool_call_id: None, + tool_name: None, + tool_calls: None, + attachments: Vec::new(), + }], + )); + assert!(app.active_turn.is_none()); + assert_eq!(app.messages.back().unwrap().content, "done"); + assert!(!app.apply_turn_commit("current", 2, Vec::new())); + } + + #[test] + fn terminal_snapshot_calibrates_when_commit_arrived_first() { + let mut app = App::new(); + app.set_current_session(Some("current".into())); + assert!(app.apply_turn_commit( + "current", + 2, + vec![HistoryMessage { + id: "message".into(), + seq: 2, + role: "assistant".into(), + content: "done".into(), + reasoning_content: None, + completion_status: crate::bus::CompletionStatus::Completed, + created_at: 1, + tool_call_id: None, + tool_name: None, + tool_calls: None, + attachments: Vec::new(), + }], + )); + + assert!(app.apply_turn_snapshot(turn(3, TurnStatus::Completed))); + assert!(app.active_turn.is_none()); + } + #[test] fn failed_turn_without_a_durable_message_remains_visible_after_history_refresh() { let mut app = App::new(); diff --git a/src/delivery/service.rs b/src/delivery/service.rs index 22eb8a5..d710ce0 100644 --- a/src/delivery/service.rs +++ b/src/delivery/service.rs @@ -2,6 +2,7 @@ use std::sync::Arc; use tokio::sync::{oneshot, watch}; +use crate::bus::CommittedTurnDelta; use crate::channels::{ChannelManager, TurnTarget}; use crate::delivery::{DeliveryCoordinator, DeliveryError}; use crate::session::TurnSnapshot; @@ -78,6 +79,22 @@ impl TurnDeliveryService { )?; Ok(TurnDeliveryHandle { completion }) } + + pub async fn commit( + &self, + target: &TurnTarget, + delta: CommittedTurnDelta, + ) -> Result<(), DeliveryError> { + let channel = self + .channels + .get_channel(&target.channel) + .await + .ok_or_else(|| DeliveryError::ChannelNotFound(target.channel.clone()))?; + channel + .commit_turn(target, delta) + .await + .map_err(DeliveryError::FinalFailed) + } } #[cfg(test)] diff --git a/src/protocol.rs b/src/protocol.rs index f1221cd..a408eac 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -77,6 +77,59 @@ pub struct HistoryMessage { pub attachments: Vec, } +impl From for HistoryMessage { + fn from(message: crate::bus::CommittedMessage) -> Self { + let attachments = message + .media_refs + .iter() + .enumerate() + .map(|(index, media_ref)| MessageAttachment::from_media_ref(index, media_ref)) + .collect(); + Self { + id: message.id, + seq: message.seq, + role: message.role, + content: message.content, + reasoning_content: message.reasoning_content, + completion_status: message.completion_status, + created_at: message.created_at, + tool_call_id: message.tool_call_id, + tool_name: message.tool_name, + tool_calls: message.tool_calls, + attachments, + } + } +} + +impl HistoryMessage { + pub fn from_message_meta(message: crate::storage::message::MessageMeta) -> Self { + let attachments = message + .media_refs + .as_deref() + .and_then(|refs| serde_json::from_str::>(refs).ok()) + .unwrap_or_default() + .iter() + .enumerate() + .map(|(index, media_ref)| MessageAttachment::from_media_ref(index, media_ref)) + .collect(); + Self { + id: message.id, + seq: message.seq, + role: message.role, + content: message.content, + reasoning_content: message.reasoning_content, + completion_status: message.completion_status, + created_at: message.created_at, + tool_call_id: message.tool_call_id, + tool_name: message.tool_name, + tool_calls: message + .tool_calls + .and_then(|calls| serde_json::from_str(&calls).ok()), + attachments, + } + } +} + #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "type")] pub enum WsInbound { @@ -148,6 +201,12 @@ pub enum WsOutbound { TurnUpdated { snapshot: crate::session::TurnSnapshot, }, + #[serde(rename = "turn_committed")] + TurnCommitted { + session_id: String, + history_revision: i64, + messages: Vec, + }, #[serde(rename = "assistant_response")] AssistantResponse { id: String, @@ -259,6 +318,32 @@ mod tests { assert_eq!(value["snapshot"]["status"], "running"); } + #[test] + fn turn_committed_serializes_revision_and_durable_delta() { + let frame = WsOutbound::TurnCommitted { + session_id: "session".to_string(), + history_revision: 7, + messages: vec![HistoryMessage { + id: "message".to_string(), + seq: 7, + role: "assistant".to_string(), + content: "done".to_string(), + reasoning_content: None, + completion_status: crate::bus::CompletionStatus::Completed, + created_at: 1, + tool_call_id: None, + tool_name: None, + tool_calls: None, + attachments: Vec::new(), + }], + }; + let value = serde_json::to_value(frame).unwrap(); + + assert_eq!(value["type"], "turn_committed"); + assert_eq!(value["history_revision"], 7); + assert_eq!(value["messages"][0]["id"], "message"); + } + #[test] fn history_defaults_new_reasoning_fields_for_old_frames() { let message: HistoryMessage = serde_json::from_value(serde_json::json!({ diff --git a/src/session/persistence.rs b/src/session/persistence.rs index 4de586f..f8723a1 100644 --- a/src/session/persistence.rs +++ b/src/session/persistence.rs @@ -44,26 +44,40 @@ pub(super) async fn append_persisted_messages( session: &Arc>, messages: Vec, ) -> Result<(), StorageError> { + append_persisted_messages_with_meta(session, messages) + .await + .map(|_| ()) +} + +pub(super) async fn append_persisted_messages_with_meta( + session: &Arc>, + messages: Vec, +) -> Result, StorageError> { if messages.is_empty() { - return Ok(()); + return Ok(Vec::new()); } let persistence_lock = { session.lock().await.persistence_lock.clone() }; let _persistence_guard = persistence_lock.lock().await; let message_ids: Vec<_> = messages.iter().map(|message| message.id.clone()).collect(); - let snapshots = { + let snapshots: Vec> = { let mut guard = session.lock().await; messages .into_iter() .map(|message| guard.add_message_in_memory(message, true)) .collect() }; + let committed = snapshots + .iter() + .flatten() + .map(|(_, _, message, _)| message.clone()) + .collect(); if let Err(error) = persist_added_messages(snapshots).await { session.lock().await.rollback_message_suffix(&message_ids); return Err(error); } - Ok(()) + Ok(committed) } /// Publish `Completed` only after the supplied durable write succeeds. diff --git a/src/session/session.rs b/src/session/session.rs index 19f66e1..c861b8a 100644 --- a/src/session/session.rs +++ b/src/session/session.rs @@ -3,7 +3,9 @@ use std::sync::Arc; use tokio::sync::{Mutex, mpsc, oneshot}; -use super::persistence::{append_persisted_messages, finalize_turn_after_persistence}; +use super::persistence::{ + append_persisted_messages, append_persisted_messages_with_meta, finalize_turn_after_persistence, +}; use super::turn::{TurnBlock, TurnController, TurnSnapshot}; use super::turn_input::prepare_turn_input; use crate::bus::{ @@ -36,6 +38,39 @@ fn outbound_turn_metadata( metadata } +fn committed_turn_delta( + session_id: &str, + messages: Vec, +) -> crate::bus::CommittedTurnDelta { + let history_revision = messages.last().map_or(0, |message| message.seq); + let messages = messages + .into_iter() + .map(|message| crate::bus::CommittedMessage { + id: message.id, + seq: message.seq, + role: message.role, + content: message.content, + reasoning_content: message.reasoning_content, + completion_status: message.completion_status, + media_refs: message + .media_refs + .and_then(|refs| serde_json::from_str(&refs).ok()) + .unwrap_or_default(), + created_at: message.created_at, + tool_call_id: message.tool_call_id, + tool_name: message.tool_name, + tool_calls: message + .tool_calls + .and_then(|calls| serde_json::from_str(&calls).ok()), + }) + .collect(); + crate::bus::CommittedTurnDelta { + session_id: session_id.to_string(), + history_revision, + messages, + } +} + tokio::task_local! { pub(super) static CURRENT_SOURCE_SESSION: Option; } @@ -2729,6 +2764,8 @@ fn spawn_agent_worker( let task_metadata2 = task_metadata.clone(); let task_reply_to2 = task_reply_to.clone(); let title_supervisor = worker_supervisor.clone(); + let commit_delivery = turn_delivery.clone(); + let commit_target = turn_target.clone(); let turn_lifecycle = &turn_controller; let process_future = async move { let response_session_id = unified_str2.clone(); @@ -2901,15 +2938,18 @@ fn spawn_agent_worker( let response = match finalize_turn_after_persistence( turn_lifecycle, usage, - append_persisted_messages(&session2, result.emitted_messages), + append_persisted_messages_with_meta( + &session2, + result.emitted_messages, + ), ) .await { - Ok(()) => { + Ok(committed_messages) => { let mut guard = session2.lock().await; let sent_count = guard.messages.len(); guard.compressor.set_last_api_info(sent_count, total_tokens); - Some(response_content) + Some((response_content, committed_messages)) } Err(e) => { tracing::error!(error = %e, "Failed to atomically persist agent turn"); @@ -2917,7 +2957,7 @@ fn spawn_agent_worker( } }; - let Some(response) = response else { + let Some((response, committed_messages)) = response else { let err_outbound = OutboundMessage { channel: chan2, chat_id: cid2, @@ -2937,6 +2977,11 @@ fn spawn_agent_worker( return; }; + let delta = committed_turn_delta(&response_session_id, committed_messages); + if let Err(error) = commit_delivery.commit(&commit_target, delta).await { + tracing::warn!(error = %error, "Failed to publish committed turn delta"); + } + schedule_title_generation( session2.clone(), title_supervisor, diff --git a/webui/src/pages/ChatPage.svelte b/webui/src/pages/ChatPage.svelte index 6f49474..77550f6 100644 --- a/webui/src/pages/ChatPage.svelte +++ b/webui/src/pages/ChatPage.svelte @@ -19,6 +19,7 @@ let commandMenuDismissed = $state(false); let thinking = $state(false); let activeTurn = $state(null); + let historyRevision = $state(0); let pendingUploads = $state([]); let fileInput; let plansBySession = $state({}); @@ -77,7 +78,7 @@ function handleFrame(frame) { switch (frame.type) { - case "session_established": currentId = frame.session_id; activeTurn = null; break; + case "session_established": currentId = frame.session_id; activeTurn = null; historyRevision = 0; break; case "session_list": sessions = frame.sessions || []; if (frame.current_session_id) currentId = frame.current_session_id; @@ -87,17 +88,20 @@ currentId = frame.session_id; messages = []; activeTurn = null; + historyRevision = 0; send({ type: "list_sessions", include_archived: false }); break; case "session_loaded": currentId = frame.session_id; activeTurn = null; + historyRevision = 0; send({ type: "get_session_history", session_id: currentId, limit: 1000 }); send({ type: "get_session_plan", session_id: currentId }); break; case "session_history": if (frame.session_id === currentId) { messages = frame.messages || []; + historyRevision = Math.max(0, ...messages.map((message) => message.seq || 0)); if (activeTurn?.status !== "running" && messages.some((message) => message.id === activeTurn?.message_id)) activeTurn = null; scrollToBottom(); } @@ -144,13 +148,31 @@ if (activeTurn?.id === next.id && activeTurn.revision >= next.revision) break; activeTurn = next; thinking = next.status === "running"; + if (next.status !== "running" && messages.some((message) => message.id === next.message_id)) { + activeTurn = null; + } scrollToBottom(); if (next.status !== "running") { - send({ type: "get_session_history", session_id: currentId, limit: 1000 }); + if (next.status !== "completed") { + send({ type: "get_session_history", session_id: currentId, limit: 1000 }); + } send({ type: "list_sessions", include_archived: false }); } break; } + case "turn_committed": { + if (frame.session_id !== currentId || frame.history_revision <= historyRevision) break; + const byId = new Map(messages.map((message) => [message.id, message])); + for (const message of frame.messages || []) byId.set(message.id, message); + messages = [...byId.values()]; + historyRevision = frame.history_revision; + if (activeTurn?.status !== "running" + && (frame.messages || []).some((message) => message.id === activeTurn?.message_id)) { + activeTurn = null; + } + scrollToBottom(); + break; + } case "system_notification": if (!frame.session_id || frame.session_id === currentId) appendMessage("assistant", frame.content); break; @@ -172,6 +194,7 @@ function loadSession(id) { if (!id) return; currentId = id; + historyRevision = 0; clearPendingUploads(); messages = []; activeTurn = null;