refactor(architecture): remove obsolete compatibility scaffolding

Remove orphaned fixtures, unused capability and resource branches, and stale credential APIs.

Retire legacy Usage metadata fallbacks and record the completed cleanup.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-08-29 22:08:34 +08:00
co-authored by Codex
parent ad20064d15
commit 4830ef093b
101 changed files with 160 additions and 1249 deletions
-8
View File
@@ -1446,7 +1446,6 @@ dependencies = [
"tempfile",
"thiserror",
"tokio",
"tower",
"tracing",
"tracing-appender",
"tracing-subscriber",
@@ -2449,7 +2448,6 @@ dependencies = [
"tracing-subscriber",
"tungstenite 0.27.0",
"url",
"urlencoding",
"uuid",
"wiremock",
"zstd",
@@ -3901,12 +3899,6 @@ dependencies = [
"serde",
]
[[package]]
name = "urlencoding"
version = "2.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"
[[package]]
name = "utf-8"
version = "0.7.6"
@@ -37,11 +37,6 @@ impl AdminAuth {
pub const fn context(&self) -> &AdminRequestContext {
&self.context
}
#[must_use]
pub fn into_context(self) -> AdminRequestContext {
self.context
}
}
impl<S> FromRequestParts<S> for AdminAuth
@@ -136,12 +136,6 @@ pub enum ContinuationIntent {
}
impl ContinuationIntent {
/// 返回是否请求了对话延续。
#[must_use]
pub const fn is_continuation(&self) -> bool {
matches!(self, Self::PreviousResponseId(_))
}
/// 返回待 history owner 解析的 response ID。
#[must_use]
pub fn previous_response_id(&self) -> Option<&str> {
@@ -107,14 +107,6 @@ fn test_tree_should_match_frozen_rust_mirror() {
assert_eq!(actual, expected);
}
#[test]
fn frozen_openai_fixture_set_should_be_complete() {
let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/openai/responses/fixtures");
let files = all_files(&root);
assert_eq!(files.len(), 69);
}
#[test]
fn admin_routes_should_use_only_get_post_and_static_paths() {
let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("src/admin");
@@ -1,26 +0,0 @@
{
"type": "response.create",
"model": "gpt-5.5",
"instructions": "answer briefly",
"input": [
{
"type": "reasoning",
"id": "rs_implicit_resume",
"status": "completed",
"summary": [
{
"type": "summary_text",
"text": "cached reasoning"
}
],
"encrypted_content": "enc_reasoning_replay"
},
{
"role": "user",
"content": "continue"
}
],
"store": false,
"stream": false,
"previous_response_id": "resp_implicit_resume_first"
}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"auth fallback hello"}
event: response.completed
data: {"response":{"id":"resp_after_401","object":"response","status":"completed","usage":{"input_tokens":3,"output_tokens":1}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"quota fallback hello"}
event: response.completed
data: {"response":{"id":"resp_after_402","object":"response","status":"completed","usage":{"input_tokens":6,"output_tokens":2}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"banned fallback hello"}
event: response.completed
data: {"response":{"id":"resp_after_403","object":"response","status":"completed","usage":{"input_tokens":5,"output_tokens":1}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"retry hello"}
event: response.completed
data: {"response":{"id":"resp_after_5xx_retry","object":"response","status":"completed","usage":{"input_tokens":4,"output_tokens":2}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"cloudflare fallback hello"}
event: response.completed
data: {"response":{"id":"resp_after_cloudflare","object":"response","status":"completed","usage":{"input_tokens":5,"output_tokens":2}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"model fallback hello"}
event: response.completed
data: {"response":{"id":"resp_after_model_unsupported","object":"response","status":"completed","usage":{"input_tokens":4,"output_tokens":2}}}
@@ -1,6 +0,0 @@
event: response.output_text.delta
data: {"delta":"hello"}
event: response.completed
data: {"response":{"usage":{"input_tokens":2,"output_tokens":3}}}
@@ -1,6 +0,0 @@
event: response.output_text.delta
data: {"delta":"hello"}
event: response.completed
data: {"response":{"id":"resp_1","usage":{"input_tokens":2,"output_tokens":3}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"type":"response.output_text.delta","delta":"hello"}
event: response.incomplete
data: {"type":"response.incomplete","response":{"id":"resp_incomplete","object":"response","status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"usage":{"input_tokens":2,"output_tokens":3},"output":[{"type":"message","status":"incomplete","role":"assistant","content":[{"type":"output_text","text":"hello","annotations":[]}]}],"output_text":"hello"}}
@@ -1,9 +0,0 @@
event: response.reasoning_summary_text.delta
data: {"delta":"I considered the context."}
event: response.output_text.delta
data: {"delta":"hello"}
event: response.completed
data: {"response":{"id":"resp_1","output_text":"hello","usage":{"input_tokens":9,"output_tokens":3}}}
@@ -1,3 +0,0 @@
event: response.completed
data: {}
@@ -1,5 +0,0 @@
event: response.completed
data: {}
data: [DONE]
@@ -1,3 +0,0 @@
event: response.completed
data: {"response":{"id":"resp_1","usage":{"input_tokens":12,"output_tokens":5,"input_tokens_details":{"cached_tokens":3}},"tool_usage":{"image_gen":{"input_tokens":31,"output_tokens":9}}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"image done"}
event: response.completed
data: {"response":{"id":"resp_image","object":"response","status":"completed","output_text":"image done","usage":{"input_tokens":7,"output_tokens":4,"input_tokens_details":{"cached_tokens":2}},"tool_usage":{"image_gen":{"input_tokens":31,"output_tokens":9}}}}
@@ -1,3 +0,0 @@
event: response.completed
data: {"response":{"id":"resp_order","status":"completed","usage":{"input_tokens":1,"output_tokens":1}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"usage hello"}
event: response.completed
data: {"response":{"id":"resp_usage","object":"response","status":"completed","output_text":"usage hello","usage":{"input_tokens":7,"output_tokens":4,"input_tokens_details":{"cached_tokens":2}}}}
@@ -1,3 +0,0 @@
event: response.completed
data: {"response":{"id":"resp_1","usage":{"input_tokens":2,"output_tokens":3,"input_tokens_details":{"cached_tokens":1}}}}
@@ -1,6 +0,0 @@
event: response.created
data: {"response":{"id":"resp_1","usage":{"input_tokens":3,"output_tokens":5,"input_tokens_details":{"cached_tokens":1}}}}
event: response.completed
data: {"response":{"id":"resp_1","usage":{"input_tokens":3,"output_tokens":5,"input_tokens_details":{"cached_tokens":1}}}}
@@ -1,2 +0,0 @@
data: {"type":"response.output_text.delta","delta":"hello"}
@@ -1,5 +0,0 @@
event: response.output_item.done
data: {"item":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"from done item"}]}}
event: response.completed
data: {"response":{"id":"resp_item","object":"response","status":"completed","output":[],"usage":{"input_tokens":2,"output_tokens":3}}}
@@ -1,2 +0,0 @@
data: [DONE]
@@ -1,2 +0,0 @@
event: response.completed
data: {"response":{"id":"resp_empty","object":"response","status":"completed"}}
@@ -1,3 +0,0 @@
event: response.output_text.delta
data: {"delta":""}
@@ -1,2 +0,0 @@
event: response.failed
data: {"response":{"id":"resp_sse_auth_failed","object":"response","status":"failed","error":{"code":"token_invalid","message":"token revoked"}}}
@@ -1,3 +0,0 @@
event: response.failed
data: {"type":"response.failed","response":{"id":"resp_cyber_policy","object":"response","status":"failed","error":{"code":"cyber_policy","message":"This request has been flagged for possible cybersecurity risk."}}}
@@ -1,2 +0,0 @@
event: response.failed
data: {"response":{"id":"resp_model_unsupported_failed","object":"response","status":"failed","error":{"code":"model_not_supported","message":"Model gpt-5.5 is not supported on this account plan"}}}
@@ -1,2 +0,0 @@
event: response.failed
data: {"response":{"id":"resp_sse_quota_failed","object":"response","status":"failed","error":{"code":"quota_exceeded","message":"quota exhausted"}}}
@@ -1,2 +0,0 @@
event: response.failed
data: {"response":{"id":"resp_server_overloaded","object":"response","status":"failed","error":{"code":"server_is_overloaded","message":"upstream is temporarily overloaded"}}}
@@ -1,3 +0,0 @@
event: response.function_call_arguments.delta
data: {"delta":"{\"path\""}
@@ -1,2 +0,0 @@
event: response.output_text.delta
data: {"delta":"hello"}
@@ -1,3 +0,0 @@
event: response.completed
data: {"response":{"id":"resp_live_stream","object":"response","status":"completed","usage":{"input_tokens":3,"output_tokens":4}}}
@@ -1,2 +0,0 @@
event: response.failed
data: {"response":{"id":"resp_live_auth_failed","object":"response","status":"failed","error":{"code":"token_invalid","message":"live token expired"}}}
@@ -1,2 +0,0 @@
event: response.failed
data: {"response":{"id":"resp_live_quota_failed","object":"response","status":"failed","error":{"code":"quota_exceeded","message":"live quota exhausted"}}}
@@ -1,3 +0,0 @@
event: response.output_text.delta
data: {"delta":"live stream hello"}
@@ -1,9 +0,0 @@
event: codex.rate_limits
data: {"type":"codex.rate_limits","rate_limits":{"primary":{"used_percent":42}}}
event: response.metadata
data: {"type":"response.metadata","x-codex-turn-state":"state"}
event: response.in_progress
data: {"type":"response.in_progress","response":{"id":"resp_1","status":"in_progress"}}
@@ -1,8 +0,0 @@
event: response.created
data: {"type":"response.created","response":{"id":"resp_http_cyber_policy","status":"in_progress"}}
event: response.output_text.delta
data: {"type":"response.output_text.delta","response_id":"resp_http_cyber_policy","output_index":0,"content_index":0,"delta":"partial output before HTTP SSE policy failure"}
event: response.failed
data: {"type":"response.failed","response":{"id":"resp_http_cyber_policy","status":"failed","error":{"code":"cyber_policy","message":"This request has been flagged for possible cybersecurity risk."}}}
@@ -1,8 +0,0 @@
event: response.created
data: {"type":"response.created","response":{"id":"resp_http_output_then_missing","status":"in_progress"}}
event: response.output_text.delta
data: {"type":"response.output_text.delta","response_id":"resp_http_output_then_missing","output_index":0,"content_index":0,"delta":"visible HTTP SSE output"}
event: response.failed
data: {"type":"response.failed","response":{"id":"resp_http_output_then_missing","status":"failed","error":{"code":"previous_response_not_found","message":"Previous response was not found"}}}
@@ -1,6 +0,0 @@
event: response.created
data: {"response":{"id":"resp_clean_close"}}
event: response.output_text.delta
data: {"delta":"partial before clean close"}
@@ -1,3 +0,0 @@
event: response.output_text.delta
data: {"delta":"partial before logged disconnect"}
@@ -1,3 +0,0 @@
event: response.output_text.delta
data: {"delta":"partial before transport failure"}
@@ -1,3 +0,0 @@
event: response.reasoning_summary_text.delta
data: {"delta":"thinking"}
@@ -1,3 +0,0 @@
event: response.created
data: {"type":"response.created","response":{"id":"resp_1","status":"in_progress"}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"auth stream fallback hello"}
event: response.completed
data: {"response":{"id":"resp_stream_after_401","object":"response","status":"completed","usage":{"input_tokens":4,"output_tokens":2}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"quota stream fallback hello"}
event: response.completed
data: {"response":{"id":"resp_stream_after_402","object":"response","status":"completed","usage":{"input_tokens":6,"output_tokens":3}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"banned stream fallback hello"}
event: response.completed
data: {"response":{"id":"resp_stream_after_403","object":"response","status":"completed","usage":{"input_tokens":5,"output_tokens":2}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"rate fallback hello"}
event: response.completed
data: {"response":{"id":"resp_stream_after_429","object":"response","status":"completed","usage":{"input_tokens":4,"output_tokens":1}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"retry stream hello"}
event: response.completed
data: {"response":{"id":"resp_stream_after_5xx_retry","object":"response","status":"completed","usage":{"input_tokens":7,"output_tokens":2}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"cloudflare stream fallback hello"}
event: response.completed
data: {"response":{"id":"resp_stream_after_cloudflare","object":"response","status":"completed","usage":{"input_tokens":6,"output_tokens":2}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"model stream fallback hello"}
event: response.completed
data: {"response":{"id":"resp_stream_after_model_unsupported","object":"response","status":"completed","usage":{"input_tokens":5,"output_tokens":2}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"stream hello"}
event: response.completed
data: {"response":{"id":"resp_stream_usage","object":"response","status":"completed","usage":{"input_tokens":3,"output_tokens":5}}}
@@ -1,8 +0,0 @@
event: response.created
data: {"type":"response.created","response":{"id":"resp_http_previous_missing","status":"in_progress"}}
event: response.content_part.added
data: {"type":"response.content_part.added","part":{"type":"output_text","text":""}}
event: response.failed
data: {"type":"response.failed","response":{"id":"resp_http_previous_missing","status":"failed","error":{"code":"previous_response_not_found","message":"Previous response was not found"}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"delta":"response hello"}
event: response.completed
data: {"response":{"id":"resp_response_1","object":"response","status":"completed","usage":{"input_tokens":5,"output_tokens":2}}}
@@ -1,6 +0,0 @@
event: response.output_text.delta
data: {"delta":"hello"}
event: response.completed
data: {"response":{"id":"resp_1","object":"response","status":"completed","usage":{"input_tokens":2,"output_tokens":3}}}
@@ -1,3 +0,0 @@
event: response.output_text.delta
data: {"type":"response.output_text.delta","delta":"hello"}
@@ -1,3 +0,0 @@
event: response.output_text.delta
data: {"delta":"hello"}
@@ -1,8 +0,0 @@
event: response.output_text.delta
data: {"delta":"hello "}
event: response.output_text.delta
data: {"delta":"from deltas"}
event: response.completed
data: {"response":{"id":"resp_text","object":"response","status":"completed","output":[],"usage":{"input_tokens":2,"output_tokens":3}}}
@@ -1,5 +0,0 @@
event: response.output_text.delta
data: {"type":"response.output_text.delta","delta":"{\"point\":{\"0\":1,\"1\":2}}"}
event: response.completed
data: {"response":{"id":"resp_tuple","object":"response","status":"completed","output_text":"{\"point\":{\"0\":0,\"1\":0}}","output":[{"type":"message","role":"assistant","content":[{"type":"output_text","text":"{\"point\":{\"0\":1,\"1\":2}}"}]}],"usage":{"input_tokens":1,"output_tokens":1}}}
@@ -1,6 +0,0 @@
event: message
id: evt_1
data: hello
data: world
retry: 10
@@ -1,32 +0,0 @@
{
"type": "response.completed",
"response": {
"id": "resp_implicit_resume_first",
"object": "response",
"output": [
{
"type": "reasoning",
"id": "rs_implicit_resume",
"status": "completed",
"summary": [
{
"type": "summary_text",
"text": "cached reasoning"
}
],
"content": [
{
"type": "reasoning_text",
"text": "raw reasoning should not be replayed"
}
],
"encrypted_content": "enc_reasoning_replay"
}
],
"usage": {
"input_tokens": 8,
"output_tokens": 2,
"total_tokens": 10
}
}
}
@@ -1 +0,0 @@
{"error":{"message":"first account limited"}}
@@ -1 +0,0 @@
{"error":{"message":"history account rate limited"}}
@@ -1 +0,0 @@
{"type":"response.failed","response":{"id":"resp_invalid_encrypted_content","object":"response","status":"failed","error":{"code":"invalid_encrypted_content","message":"Invalid encrypted content in reasoning replay"}},"error":{"code":"invalid_encrypted_content","message":"Invalid encrypted content in reasoning replay"}}
@@ -1,7 +0,0 @@
{
"type": "error",
"error": {
"code": "previous_response_not_found",
"message": "Previous response with id resp_missing was not found"
}
}
@@ -1 +0,0 @@
{"error":{"message":"rate limited"}}
@@ -1 +0,0 @@
{"error":{"message":"second account limited"}}
@@ -1 +0,0 @@
{"error":{"code":"token_revoked"}}
@@ -1,7 +0,0 @@
{
"type": "error",
"error": {
"code": "invalid_request",
"message": "No tool output found for function call call_missing"
}
}
@@ -22,7 +22,7 @@ use crate::error::{GatewayError, GatewayErrorKind, ProviderError, ProviderErrorK
use crate::event::{
GatewayEvent, ProviderEvent, ProviderResponseHeader, ProviderResponseObservation,
};
use crate::operation::{Operation, ProviderSessionState, RetrySafety};
use crate::operation::{Operation, ProviderSessionState};
use crate::routing::RoutingPlan;
use futures::future::Fuse;
use futures::{FutureExt, StreamExt, pin_mut, select_biased};
@@ -214,7 +214,6 @@ struct FailureFinalization {
upstream_status_code: Option<u16>,
provider_error_code: Option<String>,
retry_after_ms: Option<u64>,
provider_response_id: Option<String>,
}
/// API 可逐事件消费的 Core 执行会话。
@@ -477,7 +476,7 @@ where
if state.provider() != provider.as_str() {
return None;
}
let account = current.metadata.provider_account_id()?.clone();
let account = current.metadata.provider_account_id().clone();
let response_id = self.upstream_response_id.as_deref()?;
let previous_response_id = PreviousResponseId::new(response_id.to_owned());
let upstream_response_id = PreviousResponseId::new(response_id.to_owned());
@@ -602,7 +601,6 @@ where
upstream_status_code: None,
provider_error_code: None,
retry_after_ms: None,
provider_response_id: None,
})
.await?;
return Err(EngineError::EmptyRoutingPlan);
@@ -749,7 +747,6 @@ where
upstream_status_code: None,
provider_error_code: None,
retry_after_ms: None,
provider_response_id: None,
})
.await?;
return Err(EngineError::ProviderMetadataMismatch);
@@ -757,9 +754,9 @@ where
let metadata = stream.metadata().clone();
if !self.account_selection.is_diagnostic()
&& metadata
.provider_account_id()
.is_none_or(|account| !candidate.account_scope().allows(account))
&& !candidate
.account_scope()
.allows(metadata.provider_account_id())
{
let error = GatewayError::new(
GatewayErrorKind::Internal,
@@ -772,14 +769,13 @@ where
upstream_status_code: None,
provider_error_code: None,
retry_after_ms: None,
provider_response_id: None,
})
.await?;
return Err(EngineError::AccountOutsideClientScope);
}
if pinned_account
.as_ref()
.is_some_and(|required| metadata.provider_account_id() != Some(required))
.is_some_and(|required| metadata.provider_account_id() != required)
{
let error = GatewayError::new(
GatewayErrorKind::Internal,
@@ -792,7 +788,6 @@ where
upstream_status_code: None,
provider_error_code: None,
retry_after_ms: None,
provider_response_id: None,
})
.await?;
return Err(EngineError::RequiredAccountMismatch);
@@ -802,9 +797,7 @@ where
.as_ref()
.and_then(ContinuationBinding::pinned)
&& self.continuation_attempt == ContinuationAttempt::Native
&& !metadata
.provider_account_id()
.is_some_and(|account| pin.matches(metadata.provider(), account))
&& !pin.matches(metadata.provider(), metadata.provider_account_id())
{
let error = GatewayError::new(
GatewayErrorKind::Internal,
@@ -817,17 +810,14 @@ where
upstream_status_code: None,
provider_error_code: None,
retry_after_ms: None,
provider_response_id: None,
})
.await?;
return Err(EngineError::ContinuationPinMismatch);
}
if self.account_state_owner.is_none()
&& let Some(account) = metadata.provider_account_id()
{
if self.account_state_owner.is_none() {
self.account_state_owner = Some(ProviderAccountStateOwner::new(
metadata.provider().clone(),
account.clone(),
metadata.provider_account_id().clone(),
));
}
let attempt_record = AttemptRecord {
@@ -835,8 +825,8 @@ where
attempt_count: next_attempt,
trigger,
provider_kind: metadata.provider().clone(),
provider_account_id: metadata.provider_account_id().cloned(),
provider_account_ref: metadata.provider_account_id().cloned(),
provider_account_id: Some(metadata.provider_account_id().clone()),
provider_account_ref: Some(metadata.provider_account_id().clone()),
upstream_model_id: metadata.upstream_model().cloned(),
upstream_transport: metadata.transport().as_str().to_owned(),
http_version: None,
@@ -1032,8 +1022,7 @@ where
&& self.downstream_committed_at.is_none()
&& !self.delivery_pending
&& attempt_send_state != UpstreamSendState::Ambiguous
&& (self.operation.retry_safety() == RetrySafety::Idempotent
|| provider_proved_replay_safe)
&& provider_proved_replay_safe
&& self.attempts < self.plan.max_attempts().get();
let same_account_retry = error.retries_same_account()
&& provider_proved_replay_safe
@@ -1041,14 +1030,9 @@ where
&& !self.delivery_pending
&& attempt_send_state != UpstreamSendState::Ambiguous
&& self.attempts < self.plan.max_attempts().get()
&& current
.metadata
.provider_account_id()
.is_some_and(|account| {
!self
.credential_recovery_attempted_accounts
.contains(account)
});
&& !self
.credential_recovery_attempted_accounts
.contains(current.metadata.provider_account_id());
let retryable =
continuation_retry || same_account_retry || ordinary_retry || pre_delivery_retry;
@@ -1057,17 +1041,15 @@ where
// 所有权保留到下一次 attempt 成功,或最终空选路时返回客户端。
let persistence_error = self.request_persisted.then(|| error.clone());
if same_account_retry {
if let Some(account) = current.metadata.provider_account_id() {
self.credential_recovery_attempted_accounts
.insert(account.clone());
// 只钉住紧随其后的 replay attempt;replay 再遇可重试错误时,
// ordinary/continuation 重试门不受影响,仍可换号。
self.recovery_account = Some(account.clone());
}
} else if !continuation_retry
&& let Some(account) = current.metadata.provider_account_id()
{
self.excluded_accounts.insert(account.clone());
let account = current.metadata.provider_account_id().clone();
self.credential_recovery_attempted_accounts
.insert(account.clone());
// 只钉住紧随其后的 replay attempt;replay 再遇可重试错误时,
// ordinary/continuation 重试门不受影响,仍可换号。
self.recovery_account = Some(account);
} else if !continuation_retry {
self.excluded_accounts
.insert(current.metadata.provider_account_id().clone());
}
self.last_retryable_failure_events = atomic_client_events;
self.last_retryable_failure = Some(error);
@@ -1082,7 +1064,7 @@ where
attempt_index: current.index,
trigger: current.trigger,
provider_kind: current.metadata.provider().clone(),
account_id: current.metadata.provider_account_id().cloned(),
account_id: Some(current.metadata.provider_account_id().clone()),
upstream_model_id: current.metadata.upstream_model().cloned(),
upstream_status_code: current
.response_observation
@@ -1157,14 +1139,12 @@ where
}
ContinuationAttempt::Native | ContinuationAttempt::ReplayOwner => {
self.continuation_attempt = ContinuationAttempt::ReplayAny;
if let Some(account) = current.metadata.provider_account_id() {
self.excluded_accounts.insert(account.clone());
}
self.excluded_accounts
.insert(current.metadata.provider_account_id().clone());
}
ContinuationAttempt::ReplayAny => {
if let Some(account) = current.metadata.provider_account_id() {
self.excluded_accounts.insert(account.clone());
}
self.excluded_accounts
.insert(current.metadata.provider_account_id().clone());
}
ContinuationAttempt::None => return false,
}
@@ -1317,9 +1297,6 @@ where
upstream_status_code: error.upstream_status(),
provider_error_code: error.upstream_code().map(|code| code.as_str().to_owned()),
retry_after_ms: error.retry_after().map(duration_ms),
provider_response_id: error
.upstream_response_id()
.map(|value| value.as_str().to_owned()),
})
.await
}
@@ -1373,7 +1350,6 @@ where
upstream_status_code: None,
provider_error_code: None,
retry_after_ms: None,
provider_response_id: None,
})
.await
}
@@ -1433,9 +1409,7 @@ where
.or(observed_status_code),
client_response_id: self.client_response_id.clone(),
upstream_request_id,
upstream_response_id: finalization
.provider_response_id
.or_else(|| self.upstream_response_id.clone()),
upstream_response_id: self.upstream_response_id.clone(),
upstream_transport,
http_version,
websocket_pool,
@@ -11,9 +11,7 @@ use async_trait::async_trait;
use futures::Stream;
use thiserror::Error;
use crate::engine::credential::{
AccountAttemptFeedback, AccountFeedbackStats, CredentialRevision, ProviderAccountId,
};
use crate::engine::credential::{AccountAttemptFeedback, AccountFeedbackStats, ProviderAccountId};
use crate::engine::{AttemptContext, UpstreamSendState};
use crate::error::{
IdentifierError, OpaqueUpstreamValue, ProviderError, ProviderErrorKind, validate_text,
@@ -28,74 +26,6 @@ use crate::routing::{
pub type EventStream =
Pin<Box<dyn Stream<Item = Result<ProviderEvent, ProviderError>> + Send + 'static>>;
/// 匿名 credential/resource 引用。
///
/// 有 credential 的值必须是带算法/版本标识的不可逆伪名;无 credential 的
/// Provider 使用 `__none__`。邮箱、API Key prefix 等身份信息不能进入该类型。
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ResourceId(String);
impl ResourceId {
/// 无 credential 的稳定 sentinel。
#[must_use]
pub fn none() -> Self {
Self("__none__".to_owned())
}
/// 创建匿名资源 ID。
///
/// # Errors
///
/// ID 为空、过长、含控制字符或使用 `__` 保留前缀时返回错误。
pub fn anonymous(value: impl Into<String>) -> Result<Self, IdentifierError> {
let value = value.into();
validate_text(&value, 128, true, None)?;
Ok(Self(value))
}
/// 返回匿名引用。
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
/// Provider 本次调用实际使用的账号或匿名资源。
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ProviderResource {
Account {
id: ProviderAccountId,
revision: CredentialRevision,
},
Anonymous(ResourceId),
}
impl ProviderResource {
#[must_use]
pub const fn account_id(&self) -> Option<&ProviderAccountId> {
match self {
Self::Account { id, .. } => Some(id),
Self::Anonymous(_) => None,
}
}
#[must_use]
pub const fn credential_revision(&self) -> Option<CredentialRevision> {
match self {
Self::Account { revision, .. } => Some(*revision),
Self::Anonymous(_) => None,
}
}
#[must_use]
pub const fn anonymous_id(&self) -> Option<&ResourceId> {
match self {
Self::Account { .. } => None,
Self::Anonymous(resource) => Some(resource),
}
}
}
/// 上游 transport 注册名称。
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct UpstreamTransport(String);
@@ -124,7 +54,7 @@ impl UpstreamTransport {
pub struct ProviderCallMetadata {
provider: ProviderKind,
upstream_model: Option<UpstreamModelId>,
resource: ProviderResource,
provider_account_id: ProviderAccountId,
upstream_request_id: Option<OpaqueUpstreamValue>,
transport: UpstreamTransport,
}
@@ -135,13 +65,13 @@ impl ProviderCallMetadata {
pub const fn new(
provider: ProviderKind,
upstream_model: UpstreamModelId,
resource: ProviderResource,
provider_account_id: ProviderAccountId,
transport: UpstreamTransport,
) -> Self {
Self {
provider,
upstream_model: Some(upstream_model),
resource,
provider_account_id,
upstream_request_id: None,
transport,
}
@@ -151,13 +81,13 @@ impl ProviderCallMetadata {
#[must_use]
pub const fn for_provider_endpoint(
provider: ProviderKind,
resource: ProviderResource,
provider_account_id: ProviderAccountId,
transport: UpstreamTransport,
) -> Self {
Self {
provider,
upstream_model: None,
resource,
provider_account_id,
upstream_request_id: None,
transport,
}
@@ -182,22 +112,10 @@ impl ProviderCallMetadata {
self.upstream_model.as_ref()
}
/// 返回匿名资源;账号调用返回 `None`。
#[must_use]
pub const fn anonymous_resource(&self) -> Option<&ResourceId> {
self.resource.anonymous_id()
}
/// 返回 live Provider account ID。
#[must_use]
pub const fn provider_account_id(&self) -> Option<&ProviderAccountId> {
self.resource.account_id()
}
/// 返回本次冻结的 credential revision。
#[must_use]
pub const fn upstream_credential_revision(&self) -> Option<CredentialRevision> {
self.resource.credential_revision()
pub const fn provider_account_id(&self) -> &ProviderAccountId {
&self.provider_account_id
}
/// 返回安全上游 request ID。
@@ -334,16 +252,14 @@ impl ProviderStream {
/// 让公共 stream 边界统一回灌账号成功率与首个有效输出延迟。
#[must_use]
pub fn with_account_feedback(mut self, stats: Arc<AccountFeedbackStats>) -> Self {
if let Some(account_id) = self.metadata.provider_account_id().cloned() {
self.account_feedback = Some(ProviderStreamAccountFeedback {
stats,
provider_kind: self.metadata.provider().clone(),
account_id,
started_at: None,
first_output_ms: None,
reported: false,
});
}
self.account_feedback = Some(ProviderStreamAccountFeedback {
stats,
provider_kind: self.metadata.provider().clone(),
account_id: self.metadata.provider_account_id().clone(),
started_at: None,
first_output_ms: None,
reported: false,
});
self
}
-20
View File
@@ -373,7 +373,6 @@ pub struct ProviderError {
struct ProviderErrorUpstreamValues {
code: Option<OpaqueUpstreamValue>,
request_id: Option<OpaqueUpstreamValue>,
response_id: Option<OpaqueUpstreamValue>,
}
#[derive(Clone)]
@@ -424,13 +423,6 @@ impl ProviderError {
self
}
/// 附加 adapter 已分类为非 bearer 的上游 response ID。
#[must_use]
pub fn with_upstream_response_id(mut self, response_id: OpaqueUpstreamValue) -> Self {
self.upstream_values_mut().response_id = Some(response_id);
self
}
fn upstream_values_mut(&mut self) -> &mut ProviderErrorUpstreamValues {
self.upstream_values
.get_or_insert_with(|| Box::new(ProviderErrorUpstreamValues::default()))
@@ -551,14 +543,6 @@ impl ProviderError {
.and_then(|values| values.request_id.as_ref())
}
/// 返回安全分类的上游 response ID。
#[must_use]
pub fn upstream_response_id(&self) -> Option<&OpaqueUpstreamValue> {
self.upstream_values
.as_deref()
.and_then(|values| values.response_id.as_ref())
}
/// 返回 Provider 建议的冷却时间。
#[must_use]
pub const fn retry_after(&self) -> Option<Duration> {
@@ -666,10 +650,6 @@ impl fmt::Debug for ProviderError {
"upstream_request_id",
&self.upstream_request_id().map(|_| "<classified-safe>"),
)
.field(
"upstream_response_id",
&self.upstream_response_id().map(|_| "<classified-safe>"),
)
.field("retry_after", &self.retry_after)
.field("continuation_failure", &self.continuation_failure)
.field("replay_safe", &self.replay_safe)
@@ -20,14 +20,8 @@ use crate::error::{OperationError, validate_text};
pub enum OperationKind {
/// 文本、多模态和工具生成。
Generate,
/// 向量嵌入。
Embed,
/// 文档重排。
Rerank,
/// 图像生成。
GenerateImage,
/// 语音生成。
Speech,
}
impl OperationKind {
@@ -36,23 +30,11 @@ impl OperationKind {
pub const fn as_str(self) -> &'static str {
match self {
Self::Generate => "generate",
Self::Embed => "embed",
Self::Rerank => "rerank",
Self::GenerateImage => "generate_image",
Self::Speech => "speech",
}
}
}
/// 请求在 commit 前是否允许跨目标重放。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum RetrySafety {
/// 相同业务 payload 可以安全重放。
Idempotent,
/// 默认禁止跨 Provider fallback。
NonIdempotent,
}
/// Router 理解的稳定能力。
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[non_exhaustive]
@@ -473,101 +455,6 @@ fn contains_type(value: Option<&Value>, expected: &str) -> bool {
}
}
/// Embedding 请求。
#[derive(Clone, PartialEq, Eq)]
pub struct EmbedRequest {
input: Vec<String>,
dimensions: Option<u32>,
}
impl EmbedRequest {
/// 创建 embedding 请求。
///
/// # Errors
///
/// 输入为空时返回错误。
pub fn new(input: Vec<String>) -> Result<Self, OperationError> {
if input.is_empty() || input.iter().any(String::is_empty) {
return Err(OperationError::EmptyField { field: "input" });
}
Ok(Self {
input,
dimensions: None,
})
}
/// 设置输出维度。
#[must_use]
pub const fn with_dimensions(mut self, dimensions: u32) -> Self {
self.dimensions = Some(dimensions);
self
}
/// 返回输入文本。
#[must_use]
pub fn input(&self) -> &[String] {
&self.input
}
}
impl fmt::Debug for EmbedRequest {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("EmbedRequest")
.field("input_items", &self.input.len())
.field("dimensions", &self.dimensions)
.finish()
}
}
/// Rerank 请求。
#[derive(Clone, PartialEq, Eq)]
pub struct RerankRequest {
query: String,
documents: Vec<String>,
top_n: Option<u32>,
}
impl RerankRequest {
/// 创建 rerank 请求。
///
/// # Errors
///
/// Query 或文档为空时返回错误。
pub fn new(query: impl Into<String>, documents: Vec<String>) -> Result<Self, OperationError> {
let query = query.into();
if query.is_empty() {
return Err(OperationError::EmptyField { field: "query" });
}
if documents.is_empty() || documents.iter().any(String::is_empty) {
return Err(OperationError::EmptyField { field: "documents" });
}
Ok(Self {
query,
documents,
top_n: None,
})
}
/// 设置返回条数。
#[must_use]
pub const fn with_top_n(mut self, top_n: u32) -> Self {
self.top_n = Some(top_n);
self
}
}
impl fmt::Debug for RerankRequest {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("RerankRequest")
.field("query", &"<not included in Debug>")
.field("documents", &self.documents.len())
.field("top_n", &self.top_n)
.finish()
}
}
/// 图像 API 的稳定端点语义。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum ImageRequestKind {
@@ -614,56 +501,14 @@ impl fmt::Debug for ImageRequest {
}
}
/// 语音生成请求。
#[derive(Clone, PartialEq, Eq)]
pub struct SpeechRequest {
input: String,
voice: String,
}
impl SpeechRequest {
/// 创建语音生成请求。
///
/// # Errors
///
/// 输入或 voice 为空时返回错误。
pub fn new(input: impl Into<String>, voice: impl Into<String>) -> Result<Self, OperationError> {
let input = input.into();
let voice = voice.into();
if input.is_empty() {
return Err(OperationError::EmptyField { field: "input" });
}
if voice.is_empty() {
return Err(OperationError::EmptyField { field: "voice" });
}
Ok(Self { input, voice })
}
}
impl fmt::Debug for SpeechRequest {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("SpeechRequest")
.field("input", &"<not included in Debug>")
.field("voice", &self.voice)
.finish()
}
}
/// 网关内部业务请求;不包含任何客户端 wire 或 Provider SDK 类型。
#[derive(Clone, PartialEq)]
#[non_exhaustive]
pub enum Operation {
/// 生成。
Generate(GenerateRequest),
/// Embedding。
Embed(EmbedRequest),
/// Rerank。
Rerank(RerankRequest),
/// 图像生成。
GenerateImage(ImageRequest),
/// 语音生成。
Speech(SpeechRequest),
}
impl Operation {
@@ -688,10 +533,7 @@ impl Operation {
pub const fn kind(&self) -> OperationKind {
match self {
Self::Generate(_) => OperationKind::Generate,
Self::Embed(_) => OperationKind::Embed,
Self::Rerank(_) => OperationKind::Rerank,
Self::GenerateImage(_) => OperationKind::GenerateImage,
Self::Speech(_) => OperationKind::Speech,
}
}
@@ -700,21 +542,7 @@ impl Operation {
pub fn capability_requirements(&self) -> CapabilityRequirements {
match self {
Self::Generate(request) => request.requirements(),
Self::Embed(_) => CapabilityRequirements::new(OperationKind::Embed),
Self::Rerank(_) => CapabilityRequirements::new(OperationKind::Rerank),
Self::GenerateImage(_) => CapabilityRequirements::new(OperationKind::GenerateImage),
Self::Speech(_) => CapabilityRequirements::new(OperationKind::Speech),
}
}
/// 返回跨目标重放安全性。
#[must_use]
pub const fn retry_safety(&self) -> RetrySafety {
match self {
Self::Embed(_) | Self::Rerank(_) => RetrySafety::Idempotent,
Self::Generate(_) | Self::GenerateImage(_) | Self::Speech(_) => {
RetrySafety::NonIdempotent
}
}
}
@@ -724,7 +552,6 @@ impl Operation {
match self {
Self::Generate(request) => request.image_generation_requested(),
Self::GenerateImage(_) => true,
Self::Embed(_) | Self::Rerank(_) | Self::Speech(_) => false,
}
}
@@ -733,7 +560,7 @@ impl Operation {
pub fn provider_session_state(&self, provider: &str) -> Option<&ProviderSessionState> {
match self {
Self::Generate(request) => request.provider_session_state(provider),
Self::Embed(_) | Self::Rerank(_) | Self::GenerateImage(_) | Self::Speech(_) => None,
Self::GenerateImage(_) => None,
}
}
}
@@ -15,12 +15,12 @@ use gateway_core::engine::continuation::{
ContinuationBinding, NativeContinuationPin, PreviousResponseId,
};
use gateway_core::engine::credential::{
AccountSelectionPolicy, CredentialRevision, ProviderAccountId, RotationStrategy,
AccountSelectionPolicy, ProviderAccountId, RotationStrategy,
};
use gateway_core::engine::execution::gateway_error_from_engine;
use gateway_core::engine::provider::{
Provider, ProviderCallMetadata, ProviderCatalogGeneration, ProviderModelCapabilities,
ProviderRegistry, ProviderRequest, ProviderResource, ProviderStream, UpstreamTransport,
ProviderRegistry, ProviderRequest, ProviderStream, UpstreamTransport,
};
use gateway_core::engine::{
AttemptContext, AttemptCoordinator, AttemptRecord, CancellationToken, CommitRequirement,
@@ -38,9 +38,7 @@ use gateway_core::event::{
ProviderResponseObservation, ProviderResponseTimings, ResponseMeta, ToolCallDelta,
UpstreamHttpVersion, WebSocketPoolKind,
};
use gateway_core::operation::{
EmbedRequest, GenerateRequest, Operation, ProtocolPayload, ProviderSessionState, RetrySafety,
};
use gateway_core::operation::{GenerateRequest, Operation, ProtocolPayload, ProviderSessionState};
use gateway_core::policy::ClientApiKeyId;
use gateway_core::routing::{
AccountRoutingSnapshot, ClientRoutingScope, ConfigRevision, FrozenAccountScope,
@@ -321,10 +319,7 @@ impl Provider for ScriptedProvider {
.upstream_model()
.cloned()
.expect("model route candidate"),
ProviderResource::Account {
id: ProviderAccountId::new(account_id).expect("account id"),
revision: CredentialRevision::new(1).expect("revision"),
},
ProviderAccountId::new(account_id).expect("account id"),
UpstreamTransport::new("http_sse").expect("transport"),
)
.with_upstream_request_id(OpaqueUpstreamValue::new("upstream-request"));
@@ -344,10 +339,7 @@ impl Provider for ScriptedProvider {
.upstream_model()
.cloned()
.expect("model route candidate"),
ProviderResource::Account {
id: ProviderAccountId::new(account_id).expect("account id"),
revision: CredentialRevision::new(1).expect("revision"),
},
ProviderAccountId::new(account_id).expect("account id"),
UpstreamTransport::new("http_sse").expect("transport"),
);
Ok(ProviderStream::new(
@@ -367,10 +359,7 @@ impl Provider for ScriptedProvider {
.upstream_model()
.cloned()
.expect("model route candidate"),
ProviderResource::Account {
id: ProviderAccountId::new(account_id).expect("account id"),
revision: CredentialRevision::new(1).expect("revision"),
},
ProviderAccountId::new(account_id).expect("account id"),
UpstreamTransport::new("websocket").expect("planned transport"),
);
Ok(ProviderStream::new(
@@ -383,15 +372,6 @@ impl Provider for ScriptedProvider {
}
}
fn operation(retry_safety: RetrySafety) -> Operation {
if retry_safety == RetrySafety::Idempotent {
return Operation::Embed(
EmbedRequest::new(vec!["hello".to_owned()]).expect("embedding request"),
);
}
generate_operation()
}
fn generate_operation() -> Operation {
let body = json!({
"model": "gpt-5",
@@ -678,7 +658,7 @@ fn terminal_non_idempotent_failure(
items: Vec<Result<GatewayEvent, ProviderError>>,
continuation: Option<ContinuationBinding>,
) -> (Arc<FakeStore>, Arc<ScriptedProvider>) {
let operation = operation(RetrySafety::NonIdempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, provider) = coordinator(vec![
Script::Stream {
@@ -709,7 +689,7 @@ fn terminal_non_idempotent_failure(
#[test]
fn success_updates_one_model_request_and_persists_usage() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, _) = coordinator(vec![Script::Stream {
account_id: "acct_one",
@@ -775,7 +755,7 @@ fn observation_store_failures_never_block_successful_delivery() {
StoreWriteFailure::Commit,
StoreWriteFailure::Finalize,
] {
let operation = operation(RetrySafety::NonIdempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let store = Arc::new(FakeStore::failing(failure));
let (coordinator, _, provider) = coordinator_with_store(
@@ -1006,7 +986,7 @@ fn failed_image_request_should_persist_failure() {
#[test]
fn response_observation_is_persisted_but_never_delivered() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let observation = ProviderResponseObservation::new(
UpstreamTransport::new("http_sse").expect("actual transport"),
@@ -1067,7 +1047,7 @@ fn response_observation_is_persisted_but_never_delivered() {
#[test]
fn mismatched_response_observation_should_not_interrupt_client_events() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let first = ProviderEvent::observation(ProviderResponseObservation::new(
UpstreamTransport::new("websocket").expect("first transport"),
@@ -1109,7 +1089,7 @@ fn mismatched_response_observation_should_not_interrupt_client_events() {
#[test]
fn response_observation_should_persist_provider_metadata_and_output_timings() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let metadata = gateway_core::event::ProviderResponseMetadata::new(
serde_json::json!({"effectiveModel": "gpt-test", "firstTokenMs": 23}).to_string(),
@@ -1165,7 +1145,7 @@ fn response_observation_should_persist_provider_metadata_and_output_timings() {
#[test]
fn empty_tool_call_delta_should_not_preempt_provider_first_token_timing() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let transport = UpstreamTransport::new("websocket").expect("actual transport");
let initial_observation = ProviderResponseObservation::new(transport.clone());
@@ -1218,7 +1198,7 @@ fn empty_tool_call_delta_should_not_preempt_provider_first_token_timing() {
#[test]
fn unknown_wire_event_before_response_identity_is_discarded_with_retried_attempt() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let unknown = ProtocolWireEvent::json(
"openai",
@@ -1231,10 +1211,10 @@ fn unknown_wire_event_before_response_identity_is_discarded_with_retried_attempt
account_id: "acct_first",
items: vec![
Ok(ProviderEvent::wire(unknown)),
Err(ProviderError::new(
ProviderErrorKind::Unavailable,
UpstreamSendState::Sent,
)),
Err(
ProviderError::new(ProviderErrorKind::Unavailable, UpstreamSendState::Sent)
.with_replay_safe(),
),
],
},
Script::Stream {
@@ -1264,7 +1244,7 @@ fn unknown_wire_event_before_response_identity_is_discarded_with_retried_attempt
#[test]
fn discarded_attempt_observation_does_not_leak_into_retry_result() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let first_observation = ProviderResponseObservation::new(
UpstreamTransport::new("websocket").expect("first transport"),
@@ -1348,7 +1328,7 @@ fn discarded_attempt_observation_does_not_leak_into_retry_result() {
#[test]
fn websocket_success_keeps_client_http_status_absent() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, _) = coordinator(vec![Script::Stream {
account_id: "acct_one",
@@ -1374,7 +1354,7 @@ fn websocket_success_keeps_client_http_status_absent() {
#[test]
fn success_without_response_observation_persists_no_upstream_status() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, _) = coordinator(vec![Script::Stream {
account_id: "acct_one",
@@ -1731,7 +1711,7 @@ fn continuation_history_failure_without_replay_proof_is_delivered_without_hidden
#[test]
fn diagnostic_account_reaches_provider_and_matching_metadata_succeeds() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, _, provider) = coordinator(vec![Script::Stream {
account_id: "acct_required",
@@ -1758,7 +1738,7 @@ fn diagnostic_account_reaches_provider_and_matching_metadata_succeeds() {
#[test]
fn provider_metadata_for_another_account_fails_closed() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, provider) = coordinator(vec![
Script::Stream {
@@ -1795,7 +1775,7 @@ fn provider_metadata_for_another_account_fails_closed() {
#[test]
fn required_account_disables_account_retry_after_stream_creation() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, provider) = coordinator(vec![
Script::Stream {
@@ -1803,7 +1783,8 @@ fn required_account_disables_account_retry_after_stream_creation() {
items: vec![Err(ProviderError::new(
ProviderErrorKind::Unavailable,
UpstreamSendState::NotSent,
))],
)
.with_replay_safe())],
},
Script::Stream {
account_id: "acct_other",
@@ -1843,7 +1824,7 @@ fn required_account_disables_account_retry_after_stream_creation() {
#[test]
fn latest_provider_reported_cost_is_persisted_as_known_usd_total() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let events = vec![
Ok(GatewayEvent::Started(ResponseMeta::new(
@@ -1886,7 +1867,7 @@ fn latest_provider_reported_cost_is_persisted_as_known_usd_total() {
#[test]
fn calculated_cost_is_persisted_when_provider_does_not_report_cost() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let events = vec![
Ok(GatewayEvent::Started(ResponseMeta::new(
@@ -1924,7 +1905,7 @@ fn calculated_cost_is_persisted_when_provider_does_not_report_cost() {
#[test]
fn provider_reported_cost_should_not_be_replaced_by_calculated_cost() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let events = vec![
Ok(GatewayEvent::Started(ResponseMeta::new(
@@ -1973,7 +1954,7 @@ fn provider_reported_cost_should_not_be_replaced_by_calculated_cost() {
#[test]
fn discarded_attempt_cost_never_leaks_into_retry_result() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, _) = coordinator(vec![
Script::Stream {
@@ -1989,10 +1970,10 @@ fn discarded_attempt_cost_never_leaks_into_retry_result() {
Ok(GatewayEvent::ProviderCost(
ProviderReportedCost::from_usd_ticks(999).expect("discarded cost"),
)),
Err(ProviderError::new(
ProviderErrorKind::Unavailable,
UpstreamSendState::Sent,
)),
Err(
ProviderError::new(ProviderErrorKind::Unavailable, UpstreamSendState::Sent)
.with_replay_safe(),
),
],
},
Script::Stream {
@@ -2018,7 +1999,7 @@ fn discarded_attempt_cost_never_leaks_into_retry_result() {
#[test]
fn pre_commit_failure_excludes_account_and_retries_same_target() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, provider) = coordinator(vec![
Script::Stream {
@@ -2028,10 +2009,10 @@ fn pre_commit_failure_excludes_account_and_retries_same_target() {
"discarded-response",
"gpt-5",
))),
Err(ProviderError::new(
ProviderErrorKind::Unauthorized,
UpstreamSendState::Sent,
)),
Err(
ProviderError::new(ProviderErrorKind::Unauthorized, UpstreamSendState::Sent)
.with_replay_safe(),
),
],
},
Script::Stream {
@@ -2082,7 +2063,7 @@ fn pre_commit_failure_excludes_account_and_retries_same_target() {
#[test]
fn recovered_credential_retries_the_same_account_exactly_once() {
let operation = operation(RetrySafety::NonIdempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, provider) = coordinator(vec![
Script::Stream {
@@ -2127,7 +2108,7 @@ fn recovered_credential_retries_the_same_account_exactly_once() {
#[test]
fn retryable_error_after_credential_recovery_switches_account_instead_of_terminating() {
let operation = operation(RetrySafety::NonIdempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, provider) = coordinator(vec![
Script::Stream {
@@ -2183,7 +2164,7 @@ fn retryable_error_after_credential_recovery_switches_account_instead_of_termina
#[test]
fn non_idempotent_explicit_429_rejection_rotates_account_before_output() {
let operation = operation(RetrySafety::NonIdempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, provider) = coordinator(vec![
Script::Stream {
@@ -2228,7 +2209,7 @@ fn non_idempotent_explicit_429_rejection_rotates_account_before_output() {
#[test]
fn rate_limited_account_exhaustion_survives_a_later_empty_selection() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, _) = coordinator(vec![
Script::Stream {
@@ -2321,7 +2302,7 @@ fn explicit_429_with_ambiguous_send_state_is_not_retried() {
#[test]
fn explicit_429_after_structural_event_should_retry_before_commit() {
let operation = operation(RetrySafety::NonIdempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, provider) = coordinator(vec![
Script::Stream {
@@ -2684,7 +2665,7 @@ fn external_continuation_explicit_429_is_not_retried() {
#[test]
fn ambiguous_send_state_stops_retry() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, provider) = coordinator(vec![
Script::Stream {
@@ -2692,7 +2673,8 @@ fn ambiguous_send_state_stops_retry() {
items: vec![Err(ProviderError::new(
ProviderErrorKind::Transport,
UpstreamSendState::Ambiguous,
))],
)
.with_replay_safe())],
},
Script::Stream {
account_id: "acct_second",
@@ -2730,7 +2712,7 @@ fn ambiguous_send_state_stops_retry() {
#[test]
fn pre_delivery_transport_failure_rotates_account_for_non_idempotent_generation() {
let operation = operation(RetrySafety::NonIdempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, provider) = coordinator(vec![
Script::Stream {
@@ -2773,7 +2755,7 @@ fn pre_delivery_transport_failure_rotates_account_for_non_idempotent_generation(
#[test]
fn final_capacity_exhaustion_returns_the_last_retryable_upstream_failure() {
let operation = operation(RetrySafety::NonIdempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let raw_body = Bytes::from_static(b"{\"error\":\"last upstream body\"}\x00");
let upstream_error =
@@ -2840,7 +2822,7 @@ fn final_capacity_exhaustion_returns_the_last_retryable_upstream_failure() {
#[test]
fn provider_infrastructure_failure_does_not_restore_the_last_upstream_failure() {
let operation = operation(RetrySafety::NonIdempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let upstream_error =
ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous)
@@ -2895,7 +2877,7 @@ fn provider_infrastructure_failure_does_not_restore_the_last_upstream_failure()
#[test]
fn pre_delivery_retry_marker_is_ignored_after_downstream_commit() {
let operation = operation(RetrySafety::NonIdempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, provider) = coordinator(vec![
Script::Stream {
@@ -2946,7 +2928,7 @@ fn pre_delivery_retry_marker_is_ignored_after_downstream_commit() {
#[test]
fn structural_event_before_replay_safe_failure_should_switch_account_before_commit() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, provider) = coordinator(vec![
Script::Stream {
@@ -3005,7 +2987,7 @@ fn structural_event_before_replay_safe_failure_should_switch_account_before_comm
#[test]
fn cancellation_before_pending_delivery_commit_reaches_terminal_state() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, _) = coordinator(vec![Script::Stream {
account_id: "acct_first",
@@ -3039,7 +3021,7 @@ fn cancellation_before_pending_delivery_commit_reaches_terminal_state() {
#[test]
fn no_eligible_account_before_stream_does_not_create_request_detail() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, _) = coordinator(vec![Script::Error(ProviderError::new(
ProviderErrorKind::NoEligibleAccount,
@@ -3072,7 +3054,7 @@ fn no_eligible_account_before_stream_does_not_create_request_detail() {
#[test]
fn expired_deadline_finalizes_without_calling_provider() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, provider) = coordinator(vec![]);
@@ -3098,7 +3080,7 @@ fn expired_deadline_finalizes_without_calling_provider() {
#[test]
fn deadline_after_commit_is_incomplete_without_provider_circuit_failure() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, _) = coordinator(vec![Script::HangingStream {
account_id: "acct_first",
@@ -3138,7 +3120,7 @@ fn deadline_after_commit_is_incomplete_without_provider_circuit_failure() {
#[test]
fn deadline_before_first_event_records_no_provider_circuit_failure() {
let operation = operation(RetrySafety::Idempotent);
let operation = generate_operation();
let route_plan = plan(&operation);
let (coordinator, store, _) = coordinator(vec![Script::HangingStream {
account_id: "acct_first",
@@ -18,7 +18,7 @@ use gateway_core::engine::continuation::{
NativeContinuationPin, NativeContinuationPort, NativeContinuationStoreError, PreviousResponseId,
};
use gateway_core::engine::credential::{
AccountSelectionPolicy, CredentialRevision, ProviderAccountId, RotationStrategy,
AccountSelectionPolicy, ProviderAccountId, RotationStrategy,
};
use gateway_core::engine::execution::{
ClientApiKeyUsageSink, ClientTransport, DefaultExecutionService, ExecutionRequestMetadata,
@@ -28,7 +28,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, ProviderResource, ProviderStream, UpstreamTransport,
ProviderRegistry, ProviderRequest, ProviderStream, UpstreamTransport,
};
use gateway_core::engine::{
AttemptContext, AttemptRecord, ExecutionStore, IntermediateFailure, ModelRequestFinalization,
@@ -283,10 +283,7 @@ impl Provider for ColdFailingProvider {
) -> Result<ProviderStream, ProviderError> {
let metadata = ProviderCallMetadata::for_provider_endpoint(
request.candidate().provider().clone(),
ProviderResource::Account {
id: ProviderAccountId::new("acct_usage").expect("account"),
revision: CredentialRevision::new(1).expect("revision"),
},
ProviderAccountId::new("acct_usage").expect("account"),
UpstreamTransport::new("http_json").expect("transport"),
);
Ok(ProviderStream::new(
@@ -6,16 +6,14 @@ use async_trait::async_trait;
use futures::{StreamExt, stream};
use futures_timer::Delay;
use gateway_core::engine::credential::{
AccountFeedbackStats, CredentialRevision, ProviderAccountId,
};
use gateway_core::engine::credential::{AccountFeedbackStats, ProviderAccountId};
use gateway_core::engine::provider::{
EventStream, Provider, ProviderCallMetadata, ProviderCatalogGeneration,
ProviderModelCapabilities, ProviderRegistry, ProviderRequest, ProviderResource, ProviderStream,
RegistryError, ResourceId, UpstreamTransport,
ProviderModelCapabilities, ProviderRegistry, ProviderRequest, ProviderStream, RegistryError,
UpstreamTransport,
};
use gateway_core::engine::{AttemptContext, UpstreamSendState};
use gateway_core::error::{IdentifierError, ProviderError, ProviderErrorKind};
use gateway_core::error::{ProviderError, ProviderErrorKind};
use gateway_core::event::{
ContentItem, ContentKind, GatewayEvent, ProtocolWireEvent, ProviderEvent, ResponseMeta,
TextDelta,
@@ -112,7 +110,7 @@ fn provider_stream_should_release_owned_lease_on_drop() {
let metadata = ProviderCallMetadata::new(
ProviderKind::new("openai").expect("valid provider"),
UpstreamModelId::new("gpt-5").expect("valid model"),
ProviderResource::Anonymous(ResourceId::none()),
ProviderAccountId::new("acct_drop").expect("account"),
UpstreamTransport::new("http_sse").expect("valid transport"),
);
let events: EventStream = Box::pin(stream::empty());
@@ -131,10 +129,7 @@ fn provider_stream_should_report_common_account_success_and_first_output() {
let metadata = ProviderCallMetadata::new(
provider.clone(),
UpstreamModelId::new("gpt-5").expect("valid model"),
ProviderResource::Account {
id: account.clone(),
revision: CredentialRevision::new(1).expect("revision"),
},
account.clone(),
UpstreamTransport::new("http_sse").expect("valid transport"),
);
let response = ResponseMeta::new("resp_upstream", "gpt-5");
@@ -173,7 +168,7 @@ fn provider_stream_should_not_sequence_gate_canonical_observation_attached_to_wi
let metadata = ProviderCallMetadata::new(
ProviderKind::new("openai").expect("valid provider"),
UpstreamModelId::new("gpt-5").expect("valid model"),
ProviderResource::Anonymous(ResourceId::none()),
ProviderAccountId::new("acct_wire").expect("account"),
UpstreamTransport::new("http_sse").expect("valid transport"),
);
let wire = ProtocolWireEvent::json(
@@ -217,10 +212,7 @@ fn provider_stream_should_report_confirmed_sent_failure_but_ignore_unconfirmed_f
let metadata = ProviderCallMetadata::new(
provider.clone(),
UpstreamModelId::new("grok-4.5").expect("valid model"),
ProviderResource::Account {
id: account,
revision: CredentialRevision::new(1).expect("revision"),
},
account,
UpstreamTransport::new("http_sse").expect("valid transport"),
);
let events: EventStream = Box::pin(stream::iter([Err(ProviderError::new(
@@ -252,19 +244,3 @@ fn provider_stream_should_report_confirmed_sent_failure_but_ignore_unconfirmed_f
(None, None)
);
}
#[test]
fn resource_id_should_accept_versioned_hmac_pseudonym() {
let resource = ResourceId::anonymous("rr_hmac_sha256_v1:opaque-digest")
.expect("versioned pseudonym is valid");
assert_eq!(resource.as_str(), "rr_hmac_sha256_v1:opaque-digest");
}
#[test]
fn resource_id_should_reserve_none_sentinel_for_constructor() {
let error = ResourceId::anonymous("__none__")
.expect_err("sentinel must only be created by ResourceId::none");
assert_eq!(error, IdentifierError::ReservedPrefix);
}
@@ -1,9 +1,7 @@
use bytes::Bytes;
use gateway_core::error::OperationError;
use gateway_core::operation::{
EmbedRequest, Feature, GenerateRequest, ImageRequest, ImageRequestKind, Operation,
OperationKind, ProtocolPayload, ProviderSessionState, RawJsonPayload, RerankRequest,
RetrySafety, SpeechRequest,
Feature, GenerateRequest, ImageRequest, ImageRequestKind, Operation, OperationKind,
ProtocolPayload, ProviderSessionState, RawJsonPayload,
};
use serde_json::{Map, Value, json};
@@ -117,43 +115,16 @@ fn provider_session_state_should_only_be_visible_to_its_provider() {
}
#[test]
fn non_generate_requests_should_reject_missing_required_fields() {
assert_eq!(
EmbedRequest::new(Vec::new()).expect_err("input is required"),
OperationError::EmptyField { field: "input" }
);
assert_eq!(
RerankRequest::new("", vec!["document".to_owned()]).expect_err("query is required"),
OperationError::EmptyField { field: "query" }
);
assert_eq!(
SpeechRequest::new("hello", "").expect_err("voice is required"),
OperationError::EmptyField { field: "voice" }
);
}
#[test]
fn operation_kind_and_retry_safety_should_remain_stable() {
fn operation_kind_should_remain_stable() {
let generate = Operation::Generate(generate(json!({"model": "gpt-test"})));
let embed = Operation::Embed(
EmbedRequest::new(vec!["input".to_owned()]).expect("embedding request is valid"),
);
let rerank = Operation::Rerank(
RerankRequest::new("query", vec!["document".to_owned()]).expect("rerank request is valid"),
);
let image = Operation::GenerateImage(image(
ImageRequestKind::Generation,
json!({"model": "gpt-image-2", "prompt": "draw"}),
));
let speech = Operation::Speech(SpeechRequest::new("hello", "alloy").expect("speech request"));
assert_eq!(generate.kind(), OperationKind::Generate);
assert_eq!(generate.retry_safety(), RetrySafety::NonIdempotent);
assert_eq!(embed.retry_safety(), RetrySafety::Idempotent);
assert_eq!(rerank.retry_safety(), RetrySafety::Idempotent);
assert_eq!(image.kind(), OperationKind::GenerateImage);
assert!(image.image_generation_requested());
assert_eq!(speech.kind(), OperationKind::Speech);
}
#[test]
@@ -494,7 +494,7 @@ fn upstream_feature_validation_should_preserve_operation_and_limit_gates() {
.require(Feature::JsonSchema);
let oversized_output = CapabilityRequirements::new(OperationKind::Generate)
.with_requested_output_tokens(Some(16_001));
let unsupported_operation = CapabilityRequirements::new(OperationKind::Speech);
let unsupported_operation = CapabilityRequirements::new(OperationKind::GenerateImage);
assert_eq!(
(
-1
View File
@@ -33,7 +33,6 @@ tracing-subscriber = { version = "0.3.23", features = ["env-filter", "json"] }
filetime = "0.2.27"
tempfile = "3.27.0"
tokio = { version = "1.52.3", default-features = false, features = ["rt-multi-thread", "test-util"] }
tower = { version = "0.5.3", features = ["util"] }
wiremock = "0.6.5"
[[test]]
@@ -675,7 +675,6 @@ fn provider_error_bytes(error: &ProviderError) -> usize {
let mut bytes = text_bytes([
error.upstream_code().map(|value| value.as_str()),
error.upstream_request_id().map(|value| value.as_str()),
error.upstream_response_id().map(|value| value.as_str()),
]);
if let Some(client_error) = error.client_visible_upstream_error() {
bytes = bytes.saturating_add(text_bytes([
@@ -48,7 +48,6 @@ tokio-tungstenite = { git = "https://github.com/openai-oss-forks/tokio-tungsteni
tracing = "0.1.44"
tungstenite = { git = "https://github.com/openai-oss-forks/tungstenite-rs", rev = "4fffad30fe373adbdcffab9545e9e9bf4f2fc19f", features = ["deflate", "proxy"] }
url = "2.5.8"
urlencoding = "2.1.3"
uuid = { version = "1.23.3", features = ["v4", "v7"] }
# 官方 Codex app-server 对 /v1/responses 请求体默认做 zstd 压缩(level 3),
# 并携带 Content-Encoding: zstd;保持同版本同级别以对齐上游指纹。
@@ -61,5 +61,5 @@ pub(crate) use selector::{CodexCyberPolicyScope, SelectCodexProviderEndpointCred
pub use types::{
CODEX_AUTHENTICATION_KIND_OAUTH, CodexAccountProfile, CodexCookie, CodexCookieCaptureOutcome,
CodexCredentialData, CodexCredentialPrincipal, CodexOAuthCredentialData, CodexOAuthSecret,
RotateCodexCredential, RuntimeCodexCookie, UpsertCodexCookie,
RuntimeCodexCookie, UpsertCodexCookie,
};
@@ -3,10 +3,9 @@
use std::sync::Arc;
use std::time::SystemTime;
use chrono::{DateTime, Utc};
use gateway_core::engine::credential::{
AccountErrorReason, AccountStateChange, CredentialCasOutcome, CredentialCasUpdate,
CredentialRevision, CredentialState, ProviderAccount, ProviderAccountId, ProviderAccountStore,
CredentialRevision, CredentialState, ProviderAccount, ProviderAccountStore,
ProviderAccountUpdate,
};
use gateway_core::routing::ProviderKind;
@@ -14,7 +13,7 @@ use secrecy::ExposeSecret;
use thiserror::Error;
use super::security::{CodexCredentialCodec, CodexCredentialDataError, CodexRuntimeCredential};
use super::types::{CodexCredentialData, CodexOAuthSecret, RotateCodexCredential};
use super::types::{CodexCredentialData, CodexOAuthSecret};
const PROVIDER_NAME: &str = "openai";
@@ -34,51 +33,6 @@ impl CodexCredentialRepository {
&self.store
}
pub async fn rotate_oauth_secret(
&self,
input: RotateCodexCredential,
) -> Result<CredentialRevision, CredentialRepositoryError> {
let account_id = ProviderAccountId::new(input.account_id)
.map_err(|_| CredentialRepositoryError::InvalidInput("account_id"))?;
let expected = CredentialRevision::new(input.expected_credential_revision)
.map_err(|_| CredentialRepositoryError::InvalidInput("credential_revision"))?;
let current = self.store.load_credential(&account_id, expected).await?;
let mut data = CodexCredentialCodec::decode_complete(&current.credential)?;
let oauth = data
.oauth_mut()
.ok_or(CredentialRepositoryError::InvalidCredentialData)?;
oauth.access_token = input.secret.access_token.expose_secret().to_owned();
oauth.refresh_token = input
.secret
.refresh_token
.as_ref()
.map(|value| value.expose_secret().to_owned());
oauth.id_token = input
.secret
.id_token
.as_ref()
.map(|value| value.expose_secret().to_owned());
let credential = CodexCredentialCodec::encode_complete(data)?;
let update = CredentialCasUpdate::new(
account_id.clone(),
expected,
ProviderAccountUpdate {
account_id,
name: current.account.name().to_owned(),
email: input.verified_account.email.clone(),
plan_type: input.verified_account.plan_type.clone(),
},
credential,
input.secret.refresh_token.is_some(),
Some(required_time(
input.verified_account.access_token_expires_at,
)?),
optional_time(input.next_refresh_at),
)
.map_err(|_| CredentialRepositoryError::InvalidCredentialData)?;
cas_revision(self.store.compare_and_swap_credential(update).await?)
}
/// 持久化成功 RT exchange 的 token,同时保留既有账号身份投影。
///
/// Refresh endpoint 已经是这次 AT/RT 轮换的授权边界;这里仅以 revision
@@ -294,20 +248,8 @@ fn cas_revision(
}
}
fn required_time(value: Option<DateTime<Utc>>) -> Result<SystemTime, CredentialRepositoryError> {
value
.map(SystemTime::from)
.ok_or(CredentialRepositoryError::InvalidCredentialData)
}
fn optional_time(value: Option<DateTime<Utc>>) -> Option<SystemTime> {
value.map(SystemTime::from)
}
#[derive(Debug, Error)]
pub enum CredentialRepositoryError {
#[error("invalid Codex credential input: {0}")]
InvalidInput(&'static str),
#[error("Codex credential data is invalid")]
InvalidCredentialData,
#[error("Codex credential revision conflict")]
@@ -1320,8 +1320,7 @@ pub enum CredentialSelectionError {
impl From<CredentialRepositoryError> for CredentialSelectionError {
fn from(error: CredentialRepositoryError) -> Self {
match error {
CredentialRepositoryError::InvalidInput(_)
| CredentialRepositoryError::InvalidCredentialData => Self::InvalidCredential,
CredentialRepositoryError::InvalidCredentialData => Self::InvalidCredential,
CredentialRepositoryError::RevisionConflict | CredentialRepositoryError::Store => {
Self::Store
}
@@ -274,14 +274,6 @@ pub fn openai_token_client(
Ok(OpenAiTokenClient::new(client, config))
}
/// 构建固定官方 endpoint 的 Codex token client。
pub fn official_openai_token_client() -> Result<OpenAiTokenClient, TokenClientBuildError> {
openai_token_client(TokenClientConfig {
client_id: OFFICIAL_CODEX_OAUTH_CLIENT_ID.to_owned(),
token_endpoint: OFFICIAL_CODEX_TOKEN_ENDPOINT.to_owned(),
})
}
#[derive(Deserialize)]
struct RefreshTokenResponse {
access_token: Option<String>,
@@ -344,12 +344,6 @@ impl CodexCredentialData {
}
}
pub fn set_installation_id(&mut self, installation_id: String) {
match self {
Self::OAuth(data) => data.installation_id = installation_id,
}
}
#[must_use]
pub fn cookies(&self) -> &[CodexCookie] {
match self {
@@ -391,31 +385,6 @@ impl fmt::Debug for CodexCredentialData {
}
}
/// 刷新成功后的 CAS 输入。
pub struct RotateCodexCredential {
pub account_id: String,
pub expected_credential_revision: u64,
pub secret: CodexOAuthSecret,
pub verified_account: CodexAccountProfile,
pub next_refresh_at: Option<DateTime<Utc>>,
}
impl fmt::Debug for RotateCodexCredential {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("RotateCodexCredential")
.field("account_id", &self.account_id)
.field(
"expected_credential_revision",
&self.expected_credential_revision,
)
.field("secret", &self.secret)
.field("verified_account", &self.verified_account)
.field("next_refresh_at", &self.next_refresh_at)
.finish()
}
}
/// 运行时使用的 Cookie;值受 `secrecy` 保护。
pub struct RuntimeCodexCookie {
pub name: String,
@@ -12,8 +12,8 @@ use gateway_core::engine::continuation::{ContinuationBinding, NativeContinuation
use gateway_core::engine::credential::{AccountFeedbackStats, ProviderAccount};
use gateway_core::engine::provider::{
EventStream, Provider, ProviderCallMetadata, ProviderCatalogGeneration,
ProviderModelCapabilities, ProviderRequest, ProviderRequestObservation, ProviderResource,
ProviderStream, UpstreamTransport,
ProviderModelCapabilities, ProviderRequest, ProviderRequestObservation, ProviderStream,
UpstreamTransport,
};
use gateway_core::engine::{
AttemptContext, CancellationToken, ContinuationAttempt, UpstreamSendState,
@@ -386,10 +386,7 @@ impl Provider for CodexProvider {
let metadata = ProviderCallMetadata::new(
provider_kind,
upstream_model.clone(),
ProviderResource::Account {
id: lease.account_id().clone(),
revision: lease.account().revision(),
},
lease.account_id().clone(),
UpstreamTransport::new(transport_name(transport)).map_err(|_| {
provider_error(ProviderErrorKind::Protocol, UpstreamSendState::NotSent)
})?,
@@ -464,10 +461,7 @@ impl CodexProvider {
.map_err(|_| provider_error(ProviderErrorKind::Protocol, UpstreamSendState::NotSent))?;
let metadata = ProviderCallMetadata::for_provider_endpoint(
provider_kind,
ProviderResource::Account {
id: lease.account_id().clone(),
revision: lease.account().revision(),
},
lease.account_id().clone(),
UpstreamTransport::new(HTTP_JSON_TRANSPORT).map_err(|_| {
provider_error(ProviderErrorKind::Protocol, UpstreamSendState::NotSent)
})?,
@@ -6,9 +6,7 @@ use provider_openai::transport::canonical::{
use provider_openai::transport::protocol::websocket::websocket_event_to_sse_frame;
use serde_json::json;
const METADATA_PREFIX_FIXTURE: &str = include_str!(
"../../../../gateway-api/tests/openai/responses/fixtures/http_sse/metadata_only_prefix.sse"
);
const METADATA_PREFIX_FIXTURE: &str = include_str!("fixtures/metadata_only_prefix.sse");
#[test]
fn decoder_should_not_forward_codex_rate_limit_metadata_fixture_as_openai_wire() {
+1 -1
View File
@@ -455,7 +455,7 @@ fn provider_call_metadata(
Ok(ProviderCallMetadata::new(
ProviderKind::new(XAI_PROVIDER_NAME).map_err(|_| protocol_not_sent())?,
candidate_upstream_model(candidate)?.clone(),
selected.resource(),
selected.account_id().clone(),
UpstreamTransport::new(HTTP_SSE_TRANSPORT).map_err(|_| protocol_not_sent())?,
))
}
@@ -8,7 +8,6 @@ use std::time::{Duration, SystemTime};
use gateway_core::engine::credential::{
AccountEligibilityPolicy, AccountSelectionPolicy, CredentialRevision, ProviderAccountId,
};
use gateway_core::engine::provider::ProviderResource;
use gateway_core::policy::ClientApiKeyId;
use gateway_core::routing::{FrozenAccountScope, UpstreamModelId};
use sha2::{Digest as _, Sha256};
@@ -147,15 +146,6 @@ impl SelectedGrokSession {
self.credential_revision
}
/// 返回 Core 为本次上游调用记录的元数据。
#[must_use]
pub fn resource(&self) -> ProviderResource {
ProviderResource::Account {
id: self.account_id.clone(),
revision: self.credential_revision,
}
}
/// 返回用于显式构造 header 的 OAuth access token。
#[must_use]
pub const fn access_token(&self) -> &SecretValue {
@@ -1125,7 +1125,7 @@ async fn execute_forwards_required_account_to_selector() {
.await
.expect("required account stream");
assert_eq!(stream.metadata().provider_account_id(), Some(&required));
assert_eq!(stream.metadata().provider_account_id(), &required);
assert_eq!(
selector
.required_accounts
@@ -1151,7 +1151,7 @@ async fn execute_returns_cold_stream_and_records_selected_account() {
assert_eq!(transport.calls.load(Ordering::SeqCst), 0);
assert_eq!(
stream.metadata().provider_account_id(),
Some(&account_id("provider"))
&account_id("provider")
);
let events = stream.by_ref().collect::<Vec<_>>().await;
assert_eq!(transport.calls.load(Ordering::SeqCst), 1);
-8
View File
@@ -360,14 +360,6 @@ export function recoverAccount(data: AccountIdParam) {
})
}
export function getAccountQuota(data: AccountIdParam) {
return request<AccountQuotaResponse>({
url: '/api/admin/accounts/quota',
method: 'GET',
params: data,
})
}
export function getAccountProfileStatistics(data: AccountIdParam) {
return request<AccountProfileStatisticsResponse>({
url: '/api/admin/accounts/profile-statistics',
@@ -1,164 +0,0 @@
<script setup lang="ts">
import type { AccountGroup } from '@/api'
import { Search } from '@lucide/vue'
import { computed, shallowRef } from 'vue'
import BaseCheckbox from '@/components/base/BaseCheckbox.vue'
import BaseEmpty from '@/components/base/BaseEmpty.vue'
import BaseInput from '@/components/base/BaseInput.vue'
import BaseScrollbar from '@/components/base/BaseScrollbar.vue'
import { formatProviderLabel } from '@/utils/providers'
const props = withDefaults(
defineProps<{
groups: AccountGroup[]
loading?: boolean
disabled?: boolean
emptyMode: 'all-accounts' | 'ungrouped' | 'require-selection'
}>(),
{
loading: false,
disabled: false,
},
)
const selectedIds = defineModel<string[]>({ required: true })
const search = shallowRef('')
const normalizedSearch = computed(() => search.value.trim().toLocaleLowerCase())
const visibleGroups = computed(() => {
if (!normalizedSearch.value)
return props.groups
return props.groups.filter(group =>
`${group.name} ${group.description ?? ''}`.toLocaleLowerCase().includes(normalizedSearch.value),
)
})
const emptyCopy = computed(() => {
if (props.emptyMode === 'all-accounts') {
return {
title: '全部账号权限',
description: '未选择分组时,该密钥可以使用当前及以后新增的所有账号。',
tone: 'warning',
}
}
if (props.emptyMode === 'ungrouped') {
return {
title: '保持未分组',
description: '账号暂不加入任何分组,可稍后在账号管理中调整。',
tone: 'warning',
}
}
if (props.emptyMode === 'require-selection') {
return {
title: '请选择分组',
description: '选中一个或多个分组后才能执行批量操作。',
tone: 'normal',
}
}
return {
title: '请选择分组',
description: '选中一个或多个分组后才能执行批量操作。',
tone: 'normal',
}
})
function toggle(groupId: string) {
if (props.disabled)
return
const next = new Set(selectedIds.value)
if (next.has(groupId))
next.delete(groupId)
else
next.add(groupId)
selectedIds.value = [...next]
}
function providerSummary(group: AccountGroup) {
return Object.entries(group.providerCounts)
.filter(([, count]) => count > 0)
.map(([provider, count]) => `${formatProviderLabel(provider)} ${count}`)
.join(' · ')
}
</script>
<template>
<div class="grid gap-3">
<BaseInput
v-model="search"
aria-label="搜索账号分组"
placeholder="搜索分组..."
:disabled="disabled || loading"
>
<template #prefix>
<Search class="size-4 text-cp-text-tertiary" />
</template>
</BaseInput>
<BaseScrollbar
v-if="visibleGroups.length > 0"
max-height="260px"
>
<div class="grid gap-2">
<div
v-for="group in visibleGroups"
:key="group.id"
class="flex w-full items-start gap-3 rounded-cp border-0 px-3 py-3 text-left outline-none transition-colors focus-visible:ring-2 focus-visible:ring-cp-control-outline"
:class="selectedIds.includes(group.id)
? 'bg-cp-control-item-bg-active'
: 'bg-cp-fill-quaternary hover:bg-cp-bg-text-hover'"
>
<BaseCheckbox
:model-value="selectedIds.includes(group.id)"
:disabled="disabled || loading"
:label="`选择${group.name}`"
@update:model-value="toggle(group.id)"
/>
<span class="min-w-0 flex-1">
<span class="flex min-w-0 items-center gap-2">
<strong class="min-w-0 truncate text-cp text-cp-text">
{{ group.name }}
</strong>
<span
v-if="!group.enabled"
class="shrink-0 rounded-md bg-cp-warning-bg px-1.5 py-1 text-[10px] leading-none font-bold text-cp-warning-text"
>
已禁用
</span>
</span>
<span class="mt-1 block text-cp-xs font-emphasis text-cp-text-secondary">
{{ group.memberCount }} 个账号<template v-if="providerSummary(group)"> · {{ providerSummary(group) }}</template>
</span>
</span>
</div>
</div>
</BaseScrollbar>
<BaseEmpty
v-else-if="groups.length === 0 && !loading"
size="sm"
title="暂无可选分组"
description="请先在分组管理中创建分组。"
/>
<BaseEmpty
v-else-if="!loading"
size="sm"
title="没有匹配分组"
description="请调整搜索关键词。"
/>
<div
v-if="selectedIds.length === 0"
class="rounded-cp px-3.5 py-3"
:class="emptyCopy.tone === 'warning'
? 'bg-cp-warning-bg text-cp-warning-text'
: 'bg-cp-fill-quaternary text-cp-text-secondary'"
>
<p class="m-0 text-cp-sm font-bold">
{{ emptyCopy.title }}
</p>
<p class="mt-1 mb-0 text-cp-xs leading-normal font-emphasis">
{{ emptyCopy.description }}
</p>
</div>
</div>
</template>
@@ -203,5 +203,3 @@ export function useBackupRecords() {
stopPolling,
}
}
export type BackupRecordsStore = ReturnType<typeof useBackupRecords>
@@ -168,5 +168,3 @@ export function useBackupSettings() {
saveSchedule,
}
}
export type BackupSettingsStore = ReturnType<typeof useBackupSettings>
-6
View File
@@ -55,12 +55,6 @@ export function formatUsd(value?: string | number | null, precise = false) {
}).format(safeValue)
}
export function formatEstimatedAmount(currency: string, value?: string | null) {
if (value == null)
return '—'
return currency.toUpperCase() === 'USD' ? formatUsd(value, true) : `${currency} ${value}`
}
export function formatUsdAxis(value: number) {
const safeValue = finiteOrZero(value)
if (Math.abs(safeValue) >= 1_000)

Some files were not shown because too many files have changed in this diff Show More