fix(memory): scope local inbox consumers by channel

Author: efecnc

src/agent/mod.rs

@@ -2037,7 +2037,7 @@ impl AgentLogic {
         let _ = memory_node
             .send_packet(MemoryMessage::ListBackgroundJobs {
                 chat_id: Some(chat_id.to_string()),
-                channel: None,
+                channel: Some(inbound.channel.clone()),
                 limit: 10,
                 reply: SharedReply::new(tx),
             })
@@ -2052,7 +2052,7 @@ impl AgentLogic {
                     .send_packet(MemoryMessage::ListClarificationTickets {
                         job_id: Some(job.job_id.clone()),
                         chat_id: Some(chat_id.to_string()),
-                        channel: None,
+                        channel: Some(inbound.channel.clone()),
                         status: Some("waiting".to_string()),
                         limit: 1,
                         reply: SharedReply::new(tx2),

src/channels/terminal_ui/run.rs

@@ -1382,6 +1382,7 @@ pub(crate) fn run_ratatui_main(config: RatatuiMainConfig) -> io::Result<()> {
         let crons_tx = crons_tx.clone();
         let jobs_tx = jobs_tx.clone();
         let notifications_tx = notifications_tx.clone();
+        let poll_channel_name = channel_name.clone();
         let spawn_result = std::thread::Builder::new()
             .name("ui-db-poller".into())
             .spawn(move || {
@@ -1425,7 +1426,7 @@ pub(crate) fn run_ratatui_main(config: RatatuiMainConfig) -> io::Result<()> {
                         let _ = memory_node
                             .send_packet(crate::memory::MemoryMessage::ListBackgroundJobs {
                                 chat_id: None,
-                                channel: None,
+                                channel: Some(poll_channel_name.clone()),
                                 limit: 50,
                                 reply: crate::memory::SharedReply::new(jtx),
                             })
@@ -1439,7 +1440,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,
+                                channel: Some(poll_channel_name.clone()),
                                 limit: 50,
                                 unseen_only: false,
                                 reply: crate::memory::SharedReply::new(ntx),

src/memory.rs

@@ -2315,8 +2315,8 @@ fn dropped_tool_call_ids(
 #[cfg(test)]
 mod root_thread_id_tests {
     use super::{
-        is_root_session_thread_id, MemoryMessage, NotificationRecord, SharedReply,
-        SqliteMemoryActor,
+        is_root_session_thread_id, BackgroundJobRecord, ClarificationTicketRecord, MemoryMessage,
+        NotificationRecord, SharedReply, SqliteMemoryActor,
     };
     use crate::session::SessionManager;
     use crate::traits::Memory;
@@ -2391,6 +2391,98 @@ mod root_thread_id_tests {
 
         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]