chore(release): v3.22.1

Co-authored-by: Codex <noreply@openai.com>
fix(usage): parse response critical path timings
2026-11-09 16:06:27 +00:00 · 2026-10-09 14:10:51 +08:00 · 2026-10-09 13:51:40 +08:00 · 2026-10-09 13:51:14 +08:00
24 changed files with 461 additions and 333 deletions
@@ -1,4 +1,4 @@
-//! 服务模式完成全部组装后交给 Host 监听，并在请求和 Worker 排空后关闭插件
+//! 服务模式完成全部组装后交给 Host 监听，并在停止监听与 Worker 收尾后关闭插件

 use gateway_host::ConfigError;

@@ -500,7 +500,7 @@ pub struct ProviderResponseTimings {
    pub upstream_response_ms: Option<u64>,
    /// 上游排除引擎与客户端工具时间后的 API 耗时
    pub upstream_api_overhead_ms: Option<f64>,
-    /// 上游引擎服务总耗时；以下专项指标保留毫秒小数及各自内部计量层级
+    /// 上游响应级引擎耗时；以下专项指标保留毫秒小数及各自内部计量层级
    pub upstream_engine_ms: Option<f64>,
    pub upstream_engine_iapi_ttft_ms: Option<f64>,
    pub upstream_engine_service_ttft_ms: Option<f64>,
@@ -126,7 +126,7 @@ pub trait ConnectionGuard: Send + 'static {}
 pub trait ConnectionLifecycle: Send + Sync {
    /// 原子地检查 drain 状态并注册一个活跃连接
    ///
-    /// 当本方法成功时，drain 必须等待返回的 guard 被释放；
+    /// 成功注册不延长关闭流程，活跃连接可随进程退出被中断；
    /// 当 drain 已经线性化生效时，本方法必须返回 [`ConnectionDraining`]
    fn try_register(&self) -> Result<Box<dyn ConnectionGuard>, ConnectionDraining>;

@@ -81,8 +81,6 @@ pub struct HostConfig {
    pub logging: LoggingConfig,
    #[serde(default)]
    pub system_update: SystemUpdateConfig,
-    #[serde(default = "default_drain_timeout_seconds")]
-    pub drain_timeout_seconds: u64,
    #[serde(default = "default_worker_shutdown_timeout_seconds")]
    pub worker_shutdown_timeout_seconds: u64,
 }
@@ -110,9 +108,6 @@ impl HostConfig {
        if self.listen.port == 0 {
            return Err(ConfigError::InvalidField("host.listen.port"));
        }
-        if self.drain_timeout_seconds == 0 {
-            return Err(ConfigError::InvalidField("host.drain_timeout_seconds"));
-        }
        if self.worker_shutdown_timeout_seconds == 0 {
            return Err(ConfigError::InvalidField(
                "host.worker_shutdown_timeout_seconds",
@@ -136,11 +131,6 @@ impl HostConfig {
        &self.runtime_data_dir
    }

-    #[must_use]
-    pub const fn drain_timeout(&self) -> Duration {
-        Duration::from_secs(self.drain_timeout_seconds)
-    }
-
    #[must_use]
    pub const fn worker_shutdown_timeout(&self) -> Duration {
        Duration::from_secs(self.worker_shutdown_timeout_seconds)
@@ -264,10 +254,6 @@ fn resolve_relative_path(base: &Path, path: &mut PathBuf) {
    }
 }

-const fn default_drain_timeout_seconds() -> u64 {
-    30
-}
-
 const fn default_worker_shutdown_timeout_seconds() -> u64 {
    30
 }
@@ -164,7 +164,7 @@ impl HostBundle {
        Ok(())
    }

-    /// 进程唯一阻塞点；返回前完成 HTTP drain 与 worker join
+    /// 进程唯一阻塞点；停止 HTTP 监听后完成 worker join
    pub async fn serve(self, router: Router) -> Result<(), HostError> {
        let result = serve_router(
            router,
@@ -172,7 +172,6 @@ impl HostBundle {
            self.config.listen.port,
            self.cancellation.clone(),
            Arc::clone(&self.connections),
-            self.config.drain_timeout(),
        )
        .await;
        self.cancellation.cancel();
@@ -181,7 +180,7 @@ impl HostBundle {
            .await;
        match &result {
            Ok(()) => {
-                tracing::info!(target: "gateway_shutdown", pid = std::process::id(), "HTTP 服务与后台任务已停止")
+                tracing::info!(target: "gateway_shutdown", pid = std::process::id(), "HTTP 监听已关闭，后台任务已结束")
            }
            Err(error) => {
                tracing::error!(target: "gateway_shutdown", pid = std::process::id(), %error, "HTTP 服务异常停止，后台任务已清理")
@@ -1,13 +1,12 @@
-//! HTTP 监听、OS signal、原子连接注册与优雅 drain
+//! HTTP 监听、关闭信号与停止时的原子连接注册

+use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::{Arc, OnceLock};
 use std::time::Duration;

 use axum::Router;
 use gateway_core::lifecycle::CancellationToken;
 use gateway_core::lifecycle::{ConnectionDraining, ConnectionGuard, ConnectionLifecycle};
-use tokio::sync::Notify;

 const DRAINING_BIT: usize = 1usize << (usize::BITS - 1);
 const ACTIVE_MASK: usize = !DRAINING_BIT;
@@ -24,7 +23,6 @@ pub struct ConnectionTracker {

 struct ConnectionState {
    value: AtomicUsize,
-    idle: Notify,
 }

 impl ConnectionTracker {
@@ -33,36 +31,14 @@ impl ConnectionTracker {
        Self {
            state: Arc::new(ConnectionState {
                value: AtomicUsize::new(0),
-                idle: Notify::new(),
            }),
            cancellation,
        }
    }

    pub(crate) fn begin_draining(&self) {
-        let previous = self.state.value.fetch_or(DRAINING_BIT, Ordering::AcqRel);
+        self.state.value.fetch_or(DRAINING_BIT, Ordering::AcqRel);
        self.cancellation.cancel();
-        if previous & ACTIVE_MASK == 0 {
-            self.state.idle.notify_waiters();
-        }
-    }
-
-    pub async fn wait_until_idle(&self, timeout: Duration) {
-        let wait = async {
-            loop {
-                let notified = self.state.idle.notified();
-                tokio::pin!(notified);
-                // notify_waiters 不保存许可：必须先 enable 注册等待者，再读
-                // 活跃计数；否则最后一个 guard 在读数与首次 poll 之间 drop
-                // 时唤醒丢失，只能等满整个超时
-                notified.as_mut().enable();
-                if self.state.value.load(Ordering::Acquire) & ACTIVE_MASK == 0 {
-                    return;
-                }
-                notified.await;
-            }
-        };
-        let _ = tokio::time::timeout(timeout, wait).await;
    }
 }

@@ -106,10 +82,7 @@ impl ConnectionGuard for ActiveConnection {}

 impl Drop for ActiveConnection {
    fn drop(&mut self) {
-        let previous = self.state.value.fetch_sub(1, Ordering::AcqRel);
-        if previous & ACTIVE_MASK == 1 {
-            self.state.idle.notify_waiters();
-        }
+        self.state.value.fetch_sub(1, Ordering::AcqRel);
    }
 }

@@ -140,59 +113,26 @@ pub(crate) async fn serve_router(
    port: u16,
    cancellation: CancellationToken,
    connections: Arc<ConnectionTracker>,
-    drain_timeout: Duration,
 ) -> Result<(), ServeError> {
    let listener = bind_listener(&format!("{host}:{port}"))
        .await
        .map_err(ServeError::Bind)?;
    tracing::info!(target: "gateway_startup", pid = std::process::id(), host, port, "网关开始监听");

-    let shutdown_cancellation = cancellation.clone();
-    let shutdown_connections = Arc::clone(&connections);
-    let shutdown = async move {
-        wait_for_shutdown(&shutdown_cancellation).await;
-        shutdown_connections.begin_draining();
-    };
-    // axum 的优雅关闭会无限等待存量连接结束，而 SSE 等长响应体不观察
-    // 取消信号：慢客户端可以把关闭拖住直到被 SIGKILL
-    // 整个 drain（axum
-    // 优雅关闭 + 游离 guard 等待）共享同一个从取消信号起算的绝对截止点，
-    // 逾期即放弃等待，存量连接随进程退出终止
-    let drain_deadline_at = Arc::new(OnceLock::new());
-    let drain_deadline = {
-        let cancellation = cancellation.clone();
-        let deadline_at = Arc::clone(&drain_deadline_at);
-        async move {
-            cancellation.cancelled().await;
-            let deadline = tokio::time::Instant::now() + drain_timeout;
-            let _ = deadline_at.set(deadline);
-            tokio::time::sleep_until(deadline).await;
-        }
-    };
+    // 所有关闭入口都停止等待长请求；丢弃 serve 关闭监听，存量连接随进程退出终止
+    // 后台落盘由 Host 的 Worker 关闭阶段独立收尾
    let serve = axum::serve(
        listener,
        router.into_make_service_with_connect_info::<std::net::SocketAddr>(),
-    )
-    .with_graceful_shutdown(shutdown);
-    tokio::select! {
+    );
+    let result = tokio::select! {
        biased;
-        result = serve => {
-            result.map_err(ServeError::Serve)?;
-            connections.begin_draining();
-            let remaining = drain_deadline_at.get().map_or(drain_timeout, |deadline| {
-                deadline.saturating_duration_since(tokio::time::Instant::now())
-            });
-            connections.wait_until_idle(remaining).await;
-        }
-        () = drain_deadline => {
-            tracing::warn!(
-                target: "gateway_shutdown",
-                timeout_secs = drain_timeout.as_secs(),
-                "优雅 drain 超时，放弃等待存量连接"
-            );
-        }
-    }
-    Ok(())
+        () = wait_for_shutdown(&cancellation) => Ok(()),
+        result = serve => result.map_err(ServeError::Serve),
+    };
+    connections.begin_draining();
+    tracing::info!(target: "gateway_shutdown", "停止等待存量连接，后台落盘完成后退出进程");
+    result
 }

 pub(crate) async fn wait_for_shutdown(cancellation: &CancellationToken) {
@@ -538,6 +538,7 @@ impl SystemOperations for ProcessSystemOperations {
        let cancellation = self.cancellation.clone();
        drop(tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(500)).await;
+            // 留出受理响应返回时间，再走统一关闭流程
            cancellation.cancel();
        }));
        Ok(SystemOperationAccepted::Restart {
@@ -127,9 +127,9 @@ fn host_config_derives_update_paths_from_runtime_data_dir() {
 }

 #[test]
-fn host_config_rejects_zero_drain_window() {
+fn host_config_rejects_zero_worker_shutdown_window() {
    let mut config = valid_config();
-    config.drain_timeout_seconds = 0;
+    config.worker_shutdown_timeout_seconds = 0;

    assert!(
        config
@@ -223,7 +223,6 @@ fn valid_config() -> HostConfig {
            request_dump_retention_days: 1,
        },
        system_update,
-        drain_timeout_seconds: 30,
        worker_shutdown_timeout_seconds: 30,
    }
 }
@@ -51,7 +51,6 @@ fn logging_requires_at_least_one_sink() {
            request_dump_retention_days: 1,
        },
        system_update: SystemUpdateConfig::default(),
-        drain_timeout_seconds: 30,
        worker_shutdown_timeout_seconds: 30,
    };

@@ -329,7 +328,6 @@ fn logging_config(directory: PathBuf, request_dump: bool) -> HostConfig {
            request_dump_retention_days: 1,
        },
        system_update: SystemUpdateConfig::default(),
-        drain_timeout_seconds: 30,
        worker_shutdown_timeout_seconds: 30,
    }
 }
@@ -1,4 +1,4 @@
-//! 验证宿主服务关闭时的连接排空、超时与并发唤醒
+//! 验证统一关闭入口跳过会话等待、保留后台落盘及清理超时

 use std::sync::Arc;
 use std::sync::Mutex;
@@ -6,17 +6,19 @@ use std::time::Duration;

 use axum::{Router, routing::get};
 use futures::future::BoxFuture;
+use gateway_admin::ports::system::{
+    SystemOperationError, SystemRestartPreflight, SystemUpdateCandidate,
+};
 use gateway_core::diagnostics::{OperationalDiagnostics, OperationalFailure};
 use gateway_core::lifecycle::CancellationToken;
-use gateway_core::lifecycle::ConnectionLifecycle as _;
 use gateway_core::task::{
    DaemonRestartPolicy, DaemonTask, WorkerContribution, WorkerId, WorkerKind,
    WorkerLeaderLeasePort, WorkerLeaseAcquisition, WorkerLeaseError, WorkerLeaseRequest,
    WorkerRegistration, WorkerRunnable, WorkerTaskError,
 };
 use gateway_host::HostBundle;
-use gateway_host::serve::{ConnectionTracker, bind_listener};
-use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
+use gateway_host::serve::bind_listener;
+use tokio::io::AsyncWriteExt as _;
 use tokio::sync::{Semaphore, mpsc};

 use crate::support::host::{configuration, run_in_child};
@@ -28,65 +30,6 @@ fn host_bundle_serve_is_a_consuming_process_entrypoint() {
    assert_eq!(std::mem::size_of_val(&_serve), 0);
 }

-#[tokio::test(start_paused = true)]
-async fn wait_until_idle_should_return_immediately_without_active_connections() {
-    let tracker = ConnectionTracker::new(CancellationToken::new());
-    let started = tokio::time::Instant::now();
-
-    tracker.wait_until_idle(Duration::from_secs(30)).await;
-
-    assert_eq!(started.elapsed(), Duration::ZERO);
-}
-
-#[tokio::test(start_paused = true)]
-async fn wait_until_idle_should_give_up_at_timeout_while_connections_remain() {
-    let tracker = ConnectionTracker::new(CancellationToken::new());
-    let _guard = tracker.try_register().expect("register connection");
-    let started = tokio::time::Instant::now();
-
-    tracker.wait_until_idle(Duration::from_secs(30)).await;
-
-    assert_eq!(started.elapsed(), Duration::from_secs(30));
-}
-
-#[tokio::test]
-async fn wait_until_idle_should_wake_when_last_guard_drops_after_first_poll() {
-    let tracker = ConnectionTracker::new(CancellationToken::new());
-    let guard = tracker.try_register().expect("register connection");
-    let mut wait = Box::pin(tracker.wait_until_idle(Duration::from_secs(30)));
-    assert!(futures::poll!(wait.as_mut()).is_pending());
-
-    drop(guard);
-
-    tokio::time::timeout(Duration::from_secs(2), wait)
-        .await
-        .expect("woken by last guard drop instead of waiting out the timeout");
-}
-
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn wait_until_idle_should_observe_guard_drop_racing_the_idle_check() {
-    // 回归约束：notified 必须在读取活跃计数前完成注册（enable），否则最后
-    // 一个 guard 在计数检查与首次 poll 之间 drop 时唤醒丢失，只能等满整个
-    // 超时
-    // 多线程反复交错，任何一次丢唤醒都会撞上 2s 超时并触发断言
-    for _ in 0..256 {
-        let tracker = Arc::new(ConnectionTracker::new(CancellationToken::new()));
-        let guard = tracker.try_register().expect("register connection");
-        let waiter = tokio::spawn({
-            let tracker = Arc::clone(&tracker);
-            async move { tracker.wait_until_idle(Duration::from_secs(2)).await }
-        });
-        let dropper = std::thread::spawn(move || drop(guard));
-        let started = std::time::Instant::now();
-        waiter.await.expect("waiter completes");
-        dropper.join().expect("dropper completes");
-        assert!(
-            started.elapsed() < Duration::from_millis(1_900),
-            "idle wakeup was lost and the waiter slept until the drain timeout"
-        );
-    }
-}
-
 #[tokio::test]
 async fn bind_listener_should_retry_until_previous_listener_releases_port() {
    let holder = tokio::net::TcpListener::bind("127.0.0.1:0")
@@ -110,76 +53,148 @@ async fn bind_listener_should_retry_until_previous_listener_releases_port() {
 }

 #[test]
-fn http_drain_keeps_writer_running_until_the_request_finishes() {
+fn cancellation_skips_connections_but_flushes_and_joins_workers() {
    run_in_child(
-        "serve::http_drain_keeps_writer_running_until_the_request_finishes",
-        async {
-            let directory = tempfile::tempdir().unwrap();
-            let (host, address) = listening_host(directory.path()).await;
-            let cancellation = host.cancellation();
-            let connections = host.connection_lifecycle();
-            let probe = start_workers(&host, false);
-            probe.started.cancelled().await;
-            let (router, entered, finish_request) = request_router(&host, &probe);
-            let server = tokio::spawn(host.serve(router));
-            let mut client = connect_request(address).await;
-            entered.cancelled().await;
-
-            cancellation.cancel();
-            wait_for_draining(connections.as_ref()).await;
-            assert!(connections.try_register().is_err());
-            assert!(
-                tokio::time::timeout(Duration::from_millis(50), probe.cancelled.cancelled())
-                    .await
-                    .is_err(),
-                "writer cancelled before the in-flight request finished"
-            );
-            finish_request.add_permits(1);
-            let mut response = String::new();
-            client.read_to_string(&mut response).await.unwrap();
-            server.await.unwrap().unwrap();
-
-            assert!(response.starts_with("HTTP/1.1 200 OK"));
-            assert_eq!(*probe.writes.lock().unwrap(), [1]);
-            assert!(probe.finished.is_cancelled());
-        },
+        "serve::cancellation_skips_connections_but_flushes_and_joins_workers",
+        shutdown_with_pending_request(ShutdownTrigger::Cancellation),
    );
 }

 #[test]
-fn drain_timeout_starts_a_separate_worker_shutdown_phase() {
+fn self_restart_skips_connections_but_flushes_and_joins_workers() {
    run_in_child(
-        "serve::drain_timeout_starts_a_separate_worker_shutdown_phase",
+        "serve::self_restart_skips_connections_but_flushes_and_joins_workers",
+        shutdown_with_pending_request(ShutdownTrigger::SelfRestart),
+    );
+}
+
+#[cfg(unix)]
+#[test]
+fn sigterm_skips_connections_but_flushes_and_joins_workers() {
+    run_in_child(
+        "serve::sigterm_skips_connections_but_flushes_and_joins_workers",
+        shutdown_with_pending_request(ShutdownTrigger::Signal("-TERM")),
+    );
+}
+
+#[cfg(unix)]
+#[test]
+fn sigint_skips_connections_but_flushes_and_joins_workers() {
+    run_in_child(
+        "serve::sigint_skips_connections_but_flushes_and_joins_workers",
+        shutdown_with_pending_request(ShutdownTrigger::Signal("-INT")),
+    );
+}
+
+enum ShutdownTrigger {
+    Cancellation,
+    SelfRestart,
+    #[cfg(unix)]
+    Signal(&'static str),
+}
+
+async fn shutdown_with_pending_request(trigger: ShutdownTrigger) {
+    let directory = tempfile::tempdir().unwrap();
+    let reservation = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+    let address = reservation.local_addr().unwrap();
+    let mut config = configuration(directory.path());
+    config.listen.port = address.port();
+    config.system_update.deployment_mode = "docker".to_owned();
+    config.system_update.build_type = "source".to_owned();
+    config.system_update.self_restart_enabled = true;
+    config.system_update.update_lock_file = directory.path().join("update.lock");
+    let host = gateway_host::initialize(config).await.unwrap();
+    drop(reservation);
+    let system = host.system_operations();
+    let cancellation = host.cancellation();
+    let connections = host.connection_lifecycle();
+    let guard = connections.try_register().unwrap();
+    let probe = start_workers(&host, true);
+    probe.started.cancelled().await;
+    let (router, entered, _finish_request) = request_router(&host, &probe);
+    let server = tokio::spawn(host.serve(router));
+    let client = connect_request(address).await;
+    entered.cancelled().await;
+    probe.sender.send(7).unwrap();
+
+    match trigger {
+        ShutdownTrigger::Cancellation => cancellation.cancel(),
+        ShutdownTrigger::SelfRestart => {
+            system.restart(Arc::new(AllowRestart)).await.unwrap();
+        }
+        #[cfg(unix)]
+        ShutdownTrigger::Signal(signal) => {
+            // 真实请求进入后信号监听已被 poll，只向隔离的当前测试子进程发送信号
+            assert!(
+                std::process::Command::new("kill")
+                    .args([signal, &std::process::id().to_string()])
+                    .status()
+                    .unwrap()
+                    .success()
+            );
+        }
+    }
+    tokio::time::timeout(Duration::from_secs(2), probe.cancelled.cancelled())
+        .await
+        .expect("shutdown must not wait for the blocked HTTP request or live guard");
+    assert!(connections.is_draining());
+    assert!(connections.try_register().is_err());
+    drop(guard);
+    assert!(connections.try_register().is_err());
+    assert!(
+        tokio::net::TcpStream::connect(address).await.is_err(),
+        "listener must close before worker cleanup finishes"
+    );
+    assert!(
+        !server.is_finished(),
+        "worker cleanup still needs to finish"
+    );
+    probe.finish_shutdown.add_permits(1);
+    server.await.unwrap().unwrap();
+    assert_eq!(*probe.writes.lock().unwrap(), [7]);
+    assert!(probe.finished.is_cancelled());
+    // 请求仍未结束，独立进程退出时终止连接和挂起 handler
+    drop(client);
+}
+
+#[test]
+fn shutdown_aborts_workers_that_exceed_the_cleanup_budget() {
+    run_in_child(
+        "serve::shutdown_aborts_workers_that_exceed_the_cleanup_budget",
        async {
            let directory = tempfile::tempdir().unwrap();
            let (host, address) = listening_host(directory.path()).await;
            let cancellation = host.cancellation();
-            let connections = host.connection_lifecycle();
            let probe = start_workers(&host, true);
            probe.started.cancelled().await;
-            let (router, entered, finish_request) = request_router(&host, &probe);
+            let (router, entered, _finish_request) = request_router(&host, &probe);
            let server = tokio::spawn(host.serve(router));
            let client = connect_request(address).await;
            entered.cancelled().await;
-
-            let started = tokio::time::Instant::now();
            cancellation.cancel();
-            wait_for_draining(connections.as_ref()).await;
-            probe.cancelled.cancelled().await;
-            assert!(started.elapsed() >= Duration::from_secs(1));
-            assert!(!server.is_finished(), "worker cleanup was not joined");
-            assert!(!probe.finished.is_cancelled());
-            probe.finish_shutdown.add_permits(1);
-            server.await.unwrap().unwrap();
-            assert!(probe.finished.is_cancelled());

-            // 超时后真实连接可以仍在 axum 任务中；释放测试端，避免遗留挂起 handler
+            tokio::time::timeout(Duration::from_secs(3), server)
+                .await
+                .expect("worker cleanup budget must bound process shutdown")
+                .unwrap()
+                .unwrap();
+            probe.dropped.cancelled().await;
+            assert!(probe.cancelled.is_cancelled());
+            assert!(!probe.finished.is_cancelled());
            drop(client);
-            finish_request.add_permits(1);
        },
    );
 }

+struct AllowRestart;
+
+#[async_trait::async_trait]
+impl SystemRestartPreflight for AllowRestart {
+    async fn prepare(&self, _: Option<SystemUpdateCandidate>) -> Result<(), SystemOperationError> {
+        Ok(())
+    }
+}
+
 #[test]
 fn bind_failure_cancels_and_joins_workers_before_returning() {
    run_in_child(
@@ -291,12 +306,6 @@ async fn connect_request(address: std::net::SocketAddr) -> tokio::net::TcpStream
    client
 }

-async fn wait_for_draining(connections: &dyn gateway_core::lifecycle::ConnectionLifecycle) {
-    while !connections.is_draining() {
-        tokio::task::yield_now().await;
-    }
-}
-
 fn request_router(
    host: &HostBundle,
    probe: &WriterProbe,
@@ -319,7 +328,7 @@ fn request_router(
                    let _connection = connections.try_register().unwrap();
                    entered.cancel();
                    finish.acquire().await.unwrap().forget();
-                    // 对应请求结束后写入终态的时机；drain 超时后的写入允许被拒绝
+                    // 对应请求结束后写入终态的时机；关闭开始后的写入允许被拒绝
                    if sender.send(1).is_ok() {
                        "recorded"
                    } else {
@@ -59,7 +59,6 @@ pub fn configuration(directory: &Path) -> HostConfig {
            request_dump_retention_days: 1,
        },
        system_update: Default::default(),
-        drain_timeout_seconds: 1,
        worker_shutdown_timeout_seconds: 1,
    }
 }
@@ -4,8 +4,8 @@
    "version": 3,
    "replacement_version": 4,
    "introduced_in": "3.20.0",
-    "last_counted_in": "3.22.0",
-    "remaining_releases": 2,
+    "last_counted_in": "3.22.1",
+    "remaining_releases": 1,
    "migration": "将执行设置和来源中的 disable_fast 改为 fast_mode，使用 default、enabled、disabled 三态，并将 middleware 声明升级到 v4"
  },
  {
@@ -13,8 +13,8 @@
    "version": 1,
    "replacement_version": 2,
    "introduced_in": "3.20.0",
-    "last_counted_in": "3.22.0",
-    "remaining_releases": 2,
+    "last_counted_in": "3.22.1",
+    "remaining_releases": 1,
    "migration": "使用新版 SDK 读取 fast_mode 三态，并将 upstream_adapter 声明升级到 v2"
  }
 ]
@@ -1,5 +1,7 @@
 //! PostgreSQL 迁移屏障、连接池与会话级资源预算

+use std::time::Duration;
+
 use sqlx::{
    PgPool,
    postgres::{PgConnectOptions, PgPoolOptions},
@@ -37,11 +39,34 @@ pub async fn connect_and_migrate(
        )
        .await
        .map_err(|source| postgres_unavailable("connect PostgreSQL for migrations", source))?;
-    if let Err(error) = MIGRATOR.run(&migration_pool).await {
-        migration_pool.close().await;
+    let started = tokio::time::Instant::now();
+    tracing::info!(target: "gateway_startup", "开始检查并执行数据库迁移");
+    let result = {
+        let migration = MIGRATOR.run(&migration_pool);
+        tokio::pin!(migration);
+        let progress_interval = Duration::from_secs(15);
+        let mut progress = tokio::time::interval_at(started + progress_interval, progress_interval);
+        progress.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+        loop {
+            tokio::select! {
+                biased;
+                result = &mut migration => break result,
+                _ = progress.tick() => {
+                    // 迁移或迁移锁等待都不能被进度提示中断，不将已等待时间冒充完成比例
+                    tracing::info!(target: "gateway_startup", elapsed_seconds = started.elapsed().as_secs(),
+                        "数据库迁移仍在执行或等待锁，请等待完成，避免重复重启");
+                }
+            }
+        }
+    };
+    migration_pool.close().await;
+    if let Err(error) = result {
+        tracing::error!(target: "gateway_startup", elapsed_seconds = started.elapsed().as_secs_f64(),
+            "数据库迁移失败");
        return Err(postgres_unavailable("apply PostgreSQL migrations", error));
    }
-    migration_pool.close().await;
+    tracing::info!(target: "gateway_startup", elapsed_seconds = started.elapsed().as_secs_f64(),
+        "数据库迁移检查完成");

    connect_pool(connect_options, pool_config, false).await
 }
@@ -344,10 +344,15 @@ impl CodexCanonicalDecoder {
        }
        if event_type == Some("responsesapi.websocket_timing") && self.started && !self.completed {
            // 无 ID 的官方计时帧仅属于当前已开始的响应，复用连接的尾帧不能跨边界归属
-            let response_id = value
-                .get("response_id")
-                .or_else(|| value.pointer("/response/id"));
-            if response_id.is_none_or(|id| id.as_str() == self.response_id.as_deref()) {
+            let response_ids = [
+                value.get("response_id"),
+                value.pointer("/response/id"),
+                value.pointer("/timing_metrics/response_id"),
+            ];
+            if response_ids.into_iter().flatten().all(|id| {
+                id.as_str()
+                    .is_some_and(|id| Some(id) == self.response_id.as_deref())
+            }) {
                self.upstream_timing_metrics
                    .merge(ResponseTimingMetrics::from_event(&value));
            }
@@ -251,22 +251,54 @@ pub(crate) struct ResponseTimingMetrics {

 impl ResponseTimingMetrics {
    pub(crate) fn from_event(value: &Value) -> Self {
-        let metrics = value.get("timing_metrics");
-        let milliseconds = |key: &str| {
-            metrics?
-                .get(key)?
-                .as_f64()
-                .filter(|value| value.is_finite() && *value >= 0.0 && *value < i64::MAX as f64)
+        let Some(metrics) = value.get("timing_metrics") else {
+            return Self::default();
        };
+        let valid_milliseconds =
+            |value: &f64| value.is_finite() && *value >= 0.0 && *value < i64::MAX as f64;
+        let milliseconds =
+            |source: &Value, key: &str| source.get(key)?.as_f64().filter(valid_milliseconds);
+        if let Some(path) = metrics.get("critical_path") {
+            // 外层 logical_turn 可累计多次响应与工具等待，只取完整的当前响应阶段
+            if path.get("scope").and_then(Value::as_str) != Some("response")
+                || path.get("coverage").and_then(Value::as_str) != Some("complete")
+                || path.get("boundary_type").and_then(Value::as_str)
+                    != Some("actionable_output_item_done")
+            {
+                return Self::default();
+            }
+            // API 开销由推理前与其他处理构成，缺少任一分项时不能按零补齐
+            let overhead = milliseconds(path, "responses_pre_inference_ms")
+                .zip(milliseconds(path, "responses_other_ms"))
+                .map(|(before, other)| before + other)
+                .filter(valid_milliseconds);
+            return Self {
+                upstream_api_overhead_ms: overhead,
+                upstream_engine_ms: milliseconds(path, "engine_wall_ms"),
+                ..Self::default()
+            };
+        }
+        // 直接返回的专项指标保留原口径，明确标为整轮累计的值不能归入单次请求
+        if metrics
+            .get("timing_scope")
+            .is_some_and(|scope| scope.as_str() != Some("response"))
+        {
+            return Self::default();
+        }
        Self {
            upstream_api_overhead_ms: milliseconds(
+                metrics,
                "responses_duration_excl_engine_and_client_tool_time_ms",
            ),
-            upstream_engine_ms: milliseconds("engine_service_total_ms"),
-            upstream_engine_iapi_ttft_ms: milliseconds("engine_iapi_ttft_total_ms"),
-            upstream_engine_service_ttft_ms: milliseconds("engine_service_ttft_total_ms"),
-            upstream_engine_iapi_tbt_ms: milliseconds("engine_iapi_tbt_across_engine_calls_ms"),
+            upstream_engine_ms: milliseconds(metrics, "engine_service_total_ms"),
+            upstream_engine_iapi_ttft_ms: milliseconds(metrics, "engine_iapi_ttft_total_ms"),
+            upstream_engine_service_ttft_ms: milliseconds(metrics, "engine_service_ttft_total_ms"),
+            upstream_engine_iapi_tbt_ms: milliseconds(
+                metrics,
+                "engine_iapi_tbt_across_engine_calls_ms",
+            ),
            upstream_engine_service_tbt_ms: milliseconds(
+                metrics,
                "engine_service_tbt_across_engine_calls_ms",
            ),
        }
@@ -55,11 +55,71 @@ async fn finish_timings(
    .expect("complete upstream stream")
 }

+async fn observe_timing_events(websocket: bool, events: Vec<Value>) -> ProviderResponseTimings {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_provider_contract").await;
+    let (base_url, server) = if websocket {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let base_url = format!("http://{}", listener.local_addr().unwrap());
+        let server = tokio::spawn(async move {
+            let (socket, _) = listener.accept().await.unwrap();
+            let mut ws = crate::transport::accept_codex_test_websocket_with(
+                socket,
+                |request: &tokio_tungstenite::tungstenite::handshake::server::Request, response| {
+                    assert_eq!(
+                        request.headers()["x-responsesapi-include-timing-metrics"],
+                        "true"
+                    );
+                    response.headers_mut().insert(
+                        "sec-websocket-extensions",
+                        "permessage-deflate".parse().unwrap(),
+                    );
+                },
+            )
+            .await;
+            ws.next().await.unwrap().unwrap();
+            for event in events {
+                ws.send(Message::Text(event.to_string().into()))
+                    .await
+                    .unwrap();
+            }
+        });
+        (base_url, server)
+    } else {
+        let (base_url, release, _, server) = paused_chunked_sse_server(
+            events[..1]
+                .iter()
+                .map(|event| format!("data: {event}\n\n"))
+                .collect(),
+            events[1..]
+                .iter()
+                .map(|event| format!("data: {event}\n\n"))
+                .collect(),
+        )
+        .await;
+        release.send(()).unwrap();
+        (base_url, server)
+    };
+    let operation = if websocket {
+        generate_operation()
+    } else {
+        http_generate_operation()
+    };
+    let mut stream = provider_with_base_url(&store, base_url)
+        .execute(
+            planned_request("openai", operation),
+            timed_context(Instant::now() - Duration::from_secs(20), 2),
+        )
+        .await
+        .unwrap();
+    let timings = finish_timings(&mut stream, ProviderResponseTimings::default()).await;
+    server.await.unwrap();
+    timings
+}
+
 #[tokio::test]
 async fn official_response_duration_is_observed_independently_of_the_request_clock() {
    for websocket in [false, true] {
-        let store = Arc::new(MemoryAccountStore::default());
-        create_account(&store, "acct_provider_contract").await;
        let created = json!({"type":"response.created","response":{"id":"resp_official_time","created_at":100}});
        let completed = json!({"type":"response.completed","response":{"id":"resp_official_time","status":"completed","created_at":100,"completed_at":107,"output":[],"usage":{"input_tokens":1,"output_tokens":112,"total_tokens":113}}});
        let metrics = json!({"type":"responsesapi.websocket_timing","timing_metrics":{
@@ -81,62 +141,7 @@ async fn official_response_duration_is_observed_independently_of_the_request_clo
        }});
        let stale = json!({"type":"responsesapi.websocket_timing","timing_metrics":{"engine_service_total_ms":999.0}});
        let events = [stale, created, metrics, wrong_id, invalid, completed];
-        let (base_url, server) = if websocket {
-            let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-            let base_url = format!("http://{}", listener.local_addr().unwrap());
-            let server = tokio::spawn(async move {
-                let (socket, _) = listener.accept().await.unwrap();
-                let mut ws = crate::transport::accept_codex_test_websocket_with(
-                    socket,
-                    |request: &tokio_tungstenite::tungstenite::handshake::server::Request,
-                     response| {
-                        assert_eq!(
-                            request.headers()["x-responsesapi-include-timing-metrics"],
-                            "true"
-                        );
-                        response.headers_mut().insert(
-                            "sec-websocket-extensions",
-                            "permessage-deflate".parse().unwrap(),
-                        );
-                    },
-                )
-                .await;
-                ws.next().await.unwrap().unwrap();
-                for event in events {
-                    ws.send(Message::Text(event.to_string().into()))
-                        .await
-                        .unwrap();
-                }
-            });
-            (base_url, server)
-        } else {
-            let (base_url, release, _, server) = paused_chunked_sse_server(
-                events[..3]
-                    .iter()
-                    .map(|event| format!("data: {event}\n\n"))
-                    .collect(),
-                events[3..]
-                    .iter()
-                    .map(|event| format!("data: {event}\n\n"))
-                    .collect(),
-            )
-            .await;
-            release.send(()).unwrap();
-            (base_url, server)
-        };
-        let operation = if websocket {
-            generate_operation()
-        } else {
-            http_generate_operation()
-        };
-        let mut stream = provider_with_base_url(&store, base_url)
-            .execute(
-                planned_request("openai", operation),
-                timed_context(Instant::now() - Duration::from_secs(20), 2),
-            )
-            .await
-            .unwrap();
-        let timings = finish_timings(&mut stream, ProviderResponseTimings::default()).await;
+        let timings = observe_timing_events(websocket, events.to_vec()).await;
        assert_eq!(timings.upstream_response_ms, Some(7_000));
        assert_eq!(timings.upstream_api_overhead_ms, Some(120.25));
        assert_eq!(timings.upstream_engine_ms, Some(6400.0));
@@ -146,7 +151,133 @@ async fn official_response_duration_is_observed_independently_of_the_request_clo
        assert_eq!(timings.upstream_engine_service_tbt_ms, Some(20.12));
        assert_eq!(timings.first_token_ms, None);
        assert!(timings.first_event_ms.is_some_and(|value| value >= 20_000));
-        server.await.unwrap();
+    }
+}
+
+// 保留实测计时帧的层级和小数，响应身份使用虚构值，不包含请求正文
+fn response_critical_path_event() -> Value {
+    json!({"type":"responsesapi.websocket_timing","timing_metrics":{
+        "timing_scope":"logical_turn",
+        "response_id":"resp_critical_path",
+        "pre_inference_ms":211.691468,
+        "client_tool_pause_total_ms":3730.088369,
+        "total_turn_time_s":25.017284596,
+        "num_engine_calls":4,
+        "first_sampled_message_ttft_ms":null,
+        "critical_path":{
+            "scope":"response",
+            "coverage":"complete",
+            "boundary_type":"actionable_output_item_done",
+            "num_engine_calls":1,
+            "responses_harness_unblocked_ms":6792.094664,
+            "responses_pre_inference_ms":35.747883,
+            "engine_wall_ms":6710.68507,
+            "client_tool_pause_ms":0.0,
+            "responses_other_ms":45.66171099999974,
+            "engine_iapi_total_ms":null,
+            "taas_request_setup_total_ms":null,
+            "taas_provider_wait_total_ms":null,
+            "sampling_and_stream_total_ms":null,
+            "inference_residual_total_ms":null
+        }
+    }})
+}
+
+fn critical_path_response_events(metrics: Value) -> Vec<Value> {
+    vec![
+        json!({"type":"response.created","response":{"id":"resp_critical_path"}}),
+        metrics,
+        json!({"type":"response.completed","response":{"id":"resp_critical_path","status":"completed","created_at":100,"completed_at":107,"output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}),
+    ]
+}
+
+#[tokio::test]
+async fn response_critical_path_is_observed_without_logical_turn_totals_on_both_transports() {
+    for websocket in [false, true] {
+        let mut metrics = response_critical_path_event();
+        // 同帧即使携带整轮专项值，也不能与响应级 critical path 混算
+        metrics["timing_metrics"]["engine_service_total_ms"] = json!(25_000);
+        metrics["timing_metrics"]["engine_iapi_ttft_total_ms"] = json!(500);
+        let mut events = critical_path_response_events(metrics);
+        let mut wrong_id = response_critical_path_event();
+        wrong_id["timing_metrics"]["response_id"] = json!("resp_other");
+        wrong_id["timing_metrics"]["critical_path"]["engine_wall_ms"] = json!(999);
+        events.insert(2, wrong_id);
+        let timings = observe_timing_events(websocket, events).await;
+        assert_eq!(timings.upstream_engine_ms, Some(6710.68507));
+        assert!((timings.upstream_api_overhead_ms.unwrap() - 81.409594).abs() < 0.000001);
+        assert_eq!(timings.upstream_response_ms, Some(7000));
+        assert_eq!(timings.first_token_ms, None);
+        assert_eq!(timings.upstream_engine_iapi_ttft_ms, None);
+        assert_eq!(timings.upstream_engine_service_ttft_ms, None);
+        assert_eq!(timings.upstream_engine_iapi_tbt_ms, None);
+        assert_eq!(timings.upstream_engine_service_tbt_ms, None);
+    }
+}
+
+#[tokio::test]
+async fn critical_path_rejects_mismatched_identity_and_incomplete_or_cumulative_scope() {
+    for (pointer, value) in [
+        ("/timing_metrics/response_id", json!("resp_other")),
+        ("/timing_metrics/response_id", Value::Null),
+        ("/timing_metrics/critical_path/scope", json!("logical_turn")),
+        ("/timing_metrics/critical_path/scope", Value::Null),
+        ("/timing_metrics/critical_path/coverage", json!("partial")),
+        ("/timing_metrics/critical_path/coverage", Value::Null),
+        (
+            "/timing_metrics/critical_path/boundary_type",
+            json!("unknown"),
+        ),
+        ("/timing_metrics/critical_path", Value::Null),
+    ] {
+        let mut metrics = response_critical_path_event();
+        // 正确的顶层 ID 不能掩盖嵌套 ID 的冲突
+        metrics["response_id"] = json!("resp_critical_path");
+        *metrics.pointer_mut(pointer).unwrap() = value;
+        let timings = observe_timing_events(false, critical_path_response_events(metrics)).await;
+        assert_eq!(timings.upstream_engine_ms, None, "{pointer}");
+        assert_eq!(timings.upstream_api_overhead_ms, None, "{pointer}");
+    }
+    for identity in ["response_id", "response"] {
+        let mut metrics = response_critical_path_event();
+        metrics[identity] = if identity == "response" {
+            json!({"id":"resp_other"})
+        } else {
+            json!("resp_other")
+        };
+        let timings = observe_timing_events(false, critical_path_response_events(metrics)).await;
+        assert_eq!(timings.upstream_engine_ms, None);
+    }
+    let metrics = json!({"type":"responsesapi.websocket_timing","timing_metrics":{
+        "timing_scope":"logical_turn",
+        "engine_service_total_ms":25_000,
+        "responses_duration_excl_engine_and_client_tool_time_ms":500,
+        "engine_iapi_ttft_total_ms":200
+    }});
+    let timings = observe_timing_events(false, critical_path_response_events(metrics)).await;
+    assert_eq!(timings.upstream_engine_ms, None);
+    assert_eq!(timings.upstream_api_overhead_ms, None);
+    assert_eq!(timings.upstream_engine_iapi_ttft_ms, None);
+}
+
+#[tokio::test]
+async fn critical_path_keeps_missing_or_invalid_components_unknown_and_preserves_zero() {
+    for value in [Value::Null, json!(-1), json!("45"), json!(1e99)] {
+        let mut metrics = response_critical_path_event();
+        metrics["timing_metrics"]["critical_path"]["responses_other_ms"] = value;
+        let timings = observe_timing_events(false, critical_path_response_events(metrics)).await;
+        assert_eq!(timings.upstream_api_overhead_ms, None);
+        assert_eq!(timings.upstream_engine_ms, Some(6710.68507));
+    }
+    for (component, expected) in [(0.0, Some(0.0)), (5e18, None)] {
+        let mut metrics = response_critical_path_event();
+        let path = &mut metrics["timing_metrics"]["critical_path"];
+        path["responses_other_ms"] = json!(component);
+        path["responses_pre_inference_ms"] = json!(component);
+        path["engine_wall_ms"] = json!(0);
+        let timings = observe_timing_events(false, critical_path_response_events(metrics)).await;
+        assert_eq!(timings.upstream_api_overhead_ms, expected);
+        assert_eq!(timings.upstream_engine_ms, Some(0.0));
    }
 }

@@ -70,7 +70,8 @@ ENV PATH="/app/bin:/usr/lib/postgresql/18/bin:${PATH}"

 EXPOSE 8080

-HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
+# 首次升级的历史请求迁移需要启动宽限期
+HEALTHCHECK --interval=30s --timeout=5s --start-period=5m --retries=3 \
    CMD curl -fsS http://127.0.0.1:8080/healthz || exit 1

 CMD ["/app/bin/codex-proxy-rs"]
@@ -9,7 +9,7 @@
 | 安装与启动 | [手动安装](#手动安装) · [启动](#启动) · [公网访问](#公网访问) |
 | 接入客户端 | [客户端配置](#客户端配置) · [时区与客户端环境信息](#时区与客户端环境信息) · [登录与生图排查](#登录与生图排查) |
 | 日常运维 | [运行设置](#启动后的运行设置) · [小内存优化](#小内存优化) · [持久化与日志](#持久化与备份) · [请求排查](#请求错误排查) · [密码轮换](#密码语义) |
-| 更新与恢复 | [镜像升级](#镜像升级与源码构建) · [在线更新](#管理端在线更新) · [备份与恢复](#备份与恢复) · [优雅关停](#优雅关停) |
+| 更新与恢复 | [镜像升级](#镜像升级与源码构建) · [在线更新](#管理端在线更新) · [备份与恢复](#备份与恢复) · [停止与重启](#停止与重启) |
 | 运行源码 | [开发与源码联调](../docs/development.md) |

 ## 配置归属
@@ -196,6 +196,12 @@ curl -i http://127.0.0.1:8080/healthz

 `204 No Content` 表示应用、PostgreSQL、Redis 和后台任务的健康检查通过，不代表每个上游账号都可用

+HTTP 在数据库迁移和初始化完成后才监听。迁移期间启动日志每 15 秒报告已等待时间，
+包含执行迁移和等待迁移锁的时间，不代表完成比例；不要因暂时无法访问而反复重启。
+镜像和 Compose 默认提供 5 分钟健康检查启动宽限期，探测成功后立即转为健康，无需等满宽限期。
+历史数据较多时应提前延长 `healthcheck.start_period` 和部署工具的等待超时；只修改 Compose
+文件不会改变已创建容器的检查配置，需要在维护窗口重新创建容器
+
 不要把未脱敏的 `docker compose config` 或 `docker inspect` 输出上传到工单；它们会包含
 PostgreSQL/Redis 启动密码。日常校验使用 `config --quiet`

@@ -388,16 +394,14 @@ metrics_exporter = "none"
 其他客户端使用 Responses API、`/v1` Base URL 和代理密钥即可，不需要 Actor 标记。
 路由与请求格式见 [API 参考](../docs/api.md#3-openai-数据面与模型目录)

-## 优雅关停
+## 停止与重启

-收到停止信号后，应用先停止接收新连接并 drain 存量连接；整个 drain 共享一个从停止信号
-起算的绝对截止点（`host.drain_timeout_seconds`，默认 30 秒），逾期放弃等待，存量连接随
-进程退出终止。drain 结束后才关停后台 worker，预算为
-`host.worker_shutdown_timeout_seconds`（默认 30 秒），两段预算按最坏情况串联
+管理端重启、SIGTERM 和 SIGINT 使用同一关闭策略：停止接收新连接，不等待现有 HTTP、SSE、
+WebSocket 会话完成，进程退出时强制断开剩余连接。已入队的观测写入与后台清理保留
+`host.worker_shutdown_timeout_seconds`（默认 30 秒）的收尾预算；客户端在新进程就绪后重新连接

-Compose 的 `stop_grace_period` 为 75 秒，覆盖默认 30 秒 HTTP drain、30 秒 worker 收尾和额外调度
-余量。若调大任一应用超时，也必须把 `stop_grace_period` 调到大于两段超时之和；否则 Docker 会在
-宽限期结束时 SIGKILL
+Compose 的 `stop_grace_period` 为 75 秒，覆盖默认 30 秒后台收尾及进程清理余量。
+调大后台关闭预算时，也必须同步预留容器停止宽限期，否则 Docker 会在宽限期结束时 SIGKILL

 ## 本地开发

@@ -622,6 +626,8 @@ Compose 提供以下在线更新运行参数：
 - `CPR_ENABLE_SELF_RESTART=true`：更新或回滚完成后允许管理端请求重启；Docker 进程退出后由
  Compose 的 `restart: unless-stopped` 拉起新进程

+管理端自重启遵循统一的[停止与重启](#停止与重启)策略，客户端需在新进程完成迁移与初始化后重新连接
+
 Release 必须提供当前 OS/架构的 `codex-proxy-rs_<version>_<os>_<arch>.tar.gz` 与
 `checksums.txt`。服务会在替换前再次查询远端最新版本，校验下载 host、声明大小、SHA-256 和
 归档路径；二进制或静态资源任一替换失败时恢复旧文件。成功后的旧二进制和旧静态目录分别保留为
@@ -80,8 +80,7 @@ services:
    restart: unless-stopped
    user: '10001:10001'
    init: true
-    # 关闭是串行两段：HTTP drain（host.drain_timeout_seconds=30s）之后 worker join
-    # （host.worker_shutdown_timeout_seconds=30s），宽限期必须大于两者之和。
+    # 不等待存量会话；宽限期需覆盖后台收尾（默认 30s）及进程清理余量
    stop_grace_period: 75s
    depends_on:
      postgres:
@@ -116,7 +115,8 @@ services:
      interval: 30s
      timeout: 5s
      retries: 3
-      start_period: 10s
+      # 首次升级需回填历史请求，宽限期只容忍启动失败，探测成功后立即转为健康
+      start_period: 5m
    environment:
      HOME: /app
      CPR_SERVER_HOST: 0.0.0.0
@@ -37,7 +37,6 @@ host:
  #   deployment_mode: 'binary'
  #   self_restart_enabled: true
  #   executable_path: '/opt/codex-proxy-rs/codex-proxy-rs' # 可选：覆盖启动时解析的程序路径。
-  drain_timeout_seconds: 30
  worker_shutdown_timeout_seconds: 30

 store:
@@ -1806,10 +1806,15 @@ OpenAI 从当前响应的 `responsesapi.websocket_timing.timing_metrics` 读取
 WebSocket 握手默认请求专项计时，客户端显式提供的开关保留原值。
 各字段独立可选；响应开始前、终态后的计时和明确属于其他响应的事件不写入当前请求，重试时清空

+`critical_path` 仅采集 `scope=response`、`coverage=complete` 且边界为
+`actionable_output_item_done` 的响应级指标：引擎耗时取 `engine_wall_ms`，API 开销为
+`responses_pre_inference_ms` 与 `responses_other_ms` 之和，缺失分项不按零补齐。
+外层 `logical_turn` 累计时间不进入请求指标，也不用于补齐引擎 TTFT 或 Token 间隔
+
 | 字段 | 上游口径 |
 | --- | --- |
 | `upstreamApiOverheadMs` | API 排除引擎与客户端工具时间后的耗时 |
-| `upstreamEngineMs` | Engine Service 总耗时 |
+| `upstreamEngineMs` | 响应级引擎耗时，取完整 critical path 的墙钟时间或直接返回的 Engine Service 时间 |
 | `upstreamEngineIapiTtftMs` | Engine IAPI TTFT 总计 |
 | `upstreamEngineServiceTtftMs` | Engine Service TTFT 总计 |
 | `upstreamEngineIapiTbtMs` | 跨引擎调用的 IAPI Token 间隔 |
@@ -935,8 +935,10 @@ worker 复用连接测试探针执行真实上游调用，并按冻结代次处
 客户端下载解析由 Host 的 `ClientDistributionResolver` 实现，组合根注入 Admin；只在管理员请求时访问外部来源，
 不阻塞启动。失败时使用官方稳定地址，临时直链不持久化或代理下载，字段与来源规则见 [客户端下载 API](api.md#windows-客户端下载)

-关闭分为两段：先停止接收新连接并 drain HTTP/WS，再取消并等待 Worker；两段各有独立预算，Compose 的
-`stop_grace_period` 必须覆盖二者之和。超时后只丢弃仍未落盘的可恢复观测，不执行隐式业务重放
+Host 对管理端重启、OS 关闭信号和内部取消使用统一策略：停止监听并拒绝新连接注册，
+不等待存量会话，随后在独立预算内取消并等待 Worker 完成已入队观测写入和后台清理。
+剩余连接随进程退出终止，超时后只丢弃仍未落盘的可恢复观测，不执行隐式业务重放。
+Compose 的 `stop_grace_period` 必须覆盖后台关闭预算并留出进程清理余量

 ### 凭据、日志与诊断

@@ -1,23 +1,14 @@
-# v3.22.0
-
-## 用量与性能观测
-
- 用量记录统一保存请求、调度、传输与恢复观测，保留历史记录的诊断信息
- 延迟明细区分上游性能、请求观测与传输观测，展示上游返回的响应耗时、引擎耗时、TTFT 和 Token 间隔等指标
- 输出速率改用同一完成响应的上游时间戳计算，避免把网关等待与传输时间计入生成耗时；缺少有效上游时间戳时显示「—」
- 首字等待按首个输出项开始事件计时，不再把其他结构事件计为首字
+# v3.22.1

 ## 修复与改进

- 账号与 Key 表单的分组选择使用轻量查询，避免加载选择项时重复计算分组聚合统计
- Codex 配置导出统一使用 `config.toml` 中的 Provider Bearer 鉴权；CC Switch 导入对齐 4.0.4 支持的连接信息与用量查询字段
- 改进 Codex 0.162 协议兼容：Grok 转换采用最新的增量工具声明，修正同名工具及命名空间成员更新；流内错误的 `Retry-After` 解析与官方头部处理语义保持一致
- 配置发布复用未变化的插件实例，减少进程重建；完善插件状态迁移失败后的恢复与发布流程，保留进行中请求使用的实例
- 修复 PostgreSQL 事务开始被取消后，连接复用可能残留未结束事务的问题
+- 修复上游性能明细漏采：读取当前响应的关键路径计时，恢复引擎耗时与 API 开销，并校验响应归属，避免混入整轮累计时间
+- 改进数据库迁移期间的启动反馈：迁移或等待迁移锁时，每 15 秒记录已等待时间；镜像与 Compose 的健康检查启动宽限期延长至 5 分钟
+- 统一管理端重启、SIGTERM 与 SIGINT 的关闭流程，停止监听后结束剩余会话，保留后台观测写入与资源清理的收尾预算

 ## 升级事项

- 从 v3.21.3 升级会自动执行数据库迁移 `0025_request_observation.sql`，回填现有请求观测并调整存储结构；升级前备份数据库，停止共用该数据库的旧版本实例，为历史请求回填预留时间
- 迁移完成后不能仅切回旧二进制或镜像回滚；如需回退，应同时恢复升级前的数据库备份
- 历史记录缺少新增的上游性能数据时，相应指标保持空缺，输出速率显示「—」
- 手动配置 Codex 时使用完整的 `config.toml` 模板，无需另外生成 `auth.json`；CC Switch 4.0.4 会重建配置，导入连接后如需原生生图等模板设置，请核对并补齐对应配置
+- 从 v3.22.0 升级没有新增数据库迁移；从更早版本升级仍会执行相应历史迁移
+- 停止或重启会断开尚未完成的 HTTP、SSE 与 WebSocket 会话；`host.drain_timeout_seconds` 不再生效，后台收尾预算由 `host.worker_shutdown_timeout_seconds` 控制
+- 现有 Compose 部署需同步新的健康检查配置并重建容器，才能应用 5 分钟的启动宽限期
+- 新增采集的上游性能数据只影响升级后的请求，历史记录中的空缺不会自动补齐
@@ -1 +1 @@
-version: "3.22.0"
+version: "3.22.1"