From deab3699fe0cb37664015b756877967c7ff24f07 Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Wed, 5 Aug 2026 09:43:09 +0800 Subject: [PATCH] =?UTF-8?q?fix(web):=20token=20=E7=BB=9F=E8=AE=A1=E6=8C=89?= =?UTF-8?q?=20topic=5Fid=20=E8=81=9A=E5=90=88=EF=BC=8C=E9=81=BF=E5=85=8D?= =?UTF-8?q?=E5=90=8C=20session=20=E5=A4=9A=20topic=20=E5=85=B1=E4=BA=AB?= =?UTF-8?q?=E6=80=BB=E5=92=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit batch_session_token_stats 重命名为 batch_topic_token_stats，SQL 从 GROUP BY session_id 改为 GROUP BY topic_id，修复同一 session 下多个 topic 显示相同 token 总和的问题。 子代理天然隔离：子代理消息的 topic_id 属于子代理自身的 topic，不在主 topic 列表中。topic_id IN (...) 自动排除 NULL topic_id 的旧消息。 对抗性审查补充：latest assistant 消息查询的 JOIN 条件 m.topic_id = latest.topic_id AND m.seq = latest.max_seq 可能匹配到子代理消息（seq 是 session 级递增，主/子 session 可能有相同 seq）。在外层 JOIN 添加 WHERE m.session_id NOT LIKE 'sub:%'，避免重复行导致 stats.entry(tid) 被覆盖、结果不确定。 --- src/command/handlers/list_topics.rs | 25 ++++++------ src/storage/mod.rs | 62 +++++++++++++++++++---------- 2 files changed, 53 insertions(+), 34 deletions(-) diff --git a/src/command/handlers/list_topics.rs b/src/command/handlers/list_topics.rs index c57dc68..3573c72 100644 --- a/src/command/handlers/list_topics.rs +++ b/src/command/handlers/list_topics.rs @@ -40,9 +40,10 @@ pub struct TopicSummary { /// 构建 TopicSummary 列表的公共函数。 /// -/// 一次批量查询所有 topic 对应 session 的 token 统计，避免 N 次 RTT。 -/// 按 `session_id` 聚合天然分离主 agent 与子 agent（子代理 session_id 形如 -/// `sub:...`，不在此查询的主 session_id 列表中）。 +/// 一次批量查询所有 topic 的 token 统计，按 `topic_id` 聚合， +/// 避免按 session_id 聚合时同 session 下多个 topic 共享同一总和。 +/// 子代理天然隔离：子代理消息的 topic_id 属于子代理自身的 topic， +/// 不在主 topic 列表中。 /// /// `context_window_tokens` 来自最新 assistant 消息记录（LLM 调用时持久化）， /// 无需从配置链路注入，保持存储层与配置解耦。 @@ -54,24 +55,24 @@ pub fn build_topic_summaries( return Ok(Vec::new()); } - // 收集所有 topic 的 session_id（去重） - let mut session_ids: Vec = Vec::new(); + // 收集所有 topic_id（去重） + let mut topic_ids: Vec = Vec::new(); for t in &topics { - if !session_ids.contains(&t.session_id) { - session_ids.push(t.session_id.clone()); + if !topic_ids.contains(&t.id) { + topic_ids.push(t.id.clone()); } } - let session_id_refs: Vec<&str> = session_ids.iter().map(|s| s.as_str()).collect(); + let topic_id_refs: Vec<&str> = topic_ids.iter().map(|s| s.as_str()).collect(); - // 一次批量查询 token 统计 + // 一次批量查询 token 统计（按 topic_id 聚合） let stats_map: HashMap = store - .batch_session_token_stats(&session_id_refs) + .batch_topic_token_stats(&topic_id_refs) .map_err(|e| CommandError::new("TOKEN_STATS_ERROR", e.to_string()))?; let summaries = topics .into_iter() .map(|t| { - let token_stats = stats_map.get(&t.session_id).map(|s| TopicTokenStats { + let token_stats = stats_map.get(&t.id).map(|s| TopicTokenStats { prompt_tokens: s.prompt_tokens, completion_tokens: s.completion_tokens, total_tokens: s.total_tokens, @@ -140,8 +141,6 @@ async fn handle_list_topics( .list_topics(&session_id) .map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))?; - // context_window_tokens 当前未从配置链路注入（前端可从已有 config 接口获取）， - // 此处传 0 表示"后端不提供上限"，前端按需隐藏百分比。 let summaries = build_topic_summaries(handler.store.as_ref(), topics)?; let topics_json = serde_json::to_string(&summaries) diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 4226a10..b1243e7 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -1630,32 +1630,44 @@ impl SessionStore { .map_err(StorageError::from) } - /// 批量查询多个 session 的 token 消耗统计（cost 累计 + context 瞬时）。 - pub fn batch_session_token_stats( + /// 批量查询多个 topic 的 token 消耗统计（cost 累计 + context 瞬时）。 + /// + /// 按 `topic_id` 聚合而非 `session_id`：一个 session 可包含多个 topic， + /// 若按 session_id 聚合会导致同 session 下的所有 topic 显示相同的总和。 + /// + /// 子代理隔离：子代理消息持久化时 session_id='sub:...'，topic_id=父 topic_id + /// （见 task::runtime PersistingEmittedMessageHandler 构造），因此不能仅靠 + /// topic_id 隔离。此处用 `session_id NOT LIKE 'sub:%'` 显式排除子代理消息， + /// 与项目约定一致（session 列表同样过滤 'sub:%'）。子代理 token 不计入父 topic， + /// 保持"子代理分别计算"语义。 + pub fn batch_topic_token_stats( &self, - session_ids: &[&str], + topic_ids: &[&str], ) -> Result, StorageError> { - if session_ids.is_empty() { + if topic_ids.is_empty() { return Ok(HashMap::new()); } let conn = self.pool.get()?; - let placeholders = (0..session_ids.len()) + let placeholders = (0..topic_ids.len()) .map(|i| format!("?{}", i + 1)) .collect::>() .join(", "); + // topic_id IN (...) 自动排除 NULL topic_id 的旧消息； + // session_id NOT LIKE 'sub:%' 排除子代理消息（其 topic_id=父 topic_id） let sum_sql = format!( - "SELECT session_id, \ + "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 \ FROM messages \ - WHERE session_id IN ({placeholders}) AND role = 'assistant' \ - GROUP BY session_id" + WHERE topic_id IN ({placeholders}) AND role = 'assistant' \ + AND session_id NOT LIKE 'sub:%' \ + GROUP BY topic_id" ); let mut stmt = conn.prepare(&sum_sql)?; - let params: Vec<&dyn rusqlite::ToSql> = session_ids + let params: Vec<&dyn rusqlite::ToSql> = topic_ids .iter() .map(|s| s as &dyn rusqlite::ToSql) .collect(); @@ -1674,26 +1686,34 @@ impl SessionStore { let mut stats: HashMap = HashMap::new(); for row in sum_rows { - let (sid, s) = row?; - stats.insert(sid, s); + let (tid, s) = row?; + stats.insert(tid, s); } - // 查找每个 session 中最新的**有 usage 数据的** assistant 消息， + // 查找每个 topic 中最新的**有 usage 数据的** assistant 消息， // 读取其 prompt_tokens 和 context_window_tokens。 - // 过滤 prompt_tokens IS NOT NULL 确保跳过 error/cancel 消息（usage 为 NULL）。 + // 过滤 prompt_tokens IS NOT NULL 确保跳过 error/cancel 消息（usage 为 NULL）； + // session_id NOT LIKE 'sub:%' 排除子代理消息，避免取到子代理的 context_window。 + // + // 注意：seq 是 session 级递增（见 append_message_with_topic），主 session 与 + // 子代理 session 各自独立计数，可能存在相同 seq。外层 WHERE 必须再次过滤 + // session_id NOT LIKE 'sub:%'，否则 JOIN 会同时匹配主消息和子代理消息， + // 导致重复行并使 stats.entry(tid) 被覆盖，结果不确定。 let last_sql = format!( - "SELECT m.session_id, m.prompt_tokens, m.context_window_tokens \ + "SELECT m.topic_id, m.prompt_tokens, m.context_window_tokens \ FROM messages m \ INNER JOIN ( \ - SELECT session_id, MAX(seq) AS max_seq \ + SELECT topic_id, MAX(seq) AS max_seq \ FROM messages \ - WHERE session_id IN ({placeholders}) AND role = 'assistant' \ + WHERE topic_id IN ({placeholders}) AND role = 'assistant' \ AND prompt_tokens IS NOT NULL \ - GROUP BY session_id \ - ) latest ON m.session_id = latest.session_id AND m.seq = latest.max_seq" + AND session_id NOT LIKE 'sub:%' \ + GROUP BY topic_id \ + ) latest ON m.topic_id = latest.topic_id AND m.seq = latest.max_seq \ + WHERE m.session_id NOT LIKE 'sub:%'" ); let mut stmt2 = conn.prepare(&last_sql)?; - let params2: Vec<&dyn rusqlite::ToSql> = session_ids + let params2: Vec<&dyn rusqlite::ToSql> = topic_ids .iter() .map(|s| s as &dyn rusqlite::ToSql) .collect(); @@ -1705,8 +1725,8 @@ impl SessionStore { )) })?; for row in last_rows { - let (sid, last_prompt, last_ctx_window) = row?; - let entry = stats.entry(sid).or_insert(SessionTokenStats { + let (tid, last_prompt, last_ctx_window) = row?; + let entry = stats.entry(tid).or_insert(SessionTokenStats { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0,