mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-06 08:06:19 +00:00
fix(gateway): preserve OpenAI wire transparency
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
@@ -69,9 +69,9 @@ async fn handle_responses(
|
||||
Err(error) => return authentication_error_response(error),
|
||||
};
|
||||
let decoded = match if review {
|
||||
decode_review_request_with_headers(&body, &headers)
|
||||
decode_review_request_with_headers(&body, &headers, client.policy().provider_kind())
|
||||
} else {
|
||||
decode_request_with_headers(&body, &headers)
|
||||
decode_request_with_headers(&body, &headers, client.policy().provider_kind())
|
||||
} {
|
||||
Ok(decoded) => decoded,
|
||||
Err(error) => {
|
||||
|
||||
@@ -9,6 +9,7 @@ use gateway_core::operation::{
|
||||
ProviderOptions, ProviderSessionState, ReasoningEffort, ReasoningRequirement, ReasoningSummary,
|
||||
ResponsePersistence, ToolDefinition,
|
||||
};
|
||||
use gateway_core::routing::ProviderKind;
|
||||
use gateway_protocol::openai::{
|
||||
X_OPENAI_INTERNAL_CODEX_RESPONSES_LITE_HEADER, X_OPENAI_MEMGEN_REQUEST_HEADER,
|
||||
};
|
||||
@@ -20,6 +21,7 @@ use super::error::RequestDecodeError;
|
||||
pub const PROVIDER_OPTIONS_VERSION: &str = "v1";
|
||||
|
||||
const OPENAI_PROTOCOL: &str = "openai";
|
||||
const XAI_PROVIDER: &str = "xai";
|
||||
const OPENAI_TRANSPORT_OPTION: &str = "transport";
|
||||
const HTTP_SSE_TRANSPORT: &str = "http_sse";
|
||||
const WEBSOCKET_TRANSPORT: &str = "websocket";
|
||||
@@ -311,29 +313,42 @@ impl fmt::Debug for DecodedResponsesRequest {
|
||||
///
|
||||
/// JSON 非法、顶层不是 object,或网关必须解释的路由字段无效时返回安全错误。
|
||||
pub fn decode_request(body: &[u8]) -> Result<DecodedResponsesRequest, RequestDecodeError> {
|
||||
decode_request_inner(body, false, &OpenAiRequestHeaders::default())
|
||||
decode_request_inner(body, false, &OpenAiRequestHeaders::default(), None)
|
||||
}
|
||||
|
||||
/// 使用下游 OpenAI/Codex 请求头解码 `POST /v1/responses`。
|
||||
pub(super) fn decode_request_with_headers(
|
||||
body: &[u8],
|
||||
headers: &HeaderMap,
|
||||
provider_kind: &ProviderKind,
|
||||
) -> Result<DecodedResponsesRequest, RequestDecodeError> {
|
||||
decode_request_inner(body, false, &OpenAiRequestHeaders::from_headers(headers))
|
||||
decode_request_inner(
|
||||
body,
|
||||
false,
|
||||
&OpenAiRequestHeaders::from_headers(headers),
|
||||
Some(provider_kind),
|
||||
)
|
||||
}
|
||||
|
||||
/// 解码 `POST /v1/responses/review` 请求并冻结 review subagent 语义。
|
||||
pub(super) fn decode_review_request_with_headers(
|
||||
body: &[u8],
|
||||
headers: &HeaderMap,
|
||||
provider_kind: &ProviderKind,
|
||||
) -> Result<DecodedResponsesRequest, RequestDecodeError> {
|
||||
decode_request_inner(body, true, &OpenAiRequestHeaders::from_headers(headers))
|
||||
decode_request_inner(
|
||||
body,
|
||||
true,
|
||||
&OpenAiRequestHeaders::from_headers(headers),
|
||||
Some(provider_kind),
|
||||
)
|
||||
}
|
||||
|
||||
pub(super) fn decode_request_inner(
|
||||
body: &[u8],
|
||||
review: bool,
|
||||
request_headers: &OpenAiRequestHeaders,
|
||||
provider_kind: Option<&ProviderKind>,
|
||||
) -> Result<DecodedResponsesRequest, RequestDecodeError> {
|
||||
let input_token_estimate =
|
||||
u64::try_from(body.len()).map_err(|_| RequestDecodeError::InvalidValue {
|
||||
@@ -358,7 +373,9 @@ pub(super) fn decode_request_inner(
|
||||
field: "input".to_owned(),
|
||||
});
|
||||
}
|
||||
let compact_conversation = consume_compaction_trigger(&mut object);
|
||||
let compact_conversation = provider_kind
|
||||
.is_some_and(|provider| provider.as_str() == XAI_PROVIDER)
|
||||
&& consume_compaction_trigger(&mut object);
|
||||
|
||||
let stream = optional_bool(&object, "stream", "stream")?.unwrap_or(true);
|
||||
let store = optional_bool(&object, "store", "store")?.unwrap_or(false);
|
||||
|
||||
@@ -31,8 +31,8 @@ struct PendingWireEvent {
|
||||
|
||||
/// Provider event 封套到 OpenAI Responses wire 的唯一表达边界。
|
||||
///
|
||||
/// OpenAI Provider 带原生 wire 时只下发该 wire;其他 Provider 的 canonical
|
||||
/// facts 仍由 [`ResponsesCollector`] 表达。同一响应不能在两种模式之间切换。
|
||||
/// Provider 已提供 OpenAI wire 时只下发该 wire;否则由 [`ResponsesCollector`]
|
||||
/// 表达 canonical facts。同一响应不能在两种模式之间切换。
|
||||
#[derive(Debug)]
|
||||
pub struct OpenAiResponsesEncoder {
|
||||
canonical: ResponsesCollector,
|
||||
|
||||
@@ -19,6 +19,7 @@ use gateway_core::error::{GatewayError, GatewayErrorKind};
|
||||
use gateway_core::event::ProviderResponseHeader;
|
||||
use gateway_core::lifecycle::{ConnectionGuard, ConnectionLifecycle};
|
||||
use gateway_core::operation::ProviderSessionState;
|
||||
use gateway_core::routing::ProviderKind;
|
||||
use serde_json::{Map, Value, json};
|
||||
use thiserror::Error;
|
||||
|
||||
@@ -157,32 +158,35 @@ async fn serve_responses_websocket(mut socket: WebSocket, session: ResponsesWebS
|
||||
};
|
||||
request_count = request_count.saturating_add(1);
|
||||
let correlation_id = session.service.next_request_id();
|
||||
let decoded =
|
||||
match decode_response_create_with_context(payload.as_str(), &session.request_headers) {
|
||||
Ok(decoded) => {
|
||||
decoded.with_client_context(session.client_ip, session.user_agent.clone())
|
||||
let decoded = match decode_response_create_with_context(
|
||||
payload.as_str(),
|
||||
&session.request_headers,
|
||||
session.client.policy().provider_kind(),
|
||||
) {
|
||||
Ok(decoded) => {
|
||||
decoded.with_client_context(session.client_ip, session.user_agent.clone())
|
||||
}
|
||||
Err(error) => {
|
||||
tracing::info!(
|
||||
websocket_connection_id = %session.connection_id,
|
||||
request_id = %correlation_id,
|
||||
error = %error,
|
||||
"Responses WebSocket request rejected"
|
||||
);
|
||||
if send_protocol_error(
|
||||
&mut socket,
|
||||
StatusCode::BAD_REQUEST,
|
||||
error.protocol_body(),
|
||||
&correlation_id,
|
||||
)
|
||||
.await
|
||||
== ForwardOutcome::Disconnect
|
||||
{
|
||||
break;
|
||||
}
|
||||
Err(error) => {
|
||||
tracing::info!(
|
||||
websocket_connection_id = %session.connection_id,
|
||||
request_id = %correlation_id,
|
||||
error = %error,
|
||||
"Responses WebSocket request rejected"
|
||||
);
|
||||
if send_protocol_error(
|
||||
&mut socket,
|
||||
StatusCode::BAD_REQUEST,
|
||||
error.protocol_body(),
|
||||
&correlation_id,
|
||||
)
|
||||
.await
|
||||
== ForwardOutcome::Disconnect
|
||||
{
|
||||
break;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
};
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let decoded = replay.prepare(decoded);
|
||||
let started = match session
|
||||
.service
|
||||
@@ -529,19 +533,21 @@ async fn next_active_input(
|
||||
pub fn decode_response_create(
|
||||
payload: &str,
|
||||
) -> Result<DecodedResponsesRequest, ResponseCreateFrameError> {
|
||||
decode_response_create_inner(payload, &OpenAiRequestHeaders::default())
|
||||
decode_response_create_inner(payload, &OpenAiRequestHeaders::default(), None)
|
||||
}
|
||||
|
||||
fn decode_response_create_with_context(
|
||||
payload: &str,
|
||||
request_headers: &OpenAiRequestHeaders,
|
||||
provider_kind: &ProviderKind,
|
||||
) -> Result<DecodedResponsesRequest, ResponseCreateFrameError> {
|
||||
decode_response_create_inner(payload, request_headers)
|
||||
decode_response_create_inner(payload, request_headers, Some(provider_kind))
|
||||
}
|
||||
|
||||
fn decode_response_create_inner(
|
||||
payload: &str,
|
||||
request_headers: &OpenAiRequestHeaders,
|
||||
provider_kind: Option<&ProviderKind>,
|
||||
) -> Result<DecodedResponsesRequest, ResponseCreateFrameError> {
|
||||
let Value::Object(mut body) = serde_json::from_str::<Value>(payload)
|
||||
.map_err(|_| ResponseCreateFrameError::InvalidJson)?
|
||||
@@ -569,7 +575,7 @@ fn decode_response_create_inner(
|
||||
}
|
||||
let encoded = serde_json::to_vec(&Value::Object(body))
|
||||
.map_err(|_| ResponseCreateFrameError::InvalidJson)?;
|
||||
super::request::decode_request_inner(&encoded, false, request_headers)
|
||||
super::request::decode_request_inner(&encoded, false, request_headers, provider_kind)
|
||||
.map_err(ResponseCreateFrameError::Request)
|
||||
}
|
||||
|
||||
|
||||
@@ -58,8 +58,15 @@ pub(super) async fn api_router(execution: Arc<dyn ExecutionService>) -> axum::Ro
|
||||
}
|
||||
|
||||
pub(super) fn authenticated_client(plaintext: &str) -> AuthenticatedClient {
|
||||
authenticated_client_for_provider(plaintext, "openai")
|
||||
}
|
||||
|
||||
pub(super) fn authenticated_client_for_provider(
|
||||
plaintext: &str,
|
||||
provider_name: &str,
|
||||
) -> AuthenticatedClient {
|
||||
let source = DefaultExecutionService::new(
|
||||
RuntimeSnapshotHandle::new(snapshot(plaintext)),
|
||||
RuntimeSnapshotHandle::new(snapshot(plaintext, provider_name)),
|
||||
Arc::new(UnusedExecutionStore),
|
||||
ProviderRegistry::default(),
|
||||
Arc::new(UnusedAdmissions),
|
||||
@@ -71,9 +78,10 @@ pub(super) fn authenticated_client(plaintext: &str) -> AuthenticatedClient {
|
||||
.expect("authenticated client")
|
||||
}
|
||||
|
||||
fn snapshot(plaintext: &str) -> RuntimeSnapshot {
|
||||
let provider = ProviderKind::new("openai").expect("provider");
|
||||
let instance_id = ProviderInstanceId::new("inst_openai_api_test").expect("instance");
|
||||
fn snapshot(plaintext: &str, provider_name: &str) -> RuntimeSnapshot {
|
||||
let provider = ProviderKind::new(provider_name).expect("provider");
|
||||
let instance_id =
|
||||
ProviderInstanceId::new(format!("inst_{provider_name}_api_test")).expect("instance");
|
||||
let capabilities = ModelCapabilities::new(
|
||||
BTreeSet::from([gateway_core::operation::OperationKind::Generate]),
|
||||
128_000,
|
||||
|
||||
@@ -23,14 +23,14 @@ use gateway_core::event::{
|
||||
ContentItem, ContentKind, GatewayEvent, ProtocolWireEvent, ProviderEvent,
|
||||
ProviderResponseHeader, ResponseMeta, TextDelta,
|
||||
};
|
||||
use gateway_core::operation::Operation;
|
||||
use gateway_core::operation::{Operation, OperationKind};
|
||||
use gateway_core::routing::PublicModelId;
|
||||
use serde_json::{Value, json};
|
||||
|
||||
use gateway_api::openai::responses::{collect_execution_response, stream_execution_response};
|
||||
use tower::ServiceExt;
|
||||
|
||||
use crate::openai::{api_router, authenticated_client};
|
||||
use crate::openai::{api_router, authenticated_client, authenticated_client_for_provider};
|
||||
|
||||
#[derive(Default)]
|
||||
struct Trace {
|
||||
@@ -102,6 +102,8 @@ struct CapturedClientContext {
|
||||
client_ip: Option<IpAddr>,
|
||||
user_agent: Option<String>,
|
||||
endpoint: String,
|
||||
operation_kind: OperationKind,
|
||||
input: Option<Value>,
|
||||
client_metadata: Option<Value>,
|
||||
openai_options: Option<Value>,
|
||||
xai_options: Option<Value>,
|
||||
@@ -133,31 +135,42 @@ impl ExecutionService for ContextCaptureExecution {
|
||||
request: StartExecution,
|
||||
) -> BoxFuture<'_, Result<StartedExecution, GatewayError>> {
|
||||
Box::pin(async move {
|
||||
let (client_metadata, openai_options, xai_options, prompt_cache_key) =
|
||||
match &request.operation {
|
||||
Operation::Generate(generate) => (
|
||||
generate
|
||||
let operation_kind = request.operation.kind();
|
||||
let generation = match &request.operation {
|
||||
Operation::Generate(generation) => Some(generation),
|
||||
Operation::CompactConversation(compaction) => Some(compaction.generation()),
|
||||
_ => None,
|
||||
};
|
||||
let (client_metadata, openai_options, xai_options, prompt_cache_key, input) =
|
||||
generation.map_or((None, None, None, None, None), |generation| {
|
||||
(
|
||||
generation
|
||||
.protocol_payload()
|
||||
.and_then(|payload| payload.body().get("client_metadata"))
|
||||
.cloned(),
|
||||
generate
|
||||
generation
|
||||
.provider_options()
|
||||
.get("openai")
|
||||
.cloned()
|
||||
.map(Value::Object),
|
||||
generate
|
||||
generation
|
||||
.provider_options()
|
||||
.get("xai")
|
||||
.cloned()
|
||||
.map(Value::Object),
|
||||
generate.prompt_cache_key().map(ToOwned::to_owned),
|
||||
),
|
||||
_ => (None, None, None, None),
|
||||
};
|
||||
generation.prompt_cache_key().map(ToOwned::to_owned),
|
||||
generation
|
||||
.protocol_payload()
|
||||
.and_then(|payload| payload.body().get("input"))
|
||||
.cloned(),
|
||||
)
|
||||
});
|
||||
*self.observed.lock().expect("context capture lock") = Some(CapturedClientContext {
|
||||
client_ip: request.metadata.client_ip,
|
||||
user_agent: request.metadata.user_agent,
|
||||
endpoint: request.metadata.endpoint,
|
||||
operation_kind,
|
||||
input,
|
||||
client_metadata,
|
||||
openai_options,
|
||||
xai_options,
|
||||
@@ -337,8 +350,23 @@ fn delivery(event: GatewayEvent, commit_requirement: CommitRequirement) -> Coord
|
||||
}
|
||||
|
||||
async fn captured_client_context(
|
||||
mut headers: HeaderMap,
|
||||
headers: HeaderMap,
|
||||
peer_address: SocketAddr,
|
||||
) -> CapturedClientContext {
|
||||
captured_http_request(
|
||||
"openai",
|
||||
json!({"model": "smart-code", "input": "hello"}),
|
||||
headers,
|
||||
Some(peer_address),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn captured_http_request(
|
||||
provider_name: &str,
|
||||
body: Value,
|
||||
mut headers: HeaderMap,
|
||||
peer_address: Option<SocketAddr>,
|
||||
) -> CapturedClientContext {
|
||||
headers.insert(
|
||||
AUTHORIZATION,
|
||||
@@ -347,13 +375,15 @@ async fn captured_client_context(
|
||||
let observed = Arc::new(Mutex::new(None));
|
||||
let execution = Arc::new(ContextCaptureExecution {
|
||||
observed: Arc::clone(&observed),
|
||||
client: authenticated_client("sk_context_test"),
|
||||
client: authenticated_client_for_provider("sk_context_test", provider_name),
|
||||
});
|
||||
let mut request = Request::post("/v1/responses")
|
||||
.body(Body::from(r#"{"model":"smart-code","input":"hello"}"#))
|
||||
.body(Body::from(body.to_string()))
|
||||
.expect("context request");
|
||||
*request.headers_mut() = headers;
|
||||
request.extensions_mut().insert(ConnectInfo(peer_address));
|
||||
if let Some(peer_address) = peer_address {
|
||||
request.extensions_mut().insert(ConnectInfo(peer_address));
|
||||
}
|
||||
let response = api_router(execution)
|
||||
.await
|
||||
.oneshot(request)
|
||||
@@ -367,6 +397,54 @@ async fn captured_client_context(
|
||||
.expect("captured request context")
|
||||
}
|
||||
|
||||
async fn captured_http_compaction(provider_name: &str) -> CapturedClientContext {
|
||||
captured_http_request(
|
||||
provider_name,
|
||||
json!({
|
||||
"model": "smart-code",
|
||||
"input": [
|
||||
{"type": "message", "role": "user", "content": "history"},
|
||||
{"type": "compaction_trigger"}
|
||||
],
|
||||
"stream": true
|
||||
}),
|
||||
HeaderMap::new(),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn openai_http_should_preserve_compaction_trigger_as_generate() {
|
||||
let captured = captured_http_compaction("openai").await;
|
||||
|
||||
assert_eq!(
|
||||
(captured.operation_kind, captured.input),
|
||||
(
|
||||
OperationKind::Generate,
|
||||
Some(json!([
|
||||
{"type": "message", "role": "user", "content": "history"},
|
||||
{"type": "compaction_trigger"}
|
||||
])),
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn xai_http_should_consume_compaction_trigger_as_typed_operation() {
|
||||
let captured = captured_http_compaction("xai").await;
|
||||
|
||||
assert_eq!(
|
||||
(captured.operation_kind, captured.input),
|
||||
(
|
||||
OperationKind::CompactConversation,
|
||||
Some(json!([
|
||||
{"type": "message", "role": "user", "content": "history"}
|
||||
])),
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn request_context_should_resolve_forwarded_precedence_and_peer_fallback() {
|
||||
let peer = "192.0.2.10:443".parse().expect("peer address");
|
||||
@@ -385,6 +463,8 @@ async fn request_context_should_resolve_forwarded_precedence_and_peer_fallback()
|
||||
client_ip: Some("198.51.100.1".parse().expect("expected IP")),
|
||||
user_agent: Some("Codex-CLI/1.0".to_owned()),
|
||||
endpoint: "/v1/responses".to_owned(),
|
||||
operation_kind: OperationKind::Generate,
|
||||
input: Some(json!("hello")),
|
||||
client_metadata: None,
|
||||
openai_options: None,
|
||||
xai_options: None,
|
||||
|
||||
@@ -206,7 +206,7 @@ fn decoder_should_map_string_input_to_user_text() {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decoder_should_project_remote_compaction_v2_to_a_typed_operation() {
|
||||
fn decoder_should_preserve_remote_compaction_trigger_for_openai() {
|
||||
let decoded = decode_request(
|
||||
&serde_json::to_vec(&json!({
|
||||
"model": "smart-code",
|
||||
@@ -219,22 +219,15 @@ fn decoder_should_project_remote_compaction_v2_to_a_typed_operation() {
|
||||
.expect("request JSON"),
|
||||
)
|
||||
.expect("remote compaction v2 request should decode");
|
||||
let Operation::CompactConversation(request) = decoded.operation() else {
|
||||
panic!("remote compaction v2 must use the typed compact operation");
|
||||
};
|
||||
let upstream_input = request
|
||||
.generation()
|
||||
.protocol_payload()
|
||||
.expect("compaction should preserve the normalized OpenAI payload")
|
||||
.body()
|
||||
let upstream_input = openai_wire_body(&decoded)
|
||||
.get("input")
|
||||
.and_then(Value::as_array)
|
||||
.expect("compaction input should remain an array");
|
||||
.expect("OpenAI input should remain an array");
|
||||
|
||||
assert!(
|
||||
upstream_input
|
||||
.iter()
|
||||
.all(|item| { item.get("type").and_then(Value::as_str) != Some("compaction_trigger") })
|
||||
.any(|item| { item.get("type").and_then(Value::as_str) == Some("compaction_trigger") })
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -51,6 +51,33 @@ fn response_create_should_preserve_current_provider_options() {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn response_create_should_preserve_compaction_trigger_for_openai() {
|
||||
let decoded = decode_response_create(
|
||||
&json!({
|
||||
"type": "response.create",
|
||||
"model": "smart-code",
|
||||
"input": [
|
||||
{"type": "message", "role": "user", "content": "history"},
|
||||
{"type": "compaction_trigger"}
|
||||
]
|
||||
})
|
||||
.to_string(),
|
||||
)
|
||||
.expect("decode OpenAI response.create");
|
||||
let Operation::Generate(request) = decoded.operation() else {
|
||||
panic!("OpenAI response.create must remain Generate");
|
||||
};
|
||||
|
||||
assert_eq!(
|
||||
request
|
||||
.protocol_payload()
|
||||
.and_then(|payload| payload.body().get("input"))
|
||||
.and_then(|input| input.pointer("/1/type")),
|
||||
Some(&json!("compaction_trigger"))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn response_create_should_reject_explicit_non_streaming_requests() {
|
||||
let error = decode_response_create(
|
||||
|
||||
@@ -142,6 +142,7 @@ pub struct ModelCapabilities {
|
||||
features: BTreeMap<Feature, SupportLevel>,
|
||||
context_window_tokens: u64,
|
||||
max_output_tokens: Option<u64>,
|
||||
upstream_validates_features: bool,
|
||||
}
|
||||
|
||||
impl ModelCapabilities {
|
||||
@@ -156,6 +157,7 @@ impl ModelCapabilities {
|
||||
features: BTreeMap::new(),
|
||||
context_window_tokens,
|
||||
max_output_tokens,
|
||||
upstream_validates_features: false,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -165,6 +167,13 @@ impl ModelCapabilities {
|
||||
self
|
||||
}
|
||||
|
||||
/// 将请求形态 feature 的最终合法性判断交给上游 wire API。
|
||||
#[must_use]
|
||||
pub const fn with_upstream_feature_validation(mut self) -> Self {
|
||||
self.upstream_validates_features = true;
|
||||
self
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn match_requirements(
|
||||
&self,
|
||||
@@ -182,6 +191,10 @@ impl ModelCapabilities {
|
||||
return None;
|
||||
}
|
||||
|
||||
if self.upstream_validates_features {
|
||||
return Some(BTreeSet::new());
|
||||
}
|
||||
|
||||
let mut emulated = BTreeSet::new();
|
||||
for feature in requirements.features() {
|
||||
match self
|
||||
|
||||
@@ -4,7 +4,8 @@ use std::time::Duration;
|
||||
|
||||
use gateway_core::engine::credential::{AccountSelectionPolicy, RotationStrategy};
|
||||
use gateway_core::operation::{
|
||||
ContentPart, GenerateRequest, Message, MessageRole, Operation, OperationKind,
|
||||
CapabilityRequirements, ContentPart, Feature, GenerateRequest, Message, MessageRole, Operation,
|
||||
OperationKind,
|
||||
};
|
||||
use gateway_core::policy::{ClientApiKeyId, ClientPolicy, PlaintextClientApiKey, RateLimits};
|
||||
use gateway_core::routing::{
|
||||
@@ -234,7 +235,7 @@ fn blocked_instance_should_be_filtered() {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn known_unsupported_capability_should_not_be_bypassed() {
|
||||
fn known_unsupported_operation_should_not_be_bypassed() {
|
||||
let snapshot = RuntimeSnapshot::new(
|
||||
ConfigRevision::new(1).expect("revision"),
|
||||
scheduling(),
|
||||
@@ -262,6 +263,35 @@ fn known_unsupported_capability_should_not_be_bypassed() {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn upstream_feature_validation_should_preserve_operation_and_limit_gates() {
|
||||
let capabilities = ModelCapabilities::new(
|
||||
BTreeSet::from([OperationKind::Generate]),
|
||||
128_000,
|
||||
Some(16_000),
|
||||
)
|
||||
.with_feature(
|
||||
Feature::Tools,
|
||||
gateway_core::routing::SupportLevel::Unsupported,
|
||||
)
|
||||
.with_upstream_feature_validation();
|
||||
let wire_features = CapabilityRequirements::new(OperationKind::Generate)
|
||||
.require(Feature::Tools)
|
||||
.require(Feature::JsonSchema);
|
||||
let oversized =
|
||||
CapabilityRequirements::new(OperationKind::Generate).with_minimum_context_tokens(128_001);
|
||||
let unsupported_operation = CapabilityRequirements::new(OperationKind::CompactConversation);
|
||||
|
||||
assert_eq!(
|
||||
(
|
||||
capabilities.match_requirements(&wire_features),
|
||||
capabilities.match_requirements(&oversized),
|
||||
capabilities.match_requirements(&unsupported_operation),
|
||||
),
|
||||
(Some(BTreeSet::new()), None, None)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn public_catalog_should_include_discovered_models_and_aliases() {
|
||||
let models =
|
||||
|
||||
@@ -21,13 +21,11 @@ use gateway_core::event::{
|
||||
GatewayEvent, ProviderEvent, ProviderResponseHeader, ProviderResponseObservation,
|
||||
ProviderResponseTimings, UpstreamHttpVersion,
|
||||
};
|
||||
use gateway_core::operation::{
|
||||
Feature, GenerateRequest, Operation, OperationKind, ProviderSessionState,
|
||||
};
|
||||
use gateway_core::operation::{GenerateRequest, Operation, OperationKind, ProviderSessionState};
|
||||
use gateway_core::provider_ports::ProviderInstanceCatalogPort;
|
||||
use gateway_core::routing::{
|
||||
InstanceHealth, ModelCapabilities, ProviderInstance, ProviderInstanceId, ProviderKind,
|
||||
SupportLevel, UpstreamModelId,
|
||||
UpstreamModelId,
|
||||
};
|
||||
use gateway_core::task::{
|
||||
DaemonRestartPolicy, DaemonTask, ScheduledTask, WorkerContribution, WorkerCycleContext,
|
||||
@@ -1032,26 +1030,11 @@ fn compile_model_capabilities(model: &CodexCatalogModel) -> ProviderModelCapabil
|
||||
.context_window_tokens()
|
||||
.or_else(|| model.limits().max_context_window_tokens())
|
||||
.map_or(0, std::num::NonZeroU64::get);
|
||||
let mut capabilities = ModelCapabilities::new(operations, context_window, None);
|
||||
capabilities = capabilities.with_feature(
|
||||
Feature::NativeContinuation,
|
||||
support(evidence.responses_api()),
|
||||
);
|
||||
capabilities = capabilities.with_feature(Feature::Reasoning, support(evidence.reasoning()));
|
||||
capabilities =
|
||||
capabilities.with_feature(Feature::Tools, support(evidence.parallel_tool_calls()));
|
||||
capabilities = capabilities.with_feature(Feature::Vision, support(evidence.image_input()));
|
||||
let capabilities =
|
||||
ModelCapabilities::new(operations, context_window, None).with_upstream_feature_validation();
|
||||
ProviderModelCapabilities::new(model.request_model().clone(), capabilities)
|
||||
}
|
||||
|
||||
const fn support(evidence: CodexCatalogCapabilityEvidence) -> SupportLevel {
|
||||
match evidence {
|
||||
CodexCatalogCapabilityEvidence::DeclaredNative => SupportLevel::Native,
|
||||
CodexCatalogCapabilityEvidence::DeclaredUnsupported => SupportLevel::Unsupported,
|
||||
CodexCatalogCapabilityEvidence::Unknown => SupportLevel::Unknown,
|
||||
}
|
||||
}
|
||||
|
||||
fn selected_transport(
|
||||
request: &ProviderRequest,
|
||||
instance: &CodexProviderInstanceConfig,
|
||||
|
||||
@@ -19,7 +19,7 @@ use gateway_core::engine::{
|
||||
use gateway_core::error::{ProviderError, ProviderErrorKind, SafeUpstreamValue};
|
||||
use gateway_core::event::{GatewayEvent, UpstreamHttpVersion};
|
||||
use gateway_core::operation::{
|
||||
CompactConversationRequest, ContentPart, ContinuationMode, GenerateRequest, Message,
|
||||
CompactConversationRequest, ContentPart, ContinuationMode, Feature, GenerateRequest, Message,
|
||||
MessageRole, Operation, OperationKind, ProtocolPayload, ProviderOptions,
|
||||
};
|
||||
use gateway_core::routing::{
|
||||
@@ -2510,7 +2510,7 @@ async fn permission_denied_failure_keeps_every_account_ready() {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn capability_query_compiles_only_explicit_catalog_evidence() {
|
||||
async fn capability_query_keeps_model_gates_while_delegating_wire_features() {
|
||||
let server = MockServer::start().await;
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/backend-api/codex/models"))
|
||||
@@ -2560,6 +2560,26 @@ async fn capability_query_compiles_only_explicit_catalog_evidence() {
|
||||
.match_requirements(&continuation.capability_requirements())
|
||||
.is_some()
|
||||
);
|
||||
let wire_features =
|
||||
gateway_core::operation::CapabilityRequirements::new(OperationKind::Generate)
|
||||
.require(Feature::Tools)
|
||||
.require(Feature::Vision)
|
||||
.require(Feature::Reasoning)
|
||||
.require(Feature::JsonSchema);
|
||||
assert!(
|
||||
capabilities[0]
|
||||
.capabilities()
|
||||
.match_requirements(&wire_features)
|
||||
.is_some()
|
||||
);
|
||||
let oversized = gateway_core::operation::CapabilityRequirements::new(OperationKind::Generate)
|
||||
.with_minimum_context_tokens(128_001);
|
||||
assert!(
|
||||
capabilities[0]
|
||||
.capabilities()
|
||||
.match_requirements(&oversized)
|
||||
.is_none()
|
||||
);
|
||||
let compact = Operation::CompactConversation(CompactConversationRequest::new(
|
||||
GenerateRequest::new(vec![
|
||||
Message::new(
|
||||
|
||||
@@ -12,7 +12,10 @@ fn encoder_should_preserve_openai_wire_fields_without_deriving_accountless_pool_
|
||||
("model".to_owned(), json!("client-model")),
|
||||
(
|
||||
"input".to_owned(),
|
||||
json!([{"role":"user","content":"private stable prompt"}]),
|
||||
json!([
|
||||
{"role":"user","content":"private stable prompt"},
|
||||
{"type":"compaction_trigger"}
|
||||
]),
|
||||
),
|
||||
("include".to_owned(), json!(["reasoning.encrypted_content"])),
|
||||
("tool_choice".to_owned(), json!("auto")),
|
||||
@@ -42,6 +45,10 @@ fn encoder_should_preserve_openai_wire_fields_without_deriving_accountless_pool_
|
||||
);
|
||||
assert!(encoded.local_conversation_id.is_none());
|
||||
assert!(!format!("{encoded:?}").contains("private stable prompt"));
|
||||
assert_eq!(
|
||||
Value::Object(encoded.body().clone()).pointer("/input/1/type"),
|
||||
Some(&json!("compaction_trigger"))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
Reference in New Issue
Block a user