feat(channels): add threading support to message channels

Add optional thread_ts field to ChannelMessage and SendMessage for
platform-specific threading (e.g. Slack threads, Discord threads).

- ChannelMessage.thread_ts captures incoming thread context
- SendMessage.thread_ts propagates thread context to replies
- SendMessage::in_thread() builder for fluent API
- Slack: send with thread_ts, capture ts from incoming messages
- All reply paths in runtime preserve thread context via in_thread()
- All other channels initialize thread_ts: None (forward-compatible)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Daniel Willitzer 2026-02-19 00:42:29 -08:00 committed by Chummy
parent adc998429e
commit 9afe4f28e7
18 changed files with 63 additions and 8 deletions

View file

@ -47,6 +47,7 @@ impl Channel for CliChannel {
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs(), .as_secs(),
thread_ts: None,
}; };
if tx.send(msg).await.is_err() { if tx.send(msg).await.is_err() {
@ -107,6 +108,7 @@ mod tests {
content: "hello".into(), content: "hello".into(),
channel: "cli".into(), channel: "cli".into(),
timestamp: 1_234_567_890, timestamp: 1_234_567_890,
thread_ts: None,
}; };
assert_eq!(msg.id, "test-id"); assert_eq!(msg.id, "test-id");
assert_eq!(msg.sender, "user"); assert_eq!(msg.sender, "user");
@ -125,6 +127,7 @@ mod tests {
content: "c".into(), content: "c".into(),
channel: "ch".into(), channel: "ch".into(),
timestamp: 0, timestamp: 0,
thread_ts: None,
}; };
let cloned = msg.clone(); let cloned = msg.clone();
assert_eq!(cloned.id, msg.id); assert_eq!(cloned.id, msg.id);

View file

@ -274,6 +274,7 @@ impl Channel for DingTalkChannel {
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs(), .as_secs(),
thread_ts: None,
}; };
if tx.send(channel_msg).await.is_err() { if tx.send(channel_msg).await.is_err() {

View file

@ -413,6 +413,7 @@ impl Channel for DiscordChannel {
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs(), .as_secs(),
thread_ts: None,
}; };
if tx.send(channel_msg).await.is_err() { if tx.send(channel_msg).await.is_err() {

View file

@ -443,6 +443,7 @@ impl EmailChannel {
content: email.content, content: email.content,
channel: "email".to_string(), channel: "email".to_string(),
timestamp: email.timestamp, timestamp: email.timestamp,
thread_ts: None,
}; };
if tx.send(msg).await.is_err() { if tx.send(msg).await.is_err() {

View file

@ -231,6 +231,7 @@ end tell"#
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs(), .as_secs(),
thread_ts: None,
}; };
if tx.send(msg).await.is_err() { if tx.send(msg).await.is_err() {

View file

@ -574,6 +574,7 @@ impl Channel for IrcChannel {
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs(), .as_secs(),
thread_ts: None,
}; };
if tx.send(channel_msg).await.is_err() { if tx.send(channel_msg).await.is_err() {

View file

@ -474,6 +474,7 @@ impl LarkChannel {
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs(), .as_secs(),
thread_ts: None,
}; };
tracing::debug!("Lark WS: message in {}", lark_msg.chat_id); tracing::debug!("Lark WS: message in {}", lark_msg.chat_id);
@ -635,6 +636,7 @@ impl LarkChannel {
content: text, content: text,
channel: "lark".to_string(), channel: "lark".to_string(),
timestamp, timestamp,
thread_ts: None,
}); });
messages messages

View file

@ -178,6 +178,7 @@ impl LinqChannel {
content, content,
channel: "linq".to_string(), channel: "linq".to_string(),
timestamp, timestamp,
thread_ts: None,
}); });
messages messages

View file

@ -596,6 +596,7 @@ impl Channel for MatrixChannel {
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs(), .as_secs(),
thread_ts: None,
}; };
let _ = tx.send(msg).await; let _ = tx.send(msg).await;

View file

@ -321,6 +321,7 @@ impl MattermostChannel {
channel: "mattermost".to_string(), channel: "mattermost".to_string(),
#[allow(clippy::cast_sign_loss)] #[allow(clippy::cast_sign_loss)]
timestamp: (create_at / 1000) as u64, timestamp: (create_at / 1000) as u64,
thread_ts: None,
}) })
} }
} }

View file

@ -442,7 +442,7 @@ async fn handle_runtime_command_if_needed(
}; };
if let Err(err) = channel if let Err(err) = channel
.send(&SendMessage::new(response, &msg.reply_target)) .send(&SendMessage::new(response, &msg.reply_target).in_thread(msg.thread_ts.clone()))
.await .await
{ {
tracing::warn!( tracing::warn!(
@ -592,7 +592,7 @@ async fn process_channel_message(ctx: Arc<ChannelRuntimeContext>, msg: traits::C
); );
if let Some(channel) = target_channel.as_ref() { if let Some(channel) = target_channel.as_ref() {
let _ = channel let _ = channel
.send(&SendMessage::new(message, &msg.reply_target)) .send(&SendMessage::new(message, &msg.reply_target).in_thread(msg.thread_ts.clone()))
.await; .await;
} }
return; return;
@ -658,7 +658,7 @@ async fn process_channel_message(ctx: Arc<ChannelRuntimeContext>, msg: traits::C
let draft_message_id = if use_streaming { let draft_message_id = if use_streaming {
if let Some(channel) = target_channel.as_ref() { if let Some(channel) = target_channel.as_ref() {
match channel match channel
.send_draft(&SendMessage::new("...", &msg.reply_target)) .send_draft(&SendMessage::new("...", &msg.reply_target).in_thread(msg.thread_ts.clone()))
.await .await
{ {
Ok(id) => id, Ok(id) => id,
@ -769,11 +769,11 @@ async fn process_channel_message(ctx: Arc<ChannelRuntimeContext>, msg: traits::C
{ {
tracing::warn!("Failed to finalize draft: {e}; sending as new message"); tracing::warn!("Failed to finalize draft: {e}; sending as new message");
let _ = channel let _ = channel
.send(&SendMessage::new(&response, &msg.reply_target)) .send(&SendMessage::new(&response, &msg.reply_target).in_thread(msg.thread_ts.clone()))
.await; .await;
} }
} else if let Err(e) = channel } else if let Err(e) = channel
.send(&SendMessage::new(response, &msg.reply_target)) .send(&SendMessage::new(response, &msg.reply_target).in_thread(msg.thread_ts.clone()))
.await .await
{ {
eprintln!(" ❌ Failed to reply on {}: {e}", channel.name()); eprintln!(" ❌ Failed to reply on {}: {e}", channel.name());
@ -795,7 +795,7 @@ async fn process_channel_message(ctx: Arc<ChannelRuntimeContext>, msg: traits::C
.send(&SendMessage::new( .send(&SendMessage::new(
format!("⚠️ Error: {e}"), format!("⚠️ Error: {e}"),
&msg.reply_target, &msg.reply_target,
)) ).in_thread(msg.thread_ts.clone()))
.await; .await;
} }
} }
@ -816,7 +816,7 @@ async fn process_channel_message(ctx: Arc<ChannelRuntimeContext>, msg: traits::C
.await; .await;
} else { } else {
let _ = channel let _ = channel
.send(&SendMessage::new(error_text, &msg.reply_target)) .send(&SendMessage::new(error_text, &msg.reply_target).in_thread(msg.thread_ts.clone()))
.await; .await;
} }
} }
@ -2350,6 +2350,7 @@ mod tests {
content: "What is the BTC price now?".to_string(), content: "What is the BTC price now?".to_string(),
channel: "test-channel".to_string(), channel: "test-channel".to_string(),
timestamp: 1, timestamp: 1,
thread_ts: None,
}, },
) )
.await; .await;
@ -2403,6 +2404,7 @@ mod tests {
content: "What is the BTC price now?".to_string(), content: "What is the BTC price now?".to_string(),
channel: "test-channel".to_string(), channel: "test-channel".to_string(),
timestamp: 2, timestamp: 2,
thread_ts: None,
}, },
) )
.await; .await;
@ -2465,6 +2467,7 @@ mod tests {
content: "/models openrouter".to_string(), content: "/models openrouter".to_string(),
channel: "telegram".to_string(), channel: "telegram".to_string(),
timestamp: 1, timestamp: 1,
thread_ts: None,
}, },
) )
.await; .await;
@ -2548,6 +2551,7 @@ mod tests {
content: "hello routed provider".to_string(), content: "hello routed provider".to_string(),
channel: "telegram".to_string(), channel: "telegram".to_string(),
timestamp: 2, timestamp: 2,
thread_ts: None,
}, },
) )
.await; .await;
@ -2607,6 +2611,7 @@ mod tests {
content: "Loop until done".to_string(), content: "Loop until done".to_string(),
channel: "test-channel".to_string(), channel: "test-channel".to_string(),
timestamp: 1, timestamp: 1,
thread_ts: None,
}, },
) )
.await; .await;
@ -2661,6 +2666,7 @@ mod tests {
content: "Loop forever".to_string(), content: "Loop forever".to_string(),
channel: "test-channel".to_string(), channel: "test-channel".to_string(),
timestamp: 2, timestamp: 2,
thread_ts: None,
}, },
) )
.await; .await;
@ -2765,6 +2771,7 @@ mod tests {
content: "hello".to_string(), content: "hello".to_string(),
channel: "test-channel".to_string(), channel: "test-channel".to_string(),
timestamp: 1, timestamp: 1,
thread_ts: None,
}) })
.await .await
.unwrap(); .unwrap();
@ -2775,6 +2782,7 @@ mod tests {
content: "world".to_string(), content: "world".to_string(),
channel: "test-channel".to_string(), channel: "test-channel".to_string(),
timestamp: 2, timestamp: 2,
thread_ts: None,
}) })
.await .await
.unwrap(); .unwrap();
@ -2837,6 +2845,7 @@ mod tests {
content: "hello".to_string(), content: "hello".to_string(),
channel: "test-channel".to_string(), channel: "test-channel".to_string(),
timestamp: 1, timestamp: 1,
thread_ts: None,
}, },
) )
.await; .await;
@ -3097,6 +3106,7 @@ mod tests {
content: "hello".into(), content: "hello".into(),
channel: "slack".into(), channel: "slack".into(),
timestamp: 1, timestamp: 1,
thread_ts: None,
}; };
assert_eq!(conversation_memory_key(&msg), "slack_U123_msg_abc123"); assert_eq!(conversation_memory_key(&msg), "slack_U123_msg_abc123");
@ -3111,6 +3121,7 @@ mod tests {
content: "first".into(), content: "first".into(),
channel: "slack".into(), channel: "slack".into(),
timestamp: 1, timestamp: 1,
thread_ts: None,
}; };
let msg2 = traits::ChannelMessage { let msg2 = traits::ChannelMessage {
id: "msg_2".into(), id: "msg_2".into(),
@ -3119,6 +3130,7 @@ mod tests {
content: "second".into(), content: "second".into(),
channel: "slack".into(), channel: "slack".into(),
timestamp: 2, timestamp: 2,
thread_ts: None,
}; };
assert_ne!( assert_ne!(
@ -3139,6 +3151,7 @@ mod tests {
content: "I'm Paul".into(), content: "I'm Paul".into(),
channel: "slack".into(), channel: "slack".into(),
timestamp: 1, timestamp: 1,
thread_ts: None,
}; };
let msg2 = traits::ChannelMessage { let msg2 = traits::ChannelMessage {
id: "msg_2".into(), id: "msg_2".into(),
@ -3147,6 +3160,7 @@ mod tests {
content: "I'm 45".into(), content: "I'm 45".into(),
channel: "slack".into(), channel: "slack".into(),
timestamp: 2, timestamp: 2,
thread_ts: None,
}; };
mem.store( mem.store(
@ -3228,6 +3242,7 @@ mod tests {
content: "hello".to_string(), content: "hello".to_string(),
channel: "test-channel".to_string(), channel: "test-channel".to_string(),
timestamp: 1, timestamp: 1,
thread_ts: None,
}, },
) )
.await; .await;
@ -3241,6 +3256,7 @@ mod tests {
content: "follow up".to_string(), content: "follow up".to_string(),
channel: "test-channel".to_string(), channel: "test-channel".to_string(),
timestamp: 2, timestamp: 2,
thread_ts: None,
}, },
) )
.await; .await;

View file

@ -377,6 +377,7 @@ impl Channel for QQChannel {
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs(), .as_secs(),
thread_ts: None,
}; };
if tx.send(channel_msg).await.is_err() { if tx.send(channel_msg).await.is_err() {
@ -415,6 +416,7 @@ impl Channel for QQChannel {
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs(), .as_secs(),
thread_ts: None,
}; };
if tx.send(channel_msg).await.is_err() { if tx.send(channel_msg).await.is_err() {

View file

@ -265,6 +265,7 @@ impl SignalChannel {
content: text.to_string(), content: text.to_string(),
channel: "signal".to_string(), channel: "signal".to_string(),
timestamp: timestamp / 1000, // millis → secs timestamp: timestamp / 1000, // millis → secs
thread_ts: None,
}) })
} }
} }

View file

@ -54,11 +54,15 @@ impl Channel for SlackChannel {
} }
async fn send(&self, message: &SendMessage) -> anyhow::Result<()> { async fn send(&self, message: &SendMessage) -> anyhow::Result<()> {
let body = serde_json::json!({ let mut body = serde_json::json!({
"channel": message.recipient, "channel": message.recipient,
"text": message.content "text": message.content
}); });
if let Some(ref ts) = message.thread_ts {
body["thread_ts"] = serde_json::json!(ts);
}
let resp = self let resp = self
.http_client() .http_client()
.post("https://slack.com/api/chat.postMessage") .post("https://slack.com/api/chat.postMessage")
@ -170,6 +174,7 @@ impl Channel for SlackChannel {
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs(), .as_secs(),
thread_ts: Some(ts.to_string()),
}; };
if tx.send(channel_msg).await.is_err() { if tx.send(channel_msg).await.is_err() {

View file

@ -819,6 +819,7 @@ Allowlist Telegram username (without '@') or numeric user ID.",
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs(), .as_secs(),
thread_ts: None,
}) })
} }

View file

@ -9,6 +9,9 @@ pub struct ChannelMessage {
pub content: String, pub content: String,
pub channel: String, pub channel: String,
pub timestamp: u64, pub timestamp: u64,
/// Platform thread identifier (e.g. Slack `ts`, Discord thread ID).
/// When set, replies should be posted as threaded responses.
pub thread_ts: Option<String>,
} }
/// Message to send through a channel /// Message to send through a channel
@ -17,6 +20,8 @@ pub struct SendMessage {
pub content: String, pub content: String,
pub recipient: String, pub recipient: String,
pub subject: Option<String>, pub subject: Option<String>,
/// Platform thread identifier for threaded replies (e.g. Slack `thread_ts`).
pub thread_ts: Option<String>,
} }
impl SendMessage { impl SendMessage {
@ -26,6 +31,7 @@ impl SendMessage {
content: content.into(), content: content.into(),
recipient: recipient.into(), recipient: recipient.into(),
subject: None, subject: None,
thread_ts: None,
} }
} }
@ -39,8 +45,15 @@ impl SendMessage {
content: content.into(), content: content.into(),
recipient: recipient.into(), recipient: recipient.into(),
subject: Some(subject.into()), subject: Some(subject.into()),
thread_ts: None,
} }
} }
/// Set the thread identifier for threaded replies.
pub fn in_thread(mut self, thread_ts: Option<String>) -> Self {
self.thread_ts = thread_ts;
self
}
} }
/// Core channel trait — implement for any messaging platform /// Core channel trait — implement for any messaging platform
@ -129,6 +142,7 @@ mod tests {
content: "hello".into(), content: "hello".into(),
channel: "dummy".into(), channel: "dummy".into(),
timestamp: 123, timestamp: 123,
thread_ts: None,
}) })
.await .await
.map_err(|e| anyhow::anyhow!(e.to_string())) .map_err(|e| anyhow::anyhow!(e.to_string()))
@ -144,6 +158,7 @@ mod tests {
content: "ping".into(), content: "ping".into(),
channel: "dummy".into(), channel: "dummy".into(),
timestamp: 999, timestamp: 999,
thread_ts: None,
}; };
let cloned = message.clone(); let cloned = message.clone();

View file

@ -140,6 +140,7 @@ impl WhatsAppChannel {
content, content,
channel: "whatsapp".to_string(), channel: "whatsapp".to_string(),
timestamp, timestamp,
thread_ts: None,
}); });
} }
} }

View file

@ -1450,6 +1450,7 @@ mod tests {
content: "hello".into(), content: "hello".into(),
channel: "whatsapp".into(), channel: "whatsapp".into(),
timestamp: 1, timestamp: 1,
thread_ts: None,
}; };
let key = whatsapp_memory_key(&msg); let key = whatsapp_memory_key(&msg);