perf(test): reduce backend fixture and timeout overhead

Reuse immutable plugin archives across nextest processes and share admin fixtures within test matrices. Reduce log scanning, hashing, compression, database setup, and SSE scanning costs while preserving isolation and boundary coverage.

Use controllable clocks for recovery budgets and quota scheduling, retaining real socket and HTTP verification.

Validation: 3200 backend tests, 132 plugin persistence tests, 54 Redis ACL tests, 30 clock stress iterations, clippy, and formatting checks passed. Local full-suite runtime improved from 182.8s to 74.0s at unchanged concurrency.

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

backend/Cargo.lock

@@ -1834,6 +1834,7 @@ dependencies = [
 name = "gateway-protocol"
 version = "0.0.0"
 dependencies = [
+ "memchr",
  "serde",
  "serde_json",
  "thiserror",

backend/Cargo.toml

@@ -35,6 +35,20 @@ needless_collect = { level = "deny", priority = 7 }
 [profile.dev]
 debug = true
 
+# 测试保留真实密码参数和压缩数据量，优化热点依赖以减少 fixture 的计算开销。
+# test 继承 dev；workspace 代码仍保留默认调试与溢出检查。
+[profile.dev.package.argon2]
+opt-level = 2
+
+[profile.dev.package.blake2]
+opt-level = 2
+
+[profile.dev.package."sha2@0.11.0"]
+opt-level = 2
+
+[profile.dev.package."miniz_oxide@0.9.1"]
+opt-level = 2
+
 [profile.release]
 debug = false
 strip = "symbols"

backend/apps/gateway/tests/architecture.rs

@@ -498,6 +498,8 @@ fn root_test_scenario_allowed(member: &str, module: &Path) -> bool {
 
 fn assert_module_tree(root: &Path, crate_roots: &[&str]) {
     let files = super::rust_files(root);
+    // 同一父模块的所有子文件复用一次语法解析，计数仍保留重复声明检查。
+    let mut declarations = BTreeMap::<PathBuf, BTreeMap<String, usize>>::new();
     for relative in &files {
         if crate_roots
             .iter()
@@ -560,7 +562,14 @@ fn assert_module_tree(root: &Path, crate_roots: &[&str]) {
         };
         let declaration_count = declaration_parents
             .iter()
-            .map(|path| external_module_declaration_count(path, module_name))
+            .map(|path| {
+                declarations
+                    .entry(path.clone())
+                    .or_insert_with(|| external_module_declarations(path))
+                    .get(module_name)
+                    .copied()
+                    .unwrap_or_default()
+            })
             .sum::<usize>();
         assert_eq!(
             declaration_count,
@@ -571,23 +580,21 @@ fn assert_module_tree(root: &Path, crate_roots: &[&str]) {
     }
 }
 
-fn external_module_declaration_count(path: &Path, module_name: &str) -> usize {
+fn external_module_declarations(path: &Path) -> BTreeMap<String, usize> {
     if !path.is_file() {
-        return 0;
+        return BTreeMap::new();
     }
     let source = fs::read_to_string(path).expect("read parent module source");
     let syntax = syn::parse_file(&source).expect("parse parent module source");
-    syntax
-        .items
-        .iter()
-        .filter(|item| {
-            matches!(
-                item,
-                Item::Mod(module)
-                    if module.content.is_none() && module.ident == module_name
-            )
-        })
-        .count()
+    let mut declarations = BTreeMap::new();
+    for item in syntax.items {
+        if let Item::Mod(module) = item
+            && module.content.is_none()
+        {
+            *declarations.entry(module.ident.to_string()).or_default() += 1;
+        }
+    }
+    declarations
 }
 
 pub(super) fn backend_root() -> PathBuf {

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

@@ -859,6 +859,7 @@ async fn streaming_encodes_first_frame_before_commit_and_http_delivery() {
 
 #[tokio::test]
 async fn streaming_dropped_unknown_initial_batch_discards_without_committing() {
+    let admin = crate::admin::AdminTestFixture::new().await.services;
     let trace = Arc::new(Trace::default());
     let dropped_raw = Bytes::from_static(
         b"event: response.future_metadata\ndata: {\"type\":\"response.future_metadata\",\"dropped\":true}\n\n",
@@ -885,7 +886,8 @@ async fn streaming_dropped_unknown_initial_batch_discards_without_committing() {
         ResponseFrameAction::Continue,
     ]));
 
-    let response = response_with_middleware("openai", session, true, Some(middleware)).await;
+    let response =
+        response_with_middleware(&admin, "openai", session, true, Some(middleware)).await;
     let body = to_bytes(response.into_body(), usize::MAX)
         .await
         .expect("read filtered SSE response");
@@ -912,6 +914,7 @@ async fn streaming_dropped_unknown_initial_batch_discards_without_committing() {
 
 #[tokio::test]
 async fn streaming_modified_sse_rejects_out_of_order_tool_events_without_replay() {
+    let admin = crate::admin::AdminTestFixture::new().await.services;
     let trace = Arc::new(Trace::default());
     let added = super::openai_wire_event(
         vec![],
@@ -950,7 +953,8 @@ async fn streaming_modified_sse_rejects_out_of_order_tool_events_without_replay(
         ResponseFrameAction::Replace(Bytes::from(out_of_order)),
     ]));
 
-    let response = response_with_middleware("openai", session, true, Some(middleware)).await;
+    let response =
+        response_with_middleware(&admin, "openai", session, true, Some(middleware)).await;
     let body = to_bytes(response.into_body(), usize::MAX)
         .await
         .expect("read rejected transformed SSE");
@@ -970,6 +974,7 @@ async fn streaming_modified_sse_rejects_out_of_order_tool_events_without_replay(
 
 #[tokio::test]
 async fn streaming_modified_sse_rejects_wrong_tool_association_without_replay() {
+    let admin = crate::admin::AdminTestFixture::new().await.services;
     let trace = Arc::new(Trace::default());
     let added = super::openai_wire_event(
         vec![],
@@ -1022,7 +1027,8 @@ async fn streaming_modified_sse_rejects_wrong_tool_association_without_replay()
         ResponseFrameAction::Replace(Bytes::from(wrong_association)),
     ]));
 
-    let response = response_with_middleware("openai", session, true, Some(middleware)).await;
+    let response =
+        response_with_middleware(&admin, "openai", session, true, Some(middleware)).await;
     let body = to_bytes(response.into_body(), usize::MAX)
         .await
         .expect("read rejected transformed SSE");
@@ -1046,6 +1052,7 @@ async fn streaming_modified_sse_rejects_wrong_tool_association_without_replay()
 
 #[tokio::test]
 async fn streaming_modified_sse_cannot_change_terminal_classification() {
+    let admin = crate::admin::AdminTestFixture::new().await.services;
     let trace = Arc::new(Trace::default());
     let incomplete = encode_sse_event(
         "response.incomplete",
@@ -1071,7 +1078,8 @@ async fn streaming_modified_sse_cannot_change_terminal_classification() {
         ResponseFrameAction::Replace(Bytes::from(incomplete)),
     ]));
 
-    let response = response_with_middleware("openai", session, true, Some(middleware)).await;
+    let response =
+        response_with_middleware(&admin, "openai", session, true, Some(middleware)).await;
     let body = to_bytes(response.into_body(), usize::MAX)
         .await
         .expect("read terminal-class rejection");
@@ -1090,6 +1098,7 @@ async fn streaming_modified_sse_cannot_change_terminal_classification() {
 
 #[tokio::test]
 async fn streaming_terminal_drop_fails_before_commit() {
+    let admin = crate::admin::AdminTestFixture::new().await.services;
     let trace = Arc::new(Trace::default());
     let session = FakeSession::streaming(
         Arc::clone(&trace),
@@ -1104,7 +1113,8 @@ async fn streaming_terminal_drop_fails_before_commit() {
     let middleware =
         Arc::new(responses_middleware().with_response_actions(vec![ResponseFrameAction::Drop]));
 
-    let response = response_with_middleware("openai", session, true, Some(middleware)).await;
+    let response =
+        response_with_middleware(&admin, "openai", session, true, Some(middleware)).await;
 
     assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
     let body = to_bytes(response.into_body(), usize::MAX)
@@ -2373,6 +2383,7 @@ async fn buffered_response_never_merges_items_from_a_different_wire_response() {
 
 #[tokio::test]
 async fn buffered_response_middleware_replaces_body_and_filters_unsafe_headers() {
+    let admin = crate::admin::AdminTestFixture::new().await.services;
     let trace = Arc::new(Trace::default());
     let session = FakeSession::buffered(Arc::clone(&trace), vec![started(), completed()]);
     let middleware = Arc::new(
@@ -2388,7 +2399,8 @@ async fn buffered_response_middleware_replaces_body_and_filters_unsafe_headers()
             ]),
     );
 
-    let response = response_with_middleware("openai", session, false, Some(middleware)).await;
+    let response =
+        response_with_middleware(&admin, "openai", session, false, Some(middleware)).await;
     assert_eq!(response.status(), StatusCode::OK);
     assert_eq!(response.headers()["x-policy-result"], "replaced");
     assert!(response.headers().get(AUTHORIZATION).is_none());
@@ -2408,6 +2420,7 @@ async fn buffered_response_middleware_replaces_body_and_filters_unsafe_headers()
 
 #[tokio::test]
 async fn buffered_response_middleware_rejects_missing_terminal_fields_or_duplicate_tool_identity() {
+    let admin = crate::admin::AdminTestFixture::new().await.services;
     for replacement in [
         json!({"id": "resp_test", "output": []}),
         json!({"id": "resp_test", "status": "incomplete", "output": []}),
@@ -2432,7 +2445,8 @@ async fn buffered_response_middleware_rejects_missing_terminal_fields_or_duplica
             )),
         ]));
 
-        let response = response_with_middleware("openai", session, false, Some(middleware)).await;
+        let response =
+            response_with_middleware(&admin, "openai", session, false, Some(middleware)).await;
         assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
         let body = to_bytes(response.into_body(), usize::MAX)
             .await
@@ -2550,6 +2564,7 @@ struct SessionExecution {
 
 #[tokio::test]
 async fn chatgpt_remote_responses_should_only_skip_missing_desktop_versions() {
+    let admin = crate::admin::AdminTestFixture::new().await.services;
     for name in [
         "codex_chatgpt_android_remote",
         "codex_chatgpt_ios_remote",
@@ -2589,18 +2604,20 @@ async fn chatgpt_remote_responses_should_only_skip_missing_desktop_versions() {
             if let Some(version) = version {
                 request = request.header("version", version);
             }
-            let response = api_router(execution.clone())
-                .await
-                .oneshot(
-                    request
-                        .body(Body::from(
-                            json!({"model": "model-a", "input": "synthetic", "stream": true})
-                                .to_string(),
-                        ))
-                        .unwrap(),
-                )
-                .await
-                .unwrap();
+            let response = crate::openai::api_router_with_admin_and_execution(
+                admin.clone(),
+                execution.clone(),
+            )
+            .oneshot(
+                request
+                    .body(Body::from(
+                        json!({"model": "model-a", "input": "synthetic", "stream": true})
+                            .to_string(),
+                    ))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
             let status = response.status();
             let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
             if let Some(code) = expected_code {
@@ -2698,14 +2715,16 @@ impl ExecutionService for SessionExecution {
 }
 
 async fn response(
+    admin: &gateway_admin::AdminServices,
     provider: &str,
     session: FakeSession,
     streaming: bool,
 ) -> axum::response::Response {
-    response_with_middleware(provider, session, streaming, None).await
+    response_with_middleware(admin, provider, session, streaming, None).await
 }
 
 async fn response_with_middleware(
+    admin: &gateway_admin::AdminServices,
     provider: &str,
     session: FakeSession,
     streaming: bool,
@@ -2716,8 +2735,7 @@ async fn response_with_middleware(
         session: Mutex::new(Some(Box::new(session))),
         middleware: middleware.map(|middleware| middleware.frozen()),
     });
-    api_router(execution)
-        .await
+    crate::openai::api_router_with_admin_and_execution(admin.clone(), execution)
         .oneshot(
             Request::post("/v1/responses")
                 .header(AUTHORIZATION, "Bearer sk_correlation_test")
@@ -2755,6 +2773,7 @@ fn failed_session(streaming: bool, error: EngineError) -> FakeSession {
 
 #[tokio::test]
 async fn response_ids_are_provider_and_client_independent() {
+    let admin = crate::admin::AdminTestFixture::new().await.services;
     let cases: &[(&[(&str, &str)], &str)] = &[
         (&[], MODEL_REQUEST_ID),
         (&[("x-request-id", "upstream-primary")], "upstream-primary"),
@@ -2798,7 +2817,7 @@ async fn response_ids_are_provider_and_client_independent() {
                     FakeSession::buffered(trace, vec![started(), completed()])
                 }
                 .with_response_headers(headers(upstream_headers));
-                let response = response(provider, session, streaming).await;
+                let response = response(&admin, provider, session, streaming).await;
                 assert_eq!(response.status(), StatusCode::OK);
                 assert_eq!(response.headers()["x-request-id"], *expected);
                 assert_eq!(response.headers()["x-gateway-request-id"], MODEL_REQUEST_ID);
@@ -2815,6 +2834,7 @@ async fn response_ids_are_provider_and_client_independent() {
 
 #[tokio::test]
 async fn failure_ids_prefer_the_actual_error_over_session_headers() {
+    let admin = crate::admin::AdminTestFixture::new().await.services;
     for provider in ["openai", "xai"] {
         for streaming in [false, true] {
             for upstream_headers in [vec![], headers(&[("x-oai-request-id", "actual-error")])] {
@@ -2837,7 +2857,7 @@ async fn failure_ids_prefer_the_actual_error_over_session_headers() {
                         );
                 let session = failed_session(streaming, EngineError::Provider(error))
                     .with_response_headers(headers(&[("x-request-id", "earlier-session")]));
-                let response = response(provider, session, streaming).await;
+                let response = response(&admin, provider, session, streaming).await;
                 assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);
                 assert_eq!(response.headers()["x-request-id"], expected);
                 assert_eq!(response.headers()["x-gateway-request-id"], MODEL_REQUEST_ID);
@@ -2853,7 +2873,7 @@ async fn failure_ids_prefer_the_actual_error_over_session_headers() {
                     ("x-request-id", "earlier-session"),
                     ("x-oai-request-id", "earlier-alias"),
                 ]));
-            let response = response(provider, session, streaming).await;
+            let response = response(&admin, provider, session, streaming).await;
             assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
             assert_eq!(response.headers()["x-request-id"], "body-read-error");
             assert_eq!(response.headers().get_all("x-request-id").iter().count(), 1);
@@ -2865,6 +2885,7 @@ async fn failure_ids_prefer_the_actual_error_over_session_headers() {
 
 #[tokio::test]
 async fn local_execution_failure_uses_the_existing_model_id() {
+    let admin = crate::admin::AdminTestFixture::new().await.services;
     for provider in ["openai", "xai"] {
         for streaming in [false, true] {
             let session =
@@ -2873,7 +2894,7 @@ async fn local_execution_failure_uses_the_existing_model_id() {
                     ("x-oai-request-id", "opening-alias"),
                     ("x-codex-turn-state", "retained-turn-state"),
                 ]));
-            let response = response(provider, session, streaming).await;
+            let response = response(&admin, provider, session, streaming).await;
             assert_eq!(response.headers()["x-request-id"], MODEL_REQUEST_ID);
             assert_eq!(response.headers()["x-gateway-request-id"], MODEL_REQUEST_ID);
             assert!(response.headers().get("x-oai-request-id").is_none());
@@ -2887,6 +2908,7 @@ async fn local_execution_failure_uses_the_existing_model_id() {
 
 #[tokio::test]
 async fn compressed_http_request_above_default_limit_should_reach_execution_after_increase() {
+    let admin = crate::admin::AdminTestFixture::new().await.services;
     let body = json!({"model": "model-a", "input": "x".repeat(64 * 1024 * 1024)}).to_string();
     let compressed = super::request::encode_body("zstd", body.as_bytes());
     for limit in [64 * 1024 * 1024, 128 * 1024 * 1024] {
@@ -2899,8 +2921,7 @@ async fn compressed_http_request_above_default_limit_should_reach_execution_afte
                 limit,
             ),
         });
-        let response = api_router(execution)
-            .await
+        let response = crate::openai::api_router_with_admin_and_execution(admin.clone(), execution)
             .oneshot(
                 Request::post("/v1/responses")
                     .header(AUTHORIZATION, "Bearer sk_context_test")

backend/crates/gateway-api/tests/openai/router.rs

@@ -13,6 +13,11 @@ const REMOVED_BODY_LIMIT_BYTES: usize = 16 * 1024 * 1024;
 async fn browser_origin_controls_http_admin_sessions_without_configuration() {
     use axum::body::to_bytes;
     use serde_json::{Value, json};
+    let app = api_router_with_origins(ModelsExecution::new(), Vec::new())
+        .await
+        .layer(axum::Extension(axum::extract::ConnectInfo(
+            std::net::SocketAddr::from(([127, 0, 0, 1], 41000)),
+        )));
     for (origins, secure) in [
         (vec!["http://admin.example.test"], false),
         (vec!["http://192.0.2.1:8080"], false),
@@ -29,11 +34,6 @@ async fn browser_origin_controls_http_admin_sessions_without_configuration() {
             true,
         ),
     ] {
-        let app = api_router_with_origins(ModelsExecution::new(), Vec::new())
-            .await
-            .layer(axum::Extension(axum::extract::ConnectInfo(
-                std::net::SocketAddr::from(([127, 0, 0, 1], 41000)),
-            )));
         let request = |path: &str| {
             let mut builder = Request::post(path)
                 .header("content-type", "application/json")
@@ -99,6 +99,7 @@ async fn browser_origin_controls_http_admin_sessions_without_configuration() {
         }
 
         let response = app
+            .clone()
             .oneshot(
                 Request::get("/api/auth/status")
                     .header("cookie", &session)

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

@@ -14,26 +14,45 @@ struct State {
 }
 
 /// 换账号、换传输和内部重试共享同一预算；成功发送后的协议恢复不重置它。
-#[derive(Debug, Clone, Default)]
-pub struct ConnectionBudget(Arc<Mutex<State>>);
+#[derive(Debug, Clone)]
+pub struct ConnectionBudget {
+    state: Arc<Mutex<State>>,
+    now: fn() -> Instant,
+}
+
+impl Default for ConnectionBudget {
+    fn default() -> Self {
+        Self::with_clock(Instant::now)
+    }
+}
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
 #[error("connection recovery budget exhausted")]
 pub struct ConnectionBudgetExhausted;
 
 impl ConnectionBudget {
+    /// 绑定单调时钟；克隆预算继续共享状态和时钟，不依赖具体异步运行时。
+    #[must_use]
+    pub fn with_clock(now: fn() -> Instant) -> Self {
+        Self {
+            state: Arc::default(),
+            now,
+        }
+    }
+
     /// 在可能建立连接前预约一次 opening；返回本次连接允许使用的剩余时间。
     /// HTTP 连接池隐藏了是否冷建连，因此一次 HTTP opening 也计入次数上限。
     pub fn begin(&self) -> Result<Option<Duration>, ConnectionBudgetExhausted> {
         let mut state = self
-            .0
+            .state
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner);
         if state.completed {
             return Ok(None);
         }
-        let started = *state.started.get_or_insert_with(Instant::now);
-        let remaining = STARTUP_TIMEOUT.saturating_sub(started.elapsed());
+        let started = *state.started.get_or_insert_with(self.now);
+        let remaining =
+            STARTUP_TIMEOUT.saturating_sub((self.now)().saturating_duration_since(started));
         if state.openings >= MAX_OPENINGS || remaining.is_zero() {
             return Err(ConnectionBudgetExhausted);
         }
@@ -43,7 +62,7 @@ impl ConnectionBudget {
 
     /// 仅连接成功不足以清零；确认业务发送后退出新增的 NotSent 恢复策略。
     pub fn complete(&self) {
-        self.0
+        self.state
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner)
             .completed = true;
@@ -52,14 +71,14 @@ impl ConnectionBudget {
     #[must_use]
     pub fn remaining(&self) -> Option<Duration> {
         let state = self
-            .0
+            .state
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner);
         if state.completed || state.openings >= MAX_OPENINGS {
             return None;
         }
         let remaining = state.started.map_or(STARTUP_TIMEOUT, |started| {
-            STARTUP_TIMEOUT.saturating_sub(started.elapsed())
+            STARTUP_TIMEOUT.saturating_sub((self.now)().saturating_duration_since(started))
         });
         (!remaining.is_zero()).then_some(remaining)
     }
@@ -67,29 +86,29 @@ impl ConnectionBudget {
     #[must_use]
     pub fn exhausted(&self) -> bool {
         let state = self
-            .0
+            .state
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner);
         !state.completed
             && (state.openings >= MAX_OPENINGS
-                || state
-                    .started
-                    .is_some_and(|started| started.elapsed() >= STARTUP_TIMEOUT))
+                || state.started.is_some_and(|started| {
+                    (self.now)().saturating_duration_since(started) >= STARTUP_TIMEOUT
+                }))
     }
 
     /// 首次建连后的等待也计入恢复窗口；不缩短已发送请求的生成时间。
     #[must_use]
     pub fn startup_remaining(&self) -> Option<Duration> {
         let state = self
-            .0
+            .state
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner);
         if state.completed {
             return None;
         }
-        state
-            .started
-            .map(|started| STARTUP_TIMEOUT.saturating_sub(started.elapsed()))
+        state.started.map(|started| {
+            STARTUP_TIMEOUT.saturating_sub((self.now)().saturating_duration_since(started))
+        })
     }
 
     /// 额外 attempt 最多三个；抖动由请求 ID 派生，重试不依赖运行时随机源。

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

@@ -1,5 +1,30 @@
 use gateway_core::engine::connection::ConnectionBudget;
 
+#[test]
+fn elapsed_recovery_window_is_shared_without_waiting_for_real_time() {
+    use std::{
+        cell::Cell,
+        time::{Duration, Instant},
+    };
+    thread_local! {
+        static NOW: Cell<Instant> = Cell::new(Instant::now());
+    }
+    fn now() -> Instant {
+        NOW.get()
+    }
+    let request = ConnectionBudget::with_clock(now);
+    let fallback = request.clone();
+    assert_eq!(request.begin().unwrap(), Some(Duration::from_secs(30)));
+    NOW.set(now() + Duration::from_secs(29));
+    assert_eq!(fallback.begin().unwrap(), Some(Duration::from_secs(1)));
+    assert_eq!(request.startup_remaining(), Some(Duration::from_secs(1)));
+    NOW.set(now() + Duration::from_secs(1));
+    assert!(request.exhausted());
+    assert!(fallback.begin().is_err());
+    assert!(request.remaining().is_none());
+    assert_eq!(fallback.startup_remaining(), Some(Duration::ZERO));
+}
+
 #[test]
 fn all_transports_share_one_opening_budget_but_other_requests_do_not() {
     let request = ConnectionBudget::default();

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

@@ -1,6 +1,8 @@
 use super::*;
 use std::time::SystemTime;
 
+const ROTATION_RECORDS: usize = 3;
+
 #[test]
 fn retention_preserves_complete_dates_across_compression_rotation_and_restart() {
     if env::var_os(CHILD_PROCESS_ENV).is_some() {
@@ -9,7 +11,8 @@ fn retention_preserves_complete_dates_across_compression_rotation_and_restart()
             true,
             || {
                 let payload = "x".repeat(1024 * 1024);
-                for sequence in 0..25 {
+                // 三条记录已跨过 1 MiB 轮转边界；保留日期的 25 个分段由下方单独构造。
+                for sequence in 0..ROTATION_RECORDS {
                     tracing::info!(target: REQUEST_DUMP_LOG_TARGET, sequence, payload, "retention record");
                     tracing::info!(target: APPLICATION_LOG_TARGET, sequence, payload, "retention record");
                     tracing::info!(target: OAUTH_RECOVERY_LOG_TARGET, sequence, payload, "retention record");
@@ -89,18 +92,47 @@ fn retention_preserves_complete_dates_across_compression_rotation_and_restart()
             REQUEST_DUMP_LOG_FILE_PREFIX,
         ] {
             let body = read_log_file_set(directory.path(), prefix);
-            for sequence in 0..25 {
-                assert_eq!(
-                    body.matches(&format!(r#""sequence":{sequence},"#)).count(),
-                    run
-                );
-            }
-            for segment in 1..=25 {
-                assert!(
-                    body.contains(&format!("boundary-{segment}\n")),
-                    "whole boundary date must survive"
-                );
+            let mut sequences = [0; ROTATION_RECORDS];
+            let mut boundaries = std::collections::BTreeSet::new();
+            for line in body.lines() {
+                if let Some(segment) = line.strip_prefix("boundary-") {
+                    boundaries.insert(segment.parse::<usize>().unwrap());
+                } else if line.starts_with('{') {
+                    #[derive(serde::Deserialize)]
+                    struct Record {
+                        fields: Fields,
+                    }
+                    #[derive(serde::Deserialize)]
+                    struct Fields {
+                        sequence: Option<usize>,
+                    }
+                    if let Some(sequence) = serde_json::from_str::<Record>(line)
+                        .unwrap()
+                        .fields
+                        .sequence
+                    {
+                        sequences[sequence] += 1;
+                    }
+                }
             }
+            assert_eq!(sequences, [run; ROTATION_RECORDS]);
+            assert_eq!(
+                boundaries,
+                (1..=25).collect(),
+                "whole boundary date must survive"
+            );
+            let today_segments = fs::read_dir(directory.path())
+                .unwrap()
+                .filter(|entry| {
+                    entry
+                        .as_ref()
+                        .unwrap()
+                        .file_name()
+                        .to_string_lossy()
+                        .starts_with(&format!("{prefix}{today}"))
+                })
+                .count();
+            assert!(today_segments >= 2, "records must exercise size rotation");
         }
     }
 }

backend/crates/gateway-host/tests/system_update/mod.rs

@@ -212,10 +212,11 @@ async fn restart_should_use_startup_path_after_running_executable_is_renamed() {
     const CHILD_ENV: &str = "CPR_TEST_RESTART_RENAMED_EXECUTABLE_CHILD";
     let executable = std::env::current_exe().expect("test executable");
     if std::env::var_os(CHILD_ENV).is_none() {
-        // 只重命名隔离副本，避免修改 Cargo 的测试程序或影响并行测试。
-        let directory = tempfile::tempdir().expect("isolated test directory");
+        // 用例只移动目录项，不修改旧程序内容；同盘硬链接避免复制整个调试二进制。
+        let directory =
+            tempfile::tempdir_in(executable.parent().unwrap()).expect("isolated test directory");
         let child = directory.path().join("restart-test");
-        fs::copy(&executable, &child).expect("copy test executable");
+        fs::hard_link(&executable, &child).expect("link isolated test executable");
         let output = std::process::Command::new(child)
             .args([
                 "--exact",
@@ -246,7 +247,7 @@ async fn restart_should_use_startup_path_after_running_executable_is_renamed() {
         return;
     }
 
-    let fixture = Fixture::new();
+    let fixture = Fixture::new_in(executable.parent().unwrap());
     fs::rename(executable, fixture.executable()).expect("move isolated running executable");
     let mut config = fixture.config("http://127.0.0.1:1/repos");
     config.executable_path = None;
@@ -1950,7 +1951,11 @@ struct Fixture {
 
 impl Fixture {
     fn new() -> Self {
-        let root = tempfile::tempdir().expect("system update root");
+        Self::new_in(std::env::temp_dir())
+    }
+
+    fn new_in(parent: impl AsRef<std::path::Path>) -> Self {
+        let root = tempfile::tempdir_in(parent).expect("system update root");
         let fixture = Self { root };
         fixture.write_executable("old-binary");
         fs::create_dir_all(fixture.web()).expect("web dir");

backend/crates/gateway-plugin/runtime/tests/support/mod.rs

@@ -1,7 +1,7 @@
 use std::{
-    collections::{BTreeMap, BTreeSet, VecDeque},
+    collections::{BTreeMap, BTreeSet},
     io::Write as _,
-    sync::{Arc, Mutex, OnceLock},
+    sync::{Arc, OnceLock},
 };
 
 use flate2::{Compression, write::GzEncoder};
@@ -167,26 +167,33 @@ fn package_with_identity_and_state(
         }),
     };
     let manifest = serde_json::to_vec(&manifest).unwrap();
-    let key: [u8; 32] = Sha256::digest(&manifest).into();
-    type PackageCache = VecDeque<([u8; 32], Arc<[u8]>)>;
-    static PACKAGES: Mutex<PackageCache> = Mutex::new(VecDeque::new());
-    // 缓存的只是不可变制品；各用例仍独立校验、解包、启动进程和创建数据库 schema。
-    // 同时保留至多 8 个制品，避免复杂清单用例累积常驻内存。
-    let mut packages = PACKAGES.lock().unwrap();
-    if let Some(index) = packages.iter().position(|(existing, _)| *existing == key) {
-        let entry = packages.remove(index).unwrap();
-        let package = entry.1.clone();
-        packages.push_back(entry);
-        return package;
+    let key = hex::encode(Sha256::digest(&manifest));
+    // nextest 的用例分属独立进程；清单含 worker 摘要，只共享不可变归档。
+    // 私有解包目录、会话、进程及 Store 仍由每个用例独立创建。
+    let cache = std::path::Path::new(env!("CARGO_TARGET_TMPDIR")).join("plugin-packages-v1");
+    std::fs::create_dir_all(&cache).unwrap();
+    let path = cache.join(format!("{key}.tar.gz"));
+    let lock = std::fs::OpenOptions::new()
+        .create(true)
+        .truncate(false)
+        .read(true)
+        .write(true)
+        .open(cache.join(format!("{key}.lock")))
+        .unwrap();
+    lock.lock().unwrap();
+    match std::fs::read(&path) {
+        Ok(bytes) => return bytes.into(),
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+        Err(error) => panic!("read test plugin archive: {error}"),
     }
     let package = archive(BTreeMap::from([
         ("plugin.json".into(), manifest),
         ("bin/worker".into(), worker.to_vec()),
     ]));
-    if packages.len() == 8 {
-        packages.pop_front();
-    }
-    packages.push_back((key, package.clone()));
+    // 原子发布避免构建中断后其他进程读到不完整的压缩包，文件锁随进程退出释放。
+    let mut temporary = tempfile::NamedTempFile::new_in(cache).unwrap();
+    temporary.write_all(&package).unwrap();
+    temporary.persist(path).unwrap();
     package
 }
 

backend/crates/gateway-protocol/Cargo.toml

@@ -12,6 +12,7 @@ name = "main"
 path = "tests/main.rs"
 
 [dependencies]
+memchr = "2.8.3"
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = { version = "1.0.150", features = ["arbitrary_precision", "preserve_order"] }
 thiserror = "2.0.18"

backend/crates/gateway-protocol/src/openai/sse.rs

@@ -527,14 +527,8 @@ fn track_event_buffer_bytes(current_bytes: &mut usize, line: &str) -> Result<(),
 }
 
 fn sse_frame_separator_bytes(bytes: &[u8]) -> Option<(usize, usize)> {
-    let lf = bytes
-        .windows(2)
-        .position(|window| window == b"\n\n")
-        .map(|position| (position, 2));
-    let crlf = bytes
-        .windows(4)
-        .position(|window| window == b"\r\n\r\n")
-        .map(|position| (position, 4));
+    let lf = memchr::memmem::find(bytes, b"\n\n").map(|position| (position, 2));
+    let crlf = memchr::memmem::find(bytes, b"\r\n\r\n").map(|position| (position, 4));
     match (lf, crlf) {
         (Some(left), Some(right)) => Some(if left.0 <= right.0 { left } else { right }),
         (Some(found), None) | (None, Some(found)) => Some(found),

backend/crates/gateway-protocol/tests/openai/sse.rs

@@ -266,16 +266,17 @@ fn incremental_decoder_should_accept_exact_limit_and_reject_one_byte_over() {
 fn raw_frame_decoder_should_stream_oversized_unfinished_frame_and_resume_observation() {
     let oversized = vec![b'x'; MAX_SSE_EVENT_BUFFER_BYTES + 1];
     let mut decoder = SseEventDecoder::default();
-    let mut raw = decoder
-        .push_frames(&oversized)
-        .into_iter()
-        .flat_map(|frame| frame.into_parts().0)
-        .collect::<Vec<_>>();
+    let mut raw = Vec::with_capacity(oversized.len() + 2);
+    for frame in decoder.push_frames(&oversized) {
+        raw.extend_from_slice(frame.raw());
+    }
 
     assert!(!raw.is_empty());
     let tail = decoder.push_frames(b"\n\n");
     assert!(tail.iter().all(|frame| frame.events().is_empty()));
-    raw.extend(tail.into_iter().flat_map(|frame| frame.into_parts().0));
+    for frame in tail {
+        raw.extend_from_slice(frame.raw());
+    }
     assert_eq!(raw.len(), oversized.len() + 2);
     assert_eq!(&raw[..oversized.len()], oversized.as_slice());
     assert_eq!(&raw[oversized.len()..], b"\n\n");
@@ -474,7 +475,9 @@ fn frame_end_should_support_lf_and_crlf_boundaries() {
             sse_frame_end(b"data: one\n\nremaining"),
             sse_frame_end(b"data: two\r\n\r\nremaining"),
             sse_frame_end(b"data: incomplete"),
+            sse_frame_end(b"data: one\n\ndata: two\r\n\r\n"),
+            sse_frame_end(b"data: one\r\n\r\ndata: two\n\n"),
         ],
-        [Some(11), Some(13), None]
+        [Some(11), Some(13), None, Some(11), Some(13)]
     );
 }

backend/crates/gateway-store/tests/postgres/mod.rs

@@ -65,29 +65,23 @@ impl TestDatabase {
     pub(super) async fn create_through(label: &str, migration_version: i64) -> Option<Self> {
         let database_url = crate::support::test_env("CPR_TEST_DATABASE_URL")?;
         let schema = format!("cpr_store_{label}_{}", Uuid::new_v4().simple());
+        // 临时 schema 验证事务可见性与回滚，不模拟 PostgreSQL 掉电恢复。
+        // 仅这些测试连接异步刷 WAL，保持服务端配置和生产连接行为不变。
+        let options = PgConnectOptions::from_str(&database_url)
+            .expect("parse test PostgreSQL URL")
+            .options([("synchronous_commit", "off")]);
         let admin = PgPoolOptions::new()
             .max_connections(1)
-            .connect(&database_url)
+            .connect_with(options.clone())
             .await
             .expect("connect test PostgreSQL");
         sqlx::raw_sql(sqlx::AssertSqlSafe(format!("create schema \"{schema}\"")))
             .execute(&admin)
             .await
             .expect("create test schema");
-        let search_path = schema.clone();
         let pool = PgPoolOptions::new()
             .max_connections(2)
-            .after_connect(move |connection, _metadata| {
-                let search_path = search_path.clone();
-                Box::pin(async move {
-                    sqlx::query("select set_config('search_path', $1, false)")
-                        .bind(search_path)
-                        .execute(connection)
-                        .await?;
-                    Ok(())
-                })
-            })
-            .connect(&database_url)
+            .connect_with(options.options([("search_path", schema.as_str())]))
             .await
             .expect("connect isolated test schema");
         TEST_MIGRATOR

backend/crates/providers/openai/src/credential/quota/mod.rs

@@ -918,12 +918,19 @@ impl CodexCredentialQuotaService {
     }
 
     pub async fn synchronize(&self) -> Result<CodexQuotaSyncSummary, CodexCredentialQuotaError> {
+        self.synchronize_at(SystemTime::now()).await
+    }
+
+    /// 按本轮调度时刻选择到期账号；实际 HTTP 观察与落库仍使用发生时刻。
+    pub async fn synchronize_at(
+        &self,
+        now: SystemTime,
+    ) -> Result<CodexQuotaSyncSummary, CodexCredentialQuotaError> {
         let mut accounts = self.repository.list_for_provider().await?;
         accounts.retain(|account| {
             account.authentication_kind() == crate::credential::CODEX_AUTHENTICATION_KIND_OAUTH
         });
         let mut summary = CodexQuotaSyncSummary::default();
-        let now = SystemTime::now();
         let account_ids = accounts
             .iter()
             .map(|account| account.id().clone())

backend/crates/providers/openai/tests/credential/catalog.rs

@@ -1351,14 +1351,40 @@ async fn slow_api_key_catalogs_share_a_deadline_and_preserve_healthy_catalogs()
     let service = service_with_catalog_cache(&store, upstream.uri(), catalog_cache());
     let scope = client_scope(&accounts);
     let slow_id = ProviderAccountId::new("acct_slow_0").expect("account ID");
-    let (client, refresh) = tokio::time::timeout(Duration::from_secs(17), async {
-        tokio::join!(
-            service.client_model_catalog(&scope, "1.0.0"),
-            service.refresh_account_catalog(&slow_id)
-        )
+    let execution = tokio::spawn(async move {
+        tokio::time::timeout(Duration::from_secs(17), async {
+            tokio::join!(
+                service.client_model_catalog(&scope, "1.0.0"),
+                service.refresh_account_catalog(&slow_id)
+            )
+        })
+        .await
+        .expect("catalog deadline must bound both client reads and account refreshes")
+    });
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            if upstream
+                .received_requests()
+                .await
+                .unwrap()
+                .iter()
+                .filter(|request| request.url.path() == "/slow/models")
+                .count()
+                >= 5
+            {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
     })
     .await
-    .expect("catalog deadline must bound both client reads and account refreshes");
+    .expect("healthy API catalog completed and released a concurrency slot");
+    // 四个并发读取中，健康 API 目录完成后才会发出第四个慢请求，另有一次独立刷新。
+    // 只快进慢目录的共享期限，随后恢复实时时钟让 OAuth 冷请求完整通过真实 HTTP。
+    tokio::time::pause();
+    tokio::time::advance(Duration::from_secs(15)).await;
+    tokio::time::resume();
+    let (client, refresh) = execution.await.unwrap();
     let models = client.expect("healthy catalogs remain available");
     assert_eq!(
         models

backend/crates/providers/openai/tests/credential/quota/refresh_timing.rs

@@ -13,22 +13,24 @@ async fn mount_usage(server: &MockServer, value: serde_json::Value) {
         .await;
 }
 
-async fn wait_for_reset_grace(reset: i64) {
-    let due = SystemTime::UNIX_EPOCH + Duration::from_secs((reset + 120) as u64);
-    if let Ok(remaining) = due.duration_since(SystemTime::now()) {
-        tokio::time::sleep(remaining + Duration::from_millis(50)).await;
-    }
+fn reset_grace(reset: i64) -> SystemTime {
+    SystemTime::UNIX_EPOCH + Duration::from_secs((reset + 120) as u64)
 }
 
 #[tokio::test]
 async fn reset_grace_bypasses_periodic_throttle_once_then_allows_the_next_window() {
+    let mut now = SystemTime::now();
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_quota_timing").await;
     let account = store.account("acct_quota_timing").expect("account");
     let server = MockServer::start().await;
     let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
-    // 使用接近宽限期终点的真实窗口，在数秒内走完两个窗口的调度与落库路径。
-    let short_reset = Utc::now().timestamp() - 115;
+    // 显式推进调度时刻，真实 HTTP 与落库路径仍完整执行。
+    let short_reset = now
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .unwrap()
+        .as_secs() as i64
+        - 115;
     let week_reset = short_reset + 3;
     let usage = |short_used, short_reset, week_used, week_reset| {
         json!({"rate_limit": {
@@ -46,22 +48,31 @@ async fn reset_grace_bypasses_periodic_throttle_once_then_allows_the_next_window
         .refresh_account(account.id())
         .await
         .expect("seed exhaustion");
-    service.synchronize().await.expect("initial periodic check");
+    service
+        .synchronize_at(now)
+        .await
+        .expect("initial periodic check");
     let requests = server.received_requests().await.expect("requests").len();
 
     assert!(
-        Utc::now().timestamp() < short_reset + 120,
+        now < reset_grace(short_reset),
         "fixture must precede grace deadline"
     );
-    service.synchronize().await.expect("before grace deadline");
+    service
+        .synchronize_at(now)
+        .await
+        .expect("before grace deadline");
     assert_eq!(
         server.received_requests().await.expect("requests").len(),
         requests
     );
 
     mount_usage(&server, usage(0, short_reset + 18_000, 100, week_reset)).await;
-    wait_for_reset_grace(short_reset).await;
-    let partial = service.synchronize().await.expect("short reset check");
+    now = reset_grace(short_reset);
+    let partial = service
+        .synchronize_at(now)
+        .await
+        .expect("short reset check");
     assert_eq!(partial.exhausted, 1);
     assert_eq!(
         store
@@ -72,23 +83,23 @@ async fn reset_grace_bypasses_periodic_throttle_once_then_allows_the_next_window
         Some(SystemTime::UNIX_EPOCH + Duration::from_secs(week_reset as u64))
     );
     service
-        .synchronize()
+        .synchronize_at(now)
         .await
         .expect("next scan before weekly grace deadline");
     assert_eq!(server.received_requests().await.expect("requests").len(), 1);
 
     // 周窗口的到期复核仍未恢复时，也不能每轮扫描重复请求。
-    wait_for_reset_grace(week_reset).await;
+    now = reset_grace(week_reset);
     assert_eq!(
         service
-            .synchronize()
+            .synchronize_at(now)
             .await
             .expect("weekly reset check")
             .exhausted,
         1
     );
     service
-        .synchronize()
+        .synchronize_at(now)
         .await
         .expect("repeat scan after weekly check");
     assert_eq!(server.received_requests().await.expect("requests").len(), 2);
@@ -131,6 +142,7 @@ async fn periodic_checks_continue_when_reset_is_unknown_or_far_in_the_future() {
 
 #[tokio::test]
 async fn allowed_account_with_expired_window_synchronizes_at_reset_grace() {
+    let mut now = SystemTime::now();
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_allowed_expired_window").await;
     let account = store
@@ -139,7 +151,11 @@ async fn allowed_account_with_expired_window_synchronizes_at_reset_grace() {
     let server = MockServer::start().await;
     let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
 
-    let short_reset = Utc::now().timestamp() - 114;
+    let short_reset = now
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .unwrap()
+        .as_secs() as i64
+        - 114;
     let usage = |used: u32, reset: i64| {
         json!({"rate_limit": {
             "allowed": true,
@@ -158,13 +174,16 @@ async fn allowed_account_with_expired_window_synchronizes_at_reset_grace() {
         .expect("seed quota with 100% usage");
 
     assert!(
-        Utc::now().timestamp() < short_reset + 120,
+        now < reset_grace(short_reset),
         "fixture must precede grace deadline"
     );
     let requests = server.received_requests().await.expect("requests").len();
-    service.synchronize().await.expect("initial periodic check");
     service
-        .synchronize()
+        .synchronize_at(now)
+        .await
+        .expect("initial periodic check");
+    service
+        .synchronize_at(now)
         .await
         .expect("repeat check before grace");
     assert_eq!(
@@ -175,10 +194,10 @@ async fn allowed_account_with_expired_window_synchronizes_at_reset_grace() {
 
     // 当到达 reset + 120s 宽限期后，账号虽然处于 allowed 状态，但包含已到期的非零用量窗口，被调度主动同步。
     mount_usage(&server, usage(0, short_reset + 18_000)).await;
-    wait_for_reset_grace(short_reset).await;
+    now = reset_grace(short_reset);
 
     let summary = service
-        .synchronize()
+        .synchronize_at(now)
         .await
         .expect("synchronize expired window");
     assert_eq!(summary.updated, 1);
@@ -190,7 +209,10 @@ async fn allowed_account_with_expired_window_synchronizes_at_reset_grace() {
         .expect("read quota")
         .expect("snapshot");
     assert_eq!(snapshot.windows()[0].used_percent(), Some(0.0));
-    service.synchronize().await.expect("skip fresh window");
+    service
+        .synchronize_at(now)
+        .await
+        .expect("skip fresh window");
     assert_eq!(server.received_requests().await.expect("requests").len(), 1);
 }
 

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

@@ -11752,10 +11752,15 @@ async fn lost_http_response_after_payload_does_not_request_connection_replay() {
 async fn stalled_proxy_connections_share_thirty_seconds_instead_of_resetting_timeout() {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
     let proxy = format!("socks5h://{}", listener.local_addr().unwrap());
+    let (accepted, connected) = tokio::sync::oneshot::channel();
     let server = tokio::spawn(async move {
+        let mut accepted = Some(accepted);
         let mut sockets = Vec::new();
         loop {
             sockets.push(listener.accept().await.unwrap().0);
+            if let Some(accepted) = accepted.take() {
+                let _ = accepted.send(());
+            }
         }
     });
     let store = Arc::new(MemoryAccountStore::default());
@@ -11766,32 +11771,62 @@ async fn stalled_proxy_connections_share_thirty_seconds_instead_of_resetting_tim
         None,
     );
     let provider = provider_with_base_url(&store, "http://upstream.invalid".to_owned());
-    let context = fallback_transport_context("req_socks_timeout");
-    let started = std::time::Instant::now();
-    timeout(Duration::from_secs(32), async {
-        for _ in 0..4 {
-            let mut stream = provider
-                .clone()
-                .execute(
-                    planned_request("openai", generate_operation()),
-                    context.clone(),
-                )
-                .await
-                .unwrap();
-            let mut failure = None;
-            while let Some(event) = stream.next().await {
-                if let Err(error) = event {
-                    failure = Some(error);
-                    break;
+    let context = AttemptContext::new(
+        RequestAttemptContext::new(
+            ModelRequestId::new("req_socks_timeout").unwrap(),
+            ClientApiKeyId::new("key_openai_contract").unwrap(),
+        )
+        .with_request_location(Some(global_request_location()))
+        .with_connection_budget(
+            gateway_core::engine::connection::ConnectionBudget::with_clock(|| {
+                tokio::time::Instant::now().into_std()
+            }),
+        ),
+        NonZeroU32::MIN,
+        SystemTime::now() + Duration::from_secs(30),
+        account_policy(),
+        AccountAttemptContext::new(BTreeSet::new(), None, None)
+            .with_account_scope(contract_account_scope()),
+        None,
+        CancellationToken::new(),
+    )
+    .with_transport(AttemptTransport::Fallback);
+    let started = tokio::time::Instant::now();
+    let execution_context = context.clone();
+    let execution = tokio::spawn(async move {
+        timeout(Duration::from_secs(32), async {
+            for _ in 0..4 {
+                let mut stream = provider
+                    .clone()
+                    .execute(
+                        planned_request("openai", generate_operation()),
+                        execution_context.clone(),
+                    )
+                    .await
+                    .unwrap();
+                let mut failure = None;
+                while let Some(event) = stream.next().await {
+                    if let Err(error) = event {
+                        failure = Some(error);
+                        break;
+                    }
                 }
+                let failure = failure.unwrap();
+                assert_eq!(failure.kind(), ProviderErrorKind::Timeout);
+                assert_eq!(failure.send_state(), UpstreamSendState::NotSent);
             }
-            let failure = failure.unwrap();
-            assert_eq!(failure.kind(), ProviderErrorKind::Timeout);
-            assert_eq!(failure.send_state(), UpstreamSendState::NotSent);
-        }
-    })
-    .await
-    .expect("shared recovery window must not become four 15-second timeouts");
+        })
+        .await
+        .expect("shared recovery window must not become four 15-second timeouts");
+    });
+    timeout(Duration::from_secs(5), connected)
+        .await
+        .expect("proxy accepted the real socket")
+        .unwrap();
+    // 先确认真实连接已建立，再推进共享预算与 transport 的同一单调时钟。
+    tokio::time::pause();
+    tokio::time::advance(Duration::from_secs(29)).await;
+    execution.await.unwrap();
     assert!(started.elapsed() >= Duration::from_secs(25));
     assert!(context.connection_budget().exhausted());
     server.abort();

backend/migrations/README.md

@@ -48,6 +48,9 @@ export CPR_TEST_REDIS_URL='redis://:<password>@127.0.0.1:6379'
 凭据从自己的部署配置或 CI Secret 中读取，不要粘贴未脱敏的 `docker inspect` 输出。
 环境变量未设置导致的跳过不算数据库测试通过
 
+PostgreSQL 的临时 schema 测试连接使用异步提交，保留事务可见性、回滚和完整迁移检查
+需要验证数据库崩溃后的持久性时，使用独立数据库与生产连接配置，不复用这类 fixture
+
 运行包含 `StoreBundle` 初始化的完整集成测试时，两条测试 URL 都须包含密码，且密码满足 Store
 启动配置的 48 位十六进制要求；仅能连接数据库并不代表该初始化合同通过。专用服务使用对应测试密码，
 并在测试进程中清除 `CPR_DATABASE_URL`、`CPR_REDIS_URL`、`CPR_DATABASE_PASSWORD` 和

docs/architecture.md

@@ -814,6 +814,9 @@ RUST_MIN_STACK=16777216 cargo +1.97.0 test --manifest-path backend/Cargo.toml --
 若测试环境使用 Fake-IP 或私网 DNS，需通过 `CPR_PLUGIN_TEST_LIVE_NETWORK_RANGES` 显式提供逗号分隔的
 CIDR 授权。该选项只用于真实网络测试，默认为空，不改变生产网络策略或其他测试的授权
 
+测试归档缓存在 Cargo 测试临时目录的 `plugin-packages-v1/`，按含 worker 摘要的清单跨进程复用；
+每项测试独立校验、解包并创建会话、子进程与 Store，缓存不承载可变运行状态
+
 独立插件包的构建、安装与功能验证说明位于 `codex-proxy-plugins` 仓库的 `examples/workbench/README.md`。
 功能测试与性能、隔离和平台实测分别记录，不相互替代