Merge pull request #65 from altaidevorg/feat/altai-scoped-inbox-queries

feat(memory): scope inbox list queries by channel

Author: monatis

src/agent/mod.rs

@@ -2170,6 +2170,11 @@ impl AgentLogic {
         let _ = memory_node
             .send_packet(MemoryMessage::ListBackgroundJobs {
                 chat_id: Some(chat_id.to_string()),
+                // A background job may have been created through a different
+                // channel for this same chat. Native recovery deliberately
+                // remains chat-scoped; host embedders opt into channel scope
+                // when their UI needs an isolated inbox.
+                channel: None,
                 limit: 10,
                 reply: SharedReply::new(tx),
             })
@@ -2184,6 +2189,7 @@ impl AgentLogic {
                     .send_packet(MemoryMessage::ListClarificationTickets {
                         job_id: Some(job.job_id.clone()),
                         chat_id: Some(chat_id.to_string()),
+                        channel: None,
                         status: Some("waiting".to_string()),
                         limit: 1,
                         reply: SharedReply::new(tx2),

src/channels/api.rs

@@ -1957,6 +1957,8 @@ struct JobsQuery {
     #[serde(default)]
     chat_id: Option<String>,
     #[serde(default)]
+    channel: Option<String>,
+    #[serde(default)]
     limit: Option<usize>,
 }
 
@@ -1965,6 +1967,8 @@ struct NotificationsQuery {
     #[serde(default)]
     chat_id: Option<String>,
     #[serde(default)]
+    channel: Option<String>,
+    #[serde(default)]
     limit: Option<usize>,
     #[serde(default)]
     unseen_only: Option<bool>,
@@ -2373,6 +2377,7 @@ async fn handle_list_background_jobs(
     let res = memory_request(&state.memory_node, |reply| {
         MemoryMessage::ListBackgroundJobs {
             chat_id: params.chat_id,
+            channel: params.channel,
             limit: params.limit.unwrap_or(100),
             reply,
         }
@@ -2392,6 +2397,7 @@ async fn handle_list_notifications(
     let res = memory_request(&state.memory_node, |reply| {
         MemoryMessage::ListNotifications {
             chat_id: params.chat_id,
+            channel: params.channel,
             limit: params.limit.unwrap_or(100),
             unseen_only: params.unseen_only.unwrap_or(false),
             reply,

src/channels/terminal_ui/run.rs

@@ -1424,6 +1424,10 @@ pub(crate) fn run_ratatui_main(config: RatatuiMainConfig) -> io::Result<()> {
                         let _ = memory_node
                             .send_packet(crate::memory::MemoryMessage::ListBackgroundJobs {
                                 chat_id: None,
+                                // The terminal's inbox is intentionally
+                                // workspace-global: jobs may originate in a
+                                // different channel and must stay visible.
+                                channel: None,
                                 limit: 50,
                                 reply: crate::memory::SharedReply::new(jtx),
                             })
@@ -1437,6 +1441,7 @@ pub(crate) fn run_ratatui_main(config: RatatuiMainConfig) -> io::Result<()> {
                         let _ = memory_node
                             .send_packet(crate::memory::MemoryMessage::ListNotifications {
                                 chat_id: None,
+                                channel: None,
                                 limit: 50,
                                 unseen_only: false,
                                 reply: crate::memory::SharedReply::new(ntx),

src/main.rs

@@ -1341,6 +1341,7 @@ async fn recover_background_jobs_on_startup(
     if memory_node
         .send_packet(MemoryMessage::ListBackgroundJobs {
             chat_id: None,
+            channel: None,
             limit: 500,
             reply: SharedReply::new(tx),
         })

src/memory.rs

@@ -657,8 +657,11 @@ pub enum MemoryMessage {
         record: BackgroundJobRecord,
         reply: SharedReply<Result<(), String>>,
     },
+    /// List background jobs, optionally scoped to one host channel before the
+    /// result limit is applied.
     ListBackgroundJobs {
         chat_id: Option<String>,
+        channel: Option<String>,
         limit: usize,
         reply: SharedReply<Result<Vec<BackgroundJobRecord>, String>>,
     },
@@ -672,8 +675,11 @@ pub enum MemoryMessage {
         record: NotificationRecord,
         reply: SharedReply<Result<(), String>>,
     },
+    /// List notifications, optionally scoped to one host channel before the
+    /// result limit is applied.
     ListNotifications {
         chat_id: Option<String>,
+        channel: Option<String>,
         limit: usize,
         unseen_only: bool,
         reply: SharedReply<Result<Vec<NotificationRecord>, String>>,
@@ -705,10 +711,12 @@ pub enum MemoryMessage {
         response: String,
         reply: SharedReply<Result<(), String>>,
     },
-    /// List clarification tickets with optional filters.
+    /// List clarification tickets with optional filters. `channel` is applied
+    /// in SQLite before the limit so one host cannot starve another's inbox.
     ListClarificationTickets {
         job_id: Option<String>,
         chat_id: Option<String>,
+        channel: Option<String>,
         status: Option<String>,
         limit: usize,
         reply: SharedReply<Result<Vec<ClarificationTicketRecord>, String>>,
@@ -1869,6 +1877,7 @@ impl ActorLogic<MemoryMessage> for SqliteMemoryActor {
             }
             MemoryMessage::ListBackgroundJobs {
                 chat_id,
+                channel,
                 limit,
                 reply,
             } => {
@@ -1886,6 +1895,10 @@ impl ActorLogic<MemoryMessage> for SqliteMemoryActor {
                         filters.push("chat_id = ?");
                         params_vec.push(Box::new(cid));
                     }
+                    if let Some(ch) = channel {
+                        filters.push("channel = ?");
+                        params_vec.push(Box::new(ch));
+                    }
 
                     if !filters.is_empty() {
                         sql.push_str(" WHERE ");
@@ -1955,6 +1968,7 @@ impl ActorLogic<MemoryMessage> for SqliteMemoryActor {
             }
             MemoryMessage::ListNotifications {
                 chat_id,
+                channel,
                 limit,
                 unseen_only,
                 reply,
@@ -1969,6 +1983,10 @@ impl ActorLogic<MemoryMessage> for SqliteMemoryActor {
                         filters.push("chat_id = ?");
                         params_vec.push(Box::new(cid));
                     }
+                    if let Some(ch) = channel {
+                        filters.push("channel = ?");
+                        params_vec.push(Box::new(ch));
+                    }
                     if unseen_only {
                         filters.push("seen_at_ms IS NULL");
                     }
@@ -2183,6 +2201,7 @@ impl ActorLogic<MemoryMessage> for SqliteMemoryActor {
             MemoryMessage::ListClarificationTickets {
                 job_id,
                 chat_id,
+                channel,
                 status,
                 limit,
                 reply,
@@ -2201,6 +2220,10 @@ impl ActorLogic<MemoryMessage> for SqliteMemoryActor {
                         filters.push("chat_id = ?");
                         params_vec.push(Box::new(cid));
                     }
+                    if let Some(ch) = channel {
+                        filters.push("channel = ?");
+                        params_vec.push(Box::new(ch));
+                    }
                     if let Some(s) = status {
                         filters.push("status = ?");
                         params_vec.push(Box::new(s));
@@ -2298,7 +2321,8 @@ fn dropped_tool_call_ids(
 #[cfg(test)]
 mod root_thread_id_tests {
     use super::{
-        chat_id_from_root_thread_id, is_root_session_thread_id, MemoryMessage, SharedReply,
+        chat_id_from_root_thread_id, is_root_session_thread_id, BackgroundJobRecord,
+        ClarificationTicketRecord, MemoryMessage, NotificationRecord, SharedReply,
         SqliteMemoryActor,
     };
     use crate::session::SessionManager;
@@ -2339,6 +2363,151 @@ mod root_thread_id_tests {
         assert!(!is_root_session_thread_id("tauri", "tauri:s-abc:extra:"));
     }
 
+    #[tokio::test]
+    async fn channel_scoped_inbox_query_filters_before_limit() {
+        let actor = SqliteMemoryActor::new(":memory:").expect("memory actor");
+        let node = NodeHandle::new(actor, 16, 1, Duration::from_millis(1));
+
+        for (id, channel, created_at_ms) in [
+            ("tauri-notification", "tauri", 1_i64),
+            ("terminal-notification", "terminal", 2_i64),
+        ] {
+            let (tx, rx) = tokio::sync::oneshot::channel();
+            node.send_packet(MemoryMessage::InsertNotification {
+                record: NotificationRecord {
+                    notification_id: id.to_string(),
+                    chat_id: "chat-1".to_string(),
+                    channel: channel.to_string(),
+                    thread_id: None,
+                    kind: "test".to_string(),
+                    title: "Test".to_string(),
+                    body: "Test".to_string(),
+                    action_kind: None,
+                    action_payload: None,
+                    seen_at_ms: None,
+                    resolved_at_ms: None,
+                    created_at_ms,
+                },
+                reply: SharedReply::new(tx),
+            })
+            .await
+            .expect("enqueue notification");
+            rx.await
+                .expect("notification actor reply")
+                .expect("insert notification");
+        }
+
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        node.send_packet(MemoryMessage::ListNotifications {
+            chat_id: None,
+            channel: Some("tauri".to_string()),
+            limit: 1,
+            unseen_only: false,
+            reply: SharedReply::new(tx),
+        })
+        .await
+        .expect("enqueue list");
+        let rows = rx
+            .await
+            .expect("notification actor reply")
+            .expect("list notifications");
+
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0].notification_id, "tauri-notification");
+
+        for (id, channel, updated_at_ms) in [
+            ("tauri-job", "tauri", 1_i64),
+            ("terminal-job", "terminal", 2_i64),
+        ] {
+            let (tx, rx) = tokio::sync::oneshot::channel();
+            node.send_packet(MemoryMessage::UpsertBackgroundJob {
+                record: BackgroundJobRecord {
+                    job_id: id.to_string(),
+                    kind: "test".to_string(),
+                    chat_id: "chat-1".to_string(),
+                    channel: channel.to_string(),
+                    thread_id: None,
+                    state: "waiting".to_string(),
+                    payload_json: "{}".to_string(),
+                    resume_after_restart: false,
+                    detached: false,
+                    last_error: None,
+                    created_at_ms: updated_at_ms,
+                    updated_at_ms,
+                },
+                reply: SharedReply::new(tx),
+            })
+            .await
+            .expect("enqueue background job");
+            rx.await
+                .expect("background job actor reply")
+                .expect("insert background job");
+        }
+
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        node.send_packet(MemoryMessage::ListBackgroundJobs {
+            chat_id: None,
+            channel: Some("tauri".to_string()),
+            limit: 1,
+            reply: SharedReply::new(tx),
+        })
+        .await
+        .expect("enqueue background job list");
+        let jobs = rx
+            .await
+            .expect("background job actor reply")
+            .expect("list background jobs");
+        assert_eq!(jobs.len(), 1);
+        assert_eq!(jobs[0].job_id, "tauri-job");
+
+        for (id, channel, updated_at_ms) in [
+            ("tauri-ticket", "tauri", 1_i64),
+            ("terminal-ticket", "terminal", 2_i64),
+        ] {
+            let (tx, rx) = tokio::sync::oneshot::channel();
+            node.send_packet(MemoryMessage::UpsertClarificationTicket {
+                record: ClarificationTicketRecord {
+                    ticket_id: id.to_string(),
+                    job_id: id.to_string(),
+                    chat_id: "chat-1".to_string(),
+                    channel: channel.to_string(),
+                    thread_id: None,
+                    tool_call_id: None,
+                    prompt: "Test".to_string(),
+                    choices_json: None,
+                    response: None,
+                    status: "waiting".to_string(),
+                    created_at_ms: updated_at_ms,
+                    updated_at_ms,
+                },
+                reply: SharedReply::new(tx),
+            })
+            .await
+            .expect("enqueue clarification ticket");
+            rx.await
+                .expect("clarification ticket actor reply")
+                .expect("insert clarification ticket");
+        }
+
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        node.send_packet(MemoryMessage::ListClarificationTickets {
+            job_id: None,
+            chat_id: None,
+            channel: Some("tauri".to_string()),
+            status: None,
+            limit: 1,
+            reply: SharedReply::new(tx),
+        })
+        .await
+        .expect("enqueue clarification ticket list");
+        let tickets = rx
+            .await
+            .expect("clarification ticket actor reply")
+            .expect("list clarification tickets");
+        assert_eq!(tickets.len(), 1);
+        assert_eq!(tickets[0].ticket_id, "tauri-ticket");
+    }
+
     #[tokio::test]
     async fn get_context_returns_messages_in_insert_order() {
         let actor = SqliteMemoryActor::new(":memory:").expect("memory actor");