From 88992935278327381d144e6d7f78ae60693f7ccc Mon Sep 17 00:00:00 2001 From: s-salamatov <55296341+s-salamatov@users.noreply.github.com> Date: Wed, 3 Jun 2026 05:56:04 +0000 Subject: [PATCH] feat(channels): add activity log visibility settings --- crates/channels/src/activity_log.rs | 59 ++ crates/channels/src/config_view.rs | 20 +- crates/channels/src/lib.rs | 2 + crates/channels/src/plugin.rs | 33 +- .../src/plugin/tests/binding_tests.rs | 35 ++ crates/chat/src/channel_logbook.rs | 209 +++++++ crates/chat/src/channels.rs | 231 ++++---- crates/chat/src/lib.rs | 1 + crates/chat/src/runtime.rs | 11 +- crates/config/src/template.rs | 11 + crates/discord/src/commands.rs | 4 + crates/discord/src/handler/implementation.rs | 2 + crates/gateway/src/channel_agent_tools.rs | 254 ++++++++- crates/gateway/src/channel_events.rs | 122 +++- .../channel_events/commands/attachments.rs | 19 +- .../src/channel_events/commands/dispatch.rs | 5 +- .../src/channel_events/commands/mod.rs | 2 +- .../commands/session_handlers.rs | 29 +- crates/gateway/src/channel_events/dispatch.rs | 20 +- crates/gateway/src/channel_events/tests.rs | 539 +++++++++++++++++- crates/gateway/src/chat.rs | 13 +- crates/gateway/src/session/tests.rs | 2 + crates/gateway/src/state.rs | 22 +- crates/matrix/src/handler/implementation.rs | 4 + crates/msteams/src/plugin.rs | 2 + crates/nostr/src/bus.rs | 2 + crates/signal/src/inbound.rs | 2 + crates/slack/src/socket.rs | 6 + crates/slack/src/webhook.rs | 6 + crates/telegram/src/config.rs | 65 ++- crates/telegram/src/handlers/callbacks.rs | 2 + .../telegram/src/handlers/implementation.rs | 8 + crates/telephony/src/plugin.rs | 2 + crates/web/ui/src/types/channel.ts | 8 + crates/whatsapp/src/handlers.rs | 4 + docs/src/telegram.md | 15 +- 36 files changed, 1560 insertions(+), 211 deletions(-) create mode 100644 crates/channels/src/activity_log.rs create mode 100644 crates/chat/src/channel_logbook.rs diff --git a/crates/channels/src/activity_log.rs b/crates/channels/src/activity_log.rs new file mode 100644 index 0000000000..c34a1c75c0 --- /dev/null +++ b/crates/channels/src/activity_log.rs @@ -0,0 +1,59 @@ +//! User-facing channel Activity log visibility. + +/// User-facing activity log visibility for channel replies. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ActivityLogMode { + /// Show all buffered activity entries. + #[default] + All, + /// Show failed tool/activity entries only. + ErrorsOnly, + /// Do not append or send activity log entries. + Off, +} + +impl ActivityLogMode { + pub fn includes(self, kind: ChannelStatusLogKind) -> bool { + match self { + Self::All => true, + Self::ErrorsOnly => kind == ChannelStatusLogKind::Error, + Self::Off => false, + } + } + + pub fn is_all(&self) -> bool { + *self == Self::All + } +} + +/// Type of buffered channel activity entry. +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ChannelStatusLogKind { + Info, + Error, +} + +/// Buffered activity entry appended to channel replies when enabled. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct ChannelStatusLogEntry { + pub kind: ChannelStatusLogKind, + pub message: String, +} + +impl ChannelStatusLogEntry { + pub fn info(message: impl Into) -> Self { + Self { + kind: ChannelStatusLogKind::Info, + message: message.into(), + } + } + + pub fn error(message: impl Into) -> Self { + Self { + kind: ChannelStatusLogKind::Error, + message: message.into(), + } + } +} diff --git a/crates/channels/src/config_view.rs b/crates/channels/src/config_view.rs index 43a1ae1678..97ad836e73 100644 --- a/crates/channels/src/config_view.rs +++ b/crates/channels/src/config_view.rs @@ -1,4 +1,7 @@ -use crate::gating::{DmPolicy, GroupPolicy}; +use crate::{ + activity_log::ActivityLogMode, + gating::{DmPolicy, GroupPolicy}, +}; /// Typed read-only view of common channel account config fields. /// @@ -27,6 +30,11 @@ pub trait ChannelConfigView: Send + Sync + std::fmt::Debug { /// Provider name associated with the model. fn model_provider(&self) -> Option<&str>; + /// Default activity log visibility for this channel account. + fn activity_log(&self) -> ActivityLogMode { + ActivityLogMode::All + } + /// Default agent id for this channel account. fn agent_id(&self) -> Option<&str> { None @@ -59,6 +67,16 @@ pub trait ChannelConfigView: Send + Sync + std::fmt::Debug { None } + /// Activity log override for a specific channel/chat ID. + fn channel_activity_log(&self, _channel_id: &str) -> Option { + None + } + + /// Activity log override for a specific user. + fn user_activity_log(&self, _user_id: &str) -> Option { + None + } + /// Agent override for a specific user. fn user_agent_id(&self, _user_id: &str) -> Option<&str> { None diff --git a/crates/channels/src/lib.rs b/crates/channels/src/lib.rs index 78d09654d2..19c7e5d568 100644 --- a/crates/channels/src/lib.rs +++ b/crates/channels/src/lib.rs @@ -4,6 +4,7 @@ //! ChannelPlugin trait with sub-traits for config, auth, inbound/outbound //! messaging, status, and gateway lifecycle. +pub mod activity_log; pub mod channel_webhook_middleware; pub mod commands; pub mod config_view; @@ -18,6 +19,7 @@ pub mod registry; pub mod store; pub use { + activity_log::{ActivityLogMode, ChannelStatusLogEntry, ChannelStatusLogKind}, channel_webhook_middleware::{ ChannelWebhookDedupeResult, ChannelWebhookRatePolicy, ChannelWebhookRejection, ChannelWebhookVerifier, TimestampGuard, VerifiedChannelWebhook, diff --git a/crates/channels/src/plugin.rs b/crates/channels/src/plugin.rs index f805e52d98..b168e383d0 100644 --- a/crates/channels/src/plugin.rs +++ b/crates/channels/src/plugin.rs @@ -1,6 +1,7 @@ use std::sync::Arc; use { + crate::activity_log::ActivityLogMode, async_trait::async_trait, moltis_common::{hooks::ChannelBinding, types::ReplyPayload}, tokio::sync::mpsc, @@ -645,6 +646,16 @@ pub struct ChannelReplyTarget { /// top-level chat. #[serde(default, skip_serializing_if = "Option::is_none")] pub thread_id: Option, + /// User/sender ID associated with this target, when known. + /// + /// This is persisted with channel bindings so later web-originated replies + /// can re-evaluate per-user channel settings without storing stale effective + /// runtime decisions. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub sender_id: Option, + /// Effective activity log visibility for this reply target. + #[serde(default, skip_serializing_if = "ActivityLogMode::is_all")] + pub activity_log: ActivityLogMode, } impl ChannelReplyTarget { @@ -659,6 +670,16 @@ impl ChannelReplyTarget { None => std::borrow::Cow::Borrowed(&self.chat_id), } } + + /// Returns the route data safe to persist in session metadata. + /// + /// Effective settings such as `activity_log` are intentionally reset so + /// active channel-bound sessions pick up later config changes. + pub fn for_persistence(&self) -> Self { + let mut target = self.clone(); + target.activity_log = ActivityLogMode::All; + target + } } impl From<&ChannelReplyTarget> for ChannelBinding { @@ -671,7 +692,7 @@ impl From<&ChannelReplyTarget> for ChannelBinding { account_id: Some(target.account_id.clone()), chat_id: Some(target.chat_id.clone()), chat_type: target.channel_type.classify_chat(&target.chat_id), - sender_id: None, + sender_id: target.sender_id.clone(), } } } @@ -1101,6 +1122,8 @@ mod tests { chat_id: "42".into(), message_id: None, thread_id: None, + sender_id: None, + activity_log: ActivityLogMode::All, }; assert!(!sink.update_location(&target, 48.8566, 2.3522).await); } @@ -1113,6 +1136,8 @@ mod tests { chat_id: "12345".into(), message_id: None, thread_id: None, + sender_id: None, + activity_log: ActivityLogMode::All, }; assert_eq!(target.outbound_to().as_ref(), "12345"); } @@ -1125,6 +1150,8 @@ mod tests { chat_id: "-100999".into(), message_id: None, thread_id: Some("42".into()), + sender_id: None, + activity_log: ActivityLogMode::All, }; assert_eq!(target.outbound_to().as_ref(), "-100999:42"); } @@ -1137,6 +1164,8 @@ mod tests { chat_id: "-100999".into(), message_id: None, thread_id: Some("42".into()), + sender_id: None, + activity_log: ActivityLogMode::All, }; let json = serde_json::to_string(&target).unwrap(); assert!(json.contains("\"thread_id\":\"42\"")); @@ -1438,6 +1467,8 @@ mod tests { chat_id: "-100999".into(), message_id: Some("7".into()), thread_id: Some("42".into()), + sender_id: None, + activity_log: ActivityLogMode::All, }; let binding: ChannelBinding = (&target).into(); diff --git a/crates/channels/src/plugin/tests/binding_tests.rs b/crates/channels/src/plugin/tests/binding_tests.rs index 786ad3f1f0..45cf6913b8 100644 --- a/crates/channels/src/plugin/tests/binding_tests.rs +++ b/crates/channels/src/plugin/tests/binding_tests.rs @@ -26,6 +26,8 @@ fn resolve_session_channel_binding_extracts_channel_target() { chat_id: "-100123".into(), message_id: Some("11".into()), thread_id: None, + sender_id: Some("user-123".into()), + activity_log: ActivityLogMode::All, }) .unwrap_or_else(|error| panic!("serialize binding: {error}")); @@ -38,6 +40,7 @@ fn resolve_session_channel_binding_extracts_channel_target() { assert_eq!(binding.account_id.as_deref(), Some("bot-main")); assert_eq!(binding.chat_id.as_deref(), Some("-100123")); assert_eq!(binding.chat_type.as_deref(), Some("channel_or_supergroup")); + assert_eq!(binding.sender_id.as_deref(), Some("user-123")); } #[test] @@ -47,3 +50,35 @@ fn resolve_session_channel_binding_returns_error_for_invalid_json() { .unwrap_or_else(|| panic!("invalid binding json should fail")); assert!(error.is_syntax()); } + +#[test] +fn channel_reply_target_defaults_activity_log_to_all() { + let target: ChannelReplyTarget = serde_json::from_value(serde_json::json!({ + "channel_type": "telegram", + "account_id": "bot1", + "chat_id": "123", + "message_id": "42" + })) + .unwrap(); + + assert_eq!(target.activity_log, ActivityLogMode::All); + let serialized = serde_json::to_value(&target).unwrap(); + assert!(serialized.get("activity_log").is_none()); +} + +#[test] +fn reply_target_persistence_keeps_sender_but_not_activity_snapshot() { + let target = ChannelReplyTarget { + channel_type: ChannelType::Telegram, + account_id: "bot1".into(), + chat_id: "-100999".into(), + message_id: Some("7".into()), + thread_id: None, + sender_id: Some("123".into()), + activity_log: ActivityLogMode::Off, + }; + + let persisted = serde_json::to_value(target.for_persistence()).unwrap(); + assert_eq!(persisted["sender_id"], "123"); + assert!(persisted.get("activity_log").is_none()); +} diff --git a/crates/chat/src/channel_logbook.rs b/crates/chat/src/channel_logbook.rs new file mode 100644 index 0000000000..ac0d62919a --- /dev/null +++ b/crates/chat/src/channel_logbook.rs @@ -0,0 +1,209 @@ +use std::sync::Arc; + +use { + moltis_channels::{ + ActivityLogMode, ChannelReplyTarget, ChannelStatusLogEntry, ChannelType, + plugin::ChannelOutbound, + }, + tracing::warn, +}; + +/// Format buffered status log entries into a Telegram expandable blockquote HTML. +/// Returns an empty string if there are no entries after filtering. +pub(crate) fn format_logbook_html_for_mode( + entries: &[ChannelStatusLogEntry], + mode: ActivityLogMode, +) -> String { + let mut html = String::new(); + for entry in entries.iter().filter(|entry| mode.includes(entry.kind)) { + if html.is_empty() { + html.push_str("
\n\u{1f4cb} Activity log\n"); + } + let escaped = entry + .message + .replace('&', "&") + .replace('<', "<") + .replace('>', ">"); + html.push_str(&format!("\u{2022} {escaped}\n")); + } + if !html.is_empty() { + html.push_str("
"); + } + html +} + +fn format_logbook_plain_text_for_mode( + entries: &[ChannelStatusLogEntry], + mode: ActivityLogMode, +) -> String { + let mut text = String::new(); + for entry in entries.iter().filter(|entry| mode.includes(entry.kind)) { + if text.is_empty() { + text.push_str("Activity log\n"); + } + text.push_str("- "); + text.push_str(&entry.message); + text.push('\n'); + } + let _ = text.pop(); + text +} + +fn supports_html_logbook_follow_up(channel_type: ChannelType) -> bool { + matches!( + channel_type, + ChannelType::Telegram | ChannelType::Discord | ChannelType::Matrix + ) +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum ChannelLogbookFollowUp { + Html(String), + Text(String), +} + +pub(crate) fn format_channel_logbook_follow_up( + entries: &[ChannelStatusLogEntry], + target: &ChannelReplyTarget, +) -> Option { + if supports_html_logbook_follow_up(target.channel_type) { + let html = format_logbook_html_for_mode(entries, target.activity_log); + if html.is_empty() { + None + } else { + Some(ChannelLogbookFollowUp::Html(html)) + } + } else { + let text = format_logbook_plain_text_for_mode(entries, target.activity_log); + if text.is_empty() { + None + } else { + Some(ChannelLogbookFollowUp::Text(text)) + } + } +} + +pub(crate) async fn send_channel_logbook_follow_up( + outbound: &dyn ChannelOutbound, + target: &ChannelReplyTarget, + to: &str, + follow_up: ChannelLogbookFollowUp, +) -> moltis_channels::Result<()> { + match follow_up { + ChannelLogbookFollowUp::Html(html) => { + outbound + .send_html(&target.account_id, to, &html, None) + .await + }, + ChannelLogbookFollowUp::Text(text) => { + outbound + .send_text(&target.account_id, to, &text, None) + .await + }, + } +} + +pub(crate) async fn send_channel_logbook_follow_up_to_targets( + outbound: Arc, + targets: Vec, + status_log: &[ChannelStatusLogEntry], +) { + if targets.is_empty() || status_log.is_empty() { + return; + } + + let mut tasks = Vec::with_capacity(targets.len()); + for target in targets { + let outbound = Arc::clone(&outbound); + let Some(follow_up) = format_channel_logbook_follow_up(status_log, &target) else { + continue; + }; + let to = target.outbound_to().into_owned(); + tasks.push(tokio::spawn(async move { + if let Err(e) = + send_channel_logbook_follow_up(outbound.as_ref(), &target, &to, follow_up).await + { + warn!( + account_id = target.account_id, + chat_id = target.chat_id, + thread_id = target.thread_id.as_deref().unwrap_or("-"), + "failed to send logbook follow-up: {e}" + ); + } + })); + } + + for task in tasks { + if let Err(e) = task.await { + warn!(error = %e, "channel logbook follow-up task join failed"); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + use { + async_trait::async_trait, + moltis_channels::{ChannelStatusLogKind, ChannelType}, + moltis_common::types::ReplyPayload, + tokio::sync::Mutex, + }; + + #[derive(Default)] + struct RecordingOutbound { + sent_text: Mutex>, + } + + #[async_trait] + impl ChannelOutbound for RecordingOutbound { + async fn send_text( + &self, + _account_id: &str, + _to: &str, + text: &str, + _reply_to: Option<&str>, + ) -> moltis_channels::Result<()> { + self.sent_text.lock().await.push(text.to_string()); + Ok(()) + } + + async fn send_media( + &self, + _account_id: &str, + _to: &str, + _payload: &ReplyPayload, + _reply_to: Option<&str>, + ) -> moltis_channels::Result<()> { + Ok(()) + } + } + + #[tokio::test] + async fn non_html_channel_logbook_follow_up_uses_plain_text() { + let outbound = Arc::new(RecordingOutbound::default()); + let targets = vec![ChannelReplyTarget { + channel_type: ChannelType::Slack, + account_id: "slack1".into(), + chat_id: "C123".into(), + message_id: None, + thread_id: None, + sender_id: None, + activity_log: ActivityLogMode::All, + }]; + let entries = [ChannelStatusLogEntry { + kind: ChannelStatusLogKind::Info, + message: "Running: `date`".into(), + }]; + + send_channel_logbook_follow_up_to_targets(outbound.clone(), targets, &entries).await; + + let sent = outbound.sent_text.lock().await; + assert_eq!(sent.len(), 1); + assert!(sent[0].contains("Activity log")); + assert!(sent[0].contains("Running: `date`")); + assert!(!sent[0].contains("")); + } +} diff --git a/crates/chat/src/channels.rs b/crates/chat/src/channels.rs index de566b33d2..964cac3b7b 100644 --- a/crates/chat/src/channels.rs +++ b/crates/chat/src/channels.rs @@ -8,10 +8,20 @@ use { tracing::{debug, info, warn}, }; -use moltis_sessions::store::SessionStore; +use { + moltis_channels::{ChannelReplyTarget, ChannelStatusLogEntry, plugin::ChannelOutbound}, + moltis_sessions::store::SessionStore, +}; use crate::{ - agent_loop::ChannelReplyTargetKey, compaction_run, error, runtime::ChatRuntime, types::*, + agent_loop::ChannelReplyTargetKey, + channel_logbook::{ + ChannelLogbookFollowUp, format_channel_logbook_follow_up, format_logbook_html_for_mode, + send_channel_logbook_follow_up, send_channel_logbook_follow_up_to_targets, + }, + compaction_run, error, + runtime::ChatRuntime, + types::*, }; /// Build the SPA URL for a push notification click-through. @@ -130,12 +140,11 @@ pub(crate) async fn deliver_channel_replies( }; // Drain buffered status log entries to build a logbook suffix. let status_log = state.drain_channel_status_log(session_key).await; - let logbook_html = format_logbook_html(&status_log); - if !streamed_targets.is_empty() && !logbook_html.is_empty() { + if !streamed_targets.is_empty() { send_channel_logbook_follow_up_to_targets( Arc::clone(&outbound), streamed_targets, - &logbook_html, + &status_log, ) .await; } @@ -163,62 +172,6 @@ pub(crate) async fn deliver_channel_replies( .await; } -/// Format buffered status log entries into a Telegram expandable blockquote HTML. -/// Returns an empty string if there are no entries. -fn format_logbook_html(entries: &[String]) -> String { - if entries.is_empty() { - return String::new(); - } - let mut html = String::from("
\n\u{1f4cb} Activity log\n"); - for entry in entries { - // Escape HTML entities in the entry text. - let escaped = entry - .replace('&', "&") - .replace('<', "<") - .replace('>', ">"); - html.push_str(&format!("\u{2022} {escaped}\n")); - } - html.push_str("
"); - html -} - -async fn send_channel_logbook_follow_up_to_targets( - outbound: Arc, - targets: Vec, - logbook_html: &str, -) { - if targets.is_empty() || logbook_html.is_empty() { - return; - } - - let html = logbook_html.to_string(); - let mut tasks = Vec::with_capacity(targets.len()); - for target in targets { - let outbound = Arc::clone(&outbound); - let html = html.clone(); - let to = target.outbound_to().into_owned(); - tasks.push(tokio::spawn(async move { - if let Err(e) = outbound - .send_html(&target.account_id, &to, &html, None) - .await - { - warn!( - account_id = target.account_id, - chat_id = target.chat_id, - thread_id = target.thread_id.as_deref().unwrap_or("-"), - "failed to send logbook follow-up: {e}" - ); - } - })); - } - - for task in tasks { - if let Err(e) = task.await { - warn!(error = %e, "channel logbook follow-up task join failed"); - } - } -} - fn format_channel_retry_message(error_obj: &Value, retry_after: Duration) -> String { let retry_secs = ((retry_after.as_millis() as u64).saturating_add(999) / 1_000).max(1); if error_obj.get("type").and_then(|v| v.as_str()) == Some("rate_limit_exceeded") { @@ -413,12 +366,11 @@ pub(crate) async fn deliver_channel_error( }; let error_text = format_channel_error_message(error_obj); - let logbook_html = format_logbook_html(&status_log); let mut tasks = Vec::with_capacity(targets.len()); for target in targets { let outbound = Arc::clone(&outbound); let error_text = error_text.clone(); - let logbook_html = logbook_html.clone(); + let logbook_html = format_logbook_html_for_mode(&status_log, target.activity_log); let to = target.outbound_to().into_owned(); tasks.push(tokio::spawn(async move { let reply_to = target.message_id.as_deref(); @@ -456,25 +408,25 @@ pub(crate) async fn deliver_channel_error( } async fn deliver_channel_replies_to_targets( - outbound: Arc, - targets: Vec, + outbound: Arc, + targets: Vec, session_key: &str, text: &str, state: Arc, desired_reply_medium: ReplyMedium, - status_log: Vec, + status_log: Vec, streamed_target_keys: &HashSet, ) { let session_key = session_key.to_string(); let text = text.to_string(); - let logbook_html = format_logbook_html(&status_log); let mut tasks = Vec::with_capacity(targets.len()); for target in targets { let outbound = Arc::clone(&outbound); let state = Arc::clone(&state); let session_key = session_key.clone(); let text = text.clone(); - let logbook_html = logbook_html.clone(); + let logbook_html = format_logbook_html_for_mode(&status_log, target.activity_log); + let logbook_follow_up = format_channel_logbook_follow_up(&status_log, &target); // Text was already delivered via edit-in-place streaming — skip text // caption/follow-up and only send the TTS voice audio. let text_already_streamed = @@ -504,19 +456,13 @@ async fn deliver_channel_replies_to_targets( "failed to send channel voice reply: {e}" ); } - // Send logbook as a follow-up if present. - if !logbook_html.is_empty() - && let Err(e) = outbound - .send_html(&target.account_id, &to, &logbook_html, None) - .await - { - warn!( - account_id = target.account_id, - chat_id = target.chat_id, - thread_id = target.thread_id.as_deref().unwrap_or("-"), - "failed to send logbook follow-up: {e}" - ); - } + send_logbook_follow_up_if_present( + outbound.as_ref(), + &target, + &to, + logbook_follow_up.clone(), + ) + .await; } else { // Check if transcript fits as Telegram caption (when feature enabled). // When telegram feature is disabled, this evaluates to false and we @@ -541,19 +487,13 @@ async fn deliver_channel_replies_to_targets( "failed to send channel voice reply: {e}" ); } - // Send logbook as a follow-up if present. - if !logbook_html.is_empty() - && let Err(e) = outbound - .send_html(&target.account_id, &to, &logbook_html, None) - .await - { - warn!( - account_id = target.account_id, - chat_id = target.chat_id, - thread_id = target.thread_id.as_deref().unwrap_or("-"), - "failed to send logbook follow-up: {e}" - ); - } + send_logbook_follow_up_if_present( + outbound.as_ref(), + &target, + &to, + logbook_follow_up.clone(), + ) + .await; } else { // Transcript too long for a caption — send voice // without caption, then the full text as a follow-up. @@ -597,18 +537,13 @@ async fn deliver_channel_replies_to_targets( None if text_already_streamed => { // TTS disabled/failed but text was already streamed — // only send logbook follow-up if present. - if !logbook_html.is_empty() - && let Err(e) = outbound - .send_html(&target.account_id, &to, &logbook_html, None) - .await - { - warn!( - account_id = target.account_id, - chat_id = target.chat_id, - thread_id = target.thread_id.as_deref().unwrap_or("-"), - "failed to send logbook follow-up: {e}" - ); - } + send_logbook_follow_up_if_present( + outbound.as_ref(), + &target, + &to, + logbook_follow_up.clone(), + ) + .await; }, None => { let result = if logbook_html.is_empty() { @@ -653,18 +588,13 @@ async fn deliver_channel_replies_to_targets( None if text_already_streamed => { // TTS disabled/failed but text was already streamed — // only send logbook follow-up if present. - if !logbook_html.is_empty() - && let Err(e) = outbound - .send_html(&target.account_id, &to, &logbook_html, None) - .await - { - warn!( - account_id = target.account_id, - chat_id = target.chat_id, - thread_id = target.thread_id.as_deref().unwrap_or("-"), - "failed to send logbook follow-up: {e}" - ); - } + send_logbook_follow_up_if_present( + outbound.as_ref(), + &target, + &to, + logbook_follow_up.clone(), + ) + .await; }, None => { let result = if logbook_html.is_empty() { @@ -703,6 +633,25 @@ async fn deliver_channel_replies_to_targets( } } +async fn send_logbook_follow_up_if_present( + outbound: &dyn ChannelOutbound, + target: &ChannelReplyTarget, + to: &str, + follow_up: Option, +) { + let Some(follow_up) = follow_up else { + return; + }; + if let Err(e) = send_channel_logbook_follow_up(outbound, target, to, follow_up).await { + warn!( + account_id = target.account_id, + chat_id = target.chat_id, + thread_id = target.thread_id.as_deref().unwrap_or("-"), + "failed to send logbook follow-up: {e}" + ); + } +} + #[derive(Debug, Deserialize)] struct TtsStatusResponse { enabled: bool, @@ -786,7 +735,7 @@ pub(crate) async fn generate_tts_audio( async fn build_tts_payload( state: &Arc, session_key: &str, - target: &moltis_channels::ChannelReplyTarget, + target: &ChannelReplyTarget, text: &str, ) -> Option { use moltis_common::types::{MediaAttachment, ReplyPayload}; @@ -853,7 +802,9 @@ pub(crate) async fn send_tool_status_to_channels( // Buffer the status message for the logbook let message = format_tool_status_message(tool_name, arguments); - state.push_channel_status_log(session_key, message).await; + state + .push_channel_status_log(session_key, ChannelStatusLogEntry::info(message)) + .await; } /// Buffer a tool error result into the channel status log for a session. @@ -876,7 +827,9 @@ pub(crate) async fn send_tool_result_to_channels( } let message = format_tool_result_message(tool_name, error, result); - state.push_channel_status_log(session_key, message).await; + state + .push_channel_status_log(session_key, ChannelStatusLogEntry::error(message)) + .await; } /// Format a human-readable error summary for a failed tool call. @@ -1091,8 +1044,8 @@ fn build_screenshot_reply_payload( } async fn dispatch_screenshot_to_targets( - outbound: Arc, - targets: Vec, + outbound: Arc, + targets: Vec, screenshot_data: &str, caption: Option<&str>, ) { @@ -1344,7 +1297,7 @@ mod tests { use { super::*, async_trait::async_trait, - moltis_channels::{ChannelReplyTarget, ChannelType}, + moltis_channels::{ActivityLogMode, ChannelReplyTarget, ChannelType}, moltis_common::types::ReplyPayload, std::sync::{Arc, Mutex}, }; @@ -1363,7 +1316,7 @@ mod tests { } #[async_trait] - impl moltis_channels::ChannelOutbound for RecordingOutbound { + impl ChannelOutbound for RecordingOutbound { async fn send_text( &self, _account_id: &str, @@ -1413,6 +1366,30 @@ mod tests { assert_eq!(push_notification_url("main"), "/chats/main"); } + #[test] + fn logbook_filters_entries_by_activity_log_mode() { + use moltis_channels::{ActivityLogMode, ChannelStatusLogEntry}; + + let entries = vec![ + ChannelStatusLogEntry::info("Running: `date`"), + ChannelStatusLogEntry::error("exit 1 -- failed"), + ]; + + let all = format_logbook_html_for_mode(&entries, ActivityLogMode::All); + assert!(all.contains("Running: `date`")); + assert!(all.contains("exit 1")); + + let errors_only = format_logbook_html_for_mode(&entries, ActivityLogMode::ErrorsOnly); + assert!(!errors_only.contains("Running: `date`")); + assert!(errors_only.contains("exit 1")); + + let off = format_logbook_html_for_mode(&entries, ActivityLogMode::Off); + assert!(off.is_empty()); + + let no_matching = format_logbook_html_for_mode(&entries[..1], ActivityLogMode::ErrorsOnly); + assert!(no_matching.is_empty()); + } + #[tokio::test] async fn generated_image_payload_dispatches_to_telegram_as_media() { let outbound = Arc::new(RecordingOutbound::default()); @@ -1422,6 +1399,8 @@ mod tests { chat_id: "-100123".into(), message_id: Some("42".into()), thread_id: Some("7".into()), + sender_id: None, + activity_log: ActivityLogMode::All, }]; dispatch_screenshot_to_targets( @@ -1457,6 +1436,8 @@ mod tests { chat_id: "!room:example.org".into(), message_id: Some("$event".into()), thread_id: None, + sender_id: None, + activity_log: ActivityLogMode::All, }]; dispatch_screenshot_to_targets( diff --git a/crates/chat/src/lib.rs b/crates/chat/src/lib.rs index 63bcd570bd..a96e4423a0 100644 --- a/crates/chat/src/lib.rs +++ b/crates/chat/src/lib.rs @@ -1,4 +1,5 @@ mod agent_loop; +mod channel_logbook; mod channels; mod compaction; mod compaction_run; diff --git a/crates/chat/src/runtime.rs b/crates/chat/src/runtime.rs index cf430dceb1..1760870906 100644 --- a/crates/chat/src/runtime.rs +++ b/crates/chat/src/runtime.rs @@ -7,7 +7,10 @@ use std::sync::Arc; use serde_json::Value; -use {moltis_channels::ChannelReplyTarget, moltis_tools::sandbox::SandboxRouter}; +use { + moltis_channels::{ChannelReplyTarget, ChannelStatusLogEntry}, + moltis_tools::sandbox::SandboxRouter, +}; /// TTS runtime override configuration (provider/voice/model). /// @@ -62,11 +65,11 @@ pub trait ChatRuntime: Send + Sync { // ── Channel status log ─────────────────────────────────────────────── - /// Append a status line (tool use, model selection) to a session's log. - async fn push_channel_status_log(&self, session_key: &str, message: String); + /// Append a status entry (tool use, model selection, or error) to a session's log. + async fn push_channel_status_log(&self, session_key: &str, entry: ChannelStatusLogEntry); /// Drain all buffered status log entries for a session. - async fn drain_channel_status_log(&self, session_key: &str) -> Vec; + async fn drain_channel_status_log(&self, session_key: &str) -> Vec; // ── Run error tracking ─────────────────────────────────────────────── diff --git a/crates/config/src/template.rs b/crates/config/src/template.rs index 8320fd737f..14edbfbb2e 100644 --- a/crates/config/src/template.rs +++ b/crates/config/src/template.rs @@ -723,6 +723,17 @@ port = {port} # Port number (auto-generated for this i # See docs or defaults.toml for full channel configuration examples # (WhatsApp, Telegram, Teams, Discord, Slack, Matrix, Nostr, Signal). +# +# Example Telegram Activity log controls: +# [channels.telegram.my-bot] +# token = "..." +# activity_log = "all" # "all", "errors_only", or "off" +# +# [channels.telegram.my-bot.channel_overrides."-1001234567890"] +# activity_log = "off" +# +# [channels.telegram.my-bot.user_overrides."123456789"] +# activity_log = "errors_only" # ══════════════════════════════════════════════════════════════════════════════ # HOOKS diff --git a/crates/discord/src/commands.rs b/crates/discord/src/commands.rs index 92da9774a7..eeafc9fa5c 100644 --- a/crates/discord/src/commands.rs +++ b/crates/discord/src/commands.rs @@ -112,6 +112,8 @@ async fn handle_slash_command( chat_id: command.channel_id.to_string(), message_id: None, thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; let sender_id = command.user.id.to_string(); @@ -171,6 +173,8 @@ async fn handle_component_interaction( chat_id: component.channel_id.to_string(), message_id: None, thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; match sink.dispatch_interaction(callback_data, reply_to).await { diff --git a/crates/discord/src/handler/implementation.rs b/crates/discord/src/handler/implementation.rs index 7b2a07bbb9..4220c8cf0e 100644 --- a/crates/discord/src/handler/implementation.rs +++ b/crates/discord/src/handler/implementation.rs @@ -905,6 +905,8 @@ impl EventHandler for Handler { chat_id: chat_id.clone(), message_id: Some(msg.id.to_string()), thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; let Some(sink) = event_sink else { diff --git a/crates/gateway/src/channel_agent_tools.rs b/crates/gateway/src/channel_agent_tools.rs index 78d0478a5c..6ed367ba98 100644 --- a/crates/gateway/src/channel_agent_tools.rs +++ b/crates/gateway/src/channel_agent_tools.rs @@ -3,6 +3,7 @@ use { async_trait::async_trait, moltis_agents::tool_registry::AgentTool, moltis_channels::{ + ActivityLogMode, gating::{DmPolicy, GroupPolicy, MentionMode}, plugin::ChannelType, store::{ChannelStore, StoredChannel}, @@ -101,6 +102,7 @@ struct ChannelSettingsPatch { reply_to_message: Option, thread_replies: Option, stream_mode: Option, + activity_log: Option, allowlist_add: Vec, allowlist_remove: Vec, group_allowlist_add: Vec, @@ -119,6 +121,8 @@ struct ModelOverridePatch { #[serde(default)] agent_id: Option>, #[serde(default)] + activity_log: Option>, + #[serde(default)] clear: bool, } @@ -140,6 +144,37 @@ impl ChannelSettingsStreamMode { } } +fn validate_activity_log_patch_values(settings: &Value) -> Result<()> { + if let Some(value) = settings.get("activity_log") { + validate_activity_log_value("settings.activity_log", value, false)?; + } + for override_key in ["channel_override", "user_override"] { + let Some(override_value) = settings.get(override_key) else { + continue; + }; + if let Some(value) = override_value.get("activity_log") { + validate_activity_log_value( + &format!("settings.{override_key}.activity_log"), + value, + true, + )?; + } + } + Ok(()) +} + +fn validate_activity_log_value(path: &str, value: &Value, allow_null: bool) -> Result<()> { + if allow_null && value.is_null() { + return Ok(()); + } + match value.as_str() { + Some("all" | "errors_only" | "off") => Ok(()), + _ => Err(anyhow!( + "invalid '{path}': activity_log must be one of 'all', 'errors_only', or 'off'" + )), + } +} + /// Agent tool that safely updates persisted channel settings. /// /// This deliberately exposes a narrow patch surface instead of arbitrary config @@ -172,6 +207,37 @@ impl AgentTool for UpdateChannelSettingsTool { } fn parameters_schema(&self) -> Value { + let override_activity_log_schema = json!({ + "type": ["string", "null"], + "enum": ["all", "errors_only", "off", null], + }); + let channel_override_schema = json!({ + "type": "object", + "description": "Set or clear model, provider, agent, or Activity log overrides for a specific Telegram/Discord/Slack/WhatsApp channel or chat id. Activity log overrides are currently supported by Telegram only.", + "required": ["target_id"], + "properties": { + "target_id": { "type": "string" }, + "model": { "type": ["string", "null"] }, + "model_provider": { "type": ["string", "null"] }, + "agent_id": { "type": ["string", "null"] }, + "activity_log": override_activity_log_schema.clone(), + "clear": { "type": "boolean", "default": false } + } + }); + let user_override_schema = json!({ + "type": "object", + "description": "Set or clear model, provider, agent, or Activity log overrides for a specific Telegram/Discord/Slack/WhatsApp user id. Activity log overrides are currently supported by Telegram only.", + "required": ["target_id"], + "properties": { + "target_id": { "type": "string" }, + "model": { "type": ["string", "null"] }, + "model_provider": { "type": ["string", "null"] }, + "agent_id": { "type": ["string", "null"] }, + "activity_log": override_activity_log_schema, + "clear": { "type": "boolean", "default": false } + } + }); + json!({ "type": "object", "required": ["account_id", "settings"], @@ -256,30 +322,13 @@ impl AgentTool for UpdateChannelSettingsTool { "enum": ["edit_in_place", "native", "off"], "description": "Supported by Telegram (`edit_in_place`, `off`) and Slack (`edit_in_place`, `native`, `off`)." }, - "channel_override": { - "type": "object", - "description": "Set or clear model, provider, or agent overrides for a specific Telegram/Discord/Slack/WhatsApp channel or chat id.", - "required": ["target_id"], - "properties": { - "target_id": { "type": "string" }, - "model": { "type": ["string", "null"] }, - "model_provider": { "type": ["string", "null"] }, - "agent_id": { "type": ["string", "null"] }, - "clear": { "type": "boolean", "default": false } - } + "activity_log": { + "type": "string", + "enum": ["all", "errors_only", "off"], + "description": "Controls user-facing Activity log delivery for this channel account. Currently supported by Telegram only. Defaults to all." }, - "user_override": { - "type": "object", - "description": "Set or clear model, provider, or agent overrides for a specific Telegram/Discord/Slack/WhatsApp user id.", - "required": ["target_id"], - "properties": { - "target_id": { "type": "string" }, - "model": { "type": ["string", "null"] }, - "model_provider": { "type": ["string", "null"] }, - "agent_id": { "type": ["string", "null"] }, - "clear": { "type": "boolean", "default": false } - } - } + "channel_override": channel_override_schema, + "user_override": user_override_schema } } } @@ -306,6 +355,7 @@ impl AgentTool for UpdateChannelSettingsTool { .get("settings") .cloned() .ok_or_else(|| anyhow!("missing 'settings'"))?; + validate_activity_log_patch_values(&settings)?; let patch: ChannelSettingsPatch = serde_json::from_value(settings).map_err(|e| anyhow!("invalid 'settings': {e}"))?; @@ -459,6 +509,15 @@ fn apply_channel_settings_patch( config.insert("stream_mode".into(), json!(stream_mode)); changes.push("stream_mode".to_string()); } + if let Some(activity_log) = patch.activity_log { + ensure_supported( + channel_type, + "activity_log", + supports_activity_log(channel_type), + )?; + config.insert("activity_log".into(), json!(activity_log)); + changes.push("activity_log".to_string()); + } if update_string_array( config, "allowlist", @@ -481,6 +540,13 @@ fn apply_channel_settings_patch( "channel_override", supports_model_overrides(channel_type), )?; + if override_patch.activity_log.is_some() { + ensure_supported( + channel_type, + "activity_log", + supports_activity_log(channel_type), + )?; + } apply_model_override_patch(config, "channel_overrides", override_patch)?; changes.push(format!("channel_override:{}", override_patch.target_id)); } @@ -490,6 +556,13 @@ fn apply_channel_settings_patch( "user_override", supports_model_overrides(channel_type), )?; + if override_patch.activity_log.is_some() { + ensure_supported( + channel_type, + "activity_log", + supports_activity_log(channel_type), + )?; + } apply_model_override_patch(config, "user_overrides", override_patch)?; changes.push(format!("user_override:{}", override_patch.target_id)); } @@ -545,6 +618,10 @@ fn supports_model_overrides(channel_type: ChannelType) -> bool { ) } +fn supports_activity_log(channel_type: ChannelType) -> bool { + matches!(channel_type, ChannelType::Telegram) +} + fn validate_stream_mode( channel_type: ChannelType, stream_mode: &ChannelSettingsStreamMode, @@ -667,9 +744,13 @@ fn apply_model_override_patch( return Ok(()); } - if patch.model.is_none() && patch.model_provider.is_none() && patch.agent_id.is_none() { + if patch.model.is_none() + && patch.model_provider.is_none() + && patch.agent_id.is_none() + && patch.activity_log.is_none() + { return Err(anyhow!( - "'{key}' for '{}' must set 'model', 'model_provider', or 'agent_id', or use clear=true", + "'{key}' for '{}' must set 'model', 'model_provider', 'agent_id', or 'activity_log', or use clear=true", patch.target_id )); } @@ -691,6 +772,16 @@ fn apply_model_override_patch( if let Some(agent_id) = &patch.agent_id { set_optional_string(&mut override_value, "agent_id", agent_id); } + if let Some(activity_log) = patch.activity_log { + match activity_log { + Some(activity_log) => { + override_value.insert("activity_log".to_string(), json!(activity_log)); + }, + None => { + override_value.remove("activity_log"); + }, + } + } if override_value.is_empty() { overrides.remove(&patch.target_id); @@ -1114,6 +1205,119 @@ mod tests { ); } + #[tokio::test] + async fn update_channel_settings_merges_activity_log_overrides() { + let service = Arc::new(RecordingChannelService::new()); + let store = Arc::new(MemoryChannelStore::new(vec![stored_channel( + "telegram-main", + "telegram", + json!({ + "token": "telegram-secret", + "allowlist": [], + "group_allowlist": [], + "channel_overrides": { + "chat-1": { "model": "old-model" } + } + }), + )])); + let tool = UpdateChannelSettingsTool::new( + service.clone() as Arc, + Some(store as Arc), + ); + + tool.execute(json!({ + "account_id": "telegram-main", + "settings": { + "activity_log": "errors_only", + "channel_override": { + "target_id": "chat-1", + "activity_log": "off" + }, + "user_override": { + "target_id": "user-1", + "activity_log": "all" + } + } + })) + .await + .expect("activity log settings update"); + + let updated = service + .updated + .lock() + .await + .clone() + .expect("captured update payload"); + assert_eq!(updated["config"]["activity_log"], "errors_only"); + assert_eq!( + updated["config"]["channel_overrides"]["chat-1"]["model"], + "old-model" + ); + assert_eq!( + updated["config"]["channel_overrides"]["chat-1"]["activity_log"], + "off" + ); + assert_eq!( + updated["config"]["user_overrides"]["user-1"]["activity_log"], + "all" + ); + } + + #[tokio::test] + async fn update_channel_settings_rejects_invalid_activity_log_mode() { + let service = Arc::new(RecordingChannelService::new()); + let store = Arc::new(MemoryChannelStore::new(vec![stored_channel( + "telegram-main", + "telegram", + json!({ "token": "telegram-secret", "allowlist": [], "group_allowlist": [] }), + )])); + let tool = UpdateChannelSettingsTool::new( + service as Arc, + Some(store as Arc), + ); + + let err = tool + .execute(json!({ + "account_id": "telegram-main", + "settings": { "activity_log": "verbose" } + })) + .await + .expect_err("invalid activity_log should fail"); + + assert!(err.to_string().contains("activity_log")); + } + + #[tokio::test] + async fn update_channel_settings_rejects_activity_log_for_unsupported_channel() { + let service = Arc::new(RecordingChannelService::new()); + let store = Arc::new(MemoryChannelStore::new(vec![stored_channel( + "slack-main", + "slack", + json!({ "bot_token": "secret", "app_token": "secret" }), + )])); + let tool = UpdateChannelSettingsTool::new( + service as Arc, + Some(store as Arc), + ); + + let err = tool + .execute(json!({ + "account_id": "slack-main", + "type": "slack", + "settings": { + "channel_override": { + "target_id": "C123", + "activity_log": "off" + } + } + })) + .await + .expect_err("unsupported activity_log should fail"); + + assert!(err.to_string().contains("activity_log")); + assert!(err.to_string().contains("slack")); + } + #[test] fn update_string_array_ignores_noop_changes() { let mut config = Map::from_iter([("allowlist".to_string(), json!(["alice", "bob"]))]); diff --git a/crates/gateway/src/channel_events.rs b/crates/gateway/src/channel_events.rs index 30bca0da65..8f7f488d55 100644 --- a/crates/gateway/src/channel_events.rs +++ b/crates/gateway/src/channel_events.rs @@ -8,8 +8,8 @@ use { use { moltis_channels::{ - ChannelAttachment, ChannelEvent, ChannelEventSink, ChannelMessageMeta, ChannelReplyTarget, - Error as ChannelError, Result as ChannelResult, SavedChannelFile, + ActivityLogMode, ChannelAttachment, ChannelEvent, ChannelEventSink, ChannelMessageMeta, + ChannelReplyTarget, Error as ChannelError, Result as ChannelResult, SavedChannelFile, }, moltis_sessions::metadata::{SessionEntry, SqliteSessionMetadata}, moltis_tools::approval::PendingApprovalView, @@ -234,6 +234,22 @@ fn config_string(value: Option<&serde_json::Value>) -> Option { .map(str::to_string) } +fn config_activity_log(path: &str, value: Option<&serde_json::Value>) -> Option { + let value = value?; + match serde_json::from_value(value.clone()) { + Ok(mode) => Some(mode), + Err(error) if !value.is_null() => { + warn!( + path, + error = %error, + "invalid channel activity_log config value; falling back" + ); + None + }, + Err(_) => None, + } +} + fn override_map<'a>( config: &'a serde_json::Value, key: &str, @@ -246,35 +262,50 @@ fn override_map<'a>( .and_then(serde_json::Value::as_object) } +async fn channel_account_config( + state: &GatewayState, + reply_to: &ChannelReplyTarget, +) -> Option { + if let Some(registry) = state.services.channel_registry.as_ref() + && registry + .resolve_channel_type(&reply_to.account_id) + .as_deref() + == Some(reply_to.channel_type.as_str()) + && let Some(config) = registry.account_config_json(&reply_to.account_id).await + { + return Some(config); + } + + if let Some(store) = state.services.channel_store.as_ref() { + match store + .get(reply_to.channel_type.as_str(), &reply_to.account_id) + .await + { + Ok(Some(channel)) => return Some(channel.config), + Ok(None) => {}, + Err(error) => { + warn!( + error = %error, + channel_type = reply_to.channel_type.as_str(), + account_id = %reply_to.account_id, + "failed to read stored channel config; falling back to defaults" + ); + }, + } + } + None +} + async fn resolve_channel_session_defaults( state: &Arc, reply_to: &ChannelReplyTarget, sender_id: Option<&str>, ) -> ChannelSessionDefaults { - let Ok(status) = state.services.channel.status().await else { - return ChannelSessionDefaults::default(); - }; - let Some(channel) = status - .get("channels") - .and_then(serde_json::Value::as_array) - .and_then(|channels| { - channels.iter().find(|channel| { - channel - .get("account_id") - .and_then(serde_json::Value::as_str) - == Some(reply_to.account_id.as_str()) - && channel.get("type").and_then(serde_json::Value::as_str) - == Some(reply_to.channel_type.as_str()) - }) - }) - else { - return ChannelSessionDefaults::default(); - }; - let Some(config) = channel.get("config") else { + let Some(config) = channel_account_config(state, reply_to).await else { return ChannelSessionDefaults::default(); }; - resolve_channel_session_defaults_from_config(config, &reply_to.chat_id, sender_id) + resolve_channel_session_defaults_from_config(&config, &reply_to.chat_id, sender_id) } fn resolve_channel_session_defaults_from_config( @@ -309,6 +340,51 @@ fn resolve_channel_session_defaults_from_config( } } +pub(crate) async fn resolve_channel_activity_log( + state: &GatewayState, + reply_to: &ChannelReplyTarget, + sender_id: Option<&str>, +) -> ActivityLogMode { + let Some(config) = channel_account_config(state, reply_to).await else { + return ActivityLogMode::All; + }; + + resolve_channel_activity_log_from_config(&config, &reply_to.chat_id, sender_id) +} + +fn resolve_channel_activity_log_from_config( + config: &serde_json::Value, + chat_id: &str, + sender_id: Option<&str>, +) -> ActivityLogMode { + let user_override = override_map( + config, + "user_overrides", + sender_id + .filter(|sender_id| *sender_id != chat_id) + .unwrap_or(chat_id), + ); + let channel_override = override_map(config, "channel_overrides", chat_id); + + user_override + .and_then(|override_value| { + config_activity_log( + "user_overrides..activity_log", + override_value.get("activity_log"), + ) + }) + .or_else(|| { + channel_override.and_then(|override_value| { + config_activity_log( + "channel_overrides..activity_log", + override_value.get("activity_log"), + ) + }) + }) + .or_else(|| config_activity_log("activity_log", config.get("activity_log"))) + .unwrap_or(ActivityLogMode::All) +} + fn start_channel_typing_loop( state: &Arc, reply_to: &ChannelReplyTarget, diff --git a/crates/gateway/src/channel_events/commands/attachments.rs b/crates/gateway/src/channel_events/commands/attachments.rs index 3a67b511dc..4295548db2 100644 --- a/crates/gateway/src/channel_events/commands/attachments.rs +++ b/crates/gateway/src/channel_events/commands/attachments.rs @@ -18,7 +18,7 @@ pub(in crate::channel_events) async fn dispatch_to_chat_with_attachments( state: &Arc>>, text: &str, attachments: Vec, - reply_to: ChannelReplyTarget, + mut reply_to: ChannelReplyTarget, meta: ChannelMessageMeta, ) { if attachments.is_empty() { @@ -31,6 +31,7 @@ pub(in crate::channel_events) async fn dispatch_to_chat_with_attachments( warn!("channel dispatch_to_chat_with_attachments: gateway not ready"); return; }; + reply_to.sender_id = meta.sender_id.clone(); // Start typing immediately so image preprocessing/session setup doesn't // delay channel feedback. @@ -90,7 +91,7 @@ pub(in crate::channel_events) async fn dispatch_to_chat_with_attachments( // Persist channel binding (ensure session row exists first -- // set_channel_binding is an UPDATE so the row must already be present). - if let Ok(binding_json) = serde_json::to_string(&reply_to) + if let Ok(binding_json) = serde_json::to_string(&reply_to.for_persistence()) && let Some(ref session_meta) = state.services.session_metadata { let entry = session_meta.get(&session_key).await; @@ -177,7 +178,12 @@ pub(in crate::channel_events) async fn dispatch_to_chat_with_attachments( model.clone() }; let msg = format!("Using {display}. Use /model to change."); - state.push_channel_status_log(&session_key, msg).await; + state + .push_channel_status_log( + &session_key, + moltis_channels::ChannelStatusLogEntry::info(msg), + ) + .await; } } else { let session_has_model = if let Some(ref sm) = state.services.session_metadata { @@ -206,7 +212,12 @@ pub(in crate::channel_events) async fn dispatch_to_chat_with_attachments( .and_then(|v| v.as_str()) .unwrap_or(id); let msg = format!("Using {display}. Use /model to change."); - state.push_channel_status_log(&session_key, msg).await; + state + .push_channel_status_log( + &session_key, + moltis_channels::ChannelStatusLogEntry::info(msg), + ) + .await; } } diff --git a/crates/gateway/src/channel_events/commands/dispatch.rs b/crates/gateway/src/channel_events/commands/dispatch.rs index 3b0e4ce4ce..8223b9a5d6 100644 --- a/crates/gateway/src/channel_events/commands/dispatch.rs +++ b/crates/gateway/src/channel_events/commands/dispatch.rs @@ -37,9 +37,12 @@ pub(in crate::channel_events) async fn dispatch_interaction( pub(in crate::channel_events) async fn dispatch_command( state: &Arc>>, command: &str, - reply_to: ChannelReplyTarget, + mut reply_to: ChannelReplyTarget, sender_id: Option<&str>, ) -> ChannelResult { + if let Some(sender_id) = sender_id { + reply_to.sender_id = Some(sender_id.to_string()); + } let state = state .get() .ok_or_else(|| ChannelError::unavailable("gateway not ready"))?; diff --git a/crates/gateway/src/channel_events/commands/mod.rs b/crates/gateway/src/channel_events/commands/mod.rs index 6b6cbeac98..d13be072d5 100644 --- a/crates/gateway/src/channel_events/commands/mod.rs +++ b/crates/gateway/src/channel_events/commands/mod.rs @@ -5,7 +5,7 @@ pub(in crate::channel_events) mod formatting; mod location; mod media; mod quick_actions; -mod session_handlers; +pub(in crate::channel_events) mod session_handlers; // Re-export everything that `channel_events.rs` uses via `commands::*`. pub(super) use { diff --git a/crates/gateway/src/channel_events/commands/session_handlers.rs b/crates/gateway/src/channel_events/commands/session_handlers.rs index b31e542c9d..452c2dfbbe 100644 --- a/crates/gateway/src/channel_events/commands/session_handlers.rs +++ b/crates/gateway/src/channel_events/commands/session_handlers.rs @@ -1,6 +1,6 @@ use std::sync::Arc; -use tracing::info; +use tracing::{info, warn}; use { moltis_channels::{ChannelReplyTarget, Error as ChannelError, Result as ChannelResult}, @@ -28,7 +28,7 @@ pub(in crate::channel_events) async fn handle_new( ) -> ChannelResult { // Create a new session with a fresh UUID key. let new_key = format!("session:{}", uuid::Uuid::new_v4()); - let binding_json = serde_json::to_string(reply_to) + let binding_json = serde_json::to_string(&reply_to.for_persistence()) .map_err(|e| ChannelError::external("serialize channel binding", e))?; // Sequential label: count existing sessions for this chat. @@ -372,6 +372,29 @@ pub(in crate::channel_events) async fn handle_sessions( } let target_session = &sessions[n - 1]; + let mut refreshed_binding = target_session + .channel_binding + .as_deref() + .and_then(|binding_json| match serde_json::from_str::(binding_json) + { + Ok(binding) => Some(binding), + Err(error) => { + warn!( + error = %error, + session = %target_session.key, + "failed to parse channel binding while switching sessions; rebuilding from current target" + ); + None + }, + }) + .unwrap_or_else(|| reply_to.for_persistence()); + refreshed_binding.sender_id = reply_to.sender_id.clone(); + let binding_json = serde_json::to_string(&refreshed_binding.for_persistence()) + .map_err(|e| ChannelError::external("serialize channel binding", e))?; + session_metadata + .set_channel_binding(&target_session.key, Some(binding_json)) + .await; + // Update forward mapping. session_metadata .set_active_session( @@ -441,7 +464,7 @@ pub(in crate::channel_events) async fn handle_attach( } let target_session = &sessions[n - 1]; - let binding_json = serde_json::to_string(reply_to) + let binding_json = serde_json::to_string(&reply_to.for_persistence()) .map_err(|e| ChannelError::external("serialize channel binding", e))?; session_metadata diff --git a/crates/gateway/src/channel_events/dispatch.rs b/crates/gateway/src/channel_events/dispatch.rs index 2b06a03e0b..36eec6e4d3 100644 --- a/crates/gateway/src/channel_events/dispatch.rs +++ b/crates/gateway/src/channel_events/dispatch.rs @@ -3,10 +3,12 @@ use super::*; pub(in crate::channel_events) async fn dispatch_to_chat( state: &Arc>>, text: &str, - reply_to: ChannelReplyTarget, + mut reply_to: ChannelReplyTarget, meta: ChannelMessageMeta, ) { if let Some(state) = state.get() { + reply_to.sender_id = meta.sender_id.clone(); + // Start typing immediately so pre-run setup (session/model resolution) // does not delay channel feedback. let typing_done = start_channel_typing_loop(state, &reply_to); @@ -45,7 +47,7 @@ pub(in crate::channel_events) async fn dispatch_to_chat( // Persist channel binding so web UI messages on this session // can be echoed back to the channel. - if let Ok(binding_json) = serde_json::to_string(&reply_to) + if let Ok(binding_json) = serde_json::to_string(&reply_to.for_persistence()) && let Some(ref session_meta) = state.services.session_metadata { // Ensure the session row exists and label it on first use. @@ -184,7 +186,12 @@ pub(in crate::channel_events) async fn dispatch_to_chat( model.clone() }; let msg = format!("Using {display}. Use /model to change."); - state.push_channel_status_log(&session_key, msg).await; + state + .push_channel_status_log( + &session_key, + moltis_channels::ChannelStatusLogEntry::info(msg), + ) + .await; } } else { let session_has_model = if let Some(ref sm) = state.services.session_metadata { @@ -214,7 +221,12 @@ pub(in crate::channel_events) async fn dispatch_to_chat( .and_then(|v| v.as_str()) .unwrap_or(id); let msg = format!("Using {display}. Use /model to change."); - state.push_channel_status_log(&session_key, msg).await; + state + .push_channel_status_log( + &session_key, + moltis_channels::ChannelStatusLogEntry::info(msg), + ) + .await; } } diff --git a/crates/gateway/src/channel_events/tests.rs b/crates/gateway/src/channel_events/tests.rs index 2f8a0cb1d5..a640b27521 100644 --- a/crates/gateway/src/channel_events/tests.rs +++ b/crates/gateway/src/channel_events/tests.rs @@ -2,7 +2,284 @@ use {super::*, crate::channel_events::commands::formatting::unique_providers}; -use moltis_channels::ChannelType; +use { + crate::{ + auth::{AuthMode, ResolvedAuth}, + services::{ChannelService, GatewayServices, ServiceResult}, + }, + async_trait::async_trait, + moltis_channels::{ + ChannelOutbound, ChannelPlugin, ChannelRegistry, ChannelStatus, ChannelStreamOutbound, + ChannelType, + store::{ChannelStore, StoredChannel}, + }, + moltis_common::types::ReplyPayload, + serde_json::Value, + std::{collections::HashMap, sync::Arc}, + tokio::sync::RwLock, +}; + +struct MemoryChannelStore { + channels: HashMap<(String, String), StoredChannel>, +} + +#[async_trait] +impl ChannelStore for MemoryChannelStore { + async fn list(&self) -> moltis_channels::Result> { + Ok(self.channels.values().cloned().collect()) + } + + async fn get( + &self, + channel_type: &str, + account_id: &str, + ) -> moltis_channels::Result> { + Ok(self + .channels + .get(&(channel_type.to_string(), account_id.to_string())) + .cloned()) + } + + async fn upsert(&self, _channel: StoredChannel) -> moltis_channels::Result<()> { + Ok(()) + } + + async fn delete(&self, _channel_type: &str, _account_id: &str) -> moltis_channels::Result<()> { + Ok(()) + } +} + +struct StatusPanicsChannelService; + +#[async_trait] +impl ChannelService for StatusPanicsChannelService { + async fn status(&self) -> ServiceResult { + panic!("channel status should not be probed for stored config") + } + + async fn logout(&self, _params: Value) -> ServiceResult { + Ok(serde_json::json!({})) + } + + async fn send(&self, _params: Value) -> ServiceResult { + Ok(serde_json::json!({})) + } + + async fn add(&self, _params: Value) -> ServiceResult { + Ok(serde_json::json!({})) + } + + async fn remove(&self, _params: Value) -> ServiceResult { + Ok(serde_json::json!({})) + } + + async fn update(&self, _params: Value) -> ServiceResult { + Ok(serde_json::json!({})) + } + + async fn retry_ownership(&self, _params: Value) -> ServiceResult { + Ok(serde_json::json!({})) + } + + async fn senders_list(&self, _params: Value) -> ServiceResult { + Ok(serde_json::json!({})) + } + + async fn sender_approve(&self, _params: Value) -> ServiceResult { + Ok(serde_json::json!({})) + } + + async fn sender_deny(&self, _params: Value) -> ServiceResult { + Ok(serde_json::json!({})) + } +} + +struct ConfigPlugin { + id: String, + accounts: std::sync::Mutex>, +} + +impl ConfigPlugin { + fn new(id: &str) -> Self { + Self { + id: id.to_string(), + accounts: std::sync::Mutex::new(HashMap::new()), + } + } +} + +#[async_trait] +impl ChannelPlugin for ConfigPlugin { + fn id(&self) -> &str { + &self.id + } + + fn name(&self) -> &str { + &self.id + } + + async fn start_account( + &mut self, + account_id: &str, + config: Value, + ) -> moltis_channels::Result<()> { + self.accounts + .lock() + .unwrap() + .insert(account_id.to_string(), config); + Ok(()) + } + + async fn stop_account(&mut self, account_id: &str) -> moltis_channels::Result<()> { + self.accounts.lock().unwrap().remove(account_id); + Ok(()) + } + + fn outbound(&self) -> Option<&dyn ChannelOutbound> { + None + } + + fn status(&self) -> Option<&dyn ChannelStatus> { + None + } + + fn has_account(&self, account_id: &str) -> bool { + self.accounts.lock().unwrap().contains_key(account_id) + } + + fn account_ids(&self) -> Vec { + self.accounts.lock().unwrap().keys().cloned().collect() + } + + fn account_config( + &self, + _account_id: &str, + ) -> Option> { + None + } + + fn update_account_config( + &self, + account_id: &str, + config: Value, + ) -> moltis_channels::Result<()> { + if let Some(account_config) = self.accounts.lock().unwrap().get_mut(account_id) { + *account_config = config; + Ok(()) + } else { + Err(moltis_channels::Error::UnknownAccount { + account_id: account_id.to_string(), + }) + } + } + + fn shared_outbound(&self) -> Arc { + Arc::new(NullOutbound) + } + + fn shared_stream_outbound(&self) -> Arc { + Arc::new(NullStreamOutbound) + } + + fn account_config_json(&self, account_id: &str) -> Option { + self.accounts.lock().unwrap().get(account_id).cloned() + } +} + +struct NullOutbound; + +#[async_trait] +impl ChannelOutbound for NullOutbound { + async fn send_text( + &self, + _account_id: &str, + _to: &str, + _text: &str, + _reply_to: Option<&str>, + ) -> moltis_channels::Result<()> { + Ok(()) + } + + async fn send_media( + &self, + _account_id: &str, + _to: &str, + _payload: &ReplyPayload, + _reply_to: Option<&str>, + ) -> moltis_channels::Result<()> { + Ok(()) + } +} + +struct NullStreamOutbound; + +#[async_trait] +impl ChannelStreamOutbound for NullStreamOutbound { + async fn send_stream( + &self, + _account_id: &str, + _to: &str, + _reply_to: Option<&str>, + mut stream: moltis_channels::StreamReceiver, + ) -> moltis_channels::Result<()> { + while stream.recv().await.is_some() {} + Ok(()) + } +} + +fn test_state_with_channel_services( + store: Option>, + registry: Option>, +) -> Arc { + let mut services = GatewayServices::noop(); + if let Some(store) = store { + services = services.with_channel_store(store); + } + if let Some(registry) = registry { + services = services.with_channel_registry(registry); + } + services.channel = Arc::new(StatusPanicsChannelService); + + GatewayState::new( + ResolvedAuth { + mode: AuthMode::Token, + token: None, + password: None, + }, + services, + ) +} + +fn test_state_with_channel_store(store: Arc) -> Arc { + test_state_with_channel_services(Some(store), None) +} + +fn test_state_with_channel_store_and_registry( + store: Arc, + registry: Arc, +) -> Arc { + test_state_with_channel_services(Some(store), Some(registry)) +} + +async fn test_registry_with_account_config(config: Value) -> Arc { + let mut registry = ChannelRegistry::new(); + registry + .register(Arc::new(RwLock::new(ConfigPlugin::new("telegram")))) + .await; + let registry = Arc::new(registry); + registry + .start_account("telegram", "bot1", config) + .await + .unwrap(); + registry +} + +async fn sqlite_session_metadata() -> SqliteSessionMetadata { + let pool = sqlx::SqlitePool::connect("sqlite::memory:").await.unwrap(); + moltis_projects::run_migrations(&pool).await.unwrap(); + SqliteSessionMetadata::init(&pool).await.unwrap(); + SqliteSessionMetadata::new(pool) +} #[test] fn channel_event_serialization() { @@ -34,6 +311,8 @@ fn channel_session_key_format() { chat_id: "12345".into(), message_id: None, thread_id: None, + sender_id: None, + activity_log: ActivityLogMode::All, }; assert_eq!(default_channel_session_key(&target), "telegram:bot1:12345"); } @@ -46,6 +325,8 @@ fn channel_session_key_group() { chat_id: "-100999".into(), message_id: None, thread_id: None, + sender_id: None, + activity_log: ActivityLogMode::All, }; assert_eq!( default_channel_session_key(&target), @@ -61,6 +342,8 @@ fn channel_session_key_forum_topic() { chat_id: "-100999".into(), message_id: None, thread_id: Some("42".into()), + sender_id: None, + activity_log: ActivityLogMode::All, }; assert_eq!( default_channel_session_key(&target), @@ -68,6 +351,258 @@ fn channel_session_key_forum_topic() { ); } +#[test] +fn channel_activity_log_resolution_prefers_user_channel_account_and_fallback() { + let config = serde_json::json!({ + "activity_log": "errors_only", + "channel_overrides": { + "-100999": { "activity_log": "off" } + }, + "user_overrides": { + "123": { "activity_log": "all" }, + "456": { "activity_log": "verbose" } + } + }); + + assert_eq!( + resolve_channel_activity_log_from_config(&config, "-100999", Some("123")), + ActivityLogMode::All + ); + assert_eq!( + resolve_channel_activity_log_from_config(&config, "-100999", Some("789")), + ActivityLogMode::Off + ); + assert_eq!( + resolve_channel_activity_log_from_config(&config, "-100111", Some("789")), + ActivityLogMode::ErrorsOnly + ); + assert_eq!( + resolve_channel_activity_log_from_config(&config, "-100111", Some("456")), + ActivityLogMode::ErrorsOnly + ); + assert_eq!( + resolve_channel_activity_log_from_config(&serde_json::json!({}), "chat", None), + ActivityLogMode::All + ); +} + +#[tokio::test] +async fn channel_activity_log_resolution_reads_store_without_status_probe() { + let store = MemoryChannelStore { + channels: HashMap::from([( + ("telegram".to_string(), "bot1".to_string()), + StoredChannel { + account_id: "bot1".to_string(), + channel_type: "telegram".to_string(), + config: serde_json::json!({ + "activity_log": "errors_only", + "channel_overrides": { + "-100999": { "activity_log": "off" } + }, + "user_overrides": { + "123": { "activity_log": "all" } + } + }), + created_at: 1, + updated_at: 1, + }, + )]), + }; + let state = test_state_with_channel_store(Arc::new(store)); + let target = ChannelReplyTarget { + channel_type: ChannelType::Telegram, + account_id: "bot1".into(), + chat_id: "-100999".into(), + message_id: None, + thread_id: None, + sender_id: Some("789".into()), + activity_log: ActivityLogMode::All, + }; + + let mode = resolve_channel_activity_log(&state, &target, target.sender_id.as_deref()).await; + + assert_eq!(mode, ActivityLogMode::Off); +} + +#[tokio::test] +async fn channel_activity_log_resolution_falls_back_to_live_config_without_status_probe() { + let store = MemoryChannelStore { + channels: HashMap::new(), + }; + let registry = test_registry_with_account_config(serde_json::json!({ + "activity_log": "errors_only", + "channel_overrides": { + "-100999": { "activity_log": "off" } + }, + "user_overrides": { + "123": { "activity_log": "all" } + } + })) + .await; + let state = test_state_with_channel_store_and_registry(Arc::new(store), registry); + let target = ChannelReplyTarget { + channel_type: ChannelType::Telegram, + account_id: "bot1".into(), + chat_id: "-100111".into(), + message_id: None, + thread_id: None, + sender_id: Some("789".into()), + activity_log: ActivityLogMode::All, + }; + + let mode = resolve_channel_activity_log(&state, &target, target.sender_id.as_deref()).await; + + assert_eq!(mode, ActivityLogMode::ErrorsOnly); +} + +#[tokio::test] +async fn channel_activity_log_resolution_prefers_live_config_for_active_accounts() { + let store = MemoryChannelStore { + channels: HashMap::from([( + ("telegram".to_string(), "bot1".to_string()), + StoredChannel { + account_id: "bot1".to_string(), + channel_type: "telegram".to_string(), + config: serde_json::json!({ + "activity_log": "all", + "channel_overrides": { + "-100999": { "activity_log": "all" } + } + }), + created_at: 1, + updated_at: 1, + }, + )]), + }; + let registry = test_registry_with_account_config(serde_json::json!({ + "activity_log": "errors_only", + "channel_overrides": { + "-100999": { "activity_log": "off" } + } + })) + .await; + let state = test_state_with_channel_store_and_registry(Arc::new(store), registry); + let target = ChannelReplyTarget { + channel_type: ChannelType::Telegram, + account_id: "bot1".into(), + chat_id: "-100999".into(), + message_id: None, + thread_id: None, + sender_id: Some("789".into()), + activity_log: ActivityLogMode::All, + }; + + let mode = resolve_channel_activity_log(&state, &target, target.sender_id.as_deref()).await; + + assert_eq!(mode, ActivityLogMode::Off); +} + +#[tokio::test] +async fn channel_session_defaults_fall_back_to_live_config_without_status_probe() { + let store = MemoryChannelStore { + channels: HashMap::new(), + }; + let registry = test_registry_with_account_config(serde_json::json!({ + "model": "toml-model", + "agent_id": "toml-agent", + "channel_overrides": { + "-100999": { + "model": "channel-model", + "agent_id": "channel-agent" + } + } + })) + .await; + let state = test_state_with_channel_store_and_registry(Arc::new(store), registry); + let target = ChannelReplyTarget { + channel_type: ChannelType::Telegram, + account_id: "bot1".into(), + chat_id: "-100999".into(), + message_id: None, + thread_id: None, + sender_id: Some("789".into()), + activity_log: ActivityLogMode::All, + }; + + let defaults = + resolve_channel_session_defaults(&state, &target, target.sender_id.as_deref()).await; + + assert_eq!(defaults.model.as_deref(), Some("channel-model")); + assert_eq!(defaults.agent_id.as_deref(), Some("channel-agent")); +} + +#[tokio::test] +async fn handle_sessions_refreshes_target_binding_sender_id() { + let metadata = sqlite_session_metadata().await; + let state = GatewayState::new( + ResolvedAuth { + mode: AuthMode::Token, + token: None, + password: None, + }, + GatewayServices::noop(), + ); + + let current_binding = ChannelReplyTarget { + channel_type: ChannelType::Telegram, + account_id: "bot1".into(), + chat_id: "-100999".into(), + message_id: Some("1".into()), + thread_id: None, + sender_id: Some("old-user".into()), + activity_log: ActivityLogMode::All, + }; + let target_binding = ChannelReplyTarget { + message_id: Some("2".into()), + sender_id: Some("previous-user".into()), + ..current_binding.clone() + }; + metadata + .upsert("telegram:bot1:-100999", Some("Session 1".into())) + .await + .unwrap(); + metadata + .set_channel_binding( + "telegram:bot1:-100999", + Some(serde_json::to_string(¤t_binding).unwrap()), + ) + .await; + tokio::time::sleep(std::time::Duration::from_millis(2)).await; + metadata + .upsert("session:old", Some("Session 2".into())) + .await + .unwrap(); + metadata + .set_channel_binding( + "session:old", + Some(serde_json::to_string(&target_binding).unwrap()), + ) + .await; + + let reply_to = ChannelReplyTarget { + message_id: Some("99".into()), + sender_id: Some("fresh-user".into()), + ..current_binding + }; + + let result = commands::session_handlers::handle_sessions( + &state, + &metadata, + "telegram:bot1:-100999", + &reply_to, + "2", + ) + .await + .unwrap(); + + assert_eq!(result, "Switched to: Session 2"); + let updated = metadata.get("session:old").await.unwrap(); + let binding: ChannelReplyTarget = + serde_json::from_str(&updated.channel_binding.unwrap()).unwrap(); + assert_eq!(binding.sender_id.as_deref(), Some("fresh-user")); + assert_eq!(binding.message_id.as_deref(), Some("2")); +} + #[test] fn channel_event_serialization_nulls() { let event = ChannelEvent::InboundMessage { @@ -150,7 +685,7 @@ fn unique_providers_skips_entries_without_provider() { let models = vec![ serde_json::json!({"id": "m1"}), serde_json::json!({"id": "m2", "provider": "openai"}), - serde_json::json!({"id": "m3", "provider": serde_json::Value::Null}), + serde_json::json!({"id": "m3", "provider": Value::Null}), ]; assert_eq!(unique_providers(&models), vec!["openai"]); } diff --git a/crates/gateway/src/chat.rs b/crates/gateway/src/chat.rs index f27fad7cde..b89553350e 100644 --- a/crates/gateway/src/chat.rs +++ b/crates/gateway/src/chat.rs @@ -7,7 +7,10 @@ use std::sync::Arc; use {async_trait::async_trait, serde_json::Value}; -use {moltis_channels::ChannelReplyTarget, moltis_tools::sandbox::SandboxRouter}; +use { + moltis_channels::{ChannelReplyTarget, ChannelStatusLogEntry}, + moltis_tools::sandbox::SandboxRouter, +}; use crate::state::GatewayState; @@ -55,13 +58,11 @@ impl ChatRuntime for GatewayChatRuntime { // ── Channel status log ────────────────────────────────────────────────── - async fn push_channel_status_log(&self, session_key: &str, message: String) { - self.state - .push_channel_status_log(session_key, message) - .await; + async fn push_channel_status_log(&self, session_key: &str, entry: ChannelStatusLogEntry) { + self.state.push_channel_status_log(session_key, entry).await; } - async fn drain_channel_status_log(&self, session_key: &str) -> Vec { + async fn drain_channel_status_log(&self, session_key: &str) -> Vec { self.state.drain_channel_status_log(session_key).await } diff --git a/crates/gateway/src/session/tests.rs b/crates/gateway/src/session/tests.rs index 73be5cc0a1..f3d4f37baf 100644 --- a/crates/gateway/src/session/tests.rs +++ b/crates/gateway/src/session/tests.rs @@ -990,6 +990,8 @@ mod tests { chat_id: "-100123".to_string(), message_id: Some("9".to_string()), thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }) .unwrap(); metadata.set_channel_binding(key, Some(binding_json)).await; diff --git a/crates/gateway/src/state.rs b/crates/gateway/src/state.rs index 857f9f56a3..a7005cd2c6 100644 --- a/crates/gateway/src/state.rs +++ b/crates/gateway/src/state.rs @@ -76,7 +76,10 @@ use moltis_protocol::{ConnectParams, EventFrame}; use moltis_tools::sandbox::SandboxRouter; -use {moltis_channels::ChannelReplyTarget, moltis_sessions::session_events::SessionEventBus}; +use { + moltis_channels::{ChannelReplyTarget, ChannelStatusLogEntry}, + moltis_sessions::session_events::SessionEventBus, +}; use crate::{ auth::{CredentialStore, ResolvedAuth}, @@ -323,7 +326,7 @@ pub struct GatewayInner { pub cached_location: Option, /// Per-session buffer for channel status messages (tool use, model selection). /// Drained when the final response is delivered to the channel. - pub channel_status_log: HashMap>, + pub channel_status_log: HashMap>, /// Sessions currently in channel command mode (/sh passthrough). pub channel_command_mode_sessions: HashSet, /// Sessions with fast/priority mode enabled. @@ -706,7 +709,14 @@ impl GatewayState { } /// Push a reply target for a session (used when a channel message triggers chat.send). - pub async fn push_channel_reply(&self, session_key: &str, target: ChannelReplyTarget) { + pub async fn push_channel_reply(&self, session_key: &str, mut target: ChannelReplyTarget) { + let sender_id = target.sender_id.clone(); + target.activity_log = crate::channel_events::resolve_channel_activity_log( + self, + &target, + sender_id.as_deref(), + ) + .await; self.inner .write() .await @@ -767,7 +777,7 @@ impl GatewayState { /// status log for a session. These are drained and appended as a logbook /// when the final response is delivered. Capped at 100 entries per session /// to prevent unbounded growth. - pub async fn push_channel_status_log(&self, session_key: &str, message: String) { + pub async fn push_channel_status_log(&self, session_key: &str, entry: ChannelStatusLogEntry) { const MAX_STATUS_LOG_ENTRIES: usize = 100; let mut inner = self.inner.write().await; let log = inner @@ -777,11 +787,11 @@ impl GatewayState { if log.len() >= MAX_STATUS_LOG_ENTRIES { log.drain(..log.len() - MAX_STATUS_LOG_ENTRIES + 1); } - log.push(message); + log.push(entry); } /// Drain all buffered status log entries for a session. - pub async fn drain_channel_status_log(&self, session_key: &str) -> Vec { + pub async fn drain_channel_status_log(&self, session_key: &str) -> Vec { self.inner .write() .await diff --git a/crates/matrix/src/handler/implementation.rs b/crates/matrix/src/handler/implementation.rs index 36f668c5af..37075e7a9c 100644 --- a/crates/matrix/src/handler/implementation.rs +++ b/crates/matrix/src/handler/implementation.rs @@ -304,11 +304,13 @@ pub async fn handle_room_message( account_id: account_id.clone(), chat_id: room_id.clone(), thread_id: None, + sender_id: None, message_id: if config.reply_to_message { Some(event_id.clone()) } else { None }, + activity_log: moltis_channels::ActivityLogMode::All, }; let meta = ChannelMessageMeta { @@ -573,7 +575,9 @@ pub async fn handle_poll_response( account_id: account_id.clone(), chat_id: room_id, thread_id: None, + sender_id: None, message_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; if let Err(error) = sink.dispatch_interaction(&callback_data, reply_to).await { diff --git a/crates/msteams/src/plugin.rs b/crates/msteams/src/plugin.rs index fa567eeefd..c9d1e1640e 100644 --- a/crates/msteams/src/plugin.rs +++ b/crates/msteams/src/plugin.rs @@ -385,6 +385,8 @@ impl MsTeamsPlugin { chat_id: chat_id.clone(), message_id, thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; let Some(sink) = event_sink else { diff --git a/crates/nostr/src/bus.rs b/crates/nostr/src/bus.rs index c5092a887f..c63ebdfc80 100644 --- a/crates/nostr/src/bus.rs +++ b/crates/nostr/src/bus.rs @@ -314,6 +314,8 @@ async fn handle_event( chat_id: sender_hex.clone(), message_id: Some(event.id.to_hex()), thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; if let Some(cmd_text) = text.strip_prefix('/') { diff --git a/crates/signal/src/inbound.rs b/crates/signal/src/inbound.rs index 259ddd0516..72662e2e1c 100644 --- a/crates/signal/src/inbound.rs +++ b/crates/signal/src/inbound.rs @@ -123,6 +123,8 @@ pub async fn handle_event( chat_id: chat_id.clone(), message_id, thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; if let Some(cmd_text) = text.strip_prefix('/') { diff --git a/crates/slack/src/socket.rs b/crates/slack/src/socket.rs index faef2eedf0..6da4ad6943 100644 --- a/crates/slack/src/socket.rs +++ b/crates/slack/src/socket.rs @@ -277,6 +277,8 @@ async fn command_events_callback( chat_id: event.channel_id.to_string(), message_id: None, thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; match sink .dispatch_command(&full_command, reply_to, Some(&sender_id)) @@ -344,6 +346,8 @@ async fn interaction_events_callback( chat_id: channel_id, message_id: None, thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; match sink.dispatch_interaction(&action_id, reply_to).await { Ok(_response) => { @@ -558,6 +562,8 @@ pub(crate) async fn handle_inbound( chat_id: channel_id.to_string(), message_id: thread_ts, thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; let meta = ChannelMessageMeta { diff --git a/crates/slack/src/webhook.rs b/crates/slack/src/webhook.rs index 03c22d4d6a..a922c8f97c 100644 --- a/crates/slack/src/webhook.rs +++ b/crates/slack/src/webhook.rs @@ -265,6 +265,8 @@ pub async fn handle_verified_interaction_webhook( chat_id: channel_id.to_string(), message_id: None, thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; match sink.dispatch_interaction(action_id, reply_to).await { Ok(_) => {}, @@ -315,6 +317,8 @@ pub async fn handle_verified_command_webhook( chat_id: channel_id, message_id: None, thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; let sender = if user_id.is_empty() { None @@ -496,6 +500,8 @@ pub async fn handle_interaction_webhook( chat_id: channel_id.to_string(), message_id: None, thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; match sink.dispatch_interaction(action_id, reply_to).await { Ok(_) => {}, diff --git a/crates/telegram/src/config.rs b/crates/telegram/src/config.rs index e4c91a2abc..4ce77a70a5 100644 --- a/crates/telegram/src/config.rs +++ b/crates/telegram/src/config.rs @@ -2,6 +2,7 @@ use std::collections::HashMap; use { moltis_channels::{ + ActivityLogMode, config_view::ChannelConfigView, gating::{DmPolicy, GroupPolicy, MentionMode}, }, @@ -19,6 +20,8 @@ pub struct ChannelOverride { pub model_provider: Option, #[serde(skip_serializing_if = "Option::is_none")] pub agent_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub activity_log: Option, } /// Per-user model/provider override. @@ -30,6 +33,8 @@ pub struct UserOverride { pub model_provider: Option, #[serde(skip_serializing_if = "Option::is_none")] pub agent_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub activity_log: Option, } /// How streaming responses are delivered. @@ -80,6 +85,9 @@ pub struct TelegramAccountConfig { /// streamed message. Helps avoid early push notifications with tiny drafts. pub stream_min_initial_chars: usize, + /// User-facing activity log visibility for replies from this account. + pub activity_log: ActivityLogMode, + /// Default model ID for this bot's sessions (e.g. "claude-sonnet-4-5-20250929"). /// When set, channel messages use this model instead of the first registered provider. #[serde(skip_serializing_if = "Option::is_none")] @@ -132,7 +140,7 @@ pub struct RedactedConfig<'a>(pub &'a TelegramAccountConfig); impl Serialize for RedactedConfig<'_> { fn serialize(&self, serializer: S) -> Result { let c = self.0; - let mut count = 13; // always-present fields + let mut count = 14; // always-present fields count += c.model.is_some() as usize; count += c.model_provider.is_some() as usize; count += c.agent_id.is_some() as usize; @@ -149,6 +157,7 @@ impl Serialize for RedactedConfig<'_> { s.serialize_field("edit_throttle_ms", &c.edit_throttle_ms)?; s.serialize_field("stream_notify_on_complete", &c.stream_notify_on_complete)?; s.serialize_field("stream_min_initial_chars", &c.stream_min_initial_chars)?; + s.serialize_field("activity_log", &c.activity_log)?; if c.model.is_some() { s.serialize_field("model", &c.model)?; } @@ -200,6 +209,10 @@ impl ChannelConfigView for TelegramAccountConfig { self.agent_id.as_deref() } + fn activity_log(&self) -> ActivityLogMode { + self.activity_log + } + fn channel_model(&self, channel_id: &str) -> Option<&str> { self.channel_overrides .get(channel_id) @@ -218,6 +231,12 @@ impl ChannelConfigView for TelegramAccountConfig { .and_then(|o| o.agent_id.as_deref()) } + fn channel_activity_log(&self, channel_id: &str) -> Option { + self.channel_overrides + .get(channel_id) + .and_then(|o| o.activity_log) + } + fn user_model(&self, user_id: &str) -> Option<&str> { self.user_overrides .get(user_id) @@ -235,6 +254,12 @@ impl ChannelConfigView for TelegramAccountConfig { .get(user_id) .and_then(|o| o.agent_id.as_deref()) } + + fn user_activity_log(&self, user_id: &str) -> Option { + self.user_overrides + .get(user_id) + .and_then(|o| o.activity_log) + } } impl Default for TelegramAccountConfig { @@ -250,6 +275,7 @@ impl Default for TelegramAccountConfig { edit_throttle_ms: 300, stream_notify_on_complete: false, stream_min_initial_chars: 30, + activity_log: ActivityLogMode::All, model: None, model_provider: None, agent_id: None, @@ -279,6 +305,7 @@ mod tests { assert_eq!(cfg.edit_throttle_ms, 300); assert!(!cfg.stream_notify_on_complete); assert_eq!(cfg.stream_min_initial_chars, 30); + assert_eq!(cfg.activity_log, ActivityLogMode::All); } #[test] @@ -360,15 +387,47 @@ mod tests { ); } + #[test] + fn resolve_activity_log_user_overrides_channel() { + use moltis_channels::ActivityLogMode; + + let mut cfg = TelegramAccountConfig { + activity_log: ActivityLogMode::All, + ..Default::default() + }; + cfg.channel_overrides + .insert("-100123".into(), ChannelOverride { + activity_log: Some(ActivityLogMode::ErrorsOnly), + ..Default::default() + }); + cfg.user_overrides.insert("456".into(), UserOverride { + activity_log: Some(ActivityLogMode::Off), + ..Default::default() + }); + + assert_eq!(cfg.user_activity_log("456"), Some(ActivityLogMode::Off)); + assert_eq!( + cfg.channel_activity_log("-100123"), + Some(ActivityLogMode::ErrorsOnly) + ); + assert_eq!(cfg.activity_log(), ActivityLogMode::All); + } + #[test] fn overrides_round_trip() { let json = serde_json::json!({ "token": "123:ABC", + "activity_log": "errors_only", "channel_overrides": { "-100123": { "model": "gpt-4", "agent_id": "group-agent" } }, "user_overrides": { - "456": { "model": "claude-sonnet", "model_provider": "anthropic", "agent_id": "user-agent" } + "456": { + "model": "claude-sonnet", + "model_provider": "anthropic", + "agent_id": "user-agent", + "activity_log": "off" + } } }); let cfg: TelegramAccountConfig = serde_json::from_value(json).unwrap(); @@ -376,6 +435,8 @@ mod tests { assert!(cfg.channel_model_provider("-100123").is_none()); assert_eq!(cfg.channel_agent_id("-100123"), Some("group-agent")); assert_eq!(cfg.user_model("456"), Some("claude-sonnet")); + assert_eq!(cfg.activity_log, ActivityLogMode::ErrorsOnly); + assert_eq!(cfg.user_activity_log("456"), Some(ActivityLogMode::Off)); assert_eq!(cfg.user_model_provider("456"), Some("anthropic")); assert_eq!(cfg.user_agent_id("456"), Some("user-agent")); diff --git a/crates/telegram/src/handlers/callbacks.rs b/crates/telegram/src/handlers/callbacks.rs index 264780b437..20c854bda9 100644 --- a/crates/telegram/src/handlers/callbacks.rs +++ b/crates/telegram/src/handlers/callbacks.rs @@ -394,6 +394,8 @@ pub(super) async fn handle_callback_query( chat_id: chat_id.clone(), message_id: None, thread_id: callback_thread_id, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; let outbound_to = reply_target.outbound_to().into_owned(); diff --git a/crates/telegram/src/handlers/implementation.rs b/crates/telegram/src/handlers/implementation.rs index a6c24a4b18..6dd770dc86 100644 --- a/crates/telegram/src/handlers/implementation.rs +++ b/crates/telegram/src/handlers/implementation.rs @@ -58,6 +58,8 @@ fn reply_target_for_msg(account_id: &str, msg: &Message) -> ChannelReplyTarget { chat_id: msg.chat.id.0.to_string(), message_id: Some(msg.id.0.to_string()), thread_id: extract_thread_id(msg), + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, } } @@ -458,6 +460,8 @@ pub async fn handle_message_direct( chat_id: msg.chat.id.0.to_string(), message_id: Some(msg.id.0.to_string()), thread_id: extract_thread_id(&msg), + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; sink.update_location(&reply_target, lat, lon).await } else { @@ -839,6 +843,8 @@ pub async fn handle_edited_location( chat_id: msg.chat.id.0.to_string(), message_id: Some(msg.id.0.to_string()), thread_id: extract_thread_id(&msg), + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; sink.update_location(&reply_target, lat, lon).await; } @@ -931,6 +937,8 @@ pub async fn handle_callback_query( chat_id: chat_id.clone(), message_id: None, // Callback queries don't have a message to reply-thread to. thread_id: callback_thread_id, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; let outbound_to = reply_target.outbound_to().into_owned(); diff --git a/crates/telephony/src/plugin.rs b/crates/telephony/src/plugin.rs index d2bfb639e3..9937abb0b2 100644 --- a/crates/telephony/src/plugin.rs +++ b/crates/telephony/src/plugin.rs @@ -99,6 +99,8 @@ impl TelephonyPlugin { chat_id: call_id.to_string(), message_id: None, thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; let meta = moltis_channels::ChannelMessageMeta { diff --git a/crates/web/ui/src/types/channel.ts b/crates/web/ui/src/types/channel.ts index 1a1ec92c69..ce50d266ce 100644 --- a/crates/web/ui/src/types/channel.ts +++ b/crates/web/ui/src/types/channel.ts @@ -34,6 +34,12 @@ export const ChannelType = { Telephony: "telephony" as const, } satisfies Record; +/** + * User-facing Activity log visibility. + * Serialised as snake_case via `#[serde(rename_all = "snake_case")]`. + */ +export type ActivityLogMode = "all" | "errors_only" | "off"; + /** * How a channel receives inbound messages. * Serialised as snake_case via `#[serde(rename_all = "snake_case")]`. @@ -76,6 +82,8 @@ export interface ChannelReplyTarget { chat_id: string; message_id?: string; thread_id?: string; + sender_id?: string; + activity_log?: ActivityLogMode; } /** diff --git a/crates/whatsapp/src/handlers.rs b/crates/whatsapp/src/handlers.rs index 2f8528c330..491bff1469 100644 --- a/crates/whatsapp/src/handlers.rs +++ b/crates/whatsapp/src/handlers.rs @@ -400,6 +400,8 @@ async fn handle_message( chat_id: chat_id.clone(), message_id: Some(info.id.to_string()), thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; if let Some(ref sink) = state.event_sink { match sink.dispatch_command(cmd, reply_to, Some(&peer_id)).await { @@ -431,6 +433,8 @@ async fn handle_message( chat_id: chat_id.clone(), message_id: Some(info.id.to_string()), thread_id: None, + sender_id: None, + activity_log: moltis_channels::ActivityLogMode::All, }; let meta = ChannelMessageMeta { channel_type: ChannelType::Whatsapp, diff --git a/docs/src/telegram.md b/docs/src/telegram.md index b62acbbb3b..b40355de24 100644 --- a/docs/src/telegram.md +++ b/docs/src/telegram.md @@ -84,6 +84,7 @@ offered = ["telegram"] | `stream_mode` | no | `"edit_in_place"` | Streaming mode: `"edit_in_place"` or `"off"` | | `edit_throttle_ms` | no | `300` | Minimum milliseconds between streaming edit updates | | `stream_notify_on_complete` | no | `false` | Send a completion notification after streaming finishes | +| `activity_log` | no | `"all"` | Show tool/status Activity log entries after replies: `"all"`, `"errors_only"`, or `"off"` | | `stream_min_initial_chars` | no | `30` | Minimum characters before sending the first streamed message | ```admonish important title="Allowlist values are strings" @@ -113,31 +114,39 @@ agent_id = "research" otp_self_approval = true stream_mode = "edit_in_place" edit_throttle_ms = 300 +activity_log = "errors_only" ``` -### Per-User and Per-Channel Model and Agent Overrides +### Per-User and Per-Channel Overrides -You can override the model or agent for specific users or group chats: +You can override the model, agent, or Activity log visibility for specific users +or group chats: ```toml [channels.telegram.my-bot] token = "..." model = "claude-sonnet-4-20250514" model_provider = "anthropic" +activity_log = "all" [channels.telegram.my-bot.channel_overrides."-1001234567890"] model = "gpt-4o" model_provider = "openai" agent_id = "triage" +activity_log = "off" [channels.telegram.my-bot.user_overrides."123456789"] model = "claude-opus-4-20250514" model_provider = "anthropic" agent_id = "research" +activity_log = "errors_only" ``` +`activity_log = "all"` preserves the default behavior, `"errors_only"` sends +only failed tool/status entries, and `"off"` disables Activity log messages. User overrides take priority over channel overrides, which take priority over -the account default, for both model selection and agent selection. +the account default, for model selection, agent selection, and Activity log +visibility. ## Access Control