From 7e9b99bcbf1de46e95a3fe6b4791869bb33f16c9 Mon Sep 17 00:00:00 2001 From: Jason Date: Fri, 2 Oct 2026 20:34:14 +0800 Subject: [PATCH] feat(usage): estimate request duration from Claude Code and Codex session logs Requests imported from session logs had no timing, so speed was only available through the routing service. The importers now derive a duration from log timestamps and store it in latency_ms (first_token_ms stays NULL, 0 still means "no timing"). The duration includes the wait for the first token, so the resulting speed reads lower than the routed figure; provider stats sum it separately (est_speed_*) and never mix it into the exact speed. Only rows with at least 200 output tokens and a duration of at least one second count. Claude Code: a reply ends at its latest block timestamp. Its start is found by walking parentUuid up from the first block (apiBlockIndex 0), skipping attachment rows, to the user message or tool result. The previous line cannot be used: results of parallel tool calls are written between blocks of the same reply. The start usually sits before the byte cursor, so each incremental read looks back 256 KiB for chain and block info without importing it. A reply seen in two syncs gets its duration extended when the later blocks arrive. Codex: rollout layouts differ by version. A token_usage_record marks the end of the response when present; otherwise token_count does, unless it was written only after the tool outputs, in which case the last output item is used. The start is the latest boundary before the first output item. Output items that old versions wrote after their own token_count are ignored, and task_started / turn_aborted clear the state so an interrupted request does not lend its start to the next one. Rows imported earlier keep latency 0; nothing is backfilled. --- src-tauri/src/services/session_usage.rs | 371 ++++++++++++++- src-tauri/src/services/session_usage_codex.rs | 438 +++++++++++++++++- src-tauri/src/services/usage_stats.rs | 81 +++- 3 files changed, 863 insertions(+), 27 deletions(-) diff --git a/src-tauri/src/services/session_usage.rs b/src-tauri/src/services/session_usage.rs index d3549adf6..8b232445a 100644 --- a/src-tauri/src/services/session_usage.rs +++ b/src-tauri/src/services/session_usage.rs @@ -201,6 +201,167 @@ struct ParsedAssistantUsage { session_id: Option, } +/// 估算出的请求耗时短于这个毫秒数就不要：真实请求不可能这么快，多半是起点取错了。 +const MIN_PLAUSIBLE_LATENCY_MS: i64 = 100; + +/// 估算出的请求耗时长于这个毫秒数也不要：中间多半夹了休眠或长时间的重试等待。 +const MAX_PLAUSIBLE_LATENCY_MS: i64 = 60 * 60 * 1000; + +/// 会话日志没有请求计时，耗时只能拿两个时间戳相减估出来（含首字等待）。 +/// 落在合理范围之外的返回 None，入库时写 0，表示没有计时。 +/// +/// Shared by the Claude and Codex session parsers. +pub(crate) fn estimated_latency_ms(start_ms: i64, end_ms: i64) -> Option { + let latency = end_ms.checked_sub(start_ms)?; + (MIN_PLAUSIBLE_LATENCY_MS..=MAX_PLAUSIBLE_LATENCY_MS) + .contains(&latency) + .then_some(latency) +} + +pub(crate) fn parse_timestamp_millis(timestamp: &str) -> Option { + chrono::DateTime::parse_from_rfc3339(timestamp) + .ok() + .map(|dt| dt.timestamp_millis()) +} + +/// 一行日志在对话链上的位置，只留估算请求耗时用得到的三样。 +struct ChainNode { + parent: Option, + timestamp_ms: Option, + /// `attachment` 行的时间戳不可靠（有的是回复开始后才补记的），找请求起点时跳过。 + is_attachment: bool, +} + +type ChainNodes = HashMap; + +fn parse_line_uuid(value: &serde_json::Value, key: &str) -> Option { + value + .get(key) + .and_then(|v| v.as_str()) + .and_then(|s| uuid::Uuid::parse_str(s).ok()) +} + +fn record_chain_node(chain: &mut ChainNodes, value: &serde_json::Value) { + let Some(uuid) = parse_line_uuid(value, "uuid") else { + return; + }; + chain.insert( + uuid, + ChainNode { + parent: parse_line_uuid(value, "parentUuid"), + timestamp_ms: value + .get("timestamp") + .and_then(|v| v.as_str()) + .and_then(parse_timestamp_millis), + is_attachment: value.get("type").and_then(|t| t.as_str()) == Some("attachment"), + }, + ); +} + +/// 增量读取时往游标前回看的字节数。回复的第一块常常紧跟在游标后面，而它的 +/// 起点行（用户消息、工具结果）在游标前面：工具结果先落盘，回复要等整条 +/// 结束才一次写完。起点行比这个窗口还大（比如带图片的工具结果）时，那一条 +/// 请求就不估耗时。 +const CHAIN_SEED_BYTES: i64 = 256 * 1024; + +/// 把游标前一小段里的行记进对话链和回复计时（只取这两样，不导入）。回复的 +/// 前几块可能已经被上一轮读走，这一轮要靠它们才知道回复从哪一块开始。返回后 +/// 文件位置停在 `offset`。回看读不出来不算错，只是少估几条耗时。 +fn read_chain_seed( + file: &mut fs::File, + offset: i64, + chain: &mut ChainNodes, + timings: &mut HashMap, +) -> Result<(), AppError> { + let len = offset.clamp(0, CHAIN_SEED_BYTES); + if len == 0 { + return Ok(()); + } + let mut window = vec![0u8; len as usize]; + let read_ok = file + .seek(SeekFrom::Start((offset - len) as u64)) + .and_then(|_| file.read_exact(&mut window)) + .is_ok(); + file.seek(SeekFrom::Start(offset as u64)) + .map_err(|e| AppError::Config(format!("无法定位文件偏移: {e}")))?; + if !read_ok { + return Ok(()); + } + + let mut lines = window.split(|b| *b == b'\n'); + if offset > len { + // 窗口不是从文件头开始，第一段多半是半行 + lines.next(); + } + for line in lines { + if let Ok(value) = serde_json::from_slice::(line) { + record_chain_node(chain, &value); + if let Some(msg_id) = assistant_message_id(&value) { + record_block_timing(timings, &value, msg_id); + } + } + } + Ok(()) +} + +/// assistant 行的 `message.id`；别的行返回 None。 +fn assistant_message_id(value: &serde_json::Value) -> Option<&str> { + if value.get("type").and_then(|t| t.as_str()) != Some("assistant") { + return None; + } + value.get("message")?.get("id")?.as_str() +} + +/// 一条回复（同一个 message.id 的所有内容块）的计时线索。 +struct MessageTiming { + /// 文件里最先出现的那一块的父行，请求起点沿它往上找。 + first_parent: Option, + /// 最先出现的那一块是不是这条回复的第一块。不是的话（前面的块在回看窗口 + /// 之外）起点会取晚、速度偏高，宁可不估。旧版日志没有 `apiBlockIndex`，按是处理。 + starts_at_first_block: bool, + /// 各块时间戳（块写完的时刻）里最晚的一个，即回复结束。 + end_ms: Option, +} + +/// 把一个内容块记进它所属回复的计时：最先出现的块定起点，各块里最晚的定结束。 +fn record_block_timing( + timings: &mut HashMap, + value: &serde_json::Value, + msg_id: &str, +) { + let block_end_ms = value + .get("timestamp") + .and_then(|v| v.as_str()) + .and_then(parse_timestamp_millis); + let timing = timings + .entry(msg_id.to_string()) + .or_insert_with(|| MessageTiming { + first_parent: parse_line_uuid(value, "parentUuid"), + starts_at_first_block: value + .get("apiBlockIndex") + .is_none_or(|index| index.as_u64() == Some(0)), + end_ms: None, + }); + timing.end_ms = timing.end_ms.max(block_end_ms); +} + +/// 沿父链往上最多走这么多行。 +const MAX_CHAIN_HOPS: usize = 32; + +/// 请求起点：从回复第一块的父行往上，跳过 `attachment`，第一个别的行（用户 +/// 消息、工具结果、出错重试的 system 行）的时间戳。 +fn resolve_request_start_ms(chain: &ChainNodes, first_parent: Option) -> Option { + let mut cursor = first_parent?; + for _ in 0..MAX_CHAIN_HOPS { + let node = chain.get(&cursor)?; + if !node.is_attachment { + return node.timestamp_ms; + } + cursor = node.parent?; + } + None +} + /// 同步 Claude Code 会话日志到使用统计数据库 pub fn sync_claude_session_logs(db: &Database) -> Result { let projects_dir = get_claude_config_dir().join("projects"); @@ -450,6 +611,8 @@ fn sync_single_file( // 一概不重放（见函数文档：重放会把已剪明细双算进汇总）。指纹为 NULL // （升级存量行转换后的首轮之前）时无从校验，按纯追加处理——这是旧 // 行号游标本就存在的暴露面，首轮写入后即有指纹。 + let mut chain = ChainNodes::new(); + let mut timings: HashMap = HashMap::new(); let (start_byte, legacy_lines, mut tail_buf) = match last_byte_offset { Some(offset) => { let truncated = !(0..=file_size).contains(&offset); @@ -485,6 +648,7 @@ fn sync_single_file( ..Default::default() }); } + read_chain_seed(&mut file, offset, &mut chain, &mut timings)?; (offset, 0, seed.unwrap_or_default()) } // 旧行号游标：从头按行转换（下方转换段），tail 缓冲从空积累 @@ -562,6 +726,9 @@ fn sync_single_file( } } + // 每一行都记进对话链，估算请求耗时时沿它找起点 + record_chain_node(&mut chain, &value); + // 只处理 assistant 类型的消息 if value.get("type").and_then(|t| t.as_str()) != Some("assistant") { continue; @@ -577,6 +744,8 @@ fn sync_single_file( None => continue, }; + record_block_timing(&mut timings, &value, &msg_id); + let usage = match message.get("usage") { Some(u) => u, None => continue, @@ -675,7 +844,17 @@ fn sync_single_file( msg.message_id ); - match upsert_session_log_entry_on_conn(&tx, &request_id, msg) { + // 耗时估算：只给写完整了的回复算（有 stop_reason），而且要看得到它的第一块 + // （这一批里，或游标前的回看窗口里），否则起点对不上 + let latency_ms = timings + .get(&msg.message_id) + .filter(|timing| msg.stop_reason.is_some() && timing.starts_at_first_block) + .and_then(|timing| { + let start_ms = resolve_request_start_ms(&chain, timing.first_parent)?; + estimated_latency_ms(start_ms, timing.end_ms?) + }); + + match upsert_session_log_entry_on_conn(&tx, &request_id, msg, latency_ms) { // 补全已有的行也算进 imported：数据变了，界面要跟着刷新 Ok(SessionRowOutcome::Inserted | SessionRowOutcome::Updated) => imported += 1, Ok(SessionRowOutcome::Skipped) => skipped += 1, @@ -816,16 +995,17 @@ pub(crate) fn update_sync_state_on_conn( #[derive(Debug, PartialEq, Eq)] enum SessionRowOutcome { Inserted, - /// 这条回复早先按中间块入过库，这次补全了用量；或者补全后发现它和 + /// 这条回复早先按中间块入过库，这次补全了用量或耗时；或者补全后发现它和 /// 路由服务记的是同一次请求，把先前那一行删了。 Updated, Skipped, } -/// 已经在库里的那一行里，判断要不要补全用得到的两样。 +/// 已经在库里的那一行里，判断要不要补全用得到的三样。 struct StoredSessionRow { data_source: String, output_tokens: i64, + latency_ms: i64, } fn find_stored_session_row( @@ -833,7 +1013,7 @@ fn find_stored_session_row( request_id: &str, ) -> Result, AppError> { conn.prepare_cached( - "SELECT COALESCE(data_source, 'proxy'), output_tokens + "SELECT COALESCE(data_source, 'proxy'), output_tokens, latency_ms FROM proxy_request_logs WHERE request_id = ?1", ) .and_then(|mut stmt| { @@ -841,6 +1021,7 @@ fn find_stored_session_row( Ok(StoredSessionRow { data_source: row.get(0)?, output_tokens: row.get(1)?, + latency_ms: row.get(2)?, }) }) .optional() @@ -877,7 +1058,7 @@ fn session_costs(conn: &rusqlite::Connection, msg: &ParsedAssistantUsage) -> [St /// /// 子代理的日志是逐块写的，前面的块只带当时的中间用量（输出 token 只有个位数）。 /// 同步恰好落在一条回复写到一半的时候，就会按中间值入库；最终块到了以后要把 -/// 输出 token 和成本补上，否则这条回复的输出永远少记。输入和缓存 token +/// 输出 token、成本和耗时补上，否则这条回复的输出永远少记。输入和缓存 token /// 在第一块里就是准的。 /// /// 只动会话日志导入的明细行。30 天前的明细已经汇总后删除，但一条回复的各块 @@ -886,12 +1067,27 @@ fn refresh_session_log_entry_on_conn( conn: &rusqlite::Connection, request_id: &str, msg: &ParsedAssistantUsage, + latency_ms: Option, stored: &StoredSessionRow, dedup_key: &DedupKey, ) -> Result { - if stored.data_source != "session_log" || i64::from(msg.output_tokens) <= stored.output_tokens { + if stored.data_source != "session_log" { return Ok(SessionRowOutcome::Skipped); } + let output_grew = i64::from(msg.output_tokens) > stored.output_tokens; + let latency_ms = latency_ms.map_or(stored.latency_ms, |ms| ms.max(stored.latency_ms)); + if !output_grew && latency_ms == stored.latency_ms { + return Ok(SessionRowOutcome::Skipped); + } + + if !output_grew { + conn.execute( + "UPDATE proxy_request_logs SET latency_ms = ?1 WHERE request_id = ?2", + rusqlite::params![latency_ms, request_id], + ) + .map_err(|e| AppError::Database(format!("更新会话日志耗时失败: {e}")))?; + return Ok(SessionRowOutcome::Updated); + } // 入库时用量还是中间值，对不上路由服务记的那一行；补全后对上了，说明这次 // 请求路由服务已经记过，先前入库的这一行是重复的 @@ -910,8 +1106,9 @@ fn refresh_session_log_entry_on_conn( "UPDATE proxy_request_logs SET input_tokens = ?1, output_tokens = ?2, cache_read_tokens = ?3, cache_creation_tokens = ?4, input_cost_usd = ?5, output_cost_usd = ?6, - cache_read_cost_usd = ?7, cache_creation_cost_usd = ?8, total_cost_usd = ?9 - WHERE request_id = ?10", + cache_read_cost_usd = ?7, cache_creation_cost_usd = ?8, total_cost_usd = ?9, + latency_ms = ?10 + WHERE request_id = ?11", rusqlite::params![ msg.input_tokens, msg.output_tokens, @@ -922,6 +1119,7 @@ fn refresh_session_log_entry_on_conn( cache_read_cost, cache_creation_cost, total_cost, + latency_ms, request_id, ], ) @@ -932,11 +1130,13 @@ fn refresh_session_log_entry_on_conn( /// 把一条回复写进 proxy_request_logs：没入过库就插入，入过库就按需补全 /// （见 [`refresh_session_log_entry_on_conn`]）。 /// -/// 调用方持有连接锁（通常在事务内）。 +/// 调用方持有连接锁（通常在事务内）。`latency_ms` 是按日志时间戳估出来的 +/// 请求耗时（含首字等待），估不出来传 None。 fn upsert_session_log_entry_on_conn( conn: &rusqlite::Connection, request_id: &str, msg: &ParsedAssistantUsage, + latency_ms: Option, ) -> Result { let created_at = msg .timestamp @@ -963,7 +1163,9 @@ fn upsert_session_log_entry_on_conn( created_at, }; if let Some(stored) = find_stored_session_row(conn, request_id)? { - return refresh_session_log_entry_on_conn(conn, request_id, msg, &stored, &dedup_key); + return refresh_session_log_entry_on_conn( + conn, request_id, msg, latency_ms, &stored, &dedup_key, + ); } if has_matching_proxy_usage_log(conn, &dedup_key)? { return Ok(SessionRowOutcome::Skipped); @@ -996,8 +1198,8 @@ fn upsert_session_log_entry_on_conn( cache_read_cost, cache_creation_cost, total_cost, - 0i64, // latency_ms: 会话日志无此数据 - Option::::None, // first_token_ms + latency_ms.unwrap_or(0), // latency_ms: 按时间戳估算，0 = 没有计时 + Option::::None, // first_token_ms: 会话日志无此数据 200i64, // status_code: 会话日志中的请求只要产生计费 token 即视为成功 Option::::None, // error_message msg.session_id, @@ -1214,7 +1416,7 @@ mod tests { let outcome = { let conn = lock_conn!(db.conn); - upsert_session_log_entry_on_conn(&conn, "session:msg_1", &msg)? + upsert_session_log_entry_on_conn(&conn, "session:msg_1", &msg, None)? }; assert_eq!(outcome, SessionRowOutcome::Skipped); @@ -1688,6 +1890,18 @@ mod tests { ) } + /// 一条回复里的一个内容块；同一条回复的各块共用 message.id 和最终用量 + /// （主会话的写法：整条回复结束时一次写完） + fn assistant_block( + msg_id: &str, + uuid: u8, + parent: u8, + block_index: u32, + timestamp: &str, + ) -> String { + assistant_block_with(msg_id, uuid, parent, block_index, timestamp, 900, true) + } + /// 子代理的写法是逐块写：非末块只带当时的输出 token，也没有 stop_reason fn assistant_block_with( msg_id: &str, @@ -1726,6 +1940,16 @@ mod tests { .unwrap() } + fn latency_of(db: &Database, msg_id: &str) -> i64 { + let conn = db.conn.lock().unwrap(); + conn.query_row( + "SELECT latency_ms FROM proxy_request_logs WHERE request_id = ?1", + [format!("session:{msg_id}")], + |row| row.get(0), + ) + .unwrap() + } + fn temp_session_file() -> (PathBuf, PathBuf) { let tmp = std::env::temp_dir().join(format!("cc-switch-test-{}", uuid::Uuid::new_v4())); fs::create_dir_all(&tmp).unwrap(); @@ -1733,6 +1957,90 @@ mod tests { (tmp, file) } + #[test] + fn test_latency_spans_from_tool_result_to_last_block() -> Result<(), AppError> { + let db = Database::memory()?; + let (tmp, file) = temp_session_file(); + let lines = [ + // 起点：工具结果 + chain_line("user", 1, None, "2026-06-07T13:00:00.000Z"), + chain_line("attachment", 2, Some(1), "2026-06-07T13:00:00.003Z"), + // 回复开始后才补记的 attachment，时间戳比第一块还晚，不能当起点 + chain_line("attachment", 3, Some(2), "2026-06-07T13:00:05.001Z"), + assistant_block("msg_t", 4, 3, 0, "2026-06-07T13:00:05.000Z"), + assistant_block("msg_t", 5, 4, 1, "2026-06-07T13:00:07.000Z"), + // 并行工具调用：前一个工具的结果插在同一条回复的两块中间 + chain_line("user", 6, Some(5), "2026-06-07T13:00:08.000Z"), + assistant_block("msg_t", 7, 6, 2, "2026-06-07T13:00:12.500Z"), + ]; + fs::write(&file, lines.join("\n") + "\n").unwrap(); + + let file_sync = sync_single_file(&db, &file, None)?; + assert_eq!(file_sync.imported, 1); + assert_eq!(latency_of(&db, "msg_t"), 12_500); + + fs::remove_dir_all(&tmp).ok(); + Ok(()) + } + + #[test] + fn test_latency_start_is_found_before_the_cursor() -> Result<(), AppError> { + // 工具结果先落盘、被上一轮同步读走；回复结束时才整条写出， + // 它的起点行在游标前面 + let db = Database::memory()?; + let (tmp, file) = temp_session_file(); + let before = [ + chain_line("user", 1, None, "2026-06-07T13:00:00.000Z"), + chain_line("attachment", 2, Some(1), "2026-06-07T13:00:00.003Z"), + ]; + fs::write(&file, before.join("\n") + "\n").unwrap(); + let first = sync_with_cursor(&db, &file)?; + assert_eq!(first.imported, 0); + + let mut content = fs::read(&file).unwrap(); + content.extend_from_slice( + format!( + "{}\n", + assistant_block("msg_t", 3, 2, 0, "2026-06-07T13:00:09.000Z") + ) + .as_bytes(), + ); + fs::write(&file, &content).unwrap(); + bump_mtime(&file); + let second = sync_with_cursor(&db, &file)?; + assert_eq!(second.imported, 1); + assert_eq!(latency_of(&db, "msg_t"), 9_000); + + fs::remove_dir_all(&tmp).ok(); + Ok(()) + } + + #[test] + fn test_latency_is_not_estimated_without_a_reliable_start() -> Result<(), AppError> { + let db = Database::memory()?; + let (tmp, file) = temp_session_file(); + let lines = [ + chain_line("user", 1, None, "2026-06-07T13:00:00.000Z"), + // 看不到第一块：最先出现的是第 2 块，它的父行是插进来的工具结果 + assistant_block("msg_no_first", 2, 1, 2, "2026-06-07T13:00:09.000Z"), + // 父行不在文件里 + assistant_block("msg_orphan", 3, 9, 0, "2026-06-07T13:00:20.000Z"), + // 起点和结束挨得太近，不像一次真实请求 + chain_line("user", 4, None, "2026-06-07T13:01:00.000Z"), + assistant_block("msg_instant", 5, 4, 0, "2026-06-07T13:01:00.020Z"), + ]; + fs::write(&file, lines.join("\n") + "\n").unwrap(); + + let file_sync = sync_single_file(&db, &file, None)?; + assert_eq!(file_sync.imported, 3); + assert_eq!(latency_of(&db, "msg_no_first"), 0); + assert_eq!(latency_of(&db, "msg_orphan"), 0); + assert_eq!(latency_of(&db, "msg_instant"), 0); + + fs::remove_dir_all(&tmp).ok(); + Ok(()) + } + #[test] fn test_partial_subagent_reply_is_completed_by_its_final_block() -> Result<(), AppError> { let first_block = [ @@ -1772,7 +2080,7 @@ mod tests { let completed = stored_row(&db, "msg_s"); assert_eq!(completed, stored_row(&whole_db, "msg_s")); - assert_eq!(completed.map(|row| row.0), Some(4_660)); + assert_eq!(completed.map(|row| (row.0, row.2)), Some((4_660, 30_000))); // 再同步一轮不会重复补 bump_mtime(&file); @@ -1784,6 +2092,37 @@ mod tests { Ok(()) } + #[test] + fn test_latency_is_extended_when_later_blocks_arrive() -> Result<(), AppError> { + // 主会话的各块都带最终用量；同步读在一次写入的中途时，先按前几块算了 + // 较短的耗时，后面的块到了要补上 + let db = Database::memory()?; + let (tmp, file) = temp_session_file(); + append_lines( + &file, + &[ + chain_line("user", 1, None, "2026-06-07T13:00:00.000Z"), + assistant_block("msg_t", 2, 1, 0, "2026-06-07T13:00:05.000Z"), + ], + ); + sync_with_cursor(&db, &file)?; + assert_eq!(latency_of(&db, "msg_t"), 5_000); + + append_lines( + &file, + &[ + chain_line("user", 3, Some(2), "2026-06-07T13:00:06.000Z"), + assistant_block("msg_t", 4, 3, 1, "2026-06-07T13:00:12.500Z"), + ], + ); + let second = sync_with_cursor(&db, &file)?; + assert_eq!(second.imported, 1); + assert_eq!(latency_of(&db, "msg_t"), 12_500); + + fs::remove_dir_all(&tmp).ok(); + Ok(()) + } + #[test] fn test_completed_reply_matching_a_proxy_log_is_removed() -> Result<(), AppError> { let db = Database::memory()?; @@ -1821,9 +2160,7 @@ mod tests { "0.10", 30_000, 200, - chrono::DateTime::parse_from_rfc3339("2026-06-07T13:00:30Z") - .unwrap() - .timestamp(), + parse_timestamp_millis("2026-06-07T13:00:30.000Z").unwrap() / 1000, "proxy" ], )?; diff --git a/src-tauri/src/services/session_usage_codex.rs b/src-tauri/src/services/session_usage_codex.rs index cf7f5c9b3..960c698c0 100644 --- a/src-tauri/src/services/session_usage_codex.rs +++ b/src-tauri/src/services/session_usage_codex.rs @@ -19,7 +19,8 @@ use crate::error::AppError; use crate::proxy::usage::calculator::{CostCalculator, ModelPricing}; use crate::proxy::usage::parser::TokenUsage; use crate::services::session_usage::{ - metadata_modified_nanos, update_sync_state, update_sync_state_on_conn, SessionSyncResult, + estimated_latency_ms, metadata_modified_nanos, parse_timestamp_millis, update_sync_state, + update_sync_state_on_conn, SessionSyncResult, }; use crate::services::usage_stats::{ find_model_pricing, has_suspected_codex_session_duplicate, should_skip_session_insert, DedupKey, @@ -197,6 +198,199 @@ struct ParsedTokenEvent { event_index: Option, model: String, timestamp: Option, + /// 按事件时间戳估出来的请求耗时（含首字等待）；估不出来为 None。 + latency_ms: Option, +} + +/// 只看每行开头这么多字节来判断行的种类：时间戳、类型、角色都在行头， +/// 后面的正文（工具输出可能上百 KB）不用解析。 +const LINE_HEAD_BYTES: usize = 512; + +fn find_bytes(haystack: &[u8], needle: &[u8]) -> Option { + haystack + .windows(needle.len()) + .position(|window| window == needle) +} + +/// 从行头里取 `"key":"value"` 的值。只用于时间戳、类型名、角色这类不含转义的短值。 +fn head_str<'a>(head: &'a [u8], key: &str) -> Option<&'a str> { + let needle = format!("\"{key}\":\""); + let start = find_bytes(head, needle.as_bytes())? + needle.len(); + let len = head[start..].iter().position(|b| *b == b'"')?; + std::str::from_utf8(&head[start..start + len]).ok() +} + +#[derive(Debug, PartialEq, Eq)] +enum TimingLine { + /// 下一次请求只会在它之后发出：用户消息、`turn_context` + Boundary, + /// 工具结果，同样是下一次请求的起点 + ToolOutput, + /// 模型输出的一项（思考、回复、工具调用），写在这一项生成完的时刻 + ModelOutput, + /// `token_usage_record`：响应结束时写的用量记录（新版 Codex 才有） + UsageRecord, + /// 一轮开始（`task_started`）或被中断（`turn_aborted`）：没等到 `token_count` + /// 的那次请求到此作废 + TurnReset, +} + +fn classify_timing_line(line: &str) -> Option<(i64, TimingLine)> { + let bytes = line.as_bytes(); + let head = &bytes[..bytes.len().min(LINE_HEAD_BYTES)]; + let (envelope, payload) = match find_bytes(head, b"\"payload\":{") { + Some(at) => head.split_at(at), + None => (head, &[][..]), + }; + let kind = match head_str(envelope, "type")? { + "turn_context" => TimingLine::Boundary, + "token_usage_record" => TimingLine::UsageRecord, + "response_item" => match head_str(payload, "type")? { + "message" => match head_str(payload, "role")? { + "assistant" => TimingLine::ModelOutput, + _ => TimingLine::Boundary, + }, + "reasoning" => TimingLine::ModelOutput, + item if item.ends_with("_output") => TimingLine::ToolOutput, + item if item.ends_with("_call") => TimingLine::ModelOutput, + _ => return None, + }, + "event_msg" => match head_str(payload, "type")? { + "task_started" | "turn_aborted" => TimingLine::TurnReset, + _ => return None, + }, + _ => return None, + }; + let timestamp_ms = head_str(envelope, "timestamp").and_then(parse_timestamp_millis)?; + Some((timestamp_ms, kind)) +} + +/// 按事件顺序估每次请求的耗时。Codex 日志没有请求级计时，而且各版本的 +/// 事件顺序不一样： +/// +/// - 结束时刻：新版在响应结束时写 `token_usage_record`，直接用它。旧版只有 +/// `token_count`，有的版本在响应结束时写，有的要等工具跑完才写；后一种 +/// 表现为「最后一个输出项之后先出现工具结果，`token_count` 紧跟着工具结果 +/// 写出」，这时改用最后一个输出项的时刻（略早于真正结束，速度会略偏高）。 +/// - 开始时刻：看到这次请求第一个输出项时，在它之前最近的一个起点行 +/// （上一次的 `token_count`、工具结果、用户消息、`turn_context`）。不直接取 +/// 结束前最近的起点行，是因为工具结果可能在响应还没结束时就写进来了。 +/// +/// 2025 年的旧布局把上一次响应的输出项补写在它的 `token_count` 之后（时间戳 +/// 和那个 `token_count` 一样）。这些输出项不属于下一次请求，直接忽略；否则 +/// 下一次请求的起点会被钉在上一次响应结束的时刻，用户隔很久才发下一条消息时 +/// 算出来的耗时就长得离谱。 +/// +/// 请求被中断或出错时等不到 `token_count`，它留下的起点要在下一轮开始时清掉， +/// 否则会被下一次请求沿用。只在 `task_started` / `turn_aborted` 上清，不在用户 +/// 消息上清：Codex 会在响应中途插入 developer 消息，在那里清会把起点挪晚。 +#[derive(Debug, Default)] +struct RequestTimer { + last_boundary_ms: Option, + last_token_count_ms: Option, + request_start_ms: Option, + last_model_output_ms: Option, + /// 最后一个输出项之后出现的工具结果里最晚的一个；之后再有输出项就清空。 + tool_output_after_model_output_ms: Option, + usage_record: Option<(i64, CumulativeTokens)>, +} + +/// 两行的时间戳相差不超过这个毫秒数，就算是同一批写出的（实测相差 0–1 毫秒， +/// 而一次真实的请求不可能这么快）。 +const SAME_FLUSH_SLACK_MS: i64 = 100; + +impl RequestTimer { + fn observe_line(&mut self, line: &str) { + let Some((timestamp_ms, kind)) = classify_timing_line(line) else { + return; + }; + match kind { + TimingLine::Boundary => { + self.last_boundary_ms = self.last_boundary_ms.max(Some(timestamp_ms)); + } + TimingLine::ToolOutput => { + self.last_boundary_ms = self.last_boundary_ms.max(Some(timestamp_ms)); + if self.last_model_output_ms.is_some() { + self.tool_output_after_model_output_ms = Some(timestamp_ms); + } + } + TimingLine::ModelOutput => { + let trails_token_count = self.request_start_ms.is_none() + && self + .last_token_count_ms + .is_some_and(|at| timestamp_ms - at <= SAME_FLUSH_SLACK_MS); + if trails_token_count { + return; + } + if self.request_start_ms.is_none() { + self.request_start_ms = self.last_boundary_ms; + } + self.last_model_output_ms = Some(timestamp_ms); + self.tool_output_after_model_output_ms = None; + } + TimingLine::UsageRecord => { + // 行头只够判断种类，用量要解析整行；这类行很短 + let usage = serde_json::from_str::(line) + .ok() + .and_then(|value| { + value + .get("payload") + .and_then(|payload| payload.get("usage")) + .and_then(parse_cumulative_tokens) + }); + self.usage_record = usage.map(|usage| (timestamp_ms, usage)); + } + TimingLine::TurnReset => { + *self = RequestTimer { + last_boundary_ms: self.last_boundary_ms.max(Some(timestamp_ms)), + last_token_count_ms: self.last_token_count_ms, + ..RequestTimer::default() + }; + } + } + } + + /// 遇到一次有用量的 `token_count`：结算这次请求的耗时，并把它记成下一次 + /// 请求的起点。`last` 是这次请求自己的用量，用来确认 `token_usage_record` + /// 说的是同一次请求。 + fn finish_request( + &mut self, + token_count_ms: Option, + last: Option<&CumulativeTokens>, + ) -> Option { + let start_ms = self.request_start_ms.or(self.last_boundary_ms); + let record_end_ms = self + .usage_record + .take() + .filter(|(_, usage)| { + last.is_none_or(|last| { + usage.input == last.input + && usage.cached_input == last.cached_input + && usage.output == last.output + }) + }) + .map(|(timestamp_ms, _)| timestamp_ms); + let token_count_waited_for_tools = + match (self.tool_output_after_model_output_ms, token_count_ms) { + (Some(tool_output_ms), Some(token_count_ms)) => { + token_count_ms - tool_output_ms <= SAME_FLUSH_SLACK_MS + } + (Some(_), None) => true, + (None, _) => false, + }; + let end_ms = record_end_ms.or(if token_count_waited_for_tools { + self.last_model_output_ms + } else { + token_count_ms + }); + + *self = RequestTimer { + last_boundary_ms: self.last_boundary_ms.max(token_count_ms), + last_token_count_ms: token_count_ms, + ..RequestTimer::default() + }; + estimated_latency_ms(start_ms?, end_ms?) + } } #[derive(Debug)] @@ -818,6 +1012,7 @@ fn parse_codex_file( let mut line_offset = 0i64; let mut observed_bytes = 0i64; let mut has_billable_tokens = false; + let mut timer = RequestTimer::default(); loop { let mut bytes = Vec::new(); @@ -845,6 +1040,8 @@ fn parse_codex_file( continue; } + timer.observe_line(&line); + let is_event_msg = line.contains("\"event_msg\""); let is_turn_context = line.contains("\"turn_context\""); let is_session_meta = line.contains("\"session_meta\""); @@ -964,6 +1161,7 @@ fn parse_codex_file( } previous_token_signature = Some(signature.clone()); + let request_usage = last.clone(); let delta = if duplicate_snapshot { DeltaTokens { input: 0, @@ -1003,16 +1201,26 @@ fn parse_codex_file( Some(event_index) }; + let timestamp = value + .get("timestamp") + .and_then(serde_json::Value::as_str) + .map(str::to_owned); + // 重复快照（限额刷新时重发的）不是一次请求，不参与计时 + let latency_ms = nonzero_index.and_then(|_| { + timer.finish_request( + timestamp.as_deref().and_then(parse_timestamp_millis), + request_usage.as_ref(), + ) + }); + token_events.push(ParsedTokenEvent { line_offset, signature, delta, event_index: nonzero_index, model: current_model.clone(), - timestamp: value - .get("timestamp") - .and_then(serde_json::Value::as_str) - .map(str::to_owned), + timestamp, + latency_ms, }); } _ => {} @@ -1418,6 +1626,7 @@ fn sync_single_codex_file( &event.model, Some(session_thread_id), event.timestamp.as_deref(), + event.latency_ms, &mut batch_suspected, &mut pass.pricing, ) { @@ -1468,6 +1677,7 @@ fn insert_codex_session_entry( model, session_id, timestamp, + None, suspected_duplicates, &mut HashMap::new(), ) @@ -1477,7 +1687,8 @@ fn insert_codex_session_entry( /// /// 调用方负责持锁/事务；`pricing_cache` 按原始 model 字符串键控（ /// `find_codex_pricing` 是纯函数式查找，同串必同结果），全量重导时把 -/// 每事件一次的定价 SELECT 降为每模型一次。 +/// 每事件一次的定价 SELECT 降为每模型一次。`latency_ms` 是按事件时间戳估出来的 +/// 请求耗时（见 [`RequestTimer`]），估不出来传 None。 #[allow(clippy::too_many_arguments)] fn insert_codex_session_entry_on_conn( conn: &rusqlite::Connection, @@ -1486,6 +1697,7 @@ fn insert_codex_session_entry_on_conn( model: &str, session_id: Option<&str>, timestamp: Option<&str>, + latency_ms: Option, suspected_duplicates: &mut u32, pricing_cache: &mut HashMap>, ) -> Result { @@ -1584,8 +1796,8 @@ fn insert_codex_session_entry_on_conn( cache_read_cost, cache_creation_cost, total_cost, - 0i64, // latency_ms - Option::::None, // first_token_ms + latency_ms.unwrap_or(0), // latency_ms: 按时间戳估算，0 = 没有计时 + Option::::None, // first_token_ms: 会话日志无此数据 200i64, // status_code Option::::None, // error_message session_id.map(|s| s.to_string()), @@ -3451,4 +3663,214 @@ mod tests { } Ok(()) } + + /// 按 Codex 实际写出的键顺序（timestamp、type、payload）拼一行；`json!` 会把键 + /// 按字母排序，行头判断种类靠的是真实顺序 + fn raw_line(timestamp: &str, kind: &str, payload: &str) -> String { + format!(r#"{{"timestamp":"{timestamp}","type":"{kind}","payload":{payload}}}"#) + } + + fn raw_item(timestamp: &str, payload: &str) -> String { + raw_line(timestamp, "response_item", payload) + } + + fn raw_usage(input: u64, output: u64) -> String { + format!( + r#"{{"input_tokens":{input},"cached_input_tokens":0,"output_tokens":{output},"reasoning_output_tokens":0,"total_tokens":{}}}"#, + input + output + ) + } + + /// `total` 是累计用量，`last` 是这一次请求的用量 + fn raw_token_count(timestamp: &str, total: (u64, u64), last: (u64, u64)) -> String { + raw_line( + timestamp, + "event_msg", + &format!( + r#"{{"type":"token_count","info":{{"total_token_usage":{},"last_token_usage":{}}},"rate_limits":{{"limit_id":"codex"}}}}"#, + raw_usage(total.0, total.1), + raw_usage(last.0, last.1) + ), + ) + } + + fn parsed_latencies(lines: &[String]) -> Vec> { + let dir = tempdir().unwrap(); + let file = rollout_path(dir.path(), PARENT_ID); + fs::write(&file, lines.join("\n") + "\n").unwrap(); + parse_codex_file(&file, Some(PARENT_ID.to_string())) + .unwrap() + .token_events + .iter() + .filter(|event| event.event_index.is_some()) + .map(|event| event.latency_ms) + .collect() + } + + const USER_MESSAGE: &str = r#"{"type":"message","role":"user","content":[]}"#; + const ASSISTANT_MESSAGE: &str = r#"{"type":"message","role":"assistant","content":[]}"#; + const REASONING: &str = r#"{"type":"reasoning","summary":[]}"#; + const TOOL_CALL: &str = r#"{"type":"custom_tool_call","call_id":"c1","name":"shell"}"#; + const TOOL_OUTPUT: &str = r#"{"type":"custom_tool_call_output","call_id":"c1","output":"ok"}"#; + + #[test] + fn test_latency_ends_at_usage_record_not_at_delayed_token_count() -> Result<(), AppError> { + let lines = vec![ + raw_line( + "2026-10-02T03:00:00.000Z", + "turn_context", + r#"{"model":"gpt-5"}"#, + ), + raw_item("2026-10-02T03:00:01.000Z", USER_MESSAGE), + raw_item("2026-10-02T03:00:05.000Z", REASONING), + raw_item("2026-10-02T03:00:09.000Z", TOOL_CALL), + // 响应在这里结束；token_count 要等工具跑完才写 + raw_line( + "2026-10-02T03:00:09.500Z", + "token_usage_record", + &format!(r#"{{"turn_id":"t1","usage":{}}}"#, raw_usage(100, 600)), + ), + raw_item("2026-10-02T03:00:20.000Z", TOOL_OUTPUT), + raw_token_count("2026-10-02T03:00:20.001Z", (100, 600), (100, 600)), + // 第二次请求：用量记录对不上这次请求，不采信，退回 token_count 的时刻 + raw_item("2026-10-02T03:00:25.000Z", REASONING), + raw_item("2026-10-02T03:00:30.000Z", ASSISTANT_MESSAGE), + raw_line( + "2026-10-02T03:00:30.200Z", + "token_usage_record", + &format!(r#"{{"turn_id":"other","usage":{}}}"#, raw_usage(1, 999)), + ), + raw_token_count("2026-10-02T03:00:30.300Z", (300, 1300), (200, 700)), + ]; + + assert_eq!(parsed_latencies(&lines), vec![Some(8_500), Some(10_299)]); + + // 估出来的耗时要落进库里 + let dir = tempdir().unwrap(); + let file = rollout_path(dir.path(), PARENT_ID); + let mut contents = session_meta(PARENT_ID).to_string(); + contents.push('\n'); + contents.push_str(&(lines.join("\n") + "\n")); + fs::write(&file, contents).unwrap(); + let db = Database::memory()?; + sync_test_file(&db, &file, &[&file])?; + let conn = lock_conn!(db.conn); + let mut stmt = conn + .prepare("SELECT latency_ms FROM proxy_request_logs ORDER BY request_id") + .unwrap(); + let stored: Vec = stmt + .query_map([], |row| row.get(0)) + .unwrap() + .map(|row| row.unwrap()) + .collect(); + assert_eq!(stored, vec![8_500, 10_299]); + Ok(()) + } + + #[test] + fn test_latency_without_usage_record_stops_before_tool_execution() { + let lines = vec![ + raw_item("2026-08-03T03:00:01.000Z", USER_MESSAGE), + raw_item("2026-08-03T03:00:06.000Z", TOOL_CALL), + // 工具跑了 10 秒，token_count 紧跟着工具结果写出 + raw_item("2026-08-03T03:00:16.000Z", TOOL_OUTPUT), + raw_token_count("2026-08-03T03:00:16.001Z", (100, 600), (100, 600)), + // 限额刷新时重发的快照不是一次请求，不能把下一次请求的起点往后挪 + raw_token_count("2026-08-03T03:00:17.000Z", (100, 600), (100, 600)), + raw_item("2026-08-03T03:00:20.000Z", REASONING), + raw_item("2026-08-03T03:00:22.000Z", ASSISTANT_MESSAGE), + raw_token_count("2026-08-03T03:00:22.100Z", (300, 1300), (200, 700)), + ]; + + assert_eq!(parsed_latencies(&lines), vec![Some(5_000), Some(6_099)]); + } + + #[test] + fn test_latency_ignores_items_written_after_their_token_count() { + // 2025 年的布局：输出项和工具结果补写在它们那次响应的 token_count 之后 + let lines = vec![ + raw_item("2025-10-30T08:00:01.000Z", USER_MESSAGE), + raw_token_count("2025-10-30T08:00:08.000Z", (100, 600), (100, 600)), + raw_item("2025-10-30T08:00:08.000Z", REASONING), + raw_item("2025-10-30T08:00:08.001Z", TOOL_CALL), + raw_item("2025-10-30T08:00:08.001Z", TOOL_OUTPUT), + raw_line( + "2025-10-30T08:00:08.002Z", + "turn_context", + r#"{"model":"gpt-5"}"#, + ), + // 用户隔了一分多钟才发下一条消息，这段空闲不算进请求耗时 + raw_item("2025-10-30T08:01:40.000Z", USER_MESSAGE), + raw_token_count("2025-10-30T08:01:50.000Z", (300, 1300), (200, 700)), + ]; + + assert_eq!(parsed_latencies(&lines), vec![Some(7_000), Some(10_000)]); + } + + #[test] + fn test_timing_line_is_classified_from_the_line_head() { + let long_output = format!( + r#"{{"type":"function_call_output","call_id":"c1","output":"{}"}}"#, + "x".repeat(4 * LINE_HEAD_BYTES) + ); + assert_eq!( + classify_timing_line(&raw_item("2026-10-02T03:00:00.250Z", &long_output)), + Some((1_790_910_000_250, TimingLine::ToolOutput)) + ); + // 正文里出现的字样不影响判断：种类只看信封上的 type + let quoted = + r#"{"type":"message","role":"user","content":"\"type\":\"token_usage_record\""}"#; + assert_eq!( + classify_timing_line(&raw_item("2026-10-02T03:00:00.250Z", quoted)).map(|line| line.1), + Some(TimingLine::Boundary) + ); + assert_eq!( + classify_timing_line(&raw_line( + "2026-10-02T03:00:00.250Z", + "event_msg", + r#"{"type":"task_started"}"# + )) + .map(|line| line.1), + Some(TimingLine::TurnReset) + ); + assert_eq!( + classify_timing_line(&raw_line( + "2026-10-02T03:00:00.250Z", + "event_msg", + r#"{"type":"item_completed"}"# + )), + None + ); + } + + #[test] + fn test_latency_does_not_inherit_the_start_of_an_aborted_request() { + const DEVELOPER_MESSAGE: &str = r#"{"type":"message","role":"developer","content":[]}"#; + let lines = vec![ + raw_item("2026-10-02T03:00:01.000Z", USER_MESSAGE), + // 这次请求出了一段思考就被用户中断，等不到 token_count + raw_item("2026-10-02T03:00:10.000Z", REASONING), + raw_line( + "2026-10-02T03:00:20.000Z", + "event_msg", + r#"{"type":"turn_aborted","reason":"interrupted"}"#, + ), + raw_line( + "2026-10-02T03:02:00.000Z", + "event_msg", + r#"{"type":"task_started"}"#, + ), + raw_item("2026-10-02T03:02:00.010Z", USER_MESSAGE), + raw_item("2026-10-02T03:02:20.000Z", REASONING), + raw_item("2026-10-02T03:02:30.000Z", ASSISTANT_MESSAGE), + raw_token_count("2026-10-02T03:02:30.100Z", (100, 600), (100, 600)), + // 响应中途插进来的 developer 消息不是新一轮，起点不动 + raw_item("2026-10-02T03:02:35.000Z", REASONING), + raw_item("2026-10-02T03:02:36.000Z", DEVELOPER_MESSAGE), + raw_item("2026-10-02T03:02:40.000Z", ASSISTANT_MESSAGE), + raw_token_count("2026-10-02T03:02:40.200Z", (300, 1300), (200, 700)), + ]; + + assert_eq!(parsed_latencies(&lines), vec![Some(30_090), Some(10_100)]); + } } diff --git a/src-tauri/src/services/usage_stats.rs b/src-tauri/src/services/usage_stats.rs index 977eb1f00..4fe6f2086 100644 --- a/src-tauri/src/services/usage_stats.rs +++ b/src-tauri/src/services/usage_stats.rs @@ -92,6 +92,11 @@ pub struct ProviderStats { pub speed_output_tokens: u64, /// 速度的分母：同一批请求的 (latency_ms - first_token_ms) 之和，单位毫秒。 pub speed_generation_ms: u64, + /// 估算速度的分子：会话日志导入的请求里，有估算耗时、输出 >= 200 token 的那些的输出之和。 + /// 和上面那组分开累计：估算的耗时含首字等待，口径不同，不能加在一起。 + pub est_speed_output_tokens: u64, + /// 估算速度的分母：同一批请求的 latency_ms 之和，单位毫秒。 + pub est_speed_duration_ms: u64, } /// 计速度的门槛：输出少于这个数的请求（工具调用这类）不算速度，避免 0.1 秒回 15 个 token 算出离谱的数。 @@ -109,6 +114,25 @@ fn speed_eligible_sql(alias: &str) -> String { ) } +/// 估算速度的输出门槛：估算的耗时含首字等待，输出越少首字占比越大、算出来越偏低， +/// 所以比精确口径的门槛高。实测 200–300 token 的请求比长请求低约四分之一， +/// 100–200 的低约四成；而 Codex 的请求只有四分之一超过 500，门槛再高大半行都是空的。 +pub const SPEED_ESTIMATE_MIN_OUTPUT_TOKENS: i64 = 200; + +/// 估算耗时短于这个毫秒数时不估速度：输出 200 token 以上却不到 1 秒，多半是起点取晚了。 +pub const SPEED_ESTIMATE_MIN_DURATION_MS: i64 = 1000; + +/// 明细行能不能估速度的 SQL 条件（和前端 `isSpeedEstimateEligible` 同口径）： +/// 会话日志导入的行（没有首字计时），耗时是导入时按日志时间戳估的。 +fn speed_estimate_eligible_sql(alias: &str) -> String { + let data_source = data_source_expr(alias); + format!( + "{alias}.first_token_ms IS NULL AND {data_source} <> 'proxy' \ + AND {alias}.output_tokens >= {SPEED_ESTIMATE_MIN_OUTPUT_TOKENS} \ + AND {alias}.latency_ms >= {SPEED_ESTIMATE_MIN_DURATION_MS}" + ) +} + /// 模型统计 #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -1355,6 +1379,7 @@ impl Database { let fresh_input_detail = fresh_input_sql("l"); let fresh_input_rollup = fresh_input_sql("r"); let speed_ok = speed_eligible_sql("l"); + let est_ok = speed_estimate_eligible_sql("l"); let sql = format!( "SELECT provider_id, app_type, provider_name, @@ -1366,7 +1391,9 @@ impl Database { THEN SUM(latency_sum) / SUM(request_count) ELSE 0 END as avg_latency, SUM(speed_output) as speed_output, - SUM(speed_gen_ms) as speed_gen_ms + SUM(speed_gen_ms) as speed_gen_ms, + SUM(est_output) as est_output, + SUM(est_ms) as est_ms FROM ( SELECT l.provider_id, l.app_type, {detail_pname} as provider_name, @@ -1376,7 +1403,9 @@ impl Database { COALESCE(SUM(CASE WHEN l.status_code >= 200 AND l.status_code < 300 THEN 1 ELSE 0 END), 0) as success_count, COALESCE(SUM(l.latency_ms), 0) as latency_sum, COALESCE(SUM(CASE WHEN {speed_ok} THEN l.output_tokens ELSE 0 END), 0) as speed_output, - COALESCE(SUM(CASE WHEN {speed_ok} THEN l.latency_ms - l.first_token_ms ELSE 0 END), 0) as speed_gen_ms + COALESCE(SUM(CASE WHEN {speed_ok} THEN l.latency_ms - l.first_token_ms ELSE 0 END), 0) as speed_gen_ms, + COALESCE(SUM(CASE WHEN {est_ok} THEN l.output_tokens ELSE 0 END), 0) as est_output, + COALESCE(SUM(CASE WHEN {est_ok} THEN l.latency_ms ELSE 0 END), 0) as est_ms FROM proxy_request_logs l LEFT JOIN providers p ON l.provider_id = p.id AND l.app_type = p.app_type {detail_where} @@ -1390,6 +1419,8 @@ impl Database { COALESCE(SUM(r.success_count), 0), COALESCE(SUM(r.avg_latency_ms * r.request_count), 0), 0, + 0, + 0, 0 FROM usage_daily_rollups r LEFT JOIN providers p2 ON r.provider_id = p2.id AND r.app_type = p2.app_type @@ -1423,6 +1454,8 @@ impl Database { avg_latency_ms: row.get::<_, f64>(7)? as u64, speed_output_tokens: row.get::<_, i64>(8)?.max(0) as u64, speed_generation_ms: row.get::<_, i64>(9)?.max(0) as u64, + est_speed_output_tokens: row.get::<_, i64>(10)?.max(0) as u64, + est_speed_duration_ms: row.get::<_, i64>(11)?.max(0) as u64, }) }; @@ -4020,6 +4053,50 @@ mod tests { assert_eq!(stats[0].request_count, 6); assert_eq!(stats[0].speed_output_tokens, 1_300); assert_eq!(stats[0].speed_generation_ms, 13_000); + // 路由服务的行没有首字（非流式）也不算估算速度 + assert_eq!(stats[0].est_speed_output_tokens, 0); + assert_eq!(stats[0].est_speed_duration_ms, 0); + + Ok(()) + } + + #[test] + fn test_get_provider_stats_estimated_speed_sums_session_rows() -> Result<(), AppError> { + let db = Database::memory()?; + + { + let conn = lock_conn!(db.conn); + let insert = |id: &str, output: i64, latency: i64, source: &str| { + conn.execute( + "INSERT INTO proxy_request_logs ( + request_id, provider_id, app_type, model, + input_tokens, output_tokens, total_cost_usd, + latency_ms, status_code, created_at, data_source + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + params![ + id, "_session", "claude", "m", 10, output, "0", latency, 200, 1000, source + ], + ) + }; + // 计入：2000 token / 20000 ms + insert("ok-a", 2_000, 20_000, "session_log")?; + // 计入：500 token / 5000 ms + insert("ok-b", 200, 5_000, "session_log")?; + // 不计：输出不到 200 + insert("short", 199, 5_000, "session_log")?; + // 不计：没估出耗时 + insert("no-timing", 3_000, 0, "session_log")?; + // 不计：耗时不到 1 秒 + insert("too-fast", 800, 900, "session_log")?; + } + + let stats = db.get_provider_stats(None, None, None, None, None)?; + assert_eq!(stats.len(), 1); + assert_eq!(stats[0].est_speed_output_tokens, 2_200); + assert_eq!(stats[0].est_speed_duration_ms, 25_000); + // 估算的不混进精确口径 + assert_eq!(stats[0].speed_output_tokens, 0); + assert_eq!(stats[0].speed_generation_ms, 0); Ok(()) }