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

Support OpenAI Responses builtin tool stream items

Browse Source
This commit is contained in:
elky
2026-06-10 14:49:13 +08:00
parent 8edcbdcb29
commit ea76f7bb0b
3 changed files with 672 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aether-ai-formats/src/formats/openai/chat/stream.rs
+447 -106
View File
@@ -683,6 +683,136 @@ impl OpenAIResponsesProviderState {
self.emit_ready_tool_call(report_context, out, index);
}
fn emit_custom_tool_call_item(
&mut self,
report_context: &Value,
out: &mut Vec<CanonicalStreamFrame>,
item: &Map<String, Value>,
output_index: Option<usize>,
) {
if item.get("type").and_then(Value::as_str) != Some("custom_tool_call") {
return;
}
let name = item
.get("name")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("custom_tool")
.to_string();
let arguments = tool_arguments_from_maybe_json_string(
item.get("input").or_else(|| item.get("arguments")),
"input",
);
self.emit_generic_tool_call_item(report_context, out, item, output_index, name, arguments);
}
fn emit_shell_tool_call_item(
&mut self,
report_context: &Value,
out: &mut Vec<CanonicalStreamFrame>,
item: &Map<String, Value>,
output_index: Option<usize>,
) {
let item_type = item.get("type").and_then(Value::as_str).unwrap_or_default();
let name = match item_type {
"local_shell_call" => "local_shell",
"shell_call" => "shell",
_ => return,
};
let arguments = tool_arguments_from_named_fields(
item,
&[
"action",
"environment",
"status",
"created_by",
"max_output_length",
],
);
self.emit_generic_tool_call_item(
report_context,
out,
item,
output_index,
name.to_string(),
arguments,
);
}
fn emit_apply_patch_tool_call_item(
&mut self,
report_context: &Value,
out: &mut Vec<CanonicalStreamFrame>,
item: &Map<String, Value>,
output_index: Option<usize>,
) {
if item.get("type").and_then(Value::as_str) != Some("apply_patch_call") {
return;
}
let arguments = tool_arguments_from_named_fields(item, &["operation", "status"]);
self.emit_generic_tool_call_item(
report_context,
out,
item,
output_index,
"apply_patch".to_string(),
arguments,
);
}
fn emit_computer_tool_call_item(
&mut self,
report_context: &Value,
out: &mut Vec<CanonicalStreamFrame>,
item: &Map<String, Value>,
output_index: Option<usize>,
) {
if item.get("type").and_then(Value::as_str) != Some("computer_call") {
return;
}
let arguments = tool_arguments_from_named_fields(
item,
&["action", "actions", "pending_safety_checks", "status"],
);
self.emit_generic_tool_call_item(
report_context,
out,
item,
output_index,
"computer".to_string(),
arguments,
);
}
fn emit_generic_tool_call_item(
&mut self,
report_context: &Value,
out: &mut Vec<CanonicalStreamFrame>,
item: &Map<String, Value>,
output_index: Option<usize>,
name: String,
arguments: String,
) {
self.ensure_started(report_context, out);
let key = item
.get("call_id")
.or_else(|| item.get("id"))
.and_then(Value::as_str)
.map(ToOwned::to_owned);
let index = self.tool_index_for_key(key, output_index);
let state = self.tool_calls.entry(index).or_default();
state.call_id = item
.get("call_id")
.or_else(|| item.get("id"))
.and_then(Value::as_str)
.unwrap_or(state.call_id.as_str())
.to_string();
state.name = name;
Self::merge_tool_call_arguments(state, &arguments);
self.emit_ready_tool_call(report_context, out, index);
}
fn emit_missing_tool_result(
&mut self,
report_context: &Value,
@@ -756,6 +886,54 @@ impl OpenAIResponsesProviderState {
self.emit_missing_tool_result(report_context, out, index, tool_use_id, name, &content);
}
fn emit_generic_tool_result_item(
&mut self,
report_context: &Value,
out: &mut Vec<CanonicalStreamFrame>,
item: &Map<String, Value>,
output_index: Option<usize>,
) {
let item_type = item.get("type").and_then(Value::as_str).unwrap_or_default();
let is_supported_result = matches!(
item_type,
"custom_tool_call_output"
| "local_shell_call_output"
| "shell_call_output"
| "apply_patch_call_output"
| "computer_call_output"
);
if !is_supported_result {
return;
}
let tool_use_id = item
.get("call_id")
.or_else(|| item.get("tool_call_id"))
.or_else(|| item.get("id"))
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.unwrap_or("call_auto_0")
.to_string();
let index =
self.tool_index_for_key(Some(format!("{item_type}:{tool_use_id}")), output_index);
let content = openai_tool_result_content_from_value(
item.get("output")
.or_else(|| item.get("content"))
.or_else(|| item.get("delta")),
);
let name = match item_type {
"local_shell_call_output" => Some("local_shell".to_string()),
"shell_call_output" => Some("shell".to_string()),
"apply_patch_call_output" => Some("apply_patch".to_string()),
"computer_call_output" => Some("computer".to_string()),
_ => item
.get("name")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.map(ToOwned::to_owned),
};
self.emit_missing_tool_result(report_context, out, index, tool_use_id, name, &content);
}
fn emit_message_item(
&mut self,
report_context: &Value,
@@ -870,6 +1048,79 @@ impl OpenAIResponsesProviderState {
});
}
fn emit_output_item(
&mut self,
report_context: &Value,
out: &mut Vec<CanonicalStreamFrame>,
item: &Map<String, Value>,
output_index: Option<usize>,
final_item: bool,
) {
match item.get("type").and_then(Value::as_str).unwrap_or_default() {
"function_call" => self.emit_tool_call_item(report_context, out, item, output_index),
"function_call_output" => {
self.emit_tool_result_item(report_context, out, item, output_index);
}
"custom_tool_call" => {
self.emit_custom_tool_call_item(report_context, out, item, output_index);
}
"local_shell_call" | "shell_call" => {
self.emit_shell_tool_call_item(report_context, out, item, output_index);
}
"apply_patch_call" => {
self.emit_apply_patch_tool_call_item(report_context, out, item, output_index);
}
"computer_call" => {
self.emit_computer_tool_call_item(report_context, out, item, output_index);
}
"custom_tool_call_output"
| "local_shell_call_output"
| "shell_call_output"
| "apply_patch_call_output"
| "computer_call_output" => {
self.emit_generic_tool_result_item(report_context, out, item, output_index);
}
"message" => self.emit_message_item(report_context, out, item, output_index),
"reasoning" if final_item => self.emit_reasoning_item(report_context, out, item),
"reasoning" => self.ensure_started(report_context, out),
"image_generation_call" => {
self.emit_image_generation_item(
report_context,
out,
item,
output_index,
final_item,
);
}
"web_search_call" | "file_search_call" | "code_interpreter_call" | "mcp_call" => {
if !final_item {
self.ensure_started(report_context, out);
}
}
_ => out.push(self.unknown_frame(report_context, Value::Object(item.clone()))),
}
}
fn emit_response_output_items(
&mut self,
report_context: &Value,
out: &mut Vec<CanonicalStreamFrame>,
response: &Map<String, Value>,
) {
for (output_index, raw_item) in response
.get("output")
.and_then(Value::as_array)
.into_iter()
.flatten()
.enumerate()
{
let Some(item) = raw_item.as_object() else {
continue;
};
self.emit_output_item(report_context, out, item, Some(output_index), true);
}
}
pub fn push_line(
&mut self,
report_context: &Value,
@@ -988,6 +1239,27 @@ impl OpenAIResponsesProviderState {
self.emit_missing_text(report_context, &mut out, key, refusal);
}
}
"response.audio.transcript.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.audio.transcript.done" => {
let transcript = value
.get("transcript")
.and_then(Value::as_str)
.or_else(|| value.get("text").and_then(Value::as_str))
.unwrap_or_default();
if !transcript.is_empty() {
let key = Self::text_part_key_from_event(&value);
self.emit_missing_text(report_context, &mut out, key, transcript);
}
}
"response.reasoning_text.delta" | "response.reasoning_summary_text.delta" => {
let piece = value
.get("delta")
@@ -1054,32 +1326,70 @@ impl OpenAIResponsesProviderState {
.get("output_index")
.and_then(Value::as_u64)
.map(|value| value as usize);
match item.get("type").and_then(Value::as_str).unwrap_or_default() {
"function_call" => {
self.emit_tool_call_item(report_context, &mut out, item, output_index);
}
"function_call_output" => {
self.emit_tool_result_item(report_context, &mut out, item, output_index);
}
"message" => {
self.emit_message_item(report_context, &mut out, item, output_index);
}
"reasoning" => {
self.ensure_started(report_context, &mut out);
}
"image_generation_call" => {
self.emit_image_generation_item(
report_context,
&mut out,
item,
output_index,
false,
);
}
_ => {
out.push(self.unknown_frame(report_context, Value::Object(item.clone())));
}
self.emit_output_item(report_context, &mut out, item, output_index, false);
}
"response.custom_tool_call_input.delta" => {
let delta = value
.get("delta")
.and_then(Value::as_str)
.unwrap_or_default();
if delta.is_empty() {
return Ok(out);
}
self.ensure_started(report_context, &mut out);
let key = value
.get("item_id")
.or_else(|| value.get("call_id"))
.or_else(|| value.get("id"))
.and_then(Value::as_str)
.map(ToOwned::to_owned);
let output_index = value
.get("output_index")
.and_then(Value::as_u64)
.map(|value| value as usize);
let index = self.tool_index_for_key(key, output_index);
let state = self.tool_calls.entry(index).or_default();
if state.name.is_empty() {
state.name = value
.get("name")
.and_then(Value::as_str)
.unwrap_or("custom_tool")
.to_string();
}
state.arguments.push_str(delta);
self.emit_ready_tool_call(report_context, &mut out, index);
}
"response.custom_tool_call_input.done" => {
let input = value
.get("input")
.and_then(Value::as_str)
.unwrap_or_default();
self.ensure_started(report_context, &mut out);
let key = value
.get("item_id")
.or_else(|| value.get("call_id"))
.or_else(|| value.get("id"))
.and_then(Value::as_str)
.map(ToOwned::to_owned);
let output_index = value
.get("output_index")
.and_then(Value::as_u64)
.map(|value| value as usize);
let index = self.tool_index_for_key(key, output_index);
let state = self.tool_calls.entry(index).or_default();
if state.name.is_empty() {
state.name = value
.get("name")
.and_then(Value::as_str)
.unwrap_or("custom_tool")
.to_string();
}
let arguments = tool_arguments_from_maybe_json_string(
Some(&Value::String(input.to_string())),
"input",
);
Self::merge_tool_call_arguments(state, &arguments);
self.emit_ready_tool_call(report_context, &mut out, index);
}
"response.function_call_arguments.delta" => {
let delta = value
@@ -1226,32 +1536,28 @@ impl OpenAIResponsesProviderState {
.get("output_index")
.and_then(Value::as_u64)
.map(|value| value as usize);
match item.get("type").and_then(Value::as_str).unwrap_or_default() {
"function_call" => {
self.emit_tool_call_item(report_context, &mut out, item, output_index);
}
"function_call_output" => {
self.emit_tool_result_item(report_context, &mut out, item, output_index);
}
"message" => {
self.emit_message_item(report_context, &mut out, item, output_index);
}
"reasoning" => {
self.emit_reasoning_item(report_context, &mut out, item);
}
"image_generation_call" => {
self.emit_image_generation_item(
report_context,
&mut out,
item,
output_index,
true,
);
}
_ => {
out.push(self.unknown_frame(report_context, Value::Object(item.clone())));
}
}
self.emit_output_item(report_context, &mut out, item, output_index, true);
}
"response.incomplete" => {
let Some(response) = value.get("response").and_then(Value::as_object) else {
return Ok(out);
};
self.ensure_started(report_context, &mut out);
let (id, model) = self.identity(report_context);
self.emit_response_output_items(report_context, &mut out, response);
out.push(CanonicalStreamFrame {
id,
model,
event: CanonicalStreamEvent::Finish {
finish_reason: Some(openai_responses_incomplete_finish_reason(&value)),
usage: canonical_usage_from_openai_usage(response.get("usage")),
},
});
self.finished = true;
}
event_type if openai_responses_stream_event_is_known_noop(event_type) => {
self.ensure_started(report_context, &mut out);
}
event_type if openai_stream_payload_is_terminal_error(&value) => {
self.finished = true;
@@ -1276,61 +1582,7 @@ impl OpenAIResponsesProviderState {
};
self.ensure_started(report_context, &mut out);
let (id, model) = self.identity(report_context);
for (output_index, raw_item) in response
.get("output")
.and_then(Value::as_array)
.into_iter()
.flatten()
.enumerate()
{
let Some(item) = raw_item.as_object() else {
continue;
};
match item.get("type").and_then(Value::as_str).unwrap_or_default() {
"message" => {
self.emit_message_item(
report_context,
&mut out,
item,
Some(output_index),
);
}
"function_call" => {
self.emit_tool_call_item(
report_context,
&mut out,
item,
Some(output_index),
);
}
"function_call_output" => {
self.emit_tool_result_item(
report_context,
&mut out,
item,
Some(output_index),
);
}
"reasoning" => {
self.emit_reasoning_item(report_context, &mut out, item);
}
"image_generation_call" => {
self.emit_image_generation_item(
report_context,
&mut out,
item,
Some(output_index),
true,
);
}
_ => {
out.push(
self.unknown_frame(report_context, Value::Object(item.clone())),
);
}
}
}
self.emit_response_output_items(report_context, &mut out, response);
let finish_reason = if self.tool_calls.is_empty() {
Some("stop".to_string())
@@ -2791,6 +3043,95 @@ fn openai_tool_result_content_from_value(value: Option<&Value>) -> String {
}
}
fn tool_arguments_from_maybe_json_string(value: Option<&Value>, fallback_key: &str) -> String {
match value {
Some(Value::String(text)) => {
let trimmed = text.trim();
if trimmed.is_empty() {
return String::new();
}
match serde_json::from_str::<Value>(trimmed) {
Ok(Value::Object(_)) => trimmed.to_string(),
Ok(parsed) => single_field_tool_arguments(fallback_key, parsed),
Err(_) => single_field_tool_arguments(fallback_key, Value::String(text.clone())),
}
}
Some(value @ Value::Object(_)) => value.to_string(),
Some(Value::Null) | None => String::new(),
Some(value) => single_field_tool_arguments(fallback_key, value.clone()),
}
}
fn single_field_tool_arguments(key: &str, value: Value) -> String {
let mut arguments = Map::new();
arguments.insert(key.to_string(), value);
Value::Object(arguments).to_string()
}
fn tool_arguments_from_named_fields(item: &Map<String, Value>, field_names: &[&str]) -> String {
let mut arguments = Map::new();
for field_name in field_names {
if let Some(value) = item.get(*field_name) {
arguments.insert((*field_name).to_string(), value.clone());
}
}
if arguments.is_empty() {
String::new()
} else {
Value::Object(arguments).to_string()
}
}
fn openai_responses_stream_event_is_known_noop(event_type: &str) -> bool {
matches!(
event_type,
"response.queued"
| "response.output_text.annotation.added"
| "response.audio.delta"
| "response.audio.done"
| "response.code_interpreter_call.in_progress"
| "response.code_interpreter_call.interpreting"
| "response.code_interpreter_call.completed"
| "response.code_interpreter_call_code.delta"
| "response.code_interpreter_call_code.done"
| "response.file_search_call.in_progress"
| "response.file_search_call.searching"
| "response.file_search_call.completed"
| "response.image_generation_call.in_progress"
| "response.image_generation_call.generating"
| "response.image_generation_call.partial_image"
| "response.image_generation_call.completed"
| "response.mcp_call.in_progress"
| "response.mcp_call.completed"
| "response.mcp_call.failed"
| "response.mcp_call_arguments.delta"
| "response.mcp_call_arguments.done"
| "response.mcp_list_tools.in_progress"
| "response.mcp_list_tools.completed"
| "response.mcp_list_tools.failed"
| "response.web_search_call.in_progress"
| "response.web_search_call.searching"
| "response.web_search_call.completed"
)
}
fn openai_responses_incomplete_finish_reason(payload: &Value) -> String {
let reason = payload
.get("response")
.and_then(Value::as_object)
.and_then(|response| response.get("incomplete_details"))
.and_then(Value::as_object)
.and_then(|details| details.get("reason"))
.and_then(Value::as_str)
.unwrap_or_default();
match reason {
"content_filter" => "content_filter",
"tool_calls" | "function_call" => "tool_calls",
_ => "length",
}
.to_string()
}
#[cfg(test)]
mod tests {
use super::*;
crates/aether-ai-formats/src/formats/shared/stream_core/common.rs
+12 -11
View File
@@ -110,26 +110,27 @@ pub fn canonical_usage_from_openai_usage(value: Option<&Value>) -> Option<Canoni
}
pub fn openai_stream_payload_is_terminal_error(payload: &Value) -> bool {
let response = payload.get("response").and_then(Value::as_object);
if payload.get("error").is_some()
|| response
.and_then(|response| response.get("error"))
.is_some()
{
return true;
}
let event_type = payload
.get("type")
.and_then(Value::as_str)
.unwrap_or_default();
if payload.get("error").is_some() {
return true;
}
if matches!(
event_type,
"error" | "response.failed" | "response.incomplete"
) {
if matches!(event_type, "error" | "response.failed") {
return true;
}
payload
.get("response")
.and_then(Value::as_object)
response
.and_then(|response| response.get("status"))
.and_then(Value::as_str)
.is_some_and(|status| matches!(status, "failed" | "incomplete"))
.is_some_and(|status| status == "failed")
}
pub fn openai_stream_terminal_error_body(payload: &Value) -> Option<Value> {
crates/aether-ai-formats/src/formats/shared/stream_core/format_matrix.rs
+213
View File
@@ -983,6 +983,180 @@ mod tests {
}
}
#[test]
fn transforms_openai_responses_known_sidecar_events_without_unsupported_errors() {
let report_context = report_context("openai:responses", "claude:messages");
let mut matrix = StreamingStandardFormatMatrix::default();
let mut output = Vec::new();
for line in [
data_line(json!({
"type": "response.created",
"response": {
"id": "resp_sidecar_123",
"model": "gpt-5.4",
"status": "in_progress",
"output": [],
},
})),
data_line(json!({
"type": "response.output_item.added",
"response_id": "resp_sidecar_123",
"output_index": 0,
"item": {
"type": "web_search_call",
"id": "ws_123",
"status": "in_progress",
"action": {"type": "search", "query": "aether format conversion"},
},
})),
data_line(json!({
"type": "response.web_search_call.searching",
"item_id": "ws_123",
"output_index": 0,
})),
data_line(json!({
"type": "response.output_text.annotation.added",
"response_id": "resp_sidecar_123",
"output_index": 1,
"content_index": 0,
"annotation_index": 0,
"annotation": {"type": "url_citation", "url": "https://example.invalid"},
})),
data_line(json!({
"type": "response.output_text.delta",
"response_id": "resp_sidecar_123",
"output_index": 1,
"content_index": 0,
"delta": "sidecar ok",
})),
data_line(json!({
"type": "response.completed",
"response": {
"id": "resp_sidecar_123",
"object": "response",
"model": "gpt-5.4",
"status": "completed",
"output": [],
"usage": {
"input_tokens": 1,
"output_tokens": 2,
"total_tokens": 3,
},
},
})),
] {
output.extend(
matrix
.transform_line(&report_context, line)
.expect("known responses sidecar event should convert or be ignored"),
);
}
let sse = String::from_utf8(output).expect("sse should be utf8");
assert!(!sse.contains("unsupported_stream_event"), "{sse}");
assert!(!sse.contains("Unsupported provider stream event"), "{sse}");
assert!(sse.contains("sidecar ok"), "{sse}");
assert!(sse.contains("event: message_stop"), "{sse}");
}
#[test]
fn transforms_openai_responses_incomplete_max_tokens_as_normal_finish() {
let report_context = report_context("openai:responses", "claude:messages");
let mut matrix = StreamingStandardFormatMatrix::default();
let output = matrix
.transform_line(
&report_context,
data_line(json!({
"type": "response.incomplete",
"response": {
"id": "resp_incomplete_123",
"object": "response",
"model": "gpt-5.4",
"status": "incomplete",
"incomplete_details": {
"reason": "max_output_tokens",
},
"output": [{
"type": "message",
"id": "msg_incomplete_123",
"role": "assistant",
"status": "incomplete",
"content": [{
"type": "output_text",
"text": "partial answer",
}],
}],
"usage": {
"input_tokens": 10,
"output_tokens": 20,
"total_tokens": 30,
},
},
})),
)
.expect("incomplete max token response should convert as length finish");
let sse = String::from_utf8(output).expect("sse should be utf8");
assert!(!sse.contains("Response incomplete"), "{sse}");
assert!(!sse.contains("unsupported_stream_event"), "{sse}");
assert!(sse.contains("partial answer"), "{sse}");
assert!(sse.contains("\"stop_reason\":\"max_tokens\""), "{sse}");
assert!(matrix
.finish(&report_context)
.expect("finish should be terminated")
.is_empty());
}
#[test]
fn transforms_openai_responses_local_shell_call_to_claude_tool_use() {
let report_context = report_context("openai:responses", "claude:messages");
let mut matrix = StreamingStandardFormatMatrix::default();
let mut output = Vec::new();
for line in [
data_line(json!({
"type": "response.output_item.done",
"response_id": "resp_shell_123",
"output_index": 0,
"item": {
"type": "local_shell_call",
"id": "lsc_123",
"call_id": "call_shell_123",
"status": "completed",
"action": {
"type": "exec",
"command": ["pwd"],
"env": {},
},
},
})),
data_line(json!({
"type": "response.completed",
"response": {
"id": "resp_shell_123",
"object": "response",
"model": "gpt-5.4",
"status": "completed",
"output": [],
},
})),
] {
output.extend(
matrix
.transform_line(&report_context, line)
.expect("local shell call should convert to a generic tool use"),
);
}
let sse = String::from_utf8(output).expect("sse should be utf8");
assert!(!sse.contains("unsupported_stream_event"), "{sse}");
assert!(sse.contains("\"type\":\"tool_use\""), "{sse}");
assert!(sse.contains("\"name\":\"local_shell\""), "{sse}");
assert!(sse.contains("\\\"command\\\":[\\\"pwd\\\"]"), "{sse}");
assert!(sse.contains("\"stop_reason\":\"tool_use\""), "{sse}");
}
#[test]
fn transforms_unknown_stream_finish_reasons_to_visible_client_errors() {
let cases = [
@@ -1452,6 +1626,45 @@ mod tests {
assert_eq!(summary.unknown_event_count, 1);
}
#[test]
fn terminal_observer_marks_openai_responses_incomplete_as_length_finish() {
let mut report_context = report_context("openai:chat", "openai:responses");
report_context["provider_stream_event_api_format"] = json!("openai:responses");
let mut observer = StreamingStandardTerminalObserver::default();
observer
.push_line(
&report_context,
data_line(json!({
"type": "response.incomplete",
"response": {
"id": "resp_incomplete_123",
"model": "gpt-5.4",
"status": "incomplete",
"incomplete_details": {
"reason": "max_output_tokens",
},
"output": [],
"usage": {
"input_tokens": 10,
"output_tokens": 20,
"total_tokens": 30,
},
},
})),
)
.expect("incomplete event should be observed as terminal finish");
let summary = observer
.latest_summary()
.cloned()
.expect("summary should exist");
assert!(summary.observed_finish);
assert_eq!(summary.finish_reason.as_deref(), Some("length"));
assert_eq!(summary.parser_error, None);
assert_eq!(summary.unknown_event_count, 0);
}
#[test]
fn terminal_observer_tracks_openai_image_stream_usage() {
let mut report_context = report_context("openai:image", "openai:chat");