mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-06 08:06:19 +00:00
refactor(backend): simplify shared rules and redundant tests
Reuse Core error classification and token presentation, remove unused operation APIs and redundant tests, and avoid unnecessary context cloning. Fix expired reset timestamp underflow and SSE metadata frame boundaries. Consolidate architecture checks and document the backend audit results. Validation: backend tests, architecture tests, rustfmt and strict Clippy. PostgreSQL/Redis integration was not verified without dedicated services. Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
Codex
@@ -4,6 +4,7 @@ use std::{
|
||||
collections::{BTreeMap, BTreeSet},
|
||||
fs,
|
||||
path::{Path, PathBuf},
|
||||
sync::OnceLock,
|
||||
};
|
||||
|
||||
use syn::{Item, visit::Visit};
|
||||
@@ -26,22 +27,24 @@ pub(super) const WORKSPACE_MEMBERS: &[&str] = &[
|
||||
|
||||
#[test]
|
||||
fn workspace_member_list_matches_the_frozen_dag_scope() {
|
||||
let manifest =
|
||||
fs::read_to_string(backend_root().join("Cargo.toml")).expect("read workspace manifest");
|
||||
for member in WORKSPACE_MEMBERS {
|
||||
assert!(
|
||||
manifest.contains(&format!("\"{member}\"")),
|
||||
"workspace must include {member}"
|
||||
);
|
||||
}
|
||||
let member_count = manifest
|
||||
.lines()
|
||||
.map(str::trim)
|
||||
.filter(|line| line.starts_with("\"apps/") || line.starts_with("\"crates/"))
|
||||
.count();
|
||||
let metadata = cargo_metadata_json();
|
||||
let root = backend_root();
|
||||
let actual: BTreeSet<_> = metadata["packages"]
|
||||
.as_array()
|
||||
.expect("Cargo packages")
|
||||
.iter()
|
||||
.map(|package| {
|
||||
Path::new(package["manifest_path"].as_str().expect("package manifest"))
|
||||
.parent()
|
||||
.expect("package directory")
|
||||
.strip_prefix(&root)
|
||||
.expect("workspace member path")
|
||||
.to_path_buf()
|
||||
})
|
||||
.collect();
|
||||
assert_eq!(
|
||||
member_count,
|
||||
WORKSPACE_MEMBERS.len(),
|
||||
actual,
|
||||
WORKSPACE_MEMBERS.iter().map(PathBuf::from).collect(),
|
||||
"new workspace members must extend the dependency DAG rules"
|
||||
);
|
||||
}
|
||||
@@ -378,7 +381,7 @@ const ALLOWED_INTERNAL_EDGES: &[(&str, &str)] = &[
|
||||
#[test]
|
||||
fn workspace_internal_dependency_edges_match_frozen_dag() {
|
||||
let metadata = cargo_metadata_json();
|
||||
let actual = workspace_runtime_dependency_edges(&metadata);
|
||||
let actual = workspace_runtime_dependency_edges(metadata);
|
||||
let mut expected: Vec<(String, String)> = ALLOWED_INTERNAL_EDGES
|
||||
.iter()
|
||||
.map(|(from, to)| ((*from).to_owned(), (*to).to_owned()))
|
||||
@@ -474,24 +477,6 @@ fn adapter_public_module_surfaces_match_allowlist() {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn production_sources_do_not_host_tests() {
|
||||
for member in WORKSPACE_MEMBERS {
|
||||
let src = backend_root().join(member).join("src");
|
||||
let sources = super::rust_files(&src);
|
||||
assert!(!sources.is_empty(), "{member} has no production sources");
|
||||
for relative in sources {
|
||||
let path = src.join(relative);
|
||||
let source = fs::read_to_string(&path).expect("read production source");
|
||||
assert!(
|
||||
!source.contains("#[cfg(test)]"),
|
||||
"{} hosts tests in production src",
|
||||
path.display()
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn workspace_modules_follow_conventional_file_layout() {
|
||||
let targets = test_source_roots();
|
||||
@@ -748,40 +733,46 @@ pub(super) fn backend_root() -> PathBuf {
|
||||
.expect("backend workspace root")
|
||||
}
|
||||
|
||||
fn cargo_metadata_json() -> serde_json::Value {
|
||||
let output = std::process::Command::new("cargo")
|
||||
.args(["metadata", "--format-version", "1", "--no-deps"])
|
||||
.current_dir(backend_root())
|
||||
.output()
|
||||
.expect("run cargo metadata");
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"cargo metadata failed: {}",
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
serde_json::from_slice(&output.stdout).expect("parse cargo metadata")
|
||||
fn cargo_metadata_json() -> &'static serde_json::Value {
|
||||
static METADATA: OnceLock<serde_json::Value> = OnceLock::new();
|
||||
METADATA.get_or_init(|| {
|
||||
let output = std::process::Command::new("cargo")
|
||||
.args(["metadata", "--format-version", "1", "--no-deps", "--locked"])
|
||||
.current_dir(backend_root())
|
||||
.output()
|
||||
.expect("run cargo metadata");
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"cargo metadata failed: {}",
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
serde_json::from_slice(&output.stdout).expect("parse cargo metadata")
|
||||
})
|
||||
}
|
||||
|
||||
/// 提取成员 `[dependencies]` 段内声明的依赖名;段落以下一个 `[` 表头结束
|
||||
/// 复用 Cargo 解析后的实际包名，覆盖重命名依赖与平台条件，排除测试和构建依赖
|
||||
fn dependency_names(member: &str) -> Vec<String> {
|
||||
let manifest = fs::read_to_string(backend_root().join(member).join("Cargo.toml"))
|
||||
.expect("read member manifest");
|
||||
let mut in_dependencies = false;
|
||||
let mut names = Vec::new();
|
||||
for line in manifest.lines() {
|
||||
let line = line.trim();
|
||||
if line.starts_with('[') {
|
||||
in_dependencies = line == "[dependencies]";
|
||||
continue;
|
||||
}
|
||||
if !in_dependencies || line.is_empty() || line.starts_with('#') {
|
||||
continue;
|
||||
}
|
||||
if let Some((name, _)) = line.split_once('=') {
|
||||
names.push(name.trim().to_owned());
|
||||
}
|
||||
}
|
||||
names
|
||||
let manifest = backend_root().join(member).join("Cargo.toml");
|
||||
let package = cargo_metadata_json()["packages"]
|
||||
.as_array()
|
||||
.expect("Cargo packages")
|
||||
.iter()
|
||||
.find(|package| {
|
||||
Path::new(package["manifest_path"].as_str().expect("package manifest")) == manifest
|
||||
})
|
||||
.expect("workspace member");
|
||||
package["dependencies"]
|
||||
.as_array()
|
||||
.expect("package dependencies")
|
||||
.iter()
|
||||
.filter(|dependency| dependency["kind"].is_null())
|
||||
.map(|dependency| {
|
||||
dependency["name"]
|
||||
.as_str()
|
||||
.expect("dependency name")
|
||||
.to_owned()
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn assert_no_dependency(member: &str, forbidden: &[&str]) {
|
||||
|
||||
@@ -59,9 +59,16 @@ fn cargo_library_root_is_conventional_lib() {
|
||||
fn workspace_production_files_have_no_hidden_modules_or_test_hooks() {
|
||||
for member in architecture::WORKSPACE_MEMBERS {
|
||||
let root = architecture::backend_root().join(member).join("src");
|
||||
for relative in rust_files(&root) {
|
||||
let sources = rust_files(&root);
|
||||
assert!(!sources.is_empty(), "{member} has no production sources");
|
||||
for relative in sources {
|
||||
let path = root.join(relative);
|
||||
let source = fs::read_to_string(&path).expect("read production source");
|
||||
assert!(
|
||||
!source.contains("#[cfg(test)]"),
|
||||
"{} hosts tests in production src",
|
||||
path.display()
|
||||
);
|
||||
assert!(
|
||||
!source.contains("include!("),
|
||||
"{} uses include!",
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
//! 观测领域事实到管理页面的展示投影
|
||||
|
||||
use super::*;
|
||||
use gateway_core::metering::Usage;
|
||||
|
||||
pub(crate) fn display_duration(value: Option<u64>) -> String {
|
||||
let Some(value) = value.and_then(|value| i64::try_from(value).ok()) else {
|
||||
@@ -100,7 +101,7 @@ pub(crate) fn attempt_metrics_view(metrics: &domain::AttemptMetrics) -> AttemptM
|
||||
}
|
||||
|
||||
pub(crate) fn token_details(record: &domain::UsageRecord) -> TokenDetailsView {
|
||||
TokenDetailsView {
|
||||
token_details_view(&Usage {
|
||||
input_tokens: record.input_tokens,
|
||||
output_tokens: record.output_tokens,
|
||||
cached_input_tokens: record.cached_input_tokens,
|
||||
@@ -109,35 +110,11 @@ pub(crate) fn token_details(record: &domain::UsageRecord) -> TokenDetailsView {
|
||||
image_input_tokens: record.image_input_tokens,
|
||||
image_output_tokens: record.image_output_tokens,
|
||||
total_tokens: record.total_tokens,
|
||||
input_tokens_display: record
|
||||
.input_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_number),
|
||||
output_tokens_display: record
|
||||
.output_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_number),
|
||||
cached_input_tokens_display: record
|
||||
.cached_input_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_compact_number),
|
||||
cache_write_input_tokens_display: record
|
||||
.cache_write_input_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_compact_number),
|
||||
reasoning_output_tokens_display: record
|
||||
.reasoning_output_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_number),
|
||||
image_input_tokens_display: record
|
||||
.image_input_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_number),
|
||||
image_output_tokens_display: record
|
||||
.image_output_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_number),
|
||||
total_tokens_display: record
|
||||
.total_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_number),
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn usage_list_token_details(record: &domain::UsageListRecord) -> TokenDetailsView {
|
||||
TokenDetailsView {
|
||||
token_details_view(&Usage {
|
||||
input_tokens: record.input_tokens,
|
||||
output_tokens: record.output_tokens,
|
||||
cached_input_tokens: record.cached_input_tokens,
|
||||
@@ -146,28 +123,41 @@ pub(crate) fn usage_list_token_details(record: &domain::UsageListRecord) -> Toke
|
||||
image_input_tokens: record.image_input_tokens,
|
||||
image_output_tokens: record.image_output_tokens,
|
||||
total_tokens: record.total_tokens,
|
||||
input_tokens_display: record
|
||||
})
|
||||
}
|
||||
|
||||
fn token_details_view(usage: &Usage) -> TokenDetailsView {
|
||||
TokenDetailsView {
|
||||
input_tokens: usage.input_tokens,
|
||||
output_tokens: usage.output_tokens,
|
||||
cached_input_tokens: usage.cached_input_tokens,
|
||||
cache_write_input_tokens: usage.cache_write_input_tokens,
|
||||
reasoning_output_tokens: usage.reasoning_output_tokens,
|
||||
image_input_tokens: usage.image_input_tokens,
|
||||
image_output_tokens: usage.image_output_tokens,
|
||||
total_tokens: usage.total_tokens,
|
||||
input_tokens_display: usage
|
||||
.input_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_number),
|
||||
output_tokens_display: record
|
||||
output_tokens_display: usage
|
||||
.output_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_number),
|
||||
cached_input_tokens_display: record
|
||||
cached_input_tokens_display: usage
|
||||
.cached_input_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_compact_number),
|
||||
cache_write_input_tokens_display: record
|
||||
cache_write_input_tokens_display: usage
|
||||
.cache_write_input_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_compact_number),
|
||||
reasoning_output_tokens_display: record
|
||||
reasoning_output_tokens_display: usage
|
||||
.reasoning_output_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_number),
|
||||
image_input_tokens_display: record
|
||||
image_input_tokens_display: usage
|
||||
.image_input_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_number),
|
||||
image_output_tokens_display: record
|
||||
image_output_tokens_display: usage
|
||||
.image_output_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_number),
|
||||
total_tokens_display: record
|
||||
total_tokens_display: usage
|
||||
.total_tokens
|
||||
.map_or_else(|| "-".to_owned(), format_number),
|
||||
}
|
||||
|
||||
@@ -13,6 +13,8 @@ use gateway_core::{
|
||||
};
|
||||
use serde_json::{Value, json};
|
||||
|
||||
pub use gateway_core::engine::execution::gateway_error_from_engine;
|
||||
|
||||
use super::responses::ProtocolErrorBody;
|
||||
|
||||
/// OpenAI 风格错误响应
|
||||
@@ -148,39 +150,6 @@ pub fn model_not_found_response() -> (StatusCode, Json<Value>) {
|
||||
)
|
||||
}
|
||||
|
||||
/// 将 core Engine 错误收敛为稳定 Gateway 分类
|
||||
#[must_use]
|
||||
pub fn gateway_error_from_engine(error: &EngineError) -> GatewayError {
|
||||
match error {
|
||||
EngineError::Provider(error) => GatewayError::from_provider(error),
|
||||
EngineError::Cancelled => {
|
||||
GatewayError::new(GatewayErrorKind::Cancelled, "request was cancelled")
|
||||
}
|
||||
EngineError::Deadline => {
|
||||
GatewayError::new(GatewayErrorKind::Timeout, "gateway request timed out")
|
||||
}
|
||||
EngineError::ProviderNotRegistered { .. } | EngineError::EmptyRoutingPlan => {
|
||||
GatewayError::new(
|
||||
GatewayErrorKind::NoAvailableProvider,
|
||||
"no upstream provider is currently available for this request",
|
||||
)
|
||||
}
|
||||
EngineError::Store(source) => {
|
||||
GatewayError::new(GatewayErrorKind::Internal, "gateway execution failed")
|
||||
.with_source(source.clone())
|
||||
}
|
||||
EngineError::ProviderMetadataMismatch
|
||||
| EngineError::ContinuationPinMismatch
|
||||
| EngineError::RequiredAccountMismatch
|
||||
| EngineError::AccountOutsideClientScope
|
||||
| EngineError::DownstreamCommitRequired
|
||||
| EngineError::NoActiveAttempt
|
||||
| EngineError::InvalidDeliveryState => {
|
||||
GatewayError::new(GatewayErrorKind::Internal, "gateway execution failed")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Gateway 错误的 OpenAI HTTP 表达
|
||||
pub fn gateway_error_response(error: &GatewayError) -> Response {
|
||||
let (status, default_type, default_code) = gateway_error_contract(error.kind());
|
||||
|
||||
@@ -22,13 +22,6 @@ fn manifest_should_depend_only_on_core_protocol_and_http_adapter_layers() {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn workspace_should_include_gateway_api() {
|
||||
let workspace = include_str!("../../../Cargo.toml");
|
||||
|
||||
assert!(workspace.contains("\"crates/gateway-api\""));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn admin_routes_should_use_only_get_post_and_static_paths() {
|
||||
let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("src/admin");
|
||||
|
||||
@@ -548,7 +548,7 @@ fn engine_store_error_should_collapse_to_safe_internal_error() {
|
||||
|
||||
let projected = gateway_error_from_engine(&error);
|
||||
assert_eq!(projected.kind(), GatewayErrorKind::Internal);
|
||||
assert_eq!(projected.safe_message(), "gateway execution failed");
|
||||
assert_eq!(projected.safe_message(), "request execution failed");
|
||||
}
|
||||
#[test]
|
||||
fn openai_error_response_should_preserve_only_safe_contract_fields() {
|
||||
|
||||
@@ -638,7 +638,8 @@ fn provider_event_to_middleware_frame(
|
||||
|
||||
fn encode_sse_wire_event(wire: &ProtocolWireEvent) -> Result<Bytes, ProviderError> {
|
||||
let mut encoded = Vec::new();
|
||||
if let Some(id) = wire.sse_id() {
|
||||
// 源事件元数据保持不透明；仅编码可表示为单行 SSE 字段的值，避免生成额外帧
|
||||
if let Some(id) = wire.sse_id().filter(|id| !id.contains(['\r', '\n', '\0'])) {
|
||||
encoded.extend_from_slice(b"id: ");
|
||||
encoded.extend_from_slice(id.as_bytes());
|
||||
encoded.push(b'\n');
|
||||
@@ -646,7 +647,10 @@ fn encode_sse_wire_event(wire: &ProtocolWireEvent) -> Result<Bytes, ProviderErro
|
||||
if let Some(retry) = wire.sse_retry() {
|
||||
encoded.extend_from_slice(format!("retry: {retry}\n").as_bytes());
|
||||
}
|
||||
if let Some(event_type) = wire.event_type() {
|
||||
if let Some(event_type) = wire
|
||||
.event_type()
|
||||
.filter(|event_type| !event_type.is_empty() && !event_type.contains(['\r', '\n']))
|
||||
{
|
||||
encoded.extend_from_slice(b"event: ");
|
||||
encoded.extend_from_slice(event_type.as_bytes());
|
||||
encoded.push(b'\n');
|
||||
|
||||
@@ -464,29 +464,6 @@ impl GenerateRequest {
|
||||
self.payload.client_turn.as_ref()
|
||||
}
|
||||
|
||||
/// 写回原生 Provider 编码结果，保留已准入的能力、连接上下文和会话状态
|
||||
///
|
||||
/// 原生编码不消耗中间件的一次跨协议改写额度：可以保留协议名，也可以在一次
|
||||
/// 中间件改写后编码到 Provider wire
|
||||
/// 调用方仍须验证保护字段没有被改动
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// 目标协议名无效时返回错误
|
||||
pub fn with_native_encoded_body(
|
||||
mut self,
|
||||
protocol: impl Into<String>,
|
||||
body: Map<String, Value>,
|
||||
) -> Result<Self, OperationError> {
|
||||
let encoded = ProtocolPayload::json_object(protocol, body)?
|
||||
.with_context(self.protocol_payload().context().clone());
|
||||
let requirements = self.requirements();
|
||||
let payload = Arc::make_mut(&mut self.payload);
|
||||
payload.protocol_payload = encoded;
|
||||
payload.source_requirements = Some(requirements);
|
||||
Ok(self)
|
||||
}
|
||||
|
||||
/// 附着同一客户端连接上一轮由 Provider 返回的不透明状态
|
||||
#[must_use]
|
||||
pub fn with_provider_session_state(mut self, state: ProviderSessionState) -> Self {
|
||||
@@ -989,41 +966,6 @@ impl Operation {
|
||||
self.replace_protocol_wire_inner(None, body, context)
|
||||
}
|
||||
|
||||
/// 替换 Provider HTTP 正文、非 wire 上下文和已解析 headers，同时保持 endpoint、
|
||||
/// method 与 query 不变
|
||||
/// headers 继续由 [`ProviderHttpRequest::new`] 统一验证；
|
||||
/// Core 不解析插件传输使用的 base64 表示
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// 当前 operation 不是 Provider HTTP，或替换后的 headers 超出请求边界时返回错误
|
||||
pub fn replace_provider_http_wire(
|
||||
self,
|
||||
body: Option<Bytes>,
|
||||
context: Map<String, Value>,
|
||||
headers: Vec<ProviderHttpHeader>,
|
||||
) -> Result<Self, OperationError> {
|
||||
let Self::ProviderHttp(request) = self else {
|
||||
return Err(OperationError::EmptyField {
|
||||
field: "provider_http_request",
|
||||
});
|
||||
};
|
||||
let payload = request.payload;
|
||||
ProviderHttpRequest::new(
|
||||
request.endpoint,
|
||||
request.method,
|
||||
request.query,
|
||||
headers,
|
||||
RawHttpPayload {
|
||||
protocol: payload.protocol,
|
||||
body: body.unwrap_or(payload.body),
|
||||
context,
|
||||
translated: payload.translated,
|
||||
},
|
||||
)
|
||||
.map(Self::ProviderHttp)
|
||||
}
|
||||
|
||||
/// 把一次已选择的协议转换结果写回 operation，同时保留非 wire 上下文和
|
||||
/// Provider 会话状态
|
||||
/// 该方法只验证公共载荷形状，不解释目标协议字段
|
||||
|
||||
@@ -509,6 +509,84 @@ fn attempt_passthrough_preserves_parsed_wire_for_every_client_transport() {
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn attempt_sse_encoding_keeps_opaque_metadata_from_creating_extra_frames() {
|
||||
#[derive(Debug)]
|
||||
struct InspectingPlan {
|
||||
expected: &'static str,
|
||||
}
|
||||
|
||||
impl MiddlewarePlan for InspectingPlan {
|
||||
fn handle(
|
||||
&self,
|
||||
_: MiddlewareContext,
|
||||
request: MiddlewareRequest,
|
||||
next: MiddlewareNext,
|
||||
) -> BoxFuture<'static, Result<MiddlewareResponse, MiddlewareError>> {
|
||||
let expected = self.expected;
|
||||
Box::pin(async move {
|
||||
let (protocol, status, headers, mut body, envelope) =
|
||||
next.run(request).await?.into_parts();
|
||||
let frame = body.next_frame().await?.expect("provider event");
|
||||
assert_eq!(frame.bytes().as_ref(), expected.as_bytes());
|
||||
Ok(MiddlewareResponse::new(protocol, status, headers, body)
|
||||
.with_envelope(envelope.expect("provider metadata")))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
futures::executor::block_on(async {
|
||||
for (event_type, id, expected) in [
|
||||
(
|
||||
"future.event",
|
||||
"event-1",
|
||||
"id: event-1\nretry: 10\nevent: future.event\ndata: {\"value\":1}\n\n",
|
||||
),
|
||||
(
|
||||
"future.event\ndata: injected\n\n",
|
||||
"event-1\r\ndata: injected\r\n\r\n",
|
||||
"retry: 10\ndata: {\"value\":1}\n\n",
|
||||
),
|
||||
("", "event\0id", "retry: 10\ndata: {\"value\":1}\n\n"),
|
||||
] {
|
||||
let plan = FrozenMiddlewarePlan::new(
|
||||
Arc::new(InspectingPlan { expected }),
|
||||
ExtensionSetReference::new(
|
||||
ExtensionSetId::new("sse-metadata".to_owned()).unwrap(),
|
||||
Arc::new(ReadyGeneration),
|
||||
),
|
||||
);
|
||||
let wire = ProtocolWireEvent::json_with_sse_metadata(
|
||||
"openai",
|
||||
Some(event_type.to_owned()),
|
||||
json!({"value":1}),
|
||||
Some(id.to_owned()),
|
||||
Some(10),
|
||||
)
|
||||
.unwrap();
|
||||
let terminal: ProviderMiddlewareTerminal = Box::new(move |_, _| {
|
||||
Box::pin(async move {
|
||||
Ok(ProviderStream::new(
|
||||
empty_provider_stream().metadata().clone(),
|
||||
stream::iter([Ok(ProviderEvent::wire(wire))]),
|
||||
(),
|
||||
))
|
||||
})
|
||||
});
|
||||
|
||||
execute_attempt_middleware(
|
||||
Some(&plan),
|
||||
context(),
|
||||
operation(),
|
||||
ClientTransport::HttpSse,
|
||||
terminal,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn attempt_middleware_preserves_multi_value_and_authentication_headers() {
|
||||
futures::executor::block_on(async {
|
||||
|
||||
@@ -8,7 +8,6 @@ mod execution;
|
||||
mod extensions;
|
||||
mod middleware;
|
||||
mod policy;
|
||||
mod probe;
|
||||
mod provider;
|
||||
mod response_control;
|
||||
|
||||
|
||||
@@ -1,9 +0,0 @@
|
||||
//! 验证账号探测可通过对象安全的执行链端口调用
|
||||
|
||||
use gateway_core::engine::probe::AccountProbe;
|
||||
|
||||
#[test]
|
||||
fn account_probe_is_an_object_safe_execution_chain_port() {
|
||||
fn accepts(_: Option<&dyn AccountProbe>) {}
|
||||
accepts(None);
|
||||
}
|
||||
@@ -1,24 +1,10 @@
|
||||
//! 验证健康探针的对象安全接口与 Worker 身份投影
|
||||
//! 验证 Worker 健康状态的身份投影
|
||||
|
||||
use futures::future::BoxFuture;
|
||||
use gateway_core::health::{
|
||||
HealthProbe, HealthState, WorkerHealthKey, WorkerHealthSnapshot, WorkerHealthSource,
|
||||
WorkerRuntimeState,
|
||||
WorkerHealthKey, WorkerHealthSnapshot, WorkerHealthSource, WorkerRuntimeState,
|
||||
};
|
||||
use gateway_core::task::{WorkerId, WorkerKind};
|
||||
|
||||
struct HealthyProbe;
|
||||
|
||||
impl HealthProbe for HealthyProbe {
|
||||
fn name(&self) -> &'static str {
|
||||
"test"
|
||||
}
|
||||
|
||||
fn check(&self) -> BoxFuture<'_, HealthState> {
|
||||
Box::pin(async { HealthState::Healthy })
|
||||
}
|
||||
}
|
||||
|
||||
struct SingleSnapshot;
|
||||
|
||||
impl WorkerHealthSource for SingleSnapshot {
|
||||
@@ -38,13 +24,6 @@ impl WorkerHealthSource for SingleSnapshot {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn health_probe_contract_is_object_safe() {
|
||||
let probe: &dyn HealthProbe = &HealthyProbe;
|
||||
|
||||
assert_eq!(probe.name(), "test");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn worker_health_source_preserves_worker_identity() {
|
||||
let source: &dyn WorkerHealthSource = &SingleSnapshot;
|
||||
|
||||
@@ -315,14 +315,6 @@ fn connection_registration_rejects_after_drain_linearization() {
|
||||
assert!(matches!(lifecycle.try_register(), Err(ConnectionDraining)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn connection_lifecycle_contract_is_object_safe() {
|
||||
let lifecycle = TestLifecycle::new();
|
||||
let object: &dyn ConnectionLifecycle = &lifecycle;
|
||||
|
||||
assert!(!object.is_draining());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cancellation_reason_survives_cleanup_and_propagates_only_to_descendants() {
|
||||
use gateway_core::lifecycle::{CancellationReason, LeaseFailure, RequestLeaseKind};
|
||||
|
||||
@@ -109,68 +109,6 @@ fn middleware_protocol_change_keeps_source_capability_facts_without_interpreting
|
||||
assert_eq!(replaced.capability_requirements(), source_requirements);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn native_encoding_keeps_admission_context_and_session_after_optional_translation() {
|
||||
let context = Map::from_iter([("conversation_id".to_owned(), json!("conversation"))]);
|
||||
let session = ProviderSessionState::new(
|
||||
"xai",
|
||||
Map::from_iter([("session_id".to_owned(), json!("session"))]),
|
||||
)
|
||||
.unwrap();
|
||||
let request = GenerateRequest::from_protocol_payload(
|
||||
ProtocolPayload::json_object(
|
||||
"openai",
|
||||
json!({"input":"hello", "max_output_tokens":64, "reasoning":{"effort":"high"}})
|
||||
.as_object()
|
||||
.unwrap()
|
||||
.clone(),
|
||||
)
|
||||
.unwrap()
|
||||
.with_context(context.clone()),
|
||||
)
|
||||
.with_provider_session_state(session.clone());
|
||||
let original = Operation::Generate(request.clone());
|
||||
let translated = original
|
||||
.clone()
|
||||
.replace_middleware_wire("native-input", Bytes::from_static(br#"{"prompt":"hello"}"#))
|
||||
.unwrap();
|
||||
let Operation::Generate(translated) = translated else {
|
||||
panic!("generate request");
|
||||
};
|
||||
|
||||
for (request, target) in [(request, "openai"), (translated, "xai")] {
|
||||
let body = Map::from_iter([("native_prompt".to_owned(), json!("encoded"))]);
|
||||
let encoded = request
|
||||
.with_native_encoded_body(target, body.clone())
|
||||
.unwrap();
|
||||
assert_eq!(encoded.protocol_payload().protocol(), target);
|
||||
assert_eq!(encoded.protocol_payload().body(), &body);
|
||||
assert_eq!(encoded.protocol_payload().context(), &context);
|
||||
assert_eq!(encoded.provider_session_state("xai"), Some(&session));
|
||||
let operation = Operation::Generate(encoded);
|
||||
assert_eq!(
|
||||
operation.capability_requirements(),
|
||||
original.capability_requirements()
|
||||
);
|
||||
assert!(
|
||||
operation
|
||||
.replace_middleware_wire("another", Bytes::from_static(b"{}"))
|
||||
.is_err()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn native_encoding_rejects_invalid_protocol_names() {
|
||||
for name in ["", "invalid\nprotocol"] {
|
||||
assert!(
|
||||
generate(json!({"input":"hello"}))
|
||||
.with_native_encoded_body(name, Map::new())
|
||||
.is_err()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn middleware_processing_recomputes_requirements_but_protocol_change_is_direct_once() {
|
||||
let operation = Operation::Generate(generate(json!({
|
||||
@@ -336,68 +274,6 @@ fn provider_http_should_reject_urls_controls_and_unbounded_headers() {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn provider_http_wire_replacement_keeps_frozen_target_and_validates_typed_headers() {
|
||||
let operation = Operation::ProviderHttp(
|
||||
ProviderHttpRequest::new(
|
||||
"inspect",
|
||||
ProviderHttpMethod::Post,
|
||||
Some("mode=stable".to_owned()),
|
||||
vec![ProviderHttpHeader::new(
|
||||
"x-original",
|
||||
Bytes::from_static(b"old"),
|
||||
)],
|
||||
RawHttpPayload::new("provider-http", Bytes::from_static(b"old-body"))
|
||||
.expect("provider HTTP payload"),
|
||||
)
|
||||
.expect("provider HTTP request"),
|
||||
);
|
||||
assert!(
|
||||
operation
|
||||
.clone()
|
||||
.replace_provider_http_wire(
|
||||
None,
|
||||
Map::new(),
|
||||
vec![ProviderHttpHeader::new(
|
||||
"x-invalid",
|
||||
Bytes::from_static(b"line\r\nbreak"),
|
||||
)],
|
||||
)
|
||||
.is_err()
|
||||
);
|
||||
let replacement = operation
|
||||
.replace_provider_http_wire(
|
||||
Some(Bytes::from_static(&[0x00, 0xff, b'n', b'e', b'w'])),
|
||||
Map::from_iter([("connection_fact".to_owned(), json!("kept"))]),
|
||||
vec![
|
||||
ProviderHttpHeader::new("x-repeat", Bytes::from_static(&[0xff, 0xfe])),
|
||||
ProviderHttpHeader::new("x-repeat", Bytes::from_static(b"two")),
|
||||
],
|
||||
)
|
||||
.expect("typed Provider HTTP replacement");
|
||||
let Operation::ProviderHttp(request) = replacement else {
|
||||
panic!("operation kind changed")
|
||||
};
|
||||
assert_eq!(request.endpoint(), "inspect");
|
||||
assert_eq!(request.method(), ProviderHttpMethod::Post);
|
||||
assert_eq!(request.query(), Some("mode=stable"));
|
||||
assert_eq!(
|
||||
request.payload().body().as_ref(),
|
||||
&[0x00, 0xff, b'n', b'e', b'w']
|
||||
);
|
||||
assert_eq!(request.payload().context()["connection_fact"], "kept");
|
||||
assert_eq!(request.headers().len(), 2);
|
||||
assert_eq!(request.headers()[0].value().as_ref(), &[0xff, 0xfe]);
|
||||
assert_eq!(request.headers()[1].value().as_ref(), b"two");
|
||||
|
||||
let generate = Operation::Generate(generate(json!({"model":"gpt-test"})));
|
||||
assert!(
|
||||
generate
|
||||
.replace_provider_http_wire(None, Map::new(), Vec::new())
|
||||
.is_err()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn standalone_search_should_keep_raw_body_and_context_opaque() {
|
||||
let body = Bytes::from_static(
|
||||
|
||||
@@ -324,25 +324,6 @@ fn publisher_should_suspend_but_still_notify_after_committed_refresh_failure() {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn snapshot_ports_and_control_should_remain_object_safe() {
|
||||
fn accept_store(_: &dyn SnapshotStorePort) {}
|
||||
fn accept_subscriptions(_: &dyn SnapshotSubscriptionPort) {}
|
||||
fn accept_control(_: &dyn SnapshotControl) {}
|
||||
|
||||
let store = Arc::new(TestSnapshotStore::new(Ok(facts(1, 1))));
|
||||
let subscriptions = Arc::new(TestSnapshotSubscriptions::default());
|
||||
let publisher = RuntimeSnapshotPublisher::new(
|
||||
Arc::new(compiler(store.clone())),
|
||||
RuntimeSnapshotHandle::new(empty_snapshot(1)),
|
||||
subscriptions.clone(),
|
||||
);
|
||||
|
||||
accept_store(store.as_ref());
|
||||
accept_subscriptions(subscriptions.as_ref());
|
||||
accept_control(&publisher);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn publisher_should_contribute_reconciliation_and_subscription_workers() {
|
||||
let store = Arc::new(TestSnapshotStore::new(Ok(facts(1, 1))));
|
||||
|
||||
@@ -6,7 +6,7 @@ use std::time::Duration;
|
||||
use futures::future::BoxFuture;
|
||||
use gateway_core::lifecycle::CancellationToken;
|
||||
use gateway_core::task::{
|
||||
DaemonRestartPolicy, DaemonTask, ScheduledTask, WorkerCycleContext, WorkerDefinitionError,
|
||||
DaemonRestartPolicy, ScheduledTask, WorkerCycleContext, WorkerDefinitionError,
|
||||
WorkerFencingToken, WorkerId, WorkerKind, WorkerLeaseRequest, WorkerRegistration,
|
||||
WorkerRunnable, WorkerSchedule, WorkerTaskError,
|
||||
};
|
||||
@@ -22,14 +22,6 @@ impl ScheduledTask for NoopScheduledTask {
|
||||
}
|
||||
}
|
||||
|
||||
struct NoopDaemonTask;
|
||||
|
||||
impl DaemonTask for NoopDaemonTask {
|
||||
fn run(&self, _cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>> {
|
||||
Box::pin(async { Ok(()) })
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn schedule_rejects_every_invalid_duration_boundary() {
|
||||
let one = Duration::from_nanos(1);
|
||||
@@ -135,15 +127,6 @@ fn worker_cycle_context_preserves_optional_fencing_token() {
|
||||
assert_eq!(context.fencing_token(), Some(token));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scheduled_and_daemon_tasks_are_object_safe() {
|
||||
fn accept_scheduled(_task: &dyn ScheduledTask) {}
|
||||
fn accept_daemon(_task: &dyn DaemonTask) {}
|
||||
|
||||
accept_scheduled(&NoopScheduledTask);
|
||||
accept_daemon(&NoopDaemonTask);
|
||||
}
|
||||
|
||||
fn worker_id(owner: &str) -> WorkerId {
|
||||
WorkerId::try_new(WorkerKind::OAuthRefresh, owner).expect("valid worker identity")
|
||||
}
|
||||
|
||||
@@ -39,7 +39,11 @@ pub struct HttpPluginDistribution {
|
||||
|
||||
impl HttpPluginDistribution {
|
||||
pub fn new(http: Arc<HttpClient>) -> Result<Self, AdminError> {
|
||||
Self::with_transport("https://api.github.com/", http, source_network()?)
|
||||
Self::with_transport(
|
||||
"https://api.github.com/",
|
||||
http,
|
||||
NetworkPolicy::unrestricted(),
|
||||
)
|
||||
}
|
||||
|
||||
/// 测试可注入受限网络；生产来源由管理员显式配置，因此保持既有内外网可达范围
|
||||
@@ -173,10 +177,6 @@ impl PluginDistribution for HttpPluginDistribution {
|
||||
}
|
||||
}
|
||||
|
||||
fn source_network() -> Result<NetworkPolicy, AdminError> {
|
||||
NetworkPolicy::new(&["0.0.0.0/0".into(), "::/0".into()]).map_err(|_| source_error())
|
||||
}
|
||||
|
||||
fn credential_ids(credentials: &[SourceCredential]) -> Vec<String> {
|
||||
credentials
|
||||
.iter()
|
||||
|
||||
@@ -599,7 +599,7 @@ fn retry_after_seconds_from_resets_at(error: &Value) -> Option<u64> {
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.ok()?
|
||||
.as_secs();
|
||||
(resets_at > now).then_some(resets_at - now)
|
||||
(resets_at > now).then(|| resets_at - now)
|
||||
}
|
||||
|
||||
fn positive_u64(value: &Value) -> Option<u64> {
|
||||
|
||||
@@ -394,6 +394,19 @@ fn retry_after_should_read_string_reset_timestamp() {
|
||||
assert!((59..=60).contains(&retry_after));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn retry_after_should_ignore_elapsed_reset_timestamps_and_keep_later_advice() {
|
||||
for reset_at in [json!(1), json!("1")] {
|
||||
let mut body = json!({
|
||||
"error": {"type": "usage_limit_reached", "resets_at": reset_at}
|
||||
});
|
||||
assert_eq!(retry_after_seconds_from_body(&body.to_string()), None);
|
||||
|
||||
body["retry_after_seconds"] = json!(7);
|
||||
assert_eq!(retry_after_seconds_from_body(&body.to_string()), Some(7));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn retry_after_should_read_nested_error_delay() {
|
||||
let body = json!({
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
//! 请求正文与最终头部的会话身份解释，供账号亲和与传输编码共用
|
||||
|
||||
use std::borrow::Cow;
|
||||
|
||||
use serde_json::{Map, Value};
|
||||
|
||||
/// 只有能识别出的后代线程禁止迁移会话；没有身份的普通客户端沿用现有调度
|
||||
@@ -64,13 +66,15 @@ pub(crate) fn account_session_with_headers(
|
||||
{
|
||||
return Some(session);
|
||||
}
|
||||
let mut context = context.clone();
|
||||
let mut context = Cow::Borrowed(context);
|
||||
for (name, field) in [
|
||||
("session-id", "session_id"),
|
||||
("x-codex-turn-metadata", "turn_metadata"),
|
||||
] {
|
||||
if let Some(value) = header(name) {
|
||||
context.insert(field.to_owned(), Value::String(value.to_owned()));
|
||||
context
|
||||
.to_mut()
|
||||
.insert(field.to_owned(), Value::String(value.to_owned()));
|
||||
}
|
||||
}
|
||||
gateway_protocol::openai::codex_account_session_id(body, &context)
|
||||
|
||||
@@ -9979,7 +9979,7 @@ async fn successful_http_sse_rate_limit_event_persists_structured_exhaustion() {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn request_observation_reads_openai_metadata_without_changing_the_operation() {
|
||||
fn request_observation_reads_openai_metadata() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
let payload = ProtocolPayload::json_object(
|
||||
"openai",
|
||||
@@ -10004,11 +10004,10 @@ fn request_observation_reads_openai_metadata_without_changing_the_operation() {
|
||||
assert_eq!(observation.request_kind.as_deref(), Some("review"));
|
||||
assert_eq!(observation.subagent_kind.as_deref(), Some("worker"));
|
||||
assert!(!observation.compact);
|
||||
assert!(matches!(operation, Operation::Generate(_)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn request_observation_classifies_prewarm_from_generate_without_rewriting_the_payload() {
|
||||
fn request_observation_classifies_prewarm_from_generate() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
let provider = provider(&store);
|
||||
let client_key_id = ClientApiKeyId::new("key_prewarm_observation").expect("client key");
|
||||
@@ -10050,15 +10049,11 @@ fn request_observation_classifies_prewarm_from_generate_without_rewriting_the_pa
|
||||
body.get("generate"),
|
||||
);
|
||||
assert_eq!(observation.subagent_kind.as_deref(), Some("worker"));
|
||||
let Operation::Generate(generation) = operation else {
|
||||
unreachable!();
|
||||
};
|
||||
assert_eq!(generation.protocol_payload().body(), &body);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn endpoint_observation_should_read_models_without_rewriting_or_requiring_a_catalog() {
|
||||
fn endpoint_observation_should_read_models_without_requiring_a_catalog() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
let provider = provider(&store);
|
||||
let client_key = ClientApiKeyId::new("key_endpoint_model").expect("client key");
|
||||
@@ -10071,8 +10066,7 @@ fn endpoint_observation_should_read_models_without_rewriting_or_requiring_a_cata
|
||||
] {
|
||||
let body = serde_json::to_vec(&json!({"model":model,"future":9007199254740993_u64}))
|
||||
.expect("body");
|
||||
let payload =
|
||||
RawJsonPayload::new("openai", Bytes::copy_from_slice(&body)).expect("payload");
|
||||
let payload = RawJsonPayload::new("openai", Bytes::from(body)).expect("payload");
|
||||
let operations = [
|
||||
Operation::GenerateImage(ImageRequest::from_raw_json(
|
||||
ImageRequestKind::Generation,
|
||||
@@ -10093,12 +10087,6 @@ fn endpoint_observation_should_read_models_without_rewriting_or_requiring_a_cata
|
||||
.map(PublicModelId::as_str),
|
||||
model.as_str().filter(|model| !model.is_empty()),
|
||||
);
|
||||
let payload = match &operation {
|
||||
Operation::GenerateImage(request) => request.payload(),
|
||||
Operation::Search(request) => request.payload(),
|
||||
_ => unreachable!(),
|
||||
};
|
||||
assert_eq!(payload.body().as_ref(), body.as_slice());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -10123,43 +10111,6 @@ fn request_observation_preserves_the_raw_reasoning_effort() {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn request_observation_ignores_future_session_state_without_rewriting_the_protocol_body() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
let generation = GenerateRequest::from_protocol_payload(
|
||||
ProtocolPayload::json_object(
|
||||
"openai",
|
||||
Map::from_iter([
|
||||
("model".to_owned(), json!("gpt-test")),
|
||||
("previous_response_id".to_owned(), json!("resp_opaque")),
|
||||
]),
|
||||
)
|
||||
.expect("OpenAI payload"),
|
||||
)
|
||||
.with_provider_session_state(
|
||||
ProviderSessionState::new(
|
||||
"openai",
|
||||
Map::from_iter([("future_session_shape".to_owned(), json!([1, 2, 3]))]),
|
||||
)
|
||||
.expect("provider session state"),
|
||||
);
|
||||
let operation = Operation::Generate(generation);
|
||||
|
||||
let client_key_id = ClientApiKeyId::new("key_openai_observation").expect("client key");
|
||||
let _observation = provider(&store).request_observation(&operation, &client_key_id);
|
||||
|
||||
let Operation::Generate(generation) = &operation else {
|
||||
panic!("operation should remain a generate request");
|
||||
};
|
||||
assert_eq!(
|
||||
generation
|
||||
.protocol_payload()
|
||||
.body()
|
||||
.get("previous_response_id"),
|
||||
Some(&json!("resp_opaque"))
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn provider_compiles_catalog_presentation_for_codex_models() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
|
||||
Reference in New Issue
Block a user