Persist and sync the message send-later queue across restarts and clients

src-tauri/src/commands/message_queue.rs

@@ -0,0 +1,42 @@
+#[cfg(feature = "tauri-runtime")]
+use crate::db::AppDatabase;
+#[cfg(feature = "tauri-runtime")]
+use crate::db::error::DbError;
+#[cfg(feature = "tauri-runtime")]
+use crate::db::service::message_queue_service::{self, QueueItemInput};
+#[cfg(feature = "tauri-runtime")]
+use crate::models::MessageQueueItem;
+#[cfg(feature = "tauri-runtime")]
+use crate::web::event_bridge::{emit_event, EventEmitter, MessageQueueChange, MESSAGE_QUEUE_CHANGED_EVENT};
+
+#[cfg(feature = "tauri-runtime")]
+#[cfg_attr(feature = "tauri-runtime", tauri::command)]
+pub async fn message_queue_list(
+    db: tauri::State<'_, AppDatabase>,
+    owner_id: String,
+) -> Result<Vec<MessageQueueItem>, DbError> {
+    message_queue_service::list(&db.conn, &owner_id).await
+}
+
+#[cfg(feature = "tauri-runtime")]
+#[cfg_attr(feature = "tauri-runtime", tauri::command)]
+pub async fn message_queue_save(
+    db: tauri::State<'_, AppDatabase>,
+    app: tauri::AppHandle,
+    owner_id: String,
+    items: Vec<QueueItemInput>,
+    origin: String,
+) -> Result<(), DbError> {
+    let saved = message_queue_service::save(&db.conn, &owner_id, items).await?;
+    let emitter = EventEmitter::Tauri(app);
+    emit_event(
+        &emitter,
+        MESSAGE_QUEUE_CHANGED_EVENT,
+        MessageQueueChange {
+            owner_id,
+            origin,
+            items: saved,
+        },
+    );
+    Ok(())
+}

src-tauri/src/commands/mod.rs

@@ -68,6 +68,7 @@ pub mod notification;
 pub mod pet;
 pub mod project_boot;
 pub mod question;
+pub mod message_queue;
 pub mod quick_messages;
 #[cfg(feature = "tauri-runtime")]
 pub mod remote_proxy;

src-tauri/src/db/entities/message_queue_item.rs

@@ -0,0 +1,28 @@
+use sea_orm::entity::prelude::*;
+
+/// One row per queued message in a conversation's "send later" queue.
+///
+/// The queue is keyed by `owner_id` (a conversation id). `item_uid` is a
+/// client-generated UUID that is stable across edits, so reordering and updates
+/// can target the same logical item without a server round-trip for ids. The
+/// `draft` is stored as JSON because `PromptDraft` is a frontend-only type.
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel)]
+#[sea_orm(table_name = "message_queue_item")]
+pub struct Model {
+    #[sea_orm(primary_key)]
+    pub id: i32,
+    pub owner_id: String,
+    pub item_uid: String,
+    #[sea_orm(column_type = "Text")]
+    pub draft_json: String,
+    pub mode_id: Option<String>,
+    pub adopt_send_time_mode: bool,
+    pub sort_order: i32,
+    pub created_at: DateTimeUtc,
+    pub updated_at: DateTimeUtc,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {}
+
+impl ActiveModelBehavior for ActiveModel {}

src-tauri/src/db/entities/mod.rs

@@ -20,6 +20,7 @@ pub mod folder_link;
 pub mod model_provider;
 pub mod opened_tab;
 pub mod prelude;
+pub mod message_queue_item;
 pub mod quick_message;
 pub mod remote_workspace_connection;
 pub mod token_usage_sync;

src-tauri/src/db/entities/prelude.rs

@@ -21,6 +21,7 @@ pub use super::folder_group::Entity as FolderGroup;
 pub use super::folder_link::Entity as FolderLink;
 pub use super::model_provider::Entity as ModelProvider;
 pub use super::opened_tab::Entity as OpenedTab;
+pub use super::message_queue_item::Entity as MessageQueueItem;
 pub use super::quick_message::Entity as QuickMessage;
 pub use super::token_usage_sync::Entity as TokenUsageSync;
 pub use super::token_usage_turn::Entity as TokenUsageTurn;

src-tauri/src/db/migration/m20261030_000001_message_queue.rs

@@ -0,0 +1,106 @@
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        // message_queue_item: the per-conversation "send later" queue, persisted
+        // so it survives an app restart and — via the `message-queue://changed`
+        // broadcast — stays in sync across every open window / web client.
+        //
+        // One row per queued message. `item_uid` is a client-generated UUID that
+        // is stable across edits; `owner_id` is the conversation the queue belongs
+        // to. The `draft` is JSON because `PromptDraft` is a frontend-only type.
+        manager
+            .create_table(
+                Table::create()
+                    .table(MessageQueueItem::Table)
+                    .if_not_exists()
+                    .col(
+                        ColumnDef::new(MessageQueueItem::Id)
+                            .integer()
+                            .not_null()
+                            .auto_increment()
+                            .primary_key(),
+                    )
+                    .col(ColumnDef::new(MessageQueueItem::OwnerId).string().not_null())
+                    .col(ColumnDef::new(MessageQueueItem::ItemUid).string().not_null())
+                    .col(ColumnDef::new(MessageQueueItem::DraftJson).text().not_null())
+                    .col(ColumnDef::new(MessageQueueItem::ModeId).string().null())
+                    .col(
+                        ColumnDef::new(MessageQueueItem::AdoptSendTimeMode)
+                            .boolean()
+                            .not_null()
+                            .default(false),
+                    )
+                    .col(
+                        ColumnDef::new(MessageQueueItem::SortOrder)
+                            .integer()
+                            .not_null()
+                            .default(0),
+                    )
+                    .col(
+                        ColumnDef::new(MessageQueueItem::CreatedAt)
+                            .timestamp_with_time_zone()
+                            .not_null(),
+                    )
+                    .col(
+                        ColumnDef::new(MessageQueueItem::UpdatedAt)
+                            .timestamp_with_time_zone()
+                            .not_null(),
+                    )
+                    .to_owned(),
+            )
+            .await?;
+
+        // "All items for a conversation", read on every queue load.
+        manager
+            .create_index(
+                Index::create()
+                    .if_not_exists()
+                    .name("idx_message_queue_item_owner")
+                    .table(MessageQueueItem::Table)
+                    .col(MessageQueueItem::OwnerId)
+                    .to_owned(),
+            )
+            .await?;
+
+        // The upsert / delete conflict target.
+        manager
+            .create_index(
+                Index::create()
+                    .if_not_exists()
+                    .name("idx_message_queue_item_owner_uid")
+                    .table(MessageQueueItem::Table)
+                    .col(MessageQueueItem::OwnerId)
+                    .col(MessageQueueItem::ItemUid)
+                    .unique()
+                    .to_owned(),
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .drop_table(Table::drop().table(MessageQueueItem::Table).to_owned())
+            .await
+    }
+}
+
+#[derive(DeriveIden)]
+enum MessageQueueItem {
+    Table,
+    Id,
+    OwnerId,
+    ItemUid,
+    DraftJson,
+    ModeId,
+    AdoptSendTimeMode,
+    SortOrder,
+    CreatedAt,
+    UpdatedAt,
+}

src-tauri/src/db/migration/mod.rs

@@ -56,6 +56,7 @@ mod m20261017_000002_conversation_source;
 mod m20261020_000001_folder_command_run_local;
 mod m20261022_000001_tool_approval;
 mod m20261022_000001_work_task_delivery;
+mod m20261030_000001_message_queue;
 pub struct Migrator;
 
 #[async_trait::async_trait]
@@ -118,6 +119,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20261020_000001_folder_command_run_local::Migration),
             Box::new(m20261022_000001_tool_approval::Migration),
             Box::new(m20261022_000001_work_task_delivery::Migration),
+            Box::new(m20261030_000001_message_queue::Migration),
         ]
     }
 }

src-tauri/src/db/service/message_queue_service.rs

@@ -0,0 +1,81 @@
+use chrono::Utc;
+use sea_orm::{
+    ActiveModelTrait, ActiveValue::NotSet, ColumnTrait, DatabaseConnection, EntityTrait,
+    QueryFilter, QueryOrder, Set, TransactionTrait,
+};
+use serde::Deserialize;
+
+use crate::db::entities::message_queue_item;
+use crate::db::error::DbError;
+use crate::models::MessageQueueItem;
+
+/// A queue item as supplied by the frontend when saving. Field names are
+/// camelCase on the wire (both the Tauri command and the axum web handler
+/// expect it), so serde rename handles the mapping to the Rust fields.
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct QueueItemInput {
+    pub id: String,
+    pub draft_json: String,
+    pub mode_id: Option<String>,
+    pub adopt_send_time_mode: bool,
+}
+
+fn to_item(m: message_queue_item::Model) -> MessageQueueItem {
+    MessageQueueItem {
+        id: m.item_uid,
+        draft_json: m.draft_json,
+        mode_id: m.mode_id,
+        adopt_send_time_mode: m.adopt_send_time_mode,
+    }
+}
+
+pub async fn list(
+    conn: &DatabaseConnection,
+    owner_id: &str,
+) -> Result<Vec<MessageQueueItem>, DbError> {
+    let rows = message_queue_item::Entity::find()
+        .filter(message_queue_item::Column::OwnerId.eq(owner_id))
+        .order_by_asc(message_queue_item::Column::SortOrder)
+        .all(conn)
+        .await?;
+    Ok(rows.into_iter().map(to_item).collect())
+}
+
+/// Replace the entire queue for `owner_id` with `items` (in order). Simple and
+/// idempotent: last-writer-wins, and the broadcast converges every client to the
+/// same list. Multi-end sync is the goal, so the frontend is the source of
+/// truth for ordering and the backend just persists the canonical list. The
+/// whole replace runs in one transaction so a concurrent reader never sees a
+/// half-written queue.
+pub async fn save(
+    conn: &DatabaseConnection,
+    owner_id: &str,
+    items: Vec<QueueItemInput>,
+) -> Result<Vec<MessageQueueItem>, DbError> {
+    let txn = conn.begin().await?;
+    message_queue_item::Entity::delete_many()
+        .filter(message_queue_item::Column::OwnerId.eq(owner_id))
+        .exec(&txn)
+        .await?;
+
+    let mut saved = Vec::with_capacity(items.len());
+    for (idx, item) in items.into_iter().enumerate() {
+        let now = Utc::now();
+        let active = message_queue_item::ActiveModel {
+            id: NotSet,
+            owner_id: Set(owner_id.to_string()),
+            item_uid: Set(item.id),
+            draft_json: Set(item.draft_json),
+            mode_id: Set(item.mode_id),
+            adopt_send_time_mode: Set(item.adopt_send_time_mode),
+            sort_order: Set(idx as i32),
+            created_at: Set(now),
+            updated_at: Set(now),
+        };
+        let m = active.insert(&txn).await?;
+        saved.push(to_item(m));
+    }
+    txn.commit().await?;
+    Ok(saved)
+}

src-tauri/src/db/service/mod.rs

@@ -14,6 +14,7 @@ pub mod folder_link_service;
 pub mod folder_service;
 pub mod import_service;
 pub mod model_provider_service;
+pub mod message_queue_service;
 pub mod quick_message_service;
 pub mod remote_workspace_connection_service;
 pub mod sender_context_service;

src-tauri/src/lib.rs

@@ -117,6 +117,7 @@ mod tauri_app {
         folders, logging as logging_commands, mcp as mcp_commands,
         model_provider as model_provider_commands, notification, pet as pet_commands, project_boot,
         question as question_commands, quick_messages as quick_messages_commands,
+        message_queue as message_queue_commands,
         remote_proxy as remote_proxy_commands,
         remote_workspace as remote_workspace_commands, science as science_commands,
         session_info as session_info_commands,
@@ -2282,6 +2283,8 @@ mod tauri_app {
                 quick_messages_commands::quick_messages_update,
                 quick_messages_commands::quick_messages_delete,
                 quick_messages_commands::quick_messages_reorder,
+                message_queue_commands::message_queue_list,
+                message_queue_commands::message_queue_save,
                 automation_commands::automation_list,
                 automation_commands::automation_get,
                 automation_commands::automation_runs,

src-tauri/src/models/message_queue.rs

@@ -0,0 +1,14 @@
+use serde::Serialize;
+
+/// A single persisted message-queue item, returned to the frontend. The `draft`
+/// travels as a JSON string (`draft_json`) because `PromptDraft` is a
+/// frontend-only type; the client re-parses it. Field names are snake_case to
+/// match how the rest of the codebase serializes command returns (no camelCase
+/// conversion happens on the wire for event/return payloads).
+#[derive(Debug, Clone, Serialize)]
+pub struct MessageQueueItem {
+    pub id: String,
+    pub draft_json: String,
+    pub mode_id: Option<String>,
+    pub adopt_send_time_mode: bool,
+}

src-tauri/src/models/mod.rs

@@ -9,6 +9,7 @@ pub mod folder;
 pub mod message;
 pub mod model_provider;
 pub mod pet;
+pub mod message_queue;
 pub mod quick_message;
 pub mod remote_workspace_connection;
 pub mod system;
@@ -42,6 +43,7 @@ pub use message::{
     AgentExecutionStats, AgentTimelineEntry, AgentToolCall, ContentBlock, ImageData, MessageRole,
     MessageTurn, TurnRole, TurnUsage, UnifiedMessage,
 };
+pub use message_queue::MessageQueueItem;
 pub use quick_message::QuickMessageInfo;
 pub use remote_workspace_connection::{
     RemoteWorkspaceConnectionInfo, RemoteWorkspaceHeader, ToHeaderMap,

src-tauri/src/web/event_bridge.rs

@@ -394,6 +394,13 @@ pub struct ConversationsBulkChanged {
 /// webview and every WebSocket client.
 pub const TABS_CHANGED_EVENT: &str = "tabs://changed";
 
+/// Global side-channel announcing a message-queue change for a given owner
+/// (conversation). Mirrors [`CONVERSATION_CHANGED_EVENT`]: a single
+/// [`emit_event`] reaches the Tauri webview and every WebSocket client, so the
+/// "send later" queue — which is persisted in the DB and edited from any window
+/// or web client — converges across all of them. Payload: [`MessageQueueChange`].
+pub const MESSAGE_QUEUE_CHANGED_EVENT: &str = "message-queue://changed";
+
 /// Payload for the [`TABS_CHANGED_EVENT`] side-channel. Carries the full
 /// conversation-bound tab set (a snapshot, not a delta) so every client
 /// converges idempotently — matching the full-replacement save semantics.
@@ -412,6 +419,19 @@ pub struct TabsChanged {
     pub tabs: Vec<crate::models::OpenedTab>,
 }
 
+/// Payload for [`MESSAGE_QUEUE_CHANGED_EVENT`]. Carries the full queue for the
+/// owner so every client replaces its local copy idempotently (last-writer-wins).
+///
+/// `origin` is the client id that made the change; that client ignores its own
+/// echo so a concurrent local mutation is never clobbered by the broadcast it
+/// triggered.
+#[derive(Debug, Clone, Serialize)]
+pub struct MessageQueueChange {
+    pub owner_id: String,
+    pub origin: String,
+    pub items: Vec<crate::models::MessageQueueItem>,
+}
+
 /// Global side-channel for cross-client automation list + run sync. Mirrors
 /// [`CONVERSATION_CHANGED_EVENT`]: a single [`emit_event`] reaches the Tauri
 /// webview and every WebSocket client. The scheduler runs headless (no window),