From c9ac6efd69d33e0c246dd19298976c68ef41296b Mon Sep 17 00:00:00 2001 From: Jason Date: Mon, 20 Jul 2026 12:16:57 +0800 Subject: [PATCH] fix(proxy): add stable usage keys and idempotent raw-response logging Derive request ids from upstream envelope ids (Codex/OpenAI top-level id, Gemini responseId, Claude message id with non-empty filtering) scoped as session:{app_type}:{provider_id}:{id} for non-Claude sources; Claude keeps bare session:{id} to preserve session-log convergence. The logger now queries and conditionally writes under a single connection guard: identical semantic replays return without writing or notifying, session_log rows may be upgraded by proxy, and same-id different-semantic responses land on a deterministic SHA-256 collision fallback key instead of being overwritten. Fixes the random-UUID INSERT OR REPLACE duplication behind #5496. --- src-tauri/src/proxy/handlers.rs | 3 +- src-tauri/src/proxy/providers/transform.rs | 2 +- src-tauri/src/proxy/response_processor.rs | 3 +- src-tauri/src/proxy/usage/logger.rs | 323 ++++++++++++++++++--- src-tauri/src/proxy/usage/parser.rs | 108 +++++-- 5 files changed, 385 insertions(+), 54 deletions(-) diff --git a/src-tauri/src/proxy/handlers.rs b/src-tauri/src/proxy/handlers.rs index fe83a34d6..af1620eb8 100644 --- a/src-tauri/src/proxy/handlers.rs +++ b/src-tauri/src/proxy/handlers.rs @@ -2446,7 +2446,8 @@ async fn log_usage( model }; - let request_id = usage.dedup_request_id(); + let dedup_scope = (app_type != "claude").then_some((app_type, provider_id)); + let request_id = usage.dedup_request_id(dedup_scope); if let Err(e) = logger.log_with_calculation( request_id, diff --git a/src-tauri/src/proxy/providers/transform.rs b/src-tauri/src/proxy/providers/transform.rs index fd5aac801..4b948c375 100644 --- a/src-tauri/src/proxy/providers/transform.rs +++ b/src-tauri/src/proxy/providers/transform.rs @@ -1181,7 +1181,7 @@ mod tests { Some("chatcmpl-claude-compatible") ); assert_eq!( - usage.dedup_request_id(), + usage.dedup_request_id(None), "session:chatcmpl-claude-compatible" ); } diff --git a/src-tauri/src/proxy/response_processor.rs b/src-tauri/src/proxy/response_processor.rs index c63ae704b..9ab57fe37 100644 --- a/src-tauri/src/proxy/response_processor.rs +++ b/src-tauri/src/proxy/response_processor.rs @@ -642,7 +642,8 @@ async fn log_usage_internal( model }; - let request_id = usage.dedup_request_id(); + let dedup_scope = (app_type != "claude").then_some((app_type, provider_id)); + let request_id = usage.dedup_request_id(dedup_scope); log::debug!( "[{app_type}] 记录请求日志: id={request_id}, provider={provider_id}, model={model}, streaming={is_streaming}, status={status_code}, latency_ms={latency_ms}, first_token_ms={first_token_ms:?}, session={}, input={}, output={}, cache_read={}, cache_creation={}", diff --git a/src-tauri/src/proxy/usage/logger.rs b/src-tauri/src/proxy/usage/logger.rs index 549d81cf4..5baf3f8c8 100644 --- a/src-tauri/src/proxy/usage/logger.rs +++ b/src-tauri/src/proxy/usage/logger.rs @@ -6,9 +6,59 @@ use crate::database::{Database, PRICING_SOURCE_REQUEST, PRICING_SOURCE_RESPONSE} use crate::error::AppError; use crate::services::sql_helpers::{INPUT_TOKEN_SEMANTICS_FRESH, INPUT_TOKEN_SEMANTICS_TOTAL}; use crate::services::usage_stats::{find_model_pricing_row, is_placeholder_pricing_model}; +use rusqlite::OptionalExtension; use rust_decimal::Decimal; +use sha2::{Digest, Sha256}; use std::str::FromStr; +#[derive(Debug, PartialEq, Eq)] +struct UsageSemantic { + app_type: String, + provider_id: String, + model: String, + input_token_semantics: i64, + input_tokens: u32, + output_tokens: u32, + cache_read_tokens: u32, + cache_creation_tokens: u32, + status_code: u16, +} + +impl UsageSemantic { + fn from_log(log: &RequestLog, input_token_semantics: i64) -> Self { + Self { + app_type: log.app_type.clone(), + provider_id: log.provider_id.clone(), + model: log.model.clone(), + input_token_semantics, + input_tokens: log.usage.input_tokens, + output_tokens: log.usage.output_tokens, + cache_read_tokens: log.usage.cache_read_tokens, + cache_creation_tokens: log.usage.cache_creation_tokens, + status_code: log.status_code, + } + } + + fn sha256(&self) -> String { + let encoded = serde_json::to_vec(&( + &self.app_type, + &self.provider_id, + &self.model, + self.input_token_semantics, + self.input_tokens, + self.output_tokens, + self.cache_read_tokens, + self.cache_creation_tokens, + self.status_code, + )) + .expect("usage semantic tuple is serializable"); + Sha256::digest(encoded) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() + } +} + /// 请求日志 #[derive(Debug, Clone)] pub struct RequestLog { @@ -77,52 +127,132 @@ impl<'a> UsageLogger<'a> { } else { INPUT_TOKEN_SEMANTICS_FRESH }; + let semantic = UsageSemantic::from_log(log, input_token_semantics); + let existing = Self::load_existing_semantic(&conn, &log.request_id)?; - conn.execute( - "INSERT OR REPLACE INTO proxy_request_logs ( + let (request_id, replace_session_log, collision) = match existing { + None => (log.request_id.clone(), false, false), + Some((data_source, _existing_semantic)) + if data_source.as_deref() == Some("session_log") => + { + (log.request_id.clone(), true, false) + } + Some((data_source, existing_semantic)) + if data_source.as_deref().unwrap_or("proxy") == "proxy" + && existing_semantic == semantic => + { + return Ok(()); + } + Some(_) => { + let fallback = format!("{}:collision:{}", log.request_id, semantic.sha256()); + if let Some((data_source, existing_semantic)) = + Self::load_existing_semantic(&conn, &fallback)? + { + if data_source.as_deref().unwrap_or("proxy") == "proxy" + && existing_semantic == semantic + { + return Ok(()); + } + return Err(AppError::Database(format!( + "usage collision fallback 主键发生 SHA-256 冲突: {fallback}" + ))); + } + (fallback, false, true) + } + }; + + let insert_verb = if replace_session_log { + "INSERT OR REPLACE" + } else { + "INSERT OR IGNORE" + }; + let sql = format!( + "{insert_verb} INTO proxy_request_logs ( request_id, provider_id, app_type, model, request_model, pricing_model, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, input_token_semantics, input_cost_usd, output_cost_usd, cache_read_cost_usd, cache_creation_cost_usd, total_cost_usd, latency_ms, first_token_ms, status_code, error_message, session_id, provider_type, is_streaming, cost_multiplier, created_at - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25)", - rusqlite::params![ - log.request_id, - log.provider_id, - log.app_type, - log.model, - log.request_model, - log.pricing_model, - log.usage.input_tokens, - log.usage.output_tokens, - log.usage.cache_read_tokens, - log.usage.cache_creation_tokens, - input_token_semantics, - input_cost, - output_cost, - cache_read_cost, - cache_creation_cost, - total_cost, - log.latency_ms as i64, - log.first_token_ms.map(|v| v as i64), - log.status_code as i64, - log.error_message, - log.session_id, - log.provider_type, - log.is_streaming as i64, - log.cost_multiplier, - created_at, - ], - ) - .map_err(|e| AppError::Database(format!("记录请求日志失败: {e}")))?; + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25)" + ); + let affected_rows = conn + .execute( + &sql, + rusqlite::params![ + request_id, + log.provider_id, + log.app_type, + log.model, + log.request_model, + log.pricing_model, + log.usage.input_tokens, + log.usage.output_tokens, + log.usage.cache_read_tokens, + log.usage.cache_creation_tokens, + input_token_semantics, + input_cost, + output_cost, + cache_read_cost, + cache_creation_cost, + total_cost, + log.latency_ms as i64, + log.first_token_ms.map(|v| v as i64), + log.status_code as i64, + log.error_message, + log.session_id, + log.provider_type, + log.is_streaming as i64, + log.cost_multiplier, + created_at, + ], + ) + .map_err(|e| AppError::Database(format!("记录请求日志失败: {e}")))?; - // 通知前端使用统计有更新（200ms 防抖合并，不阻塞写入路径） - crate::usage_events::notify_log_recorded(); + if affected_rows > 0 { + if collision { + log::warn!( + "usage request_id collision: primary={}, fallback={request_id}", + log.request_id + ); + } + crate::usage_events::notify_log_recorded(); + } Ok(()) } + fn load_existing_semantic( + conn: &rusqlite::Connection, + request_id: &str, + ) -> Result, UsageSemantic)>, AppError> { + conn.query_row( + "SELECT data_source, app_type, provider_id, model, input_token_semantics, + input_tokens, output_tokens, cache_read_tokens, + cache_creation_tokens, status_code + FROM proxy_request_logs WHERE request_id = ?1", + [request_id], + |row| { + Ok(( + row.get(0)?, + UsageSemantic { + app_type: row.get(1)?, + provider_id: row.get(2)?, + model: row.get(3)?, + input_token_semantics: row.get(4)?, + input_tokens: row.get::<_, i64>(5)? as u32, + output_tokens: row.get::<_, i64>(6)? as u32, + cache_read_tokens: row.get::<_, i64>(7)? as u32, + cache_creation_tokens: row.get::<_, i64>(8)? as u32, + status_code: row.get::<_, i64>(9)? as u16, + }, + )) + }, + ) + .optional() + .map_err(|error| AppError::Database(format!("查询 usage request_id 失败: {error}"))) + } + /// 记录失败的请求 /// /// 用于记录无法从上游获取 usage 信息的失败请求 @@ -375,6 +505,34 @@ impl<'a> UsageLogger<'a> { mod tests { use super::*; + fn request_log(request_id: &str, input_tokens: u32) -> RequestLog { + RequestLog { + request_id: request_id.to_string(), + provider_id: "provider-1".to_string(), + app_type: "codex".to_string(), + model: "gpt-5.6".to_string(), + request_model: "gpt-5.6".to_string(), + pricing_model: "gpt-5.6".to_string(), + usage: TokenUsage { + input_tokens, + output_tokens: 5, + cache_read_tokens: 2, + cache_creation_tokens: 0, + model: None, + message_id: Some("resp-1".to_string()), + }, + cost: None, + latency_ms: 10, + first_token_ms: Some(2), + status_code: 200, + error_message: None, + session_id: None, + provider_type: Some("codex".to_string()), + is_streaming: true, + cost_multiplier: "1".to_string(), + } + } + #[test] fn test_log_request() -> Result<(), AppError> { let db = Database::memory()?; @@ -432,6 +590,103 @@ mod tests { Ok(()) } + #[test] + fn identical_replay_writes_and_notifies_once() -> Result<(), AppError> { + let db = Database::memory()?; + let logger = UsageLogger::new(&db); + crate::usage_events::take_test_notify_count(); + let log = request_log("stable-id", 10); + + logger.log_request(&log)?; + logger.log_request(&log)?; + + let conn = crate::database::lock_conn!(db.conn); + let count: i64 = conn.query_row( + "SELECT COUNT(*) FROM proxy_request_logs WHERE request_id = 'stable-id'", + [], + |row| row.get(0), + )?; + assert_eq!(count, 1); + assert_eq!(crate::usage_events::take_test_notify_count(), 1); + Ok(()) + } + + #[test] + fn semantic_collision_uses_deterministic_idempotent_fallback() -> Result<(), AppError> { + let db = Database::memory()?; + let logger = UsageLogger::new(&db); + crate::usage_events::take_test_notify_count(); + let first = request_log("shared-id", 10); + let second = request_log("shared-id", 20); + + logger.log_request(&first)?; + logger.log_request(&second)?; + logger.log_request(&second)?; + + let conn = crate::database::lock_conn!(db.conn); + let rows: Vec<(String, i64)> = conn + .prepare( + "SELECT request_id, input_tokens FROM proxy_request_logs ORDER BY input_tokens", + )? + .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))? + .collect::>()?; + assert_eq!(rows.len(), 2); + assert_eq!(rows[0], ("shared-id".to_string(), 10)); + assert!(rows[1].0.starts_with("shared-id:collision:")); + assert_eq!(rows[1].1, 20); + assert_eq!(crate::usage_events::take_test_notify_count(), 2); + Ok(()) + } + + #[test] + fn only_session_log_primary_rows_may_be_replaced() -> Result<(), AppError> { + let db = Database::memory()?; + { + let conn = crate::database::lock_conn!(db.conn); + for (request_id, data_source) in [ + ("session-primary", "session_log"), + ("codex-primary", "codex_session"), + ] { + conn.execute( + "INSERT INTO proxy_request_logs ( + request_id, provider_id, app_type, model, input_tokens, + output_tokens, cache_read_tokens, cache_creation_tokens, + latency_ms, status_code, created_at, data_source + ) VALUES (?1, '_session', 'claude', 'old', 1, 1, 0, 0, 0, 200, 1, ?2)", + rusqlite::params![request_id, data_source], + )?; + } + } + let logger = UsageLogger::new(&db); + + let mut session_replacement = request_log("session-primary", 10); + session_replacement.app_type = "claude".to_string(); + logger.log_request(&session_replacement)?; + logger.log_request(&request_log("codex-primary", 20))?; + + let conn = crate::database::lock_conn!(db.conn); + let session_source: String = conn.query_row( + "SELECT data_source FROM proxy_request_logs WHERE request_id = 'session-primary'", + [], + |row| row.get(0), + )?; + let codex_input: i64 = conn.query_row( + "SELECT input_tokens FROM proxy_request_logs WHERE request_id = 'codex-primary'", + [], + |row| row.get(0), + )?; + let fallback_count: i64 = conn.query_row( + "SELECT COUNT(*) FROM proxy_request_logs + WHERE request_id LIKE 'codex-primary:collision:%'", + [], + |row| row.get(0), + )?; + assert_eq!(session_source, "proxy"); + assert_eq!(codex_input, 1); + assert_eq!(fallback_count, 1); + Ok(()) + } + #[test] fn test_log_error() -> Result<(), AppError> { let db = Database::memory()?; diff --git a/src-tauri/src/proxy/usage/parser.rs b/src-tauri/src/proxy/usage/parser.rs index 45120b8ae..6f438595f 100644 --- a/src-tauri/src/proxy/usage/parser.rs +++ b/src-tauri/src/proxy/usage/parser.rs @@ -30,6 +30,13 @@ fn openai_cache_write_tokens(usage: &Value) -> u32 { /// Session 日志 request_id 前缀，与 `session_usage.rs` 中的格式保持一致 pub const SESSION_REQUEST_ID_PREFIX: &str = "session:"; +fn response_id(body: &Value, field: &str) -> Option { + body.get(field) + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .map(str::to_owned) +} + /// Token 使用量统计 #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct TokenUsage { @@ -47,12 +54,18 @@ pub struct TokenUsage { } impl TokenUsage { - /// 生成与 session 日志共享的 request_id，用于跨源去重。 - /// 有 message_id 时返回 `session:{id}`，否则回退到随机 UUID。 - pub fn dedup_request_id(&self) -> String { + /// 生成稳定 request_id。Claude 不加作用域，以便继续与 session JSONL 的 + /// `session:{message_id}` 主键收敛；其他协议加入 app/provider 作用域，避免 + /// 不同上游复用 envelope id 时互相覆盖。 + pub fn dedup_request_id(&self, scope: Option<(&str, &str)>) -> String { self.message_id .as_ref() - .map(|mid| format!("{SESSION_REQUEST_ID_PREFIX}{mid}")) + .map(|message_id| match scope { + Some((app_type, provider_id)) => { + format!("{SESSION_REQUEST_ID_PREFIX}{app_type}:{provider_id}:{message_id}") + } + None => format!("{SESSION_REQUEST_ID_PREFIX}{message_id}"), + }) .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()) } @@ -88,10 +101,7 @@ impl TokenUsage { .get("model") .and_then(|v| v.as_str()) .map(|s| s.to_string()); - let message_id = body - .get("id") - .and_then(|v| v.as_str()) - .map(|s| s.to_string()); + let message_id = response_id(body, "id"); Some(Self { input_tokens: usage.get("input_tokens")?.as_u64()? as u32, @@ -128,8 +138,8 @@ impl TokenUsage { } } if message_id.is_none() { - if let Some(id) = message.get("id").and_then(|v| v.as_str()) { - message_id = Some(id.to_string()); + if let Some(id) = response_id(message, "id") { + message_id = Some(id); } } } @@ -238,7 +248,7 @@ impl TokenUsage { cache_read_tokens: 0, cache_creation_tokens: 0, model: None, - message_id: None, + message_id: response_id(body, "id"), }) } @@ -277,7 +287,7 @@ impl TokenUsage { cache_read_tokens: cached_tokens, cache_creation_tokens: cache_write_tokens, model, - message_id: None, + message_id: response_id(body, "id"), }) } @@ -312,7 +322,7 @@ impl TokenUsage { cache_read_tokens: cached_tokens, cache_creation_tokens: cache_write_tokens, model, - message_id: None, + message_id: response_id(body, "id"), }) } @@ -404,7 +414,7 @@ impl TokenUsage { cache_read_tokens: cached_tokens, cache_creation_tokens: cache_write_tokens, model, - message_id: None, + message_id: response_id(body, "id"), }) } @@ -416,7 +426,12 @@ impl TokenUsage { if let Some(usage) = event.get("usage") { if !usage.is_null() { log::debug!("[Codex] 找到 usage: {usage:?}"); - return Self::from_openai_response(event); + let mut parsed = Self::from_openai_response(event)?; + if parsed.message_id.is_none() { + parsed.message_id = + events.iter().find_map(|chunk| response_id(chunk, "id")); + } + return Some(parsed); } } } @@ -449,7 +464,7 @@ impl TokenUsage { .unwrap_or(0) as u32, cache_creation_tokens: 0, model, - message_id: None, + message_id: response_id(body, "responseId"), }) } @@ -460,6 +475,7 @@ impl TokenUsage { let mut total_tokens = 0u32; let mut total_cache_read = 0u32; let mut model: Option = None; + let mut message_id: Option = None; for chunk in chunks { if let Some(usage) = chunk.get("usageMetadata") { @@ -488,6 +504,9 @@ impl TokenUsage { model = Some(model_version.to_string()); } } + if message_id.is_none() { + message_id = response_id(chunk, "responseId"); + } } // 输出 tokens = 总 tokens - 输入 tokens @@ -500,7 +519,7 @@ impl TokenUsage { cache_read_tokens: total_cache_read, cache_creation_tokens: 0, model, - message_id: None, + message_id, }) } else { None @@ -513,6 +532,61 @@ mod tests { use super::*; use serde_json::json; + #[test] + fn response_ids_produce_scoped_dedup_keys_and_empty_ids_fall_back() { + let response = json!({ + "id": "resp_123", + "model": "gpt-5.6", + "usage": { "input_tokens": 10, "output_tokens": 2 } + }); + let usage = TokenUsage::from_codex_response(&response).unwrap(); + assert_eq!(usage.message_id.as_deref(), Some("resp_123")); + assert_eq!( + usage.dedup_request_id(Some(("codex", "provider-a"))), + "session:codex:provider-a:resp_123" + ); + + let empty = json!({ + "id": "", + "usage": { "input_tokens": 10, "output_tokens": 2 } + }); + let empty_usage = TokenUsage::from_codex_response(&empty).unwrap(); + assert!(empty_usage.message_id.is_none()); + assert!(!empty_usage + .dedup_request_id(Some(("codex", "provider-a"))) + .starts_with("session:")); + } + + #[test] + fn stream_parsers_recover_ids_from_envelope_chunks() { + let openai = vec![ + json!({"id": "chatcmpl_123", "choices": []}), + json!({ + "usage": { "prompt_tokens": 10, "completion_tokens": 2 }, + "choices": [] + }), + ]; + assert_eq!( + TokenUsage::from_openai_stream_events(&openai) + .unwrap() + .message_id + .as_deref(), + Some("chatcmpl_123") + ); + + let gemini = vec![json!({ + "responseId": "gemini_123", + "usageMetadata": { "promptTokenCount": 10, "totalTokenCount": 12 } + })]; + assert_eq!( + TokenUsage::from_gemini_stream_chunks(&gemini) + .unwrap() + .message_id + .as_deref(), + Some("gemini_123") + ); + } + #[test] fn test_claude_response_parsing() { let response = json!({