mirror of
https://github.com/pchuan98/codex.git
synced 2026-07-01 00:31:56 +08:00
feat(core): store turn_id on ResponseItem metadata (#28360)
## Description This PR is a followup to https://github.com/openai/codex/pull/28355 and starts assigning `internal_chat_message_metadata_passthrough.turn_id` to durable Responses API items created during a turn. The goal is that those items keep the `turn_id` that introduced them when Codex resends stateless HTTP context, reconstructs history for resume/fork paths, or reuses websocket response state. ## What changed - Set `internal_chat_message_metadata_passthrough.turn_id` when missing as response items enter durable history, initial/replacement history, inter-agent communication history, and local compaction summaries. - Preserve existing item turn IDs instead of overwriting them during persistence, resume reconstruction, compaction, forked history, and websocket incremental reuse. - Keep `compaction_trigger` fieldless because it is a request control, not a durable response item. - Update focused history/request assertions and fixtures for stateless requests, websocket incrementals, compaction, thread injection, prompt debug, and related CI coverage.
This commit is contained in:
committed by
GitHub
Unverified
parent
7153affa0f
commit
4a82ecc3c9
@@ -887,6 +887,15 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
|
||||
parent_thread
|
||||
.inject_user_message_without_turn("parent seed context".to_string())
|
||||
.await;
|
||||
let expected_parent_seed = parent_thread
|
||||
.codex
|
||||
.session
|
||||
.clone_history()
|
||||
.await
|
||||
.raw_items()
|
||||
.first()
|
||||
.cloned()
|
||||
.expect("parent seed should be recorded");
|
||||
let turn_context = parent_thread.codex.session.new_default_turn().await;
|
||||
let parent_spawn_call_id = "spawn-call-history".to_string();
|
||||
let trigger_message = InterAgentCommunication::new(
|
||||
@@ -988,17 +997,12 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
|
||||
);
|
||||
assert_ne!(child_thread_id, parent_thread_id);
|
||||
let history = child_thread.codex.session.clone_history().await;
|
||||
let mut expected_final_answer =
|
||||
assistant_message("parent final answer", Some(MessagePhase::FinalAnswer));
|
||||
expected_final_answer.set_turn_id_if_missing(&turn_context.sub_id);
|
||||
let expected_history = [
|
||||
ResponseItem::Message {
|
||||
id: None,
|
||||
role: "user".to_string(),
|
||||
content: vec![ContentItem::InputText {
|
||||
text: "parent seed context".to_string(),
|
||||
}],
|
||||
phase: None,
|
||||
internal_chat_message_metadata_passthrough: None,
|
||||
},
|
||||
assistant_message("parent final answer", Some(MessagePhase::FinalAnswer)),
|
||||
expected_parent_seed,
|
||||
expected_final_answer,
|
||||
ResponseItem::Message {
|
||||
id: None,
|
||||
role: "developer".to_string(),
|
||||
|
||||
@@ -302,6 +302,11 @@ async fn run_compact_task_inner_impl(
|
||||
let user_messages = collect_user_messages(history_items);
|
||||
|
||||
let mut new_history = build_compacted_history(Vec::new(), &user_messages, &summary_text);
|
||||
if let Some(summary_item) = new_history.last_mut() {
|
||||
// This replacement history skips `record_conversation_items`; only the appended summary
|
||||
// belongs to this compaction turn.
|
||||
summary_item.set_turn_id_if_missing(&turn_context.sub_id);
|
||||
}
|
||||
let (window_number, window_ids) = sess.advance_auto_compact_window().await;
|
||||
|
||||
if matches!(
|
||||
|
||||
@@ -234,9 +234,7 @@ async fn run_remote_compact_task_inner_impl(
|
||||
)
|
||||
.await?;
|
||||
let mut input = prompt_input.clone();
|
||||
input.push(ResponseItem::CompactionTrigger {
|
||||
internal_chat_message_metadata_passthrough: None,
|
||||
});
|
||||
input.push(ResponseItem::CompactionTrigger {});
|
||||
let prompt = Prompt {
|
||||
input,
|
||||
tools: tool_router.model_visible_specs(),
|
||||
|
||||
@@ -2693,6 +2693,10 @@ impl Session {
|
||||
) -> Cow<'a, [ResponseItem]> {
|
||||
let mut items = Cow::Borrowed(items);
|
||||
prepare_response_items(items.to_mut());
|
||||
// Most response items get their passthrough turn ID at the durable history boundary.
|
||||
for item in items.to_mut() {
|
||||
item.set_turn_id_if_missing(&turn_context.sub_id);
|
||||
}
|
||||
if turn_context.config.features.enabled(Feature::ItemIds) {
|
||||
Self::assign_missing_response_item_ids(items)
|
||||
} else {
|
||||
@@ -2784,8 +2788,9 @@ impl Session {
|
||||
pub(crate) async fn record_inter_agent_communication(
|
||||
&self,
|
||||
turn_context: &TurnContext,
|
||||
communication: InterAgentCommunication,
|
||||
mut communication: InterAgentCommunication,
|
||||
) {
|
||||
communication.set_turn_id_if_missing(&turn_context.sub_id);
|
||||
let response_item = communication.to_model_input_item();
|
||||
let items = self.prepare_conversation_items_for_history(
|
||||
turn_context,
|
||||
@@ -3355,6 +3360,10 @@ impl Session {
|
||||
{
|
||||
items.push(guardian_developer_message);
|
||||
}
|
||||
// New context windows and compaction install these items directly into replacement history.
|
||||
for item in &mut items {
|
||||
item.set_turn_id_if_missing(&turn_context.sub_id);
|
||||
}
|
||||
items
|
||||
}
|
||||
|
||||
|
||||
@@ -152,6 +152,7 @@ use core_test_support::responses::ev_response_created;
|
||||
use core_test_support::responses::mount_sse_once;
|
||||
use core_test_support::responses::sse;
|
||||
use core_test_support::responses::start_mock_server;
|
||||
use core_test_support::responses::strip_metadata_from_items;
|
||||
use core_test_support::test_codex::local;
|
||||
use core_test_support::test_codex::test_codex;
|
||||
use core_test_support::test_path_buf;
|
||||
@@ -1692,6 +1693,67 @@ async fn record_initial_history_reconstructs_resumed_transcript() {
|
||||
assert_eq!(expected, history.raw_items());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn record_conversation_items_stamps_missing_turn_id_and_preserves_existing_turn_id() {
|
||||
let (session, turn_context) = make_session_and_context().await;
|
||||
let fresh_item = user_message("fresh");
|
||||
let mut existing_item = assistant_message("existing");
|
||||
existing_item.set_turn_id_if_missing("older-turn");
|
||||
|
||||
session
|
||||
.record_conversation_items(&turn_context, &[fresh_item.clone(), existing_item.clone()])
|
||||
.await;
|
||||
|
||||
let mut expected_fresh_item = fresh_item;
|
||||
expected_fresh_item.set_turn_id_if_missing(&turn_context.sub_id);
|
||||
let expected_items = vec![expected_fresh_item, existing_item];
|
||||
assert_eq!(
|
||||
session.clone_history().await.raw_items(),
|
||||
expected_items.as_slice()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn record_inter_agent_communication_sets_turn_id_in_rollout_and_resume() {
|
||||
let (mut session, turn_context) = make_session_and_context().await;
|
||||
let rollout_path = attach_thread_persistence(&mut session).await;
|
||||
let communication = InterAgentCommunication::new(
|
||||
AgentPath::root().join("worker").expect("worker path"),
|
||||
AgentPath::root(),
|
||||
Vec::new(),
|
||||
"child done".to_string(),
|
||||
/*trigger_turn*/ false,
|
||||
);
|
||||
let mut expected_item = communication.to_model_input_item();
|
||||
expected_item.set_turn_id_if_missing(&turn_context.sub_id);
|
||||
|
||||
session
|
||||
.record_inter_agent_communication(&turn_context, communication)
|
||||
.await;
|
||||
|
||||
assert_eq!(
|
||||
session.clone_history().await.raw_items(),
|
||||
std::slice::from_ref(&expected_item)
|
||||
);
|
||||
|
||||
session.flush_rollout().await.expect("rollout should flush");
|
||||
let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path)
|
||||
.await
|
||||
.expect("read rollout history")
|
||||
else {
|
||||
panic!("expected resumed rollout history");
|
||||
};
|
||||
|
||||
let (resumed_session, _resumed_turn_context) = make_session_and_context().await;
|
||||
resumed_session
|
||||
.record_initial_history(InitialHistory::Resumed(resumed))
|
||||
.await;
|
||||
assert_eq!(
|
||||
resumed_session.clone_history().await.raw_items(),
|
||||
std::slice::from_ref(&expected_item)
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn prepares_image_failures_before_history_insertion() {
|
||||
let (session, turn_context, _rx) = make_session_and_context_with_auth_and_config_and_rx(
|
||||
@@ -1758,7 +1820,7 @@ async fn prepares_image_failures_before_history_insertion() {
|
||||
},
|
||||
internal_chat_message_metadata_passthrough: None,
|
||||
}];
|
||||
assert_eq!(history.raw_items(), expected.as_slice());
|
||||
assert_eq!(strip_metadata_from_items(history.raw_items()), expected);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -8224,7 +8286,7 @@ async fn handle_output_item_done_records_image_save_history_message() {
|
||||
),
|
||||
);
|
||||
let expected = vec![image_message, item];
|
||||
assert_eq!(history.raw_items(), expected.as_slice());
|
||||
assert_eq!(strip_metadata_from_items(history.raw_items()), expected);
|
||||
assert_eq!(
|
||||
std::fs::read(&expected_saved_path).expect("saved file"),
|
||||
b"foo"
|
||||
@@ -8267,7 +8329,7 @@ async fn handle_output_item_done_skips_image_save_message_when_save_fails() {
|
||||
|
||||
let history = session.clone_history().await;
|
||||
let expected = vec![item];
|
||||
assert_eq!(history.raw_items(), expected.as_slice());
|
||||
assert_eq!(strip_metadata_from_items(history.raw_items()), expected);
|
||||
assert!(!expected_saved_path.exists());
|
||||
}
|
||||
|
||||
@@ -9019,7 +9081,7 @@ async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input()
|
||||
internal_chat_message_metadata_passthrough: None,
|
||||
};
|
||||
assert!(
|
||||
history.raw_items().iter().any(|item| item == &expected),
|
||||
strip_metadata_from_items(history.raw_items()).contains(&expected),
|
||||
"expected pending input to be persisted into history on turn completion"
|
||||
);
|
||||
|
||||
|
||||
@@ -334,11 +334,13 @@ mod tests {
|
||||
session
|
||||
.record_conversation_items(&turn, std::slice::from_ref(&history_item))
|
||||
.await;
|
||||
let mut expected_history_item = history_item.clone();
|
||||
expected_history_item.set_turn_id_if_missing(&turn_id);
|
||||
let raw_history_event = rx.recv().await.expect("history raw response item event");
|
||||
let EventMsg::RawResponseItem(raw_history_item) = raw_history_event.msg else {
|
||||
panic!("expected raw response item event");
|
||||
};
|
||||
assert_eq!(raw_history_item.item, history_item);
|
||||
assert_eq!(raw_history_item.item, expected_history_item);
|
||||
let invocation = ToolInvocation {
|
||||
session,
|
||||
turn,
|
||||
@@ -377,7 +379,7 @@ mod tests {
|
||||
);
|
||||
assert_eq!(
|
||||
captured_call.conversation_history.items(),
|
||||
std::slice::from_ref(&history_item)
|
||||
std::slice::from_ref(&expected_history_item)
|
||||
);
|
||||
match captured_call.payload {
|
||||
ToolPayload::Function { arguments } => {
|
||||
|
||||
@@ -345,6 +345,8 @@ async fn extension_tool_executors_are_model_visible_and_dispatchable() -> anyhow
|
||||
session
|
||||
.record_conversation_items(&turn, std::slice::from_ref(&history_item))
|
||||
.await;
|
||||
let mut expected_history_item = history_item.clone();
|
||||
expected_history_item.set_turn_id_if_missing(&turn.sub_id);
|
||||
|
||||
let router = ToolRouter::from_turn_context(
|
||||
&turn,
|
||||
@@ -404,7 +406,7 @@ async fn extension_tool_executors_are_model_visible_and_dispatchable() -> anyhow
|
||||
json!({
|
||||
"arguments": { "message": "hello" },
|
||||
"callId": "call-extension",
|
||||
"conversationHistory": [history_item],
|
||||
"conversationHistory": [expected_history_item],
|
||||
"ok": true,
|
||||
})
|
||||
);
|
||||
|
||||
Reference in New Issue
Block a user