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

fix(ai): dedupe OpenAI responses text snapshot deltas

Browse Source
This commit is contained in:
zhefox
2026-05-27 16:11:02 +08:00
parent e09d3199c1
commit 8abedecb16
3 changed files with 245 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aether-ai-formats/src/formats/openai/chat/stream.rs
+91 -15
View File
@@ -428,7 +428,7 @@ impl OpenAIResponsesProviderState {
) {
let missing = if text.starts_with(&self.text) {
text[self.text.len()..].to_string()
} else if self.text == text {
} else if self.text == text || self.text.starts_with(text) {
String::new()
} else {
text.to_string()
@@ -829,27 +829,24 @@ impl OpenAIResponsesProviderState {
"response.created" | "response.in_progress" => {
self.ensure_started(report_context, &mut out);
}
"response.output_text.delta" | "response.outtext.delta" => {
let piece = match value.get("delta") {
Some(Value::String(text)) => text.clone(),
Some(Value::Object(delta)) => delta
.get("text")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string(),
_ => String::new(),
};
if !piece.is_empty() {
"response.output_text.delta" | "response.outtext.delta" => match value.get("delta") {
Some(Value::String(piece)) if !piece.is_empty() => {
self.ensure_started(report_context, &mut out);
self.text.push_str(&piece);
self.text.push_str(piece);
let (id, model) = self.identity(report_context);
out.push(CanonicalStreamFrame {
id,
model,
event: CanonicalStreamEvent::TextDelta(piece),
event: CanonicalStreamEvent::TextDelta(piece.clone()),
});
}
}
Some(Value::Object(delta)) => {
if let Some(text) = delta.get("text").and_then(Value::as_str) {
self.emit_missing_text(report_context, &mut out, text);
}
}
_ => {}
},
"response.content_part.added" | "response.content_part.done" => {
if let Some(part) = value.get("part").and_then(Value::as_object) {
if part.get("type").and_then(Value::as_str) == Some("output_text") {
@@ -3158,6 +3155,85 @@ mod tests {
)));
}
#[test]
fn openai_responses_provider_state_does_not_duplicate_text_snapshot_deltas() {
let mut state = OpenAIResponsesProviderState::default();
let report_context = json!({});
let mut frames = Vec::new();
for event in [
json!({
"type": "response.output_text.delta",
"response_id": "resp_snapshot_delta",
"output_index": 0,
"content_index": 0,
"delta": {
"text": "Hello",
}
}),
json!({
"type": "response.output_text.delta",
"response_id": "resp_snapshot_delta",
"output_index": 0,
"content_index": 0,
"delta": {
"text": "Hello world",
}
}),
json!({
"type": "response.output_text.delta",
"response_id": "resp_snapshot_delta",
"output_index": 0,
"content_index": 0,
"delta": {
"text": "Hello",
}
}),
json!({
"type": "response.output_text.done",
"response_id": "resp_snapshot_delta",
"output_index": 0,
"content_index": 0,
"text": "Hello world",
}),
json!({
"type": "response.completed",
"response": {
"id": "resp_snapshot_delta",
"object": "response",
"model": "gpt-5.4",
"status": "completed",
"output": [{
"type": "message",
"id": "msg_snapshot_delta",
"role": "assistant",
"status": "completed",
"content": [{
"type": "output_text",
"text": "Hello world",
"annotations": [],
}]
}],
}
}),
] {
frames.extend(
state
.push_line(&report_context, data_line(event))
.expect("responses text event should parse"),
);
}
let text = frames
.iter()
.filter_map(|frame| match &frame.event {
CanonicalStreamEvent::TextDelta(text) => Some(text.as_str()),
_ => None,
})
.collect::<String>();
assert_eq!(text, "Hello world");
}
#[test]
fn openai_responses_provider_state_delays_arguments_until_tool_name_is_known() {
let mut state = OpenAIResponsesProviderState::default();
crates/aether-ai-formats/src/formats/shared/stream_core/format_matrix.rs
+77
View File
@@ -573,6 +573,83 @@ mod tests {
}
}
#[test]
fn transforms_openai_responses_text_snapshot_deltas_to_openai_chat_without_duplicates() {
let report_context = report_context("openai:responses", "openai:chat");
let mut matrix = StreamingStandardFormatMatrix::default();
let mut output = Vec::new();
for line in [
data_line(json!({
"type": "response.output_text.delta",
"response_id": "resp_snapshot_delta",
"output_index": 0,
"content_index": 0,
"delta": {
"text": "Hello",
}
})),
data_line(json!({
"type": "response.output_text.delta",
"response_id": "resp_snapshot_delta",
"output_index": 0,
"content_index": 0,
"delta": {
"text": "Hello world",
}
})),
data_line(json!({
"type": "response.output_text.done",
"response_id": "resp_snapshot_delta",
"output_index": 0,
"content_index": 0,
"text": "Hello world",
})),
data_line(json!({
"type": "response.completed",
"response": {
"id": "resp_snapshot_delta",
"object": "response",
"model": "gpt-5.4",
"status": "completed",
"output": [{
"type": "message",
"id": "msg_snapshot_delta",
"role": "assistant",
"status": "completed",
"content": [{
"type": "output_text",
"text": "Hello world",
"annotations": [],
}]
}],
}
})),
] {
output.extend(
matrix
.transform_line(&report_context, line)
.expect("responses stream line should convert"),
);
}
let sse = String::from_utf8(output).expect("sse should be utf8");
let content = sse
.lines()
.filter_map(|line| line.strip_prefix("data: "))
.filter_map(|payload| serde_json::from_str::<Value>(payload).ok())
.filter_map(|value| {
value
.pointer("/choices/0/delta/content")
.and_then(Value::as_str)
.map(ToOwned::to_owned)
})
.collect::<String>();
assert_eq!(content, "Hello world");
assert!(!sse.contains("HelloHello"));
}
#[test]
fn transforms_provider_errors_to_claude_error_events() {
let cases = [
crates/aether-ai-formats/src/formats/shared/sync_products.rs
+77 -15
View File
@@ -1720,22 +1720,25 @@ pub fn aggregate_openai_responses_stream_sync_response(body: &[u8]) -> Option<Va
"response.output_text.delta" | "response.outtext.delta" => {
let output_index = openai_responses_event_output_index(event_object).unwrap_or(0);
let content_index = openai_responses_event_content_index(event_object);
let delta = match event_object.get("delta") {
Some(Value::String(text)) => text.as_str(),
Some(Value::Object(delta)) => delta
.get("text")
.and_then(Value::as_str)
.unwrap_or_default(),
_ => "",
};
if delta.is_empty() {
continue;
match event_object.get("delta") {
Some(Value::String(delta)) => {
append_openai_responses_message_text_delta(
message_states.entry(output_index).or_default(),
content_index,
delta,
);
}
Some(Value::Object(delta)) => {
if let Some(text) = delta.get("text").and_then(Value::as_str) {
merge_openai_responses_message_text_delta_object(
message_states.entry(output_index).or_default(),
content_index,
text,
);
}
}
_ => {}
}
append_openai_responses_message_text_delta(
message_states.entry(output_index).or_default(),
content_index,
delta,
);
}
"response.output_text.done" => {
let output_index = openai_responses_event_output_index(event_object).unwrap_or(0);
@@ -2109,6 +2112,46 @@ fn append_openai_responses_message_text_delta(
.or_insert_with(|| Value::Array(Vec::new()));
}
fn merge_openai_responses_message_text_delta_object(
state: &mut OpenAIResponsesSyncMessageState,
content_index: usize,
text: &str,
) {
if text.is_empty() {
return;
}
let part = state
.parts
.entry(content_index)
.or_insert_with(default_openai_responses_output_text_part);
let Some(part) = part.as_object_mut() else {
return;
};
if !part
.get("type")
.and_then(Value::as_str)
.is_some_and(|value| matches!(value, "output_text" | "text"))
{
return;
}
let current = part
.get("text")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
let merged = if text.starts_with(current.as_str()) {
text.to_string()
} else if current == text || current.starts_with(text) {
current
} else {
format!("{current}{text}")
};
part.insert("type".to_string(), Value::String("output_text".to_string()));
part.insert("text".to_string(), Value::String(merged));
part.entry("annotations".to_string())
.or_insert_with(|| Value::Array(Vec::new()));
}
fn merge_openai_responses_message_text_part(
state: &mut OpenAIResponsesSyncMessageState,
content_index: usize,
@@ -3755,6 +3798,25 @@ mod tests {
assert_eq!(result["output"][0]["content"][1]["text"], " world");
}
#[test]
fn aggregates_openai_responses_text_snapshot_deltas_without_duplicates() {
let body = concat!(
"event: response.output_text.delta\n",
"data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"content_index\":0,\"delta\":{\"text\":\"Hello\"}}\n\n",
"event: response.output_text.delta\n",
"data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"content_index\":0,\"delta\":{\"text\":\"Hello world\"}}\n\n",
"event: response.output_text.done\n",
"data: {\"type\":\"response.output_text.done\",\"output_index\":0,\"content_index\":0,\"text\":\"Hello world\"}\n\n",
"event: response.completed\n",
"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_snapshot_123\",\"object\":\"response\",\"model\":\"gpt-5\",\"status\":\"completed\",\"output\":[]}}\n\n",
);
let result = aggregate_openai_responses_stream_sync_response(body.as_bytes())
.expect("openai-responses stream should aggregate into a sync body");
assert_eq!(result["output"][0]["content"][0]["text"], "Hello world");
}
#[test]
fn preserves_openai_responses_non_text_content_parts() {
let body = concat!(