perf: calibrate turn history incrementally

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

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 上下文会为所有媒体注入内部路径清单，客户端响应不得暴露该路径。
 

src/bus/message.rs

@@ -374,6 +374,31 @@ pub struct ChannelContext {
     pub private: HashMap<String, String>,
 }
 
+/// 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<String>,
+    pub completion_status: CompletionStatus,
+    pub media_refs: Vec<MediaRef>,
+    pub created_at: i64,
+    pub tool_call_id: Option<String>,
+    pub tool_name: Option<String>,
+    pub tool_calls: Option<Vec<ToolCall>>,
+}
+
+#[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<CommittedMessage>,
+}
+
 #[derive(Debug, Clone)]
 pub struct InboundMessage {
     pub channel: String,

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;

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>;
 

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::<Vec<crate::bus::MediaRef>>(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:?}"),
+        }
+    }
 }

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,

src/client/tui/app.rs

@@ -66,6 +66,7 @@ pub struct App {
     pub selected_session: usize,
     pub show_archived: bool,
     pub messages: VecDeque<ChatMessage>,
+    pub history_revision: i64,
     pub active_turn: Option<TurnSnapshot>,
     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<HistoryMessage>,
+    ) -> 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<SessionSummary>) {
         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();

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)]

src/protocol.rs

@@ -77,6 +77,59 @@ pub struct HistoryMessage {
     pub attachments: Vec<MessageAttachment>,
 }
 
+impl From<crate::bus::CommittedMessage> 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::<Vec<crate::bus::MediaRef>>(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<HistoryMessage>,
+    },
     #[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!({

src/session/persistence.rs

@@ -44,26 +44,40 @@ pub(super) async fn append_persisted_messages(
     session: &Arc<Mutex<Session>>,
     messages: Vec<ChatMessage>,
 ) -> Result<(), StorageError> {
+    append_persisted_messages_with_meta(session, messages)
+        .await
+        .map(|_| ())
+}
+
+pub(super) async fn append_persisted_messages_with_meta(
+    session: &Arc<Mutex<Session>>,
+    messages: Vec<ChatMessage>,
+) -> Result<Vec<crate::storage::message::MessageMeta>, 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<Option<MessagePersistSnapshot>> = {
         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.

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::storage::message::MessageMeta>,
+) -> 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<String>;
 }
@@ -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,

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;