starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-07 00:05:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(openai): measure first-token latency from first response data

Browse Source 
Record first-token timing from nonempty HTTP SSE data or WebSocket messages before JSON and event classification. Preserve semantic output, reasoning and text timing boundaries.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-19 13:59:05 +08:00
co-authored by Codex
parent c4d9576639
commit 91e8105e5d
7 changed files with 91 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/providers/openai/src/provider/observation.rs
+2 -2
View File
@@ -153,8 +153,8 @@ impl OpenAiResponseObservationState {
started_at: Instant,
) -> bool {
let mut changed = false;
// 首字采用官方输出项边界，真实推理与正文仍分别计时
if signals.output_start {
// 首字采用首响应载荷边界，真实推理与正文仍分别计时
if signals.response_data {
changed |= insert_first_timing(&mut self.timings.first_token_ms, started_at);
}
if signals.reasoning_output {
backend/crates/providers/openai/src/transport/canonical.rs
+9 -2
View File
@@ -342,6 +342,14 @@ impl CodexCanonicalDecoder {
raw_sse_frame: Option<RawResponseFrame>,
output: &mut Vec<ProviderEvent>,
) -> Result<(), CodexCanonicalError> {
// 首字只要求响应载荷，不依赖 JSON、事件类型或实际输出内容
let data = event.data.trim();
self.timing_signals.response_data |=
if matches!(raw_sse_frame, Some(RawResponseFrame::WebSocket(_))) {
!event.data.is_empty()
} else {
!data.is_empty() && !data.starts_with("[DONE]")
};
if event.data.trim() == "[DONE]"
&& !matches!(raw_sse_frame, Some(RawResponseFrame::WebSocket(_)))
{
@@ -371,7 +379,7 @@ impl CodexCanonicalDecoder {
.or_else(|| value.get("type").and_then(Value::as_str));
if event_type == Some("codex.rate_limits") {
// HTTP 传输已将此控制帧投影为本地额度事实
// 此帧不能成为客户端输出，也不能启动首个输出计时
// 此帧只贡献首响应观测，不能成为客户端输出
return Ok(());
}
if event_type == Some("responsesapi.websocket_timing") && self.started && !self.completed {
@@ -438,7 +446,6 @@ impl CodexCanonicalDecoder {
fn merge_timing_signals(&mut self, signals: ResponseEventSignals) {
self.timing_signals.protocol_progress |= signals.protocol_progress;
self.timing_signals.output_start |= signals.output_start;
self.timing_signals.semantic_output |= signals.semantic_output;
self.timing_signals.reasoning_output |= signals.reasoning_output;
self.timing_signals.text_output |= signals.text_output;
backend/crates/providers/openai/src/transport/protocol/responses.rs
+4 -5
View File
@@ -222,14 +222,14 @@ pub(crate) fn is_replaceable_response_prelude(event_type: &str, value: &Value) -
}
}
/// 单个 Responses 事件对计时系统提供的稳定语义信号
/// Responses 响应载荷与实际内容的观测信号
///
/// `output_start` 只表示官方输出项开始事件；`semantic_output` 仍要求
/// `response_data` 表示收到非空响应载荷；`semantic_output` 仍要求
/// 实际内容，供重试与交付边界使用，不能与首字观测互换
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ResponseEventSignals {
pub protocol_progress: bool,
pub output_start: bool,
pub response_data: bool,
pub semantic_output: bool,
pub reasoning_output: bool,
pub text_output: bool,
@@ -313,12 +313,11 @@ impl ResponseTimingMetrics {
/// 从已解析的 Responses 事件提取计时语义
///
/// 首字采用官方 Codex 的首个 output_item.added 边界
/// 首响应载荷由解码器在 JSON 解析前观测
/// 语义输出仍要求文本、推理、工具参数、图片结果或工具执行
pub fn response_event_signals(event_type: Option<&str>, value: &Value) -> ResponseEventSignals {
let mut signals = ResponseEventSignals {
protocol_progress: !matches!(event_type, Some("response.failed" | "error")),
output_start: event_type == Some("response.output_item.added"),
..ResponseEventSignals::default()
};
match event_type {
backend/crates/providers/openai/tests/provider/contract/timing.rs
+39 -16
View File
@@ -138,7 +138,6 @@ async fn official_response_duration_is_observed_independently_of_the_request_clo
assert_eq!(timings.upstream_response_ms, Some(7_000));
assert_eq!(timings.upstream_api_overhead_ms, Some(120.25));
assert_eq!(timings.upstream_engine_ms, Some(6400.0));
assert_eq!(timings.first_token_ms, None);
assert!(timings.first_event_ms.is_some_and(|value| value >= 20_000));
}
}
@@ -195,7 +194,6 @@ async fn response_critical_path_is_observed_without_logical_turn_totals_on_both_
assert_eq!(timings.upstream_engine_ms, Some(6710.68507));
assert!((timings.upstream_api_overhead_ms.unwrap() - 81.409594).abs() < 0.000001);
assert_eq!(timings.upstream_response_ms, Some(7000));
assert_eq!(timings.first_token_ms, None);
}
}
@@ -289,12 +287,39 @@ async fn timing_frames_outside_the_active_response_do_not_enter_request_metrics(
let timings = finish_timings(&mut stream, ProviderResponseTimings::default()).await;
assert_eq!(timings.upstream_api_overhead_ms, None);
assert_eq!(timings.upstream_engine_ms, None);
assert_eq!(timings.first_token_ms, None);
server.await.unwrap();
}
#[tokio::test]
async fn content_without_output_item_start_keeps_ttft_unknown_on_every_attempt() {
async fn sse_keepalive_and_empty_data_do_not_start_ttft() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_provider_contract").await;
let (base_url, release, _, server) = paused_chunked_sse_server(
": ping\n\nevent: response.created\n\ndata: \t\n\ndata: [DONE]\n\n".to_owned(),
CAPTURE_COMPLETED_SSE.to_owned(),
)
.await;
let mut stream = provider_with_base_url(&store, base_url)
.execute(
planned_request("openai", http_generate_operation()),
timed_context(Instant::now() - Duration::from_secs(3), 1),
)
.await
.unwrap();
let first = first_upstream_timings(&mut stream).await;
assert_eq!(first.first_token_ms, None);
release.send(()).unwrap();
let final_timings = finish_timings(&mut stream, first).await;
assert!(
final_timings
.first_token_ms
.is_some_and(|value| value >= 3_000)
);
server.await.unwrap();
}
#[tokio::test]
async fn first_response_uses_request_clock_on_every_attempt() {
for attempt in [1, 2] {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_provider_contract").await;
@@ -318,7 +343,7 @@ async fn content_without_output_item_start_keeps_ttft_unknown_on_every_attempt()
.unwrap();
let first = first_upstream_timings(&mut stream).await;
assert!(first.first_event_ms.is_some_and(|value| value >= 3_000));
assert_eq!(first.first_token_ms, None);
assert!(first.first_token_ms.is_some_and(|value| value >= 3_000));
assert!(first.first_text_ms.is_some_and(|value| value >= 3_000));
release.send(()).unwrap();
let final_timings = finish_timings(&mut stream, first).await;
@@ -328,7 +353,7 @@ async fn content_without_output_item_start_keeps_ttft_unknown_on_every_attempt()
}
#[tokio::test]
async fn structural_frames_start_ttft_before_content_on_http_and_websocket() {
async fn created_response_starts_ttft_before_output_on_http_and_websocket() {
for websocket in [false, true] {
for (output, has_content) in [
(
@@ -382,13 +407,11 @@ async fn structural_frames_start_ttft_before_content_on_http_and_websocket() {
)
.await;
ws.next().await.unwrap().unwrap();
for event in [created, added] {
ws.send(Message::Text(event.to_string().into()))
.await
.unwrap();
}
ws.send(Message::Text(created.to_string().into()))
.await
.unwrap();
released.await.unwrap();
for event in [output, completed] {
for event in [added, output, completed] {
ws.send(Message::Text(event.to_string().into()))
.await
.unwrap();
@@ -397,8 +420,8 @@ async fn structural_frames_start_ttft_before_content_on_http_and_websocket() {
(base_url, release, server)
} else {
let (base_url, release, _, server) = paused_chunked_sse_server(
format!("data: {created}\n\ndata: {added}\n\n"),
format!("data: {output}\n\ndata: {completed}\n\n"),
format!("data: {created}\n\n"),
format!("data: {added}\n\ndata: {output}\n\ndata: {completed}\n\n"),
)
.await;
(base_url, release, server)
@@ -423,10 +446,10 @@ async fn structural_frames_start_ttft_before_content_on_http_and_websocket() {
return observation.timings();
}
}
panic!("missing structural first-token observation");
panic!("missing first-response observation");
})
.await
.expect("structural first token before releasing content");
.expect("first response before releasing output");
assert!(first.first_token_ms.is_some_and(|value| value >= 3_000));
assert_eq!(first.first_text_ms, None);
assert_eq!(first.first_reasoning_ms, None);
backend/crates/providers/openai/tests/transport/canonical.rs
+35 -4
View File
@@ -141,11 +141,42 @@ fn official_response_duration_does_not_cross_response_identity_or_incomplete_bou
}
#[test]
fn official_timing_frames_do_not_start_first_output_wait() {
fn first_response_observation_uses_nonempty_data_before_json_or_event_classification() {
for data in [
r#"{"type":"response.created","response":{"id":"resp_first"}}"#,
r#"{"type":"response.failed","response":{"error":{"message":"failed"}}}"#,
r#"{"type":"future.event"}"#,
"not-json",
] {
for websocket in [false, true] {
let mut decoder = CodexCanonicalDecoder::new("fallback").with_raw_sse_passthrough();
if websocket {
let _ = decoder.push_websocket(data);
} else {
let _ = decoder
.push(b": ping\r\nevent: response.created\r\ndata: \t\r\n\r\ndata: [DONE]\n\n");
assert!(!decoder.take_timing_signals().response_data);
let frame = format!("data: {data}\r\n\r\n");
let split = frame.len() / 2;
let _ = decoder.push(&frame.as_bytes()[..split]);
assert!(!decoder.take_timing_signals().response_data);
let _ = decoder.push(&frame.as_bytes()[split..]);
}
let signals = decoder.take_timing_signals();
assert!(signals.response_data, "websocket={websocket}, data={data}");
assert!(!signals.semantic_output);
assert!(!signals.reasoning_output);
assert!(!signals.text_output);
}
}
}
#[test]
fn official_timing_frames_start_first_response_without_content() {
let mut decoder = CodexCanonicalDecoder::new("gpt-5.4");
let events = decoder.push(b"data: {\"type\":\"responsesapi.websocket_timing\",\"timing_metrics\":{\"total_turn_time_s\":20}}\n\n").expect("timing frame");
assert!(!events.is_empty());
assert!(!decoder.take_timing_signals().output_start);
assert!(decoder.take_timing_signals().response_data);
assert_eq!(decoder.upstream_response_ms(), None);
}
@@ -169,7 +200,7 @@ fn decoder_should_not_forward_codex_rate_limit_metadata_fixture_as_openai_wire()
}
#[test]
fn raw_sse_passthrough_should_drop_rate_limit_control_without_timing_side_effects() {
fn raw_sse_passthrough_should_observe_rate_limit_response_without_forwarding_it() {
let frame = concat!(
"event: codex.rate_limits\n",
"data: {\"type\":\"codex.rate_limits\",\"rate_limits\":{\"primary\":{\"used_percent\":90}}}\n\n",
@@ -183,7 +214,7 @@ fn raw_sse_passthrough_should_drop_rate_limit_control_without_timing_side_effect
assert!(events.is_empty());
assert!(!signals.protocol_progress);
assert!(!signals.output_start);
assert!(signals.response_data);
assert!(!signals.semantic_output);
}
backend/crates/providers/openai/tests/transport/protocol/responses.rs
-41
View File
@@ -88,47 +88,6 @@ fn semantic_output_should_ignore_structural_empty_and_terminal_frames() {
}
}
#[test]
fn response_signals_should_count_structural_start_without_semantic_output() {
let signals = response_event_signals(
Some("response.output_item.added"),
&json!({"item": {"type": "message", "content": []}}),
);
assert!(signals.protocol_progress);
assert!(signals.output_start);
assert!(!signals.semantic_output);
}
#[test]
fn first_token_should_use_only_the_official_output_item_added_boundary() {
for event in [
"response.created",
"response.in_progress",
"keepalive",
"codex.rate_limits",
"codex.response.metadata",
"response.metadata",
"responsesapi.websocket_timing",
"response.content_part.added",
"response.reasoning_summary_part.added",
"response.output_text.delta",
"response.output_item.done",
"response.completed",
"response.failed",
"error",
] {
let signals = response_event_signals(Some(event), &json!({"type": event}));
assert!(!signals.output_start, "not an output item start: {event}");
}
let signals = response_event_signals(
Some("response.output_item.added"),
&json!({"type": "response.output_item.added", "item": {"type": "message", "content": []}}),
);
assert!(signals.output_start);
assert!(!signals.semantic_output);
}
#[test]
fn response_signals_should_count_text_inside_a_completed_output_item() {
let signals = response_event_signals(
docs/api.md
+2 -1
View File
@@ -2047,7 +2047,8 @@ OpenAI 优先采用服务端 `openai-model` / `x-openai-model` 报告（流内
`latencyMs` 从模型执行会话开始计到终结，包含账号选择、重试及流交付等待，不包含此前的入口解析、路由和准入。
`firstTokenLatencyMs` 与它使用同一计时起点，首字边界由 Provider 协议定义。
OpenAI Responses 采用官方 Codex 的首个 `response.output_item.added` 边界；
OpenAI Responses 在 HTTP SSE 收到首个非空且非 `[DONE]` 的 `data` 载荷、
或 WebSocket 收到首个非空响应消息时记录首字，不要求实际输出内容；
xAI 采用首个非空文本、推理或工具参数输出。对应事件缺失时保留未知，不用其他事件补齐首字。
请求级首字统一通过 `firstTokenLatencyMs` 返回，由前端格式化。
`latencyDetails` 的首事件、首推理和首正文时间也使用请求级起点，首推理与首正文仍要求实际内容，