diff --git a/backend/crates/gateway-api/src/openai/middleware/http.rs b/backend/crates/gateway-api/src/openai/middleware/http.rs index a9c6d96a8..d220cc92e 100644 --- a/backend/crates/gateway-api/src/openai/middleware/http.rs +++ b/backend/crates/gateway-api/src/openai/middleware/http.rs @@ -19,6 +19,9 @@ use gateway_core::{ error::GatewayError, }; +pub(crate) const SSE_KEEPALIVE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(15); +pub(crate) const SSE_KEEPALIVE_FRAME: Bytes = Bytes::from_static(b": keep-alive\n\n"); + /// 只读目录/用量与协议错误共用的单帧响应；没有 Provider attempt 或计量副作用 pub(crate) async fn buffered_response( protocol: &str, @@ -161,8 +164,8 @@ pub(crate) async fn into_http_response( let chunk = tokio::select! { biased; chunk = stream.next() => chunk?, - () = tokio::time::sleep(std::time::Duration::from_secs(15)), if keep_alive => { - Ok(Bytes::from_static(b": keep-alive\n\n")) + () = tokio::time::sleep(SSE_KEEPALIVE_INTERVAL), if keep_alive => { + Ok(SSE_KEEPALIVE_FRAME) } }; Some((chunk, stream)) diff --git a/backend/crates/gateway-api/src/openai/middleware/mod.rs b/backend/crates/gateway-api/src/openai/middleware/mod.rs index 1eaa86d37..142573842 100644 --- a/backend/crates/gateway-api/src/openai/middleware/mod.rs +++ b/backend/crates/gateway-api/src/openai/middleware/mod.rs @@ -4,6 +4,9 @@ mod http; mod request; mod session; -pub(crate) use http::{ExpectedBody, buffered_response, error_response, into_http_response}; +pub(crate) use http::{ + ExpectedBody, SSE_KEEPALIVE_FRAME, SSE_KEEPALIVE_INTERVAL, buffered_response, error_response, + into_http_response, +}; pub(crate) use request::{RequestInput, invoke_request, query_response, request_parts}; pub(crate) use session::{PendingExecution, finalize_session, pending_execution_response}; diff --git a/backend/crates/gateway-api/src/openai/responses/http.rs b/backend/crates/gateway-api/src/openai/responses/http.rs index afed0b4f6..380e55813 100644 --- a/backend/crates/gateway-api/src/openai/responses/http.rs +++ b/backend/crates/gateway-api/src/openai/responses/http.rs @@ -141,6 +141,7 @@ pub(crate) async fn execute_prepared_responses( let cancellation = prepared.cancellation(); let started_request_id = Arc::new(OnceLock::new()); let terminal_request_id = Arc::clone(&started_request_id); + let (admitted, keepalive) = tokio::sync::oneshot::channel(); let mut middleware_headers = headers.clone(); // 中间件正文已经由入口按上限解压；传输编码和旧长度不能与改写后的正文混用 middleware_headers.remove(CONTENT_ENCODING); @@ -213,37 +214,45 @@ pub(crate) async fn execute_prepared_responses( .session .trace() .capture("client.request.body", &request_body); + let _ = admitted.send(started.session.response_commit()); execution_middleware_response(started, connection_guard, terminal_validation).await }) }), ); - let middleware = if let Some(authorization) = authorization.as_ref() { - match await_with_authorization(middleware, authorization, &cancellation).await { - Ok(result) => result, - Err(error) => Err(MiddlewareError::Gateway(error)), + let response = async move { + let middleware = if let Some(authorization) = authorization.as_ref() { + match await_with_authorization(middleware, authorization, &cancellation).await { + Ok(result) => result, + Err(error) => Err(MiddlewareError::Gateway(error)), + } + } else { + middleware.await + }; + let expected = if stream { + ExpectedBody::Sse + } else { + ExpectedBody::SingleJson + }; + match middleware { + Ok(response) => { + let response = if let Some(authorization) = authorization { + authorized_middleware_response(response, authorization, cancellation) + } else { + response + }; + into_http_response( + validated_middleware_response(response, validation, stream), + expected, + ) + .await + } + Err(error) => error_response(error), } - } else { - middleware.await }; - let expected = if stream { - ExpectedBody::Sse + let response = if stream { + super::keepalive::response(response, keepalive).await } else { - ExpectedBody::SingleJson - }; - let response = match middleware { - Ok(response) => { - let response = if let Some(authorization) = authorization { - authorized_middleware_response(response, authorization, cancellation) - } else { - response - }; - into_http_response( - validated_middleware_response(response, validation, stream), - expected, - ) - .await - } - Err(error) => error_response(error), + response.await }; super::super::with_model_request_id(response, started_request_id.get().unwrap_or(&request_id)) } @@ -519,16 +528,24 @@ pub async fn stream_execution_response( session: Box, connection_guard: Option>, ) -> Response { - match streaming_execution_middleware_response( - session, - connection_guard, - ResponseValidationFacts::default(), + let (admitted, keepalive) = tokio::sync::oneshot::channel(); + let _ = admitted.send(session.response_commit()); + super::keepalive::response( + async move { + match streaming_execution_middleware_response( + session, + connection_guard, + ResponseValidationFacts::default(), + ) + .await + { + Ok(response) => into_http_response(response, ExpectedBody::Sse).await, + Err(error) => error_response(error), + } + }, + keepalive, ) .await - { - Ok(response) => into_http_response(response, ExpectedBody::Sse).await, - Err(error) => error_response(error), - } } fn validated_middleware_response( diff --git a/backend/crates/gateway-api/src/openai/responses/keepalive.rs b/backend/crates/gateway-api/src/openai/responses/keepalive.rs new file mode 100644 index 000000000..6a66d6140 --- /dev/null +++ b/backend/crates/gateway-api/src/openai/responses/keepalive.rs @@ -0,0 +1,151 @@ +//! 准入成功后覆盖 Responses 首帧等待；已发心跳的 HTTP 错误转为流内失败 + +use std::future::Future; + +use axum::{ + body::{Body, BodyDataStream, Bytes, to_bytes}, + http::{HeaderValue, StatusCode, header}, + response::Response, +}; +use futures::{StreamExt as _, future::BoxFuture, stream}; +use gateway_core::engine::execution::ResponseCommit; +use gateway_protocol::openai::sse::{ + DONE_SSE_FRAME, encode_sse_event, response_failed_sse_data_from_error_event, + response_failed_sse_data_with_id, +}; +use serde_json::Value; +use tokio::sync::oneshot; + +use crate::openai::middleware::{SSE_KEEPALIVE_FRAME, SSE_KEEPALIVE_INTERVAL}; + +pub(super) async fn response( + response: impl Future + Send + 'static, + admitted: oneshot::Receiver, +) -> Response { + let mut response = Box::pin(response); + // 鉴权、模型门禁和准入失败仍保留原 HTTP 状态，不提前建立 SSE + let commit = tokio::select! { + biased; + response = &mut response => return response, + admitted = admitted => match admitted { + Ok(commit) => commit, + Err(_) => return response.await, + }, + }; + tokio::select! { + biased; + response = &mut response => return response, + () = tokio::time::sleep(SSE_KEEPALIVE_INTERVAL) => {} + } + if !commit.record(Some(StatusCode::OK.as_u16())) { + return response.await; + } + let pending = PendingResponse::Waiting(response); + let stream = stream::once(async { Ok::<_, axum::Error>(SSE_KEEPALIVE_FRAME) }) + .chain(stream::try_unfold(pending, PendingResponse::next)); + let mut response = Response::new(Body::from_stream(stream)); + let headers = response.headers_mut(); + headers.insert( + header::CONTENT_TYPE, + HeaderValue::from_static("text/event-stream"), + ); + headers.insert( + header::CACHE_CONTROL, + HeaderValue::from_static("no-cache, no-transform"), + ); + headers.insert("x-accel-buffering", HeaderValue::from_static("no")); + response +} + +enum PendingResponse { + Waiting(BoxFuture<'static, Response>), + Streaming(BodyDataStream), +} + +impl PendingResponse { + async fn next(mut self) -> Result, axum::Error> { + loop { + match &mut self { + Self::Waiting(response) => { + // 心跳只轮询同一个 future，不取消重建首帧等待或插件调用 + let response = tokio::select! { + biased; + response = response => response, + () = tokio::time::sleep(SSE_KEEPALIVE_INTERVAL) => return Ok(Some((SSE_KEEPALIVE_FRAME, self))), + }; + let sse = response.status().is_success() + && response + .headers() + .get(header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.split(';').next()) + .is_some_and(|value| { + value.trim().eq_ignore_ascii_case("text/event-stream") + }); + let body = if sse { + response.into_body() + } else { + stream_failure(response).await + }; + // 正式响应的正文已经带有交付期心跳，后续只转交，不叠加定时器 + self = Self::Streaming(body.into_data_stream()); + } + Self::Streaming(body) => { + return body + .next() + .await + .transpose() + .map(|chunk| chunk.map(|chunk| (chunk, self))); + } + } + } + } +} + +async fn stream_failure(response: Response) -> Body { + let status = response.status(); + let retry_after = response + .headers() + .get(header::RETRY_AFTER) + .and_then(|value| value.to_str().ok()) + .map(ToOwned::to_owned); + let bytes = to_bytes(response.into_body(), usize::MAX) + .await + .unwrap_or_default(); + let json = serde_json::from_slice::(&bytes).ok(); + let raw_message = std::str::from_utf8(&bytes) + .ok() + .map(str::trim) + .filter(|text| !text.is_empty()); + // 已完成客户端错误投影的对象原样保留，包括重试建议和策略错误的扩展字段 + let mut data = json + .as_ref() + .and_then(|json| response_failed_sse_data_from_error_event(None, None, json)) + .unwrap_or_else(|| { + response_failed_sse_data_with_id( + None, + "server_error", + "upstream_error", + raw_message.unwrap_or_else(|| { + status + .canonical_reason() + .unwrap_or("Response stream failed") + }), + ) + }); + // HTTP 头已提交，按官方 SSE 错误合同保留允许下发的重试建议 + if let Some(retry_after) = retry_after + && let Some(error) = data["response"]["error"].as_object_mut() + && let Some(headers) = error + .entry("headers") + .or_insert_with(|| serde_json::json!({})) + .as_object_mut() + { + headers + .entry("retry-after") + .or_insert(Value::String(retry_after)); + } + let mut event = encode_sse_event("response.failed", &data.to_string()); + event.push_str(DONE_SSE_FRAME); + Body::from(event) +} diff --git a/backend/crates/gateway-api/src/openai/responses/mod.rs b/backend/crates/gateway-api/src/openai/responses/mod.rs index d19b1c9d1..d66867fe9 100644 --- a/backend/crates/gateway-api/src/openai/responses/mod.rs +++ b/backend/crates/gateway-api/src/openai/responses/mod.rs @@ -2,6 +2,7 @@ mod error; mod http; +mod keepalive; mod request; mod response; mod validation; diff --git a/backend/crates/gateway-api/tests/openai/responses/http.rs b/backend/crates/gateway-api/tests/openai/responses/http.rs index 84562f86f..fac5c710d 100644 --- a/backend/crates/gateway-api/tests/openai/responses/http.rs +++ b/backend/crates/gateway-api/tests/openai/responses/http.rs @@ -23,7 +23,7 @@ use futures::StreamExt; use futures::future::{BoxFuture, pending}; use gateway_core::engine::execution::{ AuthenticatedClient, ClientAuthenticationError, ExecutionService, ExecutionSession, - StartExecution, StartProviderExecution, StartedExecution, + ResponseCommit, StartExecution, StartProviderExecution, StartedExecution, }; use gateway_core::engine::middleware::{FrozenMiddlewarePlan, MiddlewareHeader}; use gateway_core::engine::{CommitRequirement, CoordinatedEvent, EngineError, ModelRequestId}; @@ -58,6 +58,7 @@ fn responses_middleware() -> RequestMiddleware { #[derive(Default)] struct Trace { + response_commit: ResponseCommit, events: Mutex>, client_statuses: Mutex>, cancelled: AtomicBool, @@ -77,7 +78,7 @@ impl Trace { self.client_statuses .lock() .expect("client status lock") - .push(status); + .push(self.response_commit.status_code().unwrap_or(status)); } fn client_statuses(&self) -> Vec { @@ -104,6 +105,7 @@ enum NextStep { Event(CoordinatedEvent), DelayedEvent(Duration, CoordinatedEvent), Error(EngineError), + DelayedError(Duration, EngineError), FinalizeCancelled, FinalizeSuccess, End, @@ -297,6 +299,10 @@ impl FakeSession { } impl ExecutionSession for FakeSession { + fn response_commit(&self) -> ResponseCommit { + self.trace.response_commit.clone() + } + fn next_event(&mut self) -> BoxFuture<'_, Result, EngineError>> { Box::pin(async move { match self.next.pop_front().unwrap_or(NextStep::End) { @@ -310,6 +316,13 @@ impl ExecutionSession for FakeSession { self.trace.push("next_event"); Ok(Some(event)) } + NextStep::DelayedError(delay, error) => { + self.trace.push("wait_error"); + tokio::time::sleep(delay).await; + self.trace.push("next_error"); + self.finalized = true; + Err(error) + } NextStep::Error(error) => { self.trace.push("next_error"); self.finalized = true; @@ -1288,27 +1301,203 @@ async fn dropping_stream_during_keepalive_should_cancel_and_finalize_execution() } #[tokio::test(start_paused = true)] -async fn streaming_keepalive_should_wait_for_first_event_before_committing_http_status() { +async fn streaming_keepalive_should_cover_four_minutes_before_first_event() { let trace = Arc::new(Trace::default()); let session = FakeSession::streaming( Arc::clone(&trace), vec![ NextStep::DelayedEvent( - Duration::from_secs(45), - delivery(started(), CommitRequirement::CommitBeforeDelivery), + Duration::from_secs(240), + delivery(completed(), CommitRequirement::CommitBeforeDelivery), + ), + NextStep::FinalizeSuccess, + ], + ); + let start = tokio::time::Instant::now(); + let response = tokio::time::timeout( + Duration::from_secs(20), + stream_execution_response(Box::new(session), None), + ) + .await + .expect("headers and heartbeat must precede the first upstream event"); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.headers()[CONTENT_TYPE], "text/event-stream"); + assert_eq!(trace.response_commit.status_code(), Some(200)); + let mut body = response.into_body().into_data_stream(); + for heartbeat in 1..16 { + let chunk = tokio::time::timeout(Duration::from_secs(20), body.next()) + .await + .expect("first-event wait must stay alive") + .expect("heartbeat") + .expect("bytes"); + assert_eq!(chunk, Bytes::from_static(b": keep-alive\n\n")); + assert_eq!(start.elapsed(), Duration::from_secs(15 * heartbeat)); + assert_eq!(trace.snapshot(), vec!["wait_event"]); + } + let terminal = body.next().await.expect("terminal").expect("bytes"); + assert!(String::from_utf8_lossy(&terminal).contains("response.completed")); + assert_eq!(start.elapsed(), Duration::from_secs(240)); + assert_eq!( + body.next().await.unwrap().unwrap(), + Bytes::from_static(b"data: [DONE]\n\n") + ); + assert!(body.next().await.is_none()); + assert_eq!( + trace.snapshot(), + vec!["wait_event", "next_event", "commit", "next_end"] + ); + assert!(!trace.is_cancelled()); +} + +#[tokio::test(start_paused = true)] +async fn streaming_error_after_initial_keepalive_should_be_a_terminal_sse_failure() { + let trace = Arc::new(Trace::default()); + let session = FakeSession::streaming( + Arc::clone(&trace), + vec![NextStep::DelayedError( + Duration::from_secs(45), + EngineError::Deadline, + )], + ) + .with_response_headers(headers(&[("x-codex-turn-state", "private-attempt-state")])); + let response = stream_execution_response(Box::new(session), None).await; + assert_eq!(response.status(), StatusCode::OK); + assert!(!response.headers().contains_key("x-codex-turn-state")); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let events = parse_sse_events(std::str::from_utf8(&bytes).unwrap()).unwrap(); + assert!(String::from_utf8_lossy(&bytes).starts_with(": keep-alive\n\n")); + let failed = events + .iter() + .find(|event| event.event.as_deref() == Some("response.failed")) + .expect("failure event"); + let payload: Value = serde_json::from_str(&failed.data).unwrap(); + assert_eq!(payload["response"]["error"]["code"], "request_timeout"); + assert!(bytes.ends_with(b"data: [DONE]\n\n")); + assert_eq!(trace.client_statuses(), vec![200]); + assert_eq!(trace.snapshot(), vec!["wait_error", "next_error"]); +} + +#[tokio::test(start_paused = true)] +async fn streaming_late_upstream_error_should_preserve_extended_error_fields() { + let mut error = json!({ + "type": "invalid_request_error", "code": "misalignment_policy_violation", + "message": "synthetic policy error", "misalignment": {"category": "synthetic"}, + "extension": {"opaque": true} + }); + let failure = ProviderError::new(ProviderErrorKind::InvalidRequest, UpstreamSendState::Sent) + .with_client_visible_upstream_response( + ClientVisibleUpstreamResponse::new( + 400, + Some(b"application/json".to_vec()), + Bytes::from(json!({"error": error}).to_string()), + ) + .with_headers(headers(&[("retry-after", "7")])), + ); + error["headers"] = json!({"retry-after": "7"}); + let session = FakeSession::streaming( + Arc::new(Trace::default()), + vec![NextStep::DelayedError( + Duration::from_secs(30), + EngineError::Provider(failure), + )], + ); + let response = stream_execution_response(Box::new(session), None).await; + assert_eq!(response.status(), StatusCode::OK); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let events = parse_sse_events(std::str::from_utf8(&bytes).unwrap()).unwrap(); + let failed = events + .iter() + .find(|event| event.event.as_deref() == Some("response.failed")) + .unwrap(); + let payload: Value = serde_json::from_str(&failed.data).unwrap(); + assert_eq!(payload["response"]["error"], error); + assert!(bytes.ends_with(b"data: [DONE]\n\n")); +} + +#[tokio::test] +async fn responses_http_connection_should_survive_a_shorter_read_idle_timeout() { + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + + let trace = Arc::new(Trace::default()); + let session = FakeSession::streaming( + Arc::clone(&trace), + vec![ + NextStep::DelayedEvent( + Duration::from_secs(35), + delivery(completed(), CommitRequirement::CommitBeforeDelivery), + ), + NextStep::FinalizeSuccess, + ], + ); + let execution = Arc::new(SessionExecution { + client: authenticated_client_for_provider("sk_correlation_test", "openai"), + session: Mutex::new(Some(Box::new(session))), + middleware: Some(Arc::new(responses_middleware()).frozen()), + }); + let app = api_router(execution).await; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let mut client = tokio::net::TcpStream::connect(address).await.unwrap(); + let body = json!({"model": "model-a", "input": "synthetic", "stream": true}).to_string(); + client.write_all(format!( + "POST /v1/responses HTTP/1.1\r\nHost: {address}\r\nAuthorization: Bearer sk_correlation_test\r\nUser-Agent: AnotherTerminal/1.0\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len() + ).as_bytes()).await.unwrap(); + let mut received = Vec::new(); + let result = async { + loop { + let mut buffer = [0; 4096]; + // 真实 TCP 每次读取都受空闲期限约束，首个业务事件晚于该期限 + let read = + tokio::time::timeout(Duration::from_secs(20), client.read(&mut buffer)).await?; + let count = read.unwrap(); + if count == 0 { + break; + } + received.extend_from_slice(&buffer[..count]); + } + Ok::<(), tokio::time::error::Elapsed>(()) + } + .await; + server.abort(); + result.expect("HTTP connection must receive keepalive before its read idle deadline"); + let wire = String::from_utf8(received).unwrap(); + assert!(wire.starts_with("HTTP/1.1 200 OK\r\n")); + assert!(wire.contains("x-accel-buffering: no\r\n")); + assert_eq!(wire.matches(": keep-alive\n\n").count(), 2); + assert!(wire.contains("response.completed")); + assert!(wire.contains("data: [DONE]\n\n")); + assert!(!trace.is_cancelled()); + assert_eq!( + trace.snapshot(), + vec!["wait_event", "next_event", "commit", "next_end"] + ); +} + +#[tokio::test(start_paused = true)] +async fn dropping_stream_before_first_event_should_cancel_the_original_wait() { + let trace = Arc::new(Trace::default()); + let session = FakeSession::streaming( + Arc::clone(&trace), + vec![ + NextStep::DelayedEvent( + Duration::from_secs(240), + delivery(completed(), CommitRequirement::CommitBeforeDelivery), ), NextStep::FinalizeCancelled, ], ); - let response = tokio::spawn(stream_execution_response(Box::new(session), None)); + let response = stream_execution_response(Box::new(session), None).await; + let mut body = response.into_body().into_data_stream(); + assert_eq!( + body.next().await.unwrap().unwrap(), + Bytes::from_static(b": keep-alive\n\n") + ); + drop(body); tokio::task::yield_now().await; - tokio::time::advance(Duration::from_secs(30)).await; - - assert!(!response.is_finished()); - assert!(trace.client_statuses().is_empty()); - let response = response.await.expect("response task"); - assert_eq!(response.status(), StatusCode::OK); - assert_eq!(trace.snapshot(), vec!["wait_event", "next_event", "commit"]); + assert!(trace.is_cancelled()); + assert_eq!(trace.detached_finalizations(), 1); + assert_eq!(trace.snapshot(), vec!["wait_event", "cancel_finalize"]); } #[tokio::test] @@ -3086,3 +3275,53 @@ async fn streaming_middleware_preserves_heartbeats_before_and_after_rewriting() assert!(!trace.is_cancelled()); } } + +#[tokio::test(start_paused = true)] +async fn responses_route_should_keep_alive_while_request_middleware_waits_for_first_event() { + let admin = crate::admin::AdminTestFixture::new().await.services; + for invalid_output in [false, true] { + let trace = Arc::new(Trace::default()); + let session = FakeSession::streaming( + Arc::clone(&trace), + vec![ + NextStep::DelayedEvent( + Duration::from_secs(150), + delivery(started(), CommitRequirement::CommitBeforeDelivery), + ), + NextStep::Event(delivery(completed(), CommitRequirement::AlreadyCommitted)), + NextStep::FinalizeSuccess, + ], + ); + let actions = if invalid_output { + vec![ResponseFrameAction::Replace(Bytes::from_static( + b"data: {invalid-json}\n\n", + ))] + } else { + Vec::new() + }; + let middleware = Arc::new(responses_middleware().with_response_actions(actions)); + let response = tokio::time::timeout( + Duration::from_secs(20), + response_with_middleware(&admin, "openai", session, true, Some(middleware)), + ) + .await + .expect("request middleware must not block transport keepalive"); + assert_eq!(response.status(), StatusCode::OK); + assert!(response.headers().contains_key("x-gateway-request-id")); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let text = std::str::from_utf8(&bytes).unwrap(); + assert!(text.starts_with(": keep-alive\n\n")); + assert!(text.ends_with("data: [DONE]\n\n")); + if invalid_output { + assert!(text.contains("response.failed")); + assert!(text.contains("middleware_failed")); + assert!(!text.contains("{invalid-json}")); + assert!(trace.is_cancelled()); + } else { + assert!(text.contains("response.completed")); + assert!(!text.contains("response.failed")); + assert!(!trace.is_cancelled()); + } + assert_eq!(trace.client_statuses(), vec![200]); + } +} diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index 5d45eef74..a9047faae 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -307,6 +307,7 @@ where request_observation, budget_prior_attempts_usd: Decimal::ZERO, budget_attempt_already_counted: false, + response_commit: super::execution::ResponseCommit::new(trace.clone()), trace, deadline, deadline_timer: deadline.wait().fuse(), @@ -345,7 +346,7 @@ where request_profiles: BTreeMap::new(), current: None, send_state_watermark: UpstreamSendState::NotSent, - downstream_committed_at: None, + delivery_committed: false, replay_barrier_reached: false, client_status_code: None, delivery_pending: false, @@ -483,7 +484,9 @@ pub struct ResponseExecutionSession { current: Option, /// 请求级发送状态水位；跨 attempt 单调不降，终态写回不得低于此档 send_state_watermark: UpstreamSendState, - downstream_committed_at: Option, + response_commit: super::execution::ResponseCommit, + // 传输心跳可先于首个业务批次提交，两者不能共用重放屏障 + delivery_committed: bool, /// 已交付的实际输出禁止重放，可替换前导事件不跨越此屏障 replay_barrier_reached: bool, client_status_code: Option, @@ -521,6 +524,11 @@ where self } + /// 传输首字节事实可在 next_event 等待期间由协议层提交 + pub fn response_commit(&self) -> super::execution::ResponseCommit { + self.response_commit.clone() + } + /// 当前请求共享的诊断上下文 pub fn trace(&self) -> TraceContext { self.trace.clone() @@ -555,7 +563,7 @@ where || self.client_transport != super::execution::ClientTransport::WebSocket || self.response_control.is_none() || events.iter().any(|event| !event.is_replayable_prelude()); - let requirement = if self.downstream_committed_at.is_some() { + let requirement = if self.delivery_committed { CommitRequirement::AlreadyCommitted } else { self.delivery_pending = true; @@ -569,7 +577,7 @@ where error, send_state, } => { - let requirement = if self.downstream_committed_at.is_some() { + let requirement = if self.delivery_committed { CommitRequirement::AlreadyCommitted } else { self.delivery_pending = true; @@ -581,7 +589,7 @@ where return CoordinatedEvent::try_batch(events, requirement).map(Some); } PullOutcome::End => { - if self.downstream_committed_at.is_some() { + if self.delivery_committed { self.finish_success().await?; } return Ok(None); @@ -597,7 +605,7 @@ where /// 会话已提交、已有待提交结果或执行失败时返回错误 pub async fn collect_uncommitted(&mut self) -> Result, EngineError> { self.resume_finalization().await; - if self.downstream_committed_at.is_some() || self.delivery_pending { + if self.delivery_committed || self.delivery_pending { return Err(EngineError::InvalidDeliveryState); } if self.is_finalized() { @@ -636,14 +644,15 @@ where &mut self, client_status_code: Option, ) -> Result<(), EngineError> { - if !self.delivery_pending || self.downstream_committed_at.is_some() || self.is_finalized() { + if !self.delivery_pending || self.delivery_committed || self.is_finalized() { return Err(EngineError::InvalidDeliveryState); } - let committed_at = SystemTime::now(); - self.trace.record( - "downstream.committed", - json!({"status": client_status_code}), - ); + self.response_commit.record(client_status_code); + let committed_at = self + .response_commit + .committed_at() + .ok_or(EngineError::InvalidDeliveryState)?; + let client_status_code = self.response_commit.status_code(); if self.request_persisted { best_effort_store_write( "mark_downstream_committed", @@ -656,7 +665,7 @@ where ) .await; } - self.downstream_committed_at = Some(committed_at); + self.delivery_committed = true; self.client_status_code = client_status_code; self.delivery_pending = false; if self.upstream_complete { @@ -669,7 +678,7 @@ where /// 原始 Provider facts 已经观察，不回滚计量，也不把丢弃误记为客户端 commit pub fn discard_pending_delivery(&mut self) -> Result<(), EngineError> { if !self.delivery_pending - || self.downstream_committed_at.is_some() + || self.delivery_committed || self.pending_terminal_failure.is_some() || self.is_finalized() { @@ -692,7 +701,10 @@ where if self.client_status_code.is_some() { return Err(EngineError::InvalidDeliveryState); } - if self.request_persisted { + let committed_status = self.response_commit.status_code(); + let client_status_code = committed_status.unwrap_or(client_status_code); + // 保活后的失败终结已写入实际 HTTP 状态，不能再触发仅补空值的 Store 更新 + if self.request_persisted && !(self.is_finalized() && committed_status.is_some()) { best_effort_store_write( "record_client_status", &self.request_id, @@ -1664,7 +1676,7 @@ where "connectionBudgetRemainingMs": self.connection_budget.remaining().map(duration_ms), "executionEffectObserved": execution_effect_observed, "delayMs": transient_retry.or(transport_recovery.map(|(_, delay)| delay)).map(duration_ms), - "downstreamCommitted": self.downstream_committed_at.is_some(), + "downstreamCommitted": self.response_commit.committed_at().is_some(), "replayBarrierReached": self.replay_barrier_reached, "sendState": format!("{attempt_send_state:?}"), })); @@ -1963,8 +1975,11 @@ where outcome: ExecutionOutcome::Succeeded, send_state: UpstreamSendState::Sent, attempt_count: self.attempts, - downstream_committed_at: self.downstream_committed_at, - client_status_code: self.client_status_code, + downstream_committed_at: self.response_commit.committed_at(), + client_status_code: self + .response_commit + .status_code() + .or(self.client_status_code), upstream_status_code, client_response_id: self.observation.client_response_id.clone(), upstream_request_id, @@ -2012,7 +2027,7 @@ where let send_state = self.raise_send_watermark(send_state); let outcome = if error.kind() == ProviderErrorKind::Cancelled { ExecutionOutcome::Cancelled - } else if self.downstream_committed_at.is_some() { + } else if self.response_commit.committed_at().is_some() { ExecutionOutcome::Incomplete } else { ExecutionOutcome::Failed @@ -2037,13 +2052,15 @@ where } async fn finish_interruption(&mut self, error: &EngineError) -> Result<(), EngineError> { + // 结算 I/O 可能继续等待，先冻结交付事实以保持终态与实际 HTTP 状态一致 + self.response_commit.finish(); let (outcome, gateway_error) = match error { EngineError::Cancelled => ( ExecutionOutcome::Cancelled, GatewayError::new(GatewayErrorKind::Cancelled, "request was cancelled"), ), EngineError::Deadline => ( - if self.downstream_committed_at.is_some() { + if self.response_commit.committed_at().is_some() { ExecutionOutcome::Incomplete } else { ExecutionOutcome::Failed @@ -2057,7 +2074,7 @@ where }; let send_state = if self.attempts == 0 { UpstreamSendState::NotSent - } else if self.downstream_committed_at.is_some() + } else if self.delivery_committed || self .current .as_ref() @@ -2128,8 +2145,11 @@ where outcome: finalization.outcome, send_state: finalization.send_state, attempt_count: self.attempts, - downstream_committed_at: self.downstream_committed_at, - client_status_code: self.client_status_code, + downstream_committed_at: self.response_commit.committed_at(), + client_status_code: self + .response_commit + .status_code() + .or(self.client_status_code), upstream_status_code: finalization.upstream_status_code.or(observed_status_code), client_response_id: self.observation.client_response_id.clone(), upstream_request_id: finalization.upstream_request_id.or(upstream_request_id), @@ -2162,6 +2182,7 @@ where } async fn persist_finalization(&mut self, finalization: ModelRequestFinalization) { + self.response_commit.finish(); self.finalized_at = Some(finalization.completed_at); let observed_provider = self .current diff --git a/backend/crates/gateway-core/src/engine/execution/contract.rs b/backend/crates/gateway-core/src/engine/execution/contract.rs index 432821eaa..138995ea1 100644 --- a/backend/crates/gateway-core/src/engine/execution/contract.rs +++ b/backend/crates/gateway-core/src/engine/execution/contract.rs @@ -297,6 +297,11 @@ pub struct StartedExecution { } pub trait ExecutionSession: Send { + /// 允许协议层在等待业务事件期间记录传输首字节，保活不提交业务批次 + fn response_commit(&self) -> super::ResponseCommit { + super::ResponseCommit::default() + } + fn trace(&self) -> crate::diagnostics::TraceContext { crate::diagnostics::TraceContext::default() } diff --git a/backend/crates/gateway-core/src/engine/execution/delivery.rs b/backend/crates/gateway-core/src/engine/execution/delivery.rs new file mode 100644 index 000000000..dfb6c7851 --- /dev/null +++ b/backend/crates/gateway-core/src/engine/execution/delivery.rs @@ -0,0 +1,65 @@ +//! 下游首字节提交事实；传输保活不代表业务事件已交付 + +use std::sync::{Arc, OnceLock}; +use std::time::SystemTime; + +use crate::diagnostics::TraceContext; + +#[derive(Clone, Default)] +pub struct ResponseCommit { + first: Arc>, + trace: TraceContext, +} + +enum TransportState { + Committed { at: SystemTime, status: Option }, + Finished, +} + +impl ResponseCommit { + pub(crate) fn new(trace: TraceContext) -> Self { + Self { + trace, + ..Self::default() + } + } + + /// 只记录首次实际交付；已结束且未提交的执行返回 false，不改变业务重放屏障 + pub fn record(&self, status: Option) -> bool { + if self + .first + .set(TransportState::Committed { + at: SystemTime::now(), + status, + }) + .is_ok() + { + self.trace.record( + "downstream.committed", + serde_json::json!({"status": status}), + ); + } + self.committed_at().is_some() + } + + pub(crate) fn finish(&self) { + // 执行已结束且未提交时保留原 HTTP 错误，不能因外层插件迟迟返回再改成 200 + let _ = self.first.set(TransportState::Finished); + } + + #[must_use] + pub fn committed_at(&self) -> Option { + match self.first.get() { + Some(TransportState::Committed { at, .. }) => Some(*at), + _ => None, + } + } + + #[must_use] + pub fn status_code(&self) -> Option { + match self.first.get() { + Some(TransportState::Committed { status, .. }) => *status, + _ => None, + } + } +} diff --git a/backend/crates/gateway-core/src/engine/execution/mod.rs b/backend/crates/gateway-core/src/engine/execution/mod.rs index 198340669..9addf09e0 100644 --- a/backend/crates/gateway-core/src/engine/execution/mod.rs +++ b/backend/crates/gateway-core/src/engine/execution/mod.rs @@ -2,6 +2,8 @@ mod admission; mod contract; +mod delivery; +pub use delivery::ResponseCommit; mod failure; pub use contract::{ AuthenticatedClient, ClientApiKeyUsageSink, ClientAuthenticationError, ClientKeyVerifier, diff --git a/backend/crates/gateway-core/src/engine/execution/session.rs b/backend/crates/gateway-core/src/engine/execution/session.rs index cd5570413..3b1732586 100644 --- a/backend/crates/gateway-core/src/engine/execution/session.rs +++ b/backend/crates/gateway-core/src/engine/execution/session.rs @@ -106,6 +106,10 @@ impl Drop for DefaultExecutionSession { } impl ExecutionSession for DefaultExecutionSession { + fn response_commit(&self) -> super::ResponseCommit { + self.core.response_commit() + } + fn trace(&self) -> crate::diagnostics::TraceContext { self.core.trace() } diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 0b8f1ef91..cfad358fd 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -5854,3 +5854,144 @@ fn zero_rotation_budget_still_allows_retrying_the_same_account() { Some(&ProviderAccountId::new("acct_first").unwrap()) ); } + +#[test] +fn transport_keepalive_preserves_pre_output_recovery_and_first_business_commit() { + let operation = generate_operation(); + let route_plan = plan(&operation); + let (coordinator, store, provider) = coordinator(vec![ + Script::Stream { + account_id: "acct_first", + items: vec![Err(transient_rejection())], + }, + Script::Stream { + account_id: "acct_first", + items: complete_stream(None), + }, + ]); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .expect("session"); + let transport = session.response_commit(); + transport.record(Some(200)); + let first_commit = transport.committed_at(); + let event = block_on(session.next_event()) + .expect("recovery") + .expect("event"); + assert_eq!( + event.commit_requirement(), + CommitRequirement::CommitBeforeDelivery + ); + block_on(session.commit_downstream(Some(200))).expect("business commit"); + while block_on(session.next_event()) + .expect("remaining stream") + .is_some() + {} + assert_eq!(transport.committed_at(), first_commit); + assert_eq!(provider.contexts.lock().unwrap().len(), 2); + let state = store.state.lock().unwrap(); + assert_eq!(state.finalizations.len(), 1); + assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Succeeded); + assert_eq!(state.finalizations[0].client_status_code, Some(200)); + assert!(state.finalizations[0].committed); +} + +#[test] +fn transport_keepalive_records_late_failure_as_incomplete_http_200() { + let operation = generate_operation(); + let route_plan = plan(&operation); + let (coordinator, store, _) = coordinator(vec![Script::Stream { + account_id: "acct_first", + items: vec![Err(ProviderError::new( + ProviderErrorKind::Timeout, + UpstreamSendState::Sent, + ) + .with_status(504))], + }]); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .expect("session"); + session.response_commit().record(Some(200)); + block_on(session.next_event()).expect_err("upstream failure"); + block_on(session.record_client_status(504)) + .expect("late logical status cannot replace HTTP 200"); + let state = store.state.lock().unwrap(); + assert_eq!(state.finalizations.len(), 1); + assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Incomplete); + assert_eq!(state.finalizations[0].client_status_code, Some(200)); + assert_eq!(state.finalizations[0].upstream_status_code, Some(504)); + assert!(state.finalizations[0].committed); + assert!(state.recorded_statuses.is_empty()); +} + +#[test] +fn transport_keepalive_cancellation_retains_commit_without_inventing_a_send() { + let operation = generate_operation(); + let route_plan = plan(&operation); + let (coordinator, store, provider) = coordinator(Vec::new()); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .expect("session"); + session.response_commit().record(Some(200)); + block_on(session.cancel_and_finalize()).expect("cancel"); + assert!(provider.contexts.lock().unwrap().is_empty()); + let state = store.state.lock().unwrap(); + assert_eq!(state.finalizations.len(), 1); + assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Cancelled); + assert_eq!(state.finalizations[0].client_status_code, Some(200)); + assert_eq!( + state.finalizations[0].send_state, + UpstreamSendState::NotSent + ); + assert!(state.finalizations[0].committed); +} + +#[test] +fn transport_keepalive_cannot_change_an_already_finalized_failure_to_http_200() { + let operation = generate_operation(); + let route_plan = plan(&operation); + let (coordinator, store, _) = coordinator(vec![Script::Stream { + account_id: "acct_first", + items: vec![Err(ProviderError::new( + ProviderErrorKind::Timeout, + UpstreamSendState::Sent, + ) + .with_status(504))], + }]); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .unwrap(); + let transport = session.response_commit(); + block_on(session.next_event()).expect_err("original failure"); + assert!(!transport.record(Some(200))); + assert!(transport.committed_at().is_none()); + block_on(session.record_client_status(504)).unwrap(); + let state = store.state.lock().unwrap(); + assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Failed); + assert!(!state.finalizations[0].committed); + assert_eq!(state.recorded_statuses, vec![504]); +} diff --git a/deploy/README.md b/deploy/README.md index 8a5c3bf3a..a9ccf12cb 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -265,9 +265,11 @@ HTTP 传输不加密，公网部署仍建议使用 HTTPS。 客户端使用部署地址下的 `/v1`，协议与部署一致，不要使用前端开发服务的 `5173/dev/v1`。 当前应用只支持单副本，不能通过复制容器扩容 -流式响应在首个上游事件提交后，每 15 秒无输出会发送一次 SSE 注释保活， +Responses 流式请求完成准入后，每 15 秒无输出会发送一次 SSE 注释保活，覆盖首个业务事件前的等待， 并设置 `X-Accel-Buffering: no` 和 `Cache-Control: no-cache, no-transform`。 -反向代理仍需允许这些响应头生效；首个事件到达前的等待也需要足够的读取超时 +反向代理仍需允许这些响应头生效；准入前的等待需要足够的读取超时。 +心跳仅覆盖客户端到 CPR 的下游链路，不会延长账号出站 HTTP/SOCKS 代理的读取期限。 +提前发送心跳后的状态码与错误行为见 [Responses 交付合同](../docs/api.md#容量拒绝与重试) 网关默认不限制模型请求的总执行时长；OpenAI 上游流默认有 300 秒空闲超时。 `api.request_timeout_seconds` 默认 `null`，只控制 HTTP 路由返回响应前的等待，不是流式正文或 WebSocket 每轮执行的总时限。 diff --git a/docs/api.md b/docs/api.md index 839032515..4ef1f1f27 100644 --- a/docs/api.md +++ b/docs/api.md @@ -452,7 +452,11 @@ xAI 的 `reasoning.effort` 接受 `none / minimal / low / medium / high / xhigh 其他 WebSocket 与 HTTP/SSE 的前导事件保留到首次有效文本、推理、工具活动或终态再交付。 前导缓冲不另设等待时长或累计体积上限，等待沿用[请求期限](#请求期限)与传输超时配置。 -WebSocket 独立 Ping 保活；HTTP 在首个业务事件前保留状态码和响应头，SSE 注释保活从提交响应后开始 +WebSocket 独立 Ping 保活；Responses HTTP 流式请求完成准入后，执行仍在等待且连续 15 秒无输出时提交 `200` +和 SSE 注释心跳，此后的等待继续每 15 秒保活，包括请求中间件等待和前导事件缓冲。 +首次心跳前仍保留最终 HTTP 状态与响应头；心跳发出后发生的失败使用 `response.failed` 和 `[DONE]`， +后续取得的上游或插件响应头无法再补发，`Retry-After` 转入流内错误的 `response.error.headers`。 +心跳不代表业务输出，不关闭已有的安全重试窗口 WebSocket 在交付有效输出前断线或超时，可按 `stream_max_retries` 做同账号、有退避的尽力重试 （默认最多 5 次），耗尽后仅在请求允许时改用 HTTP。已发送或发送结果不明不单独阻止这项恢复； @@ -1328,7 +1332,8 @@ HTTP 请求头及新建 WS 的握手提示按当时的最终出站档位构造 队列满返回 `429` / `concurrency_queue_full`，排队超时返回 `429` / `concurrency_queue_timeout`； WebSocket 使用对应错误事件。等待期间不发送上游请求，取消后释放等待位置，排队重查不重复计入 RPM。 -SSE 在取得有效执行前不发送保活帧，因此此阶段保留 HTTP 错误状态；已开始交付的失败沿用流内错误合同 +SSE 在 Key 准入完成并取得执行会话前不发送保活帧，此阶段保留 HTTP 错误状态； +取得会话后的账号等待受 SSE 保活覆盖，已开始交付的失败沿用流内错误合同 任一已结算金额达到限额后拒绝新请求，已准入请求可完成并使金额超过阈值。 HTTP 返回 `429`，`error.code` 为 `key_daily_budget_exceeded` 或 `key_weekly_budget_exceeded`， diff --git a/docs/architecture.md b/docs/architecture.md index 9e35c0e06..fd4432b7a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -382,6 +382,8 @@ Client Key 鉴权完成后，API adapter 从有界请求头识别客户端，Cor - HTTP 明确未发送的瞬时建连失败（含 TLS 握手 EOF）共享最多 4 次 opening、30 秒的请求内预算； 证书、TLS 协议及配置错误不进入同账号建连恢复，取消和执行 deadline 优先终止等待 - downstream commit 记录不可撤回的首次交付；不可重放事件交付后禁止换号、重试和 fallback + HTTP 保活可先提交响应头，Core 记录该首字节时间与实际状态，但首个业务批次仍独立提交； + SSE 注释不构成业务输出或上游已发送的证明，也不改变重放资格 原生 Responses Lite WebSocket 及时交付正常事件，仅 Provider 明确标记的可替换前导保留重试资格；Core 单独维护重放屏障， 仍检查错误分类、续接归属、发送状态、副作用及预算。中间件链不启用此前导例外 - API 在最终错误编码出口投影客户端恢复信号；该投影不修改 Provider 上游事实，也不改变 Core 的