openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-13 11:47:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

tests

Browse Source
This commit is contained in:
Ahmed Ibrahim
2025-10-16 17:18:01 -07:00
parent 610cd53595
commit fdf9b7566f
2 changed files with 142 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

265
codex-rs/core/src/codex.rs
View File

@@ -1057,6 +1057,12 @@ impl Session {
}
}
/// Track a freshly emitted `ResponseItem` for the active turn.
///
/// When the model stream produces an item (e.g. a tool call) we register it
/// immediately so abort paths can still flush it into history even if the
/// matching tool future has not completed yet. The returned index lets the
/// caller later patch the slot when the response arrives.
pub async fn append_processed_item(
&self,
sub_id: &str,
@@ -1072,6 +1078,14 @@ impl Session {
}
}
/// Patch a previously queued `ProcessedResponseItem` with the tool
/// response.
///
/// Tool futures resolve outside of the stream loop. Once a result is
/// available we update the placeholder slot created by
/// [`append_processed_item`] so the turn state reflects the final
/// `ResponseInputItem`. If the turn has already ended (e.g. abort), we skip
/// the update—the abort handler will synthesize the appropriate output.
pub async fn update_processed_item_response(
&self,
sub_id: &str,
@@ -2023,6 +2037,132 @@ pub(crate) struct ProcessedResponseItem {
pub(crate) response: Option<ResponseInputItem>,
}
struct ProcessedItemsSummary {
history_items: Vec<ResponseItem>,
responses: Vec<ResponseInputItem>,
}
fn summarize_processed_items(processed_items: Vec<ProcessedResponseItem>) -> ProcessedItemsSummary {
let mut summary = ProcessedItemsSummary {
history_items: Vec::new(),
responses: Vec::new(),
};
for processed_response_item in processed_items {
let ProcessedResponseItem { item, response } = processed_response_item;
match (&item, &response) {
(ResponseItem::Message { role, .. }, None) if role == "assistant" => {
summary.history_items.push(item.clone());
}
(
ResponseItem::LocalShellCall { .. },
Some(ResponseInputItem::FunctionCallOutput { call_id, output }),
) => {
summary.history_items.push(item.clone());
summary
.history_items
.push(ResponseItem::FunctionCallOutput {
call_id: call_id.clone(),
output: output.clone(),
});
}
(
ResponseItem::FunctionCall { .. },
Some(ResponseInputItem::FunctionCallOutput { call_id, output }),
) => {
summary.history_items.push(item.clone());
summary
.history_items
.push(ResponseItem::FunctionCallOutput {
call_id: call_id.clone(),
output: output.clone(),
});
}
(
ResponseItem::CustomToolCall { .. },
Some(ResponseInputItem::CustomToolCallOutput { call_id, output }),
) => {
summary.history_items.push(item.clone());
summary
.history_items
.push(ResponseItem::CustomToolCallOutput {
call_id: call_id.clone(),
output: output.clone(),
});
}
(
ResponseItem::FunctionCall { .. },
Some(ResponseInputItem::McpToolCallOutput { call_id, result }),
) => {
summary.history_items.push(item.clone());
let output = match result {
Ok(call_tool_result) => {
convert_call_tool_result_to_function_call_output_payload(call_tool_result)
}
Err(err) => FunctionCallOutputPayload {
content: err.clone(),
success: Some(false),
},
};
summary
.history_items
.push(ResponseItem::FunctionCallOutput {
call_id: call_id.clone(),
output,
});
}
(
ResponseItem::LocalShellCall {
call_id: Some(call_id),
..
},
None,
) => {
summary.history_items.push(item.clone());
summary
.history_items
.push(make_aborted_function_call_output(call_id.clone()));
}
(ResponseItem::FunctionCall { call_id, .. }, None) => {
summary.history_items.push(item.clone());
summary
.history_items
.push(make_aborted_function_call_output(call_id.clone()));
}
(ResponseItem::CustomToolCall { call_id, .. }, None) => {
summary.history_items.push(item.clone());
summary
.history_items
.push(make_aborted_custom_tool_call_output(call_id.clone()));
}
(
ResponseItem::Reasoning {
id,
summary: reasoning_summary,
content,
encrypted_content,
},
None,
) => {
summary.history_items.push(ResponseItem::Reasoning {
id: id.clone(),
summary: reasoning_summary.clone(),
content: content.clone(),
encrypted_content: encrypted_content.clone(),
});
}
_ => {
warn!("Unexpected response item: {item:?} with response: {response:?}");
}
};
if let Some(response) = response {
summary.responses.push(response);
}
}
summary
}
#[derive(Debug)]
struct TurnRunResult {
processed_items: Vec<ProcessedResponseItem>,
@@ -3488,128 +3628,3 @@ mod tests {
assert!(exec_output.output.contains("hi"));
}
}
struct ProcessedItemsSummary {
history_items: Vec<ResponseItem>,
responses: Vec<ResponseInputItem>,
}
fn summarize_processed_items(processed_items: Vec<ProcessedResponseItem>) -> ProcessedItemsSummary {
let mut summary = ProcessedItemsSummary {
history_items: Vec::new(),
responses: Vec::new(),
};
for processed_response_item in processed_items {
let ProcessedResponseItem { item, response } = processed_response_item;
match (&item, &response) {
(ResponseItem::Message { role, .. }, None) if role == "assistant" => {
summary.history_items.push(item.clone());
}
(
ResponseItem::LocalShellCall { .. },
Some(ResponseInputItem::FunctionCallOutput { call_id, output }),
) => {
summary.history_items.push(item.clone());
summary
.history_items
.push(ResponseItem::FunctionCallOutput {
call_id: call_id.clone(),
output: output.clone(),
});
}
(
ResponseItem::FunctionCall { .. },
Some(ResponseInputItem::FunctionCallOutput { call_id, output }),
) => {
summary.history_items.push(item.clone());
summary
.history_items
.push(ResponseItem::FunctionCallOutput {
call_id: call_id.clone(),
output: output.clone(),
});
}
(
ResponseItem::CustomToolCall { .. },
Some(ResponseInputItem::CustomToolCallOutput { call_id, output }),
) => {
summary.history_items.push(item.clone());
summary
.history_items
.push(ResponseItem::CustomToolCallOutput {
call_id: call_id.clone(),
output: output.clone(),
});
}
(
ResponseItem::FunctionCall { .. },
Some(ResponseInputItem::McpToolCallOutput { call_id, result }),
) => {
summary.history_items.push(item.clone());
let output = match result {
Ok(call_tool_result) => {
convert_call_tool_result_to_function_call_output_payload(call_tool_result)
}
Err(err) => FunctionCallOutputPayload {
content: err.clone(),
success: Some(false),
},
};
summary
.history_items
.push(ResponseItem::FunctionCallOutput {
call_id: call_id.clone(),
output,
});
}
(
ResponseItem::LocalShellCall {
call_id: Some(call_id),
..
},
None,
) => {
summary.history_items.push(item.clone());
summary
.history_items
.push(make_aborted_function_call_output(call_id.clone()));
}
(ResponseItem::FunctionCall { call_id, .. }, None) => {
summary.history_items.push(item.clone());
summary
.history_items
.push(make_aborted_function_call_output(call_id.clone()));
}
(ResponseItem::CustomToolCall { call_id, .. }, None) => {
summary.history_items.push(item.clone());
summary
.history_items
.push(make_aborted_custom_tool_call_output(call_id.clone()));
}
(
ResponseItem::Reasoning {
id,
summary: reasoning_summary,
content,
encrypted_content,
},
None,
) => {
summary.history_items.push(ResponseItem::Reasoning {
id: id.clone(),
summary: reasoning_summary.clone(),
content: content.clone(),
encrypted_content: encrypted_content.clone(),
});
}
_ => {
warn!("Unexpected response item: {item:?} with response: {response:?}");
}
};
if let Some(response) = response {
summary.responses.push(response);
}
}
summary
}

7
codex-rs/core/tests/suite/abort_tasks.rs
View File

@@ -168,11 +168,8 @@ async fn interrupt_tool_records_history_entries() {
if next.get("type").and_then(|v| v.as_str()) == Some("function_call_output")
&& next.get("call_id").and_then(|v| v.as_str()) == Some(call_id)
{
let content_matches = next
.get("output")
.and_then(|o| o.as_str())
.map(|s| s == "aborted")
.unwrap_or(false);
let content_matches =
next.get("output").and_then(serde_json::Value::as_str) == Some("aborted");
if content_matches {
abort_seen = true;
break;