diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 3bb0ff59f..43bad2e09 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -303,7 +303,7 @@ dependencies = [ "lru", "percent-encoding", "regex-lite", - "sha2 0.11.0", + "sha2", "tracing", "url", ] @@ -325,7 +325,7 @@ dependencies = [ "hmac", "http", "percent-encoding", - "sha2 0.11.0", + "sha2", "time", "tracing", ] @@ -358,7 +358,7 @@ dependencies = [ "md-5", "pin-project-lite", "sha1 0.11.0", - "sha2 0.11.0", + "sha2", "tracing", ] @@ -877,7 +877,7 @@ dependencies = [ "gateway-plugin-sdk", "hex", "serde_json", - "sha2 0.11.0", + "sha2", "tar", "tempfile", "thiserror", @@ -1688,7 +1688,7 @@ dependencies = [ "semver", "serde", "serde_json", - "sha2 0.11.0", + "sha2", "subtle", "syn 3.0.6", "tempfile", @@ -1719,7 +1719,7 @@ dependencies = [ "secrecy", "serde", "serde_json", - "sha2 0.11.0", + "sha2", "thiserror", "tokio", "tokio-tungstenite 0.29.0", @@ -1749,7 +1749,7 @@ dependencies = [ "semver", "serde", "serde_json", - "sha2 0.11.0", + "sha2", "thiserror", "tracing", "url", @@ -1789,7 +1789,7 @@ dependencies = [ "serde_ignored", "serde_json", "serde_path_to_error", - "sha2 0.11.0", + "sha2", "tar", "tempfile", "thiserror", @@ -1830,7 +1830,7 @@ dependencies = [ "semver", "serde", "serde_json", - "sha2 0.11.0", + "sha2", "sqlx", "tar", "tempfile", @@ -1858,6 +1858,7 @@ dependencies = [ name = "gateway-protocol" version = "0.0.0" dependencies = [ + "http", "httpdate", "memchr", "serde", @@ -1886,7 +1887,7 @@ dependencies = [ "secrecy", "serde", "serde_json", - "sha2 0.11.0", + "sha2", "sqlx", "tempfile", "thiserror", @@ -2524,6 +2525,15 @@ version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "libredox" +version = "0.1.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61ff90caf6077a803a240f62fdbe88645a890bbca49ef8174c3cb0404362171d" +dependencies = [ + "libc", +] + [[package]] name = "libsqlite3-sys" version = "0.37.0" @@ -2841,6 +2851,24 @@ dependencies = [ "libc", ] +[[package]] +name = "objc2-core-foundation" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536" +dependencies = [ + "bitflags", +] + +[[package]] +name = "objc2-system-configuration" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7216bd11cbda54ccabcab84d523dc93b858ec75ecfb3a7d89513fa22464da396" +dependencies = [ + "objc2-core-foundation", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -2929,7 +2957,7 @@ dependencies = [ "elliptic-curve", "primefield", "primeorder", - "sha2 0.11.0", + "sha2", ] [[package]] @@ -3270,7 +3298,7 @@ dependencies = [ "semver", "serde", "serde_json", - "sha2 0.11.0", + "sha2", "subtle", "tar", "tempfile", @@ -3312,7 +3340,7 @@ dependencies = [ "semver", "serde", "serde_json", - "sha2 0.11.0", + "sha2", "thiserror", "tokio", "tracing", @@ -4023,17 +4051,6 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d" -[[package]] -name = "sha2" -version = "0.10.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" -dependencies = [ - "cfg-if", - "cpufeatures 0.2.17", - "digest 0.10.7", -] - [[package]] name = "sha2" version = "0.11.0" @@ -4154,8 +4171,7 @@ dependencies = [ [[package]] name = "sqlx" version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "378620ccc25c62c89d8be1c819e76a88d59bdcc3304733330788948e619bfd71" +source = "git+https://github.com/transact-rs/sqlx?rev=f1e94ec34dd5f11ad6488a3a8ac7988439a3f757#f1e94ec34dd5f11ad6488a3a8ac7988439a3f757" dependencies = [ "sqlx-core", "sqlx-macros", @@ -4167,8 +4183,7 @@ dependencies = [ [[package]] name = "sqlx-core" version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05b44e85bf579a8eeb4ceaa77a3a523baf2bf0e9bac7e40f405d537b5d2d5ccb" +source = "git+https://github.com/transact-rs/sqlx?rev=f1e94ec34dd5f11ad6488a3a8ac7988439a3f757#f1e94ec34dd5f11ad6488a3a8ac7988439a3f757" dependencies = [ "base64 0.22.1", "bytes", @@ -4191,7 +4206,7 @@ dependencies = [ "rustls", "serde", "serde_json", - "sha2 0.10.9", + "sha2", "smallvec", "thiserror", "tokio", @@ -4205,8 +4220,7 @@ dependencies = [ [[package]] name = "sqlx-macros" version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bd2b84f2bc39a5705ef27ec785a11c934a41bbd4a24941e257927cddc26b60bf" +source = "git+https://github.com/transact-rs/sqlx?rev=f1e94ec34dd5f11ad6488a3a8ac7988439a3f757#f1e94ec34dd5f11ad6488a3a8ac7988439a3f757" dependencies = [ "proc-macro2", "quote", @@ -4218,8 +4232,7 @@ dependencies = [ [[package]] name = "sqlx-macros-core" version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb8d96de5fdc85a5c4ec813432b523ec637e80ba98f046555f75f7908ddac7c3" +source = "git+https://github.com/transact-rs/sqlx?rev=f1e94ec34dd5f11ad6488a3a8ac7988439a3f757#f1e94ec34dd5f11ad6488a3a8ac7988439a3f757" dependencies = [ "cfg-if", "dotenvy", @@ -4230,7 +4243,7 @@ dependencies = [ "quote", "serde", "serde_json", - "sha2 0.10.9", + "sha2", "sqlx-core", "sqlx-mysql", "sqlx-postgres", @@ -4244,8 +4257,7 @@ dependencies = [ [[package]] name = "sqlx-mysql" version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "90b8020fe17c5f2c245bfa2505d7ef59c5604839527c740266ad2214acebea27" +source = "git+https://github.com/transact-rs/sqlx?rev=f1e94ec34dd5f11ad6488a3a8ac7988439a3f757#f1e94ec34dd5f11ad6488a3a8ac7988439a3f757" dependencies = [ "bitflags", "byteorder", @@ -4262,7 +4274,7 @@ dependencies = [ "percent-encoding", "serde", "sha1 0.11.0", - "sha2 0.11.0", + "sha2", "sqlx-core", "thiserror", "tracing", @@ -4272,8 +4284,7 @@ dependencies = [ [[package]] name = "sqlx-postgres" version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "87a2bdd6e83f6b3ea525ca9fee568030508b58355a43d0b2c1674d5f79dcd65e" +source = "git+https://github.com/transact-rs/sqlx?rev=f1e94ec34dd5f11ad6488a3a8ac7988439a3f757#f1e94ec34dd5f11ad6488a3a8ac7988439a3f757" dependencies = [ "atoi", "base64 0.22.1", @@ -4296,7 +4307,7 @@ dependencies = [ "rand 0.10.2", "serde", "serde_json", - "sha2 0.11.0", + "sha2", "smallvec", "sqlx-core", "stringprep", @@ -4309,8 +4320,7 @@ dependencies = [ [[package]] name = "sqlx-sqlite" version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "488e99c397a62007e4229aec669a179816339afc6d2620ca6fa420dbee2e982c" +source = "git+https://github.com/transact-rs/sqlx?rev=f1e94ec34dd5f11ad6488a3a8ac7988439a3f757#f1e94ec34dd5f11ad6488a3a8ac7988439a3f757" dependencies = [ "atoi", "chrono", @@ -4493,7 +4503,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.3", + "getrandom 0.3.4", "once_cell", "rustix", "windows-sys 0.61.2", @@ -5182,6 +5192,12 @@ name = "whoami" version = "2.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "998767ef88740d1f5b0682a9c53c24431453923962269c2db68ee43788c5a40d" +dependencies = [ + "libc", + "libredox", + "objc2-system-configuration", + "web-sys", +] [[package]] name = "winapi" diff --git a/backend/Cargo.toml b/backend/Cargo.toml index a8ebbc87d..84752fc63 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -59,3 +59,5 @@ panic = "abort" [patch.crates-io] # Upstream SOCKS IPv6 fix; remove once a release includes hyperium/hyper-util#302. hyper-util = { git = "https://github.com/hyperium/hyper-util", rev = "191b1acf0a6b18e807e4abf5a6d4a7c47979ecd6" } +# 固定上游 BEGIN 取消回滚修复,避免未结束事务复用;见 transact-rs/sqlx#4394 +sqlx = { git = "https://github.com/transact-rs/sqlx", rev = "f1e94ec34dd5f11ad6488a3a8ac7988439a3f757" } diff --git a/backend/crates/gateway-api/tests/openai/responses/http.rs b/backend/crates/gateway-api/tests/openai/responses/http.rs index 468315367..fff13fb84 100644 --- a/backend/crates/gateway-api/tests/openai/responses/http.rs +++ b/backend/crates/gateway-api/tests/openai/responses/http.rs @@ -2279,6 +2279,7 @@ async fn buffered_response_commits_only_after_complete_json_is_encoded() { async fn buffered_response_collects_completed_output_items_without_rewriting_streams() { let message = json!({ "id": "msg_test", "type": "message", "role": "assistant", "status": "completed", + "phase": "partial_answer", "content": [{"type": "output_text", "text": "完整文本", "annotations": []}], "future_item_field": {"keep": true} }); @@ -2288,6 +2289,7 @@ async fn buffered_response_collects_completed_output_items_without_rewriting_str }); let terminal = json!({ "id": "resp_test", "status": "completed", "output": [], + "end_turn": false, "usage": {"input_tokens": 20, "output_tokens": 8, "total_tokens": 28}, "future_terminal_field": {"keep": true} }); diff --git a/backend/crates/gateway-protocol/Cargo.toml b/backend/crates/gateway-protocol/Cargo.toml index 5af5138fe..053be5af1 100644 --- a/backend/crates/gateway-protocol/Cargo.toml +++ b/backend/crates/gateway-protocol/Cargo.toml @@ -12,6 +12,7 @@ name = "main" path = "tests/main.rs" [dependencies] +http = "1.5.0" httpdate = "1.0.3" memchr = "2.8.3" serde = { version = "1.0.228", features = ["derive"] } diff --git a/backend/crates/gateway-protocol/src/openai/events.rs b/backend/crates/gateway-protocol/src/openai/events.rs index bf6aa2f45..63e0120e0 100644 --- a/backend/crates/gateway-protocol/src/openai/events.rs +++ b/backend/crates/gateway-protocol/src/openai/events.rs @@ -275,27 +275,21 @@ fn retry_after_seconds_field(value: &Value) -> Option { } fn retry_after_seconds_header(value: &Value) -> Option { - value - .get("headers") - .and_then(Value::as_object) - .and_then(|headers| { - headers.iter().find_map(|(name, value)| { - if name.eq_ignore_ascii_case("retry-after") { - json_value_as_retry_after_seconds(value) - } else { - None - } - }) + let headers = value.get("headers")?.as_object()?; + // 官方先按 HTTP 头值校验,再覆盖大小写不同的同名头;最后才解释重试时间 + // 合法但无法解析为时间的末项也会覆盖前值,非法头值和数组不参与覆盖 + let header = headers + .iter() + .filter(|(name, _)| name.eq_ignore_ascii_case("retry-after")) + .filter_map(|(_, value)| match value { + Value::String(value) => http::HeaderValue::from_str(value).ok(), + Value::Number(_) | Value::Bool(_) => { + http::HeaderValue::from_str(&value.to_string()).ok() + } + Value::Null | Value::Array(_) | Value::Object(_) => None, }) -} - -fn json_value_as_retry_after_seconds(value: &Value) -> Option { - match value { - Value::Number(value) => value.as_u64(), - Value::String(value) => super::headers::parse_retry_after_seconds(value), - Value::Array(values) => values.first().and_then(json_value_as_retry_after_seconds), - Value::Null | Value::Bool(_) | Value::Object(_) => None, - } + .next_back()?; + super::headers::parse_retry_after_seconds(header.to_str().ok()?) } /// 标准化的单个限流窗口 diff --git a/backend/crates/gateway-protocol/tests/openai/events.rs b/backend/crates/gateway-protocol/tests/openai/events.rs index cb7a6afd4..7abeb55a3 100644 --- a/backend/crates/gateway-protocol/tests/openai/events.rs +++ b/backend/crates/gateway-protocol/tests/openai/events.rs @@ -406,14 +406,14 @@ fn retry_after_should_read_nested_error_delay() { } #[test] -fn retry_after_should_read_case_insensitive_header_array() { +fn retry_after_should_ignore_header_arrays() { let body = json!({ "type": "error", "headers": {"Retry-After": ["37"]} }) .to_string(); - assert_eq!(retry_after_seconds_from_body(&body), Some(37)); + assert_eq!(retry_after_seconds_from_body(&body), None); } #[test] @@ -466,7 +466,7 @@ fn retry_after_should_reject_malformed_json() { #[test] fn retry_after_should_prioritize_nested_headers_over_other_advice() { - for header in [json!(30), json!("30"), json!(["30"])] { + for header in [json!(30), json!("30")] { for wrap_response in [false, true] { let error = json!({ "code": "rate_limit_exceeded", "message": "Try again in 2s", @@ -482,6 +482,33 @@ fn retry_after_should_prioritize_nested_headers_over_other_advice() { } } +#[test] +fn retry_after_should_validate_http_values_before_resolving_duplicates() { + for (headers, expected) in [ + (json!("invalid"), 12), + (json!({"retry-after":"\n5\n"}), 12), + (json!({"retry-after":"\t5\t"}), 5), + (json!({"Retry-After":"5", "retry-after":"30"}), 30), + (json!({"Retry-After":"5", "retry-after":"invalid"}), 12), + (json!({"Retry-After":"5", "retry-after":"\n30\n"}), 5), + (json!({"Retry-After":"5", "retry-after":["30"]}), 5), + (json!({"Retry-After":"5", "retry-after":false}), 12), + ] { + for wrap_response in [false, true] { + let error = json!({"code":"rate_limit_exceeded", "headers":headers}); + let body = if wrap_response { + json!({"type":"response.failed", "response":{"error":error}, "headers":{"retry-after":"12"}}) + } else { + json!({"type":"error", "error":error, "headers":{"retry-after":"12"}}) + }; + assert_eq!( + retry_after_seconds_from_body(&body.to_string()), + Some(expected) + ); + } + } +} + #[test] fn retry_after_should_keep_zero_advice_and_fall_back_only_for_invalid_headers() { for (header, expected) in [ diff --git a/backend/crates/gateway-store/tests/postgres/connection.rs b/backend/crates/gateway-store/tests/postgres/connection.rs index 1b3a77e30..225bd4bc9 100644 --- a/backend/crates/gateway-store/tests/postgres/connection.rs +++ b/backend/crates/gateway-store/tests/postgres/connection.rs @@ -4,6 +4,90 @@ use std::error::Error as _; use super::*; +#[tokio::test] +async fn cancelled_transaction_begin_should_return_a_clean_connection_to_the_pool() { + let Some(database_url) = crate::support::test_env("CPR_TEST_DATABASE_URL") else { + return; + }; + let pool = PgPoolOptions::new() + .max_connections(1) + .connect(&database_url) + .await + .unwrap(); + let observer = PgPoolOptions::new() + .max_connections(1) + .connect(&database_url) + .await + .unwrap(); + let pid: i32 = sqlx::query_scalar("select pg_backend_pid()") + .fetch_one(&pool) + .await + .unwrap(); + let lock = i64::from_le_bytes(Uuid::new_v4().as_bytes()[..8].try_into().unwrap()); + sqlx::query("select pg_advisory_lock($1)") + .bind(lock) + .execute(&observer) + .await + .unwrap(); + let pending_pool = pool.clone(); + let begin = tokio::spawn(async move { + pending_pool + .begin_with(sqlx::AssertSqlSafe(format!( + "begin; select pg_advisory_xact_lock({lock})" + ))) + .await + }); + // 等服务端进入 BEGIN 的同一次往返再取消,避免依赖客户端调度时机 + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let waiting: bool = sqlx::query_scalar( + "select coalesce(wait_event = 'advisory', false) from pg_stat_activity where pid=$1", + ) + .bind(pid) + .fetch_one(&observer) + .await + .unwrap(); + if waiting { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("BEGIN must reach PostgreSQL before cancellation"); + begin.abort(); + assert!(begin.await.unwrap_err().is_cancelled()); + sqlx::query("select pg_advisory_unlock($1)") + .bind(lock) + .execute(&observer) + .await + .unwrap(); + + // 重新借出意味着池已处理被取消语句及待回滚帧;必须复用干净的原连接 + let mut connection = tokio::time::timeout(Duration::from_secs(5), pool.acquire()) + .await + .unwrap() + .unwrap(); + let reused_pid: i32 = sqlx::query_scalar("select pg_backend_pid()") + .fetch_one(&mut *connection) + .await + .unwrap(); + let state: String = sqlx::query_scalar("select state from pg_stat_activity where pid=$1") + .bind(pid) + .fetch_one(&observer) + .await + .unwrap(); + drop(connection); + pool.close().await; + observer.close().await; + + assert_eq!(reused_pid, pid); + assert_eq!( + state, "idle", + "cancelled BEGIN must not leak its transaction" + ); +} + #[tokio::test] async fn invalid_postgres_options_preserve_the_sqlx_source() { let error = gateway_store::postgres::connect_and_migrate( diff --git a/backend/crates/gateway-store/tests/redis/credential_cooldown.rs b/backend/crates/gateway-store/tests/redis/credential_cooldown.rs index 318b013de..5de927178 100644 --- a/backend/crates/gateway-store/tests/redis/credential_cooldown.rs +++ b/backend/crates/gateway-store/tests/redis/credential_cooldown.rs @@ -347,7 +347,21 @@ async fn capacity_freeze_survives_due_time_and_runtime_restart_until_probe_succe .await .expect("write due freeze"); let runtime = runtime_store(&repository, connection.clone(), &namespace); - let snapshot = runtime.active_rate_limits().await.expect("status snapshot"); + let snapshot = runtime.active_rate_limits().await.unwrap_or_else(|error| { + // 来源默认不参与 Debug;只输出错误类别,保留诊断线索且不展开连接信息 + let mut source: Option<&(dyn std::error::Error + 'static)> = Some(&error); + while let Some(cause) = source { + if let Some(redis_error) = cause.downcast_ref::() { + panic!( + "status snapshot: Redis {:?}, timeout={}", + redis_error.kind(), + redis_error.is_timeout() + ); + } + source = cause.source(); + } + panic!("status snapshot: {error}"); + }); assert!(snapshot.cooldown[&frozen.provider_account_id].is_active(SystemTime::now())); assert!( snapshot.cooldown[&frozen.provider_account_id] diff --git a/backend/crates/providers/openai/tests/transport/downstream/body.rs b/backend/crates/providers/openai/tests/transport/downstream/body.rs index 0865d240b..2ec6a49be 100644 --- a/backend/crates/providers/openai/tests/transport/downstream/body.rs +++ b/backend/crates/providers/openai/tests/transport/downstream/body.rs @@ -5,6 +5,36 @@ use provider_openai::encode_generate_request; use super::super::*; +#[test] +fn encoder_should_preserve_lite_input_without_top_level_instructions_or_tools() { + let body = json!({ + "model": "gpt-test", + "store": false, + "input": [ + {"type": "message", "role": "developer", "content": [ + {"type": "input_text", "text": "Base instructions."} + ]}, + {"type": "additional_tools", "tools": [ + {"type": "function", "name": "lookup", "parameters": {"type": "object"}} + ]}, + {"type": "message", "role": "assistant", "phase": "partial_answer", + "content": [{"type": "output_text", "text": "Still working."}]}, + {"type": "message", "role": "user", "content": "Continue."} + ] + }); + let encoded = encode_downstream_request(body.clone()); + + assert_eq!(Value::Object(encoded.body().clone()), body); + let websocket = + provider_openai::transport::protocol::websocket::websocket_response_create_payload_text( + &encoded, + ) + .expect("encode Lite WebSocket request"); + let mut expected = body; + expected["type"] = json!("response.create"); + assert_eq!(serde_json::from_str::(&websocket).unwrap(), expected); +} + #[test] fn encoder_should_adapt_pi_responses_parameters_without_losing_codex_fields() { // 对照本机 Pi 0.79.0 普通 Responses 适配在 onPayload 阶段生成的正文; diff --git a/backend/crates/providers/xai/src/transport/request/tools.rs b/backend/crates/providers/xai/src/transport/request/tools.rs index fdac10f73..b7dbeb4b4 100644 --- a/backend/crates/providers/xai/src/transport/request/tools.rs +++ b/backend/crates/providers/xai/src/transport/request/tools.rs @@ -11,18 +11,6 @@ pub(super) struct ToolNormalizer { server_search_eager: bool, } -pub(super) struct NormalizedInputItems { - items: Vec, - loaded_tools: Vec, - visible_tools: Vec, -} - -pub(super) struct NormalizedToolSearchOutput { - history: Map, - loaded_tools: Vec, - visible_tools: Vec, -} - impl ToolNormalizer { pub(super) fn new() -> Self { Self { @@ -55,10 +43,13 @@ impl ToolNormalizer { ) -> Result { let (tools, had_tools) = optional_array(payload.get("tools")).map_err(|error| error.at_field("tools"))?; - if had_tools { - self.response.visible_tools.clone_from(&tools); - } - let client_search = inspect_tool_search(&tools).map_err(|error| error.at_field("tools"))?; + // 先收敛历史中的增量声明,再生成别名和响应转换事实,避免旧 schema 残留 + let input_tools = collect_input_tool_declarations(payload.get("input")) + .map_err(|error| error.at_field("input"))?; + self.response.visible_tools = + merge_tool_declarations(tools.iter().chain(&input_tools).cloned().collect()); + let client_search = inspect_tool_search(&tools).map_err(|error| error.at_field("tools"))? + || inspect_tool_search(&input_tools).map_err(|error| error.at_field("input"))?; self.normalize_client_search_parallel(payload, client_search) .map_err(|error| error.at_field("parallel_tool_calls"))?; @@ -69,14 +60,18 @@ impl ToolNormalizer { .map_err(|error| error.at_field("tools"))?, ); } + for raw_tool in &input_tools { + normalized_tools.extend( + self.normalize_tool(raw_tool, "", client_search, true) + .map_err(|error| error.at_field("input"))?, + ); + } if let Some(Value::Array(items)) = payload.get("input") { let normalized = self .normalize_input_items(items) .map_err(|error| error.at_field("input"))?; - normalized_tools.extend(normalized.loaded_tools); - self.response.visible_tools.extend(normalized.visible_tools); - payload.insert("input".to_owned(), Value::Array(normalized.items)); + payload.insert("input".to_owned(), Value::Array(normalized)); } else if payload .get("input") .is_some_and(|input| !input.is_null() && !input.is_string()) @@ -251,6 +246,8 @@ impl ToolNormalizer { let alias = self.alias(ToolIdentity::new(ToolKind::Function, namespace, name)); if let Some(schema) = function_schema { self.response.function_schemas.insert(alias.clone(), schema); + } else { + self.response.function_schemas.remove(&alias); } converted.insert("name".to_owned(), Value::String(alias)); Ok(vec![Value::Object(converted)]) @@ -513,11 +510,70 @@ pub(super) fn short_tool_hash(value: &str) -> String { pub(super) fn dedupe_normalized_tools(tools: Vec) -> Vec { let mut result = Vec::with_capacity(tools.len()); - let mut seen = BTreeSet::new(); + let mut positions = BTreeMap::new(); for tool in tools { - if normalized_tool_dedupe_key(&tool).is_none_or(|key| seen.insert(key)) { - result.push(tool); + if let Some(key) = normalized_tool_dedupe_key(&tool) { + if let Some(&index) = positions.get(&key) { + result[index] = tool; + continue; + } + positions.insert(key, result.len()); } + result.push(tool); + } + result +} + +fn collect_input_tool_declarations( + input: Option<&Value>, +) -> Result, GrokRequestEncodeError> { + let mut declarations = Vec::new(); + for item in input.and_then(Value::as_array).into_iter().flatten() { + if !matches!( + item.get("type").and_then(Value::as_str).map(str::trim), + Some("additional_tools" | "tool_search_output") + ) { + continue; + } + let tools = item + .get("tools") + .and_then(Value::as_array) + .ok_or(GrokRequestEncodeError::InvalidRequestNormalization)?; + declarations.extend(tools.iter().cloned()); + } + Ok(merge_tool_declarations(declarations)) +} + +/// 官方增量声明按名字覆盖,namespace 只更新本次声明的成员,保留其他成员 +fn merge_tool_declarations(tools: Vec) -> Vec { + let mut result: Vec = Vec::with_capacity(tools.len()); + let mut positions = BTreeMap::::new(); + for mut tool in tools { + let key = match tool.get("type").and_then(Value::as_str) { + Some("namespace" | "function" | "custom") => tool + .get("name") + .and_then(Value::as_str) + .map(|name| format!("name:{name}")), + _ => normalized_tool_dedupe_key(&tool), + }; + if let Some(key) = key { + if let Some(&index) = positions.get(&key) { + let previous = &result[index]; + if previous.get("type").and_then(Value::as_str) == Some("namespace") + && tool.get("type").and_then(Value::as_str) == Some("namespace") + && let Some(previous_members) = previous.get("tools").and_then(Value::as_array) + && let Some(members) = tool.get_mut("tools").and_then(Value::as_array_mut) + { + let mut combined = previous_members.clone(); + combined.append(members); + *members = merge_tool_declarations(combined); + } + result[index] = tool; + continue; + } + positions.insert(key, result.len()); + } + result.push(tool); } result } @@ -918,10 +974,8 @@ impl ToolNormalizer { fn normalize_input_items( &mut self, items: &[Value], - ) -> Result { + ) -> Result, GrokRequestEncodeError> { let mut rewritten = Vec::with_capacity(items.len()); - let mut loaded_tools = Vec::new(); - let mut visible_tools = Vec::new(); for raw_item in items { let Some(item) = raw_item.as_object() else { rewritten.push(raw_item.clone()); @@ -960,10 +1014,7 @@ impl ToolNormalizer { rewritten.push(Value::Object(self.normalize_tool_search_call(item)?)); } "tool_search_output" => { - let normalized = self.normalize_tool_search_output(item)?; - rewritten.push(Value::Object(normalized.history)); - loaded_tools.extend(normalized.loaded_tools); - visible_tools.extend(normalized.visible_tools); + rewritten.push(Value::Object(self.normalize_tool_search_output(item)?)); } "custom_tool_call" => { rewritten.push(Value::Object(self.normalize_custom_tool_call_input(item)?)); @@ -988,22 +1039,14 @@ impl ToolNormalizer { "compaction_trigger" => { return Err(GrokRequestEncodeError::InvalidRequestNormalization); } - "additional_tools" => { - let (tools, visible) = self.normalize_additional_tools_input(item)?; - loaded_tools.extend(tools); - visible_tools.extend(visible); - } + "additional_tools" => {} "" => rewritten.push(raw_item.clone()), unsupported => { rewritten.push(unsupported_input_history_boundary(item, unsupported)) } } } - Ok(NormalizedInputItems { - items: rewritten, - loaded_tools, - visible_tools, - }) + Ok(rewritten) } fn normalize_message_input( @@ -1269,7 +1312,7 @@ impl ToolNormalizer { fn normalize_tool_search_output( &mut self, item: &Map, - ) -> Result { + ) -> Result, GrokRequestEncodeError> { let execution = string_field(item, "execution").trim().to_ascii_lowercase(); if !matches!(execution.as_str(), "" | "client" | "server") { return Err(GrokRequestEncodeError::InvalidRequestNormalization); @@ -1279,10 +1322,6 @@ impl ToolNormalizer { .get("tools") .and_then(Value::as_array) .ok_or(GrokRequestEncodeError::InvalidRequestNormalization)?; - let mut normalized = Vec::new(); - for tool in tools { - normalized.extend(self.normalize_tool(tool, "", false, true)?); - } let message = format!( "Tool search completed; {} selected tool definitions are now available.", tools.len() @@ -1300,11 +1339,7 @@ impl ToolNormalizer { self.server_search_eager = true; boundary_message(&message) }; - Ok(NormalizedToolSearchOutput { - history, - loaded_tools: normalized, - visible_tools: tools.clone(), - }) + Ok(history) } fn normalize_apply_patch_call_input( @@ -1324,21 +1359,6 @@ impl ToolNormalizer { ("arguments".to_owned(), Value::String(arguments)), ])) } - - fn normalize_additional_tools_input( - &mut self, - item: &Map, - ) -> Result<(Vec, Vec), GrokRequestEncodeError> { - let tools = item - .get("tools") - .and_then(Value::as_array) - .ok_or(GrokRequestEncodeError::InvalidRequestNormalization)?; - let mut normalized = Vec::new(); - for raw_tool in tools { - normalized.extend(self.normalize_tool(raw_tool, "", false, true)?); - } - Ok((normalized, tools.clone())) - } } pub(super) fn normalize_input_file_part(item: &Map) -> Map { diff --git a/backend/crates/providers/xai/tests/transport/request/incremental_tools.rs b/backend/crates/providers/xai/tests/transport/request/incremental_tools.rs new file mode 100644 index 000000000..65927c249 --- /dev/null +++ b/backend/crates/providers/xai/tests/transport/request/incremental_tools.rs @@ -0,0 +1,112 @@ +//! 验证 Codex 增量工具声明覆盖、namespace 合并及响应 schema 同步 + +use super::*; +use provider_xai::GrokCanonicalDecoder; + +fn declaration(tools: Vec) -> Value { + json!({"type":"additional_tools", "role":"developer", "tools":tools}) +} + +fn function(name: &str, parameter_type: &str) -> Value { + json!({"type":"function", "name":name, "parameters":{ + "type":"object", "properties":{"value":{"type":parameter_type}} + }}) +} + +#[test] +fn namespace_updates_should_replace_members_and_preserve_undeclared_members() { + let old = json!({"type":"namespace", "name":"workspace", "tools":[ + function("lookup", "integer"), function("read", "string") + ]}); + let latest = json!({"type":"namespace", "name":"workspace", "tools":[ + function("lookup", "string"), function("write", "string") + ]}); + let encoded = GrokResponsesRequest::encode( + &raw_request(json!({"input":[declaration(vec![old]), declaration(vec![latest])]})), + "grok-4.5", + &client_key(), + ) + .unwrap(); + let tools = encoded.body()["tools"].as_array().unwrap(); + assert_eq!(tools.len(), 3); + assert_eq!(tools[0]["name"], "workspace__lookup"); + assert_eq!( + tools[0]["parameters"]["properties"]["value"]["type"], + "string" + ); + assert_eq!(tools[1]["name"], "workspace__read"); + assert_eq!(tools[2]["name"], "workspace__write"); + + let created = json!({"type":"response.created", "response":{"id":"resp_incremental"}}); + let completed = json!({"type":"response.completed", "response":{ + "id":"resp_incremental", "status":"completed", "tools":tools, "output":[] + }}); + let events = GrokCanonicalDecoder::for_request("grok-4.5", &encoded) + .push(format!("data: {created}\n\ndata: {completed}\n\n").as_bytes()) + .unwrap(); + let restored = events + .iter() + .filter_map(|event| event.wire_event()) + .find_map(|wire| wire.data().pointer("/response/tools")) + .unwrap(); + assert_eq!( + restored, + &json!([{"type":"namespace", "name":"workspace", "tools":[ + function("lookup", "string"), function("read", "string"), function("write", "string") + ]}]) + ); +} + +#[test] +fn replaced_integer_schema_should_not_rewrite_new_number_arguments() { + let encoded = GrokResponsesRequest::encode( + &raw_request(json!({"tools":[function("lookup", "integer")], "input":[ + declaration(vec![function("lookup", "number")]) + ]})), + "grok-4.5", + &client_key(), + ) + .unwrap(); + let event = json!({"type":"response.output_item.done", "output_index":0, "item":{ + "type":"function_call", "id":"fc_lookup", "call_id":"call_lookup", + "name":"lookup", "arguments":"{\"value\":1.0}" + }}); + let events = GrokCanonicalDecoder::for_request("grok-4.5", &encoded) + .push(format!("data: {event}\n\n").as_bytes()) + .unwrap(); + let arguments = events + .iter() + .filter_map(|event| event.wire_event()) + .find_map(|wire| wire.data().pointer("/item/arguments")) + .unwrap(); + assert_eq!(arguments, &json!("{\"value\":1.0}")); +} + +#[test] +fn later_declaration_should_replace_the_tool_kind_before_alias_generation() { + let encoded = GrokResponsesRequest::encode( + &raw_request(json!({"input":[ + declaration(vec![function("lookup", "integer")]), + declaration(vec![json!({"type":"custom", "name":"lookup"})]) + ]})), + "grok-4.5", + &client_key(), + ) + .unwrap(); + let tools = encoded.body()["tools"].as_array().unwrap(); + assert_eq!(tools.len(), 1); + assert_eq!(tools[0]["name"], "lookup"); + assert_eq!(tools[0]["parameters"]["required"], json!(["input"])); +} + +#[test] +fn tool_search_results_should_follow_the_same_declaration_order() { + let latest = function("lookup", "number"); + let encoded = GrokResponsesRequest::encode( + &raw_request(json!({"input":[ + declaration(vec![function("lookup", "integer")]), + {"type":"tool_search_output", "call_id":"search_1", "execution":"server", "tools":[latest]} + ]})), "grok-4.5", &client_key(), + ).unwrap(); + assert_eq!(encoded.body()["tools"], json!([latest])); +} diff --git a/backend/crates/providers/xai/tests/transport/request/mod.rs b/backend/crates/providers/xai/tests/transport/request/mod.rs index 3f7813f27..8956405e0 100644 --- a/backend/crates/providers/xai/tests/transport/request/mod.rs +++ b/backend/crates/providers/xai/tests/transport/request/mod.rs @@ -7,6 +7,7 @@ use serde_json::{Map, Value, json}; use provider_xai::{GrokRequestEncodeError, GrokResponsesRequest}; mod hosted_names; +mod incremental_tools; mod model_identity; mod parameters; mod web_search; @@ -713,14 +714,14 @@ fn explicit_session_should_add_x_search_after_codex_additional_tools_normalizati } #[test] -fn additional_tools_should_keep_top_level_definition_and_drop_unsupported_types() { +fn additional_tools_should_replace_earlier_definition_and_drop_unsupported_types() { let request = raw_request(json!({ "model": "client", "tools": [ { "type": "function", "name": "existing", - "description": "top-level wins", + "description": "initial definition", "parameters": {"type": "object"} }, {"type": "image_generation", "model": "gpt-image-2"} @@ -733,7 +734,7 @@ fn additional_tools_should_keep_top_level_definition_and_drop_unsupported_types( { "type": "function", "name": "existing", - "description": "carrier duplicate", + "description": "updated definition", "parameters": {"type": "object"} }, {"type": "function", "name": "wait"}, @@ -753,7 +754,7 @@ fn additional_tools_should_keep_top_level_definition_and_drop_unsupported_types( assert_eq!(body.pointer("/tools/0/name"), Some(&json!("existing"))); assert_eq!( body.pointer("/tools/0/description"), - Some(&json!("top-level wins")) + Some(&json!("updated definition")) ); assert_eq!(body.pointer("/tools/1/name"), Some(&json!("wait"))); assert_eq!(body.pointer("/tools/2"), None);