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

Merge pull request #441 from zhefox/aether-rust-pioneer

Browse Source 
fix: add codex reasoning defaults and stream rewrite tests
This commit is contained in:
fawney19
2026-05-14 00:35:45 +08:00
committed by GitHub
parent acd44328d6 697b6e0653
commit eb654c2fe7
9 changed files with 828 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
crates/aether-ai-formats/src/formats/claude/messages/stream.rs
View File

@@ -25,6 +25,11 @@ pub struct ClaudeProviderState {
finished: bool,
usage: Option<CanonicalUsage>,
tool_calls: BTreeMap<usize, ClaudeProviderToolState>,
/// True while we are inside a thinking content block. Used to emit
/// `ReasoningSummaryDone` when the block closes, so that downstream
/// emitters can insert paragraph separators between distinct thinking
/// blocks (CPA strategy).
in_thinking_block: bool,
}
impl ClaudeProviderState {
@@ -215,6 +220,7 @@ impl ClaudeProviderState {
.and_then(Value::as_str)
.unwrap_or_default();
if block_type == "thinking" {
self.in_thinking_block = true;
let Some(piece) = block
.get("thinking")
.and_then(Value::as_str)
@@ -331,7 +337,22 @@ impl ClaudeProviderState {
});
self.finished = true;
}
"content_block_stop" | "message_stop" | "ping" => {}
"content_block_stop" => {
// CPA strategy: when a thinking block closes, emit
// ReasoningSummaryDone so downstream emitters can insert
// paragraph separators between distinct thinking blocks.
if self.in_thinking_block {
self.in_thinking_block = false;
self.ensure_started(report_context, &mut out);
let (id, model) = self.identity(report_context);
out.push(CanonicalStreamFrame {
id,
model,
event: CanonicalStreamEvent::ReasoningSummaryDone,
});
}
}
"message_stop" | "ping" => {}
_ => {
out.push(self.unknown_frame(report_context, value.clone()));
}
@@ -746,6 +767,12 @@ impl ClaudeClientEmitter {
self.finished = true;
Ok(out)
}
CanonicalStreamEvent::ReasoningSummaryDone => {
// CPA strategy: close the current thinking block so the next
// ReasoningDelta opens a fresh one. Each reasoning paragraph
// becomes its own thinking block in Claude's wire format.
Ok(self.close_open_block().unwrap_or_default())
}
}
}

1
crates/aether-ai-formats/src/formats/gemini/generate_content/stream.rs
View File

@@ -551,6 +551,7 @@ impl GeminiClientEmitter {
self.finished = true;
Ok(out)
}
CanonicalStreamEvent::ReasoningSummaryDone => Ok(Vec::new()),
}
}

511
crates/aether-ai-formats/src/formats/openai/chat/stream.rs
View File

@@ -46,6 +46,7 @@ pub struct OpenAIResponsesProviderState {
finished: bool,
text: String,
reasoning: String,
reasoning_parts: BTreeMap<usize, String>,
tool_calls: BTreeMap<usize, OpenAIResponsesProviderToolState>,
tool_results: BTreeMap<usize, OpenAIResponsesProviderToolResultState>,
tool_index_by_key: BTreeMap<String, usize>,
@@ -469,6 +470,45 @@ impl OpenAIResponsesProviderState {
});
}
fn emit_missing_reasoning_part_text(
&mut self,
report_context: &Value,
out: &mut Vec<CanonicalStreamFrame>,
summary_index: usize,
text: &str,
) {
if text.is_empty() {
return;
}
let missing = {
let current = self.reasoning_parts.entry(summary_index).or_default();
let missing = if text.starts_with(current.as_str()) {
text[current.len()..].to_string()
} else if current.as_str() == text {
String::new()
} else if current.is_empty() {
text.to_string()
} else {
String::new()
};
if !missing.is_empty() {
current.push_str(&missing);
}
missing
};
if missing.is_empty() {
return;
}
self.ensure_started(report_context, out);
self.reasoning.push_str(&missing);
let (id, model) = self.identity(report_context);
out.push(CanonicalStreamFrame {
id,
model,
event: CanonicalStreamEvent::ReasoningDelta(missing),
});
}
fn emit_tool_call_item(
&mut self,
report_context: &Value,
@@ -744,9 +784,17 @@ impl OpenAIResponsesProviderState {
if let Some(part) = value.get("part").and_then(Value::as_object) {
if part.get("type").and_then(Value::as_str) == Some("summary_text") {
if let Some(text) = part.get("text").and_then(Value::as_str) {
if !text.is_empty() {
self.emit_missing_reasoning(report_context, &mut out, text);
}
let summary_index = value
.get("summary_index")
.and_then(Value::as_u64)
.map(|value| value as usize)
.unwrap_or(0);
self.emit_missing_reasoning_part_text(
report_context,
&mut out,
summary_index,
text,
);
}
}
}
@@ -773,8 +821,17 @@ impl OpenAIResponsesProviderState {
.and_then(Value::as_str)
.unwrap_or_default();
if !piece.is_empty() {
let summary_index = value
.get("summary_index")
.and_then(Value::as_u64)
.map(|value| value as usize)
.unwrap_or(0);
self.ensure_started(report_context, &mut out);
self.reasoning.push_str(piece);
self.reasoning_parts
.entry(summary_index)
.or_default()
.push_str(piece);
let (id, model) = self.identity(report_context);
out.push(CanonicalStreamFrame {
id,
@@ -796,8 +853,25 @@ impl OpenAIResponsesProviderState {
})
.unwrap_or_default();
if !text.is_empty() {
self.emit_missing_reasoning(report_context, &mut out, text);
let summary_index = value
.get("summary_index")
.and_then(Value::as_u64)
.map(|value| value as usize)
.unwrap_or(0);
self.emit_missing_reasoning_part_text(
report_context,
&mut out,
summary_index,
text,
);
}
self.ensure_started(report_context, &mut out);
let (id, model) = self.identity(report_context);
out.push(CanonicalStreamFrame {
id,
model,
event: CanonicalStreamEvent::ReasoningSummaryDone,
});
}
"response.output_item.added" => {
let Some(item) = value.get("item").and_then(Value::as_object) else {
@@ -818,7 +892,7 @@ impl OpenAIResponsesProviderState {
self.emit_message_item(report_context, &mut out, item);
}
"reasoning" => {
self.emit_reasoning_item(report_context, &mut out, item);
self.ensure_started(report_context, &mut out);
}
_ => {
out.push(self.unknown_frame(report_context, Value::Object(item.clone())));
@@ -1177,6 +1251,8 @@ pub struct OpenAIResponsesClientEmitter {
message_output_index: Option<usize>,
text: String,
reasoning: String,
reasoning_part: String,
reasoning_summary_parts: Vec<String>,
tool_calls: BTreeMap<usize, OpenAIResponsesClientToolState>,
tool_results: BTreeMap<usize, OpenAIResponsesClientToolResultState>,
}
@@ -1244,6 +1320,29 @@ impl OpenAIChatClientEmitter {
)?);
Ok(out)
}
CanonicalStreamEvent::ReasoningSummaryDone => {
// CPA strategy: emit "\n\n" as paragraph separator between
// reasoning sections, matching CPA's Chat downstream behavior.
let mut out = self.ensure_started()?;
out.extend(encode_json_sse(
None,
&json!({
"id": self.response_id
.as_deref()
.unwrap_or("chatcmpl-local-stream"),
"object": "chat.completion.chunk",
"model": self.model.as_deref().unwrap_or("unknown"),
"choices": [{
"index": 0,
"delta": {
"reasoning_content": "\n\n",
},
"finish_reason": Value::Null
}]
}),
)?);
Ok(out)
}
CanonicalStreamEvent::ReasoningSignature(_) => Ok(Vec::new()),
CanonicalStreamEvent::ContentPart(part) => {
let placeholder = openai_stream_placeholder_for_content_part(&part);
@@ -1317,10 +1416,10 @@ impl OpenAIChatClientEmitter {
Ok(out)
}
CanonicalStreamEvent::ToolResultDelta {
index: _,
tool_use_id,
name,
content,
..
} => {
let mut out = self.ensure_started()?;
let mut delta = Map::new();
@@ -1515,6 +1614,10 @@ impl OpenAIResponsesClientEmitter {
output_index
}
fn current_reasoning_summary_index(&self) -> usize {
self.reasoning_summary_parts.len()
}
fn ensure_message_output_index(&mut self) -> usize {
if let Some(output_index) = self.message_output_index {
return output_index;
@@ -1571,6 +1674,7 @@ impl OpenAIResponsesClientEmitter {
self.reasoning_item_started = true;
}
if !self.reasoning_part_started {
let summary_index = self.current_reasoning_summary_index();
out.extend(self.encode_response_event(
"response.reasoning_summary_part.added",
json!({
@@ -1578,7 +1682,7 @@ impl OpenAIResponsesClientEmitter {
"response_id": self.response_id(),
"item_id": item_id,
"output_index": output_index,
"summary_index": 0,
"summary_index": summary_index,
"part": {
"type": "summary_text",
"text": "",
@@ -1698,6 +1802,8 @@ impl OpenAIResponsesClientEmitter {
let item_id = self.reasoning_item_id();
let mut out = Vec::new();
if self.reasoning_part_started {
let summary_index = self.current_reasoning_summary_index();
let part_text = self.reasoning_part.clone();
out.extend(self.encode_response_event(
"response.reasoning_summary_text.done",
json!({
@@ -1705,8 +1811,8 @@ impl OpenAIResponsesClientEmitter {
"response_id": self.response_id(),
"item_id": item_id.clone(),
"output_index": output_index,
"summary_index": 0,
"text": self.reasoning.as_str(),
"summary_index": summary_index,
"text": part_text.as_str(),
}),
)?);
out.extend(self.encode_response_event(
@@ -1716,14 +1822,37 @@ impl OpenAIResponsesClientEmitter {
"response_id": self.response_id(),
"item_id": item_id.clone(),
"output_index": output_index,
"summary_index": 0,
"summary_index": summary_index,
"part": {
"type": "summary_text",
"text": self.reasoning.as_str(),
"text": part_text.as_str(),
}
}),
)?);
self.reasoning_summary_parts.push(part_text);
self.reasoning_part.clear();
self.reasoning_part_started = false;
}
let summary = if self.reasoning_summary_parts.is_empty() {
if self.reasoning.trim().is_empty() {
Vec::new()
} else {
vec![json!({
"type": "summary_text",
"text": self.reasoning.as_str(),
})]
}
} else {
self.reasoning_summary_parts
.iter()
.map(|text| {
json!({
"type": "summary_text",
"text": text,
})
})
.collect::<Vec<_>>()
};
out.extend(self.encode_response_event(
"response.output_item.done",
json!({
@@ -1733,10 +1862,7 @@ impl OpenAIResponsesClientEmitter {
"item": {
"type": "reasoning",
"id": item_id,
"summary": [{
"type": "summary_text",
"text": self.reasoning.as_str(),
}],
"summary": summary,
}
}),
)?);
@@ -1808,10 +1934,15 @@ impl OpenAIResponsesClientEmitter {
);
item.insert("id".to_string(), Value::String(format!("{item_id}_output")));
item.insert("call_id".to_string(), Value::String(item_id));
if let Some(name) = state.name.filter(|value| !value.trim().is_empty()) {
if let Some(name) = state
.name
.as_ref()
.filter(|value| !value.trim().is_empty())
.cloned()
{
item.insert("name".to_string(), Value::String(name));
}
item.insert("output".to_string(), Value::String(state.content));
item.insert("output".to_string(), Value::String(state.content.clone()));
out.extend(self.encode_response_event(
"response.output_item.done",
json!({
@@ -1827,17 +1958,34 @@ impl OpenAIResponsesClientEmitter {
fn completed_response(&self, usage: CanonicalUsage) -> Value {
let mut ordered_output = Vec::new();
if !self.reasoning.trim().is_empty() {
let summary = if self.reasoning_summary_parts.is_empty() {
if self.reasoning.trim().is_empty() {
Vec::new()
} else {
vec![json!({
"type": "summary_text",
"text": self.reasoning.as_str(),
})]
}
} else {
self.reasoning_summary_parts
.iter()
.map(|text| {
json!({
"type": "summary_text",
"text": text,
})
})
.collect::<Vec<_>>()
};
if !summary.is_empty() {
ordered_output.push((
self.reasoning_output_index.unwrap_or(0),
json!({
"type": "reasoning",
"id": self.reasoning_item_id(),
"status": "completed",
"summary": [{
"type": "summary_text",
"text": self.reasoning.as_str(),
}]
"summary": summary,
}),
));
}
@@ -1962,6 +2110,7 @@ impl OpenAIResponsesClientEmitter {
CanonicalStreamEvent::ReasoningDelta(text) => {
let mut out = self.ensure_reasoning_item_started()?;
self.reasoning.push_str(&text);
self.reasoning_part.push_str(&text);
out.extend(self.encode_response_event(
"response.reasoning_summary_text.delta",
json!({
@@ -1969,12 +2118,53 @@ impl OpenAIResponsesClientEmitter {
"response_id": self.response_id(),
"item_id": self.reasoning_item_id(),
"output_index": self.reasoning_output_index.unwrap_or(0),
"summary_index": 0,
"summary_index": self.current_reasoning_summary_index(),
"delta": text,
}),
)?);
Ok(out)
}
CanonicalStreamEvent::ReasoningSummaryDone => {
// Close the current reasoning part and reset state so the next
// ReasoningDelta starts a fresh part within the same item.
if !self.reasoning_item_started || !self.reasoning_part_started {
return Ok(Vec::new());
}
let output_index = self.reasoning_output_index.unwrap_or(0);
let item_id = self.reasoning_item_id();
let summary_index = self.current_reasoning_summary_index();
let part_text = self.reasoning_part.clone();
let mut out = Vec::new();
out.extend(self.encode_response_event(
"response.reasoning_summary_text.done",
json!({
"type": "response.reasoning_summary_text.done",
"response_id": self.response_id(),
"item_id": item_id.clone(),
"output_index": output_index,
"summary_index": summary_index,
"text": part_text.as_str(),
}),
)?);
out.extend(self.encode_response_event(
"response.reasoning_summary_part.done",
json!({
"type": "response.reasoning_summary_part.done",
"response_id": self.response_id(),
"item_id": item_id,
"output_index": output_index,
"summary_index": summary_index,
"part": {
"type": "summary_text",
"text": part_text.as_str(),
}
}),
)?);
self.reasoning_summary_parts.push(part_text);
self.reasoning_part.clear();
self.reasoning_part_started = false;
Ok(out)
}
CanonicalStreamEvent::ReasoningSignature(_) => Ok(Vec::new()),
CanonicalStreamEvent::ContentPart(part) => {
let placeholder = openai_stream_placeholder_for_content_part(&part);
@@ -2238,6 +2428,38 @@ mod tests {
sequence_numbers
}
fn response_reasoning_text_done_parts(sse: &str) -> Vec<(u64, String)> {
let mut parts = Vec::new();
for block in sse.split("\n\n") {
let mut event_name = None;
let mut data = None;
for line in block.lines() {
if let Some(value) = line.strip_prefix("event: ") {
event_name = Some(value);
} else if let Some(value) = line.strip_prefix("data: ") {
data = Some(value);
}
}
if event_name != Some("response.reasoning_summary_text.done") {
continue;
}
let Some(data) = data else {
continue;
};
let Ok(value) = serde_json::from_str::<Value>(data) else {
continue;
};
let Some(summary_index) = value.get("summary_index").and_then(Value::as_u64) else {
continue;
};
let Some(text) = value.get("text").and_then(Value::as_str) else {
continue;
};
parts.push((summary_index, text.to_string()));
}
parts
}
#[test]
fn openai_chat_provider_state_emits_unknown_events_for_unrecognized_deltas() {
let mut state = OpenAIChatProviderState::default();
@@ -2979,6 +3201,72 @@ mod tests {
assert_eq!(response_sequence_numbers(&sse), (1..=9).collect::<Vec<_>>());
}
#[test]
fn openai_responses_client_emitter_closes_distinct_reasoning_parts() {
let mut emitter = OpenAIResponsesClientEmitter::default();
let mut bytes = emitter
.emit(CanonicalStreamFrame {
id: "resp_789".to_string(),
model: "gpt-5.4".to_string(),
event: CanonicalStreamEvent::Start,
})
.expect("start should encode");
bytes.extend(
emitter
.emit(CanonicalStreamFrame {
id: "resp_789".to_string(),
model: "gpt-5.4".to_string(),
event: CanonicalStreamEvent::ReasoningDelta("alpha".to_string()),
})
.expect("first reasoning should encode"),
);
bytes.extend(
emitter
.emit(CanonicalStreamFrame {
id: "resp_789".to_string(),
model: "gpt-5.4".to_string(),
event: CanonicalStreamEvent::ReasoningSummaryDone,
})
.expect("first boundary should encode"),
);
bytes.extend(
emitter
.emit(CanonicalStreamFrame {
id: "resp_789".to_string(),
model: "gpt-5.4".to_string(),
event: CanonicalStreamEvent::ReasoningDelta("beta".to_string()),
})
.expect("second reasoning should encode"),
);
bytes.extend(
emitter
.emit(CanonicalStreamFrame {
id: "resp_789".to_string(),
model: "gpt-5.4".to_string(),
event: CanonicalStreamEvent::ReasoningSummaryDone,
})
.expect("second boundary should encode"),
);
bytes.extend(
emitter
.emit(CanonicalStreamFrame {
id: "resp_789".to_string(),
model: "gpt-5.4".to_string(),
event: CanonicalStreamEvent::Finish {
finish_reason: Some("stop".to_string()),
usage: None,
},
})
.expect("finish should encode"),
);
let sse = String::from_utf8(bytes).expect("sse should be utf8");
assert_eq!(
response_reasoning_text_done_parts(&sse),
vec![(0, "alpha".to_string()), (1, "beta".to_string())]
);
}
#[test]
fn openai_responses_client_emitter_emits_failed_event_with_sequence_number() {
let mut emitter = OpenAIResponsesClientEmitter::default();
@@ -3096,4 +3384,181 @@ mod tests {
CanonicalStreamEvent::ReasoningDelta(ref text) if text == "step"
));
}
#[test]
fn openai_responses_provider_state_accepts_reasoning_done_without_delta() {
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.created",
"response": {
"id": "resp_done_only",
"model": "gpt-5.4",
}
})),
)
.expect("created should parse"),
);
frames.extend(
state
.push_line(
&report_context,
data_line(json!({
"type": "response.reasoning_summary_text.done",
"response_id": "resp_done_only",
"item_id": "resp_done_only_rs_0",
"output_index": 0,
"summary_index": 0,
"text": "fallback reasoning",
})),
)
.expect("reasoning done should parse"),
);
assert!(frames.iter().any(|frame| matches!(
frame.event,
CanonicalStreamEvent::ReasoningDelta(ref text) if text == "fallback reasoning"
)));
assert!(frames
.iter()
.any(|frame| matches!(frame.event, CanonicalStreamEvent::ReasoningSummaryDone)));
}
#[test]
fn openai_responses_provider_state_does_not_duplicate_part_scoped_reasoning_done() {
let mut state = OpenAIResponsesProviderState::default();
let report_context = json!({});
let mut frames = Vec::new();
for event in [
json!({
"type": "response.reasoning_summary_text.delta",
"response_id": "resp_parts",
"item_id": "resp_parts_rs_0",
"output_index": 0,
"summary_index": 0,
"delta": "alpha",
}),
json!({
"type": "response.reasoning_summary_text.done",
"response_id": "resp_parts",
"item_id": "resp_parts_rs_0",
"output_index": 0,
"summary_index": 0,
"text": "alpha",
}),
json!({
"type": "response.reasoning_summary_text.delta",
"response_id": "resp_parts",
"item_id": "resp_parts_rs_0",
"output_index": 0,
"summary_index": 1,
"delta": "beta",
}),
json!({
"type": "response.reasoning_summary_text.done",
"response_id": "resp_parts",
"item_id": "resp_parts_rs_0",
"output_index": 0,
"summary_index": 1,
"text": "beta",
}),
] {
frames.extend(
state
.push_line(&report_context, data_line(event))
.expect("reasoning event should parse"),
);
}
let reasoning = frames
.iter()
.filter_map(|frame| match &frame.event {
CanonicalStreamEvent::ReasoningDelta(text) => Some(text.as_str()),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(reasoning, vec!["alpha", "beta"]);
}
#[test]
fn openai_responses_provider_state_does_not_duplicate_added_reasoning_item_summary() {
let mut state = OpenAIResponsesProviderState::default();
let report_context = json!({});
let mut frames = Vec::new();
for event in [
json!({
"type": "response.output_item.added",
"response_id": "resp_added_summary",
"output_index": 0,
"item": {
"type": "reasoning",
"id": "resp_added_summary_rs_0",
"summary": [{
"type": "summary_text",
"text": "alpha",
}]
}
}),
json!({
"type": "response.reasoning_summary_text.delta",
"response_id": "resp_added_summary",
"item_id": "resp_added_summary_rs_0",
"output_index": 0,
"summary_index": 0,
"delta": "alpha",
}),
] {
frames.extend(
state
.push_line(&report_context, data_line(event))
.expect("reasoning event should parse"),
);
}
let reasoning = frames
.iter()
.filter_map(|frame| match &frame.event {
CanonicalStreamEvent::ReasoningDelta(text) => Some(text.as_str()),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(reasoning, vec!["alpha"]);
}
#[test]
fn openai_responses_provider_state_uses_reasoning_item_as_fallback() {
let mut state = OpenAIResponsesProviderState::default();
let report_context = json!({});
let frames = state
.push_line(
&report_context,
data_line(json!({
"type": "response.output_item.done",
"response_id": "resp_item_fallback",
"output_index": 0,
"item": {
"type": "reasoning",
"id": "resp_item_fallback_rs_0",
"summary": [{
"type": "summary_text",
"text": "item fallback reasoning",
}]
}
})),
)
.expect("reasoning item should parse");
assert!(frames.iter().any(|frame| matches!(
frame.event,
CanonicalStreamEvent::ReasoningDelta(ref text) if text == "item fallback reasoning"
)));
}
}

106
crates/aether-ai-formats/src/formats/openai/responses/codex.rs
View File

@@ -139,6 +139,50 @@ fn inject_codex_default_variation_prompt(body_object: &mut serde_json::Map<Strin
);
}
fn ensure_codex_reasoning_summary(body_object: &mut serde_json::Map<String, Value>) {
let reasoning = body_object
.entry("reasoning".to_string())
.or_insert_with(|| json!({}));
if !reasoning.is_object() {
*reasoning = json!({});
}
let Some(reasoning_object) = reasoning.as_object_mut() else {
return;
};
reasoning_object
.entry("effort".to_string())
.or_insert_with(|| json!("medium"));
reasoning_object
.entry("summary".to_string())
.or_insert_with(|| json!("auto"));
}
fn ensure_codex_reasoning_include(body_object: &mut serde_json::Map<String, Value>) {
const REASONING_ENCRYPTED_CONTENT: &str = "reasoning.encrypted_content";
match body_object.get_mut("include") {
Some(Value::Array(include)) => {
let has_reasoning_encrypted_content = include
.iter()
.any(|value| value.as_str() == Some(REASONING_ENCRYPTED_CONTENT));
if !has_reasoning_encrypted_content {
include.push(json!(REASONING_ENCRYPTED_CONTENT));
}
}
Some(_) | None => {
body_object.insert("include".to_string(), json!([REASONING_ENCRYPTED_CONTENT]));
}
}
}
fn ensure_codex_responses_defaults(body_object: &mut serde_json::Map<String, Value>) {
ensure_codex_reasoning_summary(body_object);
ensure_codex_reasoning_include(body_object);
body_object
.entry("parallel_tool_calls".to_string())
.or_insert_with(|| json!(true));
}
fn build_stable_codex_prompt_cache_key(user_api_key_id: &str) -> Option<String> {
let normalized = user_api_key_id.trim();
if normalized.is_empty() {
@@ -318,6 +362,7 @@ pub fn apply_codex_openai_responses_special_body_edits(
} else if !body_rules_handle_path(body_rules, "store") {
body_object.insert("store".to_string(), json!(false));
}
ensure_codex_responses_defaults(body_object);
if !body_rules_handle_path(body_rules, "instructions")
&& !body_object.contains_key("instructions")
{
@@ -423,6 +468,67 @@ mod tests {
};
use serde_json::json;
#[test]
fn codex_responses_body_edits_request_reasoning_summary_stream() {
let mut provider_request_body = json!({
"input": [{
"role": "user",
"content": "hello"
}],
"model": "gpt-5.4",
"stream": true
});
apply_codex_openai_responses_special_body_edits(
&mut provider_request_body,
"codex",
"openai:responses",
None,
None,
);
assert_eq!(
provider_request_body["reasoning"]["effort"],
json!("medium")
);
assert_eq!(provider_request_body["reasoning"]["summary"], json!("auto"));
assert_eq!(
provider_request_body["include"],
json!(["reasoning.encrypted_content"])
);
assert_eq!(provider_request_body["parallel_tool_calls"], json!(true));
}
#[test]
fn codex_responses_body_edits_preserve_existing_reasoning_and_include() {
let mut provider_request_body = json!({
"input": [],
"model": "gpt-5.4",
"include": ["file_search_call.results"],
"reasoning": {"effort": "high", "summary": "detailed"},
"parallel_tool_calls": false
});
apply_codex_openai_responses_special_body_edits(
&mut provider_request_body,
"codex",
"openai:responses",
None,
None,
);
assert_eq!(provider_request_body["reasoning"]["effort"], json!("high"));
assert_eq!(
provider_request_body["reasoning"]["summary"],
json!("detailed")
);
assert_eq!(
provider_request_body["include"],
json!(["file_search_call.results", "reasoning.encrypted_content"])
);
assert_eq!(provider_request_body["parallel_tool_calls"], json!(false));
}
#[test]
fn codex_image_body_edits_force_tool_choice_and_default_generate_tool_fields() {
let mut provider_request_body = json!({

172
crates/aether-ai-formats/src/formats/shared/stream_rewrite.rs
View File

@@ -58,6 +58,14 @@ pub fn resolve_finalize_stream_rewrite_mode(
}
if needs_conversion {
// CPA strategy: when provider and client share the same wire format
// (exact match or same family), pass through the stream verbatim.
// 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()) {
return model_directive_display_model_from_report_context(report_context)
.map(|_| FinalizeStreamRewriteMode::ModelDirectiveDisplay);
}
return supports_standard_stream_rewrite(
provider_api_format.as_str(),
client_api_format.as_str(),
@@ -336,6 +344,31 @@ fn supports_standard_stream_rewrite(provider_api_format: &str, client_api_format
|| is_standard_cli_client_api_format(client_api_format))
}
/// Returns true for OpenAI Responses family formats that share the same SSE
/// wire format and can be passed through without parsing→rebuilding.
fn is_openai_responses_family(api_format: &str) -> bool {
matches!(
aether_ai_formats::normalize_api_format_alias(api_format).as_str(),
"openai:responses" | "openai:responses:compact"
)
}
/// Returns true when two API formats share the same SSE wire format and
/// can be passed through without parsing→rebuilding. This covers:
///
/// - Exact matches after normalisation (e.g. `claude:messages` ↔ `claude:messages`)
/// - OpenAI Responses family (`openai:responses` ↔ `openai:responses:compact`)
fn is_same_format_family(provider_format: &str, client_format: &str) -> bool {
let provider = aether_ai_formats::normalize_api_format_alias(provider_format);
let client = aether_ai_formats::normalize_api_format_alias(client_format);
if provider == client {
return true;
}
// OpenAI Responses family shares the same wire format despite having
// distinct format IDs.
is_openai_responses_family(provider_format) && is_openai_responses_family(client_format)
}
fn is_standard_provider_api_format(api_format: &str) -> bool {
matches!(
aether_ai_formats::normalize_api_format_alias(api_format).as_str(),
@@ -502,6 +535,145 @@ data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_123\",\"object\
assert!(!output.contains("\"model\":\"gpt-5.5\""));
}
#[test]
fn standard_rewriter_converts_openai_responses_reasoning_delta_to_chat() {
let report_context = json!({
"provider_api_format": "openai:responses",
"client_api_format": "openai:chat",
"needs_conversion": true,
"mapped_model": "gpt-5.4",
});
let mut rewriter = maybe_build_ai_surface_stream_rewriter(Some(&report_context))
.expect("rewriter should exist");
let output = rewriter
.push_chunk(
b"event: response.reasoning_summary_text.delta\n\
data: {\"type\":\"response.reasoning_summary_text.delta\",\"response_id\":\"resp_reasoning_stream_123\",\"item_id\":\"rs_123\",\"output_index\":0,\"summary_index\":0,\"delta\":\"Need to inspect first.\"}\n\n",
)
.expect("rewrite should succeed");
let output = String::from_utf8(output).expect("output should be utf8");
assert!(output.contains("\"object\":\"chat.completion.chunk\""));
assert!(output.contains("\"reasoning_content\":\"Need to inspect first.\""));
assert!(!output.contains("\"content\""));
assert!(!output.contains("data: [DONE]"));
}
#[test]
fn same_family_responses_passthrough_preserves_encrypted_content() {
// When provider and client are both OpenAI Responses family,
// the stream should pass through verbatim (only model name rewrite).
// This preserves encrypted_content, original item IDs, etc.
let report_context = json!({
"provider_api_format": "openai:responses",
"client_api_format": "openai:responses:compact",
"needs_conversion": true,
"model": "gpt-5.5-xhigh",
"mapped_model": "gpt-5.5",
});
let mut rewriter = maybe_build_ai_surface_stream_rewriter(Some(&report_context))
.expect("rewriter should exist");
let 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\",\"summary\":[],\"encrypted_content\":\"EWxvY2tlZENvbnRlbnQ=\"}}\n\n",
)
.expect("rewrite should succeed");
let output = String::from_utf8(output).expect("output should be utf8");
// Passthrough preserves the full payload structure
assert!(output.contains("event: response.output_item.added"));
assert!(output.contains("\"encrypted_content\":\"EWxvY2tlZENvbnRlbnQ=\""));
assert!(output.contains("\"id\":\"rs_abc\""));
assert!(output.contains("\"type\":\"reasoning\""));
}
#[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).
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());
}
#[test]
fn same_format_claude_passthrough_with_display_model() {
// Claude→Claude with needs_conversion=true should pass through
// (only model name rewrite), not parse→rebuild.
let report_context = json!({
"provider_api_format": "claude:messages",
"client_api_format": "claude:messages",
"needs_conversion": true,
"model": "claude-sonnet-4.5-high",
"mapped_model": "claude-sonnet-4.5",
});
let mut rewriter = maybe_build_ai_surface_stream_rewriter(Some(&report_context))
.expect("rewriter should exist");
let output = rewriter
.push_chunk(
b"event: content_block_delta\n\
data: {\"type\":\"content_block_delta\",\"index\":1,\"delta\":{\"type\":\"thinking_delta\",\"thinking\":\"Let me reason...\"}}\n\n",
)
.expect("rewrite should succeed");
let output = String::from_utf8(output).expect("output should be utf8");
// Passthrough preserves the exact wire format
assert!(output.contains("event: content_block_delta"));
assert!(output.contains("\"thinking\":\"Let me reason...\""));
assert!(output.contains("\"type\":\"thinking_delta\""));
}
#[test]
fn same_format_claude_without_display_model_passes_through_verbatim() {
// Claude→Claude without display model: no rewriter needed at all.
let report_context = json!({
"provider_api_format": "claude:messages",
"client_api_format": "claude:messages",
"needs_conversion": true,
});
assert!(maybe_build_ai_surface_stream_rewriter(Some(&report_context)).is_none());
}
#[test]
fn same_format_gemini_passthrough_with_display_model() {
// Gemini→Gemini with needs_conversion=true should pass through.
let report_context = json!({
"provider_api_format": "gemini:generate_content",
"client_api_format": "gemini:generate_content",
"needs_conversion": true,
"model": "gemini-2.5-pro-high",
"mapped_model": "gemini-2.5-pro",
});
let mut rewriter = maybe_build_ai_surface_stream_rewriter(Some(&report_context))
.expect("rewriter should exist");
let output = rewriter
.push_chunk(
b"data: {\"candidates\":[{\"content\":{\"parts\":[{\"text\":\"Hello\"}],\"role\":\"model\"}}],\"modelVersion\":\"gemini-2.5-pro\"}\n\n",
)
.expect("rewrite should succeed");
let output = String::from_utf8(output).expect("output should be utf8");
// Model version should be rewritten
assert!(output.contains("\"modelVersion\":\"gemini-2.5-pro-high\""));
assert!(!output.contains("\"modelVersion\":\"gemini-2.5-pro\""));
}
#[test]
fn same_format_gemini_without_display_model_passes_through_verbatim() {
// Gemini→Gemini without display model: no rewriter needed.
let report_context = json!({
"provider_api_format": "gemini:generate_content",
"client_api_format": "gemini:generate_content",
"needs_conversion": true,
});
assert!(maybe_build_ai_surface_stream_rewriter(Some(&report_context)).is_none());
}
#[test]
fn resolves_openai_image_mode_for_same_format_image_streams() {
let report_context = json!({

6
crates/aether-ai-formats/src/formats/shared/sync_products.rs
View File

@@ -2593,6 +2593,7 @@ pub fn aggregate_gemini_stream_sync_response(body: &[u8]) -> Option<Value> {
));
}
CanonicalStreamEvent::UnknownEvent(_) => {}
CanonicalStreamEvent::ReasoningSummaryDone => {}
CanonicalStreamEvent::Finish {
finish_reason: frame_finish_reason,
usage,
@@ -3630,8 +3631,9 @@ mod tests {
fn rejects_openai_responses_same_family_error_body_json() {
let report_context = json!({
"provider_api_format": "openai:responses",
"client_api_format": "openai:responses",
"needs_conversion": false,
"client_api_format": "openai:responses:compact",
"model": "gpt-5",
"mapped_model": "gpt-5",
});
let provider_body_json = json!({
"error": {

1
crates/aether-ai-formats/src/protocol/stream.rs
View File

@@ -35,6 +35,7 @@ pub enum CanonicalStreamEvent {
Start,
TextDelta(String),
ReasoningDelta(String),
ReasoningSummaryDone,
ReasoningSignature(String),
ContentPart(CanonicalContentPart),
ToolCallStart {