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

fix(provider): normalize OpenAI Responses modern fields and stream events

Browse Source
This commit is contained in:
zhefox
2026-06-04 15:45:37 +08:00
parent 69b8b96fb8
commit fd27f55fe5
7 changed files with 774 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aether-ai-formats/src/formats/conversion/request.rs
+28
View File
@@ -814,6 +814,34 @@ mod tests {
}
}
#[test]
fn openai_responses_request_normalizer_strips_content_cache_control() {
let body = json!({
"model": "gpt-5.1",
"input": [{
"type": "message",
"role": "user",
"content": [{
"type": "input_text",
"text": "stable project brief",
"cache_control": {"type": "ephemeral"}
}]
}],
"prompt_cache_key": "cache_123"
});
let converted = registry::convert_request(
"openai:responses",
"openai:responses",
&body,
&FormatContext::default(),
)
.expect("responses request");
assert_eq!(converted["prompt_cache_key"], "cache_123");
assert!(!converted["input"].to_string().contains("cache_control"));
}
#[test]
fn claude_output_config_effort_controls_responses_reasoning() {
let body = json!({
crates/aether-ai-formats/src/formats/conversion/response.rs
+30 -3
View File
@@ -6,7 +6,10 @@
use serde_json::{json, Value};
use crate::formats::{context::FormatContext, registry};
use crate::formats::{
context::FormatContext,
openai::responses::response::ensure_modern_openai_responses_response_fields, registry,
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct OpenAiResponsesResponseUsage {
@@ -218,7 +221,7 @@ pub fn build_openai_responses_response_with_content(
}));
}
output.extend(function_calls);
json!({
let mut response = json!({
"id": response_id,
"object": "response",
"status": "completed",
@@ -229,7 +232,11 @@ pub fn build_openai_responses_response_with_content(
"output_tokens": usage.output_tokens,
"total_tokens": usage.total_tokens,
}
})
});
if let Some(response_object) = response.as_object_mut() {
ensure_modern_openai_responses_response_fields(response_object);
}
response
}
fn response_context(report_context: &Value) -> FormatContext {
@@ -273,6 +280,26 @@ mod tests {
assert_eq!(converted["object"], "response");
assert_eq!(converted["output"][0]["type"], "message");
assert_eq!(converted["output_text"], "hello");
assert!(converted["created_at"].as_i64().is_some());
assert!(converted["completed_at"].as_i64().is_some());
}
#[test]
fn manual_responses_response_builder_emits_modern_fields() {
let response = super::build_openai_responses_response(
"resp_manual_123",
"gpt-5",
"Hello manual",
Vec::new(),
1,
2,
3,
);
assert_eq!(response["output_text"], "Hello manual");
assert!(response["created_at"].as_i64().is_some());
assert!(response["completed_at"].as_i64().is_some());
}
#[test]
crates/aether-ai-formats/src/formats/openai/chat/stream.rs
+112 -7
View File
@@ -2,6 +2,9 @@ use std::collections::{BTreeMap, BTreeSet};
use serde_json::{json, Map, Value};
use crate::formats::openai::responses::response::{
ensure_modern_openai_responses_response_fields, openai_responses_current_timestamp,
};
use crate::formats::shared::response::build_generated_tool_call_id;
use crate::formats::shared::sse::{encode_done_sse, encode_json_sse};
use crate::formats::shared::stream_core::common::*;
@@ -958,7 +961,34 @@ impl OpenAIResponsesProviderState {
self.emit_missing_text(report_context, &mut out, key, text);
}
}
"response.reasoning_summary_text.delta" => {
"response.refusal.delta" => {
let piece = value
.get("delta")
.and_then(Value::as_str)
.unwrap_or_default();
if !piece.is_empty() {
let key = Self::text_part_key_from_event(&value);
self.emit_text_delta(report_context, &mut out, key, piece);
}
}
"response.refusal.done" => {
let refusal = value
.get("refusal")
.and_then(Value::as_str)
.or_else(|| {
value
.get("part")
.and_then(Value::as_object)
.and_then(|part| part.get("refusal"))
.and_then(Value::as_str)
})
.unwrap_or_default();
if !refusal.is_empty() {
let key = Self::text_part_key_from_event(&value);
self.emit_missing_text(report_context, &mut out, key, refusal);
}
}
"response.reasoning_text.delta" | "response.reasoning_summary_text.delta" => {
let piece = value
.get("delta")
.and_then(Value::as_str)
@@ -983,7 +1013,7 @@ impl OpenAIResponsesProviderState {
});
}
}
"response.reasoning_summary_text.done" => {
"response.reasoning_text.done" | "response.reasoning_summary_text.done" => {
let text = value
.get("text")
.and_then(Value::as_str)
@@ -1240,7 +1270,7 @@ impl OpenAIResponsesProviderState {
}
out.push(self.unknown_frame(report_context, payload));
}
"response.completed" => {
"response.completed" | "response.done" => {
let Some(response) = value.get("response").and_then(Value::as_object) else {
return Ok(out);
};
@@ -1399,6 +1429,7 @@ fn web_search_query_from_arguments(arguments: &str) -> String {
pub struct OpenAIResponsesClientEmitter {
response_id: Option<String>,
model: Option<String>,
created_at: Option<i64>,
message_item_id: Option<String>,
reasoning_item_id: Option<String>,
started: bool,
@@ -1744,13 +1775,28 @@ impl OpenAIResponsesClientEmitter {
}
fn in_progress_response(&self) -> Value {
json!({
let mut response = json!({
"id": self.response_id(),
"object": "response",
"model": self.model(),
"status": "in_progress",
"output": [],
})
});
if let (Some(created_at), Some(response_object)) =
(self.created_at, response.as_object_mut())
{
response_object.insert("created_at".to_string(), Value::from(created_at));
}
response
}
fn ensure_created_at(&mut self) -> i64 {
if let Some(created_at) = self.created_at {
return created_at;
}
let created_at = openai_responses_current_timestamp();
self.created_at = Some(created_at);
created_at
}
fn allocate_output_index(&mut self) -> usize {
@@ -1787,6 +1833,7 @@ impl OpenAIResponsesClientEmitter {
if self.started {
return Ok(Vec::new());
}
self.ensure_created_at();
self.started = true;
let mut out = self.encode_response_event(
"response.created",
@@ -2118,6 +2165,7 @@ impl OpenAIResponsesClientEmitter {
"output_index": output_index,
"item_id": item_id.clone(),
"call_id": item_id.clone(),
"name": name,
"arguments": state.arguments.as_str(),
}),
)?);
@@ -2336,7 +2384,7 @@ impl OpenAIResponsesClientEmitter {
}
ordered_output.sort_by_key(|(output_index, _)| *output_index);
json!({
let mut response = json!({
"id": self.response_id(),
"object": "response",
"status": "completed",
@@ -2346,7 +2394,14 @@ impl OpenAIResponsesClientEmitter {
.map(|(_, item)| item)
.collect::<Vec<_>>(),
"usage": openai_responses_usage_from_usage(&usage),
})
});
if let Some(response_object) = response.as_object_mut() {
if let Some(created_at) = self.created_at {
response_object.insert("created_at".to_string(), Value::from(created_at));
}
ensure_modern_openai_responses_response_fields(response_object);
}
response
}
pub fn emit(&mut self, frame: CanonicalStreamFrame) -> Result<Vec<u8>, AiSurfaceFinalizeError> {
@@ -3043,6 +3098,9 @@ mod tests {
assert!(sse.contains("\"response_id\":\"resp_stream_123\""));
assert!(sse.contains("\"item_id\":\"resp_stream_123_msg\""));
assert!(sse.contains("\"text\":\"Hello\""));
assert!(sse.contains("\"output_text\":\"Hello\""));
assert!(sse.contains("\"created_at\":"));
assert!(sse.contains("\"completed_at\":"));
assert_eq!(response_sequence_numbers(&sse), (1..=9).collect::<Vec<_>>());
}
@@ -3226,6 +3284,53 @@ mod tests {
)));
}
#[test]
fn openai_responses_provider_state_accepts_refusal_events() {
let mut state = OpenAIResponsesProviderState::default();
let report_context = json!({});
let mut frames = Vec::new();
frames.extend(
state
.push_line(
&report_context,
data_line(json!({
"type": "response.refusal.delta",
"response_id": "resp_refusal_123",
"output_index": 0,
"item_id": "msg_refusal_123",
"content_index": 0,
"delta": "I can't",
})),
)
.expect("refusal delta should parse"),
);
frames.extend(
state
.push_line(
&report_context,
data_line(json!({
"type": "response.refusal.done",
"response_id": "resp_refusal_123",
"output_index": 0,
"item_id": "msg_refusal_123",
"content_index": 0,
"refusal": "I can't help with that.",
})),
)
.expect("refusal done should parse"),
);
assert!(frames.iter().any(|frame| matches!(
&frame.event,
CanonicalStreamEvent::TextDelta(text) if text == "I can't"
)));
assert!(frames.iter().any(|frame| matches!(
&frame.event,
CanonicalStreamEvent::TextDelta(text) if text == " help with that."
)));
}
#[test]
fn openai_responses_provider_state_does_not_duplicate_text_snapshot_deltas() {
let mut state = OpenAIResponsesProviderState::default();
crates/aether-ai-formats/src/formats/openai/responses/response.rs
+117 -1
View File
@@ -1,4 +1,7 @@
use std::collections::BTreeMap;
use std::{
collections::BTreeMap,
time::{SystemTime, UNIX_EPOCH},
};
use serde_json::{json, Map, Value};
@@ -282,9 +285,86 @@ pub fn to_raw(canonical: &CanonicalResponse, report_context: &Value, _compact: b
OPENAI_RESPONSES_LEGACY_EXTENSION_NAMESPACE,
&response,
));
ensure_modern_openai_responses_response_fields(&mut response);
Value::Object(response)
}
pub(crate) fn ensure_modern_openai_responses_response_fields(
response: &mut Map<String, Value>,
) -> bool {
let mut changed = false;
if !response
.get("output")
.is_some_and(|value| matches!(value, Value::Array(_)))
{
response.insert("output".to_string(), Value::Array(Vec::new()));
changed = true;
}
if !response.contains_key("created_at") {
let created_at = response
.get("created")
.and_then(openai_responses_timestamp_value)
.unwrap_or_else(openai_responses_current_timestamp);
response.insert("created_at".to_string(), Value::from(created_at));
changed = true;
}
if response
.get("status")
.and_then(Value::as_str)
.is_none_or(|status| status == "completed")
&& !response.contains_key("completed_at")
{
let completed_at = response
.get("created_at")
.and_then(openai_responses_timestamp_value)
.unwrap_or_else(openai_responses_current_timestamp);
response.insert("completed_at".to_string(), Value::from(completed_at));
changed = true;
}
if !response.contains_key("output_text") {
let output_text = openai_responses_output_text_from_output(response.get("output"));
response.insert("output_text".to_string(), Value::String(output_text));
changed = true;
}
changed
}
pub(crate) fn openai_responses_output_text_from_output(output: Option<&Value>) -> String {
output
.and_then(Value::as_array)
.into_iter()
.flatten()
.filter_map(Value::as_object)
.flat_map(|item| {
item.get("content")
.and_then(Value::as_array)
.into_iter()
.flatten()
})
.filter_map(|part| {
let part = part.as_object()?;
matches!(
part.get("type").and_then(Value::as_str),
Some("output_text" | "text")
)
.then(|| part.get("text").and_then(Value::as_str).unwrap_or_default())
})
.collect::<String>()
}
pub(crate) fn openai_responses_current_timestamp() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_secs() as i64)
.unwrap_or_default()
}
fn openai_responses_timestamp_value(value: &Value) -> Option<i64> {
value
.as_i64()
.or_else(|| value.as_u64().and_then(|value| i64::try_from(value).ok()))
}
fn image_block_is_generation_call(extensions: &BTreeMap<String, Value>) -> bool {
extensions
.get(OPENAI_RESPONSES_EXTENSION_NAMESPACE)
@@ -379,6 +459,42 @@ mod tests {
assert_eq!(body["output"][0]["status"], "completed");
assert_eq!(body["output"][0]["action"]["type"], "search");
assert_eq!(body["output"][0]["action"]["query"], "today tech");
assert_eq!(body["output_text"], "");
assert!(body["created_at"].as_i64().is_some());
assert!(body["completed_at"].as_i64().is_some());
}
#[test]
fn responses_response_builder_emits_modern_output_text_and_preserves_source_fields() {
let mut extensions = BTreeMap::new();
extensions.insert(
OPENAI_RESPONSES_EXTENSION_NAMESPACE.to_string(),
json!({
"created_at": 111,
"completed_at": 222,
"output_text": "source text",
"conversation": {"id": "conv_123"}
}),
);
let response = CanonicalResponse {
id: "resp_text".to_string(),
model: "gpt-5".to_string(),
content: vec![CanonicalContentBlock::Text {
text: "generated text".to_string(),
extensions: BTreeMap::new(),
}],
outputs: Vec::new(),
stop_reason: Some(CanonicalStopReason::EndTurn),
usage: None,
extensions,
};
let body = to_raw(&response, &json!({}), false);
assert_eq!(body["output_text"], "source text");
assert_eq!(body["created_at"], 111);
assert_eq!(body["completed_at"], 222);
assert_eq!(body["conversation"]["id"], "conv_123");
}
#[test]
crates/aether-ai-formats/src/formats/shared/standard_matrix.rs
+148
View File
@@ -150,6 +150,10 @@ pub fn build_standard_request_body_with_model_directives_and_request_headers(
&mut provider_request_body,
provider_api_format,
);
strip_openai_responses_input_content_cache_control(
&mut provider_request_body,
provider_api_format,
);
let require_body_stream_field = body_json
.as_object()
.is_some_and(|object| object.contains_key("stream"))
@@ -246,9 +250,61 @@ pub fn build_standard_request_body_from_canonical_with_model_directives(
None,
);
}
strip_openai_responses_input_content_cache_control(
&mut provider_request_body,
provider_api_format,
);
Some(provider_request_body)
}
fn strip_openai_responses_input_content_cache_control(
provider_request_body: &mut Value,
provider_api_format: &str,
) {
if !matches!(
aether_ai_formats::normalize_api_format_alias(provider_api_format).as_str(),
"openai:responses" | "openai:responses:compact"
) {
return;
}
let Some(input) = provider_request_body.get_mut("input") else {
return;
};
strip_responses_input_items_content_cache_control(input);
}
fn strip_responses_input_items_content_cache_control(value: &mut Value) {
match value {
Value::Array(items) => {
for item in items {
strip_responses_input_items_content_cache_control(item);
}
}
Value::Object(item) => {
if let Some(content) = item.get_mut("content") {
strip_responses_content_cache_control(content);
}
}
_ => {}
}
}
fn strip_responses_content_cache_control(content: &mut Value) {
match content {
Value::Array(parts) => {
for part in parts {
if let Some(part) = part.as_object_mut() {
part.remove("cache_control");
}
}
}
Value::Object(part) => {
part.remove("cache_control");
}
_ => {}
}
}
pub fn normalize_standard_request_to_openai_chat_request(
body_json: &Value,
client_api_format: &str,
@@ -1236,6 +1292,98 @@ mod tests {
}
}
#[test]
fn standard_openai_responses_strips_content_cache_control_after_body_rules() {
let request = json!({
"model": "gpt-5.1",
"input": [{
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "hello"}]
}],
"prompt_cache_key": "cache_123"
});
let body_rules = json!([
{
"action": "set",
"path": "input[0].content[0].cache_control",
"value": {"type": "ephemeral"}
}
]);
let converted = build_standard_request_body(
&request,
"openai:responses",
"gpt-5.1",
"openai",
"openai:responses",
"/v1/responses",
false,
Some(&body_rules),
None,
)
.expect("responses request should build");
assert_eq!(converted["prompt_cache_key"], "cache_123");
assert!(!converted["input"].to_string().contains("cache_control"));
}
#[test]
fn standard_codex_responses_derives_prompt_cache_key_before_stripping_cache_control() {
fn claude_request(user_text: &str) -> Value {
json!({
"model": "claude-sonnet",
"system": [{
"type": "text",
"text": "stable system brief",
"cache_control": {"type": "ephemeral"}
}],
"messages": [{
"role": "user",
"content": [{"type": "text", "text": user_text}]
}],
"max_tokens": 128
})
}
let body_a = claude_request("new turn A");
let body_b = claude_request("new turn B");
let converted_a = build_standard_request_body(
&body_a,
"claude:messages",
"gpt-5.4",
"codex",
"openai:responses",
"/v1/messages",
true,
None,
Some("key-a"),
)
.expect("claude to codex responses request should build");
let converted_b = build_standard_request_body(
&body_b,
"claude:messages",
"gpt-5.4",
"codex",
"openai:responses",
"/v1/messages",
true,
None,
Some("key-a"),
)
.expect("claude to codex responses request should build");
assert!(converted_a["prompt_cache_key"]
.as_str()
.is_some_and(|value| !value.trim().is_empty()));
assert_eq!(
converted_a["prompt_cache_key"],
converted_b["prompt_cache_key"]
);
assert!(!converted_a.to_string().contains("cache_control"));
assert!(!converted_b.to_string().contains("cache_control"));
}
#[test]
fn builds_openai_chat_request_from_claude_chat_source() {
let request = json!({
crates/aether-ai-formats/src/formats/shared/stream_rewrite.rs
+90 -5
View File
@@ -3,6 +3,7 @@ use std::collections::BTreeMap;
use serde_json::{json, Map, Value};
use crate::formats::openai::image::stream::{OpenAiImageChatStreamState, OpenAiImageStreamState};
use crate::formats::openai::responses::response::ensure_modern_openai_responses_response_fields;
use crate::formats::shared::model_directives::model_directive_display_model_from_report_context;
use crate::formats::shared::response::{
remove_empty_pages_from_tool_arguments, remove_empty_pages_from_tool_input_value,
@@ -20,6 +21,7 @@ use crate::provider_compat::surfaces::{
pub enum FinalizeStreamRewriteMode {
EnvelopeUnwrap,
ModelDirectiveDisplay,
OpenAiResponsesCompat,
OpenAiImage,
OpenAiImageToOpenAiChat,
ClaudeReadToolSanitize,
@@ -91,6 +93,11 @@ pub fn resolve_finalize_stream_rewrite_mode(
// Parsing→rebuilding only adds overhead and may lose information
// (encrypted_content, original item IDs, etc.).
if is_same_format_family(provider_api_format.as_str(), client_api_format.as_str()) {
if is_openai_responses_family(provider_api_format.as_str())
&& is_openai_responses_family(client_api_format.as_str())
{
return Some(FinalizeStreamRewriteMode::OpenAiResponsesCompat);
}
if provider_api_format == "claude:messages" && client_api_format == "claude:messages" {
return Some(FinalizeStreamRewriteMode::ClaudeReadToolSanitize);
}
@@ -128,6 +135,12 @@ pub fn resolve_finalize_stream_rewrite_mode(
return Some(FinalizeStreamRewriteMode::ClaudeReadToolSanitize);
}
if provider_api_format == client_api_format
&& is_openai_responses_family(provider_api_format.as_str())
{
return Some(FinalizeStreamRewriteMode::OpenAiResponsesCompat);
}
(provider_api_format == client_api_format
&& provider_adaptation_should_unwrap_stream_envelope(
envelope_name.as_str(),
@@ -159,6 +172,7 @@ fn client_consumes_same_private_stream_envelope(
enum AiSurfaceStreamRewriteState {
EnvelopeUnwrap,
ModelDirectiveDisplay,
OpenAiResponsesCompat,
OpenAiImage(Box<OpenAiImageStreamState>),
OpenAiImageToOpenAiChat(Box<OpenAiImageChatStreamState>),
ClaudeReadToolSanitize(Box<ClaudeReadToolStreamSanitizer>),
@@ -185,6 +199,9 @@ pub fn maybe_build_ai_surface_stream_rewriter<'a>(
FinalizeStreamRewriteMode::ModelDirectiveDisplay => {
AiSurfaceStreamRewriteState::ModelDirectiveDisplay
}
FinalizeStreamRewriteMode::OpenAiResponsesCompat => {
AiSurfaceStreamRewriteState::OpenAiResponsesCompat
}
FinalizeStreamRewriteMode::OpenAiImage => {
AiSurfaceStreamRewriteState::OpenAiImage(Box::<OpenAiImageStreamState>::default())
}
@@ -240,6 +257,7 @@ impl AiSurfaceStreamRewriter<'_> {
}
AiSurfaceStreamRewriteState::EnvelopeUnwrap
| AiSurfaceStreamRewriteState::ModelDirectiveDisplay
| AiSurfaceStreamRewriteState::OpenAiResponsesCompat
| AiSurfaceStreamRewriteState::Standard(_) => {
self.buffered.extend_from_slice(chunk);
let mut output = Vec::new();
@@ -275,6 +293,7 @@ impl AiSurfaceStreamRewriter<'_> {
}
AiSurfaceStreamRewriteState::EnvelopeUnwrap
| AiSurfaceStreamRewriteState::ModelDirectiveDisplay
| AiSurfaceStreamRewriteState::OpenAiResponsesCompat
| AiSurfaceStreamRewriteState::Standard(_) => {
if self.buffered.is_empty() {
if let AiSurfaceStreamRewriteState::Standard(state) = &mut self.state {
@@ -302,6 +321,9 @@ impl AiSurfaceStreamRewriter<'_> {
AiSurfaceStreamRewriteState::ModelDirectiveDisplay => {
rewrite_model_directive_stream_line(self.report_context, line)
}
AiSurfaceStreamRewriteState::OpenAiResponsesCompat => {
rewrite_openai_responses_compat_stream_line(self.report_context, line)
}
AiSurfaceStreamRewriteState::Standard(state) => {
transform_standard_line(state, self.report_context, line)
}
@@ -611,6 +633,50 @@ fn rewrite_model_directive_stream_line(
Ok(output)
}
fn rewrite_openai_responses_compat_stream_line(
report_context: &Value,
line: Vec<u8>,
) -> Result<Vec<u8>, AiSurfaceFinalizeError> {
let text = match std::str::from_utf8(&line) {
Ok(text) => text,
Err(_) => return Ok(line),
};
let trimmed_line_end = text.trim_end_matches(['\r', '\n']);
let trailing = &text[trimmed_line_end.len()..];
let Some((prefix, payload)) = trimmed_line_end.split_once(':') else {
return Ok(line);
};
if prefix.trim() != "data" {
return Ok(line);
}
let payload = payload.trim_start();
if payload.is_empty() || payload == "[DONE]" {
return Ok(line);
}
let mut value = match serde_json::from_str::<Value>(payload) {
Ok(value) => value,
Err(_) => return Ok(line),
};
let mut changed = rewrite_stream_payload_model_from_context(report_context, &mut value);
let event_type = value
.get("type")
.and_then(Value::as_str)
.unwrap_or_default();
if matches!(event_type, "response.completed" | "response.done") {
if let Some(response) = value.get_mut("response").and_then(Value::as_object_mut) {
changed |= ensure_modern_openai_responses_response_fields(response);
}
}
if !changed {
return Ok(line);
}
let mut output = Vec::new();
output.extend_from_slice(b"data: ");
output.extend(serde_json::to_vec(&value)?);
output.extend_from_slice(trailing.as_bytes());
Ok(output)
}
fn rewrite_stream_payload_model(value: &mut Value, display_model: &str) -> bool {
let Some(object) = value.as_object_mut() else {
return false;
@@ -975,16 +1041,35 @@ data: {\"type\":\"response.output_item.added\",\"response_id\":\"resp_123\",\"ou
}
#[test]
fn same_family_responses_without_display_model_passes_through_verbatim() {
// When provider and client are both OpenAI Responses family but
// there is no display model override, the rewriter returns None
// (complete passthrough, no interception at all).
fn same_family_responses_without_display_model_runs_terminal_compat_only() {
let report_context = json!({
"provider_api_format": "openai:responses",
"client_api_format": "openai:responses:compact",
"needs_conversion": true,
});
assert!(maybe_build_ai_surface_stream_rewriter(Some(&report_context)).is_none());
let mut rewriter = maybe_build_ai_surface_stream_rewriter(Some(&report_context))
.expect("responses compat rewriter should exist");
let mut output = rewriter
.push_chunk(
b"event: response.output_item.added\n\
data: {\"type\":\"response.output_item.added\",\"response_id\":\"resp_123\",\"output_index\":0,\"item\":{\"type\":\"reasoning\",\"id\":\"rs_abc\",\"encrypted_content\":\"EWxvY2tlZA==\"}}\n\n",
)
.expect("non-terminal event should pass through");
output.extend(
rewriter
.push_chunk(
b"event: response.completed\n\
data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_123\",\"object\":\"response\",\"model\":\"gpt-5\",\"status\":\"completed\"}}\n\n",
)
.expect("terminal event should be normalized"),
);
let output = String::from_utf8(output).expect("output should be utf8");
assert!(output.contains("\"encrypted_content\":\"EWxvY2tlZA==\""));
assert!(output.contains("event: response.completed"));
assert!(output.contains("\"output\":[]"));
assert!(output.contains("\"output_text\":\"\""));
assert!(output.contains("\"completed_at\":"));
}
#[test]
crates/aether-ai-formats/src/formats/shared/sync_products.rs
+249 -3
View File
@@ -7,6 +7,7 @@ use aether_ai_formats::formats::conversion::response::{
convert_openai_chat_response_to_openai_responses,
convert_openai_responses_response_to_openai_chat,
};
use aether_ai_formats::formats::openai::responses::response::ensure_modern_openai_responses_response_fields;
use aether_ai_formats::formats::registry::{convert_response, FormatContext};
use aether_ai_formats::{
canonical_to_claude_response, canonical_to_embedding_response, canonical_to_gemini_response,
@@ -1764,6 +1765,50 @@ pub fn aggregate_openai_responses_stream_sync_response(body: &[u8]) -> Option<Va
part,
);
}
"response.output_text.annotation.added" => {
let output_index = openai_responses_event_output_index(event_object).unwrap_or(0);
let content_index = openai_responses_event_content_index(event_object);
merge_openai_responses_message_text_annotation(
message_states.entry(output_index).or_default(),
content_index,
event_object,
);
}
"response.refusal.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 = event_object
.get("delta")
.and_then(Value::as_str)
.unwrap_or_default();
append_openai_responses_message_refusal_delta(
message_states.entry(output_index).or_default(),
content_index,
delta,
);
}
"response.refusal.done" => {
let output_index = openai_responses_event_output_index(event_object).unwrap_or(0);
let content_index = openai_responses_event_content_index(event_object);
let part = event_object.get("part").and_then(Value::as_object);
let refusal = event_object
.get("refusal")
.and_then(Value::as_str)
.or_else(|| {
event_object
.get("part")
.and_then(Value::as_object)
.and_then(|part| part.get("refusal"))
.and_then(Value::as_str)
})
.unwrap_or_default();
merge_openai_responses_message_refusal_part(
message_states.entry(output_index).or_default(),
content_index,
refusal,
part,
);
}
"response.content_part.added" | "response.content_part.done" => {
let Some(part) = event_object.get("part").and_then(Value::as_object) else {
continue;
@@ -1776,7 +1821,7 @@ pub fn aggregate_openai_responses_stream_sync_response(body: &[u8]) -> Option<Va
part,
);
}
"response.reasoning_summary_text.delta" => {
"response.reasoning_text.delta" | "response.reasoning_summary_text.delta" => {
let output_index = openai_responses_event_output_index(event_object).unwrap_or(0);
let delta = event_object
.get("delta")
@@ -1791,7 +1836,7 @@ pub fn aggregate_openai_responses_stream_sync_response(body: &[u8]) -> Option<Va
.summary_text
.push_str(delta);
}
"response.reasoning_summary_text.done" => {
"response.reasoning_text.done" | "response.reasoning_summary_text.done" => {
let output_index = openai_responses_event_output_index(event_object).unwrap_or(0);
let text = event_object
.get("text")
@@ -1917,7 +1962,7 @@ pub fn aggregate_openai_responses_stream_sync_response(body: &[u8]) -> Option<Va
output_index,
);
}
"response.completed" => {
"response.completed" | "response.done" => {
response_object = event_object
.get("response")
.and_then(Value::as_object)
@@ -1988,6 +2033,7 @@ pub fn aggregate_openai_responses_stream_sync_response(body: &[u8]) -> Option<Va
.and_then(Value::as_array)
.is_some_and(|output| !output.is_empty())
{
ensure_modern_openai_responses_response_fields(&mut response);
return Some(Value::Object(response));
}
@@ -2026,6 +2072,8 @@ pub fn aggregate_openai_responses_stream_sync_response(body: &[u8]) -> Option<Va
response.insert("output".to_string(), Value::Array(output));
}
ensure_modern_openai_responses_response_fields(&mut response);
Some(Value::Object(response))
}
@@ -2081,6 +2129,13 @@ fn default_openai_responses_output_text_part() -> Value {
})
}
fn default_openai_responses_refusal_part() -> Value {
json!({
"type": "refusal",
"refusal": "",
})
}
fn append_openai_responses_message_text_delta(
state: &mut OpenAIResponsesSyncMessageState,
content_index: usize,
@@ -2190,6 +2245,115 @@ fn merge_openai_responses_message_text_part(
.or_insert_with(|| Value::Array(Vec::new()));
}
fn append_openai_responses_message_refusal_delta(
state: &mut OpenAIResponsesSyncMessageState,
content_index: usize,
delta: &str,
) {
if delta.is_empty() {
return;
}
let part = state
.parts
.entry(content_index)
.or_insert_with(default_openai_responses_refusal_part);
let Some(part) = part.as_object_mut() else {
return;
};
if part.get("type").and_then(Value::as_str) != Some("refusal") {
return;
}
let current = part
.get("refusal")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
part.insert("type".to_string(), Value::String("refusal".to_string()));
part.insert(
"refusal".to_string(),
Value::String(format!("{current}{delta}")),
);
}
fn merge_openai_responses_message_refusal_part(
state: &mut OpenAIResponsesSyncMessageState,
content_index: usize,
refusal: &str,
template_part: Option<&Map<String, Value>>,
) {
if refusal.is_empty() && template_part.is_none() {
return;
}
let part = state.parts.entry(content_index).or_insert_with(|| {
template_part
.map(|part| Value::Object(part.clone()))
.unwrap_or_else(default_openai_responses_refusal_part)
});
let Some(part) = part.as_object_mut() else {
return;
};
if let Some(template_part) = template_part {
for (key, value) in template_part {
if key != "refusal" {
part.insert(key.clone(), value.clone());
}
}
}
part.insert("type".to_string(), Value::String("refusal".to_string()));
let mut current = part
.get("refusal")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
reconcile_openai_responses_authoritative_text(&mut current, refusal);
part.insert("refusal".to_string(), Value::String(current));
}
fn merge_openai_responses_message_text_annotation(
state: &mut OpenAIResponsesSyncMessageState,
content_index: usize,
event: &Map<String, Value>,
) {
let Some(annotation) = event.get("annotation") else {
return;
};
let annotation_index = event
.get("annotation_index")
.and_then(Value::as_u64)
.map(|value| value as usize);
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;
}
part.insert("type".to_string(), Value::String("output_text".to_string()));
part.entry("text".to_string())
.or_insert_with(|| Value::String(String::new()));
let annotations = part
.entry("annotations".to_string())
.or_insert_with(|| Value::Array(Vec::new()));
let Some(annotations) = annotations.as_array_mut() else {
return;
};
if let Some(annotation_index) = annotation_index {
if annotations.len() <= annotation_index {
annotations.resize(annotation_index + 1, Value::Null);
}
annotations[annotation_index] = annotation.clone();
} else {
annotations.push(annotation.clone());
}
}
fn merge_openai_responses_message_part(
state: &mut OpenAIResponsesSyncMessageState,
content_index: usize,
@@ -2202,6 +2366,12 @@ fn merge_openai_responses_message_part(
{
let text = part.get("text").and_then(Value::as_str).unwrap_or_default();
merge_openai_responses_message_text_part(state, content_index, text, Some(part));
} else if part.get("type").and_then(Value::as_str) == Some("refusal") {
let refusal = part
.get("refusal")
.and_then(Value::as_str)
.unwrap_or_default();
merge_openai_responses_message_refusal_part(state, content_index, refusal, Some(part));
} else {
state
.parts
@@ -3691,6 +3861,9 @@ mod tests {
assert_eq!(body_json.get("id"), Some(&json!("resp_123")));
assert_eq!(body_json.get("status"), Some(&json!("completed")));
assert_eq!(body_json["output"][0]["content"][0]["text"], json!("Hello"));
assert_eq!(body_json["output_text"], "Hello");
assert!(body_json["created_at"].as_i64().is_some());
assert!(body_json["completed_at"].as_i64().is_some());
}
#[test]
@@ -3818,6 +3991,58 @@ mod tests {
assert_eq!(result["output"][0]["content"][0]["refusal"], "blocked");
}
#[test]
fn aggregates_official_refusal_stream_events() {
let body = concat!(
"event: response.output_item.added\n",
"data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"message\",\"id\":\"msg_refusal_123\",\"role\":\"assistant\",\"status\":\"in_progress\",\"content\":[]}}\n\n",
"event: response.content_part.added\n",
"data: {\"type\":\"response.content_part.added\",\"output_index\":0,\"content_index\":0,\"part\":{\"type\":\"refusal\",\"refusal\":\"\"}}\n\n",
"event: response.refusal.delta\n",
"data: {\"type\":\"response.refusal.delta\",\"output_index\":0,\"content_index\":0,\"item_id\":\"msg_refusal_123\",\"delta\":\"I can't\"}\n\n",
"event: response.refusal.done\n",
"data: {\"type\":\"response.refusal.done\",\"output_index\":0,\"content_index\":0,\"item_id\":\"msg_refusal_123\",\"refusal\":\"I can't help with that.\"}\n\n",
"event: response.completed\n",
"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_refusal_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 refusal stream should aggregate into a sync body");
assert_eq!(result["output"][0]["content"][0]["type"], "refusal");
assert_eq!(
result["output"][0]["content"][0]["refusal"],
"I can't help with that."
);
assert_eq!(result["output_text"], "");
}
#[test]
fn aggregates_official_output_text_annotation_added_event() {
let body = concat!(
"event: response.output_text.delta\n",
"data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"content_index\":0,\"delta\":\"Hello annotated\"}\n\n",
"event: response.output_text.annotation.added\n",
"data: {\"type\":\"response.output_text.annotation.added\",\"output_index\":0,\"content_index\":0,\"annotation_index\":0,\"annotation\":{\"type\":\"text_annotation\",\"text\":\"annotated\",\"start\":6,\"end\":15}}\n\n",
"event: response.completed\n",
"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_annotation_added_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 annotation stream should aggregate into a sync body");
assert_eq!(result["output"][0]["content"][0]["text"], "Hello annotated");
assert_eq!(
result["output"][0]["content"][0]["annotations"][0]["type"],
"text_annotation"
);
assert_eq!(
result["output"][0]["content"][0]["annotations"][0]["start"],
6
);
assert_eq!(result["output_text"], "Hello annotated");
}
#[test]
fn authoritative_output_text_done_preserves_annotations() {
let body = concat!(
@@ -3863,6 +4088,27 @@ mod tests {
assert_eq!(result["output"][0]["arguments"], r#"{"location": "Tokyo"}"#);
}
#[test]
fn aggregates_modern_reasoning_text_and_response_done_alias() {
let body = concat!(
"event: response.reasoning_text.delta\n",
"data: {\"type\":\"response.reasoning_text.delta\",\"output_index\":0,\"delta\":\"Need\"}\n\n",
"event: response.reasoning_text.done\n",
"data: {\"type\":\"response.reasoning_text.done\",\"output_index\":0,\"text\":\"Need care\"}\n\n",
"event: response.done\n",
"data: {\"type\":\"response.done\",\"response\":{\"id\":\"resp_done_alias_123\",\"object\":\"response\",\"model\":\"gpt-5\",\"status\":\"completed\"}}\n\n",
);
let result = aggregate_openai_responses_stream_sync_response(body.as_bytes())
.expect("modern response.done stream should aggregate");
assert_eq!(result["output"][0]["type"], "reasoning");
assert_eq!(result["output"][0]["summary"][0]["text"], "Need care");
assert!(result["output"].as_array().is_some());
assert_eq!(result["output_text"], "");
assert!(result["completed_at"].as_i64().is_some());
}
#[test]
fn accepts_openai_responses_same_family_stream_when_needs_conversion_is_true() {
let body = concat!(