fix(responses): keep SSE alive before first output

Start downstream keepalives while admitted streaming requests await their first output, including middleware processing. Separate transport commitment from the replay barrier so heartbeat frames preserve safe retries and execution accounting.

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

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))

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};

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<dyn ExecutionSession>,
     connection_guard: Option<Box<dyn ConnectionGuard>>,
 ) -> 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(

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<Output = Response> + Send + 'static,
+    admitted: oneshot::Receiver<ResponseCommit>,
+) -> 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<Option<(Bytes, Self)>, 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::<Value>(&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)
+}

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;

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<Vec<&'static str>>,
     client_statuses: Mutex<Vec<u16>>,
     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<u16> {
@@ -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<Option<CoordinatedEvent>, 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]);
+    }
+}

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<S: ?Sized> {
     current: Option<CurrentAttempt>,
     /// 请求级发送状态水位；跨 attempt 单调不降，终态写回不得低于此档
     send_state_watermark: UpstreamSendState,
-    downstream_committed_at: Option<SystemTime>,
+    response_commit: super::execution::ResponseCommit,
+    // 传输心跳可先于首个业务批次提交，两者不能共用重放屏障
+    delivery_committed: bool,
     /// 已交付的实际输出禁止重放，可替换前导事件不跨越此屏障
     replay_barrier_reached: bool,
     client_status_code: Option<u16>,
@@ -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<Vec<ProviderEvent>, 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<u16>,
     ) -> 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

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()
     }

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<OnceLock<TransportState>>,
+    trace: TraceContext,
+}
+
+enum TransportState {
+    Committed { at: SystemTime, status: Option<u16> },
+    Finished,
+}
+
+impl ResponseCommit {
+    pub(crate) fn new(trace: TraceContext) -> Self {
+        Self {
+            trace,
+            ..Self::default()
+        }
+    }
+
+    /// 只记录首次实际交付；已结束且未提交的执行返回 false，不改变业务重放屏障
+    pub fn record(&self, status: Option<u16>) -> 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<SystemTime> {
+        match self.first.get() {
+            Some(TransportState::Committed { at, .. }) => Some(*at),
+            _ => None,
+        }
+    }
+
+    #[must_use]
+    pub fn status_code(&self) -> Option<u16> {
+        match self.first.get() {
+            Some(TransportState::Committed { status, .. }) => *status,
+            _ => None,
+        }
+    }
+}

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,

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()
     }

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]);
+}

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 每轮执行的总时限。

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`，

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 的