From 95d3829e9279db698fb644775caeba0406e31fd9 Mon Sep 17 00:00:00 2001 From: Darkatse <39111369+Darkatse@users.noreply.github.com> Date: Wed, 15 Jul 2026 22:23:22 -0400 Subject: [PATCH] perf(chat): coalesce automatic backups after stable checkpoints #128 #131 --- src-tauri/crates/tauritavern/src/app.rs | 6 + .../composition/adapters/cache_reconciler.rs | 6 + .../src/app/composition/services/mod.rs | 12 + .../tauritavern/src/app/contract_tests/mod.rs | 9 +- .../tauritavern/src/app/host/shutdown.rs | 1 + src-tauri/crates/tauritavern/src/app/state.rs | 4 + .../presentation/commands/chat_commands.rs | 35 +- .../commands/group_chat_commands.rs | 3 +- .../src/presentation/commands/registry.rs | 2 + .../file_chat_repository/backup.rs | 64 +- .../file_chat_repository/backup_inventory.rs | 5 +- .../file_chat_repository/cache.rs | 30 - .../group_chat_repository_impl.rs | 22 +- .../repositories/file_chat_repository/mod.rs | 6 +- .../file_chat_repository/paths.rs | 24 +- .../file_chat_repository/payload.rs | 11 +- .../file_chat_repository/repository_impl.rs | 30 +- .../file_chat_repository/tests.rs | 88 +- .../file_chat_repository/windowed_hide.rs | 6 - .../file_chat_repository/windowed_patch.rs | 6 - .../crates/tt-application/src/dto/chat_dto.rs | 6 + .../src/dto/chat_history_dto.rs | 60 + .../crates/tt-application/src/dto/mod.rs | 1 + .../src/services/character_service.rs | 32 + .../src/services/chat_history_coordinator.rs | 1163 +++++++++++++++++ .../src/services/chat_service.rs | 82 +- .../src/services/group_chat_service.rs | 52 +- .../src/services/group_service.rs | 24 + .../crates/tt-application/src/services/mod.rs | 1 + .../src/repositories/chat_repository.rs | 10 + .../src/repositories/group_chat_repository.rs | 6 + src/script.js | 93 +- src/scripts/chat-payload-transport.js | 1 + src/scripts/group-chats.js | 16 +- src/scripts/tauri/chat/transport.js | 19 +- .../main/api/agent-chat-commit-bridge.js | 16 +- src/tauri/main/routes/chat-routes.js | 3 +- tests/active-chat-ref-contract.test.mjs | 26 + tests/generation-lifecycle-contract.test.mjs | 35 +- tests/itemized-prompts-contract.test.mjs | 2 +- 40 files changed, 1844 insertions(+), 174 deletions(-) create mode 100644 src-tauri/crates/tt-application/src/dto/chat_history_dto.rs create mode 100644 src-tauri/crates/tt-application/src/services/chat_history_coordinator.rs diff --git a/src-tauri/crates/tauritavern/src/app.rs b/src-tauri/crates/tauritavern/src/app.rs index 7dd1f047..33de46c0 100644 --- a/src-tauri/crates/tauritavern/src/app.rs +++ b/src-tauri/crates/tauritavern/src/app.rs @@ -84,6 +84,12 @@ pub fn spawn_initialization( .await; }); + let chat_history_coordinator = state.services.chat_history_coordinator.clone(); + let chat_history_cancel = state.lifecycle.chat_history_cancel.clone(); + tauri::async_runtime::spawn(async move { + chat_history_coordinator.run(chat_history_cancel).await; + }); + match app_handle.emit("app-ready", ()) { Ok(_) => tracing::debug!("Application is ready"), Err(error) => tracing::error!("Failed to emit app-ready event: {}", error), diff --git a/src-tauri/crates/tauritavern/src/app/composition/adapters/cache_reconciler.rs b/src-tauri/crates/tauritavern/src/app/composition/adapters/cache_reconciler.rs index 56a77253..e01367df 100644 --- a/src-tauri/crates/tauritavern/src/app/composition/adapters/cache_reconciler.rs +++ b/src-tauri/crates/tauritavern/src/app/composition/adapters/cache_reconciler.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use async_trait::async_trait; use tt_application::services::character_service::CharacterService; +use tt_application::services::chat_history_coordinator::ChatHistoryCoordinator; use tt_application::services::chat_service::ChatService; use tt_application::services::group_chat_service::GroupChatService; use tt_application::services::group_service::GroupService; @@ -18,6 +19,7 @@ pub(in crate::app::composition) fn data_change_reconciler( group_service: Arc, secret_service: Arc, settings_service: Arc, + chat_history_coordinator: Arc, ) -> Arc { Arc::new(ServiceCacheReconciler { character_service, @@ -26,6 +28,7 @@ pub(in crate::app::composition) fn data_change_reconciler( group_service, secret_service, settings_service, + chat_history_coordinator, }) } @@ -36,6 +39,7 @@ struct ServiceCacheReconciler { group_service: Arc, secret_service: Arc, settings_service: Arc, + chat_history_coordinator: Arc, } #[async_trait] @@ -46,6 +50,8 @@ impl DataChangeReconciler for ServiceCacheReconciler { "Refreshing runtime caches after external data change" ); + self.chat_history_coordinator.invalidate_all_pending().await; + self.character_service.clear_cache().await?; self.chat_service.clear_cache().await?; self.group_chat_service.clear_cache().await?; diff --git a/src-tauri/crates/tauritavern/src/app/composition/services/mod.rs b/src-tauri/crates/tauritavern/src/app/composition/services/mod.rs index 872c3e5b..7d81843d 100644 --- a/src-tauri/crates/tauritavern/src/app/composition/services/mod.rs +++ b/src-tauri/crates/tauritavern/src/app/composition/services/mod.rs @@ -16,6 +16,7 @@ use tt_application::services::avatar_service::AvatarService; use tt_application::services::background_service::BackgroundService; use tt_application::services::character_service::CharacterService; use tt_application::services::chat_completion_service::ChatCompletionService; +use tt_application::services::chat_history_coordinator::ChatHistoryCoordinator; use tt_application::services::chat_service::ChatService; use tt_application::services::content_service::ContentService; use tt_application::services::extension_service::ExtensionService; @@ -141,24 +142,33 @@ pub(super) async fn build( crate::product::VERSION, )); + let chat_history_coordinator = Arc::new(ChatHistoryCoordinator::new( + repositories.chat_repository.clone(), + repositories.group_chat_repository.clone(), + )); + let group_service = Arc::new(GroupService::new( repositories.group_repository.clone(), agent_services.agent_workspace_lifecycle_service.clone(), + chat_history_coordinator.clone(), )); let character_service = Arc::new(CharacterService::new( repositories.character_repository.clone(), repositories.chat_repository.clone(), repositories.world_info_repository.clone(), agent_services.agent_workspace_lifecycle_service.clone(), + chat_history_coordinator.clone(), )); let chat_service = Arc::new(ChatService::new( repositories.chat_repository.clone(), repositories.character_repository.clone(), agent_services.agent_workspace_lifecycle_service.clone(), + chat_history_coordinator.clone(), )); let group_chat_service = Arc::new(GroupChatService::new( repositories.group_chat_repository.clone(), agent_services.agent_workspace_lifecycle_service, + chat_history_coordinator.clone(), )); let secret_service = Arc::new(SecretService::new( repositories.secret_repository.clone(), @@ -180,6 +190,7 @@ pub(super) async fn build( group_service.clone(), secret_service.clone(), settings_service.clone(), + chat_history_coordinator.clone(), ); let data_archive_service = archive::build(app_handle, data_change_reconciler.clone()); let sync_services = sync::build( @@ -194,6 +205,7 @@ pub(super) async fn build( character_service, chat_service, group_chat_service, + chat_history_coordinator, user_service, settings_service, user_directory_service, diff --git a/src-tauri/crates/tauritavern/src/app/contract_tests/mod.rs b/src-tauri/crates/tauritavern/src/app/contract_tests/mod.rs index 24a9420c..97ed1173 100644 --- a/src-tauri/crates/tauritavern/src/app/contract_tests/mod.rs +++ b/src-tauri/crates/tauritavern/src/app/contract_tests/mod.rs @@ -48,6 +48,7 @@ use tt_application::services::agent_workspace_lifecycle_service::{ AgentRunActivity, AgentWorkspaceLifecycleService, }; use tt_application::services::character_service::CharacterService; +use tt_application::services::chat_history_coordinator::ChatHistoryCoordinator; use tt_application::services::llm_connection_service::LlmConnectionService; use tt_application::services::prompt_assembly_service::PromptAssemblyService; use tt_application::services::skill_service::SkillService; @@ -127,7 +128,12 @@ async fn character_service_with_world_repository( aliases, file_chat_repository.clone(), )); - let chat_repository: Arc = file_chat_repository; + let chat_repository: Arc = file_chat_repository.clone(); + let group_chat_repository: Arc = file_chat_repository; + let chat_history_coordinator = Arc::new(ChatHistoryCoordinator::new( + chat_repository.clone(), + group_chat_repository, + )); let world_repository = Arc::new(FileWorldInfoRepository::new(default_user.join("worlds"))); let agent_repository = Arc::new(FileAgentRepository::new( root.join("_tauritavern/agent-workspaces"), @@ -144,6 +150,7 @@ async fn character_service_with_world_repository( chat_repository, world_repository.clone(), lifecycle_service, + chat_history_coordinator, ), world_repository, ) diff --git a/src-tauri/crates/tauritavern/src/app/host/shutdown.rs b/src-tauri/crates/tauritavern/src/app/host/shutdown.rs index f5881435..dfa49255 100644 --- a/src-tauri/crates/tauritavern/src/app/host/shutdown.rs +++ b/src-tauri/crates/tauritavern/src/app/host/shutdown.rs @@ -60,6 +60,7 @@ pub(super) fn handle_run_event(app_handle: &tauri::AppHandle, event: tauri::RunE .lifecycle .agent_run_retention_automation_cancel .cancel(); + state.lifecycle.chat_history_cancel.cancel(); let lan_sync_service = state.services.lan_sync_service.clone(); tauri::async_runtime::spawn(async move { if let Err(error) = lan_sync_service.shutdown().await { diff --git a/src-tauri/crates/tauritavern/src/app/state.rs b/src-tauri/crates/tauritavern/src/app/state.rs index bde98e40..506a0886 100644 --- a/src-tauri/crates/tauritavern/src/app/state.rs +++ b/src-tauri/crates/tauritavern/src/app/state.rs @@ -14,6 +14,7 @@ use tt_application::services::avatar_service::AvatarService; use tt_application::services::background_service::BackgroundService; use tt_application::services::character_service::CharacterService; use tt_application::services::chat_completion_service::ChatCompletionService; +use tt_application::services::chat_history_coordinator::ChatHistoryCoordinator; use tt_application::services::chat_service::ChatService; use tt_application::services::content_service::ContentService; use tt_application::services::data_archive_service::DataArchiveService; @@ -57,6 +58,7 @@ pub struct AppState { pub(crate) struct AppLifecycle { pub(crate) sync_automation_cancel: CancellationToken, pub(crate) agent_run_retention_automation_cancel: CancellationToken, + pub(crate) chat_history_cancel: CancellationToken, } impl AppLifecycle { @@ -64,6 +66,7 @@ impl AppLifecycle { Self { sync_automation_cancel: CancellationToken::new(), agent_run_retention_automation_cancel: CancellationToken::new(), + chat_history_cancel: CancellationToken::new(), } } } @@ -72,6 +75,7 @@ pub(crate) struct AppServices { pub(crate) character_service: Arc, pub(crate) chat_service: Arc, pub(crate) group_chat_service: Arc, + pub(crate) chat_history_coordinator: Arc, pub(crate) user_service: Arc, pub(crate) settings_service: Arc, pub(crate) user_directory_service: Arc, diff --git a/src-tauri/crates/tauritavern/src/presentation/commands/chat_commands.rs b/src-tauri/crates/tauritavern/src/presentation/commands/chat_commands.rs index 25e6fcb1..00c0bc2f 100644 --- a/src-tauri/crates/tauritavern/src/presentation/commands/chat_commands.rs +++ b/src-tauri/crates/tauritavern/src/presentation/commands/chat_commands.rs @@ -11,6 +11,7 @@ use tt_application::dto::chat_dto::{ HideChatBeforeCursorDto, ImportCharacterChatsDto, ImportChatDto, PatchChatWindowedDto, PinnedCharacterChatDto, RenameChatDto, SaveChatFromFileDto, }; +use tt_application::dto::chat_history_dto::ChatHistoryLocator; use tt_application::errors::ApplicationError; use tt_ports::repositories::chat_repository::{ ChatPayloadChunk, ChatPayloadCursor, ChatPayloadTail, ChatPayloadWindowPatchRequest, @@ -30,6 +31,32 @@ pub async fn get_all_chats( .map_err(map_command_error("Failed to get all chats")) } +#[tauri::command] +pub async fn chat_history_generation_started( + locator: ChatHistoryLocator, + app_state: State<'_, Arc>, +) -> Result<(), CommandError> { + app_state + .services + .chat_history_coordinator + .generation_started(locator) + .await + .map_err(Into::into) +} + +#[tauri::command] +pub async fn chat_history_generation_finished( + locator: ChatHistoryLocator, + app_state: State<'_, Arc>, +) -> Result<(), CommandError> { + app_state + .services + .chat_history_coordinator + .generation_finished(locator) + .await + .map_err(Into::into) +} + #[tauri::command] pub async fn get_chat( character_name: String, @@ -443,6 +470,7 @@ pub async fn patch_chat_payload_windowed( patch, expected_window_line_count, force, + commit_reason, } = dto; log_command(format!( "patch_chat_payload_windowed {}/{}", @@ -460,7 +488,12 @@ pub async fn patch_chat_payload_windowed( app_state .services .chat_service - .patch_chat_payload_windowed(&character_name, &file_name, request) + .patch_chat_payload_windowed( + &character_name, + &file_name, + request, + commit_reason.unwrap_or_default(), + ) .await .map_err(map_command_error("Failed to patch windowed chat payload")) } diff --git a/src-tauri/crates/tauritavern/src/presentation/commands/group_chat_commands.rs b/src-tauri/crates/tauritavern/src/presentation/commands/group_chat_commands.rs index 9618c26c..2dfc5c44 100644 --- a/src-tauri/crates/tauritavern/src/presentation/commands/group_chat_commands.rs +++ b/src-tauri/crates/tauritavern/src/presentation/commands/group_chat_commands.rs @@ -184,6 +184,7 @@ pub async fn patch_group_chat_payload_windowed( patch, expected_window_line_count, force, + commit_reason, } = dto; log_command(format!("patch_group_chat_payload_windowed {}", id)); @@ -198,7 +199,7 @@ pub async fn patch_group_chat_payload_windowed( app_state .services .group_chat_service - .patch_group_chat_payload_windowed(&id, request) + .patch_group_chat_payload_windowed(&id, request, commit_reason.unwrap_or_default()) .await .map_err(map_command_error( "Failed to patch windowed group chat payload", diff --git a/src-tauri/crates/tauritavern/src/presentation/commands/registry.rs b/src-tauri/crates/tauritavern/src/presentation/commands/registry.rs index 467ea5ee..ab77301c 100644 --- a/src-tauri/crates/tauritavern/src/presentation/commands/registry.rs +++ b/src-tauri/crates/tauritavern/src/presentation/commands/registry.rs @@ -22,6 +22,8 @@ pub fn invoke_handler() -> impl Fn(tauri::ipc::Invoke) -> bool + Sen super::character_commands::get_character_chats_by_id, super::character_commands::clear_character_cache, // Chat commands + super::chat_commands::chat_history_generation_started, + super::chat_commands::chat_history_generation_finished, super::chat_commands::get_all_chats, super::chat_commands::get_chat, super::chat_commands::get_character_chats, diff --git a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/backup.rs b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/backup.rs index 3b69af74..e3677248 100644 --- a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/backup.rs +++ b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/backup.rs @@ -20,40 +20,50 @@ impl FileChatRepository { &self, chat_path: &Path, backup_name: &str, - backup_key: &str, - ) { - let Ok(mut state) = self.backup_history.try_lock() else { - return; + ) -> Result<(), DomainError> { + let policy = match self.backup_policy.try_read() { + Ok(policy) => *policy, + Err(_) => { + return Err(DomainError::transient( + "Chat backup policy is being updated", + )); + } }; - let Ok(policy) = self.backup_policy.try_read() else { - return; - }; - let policy = *policy; - if !policy.automatic_enabled - || policy.history_disabled() - || !state.throttle.should_backup(backup_key) - { - return; + if !policy.automatic_enabled || policy.history_disabled() { + return Ok(()); } - let result = match &mut state.inventory { - BackupInventoryState::Ready(inventory) => { - self.publish_chat_backup(chat_path, backup_name, policy, inventory) - .await + let Ok(mut state) = self.backup_history.try_lock() else { + return Err(DomainError::transient("Chat backup history is busy")); + }; + let inventory = match &mut state.inventory { + BackupInventoryState::Ready(inventory) => inventory, + BackupInventoryState::Uninitialized => { + return Err(DomainError::transient( + "Chat backup inventory is initializing", + )); + } + BackupInventoryState::Failed(message) => { + return Err(DomainError::InternalError(format!( + "Chat backup inventory is unavailable: {}", + message + ))); } - BackupInventoryState::Uninitialized | BackupInventoryState::Failed(_) => return, }; - match result { - Ok(()) => state.throttle.update(backup_key), - Err(DomainError::Conflict(message)) => { - tracing::warn!("Skipping automatic chat backup: {}", message) + match self + .publish_chat_backup(chat_path, backup_name, policy, inventory) + .await + { + Ok(()) => { + tracing::info!(source = ?chat_path, "Created automatic chat backup"); + Ok(()) } - Err(error) => tracing::error!( - target: tt_contracts::observability::USER_VISIBLE_ERROR, - "Automatic chat backup failed after the current chat was saved: {}", - error - ), + Err(DomainError::Conflict(message)) => { + tracing::warn!(reason = %message, "Skipping automatic chat backup"); + Ok(()) + } + Err(error) => Err(error), } } diff --git a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/backup_inventory.rs b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/backup_inventory.rs index c2d51091..0ab8f378 100644 --- a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/backup_inventory.rs +++ b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/backup_inventory.rs @@ -7,7 +7,6 @@ use tt_domain::errors::DomainError; use tt_domain::models::settings::ChatBackupSettings; use super::FileChatRepository; -use super::cache::ThrottledBackup; pub(super) const BACKUP_TEMP_PREFIX: &str = ".tmp-chat-backup-"; @@ -63,14 +62,12 @@ pub(super) enum BackupInventoryState { pub(super) struct BackupHistoryState { pub inventory: BackupInventoryState, - pub throttle: ThrottledBackup, } impl BackupHistoryState { - pub fn new(throttle: ThrottledBackup) -> Self { + pub fn new() -> Self { Self { inventory: BackupInventoryState::Uninitialized, - throttle, } } } diff --git a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/cache.rs b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/cache.rs index c80131cc..3ad89611 100644 --- a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/cache.rs +++ b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/cache.rs @@ -57,33 +57,3 @@ impl MemoryCache { self.chats.clear(); } } - -/// Throttled function for backups -pub(super) struct ThrottledBackup { - last_backup: HashMap, - interval: Duration, -} - -impl ThrottledBackup { - /// Create a new throttled backup with the specified interval - pub(super) fn new(interval_seconds: u64) -> Self { - Self { - last_backup: HashMap::new(), - interval: Duration::from_secs(interval_seconds), - } - } - - /// Check if a backup should be performed - pub(super) fn should_backup(&self, key: &str) -> bool { - if let Some(last) = self.last_backup.get(key) { - last.elapsed() >= self.interval - } else { - true - } - } - - /// Update the last backup time - pub(super) fn update(&mut self, key: &str) { - self.last_backup.insert(key.to_string(), Instant::now()); - } -} diff --git a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/group_chat_repository_impl.rs b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/group_chat_repository_impl.rs index 0e5820ba..44ec6605 100644 --- a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/group_chat_repository_impl.rs +++ b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/group_chat_repository_impl.rs @@ -189,15 +189,33 @@ impl GroupChatRepository for FileChatRepository { ) -> Result<(), DomainError> { self.ensure_directory_exists().await?; let path = self.get_group_chat_path(chat_id)?; - let backup_key = Self::get_group_backup_key(chat_id)?; - self.write_payload_file_to_path(&path, source_path, force, chat_id, &backup_key) + self.write_payload_file_to_path(&path, source_path, force) .await?; self.remove_summary_cache_for_path(&path).await; Ok(()) } + async fn backup_group_chat_automatic(&self, chat_id: &str) -> Result<(), DomainError> { + let path = self.get_group_chat_path(chat_id)?; + let Some(_write_guard) = self.try_acquire_payload_write_lock(&path).await else { + return Err(DomainError::transient(format!( + "Group chat current is busy: {}", + path.display() + ))); + }; + if !path.exists() { + return Err(DomainError::NotFound(format!( + "Group chat not found: {}", + chat_id + ))); + } + + self.backup_chat_file_automatic(&path, chat_id).await + } + async fn delete_group_chat_payload(&self, chat_id: &str) -> Result<(), DomainError> { let path = self.get_group_chat_path(chat_id)?; + let _write_guard = self.acquire_payload_write_lock(&path).await; if !path.exists() { return Err(DomainError::NotFound(format!( "Group chat not found: {}", diff --git a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/mod.rs b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/mod.rs index ea001086..15e028f5 100644 --- a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/mod.rs +++ b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/mod.rs @@ -31,7 +31,7 @@ mod windowed_payload_io; mod tests; use self::backup_inventory::BackupHistoryState; -use self::cache::{MemoryCache, ThrottledBackup}; +use self::cache::MemoryCache; use self::summary::SummaryCache; use crate::chat_directory_identity::{ SharedChatAliasStore, chat_alias_path_for_user_dir, new_shared_chat_alias_store, @@ -138,9 +138,7 @@ impl FileChatRepository { summary_cache, chat_aliases, backup_policy: Arc::new(RwLock::new(backup_settings)), - backup_history: Arc::new(Mutex::new(BackupHistoryState::new(ThrottledBackup::new( - 10, - )))), + backup_history: Arc::new(Mutex::new(BackupHistoryState::new())), } } } diff --git a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/paths.rs b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/paths.rs index cc2da116..48b4bd5a 100644 --- a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/paths.rs +++ b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/paths.rs @@ -53,11 +53,11 @@ impl FileChatRepository { sanitize_chat_dir_key(value, fallback) } - pub(super) async fn acquire_payload_write_lock(&self, path: &Path) -> OwnedMutexGuard<()> { + async fn payload_write_lock(&self, path: &Path) -> Arc> { const MAX_RETAINED_LOCK_ENTRIES: usize = 2048; let key = path.to_path_buf(); - let lock = { + { let mut locks = self.path_write_locks.lock().await; if locks.len() > MAX_RETAINED_LOCK_ENTRIES { locks.retain(|_, value| value.strong_count() > 0); @@ -71,9 +71,18 @@ impl FileChatRepository { created } } - }; + } + } - lock.lock_owned().await + pub(super) async fn acquire_payload_write_lock(&self, path: &Path) -> OwnedMutexGuard<()> { + self.payload_write_lock(path).await.lock_owned().await + } + + pub(super) async fn try_acquire_payload_write_lock( + &self, + path: &Path, + ) -> Option> { + self.payload_write_lock(path).await.try_lock_owned().ok() } pub(super) async fn acquire_payload_rename_locks( @@ -276,11 +285,4 @@ impl FileChatRepository { Self::normalize_jsonl_file_stem(file_name)? )) } - - pub(super) fn get_group_backup_key(chat_id: &str) -> Result { - Ok(format!( - "group:{}", - Self::normalize_jsonl_file_stem(chat_id)? - )) - } } diff --git a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/payload.rs b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/payload.rs index d3f972d1..53c38068 100644 --- a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/payload.rs +++ b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/payload.rs @@ -174,8 +174,6 @@ impl FileChatRepository { path: &Path, payload: &[Value], force: bool, - backup_name: &str, - backup_key: &str, ) -> Result<(), DomainError> { if payload.is_empty() { return Err(DomainError::InvalidData( @@ -187,8 +185,6 @@ impl FileChatRepository { self.verify_chat_integrity_if_needed(path, payload, force) .await?; write_jsonl_file(path, payload).await?; - self.backup_chat_file_automatic(path, backup_name, backup_key) - .await; Ok(()) } @@ -198,8 +194,6 @@ impl FileChatRepository { path: &Path, source_path: &Path, force: bool, - backup_name: &str, - backup_key: &str, ) -> Result<(), DomainError> { if !source_path.exists() { return Err(DomainError::NotFound(format!( @@ -229,8 +223,6 @@ impl FileChatRepository { })?; replace_file_with_fallback(&temp_path, path).await?; - self.backup_chat_file_automatic(path, backup_name, backup_key) - .await; Ok(()) } @@ -295,8 +287,7 @@ impl FileChatRepository { let objects = Self::build_payload_from_chat(chat)?; - self.write_payload_to_path(&path, &objects, force, &chat.character_name, &backup_key) - .await?; + self.write_payload_to_path(&path, &objects, force).await?; // Update cache let mut cache = self.memory_cache.lock().await; diff --git a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/repository_impl.rs b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/repository_impl.rs index 5ac09185..ba68faa0 100644 --- a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/repository_impl.rs +++ b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/repository_impl.rs @@ -164,6 +164,7 @@ impl ChatRepository for FileChatRepository { let path = self .resolve_character_chat_path(character_name, file_name) .await?; + let _write_guard = self.acquire_payload_write_lock(&path).await; if !path.exists() { return Err(DomainError::NotFound(format!( @@ -470,6 +471,7 @@ impl ChatRepository for FileChatRepository { let chat_path = self .resolve_character_chat_path(character_name, file_name) .await?; + let _write_guard = self.acquire_payload_write_lock(&chat_path).await; if !chat_path.exists() { return Err(DomainError::NotFound(format!( "Chat not found: {}/{}", @@ -477,11 +479,35 @@ impl ChatRepository for FileChatRepository { ))); } - let _write_guard = self.acquire_payload_write_lock(&chat_path).await; self.backup_chat_file_explicit(&chat_path, character_name) .await } + async fn backup_chat_automatic( + &self, + character_name: &str, + file_name: &str, + ) -> Result<(), DomainError> { + let chat_path = self + .resolve_character_chat_path(character_name, file_name) + .await?; + let Some(_write_guard) = self.try_acquire_payload_write_lock(&chat_path).await else { + return Err(DomainError::transient(format!( + "Chat current is busy: {}", + chat_path.display() + ))); + }; + if !chat_path.exists() { + return Err(DomainError::NotFound(format!( + "Chat not found: {}/{}", + character_name, file_name + ))); + } + + self.backup_chat_file_automatic(&chat_path, character_name) + .await + } + async fn list_chat_backups(&self) -> Result, DomainError> { let descriptors = self.list_chat_backup_files().await?; let mut results = Vec::with_capacity(descriptors.len()); @@ -645,7 +671,7 @@ impl ChatRepository for FileChatRepository { })?; } - self.write_payload_file_to_path(&path, source_path, force, character_name, &backup_key) + self.write_payload_file_to_path(&path, source_path, force) .await?; { diff --git a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/tests.rs b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/tests.rs index 3f99efae..972e1d12 100644 --- a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/tests.rs +++ b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/tests.rs @@ -213,18 +213,6 @@ fn backup_name_reserved_windows_name_becomes_empty() { assert_eq!(key, ""); } -#[test] -fn group_backup_key_uses_committed_chat_file_stem() { - assert_eq!( - FileChatRepository::get_group_backup_key("group/name.jsonl").expect("group backup key"), - "group:groupname" - ); - assert_eq!( - FileChatRepository::get_group_backup_key("Story.JSONL").expect("group backup key"), - "group:Story.JSONL" - ); -} - #[test] fn backup_file_prefix_matches_sillytavern_pattern() { let prefix = FileChatRepository::backup_file_prefix("A:li*ce Name"); @@ -373,6 +361,12 @@ async fn automatic_quota_rejection_does_not_fail_current_save() { .expect("current save must succeed"); assert!(backup_file_names(&root).await.is_empty()); + repository + .backup_chat_automatic("Alice", "session") + .await + .expect("automatic quota rejection is an expected skip"); + assert!(backup_file_names(&root).await.is_empty()); + let error = repository .backup_chat("Alice", "session") .await @@ -387,6 +381,76 @@ async fn automatic_quota_rejection_does_not_fail_current_save() { ); } +#[tokio::test] +async fn automatic_character_and_group_snapshots_run_only_when_requested() { + let (repository, root) = setup_repository().await; + apply_and_reconcile_backups(&repository, backup_policy(-1, -1, -1)).await; + let source = root.join("source.jsonl"); + fs::write( + &source, + payload_to_jsonl(&payload_with_integrity("automatic")), + ) + .await + .expect("write source"); + + repository + .save_chat_payload_from_path("Alice", "session", &source, false) + .await + .expect("save character current"); + repository + .save_group_chat_payload_from_path("group-session", &source, false) + .await + .expect("save group current"); + assert!(backup_file_names(&root).await.is_empty()); + + repository + .backup_chat_automatic("Alice", "session") + .await + .expect("automatic character snapshot"); + repository + .backup_group_chat_automatic("group-session") + .await + .expect("automatic group snapshot"); + + assert_eq!(backup_file_names(&root).await.len(), 2); +} + +#[tokio::test] +async fn automatic_snapshot_defers_instead_of_waiting_for_a_busy_current() { + let (repository, root) = setup_repository().await; + apply_and_reconcile_backups(&repository, backup_policy(-1, -1, -1)).await; + let source = root.join("source.jsonl"); + fs::write( + &source, + payload_to_jsonl(&payload_with_integrity("automatic-busy")), + ) + .await + .expect("write source"); + repository + .save_chat_payload_from_path("Alice", "session", &source, false) + .await + .expect("save character current"); + + let current_path = repository + .get_chat_payload_path("Alice", "session") + .await + .expect("resolve current path"); + let current_guard = repository.acquire_payload_write_lock(¤t_path).await; + let error = repository + .backup_chat_automatic("Alice", "session") + .await + .expect_err("busy current should defer the automatic snapshot"); + assert!(matches!(error, DomainError::Transient(_))); + assert!(backup_file_names(&root).await.is_empty()); + + drop(current_guard); + repository + .backup_chat_automatic("Alice", "session") + .await + .expect("automatic snapshot after current writer finishes"); + assert_eq!(backup_file_names(&root).await.len(), 1); +} + #[tokio::test] async fn automatic_toggle_does_not_disable_explicit_backups() { let (repository, root) = setup_repository().await; diff --git a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/windowed_hide.rs b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/windowed_hide.rs index 63a5f47a..a79e4cc6 100644 --- a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/windowed_hide.rs +++ b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/windowed_hide.rs @@ -47,9 +47,6 @@ impl FileChatRepository { } self.remove_summary_cache_for_path(&path).await; - self.backup_chat_file_automatic(&path, character_name, &backup_key) - .await; - Ok(result) } @@ -65,7 +62,6 @@ impl FileChatRepository { let path = self.get_group_chat_path(chat_id)?; let _write_guard = self.acquire_payload_write_lock(&path).await; - let backup_key = Self::get_group_backup_key(chat_id)?; let result = hide_payload_before_cursor_internal( &path, cursor, @@ -76,8 +72,6 @@ impl FileChatRepository { .await?; self.remove_summary_cache_for_path(&path).await; - self.backup_chat_file_automatic(&path, chat_id, &backup_key) - .await; Ok(result) } diff --git a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/windowed_patch.rs b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/windowed_patch.rs index f79a1e92..44413906 100644 --- a/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/windowed_patch.rs +++ b/src-tauri/crates/tt-adapter-storage-core/src/repositories/file_chat_repository/windowed_patch.rs @@ -42,9 +42,6 @@ impl FileChatRepository { } self.remove_summary_cache_for_path(&path).await; - self.backup_chat_file_automatic(&path, character_name, &backup_key) - .await; - Ok(result) } @@ -57,12 +54,9 @@ impl FileChatRepository { let path = self.get_group_chat_path(chat_id)?; let _write_guard = self.acquire_payload_write_lock(&path).await; - let backup_key = Self::get_group_backup_key(chat_id)?; let result = patch_payload_windowed_internal(&path, request).await?; self.remove_summary_cache_for_path(&path).await; - self.backup_chat_file_automatic(&path, chat_id, &backup_key) - .await; Ok(result) } diff --git a/src-tauri/crates/tt-application/src/dto/chat_dto.rs b/src-tauri/crates/tt-application/src/dto/chat_dto.rs index 813112ab..5273beb1 100644 --- a/src-tauri/crates/tt-application/src/dto/chat_dto.rs +++ b/src-tauri/crates/tt-application/src/dto/chat_dto.rs @@ -5,6 +5,8 @@ use tt_ports::repositories::chat_repository::{ ChatPayloadCursor, ChatPayloadPatchOp, ChatSearchResult, PinnedCharacterChat, PinnedGroupChat, }; +use super::chat_history_dto::CurrentCommitReason; + /// DTO for chat message extra data #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MessageExtraDto { @@ -160,6 +162,7 @@ pub struct SaveChatFromFileDto { pub file_name: String, pub file_path: String, pub force: Option, + pub commit_reason: Option, } /// DTO for patching a windowed character chat payload. @@ -173,6 +176,7 @@ pub struct PatchChatWindowedDto { pub patch: ChatPayloadPatchOp, pub expected_window_line_count: usize, pub force: Option, + pub commit_reason: Option, } /// DTO for toggling the hidden flag on messages before the window cursor. @@ -203,6 +207,7 @@ pub struct SaveGroupChatFromFileDto { pub id: String, pub file_path: String, pub force: Option, + pub commit_reason: Option, } /// DTO for patching a windowed group chat payload. @@ -214,6 +219,7 @@ pub struct PatchGroupChatWindowedDto { pub patch: ChatPayloadPatchOp, pub expected_window_line_count: usize, pub force: Option, + pub commit_reason: Option, } /// DTO for deleting a group chat payload diff --git a/src-tauri/crates/tt-application/src/dto/chat_history_dto.rs b/src-tauri/crates/tt-application/src/dto/chat_history_dto.rs new file mode 100644 index 00000000..f39a354b --- /dev/null +++ b/src-tauri/crates/tt-application/src/dto/chat_history_dto.rs @@ -0,0 +1,60 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)] +#[serde(tag = "kind")] +pub enum ChatHistoryLocator { + #[serde(rename = "character")] + Character { + #[serde(rename = "characterId")] + character_id: String, + #[serde(rename = "fileName")] + file_name: String, + }, + #[serde(rename = "group")] + Group { + #[serde(rename = "chatId")] + chat_id: String, + }, +} + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum CurrentCommitReason { + #[default] + Mutation, + ProviderBarrier, + GenerationCheckpoint, + Maintenance, +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + #[test] + fn frontend_locator_and_reason_shape_is_camel_case() { + let locator: ChatHistoryLocator = serde_json::from_value(json!({ + "kind": "character", + "characterId": "Alice#1", + "fileName": "Story", + })) + .expect("deserialize character locator"); + assert_eq!( + locator, + ChatHistoryLocator::Character { + character_id: "Alice#1".to_string(), + file_name: "Story".to_string(), + } + ); + + let reason: CurrentCommitReason = serde_json::from_value(json!("generationCheckpoint")) + .expect("deserialize commit reason"); + assert_eq!(reason, CurrentCommitReason::GenerationCheckpoint); + + let maintenance: CurrentCommitReason = + serde_json::from_value(json!("maintenance")).expect("deserialize maintenance reason"); + assert_eq!(maintenance, CurrentCommitReason::Maintenance); + } +} diff --git a/src-tauri/crates/tt-application/src/dto/mod.rs b/src-tauri/crates/tt-application/src/dto/mod.rs index f00d3e94..4cc410d5 100644 --- a/src-tauri/crates/tt-application/src/dto/mod.rs +++ b/src-tauri/crates/tt-application/src/dto/mod.rs @@ -5,6 +5,7 @@ pub mod bootstrap_dto; pub mod character_dto; pub mod chat_completion_dto; pub mod chat_dto; +pub mod chat_history_dto; pub mod data_archive_dto; pub mod dev_observability_dto; pub mod group_dto; diff --git a/src-tauri/crates/tt-application/src/services/character_service.rs b/src-tauri/crates/tt-application/src/services/character_service.rs index 6e817859..804f1a04 100644 --- a/src-tauri/crates/tt-application/src/services/character_service.rs +++ b/src-tauri/crates/tt-application/src/services/character_service.rs @@ -15,6 +15,7 @@ use crate::errors::ApplicationError; use crate::services::agent_workspace_lifecycle_service::{ AgentChatWorkspaceTarget, AgentWorkspaceLifecycleService, }; +use crate::services::chat_history_coordinator::ChatHistoryCoordinator; use serde_json::Value; use sha2::{Digest, Sha256}; use std::collections::HashSet; @@ -38,6 +39,7 @@ pub struct CharacterService { chat_repository: Arc, world_info_repository: Arc, agent_workspace_lifecycle_service: Arc, + chat_history_coordinator: Arc, } #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -59,12 +61,14 @@ impl CharacterService { chat_repository: Arc, world_info_repository: Arc, agent_workspace_lifecycle_service: Arc, + chat_history_coordinator: Arc, ) -> Self { Self { repository, chat_repository, world_info_repository, agent_workspace_lifecycle_service, + chat_history_coordinator, } } @@ -471,7 +475,18 @@ impl CharacterService { .ensure_chat_workspaces_inactive(&workspace_targets) .await?; + let execution_guard = self + .chat_history_coordinator + .lock_snapshot_execution() + .await; + self.chat_history_coordinator + .invalidate_character(&dto.name) + .await; self.repository.delete(&dto.name, dto.delete_chats).await?; + self.chat_history_coordinator + .invalidate_character(&dto.name) + .await; + drop(execution_guard); self.agent_workspace_lifecycle_service .delete_chat_workspaces(&workspace_targets) .await?; @@ -510,7 +525,24 @@ impl CharacterService { self.validate_character_name(&dto.new_name)?; tracing::debug!("Renaming character: {} -> {}", dto.old_name, dto.new_name); + let execution_guard = self + .chat_history_coordinator + .lock_snapshot_execution() + .await; + self.chat_history_coordinator + .invalidate_character(&dto.old_name) + .await; + self.chat_history_coordinator + .invalidate_character(&dto.new_name) + .await; let character = self.repository.rename(&dto.old_name, &dto.new_name).await?; + self.chat_history_coordinator + .invalidate_character(&dto.old_name) + .await; + self.chat_history_coordinator + .invalidate_character(&dto.new_name) + .await; + drop(execution_guard); Ok(CharacterDto::from(character)) } diff --git a/src-tauri/crates/tt-application/src/services/chat_history_coordinator.rs b/src-tauri/crates/tt-application/src/services/chat_history_coordinator.rs new file mode 100644 index 00000000..32c107de --- /dev/null +++ b/src-tauri/crates/tt-application/src/services/chat_history_coordinator.rs @@ -0,0 +1,1163 @@ +use std::cmp::max; +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::{Mutex, MutexGuard, Notify}; +use tokio::time::{Instant, sleep_until}; +use tokio_util::sync::CancellationToken; +use tt_domain::errors::DomainError; +use tt_ports::repositories::chat_repository::ChatRepository; +use tt_ports::repositories::group_chat_repository::GroupChatRepository; + +use crate::dto::chat_history_dto::{ChatHistoryLocator, CurrentCommitReason}; +use crate::errors::ApplicationError; +use crate::services::chat_file_validation::{ + validate_character_path_component, validate_chat_file_name, +}; + +const DEFAULT_QUIET_PERIOD: Duration = Duration::from_secs(2); +const DEFAULT_MINIMUM_INTERVAL: Duration = Duration::from_secs(5 * 60); +const DEFAULT_MAX_PENDING: usize = 32; + +/// Owns automatic chat-history scheduling. Current writes only report successful +/// commits; this service coalesces them and performs history copies off the write path. +pub struct ChatHistoryCoordinator { + chat_repository: Arc, + group_chat_repository: Arc, + state: Mutex, + wake: Notify, + execution_gate: Mutex<()>, + quiet_period: Duration, + minimum_interval: Duration, + max_pending: usize, +} + +impl ChatHistoryCoordinator { + pub fn new( + chat_repository: Arc, + group_chat_repository: Arc, + ) -> Self { + Self { + chat_repository, + group_chat_repository, + state: Mutex::new(CoordinatorState::new(Instant::now())), + wake: Notify::new(), + execution_gate: Mutex::new(()), + quiet_period: DEFAULT_QUIET_PERIOD, + minimum_interval: DEFAULT_MINIMUM_INTERVAL, + max_pending: DEFAULT_MAX_PENDING, + } + } + + pub async fn generation_started( + &self, + locator: ChatHistoryLocator, + ) -> Result<(), ApplicationError> { + validate_locator(&locator)?; + let mut state = self.state.lock().await; + let replaced_stale_generation = state.begin_generation(locator); + drop(state); + if replaced_stale_generation { + tracing::warn!("Replacing stale outer chat-history generation state"); + } + self.wake.notify_one(); + Ok(()) + } + + pub async fn generation_finished( + &self, + locator: ChatHistoryLocator, + ) -> Result<(), ApplicationError> { + validate_locator(&locator)?; + let mut state = self.state.lock().await; + let outcome = state.finish_generation( + &locator, + Instant::now(), + self.quiet_period, + self.max_pending, + )?; + drop(state); + if outcome == NoteOutcome::PendingCapacityReached { + tracing::warn!( + max_pending = self.max_pending, + "Automatic chat-history queue is full; skipping a completed generation" + ); + } + self.wake.notify_one(); + Ok(()) + } + + /// Report a current commit only after its repository operation has succeeded. + /// History scheduling remains best-effort and never changes current-write results. + pub async fn note_current_committed( + &self, + locator: ChatHistoryLocator, + reason: CurrentCommitReason, + ) { + let mut state = self.state.lock().await; + let outcome = state.note_commit( + locator, + reason, + Instant::now(), + self.quiet_period, + self.max_pending, + ); + drop(state); + + match outcome { + NoteOutcome::WakeWorker => self.wake.notify_one(), + NoteOutcome::PendingCapacityReached => tracing::warn!( + max_pending = self.max_pending, + "Automatic chat-history queue is full; skipping a new locator" + ), + NoteOutcome::NoWake => {} + } + } + + /// Explicit history snapshots stay immediate and observable, but share the + /// execution gate and global rate boundary with automatic snapshots. + pub async fn backup_character_explicit( + &self, + character_id: &str, + file_name: &str, + ) -> Result<(), ApplicationError> { + validate_character_path_component(character_id)?; + validate_chat_file_name(file_name, "Chat file name")?; + + let locator = ChatHistoryLocator::Character { + character_id: character_id.to_string(), + file_name: file_name.to_string(), + }; + let _execution_guard = self.execution_gate.lock().await; + let committed_before = self.state.lock().await.next_commit_seq; + + self.chat_repository + .backup_chat(character_id, file_name) + .await?; + + let mut state = self.state.lock().await; + state.complete_explicit( + &locator, + committed_before, + Instant::now(), + self.minimum_interval, + ); + drop(state); + self.wake.notify_one(); + Ok(()) + } + + pub async fn invalidate(&self, locator: &ChatHistoryLocator) { + let mut state = self.state.lock().await; + state.invalidate(locator); + drop(state); + self.wake.notify_one(); + } + + pub async fn invalidate_character(&self, character_id: &str) { + let mut state = self.state.lock().await; + state.invalidate_character(character_id); + drop(state); + self.wake.notify_one(); + } + + pub async fn invalidate_all_pending(&self) { + let mut state = self.state.lock().await; + state.invalidate_all_pending(); + drop(state); + self.wake.notify_one(); + } + + pub async fn lock_snapshot_execution(&self) -> MutexGuard<'_, ()> { + self.execution_gate.lock().await + } + + pub async fn run(self: Arc, cancel: CancellationToken) { + loop { + let notified = self.wake.notified(); + tokio::pin!(notified); + + match self.worker_decision(Instant::now()).await { + WorkerDecision::Stopped => { + tokio::select! { + _ = &mut notified => continue, + _ = cancel.cancelled() => break, + } + } + WorkerDecision::WaitUntil(deadline) => { + tokio::select! { + _ = sleep_until(deadline) => {} + _ = &mut notified => {} + _ = cancel.cancelled() => break, + } + } + WorkerDecision::Ready => { + let execution_guard = tokio::select! { + guard = self.execution_gate.lock() => guard, + _ = cancel.cancelled() => break, + }; + if cancel.is_cancelled() { + break; + } + + let attempt = { + let mut state = self.state.lock().await; + state.take_ready(Instant::now()) + }; + let Some(attempt) = attempt else { + drop(execution_guard); + continue; + }; + + let result = self.create_automatic_snapshot(&attempt.locator).await; + drop(execution_guard); + + let mut state = self.state.lock().await; + match &result { + Err(DomainError::Transient(_)) => { + state.defer_automatic(&attempt, Instant::now() + self.quiet_period) + } + _ => state.complete_automatic( + &attempt, + Instant::now(), + self.minimum_interval, + ), + } + drop(state); + + match result { + Ok(()) => {} + Err(DomainError::Transient(message)) => tracing::debug!( + locator = ?attempt.locator, + commit_seq = attempt.commit_seq, + reason = %message, + "Deferring busy automatic chat-history request" + ), + Err(DomainError::NotFound(message)) => tracing::debug!( + locator = ?attempt.locator, + commit_seq = attempt.commit_seq, + reason = %message, + "Skipping stale automatic chat-history request" + ), + Err(error) => tracing::error!( + target: tt_contracts::observability::USER_VISIBLE_ERROR, + locator = ?attempt.locator, + commit_seq = attempt.commit_seq, + error = %error, + "Automatic chat-history snapshot failed" + ), + } + self.wake.notify_one(); + } + } + } + } + + async fn worker_decision(&self, now: Instant) -> WorkerDecision { + self.state.lock().await.worker_decision(now) + } + + async fn create_automatic_snapshot( + &self, + locator: &ChatHistoryLocator, + ) -> Result<(), DomainError> { + match locator { + ChatHistoryLocator::Character { + character_id, + file_name, + } => { + self.chat_repository + .backup_chat_automatic(character_id, file_name) + .await + } + ChatHistoryLocator::Group { chat_id } => { + self.group_chat_repository + .backup_group_chat_automatic(chat_id) + .await + } + } + } +} + +fn validate_locator(locator: &ChatHistoryLocator) -> Result<(), ApplicationError> { + match locator { + ChatHistoryLocator::Character { + character_id, + file_name, + } => { + validate_character_path_component(character_id)?; + validate_chat_file_name(file_name, "Chat file name") + } + ChatHistoryLocator::Group { chat_id } => validate_chat_file_name(chat_id, "Group chat id"), + } +} + +#[derive(Debug)] +struct CoordinatorState { + active_generation: Option, + pending: HashMap, + in_flight: Option, + next_commit_seq: u64, + next_order: u64, + next_automatic_at: Instant, +} + +impl CoordinatorState { + fn new(now: Instant) -> Self { + Self { + active_generation: None, + pending: HashMap::new(), + in_flight: None, + next_commit_seq: 0, + next_order: 0, + next_automatic_at: now, + } + } + + fn begin_generation(&mut self, locator: ChatHistoryLocator) -> bool { + let replaced_stale_generation = self.active_generation.take().is_some(); + let suspended = self.pending.remove(&locator); + self.active_generation = Some(ActiveGeneration { + locator, + suspended, + latest_commit_seq: self.next_commit_seq, + saw_commit: false, + saw_checkpoint: false, + }); + replaced_stale_generation + } + + fn finish_generation( + &mut self, + locator: &ChatHistoryLocator, + now: Instant, + quiet_period: Duration, + max_pending: usize, + ) -> Result { + let Some(active) = self.active_generation.as_ref() else { + return Err(ApplicationError::Conflict( + "No outer generation is active".to_string(), + )); + }; + if &active.locator != locator { + return Err(ApplicationError::Conflict( + "Generation finish locator does not match its start locator".to_string(), + )); + } + + let active = self + .active_generation + .take() + .expect("active generation was checked above"); + let outcome = if active.saw_checkpoint { + self.schedule( + active.locator, + active.latest_commit_seq, + now + quiet_period, + max_pending, + ) + } else if !active.saw_commit + && let Some(suspended) = active.suspended + { + if self.pending.len() >= max_pending { + NoteOutcome::PendingCapacityReached + } else { + self.pending.insert(active.locator, suspended); + NoteOutcome::WakeWorker + } + } else { + NoteOutcome::NoWake + }; + Ok(outcome) + } + + fn note_commit( + &mut self, + locator: ChatHistoryLocator, + reason: CurrentCommitReason, + now: Instant, + quiet_period: Duration, + max_pending: usize, + ) -> NoteOutcome { + self.next_commit_seq += 1; + let commit_seq = self.next_commit_seq; + + if let Some(active) = self.active_generation.as_mut() + && active.locator == locator + { + active.latest_commit_seq = commit_seq; + active.saw_commit = true; + active.saw_checkpoint |= reason == CurrentCommitReason::GenerationCheckpoint; + return NoteOutcome::NoWake; + } + + match reason { + CurrentCommitReason::ProviderBarrier => { + self.pending.remove(&locator); + NoteOutcome::WakeWorker + } + CurrentCommitReason::Mutation | CurrentCommitReason::GenerationCheckpoint => { + self.schedule(locator, commit_seq, now + quiet_period, max_pending) + } + CurrentCommitReason::Maintenance => { + if let Some(pending) = self.pending.get_mut(&locator) { + pending.commit_seq = commit_seq; + } + NoteOutcome::NoWake + } + } + } + + fn schedule( + &mut self, + locator: ChatHistoryLocator, + commit_seq: u64, + due_at: Instant, + max_pending: usize, + ) -> NoteOutcome { + if let Some(pending) = self.pending.get_mut(&locator) { + pending.commit_seq = commit_seq; + pending.due_at = due_at; + return NoteOutcome::WakeWorker; + } + if self.pending.len() >= max_pending { + return NoteOutcome::PendingCapacityReached; + } + + let order = self.next_order; + self.next_order += 1; + self.pending.insert( + locator, + PendingSnapshot { + commit_seq, + due_at, + order, + }, + ); + NoteOutcome::WakeWorker + } + + fn worker_decision(&self, now: Instant) -> WorkerDecision { + if self.active_generation.is_some() || self.pending.is_empty() { + return WorkerDecision::Stopped; + } + + let deadline = self + .pending + .values() + .map(|pending| max(pending.due_at, self.next_automatic_at)) + .min() + .expect("pending map is not empty"); + if deadline <= now { + WorkerDecision::Ready + } else { + WorkerDecision::WaitUntil(deadline) + } + } + + fn take_ready(&mut self, now: Instant) -> Option { + if self.active_generation.is_some() || self.in_flight.is_some() { + return None; + } + + let (locator, _) = self + .pending + .iter() + .filter_map(|(locator, pending)| { + let deadline = max(pending.due_at, self.next_automatic_at); + (deadline <= now).then_some((locator, (deadline, pending.order))) + }) + .min_by_key(|(_, key)| *key)?; + let locator = locator.clone(); + let pending = self + .pending + .remove(&locator) + .expect("selected pending snapshot must exist"); + let attempt = SnapshotAttempt { + locator, + commit_seq: pending.commit_seq, + }; + self.in_flight = Some(attempt.clone()); + Some(attempt) + } + + fn complete_automatic( + &mut self, + attempt: &SnapshotAttempt, + now: Instant, + minimum_interval: Duration, + ) { + let current = self + .in_flight + .take() + .expect("automatic completion requires an in-flight snapshot"); + assert_eq!(¤t, attempt, "automatic snapshot completion mismatch"); + self.next_automatic_at = max(self.next_automatic_at, now + minimum_interval); + } + + fn defer_automatic(&mut self, attempt: &SnapshotAttempt, retry_at: Instant) { + let current = self + .in_flight + .take() + .expect("automatic deferral requires an in-flight snapshot"); + assert_eq!(¤t, attempt, "automatic snapshot deferral mismatch"); + + if self.pending.contains_key(&attempt.locator) { + return; + } + + // The in-flight attempt owns one reserved slot, so a deferral stays bounded + // at max_pending + 1 even if another locator filled the pending map meanwhile. + let order = self.next_order; + self.next_order += 1; + self.pending.insert( + attempt.locator.clone(), + PendingSnapshot { + commit_seq: attempt.commit_seq, + due_at: retry_at, + order, + }, + ); + } + + fn complete_explicit( + &mut self, + locator: &ChatHistoryLocator, + committed_before: u64, + now: Instant, + minimum_interval: Duration, + ) { + if self + .pending + .get(locator) + .is_some_and(|pending| pending.commit_seq <= committed_before) + { + self.pending.remove(locator); + } + if let Some(active) = self.active_generation.as_mut() + && &active.locator == locator + && active + .suspended + .as_ref() + .is_some_and(|pending| pending.commit_seq <= committed_before) + { + active.suspended = None; + } + self.next_automatic_at = max(self.next_automatic_at, now + minimum_interval); + } + + fn invalidate(&mut self, locator: &ChatHistoryLocator) { + self.pending.remove(locator); + if let Some(active) = self.active_generation.as_mut() + && &active.locator == locator + { + active.suspended = None; + active.saw_checkpoint = false; + } + } + + fn invalidate_character(&mut self, character_id: &str) { + self.pending.retain(|locator, _| { + !matches!( + locator, + ChatHistoryLocator::Character { + character_id: pending_character_id, + .. + } if pending_character_id == character_id + ) + }); + if let Some(active) = self.active_generation.as_mut() + && matches!( + &active.locator, + ChatHistoryLocator::Character { + character_id: active_character_id, + .. + } if active_character_id == character_id + ) + { + active.suspended = None; + active.saw_checkpoint = false; + } + } + + fn invalidate_all_pending(&mut self) { + self.pending.clear(); + if let Some(active) = self.active_generation.as_mut() { + active.suspended = None; + active.saw_checkpoint = false; + } + } +} + +#[derive(Debug)] +struct ActiveGeneration { + locator: ChatHistoryLocator, + suspended: Option, + latest_commit_seq: u64, + saw_commit: bool, + saw_checkpoint: bool, +} + +#[derive(Clone, Debug)] +struct PendingSnapshot { + commit_seq: u64, + due_at: Instant, + order: u64, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct SnapshotAttempt { + locator: ChatHistoryLocator, + commit_seq: u64, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum NoteOutcome { + WakeWorker, + NoWake, + PendingCapacityReached, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum WorkerDecision { + Stopped, + WaitUntil(Instant), + Ready, +} + +#[cfg(test)] +mod tests { + use super::*; + + fn character(file_name: &str) -> ChatHistoryLocator { + ChatHistoryLocator::Character { + character_id: "alice".to_string(), + file_name: file_name.to_string(), + } + } + + #[test] + fn provider_only_generation_does_not_schedule_history() { + let now = Instant::now(); + let locator = character("chat"); + let mut state = CoordinatorState::new(now); + + state.begin_generation(locator.clone()); + state.note_commit( + locator.clone(), + CurrentCommitReason::ProviderBarrier, + now, + Duration::from_secs(2), + 32, + ); + state + .finish_generation(&locator, now, Duration::from_secs(2), 32) + .unwrap(); + + assert!(state.pending.is_empty()); + } + + #[test] + fn checkpoint_and_outer_finish_schedule_the_latest_commit() { + let now = Instant::now(); + let locator = character("chat"); + let mut state = CoordinatorState::new(now); + + state.begin_generation(locator.clone()); + state.note_commit( + locator.clone(), + CurrentCommitReason::GenerationCheckpoint, + now, + Duration::from_secs(2), + 32, + ); + state.note_commit( + locator.clone(), + CurrentCommitReason::Mutation, + now, + Duration::from_secs(2), + 32, + ); + state + .finish_generation(&locator, now, Duration::from_secs(2), 32) + .unwrap(); + + let pending = state.pending.get(&locator).unwrap(); + assert_eq!(pending.commit_seq, 2); + assert_eq!(pending.due_at, now + Duration::from_secs(2)); + } + + #[test] + fn generation_without_a_commit_restores_existing_pending_request() { + let now = Instant::now(); + let locator = character("chat"); + let mut state = CoordinatorState::new(now); + state.note_commit( + locator.clone(), + CurrentCommitReason::Mutation, + now, + Duration::from_secs(2), + 32, + ); + let expected_due = state.pending[&locator].due_at; + + state.begin_generation(locator.clone()); + state + .finish_generation(&locator, now, Duration::from_secs(2), 32) + .unwrap(); + + assert_eq!(state.pending[&locator].due_at, expected_due); + } + + #[test] + fn non_checkpoint_commit_invalidates_pre_generation_pending() { + let now = Instant::now(); + let locator = character("chat"); + let mut state = CoordinatorState::new(now); + state.note_commit( + locator.clone(), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + + state.begin_generation(locator.clone()); + state.note_commit( + locator.clone(), + CurrentCommitReason::ProviderBarrier, + now, + Duration::ZERO, + 32, + ); + state + .finish_generation(&locator, now, Duration::ZERO, 32) + .unwrap(); + + assert!(state.pending.is_empty()); + } + + #[test] + fn active_generation_globally_stops_worker_admission() { + let now = Instant::now(); + let mut state = CoordinatorState::new(now); + state.note_commit( + character("waiting"), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + state.begin_generation(character("active")); + + assert_eq!(state.worker_decision(now), WorkerDecision::Stopped); + assert!(state.take_ready(now).is_none()); + } + + #[test] + fn unrelated_commit_waits_without_being_lost_during_generation() { + let now = Instant::now(); + let active = character("active"); + let waiting = character("waiting"); + let mut state = CoordinatorState::new(now); + state.begin_generation(active.clone()); + + state.note_commit( + waiting.clone(), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + + assert!(state.pending.contains_key(&waiting)); + assert_eq!(state.worker_decision(now), WorkerDecision::Stopped); + state + .finish_generation(&active, now, Duration::ZERO, 32) + .unwrap(); + assert_eq!(state.worker_decision(now), WorkerDecision::Ready); + } + + #[test] + fn new_generation_replaces_stale_generation_state() { + let now = Instant::now(); + let stale = character("stale"); + let current = character("current"); + let mut state = CoordinatorState::new(now); + assert!(!state.begin_generation(stale)); + + assert!(state.begin_generation(current.clone())); + state.note_commit( + current.clone(), + CurrentCommitReason::GenerationCheckpoint, + now, + Duration::ZERO, + 32, + ); + state + .finish_generation(¤t, now, Duration::ZERO, 32) + .unwrap(); + + assert!(state.pending.contains_key(¤t)); + } + + #[test] + fn maintenance_only_refreshes_an_existing_pending_request() { + let now = Instant::now(); + let locator = character("chat"); + let mut state = CoordinatorState::new(now); + + state.note_commit( + locator.clone(), + CurrentCommitReason::Maintenance, + now, + Duration::from_secs(30), + 32, + ); + assert!(state.pending.is_empty()); + + state.note_commit( + locator.clone(), + CurrentCommitReason::Mutation, + now, + Duration::from_secs(10), + 32, + ); + let due_at = state.pending[&locator].due_at; + state.note_commit( + locator.clone(), + CurrentCommitReason::Maintenance, + now + Duration::from_secs(1), + Duration::from_secs(30), + 32, + ); + + assert_eq!(state.pending[&locator].commit_seq, 3); + assert_eq!(state.pending[&locator].due_at, due_at); + } + + #[test] + fn repeated_commits_are_latest_wins_without_growing_the_map() { + let now = Instant::now(); + let locator = character("chat"); + let mut state = CoordinatorState::new(now); + state.note_commit( + locator.clone(), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + state.note_commit( + locator.clone(), + CurrentCommitReason::Mutation, + now + Duration::from_secs(1), + Duration::ZERO, + 32, + ); + + assert_eq!(state.pending.len(), 1); + assert_eq!(state.pending[&locator].commit_seq, 2); + } + + #[test] + fn global_interval_applies_after_an_attempt_even_with_trailing_work() { + let now = Instant::now(); + let first = character("first"); + let second = character("second"); + let mut state = CoordinatorState::new(now); + state.note_commit( + first, + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + state.note_commit( + second, + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + + let attempt = state.take_ready(now).unwrap(); + state.complete_automatic(&attempt, now, Duration::from_secs(60)); + + assert_eq!( + state.worker_decision(now), + WorkerDecision::WaitUntil(now + Duration::from_secs(60)) + ); + } + + #[test] + fn explicit_completion_preserves_a_newer_commit() { + let now = Instant::now(); + let locator = character("chat"); + let mut state = CoordinatorState::new(now); + state.note_commit( + locator.clone(), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + let explicit_started_at = state.next_commit_seq; + state.note_commit( + locator.clone(), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + + state.complete_explicit(&locator, explicit_started_at, now, Duration::from_secs(60)); + + assert_eq!(state.pending[&locator].commit_seq, 2); + } + + #[test] + fn explicit_completion_clears_an_older_suspended_request() { + let now = Instant::now(); + let locator = character("chat"); + let mut state = CoordinatorState::new(now); + state.note_commit( + locator.clone(), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + let explicit_started_at = state.next_commit_seq; + state.begin_generation(locator.clone()); + + state.complete_explicit(&locator, explicit_started_at, now, Duration::from_secs(60)); + state + .finish_generation(&locator, now, Duration::ZERO, 32) + .unwrap(); + + assert!(state.pending.is_empty()); + } + + #[test] + fn in_flight_commit_creates_a_trailing_latest_request() { + let now = Instant::now(); + let locator = character("chat"); + let mut state = CoordinatorState::new(now); + state.note_commit( + locator.clone(), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + let attempt = state.take_ready(now).unwrap(); + + state.note_commit( + locator.clone(), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + state.complete_automatic(&attempt, now, Duration::from_secs(60)); + + assert_eq!(state.pending[&locator].commit_seq, 2); + } + + #[test] + fn transient_attempt_is_deferred_without_consuming_the_global_interval() { + let now = Instant::now(); + let locator = character("chat"); + let mut state = CoordinatorState::new(now); + state.note_commit( + locator.clone(), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + let attempt = state.take_ready(now).unwrap(); + + state.defer_automatic(&attempt, now + Duration::from_secs(2)); + + assert_eq!(state.pending[&locator].commit_seq, attempt.commit_seq); + assert_eq!( + state.worker_decision(now), + WorkerDecision::WaitUntil(now + Duration::from_secs(2)) + ); + assert_eq!(state.next_automatic_at, now); + } + + #[test] + fn transient_attempt_does_not_overwrite_a_newer_trailing_request() { + let now = Instant::now(); + let locator = character("chat"); + let mut state = CoordinatorState::new(now); + state.note_commit( + locator.clone(), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + let attempt = state.take_ready(now).unwrap(); + state.note_commit( + locator.clone(), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + + state.defer_automatic(&attempt, now + Duration::from_secs(2)); + + assert_eq!(state.pending[&locator].commit_seq, 2); + assert_eq!(state.pending[&locator].due_at, now); + } + + #[test] + fn pending_capacity_still_allows_existing_locator_coalescing() { + let now = Instant::now(); + let first = character("first"); + let second = character("second"); + let mut state = CoordinatorState::new(now); + + assert_eq!( + state.note_commit( + first.clone(), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 1, + ), + NoteOutcome::WakeWorker + ); + assert_eq!( + state.note_commit( + second, + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 1, + ), + NoteOutcome::PendingCapacityReached + ); + assert_eq!( + state.note_commit( + first.clone(), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 1, + ), + NoteOutcome::WakeWorker + ); + assert_eq!(state.pending.len(), 1); + assert_eq!(state.pending[&first].commit_seq, 3); + } + + #[test] + fn invalidate_all_preserves_generation_gate_and_global_cooldown() { + let now = Instant::now(); + let active = character("active"); + let waiting = character("waiting"); + let mut state = CoordinatorState::new(now); + state.next_automatic_at = now + Duration::from_secs(60); + state.note_commit( + waiting, + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 32, + ); + state.begin_generation(active.clone()); + + state.invalidate_all_pending(); + + assert!(state.pending.is_empty()); + assert_eq!(state.next_automatic_at, now + Duration::from_secs(60)); + assert_eq!(state.active_generation.as_ref().unwrap().locator, active); + } + + #[test] + fn invalidation_revokes_an_active_checkpoint() { + let now = Instant::now(); + let locator = character("active"); + let mut state = CoordinatorState::new(now); + state.begin_generation(locator.clone()); + state.note_commit( + locator.clone(), + CurrentCommitReason::GenerationCheckpoint, + now, + Duration::ZERO, + 32, + ); + + state.invalidate(&locator); + state + .finish_generation(&locator, now, Duration::ZERO, 32) + .unwrap(); + + assert!(state.pending.is_empty()); + } + + #[test] + fn a_new_checkpoint_after_invalidation_authorizes_history_again() { + let now = Instant::now(); + let locator = character("active"); + let mut state = CoordinatorState::new(now); + state.begin_generation(locator.clone()); + state.note_commit( + locator.clone(), + CurrentCommitReason::GenerationCheckpoint, + now, + Duration::ZERO, + 32, + ); + state.invalidate(&locator); + + state.note_commit( + locator.clone(), + CurrentCommitReason::GenerationCheckpoint, + now, + Duration::ZERO, + 32, + ); + state + .finish_generation(&locator, now, Duration::ZERO, 32) + .unwrap(); + + assert_eq!(state.pending[&locator].commit_seq, 2); + } + + #[test] + fn completed_generation_reports_a_full_pending_map() { + let now = Instant::now(); + let locator = character("active"); + let mut state = CoordinatorState::new(now); + state.note_commit( + character("waiting"), + CurrentCommitReason::Mutation, + now, + Duration::ZERO, + 1, + ); + state.begin_generation(locator.clone()); + state.note_commit( + locator.clone(), + CurrentCommitReason::GenerationCheckpoint, + now, + Duration::ZERO, + 1, + ); + + assert_eq!( + state + .finish_generation(&locator, now, Duration::ZERO, 1) + .unwrap(), + NoteOutcome::PendingCapacityReached, + ); + } +} diff --git a/src-tauri/crates/tt-application/src/services/chat_service.rs b/src-tauri/crates/tt-application/src/services/chat_service.rs index d0e0b265..393587e2 100644 --- a/src-tauri/crates/tt-application/src/services/chat_service.rs +++ b/src-tauri/crates/tt-application/src/services/chat_service.rs @@ -7,6 +7,7 @@ use crate::dto::chat_dto::{ AddMessageDto, ChatDto, ChatSearchResultDto, CreateChatDto, ExportChatDto, ImportCharacterChatsDto, ImportChatDto, RenameChatDto, SaveChatFromFileDto, }; +use crate::dto::chat_history_dto::{ChatHistoryLocator, CurrentCommitReason}; use crate::errors::ApplicationError; use crate::services::agent_workspace_lifecycle_service::{ AgentChatWorkspaceTarget, AgentWorkspaceLifecycleService, @@ -14,6 +15,7 @@ use crate::services::agent_workspace_lifecycle_service::{ use crate::services::chat_file_validation::{ validate_character_path_component, validate_chat_file_name, }; +use crate::services::chat_history_coordinator::ChatHistoryCoordinator; use tt_domain::errors::DomainError; use tt_domain::models::chat::{Chat, ChatMessage, MessageExtra}; use tt_ports::repositories::agent_workspace_lifecycle_repository::{ @@ -32,6 +34,7 @@ pub struct ChatService { chat_repository: Arc, character_repository: Arc, agent_workspace_lifecycle_service: Arc, + chat_history_coordinator: Arc, } impl ChatService { @@ -40,11 +43,13 @@ impl ChatService { chat_repository: Arc, character_repository: Arc, agent_workspace_lifecycle_service: Arc, + chat_history_coordinator: Arc, ) -> Self { Self { chat_repository, character_repository, agent_workspace_lifecycle_service, + chat_history_coordinator, } } @@ -68,6 +73,14 @@ impl ChatService { // Save the chat self.chat_repository.save(&chat).await?; + if let Some(file_name) = chat.file_name.as_deref() { + self.note_current_committed( + &chat.character_name, + file_name, + CurrentCommitReason::Maintenance, + ) + .await; + } Ok(ChatDto::from(chat)) } @@ -152,6 +165,12 @@ impl ChatService { .chat_repository .add_message(&dto.character_name, &dto.file_name, message) .await?; + self.note_current_committed( + &dto.character_name, + &dto.file_name, + CurrentCommitReason::Mutation, + ) + .await; Ok(ChatDto::from(chat)) } @@ -175,6 +194,16 @@ impl ChatService { .rename_chat(&dto.character_name, &dto.old_file_name, &dto.new_file_name) .await?; + self.chat_history_coordinator + .invalidate(&character_locator(&dto.character_name, &dto.old_file_name)) + .await; + self.chat_history_coordinator + .invalidate(&character_locator( + &dto.character_name, + &committed_file_name, + )) + .await; + Ok(committed_file_name) } @@ -207,6 +236,9 @@ impl ChatService { self.chat_repository .delete_chat(character_name, file_name) .await?; + self.chat_history_coordinator + .invalidate(&character_locator(character_name, file_name)) + .await; if let Some(target) = target { self.agent_workspace_lifecycle_service @@ -325,8 +357,8 @@ impl ChatService { ) -> Result<(), ApplicationError> { tracing::info!("Backing up chat: {}/{}", character_name, file_name); - self.chat_repository - .backup_chat(character_name, file_name) + self.chat_history_coordinator + .backup_character_explicit(character_name, file_name) .await?; Ok(()) @@ -405,6 +437,8 @@ impl ChatService { self.chat_repository .set_character_chat_metadata_extension(character_name, file_name, namespace, value) .await?; + self.note_current_committed(character_name, file_name, CurrentCommitReason::Mutation) + .await; Ok(()) } @@ -624,14 +658,18 @@ impl ChatService { character_name: &str, file_name: &str, request: ChatPayloadWindowPatchRequest, + commit_reason: CurrentCommitReason, ) -> Result { validate_character_path_component(character_name)?; validate_chat_file_name(file_name, "Chat file name")?; - self.chat_repository + let cursor = self + .chat_repository .patch_chat_payload_windowed(character_name, file_name, request) - .await - .map_err(Into::into) + .await?; + self.note_current_committed(character_name, file_name, commit_reason) + .await; + Ok(cursor) } /// Set the hidden flag on all messages stored before the window cursor. @@ -647,7 +685,8 @@ impl ChatService { validate_character_path_component(character_name)?; validate_chat_file_name(file_name, "Chat file name")?; - self.chat_repository + let cursor = self + .chat_repository .hide_chat_payload_before_cursor( character_name, file_name, @@ -656,8 +695,10 @@ impl ChatService { name_filter, expected_window_line_count, ) - .await - .map_err(Into::into) + .await?; + self.note_current_committed(character_name, file_name, CurrentCommitReason::Mutation) + .await; + Ok(cursor) } /// Save a character chat payload from a JSONL file path. @@ -668,6 +709,7 @@ impl ChatService { validate_character_path_component(&dto.character_name)?; validate_chat_file_name(&dto.file_name, "Chat file name")?; + let commit_reason = dto.commit_reason.unwrap_or_default(); self.chat_repository .save_chat_payload_from_path( &dto.character_name, @@ -675,8 +717,10 @@ impl ChatService { Path::new(&dto.file_path), dto.force.unwrap_or(false), ) - .await - .map_err(Into::into) + .await?; + self.note_current_committed(&dto.character_name, &dto.file_name, commit_reason) + .await; + Ok(()) } /// Import one or more character chats from an uploaded file. @@ -712,4 +756,22 @@ impl ChatService { .await .map_err(Into::into) } + + async fn note_current_committed( + &self, + character_id: &str, + file_name: &str, + reason: CurrentCommitReason, + ) { + self.chat_history_coordinator + .note_current_committed(character_locator(character_id, file_name), reason) + .await; + } +} + +fn character_locator(character_id: &str, file_name: &str) -> ChatHistoryLocator { + ChatHistoryLocator::Character { + character_id: character_id.to_string(), + file_name: file_name.to_string(), + } } diff --git a/src-tauri/crates/tt-application/src/services/group_chat_service.rs b/src-tauri/crates/tt-application/src/services/group_chat_service.rs index 6acaf4f5..ab2cfed1 100644 --- a/src-tauri/crates/tt-application/src/services/group_chat_service.rs +++ b/src-tauri/crates/tt-application/src/services/group_chat_service.rs @@ -7,9 +7,11 @@ use crate::dto::chat_dto::{ ChatSearchResultDto, DeleteGroupChatDto, ImportGroupChatDto, RenameGroupChatDto, SaveGroupChatFromFileDto, }; +use crate::dto::chat_history_dto::{ChatHistoryLocator, CurrentCommitReason}; use crate::errors::ApplicationError; use crate::services::agent_workspace_lifecycle_service::AgentWorkspaceLifecycleService; use crate::services::chat_file_validation::validate_chat_file_name; +use crate::services::chat_history_coordinator::ChatHistoryCoordinator; use tt_domain::errors::DomainError; use tt_ports::repositories::chat_types::{ ChatMessageSearchHit, ChatMessageSearchQuery, ChatPayloadChunk, ChatPayloadCursor, @@ -22,16 +24,19 @@ use tt_ports::repositories::group_chat_repository::GroupChatRepository; pub struct GroupChatService { group_chat_repository: Arc, agent_workspace_lifecycle_service: Arc, + chat_history_coordinator: Arc, } impl GroupChatService { pub fn new( group_chat_repository: Arc, agent_workspace_lifecycle_service: Arc, + chat_history_coordinator: Arc, ) -> Self { Self { group_chat_repository, agent_workspace_lifecycle_service, + chat_history_coordinator, } } @@ -107,6 +112,8 @@ impl GroupChatService { self.group_chat_repository .set_group_chat_metadata_extension(chat_id, namespace, value) .await?; + self.note_current_committed(chat_id, CurrentCommitReason::Mutation) + .await; Ok(()) } @@ -297,13 +304,16 @@ impl GroupChatService { &self, chat_id: &str, request: ChatPayloadWindowPatchRequest, + commit_reason: CurrentCommitReason, ) -> Result { validate_chat_file_name(chat_id, "Group chat id")?; - self.group_chat_repository + let cursor = self + .group_chat_repository .patch_group_chat_payload_windowed(chat_id, request) - .await - .map_err(Into::into) + .await?; + self.note_current_committed(chat_id, commit_reason).await; + Ok(cursor) } /// Set the hidden flag on all messages stored before the window cursor. @@ -317,7 +327,8 @@ impl GroupChatService { ) -> Result { validate_chat_file_name(chat_id, "Group chat id")?; - self.group_chat_repository + let cursor = self + .group_chat_repository .hide_group_chat_payload_before_cursor( chat_id, cursor, @@ -325,8 +336,10 @@ impl GroupChatService { name_filter, expected_window_line_count, ) - .await - .map_err(Into::into) + .await?; + self.note_current_committed(chat_id, CurrentCommitReason::Mutation) + .await; + Ok(cursor) } /// Save a group chat payload from a JSONL file path. @@ -336,14 +349,16 @@ impl GroupChatService { ) -> Result<(), ApplicationError> { validate_chat_file_name(&dto.id, "Group chat id")?; + let commit_reason = dto.commit_reason.unwrap_or_default(); self.group_chat_repository .save_group_chat_payload_from_path( &dto.id, Path::new(&dto.file_path), dto.force.unwrap_or(false), ) - .await - .map_err(Into::into) + .await?; + self.note_current_committed(&dto.id, commit_reason).await; + Ok(()) } /// Delete a group chat payload file. @@ -358,6 +373,9 @@ impl GroupChatService { self.group_chat_repository .delete_group_chat_payload(&dto.id) .await?; + self.chat_history_coordinator + .invalidate(&group_locator(&dto.id)) + .await; self.agent_workspace_lifecycle_service .delete_chat_workspace(&target) .await?; @@ -376,6 +394,12 @@ impl GroupChatService { .group_chat_repository .rename_group_chat_payload(&dto.old_file_name, &dto.new_file_name) .await?; + self.chat_history_coordinator + .invalidate(&group_locator(&dto.old_file_name)) + .await; + self.chat_history_coordinator + .invalidate(&group_locator(&committed_file_name)) + .await; Ok(committed_file_name) } @@ -389,4 +413,16 @@ impl GroupChatService { .await .map_err(Into::into) } + + async fn note_current_committed(&self, chat_id: &str, reason: CurrentCommitReason) { + self.chat_history_coordinator + .note_current_committed(group_locator(chat_id), reason) + .await; + } +} + +fn group_locator(chat_id: &str) -> ChatHistoryLocator { + ChatHistoryLocator::Group { + chat_id: chat_id.to_string(), + } } diff --git a/src-tauri/crates/tt-application/src/services/group_service.rs b/src-tauri/crates/tt-application/src/services/group_service.rs index 342ad34f..6c39444a 100644 --- a/src-tauri/crates/tt-application/src/services/group_service.rs +++ b/src-tauri/crates/tt-application/src/services/group_service.rs @@ -1,9 +1,11 @@ use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; +use crate::dto::chat_history_dto::ChatHistoryLocator; use crate::dto::group_dto::{CreateGroupDto, DeleteGroupDto, UpdateGroupDto}; use crate::errors::ApplicationError; use crate::services::agent_workspace_lifecycle_service::AgentWorkspaceLifecycleService; +use crate::services::chat_history_coordinator::ChatHistoryCoordinator; use tt_domain::errors::DomainError; use tt_domain::models::group::Group; use tt_ports::repositories::group_repository::GroupRepository; @@ -13,6 +15,7 @@ pub struct GroupService { /// Repository for group data repository: Arc, agent_workspace_lifecycle_service: Arc, + chat_history_coordinator: Arc, } impl GroupService { @@ -20,10 +23,12 @@ impl GroupService { pub fn new( repository: Arc, agent_workspace_lifecycle_service: Arc, + chat_history_coordinator: Arc, ) -> Self { Self { repository, agent_workspace_lifecycle_service, + chat_history_coordinator, } } @@ -112,7 +117,26 @@ impl GroupService { .ensure_chat_workspaces_inactive(&targets) .await?; + let execution_guard = self + .chat_history_coordinator + .lock_snapshot_execution() + .await; + for chat_id in &group.chats { + self.chat_history_coordinator + .invalidate(&ChatHistoryLocator::Group { + chat_id: chat_id.clone(), + }) + .await; + } self.repository.delete_group(&dto.id).await?; + for chat_id in &group.chats { + self.chat_history_coordinator + .invalidate(&ChatHistoryLocator::Group { + chat_id: chat_id.clone(), + }) + .await; + } + drop(execution_guard); self.agent_workspace_lifecycle_service .delete_chat_workspaces(&targets) .await?; diff --git a/src-tauri/crates/tt-application/src/services/mod.rs b/src-tauri/crates/tt-application/src/services/mod.rs index b99d0859..9f8f9a7f 100644 --- a/src-tauri/crates/tt-application/src/services/mod.rs +++ b/src-tauri/crates/tt-application/src/services/mod.rs @@ -19,6 +19,7 @@ pub mod bundled_template_service; pub mod character_service; pub mod chat_completion_service; mod chat_file_validation; +pub mod chat_history_coordinator; pub mod chat_service; pub mod content_service; pub mod data_archive_service; diff --git a/src-tauri/crates/tt-ports/src/repositories/chat_repository.rs b/src-tauri/crates/tt-ports/src/repositories/chat_repository.rs index 77726f76..d746ceb3 100644 --- a/src-tauri/crates/tt-ports/src/repositories/chat_repository.rs +++ b/src-tauri/crates/tt-ports/src/repositories/chat_repository.rs @@ -94,6 +94,16 @@ pub trait ChatRepository: Send + Sync { /// Backup a chat async fn backup_chat(&self, character_name: &str, file_name: &str) -> Result<(), DomainError>; + /// Create an automatic history snapshot for a character chat. + /// + /// Scheduling is owned by the application layer; the repository only + /// serializes the source payload and publishes the snapshot. + async fn backup_chat_automatic( + &self, + character_name: &str, + file_name: &str, + ) -> Result<(), DomainError>; + /// List all chat backup files. async fn list_chat_backups(&self) -> Result, DomainError>; diff --git a/src-tauri/crates/tt-ports/src/repositories/group_chat_repository.rs b/src-tauri/crates/tt-ports/src/repositories/group_chat_repository.rs index da1bc876..ed719cd4 100644 --- a/src-tauri/crates/tt-ports/src/repositories/group_chat_repository.rs +++ b/src-tauri/crates/tt-ports/src/repositories/group_chat_repository.rs @@ -80,6 +80,12 @@ pub trait GroupChatRepository: Send + Sync { force: bool, ) -> Result<(), DomainError>; + /// Create an automatic history snapshot for a group chat. + /// + /// Scheduling is owned by the application layer; the repository only + /// serializes the source payload and publishes the snapshot. + async fn backup_group_chat_automatic(&self, chat_id: &str) -> Result<(), DomainError>; + /// Delete a group chat payload file. async fn delete_group_chat_payload(&self, chat_id: &str) -> Result<(), DomainError>; diff --git a/src/script.js b/src/script.js index 8e47a39e..4ca73f39 100644 --- a/src/script.js +++ b/src/script.js @@ -9,13 +9,14 @@ import { default as libs, lodash, } from './lib.js'; -import { getClientVersion as getBridgeClientVersion } from './tauri-bridge.js'; +import { getClientVersion as getBridgeClientVersion, invoke } from './tauri-bridge.js'; import { SILLYTAVERN_COMPAT_VERSION } from './compat-version.js'; import { registerLifecycleFlushHandler } from './tauri/main/services/lifecycle/lifecycle-flush-service.js'; import { replaceMesTextHtmlWithRuntimePolicy } from './scripts/tauri/message/mes-text-write.js'; import { getCodeHighlightCoordinator } from './scripts/tauri/perf/code-highlight-coordinator.js'; import { isInlineDrawerContentOpen, setInlineDrawerContentOpen } from './scripts/tauri/perf/inline-drawer-motion.js'; import { + CHAT_COMMIT_REASON, isTauriChatPayloadTransportEnabled, loadCharacterChatPayload, loadCharacterChatPayloadTail, @@ -25,6 +26,7 @@ import { saveCharacterChatPayload, patchCharacterChatPayloadWindowed, } from './scripts/chat-payload-transport.js'; +import { getActiveChatSnapshot } from './tauri/main/adapters/st/active-chat-ref.js'; import { buildWindowedPayloadPatch, clearWindowedChatState, @@ -4554,7 +4556,9 @@ class StreamingProcessor { if (!isAborted && power_user.auto_swipe && generatedTextFiltered(text)) { return await swipe(null, SWIPE_DIRECTION.RIGHT, { source: SWIPE_SOURCE.AUTO_SWIPE, repeated: true, forceMesId: chat.length - 1 }); } - await saveChatConditional(); + await saveChatConditional(this.type === 'impersonate' + ? CHAT_COMMIT_REASON.MUTATION + : CHAT_COMMIT_REASON.GENERATION_CHECKPOINT); playMessageSound(); } @@ -5033,26 +5037,58 @@ function removeLastMessage() { const generationIdleGate = createGenerationIdleGate(); let generationInFlightCount = 0; +let generationHistoryLocator = null; export function waitForGenerationIdle() { return generationIdleGate.wait(); } -function enterGeneration() { +async function enterGeneration(dryRun) { if (generationInFlightCount === 0) { generationIdleGate.markBusy(); + generationInFlightCount = 1; + generationHistoryLocator = null; + + // Prompt-assembly dry runs do not define a chat-history transaction. + if (dryRun) return; + + try { + generationHistoryLocator = getActiveChatSnapshot().ref; + } catch (error) { + generationHistoryLocator = null; + console.warn('Failed to capture chat-history generation locator', error); + } + + if (generationHistoryLocator) { + try { + await invoke('chat_history_generation_started', { locator: generationHistoryLocator }); + } catch (error) { + console.error('Failed to report chat-history generation start', error); + } + } + return; } generationInFlightCount += 1; } -function exitGeneration() { +async function exitGeneration() { if (generationInFlightCount <= 0) { throw new Error('Generation in-flight counter underflow'); } generationInFlightCount -= 1; if (generationInFlightCount === 0) { - generationIdleGate.markIdle(); + const locator = generationHistoryLocator; + generationHistoryLocator = null; + try { + if (locator) { + await invoke('chat_history_generation_finished', { locator }); + } + } catch (error) { + console.error('Failed to report chat-history generation finish', error); + } finally { + generationIdleGate.markIdle(); + } } } @@ -5065,14 +5101,14 @@ function exitGeneration() { * @returns {Promise} Returns a promise that resolves when the text is done generating. */ export async function Generate(type, options = {}, dryRun = false) { - enterGeneration(); + await enterGeneration(dryRun); try { return await GenerateInternal(type, options, dryRun); } catch (error) { cleanupGenerationAfterUnhandledError(type, dryRun); throw error; } finally { - exitGeneration(); + await exitGeneration(); } } @@ -5259,11 +5295,11 @@ async function GenerateInternal(type, { automatic_trigger, force_name2, quiet_pr if (messageBias && !removeMacros(textareaText)) { sendSystemMessage(system_message_types.GENERIC, ' ', { bias: messageBias }); } else { - await sendMessageAsUser(textareaText, messageBias); + await sendMessageAsUserAtProviderBarrier(textareaText, messageBias); } } else if (textareaText == '' && !automatic_trigger && !dryRun && [undefined, 'normal'].includes(type) && main_api == 'openai' && oai_settings.send_if_empty.trim().length > 0 && !depth) { // Use send_if_empty if set and the user message is empty. Only when sending messages normally - await sendMessageAsUser(oai_settings.send_if_empty.trim(), messageBias); + await sendMessageAsUserAtProviderBarrier(oai_settings.send_if_empty.trim(), messageBias); } let { @@ -6487,7 +6523,9 @@ async function GenerateInternal(type, { automatic_trigger, force_name2, quiet_pr } console.debug('/api/chats/save called by /Generate'); - await saveChatConditional(); + await saveChatConditional(isImpersonate + ? CHAT_COMMIT_REASON.MUTATION + : CHAT_COMMIT_REASON.GENERATION_CHECKPOINT); unblockGeneration(type); streamingProcessor = null; @@ -6955,9 +6993,10 @@ export function removeMacros(str) { * @param {boolean} [compact] Send as a compact display message. * @param {string} [name] Name of the user sending the message. Defaults to name1. * @param {string} [avatar] Avatar of the user sending the message. Defaults to user_avatar. + * @param {string} [commitReason] Current/history scheduling reason. * @returns {Promise} A promise that resolves to the message when it is inserted. */ -export async function sendMessageAsUser(messageText, messageBias, insertAt = null, compact = false, name = name1, avatar = user_avatar) { +export async function sendMessageAsUser(messageText, messageBias, insertAt = null, compact = false, name = name1, avatar = user_avatar, commitReason = CHAT_COMMIT_REASON.MUTATION) { messageText = getRegexedString(messageText, regex_placement.USER_INPUT); const message = { @@ -6993,13 +7032,13 @@ export async function sendMessageAsUser(messageText, messageBias, insertAt = nul if (typeof insertAt === 'number' && insertAt >= 0 && insertAt <= chat.length) { chat.splice(insertAt, 0, message); markWindowedChatDirtyFromIndex(insertAt); - await saveChatConditional(); + await saveChatConditional(commitReason); await eventSource.emit(event_types.MESSAGE_SENT, insertAt); await reloadCurrentChat(); await eventSource.emit(event_types.USER_MESSAGE_RENDERED, insertAt); } else { chat.push(message); - await saveChatConditional(); + await saveChatConditional(commitReason); const chat_id = (chat.length - 1); await eventSource.emit(event_types.MESSAGE_SENT, chat_id); addOneMessage(message); @@ -7009,6 +7048,18 @@ export async function sendMessageAsUser(messageText, messageBias, insertAt = nul return message; } +export function sendMessageAsUserAtProviderBarrier(messageText, messageBias) { + return sendMessageAsUser( + messageText, + messageBias, + null, + false, + name1, + user_avatar, + CHAT_COMMIT_REASON.PROVIDER_BARRIER, + ); +} + /** * Gets the maximum context token limit (the full context window size before subtracting response length). * @returns {number} The maximum context token limit for the current API. @@ -8555,6 +8606,7 @@ export function markWindowedChatDirtyFromIndex(messageId) { * @param {number} [options.mesId] The message ID to save the chat up to * @param {boolean} [options.force] Force the saving despite the integrity check result * @param {ChatMessage[]} [options.chatData] Chat snapshot to save instead of the current in-memory chat + * @param {string} [options.commitReason] Current/history scheduling reason. * * @returns {Promise} */ @@ -8567,7 +8619,7 @@ export async function saveChat(...args) { return enqueueChatSave(() => saveChatUnsafe(...args)); } -async function saveChatUnsafe({ chatName, withMetadata, mesId, force = false, chatData = undefined } = {}) { +async function saveChatUnsafe({ chatName, withMetadata, mesId, force = false, chatData = undefined, commitReason = CHAT_COMMIT_REASON.MUTATION } = {}) { if (arguments.length > 0 && typeof arguments[0] !== 'object') { console.trace('saveChat called with positional arguments. Please use an object instead.'); [chatName, withMetadata, mesId, force] = arguments; @@ -8625,6 +8677,7 @@ async function saveChatUnsafe({ chatName, withMetadata, mesId, force = false, ch patch, expectedWindowLineCount, force: Boolean(force), + commitReason, }); const activeWindowState = getWindowedChatState(); @@ -8650,6 +8703,7 @@ async function saveChatUnsafe({ chatName, withMetadata, mesId, force = false, ch fileName, payload, force: Boolean(force), + commitReason, }); } return; @@ -8665,6 +8719,7 @@ async function saveChatUnsafe({ chatName, withMetadata, mesId, force = false, ch chat: payload, avatar_url: characters[this_chid].avatar, force: force, + commit_reason: commitReason, }), }); const result = await fetch('/api/chats/save', saveChatRequest); @@ -8708,7 +8763,7 @@ async function saveChatUnsafe({ chatName, withMetadata, mesId, force = false, ch return; } - await saveChatUnsafe({ chatName, withMetadata, mesId, force: true }); + await saveChatUnsafe({ chatName, withMetadata, mesId, force: true, chatData, commitReason }); } } @@ -8993,7 +9048,7 @@ async function getChatResult({ allowNewChat = false } = {}) { freshChat = true; } // Make sure the chat appears on the server - await saveChatConditional(); + await saveChatConditional(CHAT_COMMIT_REASON.MAINTENANCE); } await loadItemizedPrompts(getCurrentChatId()); await printMessages({ frontendSourceHandoffEvent: event_types.CHAT_LOADED }); @@ -10859,13 +10914,13 @@ export async function saveMetadata() { return await saveChatConditional(); } -export async function saveChatConditional() { +export async function saveChatConditional(commitReason = CHAT_COMMIT_REASON.MUTATION) { try { cancelDebouncedChatSave(); const savePromise = selected_group - ? saveGroupChat(selected_group, true) - : saveChat(); + ? saveGroupChat(selected_group, true, false, commitReason) + : saveChat({ commitReason }); // Keep prompt/token persistence serialized with chat writes to avoid // chat switches corrupting per-chat IndexedDB state. diff --git a/src/scripts/chat-payload-transport.js b/src/scripts/chat-payload-transport.js index 4b9d30de..2cba3af0 100644 --- a/src/scripts/chat-payload-transport.js +++ b/src/scripts/chat-payload-transport.js @@ -1,5 +1,6 @@ export { payloadToJsonl, jsonlToPayload } from './tauri/chat/jsonl.js'; export { + CHAT_COMMIT_REASON, isTauriChatPayloadTransportEnabled, normalizeChatFileName, resolveCharacterDirectoryId, diff --git a/src/scripts/group-chats.js b/src/scripts/group-chats.js index f8d3f934..a247b3ea 100644 --- a/src/scripts/group-chats.js +++ b/src/scripts/group-chats.js @@ -88,6 +88,7 @@ import { POPUP_TYPE, Popup, callGenericPopup } from './popup.js'; import { t } from './i18n.js'; import { accountStorage } from './util/AccountStorage.js'; import { + CHAT_COMMIT_REASON, isTauriChatPayloadTransportEnabled, loadGroupChatPayloadTail, saveGroupChatPayload, @@ -395,7 +396,7 @@ export async function getGroupChat(groupId, reload = false, { allowNewChat = fal addOneMessage(mes); await eventSource.emit(event_types.CHARACTER_MESSAGE_RENDERED, (chat.length - 1), 'first_message'); } - await saveGroupChat(groupId, false); + await saveGroupChat(groupId, false, false, CHAT_COMMIT_REASON.MAINTENANCE); } else if (Array.isArray(data) && data.length) { if (!isStillActive()) { return; @@ -731,13 +732,14 @@ function resetSelectedGroup() { * @param {string} groupId Group ID * @param {boolean} shouldSaveGroup Whether to save the group after saving the chat * @param {boolean} force Force the saving on integrity error + * @param {string} commitReason Current/history scheduling reason * @returns {Promise} A promise that resolves when the group chat has been saved. */ -async function saveGroupChat(groupId, shouldSaveGroup, force = false) { - return enqueueChatSave(() => saveGroupChatUnsafe(groupId, shouldSaveGroup, force)); +async function saveGroupChat(groupId, shouldSaveGroup, force = false, commitReason = CHAT_COMMIT_REASON.MUTATION) { + return enqueueChatSave(() => saveGroupChatUnsafe(groupId, shouldSaveGroup, force, commitReason)); } -async function saveGroupChatUnsafe(groupId, shouldSaveGroup, force = false) { +async function saveGroupChatUnsafe(groupId, shouldSaveGroup, force = false, commitReason = CHAT_COMMIT_REASON.MUTATION) { const group = groups.find(x => x.id == groupId); if (!group) { console.warn('Group not found', groupId); @@ -778,6 +780,7 @@ async function saveGroupChatUnsafe(groupId, shouldSaveGroup, force = false) { patch, expectedWindowLineCount, force: Boolean(force), + commitReason, }); const activeWindowState = getWindowedChatState(); @@ -800,13 +803,14 @@ async function saveGroupChatUnsafe(groupId, shouldSaveGroup, force = false) { id: chatId, payload, force: Boolean(force), + commitReason, }); } } else { const response = await fetch('/api/chats/group/save', { method: 'POST', headers: getRequestHeaders(), - body: JSON.stringify({ id: chatId, chat: payload, force: force }), + body: JSON.stringify({ id: chatId, chat: payload, force: force, commit_reason: commitReason }), }); if (response.ok) { @@ -849,7 +853,7 @@ async function saveGroupChatUnsafe(groupId, shouldSaveGroup, force = false) { return; } - await saveGroupChatUnsafe(groupId, shouldSaveGroup, true); + await saveGroupChatUnsafe(groupId, shouldSaveGroup, true, commitReason); } if (shouldSaveGroup) { diff --git a/src/scripts/tauri/chat/transport.js b/src/scripts/tauri/chat/transport.js index bee7c61c..42acc602 100644 --- a/src/scripts/tauri/chat/transport.js +++ b/src/scripts/tauri/chat/transport.js @@ -11,6 +11,13 @@ import { getChatHistoryBootstrapModeName, } from '../../../tauri/main/services/chat-history/chat-history-mode-state.js'; +export const CHAT_COMMIT_REASON = Object.freeze({ + MUTATION: 'mutation', + PROVIDER_BARRIER: 'providerBarrier', + GENERATION_CHECKPOINT: 'generationCheckpoint', + MAINTENANCE: 'maintenance', +}); + export function normalizeChatFileName(fileName) { return stripJsonl(fileName); } @@ -157,7 +164,7 @@ export async function loadCharacterChatPayloadBeforePages({ characterName, avata }); } -export async function saveCharacterChatPayload({ characterName, avatarUrl, fileName, payload, force = false }) { +export async function saveCharacterChatPayload({ characterName, avatarUrl, fileName, payload, force = false, commitReason = CHAT_COMMIT_REASON.MUTATION }) { const normalizedCharacter = resolveCharacterDirectoryId(characterName, avatarUrl); const normalizedFile = normalizeChatFileName(fileName); if (!Array.isArray(payload) || payload.length === 0 || !normalizedCharacter || !normalizedFile.trim()) { @@ -170,6 +177,7 @@ export async function saveCharacterChatPayload({ characterName, avatarUrl, fileN file_name: normalizedFile, file_path: filePath, force, + commit_reason: commitReason, }, })); } @@ -182,7 +190,7 @@ function normalizeExpectedWindowLineCount(value) { return normalized; } -export async function patchCharacterChatPayloadWindowed({ characterName, avatarUrl, fileName, cursor, header, patch, expectedWindowLineCount, force = false }) { +export async function patchCharacterChatPayloadWindowed({ characterName, avatarUrl, fileName, cursor, header, patch, expectedWindowLineCount, force = false, commitReason = CHAT_COMMIT_REASON.MUTATION }) { const normalizedCharacter = resolveCharacterDirectoryId(characterName, avatarUrl); const normalizedFile = normalizeChatFileName(fileName); if (!normalizedCharacter || !normalizedFile.trim()) { @@ -198,6 +206,7 @@ export async function patchCharacterChatPayloadWindowed({ characterName, avatarU patch, expected_window_line_count: normalizeExpectedWindowLineCount(expectedWindowLineCount), force, + commit_reason: commitReason, }, }); } @@ -305,7 +314,7 @@ export async function loadGroupChatPayloadBeforePages({ id, cursor, maxLines, ma }); } -export async function saveGroupChatPayload({ id, payload, force = false }) { +export async function saveGroupChatPayload({ id, payload, force = false, commitReason = CHAT_COMMIT_REASON.MUTATION }) { const normalizedId = normalizeChatFileName(id); if (!Array.isArray(payload) || payload.length === 0 || !normalizedId.trim()) { throw new Error('Invalid group chat payload'); @@ -316,11 +325,12 @@ export async function saveGroupChatPayload({ id, payload, force = false }) { id: normalizedId, file_path: filePath, force, + commit_reason: commitReason, }, })); } -export async function patchGroupChatPayloadWindowed({ id, cursor, header, patch, expectedWindowLineCount, force = false }) { +export async function patchGroupChatPayloadWindowed({ id, cursor, header, patch, expectedWindowLineCount, force = false, commitReason = CHAT_COMMIT_REASON.MUTATION }) { const normalizedId = normalizeChatFileName(id); if (!normalizedId.trim()) { throw new Error('Invalid group chat payload patch request'); @@ -334,6 +344,7 @@ export async function patchGroupChatPayloadWindowed({ id, cursor, header, patch, patch, expected_window_line_count: normalizeExpectedWindowLineCount(expectedWindowLineCount), force, + commit_reason: commitReason, }, }); } diff --git a/src/tauri/main/api/agent-chat-commit-bridge.js b/src/tauri/main/api/agent-chat-commit-bridge.js index 61f25fa9..26c5d445 100644 --- a/src/tauri/main/api/agent-chat-commit-bridge.js +++ b/src/tauri/main/api/agent-chat-commit-bridge.js @@ -1,6 +1,7 @@ // @ts-check import { resolveStableChatId } from './agent-chat-identity.js'; +import { CHAT_COMMIT_REASON } from '../../../scripts/chat-payload-transport.js'; const activeCommitBridges = new Map(); const TERMINAL_EVENTS = new Set(['run_completed', 'run_partial_success', 'run_cancelled', 'run_failed']); @@ -149,7 +150,7 @@ async function handleChatCommitRequested({ state, event, safeInvoke, readWorkspa createdMessage: state.createdMessage, firstSwipeId: state.firstSwipeId, }); - await state.persistChat(script); + await state.persistChat(script, CHAT_COMMIT_REASON.GENERATION_CHECKPOINT); await safeInvoke('resolve_agent_chat_commit', { dto: { @@ -183,7 +184,7 @@ async function handlePersistentStateMetadataUpdateRequested({ state, event, safe const messageId = normalizeMessageId(payload.messageId ?? state.messageId); const stateId = requirePayloadString(payload, 'stateId'); mergePersistentStateExtraIntoMessage(script.chat, messageId, payload, stateId); - await state.persistChat(script); + await state.persistChat(script, CHAT_COMMIT_REASON.MUTATION); await safeInvoke('resolve_agent_persistent_state_metadata_update', { dto: { @@ -438,18 +439,23 @@ function isPlainObject(value) { return Boolean(value) && typeof value === 'object' && !Array.isArray(value); } -async function persistActiveChat(script) { +async function persistActiveChat(script, commitReason) { const groupChats = await import('../../../scripts/group-chats.js'); if (groupChats.selected_group) { if (typeof groupChats.saveGroupChat !== 'function') { throw new Error('saveGroupChat is not available'); } - await groupChats.saveGroupChat(groupChats.selected_group, true); + await groupChats.saveGroupChat( + groupChats.selected_group, + true, + false, + commitReason, + ); return; } if (typeof script.saveChat !== 'function') { throw new Error('saveChat is not available'); } - await script.saveChat(); + await script.saveChat({ commitReason }); } diff --git a/src/tauri/main/routes/chat-routes.js b/src/tauri/main/routes/chat-routes.js index 5e29e7f9..9c617eca 100644 --- a/src/tauri/main/routes/chat-routes.js +++ b/src/tauri/main/routes/chat-routes.js @@ -77,7 +77,6 @@ export function registerChatRoutes(router, context, { jsonResponse }) { if (!characterId || !fileName.trim() || !Array.isArray(body?.chat)) { return jsonResponse({ error: 'Invalid chat payload' }, 400); } - try { await saveCharacterChatPayload({ characterName: characterId, @@ -85,6 +84,7 @@ export function registerChatRoutes(router, context, { jsonResponse }) { fileName, payload: body.chat, force: Boolean(body?.force), + commitReason: body?.commit_reason, }); return jsonResponse({ ok: true }); } catch (error) { @@ -431,6 +431,7 @@ export function registerChatRoutes(router, context, { jsonResponse }) { id, payload: body.chat, force: Boolean(body?.force), + commitReason: body?.commit_reason, }); return jsonResponse({ ok: true }); } catch (error) { diff --git a/tests/active-chat-ref-contract.test.mjs b/tests/active-chat-ref-contract.test.mjs index 68f4b5c3..de99c1ff 100644 --- a/tests/active-chat-ref-contract.test.mjs +++ b/tests/active-chat-ref-contract.test.mjs @@ -52,6 +52,32 @@ test('active chat ref resolves character identity from exact avatar filename', a }); }); +test('active chat ref resolves a group locator from the committed chat id', async () => { + const { getActiveChatSnapshot } = await importFresh( + path.join(REPO_ROOT, 'src/tauri/main/adapters/st/active-chat-ref.js'), + ); + + const snapshot = withWindow({ + SillyTavern: { + getContext: () => ({ + chat: [{ mes: 'hello' }, { mes: 'world' }], + chatId: 'party-session.jsonl', + groupId: 'party', + characters: [], + characterId: null, + }), + }, + }, () => getActiveChatSnapshot()); + + assert.deepEqual(snapshot, { + ref: { + kind: 'group', + chatId: 'party-session', + }, + windowLength: 2, + }); +}); + test('active chat ref rejects URL-like active character avatars', async () => { const { getActiveChatSnapshot } = await importFresh( path.join(REPO_ROOT, 'src/tauri/main/adapters/st/active-chat-ref.js'), diff --git a/tests/generation-lifecycle-contract.test.mjs b/tests/generation-lifecycle-contract.test.mjs index 7593f110..985697f9 100644 --- a/tests/generation-lifecycle-contract.test.mjs +++ b/tests/generation-lifecycle-contract.test.mjs @@ -11,10 +11,11 @@ test('Generate wrapper uses an in-flight lifecycle gate around GenerateInternal( assert.match(source, /const generationIdleGate = createGenerationIdleGate\(\);/); assert.match(source, /let generationInFlightCount = 0;/); + assert.match(source, /let generationHistoryLocator = null;/); assert.match(source, /export function waitForGenerationIdle\(\)\s*{\s*return generationIdleGate\.wait\(\);\s*}/s); - assert.match(source, /function enterGeneration\(\)\s*{\s*if \(generationInFlightCount === 0\) {\s*generationIdleGate\.markBusy\(\);\s*}\s*generationInFlightCount \+= 1;\s*}/s); - assert.match(source, /function exitGeneration\(\)\s*{\s*if \(generationInFlightCount <= 0\) {\s*throw new Error\('Generation in-flight counter underflow'\);\s*}\s*generationInFlightCount -= 1;\s*if \(generationInFlightCount === 0\) {\s*generationIdleGate\.markIdle\(\);\s*}\s*}/s); - assert.match(source, /export async function Generate\(type, options = \{\}, dryRun = false\)\s*{\s*enterGeneration\(\);\s*try {\s*return await GenerateInternal\(type, options, dryRun\);\s*} catch \(error\) {\s*cleanupGenerationAfterUnhandledError\(type, dryRun\);\s*throw error;\s*} finally {\s*exitGeneration\(\);\s*}\s*}/s); + assert.match(source, /async function enterGeneration\(dryRun\)[\s\S]*generationInFlightCount = 1;[\s\S]*generationHistoryLocator = null;[\s\S]*if \(dryRun\) return;[\s\S]*getActiveChatSnapshot\(\)\.ref;[\s\S]*invoke\('chat_history_generation_started', \{ locator: generationHistoryLocator \}\);/); + assert.match(source, /async function exitGeneration\(\)[\s\S]*const locator = generationHistoryLocator;[\s\S]*generationHistoryLocator = null;[\s\S]*invoke\('chat_history_generation_finished', \{ locator \}\);[\s\S]*finally \{\s*generationIdleGate\.markIdle\(\);/s); + assert.match(source, /export async function Generate\(type, options = \{\}, dryRun = false\)\s*{\s*await enterGeneration\(dryRun\);\s*try {\s*return await GenerateInternal\(type, options, dryRun\);\s*} catch \(error\) {\s*cleanupGenerationAfterUnhandledError\(type, dryRun\);\s*throw error;\s*} finally {\s*await exitGeneration\(\);\s*}\s*}/s); assert.match(source, /async function GenerateInternal\(/); }); @@ -26,6 +27,34 @@ test('Unhandled foreground Generate errors reuse the legacy unblock path', async assert.match(source, /cleanupGenerationAfterUnhandledError[\s\S]*unblockGeneration\(type\);/); }); +test('stable chat-history reasons cover provider, final, transport, and agent commits', async () => { + const [script, groups, transport, routes, agentBridge] = await Promise.all([ + readFile(path.join(REPO_ROOT, 'src/script.js'), 'utf8'), + readFile(path.join(REPO_ROOT, 'src/scripts/group-chats.js'), 'utf8'), + readFile(path.join(REPO_ROOT, 'src/scripts/tauri/chat/transport.js'), 'utf8'), + readFile(path.join(REPO_ROOT, 'src/tauri/main/routes/chat-routes.js'), 'utf8'), + readFile(path.join(REPO_ROOT, 'src/tauri/main/api/agent-chat-commit-bridge.js'), 'utf8'), + ]); + + assert.match(script, /sendMessageAsUserAtProviderBarrier\(textareaText, messageBias\)/); + assert.match(script, /sendMessageAsUserAtProviderBarrier\(oai_settings\.send_if_empty\.trim\(\), messageBias\)/); + assert.match(script, /saveChatConditional\(this\.type === 'impersonate'[\s\S]*CHAT_COMMIT_REASON\.MUTATION[\s\S]*CHAT_COMMIT_REASON\.GENERATION_CHECKPOINT\)/); + assert.match(script, /saveChatConditional\(isImpersonate[\s\S]*CHAT_COMMIT_REASON\.MUTATION[\s\S]*CHAT_COMMIT_REASON\.GENERATION_CHECKPOINT\)/); + assert.match(script, /saveChatUnsafe\(\{ chatName, withMetadata, mesId, force: true, chatData, commitReason \}\)/); + assert.match(groups, /saveGroupChatUnsafe\(groupId, shouldSaveGroup, true, commitReason\)/); + assert.match(groups, /sendMessageAsUser\(userInput, bias\.messageBias\)/); + assert.match(script, /saveChatConditional\(CHAT_COMMIT_REASON\.MAINTENANCE\)/); + assert.match(groups, /saveGroupChat\(groupId, false, false, CHAT_COMMIT_REASON\.MAINTENANCE\)/); + assert.match(transport, /MAINTENANCE: 'maintenance'/); + + assert.match(transport, /commit_reason: commitReason/g); + assert.equal(transport.match(/commit_reason: commitReason/g)?.length, 4); + assert.equal(routes.match(/commitReason: body\?\.commit_reason/g)?.length, 2); + assert.match(agentBridge, /handleChatCommitRequested[\s\S]*state\.persistChat\(script, CHAT_COMMIT_REASON\.GENERATION_CHECKPOINT\)/); + assert.match(agentBridge, /handlePersistentStateMetadataUpdateRequested[\s\S]*state\.persistChat\(script, CHAT_COMMIT_REASON\.MUTATION\)/); + assert.match(agentBridge, /saveGroupChat\([\s\S]*commitReason,[\s\S]*script\.saveChat\(\{ commitReason \}\)/); +}); + test('Unhandled error cleanup only targets foreground UI lifecycle leaks', async () => { const { shouldUnblockGenerationAfterUnhandledError } = await import('../src/scripts/util/generation-lifecycle.js'); diff --git a/tests/itemized-prompts-contract.test.mjs b/tests/itemized-prompts-contract.test.mjs index f65ba68e..f20b7451 100644 --- a/tests/itemized-prompts-contract.test.mjs +++ b/tests/itemized-prompts-contract.test.mjs @@ -70,7 +70,7 @@ test('Chat rendering and generation rely on index presence, not whole records', const source = await readFile(path.join(REPO_ROOT, 'src/script.js'), 'utf8'); const saveConditionalFn = extractBetween( source, - 'export async function saveChatConditional() {', + 'export async function saveChatConditional(commitReason = CHAT_COMMIT_REASON.MUTATION) {', '/**\n * Saves the chat to the server.', );