From 740d0f4f332d0c972222cb6277e17d8e167b4527 Mon Sep 17 00:00:00 2001 From: xiaoxixi Date: Fri, 17 Jul 2026 14:37:10 +0800 Subject: [PATCH] feat: persist streaming turn metadata --- src/bus/message.rs | 105 ++++++++++++ src/bus/mod.rs | 4 +- src/session/session.rs | 19 +++ src/storage/message.rs | 6 + src/storage/mod.rs | 344 +++++++++++++++++++++++++------------- src/tools/chat_manager.rs | 12 ++ 6 files changed, 374 insertions(+), 116 deletions(-) diff --git a/src/bus/message.rs b/src/bus/message.rs index 56ae5ef..21c5659 100644 --- a/src/bus/message.rs +++ b/src/bus/message.rs @@ -3,6 +3,52 @@ use std::collections::HashMap; use crate::providers::ToolCall; +/// Provider-private state required to faithfully replay an assistant message. +/// +/// This is durable conversation data, but it is never presentation data. UI and +/// channel projections must not serialize it to end users. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct ProviderReasoningState { + pub provider: String, + pub payload: serde_json::Value, +} + +impl ProviderReasoningState { + /// Decode persisted provider state without making conversation history + /// unreadable when an old or damaged payload is encountered. + pub fn from_json_lossy(value: &str) -> Option { + serde_json::from_str(value).ok() + } +} + +/// Describes whether a persisted message represents a complete model result. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum CompletionStatus { + #[default] + Completed, + Cancelled, + Interrupted, +} + +impl CompletionStatus { + pub fn as_str(self) -> &'static str { + match self { + Self::Completed => "completed", + Self::Cancelled => "cancelled", + Self::Interrupted => "interrupted", + } + } + + pub fn from_storage(value: &str) -> Self { + match value { + "cancelled" => Self::Cancelled, + "interrupted" => Self::Interrupted, + _ => Self::Completed, + } + } +} + // ============================================================================ // ContentBlock - Multimodal content representation (OpenAI-style) // ============================================================================ @@ -85,6 +131,15 @@ pub struct ChatMessage { pub role: String, pub content: String, pub reasoning_content: Option, + /// Opaque state used only when replaying history to the same provider. + #[serde(skip_serializing_if = "Option::is_none")] + pub provider_state: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub turn_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub iteration: Option, + #[serde(default)] + pub completion_status: CompletionStatus, pub media_refs: Vec, pub timestamp: i64, #[serde(skip_serializing_if = "Option::is_none")] @@ -124,6 +179,10 @@ impl ChatMessage { role: "user".to_string(), content: content.into(), reasoning_content: None, + provider_state: None, + turn_id: None, + iteration: None, + completion_status: CompletionStatus::Completed, media_refs: Vec::new(), timestamp: current_timestamp(), tool_call_id: None, @@ -139,6 +198,10 @@ impl ChatMessage { role: "user".to_string(), content: content.into(), reasoning_content: None, + provider_state: None, + turn_id: None, + iteration: None, + completion_status: CompletionStatus::Completed, media_refs, timestamp: current_timestamp(), tool_call_id: None, @@ -154,6 +217,10 @@ impl ChatMessage { role: "assistant".to_string(), content: content.into(), reasoning_content: None, + provider_state: None, + turn_id: None, + iteration: None, + completion_status: CompletionStatus::Completed, media_refs: Vec::new(), timestamp: current_timestamp(), tool_call_id: None, @@ -172,6 +239,10 @@ impl ChatMessage { role: "assistant".to_string(), content: content.into(), reasoning_content: None, + provider_state: None, + turn_id: None, + iteration: None, + completion_status: CompletionStatus::Completed, media_refs: Vec::new(), timestamp: current_timestamp(), tool_call_id: None, @@ -187,6 +258,10 @@ impl ChatMessage { role: "assistant".to_string(), content: content.into(), reasoning_content: None, + provider_state: None, + turn_id: None, + iteration: None, + completion_status: CompletionStatus::Completed, media_refs: Vec::new(), timestamp: current_timestamp(), tool_call_id: None, @@ -202,6 +277,10 @@ impl ChatMessage { role: "system".to_string(), content: content.into(), reasoning_content: None, + provider_state: None, + turn_id: None, + iteration: None, + completion_status: CompletionStatus::Completed, media_refs: Vec::new(), timestamp: current_timestamp(), tool_call_id: None, @@ -230,6 +309,10 @@ impl ChatMessage { role: "tool".to_string(), content: content.into(), reasoning_content: None, + provider_state: None, + turn_id: None, + iteration: None, + completion_status: CompletionStatus::Completed, media_refs, timestamp: current_timestamp(), tool_call_id: Some(tool_call_id.into()), @@ -245,6 +328,10 @@ impl ChatMessage { role: "user".to_string(), content: content.into(), reasoning_content: None, + provider_state: None, + turn_id: None, + iteration: None, + completion_status: CompletionStatus::Completed, media_refs: Vec::new(), timestamp: current_timestamp(), tool_call_id: None, @@ -255,6 +342,24 @@ impl ChatMessage { } } +#[cfg(test)] +mod conversation_message_tests { + use super::*; + + #[test] + fn damaged_provider_state_is_ignored() { + assert!(ProviderReasoningState::from_json_lossy("not-json").is_none()); + } + + #[test] + fn unknown_completion_status_is_backward_compatible() { + assert_eq!( + CompletionStatus::from_storage("future-status"), + CompletionStatus::Completed + ); + } +} + // ============================================================================ // InboundMessage - Message from Channel to Bus (user input) // ============================================================================ diff --git a/src/bus/mod.rs b/src/bus/mod.rs index bde9eb4..f54fd1d 100644 --- a/src/bus/mod.rs +++ b/src/bus/mod.rs @@ -3,8 +3,8 @@ pub mod message; pub use dispatcher::OutboundDispatcher; pub use message::{ - ChatMessage, ContentBlock, ControlMessage, InboundMessage, MediaItem, MediaRef, MessageSource, - OutboundMessage, SourceKind, + ChatMessage, CompletionStatus, ContentBlock, ControlMessage, InboundMessage, MediaItem, + MediaRef, MessageSource, OutboundMessage, ProviderReasoningState, SourceKind, }; use std::sync::Arc; diff --git a/src/session/session.rs b/src/session/session.rs index 5ab3d5b..ee17eca 100644 --- a/src/session/session.rs +++ b/src/session/session.rs @@ -258,6 +258,12 @@ impl Session { role: m.role, content: m.content, reasoning_content: m.reasoning_content, + provider_state: m.provider_state.and_then(|state| { + crate::bus::ProviderReasoningState::from_json_lossy(&state) + }), + turn_id: m.turn_id, + iteration: m.iteration.and_then(|value| u32::try_from(value).ok()), + completion_status: m.completion_status, media_refs: m .media_refs .map(|refs| serde_json::from_str(&refs).unwrap_or_default()) @@ -293,6 +299,12 @@ impl Session { role: m.role, content: m.content, reasoning_content: m.reasoning_content, + provider_state: m.provider_state.and_then(|state| { + crate::bus::ProviderReasoningState::from_json_lossy(&state) + }), + turn_id: m.turn_id, + iteration: m.iteration.and_then(|value| u32::try_from(value).ok()), + completion_status: m.completion_status, media_refs: m .media_refs .map(|refs| serde_json::from_str(&refs).unwrap_or_default()) @@ -386,6 +398,13 @@ impl Session { role: message.role.clone(), content: message.content.clone(), reasoning_content: message.reasoning_content.clone(), + provider_state: message + .provider_state + .as_ref() + .and_then(|state| serde_json::to_string(state).ok()), + turn_id: message.turn_id.clone(), + iteration: message.iteration.map(i64::from), + completion_status: message.completion_status, media_refs: if message.media_refs.is_empty() { None } else { diff --git a/src/storage/message.rs b/src/storage/message.rs index 6ae0c29..11b3123 100644 --- a/src/storage/message.rs +++ b/src/storage/message.rs @@ -1,5 +1,7 @@ use serde::{Deserialize, Serialize}; +use crate::bus::CompletionStatus; + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MessageMeta { pub id: String, @@ -8,6 +10,10 @@ pub struct MessageMeta { pub role: String, pub content: String, pub reasoning_content: Option, + pub provider_state: Option, + pub turn_id: Option, + pub iteration: Option, + pub completion_status: CompletionStatus, pub media_refs: Option, pub tool_call_id: Option, pub tool_name: Option, diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 1d3612d..c0331bb 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -9,15 +9,21 @@ pub use background_task::BackgroundTask; pub use error::StorageError; pub use scheduler::{DeliveryPolicy, JobKind, JobRun, ScheduledJob}; -use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous}; +use sqlx::sqlite::{ + SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteRow, SqliteSynchronous, +}; use sqlx::{Pool, Row, Sqlite}; use std::path::Path; use tokio::time::{Duration, sleep}; -const SCHEMA_VERSION: i64 = 3; +const SCHEMA_VERSION: i64 = 4; const INSERT_MESSAGE_SQL: &str = r#" - INSERT INTO messages (id, session_id, seq, role, content, reasoning_content, media_refs, tool_call_id, tool_name, tool_calls, source, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + INSERT INTO messages ( + id, session_id, seq, role, content, reasoning_content, provider_state, + turn_id, iteration, completion_status, media_refs, tool_call_id, + tool_name, tool_calls, source, created_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "#; fn insert_message_query<'a>( @@ -31,6 +37,10 @@ fn insert_message_query<'a>( .bind(&msg.role) .bind(&msg.content) .bind(&msg.reasoning_content) + .bind(&msg.provider_state) + .bind(&msg.turn_id) + .bind(msg.iteration) + .bind(msg.completion_status.as_str()) .bind(&msg.media_refs) .bind(&msg.tool_call_id) .bind(&msg.tool_name) @@ -39,6 +49,28 @@ fn insert_message_query<'a>( .bind(msg.created_at) } +fn message_meta_from_row(row: SqliteRow) -> crate::storage::message::MessageMeta { + let completion_status: String = row.get("completion_status"); + crate::storage::message::MessageMeta { + id: row.get("id"), + session_id: row.get("session_id"), + seq: row.get("seq"), + role: row.get("role"), + content: row.get("content"), + reasoning_content: row.get("reasoning_content"), + provider_state: row.get("provider_state"), + turn_id: row.get("turn_id"), + iteration: row.get("iteration"), + completion_status: crate::bus::CompletionStatus::from_storage(&completion_status), + media_refs: row.get("media_refs"), + tool_call_id: row.get("tool_call_id"), + tool_name: row.get("tool_name"), + tool_calls: row.get("tool_calls"), + source: row.get("source"), + created_at: row.get("created_at"), + } +} + pub struct Storage { pub(crate) pool: Pool, } @@ -111,6 +143,10 @@ impl Storage { tool_calls TEXT, source TEXT, reasoning_content TEXT, + provider_state TEXT, + turn_id TEXT, + iteration INTEGER, + completion_status TEXT NOT NULL DEFAULT 'completed', created_at INTEGER NOT NULL, FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE ) @@ -351,6 +387,14 @@ impl Storage { for (table, column, definition) in [ ("messages", "source", "source TEXT"), ("messages", "reasoning_content", "reasoning_content TEXT"), + ("messages", "provider_state", "provider_state TEXT"), + ("messages", "turn_id", "turn_id TEXT"), + ("messages", "iteration", "iteration INTEGER"), + ( + "messages", + "completion_status", + "completion_status TEXT NOT NULL DEFAULT 'completed'", + ), ("sessions", "archived_at", "archived_at INTEGER"), ( "sessions", @@ -829,7 +873,9 @@ impl Storage { ) -> Result, StorageError> { let rows = sqlx::query( r#" - SELECT id, session_id, seq, role, content, reasoning_content, media_refs, tool_call_id, tool_name, tool_calls, source, created_at + SELECT id, session_id, seq, role, content, reasoning_content, provider_state, + turn_id, iteration, completion_status, media_refs, tool_call_id, + tool_name, tool_calls, source, created_at FROM messages WHERE session_id = ? AND seq >= ? ORDER BY seq ASC @@ -840,23 +886,7 @@ impl Storage { .fetch_all(self.pool()) .await?; - Ok(rows - .into_iter() - .map(|row| crate::storage::message::MessageMeta { - id: row.get("id"), - session_id: row.get("session_id"), - seq: row.get("seq"), - role: row.get("role"), - content: row.get("content"), - reasoning_content: row.get("reasoning_content"), - media_refs: row.get("media_refs"), - tool_call_id: row.get("tool_call_id"), - tool_name: row.get("tool_name"), - tool_calls: row.get("tool_calls"), - source: row.get("source"), - created_at: row.get("created_at"), - }) - .collect()) + Ok(rows.into_iter().map(message_meta_from_row).collect()) } pub async fn get_message( @@ -866,8 +896,9 @@ impl Storage { ) -> Result, StorageError> { let row = sqlx::query( r#" - SELECT id, session_id, seq, role, content, reasoning_content, media_refs, - tool_call_id, tool_name, tool_calls, source, created_at + SELECT id, session_id, seq, role, content, reasoning_content, provider_state, + turn_id, iteration, completion_status, media_refs, tool_call_id, + tool_name, tool_calls, source, created_at FROM messages WHERE session_id = ? AND id = ? "#, @@ -877,20 +908,7 @@ impl Storage { .fetch_optional(self.pool()) .await?; - Ok(row.map(|row| crate::storage::message::MessageMeta { - id: row.get("id"), - session_id: row.get("session_id"), - seq: row.get("seq"), - role: row.get("role"), - content: row.get("content"), - reasoning_content: row.get("reasoning_content"), - media_refs: row.get("media_refs"), - tool_call_id: row.get("tool_call_id"), - tool_name: row.get("tool_name"), - tool_calls: row.get("tool_calls"), - source: row.get("source"), - created_at: row.get("created_at"), - })) + Ok(row.map(message_meta_from_row)) } pub async fn get_max_message_seq(&self, session_id: &str) -> Result { @@ -912,11 +930,13 @@ impl Storage { let limit = limit.clamp(1, 2_000); let rows = sqlx::query( r#" - SELECT id, session_id, seq, role, content, reasoning_content, media_refs, - tool_call_id, tool_name, tool_calls, source, created_at + SELECT id, session_id, seq, role, content, reasoning_content, provider_state, + turn_id, iteration, completion_status, media_refs, tool_call_id, + tool_name, tool_calls, source, created_at FROM ( - SELECT id, session_id, seq, role, content, reasoning_content, media_refs, - tool_call_id, tool_name, tool_calls, source, created_at + SELECT id, session_id, seq, role, content, reasoning_content, provider_state, + turn_id, iteration, completion_status, media_refs, tool_call_id, + tool_name, tool_calls, source, created_at FROM messages WHERE session_id = ? ORDER BY seq DESC @@ -930,23 +950,7 @@ impl Storage { .fetch_all(self.pool()) .await?; - Ok(rows - .into_iter() - .map(|row| crate::storage::message::MessageMeta { - id: row.get("id"), - session_id: row.get("session_id"), - seq: row.get("seq"), - role: row.get("role"), - content: row.get("content"), - reasoning_content: row.get("reasoning_content"), - media_refs: row.get("media_refs"), - tool_call_id: row.get("tool_call_id"), - tool_name: row.get("tool_name"), - tool_calls: row.get("tool_calls"), - source: row.get("source"), - created_at: row.get("created_at"), - }) - .collect()) + Ok(rows.into_iter().map(message_meta_from_row).collect()) } pub async fn load_messages_after_timestamp( @@ -956,7 +960,9 @@ impl Storage { ) -> Result, StorageError> { let rows = sqlx::query( r#" - SELECT id, session_id, seq, role, content, reasoning_content, media_refs, tool_call_id, tool_name, tool_calls, source, created_at + SELECT id, session_id, seq, role, content, reasoning_content, provider_state, + turn_id, iteration, completion_status, media_refs, tool_call_id, + tool_name, tool_calls, source, created_at FROM messages WHERE session_id = ? AND created_at > ? ORDER BY seq ASC @@ -967,23 +973,7 @@ impl Storage { .fetch_all(self.pool()) .await?; - Ok(rows - .into_iter() - .map(|row| crate::storage::message::MessageMeta { - id: row.get("id"), - session_id: row.get("session_id"), - seq: row.get("seq"), - role: row.get("role"), - content: row.get("content"), - reasoning_content: row.get("reasoning_content"), - media_refs: row.get("media_refs"), - tool_call_id: row.get("tool_call_id"), - tool_name: row.get("tool_name"), - tool_calls: row.get("tool_calls"), - source: row.get("source"), - created_at: row.get("created_at"), - }) - .collect()) + Ok(rows.into_iter().map(message_meta_from_row).collect()) } pub async fn query_sessions_range( @@ -1040,7 +1030,9 @@ impl Storage { ) -> Result, StorageError> { let rows = sqlx::query( r#" - SELECT id, session_id, seq, role, content, reasoning_content, media_refs, tool_call_id, tool_name, tool_calls, source, created_at + SELECT id, session_id, seq, role, content, reasoning_content, provider_state, + turn_id, iteration, completion_status, media_refs, tool_call_id, + tool_name, tool_calls, source, created_at FROM messages WHERE session_id = ? ORDER BY seq DESC @@ -1052,23 +1044,7 @@ impl Storage { .fetch_all(self.pool()) .await?; - let mut messages: Vec<_> = rows - .into_iter() - .map(|row| crate::storage::message::MessageMeta { - id: row.get("id"), - session_id: row.get("session_id"), - seq: row.get("seq"), - role: row.get("role"), - content: row.get("content"), - reasoning_content: row.get("reasoning_content"), - media_refs: row.get("media_refs"), - tool_call_id: row.get("tool_call_id"), - tool_name: row.get("tool_name"), - tool_calls: row.get("tool_calls"), - source: row.get("source"), - created_at: row.get("created_at"), - }) - .collect(); + let mut messages: Vec<_> = rows.into_iter().map(message_meta_from_row).collect(); messages.reverse(); Ok(messages) } @@ -1095,7 +1071,9 @@ impl Storage { ); let select_sql = format!( r#" - SELECT id, session_id, seq, role, content, reasoning_content, media_refs, tool_call_id, tool_name, tool_calls, source, created_at + SELECT id, session_id, seq, role, content, reasoning_content, provider_state, + turn_id, iteration, completion_status, media_refs, tool_call_id, + tool_name, tool_calls, source, created_at FROM messages WHERE session_id = ?{} ORDER BY seq ASC @@ -1127,23 +1105,7 @@ impl Storage { .fetch_all(self.pool()) .await?; - let messages: Vec<_> = rows - .into_iter() - .map(|row| crate::storage::message::MessageMeta { - id: row.get("id"), - session_id: row.get("session_id"), - seq: row.get("seq"), - role: row.get("role"), - content: row.get("content"), - reasoning_content: row.get("reasoning_content"), - media_refs: row.get("media_refs"), - tool_call_id: row.get("tool_call_id"), - tool_name: row.get("tool_name"), - tool_calls: row.get("tool_calls"), - source: row.get("source"), - created_at: row.get("created_at"), - }) - .collect(); + let messages: Vec<_> = rows.into_iter().map(message_meta_from_row).collect(); Ok((messages, total)) } @@ -1616,7 +1578,17 @@ mod tests { let storage = Storage::new(&db_path).await.unwrap(); for (table, expected) in [ - ("messages", vec!["source", "reasoning_content"]), + ( + "messages", + vec![ + "source", + "reasoning_content", + "provider_state", + "turn_id", + "iteration", + "completion_status", + ], + ), ( "sessions", vec![ @@ -1660,6 +1632,81 @@ mod tests { } } + #[tokio::test] + async fn v3_migration_preserves_existing_reasoning_and_defaults_completion() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("v3.db"); + let pool = SqlitePoolOptions::new() + .connect_with( + SqliteConnectOptions::new() + .filename(&db_path) + .create_if_missing(true), + ) + .await + .unwrap(); + sqlx::query( + r#" + CREATE TABLE sessions ( + id TEXT PRIMARY KEY, channel TEXT NOT NULL, chat_id TEXT NOT NULL, + dialog_id TEXT NOT NULL, title TEXT NOT NULL DEFAULT 'new', + created_at INTEGER NOT NULL, last_active_at INTEGER NOT NULL, + message_count INTEGER DEFAULT 0, routing_info TEXT, archived_at INTEGER, + deleted_at INTEGER, last_consolidated_at INTEGER, + last_compressed_message_at INTEGER, + UNIQUE(channel, chat_id, dialog_id) + ) + "#, + ) + .execute(&pool) + .await + .unwrap(); + sqlx::query( + r#" + CREATE TABLE messages ( + id TEXT PRIMARY KEY, session_id TEXT NOT NULL, seq INTEGER NOT NULL, + role TEXT NOT NULL, content TEXT NOT NULL, reasoning_content TEXT, + media_refs TEXT, tool_call_id TEXT, tool_name TEXT, tool_calls TEXT, + source TEXT, created_at INTEGER NOT NULL + ) + "#, + ) + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO sessions (id, channel, chat_id, dialog_id, created_at, last_active_at) VALUES ('cli:c:d', 'cli', 'c', 'd', 1, 1)", + ) + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO messages (id, session_id, seq, role, content, reasoning_content, created_at) VALUES ('m1', 'cli:c:d', 1, 'assistant', 'answer', 'existing reasoning', 1)", + ) + .execute(&pool) + .await + .unwrap(); + sqlx::query("PRAGMA user_version = 3") + .execute(&pool) + .await + .unwrap(); + drop(pool); + + let storage = Storage::new(&db_path).await.unwrap(); + let messages = storage.load_messages("cli:c:d", 0).await.unwrap(); + assert_eq!(messages.len(), 1); + assert_eq!( + messages[0].reasoning_content.as_deref(), + Some("existing reasoning") + ); + assert_eq!(messages[0].provider_state, None); + assert_eq!(messages[0].turn_id, None); + assert_eq!(messages[0].iteration, None); + assert_eq!( + messages[0].completion_status, + crate::bus::CompletionStatus::Completed + ); + } + #[tokio::test] async fn test_upsert_and_get_session() { let (storage, _dir) = create_test_storage().await; @@ -1784,6 +1831,10 @@ mod tests { role: "user".to_string(), content: "你好".to_string(), reasoning_content: None, + provider_state: None, + turn_id: None, + iteration: None, + completion_status: crate::bus::CompletionStatus::Completed, media_refs: None, tool_call_id: None, tool_name: None, @@ -1821,6 +1872,67 @@ mod tests { assert_eq!(recent[1].seq, 5); } + #[tokio::test] + async fn streaming_message_metadata_round_trips() { + let (storage, _dir) = create_test_storage().await; + let session_meta = crate::storage::session::SessionMeta { + id: "cli_chat:stream:dialog1".to_string(), + channel: "cli_chat".to_string(), + chat_id: "stream".to_string(), + dialog_id: "dialog1".to_string(), + title: "Stream metadata".to_string(), + created_at: 1000, + last_active_at: 1000, + message_count: 0, + routing_info: None, + archived_at: None, + deleted_at: None, + last_consolidated_at: None, + last_compressed_message_at: None, + }; + storage.upsert_session(&session_meta).await.unwrap(); + + let message = crate::storage::message::MessageMeta { + id: "assistant-1".to_string(), + session_id: session_meta.id.clone(), + seq: 1, + role: "assistant".to_string(), + content: "partial".to_string(), + reasoning_content: Some("visible reasoning".to_string()), + provider_state: Some( + serde_json::json!({"provider":"anthropic","payload":{"signature":"opaque"}}) + .to_string(), + ), + turn_id: Some("turn-1".to_string()), + iteration: Some(2), + completion_status: crate::bus::CompletionStatus::Interrupted, + media_refs: None, + tool_call_id: None, + tool_name: None, + tool_calls: None, + source: None, + created_at: 1001, + }; + storage + .append_message(&session_meta.id, &message) + .await + .unwrap(); + + let loaded = storage.load_messages(&session_meta.id, 0).await.unwrap(); + assert_eq!(loaded.len(), 1); + assert_eq!( + loaded[0].reasoning_content.as_deref(), + Some("visible reasoning") + ); + assert_eq!(loaded[0].provider_state, message.provider_state); + assert_eq!(loaded[0].turn_id.as_deref(), Some("turn-1")); + assert_eq!(loaded[0].iteration, Some(2)); + assert_eq!( + loaded[0].completion_status, + crate::bus::CompletionStatus::Interrupted + ); + } + #[tokio::test] async fn test_persist_message_batch_is_atomic() { let (storage, _dir) = create_test_storage().await; @@ -1848,6 +1960,10 @@ mod tests { role: "assistant".to_string(), content: "must roll back".to_string(), reasoning_content: None, + provider_state: None, + turn_id: None, + iteration: None, + completion_status: crate::bus::CompletionStatus::Completed, media_refs: None, tool_call_id: None, tool_name: None, diff --git a/src/tools/chat_manager.rs b/src/tools/chat_manager.rs index 06497d9..39d6e8e 100644 --- a/src/tools/chat_manager.rs +++ b/src/tools/chat_manager.rs @@ -358,6 +358,10 @@ mod tests { }, content: format!("消息内容 {}", i), reasoning_content: None, + provider_state: None, + turn_id: None, + iteration: None, + completion_status: crate::bus::CompletionStatus::Completed, media_refs: None, tool_call_id: None, tool_name: None, @@ -420,6 +424,10 @@ mod tests { }, content: format!("消息内容 {}", i), reasoning_content: None, + provider_state: None, + turn_id: None, + iteration: None, + completion_status: crate::bus::CompletionStatus::Completed, media_refs: None, tool_call_id: None, tool_name: None, @@ -476,6 +484,10 @@ mod tests { role: "user".to_string(), content: format!("消息内容 {}", i), reasoning_content: None, + provider_state: None, + turn_id: None, + iteration: None, + completion_status: crate::bus::CompletionStatus::Completed, media_refs: None, tool_call_id: None, tool_name: None,