Anthropic 路由改回原生优先、桥接兜底

- llm：/api/llm/anthropic/* 先按 Anthropic 原样直通 Router 的 /v1/messages（20 秒首包上限），拿到 2xx 就直接透传
- llm：直通返回非 2xx、连接失败或首包超时才回退到 Anthropic→Chat 协议桥接，并记录 model/status 便于定位
- 测试：两条 Anthropic 路由用例改为「原生先被拒 + 桥接命中」的两段式 mock

server-rs/crates/api-server/src/llm/mod.rs

@@ -30,6 +30,14 @@ use crate::{
 
 pub(crate) const LLM_REQUEST_MAX_BODY_BYTES: usize = 32 * 1024 * 1024;
 
+/// 原生 Anthropic 直通的首包等待上限。
+///
+/// 有些部署（按账号分组）只有 Anthropic 原生渠道，这时直通 `/v1/messages` 是正确路径；
+/// 另一些部署只有 OpenAI 形状的渠道，`/v1/messages` 会快速失败或长时间不响应。这里给直通
+/// 一个有限等待窗口：拿到 2xx 就用原生，否则回退到协议桥接。
+const LLM_ANTHROPIC_NATIVE_FIRST_BYTE_TIMEOUT: std::time::Duration =
+    std::time::Duration::from_secs(20);
+
 mod anthropic_bridge;
 
 pub(crate) mod icon_specs;
@@ -415,11 +423,11 @@ pub async fn proxy_llm_responses(
             );
         }
     }
-    let upstream_headers = upstream.headers().clone();
     let is_stream = payload
         .get("stream")
         .and_then(Value::as_bool)
         .unwrap_or(false);
+    let upstream_headers = upstream.headers().clone();
     if !is_stream {
         let body = upstream.bytes().await.map_err(|error| {
             llm_error_response(
@@ -487,7 +495,7 @@ pub async fn proxy_llm_messages(
     State(state): State<AppState>,
     Extension(request_context): Extension<RequestContext>,
     Extension(authenticated): Extension<AuthenticatedAccessToken>,
-    _headers: HeaderMap,
+    headers: HeaderMap,
     body: Bytes,
 ) -> Result<Response, Response> {
     if body.len() > LLM_REQUEST_MAX_BODY_BYTES {
@@ -558,8 +566,8 @@ pub async fn proxy_llm_messages(
             )
         })?
         .to_string();
-    // 这里不再改写 payload 的 model：出站请求由 Anthropic→Chat 转换器按解析后的模型名重建。
-    object.remove("model");
+    // 原生直通要用目录解析后的上游模型名；桥接路径由转换器显式接收同一个名字。
+    object.insert("model".to_string(), Value::String(selected_model.clone()));
 
     let (base_url, api_key, key_id) =
         resolve_llm_router_credentials(&state, authenticated.claims().user_id())
@@ -587,9 +595,122 @@ pub async fn proxy_llm_messages(
                     .with_message(format!("创建 LLM Router 请求客户端失败：{error}")),
             )
         })?;
-    // 账号 Router 的 `/v1/messages` 在现网不可用（最小请求也回 `not implemented`），
-    // 所以这里把 Anthropic Messages 转成 Router 可用的 Chat Completions，再把回程的
-    // SSE 翻译回 Anthropic 事件流。Claude Code 只认 Anthropic 形状，转换发生在网关内。
+    // 先试原生 Anthropic 直通：账号所在分组有 Anthropic 渠道时这是最短、最保真的路径。
+    // 直通失败（非 2xx / 首包超时）再回退到协议桥接——现网有些分组只有 OpenAI 形状的
+    // 渠道，`/v1/messages` 在那里会直接报 `not implemented`。
+    let is_stream = payload
+        .get("stream")
+        .and_then(Value::as_bool)
+        .unwrap_or(false);
+    let native_body = payload.to_string();
+    let mut native_request = client
+        .post(router_protocol_url(&base_url, "messages"))
+        .header("x-api-key", api_key.clone())
+        .bearer_auth(api_key.clone())
+        .header("content-type", "application/json");
+    for name in ["anthropic-version", "anthropic-beta", "accept"] {
+        if let Some(value) = headers.get(name) {
+            native_request = native_request.header(name, value);
+        }
+    }
+    let native = tokio::time::timeout(
+        LLM_ANTHROPIC_NATIVE_FIRST_BYTE_TIMEOUT,
+        native_request.body(native_body).send(),
+    )
+    .await;
+    let native_response = match native {
+        Ok(Ok(response)) if response.status().is_success() => Some(response),
+        Ok(Ok(response)) => {
+            tracing::warn!(
+                request_id = request_context.request_id(),
+                user_id = %authenticated.claims().user_id(),
+                model = %selected_model,
+                status = response.status().as_u16(),
+                "LLM Router 原生 Anthropic 直通失败，回退协议桥接"
+            );
+            None
+        }
+        Ok(Err(error)) => {
+            tracing::warn!(
+                request_id = request_context.request_id(),
+                user_id = %authenticated.claims().user_id(),
+                model = %selected_model,
+                error = %error,
+                "LLM Router 原生 Anthropic 请求失败，回退协议桥接"
+            );
+            None
+        }
+        Err(_) => {
+            tracing::warn!(
+                request_id = request_context.request_id(),
+                user_id = %authenticated.claims().user_id(),
+                model = %selected_model,
+                "LLM Router 原生 Anthropic 首包超时，回退协议桥接"
+            );
+            None
+        }
+    };
+    if let Some(upstream) = native_response {
+        let status = upstream.status();
+        let upstream_headers = upstream.headers().clone();
+        if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) {
+            if let Err(error) = crate::external_api_keys::revoke_llm_router_account(
+                &state,
+                authenticated.claims().user_id(),
+                &key_id,
+            )
+            .await
+            {
+                tracing::warn!(
+                    request_id = request_context.request_id(),
+                    user_id = %authenticated.claims().user_id(),
+                    key_id = %key_id,
+                    error = %error,
+                    "LLM Router 返回确定鉴权失败，但本地账号 Key 失效标记未完成"
+                );
+            }
+        }
+        if !is_stream {
+            let body = upstream.bytes().await.map_err(|error| {
+                llm_error_response(
+                    &request_context,
+                    AppError::from_status(StatusCode::BAD_GATEWAY)
+                        .with_message(format!("读取 LLM Router 响应失败：{error}")),
+                )
+            })?;
+            if let Err(error) =
+                settle_llm_router_usage(&state, authenticated.claims().user_id()).await
+            {
+                tracing::error!(
+                    request_id = request_context.request_id(),
+                    user_id = %authenticated.claims().user_id(),
+                    error = %error,
+                    "LLM Router Messages 成功但累计额度同步未完成"
+                );
+            }
+            return build_upstream_response(
+                status,
+                &upstream_headers,
+                Body::from(body),
+                &request_context,
+            );
+        }
+        let stream = stream_messages_with_billing(
+            upstream.bytes_stream(),
+            state.clone(),
+            authenticated.claims().user_id().to_string(),
+            request_context.request_id().to_string(),
+        );
+        return build_upstream_response(
+            status,
+            &upstream_headers,
+            Body::from_stream(stream),
+            &request_context,
+        );
+    }
+
+    // 回退路径：把 Anthropic Messages 转成 Router 可用的 Chat Completions，再把回程的
+    // SSE 翻译回 Anthropic 事件流（Claude Code 只认 Anthropic 形状）。
     let (converted, tool_names) =
         anthropic_bridge::anthropic_messages_to_chat_completions(&payload, &selected_model)
             .map_err(|message| {
@@ -1081,6 +1202,77 @@ fn stream_responses_with_billing(
 ///
 /// The bytes are forwarded verbatim so the Claude Agent SDK keeps parsing the
 /// same event shapes it received before the platform gateway was introduced.
+/// 原生 Anthropic 直通的流式结算：`message_stop` 是唯一结算点。
+fn stream_messages_with_billing(
+    mut upstream: impl futures_util::Stream<Item = Result<Bytes, reqwest::Error>> + Unpin,
+    state: AppState,
+    owner_user_id: String,
+    request_id: String,
+) -> impl futures_util::Stream<Item = Result<Bytes, std::io::Error>> {
+    async_stream::stream! {
+        let mut pending = String::new();
+        let mut utf8_pending = Vec::new();
+        let mut billing_result: Option<Result<(), AppError>> = None;
+        while let Some(chunk) = upstream.next().await {
+            match chunk {
+                Ok(bytes) => {
+                    append_utf8_chunk(&mut pending, &mut utf8_pending, bytes.as_ref());
+                    pending = pending.replace("\r\n", "\n");
+                    while let Some(separator) = pending.find("\n\n") {
+                        let raw_event = pending[..separator + 2].to_string();
+                        let event = pending[..separator].to_string();
+                        pending.drain(..separator + 2);
+                        if is_messages_terminal_sse_event(&event) && billing_result.is_none() {
+                            billing_result = Some(
+                                settle_llm_router_usage(&state, owner_user_id.as_str()).await,
+                            );
+                            if let Some(Err(error)) = billing_result.as_ref() {
+                                tracing::error!(
+                                    request_id = %request_id,
+                                    user_id = %owner_user_id,
+                                    error = %error,
+                                    "LLM Router 流式响应已完成但累计额度同步未完成"
+                                );
+                            }
+                        }
+                        yield Ok(Bytes::from(raw_event));
+                    }
+                }
+                Err(error) => {
+                    if billing_result.is_none()
+                        && let Err(billing_error) =
+                            settle_llm_router_usage(&state, owner_user_id.as_str()).await
+                    {
+                        tracing::error!(
+                            request_id = %request_id,
+                            user_id = %owner_user_id,
+                            error = %billing_error,
+                            "LLM Router 流式响应中断且累计额度同步未完成"
+                        );
+                    }
+                    yield Err(std::io::Error::other(format!(
+                        "LLM Router 响应流读取失败：{error}"
+                    )));
+                    return;
+                }
+            }
+        }
+        if !pending.is_empty() {
+            yield Ok(Bytes::from(pending));
+        }
+    }
+}
+
+/// Anthropic Messages 以 `message_stop` 结束，没有 `[DONE]` 哨兵。
+fn is_messages_terminal_sse_event(event: &str) -> bool {
+    event.lines().any(|line| {
+        let line = line.trim();
+        line == "event: message_stop"
+            || line.contains("\"type\":\"message_stop\"")
+            || line.contains("\"type\": \"message_stop\"")
+    })
+}
+
 /// Router 的 Chat Completions SSE → Anthropic Messages SSE（Claude Code 只认后者）。
 ///
 /// 结算点放在 `message_stop` 之后或流结束：Anthropic 的终止事件由翻译器产出，
@@ -2084,7 +2276,7 @@ mod tests {
 
     #[tokio::test]
     async fn llm_anthropic_messages_uses_account_router_credential() {
-        let (server_url, captured) = spawn_capturing_mock_server(MockResponse {
+        let (server_url, captured) = spawn_native_rejecting_capturing_mock_server(MockResponse {
             status_line: "200 OK",
             content_type: "application/json; charset=utf-8",
             body: r#"{"id":"chatcmpl_api_server_01","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"pong"},"finish_reason":"stop"}],"usage":{"prompt_tokens":5,"completion_tokens":2}}"#.to_string(),
@@ -2170,7 +2362,7 @@ mod tests {
             "data: {\"id\":\"chatcmpl_stream_01\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":5,\"completion_tokens\":2}}\n\n",
             "data: [DONE]\n\n",
         );
-        let (server_url, _captured) = spawn_capturing_mock_server(MockResponse {
+        let (server_url, _captured) = spawn_native_rejecting_capturing_mock_server(MockResponse {
             status_line: "200 OK",
             content_type: "text/event-stream; charset=utf-8",
             body: streamed.to_string(),
@@ -2259,6 +2451,36 @@ mod tests {
         (format!("http://{address}"), captured)
     }
 
+    /// 原生 Anthropic 直通先被拒（500），再由桥接路径命中；捕获的是第二条（桥接）请求。
+    fn spawn_native_rejecting_capturing_mock_server(
+        response: MockResponse,
+    ) -> (String, Arc<Mutex<Option<String>>>) {
+        let listener = TcpListener::bind("127.0.0.1:0").expect("listener should bind");
+        let address = listener.local_addr().expect("listener should have addr");
+        let captured = Arc::new(Mutex::new(None));
+        let captured_for_thread = Arc::clone(&captured);
+
+        thread::spawn(move || {
+            let (mut native_stream, _) = listener.accept().expect("native request should connect");
+            let _ = read_request(&mut native_stream);
+            write_response(
+                &mut native_stream,
+                MockResponse {
+                    status_line: "500 Internal Server Error",
+                    content_type: "application/json; charset=utf-8",
+                    body: r#"{"error":"native anthropic route unavailable"}"#.to_string(),
+                    extra_headers: Vec::new(),
+                },
+            );
+            let (mut stream, _) = listener.accept().expect("request should connect");
+            let request = read_request(&mut stream);
+            *captured_for_thread.lock().expect("captured request lock") = Some(request);
+            write_response(&mut stream, response);
+        });
+
+        (format!("http://{address}"), captured)
+    }
+
     fn read_request(stream: &mut std::net::TcpStream) -> String {
         stream
             .set_read_timeout(Some(StdDuration::from_secs(1)))