mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-26 15:40:18 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3f970252d5 | ||
|
|
3930184e71 | ||
|
|
4fb0d53206 | ||
|
|
5f50df7583 | ||
|
|
30feaabb7c | ||
|
|
427f908e71 | ||
|
|
8bd30a970e | ||
|
|
5d1d504e11 |
@@ -0,0 +1,4 @@
|
|||||||
|
-- Add source_channel to conversations for cross-channel approval authorization.
|
||||||
|
-- Tracks which channel originally created a conversation so that approval
|
||||||
|
-- messages from other channels can be validated.
|
||||||
|
ALTER TABLE conversations ADD COLUMN source_channel TEXT;
|
||||||
+35
-2
@@ -843,7 +843,10 @@ impl Agent {
|
|||||||
{
|
{
|
||||||
use crate::agent::session::Thread;
|
use crate::agent::session::Thread;
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = Thread::with_id(id, sess.id);
|
// Bootstrap thread has no incoming message -- use the
|
||||||
|
// "__bootstrap__" sentinel so approvals from any channel are
|
||||||
|
// permitted. None means "deny by default" (fail-closed).
|
||||||
|
let thread = Thread::with_id(id, sess.id, Some("__bootstrap__"));
|
||||||
sess.active_thread = Some(id);
|
sess.active_thread = Some(id);
|
||||||
sess.threads.entry(id).or_insert(thread);
|
sess.threads.entry(id).or_insert(thread);
|
||||||
}
|
}
|
||||||
@@ -1148,7 +1151,37 @@ impl Agent {
|
|||||||
.get_or_create_session(&message.user_id)
|
.get_or_create_session(&message.user_id)
|
||||||
.await;
|
.await;
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
if sess.threads.contains_key(&target_thread_id) {
|
if let Some(thread) = sess.threads.get(&target_thread_id) {
|
||||||
|
// Verify the thread actually has a pending approval before
|
||||||
|
// allowing approval-shaped messages to target it. Without this
|
||||||
|
// check, an attacker could use approval messages to hijack any
|
||||||
|
// thread by UUID.
|
||||||
|
if thread.pending_approval.is_none() {
|
||||||
|
tracing::warn!(
|
||||||
|
%target_thread_id,
|
||||||
|
approval_channel = %message.channel,
|
||||||
|
"Blocked approval for thread with no pending approval"
|
||||||
|
);
|
||||||
|
drop(sess);
|
||||||
|
return Ok(Some("Error: no pending approval on this thread".into()));
|
||||||
|
}
|
||||||
|
|
||||||
|
let authorized = crate::agent::session::is_approval_authorized(
|
||||||
|
thread.source_channel.as_deref(),
|
||||||
|
&message.channel,
|
||||||
|
);
|
||||||
|
if !authorized {
|
||||||
|
tracing::warn!(
|
||||||
|
%target_thread_id,
|
||||||
|
source_channel = ?thread.source_channel,
|
||||||
|
approval_channel = %message.channel,
|
||||||
|
"Blocked cross-channel approval attempt"
|
||||||
|
);
|
||||||
|
drop(sess);
|
||||||
|
return Ok(Some(
|
||||||
|
"Error: approval not authorized for this channel".into(),
|
||||||
|
));
|
||||||
|
}
|
||||||
sess.active_thread = Some(target_thread_id);
|
sess.active_thread = Some(target_thread_id);
|
||||||
sess.last_active_at = chrono::Utc::now();
|
sess.last_active_at = chrono::Utc::now();
|
||||||
drop(sess);
|
drop(sess);
|
||||||
|
|||||||
@@ -319,7 +319,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_format_turns() {
|
fn test_format_turns() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
thread.start_turn("Hello");
|
thread.start_turn("Hello");
|
||||||
thread.complete_turn("Hi there");
|
thread.complete_turn("Hi there");
|
||||||
thread.start_turn("How are you?");
|
thread.start_turn("How are you?");
|
||||||
@@ -351,7 +351,7 @@ mod tests {
|
|||||||
/// Helper: build a thread with `n` completed turns.
|
/// Helper: build a thread with `n` completed turns.
|
||||||
/// Turn `i` has user_input "msg-{i}" and response "resp-{i}".
|
/// Turn `i` has user_input "msg-{i}" and response "resp-{i}".
|
||||||
fn make_thread(n: usize) -> Thread {
|
fn make_thread(n: usize) -> Thread {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
for i in 0..n {
|
for i in 0..n {
|
||||||
thread.start_turn(format!("msg-{}", i));
|
thread.start_turn(format!("msg-{}", i));
|
||||||
thread.complete_turn(format!("resp-{}", i));
|
thread.complete_turn(format!("resp-{}", i));
|
||||||
@@ -457,7 +457,7 @@ mod tests {
|
|||||||
async fn test_compact_truncate_empty_turns() {
|
async fn test_compact_truncate_empty_turns() {
|
||||||
let llm = Arc::new(StubLlm::new("unused"));
|
let llm = Arc::new(StubLlm::new("unused"));
|
||||||
let compactor = make_compactor(llm);
|
let compactor = make_compactor(llm);
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
assert!(thread.turns.is_empty());
|
assert!(thread.turns.is_empty());
|
||||||
|
|
||||||
let result = compactor
|
let result = compactor
|
||||||
@@ -698,7 +698,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_format_turns_for_storage_with_tool_calls() {
|
fn test_format_turns_for_storage_with_tool_calls() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
thread.start_turn("Search for X");
|
thread.start_turn("Search for X");
|
||||||
// Record a tool call on the current turn
|
// Record a tool call on the current turn
|
||||||
if let Some(turn) = thread.turns.last_mut() {
|
if let Some(turn) = thread.turns.last_mut() {
|
||||||
@@ -719,7 +719,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_format_turns_for_storage_incomplete_turn() {
|
fn test_format_turns_for_storage_incomplete_turn() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
thread.start_turn("In progress message");
|
thread.start_turn("In progress message");
|
||||||
// Don't complete the turn
|
// Don't complete the turn
|
||||||
|
|
||||||
|
|||||||
@@ -2299,7 +2299,7 @@ mod tests {
|
|||||||
// Initialize a thread in the session so the loop can record tool calls.
|
// Initialize a thread in the session so the loop can record tool calls.
|
||||||
let thread_id = {
|
let thread_id = {
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
sess.create_thread().id
|
sess.create_thread(Some("test")).id
|
||||||
};
|
};
|
||||||
|
|
||||||
let message = IncomingMessage::new("test", "test-user", "do something");
|
let message = IncomingMessage::new("test", "test-user", "do something");
|
||||||
@@ -2412,7 +2412,7 @@ mod tests {
|
|||||||
let session = Arc::new(Mutex::new(Session::new("test-user")));
|
let session = Arc::new(Mutex::new(Session::new("test-user")));
|
||||||
let thread_id = {
|
let thread_id = {
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
sess.create_thread().id
|
sess.create_thread(Some("test")).id
|
||||||
};
|
};
|
||||||
|
|
||||||
let message = IncomingMessage::new("test", "test-user", "keep calling tools");
|
let message = IncomingMessage::new("test", "test-user", "keep calling tools");
|
||||||
|
|||||||
+169
-54
@@ -68,8 +68,8 @@ impl Session {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Create a new thread in this session.
|
/// Create a new thread in this session.
|
||||||
pub fn create_thread(&mut self) -> &mut Thread {
|
pub fn create_thread(&mut self, channel: Option<&str>) -> &mut Thread {
|
||||||
let thread = Thread::new(self.id);
|
let thread = Thread::new(self.id, channel);
|
||||||
let thread_id = thread.id;
|
let thread_id = thread.id;
|
||||||
self.active_thread = Some(thread_id);
|
self.active_thread = Some(thread_id);
|
||||||
self.last_active_at = Utc::now();
|
self.last_active_at = Utc::now();
|
||||||
@@ -87,9 +87,9 @@ impl Session {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Get or create the active thread.
|
/// Get or create the active thread.
|
||||||
pub fn get_or_create_thread(&mut self) -> &mut Thread {
|
pub fn get_or_create_thread(&mut self, channel: Option<&str>) -> &mut Thread {
|
||||||
match self.active_thread {
|
match self.active_thread {
|
||||||
None => self.create_thread(),
|
None => self.create_thread(channel),
|
||||||
Some(id) => {
|
Some(id) => {
|
||||||
if self.threads.contains_key(&id) {
|
if self.threads.contains_key(&id) {
|
||||||
// Entry existence confirmed by contains_key above.
|
// Entry existence confirmed by contains_key above.
|
||||||
@@ -100,7 +100,7 @@ impl Session {
|
|||||||
} else {
|
} else {
|
||||||
// Stale active_thread ID: create a new thread, which
|
// Stale active_thread ID: create a new thread, which
|
||||||
// updates self.active_thread to the new thread's ID.
|
// updates self.active_thread to the new thread's ID.
|
||||||
self.create_thread()
|
self.create_thread(channel)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -225,6 +225,9 @@ pub struct Thread {
|
|||||||
/// Messages queued while the thread was processing a turn.
|
/// Messages queued while the thread was processing a turn.
|
||||||
#[serde(default, skip_serializing_if = "VecDeque::is_empty")]
|
#[serde(default, skip_serializing_if = "VecDeque::is_empty")]
|
||||||
pub pending_messages: VecDeque<String>,
|
pub pending_messages: VecDeque<String>,
|
||||||
|
/// Channel that created this thread (for approval authorization).
|
||||||
|
#[serde(default)]
|
||||||
|
pub source_channel: Option<String>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Maximum number of messages that can be queued while a thread is processing.
|
/// Maximum number of messages that can be queued while a thread is processing.
|
||||||
@@ -233,9 +236,29 @@ pub struct Thread {
|
|||||||
/// rapid follow-ups. The drain loop processes them as one newline-delimited turn.
|
/// rapid follow-ups. The drain loop processes them as one newline-delimited turn.
|
||||||
pub const MAX_PENDING_MESSAGES: usize = 10;
|
pub const MAX_PENDING_MESSAGES: usize = 10;
|
||||||
|
|
||||||
|
/// Sentinel value for bootstrap threads that accept approvals from any channel.
|
||||||
|
pub const BOOTSTRAP_SOURCE_CHANNEL: &str = "__bootstrap__";
|
||||||
|
|
||||||
|
/// Check whether an approval from `requesting_channel` is authorized for a
|
||||||
|
/// thread whose `source_channel` is `source`.
|
||||||
|
///
|
||||||
|
/// Rules:
|
||||||
|
/// - `None` (unknown origin) -> denied (fail-closed)
|
||||||
|
/// - `Some("__bootstrap__")` -> authorized from any channel
|
||||||
|
/// - `Some(src) == requesting` -> same channel, authorized
|
||||||
|
/// - requesting is "web" or "gateway" -> always authorized (trusted UI)
|
||||||
|
/// - Otherwise -> denied
|
||||||
|
pub fn is_approval_authorized(source: Option<&str>, requesting: &str) -> bool {
|
||||||
|
match source {
|
||||||
|
None => false,
|
||||||
|
Some(src) if src == BOOTSTRAP_SOURCE_CHANNEL => true,
|
||||||
|
Some(src) => src == requesting || requesting == "web" || requesting == "gateway",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
impl Thread {
|
impl Thread {
|
||||||
/// Create a new thread.
|
/// Create a new thread.
|
||||||
pub fn new(session_id: Uuid) -> Self {
|
pub fn new(session_id: Uuid, source_channel: Option<&str>) -> Self {
|
||||||
let now = Utc::now();
|
let now = Utc::now();
|
||||||
Self {
|
Self {
|
||||||
id: Uuid::new_v4(),
|
id: Uuid::new_v4(),
|
||||||
@@ -248,11 +271,12 @@ impl Thread {
|
|||||||
pending_approval: None,
|
pending_approval: None,
|
||||||
pending_auth: None,
|
pending_auth: None,
|
||||||
pending_messages: VecDeque::new(),
|
pending_messages: VecDeque::new(),
|
||||||
|
source_channel: source_channel.map(String::from),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Create a thread with a specific ID (for DB hydration).
|
/// Create a thread with a specific ID (for DB hydration).
|
||||||
pub fn with_id(id: Uuid, session_id: Uuid) -> Self {
|
pub fn with_id(id: Uuid, session_id: Uuid, source_channel: Option<&str>) -> Self {
|
||||||
let now = Utc::now();
|
let now = Utc::now();
|
||||||
Self {
|
Self {
|
||||||
id,
|
id,
|
||||||
@@ -265,6 +289,7 @@ impl Thread {
|
|||||||
pending_approval: None,
|
pending_approval: None,
|
||||||
pending_auth: None,
|
pending_auth: None,
|
||||||
pending_messages: VecDeque::new(),
|
pending_messages: VecDeque::new(),
|
||||||
|
source_channel: source_channel.map(String::from),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -787,13 +812,13 @@ mod tests {
|
|||||||
let mut session = Session::new("user-123");
|
let mut session = Session::new("user-123");
|
||||||
assert!(session.active_thread.is_none());
|
assert!(session.active_thread.is_none());
|
||||||
|
|
||||||
session.create_thread();
|
session.create_thread(None);
|
||||||
assert!(session.active_thread.is_some());
|
assert!(session.active_thread.is_some());
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_thread_turns() {
|
fn test_thread_turns() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
thread.start_turn("Hello");
|
thread.start_turn("Hello");
|
||||||
assert_eq!(thread.state, ThreadState::Processing);
|
assert_eq!(thread.state, ThreadState::Processing);
|
||||||
@@ -806,7 +831,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_thread_messages() {
|
fn test_thread_messages() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
thread.start_turn("First message");
|
thread.start_turn("First message");
|
||||||
thread.complete_turn("First response");
|
thread.complete_turn("First response");
|
||||||
@@ -829,7 +854,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_restore_from_messages() {
|
fn test_restore_from_messages() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// First add some turns
|
// First add some turns
|
||||||
thread.start_turn("Original message");
|
thread.start_turn("Original message");
|
||||||
@@ -855,7 +880,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_restore_from_messages_incomplete_turn() {
|
fn test_restore_from_messages_incomplete_turn() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Messages with incomplete last turn (no assistant response)
|
// Messages with incomplete last turn (no assistant response)
|
||||||
let messages = vec![
|
let messages = vec![
|
||||||
@@ -874,7 +899,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn test_enter_auth_mode() {
|
fn test_enter_auth_mode() {
|
||||||
let before = Utc::now();
|
let before = Utc::now();
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
assert!(thread.pending_auth.is_none());
|
assert!(thread.pending_auth.is_none());
|
||||||
|
|
||||||
thread.enter_auth_mode("telegram".to_string());
|
thread.enter_auth_mode("telegram".to_string());
|
||||||
@@ -887,7 +912,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_take_pending_auth() {
|
fn test_take_pending_auth() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
thread.enter_auth_mode("notion".to_string());
|
thread.enter_auth_mode("notion".to_string());
|
||||||
|
|
||||||
let pending = thread.take_pending_auth();
|
let pending = thread.take_pending_auth();
|
||||||
@@ -902,7 +927,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_pending_auth_serialization() {
|
fn test_pending_auth_serialization() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
thread.enter_auth_mode("openai".to_string());
|
thread.enter_auth_mode("openai".to_string());
|
||||||
|
|
||||||
let json = serde_json::to_string(&thread).expect("should serialize");
|
let json = serde_json::to_string(&thread).expect("should serialize");
|
||||||
@@ -932,7 +957,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn test_pending_auth_default_none() {
|
fn test_pending_auth_default_none() {
|
||||||
// Deserialization of old data without pending_auth should default to None
|
// Deserialization of old data without pending_auth should default to None
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
thread.pending_auth = None;
|
thread.pending_auth = None;
|
||||||
let json = serde_json::to_string(&thread).expect("serialize");
|
let json = serde_json::to_string(&thread).expect("serialize");
|
||||||
|
|
||||||
@@ -946,7 +971,7 @@ mod tests {
|
|||||||
fn test_thread_with_id() {
|
fn test_thread_with_id() {
|
||||||
let specific_id = Uuid::new_v4();
|
let specific_id = Uuid::new_v4();
|
||||||
let session_id = Uuid::new_v4();
|
let session_id = Uuid::new_v4();
|
||||||
let thread = Thread::with_id(specific_id, session_id);
|
let thread = Thread::with_id(specific_id, session_id, None);
|
||||||
|
|
||||||
assert_eq!(thread.id, specific_id);
|
assert_eq!(thread.id, specific_id);
|
||||||
assert_eq!(thread.session_id, session_id);
|
assert_eq!(thread.session_id, session_id);
|
||||||
@@ -958,7 +983,7 @@ mod tests {
|
|||||||
fn test_thread_with_id_restore_messages() {
|
fn test_thread_with_id_restore_messages() {
|
||||||
let thread_id = Uuid::new_v4();
|
let thread_id = Uuid::new_v4();
|
||||||
let session_id = Uuid::new_v4();
|
let session_id = Uuid::new_v4();
|
||||||
let mut thread = Thread::with_id(thread_id, session_id);
|
let mut thread = Thread::with_id(thread_id, session_id, None);
|
||||||
|
|
||||||
let messages = vec![
|
let messages = vec![
|
||||||
ChatMessage::user("Hello from DB"),
|
ChatMessage::user("Hello from DB"),
|
||||||
@@ -977,7 +1002,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_restore_from_messages_empty() {
|
fn test_restore_from_messages_empty() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Add a turn first, then restore with empty vec
|
// Add a turn first, then restore with empty vec
|
||||||
thread.start_turn("hello");
|
thread.start_turn("hello");
|
||||||
@@ -993,7 +1018,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_restore_from_messages_only_assistant_messages() {
|
fn test_restore_from_messages_only_assistant_messages() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Only assistant messages (no user messages to anchor turns)
|
// Only assistant messages (no user messages to anchor turns)
|
||||||
let messages = vec![
|
let messages = vec![
|
||||||
@@ -1010,7 +1035,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_restore_from_messages_multiple_user_messages_in_a_row() {
|
fn test_restore_from_messages_multiple_user_messages_in_a_row() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Two user messages with no assistant response between them
|
// Two user messages with no assistant response between them
|
||||||
let messages = vec![
|
let messages = vec![
|
||||||
@@ -1037,8 +1062,8 @@ mod tests {
|
|||||||
fn test_thread_switch() {
|
fn test_thread_switch() {
|
||||||
let mut session = Session::new("user-1");
|
let mut session = Session::new("user-1");
|
||||||
|
|
||||||
let t1_id = session.create_thread().id;
|
let t1_id = session.create_thread(None).id;
|
||||||
let t2_id = session.create_thread().id;
|
let t2_id = session.create_thread(None).id;
|
||||||
|
|
||||||
// After creating two threads, active should be the last one
|
// After creating two threads, active should be the last one
|
||||||
assert_eq!(session.active_thread, Some(t2_id));
|
assert_eq!(session.active_thread, Some(t2_id));
|
||||||
@@ -1058,8 +1083,8 @@ mod tests {
|
|||||||
fn test_get_or_create_thread_idempotent() {
|
fn test_get_or_create_thread_idempotent() {
|
||||||
let mut session = Session::new("user-1");
|
let mut session = Session::new("user-1");
|
||||||
|
|
||||||
let tid1 = session.get_or_create_thread().id;
|
let tid1 = session.get_or_create_thread(None).id;
|
||||||
let tid2 = session.get_or_create_thread().id;
|
let tid2 = session.get_or_create_thread(None).id;
|
||||||
|
|
||||||
// Should return the same thread (not create a new one each time)
|
// Should return the same thread (not create a new one each time)
|
||||||
assert_eq!(tid1, tid2);
|
assert_eq!(tid1, tid2);
|
||||||
@@ -1068,7 +1093,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_truncate_turns() {
|
fn test_truncate_turns() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
for i in 0..5 {
|
for i in 0..5 {
|
||||||
thread.start_turn(format!("msg-{}", i));
|
thread.start_turn(format!("msg-{}", i));
|
||||||
@@ -1092,7 +1117,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_truncate_turns_noop_when_fewer() {
|
fn test_truncate_turns_noop_when_fewer() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
thread.start_turn("only one");
|
thread.start_turn("only one");
|
||||||
thread.complete_turn("response");
|
thread.complete_turn("response");
|
||||||
@@ -1104,7 +1129,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_thread_interrupt_and_resume() {
|
fn test_thread_interrupt_and_resume() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
thread.start_turn("do something");
|
thread.start_turn("do something");
|
||||||
assert_eq!(thread.state, ThreadState::Processing);
|
assert_eq!(thread.state, ThreadState::Processing);
|
||||||
@@ -1122,7 +1147,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_resume_only_from_interrupted() {
|
fn test_resume_only_from_interrupted() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Idle thread: resume should be a no-op
|
// Idle thread: resume should be a no-op
|
||||||
assert_eq!(thread.state, ThreadState::Idle);
|
assert_eq!(thread.state, ThreadState::Idle);
|
||||||
@@ -1138,7 +1163,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_turn_fail() {
|
fn test_turn_fail() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
thread.start_turn("risky operation");
|
thread.start_turn("risky operation");
|
||||||
thread.fail_turn("connection timed out");
|
thread.fail_turn("connection timed out");
|
||||||
@@ -1154,7 +1179,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_messages_with_incomplete_last_turn() {
|
fn test_messages_with_incomplete_last_turn() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
thread.start_turn("first");
|
thread.start_turn("first");
|
||||||
thread.complete_turn("first reply");
|
thread.complete_turn("first reply");
|
||||||
@@ -1170,7 +1195,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_thread_serialization_round_trip() {
|
fn test_thread_serialization_round_trip() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
thread.start_turn("hello");
|
thread.start_turn("hello");
|
||||||
thread.complete_turn("world");
|
thread.complete_turn("world");
|
||||||
@@ -1188,7 +1213,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn test_session_serialization_round_trip() {
|
fn test_session_serialization_round_trip() {
|
||||||
let mut session = Session::new("user-ser");
|
let mut session = Session::new("user-ser");
|
||||||
session.create_thread();
|
session.create_thread(None);
|
||||||
session.auto_approve_tool("echo");
|
session.auto_approve_tool("echo");
|
||||||
|
|
||||||
let json = serde_json::to_string(&session).unwrap();
|
let json = serde_json::to_string(&session).unwrap();
|
||||||
@@ -1226,7 +1251,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_turn_number_increments() {
|
fn test_turn_number_increments() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Before any turns, turn_number() is 1 (1-indexed for display)
|
// Before any turns, turn_number() is 1 (1-indexed for display)
|
||||||
assert_eq!(thread.turn_number(), 1);
|
assert_eq!(thread.turn_number(), 1);
|
||||||
@@ -1241,7 +1266,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_complete_turn_on_empty_thread() {
|
fn test_complete_turn_on_empty_thread() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Completing a turn when there are no turns should be a safe no-op
|
// Completing a turn when there are no turns should be a safe no-op
|
||||||
thread.complete_turn("phantom response");
|
thread.complete_turn("phantom response");
|
||||||
@@ -1251,7 +1276,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_fail_turn_on_empty_thread() {
|
fn test_fail_turn_on_empty_thread() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Failing a turn when there are no turns should be a safe no-op
|
// Failing a turn when there are no turns should be a safe no-op
|
||||||
thread.fail_turn("phantom error");
|
thread.fail_turn("phantom error");
|
||||||
@@ -1261,7 +1286,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_pending_approval_flow() {
|
fn test_pending_approval_flow() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
let approval = PendingApproval {
|
let approval = PendingApproval {
|
||||||
request_id: Uuid::new_v4(),
|
request_id: Uuid::new_v4(),
|
||||||
@@ -1288,7 +1313,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_clear_pending_approval() {
|
fn test_clear_pending_approval() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
let approval = PendingApproval {
|
let approval = PendingApproval {
|
||||||
request_id: Uuid::new_v4(),
|
request_id: Uuid::new_v4(),
|
||||||
@@ -1317,7 +1342,7 @@ mod tests {
|
|||||||
assert!(session.active_thread().is_none());
|
assert!(session.active_thread().is_none());
|
||||||
assert!(session.active_thread_mut().is_none());
|
assert!(session.active_thread_mut().is_none());
|
||||||
|
|
||||||
let tid = session.create_thread().id;
|
let tid = session.create_thread(None).id;
|
||||||
|
|
||||||
assert!(session.active_thread().is_some());
|
assert!(session.active_thread().is_some());
|
||||||
assert_eq!(session.active_thread().unwrap().id, tid);
|
assert_eq!(session.active_thread().unwrap().id, tid);
|
||||||
@@ -1334,7 +1359,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_messages_includes_tool_calls() {
|
fn test_messages_includes_tool_calls() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
thread.start_turn("Search for X");
|
thread.start_turn("Search for X");
|
||||||
{
|
{
|
||||||
@@ -1366,7 +1391,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_messages_multiple_tool_calls_per_turn() {
|
fn test_messages_multiple_tool_calls_per_turn() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
thread.start_turn("Do two things");
|
thread.start_turn("Do two things");
|
||||||
{
|
{
|
||||||
@@ -1393,7 +1418,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_restore_from_messages_with_tool_calls() {
|
fn test_restore_from_messages_with_tool_calls() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Build a message sequence with tool calls
|
// Build a message sequence with tool calls
|
||||||
let tc = ToolCall {
|
let tc = ToolCall {
|
||||||
@@ -1425,7 +1450,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_restore_from_messages_with_tool_error() {
|
fn test_restore_from_messages_with_tool_error() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
let tc = ToolCall {
|
let tc = ToolCall {
|
||||||
id: "call_0".to_string(),
|
id: "call_0".to_string(),
|
||||||
@@ -1456,7 +1481,7 @@ mod tests {
|
|||||||
fn test_messages_round_trip_with_tools() {
|
fn test_messages_round_trip_with_tools() {
|
||||||
// Build a thread with tool calls, get messages(), restore, get messages() again
|
// Build a thread with tool calls, get messages(), restore, get messages() again
|
||||||
// The two message sequences should be equivalent.
|
// The two message sequences should be equivalent.
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
thread.start_turn("Do search");
|
thread.start_turn("Do search");
|
||||||
{
|
{
|
||||||
@@ -1469,7 +1494,7 @@ mod tests {
|
|||||||
let messages_original = thread.messages();
|
let messages_original = thread.messages();
|
||||||
|
|
||||||
// Restore into a new thread
|
// Restore into a new thread
|
||||||
let mut thread2 = Thread::new(Uuid::new_v4());
|
let mut thread2 = Thread::new(Uuid::new_v4(), None);
|
||||||
thread2.restore_from_messages(messages_original.clone());
|
thread2.restore_from_messages(messages_original.clone());
|
||||||
|
|
||||||
let messages_restored = thread2.messages();
|
let messages_restored = thread2.messages();
|
||||||
@@ -1491,7 +1516,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_restore_multi_stage_tool_calls() {
|
fn test_restore_multi_stage_tool_calls() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
let tc1 = ToolCall {
|
let tc1 = ToolCall {
|
||||||
id: "call_a".to_string(),
|
id: "call_a".to_string(),
|
||||||
@@ -1534,7 +1559,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_messages_truncates_large_tool_results() {
|
fn test_messages_truncates_large_tool_results() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
thread.start_turn("Read big file");
|
thread.start_turn("Read big file");
|
||||||
{
|
{
|
||||||
@@ -1557,7 +1582,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_thread_message_queue() {
|
fn test_thread_message_queue() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Queue is initially empty
|
// Queue is initially empty
|
||||||
assert!(thread.pending_messages.is_empty());
|
assert!(thread.pending_messages.is_empty());
|
||||||
@@ -1593,7 +1618,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_thread_message_queue_serialization() {
|
fn test_thread_message_queue_serialization() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Empty queue should not appear in serialization (skip_serializing_if)
|
// Empty queue should not appear in serialization (skip_serializing_if)
|
||||||
let json = serde_json::to_string(&thread).unwrap();
|
let json = serde_json::to_string(&thread).unwrap();
|
||||||
@@ -1613,7 +1638,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn test_thread_message_queue_default_on_old_data() {
|
fn test_thread_message_queue_default_on_old_data() {
|
||||||
// Deserialization of old data without pending_messages should default to empty
|
// Deserialization of old data without pending_messages should default to empty
|
||||||
let thread = Thread::new(Uuid::new_v4());
|
let thread = Thread::new(Uuid::new_v4(), None);
|
||||||
let json = serde_json::to_string(&thread).unwrap();
|
let json = serde_json::to_string(&thread).unwrap();
|
||||||
|
|
||||||
// The field is absent (skip_serializing_if), simulating old data
|
// The field is absent (skip_serializing_if), simulating old data
|
||||||
@@ -1624,7 +1649,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_interrupt_clears_pending_messages() {
|
fn test_interrupt_clears_pending_messages() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Start a turn so there's something to interrupt
|
// Start a turn so there's something to interrupt
|
||||||
thread.start_turn("initial input");
|
thread.start_turn("initial input");
|
||||||
@@ -1643,7 +1668,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_thread_state_idle_after_full_drain() {
|
fn test_thread_state_idle_after_full_drain() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Simulate a full drain cycle: start turn, queue messages, complete turn,
|
// Simulate a full drain cycle: start turn, queue messages, complete turn,
|
||||||
// then drain all queued messages as a single merged turn (#259).
|
// then drain all queued messages as a single merged turn (#259).
|
||||||
@@ -1671,7 +1696,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_drain_pending_messages_merges_with_newlines() {
|
fn test_drain_pending_messages_merges_with_newlines() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Empty queue returns None
|
// Empty queue returns None
|
||||||
assert!(thread.drain_pending_messages().is_none());
|
assert!(thread.drain_pending_messages().is_none());
|
||||||
@@ -1700,7 +1725,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_requeue_drained_preserves_content_at_front() {
|
fn test_requeue_drained_preserves_content_at_front() {
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
|
||||||
// Re-queue into empty queue
|
// Re-queue into empty queue
|
||||||
thread.requeue_drained("failed batch".to_string());
|
thread.requeue_drained("failed batch".to_string());
|
||||||
@@ -1811,4 +1836,94 @@ mod tests {
|
|||||||
&serde_json::json!("done")
|
&serde_json::json!("done")
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_thread_new_stores_source_channel() {
|
||||||
|
let thread = Thread::new(Uuid::new_v4(), Some("telegram"));
|
||||||
|
assert_eq!(thread.source_channel.as_deref(), Some("telegram"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_thread_new_none_channel() {
|
||||||
|
let thread = Thread::new(Uuid::new_v4(), None);
|
||||||
|
assert!(thread.source_channel.is_none());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_source_channel_serde_backcompat() {
|
||||||
|
// Simulate deserializing a Thread from older DB records that lack source_channel.
|
||||||
|
let thread = Thread::new(Uuid::new_v4(), Some("cli"));
|
||||||
|
let json = serde_json::to_string(&thread).unwrap();
|
||||||
|
|
||||||
|
// Remove the source_channel field to simulate an old record.
|
||||||
|
let mut value: serde_json::Value = serde_json::from_str(&json).unwrap();
|
||||||
|
value.as_object_mut().unwrap().remove("source_channel");
|
||||||
|
let old_json = serde_json::to_string(&value).unwrap();
|
||||||
|
|
||||||
|
let deserialized: Thread = serde_json::from_str(&old_json).unwrap();
|
||||||
|
assert!(
|
||||||
|
deserialized.source_channel.is_none(),
|
||||||
|
"missing source_channel should deserialize as None"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_approval_authorized_same_channel() {
|
||||||
|
assert!(
|
||||||
|
is_approval_authorized(Some("telegram"), "telegram"),
|
||||||
|
"same channel should be authorized"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_approval_authorized_different_channel_blocked() {
|
||||||
|
assert!(
|
||||||
|
!is_approval_authorized(Some("telegram"), "http"),
|
||||||
|
"different channel should be blocked"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_approval_authorized_web_always_allowed() {
|
||||||
|
assert!(
|
||||||
|
is_approval_authorized(Some("telegram"), "web"),
|
||||||
|
"web channel should always be authorized"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_approval_authorized_gateway_always_allowed() {
|
||||||
|
assert!(
|
||||||
|
is_approval_authorized(Some("telegram"), "gateway"),
|
||||||
|
"gateway channel should always be authorized"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_approval_authorized_none_denied() {
|
||||||
|
assert!(
|
||||||
|
!is_approval_authorized(None, "telegram"),
|
||||||
|
"None source_channel should be denied (fail-closed)"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
!is_approval_authorized(None, "web"),
|
||||||
|
"None source_channel should be denied even for web"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_approval_authorized_bootstrap_any_channel() {
|
||||||
|
assert!(
|
||||||
|
is_approval_authorized(Some(BOOTSTRAP_SOURCE_CHANNEL), "telegram"),
|
||||||
|
"__bootstrap__ should be authorized from any channel"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
is_approval_authorized(Some(BOOTSTRAP_SOURCE_CHANNEL), "http"),
|
||||||
|
"__bootstrap__ should be authorized from any channel"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
is_approval_authorized(Some(BOOTSTRAP_SOURCE_CHANNEL), "cli"),
|
||||||
|
"__bootstrap__ should be authorized from any channel"
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -200,7 +200,7 @@ impl SessionManager {
|
|||||||
// Create new thread (always create a new one for a new key)
|
// Create new thread (always create a new one for a new key)
|
||||||
let thread_id = {
|
let thread_id = {
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = sess.create_thread();
|
let thread = sess.create_thread(Some(channel));
|
||||||
thread.id
|
thread.id
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -476,7 +476,7 @@ mod tests {
|
|||||||
let session = Arc::new(Mutex::new(Session::new("user-hydrate")));
|
let session = Arc::new(Mutex::new(Session::new("user-hydrate")));
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = Thread::with_id(thread_id, sess.id);
|
let thread = Thread::with_id(thread_id, sess.id, None);
|
||||||
sess.threads.insert(thread_id, thread);
|
sess.threads.insert(thread_id, thread);
|
||||||
sess.active_thread = Some(thread_id);
|
sess.active_thread = Some(thread_id);
|
||||||
}
|
}
|
||||||
@@ -600,7 +600,7 @@ mod tests {
|
|||||||
// Simulate hydration: create thread with a known UUID
|
// Simulate hydration: create thread with a known UUID
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = Thread::with_id(known_uuid, session_id);
|
let thread = Thread::with_id(known_uuid, session_id, None);
|
||||||
sess.threads.insert(known_uuid, thread);
|
sess.threads.insert(known_uuid, thread);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -627,7 +627,7 @@ mod tests {
|
|||||||
let session = Arc::new(Mutex::new(Session::new("user-idem")));
|
let session = Arc::new(Mutex::new(Session::new("user-idem")));
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = Thread::with_id(tid, sess.id);
|
let thread = Thread::with_id(tid, sess.id, None);
|
||||||
sess.threads.insert(tid, thread);
|
sess.threads.insert(tid, thread);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -656,7 +656,7 @@ mod tests {
|
|||||||
let session = Arc::new(Mutex::new(Session::new("user-undo")));
|
let session = Arc::new(Mutex::new(Session::new("user-undo")));
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = Thread::with_id(tid, sess.id);
|
let thread = Thread::with_id(tid, sess.id, None);
|
||||||
sess.threads.insert(tid, thread);
|
sess.threads.insert(tid, thread);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -680,7 +680,7 @@ mod tests {
|
|||||||
let session = Arc::new(Mutex::new(Session::new("user-new")));
|
let session = Arc::new(Mutex::new(Session::new("user-new")));
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = Thread::with_id(tid, sess.id);
|
let thread = Thread::with_id(tid, sess.id, None);
|
||||||
sess.threads.insert(tid, thread);
|
sess.threads.insert(tid, thread);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -788,7 +788,7 @@ mod tests {
|
|||||||
let session = Arc::new(Mutex::new(Session::new("user-cross")));
|
let session = Arc::new(Mutex::new(Session::new("user-cross")));
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = Thread::with_id(tid, sess.id);
|
let thread = Thread::with_id(tid, sess.id, None);
|
||||||
sess.threads.insert(tid, thread);
|
sess.threads.insert(tid, thread);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -815,7 +815,7 @@ mod tests {
|
|||||||
let session = Arc::new(Mutex::new(Session::new("user-cross")));
|
let session = Arc::new(Mutex::new(Session::new("user-cross")));
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = Thread::with_id(tid, sess.id);
|
let thread = Thread::with_id(tid, sess.id, None);
|
||||||
sess.threads.insert(tid, thread);
|
sess.threads.insert(tid, thread);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -966,7 +966,7 @@ mod tests {
|
|||||||
let adopted_id = Uuid::new_v4();
|
let adopted_id = Uuid::new_v4();
|
||||||
{
|
{
|
||||||
let mut sess = session1.lock().await;
|
let mut sess = session1.lock().await;
|
||||||
let thread = Thread::with_id(adopted_id, sess.id);
|
let thread = Thread::with_id(adopted_id, sess.id, None);
|
||||||
sess.threads.insert(adopted_id, thread);
|
sess.threads.insert(adopted_id, thread);
|
||||||
}
|
}
|
||||||
// Resolve with the UUID as external_thread_id -- should adopt it
|
// Resolve with the UUID as external_thread_id -- should adopt it
|
||||||
@@ -992,7 +992,7 @@ mod tests {
|
|||||||
let session = Arc::new(Mutex::new(Session::new("user-direct")));
|
let session = Arc::new(Mutex::new(Session::new("user-direct")));
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = Thread::with_id(tid, sess.id);
|
let thread = Thread::with_id(tid, sess.id, None);
|
||||||
sess.threads.insert(tid, thread);
|
sess.threads.insert(tid, thread);
|
||||||
}
|
}
|
||||||
{
|
{
|
||||||
@@ -1030,7 +1030,7 @@ mod tests {
|
|||||||
let known_id = Uuid::new_v4();
|
let known_id = Uuid::new_v4();
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = Thread::with_id(known_id, sess.id);
|
let thread = Thread::with_id(known_id, sess.id, None);
|
||||||
sess.threads.insert(known_id, thread);
|
sess.threads.insert(known_id, thread);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1057,7 +1057,7 @@ mod tests {
|
|||||||
let known_id = Uuid::new_v4();
|
let known_id = Uuid::new_v4();
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = Thread::with_id(known_id, sess.id);
|
let thread = Thread::with_id(known_id, sess.id, None);
|
||||||
sess.threads.insert(known_id, thread);
|
sess.threads.insert(known_id, thread);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1081,7 +1081,7 @@ mod tests {
|
|||||||
let known_id = Uuid::new_v4();
|
let known_id = Uuid::new_v4();
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = Thread::with_id(known_id, sess.id);
|
let thread = Thread::with_id(known_id, sess.id, None);
|
||||||
sess.threads.insert(known_id, thread);
|
sess.threads.insert(known_id, thread);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1102,4 +1102,19 @@ mod tests {
|
|||||||
"should NOT adopt UUID when external_thread_id is None"
|
"should NOT adopt UUID when external_thread_id is None"
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_thread_stores_source_channel() {
|
||||||
|
let manager = SessionManager::new();
|
||||||
|
|
||||||
|
let (session, thread_id) = manager.resolve_thread("user-1", "telegram", None).await;
|
||||||
|
|
||||||
|
let sess = session.lock().await;
|
||||||
|
let thread = sess.threads.get(&thread_id).unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
thread.source_channel.as_deref(),
|
||||||
|
Some("telegram"),
|
||||||
|
"resolve_thread should store source_channel from the channel parameter"
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+25
-9
@@ -135,13 +135,29 @@ impl Agent {
|
|||||||
msg_count = 0;
|
msg_count = 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Create thread with the historical ID and restore messages
|
// Create thread with the historical ID and restore messages.
|
||||||
|
// Read source_channel from DB so the authorization check uses the
|
||||||
|
// original creator's channel, not the requesting message's channel.
|
||||||
|
let db_source_channel = if let Some(store) = self.store() {
|
||||||
|
store
|
||||||
|
.get_conversation_source_channel(thread_uuid)
|
||||||
|
.await
|
||||||
|
.unwrap_or(None)
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
};
|
||||||
|
let effective_source_channel = db_source_channel.as_deref();
|
||||||
|
|
||||||
let session_id = {
|
let session_id = {
|
||||||
let sess = session.lock().await;
|
let sess = session.lock().await;
|
||||||
sess.id
|
sess.id
|
||||||
};
|
};
|
||||||
|
|
||||||
let mut thread = crate::agent::session::Thread::with_id(thread_uuid, session_id);
|
let mut thread = crate::agent::session::Thread::with_id(
|
||||||
|
thread_uuid,
|
||||||
|
session_id,
|
||||||
|
effective_source_channel,
|
||||||
|
);
|
||||||
if !chat_messages.is_empty() {
|
if !chat_messages.is_empty() {
|
||||||
thread.restore_from_messages(chat_messages);
|
thread.restore_from_messages(chat_messages);
|
||||||
}
|
}
|
||||||
@@ -636,7 +652,7 @@ impl Agent {
|
|||||||
user_id: &str,
|
user_id: &str,
|
||||||
) -> bool {
|
) -> bool {
|
||||||
match store
|
match store
|
||||||
.ensure_conversation(thread_id, channel, user_id, None)
|
.ensure_conversation(thread_id, channel, user_id, None, Some(channel))
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
Ok(true) => true,
|
Ok(true) => true,
|
||||||
@@ -1781,7 +1797,7 @@ impl Agent {
|
|||||||
.get_or_create_session(&message.user_id)
|
.get_or_create_session(&message.user_id)
|
||||||
.await;
|
.await;
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = sess.create_thread();
|
let thread = sess.create_thread(Some(&message.channel));
|
||||||
let thread_id = thread.id;
|
let thread_id = thread.id;
|
||||||
Ok(SubmissionResult::ok_with_message(format!(
|
Ok(SubmissionResult::ok_with_message(format!(
|
||||||
"New thread: {}",
|
"New thread: {}",
|
||||||
@@ -2117,7 +2133,7 @@ mod tests {
|
|||||||
|
|
||||||
let session_id = Uuid::new_v4();
|
let session_id = Uuid::new_v4();
|
||||||
let thread_id = Uuid::new_v4();
|
let thread_id = Uuid::new_v4();
|
||||||
let mut thread = Thread::with_id(thread_id, session_id);
|
let mut thread = Thread::with_id(thread_id, session_id, None);
|
||||||
|
|
||||||
// Set thread to AwaitingApproval with a pending tool approval
|
// Set thread to AwaitingApproval with a pending tool approval
|
||||||
let pending = PendingApproval {
|
let pending = PendingApproval {
|
||||||
@@ -2185,7 +2201,7 @@ mod tests {
|
|||||||
use crate::agent::session::{MAX_PENDING_MESSAGES, Thread, ThreadState};
|
use crate::agent::session::{MAX_PENDING_MESSAGES, Thread, ThreadState};
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
thread.start_turn("processing something");
|
thread.start_turn("processing something");
|
||||||
assert_eq!(thread.state, ThreadState::Processing);
|
assert_eq!(thread.state, ThreadState::Processing);
|
||||||
|
|
||||||
@@ -2211,7 +2227,7 @@ mod tests {
|
|||||||
use crate::agent::session::{Thread, ThreadState};
|
use crate::agent::session::{Thread, ThreadState};
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
let mut thread = Thread::new(Uuid::new_v4());
|
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||||
thread.start_turn("processing");
|
thread.start_turn("processing");
|
||||||
|
|
||||||
thread.queue_message("pending-1".to_string());
|
thread.queue_message("pending-1".to_string());
|
||||||
@@ -2241,7 +2257,7 @@ mod tests {
|
|||||||
|
|
||||||
let thread_id = Uuid::new_v4();
|
let thread_id = Uuid::new_v4();
|
||||||
let session_id = Uuid::new_v4();
|
let session_id = Uuid::new_v4();
|
||||||
let mut thread = Thread::with_id(thread_id, session_id);
|
let mut thread = Thread::with_id(thread_id, session_id, None);
|
||||||
thread.start_turn("working");
|
thread.start_turn("working");
|
||||||
assert_eq!(thread.state, ThreadState::Processing);
|
assert_eq!(thread.state, ThreadState::Processing);
|
||||||
|
|
||||||
@@ -2268,7 +2284,7 @@ mod tests {
|
|||||||
|
|
||||||
let thread_id = Uuid::new_v4();
|
let thread_id = Uuid::new_v4();
|
||||||
let session_id = Uuid::new_v4();
|
let session_id = Uuid::new_v4();
|
||||||
let mut thread = Thread::with_id(thread_id, session_id);
|
let mut thread = Thread::with_id(thread_id, session_id, None);
|
||||||
thread.start_turn("working");
|
thread.start_turn("working");
|
||||||
assert_eq!(thread.state, ThreadState::Processing);
|
assert_eq!(thread.state, ThreadState::Processing);
|
||||||
|
|
||||||
|
|||||||
@@ -34,6 +34,7 @@ pub async fn setup_wasm_channels(
|
|||||||
secrets_store: &Option<Arc<dyn SecretsStore + Send + Sync>>,
|
secrets_store: &Option<Arc<dyn SecretsStore + Send + Sync>>,
|
||||||
extension_manager: Option<&Arc<ExtensionManager>>,
|
extension_manager: Option<&Arc<ExtensionManager>>,
|
||||||
database: Option<&Arc<dyn Database>>,
|
database: Option<&Arc<dyn Database>>,
|
||||||
|
registered_channel_names: &[String],
|
||||||
) -> Option<WasmChannelSetup> {
|
) -> Option<WasmChannelSetup> {
|
||||||
let runtime = match WasmChannelRuntime::new(WasmChannelRuntimeConfig::default()) {
|
let runtime = match WasmChannelRuntime::new(WasmChannelRuntimeConfig::default()) {
|
||||||
Ok(r) => Arc::new(r),
|
Ok(r) => Arc::new(r),
|
||||||
@@ -71,7 +72,46 @@ pub async fn setup_wasm_channels(
|
|||||||
let mut channels: Vec<(String, Box<dyn crate::channels::Channel>)> = Vec::new();
|
let mut channels: Vec<(String, Box<dyn crate::channels::Channel>)> = Vec::new();
|
||||||
let mut channel_names: Vec<String> = Vec::new();
|
let mut channel_names: Vec<String> = Vec::new();
|
||||||
|
|
||||||
|
// Reserved channel names that WASM modules must not claim.
|
||||||
|
// A malicious module could otherwise register as a trusted built-in
|
||||||
|
// channel and bypass cross-channel authorization checks.
|
||||||
|
// This list must cover every built-in channel name to prevent a WASM
|
||||||
|
// module from impersonating a built-in and satisfying same-channel
|
||||||
|
// approval checks.
|
||||||
|
const RESERVED_CHANNEL_NAMES: &[&str] = &[
|
||||||
|
"web",
|
||||||
|
"gateway",
|
||||||
|
"cli",
|
||||||
|
"repl",
|
||||||
|
"http",
|
||||||
|
"signal",
|
||||||
|
"slack-relay",
|
||||||
|
"secret_save",
|
||||||
|
];
|
||||||
|
|
||||||
for loaded in results.loaded {
|
for loaded in results.loaded {
|
||||||
|
let name_lower = loaded.name().to_ascii_lowercase();
|
||||||
|
if RESERVED_CHANNEL_NAMES.contains(&name_lower.as_str()) {
|
||||||
|
tracing::warn!(
|
||||||
|
channel = %loaded.name(),
|
||||||
|
"Rejected WASM channel with reserved name"
|
||||||
|
);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
// Also reject any name that collides with an already-registered
|
||||||
|
// channel to prevent a WASM module from shadowing a channel that
|
||||||
|
// was registered earlier in the startup sequence.
|
||||||
|
if registered_channel_names
|
||||||
|
.iter()
|
||||||
|
.any(|n| n.to_ascii_lowercase() == name_lower)
|
||||||
|
{
|
||||||
|
tracing::warn!(
|
||||||
|
channel = %loaded.name(),
|
||||||
|
"Rejected WASM channel that collides with already-registered channel"
|
||||||
|
);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
let (name, channel) = register_channel(
|
let (name, channel) = register_channel(
|
||||||
loaded,
|
loaded,
|
||||||
config,
|
config,
|
||||||
|
|||||||
@@ -574,7 +574,7 @@ pub async fn chat_new_thread_handler(
|
|||||||
.await;
|
.await;
|
||||||
let (thread_id, info) = {
|
let (thread_id, info) = {
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = sess.create_thread();
|
let thread = sess.create_thread(Some("web"));
|
||||||
let id = thread.id;
|
let id = thread.id;
|
||||||
let info = ThreadInfo {
|
let info = ThreadInfo {
|
||||||
id: thread.id,
|
id: thread.id,
|
||||||
@@ -593,7 +593,13 @@ pub async fn chat_new_thread_handler(
|
|||||||
// so that the subsequent loadThreads() call from the frontend sees it.
|
// so that the subsequent loadThreads() call from the frontend sees it.
|
||||||
if let Some(ref store) = state.store {
|
if let Some(ref store) = state.store {
|
||||||
match store
|
match store
|
||||||
.ensure_conversation(thread_id, "gateway", &identity.user_id, None)
|
.ensure_conversation(
|
||||||
|
thread_id,
|
||||||
|
"gateway",
|
||||||
|
&identity.user_id,
|
||||||
|
None,
|
||||||
|
Some("gateway"),
|
||||||
|
)
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
Ok(true) => {}
|
Ok(true) => {}
|
||||||
|
|||||||
@@ -2014,7 +2014,7 @@ async fn chat_new_thread_handler(
|
|||||||
let session = session_manager.get_or_create_session(&user.user_id).await;
|
let session = session_manager.get_or_create_session(&user.user_id).await;
|
||||||
let (thread_id, info) = {
|
let (thread_id, info) = {
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = sess.create_thread();
|
let thread = sess.create_thread(Some("gateway"));
|
||||||
let id = thread.id;
|
let id = thread.id;
|
||||||
let info = ThreadInfo {
|
let info = ThreadInfo {
|
||||||
id: thread.id,
|
id: thread.id,
|
||||||
@@ -2033,7 +2033,7 @@ async fn chat_new_thread_handler(
|
|||||||
// so that the subsequent loadThreads() call from the frontend sees it.
|
// so that the subsequent loadThreads() call from the frontend sees it.
|
||||||
if let Some(ref store) = state.store {
|
if let Some(ref store) = state.store {
|
||||||
match store
|
match store
|
||||||
.ensure_conversation(thread_id, "gateway", &user.user_id, None)
|
.ensure_conversation(thread_id, "gateway", &user.user_id, None, Some("gateway"))
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
Ok(true) => {}
|
Ok(true) => {}
|
||||||
|
|||||||
@@ -67,19 +67,20 @@ impl ConversationStore for LibSqlBackend {
|
|||||||
channel: &str,
|
channel: &str,
|
||||||
user_id: &str,
|
user_id: &str,
|
||||||
thread_id: Option<&str>,
|
thread_id: Option<&str>,
|
||||||
|
source_channel: Option<&str>,
|
||||||
) -> Result<bool, DatabaseError> {
|
) -> Result<bool, DatabaseError> {
|
||||||
let conn = self.connect().await?;
|
let conn = self.connect().await?;
|
||||||
let now = fmt_ts(&Utc::now());
|
let now = fmt_ts(&Utc::now());
|
||||||
let affected = conn
|
let affected = conn
|
||||||
.execute(
|
.execute(
|
||||||
r#"
|
r#"
|
||||||
INSERT INTO conversations (id, channel, user_id, thread_id, started_at, last_activity)
|
INSERT INTO conversations (id, channel, user_id, thread_id, source_channel, started_at, last_activity)
|
||||||
VALUES (?1, ?2, ?3, ?4, ?5, ?5)
|
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?6)
|
||||||
ON CONFLICT (id) DO UPDATE SET last_activity = excluded.last_activity
|
ON CONFLICT (id) DO UPDATE SET last_activity = excluded.last_activity
|
||||||
WHERE conversations.user_id = excluded.user_id
|
WHERE conversations.user_id = excluded.user_id
|
||||||
AND conversations.channel = excluded.channel
|
AND conversations.channel = excluded.channel
|
||||||
"#,
|
"#,
|
||||||
params![id.to_string(), channel, user_id, opt_text(thread_id), now],
|
params![id.to_string(), channel, user_id, opt_text(thread_id), opt_text(source_channel), now],
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||||
@@ -565,6 +566,28 @@ impl ConversationStore for LibSqlBackend {
|
|||||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||||
Ok(found.is_some())
|
Ok(found.is_some())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn get_conversation_source_channel(
|
||||||
|
&self,
|
||||||
|
conversation_id: Uuid,
|
||||||
|
) -> Result<Option<String>, DatabaseError> {
|
||||||
|
let conn = self.connect().await?;
|
||||||
|
let mut rows = conn
|
||||||
|
.query(
|
||||||
|
"SELECT source_channel FROM conversations WHERE id = ?1",
|
||||||
|
params![conversation_id.to_string()],
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||||
|
match rows
|
||||||
|
.next()
|
||||||
|
.await
|
||||||
|
.map_err(|e| DatabaseError::Query(e.to_string()))?
|
||||||
|
{
|
||||||
|
Some(row) => Ok(get_opt_text(&row, 0)),
|
||||||
|
None => Ok(None),
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
|
|||||||
@@ -981,7 +981,7 @@ mod tests {
|
|||||||
assert!(alice_stats.last_active_at.is_some());
|
assert!(alice_stats.last_active_at.is_some());
|
||||||
|
|
||||||
// Bob has no LLM calls so doesn't appear in summary stats
|
// Bob has no LLM calls so doesn't appear in summary stats
|
||||||
assert!(stats.iter().find(|s| s.user_id == "bob").is_none());
|
assert!(!stats.iter().any(|s| s.user_id == "bob"));
|
||||||
|
|
||||||
// Filter to single user
|
// Filter to single user
|
||||||
let alice_only = db.user_summary_stats(Some("alice")).await.unwrap();
|
let alice_only = db.user_summary_stats(Some("alice")).await.unwrap();
|
||||||
|
|||||||
@@ -785,6 +785,14 @@ CREATE TABLE IF NOT EXISTS api_tokens (
|
|||||||
);
|
);
|
||||||
CREATE INDEX IF NOT EXISTS idx_api_tokens_user ON api_tokens(user_id);
|
CREATE INDEX IF NOT EXISTS idx_api_tokens_user ON api_tokens(user_id);
|
||||||
CREATE INDEX IF NOT EXISTS idx_api_tokens_hash ON api_tokens(token_hash);
|
CREATE INDEX IF NOT EXISTS idx_api_tokens_hash ON api_tokens(token_hash);
|
||||||
|
"#,
|
||||||
|
),
|
||||||
|
(
|
||||||
|
15,
|
||||||
|
"conversation_source_channel",
|
||||||
|
// Add source_channel to conversations for cross-channel approval authorization.
|
||||||
|
r#"
|
||||||
|
ALTER TABLE conversations ADD COLUMN source_channel TEXT;
|
||||||
"#,
|
"#,
|
||||||
),
|
),
|
||||||
];
|
];
|
||||||
|
|||||||
@@ -373,6 +373,7 @@ pub trait ConversationStore: Send + Sync {
|
|||||||
channel: &str,
|
channel: &str,
|
||||||
user_id: &str,
|
user_id: &str,
|
||||||
thread_id: Option<&str>,
|
thread_id: Option<&str>,
|
||||||
|
source_channel: Option<&str>,
|
||||||
) -> Result<bool, DatabaseError>;
|
) -> Result<bool, DatabaseError>;
|
||||||
async fn list_conversations_with_preview(
|
async fn list_conversations_with_preview(
|
||||||
&self,
|
&self,
|
||||||
@@ -431,6 +432,11 @@ pub trait ConversationStore: Send + Sync {
|
|||||||
conversation_id: Uuid,
|
conversation_id: Uuid,
|
||||||
user_id: &str,
|
user_id: &str,
|
||||||
) -> Result<bool, DatabaseError>;
|
) -> Result<bool, DatabaseError>;
|
||||||
|
/// Get the source_channel for a conversation (the channel that created it).
|
||||||
|
async fn get_conversation_source_channel(
|
||||||
|
&self,
|
||||||
|
conversation_id: Uuid,
|
||||||
|
) -> Result<Option<String>, DatabaseError>;
|
||||||
}
|
}
|
||||||
|
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
|
|||||||
+11
-1
@@ -99,9 +99,10 @@ impl ConversationStore for PgBackend {
|
|||||||
channel: &str,
|
channel: &str,
|
||||||
user_id: &str,
|
user_id: &str,
|
||||||
thread_id: Option<&str>,
|
thread_id: Option<&str>,
|
||||||
|
source_channel: Option<&str>,
|
||||||
) -> Result<bool, DatabaseError> {
|
) -> Result<bool, DatabaseError> {
|
||||||
self.store
|
self.store
|
||||||
.ensure_conversation(id, channel, user_id, thread_id)
|
.ensure_conversation(id, channel, user_id, thread_id, source_channel)
|
||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -212,6 +213,15 @@ impl ConversationStore for PgBackend {
|
|||||||
.conversation_belongs_to_user(conversation_id, user_id)
|
.conversation_belongs_to_user(conversation_id, user_id)
|
||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn get_conversation_source_channel(
|
||||||
|
&self,
|
||||||
|
conversation_id: Uuid,
|
||||||
|
) -> Result<Option<String>, DatabaseError> {
|
||||||
|
self.store
|
||||||
|
.get_conversation_source_channel(conversation_id)
|
||||||
|
.await
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// ==================== JobStore ====================
|
// ==================== JobStore ====================
|
||||||
|
|||||||
+19
-3
@@ -1581,19 +1581,20 @@ impl Store {
|
|||||||
channel: &str,
|
channel: &str,
|
||||||
user_id: &str,
|
user_id: &str,
|
||||||
thread_id: Option<&str>,
|
thread_id: Option<&str>,
|
||||||
|
source_channel: Option<&str>,
|
||||||
) -> Result<bool, DatabaseError> {
|
) -> Result<bool, DatabaseError> {
|
||||||
let conn = self.conn().await?;
|
let conn = self.conn().await?;
|
||||||
let affected = conn
|
let affected = conn
|
||||||
.execute(
|
.execute(
|
||||||
r#"
|
r#"
|
||||||
INSERT INTO conversations (id, channel, user_id, thread_id)
|
INSERT INTO conversations (id, channel, user_id, thread_id, source_channel)
|
||||||
VALUES ($1, $2, $3, $4)
|
VALUES ($1, $2, $3, $4, $5)
|
||||||
ON CONFLICT (id) DO UPDATE
|
ON CONFLICT (id) DO UPDATE
|
||||||
SET last_activity = NOW()
|
SET last_activity = NOW()
|
||||||
WHERE conversations.user_id = EXCLUDED.user_id
|
WHERE conversations.user_id = EXCLUDED.user_id
|
||||||
AND conversations.channel = EXCLUDED.channel
|
AND conversations.channel = EXCLUDED.channel
|
||||||
"#,
|
"#,
|
||||||
&[&id, &channel, &user_id, &thread_id],
|
&[&id, &channel, &user_id, &thread_id, &source_channel],
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
Ok(affected > 0)
|
Ok(affected > 0)
|
||||||
@@ -1892,6 +1893,21 @@ impl Store {
|
|||||||
Ok(row.is_some())
|
Ok(row.is_some())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Get the source_channel for a conversation.
|
||||||
|
pub async fn get_conversation_source_channel(
|
||||||
|
&self,
|
||||||
|
conversation_id: Uuid,
|
||||||
|
) -> Result<Option<String>, DatabaseError> {
|
||||||
|
let conn = self.conn().await?;
|
||||||
|
let row = conn
|
||||||
|
.query_opt(
|
||||||
|
"SELECT source_channel FROM conversations WHERE id = $1",
|
||||||
|
&[&conversation_id],
|
||||||
|
)
|
||||||
|
.await?;
|
||||||
|
Ok(row.and_then(|r| r.get::<_, Option<String>>(0)))
|
||||||
|
}
|
||||||
|
|
||||||
/// Load messages for a conversation with cursor-based pagination.
|
/// Load messages for a conversation with cursor-based pagination.
|
||||||
///
|
///
|
||||||
/// Returns `(messages_oldest_first, has_more)`.
|
/// Returns `(messages_oldest_first, has_more)`.
|
||||||
|
|||||||
@@ -449,6 +449,7 @@ async fn async_main() -> anyhow::Result<()> {
|
|||||||
&components.secrets_store,
|
&components.secrets_store,
|
||||||
components.extension_manager.as_ref(),
|
components.extension_manager.as_ref(),
|
||||||
components.db.as_ref(),
|
components.db.as_ref(),
|
||||||
|
&channel_names,
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
|
|
||||||
|
|||||||
+1
-1
@@ -278,7 +278,7 @@ impl TenantScope {
|
|||||||
thread_id: Option<&str>,
|
thread_id: Option<&str>,
|
||||||
) -> Result<bool, DatabaseError> {
|
) -> Result<bool, DatabaseError> {
|
||||||
self.inner
|
self.inner
|
||||||
.ensure_conversation(id, channel, &self.user_id, thread_id)
|
.ensure_conversation(id, channel, &self.user_id, thread_id, None)
|
||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+3
-3
@@ -753,7 +753,7 @@ mod tests {
|
|||||||
|
|
||||||
// ensure_conversation should create the row.
|
// ensure_conversation should create the row.
|
||||||
assert!(
|
assert!(
|
||||||
db.ensure_conversation(conv_id, "web", "carol", None)
|
db.ensure_conversation(conv_id, "web", "carol", None, Some("web"))
|
||||||
.await
|
.await
|
||||||
.expect("ensure first"),
|
.expect("ensure first"),
|
||||||
"first ensure_conversation should create the row"
|
"first ensure_conversation should create the row"
|
||||||
@@ -761,7 +761,7 @@ mod tests {
|
|||||||
|
|
||||||
// Calling again with the same ID should not error.
|
// Calling again with the same ID should not error.
|
||||||
assert!(
|
assert!(
|
||||||
db.ensure_conversation(conv_id, "web", "carol", None)
|
db.ensure_conversation(conv_id, "web", "carol", None, Some("web"))
|
||||||
.await
|
.await
|
||||||
.expect("ensure second (idempotent)"),
|
.expect("ensure second (idempotent)"),
|
||||||
"second ensure_conversation should touch owned row"
|
"second ensure_conversation should touch owned row"
|
||||||
@@ -806,7 +806,7 @@ mod tests {
|
|||||||
tokio::time::sleep(std::time::Duration::from_millis(25)).await;
|
tokio::time::sleep(std::time::Duration::from_millis(25)).await;
|
||||||
|
|
||||||
assert!(
|
assert!(
|
||||||
!db.ensure_conversation(conv_id, "web", "mallory", None)
|
!db.ensure_conversation(conv_id, "web", "mallory", None, None)
|
||||||
.await
|
.await
|
||||||
.expect("foreign ensure should not error"),
|
.expect("foreign ensure should not error"),
|
||||||
"foreign ensure_conversation should report not ensured"
|
"foreign ensure_conversation should report not ensured"
|
||||||
|
|||||||
@@ -48,7 +48,13 @@ mod tests {
|
|||||||
let store = rig.database();
|
let store = rig.database();
|
||||||
assert!(
|
assert!(
|
||||||
store
|
store
|
||||||
.ensure_conversation(foreign_thread_id, "gateway", "victim-user", None)
|
.ensure_conversation(
|
||||||
|
foreign_thread_id,
|
||||||
|
"gateway",
|
||||||
|
"victim-user",
|
||||||
|
None,
|
||||||
|
Some("gateway")
|
||||||
|
)
|
||||||
.await
|
.await
|
||||||
.expect("failed to create victim conversation"),
|
.expect("failed to create victim conversation"),
|
||||||
"test setup failed: victim conversation was not created"
|
"test setup failed: victim conversation was not created"
|
||||||
|
|||||||
Reference in New Issue
Block a user