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

fix(windsurf): 修复 native 工具流式回程

Browse Source
This commit is contained in:
Entropy.Xu
2026-05-21 02:55:05 +08:00
parent 4112a8b2ea
commit 6b04a0a3a6
2 changed files with 82 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/aether-gateway/src/execution_runtime/windsurf.rs
+55 -8
View File
@@ -573,6 +573,10 @@ fn build_windsurf_stream_frame_stream(
.filter(|tool_call| !streamed_native_call_ids.contains(&tool_call.id))
.collect::<Vec<_>>();
tool_calls.extend(parsed_tool_calls.tool_calls);
let finish_reason = windsurf_stream_finish_reason(
!streamed_native_call_ids.is_empty(),
!tool_calls.is_empty(),
);
if !tool_calls.is_empty() {
for frame in sse_tool_call_frames_from_index(
&prepared.request_id,
@@ -595,9 +599,10 @@ fn build_windsurf_stream_frame_stream(
&content,
)));
}
let _ = tx.send(encode_stream_frame_ndjson(&sse_finish_frame(
let _ = tx.send(encode_stream_frame_ndjson(&sse_finish_frame_with_reason(
&prepared.request_id,
&prepared.model,
finish_reason,
)));
}
let _ = tx.send(encode_stream_frame_ndjson(&raw_sse_data_frame(b"data: [DONE]\n\n")));
@@ -616,7 +621,7 @@ fn build_windsurf_stream_frame_stream(
windsurf_terminal_summary(
poll_result.usage,
Some(prepared.model.as_str()),
Some(if tool_calls.is_empty() { "stop" } else { "tool_calls" }),
Some(finish_reason),
),
)));
}
@@ -1092,8 +1097,7 @@ where
step.response_text.as_str()
};
let previous = yielded_by_step.get(&index).copied().unwrap_or_default();
if live_text.len() > previous {
let delta = live_text[previous..].to_string();
if let Some(delta) = windsurf_text_delta_from_cursor(live_text, previous) {
yielded_by_step.insert(index, live_text.len());
grew = true;
on_delta(delta)?;
@@ -1104,8 +1108,7 @@ where
&& step.modified_text.starts_with(live_text)
{
let cursor = yielded_by_step.get(&index).copied().unwrap_or_default();
if step.modified_text.len() > cursor {
let delta = step.modified_text[cursor..].to_string();
if let Some(delta) = windsurf_text_delta_from_cursor(&step.modified_text, cursor) {
yielded_by_step.insert(index, step.modified_text.len());
grew = true;
on_delta(delta)?;
@@ -1115,6 +1118,18 @@ where
Ok(grew)
}
fn windsurf_text_delta_from_cursor(text: &str, cursor: usize) -> Option<String> {
if text.len() <= cursor {
return None;
}
let cursor = if text.is_char_boundary(cursor) {
cursor
} else {
0
};
Some(text[cursor..].to_string())
}
async fn ensure_windsurf_language_server(
plan: &ExecutionPlan,
) -> Result<LsHandle, ExecutionRuntimeTransportError> {
@@ -3111,8 +3126,15 @@ fn sse_data_frame(request_id: &str, model: &str, delta: &str) -> StreamFrame {
raw_sse_data_frame(&body)
}
fn sse_finish_frame(request_id: &str, model: &str) -> StreamFrame {
sse_finish_frame_with_reason(request_id, model, "stop")
fn windsurf_stream_finish_reason(
streamed_native_tool_call: bool,
pending_tool_call: bool,
) -> &'static str {
if streamed_native_tool_call || pending_tool_call {
"tool_calls"
} else {
"stop"
}
}
fn sse_finish_frame_with_reason(request_id: &str, model: &str, finish_reason: &str) -> StreamFrame {
@@ -4429,6 +4451,19 @@ mod tests {
assert!(text.contains(r#""arguments":"{\"query\":\"today tech\"}""#));
}
#[test]
fn native_tool_stream_finish_reason_stays_tool_calls_after_live_delta() {
assert_eq!(
super::windsurf_stream_finish_reason(true, false),
"tool_calls"
);
assert_eq!(
super::windsurf_stream_finish_reason(false, true),
"tool_calls"
);
assert_eq!(super::windsurf_stream_finish_reason(false, false), "stop");
}
#[test]
fn windsurf_rate_limit_errors_classify_as_upstream_429() {
let err = ExecutionRuntimeTransportError::UpstreamRequest(
@@ -4893,6 +4928,18 @@ mod tests {
assert_eq!(yielded_by_step.get(&0), Some(&12));
}
#[test]
fn text_delta_cursor_resets_on_non_char_boundary() {
assert_eq!(
super::windsurf_text_delta_from_cursor("aé", 2).as_deref(),
Some("aé")
);
assert_eq!(
super::windsurf_text_delta_from_cursor("hello", 2).as_deref(),
Some("llo")
);
}
#[test]
fn windsurf_warmup_treats_http_404_as_non_transport_error() {
let err = ExecutionRuntimeTransportError::UpstreamRequest(
crates/aether-usage-runtime/src/write.rs
+27 -1
View File
@@ -2157,6 +2157,11 @@ fn decode_body_for_storage(body_base64: Option<&str>) -> Option<Value> {
let bytes = base64::engine::general_purpose::STANDARD
.decode(body_base64)
.ok()?;
if let Some(error_body) =
aether_ai_formats::api::extract_provider_private_stream_error_body(None, &bytes)
{
return Some(error_body);
}
if let Ok(json_body) = serde_json::from_slice::<Value>(&bytes) {
return Some(json_body);
}
@@ -3014,7 +3019,7 @@ mod tests {
build_stream_terminal_usage_event, build_streaming_usage_record,
build_sync_terminal_usage_event, build_sync_terminal_usage_payload_seed,
build_sync_terminal_usage_seed, build_terminal_usage_context_seed,
build_terminal_usage_event_from_seed, build_usage_event_data_seed,
build_terminal_usage_event_from_seed, build_usage_event_data_seed, decode_body_for_storage,
extract_token_counts_from_json, extract_token_counts_from_value, headers_to_json,
mask_header_value, mask_sensitive_body_fields, mask_sensitive_headers_in_json_value,
parse_sse_body_for_storage, resolve_error_message, trim_owned_non_empty_string,
@@ -5403,6 +5408,27 @@ mod tests {
);
}
#[test]
fn decode_body_for_storage_extracts_connect_json_error_frames() {
let payload = br#"{"error":{"code":"resource_exhausted","message":"quota exhausted"}}"#;
let mut framed = Vec::new();
framed.push(2);
framed.extend_from_slice(&(payload.len() as u32).to_be_bytes());
framed.extend_from_slice(payload);
let body_base64 = base64::engine::general_purpose::STANDARD.encode(framed);
assert_eq!(
decode_body_for_storage(Some(body_base64.as_str())),
Some(json!({
"error": {
"code": "resource_exhausted",
"message": "quota exhausted",
"type": "resource_exhausted"
}
}))
);
}
#[test]
fn parse_sse_body_for_storage_handles_crlf_and_cr_line_endings() {
let sse_body = concat!(