fix(diagnostics): align trace snapshot persistence limits

Co-authored-by: Codex <noreply@openai.com>

backend/crates/gateway-core/src/diagnostics/mod.rs

@@ -9,4 +9,4 @@ mod trace;
 pub use capture::{body_fingerprint, diagnostic_headers, diagnostic_json, json_fingerprint};
 pub use failure::{OperationalDiagnostics, OperationalFailure};
 pub use stream::{StreamCapture, StreamFormat};
-pub use trace::TraceContext;
+pub use trace::{MAX_TRACE_SNAPSHOT_BYTES, TraceContext};

backend/crates/gateway-core/src/diagnostics/trace.rs

@@ -15,13 +15,16 @@ use crate::error::ProviderError;
 use super::capture::{bounded, diagnostic_event_json, diagnostic_event_type};
 use super::{body_fingerprint, diagnostic_headers};
 
+/// 完整诊断快照的编码上限，生产者与持久化适配器共用
+pub const MAX_TRACE_SNAPSHOT_BYTES: usize = 128 * 1024;
+
 const MAX_EVENTS: usize = 128;
 const HEAD_EVENTS: usize = 8;
 const MAX_DATA_BYTES: usize = 4096;
-const MAX_BUFFER_BYTES: usize = 64 * 1024;
+const MAX_BUFFER_BYTES: usize = MAX_TRACE_SNAPSHOT_BYTES / 2;
 const MAX_EVIDENCE_EVENTS: usize = 64;
 const MAX_EVIDENCE_DATA_BYTES: usize = 16 * 1024;
-const MAX_EVIDENCE_BUFFER_BYTES: usize = 64 * 1024;
+const MAX_EVIDENCE_BUFFER_BYTES: usize = MAX_TRACE_SNAPSHOT_BYTES / 2;
 
 /// 显式传给异步任务的关联上下文；默认值禁用捕获，避免全局请求映射
 #[derive(Clone, Default)]
@@ -284,27 +287,22 @@ impl TraceContext {
     /// 与请求终态原子持久化；旧记录没有此字段，不伪造历史时间线
     #[must_use]
     pub fn snapshot(&self) -> Option<Value> {
-        let state = self
+        let mut state = self
             .state
             .as_ref()?
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner);
-        Some(json!({
-            "schemaVersion": 3, "requestId": state.request_id,
-            "startedAtUnixMs": state.started_at_ms, "totalEvents": state.sequence,
-            "droppedEvents": state.dropped_events, "maxEvents": MAX_EVENTS,
-            "captureMode": "sanitized", "events": state.events,
-            "wireDumpEnabled": tracing::enabled!(target: "request_dump", tracing::Level::INFO),
-            "wireFrames": state.wire_frames, "wireBytes": state.wire_bytes,
-            "maxBufferBytes": MAX_BUFFER_BYTES + MAX_EVIDENCE_BUFFER_BYTES, "headEvents": HEAD_EVENTS,
-            "maxEvidenceEvents": MAX_EVIDENCE_EVENTS,
-            "maxEvidenceDataBytes": MAX_EVIDENCE_DATA_BYTES,
-            "maxEvidenceBufferBytes": MAX_EVIDENCE_BUFFER_BYTES,
-            "droppedEvidenceEvents": state.dropped_evidence_events,
-            "stageCounts": state.stage_counts,
-            "stageCountOverflow": state.stage_count_overflow,
-            "snapshotBoundary": "execution_finalization",
-        }))
+        loop {
+            let snapshot = state.snapshot();
+            if snapshot.to_string().len() <= MAX_TRACE_SNAPSHOT_BYTES {
+                return Some(snapshot);
+            }
+            // 完整编码还包含计数与外层字段，超限时先淘汰普通事件并保留丢弃计数
+            // 元数据本身无法容纳时省略诊断，不能让辅助观测阻断请求终态写入
+            let index =
+                regular_eviction_index(&state).or_else(|| evidence_eviction_index(&state))?;
+            remove_event(&mut state, index);
+        }
     }
 
     fn push(&self, stage: &'static str, data: Value, coalesce: bool) {
@@ -439,26 +437,14 @@ impl TraceContext {
             if state.evidence_bytes > MAX_EVIDENCE_BUFFER_BYTES
                 || state.evidence_events > MAX_EVIDENCE_EVENTS
             {
-                if let Some(index) = state
-                    .events
-                    .iter()
-                    .position(|event| event.evidence && event.stage != "request.routed")
-                    .or_else(|| state.events.iter().position(|event| event.evidence))
-                {
+                if let Some(index) = evidence_eviction_index(&state) {
                     remove_event(&mut state, index);
                 }
                 continue;
             }
             // 相比流中间的常规事件，优先保留失败与重试决策
             // 关键事实使用独立预算，只有自身超限才会淘汰
-            let index = (HEAD_EVENTS..state.events.len().saturating_sub(8))
-                .find(|index| !state.events[*index].evidence)
-                .or_else(|| {
-                    (HEAD_EVENTS..state.events.len())
-                        .chain(0..HEAD_EVENTS.min(state.events.len()))
-                        .find(|index| !state.events[*index].evidence)
-                });
-            if let Some(index) = index {
+            if let Some(index) = regular_eviction_index(&state) {
                 remove_event(&mut state, index);
             } else {
                 break;
@@ -467,6 +453,45 @@ impl TraceContext {
     }
 }
 
+impl TraceState {
+    fn snapshot(&self) -> Value {
+        json!({
+            "schemaVersion": 3, "requestId": self.request_id,
+            "startedAtUnixMs": self.started_at_ms, "totalEvents": self.sequence,
+            "droppedEvents": self.dropped_events, "maxEvents": MAX_EVENTS,
+            "captureMode": "sanitized", "events": self.events,
+            "wireDumpEnabled": tracing::enabled!(target: "request_dump", tracing::Level::INFO),
+            "wireFrames": self.wire_frames, "wireBytes": self.wire_bytes,
+            "maxBufferBytes": MAX_BUFFER_BYTES + MAX_EVIDENCE_BUFFER_BYTES, "headEvents": HEAD_EVENTS,
+            "maxEvidenceEvents": MAX_EVIDENCE_EVENTS,
+            "maxEvidenceDataBytes": MAX_EVIDENCE_DATA_BYTES,
+            "maxEvidenceBufferBytes": MAX_EVIDENCE_BUFFER_BYTES,
+            "droppedEvidenceEvents": self.dropped_evidence_events,
+            "stageCounts": self.stage_counts,
+            "stageCountOverflow": self.stage_count_overflow,
+            "snapshotBoundary": "execution_finalization",
+        })
+    }
+}
+
+fn regular_eviction_index(state: &TraceState) -> Option<usize> {
+    (HEAD_EVENTS..state.events.len().saturating_sub(8))
+        .find(|index| !state.events[*index].evidence)
+        .or_else(|| {
+            (HEAD_EVENTS..state.events.len())
+                .chain(0..HEAD_EVENTS.min(state.events.len()))
+                .find(|index| !state.events[*index].evidence)
+        })
+}
+
+fn evidence_eviction_index(state: &TraceState) -> Option<usize> {
+    state
+        .events
+        .iter()
+        .position(|event| event.evidence && event.stage != "request.routed")
+        .or_else(|| state.events.iter().position(|event| event.evidence))
+}
+
 fn remove_event(state: &mut TraceState, index: usize) {
     if let Some(event) = state.events.remove(index) {
         state.bytes = state.bytes.saturating_sub(event.bytes);

backend/crates/gateway-core/tests/diagnostics/trace.rs

@@ -5,7 +5,7 @@ use std::{
     sync::{Arc, Mutex},
 };
 
-use gateway_core::diagnostics::{TraceContext, body_fingerprint};
+use gateway_core::diagnostics::{MAX_TRACE_SNAPSHOT_BYTES, TraceContext, body_fingerprint};
 use gateway_core::error::{ProviderDiagnostic, ProviderError, ProviderErrorKind, RawUpstreamError};
 use gateway_core::upstream::UpstreamSendState;
 use serde_json::json;
@@ -47,6 +47,7 @@ fn critical_evidence_survives_stream_pressure_and_preserves_stage_coverage() {
     );
     assert_eq!(snapshot["droppedEvidenceEvents"], 0);
     assert!(snapshot["droppedEvents"].as_u64().unwrap() > 0);
+    assert!(snapshot.to_string().len() <= MAX_TRACE_SNAPSHOT_BYTES);
 }
 
 #[test]
@@ -67,8 +68,42 @@ fn critical_evidence_has_an_independent_bound_and_reports_its_own_loss() {
         events.len() as u64 + snapshot["droppedEvents"].as_u64().unwrap(),
         100
     );
-    assert!(snapshot.to_string().len() < 128 * 1024);
+    assert!(snapshot.to_string().len() <= MAX_TRACE_SNAPSHOT_BYTES);
 }
+
+#[test]
+fn snapshot_budget_includes_stage_coverage_and_event_envelopes() {
+    let trace = TraceContext::new("req_snapshot_budget");
+    trace.record("request.routed", json!({"scope": "all"}));
+    for index in 0..10 {
+        trace.record_evidence(
+            "provider.request.input",
+            json!({"index": index, "hashes": "a".repeat(6000)}),
+        );
+    }
+    for index in 0..64 {
+        let stage = Box::leak(format!("stage.{index}.{}", "x".repeat(512)).into_boxed_str());
+        trace.record(stage, json!({"detail": "x".repeat(2000)}));
+    }
+    trace.record("request.finished", json!({"outcome": "succeeded"}));
+    let snapshot = trace.snapshot().unwrap();
+    let events = snapshot["events"].as_array().unwrap();
+    assert!(snapshot.to_string().len() <= MAX_TRACE_SNAPSHOT_BYTES);
+    assert_eq!(events.first().unwrap()["stage"], "request.routed");
+    assert_eq!(events.last().unwrap()["stage"], "request.finished");
+    assert_eq!(snapshot["droppedEvidenceEvents"], 0);
+    assert!(snapshot["droppedEvents"].as_u64().unwrap() > 0);
+    let retained: u64 = events
+        .iter()
+        .map(|event| event["count"].as_u64().unwrap())
+        .sum();
+    assert_eq!(
+        retained + snapshot["droppedEvents"].as_u64().unwrap(),
+        snapshot["totalEvents"]
+    );
+    assert_eq!(trace.snapshot(), Some(snapshot));
+}
+
 #[test]
 fn bounded_history_preserves_start_failure_and_final_result() {
     let trace = TraceContext::new("req_trace");

backend/crates/gateway-store/src/postgres/execution/mod.rs

@@ -8,6 +8,7 @@ use serde_json::Value;
 use sqlx::PgPool;
 use uuid::Uuid;
 
+use gateway_core::diagnostics::MAX_TRACE_SNAPSHOT_BYTES;
 use gateway_core::engine::{
     AttemptRecord as CoreAttemptRecord, ExecutionStore, IntermediateFailure,
     ModelRequestFinalization as CoreModelRequestFinalization, ModelRequestId,
@@ -326,13 +327,11 @@ impl ModelRequestFinalization {
         {
             return Err(invalid("provider observation must be a JSON object"));
         }
-        if self
-            .diagnostic_trace_json
-            .as_ref()
-            .is_some_and(|trace| !trace.is_object() || trace.to_string().len() > 64 * 1024)
-        {
+        if self.diagnostic_trace_json.as_ref().is_some_and(|trace| {
+            !trace.is_object() || trace.to_string().len() > MAX_TRACE_SNAPSHOT_BYTES
+        }) {
             return Err(invalid(
-                "diagnostic trace must be a JSON object within 64 KiB",
+                "diagnostic trace must be a JSON object within the snapshot size limit",
             ));
         }
         validate_optional_stable_reason(

backend/crates/gateway-store/tests/postgres/execution/mod.rs

@@ -8,7 +8,7 @@ use chrono::{DateTime, Duration, Utc};
 use gateway_admin::{
     model::observability as admin_observability, ports::store::ObservabilityStore as _,
 };
-use gateway_core::diagnostics::TraceContext;
+use gateway_core::diagnostics::{MAX_TRACE_SNAPSHOT_BYTES, TraceContext};
 use gateway_core::engine::{
     ExecutionOutcome, ExecutionStore, ModelRequestFailureObservation,
     ModelRequestFinalization as CoreModelRequestFinalization, ModelRequestId,
@@ -1429,54 +1429,100 @@ async fn seed_running_request(pool: &sqlx::PgPool, id: &str) -> Result<(), sqlx:
 }
 
 #[tokio::test]
-async fn diagnostic_trace_is_finalized_atomically_and_available_for_failed_request_details() {
+async fn diagnostic_trace_pressure_preserves_terminal_outcomes_and_usage() {
     let Some(database) = TestDatabase::create("diagnostic_trace").await else {
         return;
     };
-    seed_running_request(&database.pool, "req_diagnostic_failed")
-        .await
-        .unwrap();
     let store = PgExecutionStore::new(database.pool.clone());
-    let mut finalization = successful_core_finalization("req_diagnostic_failed");
-    finalization.outcome = ExecutionOutcome::Failed;
-    finalization.client_status_code = Some(502);
-    let diagnostic = gateway_core::error::ProviderDiagnostic::new(
-        "OpenAI WebSocket receive idle timeout after 300s",
-    )
-    .with_classification("receive", "receive_idle_timeout");
-    let provider_error = gateway_core::error::ProviderError::new(
-        gateway_core::error::ProviderErrorKind::Timeout,
-        UpstreamSendState::Ambiguous,
-    )
-    .with_diagnostic(diagnostic);
-    finalization.error = Some(GatewayError::from_provider(&provider_error));
-    let trace = serde_json::json!({"schemaVersion": 1, "requestId": "req_diagnostic_failed", "events": [
-        {"attemptIndex": 1, "stage": "attempt.failed", "data": {"diagnostic": {
-            "stage": "receive", "code": "receive_idle_timeout", "message": "OpenAI WebSocket receive idle timeout after 300s"
-        }}},
-    ]});
-    finalization.diagnostic_trace_json = Some(trace.to_string());
-    ExecutionStore::finalize_model_request(&store, finalization)
-        .await
-        .unwrap();
-    let persisted: (String, serde_json::Value, String) = sqlx::query_as(
-        "select outcome, diagnostic_trace_json, error_message from model_request_observations where id = 'req_diagnostic_failed'"
-    ).fetch_one(&database.pool).await.unwrap();
-    assert_eq!(
-        persisted,
-        (
-            "failed".to_owned(),
-            trace.clone(),
-            "OpenAI WebSocket receive idle timeout after 300s".to_owned()
-        )
-    );
     let repository = super::observability_repository(&database.pool);
-    let detail = repository
-        .usage_record_detail("req_diagnostic_failed")
+    for (id, outcome, expected_outcome) in [
+        (
+            "req_diagnostic_success",
+            ExecutionOutcome::Succeeded,
+            "succeeded",
+        ),
+        ("req_diagnostic_failed", ExecutionOutcome::Failed, "failed"),
+    ] {
+        seed_running_request(&database.pool, id).await.unwrap();
+        let mut finalization = successful_core_finalization(id);
+        finalization.outcome = outcome;
+        finalization.usage.input_tokens = Some(100_000);
+        finalization.usage.output_tokens = Some(2000);
+        finalization.usage.total_tokens = Some(102_000);
+        finalization.cost = CalculatedCost::from_usd_ticks(10_000_000_000)
+            .unwrap()
+            .into_estimate();
+        finalization.timings.first_token_ms = Some(120);
+        finalization.timings.latency_ms = Some(42_000);
+        let trace = TraceContext::new(id);
+        trace.record("request.routed", json!({"scope": "all"}));
+        for stage in ["provider.request.input", "provider.request.outbound"] {
+            trace.record_evidence(stage, json!({"hashes": "a".repeat(6000)}));
+        }
+        for index in 0..500 {
+            trace.record(
+                "stream.sample",
+                json!({"index": index, "detail": "x".repeat(2000)}),
+            );
+        }
+        let error_message = if outcome == ExecutionOutcome::Failed {
+            finalization.client_status_code = Some(502);
+            let message = "OpenAI WebSocket receive idle timeout after 300s";
+            let provider_error = gateway_core::error::ProviderError::new(
+                gateway_core::error::ProviderErrorKind::Timeout,
+                UpstreamSendState::Ambiguous,
+            )
+            .with_diagnostic(
+                gateway_core::error::ProviderDiagnostic::new(message)
+                    .with_classification("receive", "receive_idle_timeout"),
+            );
+            trace.record_provider_failure(&provider_error);
+            finalization.error = Some(GatewayError::from_provider(&provider_error));
+            Some(message)
+        } else {
+            None
+        };
+        trace.record("request.finished", json!({"outcome": expected_outcome}));
+        let snapshot = trace.snapshot().unwrap();
+        let encoded = snapshot.to_string();
+        assert!(encoded.len() > 64 * 1024);
+        assert!(encoded.len() <= MAX_TRACE_SNAPSHOT_BYTES);
+        assert!(snapshot["droppedEvents"].as_u64().unwrap() > 0);
+        finalization.diagnostic_trace_json = Some(encoded);
+        ExecutionStore::finalize_model_request(&store, finalization)
+            .await
+            .expect("trace pressure must not discard the terminal result");
+        let persisted: Value = sqlx::query_scalar(
+            "select jsonb_build_object(
+               'outcome', outcome, 'input', input_tokens, 'output', output_tokens,
+               'cost', cost_amount::text, 'firstToken', first_token_ms,
+               'latency', latency_ms, 'error', error_message)
+             from model_request_observations where id = $1",
+        )
+        .bind(id)
+        .fetch_one(&database.pool)
         .await
         .unwrap();
-    assert_eq!(detail.trace, Some(trace));
-    assert!(detail.related_requests.is_empty());
+        assert_eq!(
+            persisted,
+            json!({
+                "outcome": expected_outcome, "input": 100_000, "output": 2000,
+                "cost": "1.0000000000", "firstToken": 120, "latency": 42_000,
+                "error": error_message,
+            })
+        );
+        let detail = repository.usage_record_detail(id).await.unwrap();
+        assert_eq!(detail.trace, Some(snapshot));
+        assert!(detail.related_requests.is_empty());
+    }
+    assert_eq!(
+        store
+            .recover_expired_model_requests(Utc::now() + Duration::minutes(20))
+            .await
+            .unwrap()
+            .requests,
+        0,
+    );
     database.close().await;
 }
 
@@ -1784,7 +1830,7 @@ async fn zero_attempt_finalization_rejects_sent_or_missing_attempt_facts_atomica
     for invalid_trace in [
         "not JSON".to_owned(),
         "[]".to_owned(),
-        json!({"data": "x".repeat(65_536)}).to_string(),
+        json!({"data": "x".repeat(MAX_TRACE_SNAPSHOT_BYTES)}).to_string(),
     ] {
         let mut finalization = early_failure(&request);
         finalization.diagnostic_trace_json = Some(invalid_trace);

docs/api.md

@@ -2188,7 +2188,9 @@ WebSocket 握手不主动添加专项计时请求头，客户端显式提供的
 已有的 `provider.precommit.released` 数据保留释放原因、预取字节数与等待时间，不等同于下游已提交
 
 Trace v3 为路由、账号选择、失败、重试、响应边界及终态事实预留独立的 64 KiB 预算，
-普通事件另有 64 KiB 预算，总事件数最多 128。`stageCounts` 记录至多 64 类已采集阶段的次数，
+普通事件另有 64 KiB 预算，总事件数最多 128。包含元数据的完整 JSON 快照最多 128 KiB，
+生成与持久化共用同一上限；完整编码超限时先淘汰普通事件，再淘汰关键证据。
+`stageCounts` 记录至多 64 类已采集阶段的次数，
 事件淘汰不抹除阶段计数；`droppedEvidenceEvents` 与 `stageCountOverflow` 分别报告关键证据淘汰和阶段统计超限
 
 OpenAI Responses 的 `provider.request.input` 与 `provider.request.outbound` 分别记录 Provider 收到的