diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 7dc9fdbc9..ddb16b673 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1746,6 +1746,7 @@ dependencies = [ "http", "http-body", "http-body-util", + "regex", "secrecy", "serde", "serde_json", diff --git a/backend/crates/gateway-admin/src/lib.rs b/backend/crates/gateway-admin/src/lib.rs index f18f63843..00ff49590 100644 --- a/backend/crates/gateway-admin/src/lib.rs +++ b/backend/crates/gateway-admin/src/lib.rs @@ -521,6 +521,7 @@ async fn initialize_inner( store.settings(), registry.clone(), timezone, + system.clone(), )), settings, usage_data, diff --git a/backend/crates/gateway-admin/src/model/diagnostics.rs b/backend/crates/gateway-admin/src/model/diagnostics.rs new file mode 100644 index 000000000..45ce3e530 --- /dev/null +++ b/backend/crates/gateway-admin/src/model/diagnostics.rs @@ -0,0 +1,44 @@ +//! 故障诊断导出的目标、已采集证据与缺口 + +use chrono::{DateTime, Utc}; + +use super::{ + observability::{OpsError, UsageDetail}, + system::SystemVersion, +}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum DiagnosticTarget { + Request(String), + Operation(String), +} + +impl DiagnosticTarget { + #[must_use] + pub fn id(&self) -> &str { + match self { + Self::Request(id) | Self::Operation(id) => id, + } + } +} + +pub struct DiagnosticGap { + pub section: String, + pub reason: &'static str, +} + +pub struct DiagnosticRequest { + pub request_id: String, + pub detail: Option, + pub errors: Vec, +} + +/// 原始受控详情只能经导出投影脱敏后交付,不参与普通日志或 Debug +pub struct DiagnosticBundle { + pub target: DiagnosticTarget, + pub collected_at: DateTime, + pub environment: Option, + pub error: Option, + pub requests: Vec, + pub gaps: Vec, +} diff --git a/backend/crates/gateway-admin/src/model/mod.rs b/backend/crates/gateway-admin/src/model/mod.rs index cc57d9b46..dcd37e802 100644 --- a/backend/crates/gateway-admin/src/model/mod.rs +++ b/backend/crates/gateway-admin/src/model/mod.rs @@ -10,6 +10,7 @@ pub mod auth; pub mod backup; pub mod client_distribution; pub mod client_keys; +pub mod diagnostics; pub mod import_tasks; pub mod key_usage; pub mod observability; diff --git a/backend/crates/gateway-admin/src/ports/store.rs b/backend/crates/gateway-admin/src/ports/store.rs index de16bbd8b..fcf5cc9aa 100644 --- a/backend/crates/gateway-admin/src/ports/store.rs +++ b/backend/crates/gateway-admin/src/ports/store.rs @@ -485,6 +485,13 @@ pub trait ObservabilityStore: Send + Sync { ) -> AdminStoreResult; async fn list_ops_errors(&self, query: OpsErrorQuery) -> AdminStoreResult; + + /// 按精确目标读取受控错误,不受列表时间窗口与当前页面影响 + async fn diagnostic_errors( + &self, + target: &crate::model::diagnostics::DiagnosticTarget, + limit: u16, + ) -> AdminStoreResult>; } /// Runtime settings 与管理员 API Key 写入 diff --git a/backend/crates/gateway-admin/src/use_case/observability/export.rs b/backend/crates/gateway-admin/src/use_case/observability/export.rs new file mode 100644 index 000000000..5d455f807 --- /dev/null +++ b/backend/crates/gateway-admin/src/use_case/observability/export.rs @@ -0,0 +1,227 @@ +//! 按故障目标收集诊断证据,补充读取失败不丢弃已取得的记录 + +use std::{ + collections::{HashSet, VecDeque}, + time::Duration, +}; + +use tokio::time::{Instant, timeout_at}; + +use super::{DefaultObservabilityService, map_store_error}; +use crate::model::{ + AdminError, AdminErrorKind, + diagnostics::{DiagnosticBundle, DiagnosticGap, DiagnosticRequest, DiagnosticTarget}, + observability::OpsError, +}; + +const MAX_REQUESTS: usize = 8; +const MAX_ERRORS: u16 = 128; + +impl DefaultObservabilityService { + pub(super) async fn collect_diagnostics( + &self, + target: DiagnosticTarget, + ) -> Result { + if target.id().trim().is_empty() || target.id().len() > 256 { + return Err(AdminError::invalid("诊断目标 ID 不合法")); + } + let deadline = Instant::now() + Duration::from_secs(20); + let (collected, environment) = tokio::join!( + self.collect_records(target, deadline), + timeout_at(deadline, self.system.version()), + ); + let mut bundle = collected?; + match environment { + Ok(Ok(version)) => bundle.environment = Some(version), + _ => gap(&mut bundle, "environment", "collection_failed"), + } + Ok(bundle) + } + + async fn collect_records( + &self, + target: DiagnosticTarget, + deadline: Instant, + ) -> Result { + let mut bundle = DiagnosticBundle { + target, + collected_at: chrono::Utc::now(), + environment: None, + error: None, + requests: Vec::new(), + gaps: Vec::new(), + }; + let mut remaining_errors = MAX_ERRORS; + let first_id = match &bundle.target { + DiagnosticTarget::Request(id) => Some(id.clone()), + DiagnosticTarget::Operation(_) => { + let mut errors = self.export_errors(&bundle.target, 1, deadline).await?; + let error = errors + .pop() + .ok_or_else(|| AdminError::not_found("错误记录不存在或已过期"))?; + let id = error.request_id.clone(); + bundle.error = Some(error); + remaining_errors -= 1; + id + } + }; + let mut pending = VecDeque::from_iter(first_id); + let mut seen = HashSet::new(); + while let Some(id) = pending.pop_front() { + if !seen.insert(id.clone()) { + continue; + } + if bundle.requests.len() >= MAX_REQUESTS { + gap(&mut bundle, "requests", "related_request_limit"); + break; + } + let section = format!("requests.{id}"); + let target = DiagnosticTarget::Request(id.clone()); + let selected = bundle + .error + .as_ref() + .filter(|error| error.request_id.as_deref() == Some(id.as_str())); + // 事件导出已经持有选中记录,额外读取一个槽位用于去重后判断是否超限 + let read_limit = remaining_errors + 1 + u16::from(selected.is_some()); + let (detail, errors) = tokio::join!( + timeout_at(deadline, self.store.usage_record_detail(&id)), + self.export_errors(&target, read_limit, deadline), + ); + let (detail, detail_error) = match detail { + Ok(Ok(detail)) => (Some(detail), None), + Ok(Err(error)) => (None, Some(map_store_error(error, "diagnostic request"))), + Err(_) => ( + None, + Some(AdminError::new(AdminErrorKind::Unavailable, "诊断采集超时")), + ), + }; + let mut errors = match errors { + Ok(mut errors) => { + if let Some(selected) = selected { + errors.retain(|error| { + error.source != selected.source || error.event_id != selected.event_id + }); + } + errors + } + Err(error) => { + if detail.is_none() && bundle.requests.is_empty() && bundle.error.is_none() { + return Err(error); + } + gap( + &mut bundle, + &format!("{section}.errors"), + "collection_failed", + ); + Vec::new() + } + }; + if detail.is_none() + && errors.is_empty() + && bundle.requests.is_empty() + && bundle.error.is_none() + { + return Err( + detail_error.unwrap_or_else(|| AdminError::not_found("请求记录不存在或已过期")) + ); + } + if errors.len() > usize::from(remaining_errors) { + gap( + &mut bundle, + &format!("{section}.errors"), + "error_record_limit", + ); + errors.truncate(usize::from(remaining_errors)); + } + remaining_errors = + remaining_errors.saturating_sub(u16::try_from(errors.len()).unwrap_or(MAX_ERRORS)); + if let Some(record) = &detail { + for related in &record.related_requests { + if let Some(id) = related.get("requestId").and_then(serde_json::Value::as_str) { + pending.push_back(id.to_owned()); + } + } + if record.request.outcome == crate::model::observability::RequestOutcome::Running { + gap(&mut bundle, §ion, "request_not_finalized"); + } + match &record.trace { + None => gap( + &mut bundle, + &format!("{section}.trace"), + "not_recorded_or_retained", + ), + Some(trace) => { + if trace["droppedEvents"].as_u64().unwrap_or(0) > 0 { + gap(&mut bundle, &format!("{section}.trace"), "events_evicted"); + } + let events = trace["events"].as_array(); + if !events.is_some_and(|events| { + events.iter().any(|event| { + event["stage"] == "request.routed" + && event["data"]["accountPolicy"].is_object() + }) + }) { + gap(&mut bundle, §ion, "incident_policy_not_recorded"); + } + for event in events.into_iter().flatten() { + if event["stage"] == "capture.gap" || event["data"]["truncated"] == true + { + gap( + &mut bundle, + &format!("{section}.trace"), + "capture_gap_or_truncation", + ); + } + if event["stage"] == "retry.decided" && event["data"]["phase"].is_null() + { + gap( + &mut bundle, + &format!("{section}.trace"), + "legacy_retry_facts_not_recorded", + ); + } + } + } + } + gap( + &mut bundle, + &format!("{section}.attempts"), + "best_effort_persistence", + ); + } else { + gap(&mut bundle, §ion, "request_detail_unavailable"); + } + bundle.requests.push(DiagnosticRequest { + request_id: id, + detail, + errors, + }); + } + Ok(bundle) + } + + async fn export_errors( + &self, + target: &DiagnosticTarget, + limit: u16, + deadline: Instant, + ) -> Result, AdminError> { + timeout_at(deadline, self.store.diagnostic_errors(target, limit)) + .await + .map_err(|_| AdminError::new(AdminErrorKind::Unavailable, "诊断采集超时"))? + .map_err(|error| map_store_error(error, "diagnostic errors")) + } +} + +fn gap(bundle: &mut DiagnosticBundle, section: &str, reason: &'static str) { + if !bundle + .gaps + .iter() + .any(|gap| gap.section == section && gap.reason == reason) + { + bundle.gaps.push(DiagnosticGap { + section: section.to_owned(), + reason, + }); + } +} diff --git a/backend/crates/gateway-admin/src/use_case/observability/mod.rs b/backend/crates/gateway-admin/src/use_case/observability/mod.rs index 513b127f2..17664afa9 100644 --- a/backend/crates/gateway-admin/src/use_case/observability/mod.rs +++ b/backend/crates/gateway-admin/src/use_case/observability/mod.rs @@ -38,6 +38,7 @@ const HEALTH_TIMELINE_SLOT_MINUTES: i64 = 15; const HEALTH_TIMELINE_MIN_SAMPLE_SIZE: u64 = 10; const HEALTH_TIMELINE_UNAVAILABLE_FAILURE_THRESHOLD: u64 = 3; const HEALTH_TIMELINE_STABLE_RELIABILITY: f64 = 95.0; +mod export; mod store_cache; pub(crate) use store_cache::CachedObservabilityStore; @@ -70,6 +71,10 @@ pub trait ObservabilityService: Send + Sync { dimension: DiagnosticDimension, ) -> Result; async fn ops_errors(&self, query: OpsErrorQuery) -> Result; + async fn export_diagnostics( + &self, + target: crate::model::diagnostics::DiagnosticTarget, + ) -> Result; } pub(crate) struct DefaultObservabilityService { @@ -78,6 +83,7 @@ pub(crate) struct DefaultObservabilityService { accounts: Arc, settings: Arc, providers: ProviderAdminRegistry, + system: Arc, } impl DefaultObservabilityService { @@ -88,6 +94,7 @@ impl DefaultObservabilityService { settings: Arc, providers: ProviderAdminRegistry, timezone: gateway_core::time::DeploymentTimeZone, + system: Arc, ) -> Self { Self { timezone, @@ -95,6 +102,7 @@ impl DefaultObservabilityService { accounts, settings, providers, + system, } } @@ -192,6 +200,13 @@ impl DefaultObservabilityService { #[async_trait] impl ObservabilityService for DefaultObservabilityService { + async fn export_diagnostics( + &self, + target: crate::model::diagnostics::DiagnosticTarget, + ) -> Result { + self.collect_diagnostics(target).await + } + async fn dashboard_summary( &self, range: TimeRange, diff --git a/backend/crates/gateway-admin/src/use_case/observability/store_cache.rs b/backend/crates/gateway-admin/src/use_case/observability/store_cache.rs index 44dbf4b02..b8828179b 100644 --- a/backend/crates/gateway-admin/src/use_case/observability/store_cache.rs +++ b/backend/crates/gateway-admin/src/use_case/observability/store_cache.rs @@ -289,4 +289,12 @@ impl ObservabilityStore for CachedObservabilityStore { async fn list_ops_errors(&self, query: OpsErrorQuery) -> AdminStoreResult { self.inner.list_ops_errors(query).await } + + async fn diagnostic_errors( + &self, + target: &crate::model::diagnostics::DiagnosticTarget, + limit: u16, + ) -> AdminStoreResult> { + self.inner.diagnostic_errors(target, limit).await + } } diff --git a/backend/crates/gateway-admin/tests/use_case/mod.rs b/backend/crates/gateway-admin/tests/use_case/mod.rs index dc65dfb7a..208319649 100644 --- a/backend/crates/gateway-admin/tests/use_case/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/mod.rs @@ -749,6 +749,14 @@ impl ClientKeyStore for UnavailableStore { #[async_trait] impl ObservabilityStore for UnavailableStore { + async fn diagnostic_errors( + &self, + _target: &gateway_admin::model::diagnostics::DiagnosticTarget, + _limit: u16, + ) -> AdminStoreResult> { + Err(unavailable("diagnostic errors")) + } + fn query_timeout(&self) -> std::time::Duration { std::time::Duration::from_secs(10) } diff --git a/backend/crates/gateway-admin/tests/use_case/observability/mod.rs b/backend/crates/gateway-admin/tests/use_case/observability/mod.rs index d9c4674dc..327ae943e 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability/mod.rs @@ -945,6 +945,14 @@ impl FixtureObservabilityStore { #[async_trait] impl ObservabilityStore for FixtureObservabilityStore { + async fn diagnostic_errors( + &self, + _target: &gateway_admin::model::diagnostics::DiagnosticTarget, + _limit: u16, + ) -> AdminStoreResult> { + Err(super::unavailable("diagnostic errors")) + } + fn query_timeout(&self) -> std::time::Duration { std::time::Duration::from_secs(10) } diff --git a/backend/crates/gateway-api/Cargo.toml b/backend/crates/gateway-api/Cargo.toml index 788cb70a1..d918fe42d 100644 --- a/backend/crates/gateway-api/Cargo.toml +++ b/backend/crates/gateway-api/Cargo.toml @@ -25,6 +25,7 @@ gateway-admin = { path = "../gateway-admin" } gateway-core = { path = "../gateway-core" } gateway-protocol = { path = "../gateway-protocol" } headers = "0.4.1" +regex = "1.13.1" secrecy = "0.10" serde = { version = "1.0.228", features = ["derive"] } serde_json = { version = "1.0.150", features = ["arbitrary_precision", "preserve_order"] } diff --git a/backend/crates/gateway-api/src/admin/observability/export/mod.rs b/backend/crates/gateway-api/src/admin/observability/export/mod.rs new file mode 100644 index 000000000..77f55892e --- /dev/null +++ b/backend/crates/gateway-api/src/admin/observability/export/mod.rs @@ -0,0 +1,136 @@ +//! 诊断文件的 HTTP 下载与受控数据投影,普通详情响应不携带时间线 + +use axum::{extract::State, http::header, response::IntoResponse}; +use gateway_admin::model::{ + diagnostics::{DiagnosticBundle, DiagnosticTarget}, + observability::OpsError, +}; +use serde::Deserialize; +use serde_json::{Value, json}; + +use super::{ + AdminAuth, AdminError, AdminQuery, map_service_error, ops_error_view, usage_detail_view, +}; +use crate::{auth::SessionState, time::TimePresenter}; + +mod privacy; + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub(super) struct ExportQuery { + request_id: Option, + error_id: Option, +} + +pub(super) async fn download( + _auth: AdminAuth, + State(state): State, + AdminQuery(query): AdminQuery, +) -> Result +where + S: SessionState + Send + Sync, +{ + let target = match (query.request_id, query.error_id) { + (Some(id), None) => DiagnosticTarget::Request(id), + (None, Some(id)) => DiagnosticTarget::Operation(id), + _ => { + return Err(map_service_error( + gateway_admin::model::AdminError::invalid("请指定一个请求 ID 或错误记录 ID"), + )); + } + }; + let bundle = state + .admin_services() + .observability() + .export_diagnostics(target) + .await + .map_err(map_service_error)?; + let time = TimePresenter::new(state.admin_services().timezone()); + let document = privacy::redact(document(bundle, time)); + let body = serde_json::to_vec_pretty(&document).map_err(|_| { + map_service_error(gateway_admin::model::AdminError::internal( + "诊断文件生成失败", + )) + })?; + Ok(( + [ + (header::CONTENT_TYPE, "application/json; charset=utf-8"), + ( + header::CONTENT_DISPOSITION, + "attachment; filename=diagnostics.json", + ), + (header::CACHE_CONTROL, "private, no-store"), + (header::X_CONTENT_TYPE_OPTIONS, "nosniff"), + ], + body, + )) +} + +fn document(bundle: DiagnosticBundle, time: TimePresenter) -> Value { + let target = match &bundle.target { + DiagnosticTarget::Request(id) => json!({"requestId": id}), + DiagnosticTarget::Operation(id) => json!({"errorId": id}), + }; + let requests = bundle.requests.into_iter().map(|record| { + let evidence = record.detail.map(|mut detail| { + let trace = detail.trace.take(); + let related = std::mem::take(&mut detail.related_requests); + let configuration = json!({ + "revision": detail.request.config_revision, + "routingScope": detail.request.routing_scope, + "routingGroupIds": detail.request.routing_group_refs, + "deadlineAt": detail.request.deadline_at, + }); + // 页面 DTO 没有展示的尝试原因与等待事实仍属于导出证据 + let attempt_details = detail.attempts.iter().map(|attempt| json!({ + "id": attempt.id, "attemptIndex": attempt.attempt_index, + "component": attempt.component, "operation": attempt.operation, + "message": attempt.message, "retryAfterMs": attempt.retry_after_ms, + })).collect::>(); + json!({ + "detail": usage_detail_view(detail, time), "trace": trace, + "relatedRequests": related, "incidentConfiguration": configuration, + "attemptDetails": attempt_details, + }) + }); + json!({ + "requestId": record.request_id, + "evidence": evidence, + "errors": record.errors.into_iter().map(|error| export_error(error, time)).collect::>(), + }) + }).collect::>(); + json!({ + "schemaVersion": 5, + "source": "gateway.diagnostics", + "exportPolicy": "support_credentials_redacted", + "target": target, + "collectedAt": bundle.collected_at, + "environment": { + "scope": "export_time", "observedAt": bundle.collected_at, + "gateway": bundle.environment.map(|version| json!({ + "version": version.version, "gitSha": version.git_sha, + "buildTime": version.build_time, "deploymentMode": version.deployment_mode, + })), + }, + "error": bundle.error.map(|error| export_error(error, time)), + "requests": requests, + "availability": {"gaps": bundle.gaps.into_iter().map(|gap| json!({"section": gap.section, "reason": gap.reason})).collect::>()}, + "omitted": ["credentials_and_authentication", "account_contact_details", "uncollected_wire_payloads", "unrelated_requests_and_system_secrets"], + "notes": [ + "诊断包包含错误原因链、上游错误与业务上下文,用于故障分析", + "认证字段、账号联系信息与可识别的内嵌凭据已脱敏,任意业务内容不属于自动识别范围", + "environment 为导出时环境,incidentConfiguration 与 request.routed 为事发配置事实,当前版本不能代替事发版本", + "缺失、截断、淘汰与读取失败见 availability.gaps,没有记录不表示没有发生", + "attempts 为异步 best-effort 投影,firstTokenMs 为空不证明没有输出", + "未采集的请求正文、过期记录、上游内部状态与退出前未落盘信息无法由导出补回" + ], + }) +} + +// 复用错误字段投影,原始详情直接进入脱敏遍历,避免先为页面缩进再解析一遍 +fn export_error(mut error: OpsError, time: TimePresenter) -> Value { + let details = error.error_details.take(); + let mut value = json!(ops_error_view(error, time)); + value["errorDetails"] = json!(details); + value +} diff --git a/backend/crates/gateway-api/src/admin/observability/export/privacy.rs b/backend/crates/gateway-api/src/admin/observability/export/privacy.rs new file mode 100644 index 000000000..ac397738d --- /dev/null +++ b/backend/crates/gateway-api/src/admin/observability/export/privacy.rs @@ -0,0 +1,220 @@ +//! 诊断文件按认证与观测合同移除敏感字段,保留错误事实并明确导出截断 + +use regex::Regex; +use serde::Serialize; +use serde_json::{Value, json}; +use std::sync::LazyLock; + +const REDACTED: &str = ""; +const MAX_TEXT: usize = 256 * 1024; +static INLINE: LazyLock> = LazyLock::new(|| { + [ + (r"(?i)\b((?:set-cookie|cookie2?)\s*:\s*)[^\r\n]+", "${1}"), + (r#"(?i)\b(Bearer|Basic)\s+[^\s"'<>;,]+"#, "${1} "), + // 官方 TokenData 的 access_token / id_token 为 JWT,不按猜测的令牌前缀过滤普通文本 + (r"\beyJ[\w-]+\.[\w-]+\.[\w-]+\b", ""), + (r"(?i)([a-z][a-z0-9+.-]*://)[^\s/@]+@", "${1}@"), + (r#"(?i)\b((?:(?:personal_?)?access_?token|refresh_?token|id_?token|(?:openai_)?api_?key|(?:current_|new_)?password|authorization|proxy-authorization)["']?\s*[=:]\s*)("[^"\r\n]*"|'[^'\r\n]*'|[^\s&,;]+)"#, "${1}"), + ].into_iter().map(|(pattern, replacement)| (Regex::new(pattern).expect("static credential pattern"), replacement)).collect() +}); + +#[derive(Default, Serialize)] +#[serde(rename_all = "camelCase")] +struct Report { + credential_fields: usize, + personal_fields: usize, + inline_credentials: usize, + truncated_values: usize, +} + +struct Redactor { + report: Report, + bytes: usize, + nodes: usize, +} + +pub(super) fn redact(mut value: Value) -> Value { + let mut redactor = Redactor { + report: Report::default(), + bytes: 8 * 1024 * 1024, + nodes: 100_000, + }; + // 先处理有界文件信息,正文耗尽预算也不能删除 schema、目标或既有缺口 + let mut error = value["error"].take(); + let mut requests = value["requests"].take(); + redactor.visit(&mut value, 0); + redactor.visit(&mut error, 0); + redactor.visit(&mut requests, 0); + value["error"] = error; + value["requests"] = requests; + if redactor.report.truncated_values > 0 { + // 缺口清单可能也触及预算,顶层统计始终保留以免截断变成静默丢失 + value["exportTruncated"] = json!(true); + if !value["availability"]["gaps"].is_array() { + value["availability"] = json!({"gaps": []}); + } + value["availability"]["gaps"] + .as_array_mut() + .expect("gap array") + .push(json!({ + "section": "export", "reason": "export_size_limit", + })); + } + value["redaction"] = json!(redactor.report); + value +} + +fn normalized(key: &str) -> String { + key.chars() + .filter(|character| !matches!(character, '_' | '-')) + .flat_map(char::to_lowercase) + .collect() +} + +// OAuth 字段来自 Provider credential schema 与官方 Codex TokenData;API Key 来自 +// CodexCprApiKeyExportAccount / AuthDotJson;密码来自管理登录合同,头部来自传输身份合同 +// 不用 token、secret、credential 等泛称推断敏感性,避免删除业务错误上下文 +fn secret(key: &str) -> bool { + matches!( + key, + "accesstoken" + | "personalaccesstoken" + | "refreshtoken" + | "idtoken" + | "apikey" + | "openaiapikey" + | "password" + | "currentpassword" + | "newpassword" + | "authorization" + | "proxyauthorization" + | "xapikey" + | "cookie" + | "cookie2" + | "setcookie" + | "xoaiis" + | "xoaiisupdate" + | "xoaiattestation" + ) +} + +impl Redactor { + fn omitted(&mut self) -> Value { + self.report.truncated_values += 1; + json!({"omitted": "export_size_limit"}) + } + + fn text(&mut self, text: &str) -> String { + let mut result = text.to_owned(); + for (pattern, replacement) in INLINE.iter() { + self.report.inline_credentials += pattern.find_iter(&result).count(); + result = pattern.replace_all(&result, *replacement).into_owned(); + } + result + } + + fn visit(&mut self, value: &mut Value, depth: usize) { + if self.nodes == 0 || self.bytes == 0 || depth > 32 { + *value = self.omitted(); + return; + } + self.nodes -= 1; + // 预算同时计入缩进和容器分隔符,避免深层 JSON 的下载体积显著超过正文预算 + self.bytes = self.bytes.saturating_sub(24 + depth * 4); + match value { + Value::String(text) => { + if text.trim_start().starts_with('{') || text.trim_start().starts_with('[') { + if text.len() > MAX_TEXT { + *value = self.omitted(); + return; + } + if let Ok(parsed) = serde_json::from_str::(text) { + *value = parsed; + self.visit(value, depth + 1); + return; + } + } + let cleaned = self.text(text); + let limit = MAX_TEXT.min(self.bytes / 2); + if cleaned.len() > limit { + let boundary = cleaned.floor_char_boundary(limit); + *value = json!({"text": &cleaned[..boundary], "omittedBytes": cleaned.len() - boundary}); + self.report.truncated_values += 1; + } else { + *text = cleaned; + } + self.bytes = self.bytes.saturating_sub(value.to_string().len()); + } + Value::Array(items) => { + // 头部也可能以二元组承载,不能只识别对象字段名 + if items.len() == 2 + && items[0] + .as_str() + .is_some_and(|name| secret(&normalized(name))) + { + items[1] = json!(REDACTED); + self.report.credential_fields += 1; + } + let original_len = items.len(); + let mut kept = 0; + for item in items.iter_mut().take(4096) { + if self.nodes == 0 || self.bytes == 0 { + break; + } + self.visit(item, depth + 1); + kept += 1; + } + if kept < original_len { + items.truncate(kept); + items.push(json!({"omittedItems": original_len - kept})); + self.report.truncated_values += 1; + } + } + Value::Object(object) => { + if object + .get("name") + .and_then(Value::as_str) + .is_some_and(|name| secret(&normalized(name))) + && object.contains_key("value") + { + object.insert("value".to_owned(), json!(REDACTED)); + self.report.credential_fields += 1; + } + let fields = std::mem::take(object); + let total = fields.len(); + for (key, mut item) in fields { + if object.len() >= 4096 || self.nodes == 0 || self.bytes == 0 { + break; + } + let normalized = normalized(&key); + if secret(&normalized) { + item = json!(REDACTED); + self.report.credential_fields += 1; + } else if matches!( + normalized.as_str(), + "email" + | "accountemail" + | "clientip" + | "accountname" + | "accountnotes" + | "credentialname" + | "clientapikeyname" + | "accountlabel" + ) { + item = json!(REDACTED); + self.report.personal_fields += 1; + } else { + self.visit(&mut item, depth + 1); + } + self.bytes = self.bytes.saturating_sub(key.len()); + object.insert(self.text(&key), item); + } + if object.len() < total { + object.insert("_omittedFields".to_owned(), json!(total - object.len())); + self.report.truncated_values += 1; + } + } + _ => {} + } + } +} diff --git a/backend/crates/gateway-api/src/admin/observability/mod.rs b/backend/crates/gateway-api/src/admin/observability/mod.rs index 29e953815..b9addf717 100644 --- a/backend/crates/gateway-api/src/admin/observability/mod.rs +++ b/backend/crates/gateway-api/src/admin/observability/mod.rs @@ -16,6 +16,7 @@ use super::{ wire::map_admin_service_error, }; +mod export; mod presenter; mod query; mod routes; diff --git a/backend/crates/gateway-api/src/admin/observability/presenter.rs b/backend/crates/gateway-api/src/admin/observability/presenter.rs index 8064b5603..0d0950752 100644 --- a/backend/crates/gateway-api/src/admin/observability/presenter.rs +++ b/backend/crates/gateway-api/src/admin/observability/presenter.rs @@ -535,8 +535,6 @@ pub(crate) fn usage_detail_view( time: crate::time::TimePresenter, ) -> UsageRecordDetailView { UsageRecordDetailView { - trace: detail.trace, - related_requests: detail.related_requests, request: usage_record_view(detail.request, time), attempts: detail .attempts @@ -616,7 +614,7 @@ pub(crate) fn ops_error_view( subagent_kind: error.subagent_kind, compact: error.compact, message: error.message, - error_details: error.error_details, + error_details: error.error_details.map(error_details_view), metadata: OpsErrorMetadataView { source: error.source, component: error.component, @@ -640,6 +638,20 @@ pub(crate) fn ops_error_view( } } +// 受控错误正文保持原文语义,嵌套 JSON 只在展示投影中展开缩进 +fn error_details_view(raw: String) -> String { + let Ok(mut value) = serde_json::from_str::(&raw) else { + return raw; + }; + if let Some(upstream) = value.get_mut("upstream") + && let Some(text) = upstream.as_str() + && let Ok(parsed) = serde_json::from_str::(text) + { + *upstream = parsed; + } + serde_json::to_string_pretty(&value).unwrap_or(raw) +} + pub(crate) fn ops_page_view( page: domain::OpsErrorPage, time: crate::time::TimePresenter, diff --git a/backend/crates/gateway-api/src/admin/observability/routes.rs b/backend/crates/gateway-api/src/admin/observability/routes.rs index 1a88f5076..1d42f31cf 100644 --- a/backend/crates/gateway-api/src/admin/observability/routes.rs +++ b/backend/crates/gateway-api/src/admin/observability/routes.rs @@ -29,6 +29,10 @@ where get(usage_insights_diagnostics::), ) .route("/api/admin/operations/errors", get(ops_errors::)) + .route( + "/api/admin/diagnostics/export", + get(super::export::download::), + ) } pub(crate) async fn dashboard_summary( diff --git a/backend/crates/gateway-api/src/admin/observability/wire.rs b/backend/crates/gateway-api/src/admin/observability/wire.rs index d1baa0aa8..c36dae1e1 100644 --- a/backend/crates/gateway-api/src/admin/observability/wire.rs +++ b/backend/crates/gateway-api/src/admin/observability/wire.rs @@ -275,8 +275,6 @@ pub struct UsageAttemptView { #[derive(Serialize)] #[serde(rename_all = "camelCase")] pub struct UsageRecordDetailView { - pub trace: Option, - pub related_requests: Vec, #[serde(flatten)] pub request: UsageRecordView, pub attempts: Vec, diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index 3408b6728..40eda0870 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -1381,6 +1381,27 @@ impl AccountRuntimeStore for UnusedStore { #[async_trait] impl ObservabilityStore for UnusedStore { + async fn diagnostic_errors( + &self, + target: &gateway_admin::model::diagnostics::DiagnosticTarget, + limit: u16, + ) -> AdminStoreResult> { + let errors = self.ops_errors.lock().expect("ops errors"); + Ok(errors + .iter() + .filter(|error| match target { + gateway_admin::model::diagnostics::DiagnosticTarget::Request(id) => { + error.request_id.as_ref() == Some(id) + } + gateway_admin::model::diagnostics::DiagnosticTarget::Operation(id) => { + &error.event_id == id && error.source == "ops_event" + } + }) + .take(usize::from(limit)) + .cloned() + .collect()) + } + fn query_timeout(&self) -> std::time::Duration { std::time::Duration::from_secs(10) } @@ -1470,11 +1491,12 @@ impl ObservabilityStore for UnusedStore { }) } - async fn usage_record_detail(&self, _: &str) -> AdminStoreResult { + async fn usage_record_detail(&self, request_id: &str) -> AdminStoreResult { self.usage_detail .lock() .expect("usage detail") .clone() + .filter(|detail| detail.request.id == request_id) .ok_or_else(|| unavailable("usage detail")) } diff --git a/backend/crates/gateway-api/tests/admin/observability/export/mod.rs b/backend/crates/gateway-api/tests/admin/observability/export/mod.rs new file mode 100644 index 000000000..b097cd15a --- /dev/null +++ b/backend/crates/gateway-api/tests/admin/observability/export/mod.rs @@ -0,0 +1,372 @@ +//! 诊断下载的权限、证据完整性、脱敏与部分失败合同 + +use axum::{ + body::{Body, to_bytes}, + http::{Request, StatusCode, header}, +}; +use chrono::Utc; +use gateway_admin::model::observability::UsageDetail; +use gateway_api::admin::observability; +use serde_json::{Value, json}; +use tower::ServiceExt as _; + +use super::response::{ops_error_record, usage_record_with_account}; +use crate::admin::{AdminTestFixture, AdminTestState}; + +async fn export( + fixture: &AdminTestFixture, + query: &str, + authenticated: bool, +) -> (StatusCode, axum::http::HeaderMap, Value) { + let mut request = Request::builder() + .uri(format!("/api/admin/diagnostics/export{query}")) + .header("x-request-id", "diagnostics-export-test"); + if authenticated { + request = request.header(header::COOKIE, "cpr_session=valid-session"); + } + let response = observability::router::() + .with_state(fixture.state()) + .oneshot(request.body(Body::empty()).unwrap()) + .await + .unwrap(); + let status = response.status(); + let headers = response.headers().clone(); + let body = to_bytes(response.into_body(), 20 * 1024 * 1024) + .await + .unwrap(); + (status, headers, serde_json::from_slice(&body).unwrap()) +} + +#[tokio::test] +async fn diagnostic_export_requires_admin_and_exactly_one_target() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + assert_eq!( + export(&fixture, "?errorId=missing", false).await.0, + StatusCode::UNAUTHORIZED + ); + for query in [ + "", + "?requestId=a&errorId=b", + "?requestId=", + "?errorId=a&payload=untrusted", + ] { + assert_eq!( + export(&fixture, query, true).await.0, + StatusCode::BAD_REQUEST, + "{query}" + ); + } + assert_eq!( + export(&fixture, "?errorId=missing", true).await.0, + StatusCode::NOT_FOUND + ); +} + +#[tokio::test] +async fn diagnostic_export_keeps_retry_evidence_and_original_errors_but_not_credentials() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let mut error = ops_error_record(); + error.error_details = Some(json!({ + "causes": ["TLS certificate verification failed", "pool timeout after 5 seconds"], + "upstream": json!({"error": {"code": "unknown_vendor_failure", "message": "quota exhausted"}, + "headers": {"Authorization": "fixture-authorization", "set-cookie": "fixture-cookie"}, + "access_token": "fixture-access", "refreshToken": "fixture-refresh", + }).to_string(), + "endpoint": "postgres://fixture-user:fixture-password@database:5432/main", + "message": "connect failed: Bearer fixture-bearer, retry after 30 seconds", + }).to_string()); + let mut unrelated = error.clone(); + unrelated.request_id = Some("another-request".into()); + unrelated.message = "unrelated evidence must not leak".into(); + fixture + .ops_errors + .lock() + .unwrap() + .extend([error, unrelated]); + let trace = json!({"droppedEvents": 0, "events": [ + {"stage": "request.routed", "data": {"accountPolicy": {"maxAccountRotations": 3}}}, + {"stage": "account.selection", "data": {"candidates": [{"id": "acct_high", "weight": 2}, {"id": "acct_low", "weight": 1}]}}, + {"stage": "attempt.failed", "data": {"kind": "unavailable", "recovery": {"replaySafe": true}}}, + {"stage": "retry.decided", "data": {"phase": "execute", "retryable": false, "clientRetrySuppressed": true, "replayBarrierReached": false}}, + ]}); + fixture.usage_detail.lock().unwrap().replace(UsageDetail { + request: usage_record_with_account( + "req_err", + "acct_err", + "private name", + "fixture@example.invalid", + "api_key", + Utc::now(), + ), + attempts: Vec::new(), + trace: Some(trace.clone()), + related_requests: Vec::new(), + }); + let (status, headers, file) = export(&fixture, "?requestId=req_err", true).await; + assert_eq!(status, StatusCode::OK, "{file}"); + assert_eq!( + headers[header::CONTENT_DISPOSITION], + "attachment; filename=diagnostics.json" + ); + assert_eq!(headers[header::CACHE_CONTROL], "private, no-store"); + assert_eq!(file["schemaVersion"], 5); + assert_eq!(file["requests"][0]["evidence"]["trace"], trace); + assert_eq!( + file["requests"][0]["evidence"]["incidentConfiguration"]["revision"], + 1 + ); + let details = &file["requests"][0]["errors"][0]["errorDetails"]; + assert_eq!( + details["upstream"]["error"]["code"], + "unknown_vendor_failure" + ); + assert_eq!(details["causes"][0], "TLS certificate verification failed"); + assert_eq!(file["requests"][0]["errors"][0]["accountId"], "acct_err"); + assert!( + file["availability"]["gaps"] + .as_array() + .unwrap() + .iter() + .any(|gap| gap["section"] == "environment") + ); + let text = file.to_string(); + for marker in [ + "fixture-authorization", + "fixture-cookie", + "fixture-access", + "fixture-refresh", + "fixture-password", + "fixture-user", + "fixture-bearer", + "fixture@example.invalid", + "private name", + "unrelated evidence must not leak", + "203.0.113.8", + ] { + assert!(!text.contains(marker), "leaked {marker}"); + } + assert!(text.contains("retry after 30 seconds")); +} + +#[tokio::test] +async fn background_error_exports_without_fabricating_a_model_request() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let mut error = ops_error_record(); + error.source = "ops_event".into(); + error.request_id = None; + error.component = "worker".into(); + error.error_details = + Some(json!({"causes": ["database connection refused"], "upstream": null}).to_string()); + fixture.ops_errors.lock().unwrap().push(error); + let (status, _, file) = export(&fixture, "?errorId=err_snapshot", true).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(file["requests"], json!([])); + assert_eq!(file["error"]["requestId"], Value::Null); + assert_eq!( + file["error"]["errorDetails"]["causes"][0], + "database connection refused" + ); +} + +#[tokio::test] +async fn error_survives_missing_request_and_nested_credentials_are_redacted() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let mut error = ops_error_record(); + error.error_details = Some(json!({ + "cookies": "Cookie: a=fixture-cookie-a; b=fixture-cookie-b\nupstream timeout", + "uri": "https://provider.invalid/path?api_key=fixture-query&model=gpt-test", + "headers": [["Authorization", "fixture-tuple"], {"name": "Cookie", "value": "fixture-header"}], + }).to_string()); + fixture.ops_errors.lock().unwrap().push(error); + let (status, _, file) = export(&fixture, "?requestId=req_err", true).await; + assert_eq!(status, StatusCode::OK); + assert!(file["requests"][0]["evidence"].is_null()); + assert!( + file["availability"]["gaps"] + .as_array() + .unwrap() + .iter() + .any(|gap| gap["reason"] == "request_detail_unavailable") + ); + let text = file.to_string(); + for marker in [ + "fixture-cookie-a", + "fixture-cookie-b", + "fixture-query", + "fixture-tuple", + "fixture-header", + ] { + assert!(!text.contains(marker), "leaked {marker}"); + } + assert!(text.contains("upstream timeout")); + assert!(text.contains("model=gpt-test")); +} + +#[tokio::test] +async fn oversized_error_export_reports_truncation_instead_of_silent_loss() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let mut error = ops_error_record(); + error.source = "ops_event".into(); + error.request_id = None; + error.message = "诊断文本".repeat(100_000); + fixture.ops_errors.lock().unwrap().push(error); + let (status, _, file) = export(&fixture, "?errorId=err_snapshot", true).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(file["exportTruncated"], true); + assert!(file["redaction"]["truncatedValues"].as_u64().unwrap() > 0); + assert!(file["error"]["message"]["omittedBytes"].as_u64().unwrap() > 0); +} + +#[tokio::test] +async fn related_read_failures_and_limits_keep_primary_evidence() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let mut related = vec![json!({"requestId": "req_err", "relation": "recovers"})]; + related.extend( + (0..10).map( + |index| json!({"requestId": format!("missing_{index}"), "relation": "recovered_by"}), + ), + ); + fixture.usage_detail.lock().unwrap().replace(UsageDetail { + request: usage_record_with_account( + "req_err", + "acct_err", + "fixture", + "fixture@example.invalid", + "api_key", + Utc::now(), + ), + attempts: Vec::new(), + trace: None, + related_requests: related, + }); + fixture + .ops_errors + .lock() + .unwrap() + .extend((0..130).map(|index| { + let mut error = ops_error_record(); + error.event_id = format!("error_{index}"); + error + })); + let (status, _, file) = export(&fixture, "?requestId=req_err", true).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(file["requests"].as_array().unwrap().len(), 8); + assert_eq!(file["requests"][0]["errors"].as_array().unwrap().len(), 128); + assert_eq!( + file["requests"][0]["evidence"]["detail"]["requestId"], + "req_err" + ); + assert!(file["requests"][1]["evidence"].is_null()); + let gaps = file["availability"]["gaps"].as_array().unwrap(); + for reason in [ + "related_request_limit", + "error_record_limit", + "request_detail_unavailable", + "not_recorded_or_retained", + ] { + assert!( + gaps.iter().any(|gap| gap["reason"] == reason), + "missing {reason}" + ); + } +} + +#[tokio::test] +async fn credential_schemas_are_redacted_without_guessing_business_field_names() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let mut error = ops_error_record(); + let business = json!({ + "token": {"input": 128, "output": 64}, + "credentials": {"status": "expired", "provider": "openai"}, + "secret": "tool argument name", "clientSecret": "business field description", + "accountId": "account-correlation", "userId": "user-correlation", + "planType": "plus", "status": "active", "accessTokenExpiresAt": "2026-10-16T00:00:00Z", + "error": {"code": "unknown_code", "message": "upstream capacity exceeded"}, + }); + error.error_details = Some(json!({ + "business": business, + "account": {"accessToken": "fixture-access", "refreshToken": "fixture-refresh", "idToken": "fixture-id"}, + "auth": {"OPENAI_API_KEY": "fixture-key", "personal_access_token": "fixture-pat", "tokens": { + "access_token": "fixture-official-access", "refresh_token": "fixture-official-refresh", + "id_token": "fixture-official-id", "account_id": "account-correlation", + }}, + }).to_string()); + fixture.ops_errors.lock().unwrap().push(error); + let (status, _, file) = export(&fixture, "?requestId=req_err", true).await; + assert_eq!(status, StatusCode::OK); + let details = &file["requests"][0]["errors"][0]["errorDetails"]; + assert_eq!(details["business"], business); + assert_eq!( + details["auth"]["tokens"]["account_id"], + "account-correlation" + ); + assert!(!file.to_string().contains("fixture-")); +} + +#[tokio::test] +async fn export_budget_keeps_format_target_and_existing_gap_records() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + fixture.usage_detail.lock().unwrap().replace(UsageDetail { + request: usage_record_with_account( + "req_err", + "acct_err", + "QA", + "qa@example.invalid", + "oauth", + Utc::now(), + ), + attempts: Vec::new(), + trace: Some(json!({"events": vec![vec![true; 32]; 3500]})), + related_requests: Vec::new(), + }); + let (status, _, file) = export(&fixture, "?requestId=req_err", true).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(file["schemaVersion"], 5); + assert_eq!(file["source"], "gateway.diagnostics"); + assert_eq!(file["target"]["requestId"], "req_err"); + assert_eq!(file["exportTruncated"], true); + let gaps = file["availability"]["gaps"].as_array().unwrap(); + assert!(gaps.iter().any(|gap| gap["reason"] == "export_size_limit")); + assert!( + gaps.iter() + .any(|gap| gap["reason"] == "incident_policy_not_recorded") + ); + assert!(serde_json::to_vec_pretty(&file).unwrap().len() < 16 * 1024 * 1024); +} + +#[tokio::test] +async fn selected_event_is_exported_once_and_shares_the_total_error_budget() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let mut selected = ops_error_record(); + selected.source = "ops_event".into(); + let mut errors = vec![selected]; + errors.extend((0..129).map(|index| { + let mut error = ops_error_record(); + error.source = "ops_event".into(); + error.event_id = format!("related_{index}"); + error + })); + fixture.ops_errors.lock().unwrap().extend(errors); + let (status, _, file) = export(&fixture, "?errorId=err_snapshot", true).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(file["error"]["id"], "err_snapshot"); + let errors = file["requests"][0]["errors"].as_array().unwrap(); + assert_eq!(errors.len(), 127); + assert!(errors.iter().all(|error| error["id"] != "err_snapshot")); + assert!( + file["availability"]["gaps"] + .as_array() + .unwrap() + .iter() + .any(|gap| gap["reason"] == "error_record_limit") + ); +} diff --git a/backend/crates/gateway-api/tests/admin/observability/mod.rs b/backend/crates/gateway-api/tests/admin/observability/mod.rs index c03da4c91..b4baee239 100644 --- a/backend/crates/gateway-api/tests/admin/observability/mod.rs +++ b/backend/crates/gateway-api/tests/admin/observability/mod.rs @@ -1,4 +1,5 @@ //! Admin 观测 wire/路由的固定合同测试 +mod export; mod query; mod response; diff --git a/backend/crates/gateway-api/tests/admin/observability/response.rs b/backend/crates/gateway-api/tests/admin/observability/response.rs index 2eb1eb001..2d777f268 100644 --- a/backend/crates/gateway-api/tests/admin/observability/response.rs +++ b/backend/crates/gateway-api/tests/admin/observability/response.rs @@ -398,7 +398,7 @@ async fn usage_detail_should_keep_attempt_snapshot_contract() { } #[tokio::test] -async fn zero_attempt_failure_detail_keeps_missing_upstream_facts_and_preparation_trace() { +async fn zero_attempt_failure_detail_omits_trace_and_keeps_missing_upstream_facts() { use axum::{ body::{Body, to_bytes}, http::{Request, StatusCode, header}, @@ -470,7 +470,8 @@ async fn zero_attempt_failure_detail_keeps_missing_upstream_facts_and_preparatio assert_eq!(data["attempts"], json!([])); // 明细仍沿用全局 best-effort 合同,不从预备 trace 推断完整尝试列表 assert_eq!(data["attemptsComplete"], false); - assert_eq!(data["trace"], trace); + assert!(data.get("trace").is_none()); + assert!(data.get("relatedRequests").is_none()); assert_eq!(data["logicalOutcome"], "failed"); assert_eq!(data["clientStatusCode"], 503); assert_eq!(data["message"], "no upstream account is available"); @@ -495,8 +496,6 @@ async fn ops_errors_should_keep_account_label_and_authentication_contract() { body::{Body, to_bytes}, http::{Request, StatusCode, header}, }; - use chrono::Utc; - use gateway_admin::model::observability::OpsError; use gateway_api::admin::observability; use tower::ServiceExt as _; @@ -508,63 +507,7 @@ async fn ops_errors_should_keep_account_label_and_authentication_contract() { .ops_errors .lock() .expect("ops errors") - .push(OpsError { - client_api_key_name: Some("Production".to_owned()), - source: "model_request".to_owned(), - event_id: "err_snapshot".to_owned(), - request_id: Some("req_err".to_owned()), - attempt_index: Some(1), - client_api_key_ref: Some("key_err".to_owned()), - component: "model_request".to_owned(), - operation: "responses".to_owned(), - protocol: Some("openai".to_owned()), - client_transport: Some("http_sse".to_owned()), - requested_model_id: Some("gpt-5.4".to_owned()), - service_tier: Some("priority".to_owned()), - endpoint: Some("/v1/responses".to_owned()), - provider_kind: Some("openai".to_owned()), - provider_account_ref: Some("acct_err".to_owned()), - provider_account_name: None, - provider_account_email: Some("err@example.invalid".to_owned()), - provider_account_plan_type: Some("pro".to_owned()), - provider_account_plan_type_display: Some("Pro".to_owned()), - provider_account_authentication_kind: Some("api_key".to_owned()), - upstream_model_id: Some("upstream-err".to_owned()), - upstream_transport: Some("http_sse".to_owned()), - failure_kind: "upstream_error".to_owned(), - upstream_send_state: Some("sent".to_owned()), - client_status_code: Some(502), - upstream_status_code: Some(502), - provider_error_code: Some("upstream".to_owned()), - client_response_id: None, - upstream_request_id: None, - latency_ms: Some(90), - message: "snapshot error".to_owned(), - error_details: Some( - r#"{"error":{"code":"upstream","message":"raw upstream marker"}}"#.to_owned(), - ), - client_ip: Some("203.0.113.8".to_owned()), - user_agent: Some("codex-cli/0.144.0".to_owned()), - reasoning_effort: Some("medium".to_owned()), - reasoning_preset: None, - request_kind: Some("root".to_owned()), - subagent_kind: None, - compact: Some(false), - continuation_affinity_hash: Some("affinity-hash".to_owned()), - continuation_previous_response_id_hash: Some("response-hash".to_owned()), - continuation_unavailable_reason: Some("reused_connection_lost".to_owned()), - upstream_connection_id: Some("connection-id".to_owned()), - upstream_connection_exit_reason: Some("tcp_reset".to_owned()), - upstream_connection_age_ms: Some(12_000), - upstream_connection_idle_ms: Some(4_000), - recovery_request_id: None, - recovered_at: None, - recovery_attempt_count: 0, - recovery_retry_delay_ms: None, - recovery_total_latency_ms: None, - occurred_at: Utc::now(), - stable_sort_id: "model_request:req_err".to_owned(), - }); + .push(ops_error_record()); let response = observability::router::() .with_state(fixture.state()) .oneshot( @@ -621,7 +564,7 @@ async fn ops_errors_should_keep_account_label_and_authentication_contract() { "userAgent": "codex-cli/0.144.0", "providerErrorCode": "upstream", "upstreamSendState": "sent", - "errorDetails": "{\"error\":{\"code\":\"upstream\",\"message\":\"raw upstream marker\"}}", + "errorDetails": serde_json::to_string_pretty(&json!({"error": {"code": "upstream", "message": "raw upstream marker"}})).unwrap(), "continuationUnavailableReason": "reused_connection_lost", "upstreamConnectionExitReason": "tcp_reset", "upstreamConnectionAgeMs": 12000, @@ -707,7 +650,7 @@ async fn diagnostics_should_keep_stable_key_and_display_name_contract() { ); } -fn usage_record_with_account( +pub(super) fn usage_record_with_account( id: &str, account_id: &str, name: &str, @@ -1303,3 +1246,64 @@ async fn observability_queries_reject_group_filters_and_analytics_search() { assert!(observations.records.is_empty()); assert!(observations.errors.is_empty()); } + +pub(super) fn ops_error_record() -> gateway_admin::model::observability::OpsError { + use gateway_admin::model::observability::OpsError; + OpsError { + client_api_key_name: Some("Production".to_owned()), + source: "model_request".to_owned(), + event_id: "err_snapshot".to_owned(), + request_id: Some("req_err".to_owned()), + attempt_index: Some(1), + client_api_key_ref: Some("key_err".to_owned()), + component: "model_request".to_owned(), + operation: "responses".to_owned(), + protocol: Some("openai".to_owned()), + client_transport: Some("http_sse".to_owned()), + requested_model_id: Some("gpt-5.4".to_owned()), + service_tier: Some("priority".to_owned()), + endpoint: Some("/v1/responses".to_owned()), + provider_kind: Some("openai".to_owned()), + provider_account_ref: Some("acct_err".to_owned()), + provider_account_name: None, + provider_account_email: Some("err@example.invalid".to_owned()), + provider_account_plan_type: Some("pro".to_owned()), + provider_account_plan_type_display: Some("Pro".to_owned()), + provider_account_authentication_kind: Some("api_key".to_owned()), + upstream_model_id: Some("upstream-err".to_owned()), + upstream_transport: Some("http_sse".to_owned()), + failure_kind: "upstream_error".to_owned(), + upstream_send_state: Some("sent".to_owned()), + client_status_code: Some(502), + upstream_status_code: Some(502), + provider_error_code: Some("upstream".to_owned()), + client_response_id: None, + upstream_request_id: None, + latency_ms: Some(90), + message: "snapshot error".to_owned(), + error_details: Some( + r#"{"error":{"code":"upstream","message":"raw upstream marker"}}"#.to_owned(), + ), + client_ip: Some("203.0.113.8".to_owned()), + user_agent: Some("codex-cli/0.144.0".to_owned()), + reasoning_effort: Some("medium".to_owned()), + reasoning_preset: None, + request_kind: Some("root".to_owned()), + subagent_kind: None, + compact: Some(false), + continuation_affinity_hash: Some("affinity-hash".to_owned()), + continuation_previous_response_id_hash: Some("response-hash".to_owned()), + continuation_unavailable_reason: Some("reused_connection_lost".to_owned()), + upstream_connection_id: Some("connection-id".to_owned()), + upstream_connection_exit_reason: Some("tcp_reset".to_owned()), + upstream_connection_age_ms: Some(12_000), + upstream_connection_idle_ms: Some(4_000), + recovery_request_id: None, + recovered_at: None, + recovery_attempt_count: 0, + recovery_retry_delay_ms: None, + recovery_total_latency_ms: None, + occurred_at: Utc::now(), + stable_sort_id: "model_request:req_err".to_owned(), + } +} diff --git a/backend/crates/gateway-core/src/diagnostics/trace.rs b/backend/crates/gateway-core/src/diagnostics/trace.rs index 8874c5166..9bad2ee6e 100644 --- a/backend/crates/gateway-core/src/diagnostics/trace.rs +++ b/backend/crates/gateway-core/src/diagnostics/trace.rs @@ -132,6 +132,15 @@ impl TraceContext { } let data = json!({ "kind": error.kind().as_str(), "sendState": error.send_state().as_str(), + "recovery": { + "replaySafe": error.replay_is_safe(), + "retryProhibited": error.retry_is_prohibited(), + "sameAccountCredentialRetry": error.retries_same_account(), + "intent": error.pre_delivery_retry().map(|intent| format!("{intent:?}")), + "continuationFailure": error.continuation_failure().map(|failure| format!("{failure:?}")), + "continuationDisposition": error.continuation_recovery_disposition().map(|disposition| format!("{disposition:?}")), + "retryAfterMs": error.retry_after().map(|delay| delay.as_millis()), + }, "diagnostic": error.diagnostic().map(|diagnostic| json!({ "stage": diagnostic.stage(), "code": diagnostic.code(), "message": diagnostic.as_str(), })), @@ -291,6 +300,27 @@ impl TraceContext { } summary["upstreamStatus"] = json!(data.get("upstreamStatus").and_then(Value::as_u64)); + if let Some(recovery) = data.get("recovery") { + let mut kept = json!({}); + for key in [ + "replaySafe", + "retryProhibited", + "sameAccountCredentialRetry", + ] { + kept[key] = json!(recovery.get(key).and_then(Value::as_bool)); + } + for key in ["intent", "continuationFailure", "continuationDisposition"] { + kept[key] = json!( + recovery + .get(key) + .and_then(Value::as_str) + .map(|value| bounded(value, 256)) + ); + } + kept["retryAfterMs"] = + json!(recovery.get("retryAfterMs").and_then(Value::as_u64)); + summary["recovery"] = kept; + } if let Some(diagnostic) = data.get("diagnostic").and_then(Value::as_object) { let mut kept = json!({}); for (key, limit) in [("stage", 96), ("code", 96), ("message", 256)] { @@ -359,6 +389,7 @@ impl TraceContext { | "account.selected" | "attempt.failed" | "retry.decided" + | "retry.policy" | "upstream.close" | "upstream.read.failed" | "capture.gap" diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index afdada6a3..9da9e9713 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -294,6 +294,21 @@ where "admissionMs": request.admission_decision_ms, "continuation": format!("{continuation_attempt:?}"), "candidateCount": plan.candidates().len(), + "accountPolicy": { + "strategy": plan.account_selection_policy().strategy().as_str(), + "affinity": plan.account_selection_policy().openai_account_affinity().as_str(), + "maxAccountRotations": plan.account_selection_policy().max_account_rotations(), + "maxRoutingAttempts": plan.max_attempts().get(), + "maxConcurrentPerAccount": plan.account_selection_policy().max_concurrent_per_account().get(), + "queueMaxWaiting": plan.account_selection_policy().queue_policy().max_waiting, + "queueTimeoutMs": plan.account_selection_policy().queue_policy().timeout.as_millis(), + "sessionAffinityTtlMs": plan.account_selection_policy().openai_session_affinity_ttl().as_millis(), + "smartWeights": plan.account_selection_policy().smart_scheduling().weights(), + "preferHigherWeight": plan.account_selection_policy().smart_scheduling().prefer_higher_weight(), + }, + "requiredAccount": account_selection.required_account().map(|account| account.as_str()), + "hasMiddleware": middleware.is_some(), + "hasRetryPolicy": request_policy.is_some(), }), ); let image_generation_requested = operation.image_generation_requested(); @@ -1200,6 +1215,17 @@ where retry_allowed: continuation_retry || candidate_retry, }) .await?; + attempt_trace.record("retry.decided", json!({ + "phase": "prepare", "retryable": retryable, + "continuationRetry": continuation_retry, "candidateRetry": candidate_retry, + "retryProhibited": error.retry_is_prohibited(), + "sendState": error.send_state().as_str(), + "requestSendState": self.current_send_state().as_str(), + "continuation": format!("{:?}", self.continuation_attempt), + "accountRotations": self.account_rotations, + "maxAccountRotations": self.plan.account_selection_policy().max_account_rotations(), + "remainingRoutingAttempts": self.plan.max_attempts().get().saturating_sub(self.routing_attempts), + })); if retryable { return Ok(Some(PullOutcome::AttemptDiscarded)); } @@ -1706,6 +1732,7 @@ where }; self.trace.attempt(current.index.get()).record("retry.decided", json!({ + "phase": "execute", "retryable": retryable, "continuationRetry": continuation_retry, "sameAccountRetry": same_account_retry, "accountRotationRetry": account_rotation_retry, "ordinaryRetry": ordinary_retry, "transportRecovery": transport_recovery.is_some(), @@ -1719,6 +1746,17 @@ where "delayMs": transient_retry.or(transport_recovery.map(|(_, delay)| delay)).map(duration_ms), "downstreamCommitted": self.response_commit.committed_at().is_some(), "replayBarrierReached": self.replay_barrier_reached, + "deliveryPending": self.delivery_pending, + "providerProvedReplaySafe": provider_proved_replay_safe, + "retryProhibited": error.retry_is_prohibited(), + "requiredAccount": self.account_selection.required_account().map(|account| account.as_str()), + "continuation": format!("{:?}", self.continuation_attempt), + "connectionBudgetExhausted": self.connection_budget.exhausted(), + "remainingRoutingAttempts": self.plan.max_attempts().get().saturating_sub(self.routing_attempts), + "maxAccountRotations": self.plan.account_selection_policy().max_account_rotations(), + "transientRetriesUsed": self.transient_retry_counts.get(current.metadata.provider_account_id()).copied().unwrap_or_default(), + "credentialRecoveryAttempted": self.credential_recovery_attempted_accounts.contains(current.metadata.provider_account_id()), + "hasRetryPolicy": self.request_policy.is_some(), "sendState": format!("{attempt_send_state:?}"), })); if retryable { @@ -1818,6 +1856,7 @@ where && !self.cancellation.is_cancelled() && !self.execution_effect_observed(); let allowed = facts.retry_allowed; + let attempt_index = facts.attempt_index; let decision = { let cancellation = self.cancellation.clone(); let cancelled = cancellation.cancelled().fuse(); @@ -1830,6 +1869,16 @@ where decision = decision => Ok(decision), } }?; + self.trace.attempt(attempt_index.get()).record( + "retry.policy", + json!({ + "hostAllowed": allowed, + "policyStopped": decision == super::policy::RetryDecision::Stop, + "executionEffectObserved": self.execution_effect_observed(), + "cancelled": self.cancellation.is_cancelled(), + "deadlineElapsed": self.deadline.is_elapsed(), + }), + ); // 策略只收窄宿主已允许的恢复,返回后仍复核期限和外部副作用 Ok(allowed && decision != super::policy::RetryDecision::Stop diff --git a/backend/crates/gateway-core/tests/diagnostics/trace.rs b/backend/crates/gateway-core/tests/diagnostics/trace.rs index f08fb6745..d4b897623 100644 --- a/backend/crates/gateway-core/tests/diagnostics/trace.rs +++ b/backend/crates/gateway-core/tests/diagnostics/trace.rs @@ -66,6 +66,8 @@ fn provider_failure_trace_keeps_classification_when_diagnostic_is_truncated() { assert_eq!(data["diagnostic"]["stage"], "prepare"); assert_eq!(data["diagnostic"]["code"], "account_proxy_invalid"); assert_eq!(data["diagnostic"]["truncated"], true); + assert_eq!(data["recovery"]["replaySafe"], false); + assert_eq!(data["recovery"]["retryProhibited"], false); assert!(!snapshot.to_string().contains("PRIVATE_RAW")); } diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 494298929..93b37ce6b 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -215,6 +215,19 @@ fn client_retry_should_skip_repeated_budget_and_clear_before_completed_delivery( let mut session = retry_session(&coordinator, id, operation.clone(), "key_client_1"); assert!(block_on(session.collect_uncommitted()).is_err()); assert_eq!(provider.contexts.lock().unwrap().len(), expected); + let trace = session.trace().snapshot().unwrap(); + let decision = trace["events"] + .as_array() + .unwrap() + .iter() + .rev() + .find(|event| event["stage"] == "retry.decided") + .unwrap(); + assert_eq!(decision["data"]["clientRetrySuppressed"], id == "req_retry"); + assert_eq!(decision["data"]["providerProvedReplaySafe"], true); + assert_eq!(decision["data"]["retryProhibited"], false); + assert_eq!(decision["data"]["replayBarrierReached"], false); + assert_eq!(decision["data"]["retryable"], false); } let mut recovered = retry_session( &coordinator, @@ -4197,7 +4210,7 @@ fn terminal_rejection_should_prevent_not_sent_rotation_and_explicit_transport_re error = error.with_pre_delivery_transport_fallback(); } assert!(error.stable_snapshot().retry_is_prohibited()); - let (coordinator, _, provider) = coordinator(vec![Script::Stream { + let (coordinator, store, provider) = coordinator(vec![Script::Stream { account_id: "acct_first", items: vec![Err(error)], }]); @@ -4215,6 +4228,27 @@ fn terminal_rejection_should_prevent_not_sent_rotation_and_explicit_transport_re Err(EngineError::Provider(_)) )); assert_eq!(provider.contexts.lock().unwrap().len(), 1); + let state = store.state.lock().unwrap(); + let trace: Value = serde_json::from_str( + state.finalizations[0] + .diagnostic_trace_json + .as_deref() + .unwrap(), + ) + .unwrap(); + let events = trace["events"].as_array().unwrap(); + let failed = events + .iter() + .find(|event| event["stage"] == "attempt.failed") + .unwrap(); + assert_eq!(failed["data"]["recovery"]["retryProhibited"], true); + let decision = events + .iter() + .find(|event| event["stage"] == "retry.decided") + .unwrap(); + assert_eq!(decision["data"]["phase"], "execute"); + assert_eq!(decision["data"]["retryable"], false); + assert_eq!(decision["data"]["retryProhibited"], true); } } @@ -5021,6 +5055,13 @@ fn no_eligible_account_before_stream_records_failure_without_fabricating_an_atte .iter() .any(|event| event["stage"] == "attempt.failed") ); + let decision = events + .iter() + .find(|event| event["stage"] == "retry.decided") + .unwrap(); + assert_eq!(decision["data"]["phase"], "prepare"); + assert_eq!(decision["data"]["retryable"], false); + assert_eq!(decision["data"]["requestSendState"], "not_sent"); assert_eq!(events.last().unwrap()["stage"], "request.finished"); } } @@ -6404,6 +6445,29 @@ fn configured_rotation_budget_limits_actual_account_changes() { .openai_session_affinity_ttl() == Duration::from_secs(168 * 3600) })); + let state = store.state.lock().unwrap(); + let trace: Value = serde_json::from_str( + state.finalizations[0] + .diagnostic_trace_json + .as_deref() + .unwrap(), + ) + .unwrap(); + let routed = trace["events"] + .as_array() + .unwrap() + .iter() + .find(|event| event["stage"] == "request.routed") + .unwrap(); + assert_eq!( + routed["data"]["accountPolicy"]["maxAccountRotations"], + budget + ); + assert_eq!( + routed["data"]["accountPolicy"]["sessionAffinityTtlMs"], + 168 * 3600 * 1000_u64 + ); + drop(state); assert_eq!( store.state.lock().unwrap().finalizations[0].attempt_count, budget + 1 diff --git a/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs b/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs index baa0e77bb..01cf2a049 100644 --- a/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs +++ b/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs @@ -400,6 +400,17 @@ impl PgObservabilityRepository { #[async_trait] impl AdminObservabilityStore for PgAdminObservabilityStore { + async fn diagnostic_errors( + &self, + target: &gateway_admin::model::diagnostics::DiagnosticTarget, + limit: u16, + ) -> AdminStoreResult> { + let target = target.clone(); + self.repository.query_budget.run(&self.repository.pool, "export diagnostic errors", |mut connection| async move { + diagnostic_errors(&mut connection, &target, limit).await + }).await.map_err(observability_error) + } + fn query_timeout(&self) -> std::time::Duration { self.repository.query_budget.query_timeout() } diff --git a/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs b/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs index dd72026a0..0ee803f5a 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs +++ b/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs @@ -104,6 +104,45 @@ pub(crate) async fn list_ops_errors( }) } +/// 精确故障导出复用错误投影,不经过列表时间窗口或聚合缓存 +pub(crate) async fn diagnostic_errors( + connection: &mut PgConnection, + target: &gateway_admin::model::diagnostics::DiagnosticTarget, + limit: u16, +) -> StoreResult> { + use gateway_admin::model::diagnostics::DiagnosticTarget; + require_nonempty("diagnostic target", "id", target.id())?; + validate_text(target.id(), MAX_FILTER_BYTES, "diagnostic target ID")?; + let mut statement = QueryBuilder::::new("select * from ("); + match target { + DiagnosticTarget::Request(id) => { + statement + .push(REQUEST_ERROR_SELECT) + .push(" and mr.id = ") + .push_bind(id) + .push(" and mr.error_kind is not null union all ") + .push(OPS_EVENT_SELECT) + .push(" and oe.model_request_id = ") + .push_bind(id); + } + DiagnosticTarget::Operation(id) => { + statement + .push(OPS_EVENT_SELECT) + .push(" and oe.id = ") + .push_bind(id); + } + } + statement + .push(") errors order by (source = 'model_request') desc, occurred_at desc, stable_sort_id desc limit ") + .push_bind(i64::from(limit)); + let rows = statement + .build() + .fetch_all(connection) + .await + .map_err(|source| postgres_unavailable("export diagnostic errors", source))?; + rows.iter().map(ops_error_from_row).collect() +} + async fn list_ops_error_items_at( connection: &mut PgConnection, query: &OpsErrorQuery, diff --git a/backend/crates/gateway-store/tests/postgres/observability/mod.rs b/backend/crates/gateway-store/tests/postgres/observability/mod.rs index c8fea5458..83c6b39f0 100644 --- a/backend/crates/gateway-store/tests/postgres/observability/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/observability/mod.rs @@ -2950,3 +2950,47 @@ fn assert_percentile(actual: Option, expected: Option) { values => assert_eq!(values.0, values.1), } } + +#[tokio::test] +async fn diagnostic_export_reads_exact_targets_and_prioritizes_final_error() { + use gateway_admin::model::diagnostics::DiagnosticTarget; + let Some(database) = TestDatabase::create("diagnostic_export").await else { + return; + }; + seed_observability_facts(&database.pool, Utc::now()) + .await + .unwrap(); + let store = admin_observability_store(&database.pool); + let target = DiagnosticTarget::Request("req_observe_failed".into()); + let errors = store.diagnostic_errors(&target, 129).await.unwrap(); + assert_eq!(errors.len(), 2); + assert!( + errors + .iter() + .all(|error| error.request_id.as_deref() == Some("req_observe_failed")) + ); + assert_eq!(errors[0].source, "model_request"); + assert_eq!( + store.diagnostic_errors(&target, 1).await.unwrap()[0].source, + "model_request" + ); + assert!( + store + .diagnostic_errors(&DiagnosticTarget::Request("unrelated".into()), 129) + .await + .unwrap() + .is_empty() + ); + sqlx::query("update ops_events set model_request_id = null, attempt_index = null where id = 'ops_observe_retry'") + .execute(&database.pool) + .await + .unwrap(); + let errors = store + .diagnostic_errors(&DiagnosticTarget::Operation("ops_observe_retry".into()), 1) + .await + .unwrap(); + assert_eq!(errors.len(), 1); + assert_eq!(errors[0].event_id, "ops_observe_retry"); + assert!(errors[0].request_id.is_none()); + database.close().await; +} diff --git a/docs/api.md b/docs/api.md index d30eccce8..c99936c88 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1991,28 +1991,32 @@ WebSocket 握手默认请求专项计时,客户端显式提供的开关保留 ### 诊断与恢复关联 -详情接口按 `id` 可读取成功、失败或未完成请求,返回 `trace`(未采集记录为 `null`)和 -`relatedRequests[]`(`requestId / relation / outcome / completedAt`);`relation` 为 `recovered_by` 或 -`recovers`。`trace` 是执行终态时的有界脱敏时间线,包含 request、attempt 和 exchange 关联、阶段、 -事件摘要及淘汰计数;普通用量列表不携带此字段。 -采集时,未知 JSON 键名与值只保留结构和摘要;事件摘要中的 `eventType` 为已知事件名称字符串、 -未知名称的 `{ bytes, sha256 }` 摘要,或缺失时的 `null`。已保存的 trace 不自动清理或回填, -其中的 `sanitized` 标记不能作为可直接公开的保证 +详情接口按 `id` 可读取成功、失败或未完成请求,普通详情与列表不传输时间线。 +管理员通过 `GET /api/admin/diagnostics/export` 下载 JSON 附件,必须且只能指定 +`requestId` 或 `errorId`;`errorId` 指运维错误事件 ID,无模型请求的后台错误也能导出。 +响应不使用普通 JSON 信封,设置 `Content-Disposition: attachment` 与 `Cache-Control: private, no-store` -`attempt.failed.data` 的 `kind / sendState / upstreamStatus` 与 `diagnostic.stage / diagnostic.code` -由 Core 错误类型与 Provider 静态诊断生成;`diagnostic.message` 保存安全摘要。展示与导出共用这些字段, -`sendState` 为 `not_sent / sent / ambiguous`,摘要被截断时带有 `truncated` 标记 +诊断包 `schemaVersion: 5` 用于故障分析,不是备份或导入格式。后端按目标 ID 收集请求、尝试、 +完整已存时间线、错误原因链、原始上游错误及恢复关联;前端只负责下载。 +`requests[].evidence` 保存详情、`trace`、`relatedRequests`、事发配置版本和尝试原因; +`requests[].errors` 保存同一请求的错误记录,`error` 保存按事件 ID 选中的错误,该事件不在请求内重复导出。 +`environment.scope = export_time` 明确版本、构建与部署信息是导出时状态,不能代替事发环境 -管理端下载的诊断包 `schemaVersion: 4` 用于人工反馈,不是备份或导入格式。它包含关联 ID、错误分类摘要、 -请求与错误事件各自的状态、attempt、时间线阶段、计时和上述失败分类。 -诊断包兼容已有 `provider.precommit.released` 记录,其 `precommitRelease` 保留 `reason / prefetchedBytes / waitMs`, -分别表示 Provider 释放暂存的原因、累计预取字节数和等待毫秒数,不等同于下游已提交。 -导出接受记录中的 `semantic_output / terminal / eof / grace_timeout / byte_limit`。 -数值只允许非负安全整数,缺失或未通过校验的字段为 `null`;其他事件的 `precommitRelease` 为 `null` +`request.routed.data.accountPolicy` 保存执行时冻结的账号策略与预算。 +`attempt.failed.data.recovery` 保存 Provider 的重放安全性和恢复意图;`retry.decided` 保存当时 +实际参与决策的门槛、剩余预算和发送状态,`retry.policy` 保存宿主及插件策略的停止事实。 +已有的 `provider.precommit.released` 数据保留释放原因、预取字节数与等待时间,不等同于下游已提交 -诊断包不自动导出 message/raw error、任意 metadata、其他 trace event data、请求响应正文和头部。 -`availability` 与 `omitted` 明示未采集、不完整或主动省略的内容,`null` 不代表没有发生错误。 -版本、环境及原始错误片段仍需操作者另行补充并审阅脱敏 +采集阶段仍按有界 trace 合同处理未知 JSON 键和值,未知事件名称保留摘要;导出不会补造被省略的原文。 +下载端按 Provider 凭据、管理认证与传输头合同移除令牌、密钥、认证头、Cookie 和 URL 认证部分, +移除账号联系信息及客户端 IP,保留关联 ID、错误 code/message、套餐、状态和重试事实。 +OAuth 字段依据 CPR 导出结构及官方 Codex `TokenData`,不按 `token / secret / credential` 等泛称删除业务字段。 +原始错误中可解析的嵌套 JSON 展开后脱敏;任意业务内容不属于凭据识别范围 + +收集共享 20 秒预算,最多包含 8 个关联请求和 128 条错误记录,优先保留主请求的最终错误。 +导出另有文本、节点与深度预算,正文截断仍保留格式版本、目标和缺口;超限会写入 `exportTruncated`、`redaction.truncatedValues` 和缺口记录。 +`availability.gaps` 明示未记录、淘汰、截断、读取失败与旧记录缺少的决策事实,`omitted` 说明主动省略范围。 +异步观测尚未落盘、已过期记录、未采集的请求正文与上游内部状态无法由导出补回,`null` 不表示没有发生错误 错误记录中的“已自动恢复”表示系统关联到了后续成功请求,不会把原来的失败记录改为成功。 `upstreamSendState = ambiguous` 表示无法确认该次上游执行结果,不代表后续恢复请求失败; diff --git a/docs/architecture.md b/docs/architecture.md index 9f67d2986..72caff425 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -570,7 +570,9 @@ OAuth 响应中的令牌字段和回显的刷新令牌在进入受控详情前 任意上游 code 或关闭原因。受控详情无法写入时不能回退到普通日志打印原文 Core 从执行错误生成一份失败诊断,展示与导出共用其中的分类;事件摘要截断时保留有界的失败诊断, -并显式标记截断。完整上游错误仍归运维错误记录,不进入诊断包 +并显式标记截断。完整上游错误归运维错误记录;Admin 按故障目标与关联请求收集已存证据, +API 在认证后的下载边界统一移除凭据并生成诊断文件,Vue 只发起下载。 +采集使用共享截止时间和记录上限,部分读取失败保留已有证据并报告缺口;当前环境与事发配置分别标记 Provider 管理适配在仍持有结构化失败事实时选择静态 `public_message`;Admin 将其转为可安全展示的 `AdminError.message`,API 保留其中的 502/503 具体原因,无公开消息时才使用通用回退。认证与未知 @@ -1033,8 +1035,9 @@ Compose 的 `stop_grace_period` 必须覆盖后台关闭预算并留出进程清 `attempt.failed` 保存这些字段,最终错误记录优先持久化诊断消息。重试包装不能覆盖底层诊断, 也不能因为错误更详细而改变已有重试安全边界 - 默认诊断区分 transport 采集的真实头部与用户 JSON;未知 map 的键和值均不原样保留, - 嵌套的同名 header 不获得头部白名单权限。管理端反馈包只导出明确允许的关联、分类、阶段和计时, - 不复制 trace 的事件 data,也不因历史 `sanitized` 标记而信任旧正文 + 嵌套的同名 header 不获得头部白名单权限。诊断下载保留已存 trace 事件 data 与错误原因, + 在服务端按实际凭据与认证合同重新脱敏,不因历史 `sanitized` 标记而信任旧正文。 + 普通详情不传输时间线,采集或导出丢失的信息必须明确标记,接口范围见 [诊断与恢复关联](api.md#诊断与恢复关联) - 真实 secret 不进入普通日志、Debug、fixture 或 audit details;明文只能通过账号导出、Key reveal、 备份设置等明确的敏感 Admin 合同返回 - OAuth pending flow 使用有期限、带 owner 的一次性 claim;事务成功后才消费,失败释放 claim diff --git a/frontend/src/api/modules/usage.ts b/frontend/src/api/modules/usage.ts index e1b71f503..acbf3e473 100644 --- a/frontend/src/api/modules/usage.ts +++ b/frontend/src/api/modules/usage.ts @@ -1,6 +1,6 @@ import type { RequestOptions } from '../request' import type { AnalyticsMetadata } from './analytics' -import request from '../request' +import request, { requestRaw } from '../request' export interface UsageTokenDetails { inputTokens: number | null @@ -205,50 +205,7 @@ export interface UsageAttempt { completedAt: string | null } -export interface RequestTraceEvent { - sequence: number - lastSequence: number - elapsedMs: number - lastElapsedMs: number - attemptIndex: number - exchangeId: number | null - stage: string - count: number - data: Record & { - /** attempt.failed 由 Core 从 ProviderError 生成,分类与诊断只有这一份来源 */ - kind?: string - sendState?: 'not_sent' | 'sent' | 'ambiguous' - upstreamStatus?: number | null - diagnostic?: { - stage: string | null - code: string | null - message: string - truncated?: boolean - } | null - } -} - -export interface RequestTrace { - schemaVersion: number - requestId: string - startedAtUnixMs: number - totalEvents: number - droppedEvents: number - maxEvents: number - captureMode: string - events: RequestTraceEvent[] -} - -export interface RelatedRequest { - requestId: string - relation: 'recovered_by' | 'recovers' - outcome: string - completedAt: string | null -} - export type UsageRecordDetail = UsageRecord & { - trace: RequestTrace | null - relatedRequests: RelatedRequest[] attempts: UsageAttempt[] /** 尝试列表是否完整;best-effort 下恒为 false。 */ attemptsComplete: boolean @@ -560,3 +517,16 @@ export function getUsageRecordInsightsDiagnostics(data: UsageDiagnosticsQuery, o ...options, }) } + +export function downloadDiagnostics(data: { requestId: string } | { errorId: string }, options: RequestOptions = {}) { + return requestRaw({ + url: '/api/admin/diagnostics/export', + method: 'GET', + params: data, + timeout: 25_000, + ...options, + }, { + maximumBytes: 16 * 1024 * 1024, + accept: response => response.status === 200 && !!response.header('content-type')?.startsWith('application/json'), + }) +} diff --git a/frontend/src/composables/useDownload.ts b/frontend/src/composables/useDownload.ts index 702546412..99eb8a532 100644 --- a/frontend/src/composables/useDownload.ts +++ b/frontend/src/composables/useDownload.ts @@ -6,9 +6,13 @@ export function useDownload() { const objectUrl = useObjectUrl(object) async function downloadJson(payload: unknown, fileName: string) { - object.value = new Blob([`${JSON.stringify(payload, null, 2)}\n`], { + return downloadBlob(new Blob([`${JSON.stringify(payload, null, 2)}\n`], { type: 'application/json;charset=utf-8', - }) + }), fileName) + } + + async function downloadBlob(blob: Blob, fileName: string) { + object.value = blob await nextTick() if (!objectUrl.value) @@ -24,5 +28,6 @@ export function useDownload() { return { downloadJson, + downloadBlob, } } diff --git a/frontend/src/views/usage/components/DiagnosticsDownloadButton.vue b/frontend/src/views/usage/components/DiagnosticsDownloadButton.vue new file mode 100644 index 000000000..2562f99be --- /dev/null +++ b/frontend/src/views/usage/components/DiagnosticsDownloadButton.vue @@ -0,0 +1,49 @@ + + + diff --git a/frontend/src/views/usage/components/OpsErrorDetailModal.vue b/frontend/src/views/usage/components/OpsErrorDetailModal.vue index 4123396e7..180da48dd 100644 --- a/frontend/src/views/usage/components/OpsErrorDetailModal.vue +++ b/frontend/src/views/usage/components/OpsErrorDetailModal.vue @@ -4,7 +4,7 @@ import type { OpsError } from '@/api' import { ZButton, ZDialog } from '@codex-proxy/ui' import { computed } from 'vue' import { failureClassText } from '../utils/opsErrorPresentation' -import RequestDiagnosticsPanel from './RequestDiagnosticsPanel.vue' +import DiagnosticsDownloadButton from './DiagnosticsDownloadButton.vue' import UsageDetailCodePanel from './UsageDetailCodePanel.vue' import UsageDetailFieldGrid from './UsageDetailFieldGrid.vue' @@ -129,7 +129,6 @@ function visibleFields(items: DetailField[]) { width="64rem" > diff --git a/frontend/src/views/usage/components/UsageRecordDetailModal.vue b/frontend/src/views/usage/components/UsageRecordDetailModal.vue index 3203689c5..8e099cc9d 100644 --- a/frontend/src/views/usage/components/UsageRecordDetailModal.vue +++ b/frontend/src/views/usage/components/UsageRecordDetailModal.vue @@ -19,7 +19,7 @@ import { import { useChartPalette } from '@/composables/useChartPalette' import { formatDuration, formatInteger } from '@/utils/format' import { displayValue, fieldLabelClass, fieldValueBaseClass, fieldValueClass, visibleRequestText, visibleResponseText } from '../utils/detail' -import RequestDiagnosticsPanel from './RequestDiagnosticsPanel.vue' +import DiagnosticsDownloadButton from './DiagnosticsDownloadButton.vue' import UsageDetailCodePanel from './UsageDetailCodePanel.vue' import UsageDetailFieldGrid from './UsageDetailFieldGrid.vue' import UsageStatusCodeBadge from './UsageStatusCodeBadge.vue' @@ -510,9 +510,8 @@ const tokenDonutOption = computed(() => { - -