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