chore: bump version to 3.3.23

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

backend/crates/gateway-api/src/openai/responses/http.rs

@@ -83,6 +83,19 @@ pub(crate) async fn responses(
         Ok(started) => started,
         Err(error) => return gateway_error_response(&error),
     };
+    tracing::info!(
+        target: "request_dump",
+        provider = "openai",
+        request_id = %started.request_id,
+        request_direction = "downstream",
+        request_transport = if streaming { "http_sse" } else { "http_json" },
+        request_method = "POST",
+        request_path = "/v1/responses",
+        request_headers = ?headers,
+        request_body = %String::from_utf8_lossy(&body),
+        contains_sensitive_data = true,
+        "request dump (unredacted)"
+    );
     let StartedExecution {
         stream, session, ..
     } = started;

backend/crates/gateway-api/src/openai/responses/websocket/mod.rs

@@ -60,12 +60,14 @@ pub(crate) async fn responses_websocket(
         &headers,
         connect_info.map(|Extension(ConnectInfo(address))| address),
     );
+    let request_headers = OpenAiRequestHeaders::from_headers(&headers);
     ResponsesWebSocketAdapter::new(service).upgrade_with_client_context(
         websocket,
         client,
         client_ip,
         user_agent,
-        OpenAiRequestHeaders::from_headers(&headers),
+        request_headers,
+        headers,
     )
 }
 
@@ -89,15 +91,29 @@ impl ResponsesWebSocketAdapter {
         client_ip: Option<IpAddr>,
         user_agent: Option<String>,
         request_headers: OpenAiRequestHeaders,
+        raw_headers: HeaderMap,
     ) -> Response {
         let connection_guard = match self.service.try_register_connection() {
             Ok(guard) => guard,
             Err(_) => return runtime_unavailable_response().into_response(),
         };
+        let connection_id = self.service.next_request_id().replacen("req_", "ws_", 1);
+        tracing::info!(
+            target: "request_dump",
+            provider = "openai",
+            websocket_connection_id = %connection_id,
+            request_direction = "downstream",
+            request_transport = "websocket_handshake",
+            request_method = "GET",
+            request_path = "/v1/responses",
+            request_headers = ?raw_headers,
+            contains_sensitive_data = true,
+            "request dump (unredacted)"
+        );
         let session = ResponsesWebSocketSession {
             service: self.service.clone(),
             client,
-            connection_id: self.service.next_request_id().replacen("req_", "ws_", 1),
+            connection_id,
             client_ip,
             user_agent,
             request_headers,
@@ -171,6 +187,18 @@ async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocke
         }
         request_count = request_count.saturating_add(1);
         let correlation_id = Arc::<str>::from(service.next_request_id());
+        tracing::info!(
+            target: "request_dump",
+            provider = "openai",
+            websocket_connection_id = connection.id(),
+            correlation_id = %correlation_id,
+            request_index = request_count,
+            request_direction = "downstream",
+            request_transport = "websocket",
+            request_payload = %payload,
+            contains_sensitive_data = true,
+            "request dump (unredacted)"
+        );
         let decoded = match decode_response_create_with_context(&payload, &request_headers) {
             Ok(decoded) => decoded.with_client_context(client_ip, user_agent.clone()),
             Err(error) => {
@@ -219,6 +247,14 @@ async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocke
                 continue;
             }
         };
+        tracing::info!(
+            target: "request_dump",
+            provider = "openai",
+            websocket_connection_id = connection.id(),
+            correlation_id = %correlation_id,
+            request_id = %started.request_id,
+            "request dump correlation"
+        );
 
         if forward_execution(&mut connection, started, &mut replay).await
             == ForwardOutcome::Disconnect

backend/crates/gateway-core/src/engine/coordinator.rs

@@ -210,6 +210,8 @@ struct CurrentAttempt {
     index: NonZeroU32,
     started_at: SystemTime,
     send_observed: bool,
+    /// Provider 已为当前 attempt 交付可供同请求恢复的不透明状态检查点。
+    session_checkpoint_observed: bool,
     response_observation: Option<ProviderResponseObservation>,
 }
 
@@ -570,6 +572,9 @@ where
                     return Err(EngineError::Deadline);
                 }
                 PollBoundary::Item(Some(Ok(mut event))) => {
+                    if let Some(state) = event.session_update() {
+                        self.observe_session_checkpoint(state);
+                    }
                     if let Some(observation) = event.take_observation() {
                         self.observe_response(observation);
                     }
@@ -931,6 +936,7 @@ where
             index: next_attempt,
             started_at: SystemTime::now(),
             send_observed: false,
+            session_checkpoint_observed: false,
             response_observation: None,
         });
         Ok(None)
@@ -972,6 +978,17 @@ where
         self.mark_send_observed().await;
     }
 
+    fn observe_session_checkpoint(&mut self, state: &ProviderSessionState) {
+        let Some(current) = self.current.as_mut() else {
+            return;
+        };
+        if state.provider() != current.metadata.provider().as_str() {
+            return;
+        }
+        current.session_checkpoint_observed = true;
+        self.operation.set_provider_session_state(state.clone());
+    }
+
     async fn mark_send_observed(&mut self) {
         let Some(current) = self.current.as_mut() else {
             return;
@@ -1071,6 +1088,11 @@ where
             .await;
         }
         let provider_proved_replay_safe = provider_proved_replay_safe(&error);
+        // Ambiguous 仍是默认 fail-closed。唯一例外是 Provider 在同一 attempt 中先
+        // 交付了不透明恢复检查点，并明确证明从该检查点重放安全；该例外只用于
+        // 固定原账号的 transport recovery，不放开普通换号或 continuation 重放。
+        let checkpointed_transport_replay =
+            current.session_checkpoint_observed && error.replay_is_safe();
         let continuation_retry = self.prepare_continuation_retry(
             &current,
             &error,
@@ -1093,14 +1115,16 @@ where
                 delay,
             }) if self.downstream_committed_at.is_none()
                 && !self.delivery_pending
-                && attempt_send_state != UpstreamSendState::Ambiguous =>
+                && (attempt_send_state != UpstreamSendState::Ambiguous
+                    || checkpointed_transport_replay) =>
             {
                 Some((AttemptTransport::Retry(retry_index), delay))
             }
             Some(crate::error::PreDeliveryRetry::SameAccountTransportFallback)
                 if self.downstream_committed_at.is_none()
                     && !self.delivery_pending
-                    && attempt_send_state != UpstreamSendState::Ambiguous =>
+                    && (attempt_send_state != UpstreamSendState::Ambiguous
+                        || checkpointed_transport_replay) =>
             {
                 Some((AttemptTransport::Fallback, Duration::ZERO))
             }

backend/crates/gateway-core/src/error.rs

@@ -623,13 +623,21 @@ impl ProviderError {
             .get_or_insert_with(|| Box::new(ProviderErrorFailureObservation::default()))
     }
 
-    /// 标记 Provider 已证明本次拒绝没有执行生成，可在下游提交前重放。
+    /// 标记 Provider 已证明本次失败可在下游提交前安全重放。
+    ///
+    /// 证明可以来自明确的“未执行”拒绝，也可以来自同 attempt 已交付、可恢复
+    /// 同一业务请求的状态检查点；后者仍须由 Core 的同账号恢复门禁单独校验。
     #[must_use]
     pub const fn with_replay_safe(mut self) -> Self {
         self.replay_safe = true;
         self
     }
 
+    /// 原地标记 Provider 已证明本次失败可从已交付的状态检查点安全恢复。
+    pub const fn set_replay_safe(&mut self) {
+        self.replay_safe = true;
+    }
+
     /// 允许 Core 仅在客户端尚未收到任何事件时执行一次受预算约束的换号恢复。
     ///
     /// 该标记不证明上游未执行，也不等价于 [`Self::with_replay_safe`]；它只表达

backend/crates/gateway-core/src/event.rs

@@ -763,7 +763,8 @@ impl ProviderEvent {
         }
     }
 
-    /// 把本事件标记为 Provider 连接内状态的提交边界。
+    /// 附加 Provider 私有状态检查点；Core 可将其用于同请求恢复，协议连接可在
+    /// terminal 事件上持有它供下一轮使用。
     pub fn attach_session_update(&mut self, state: ProviderSessionState) {
         self.session_update = Some(Box::new(state));
     }

backend/crates/gateway-core/tests/engine/coordinator.rs

@@ -270,6 +270,7 @@ enum Script {
 struct ScriptedProvider {
     scripts: Mutex<VecDeque<Script>>,
     contexts: Mutex<Vec<AttemptContext>>,
+    operations: Mutex<Vec<Operation>>,
 }
 
 impl ScriptedProvider {
@@ -277,6 +278,7 @@ impl ScriptedProvider {
         Self {
             scripts: Mutex::new(scripts.into()),
             contexts: Mutex::new(Vec::new()),
+            operations: Mutex::new(Vec::new()),
         }
     }
 }
@@ -302,6 +304,10 @@ impl Provider for ScriptedProvider {
         request: ProviderRequest,
         context: AttemptContext,
     ) -> Result<ProviderStream, ProviderError> {
+        self.operations
+            .lock()
+            .expect("operations lock")
+            .push(request.operation().clone());
         self.contexts.lock().expect("contexts lock").push(context);
         let script = self
             .scripts
@@ -2838,14 +2844,25 @@ fn ambiguous_send_state_stops_retry() {
 fn ambiguous_pre_delivery_retry_marker_does_not_rotate_account() {
     let operation = generate_operation();
     let route_plan = plan(&operation);
+    let checkpoint = ProviderSessionState::new(
+        "openai",
+        Map::from_iter([("checkpoint".to_owned(), json!("opaque-state"))]),
+    )
+    .expect("provider checkpoint");
+    let mut checkpoint_event = ProviderEvent::observation(ProviderResponseObservation::new(
+        UpstreamTransport::new("websocket").expect("transport"),
+    ));
+    checkpoint_event.attach_session_update(checkpoint);
     let (coordinator, store, provider) = coordinator(vec![
-        Script::Stream {
+        Script::ObservedStream {
             account_id: "acct_first",
-            items: vec![Err(ProviderError::new(
-                ProviderErrorKind::Transport,
-                UpstreamSendState::Ambiguous,
-            )
-            .with_pre_delivery_retry())],
+            items: vec![
+                Ok(checkpoint_event),
+                Err(
+                    ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous)
+                        .with_pre_delivery_retry(),
+                ),
+            ],
         },
         Script::Stream {
             account_id: "acct_second",
@@ -2874,6 +2891,70 @@ fn ambiguous_pre_delivery_retry_marker_does_not_rotate_account() {
     assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Failed);
 }
 
+#[test]
+fn ambiguous_transport_retry_uses_provider_session_checkpoint_on_same_account() {
+    let operation = generate_operation();
+    let route_plan = plan(&operation);
+    let checkpoint = ProviderSessionState::new(
+        "openai",
+        Map::from_iter([("checkpoint".to_owned(), json!("opaque-state"))]),
+    )
+    .expect("provider checkpoint");
+    let mut checkpoint_event = ProviderEvent::observation(ProviderResponseObservation::new(
+        UpstreamTransport::new("websocket").expect("transport"),
+    ));
+    checkpoint_event.attach_session_update(checkpoint.clone());
+    let retry_index = NonZeroU32::new(1).expect("retry index");
+    let (coordinator, store, provider) = coordinator(vec![
+        Script::ObservedStream {
+            account_id: "acct_first",
+            items: vec![
+                Ok(checkpoint_event),
+                Err(
+                    ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous)
+                        .with_replay_safe()
+                        .with_pre_delivery_transport_retry(retry_index, Duration::ZERO),
+                ),
+            ],
+        },
+        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("start execution");
+    block_on(session.collect_uncommitted()).expect("checkpointed retry succeeds");
+    block_on(session.commit_downstream(Some(200))).expect("commit winning response");
+
+    let original = ProviderAccountId::new("acct_first").expect("account id");
+    let contexts = provider.contexts.lock().expect("contexts lock");
+    assert_eq!(contexts.len(), 2);
+    assert_eq!(
+        contexts[1].transport(),
+        AttemptTransport::Retry(retry_index)
+    );
+    assert_eq!(contexts[1].required_account(), Some(&original));
+    drop(contexts);
+    let operations = provider.operations.lock().expect("operations lock");
+    assert_eq!(
+        operations[1].provider_session_state("openai"),
+        Some(&checkpoint)
+    );
+    let state = store.state.lock().expect("store lock");
+    assert_eq!(state.intermediate_failures, 1);
+    assert_eq!(state.finalizations[0].attempt_count, 2);
+    assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Succeeded);
+}
+
 #[test]
 fn pre_delivery_transport_fallback_retries_the_same_account_with_fallback_transport() {
     let operation = generate_operation();

backend/crates/gateway-host/src/config.rs

@@ -118,6 +118,9 @@ pub struct LoggingConfig {
     pub level: String,
     pub stdout: bool,
     pub file: FileLoggingConfig,
+    /// 将完整请求原文写入独立诊断日志；默认关闭，因为内容包含凭据。
+    #[serde(default)]
+    pub request_dump: bool,
 }
 
 impl LoggingConfig {

backend/crates/gateway-host/src/logging.rs

@@ -6,16 +6,19 @@ use std::io::{self, Write};
 use std::path::{Path, PathBuf};
 
 use chrono::Utc;
-use tracing_appender::non_blocking::{NonBlockingBuilder, WorkerGuard};
+use tracing_appender::non_blocking::{NonBlocking, NonBlockingBuilder, WorkerGuard};
 use tracing_subscriber::EnvFilter;
 use tracing_subscriber::{Layer as _, layer::SubscriberExt as _, util::SubscriberInitExt as _};
 
-use crate::config::LoggingConfig;
+use crate::config::{FileLoggingConfig, LoggingConfig};
 
 const LOG_FILE_PREFIX: &str = "codex-proxy-rs";
 const OAUTH_RECOVERY_LOG_FILE_PREFIX: &str = "codex-proxy-rs-oath";
 /// OAuth 恢复记录含原始 AT/RT，文件日志开启时必须落到独立文件。
 const OAUTH_RECOVERY_LOG_TARGET: &str = "oauth_recovery";
+const REQUEST_DUMP_LOG_FILE_PREFIX: &str = "codex-proxy-rs-request-dump";
+/// 请求转储含完整凭据与正文，只允许在显式开启时写入独立文件。
+const REQUEST_DUMP_LOG_TARGET: &str = "request_dump";
 
 /// non-blocking 日志 writer 的进程级守卫。
 pub struct LogGuard {
@@ -39,6 +42,7 @@ pub fn initialize_logging(config: &LoggingConfig) -> Result<LogGuard, LogError>
     let directive = env::var("RUST_LOG").unwrap_or_else(|_| config.level.clone());
     let file_filter = application_file_filter(&directive)?;
     let recovery_file_filter = oauth_recovery_file_filter();
+    let request_dump_file_filter = request_dump_file_filter();
     let stdout_filter =
         EnvFilter::try_new(stdout_filter_directive(&directive, config.file.enabled))
             .map_err(|_| LogError::InvalidFilter)?;
@@ -52,37 +56,30 @@ pub fn initialize_logging(config: &LoggingConfig) -> Result<LogGuard, LogError>
         writer
     });
     let (file_writer, recovery_file_writer) = if config.file.enabled {
-        let maximum_bytes = config
-            .file
-            .max_file_size_mb
-            .checked_mul(1024 * 1024)
-            .ok_or(LogError::SizeOverflow)?;
-        let file_writer = RotatingLogWriter::open(
-            config.file.directory.clone(),
-            LOG_FILE_PREFIX,
-            maximum_bytes,
-            config.file.retention_days,
-            config.file.max_files,
-        )?;
-        let recovery_file_writer = RotatingLogWriter::open(
-            config.file.directory.clone(),
+        let (file_writer, file_guard) =
+            rotating_file_writer(&config.file, LOG_FILE_PREFIX, "gateway-log-file")?;
+        let (recovery_file_writer, recovery_file_guard) = rotating_file_writer(
+            &config.file,
             OAUTH_RECOVERY_LOG_FILE_PREFIX,
-            maximum_bytes,
-            config.file.retention_days,
-            config.file.max_files,
+            "gateway-log-oauth-recovery-file",
         )?;
-        let (file_writer, file_guard) = NonBlockingBuilder::default()
-            .thread_name("gateway-log-file")
-            .finish(file_writer);
-        let (recovery_file_writer, recovery_file_guard) = NonBlockingBuilder::default()
-            .thread_name("gateway-log-oauth-recovery-file")
-            .finish(recovery_file_writer);
         guards.push(file_guard);
         guards.push(recovery_file_guard);
         (Some(file_writer), Some(recovery_file_writer))
     } else {
         (None, None)
     };
+    let request_dump_file_writer = if config.request_dump {
+        let (writer, guard) = rotating_file_writer(
+            &config.file,
+            REQUEST_DUMP_LOG_FILE_PREFIX,
+            "gateway-log-request-dump-file",
+        )?;
+        guards.push(guard);
+        Some(writer)
+    } else {
+        None
+    };
 
     let stdout_layer = stdout_writer.map(|writer| {
         tracing_subscriber::fmt::layer()
@@ -118,10 +115,24 @@ pub fn initialize_logging(config: &LoggingConfig) -> Result<LogGuard, LogError>
             .with_span_list(true)
             .with_filter(recovery_file_filter)
     });
+    let request_dump_file_layer = request_dump_file_writer.map(|writer| {
+        tracing_subscriber::fmt::layer()
+            .json()
+            .with_writer(writer)
+            .with_target(true)
+            .with_file(true)
+            .with_line_number(true)
+            .with_thread_ids(true)
+            .with_thread_names(true)
+            .with_current_span(true)
+            .with_span_list(true)
+            .with_filter(request_dump_file_filter)
+    });
     tracing_subscriber::registry()
         .with(stdout_layer)
         .with(file_layer)
         .with(recovery_file_layer)
+        .with(request_dump_file_layer)
         .try_init()
         .map_err(|_| LogError::AlreadyInitialized)?;
     Ok(LogGuard { _writers: guards })
@@ -131,8 +142,12 @@ fn application_file_filter(directive: &str) -> Result<EnvFilter, LogError> {
     let recovery_directive = format!("{OAUTH_RECOVERY_LOG_TARGET}=off")
         .parse()
         .expect("static OAuth recovery log directive is valid");
+    let request_dump_directive = format!("{REQUEST_DUMP_LOG_TARGET}=off")
+        .parse()
+        .expect("static request dump log directive is valid");
     EnvFilter::try_new(directive)
         .map(|filter| filter.add_directive(recovery_directive))
+        .map(|filter| filter.add_directive(request_dump_directive))
         .map_err(|_| LogError::InvalidFilter)
 }
 
@@ -142,14 +157,41 @@ fn oauth_recovery_file_filter() -> EnvFilter {
         .expect("static OAuth recovery log filter is valid")
 }
 
+fn request_dump_file_filter() -> EnvFilter {
+    format!("off,{REQUEST_DUMP_LOG_TARGET}=info")
+        .parse()
+        .expect("static request dump log filter is valid")
+}
+
 fn stdout_filter_directive(directive: &str, persistent_log_enabled: bool) -> String {
     if !persistent_log_enabled {
-        directive.to_owned()
+        format!("{directive},{REQUEST_DUMP_LOG_TARGET}=off")
     } else {
         "off,gateway_startup=info".to_owned()
     }
 }
 
+fn rotating_file_writer(
+    config: &FileLoggingConfig,
+    file_prefix: &'static str,
+    thread_name: &'static str,
+) -> Result<(NonBlocking, WorkerGuard), LogError> {
+    let maximum_bytes = config
+        .max_file_size_mb
+        .checked_mul(1024 * 1024)
+        .ok_or(LogError::SizeOverflow)?;
+    let writer = RotatingLogWriter::open(
+        config.directory.clone(),
+        file_prefix,
+        maximum_bytes,
+        config.retention_days,
+        config.max_files,
+    )?;
+    Ok(NonBlockingBuilder::default()
+        .thread_name(thread_name)
+        .finish(writer))
+}
+
 struct RotatingLogWriter {
     directory: PathBuf,
     file_prefix: &'static str,

backend/crates/gateway-host/tests/config.rs

@@ -103,6 +103,7 @@ fn valid_config() -> HostConfig {
                 max_file_size_mb: 100,
                 max_files: 30,
             },
+            request_dump: false,
         },
         system_update,
         drain_timeout_seconds: 30,

backend/crates/gateway-host/tests/logging.rs

@@ -8,13 +8,18 @@ use gateway_host::system_update::SystemUpdateConfig;
 
 const LOG_DIRECTORY_ENV: &str = "CPR_LOGGING_TEST_DIRECTORY";
 const CHILD_PROCESS_ENV: &str = "CPR_LOGGING_TEST_CHILD";
+const REQUEST_DUMP_ENABLED_ENV: &str = "CPR_LOGGING_TEST_REQUEST_DUMP_ENABLED";
 const APPLICATION_LOG_FILE_PREFIX: &str = "codex-proxy-rs.";
 const OAUTH_RECOVERY_LOG_FILE_PREFIX: &str = "codex-proxy-rs-oath.";
+const REQUEST_DUMP_LOG_FILE_PREFIX: &str = "codex-proxy-rs-request-dump.";
 const APPLICATION_LOG_TARGET: &str = "logging_test_application";
 const APPLICATION_LOG_MARKER: &str = "application-file-filter-test";
 const OAUTH_RECOVERY_LOG_TARGET: &str = "oauth_recovery";
 const OAUTH_RECOVERY_LOG_MARKER: &str = "oauth-recovery-file-filter-test";
 const OAUTH_RECOVERY_PROVIDER: &str = "openai";
+const REQUEST_DUMP_LOG_TARGET: &str = "request_dump";
+const REQUEST_DUMP_LOG_MARKER: &str = "request-dump-file-filter-test";
+const REQUEST_DUMP_SECRET: &str = "unredacted-authorization-value";
 
 #[test]
 fn logging_requires_at_least_one_sink() {
@@ -34,6 +39,7 @@ fn logging_requires_at_least_one_sink() {
                 max_file_size_mb: 100,
                 max_files: 30,
             },
+            request_dump: false,
         },
         system_update: SystemUpdateConfig::default(),
         drain_timeout_seconds: 30,
@@ -48,11 +54,12 @@ fn logging_requires_at_least_one_sink() {
 }
 
 #[test]
-fn oauth_recovery_file_logging_is_separate_and_overrides_global_log_level() {
+fn sensitive_file_logging_is_separate_and_overrides_global_log_level() {
     if env::var_os(CHILD_PROCESS_ENV).is_some() {
-        write_oauth_recovery_log(PathBuf::from(
-            env::var_os(LOG_DIRECTORY_ENV).expect("child log directory"),
-        ));
+        write_sensitive_logs(
+            PathBuf::from(env::var_os(LOG_DIRECTORY_ENV).expect("child log directory")),
+            env::var(REQUEST_DUMP_ENABLED_ENV).is_ok_and(|value| value == "true"),
+        );
         return;
     }
 
@@ -60,14 +67,15 @@ fn oauth_recovery_file_logging_is_separate_and_overrides_global_log_level() {
     let output = Command::new(env::current_exe().expect("current test executable"))
         .args([
             "--exact",
-            "logging::oauth_recovery_file_logging_is_separate_and_overrides_global_log_level",
+            "logging::sensitive_file_logging_is_separate_and_overrides_global_log_level",
             "--nocapture",
         ])
         .env(CHILD_PROCESS_ENV, "1")
         .env(LOG_DIRECTORY_ENV, directory.path())
+        .env(REQUEST_DUMP_ENABLED_ENV, "true")
         .env(
             "RUST_LOG",
-            "off,logging_test_application=info,oauth_recovery=off",
+            "off,logging_test_application=info,oauth_recovery=off,request_dump=trace",
         )
         .output()
         .expect("run logging child process");
@@ -84,6 +92,8 @@ fn oauth_recovery_file_logging_is_separate_and_overrides_global_log_level() {
     assert!(application_log.contains(APPLICATION_LOG_MARKER));
     assert!(!application_log.contains(&json_target_field(OAUTH_RECOVERY_LOG_TARGET)));
     assert!(!application_log.contains(OAUTH_RECOVERY_LOG_MARKER));
+    assert!(!application_log.contains(&json_target_field(REQUEST_DUMP_LOG_TARGET)));
+    assert!(!application_log.contains(REQUEST_DUMP_SECRET));
 
     let recovery_log = read_log_file_set(directory.path(), OAUTH_RECOVERY_LOG_FILE_PREFIX);
     assert!(recovery_log.contains(&json_target_field(OAUTH_RECOVERY_LOG_TARGET)));
@@ -91,9 +101,49 @@ fn oauth_recovery_file_logging_is_separate_and_overrides_global_log_level() {
     assert!(!recovery_log.contains(&json_target_field(APPLICATION_LOG_TARGET)));
     assert!(!recovery_log.contains(APPLICATION_LOG_MARKER));
     assert!(recovery_log.contains(&format!(r#""provider":"{OAUTH_RECOVERY_PROVIDER}""#)));
+    assert!(!recovery_log.contains(&json_target_field(REQUEST_DUMP_LOG_TARGET)));
+    assert!(!recovery_log.contains(REQUEST_DUMP_SECRET));
+
+    let request_dump_log = read_log_file_set(directory.path(), REQUEST_DUMP_LOG_FILE_PREFIX);
+    assert!(request_dump_log.contains(&json_target_field(REQUEST_DUMP_LOG_TARGET)));
+    assert!(request_dump_log.contains(REQUEST_DUMP_LOG_MARKER));
+    assert!(request_dump_log.contains(REQUEST_DUMP_SECRET));
+    assert!(!request_dump_log.contains(&json_target_field(APPLICATION_LOG_TARGET)));
+    assert!(!request_dump_log.contains(APPLICATION_LOG_MARKER));
+    assert!(!request_dump_log.contains(&json_target_field(OAUTH_RECOVERY_LOG_TARGET)));
+    assert!(!request_dump_log.contains(OAUTH_RECOVERY_LOG_MARKER));
+
+    let disabled_directory = tempfile::tempdir().expect("create disabled log directory");
+    let output = Command::new(env::current_exe().expect("current test executable"))
+        .args([
+            "--exact",
+            "logging::sensitive_file_logging_is_separate_and_overrides_global_log_level",
+            "--nocapture",
+        ])
+        .env(CHILD_PROCESS_ENV, "1")
+        .env(LOG_DIRECTORY_ENV, disabled_directory.path())
+        .env(REQUEST_DUMP_ENABLED_ENV, "false")
+        .env(
+            "RUST_LOG",
+            "off,logging_test_application=info,oauth_recovery=off,request_dump=trace",
+        )
+        .output()
+        .expect("run logging child process with request dump disabled");
+    assert!(
+        output.status.success(),
+        "logging child process failed:\nstdout:\n{}\nstderr:\n{}",
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr),
+    );
+    assert!(!log_file_set_exists(
+        disabled_directory.path(),
+        REQUEST_DUMP_LOG_FILE_PREFIX
+    ));
+    let application_log = read_log_file_set(disabled_directory.path(), APPLICATION_LOG_FILE_PREFIX);
+    assert!(!application_log.contains(REQUEST_DUMP_SECRET));
 }
 
-fn write_oauth_recovery_log(directory: PathBuf) {
+fn write_sensitive_logs(directory: PathBuf, request_dump: bool) {
     let config = HostConfig {
         listen: ListenConfig {
             host: "127.0.0.1".to_owned(),
@@ -110,6 +160,7 @@ fn write_oauth_recovery_log(directory: PathBuf) {
                 max_file_size_mb: 1,
                 max_files: 1,
             },
+            request_dump,
         },
         system_update: SystemUpdateConfig::default(),
         drain_timeout_seconds: 30,
@@ -134,6 +185,12 @@ fn write_oauth_recovery_log(directory: PathBuf) {
         marker = OAUTH_RECOVERY_LOG_MARKER,
         "OAuth recovery test record"
     );
+    tracing::info!(
+        target: REQUEST_DUMP_LOG_TARGET,
+        marker = REQUEST_DUMP_LOG_MARKER,
+        authorization = REQUEST_DUMP_SECRET,
+        "request dump test record"
+    );
     drop(bundle);
 }
 
@@ -156,6 +213,18 @@ fn read_log_file_set(directory: &Path, file_prefix: &str) -> String {
         .collect()
 }
 
+fn log_file_set_exists(directory: &Path, file_prefix: &str) -> bool {
+    fs::read_dir(directory)
+        .expect("read log directory")
+        .filter_map(Result::ok)
+        .any(|entry| {
+            entry
+                .file_name()
+                .to_str()
+                .is_some_and(|name| name.starts_with(file_prefix))
+        })
+}
+
 fn json_target_field(target: &str) -> String {
     format!(r#""target":"{target}""#)
 }

backend/crates/providers/openai/src/provider/execution.rs

@@ -186,6 +186,13 @@ pub(super) struct OpenAiSessionState {
     #[serde(default)]
     pub(super) client_turn_id: Option<String>,
     pub(super) continuation_scope: OpenAiContinuationScope,
+    /// 该状态来自同一业务请求的 metadata 检查点，只允许被紧随其后的恢复 attempt 消费。
+    #[serde(default, skip_serializing_if = "is_false")]
+    pub(super) retry_checkpoint: bool,
+}
+
+const fn is_false(value: &bool) -> bool {
+    !*value
 }
 
 #[derive(Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -232,6 +239,26 @@ pub(super) fn encode_openai_session_state(
         .map_err(|_| provider_error(ProviderErrorKind::Protocol, UpstreamSendState::Sent))
 }
 
+fn encode_openai_session_capture(
+    capture: &OpenAiSessionCapture,
+    retry_checkpoint: bool,
+) -> Result<ProviderSessionState, ProviderError> {
+    let Some(continuation_scope) = capture.continuation_scope else {
+        return Err(provider_error(
+            ProviderErrorKind::Protocol,
+            UpstreamSendState::Sent,
+        ));
+    };
+    encode_openai_session_state(OpenAiSessionState {
+        account_id: capture.account_id.clone(),
+        conversation_id: capture.conversation_id.clone(),
+        turn_state: capture.turn_state.clone(),
+        client_turn_id: capture.client_turn_id.clone(),
+        continuation_scope,
+        retry_checkpoint,
+    })
+}
+
 pub(super) fn attach_openai_session_update(
     events: &mut [ProviderEvent],
     capture: &mut Option<OpenAiSessionCapture>,
@@ -245,16 +272,7 @@ pub(super) fn attach_openai_session_update(
     let Some(capture) = capture.take() else {
         return;
     };
-    let Some(continuation_scope) = capture.continuation_scope else {
-        return;
-    };
-    let Ok(update) = encode_openai_session_state(OpenAiSessionState {
-        account_id: capture.account_id,
-        conversation_id: capture.conversation_id,
-        turn_state: capture.turn_state,
-        client_turn_id: capture.client_turn_id,
-        continuation_scope,
-    }) else {
+    let Ok(update) = encode_openai_session_capture(&capture, false) else {
         return;
     };
     events[terminal_index].attach_session_update(update);
@@ -563,6 +581,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                         WebSocketRecoveryContext {
                             policy,
                             requirement: request_transport_requirement,
+                            replay_checkpoint_available: false,
                             retry_count: websocket_retry_count,
                             max_retries: stream_max_retries,
                             request_id: context.request_id().as_str(),
@@ -594,6 +613,9 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
             Err(failure.error)?;
             return;
         }
+        let mut turn_state_checkpoint_observed = response.transport
+            == CodexBackendTransport::WebSocket
+            && response.turn_state.is_some();
         if let Some(capture) = session_capture.as_mut() {
             capture.continuation_scope = Some(if capture.response_store {
                 OpenAiContinuationScope::Persisted
@@ -707,6 +729,49 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                     continue;
                 }
                 Err(mut failure) => {
+                    let updates = take_rate_limit_updates(rate_limit_updates.as_ref()).await;
+                    let rate_limits_changed = if updates.is_empty() {
+                        false
+                    } else {
+                        passive_quota_observation.observe(&updates);
+                        let update_headers = rate_limit_update_headers(&updates);
+                        observation_state.merge_rate_limit_headers(&update_headers)
+                    };
+                    let turn_state_merge = merge_turn_state_update(
+                        turn_state_updates.as_ref(),
+                        &mut session_capture,
+                        &mut observation_state,
+                    )
+                    .await;
+                    turn_state_checkpoint_observed |= turn_state_merge.is_some();
+                    let retry_checkpoint_update = if turn_state_checkpoint_observed
+                        && failure.websocket_checkpoint_retryable
+                    {
+                        session_capture
+                            .as_ref()
+                            .and_then(|capture| encode_openai_session_capture(capture, true).ok())
+                    } else {
+                        None
+                    };
+                    let mut observation_event = if rate_limits_changed
+                        || turn_state_merge.is_some()
+                        || retry_checkpoint_update.is_some()
+                    {
+                        observation_state
+                            .observation()
+                            .map(ProviderEvent::observation)
+                    } else {
+                        None
+                    };
+                    if let (Some(event), Some(update)) =
+                        (observation_event.as_mut(), retry_checkpoint_update)
+                    {
+                        event.attach_session_update(update);
+                    }
+                    let replay_checkpoint_available = observation_event
+                        .as_ref()
+                        .and_then(ProviderEvent::session_update)
+                        .is_some();
                     if failure.websocket_transport_retryable
                         && response_transport == CodexBackendTransport::WebSocket
                         && (matches!(
@@ -719,6 +784,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                             WebSocketRecoveryContext {
                                 policy: WebSocketFailurePolicy::Budgeted,
                                 requirement: request_transport_requirement,
+                                replay_checkpoint_available,
                                 retry_count: websocket_retry_count,
                                 max_retries: stream_max_retries,
                                 request_id: context.request_id().as_str(),
@@ -731,24 +797,8 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                             },
                         );
                     }
-                    let updates = take_rate_limit_updates(rate_limit_updates.as_ref()).await;
-                    let rate_limits_changed = if updates.is_empty() {
-                        false
-                    } else {
-                        passive_quota_observation.observe(&updates);
-                        let update_headers = rate_limit_update_headers(&updates);
-                        observation_state.merge_rate_limit_headers(&update_headers)
-                    };
-                    let turn_state_changed = merge_turn_state_update(
-                        turn_state_updates.as_ref(),
-                        &mut session_capture,
-                        &mut observation_state,
-                    )
-                    .await;
-                    if (rate_limits_changed || turn_state_changed)
-                        && let Some(observation) = observation_state.observation()
-                    {
-                        yield ProviderEvent::observation(observation);
+                    if let Some(event) = observation_event {
+                        yield event;
                     }
                     if allows_account_state_mutation {
                         synchronize_passive_quota(
@@ -772,12 +822,14 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                 passive_quota_observation.observe(&updates);
                 observation_state.merge_rate_limit_headers(&rate_limit_update_headers(&updates))
             };
-            let turn_state_changed = merge_turn_state_update(
+            let turn_state_merge = merge_turn_state_update(
                 turn_state_updates.as_ref(),
                 &mut session_capture,
                 &mut observation_state,
             )
             .await;
+            turn_state_checkpoint_observed |= turn_state_merge.is_some();
+            let turn_state_changed = turn_state_merge.unwrap_or(false);
             let first_event_changed =
                 observation_state.observe_stream_chunk(&chunk, output_started_at);
             let chunk_len = chunk.len();
@@ -959,7 +1011,8 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
             &mut session_capture,
             &mut observation_state,
         )
-        .await;
+        .await
+        .unwrap_or(false);
         attach_openai_session_update(&mut events, &mut session_capture);
         let completed = events
             .iter()
@@ -1030,15 +1083,11 @@ async fn merge_turn_state_update(
     updates: Option<&CodexTurnStateUpdate>,
     session_capture: &mut Option<OpenAiSessionCapture>,
     observation_state: &mut OpenAiResponseObservationState,
-) -> bool {
-    let Some(updates) = updates else {
-        return false;
-    };
-    let Some(turn_state) = updates.lock().await.take() else {
-        return false;
-    };
+) -> Option<bool> {
+    let updates = updates?;
+    let turn_state = updates.lock().await.take()?;
     if let Some(capture) = session_capture.as_mut() {
         capture.turn_state = Some(turn_state.clone());
     }
-    observation_state.merge_client_header("x-codex-turn-state", &turn_state)
+    Some(observation_state.merge_client_header("x-codex-turn-state", &turn_state))
 }

backend/crates/providers/openai/src/provider/failure.rs

@@ -39,6 +39,8 @@ pub fn openai_failure_affects_account_score(error: &ProviderError) -> bool {
 pub(super) struct MappedProviderFailure {
     pub(super) error: ProviderError,
     pub(super) websocket_transport_retryable: bool,
+    /// 兼容少数链路在 metadata 后、终态前关闭；仅在捕获 turn-state 时允许安全恢复。
+    pub(super) websocket_checkpoint_retryable: bool,
     pub(super) account_failure: Option<CodexAccountFailure>,
     /// 原始上游错误描述，仅在凭据错误状态下持久化。
     pub(super) error_message: Option<String>,
@@ -54,6 +56,7 @@ impl MappedProviderFailure {
         Self {
             error,
             websocket_transport_retryable: false,
+            websocket_checkpoint_retryable: false,
             account_failure: None,
             error_message: None,
             cyber_policy_failure: false,
@@ -451,6 +454,8 @@ pub(super) enum WebSocketFailurePolicy {
 pub(super) struct WebSocketRecoveryContext<'a> {
     pub(super) policy: WebSocketFailurePolicy,
     pub(super) requirement: TransportRequirement,
+    /// 本 attempt 已从 metadata 捕获可恢复同一请求的 turn-state 检查点。
+    pub(super) replay_checkpoint_available: bool,
     pub(super) retry_count: u32,
     pub(super) max_retries: u32,
     pub(super) request_id: &'a str,
@@ -505,10 +510,11 @@ pub(super) fn apply_websocket_recovery_policy(
     context: WebSocketRecoveryContext<'_>,
 ) {
     let send_state = failure.error.send_state();
-    if matches!(
+    let post_send = matches!(
         send_state,
         UpstreamSendState::Sent | UpstreamSendState::Ambiguous
-    ) {
+    );
+    if post_send && !context.replay_checkpoint_available {
         // payload 已经发送或发送结果不确定时，代理不能用同一业务输入自动重放。
         // 当前请求保持终态失败；恢复状态只裁决客户端主动发起的下一次请求。
         let session_recovery_transition = match context.session_recovery_probe.take() {
@@ -555,6 +561,11 @@ pub(super) fn apply_websocket_recovery_policy(
     // 传输恢复必须保持原账号可调度；最终 HTTP attempt 若仍失败，再按真实 HTTP
     // 结果更新账号健康度，避免中间 WS 错误把同账号钉选提前冷却掉。
     failure.account_failure = None;
+    if post_send {
+        // Core 对 Ambiguous 默认仍然 fail-closed；只有同一 attempt 的 session
+        // checkpoint 与这个 Provider proof 同时存在时，才允许同账号传输恢复。
+        failure.error.set_replay_safe();
+    }
     let fallback_now = context.policy == WebSocketFailurePolicy::ImmediateFallback
         || context.retry_count >= context.max_retries;
     if !fallback_now {
@@ -583,6 +594,7 @@ pub(super) fn apply_websocket_recovery_policy(
             websocket_max_retries = context.max_retries,
             websocket_retry_delay_ms = u64::try_from(delay.as_millis()).unwrap_or(u64::MAX),
             transport_requirement = context.requirement.as_str(),
+            replay_checkpoint_available = context.replay_checkpoint_available,
             "OpenAI upstream WebSocket failed; retrying the same account"
         );
         return;
@@ -621,6 +633,7 @@ pub(super) fn apply_websocket_recovery_policy(
             WebSocketFailurePolicy::Budgeted => "retry_budget_exhausted",
             WebSocketFailurePolicy::ImmediateFallback => "upgrade_required",
         },
+        replay_checkpoint_available = context.replay_checkpoint_available,
         session_affinity_present = context.session_affinity_key.is_some(),
         session_affinity_key_hash = context.session_affinity_key_hash.unwrap_or(""),
         session_transport_action =
@@ -673,14 +686,31 @@ pub(super) fn continuation_replay_required_error(reason: &'static str) -> Provid
 pub(super) fn map_stream_error(error: CodexClientError) -> MappedProviderFailure {
     let allows_pre_delivery_retry = stream_transport_allows_pre_delivery_retry(&error);
     let websocket_failure = error.transport() == Some(CodexBackendTransport::WebSocket);
+    let websocket_checkpoint_retryable = websocket_metadata_close_can_resume(&error);
     let mut failure = map_client_error(error, UpstreamSendState::Sent, false);
     failure.websocket_transport_retryable = allows_pre_delivery_retry && websocket_failure;
+    failure.websocket_checkpoint_retryable =
+        allows_pre_delivery_retry && websocket_checkpoint_retryable;
     if allows_pre_delivery_retry && !websocket_failure {
         failure.error = failure.error.with_pre_delivery_retry();
     }
     failure
 }
 
+fn websocket_metadata_close_can_resume(error: &CodexClientError) -> bool {
+    let CodexClientError::WebSocket(error) = error else {
+        return false;
+    };
+    let Some(close) = error.close_before_terminal() else {
+        return false;
+    };
+    close.code() == Some(1000)
+        && matches!(
+            close.last_event_type(),
+            Some("codex.response.metadata" | "response.metadata")
+        )
+}
+
 pub(super) fn stream_transport_allows_pre_delivery_retry(error: &CodexClientError) -> bool {
     match error {
         CodexClientError::Http(_)
@@ -1068,6 +1098,7 @@ pub(super) fn map_upstream_failure(
     MappedProviderFailure {
         error,
         websocket_transport_retryable: false,
+        websocket_checkpoint_retryable: false,
         account_failure: account_failure(
             category,
             failure.retry_after_seconds,

backend/crates/providers/openai/src/provider/mod.rs

@@ -321,12 +321,14 @@ impl Provider for CodexProvider {
             identity.prepare_local_conversation(&mut upstream_request);
         }
         if let Some(previous_session) = previous_session.as_ref() {
-            upstream_request.turn_state = if same_client_turn(
-                previous_session.client_turn_id.as_deref(),
-                upstream_request.client_turn_id.as_deref(),
-            ) {
+            upstream_request.turn_state = if previous_session.retry_checkpoint
+                || same_client_turn(
+                    previous_session.client_turn_id.as_deref(),
+                    upstream_request.client_turn_id.as_deref(),
+                ) {
                 // 客户端可能携带 metadata-close 失败后才拿到的新状态；同一
-                // turn 内显式回传值比最后一次成功响应保存的状态更新。
+                // turn 内显式回传值比最后一次成功响应保存的状态更新。同请求
+                // 内部恢复由一次性 retry checkpoint 明确证明，不依赖客户端 turn ID。
                 upstream_request
                     .turn_state
                     .take()

backend/crates/providers/openai/src/transport/client_sse.rs

@@ -37,10 +37,11 @@ use crate::transport::{
     response_meta,
     websocket::{
         CodexWebSocketConnection, CodexWebSocketExchangeError, CodexWebSocketPool,
-        CodexWebSocketPoolKey, CodexWebSocketStreamingExchange, DEFAULT_INITIAL_EVENT_TIMEOUT,
-        WEBSOCKET_FAST_PATH_BUDGET, WebSocketFastPath, WebSocketOriginBreaker,
-        WebSocketPoolDecision, execute_prepared_response_create_request_stream,
-        post_send_ambiguous, prepare_response_create_request_with_pool, websocket_audit_dir,
+        CodexWebSocketPoolKey, CodexWebSocketRequest, CodexWebSocketStreamingExchange,
+        DEFAULT_INITIAL_EVENT_TIMEOUT, WEBSOCKET_FAST_PATH_BUDGET, WebSocketFastPath,
+        WebSocketOriginBreaker, WebSocketPoolDecision,
+        execute_prepared_response_create_request_stream, post_send_ambiguous,
+        prepare_response_create_request_with_pool, websocket_audit_dir,
         write_websocket_audit_artifact_from_env,
     },
 };
@@ -104,11 +105,26 @@ impl CodexBackendClient {
         upstream_body.insert("stream".to_owned(), serde_json::Value::Bool(true));
         let body =
             serde_json::to_vec(&upstream_body).map_err(CodexClientError::RequestBodyEncode)?;
+        let endpoint = endpoint_url(&self.base_url, CODEX_RESPONSES_PATH);
+        tracing::info!(
+            target: "request_dump",
+            provider = "openai",
+            request_id = %context.request_id,
+            account_id = context.account_id.unwrap_or_default(),
+            request_direction = "upstream",
+            request_transport = "http_sse",
+            request_method = "POST",
+            request_url = %endpoint,
+            request_headers = ?headers,
+            request_body = %String::from_utf8_lossy(&body),
+            contains_sensitive_data = true,
+            "request dump (unredacted)"
+        );
         let body = zstd::stream::encode_all(std::io::Cursor::new(body), 3)
             .map_err(CodexClientError::RequestCompression)?;
         let response = self
             .client
-            .post(endpoint_url(&self.base_url, CODEX_RESPONSES_PATH))
+            .post(endpoint)
             .headers(headers)
             .header(CONTENT_ENCODING, HeaderValue::from_static("zstd"))
             .body(body)
@@ -425,6 +441,7 @@ impl CodexBackendClient {
                     prepared,
                 } = *route;
                 let delivery_wait_started_at = Instant::now();
+                log_unredacted_websocket_request(&websocket_request, context);
                 let mut exchange = match execute_prepared_response_create_request_stream(
                     &websocket_request,
                     prepared,
@@ -495,6 +512,7 @@ impl CodexBackendClient {
                                         route: PreparedResponseRoute::WebSocket(route),
                                         ..
                                     }) => {
+                                        log_unredacted_websocket_request(&route.request, context);
                                         match execute_prepared_response_create_request_stream(
                                             &route.request,
                                             route.prepared,
@@ -891,6 +909,26 @@ impl CodexBackendClient {
     }
 }
 
+fn log_unredacted_websocket_request(
+    request: &CodexWebSocketRequest,
+    context: CodexRequestContext<'_>,
+) {
+    tracing::info!(
+        target: "request_dump",
+        provider = "openai",
+        request_id = %context.request_id,
+        account_id = context.account_id.unwrap_or_default(),
+        request_direction = "upstream",
+        request_transport = "websocket",
+        request_method = "GET + response.create",
+        request_url = request.connection().endpoint(),
+        request_headers = ?request.connection().headers(),
+        request_payload = request.payload_text(),
+        contains_sensitive_data = true,
+        "request dump (unredacted)"
+    );
+}
+
 /// 首个可投递帧前的交付边界结果。
 enum DeliveryBoundary {
     /// 已越过边界，可开始向下游投递。

backend/crates/providers/openai/tests/provider/contract.rs

@@ -665,6 +665,7 @@ async fn capture_turn_state_request(
     previous_turn_id: Option<&str>,
     current_turn_id: Option<&str>,
     client_turn_state: Option<&str>,
+    retry_checkpoint: bool,
 ) -> wiremock::Request {
     let account_id = "acct_session_affinity";
     let store = Arc::new(MemoryAccountStore::default());
@@ -689,6 +690,9 @@ async fn capture_turn_state_request(
     if let Some(previous_turn_id) = previous_turn_id {
         session_state.insert("client_turn_id".to_owned(), json!(previous_turn_id));
     }
+    if retry_checkpoint {
+        session_state.insert("retry_checkpoint".to_owned(), json!(true));
+    }
     let mut protocol_context = Map::from_iter([("use_websocket".to_owned(), json!(false))]);
     if let Some(current_turn_id) = current_turn_id {
         protocol_context.insert("turn_id".to_owned(), json!(current_turn_id));
@@ -2521,7 +2525,7 @@ async fn websocket_turn_state_metadata_is_exposed_through_response_observation()
 }
 
 #[tokio::test]
-async fn websocket_turn_state_metadata_survives_close_1000_without_changing_failure_semantics() {
+async fn websocket_turn_state_metadata_close_requests_checkpointed_retry() {
     const ACCOUNT_ID: &str = "acct_websocket_metadata_close";
     const TURN_STATE: &str = "turn-state-before-close";
 
@@ -2579,6 +2583,7 @@ async fn websocket_turn_state_metadata_survives_close_1000_without_changing_fail
         .await
         .expect("prepare WebSocket provider stream");
     let mut observed_turn_state = false;
+    let mut replay_checkpoint = None;
     let error = loop {
         match stream.next().await {
             Some(Ok(event)) => {
@@ -2588,6 +2593,9 @@ async fn websocket_turn_state_metadata_survives_close_1000_without_changing_fail
                             && header.value().as_ref() == TURN_STATE.as_bytes()
                     });
                 }
+                if let Some(update) = event.session_update() {
+                    replay_checkpoint = Some(update.clone());
+                }
             }
             Some(Err(error)) => break error,
             None => panic!("metadata close must surface a provider error"),
@@ -2596,8 +2604,22 @@ async fn websocket_turn_state_metadata_survives_close_1000_without_changing_fail
     server.await.expect("WebSocket server");
 
     assert!(observed_turn_state);
+    let replay_checkpoint = replay_checkpoint.expect("metadata close checkpoint");
+    assert_eq!(
+        replay_checkpoint.payload().get("turn_state"),
+        Some(&json!(TURN_STATE))
+    );
+    assert_eq!(
+        replay_checkpoint.payload().get("retry_checkpoint"),
+        Some(&json!(true))
+    );
     assert_eq!(error.send_state(), UpstreamSendState::Ambiguous);
-    assert_eq!(error.pre_delivery_retry(), None);
+    assert!(error.replay_is_safe());
+    assert!(matches!(
+        error.pre_delivery_retry(),
+        Some(PreDeliveryRetry::SameAccountTransportRetry { retry_index, .. })
+            if retry_index.get() == 1
+    ));
     assert_eq!(
         error.upstream_code().map(|code| code.as_str()),
         Some("websocket_close_1000")
@@ -3036,6 +3058,7 @@ async fn matching_turn_id_should_restore_previous_turn_state() {
         Some("turn-same"),
         Some("turn-same"),
         None,
+        false,
     )
     .await;
 
@@ -3045,6 +3068,17 @@ async fn matching_turn_id_should_restore_previous_turn_state() {
     );
 }
 
+#[tokio::test]
+async fn retry_checkpoint_should_restore_turn_state_without_client_turn_id() {
+    let request =
+        capture_turn_state_request("req_restore_retry_checkpoint", None, None, None, true).await;
+
+    assert_eq!(
+        captured_header_values(&request, "x-codex-turn-state"),
+        vec![b"previous-turn-state".to_vec()]
+    );
+}
+
 #[tokio::test]
 async fn matching_turn_id_should_prefer_an_explicit_client_echo_over_saved_provider_state() {
     let request = capture_turn_state_request(
@@ -3052,6 +3086,7 @@ async fn matching_turn_id_should_prefer_an_explicit_client_echo_over_saved_provi
         Some("turn-same"),
         Some("turn-same"),
         Some("client-turn-state"),
+        false,
     )
     .await;
 
@@ -3083,6 +3118,7 @@ async fn new_or_unidentified_turn_should_not_restore_previous_turn_state() {
             previous_turn_id,
             current_turn_id,
             client_turn_state,
+            false,
         )
         .await;
         assert!(captured_header_values(&request, "x-codex-turn-state").is_empty());

deploy/config.example.yaml

@@ -20,6 +20,8 @@ host:
   logging:
     level: 'info'
     stdout: true
+    # 将完整请求头和正文写入独立文件；包含凭据，仅排障时开启。
+    request_dump: false
     file:
       enabled: true
       # 相对路径以 deploy/config.yaml 所在目录为基准。

release/version.yaml

@@ -1 +1 @@
-version: "3.3.22"
+version: "3.3.23"