Compare commits

..

11 Commits

Author SHA1 Message Date
oudecheng
b26a2c2512 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 占用方向相反)
2026-08-15 19:21:51 +08:00
oudecheng
82b6a882a2 test(scheduler): 调度器非阻塞派发与并发容量回归测试
- 慢执行器(400ms sleep)双任务:process_tick 必须立即返回且两任务并发完成(串行需 >=800ms,断言 <700ms)

- worker_queue_capacity=1:槽位耗尽时多余到期任务被推迟(保持 Scheduled),槽位释放后下个 tick 正常派发

- 执行完成后状态机验证:回到 Scheduled、status=ok、next_fire_at 推进
2026-08-15 19:04:54 +08:00
oudecheng
9b8c64bd21 refactor(scheduler): 任务执行移出 tick 循环,worker_queue_capacity 并发约束生效
原 process_tick 内联 await execute_job:长耗时 agent 任务(可达数分钟)会串行阻塞整个调度循环,拖后所有其他 job 的触发;worker_queue_capacity 配置项从未被使用。

- 到期 job 的执行与事后状态写入移入 tokio::spawn 后台任务,tick 循环只做派发

- 新增 worker_semaphore(容量=worker_queue_capacity):槽位耗尽时 job 保持 Scheduled 留待下个 tick,形成有界背压

- job 派发前置为 Running 状态,is_due 要求 Scheduled,天然防止重复派发

- after_execution 状态推进失败时回退为 Scheduled,避免 job 永久卡在 Running

- execute_job 拆分为 execute_job_inner(不依赖 &self,可移入后台任务)
2026-08-15 18:50:40 +08:00
oudecheng
4b93c84447 test(gateway): 定时任务 topic 补齐与消息送达回归测试
- 新增捕获请求的 mock OpenAI server,直接断言任务 prompt 进入 LLM 输入

- 静默任务按生产默认 fresh_session=true 连续执行两轮,验证清空-重建时序下 prompt 仍送达、topic 自动创建并持久化、系统提示词含送达提示

- ensure_topic_for_chat 幂等性与全新 chat 默认 topic 创建测试

- append_persisted_message 无 topic 回退 chat_id 键内存历史测试

- SessionManager/SessionLifecycleService 暴露 get_scheduler_session 供测试断言
2026-08-15 18:17:29 +08:00
oudecheng
cbe2ff1339 feat(gateway): 静默定时任务增加主动送达提示,避免结果静默丢失
silent_agent_task 的最终响应发往 scheduler/ 虚拟会话、被出站分发器跳过,成功时用户收不到任何消息,只能依赖 agent 主动调用 send_session_message。

- 定时任务系统提示词仅对 scheduler/ 虚拟会话追加送达提示(避免普通 agent_task 重复发送)

- scheduler_manage 工具描述新增 Silent Mode Delivery 指引:用户期望收到结果时,prompt 必须要求用 send_session_message 交付
2026-08-15 18:17:07 +08:00
oudecheng
508806a408 fix(gateway): 定时任务执行前自动补齐 topic,修复 agent 空历史执行
scheduler/ 虚拟会话从不经用户消息分配 topic,导致定时任务的系统提示词与任务 prompt 只落库、不进内存历史(append_persisted_message 在 topic=None 时跳过内存写入),agent 以空历史执行,任务实际未运行、消息无法送达会话。

- Session 新增 ensure_topic_for_chat:内存无 topic 时从 DB 恢复最近活跃 topic,无则自动创建(先确保持久会话以规避外键约束)

- prepare_and_execute_scheduled_task 捕获 topic 前调用 ensure,保证 original_topic_id 恒为 Some

- append_persisted_message 无 topic 时回退到 chat_id 键内存历史,兜底所有无 topic 路径
2026-08-15 18:16:50 +08:00
oudecheng
73c25e5a20 feat(web): Task 卡片差异显示子代理模型名 + 子代理完成状态转发修复
- Task/tool 卡片头部在子代理模型与主代理不同(model 或 provider 任一不同)时显示模型徽章
- useChat 放行带 subagent_task_id 的 execution_completed 消息,更新卡片 running→completed 状态
2026-08-15 15:35:32 +08:00
oudecheng
53e2fb6dc6 feat(web): per-topic 输入草稿记忆(切换暂存/恢复 + localStorage 持久化 + 删除清理)
- 文本草稿 debounce 持久化到 localStorage(key 按 topicId),附件仅存内存
- 切换话题时暂存当前草稿并恢复目标话题草稿,组件卸载时补齐落盘
- 删除话题时同步清除其持久化草稿,避免孤儿 key
2026-08-15 15:35:07 +08:00
oudecheng
8f59b6b93a feat(web): 话题级模型选择前端(ModelSelector topicId 化 + 竞态防护)+ 专家系统开关隐藏
- ModelSelector 支持按 topicId 读写话题级选择(topic 优先,session 兜底)
- 快速切换话题时用递增令牌丢弃过期响应,防止乱序覆盖
- ChatContainer 透传 topicId,向后端双写新话题默认模型
- ExpertSelector 依据后端 experts_system_enabled 隐藏选择器并清除已选专家
2026-08-15 15:34:38 +08:00
oudecheng
dc693fa80b feat(model): 主代理/子代理模型解析链 topic>session>expert>config + TaskToolResult 附带实际模型
- 主代理解析链新增 topic 级选择(最高优先级),session 级首命中时物化固化为话题选择
- 子代理解析链共享 helper:def frontmatter > 话题级 > session 级 > 全局基础配置
- TaskToolResult 携带子代理实际使用的 provider/model,供前端差异展示
2026-08-15 15:34:18 +08:00
oudecheng
414105d419 feat(model): 话题级模型选择持久化与 API(topics 表加 provider/model 列 + 双写内存缓存)
- topics 表迁移新增 provider/model 列,记录话题级显式模型选择
- storage 层新增 update_topic_model / list_topic_model_selections
- 新增 POST /api/topic/select-model 与 GET /api/topic/selected-model
- 双写 key 取 topic 行自带 session_id(不信任请求体,防污染其他会话)
- 清除时同步清 session 级选择,避免'重置为默认'被物化逻辑重新覆盖
2026-08-15 15:33:58 +08:00
35 changed files with 1564 additions and 175 deletions

View File

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

View File

@ -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,
}
}

View File

@ -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),
});

View File

@ -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),
});

View File

@ -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"

View File

@ -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 尚无 topicscheduler/ 虚拟会话从不经用户消息分配
// 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
.current_topic(request.chat_id)
.map(|s| s.to_string());
let tid = match session_guard.current_topic(request.chat_id) {
Some(topic_id) => Some(topic_id.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(),

View File

@ -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 storekey 一律取 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 })
}

View File

@ -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));

View File

@ -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,
))
}

View File

@ -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,12 +535,15 @@ 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 {
if let Some(ref tid) = topic_id {
match topic_id.as_deref() {
Some(tid) if Some(tid) == current_chat_topic => {
self.add_message(tid, message);
}
} else {
Some(_) => {
tracing::info!(
chat_id = %chat_id,
write_topic_id = ?topic_id,
@ -498,6 +551,10 @@ impl Session {
"Skipping memory history update: message belongs to a different topic"
);
}
None => {
self.add_message(chat_id, message);
}
}
// 更新 topic 的最后活跃时间
if let Some(ref topic_id) = topic_id {
@ -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 自动拥有 topicensure_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 {

View File

@ -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;
}

View File

@ -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!(

View File

@ -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(),
},
})
}

View File

@ -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,
}
/// 流式响应中的增量事件

View File

@ -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,13 +277,28 @@ impl Scheduler {
job.completed_at,
)?;
let execution_result = self.execute_job(&job).await;
job.after_execution(
now,
execution_result.as_ref().err().map(|err| err.to_string()),
self.config.misfire_policy,
self.timezone,
)?;
// 执行与事后状态写入移入后台任务tick 循环只做派发,
// 长耗时任务agent_task 可能长达数分钟)不再串行阻塞其他 job 的触发。
// job 在 DB 中已是 Running 状态is_due 要求 Scheduled因此不会被重复派发。
let bus = self.bus.clone();
let jobs_repo = self.jobs.clone();
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;
if let Err(error) = &execution_result {
tracing::error!(job_id = %job.id, error = %error, "Scheduler job failed");
}
let status = if execution_result.is_ok() {
Some(SchedulerJobStatus::Ok)
@ -274,11 +306,15 @@ impl Scheduler {
Some(SchedulerJobStatus::Error)
};
if let Err(error) = &execution_result {
tracing::error!(job_id = %job.id, error = %error, "Scheduler job failed");
}
self.jobs.update_scheduler_job_runtime(
let mut job = job;
match job.after_execution(
fire_at,
execution_result.as_ref().err().map(|err| err.to_string()),
misfire_policy,
timezone,
) {
Ok(()) => {
if let Err(error) = jobs_repo.update_scheduler_job_runtime(
&job.id,
job.state.clone(),
status,
@ -288,34 +324,89 @@ impl Scheduler {
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
.agent_task_executor
if let Err(error) = 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
.bus
if let Err(e) = 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;
}
}

View File

@ -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=0LLM 可见),通过 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(

View File

@ -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
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)
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, ?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
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)
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, ?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, \
COALESCE(SUM(prompt_tokens), 0) AS sum_prompt, \
COALESCE(SUM(completion_tokens), 0) AS sum_completion, \
COALESCE(SUM(total_tokens), 0) AS sum_total \
"SELECT topic_id, {USAGE_SUM_COLUMNS} \
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((
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,
},
))
Ok((row.get::<_, String>(0)?, read_usage_sum_row(row, 1)?))
})?;
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
let sum_sql = "SELECT \
COALESCE(SUM(prompt_tokens), 0), \
COALESCE(SUM(completion_tokens), 0), \
COALESCE(SUM(total_tokens), 0) \
// 1. SUM 查询:累计 prompt/completion/total/cached
// 列清单与行映射复用 USAGE_SUM_COLUMNS / read_usage_sum_row与 topic 聚合共享)
let sum_sql = format!(
"SELECT {USAGE_SUM_COLUMNS} \
FROM messages \
WHERE session_id = ?1 AND role = 'assistant'";
let mut stmt = conn.prepare(sum_sql)?;
let sum_row = stmt.query_row(params![session_id], |row| {
Ok((
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;
WHERE session_id = ?1 AND role = 'assistant'"
);
let mut stmt = conn.prepare(&sum_sql)?;
let sum_stats = stmt.query_row(params![session_id], |row| read_usage_sum_row(row, 0))?;
// 无 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,
completion_tokens,
total_tokens,
prompt_tokens: sum_stats.prompt_tokens,
completion_tokens: sum_stats.completion_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)?,
})
})?;

View File

@ -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>,
}

View File

@ -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)?,
})
}

View File

@ -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. \

View File

@ -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 {
Some(d) if d.provider.is_some() || d.model.is_some() => self
.model_resolver
.resolve(
d.provider.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(),
};
let effective_provider_config = self.resolve_effective_provider_config(
session,
def.map(|d| d.name.as_str()),
def.and_then(|d| d.provider.as_deref()),
def.and_then(|d| d.model.as_deref()),
)?;
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 覆盖,保证环境提示中的模型名与实际使用的模型一致
let effective_provider_config = match (def.provider.is_some(), def.model.is_some()) {
(true, _) | (_, true) => self
.model_resolver
.resolve(
// 同步解析生效配置def > topic 级 > session 级 > 全局),
// 与 create_subagent 共用 helper保证环境提示中的模型名与实际使用的模型一致
let effective_provider_config = self.resolve_effective_provider_config(
&session,
Some(def.name.as_str()),
def.provider.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()),
});
}

View File

@ -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>,
}
/// 异步子代理完成状态

View File

@ -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';

View File

@ -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();
}

View File

@ -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>
)}

View File

@ -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) return null;
// If sessionId is null or expert system disabled in settings, render nothing
if (!sessionId || !systemEnabled) return null;
return (
<div ref={containerRef} className="relative shrink-0 flex items-center gap-2">

View File

@ -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

View File

@ -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;
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);

View File

@ -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>

View File

@ -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));
}, [sessionId]);
.finally(() => {
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);

View File

@ -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,7 +71,37 @@ 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 && (

View File

@ -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 };
}, []);

View File

@ -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) {

View File

@ -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 {

View File

@ -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)]';
}