feat(diagnostics): generate support bundles on the backend

Collect request, retry, error and recovery evidence with bounded reads and
credential redaction. Keep ordinary detail responses small and expose a
shared download action next to the close button.

Preserve export metadata when limits are reached and deduplicate selected
errors without discarding business context based on generic field names.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-16 12:49:18 +08:00
co-authored by Codex
parent 7b8339ede3
commit a8a96d8919
40 changed files with 1505 additions and 568 deletions
+1
View File
@@ -1746,6 +1746,7 @@ dependencies = [
"http",
"http-body",
"http-body-util",
"regex",
"secrecy",
"serde",
"serde_json",
+1
View File
@@ -521,6 +521,7 @@ async fn initialize_inner(
store.settings(),
registry.clone(),
timezone,
system.clone(),
)),
settings,
usage_data,
@@ -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<UsageDetail>,
pub errors: Vec<OpsError>,
}
/// 原始受控详情只能经导出投影脱敏后交付,不参与普通日志或 Debug
pub struct DiagnosticBundle {
pub target: DiagnosticTarget,
pub collected_at: DateTime<Utc>,
pub environment: Option<SystemVersion>,
pub error: Option<OpsError>,
pub requests: Vec<DiagnosticRequest>,
pub gaps: Vec<DiagnosticGap>,
}
@@ -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;
@@ -485,6 +485,13 @@ pub trait ObservabilityStore: Send + Sync {
) -> AdminStoreResult<DiagnosticsObservation>;
async fn list_ops_errors(&self, query: OpsErrorQuery) -> AdminStoreResult<OpsErrorPage>;
/// 按精确目标读取受控错误,不受列表时间窗口与当前页面影响
async fn diagnostic_errors(
&self,
target: &crate::model::diagnostics::DiagnosticTarget,
limit: u16,
) -> AdminStoreResult<Vec<crate::model::observability::OpsError>>;
}
/// Runtime settings 与管理员 API Key 写入
@@ -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<DiagnosticBundle, AdminError> {
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<DiagnosticBundle, AdminError> {
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, &section, "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, &section, "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, &section, "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<Vec<OpsError>, 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,
});
}
}
@@ -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<DiagnosticsResult, AdminError>;
async fn ops_errors(&self, query: OpsErrorQuery) -> Result<OpsErrorPage, AdminError>;
async fn export_diagnostics(
&self,
target: crate::model::diagnostics::DiagnosticTarget,
) -> Result<crate::model::diagnostics::DiagnosticBundle, AdminError>;
}
pub(crate) struct DefaultObservabilityService {
@@ -78,6 +83,7 @@ pub(crate) struct DefaultObservabilityService {
accounts: Arc<dyn AccountStore>,
settings: Arc<dyn SettingsStore>,
providers: ProviderAdminRegistry,
system: Arc<dyn super::system::SystemService>,
}
impl DefaultObservabilityService {
@@ -88,6 +94,7 @@ impl DefaultObservabilityService {
settings: Arc<dyn SettingsStore>,
providers: ProviderAdminRegistry,
timezone: gateway_core::time::DeploymentTimeZone,
system: Arc<dyn super::system::SystemService>,
) -> 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<crate::model::diagnostics::DiagnosticBundle, AdminError> {
self.collect_diagnostics(target).await
}
async fn dashboard_summary(
&self,
range: TimeRange,
@@ -289,4 +289,12 @@ impl ObservabilityStore for CachedObservabilityStore {
async fn list_ops_errors(&self, query: OpsErrorQuery) -> AdminStoreResult<OpsErrorPage> {
self.inner.list_ops_errors(query).await
}
async fn diagnostic_errors(
&self,
target: &crate::model::diagnostics::DiagnosticTarget,
limit: u16,
) -> AdminStoreResult<Vec<crate::model::observability::OpsError>> {
self.inner.diagnostic_errors(target, limit).await
}
}
@@ -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<Vec<gateway_admin::model::observability::OpsError>> {
Err(unavailable("diagnostic errors"))
}
fn query_timeout(&self) -> std::time::Duration {
std::time::Duration::from_secs(10)
}
@@ -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<Vec<gateway_admin::model::observability::OpsError>> {
Err(super::unavailable("diagnostic errors"))
}
fn query_timeout(&self) -> std::time::Duration {
std::time::Duration::from_secs(10)
}
+1
View File
@@ -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"] }
@@ -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<String>,
error_id: Option<String>,
}
pub(super) async fn download<S>(
_auth: AdminAuth,
State(state): State<S>,
AdminQuery(query): AdminQuery<ExportQuery>,
) -> Result<impl IntoResponse, AdminError>
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::<Vec<_>>();
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::<Vec<_>>(),
})
}).collect::<Vec<_>>();
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::<Vec<_>>()},
"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
}
@@ -0,0 +1,220 @@
//! 诊断文件按认证与观测合同移除敏感字段,保留错误事实并明确导出截断
use regex::Regex;
use serde::Serialize;
use serde_json::{Value, json};
use std::sync::LazyLock;
const REDACTED: &str = "<redacted>";
const MAX_TEXT: usize = 256 * 1024;
static INLINE: LazyLock<Vec<(Regex, &str)>> = LazyLock::new(|| {
[
(r"(?i)\b((?:set-cookie|cookie2?)\s*:\s*)[^\r\n]+", "${1}<redacted>"),
(r#"(?i)\b(Bearer|Basic)\s+[^\s"'<>;,]+"#, "${1} <redacted>"),
// 官方 TokenData 的 access_token / id_token 为 JWT,不按猜测的令牌前缀过滤普通文本
(r"\beyJ[\w-]+\.[\w-]+\.[\w-]+\b", "<redacted>"),
(r"(?i)([a-z][a-z0-9+.-]*://)[^\s/@]+@", "${1}<redacted>@"),
(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}<redacted>"),
].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::<Value>(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;
}
}
_ => {}
}
}
}
@@ -16,6 +16,7 @@ use super::{
wire::map_admin_service_error,
};
mod export;
mod presenter;
mod query;
mod routes;
@@ -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::<Value>(&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::<Value>(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,
@@ -29,6 +29,10 @@ where
get(usage_insights_diagnostics::<S>),
)
.route("/api/admin/operations/errors", get(ops_errors::<S>))
.route(
"/api/admin/diagnostics/export",
get(super::export::download::<S>),
)
}
pub(crate) async fn dashboard_summary<S>(
@@ -275,8 +275,6 @@ pub struct UsageAttemptView {
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UsageRecordDetailView {
pub trace: Option<serde_json::Value>,
pub related_requests: Vec<serde_json::Value>,
#[serde(flatten)]
pub request: UsageRecordView,
pub attempts: Vec<UsageAttemptView>,
+23 -1
View File
@@ -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<Vec<gateway_admin::model::observability::OpsError>> {
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<UsageDetail> {
async fn usage_record_detail(&self, request_id: &str) -> AdminStoreResult<UsageDetail> {
self.usage_detail
.lock()
.expect("usage detail")
.clone()
.filter(|detail| detail.request.id == request_id)
.ok_or_else(|| unavailable("usage detail"))
}
@@ -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::<AdminTestState>()
.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")
);
}
@@ -1,4 +1,5 @@
//! Admin 观测 wire/路由的固定合同测试
mod export;
mod query;
mod response;
@@ -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::<AdminTestState>()
.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(),
}
}
@@ -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"
@@ -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
@@ -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"));
}
@@ -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
@@ -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<Vec<admin_observability::OpsError>> {
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()
}
@@ -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<Vec<OpsErrorRecord>> {
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::<Postgres>::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,
@@ -2950,3 +2950,47 @@ fn assert_percentile(actual: Option<f64>, expected: Option<f64>) {
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;
}
+23 -19
View File
@@ -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` 表示无法确认该次上游执行结果,不代表后续恢复请求失败;
+6 -3
View File
@@ -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
+14 -44
View File
@@ -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<string, unknown> & {
/** 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'),
})
}
+7 -2
View File
@@ -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,
}
}
@@ -0,0 +1,49 @@
<script setup lang="ts">
import { ZButton } from '@codex-proxy/ui'
import { Download } from '@lucide/vue'
import { onScopeDispose, watch } from 'vue'
import { downloadDiagnostics } from '@/api'
import { useAsyncAction } from '@/composables/useAsyncAction'
import { useDownload } from '@/composables/useDownload'
const props = defineProps<{ requestId?: string | null, errorId?: string | null, active: boolean }>()
const { downloadBlob } = useDownload()
const { loading, run } = useAsyncAction()
let controller: AbortController | undefined
watch(() => [props.requestId, props.errorId, props.active], () => controller?.abort())
onScopeDispose(() => controller?.abort())
function download() {
const id = props.requestId || props.errorId
if (!id || loading.value || !props.active)
return
const target = props.requestId ? { requestId: props.requestId } : { errorId: id }
const current = new AbortController()
controller = current
return run(async () => {
try {
const file = await downloadDiagnostics(target, { signal: current.signal })
if (!current.signal.aborted)
await downloadBlob(new Blob([file.body], { type: file.contentType }), `diagnostics-${id.replace(/[^\w-]/g, '_')}.json`)
}
catch (cause) {
if (!current.signal.aborted)
throw cause
}
finally {
if (controller === current)
controller = undefined
}
}, { errorText: '下载诊断包失败', minimumMs: 400 })
}
</script>
<template>
<ZButton type="primary" variant="plain" :loading="loading" :disabled="!requestId && !errorId" @click="download">
<template #icon>
<Download :size="16" />
</template>
{{ loading ? '正在生成诊断包' : '下载诊断包' }}
</ZButton>
</template>
@@ -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"
>
<template v-if="record">
<RequestDiagnosticsPanel v-if="record.requestId" class="mb-3" :request-id="record.requestId" :metadata="record.metadata" :error-record="record" :active="open" />
<section :class="panelClass">
<h3 :class="panelTitleClass">
错误
@@ -180,6 +179,7 @@ function visibleFields(items: DetailField[]) {
</template>
<template #footer>
<DiagnosticsDownloadButton v-if="record" :request-id="record.requestId" :error-id="record.id" :active="open" />
<ZButton type="primary" @click="open = false">
关闭
</ZButton>
@@ -1,107 +0,0 @@
<script setup lang="ts">
import type { OpsError, OpsErrorMetadata } from '@/api'
import { ZButton, ZEmpty, ZScrollbar } from '@codex-proxy/ui'
import { Download, RefreshCw } from '@lucide/vue'
import { computed } from 'vue'
import { useAsyncAction } from '@/composables/useAsyncAction'
import { useDownload } from '@/composables/useDownload'
import { useRequestDiagnostics } from '../composables/useRequestDiagnostics'
import { requestDiagnosticsBundle } from '../utils/diagnosticsBundle'
import RequestTransportFailure from './RequestTransportFailure.vue'
import UsageDetailCodePanel from './UsageDetailCodePanel.vue'
const props = withDefaults(defineProps<{ requestId: string, metadata?: OpsErrorMetadata, errorRecord?: OpsError, active?: boolean }>(), { active: true })
const { downloadJson } = useDownload()
const { loading: exporting, run: runExport } = useAsyncAction()
const { selectedId, detail, loading, error, refresh } = useRequestDiagnostics(() => props.requestId, () => props.active)
const trace = computed(() => detail.value?.trace)
const events = computed(() => (trace.value?.events ?? []).map(event => ({
...event,
content: JSON.stringify(event.data, null, 2),
})))
const selectedError = computed(() => props.errorRecord?.requestId === selectedId.value ? props.errorRecord : undefined)
const canExport = computed(() => !loading.value && (detail.value?.requestId === selectedId.value || !!selectedError.value))
function download() {
if (!canExport.value || exporting.value)
return
const bundle = requestDiagnosticsBundle(selectedId.value, detail.value, selectedError.value)
return runExport(
() => downloadJson(bundle, `diagnostics-${selectedId.value.replace(/[^\w-]/g, '_')}.json`),
{ errorText: '导出诊断包失败', minimumMs: 400 },
)
}
</script>
<template>
<section class="mt-3 min-w-0 rounded-cp-card bg-cp-fill-quaternary px-4 py-3.5" aria-label="请求诊断">
<div class="flex flex-wrap items-center justify-between gap-3">
<h3 class="m-0 text-cp-sm font-heavy text-cp-text-secondary">
请求诊断
</h3>
<div class="flex flex-wrap gap-2">
<ZButton v-if="selectedId !== requestId" type="primary" variant="plain" size="small" @click="selectedId = requestId">
返回本次请求
</ZButton>
<ZButton type="primary" variant="plain" size="small" :loading="loading" @click="refresh">
<template #icon>
<RefreshCw :size="14" />
</template>
刷新
</ZButton>
<ZButton type="primary" variant="plain" size="small" :loading="exporting" :disabled="!canExport" @click="download">
<template #icon>
<Download :size="14" />
</template>
导出诊断包
</ZButton>
</div>
</div>
<p class="mt-1 mb-3 break-all font-mono text-cp-xs leading-relaxed text-cp-text-secondary">
{{ selectedId }}
</p>
<p v-if="loading" role="status" class="text-cp-sm text-cp-text-secondary">
正在加载诊断记录…
</p>
<p v-else-if="error" role="alert" class="text-cp-sm text-cp-error-text">
{{ error }}
</p>
<template v-else-if="detail">
<div v-if="detail.relatedRequests?.length" class="mb-3 flex flex-wrap gap-2">
<ZButton v-for="related in detail.relatedRequests" :key="related.requestId" type="primary" variant="plain" size="small" class="max-w-full" @click="selectedId = related.requestId">
{{ related.relation === 'recovered_by' ? '查看恢复请求' : '查看先前失败' }} · {{ related.requestId }}
</ZButton>
</div>
<RequestTransportFailure :events="trace?.events ?? []" :metadata="selectedId === requestId ? metadata : undefined" />
<ZEmpty v-if="!events.length" title="暂无诊断时间线" size="small" surface="none" />
<template v-else-if="trace">
<div class="mb-3 grid gap-1 text-cp-xs leading-relaxed">
<p class="m-0 text-cp-text-secondary">
已观测 {{ trace.totalEvents }} 个阶段或事件,展示 {{ events.length }} 条记录
</p>
<p v-if="trace.droppedEvents" role="status" class="m-0 text-cp-warning-text">
已达保存上限,{{ trace.droppedEvents }} 个事件未保留
</p>
</div>
<ZScrollbar max-height="32rem">
<ol class="m-0 grid list-none gap-2 p-0 pr-3">
<li v-for="event in events" :key="event.sequence" class="min-w-0">
<details class="group overflow-hidden rounded-cp bg-cp-bg-container">
<summary class="cursor-pointer px-3 py-2.5 break-all text-cp-xs leading-relaxed transition-colors hover:bg-cp-primary-container group-open:bg-cp-primary-container focus-visible:outline-2 focus-visible:-outline-offset-2 focus-visible:outline-cp-primary motion-reduce:transition-none">
<span class="font-mono tabular-nums text-cp-text-secondary">+{{ event.elapsedMs }} ms · #{{ event.sequence }}</span>
<span class="mx-2 font-mono text-cp-text">{{ event.stage }}</span>
<span v-if="event.attemptIndex" class="text-cp-text-secondary">尝试 {{ event.attemptIndex }}</span>
<span v-if="event.exchangeId" class="text-cp-text-secondary"> · 交换 {{ event.exchangeId }}</span>
<span v-if="event.count > 1" class="text-cp-text-secondary"> · {{ event.count }} 次(至 +{{ event.lastElapsedMs }} ms)</span>
</summary>
<div class="mx-3 py-3">
<UsageDetailCodePanel title="诊断事实" :content="event.content" max-height="280px" />
</div>
</details>
</li>
</ol>
</ZScrollbar>
</template>
</template>
</section>
</template>
@@ -1,115 +0,0 @@
<script setup lang="ts">
import type { OpsErrorMetadata, RequestTraceEvent } from '@/api'
import { CircleAlert } from '@lucide/vue'
import { computed } from 'vue'
const props = defineProps<{ events: RequestTraceEvent[], metadata?: OpsErrorMetadata }>()
const reasonLabels: Record<string, string> = {
reset_without_closing_handshake: '上游连接中断,未收到 WebSocket 关闭握手',
tcp_reset: '上游 TCP 连接被重置',
connection_aborted: '上游连接被中止',
broken_pipe: '写入失败,上游连接已断开',
unexpected_eof: '上游连接意外结束',
transport_timeout: '上游连接传输超时',
receive_idle_timeout: '等待上游事件超时',
tls_error: '上游 TLS 连接异常',
protocol_error: '上游 WebSocket 协议异常',
capacity_error: '上游 WebSocket 数据超出容量限制',
io_error: '上游连接读写失败',
transport_error: '上游连接传输失败',
outbound_transport_error: '向上游连接发送请求失败',
}
function text(value: unknown) {
return typeof value === 'string' && value ? value : undefined
}
function duration(value: unknown) {
return typeof value === 'number' && Number.isFinite(value) && value >= 0
? `${(value / 1000).toLocaleString('zh-CN', { maximumFractionDigits: 2 })} 秒`
: undefined
}
const diagnosis = computed(() => {
const newestFirst = [...props.events].reverse()
const attempt = newestFirst.find(event => event.stage === 'attempt.failed')
const rawDiagnostic = attempt?.data.diagnostic
const diagnostic = rawDiagnostic && typeof rawDiagnostic === 'object'
? rawDiagnostic as Record<string, unknown>
: undefined
const failure = newestFirst.find(event => [
'upstream.read.failed',
'upstream.send.failed',
'upstream.eof',
'upstream.close',
].includes(event.stage) && (!attempt || event.attemptIndex === attempt.attemptIndex))
const connection = failure && newestFirst.find(event => event.stage === 'upstream.connection'
&& event.sequence < failure.sequence
&& event.attemptIndex === failure.attemptIndex
&& event.exchangeId === failure.exchangeId)
const metadata = props.metadata
const connectionId = text(failure?.data.connectionId) ?? text(connection?.data.connectionId)
?? metadata?.upstreamConnectionId
const storedReason = metadata?.upstreamConnectionExitReason
const preciseReason = text(diagnostic?.code) ?? text(failure?.data.failureReason)
const reason = preciseReason ?? storedReason
const message = text(diagnostic?.message)
if (!failure && !message && !storedReason)
return null
const reused = failure?.data.reused ?? connection?.data.reused
const closeCode = failure?.data.code
const fallback = failure?.stage === 'upstream.close'
? `上游在响应完成前关闭连接${typeof closeCode === 'number' ? `(关闭码 ${closeCode})` : ''}`
: failure?.stage === 'upstream.eof'
? '上游连接在响应完成前结束'
: failure?.stage === 'upstream.send.failed' ? '向上游发送请求失败' : '读取上游响应失败'
return {
message: message ?? (!preciseReason && storedReason === 'tcp_reset'
? '上游连接异常结束(旧版汇总分类,无法区分 TCP 重置与缺失关闭握手)'
: reason ? reasonLabels[reason] ?? fallback : fallback),
reason,
sequence: (attempt ?? failure)?.sequence,
elapsed: duration((attempt ?? failure)?.elapsedMs),
fields: [
{ label: '失败阶段', value: text(diagnostic?.stage) },
{ label: '连接使用', value: reused === true ? '复用已有连接' : reused === false ? '新建连接' : undefined },
{ label: '连接年龄', value: duration(failure?.data.connectionAgeMs ?? metadata?.upstreamConnectionAgeMs) },
{ label: '距最近活动', value: duration(failure?.data.connectionIdleMs ?? metadata?.upstreamConnectionIdleMs) },
{ label: '最后事件', value: failure?.data.lastEventType === null ? '尚未收到事件' : text(failure?.data.lastEventType) },
{ label: '连接 ID', value: connectionId },
].filter(field => field.value !== undefined),
}
})
</script>
<template>
<aside v-if="diagnosis" class="mb-3 rounded-cp bg-cp-warning-container p-3" aria-label="上游失败诊断">
<div class="flex items-start gap-2">
<CircleAlert :size="16" class="mt-0.5 shrink-0 text-cp-warning-text" aria-hidden="true" />
<div class="min-w-0 flex-1">
<p class="m-0 text-cp-sm font-bold text-cp-text">
{{ diagnosis.message }}
</p>
<p class="mt-1 mb-0 break-all font-mono text-cp-xs tabular-nums text-cp-text-secondary">
<span v-if="diagnosis.sequence !== undefined">#{{ diagnosis.sequence }} · +{{ diagnosis.elapsed }} · </span>{{ diagnosis.reason }}
</p>
</div>
</div>
<dl v-if="diagnosis.fields.length" class="mt-3 mb-0 grid gap-x-5 gap-y-2 text-cp-xs sm:grid-cols-2">
<div v-for="field in diagnosis.fields" :key="field.label" class="min-w-0">
<dt class="text-cp-text-secondary">
{{ field.label }}
</dt>
<dd class="mt-0.5 ml-0 break-all font-mono leading-relaxed text-cp-text">
{{ field.value }}
</dd>
</div>
</dl>
<p v-if="!diagnosis.reason" class="mt-3 mb-0 text-cp-xs leading-relaxed text-cp-text-secondary">
这条记录未保存底层错误分类,需用请求 ID 检索留存的服务器日志
</p>
</aside>
</template>
@@ -19,7 +19,7 @@ const codeBlockClass
</h3>
<ZScrollbar :max-height="maxHeight">
<div class="rounded-cp bg-cp-bg-container px-3 py-2.5">
<pre :class="codeBlockClass">{{ content }}</pre>
<pre :class="codeBlockClass"><code>{{ content }}</code></pre>
</div>
</ZScrollbar>
</template>
@@ -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<EChartsOption>(() => {
</section>
</div>
<RequestDiagnosticsPanel v-if="record" :request-id="record.requestId" :active="open" />
<template #footer>
<DiagnosticsDownloadButton v-if="record" :request-id="record.requestId" :active="open" />
<ZButton type="primary" @click="open = false">
关闭
</ZButton>
@@ -1,43 +0,0 @@
import type { UsageRecordDetail } from '@/api'
import { shallowRef, watch } from 'vue'
import { getUsageRecordDetail } from '@/api'
import { errorMessage } from '@/utils/operation'
// 切换请求或开始退场时取消旧查询,退场期间保留最后一次诊断画面。
export function useRequestDiagnostics(requestId: () => string, isActive: () => boolean = () => true) {
const selectedId = shallowRef(requestId())
const revision = shallowRef(0)
const detail = shallowRef<UsageRecordDetail | null>(null)
const loading = shallowRef(false)
const error = shallowRef('')
watch(requestId, id => selectedId.value = id)
watch([selectedId, revision, isActive], async ([id, , enabled], _previous, onCleanup) => {
if (!enabled)
return
let active = true
const controller = new AbortController()
onCleanup(() => {
active = false
controller.abort()
})
detail.value = null
error.value = ''
loading.value = true
try {
const result = await getUsageRecordDetail({ id }, { signal: controller.signal })
if (active)
detail.value = result
}
catch (cause: unknown) {
if (active)
error.value = errorMessage(cause)
}
finally {
if (active)
loading.value = false
}
}, { immediate: true })
return { selectedId, detail, loading, error, refresh: () => revision.value++ }
}
@@ -1,159 +0,0 @@
import type { OpsError, UsageRecordDetail } from '@/api'
import { failureClassText } from './opsErrorPresentation'
const precommitReleaseReasons = new Set(['semantic_output', 'terminal', 'byte_limit', 'grace_timeout', 'eof'])
function nonNegativeInteger(value: unknown) {
return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 ? value : null
}
export function requestDiagnosticsBundle(
requestId: string,
detail: UsageRecordDetail | null,
errorRecord?: OpsError,
) {
// 关联请求切换后,不能把原弹窗的错误归到另一个请求。
const record = detail?.requestId === requestId ? detail : null
const error = errorRecord?.requestId === requestId ? errorRecord : null
const trace = record?.trace
// 请求终态与错误事件可能来自不同尝试,不用事件的值填补终态空字段。
const request = record ?? error
return {
schemaVersion: 4,
source: 'usage.request_diagnostics',
exportPolicy: 'allowlisted_fields_without_payloads',
request: {
requestId,
source: record ? 'request_detail' : error ? 'error_record' : 'request_id_only',
route: request?.route ?? null,
provider: request?.provider ?? null,
accountId: request?.accountId ?? null,
clientTransport: request?.clientTransport ?? null,
upstreamTransport: record ? record.upstreamTransport : error?.transport ?? null,
upstreamRequestId: request?.upstreamRequestId ?? null,
responseId: request?.responseId ?? null,
statusCode: record?.statusCode ?? null,
clientStatusCode: request?.clientStatusCode ?? null,
upstreamStatusCode: request?.upstreamStatusCode ?? null,
logicalOutcome: record?.logicalOutcome ?? null,
attemptCount: record?.attemptCount ?? null,
createdAt: request?.createdAt ?? null,
},
error: error
? {
id: error.id,
requestId: error.requestId,
source: error.metadata.source,
component: error.metadata.component,
summary: failureClassText(error.failureClass),
summarySource: 'failure_class',
failureClass: error.failureClass,
upstreamSendState: error.upstreamSendState,
attemptIndex: error.attemptIndex,
clientStatusCode: error.clientStatusCode,
upstreamStatusCode: error.upstreamStatusCode,
upstreamRequestId: error.upstreamRequestId,
responseId: error.responseId,
latencyMs: error.latencyMs,
createdAt: error.createdAt,
}
: null,
// 只导出失败分类与缓存释放事实的白名单,正文与任意事件 data 不进入诊断包
trace: trace
? {
schemaVersion: trace.schemaVersion,
startedAtUnixMs: trace.startedAtUnixMs,
totalEvents: trace.totalEvents,
droppedEvents: trace.droppedEvents,
events: trace.events.map(event => ({
sequence: event.sequence,
lastSequence: event.lastSequence,
elapsedMs: event.elapsedMs,
lastElapsedMs: event.lastElapsedMs,
attemptIndex: event.attemptIndex,
exchangeId: event.exchangeId,
stage: event.stage,
count: event.count,
failure: event.stage === 'attempt.failed'
? {
kind: event.data.kind,
sendState: event.data.sendState,
stage: event.data.diagnostic?.stage ?? null,
code: event.data.diagnostic?.code ?? null,
upstreamStatus: event.data.upstreamStatus,
}
: null,
precommitRelease: event.stage === 'provider.precommit.released'
? {
reason: typeof event.data.reason === 'string' && precommitReleaseReasons.has(event.data.reason)
? event.data.reason
: null,
prefetchedBytes: nonNegativeInteger(event.data.prefetchedBytes),
waitMs: nonNegativeInteger(event.data.waitMs),
}
: null,
})),
}
: null,
attempts: record?.attempts.map(attempt => ({
id: attempt.id,
attemptIndex: attempt.attemptIndex,
trigger: attempt.trigger,
provider: attempt.provider,
transport: attempt.transport,
sendState: attempt.sendState,
outcome: attempt.outcome,
downstreamCommitted: attempt.downstreamCommitted,
statusCode: attempt.statusCode,
failureClass: attempt.failureClass,
accountId: attempt.accountId,
firstTokenMs: attempt.firstTokenMs,
latencyMs: attempt.latencyMs,
startedAt: attempt.startedAt,
completedAt: attempt.completedAt,
})) ?? [],
relatedRequests: record?.relatedRequests.map(related => ({
requestId: related.requestId,
relation: related.relation,
outcome: related.outcome,
completedAt: related.completedAt,
})) ?? [],
availability: {
requestDetail: record ? 'available' : 'unavailable',
errorSummary: error ? 'available' : 'not_available_for_selected_request',
trace: !record
? 'unknown'
: !trace
? 'not_recorded'
: trace.events.some(event => event.stage === 'attempt.failed' || event.stage === 'provider.precommit.released')
? 'stages_and_allowlisted_facts'
: 'stages_only',
traceEventsDropped: trace?.droppedEvents ?? null,
attemptsComplete: record?.attemptsComplete ?? null,
relatedRequests: record ? 'available' : 'unknown',
environment: 'not_collected',
},
omitted: [
'error.message',
'error.errorDetails',
'error.providerErrorCode',
'attempts.providerErrorCode',
'request.message',
'request.metadata',
'trace.events.data_except_failure_classification_and_precommit_release',
'request_and_response_headers_and_bodies',
'user_supplied_model_names',
'account_names_emails_and_credential_names',
'client_ip_user_agent_and_api_key',
],
notes: [
'schemaVersion 4 仅供人工排障,null 表示未知、未采集或未通过导出校验,不代表没有发生错误',
'错误摘要只使用稳定分类,不含上游错误码或原文,error 为 null 时请结合 attempts 的分类与状态',
'时间线导出阶段、顺序、计时、Core 记录的失败分类和 Provider 缓存释放的原因、预取字节数及等待毫秒数,不含错误原文和其他事件 data',
'attemptsComplete 不为 true 时,尝试列表不完整,没有时间线的旧记录无法补回未采集事件',
'未自动采集网关版本、客户端版本、部署环境和故障发生时区,请另行补充',
'关联 ID 仍可能属于内部信息,分享前请审阅,原始错误、正文和日志需另行审阅脱敏,勿直接公开',
],
}
}