feat(token): 新增缓存占比统计，聚合查询收敛到共享列清单

- 解析: OpenAIUsage 支持 DeepSeek prompt_cache_hit_tokens + OpenAI prompt_tokens_details.cached_tokens fallback；Anthropic 接入 cache_read_input_tokens

- 存储: messages 表新增 cached_tokens 列(幂等迁移)，贯穿 INSERT/SELECT/聚合

- 解耦加固: 抽取 MESSAGE_LOAD_COLUMNS 收敛 6 处消息加载 SELECT；抽取 USAGE_SUM_COLUMNS + read_usage_sum_row 收敛 topic/session 两份重复 SUM 聚合

- 协议: SessionTokenStats/TopicTokenStats 新增 cached_tokens 字段

- 前端: 面板新增缓存命中行 + 缓存命中率进度条(命中率越高越绿，与 ctx 占用方向相反)

src/agent/agent_loop.rs

@@ -3353,6 +3353,7 @@ mod tests {
                 prompt_tokens: 10,
                 completion_tokens: 10,
                 total_tokens: 20,
+                cached_tokens: 0,
             },
         }
     }

src/bus/message.rs

@@ -78,6 +78,10 @@ pub struct MessageUsage {
     pub prompt_tokens: u32,
     pub completion_tokens: u32,
     pub total_tokens: u32,
+    /// 输入中命中服务端缓存的 tokens 数（DeepSeek/Anthropic 等）。
+    /// 用于计算缓存命中率，反映成本节省效率。
+    #[serde(default)]
+    pub cached_tokens: u32,
     /// 本次调用所用模型的上下文窗口大小（来自 AgentRuntimeConfig）。
     /// 与 prompt_tokens 一起持久化，用于计算上下文占用率。
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -90,6 +94,7 @@ impl MessageUsage {
             prompt_tokens: u.prompt_tokens,
             completion_tokens: u.completion_tokens,
             total_tokens: u.total_tokens,
+            cached_tokens: u.cached_tokens,
             context_window_tokens: None,
         }
     }

src/command/handlers/list_topics.rs

@@ -19,6 +19,8 @@ pub struct TopicTokenStats {
     pub prompt_tokens: u64,
     pub completion_tokens: u64,
     pub total_tokens: u64,
+    /// 累计缓存命中的输入 tokens 数（老数据为 0）
+    pub cached_tokens: u64,
     pub last_prompt_tokens: Option<u32>,
     pub context_window_tokens: u32,
 }
@@ -76,6 +78,7 @@ pub fn build_topic_summaries(
                 prompt_tokens: s.prompt_tokens,
                 completion_tokens: s.completion_tokens,
                 total_tokens: s.total_tokens,
+                cached_tokens: s.cached_tokens,
                 last_prompt_tokens: s.last_prompt_tokens,
                 context_window_tokens: s.context_window_tokens.unwrap_or(0),
             });

src/command/handlers/load_task_messages.rs

@@ -110,6 +110,7 @@ async fn handle_load_task_messages(
             prompt_tokens: s.prompt_tokens,
             completion_tokens: s.completion_tokens,
             total_tokens: s.total_tokens,
+            cached_tokens: s.cached_tokens,
             last_prompt_tokens: s.last_prompt_tokens,
             context_window_tokens: s.context_window_tokens.unwrap_or(0),
         });

src/gateway/agent_factory.rs

@@ -15,6 +15,7 @@ use crate::observability::Observer;
 use crate::skills::{SkillPromptProvider, SkillRuntime};
 use crate::storage::PromptInjectionRepository;
 use crate::storage::persistent_session_id;
+use crate::storage::SessionStore;
 use crate::tools::task::runtime::{SubagentPromptProvider, SubagentRuntime};
 use crate::tools::task::SubagentResult;
 use crate::tools::{ToolContext, ToolRegistry, WaitCoordinator};
@@ -56,8 +57,12 @@ pub(crate) struct AgentFactory {
     prompt_repository: Arc<dyn PromptInjectionRepository>,
     /// Provider/Model 解析器：按专家 frontmatter 中的 provider/model 字段覆盖基础配置
     model_resolver: Arc<ModelResolver>,
-    /// per-session 的用户模型选择（最高优先级，覆盖专家配置）
+    /// per-session 的用户模型选择（覆盖专家配置）
     model_selections: Arc<ModelSelectionStore>,
+    /// per-topic 的用户模型选择（最高优先级；物化后话题模型不再随 session 选择漂移）
+    topic_model_selections: Arc<ModelSelectionStore>,
+    /// 持久化存储：session 级选择首次被话题命中时物化回写 topics 行
+    store: Arc<SessionStore>,
     /// 上下文压缩算法配置（所有 agent 共享）
     compaction_config: CompactionConfig,
     /// 可观测性 Observer（依赖注入到 AgentLoop，业务层不感知具体实现）
@@ -96,6 +101,8 @@ impl AgentFactory {
         prompt_repository: Arc<dyn PromptInjectionRepository>,
         model_resolver: Arc<ModelResolver>,
         model_selections: Arc<ModelSelectionStore>,
+        topic_model_selections: Arc<ModelSelectionStore>,
+        store: Arc<SessionStore>,
         compaction_config: CompactionConfig,
         observer: Option<Arc<dyn Observer>>,
     ) -> Self {
@@ -115,6 +122,8 @@ impl AgentFactory {
             prompt_repository,
             model_resolver,
             model_selections,
+            topic_model_selections,
+            store,
             compaction_config,
             observer,
             instance_id,
@@ -164,9 +173,20 @@ impl AgentFactory {
             _ => request.provider_config.clone(),
         };
 
-        // 按用户手动选择的 provider/model 覆盖（最高优先级，覆盖专家配置）。
+        // 用户手动选择的 provider/model 覆盖（最高优先级，覆盖专家配置）。
+        // 优先级：topic 级选择 > session 级选择；均未设置时保持专家/基础配置。
+        // 物化规则：session 级选择首次被话题命中时回写 topics 行固化——此后该话题的
+        // 模型只能被"在该话题内显式改选"改变，不再随 session 级选择漂移；
+        // 专家/config 默认不物化（保持继承活性）。
         // 引用不存在的 provider/model 名时报错并阻止会话（用户主动选择，配置错误应明确反馈）。
-        let effective_provider_config = match self.model_selections.get(&session_id) {
+        let topic_selection = request
+            .topic_id
+            .as_deref()
+            .and_then(|tid| self.topic_model_selections.get(tid));
+        let from_topic = topic_selection.is_some();
+        let user_selection = topic_selection.or_else(|| self.model_selections.get(&session_id));
+
+        let effective_provider_config = match user_selection {
             Some((user_provider, user_model))
                 if user_provider.is_some() || user_model.is_some() =>
             {
@@ -178,9 +198,32 @@ impl AgentFactory {
                         &expert_provider_config,
                     )
                     .map_err(|e| AgentError::Other(e.to_string()))?;
+
+                // 物化：命中 session 级选择且话题无固化值时，将解析后的具体
+                // (provider, model) 写入 topics 行（持久化 + 内存缓存）
+                if !from_topic {
+                    if let Some(tid) = request.topic_id.as_deref() {
+                        let provider = resolved.name.clone();
+                        let model = resolved.model_id.clone();
+                        self.topic_model_selections
+                            .set(tid, Some(provider.clone()), Some(model.clone()));
+                        if let Err(err) =
+                            self.store.update_topic_model(tid, Some(&provider), Some(&model))
+                        {
+                            tracing::warn!(
+                                error = %err,
+                                topic_id = %tid,
+                                "AgentFactory: failed to materialize topic model selection"
+                            );
+                        }
+                    }
+                }
+
                 tracing::info!(
                     instance_id = self.instance_id,
                     session_id = %session_id,
+                    topic_id = request.topic_id.as_deref().unwrap_or(""),
+                    source = if from_topic { "topic" } else { "session" },
                     provider = %resolved.name,
                     model_id = %resolved.model_id,
                     "AgentFactory: applied user model override"

src/gateway/execution.rs

@@ -15,6 +15,7 @@ use tokio::sync::Mutex;
 use super::compaction::schedule_background_history_compaction;
 use super::message_prepare::enrich_user_content_with_media_refs;
 use super::session::Session;
+use super::session_pool::is_scheduler_chat_id;
 use super::wait_coordinator::SessionWaitCoordinator;
 use crate::tools::WaitCoordinator;
 
@@ -73,6 +74,11 @@ impl CompactionSink for CompactionSinkImpl {
 
 const SCHEDULED_TASK_EXECUTION_SYSTEM_PROMPT: &str = "系统说明：当前输入来自一次已经触发的定时任务执行。你现在需要执行任务内容本身，而不是创建、修改、恢复、暂停或查询新的定时任务。除非当前任务内容明确要求管理调度器，否则不要调用任何定时任务管理工具；像“每小时”、“每天”、“cron”、“定时”等词，只应视为任务背景，不应再解释为新的建任务请求。";
 
+/// 静默（后台）定时任务的送达提示：最终响应发往虚拟会话，不会自动推送给用户，
+/// 必须显式调用 send_session_message 才能把结果送达用户会话。
+/// 仅在 scheduler/ 虚拟会话（silent_agent_task）中追加，避免普通 agent_task 重复发送。
+const SCHEDULED_TASK_SILENT_DELIVERY_HINT: &str = "特别注意：本次定时任务运行在后台独立会话中，你的最终响应不会自动发送给用户。如果任务需要向用户交付结果或发出通知，必须主动调用 send_session_message 工具把内容发送到当前会话，否则用户不会收到任何消息。";
+
 pub(crate) fn compose_scheduled_task_system_prompt(system_prompt: Option<&str>) -> String {
     match system_prompt
         .map(str::trim)
@@ -442,11 +448,16 @@ impl AgentExecutionService {
         // 获取该 topic 的串行锁（与普通消息路径共享，保证串行执行）
         // 定时任务由调度器触发，无用户消息竞态；在锁前一次性捕获 topic_id，
         // 锁后复用同一值作为 original_topic_id，保证锁键与写入目标一致。
+        //
+        // 关键：若该 chat 尚无 topic（scheduler/ 虚拟会话从不经用户消息分配
+        // topic），必须先补齐，否则后续消息无法进入按 topic 键化的内存历史，
+        // agent 将以空历史执行（任务 prompt 丢失）。
         let (serial_lock, session_store, lock_key, lock_time_topic_id) = {
             let mut session_guard = request.session.lock().await;
-            let tid = session_guard
+            let tid = match session_guard.current_topic(request.chat_id) {
-                .current_topic(request.chat_id)
+                Some(topic_id) => Some(topic_id.to_string()),
-                .map(|s| s.to_string());
+                None => Some(session_guard.ensure_topic_for_chat(request.chat_id)?),
+            };
             let lock_key = tid.as_deref().unwrap_or(request.chat_id).to_string();
             session_guard.ensure_sub_done_channel(&lock_key);
             (
@@ -491,8 +502,16 @@ impl AgentExecutionService {
             session_guard.ensure_chat_loaded(request.chat_id, original_topic_id.as_deref())?;
             session_guard.ensure_agent_prompt_before_user_message(request.chat_id)?;
 
-            let scheduled_system_prompt =
+            let mut scheduled_system_prompt =
                 compose_scheduled_task_system_prompt(request.system_prompt);
+            // 静默任务（scheduler/ 虚拟会话）的最终响应不会送达用户，
+            // 提示 agent 必须用 send_session_message 主动交付结果
+            if is_scheduler_chat_id(request.chat_id) {
+                scheduled_system_prompt = format!(
+                    "{}\n{}",
+                    scheduled_system_prompt, SCHEDULED_TASK_SILENT_DELIVERY_HINT
+                );
+            }
             session_guard.append_persisted_message(
                 request.chat_id,
                 original_topic_id.as_deref(),

src/gateway/http.rs

@@ -1217,3 +1217,154 @@ pub async fn session_selected_model(
         .unwrap_or((None, None));
     Json(SessionSelectedModelResponse { provider, model })
 }
+
+#[derive(Deserialize)]
+pub struct SelectTopicModelRequest {
+    pub session_id: String,
+    pub topic_id: String,
+    pub provider: Option<String>,
+    pub model: Option<String>,
+}
+
+/// POST /api/topic/select-model — 设置（或清除）话题级用户模型覆盖。
+///
+/// 双写语义（session store 的 key 一律取 topic 行自带的 session_id，不信任请求体）：
+/// - 设置：写 topics 行（话题记忆，物化）+ 内存缓存 + session store（成为新话题默认）
+/// - 清除（provider/model 均空）：移除话题级选择，并同步清除 session 级选择
+///   （否则 agent 会在下一条消息把 session 级重新物化回 topic 行，重置永远无效）
+pub async fn topic_select_model(
+    State(state): State<Arc<GatewayState>>,
+    Json(req): Json<SelectTopicModelRequest>,
+) -> (StatusCode, Json<SelectModelResponse>) {
+    // 规范化：trim 后空字符串视为 None（与 frontmatter 解析逻辑一致）
+    let provider = req
+        .provider
+        .map(|s| s.trim().to_string())
+        .filter(|s| !s.is_empty());
+    let model = req
+        .model
+        .map(|s| s.trim().to_string())
+        .filter(|s| !s.is_empty());
+
+    // 话题必须存在（防止对已删除话题静默写空）。
+    // 同时取 topic 行自带的 session_id 作为双写目标——不信任请求体中的 session_id，
+    // 防止客户端误传导致污染其他 session 的默认模型。
+    let store = state.session_manager.store();
+    let topic_session_id = match store.get_topic(&req.topic_id) {
+        Ok(Some(topic)) => topic.session_id,
+        Ok(None) => {
+            return (
+                StatusCode::NOT_FOUND,
+                Json(SelectModelResponse {
+                    success: false,
+                    error: Some(format!("topic '{}' not found", req.topic_id)),
+                }),
+            );
+        }
+        Err(e) => {
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(SelectModelResponse {
+                    success: false,
+                    error: Some(format!("failed to load topic: {}", e)),
+                }),
+            );
+        }
+    };
+
+    // 校验：provider/model 名必须在 config 的 providers/models 表中存在
+    let config = state.config.read().await;
+    if let Some(name) = provider.as_ref() {
+        if !config.providers.contains_key(name) {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(SelectModelResponse {
+                    success: false,
+                    error: Some(format!("provider '{}' not found in config", name)),
+                }),
+            );
+        }
+    }
+    if let Some(name) = model.as_ref() {
+        if !config.models.contains_key(name) {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(SelectModelResponse {
+                    success: false,
+                    error: Some(format!("model '{}' not found in config", name)),
+                }),
+            );
+        }
+    }
+    drop(config);
+
+    let is_clear = provider.is_none() && model.is_none();
+    if let Err(e) = store.update_topic_model(&req.topic_id, provider.as_deref(), model.as_deref()) {
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(SelectModelResponse {
+                success: false,
+                error: Some(format!("failed to persist topic model: {}", e)),
+            }),
+        );
+    }
+
+    state
+        .topic_model_selections
+        .set(&req.topic_id, provider.clone(), model.clone());
+
+    // 双写 session store（key 一律取 topic 行的 session_id）：
+    // - 显式设置：成为新话题的初始默认（"最近使用"）
+    // - 清除：同步移除 session 级选择。否则"重置为默认"会陷入死循环——
+    //   topic 行清空后 agent 命中 session 级又重新物化回 topic 行，重置形同虚设
+    if is_clear {
+        state.model_selections.set(&topic_session_id, None, None);
+    } else {
+        state.model_selections.set(&topic_session_id, provider, model);
+    }
+
+    (
+        StatusCode::OK,
+        Json(SelectModelResponse {
+            success: true,
+            error: None,
+        }),
+    )
+}
+
+/// GET /api/topic/selected-model?topic_id=... — 返回话题生效的用户模型选择。
+///
+/// 语义与 session 端点一致：只反映用户选择（topic 级优先，miss 回退 session 级），
+/// 不解析 expert/config 默认。
+#[derive(Deserialize)]
+pub struct TopicSelectedModelQuery {
+    pub topic_id: String,
+}
+
+pub async fn topic_selected_model(
+    State(state): State<Arc<GatewayState>>,
+    Query(q): Query<TopicSelectedModelQuery>,
+) -> Json<SessionSelectedModelResponse> {
+    // topic 级命中直接返回；miss 时按 topic 行的 session_id 回退 session 级
+    let (provider, model) = match state.topic_model_selections.get(&q.topic_id) {
+        Some(selection) => selection,
+        None => match state.session_manager.store().get_topic(&q.topic_id) {
+            Ok(Some(topic)) => {
+                // 双保险：SQLite 有物化值但缓存 miss（理论上不会发生）时回填缓存
+                if topic.provider.is_some() || topic.model.is_some() {
+                    state.topic_model_selections.set(
+                        &q.topic_id,
+                        topic.provider.clone(),
+                        topic.model.clone(),
+                    );
+                }
+                state
+                    .model_selections
+                    .get(&topic.session_id)
+                    .unwrap_or((None, None))
+            }
+            _ => (None, None),
+        },
+    };
+    Json(SessionSelectedModelResponse { provider, model })
+}

src/gateway/mod.rs

@@ -75,6 +75,8 @@ pub struct GatewayState {
     pub subagent_executor: Option<Arc<dyn crate::tools::SubAgentRuntime>>,
     /// per-session 的用户模型选择（覆盖专家配置）
     pub model_selections: Arc<model_selection::ModelSelectionStore>,
+    /// per-topic 的用户模型选择（最高优先级；物化保证话题模型不随 session 选择漂移）
+    pub topic_model_selections: Arc<model_selection::ModelSelectionStore>,
     /// Prometheus metrics handle（/metrics 端点渲染用）。
     /// None 表示 recorder 安装失败；热重启时从 OnceLock 缓存复用，不会因重复安装而变为 None。
     pub prometheus_handle: Option<metrics_exporter_prometheus::PrometheusHandle>,
@@ -108,7 +110,7 @@ impl GatewayState {
             mcp_servers: config.mcp_servers.clone(),
         };
 
-        let (session_manager, task_repository, mcp_manager, subagent_runtime, model_selections, subagent_executor) =
+        let (session_manager, task_repository, mcp_manager, subagent_runtime, model_selections, topic_model_selections, subagent_executor) =
             build_session_manager_with_sender(
                 agent_prompt_reinject_every,
                 show_tool_results,
@@ -155,6 +157,7 @@ impl GatewayState {
             subagent_runtime,
             subagent_executor,
             model_selections,
+            topic_model_selections,
             prometheus_handle,
         })
     }
@@ -369,6 +372,14 @@ pub async fn run(
             "/api/session/selected-model",
             routing::get(http::session_selected_model),
         )
+        .route(
+            "/api/topic/select-model",
+            routing::post(http::topic_select_model),
+        )
+        .route(
+            "/api/topic/selected-model",
+            routing::get(http::topic_selected_model),
+        )
         .route("/ws", routing::get(ws::ws_handler))
         .route("/metrics", routing::get(http::metrics_handler));
 

src/gateway/runtime.rs

@@ -67,6 +67,7 @@ pub(crate) fn build_session_manager(
         Option<Arc<McpClientManager>>,
         Arc<SubagentRuntime>,
         Arc<ModelSelectionStore>,
+        Arc<ModelSelectionStore>,
         Option<Arc<dyn SubAgentRuntime>>,
     ),
     AgentError,
@@ -120,6 +121,7 @@ pub(crate) fn build_session_manager_with_sender(
         Option<Arc<McpClientManager>>,
         Arc<SubagentRuntime>,
         Arc<ModelSelectionStore>,
+        Arc<ModelSelectionStore>,
         Option<Arc<dyn SubAgentRuntime>>,
     ),
     AgentError,
@@ -128,6 +130,22 @@ pub(crate) fn build_session_manager_with_sender(
         SessionStore::new()
             .map_err(|err| AgentError::Other(format!("session store init error: {}", err)))?,
     );
+    // 模型选择内存缓存：session 级 + topic 级（topic 级启动时从 topics 表预热）
+    let model_selections = Arc::new(ModelSelectionStore::new());
+    let topic_model_selections = Arc::new(ModelSelectionStore::new());
+    match store.list_topic_model_selections() {
+        Ok(entries) => {
+            for (topic_id, provider, model) in entries {
+                topic_model_selections.set(&topic_id, provider, model);
+            }
+        }
+        Err(err) => {
+            tracing::warn!(
+                error = %err,
+                "build_session_manager: failed to preload topic model selections"
+            );
+        }
+    }
     let known_agents = provider_configs.keys().cloned().collect::<HashSet<_>>();
     let provider_configs = ProviderConfigService::new(
         provider_config.clone(),
@@ -254,6 +272,8 @@ pub(crate) fn build_session_manager_with_sender(
             bus.clone(),
             store.clone(),
             skills.clone(),
+            Some(model_selections.clone()),
+            Some(topic_model_selections.clone()),
         ));
 
         // 注册 task 工具到子代理工具集（需在 runtime 创建之后，打破循环依赖）
@@ -320,7 +340,6 @@ pub(crate) fn build_session_manager_with_sender(
     );
 
     let prompt_repository: Arc<dyn PromptInjectionRepository> = store.clone();
-    let model_selections = Arc::new(ModelSelectionStore::new());
     let observer: Arc<dyn crate::observability::Observer> =
         crate::observability::metrics::default_observer();
     let agent_factory = AgentFactory::new(
@@ -332,6 +351,8 @@ pub(crate) fn build_session_manager_with_sender(
         prompt_repository.clone(),
         model_resolver.clone(),
         model_selections.clone(),
+        topic_model_selections.clone(),
+        store.clone(),
         compaction_config,
         Some(observer),
     );
@@ -376,6 +397,7 @@ pub(crate) fn build_session_manager_with_sender(
         mcp_manager,
         subagent_runtime,
         model_selections,
+        topic_model_selections,
         subagent_executor,
     ))
 }

src/gateway/session.rs

@@ -309,6 +309,8 @@ impl Session {
             prompt_repository.clone(),
             model_resolver,
             Arc::new(super::model_selection::ModelSelectionStore::new()),
+            Arc::new(super::model_selection::ModelSelectionStore::new()),
+            store.clone(),
             crate::config::CompactionConfig::default(),
             None,
         );
@@ -386,6 +388,54 @@ impl Session {
         self.history.chat_topic(chat_id)
     }
 
+    /// 确保指定 chat 存在当前话题，返回话题 ID。
+    ///
+    /// 内存中无当前话题时，先从数据库恢复最近活跃的话题；数据库中也没有则
+    /// 自动创建默认话题。定时任务路径（尤其 scheduler/ 虚拟会话）不经过用户
+    /// 消息的 topic 分配流程，若不补齐话题，消息将无法进入按 topic 键化的
+    /// 内存历史，导致 agent 以空历史执行。
+    pub fn ensure_topic_for_chat(&mut self, chat_id: &str) -> Result<String, AgentError> {
+        if let Some(topic_id) = self.history.chat_topic(chat_id) {
+            return Ok(topic_id.to_string());
+        }
+
+        // create_topic 依赖 sessions 行存在（外键约束），先确保持久会话已建立
+        self.ensure_persistent_session(chat_id)?;
+
+        let session_id = self.persistent_session_id(chat_id);
+        let topics = self
+            .store
+            .list_topics(&session_id)
+            .map_err(|e| AgentError::Other(format!("Failed to list topics: {}", e)))?;
+
+        if let Some(latest_topic) = topics.first() {
+            let topic_id = latest_topic.id.clone();
+            self.history.set_chat_topic(chat_id, topic_id.clone());
+            tracing::info!(
+                chat_id = %chat_id,
+                topic_id = %topic_id,
+                "Restored current topic from database"
+            );
+            return Ok(topic_id);
+        }
+
+        let title = format!("话题 {}", chrono::Local::now().format("%m/%d %H:%M"));
+        let topic = self
+            .store
+            .create_topic(&session_id, &title, None)
+            .map_err(|e| {
+                AgentError::Other(format!("Failed to auto-create default topic: {}", e))
+            })?;
+        self.history.set_chat_topic(chat_id, topic.id.clone());
+        tracing::info!(
+            chat_id = %chat_id,
+            topic_id = %topic.id,
+            session_id = %session_id,
+            "Auto-created default topic for chat"
+        );
+        Ok(topic.id)
+    }
+
     /// 切换话题 - 设置当前 topic 并加载新话题的历史到内存
     /// 不同 topic 的历史在 topic_histories 中独立存储，切换不互斥。
     pub fn switch_topic(&mut self, chat_id: &str, topic_id: &str) -> Result<(), AgentError> {
@@ -485,18 +535,25 @@ impl Session {
 
         // 只有当写入的 topic 匹配当前活跃 topic 时才更新内存历史。
         // 当用户已切换到新 topic 时，旧 topic 的排队消息不应污染新 topic 的内存历史。
+        // 完全无 topic 时回退到以 chat_id 为键的内存历史（与调用方
+        // history_key = topic.unwrap_or(chat_id) 的约定一致），避免无 topic
+        // 路径的消息只落库、不进内存，导致 agent 拿到空历史。
         let current_chat_topic = self.history.chat_topic(chat_id);
-        if topic_id.as_deref() == current_chat_topic {
+        match topic_id.as_deref() {
-            if let Some(ref tid) = topic_id {
+            Some(tid) if Some(tid) == current_chat_topic => {
                 self.add_message(tid, message);
             }
-        } else {
+            Some(_) => {
-            tracing::info!(
+                tracing::info!(
-                chat_id = %chat_id,
+                    chat_id = %chat_id,
-                write_topic_id = ?topic_id,
+                    write_topic_id = ?topic_id,
-                current_topic_id = ?current_chat_topic,
+                    current_topic_id = ?current_chat_topic,
-                "Skipping memory history update: message belongs to a different topic"
+                    "Skipping memory history update: message belongs to a different topic"
-            );
+                );
+            }
+            None => {
+                self.add_message(chat_id, message);
+            }
         }
 
         // 更新 topic 的最后活跃时间
@@ -994,7 +1051,7 @@ impl SessionManager {
             model_resolver,
             crate::config::CompactionConfig::default(),
         )
-        .map(|(session_manager, _, _, _, _, _)| session_manager)
+        .map(|(session_manager, _, _, _, _, _, _)| session_manager)
     }
 
     pub fn tools(&self) -> Arc<ToolRegistry> {
@@ -1055,6 +1112,11 @@ impl SessionManager {
         self.lifecycle.get(channel_name).await
     }
 
+    /// 获取定时任务专用 Session（不自动创建）
+    pub async fn get_scheduler_session(&self, channel_name: &str) -> Option<Arc<Mutex<Session>>> {
+        self.lifecycle.get_scheduler_session(channel_name).await
+    }
+
     /// 获取指定 chat 的当前话题（确保 session 存在，自动从数据库恢复）
     pub async fn get_current_topic(
         &self,
@@ -1430,6 +1492,55 @@ mod tests {
         format!("http://{}", address)
     }
 
+    /// 捕获所有 LLM 请求体的 mock server，用于断言"哪些消息真正进入了 LLM 输入"
+    async fn start_mock_openai_server_capturing() -> (String, StdArc<std::sync::Mutex<Vec<Value>>>)
+    {
+        let captured: StdArc<std::sync::Mutex<Vec<Value>>> =
+            StdArc::new(std::sync::Mutex::new(Vec::new()));
+        let state = captured.clone();
+
+        async fn handle(
+            axum::extract::State(state): axum::extract::State<StdArc<std::sync::Mutex<Vec<Value>>>>,
+            Json(body): Json<Value>,
+        ) -> Json<Value> {
+            if let Ok(mut guard) = state.lock() {
+                guard.push(body.clone());
+            }
+            let model = body
+                .get("model")
+                .and_then(|value| value.as_str())
+                .unwrap_or("unknown-model");
+
+            Json(json!({
+                "id": "mock-response",
+                "model": model,
+                "choices": [
+                    {
+                        "message": {
+                            "content": "任务已完成",
+                            "tool_calls": []
+                        }
+                    }
+                ],
+                "usage": {
+                    "prompt_tokens": 1,
+                    "completion_tokens": 1,
+                    "total_tokens": 2
+                }
+            }))
+        }
+
+        let app = Router::new()
+            .route("/chat/completions", post(handle))
+            .with_state(state);
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+        (format!("http://{}", address), captured)
+    }
+
     async fn start_mock_openai_504_server() -> String {
         async fn handle() -> (StatusCode, &'static str) {
             (StatusCode::GATEWAY_TIMEOUT, "stream timeout")
@@ -1711,6 +1822,281 @@ mod tests {
         assert!(scheduled_prompt.content.contains("你是邮箱待办同步助手。"));
     }
 
+    /// 回归：静默定时任务（scheduler/ 虚拟会话）必须——
+    /// 1. 自动创建 topic，任务 prompt 进入内存历史并送达 LLM（修复前为空历史执行）；
+    /// 2. 持久化的定时系统提示词包含 send_session_message 主动送达提示。
+    #[tokio::test]
+    async fn test_run_silent_agent_task_passes_prompt_and_creates_topic() {
+        let (base_url, captured) = start_mock_openai_server_capturing().await;
+        let provider_config = LLMProviderConfig {
+            provider_type: "openai".to_string(),
+            name: "default-provider".to_string(),
+            base_url,
+            api_key: "test-key".to_string(),
+            extra_headers: HashMap::new(),
+            model_id: "default-model".to_string(),
+            temperature: Some(0.0),
+            max_tokens: Some(32),
+            context_window_tokens: None,
+            model_extra: HashMap::new(),
+            max_tool_iterations: 1,
+            llm_timeout_secs: 30,
+            memory_maintenance_timeout_secs: 600,
+            max_retries: 3,
+            tool_result_max_chars: 100_000,
+            context_tool_result_trim_chars: 100_000,
+            max_images_in_context: 1,
+            max_image_age_rounds: 10,
+        };
+
+        let session_manager = SessionManager::new(
+            100,
+            false,
+            "Asia/Shanghai".to_string(),
+            provider_config.clone(),
+            HashMap::from([("default".to_string(), provider_config)]),
+            Arc::new(SkillRuntime::default()),
+            HashSet::new(),
+            crate::config::TaskConfig::default(),
+            crate::config::SubagentsConfig::default(),
+            test_maintenance_config(),
+            Some(24),
+            crate::mcp::McpConfig::default(),
+        )
+        .unwrap();
+
+        let outbound = session_manager
+            .run_silent_agent_task(
+                "test-channel",
+                "scheduler/silent-delivery-check",
+                Some("oc_notification_target"),
+                "检查网关连通性并用 send_session_message 通知用户",
+                // fresh_session=true 是 scheduler 生产路径的默认值：每次运行清空历史，
+                // 必须保证清空后任务 prompt 仍进入内存历史
+                ScheduledAgentTaskOptions {
+                    fresh_session: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+        assert_eq!(outbound.len(), 1);
+
+        // 第二次运行（fresh_session 再次清空历史）prompt 仍必须送达 LLM
+        session_manager
+            .run_silent_agent_task(
+                "test-channel",
+                "scheduler/silent-delivery-check",
+                Some("oc_notification_target"),
+                "第二轮执行：检查网关连通性",
+                ScheduledAgentTaskOptions {
+                    fresh_session: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+
+        // 1. 两次运行都应发起 LLM 请求，且各自的任务 prompt 出现在 user 消息中（非空历史执行）
+        let requests = captured.lock().unwrap().clone();
+        assert!(requests.len() >= 2, "两次定时任务应各发起至少一次 LLM 请求");
+        for expected in [
+            "检查网关连通性并用 send_session_message 通知用户",
+            "第二轮执行：检查网关连通性",
+        ] {
+            let prompt_delivered = requests.iter().any(|request| {
+                request
+                    .get("messages")
+                    .and_then(|value| value.as_array())
+                    .map(|messages| {
+                        messages.iter().any(|message| {
+                            message.get("role").and_then(|role| role.as_str()) == Some("user")
+                                && message
+                                    .get("content")
+                                    .and_then(|content| content.as_str())
+                                    .is_some_and(|content| content.contains(expected))
+                        })
+                    })
+                    .unwrap_or(false)
+            });
+            assert!(
+                prompt_delivered,
+                "定时任务 prompt “{}” 必须进入 LLM 输入，实际请求：{:?}",
+                expected, requests
+            );
+        }
+
+        // 2. scheduler/ 虚拟会话自动创建 topic，且消息挂在该 topic 下
+        let scheduler_session = session_manager
+            .get_scheduler_session("test-channel")
+            .await
+            .expect("scheduler session should exist");
+        let guard = scheduler_session.lock().await;
+        assert!(
+            guard
+                .current_topic("scheduler/silent-delivery-check")
+                .is_some(),
+            "定时任务应为虚拟会话设置当前 topic"
+        );
+        let session_id = guard.persistent_session_id("scheduler/silent-delivery-check");
+        let topics = guard.session_store().list_topics(&session_id).unwrap();
+        assert!(!topics.is_empty(), "scheduler 虚拟会话应自动创建 topic");
+
+        let messages = guard
+            .store()
+            .load_messages_for_topic(&topics[0].id, Some(&session_id))
+            .unwrap();
+        assert!(
+            messages
+                .iter()
+                .any(|message| message.role == "user" && message.content.contains("检查网关连通性")),
+            "任务 prompt 应持久化到 topic"
+        );
+
+        // 3. 静默任务的系统提示词包含主动送达提示
+        let scheduled_prompt = messages
+            .iter()
+            .find(|message| message.has_system_context(SYSTEM_CONTEXT_SCHEDULED_PROMPT))
+            .expect("missing scheduled system prompt");
+        assert!(
+            scheduled_prompt.content.contains("send_session_message"),
+            "静默任务系统提示词应包含 send_session_message 送达提示"
+        );
+    }
+
+    /// 回归：定时任务运行后 chat 自动拥有 topic；ensure_topic_for_chat 幂等，
+    /// 对全新 chat 会创建并持久化默认 topic。
+    #[tokio::test]
+    async fn test_ensure_topic_for_chat_creates_and_is_idempotent() {
+        let base_url = start_mock_openai_server().await;
+        let provider_config = LLMProviderConfig {
+            provider_type: "openai".to_string(),
+            name: "default-provider".to_string(),
+            base_url,
+            api_key: "test-key".to_string(),
+            extra_headers: HashMap::new(),
+            model_id: "default-model".to_string(),
+            temperature: Some(0.0),
+            max_tokens: Some(32),
+            context_window_tokens: None,
+            model_extra: HashMap::new(),
+            max_tool_iterations: 1,
+            llm_timeout_secs: 30,
+            memory_maintenance_timeout_secs: 600,
+            max_retries: 3,
+            tool_result_max_chars: 100_000,
+            context_tool_result_trim_chars: 100_000,
+            max_images_in_context: 1,
+            max_image_age_rounds: 10,
+        };
+
+        let session_manager = SessionManager::new(
+            100,
+            false,
+            "Asia/Shanghai".to_string(),
+            provider_config.clone(),
+            HashMap::from([("default".to_string(), provider_config)]),
+            Arc::new(SkillRuntime::default()),
+            HashSet::new(),
+            crate::config::TaskConfig::default(),
+            crate::config::SubagentsConfig::default(),
+            test_maintenance_config(),
+            Some(24),
+            crate::mcp::McpConfig::default(),
+        )
+        .unwrap();
+
+        session_manager
+            .run_scheduled_agent_task(
+                "test-channel",
+                "chat-topic-auto",
+                "执行任务A",
+                ScheduledAgentTaskOptions::default(),
+            )
+            .await
+            .unwrap();
+
+        let session = session_manager.get("test-channel").await.unwrap();
+        let mut guard = session.lock().await;
+
+        let tid1 = guard
+            .current_topic("chat-topic-auto")
+            .map(str::to_string)
+            .expect("定时任务运行后 chat 应自动拥有 topic");
+        let tid2 = guard.ensure_topic_for_chat("chat-topic-auto").unwrap();
+        assert_eq!(tid1, tid2, "ensure_topic_for_chat 应幂等返回当前 topic");
+
+        let tid3 = guard.ensure_topic_for_chat("chat-brand-new").unwrap();
+        let session_id = guard.persistent_session_id("chat-brand-new");
+        let topics = guard.session_store().list_topics(&session_id).unwrap();
+        assert!(
+            topics.iter().any(|topic| topic.id == tid3),
+            "对无历史的 chat 应创建并持久化默认 topic"
+        );
+        assert!(guard.current_topic("chat-brand-new").is_some());
+    }
+
+    /// 回归：无 topic 时 append_persisted_message 应回退到 chat_id 键的内存历史，
+    /// 避免消息只落库、不进内存，导致 agent 拿到空历史。
+    #[tokio::test]
+    async fn test_append_persisted_message_without_topic_updates_chat_keyed_history() {
+        let base_url = start_mock_openai_server().await;
+        let provider_config = LLMProviderConfig {
+            provider_type: "openai".to_string(),
+            name: "default-provider".to_string(),
+            base_url,
+            api_key: "test-key".to_string(),
+            extra_headers: HashMap::new(),
+            model_id: "default-model".to_string(),
+            temperature: Some(0.0),
+            max_tokens: Some(32),
+            context_window_tokens: None,
+            model_extra: HashMap::new(),
+            max_tool_iterations: 1,
+            llm_timeout_secs: 30,
+            memory_maintenance_timeout_secs: 600,
+            max_retries: 3,
+            tool_result_max_chars: 100_000,
+            context_tool_result_trim_chars: 100_000,
+            max_images_in_context: 1,
+            max_image_age_rounds: 10,
+        };
+
+        let session_manager = SessionManager::new(
+            100,
+            false,
+            "Asia/Shanghai".to_string(),
+            provider_config.clone(),
+            HashMap::from([("default".to_string(), provider_config)]),
+            Arc::new(SkillRuntime::default()),
+            HashSet::new(),
+            crate::config::TaskConfig::default(),
+            crate::config::SubagentsConfig::default(),
+            test_maintenance_config(),
+            Some(24),
+            crate::mcp::McpConfig::default(),
+        )
+        .unwrap();
+
+        session_manager
+            .ensure_session("test-channel")
+            .await
+            .unwrap();
+        let session = session_manager.get("test-channel").await.unwrap();
+        let mut guard = session.lock().await;
+        guard.ensure_persistent_session("chat-no-topic").unwrap();
+
+        guard
+            .append_persisted_message("chat-no-topic", None, ChatMessage::user("hello-no-topic"))
+            .unwrap();
+
+        let history = guard
+            .get_history("chat-no-topic")
+            .expect("无 topic 消息应进入以 chat_id 为键的内存历史");
+        assert_eq!(history.len(), 1);
+        assert_eq!(history[0].content, "hello-no-topic");
+    }
+
     /// 测试专用的 MemoryMaintenanceConfig，降低 min_memories_to_keep 以便于单条记忆测试
     fn test_maintenance_config() -> crate::config::MemoryMaintenanceConfig {
         crate::config::MemoryMaintenanceConfig {

src/gateway/session_lifecycle.rs

@@ -28,6 +28,14 @@ impl SessionLifecycleService {
         self.session_pool.get(channel_name).await
     }
 
+    /// 获取定时任务专用 Session（不自动创建）
+    pub(crate) async fn get_scheduler_session(
+        &self,
+        channel_name: &str,
+    ) -> Option<Arc<Mutex<Session>>> {
+        self.session_pool.get_scheduler_session(channel_name).await
+    }
+
     pub(crate) async fn touch(&self, channel_name: &str) {
         self.session_pool.touch(channel_name).await;
     }

src/providers/anthropic.rs

@@ -248,6 +248,9 @@ enum AnthropicContent {
 struct AnthropicUsage {
     input_tokens: u32,
     output_tokens: u32,
+    /// 从服务端缓存读取的输入 tokens 数（命中缓存部分）
+    #[serde(default)]
+    cache_read_input_tokens: Option<u32>,
 }
 
 #[async_trait]
@@ -388,6 +391,7 @@ impl LLMProvider for AnthropicProvider {
             prompt_tokens: anthropic_resp.usage.input_tokens,
             completion_tokens: anthropic_resp.usage.output_tokens,
             total_tokens: anthropic_resp.usage.input_tokens + anthropic_resp.usage.output_tokens,
+            cached_tokens: anthropic_resp.usage.cache_read_input_tokens.unwrap_or(0),
         };
 
         tracing::info!(

src/providers/openai.rs

@@ -134,11 +134,13 @@ impl StreamingAccumulator {
                     prompt_tokens: u.prompt_tokens,
                     completion_tokens: u.completion_tokens,
                     total_tokens: u.total_tokens,
+                    cached_tokens: u.cached_tokens(),
                 })
                 .unwrap_or(Usage {
                     prompt_tokens: 0,
                     completion_tokens: 0,
                     total_tokens: 0,
+                    cached_tokens: 0,
                 }),
         }
     }
@@ -726,6 +728,7 @@ impl OpenAIProvider {
                         prompt_tokens: openai_resp.usage.prompt_tokens,
                         completion_tokens: openai_resp.usage.completion_tokens,
                         total_tokens: openai_resp.usage.total_tokens,
+                        cached_tokens: openai_resp.usage.cached_tokens(),
                     };
                 }
             }
@@ -1077,6 +1080,32 @@ struct OpenAIUsage {
     completion_tokens: u32,
     #[serde(default)]
     total_tokens: u32,
+    /// DeepSeek 原生缓存字段：本次请求输入中命中缓存的 tokens 数
+    #[serde(default)]
+    prompt_cache_hit_tokens: Option<u32>,
+    /// OpenAI 兼容嵌套字段：prompt_tokens_details.cached_tokens
+    #[serde(default)]
+    prompt_tokens_details: Option<OpenAIPromptTokensDetails>,
+}
+
+#[derive(Deserialize, Default, Clone, Debug)]
+struct OpenAIPromptTokensDetails {
+    #[serde(default)]
+    cached_tokens: Option<u32>,
+}
+
+impl OpenAIUsage {
+    /// 缓存命中的输入 tokens 数。
+    /// 两种 API 形态互斥：优先 DeepSeek 顶层字段，回退 OpenAI 嵌套字段。
+    fn cached_tokens(&self) -> u32 {
+        self.prompt_cache_hit_tokens
+            .or_else(|| {
+                self.prompt_tokens_details
+                    .as_ref()
+                    .and_then(|d| d.cached_tokens)
+            })
+            .unwrap_or(0)
+    }
 }
 
 #[async_trait]
@@ -1238,6 +1267,7 @@ impl LLMProvider for OpenAIProvider {
                 prompt_tokens: openai_resp.usage.prompt_tokens,
                 completion_tokens: openai_resp.usage.completion_tokens,
                 total_tokens: openai_resp.usage.total_tokens,
+                cached_tokens: openai_resp.usage.cached_tokens(),
             },
         })
     }

src/providers/traits.rs

@@ -136,6 +136,10 @@ pub struct Usage {
     pub prompt_tokens: u32,
     pub completion_tokens: u32,
     pub total_tokens: u32,
+    /// 输入中命中服务端缓存的 tokens 数（DeepSeek prompt_cache_hit_tokens /
+    /// OpenAI prompt_tokens_details.cached_tokens）。不支持缓存的 provider 为 0。
+    #[serde(default)]
+    pub cached_tokens: u32,
 }
 
 /// 流式响应中的增量事件

src/scheduler/mod.rs

@@ -5,7 +5,7 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use chrono::{DateTime, Duration as ChronoDuration, TimeZone, Utc};
 use chrono_tz::Tz;
-use tokio::sync::watch;
+use tokio::sync::{Semaphore, watch};
 
 use crate::bus::{MessageBus, OutboundMessage};
 use crate::config::{
@@ -71,6 +71,9 @@ pub struct Scheduler {
     jobs: Arc<dyn SchedulerJobRepository>,
     agent_task_executor: Arc<dyn AgentTaskExecutor>,
     maintenance_executor: Arc<dyn MaintenanceExecutor>,
+    /// 并发执行槽位：限制同时执行的 job 数量（worker_queue_capacity）。
+    /// tick 循环只负责派发，job 执行在后台任务中进行，长任务不再阻塞其他 job。
+    worker_semaphore: Arc<Semaphore>,
 }
 
 impl Scheduler {
@@ -86,6 +89,7 @@ impl Scheduler {
         A: AgentTaskExecutor + 'static,
         M: MaintenanceExecutor + 'static,
     {
+        let worker_capacity = config.worker_queue_capacity.max(1);
         Self {
             bus,
             config,
@@ -93,6 +97,7 @@ impl Scheduler {
             jobs,
             agent_task_executor: Arc::new(agent_task_executor),
             maintenance_executor: Arc::new(maintenance_executor),
+            worker_semaphore: Arc::new(Semaphore::new(worker_capacity)),
         }
     }
 
@@ -224,7 +229,7 @@ impl Scheduler {
         let jobs = self.jobs.list_scheduler_jobs(true)?;
 
         for record in jobs {
-            let Some(mut job) =
+            let Some(job) =
                 RuntimeJob::from_record(&record, self.config.misfire_policy, self.timezone)?
             else {
                 continue;
@@ -248,6 +253,18 @@ impl Scheduler {
                 continue;
             }
 
+            // 尝试获取一个 worker 槽位：并发达到上限（worker_queue_capacity）时
+            // 不启动该 job——保持 Scheduled 状态，留待下一个 tick 重试。
+            // 这样长任务不会阻塞 tick 循环，也不会无界堆积并发执行。
+            let Ok(permit) = self.worker_semaphore.clone().try_acquire_owned() else {
+                tracing::warn!(
+                    job_id = %job.id,
+                    capacity = self.config.worker_queue_capacity,
+                    "Scheduler worker capacity exhausted, deferring job to next tick"
+                );
+                continue;
+            };
+
             self.jobs.update_scheduler_job_runtime(
                 &job.id,
                 SchedulerJobState::Running,
@@ -260,62 +277,136 @@ impl Scheduler {
                 job.completed_at,
             )?;
 
-            let execution_result = self.execute_job(&job).await;
+            // 执行与事后状态写入移入后台任务：tick 循环只做派发，
-            job.after_execution(
+            // 长耗时任务（agent_task 可能长达数分钟）不再串行阻塞其他 job 的触发。
-                now,
+            // job 在 DB 中已是 Running 状态，is_due 要求 Scheduled，因此不会被重复派发。
-                execution_result.as_ref().err().map(|err| err.to_string()),
+            let bus = self.bus.clone();
-                self.config.misfire_policy,
+            let jobs_repo = self.jobs.clone();
-                self.timezone,
+            let agent_executor = self.agent_task_executor.clone();
-            )?;
+            let maintenance_executor = self.maintenance_executor.clone();
+            let misfire_policy = self.config.misfire_policy;
+            let timezone = self.timezone;
+            let fire_at = now;
+            tokio::spawn(async move {
+                let execution_result = Scheduler::execute_job_inner(
+                    &bus,
+                    &*agent_executor,
+                    &*maintenance_executor,
+                    &job,
+                )
+                .await;
 
-            let status = if execution_result.is_ok() {
+                if let Err(error) = &execution_result {
-                Some(SchedulerJobStatus::Ok)
+                    tracing::error!(job_id = %job.id, error = %error, "Scheduler job failed");
-            } else {
+                }
-                Some(SchedulerJobStatus::Error)
-            };
 
-            if let Err(error) = &execution_result {
+                let status = if execution_result.is_ok() {
-                tracing::error!(job_id = %job.id, error = %error, "Scheduler job failed");
+                    Some(SchedulerJobStatus::Ok)
-            }
+                } else {
+                    Some(SchedulerJobStatus::Error)
+                };
 
-            self.jobs.update_scheduler_job_runtime(
+                let mut job = job;
-                &job.id,
+                match job.after_execution(
-                job.state.clone(),
+                    fire_at,
-                status,
+                    execution_result.as_ref().err().map(|err| err.to_string()),
-                job.last_error.as_deref(),
+                    misfire_policy,
-                job.run_count,
+                    timezone,
-                job.last_fired_at,
+                ) {
-                job.next_fire_at,
+                    Ok(()) => {
-                job.paused_at,
+                        if let Err(error) = jobs_repo.update_scheduler_job_runtime(
-                job.completed_at,
+                            &job.id,
-            )?;
+                            job.state.clone(),
+                            status,
+                            job.last_error.as_deref(),
+                            job.run_count,
+                            job.last_fired_at,
+                            job.next_fire_at,
+                            job.paused_at,
+                            job.completed_at,
+                        ) {
+                            tracing::error!(
+                                job_id = %job.id,
+                                error = %error,
+                                "Failed to persist scheduler job state after execution"
+                            );
+                        }
+                    }
+                    Err(error) => {
+                        // 兜底：状态推进失败时回退为 Scheduled，避免 job 永远卡在 Running
+                        tracing::error!(
+                            job_id = %job.id,
+                            error = %error,
+                            "Failed to compute post-execution scheduler state, resetting to Scheduled"
+                        );
+                        if let Err(update_error) = jobs_repo.update_scheduler_job_runtime(
+                            &job.id,
+                            SchedulerJobState::Scheduled,
+                            Some(SchedulerJobStatus::Error),
+                            Some(&error.to_string()),
+                            job.run_count,
+                            job.last_fired_at,
+                            job.next_fire_at,
+                            job.paused_at,
+                            job.completed_at,
+                        ) {
+                            tracing::error!(
+                                job_id = %job.id,
+                                error = %update_error,
+                                "Failed to persist scheduler job state after execution failure"
+                            );
+                        }
+                    }
+                }
+
+                // permit 持有到执行与状态写入全部完成后才释放
+                drop(permit);
+            });
         }
 
         Ok(())
     }
 
+    /// 执行单个 job（测试直接调用入口；生产派发走 process_tick 的后台任务）。
+    #[cfg(test)]
     async fn execute_job(&self, job: &RuntimeJob) -> anyhow::Result<()> {
+        Self::execute_job_inner(
+            &self.bus,
+            self.agent_task_executor.as_ref(),
+            self.maintenance_executor.as_ref(),
+            job,
+        )
+        .await
+    }
+
+    /// job 执行主体：不依赖 &self，便于移入 tokio::spawn 的后台任务。
+    async fn execute_job_inner(
+        bus: &Arc<MessageBus>,
+        agent_task_executor: &dyn AgentTaskExecutor,
+        maintenance_executor: &dyn MaintenanceExecutor,
+        job: &RuntimeJob,
+    ) -> anyhow::Result<()> {
         match job.kind {
             SchedulerJobKind::OutboundMessage => {
                 let message = build_outbound_message(job)?;
                 // publish_outbound 失败（bus 满或关闭）不视为 job 失败：
                 // 通知丢弃是预期的背压行为，标记 job 失败会触发 misfire 重试风暴
-                if let Err(e) = self.bus.publish_outbound(message).await {
+                if let Err(e) = bus.publish_outbound(message).await {
                     tracing::warn!(error = %e, job_id = %job.id, "Dropping outbound for scheduler job");
                 }
             }
             SchedulerJobKind::InternalEvent => {
-                execute_internal_event(self.maintenance_executor.as_ref(), job).await?;
+                execute_internal_event(maintenance_executor, job).await?;
             }
             SchedulerJobKind::AgentTask => {
                 let outbound_messages = execute_agent_task(
-                    self.agent_task_executor.as_ref(),
+                    agent_task_executor,
                     job,
                     required_notification_chat_id(job, "agent_task")?,
                 )
                 .await?;
                 for message in outbound_messages {
-                    if let Err(e) = self.bus.publish_outbound(message).await {
+                    if let Err(e) = bus.publish_outbound(message).await {
                         tracing::warn!(error = %e, job_id = %job.id, "Dropping outbound for scheduler agent task");
                     }
                 }
@@ -339,7 +430,7 @@ impl Scheduler {
                     Ok(p) => p,
                     Err(e) => {
                         if let Err(notify_error) =
-                            self.notify_silent_agent_task_failure(job, &e).await
+                            Self::notify_silent_agent_task_failure(bus, job, &e).await
                         {
                             tracing::error!(
                                 job_id = %job.id,
@@ -354,7 +445,7 @@ impl Scheduler {
                     Ok(o) => o,
                     Err(e) => {
                         if let Err(notify_error) =
-                            self.notify_silent_agent_task_failure(job, &e).await
+                            Self::notify_silent_agent_task_failure(bus, job, &e).await
                         {
                             tracing::error!(
                                 job_id = %job.id,
@@ -366,8 +457,7 @@ impl Scheduler {
                     }
                 };
 
-                if let Err(error) = self
+                if let Err(error) = agent_task_executor
-                    .agent_task_executor
                     .execute_silent(
                         job.target.channel.as_deref().unwrap_or_default(),
                         &session_chat_id,
@@ -378,7 +468,7 @@ impl Scheduler {
                     .await
                 {
                     if let Err(notify_error) =
-                        self.notify_silent_agent_task_failure(job, &error).await
+                        Self::notify_silent_agent_task_failure(bus, job, &error).await
                     {
                         tracing::error!(
                             job_id = %job.id,
@@ -395,7 +485,7 @@ impl Scheduler {
     }
 
     async fn notify_silent_agent_task_failure(
-        &self,
+        bus: &Arc<MessageBus>,
         job: &RuntimeJob,
         error: &anyhow::Error,
     ) -> anyhow::Result<()> {
@@ -413,8 +503,7 @@ impl Scheduler {
             "silent_agent_task".to_string(),
         );
 
-        if let Err(e) = self
+        if let Err(e) = bus
-            .bus
             .publish_outbound(OutboundMessage::error_notification(
                 channel,
                 chat_id,
@@ -1943,4 +2032,194 @@ mod tests {
         assert_eq!(convert_cron_weekday("*"), "*");
         assert_eq!(convert_cron_weekday("?"), "?");
     }
+
+    #[derive(Clone)]
+    struct SlowAgentTaskExecutor {
+        delay: std::time::Duration,
+    }
+
+    #[async_trait::async_trait]
+    impl AgentTaskExecutor for SlowAgentTaskExecutor {
+        async fn execute(
+            &self,
+            _channel_name: &str,
+            _chat_id: &str,
+            _prompt: &str,
+            _options: ScheduledAgentTaskOptions,
+        ) -> anyhow::Result<Vec<OutboundMessage>> {
+            tokio::time::sleep(self.delay).await;
+            Ok(Vec::new())
+        }
+
+        async fn execute_silent(
+            &self,
+            _channel_name: &str,
+            _session_chat_id: &str,
+            _notification_chat_id: Option<&str>,
+            _prompt: &str,
+            _options: ScheduledAgentTaskOptions,
+        ) -> anyhow::Result<Vec<OutboundMessage>> {
+            tokio::time::sleep(self.delay).await;
+            Ok(Vec::new())
+        }
+    }
+
+    /// 构造一个已到期（next_fire_at 在过去）的 agent_task
+    fn upsert_due_agent_task(store: &SessionStore, job_id: &str) {
+        store
+            .upsert_scheduler_job(&SchedulerJobUpsert {
+                id: job_id.to_string(),
+                kind: "agent_task".to_string(),
+                schedule: serde_json::json!({
+                    "type": "interval",
+                    "seconds": 3600,
+                    "startup_delay_secs": 0
+                }),
+                interval_secs: 3600,
+                startup_delay_secs: 0,
+                target: serde_json::json!({
+                    "channel": "test-channel",
+                    "chat_id": "oc_demo"
+                }),
+                payload: serde_json::json!({ "prompt": "测试任务" }),
+                enabled: true,
+                state: SchedulerJobState::Scheduled,
+                last_status: None,
+                last_error: None,
+                run_count: 0,
+                max_runs: None,
+                last_fired_at: None,
+                next_fire_at: Some(1),
+                paused_at: None,
+                completed_at: None,
+            })
+            .unwrap();
+    }
+
+    fn scheduler_config_with_capacity(capacity: usize) -> SchedulerConfig {
+        SchedulerConfig {
+            enabled: true,
+            tick_resolution_ms: 1000,
+            worker_queue_capacity: capacity,
+            misfire_policy: SchedulerMisfirePolicy::Skip,
+            jobs: Vec::new(),
+        }
+    }
+
+    async fn wait_for_run_count(store: &Arc<SessionStore>, job_id: &str, expected: i64) {
+        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+        loop {
+            let record = store.get_scheduler_job(job_id).unwrap().unwrap();
+            if record.run_count == expected {
+                return;
+            }
+            assert!(
+                std::time::Instant::now() < deadline,
+                "job {} did not reach run_count {} in time (current: {})",
+                job_id,
+                expected,
+                record.run_count
+            );
+            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+        }
+    }
+
+    /// 回归：process_tick 不得内联等待任务执行而阻塞 tick 循环；
+    /// 多个到期任务应并发执行，且各自正确推进到执行完成状态。
+    #[tokio::test]
+    async fn process_tick_dispatches_jobs_without_blocking() {
+        let store = Arc::new(SessionStore::in_memory().unwrap());
+        upsert_due_agent_task(&store, "job-a");
+        upsert_due_agent_task(&store, "job-b");
+
+        let (_, maintenance_service) = test_scheduler_services();
+        let scheduler = Scheduler::new(
+            MessageBus::new(8),
+            scheduler_config_with_capacity(64),
+            chrono_tz::Asia::Shanghai,
+            store.clone(),
+            SlowAgentTaskExecutor {
+                delay: std::time::Duration::from_millis(400),
+            },
+            maintenance_service,
+        );
+
+        let started = std::time::Instant::now();
+        scheduler.process_tick().await.unwrap();
+        let tick_elapsed = started.elapsed();
+        // tick 循环必须立即返回（远小于 400ms 的任务执行时长），
+        // 否则说明仍在串行等待任务执行
+        assert!(
+            tick_elapsed < std::time::Duration::from_millis(150),
+            "process_tick blocked on slow jobs: {:?}",
+            tick_elapsed
+        );
+
+        // 两个任务各 sleep 400ms：并发执行约 400ms 完成，串行需 >=800ms。
+        // 要求 700ms 内全部完成，证明并发执行。
+        wait_for_run_count(&store, "job-a", 1).await;
+        wait_for_run_count(&store, "job-b", 1).await;
+        assert!(
+            started.elapsed() < std::time::Duration::from_millis(700),
+            "jobs appear to run serially: {:?}",
+            started.elapsed()
+        );
+
+        // 执行完成：状态回到 Scheduled、status=ok、下次触发时间已推进
+        let record = store.get_scheduler_job("job-a").unwrap().unwrap();
+        assert_eq!(record.state, SchedulerJobState::Scheduled);
+        assert_eq!(record.last_status, Some(SchedulerJobStatus::Ok));
+        assert!(record.next_fire_at.unwrap() > 1);
+    }
+
+    /// 回归：并发槽位耗尽（worker_queue_capacity）时，多余的到期任务被推迟
+    /// （保持 Scheduled 不执行），槽位释放后的 tick 能正常派发。
+    #[tokio::test]
+    async fn process_tick_defers_jobs_when_worker_capacity_exhausted() {
+        let store = Arc::new(SessionStore::in_memory().unwrap());
+        upsert_due_agent_task(&store, "job-a");
+        upsert_due_agent_task(&store, "job-b");
+
+        let (_, maintenance_service) = test_scheduler_services();
+        let scheduler = Scheduler::new(
+            MessageBus::new(8),
+            scheduler_config_with_capacity(1),
+            chrono_tz::Asia::Shanghai,
+            store.clone(),
+            SlowAgentTaskExecutor {
+                delay: std::time::Duration::from_millis(500),
+            },
+            maintenance_service,
+        );
+
+        // 第一个 tick：只有一个槽位，一个任务开始执行，另一个必须被推迟
+        scheduler.process_tick().await.unwrap();
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        let a = store.get_scheduler_job("job-a").unwrap().unwrap();
+        let b = store.get_scheduler_job("job-b").unwrap().unwrap();
+        let (started, deferred) = if a.state == SchedulerJobState::Running {
+            (a, b)
+        } else {
+            assert_eq!(
+                b.state,
+                SchedulerJobState::Running,
+                "exactly one job should hold the only worker slot"
+            );
+            (b, a)
+        };
+        assert_eq!(started.run_count, 0, "running job has not finished yet");
+        assert_eq!(deferred.run_count, 0, "deferred job must not have executed");
+        assert_eq!(deferred.state, SchedulerJobState::Scheduled);
+
+        // 第二个 tick：槽位仍被占用，被推迟任务继续等待（不被派发也不报错）
+        scheduler.process_tick().await.unwrap();
+        let deferred_again = store.get_scheduler_job(&deferred.id).unwrap().unwrap();
+        assert_eq!(deferred_again.run_count, 0);
+        assert_eq!(deferred_again.state, SchedulerJobState::Scheduled);
+
+        // 第一个任务完成释放槽位后，下一个 tick 派发被推迟任务
+        wait_for_run_count(&store, &started.id, 1).await;
+        scheduler.process_tick().await.unwrap();
+        wait_for_run_count(&store, &deferred.id, 1).await;
+    }
 }

src/storage/migrations.rs

@@ -73,6 +73,13 @@ pub(super) fn ensure_messages_schema(conn: &Connection) -> Result<(), StorageErr
             "ALTER TABLE messages ADD COLUMN context_window_tokens INTEGER",
         )?;
     }
+    // 缓存命中的输入 tokens 数（老数据为 NULL，聚合时 COALESCE 为 0）
+    if !has_column(conn, "messages", "cached_tokens")? {
+        add_column_if_missing(
+            conn,
+            "ALTER TABLE messages ADD COLUMN cached_tokens INTEGER",
+        )?;
+    }
 
     // is_compacted: 1 表示该消息是被压缩消费掉的原始消息（前端可见、LLM 不可见）。
     // 压缩摘要消息 is_compacted=0（LLM 可见），通过 system_context='history_compaction*'
@@ -93,6 +100,18 @@ pub(super) fn ensure_messages_schema(conn: &Connection) -> Result<(), StorageErr
     Ok(())
 }
 
+/// topics 表：话题级模型选择列（用户在特定话题内显式选择的 provider/model）。
+/// NULL 表示该话题无显式选择（运行时按 session 级 → expert → config 链解析）。
+pub(super) fn ensure_topics_schema(conn: &Connection) -> Result<(), StorageError> {
+    if !has_column(conn, "topics", "provider")? {
+        add_column_if_missing(conn, "ALTER TABLE topics ADD COLUMN provider TEXT")?;
+    }
+    if !has_column(conn, "topics", "model")? {
+        add_column_if_missing(conn, "ALTER TABLE topics ADD COLUMN model TEXT")?;
+    }
+    Ok(())
+}
+
 pub(super) fn ensure_scheduler_schema(conn: &Connection) -> Result<(), StorageError> {
     if !has_column(conn, "scheduler_jobs", "schedule_json")? {
         conn.execute(

src/storage/mod.rs

@@ -111,6 +111,7 @@ impl SessionStore {
                 completion_tokens INTEGER,
                 total_tokens INTEGER,
                 context_window_tokens INTEGER,
+                cached_tokens INTEGER,
                 created_at INTEGER NOT NULL,
                 FOREIGN KEY(session_id) REFERENCES sessions(id) ON DELETE CASCADE,
                 FOREIGN KEY(topic_id) REFERENCES topics(id) ON DELETE SET NULL,
@@ -231,6 +232,7 @@ impl SessionStore {
 
         ensure_sessions_schema(&conn)?;
         ensure_messages_schema(&conn)?;
+        ensure_topics_schema(&conn)?;
         ensure_scheduler_schema(&conn)?;
         ensure_memory_scope_key_migration(&conn)?;
         ensure_todos_schema(&conn)?;
@@ -468,7 +470,7 @@ impl SessionStore {
     pub fn get_topic(&self, topic_id: &str) -> Result<Option<TopicRecord>, StorageError> {
         let conn = self.pool.get()?;
         let mut stmt = conn.prepare(
-            "SELECT id, session_id, title, description, created_at, updated_at, last_active_at, message_count FROM topics WHERE id = ?1",
+            "SELECT id, session_id, title, description, created_at, updated_at, last_active_at, message_count, provider, model FROM topics WHERE id = ?1",
         )?;
 
         stmt.query_row(params![topic_id], |row| {
@@ -481,6 +483,8 @@ impl SessionStore {
                 updated_at: row.get(5)?,
                 last_active_at: row.get(6)?,
                 message_count: row.get(7)?,
+                provider: row.get(8)?,
+                model: row.get(9)?,
             })
         })
         .optional()
@@ -490,7 +494,7 @@ impl SessionStore {
     pub fn list_topics(&self, session_id: &str) -> Result<Vec<TopicRecord>, StorageError> {
         let conn = self.pool.get()?;
         let mut stmt = conn.prepare(
-            "SELECT id, session_id, title, description, created_at, updated_at, last_active_at, message_count FROM topics WHERE session_id = ?1 ORDER BY last_active_at DESC"
+            "SELECT id, session_id, title, description, created_at, updated_at, last_active_at, message_count, provider, model FROM topics WHERE session_id = ?1 ORDER BY last_active_at DESC"
         )?;
 
         let rows = stmt.query_map(params![session_id], |row| {
@@ -503,6 +507,8 @@ impl SessionStore {
                 updated_at: row.get(5)?,
                 last_active_at: row.get(6)?,
                 message_count: row.get(7)?,
+                provider: row.get(8)?,
+                model: row.get(9)?,
             })
         })?;
 
@@ -544,6 +550,40 @@ impl SessionStore {
         Ok(())
     }
 
+    /// 设置/清除话题级模型选择。provider 与 model 均为 None 时清除（恢复继承）。
+    pub fn update_topic_model(
+        &self,
+        topic_id: &str,
+        provider: Option<&str>,
+        model: Option<&str>,
+    ) -> Result<(), StorageError> {
+        let now = current_timestamp();
+        let conn = self.pool.get()?;
+        conn.execute(
+            "UPDATE topics SET provider = ?2, model = ?3, updated_at = ?4 WHERE id = ?1",
+            params![topic_id, provider, model, now],
+        )?;
+        Ok(())
+    }
+
+    /// 全量读出话题级模型选择，供启动时预热内存缓存。
+    pub fn list_topic_model_selections(
+        &self,
+    ) -> Result<Vec<(String, Option<String>, Option<String>)>, StorageError> {
+        let conn = self.pool.get()?;
+        let mut stmt = conn.prepare(
+            "SELECT id, provider, model FROM topics WHERE provider IS NOT NULL OR model IS NOT NULL",
+        )?;
+        let rows = stmt.query_map([], |row| {
+            Ok((row.get(0)?, row.get(1)?, row.get(2)?))
+        })?;
+        let mut result = Vec::new();
+        for row in rows {
+            result.push(row?);
+        }
+        Ok(result)
+    }
+
     pub fn touch_topic(&self, topic_id: &str) -> Result<(), StorageError> {
         let now = current_timestamp();
         let conn = self.pool.get()?;
@@ -609,8 +649,8 @@ impl SessionStore {
             "
             INSERT INTO messages (
                 id, session_id, topic_id, seq, role, content,
-                system_context, reasoning_content, media_refs_json, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens, created_at
+                system_context, reasoning_content, media_refs_json, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens, cached_tokens, created_at
-            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)
+            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19)
             ",
             params![
                 message.id,
@@ -630,6 +670,7 @@ impl SessionStore {
                 message.usage.as_ref().map(|u| u.completion_tokens as i64),
                 message.usage.as_ref().map(|u| u.total_tokens as i64),
                 message.usage.as_ref().and_then(|u| u.context_window_tokens.map(|v| v as i64)),
+                message.usage.as_ref().map(|u| u.cached_tokens as i64),
                 message.timestamp,
             ],
         )?;
@@ -691,8 +732,8 @@ impl SessionStore {
                 INSERT INTO messages (
                     id, session_id, topic_id, seq, role, content,
                     system_context, reasoning_content, media_refs_json,
-                    tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens, created_at
+                    tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens, cached_tokens, created_at
-                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)
+                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19)
                 ",
                 params![
                     message.id,
@@ -712,6 +753,7 @@ impl SessionStore {
                     message.usage.as_ref().map(|u| u.completion_tokens as i64),
                     message.usage.as_ref().map(|u| u.total_tokens as i64),
                     message.usage.as_ref().and_then(|u| u.context_window_tokens.map(|v| v as i64)),
+                    message.usage.as_ref().map(|u| u.cached_tokens as i64),
                     message.timestamp,
                 ],
             )?;
@@ -1668,14 +1710,14 @@ impl SessionStore {
         let conn = self.pool.get()?;
 
         if let Some(sid) = session_id {
-            let mut stmt = conn.prepare(
+            let mut stmt = conn.prepare(&format!(
                 "
-                SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens
+                SELECT {MESSAGE_LOAD_COLUMNS}
                 FROM messages
                 WHERE topic_id = ?1 AND session_id = ?2 AND is_compacted = 0
                 ORDER BY seq ASC
                 ",
-            )?;
+            ))?;
             let rows = stmt.query_map(params![topic_id, sid], map_chat_message_row)?;
             let mut messages = Vec::new();
             for row in rows {
@@ -1683,14 +1725,14 @@ impl SessionStore {
             }
             Ok(messages)
         } else {
-            let mut stmt = conn.prepare(
+            let mut stmt = conn.prepare(&format!(
                 "
-                SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens
+                SELECT {MESSAGE_LOAD_COLUMNS}
                 FROM messages
                 WHERE topic_id = ?1 AND is_compacted = 0
                 ORDER BY seq ASC
                 ",
-            )?;
+            ))?;
             let rows = stmt.query_map(params![topic_id], map_chat_message_row)?;
             let mut messages = Vec::new();
             for row in rows {
@@ -1711,15 +1753,15 @@ impl SessionStore {
         let conn = self.pool.get()?;
 
         if let Some(sid) = session_id {
-            let mut stmt = conn.prepare(
+            let mut stmt = conn.prepare(&format!(
                 "
-                SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens
+                SELECT {MESSAGE_LOAD_COLUMNS}
                 FROM messages
                 WHERE topic_id = ?1 AND session_id = ?2
                   AND (system_context IS NULL OR system_context NOT LIKE 'history_compaction%')
                 ORDER BY seq ASC
                 ",
-            )?;
+            ))?;
             let rows = stmt.query_map(params![topic_id, sid], map_chat_message_row)?;
             let mut messages = Vec::new();
             for row in rows {
@@ -1727,15 +1769,15 @@ impl SessionStore {
             }
             Ok(messages)
         } else {
-            let mut stmt = conn.prepare(
+            let mut stmt = conn.prepare(&format!(
                 "
-                SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens
+                SELECT {MESSAGE_LOAD_COLUMNS}
                 FROM messages
                 WHERE topic_id = ?1
                   AND (system_context IS NULL OR system_context NOT LIKE 'history_compaction%')
                 ORDER BY seq ASC
                 ",
-            )?;
+            ))?;
             let rows = stmt.query_map(params![topic_id], map_chat_message_row)?;
             let mut messages = Vec::new();
             for row in rows {
@@ -1803,12 +1845,10 @@ impl SessionStore {
             .collect::<Vec<_>>()
             .join(", ");
         // topic_id IN (...) 自动排除 NULL topic_id 的旧消息；
-        // session_id NOT LIKE 'sub:%' 排除子代理消息（其 topic_id=父 topic_id）
+        // session_id NOT LIKE 'sub:%' 排除子代理消息（其 topic_id=父 topic_id）。
+        // SUM 列清单与行映射见 USAGE_SUM_COLUMNS / read_usage_sum_row（共享于子代理查询）。
         let sum_sql = format!(
-            "SELECT topic_id, \
+            "SELECT topic_id, {USAGE_SUM_COLUMNS} \
-                COALESCE(SUM(prompt_tokens), 0) AS sum_prompt, \
-                COALESCE(SUM(completion_tokens), 0) AS sum_completion, \
-                COALESCE(SUM(total_tokens), 0) AS sum_total \
              FROM messages \
              WHERE topic_id IN ({placeholders}) AND role = 'assistant' \
                 AND session_id NOT LIKE 'sub:%' \
@@ -1821,16 +1861,7 @@ impl SessionStore {
             .map(|s| s as &dyn rusqlite::ToSql)
             .collect();
         let sum_rows = stmt.query_map(params.as_slice(), |row| {
-            Ok((
+            Ok((row.get::<_, String>(0)?, read_usage_sum_row(row, 1)?))
-                row.get::<_, String>(0)?,
-                SessionTokenStats {
-                    prompt_tokens: row.get::<_, i64>(1)? as u64,
-                    completion_tokens: row.get::<_, i64>(2)? as u64,
-                    total_tokens: row.get::<_, i64>(3)? as u64,
-                    last_prompt_tokens: None,
-                    context_window_tokens: None,
-                },
-            ))
         })?;
 
         let mut stats: HashMap<String, SessionTokenStats> = HashMap::new();
@@ -1879,6 +1910,7 @@ impl SessionStore {
                 prompt_tokens: 0,
                 completion_tokens: 0,
                 total_tokens: 0,
+                cached_tokens: 0,
                 last_prompt_tokens: None,
                 context_window_tokens: None,
             });
@@ -1903,25 +1935,22 @@ impl SessionStore {
     ) -> Result<Option<SessionTokenStats>, StorageError> {
         let conn = self.pool.get()?;
 
-        // 1. SUM 查询：累计 prompt/completion/total
+        // 1. SUM 查询：累计 prompt/completion/total/cached
-        let sum_sql = "SELECT \
+        // 列清单与行映射复用 USAGE_SUM_COLUMNS / read_usage_sum_row（与 topic 聚合共享）
-                COALESCE(SUM(prompt_tokens), 0), \
+        let sum_sql = format!(
-                COALESCE(SUM(completion_tokens), 0), \
+            "SELECT {USAGE_SUM_COLUMNS} \
-                COALESCE(SUM(total_tokens), 0) \
              FROM messages \
-             WHERE session_id = ?1 AND role = 'assistant'";
+             WHERE session_id = ?1 AND role = 'assistant'"
-        let mut stmt = conn.prepare(sum_sql)?;
+        );
-        let sum_row = stmt.query_row(params![session_id], |row| {
+        let mut stmt = conn.prepare(&sum_sql)?;
-            Ok((
+        let sum_stats = stmt.query_row(params![session_id], |row| read_usage_sum_row(row, 0))?;
-                row.get::<_, i64>(0)? as u64,
-                row.get::<_, i64>(1)? as u64,
-                row.get::<_, i64>(2)? as u64,
-            ))
-        })?;
-        let (prompt_tokens, completion_tokens, total_tokens) = sum_row;
 
         // 无 assistant 消息时直接返回 None
-        if total_tokens == 0 && prompt_tokens == 0 && completion_tokens == 0 {
+        if sum_stats.total_tokens == 0
+            && sum_stats.prompt_tokens == 0
+            && sum_stats.completion_tokens == 0
+            && sum_stats.cached_tokens == 0
+        {
             // 需要二次确认是否真的没有 assistant 消息（usage 全 0 也可能是合法的）
             let count_sql =
                 "SELECT COUNT(*) FROM messages WHERE session_id = ?1 AND role = 'assistant'";
@@ -1949,9 +1978,10 @@ impl SessionStore {
         };
 
         Ok(Some(SessionTokenStats {
-            prompt_tokens,
+            prompt_tokens: sum_stats.prompt_tokens,
-            completion_tokens,
+            completion_tokens: sum_stats.completion_tokens,
-            total_tokens,
+            total_tokens: sum_stats.total_tokens,
+            cached_tokens: sum_stats.cached_tokens,
             last_prompt_tokens,
             context_window_tokens,
         }))
@@ -2326,14 +2356,14 @@ fn load_messages_between(
     start_seq_exclusive: i64,
     end_seq_inclusive: i64,
 ) -> Result<Vec<ChatMessage>, StorageError> {
-    let mut stmt = conn.prepare(
+    let mut stmt = conn.prepare(&format!(
         "
-        SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens
+        SELECT {MESSAGE_LOAD_COLUMNS}
         FROM messages
         WHERE session_id = ?1 AND seq > ?2 AND seq <= ?3
         ORDER BY seq ASC
         ",
-    )?;
+    ))?;
 
     let rows = stmt.query_map(
         params![session_id, start_seq_exclusive, end_seq_inclusive],
@@ -2374,7 +2404,7 @@ fn load_messages_between(
                 tool_state: None,
                 tool_duration_ms: row.get::<_, Option<i64>>(10)?.map(|v| v as u64),
                 tool_calls,
-                usage: map_usage_row(row, 11, 12, 13, 14)?,
+                usage: map_usage_row(row, 11, 12, 13, 14, 15)?,
             })
         },
     )?;
@@ -2391,14 +2421,14 @@ fn load_messages_after(
     session_id: &str,
     cutoff_seq: i64,
 ) -> Result<Vec<ChatMessage>, StorageError> {
-    let mut stmt = conn.prepare(
+    let mut stmt = conn.prepare(&format!(
         "
-        SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens
+        SELECT {MESSAGE_LOAD_COLUMNS}
         FROM messages
         WHERE session_id = ?1 AND seq > ?2
         ORDER BY seq ASC
         ",
-    )?;
+    ))?;
 
     let rows = stmt.query_map(params![session_id, cutoff_seq], |row| {
         let media_refs_json: String = row.get(5)?;
@@ -2436,7 +2466,7 @@ fn load_messages_after(
             tool_state: None,
             tool_duration_ms: row.get::<_, Option<i64>>(10)?.map(|v| v as u64),
             tool_calls,
-            usage: map_usage_row(row, 11, 12, 13, 14)?,
+            usage: map_usage_row(row, 11, 12, 13, 14, 15)?,
         })
     })?;
 

src/storage/records.rs

@@ -109,6 +109,9 @@ pub struct TopicRecord {
     pub updated_at: i64,
     pub last_active_at: i64,
     pub message_count: i64,
+    /// 话题级用户模型选择（NULL 表示无显式选择，运行时按继承链解析）
+    pub provider: Option<String>,
+    pub model: Option<String>,
 }
 
 /// pending_subagents 表的记录，跟踪异步子代理执行状态。
@@ -142,6 +145,8 @@ pub struct SessionTokenStats {
     pub prompt_tokens: u64,
     pub completion_tokens: u64,
     pub total_tokens: u64,
+    /// 累计缓存命中的输入 tokens 数（老数据为 0）
+    pub cached_tokens: u64,
     pub last_prompt_tokens: Option<u32>,
     pub context_window_tokens: Option<u32>,
 }

src/storage/row_mapping.rs

@@ -12,33 +12,69 @@ use crate::bus::message::MessageUsage;
 
 use super::{
     MemoryRecord, SchedulerJobRecord, SchedulerJobState, SchedulerJobStatus, SessionRecord,
-    SkillEventRecord, StorageError,
+    SessionTokenStats, SkillEventRecord, StorageError,
 };
 
-/// 从指定列索引读取 token usage 四元组（含 context_window_tokens）。
+/// 消息加载查询的共享列清单（列序与 map_chat_message_row / map_usage_row 的下标一一对应）。
+/// 新增 usage 列时只需改这里 + map_usage_row，无需逐条 SELECT 手工对齐。
+pub(super) const MESSAGE_LOAD_COLUMNS: &str = "id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens, cached_tokens";
+
+/// 从指定列索引读取 token usage 五元组（含 context_window_tokens、cached_tokens）。
 pub(super) fn map_usage_row(
     row: &rusqlite::Row<'_>,
     prompt_idx: usize,
     completion_idx: usize,
     total_idx: usize,
     context_window_idx: usize,
+    cached_idx: usize,
 ) -> rusqlite::Result<Option<MessageUsage>> {
     let prompt: Option<i64> = row.get(prompt_idx)?;
     let completion: Option<i64> = row.get(completion_idx)?;
     let total: Option<i64> = row.get(total_idx)?;
     let context_window: Option<i64> = row.get(context_window_idx)?;
-    if prompt.is_none() && completion.is_none() && total.is_none() && context_window.is_none() {
+    let cached: Option<i64> = row.get(cached_idx)?;
+    if prompt.is_none()
+        && completion.is_none()
+        && total.is_none()
+        && context_window.is_none()
+        && cached.is_none()
+    {
         Ok(None)
     } else {
         Ok(Some(MessageUsage {
             prompt_tokens: prompt.unwrap_or(0) as u32,
             completion_tokens: completion.unwrap_or(0) as u32,
             total_tokens: total.unwrap_or(0) as u32,
+            cached_tokens: cached.unwrap_or(0) as u32,
             context_window_tokens: context_window.map(|v| v as u32),
         }))
     }
 }
 
+/// token 用量聚合的共享 SUM 列清单：batch_topic_token_stats 与
+/// get_session_token_stats 共用。新增累计指标只需改这里 + read_usage_sum_row，
+/// 避免两份聚合 SQL 手工对齐（shotgun surgery）。
+pub(super) const USAGE_SUM_COLUMNS: &str = "COALESCE(SUM(prompt_tokens), 0), \
+     COALESCE(SUM(completion_tokens), 0), \
+     COALESCE(SUM(total_tokens), 0), \
+     COALESCE(SUM(cached_tokens), 0)";
+
+/// 从聚合行读取累计 usage 字段（SUM 列从 offset 开始）。
+/// last_* 瞬时字段不在 SUM 中，此处置 None，由调用方在 last 查询后回填。
+pub(super) fn read_usage_sum_row(
+    row: &rusqlite::Row<'_>,
+    offset: usize,
+) -> rusqlite::Result<SessionTokenStats> {
+    Ok(SessionTokenStats {
+        prompt_tokens: row.get::<_, i64>(offset)? as u64,
+        completion_tokens: row.get::<_, i64>(offset + 1)? as u64,
+        total_tokens: row.get::<_, i64>(offset + 2)? as u64,
+        cached_tokens: row.get::<_, i64>(offset + 3)? as u64,
+        last_prompt_tokens: None,
+        context_window_tokens: None,
+    })
+}
+
 pub(super) fn get_session_with_conn(
     conn: &Connection,
     session_id: &str,
@@ -172,7 +208,7 @@ pub(super) fn map_chat_message_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<
         tool_state: None,
         tool_duration_ms: row.get::<_, Option<i64>>(10)?.map(|v| v as u64),
         tool_calls,
-        usage: map_usage_row(row, 11, 12, 13, 14)?,
+        usage: map_usage_row(row, 11, 12, 13, 14, 15)?,
     })
 }
 

src/tools/scheduler_manage.rs

@@ -41,6 +41,9 @@ impl Tool for SchedulerManageTool {
         IMPORTANT - Default to Silent Mode: \
         When users request scheduled tasks without explicitly specifying the mode, default to silent_agent_task instead of agent_task. Silent mode is preferred for most automated tasks because it runs in a dedicated background session without cluttering the main conversation. Only use agent_task when the user explicitly wants interactive execution in the main chat. \
         \
+        IMPORTANT - Silent Mode Delivery: \
+        A silent_agent_task runs in a background session and its final response is NOT automatically sent to the user's conversation; only execution failures trigger a notification. Therefore, when the user expects to receive the task's result (a report, a status, an alert, etc.), the prompt MUST explicitly instruct the task to deliver its result by calling the send_session_message tool. Without this instruction the user will receive nothing on success. \
+        \
         IMPORTANT - Target Configuration: \
         For agent_task and silent_agent_task, the target.channel and target.chat_id determine where notifications are sent. \
         - If target is omitted or fields are empty, they are automatically filled from the current conversation context. \

src/tools/task/runtime.rs

@@ -462,6 +462,10 @@ pub struct DefaultSubAgentRuntime {
     /// task_id → CancellationToken 映射，用于取消传播
     /// Arc 包装以便 spawned task 完成后清理自身条目
     cancel_registry: Arc<parking_lot::Mutex<HashMap<String, tokio_util::sync::CancellationToken>>>,
+    /// per-session 的用户模型选择（子代理 def 未显式设定模型时继承）
+    model_selections: Option<Arc<crate::gateway::model_selection::ModelSelectionStore>>,
+    /// per-topic 的用户模型选择（优先于 session 级继承）
+    topic_model_selections: Option<Arc<crate::gateway::model_selection::ModelSelectionStore>>,
 }
 
 impl DefaultSubAgentRuntime {
@@ -476,6 +480,8 @@ impl DefaultSubAgentRuntime {
         bus: Option<Arc<MessageBus>>,
         store: Arc<SessionStore>,
         skills: Arc<SkillRuntime>,
+        model_selections: Option<Arc<crate::gateway::model_selection::ModelSelectionStore>>,
+        topic_model_selections: Option<Arc<crate::gateway::model_selection::ModelSelectionStore>>,
     ) -> Self {
         let max_concurrent = config.max_concurrent.max(1);
         let semaphore = Arc::new(tokio::sync::Semaphore::new(max_concurrent));
@@ -492,6 +498,8 @@ impl DefaultSubAgentRuntime {
             skills,
             semaphore,
             cancel_registry: Arc::new(parking_lot::Mutex::new(HashMap::new())),
+            model_selections,
+            topic_model_selections,
         }
     }
 
@@ -502,6 +510,55 @@ impl DefaultSubAgentRuntime {
             .ok_or_else(|| format!("subagent type '{}' is disabled or not found", type_name))
     }
 
+    /// 解析子代理最终生效的 provider 配置（create_subagent 与 spawn 共用，保证一致）。
+    ///
+    /// 优先级：def frontmatter > 话题级用户选择 > session 级用户选择 > 全局基础配置。
+    /// def 仅设定部分字段时，缺失字段从继承链补齐（而非全局），
+    /// 与主代理"专家覆盖 → 用户选择再覆盖"的解析语义对称。
+    fn resolve_effective_provider_config(
+        &self,
+        session: &TaskSession,
+        def_name: Option<&str>,
+        def_provider: Option<&str>,
+        def_model: Option<&str>,
+    ) -> Result<LLMProviderConfig, TaskError> {
+        let resolve_err = |scope: &str, e: crate::config::ConfigError| {
+            TaskError::AgentCreationFailed(format!(
+                "subagent '{}' {} model resolution failed: {}",
+                def_name.unwrap_or("?"),
+                scope,
+                e
+            ))
+        };
+
+        // 继承链：topic 级 > session 级（key 为主代理的 persistent session id）
+        let inherited = session
+            .parent_topic_id
+            .as_deref()
+            .and_then(|tid| self.topic_model_selections.as_ref().and_then(|s| s.get(tid)))
+            .or_else(|| {
+                self.model_selections
+                    .as_ref()
+                    .and_then(|s| s.get(&session.parent_session_id))
+            });
+
+        let base = match inherited {
+            Some((p, m)) if p.is_some() || m.is_some() => self
+                .model_resolver
+                .resolve(p.as_deref(), m.as_deref(), &self.provider_config)
+                .map_err(|e| resolve_err("inherited", e))?,
+            _ => self.provider_config.clone(),
+        };
+
+        match (def_provider.is_some(), def_model.is_some()) {
+            (true, _) | (_, true) => self
+                .model_resolver
+                .resolve(def_provider, def_model, &base)
+                .map_err(|e| resolve_err("", e)),
+            _ => Ok(base),
+        }
+    }
+
     /// 获取实际执行时间
     fn effective_max_execution_secs(&self, def: &SubagentDef) -> u64 {
         def.max_execution_secs
@@ -582,25 +639,14 @@ impl DefaultSubAgentRuntime {
         let child_depth = parent_nesting_depth + 1;
         let tools = self.build_subagent_tools_registry(def, child_depth);
 
-        // 按 def 中的 provider/model 字段解析覆盖基础 provider_config。
+        // 解析优先级：def frontmatter > topic 级用户选择 > session 级用户选择 > 全局基础配置。
         // 引用不存在的 provider/model 名时返回错误（反馈给 LLM 重试，与 def 缺失即拒绝的安全范式一致）。
-        let effective_provider_config = match def {
+        let effective_provider_config = self.resolve_effective_provider_config(
-            Some(d) if d.provider.is_some() || d.model.is_some() => self
+            session,
-                .model_resolver
+            def.map(|d| d.name.as_str()),
-                .resolve(
+            def.and_then(|d| d.provider.as_deref()),
-                    d.provider.as_deref(),
+            def.and_then(|d| d.model.as_deref()),
-                    d.model.as_deref(),
+        )?;
-                    &self.provider_config,
-                )
-                .map_err(|e| {
-                    TaskError::AgentCreationFailed(format!(
-                        "subagent '{}' model resolution failed: {}",
-                        def.map(|d| d.name.as_str()).unwrap_or("?"),
-                        e
-                    ))
-                })?,
-            _ => self.provider_config.clone(),
-        };
 
         AgentLoop::with_tools_and_system_prompt_provider(
             AgentRuntimeConfig::from(effective_provider_config),
@@ -714,6 +760,8 @@ impl DefaultSubAgentRuntime {
                     summary: extract_summary(&final_message.content),
                     output: final_message.content,
                     task_id: session.id.clone(),
+                    provider: None,
+                    model: None,
                 })
             }
             Ok(Err(e)) => Err(TaskError::ExecutionFailed(e.to_string())),
@@ -759,6 +807,8 @@ impl DefaultSubAgentRuntime {
                     summary: extract_summary(&final_message.content),
                     output: final_message.content,
                     task_id: session.id.clone(),
+                    provider: None,
+                    model: None,
                 })
             }
             Ok(Err(e)) => Err(TaskError::ExecutionFailed(e.to_string())),
@@ -797,6 +847,8 @@ impl DefaultSubAgentRuntime {
             summary: error.to_string(),
             output: String::new(),
             task_id: session.id.clone(),
+            provider: None,
+            model: None,
         })
     }
 }
@@ -926,23 +978,14 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
         } else {
             self.skills.system_index_prompt()
         };
-        // 同步解析 def 中的 provider/model 覆盖，保证环境提示中的模型名与实际使用的模型一致
+        // 同步解析生效配置（def > topic 级 > session 级 > 全局），
-        let effective_provider_config = match (def.provider.is_some(), def.model.is_some()) {
+        // 与 create_subagent 共用 helper，保证环境提示中的模型名与实际使用的模型一致
-            (true, _) | (_, true) => self
+        let effective_provider_config = self.resolve_effective_provider_config(
-                .model_resolver
+            &session,
-                .resolve(
+            Some(def.name.as_str()),
-                    def.provider.as_deref(),
+            def.provider.as_deref(),
-                    def.model.as_deref(),
+            def.model.as_deref(),
-                    &self.provider_config,
+        )?;
-                )
-                .map_err(|e| {
-                    TaskError::AgentCreationFailed(format!(
-                        "subagent '{}' model resolution failed: {}",
-                        def.name, e
-                    ))
-                })?,
-            _ => self.provider_config.clone(),
-        };
         let system_prompt = SubagentPromptBuilder::build(
             &def,
             &task.description,
@@ -1177,6 +1220,7 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
 
             // 8c. 立即返回 running 占位结果
             // 注意: summary 留空，output 只含引导信息。LLM 看到 running 后应调 wait_for_subagents。
+            // provider/model 携带实际生效配置，供前端在偏离主代理模型时差异显示。
             return Ok(TaskToolResult {
                 status: "running".to_string(),
                 summary: format!("Task {} spawned asynchronously", task_id),
@@ -1185,6 +1229,8 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
                     task_id
                 ),
                 task_id,
+                provider: Some(effective_provider_config.name.clone()),
+                model: Some(effective_provider_config.model_id.clone()),
             });
         }
 

src/tools/task/types.rs

@@ -301,6 +301,12 @@ pub struct TaskToolResult {
     pub output: String,
     /// 会话 ID（用于恢复）
     pub task_id: String,
+    /// 子代理实际使用的 provider 名（running 占位时携带，供前端差异显示；其余场景可省略）
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub provider: Option<String>,
+    /// 子代理实际使用的 model id（同上）
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub model: Option<String>,
 }
 
 /// 异步子代理完成状态

web/src/api/client.ts

@@ -22,6 +22,8 @@ export const API = {
   expertsSelect: '/api/experts/select',
   sessionSelectModel: '/api/session/select-model',
   sessionSelectedModel: '/api/session/selected-model',
+  topicSelectModel: '/api/topic/select-model',
+  topicSelectedModel: '/api/topic/selected-model',
 } as const;
 
 const TOKEN_KEY = 'picobot-gateway-token';

web/src/api/experts.ts

@@ -106,3 +106,35 @@ export async function getSelectedModel(
   if (!resp.ok) return { provider: null, model: null };
   return resp.json();
 }
+
+/**
+ * 设置（或清除）话题级模型选择。后端双写：topics 行 + session store（新话题默认值）。
+ * provider/model 均为 null 时清除话题级选择（恢复继承 session 级）。
+ */
+export async function selectTopicModel(
+  sessionId: string,
+  topicId: string,
+  provider: string | null,
+  model: string | null,
+): Promise<{ success: boolean; error?: string }> {
+  const resp = await authedFetch(API.topicSelectModel, {
+    method: 'POST',
+    body: { session_id: sessionId, topic_id: topicId, provider, model },
+  });
+  const data = await resp.json().catch(() => ({}));
+  if (!resp.ok || !data.success) return { success: false, error: data.error || '切换模型失败' };
+  return { success: true };
+}
+
+/**
+ * 读取话题生效的用户模型选择（topic 级优先，miss 回退 session 级）。
+ * 与 session 端点同语义：只反映用户选择，不解析 expert/config 默认。
+ */
+export async function getSelectedTopicModel(
+  topicId: string,
+): Promise<{ provider: string | null; model: string | null }> {
+  const params = new URLSearchParams({ topic_id: topicId });
+  const resp = await authedFetch(`${API.topicSelectedModel}?${params}`);
+  if (!resp.ok) return { provider: null, model: null };
+  return resp.json();
+}

web/src/components/Chat/ChatContainer.tsx

@@ -49,6 +49,11 @@ export function ChatContainer({
     name: string;
     description: string;
   } | null>(null);
+  // 当前生效模型（供 Task 卡片差异显示：子代理模型 ≠ 主代理模型时提示）
+  const [effectiveModel, setEffectiveModel] = useState<{
+    provider: string;
+    model: string;
+  } | null>(null);
 
   const selectors = (
     <div className="flex flex-wrap items-center gap-1 px-3 pt-2">
@@ -58,7 +63,14 @@ export function ChatContainer({
         onSelectionChange={setSelectedExpert}
         settingsClosedTick={settingsClosedTick}
       />
-      <ModelSelector sessionId={sessionId ?? null} settingsClosedTick={settingsClosedTick} />
+      <ModelSelector
+        sessionId={sessionId ?? null}
+        topicId={topicId ?? null}
+        settingsClosedTick={settingsClosedTick}
+        onSelectionChange={(effective) =>
+          setEffectiveModel({ provider: effective.provider, model: effective.model })
+        }
+      />
     </div>
   );
 
@@ -107,6 +119,7 @@ export function ChatContainer({
             showThinking={showThinking}
             viewKey={viewKey}
             highlightedMessageId={highlightedMessageId}
+            effectiveModel={effectiveModel}
           />
         </div>
       )}

web/src/components/Chat/ExpertSelector.tsx

@@ -36,6 +36,8 @@ export function ExpertSelector({
   const [loading, setLoading] = useState(false);
   const [listLoading, setListLoading] = useState(false);
   const [error, setError] = useState<string | null>(null);
+  // 专家系统开关（设置页可关闭）；默认 true 避免加载期间闪烁隐藏
+  const [systemEnabled, setSystemEnabled] = useState(true);
 
   const containerRef = useRef<HTMLDivElement>(null);
 
@@ -72,10 +74,27 @@ export function ExpertSelector({
     refreshSelection();
   }, [refreshSelection]);
 
-  // 设置弹窗关闭时刷新选中状态（处理已选专家被禁用/删除的情况）
+  // 检查专家系统是否启用（设置页 experts.enabled 开关）；
+  // 关闭时隐藏输入框上方的专家选择器，并清除已选专家
+  const checkSystemEnabled = useCallback(async () => {
+    const data = await listExperts();
+    if (!data) return; // 网络失败保持现状
+    setSystemEnabled(data.experts_system_enabled);
+    if (!data.experts_system_enabled) {
+      setSelectedExpert(null);
+      onSelectionChange?.(null);
+    }
+  }, [onSelectionChange]);
+
+  useEffect(() => {
+    checkSystemEnabled();
+  }, [checkSystemEnabled]);
+
+  // 设置弹窗关闭时刷新选中状态（处理已选专家被禁用/删除/系统开关切换的情况）
   useEffect(() => {
     if (settingsClosedTick === undefined) return;
     refreshSelection();
+    checkSystemEnabled();
     // eslint-disable-next-line react-hooks/exhaustive-deps
   }, [settingsClosedTick]);
 
@@ -143,8 +162,8 @@ export function ExpertSelector({
     onManageExperts?.();
   };
 
-  // If sessionId is null, render nothing
+  // If sessionId is null or expert system disabled in settings, render nothing
-  if (!sessionId) return null;
+  if (!sessionId || !systemEnabled) return null;
 
   return (
     <div ref={containerRef} className="relative shrink-0 flex items-center gap-2">

web/src/components/Chat/MessageBubble.tsx

@@ -89,6 +89,8 @@ interface MessageBubbleProps {
   message: ChatMessage;
   onNavigateToSubAgent?: (taskId: string, description: string, subagentType?: string) => void;
   showThinking?: boolean;
+  /** 主代理当前生效模型（Task 卡片差异显示：子代理模型不同才展示模型徽章） */
+  effectiveModel?: { provider: string; model: string } | null;
 }
 
 function getAttachmentIcon(mediaType: string) {
@@ -352,6 +354,7 @@ export const MessageBubble = memo(function MessageBubble({
   message,
   onNavigateToSubAgent,
   showThinking = true,
+  effectiveModel,
 }: MessageBubbleProps) {
   const isUser = message.role === 'user';
   const isTool = message.role === 'tool';
@@ -484,6 +487,19 @@ export const MessageBubble = memo(function MessageBubble({
     // 安全获取 task 状态配色，未知状态回退到默认（避免 undefined.borderColor 崩溃）
     const taskStyle = taskResult ? (taskStatusConfig[taskResult.status] ?? taskStatusConfig.failed) : null;
 
+    // 子代理模型徽章：结果携带模型信息，且（无主代理参照 或 model/provider 任一不同）时展示。
+    // provider 不同即使 model 同名也显示——不同 provider 的同名模型在路由/计费上已不同。
+    const modelDiffers =
+      !effectiveModel ||
+      effectiveModel.model !== taskResult?.model ||
+      effectiveModel.provider !== taskResult?.provider;
+    const subagentModelLabel =
+      taskResult?.model && modelDiffers
+        ? effectiveModel && effectiveModel.provider === taskResult.provider
+          ? taskResult.model
+          : `${taskResult.provider ?? '?'}/${taskResult.model}`
+        : null;
+
     return (
       <div data-message-id={message.id} className="animate-slide-in">
         <div
@@ -504,6 +520,14 @@ export const MessageBubble = memo(function MessageBubble({
                   ? `${message.toolName || 'Tool'}${taskDescription ? ` · ${taskDescription}` : ''}`
                   : message.toolName || 'Tool'}
               </span>
+              {subagentModelLabel && (
+                <span
+                  className="flex-shrink-0 rounded-full border border-[var(--border-color)] bg-[var(--bg-tertiary)] px-1.5 py-px text-[10px] leading-4 text-[var(--text-muted)] max-w-[160px] truncate"
+                  title={`子代理模型：${taskResult?.provider ?? '?'} / ${taskResult?.model}`}
+                >
+                  {subagentModelLabel}
+                </span>
+              )}
               <span
                 className={`flex-shrink-0 transition-all duration-300 ${
                   taskStyle ? taskStyle.iconColor : statusConfig.iconColor

web/src/components/Chat/MessageInput.tsx

@@ -15,6 +15,40 @@ import type { Attachment } from '../../types/protocol';
 
 const MAX_FILE_SIZE = 50 * 1024 * 1024; // 50MB
 
+/** per-topic 文本草稿的 localStorage key（附件 File 对象不可序列化，仅存内存） */
+export function draftStorageKey(topicId: string): string {
+  return `picobot:draft:${topicId}`;
+}
+
+function loadDraftText(topicId: string): string {
+  try {
+    return localStorage.getItem(draftStorageKey(topicId)) ?? '';
+  } catch {
+    return '';
+  }
+}
+
+function saveDraftText(topicId: string, text: string): void {
+  try {
+    if (text.trim()) {
+      localStorage.setItem(draftStorageKey(topicId), text);
+    } else {
+      localStorage.removeItem(draftStorageKey(topicId));
+    }
+  } catch {
+    // 隐私模式/配额满：静默降级为仅内存草稿
+  }
+}
+
+/** 删除话题时调用：清除其持久化草稿 */
+export function clearTopicDraft(topicId: string): void {
+  try {
+    localStorage.removeItem(draftStorageKey(topicId));
+  } catch {
+    // ignore
+  }
+}
+
 interface MessageInputProps {
   onSend: (content: string, attachments: Attachment[]) => void;
   onStop?: () => void;
@@ -24,7 +58,7 @@ interface MessageInputProps {
   isReadOnly?: boolean;
   channelName?: string;
   selectedExpert?: { name: string; description: string } | null;
-  /** 当前话题 ID，切换话题时自动清空草稿 */
+  /** 当前话题 ID：切换话题时保留各自草稿（文本持久化到 localStorage，附件仅内存） */
   topicId?: string | null;
 }
 
@@ -65,16 +99,77 @@ export function MessageInput({
   const fileInputRef = useRef<HTMLInputElement>(null);
   const wasLoadingRef = useRef(false);
   const prevTopicIdRef = useRef<string | null | undefined>(topicId);
+  // per-topic 草稿缓存：切换话题时暂存/恢复（附件 File 对象仅内存，刷新丢失）
+  const draftsRef = useRef<Map<string, { content: string; attachments: FileAttachment[] }>>(
+    new Map(),
+  );
+  const draftDebounceRef = useRef<number | null>(null);
 
-  // 切换话题时清空草稿（替代原来通过 key remount 的重置机制）
+  // 切换话题时：暂存当前草稿 → 恢复目标话题草稿（内存 miss 时读 localStorage）
   useEffect(() => {
-    if (prevTopicIdRef.current !== topicId) {
+    if (prevTopicIdRef.current === topicId) return;
-      prevTopicIdRef.current = topicId;
+    const prevTopicId = prevTopicIdRef.current;
+    prevTopicIdRef.current = topicId;
+
+    // 离开话题：立即持久化（补齐 debounce 未落盘部分）+ 暂存到内存（含附件）
+    if (prevTopicId != null) {
+      saveDraftText(prevTopicId, content);
+      draftsRef.current.set(prevTopicId, { content, attachments });
+    }
+
+    // 恢复目标话题草稿
+    if (topicId == null) {
       setContent('');
       setAttachments([]);
+      return;
     }
+    const cached = draftsRef.current.get(topicId);
+    if (cached) {
+      setContent(cached.content);
+      setAttachments(cached.attachments);
+    } else {
+      setContent(loadDraftText(topicId));
+      setAttachments([]);
+    }
+    // eslint-disable-next-line react-hooks/exhaustive-deps
   }, [topicId]);
 
+  // 文本草稿 debounce 持久化（300ms）
+  useEffect(() => {
+    if (topicId == null) return;
+    if (draftDebounceRef.current != null) {
+      window.clearTimeout(draftDebounceRef.current);
+    }
+    draftDebounceRef.current = window.setTimeout(() => {
+      saveDraftText(topicId, content);
+    }, 300);
+    return () => {
+      if (draftDebounceRef.current != null) {
+        window.clearTimeout(draftDebounceRef.current);
+      }
+    };
+  }, [content, topicId]);
+
+  // 卸载时补齐持久化（debounce 可能未落盘）+ 暂存内存草稿（含附件，供同 SPA 会话恢复）
+  const latestContentRef = useRef(content);
+  const latestAttachmentsRef = useRef(attachments);
+  latestContentRef.current = content;
+  latestAttachmentsRef.current = attachments;
+  useEffect(() => {
+    const drafts = draftsRef.current;
+    return () => {
+      const tid = prevTopicIdRef.current;
+      if (tid != null) {
+        saveDraftText(tid, latestContentRef.current);
+        drafts.set(tid, {
+          content: latestContentRef.current,
+          attachments: latestAttachmentsRef.current,
+        });
+      }
+    };
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, []);
+
   useEffect(() => {
     const textarea = textareaRef.current;
     if (textarea) {
@@ -250,6 +345,11 @@ export function MessageInput({
         content.trim(),
         attachments.map((a) => a.attachment),
       );
+      // 发送成功：清除该话题的草稿（内存 + localStorage）
+      if (topicId != null) {
+        draftsRef.current.delete(topicId);
+        saveDraftText(topicId, '');
+      }
       setContent('');
       setAttachments([]);
       setError(null);

web/src/components/Chat/MessageList.tsx

@@ -12,6 +12,8 @@ interface MessageListProps {
   viewKey?: string;
   /** 高亮的消息 ID，点击待办项后滚动并高亮显示 */
   highlightedMessageId?: string | null;
+  /** 主代理当前生效模型（透传给 MessageBubble 做 Task 卡片差异显示） */
+  effectiveModel?: { provider: string; model: string } | null;
 }
 
 export function MessageList({
@@ -20,6 +22,7 @@ export function MessageList({
   showThinking = true,
   viewKey,
   highlightedMessageId,
+  effectiveModel,
 }: MessageListProps) {
   const containerRef = useRef<HTMLDivElement>(null);
   const isAtBottomRef = useRef(true);
@@ -274,6 +277,7 @@ export function MessageList({
                     message={message}
                     onNavigateToSubAgent={onNavigateToSubAgent}
                     showThinking={showThinking}
+                    effectiveModel={effectiveModel}
                   />
                 </div>
               </div>

web/src/components/Chat/ModelSelector.tsx

@@ -1,10 +1,18 @@
 import { useState, useEffect, useRef, useCallback } from 'react';
 import { Cpu, ChevronDown, Loader2, Check } from 'lucide-react';
-import { listModelOptions, selectModel, getSelectedModel } from '../../api/experts';
+import {
+  listModelOptions,
+  selectModel,
+  getSelectedModel,
+  selectTopicModel,
+  getSelectedTopicModel,
+} from '../../api/experts';
 import type { ModelOptionsResponse } from '../Settings/types';
 
 interface ModelSelectorProps {
   sessionId: string | null;
+  /** 当前话题 ID：提供时按话题级选择读写（topic 优先，session 兜底） */
+  topicId?: string | null;
   /** 设置弹窗关闭信号（每次关闭递增，用于触发刷新） */
   settingsClosedTick?: number;
   /** 选择变化回调（参数为生效的 provider/model，未覆盖时为 current 默认） */
@@ -13,6 +21,7 @@ interface ModelSelectorProps {
 
 export function ModelSelector({
   sessionId,
+  topicId,
   settingsClosedTick,
   onSelectionChange,
 }: ModelSelectorProps) {
@@ -29,27 +38,36 @@ export function ModelSelector({
   const [draftModel, setDraftModel] = useState<string>('');
 
   const containerRef = useRef<HTMLDivElement>(null);
+  // 竞态防护：快速切换话题时，旧请求的响应晚于新请求返回会覆盖新状态。
+  // 每次发起刷新递增 token，响应落地时校验 token 未变才应用。
+  const refreshTokenRef = useRef(0);
 
-  // 刷新当前会话的用户模型覆盖
+  // 刷新当前话题/会话的用户模型覆盖（topic 级优先，session 级兜底）
   const refreshSelection = useCallback(() => {
     if (!sessionId) {
       setUserProvider(null);
       setUserModel(null);
       return;
     }
+    const token = ++refreshTokenRef.current;
     setLoading(true);
     setError(null);
-    getSelectedModel(sessionId)
+    const fetcher = topicId ? getSelectedTopicModel(topicId) : getSelectedModel(sessionId);
+    fetcher
       .then((data) => {
+        if (refreshTokenRef.current !== token) return; // 已被更新的刷新取代，丢弃
         setUserProvider(data.provider);
         setUserModel(data.model);
       })
       .catch(() => {
+        if (refreshTokenRef.current !== token) return;
         setUserProvider(null);
         setUserModel(null);
       })
-      .finally(() => setLoading(false));
+      .finally(() => {
-  }, [sessionId]);
+        if (refreshTokenRef.current === token) setLoading(false);
+      });
+  }, [sessionId, topicId]);
 
   // 加载模型选项（全局缓存，仅加载一次）
   useEffect(() => {
@@ -120,7 +138,10 @@ export function ModelSelector({
     setSaving(true);
     setError(null);
     try {
-      const result = await selectModel(sessionId, provider, model);
+      // 有话题时写话题级（后端双写 topics 行 + session store）；否则写 session 级
+      const result = topicId
+        ? await selectTopicModel(sessionId, topicId, provider, model)
+        : await selectModel(sessionId, provider, model);
       if (!result.success) {
         setError(result.error || '切换模型失败');
         setTimeout(() => setError(null), 3000);

web/src/components/Panel/TopicTokenStatsPanel.tsx

@@ -1,6 +1,12 @@
 import { Coins } from 'lucide-react';
 import type { TopicTokenStats } from '../../types/protocol';
-import { formatTokenCount, contextOccupancyPct, occupancyColor } from '../../utils/tokenStats';
+import {
+  formatTokenCount,
+  contextOccupancyPct,
+  occupancyColor,
+  cacheHitRatePct,
+  cacheHitColor,
+} from '../../utils/tokenStats';
 
 interface TopicTokenStatsPanelProps {
   tokenStats?: TopicTokenStats | null;
@@ -21,6 +27,7 @@ export function TopicTokenStatsPanel({ tokenStats }: TopicTokenStatsPanelProps)
   }
 
   const pct = contextOccupancyPct(tokenStats);
+  const cachePct = cacheHitRatePct(tokenStats);
 
   return (
     <div className="shrink-0 border-b border-[var(--border-color)] p-3">
@@ -64,8 +71,38 @@ export function TopicTokenStatsPanel({ tokenStats }: TopicTokenStatsPanelProps)
             {formatTokenCount(tokenStats.context_window_tokens)}
           </span>
         </div>
+        <div className="flex justify-between col-span-2">
+          <span className="text-[var(--text-muted)]">缓存命中</span>
+          <span className="text-[var(--text-secondary)] font-mono">
+            {formatTokenCount(tokenStats.cached_tokens ?? 0)}
+          </span>
+        </div>
       </div>
 
+      {/* 缓存命中率 + 进度条（越高越省钱 → 绿） */}
+      {cachePct != null && (
+        <div className="mt-2">
+          <div className="flex items-center justify-between text-xs mb-1">
+            <span className="text-[var(--text-muted)]">缓存命中率</span>
+            <span className={`font-mono font-medium ${cacheHitColor(cachePct)}`}>
+              cache {cachePct}%
+            </span>
+          </div>
+          <div className="h-1.5 rounded-full bg-[var(--overlay-subtle)] overflow-hidden">
+            <div
+              className={`h-full rounded-full transition-all ${
+                cachePct >= 50
+                  ? 'bg-[var(--accent-green)]'
+                  : cachePct > 0
+                    ? 'bg-[var(--accent-amber)]'
+                    : 'bg-[var(--overlay-subtle)]'
+              }`}
+              style={{ width: `${Math.max(cachePct, 1)}%` }}
+            />
+          </div>
+        </div>
+      )}
+
       {/* 上下文占用百分比 + 进度条 */}
       {pct != null && (
         <div className="mt-2">

web/src/hooks/chat/useTopics.ts

@@ -8,6 +8,7 @@ import {
   type MutableRefObject,
 } from 'react';
 import type { Topic, TopicList, TopicRenamed, TopicSummary, Command } from '../../types/protocol';
+import { clearTopicDraft } from '../../components/Chat/MessageInput';
 
 export interface UseTopicsReturn {
   topics: Topic[];
@@ -101,6 +102,8 @@ export function useTopics(): UseTopicsReturn {
   }, []);
 
   const deleteTopic = useCallback((topicId: string): Command => {
+    // 同步清除该话题的持久化草稿（删除后端话题成功与否均无碍：残留草稿 key 无话题可挂载）
+    clearTopicDraft(topicId);
     return { type: 'delete_topic', topic_id: topicId };
   }, []);
 

web/src/hooks/useChat.ts

@@ -182,7 +182,9 @@ export function useChat(): UseChatReturn {
 
     // Tier 3: 主视图路由
     // 3a: 带 subagent_task_id 的消息在主视图直接丢弃（已在 Tier 2 未命中）
-    if (getSubagentTaskId(message)) return;
+    // 例外：execution_completed 携带 subagent_status，需转发到 handleMainViewMessage
+    // 更新主视图 task 卡片占位状态（running → completed 等），否则卡片永远显示运行中
+    if (getSubagentTaskId(message) && message.type !== 'execution_completed') return;
 
     // 3b: 非 chat 消息的 case 分发
     switch (message.type) {

web/src/types/protocol.ts

@@ -153,6 +153,8 @@ export interface TopicTokenStats {
   prompt_tokens: number;
   completion_tokens: number;
   total_tokens: number;
+  /** 累计缓存命中的输入 tokens 数（老数据为 0） */
+  cached_tokens: number;
   last_prompt_tokens?: number;
   context_window_tokens: number;
 }
@@ -519,6 +521,10 @@ export interface TaskToolResult {
   summary: string;
   output: string;
   task_id: string;
+  /** 子代理实际使用的 provider 名（running 占位结果携带，供卡片差异显示） */
+  provider?: string;
+  /** 子代理实际使用的 model id（同上） */
+  model?: string;
 }
 
 export interface Topic {

web/src/utils/tokenStats.ts

@@ -15,9 +15,23 @@ export function contextOccupancyPct(stats: TopicTokenStats): number | null {
   return Math.min(100, Math.round((last / stats.context_window_tokens) * 100));
 }
 
-/** 根据占用率返回颜色 class */
+/** 根据占用率返回颜色 class（占用越高越危险 → 红） */
 export function occupancyColor(pct: number): string {
   if (pct >= 80) return 'text-[rgb(242,90,90)]';
   if (pct >= 50) return 'text-[var(--accent-amber)]';
   return 'text-[var(--accent-green)]';
 }
+
+/** 计算累计缓存命中率：SUM(cached_tokens) / SUM(prompt_tokens)，返回 0-100 */
+export function cacheHitRatePct(stats: TopicTokenStats): number | null {
+  if (!stats.prompt_tokens || stats.prompt_tokens === 0) return null;
+  return Math.min(100, Math.round(((stats.cached_tokens ?? 0) / stats.prompt_tokens) * 100));
+}
+
+/** 缓存命中率颜色（与 ctx 占用方向相反：命中率越高越省钱 → 绿）。
+ * 0% 返回 muted 色（新话题首轮必然为 0，不算异常） */
+export function cacheHitColor(pct: number): string {
+  if (pct <= 0) return 'text-[var(--text-muted)]';
+  if (pct >= 50) return 'text-[var(--accent-green)]';
+  return 'text-[var(--accent-amber)]';
+}