Stamp conversation history items with creation times (#38272)

## What changed

- Add fractional Unix creation times to locally authored user, developer, agent, and tool-output items when they enter durable conversation history.
- Preserve any creation time already supplied with an item across subsequent requests, remote compaction, and resumed history.

## Testing

- Cover timestamp assignment and preservation in session, client, tool, agent-spawn, and remote-compaction tests.

GitOrigin-RevId: 431d1770b9746005c808bc0a43953845166d68ad
This commit is contained in:
ashwinnathan-openai
2026-08-12 23:04:42 +00:00
committed by copyberry
parent 842fae26c9
commit 361fe2d202
10 changed files with 196 additions and 22 deletions
+1
View File
@@ -17,6 +17,7 @@ struct SpawnAgentThreadInheritance {
/// submission and lifecycle logging cannot receive one without the other. Other spawn sources
/// provide user input directly, making an uncontextualized inter-agent communication
/// unrepresentable.
#[allow(clippy::large_enum_variant)]
enum SpawnInitialInput {
UserInput(Vec<UserInput>),
InterAgentCommunication(InterAgentCommunication, AgentCommunicationContext),
+7 -1
View File
@@ -1477,6 +1477,7 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
ThreadHistoryMode::Legacy
);
let history = child_thread.session.clone_history().await;
let history_items = history.raw_items().cloned().collect::<Vec<_>>();
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);
@@ -1496,6 +1497,12 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
internal_chat_message_metadata_passthrough: None,
};
expected_developer_message.set_turn_id_if_missing(&turn_context.sub_id);
expected_developer_message.set_create_time_if_missing(
history_items[1]
.executed_tool_call_metadata()
.and_then(|metadata| metadata.create_time.clone())
.expect("recorded developer message should have a creation timestamp"),
);
let expected_history = [
expected_parent_seed,
expected_developer_message,
@@ -1510,7 +1517,6 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
internal_chat_message_metadata_passthrough: None,
},
];
let history_items = history.raw_items().cloned().collect::<Vec<_>>();
assert_eq!(
strip_response_item_ids(&history_items),
strip_response_item_ids(&expected_history),
+16 -3
View File
@@ -2948,6 +2948,20 @@ impl Session {
}
}
pub(crate) fn response_item_create_time() -> serde_json::Number {
let create_time = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default();
serde_json::Number::from_f64(create_time.as_secs_f64())
.unwrap_or_else(|| serde_json::Number::from(create_time.as_secs()))
}
/// Adds the turn ID and creation time to newly created conversation items.
pub(crate) fn stamp_response_item_for_history(item: &mut ResponseItem, turn_id: &str) {
item.set_turn_id_if_missing(turn_id);
item.set_create_time_if_missing(Self::response_item_create_time());
}
/// Records conversation items: append to history, persist to rollout, and
/// notify clients observing raw response items.
pub(crate) fn prepare_conversation_items_for_history<'a>(
@@ -2981,7 +2995,7 @@ impl Session {
prepare_audio_response_items(&mut items);
// Most response items get their passthrough turn ID at the durable history boundary.
for item in &mut items {
item.set_turn_id_if_missing(&turn_context.sub_id);
Self::stamp_response_item_for_history(item, &turn_context.sub_id);
}
let items = Cow::Owned(items);
(
@@ -3223,9 +3237,8 @@ impl Session {
pub(crate) async fn record_inter_agent_communication(
&self,
turn_context: &TurnContext,
mut communication: InterAgentCommunication,
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,
+32 -11
View File
@@ -2085,13 +2085,23 @@ async fn record_conversation_items_stamps_missing_turn_id_and_preserves_existing
.record_conversation_items(&turn_context, &[fresh_item.clone(), existing_item.clone()])
.await;
let history = session.clone_history().await;
let recorded_items = raw_history_items(&history);
let fresh_create_time = recorded_items[0]
.executed_tool_call_metadata()
.and_then(|metadata| metadata.create_time.clone())
.expect("harness-authored items should receive creation timestamps");
assert!(
fresh_create_time
.as_f64()
.is_some_and(|seconds| seconds > 0.0)
);
let mut expected_fresh_item = fresh_item;
expected_fresh_item.set_turn_id_if_missing(&turn_context.sub_id);
expected_fresh_item.set_create_time_if_missing(fresh_create_time);
let expected_items = vec![expected_fresh_item, existing_item];
assert_eq!(
raw_history_items(&session.clone_history().await),
expected_items
);
assert_eq!(recorded_items, expected_items);
}
#[tokio::test]
@@ -2191,8 +2201,17 @@ async fn record_inter_agent_communication_sets_turn_id_in_rollout_and_resume() {
.record_inter_agent_communication(&turn_context, communication)
.await;
let recorded_history = session.clone_history().await;
let recorded_items = raw_history_items(&recorded_history);
let create_time = recorded_items[0]
.executed_tool_call_metadata()
.and_then(|metadata| metadata.create_time.clone())
.expect("locally authored agent message should receive a creation timestamp");
assert!(create_time.as_f64().is_some_and(|seconds| seconds > 0.0));
expected_item.set_create_time_if_missing(create_time);
assert_eq!(
strip_response_item_ids(&raw_history_items(&session.clone_history().await)),
strip_response_item_ids(&recorded_items),
strip_response_item_ids(std::slice::from_ref(&expected_item))
);
@@ -2564,8 +2583,10 @@ async fn resumed_history_injects_initial_context_on_first_context_update_only()
expected.extend(initial_context);
let history_after_seed = session.clone_history().await;
assert_eq!(
strip_response_item_ids(&expected),
strip_response_item_ids(&raw_history_items(&history_after_seed))
strip_response_item_ids(&strip_metadata_from_items(&expected)),
strip_response_item_ids(&strip_metadata_from_items(&raw_history_items(
&history_after_seed
)))
);
session
@@ -9606,8 +9627,8 @@ async fn record_context_updates_and_set_reference_context_item_injects_full_cont
let history = session.clone_history().await;
let initial_context = build_initial_context(&session, &turn_context).await;
assert_eq!(
strip_response_item_ids(&raw_history_items(&history)),
strip_response_item_ids(&initial_context)
strip_response_item_ids(&strip_metadata_from_items(&raw_history_items(&history))),
strip_response_item_ids(&strip_metadata_from_items(&initial_context))
);
let current_context = session.reference_context_item().await;
@@ -9661,8 +9682,8 @@ async fn record_context_updates_and_set_reference_context_item_reinjects_full_co
let initial_context = build_initial_context(&session, &turn_context).await;
expected_history.extend(initial_context);
assert_eq!(
strip_response_item_ids(&raw_history_items(&history)),
strip_response_item_ids(&expected_history)
strip_response_item_ids(&strip_metadata_from_items(&raw_history_items(&history))),
strip_response_item_ids(&strip_metadata_from_items(&expected_history))
);
}
@@ -382,8 +382,15 @@ 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 expected_history_item = strip_response_item_id(
session
.clone_history()
.await
.raw_items()
.next()
.expect("history item")
.clone(),
);
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");
+7 -2
View File
@@ -545,8 +545,13 @@ 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 expected_history_item = session
.clone_history()
.await
.raw_items()
.next()
.expect("history item")
.clone();
let router = test_tool_router(
step_context.as_ref(),
+34 -1
View File
@@ -196,12 +196,17 @@ fn assert_codex_client_metadata(
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn openai_stateless_responses_requests_preserve_item_turn_metadata_across_turns() {
let server = MockServer::start().await;
let assistant_create_time = 1_785_276_138.422709;
let mut assistant_message = ev_assistant_message("msg-1", "first answer");
assistant_message["item"]["internal_chat_message_metadata_passthrough"] = json!({
"create_time": assistant_create_time,
});
let response_mock = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_response_created("resp1"),
ev_assistant_message("msg-1", "first answer"),
assistant_message,
ev_completed("resp1"),
]),
sse(vec![ev_response_created("resp2"), ev_completed("resp2")]),
@@ -234,6 +239,14 @@ async fn openai_stateless_responses_requests_preserve_item_turn_metadata_across_
Some(first_turn_id)
);
}
for role in ["user", "developer"] {
assert!(first_input.iter().any(|item| {
item["role"].as_str() == Some(role)
&& item["internal_chat_message_metadata_passthrough"]["create_time"]
.as_f64()
.is_some_and(|create_time| create_time > 0.0)
}));
}
let item_turn_id = |text: &str| {
second_input
@@ -250,6 +263,26 @@ async fn openai_stateless_responses_requests_preserve_item_turn_metadata_across_
assert_eq!(item_turn_id("turn one"), Some(first_turn_id));
assert_eq!(item_turn_id("first answer"), Some(first_turn_id));
assert_eq!(item_turn_id("turn two"), Some(second_turn_id));
let item_create_time = |text: &str| {
second_input
.iter()
.find(|item| {
item["content"].as_array().is_some_and(|content| {
content
.iter()
.any(|content_item| content_item["text"].as_str() == Some(text))
})
})
.and_then(|item| {
item["internal_chat_message_metadata_passthrough"]["create_time"].as_f64()
})
};
assert_eq!(
item_create_time("first answer"),
Some(assistant_create_time)
);
assert!(item_create_time("turn two").is_some_and(|create_time| create_time > 0.0));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -1380,6 +1380,27 @@ async fn remote_compact_v2_reuses_compaction_trigger_for_followups() -> Result<(
let response_requests = responses_mock.requests();
let compact_request = &response_requests[2];
let item_create_time = |request: &responses::ResponsesRequest, text: &str| {
request
.input()
.into_iter()
.find(|item| {
item["content"].as_array().is_some_and(|content| {
content.iter().any(|part| {
part["text"].as_str() == Some(text)
|| part["encrypted_content"].as_str() == Some(text)
})
})
})
.and_then(|item| {
item.pointer("/internal_chat_message_metadata_passthrough/create_time")
.cloned()
})
.expect("matching message should include a creation timestamp")
};
let original_user_create_time = item_create_time(&response_requests[0], "hello remote compact");
let delegated_task_create_time =
item_create_time(&response_requests[1], &delegated_task_ciphertext);
assert!(
compact_request
.inputs_of_type("agent_message")
@@ -1444,6 +1465,19 @@ async fn remote_compact_v2_reuses_compaction_trigger_for_followups() -> Result<(
);
let follow_up_request = response_requests.last().expect("follow-up request missing");
assert_eq!(
item_create_time(follow_up_request, "hello remote compact"),
original_user_create_time
);
assert_eq!(
item_create_time(follow_up_request, &delegated_task_ciphertext),
delegated_task_create_time
);
assert!(
item_create_time(follow_up_request, "after compact")
.as_f64()
.is_some_and(|create_time| create_time > 0.0)
);
assert!(
follow_up_request
.inputs_of_type("agent_message")
+7 -1
View File
@@ -409,10 +409,15 @@ async fn namespaced_custom_tool_call_preserves_namespace_through_dispatch_and_re
.and_then(Value::as_str)
.map(str::to_string)
.expect("custom tool call should include turn metadata");
let custom_tool_output = request.custom_tool_call_output(call_id);
let output_create_time = custom_tool_output
.pointer("/internal_chat_message_metadata_passthrough/create_time")
.and_then(Value::as_f64)
.expect("custom tool output should include a creation timestamp");
assert_eq!(
(
strip_response_item_ids_from_json(Value::Array(custom_tool_calls)),
strip_response_item_ids_from_json(request.custom_tool_call_output(call_id)),
strip_response_item_ids_from_json(custom_tool_output),
),
(
Value::Array(vec![json!({
@@ -431,6 +436,7 @@ async fn namespaced_custom_tool_call_preserves_namespace_through_dispatch_and_re
"output": format!("unsupported custom tool call: {namespace}{tool_name}"),
"internal_chat_message_metadata_passthrough": {
"turn_id": turn_id,
"create_time": output_create_time,
"executed_tool_calls": [{
"name": format!("{namespace}__{tool_name}"),
"arguments": input,
+49 -1
View File
@@ -790,6 +790,11 @@ pub struct InternalChatMessageMetadataPassthrough {
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional)]
pub turn_id: Option<String>,
/// Message creation time in fractional Unix seconds.
#[serde(default, skip_serializing_if = "Option::is_none")]
#[schemars(skip)]
#[ts(skip)]
pub create_time: Option<serde_json::Number>,
/// Warehouse-only Responses metadata, not part of the public app-server protocol.
#[serde(default, skip_deserializing, skip_serializing_if = "Option::is_none")]
#[schemars(skip)]
@@ -1133,6 +1138,39 @@ impl ResponseItem {
InternalChatMessageMetadataPassthrough::set_turn_id_if_missing(metadata, turn_id);
}
/// Stamps a harness-authored durable item without replacing its creation time.
pub fn set_create_time_if_missing(&mut self, create_time: serde_json::Number) {
let metadata = match self {
Self::Message {
role,
internal_chat_message_metadata_passthrough: metadata,
..
} if matches!(role.as_str(), "user" | "developer") => metadata,
Self::AgentMessage {
internal_chat_message_metadata_passthrough: metadata,
..
}
| Self::FunctionCallOutput {
internal_chat_message_metadata_passthrough: metadata,
..
}
| Self::CustomToolCallOutput {
internal_chat_message_metadata_passthrough: metadata,
..
}
| Self::ToolSearchOutput {
internal_chat_message_metadata_passthrough: metadata,
..
} => metadata,
_ => return,
};
metadata
.get_or_insert_default()
.create_time
.get_or_insert(create_time);
}
/// Removes internal chat message metadata passthrough before sending to a provider that does
/// not accept it.
pub fn clear_internal_chat_message_metadata_passthrough(&mut self) {
@@ -2357,9 +2395,19 @@ mod tests {
let mut missing_turn_id = response_item_with_passthrough_metadata(
/*internal_chat_message_metadata_passthrough*/ None,
);
let create_time = serde_json::Number::from(123);
missing_turn_id.set_create_time_if_missing(create_time.clone());
missing_turn_id.set_turn_id_if_missing("");
missing_turn_id.set_turn_id_if_missing("turn-1");
assert_eq!(missing_turn_id.turn_id(), Some("turn-1"));
missing_turn_id.set_create_time_if_missing(serde_json::Number::from(456));
assert_eq!(
missing_turn_id.executed_tool_call_metadata(),
Some(&InternalChatMessageMetadataPassthrough {
turn_id: Some("turn-1".to_string()),
create_time: Some(create_time),
..Default::default()
})
);
let mut other = ResponseItem::Other;
other.set_turn_id_if_missing("turn-1");