src/gateway/mod.rs

@@ -228,6 +228,7 @@ impl GatewayState {
             let sched = Arc::new(Scheduler::new(
                 self.storage.clone(),
                 self.session_manager.clone(),
+                self.channel_manager.bus(),
                 scheduler_config,
             ));
             tokio::spawn(async move {

src/providers/anthropic.rs

@@ -2,7 +2,6 @@ use async_trait::async_trait;
 use reqwest::Client;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
-use std::time::Duration;
 
 use crate::bus::message::ContentBlock;
 use super::{ChatCompletionRequest, ChatCompletionResponse, LLMProvider, Tool, ToolCall};
@@ -10,8 +9,6 @@ use super::traits::Usage;
 use std::sync::Arc;
 use crate::storage::Storage;
 
-const LLM_REQUEST_TIMEOUT_SECS: u64 = 300;
-
 fn convert_content_blocks(blocks: &[ContentBlock]) -> Vec<serde_json::Value> {
     blocks.iter().map(|b| match b {
         ContentBlock::Text { text } => {
@@ -75,10 +72,7 @@ impl AnthropicProvider {
         model_extra: HashMap<String, serde_json::Value>,
     ) -> Self {
         Self {
-            client: Client::builder()
-                .timeout(Duration::from_secs(LLM_REQUEST_TIMEOUT_SECS))
-                .build()
-                .unwrap_or_else(|_| Client::new()),
+            client: Client::new(),
             name,
             api_key,
             base_url,
@@ -244,19 +238,7 @@ impl LLMProvider for AnthropicProvider {
         let req_body_str = serde_json::to_string_pretty(&body).unwrap_or_default();
         tracing::debug!(req_body = %req_body_str, "LLM request");
 
-        let resp = req_builder.json(&body).send().await
-            .inspect_err(|e| {
-                let is_timeout = e.is_timeout();
-                tracing::error!(
-                    provider = %self.name,
-                    model = %self.model_id,
-                    url = %url,
-                    timeout = is_timeout,
-                    error = %e,
-                    elapsed_ms = %start.elapsed().as_millis(),
-                    "LLM API request failed"
-                );
-            })?;
+        let resp = req_builder.json(&body).send().await?;
 
         let status = resp.status();
         let body_text = resp.text().await?;
@@ -272,14 +254,6 @@ impl LLMProvider for AnthropicProvider {
                         .map(|s| s.to_string())
                 })
                 .unwrap_or_else(|| body_text.clone());
-            tracing::error!(
-                provider = %self.name,
-                model = %self.model_id,
-                http_status = %status,
-                error = %error_msg,
-                elapsed_ms = %start.elapsed().as_millis(),
-                "LLM API returned error"
-            );
             if let Some(ref storage) = self.storage {
                 let _ = storage.append_llm_call(
                     &self.name, &self.model_id, &req_body_str,

src/providers/openai.rs

@@ -3,7 +3,6 @@ use reqwest::Client;
 use serde::Deserialize;
 use serde_json::{json, Value};
 use std::collections::HashMap;
-use std::time::Duration;
 
 use crate::bus::message::ContentBlock;
 use super::{ChatCompletionRequest, ChatCompletionResponse, LLMProvider, ToolCall};
@@ -11,8 +10,6 @@ use super::traits::Usage;
 use std::sync::Arc;
 use crate::storage::Storage;
 
-const LLM_REQUEST_TIMEOUT_SECS: u64 = 300;
-
 fn convert_content_blocks(blocks: &[ContentBlock]) -> Value {
     if blocks.len() == 1
         && let ContentBlock::Text { text } = &blocks[0] {
@@ -51,10 +48,7 @@ impl OpenAIProvider {
         model_extra: HashMap<String, serde_json::Value>,
     ) -> Self {
         Self {
-            client: Client::builder()
-                .timeout(Duration::from_secs(LLM_REQUEST_TIMEOUT_SECS))
-                .build()
-                .unwrap_or_else(|_| Client::new()),
+            client: Client::new(),
             name,
             api_key,
             base_url,
@@ -212,19 +206,7 @@ impl LLMProvider for OpenAIProvider {
         let req_body_str = serde_json::to_string_pretty(&body).unwrap_or_default();
         tracing::debug!(req_body = %req_body_str, "LLM request");
 
-        let resp = req_builder.json(&body).send().await
-            .inspect_err(|e| {
-                let is_timeout = e.is_timeout();
-                tracing::error!(
-                    provider = %self.name,
-                    model = %self.model_id,
-                    url = %url,
-                    timeout = is_timeout,
-                    error = %e,
-                    elapsed_ms = %start.elapsed().as_millis(),
-                    "LLM API request failed"
-                );
-            })?;
+        let resp = req_builder.json(&body).send().await?;
 
         let status = resp.status();
         let text = resp.text().await?;
@@ -232,14 +214,6 @@ impl LLMProvider for OpenAIProvider {
 
         if !status.is_success() {
             let error = format!("API error {}: {}", status, text);
-            tracing::error!(
-                provider = %self.name,
-                model = %self.model_id,
-                http_status = %status,
-                error = %error,
-                elapsed_ms = %start.elapsed().as_millis(),
-                "LLM API returned error"
-            );
             if let Some(ref storage) = self.storage
                 && let Err(e) = storage.append_llm_call(
                     &self.name, &self.model_id, &req_body_str,

src/scheduler/mod.rs

@@ -4,12 +4,11 @@ use std::sync::Arc;
 use std::time::Instant;
 use tokio::time;
 
+use crate::bus::MessageBus;
 use crate::config::SchedulerConfig;
 use crate::session::session::HandleResult;
 use crate::session::SessionManager;
-use crate::storage::ScheduledJob;
-use crate::storage::Storage;
-use crate::storage::JobRun;
+use crate::storage::{JobRun, ScheduledJob, Storage};
 
 pub use types::Schedule;
 
@@ -54,6 +53,7 @@ fn now_ms() -> i64 {
 pub struct Scheduler {
     storage: Arc<Storage>,
     session_manager: Arc<SessionManager>,
+    bus: Arc<MessageBus>,
     config: SchedulerConfig,
 }
 
@@ -61,11 +61,13 @@ impl Scheduler {
     pub fn new(
         storage: Arc<Storage>,
         session_manager: Arc<SessionManager>,
+        bus: Arc<MessageBus>,
         config: SchedulerConfig,
     ) -> Self {
         Self {
             storage,
             session_manager,
+            bus,
             config,
         }
     }
@@ -134,6 +136,18 @@ impl Scheduler {
 
                 match result {
                     Ok(HandleResult::AgentResponse(output)) => {
+                        let outbound = crate::bus::OutboundMessage {
+                            channel: job.channel.clone(),
+                            chat_id: job.chat_id.clone(),
+                            content: output.clone(),
+                            reply_to: None,
+                            media: vec![],
+                            metadata: std::collections::HashMap::new(),
+                        };
+                        if let Err(e) = self.bus.publish_outbound(outbound).await {
+                            tracing::warn!(job_id = %job.id, "scheduler: failed to publish outbound: {}", e);
+                        }
+
                         let output_truncated = if output.len() > 8000 {
                             format!("{}...[truncated]", &output[..output.ceil_char_boundary(8000)])
                         } else {
@@ -166,6 +180,18 @@ impl Scheduler {
                         );
                     }
                     Ok(HandleResult::CommandOutput(output)) => {
+                        let outbound = crate::bus::OutboundMessage {
+                            channel: job.channel.clone(),
+                            chat_id: job.chat_id.clone(),
+                            content: output.clone(),
+                            reply_to: None,
+                            media: vec![],
+                            metadata: std::collections::HashMap::new(),
+                        };
+                        if let Err(e) = self.bus.publish_outbound(outbound).await {
+                            tracing::warn!(job_id = %job.id, "scheduler: failed to publish outbound: {}", e);
+                        }
+
                         let run = JobRun {
                             id: 0,
                             job_id: job.id.clone(),

src/session/session.rs

@@ -7,10 +7,6 @@ use crate::bus::{ChatMessage, MediaItem, MessageSource, OutboundMessage, SourceK
 use crate::storage::{Storage, StorageError};
 use std::sync::Arc as StdArc;
 
-tokio::task_local! {
-    static CURRENT_SOURCE_SESSION: Option<String>;
-}
-
 /// Result of handling a message - either an AI response or a command output
 pub enum HandleResult {
     /// AI response to be sent as AssistantResponse
@@ -721,6 +717,7 @@ pub struct SessionManager {
     skills_loader: Arc<SkillsLoader>,
     storage: Arc<Storage>,
     bus: Arc<MessageBus>,
+    current_source_session: Arc<Mutex<Option<String>>>,
     memory_manager: Arc<crate::memory::MemoryManager>,
 }
 
@@ -825,6 +822,7 @@ impl SessionManager {
             skills_loader,
             storage,
             bus,
+            current_source_session: Arc::new(Mutex::new(None)),
             memory_manager,
         })
     }
@@ -839,19 +837,6 @@ impl SessionManager {
         self.tools.clone()
     }
 
-    /// 为定时任务创建一个无 session 绑定的 AgentLoop
-    pub fn create_cron_agent(&self) -> Result<AgentLoop, AgentError> {
-        let provider = create_provider(self.provider_config.clone())
-            .map_err(|e| AgentError::Other(format!("failed to create cron provider: {}", e)))?;
-        Ok(AgentLoop::with_provider_and_tools(
-            Arc::from(provider),
-            self.tools.clone(),
-            self.provider_config.max_tool_iterations,
-            self.provider_config.model_id.clone(),
-            self.provider_config.workspace_dir.clone(),
-        ).with_context_window(self.provider_config.token_limit))
-    }
-
     /// 获取所有可用的斜杠命令
     pub fn get_slash_commands(&self) -> &[SlashCommand] {
         SLASH_COMMANDS
@@ -1288,10 +1273,10 @@ impl SessionManager {
         media: Vec<MediaItem>,
     ) -> Result<HandleResult, AgentError> {
         let unified_id = self.resolve_dialog_id(channel, chat_id).await?;
+        *self.current_source_session.lock().await = Some(unified_id.to_string());
         tracing::debug!(unified_id = %unified_id, "handle_message resolved unified_id");
         let session = self.get_or_create_session(&unified_id).await?;
 
-        CURRENT_SOURCE_SESSION.scope(Some(unified_id.to_string()), async {
         // Check for slash command
         if let Some((cmd_name, cmd_args)) = parse_slash_command(content) {
             let result = self.execute_slash_command(
@@ -1302,10 +1287,16 @@ impl SessionManager {
                 Some(&unified_id),
             ).await;
 
-                return match result {
-                    Ok((_new_session_id, response)) => Ok(HandleResult::CommandOutput(response)),
-                    Err(e) => Ok(HandleResult::CommandOutput(e.to_string())),
-                };
+            match result {
+                Ok((_new_session_id, response)) => {
+                    *self.current_source_session.lock().await = None;
+                    return Ok(HandleResult::CommandOutput(response));
+                }
+                Err(e) => {
+                    *self.current_source_session.lock().await = None;
+                    return Ok(HandleResult::CommandOutput(e.to_string()));
+                }
+            }
         }
 
         // Normal message handling through LLM
@@ -1333,8 +1324,7 @@ impl SessionManager {
             });
         }
 
-            // Phase 1: prepare data under session lock
-            let (agent, history, system_prompt) = {
+        let response: String = {
             let mut session_guard = session.lock().await;
 
             let media_refs: Vec<String> = media.iter().map(|m| m.path.clone()).collect();
@@ -1349,8 +1339,10 @@ impl SessionManager {
 
             let history = session_guard.get_history().to_vec();
 
+            // Build skills prompt
             let skills_prompt = self.skills_loader.build_skills_prompt();
 
+            // Fetch memory context
             let memory_context = match self.memory_manager.recall(content, 5, Some(crate::memory::MemoryCategory::Knowledge), None).await {
                 Ok(entries) if !entries.is_empty() => {
                     Some(entries.iter()
@@ -1365,14 +1357,14 @@ impl SessionManager {
                 _ => None,
             };
 
+            // Build combined system prompt and inject at position 0 AFTER compression.
+            // This ensures AgentLoop.process() sees a system message without it participating
+            // in context compression (system prompt is dynamic and should not be persisted).
             let system_prompt = session_guard.build_system_prompt(&skills_prompt, memory_context.as_deref());
 
             let result = session_guard.compressor
                 .compress_if_needed(history)
-                    .await
-                    .inspect_err(|e| {
-                        tracing::warn!(error = %e, "Context compression failed in handle_message");
-                    })?;
+                .await?;
             if result.created_timelines {
                 session_guard.last_compressed_message_at = Some(chrono::Utc::now().timestamp_millis());
             }
@@ -1380,6 +1372,7 @@ impl SessionManager {
 
             history.insert(0, ChatMessage::system(system_prompt.clone()));
 
+            // Persist consolidation state
             let now = chrono::Utc::now().timestamp_millis();
             session_guard.last_consolidated_at = Some(now);
             if let Err(e) = session_guard.persist_session_meta().await {
@@ -1387,17 +1380,13 @@ impl SessionManager {
             }
 
             let agent = session_guard.create_agent_with_notify(notify_tx)?;
-                (agent, history, system_prompt)
-            }; // session lock released — send_message can now lock freely
 
-            // Phase 2: LLM call (no session lock held)
+            // Try LLM call; on context overflow, re-compress with tighter window and retry once.
             let result = match agent.process(history).await {
                 Ok(r) => r,
                 Err(AgentError::LlmError(ref msg))
                     if is_context_overflow_error(msg) =>
                 {
-                    let retry_history = {
-                        let mut session_guard = session.lock().await;
                     let new_window = crate::agent::ContextCompressor::parse_context_limit_from_error(msg)
                         .unwrap_or(session_guard.compressor_threshold());
                     tracing::warn!(
@@ -1415,30 +1404,18 @@ impl SessionManager {
                         }
                     }
                     let mut retry = retry_result.history;
-                        retry.insert(0, ChatMessage::system(system_prompt.clone()));
-                        retry
-                    }; // lock released again for retry
-
-                    agent.process(retry_history).await
-                        .inspect_err(|e| {
-                            tracing::error!(error = %e, "Agent retry after context compression failed");
-                        })?
+                    retry.insert(0, ChatMessage::system(system_prompt));
+                    agent.process(retry).await?
                 }
-                Err(e) => {
-                    tracing::error!(error = %e, "Agent processing error — propagating to caller");
-                    return Err(e);
-                },
+                Err(e) => return Err(e),
             };
 
-            // Phase 3: persist results under session lock
-            let response: String = {
-                let mut session_guard = session.lock().await;
-
             for msg in result.emitted_messages {
                 session_guard.add_message(msg, true).await
                     .map_err(|e| AgentError::Other(format!("persist error: {}", e)))?;
             }
 
+            // Check if we need to generate a title (after 10 user messages)
             if session_guard.should_generate_title()
                 && let Err(e) = session_guard.generate_title().await {
                     tracing::warn!("failed to generate title: {}", e);
@@ -1455,16 +1432,16 @@ impl SessionManager {
             "Agent response received"
         );
 
+        *self.current_source_session.lock().await = None;
+
         Ok(HandleResult::AgentResponse(response))
-        }).await
     }
 
     /// Handle a message triggered by a scheduled cron job.
     ///
-    /// Runs in a stateless manner: no session creation, no history persistence.
-    /// The cron system prompt instructs the LLM to deliver results via the
-    /// `send_message` tool, which handles both delivery and history writing
-    /// on the target session.
+    /// This is similar to `handle_message`, but the user message is created with
+    /// `SourceKind::ExternalTrigger` source metadata so that the cron job identity
+    /// is preserved in the conversation history and database.
     pub async fn handle_cron_message(
         &self,
         channel: &str,
@@ -1473,45 +1450,128 @@ impl SessionManager {
         job_id: &str,
         job_name: &str,
     ) -> Result<HandleResult, AgentError> {
+        use crate::bus::{MessageSource, SourceKind};
+
+        let unified_id = self.resolve_dialog_id(channel, chat_id).await?;
+        *self.current_source_session.lock().await = Some(unified_id.to_string());
+        tracing::debug!(unified_id = %unified_id, job_id = %job_id, "handle_cron_message resolved");
+
+        let session = self.get_or_create_session(&unified_id).await?;
+
+        let (notify_tx, mut notify_rx) = tokio::sync::mpsc::unbounded_channel();
+
+        {
+            use std::collections::HashMap;
+            use crate::bus::OutboundMessage;
+            let bus = self.bus.clone();
+            let ch = channel.to_string();
+            let cid = chat_id.to_string();
+            tokio::spawn(async move {
+                while let Some(notif) = notify_rx.recv().await {
+                    let mut metadata = HashMap::new();
+                    metadata.insert("_type".to_string(), "notification".to_string());
+                    let outbound = OutboundMessage {
+                        channel: ch.clone(),
+                        chat_id: cid.clone(),
+                        content: notif,
+                        reply_to: None,
+                        media: vec![],
+                        metadata,
+                    };
+                    let _ = bus.publish_outbound(outbound).await;
+                }
+            });
+        }
+
+        let response: String = {
+            let mut session_guard = session.lock().await;
+
+            let source = MessageSource {
+                kind: SourceKind::ExternalTrigger,
+                from_channel: Some(channel.to_string()),
+                from_session: None,
+                from_user_id: None,
+                system_name: Some(job_name.to_string()),
+                task_id: Some(job_id.to_string()),
+            };
+            let user_message = session_guard.create_user_message_with_source(prompt, vec![], source);
+            session_guard.add_message(user_message, true).await
+                .map_err(|e| AgentError::Other(format!("persist error: {}", e)))?;
+
+            let history = session_guard.get_history().to_vec();
+
             let skills_prompt = self.skills_loader.build_skills_prompt();
-
-        let base_prompt = build_system_prompt(
-            &self.provider_config.workspace_dir,
-            &self.provider_config.model_id,
-            &self.tools,
-            Some(&format!("cron:{}:{}", job_name, job_id)),
-            None,
-            false,
-        );
+            let system_prompt = session_guard.build_system_prompt(&skills_prompt, None);
             let cron_context = format!(
-            "## 定时任务执行\n\n\
-             你正在执行定时任务「{job_name}」({job_id})。\n\
-             目标渠道: {channel}:{chat_id}\n\n\
-             规则:\n\
-             - 这不是聊天对话，没有用户会直接看到你的输出\n\
-             - 你必须使用 send_message 工具将最终结果发送到目标渠道\n\
-             - send_message 格式: target_chat_id=\"{channel}:{chat_id}\", content=\"消息内容\"\n\
-             - 可以调用其他工具收集信息、处理任务，但最终消息必须通过 send_message 发送\n\
-             - 只输出最终消息内容，不要输出中间思考过程或分析！"
+                "\n\n## 定时任务执行\n\n\
+                 你正在执行定时任务「{}」({})。\n\
+                 目标渠道: {}:{}\n\n\
+                 定时任务执行规则：\n\
+                 - 这不是聊天对话，没有人会回复你，不要等待用户输入\n\
+                 - 你的职责是根据任务指令直接生成要发送的消息内容\n\
+                 - 只输出最终消息，不要输出中间思考过程或分析\n\
+                 - 系统会自动将你的回复推送到目标渠道，不要使用 send_message 工具\n\
+                 - 你的最终回复就是推送给用户的消息原文",
+                job_name, job_id, channel, chat_id
             );
-        let full_system_prompt = format!("{}\n\n{}\n\n{}", base_prompt, skills_prompt, cron_context);
+            let full_system_prompt = format!("{}{}", system_prompt, cron_context);
 
-        let history = vec![
-            ChatMessage::system(full_system_prompt),
-            ChatMessage::user(prompt),
-        ];
+            // Inject system prompt AFTER compression so it doesn't participate
+            // in context compression (system prompt is dynamic and should not be persisted).
+            let mut history = session_guard.compressor
+                .compress_if_needed(history)
+                .await?
+                .history;
 
-        let agent = self.create_cron_agent()?;
-        let source_session = format!("cron:{}", job_name);
-        let result = CURRENT_SOURCE_SESSION.scope(Some(source_session), async {
-            agent.process(history).await
-        })
-        .await
-        .inspect_err(|e| {
-            tracing::error!(error = %e, job_id = %job_id, "Cron agent processing error");
-        })?;
+            history.insert(0, ChatMessage::system(full_system_prompt));
 
-        Ok(HandleResult::AgentResponse(result.final_response.content))
+            let agent = session_guard.create_agent_with_notify(notify_tx)?;
+            let result = agent.process(history).await?;
+
+            for msg in result.emitted_messages {
+                session_guard.add_message(msg, true).await
+                    .map_err(|e| AgentError::Other(format!("persist error: {}", e)))?;
+            }
+
+            if session_guard.should_generate_title()
+                && let Err(e) = session_guard.generate_title().await {
+                    tracing::warn!("failed to generate title: {}", e);
+                }
+
+            let raw_response = result.final_response.content;
+            let prefix = format!(
+                "[message from cron:{}({})]\n",
+                job_name, job_id
+            );
+            let prefixed_response = format!("{}{}", prefix, raw_response);
+
+            let source = MessageSource {
+                kind: SourceKind::CrossChannel,
+                from_channel: Some("cron".to_string()),
+                from_session: Some(format!("{}:{}", job_name, job_id)),
+                from_user_id: None,
+                system_name: Some(job_name.to_string()),
+                task_id: Some(job_id.to_string()),
+            };
+            let msg = ChatMessage::assistant_with_source(prefixed_response.clone(), source);
+            session_guard.add_message(msg, true).await
+                .map_err(|e| AgentError::Other(format!("persist error: {}", e)))?;
+
+            prefixed_response
+        };
+
+        #[cfg(debug_assertions)]
+        tracing::debug!(
+            channel = %channel,
+            chat_id = %chat_id,
+            job_id = %job_id,
+            response_len = %response.len(),
+            "Cron agent response received"
+        );
+
+        *self.current_source_session.lock().await = None;
+
+        Ok(HandleResult::AgentResponse(response))
     }
 
     pub async fn clear_session_history(&self, unified_id: &UnifiedSessionId) -> Result<(), AgentError> {
@@ -1543,7 +1603,7 @@ impl OutboundMessenger for SessionManager {
     ) -> Result<(), String> {
         // Fill origin from current source session if not provided
         if source.from_session.is_none() {
-            source.from_session = CURRENT_SOURCE_SESSION.try_with(|v| v.clone()).ok().flatten();
+            source.from_session = self.current_source_session.lock().await.clone();
         }
 
         let (target_sid, session) = if let Some(did) = dialog_id {