Compare commits
11 Commits
9e503f2672
...
b26a2c2512
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b26a2c2512 | ||
|
|
82b6a882a2 | ||
|
|
9b8c64bd21 | ||
|
|
4b93c84447 | ||
|
|
cbe2ff1339 | ||
|
|
508806a408 | ||
|
|
73c25e5a20 | ||
|
|
53e2fb6dc6 | ||
|
|
8f59b6b93a | ||
|
|
dc693fa80b | ||
|
|
414105d419 |
@ -3353,6 +3353,7 @@ mod tests {
|
||||
prompt_tokens: 10,
|
||||
completion_tokens: 10,
|
||||
total_tokens: 20,
|
||||
cached_tokens: 0,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@ -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,
|
||||
}
|
||||
}
|
||||
|
||||
@ -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),
|
||||
});
|
||||
|
||||
@ -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),
|
||||
});
|
||||
|
||||
@ -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"
|
||||
|
||||
@ -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
|
||||
.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(),
|
||||
|
||||
@ -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 })
|
||||
}
|
||||
|
||||
@ -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));
|
||||
|
||||
|
||||
@ -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,
|
||||
))
|
||||
}
|
||||
|
||||
@ -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 自动拥有 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 {
|
||||
|
||||
@ -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;
|
||||
}
|
||||
|
||||
@ -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!(
|
||||
|
||||
@ -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(),
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
@ -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,
|
||||
}
|
||||
|
||||
/// 流式响应中的增量事件
|
||||
|
||||
@ -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;
|
||||
}
|
||||
}
|
||||
|
||||
@ -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(
|
||||
|
||||
@ -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)?,
|
||||
})
|
||||
})?;
|
||||
|
||||
|
||||
@ -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>,
|
||||
}
|
||||
|
||||
@ -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)?,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@ -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. \
|
||||
|
||||
@ -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()),
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@ -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>,
|
||||
}
|
||||
|
||||
/// 异步子代理完成状态
|
||||
|
||||
@ -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';
|
||||
|
||||
@ -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();
|
||||
}
|
||||
|
||||
@ -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>
|
||||
)}
|
||||
|
||||
@ -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">
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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);
|
||||
|
||||
@ -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>
|
||||
|
||||
@ -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);
|
||||
|
||||
@ -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 && (
|
||||
|
||||
@ -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 };
|
||||
}, []);
|
||||
|
||||
|
||||
@ -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) {
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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)]';
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user