i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 01:10:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(provider): 修复 Windsurf Connect 请求与端点计数

Browse Source
This commit is contained in:
Entropy.Xu
2026-05-18 21:23:02 +08:00
parent 931c577345
commit 82637ad882
6 changed files with 301 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
apps/aether-gateway/src/execution_runtime/transport.rs
View File

@@ -1470,11 +1470,16 @@ pub(crate) fn decode_response_body_bytes(
}
pub(crate) fn response_body_is_json(headers: &BTreeMap<String, String>, body_bytes: &[u8]) -> bool {
if headers
let content_type = headers
.get("content-type")
.map(|value| value.to_ascii_lowercase())
.is_some_and(|value| value.contains("json"))
.unwrap_or_default();
if content_type.contains("application/connect+json")
|| content_type.contains("application/connect+proto")
{
return false;
}
if content_type.contains("json") {
return true;
}
@@ -1508,7 +1513,7 @@ mod tests {
build_browser_wreq_client, build_client, build_request_headers, execute_sync_plan,
record_manual_proxy_request_failure, record_manual_proxy_request_outcome,
record_manual_proxy_request_success, record_manual_proxy_stream_error,
resolve_execution_transport_controls, DirectSyncExecutionRuntime,
resolve_execution_transport_controls, response_body_is_json, DirectSyncExecutionRuntime,
ExecutionRuntimeTransportError, ExecutionTransportControls,
};
use crate::constants::{
@@ -2764,6 +2769,17 @@ mod tests {
));
}
#[test]
fn connect_json_response_is_not_treated_as_plain_json() {
let headers = BTreeMap::from([(
"content-type".to_string(),
"application/connect+json".to_string(),
)]);
let body = [2, 0, 0, 0, 2, b'{', b'}'];
assert!(!response_body_is_json(&headers, &body));
}
#[tokio::test]
async fn direct_sync_execution_runtime_compresses_json_body_when_requested() {
let listener = crate::test_support::bind_loopback_listener()

2
apps/aether-gateway/src/handlers/admin/provider/endpoints_admin/reads.rs
View File

@@ -1,6 +1,6 @@
use crate::handlers::admin::request::AdminAppState;
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogProvider,
StoredProviderCatalogEndpoint, StoredProviderCatalogKey, StoredProviderCatalogProvider,
};
use std::time::{SystemTime, UNIX_EPOCH};

68
apps/aether-gateway/src/handlers/admin/provider/query/models/model_test.rs
View File

@@ -14,7 +14,7 @@ use super::{provider_query_key_display_name, provider_query_provider_payload};
use crate::ai_serving::{
maybe_build_sync_finalize_outcome, GatewayControlDecision,
ANTIGRAVITY_V1INTERNAL_ENVELOPE_NAME, GEMINI_CHAT_SYNC_FINALIZE_REPORT_KIND,
OPENAI_IMAGE_SYNC_FINALIZE_REPORT_KIND,
OPENAI_CHAT_SYNC_FINALIZE_REPORT_KIND, OPENAI_IMAGE_SYNC_FINALIZE_REPORT_KIND,
};
use crate::clock::current_unix_ms;
use crate::execution_runtime;
@@ -1786,6 +1786,56 @@ async fn provider_query_execute_kiro_test_candidate(
})
}
async fn provider_query_finalize_windsurf_result(
route_path: &str,
trace_id: &str,
requested_model: &str,
mapped_model: &str,
original_request_body: &Value,
result: &aether_contracts::ExecutionResult,
) -> Result<Option<Value>, GatewayError> {
let decision = GatewayControlDecision::synthetic(
route_path,
Some("admin_proxy".to_string()),
Some("provider_query_manage".to_string()),
Some("test_model_failover".to_string()),
Some("openai:chat".to_string()),
);
let payload = GatewaySyncReportRequest {
trace_id: trace_id.to_string(),
report_kind: OPENAI_CHAT_SYNC_FINALIZE_REPORT_KIND.to_string(),
report_context: Some(json!({
"client_api_format": "openai:chat",
"provider_api_format": "openai:chat",
"model": requested_model,
"mapped_model": mapped_model,
"needs_conversion": false,
"has_envelope": true,
"envelope_name": crate::provider_transport::windsurf::WINDSURF_ENVELOPE_NAME,
"original_request_body": original_request_body,
})),
status_code: result.status_code,
headers: result.headers.clone(),
body_json: result.body.as_ref().and_then(|body| body.json_body.clone()),
client_body_json: None,
body_base64: result
.body
.as_ref()
.and_then(|body| body.body_bytes_b64.clone()),
telemetry: result.telemetry.clone(),
};
let Some(outcome) = maybe_build_sync_finalize_outcome(trace_id, &decision, &payload)? else {
return Ok(None);
};
let bytes = to_bytes(outcome.response.into_body(), usize::MAX)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?;
serde_json::from_slice::<Value>(&bytes)
.map(Some)
.map_err(|err| GatewayError::Internal(err.to_string()))
}
fn provider_query_build_openai_image_test_request_body_for_route(
payload: &Value,
model: &str,
@@ -3194,7 +3244,7 @@ async fn provider_query_execute_windsurf_test_candidate(
method: "POST".to_string(),
url: request_url.clone(),
headers: request_headers.clone(),
content_type: Some("application/json".to_string()),
content_type: Some("application/connect+json".to_string()),
content_encoding: None,
body: RequestBody::from_json(provider_request_body.clone()),
stream: upstream_is_stream,
@@ -3211,7 +3261,19 @@ async fn provider_query_execute_windsurf_test_candidate(
let result = state
.execute_execution_runtime_sync_plan(Some(trace_id), &plan)
.await?;
let response_body = result.body.as_ref().and_then(|body| body.json_body.clone());
let response_body = if result.status_code < 400 {
provider_query_finalize_windsurf_result(
route_path,
trace_id,
request_model,
request_model,
&request_body,
&result,
)
.await?
} else {
result.body.as_ref().and_then(|body| body.json_body.clone())
};
let missing_success_body = result.status_code < 400 && response_body.is_none();
let did_fail = result.status_code >= 400 || missing_success_body;
let error_message = if did_fail {

72
apps/aether-gateway/src/tests/control/admin/endpoints/routes.rs
View File

@@ -347,6 +347,78 @@ async fn gateway_counts_keys_with_null_api_formats_for_each_fixed_provider_endpo
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_counts_inherited_windsurf_key_formats_for_admin_provider_endpoints() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/providers/provider-windsurf/endpoints",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let mut key = sample_key(
"key-windsurf-a",
"provider-windsurf",
"openai:chat",
"oauth-secret",
);
key.auth_type = "oauth".to_string();
key.api_formats = None;
let mut provider = sample_provider("provider-windsurf", "windsurf", 10);
provider.provider_type = "windsurf".to_string();
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
vec![sample_endpoint(
"endpoint-windsurf-chat",
"provider-windsurf",
"openai:chat",
"https://server.codeium.com",
)],
vec![key],
));
let (_, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
)),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-windsurf/endpoints?skip=0&limit=50"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
let items = payload.as_array().expect("payload should be an array");
assert_eq!(items.len(), 1);
assert_eq!(items[0]["id"], "endpoint-windsurf-chat");
assert_eq!(items[0]["total_keys"], 1);
assert_eq!(items[0]["active_keys"], 1);
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_returns_service_unavailable_for_admin_provider_endpoint_create_when_catalog_writer_unavailable(
) {

133
crates/aether-ai-formats/src/provider_compat/private_envelope.rs
View File

@@ -182,6 +182,96 @@ fn transform_provider_private_stream_line_with_event_state(
Ok(out)
}
const CONNECT_FRAME_HEADER_BYTES: usize = 5;
const MAX_CONNECT_JSON_FRAME_BYTES: usize = 16 * 1024 * 1024;
fn report_context_is_windsurf_envelope(report_context: &Value) -> bool {
report_context
.get("envelope_name")
.and_then(Value::as_str)
.is_some_and(|value| value.eq_ignore_ascii_case(WINDSURF_ENVELOPE_NAME))
}
fn buffer_looks_like_connect_frame(buffer: &[u8]) -> bool {
let Some(flags) = buffer.first().copied() else {
return false;
};
if flags & !0x03 != 0 {
return false;
}
if buffer.len() < CONNECT_FRAME_HEADER_BYTES {
return true;
}
let len = u32::from_be_bytes([buffer[1], buffer[2], buffer[3], buffer[4]]) as usize;
len <= MAX_CONNECT_JSON_FRAME_BYTES
}
fn drain_windsurf_connect_json_frames(
buffer: &mut Vec<u8>,
) -> Result<Vec<u8>, AiSurfaceFinalizeError> {
let mut output = Vec::new();
while buffer.len() >= CONNECT_FRAME_HEADER_BYTES {
let flags = buffer[0];
if flags & !0x03 != 0 {
return Err(AiSurfaceFinalizeError::new(format!(
"invalid Connect frame flags: {flags}"
)));
}
let len = u32::from_be_bytes([buffer[1], buffer[2], buffer[3], buffer[4]]) as usize;
if len > MAX_CONNECT_JSON_FRAME_BYTES {
return Err(AiSurfaceFinalizeError::new(format!(
"Connect frame size {len} exceeds {MAX_CONNECT_JSON_FRAME_BYTES}"
)));
}
if buffer.len() < CONNECT_FRAME_HEADER_BYTES + len {
break;
}
let payload = buffer[CONNECT_FRAME_HEADER_BYTES..CONNECT_FRAME_HEADER_BYTES + len].to_vec();
buffer.drain(..CONNECT_FRAME_HEADER_BYTES + len);
if flags & 0x01 != 0 {
return Err(AiSurfaceFinalizeError::new(
"compressed Connect JSON frames are not supported for Windsurf chat",
));
}
if payload.is_empty() {
continue;
}
let body: Value = serde_json::from_slice(&payload)?;
if flags & 0x02 != 0 {
if let Some(error) = body.get("error") {
output.extend_from_slice(b"event: error\n");
output.extend_from_slice(b"data: ");
output.extend(serde_json::to_vec(error)?);
output.extend_from_slice(b"\n\n");
}
continue;
}
if looks_like_windsurf_error(&body) {
output.extend_from_slice(b"event: error\n");
output.extend_from_slice(b"data: ");
output.extend(serde_json::to_vec(&body)?);
output.extend_from_slice(b"\n\n");
continue;
}
let unwrapped = normalize_windsurf_stream_event_value(&body).unwrap_or(body);
let mut line = b"data: ".to_vec();
line.extend(serde_json::to_vec(&unwrapped)?);
line.extend_from_slice(b"\n\n");
output.extend(line);
}
if !buffer.is_empty()
&& buffer.len() < CONNECT_FRAME_HEADER_BYTES
&& !buffer_looks_like_connect_frame(buffer)
{
return Err(AiSurfaceFinalizeError::new(
"invalid partial Connect JSON frame",
));
}
Ok(output)
}
enum ProviderPrivateStreamNormalizeMode {
EnvelopeUnwrap,
KiroToClaudeCli(Box<KiroToClaudeCliStreamState>),
@@ -243,6 +333,11 @@ impl ProviderPrivateStreamNormalizer<'_> {
}
ProviderPrivateStreamNormalizeMode::EnvelopeUnwrap => {
self.buffered.extend_from_slice(chunk);
if report_context_is_windsurf_envelope(self.report_context)
&& buffer_looks_like_connect_frame(&self.buffered)
{
return drain_windsurf_connect_json_frames(&mut self.buffered);
}
let mut output = Vec::new();
while let Some(line_end) = self.buffered.iter().position(|byte| *byte == b'\n') {
let line = self.buffered.drain(..=line_end).collect::<Vec<_>>();
@@ -269,6 +364,11 @@ impl ProviderPrivateStreamNormalizer<'_> {
if self.buffered.is_empty() {
return Ok(Vec::new());
}
if report_context_is_windsurf_envelope(self.report_context)
&& buffer_looks_like_connect_frame(&self.buffered)
{
return drain_windsurf_connect_json_frames(&mut self.buffered);
}
let line = std::mem::take(&mut self.buffered);
transform_provider_private_stream_line_with_event_state(
self.report_context,
@@ -728,6 +828,39 @@ mod tests {
assert!(text.contains(r#""content":"chunk""#));
}
#[test]
fn unwraps_windsurf_connect_json_stream_frames() {
let report_context = json!({
"has_envelope": true,
"envelope_name": "windsurf:GetChatMessage",
"provider_api_format": "openai:chat",
});
let mut normalizer = maybe_build_provider_private_stream_normalizer(Some(&report_context))
.expect("normalizer should exist");
let mut framed = connect_json_frame(
0,
br#"{"responseId":"ws-3","response":{"text":"frame chunk"}}"#,
);
framed.extend(connect_json_frame(2, b"{}"));
let mut output = normalizer
.push_chunk(&framed)
.expect("connect frame should normalize");
output.extend(normalizer.finish().expect("finish should succeed"));
let text = String::from_utf8(output).expect("utf8");
assert!(text.contains(r#""object":"chat.completion.chunk""#));
assert!(text.contains(r#""content":"frame chunk""#));
}
fn connect_json_frame(flags: u8, payload: &[u8]) -> Vec<u8> {
let mut out = Vec::with_capacity(5 + payload.len());
out.push(flags);
out.extend_from_slice(&(payload.len() as u32).to_be_bytes());
out.extend_from_slice(payload);
out
}
#[test]
fn private_stream_normalizer_unwraps_antigravity_stream() {
let report_context = json!({

22
crates/aether-provider-transport/src/windsurf.rs
View File

@@ -170,7 +170,7 @@ pub fn build_windsurf_cascade_headers(
header_rules: Option<&Value>,
auth_header: &str,
auth_value: &str,
upstream_is_stream: bool,
_upstream_is_stream: bool,
) -> Option<BTreeMap<String, String>> {
let mut out = BTreeMap::new();
for (name, value) in headers {
@@ -203,20 +203,16 @@ pub fn build_windsurf_cascade_headers(
return None;
}
out.insert("content-type".to_string(), "application/json".to_string());
out.insert(
"content-type".to_string(),
"application/connect+json".to_string(),
);
out.insert("connect-protocol-version".to_string(), "1".to_string());
out.insert(
"user-agent".to_string(),
format!("windsurf/{DEFAULT_IDE_VERSION}"),
);
out.insert(
"accept".to_string(),
if upstream_is_stream {
"text/event-stream".to_string()
} else {
"application/json".to_string()
},
);
out.insert("accept".to_string(), "application/connect+json".to_string());
if !auth_header.is_empty() {
out.insert(auth_header, auth_value.trim().to_string());
}
@@ -441,9 +437,13 @@ mod tests {
headers.get("authorization").map(String::as_str),
Some("Bearer secret")
);
assert_eq!(
headers.get("content-type").map(String::as_str),
Some("application/connect+json")
);
assert_eq!(
headers.get("accept").map(String::as_str),
Some("application/json")
Some("application/connect+json")
);
}