feat: delegate 客户端替换为 hyper 原生实现，新增代理管理功能

aether-proxy:
- 用 hyper-util Client + 自定义 InstrumentedConnector 替换 reqwest delegate 客户端
- 支持 HTTP/HTTPS 自动 TLS，ALPN h2 协商，connect/tls 分阶段计时
- ConnectTiming 通过 hyper extensions 传递，上游响应细分 connect_ms/tls_ms/ttfb_ms
- upgrade 命令在非 root 下跳过 systemd restart 并提示手动操作

后端:
- 新增 /admin/proxy-nodes/test-url 接口，支持直接测试代理 URL 连通性
- 新增 /admin/proxy-nodes/hmac-key 接口，获取 HMAC Key 供部署使用
- 提取 _test_proxy_connectivity 公共函数，消除 test_node 中的重复代码
- candidate_resolver 在 extra_data 中输出 needs_conversion/provider_api_format
- stats_aggregator 小时聚合增加 IntegrityError 冲突重试

前端:
- 请求时间线组件展示代理 timing 细分（DNS/连接/TLS/TTFB/上游处理）
- 请求时间线增加格式转换分界标记和 conversion badge
- ProxyNodes 页面新增代理 URL 测试和 HMAC Key 复制功能
- HardwareTooltip 从 Popover 改为 Tooltip 组件

aether-proxy/Cargo.lock

@@ -10,7 +10,7 @@ checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
 
 [[package]]
 name = "aether-proxy"
-version = "0.1.3"
+version = "0.1.4"
 dependencies = [
  "anyhow",
  "base64",
@@ -41,9 +41,11 @@ dependencies = [
  "tokio",
  "tokio-rustls",
  "toml",
+ "tower-service",
  "tracing",
  "tracing-subscriber",
  "url",
+ "webpki-roots",
 ]
 
 [[package]]

aether-proxy/Cargo.toml

@@ -7,7 +7,8 @@ description = "Forward proxy for Aether with HMAC authentication"
 [dependencies]
 tokio = { version = "1", features = ["full"] }
 hyper = { version = "1", features = ["http1", "server"] }
-hyper-util = { version = "0.1", features = ["tokio", "http1", "server"] }
+hyper-util = { version = "0.1", features = ["tokio", "http1", "http2", "server", "client-legacy"] }
+tower-service = "0.3"
 http-body-util = "0.1"
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls", "stream", "http2"] }
 futures-util = "0.3"
@@ -26,6 +27,7 @@ hex = "0.4"
 anyhow = "1"
 toml = "0.8"
 tokio-rustls = "0.26"
+webpki-roots = "1"
 rustls = { version = "0.23", features = ["ring"] }
 rustls-pki-types = "1"
 rustls-pemfile = "2"

aether-proxy/src/app.rs

@@ -97,26 +97,7 @@ pub async fn run(mut config: Config) -> anyhow::Result<()> {
     let dynamic = Arc::new(RwLock::new(DynamicConfig::from_config(&config)));
 
     // Build delegate HTTP client (for proxy-initiated upstream requests).
-    // No overall timeout — SSE streams can last indefinitely.
+    let delegate_client = proxy::delegate_client::build_delegate_client(&config);
-    // Connect timeout limits connection establishment; Aether controls
-    // first-byte / idle timeouts on its own side.
-    let mut delegate_builder = reqwest::Client::builder()
-        .connect_timeout(Duration::from_secs(config.delegate_connect_timeout_secs))
-        .pool_max_idle_per_host(config.delegate_pool_max_idle_per_host)
-        .pool_idle_timeout(Duration::from_secs(config.delegate_pool_idle_timeout_secs))
-        .tcp_nodelay(config.delegate_tcp_nodelay);
-
-    if config.delegate_tcp_keepalive_secs > 0 {
-        delegate_builder = delegate_builder.tcp_keepalive(Some(Duration::from_secs(
-            config.delegate_tcp_keepalive_secs,
-        )));
-    } else {
-        delegate_builder = delegate_builder.tcp_keepalive(None);
-    }
-
-    let delegate_client = delegate_builder
-        .build()
-        .expect("failed to create delegate HTTP client");
 
     // Build shared application state
     let state = Arc::new(AppState {

aether-proxy/src/proxy/delegate.rs

@@ -1,18 +1,21 @@
 use std::collections::HashMap;
 use std::collections::HashSet;
+use std::error::Error as StdError;
 use std::sync::Arc;
 use std::time::Instant;
 
-use futures_util::{StreamExt, TryStreamExt};
+use futures_util::StreamExt;
 use http_body_util::{BodyExt, Full, Limited, StreamBody};
 use hyper::body::{Frame, Incoming};
-use hyper::{Request, Response};
+use hyper::header::{HeaderName, HeaderValue};
+use hyper::{Method, Request, Response, Uri};
 use tracing::{debug, warn};
 use url::Url;
 
 use super::BoxBody;
 use crate::auth;
 use crate::config::Config;
+use crate::proxy::delegate_client::{ConnectTiming, DelegateClient};
 use crate::proxy::target_filter::{self, DnsCache};
 
 /// Handle delegation requests: Aether sends a full request description,
@@ -35,7 +38,7 @@ pub async fn handle_delegate(
     allowed_ports: &HashSet<u16>,
     timestamp_tolerance: u64,
     dns_cache: &DnsCache,
-    http_client: &reqwest::Client,
+    http_client: &DelegateClient,
 ) -> Response<BoxBody> {
     let total_start = Instant::now();
 
@@ -150,7 +153,7 @@ pub async fn handle_delegate(
     debug!(method = %method_str, url = %target_url, is_gzip, "delegate request");
 
     // ── Build upstream request ──
-    let method = match method_str.parse::<reqwest::Method>() {
+    let method = match method_str.parse::<Method>() {
         Ok(m) => m,
         Err(e) => {
             warn!(error = %e, method = %method_str, "delegate invalid HTTP method");
@@ -158,34 +161,32 @@ pub async fn handle_delegate(
         }
     };
 
-    let mut upstream_req = http_client.request(method, &target_url);
+    let uri = match target_url.parse::<Uri>() {
-
+        Ok(u) => u,
-    // Set headers (skip `host` — reqwest sets it from the URL automatically,
+        Err(e) => {
-    // and a duplicate Host header can confuse certain upstreams)
+            warn!(error = %e, url = %target_url, "delegate invalid target URI");
-    for (name, value) in &upstream_headers {
+            return error_response(400, "bad_request", &format!("invalid URL: {}", e));
-        if name.eq_ignore_ascii_case("host") {
-            continue;
         }
-        upstream_req = upstream_req.header(name.as_str(), value.as_str());
+    };
-    }
 
     // ── Stream body passthrough ──
     // When body is gzip-compressed, forward it directly to upstream with
     // Content-Encoding: gzip header — no collect/decompress needed.
     // All major AI API providers (Anthropic, OpenAI, Google) accept gzip request bodies.
     let wire_size: u64;
+    let upstream_body: BoxBody;
     if is_gzip {
-        // Passthrough: stream the gzip body directly to upstream
+        let body_stream =
-        upstream_req = upstream_req.header("content-encoding", "gzip");
+            http_body_util::BodyStream::new(req.into_body()).filter_map(|result| async {
-        let body_stream = req.into_body();
+                match result {
-        let byte_stream = http_body_util::BodyStream::new(body_stream).filter_map(|result| async {
+                    Ok(frame) => frame.into_data().ok().map(|data| {
-            match result {
+                        Ok::<_, Box<dyn std::error::Error + Send + Sync>>(Frame::data(data))
-                Ok(frame) => frame.into_data().ok().map(Ok),
+                    }),
-                Err(e) => Some(Err(e)),
+                    Err(e) => Some(Err(Box::new(e) as Box<dyn std::error::Error + Send + Sync>)),
-            }
+                }
-        });
+            });
-        let reqwest_body = reqwest::Body::wrap_stream(byte_stream);
+        let stream_body = StreamBody::new(body_stream);
-        upstream_req = upstream_req.body(reqwest_body);
+        upstream_body = BodyExt::boxed(stream_body);
         // wire_size will be reported from Content-Length if available, otherwise 0
         wire_size = req_content_length;
     } else {
@@ -199,41 +200,85 @@ pub async fn handle_delegate(
             }
         };
         wire_size = body_bytes.len() as u64;
-        if !body_bytes.is_empty() {
+        let body = Full::new(body_bytes)
-            upstream_req = upstream_req.body(body_bytes.to_vec());
+            .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { match e {} })
+            .boxed();
+        upstream_body = body;
+    }
+
+    let mut upstream_req = Request::new(upstream_body);
+    *upstream_req.method_mut() = method;
+    *upstream_req.uri_mut() = uri;
+
+    {
+        let headers = upstream_req.headers_mut();
+        // Set headers (skip `host` — hyper sets it from the URI automatically,
+        // and a duplicate Host header can confuse certain upstreams)
+        for (name, value) in &upstream_headers {
+            if name.eq_ignore_ascii_case("host") {
+                continue;
+            }
+            let header_name = match HeaderName::from_bytes(name.as_bytes()) {
+                Ok(n) => n,
+                Err(_) => {
+                    warn!(header = %name, "delegate invalid header name");
+                    return error_response(400, "bad_request", "invalid header name");
+                }
+            };
+            let header_value = match HeaderValue::from_str(value) {
+                Ok(v) => v,
+                Err(_) => {
+                    warn!(header = %name, "delegate invalid header value");
+                    return error_response(400, "bad_request", "invalid header value");
+                }
+            };
+            headers.insert(header_name, header_value);
+        }
+        if is_gzip {
+            headers.insert(
+                hyper::header::CONTENT_ENCODING,
+                HeaderValue::from_static("gzip"),
+            );
         }
     }
 
     // ── Send upstream request ──
     // NOTE: We intentionally do NOT set a per-request timeout here.
-    // reqwest's `.timeout()` caps the *entire* request including body streaming,
+    // Connect timeout limits connection establishment; Aether controls
-    // which would truncate long-lived SSE streams. The delegate_client already
-    // has a configured connect_timeout for connection establishment, and Aether controls
     // first-byte / idle timeouts on its own side via asyncio.
     let upstream_start = Instant::now();
-    let upstream_resp = match upstream_req.send().await {
+    let upstream_resp = match http_client.request(upstream_req).await {
         Ok(resp) => resp,
         Err(e) => {
             warn!(url = %target_url, error = %e, "delegate upstream request failed");
-            let safe_detail = sanitize_upstream_error(&e.to_string());
+            let safe_detail = sanitize_upstream_error(&root_error_message(&e));
-            if e.is_timeout() {
+            if is_timeout_error(&e) {
                 return error_response(504, "upstream_timeout", &safe_detail);
             }
             return error_response(502, "upstream_connection_failed", &safe_detail);
         }
     };
-    let upstream_ms = upstream_start.elapsed().as_millis() as u64;
+    let ttfb_ms = upstream_start.elapsed().as_millis() as u64;
 
     // ── Build response ──
     let status = upstream_resp.status().as_u16();
     let resp_headers = upstream_resp.headers().clone();
+    let (connect_ms, tls_ms) = upstream_resp
+        .extensions()
+        .get::<ConnectTiming>()
+        .map(|t| (t.connect_ms, t.tls_ms))
+        .unwrap_or((0, 0));
+    let upstream_processing_ms = ttfb_ms.saturating_sub(connect_ms.saturating_add(tls_ms));
     let total_ms = total_start.elapsed().as_millis() as u64;
 
     debug!(
         url = %target_url,
         status,
         dns_ms,
-        upstream_ms,
+        connect_ms,
+        tls_ms,
+        ttfb_ms,
+        upstream_processing_ms,
         total_ms,
         wire_size,
         is_gzip,
@@ -246,16 +291,18 @@ pub async fn handle_delegate(
         "wire_size": wire_size,
         "passthrough": is_gzip,
         "dns_ms": dns_ms,
-        "upstream_ms": upstream_ms,
+        "connect_ms": connect_ms,
+        "tls_ms": tls_ms,
+        "ttfb_ms": ttfb_ms,
+        "upstream_ms": ttfb_ms,
+        "upstream_processing_ms": upstream_processing_ms,
         "total_ms": total_ms,
     });
 
-    let body_stream = upstream_resp
+    let stream_body: BoxBody = upstream_resp
-        .bytes_stream()
+        .into_body()
-        .map_ok(Frame::data)
+        .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { Box::new(e) })
-        .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { Box::new(e) });
+        .boxed();
-
-    let stream_body: BoxBody = BodyExt::boxed(StreamBody::new(body_stream));
 
     let mut builder = Response::builder().status(status);
     for (name, value) in resp_headers.iter() {
@@ -271,6 +318,30 @@ pub async fn handle_delegate(
     })
 }
 
+fn root_error_message(err: &dyn StdError) -> String {
+    let mut current = err;
+    while let Some(source) = current.source() {
+        current = source;
+    }
+    current.to_string()
+}
+
+fn is_timeout_error(err: &(dyn StdError + 'static)) -> bool {
+    if err.is::<tokio::time::error::Elapsed>() {
+        return true;
+    }
+    if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
+        if io_err.kind() == std::io::ErrorKind::TimedOut {
+            return true;
+        }
+    }
+    if let Some(source) = err.source() {
+        // source() returns &(dyn Error + 'static), so this is safe
+        return is_timeout_error(source);
+    }
+    false
+}
+
 // ── Sanitisation ─────────────────────────────────────────────────────────────
 
 /// Strip full URLs from error messages to prevent leaking upstream API keys,

aether-proxy/src/proxy/delegate_client.rs

@@ -0,0 +1,282 @@
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use std::time::{Duration, Instant};
+
+use hyper::rt;
+use hyper::Uri;
+use hyper_util::client::legacy::connect::{Connected, Connection, HttpConnector};
+use hyper_util::client::legacy::Client;
+use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
+use rustls::ClientConfig;
+use rustls_pki_types::ServerName;
+use tokio_rustls::TlsConnector;
+use tower_service::Service;
+
+use crate::config::Config;
+use crate::proxy::BoxBody;
+
+type BoxError = Box<dyn std::error::Error + Send + Sync>;
+
+type DelegateStream = MaybeHttpsStream<TokioIo<tokio::net::TcpStream>>;
+
+type DelegateConn = TimedConn<DelegateStream>;
+
+pub(crate) type DelegateClient = Client<InstrumentedConnector, BoxBody>;
+
+#[derive(Clone, Copy, Debug, Default)]
+pub(crate) struct ConnectTiming {
+    pub connect_ms: u64,
+    pub tls_ms: u64,
+}
+
+pub(crate) fn build_delegate_client(config: &Config) -> DelegateClient {
+    let mut http = HttpConnector::new();
+    http.enforce_http(false);
+    http.set_connect_timeout(Some(Duration::from_secs(
+        config.delegate_connect_timeout_secs,
+    )));
+    http.set_nodelay(config.delegate_tcp_nodelay);
+    if config.delegate_tcp_keepalive_secs > 0 {
+        http.set_keepalive(Some(Duration::from_secs(
+            config.delegate_tcp_keepalive_secs,
+        )));
+    } else {
+        http.set_keepalive(None);
+    }
+
+    let connector = InstrumentedConnector {
+        http,
+        tls_config: build_tls_config(),
+    };
+
+    let mut builder = Client::builder(TokioExecutor::new());
+    builder.pool_max_idle_per_host(config.delegate_pool_max_idle_per_host);
+    builder.pool_idle_timeout(Duration::from_secs(config.delegate_pool_idle_timeout_secs));
+    builder.pool_timer(TokioTimer::new());
+
+    builder.build::<_, BoxBody>(connector)
+}
+
+fn build_tls_config() -> Arc<ClientConfig> {
+    let root_store =
+        rustls::RootCertStore::from_iter(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
+    let mut config = ClientConfig::builder()
+        .with_root_certificates(root_store)
+        .with_no_client_auth();
+    config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
+    Arc::new(config)
+}
+
+#[derive(Clone)]
+pub(crate) struct InstrumentedConnector {
+    http: HttpConnector,
+    tls_config: Arc<ClientConfig>,
+}
+
+impl Service<Uri> for InstrumentedConnector {
+    type Response = DelegateConn;
+    type Error = BoxError;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, BoxError>> + Send>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.http.poll_ready(cx).map_err(Into::into)
+    }
+
+    fn call(&mut self, dst: Uri) -> Self::Future {
+        let scheme = dst.scheme_str().map(|s| s.to_ascii_lowercase());
+        let tls_config = self.tls_config.clone();
+        let connecting = self.http.call(dst.clone());
+        let connect_start = Instant::now();
+
+        Box::pin(async move {
+            match scheme.as_deref() {
+                Some("http") => {
+                    let tcp = connecting.await.map_err(|e| Box::new(e) as BoxError)?;
+                    let connect_ms = connect_start.elapsed().as_millis() as u64;
+                    Ok(TimedConn::new(
+                        MaybeHttpsStream::Http(tcp),
+                        ConnectTiming {
+                            connect_ms,
+                            tls_ms: 0,
+                        },
+                    ))
+                }
+                Some("https") => {
+                    let server_name = resolve_server_name(&dst)?;
+                    let tcp = connecting.await.map_err(|e| Box::new(e) as BoxError)?;
+                    let connect_ms = connect_start.elapsed().as_millis() as u64;
+
+                    let tls_start = Instant::now();
+                    let tls_stream = TlsConnector::from(tls_config)
+                        .connect(server_name, TokioIo::new(tcp))
+                        .await
+                        .map_err(std::io::Error::other)?;
+                    let tls_ms = tls_start.elapsed().as_millis() as u64;
+
+                    Ok(TimedConn::new(
+                        MaybeHttpsStream::Https(TokioIo::new(tls_stream)),
+                        ConnectTiming { connect_ms, tls_ms },
+                    ))
+                }
+                Some(other) => {
+                    Err(std::io::Error::other(format!("unsupported scheme {other}")).into())
+                }
+                None => Err(std::io::Error::other("missing scheme").into()),
+            }
+        })
+    }
+}
+
+fn resolve_server_name(uri: &Uri) -> Result<ServerName<'static>, BoxError> {
+    let host = uri.host().ok_or("missing host")?;
+    let host = host.trim_start_matches('[').trim_end_matches(']');
+    Ok(ServerName::try_from(host.to_string())?)
+}
+
+pub(crate) struct TimedConn<T> {
+    inner: T,
+    timing: ConnectTiming,
+}
+
+impl<T> TimedConn<T> {
+    fn new(inner: T, timing: ConnectTiming) -> Self {
+        Self { inner, timing }
+    }
+}
+
+impl<T: Connection> Connection for TimedConn<T> {
+    fn connected(&self) -> Connected {
+        self.inner.connected().extra(self.timing)
+    }
+}
+
+impl<T: rt::Read + Unpin> rt::Read for TimedConn<T> {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: rt::ReadBufCursor<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        Pin::new(&mut self.inner).poll_read(cx, buf)
+    }
+}
+
+impl<T: rt::Write + Unpin> rt::Write for TimedConn<T> {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        Pin::new(&mut self.inner).poll_write(cx, buf)
+    }
+
+    fn poll_flush(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        Pin::new(&mut self.inner).poll_flush(cx)
+    }
+
+    fn poll_shutdown(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        Pin::new(&mut self.inner).poll_shutdown(cx)
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        self.inner.is_write_vectored()
+    }
+
+    fn poll_write_vectored(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[std::io::IoSlice<'_>],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        Pin::new(&mut self.inner).poll_write_vectored(cx, bufs)
+    }
+}
+
+#[allow(clippy::large_enum_variant)]
+pub(crate) enum MaybeHttpsStream<T> {
+    Http(T),
+    Https(TokioIo<tokio_rustls::client::TlsStream<TokioIo<T>>>),
+}
+
+impl<T: rt::Read + rt::Write + Connection + Unpin> Connection for MaybeHttpsStream<T> {
+    fn connected(&self) -> Connected {
+        match self {
+            Self::Http(stream) => stream.connected(),
+            Self::Https(stream) => {
+                let (tcp, tls) = stream.inner().get_ref();
+                if tls.alpn_protocol() == Some(b"h2") {
+                    tcp.inner().connected().negotiated_h2()
+                } else {
+                    tcp.inner().connected()
+                }
+            }
+        }
+    }
+}
+
+impl<T: rt::Read + rt::Write + Unpin> rt::Read for MaybeHttpsStream<T> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: rt::ReadBufCursor<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        match Pin::get_mut(self) {
+            Self::Http(stream) => Pin::new(stream).poll_read(cx, buf),
+            Self::Https(stream) => Pin::new(stream).poll_read(cx, buf),
+        }
+    }
+}
+
+impl<T: rt::Write + rt::Read + Unpin> rt::Write for MaybeHttpsStream<T> {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        match Pin::get_mut(self) {
+            Self::Http(stream) => Pin::new(stream).poll_write(cx, buf),
+            Self::Https(stream) => Pin::new(stream).poll_write(cx, buf),
+        }
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
+        match Pin::get_mut(self) {
+            Self::Http(stream) => Pin::new(stream).poll_flush(cx),
+            Self::Https(stream) => Pin::new(stream).poll_flush(cx),
+        }
+    }
+
+    fn poll_shutdown(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        match Pin::get_mut(self) {
+            Self::Http(stream) => Pin::new(stream).poll_shutdown(cx),
+            Self::Https(stream) => Pin::new(stream).poll_shutdown(cx),
+        }
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        match self {
+            Self::Http(stream) => stream.is_write_vectored(),
+            Self::Https(stream) => stream.is_write_vectored(),
+        }
+    }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[std::io::IoSlice<'_>],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        match Pin::get_mut(self) {
+            Self::Http(stream) => Pin::new(stream).poll_write_vectored(cx, bufs),
+            Self::Https(stream) => Pin::new(stream).poll_write_vectored(cx, bufs),
+        }
+    }
+}

aether-proxy/src/proxy/mod.rs

@@ -1,5 +1,6 @@
 pub mod connect;
 pub mod delegate;
+pub mod delegate_client;
 pub mod server;
 pub mod target_filter;
 pub mod tls;

aether-proxy/src/setup/upgrade.rs

@@ -343,9 +343,15 @@ pub async fn cmd_upgrade(version: Option<String>) -> anyhow::Result<()> {
 
     // Restart systemd service if running
     if super::service::is_service_active() {
-        eprintln!("  Restarting systemd service...");
+        if super::service::is_root() {
-        super::service::run_cmd("systemctl", &["restart", "aether-proxy"])?;
+            eprintln!("  Restarting systemd service...");
-        eprintln!("  Service restarted.");
+            super::service::run_cmd("systemctl", &["restart", "aether-proxy"])?;
+            eprintln!("  Service restarted.");
+        } else {
+            eprintln!("  Systemd service is active, but restart requires root.");
+            eprintln!("  Run: sudo aether-proxy restart");
+            eprintln!("  Skipping restart.");
+        }
     } else {
         eprintln!("  No active systemd service detected, skipping restart.");
     }

aether-proxy/src/state.rs

@@ -12,6 +12,7 @@ use tokio_rustls::TlsAcceptor;
 
 use crate::config::Config;
 use crate::hardware::HardwareInfo;
+use crate::proxy::delegate_client::DelegateClient;
 use crate::proxy::target_filter::DnsCache;
 use crate::registration::client::AetherClient;
 use crate::runtime::SharedDynamicConfig;
@@ -26,8 +27,8 @@ pub struct AppState {
     pub public_ip: String,
     pub tls_fingerprint: Option<String>,
     pub tls_acceptor: Option<TlsAcceptor>,
-    /// Shared reqwest client for delegate mode (proxy issues upstream requests directly).
+    /// Shared delegate client for proxy-initiated upstream requests.
-    pub delegate_client: reqwest::Client,
+    pub delegate_client: DelegateClient,
     /// Active connection count for metrics reporting.
     pub active_connections: Arc<AtomicU64>,
     /// Connection concurrency limiter.

frontend/src/api/proxy-nodes.ts

@@ -96,4 +96,14 @@ export const proxyNodesApi = {
     const response = await apiClient.put<{ node_id: string; config_version: number; remote_config: ProxyNodeRemoteConfig; node: ProxyNode }>(`/api/admin/proxy-nodes/${nodeId}/config`, data)
     return response.data
   },
+
+  async testProxyUrl(data: { proxy_url: string; username?: string; password?: string }): Promise<ProxyNodeTestResult> {
+    const response = await apiClient.post<ProxyNodeTestResult>('/api/admin/proxy-nodes/test-url', data)
+    return response.data
+  },
+
+  async getHmacKey(): Promise<{ proxy_hmac_key: string }> {
+    const response = await apiClient.get<{ proxy_hmac_key: string }>('/api/admin/proxy-nodes/hmac-key')
+    return response.data
+  },
 }

frontend/src/features/usage/components/HorizontalRequestTimeline.vue

@@ -92,11 +92,30 @@
                 </div>
               </div>
 
+              <!-- 格式转换标记（节点下方） -->
+              <div
+                v-if="group.hasConversion"
+                class="conversion-indicator"
+              >
+                {{ group.primary.extra_data?.provider_api_format || '转换' }}
+              </div>
+
               <!-- 连接线 -->
               <div
                 v-if="groupIndex < groupedTimeline.length - 1"
-                class="node-line"
+                class="node-line-wrapper"
-              />
+              >
+                <div
+                  class="node-line"
+                  :class="{ 'conversion-boundary': groupIndex + 1 === conversionBoundaryIndex }"
+                />
+                <span
+                  v-if="groupIndex + 1 === conversionBoundaryIndex"
+                  class="boundary-label"
+                >
+                  格式转换
+                </span>
+              </div>
             </div>
           </div>
 
@@ -182,6 +201,19 @@
                     <span class="info-label">首字 (TTFB)</span>
                     <span class="info-value mono">{{ formatLatency(currentAttempt.extra_data.first_byte_time_ms) }}</span>
                   </div>
+                  <div
+                    v-if="currentAttempt.extra_data?.needs_conversion"
+                    class="info-item"
+                  >
+                    <span class="info-label">格式</span>
+                    <span class="info-value">
+                      <span class="conversion-badge">格式转换</span>
+                      <code
+                        v-if="currentAttempt.extra_data?.provider_api_format"
+                        class="ml-1.5 text-xs"
+                      >{{ currentAttempt.extra_data.provider_api_format }}</code>
+                    </span>
+                  </div>
                   <div
                     v-if="currentAttempt.key_name || currentAttempt.key_id"
                     class="info-item"
@@ -353,6 +385,7 @@ interface NodeGroup {
   totalLatency: number  // 所有尝试的总延迟
   startIndex: number
   endIndex: number
+  hasConversion: boolean  // 组内是否有格式转换候选
 }
 
 // 用量数据类型
@@ -478,8 +511,28 @@ const proxyTimingBreakdown = (proxy: Record<string, any>): string => {
     parts.push(label)
   }
 
-  parts.push(`DNS ${formatLatency(t.dns_ms)}`)
+  const ttfbMs = t.ttfb_ms ?? t.upstream_ms
-  parts.push(`上游 ${formatLatency(t.upstream_ms)}`)
+  const processingMs = t.upstream_processing_ms ?? (
+    ttfbMs != null && t.connect_ms != null && t.tls_ms != null
+      ? Math.max(0, ttfbMs - t.connect_ms - t.tls_ms)
+      : null
+  )
+
+  if (t.dns_ms != null && t.dns_ms > 0) {
+    parts.push(`DNS ${formatLatency(t.dns_ms)}`)
+  }
+  if (t.connect_ms != null && t.connect_ms > 0) {
+    parts.push(`连接 ${formatLatency(t.connect_ms)}`)
+  }
+  if (t.tls_ms != null && t.tls_ms > 0) {
+    parts.push(`TLS ${formatLatency(t.tls_ms)}`)
+  }
+  if (ttfbMs != null && ttfbMs > 0) {
+    parts.push(`TTFB ${formatLatency(ttfbMs)}`)
+  }
+  if (processingMs != null && processingMs > 0) {
+    parts.push(`上游处理 ${formatLatency(Math.round(processingMs))}`)
+  }
 
   // 计算 Aether→代理 之间无法解释的耗时差
   if (proxy.ttfb_ms != null && t.total_ms != null) {
@@ -535,6 +588,9 @@ const groupedTimeline = computed<NodeGroup[]>(() => {
       currentGroup.retryCount++
       currentGroup.endIndex = index
       currentGroup.totalLatency += candidate.latency_ms || 0
+      if (candidate.extra_data?.needs_conversion) {
+        currentGroup.hasConversion = true
+      }
       // 按优先级提升组状态：success > streaming/pending > failed/cancelled/stream_interrupted > skipped > available/unused
       const statusPriority: Record<string, number> = {
         available: 0, unused: 0, skipped: 1,
@@ -557,7 +613,8 @@ const groupedTimeline = computed<NodeGroup[]>(() => {
         retryCount: 0,
         totalLatency: candidate.latency_ms || 0,
         startIndex: index,
-        endIndex: index
+        endIndex: index,
+        hasConversion: candidate.extra_data?.needs_conversion === true,
       }
       groups.push(currentGroup)
     }
@@ -566,6 +623,16 @@ const groupedTimeline = computed<NodeGroup[]>(() => {
   return groups
 })
 
+// 格式转换分界点索引（首个 hasConversion=true 的 group index）
+const conversionBoundaryIndex = computed(() => {
+  const groups = groupedTimeline.value
+  if (!groups || groups.length === 0) return -1
+  const idx = groups.findIndex(g => g.hasConversion)
+  // 只有当分界点不在最开头时才有意义（前面有 exact 候选）
+  if (idx <= 0) return -1
+  return idx
+})
+
 // 计算链路总耗时（使用成功候选的 latency_ms 字段）
 // 优先使用 latency_ms，因为它与 Usage.response_time_ms 使用相同的时间基准
 // 避免 finished_at - started_at 带来的额外延迟（数据库操作时间）
@@ -1076,15 +1143,69 @@ const getStatusColorClass = (status: string) => {
 .node-dot.status-skipped { color: hsl(var(--primary)); }
 .node-dot.status-available { color: #d1d5db; }
 
-.node-line {
+/* 格式转换标记（节点下方） */
+.conversion-indicator {
+  position: absolute;
+  top: calc(100% + 6px);
+  left: 50%;
+  transform: translateX(-50%);
+  font-size: 0.55rem;
+  color: hsl(var(--muted-foreground) / 0.7);
+  white-space: nowrap;
+  max-width: 80px;
+  overflow: hidden;
+  text-overflow: ellipsis;
+  padding: 1px 4px;
+  border: 1px dashed hsl(var(--border));
+  border-radius: 3px;
+  background: hsl(var(--muted) / 0.3);
+}
+
+/* 连接线容器 */
+.node-line-wrapper {
   position: absolute;
   right: -64px;
   top: 50%;
   transform: translateY(-50%);
   width: 64px;
+  z-index: 1;
+  display: flex;
+  align-items: center;
+  justify-content: center;
+}
+
+.node-line {
+  width: 100%;
   height: 2px;
   background: hsl(var(--border));
-  z-index: 1;
+}
+
+/* 格式转换分界线 */
+.node-line.conversion-boundary {
+  background: none;
+  height: 0;
+  border-top: 2px dashed hsl(var(--muted-foreground) / 0.4);
+}
+
+.boundary-label {
+  position: absolute;
+  top: -14px;
+  font-size: 0.55rem;
+  color: hsl(var(--muted-foreground) / 0.6);
+  white-space: nowrap;
+}
+
+/* 详情面板中的格式转换标签 */
+.conversion-badge {
+  display: inline-flex;
+  align-items: center;
+  padding: 0.1rem 0.4rem;
+  font-size: 0.65rem;
+  font-weight: 500;
+  color: hsl(var(--muted-foreground));
+  background: hsl(var(--muted) / 0.5);
+  border: 1px dashed hsl(var(--border));
+  border-radius: 4px;
 }
 
 /* 详情面板 */

frontend/src/views/admin/ProxyNodes.vue

@@ -96,14 +96,24 @@
             </Select>
             <div class="h-4 w-px bg-border" />
             <Button
-              size="sm"
+              variant="ghost"
-              class="h-8 text-xs"
+              size="icon"
-              @click="showAddDialog = true"
+              class="h-8 w-8"
+              title="复制 HMAC Key"
+              @click="copyHmacKey"
             >
-              <Plus class="w-3.5 h-3.5 mr-1" />
+              <Copy class="w-3.5 h-3.5" />
-              手动添加
             </Button>
             <div class="h-4 w-px bg-border" />
+            <Button
+              variant="ghost"
+              size="icon"
+              class="h-8 w-8"
+              title="手动添加"
+              @click="showAddDialog = true"
+            >
+              <Plus class="w-3.5 h-3.5" />
+            </Button>
             <RefreshButton
               :loading="store.loading"
               @click="refresh"
@@ -429,18 +439,29 @@
       </form>
 
       <template #footer>
-        <Button
+        <div class="flex items-center justify-between w-full">
-          variant="outline"
+          <Button
-          @click="handleDialogClose(false)"
+            variant="outline"
-        >
+            :disabled="testingUrl || !addForm.proxy_url"
-          取消
+            @click="handleTestUrl"
-        </Button>
+          >
-        <Button
+            {{ testingUrl ? '测试中...' : '测试' }}
-          :disabled="addingNode || !addForm.name || !addForm.proxy_url"
+          </Button>
-          @click="editingNode ? handleUpdateManualNode() : handleAddManualNode()"
+          <div class="flex items-center gap-2">
-        >
+            <Button
-          {{ addingNode ? (editingNode ? '保存中...' : '添加中...') : (editingNode ? '保存' : '添加') }}
+              variant="outline"
-        </Button>
+              @click="handleDialogClose(false)"
+            >
+              取消
+            </Button>
+            <Button
+              :disabled="addingNode || !addForm.name || !addForm.proxy_url"
+              @click="editingNode ? handleUpdateManualNode() : handleAddManualNode()"
+            >
+              {{ addingNode ? (editingNode ? '保存中...' : '添加中...') : (editingNode ? '保存' : '添加') }}
+            </Button>
+          </div>
+        </div>
       </template>
     </Dialog>
 
@@ -539,6 +560,7 @@
 import { ref, computed, onMounted, watch } from 'vue'
 import { useProxyNodesStore } from '@/stores/proxy-nodes'
 import { useToast } from '@/composables/useToast'
+import { useClipboard } from '@/composables/useClipboard'
 import { useConfirm } from '@/composables/useConfirm'
 import { proxyNodesApi, type ProxyNode, type ProxyNodeRemoteConfig } from '@/api/proxy-nodes'
 
@@ -564,11 +586,12 @@ import {
   Dialog,
 } from '@/components/ui'
 
-import { Search, Trash2, Plus, SquarePen, Activity, Loader2, Settings } from 'lucide-vue-next'
+import { Search, Trash2, Plus, SquarePen, Activity, Loader2, Settings, Copy } from 'lucide-vue-next'
 import { formatRegion } from '@/utils/region'
 import HardwareTooltip from './components/HardwareTooltip.vue'
 
 const { success, error: toastError } = useToast()
+const { copyToClipboard } = useClipboard()
 const { confirmDanger } = useConfirm()
 const store = useProxyNodesStore()
 
@@ -602,6 +625,7 @@ const configForm = ref({
 
 // 测试连通性
 const testingNodes = ref(new Set<string>())
+const testingUrl = ref(false)
 
 const filteredNodes = computed(() => {
   let filtered = [...store.nodes]
@@ -638,6 +662,38 @@ async function refresh() {
   await store.fetchNodes()
 }
 
+async function handleTestUrl() {
+  if (!addForm.value.proxy_url || testingUrl.value) return
+  testingUrl.value = true
+  try {
+    const result = await proxyNodesApi.testProxyUrl({
+      proxy_url: addForm.value.proxy_url,
+      username: addForm.value.username || undefined,
+      password: addForm.value.password || undefined,
+    })
+    if (result.success) {
+      const parts = [`延迟: ${result.latency_ms}ms`]
+      if (result.exit_ip) parts.push(`出口IP: ${result.exit_ip}`)
+      success(`连通性测试通过，${parts.join('，')}`)
+    } else {
+      toastError(`连通性测试失败: ${result.error || '未知错误'}`)
+    }
+  } catch (err: any) {
+    toastError(err.response?.data?.error?.message || '测试请求失败')
+  } finally {
+    testingUrl.value = false
+  }
+}
+
+async function copyHmacKey() {
+  try {
+    const { proxy_hmac_key } = await proxyNodesApi.getHmacKey()
+    await copyToClipboard(proxy_hmac_key)
+  } catch (err: any) {
+    toastError(err.response?.data?.error?.message || err.response?.data?.detail || '获取 HMAC Key 失败')
+  }
+}
+
 function handleEdit(node: ProxyNode) {
   editingNode.value = node
   addForm.value = {

frontend/src/views/admin/components/HardwareTooltip.vue

@@ -1,9 +1,10 @@
 <script setup lang="ts">
 import type { ProxyNode } from '@/api/proxy-nodes'
 import {
-  Popover,
+  Tooltip,
-  PopoverContent,
+  TooltipContent,
-  PopoverTrigger,
+  TooltipProvider,
+  TooltipTrigger,
 } from '@/components/ui'
 import { Cpu } from 'lucide-vue-next'
 import { computed } from 'vue'
@@ -76,36 +77,37 @@ function formatNumber(n: number) {
 </script>
 
 <template>
-  <Popover v-if="showHardwareInfo">
+  <TooltipProvider v-if="showHardwareInfo">
-    <PopoverTrigger as-child>
+    <Tooltip>
-      <button
+      <TooltipTrigger as-child>
-        type="button"
+        <button
-        title="Hardware info"
+          type="button"
-        aria-label="Hardware info"
+          aria-label="硬件信息"
-        class="inline-flex items-center justify-center rounded-sm p-0.5 hover:bg-muted/60 transition-colors"
+          class="inline-flex items-center justify-center rounded-sm p-0.5 hover:bg-muted/60 transition-colors"
-      >
-        <Cpu class="h-3.5 w-3.5 text-muted-foreground" />
-      </button>
-    </PopoverTrigger>
-    <PopoverContent
-      side="right"
-      :side-offset="8"
-      class="w-auto p-3 text-xs space-y-1"
-    >
-      <div
-        v-if="hardwareRows.length === 0"
-        class="text-muted-foreground"
-      >
-        No hardware info reported.
-      </div>
-      <template v-else>
-        <div
-          v-for="row in hardwareRows"
-          :key="row.label"
         >
-          {{ row.label }}: {{ row.value }}
+          <Cpu class="h-3.5 w-3.5 text-muted-foreground" />
+        </button>
+      </TooltipTrigger>
+      <TooltipContent
+        side="right"
+        :side-offset="8"
+        class="w-auto px-3 py-2 text-xs space-y-1"
+      >
+        <div
+          v-if="hardwareRows.length === 0"
+          class="text-muted-foreground"
+        >
+          No hardware info reported.
         </div>
-      </template>
+        <template v-else>
-    </PopoverContent>
+          <div
-  </Popover>
+            v-for="row in hardwareRows"
+            :key="row.label"
+          >
+            {{ row.label }}: {{ row.value }}
+          </div>
+        </template>
+      </TooltipContent>
+    </Tooltip>
+  </TooltipProvider>
 </template>

src/api/admin/proxy_nodes/routes.py

@@ -216,6 +216,18 @@ async def test_proxy_node(node_id: str, request: Request, db: Session = Depends(
     return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
 
 
+@router.get("/hmac-key")
+async def get_proxy_hmac_key(request: Request, db: Session = Depends(get_db)) -> Any:
+    adapter = AdminGetProxyHmacKeyAdapter()
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
+@router.post("/test-url")
+async def test_proxy_url(request: Request, db: Session = Depends(get_db)) -> Any:
+    adapter = AdminTestProxyUrlAdapter()
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
 @router.put("/{node_id}/config")
 async def update_proxy_node_config(
     node_id: str, request: Request, db: Session = Depends(get_db)
@@ -485,3 +497,46 @@ class AdminUpdateProxyNodeConfigAdapter(AdminApiAdapter):
             "remote_config": node.remote_config,
             "node": node_to_dict(node),
         }
+
+
+@dataclass
+class AdminGetProxyHmacKeyAdapter(AdminApiAdapter):
+    """获取 proxy_hmac_key 供管理员复制到 aether-proxy 部署"""
+
+    name: str = "admin_get_proxy_hmac_key"
+
+    async def handle(self, context: ApiRequestContext) -> Any:
+        from src.config.settings import config
+
+        key = config.proxy_hmac_key
+        if not key:
+            raise InvalidRequestException(
+                "PROXY_HMAC_KEY 未配置（也未设置 ENCRYPTION_KEY 用于自动派生）"
+            )
+        return {"proxy_hmac_key": key}
+
+
+class TestProxyUrlRequest(BaseModel):
+    proxy_url: str = Field(..., min_length=1, max_length=500)
+    username: str | None = Field(None, max_length=255)
+    password: str | None = Field(None, max_length=500)
+
+
+@dataclass
+class AdminTestProxyUrlAdapter(AdminApiAdapter):
+    """通过 proxy_url 直接测试代理连通性（无需已注册节点）"""
+
+    name: str = "admin_test_proxy_url"
+
+    async def handle(self, context: ApiRequestContext) -> Any:
+        payload = context.ensure_json_body()
+        try:
+            req = TestProxyUrlRequest.model_validate(payload)
+        except ValidationError as exc:
+            raise InvalidRequestException("输入验证失败: " + _format_validation_error(exc))
+
+        return await ProxyNodeService.test_proxy_url(
+            proxy_url=req.proxy_url,
+            username=req.username,
+            password=req.password,
+        )

src/services/orchestration/candidate_resolver.py

@@ -205,7 +205,10 @@ class CandidateResolver:
                         "status": "skipped",
                         "skip_reason": candidate.skip_reason,
                         "is_cached": candidate.is_cached,
-                        "extra_data": {},
+                        "extra_data": {
+                            "needs_conversion": candidate.needs_conversion,
+                            "provider_api_format": candidate.provider_api_format or None,
+                        },
                         "required_capabilities": active_capabilities,
                         "created_at": datetime.now(timezone.utc),
                     }
@@ -235,7 +238,10 @@ class CandidateResolver:
                             "key_id": key.id,
                             "status": "available",
                             "is_cached": candidate.is_cached,
-                            "extra_data": {},
+                            "extra_data": {
+                                "needs_conversion": candidate.needs_conversion,
+                                "provider_api_format": candidate.provider_api_format or None,
+                            },
                             "required_capabilities": active_capabilities,
                             "created_at": datetime.now(timezone.utc),
                         }

src/services/proxy_node/service.py

@@ -91,6 +91,69 @@ def _sanitize_proxy_error(err: Exception) -> str:
     return re.sub(r"://[^@/]+@", "://***@", str(err))
 
 
+async def _test_proxy_connectivity(proxy_url: str) -> dict[str, Any]:
+    """通过代理 URL 测试连通性，返回标准化结果 dict"""
+    import time as _time
+
+    test_url = "https://1.1.1.1/cdn-cgi/trace"
+    start = _time.monotonic()
+    proxy_param = make_proxy_param(proxy_url)
+
+    try:
+        async with httpx.AsyncClient(
+            proxy=proxy_param,
+            timeout=httpx.Timeout(15.0, connect=10.0),
+        ) as client:
+            response = await client.get(test_url)
+            elapsed_ms = round((_time.monotonic() - start) * 1000, 1)
+
+            exit_ip = None
+            if response.status_code == 200:
+                for line in response.text.splitlines():
+                    if line.startswith("ip="):
+                        exit_ip = line.split("=", 1)[1].strip()
+                        break
+
+            return {
+                "success": True,
+                "latency_ms": elapsed_ms,
+                "exit_ip": exit_ip,
+                "error": None,
+            }
+    except httpx.ProxyError as exc:
+        elapsed_ms = round((_time.monotonic() - start) * 1000, 1)
+        return {
+            "success": False,
+            "latency_ms": elapsed_ms,
+            "exit_ip": None,
+            "error": f"代理连接失败: {_sanitize_proxy_error(exc)}",
+        }
+    except httpx.ConnectError as exc:
+        elapsed_ms = round((_time.monotonic() - start) * 1000, 1)
+        return {
+            "success": False,
+            "latency_ms": elapsed_ms,
+            "exit_ip": None,
+            "error": f"连接失败: {_sanitize_proxy_error(exc)}",
+        }
+    except httpx.TimeoutException:
+        elapsed_ms = round((_time.monotonic() - start) * 1000, 1)
+        return {
+            "success": False,
+            "latency_ms": elapsed_ms,
+            "exit_ip": None,
+            "error": "连接超时（15秒）",
+        }
+    except Exception as exc:
+        elapsed_ms = round((_time.monotonic() - start) * 1000, 1)
+        return {
+            "success": False,
+            "latency_ms": elapsed_ms,
+            "exit_ip": None,
+            "error": _sanitize_proxy_error(exc),
+        }
+
+
 def _build_test_proxy_url(node: ProxyNode) -> str:
     """为测试连通性构建代理 URL（无需节点在线）"""
     if node.is_manual:
@@ -377,76 +440,25 @@ class ProxyNodeService:
     @staticmethod
     async def test_node(db: Session, *, node_id: str) -> dict[str, Any]:
         """测试代理节点连通性和延迟"""
-        import time as _time
-
         node = db.query(ProxyNode).filter(ProxyNode.id == node_id).first()
         if not node:
             raise NotFoundException(f"ProxyNode {node_id} 不存在", "proxy_node")
 
-        # 构建代理 URL
         try:
             proxy_url = _build_test_proxy_url(node)
         except Exception as exc:
             return {"success": False, "latency_ms": None, "exit_ip": None, "error": str(exc)}
 
-        test_url = "https://1.1.1.1/cdn-cgi/trace"
+        return await _test_proxy_connectivity(proxy_url)
-        start = _time.monotonic()
 
-        proxy_param = make_proxy_param(proxy_url)
+    @staticmethod
-
+    async def test_proxy_url(
-        try:
+        *, proxy_url: str, username: str | None = None, password: str | None = None
-            async with httpx.AsyncClient(
+    ) -> dict[str, Any]:
-                proxy=proxy_param,
+        """直接通过 proxy_url 测试代理连通性（无需已注册节点）"""
-                timeout=httpx.Timeout(15.0, connect=10.0),
+        if username:
-            ) as client:
+            proxy_url = inject_auth_into_proxy_url(proxy_url, username, password)
-                response = await client.get(test_url)
+        return await _test_proxy_connectivity(proxy_url)
-                elapsed_ms = round((_time.monotonic() - start) * 1000, 1)
-
-                exit_ip = None
-                if response.status_code == 200:
-                    for line in response.text.splitlines():
-                        if line.startswith("ip="):
-                            exit_ip = line.split("=", 1)[1].strip()
-                            break
-
-                return {
-                    "success": True,
-                    "latency_ms": elapsed_ms,
-                    "exit_ip": exit_ip,
-                    "error": None,
-                }
-        except httpx.ProxyError as exc:
-            elapsed_ms = round((_time.monotonic() - start) * 1000, 1)
-            return {
-                "success": False,
-                "latency_ms": elapsed_ms,
-                "exit_ip": None,
-                "error": f"代理连接失败: {_sanitize_proxy_error(exc)}",
-            }
-        except httpx.ConnectError as exc:
-            elapsed_ms = round((_time.monotonic() - start) * 1000, 1)
-            return {
-                "success": False,
-                "latency_ms": elapsed_ms,
-                "exit_ip": None,
-                "error": f"连接失败: {_sanitize_proxy_error(exc)}",
-            }
-        except httpx.TimeoutException:
-            elapsed_ms = round((_time.monotonic() - start) * 1000, 1)
-            return {
-                "success": False,
-                "latency_ms": elapsed_ms,
-                "exit_ip": None,
-                "error": "连接超时（15秒）",
-            }
-        except Exception as exc:
-            elapsed_ms = round((_time.monotonic() - start) * 1000, 1)
-            return {
-                "success": False,
-                "latency_ms": elapsed_ms,
-                "exit_ip": None,
-                "error": _sanitize_proxy_error(exc),
-            }
 
     @staticmethod
     def update_node_config(

src/services/system/stats_aggregator.py

@@ -12,6 +12,7 @@ from datetime import date, datetime, time, timedelta, timezone
 from typing import Any
 
 from sqlalchemy import Float, and_, case, cast, func
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from src.core.logger import logger
@@ -879,15 +880,23 @@ class StatsAggregatorService:
     @staticmethod
     def aggregate_hourly_stats_bundle(db: Session, hour_utc: datetime) -> StatsHourly:
         """聚合单小时所有统计（原子提交）"""
-        stats = StatsAggregatorService.aggregate_hourly_stats(db, hour_utc, commit=False)
-        StatsAggregatorService.aggregate_hourly_user_stats(db, hour_utc, commit=False)
-        StatsAggregatorService.aggregate_hourly_model_stats(db, hour_utc, commit=False)
-        StatsAggregatorService.aggregate_hourly_provider_stats(db, hour_utc, commit=False)
 
-        stats.is_complete = True
+        def _do_aggregate() -> StatsHourly:
-        stats.aggregated_at = datetime.now(timezone.utc)
+            stats = StatsAggregatorService.aggregate_hourly_stats(db, hour_utc, commit=False)
-        db.commit()
+            StatsAggregatorService.aggregate_hourly_user_stats(db, hour_utc, commit=False)
-        return stats
+            StatsAggregatorService.aggregate_hourly_model_stats(db, hour_utc, commit=False)
+            StatsAggregatorService.aggregate_hourly_provider_stats(db, hour_utc, commit=False)
+            stats.is_complete = True
+            stats.aggregated_at = datetime.now(timezone.utc)
+            db.commit()
+            return stats
+
+        try:
+            return _do_aggregate()
+        except IntegrityError:
+            db.rollback()
+            logger.warning("小时统计聚合冲突，重试更新: {}", hour_utc)
+            return _do_aggregate()
 
     @staticmethod
     def update_summary(db: Session) -> StatsSummary: