mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-07 08:05:32 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f905e4b3fb | ||
|
|
ed3e23887b | ||
|
|
91f35d88cf | ||
|
|
f465a337e8 | ||
|
|
37465e2872 | ||
|
|
70c18dadd6 | ||
|
|
881a0eaef6 | ||
|
|
95f7a5a636 | ||
|
|
871406ab24 | ||
|
|
38a1cc6933 | ||
|
|
d38e0d4296 | ||
|
|
d12d75a037 | ||
|
|
a2922b410e | ||
|
|
d303efc6f7 |
@@ -626,7 +626,7 @@ jobs:
|
||||
cosign sign --yes "${images[@]}"
|
||||
|
||||
- name: Publish GitHub Release
|
||||
uses: softprops/action-gh-release@3d0d9888cb7fd7b750713d6e236d1fcb99157228 # v3.0.2
|
||||
uses: softprops/action-gh-release@efb35369e0ad2afab669f228072c1b0d510eae64 # v3.0.3
|
||||
with:
|
||||
tag_name: ${{ needs.release-info.outputs.tag }}
|
||||
generate_release_notes: true
|
||||
|
||||
Generated
+25
-25
@@ -154,9 +154,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "aws-lc-rs"
|
||||
version = "1.16.2"
|
||||
version = "1.18.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a054912289d18629dc78375ba2c3726a3afe3ff71b4edba9dedfca0e3446d1fc"
|
||||
checksum = "b281d307588d634de920874890732659e2e7672f72b5e10e81badc1a8a83621e"
|
||||
dependencies = [
|
||||
"aws-lc-sys",
|
||||
"untrusted 0.7.1",
|
||||
@@ -165,14 +165,15 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "aws-lc-sys"
|
||||
version = "0.39.0"
|
||||
version = "0.45.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1fa7e52a4c5c547c741610a2c6f123f3881e409b714cd27e6798ef020c514f0a"
|
||||
checksum = "9bff6c3b54fad79a2e60b8102caf565819711497c1f5f092f49508e2f5c31b27"
|
||||
dependencies = [
|
||||
"cc",
|
||||
"cmake",
|
||||
"dunce",
|
||||
"fs_extra",
|
||||
"pkg-config",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -400,9 +401,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "aws-smithy-runtime-api"
|
||||
version = "1.15.0"
|
||||
version = "1.16.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "954c563ce84507722d2679f07a35d21b9c6466b3872d513020d0281fc8112ac9"
|
||||
checksum = "9c054752dd9e4dc73d0b75748c99ac2d0feafbf2f25c7b0516f03a3534161223"
|
||||
dependencies = [
|
||||
"aws-smithy-async",
|
||||
"aws-smithy-runtime-api-macros",
|
||||
@@ -440,9 +441,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "aws-smithy-types"
|
||||
version = "1.6.2"
|
||||
version = "1.6.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "fce83ce9abbb198d25bc7131e468d0f9fe1257125e58c39f3f9fc9f5098c9647"
|
||||
checksum = "8f94d16e797ec62cd999fc9d5942b48fa7050c3093ddadff48e4d7528d16fcb9"
|
||||
dependencies = [
|
||||
"base64-simd",
|
||||
"bytes",
|
||||
@@ -1257,9 +1258,9 @@ checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
|
||||
|
||||
[[package]]
|
||||
name = "flate2"
|
||||
version = "1.1.9"
|
||||
version = "1.1.10"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c"
|
||||
checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb"
|
||||
dependencies = [
|
||||
"crc32fast",
|
||||
"miniz_oxide",
|
||||
@@ -1466,7 +1467,7 @@ dependencies = [
|
||||
"tokio",
|
||||
"tokio-tungstenite 0.29.0",
|
||||
"tower",
|
||||
"tower-http 0.7.0",
|
||||
"tower-http 0.7.1",
|
||||
"tracing",
|
||||
"url",
|
||||
"uuid",
|
||||
@@ -1835,9 +1836,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "hyper"
|
||||
version = "1.8.1"
|
||||
version = "1.11.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2ab2d4f250c3d7b1c9fcdff1cece94ea4e2dfbec68614f7b87cb205f24ca9d11"
|
||||
checksum = "27b501faa50e7a26c3d3560ca625132f4078a17771f4810baf70475ae48cbe43"
|
||||
dependencies = [
|
||||
"atomic-waker",
|
||||
"bytes",
|
||||
@@ -1850,7 +1851,6 @@ dependencies = [
|
||||
"httpdate",
|
||||
"itoa",
|
||||
"pin-project-lite",
|
||||
"pin-utils",
|
||||
"smallvec",
|
||||
"tokio",
|
||||
"want",
|
||||
@@ -2213,9 +2213,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "miniz_oxide"
|
||||
version = "0.8.9"
|
||||
version = "0.9.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316"
|
||||
checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c"
|
||||
dependencies = [
|
||||
"adler2",
|
||||
"simd-adler32",
|
||||
@@ -3026,9 +3026,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "rustls-native-certs"
|
||||
version = "0.8.3"
|
||||
version = "0.8.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "612460d5f7bea540c490b2b6395d8e34a953e52b491accd6c86c8164c5932a63"
|
||||
checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d"
|
||||
dependencies = [
|
||||
"openssl-probe",
|
||||
"rustls-pki-types",
|
||||
@@ -3038,9 +3038,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "rustls-pki-types"
|
||||
version = "1.14.0"
|
||||
version = "1.15.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "be040f8b0a225e40375822a563fa9524378b9d63112f53e19ffff34df5d33fdd"
|
||||
checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96"
|
||||
dependencies = [
|
||||
"web-time",
|
||||
"zeroize",
|
||||
@@ -3704,7 +3704,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
|
||||
dependencies = [
|
||||
"fastrand",
|
||||
"getrandom 0.3.4",
|
||||
"getrandom 0.4.3",
|
||||
"once_cell",
|
||||
"rustix",
|
||||
"windows-sys 0.61.2",
|
||||
@@ -3805,9 +3805,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
|
||||
|
||||
[[package]]
|
||||
name = "tokio"
|
||||
version = "1.52.3"
|
||||
version = "1.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe"
|
||||
checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"libc",
|
||||
@@ -3928,9 +3928,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tower-http"
|
||||
version = "0.7.0"
|
||||
version = "0.7.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b11f75e912b0c2be01b63d8cf8057b8c3f97cf34abb3d431a3a4c8675498e233"
|
||||
checksum = "08a05a66a4fdd61cbbe0a1d755ffe0ca6aba159dd4820936a0ff8a8278245b9c"
|
||||
dependencies = [
|
||||
"bitflags",
|
||||
"bytes",
|
||||
|
||||
@@ -213,13 +213,17 @@ fn capitalize_first(value: &str) -> String {
|
||||
|
||||
pub(crate) fn billing_view(billing: Option<&domain::UsageBilling>) -> Option<BillingView> {
|
||||
match billing? {
|
||||
domain::UsageBilling::Total { total, .. } => Some(BillingView {
|
||||
domain::UsageBilling::Total { source, total } => Some(BillingView {
|
||||
input_amount_display: "—".to_owned(),
|
||||
output_amount_display: "—".to_owned(),
|
||||
cache_read_amount_display: "—".to_owned(),
|
||||
cache_write_amount_display: "—".to_owned(),
|
||||
standard_amount_display: "—".to_owned(),
|
||||
total_amount_display: format_money(total),
|
||||
total_amount_display: if source == "calculated" {
|
||||
format!("≈ {}", format_money(total))
|
||||
} else {
|
||||
format_money(total)
|
||||
},
|
||||
input_price_display: "—".to_owned(),
|
||||
output_price_display: "—".to_owned(),
|
||||
cache_read_price_display: "—".to_owned(),
|
||||
|
||||
@@ -15,8 +15,8 @@ use serde_json::Value;
|
||||
use crate::ApiState;
|
||||
use crate::openai::{
|
||||
auth::{authenticate_client, client_access_error_response},
|
||||
endpoint::collect_raw_json_response,
|
||||
error::gateway_error_response,
|
||||
provider_endpoint::collect_raw_json_response,
|
||||
responses::{OpenAiRequestHeaders, request_client_context},
|
||||
};
|
||||
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
//! OpenAI 客户端协议 adapter。
|
||||
|
||||
pub mod auth;
|
||||
mod endpoint;
|
||||
pub mod error;
|
||||
pub mod images;
|
||||
pub mod models;
|
||||
mod provider_endpoint;
|
||||
pub mod responses;
|
||||
pub mod router;
|
||||
pub mod search;
|
||||
|
||||
@@ -23,6 +23,7 @@ const OUTBOUND_COMMAND_BUFFER: usize = 32;
|
||||
const INBOUND_EVENT_BUFFER: usize = 32;
|
||||
const DOWNSTREAM_WRITE_TIMEOUT: Duration = Duration::from_secs(5 * 60);
|
||||
const CONNECTION_MAX_AGE: Duration = Duration::from_secs(60 * 60);
|
||||
const DOWNSTREAM_PING_INTERVAL: Duration = Duration::from_secs(25);
|
||||
|
||||
/// WebSocket pump 的写入与生命周期预算。
|
||||
#[derive(Clone, Copy)]
|
||||
@@ -171,9 +172,27 @@ struct ConnectionStats {
|
||||
command_backpressure_count: AtomicU64,
|
||||
ping_received_count: AtomicU64,
|
||||
pong_written_count: AtomicU64,
|
||||
ping_written_count: AtomicU64,
|
||||
pong_received_count: AtomicU64,
|
||||
last_read_ms: AtomicU64,
|
||||
last_write_ms: AtomicU64,
|
||||
}
|
||||
|
||||
impl ConnectionStats {
|
||||
fn record_read(&self, opened_at: Instant) {
|
||||
self.last_read_ms.store(
|
||||
u64::try_from(opened_at.elapsed().as_millis()).unwrap_or(u64::MAX),
|
||||
Ordering::Relaxed,
|
||||
);
|
||||
}
|
||||
|
||||
fn record_write(&self, opened_at: Instant) {
|
||||
self.last_write_ms.store(
|
||||
u64::try_from(opened_at.elapsed().as_millis()).unwrap_or(u64::MAX),
|
||||
Ordering::Relaxed,
|
||||
);
|
||||
}
|
||||
|
||||
fn observe_command_queue(&self, sender: &mpsc::Sender<ConnectionCommand>) {
|
||||
let queued = OUTBOUND_COMMAND_BUFFER
|
||||
.saturating_sub(sender.capacity())
|
||||
@@ -400,10 +419,15 @@ impl ResponsesWebSocketConnection {
|
||||
}
|
||||
|
||||
pub(super) fn log_summary(&self, request_count: u64) {
|
||||
let connection_age_ms = self.age().as_millis();
|
||||
let read_idle_ms = connection_age_ms
|
||||
.saturating_sub(u128::from(self.stats.last_read_ms.load(Ordering::Relaxed)));
|
||||
let write_idle_ms = connection_age_ms
|
||||
.saturating_sub(u128::from(self.stats.last_write_ms.load(Ordering::Relaxed)));
|
||||
tracing::info!(
|
||||
websocket_connection_id = %self.connection_id,
|
||||
request_count,
|
||||
connection_age_ms = self.age().as_millis(),
|
||||
connection_age_ms,
|
||||
pump_exit_reason = self
|
||||
.exit_reason
|
||||
.map_or("unknown", PumpExitReason::as_str),
|
||||
@@ -417,6 +441,10 @@ impl ResponsesWebSocketConnection {
|
||||
.load(Ordering::Relaxed),
|
||||
ping_received_count = self.stats.ping_received_count.load(Ordering::Relaxed),
|
||||
pong_written_count = self.stats.pong_written_count.load(Ordering::Relaxed),
|
||||
ping_written_count = self.stats.ping_written_count.load(Ordering::Relaxed),
|
||||
pong_received_count = self.stats.pong_received_count.load(Ordering::Relaxed),
|
||||
read_idle_ms,
|
||||
write_idle_ms,
|
||||
"Responses WebSocket disconnected"
|
||||
);
|
||||
}
|
||||
@@ -493,6 +521,12 @@ async fn run_pump<S, E>(
|
||||
let deadline = tokio::time::sleep_until(opened_at + config.max_age);
|
||||
tokio::pin!(deadline);
|
||||
let mut deadline_elapsed = false;
|
||||
let mut heartbeat = tokio::time::interval_at(
|
||||
opened_at + DOWNSTREAM_PING_INTERVAL,
|
||||
DOWNSTREAM_PING_INTERVAL,
|
||||
);
|
||||
heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
||||
let mut ping_sequence = 0_u64;
|
||||
let reason = loop {
|
||||
tokio::select! {
|
||||
biased;
|
||||
@@ -531,6 +565,9 @@ async fn run_pump<S, E>(
|
||||
ConnectionWriteError::Transport { message }
|
||||
});
|
||||
let failed = result.is_err();
|
||||
if !failed {
|
||||
stats.record_write(opened_at);
|
||||
}
|
||||
let _ = acknowledged.send(result);
|
||||
if failed {
|
||||
break PumpExitReason::WriteError;
|
||||
@@ -539,45 +576,79 @@ async fn run_pump<S, E>(
|
||||
break PumpExitReason::ServerClose;
|
||||
}
|
||||
}
|
||||
message = socket.next() => match message {
|
||||
Some(Ok(Message::Text(payload))) => {
|
||||
if let Err(reason) = emit_incoming(&incoming, ConnectionEvent::Text(payload.to_string())) {
|
||||
break reason;
|
||||
// 官方 Codex WsStream 在独立 pump 中回复 Pong,且不将控制帧交给业务流。
|
||||
// 这里只为下游链路保活,不设置 Pong deadline,也不延长业务请求预算。
|
||||
_ = heartbeat.tick() => {
|
||||
ping_sequence = ping_sequence.wrapping_add(1);
|
||||
let ping = Message::Ping(ping_sequence.to_be_bytes().to_vec().into());
|
||||
let result = tokio::select! {
|
||||
biased;
|
||||
() = cancellation.cancelled() => break PumpExitReason::LifecycleShutdown,
|
||||
result = timeout(config.write_timeout, socket.send(ping)) => result,
|
||||
};
|
||||
match result {
|
||||
Ok(Ok(())) => {
|
||||
stats.ping_written_count.fetch_add(1, Ordering::Relaxed);
|
||||
stats.record_write(opened_at);
|
||||
}
|
||||
}
|
||||
Some(Ok(Message::Binary(_))) => {
|
||||
if let Err(reason) = emit_incoming(&incoming, ConnectionEvent::Binary) {
|
||||
break reason;
|
||||
Ok(Err(error)) => {
|
||||
tracing::info!(
|
||||
websocket_connection_id = %connection_id,
|
||||
error = %error,
|
||||
"Responses WebSocket Ping write failed"
|
||||
);
|
||||
break PumpExitReason::WriteError;
|
||||
}
|
||||
Err(_) => break PumpExitReason::WriteTimeout,
|
||||
}
|
||||
Some(Ok(Message::Ping(payload))) => {
|
||||
stats.ping_received_count.fetch_add(1, Ordering::Relaxed);
|
||||
match timeout(config.write_timeout, socket.send(Message::Pong(payload))).await {
|
||||
Ok(Ok(())) => {
|
||||
stats.pong_written_count.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
message = socket.next() => {
|
||||
if matches!(&message, Some(Ok(_))) {
|
||||
stats.record_read(opened_at);
|
||||
}
|
||||
match message {
|
||||
Some(Ok(Message::Text(payload))) => {
|
||||
if let Err(reason) = emit_incoming(&incoming, ConnectionEvent::Text(payload.to_string())) {
|
||||
break reason;
|
||||
}
|
||||
Ok(Err(error)) => {
|
||||
tracing::info!(
|
||||
websocket_connection_id = %connection_id,
|
||||
error = %error,
|
||||
"Responses WebSocket Pong write failed"
|
||||
);
|
||||
break PumpExitReason::WriteError;
|
||||
}
|
||||
Err(_) => break PumpExitReason::WriteTimeout,
|
||||
}
|
||||
Some(Ok(Message::Binary(_))) => {
|
||||
if let Err(reason) = emit_incoming(&incoming, ConnectionEvent::Binary) {
|
||||
break reason;
|
||||
}
|
||||
}
|
||||
Some(Ok(Message::Ping(payload))) => {
|
||||
stats.ping_received_count.fetch_add(1, Ordering::Relaxed);
|
||||
match timeout(config.write_timeout, socket.send(Message::Pong(payload))).await {
|
||||
Ok(Ok(())) => {
|
||||
stats.pong_written_count.fetch_add(1, Ordering::Relaxed);
|
||||
stats.record_write(opened_at);
|
||||
}
|
||||
Ok(Err(error)) => {
|
||||
tracing::info!(
|
||||
websocket_connection_id = %connection_id,
|
||||
error = %error,
|
||||
"Responses WebSocket Pong write failed"
|
||||
);
|
||||
break PumpExitReason::WriteError;
|
||||
}
|
||||
Err(_) => break PumpExitReason::WriteTimeout,
|
||||
}
|
||||
}
|
||||
Some(Ok(Message::Pong(_))) => {
|
||||
stats.pong_received_count.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
Some(Ok(Message::Close(_))) => break PumpExitReason::ClientClose,
|
||||
Some(Err(error)) => {
|
||||
tracing::info!(
|
||||
websocket_connection_id = %connection_id,
|
||||
error = %error,
|
||||
"Responses WebSocket receive failed"
|
||||
);
|
||||
break PumpExitReason::ReadError;
|
||||
}
|
||||
None => break PumpExitReason::PeerEof,
|
||||
}
|
||||
Some(Ok(Message::Pong(_))) => {}
|
||||
Some(Ok(Message::Close(_))) => break PumpExitReason::ClientClose,
|
||||
Some(Err(error)) => {
|
||||
tracing::info!(
|
||||
websocket_connection_id = %connection_id,
|
||||
error = %error,
|
||||
"Responses WebSocket receive failed"
|
||||
);
|
||||
break PumpExitReason::ReadError;
|
||||
}
|
||||
None => break PumpExitReason::PeerEof,
|
||||
},
|
||||
}
|
||||
};
|
||||
|
||||
@@ -247,7 +247,7 @@ async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocke
|
||||
}
|
||||
};
|
||||
started.session.trace().record("client.connection", serde_json::json!({
|
||||
"transport": "websocket", "connectionId": connection.id(), "correlationId": correlation_id,
|
||||
"transport": "websocket", "connectionId": connection.id(), "correlationId": correlation_id.as_ref(),
|
||||
"requestIndex": request_count,
|
||||
}));
|
||||
started
|
||||
|
||||
@@ -14,8 +14,8 @@ use gateway_core::operation::{Operation, RawJsonPayload, StandaloneSearchRequest
|
||||
use crate::ApiState;
|
||||
use crate::openai::{
|
||||
auth::{authenticate_client, client_access_error_response},
|
||||
endpoint::collect_raw_json_response,
|
||||
error::gateway_error_response,
|
||||
provider_endpoint::collect_raw_json_response,
|
||||
responses::{OpenAiRequestHeaders, request_client_context},
|
||||
};
|
||||
|
||||
|
||||
@@ -919,6 +919,22 @@ async fn usage_route_should_expose_table_facts_without_detail_payload() {
|
||||
compact: true,
|
||||
started_at: Utc::now(),
|
||||
});
|
||||
{
|
||||
let mut records = fixture.usage_records.lock().expect("usage records");
|
||||
for source in ["calculated", "provider_reported"] {
|
||||
let mut image = records[0].clone();
|
||||
image.id = format!("image_{source}");
|
||||
image.endpoint = "/v1/images/generations".to_owned();
|
||||
image.provider_kind = Some("openai".to_owned());
|
||||
image.requested_model_id = Some("gpt-image-2".to_owned());
|
||||
image.upstream_model_id = None;
|
||||
image.billing = Some(UsageBilling::Total {
|
||||
source: source.to_owned(),
|
||||
total: usd("0.00696"),
|
||||
});
|
||||
records.push(image);
|
||||
}
|
||||
}
|
||||
let response = observability::router::<AdminTestState>()
|
||||
.with_state(fixture.state())
|
||||
.oneshot(
|
||||
@@ -1016,4 +1032,12 @@ async fn usage_route_should_expose_table_facts_without_detail_payload() {
|
||||
})
|
||||
);
|
||||
assert!(value["data"]["items"][0].get("metadata").is_none());
|
||||
assert_eq!(
|
||||
value["data"]["items"][1]["billing"]["totalAmountDisplay"],
|
||||
"≈ $0.007"
|
||||
);
|
||||
assert_eq!(
|
||||
value["data"]["items"][2]["billing"]["totalAmountDisplay"],
|
||||
"$0.007"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -61,7 +61,7 @@ fn source_tree_should_match_frozen_machine_manifest() {
|
||||
"src/openai/images/mod.rs",
|
||||
"src/openai/mod.rs",
|
||||
"src/openai/models.rs",
|
||||
"src/openai/provider_endpoint.rs",
|
||||
"src/openai/endpoint.rs",
|
||||
"src/openai/responses/error.rs",
|
||||
"src/openai/responses/http.rs",
|
||||
"src/openai/responses/mod.rs",
|
||||
@@ -107,7 +107,7 @@ fn test_tree_should_match_frozen_rust_mirror() {
|
||||
"tests/openai/images/mod.rs",
|
||||
"tests/openai/mod.rs",
|
||||
"tests/openai/models.rs",
|
||||
"tests/openai/provider_endpoint.rs",
|
||||
"tests/openai/endpoint.rs",
|
||||
"tests/openai/responses/http.rs",
|
||||
"tests/openai/responses/mod.rs",
|
||||
"tests/openai/responses/websocket/connection.rs",
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
mod auth;
|
||||
mod endpoint;
|
||||
mod error;
|
||||
mod images;
|
||||
mod models;
|
||||
mod provider_endpoint;
|
||||
mod responses;
|
||||
mod router;
|
||||
mod search;
|
||||
|
||||
@@ -444,21 +444,6 @@ async fn inbound_overload_should_close_without_executing_queued_frames() {
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn pump_does_not_emit_an_active_ping() {
|
||||
let PumpHarness {
|
||||
connection,
|
||||
incoming: _incoming,
|
||||
mut written,
|
||||
..
|
||||
} = test_connection(false);
|
||||
tokio::time::advance(Duration::from_secs(5 * 60)).await;
|
||||
tokio::task::yield_now().await;
|
||||
|
||||
assert!(matches!(written.try_recv(), Err(TryRecvError::Empty)));
|
||||
drop(connection);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn outbound_commands_are_written_in_acknowledged_order() {
|
||||
let PumpHarness {
|
||||
@@ -582,13 +567,17 @@ async fn idle_connection_reaches_the_official_limit_without_starting_an_executio
|
||||
|
||||
tokio::time::advance(Duration::from_secs(60 * 60)).await;
|
||||
tokio::task::yield_now().await;
|
||||
let message = tokio::time::timeout(Duration::from_secs(1), socket.next())
|
||||
.await
|
||||
.expect("connection limit response timeout")
|
||||
.expect("connection remains available for the limit error")
|
||||
.expect("valid WebSocket frame");
|
||||
let ClientMessage::Text(text) = message else {
|
||||
panic!("connection limit must be a text error event");
|
||||
let text = loop {
|
||||
let message = tokio::time::timeout(Duration::from_secs(1), socket.next())
|
||||
.await
|
||||
.expect("connection limit response timeout")
|
||||
.expect("connection remains available for the limit error")
|
||||
.expect("valid WebSocket frame");
|
||||
match message {
|
||||
ClientMessage::Text(text) => break text,
|
||||
ClientMessage::Ping(_) => {}
|
||||
other => panic!("unexpected frame before connection limit: {other:?}"),
|
||||
}
|
||||
};
|
||||
let value = serde_json::from_str::<Value>(&text).expect("connection limit JSON");
|
||||
|
||||
@@ -606,7 +595,8 @@ async fn idle_connection_reaches_the_official_limit_without_starting_an_executio
|
||||
);
|
||||
assert_eq!(trace.starts.load(Ordering::Acquire), 0);
|
||||
|
||||
socket.close(None).await.expect("close WebSocket");
|
||||
// 服务端已因生命周期到期关闭;不再向关闭中的连接写入。
|
||||
drop(socket);
|
||||
server.abort();
|
||||
}
|
||||
|
||||
@@ -949,3 +939,134 @@ async fn client_close_should_cancel_active_execution_without_starting_queued_req
|
||||
assert!(trace.cancelled.load(Ordering::Acquire));
|
||||
server.abort();
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn idle_pump_sends_heartbeat_without_bursting_after_a_delay() {
|
||||
let PumpHarness {
|
||||
connection,
|
||||
incoming: _incoming,
|
||||
mut written,
|
||||
..
|
||||
} = test_connection(false);
|
||||
tokio::task::yield_now().await;
|
||||
assert!(matches!(written.try_recv(), Err(TryRecvError::Empty)));
|
||||
|
||||
tokio::time::advance(Duration::from_secs(25)).await;
|
||||
let first = written.recv().await.expect("first heartbeat");
|
||||
assert!(matches!(first, Message::Ping(_)));
|
||||
|
||||
tokio::time::advance(Duration::from_secs(120)).await;
|
||||
let second = written.recv().await.expect("heartbeat after delay");
|
||||
assert!(matches!(&second, Message::Ping(_)));
|
||||
assert_ne!(first, second);
|
||||
tokio::task::yield_now().await;
|
||||
assert!(matches!(written.try_recv(), Err(TryRecvError::Empty)));
|
||||
drop(connection);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn missing_pongs_do_not_cancel_the_connection_or_block_business_frames() {
|
||||
let PumpHarness {
|
||||
mut connection,
|
||||
incoming,
|
||||
mut written,
|
||||
..
|
||||
} = test_connection(false);
|
||||
for _ in 0..13 {
|
||||
tokio::time::advance(Duration::from_secs(25)).await;
|
||||
assert!(matches!(written.recv().await, Some(Message::Ping(_))));
|
||||
}
|
||||
|
||||
incoming
|
||||
.send(Ok(Message::Text("still connected".into())))
|
||||
.expect("send business frame after 325 seconds without Pong");
|
||||
assert!(matches!(
|
||||
connection.next_event().await,
|
||||
Some(ConnectionEvent::Text(text)) if text == "still connected"
|
||||
));
|
||||
incoming.send(Ok(Message::Close(None))).expect("close peer");
|
||||
assert_eq!(
|
||||
connection.wait_for_exit().await,
|
||||
PumpExitReason::ClientClose
|
||||
);
|
||||
tokio::time::advance(Duration::from_secs(25)).await;
|
||||
assert!(written.recv().await.is_none());
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn stalled_heartbeat_can_be_cancelled_without_waiting_for_write_timeout() {
|
||||
let PumpHarness {
|
||||
mut connection,
|
||||
incoming: _incoming,
|
||||
written: _written,
|
||||
dropped,
|
||||
cancellation,
|
||||
} = test_connection(true);
|
||||
tokio::time::advance(Duration::from_secs(25)).await;
|
||||
tokio::task::yield_now().await;
|
||||
cancellation.cancel();
|
||||
let reason = tokio::time::timeout(Duration::from_secs(1), connection.wait_for_exit())
|
||||
.await
|
||||
.expect("cancel a blocked heartbeat promptly");
|
||||
assert_eq!(reason, PumpExitReason::LifecycleShutdown);
|
||||
assert!(dropped.load(Ordering::Acquire));
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn stalled_heartbeat_obeys_the_existing_write_timeout() {
|
||||
let PumpHarness {
|
||||
mut connection,
|
||||
incoming: _incoming,
|
||||
written: _written,
|
||||
dropped,
|
||||
..
|
||||
} = test_connection(true);
|
||||
let started = tokio::time::Instant::now();
|
||||
assert_eq!(
|
||||
connection.wait_for_exit().await,
|
||||
PumpExitReason::WriteTimeout
|
||||
);
|
||||
assert_eq!(started.elapsed(), Duration::from_secs(25 + 300));
|
||||
assert!(dropped.load(Ordering::Acquire));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn active_response_receives_heartbeats_during_two_hundred_seconds_of_silence() {
|
||||
let (trace, mut socket, server) = start_active_response().await;
|
||||
for _ in 0..8 {
|
||||
tokio::time::pause();
|
||||
tokio::time::advance(Duration::from_secs(25)).await;
|
||||
tokio::task::yield_now().await;
|
||||
tokio::time::resume();
|
||||
let message = tokio::time::timeout(Duration::from_secs(2), socket.next())
|
||||
.await
|
||||
.expect("heartbeat while the upstream response is pending")
|
||||
.expect("connection remains open")
|
||||
.expect("valid heartbeat frame");
|
||||
let ClientMessage::Ping(payload) = message else {
|
||||
panic!("expected heartbeat while business stream is silent: {message:?}");
|
||||
};
|
||||
// 与官方 WsStream 一样,在业务响应未完成时回复同 payload 的 Pong。
|
||||
socket
|
||||
.send(ClientMessage::Pong(payload))
|
||||
.await
|
||||
.expect("reply to heartbeat");
|
||||
}
|
||||
assert_eq!(trace.starts.load(Ordering::Acquire), 1);
|
||||
assert!(!trace.cancelled.load(Ordering::Acquire));
|
||||
|
||||
trace.release_terminal.notify_one();
|
||||
let message = tokio::time::timeout(Duration::from_secs(2), socket.next())
|
||||
.await
|
||||
.expect("terminal response after heartbeats")
|
||||
.expect("connection remains open")
|
||||
.expect("valid terminal frame");
|
||||
let ClientMessage::Text(text) = message else {
|
||||
panic!("expected completed response: {message:?}");
|
||||
};
|
||||
let value: Value = serde_json::from_str(&text).expect("terminal JSON");
|
||||
assert_eq!(value["type"], "response.completed");
|
||||
assert_eq!(trace.starts.load(Ordering::Acquire), 1);
|
||||
socket.close(None).await.expect("close WebSocket");
|
||||
server.abort();
|
||||
}
|
||||
|
||||
@@ -22,7 +22,7 @@ use gateway_core::upstream::UpstreamSendState;
|
||||
use serde_json::{Value, json};
|
||||
use tower::ServiceExt;
|
||||
|
||||
use super::provider_endpoint::BufferedJsonSession;
|
||||
use super::endpoint::BufferedJsonSession;
|
||||
use super::{api_router, authenticated_client};
|
||||
|
||||
const SEARCH_RESPONSE: &[u8] = br#"{ "encrypted_output":"ciphertext", "output":"search result", "results":[{"type":"text_result","ref_id":"turn0search0","future":9007199254740993}] }"#;
|
||||
|
||||
@@ -524,7 +524,7 @@ impl DefaultExecutionService {
|
||||
operation: operation.kind(),
|
||||
endpoint: metadata.endpoint,
|
||||
client_transport: metadata.transport.as_str().to_owned(),
|
||||
requested_model: target.into_public_model(),
|
||||
requested_model: target.into_public_model().or(observation.requested_model),
|
||||
client_ip: metadata.client_ip,
|
||||
user_agent: metadata.user_agent,
|
||||
reasoning_effort: observation.reasoning_effort,
|
||||
|
||||
@@ -20,7 +20,9 @@ use crate::event::{EventSequenceValidator, ProviderEvent};
|
||||
use crate::identity::ProviderKind;
|
||||
use crate::operation::Operation;
|
||||
use crate::policy::ClientApiKeyId;
|
||||
use crate::routing::{ModelCapabilities, ModelPresentation, ProviderCandidate, UpstreamModelId};
|
||||
use crate::routing::{
|
||||
ModelCapabilities, ModelPresentation, ProviderCandidate, PublicModelId, UpstreamModelId,
|
||||
};
|
||||
use crate::upstream::OpaqueUpstreamValue;
|
||||
use crate::upstream::{UpstreamSendState, UpstreamTransport};
|
||||
|
||||
@@ -416,6 +418,8 @@ pub struct ProviderRequest {
|
||||
/// Provider 对公共观测表可解释的请求语义;未知字段保持空值。
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||
pub struct ProviderRequestObservation {
|
||||
/// 独立 Provider 端点声明的请求模型,仅用于观测,不参与文本模型目录路由。
|
||||
pub requested_model: Option<PublicModelId>,
|
||||
/// 客户端原始请求中的推理强度。
|
||||
pub reasoning_effort: Option<String>,
|
||||
pub reasoning_preset: Option<String>,
|
||||
|
||||
@@ -26,7 +26,7 @@ use gateway_core::engine::execution::{
|
||||
use gateway_core::engine::probe::{AccountProbe, AccountProbeErrorSource, AccountProbeRequest};
|
||||
use gateway_core::engine::provider::{
|
||||
Provider, ProviderCallMetadata, ProviderCatalogGeneration, ProviderModelCapabilities,
|
||||
ProviderRegistry, ProviderRequest, ProviderStream,
|
||||
ProviderRegistry, ProviderRequest, ProviderRequestObservation, ProviderStream,
|
||||
};
|
||||
use gateway_core::engine::{
|
||||
AttemptContext, AttemptRecord, ExecutionStore, IntermediateFailure, ModelRequestFinalization,
|
||||
@@ -260,7 +260,9 @@ impl Provider for LocalFailingProvider {
|
||||
}
|
||||
}
|
||||
|
||||
struct ColdFailingProvider;
|
||||
struct ColdFailingProvider {
|
||||
requested_model: Option<PublicModelId>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Provider for ColdFailingProvider {
|
||||
@@ -272,6 +274,13 @@ impl Provider for ColdFailingProvider {
|
||||
ProviderCatalogGeneration::default()
|
||||
}
|
||||
|
||||
fn request_observation(&self, _: &Operation, _: &ClientApiKeyId) -> ProviderRequestObservation {
|
||||
ProviderRequestObservation {
|
||||
requested_model: self.requested_model.clone(),
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
async fn query_model_capabilities(
|
||||
&self,
|
||||
) -> Result<Vec<ProviderModelCapabilities>, ProviderError> {
|
||||
@@ -323,12 +332,19 @@ fn successful_authentication_should_record_client_key_usage() {
|
||||
|
||||
#[test]
|
||||
fn provider_endpoint_should_persist_its_real_v1_endpoint() {
|
||||
assert_provider_endpoint_observation(None);
|
||||
assert_provider_endpoint_observation(Some("gpt-image-2"));
|
||||
}
|
||||
|
||||
fn assert_provider_endpoint_observation(model: Option<&str>) {
|
||||
let store = Arc::new(TrackingExecutionStore::default());
|
||||
let service = DefaultExecutionService::new(
|
||||
RuntimeSnapshotHandle::new(client_snapshot()),
|
||||
store.clone(),
|
||||
ProviderRegistry::new([Arc::new(ColdFailingProvider) as Arc<dyn Provider>])
|
||||
.expect("provider registry"),
|
||||
ProviderRegistry::new([Arc::new(ColdFailingProvider {
|
||||
requested_model: model.map(|model| PublicModelId::new(model).expect("model")),
|
||||
}) as Arc<dyn Provider>])
|
||||
.expect("provider registry"),
|
||||
Arc::new(UnusedAdmissions),
|
||||
Arc::new(UnusedCircuits),
|
||||
Arc::new(UnusedContinuation),
|
||||
@@ -369,7 +385,7 @@ fn provider_endpoint_should_persist_its_real_v1_endpoint() {
|
||||
vec!["/v1/images/generations".to_owned()],
|
||||
"execution error: {error:?}"
|
||||
);
|
||||
assert_eq!(store.requested_models(), vec![None]);
|
||||
assert_eq!(store.requested_models(), vec![model.map(str::to_owned)]);
|
||||
let upstream_models = store.upstream_models();
|
||||
assert!(!upstream_models.is_empty());
|
||||
assert!(upstream_models.iter().all(Option::is_none));
|
||||
|
||||
@@ -13,12 +13,12 @@ const REQUEST_ERROR_SELECT: &str = "select 'model_request'::text as source,
|
||||
mr.provider_account_authentication_kind_snapshot
|
||||
as provider_account_authentication_kind,
|
||||
mr.upstream_model_id, mr.upstream_transport,
|
||||
coalesce(mr.error_kind, 'failed') as failure_kind,
|
||||
mr.error_kind as failure_kind,
|
||||
mr.upstream_send_state,
|
||||
mr.client_status_code, mr.upstream_status_code,
|
||||
mr.provider_error_code, mr.client_response_id, mr.upstream_request_id,
|
||||
mr.latency_ms,
|
||||
coalesce(mr.error_message, mr.error_kind, 'request failed') as message,
|
||||
coalesce(mr.error_message, mr.error_kind) as message,
|
||||
mr.raw_upstream_error,
|
||||
host(mr.client_ip) as client_ip, mr.user_agent,
|
||||
mr.reasoning_effort, mr.reasoning_preset, mr.request_kind,
|
||||
@@ -32,7 +32,7 @@ const REQUEST_ERROR_SELECT: &str = "select 'model_request'::text as source,
|
||||
mr.completed_at as occurred_at,
|
||||
'model_request:' || mr.id as stable_sort_id
|
||||
from model_requests mr
|
||||
where mr.outcome = 'failed'";
|
||||
where true";
|
||||
|
||||
const OPS_EVENT_SELECT: &str = "select 'ops_event'::text as source,
|
||||
oe.id as event_id, oe.model_request_id as request_id, oe.attempt_index,
|
||||
@@ -106,7 +106,7 @@ pub(crate) async fn count_ops_errors(
|
||||
filter: &OpsErrorFilter,
|
||||
) -> StoreResult<u64> {
|
||||
let mut statement = QueryBuilder::<Postgres>::new(
|
||||
"select coalesce(sum(source_count), 0)::bigint from (select count(*)::bigint as source_count from model_requests mr where mr.outcome = 'failed'",
|
||||
"select coalesce(sum(source_count), 0)::bigint from (select count(*)::bigint as source_count from model_requests mr where true",
|
||||
);
|
||||
push_request_error_predicates(&mut statement, range, filter);
|
||||
statement.push(
|
||||
@@ -127,6 +127,9 @@ fn push_request_error_predicates(
|
||||
range: ObservabilityRange,
|
||||
filter: &OpsErrorFilter,
|
||||
) {
|
||||
// 错误事实独立于请求结束状态;主动取消不属于需要排查的错误。
|
||||
// 列表和总数共用此条件,避免流式响应中的错误因 outcome 被漏掉。
|
||||
statement.push(" and mr.error_kind is not null and mr.error_kind <> 'cancelled'");
|
||||
push_range(statement, "mr.completed_at", range);
|
||||
for (column, value) in [
|
||||
("mr.client_api_key_ref", &filter.client_api_key_ref),
|
||||
|
||||
@@ -258,6 +258,119 @@ async fn ops_search_should_treat_sql_wildcards_as_literals() {
|
||||
database.close().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ops_should_include_incomplete_upstream_errors() {
|
||||
let Some(database) = TestDatabase::create("ops_incomplete_error").await else {
|
||||
return;
|
||||
};
|
||||
let now = Utc::now();
|
||||
seed_observability_facts(&database.pool, now).await.unwrap();
|
||||
let raw_error = r#"{"type":"error","status":400,"error":{"type":"invalid_request_error","message":"Invalid `previous_response_id`."}}"#;
|
||||
sqlx::query(
|
||||
"update model_requests
|
||||
set outcome = 'incomplete', error_kind = 'invalid_request',
|
||||
client_transport = 'websocket', upstream_transport = 'websocket',
|
||||
downstream_committed_at = completed_at,
|
||||
client_status_code = null, upstream_status_code = 400,
|
||||
provider_error_code = null, raw_upstream_error = $1
|
||||
where id = 'req_observe_failed'",
|
||||
)
|
||||
.bind(raw_error)
|
||||
.execute(&database.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let page = admin_observability_store(&database.pool)
|
||||
.list_ops_errors(admin_observability::OpsErrorQuery {
|
||||
range: admin_observability::TimeRange::new(
|
||||
now - TimeDelta::hours(1),
|
||||
now + TimeDelta::hours(1),
|
||||
)
|
||||
.unwrap(),
|
||||
filter: admin_observability::OpsErrorFilter {
|
||||
request_id: Some("req_observe_failed".to_owned()),
|
||||
status_code: Some(400),
|
||||
..admin_observability::OpsErrorFilter::default()
|
||||
},
|
||||
current_page: 1,
|
||||
page_size: PageSize::new(10).unwrap(),
|
||||
})
|
||||
.await
|
||||
.expect("incomplete request must be available in admin error troubleshooting");
|
||||
database.close().await;
|
||||
|
||||
assert_eq!(page.total, 1);
|
||||
assert_eq!(page.items.len(), 1);
|
||||
let error = &page.items[0];
|
||||
assert_eq!(error.source, "model_request");
|
||||
assert_eq!(error.failure_kind, "invalid_request");
|
||||
assert_eq!(error.upstream_status_code, Some(400));
|
||||
assert_eq!(error.client_status_code, None);
|
||||
assert_eq!(error.raw_upstream_error.as_deref(), Some(raw_error));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ops_should_select_errors_independently_of_request_outcome() {
|
||||
let Some(database) = TestDatabase::create("ops_error_outcomes").await else {
|
||||
return;
|
||||
};
|
||||
let now = Utc::now();
|
||||
seed_observability_facts(&database.pool, now).await.unwrap();
|
||||
let range =
|
||||
ObservabilityRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)).unwrap();
|
||||
let repository = observability_repository(&database.pool);
|
||||
let mut results = Vec::new();
|
||||
for (outcome, error_kind, expected_count) in [
|
||||
("incomplete", Some("upstream_unavailable"), 1),
|
||||
("succeeded", Some("upstream_unavailable"), 1),
|
||||
("failed", Some("upstream_unavailable"), 1),
|
||||
("failed", None, 0),
|
||||
("incomplete", None, 0),
|
||||
("cancelled", Some("cancelled"), 0),
|
||||
("incomplete", Some("cancelled"), 0),
|
||||
("failed", Some("cancelled"), 0),
|
||||
("succeeded", None, 0),
|
||||
] {
|
||||
sqlx::query(
|
||||
"update model_requests set outcome = $1, error_kind = $2
|
||||
where id = 'req_observe_success'",
|
||||
)
|
||||
.bind(outcome)
|
||||
.bind(error_kind)
|
||||
.execute(&database.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let page = repository
|
||||
.list_ops_errors(OpsErrorQuery {
|
||||
range,
|
||||
filter: OpsErrorFilter {
|
||||
request_id: Some("req_observe_success".to_owned()),
|
||||
..OpsErrorFilter::default()
|
||||
},
|
||||
current_page: 1,
|
||||
page_size: ObservabilityPageSize::new(10).unwrap(),
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
results.push((outcome, error_kind, expected_count, page));
|
||||
}
|
||||
database.close().await;
|
||||
|
||||
for (outcome, error_kind, expected_count, page) in results {
|
||||
assert_eq!(
|
||||
page.total, expected_count,
|
||||
"total for outcome={outcome}, error_kind={error_kind:?}"
|
||||
);
|
||||
assert_eq!(
|
||||
page.items.len() as u64,
|
||||
expected_count,
|
||||
"items for outcome={outcome}, error_kind={error_kind:?}"
|
||||
);
|
||||
if let Some(error) = page.items.first() {
|
||||
assert_eq!(Some(error.failure_kind.as_str()), error_kind);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn recovered_continuation_failure_should_be_visible_in_ops_but_hidden_from_business_metrics()
|
||||
{
|
||||
|
||||
@@ -23,8 +23,8 @@ getrandom = "0.4.3"
|
||||
hex = "0.4.3"
|
||||
# TLS 依赖按官方 Core 锁文件固定,包含用于约束传递依赖的直接声明。
|
||||
# 升级前须重复采样 ClientHello,核对算法、扩展集合及排序约束、key_share 与 ALPN。
|
||||
aws-lc-rs = "=1.16.2"
|
||||
hyper = { version = "=1.8.1", default-features = false }
|
||||
aws-lc-rs = "=1.18.1"
|
||||
hyper = { version = "=1.11.1", default-features = false }
|
||||
hyper-rustls = "=0.27.7"
|
||||
# 与官方 Codex Desktop 一致:reqwest 不启用压缩特性,上游请求不带 accept-encoding。
|
||||
reqwest = { version = "=0.12.28", default-features = false, features = ["json", "stream", "cookies", "rustls-tls-native-roots", "http2"] }
|
||||
@@ -33,15 +33,15 @@ semver = "1.0.27"
|
||||
# 与官方 rustls 路径一致:aws-lc 支持 P-521,优先提供 X25519MLKEM768。
|
||||
# reqwest 的 roots feature 仍会编译 ring;实际 provider 由 TLS 初始化显式选择。
|
||||
rustls = { version = "=0.23.36", default-features = false, features = ["aws_lc_rs", "prefer-post-quantum", "std", "tls12"] }
|
||||
rustls-native-certs = "=0.8.3"
|
||||
rustls-pki-types = "=1.14.0"
|
||||
rustls-native-certs = "=0.8.4"
|
||||
rustls-pki-types = "=1.15.1"
|
||||
secrecy = "0.10.3"
|
||||
serde = { version = "1.0.228", features = ["derive"] }
|
||||
serde_json = { version = "1.0.150", features = ["arbitrary_precision", "preserve_order"] }
|
||||
sha2 = "0.11.0"
|
||||
subtle = "2.6.1"
|
||||
thiserror = "2.0.18"
|
||||
tokio = { version = "=1.52.3", default-features = false, features = ["fs", "macros", "rt", "sync", "time"] }
|
||||
tokio = { version = "=1.53.1", default-features = false, features = ["fs", "macros", "rt", "sync", "time"] }
|
||||
tokio-rustls = { version = "=0.26.4", default-features = false }
|
||||
tokio-util = { version = "0.7.18", features = ["rt"] }
|
||||
# 官方 Codex WebSocket 需要上游尚未发布的 proxy 与 permessage-deflate 实现,固定到已审计提交。
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
//! OpenAI attempt 的选择、发送与响应流执行。
|
||||
|
||||
use gateway_core::metering::{CalculatedCost, Usage};
|
||||
|
||||
use super::*;
|
||||
|
||||
impl CodexProvider {
|
||||
@@ -490,6 +492,14 @@ pub(super) fn cold_json_response_stream(request: ColdJsonResponse) -> EventStrea
|
||||
let response_meta =
|
||||
ResponseMeta::for_provider_endpoint(request.context.request_id().as_str());
|
||||
yield ProviderEvent::canonical(GatewayEvent::Started(response_meta.clone()));
|
||||
if matches!(request.endpoint_path, CODEX_IMAGE_GENERATIONS_PATH | CODEX_IMAGE_EDITS_PATH)
|
||||
&& let Some((usage, cost)) = image_response_metering(&request.body, &response.body)
|
||||
{
|
||||
yield ProviderEvent::canonical(GatewayEvent::Usage(usage));
|
||||
if let Some(cost) = cost {
|
||||
yield ProviderEvent::canonical(GatewayEvent::CalculatedCost(cost));
|
||||
}
|
||||
}
|
||||
let wire = ProtocolWireEvent::raw_json(PROVIDER_NAME, response.body).map_err(|_| {
|
||||
provider_error(ProviderErrorKind::Protocol, UpstreamSendState::Sent)
|
||||
})?;
|
||||
@@ -500,6 +510,37 @@ pub(super) fn cold_json_response_stream(request: ColdJsonResponse) -> EventStrea
|
||||
})
|
||||
}
|
||||
|
||||
fn image_response_metering(
|
||||
request_body: &[u8],
|
||||
body: &[u8],
|
||||
) -> Option<(Usage, Option<CalculatedCost>)> {
|
||||
// 只保留 usage,跳过通常很大的 base64 图片;原始响应仍按字节透传。
|
||||
#[derive(Deserialize)]
|
||||
struct ImageUsageEnvelope {
|
||||
usage: Option<Value>,
|
||||
}
|
||||
|
||||
let raw = serde_json::from_slice::<ImageUsageEnvelope>(body)
|
||||
.ok()?
|
||||
.usage?;
|
||||
let mut usage = Usage::new();
|
||||
usage.input_tokens = raw.get("input_tokens").and_then(Value::as_u64);
|
||||
usage.output_tokens = raw.get("output_tokens").and_then(Value::as_u64);
|
||||
usage.cached_tokens = raw
|
||||
.pointer("/input_tokens_details/cached_tokens")
|
||||
.and_then(Value::as_u64);
|
||||
usage.image_input_tokens = raw
|
||||
.pointer("/input_tokens_details/image_tokens")
|
||||
.and_then(Value::as_u64);
|
||||
usage.image_output_tokens = raw
|
||||
.pointer("/output_tokens_details/image_tokens")
|
||||
.and_then(Value::as_u64);
|
||||
// 总量是上游独立报告的事实;图片明细是总输入/输出的子集,不能再次相加。
|
||||
usage.total_tokens = raw.get("total_tokens").and_then(Value::as_u64);
|
||||
let cost = crate::transport::usage::image_calculated_cost(request_body, &raw);
|
||||
(usage != Usage::default()).then_some((usage, cost))
|
||||
}
|
||||
|
||||
pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
|
||||
let ColdResponse {
|
||||
client,
|
||||
|
||||
@@ -237,6 +237,15 @@ impl Provider for CodexProvider {
|
||||
_ => None,
|
||||
};
|
||||
return ProviderRequestObservation {
|
||||
requested_model: match operation {
|
||||
Operation::Search(request) => {
|
||||
observation::endpoint_requested_model(request.payload())
|
||||
}
|
||||
Operation::GenerateImage(request) => {
|
||||
observation::endpoint_requested_model(request.payload())
|
||||
}
|
||||
_ => None,
|
||||
},
|
||||
continuation: ContinuationRequestObservation {
|
||||
affinity_hash: affinity.map(|affinity| affinity.persistence_hash().to_owned()),
|
||||
..Default::default()
|
||||
@@ -259,6 +268,7 @@ impl Provider for CodexProvider {
|
||||
requested: previous_response_id.is_some(),
|
||||
};
|
||||
ProviderRequestObservation {
|
||||
requested_model: None,
|
||||
reasoning_effort,
|
||||
reasoning_preset: semantics.reasoning_preset.map(str::to_owned),
|
||||
request_kind: semantics.request_kind,
|
||||
|
||||
@@ -2,6 +2,23 @@
|
||||
|
||||
use super::*;
|
||||
|
||||
pub(super) fn endpoint_requested_model(
|
||||
payload: &gateway_core::operation::RawJsonPayload,
|
||||
) -> Option<gateway_core::routing::PublicModelId> {
|
||||
if payload.protocol() != PROVIDER_NAME {
|
||||
return None;
|
||||
}
|
||||
// 只读取 model,避免把编辑请求中的整张图片复制到观测数据。
|
||||
#[derive(Deserialize)]
|
||||
struct RequestModel {
|
||||
model: Option<Value>,
|
||||
}
|
||||
let model = serde_json::from_slice::<RequestModel>(payload.body())
|
||||
.ok()?
|
||||
.model?;
|
||||
gateway_core::routing::PublicModelId::new(model.as_str()?.to_owned()).ok()
|
||||
}
|
||||
|
||||
/// OpenAI Responses 的观测状态完全归 Provider 所有。
|
||||
///
|
||||
/// 每次原始 SSE/WS 事件推进状态后重新生成不可变 observation,Core 只负责携带、
|
||||
|
||||
@@ -851,7 +851,9 @@ pub(super) fn websocket_success_decision(
|
||||
) -> CodexTransportDecision {
|
||||
match requirement {
|
||||
TransportRequirement::ExactWebSocketContinuation => CodexTransportDecision::ExactWebSocket,
|
||||
TransportRequirement::ExplicitWebSocketWarmup => CodexTransportDecision::RequiredWebSocket,
|
||||
TransportRequirement::ExplicitWebSocketWarmup | TransportRequirement::WebSocketNewChain => {
|
||||
CodexTransportDecision::RequiredWebSocket
|
||||
}
|
||||
_ if prepared.reused() => CodexTransportDecision::ReusedWebSocket,
|
||||
_ => CodexTransportDecision::ConnectedWebSocket,
|
||||
}
|
||||
|
||||
@@ -272,6 +272,7 @@ impl CodexBackendClient {
|
||||
Some(WEBSOCKET_FAST_PATH_BUDGET)
|
||||
}
|
||||
TransportRequirement::ExplicitWebSocketWarmup
|
||||
| TransportRequirement::WebSocketNewChain
|
||||
| TransportRequirement::ExactWebSocketContinuation
|
||||
| TransportRequirement::ExternalUnknown => None,
|
||||
TransportRequirement::HttpRequired => None,
|
||||
|
||||
@@ -121,6 +121,8 @@ pub enum TransportRequirement {
|
||||
HttpRequired,
|
||||
/// `generate=false + store=false` 预热必须保留在同一条 WebSocket。
|
||||
ExplicitWebSocketWarmup,
|
||||
/// 客户端 WebSocket 的非持久化新链,必须在池化连接上建立后续续接状态。
|
||||
WebSocketNewChain,
|
||||
/// 只能使用持有指定 connection-local response 的精确 WebSocket。
|
||||
ExactWebSocketContinuation,
|
||||
/// previous response 已持久化,允许 WebSocket 或 HTTP/2。
|
||||
@@ -136,7 +138,9 @@ impl TransportRequirement {
|
||||
pub fn requires_websocket(self) -> bool {
|
||||
matches!(
|
||||
self,
|
||||
Self::ExplicitWebSocketWarmup | Self::ExactWebSocketContinuation
|
||||
Self::ExplicitWebSocketWarmup
|
||||
| Self::WebSocketNewChain
|
||||
| Self::ExactWebSocketContinuation
|
||||
)
|
||||
}
|
||||
|
||||
@@ -150,7 +154,7 @@ impl TransportRequirement {
|
||||
|
||||
/// 无续接依赖的新请求可以在明确的连接级拒绝后重新建连。
|
||||
pub fn allows_connection_restart(self) -> bool {
|
||||
matches!(self, Self::NewChain)
|
||||
matches!(self, Self::NewChain | Self::WebSocketNewChain)
|
||||
}
|
||||
|
||||
/// 用于审计与遥测的稳定名称。
|
||||
@@ -158,6 +162,7 @@ impl TransportRequirement {
|
||||
match self {
|
||||
Self::HttpRequired => "http_required",
|
||||
Self::ExplicitWebSocketWarmup => "explicit_websocket_warmup",
|
||||
Self::WebSocketNewChain => "websocket_new_chain",
|
||||
Self::ExactWebSocketContinuation => "exact_websocket_continuation",
|
||||
Self::PersistedContinuation => "persisted_continuation",
|
||||
Self::ExternalUnknown => "external_unknown",
|
||||
@@ -184,6 +189,11 @@ pub fn transport_requirement(request: &CodexResponsesRequest) -> TransportRequir
|
||||
TransportRequirement::ExternalUnknown
|
||||
}
|
||||
},
|
||||
// 客户端会在下一轮提交 response ID;HTTP store=false 的成功响应无法
|
||||
// 在池化 WebSocket 上续接,所以首轮也不能按普通快路径降级到 HTTP。
|
||||
None if request.downstream_websocket_connection_id.is_some() && !request.store() => {
|
||||
TransportRequirement::WebSocketNewChain
|
||||
}
|
||||
None => TransportRequirement::NewChain,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,9 +1,12 @@
|
||||
//! 核心 Generate operation 到 Codex Responses wire request 的严格编码。
|
||||
|
||||
use std::io;
|
||||
|
||||
use base64::{Engine as _, engine::general_purpose::STANDARD};
|
||||
use gateway_core::operation::GenerateRequest;
|
||||
use gateway_protocol::openai::WS_REQUEST_HEADER_RESPONSES_LITE_CLIENT_METADATA_KEY;
|
||||
use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
|
||||
use serde::Serialize as _;
|
||||
use serde_json::{Map, Value};
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
@@ -447,7 +450,8 @@ pub(crate) fn scope_turn_metadata(
|
||||
changed |= metadata.remove(*key).is_some();
|
||||
}
|
||||
}
|
||||
if !cross_account
|
||||
if raw.is_ascii()
|
||||
&& !cross_account
|
||||
&& !changed
|
||||
&& !INSTALLATION_ID_KEYS
|
||||
.iter()
|
||||
@@ -460,7 +464,40 @@ pub(crate) fn scope_turn_metadata(
|
||||
metadata.insert((*key).to_owned(), Value::String(installation_id.to_owned()));
|
||||
}
|
||||
}
|
||||
serde_json::to_string(&metadata).ok()
|
||||
// Codex 的 turn metadata 同时承载于 HTTP header 与 WS client_metadata。
|
||||
// 改写安装 ID 后仍须保持官方 to_ascii_json_string 的编码合同;普通
|
||||
// to_string 会把中文工作区路径还原成 UTF-8,触发上游 WS metadata 后 Close 1000。
|
||||
let mut bytes = Vec::new();
|
||||
metadata
|
||||
.serialize(&mut serde_json::Serializer::with_formatter(
|
||||
&mut bytes,
|
||||
AsciiTurnMetadataFormatter,
|
||||
))
|
||||
.ok()?;
|
||||
String::from_utf8(bytes).ok()
|
||||
}
|
||||
|
||||
struct AsciiTurnMetadataFormatter;
|
||||
|
||||
impl serde_json::ser::Formatter for AsciiTurnMetadataFormatter {
|
||||
fn write_string_fragment<W: ?Sized + io::Write>(
|
||||
&mut self,
|
||||
writer: &mut W,
|
||||
fragment: &str,
|
||||
) -> io::Result<()> {
|
||||
let mut start = 0;
|
||||
for (index, ch) in fragment.char_indices() {
|
||||
if ch.is_ascii() {
|
||||
continue;
|
||||
}
|
||||
writer.write_all(&fragment.as_bytes()[start..index])?;
|
||||
for unit in ch.encode_utf16(&mut [0; 2]) {
|
||||
write!(writer, "\\u{unit:04x}")?;
|
||||
}
|
||||
start = index + ch.len_utf8();
|
||||
}
|
||||
writer.write_all(&fragment.as_bytes()[start..])
|
||||
}
|
||||
}
|
||||
|
||||
fn replace_existing_body_string(
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
use gateway_core::metering::{
|
||||
CalculatedCostAmounts, CalculatedCostBreakdown, CalculatedCostRates, CurrencyCode, Decimal,
|
||||
Money,
|
||||
CalculatedCost, CalculatedCostAmounts, CalculatedCostBreakdown, CalculatedCostRates,
|
||||
CurrencyCode, Decimal, Money,
|
||||
};
|
||||
use gateway_protocol::openai::events::{TokenUsage, retry_after_seconds_from_body};
|
||||
use reqwest::StatusCode;
|
||||
@@ -493,6 +493,76 @@ fn openai_billing_breakdown_with_context(
|
||||
))
|
||||
}
|
||||
|
||||
/// 独立 Images 端点按公开标准 API 单价估算;不是 ChatGPT 账号实际扣费。
|
||||
pub(crate) fn image_calculated_cost(request_body: &[u8], usage: &Value) -> Option<CalculatedCost> {
|
||||
#[derive(serde::Deserialize)]
|
||||
struct ImageModel {
|
||||
model: String,
|
||||
}
|
||||
// 只读取模型,跳过编辑请求中可能很大的 base64 图片。
|
||||
let request = serde_json::from_slice::<ImageModel>(request_body).ok()?;
|
||||
if !matches!(
|
||||
request.model.as_str(),
|
||||
"gpt-image-2" | "gpt-image-2-2026-04-21"
|
||||
) {
|
||||
return None;
|
||||
}
|
||||
let input = usage.get("input_tokens")?.as_u64()?;
|
||||
let output = usage.get("output_tokens")?.as_u64()?;
|
||||
let details = usage.get("input_tokens_details")?;
|
||||
let text_input = details.get("text_tokens")?.as_u64()?;
|
||||
let image_input = details.get("image_tokens")?.as_u64()?;
|
||||
if text_input.checked_add(image_input)? != input {
|
||||
return None;
|
||||
}
|
||||
if let Some(total) = usage.get("total_tokens")
|
||||
&& total.as_u64()? != input.checked_add(output)?
|
||||
{
|
||||
return None;
|
||||
}
|
||||
// 此模型仅输出图片;上游如报告其他输出模态,不能套用图片单价。
|
||||
if let Some(details) = usage.get("output_tokens_details")
|
||||
&& (details.get("image_tokens")?.as_u64()? != output
|
||||
|| details.get("text_tokens")?.as_u64()? != 0)
|
||||
{
|
||||
return None;
|
||||
}
|
||||
let cached = match details.get("cached_tokens") {
|
||||
None => 0,
|
||||
Some(value) => value.as_u64()?,
|
||||
};
|
||||
if cached > input {
|
||||
return None;
|
||||
}
|
||||
// 缓存只给总量时,混合输入无法判定应套用哪种缓存单价。
|
||||
let (cached_text, cached_image) = match (text_input, image_input, cached) {
|
||||
(_, _, 0) => (0, 0),
|
||||
(_, 0, cached) => (cached, 0),
|
||||
(0, _, cached) => (0, cached),
|
||||
(_, _, cached) if cached == input => (text_input, image_input),
|
||||
_ => return None,
|
||||
};
|
||||
// https://developers.openai.com/api/docs/pricing (2026-09-08)
|
||||
// USD / 1M tokens: text 5 / cached 1.25; image 8 / cached 2 / output 30.
|
||||
let text = token_amounts(
|
||||
TokenRates::new(50_000, 0, 12_500),
|
||||
0,
|
||||
text_input,
|
||||
0,
|
||||
cached_text,
|
||||
0,
|
||||
)?;
|
||||
let image = token_amounts(
|
||||
TokenRates::new(80_000, 300_000, 20_000),
|
||||
0,
|
||||
image_input,
|
||||
output,
|
||||
cached_image,
|
||||
0,
|
||||
)?;
|
||||
CalculatedCost::from_usd_ticks(text.total_ticks.checked_add(image.total_ticks)?).ok()
|
||||
}
|
||||
|
||||
fn model_pricing(model: &str) -> Option<ModelPricing> {
|
||||
let normalized = normalize_model_name(model);
|
||||
if UNPRICED_MODELS
|
||||
|
||||
@@ -151,6 +151,9 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
|
||||
.map(|pool_return| std::mem::take(&mut pool_return.continuation))
|
||||
.unwrap_or_default();
|
||||
let mut last_event_type = None;
|
||||
// 官方 run_websocket_response_stream 按响应消费 metadata 事件;它们不属于握手头。
|
||||
// 复用连接时恢复握手快照,避免上一轮模型信息及普通响应头随每次请求累积。
|
||||
let opening_response_metadata = metadata.response_metadata.clone();
|
||||
loop {
|
||||
let message = tokio::select! {
|
||||
biased;
|
||||
@@ -323,6 +326,7 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
|
||||
);
|
||||
match terminal {
|
||||
WebSocketTerminalKind::Completed => {
|
||||
metadata.response_metadata = opening_response_metadata;
|
||||
finish_stream_websocket(websocket, metadata, continuation, pool_return.take())
|
||||
.await;
|
||||
}
|
||||
|
||||
@@ -26,6 +26,7 @@ use gateway_core::error::{
|
||||
};
|
||||
use gateway_core::event::{GatewayEvent, WebSocketPoolKind};
|
||||
use gateway_core::lifecycle::CancellationToken;
|
||||
use gateway_core::metering::Usage;
|
||||
use gateway_core::operation::{
|
||||
CapabilityRequirements, GenerateRequest, ImageRequest, ImageRequestKind, Operation,
|
||||
OperationKind, ProtocolPayload, ProviderSessionState, RawJsonPayload, StandaloneSearchRequest,
|
||||
@@ -1051,6 +1052,13 @@ async fn image_endpoints_bypass_only_the_text_catalog_and_preserve_the_current_c
|
||||
let mut observed_http_json = false;
|
||||
while let Some(event) = stream.next().await {
|
||||
let event = event.expect("image provider event");
|
||||
assert!(
|
||||
event
|
||||
.canonical_facts()
|
||||
.iter()
|
||||
.all(|fact| !matches!(fact, GatewayEvent::Usage(_))),
|
||||
"missing upstream usage must remain unknown"
|
||||
);
|
||||
completed |= event
|
||||
.canonical_facts()
|
||||
.iter()
|
||||
@@ -1075,6 +1083,289 @@ async fn image_endpoints_bypass_only_the_text_catalog_and_preserve_the_current_c
|
||||
server.verify().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn image_endpoints_should_report_usage_before_delivering_the_unchanged_body() {
|
||||
let response = br#"{ "created":1778832973,"data":[{"b64_json":"AAEC"}],"quality":"medium","size":"1024x1536","usage":{"input_tokens":1474,"input_tokens_details":{"image_tokens":1457,"text_tokens":17,"cached_tokens":100},"output_tokens":1372,"output_tokens_details":{"image_tokens":1372,"text_tokens":0},"total_tokens":2846},"future":9007199254740993 }"#;
|
||||
let expected = Usage {
|
||||
input_tokens: Some(1474),
|
||||
output_tokens: Some(1372),
|
||||
cached_tokens: Some(100),
|
||||
image_input_tokens: Some(1457),
|
||||
image_output_tokens: Some(1372),
|
||||
total_tokens: Some(2846),
|
||||
..Usage::default()
|
||||
};
|
||||
for kind in [ImageRequestKind::Generation, ImageRequestKind::Edit] {
|
||||
let events = image_usage_events(kind, response).await;
|
||||
let usage = events
|
||||
.iter()
|
||||
.flat_map(|event| event.canonical_facts())
|
||||
.filter_map(|fact| match fact {
|
||||
GatewayEvent::Usage(usage) => Some(usage.clone()),
|
||||
_ => None,
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(usage.as_slice(), std::slice::from_ref(&expected));
|
||||
let usage_index = events
|
||||
.iter()
|
||||
.position(|event| {
|
||||
event
|
||||
.canonical_facts()
|
||||
.iter()
|
||||
.any(|fact| matches!(fact, GatewayEvent::Usage(_)))
|
||||
})
|
||||
.expect("usage event");
|
||||
let wire_index = events
|
||||
.iter()
|
||||
.position(|event| {
|
||||
event
|
||||
.wire_event()
|
||||
.and_then(|wire| wire.raw_json_body())
|
||||
.is_some()
|
||||
})
|
||||
.expect("raw image response");
|
||||
assert!(
|
||||
usage_index < wire_index,
|
||||
"usage must be observed before image delivery"
|
||||
);
|
||||
assert_eq!(
|
||||
events[wire_index]
|
||||
.wire_event()
|
||||
.and_then(|wire| wire.raw_json_body())
|
||||
.map(|body| body.as_ref()),
|
||||
Some(response.as_slice())
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn image_usage_should_preserve_unknown_fields_and_explicit_zero_counts() {
|
||||
let cases = [
|
||||
(json!(null), None),
|
||||
(json!({}), None),
|
||||
(json!([]), None),
|
||||
(
|
||||
json!({"input_tokens": -1, "output_tokens": "42", "total_tokens": 1.5}),
|
||||
None,
|
||||
),
|
||||
(
|
||||
json!({"input_tokens": 0, "output_tokens": 0, "total_tokens": 0, "input_tokens_details": {"image_tokens": 0, "cached_tokens": 0}, "output_tokens_details": {"image_tokens": 0}}),
|
||||
Some(Usage {
|
||||
input_tokens: Some(0),
|
||||
output_tokens: Some(0),
|
||||
total_tokens: Some(0),
|
||||
cached_tokens: Some(0),
|
||||
image_input_tokens: Some(0),
|
||||
image_output_tokens: Some(0),
|
||||
..Usage::default()
|
||||
}),
|
||||
),
|
||||
(
|
||||
json!({"input_tokens": 17, "output_tokens": 1372, "input_tokens_details": {"image_tokens": "invalid"}, "output_tokens_details": {"image_tokens": 1372}}),
|
||||
Some(Usage {
|
||||
input_tokens: Some(17),
|
||||
output_tokens: Some(1372),
|
||||
image_output_tokens: Some(1372),
|
||||
..Usage::default()
|
||||
}),
|
||||
),
|
||||
];
|
||||
for (usage, expected) in cases {
|
||||
let response = serde_json::to_vec(
|
||||
&json!({"created":1778832973,"data":[{"b64_json":"AAEC"}],"usage":usage}),
|
||||
)
|
||||
.expect("image response JSON");
|
||||
let events = image_usage_events(ImageRequestKind::Generation, &response).await;
|
||||
let actual = events
|
||||
.iter()
|
||||
.flat_map(|event| event.canonical_facts())
|
||||
.filter_map(|fact| match fact {
|
||||
GatewayEvent::Usage(usage) => Some(usage.clone()),
|
||||
_ => None,
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(actual, expected.into_iter().collect::<Vec<_>>());
|
||||
let raw = events
|
||||
.iter()
|
||||
.find_map(|event| event.wire_event().and_then(|wire| wire.raw_json_body()));
|
||||
assert_eq!(raw.map(|body| body.as_ref()), Some(response.as_slice()));
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn image_prices_should_use_modality_rates_and_precede_delivery() {
|
||||
// First case is the real gpt-image-2 response verified on 2026-09-08.
|
||||
let cases = [
|
||||
(18, 0, 229, None, 69_600_000_u128),
|
||||
(17, 1457, 1372, None, 529_010_000),
|
||||
(100, 0, 10, Some(40), 6_500_000),
|
||||
(0, 100, 10, Some(40), 8_600_000),
|
||||
(20, 80, 10, Some(100), 4_850_000),
|
||||
(0, 0, 0, Some(0), 0),
|
||||
];
|
||||
for (text, image, output, cached, expected_ticks) in cases {
|
||||
let mut usage = json!({
|
||||
"input_tokens":text+image,
|
||||
"input_tokens_details":{"text_tokens":text,"image_tokens":image},
|
||||
"output_tokens":output,
|
||||
"output_tokens_details":{"text_tokens":0,"image_tokens":output},
|
||||
"total_tokens":text+image+output,
|
||||
});
|
||||
if let Some(cached) = cached {
|
||||
usage["input_tokens_details"]["cached_tokens"] = json!(cached);
|
||||
}
|
||||
let response = serde_json::to_vec(&json!({"data":[{"b64_json":"AAEC"}],"usage":usage}))
|
||||
.expect("response");
|
||||
for (kind, model) in [
|
||||
(ImageRequestKind::Generation, "gpt-image-2"),
|
||||
(ImageRequestKind::Edit, "gpt-image-2-2026-04-21"),
|
||||
] {
|
||||
let events = image_metering_events(kind, &response, model).await;
|
||||
let costs = events
|
||||
.iter()
|
||||
.enumerate()
|
||||
.flat_map(|(index, event)| {
|
||||
event
|
||||
.canonical_facts()
|
||||
.iter()
|
||||
.filter_map(move |fact| match fact {
|
||||
GatewayEvent::CalculatedCost(cost) => Some((index, *cost)),
|
||||
_ => None,
|
||||
})
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(costs.len(), 1);
|
||||
let (cost_index, cost) = costs[0];
|
||||
assert_eq!(cost.total().amount().scaled(), expected_ticks);
|
||||
assert_eq!(cost.into_estimate().source().as_str(), "calculated");
|
||||
let wire_index = events
|
||||
.iter()
|
||||
.position(|event| event.wire_event().is_some())
|
||||
.expect("wire event");
|
||||
assert!(cost_index < wire_index);
|
||||
assert_eq!(
|
||||
events[wire_index]
|
||||
.wire_event()
|
||||
.and_then(|wire| wire.raw_json_body())
|
||||
.map(|body| body.as_ref()),
|
||||
Some(response.as_slice())
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn image_prices_should_remain_unknown_when_modality_or_model_is_uncertain() {
|
||||
let valid = json!({
|
||||
"input_tokens":100,
|
||||
"input_tokens_details":{"text_tokens":20,"image_tokens":80},
|
||||
"output_tokens":10,
|
||||
"output_tokens_details":{"text_tokens":0,"image_tokens":10},
|
||||
"total_tokens":110,
|
||||
});
|
||||
let mut cases = vec![
|
||||
("gpt-image-future", valid.clone()),
|
||||
("gpt-image-2-future", valid.clone()),
|
||||
("gpt-image-2", json!(null)),
|
||||
(
|
||||
"gpt-image-2",
|
||||
json!({"input_tokens":100,"output_tokens":10}),
|
||||
),
|
||||
];
|
||||
for (pointer, value) in [
|
||||
("/input_tokens", json!(101)),
|
||||
("/input_tokens_details/text_tokens", json!(null)),
|
||||
("/input_tokens_details/image_tokens", json!("80")),
|
||||
("/output_tokens_details/text_tokens", json!(1)),
|
||||
("/output_tokens_details/image_tokens", json!(9)),
|
||||
("/output_tokens", json!(-1)),
|
||||
("/total_tokens", json!(111)),
|
||||
] {
|
||||
let mut usage = valid.clone();
|
||||
*usage.pointer_mut(pointer).expect("existing field") = value;
|
||||
cases.push(("gpt-image-2", usage));
|
||||
}
|
||||
for cached in [json!(40), json!(101), json!(null), json!("40")] {
|
||||
let mut usage = valid.clone();
|
||||
usage["input_tokens_details"]["cached_tokens"] = cached;
|
||||
cases.push(("gpt-image-2", usage));
|
||||
}
|
||||
cases.push(("gpt-image-2", json!({
|
||||
"input_tokens":u64::MAX,"input_tokens_details":{"text_tokens":u64::MAX,"image_tokens":0},
|
||||
"output_tokens":10,
|
||||
})));
|
||||
for (model, usage) in cases {
|
||||
let response = serde_json::to_vec(&json!({"data":[{"b64_json":"AAEC"}],"usage":usage}))
|
||||
.expect("response");
|
||||
let events = image_metering_events(ImageRequestKind::Generation, &response, model).await;
|
||||
assert!(
|
||||
events
|
||||
.iter()
|
||||
.flat_map(|event| event.canonical_facts())
|
||||
.all(|fact| !matches!(fact, GatewayEvent::CalculatedCost(_)))
|
||||
);
|
||||
assert_eq!(
|
||||
events
|
||||
.iter()
|
||||
.find_map(|event| event.wire_event().and_then(|wire| wire.raw_json_body()))
|
||||
.map(|body| body.as_ref()),
|
||||
Some(response.as_slice())
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async fn image_usage_events(
|
||||
kind: ImageRequestKind,
|
||||
response: &[u8],
|
||||
) -> Vec<gateway_core::event::ProviderEvent> {
|
||||
image_metering_events(kind, response, "gpt-image-2").await
|
||||
}
|
||||
|
||||
async fn image_metering_events(
|
||||
kind: ImageRequestKind,
|
||||
response: &[u8],
|
||||
model: &str,
|
||||
) -> Vec<gateway_core::event::ProviderEvent> {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
create_account(&store, "acct_provider_contract").await;
|
||||
let server = MockServer::start().await;
|
||||
let endpoint = match kind {
|
||||
ImageRequestKind::Generation => "/codex/images/generations",
|
||||
ImageRequestKind::Edit => "/codex/images/edits",
|
||||
};
|
||||
Mock::given(method("POST"))
|
||||
.and(path(endpoint))
|
||||
.respond_with(
|
||||
ResponseTemplate::new(200).set_body_raw(response.to_vec(), "application/json"),
|
||||
)
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
let provider = provider_with_base_url(&store, server.uri());
|
||||
let payload = RawJsonPayload::new(
|
||||
"openai",
|
||||
Bytes::from(
|
||||
serde_json::to_vec(&json!({"model":model,"prompt":"draw a tree"}))
|
||||
.expect("request JSON"),
|
||||
),
|
||||
)
|
||||
.expect("image payload");
|
||||
let operation = Operation::GenerateImage(ImageRequest::from_raw_json(kind, payload));
|
||||
let mut stream = provider
|
||||
.execute(
|
||||
planned_provider_endpoint_request("openai", operation),
|
||||
context("req_image_usage", CancellationToken::new()),
|
||||
)
|
||||
.await
|
||||
.expect("prepare image stream");
|
||||
let mut events = Vec::new();
|
||||
while let Some(event) = stream.next().await {
|
||||
events.push(event.expect("image provider event"));
|
||||
}
|
||||
server.verify().await;
|
||||
events
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn image_endpoint_returns_the_exact_upstream_error_response() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
@@ -2521,6 +2812,127 @@ async fn fallback_attempt_transport_forces_http_sse_for_a_websocket_request() {
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn downstream_websocket_new_chain_should_override_session_and_attempt_http_fallback() {
|
||||
const SESSION_ID: &str = "downstream-required-session";
|
||||
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
create_account(&store, "acct_http_sse_exhausted").await;
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let base_url = format!("http://{}", listener.local_addr().unwrap());
|
||||
let server = tokio::spawn(async move {
|
||||
let (mut opening, _) = listener.accept().await.unwrap();
|
||||
let request = capture_http_request(&mut opening).await;
|
||||
assert!(String::from_utf8_lossy(&request).starts_with("GET /codex/responses"));
|
||||
opening
|
||||
.write_all(
|
||||
b"HTTP/1.1 426 Upgrade Required\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let (stream, _) = listener.accept().await.unwrap();
|
||||
let mut websocket = accept_codex_test_websocket(stream).await;
|
||||
for id in ["resp_required_session", "resp_required_attempt"] {
|
||||
let request = websocket.next().await.unwrap().unwrap();
|
||||
let payload: Value = serde_json::from_str(request.to_text().unwrap()).unwrap();
|
||||
assert_eq!(payload["store"], false);
|
||||
websocket
|
||||
.send(Message::Text(
|
||||
json!({
|
||||
"type": "response.completed",
|
||||
"response": {
|
||||
"id": id,
|
||||
"model": "gpt-5.4",
|
||||
"status": "completed",
|
||||
"store": false,
|
||||
"output": [],
|
||||
"usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}
|
||||
}
|
||||
})
|
||||
.to_string()
|
||||
.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
});
|
||||
let provider = provider_with_base_url(&store, base_url);
|
||||
let mut seed = provider
|
||||
.execute(
|
||||
planned_request(
|
||||
"openai",
|
||||
Operation::Generate(generate_with_session_context(SESSION_ID, None, None)),
|
||||
),
|
||||
context("req_disable_optional_websocket", CancellationToken::new()),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let error = loop {
|
||||
match seed.next().await {
|
||||
Some(Ok(_)) => {}
|
||||
Some(Err(error)) => break error,
|
||||
None => panic!("426 opening must enable session HTTP fallback"),
|
||||
}
|
||||
};
|
||||
assert_eq!(
|
||||
error.pre_delivery_retry(),
|
||||
Some(PreDeliveryRetry::SameAccountTransportFallback)
|
||||
);
|
||||
drop(seed);
|
||||
|
||||
for attempt in [
|
||||
context("req_required_session", CancellationToken::new()),
|
||||
fallback_transport_context("req_required_attempt"),
|
||||
] {
|
||||
let payload = ProtocolPayload::json_object(
|
||||
"openai",
|
||||
Map::from_iter([
|
||||
("model".to_owned(), json!("gpt-5.4")),
|
||||
("input".to_owned(), json!("hello")),
|
||||
("store".to_owned(), json!(false)),
|
||||
("session_id".to_owned(), json!(SESSION_ID)),
|
||||
]),
|
||||
)
|
||||
.unwrap()
|
||||
.with_context(Map::from_iter([(
|
||||
"downstream_websocket_connection_id".to_owned(),
|
||||
json!("ws_required_session"),
|
||||
)]));
|
||||
let request = GenerateRequest::from_protocol_payload(payload).with_provider_session_state(
|
||||
ProviderSessionState::new(
|
||||
"openai",
|
||||
Map::from_iter([
|
||||
("account_id".to_owned(), json!("acct_http_sse_exhausted")),
|
||||
("conversation_id".to_owned(), json!(SESSION_ID)),
|
||||
("continuation_scope".to_owned(), json!("persisted")),
|
||||
]),
|
||||
)
|
||||
.unwrap(),
|
||||
);
|
||||
let mut stream = provider
|
||||
.execute(
|
||||
planned_request("openai", Operation::Generate(request)),
|
||||
attempt,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(stream.metadata().transport().as_str(), "websocket");
|
||||
let mut session = None;
|
||||
while let Some(event) = stream.next().await {
|
||||
let event = event.expect("required WebSocket response");
|
||||
if let Some(update) = event.session_update() {
|
||||
session = Some(update.clone());
|
||||
}
|
||||
}
|
||||
assert_eq!(
|
||||
session.unwrap().payload().get("continuation_scope"),
|
||||
Some(&json!("connection_local"))
|
||||
);
|
||||
}
|
||||
server.await.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn websocket_turn_state_metadata_is_exposed_through_response_observation() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
@@ -2942,6 +3354,108 @@ async fn cross_account_scope_sanitizes_only_known_turn_metadata_fields() {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn websocket_account_scoping_preserves_ascii_turn_metadata_and_unicode_input() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
create_account(&store, "acct_scope_same").await;
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.expect("listener");
|
||||
let base_url = format!("http://{}", listener.local_addr().expect("address"));
|
||||
let server = tokio::spawn(async move {
|
||||
let (socket, _) = listener.accept().await.expect("accept websocket");
|
||||
let mut socket = accept_codex_test_websocket(socket).await;
|
||||
let message = socket.next().await.expect("request").expect("valid frame");
|
||||
let body: Value = serde_json::from_str(message.to_text().expect("text")).expect("JSON");
|
||||
socket.send(Message::Text(json!({
|
||||
"type": "response.completed",
|
||||
"response": {"id": "resp_ascii_metadata", "model": "gpt-5.4", "status": "completed", "output": []}
|
||||
}).to_string().into())).await.expect("complete response");
|
||||
body
|
||||
});
|
||||
// Official Codex keeps embedded turn metadata ASCII even for Unicode workspaces.
|
||||
let raw = r#"{"installation_id":"client-installation","workspaces":{"C:\\Users\\\u9879\u76ee\\\ud83d\ude80":{"label":"caf\u00e9","literal":"\\u4e2d","quoted":"\"line\n"}}}"#;
|
||||
let input = json!([{"role": "user", "content": "中文正文 🚀"}]);
|
||||
let payload = ProtocolPayload::json_object(
|
||||
"openai",
|
||||
Map::from_iter([
|
||||
("model".to_owned(), json!("gpt-5.4")),
|
||||
("input".to_owned(), input.clone()),
|
||||
(
|
||||
"client_metadata".to_owned(),
|
||||
json!({"x-codex-turn-metadata": raw}),
|
||||
),
|
||||
]),
|
||||
)
|
||||
.expect("payload")
|
||||
.with_context(Map::from_iter([("turn_metadata".to_owned(), json!(raw))]));
|
||||
let mut stream = provider_with_base_url(&store, base_url)
|
||||
.execute(
|
||||
planned_request(
|
||||
"openai",
|
||||
Operation::Generate(GenerateRequest::from_protocol_payload(payload)),
|
||||
),
|
||||
context_with_state_owner("req_ascii_metadata", "acct_scope_same"),
|
||||
)
|
||||
.await
|
||||
.expect("provider stream");
|
||||
while let Some(event) = stream.next().await {
|
||||
event.expect("successful websocket response");
|
||||
}
|
||||
let body = server.await.expect("server");
|
||||
let encoded = body
|
||||
.pointer("/client_metadata/x-codex-turn-metadata")
|
||||
.and_then(Value::as_str)
|
||||
.expect("turn metadata");
|
||||
assert!(encoded.is_ascii(), "embedded header JSON must remain ASCII");
|
||||
let mut expected: Value = serde_json::from_str(raw).expect("original metadata");
|
||||
expected["installation_id"] = body["client_metadata"]["installation_id"].clone();
|
||||
assert_eq!(
|
||||
serde_json::from_str::<Value>(encoded).expect("metadata JSON"),
|
||||
expected
|
||||
);
|
||||
assert_eq!(body["input"], input);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn http_account_scoping_keeps_unicode_metadata_ascii_in_headers_and_body() {
|
||||
for owner in ["acct_scope_same", "acct_scope_old"] {
|
||||
let raw = r#"{"installation_id":"client-installation","workspaces":{"/tmp/\u4e2d\u6587/\ud83d\ude80":{"label":"caf\u00e9"}}}"#;
|
||||
let request = capture_scoped_http_request(
|
||||
"req_ascii_http",
|
||||
"acct_scope_same",
|
||||
owner,
|
||||
Map::from_iter([
|
||||
("model".to_owned(), json!("gpt-5.4")),
|
||||
("input".to_owned(), json!("中文正文 🚀")),
|
||||
("turnMetadata".to_owned(), json!(raw)),
|
||||
(
|
||||
"client_metadata".to_owned(),
|
||||
json!({"x-codex-turn-metadata": raw}),
|
||||
),
|
||||
]),
|
||||
Map::from_iter([("turn_metadata".to_owned(), json!(raw))]),
|
||||
)
|
||||
.await;
|
||||
let body = captured_request_body(&request);
|
||||
let headers = captured_header_values(&request, "x-codex-turn-metadata");
|
||||
assert_eq!(headers.len(), 1);
|
||||
let mut expected: Value = serde_json::from_str(raw).expect("original metadata");
|
||||
expected["installation_id"] = body["client_metadata"]["installation_id"].clone();
|
||||
for encoded in [
|
||||
std::str::from_utf8(&headers[0]).expect("UTF-8 header"),
|
||||
body["turnMetadata"].as_str().expect("body turn metadata"),
|
||||
body["client_metadata"]["x-codex-turn-metadata"]
|
||||
.as_str()
|
||||
.expect("client metadata"),
|
||||
] {
|
||||
assert!(encoded.is_ascii(), "scoped header JSON must remain ASCII");
|
||||
assert_eq!(
|
||||
serde_json::from_str::<Value>(encoded).expect("metadata JSON"),
|
||||
expected
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cross_account_continuation_should_require_client_replay_without_an_upstream_probe() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
@@ -5408,6 +5922,52 @@ fn request_observation_reads_openai_metadata_without_changing_the_operation() {
|
||||
assert!(matches!(operation, Operation::Generate(_)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn endpoint_observation_should_read_models_without_rewriting_or_requiring_a_catalog() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
let provider = provider(&store);
|
||||
let client_key = ClientApiKeyId::new("key_endpoint_model").expect("client key");
|
||||
for model in [
|
||||
json!("gpt-image-future"),
|
||||
json!("gpt-5.6-sol"),
|
||||
json!(null),
|
||||
json!(42),
|
||||
json!(""),
|
||||
] {
|
||||
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 operations = [
|
||||
Operation::GenerateImage(ImageRequest::from_raw_json(
|
||||
ImageRequestKind::Generation,
|
||||
payload.clone(),
|
||||
)),
|
||||
Operation::GenerateImage(ImageRequest::from_raw_json(
|
||||
ImageRequestKind::Edit,
|
||||
payload.clone(),
|
||||
)),
|
||||
Operation::Search(StandaloneSearchRequest::from_raw_json(payload)),
|
||||
];
|
||||
for operation in operations {
|
||||
let observation = provider.request_observation(&operation, &client_key);
|
||||
assert_eq!(
|
||||
observation
|
||||
.requested_model
|
||||
.as_ref()
|
||||
.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());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn request_observation_preserves_the_raw_reasoning_effort() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
|
||||
@@ -115,6 +115,79 @@ async fn cold_websocket_should_fall_back_without_recording_a_successful_connect(
|
||||
pool.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn downstream_websocket_new_chain_should_preserve_continuation_after_slow_opening() {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let server = tokio::spawn(async move {
|
||||
let (stream, _) = listener.accept().await.unwrap();
|
||||
tokio::select! {
|
||||
() = tokio::time::sleep(Duration::from_secs(1)) => {}
|
||||
accepted = listener.accept() => {
|
||||
let (mut http, _) = accepted.unwrap();
|
||||
let request = read_http_request(&mut http).await;
|
||||
assert!(request.starts_with("POST /codex/responses HTTP/1.1"));
|
||||
write_completed_sse_response(&mut http).await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
let mut websocket = accept_codex_test_websocket(stream).await;
|
||||
let _first = websocket.next().await.unwrap().unwrap();
|
||||
websocket
|
||||
.send(Message::Text(
|
||||
completed_websocket_response("resp_slow_seed", 2, 1).into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let second = websocket.next().await.unwrap().unwrap();
|
||||
let second: Value = serde_json::from_str(second.to_text().unwrap()).unwrap();
|
||||
assert_eq!(second["previous_response_id"], "resp_slow_seed");
|
||||
websocket
|
||||
.send(Message::Text(
|
||||
completed_websocket_response("resp_slow_continuation", 2, 1).into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
});
|
||||
let pool = Arc::new(CodexWebSocketPool::default());
|
||||
let backend = CodexBackendClient::new(
|
||||
reqwest::Client::builder().no_proxy().build().unwrap(),
|
||||
format!("http://{addr}"),
|
||||
test_wire_profile(),
|
||||
)
|
||||
.with_websocket_pool(Arc::clone(&pool));
|
||||
let mut request = new_chain_request("conversation-downstream-slow");
|
||||
request.downstream_websocket_connection_id = Some("ws_downstream_slow".to_owned());
|
||||
|
||||
let first = backend
|
||||
.create_response(
|
||||
&request,
|
||||
request_context("req_slow_seed", Some("chatgpt-account")),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(first.transport, CodexBackendTransport::WebSocket);
|
||||
assert!(first.connection_local_continuation);
|
||||
|
||||
request.set_previous_response_id(Some("resp_slow_seed".to_owned()));
|
||||
request.previous_response_scope = Some(PreviousResponseScope::ConnectionLocal);
|
||||
let second = backend
|
||||
.create_response(
|
||||
&request,
|
||||
request_context("req_slow_continuation", Some("chatgpt-account")),
|
||||
)
|
||||
.await
|
||||
.expect("the next turn must use the socket that generated the previous response");
|
||||
assert_eq!(
|
||||
second.transport_metrics.decision,
|
||||
Some(CodexTransportDecision::ExactWebSocket)
|
||||
);
|
||||
assert!(second.body.contains("resp_slow_continuation"));
|
||||
server.await.unwrap();
|
||||
pool.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn external_continuation_should_wait_for_a_cold_websocket() {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
|
||||
@@ -273,6 +273,34 @@ fn transport_should_prefer_websocket_when_requested_without_history() {
|
||||
assert!(transport_requirement(&request).allows_pre_send_http_fallback());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn downstream_websocket_new_chain_should_require_pooled_websocket_without_storage() {
|
||||
for store in [None, Some(false)] {
|
||||
let mut body = serde_json::Map::new();
|
||||
if let Some(store) = store {
|
||||
body.insert("store".to_owned(), json!(store));
|
||||
}
|
||||
let mut request = CodexResponsesRequest::from_body(body);
|
||||
request.downstream_websocket_connection_id = Some("ws_downstream".to_owned());
|
||||
|
||||
let requirement = transport_requirement(&request);
|
||||
assert!(requirement.requires_websocket());
|
||||
assert!(!requirement.allows_pre_send_http_fallback());
|
||||
assert!(requirement.allows_connection_restart());
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn downstream_websocket_new_chain_should_allow_http_when_response_is_stored() {
|
||||
let mut request = CodexResponsesRequest::from_body(serde_json::Map::from_iter([(
|
||||
"store".to_owned(),
|
||||
json!(true),
|
||||
)]));
|
||||
request.downstream_websocket_connection_id = Some("ws_downstream".to_owned());
|
||||
|
||||
assert!(transport_requirement(&request).allows_pre_send_http_fallback());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unspecified_transport_without_history_should_still_be_a_new_chain() {
|
||||
let request = codex_request("gpt-test", "be brief", Vec::new());
|
||||
|
||||
@@ -0,0 +1,92 @@
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn reused_websocket_should_keep_response_metadata_scoped_to_each_exchange() {
|
||||
const ROUNDS: usize = 40;
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let server = tokio::spawn(async move {
|
||||
let (stream, _) = listener.accept().await.unwrap();
|
||||
let mut websocket = accept_codex_test_websocket_with(stream, |_, response| {
|
||||
for value in ["first", "second"] {
|
||||
response
|
||||
.headers_mut()
|
||||
.append("x-opening-multi", value.parse().unwrap());
|
||||
}
|
||||
})
|
||||
.await;
|
||||
for round in 0..ROUNDS {
|
||||
websocket.next().await.unwrap().unwrap();
|
||||
websocket
|
||||
.send(Message::Text(
|
||||
json!({
|
||||
"type": "codex.response.metadata",
|
||||
"headers": {
|
||||
"x-models-etag": format!("etag-{round}"),
|
||||
"openai-model": format!("model-{round}"),
|
||||
"x-reasoning-included": "true",
|
||||
"x-codex-safety-buffering-enabled": "true",
|
||||
"x-codex-safety-buffering-faster-model": "faster-model",
|
||||
"x-future-response-metadata": format!("future-{round}"),
|
||||
"x-codex-turn-state": format!("turn-{round}")
|
||||
}
|
||||
})
|
||||
.to_string()
|
||||
.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
websocket
|
||||
.send(Message::Text(
|
||||
completed_websocket_response(&format!("resp_metadata_{round}"), 1, 1).into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
});
|
||||
let pool = Arc::new(CodexWebSocketPool::new(Duration::from_mins(1)));
|
||||
let backend = CodexBackendClient::new(
|
||||
reqwest::Client::builder().no_proxy().build().unwrap(),
|
||||
format!("http://{addr}"),
|
||||
test_wire_profile(),
|
||||
)
|
||||
.with_websocket_pool(pool);
|
||||
let request = pooled_websocket_request("conversation-response-metadata");
|
||||
let expected = CodexResponseMetadata {
|
||||
client_headers: vec![
|
||||
(
|
||||
"x-opening-multi".to_owned(),
|
||||
bytes::Bytes::from_static(b"first"),
|
||||
),
|
||||
(
|
||||
"x-opening-multi".to_owned(),
|
||||
bytes::Bytes::from_static(b"second"),
|
||||
),
|
||||
],
|
||||
..CodexResponseMetadata::default()
|
||||
};
|
||||
|
||||
for round in 0..ROUNDS {
|
||||
let response = timeout(
|
||||
Duration::from_secs(10),
|
||||
backend.create_response(
|
||||
&request,
|
||||
request_context("req_response_metadata", Some("chatgpt-account")),
|
||||
),
|
||||
)
|
||||
.await
|
||||
.expect("response within timeout")
|
||||
.expect("response should complete on the same socket");
|
||||
assert_eq!(response.response_metadata, expected, "exchange {round}");
|
||||
assert!(response.body.contains(&format!("etag-{round}")));
|
||||
assert_eq!(response.turn_state, Some(format!("turn-{round}")));
|
||||
if round > 0 {
|
||||
assert!(
|
||||
response
|
||||
.websocket_pool_decision
|
||||
.is_some_and(WebSocketPoolDecision::is_reuse)
|
||||
);
|
||||
}
|
||||
}
|
||||
server.await.unwrap();
|
||||
}
|
||||
@@ -4,6 +4,8 @@ use provider_openai::transport::{
|
||||
websocket::write_websocket_audit_artifact_for_dir,
|
||||
};
|
||||
|
||||
mod metadata;
|
||||
|
||||
fn rate_limit_event(used_percent: u64) -> String {
|
||||
json!({
|
||||
"type": "codex.rate_limits",
|
||||
|
||||
@@ -152,6 +152,7 @@ impl Provider for GrokBuildProvider {
|
||||
}
|
||||
let semantics = codex_responses_request_semantics(payload.body(), payload.context());
|
||||
ProviderRequestObservation {
|
||||
requested_model: None,
|
||||
reasoning_effort: semantics.reasoning_effort,
|
||||
reasoning_preset: semantics.reasoning_preset.map(str::to_owned),
|
||||
request_kind: semantics.request_kind,
|
||||
|
||||
@@ -19,7 +19,7 @@
|
||||
"@boxicons/vue": "1.0.1",
|
||||
"@fontsource-variable/inter": "^5.3.0",
|
||||
"@fontsource-variable/jetbrains-mono": "^5.3.0",
|
||||
"@lucide/vue": "^1.37.0",
|
||||
"@lucide/vue": "^1.41.0",
|
||||
"@vueuse/core": "^14.4.0",
|
||||
"axios": "^1.20.0",
|
||||
"dayjs": "^1.11.23",
|
||||
@@ -31,15 +31,15 @@
|
||||
"pinia": "^4.0.3",
|
||||
"pinia-plugin-persistedstate": "^4.7.1",
|
||||
"vue": "^3.5.42",
|
||||
"vue-router": "^5.3.0"
|
||||
"vue-router": "^5.3.1"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@antfu/eslint-config": "^9.3.0",
|
||||
"@antfu/eslint-config": "^9.5.1",
|
||||
"@tailwindcss/vite": "^4.3.3",
|
||||
"@types/node": "^26.4.0",
|
||||
"@types/node": "^26.4.1",
|
||||
"@vitejs/plugin-vue": "^6.0.8",
|
||||
"@vue/tsconfig": "^0.9.1",
|
||||
"eslint": "^10.9.1",
|
||||
"eslint": "^10.10.0",
|
||||
"eslint-plugin-format": "^2.0.1",
|
||||
"eslint-plugin-vuejs-accessibility": "^2.6.0",
|
||||
"tailwindcss": "^4.3.3",
|
||||
|
||||
Generated
+465
-376
File diff suppressed because it is too large
Load Diff
@@ -1 +1 @@
|
||||
version: "3.3.25"
|
||||
version: "3.3.28"
|
||||
|
||||
Reference in New Issue
Block a user