From e0a022518cd2ff7437d2eac535ee0a132691fc44 Mon Sep 17 00:00:00 2001 From: Zyy Date: Thu, 15 Oct 2026 13:38:25 +0800 Subject: [PATCH] refactor(backend): simplify shared rules and redundant tests Reuse Core error classification and token presentation, remove unused operation APIs and redundant tests, and avoid unnecessary context cloning. Fix expired reset timestamp underflow and SSE metadata frame boundaries. Consolidate architecture checks and document the backend audit results. Validation: backend tests, architecture tests, rustfmt and strict Clippy. PostgreSQL/Redis integration was not verified without dedicated services. Co-authored-by: Codex --- backend/apps/gateway/tests/architecture.rs | 121 ++++++++--------- backend/apps/gateway/tests/main.rs | 9 +- .../src/admin/observability/presenter.rs | 60 ++++----- .../crates/gateway-api/src/openai/error.rs | 35 +---- .../crates/gateway-api/tests/architecture.rs | 7 - .../crates/gateway-api/tests/openai/error.rs | 2 +- .../gateway-core/src/engine/provider.rs | 8 +- backend/crates/gateway-core/src/operation.rs | 58 -------- .../tests/engine/middleware/mod.rs | 78 +++++++++++ .../crates/gateway-core/tests/engine/mod.rs | 1 - .../crates/gateway-core/tests/engine/probe.rs | 9 -- backend/crates/gateway-core/tests/health.rs | 25 +--- .../crates/gateway-core/tests/lifecycle.rs | 8 -- .../crates/gateway-core/tests/operation.rs | 124 ------------------ .../crates/gateway-core/tests/runtime/mod.rs | 19 --- backend/crates/gateway-core/tests/task.rs | 19 +-- .../src/plugin_distribution/mod.rs | 10 +- .../gateway-protocol/src/openai/events.rs | 2 +- .../gateway-protocol/tests/openai/events.rs | 13 ++ .../providers/openai/src/request_identity.rs | 8 +- .../openai/tests/provider/contract/mod.rs | 57 +------- 21 files changed, 208 insertions(+), 465 deletions(-) delete mode 100644 backend/crates/gateway-core/tests/engine/probe.rs diff --git a/backend/apps/gateway/tests/architecture.rs b/backend/apps/gateway/tests/architecture.rs index cfafa8811..ad690bdd3 100644 --- a/backend/apps/gateway/tests/architecture.rs +++ b/backend/apps/gateway/tests/architecture.rs @@ -4,6 +4,7 @@ use std::{ collections::{BTreeMap, BTreeSet}, fs, path::{Path, PathBuf}, + sync::OnceLock, }; use syn::{Item, visit::Visit}; @@ -26,22 +27,24 @@ pub(super) const WORKSPACE_MEMBERS: &[&str] = &[ #[test] fn workspace_member_list_matches_the_frozen_dag_scope() { - let manifest = - fs::read_to_string(backend_root().join("Cargo.toml")).expect("read workspace manifest"); - for member in WORKSPACE_MEMBERS { - assert!( - manifest.contains(&format!("\"{member}\"")), - "workspace must include {member}" - ); - } - let member_count = manifest - .lines() - .map(str::trim) - .filter(|line| line.starts_with("\"apps/") || line.starts_with("\"crates/")) - .count(); + let metadata = cargo_metadata_json(); + let root = backend_root(); + let actual: BTreeSet<_> = metadata["packages"] + .as_array() + .expect("Cargo packages") + .iter() + .map(|package| { + Path::new(package["manifest_path"].as_str().expect("package manifest")) + .parent() + .expect("package directory") + .strip_prefix(&root) + .expect("workspace member path") + .to_path_buf() + }) + .collect(); assert_eq!( - member_count, - WORKSPACE_MEMBERS.len(), + actual, + WORKSPACE_MEMBERS.iter().map(PathBuf::from).collect(), "new workspace members must extend the dependency DAG rules" ); } @@ -378,7 +381,7 @@ const ALLOWED_INTERNAL_EDGES: &[(&str, &str)] = &[ #[test] fn workspace_internal_dependency_edges_match_frozen_dag() { let metadata = cargo_metadata_json(); - let actual = workspace_runtime_dependency_edges(&metadata); + let actual = workspace_runtime_dependency_edges(metadata); let mut expected: Vec<(String, String)> = ALLOWED_INTERNAL_EDGES .iter() .map(|(from, to)| ((*from).to_owned(), (*to).to_owned())) @@ -474,24 +477,6 @@ fn adapter_public_module_surfaces_match_allowlist() { } } -#[test] -fn production_sources_do_not_host_tests() { - for member in WORKSPACE_MEMBERS { - let src = backend_root().join(member).join("src"); - let sources = super::rust_files(&src); - assert!(!sources.is_empty(), "{member} has no production sources"); - for relative in sources { - let path = src.join(relative); - let source = fs::read_to_string(&path).expect("read production source"); - assert!( - !source.contains("#[cfg(test)]"), - "{} hosts tests in production src", - path.display() - ); - } - } -} - #[test] fn workspace_modules_follow_conventional_file_layout() { let targets = test_source_roots(); @@ -748,40 +733,46 @@ pub(super) fn backend_root() -> PathBuf { .expect("backend workspace root") } -fn cargo_metadata_json() -> serde_json::Value { - let output = std::process::Command::new("cargo") - .args(["metadata", "--format-version", "1", "--no-deps"]) - .current_dir(backend_root()) - .output() - .expect("run cargo metadata"); - assert!( - output.status.success(), - "cargo metadata failed: {}", - String::from_utf8_lossy(&output.stderr) - ); - serde_json::from_slice(&output.stdout).expect("parse cargo metadata") +fn cargo_metadata_json() -> &'static serde_json::Value { + static METADATA: OnceLock = OnceLock::new(); + METADATA.get_or_init(|| { + let output = std::process::Command::new("cargo") + .args(["metadata", "--format-version", "1", "--no-deps", "--locked"]) + .current_dir(backend_root()) + .output() + .expect("run cargo metadata"); + assert!( + output.status.success(), + "cargo metadata failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + serde_json::from_slice(&output.stdout).expect("parse cargo metadata") + }) } -/// 提取成员 `[dependencies]` 段内声明的依赖名;段落以下一个 `[` 表头结束 +/// 复用 Cargo 解析后的实际包名，覆盖重命名依赖与平台条件，排除测试和构建依赖 fn dependency_names(member: &str) -> Vec { - let manifest = fs::read_to_string(backend_root().join(member).join("Cargo.toml")) - .expect("read member manifest"); - let mut in_dependencies = false; - let mut names = Vec::new(); - for line in manifest.lines() { - let line = line.trim(); - if line.starts_with('[') { - in_dependencies = line == "[dependencies]"; - continue; - } - if !in_dependencies || line.is_empty() || line.starts_with('#') { - continue; - } - if let Some((name, _)) = line.split_once('=') { - names.push(name.trim().to_owned()); - } - } - names + let manifest = backend_root().join(member).join("Cargo.toml"); + let package = cargo_metadata_json()["packages"] + .as_array() + .expect("Cargo packages") + .iter() + .find(|package| { + Path::new(package["manifest_path"].as_str().expect("package manifest")) == manifest + }) + .expect("workspace member"); + package["dependencies"] + .as_array() + .expect("package dependencies") + .iter() + .filter(|dependency| dependency["kind"].is_null()) + .map(|dependency| { + dependency["name"] + .as_str() + .expect("dependency name") + .to_owned() + }) + .collect() } fn assert_no_dependency(member: &str, forbidden: &[&str]) { diff --git a/backend/apps/gateway/tests/main.rs b/backend/apps/gateway/tests/main.rs index 764277aa9..d21a64ff2 100644 --- a/backend/apps/gateway/tests/main.rs +++ b/backend/apps/gateway/tests/main.rs @@ -59,9 +59,16 @@ fn cargo_library_root_is_conventional_lib() { fn workspace_production_files_have_no_hidden_modules_or_test_hooks() { for member in architecture::WORKSPACE_MEMBERS { let root = architecture::backend_root().join(member).join("src"); - for relative in rust_files(&root) { + let sources = rust_files(&root); + assert!(!sources.is_empty(), "{member} has no production sources"); + for relative in sources { let path = root.join(relative); let source = fs::read_to_string(&path).expect("read production source"); + assert!( + !source.contains("#[cfg(test)]"), + "{} hosts tests in production src", + path.display() + ); assert!( !source.contains("include!("), "{} uses include!", diff --git a/backend/crates/gateway-api/src/admin/observability/presenter.rs b/backend/crates/gateway-api/src/admin/observability/presenter.rs index f16eb89d9..8064b5603 100644 --- a/backend/crates/gateway-api/src/admin/observability/presenter.rs +++ b/backend/crates/gateway-api/src/admin/observability/presenter.rs @@ -1,6 +1,7 @@ //! 观测领域事实到管理页面的展示投影 use super::*; +use gateway_core::metering::Usage; pub(crate) fn display_duration(value: Option) -> String { let Some(value) = value.and_then(|value| i64::try_from(value).ok()) else { @@ -100,7 +101,7 @@ pub(crate) fn attempt_metrics_view(metrics: &domain::AttemptMetrics) -> AttemptM } pub(crate) fn token_details(record: &domain::UsageRecord) -> TokenDetailsView { - TokenDetailsView { + token_details_view(&Usage { input_tokens: record.input_tokens, output_tokens: record.output_tokens, cached_input_tokens: record.cached_input_tokens, @@ -109,35 +110,11 @@ pub(crate) fn token_details(record: &domain::UsageRecord) -> TokenDetailsView { image_input_tokens: record.image_input_tokens, image_output_tokens: record.image_output_tokens, total_tokens: record.total_tokens, - input_tokens_display: record - .input_tokens - .map_or_else(|| "-".to_owned(), format_number), - output_tokens_display: record - .output_tokens - .map_or_else(|| "-".to_owned(), format_number), - cached_input_tokens_display: record - .cached_input_tokens - .map_or_else(|| "-".to_owned(), format_compact_number), - cache_write_input_tokens_display: record - .cache_write_input_tokens - .map_or_else(|| "-".to_owned(), format_compact_number), - reasoning_output_tokens_display: record - .reasoning_output_tokens - .map_or_else(|| "-".to_owned(), format_number), - image_input_tokens_display: record - .image_input_tokens - .map_or_else(|| "-".to_owned(), format_number), - image_output_tokens_display: record - .image_output_tokens - .map_or_else(|| "-".to_owned(), format_number), - total_tokens_display: record - .total_tokens - .map_or_else(|| "-".to_owned(), format_number), - } + }) } pub(crate) fn usage_list_token_details(record: &domain::UsageListRecord) -> TokenDetailsView { - TokenDetailsView { + token_details_view(&Usage { input_tokens: record.input_tokens, output_tokens: record.output_tokens, cached_input_tokens: record.cached_input_tokens, @@ -146,28 +123,41 @@ pub(crate) fn usage_list_token_details(record: &domain::UsageListRecord) -> Toke image_input_tokens: record.image_input_tokens, image_output_tokens: record.image_output_tokens, total_tokens: record.total_tokens, - input_tokens_display: record + }) +} + +fn token_details_view(usage: &Usage) -> TokenDetailsView { + TokenDetailsView { + input_tokens: usage.input_tokens, + output_tokens: usage.output_tokens, + cached_input_tokens: usage.cached_input_tokens, + cache_write_input_tokens: usage.cache_write_input_tokens, + reasoning_output_tokens: usage.reasoning_output_tokens, + image_input_tokens: usage.image_input_tokens, + image_output_tokens: usage.image_output_tokens, + total_tokens: usage.total_tokens, + input_tokens_display: usage .input_tokens .map_or_else(|| "-".to_owned(), format_number), - output_tokens_display: record + output_tokens_display: usage .output_tokens .map_or_else(|| "-".to_owned(), format_number), - cached_input_tokens_display: record + cached_input_tokens_display: usage .cached_input_tokens .map_or_else(|| "-".to_owned(), format_compact_number), - cache_write_input_tokens_display: record + cache_write_input_tokens_display: usage .cache_write_input_tokens .map_or_else(|| "-".to_owned(), format_compact_number), - reasoning_output_tokens_display: record + reasoning_output_tokens_display: usage .reasoning_output_tokens .map_or_else(|| "-".to_owned(), format_number), - image_input_tokens_display: record + image_input_tokens_display: usage .image_input_tokens .map_or_else(|| "-".to_owned(), format_number), - image_output_tokens_display: record + image_output_tokens_display: usage .image_output_tokens .map_or_else(|| "-".to_owned(), format_number), - total_tokens_display: record + total_tokens_display: usage .total_tokens .map_or_else(|| "-".to_owned(), format_number), } diff --git a/backend/crates/gateway-api/src/openai/error.rs b/backend/crates/gateway-api/src/openai/error.rs index d6bf964bd..ff28b3443 100644 --- a/backend/crates/gateway-api/src/openai/error.rs +++ b/backend/crates/gateway-api/src/openai/error.rs @@ -13,6 +13,8 @@ use gateway_core::{ }; use serde_json::{Value, json}; +pub use gateway_core::engine::execution::gateway_error_from_engine; + use super::responses::ProtocolErrorBody; /// OpenAI 风格错误响应 @@ -148,39 +150,6 @@ pub fn model_not_found_response() -> (StatusCode, Json) { ) } -/// 将 core Engine 错误收敛为稳定 Gateway 分类 -#[must_use] -pub fn gateway_error_from_engine(error: &EngineError) -> GatewayError { - match error { - EngineError::Provider(error) => GatewayError::from_provider(error), - EngineError::Cancelled => { - GatewayError::new(GatewayErrorKind::Cancelled, "request was cancelled") - } - EngineError::Deadline => { - GatewayError::new(GatewayErrorKind::Timeout, "gateway request timed out") - } - EngineError::ProviderNotRegistered { .. } | EngineError::EmptyRoutingPlan => { - GatewayError::new( - GatewayErrorKind::NoAvailableProvider, - "no upstream provider is currently available for this request", - ) - } - EngineError::Store(source) => { - GatewayError::new(GatewayErrorKind::Internal, "gateway execution failed") - .with_source(source.clone()) - } - EngineError::ProviderMetadataMismatch - | EngineError::ContinuationPinMismatch - | EngineError::RequiredAccountMismatch - | EngineError::AccountOutsideClientScope - | EngineError::DownstreamCommitRequired - | EngineError::NoActiveAttempt - | EngineError::InvalidDeliveryState => { - GatewayError::new(GatewayErrorKind::Internal, "gateway execution failed") - } - } -} - /// Gateway 错误的 OpenAI HTTP 表达 pub fn gateway_error_response(error: &GatewayError) -> Response { let (status, default_type, default_code) = gateway_error_contract(error.kind()); diff --git a/backend/crates/gateway-api/tests/architecture.rs b/backend/crates/gateway-api/tests/architecture.rs index 32a05ed04..2f13d5170 100644 --- a/backend/crates/gateway-api/tests/architecture.rs +++ b/backend/crates/gateway-api/tests/architecture.rs @@ -22,13 +22,6 @@ fn manifest_should_depend_only_on_core_protocol_and_http_adapter_layers() { ); } -#[test] -fn workspace_should_include_gateway_api() { - let workspace = include_str!("../../../Cargo.toml"); - - assert!(workspace.contains("\"crates/gateway-api\"")); -} - #[test] fn admin_routes_should_use_only_get_post_and_static_paths() { let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("src/admin"); diff --git a/backend/crates/gateway-api/tests/openai/error.rs b/backend/crates/gateway-api/tests/openai/error.rs index b376d2fb0..d6288b00c 100644 --- a/backend/crates/gateway-api/tests/openai/error.rs +++ b/backend/crates/gateway-api/tests/openai/error.rs @@ -548,7 +548,7 @@ fn engine_store_error_should_collapse_to_safe_internal_error() { let projected = gateway_error_from_engine(&error); assert_eq!(projected.kind(), GatewayErrorKind::Internal); - assert_eq!(projected.safe_message(), "gateway execution failed"); + assert_eq!(projected.safe_message(), "request execution failed"); } #[test] fn openai_error_response_should_preserve_only_safe_contract_fields() { diff --git a/backend/crates/gateway-core/src/engine/provider.rs b/backend/crates/gateway-core/src/engine/provider.rs index 4b0a0e524..2b59ed58d 100644 --- a/backend/crates/gateway-core/src/engine/provider.rs +++ b/backend/crates/gateway-core/src/engine/provider.rs @@ -638,7 +638,8 @@ fn provider_event_to_middleware_frame( fn encode_sse_wire_event(wire: &ProtocolWireEvent) -> Result { let mut encoded = Vec::new(); - if let Some(id) = wire.sse_id() { + // 源事件元数据保持不透明；仅编码可表示为单行 SSE 字段的值，避免生成额外帧 + if let Some(id) = wire.sse_id().filter(|id| !id.contains(['\r', '\n', '\0'])) { encoded.extend_from_slice(b"id: "); encoded.extend_from_slice(id.as_bytes()); encoded.push(b'\n'); @@ -646,7 +647,10 @@ fn encode_sse_wire_event(wire: &ProtocolWireEvent) -> Result, - body: Map, - ) -> Result { - let encoded = ProtocolPayload::json_object(protocol, body)? - .with_context(self.protocol_payload().context().clone()); - let requirements = self.requirements(); - let payload = Arc::make_mut(&mut self.payload); - payload.protocol_payload = encoded; - payload.source_requirements = Some(requirements); - Ok(self) - } - /// 附着同一客户端连接上一轮由 Provider 返回的不透明状态 #[must_use] pub fn with_provider_session_state(mut self, state: ProviderSessionState) -> Self { @@ -989,41 +966,6 @@ impl Operation { self.replace_protocol_wire_inner(None, body, context) } - /// 替换 Provider HTTP 正文、非 wire 上下文和已解析 headers，同时保持 endpoint、 - /// method 与 query 不变 - /// headers 继续由 [`ProviderHttpRequest::new`] 统一验证； - /// Core 不解析插件传输使用的 base64 表示 - /// - /// # Errors - /// - /// 当前 operation 不是 Provider HTTP，或替换后的 headers 超出请求边界时返回错误 - pub fn replace_provider_http_wire( - self, - body: Option, - context: Map, - headers: Vec, - ) -> Result { - let Self::ProviderHttp(request) = self else { - return Err(OperationError::EmptyField { - field: "provider_http_request", - }); - }; - let payload = request.payload; - ProviderHttpRequest::new( - request.endpoint, - request.method, - request.query, - headers, - RawHttpPayload { - protocol: payload.protocol, - body: body.unwrap_or(payload.body), - context, - translated: payload.translated, - }, - ) - .map(Self::ProviderHttp) - } - /// 把一次已选择的协议转换结果写回 operation，同时保留非 wire 上下文和 /// Provider 会话状态 /// 该方法只验证公共载荷形状，不解释目标协议字段 diff --git a/backend/crates/gateway-core/tests/engine/middleware/mod.rs b/backend/crates/gateway-core/tests/engine/middleware/mod.rs index 8d9ddc034..939457333 100644 --- a/backend/crates/gateway-core/tests/engine/middleware/mod.rs +++ b/backend/crates/gateway-core/tests/engine/middleware/mod.rs @@ -509,6 +509,84 @@ fn attempt_passthrough_preserves_parsed_wire_for_every_client_transport() { }); } +#[test] +fn attempt_sse_encoding_keeps_opaque_metadata_from_creating_extra_frames() { + #[derive(Debug)] + struct InspectingPlan { + expected: &'static str, + } + + impl MiddlewarePlan for InspectingPlan { + fn handle( + &self, + _: MiddlewareContext, + request: MiddlewareRequest, + next: MiddlewareNext, + ) -> BoxFuture<'static, Result> { + let expected = self.expected; + Box::pin(async move { + let (protocol, status, headers, mut body, envelope) = + next.run(request).await?.into_parts(); + let frame = body.next_frame().await?.expect("provider event"); + assert_eq!(frame.bytes().as_ref(), expected.as_bytes()); + Ok(MiddlewareResponse::new(protocol, status, headers, body) + .with_envelope(envelope.expect("provider metadata"))) + }) + } + } + + futures::executor::block_on(async { + for (event_type, id, expected) in [ + ( + "future.event", + "event-1", + "id: event-1\nretry: 10\nevent: future.event\ndata: {\"value\":1}\n\n", + ), + ( + "future.event\ndata: injected\n\n", + "event-1\r\ndata: injected\r\n\r\n", + "retry: 10\ndata: {\"value\":1}\n\n", + ), + ("", "event\0id", "retry: 10\ndata: {\"value\":1}\n\n"), + ] { + let plan = FrozenMiddlewarePlan::new( + Arc::new(InspectingPlan { expected }), + ExtensionSetReference::new( + ExtensionSetId::new("sse-metadata".to_owned()).unwrap(), + Arc::new(ReadyGeneration), + ), + ); + let wire = ProtocolWireEvent::json_with_sse_metadata( + "openai", + Some(event_type.to_owned()), + json!({"value":1}), + Some(id.to_owned()), + Some(10), + ) + .unwrap(); + let terminal: ProviderMiddlewareTerminal = Box::new(move |_, _| { + Box::pin(async move { + Ok(ProviderStream::new( + empty_provider_stream().metadata().clone(), + stream::iter([Ok(ProviderEvent::wire(wire))]), + (), + )) + }) + }); + + execute_attempt_middleware( + Some(&plan), + context(), + operation(), + ClientTransport::HttpSse, + terminal, + ) + .await + .unwrap(); + } + }); +} + #[test] fn attempt_middleware_preserves_multi_value_and_authentication_headers() { futures::executor::block_on(async { diff --git a/backend/crates/gateway-core/tests/engine/mod.rs b/backend/crates/gateway-core/tests/engine/mod.rs index 6553e9cf8..9341c49df 100644 --- a/backend/crates/gateway-core/tests/engine/mod.rs +++ b/backend/crates/gateway-core/tests/engine/mod.rs @@ -8,7 +8,6 @@ mod execution; mod extensions; mod middleware; mod policy; -mod probe; mod provider; mod response_control; diff --git a/backend/crates/gateway-core/tests/engine/probe.rs b/backend/crates/gateway-core/tests/engine/probe.rs deleted file mode 100644 index 13d5d9d30..000000000 --- a/backend/crates/gateway-core/tests/engine/probe.rs +++ /dev/null @@ -1,9 +0,0 @@ -//! 验证账号探测可通过对象安全的执行链端口调用 - -use gateway_core::engine::probe::AccountProbe; - -#[test] -fn account_probe_is_an_object_safe_execution_chain_port() { - fn accepts(_: Option<&dyn AccountProbe>) {} - accepts(None); -} diff --git a/backend/crates/gateway-core/tests/health.rs b/backend/crates/gateway-core/tests/health.rs index 7cfce97c1..42666349e 100644 --- a/backend/crates/gateway-core/tests/health.rs +++ b/backend/crates/gateway-core/tests/health.rs @@ -1,24 +1,10 @@ -//! 验证健康探针的对象安全接口与 Worker 身份投影 +//! 验证 Worker 健康状态的身份投影 -use futures::future::BoxFuture; use gateway_core::health::{ - HealthProbe, HealthState, WorkerHealthKey, WorkerHealthSnapshot, WorkerHealthSource, - WorkerRuntimeState, + WorkerHealthKey, WorkerHealthSnapshot, WorkerHealthSource, WorkerRuntimeState, }; use gateway_core::task::{WorkerId, WorkerKind}; -struct HealthyProbe; - -impl HealthProbe for HealthyProbe { - fn name(&self) -> &'static str { - "test" - } - - fn check(&self) -> BoxFuture<'_, HealthState> { - Box::pin(async { HealthState::Healthy }) - } -} - struct SingleSnapshot; impl WorkerHealthSource for SingleSnapshot { @@ -38,13 +24,6 @@ impl WorkerHealthSource for SingleSnapshot { } } -#[test] -fn health_probe_contract_is_object_safe() { - let probe: &dyn HealthProbe = &HealthyProbe; - - assert_eq!(probe.name(), "test"); -} - #[test] fn worker_health_source_preserves_worker_identity() { let source: &dyn WorkerHealthSource = &SingleSnapshot; diff --git a/backend/crates/gateway-core/tests/lifecycle.rs b/backend/crates/gateway-core/tests/lifecycle.rs index cec7191e6..c843cffbc 100644 --- a/backend/crates/gateway-core/tests/lifecycle.rs +++ b/backend/crates/gateway-core/tests/lifecycle.rs @@ -315,14 +315,6 @@ fn connection_registration_rejects_after_drain_linearization() { assert!(matches!(lifecycle.try_register(), Err(ConnectionDraining))); } -#[test] -fn connection_lifecycle_contract_is_object_safe() { - let lifecycle = TestLifecycle::new(); - let object: &dyn ConnectionLifecycle = &lifecycle; - - assert!(!object.is_draining()); -} - #[test] fn cancellation_reason_survives_cleanup_and_propagates_only_to_descendants() { use gateway_core::lifecycle::{CancellationReason, LeaseFailure, RequestLeaseKind}; diff --git a/backend/crates/gateway-core/tests/operation.rs b/backend/crates/gateway-core/tests/operation.rs index ba11b460d..ff1d71f84 100644 --- a/backend/crates/gateway-core/tests/operation.rs +++ b/backend/crates/gateway-core/tests/operation.rs @@ -109,68 +109,6 @@ fn middleware_protocol_change_keeps_source_capability_facts_without_interpreting assert_eq!(replaced.capability_requirements(), source_requirements); } -#[test] -fn native_encoding_keeps_admission_context_and_session_after_optional_translation() { - let context = Map::from_iter([("conversation_id".to_owned(), json!("conversation"))]); - let session = ProviderSessionState::new( - "xai", - Map::from_iter([("session_id".to_owned(), json!("session"))]), - ) - .unwrap(); - let request = GenerateRequest::from_protocol_payload( - ProtocolPayload::json_object( - "openai", - json!({"input":"hello", "max_output_tokens":64, "reasoning":{"effort":"high"}}) - .as_object() - .unwrap() - .clone(), - ) - .unwrap() - .with_context(context.clone()), - ) - .with_provider_session_state(session.clone()); - let original = Operation::Generate(request.clone()); - let translated = original - .clone() - .replace_middleware_wire("native-input", Bytes::from_static(br#"{"prompt":"hello"}"#)) - .unwrap(); - let Operation::Generate(translated) = translated else { - panic!("generate request"); - }; - - for (request, target) in [(request, "openai"), (translated, "xai")] { - let body = Map::from_iter([("native_prompt".to_owned(), json!("encoded"))]); - let encoded = request - .with_native_encoded_body(target, body.clone()) - .unwrap(); - assert_eq!(encoded.protocol_payload().protocol(), target); - assert_eq!(encoded.protocol_payload().body(), &body); - assert_eq!(encoded.protocol_payload().context(), &context); - assert_eq!(encoded.provider_session_state("xai"), Some(&session)); - let operation = Operation::Generate(encoded); - assert_eq!( - operation.capability_requirements(), - original.capability_requirements() - ); - assert!( - operation - .replace_middleware_wire("another", Bytes::from_static(b"{}")) - .is_err() - ); - } -} - -#[test] -fn native_encoding_rejects_invalid_protocol_names() { - for name in ["", "invalid\nprotocol"] { - assert!( - generate(json!({"input":"hello"})) - .with_native_encoded_body(name, Map::new()) - .is_err() - ); - } -} - #[test] fn middleware_processing_recomputes_requirements_but_protocol_change_is_direct_once() { let operation = Operation::Generate(generate(json!({ @@ -336,68 +274,6 @@ fn provider_http_should_reject_urls_controls_and_unbounded_headers() { ); } -#[test] -fn provider_http_wire_replacement_keeps_frozen_target_and_validates_typed_headers() { - let operation = Operation::ProviderHttp( - ProviderHttpRequest::new( - "inspect", - ProviderHttpMethod::Post, - Some("mode=stable".to_owned()), - vec![ProviderHttpHeader::new( - "x-original", - Bytes::from_static(b"old"), - )], - RawHttpPayload::new("provider-http", Bytes::from_static(b"old-body")) - .expect("provider HTTP payload"), - ) - .expect("provider HTTP request"), - ); - assert!( - operation - .clone() - .replace_provider_http_wire( - None, - Map::new(), - vec![ProviderHttpHeader::new( - "x-invalid", - Bytes::from_static(b"line\r\nbreak"), - )], - ) - .is_err() - ); - let replacement = operation - .replace_provider_http_wire( - Some(Bytes::from_static(&[0x00, 0xff, b'n', b'e', b'w'])), - Map::from_iter([("connection_fact".to_owned(), json!("kept"))]), - vec![ - ProviderHttpHeader::new("x-repeat", Bytes::from_static(&[0xff, 0xfe])), - ProviderHttpHeader::new("x-repeat", Bytes::from_static(b"two")), - ], - ) - .expect("typed Provider HTTP replacement"); - let Operation::ProviderHttp(request) = replacement else { - panic!("operation kind changed") - }; - assert_eq!(request.endpoint(), "inspect"); - assert_eq!(request.method(), ProviderHttpMethod::Post); - assert_eq!(request.query(), Some("mode=stable")); - assert_eq!( - request.payload().body().as_ref(), - &[0x00, 0xff, b'n', b'e', b'w'] - ); - assert_eq!(request.payload().context()["connection_fact"], "kept"); - assert_eq!(request.headers().len(), 2); - assert_eq!(request.headers()[0].value().as_ref(), &[0xff, 0xfe]); - assert_eq!(request.headers()[1].value().as_ref(), b"two"); - - let generate = Operation::Generate(generate(json!({"model":"gpt-test"}))); - assert!( - generate - .replace_provider_http_wire(None, Map::new(), Vec::new()) - .is_err() - ); -} - #[test] fn standalone_search_should_keep_raw_body_and_context_opaque() { let body = Bytes::from_static( diff --git a/backend/crates/gateway-core/tests/runtime/mod.rs b/backend/crates/gateway-core/tests/runtime/mod.rs index 67a20551b..6189181be 100644 --- a/backend/crates/gateway-core/tests/runtime/mod.rs +++ b/backend/crates/gateway-core/tests/runtime/mod.rs @@ -324,25 +324,6 @@ fn publisher_should_suspend_but_still_notify_after_committed_refresh_failure() { ); } -#[test] -fn snapshot_ports_and_control_should_remain_object_safe() { - fn accept_store(_: &dyn SnapshotStorePort) {} - fn accept_subscriptions(_: &dyn SnapshotSubscriptionPort) {} - fn accept_control(_: &dyn SnapshotControl) {} - - let store = Arc::new(TestSnapshotStore::new(Ok(facts(1, 1)))); - let subscriptions = Arc::new(TestSnapshotSubscriptions::default()); - let publisher = RuntimeSnapshotPublisher::new( - Arc::new(compiler(store.clone())), - RuntimeSnapshotHandle::new(empty_snapshot(1)), - subscriptions.clone(), - ); - - accept_store(store.as_ref()); - accept_subscriptions(subscriptions.as_ref()); - accept_control(&publisher); -} - #[test] fn publisher_should_contribute_reconciliation_and_subscription_workers() { let store = Arc::new(TestSnapshotStore::new(Ok(facts(1, 1)))); diff --git a/backend/crates/gateway-core/tests/task.rs b/backend/crates/gateway-core/tests/task.rs index 76f673fc8..84517eae5 100644 --- a/backend/crates/gateway-core/tests/task.rs +++ b/backend/crates/gateway-core/tests/task.rs @@ -6,7 +6,7 @@ use std::time::Duration; use futures::future::BoxFuture; use gateway_core::lifecycle::CancellationToken; use gateway_core::task::{ - DaemonRestartPolicy, DaemonTask, ScheduledTask, WorkerCycleContext, WorkerDefinitionError, + DaemonRestartPolicy, ScheduledTask, WorkerCycleContext, WorkerDefinitionError, WorkerFencingToken, WorkerId, WorkerKind, WorkerLeaseRequest, WorkerRegistration, WorkerRunnable, WorkerSchedule, WorkerTaskError, }; @@ -22,14 +22,6 @@ impl ScheduledTask for NoopScheduledTask { } } -struct NoopDaemonTask; - -impl DaemonTask for NoopDaemonTask { - fn run(&self, _cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>> { - Box::pin(async { Ok(()) }) - } -} - #[test] fn schedule_rejects_every_invalid_duration_boundary() { let one = Duration::from_nanos(1); @@ -135,15 +127,6 @@ fn worker_cycle_context_preserves_optional_fencing_token() { assert_eq!(context.fencing_token(), Some(token)); } -#[test] -fn scheduled_and_daemon_tasks_are_object_safe() { - fn accept_scheduled(_task: &dyn ScheduledTask) {} - fn accept_daemon(_task: &dyn DaemonTask) {} - - accept_scheduled(&NoopScheduledTask); - accept_daemon(&NoopDaemonTask); -} - fn worker_id(owner: &str) -> WorkerId { WorkerId::try_new(WorkerKind::OAuthRefresh, owner).expect("valid worker identity") } diff --git a/backend/crates/gateway-host/src/plugin_distribution/mod.rs b/backend/crates/gateway-host/src/plugin_distribution/mod.rs index 82cde084a..fc7932bde 100644 --- a/backend/crates/gateway-host/src/plugin_distribution/mod.rs +++ b/backend/crates/gateway-host/src/plugin_distribution/mod.rs @@ -39,7 +39,11 @@ pub struct HttpPluginDistribution { impl HttpPluginDistribution { pub fn new(http: Arc) -> Result { - Self::with_transport("https://api.github.com/", http, source_network()?) + Self::with_transport( + "https://api.github.com/", + http, + NetworkPolicy::unrestricted(), + ) } /// 测试可注入受限网络；生产来源由管理员显式配置，因此保持既有内外网可达范围 @@ -173,10 +177,6 @@ impl PluginDistribution for HttpPluginDistribution { } } -fn source_network() -> Result { - NetworkPolicy::new(&["0.0.0.0/0".into(), "::/0".into()]).map_err(|_| source_error()) -} - fn credential_ids(credentials: &[SourceCredential]) -> Vec { credentials .iter() diff --git a/backend/crates/gateway-protocol/src/openai/events.rs b/backend/crates/gateway-protocol/src/openai/events.rs index f30672018..18057f62d 100644 --- a/backend/crates/gateway-protocol/src/openai/events.rs +++ b/backend/crates/gateway-protocol/src/openai/events.rs @@ -599,7 +599,7 @@ fn retry_after_seconds_from_resets_at(error: &Value) -> Option { .duration_since(std::time::UNIX_EPOCH) .ok()? .as_secs(); - (resets_at > now).then_some(resets_at - now) + (resets_at > now).then(|| resets_at - now) } fn positive_u64(value: &Value) -> Option { diff --git a/backend/crates/gateway-protocol/tests/openai/events.rs b/backend/crates/gateway-protocol/tests/openai/events.rs index 2d6289b65..31b94a422 100644 --- a/backend/crates/gateway-protocol/tests/openai/events.rs +++ b/backend/crates/gateway-protocol/tests/openai/events.rs @@ -394,6 +394,19 @@ fn retry_after_should_read_string_reset_timestamp() { assert!((59..=60).contains(&retry_after)); } +#[test] +fn retry_after_should_ignore_elapsed_reset_timestamps_and_keep_later_advice() { + for reset_at in [json!(1), json!("1")] { + let mut body = json!({ + "error": {"type": "usage_limit_reached", "resets_at": reset_at} + }); + assert_eq!(retry_after_seconds_from_body(&body.to_string()), None); + + body["retry_after_seconds"] = json!(7); + assert_eq!(retry_after_seconds_from_body(&body.to_string()), Some(7)); + } +} + #[test] fn retry_after_should_read_nested_error_delay() { let body = json!({ diff --git a/backend/crates/providers/openai/src/request_identity.rs b/backend/crates/providers/openai/src/request_identity.rs index 81e456d4e..bc322c8a1 100644 --- a/backend/crates/providers/openai/src/request_identity.rs +++ b/backend/crates/providers/openai/src/request_identity.rs @@ -1,5 +1,7 @@ //! 请求正文与最终头部的会话身份解释，供账号亲和与传输编码共用 +use std::borrow::Cow; + use serde_json::{Map, Value}; /// 只有能识别出的后代线程禁止迁移会话；没有身份的普通客户端沿用现有调度 @@ -64,13 +66,15 @@ pub(crate) fn account_session_with_headers( { return Some(session); } - let mut context = context.clone(); + let mut context = Cow::Borrowed(context); for (name, field) in [ ("session-id", "session_id"), ("x-codex-turn-metadata", "turn_metadata"), ] { if let Some(value) = header(name) { - context.insert(field.to_owned(), Value::String(value.to_owned())); + context + .to_mut() + .insert(field.to_owned(), Value::String(value.to_owned())); } } gateway_protocol::openai::codex_account_session_id(body, &context) diff --git a/backend/crates/providers/openai/tests/provider/contract/mod.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs index a1b876498..81797920b 100644 --- a/backend/crates/providers/openai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/openai/tests/provider/contract/mod.rs @@ -9979,7 +9979,7 @@ async fn successful_http_sse_rate_limit_event_persists_structured_exhaustion() { } #[test] -fn request_observation_reads_openai_metadata_without_changing_the_operation() { +fn request_observation_reads_openai_metadata() { let store = Arc::new(MemoryAccountStore::default()); let payload = ProtocolPayload::json_object( "openai", @@ -10004,11 +10004,10 @@ fn request_observation_reads_openai_metadata_without_changing_the_operation() { assert_eq!(observation.request_kind.as_deref(), Some("review")); assert_eq!(observation.subagent_kind.as_deref(), Some("worker")); assert!(!observation.compact); - assert!(matches!(operation, Operation::Generate(_))); } #[test] -fn request_observation_classifies_prewarm_from_generate_without_rewriting_the_payload() { +fn request_observation_classifies_prewarm_from_generate() { let store = Arc::new(MemoryAccountStore::default()); let provider = provider(&store); let client_key_id = ClientApiKeyId::new("key_prewarm_observation").expect("client key"); @@ -10050,15 +10049,11 @@ fn request_observation_classifies_prewarm_from_generate_without_rewriting_the_pa body.get("generate"), ); assert_eq!(observation.subagent_kind.as_deref(), Some("worker")); - let Operation::Generate(generation) = operation else { - unreachable!(); - }; - assert_eq!(generation.protocol_payload().body(), &body); } } #[test] -fn endpoint_observation_should_read_models_without_rewriting_or_requiring_a_catalog() { +fn endpoint_observation_should_read_models_without_requiring_a_catalog() { let store = Arc::new(MemoryAccountStore::default()); let provider = provider(&store); let client_key = ClientApiKeyId::new("key_endpoint_model").expect("client key"); @@ -10071,8 +10066,7 @@ fn endpoint_observation_should_read_models_without_rewriting_or_requiring_a_cata ] { let body = serde_json::to_vec(&json!({"model":model,"future":9007199254740993_u64})) .expect("body"); - let payload = - RawJsonPayload::new("openai", Bytes::copy_from_slice(&body)).expect("payload"); + let payload = RawJsonPayload::new("openai", Bytes::from(body)).expect("payload"); let operations = [ Operation::GenerateImage(ImageRequest::from_raw_json( ImageRequestKind::Generation, @@ -10093,12 +10087,6 @@ fn endpoint_observation_should_read_models_without_rewriting_or_requiring_a_cata .map(PublicModelId::as_str), model.as_str().filter(|model| !model.is_empty()), ); - let payload = match &operation { - Operation::GenerateImage(request) => request.payload(), - Operation::Search(request) => request.payload(), - _ => unreachable!(), - }; - assert_eq!(payload.body().as_ref(), body.as_slice()); } } } @@ -10123,43 +10111,6 @@ fn request_observation_preserves_the_raw_reasoning_effort() { } } -#[test] -fn request_observation_ignores_future_session_state_without_rewriting_the_protocol_body() { - let store = Arc::new(MemoryAccountStore::default()); - let generation = GenerateRequest::from_protocol_payload( - ProtocolPayload::json_object( - "openai", - Map::from_iter([ - ("model".to_owned(), json!("gpt-test")), - ("previous_response_id".to_owned(), json!("resp_opaque")), - ]), - ) - .expect("OpenAI payload"), - ) - .with_provider_session_state( - ProviderSessionState::new( - "openai", - Map::from_iter([("future_session_shape".to_owned(), json!([1, 2, 3]))]), - ) - .expect("provider session state"), - ); - let operation = Operation::Generate(generation); - - let client_key_id = ClientApiKeyId::new("key_openai_observation").expect("client key"); - let _observation = provider(&store).request_observation(&operation, &client_key_id); - - let Operation::Generate(generation) = &operation else { - panic!("operation should remain a generate request"); - }; - assert_eq!( - generation - .protocol_payload() - .body() - .get("previous_response_id"), - Some(&json!("resp_opaque")) - ); -} - #[tokio::test] async fn provider_compiles_catalog_presentation_for_codex_models() { let store = Arc::new(MemoryAccountStore::default());