mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-06 08:06:19 +00:00
refactor(bootstrap): separate startup modes and plugin wiring
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
Codex
@@ -1,463 +0,0 @@
|
||||
//! 网关唯一组合根：加载包级配置并连接各 Bundle。
|
||||
|
||||
use gateway_core::engine::provider::{ProviderRegistry, RegistryError};
|
||||
use gateway_host::{ConfigError, HostConfig, LoadableConfig};
|
||||
use serde::Deserialize;
|
||||
|
||||
const CONFIG_SCHEMA_VERSION: u32 = 1;
|
||||
|
||||
/// 顶层配置只组合各包拥有的配置段，不解释任何业务字段。
|
||||
/// 各启动配置忽略未知字段，允许升级时保留旧配置；已知字段仍按类型和业务约束校验。
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub struct GatewayConfig {
|
||||
schema_version: u32,
|
||||
host: HostConfig,
|
||||
store: gateway_store::StoreConfig,
|
||||
admin: gateway_admin::AdminConfig,
|
||||
#[serde(default)]
|
||||
client: gateway_admin::ClientConfig,
|
||||
api: gateway_api::ApiConfig,
|
||||
#[serde(default)]
|
||||
openai: provider_openai::OpenAiConfig,
|
||||
}
|
||||
|
||||
impl LoadableConfig for GatewayConfig {
|
||||
const EXTERNAL_SECTIONS: &'static [&'static str] = &["services"];
|
||||
|
||||
fn resolve_and_validate(&mut self, source_dir: &std::path::Path) -> Result<(), ConfigError> {
|
||||
if self.schema_version != CONFIG_SCHEMA_VERSION {
|
||||
return Err(ConfigError::InvalidField("schema_version"));
|
||||
}
|
||||
self.api
|
||||
.resolve_and_validate(source_dir)
|
||||
.map_err(|_| ConfigError::InvalidField("api"))?;
|
||||
self.host
|
||||
.resolve_and_validate(source_dir, &self.api.asset_directory)?;
|
||||
let runtime_data_dir = self.host.runtime_data_dir().to_path_buf();
|
||||
self.store
|
||||
.resolve_and_validate(&runtime_data_dir)
|
||||
.map_err(|_| ConfigError::InvalidField("store"))?;
|
||||
self.admin
|
||||
.resolve_and_validate(source_dir)
|
||||
.map_err(|_| ConfigError::InvalidField("admin"))?;
|
||||
self.client
|
||||
.resolve_and_validate(source_dir)
|
||||
.map_err(|_| ConfigError::InvalidField("client"))?;
|
||||
self.openai
|
||||
.resolve_and_validate(&runtime_data_dir)
|
||||
.map_err(|_| ConfigError::InvalidField("openai"))?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// 按冻结顺序初始化全部 Bundle，并把进程阻塞权交给 Host。
|
||||
pub async fn run() -> Result<(), BootstrapError> {
|
||||
launch(None).await.map(|_| ())
|
||||
}
|
||||
|
||||
/// 这里只接收插件命名空间内的参数，不转交宿主配置或其他启动 secret。
|
||||
pub struct PluginCommand {
|
||||
pub instance_id: Option<String>,
|
||||
pub name: Option<String>,
|
||||
pub arguments: Vec<String>,
|
||||
}
|
||||
|
||||
impl PluginCommand {
|
||||
fn is_help(&self) -> bool {
|
||||
self.name.is_none()
|
||||
|| self
|
||||
.arguments
|
||||
.iter()
|
||||
.any(|value| matches!(value.as_str(), "--help" | "-h"))
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn plugin_command(
|
||||
command: PluginCommand,
|
||||
) -> Result<gateway_plugin_runtime::PluginCommandOutput, BootstrapError> {
|
||||
launch(Some(command)).await
|
||||
}
|
||||
|
||||
async fn launch(
|
||||
command: Option<PluginCommand>,
|
||||
) -> Result<gateway_plugin_runtime::PluginCommandOutput, BootstrapError> {
|
||||
macro_rules! or_shutdown {
|
||||
($runtime:expr, $result:expr) => {
|
||||
match $result {
|
||||
Ok(value) => value,
|
||||
Err(error) => {
|
||||
$runtime.shutdown().await;
|
||||
return Err(error.into());
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
let config = gateway_host::load_config::<GatewayConfig>()?;
|
||||
let GatewayConfig {
|
||||
schema_version: _,
|
||||
host,
|
||||
store,
|
||||
admin,
|
||||
client,
|
||||
api,
|
||||
openai,
|
||||
} = config;
|
||||
|
||||
let timezone = host.timezone;
|
||||
let store = store.with_timezone(timezone);
|
||||
let plugin_cache = host.runtime_data_dir().join("plugins");
|
||||
let host = if command.is_some() {
|
||||
gateway_host::initialize_command_line(host).await?
|
||||
} else {
|
||||
gateway_host::initialize(host).await?
|
||||
};
|
||||
host.report_startup_ready("Host");
|
||||
let help = command.as_ref().is_some_and(PluginCommand::is_help);
|
||||
let mut store = if help {
|
||||
gateway_store::initialize_read_only(store).await?
|
||||
} else if command.is_some() {
|
||||
gateway_store::initialize_command_line(store).await?
|
||||
} else {
|
||||
gateway_store::initialize(store).await?
|
||||
};
|
||||
host.report_startup_ready("Store");
|
||||
let version = host
|
||||
.system_operations()
|
||||
.version()
|
||||
.await
|
||||
.map_err(|_| ConfigError::InvalidField("host.version"))?
|
||||
.version;
|
||||
let plugin_http = std::sync::Arc::new(gateway_host::outbound::HttpClient::new()?);
|
||||
if let Some(command) = command.as_ref().filter(|_| help) {
|
||||
let runtime = prepare_plugins(&store, plugin_cache, &version, plugin_http.clone())?;
|
||||
let commands = or_shutdown!(runtime, runtime.prepare_command_line().await);
|
||||
let stdout = commands.help(command.instance_id.as_deref(), command.name.as_deref());
|
||||
commands.shutdown().await;
|
||||
runtime.shutdown().await;
|
||||
return Ok(gateway_plugin_runtime::PluginCommandOutput {
|
||||
stdout: stdout?,
|
||||
stderr: String::new(),
|
||||
exit_code: 0,
|
||||
saved_accounts: 0,
|
||||
});
|
||||
}
|
||||
let provider_ports = store.provider_ports();
|
||||
let mut openai =
|
||||
provider_openai::initialize(openai.with_timezone(timezone), provider_ports.clone()).await?;
|
||||
host.report_startup_ready("OpenAI Provider");
|
||||
let mut xai = provider_xai::initialize(provider_ports).await?;
|
||||
host.report_startup_ready("xAI Provider");
|
||||
let providers = ProviderRegistry::new([openai.core_provider(), xai.core_provider()])?;
|
||||
let plugin_runtime = prepare_plugins(&store, plugin_cache, &version, plugin_http.clone())?;
|
||||
let admin_providers = gateway_admin::ports::provider::ProviderAdminRegistry::new([
|
||||
openai.admin_provider(),
|
||||
xai.admin_provider(),
|
||||
])
|
||||
.map_err(|_| gateway_admin::model::AdminError::invalid("Provider 管理身份冲突"))?;
|
||||
if let Some(command) = command {
|
||||
let instance_id = or_shutdown!(
|
||||
plugin_runtime,
|
||||
command
|
||||
.instance_id
|
||||
.as_deref()
|
||||
.ok_or_else(|| gateway_admin::model::AdminError::invalid("插件实例 ID 缺失"))
|
||||
);
|
||||
let name = or_shutdown!(
|
||||
plugin_runtime,
|
||||
command
|
||||
.name
|
||||
.as_deref()
|
||||
.ok_or_else(|| gateway_admin::model::AdminError::invalid("插件命令名缺失"))
|
||||
);
|
||||
let core = gateway_core::prepare_command_plane(
|
||||
store.core_ports(),
|
||||
providers.clone(),
|
||||
Some(plugin_runtime.clone()),
|
||||
Some(plugin_runtime.observer_registry()),
|
||||
Some(plugin_runtime.policy_registry()),
|
||||
Some(plugin_runtime.middleware_registry()),
|
||||
);
|
||||
let accounts = gateway_admin::initialize_plugin_accounts(
|
||||
admin_providers.clone(),
|
||||
store.admin_ports().accounts(),
|
||||
core.snapshot_control(),
|
||||
);
|
||||
or_shutdown!(plugin_runtime, plugin_runtime.bind_account_ports(&accounts));
|
||||
let keys = gateway_admin::initialize_plugin_client_keys(
|
||||
admin_providers.clone(),
|
||||
store.admin_ports().client_keys(),
|
||||
core.snapshot_control(),
|
||||
);
|
||||
or_shutdown!(plugin_runtime, plugin_runtime.bind_client_key_ports(&keys));
|
||||
let plugin_resources = gateway_admin::initialize_plugin_resources(
|
||||
store.admin_ports().plugin_resources(),
|
||||
core.snapshot_control(),
|
||||
);
|
||||
or_shutdown!(
|
||||
plugin_runtime,
|
||||
plugin_runtime.bind_resource_ports(&plugin_resources)
|
||||
);
|
||||
let core = or_shutdown!(plugin_runtime, core.activate().await);
|
||||
let nested_models = core.nested_model_execution_port();
|
||||
let affinity_lookup = core.affinity_lookup_port();
|
||||
or_shutdown!(
|
||||
plugin_runtime,
|
||||
plugin_runtime.bind_model_ports(&nested_models, &affinity_lookup)
|
||||
);
|
||||
let settings = gateway_admin::initialize_settings(
|
||||
store.admin_ports().settings(),
|
||||
core.snapshot_control(),
|
||||
admin_providers,
|
||||
std::sync::Arc::new(gateway_host::pricing::ModelsDevPricing),
|
||||
);
|
||||
let mut services = gateway_admin::service::Registry::new({
|
||||
let snapshots = core.snapshots();
|
||||
let middleware = plugin_runtime.middleware_registry();
|
||||
std::sync::Arc::new(move || {
|
||||
let snapshot = snapshots.snapshot_for_diagnostics()?;
|
||||
middleware.resolve(snapshot.extensions()?)
|
||||
})
|
||||
});
|
||||
or_shutdown!(plugin_runtime, services.register_settings(&settings));
|
||||
let services = std::sync::Arc::new(services);
|
||||
or_shutdown!(plugin_runtime, plugin_runtime.bind_services(&services));
|
||||
let commands = or_shutdown!(plugin_runtime, plugin_runtime.prepare_command_line().await);
|
||||
or_shutdown!(plugin_runtime, store.start_command_line_writes());
|
||||
let result = commands
|
||||
.execute_cancellable(instance_id, name, &command.arguments, &host.cancellation())
|
||||
.await;
|
||||
commands.shutdown().await;
|
||||
plugin_runtime.shutdown().await;
|
||||
store.shutdown_command_line_writes().await?;
|
||||
return Ok(result?);
|
||||
}
|
||||
let core = gateway_core::prepare(
|
||||
store.core_ports(),
|
||||
providers,
|
||||
Some(plugin_runtime.clone()),
|
||||
Some(plugin_runtime.observer_registry()),
|
||||
Some(plugin_runtime.policy_registry()),
|
||||
Some(plugin_runtime.middleware_registry()),
|
||||
Some(plugin_runtime.frontend_authentication_registry()),
|
||||
);
|
||||
let plugin_accounts = gateway_admin::initialize_plugin_accounts(
|
||||
admin_providers.clone(),
|
||||
store.admin_ports().accounts(),
|
||||
core.snapshot_control(),
|
||||
);
|
||||
or_shutdown!(
|
||||
plugin_runtime,
|
||||
plugin_runtime.bind_account_ports(&plugin_accounts)
|
||||
);
|
||||
let plugin_keys = gateway_admin::initialize_plugin_client_keys(
|
||||
admin_providers.clone(),
|
||||
store.admin_ports().client_keys(),
|
||||
core.snapshot_control(),
|
||||
);
|
||||
or_shutdown!(
|
||||
plugin_runtime,
|
||||
plugin_runtime.bind_client_key_ports(&plugin_keys)
|
||||
);
|
||||
let plugin_resources = gateway_admin::initialize_plugin_resources(
|
||||
store.admin_ports().plugin_resources(),
|
||||
core.snapshot_control(),
|
||||
);
|
||||
or_shutdown!(
|
||||
plugin_runtime,
|
||||
plugin_runtime.bind_resource_ports(&plugin_resources)
|
||||
);
|
||||
let mut core = or_shutdown!(plugin_runtime, core.activate().await);
|
||||
let nested_models = core.nested_model_execution_port();
|
||||
let affinity_lookup = core.affinity_lookup_port();
|
||||
or_shutdown!(
|
||||
plugin_runtime,
|
||||
plugin_runtime.bind_model_ports(&nested_models, &affinity_lookup)
|
||||
);
|
||||
host.report_startup_ready("Core");
|
||||
let host_version = or_shutdown!(
|
||||
plugin_runtime,
|
||||
version
|
||||
.trim_start_matches('v')
|
||||
.parse()
|
||||
.map_err(|_| ConfigError::InvalidField("host.version"))
|
||||
);
|
||||
let plugin_inspector = std::sync::Arc::new(gateway_plugin_runtime::PackageInspector::new(
|
||||
gateway_plugin_runtime::PackageLimits::default(),
|
||||
host_version,
|
||||
));
|
||||
let plugin_distribution = std::sync::Arc::new(or_shutdown!(
|
||||
plugin_runtime,
|
||||
gateway_host::plugin_distribution::HttpPluginDistribution::new(plugin_http)
|
||||
));
|
||||
let mut admin = or_shutdown!(
|
||||
plugin_runtime,
|
||||
gateway_admin::initialize_with_plugin_accounts(
|
||||
admin,
|
||||
client,
|
||||
store.admin_ports(),
|
||||
gateway_admin::AdminRuntimePorts {
|
||||
timezone,
|
||||
service_middleware: {
|
||||
let snapshots = core.snapshots();
|
||||
let middleware = plugin_runtime.middleware_registry();
|
||||
std::sync::Arc::new(move || {
|
||||
let snapshot = snapshots.snapshot_for_diagnostics()?;
|
||||
middleware.resolve(snapshot.extensions()?)
|
||||
})
|
||||
},
|
||||
providers: admin_providers,
|
||||
plugin_preparation: plugin_runtime.clone(),
|
||||
plugin_management: plugin_runtime.clone(),
|
||||
published_snapshot: core.snapshots(),
|
||||
plugin_inspector,
|
||||
plugin_distribution,
|
||||
pricing_source: std::sync::Arc::new(gateway_host::pricing::ModelsDevPricing),
|
||||
snapshot: core.snapshot_control(),
|
||||
account_probe: core.account_probe(),
|
||||
proxy_probe: host.proxy_probe(provider_openai::build_reqwest_client_with_custom_ca),
|
||||
client_distribution: host.client_distribution_resolver(),
|
||||
system: host.system_operations(),
|
||||
client_key_verifier: core.client_key_verifier(),
|
||||
},
|
||||
plugin_accounts,
|
||||
)
|
||||
.await
|
||||
);
|
||||
let official_files = host.official_plugin_release_files();
|
||||
let official_identity = host.official_plugin_release_identity();
|
||||
let official_import = or_shutdown!(
|
||||
plugin_runtime,
|
||||
admin
|
||||
.services()
|
||||
.plugins()
|
||||
.import_official_release(
|
||||
official_files.as_ref(),
|
||||
&official_identity,
|
||||
&gateway_admin::model::MutationContext {
|
||||
actor: gateway_admin::model::MutationActor::System,
|
||||
request_id: "startup-official-plugin-import".to_owned(),
|
||||
},
|
||||
)
|
||||
.await
|
||||
);
|
||||
if official_import.artifacts > 0 {
|
||||
host.report_startup_ready("Official plugins");
|
||||
}
|
||||
host.report_startup_ready("Admin");
|
||||
|
||||
let mut probes = store.health_probes();
|
||||
probes.extend(core.health_probes());
|
||||
probes.push(host.logging_health_probe());
|
||||
let api = or_shutdown!(
|
||||
plugin_runtime,
|
||||
gateway_api::initialize(
|
||||
api,
|
||||
core.execution_service(),
|
||||
admin.services(),
|
||||
probes,
|
||||
host.worker_health(),
|
||||
host.connection_lifecycle(),
|
||||
)
|
||||
)
|
||||
.with_middleware({
|
||||
let snapshots = core.snapshots();
|
||||
let middleware = plugin_runtime.middleware_registry();
|
||||
move |snapshot| {
|
||||
if let Some(snapshot) = snapshot {
|
||||
return middleware.resolve(snapshot.extensions()?);
|
||||
}
|
||||
// 数据面未就绪时，管理与诊断路由仍可使用发布候选中的插件。
|
||||
let diagnostic = snapshots.snapshot_for_diagnostics()?;
|
||||
middleware.resolve(diagnostic.extensions()?)
|
||||
}
|
||||
});
|
||||
let http_dispatcher = api.dispatcher();
|
||||
or_shutdown!(plugin_runtime, plugin_runtime.bind_http(&http_dispatcher));
|
||||
host.report_startup_ready("API");
|
||||
|
||||
or_shutdown!(
|
||||
plugin_runtime,
|
||||
plugin_runtime.bind_services(&admin.services().public_services())
|
||||
);
|
||||
|
||||
let mut plan = store.take_worker_contributions();
|
||||
plan.push(gateway_host::retention::worker(store.retention())?);
|
||||
plan.extend(core.take_worker_contributions());
|
||||
plan.extend(openai.take_worker_contributions());
|
||||
plan.extend(xai.take_worker_contributions());
|
||||
plan.extend(admin.take_worker_contributions());
|
||||
plan.push(plugin_runtime.maintenance_worker(core.snapshots())?);
|
||||
or_shutdown!(
|
||||
plugin_runtime,
|
||||
host.start_workers(plan, store.worker_leader_lease())
|
||||
);
|
||||
host.report_startup_ready("Workers");
|
||||
let served = host.serve(api.router()).await;
|
||||
plugin_runtime.shutdown().await;
|
||||
served?;
|
||||
Ok(gateway_plugin_runtime::PluginCommandOutput {
|
||||
stdout: String::new(),
|
||||
stderr: String::new(),
|
||||
exit_code: 0,
|
||||
saved_accounts: 0,
|
||||
})
|
||||
}
|
||||
|
||||
fn prepare_plugins(
|
||||
store: &gateway_store::StoreBundle,
|
||||
plugin_cache: std::path::PathBuf,
|
||||
version: &str,
|
||||
http: std::sync::Arc<gateway_host::outbound::HttpClient>,
|
||||
) -> Result<std::sync::Arc<gateway_plugin_runtime::PluginRuntime>, BootstrapError> {
|
||||
Ok(std::sync::Arc::new(
|
||||
gateway_plugin_runtime::PluginRuntime::new(
|
||||
store.admin_ports().plugins(),
|
||||
store.admin_ports().plugin_state(),
|
||||
gateway_plugin_runtime::PluginRuntimeConfig {
|
||||
cache_directory: plugin_cache,
|
||||
host_version: version
|
||||
.trim_start_matches('v')
|
||||
.parse()
|
||||
.map_err(|_| ConfigError::InvalidField("host.version"))?,
|
||||
package_limits: gateway_plugin_runtime::PackageLimits::default(),
|
||||
rpc_limits: gateway_plugin_runtime::RpcLimits::default(),
|
||||
restart_circuit: Default::default(),
|
||||
},
|
||||
http,
|
||||
std::sync::Arc::new(gateway_host::process::ProcessSupervisor::default()),
|
||||
)
|
||||
.with_oauth_pending(store.provider_ports().oauth_pending()),
|
||||
))
|
||||
}
|
||||
|
||||
/// 组合根只保留包级错误分类，不展开内部实现或敏感配置。
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum BootstrapError {
|
||||
#[error(transparent)]
|
||||
Worker(#[from] gateway_core::task::WorkerDefinitionError),
|
||||
#[error(transparent)]
|
||||
PluginCommand(#[from] gateway_plugin_runtime::PluginCommandError),
|
||||
#[error(transparent)]
|
||||
Config(#[from] gateway_host::ConfigError),
|
||||
#[error(transparent)]
|
||||
Host(#[from] gateway_host::HostError),
|
||||
#[error(transparent)]
|
||||
Outbound(#[from] gateway_host::outbound::HttpError),
|
||||
#[error(transparent)]
|
||||
Store(#[from] gateway_store::StoreError),
|
||||
#[error(transparent)]
|
||||
CommandStoreDrain(#[from] gateway_store::CommandStoreDrainError),
|
||||
#[error(transparent)]
|
||||
OpenAi(#[from] provider_openai::OpenAiInitializeError),
|
||||
#[error(transparent)]
|
||||
Xai(#[from] provider_xai::XaiInitializeError),
|
||||
#[error(transparent)]
|
||||
Registry(#[from] RegistryError),
|
||||
#[error(transparent)]
|
||||
Core(#[from] gateway_core::CoreError),
|
||||
#[error(transparent)]
|
||||
Admin(#[from] gateway_admin::model::AdminError),
|
||||
#[error(transparent)]
|
||||
Api(#[from] gateway_api::ApiError),
|
||||
}
|
||||
@@ -0,0 +1,134 @@
|
||||
//! 插件帮助只读启动；命令仅组装命令平面和必要写泵。
|
||||
|
||||
use super::{
|
||||
BootstrapError, GatewayConfig,
|
||||
plugins::{ManagementPorts, or_shutdown, prepare_plugins, service_middleware},
|
||||
startup::{Environment, Mode, Providers},
|
||||
};
|
||||
|
||||
/// 这里只接收插件命名空间内的参数，不转交宿主配置或其他启动 secret。
|
||||
pub struct PluginCommand {
|
||||
pub instance_id: Option<String>,
|
||||
pub name: Option<String>,
|
||||
pub arguments: Vec<String>,
|
||||
}
|
||||
|
||||
impl PluginCommand {
|
||||
fn is_help(&self) -> bool {
|
||||
self.name.is_none()
|
||||
|| self
|
||||
.arguments
|
||||
.iter()
|
||||
.any(|value| matches!(value.as_str(), "--help" | "-h"))
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn plugin_command(
|
||||
command: PluginCommand,
|
||||
) -> Result<gateway_plugin_runtime::PluginCommandOutput, BootstrapError> {
|
||||
let config = gateway_host::load_config::<GatewayConfig>()?;
|
||||
if command.is_help() {
|
||||
let environment = Environment::initialize(config.host, config.store, Mode::Help).await?;
|
||||
help(command, environment).await
|
||||
} else {
|
||||
let environment = Environment::initialize(config.host, config.store, Mode::Command).await?;
|
||||
let providers = Providers::initialize(config.openai, &environment).await?;
|
||||
execute(command, environment, providers).await
|
||||
}
|
||||
}
|
||||
|
||||
async fn help(
|
||||
command: PluginCommand,
|
||||
environment: Environment,
|
||||
) -> Result<gateway_plugin_runtime::PluginCommandOutput, BootstrapError> {
|
||||
let runtime = prepare_plugins(
|
||||
&environment.store,
|
||||
environment.plugin_cache,
|
||||
&environment.version,
|
||||
environment.plugin_http,
|
||||
)?;
|
||||
let commands = or_shutdown!(runtime, runtime.prepare_command_line().await);
|
||||
let stdout = commands.help(command.instance_id.as_deref(), command.name.as_deref());
|
||||
commands.shutdown().await;
|
||||
runtime.shutdown().await;
|
||||
Ok(gateway_plugin_runtime::PluginCommandOutput {
|
||||
stdout: stdout?,
|
||||
stderr: String::new(),
|
||||
exit_code: 0,
|
||||
saved_accounts: 0,
|
||||
})
|
||||
}
|
||||
|
||||
async fn execute(
|
||||
command: PluginCommand,
|
||||
environment: Environment,
|
||||
providers: Providers,
|
||||
) -> Result<gateway_plugin_runtime::PluginCommandOutput, BootstrapError> {
|
||||
let Environment {
|
||||
host,
|
||||
mut store,
|
||||
plugin_cache,
|
||||
version,
|
||||
plugin_http,
|
||||
..
|
||||
} = environment;
|
||||
let plugin_runtime = prepare_plugins(&store, plugin_cache, &version, plugin_http)?;
|
||||
let instance_id = or_shutdown!(
|
||||
plugin_runtime,
|
||||
command
|
||||
.instance_id
|
||||
.as_deref()
|
||||
.ok_or_else(|| gateway_admin::model::AdminError::invalid("插件实例 ID 缺失"))
|
||||
);
|
||||
let name = or_shutdown!(
|
||||
plugin_runtime,
|
||||
command
|
||||
.name
|
||||
.as_deref()
|
||||
.ok_or_else(|| gateway_admin::model::AdminError::invalid("插件命令名缺失"))
|
||||
);
|
||||
let core = gateway_core::prepare_command_plane(
|
||||
store.core_ports(),
|
||||
providers.core,
|
||||
Some(plugin_runtime.clone()),
|
||||
Some(plugin_runtime.observer_registry()),
|
||||
Some(plugin_runtime.policy_registry()),
|
||||
Some(plugin_runtime.middleware_registry()),
|
||||
);
|
||||
let management = ManagementPorts::new(&store, providers.admin.clone(), core.snapshot_control());
|
||||
or_shutdown!(plugin_runtime, management.bind(&plugin_runtime));
|
||||
let core = or_shutdown!(plugin_runtime, core.activate().await);
|
||||
let nested_models = core.nested_model_execution_port();
|
||||
let affinity_lookup = core.affinity_lookup_port();
|
||||
or_shutdown!(
|
||||
plugin_runtime,
|
||||
plugin_runtime.bind_model_ports(&nested_models, &affinity_lookup)
|
||||
);
|
||||
let settings = gateway_admin::initialize_settings(
|
||||
store.admin_ports().settings(),
|
||||
core.snapshot_control(),
|
||||
providers.admin,
|
||||
std::sync::Arc::new(gateway_host::pricing::ModelsDevPricing),
|
||||
);
|
||||
let mut services = gateway_admin::service::Registry::new(service_middleware(
|
||||
core.snapshots(),
|
||||
&plugin_runtime,
|
||||
));
|
||||
or_shutdown!(plugin_runtime, services.register_settings(&settings));
|
||||
let services = std::sync::Arc::new(services);
|
||||
or_shutdown!(plugin_runtime, plugin_runtime.bind_services(&services));
|
||||
let commands = or_shutdown!(plugin_runtime, plugin_runtime.prepare_command_line().await);
|
||||
let result = async {
|
||||
store.start_command_line_writes()?;
|
||||
Ok::<_, BootstrapError>(
|
||||
commands
|
||||
.execute_cancellable(instance_id, name, &command.arguments, &host.cancellation())
|
||||
.await?,
|
||||
)
|
||||
}
|
||||
.await;
|
||||
commands.shutdown().await;
|
||||
plugin_runtime.shutdown().await;
|
||||
store.shutdown_command_line_writes().await?;
|
||||
result
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
use gateway_host::{ConfigError, HostConfig, LoadableConfig};
|
||||
use serde::Deserialize;
|
||||
|
||||
const CONFIG_SCHEMA_VERSION: u32 = 1;
|
||||
|
||||
/// 顶层配置只组合各包拥有的配置段，不解释任何业务字段。
|
||||
/// 各启动配置忽略未知字段，允许升级时保留旧配置；已知字段仍按类型和业务约束校验。
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub struct GatewayConfig {
|
||||
schema_version: u32,
|
||||
pub(super) host: HostConfig,
|
||||
pub(super) store: gateway_store::StoreConfig,
|
||||
pub(super) admin: gateway_admin::AdminConfig,
|
||||
#[serde(default)]
|
||||
pub(super) client: gateway_admin::ClientConfig,
|
||||
pub(super) api: gateway_api::ApiConfig,
|
||||
#[serde(default)]
|
||||
pub(super) openai: provider_openai::OpenAiConfig,
|
||||
}
|
||||
|
||||
impl LoadableConfig for GatewayConfig {
|
||||
const EXTERNAL_SECTIONS: &'static [&'static str] = &["services"];
|
||||
|
||||
fn resolve_and_validate(&mut self, source_dir: &std::path::Path) -> Result<(), ConfigError> {
|
||||
if self.schema_version != CONFIG_SCHEMA_VERSION {
|
||||
return Err(ConfigError::InvalidField("schema_version"));
|
||||
}
|
||||
self.api
|
||||
.resolve_and_validate(source_dir)
|
||||
.map_err(|_| ConfigError::InvalidField("api"))?;
|
||||
self.host
|
||||
.resolve_and_validate(source_dir, &self.api.asset_directory)?;
|
||||
let runtime_data_dir = self.host.runtime_data_dir().to_path_buf();
|
||||
self.store
|
||||
.resolve_and_validate(&runtime_data_dir)
|
||||
.map_err(|_| ConfigError::InvalidField("store"))?;
|
||||
self.admin
|
||||
.resolve_and_validate(source_dir)
|
||||
.map_err(|_| ConfigError::InvalidField("admin"))?;
|
||||
self.client
|
||||
.resolve_and_validate(source_dir)
|
||||
.map_err(|_| ConfigError::InvalidField("client"))?;
|
||||
self.openai
|
||||
.resolve_and_validate(&runtime_data_dir)
|
||||
.map_err(|_| ConfigError::InvalidField("openai"))?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
//! 网关唯一组合根：按启动模式连接各 Bundle。
|
||||
|
||||
mod command;
|
||||
mod config;
|
||||
mod plugins;
|
||||
mod server;
|
||||
mod startup;
|
||||
|
||||
pub use command::{PluginCommand, plugin_command};
|
||||
pub use config::GatewayConfig;
|
||||
pub use server::run;
|
||||
|
||||
/// 组合根只保留包级错误分类，不展开内部实现或敏感配置。
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum BootstrapError {
|
||||
#[error(transparent)]
|
||||
Worker(#[from] gateway_core::task::WorkerDefinitionError),
|
||||
#[error(transparent)]
|
||||
PluginCommand(#[from] gateway_plugin_runtime::PluginCommandError),
|
||||
#[error(transparent)]
|
||||
Config(#[from] gateway_host::ConfigError),
|
||||
#[error(transparent)]
|
||||
Host(#[from] gateway_host::HostError),
|
||||
#[error(transparent)]
|
||||
Outbound(#[from] gateway_host::outbound::HttpError),
|
||||
#[error(transparent)]
|
||||
Store(#[from] gateway_store::StoreError),
|
||||
#[error(transparent)]
|
||||
CommandStoreDrain(#[from] gateway_store::CommandStoreDrainError),
|
||||
#[error(transparent)]
|
||||
OpenAi(#[from] provider_openai::OpenAiInitializeError),
|
||||
#[error(transparent)]
|
||||
Xai(#[from] provider_xai::XaiInitializeError),
|
||||
#[error(transparent)]
|
||||
Registry(#[from] gateway_core::engine::provider::RegistryError),
|
||||
#[error(transparent)]
|
||||
Core(#[from] gateway_core::CoreError),
|
||||
#[error(transparent)]
|
||||
Admin(#[from] gateway_admin::model::AdminError),
|
||||
#[error(transparent)]
|
||||
Api(#[from] gateway_api::ApiError),
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
//! 插件 Runtime 的组装与回调资源持有；服务和命令模式复用同一组管理端口。
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use gateway_admin::{
|
||||
model::AdminError,
|
||||
ports::{
|
||||
plugin_accounts::PluginAccountAccess, plugin_client_keys::PluginClientKeyAccess,
|
||||
plugin_resources::PluginResourceAccess, provider::ProviderAdminRegistry,
|
||||
},
|
||||
};
|
||||
use gateway_core::runtime::{RuntimeSnapshotHandle, SnapshotControl};
|
||||
use gateway_host::ConfigError;
|
||||
use gateway_plugin_runtime::PluginRuntime;
|
||||
|
||||
use super::BootstrapError;
|
||||
|
||||
// 必须在调用方仍持有 Core 和回调资源时异步关闭，不能先退出该作用域再清理。
|
||||
macro_rules! or_shutdown {
|
||||
($runtime:expr, $result:expr) => {
|
||||
match $result {
|
||||
Ok(value) => value,
|
||||
Err(error) => {
|
||||
$runtime.shutdown().await;
|
||||
return Err(error.into());
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
pub(super) use or_shutdown;
|
||||
|
||||
pub(super) struct ManagementPorts {
|
||||
// Runtime 保存 Weak；组合根持有这些 Arc，直到插件全部关闭。
|
||||
pub(super) accounts: Arc<dyn PluginAccountAccess>,
|
||||
keys: Arc<dyn PluginClientKeyAccess>,
|
||||
resources: Arc<dyn PluginResourceAccess>,
|
||||
}
|
||||
|
||||
impl ManagementPorts {
|
||||
pub(super) fn new(
|
||||
store: &gateway_store::StoreBundle,
|
||||
providers: ProviderAdminRegistry,
|
||||
snapshot: Arc<dyn SnapshotControl>,
|
||||
) -> Self {
|
||||
let ports = store.admin_ports();
|
||||
Self {
|
||||
accounts: gateway_admin::initialize_plugin_accounts(
|
||||
providers.clone(),
|
||||
ports.accounts(),
|
||||
snapshot.clone(),
|
||||
),
|
||||
keys: gateway_admin::initialize_plugin_client_keys(
|
||||
providers,
|
||||
ports.client_keys(),
|
||||
snapshot.clone(),
|
||||
),
|
||||
resources: gateway_admin::initialize_plugin_resources(
|
||||
ports.plugin_resources(),
|
||||
snapshot,
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn bind(&self, runtime: &PluginRuntime) -> Result<(), AdminError> {
|
||||
runtime.bind_account_ports(&self.accounts)?;
|
||||
runtime.bind_client_key_ports(&self.keys)?;
|
||||
runtime.bind_resource_ports(&self.resources)
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn service_middleware(
|
||||
snapshots: RuntimeSnapshotHandle,
|
||||
runtime: &PluginRuntime,
|
||||
) -> gateway_admin::service::PlanSource {
|
||||
let middleware = runtime.middleware_registry();
|
||||
Arc::new(move || {
|
||||
let snapshot = snapshots.snapshot_for_diagnostics()?;
|
||||
middleware.resolve(snapshot.extensions()?)
|
||||
})
|
||||
}
|
||||
|
||||
pub(super) fn prepare_plugins(
|
||||
store: &gateway_store::StoreBundle,
|
||||
plugin_cache: std::path::PathBuf,
|
||||
version: &str,
|
||||
http: Arc<gateway_host::outbound::HttpClient>,
|
||||
) -> Result<Arc<gateway_plugin_runtime::PluginRuntime>, BootstrapError> {
|
||||
Ok(Arc::new(
|
||||
gateway_plugin_runtime::PluginRuntime::new(
|
||||
store.admin_ports().plugins(),
|
||||
store.admin_ports().plugin_state(),
|
||||
gateway_plugin_runtime::PluginRuntimeConfig {
|
||||
cache_directory: plugin_cache,
|
||||
host_version: version
|
||||
.trim_start_matches('v')
|
||||
.parse()
|
||||
.map_err(|_| ConfigError::InvalidField("host.version"))?,
|
||||
package_limits: gateway_plugin_runtime::PackageLimits::default(),
|
||||
rpc_limits: gateway_plugin_runtime::RpcLimits::default(),
|
||||
restart_circuit: Default::default(),
|
||||
},
|
||||
http,
|
||||
Arc::new(gateway_host::process::ProcessSupervisor::default()),
|
||||
)
|
||||
.with_oauth_pending(store.provider_ports().oauth_pending()),
|
||||
))
|
||||
}
|
||||
@@ -0,0 +1,164 @@
|
||||
//! 服务模式完成全部组装后交给 Host 监听，并在请求和 Worker 排空后关闭插件。
|
||||
|
||||
use gateway_host::ConfigError;
|
||||
|
||||
use super::{
|
||||
BootstrapError, GatewayConfig,
|
||||
plugins::{ManagementPorts, or_shutdown, prepare_plugins, service_middleware},
|
||||
startup::{Environment, Mode, Providers},
|
||||
};
|
||||
|
||||
/// 按冻结顺序初始化全部 Bundle，并把进程阻塞权交给 Host。
|
||||
pub async fn run() -> Result<(), BootstrapError> {
|
||||
let config = gateway_host::load_config::<GatewayConfig>()?;
|
||||
let environment = Environment::initialize(config.host, config.store, Mode::Server).await?;
|
||||
let mut providers = Providers::initialize(config.openai, &environment).await?;
|
||||
let Environment {
|
||||
host,
|
||||
mut store,
|
||||
timezone,
|
||||
plugin_cache,
|
||||
version,
|
||||
plugin_http,
|
||||
} = environment;
|
||||
let plugin_runtime = prepare_plugins(&store, plugin_cache, &version, plugin_http.clone())?;
|
||||
let core = gateway_core::prepare(
|
||||
store.core_ports(),
|
||||
providers.core,
|
||||
Some(plugin_runtime.clone()),
|
||||
Some(plugin_runtime.observer_registry()),
|
||||
Some(plugin_runtime.policy_registry()),
|
||||
Some(plugin_runtime.middleware_registry()),
|
||||
Some(plugin_runtime.frontend_authentication_registry()),
|
||||
);
|
||||
let management = ManagementPorts::new(&store, providers.admin.clone(), core.snapshot_control());
|
||||
or_shutdown!(plugin_runtime, management.bind(&plugin_runtime));
|
||||
let mut core = or_shutdown!(plugin_runtime, core.activate().await);
|
||||
let nested_models = core.nested_model_execution_port();
|
||||
let affinity_lookup = core.affinity_lookup_port();
|
||||
or_shutdown!(
|
||||
plugin_runtime,
|
||||
plugin_runtime.bind_model_ports(&nested_models, &affinity_lookup)
|
||||
);
|
||||
host.report_startup_ready("Core");
|
||||
let host_version = or_shutdown!(
|
||||
plugin_runtime,
|
||||
version
|
||||
.trim_start_matches('v')
|
||||
.parse()
|
||||
.map_err(|_| ConfigError::InvalidField("host.version"))
|
||||
);
|
||||
let plugin_inspector = std::sync::Arc::new(gateway_plugin_runtime::PackageInspector::new(
|
||||
gateway_plugin_runtime::PackageLimits::default(),
|
||||
host_version,
|
||||
));
|
||||
let plugin_distribution = std::sync::Arc::new(or_shutdown!(
|
||||
plugin_runtime,
|
||||
gateway_host::plugin_distribution::HttpPluginDistribution::new(plugin_http)
|
||||
));
|
||||
let mut admin = or_shutdown!(
|
||||
plugin_runtime,
|
||||
gateway_admin::initialize_with_plugin_accounts(
|
||||
config.admin,
|
||||
config.client,
|
||||
store.admin_ports(),
|
||||
gateway_admin::AdminRuntimePorts {
|
||||
timezone,
|
||||
service_middleware: service_middleware(core.snapshots(), &plugin_runtime),
|
||||
providers: providers.admin,
|
||||
plugin_preparation: plugin_runtime.clone(),
|
||||
plugin_management: plugin_runtime.clone(),
|
||||
published_snapshot: core.snapshots(),
|
||||
plugin_inspector,
|
||||
plugin_distribution,
|
||||
pricing_source: std::sync::Arc::new(gateway_host::pricing::ModelsDevPricing),
|
||||
snapshot: core.snapshot_control(),
|
||||
account_probe: core.account_probe(),
|
||||
proxy_probe: host.proxy_probe(provider_openai::build_reqwest_client_with_custom_ca),
|
||||
client_distribution: host.client_distribution_resolver(),
|
||||
system: host.system_operations(),
|
||||
client_key_verifier: core.client_key_verifier(),
|
||||
},
|
||||
management.accounts.clone(),
|
||||
)
|
||||
.await
|
||||
);
|
||||
let official_files = host.official_plugin_release_files();
|
||||
let official_identity = host.official_plugin_release_identity();
|
||||
let official_import = or_shutdown!(
|
||||
plugin_runtime,
|
||||
admin
|
||||
.services()
|
||||
.plugins()
|
||||
.import_official_release(
|
||||
official_files.as_ref(),
|
||||
&official_identity,
|
||||
&gateway_admin::model::MutationContext {
|
||||
actor: gateway_admin::model::MutationActor::System,
|
||||
request_id: "startup-official-plugin-import".to_owned(),
|
||||
},
|
||||
)
|
||||
.await
|
||||
);
|
||||
if official_import.artifacts > 0 {
|
||||
host.report_startup_ready("Official plugins");
|
||||
}
|
||||
host.report_startup_ready("Admin");
|
||||
|
||||
let mut probes = store.health_probes();
|
||||
probes.extend(core.health_probes());
|
||||
probes.push(host.logging_health_probe());
|
||||
let api = or_shutdown!(
|
||||
plugin_runtime,
|
||||
gateway_api::initialize(
|
||||
config.api,
|
||||
core.execution_service(),
|
||||
admin.services(),
|
||||
probes,
|
||||
host.worker_health(),
|
||||
host.connection_lifecycle(),
|
||||
)
|
||||
)
|
||||
.with_middleware({
|
||||
let snapshots = core.snapshots();
|
||||
let middleware = plugin_runtime.middleware_registry();
|
||||
move |snapshot| {
|
||||
if let Some(snapshot) = snapshot {
|
||||
return middleware.resolve(snapshot.extensions()?);
|
||||
}
|
||||
// 数据面未就绪时，管理与诊断路由仍可使用发布候选中的插件。
|
||||
let diagnostic = snapshots.snapshot_for_diagnostics()?;
|
||||
middleware.resolve(diagnostic.extensions()?)
|
||||
}
|
||||
});
|
||||
let http_dispatcher = api.dispatcher();
|
||||
or_shutdown!(plugin_runtime, plugin_runtime.bind_http(&http_dispatcher));
|
||||
host.report_startup_ready("API");
|
||||
|
||||
or_shutdown!(
|
||||
plugin_runtime,
|
||||
plugin_runtime.bind_services(&admin.services().public_services())
|
||||
);
|
||||
|
||||
let mut plan = store.take_worker_contributions();
|
||||
plan.push(or_shutdown!(
|
||||
plugin_runtime,
|
||||
gateway_host::retention::worker(store.retention())
|
||||
));
|
||||
plan.extend(core.take_worker_contributions());
|
||||
plan.extend(providers.openai.take_worker_contributions());
|
||||
plan.extend(providers.xai.take_worker_contributions());
|
||||
plan.extend(admin.take_worker_contributions());
|
||||
plan.push(or_shutdown!(
|
||||
plugin_runtime,
|
||||
plugin_runtime.maintenance_worker(core.snapshots())
|
||||
));
|
||||
or_shutdown!(
|
||||
plugin_runtime,
|
||||
host.start_workers(plan, store.worker_leader_lease())
|
||||
);
|
||||
host.report_startup_ready("Workers");
|
||||
let served = host.serve(api.router()).await;
|
||||
plugin_runtime.shutdown().await;
|
||||
Ok(served?)
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
//! 启动模式共享的基础设施与 Provider 组装，不启动 HTTP 或 Worker。
|
||||
|
||||
use std::{path::PathBuf, sync::Arc};
|
||||
|
||||
use gateway_admin::ports::provider::ProviderAdminRegistry;
|
||||
use gateway_core::{engine::provider::ProviderRegistry, time::DeploymentTimeZone};
|
||||
use gateway_host::{ConfigError, HostBundle, HostConfig, outbound::HttpClient};
|
||||
use gateway_store::{StoreBundle, StoreConfig};
|
||||
|
||||
use super::BootstrapError;
|
||||
|
||||
pub(super) enum Mode {
|
||||
Server,
|
||||
Command,
|
||||
Help,
|
||||
}
|
||||
|
||||
pub(super) struct Environment {
|
||||
pub(super) host: HostBundle,
|
||||
pub(super) store: StoreBundle,
|
||||
pub(super) timezone: DeploymentTimeZone,
|
||||
pub(super) plugin_cache: PathBuf,
|
||||
pub(super) version: String,
|
||||
pub(super) plugin_http: Arc<HttpClient>,
|
||||
}
|
||||
|
||||
impl Environment {
|
||||
pub(super) async fn initialize(
|
||||
host: HostConfig,
|
||||
store: StoreConfig,
|
||||
mode: Mode,
|
||||
) -> Result<Self, BootstrapError> {
|
||||
let timezone = host.timezone;
|
||||
let store = store.with_timezone(timezone);
|
||||
let plugin_cache = host.runtime_data_dir().join("plugins");
|
||||
let host = match mode {
|
||||
Mode::Server => gateway_host::initialize(host).await?,
|
||||
Mode::Command | Mode::Help => gateway_host::initialize_command_line(host).await?,
|
||||
};
|
||||
host.report_startup_ready("Host");
|
||||
let store = match mode {
|
||||
Mode::Server => gateway_store::initialize(store).await?,
|
||||
Mode::Command => gateway_store::initialize_command_line(store).await?,
|
||||
Mode::Help => gateway_store::initialize_read_only(store).await?,
|
||||
};
|
||||
host.report_startup_ready("Store");
|
||||
let version = host
|
||||
.system_operations()
|
||||
.version()
|
||||
.await
|
||||
.map_err(|_| ConfigError::InvalidField("host.version"))?
|
||||
.version;
|
||||
let plugin_http = Arc::new(HttpClient::new()?);
|
||||
Ok(Self {
|
||||
host,
|
||||
store,
|
||||
timezone,
|
||||
plugin_cache,
|
||||
version,
|
||||
plugin_http,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) struct Providers {
|
||||
pub(super) openai: provider_openai::ProviderBundle,
|
||||
pub(super) xai: provider_xai::ProviderBundle,
|
||||
pub(super) core: ProviderRegistry,
|
||||
pub(super) admin: ProviderAdminRegistry,
|
||||
}
|
||||
|
||||
impl Providers {
|
||||
pub(super) async fn initialize(
|
||||
config: provider_openai::OpenAiConfig,
|
||||
environment: &Environment,
|
||||
) -> Result<Self, BootstrapError> {
|
||||
let ports = environment.store.provider_ports();
|
||||
let openai =
|
||||
provider_openai::initialize(config.with_timezone(environment.timezone), ports.clone())
|
||||
.await?;
|
||||
environment.host.report_startup_ready("OpenAI Provider");
|
||||
let xai = provider_xai::initialize(ports).await?;
|
||||
environment.host.report_startup_ready("xAI Provider");
|
||||
let core = ProviderRegistry::new([openai.core_provider(), xai.core_provider()])?;
|
||||
let admin = ProviderAdminRegistry::new([openai.admin_provider(), xai.admin_provider()])
|
||||
.map_err(|_| gateway_admin::model::AdminError::invalid("Provider 管理身份冲突"))?;
|
||||
Ok(Self {
|
||||
openai,
|
||||
xai,
|
||||
core,
|
||||
admin,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
use std::{fs, process::Command};
|
||||
|
||||
#[test]
|
||||
fn basic_cli_options_do_not_load_configuration_and_reject_unexpected_arguments() {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
fs::create_dir(directory.path().join("deploy")).unwrap();
|
||||
fs::write(directory.path().join("deploy/config.yaml"), "invalid: [").unwrap();
|
||||
for flag in ["--help", "-h", "help", "--version", "-V"] {
|
||||
let output = Command::new(env!("CARGO_BIN_EXE_codex-proxy-rs"))
|
||||
.arg(flag)
|
||||
.current_dir(directory.path())
|
||||
.env_clear()
|
||||
.output()
|
||||
.unwrap();
|
||||
assert!(output.status.success(), "{flag}");
|
||||
assert!(!output.stdout.is_empty());
|
||||
assert!(output.stderr.is_empty());
|
||||
}
|
||||
for args in [
|
||||
vec!["unknown"],
|
||||
vec!["--help", "extra"],
|
||||
vec!["serve", "extra"],
|
||||
] {
|
||||
let output = Command::new(env!("CARGO_BIN_EXE_codex-proxy-rs"))
|
||||
.args(args)
|
||||
.current_dir(directory.path())
|
||||
.env_clear()
|
||||
.output()
|
||||
.unwrap();
|
||||
assert!(!output.status.success());
|
||||
assert!(output.stdout.is_empty());
|
||||
assert!(!String::from_utf8_lossy(&output.stderr).contains("configuration"));
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn non_utf8_command_is_not_interpreted_as_serve() {
|
||||
use std::os::unix::ffi::OsStringExt as _;
|
||||
|
||||
let output = Command::new(env!("CARGO_BIN_EXE_codex-proxy-rs"))
|
||||
.arg(std::ffi::OsString::from_vec(vec![0xff]))
|
||||
.env_clear()
|
||||
.output()
|
||||
.unwrap();
|
||||
assert!(!output.status.success());
|
||||
assert!(output.stdout.is_empty());
|
||||
assert!(String::from_utf8_lossy(&output.stderr).contains("command must be UTF-8"));
|
||||
}
|
||||
+4
-197
@@ -3,205 +3,12 @@ use std::{fs, process::Command};
|
||||
use codex_proxy_rs::bootstrap::GatewayConfig;
|
||||
use gateway_host::LoadableConfig;
|
||||
|
||||
const CONFIG_EXAMPLE: &str = include_str!("../../../../deploy/config.example.yaml");
|
||||
const CONFIG_EXAMPLE: &str = include_str!("../../../../../deploy/config.example.yaml");
|
||||
const POSTGRES_PASSWORD: &str = "111111111111111111111111111111111111111111111111";
|
||||
const REDIS_PASSWORD: &str = "222222222222222222222222222222222222222222222222";
|
||||
const ADMIN_PASSWORD: &str = "test-admin-password";
|
||||
const TOPOLOGY_CHILD_ENV: &str = "CPR_TEST_TOPOLOGY_CHILD";
|
||||
|
||||
#[test]
|
||||
fn basic_cli_options_do_not_load_configuration_and_reject_unexpected_arguments() {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
fs::create_dir(directory.path().join("deploy")).unwrap();
|
||||
fs::write(directory.path().join("deploy/config.yaml"), "invalid: [").unwrap();
|
||||
for flag in ["--help", "-h", "help", "--version", "-V"] {
|
||||
let output = Command::new(env!("CARGO_BIN_EXE_codex-proxy-rs"))
|
||||
.arg(flag)
|
||||
.current_dir(directory.path())
|
||||
.env_clear()
|
||||
.output()
|
||||
.unwrap();
|
||||
assert!(output.status.success(), "{flag}");
|
||||
assert!(!output.stdout.is_empty());
|
||||
assert!(output.stderr.is_empty());
|
||||
}
|
||||
for args in [
|
||||
vec!["unknown"],
|
||||
vec!["--help", "extra"],
|
||||
vec!["serve", "extra"],
|
||||
] {
|
||||
let output = Command::new(env!("CARGO_BIN_EXE_codex-proxy-rs"))
|
||||
.args(args)
|
||||
.current_dir(directory.path())
|
||||
.env_clear()
|
||||
.output()
|
||||
.unwrap();
|
||||
assert!(!output.status.success());
|
||||
assert!(output.stdout.is_empty());
|
||||
assert!(!String::from_utf8_lossy(&output.stderr).contains("configuration"));
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn non_utf8_command_is_not_interpreted_as_serve() {
|
||||
use std::os::unix::ffi::OsStringExt as _;
|
||||
|
||||
let output = Command::new(env!("CARGO_BIN_EXE_codex-proxy-rs"))
|
||||
.arg(std::ffi::OsString::from_vec(vec![0xff]))
|
||||
.env_clear()
|
||||
.output()
|
||||
.unwrap();
|
||||
assert!(!output.status.success());
|
||||
assert!(output.stdout.is_empty());
|
||||
assert!(String::from_utf8_lossy(&output.stderr).contains("command must be UTF-8"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn proxy_probe_should_use_provider_custom_ca_for_https_proxies() {
|
||||
use gateway_admin::ports::proxy::ProxyProbe;
|
||||
use gateway_core::account::OutboundProxy;
|
||||
use gateway_host::proxy_probe::HttpProxyProbe;
|
||||
use std::{sync::Arc, time::Duration};
|
||||
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
|
||||
use tokio_rustls::{
|
||||
TlsAcceptor,
|
||||
rustls::{
|
||||
ServerConfig,
|
||||
pki_types::{CertificateDer, PrivateKeyDer, pem::PemObject as _},
|
||||
},
|
||||
};
|
||||
|
||||
const CHILD_ENV: &str = "CPR_TEST_PROXY_TLS_DIRECTORY";
|
||||
let Ok(directory) = std::env::var(CHILD_ENV) else {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
generate_proxy_test_certificates(directory.path());
|
||||
// 使用子进程隔离环境变量，避免并行测试读取到临时 CA 配置。
|
||||
for ca_env in ["CODEX_CA_CERTIFICATE", "SSL_CERT_FILE"] {
|
||||
let mut child = Command::new(std::env::current_exe().unwrap());
|
||||
child
|
||||
.args([
|
||||
"--exact",
|
||||
"bootstrap::proxy_probe_should_use_provider_custom_ca_for_https_proxies",
|
||||
"--nocapture",
|
||||
])
|
||||
.env(CHILD_ENV, directory.path())
|
||||
.env_remove("CODEX_CA_CERTIFICATE")
|
||||
.env_remove("SSL_CERT_FILE")
|
||||
.env(ca_env, directory.path().join("ca.pem"));
|
||||
if ca_env == "CODEX_CA_CERTIFICATE" {
|
||||
child.env(
|
||||
"SSL_CERT_FILE",
|
||||
directory.path().join("missing-fallback.pem"),
|
||||
);
|
||||
}
|
||||
let output = child.output().unwrap();
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"{ca_env}: {}\n{}",
|
||||
String::from_utf8_lossy(&output.stdout),
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
}
|
||||
return;
|
||||
};
|
||||
provider_openai::ensure_rustls_provider();
|
||||
let directory = std::path::Path::new(&directory);
|
||||
let certificate = CertificateDer::from_pem_file(directory.join("server.pem")).unwrap();
|
||||
let key = PrivateKeyDer::from_pem_file(directory.join("server.key")).unwrap();
|
||||
let acceptor = TlsAcceptor::from(Arc::new(
|
||||
ServerConfig::builder()
|
||||
.with_no_client_auth()
|
||||
.with_single_cert(vec![certificate], key)
|
||||
.unwrap(),
|
||||
));
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let proxy =
|
||||
OutboundProxy::parse(&format!("https://{}", listener.local_addr().unwrap())).unwrap();
|
||||
let server = tokio::spawn(async move {
|
||||
let (socket, _) = listener.accept().await.unwrap();
|
||||
let mut stream = acceptor.accept(socket).await.unwrap();
|
||||
let mut request = Vec::new();
|
||||
while !request.ends_with(b"\r\n\r\n") {
|
||||
assert!(request.len() < 8192);
|
||||
request.push(stream.read_u8().await.unwrap());
|
||||
}
|
||||
assert!(request.starts_with(b"GET http://unresolvable.invalid/ip "));
|
||||
let body = "{\"ip\":\"203.0.113.8\"}";
|
||||
stream.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap();
|
||||
stream.shutdown().await.unwrap();
|
||||
});
|
||||
let probe = HttpProxyProbe::new("http://unresolvable.invalid/ip")
|
||||
.with_client_builder(provider_openai::build_reqwest_client_with_custom_ca);
|
||||
let result = probe.test(&proxy, false).await;
|
||||
assert!(result.success, "{}", result.message);
|
||||
assert_eq!(result.exit_ip.unwrap().to_string(), "203.0.113.8");
|
||||
tokio::time::timeout(Duration::from_secs(5), server)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
fn generate_proxy_test_certificates(directory: &std::path::Path) {
|
||||
let openssl = |args: &[&str]| {
|
||||
let output = Command::new("openssl")
|
||||
.args(args)
|
||||
.current_dir(directory)
|
||||
.output()
|
||||
.unwrap();
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"{}",
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
};
|
||||
openssl(&[
|
||||
"req",
|
||||
"-x509",
|
||||
"-newkey",
|
||||
"rsa:2048",
|
||||
"-nodes",
|
||||
"-keyout",
|
||||
"ca.key",
|
||||
"-out",
|
||||
"ca.pem",
|
||||
"-days",
|
||||
"1",
|
||||
"-subj",
|
||||
"/CN=Proxy Test CA",
|
||||
]);
|
||||
openssl(&[
|
||||
"req",
|
||||
"-newkey",
|
||||
"rsa:2048",
|
||||
"-nodes",
|
||||
"-keyout",
|
||||
"server.key",
|
||||
"-out",
|
||||
"server.csr",
|
||||
"-subj",
|
||||
"/CN=localhost",
|
||||
]);
|
||||
fs::write(directory.join("extensions"), "subjectAltName=IP:127.0.0.1\nbasicConstraints=critical,CA:FALSE\nkeyUsage=digitalSignature,keyEncipherment\nextendedKeyUsage=serverAuth\n").unwrap();
|
||||
openssl(&[
|
||||
"x509",
|
||||
"-req",
|
||||
"-in",
|
||||
"server.csr",
|
||||
"-CA",
|
||||
"ca.pem",
|
||||
"-CAkey",
|
||||
"ca.key",
|
||||
"-CAcreateserial",
|
||||
"-out",
|
||||
"server.pem",
|
||||
"-days",
|
||||
"1",
|
||||
"-extfile",
|
||||
"extensions",
|
||||
]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn config_loader_should_load_complete_terminal_example() {
|
||||
parse_config(&valid_config()).expect("terminal config example");
|
||||
@@ -230,7 +37,7 @@ fn config_loader_should_share_resolved_assets_with_system_update() {
|
||||
child
|
||||
.args([
|
||||
"--exact",
|
||||
"bootstrap::config_loader_should_share_resolved_assets_with_system_update",
|
||||
"bootstrap::config::config_loader_should_share_resolved_assets_with_system_update",
|
||||
])
|
||||
.env(CHILD_ENV, case)
|
||||
.env_remove("CPR_WEB_DIST_DIR");
|
||||
@@ -292,7 +99,7 @@ fn config_loader_should_apply_only_explicit_topology_overrides() {
|
||||
let status = Command::new(std::env::current_exe().expect("current test executable"))
|
||||
.args([
|
||||
"--exact",
|
||||
"bootstrap::config_loader_should_apply_only_explicit_topology_overrides",
|
||||
"bootstrap::config::config_loader_should_apply_only_explicit_topology_overrides",
|
||||
])
|
||||
.env(TOPOLOGY_CHILD_ENV, "1")
|
||||
.env("CPR_SERVER_HOST", "127.0.0.1")
|
||||
@@ -410,7 +217,7 @@ fn config_loader_should_report_startup_configuration_diagnostics() {
|
||||
let output = Command::new(std::env::current_exe().unwrap())
|
||||
.args([
|
||||
"--exact",
|
||||
"bootstrap::config_loader_should_report_startup_configuration_diagnostics",
|
||||
"bootstrap::config::config_loader_should_report_startup_configuration_diagnostics",
|
||||
"--nocapture",
|
||||
])
|
||||
.env(CHILD_ENV, "1")
|
||||
@@ -0,0 +1,3 @@
|
||||
mod command;
|
||||
mod config;
|
||||
mod server;
|
||||
@@ -0,0 +1,146 @@
|
||||
use std::{fs, process::Command};
|
||||
|
||||
#[tokio::test]
|
||||
async fn proxy_probe_should_use_provider_custom_ca_for_https_proxies() {
|
||||
use gateway_admin::ports::proxy::ProxyProbe;
|
||||
use gateway_core::account::OutboundProxy;
|
||||
use gateway_host::proxy_probe::HttpProxyProbe;
|
||||
use std::{sync::Arc, time::Duration};
|
||||
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
|
||||
use tokio_rustls::{
|
||||
TlsAcceptor,
|
||||
rustls::{
|
||||
ServerConfig,
|
||||
pki_types::{CertificateDer, PrivateKeyDer, pem::PemObject as _},
|
||||
},
|
||||
};
|
||||
|
||||
const CHILD_ENV: &str = "CPR_TEST_PROXY_TLS_DIRECTORY";
|
||||
let Ok(directory) = std::env::var(CHILD_ENV) else {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
generate_proxy_test_certificates(directory.path());
|
||||
// 使用子进程隔离环境变量，避免并行测试读取到临时 CA 配置。
|
||||
for ca_env in ["CODEX_CA_CERTIFICATE", "SSL_CERT_FILE"] {
|
||||
let mut child = Command::new(std::env::current_exe().unwrap());
|
||||
child
|
||||
.args([
|
||||
"--exact",
|
||||
"bootstrap::server::proxy_probe_should_use_provider_custom_ca_for_https_proxies",
|
||||
"--nocapture",
|
||||
])
|
||||
.env(CHILD_ENV, directory.path())
|
||||
.env_remove("CODEX_CA_CERTIFICATE")
|
||||
.env_remove("SSL_CERT_FILE")
|
||||
.env(ca_env, directory.path().join("ca.pem"));
|
||||
if ca_env == "CODEX_CA_CERTIFICATE" {
|
||||
child.env(
|
||||
"SSL_CERT_FILE",
|
||||
directory.path().join("missing-fallback.pem"),
|
||||
);
|
||||
}
|
||||
let output = child.output().unwrap();
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"{ca_env}: {}\n{}",
|
||||
String::from_utf8_lossy(&output.stdout),
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
}
|
||||
return;
|
||||
};
|
||||
provider_openai::ensure_rustls_provider();
|
||||
let directory = std::path::Path::new(&directory);
|
||||
let certificate = CertificateDer::from_pem_file(directory.join("server.pem")).unwrap();
|
||||
let key = PrivateKeyDer::from_pem_file(directory.join("server.key")).unwrap();
|
||||
let acceptor = TlsAcceptor::from(Arc::new(
|
||||
ServerConfig::builder()
|
||||
.with_no_client_auth()
|
||||
.with_single_cert(vec![certificate], key)
|
||||
.unwrap(),
|
||||
));
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let proxy =
|
||||
OutboundProxy::parse(&format!("https://{}", listener.local_addr().unwrap())).unwrap();
|
||||
let server = tokio::spawn(async move {
|
||||
let (socket, _) = listener.accept().await.unwrap();
|
||||
let mut stream = acceptor.accept(socket).await.unwrap();
|
||||
let mut request = Vec::new();
|
||||
while !request.ends_with(b"\r\n\r\n") {
|
||||
assert!(request.len() < 8192);
|
||||
request.push(stream.read_u8().await.unwrap());
|
||||
}
|
||||
assert!(request.starts_with(b"GET http://unresolvable.invalid/ip "));
|
||||
let body = "{\"ip\":\"203.0.113.8\"}";
|
||||
stream.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap();
|
||||
stream.shutdown().await.unwrap();
|
||||
});
|
||||
let probe = HttpProxyProbe::new("http://unresolvable.invalid/ip")
|
||||
.with_client_builder(provider_openai::build_reqwest_client_with_custom_ca);
|
||||
let result = probe.test(&proxy, false).await;
|
||||
assert!(result.success, "{}", result.message);
|
||||
assert_eq!(result.exit_ip.unwrap().to_string(), "203.0.113.8");
|
||||
tokio::time::timeout(Duration::from_secs(5), server)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
fn generate_proxy_test_certificates(directory: &std::path::Path) {
|
||||
let openssl = |args: &[&str]| {
|
||||
let output = Command::new("openssl")
|
||||
.args(args)
|
||||
.current_dir(directory)
|
||||
.output()
|
||||
.unwrap();
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"{}",
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
};
|
||||
openssl(&[
|
||||
"req",
|
||||
"-x509",
|
||||
"-newkey",
|
||||
"rsa:2048",
|
||||
"-nodes",
|
||||
"-keyout",
|
||||
"ca.key",
|
||||
"-out",
|
||||
"ca.pem",
|
||||
"-days",
|
||||
"1",
|
||||
"-subj",
|
||||
"/CN=Proxy Test CA",
|
||||
]);
|
||||
openssl(&[
|
||||
"req",
|
||||
"-newkey",
|
||||
"rsa:2048",
|
||||
"-nodes",
|
||||
"-keyout",
|
||||
"server.key",
|
||||
"-out",
|
||||
"server.csr",
|
||||
"-subj",
|
||||
"/CN=localhost",
|
||||
]);
|
||||
fs::write(directory.join("extensions"), "subjectAltName=IP:127.0.0.1\nbasicConstraints=critical,CA:FALSE\nkeyUsage=digitalSignature,keyEncipherment\nextendedKeyUsage=serverAuth\n").unwrap();
|
||||
openssl(&[
|
||||
"x509",
|
||||
"-req",
|
||||
"-in",
|
||||
"server.csr",
|
||||
"-CA",
|
||||
"ca.pem",
|
||||
"-CAkey",
|
||||
"ca.key",
|
||||
"-CAcreateserial",
|
||||
"-out",
|
||||
"server.pem",
|
||||
"-days",
|
||||
"1",
|
||||
"-extfile",
|
||||
"extensions",
|
||||
]);
|
||||
}
|
||||
@@ -18,7 +18,12 @@ fn app_tree_matches_frozen_terminal_manifest() {
|
||||
assert_eq!(
|
||||
rust_files(&root.join("src")),
|
||||
BTreeSet::from([
|
||||
PathBuf::from("bootstrap.rs"),
|
||||
PathBuf::from("bootstrap/command.rs"),
|
||||
PathBuf::from("bootstrap/config.rs"),
|
||||
PathBuf::from("bootstrap/mod.rs"),
|
||||
PathBuf::from("bootstrap/plugins.rs"),
|
||||
PathBuf::from("bootstrap/server.rs"),
|
||||
PathBuf::from("bootstrap/startup.rs"),
|
||||
PathBuf::from("lib.rs"),
|
||||
PathBuf::from("main.rs"),
|
||||
]),
|
||||
@@ -27,7 +32,10 @@ fn app_tree_matches_frozen_terminal_manifest() {
|
||||
rust_files(&root.join("tests")),
|
||||
BTreeSet::from([
|
||||
PathBuf::from("architecture.rs"),
|
||||
PathBuf::from("bootstrap.rs"),
|
||||
PathBuf::from("bootstrap/command.rs"),
|
||||
PathBuf::from("bootstrap/config.rs"),
|
||||
PathBuf::from("bootstrap/mod.rs"),
|
||||
PathBuf::from("bootstrap/server.rs"),
|
||||
PathBuf::from("main.rs"),
|
||||
]),
|
||||
);
|
||||
|
||||
Reference in New Issue
Block a user